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 a1b46e003520..a5647bc8f7d4 100644 --- a/apps/server/src/provider/Layers/CodexCollabRuntime.integration.test.ts +++ b/apps/server/src/provider/Layers/CodexCollabRuntime.integration.test.ts @@ -40,16 +40,28 @@ function buildScript() { method: "item/completed", params: { threadId: ROOT, + turnId: "root-turn-error", + completedAtMs: 0, item: { type: "collabAgentToolCall", id: "call_fixture_wait", tool: "wait", status: "completed", senderThreadId: ROOT, - receiverThreadIds: [CHILD_A, CHILD_B], + receiverThreadIds: [ROOT, CHILD_A, CHILD_B], + agentsStates: {}, }, }, }, + { + 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 +121,13 @@ describe("CodexSessionRuntime collab integration", () => { assert.include(methods, "collabAgent/activity"); assert.include(methods, "collabAgent/turnCompleted"); assert.include(methods, "collabAgent/closed"); + 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) => @@ -148,6 +167,269 @@ 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 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; + }); + assert.isDefined(turnStartedA); + 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 childC = "child-terminal-thread-first"; + 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 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, + notifications: [ + turnStartedA, + { + method: "error", + params: { + threadId: CHILD_A, + turnId: turnIdA, + error: { message: "child failed before registration" }, + willRetry: false, + }, + }, + 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 } }, + turnStartedC, + { + method: "error", + params: { + threadId: childC, + turnId: `${childC}-turn`, + error: { message: "thread-first child failed before registration" }, + willRetry: false, + }, + }, + threadRegistrationC, + registrationC, + 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); + const childARegistrationEvents = events.filter( + (event) => + (event.method === "collabAgent/started" || event.method === "collabAgent/activity") && + (event.payload as { agentThreadId?: string }).agentThreadId === CHILD_A, + ); + const childAFailures = events.filter( + (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", + ); + 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")) + ); + }); + 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, + CHILD_A, + "a terminal child turn must not replay as live when activity registers it later", + ); + assert.deepEqual( + childARegistrationEvents.map((event) => event.method), + ["collabAgent/started"], + "a failed child needs one start anchor, but later registration must not duplicate it", + ); + assert.lengthOf( + childAFailures, + 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.deepEqual( + childATerminalOverrides.map((event) => event.method), + [], + "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, + "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", () => @@ -181,6 +463,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 +496,7 @@ describe("CodexSessionRuntime collab integration", () => { hangInterruptFor: CHILD_A, notifications: [ turnStartedA, - registrationA, + interactedRegistrationA, memoryThreadStarted, memoryTurnStarted, registrationB, @@ -233,26 +522,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..515d5f3fb923 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 @@ -685,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; @@ -753,6 +767,7 @@ function shouldSuppressChildConversationNotification( method === "thread/tokenUsage/updated" || method === "turn/started" || method === "turn/completed" || + method === "error" || method === "turn/plan/updated" || method === "item/plan/delta" ); @@ -909,6 +924,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); @@ -1001,6 +1018,36 @@ 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", + 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 } : {}), + ...(child.parentThreadId ? { parentThreadId: child.parentThreadId } : {}), + status: { type: "systemError" }, + }, + }); const settlePendingApprovals = (decision: ProviderApprovalDecision) => Ref.get(pendingApprovalsRef).pipe( @@ -1050,6 +1097,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); @@ -1062,26 +1112,32 @@ 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* Ref.update(collabChildPreRegistrationFailuresRef, (current) => { + const next = new Set(current); + next.delete(thread.id); + return next; + }); + } + if (state.terminalError) { + if (!existingChild) { + yield* emitCollabChildStarted(state); + yield* emitCollabChildSystemError(state); + } else if (collabChildIdentityChanged(existingChild, state)) { + // Keep the terminal status authoritative while propagating + // identity that arrived after the first registration path. + yield* emitCollabChildSystemError(state); + } + } else { + yield* emitCollabChildStarted(state); + } return true; } @@ -1107,8 +1163,11 @@ export const makeCodexSessionRuntime = ( return false; } 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 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,28 +1179,68 @@ 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, + 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 + // 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 (registeredChild?.terminalError) { + if (!existingChild) { + yield* emitCollabChildStarted(registeredChild); + yield* emitCollabChildSystemError(registeredChild); + } else if (collabChildIdentityChanged(existingChild, registeredChild)) { + yield* emitCollabChildSystemError(registeredChild); + } + } else { + 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 (!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; } @@ -1175,11 +1274,21 @@ 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); 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", @@ -1196,6 +1305,9 @@ export const makeCodexSessionRuntime = ( next.delete(child.agentThreadId); return next; }); + if (child.terminalError) { + return true; + } yield* emitEvent({ kind: "notification", threadId: options.threadId, @@ -1208,6 +1320,9 @@ export const makeCodexSessionRuntime = ( }); return true; case "thread/status/changed": + if (child.terminalError) { + return true; + } yield* emitEvent({ kind: "notification", threadId: options.threadId, @@ -1253,6 +1368,9 @@ export const makeCodexSessionRuntime = ( next.delete(child.agentThreadId); return next; }); + if (child.terminalError) { + return true; + } yield* emitEvent({ kind: "notification", threadId: options.threadId, @@ -1279,16 +1397,12 @@ 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* Ref.update(collabChildAgentsRef, (current) => { + const next = new Map(current); + next.set(child.agentThreadId, { ...child, terminalError: true }); + return next; }); + yield* emitCollabChildSystemError(child); return true; } default: @@ -1308,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 @@ -1334,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) ) { @@ -1361,6 +1474,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); @@ -1369,13 +1487,21 @@ 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); 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);