diff --git a/apps/server/src/provider/Layers/CodexCollabRuntime.integration.test.ts b/apps/server/src/provider/Layers/CodexCollabRuntime.integration.test.ts index 3a02c45b23ff..53f130447afd 100644 --- a/apps/server/src/provider/Layers/CodexCollabRuntime.integration.test.ts +++ b/apps/server/src/provider/Layers/CodexCollabRuntime.integration.test.ts @@ -13,7 +13,7 @@ import * as NodePath from "node:path"; import * as NodeServices from "@effect/platform-node/NodeServices"; import { it } from "@effect/vitest"; -import { ThreadId } from "@t3tools/contracts"; +import { type ProviderEvent, ThreadId } from "@t3tools/contracts"; import * as Effect from "effect/Effect"; import * as Fiber from "effect/Fiber"; import * as Stream from "effect/Stream"; @@ -73,6 +73,125 @@ function buildScript() { const scriptPath = NodePath.join(import.meta.dirname, "../testFixtures/.collab-script.json"); const peerPath = NodePath.join(import.meta.dirname, "../testFixtures/codexCollabMockPeer.sh"); +interface MockCollabNotification { + readonly method: string; + readonly params: unknown; +} + +function mockChildActivity( + childThreadId: string, + itemId: string, + kind: "started" | "interacted" | "interrupted" = "started", +) { + return { + method: "item/completed", + params: { + threadId: ROOT, + turnId: "00000000-0000-4000-8000-000000000100", + completedAtMs: 1_700_000_000_000, + item: { + type: "subAgentActivity", + id: itemId, + kind, + agentThreadId: childThreadId, + agentPath: "/root/mock_worker", + }, + }, + }; +} + +function mockChildCompletion( + childThreadId: string, + turnId: string, + status: "completed" | "failed" | "interrupted" = "completed", +) { + return { + method: "turn/completed", + params: { + threadId: childThreadId, + turn: { id: turnId, status, items: [] }, + }, + }; +} + +function mockChildThreadRegistration(childThreadId: string) { + return { + method: "thread/started", + params: { + thread: { + agentNickname: "mock_worker", + agentRole: "worker", + cliVersion: "0.0.0-test", + createdAt: 1_700_000_000, + cwd: "/workspace/mock", + ephemeral: false, + forkedFromId: null, + gitInfo: null, + id: childThreadId, + modelProvider: "mock", + name: null, + parentThreadId: ROOT, + path: "/tmp/mock-rollout.jsonl", + preview: "", + recencyAt: 1_700_000_000, + sessionId: ROOT, + source: { + subAgent: { + thread_spawn: { + agent_nickname: "mock_worker", + agent_path: "/root/mock_worker", + agent_role: "worker", + depth: 1, + parent_thread_id: ROOT, + }, + }, + }, + status: { type: "idle" }, + threadSource: null, + turns: [], + updatedAt: 1_700_000_000, + }, + }, + }; +} + +function eventsForChild(events: ReadonlyArray, childThreadId: string) { + return events.filter( + (event) => + (event.payload as { agentThreadId?: string } | undefined)?.agentThreadId === childThreadId, + ); +} + +const runMockCollabScript = Effect.fn("CodexCollabRuntimeTest.runMockCollabScript")(function* ( + runtimeThreadId: string, + input: string, + notifications: ReadonlyArray, +) { + const script = { rootThreadId: ROOT, notifications }; + // @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(runtimeThreadId), + 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 }); + const events = Array.from(yield* Fiber.join(eventsFiber)); + yield* runtime.close; + return events; +}); + describe("CodexSessionRuntime collab integration", () => { it.effect("replays the captured fan-out into synthetic agent events without child leaks", () => Effect.gen(function* () { @@ -123,6 +242,15 @@ describe("CodexSessionRuntime collab integration", () => { ); assert.isDefined(childClosed, "child B's close becomes an agent event"); + const childAStatuses = eventsForChild(events, CHILD_A) + .filter((event) => event.method === "collabAgent/statusChanged") + .map((event) => (event.payload as { status?: { type?: string } }).status?.type); + assert.deepEqual( + childAStatuses, + ["active", "idle"], + "the child's initial pre-registration idle is not a completed run", + ); + // Parent-owned resolution passes through — not swallowed, not // re-labelled as an agent event. assert.include(methods, "serverRequest/resolved"); @@ -147,6 +275,244 @@ describe("CodexSessionRuntime collab integration", () => { }).pipe(Effect.scoped, Effect.provide(NodeServices.layer)), ); + it.effect("replays child completion received before registration", () => + Effect.gen(function* () { + const childThreadId = "00000000-0000-4000-8000-000000000101"; + const events = yield* runMockCollabScript( + "thread-collab-preregistration", + "finish before registration", + [ + mockChildCompletion(childThreadId, "00000000-0000-4000-8000-000000000102"), + mockChildActivity(childThreadId, "call_fixture_preregistration"), + ], + ); + const childMethods = eventsForChild(events, childThreadId).map((event) => event.method); + assert.deepEqual(childMethods, ["collabAgent/activity", "collabAgent/turnCompleted"]); + }).pipe(Effect.scoped, Effect.provide(NodeServices.layer)), + ); + + it.effect("replays a terminal child error received before registration", () => + Effect.gen(function* () { + const childThreadId = "00000000-0000-4000-8000-000000000801"; + const childTurnId = "00000000-0000-4000-8000-000000000802"; + const events = yield* runMockCollabScript( + "thread-collab-preregistration-error", + "fail before registration", + [ + { + method: "turn/started", + params: { + threadId: childThreadId, + turn: { id: childTurnId, status: "inProgress", items: [] }, + }, + }, + { + method: "error", + params: { + threadId: childThreadId, + turnId: childTurnId, + error: { message: "Synthetic terminal child error" }, + willRetry: false, + }, + }, + mockChildActivity(childThreadId, "call_fixture_preregistration_error"), + ], + ); + const childEvents = eventsForChild(events, childThreadId); + assert.deepEqual( + childEvents.map((event) => event.method), + ["collabAgent/activity", "collabAgent/statusChanged"], + ); + assert.equal( + (childEvents.at(-1)?.payload as { status?: { type?: string } } | undefined)?.status?.type, + "systemError", + ); + }).pipe(Effect.scoped, Effect.provide(NodeServices.layer)), + ); + + it.effect("replays settlement after duplicate child registration signals", () => + Effect.gen(function* () { + const childThreadId = "00000000-0000-4000-8000-000000000401"; + const events = yield* runMockCollabScript( + "thread-collab-preregistration-duplicate", + "register a completed child twice", + [ + mockChildCompletion(childThreadId, "00000000-0000-4000-8000-000000000402"), + mockChildThreadRegistration(childThreadId), + mockChildActivity(childThreadId, "call_fixture_preregistration_duplicate"), + ], + ); + assert.deepEqual( + eventsForChild(events, childThreadId).map((event) => event.method), + [ + "collabAgent/started", + "collabAgent/turnCompleted", + "collabAgent/activity", + "collabAgent/turnCompleted", + ], + ); + }).pipe(Effect.scoped, Effect.provide(NodeServices.layer)), + ); + + it.effect("does not replay an old settlement after a registered child restarts", () => + Effect.gen(function* () { + const childThreadId = "00000000-0000-4000-8000-000000000501"; + const events = yield* runMockCollabScript( + "thread-collab-preregistration-restarted", + "restart between duplicate registrations", + [ + mockChildCompletion(childThreadId, "00000000-0000-4000-8000-000000000502"), + mockChildThreadRegistration(childThreadId), + { + method: "turn/started", + params: { + threadId: childThreadId, + turn: { + id: "00000000-0000-4000-8000-000000000503", + status: "inProgress", + items: [], + }, + }, + }, + mockChildActivity(childThreadId, "call_fixture_preregistration_restarted"), + ], + ); + assert.deepEqual( + eventsForChild(events, childThreadId).map((event) => event.method), + [ + "collabAgent/started", + "collabAgent/turnCompleted", + "collabAgent/turnStarted", + "collabAgent/activity", + ], + ); + }).pipe(Effect.scoped, Effect.provide(NodeServices.layer)), + ); + + it.effect("keeps child interaction liveness-neutral across later registration", () => + Effect.gen(function* () { + const childThreadId = "00000000-0000-4000-8000-000000000601"; + const events = yield* runMockCollabScript( + "thread-collab-preregistration-interacted", + "interact after completion", + [ + mockChildThreadRegistration(childThreadId), + mockChildCompletion(childThreadId, "00000000-0000-4000-8000-000000000602"), + mockChildActivity(childThreadId, "call_fixture_preregistration_interacted", "interacted"), + mockChildThreadRegistration(childThreadId), + ], + ); + assert.deepEqual( + eventsForChild(events, childThreadId).map((event) => event.method), + [ + "collabAgent/started", + "collabAgent/turnCompleted", + "collabAgent/activity", + "collabAgent/started", + "collabAgent/turnCompleted", + ], + ); + }).pipe(Effect.scoped, Effect.provide(NodeServices.layer)), + ); + + it.effect("preserves a newer child interruption across later registration", () => + Effect.gen(function* () { + const childThreadId = "00000000-0000-4000-8000-000000000701"; + const events = yield* runMockCollabScript( + "thread-collab-preregistration-interrupted", + "interrupt before duplicate registration", + [ + mockChildThreadRegistration(childThreadId), + mockChildCompletion(childThreadId, "00000000-0000-4000-8000-000000000702"), + mockChildActivity( + childThreadId, + "call_fixture_preregistration_interrupted", + "interrupted", + ), + mockChildThreadRegistration(childThreadId), + ], + ); + const childEvents = eventsForChild(events, childThreadId); + assert.deepEqual( + childEvents.map((event) => event.method), + [ + "collabAgent/started", + "collabAgent/turnCompleted", + "collabAgent/activity", + "collabAgent/started", + "collabAgent/turnCompleted", + ], + ); + const replayedTurn = childEvents.at(-1)?.payload as + | { turn?: { status?: string } } + | undefined; + assert.equal(replayedTurn?.turn?.status, "interrupted"); + }).pipe(Effect.scoped, Effect.provide(NodeServices.layer)), + ); + + it.effect("replays child idle received before registration after running", () => + Effect.gen(function* () { + const childThreadId = "00000000-0000-4000-8000-000000000301"; + const events = yield* runMockCollabScript( + "thread-collab-preregistration-idle", + "become idle before registration", + [ + { + method: "thread/status/changed", + params: { + threadId: childThreadId, + status: { type: "active", activeFlags: [] }, + }, + }, + { + method: "thread/status/changed", + params: { + threadId: childThreadId, + status: { type: "idle" }, + }, + }, + mockChildActivity(childThreadId, "call_fixture_preregistration_idle"), + ], + ); + const childEvents = eventsForChild(events, childThreadId); + assert.deepEqual( + childEvents.map((event) => event.method), + ["collabAgent/activity", "collabAgent/statusChanged"], + ); + assert.equal( + (childEvents[1]?.payload as { status?: { type?: string } } | undefined)?.status?.type, + "idle", + ); + }).pipe(Effect.scoped, Effect.provide(NodeServices.layer)), + ); + + it.effect("drops buffered completion superseded by a newer child turn", () => + Effect.gen(function* () { + const childThreadId = "00000000-0000-4000-8000-000000000201"; + const events = yield* runMockCollabScript( + "thread-collab-preregistration-order", + "restart before registration", + [ + mockChildCompletion(childThreadId, "00000000-0000-4000-8000-000000000202"), + { + method: "turn/started", + params: { + threadId: childThreadId, + turn: { + id: "00000000-0000-4000-8000-000000000203", + status: "inProgress", + items: [], + }, + }, + }, + mockChildActivity(childThreadId, "call_fixture_preregistration_order"), + ], + ); + const childMethods = eventsForChild(events, childThreadId).map((event) => event.method); + assert.deepEqual(childMethods, ["collabAgent/activity"]); + }).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 57a1162dd08e..170002de19f9 100644 --- a/apps/server/src/provider/Layers/CodexSessionRuntime.ts +++ b/apps/server/src/provider/Layers/CodexSessionRuntime.ts @@ -613,9 +613,10 @@ function readRouteFields(notification: CodexServerNotification): { * WIP, probe-gated: registration is deliberately explicit-signals-only. The * spec's "provisionally treat unknown foreign thread ids as v2 children" rule * needs a live wire capture of the packaged binary before it lands — blind - * capture risks eating unrelated traffic. Until then a child whose first - * notification precedes registration passes through as today (no regression - * vs main, which passes everything through). + * capture risks eating unrelated traffic. Until then foreign child lifecycle + * stays out of the parent timeline. The runtime retains only its latest + * meaningful liveness state and replays a settlement after explicit + * registration; traffic alone never creates an agent identity. */ interface CollabChildAgentState { readonly agentThreadId: string; @@ -633,6 +634,44 @@ interface CollabChildAgentState { readonly spawnTurnId: TurnId | undefined; } +function toCollabChildSettlement(notification: CodexServerNotification) { + switch (notification.method) { + case "turn/completed": + return { + method: "collabAgent/turnCompleted", + // The adapter only needs the terminal status to reconstruct task + // liveness. Do not retain a completed turn's potentially large item + // history for the rest of the session. + payload: { turn: { status: notification.params.turn.status } }, + } as const; + case "thread/status/changed": { + const statusType = notification.params.status.type; + if (statusType !== "idle" && statusType !== "systemError") { + return undefined; + } + return { + method: "collabAgent/statusChanged", + payload: { status: { type: statusType } }, + } as const; + } + case "thread/closed": + return { method: "collabAgent/closed", payload: {} } as const; + case "error": + if (notification.params.willRetry) { + return undefined; + } + return { + method: "collabAgent/statusChanged", + payload: { status: { type: "systemError" } }, + } as const; + default: + return undefined; + } +} + +type CollabChildSettlement = NonNullable>; +type CollabChildLifecycleState = CollabChildSettlement | "active"; + function readThreadSpawnSource(thread: { readonly source: unknown }): | { nickname: string | undefined; @@ -857,6 +896,9 @@ 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()); + const collabChildLifecycleStatesRef = yield* Ref.make( + new Map(), + ); const closedRef = yield* Ref.make(false); // `~` is not shell-expanded when env vars are set via @@ -949,6 +991,44 @@ export const makeCodexSessionRuntime = ( message, }); + const setCollabChildLifecycleState = Effect.fn( + "CodexSessionRuntime.setCollabChildLifecycleState", + )(function* (agentThreadId: string, state: CollabChildLifecycleState) { + yield* Ref.update(collabChildLifecycleStatesRef, (current) => { + const next = new Map(current); + next.set(agentThreadId, state); + return next; + }); + }); + + const emitCollabChildSettlementAfterRegistration = Effect.fn( + "CodexSessionRuntime.emitCollabChildSettlementAfterRegistration", + )(function* (child: CollabChildAgentState) { + const lifecycleState = (yield* Ref.get(collabChildLifecycleStatesRef)).get( + child.agentThreadId, + ); + // Registration itself represents a still-active child. Only a newer + // settlement needs a second synthetic event. Keep that settlement as + // the latest known state because Codex can emit more than one explicit + // registration signal for the same child. + if (!lifecycleState || lifecycleState === "active") { + return; + } + yield* emitEvent({ + kind: "notification", + threadId: options.threadId, + ...(child.spawnTurnId ? { turnId: child.spawnTurnId } : {}), + method: lifecycleState.method, + payload: { + agentThreadId: child.agentThreadId, + ...(child.nickname ? { nickname: child.nickname } : {}), + ...(child.role ? { role: child.role } : {}), + ...(child.agentPath ? { agentPath: child.agentPath } : {}), + ...lifecycleState.payload, + }, + }); + }); + const settlePendingApprovals = (decision: ProviderApprovalDecision) => Ref.get(pendingApprovalsRef).pipe( Effect.flatMap((pendingApprovals) => @@ -1029,6 +1109,7 @@ export const makeCodexSessionRuntime = ( ...(state.parentThreadId ? { parentThreadId: state.parentThreadId } : {}), }, }); + yield* emitCollabChildSettlementAfterRegistration(state); return true; } @@ -1089,6 +1170,20 @@ export const makeCodexSessionRuntime = ( activityKind: item.kind, }, }); + if (registeredChild && item.kind === "started") { + // A started activity is an explicit registration signal and can + // be the second half of thread/started registration. Reapply the + // latest settlement so that duplicate starts cannot resurrect a + // child whose lifecycle arrived early. Later activity events are + // not registration and must not be overwritten by an older + // settlement. + yield* emitCollabChildSettlementAfterRegistration(registeredChild); + } else if (registeredChild && item.kind === "interrupted") { + yield* setCollabChildLifecycleState(registeredChild.agentThreadId, { + method: "collabAgent/turnCompleted", + payload: { turn: { status: "interrupted" } }, + }); + } return true; } @@ -1117,6 +1212,7 @@ export const makeCodexSessionRuntime = ( }; switch (notification.method) { case "turn/started": { + yield* setCollabChildLifecycleState(child.agentThreadId, "active"); const childTurnId = typeof (notification.params as { turn?: { id?: unknown } }).turn?.id === "string" ? ((notification.params as { turn: { id: string } }).turn.id as string) @@ -1137,7 +1233,11 @@ export const makeCodexSessionRuntime = ( }); return true; } - case "turn/completed": + case "turn/completed": { + const settlement = toCollabChildSettlement(notification); + if (settlement) { + yield* setCollabChildLifecycleState(child.agentThreadId, settlement); + } yield* Ref.update(collabChildLiveTurnsRef, (current) => { const next = new Map(current); next.delete(child.agentThreadId); @@ -1154,7 +1254,15 @@ export const makeCodexSessionRuntime = ( }, }); return true; - case "thread/status/changed": + } + case "thread/status/changed": { + const lifecycleState = + notification.params.status.type === "active" + ? "active" + : toCollabChildSettlement(notification); + if (lifecycleState) { + yield* setCollabChildLifecycleState(child.agentThreadId, lifecycleState); + } yield* emitEvent({ kind: "notification", threadId: options.threadId, @@ -1166,6 +1274,7 @@ export const makeCodexSessionRuntime = ( }, }); return true; + } case "thread/tokenUsage/updated": yield* emitEvent({ kind: "notification", @@ -1191,7 +1300,11 @@ export const makeCodexSessionRuntime = ( }, }); return true; - case "thread/closed": + case "thread/closed": { + const settlement = toCollabChildSettlement(notification); + if (settlement) { + yield* setCollabChildLifecycleState(child.agentThreadId, settlement); + } // The child is gone: drop its live-turn entry so a later Stop // doesn't waste a turn/interrupt RPC on a closed thread before // reaching the parent (review finding). @@ -1208,6 +1321,7 @@ export const makeCodexSessionRuntime = ( payload: childIdentity, }); return true; + } case "error": { // A child error must surface as a failed agent, not vanish into // the default swallow (review finding: the child stayed @@ -1219,8 +1333,13 @@ export const makeCodexSessionRuntime = ( // path. const willRetry = (notification.params as { willRetry?: boolean }).willRetry === true; if (willRetry) { + yield* setCollabChildLifecycleState(child.agentThreadId, "active"); return true; } + const settlement = toCollabChildSettlement(notification); + if (settlement) { + yield* setCollabChildLifecycleState(child.agentThreadId, settlement); + } yield* Ref.update(collabChildLiveTurnsRef, (current) => { const next = new Map(current); next.delete(child.agentThreadId); @@ -1286,9 +1405,11 @@ export const makeCodexSessionRuntime = ( providerConversationId !== suppressRootId ); })(); + const isTerminalChildError = + notification.method === "error" && notification.params.willRetry === false; if ( (childParentTurnId !== undefined || foreignConversation) && - shouldSuppressChildConversationNotification(notification.method) + (shouldSuppressChildConversationNotification(notification.method) || isTerminalChildError) ) { // Stop-everything must not depend on registration timing: a // child's turn/started can arrive before the subAgentActivity that @@ -1299,6 +1420,33 @@ export const makeCodexSessionRuntime = ( // false-positive entry costs one ignored RPC at worst. const foreignThreadId = readNotificationThreadId(notification); if (foreignThreadId !== undefined) { + const isActiveStatus = + notification.method === "thread/status/changed" && + notification.params.status.type === "active"; + if (notification.method === "turn/started" || isActiveStatus) { + yield* setCollabChildLifecycleState(foreignThreadId, "active"); + } else { + const pendingSettlement = toCollabChildSettlement(notification); + if (pendingSettlement) { + const liveChildTurns = yield* Ref.get(collabChildLiveTurnsRef); + const isIdleStatus = + notification.method === "thread/status/changed" && + notification.params.status.type === "idle"; + yield* Ref.update(collabChildLifecycleStatesRef, (current) => { + const existing = current.get(foreignThreadId); + if ( + isIdleStatus && + ((existing !== undefined && existing !== "active") || + (existing === undefined && !liveChildTurns.has(foreignThreadId))) + ) { + return current; + } + const next = new Map(current); + next.set(foreignThreadId, pendingSettlement); + return next; + }); + } + } if (notification.method === "turn/started") { const foreignTurnId = typeof (notification.params as { turn?: { id?: unknown } }).turn?.id === "string" @@ -1313,7 +1461,8 @@ export const makeCodexSessionRuntime = ( } } else if ( notification.method === "turn/completed" || - notification.method === "thread/closed" + notification.method === "thread/closed" || + isTerminalChildError ) { yield* Ref.update(collabChildLiveTurnsRef, (current) => { const next = new Map(current);