fix(sessions): keep tail follow cursor aligned with bytes actually read (#108127)

sunlit-deng · steipete · web-flow · commit 55a822dc6877 · 2026-07-15T16:00:50.000-07:00
* fix(sessions): keep tail follow cursor aligned with bytes actually read

* refactor(infra): share synchronous file window reads

Co-authored-by: sunlit-deng &lt;yang.jiajun1@xydigit.com&gt;

---------

Co-authored-by: Peter Steinberger &lt;peter@steipete.me&gt;
Co-authored-by: Peter Steinberger &lt;steipete@gmail.com&gt;
diff --git a/src/commands/sessions-tail.test.ts b/src/commands/sessions-tail.test.ts
@@ -352,6 +352,70 @@ describe("sessionsTailCommand", () => {
     expect(output).toContain("bash ok");
   });
 
+  it("delivers appended events across short follow reads without skipping bytes", async () => {
+    const runtime = makeRuntime();
+    await writeSessionEntry();
+    await appendEvents([
+      makeEvent({
+        sourceSeq: 1,
+        type: "session.started",
+        ts: "2026-05-18T12:04:17.000Z",
+      }),
+    ]);
+    const appendedEvent = makeEvent({
+      sourceSeq: 2,
+      type: "tool.result",
+      ts: "2026-05-18T12:04:21.000Z",
+      data: { name: "python", success: true },
+    });
+    // POSIX positional reads may return fewer bytes than requested; cap each
+    // call to prove the bounded delta is filled without skipping bytes.
+    let capReads = false;
+    let shortReadCalls = 0;
+    const realReadSync = fs.readSync.bind(fs);
+    const cappedReadSync = (
+      fd: number,
+      buffer: NodeJS.ArrayBufferView,
+      offset: number,
+      length: number,
+      position: fs.ReadPosition | null,
+    ): number => {
+      const cappedLength = capReads ? Math.min(length, 16) : length;
+      if (capReads) {
+        shortReadCalls += 1;
+      }
+      return realReadSync(fd, buffer, offset, cappedLength, position);
+    };
+    const readSpy = vi
+      .spyOn(fs, "readSync")
+      .mockImplementation(cappedReadSync as typeof fs.readSync);
+    let appended = false;
+    vi.mocked(runtime.log).mockImplementation((message) => {
+      if (!appended && String(message).includes("session.started")) {
+        appended = true;
+        capReads = true;
+        appendJsonl(trajectoryPath, appendedEvent);
+      }
+    });
+
+    const run = sessionsTailCommand(
+      { store: storePath, sessionKey, tail: "1", follow: true },
+      runtime,
+    );
+    try {
+      await waitForRuntimeOutput(runtime, "python ok");
+    } finally {
+      readSpy.mockRestore();
+      process.emit("SIGTERM", "SIGTERM");
+      await run;
+    }
+
+    const output = runtimeOutput(runtime);
+    expect(shortReadCalls).toBeGreaterThan(1);
+    expect(output).toContain("tool.result");
+    expect(output).toContain("python ok");
+  });
+
   it("continues following when later trajectory events are appended", async () => {
     const runtime = makeRuntime();
     await writeSessionEntry();
diff --git a/src/commands/sessions-tail.ts b/src/commands/sessions-tail.ts
@@ -16,6 +16,7 @@ import { parseSqliteSessionFileMarker } from "../config/sessions/sqlite-marker.j
 import type { SessionEntry } from "../config/sessions/types.js";
 import { resolveStoredSessionKeyForAgentStore } from "../gateway/session-store-key.js";
 import { formatErrorMessage } from "../infra/errors.js";
+import { readFileWindowFullySync } from "../infra/file-read.js";
 import { parseStrictNonNegativeInteger } from "../infra/parse-finite-number.js";
 import { readRegularFileSync } from "../infra/regular-file.js";
 import { resolveAgentIdFromSessionKey } from "../routing/session-key.js";
@@ -544,10 +545,10 @@ function readNewFileFollowEvents(state: FileFollowState): TrajectoryEvent[] {
       );
     }
     const buffer = Buffer.alloc(deltaBytes);
-    fs.readSync(fd, buffer, 0, buffer.length, state.offset);
-    state.offset = fileState.size;
+    const bytesRead = readFileWindowFullySync(fd, buffer, state.offset);
+    state.offset += bytesRead;
     state.fileState = fileState;
-    const combined = `${state.pending}${state.decoder.write(buffer)}`;
+    const combined = `${state.pending}${state.decoder.write(buffer.subarray(0, bytesRead))}`;
     // Keep an incomplete trailing JSON line until the next poll, matching
     // append-only writers that flush in chunks.
     const lines = combined.split(/\r?\n/u);
diff --git a/src/infra/file-read.ts b/src/infra/file-read.ts
@@ -1,3 +1,4 @@
+import fs from "node:fs";
 import type { FileHandle } from "node:fs/promises";
 
 /** Fills a bounded positional-read buffer unless the file reaches EOF. */
@@ -21,3 +22,26 @@ export async function readFileWindowFully(
   }
   return bytesRead;
 }
+
+/** Synchronously fills a bounded positional-read buffer unless the file reaches EOF. */
+export function readFileWindowFullySync(
+  fd: number,
+  buffer: Buffer,
+  position: number,
+): number {
+  let bytesRead = 0;
+  while (bytesRead < buffer.length) {
+    const count = fs.readSync(
+      fd,
+      buffer,
+      bytesRead,
+      buffer.length - bytesRead,
+      position + bytesRead,
+    );
+    if (count === 0) {
+      break;
+    }
+    bytesRead += count;
+  }
+  return bytesRead;
+}
