From 373b609ff1abadfeeb57d3ba2edb051d813a9a88 Mon Sep 17 00:00:00 2001 From: Dotta Date: Sat, 8 Aug 2026 02:25:41 +0000 Subject: [PATCH] fix(runner): reject cross-channel result conflicts Co-Authored-By: Paperclip --- .../paperclip-native-runner-spike-spec.md | 4 +- .../codex/codex-app-server-driver.test.ts | 208 ++++++++++++++++++ .../drivers/codex/codex-app-server-driver.ts | 124 ++++++++--- 3 files changed, 308 insertions(+), 28 deletions(-) diff --git a/packages/paperclip-runner/spec/paperclip-native-runner-spike-spec.md b/packages/paperclip-runner/spec/paperclip-native-runner-spike-spec.md index 03aa4a39cf7..1d8d13de146 100644 --- a/packages/paperclip-runner/spec/paperclip-native-runner-spike-spec.md +++ b/packages/paperclip-runner/spec/paperclip-native-runner-spike-spec.md @@ -2257,7 +2257,9 @@ Preferred order: 1. app-server structured output schema for terminal run result, when supported; 2. client-defined `paperclip_finish` and `paperclip_block` dynamic tools; -3. invalid/no result becomes `needs_review`, never inferred `done`. +3. invalid or missing proposals are rejected/preserved as runtime facts and enter + server-owned assessment/recovery without manufacturing a model disposition or + issue status. The task envelope tells the model the expected result shape but does not teach Paperclip API mechanics. diff --git a/packages/paperclip-runner/src/drivers/codex/codex-app-server-driver.test.ts b/packages/paperclip-runner/src/drivers/codex/codex-app-server-driver.test.ts index 2f21707a031..87662f58932 100644 --- a/packages/paperclip-runner/src/drivers/codex/codex-app-server-driver.test.ts +++ b/packages/paperclip-runner/src/drivers/codex/codex-app-server-driver.test.ts @@ -410,6 +410,158 @@ describe("Codex app-server Phase 4 driver", () => { expect(events.filter((event) => event.eventType === "turn.completed")).toHaveLength(1); }); + it("fails closed when an agent message changes a tool-committed result", async () => { + const transport = new FakeCodexTransport(); + const session = await makeDriver([transport]).openSession({ + runId: "run-tool-then-message-conflict", + normalizedSessionId: "normalized-tool-then-message-conflict", + workingDirectory: "/workspace", + }); + await session.startTurn({ message: { role: "user", text: "Complete." } }); + expect(await transport.invoke({ + id: "tool-result", + method: "item/tool/call", + params: { + threadId: "thread-1", + turnId: "turn-1", + callId: "tool-result", + tool: "paperclip_finish", + arguments: result, + }, + })).toMatchObject({ success: true }); + const changed = structuredClone(result); + changed.summary = "Conflicting agent-message result."; + transport.push("item/completed", { + threadId: "thread-1", + turnId: "turn-1", + item: { id: "message-result", type: "agentMessage", text: JSON.stringify(changed) }, + }); + + const events: PrpEvent[] = []; + for await (const event of session.events()) events.push(event); + expect(events.filter((event) => event.eventType === "run.result.proposed")).toHaveLength(1); + expect(events.find((event) => event.eventType === "session.failed")?.payload) + .toMatchObject({ code: "conflicting_semantic_result", recoverable: false }); + expect((await session.snapshot()).semanticResult?.result).toEqual(result); + }); + + it("rejects a tool result that changes an agent-message commitment", async () => { + const transport = new FakeCodexTransport(); + const session = await makeDriver([transport]).openSession({ + runId: "run-message-then-tool-conflict", + normalizedSessionId: "normalized-message-then-tool-conflict", + workingDirectory: "/workspace", + }); + await session.startTurn({ message: { role: "user", text: "Complete." } }); + transport.push("item/completed", { + threadId: "thread-1", + turnId: "turn-1", + item: { id: "message-result", type: "agentMessage", text: JSON.stringify(result) }, + }); + await new Promise((resolve) => setTimeout(resolve, 0)); + const changed = structuredClone(result); + changed.summary = "Conflicting tool result."; + expect(await transport.invoke({ + id: "tool-result", + method: "item/tool/call", + params: { + threadId: "thread-1", + turnId: "turn-1", + callId: "tool-result", + tool: "paperclip_finish", + arguments: changed, + }, + })).toMatchObject({ success: false }); + transport.push("turn/completed", { + threadId: "thread-1", + turn: { id: "turn-1", status: "completed", items: [] }, + }); + + const events = await collectUntilTerminal(session.events()); + expect(events.filter((event) => event.eventType === "run.result.proposed")).toHaveLength(1); + expect(events.some((event) => event.eventType === "session.failed")).toBe(false); + expect((await session.snapshot()).semanticResult).toMatchObject({ + callId: "message-result", + result, + }); + }); + + it("treats a canonically identical cross-channel result as a no-op", async () => { + const transport = new FakeCodexTransport(); + const session = await makeDriver([transport]).openSession({ + runId: "run-identical-cross-channel", + normalizedSessionId: "normalized-identical-cross-channel", + workingDirectory: "/workspace", + }); + await session.startTurn({ message: { role: "user", text: "Complete." } }); + expect(await transport.invoke({ + id: "tool-result", + method: "item/tool/call", + params: { + threadId: "thread-1", + turnId: "turn-1", + callId: "tool-result", + tool: "paperclip_finish", + arguments: result, + }, + })).toMatchObject({ success: true }); + transport.push("item/completed", { + threadId: "thread-1", + turnId: "turn-1", + item: { + id: "message-result", + type: "agentMessage", + text: JSON.stringify(reverseObjectKeys(result)), + }, + }); + transport.push("turn/completed", { + threadId: "thread-1", + turn: { id: "turn-1", status: "completed", items: [] }, + }); + + const events = await collectUntilTerminal(session.events()); + expect(events.filter((event) => event.eventType === "run.result.proposed")).toHaveLength(1); + expect((await session.snapshot()).semanticResult?.callId).toBe("tool-result"); + }); + + it("fails closed when a live terminal embeds a changed semantic result", async () => { + const transport = new FakeCodexTransport(); + const session = await makeDriver([transport]).openSession({ + runId: "run-terminal-result-conflict", + normalizedSessionId: "normalized-terminal-result-conflict", + workingDirectory: "/workspace", + }); + await session.startTurn({ message: { role: "user", text: "Complete." } }); + expect(await transport.invoke({ + id: "tool-result", + method: "item/tool/call", + params: { + threadId: "thread-1", + turnId: "turn-1", + callId: "tool-result", + tool: "paperclip_finish", + arguments: result, + }, + })).toMatchObject({ success: true }); + const changed = structuredClone(result); + changed.summary = "Conflicting terminal result."; + transport.push("turn/completed", { + threadId: "thread-1", + turn: { + id: "turn-1", + status: "completed", + items: [{ id: "terminal-result", type: "agentMessage", text: JSON.stringify(changed) }], + }, + }); + + const events: PrpEvent[] = []; + for await (const event of session.events()) events.push(event); + expect(events.find((event) => event.eventType === "session.failed")?.payload) + .toMatchObject({ code: "conflicting_semantic_result", recoverable: false }); + expect(events.some((event) => event.eventType === "turn.completed")).toBe(false); + expect((await session.snapshot()).semanticResult?.result).toEqual(result); + }); + it.each([ { threadId: "other-thread", turnId: "turn-1", label: "another thread" }, { threadId: "thread-1", turnId: "other-turn", label: "another turn" }, @@ -716,6 +868,62 @@ describe("Codex app-server Phase 4 driver", () => { expect(events.filter((event) => event.eventType === "turn.completed")).toHaveLength(1); }); + it("rejects changed terminal result content discovered during recovery", async () => { + const first = new FakeCodexTransport(); + const second = new FakeCodexTransport(); + const changed = structuredClone(result); + changed.summary = "Conflicting recovered terminal result."; + second.readResponse = { + thread: { + id: "thread-1", + sessionId: "provider-session-1", + cwd: "/workspace", + turns: [{ + id: "turn-1", + status: "completed", + items: [{ id: "terminal-result", type: "agentMessage", text: JSON.stringify(changed) }], + }], + }, + }; + const driver = makeDriver([first, second]); + const original = await driver.openSession({ + runId: "run-recovered-result-conflict", + normalizedSessionId: "normalized-recovered-result-conflict", + workingDirectory: "/workspace", + }); + await original.startTurn({ message: { role: "user", text: "Complete." } }); + expect(await first.invoke({ + id: "tool-result", + method: "item/tool/call", + params: { + threadId: "thread-1", + turnId: "turn-1", + callId: "tool-result", + tool: "paperclip_finish", + arguments: result, + }, + })).toMatchObject({ success: true }); + first.push("turn/completed", { + threadId: "thread-1", + turn: { id: "turn-1", status: "completed", items: [] }, + }); + await collectUntilTerminal(original.events()); + const snapshot = await original.snapshot(); + expect(snapshot.terminalTurns).toHaveLength(1); + await original.close({ reason: "transport lost after terminal" }); + + const recovery = await driver.recoverSession?.(snapshot); + expect(recovery).toMatchObject({ recovered: true }); + await expect(recovery!.session!.reconcile!()) + .rejects.toMatchObject({ + name: "HarnessReconciliationError", + recoverable: true, + message: expect.stringContaining("contains a conflicting semantic result"), + }); + expect((await recovery!.session!.snapshot()).semanticResult?.result).toEqual(result); + await recovery!.session!.close({ reason: "test complete" }); + }); + it("does not re-emit an already observed terminal after recovery", async () => { const first = new FakeCodexTransport(); const second = new FakeCodexTransport(); diff --git a/packages/paperclip-runner/src/drivers/codex/codex-app-server-driver.ts b/packages/paperclip-runner/src/drivers/codex/codex-app-server-driver.ts index d024ec90dd7..1aed3e647f0 100644 --- a/packages/paperclip-runner/src/drivers/codex/codex-app-server-driver.ts +++ b/packages/paperclip-runner/src/drivers/codex/codex-app-server-driver.ts @@ -98,6 +98,13 @@ type CodexCapabilities = Required< NonNullable >; +type SemanticResultAdmission = "committed" | "identical" | "conflict"; + +interface TerminalReplayConflict { + code: "conflicting_semantic_result" | "conflicting_turn_terminal"; + message: string; +} + interface OpenedCodexThread { threadId: string; providerSessionId: string | null; @@ -723,7 +730,7 @@ class CodexHarnessSession implements HarnessSession { this.#activeTurnId = expectedTurnId; } else if (["completed", "failed", "interrupted", "cancelled"].includes(status)) { reconciledTerminalTurnId = expectedTurnId; - this.#mapTerminalTurn(expectedTurn, expectedTurnId); + this.#mapTerminalTurn(expectedTurn, expectedTurnId, true); } else { throw new HarnessReconciliationError( `persisted active turn ${expectedTurnId} has unreconcilable status ${status || "missing"}`, @@ -739,9 +746,10 @@ class CodexHarnessSession implements HarnessSession { `previously terminal turn ${turnId} is no longer terminal in thread/read`, ); } - if (this.#terminalFingerprint(observed) !== fingerprint) { + const conflict = this.#terminalReplayConflict(observed, fingerprint); + if (conflict !== null) { throw new HarnessReconciliationError( - `previously terminal turn ${turnId} changed after recovery`, + `previously terminal turn ${turnId} ${conflict.message}`, ); } } @@ -864,7 +872,7 @@ class CodexHarnessSession implements HarnessSession { } if (notification.method === "item/completed") { if (!this.#notificationNamesActiveTurn(turnId, "item completion")) return; - this.#captureResultFromItem(item, turnId); + if (!this.#captureResultFromItem(item, turnId)) return; this.#emit("item.completed", boundedPayload({ kind: text(item.type, "unknown"), text: itemText(item), @@ -947,13 +955,10 @@ class CodexHarnessSession implements HarnessSession { }], }; } - const fingerprint = canonicalJson(validation.result); - if (this.#resultFingerprint !== null && this.#resultFingerprint !== fingerprint) { + const admission = this.#admitResult(validation.result, callId, turnId); + if (admission === "conflict") { return rejectedToolCall("A different semantic result was already committed."); } - if (this.#result === null) { - this.#commitResult(validation.result, callId, turnId); - } return { success: true, contentItems: [{ type: "inputText", text: "Semantic completion accepted." }] }; } @@ -995,19 +1000,33 @@ class CodexHarnessSession implements HarnessSession { return response; } - #captureResultFromItem(item: Record, turnId: string): void { - if (text(item.type) !== "agentMessage" || this.#result !== null) return; - if (!isRetainableCodexPayload(item.text)) return; + #captureResultFromItem(item: Record, turnId: string): boolean { + if (text(item.type) !== "agentMessage" || !isRetainableCodexPayload(item.text)) return true; const result = tryParseResult(item.text); if (result !== null && isRetainableCodexPayload(result)) { - this.#commitResult(result, text(item.id), turnId); + const admission = this.#admitResult(result, text(item.id), turnId); + if (admission === "conflict") { + this.#failProtocol( + "conflicting_semantic_result", + "Provider agentMessage supplied a different schema-valid semantic result after one was committed.", + ); + return false; + } } + return true; } - #commitResult(result: PrpStructuredRunResult, itemId: string, turnId?: string): void { - if (this.#result !== null) return; + #admitResult( + result: PrpStructuredRunResult, + itemId: string, + turnId?: string, + ): SemanticResultAdmission { + const fingerprint = canonicalJson(result); + if (this.#resultFingerprint !== null) { + return this.#resultFingerprint === fingerprint ? "identical" : "conflict"; + } this.#result = structuredClone(result); - this.#resultFingerprint = canonicalJson(result); + this.#resultFingerprint = fingerprint; this.#resultCallId = itemId || null; this.#resultTurnId = turnId || this.#activeTurnId; result.verification.forEach((verification, index) => { @@ -1024,6 +1043,7 @@ class CodexHarnessSession implements HarnessSession { turnId: turnId || this.#activeTurnId || undefined, itemId: itemId || undefined, }); + return "committed"; } #finalize(turnStatus: string): void { @@ -1037,25 +1057,47 @@ class CodexHarnessSession implements HarnessSession { this.#terminal = true; } - #mapTerminalTurn(turn: Record, fallbackTurnId: string): void { + #mapTerminalTurn( + turn: Record, + fallbackTurnId: string, + reconciling = false, + ): void { const turnId = text(turn.id, fallbackTurnId || this.#activeTurnId || ""); const status = text(turn.status, "completed"); - const candidate = this.#resultFromTurn(turn) ?? this.#result; - const fingerprint = this.#terminalFingerprint(turn, candidate); const previous = this.#terminalTurns.get(turnId); if (previous !== undefined) { - if (previous !== fingerprint) { + const conflict = this.#terminalReplayConflict(turn, previous); + if (conflict !== null) { + if (reconciling) { + throw new HarnessReconciliationError( + `previously terminal turn ${turnId} ${conflict.message}`, + ); + } this.#failProtocol( - "conflicting_turn_terminal", - "Provider changed a terminal fact for an already terminal turn.", + conflict.code, + `Provider terminal for turn ${turnId} ${conflict.message}.`, ); } return; } - this.#terminalTurns.set(turnId, fingerprint); - if (candidate !== null && this.#result === null) { - this.#commitResult(candidate, text(this.#resultItemFromTurn(turn)?.id), turnId); + const candidate = this.#resultFromTurn(turn); + if (candidate !== null) { + const admission = this.#admitResult( + candidate, + text(this.#resultItemFromTurn(turn)?.id), + turnId, + ); + if (admission === "conflict") { + const message = `terminal turn ${turnId} contains a conflicting semantic result`; + if (reconciling) throw new HarnessReconciliationError(message); + this.#failProtocol( + "conflicting_semantic_result", + `Provider ${message}.`, + ); + return; + } } + this.#terminalTurns.set(turnId, this.#terminalFingerprint(turn)); const eventType = status === "failed" ? "turn.failed" @@ -1074,15 +1116,43 @@ class CodexHarnessSession implements HarnessSession { #terminalFingerprint( turn: Record, - candidate: PrpStructuredRunResult | null = this.#resultFromTurn(turn) ?? this.#result, + semanticResult: PrpStructuredRunResult | null = this.#result, ): string { return canonicalJson({ terminalState: terminalState(text(turn.status, "completed")), error: turn.error ?? null, - result: candidate, + result: semanticResult, }); } + #terminalReplayConflict( + turn: Record, + expectedFingerprint: string, + ): TerminalReplayConflict | null { + const candidate = this.#resultFromTurn(turn); + if ( + candidate !== null && + this.#resultFingerprint !== null && + canonicalJson(candidate) !== this.#resultFingerprint + ) { + return { + code: "conflicting_semantic_result", + message: "contains a conflicting semantic result", + }; + } + const semanticResult = this.#result ?? candidate; + if (this.#terminalFingerprint(turn, semanticResult) !== expectedFingerprint) { + return { + code: "conflicting_turn_terminal", + message: "changed from its committed terminal fingerprint", + }; + } + if (candidate !== null && this.#result === null) { + this.#admitResult(candidate, text(this.#resultItemFromTurn(turn)?.id), text(turn.id)); + } + return null; + } + #resultItemFromTurn(turn: Record): Record | null { if (!Array.isArray(turn.items)) return null; return (