diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index 304f312b917..fb76d65e902 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -4197,8 +4197,12 @@ through the ordinary control-plane observers without starting another model turn When a later heartbeat run reuses the provider session, it clears that prior run's wait marker along with its terminal, result and pending-request authority. Already-numbered cancellation acknowledgements and cancelled/expired runtime requests -are retained during shutdown so recovery can replay a contiguous journal. They -carry no permission to start work. Stream closure, stream failure or drain timeout +are retained during shutdown with usage and the terminal. They carry no permission +to start work. The drain discards the turn's other numbered events, so the retained +journal can have sequence gaps after the trigger. Recovery accepts a gap only when +the trigger is the first replayed event and every event after the gap is one of +those retained facts of the same turn. Any run, session or source mismatch still +fails closed. Stream closure, stream failure or drain timeout after complete usage leaves the saved wait unfinalized until execution is proven stopped; complete token counters alone cannot certify that boundary. Codex allows up to 30 seconds for its interruption terminal, matching the transport command diff --git a/packages/paperclip-runner/src/native-session-runtime.test.ts b/packages/paperclip-runner/src/native-session-runtime.test.ts index c8386aaf6d5..f6eca1ad66d 100644 --- a/packages/paperclip-runner/src/native-session-runtime.test.ts +++ b/packages/paperclip-runner/src/native-session-runtime.test.ts @@ -9611,6 +9611,120 @@ describe("executeNativeSession recovery", () => { expect(start).not.toHaveBeenCalled(); expect(complete).not.toHaveBeenCalled(); }); + // Restart chaos (Codex Q1 with PV1/PV2 + K1): the provider held or dropped + // the parked turn's terminal and the controller died during the accounting + // drain. The drain had already discarded numbered non-accounting events, so + // the saved replay has gaps after the trigger. + it.each(["terminal-after-restart", "late-terminal-saved"])("settles a saved governed wait whose drain left sequence gaps before a restart (%s)", async variant => { + const lateTerminalSaved = variant === "late-terminal-saved"; + const turn = (event: PrpEvent) => ({ ...event, turnId: "turn-waiting" }); + const trigger = turn(runnerEvent(1, "item.completed", { kind: "dynamicToolCall", item: { id: "ask-1", name: "request_human_input" } })); + const acknowledgement = turn(runnerEvent(2, "item.completed", { kind: "interrupt_acknowledgement", status: "acknowledged" })); + const discardedMessage = turn(runnerEvent(3, "item.completed", { kind: "agentMessage", text: "I asked the board." })); + const discardedToolEnd = turn(runnerEvent(4, "tool.execution.completed", { name: "request_human_input", status: "completed" })); + const usage = turn(runnerEvent(5, "item.completed", { kind: "usage", usage: { runDelta: { inputTokens: 7, outputTokens: 3 }, runDeltaComplete: true } })); + const stopped = turn(runnerEvent(6, "turn.interrupted", { providerTerminalObserved: true })); + const history: PrpEvent[] = []; + let checkpoint: PersistedNativeSession | undefined; + const start = vi.fn(async () => ({ turnId: trigger.turnId })); + const cancel = vi.fn(() => ({ cleanup: Promise.resolve() })); + const complete = vi.fn(async () => undefined); + const snapshot = (): PersistedNativeSession => ({ backendKind: "mock", sessionId: identity.sessionId, identity, + providerSessionId: "same-provider", activeTurnId: trigger.turnId, cursor: String(history.at(-1)?.sourceSeq ?? 0) }); + const makeSession = (recovering: boolean): NativeSession => ({ identity: () => identity, + async capabilities() { return { resume: true, typedEvents: true, steering: false, interruption: true, structuredResult: true }; }, + startTurn: start, revokeTurnPublication: vi.fn(), cancel, + async snapshot() { return snapshot(); }, async result() { return null; }, async close() {}, + async *events() { + if (!recovering) { + // The terminal is held (or dropped) and the process then dies. + yield trigger; yield acknowledgement; yield discardedMessage; yield discardedToolEnd; yield usage; + return; + } + if (!lateTerminalSaved) yield stopped; + }, + }); + const backend: NativeSessionBackend = { + async descriptor() { return { kind: "mock", name: "drain-gap", version: "1", capabilities: await makeSession(false).capabilities() }; }, + async openSession() { return makeSession(false); }, + async recoverSession() { return { recovered: true, session: makeSession(true) }; }, + }; + const port: ControlPlanePort = { + async openRun() {}, completeRun: complete, + async checkpointSession(value) { checkpoint = structuredClone(value); }, + async appendEvent(event: PrpEvent) { + const duplicate = history.some(row => row.sourceEventId === event.sourceEventId); + if (!duplicate) history.push(event); + return { cursor: history.length, highestContiguousSourceSeq: event.sourceSeq, disposition: duplicate ? "duplicate" : "committed" }; + }, + async replayEvents(request) { + const events = history.filter(row => row.sourceInstanceId === request.sourceInstanceId && row.sourceSeq > request.afterSourceSeq) + .sort((left, right) => left.sourceSeq - right.sourceSeq); + return { events, highestContiguousSourceSeq: events.at(-1)?.sourceSeq ?? request.afterSourceSeq }; + }, + }; + const options = { input, backend, controlPlane: port, runnerInstanceId: "runner-recovery", controlPlaneInstanceId: "control-recovery" }; + await expect(executeNativeSession({ ...options, resolveGovernedWait: ({ event }) => event.sourceSeq === 1 ? yieldedResult : null })) + .rejects.toThrow(); + // The question is saved, and the drain kept only accounting facts. + expect(checkpoint?.governedWait).toEqual({ sourceEvent: trigger, result: yieldedResult }); + expect(history).toEqual([trigger, acknowledgement, usage]); + expect(complete).not.toHaveBeenCalled(); + cancel.mockClear(); + if (lateTerminalSaved) history.push(stopped); + + const answerAlreadySaved = vi.fn(() => null); + await expect(executeNativeSession({ ...options, persistedSession: checkpoint, resolveGovernedWait: answerAlreadySaved })) + .resolves.toMatchObject({ result: yieldedResult, turnId: trigger.turnId, terminal: { runTerminalState: "succeeded" } }); + expect(start).toHaveBeenCalledOnce(); + // A saved terminal settles without interrupting provider work again. + expect(cancel).toHaveBeenCalledTimes(lateTerminalSaved ? 0 : 1); + expect(answerAlreadySaved).not.toHaveBeenCalled(); + expect(complete).toHaveBeenCalledOnce(); + expect(history.filter(event => event.sourceInstanceId === trigger.sourceInstanceId)).toEqual([trigger, acknowledgement, usage, stopped]); + + // A further recovery of the settled wait neither submits nor settles again. + const settledCheckpoint = structuredClone(checkpoint); + await expect(executeNativeSession({ ...options, persistedSession: settledCheckpoint })) + .resolves.toMatchObject({ result: yieldedResult, turnId: trigger.turnId, terminal: { runTerminalState: "succeeded" } }); + expect(start).toHaveBeenCalledOnce(); + expect(cancel).toHaveBeenCalledTimes(lateTerminalSaved ? 0 : 1); + expect(history.filter(event => event.sourceInstanceId === trigger.sourceInstanceId)).toEqual([trigger, acknowledgement, usage, stopped]); + }); + + it.each([ + "gap-before-trigger", "gap-foreign-run", "gap-foreign-session", "gap-foreign-instance", + "gap-other-turn-usage", "gap-unaccounted-event", "unaccounted-event-after-gap", + ])("keeps rejecting an unproven governed-wait replay with a sequence gap: %s", async fault => { + const trigger = { ...runnerEvent(1, "item.completed", { kind: "dynamicToolCall" }), turnId: "turn-waiting" }; + const afterGap = (sourceSeq: number, eventType: PrpEvent["eventType"], payload: Record, overrides: Partial = {}) => + ({ ...runnerEvent(sourceSeq, eventType, payload), turnId: trigger.turnId, ...overrides }) as PrpEvent; + const usagePayload = { kind: "usage", usage: { runDelta: { inputTokens: 1, outputTokens: 1 }, runDeltaComplete: true } }; + const history: PrpEvent[] = fault === "gap-before-trigger" ? [{ ...trigger, sourceSeq: 2, sourceEventId: "late-trigger" }] : [trigger]; + if (fault === "gap-foreign-run") history.push(afterGap(3, "item.completed", usagePayload, { runId: "another-run" })); + if (fault === "gap-foreign-session") history.push(afterGap(3, "item.completed", usagePayload, { normalizedSessionId: "another-session" })); + if (fault === "gap-foreign-instance") history.push(afterGap(3, "item.completed", usagePayload, { sourceInstanceId: "another-runner" })); + if (fault === "gap-other-turn-usage") history.push(afterGap(3, "item.completed", usagePayload, { turnId: "another-turn" })); + if (fault === "gap-unaccounted-event") history.push(afterGap(3, "item.completed", { kind: "agentMessage", text: "more work" })); + if (fault === "unaccounted-event-after-gap") { + history.push(afterGap(3, "item.completed", usagePayload), afterGap(4, "item.delta", { kind: "agentMessage", delta: "more" })); + } + const checkpoint: PersistedNativeSession = { backendKind: "mock", sessionId: identity.sessionId, identity, + activeTurnId: trigger.turnId, governedWait: { sourceEvent: trigger, result: yieldedResult } }; + const start = vi.fn(); + const complete = vi.fn(); + const session: NativeSession = { identity: () => identity, startTurn: start, async result() { return null; }, + async capabilities() { return { resume: true, typedEvents: true, steering: false, interruption: true, structuredResult: true }; }, + async snapshot() { return checkpoint; }, async *events() {}, async close() {} }; + const backend: NativeSessionBackend = { async descriptor() { return { kind: "mock", name: "unsafe-gap", version: "1", capabilities: await session.capabilities() }; }, + async openSession() { return session; }, async recoverSession() { return { recovered: true, session }; } }; + const port: ControlPlanePort = { async openRun() {}, async checkpointSession() {}, async appendEvent() { throw new Error("unexpected append"); }, + completeRun: complete, async replayEvents(request) { return { events: history.filter(row => row.sourceSeq > request.afterSourceSeq), highestContiguousSourceSeq: request.afterSourceSeq }; } }; + await expect(executeNativeSession({ input, backend, controlPlane: port, persistedSession: checkpoint, + runnerInstanceId: "runner-recovery", controlPlaneInstanceId: "control-recovery" })).rejects.toThrow("native_governed_wait_replay_mismatch"); + expect(start).not.toHaveBeenCalled(); expect(complete).not.toHaveBeenCalled(); + }); + it.each([0, 6_000, 12_000])("retains actual shutdown usage after %s ms before finalizing a governed wait", async (receiptDelayMs) => { vi.useFakeTimers(); try { diff --git a/packages/paperclip-runner/src/native-session-runtime.ts b/packages/paperclip-runner/src/native-session-runtime.ts index 433b8fbb8ec..f06e80af74c 100644 --- a/packages/paperclip-runner/src/native-session-runtime.ts +++ b/packages/paperclip-runner/src/native-session-runtime.ts @@ -418,6 +418,22 @@ function isTurnTerminal(event: PrpEvent): boolean { ].includes(event.eventType); } +/** + * The only provider facts a governed-wait accounting drain persists after its + * trigger: usage, the turn terminal and cancellation acknowledgements of the + * parked turn. The drain discards every other numbered event, so the saved + * replay has sequence gaps exactly where those discarded events were. + */ +function isGovernedDrainRetainedEvent(event: PrpEvent, turnId: string | undefined): boolean { + if (event.turnId !== turnId) return false; + return isTurnTerminal(event) + || (event.eventType === "item.completed" + && (event.payload.kind === "usage" || event.payload.kind === "interrupt_acknowledgement")) + || event.eventType === "runtime_request.cancelled" + || event.eventType === "runtime_request.expired" + || (event.eventType === "run.terminal" && ["failed", "cancelled"].includes(String(event.payload.runTerminalState))); +} + function terminalFromEvent( event: PrpEvent, disposition: PrpTerminalState["reportedWorkDisposition"], @@ -1206,14 +1222,10 @@ async function consumeTurn( if (next.done || drainAbort.signal.aborted) return; const trailing = next.value; const terminal = isTurnTerminal(trailing); - const usage = trailing.eventType === "item.completed" && trailing.payload.kind === "usage"; - // These already-numbered cancellation facts carry no new work - // authority. Retain them so later receipts have contiguous replay. - const cancellation = trailing.eventType === "runtime_request.cancelled" - || trailing.eventType === "runtime_request.expired" - || (trailing.eventType === "item.completed" && trailing.payload.kind === "interrupt_acknowledgement") - || (trailing.eventType === "run.terminal" && ["failed", "cancelled"].includes(String(trailing.payload.runTerminalState))); - if (trailing.turnId === event.turnId && (usage || terminal || cancellation)) { + // Usage, the terminal and already-numbered cancellation facts carry + // no new work authority. Everything else is discarded, leaving gaps + // that governed-wait recovery replay accepts only for these facts. + if (isGovernedDrainRetainedEvent(trailing, event.turnId)) { const receipt = await controlPlane.appendEvent(trailing, { signal: drainAbort.signal, ...(terminal && trailing.turnId && trailing.eventType !== "turn.failed" && trailing.payload.providerTerminalObserved !== false @@ -2333,14 +2345,22 @@ export async function executeNativeSession( timeoutMessage: "native governed-wait recovery replay timed out", operation: async signal => { let afterSourceSeq = event.sourceSeq - 1; + let gapObserved = false; while (true) { const page = await options.controlPlane.replayEvents({ runId: identity.runId, sourceInstanceId: options.runnerInstanceId, afterSourceSeq, limit: 1_000 }, { signal }); signal.throwIfAborted(); if (!page.events.length) break; for (const entry of page.events) { + // The trigger must be the first replayed event. After it, the + // accounting drain persisted only facts of the parked turn and + // discarded the rest, so a gap is legitimate only when every later + // event is one of those facts. Identity never relaxes. + if (entry.sourceSeq !== afterSourceSeq + 1) gapObserved = true; if (entry.runId !== identity.runId || entry.normalizedSessionId !== identity.sessionId - || entry.sourceInstanceId !== event.sourceInstanceId || entry.sourceSeq !== afterSourceSeq + 1 + || entry.sourceInstanceId !== event.sourceInstanceId || !Number.isSafeInteger(entry.sourceSeq) + || entry.sourceSeq <= afterSourceSeq || (gapObserved && (governedReplay.length === 0 || !event.turnId + || !isGovernedDrainRetainedEvent(entry, event.turnId))) || governedReplay.length >= 10_000) throw new Error("native_governed_wait_replay_mismatch"); governedReplay.push(entry); afterSourceSeq = entry.sourceSeq;