fix: harden live workflow log streaming

src/workflow-web-assets.ts

@@ -1788,6 +1788,31 @@ export function getAssetsHtml(): string {
         if (container && parsed.length) {
           const frag = document.createDocumentFragment();
           parsed.forEach(entry => {
+            if (entry.type === 'progress') {
+              // Container already has a progress row from a newer page —
+              // skip older progress so we never overwrite newest values.
+              const containerRow = container.querySelector('[data-progress-key="progress"]');
+              if (containerRow) {
+                // The container row holds the most recent progress value;
+                // historical entries must not overwrite it.
+                return;
+              }
+              // No container row — check whether this batch already queued
+              // one (fragment rows are invisible to container.querySelector).
+              const fragRow = frag.querySelector('[data-progress-key="progress"]');
+              if (fragRow) {
+                // Update fragment-local row — entries arrive in file order
+                // so the last one is most recent.
+                const content = fragRow.querySelector('.item-content');
+                if (content) content.textContent = entry.content || entry.raw || '';
+                return;
+              }
+              // First progress entry we have seen — create a keyed row.
+              entry._progressKey = 'progress';
+              const row = buildTimelineRow(entry);
+              if (row) frag.appendChild(row);
+              return;
+            }
             const row = buildTimelineRow(entry);
             if (row) frag.appendChild(row);
           });
@@ -1899,18 +1924,30 @@ export function getAssetsHtml(): string {
       const container = document.getElementById('timeline-items-container');
       if (!container || !entries || !entries.length) return;
       const frag = document.createDocumentFragment();
+      // Track the most recent progress row in THIS batch so multiple progress
+      // entries arriving in one call still merge into one row.
+      let batchProgressRow = null;
       entries.forEach(entry => {
         // Merge high-frequency progress counters into one updatable row.
         if (entry.type === 'progress') {
           const key = 'progress';
           entry._progressKey = key;
-          const existing = container.querySelector('[data-progress-key="progress"]');
+          // Check batch-local row first (same call), then container (prior calls).
+          const existing = batchProgressRow || container.querySelector('[data-progress-key="progress"]');
           if (existing) {
             const content = existing.querySelector('.item-content');
             if (content) content.textContent = entry.content || entry.raw || '';
             timelineProgressRow = existing;
+            batchProgressRow = existing;
             return;
           }
+          // First progress in this batch – create the row.
+          const row = buildTimelineRow(entry);
+          if (row) {
+            batchProgressRow = row;
+            frag.appendChild(row);
+          }
+          return;
         }
         const row = buildTimelineRow(entry);
         if (row) frag.appendChild(row);
@@ -1945,13 +1982,30 @@ export function getAssetsHtml(): string {
       container.innerHTML = '';
       timelineProgressRow = null;
       const frag = document.createDocumentFragment();
+      // Merge high-frequency progress entries into a single updatable row
+      // (same logic as appendTimelineEntries).
+      let progressMerged = false;
       timelineEntries.forEach(entry => {
-        if (entry.type === 'progress') entry._progressKey = 'progress';
-        const row = buildTimelineRow(entry);
-        if (row) {
-          if (entry.type === 'progress') timelineProgressRow = row;
-          frag.appendChild(row);
+        if (entry.type === 'progress') {
+          if (!progressMerged) {
+            progressMerged = true;
+            entry._progressKey = 'progress';
+            const row = buildTimelineRow(entry);
+            if (row) {
+              timelineProgressRow = row;
+              frag.appendChild(row);
+            }
+          } else {
+            // Update the existing row's text instead of creating a new one.
+            if (timelineProgressRow) {
+              const contentEl = timelineProgressRow.querySelector('.item-content');
+              if (contentEl) contentEl.textContent = entry.content || entry.raw || '';
+            }
+          }
+          return;
         }
+        const row = buildTimelineRow(entry);
+        if (row) frag.appendChild(row);
       });
       container.appendChild(frag);
       timelineRenderedCount = timelineEntries.length;

src/workflow-web.ts

@@ -249,33 +249,100 @@ export function readLogSlice(
   let usable = buffer;
   let startOffset = start;
 
-  // Skip leading continuation bytes when reading mid-file.
-  if (startOffset > 0) {
-    const skip = leadingIncompleteUtf8Bytes(usable);
-    if (skip > 0) {
-      usable = Buffer.from(usable.subarray(skip));
-      startOffset += skip;
+  const isBeforeMode = opts.before !== undefined;
+
+  if (!isBeforeMode) {
+    // Skip leading continuation bytes when reading mid-file (forward/tail only).
+    if (startOffset > 0) {
+      const skip = leadingIncompleteUtf8Bytes(usable);
+      if (skip > 0) {
+        usable = Buffer.from(usable.subarray(skip));
+        startOffset += skip;
+      }
     }
+
+    // Drop trailing incomplete multi-byte sequence (forward/tail only).
+    const { complete } = splitCompleteUtf8(usable);
+    let contentBuf = complete.length > 0 ? complete : usable;
+    // If complete is empty but we have bytes at EOF, still decode (replacement chars).
+    if (complete.length === 0 && end >= size && usable.length > 0) {
+      contentBuf = usable;
+    }
+    const endOffset = startOffset + contentBuf.length;
+    const content = contentBuf.toString("utf8");
+
+    return {
+      content,
+      startOffset,
+      endOffset,
+      nextOffset: endOffset,
+      size,
+      hasMore: endOffset < size,
+      hasMoreBefore: startOffset > 0,
+    };
   }
 
-  // Drop trailing incomplete multi-byte sequence.
-  const { complete } = splitCompleteUtf8(usable);
-  let contentBuf = complete.length > 0 ? complete : usable;
-  // If complete is empty but we have bytes at EOF, still decode (replacement chars).
-  if (complete.length === 0 && end >= size && usable.length > 0) {
-    contentBuf = usable;
+  // Before mode: extend the read range backward (up to 3 bytes) so the start
+  // boundary falls on a complete UTF-8 code point.  The end boundary (`end =
+  // before`) is inherently at a non-continuation byte because it was set from
+  // an adjacent page's startOffset (either a tail page that already skipped
+  // leading continuations, or a prior before page that was extended backward
+  // to a lead byte).  This eliminates replacement characters at boundaries
+  // while keeping startOffset/endOffset continuous with adjacent pages.
+  let adjustedStart = start;
+
+  if (adjustedStart > 0) {
+    const backtrack = Math.min(adjustedStart, 3);
+    const probeBuf = Buffer.alloc(backtrack);
+    const pfd = fs.openSync(logPath, fs.constants.O_RDONLY | (fs.constants.O_NOFOLLOW || 0));
+    try {
+      fs.readSync(pfd, probeBuf, 0, backtrack, adjustedStart - backtrack);
+    } finally {
+      fs.closeSync(pfd);
+    }
+    // Scan backward from the byte just before `start` looking for a lead byte.
+    for (let i = probeBuf.length - 1; i >= 0; i--) {
+      if ((probeBuf[i] & 0xC0) === 0xC0) {
+        // Lead byte found – extend start to include the full multi-byte character.
+        adjustedStart = adjustedStart - backtrack + i;
+        break;
+      }
+      if ((probeBuf[i] & 0x80) === 0x00) {
+        // ASCII byte – the character at `start` starts fresh, no adjustment.
+        break;
+      }
+      // Continuation byte – the lead is further back.
+    }
+    // If the probe only found continuation bytes (invalid >3B chain),
+    // leave adjustedStart unchanged; splitCompleteUtf8 will handle it.
   }
-  const endOffset = startOffset + contentBuf.length;
+
+  // Read the adjusted range
+  const readLen = end - adjustedStart;
+  const readBuf = Buffer.alloc(readLen);
+  const rfd = fs.openSync(logPath, fs.constants.O_RDONLY | (fs.constants.O_NOFOLLOW || 0));
+  try {
+    fs.readSync(rfd, readBuf, 0, readLen, adjustedStart);
+  } finally {
+    fs.closeSync(rfd);
+  }
+
+  // Drop trailing incomplete sequence (e.g. lead byte at EOF whose
+  // continuation bytes haven't been written yet).
+  const { complete } = splitCompleteUtf8(readBuf);
+  let contentBuf = complete.length > 0 ? complete : readBuf;
+  const contentEnd = adjustedStart + contentBuf.length;
+
   const content = contentBuf.toString("utf8");
 
   return {
     content,
-    startOffset,
-    endOffset,
-    nextOffset: endOffset,
+    startOffset: adjustedStart,
+    endOffset: contentEnd,
+    nextOffset: contentEnd,
     size,
-    hasMore: endOffset < size,
-    hasMoreBefore: startOffset > 0,
+    hasMore: contentEnd < size,
+    hasMoreBefore: adjustedStart > 0,
   };
 }