From 40f65a44576aa8ffdc17be973024dd46c3202e89 Mon Sep 17 00:00:00 2001 From: Peter Fotinis Date: Thu, 27 Aug 2026 16:42:37 +1000 Subject: [PATCH 1/5] Fix Codex subagent relinking after session resume --- packages/host-daemon-contract/src/protocol.ts | 7 +- .../test/contract.test.ts | 2 +- .../test/payload-size.test.ts | 12 +- plugins/provider-codex/src/translator.test.ts | 102 +++++++++++++ plugins/provider-codex/src/translator.ts | 139 +++++++++++++----- 5 files changed, 220 insertions(+), 42 deletions(-) diff --git a/packages/host-daemon-contract/src/protocol.ts b/packages/host-daemon-contract/src/protocol.ts index c398b5d911..ca4f555860 100644 --- a/packages/host-daemon-contract/src/protocol.ts +++ b/packages/host-daemon-contract/src/protocol.ts @@ -335,9 +335,14 @@ // capacity implied by each model id. The usage event meaning changed across // the daemon boundary, so enrolled daemons must update with the server. // +// Version 172 restores Codex native-subagent correlation after a provider +// session resume. A followup to an agent unknown to the fresh bridge now emits +// a nested delegation and parents the resumed child turn instead of projecting +// it as competing root work; send_message remains non-turn-producing. +// // The version mismatch is what triggers the enrolled daemon's automatic update // instead of an `invalid-message` reconnect loop. -export const HOST_DAEMON_PROTOCOL_VERSION = 171 as const; +export const HOST_DAEMON_PROTOCOL_VERSION = 172 as const; /** * Absolute ceiling for any executable artifact delivered to a host daemon — diff --git a/packages/host-daemon-contract/test/contract.test.ts b/packages/host-daemon-contract/test/contract.test.ts index 2b3852f9c7..a8df5dea0d 100644 --- a/packages/host-daemon-contract/test/contract.test.ts +++ b/packages/host-daemon-contract/test/contract.test.ts @@ -1046,7 +1046,7 @@ describe("host-daemon command schemas", () => { // mixed version. Version 113 carried the Devin Desktop open target rename // and remains part of the protocol lineage. it("uses the current host-daemon protocol version", () => { - expect(HOST_DAEMON_PROTOCOL_VERSION).toBe(171); + expect(HOST_DAEMON_PROTOCOL_VERSION).toBe(172); expect(HOST_ARTIFACT_MAX_BYTES).toBe(256 * 1024 * 1024); }); diff --git a/packages/host-daemon-contract/test/payload-size.test.ts b/packages/host-daemon-contract/test/payload-size.test.ts index d8e1fdc83b..50e4807ada 100644 --- a/packages/host-daemon-contract/test/payload-size.test.ts +++ b/packages/host-daemon-contract/test/payload-size.test.ts @@ -76,18 +76,18 @@ describe("daemon-to-server event payload sizes", () => { expect(measurements).toEqual([ { eventCount: 1, - legacyEnvelope: { gzipBytes: 194, jsonBytes: 413 }, - grouped: { gzipBytes: 198, jsonBytes: 421 }, + legacyEnvelope: { gzipBytes: 192, jsonBytes: 413 }, + grouped: { gzipBytes: 196, jsonBytes: 421 }, }, { eventCount: 10, - legacyEnvelope: { gzipBytes: 246, jsonBytes: 3_554 }, - grouped: { gzipBytes: 247, jsonBytes: 3_049 }, + legacyEnvelope: { gzipBytes: 243, jsonBytes: 3_554 }, + grouped: { gzipBytes: 243, jsonBytes: 3_049 }, }, { eventCount: 50, - legacyEnvelope: { gzipBytes: 406, jsonBytes: 17_554 }, - grouped: { gzipBytes: 407, jsonBytes: 14_769 }, + legacyEnvelope: { gzipBytes: 405, jsonBytes: 17_554 }, + grouped: { gzipBytes: 406, jsonBytes: 14_769 }, }, ]); diff --git a/plugins/provider-codex/src/translator.test.ts b/plugins/provider-codex/src/translator.test.ts index 787ac69912..94e0da04a0 100644 --- a/plugins/provider-codex/src/translator.test.ts +++ b/plugins/provider-codex/src/translator.test.ts @@ -728,6 +728,22 @@ describe("codex command output capture across reordering", () => { describe("codex subagent activity correlation", () => { const rootProviderThreadId = "root-provider-thread"; + function rawCollaborationCall(args: { + callId: string; + name: "followup_task" | "send_message"; + }) { + return codexEvent("rawResponseItem/completed", { + threadId: rootProviderThreadId, + turnId: "parent-turn", + item: { + type: "function_call", + name: args.name, + arguments: '{"target":"/root/lifecycle_child"}', + call_id: args.callId, + }, + }); + } + function subAgentActivity(args: { agentThreadId?: string; id: string; @@ -947,6 +963,92 @@ describe("codex subagent activity correlation", () => { ]); }); + // A provider-session reap discards the translator's in-memory spawn map, + // but Codex can resume an agent from the durable rollout. The raw collab + // call is the surviving authority that distinguishes a turn-producing + // followup from a message that must not reserve the next native turn. + it("links an unknown resumed subagent from the raw followup intent after translator restart", () => { + const harness = createHarness(); + + expect( + harness.translate( + rawCollaborationCall({ + callId: "message-call", + name: "send_message", + }), + ), + ).toEqual([]); + expect( + harness.translate( + subAgentActivity({ id: "message-call", kind: "interacted" }), + ), + ).toEqual([]); + + expect( + harness.translate( + rawCollaborationCall({ + callId: "followup-call", + name: "followup_task", + }), + ), + ).toEqual([]); + expect( + harness.translate( + subAgentActivity({ id: "followup-call", kind: "interacted" }), + ), + ).toEqual([ + expect.objectContaining({ + type: "item/started", + scope: turnScope(harness.turnId("parent-turn")), + item: expect.objectContaining({ + type: "delegation", + id: harness.itemId("followup-call"), + childRef: "agent-thread-1", + status: "pending", + }), + }), + ]); + + expect( + harness.translate(childTurnStarted("resumed-child-turn")), + ).toContainEqual( + expect.objectContaining({ + type: "turn/started", + scope: turnScope(harness.turnId("resumed-child-turn")), + parentToolCallId: harness.itemId("followup-call"), + }), + ); + }); + + it("does not reopen a known terminal subagent for send_message", () => { + const harness = createHarness(); + harness.translate( + subAgentActivity({ id: "subagent-call-1", kind: "started" }), + ); + harness.translate(childTurnStarted("child-turn-1")); + harness.translate(childTurnCompleted("child-turn-1")); + + harness.translate( + rawCollaborationCall({ callId: "message-call", name: "send_message" }), + ); + expect( + harness.translate( + subAgentActivity({ id: "message-call", kind: "interacted" }), + ), + ).toEqual([]); + + expect( + harness.translator.prepareTurnStart({ + clientRequestId: "creq_after_message", + providerThreadId: rootProviderThreadId, + }), + ).not.toBeNull(); + const nextRootTurn = harness + .translate(childTurnStarted("next-root-turn")) + .find((event) => event.type === "turn/started"); + expect(nextRootTurn).not.toHaveProperty("parentToolCallId"); + }); + // Follow-ups queue: two interactions owe two more child turns. The re-arm is // counted, so terminalizing the agent after the first follow-up must not // discard the link the second one still needs. diff --git a/plugins/provider-codex/src/translator.ts b/plugins/provider-codex/src/translator.ts index ed8af4cbec..0d5efcbcf6 100644 --- a/plugins/provider-codex/src/translator.ts +++ b/plugins/provider-codex/src/translator.ts @@ -115,6 +115,8 @@ interface CodexPendingDelegationTurnLink { parentTurnId: string; } +type CodexInteractionKind = "followup" | "message"; + /** The collab arguments a receiver-less spawn/resume tool call carries. */ const codexDelegationArgsSchema = z .object({ @@ -226,7 +228,10 @@ function withDeltaParentRef( const codexProviderThreadIdParamsSchema = z .object({ threadId: z.string().min(1).optional(), - thread: z.object({ id: z.string().min(1) }).passthrough().optional(), + thread: z + .object({ id: z.string().min(1) }) + .passthrough() + .optional(), }) .passthrough(); @@ -437,6 +442,10 @@ export function createCodexEventTranslator( const pendingDelegationCallIds = new Set(); const pendingDelegationProviderThreadIdByCallId = new Map(); const processedSubAgentInteractionIds = new Set(); + const interactionKindsByProviderThreadId = new Map< + string, + Map + >(); const trackedSubAgentsByCallId = new Map(); const trackedSubAgentCallIdsByAgentThreadId = new Map(); @@ -599,6 +608,7 @@ export function createCodexEventTranslator( function clearCodexDelegationParentState( providerThreadId: string, ): ThreadDelta[] { + interactionKindsByProviderThreadId.delete(providerThreadId); delegationParentToolCallIdsByProviderThreadId.delete(providerThreadId); pendingDelegationTurnLinksByProviderThreadId.delete(providerThreadId); const closes: ThreadDelta[] = []; @@ -610,7 +620,9 @@ export function createCodexEventTranslator( continue; } if (isTrackedSubAgentOpen(tracked)) { - closes.push(buildCodexSubAgentCloseDelta({ status: "failed", tracked })); + closes.push( + buildCodexSubAgentCloseDelta({ status: "failed", tracked }), + ); } tracked.terminal = true; tracked.pendingFollowups = 0; @@ -1001,6 +1013,60 @@ export function createCodexEventTranslator( return trackedSubAgentsByCallId.get(callId); } + function beginCodexTrackedSubAgent( + activity: CodexSubAgentActivityEvent, + ): ThreadDelta[] { + const tracked: CodexTrackedSubAgent = { + agentPath: activity.item.agentPath, + agentThreadId: activity.item.agentThreadId, + callId: activity.item.id, + parentProviderThreadId: activity.providerThreadId, + parentTurnId: activity.turnId, + pendingFollowups: 0, + terminal: false, + }; + trackedSubAgentsByCallId.set(tracked.callId, tracked); + trackedSubAgentCallIdsByAgentThreadId.set( + tracked.agentThreadId, + tracked.callId, + ); + + const [openDelta] = attachCodexDelegationParentLinks( + [buildCodexSubAgentOpenDelta(tracked)], + activity.providerThreadId, + ); + if (openDelta?.kind === "item.open") { + tracked.parentToolCallId = openDelta.key.parentRef; + } + // Codex currently multiplexes child turns onto the root provider thread, + // even though the activity includes a distinct agent thread id. Queue a + // FIFO fallback in addition to the explicit id mapping. + enqueuePendingDelegationTurnLink({ + callId: tracked.callId, + parentTurnId: tracked.parentTurnId, + providerThreadId: tracked.parentProviderThreadId, + }); + return openDelta ? [openDelta] : []; + } + + function consumeCodexInteractionKind(args: { + callId: string; + providerThreadId: string; + }): CodexInteractionKind | undefined { + const interactionKinds = interactionKindsByProviderThreadId.get( + args.providerThreadId, + ); + const kind = interactionKinds?.get(args.callId); + if (!kind || !interactionKinds) { + return undefined; + } + interactionKinds.delete(args.callId); + if (interactionKinds.size === 0) { + interactionKindsByProviderThreadId.delete(args.providerThreadId); + } + return kind; + } + function rearmTrackedSubAgent(tracked: CodexTrackedSubAgent): void { trackedSubAgentCallIdsByAgentThreadId.set( tracked.agentThreadId, @@ -1063,37 +1129,7 @@ export function createCodexEventTranslator( if (trackedSubAgentsByCallId.has(activity.item.id)) { return []; } - const tracked: CodexTrackedSubAgent = { - agentPath: activity.item.agentPath, - agentThreadId: activity.item.agentThreadId, - callId: activity.item.id, - parentProviderThreadId: activity.providerThreadId, - parentTurnId: activity.turnId, - pendingFollowups: 0, - terminal: false, - }; - trackedSubAgentsByCallId.set(tracked.callId, tracked); - trackedSubAgentCallIdsByAgentThreadId.set( - tracked.agentThreadId, - tracked.callId, - ); - - const [openDelta] = attachCodexDelegationParentLinks( - [buildCodexSubAgentOpenDelta(tracked)], - activity.providerThreadId, - ); - if (openDelta?.kind === "item.open") { - tracked.parentToolCallId = openDelta.key.parentRef; - } - // Codex currently multiplexes child turns onto the root provider - // thread, even though the activity includes a distinct agent thread - // id. Queue a FIFO fallback in addition to the explicit id mapping. - enqueuePendingDelegationTurnLink({ - callId: tracked.callId, - parentTurnId: tracked.parentTurnId, - providerThreadId: tracked.parentProviderThreadId, - }); - return openDelta ? [openDelta] : []; + return beginCodexTrackedSubAgent(activity); } case "interacted": { // Messaging an existing agent is activity within the original @@ -1104,6 +1140,13 @@ export function createCodexEventTranslator( return []; } processedSubAgentInteractionIds.add(activity.item.id); + const interactionKind = consumeCodexInteractionKind({ + callId: activity.item.id, + providerThreadId: activity.providerThreadId, + }); + if (interactionKind === "message") { + return []; + } const tracked = findTrackedSubAgentByAgentThreadId( activity.item.agentThreadId, ); @@ -1117,6 +1160,12 @@ export function createCodexEventTranslator( return [buildCodexSubAgentOpenDelta(tracked)]; } } + // A resumed provider session has no in-memory spawn record for an + // older agent. The raw call id still proves this interaction starts a + // turn, so materialize a fresh delegation for the resumed work. + if (!tracked && interactionKind === "followup") { + return beginCodexTrackedSubAgent(activity); + } return []; } case "interrupted": { @@ -1144,7 +1193,10 @@ export function createCodexEventTranslator( const completedDeltas: ThreadDelta[] = []; for (const delta of deltas) { completedDeltas.push(delta); - if (delta.kind !== "turn.boundary" || delta.providerTurnId === undefined) { + if ( + delta.kind !== "turn.boundary" || + delta.providerTurnId === undefined + ) { continue; } const callId = delegationParentToolCallIdsByTurnId.get( @@ -1183,6 +1235,25 @@ export function createCodexEventTranslator( const { threadId: providerThreadId, item } = paramsResult.data; if (item.type === "function_call") { + // subAgentActivity collapses both verbs to `interacted`; retain the raw + // intent so send_message cannot reserve a turn that only followup_task + // will start. + if (item.name === "followup_task" || item.name === "send_message") { + if (!processedSubAgentInteractionIds.has(item.call_id)) { + const interactionKinds = + interactionKindsByProviderThreadId.get(providerThreadId) ?? + new Map(); + interactionKinds.set( + item.call_id, + item.name === "followup_task" ? "followup" : "message", + ); + interactionKindsByProviderThreadId.set( + providerThreadId, + interactionKinds, + ); + } + return []; + } if (!CODEX_SHELL_TOOL_NAMES.has(item.name)) { return []; } From 149c579c13b07e97186e158d1465e5f691ff2a1e Mon Sep 17 00:00:00 2001 From: Peter Fotinis Date: Thu, 27 Aug 2026 22:57:31 +1000 Subject: [PATCH 2/5] Handle rawless Codex subagent resumes --- plugins/provider-codex/src/translator.test.ts | 91 ++++++-- plugins/provider-codex/src/translator.ts | 206 ++++++++++++++++-- 2 files changed, 261 insertions(+), 36 deletions(-) diff --git a/plugins/provider-codex/src/translator.test.ts b/plugins/provider-codex/src/translator.test.ts index 94e0da04a0..6f53a8b89a 100644 --- a/plugins/provider-codex/src/translator.test.ts +++ b/plugins/provider-codex/src/translator.test.ts @@ -916,13 +916,15 @@ describe("codex subagent activity correlation", () => { ); harness.translate(childTurnCompleted("child-turn-1")); - // A follow-up to a settled agent re-opens its delegation row (same item - // id): the agent works again, and an open delegation is open work. + // `interacted` alone is ambiguous: Codex uses it for both followup_task + // and send_message. Wait for the child turn before reopening the row. expect( harness.translate( subAgentActivity({ id: "interaction-1", kind: "interacted" }), ), - ).toEqual([ + ).toEqual([]); + + expect(harness.translate(childTurnStarted("child-turn-2"))).toEqual([ expect.objectContaining({ type: "item/started", scope: turnScope(harness.turnId("parent-turn")), @@ -932,15 +934,12 @@ describe("codex subagent activity correlation", () => { status: "pending", }), }), - ]); - - expect(harness.translate(childTurnStarted("child-turn-2"))).toContainEqual( expect.objectContaining({ type: "turn/started", scope: turnScope(harness.turnId("child-turn-2")), parentToolCallId: harness.itemId("subagent-call-1"), }), - ); + ]); // The resumed turn settles the re-opened delegation again. const resumedTurnCompleted = harness.translate( @@ -1049,9 +1048,60 @@ describe("codex subagent activity correlation", () => { expect(nextRootTurn).not.toHaveProperty("parentToolCallId"); }); - // Follow-ups queue: two interactions owe two more child turns. The re-arm is - // counted, so terminalizing the agent after the first follow-up must not - // discard the link the second one still needs. + it("links a rawless resumed subagent when its child turn starts", () => { + const harness = createHarness(); + + expect( + harness.translate( + subAgentActivity({ id: "rawless-followup", kind: "interacted" }), + ), + ).toEqual([]); + + const resumedEvents = harness.translate( + childTurnStarted("rawless-child-turn"), + ); + expect(resumedEvents).toEqual([ + expect.objectContaining({ + type: "item/started", + scope: turnScope(harness.turnId("parent-turn")), + item: expect.objectContaining({ + type: "delegation", + id: harness.itemId("rawless-followup"), + childRef: "agent-thread-1", + }), + }), + expect.objectContaining({ + type: "turn/started", + scope: turnScope(harness.turnId("rawless-child-turn")), + parentToolCallId: harness.itemId("rawless-followup"), + }), + ]); + expect(resumedEvents[0]).not.toHaveProperty("parentToolCallId"); + expect(resumedEvents[0]).not.toHaveProperty("item.parentToolCallId"); + }); + + it("discards a rawless message interaction at its parent boundary", () => { + const harness = createHarness(); + expect( + harness.translate( + subAgentActivity({ id: "rawless-message", kind: "interacted" }), + ), + ).toEqual([]); + harness.translate(childTurnCompleted("parent-turn")); + + harness.translator.prepareTurnStart({ + clientRequestId: "creq_after_rawless_message", + providerThreadId: rootProviderThreadId, + }); + const nextRootTurn = harness + .translate(childTurnStarted("next-root-after-message")) + .find((event) => event.type === "turn/started"); + expect(nextRootTurn).not.toHaveProperty("parentToolCallId"); + }); + + // Each child turn consumes one ambiguous interaction. Settling the first + // resumed turn must not discard the second interaction that still awaits + // its own turn-producing proof. it("preserves the parent link across queued follow-up resumes", () => { const harness = createHarness(); harness.translate( @@ -1060,14 +1110,15 @@ describe("codex subagent activity correlation", () => { harness.translate(childTurnStarted("child-turn-1")); harness.translate(childTurnCompleted("child-turn-1")); - // The first follow-up re-opens the delegation; the second finds it open. + // Neither ambiguous interaction re-opens the delegation until a child + // turn proves that it was a turn-producing follow-up. expect( harness .translate( subAgentActivity({ id: "interaction-1", kind: "interacted" }), ) .map((event) => event.type), - ).toEqual(["item/started"]); + ).toEqual([]); expect( harness.translate( subAgentActivity({ id: "interaction-2", kind: "interacted" }), @@ -1077,21 +1128,25 @@ describe("codex subagent activity correlation", () => { for (const index of [2, 3]) { expect( harness.translate(childTurnStarted(`child-turn-${index}`)), - ).toContainEqual( + ).toEqual([ + expect.objectContaining({ + type: "item/started", + item: expect.objectContaining({ + type: "delegation", + id: harness.itemId("subagent-call-1"), + }), + }), expect.objectContaining({ type: "turn/started", scope: turnScope(harness.turnId(`child-turn-${index}`)), parentToolCallId: harness.itemId("subagent-call-1"), }), - ); - // The delegation closes only once the last owed follow-up turn settles. + ]); expect( harness .translate(childTurnCompleted(`child-turn-${index}`)) .map((event) => event.type), - ).toEqual( - index === 3 ? ["turn/completed", "item/completed"] : ["turn/completed"], - ); + ).toEqual(["turn/completed", "item/completed"]); } }); diff --git a/plugins/provider-codex/src/translator.ts b/plugins/provider-codex/src/translator.ts index 0d5efcbcf6..90408aa072 100644 --- a/plugins/provider-codex/src/translator.ts +++ b/plugins/provider-codex/src/translator.ts @@ -446,6 +446,10 @@ export function createCodexEventTranslator( string, Map >(); + const unclassifiedInteractionsByProviderThreadId = new Map< + string, + CodexSubAgentActivityEvent[] + >(); const trackedSubAgentsByCallId = new Map(); const trackedSubAgentCallIdsByAgentThreadId = new Map(); @@ -609,6 +613,7 @@ export function createCodexEventTranslator( providerThreadId: string, ): ThreadDelta[] { interactionKindsByProviderThreadId.delete(providerThreadId); + unclassifiedInteractionsByProviderThreadId.delete(providerThreadId); delegationParentToolCallIdsByProviderThreadId.delete(providerThreadId); pendingDelegationTurnLinksByProviderThreadId.delete(providerThreadId); const closes: ThreadDelta[] = []; @@ -1085,6 +1090,171 @@ export function createCodexEventTranslator( }); } + function queueUnclassifiedCodexInteraction( + activity: CodexSubAgentActivityEvent, + ): void { + const pending = + unclassifiedInteractionsByProviderThreadId.get( + activity.providerThreadId, + ) ?? []; + pending.push(activity); + unclassifiedInteractionsByProviderThreadId.set( + activity.providerThreadId, + pending, + ); + } + + function takeUnclassifiedCodexInteraction(args: { + callId?: string; + providerThreadId: string; + startedTurnId?: string; + }): CodexSubAgentActivityEvent | undefined { + let interactionProviderThreadId = args.providerThreadId; + let pending = unclassifiedInteractionsByProviderThreadId.get( + interactionProviderThreadId, + ); + if (!pending && args.callId === undefined) { + for (const [ + candidateProviderThreadId, + candidates, + ] of unclassifiedInteractionsByProviderThreadId) { + if ( + candidates.some( + (activity) => activity.item.agentThreadId === args.providerThreadId, + ) + ) { + interactionProviderThreadId = candidateProviderThreadId; + pending = candidates; + break; + } + } + } + if (!pending) { + return undefined; + } + let index = + args.callId !== undefined + ? pending.findIndex((activity) => activity.item.id === args.callId) + : -1; + if (args.callId === undefined) { + for ( + let candidateIndex = pending.length - 1; + candidateIndex >= 0; + --candidateIndex + ) { + const activity = pending[candidateIndex]; + if ( + activity && + activity.turnId !== args.startedTurnId && + (interactionProviderThreadId === args.providerThreadId || + activity.item.agentThreadId === args.providerThreadId) + ) { + index = candidateIndex; + break; + } + } + } + if (index === -1) { + return undefined; + } + const [activity] = pending.splice(index, 1); + if (pending.length === 0) { + unclassifiedInteractionsByProviderThreadId.delete( + interactionProviderThreadId, + ); + } + return activity; + } + + function materializeCodexFollowup( + activity: CodexSubAgentActivityEvent, + ): ThreadDelta[] { + const tracked = findTrackedSubAgentByAgentThreadId( + activity.item.agentThreadId, + ); + if (!tracked) { + return beginCodexTrackedSubAgent(activity); + } + if (!tracked.terminal) { + return []; + } + const wasOpen = isTrackedSubAgentOpen(tracked); + tracked.pendingFollowups += 1; + rearmTrackedSubAgent(tracked); + return wasOpen ? [] : [buildCodexSubAgentOpenDelta(tracked)]; + } + + function hasConsumablePendingDelegationLink(args: { + providerThreadId: string; + startedTurnId: string; + }): boolean { + return ( + pendingDelegationTurnLinksByProviderThreadId + .get(args.providerThreadId) + ?.some((link) => link.parentTurnId !== args.startedTurnId) ?? false + ); + } + + /** + * Codex app-server does not currently forward the raw collaboration call + * for resumed agents. It does forward `interacted`, followed by the child + * `turn/started` before the invoking parent turn completes. Hold an + * unclassified interaction until that observable turn proves it was a + * followup; a message-only interaction is discarded at its parent boundary. + */ + function materializeUnclassifiedCodexInteractions( + deltas: ThreadDelta[], + providerThreadId: string | undefined, + ): ThreadDelta[] { + if (!providerThreadId) { + return deltas; + } + const materialized: ThreadDelta[] = []; + for (const delta of deltas) { + if ( + delta.kind === "turn.open" && + delta.providerTurnId !== undefined && + !hasPendingNativeTurnStart(providerThreadId) && + !hasConsumablePendingDelegationLink({ + providerThreadId, + startedTurnId: delta.providerTurnId, + }) + ) { + const activity = takeUnclassifiedCodexInteraction({ + providerThreadId, + startedTurnId: delta.providerTurnId, + }); + if (activity) { + materialized.push(...materializeCodexFollowup(activity)); + } + } + materialized.push( + ...attachCodexDelegationParentLinks([delta], providerThreadId), + ); + if ( + delta.kind === "turn.boundary" && + delta.providerTurnId !== undefined + ) { + const pending = + unclassifiedInteractionsByProviderThreadId.get(providerThreadId); + if (pending) { + const remaining = pending.filter( + (activity) => activity.turnId !== delta.providerTurnId, + ); + if (remaining.length === 0) { + unclassifiedInteractionsByProviderThreadId.delete(providerThreadId); + } else if (remaining.length !== pending.length) { + unclassifiedInteractionsByProviderThreadId.set( + providerThreadId, + remaining, + ); + } + } + } + } + return materialized; + } + /** * A tracked sub-agent is open work while it has not reached a terminal * turn, or while it still owes a followup turn it was re-armed for. The @@ -1133,9 +1303,9 @@ export function createCodexEventTranslator( } case "interacted": { // Messaging an existing agent is activity within the original - // delegation, not a new timeline row. A completed agent can receive - // followup_task; re-arm the original parent so the next child turn - // is not projected as a root turn. + // delegation, not a new timeline row. Because app-server may omit the + // raw verb, a completed or historical agent waits for a child turn to + // prove this interaction was a followup before it is re-armed. if (processedSubAgentInteractionIds.has(activity.item.id)) { return []; } @@ -1150,22 +1320,13 @@ export function createCodexEventTranslator( const tracked = findTrackedSubAgentByAgentThreadId( activity.item.agentThreadId, ); - if (tracked?.terminal) { - const wasOpen = isTrackedSubAgentOpen(tracked); - tracked.pendingFollowups += 1; - rearmTrackedSubAgent(tracked); - if (!wasOpen) { - // The agent works again: re-open its delegation row (the - // assembler reuses the minted item id for a known provider id). - return [buildCodexSubAgentOpenDelta(tracked)]; - } + if (tracked && !tracked.terminal) { + return []; } - // A resumed provider session has no in-memory spawn record for an - // older agent. The raw call id still proves this interaction starts a - // turn, so materialize a fresh delegation for the resumed work. - if (!tracked && interactionKind === "followup") { - return beginCodexTrackedSubAgent(activity); + if (interactionKind === "followup") { + return materializeCodexFollowup(activity); } + queueUnclassifiedCodexInteraction(activity); return []; } case "interrupted": { @@ -1239,6 +1400,15 @@ export function createCodexEventTranslator( // intent so send_message cannot reserve a turn that only followup_task // will start. if (item.name === "followup_task" || item.name === "send_message") { + const pendingActivity = takeUnclassifiedCodexInteraction({ + callId: item.call_id, + providerThreadId, + }); + if (pendingActivity) { + return item.name === "followup_task" + ? materializeCodexFollowup(pendingActivity) + : []; + } if (!processedSubAgentInteractionIds.has(item.call_id)) { const interactionKinds = interactionKindsByProviderThreadId.get(providerThreadId) ?? @@ -1467,7 +1637,7 @@ export function createCodexEventTranslator( ); } - const parentLinkedDeltas = attachCodexDelegationParentLinks( + const parentLinkedDeltas = materializeUnclassifiedCodexInteractions( translateCodexEventToDeltas(event, eventTranslationState), providerThreadId, ); From 222882c6c080304592529cb2144d4e38c9cefe31 Mon Sep 17 00:00:00 2001 From: Peter Fotinis Date: Thu, 27 Aug 2026 23:42:46 +1000 Subject: [PATCH 3/5] Test resumed Codex child thread isolation --- plugins/provider-codex/src/translator.test.ts | 75 ++++++++++++++++++- plugins/provider-codex/src/translator.ts | 6 +- 2 files changed, 74 insertions(+), 7 deletions(-) diff --git a/plugins/provider-codex/src/translator.test.ts b/plugins/provider-codex/src/translator.test.ts index 6f53a8b89a..f28843f7ce 100644 --- a/plugins/provider-codex/src/translator.test.ts +++ b/plugins/provider-codex/src/translator.test.ts @@ -962,10 +962,10 @@ describe("codex subagent activity correlation", () => { ]); }); - // A provider-session reap discards the translator's in-memory spawn map, - // but Codex can resume an agent from the durable rollout. The raw collab - // call is the surviving authority that distinguishes a turn-producing - // followup from a message that must not reserve the next native turn. + // When app-server supplies the raw collaboration call, it distinguishes a + // turn-producing followup from a message that must not reserve the next + // native turn. Resumed sessions can omit this notification; the rawless + // cases below cover that event shape. it("links an unknown resumed subagent from the raw followup intent after translator restart", () => { const harness = createHarness(); @@ -1080,6 +1080,73 @@ describe("codex subagent activity correlation", () => { expect(resumedEvents[0]).not.toHaveProperty("item.parentToolCallId"); }); + // Production resumes report the interaction on the root thread and the + // resulting child turn on the agent's own provider thread. Once the parent + // settles, a new root prompt can start while that child is still running; + // the two provider-thread-scoped correlations must remain independent. + it("keeps root input correlation independent from a rawless resumed child thread", () => { + const harness = createHarness(); + + expect( + harness.translate( + subAgentActivity({ id: "rawless-followup", kind: "interacted" }), + ), + ).toEqual([]); + + expect( + harness.translate( + childTurnStarted("rawless-child-turn", "agent-thread-1"), + ), + ).toEqual([ + expect.objectContaining({ + type: "item/started", + scope: turnScope(harness.turnId("parent-turn")), + item: expect.objectContaining({ + type: "delegation", + id: harness.itemId("rawless-followup"), + childRef: "agent-thread-1", + }), + }), + expect.objectContaining({ + type: "turn/started", + scope: turnScope(harness.turnId("rawless-child-turn")), + parentToolCallId: harness.itemId("rawless-followup"), + }), + ]); + + harness.translate(childTurnCompleted("parent-turn")); + expect( + harness.translator.prepareTurnStart({ + clientRequestId: "creq_while_child_running", + providerThreadId: rootProviderThreadId, + }), + ).not.toBeNull(); + + const rootEvents = harness.translate(childTurnStarted("next-root-turn")); + expect(rootEvents).toContainEqual( + expect.objectContaining({ + type: "turn/started", + scope: turnScope(harness.turnId("next-root-turn")), + }), + ); + expect(rootEvents).toContainEqual( + expect.objectContaining({ + type: "turn/input/accepted", + scope: turnScope(harness.turnId("next-root-turn")), + clientRequestId: "creq_while_child_running", + }), + ); + expect( + rootEvents.find((event) => event.type === "turn/started"), + ).not.toHaveProperty("parentToolCallId"); + + expect( + harness + .translate(childTurnCompleted("rawless-child-turn", "agent-thread-1")) + .map((event) => event.type), + ).toEqual(["turn/completed", "item/completed"]); + }); + it("discards a rawless message interaction at its parent boundary", () => { const harness = createHarness(); expect( diff --git a/plugins/provider-codex/src/translator.ts b/plugins/provider-codex/src/translator.ts index 90408aa072..ea0a0824b1 100644 --- a/plugins/provider-codex/src/translator.ts +++ b/plugins/provider-codex/src/translator.ts @@ -1043,9 +1043,9 @@ export function createCodexEventTranslator( if (openDelta?.kind === "item.open") { tracked.parentToolCallId = openDelta.key.parentRef; } - // Codex currently multiplexes child turns onto the root provider thread, - // even though the activity includes a distinct agent thread id. Queue a - // FIFO fallback in addition to the explicit id mapping. + // Codex can report child turns on the agent's own provider thread. Retain + // a FIFO fallback as well for older/multiplexed event shapes that report + // them on the root provider thread. enqueuePendingDelegationTurnLink({ callId: tracked.callId, parentTurnId: tracked.parentTurnId, From 37397cc67735dd241ee7f8611b8912e40b889df7 Mon Sep 17 00:00:00 2001 From: Peter Fotinis Date: Fri, 28 Aug 2026 09:22:11 +1000 Subject: [PATCH 4/5] Cover rawless Codex message isolation --- .../test/payload-size.test.ts | 12 ++--- plugins/provider-codex/src/translator.test.ts | 51 +++++++++++++++++++ 2 files changed, 57 insertions(+), 6 deletions(-) diff --git a/packages/host-daemon-contract/test/payload-size.test.ts b/packages/host-daemon-contract/test/payload-size.test.ts index 50e4807ada..d8e1fdc83b 100644 --- a/packages/host-daemon-contract/test/payload-size.test.ts +++ b/packages/host-daemon-contract/test/payload-size.test.ts @@ -76,18 +76,18 @@ describe("daemon-to-server event payload sizes", () => { expect(measurements).toEqual([ { eventCount: 1, - legacyEnvelope: { gzipBytes: 192, jsonBytes: 413 }, - grouped: { gzipBytes: 196, jsonBytes: 421 }, + legacyEnvelope: { gzipBytes: 194, jsonBytes: 413 }, + grouped: { gzipBytes: 198, jsonBytes: 421 }, }, { eventCount: 10, - legacyEnvelope: { gzipBytes: 243, jsonBytes: 3_554 }, - grouped: { gzipBytes: 243, jsonBytes: 3_049 }, + legacyEnvelope: { gzipBytes: 246, jsonBytes: 3_554 }, + grouped: { gzipBytes: 247, jsonBytes: 3_049 }, }, { eventCount: 50, - legacyEnvelope: { gzipBytes: 405, jsonBytes: 17_554 }, - grouped: { gzipBytes: 406, jsonBytes: 14_769 }, + legacyEnvelope: { gzipBytes: 406, jsonBytes: 17_554 }, + grouped: { gzipBytes: 407, jsonBytes: 14_769 }, }, ]); diff --git a/plugins/provider-codex/src/translator.test.ts b/plugins/provider-codex/src/translator.test.ts index f28843f7ce..aa7ccf500d 100644 --- a/plugins/provider-codex/src/translator.test.ts +++ b/plugins/provider-codex/src/translator.test.ts @@ -1166,6 +1166,57 @@ describe("codex subagent activity correlation", () => { expect(nextRootTurn).not.toHaveProperty("parentToolCallId"); }); + // A rawless message is not evidence that the next turn belongs to its + // target. If a different known child starts on the multiplexed root thread, + // that child's explicit pending delegation must win; the message is then + // discarded with its parent and cannot claim a later root turn either. + it("does not attach a rawless message to an unrelated multiplexed child", () => { + const harness = createHarness(); + expect( + harness.translate( + subAgentActivity({ + agentThreadId: "message-target-thread", + id: "rawless-message", + kind: "interacted", + }), + ), + ).toEqual([]); + + harness.translate( + subAgentActivity({ + agentThreadId: "unrelated-agent-thread", + id: "unrelated-subagent-call", + kind: "started", + }), + ); + const unrelatedChild = harness + .translate(childTurnStarted("unrelated-child-turn")) + .find((event) => event.type === "turn/started"); + expect(unrelatedChild).toEqual( + expect.objectContaining({ + type: "turn/started", + parentToolCallId: harness.itemId("unrelated-subagent-call"), + }), + ); + expect(unrelatedChild).not.toHaveProperty( + "parentToolCallId", + harness.itemId("rawless-message"), + ); + + harness.translate(childTurnCompleted("unrelated-child-turn")); + harness.translate(childTurnCompleted("parent-turn")); + expect( + harness.translator.prepareTurnStart({ + clientRequestId: "creq_after_unrelated_child", + providerThreadId: rootProviderThreadId, + }), + ).not.toBeNull(); + const nextRootTurn = harness + .translate(childTurnStarted("next-root-after-unrelated-child")) + .find((event) => event.type === "turn/started"); + expect(nextRootTurn).not.toHaveProperty("parentToolCallId"); + }); + // Each child turn consumes one ambiguous interaction. Settling the first // resumed turn must not discard the second interaction that still awaits // its own turn-producing proof. From ee7bc328cb299b12c2bd2023dd5267bfa75c0c2b Mon Sep 17 00:00:00 2001 From: Sawyer Hood Date: Fri, 28 Aug 2026 01:57:17 +0000 Subject: [PATCH 5/5] fix(codex): keep daemon protocol version stable --- packages/host-daemon-contract/src/protocol.ts | 7 +------ packages/host-daemon-contract/test/contract.test.ts | 2 +- 2 files changed, 2 insertions(+), 7 deletions(-) diff --git a/packages/host-daemon-contract/src/protocol.ts b/packages/host-daemon-contract/src/protocol.ts index ca4f555860..c398b5d911 100644 --- a/packages/host-daemon-contract/src/protocol.ts +++ b/packages/host-daemon-contract/src/protocol.ts @@ -335,14 +335,9 @@ // capacity implied by each model id. The usage event meaning changed across // the daemon boundary, so enrolled daemons must update with the server. // -// Version 172 restores Codex native-subagent correlation after a provider -// session resume. A followup to an agent unknown to the fresh bridge now emits -// a nested delegation and parents the resumed child turn instead of projecting -// it as competing root work; send_message remains non-turn-producing. -// // The version mismatch is what triggers the enrolled daemon's automatic update // instead of an `invalid-message` reconnect loop. -export const HOST_DAEMON_PROTOCOL_VERSION = 172 as const; +export const HOST_DAEMON_PROTOCOL_VERSION = 171 as const; /** * Absolute ceiling for any executable artifact delivered to a host daemon — diff --git a/packages/host-daemon-contract/test/contract.test.ts b/packages/host-daemon-contract/test/contract.test.ts index a8df5dea0d..2b3852f9c7 100644 --- a/packages/host-daemon-contract/test/contract.test.ts +++ b/packages/host-daemon-contract/test/contract.test.ts @@ -1046,7 +1046,7 @@ describe("host-daemon command schemas", () => { // mixed version. Version 113 carried the Devin Desktop open target rename // and remains part of the protocol lineage. it("uses the current host-daemon protocol version", () => { - expect(HOST_DAEMON_PROTOCOL_VERSION).toBe(172); + expect(HOST_DAEMON_PROTOCOL_VERSION).toBe(171); expect(HOST_ARTIFACT_MAX_BYTES).toBe(256 * 1024 * 1024); });