From 58cb5628663fe64f63c3bc41629ff1b175f33750 Mon Sep 17 00:00:00 2001 From: Dotta Date: Fri, 9 Oct 2026 06:32:31 -0500 Subject: [PATCH] fix(runner): preserve validated native assistant delta channels Co-Authored-By: Paperclip --- .../codex-app-server-driver.events.test.ts | 24 +++++++++++++++++++ .../codex/codex-session-notifications.ts | 11 ++++++++- .../src/live/runnerd-codex-transport.test.ts | 5 ++-- .../src/live/runnerd-codex-transport.ts | 1 + 4 files changed, 38 insertions(+), 3 deletions(-) diff --git a/packages/paperclip-runner/src/drivers/codex/codex-app-server-driver.events.test.ts b/packages/paperclip-runner/src/drivers/codex/codex-app-server-driver.events.test.ts index fbf45dfea9..bced8b3788 100644 --- a/packages/paperclip-runner/src/drivers/codex/codex-app-server-driver.events.test.ts +++ b/packages/paperclip-runner/src/drivers/codex/codex-app-server-driver.events.test.ts @@ -42,8 +42,32 @@ import { type PrpEvent, type PrpStructuredRunResult, } from "./codex-app-server-driver.test-support.js"; +import { rehydrateRunnerdDeltaNotification } from "../../live/runnerd-codex-transport.js"; describe("Codex app-server Codex driver", () => { + it.each(["progress", "final"] as const)("preserves the trusted canonical %s delta channel without a preceding item start", async channel => { + const transport = new FakeCodexTransport(); + const session = await makeDriver([transport]).openSession({ + runId: "run-canonical-delta-channel", normalizedSessionId: "session-canonical-delta-channel", workingDirectory: WORKSPACE, + }); + const { turnId } = await session.startTurn({ message: { role: "user", text: "Stream native commentary." } }); + transport.push("turn/started", { threadId: "thread-1", turn: { id: turnId, status: "inProgress" } }); + transport.push("item/agentMessage/delta", rehydrateRunnerdDeltaNotification({ + itemId: "native-message-1", kind: "agentMessage", channel, text: "Native message before steering.", + }, "thread-1", turnId)); + transport.push("item/agentMessage/delta", { + threadId: "thread-1", turnId, itemId: "unmarked-message-1", kind: "agentMessage", channel, + delta: "Unmarked provider channel.", canonicalItem: true, + }); + transport.push("turn/completed", { threadId: "thread-1", turn: { id: turnId, status: "interrupted", items: [] } }); + const events = await collectUntilTerminal(session.events()); + expect(events.find(event => event.eventType === "item.delta" && event.itemId === "native-message-1")?.payload) + .toMatchObject({ text: "Native message before steering.", channel }); + expect(events.find(event => event.eventType === "item.delta" && event.itemId === "unmarked-message-1")?.payload) + .toMatchObject({ text: "Unmarked provider channel.", channel: "unknown" }); + await session.close({ reason: "canonical delta channel verified" }); + }); + it("admits a strictly bound semantic result from the durable runner", async () => { const transport = new FakeCodexTransport(); const session = await makeDriver([transport]).openSession({ diff --git a/packages/paperclip-runner/src/drivers/codex/codex-session-notifications.ts b/packages/paperclip-runner/src/drivers/codex/codex-session-notifications.ts index 525feb8dce..7163a40f37 100644 --- a/packages/paperclip-runner/src/drivers/codex/codex-session-notifications.ts +++ b/packages/paperclip-runner/src/drivers/codex/codex-session-notifications.ts @@ -29,6 +29,7 @@ import { } from "./codex-session-workspace.js"; import { boundedText, + RUNNERD_CANONICAL_ITEM, differingJsonPaths, itemFromParams, itemText, @@ -506,7 +507,7 @@ async function mapNotificationBody(state: CodexSessionState, notification: Codex const deltaKind = deltaKinds[notification.method]; if (deltaKind !== undefined) { if (!state.notificationNamesActiveTurn(turnId, "item update")) return; - const methodChannel = channelForDelta(state, notification.method); + const methodChannel = channelForDelta(state, notification.method, params); const channel = methodChannel !== "unknown" ? methodChannel @@ -581,8 +582,16 @@ function channelForStartedItem( function channelForDelta( state: CodexSessionState, method: string, + params: Record, ): "progress" | "final" | "summary" | "detail" | "unknown" { if (method === "item/reasoning/summaryTextDelta") return "summary"; if (method === "item/reasoning/textDelta") return "detail"; + // Native ACPX may emit a delta without an item-start notification. Preserve + // its validated PRP channel through this compatibility facade. The local + // symbol cannot be forged by a provider's JSON message. + if (method === "item/agentMessage/delta" + && Reflect.get(params, RUNNERD_CANONICAL_ITEM) === true + && params.kind === "agentMessage" + && (params.channel === "progress" || params.channel === "final")) return params.channel; return "unknown"; } diff --git a/packages/paperclip-runner/src/live/runnerd-codex-transport.test.ts b/packages/paperclip-runner/src/live/runnerd-codex-transport.test.ts index 612991674a..50a50347ec 100644 --- a/packages/paperclip-runner/src/live/runnerd-codex-transport.test.ts +++ b/packages/paperclip-runner/src/live/runnerd-codex-transport.test.ts @@ -2046,8 +2046,9 @@ it("classifies each coalesced canonical delta without mixing reasoning into assi }); it("restores provider identity and streamed text from a canonical delta", () => { - expect(rehydrateRunnerdDeltaNotification({ text: "Reading Gmail", itemId: "message-1", turnId: "controller-turn" }, "root-thread", "provider-turn")) - .toMatchObject({ threadId: "root-thread", turnId: "provider-turn", delta: "Reading Gmail", itemId: "message-1" }); + const params = rehydrateRunnerdDeltaNotification({ kind: "agentMessage", channel: "progress", text: "Reading Gmail", itemId: "message-1", turnId: "controller-turn" }, "root-thread", "provider-turn"); + expect(params).toMatchObject({ threadId: "root-thread", turnId: "provider-turn", delta: "Reading Gmail", itemId: "message-1", channel: "progress" }); + expect(Reflect.get(params, RUNNERD_CANONICAL_ITEM)).toBe(true); }); it("rehydrates a canonical agent item for the strict Codex facade", () => { diff --git a/packages/paperclip-runner/src/live/runnerd-codex-transport.ts b/packages/paperclip-runner/src/live/runnerd-codex-transport.ts index ba6f3f3248..0dfee2be5d 100644 --- a/packages/paperclip-runner/src/live/runnerd-codex-transport.ts +++ b/packages/paperclip-runner/src/live/runnerd-codex-transport.ts @@ -1873,6 +1873,7 @@ export function rehydrateRunnerdDeltaNotification( // turn is deliberately different from the provider's turn ID. return { ...rawParams, + [RUNNERD_CANONICAL_ITEM]: true, threadId: openedThreadId, turnId: activeTurnId, delta: rawParams.delta ?? rawParams.text,