feat: optimize live workflow log monitoring

src/workflow-web.ts

@@ -174,6 +174,111 @@ export function splitCompleteUtf8(buffer: Buffer): { complete: Buffer; residual:
   return { complete: buffer, residual: Buffer.alloc(0) as Buffer };
 }
 
+/**
+ * Advance past a leading incomplete UTF-8 sequence when reading mid-file.
+ * Returns the number of leading bytes to skip (0-3).
+ */
+export function leadingIncompleteUtf8Bytes(buffer: Buffer): number {
+  if (buffer.length === 0) return 0;
+  // Continuation bytes 10xxxxxx at the start mean we began mid-codepoint.
+  let i = 0;
+  while (i < Math.min(buffer.length, 3) && (buffer[i] & 0xC0) === 0x80) i += 1;
+  return i;
+}
+
+export type LogSliceResult = {
+  content: string;
+  startOffset: number;
+  endOffset: number;
+  nextOffset: number;
+  size: number;
+  hasMore: boolean;
+  hasMoreBefore: boolean;
+};
+
+/**
+ * Read a UTF-8-safe byte slice of a log file.
+ * - Forward: offset + limit
+ * - Tail: last `limit` bytes (tail=true)
+ * - Reverse: bytes ending at `before` (before exclusive)
+ */
+export function readLogSlice(
+  logPath: string,
+  opts: { offset?: number; limit: number; tail?: boolean; before?: number },
+): LogSliceResult {
+  const stat = fs.statSync(logPath);
+  const size = stat.size;
+  const limit = Math.max(1, opts.limit);
+
+  let start: number;
+  let end: number;
+
+  if (opts.tail) {
+    end = size;
+    start = Math.max(0, size - limit);
+  } else if (opts.before !== undefined) {
+    end = Math.max(0, Math.min(opts.before, size));
+    start = Math.max(0, end - limit);
+  } else {
+    start = Math.max(0, Math.min(opts.offset ?? 0, size));
+    end = Math.min(start + limit, size);
+  }
+
+  if (end <= start || size === 0) {
+    const emptyStart = opts.before !== undefined ? Math.min(opts.before, size) : (opts.tail ? size : Math.min(opts.offset ?? 0, size));
+    return {
+      content: "",
+      startOffset: emptyStart,
+      endOffset: emptyStart,
+      nextOffset: emptyStart,
+      size,
+      hasMore: emptyStart < size,
+      hasMoreBefore: emptyStart > 0,
+    };
+  }
+
+  const readLength = end - start;
+  const buffer = Buffer.alloc(readLength);
+  const fd = fs.openSync(logPath, fs.constants.O_RDONLY | (fs.constants.O_NOFOLLOW || 0));
+  try {
+    fs.readSync(fd, buffer, 0, readLength, start);
+  } finally {
+    fs.closeSync(fd);
+  }
+
+  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;
+    }
+  }
+
+  // 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;
+  }
+  const endOffset = startOffset + contentBuf.length;
+  const content = contentBuf.toString("utf8");
+
+  return {
+    content,
+    startOffset,
+    endOffset,
+    nextOffset: endOffset,
+    size,
+    hasMore: endOffset < size,
+    hasMoreBefore: startOffset > 0,
+  };
+}
+
 /**
  * Verify that no segment of targetPath from rootDir downwards is a symlink.
  * Blocks all symlinks (including in-root symlink aliases).
@@ -354,6 +459,9 @@ function handleSseStream(req: http.IncomingMessage, res: http.ServerResponse, ru
   let lastStateStr = "";
   let lastEventsLineCount = 0;
   let currentLogPath = "";
+  let currentLogCycle = 0;
+  let currentLogAttempt = 0;
+  let currentLogType: "exec" | "agy" = "exec";
   let logBytesRead = 0;
   let logResidualBuffer = Buffer.alloc(0);
 
@@ -371,9 +479,14 @@ function handleSseStream(req: http.IncomingMessage, res: http.ServerResponse, ru
     }
   }
 
-  // Helper to resolve active log path from state
+  // Helper to resolve active log identity from state
-  const getActiveLogPath = (stateObj: any) => {
+  const getActiveLogInfo = (stateObj: any): {
-    if (!stateObj || stateObj.currentCycle === undefined) return "";
+    path: string;
+    cycleIndex: number;
+    attemptIndex: number;
+    fileType: "exec" | "agy";
+  } | null => {
+    if (!stateObj || stateObj.currentCycle === undefined) return null;
     const cycle = stateObj.currentCycle;
     let attempt = 0;
     const progressFile = path.join(runDir, ".progress.json");
@@ -389,15 +502,20 @@ function handleSseStream(req: http.IncomingMessage, res: http.ServerResponse, ru
       } catch {
         // ignore
       }
-    } else {
+    } else if (stateObj.cycles) {
-      if (stateObj.cycles) {
       const cyRec = stateObj.cycles.find((cy: any) => cy.cycleIndex === cycle);
       if (cyRec && cyRec.executorAttemptLogs) {
         attempt = Math.max(0, cyRec.executorAttemptLogs.length - 1);
       }
     }
-    }
+    // Stream the primary executor attempt log (exec). Agy session logs are
-    return attemptLogPath(runDir, cycle, attempt);
+    // available via the REST API with type=agy and must not be mixed here.
+    return {
+      path: attemptLogPath(runDir, cycle, attempt),
+      cycleIndex: cycle,
+      attemptIndex: attempt,
+      fileType: "exec",
+    };
   };
 
   const poll = () => {
@@ -437,20 +555,25 @@ function handleSseStream(req: http.IncomingMessage, res: http.ServerResponse, ru
       }
 
       if (stateObj) {
-        const activeLog = getActiveLogPath(stateObj);
+        const activeLog = getActiveLogInfo(stateObj);
-        if (activeLog && activeLog !== currentLogPath) {
+        if (activeLog && activeLog.path !== currentLogPath) {
-          currentLogPath = activeLog;
+          currentLogPath = activeLog.path;
+          currentLogCycle = activeLog.cycleIndex;
+          currentLogAttempt = activeLog.attemptIndex;
+          currentLogType = activeLog.fileType;
           logBytesRead = 0;
           logResidualBuffer = Buffer.alloc(0);
 
-          if (fs.existsSync(activeLog) && !fs.lstatSync(activeLog).isSymbolicLink()) {
+          if (fs.existsSync(activeLog.path) && !fs.lstatSync(activeLog.path).isSymbolicLink()) {
-            logBytesRead = fs.statSync(activeLog).size;
+            logBytesRead = fs.statSync(activeLog.path).size;
           }
 
           pushEvent("log_rotation", JSON.stringify({
-            cycleIndex: stateObj.currentCycle,
+            cycleIndex: activeLog.cycleIndex,
-            attemptIndex: stateObj.cycles && stateObj.cycles[stateObj.currentCycle - 1] && stateObj.cycles[stateObj.currentCycle - 1].executorAttemptLogs ? Math.max(0, stateObj.cycles[stateObj.currentCycle - 1].executorAttemptLogs.length - 1) : 0,
+            attemptIndex: activeLog.attemptIndex,
-            logFilePath: activeLog,
+            fileType: activeLog.fileType,
+            logFilePath: activeLog.path,
+            size: logBytesRead,
           }));
         }
       }
@@ -468,6 +591,7 @@ function handleSseStream(req: http.IncomingMessage, res: http.ServerResponse, ru
           fs.readSync(fd, buffer, 0, sizeToRead, logBytesRead);
           fs.closeSync(fd);
 
+          const startOffset = logBytesRead;
           logBytesRead += sizeToRead;
 
           const combined = Buffer.concat([logResidualBuffer, buffer]) as Buffer;
@@ -475,7 +599,18 @@ function handleSseStream(req: http.IncomingMessage, res: http.ServerResponse, ru
           logResidualBuffer = residual as any;
 
           if (complete.length > 0) {
-            pushEvent("log_append", complete.toString("utf8"));
+            const endOffset = startOffset + complete.length - residual.length;
+            // endOffset is the exclusive byte offset of complete content from original file start:
+            // startOffset is where this read began; residual is incomplete trailing bytes still buffered.
+            const completeEnd = logBytesRead - residual.length;
+            pushEvent("log_append", JSON.stringify({
+              cycleIndex: currentLogCycle,
+              attemptIndex: currentLogAttempt,
+              fileType: currentLogType,
+              startOffset: completeEnd - complete.length,
+              endOffset: completeEnd,
+              content: complete.toString("utf8"),
+            }));
           }
         }
       }
@@ -497,11 +632,17 @@ function handleSseStream(req: http.IncomingMessage, res: http.ServerResponse, ru
       const stateStr = fs.readFileSync(fd, "utf8");
       fs.closeSync(fd);
       const stateObj = JSON.parse(stateStr);
-      currentLogPath = getActiveLogPath(stateObj);
+      const info = getActiveLogInfo(stateObj);
+      if (info) {
+        currentLogPath = info.path;
+        currentLogCycle = info.cycleIndex;
+        currentLogAttempt = info.attemptIndex;
+        currentLogType = info.fileType;
         if (currentLogPath && fs.existsSync(currentLogPath) && !fs.lstatSync(currentLogPath).isSymbolicLink()) {
           logBytesRead = fs.statSync(currentLogPath).size;
         }
       }
+    }
   } catch {
     // ignore
   }
@@ -677,22 +818,34 @@ export function startWorkflowWebServer(port = 4317): Promise<http.Server> {
 
         const cycleStr = parsedUrl.searchParams.get("cycle") || "1";
         const attemptStr = parsedUrl.searchParams.get("attempt") || "0";
-        const offsetStr = parsedUrl.searchParams.get("offset") || "0";
+        const offsetStr = parsedUrl.searchParams.get("offset");
+        const beforeStr = parsedUrl.searchParams.get("before");
+        const tailStr = parsedUrl.searchParams.get("tail");
         const limitStr = parsedUrl.searchParams.get("limit") || "100000";
 
         const cycle = parseInt(cycleStr, 10);
         const attempt = parseInt(attemptStr, 10);
-        const offset = parseInt(offsetStr, 10);
         const limit = parseInt(limitStr, 10);
+        const wantsTail = tailStr === "1" || tailStr === "true";
+        const wantsBefore = beforeStr !== null && beforeStr !== undefined && beforeStr !== "";
 
         if (
           !/^\d+$/.test(cycleStr) || cycle < 1 || cycle > 1000 ||
           !/^\d+$/.test(attemptStr) || attempt < 0 || attempt > 1000 ||
-          !/^\d+$/.test(offsetStr) || offset < 0 || offset > 100 * 1024 * 1024 ||
           !/^\d+$/.test(limitStr) || limit < 4 || limit > 10 * 1024 * 1024
         ) {
           return sendError(res, 400, "Bad Request: Invalid numeric parameters");
         }
+        if (wantsBefore) {
+          if (!/^\d+$/.test(beforeStr!) || parseInt(beforeStr!, 10) < 0 || parseInt(beforeStr!, 10) > 100 * 1024 * 1024) {
+            return sendError(res, 400, "Bad Request: Invalid before parameter");
+          }
+        } else if (!wantsTail) {
+          const effectiveOffset = offsetStr ?? "0";
+          if (!/^\d+$/.test(effectiveOffset) || parseInt(effectiveOffset, 10) < 0 || parseInt(effectiveOffset, 10) > 100 * 1024 * 1024) {
+            return sendError(res, 400, "Bad Request: Invalid numeric parameters");
+          }
+        }
 
         const type = parsedUrl.searchParams.get("type") || "exec";
         if (type !== "exec" && type !== "agy") {
@@ -715,49 +868,42 @@ export function startWorkflowWebServer(port = 4317): Promise<http.Server> {
             res.writeHead(200, { "Content-Type": "application/json" });
             return res.end();
           }
-          return sendJson(res, 200, { content: "", nextOffset: 0, size: 0, hasMore: false });
+          return sendJson(res, 200, {
+            content: "",
+            nextOffset: 0,
+            startOffset: 0,
+            endOffset: 0,
+            size: 0,
+            hasMore: false,
+            hasMoreBefore: false,
+          });
         }
 
-        let logContent = "";
-        let fileSize = 0;
-        let nextOffset = offset;
         try {
-          const stat = fs.statSync(logPath);
+          const slice = readLogSlice(logPath, {
-          fileSize = stat.size;
+            limit,
-          const start = Math.min(offset, fileSize);
+            ...(wantsTail
-          const end = Math.min(start + limit, fileSize);
+              ? { tail: true }
-          const readLength = end - start;
+              : wantsBefore
-
+                ? { before: parseInt(beforeStr!, 10) }
-          if (readLength > 0) {
+                : { offset: parseInt(offsetStr ?? "0", 10) }),
-            const fd = fs.openSync(logPath, fs.constants.O_RDONLY | (fs.constants.O_NOFOLLOW || 0));
+          });
-            const buffer = Buffer.alloc(readLength);
-            fs.readSync(fd, buffer, 0, readLength, start);
-            fs.closeSync(fd);
-            const { complete } = splitCompleteUtf8(buffer);
-            if (complete.length === 0) {
-              logContent = buffer.toString("utf8");
-              nextOffset = start + buffer.length;
-            } else {
-              logContent = complete.toString("utf8");
-              nextOffset = start + complete.length;
-            }
-          } else {
-            nextOffset = end;
-          }
-        } catch (err: any) {
-          return sendError(res, 500, `Error reading log file: ${err.message}`);
-        }
-
           if (method === "HEAD") {
             res.writeHead(200, { "Content-Type": "application/json" });
             return res.end();
           }
           return sendJson(res, 200, {
-          content: logContent,
+            content: slice.content,
-          nextOffset,
+            nextOffset: slice.nextOffset,
-          size: fileSize,
+            startOffset: slice.startOffset,
-          hasMore: nextOffset < fileSize,
+            endOffset: slice.endOffset,
+            size: slice.size,
+            hasMore: slice.hasMore,
+            hasMoreBefore: slice.hasMoreBefore,
           });
+        } catch (err: any) {
+          return sendError(res, 500, `Error reading log file: ${err.message}`);
+        }
       }
 
       const streamMatch = pathname.match(/^\/api\/runs\/([a-zA-Z0-9_\-\:]+)\/stream$/);