diff --git a/src/agents/subagent-registry-lifecycle-completion.ts b/src/agents/subagent-registry-lifecycle-completion.ts index 310fc98c4f4a..42cabda69d13 100644 --- a/src/agents/subagent-registry-lifecycle-completion.ts +++ b/src/agents/subagent-registry-lifecycle-completion.ts @@ -569,7 +569,18 @@ export function createSubagentRegistryLifecycleCompletion( } const suppressedForSteerRestart = params.suppressAnnounceForSteerRestart(entry); - if (mutated && !suppressedForSteerRestart && !completeParams.suppressSessionEffects) { + // Recovery persists its terminal state before draining this callback, so an + // unchanged row still needs its first session-status and progress events. + const shouldPublishTerminalStatus = + mutated || + (completeParams.recoverInterrupted === true && + !isProvisionalKill && + !progressEndedEntries.has(entry)); + if ( + shouldPublishTerminalStatus && + !suppressedForSteerRestart && + !completeParams.suppressSessionEffects + ) { emitSessionLifecycleEvent({ sessionKey: entry.childSessionKey, reason: "subagent-status", diff --git a/src/agents/subagent-registry-lifecycle.test.ts b/src/agents/subagent-registry-lifecycle.test.ts index e7298b033dbc..28f36c47db9c 100644 --- a/src/agents/subagent-registry-lifecycle.test.ts +++ b/src/agents/subagent-registry-lifecycle.test.ts @@ -337,6 +337,67 @@ describe("subagent registry lifecycle hardening", () => { expect(emitSubagentProgressEndedForRun).toHaveBeenCalledWith(entry); }); + it("publishes a recovered terminal session status exactly once", async () => { + const entry = createRunEntry(); + const emitSubagentProgressEndedForRun = vi.fn(async () => {}); + const controller = createLifecycleController({ entry, emitSubagentProgressEndedForRun }); + const completion = { + runId: entry.runId, + endedAt: 4_000, + outcome: { status: "error" as const, error: "restart interrupted run" }, + reason: SUBAGENT_ENDED_REASON_ERROR, + triggerCleanup: false, + recoverInterrupted: true, + } satisfies SubagentCompletionParams; + + await controller.completeSubagentRun(completion); + await controller.completeSubagentRun(completion); + + expect(lifecycleEventMocks.emitSessionLifecycleEvent).toHaveBeenCalledExactlyOnceWith({ + sessionKey: entry.childSessionKey, + reason: "subagent-status", + parentSessionKey: entry.requesterSessionKey, + label: entry.label, + }); + expect(emitSubagentProgressEndedForRun).toHaveBeenCalledExactlyOnceWith(entry); + }); + + it("does not publish recovered terminal events for an ordinary completion", async () => { + const outcome = { + status: "error" as const, + error: "restart interrupted run", + startedAt: 2_000, + endedAt: 4_000, + elapsedMs: 2_000, + }; + const entry = createRunEntry({ + endedAt: 4_000, + endedReason: SUBAGENT_ENDED_REASON_ERROR, + terminalOwner: "interrupted-recovery", + outcome, + execution: { + status: "terminal", + startedAt: 2_000, + endedAt: 4_000, + outcome, + }, + completion: { required: false, resultText: null, capturedAt: 4_000 }, + }); + const emitSubagentProgressEndedForRun = vi.fn(async () => {}); + const controller = createLifecycleController({ entry, emitSubagentProgressEndedForRun }); + + await controller.completeSubagentRun({ + runId: entry.runId, + endedAt: 4_000, + outcome: { status: "error", error: "restart interrupted run" }, + reason: SUBAGENT_ENDED_REASON_ERROR, + triggerCleanup: false, + }); + + expect(lifecycleEventMocks.emitSessionLifecycleEvent).not.toHaveBeenCalled(); + expect(emitSubagentProgressEndedForRun).not.toHaveBeenCalled(); + }); + it("keeps task finalization, resource retirement, and announce cleanup root-admitted", async () => { const entry = createRunEntry({ expectsCompletionMessage: true }); let releaseBrowserCleanup: (() => void) | undefined; diff --git a/src/gateway/session-message-events.test.ts b/src/gateway/session-message-events.test.ts index 16179e6e3117..9ad384bba519 100644 --- a/src/gateway/session-message-events.test.ts +++ b/src/gateway/session-message-events.test.ts @@ -11,6 +11,9 @@ import { GATEWAY_CLIENT_IDS, GATEWAY_CLIENT_MODES, } from "../../packages/gateway-protocol/src/client-info.js"; +import { SUBAGENT_ENDED_REASON_ERROR } from "../agents/subagent-lifecycle-events.js"; +import { createSubagentRegistryLifecycleController } from "../agents/subagent-registry-lifecycle.js"; +import type { SubagentRunRecord } from "../agents/subagent-registry.types.js"; import { loadTranscriptEvents, persistSessionTranscriptTurn, @@ -518,6 +521,98 @@ describe("session.message websocket events", () => { } }); + test("broadcasts a recovered subagent terminal session to a subscribed gateway exactly once", async () => { + const storePath = await createSessionStoreFile(); + const entry: SubagentRunRecord = { + runId: "run-recovered-subscriber", + childSessionKey: "agent:main:subagent:recovered-subscriber", + requesterSessionKey: "agent:main:parent", + requesterDisplayKey: "parent", + task: "finish recovered child work", + cleanup: "keep", + createdAt: 1_000, + startedAt: 2_000, + }; + await writeSessionStore({ + entries: { + [entry.childSessionKey]: { + sessionId: "sess-recovered-subscriber", + spawnedBy: entry.requesterSessionKey, + updatedAt: Date.now(), + }, + }, + storePath, + }); + + const emitSubagentProgressEndedForRun = vi.fn(async () => {}); + const controller = createSubagentRegistryLifecycleController({ + runs: new Map([[entry.runId, entry]]), + resumedRuns: new Set(), + subagentAnnounceTimeoutMs: 1_000, + getRuntimeConfig: () => ({}), + persist: vi.fn(), + persistOrThrow: vi.fn(), + clearPendingLifecycleError: vi.fn(), + countPendingDescendantRuns: () => 0, + suppressAnnounceForSteerRestart: () => false, + resolveSubagentTask: () => ({ lookup: "available" }), + shouldEmitEndedHookForRun: () => false, + emitSubagentEndedHookForRun: vi.fn(async () => {}), + emitSubagentProgressEndedForRun, + notifyContextEngineSubagentEnded: vi.fn(async () => {}), + retireSupersededRun: vi.fn(async () => {}), + resumeSubagentRun: vi.fn(), + callGateway: async >() => ({}) as T, + captureSubagentCompletionReply: vi.fn(async () => undefined), + runSubagentAnnounceFlow: vi.fn(async () => false), + maybeWakeRequesterAfterAllChildrenSettled: vi.fn(async () => false), + warn: vi.fn(), + }); + const completion = { + runId: entry.runId, + endedAt: 4_000, + outcome: { status: "error" as const, error: "restart interrupted run" }, + reason: SUBAGENT_ENDED_REASON_ERROR, + triggerCleanup: false, + recoverInterrupted: true, + } satisfies Parameters[0]; + + await withOperatorSessionSubscriber(async (ws) => { + const waitForRecoveredTerminal = (timeoutMs?: number) => + onceMessage( + ws, + (message) => + message.type === "event" && + message.event === "sessions.changed" && + (message.payload as { sessionKey?: string; reason?: string } | undefined) + ?.sessionKey === entry.childSessionKey && + (message.payload as { reason?: string } | undefined)?.reason === "subagent-status", + timeoutMs, + ); + const changedEvent = waitForRecoveredTerminal(); + + await controller.completeSubagentRun(completion); + + const event = await changedEvent; + expectRecordFields(event.payload, { + sessionKey: entry.childSessionKey, + reason: "subagent-status", + status: "failed", + endedAt: completion.endedAt, + spawnedBy: entry.requesterSessionKey, + }); + expect(emitSubagentProgressEndedForRun).toHaveBeenCalledExactlyOnceWith(entry); + + // A resumed callback must not publish a second terminal event to an + // already-subscribed Control UI client for the same child generation. + await expectNoMessageWithin({ + action: () => controller.completeSubagentRun(completion), + watch: waitForRecoveredTerminal, + }); + expect(emitSubagentProgressEndedForRun).toHaveBeenCalledExactlyOnceWith(entry); + }); + }); + test("includes spawned session ownership metadata on lifecycle sessions.changed events", async () => { const storePath = await createSessionStoreFile(); await writeSessionStore({