diff --git a/changelog.d/pi-transcript-confirmation.md b/changelog.d/pi-transcript-confirmation.md new file mode 100644 index 0000000000..b20fc0cd4a --- /dev/null +++ b/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. diff --git a/server/core/pending-identity-race.test.ts b/server/core/pending-identity-race.test.ts index f485509d00..6e3fb6fb2e 100644 --- a/server/core/pending-identity-race.test.ts +++ b/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 = { pane_id: "p", terminal_id: "p", agent: "pi", agent_status: "working" }; + const frames: ServerMessage[] = [], writes: string[] = []; + const settled = Promise.withResolvers(); + 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>); + 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 = Promise.resolve(); + const serialize = (_pane: string, task: () => T | Promise): Promise => { 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 }); } + }); diff --git a/server/core/pending.ts b/server/core/pending.ts index 99c60f5c09..25c9666032 100644 --- a/server/core/pending.ts +++ b/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. diff --git a/server/pi-delivery.test.ts b/server/pi-delivery.test.ts index f130f2c834..089441b5ae 100644 --- a/server/pi-delivery.test.ts +++ b/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, diff --git a/server/pi-delivery.ts b/server/pi-delivery.ts index 9e5cc47374..cc1188099a 100644 --- a/server/pi-delivery.ts +++ b/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>): 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