diff --git a/apps/server/src/provider/Layers/CodexAdapter.test.ts b/apps/server/src/provider/Layers/CodexAdapter.test.ts index 7b8fbec5666..637deac19e0 100644 --- a/apps/server/src/provider/Layers/CodexAdapter.test.ts +++ b/apps/server/src/provider/Layers/CodexAdapter.test.ts @@ -555,6 +555,41 @@ lifecycleLayer("CodexAdapterLive lifecycle", (it) => { }), ); + it.effect("renames an agent without overwriting its existing identity metadata", () => + Effect.gen(function* () { + const { adapter, runtime } = yield* startLifecycleRuntime(); + const firstEventFiber = yield* Stream.runHead(adapter.streamEvents).pipe(Effect.forkChild); + + yield* runtime.emit({ + id: asEventId("evt-agent-renamed"), + kind: "notification", + provider: ProviderDriverKind.make("codex"), + createdAt: "2026-01-01T00:00:00.000Z", + method: "collabAgent/renamed", + threadId: asThreadId("thread-1"), + payload: { + agentThreadId: "child-rename", + nickname: "Alpha", + }, + } satisfies ProviderEvent); + + const firstEvent = yield* Fiber.join(firstEventFiber); + NodeAssert.equal(firstEvent._tag, "Some"); + if (firstEvent._tag !== "Some") { + return; + } + NodeAssert.equal(firstEvent.value.type, "task.updated"); + if (firstEvent.value.type !== "task.updated") { + return; + } + NodeAssert.deepStrictEqual(firstEvent.value.payload, { + taskId: "child-rename", + title: "Alpha", + timelineBypass: true, + }); + }), + ); + it.effect("labels MCP lifecycle entries with server and tool names", () => 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 6b99bf52b1e..6efabfd34b3 100644 --- a/apps/server/src/provider/Layers/CodexAdapter.ts +++ b/apps/server/src/provider/Layers/CodexAdapter.ts @@ -605,6 +605,18 @@ function mapCollabAgentEvent( payload: { taskId, status: "running", ...statusLinkage }, }, ]; + case "collabAgent/renamed": + return [ + { + ...base, + type: "task.updated", + payload: { + taskId, + ...(nickname ? { title: nickname } : {}), + timelineBypass: true, + }, + }, + ]; case "collabAgent/turnCompleted": { // Idle, not terminal: the identity is resumable via sendInput/resume. const turn = diff --git a/apps/server/src/provider/Layers/CodexCollabRuntime.integration.test.ts b/apps/server/src/provider/Layers/CodexCollabRuntime.integration.test.ts index 38e0e0a7b2c..42dccaf98a0 100644 --- a/apps/server/src/provider/Layers/CodexCollabRuntime.integration.test.ts +++ b/apps/server/src/provider/Layers/CodexCollabRuntime.integration.test.ts @@ -24,6 +24,8 @@ import { makeCodexSessionRuntime } from "./CodexSessionRuntime.ts"; const ROOT = wireFixture.rootThreadId; const [CHILD_A, CHILD_B] = wireFixture.childThreadIds as [string, string]; +const CHILD_C = "019fcfd6-1806-7de1-8564-de69fd55bff3"; +const CHILD_D = "019fcfd6-1806-7de1-8564-de69fd55bff4"; /** * The captured sequence, extended with the shapes the live capture didn't @@ -70,10 +72,423 @@ function buildScript() { }; } +function buildDirectChildScript() { + const rootTurnId = "019fcfd6-1806-7de1-8564-de69fd55bffb"; + const childTurnId = `${CHILD_A}-direct-turn`; + const childItemId = `${CHILD_A}-direct-message`; + return { + rootThreadId: ROOT, + notifications: [ + { + method: "item/completed", + params: { + threadId: ROOT, + turnId: rootTurnId, + item: { + type: "collabAgentToolCall", + id: "call_direct_spawn", + tool: "spawnAgent", + status: "completed", + senderThreadId: ROOT, + receiverThreadIds: [CHILD_A], + prompt: "Return one concise result.", + agentsStates: { + [CHILD_A]: { status: "pendingInit", message: null }, + }, + }, + completedAtMs: 1785898350000, + }, + }, + { + method: "item/started", + params: { + threadId: CHILD_A, + turnId: childTurnId, + item: { type: "agentMessage", id: childItemId, text: "" }, + }, + }, + { + method: "item/agentMessage/delta", + params: { + threadId: CHILD_A, + turnId: childTurnId, + itemId: childItemId, + delta: "child narration must not enter the parent transcript", + }, + }, + { + method: "item/completed", + params: { + threadId: CHILD_A, + turnId: childTurnId, + completedAtMs: 1785898350000, + item: { + type: "agentMessage", + id: childItemId, + phase: "final_answer", + text: "child result is consumed by the parent model", + }, + }, + }, + { + method: "item/completed", + params: { + threadId: ROOT, + turnId: rootTurnId, + item: { + type: "agentMessage", + id: "root-summary", + phase: "final_answer", + text: "The agent completed:\n- Direct Child Researcher: returned one concise result.", + }, + completedAtMs: 1785898350000, + }, + }, + ], + }; +} + +function buildOutOfOrderNamingScript() { + const rootTurnId = "019fcfd6-1806-7de1-8564-de69fd55bffb"; + return { + rootThreadId: ROOT, + notifications: [ + { + method: "item/completed", + params: { + threadId: ROOT, + turnId: rootTurnId, + item: { + type: "collabAgentToolCall", + id: "call_direct_spawn_ordered", + tool: "spawnAgent", + status: "completed", + senderThreadId: ROOT, + receiverThreadIds: [CHILD_A, CHILD_B], + prompt: "Return one concise result.", + agentsStates: { + [CHILD_A]: { status: "pendingInit", message: null }, + [CHILD_B]: { status: "pendingInit", message: null }, + }, + }, + completedAtMs: 1785898350000, + }, + }, + { + method: "item/completed", + params: { + // A nested spawn reports the child agent's own turn id. It must + // still join the root fleet's naming batch. + threadId: CHILD_A, + turnId: `${CHILD_A}-nested-turn`, + item: { + type: "collabAgentToolCall", + id: "call_nested_spawn_ordered", + tool: "spawnAgent", + status: "completed", + senderThreadId: CHILD_A, + receiverThreadIds: [CHILD_C, CHILD_D], + prompt: "Return one concise result.", + agentsStates: { + [CHILD_C]: { status: "pendingInit", message: null }, + [CHILD_D]: { status: "pendingInit", message: null }, + }, + }, + completedAtMs: 1785898350000, + }, + }, + { + method: "item/completed", + params: { + threadId: ROOT, + turnId: rootTurnId, + item: { + type: "collabAgentToolCall", + id: "call_direct_wait_reordered", + tool: "wait", + status: "completed", + senderThreadId: ROOT, + // Non-spawn calls can list receivers in any order, but they must + // not change the original spawn positions. + receiverThreadIds: [CHILD_D, CHILD_C], + agentsStates: { + [CHILD_C]: { status: "running", message: null }, + [CHILD_D]: { status: "running", message: null }, + }, + }, + completedAtMs: 1785898350000, + }, + }, + { + method: "item/completed", + params: { + threadId: ROOT, + turnId: rootTurnId, + item: { + type: "agentMessage", + id: "root-order-summary", + phase: "final_answer", + text: "Agents:\n- Alpha\n- Beta\n- Gamma\n- Delta", + }, + completedAtMs: 1785898350000, + }, + }, + ], + }; +} + +function buildUnscopedExistingChildScript() { + const rootTurnId = "019fcfd6-1806-7de1-8564-de69fd55bffb"; + return { + rootThreadId: ROOT, + preTurnNotifications: [ + { + // Establish the root identity before replaying early child traffic; + // this models the provider's root thread notification and prevents + // the fixture from racing the runtime's initial session setup. + method: "thread/started", + params: { thread: wireFixture.responses.threadStart.thread }, + }, + { + method: "turn/started", + params: { + threadId: CHILD_B, + turn: { + id: `${CHILD_B}-pre-turn`, + items: [], + itemsView: "notLoaded", + status: "inProgress", + error: null, + startedAt: 1785898342, + completedAt: null, + durationMs: null, + }, + }, + }, + ], + notifications: [ + { + method: "item/completed", + params: { + threadId: ROOT, + turnId: rootTurnId, + item: { + type: "collabAgentToolCall", + id: "call_unscoped_spawn", + tool: "spawnAgent", + status: "completed", + senderThreadId: ROOT, + receiverThreadIds: [CHILD_A, CHILD_B], + prompt: "Return one concise result.", + agentsStates: { + [CHILD_A]: { status: "pendingInit", message: null }, + [CHILD_B]: { status: "pendingInit", message: null }, + }, + }, + completedAtMs: 1785898350000, + }, + }, + { + method: "item/completed", + params: { + threadId: ROOT, + turnId: rootTurnId, + item: { + type: "agentMessage", + id: "root-unscoped-summary", + phase: "final_answer", + text: "Agents:\n- Alpha\n- Beta", + }, + completedAtMs: 1785898350000, + }, + }, + ], + }; +} + +function buildLogicalRootItemScript() { + const rootTurnId = "019fcfd6-1806-7de1-8564-de69fd55bffb"; + const logicalRootId = "019fcfd6-1806-7de1-8564-de69fd55bfff"; + return { + rootThreadId: ROOT, + notifications: [ + { + method: "item/completed", + params: { + // Some provider versions address coordinator items with a logical + // root id that differs from the thread/start response id. + threadId: logicalRootId, + turnId: rootTurnId, + item: { + type: "agentMessage", + id: "logical-root-summary", + phase: "final_answer", + text: "The coordinator kept the parent timeline intact.", + }, + completedAtMs: 1785898350000, + }, + }, + ], + }; +} + const scriptPath = NodePath.join(import.meta.dirname, "../testFixtures/.collab-script.json"); const peerPath = NodePath.join(import.meta.dirname, "../testFixtures/codexCollabMockPeer.sh"); describe("CodexSessionRuntime collab integration", () => { + it.effect("registers receiver ids and keeps child narration out of the parent stream", () => + Effect.gen(function* () { + // @effect-diagnostics-next-line preferSchemaOverJson:off + NodeFS.writeFileSync(scriptPath, JSON.stringify(buildDirectChildScript()), "utf8"); + yield* Effect.addFinalizer(() => + Effect.sync(() => NodeFS.rmSync(scriptPath, { force: true })), + ); + + const runtime = yield* makeCodexSessionRuntime({ + threadId: ThreadId.make("thread-collab-direct-child"), + 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: "direct child" }); + + const events = Array.from(yield* Fiber.join(eventsFiber)); + const methods = events.map((event) => event.method); + assert.include(methods, "collabAgent/started"); + assert.include(methods, "collabAgent/item"); + assert.include(methods, "collabAgent/statusChanged"); + assert.include(methods, "collabAgent/renamed"); + assert.include(methods, "turn/completed"); + assert.notInclude(methods, "item/agentMessage/delta"); + const started = events.find((event) => event.method === "collabAgent/started"); + assert.isUndefined((started?.payload as { nickname?: string } | undefined)?.nickname); + const renamed = events.find((event) => event.method === "collabAgent/renamed"); + assert.equal( + (renamed?.payload as { nickname?: string } | undefined)?.nickname, + "Direct Child Researcher", + ); + const statusChanged = events.find( + (event) => + event.method === "collabAgent/statusChanged" && + (event.payload as { status?: { type?: string } } | undefined)?.status?.type === "idle", + ); + assert.deepEqual((statusChanged?.payload as { status?: unknown } | undefined)?.status, { + type: "idle", + }); + + const leakedChildEvents = events.filter((event) => { + if (event.method === "item/agentMessage/delta") return true; + const payload = event.payload as { threadId?: string } | undefined; + return payload?.threadId === CHILD_A; + }); + assert.deepEqual( + leakedChildEvents.map((event) => event.method), + [], + "child notifications must not be emitted as parent-timeline events", + ); + + yield* runtime.close; + }).pipe(Effect.scoped, Effect.provide(NodeServices.layer)), + ); + + it.effect("assigns names in global spawn order and ignores later non-spawn receiver order", () => + Effect.gen(function* () { + // @effect-diagnostics-next-line preferSchemaOverJson:off + NodeFS.writeFileSync(scriptPath, JSON.stringify(buildOutOfOrderNamingScript()), "utf8"); + yield* Effect.addFinalizer(() => + Effect.sync(() => NodeFS.rmSync(scriptPath, { force: true })), + ); + + const runtime = yield* makeCodexSessionRuntime({ + threadId: ThreadId.make("thread-collab-name-order"), + 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: "name the children" }); + + const events = Array.from(yield* Fiber.join(eventsFiber)); + const renamed = events.filter((event) => event.method === "collabAgent/renamed"); + assert.deepEqual( + renamed.map((event) => { + const payload = event.payload as { agentThreadId?: string; nickname?: string }; + return [payload.agentThreadId, payload.nickname]; + }), + [ + [CHILD_A, "Alpha"], + [CHILD_B, "Beta"], + [CHILD_C, "Gamma"], + [CHILD_D, "Delta"], + ], + ); + + yield* runtime.close; + }).pipe(Effect.scoped, Effect.provide(NodeServices.layer)), + ); + + it.effect("does not backfill an unscoped child into a later parent turn", () => + Effect.gen(function* () { + // @effect-diagnostics-next-line preferSchemaOverJson:off + NodeFS.writeFileSync(scriptPath, JSON.stringify(buildUnscopedExistingChildScript()), "utf8"); + yield* Effect.addFinalizer(() => + Effect.sync(() => NodeFS.rmSync(scriptPath, { force: true })), + ); + + const runtime = yield* makeCodexSessionRuntime({ + threadId: ThreadId.make("thread-collab-unscoped-child"), + 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: "keep the unscoped child separate" }); + + const events = Array.from(yield* Fiber.join(eventsFiber)); + const childBEvents = events.filter( + (event) => + (event.payload as { agentThreadId?: string } | undefined)?.agentThreadId === CHILD_B, + ); + assert.isAbove(childBEvents.length, 0); + assert.isTrue( + childBEvents.every((event) => event.turnId === undefined), + "an existing child without a spawn turn must not inherit a later turn id", + ); + assert.notInclude( + events.map((event) => event.method), + "collabAgent/renamed", + "the unscoped child must not be included in the later fleet name batch", + ); + + yield* runtime.close; + }).pipe(Effect.scoped, Effect.provide(NodeServices.layer)), + ); + it.effect("replays the captured fan-out into synthetic agent events without child leaks", () => Effect.gen(function* () { // @effect-diagnostics-next-line preferSchemaOverJson:off @@ -294,4 +709,44 @@ describe("CodexSessionRuntime collab integration", () => { yield* runtime.close; }).pipe(Effect.scoped, Effect.provide(NodeServices.layer)), ); + + it.effect("keeps logical-root items on the coordinator timeline", () => + Effect.gen(function* () { + // @effect-diagnostics-next-line preferSchemaOverJson:off + NodeFS.writeFileSync(scriptPath, JSON.stringify(buildLogicalRootItemScript()), "utf8"); + yield* Effect.addFinalizer(() => + Effect.sync(() => NodeFS.rmSync(scriptPath, { force: true })), + ); + + const runtime = yield* makeCodexSessionRuntime({ + threadId: ThreadId.make("thread-collab-logical-root"), + 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 === "item/completed"), + Stream.runCollect, + Effect.forkScoped, + ); + + yield* runtime.start(); + yield* runtime.sendTurn({ input: "preserve the parent item" }); + + const events = Array.from(yield* Fiber.join(eventsFiber)); + assert.include( + events.map((event) => event.method), + "item/completed", + ); + assert.notInclude( + events.map((event) => event.method), + "collabAgent/item", + "an unknown logical-root item must not become a child-agent event", + ); + + yield* runtime.close; + }).pipe(Effect.scoped, Effect.provide(NodeServices.layer)), + ); }); diff --git a/apps/server/src/provider/Layers/CodexCollabWire.test.ts b/apps/server/src/provider/Layers/CodexCollabWire.test.ts index 50e5e819d1f..589d7e71b00 100644 --- a/apps/server/src/provider/Layers/CodexCollabWire.test.ts +++ b/apps/server/src/provider/Layers/CodexCollabWire.test.ts @@ -15,7 +15,7 @@ import { assert, describe, it } from "vite-plus/test"; import fixture from "../testFixtures/codexMultiAgentWire.json" with { type: "json" }; -import { routeCodexChildNotification } from "./CodexSessionRuntime.ts"; +import { readCoordinatorAgentNames, routeCodexChildNotification } from "./CodexSessionRuntime.ts"; interface WireNotification { readonly method: string; @@ -61,8 +61,8 @@ describe("codex multi-agent wire capture", () => { it("emits child traffic BEFORE the item that registers the child", () => { // Ordering hazard: the child's own thread/status/changed arrives before // the parent-side subAgentActivity naming it. Registration must tolerate - // child-first arrival, so unregistered child traffic passes through - // rather than being eaten (no regression vs. pre-feature behavior). + // child-first arrival without leaking the child's conversation into the + // parent timeline. const firstChildTraffic = notifications.findIndex((entry) => { const threadId = notificationThreadId(entry); return threadId !== undefined && childThreadIds.has(threadId); @@ -128,11 +128,35 @@ describe("routeCodexChildNotification", () => { } }); + it("reads coordinator-assigned names from fleet summaries", () => { + assert.deepEqual( + readCoordinatorAgentNames( + "Three agents are running in parallel: Halley, Banach, and Parfit.", + ), + ["Halley", "Banach", "Parfit"], + ); + assert.deepEqual( + readCoordinatorAgentNames( + "All three agents completed successfully:\n\n- Halley: generated names\n- Banach: wrote a riddle\n- Parfit: sorted terms", + ), + ["Halley", "Banach", "Parfit"], + ); + assert.deepEqual( + readCoordinatorAgentNames( + "Started 3 Luna medium sub-agents:\n\n- Planck\n- Parfit\n- Avicenna", + ), + ["Planck", "Parfit", "Avicenna"], + ); + assert.deepEqual(readCoordinatorAgentNames("Summary:\n\n- Tests: passed\n- Build: passed"), []); + }); + it("drops only enumerated child chatter", () => { for (const method of [ "item/agentMessage/delta", "item/reasoning/textDelta", "item/commandExecution/outputDelta", + "item/commandExecution/terminalInteraction", + "item/mcpToolCall/progress", "turn/plan/updated", "thread/name/updated", ]) { diff --git a/apps/server/src/provider/Layers/CodexSessionRuntime.ts b/apps/server/src/provider/Layers/CodexSessionRuntime.ts index 58c012bd63e..c76d92329e2 100644 --- a/apps/server/src/provider/Layers/CodexSessionRuntime.ts +++ b/apps/server/src/provider/Layers/CodexSessionRuntime.ts @@ -610,12 +610,12 @@ function readRouteFields(notification: CodexServerNotification): { * synthetic `collabAgent/*` provider events the adapter turns into task.* * runtime events (timelineBypass keeps them out of the parent chat). * - * 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). + * Child identity is normally explicit, but some Codex builds emit a + * `collabAgentToolCall` with receiver thread ids and then immediately stream + * child notifications without a `thread/started` or `subAgentActivity` row. + * Unknown foreign thread ids are therefore provisionally registered once the + * root thread is known. Parent-owned methods still use the routing table and + * are allowed through; child chatter never reaches parent-timeline mapping. */ interface CollabChildAgentState { readonly agentThreadId: string; @@ -631,6 +631,19 @@ interface CollabChildAgentState { * "direct:no-turn" CTA (review finding). */ readonly spawnTurnId: TurnId | undefined; + /** Global position in the parent turn's spawn receiver order. */ + readonly spawnIndex: number | undefined; +} + +interface CollabChildRegistrationInput { + readonly agentThreadId: string; + readonly nickname?: string | undefined; + readonly role?: string | undefined; + readonly agentPath?: string | undefined; + readonly depth?: number | undefined; + readonly parentThreadId?: string | undefined; + readonly spawnTurnId?: TurnId | undefined; + readonly spawnIndex?: number | undefined; } function readThreadSpawnSource(thread: { readonly source: unknown }): @@ -665,6 +678,70 @@ function readThreadSpawnSource(thread: { readonly source: unknown }): }; } +function readCollabPromptNickname(prompt: string | null | undefined): string | undefined { + if (!prompt) { + return undefined; + } + const firstLine = prompt.trim().split(/\r?\n/, 1)[0]?.trim(); + if (!firstLine) { + return undefined; + } + const match = firstLine.match(/^You are (?:the )?(.+?)(?:\s+for\s+|[.!?](?:\s|$))/i); + const nickname = match?.[1]?.trim(); + return nickname && nickname.length <= 120 ? nickname : undefined; +} + +function coordinatorNameCandidate(value: string): string | undefined { + const name = value.replaceAll(/[*`_]/g, "").trim(); + return /^(?:[A-Z][A-Za-z0-9_-]{1,40})(?:\s+[A-Z][A-Za-z0-9_-]{1,40}){0,4}$/.test(name) + ? name + : undefined; +} + +/** + * Reads explicit agent names from a coordinator-authored status or summary. + * Codex's direct collab wire sometimes carries only receiver UUIDs; when the + * coordinator later says "Halley, Banach, and Parfit" or writes + * "- Halley: ...", the names can still be joined to the current spawn batch. + */ +export function readCoordinatorAgentNames(text: string): ReadonlyArray { + const names: string[] = []; + const add = (value: string) => { + const name = coordinatorNameCandidate(value); + if (name && !names.includes(name)) { + names.push(name); + } + }; + let readingAgentBulletList = false; + + for (const line of text.split(/\r?\n/)) { + const agentHeader = /\b(?:agents?|sub[- ]?agents?)\b[^:]{0,100}:/i.test(line); + if (agentHeader) { + readingAgentBulletList = true; + } + + const agentList = line.match(/\b(?:agents?|sub[- ]?agents?)\b[^:]{0,100}:\s*(.+)$/i)?.[1]; + if (agentList) { + const sentence = agentList.split(/[.!?](?:\s|$)/, 1)[0] ?? agentList; + for (const candidate of sentence.split(/\s*(?:,|\band\b)\s*/i)) { + add(candidate); + } + } + + const bullet = line.match(/^\s*(?:[-*]|\d+[.)])\s+(?:\*\*)?([^\n]+?)(?:\*\*)?\s*$/)?.[1]; + if (readingAgentBulletList && bullet) { + add(bullet.replace(/\s*:.*/, "")); + continue; + } + + if (readingAgentBulletList && !agentHeader && line.trim().length > 0) { + readingAgentBulletList = false; + } + } + + return names; +} + function rememberCollabReceiverTurns( collabReceiverTurns: Map, notification: CodexServerNotification, @@ -741,8 +818,10 @@ const CHILD_CHATTER_METHODS: ReadonlySet = new Set([ "item/reasoning/summaryTextDelta", "item/reasoning/summaryPartAdded", "item/commandExecution/outputDelta", + "item/commandExecution/terminalInteraction", "item/fileChange/outputDelta", "item/fileChange/patchUpdated", + "item/mcpToolCall/progress", "item/plan/delta", "turn/plan/updated", "turn/diff/updated", @@ -973,6 +1052,120 @@ export const makeCodexSessionRuntime = ( ), ); + const emitCollabAgentStarted = (state: CollabChildAgentState) => + emitEvent({ + kind: "notification", + threadId: options.threadId, + ...(state.spawnTurnId ? { turnId: state.spawnTurnId } : {}), + method: "collabAgent/started", + 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 } : {}), + }, + }); + + const emitCollabAgentStatusChanged = ( + state: CollabChildAgentState, + status: + | { readonly type: "active"; readonly activeFlags: ReadonlyArray } + | { readonly type: "idle" } + | { readonly type: "systemError" }, + ) => + emitEvent({ + kind: "notification", + threadId: options.threadId, + ...(state.spawnTurnId ? { turnId: state.spawnTurnId } : {}), + method: "collabAgent/statusChanged", + payload: { + agentThreadId: state.agentThreadId, + ...(state.nickname ? { nickname: state.nickname } : {}), + ...(state.role ? { role: state.role } : {}), + ...(state.agentPath ? { agentPath: state.agentPath } : {}), + status, + }, + }); + + const emitCollabAgentRenamed = (state: CollabChildAgentState) => + emitEvent({ + kind: "notification", + threadId: options.threadId, + ...(state.spawnTurnId ? { turnId: state.spawnTurnId } : {}), + method: "collabAgent/renamed", + payload: { + agentThreadId: state.agentThreadId, + ...(state.nickname ? { nickname: state.nickname } : {}), + }, + }); + + const registerCollabChild = (input: CollabChildRegistrationInput) => + Effect.gen(function* () { + const existing = (yield* Ref.get(collabChildAgentsRef)).get(input.agentThreadId); + const session = yield* Ref.get(sessionRef); + const state: CollabChildAgentState = { + agentThreadId: input.agentThreadId, + nickname: input.nickname ?? existing?.nickname, + role: input.role ?? existing?.role, + agentPath: input.agentPath ?? existing?.agentPath, + depth: input.depth ?? existing?.depth, + parentThreadId: input.parentThreadId ?? existing?.parentThreadId, + // A child can be mentioned again by wait/resume/sendInput/close + // calls. Once a spawn position is known, later tool calls must not + // overwrite it with their unrelated receiver ordering. + spawnIndex: existing?.spawnIndex ?? input.spawnIndex, + // Registration-time-only: a late metadata signal must not attach + // an old child to an unrelated parent turn. + spawnTurnId: existing + ? existing.spawnTurnId + : (input.spawnTurnId ?? session.activeTurnId ?? undefined), + }; + yield* Ref.update(collabChildAgentsRef, (current) => { + const next = new Map(current); + next.set(state.agentThreadId, state); + return next; + }); + if (!existing) { + yield* emitCollabAgentStarted(state); + } + return state; + }); + + const applyCoordinatorAgentNames = ( + names: ReadonlyArray, + parentTurnId: TurnId | undefined, + ) => + Effect.gen(function* () { + if (!parentTurnId || names.length === 0) { + return; + } + const renamed = yield* Ref.modify(collabChildAgentsRef, (current) => { + const candidates = Array.from(current.values()) + .filter((state) => state.spawnTurnId === parentTurnId) + .sort( + (left, right) => + (left.spawnIndex ?? Number.MAX_SAFE_INTEGER) - + (right.spawnIndex ?? Number.MAX_SAFE_INTEGER), + ); + if (candidates.length !== names.length) { + const noUpdates: CollabChildAgentState[] = []; + return [noUpdates, current] as const; + } + const next = new Map(current); + const updates: CollabChildAgentState[] = candidates.map((state, index) => { + const updated = { ...state, nickname: names[index] }; + next.set(updated.agentThreadId, updated); + return updated; + }); + return [updates, next] as const; + }); + for (const state of renamed) { + yield* emitCollabAgentRenamed(state); + } + }); + /** * Registers v2 collab children and re-emits their notifications as * synthetic `collabAgent/*` events for the adapter's task.* synthesis. @@ -989,49 +1182,123 @@ export const makeCodexSessionRuntime = ( if (!spawn) { return false; } - // Merge with any subAgentActivity registration that got here - // first. spawnTurnId is REGISTRATION-time-only on both paths: for - // an already-known child we keep its value (set or unset) — a - // later thread/started during an unrelated parent turn must not - // backfill that turn as the spawn batch, which would stamp an old - // 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 spawnTurnId = existingChild - ? existingChild.spawnTurnId - : ((yield* Ref.get(sessionRef)).activeTurnId ?? undefined); - const state: CollabChildAgentState = { + // Merge with any provisional/subAgentActivity registration that got + // here first. The helper keeps spawnTurnId registration-scoped and + // emits a start row only for a genuinely new child. + yield* registerCollabChild({ agentThreadId: thread.id, - nickname: spawn.nickname ?? thread.agentNickname ?? existingChild?.nickname, - role: spawn.role ?? thread.agentRole ?? existingChild?.role, - agentPath: spawn.agentPath ?? existingChild?.agentPath, - depth: spawn.depth ?? existingChild?.depth, - parentThreadId: - spawn.parentThreadId ?? thread.parentThreadId ?? existingChild?.parentThreadId, - spawnTurnId, - }; - 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 } : {}), - }, + nickname: spawn.nickname ?? thread.agentNickname ?? undefined, + role: spawn.role ?? thread.agentRole ?? undefined, + agentPath: spawn.agentPath, + depth: spawn.depth, + parentThreadId: spawn.parentThreadId ?? thread.parentThreadId ?? undefined, }); return true; } + // Registration path 2b: newer Codex builds expose the child ids on + // the parent-side collabAgentToolCall, then stream the child thread + // directly without a thread/started or subAgentActivity row. Keep + // the parent tool item visible, but make every receiver a known + // child before its conversation can be mapped. + if ( + (notification.method === "item/started" || notification.method === "item/completed") && + notification.params.item.type === "collabAgentToolCall" + ) { + const item = notification.params.item; + const rootProviderThreadId = currentProviderThreadId(yield* Ref.get(sessionRef)); + const currentChildren = yield* Ref.get(collabChildAgentsRef); + const senderChildState = + rootProviderThreadId !== undefined && item.senderThreadId !== rootProviderThreadId + ? currentChildren.get(item.senderThreadId) + : undefined; + // A logical root id can differ from thread/start's id. Treat a + // sender as nested only after its parent-side spawn registered it; + // otherwise a root collab tool item could be mistaken for child + // chatter and disappear from the coordinator timeline. + const senderIsChild = senderChildState !== undefined; + // Nested children belong to the same fleet as their sender. The + // provider's notification turn id is the nested child's own turn, + // not the root coordinator turn used by the Agents panel. + const spawnTurnId = + senderChildState?.spawnTurnId ?? TurnId.make(notification.params.turnId); + let nextSpawnIndex: number | undefined; + if (item.tool === "spawnAgent") { + nextSpawnIndex = + Array.from(currentChildren.values()) + .filter( + (state) => state.spawnTurnId === spawnTurnId && state.spawnIndex !== undefined, + ) + .reduce((highest, state) => Math.max(highest, state.spawnIndex ?? -1), -1) + 1; + } + for (const receiverThreadId of item.receiverThreadIds) { + if (!receiverThreadId || receiverThreadId === rootProviderThreadId) { + continue; + } + const existing = (yield* Ref.get(collabChildAgentsRef)).get(receiverThreadId); + const spawnIndex = + item.tool === "spawnAgent" ? (existing?.spawnIndex ?? nextSpawnIndex) : undefined; + if (item.tool === "spawnAgent" && existing?.spawnIndex === undefined) { + nextSpawnIndex = (nextSpawnIndex ?? 0) + 1; + } + const child = yield* registerCollabChild({ + agentThreadId: receiverThreadId, + nickname: + item.tool === "spawnAgent" ? readCollabPromptNickname(item.prompt) : undefined, + parentThreadId: senderIsChild ? item.senderThreadId : undefined, + spawnTurnId, + ...(spawnIndex !== undefined && spawnIndex >= 0 ? { spawnIndex } : {}), + }); + + // `agentsStates` is the only terminal signal emitted by some + // Codex builds: the child may finish with a final agentMessage + // but never send its own turn/completed/status notification. + // Fold that state into the same synthetic lifecycle used by the + // explicit child-thread path so the Agents panel does not stay + // permanently in "working". + const agentStatus = item.agentsStates[receiverThreadId]?.status; + if (agentStatus === "pendingInit" || agentStatus === "running") { + yield* emitCollabAgentStatusChanged(child, { + type: "active", + activeFlags: [], + }); + } else if ( + agentStatus === "completed" || + agentStatus === "interrupted" || + agentStatus === "shutdown" + ) { + yield* emitCollabAgentStatusChanged(child, { type: "idle" }); + } else if (agentStatus === "errored" || agentStatus === "notFound") { + yield* emitCollabAgentStatusChanged(child, { type: "systemError" }); + } + } + // A collab tool call owned by a child is itself child chatter. A + // root-owned call remains on the parent timeline for context. + return senderIsChild; + } + + // Some direct-collab Codex builds expose only receiver UUIDs at spawn + // time. If the parent later explicitly names the fleet in a completed + // assistant message, attach those names to the children from the same + // parent turn before the message reaches the normal parent mapper. + if ( + notification.method === "item/completed" && + notification.params.item.type === "agentMessage" + ) { + const text = notification.params.item.text; + if (typeof text === "string") { + // The name summary is parent-authored in normal Codex traffic, + // but some app-server fixtures/builds use a logical root id that + // differs from the thread id returned by thread/start. Matching + // only the exact spawn-turn batch keeps this safe for child final + // answers while avoiding a brittle root-id comparison. + yield* applyCoordinatorAgentNames( + readCoordinatorAgentNames(text), + TurnId.make(notification.params.turnId), + ); + } + } + // Registration path 2: parent-side subAgentActivity item names the // child thread (may arrive before or after thread/started). if ( @@ -1074,6 +1341,7 @@ export const makeCodexSessionRuntime = ( depth: existing?.depth, parentThreadId: existing?.parentThreadId, spawnTurnId: existing ? existing.spawnTurnId : activitySpawnTurnId, + spawnIndex: existing?.spawnIndex, }); return next; }); @@ -1104,10 +1372,25 @@ export const makeCodexSessionRuntime = ( if (providerConversationId === interceptRootId) { return false; } - const children = yield* Ref.get(collabChildAgentsRef); - const child = children.get(providerConversationId); + const foreignConversation = + interceptRootId !== undefined && providerConversationId !== interceptRootId; + let child = (yield* Ref.get(collabChildAgentsRef)).get(providerConversationId); if (!child) { - return false; + // A few Codex versions omit both child registration signals and + // only expose the foreign thread id on its first notification. The + // root/foreign distinction is useful for lifecycle chatter, but an + // unknown item can still be coordinator content when the provider + // uses a logical root id different from thread/start. Keep item + // rows on the parent path until a collab spawn identifies them. + if ( + !foreignConversation || + routeCodexChildNotification(notification.method) === "parent" || + notification.method === "item/started" || + notification.method === "item/completed" + ) { + return false; + } + child = yield* registerCollabChild({ agentThreadId: providerConversationId }); } const childIdentity = { agentThreadId: child.agentThreadId, @@ -1190,6 +1473,17 @@ export const makeCodexSessionRuntime = ( item: notification.params.item, }, }); + if ( + notification.method === "item/completed" && + notification.params.item.type === "agentMessage" && + notification.params.item.phase === "final_answer" + ) { + // The child final answer is consumed by Codex's parent model; + // it is not parent-chat content. Some versions omit the child + // turn/status lifecycle, so this terminal item must also mark + // the corresponding Agents row idle. + yield* emitCollabAgentStatusChanged(child, { type: "idle" }); + } return true; case "thread/closed": // The child is gone: drop its live-turn entry so a later Stop @@ -1386,7 +1680,9 @@ export const makeCodexSessionRuntime = ( yield* client.handleServerNotification("turn/started", (payload) => currentSessionProviderThreadId.pipe( Effect.flatMap((providerThreadId) => { - if (providerThreadId && payload.threadId !== providerThreadId) { + // Do not let an early child notification claim the root session + // before thread/start has stored the provider thread id. + if (!providerThreadId || payload.threadId !== providerThreadId) { return Effect.void; } return updateSession(sessionRef, { diff --git a/apps/server/src/provider/testFixtures/codexCollabMockPeer.mjs b/apps/server/src/provider/testFixtures/codexCollabMockPeer.mjs index f06e984c9aa..e827c04ae77 100644 --- a/apps/server/src/provider/testFixtures/codexCollabMockPeer.mjs +++ b/apps/server/src/provider/testFixtures/codexCollabMockPeer.mjs @@ -41,6 +41,13 @@ rl.on("line", (line) => { } if (method === "thread/start" || method === "thread/resume") { write({ id, result: fixture.responses.threadStart }); + // Some ordering regressions need traffic before the first parent turn has + // started, when the runtime has no active spawn turn yet. Scripts should + // put a root thread/started notification first when they need to model + // that the provider has already identified the root conversation. + for (const notification of script.preTurnNotifications ?? []) { + write({ jsonrpc: "2.0", method: notification.method, params: notification.params }); + } return; } if (method === "turn/start") { diff --git a/apps/web/src/components/AgentsPanel.tsx b/apps/web/src/components/AgentsPanel.tsx index 4eeff67ce5f..380caaa4214 100644 --- a/apps/web/src/components/AgentsPanel.tsx +++ b/apps/web/src/components/AgentsPanel.tsx @@ -79,6 +79,32 @@ function elapsedBetween(startedAt: string, endIso: string | null): string { return formatElapsedSeconds((end - start) / 1000); } +function isOpaqueAgentTitle(title: string, id: string): boolean { + return ( + title === id || /^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/i.test(title) + ); +} + +function readableActivity(activity: string | null): string | null { + if (!activity) { + return null; + } + if (/^(?:\/usr\/bin\/)?(?:zsh|bash|sh)\s+-lc\b/i.test(activity)) { + return "Running a shell command"; + } + if (activity.trim().toLocaleLowerCase() === "reasoning") { + return "Thinking"; + } + return activity; +} + +function readableRole(role: string | null): string | null { + if (!role) { + return null; + } + return role.trim().toLocaleLowerCase() === "general-purpose" ? "general" : role; +} + /** * Elapsed time for the current activation. Live agents self-tick via DOM * writes (zero React commits per tick); settled agents freeze at completedAt. @@ -137,28 +163,45 @@ function agentActivityText(agent: RuntimeSubagent): string | null { } /** Flat, non-interactive agent status line. No unfold. */ -function AgentRow({ agent }: { agent: RuntimeSubagent }) { +function AgentRow({ + agent, + fallbackLabel = "Agent", +}: { + agent: RuntimeSubagent; + fallbackLabel?: string; +}) { const visuals = STATUS_VISUALS[agent.status]; - const activity = agentActivityText(agent); + const rawActivity = agentActivityText(agent); + const activity = readableActivity(rawActivity); const modelLabel = formatSubagentModelLabel(agent.model, agent.effort); + const titleIsOpaque = isOpaqueAgentTitle(agent.title, agent.id); + const displayTitle = titleIsOpaque ? fallbackLabel : agent.title; const role = - agent.role?.trim().toLocaleLowerCase() === agent.title.trim().toLocaleLowerCase() + readableRole(agent.role)?.trim().toLocaleLowerCase() === displayTitle.trim().toLocaleLowerCase() ? null - : agent.role; + : readableRole(agent.role); const metadata = [ modelLabel, - agent.usage ? `${formatSubagentTokenCount(agent.usage.totalTokens)} tok` : "— tok", - agent.usage?.toolUses !== undefined ? `${agent.usage.toolUses} tools` : null, + agent.usage ? `${formatSubagentTokenCount(agent.usage.totalTokens)} tokens` : null, + agent.usage?.toolUses !== undefined + ? `${agent.usage.toolUses} tool${agent.usage.toolUses === 1 ? "" : "s"}` + : null, agent.activationCount > 1 ? `run ${agent.activationCount}` : null, ].filter((value): value is string => value !== null); + const identityLabel = titleIsOpaque ? `${displayTitle} · ${agent.id}` : displayTitle; return ( -
+
- {agent.title} + {displayTitle} + {visuals.label} {role ? ( {role} @@ -178,6 +221,7 @@ function AgentRow({ agent }: { agent: RuntimeSubagent }) { "col-start-2 col-end-4 row-start-2 block truncate text-xs", agent.status === "failed" ? "text-destructive-foreground" : "text-muted-foreground", )} + title={rawActivity ?? undefined} > {activity ?? visuals.label} @@ -366,7 +410,11 @@ function PhaseSection({ ) : null} - {open ? phase.members.map((member) => ) : null} + {open + ? phase.members.map((member, index) => ( + + )) + : null}
); } @@ -438,8 +486,8 @@ function ExpandedWorkflowSection({ {group.phases.map((phase) => ( ))} - {group.unphasedMembers.map((member) => ( - + {group.unphasedMembers.map((member, index) => ( + ))} {group.phases.length === 0 && group.unphasedMembers.length === 0 ? ( @@ -486,7 +534,7 @@ function CollapsedWorkflowSection({ {failed > 0 ? {failed} failed : null} {members.length} agents - · {formatSubagentTokenCount(totalTokens)} tok + · {formatSubagentTokenCount(totalTokens)} tokens {elapsed ? · {elapsed} : null} @@ -554,11 +602,14 @@ export function AgentsPanel({ ))} {model.directAgents.length > 0 ? (
-
- Direct spawns +
+ Direct agents + + {model.directAgents.length} +
- {model.directAgents.map((agent) => ( - + {model.directAgents.map((agent, index) => ( + ))}
) : null} @@ -574,7 +625,9 @@ export function AgentsPanel({ {model.idleCount > 0 ? {model.idleCount} idle : null} {model.settledCount > 0 ? {model.settledCount} settled : null} - Σ {formatSubagentTokenCount(model.totalTokens)} tok + + Total {formatSubagentTokenCount(model.totalTokens)} tokens +
);