fix(runner): preserve validated native assistant delta channels

Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
DottaandPaperclip committed 2026-10-09 06:33:59 -05:00
1 parent 7d0a36d393
commit 58cb562866
4 files changed
+38 -3

No files matched your search

@@ -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({
@@ -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<string, unknown>,
): "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";
}
@@ -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", () => {
@@ -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,