diff --git a/apps/server/src/orchestration/ThreadBackgroundLiveness.test.ts b/apps/server/src/orchestration/ThreadBackgroundLiveness.test.ts index 02dbfe2f44ab..4a4b68ced598 100644 --- a/apps/server/src/orchestration/ThreadBackgroundLiveness.test.ts +++ b/apps/server/src/orchestration/ThreadBackgroundLiveness.test.ts @@ -28,58 +28,6 @@ describe("ThreadBackgroundLiveness", () => { expect(liveness.getThreadBackgroundLiveness("thread")).toBeNull(); }); - it("does not let status-free updated restart an idle task", () => { - const liveness = ThreadBackgroundLiveness.make(); - liveness.recordTaskLiveness({ - threadId: "thread", - taskId: "task", - taskType: undefined, - status: undefined, - kind: "started", - }); - liveness.recordTaskLiveness({ - threadId: "thread", - taskId: "task", - taskType: undefined, - status: "idle", - kind: "updated", - }); - liveness.recordTaskLiveness({ - threadId: "thread", - taskId: "task", - taskType: undefined, - status: undefined, - kind: "updated", - }); - expect(liveness.getThreadBackgroundLiveness("thread")).toBeNull(); - }); - - it("a genuine restart (kind: started) still revives an idle task", () => { - const liveness = ThreadBackgroundLiveness.make(); - liveness.recordTaskLiveness({ - threadId: "thread", - taskId: "task", - taskType: undefined, - status: undefined, - kind: "started", - }); - liveness.recordTaskLiveness({ - threadId: "thread", - taskId: "task", - taskType: undefined, - status: "idle", - kind: "updated", - }); - liveness.recordTaskLiveness({ - threadId: "thread", - taskId: "task", - taskType: undefined, - status: undefined, - kind: "started", - }); - expect(liveness.getThreadBackgroundLiveness("thread")).toBe("working"); - }); - it("agents present as working; monitors as monitoring; agents win", () => { const liveness = ThreadBackgroundLiveness.make(); const threadId = "t-live-1"; diff --git a/apps/server/src/orchestration/ThreadBackgroundLiveness.ts b/apps/server/src/orchestration/ThreadBackgroundLiveness.ts index e2fbeff867ef..d4d6da06dfcd 100644 --- a/apps/server/src/orchestration/ThreadBackgroundLiveness.ts +++ b/apps/server/src/orchestration/ThreadBackgroundLiveness.ts @@ -130,11 +130,10 @@ export function make(): ThreadBackgroundLivenessService["Service"] { return; } - // Status-free progress/updated is a description tick, not a restart. A - // delayed progress or updated event after idle must not put the task - // back in the live set (#7128, #7172). "started" is excluded on - // purpose: a genuine restart must still revive the task. - if ((input.kind === "progress" || input.kind === "updated") && input.status === undefined) { + // Status-free progress is a description tick, not a restart. A delayed + // progress event after idle must not put the task back in the live set + // (#7128). + if (input.kind === "progress" && input.status === undefined) { const existing = stateByThreadId.get(input.threadId); const stillLive = existing !== undefined && diff --git a/apps/server/src/provider/Layers/CodexAdapter.test.ts b/apps/server/src/provider/Layers/CodexAdapter.test.ts index 3413be6db8cd..26fb1b166f61 100644 --- a/apps/server/src/provider/Layers/CodexAdapter.test.ts +++ b/apps/server/src/provider/Layers/CodexAdapter.test.ts @@ -617,6 +617,7 @@ lifecycleLayer("CodexAdapterLive lifecycle", (it) => { ); }), ); + 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/CodexCollabRuntime.integration.test.ts b/apps/server/src/provider/Layers/CodexCollabRuntime.integration.test.ts index ae8eb3b6d490..a1b46e003520 100644 --- a/apps/server/src/provider/Layers/CodexCollabRuntime.integration.test.ts +++ b/apps/server/src/provider/Layers/CodexCollabRuntime.integration.test.ts @@ -181,13 +181,6 @@ 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: { @@ -214,7 +207,7 @@ describe("CodexSessionRuntime collab integration", () => { hangInterruptFor: CHILD_A, notifications: [ turnStartedA, - interactedRegistrationA, + registrationA, memoryThreadStarted, memoryTurnStarted, registrationB, @@ -240,38 +233,26 @@ describe("CodexSessionRuntime collab integration", () => { environment: { ...process.env, T3_CODEX_COLLAB_SCRIPT: scriptPath }, }); - // 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( + // 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( Stream.filter( (event) => event.method === "collabAgent/turnStarted" && - [CHILD_A, CHILD_B].includes( - (event.payload as { agentThreadId?: string }).agentThreadId ?? "", - ), + (event.payload as { agentThreadId?: string }).agentThreadId === CHILD_B, ), - Stream.take(2), + Stream.take(1), Stream.runCollect, Effect.forkScoped, ); yield* runtime.start(); yield* runtime.sendTurn({ input: "fan out and hang" }); - const childrenStarted = yield* Fiber.join(childrenStartedFiber).pipe( + const childBStarted = yield* Fiber.join(childBStartedFiber).pipe( Effect.timeoutOption("15 seconds"), ); - 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"); - } + assert.isTrue(childBStarted._tag === "Some", "child B turnStarted never arrived"); // 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 0618ab87a2d7..fd926e43d7bf 100644 --- a/apps/server/src/provider/Layers/CodexSessionRuntime.ts +++ b/apps/server/src/provider/Layers/CodexSessionRuntime.ts @@ -1107,8 +1107,8 @@ export const makeCodexSessionRuntime = ( return false; } const activitySpawnTurnId = (yield* Ref.get(sessionRef)).activeTurnId ?? undefined; - const existingChild = (yield* Ref.get(collabChildAgentsRef)).get(item.agentThreadId); yield* Ref.update(collabChildAgentsRef, (current) => { + const existing = current.get(item.agentThreadId); const next = new Map(current); // Merge-late semantics: when thread/started registered first, a // later subAgentActivity still carries the real agentPath (and a @@ -1120,13 +1120,13 @@ export const makeCodexSessionRuntime = ( next.set(item.agentThreadId, { agentThreadId: item.agentThreadId, nickname: - existingChild?.nickname ?? + existing?.nickname ?? item.agentPath.split("/").findLast((segment) => segment.length > 0), - role: existingChild?.role, - agentPath: existingChild?.agentPath ?? item.agentPath, - depth: existingChild?.depth, - parentThreadId: existingChild?.parentThreadId, - spawnTurnId: existingChild ? existingChild.spawnTurnId : activitySpawnTurnId, + role: existing?.role, + agentPath: existing?.agentPath ?? item.agentPath, + depth: existing?.depth, + parentThreadId: existing?.parentThreadId, + spawnTurnId: existing ? existing.spawnTurnId : activitySpawnTurnId, }); return next; }); @@ -1142,29 +1142,6 @@ export const makeCodexSessionRuntime = ( activityKind: item.kind, }, }); - // A child turn can start before this activity registers the child. - // The foreign-notification suppressor records that live turn but - // cannot emit agent lifecycle until identity is known. Replay the - // explicit start after first registration so sidebar liveness sees - // genuine work; a trailing interaction with no live turn remains - // the status-free metadata update mapped by CodexAdapter. - const preRegistrationLiveTurn = (yield* Ref.get(collabChildLiveTurnsRef)).get( - item.agentThreadId, - ); - if (!existingChild && item.kind === "interacted" && preRegistrationLiveTurn) { - yield* emitEvent({ - kind: "notification", - threadId: options.threadId, - ...(registeredChild?.spawnTurnId ? { turnId: registeredChild.spawnTurnId } : {}), - method: "collabAgent/turnStarted", - payload: { - agentThreadId: item.agentThreadId, - ...(registeredChild?.nickname ? { nickname: registeredChild.nickname } : {}), - ...(registeredChild?.role ? { role: registeredChild.role } : {}), - agentPath: item.agentPath, - }, - }); - } return true; }