fix(runner): reject cross-channel result conflicts

Co-Authored-By: Paperclip <noreply@paperclip.ing>

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.
 

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<HarnessReconciliationError>({
+        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();

packages/paperclip-runner/src/drivers/codex/codex-app-server-driver.ts

@@ -98,6 +98,13 @@ type CodexCapabilities = Required<
   NonNullable<CodexAppServerDriverOptions["capabilities"]>
 >;
 
+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<string, unknown>, turnId: string): void {
-    if (text(item.type) !== "agentMessage" || this.#result !== null) return;
-    if (!isRetainableCodexPayload(item.text)) return;
+  #captureResultFromItem(item: Record<string, unknown>, 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<string, unknown>, fallbackTurnId: string): void {
+  #mapTerminalTurn(
+    turn: Record<string, unknown>,
+    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<string, unknown>,
-    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<string, unknown>,
+    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<string, unknown>): Record<string, unknown> | null {
     if (!Array.isArray(turn.items)) return null;
     return (