fix(chat): retry incomplete Pi transcript confirmation snapshots

changelog.d/pi-transcript-confirmation.md

@@ -0,0 +1,3 @@
+### Fixed
+- Pi sends retry a briefly incomplete conversation record before giving up transcript confirmation,
+  including when the writer paused before Ovis started reading the file.

server/core/pending-identity-race.test.ts

@@ -1,8 +1,13 @@
 import { afterEach, expect, it, mock, spyOn } from "bun:test";
+import { appendFileSync, mkdtempSync, rmSync, writeFileSync } from "node:fs";
+import { tmpdir } from "node:os";
+import { join } from "node:path";
 
 import type { HerdrPane, ServerMessage, SessionSnapshot } from "../../shared/protocol.ts";
 import * as client from "../herdr/client.ts";
 import * as opencode from "../opencode-native.ts";
+import * as pi from "../pi.ts";
+import * as piDelivery from "../pi-delivery.ts";
 import * as processStart from "../process-start.ts";
 import { connectionData, type Client } from "./connections.ts";
 import { createPendingDelivery } from "./pending.ts";
@@ -81,3 +86,74 @@ it.each([
   expect(frame.removed ?? []).toEqual([]);
   expect(frame.messages.map((message) => message.state)).toEqual([state]);
 });
+
+it.each(["blocked", "working", "queue held", "foreground changed", "foreground disappeared", "lease lost", "unchanged"] as const)(
+  "rechecks queued Pi delivery when capture retries and the pane becomes %s", async (change) => {
+    const root = mkdtempSync(join(tmpdir(), "ovis-pi-capture-race-"));
+    try {
+      const path = join(root, "session.jsonl");
+      writeFileSync(path, JSON.stringify({ type: "session", version: 3, id: "session" }) + "\n"
+        + JSON.stringify({ type: "message", id: "base", parentId: null, message: { role: "assistant", content: [] } }));
+      const pane: Pick<HerdrPane, "pane_id" | "terminal_id" | "agent" | "agent_status"> = { pane_id: "p", terminal_id: "p", agent: "pi", agent_status: "working" };
+      const frames: ServerMessage[] = [], writes: string[] = [];
+      const settled = Promise.withResolvers<void>();
+      const owner = {
+        data: connectionData({ deviceId: undefined, readOnly: false }),
+        getBufferedAmount: () => 0,
+        send: (value: string) => {
+          const frame = JSON.parse(value) as ServerMessage;
+          frames.push(frame);
+          if (frame.type === "pending-messages" && (frame.removed?.length || frame.messages.some((message) => message.state === "uncertain"))) settled.resolve();
+          return 1;
+        },
+      } as Client;
+      owner.data.attached.set("p", { keepSize: false });
+      let foreground = [{ pid: 41001, name: "pi" }], held = false;
+      const facts = { attached: true, origin: true, sameAttachment: true, samePty: true, sameClaim: true, claimed: true, member: true, ready: true, held: false, currentHeld: false, mirror: false, cols: 120 };
+      spyOn(processStart, "processStartedAtAsync").mockResolvedValue(1_000);
+      spyOn(client, "paneProcessInfo").mockImplementation(async () => ({ pane_id: "p", shell_pid: 41000, foreground_processes: foreground }));
+      spyOn(client, "paneRead").mockResolvedValue({ text: "" } as Awaited<ReturnType<typeof client.paneRead>>);
+      spyOn(client, "paneSendText").mockImplementation(async () => { writes.push("paste"); });
+      spyOn(client, "paneSendKeys").mockImplementation(async (_pane, keys, _options, allowed) => {
+        if (allowed && !allowed()) throw new client.HerdrError("cancelled", "Input guard refused the key");
+        writes.push(...keys);
+      });
+      spyOn(pi, "piTranscriptPath").mockResolvedValue({ path, agentDir: null });
+      const capture = piDelivery.capturePiDeliveryAnchorRetrying;
+      let waits = 0;
+      spyOn(piDelivery, "capturePiDeliveryAnchorRetrying").mockImplementation((file, allowMissing, expected) =>
+        capture(file, allowMissing, expected, piDelivery.capturePiDeliveryAnchorAttempt, async () => {
+          waits++;
+          appendFileSync(path, "\n");
+          if (change === "blocked" || change === "working") pane.agent_status = change;
+          if (change === "queue held") held = true;
+          if (change === "foreground changed") foreground = [{ pid: 41002, name: "pi" }];
+          if (change === "foreground disappeared") foreground = [];
+          if (change === "lease lost") facts.samePty = false;
+        }));
+      let tail: Promise<unknown> = Promise.resolve();
+      const serialize = <T>(_pane: string, task: () => T | Promise<T>): Promise<T> => { const run = tail.catch(() => {}).then(task); tail = run; return run; };
+      const delivery = createPendingDelivery({
+        options: { submitDelayMs: 0 }, clients: new Set([owner]), attachments: new Map(),
+        sight: () => ({ origin: null, pty: null, claim: null, facts: () => facts, write: () => {} }),
+        serialize, lastTyped: new Map(), submitText: async () => {}, authorizeSocket: () => {},
+        clientSnapshot: async () => ({ panes: [{ ...pane }] }) as SessionSnapshot,
+        queueHeld: () => held, omo: { sessionOf: () => null },
+        SUBMIT_DEADLINE_MS: 45_000, SUBMIT_DELAY_MS: 0, TYPED_SETTLE_MS: 0,
+      });
+
+      await delivery.pendingFrames.submit(owner, { type: "submit", id: 1, pane_id: "p", text: "FOLLOWUP", payload: "FOLLOWUP", delivery: "queue" });
+      pane.agent_status = "idle";
+      delivery.pending.status("p", "idle");
+      delivery.drainPending("p");
+      await settled.promise;
+      await serialize("p", () => {});
+
+      expect(waits).toBe(1);
+      expect(writes).toEqual(change === "unchanged" ? ["paste", "Enter"] : ["paste"]);
+      if (change !== "unchanged") {
+        expect(delivery.pending.list(owner, "p")).toMatchObject([{ state: "uncertain", error: { code: "submit_changed" } }]);
+        expect(frames.flatMap((frame) => frame.type === "pending-messages" ? frame.removed ?? [] : [])).toEqual([]);
+      }
+    } finally { rmSync(root, { recursive: true, force: true }); }
+  });

server/core/pending.ts

@@ -530,14 +530,19 @@ export function createPendingDelivery({ options, clients, attachments, sight, se
       }
       // fixed wait: the TUI takes the paste before its Enter, which arriving with it could read as a newline
       await Bun.sleep(options.submitDelayMs ?? SUBMIT_DELAY_MS);
-      const beforeEnter = await pendingContext(owner, paneId, lease, identity, true);
+      let beforeEnter = await pendingContext(owner, paneId, lease, identity, true);
       if (automatic && (beforeEnter.working || queueHeld(paneId))) throw new HerdrError("pending_wait", "The agent started another turn before this queued message could be committed");
       authorizePending(owner, paneId, lease);
-      const steering = !automatic && beforeEnter.working;
       // Capture Pi's exact active transcript position before Enter. Status is deliberately not
       // evidence: Pi can report idle/done while its TUI is staging an automatic-compaction input.
+      const beforeCapturePid = beforeEnter.identity.processPid;
       const piAnchor = await preparePiReceipt(paneId, beforeEnter.identity).catch(() => null);
-      if (piAnchor !== undefined) await stillInFront(paneId, beforeEnter.identity.processPid);
+      if (piAnchor !== undefined) {
+        beforeEnter = await pendingContext(owner, paneId, lease, identity, true);
+        if (automatic && (beforeEnter.working || queueHeld(paneId))) throw new HerdrError("pending_wait", "The agent started another turn before this queued message could be committed");
+        await stillInFront(paneId, beforeCapturePid);
+      }
+      const steering = !automatic && beforeEnter.working;
       committing(steering);
       // Claude Code queues what Enter commits during a turn; Ctrl+Enter sends it now (#737). herdr
       // encodes the key as the pane asked, and as a plain CR for a program that asked for nothing.

server/pi-delivery.test.ts

@@ -4,7 +4,7 @@ import { appendFileSync, mkdtempSync, writeFileSync } from "node:fs";
 import { join } from "node:path";
 import { tmpdir } from "node:os";
 
-import { capturePiDeliveryAnchor, capturePiDeliveryAnchorRetrying, matchPiDeliveries, PiReceiptCapacity, type PiDeliveryAnchor, type PiDeliveryCandidate } from "./pi-delivery.ts";
+import { capturePiDeliveryAnchor, capturePiDeliveryAnchorAttempt, capturePiDeliveryAnchorRetrying, matchPiDeliveries, PiReceiptCapacity, type PiDeliveryAnchor, type PiDeliveryCandidate } from "./pi-delivery.ts";
 
 const roots: string[] = [];
 const dir = () => { const root = mkdtempSync(join(tmpdir(), "ovis-pi-delivery-")); roots.push(root); return root; };
@@ -182,6 +182,41 @@ describe("Pi delivery transcript matching", () => {
     expect({ captures, waits }).toEqual({ captures: 3, waits: 2 });
   });
 
+  it.each(["empty file", "partial header", "partial entry"])("retries capture with %s already present until the writer completes it", async (kind) => {
+    const path = join(dir(), "in-flight.jsonl");
+    const start = header(), message = entry("base", null, "assistant", "working");
+    const complete = start + message;
+    const prefix = kind === "empty file" ? "" : kind === "partial header" ? start.slice(0, -1) : complete.slice(0, -1);
+    writeFileSync(path, prefix);
+    // No write occurs during the attempt: equal before/after stats do not prove the writer is done.
+    expect(capturePiDeliveryAnchorAttempt(path)).toEqual({ anchor: null, retry: true });
+    let waits = 0;
+    const anchor = await capturePiDeliveryAnchorRetrying(path, false, undefined, capturePiDeliveryAnchorAttempt,
+      async () => { waits++; appendFileSync(path, complete.slice(prefix.length)); });
+    expect(waits).toBe(1);
+    expect(anchor?.leaf).toBe("base");
+    appendFileSync(path, entry("ours", "base", "user", "after the write"));
+    expect(matchPiDeliveries(path, [candidate("one", "after the write", anchor!)])).toEqual({ consumed: ["one"], changed: [], retry: false });
+  });
+
+  it("bounds retries for a partial record that never completes", async () => {
+    const path = join(dir(), "stalled.jsonl");
+    writeFileSync(path, header() + entry("base", null, "assistant", "working").slice(0, -1));
+    let captures = 0, waits = 0;
+    const anchor = await capturePiDeliveryAnchorRetrying(path, false, undefined,
+      (...args) => { captures++; return capturePiDeliveryAnchorAttempt(...args); }, async () => { waits++; });
+    expect(anchor).toBeNull();
+    expect({ captures, waits }).toEqual({ captures: 3, waits: 2 });
+  });
+
+  it("does not retry a complete malformed record or a mismatching session", () => {
+    const path = join(dir(), "invalid.jsonl"), session = crypto.randomUUID();
+    writeFileSync(path, header(session) + "not JSON\n");
+    expect(capturePiDeliveryAnchorAttempt(path)).toEqual({ anchor: null, retry: false });
+    writeFileSync(path, header(session) + entry("base", null, "assistant", "working"));
+    expect(capturePiDeliveryAnchorAttempt(path, false, "other-session")).toEqual({ anchor: null, retry: false });
+  });
+
   it("does not retry a genuinely unreadable transcript capture", async () => {
     let captures = 0, waits = 0;
     const result = await capturePiDeliveryAnchorRetrying("missing.jsonl", false, undefined,

server/pi-delivery.ts

@@ -64,6 +64,8 @@ const TAIL_BYTES = 64 * 1024;
 const MAX_APPEND_BYTES = 32 * 1024 * 1024;
 /** Permanent for this anchor: retrying cannot shrink its append or retained history. */
 class PiDeliveryBudgetExceeded extends Error {}
+/** A writer may pause between chunks without changing stat during this capture attempt. */
+class PiDeliveryIncomplete extends Error {}
 const identity = (stat: NonNullable<ReturnType<typeof statSync>>): string => `${stat.dev}:${stat.ino}`;
 const fileId = (path: string): string | null => {
   try { const stat = statSync(path); return stat.isFile() ? identity(stat) : null; }
@@ -114,7 +116,7 @@ function parseEntry(bytes: Buffer, start: number, end: number): BranchEntry {
 
 /** Read just the final complete record, not the old branch or the shared tree index. */
 function lastEntry(fd: number, size: number): BranchEntry {
-  if (size === 0 || bytesAt(fd, size - 1, 1)[0] !== 0x0a) throw new Error("Partial Pi entry");
+  if (size === 0 || bytesAt(fd, size - 1, 1)[0] !== 0x0a) throw new PiDeliveryIncomplete("Partial Pi entry");
   let start = size - 1;
   const parts: Buffer[] = [];
   while (start > 0) {
@@ -168,9 +170,9 @@ export function capturePiDeliveryAnchorAttempt(path: string, allowMissing = fals
     const prefix = tailAt(fd, before.size), file = identity(before);
     if (!stable(fd, path, file, before.size, prefix)) return { anchor: null, retry: true };
     return { anchor: { path, file, session, leaf: leaf.start === 0 ? null : leaf.id, size: before.size, prefix }, retry: false };
-  } catch {
+  } catch (error) {
     const current = (() => { try { return statSync(path); } catch { return null; } })();
-    const retry = before !== undefined && current !== null && (identity(current) !== identity(before) || current.size !== before.size || current.mtimeMs !== before.mtimeMs || current.ctimeMs !== before.ctimeMs);
+    const retry = error instanceof PiDeliveryIncomplete || before !== undefined && current !== null && (identity(current) !== identity(before) || current.size !== before.size || current.mtimeMs !== before.mtimeMs || current.ctimeMs !== before.ctimeMs);
     return { anchor: allowMissing && expectedSession && before === undefined && fileId(path) === null
       ? { path, file: null, session: expectedSession, leaf: null, size: 0, prefix: "" } : null, retry };
   } finally { if (fd !== undefined) closeSync(fd); }
@@ -181,7 +183,7 @@ export function capturePiDeliveryAnchor(path: string, allowMissing = false, expe
   return capturePiDeliveryAnchorAttempt(path, allowMissing, expectedSession).anchor;
 }
 
-/** Retry only snapshots invalidated by a concurrent write. An unreadable transcript fails at once. */
+/** Retry incomplete records and snapshots invalidated by a concurrent write. Unreadable transcripts fail at once. */
 export async function capturePiDeliveryAnchorRetrying(path: string, allowMissing = false, expectedSession?: string,
   capture: typeof capturePiDeliveryAnchorAttempt = capturePiDeliveryAnchorAttempt,
   // fixed wait: bounded backoff lets an in-flight Pi append settle before retrying its snapshot