From ea59c91adebb6fe9c0a84e8f5f104d03f7a48751 Mon Sep 17 00:00:00 2001 From: Michael Johnston Date: Fri, 21 Aug 2026 15:05:41 -0700 Subject: [PATCH 01/10] fix(codex): restore pre-registration child liveness (cherry picked from commit 75b66cae0de74a7ebfa5a8d017a7444660cabe57) --- .../CodexCollabRuntime.integration.test.ts | 37 ++++++++++++++----- .../provider/Layers/CodexSessionRuntime.ts | 37 +++++++++++++++---- 2 files changed, 58 insertions(+), 16 deletions(-) diff --git a/apps/server/src/provider/Layers/CodexCollabRuntime.integration.test.ts b/apps/server/src/provider/Layers/CodexCollabRuntime.integration.test.ts index a1b46e003520..ae8eb3b6d490 100644 --- a/apps/server/src/provider/Layers/CodexCollabRuntime.integration.test.ts +++ b/apps/server/src/provider/Layers/CodexCollabRuntime.integration.test.ts @@ -181,6 +181,13 @@ describe("CodexSessionRuntime collab integration", () => { assert.isDefined(registrationA); assert.isDefined(registrationB); assert.isDefined(rootThreadStarted); + const interactedRegistrationA = { + ...registrationA, + params: { + ...registrationA.params, + item: { ...registrationA.params.item, kind: "interacted" }, + }, + }; const memoryThreadStarted = { ...rootThreadStarted, params: { @@ -207,7 +214,7 @@ describe("CodexSessionRuntime collab integration", () => { hangInterruptFor: CHILD_A, notifications: [ turnStartedA, - registrationA, + interactedRegistrationA, memoryThreadStarted, memoryTurnStarted, registrationB, @@ -233,26 +240,38 @@ describe("CodexSessionRuntime collab integration", () => { environment: { ...process.env, T3_CODEX_COLLAB_SCRIPT: scriptPath }, }); - // Wait for both children's turnStarted signals to be processed before - // stopping (B via the registered-child path; A only produces live-turn - // bookkeeping, so key on B's synthetic event). - const childBStartedFiber = yield* runtime.events.pipe( + // Wait for both children's synthetic turnStarted signals before + // stopping. B arrives through the registered-child path; A is replayed + // when its later activity registration finds the pre-registration live + // turn recorded by the foreign-notification suppressor. + const childrenStartedFiber = yield* runtime.events.pipe( Stream.filter( (event) => event.method === "collabAgent/turnStarted" && - (event.payload as { agentThreadId?: string }).agentThreadId === CHILD_B, + [CHILD_A, CHILD_B].includes( + (event.payload as { agentThreadId?: string }).agentThreadId ?? "", + ), ), - Stream.take(1), + Stream.take(2), Stream.runCollect, Effect.forkScoped, ); yield* runtime.start(); yield* runtime.sendTurn({ input: "fan out and hang" }); - const childBStarted = yield* Fiber.join(childBStartedFiber).pipe( + const childrenStarted = yield* Fiber.join(childrenStartedFiber).pipe( Effect.timeoutOption("15 seconds"), ); - assert.isTrue(childBStarted._tag === "Some", "child B turnStarted never arrived"); + assert.isTrue(childrenStarted._tag === "Some", "child turnStarted replay never arrived"); + if (childrenStarted._tag === "Some") { + const startedThreadIds = new Set( + Array.from(childrenStarted.value).map( + (event) => (event.payload as { agentThreadId?: string }).agentThreadId, + ), + ); + assert.isTrue(startedThreadIds.has(CHILD_A), "child A start must replay on registration"); + assert.isTrue(startedThreadIds.has(CHILD_B), "child B start must flow after registration"); + } // Stop everything. A's interrupt hangs forever — the bounded child // deadline must expire and the parent interrupt must still be sent. diff --git a/apps/server/src/provider/Layers/CodexSessionRuntime.ts b/apps/server/src/provider/Layers/CodexSessionRuntime.ts index fd926e43d7bf..d6ba1d7fc0bf 100644 --- a/apps/server/src/provider/Layers/CodexSessionRuntime.ts +++ b/apps/server/src/provider/Layers/CodexSessionRuntime.ts @@ -1107,8 +1107,8 @@ export const makeCodexSessionRuntime = ( return false; } const activitySpawnTurnId = (yield* Ref.get(sessionRef)).activeTurnId ?? undefined; + const existingChild = (yield* Ref.get(collabChildAgentsRef)).get(item.agentThreadId); yield* Ref.update(collabChildAgentsRef, (current) => { - const existing = current.get(item.agentThreadId); const next = new Map(current); // Merge-late semantics: when thread/started registered first, a // later subAgentActivity still carries the real agentPath (and a @@ -1120,13 +1120,13 @@ export const makeCodexSessionRuntime = ( next.set(item.agentThreadId, { agentThreadId: item.agentThreadId, nickname: - existing?.nickname ?? + existingChild?.nickname ?? item.agentPath.split("/").findLast((segment) => segment.length > 0), - role: existing?.role, - agentPath: existing?.agentPath ?? item.agentPath, - depth: existing?.depth, - parentThreadId: existing?.parentThreadId, - spawnTurnId: existing ? existing.spawnTurnId : activitySpawnTurnId, + role: existingChild?.role, + agentPath: existingChild?.agentPath ?? item.agentPath, + depth: existingChild?.depth, + parentThreadId: existingChild?.parentThreadId, + spawnTurnId: existingChild ? existingChild.spawnTurnId : activitySpawnTurnId, }); return next; }); @@ -1142,6 +1142,29 @@ export const makeCodexSessionRuntime = ( activityKind: item.kind, }, }); + // A child turn can start before this activity registers the child. + // The foreign-notification suppressor records that live turn but + // cannot emit agent lifecycle until identity is known. Replay the + // explicit start after first registration so sidebar liveness sees + // genuine work; a trailing interaction with no live turn remains + // ignored by CodexAdapter. + const preRegistrationLiveTurn = (yield* Ref.get(collabChildLiveTurnsRef)).get( + item.agentThreadId, + ); + if (!existingChild && item.kind === "interacted" && preRegistrationLiveTurn) { + yield* emitEvent({ + kind: "notification", + threadId: options.threadId, + ...(registeredChild?.spawnTurnId ? { turnId: registeredChild.spawnTurnId } : {}), + method: "collabAgent/turnStarted", + payload: { + agentThreadId: item.agentThreadId, + ...(registeredChild?.nickname ? { nickname: registeredChild.nickname } : {}), + ...(registeredChild?.role ? { role: registeredChild.role } : {}), + agentPath: item.agentPath, + }, + }); + } return true; } From a433591b001c7410cdb68deef2d011d14f28b65c Mon Sep 17 00:00:00 2001 From: Michael Johnston Date: Sun, 23 Aug 2026 13:57:00 -0700 Subject: [PATCH 02/10] fix(codex): clear failed pre-registration turns --- .../CodexCollabRuntime.integration.test.ts | 113 ++++++++++++++++++ .../provider/Layers/CodexSessionRuntime.ts | 4 +- 2 files changed, 116 insertions(+), 1 deletion(-) diff --git a/apps/server/src/provider/Layers/CodexCollabRuntime.integration.test.ts b/apps/server/src/provider/Layers/CodexCollabRuntime.integration.test.ts index ae8eb3b6d490..5c9d57357688 100644 --- a/apps/server/src/provider/Layers/CodexCollabRuntime.integration.test.ts +++ b/apps/server/src/provider/Layers/CodexCollabRuntime.integration.test.ts @@ -148,6 +148,119 @@ describe("CodexSessionRuntime collab integration", () => { }).pipe(Effect.scoped, Effect.provide(NodeServices.layer)), ); + it.effect("replays only retrying pre-registration child turns after errors", () => + Effect.gen(function* () { + const byIndex = wireFixture.notifications; + const turnStartedA = byIndex.find( + (entry) => + entry.method === "turn/started" && + (entry.params as { threadId?: string }).threadId === CHILD_A, + ); + const turnStartedB = byIndex.find( + (entry) => + entry.method === "turn/started" && + (entry.params as { threadId?: string }).threadId === CHILD_B, + ); + const registrationA = byIndex.find((entry) => { + const item = (entry.params as { item?: { type?: string; agentThreadId?: string } }).item; + return item?.type === "subAgentActivity" && item.agentThreadId === CHILD_A; + }); + const registrationB = byIndex.find((entry) => { + const item = (entry.params as { item?: { type?: string; agentThreadId?: string } }).item; + return item?.type === "subAgentActivity" && item.agentThreadId === CHILD_B; + }); + assert.isDefined(turnStartedA); + assert.isDefined(turnStartedB); + assert.isDefined(registrationA); + assert.isDefined(registrationB); + const turnIdA = (turnStartedA.params as { turn: { id: string } }).turn.id; + const turnIdB = (turnStartedB.params as { turn: { id: string } }).turn.id; + + const script = { + rootThreadId: ROOT, + notifications: [ + turnStartedA, + { + method: "error", + params: { + threadId: CHILD_A, + turnId: turnIdA, + error: { message: "child failed before registration" }, + willRetry: false, + }, + }, + { + ...registrationA, + params: { + ...registrationA.params, + item: { ...registrationA.params.item, kind: "interacted" }, + }, + }, + turnStartedB, + { + method: "error", + params: { + threadId: CHILD_B, + turnId: turnIdB, + error: { message: "child will retry before registration" }, + willRetry: true, + }, + }, + { + ...registrationB, + params: { + ...registrationB.params, + item: { ...registrationB.params.item, kind: "interacted" }, + }, + }, + ], + }; + // @effect-diagnostics-next-line preferSchemaOverJson:off + NodeFS.writeFileSync(scriptPath, JSON.stringify(script), "utf8"); + yield* Effect.addFinalizer(() => + Effect.sync(() => NodeFS.rmSync(scriptPath, { force: true })), + ); + + const runtime = yield* makeCodexSessionRuntime({ + threadId: ThreadId.make("thread-collab-terminal-before-registration"), + binaryPath: peerPath, + cwd: "/tmp", + runtimeMode: "full-access", + environment: { ...process.env, T3_CODEX_COLLAB_SCRIPT: scriptPath }, + }); + const eventsFiber = yield* runtime.events.pipe( + Stream.takeUntil((event) => event.method === "turn/completed"), + Stream.runCollect, + Effect.forkScoped, + ); + + yield* runtime.start(); + yield* runtime.sendTurn({ input: "error before registration" }); + const events = Array.from(yield* Fiber.join(eventsFiber)); + const startedThreadIds = events + .filter((event) => event.method === "collabAgent/turnStarted") + .map((event) => (event.payload as { agentThreadId?: string }).agentThreadId); + + assert.notInclude( + startedThreadIds, + CHILD_A, + "a terminal child turn must not replay as live when activity registers it later", + ); + assert.include( + startedThreadIds, + CHILD_B, + "a retrying child turn must remain live when activity registers it later", + ); + assert.notInclude( + events.map((event) => event.method), + "error", + "pre-registration child errors must not leak onto the parent event stream", + ); + + yield* runtime.close; + }).pipe(Effect.scoped, Effect.provide(NodeServices.layer)), + ); + // it.live: the runtime talks to a real child process; under it.effect's // TestClock the internal timers freeze and the join never completes. it.live("Stop interrupts every live child regardless of registration timing", () => diff --git a/apps/server/src/provider/Layers/CodexSessionRuntime.ts b/apps/server/src/provider/Layers/CodexSessionRuntime.ts index d6ba1d7fc0bf..74defd4754ff 100644 --- a/apps/server/src/provider/Layers/CodexSessionRuntime.ts +++ b/apps/server/src/provider/Layers/CodexSessionRuntime.ts @@ -753,6 +753,7 @@ function shouldSuppressChildConversationNotification( method === "thread/tokenUsage/updated" || method === "turn/started" || method === "turn/completed" || + method === "error" || method === "turn/plan/updated" || method === "item/plan/delta" ); @@ -1392,7 +1393,8 @@ export const makeCodexSessionRuntime = ( } } else if ( notification.method === "turn/completed" || - notification.method === "thread/closed" + notification.method === "thread/closed" || + (notification.method === "error" && !notification.params.willRetry) ) { yield* Ref.update(collabChildLiveTurnsRef, (current) => { const next = new Map(current); From 2f0401813558c732f58ace31b3c7b9aab844f8cd Mon Sep 17 00:00:00 2001 From: Michael Johnston Date: Sun, 23 Aug 2026 15:51:37 -0700 Subject: [PATCH 03/10] fix(codex): keep late child failures terminal --- .../CodexCollabRuntime.integration.test.ts | 26 ++++++--- .../provider/Layers/CodexSessionRuntime.ts | 57 +++++++++++++++---- 2 files changed, 65 insertions(+), 18 deletions(-) diff --git a/apps/server/src/provider/Layers/CodexCollabRuntime.integration.test.ts b/apps/server/src/provider/Layers/CodexCollabRuntime.integration.test.ts index 5c9d57357688..dd8c524847d9 100644 --- a/apps/server/src/provider/Layers/CodexCollabRuntime.integration.test.ts +++ b/apps/server/src/provider/Layers/CodexCollabRuntime.integration.test.ts @@ -189,13 +189,7 @@ describe("CodexSessionRuntime collab integration", () => { willRetry: false, }, }, - { - ...registrationA, - params: { - ...registrationA.params, - item: { ...registrationA.params.item, kind: "interacted" }, - }, - }, + registrationA, turnStartedB, { method: "error", @@ -240,12 +234,30 @@ describe("CodexSessionRuntime collab integration", () => { const startedThreadIds = events .filter((event) => event.method === "collabAgent/turnStarted") .map((event) => (event.payload as { agentThreadId?: string }).agentThreadId); + const childAActivityIndex = events.findIndex( + (event) => + event.method === "collabAgent/activity" && + (event.payload as { agentThreadId?: string }).agentThreadId === CHILD_A, + ); + const childAFailureIndex = events.findIndex( + (event) => + event.method === "collabAgent/statusChanged" && + (event.payload as { agentThreadId?: string; status?: { type?: string } }) + .agentThreadId === CHILD_A && + (event.payload as { status?: { type?: string } }).status?.type === "systemError", + ); assert.notInclude( startedThreadIds, CHILD_A, "a terminal child turn must not replay as live when activity registers it later", ); + assert.isAtLeast(childAActivityIndex, 0, "the late child registration must still surface"); + assert.isAbove( + childAFailureIndex, + childAActivityIndex, + "terminal state must follow a late started registration so liveness settles failed", + ); assert.include( startedThreadIds, CHILD_B, diff --git a/apps/server/src/provider/Layers/CodexSessionRuntime.ts b/apps/server/src/provider/Layers/CodexSessionRuntime.ts index 74defd4754ff..575c7a621627 100644 --- a/apps/server/src/provider/Layers/CodexSessionRuntime.ts +++ b/apps/server/src/provider/Layers/CodexSessionRuntime.ts @@ -910,6 +910,8 @@ export const makeCodexSessionRuntime = ( const collabChildAgentsRef = yield* Ref.make(new Map()); /** Child provider-thread id → its currently running provider turn id. */ const collabChildLiveTurnsRef = yield* Ref.make(new Map()); + /** Unregistered child threads whose latest observed turn failed terminally. */ + const collabChildPreRegistrationFailuresRef = yield* Ref.make(new Set()); const suppressMemoryConsolidationNotification = makeMemoryConsolidationNotificationFilter(); const closedRef = yield* Ref.make(false); @@ -1002,6 +1004,20 @@ export const makeCodexSessionRuntime = ( method, message, }); + const emitCollabChildSystemError = (child: CollabChildAgentState) => + emitEvent({ + kind: "notification", + threadId: options.threadId, + ...(child.spawnTurnId ? { turnId: child.spawnTurnId } : {}), + method: "collabAgent/statusChanged", + payload: { + agentThreadId: child.agentThreadId, + ...(child.nickname ? { nickname: child.nickname } : {}), + ...(child.role ? { role: child.role } : {}), + ...(child.agentPath ? { agentPath: child.agentPath } : {}), + status: { type: "systemError" }, + }, + }); const settlePendingApprovals = (decision: ProviderApprovalDecision) => Ref.get(pendingApprovalsRef).pipe( @@ -1051,6 +1067,9 @@ export const makeCodexSessionRuntime = ( // child onto a new fleet's CTA (review finding). Only a genuinely // new registration captures the current turn. const existingChild = (yield* Ref.get(collabChildAgentsRef)).get(thread.id); + const preRegistrationFailure = (yield* Ref.get( + collabChildPreRegistrationFailuresRef, + )).has(thread.id); const spawnTurnId = existingChild ? existingChild.spawnTurnId : ((yield* Ref.get(sessionRef)).activeTurnId ?? undefined); @@ -1083,6 +1102,9 @@ export const makeCodexSessionRuntime = ( ...(state.parentThreadId ? { parentThreadId: state.parentThreadId } : {}), }, }); + if (preRegistrationFailure) { + yield* emitCollabChildSystemError(state); + } return true; } @@ -1109,6 +1131,9 @@ export const makeCodexSessionRuntime = ( } const activitySpawnTurnId = (yield* Ref.get(sessionRef)).activeTurnId ?? undefined; const existingChild = (yield* Ref.get(collabChildAgentsRef)).get(item.agentThreadId); + const preRegistrationFailure = (yield* Ref.get( + collabChildPreRegistrationFailuresRef, + )).has(item.agentThreadId); yield* Ref.update(collabChildAgentsRef, (current) => { const next = new Map(current); // Merge-late semantics: when thread/started registered first, a @@ -1152,7 +1177,9 @@ export const makeCodexSessionRuntime = ( const preRegistrationLiveTurn = (yield* Ref.get(collabChildLiveTurnsRef)).get( item.agentThreadId, ); - if (!existingChild && item.kind === "interacted" && preRegistrationLiveTurn) { + if (preRegistrationFailure && registeredChild) { + yield* emitCollabChildSystemError(registeredChild); + } else if (!existingChild && item.kind === "interacted" && preRegistrationLiveTurn) { yield* emitEvent({ kind: "notification", threadId: options.threadId, @@ -1199,6 +1226,11 @@ export const makeCodexSessionRuntime = ( ? ((notification.params as { turn: { id: string } }).turn.id as string) : undefined; if (childTurnId) { + yield* Ref.update(collabChildPreRegistrationFailuresRef, (current) => { + const next = new Set(current); + next.delete(child.agentThreadId); + return next; + }); yield* Ref.update(collabChildLiveTurnsRef, (current) => { const next = new Map(current); next.set(child.agentThreadId, childTurnId); @@ -1303,16 +1335,7 @@ export const makeCodexSessionRuntime = ( next.delete(child.agentThreadId); return next; }); - yield* emitEvent({ - kind: "notification", - threadId: options.threadId, - ...(child.spawnTurnId ? { turnId: child.spawnTurnId } : {}), - method: "collabAgent/statusChanged", - payload: { - ...childIdentity, - status: { type: "systemError" }, - }, - }); + yield* emitCollabChildSystemError(child); return true; } default: @@ -1385,6 +1408,11 @@ export const makeCodexSessionRuntime = ( ? (notification.params as { turn: { id: string } }).turn.id : undefined; if (foreignTurnId) { + yield* Ref.update(collabChildPreRegistrationFailuresRef, (current) => { + const next = new Set(current); + next.delete(foreignThreadId); + return next; + }); yield* Ref.update(collabChildLiveTurnsRef, (current) => { const next = new Map(current); next.set(foreignThreadId, foreignTurnId); @@ -1401,6 +1429,13 @@ export const makeCodexSessionRuntime = ( next.delete(foreignThreadId); return next; }); + if (notification.method === "error" && !notification.params.willRetry) { + yield* Ref.update(collabChildPreRegistrationFailuresRef, (current) => { + const next = new Set(current); + next.add(foreignThreadId); + return next; + }); + } } } yield* Ref.set(collabReceiverTurnsRef, collabReceiverTurns); From b422b3d707b174aa9c278b907b5ddb4fda18016f Mon Sep 17 00:00:00 2001 From: Michael Johnston Date: Sun, 23 Aug 2026 15:57:43 -0700 Subject: [PATCH 04/10] fix(codex): consume late child failure state --- .../CodexCollabRuntime.integration.test.ts | 44 +++++++-- .../provider/Layers/CodexSessionRuntime.ts | 99 +++++++++++++------ 2 files changed, 103 insertions(+), 40 deletions(-) diff --git a/apps/server/src/provider/Layers/CodexCollabRuntime.integration.test.ts b/apps/server/src/provider/Layers/CodexCollabRuntime.integration.test.ts index dd8c524847d9..0428060039ec 100644 --- a/apps/server/src/provider/Layers/CodexCollabRuntime.integration.test.ts +++ b/apps/server/src/provider/Layers/CodexCollabRuntime.integration.test.ts @@ -165,6 +165,7 @@ describe("CodexSessionRuntime collab integration", () => { const item = (entry.params as { item?: { type?: string; agentThreadId?: string } }).item; return item?.type === "subAgentActivity" && item.agentThreadId === CHILD_A; }); + const rootThreadStarted = byIndex.find((entry) => entry.method === "thread/started"); const registrationB = byIndex.find((entry) => { const item = (entry.params as { item?: { type?: string; agentThreadId?: string } }).item; return item?.type === "subAgentActivity" && item.agentThreadId === CHILD_B; @@ -173,8 +174,30 @@ describe("CodexSessionRuntime collab integration", () => { assert.isDefined(turnStartedB); assert.isDefined(registrationA); assert.isDefined(registrationB); + assert.isDefined(rootThreadStarted); const turnIdA = (turnStartedA.params as { turn: { id: string } }).turn.id; const turnIdB = (turnStartedB.params as { turn: { id: string } }).turn.id; + const threadRegistrationA = { + ...rootThreadStarted, + params: { + thread: { + ...rootThreadStarted.params.thread, + id: CHILD_A, + sessionId: CHILD_A, + parentThreadId: ROOT, + source: { + subAgent: { + thread_spawn: { + agent_nickname: "alpha", + agent_path: "/root/alpha", + depth: 1, + parent_thread_id: ROOT, + }, + }, + }, + }, + }, + }; const script = { rootThreadId: ROOT, @@ -189,6 +212,7 @@ describe("CodexSessionRuntime collab integration", () => { willRetry: false, }, }, + threadRegistrationA, registrationA, turnStartedB, { @@ -234,12 +258,12 @@ describe("CodexSessionRuntime collab integration", () => { const startedThreadIds = events .filter((event) => event.method === "collabAgent/turnStarted") .map((event) => (event.payload as { agentThreadId?: string }).agentThreadId); - const childAActivityIndex = events.findIndex( + const childARegistrationEvents = events.filter( (event) => - event.method === "collabAgent/activity" && + (event.method === "collabAgent/started" || event.method === "collabAgent/activity") && (event.payload as { agentThreadId?: string }).agentThreadId === CHILD_A, ); - const childAFailureIndex = events.findIndex( + const childAFailures = events.filter( (event) => event.method === "collabAgent/statusChanged" && (event.payload as { agentThreadId?: string; status?: { type?: string } }) @@ -252,11 +276,15 @@ describe("CodexSessionRuntime collab integration", () => { CHILD_A, "a terminal child turn must not replay as live when activity registers it later", ); - assert.isAtLeast(childAActivityIndex, 0, "the late child registration must still surface"); - assert.isAbove( - childAFailureIndex, - childAActivityIndex, - "terminal state must follow a late started registration so liveness settles failed", + assert.deepEqual( + childARegistrationEvents.map((event) => event.method), + [], + "late registration through either path must not restart a failed child", + ); + assert.lengthOf( + childAFailures, + 1, + "the terminal child state must surface once across both registration paths", ); assert.include( startedThreadIds, diff --git a/apps/server/src/provider/Layers/CodexSessionRuntime.ts b/apps/server/src/provider/Layers/CodexSessionRuntime.ts index 575c7a621627..acefc685ac4f 100644 --- a/apps/server/src/provider/Layers/CodexSessionRuntime.ts +++ b/apps/server/src/provider/Layers/CodexSessionRuntime.ts @@ -676,6 +676,8 @@ interface CollabChildAgentState { readonly agentPath: string | undefined; readonly depth: number | undefined; readonly parentThreadId: string | undefined; + /** A terminal error remains authoritative until a genuine new turn starts. */ + readonly terminalError: boolean; /** * Parent canonical turn active when the child registered. Stamped on every * synthetic collabAgent/* event so clients can batch a fleet by its spawn @@ -1082,28 +1084,39 @@ export const makeCodexSessionRuntime = ( parentThreadId: spawn.parentThreadId ?? thread.parentThreadId ?? existingChild?.parentThreadId, spawnTurnId, + terminalError: existingChild?.terminalError ?? preRegistrationFailure, }; yield* Ref.update(collabChildAgentsRef, (current) => { const next = new Map(current); next.set(thread.id, state); return next; }); - yield* emitEvent({ - kind: "notification", - threadId: options.threadId, - method: "collabAgent/started", - ...(state.spawnTurnId ? { turnId: state.spawnTurnId } : {}), - payload: { - agentThreadId: state.agentThreadId, - ...(state.nickname ? { nickname: state.nickname } : {}), - ...(state.role ? { role: state.role } : {}), - ...(state.agentPath ? { agentPath: state.agentPath } : {}), - ...(state.depth !== undefined ? { depth: state.depth } : {}), - ...(state.parentThreadId ? { parentThreadId: state.parentThreadId } : {}), - }, - }); if (preRegistrationFailure) { - yield* emitCollabChildSystemError(state); + yield* Ref.update(collabChildPreRegistrationFailuresRef, (current) => { + const next = new Set(current); + next.delete(thread.id); + return next; + }); + } + if (state.terminalError) { + if (!existingChild) { + yield* emitCollabChildSystemError(state); + } + } else { + yield* emitEvent({ + kind: "notification", + threadId: options.threadId, + method: "collabAgent/started", + ...(state.spawnTurnId ? { turnId: state.spawnTurnId } : {}), + payload: { + agentThreadId: state.agentThreadId, + ...(state.nickname ? { nickname: state.nickname } : {}), + ...(state.role ? { role: state.role } : {}), + ...(state.agentPath ? { agentPath: state.agentPath } : {}), + ...(state.depth !== undefined ? { depth: state.depth } : {}), + ...(state.parentThreadId ? { parentThreadId: state.parentThreadId } : {}), + }, + }); } return true; } @@ -1153,21 +1166,18 @@ export const makeCodexSessionRuntime = ( depth: existingChild?.depth, parentThreadId: existingChild?.parentThreadId, spawnTurnId: existingChild ? existingChild.spawnTurnId : activitySpawnTurnId, + terminalError: existingChild?.terminalError ?? preRegistrationFailure, }); return next; }); const registeredChild = (yield* Ref.get(collabChildAgentsRef)).get(item.agentThreadId); - yield* emitEvent({ - kind: "notification", - threadId: options.threadId, - method: "collabAgent/activity", - ...(registeredChild?.spawnTurnId ? { turnId: registeredChild.spawnTurnId } : {}), - payload: { - agentThreadId: item.agentThreadId, - agentPath: item.agentPath, - activityKind: item.kind, - }, - }); + if (preRegistrationFailure) { + yield* Ref.update(collabChildPreRegistrationFailuresRef, (current) => { + const next = new Set(current); + next.delete(item.agentThreadId); + return next; + }); + } // A child turn can start before this activity registers the child. // The foreign-notification suppressor records that live turn but // cannot emit agent lifecycle until identity is known. Replay the @@ -1177,21 +1187,36 @@ export const makeCodexSessionRuntime = ( const preRegistrationLiveTurn = (yield* Ref.get(collabChildLiveTurnsRef)).get( item.agentThreadId, ); - if (preRegistrationFailure && registeredChild) { - yield* emitCollabChildSystemError(registeredChild); - } else if (!existingChild && item.kind === "interacted" && preRegistrationLiveTurn) { + if (registeredChild?.terminalError) { + if (!existingChild) { + yield* emitCollabChildSystemError(registeredChild); + } + } else { yield* emitEvent({ kind: "notification", threadId: options.threadId, + method: "collabAgent/activity", ...(registeredChild?.spawnTurnId ? { turnId: registeredChild.spawnTurnId } : {}), - method: "collabAgent/turnStarted", payload: { agentThreadId: item.agentThreadId, - ...(registeredChild?.nickname ? { nickname: registeredChild.nickname } : {}), - ...(registeredChild?.role ? { role: registeredChild.role } : {}), agentPath: item.agentPath, + activityKind: item.kind, }, }); + if (!existingChild && item.kind === "interacted" && preRegistrationLiveTurn) { + yield* emitEvent({ + kind: "notification", + threadId: options.threadId, + ...(registeredChild?.spawnTurnId ? { turnId: registeredChild.spawnTurnId } : {}), + method: "collabAgent/turnStarted", + payload: { + agentThreadId: item.agentThreadId, + ...(registeredChild?.nickname ? { nickname: registeredChild.nickname } : {}), + ...(registeredChild?.role ? { role: registeredChild.role } : {}), + agentPath: item.agentPath, + }, + }); + } } return true; } @@ -1236,6 +1261,11 @@ export const makeCodexSessionRuntime = ( next.set(child.agentThreadId, childTurnId); return next; }); + yield* Ref.update(collabChildAgentsRef, (current) => { + const next = new Map(current); + next.set(child.agentThreadId, { ...child, terminalError: false }); + return next; + }); } yield* emitEvent({ kind: "notification", @@ -1335,6 +1365,11 @@ export const makeCodexSessionRuntime = ( next.delete(child.agentThreadId); return next; }); + yield* Ref.update(collabChildAgentsRef, (current) => { + const next = new Map(current); + next.set(child.agentThreadId, { ...child, terminalError: true }); + return next; + }); yield* emitCollabChildSystemError(child); return true; } From 79688466b49ed866a494b548878ad5012603e815 Mon Sep 17 00:00:00 2001 From: Michael Johnston Date: Sun, 23 Aug 2026 16:08:51 -0700 Subject: [PATCH 05/10] fix(codex): anchor late child failures --- .../CodexCollabRuntime.integration.test.ts | 4 +-- .../provider/Layers/CodexSessionRuntime.ts | 32 +++++++++++-------- 2 files changed, 20 insertions(+), 16 deletions(-) diff --git a/apps/server/src/provider/Layers/CodexCollabRuntime.integration.test.ts b/apps/server/src/provider/Layers/CodexCollabRuntime.integration.test.ts index 0428060039ec..66df6c32cb41 100644 --- a/apps/server/src/provider/Layers/CodexCollabRuntime.integration.test.ts +++ b/apps/server/src/provider/Layers/CodexCollabRuntime.integration.test.ts @@ -278,8 +278,8 @@ describe("CodexSessionRuntime collab integration", () => { ); assert.deepEqual( childARegistrationEvents.map((event) => event.method), - [], - "late registration through either path must not restart a failed child", + ["collabAgent/started"], + "a failed child needs one start anchor, but later registration must not duplicate it", ); assert.lengthOf( childAFailures, diff --git a/apps/server/src/provider/Layers/CodexSessionRuntime.ts b/apps/server/src/provider/Layers/CodexSessionRuntime.ts index acefc685ac4f..1461bc4905df 100644 --- a/apps/server/src/provider/Layers/CodexSessionRuntime.ts +++ b/apps/server/src/provider/Layers/CodexSessionRuntime.ts @@ -1006,6 +1006,21 @@ export const makeCodexSessionRuntime = ( method, message, }); + const emitCollabChildStarted = (child: CollabChildAgentState) => + emitEvent({ + kind: "notification", + threadId: options.threadId, + method: "collabAgent/started", + ...(child.spawnTurnId ? { turnId: child.spawnTurnId } : {}), + payload: { + agentThreadId: child.agentThreadId, + ...(child.nickname ? { nickname: child.nickname } : {}), + ...(child.role ? { role: child.role } : {}), + ...(child.agentPath ? { agentPath: child.agentPath } : {}), + ...(child.depth !== undefined ? { depth: child.depth } : {}), + ...(child.parentThreadId ? { parentThreadId: child.parentThreadId } : {}), + }, + }); const emitCollabChildSystemError = (child: CollabChildAgentState) => emitEvent({ kind: "notification", @@ -1100,23 +1115,11 @@ export const makeCodexSessionRuntime = ( } if (state.terminalError) { if (!existingChild) { + yield* emitCollabChildStarted(state); yield* emitCollabChildSystemError(state); } } else { - yield* emitEvent({ - kind: "notification", - threadId: options.threadId, - method: "collabAgent/started", - ...(state.spawnTurnId ? { turnId: state.spawnTurnId } : {}), - payload: { - agentThreadId: state.agentThreadId, - ...(state.nickname ? { nickname: state.nickname } : {}), - ...(state.role ? { role: state.role } : {}), - ...(state.agentPath ? { agentPath: state.agentPath } : {}), - ...(state.depth !== undefined ? { depth: state.depth } : {}), - ...(state.parentThreadId ? { parentThreadId: state.parentThreadId } : {}), - }, - }); + yield* emitCollabChildStarted(state); } return true; } @@ -1189,6 +1192,7 @@ export const makeCodexSessionRuntime = ( ); if (registeredChild?.terminalError) { if (!existingChild) { + yield* emitCollabChildStarted(registeredChild); yield* emitCollabChildSystemError(registeredChild); } } else { From 8a913395970ca4692e94ad7fa3ce778132eeb038 Mon Sep 17 00:00:00 2001 From: Michael Johnston Date: Sun, 23 Aug 2026 16:34:44 -0700 Subject: [PATCH 06/10] fix(codex): retain terminal child metadata --- .../src/provider/Layers/CodexAdapter.test.ts | 38 +++++++++++++++++++ .../src/provider/Layers/CodexAdapter.ts | 3 ++ .../CodexCollabRuntime.integration.test.ts | 13 +++++-- .../provider/Layers/CodexSessionRuntime.ts | 10 +++++ 4 files changed, 61 insertions(+), 3 deletions(-) diff --git a/apps/server/src/provider/Layers/CodexAdapter.test.ts b/apps/server/src/provider/Layers/CodexAdapter.test.ts index 26fb1b166f61..15aceec5e83f 100644 --- a/apps/server/src/provider/Layers/CodexAdapter.test.ts +++ b/apps/server/src/provider/Layers/CodexAdapter.test.ts @@ -618,6 +618,44 @@ lifecycleLayer("CodexAdapterLive lifecycle", (it) => { }), ); + it.effect("preserves parent linkage on terminal child metadata patches", () => + Effect.gen(function* () { + const { adapter, runtime } = yield* startLifecycleRuntime(); + const eventFiber = yield* Stream.runHead(adapter.streamEvents).pipe(Effect.forkChild); + + yield* runtime.emit({ + id: asEventId("evt-child-failed-metadata"), + kind: "notification", + provider: ProviderDriverKind.make("codex"), + createdAt: "2026-01-01T00:00:00.000Z", + method: "collabAgent/statusChanged", + threadId: asThreadId("thread-1"), + turnId: asTurnId("turn-1"), + payload: { + agentThreadId: "child-1", + agentPath: "/root/audit", + parentThreadId: "workflow-1", + status: { type: "systemError" }, + }, + }); + + const event = yield* Fiber.join(eventFiber); + NodeAssert.equal(event._tag, "Some"); + if (event._tag === "Some") { + NodeAssert.equal(event.value.type, "task.updated"); + NodeAssert.deepStrictEqual(event.value.payload, { + taskId: "child-1", + status: "failed", + role: "audit", + title: "audit", + agentPath: "/root/audit", + parentAgentId: "workflow-1", + timelineBypass: true, + }); + } + }), + ); + it.effect("maps completed agent message items to canonical item.completed events", () => Effect.gen(function* () { const { adapter, runtime } = yield* startLifecycleRuntime(); diff --git a/apps/server/src/provider/Layers/CodexAdapter.ts b/apps/server/src/provider/Layers/CodexAdapter.ts index bc48f94b3866..5e8810b41a61 100644 --- a/apps/server/src/provider/Layers/CodexAdapter.ts +++ b/apps/server/src/provider/Layers/CodexAdapter.ts @@ -537,6 +537,9 @@ function mapCollabAgentEvent( role, ...(knownName ? { title: knownName } : {}), ...(agentPath ? { agentPath } : {}), + ...(typeof payload.parentThreadId === "string" + ? { parentAgentId: payload.parentThreadId } + : {}), timelineBypass: true, } as const; diff --git a/apps/server/src/provider/Layers/CodexCollabRuntime.integration.test.ts b/apps/server/src/provider/Layers/CodexCollabRuntime.integration.test.ts index 66df6c32cb41..ec6cda75a464 100644 --- a/apps/server/src/provider/Layers/CodexCollabRuntime.integration.test.ts +++ b/apps/server/src/provider/Layers/CodexCollabRuntime.integration.test.ts @@ -212,8 +212,8 @@ describe("CodexSessionRuntime collab integration", () => { willRetry: false, }, }, - threadRegistrationA, registrationA, + threadRegistrationA, turnStartedB, { method: "error", @@ -283,8 +283,15 @@ describe("CodexSessionRuntime collab integration", () => { ); assert.lengthOf( childAFailures, - 1, - "the terminal child state must surface once across both registration paths", + 2, + "late thread metadata must enrich the terminal state without restarting the child", + ); + const terminalMetadataFailure = childAFailures.at(-1); + assert.isDefined(terminalMetadataFailure); + assert.equal( + (terminalMetadataFailure.payload as { parentThreadId?: string }).parentThreadId, + ROOT, + "the terminal metadata patch must preserve parent linkage from thread registration", ); assert.include( startedThreadIds, diff --git a/apps/server/src/provider/Layers/CodexSessionRuntime.ts b/apps/server/src/provider/Layers/CodexSessionRuntime.ts index 1461bc4905df..ddb2efb9faff 100644 --- a/apps/server/src/provider/Layers/CodexSessionRuntime.ts +++ b/apps/server/src/provider/Layers/CodexSessionRuntime.ts @@ -1032,6 +1032,7 @@ export const makeCodexSessionRuntime = ( ...(child.nickname ? { nickname: child.nickname } : {}), ...(child.role ? { role: child.role } : {}), ...(child.agentPath ? { agentPath: child.agentPath } : {}), + ...(child.parentThreadId ? { parentThreadId: child.parentThreadId } : {}), status: { type: "systemError" }, }, }); @@ -1117,6 +1118,15 @@ export const makeCodexSessionRuntime = ( if (!existingChild) { yield* emitCollabChildStarted(state); yield* emitCollabChildSystemError(state); + } else if ( + existingChild.nickname !== state.nickname || + existingChild.role !== state.role || + existingChild.agentPath !== state.agentPath || + existingChild.parentThreadId !== state.parentThreadId + ) { + // Keep the terminal status authoritative while propagating + // identity that arrived after the first registration path. + yield* emitCollabChildSystemError(state); } } else { yield* emitCollabChildStarted(state); From 570497487bf59e54632013786e51117d3c2630f2 Mon Sep 17 00:00:00 2001 From: Michael Johnston Date: Sun, 23 Aug 2026 17:21:49 -0700 Subject: [PATCH 07/10] fix(codex): preserve terminal child failures --- .../CodexCollabRuntime.integration.test.ts | 33 +++++++++++++++++++ .../provider/Layers/CodexSessionRuntime.ts | 9 +++++ 2 files changed, 42 insertions(+) diff --git a/apps/server/src/provider/Layers/CodexCollabRuntime.integration.test.ts b/apps/server/src/provider/Layers/CodexCollabRuntime.integration.test.ts index ec6cda75a464..c990d79d706a 100644 --- a/apps/server/src/provider/Layers/CodexCollabRuntime.integration.test.ts +++ b/apps/server/src/provider/Layers/CodexCollabRuntime.integration.test.ts @@ -214,6 +214,18 @@ describe("CodexSessionRuntime collab integration", () => { }, registrationA, threadRegistrationA, + { + method: "thread/status/changed", + params: { threadId: CHILD_A, status: { type: "idle" } }, + }, + { + method: "turn/completed", + params: { + threadId: CHILD_A, + turn: { id: turnIdA, status: "completed", items: [] }, + }, + }, + { method: "thread/closed", params: { threadId: CHILD_A } }, turnStartedB, { method: "error", @@ -270,6 +282,22 @@ describe("CodexSessionRuntime collab integration", () => { .agentThreadId === CHILD_A && (event.payload as { status?: { type?: string } }).status?.type === "systemError", ); + const childATerminalOverrides = events.filter((event) => { + if (!event.payload || typeof event.payload !== "object") { + return false; + } + const payload = event.payload as { + agentThreadId?: string; + status?: { type?: string }; + }; + return ( + payload.agentThreadId === CHILD_A && + (event.method === "collabAgent/turnCompleted" || + event.method === "collabAgent/closed" || + (event.method === "collabAgent/statusChanged" && + payload.status?.type !== "systemError")) + ); + }); assert.notInclude( startedThreadIds, @@ -293,6 +321,11 @@ describe("CodexSessionRuntime collab integration", () => { ROOT, "the terminal metadata patch must preserve parent linkage from thread registration", ); + assert.deepEqual( + childATerminalOverrides.map((event) => event.method), + [], + "trailing lifecycle must not overwrite a terminal child error", + ); assert.include( startedThreadIds, CHILD_B, diff --git a/apps/server/src/provider/Layers/CodexSessionRuntime.ts b/apps/server/src/provider/Layers/CodexSessionRuntime.ts index ddb2efb9faff..3240ace565fa 100644 --- a/apps/server/src/provider/Layers/CodexSessionRuntime.ts +++ b/apps/server/src/provider/Layers/CodexSessionRuntime.ts @@ -1296,6 +1296,9 @@ export const makeCodexSessionRuntime = ( next.delete(child.agentThreadId); return next; }); + if (child.terminalError) { + return true; + } yield* emitEvent({ kind: "notification", threadId: options.threadId, @@ -1308,6 +1311,9 @@ export const makeCodexSessionRuntime = ( }); return true; case "thread/status/changed": + if (child.terminalError) { + return true; + } yield* emitEvent({ kind: "notification", threadId: options.threadId, @@ -1353,6 +1359,9 @@ export const makeCodexSessionRuntime = ( next.delete(child.agentThreadId); return next; }); + if (child.terminalError) { + return true; + } yield* emitEvent({ kind: "notification", threadId: options.threadId, From b8bf98c1ae20adee6f0b3d1d839148e9605fc95d Mon Sep 17 00:00:00 2001 From: Michael Johnston Date: Sun, 23 Aug 2026 17:33:52 -0700 Subject: [PATCH 08/10] fix(codex): enrich late terminal identity --- .../CodexCollabRuntime.integration.test.ts | 70 +++++++++++++++++++ .../provider/Layers/CodexSessionRuntime.ts | 21 ++++-- 2 files changed, 85 insertions(+), 6 deletions(-) diff --git a/apps/server/src/provider/Layers/CodexCollabRuntime.integration.test.ts b/apps/server/src/provider/Layers/CodexCollabRuntime.integration.test.ts index c990d79d706a..b89c649ad566 100644 --- a/apps/server/src/provider/Layers/CodexCollabRuntime.integration.test.ts +++ b/apps/server/src/provider/Layers/CodexCollabRuntime.integration.test.ts @@ -177,6 +177,7 @@ describe("CodexSessionRuntime collab integration", () => { assert.isDefined(rootThreadStarted); const turnIdA = (turnStartedA.params as { turn: { id: string } }).turn.id; const turnIdB = (turnStartedB.params as { turn: { id: string } }).turn.id; + const childC = "child-terminal-thread-first"; const threadRegistrationA = { ...rootThreadStarted, params: { @@ -198,6 +199,44 @@ describe("CodexSessionRuntime collab integration", () => { }, }, }; + const turnStartedC = { + ...turnStartedA, + params: { + ...turnStartedA.params, + threadId: childC, + turn: { ...turnStartedA.params.turn, id: `${childC}-turn` }, + }, + }; + const threadRegistrationC = { + ...threadRegistrationA, + params: { + thread: { + ...threadRegistrationA.params.thread, + id: childC, + sessionId: childC, + source: { + subAgent: { + thread_spawn: { + agent_nickname: "gamma", + depth: 1, + parent_thread_id: ROOT, + }, + }, + }, + }, + }, + }; + const registrationC = { + ...registrationA, + params: { + ...registrationA.params, + item: { + ...registrationA.params.item, + agentThreadId: childC, + agentPath: "/root/gamma", + }, + }, + }; const script = { rootThreadId: ROOT, @@ -226,6 +265,18 @@ describe("CodexSessionRuntime collab integration", () => { }, }, { method: "thread/closed", params: { threadId: CHILD_A } }, + turnStartedC, + { + method: "error", + params: { + threadId: childC, + turnId: `${childC}-turn`, + error: { message: "thread-first child failed before registration" }, + willRetry: false, + }, + }, + threadRegistrationC, + registrationC, turnStartedB, { method: "error", @@ -298,6 +349,13 @@ describe("CodexSessionRuntime collab integration", () => { payload.status?.type !== "systemError")) ); }); + const childCFailures = events.filter( + (event) => + event.method === "collabAgent/statusChanged" && + (event.payload as { agentThreadId?: string; status?: { type?: string } }) + .agentThreadId === childC && + (event.payload as { status?: { type?: string } }).status?.type === "systemError", + ); assert.notInclude( startedThreadIds, @@ -326,6 +384,18 @@ describe("CodexSessionRuntime collab integration", () => { [], "trailing lifecycle must not overwrite a terminal child error", ); + assert.lengthOf( + childCFailures, + 2, + "late activity identity must enrich a thread-first terminal child", + ); + const childCMetadataFailure = childCFailures.at(-1); + assert.isDefined(childCMetadataFailure); + assert.equal( + (childCMetadataFailure.payload as { agentPath?: string }).agentPath, + "/root/gamma", + "the terminal metadata patch must preserve a path learned from late activity", + ); assert.include( startedThreadIds, CHILD_B, diff --git a/apps/server/src/provider/Layers/CodexSessionRuntime.ts b/apps/server/src/provider/Layers/CodexSessionRuntime.ts index 3240ace565fa..c55088c1388c 100644 --- a/apps/server/src/provider/Layers/CodexSessionRuntime.ts +++ b/apps/server/src/provider/Layers/CodexSessionRuntime.ts @@ -687,6 +687,18 @@ interface CollabChildAgentState { readonly spawnTurnId: TurnId | undefined; } +function collabChildIdentityChanged( + before: CollabChildAgentState, + after: CollabChildAgentState, +): boolean { + return ( + before.nickname !== after.nickname || + before.role !== after.role || + before.agentPath !== after.agentPath || + before.parentThreadId !== after.parentThreadId + ); +} + function readThreadSpawnSource(thread: { readonly source: unknown }): | { nickname: string | undefined; @@ -1118,12 +1130,7 @@ export const makeCodexSessionRuntime = ( if (!existingChild) { yield* emitCollabChildStarted(state); yield* emitCollabChildSystemError(state); - } else if ( - existingChild.nickname !== state.nickname || - existingChild.role !== state.role || - existingChild.agentPath !== state.agentPath || - existingChild.parentThreadId !== state.parentThreadId - ) { + } else if (collabChildIdentityChanged(existingChild, state)) { // Keep the terminal status authoritative while propagating // identity that arrived after the first registration path. yield* emitCollabChildSystemError(state); @@ -1204,6 +1211,8 @@ export const makeCodexSessionRuntime = ( if (!existingChild) { yield* emitCollabChildStarted(registeredChild); yield* emitCollabChildSystemError(registeredChild); + } else if (collabChildIdentityChanged(existingChild, registeredChild)) { + yield* emitCollabChildSystemError(registeredChild); } } else { yield* emitEvent({ From e494716b9367fb248fe47cab7a371256b3c97e5b Mon Sep 17 00:00:00 2001 From: Michael Johnston Date: Sun, 23 Aug 2026 17:46:14 -0700 Subject: [PATCH 09/10] fix(codex): keep root errors visible --- .../CodexCollabRuntime.integration.test.ts | 12 ++++++++- .../provider/Layers/CodexSessionRuntime.ts | 27 +++++++++---------- 2 files changed, 24 insertions(+), 15 deletions(-) diff --git a/apps/server/src/provider/Layers/CodexCollabRuntime.integration.test.ts b/apps/server/src/provider/Layers/CodexCollabRuntime.integration.test.ts index b89c649ad566..0d04a46c75e0 100644 --- a/apps/server/src/provider/Layers/CodexCollabRuntime.integration.test.ts +++ b/apps/server/src/provider/Layers/CodexCollabRuntime.integration.test.ts @@ -46,10 +46,19 @@ function buildScript() { tool: "wait", status: "completed", senderThreadId: ROOT, - receiverThreadIds: [CHILD_A, CHILD_B], + receiverThreadIds: [ROOT, CHILD_A, CHILD_B], }, }, }, + { + method: "error", + params: { + threadId: ROOT, + turnId: "root-turn-error", + error: { message: "root error must stay visible" }, + willRetry: false, + }, + }, // Child terminal lifecycle AFTER the receiver map knows the children — // pre-fix, the legacy suppressor dropped these before interception saw // them, so no synthetic agent events were emitted. @@ -109,6 +118,7 @@ describe("CodexSessionRuntime collab integration", () => { assert.include(methods, "collabAgent/activity"); assert.include(methods, "collabAgent/turnCompleted"); assert.include(methods, "collabAgent/closed"); + assert.include(methods, "error", "receiver bookkeeping must not suppress a root error"); const childTurnCompleted = events.find( (event) => diff --git a/apps/server/src/provider/Layers/CodexSessionRuntime.ts b/apps/server/src/provider/Layers/CodexSessionRuntime.ts index c55088c1388c..515d5f3fb923 100644 --- a/apps/server/src/provider/Layers/CodexSessionRuntime.ts +++ b/apps/server/src/provider/Layers/CodexSessionRuntime.ts @@ -1422,12 +1422,10 @@ export const makeCodexSessionRuntime = ( const payload = notification.params; const route = readRouteFields(notification); const collabReceiverTurns = yield* Ref.get(collabReceiverTurnsRef); - const childParentTurnId = (() => { - const providerConversationId = readNotificationThreadId(notification); - return providerConversationId - ? collabReceiverTurns.get(providerConversationId) - : undefined; - })(); + const notificationConversationId = readNotificationThreadId(notification); + const childParentTurnId = notificationConversationId + ? collabReceiverTurns.get(notificationConversationId) + : undefined; rememberCollabReceiverTurns(collabReceiverTurns, notification, route.turnId); // Interception FIRST: a registered v2 child is usually also in the @@ -1448,15 +1446,16 @@ export const makeCodexSessionRuntime = ( // thread/* onto parent session state. Root-id-known guard keeps the // root's own early notifications flowing during session open. const suppressRootId = currentProviderThreadId(yield* Ref.get(sessionRef)); - const foreignConversation = (() => { - const providerConversationId = readNotificationThreadId(notification); - return ( - providerConversationId !== undefined && - suppressRootId !== undefined && - providerConversationId !== suppressRootId - ); - })(); + const rootConversation = + notificationConversationId !== undefined && + suppressRootId !== undefined && + notificationConversationId === suppressRootId; + const foreignConversation = + notificationConversationId !== undefined && + suppressRootId !== undefined && + notificationConversationId !== suppressRootId; if ( + !rootConversation && (childParentTurnId !== undefined || foreignConversation) && shouldSuppressChildConversationNotification(notification.method) ) { From 1123f75e9f80875ed63c88cf2e234669961d4c84 Mon Sep 17 00:00:00 2001 From: Michael Johnston Date: Sun, 23 Aug 2026 18:08:25 -0700 Subject: [PATCH 10/10] test(codex): exercise root receiver routing --- .../Layers/CodexCollabRuntime.integration.test.ts | 11 ++++++++++- 1 file changed, 10 insertions(+), 1 deletion(-) diff --git a/apps/server/src/provider/Layers/CodexCollabRuntime.integration.test.ts b/apps/server/src/provider/Layers/CodexCollabRuntime.integration.test.ts index 0d04a46c75e0..a5647bc8f7d4 100644 --- a/apps/server/src/provider/Layers/CodexCollabRuntime.integration.test.ts +++ b/apps/server/src/provider/Layers/CodexCollabRuntime.integration.test.ts @@ -40,6 +40,8 @@ function buildScript() { method: "item/completed", params: { threadId: ROOT, + turnId: "root-turn-error", + completedAtMs: 0, item: { type: "collabAgentToolCall", id: "call_fixture_wait", @@ -47,6 +49,7 @@ function buildScript() { status: "completed", senderThreadId: ROOT, receiverThreadIds: [ROOT, CHILD_A, CHILD_B], + agentsStates: {}, }, }, }, @@ -118,7 +121,13 @@ describe("CodexSessionRuntime collab integration", () => { assert.include(methods, "collabAgent/activity"); assert.include(methods, "collabAgent/turnCompleted"); assert.include(methods, "collabAgent/closed"); - assert.include(methods, "error", "receiver bookkeeping must not suppress a root error"); + const rootError = events.find( + (event) => + event.method === "error" && + (event.payload as { error?: { message?: string } }).error?.message === + "root error must stay visible", + ); + assert.isDefined(rootError, "receiver bookkeeping must not suppress a root error"); const childTurnCompleted = events.find( (event) =>