diff --git a/packages/acp/acp/src/index.ts b/packages/acp/acp/src/index.ts index d19a2e5753..2c877df7c4 100644 --- a/packages/acp/acp/src/index.ts +++ b/packages/acp/acp/src/index.ts @@ -336,6 +336,13 @@ export function apply(ctx: Context, config: AcpConfig): void { closed = true const records = [...sessions.values()] sessions.clear() + // Stop the bridge's own work before any await: a descendant drain can block + // on persistence or scoped cleanup, and the top-level agents must not keep + // running model and tool calls for its whole duration. + for (const record of records) { + record.agent.cancel({ kind: 'user' }) + settlePrompt(record, 'cancelled') + } quiescing = (async () => { // Continuable subagents outlive the turn that started them, and their // Activations own descendant teardown. Drain that forest child-first @@ -351,10 +358,7 @@ export function apply(ctx: Context, config: AcpConfig): void { logger.warn(`acp: continuable subagent teardown failed: ${String(error)}`) } } - await Promise.all(records.map(async (record) => { - settlePrompt(record, 'cancelled') - await record.dispose() - })) + await Promise.all(records.map(record => record.dispose())) })() return quiescing } diff --git a/packages/acp/acp/tests/dispose.spec.ts b/packages/acp/acp/tests/dispose.spec.ts index b7b38b0ffb..4be0810513 100644 --- a/packages/acp/acp/tests/dispose.spec.ts +++ b/packages/acp/acp/tests/dispose.spec.ts @@ -46,6 +46,34 @@ describe('ACP connection ownership', () => { expect(harness.ctx.agents.get(SessionId(sessionId))).toBeUndefined() }) + it('cancels its own prompt before awaiting the descendant drain', async () => { + harness = await makeBridgeHarness({ script: ['hang'] }) + const order: string[] = [] + const release = Promise.withResolvers() + harness.ctx.provide('subagents', { + drainContinuable: async () => { + order.push('drain started') + await release.promise + order.push('drain finished') + }, + } as never) + await harness.client.initialize({ protocolVersion: PROTOCOL_VERSION, clientCapabilities: {} }) + const { sessionId } = await harness.client.newSession({ cwd: process.cwd(), mcpServers: [] }) + const agent = harness.ctx.agents.get(SessionId(sessionId))! + void harness.client.prompt({ sessionId, prompt: [{ type: 'text', text: 'go' }] }).catch(() => {}) + await vi.waitFor(() => { expect(agent.status).toBe('running') }) + harness.ctx.on('agent/cancel-requested', () => { order.push('parent cancelled') }) + + const disposal = harness.acpFiber.dispose() + // A drain can block on persistence, so the bridge's own turn must already be + // cancelled rather than running for its whole duration. + await vi.waitFor(() => { expect(order).toContain('drain started') }) + expect(order).toEqual(['parent cancelled', 'drain started']) + release.resolve(undefined) + await disposal + expect(harness.ctx.agents.get(SessionId(sessionId))).toBeUndefined() + }) + it('reports a failed continuable drain and still disposes its sessions', async () => { harness = await makeBridgeHarness() const warnings: string[] = [] diff --git a/packages/subagent/subagent/src/continuation.ts b/packages/subagent/subagent/src/continuation.ts index b7e0d87274..9974a355a4 100644 --- a/packages/subagent/subagent/src/continuation.ts +++ b/packages/subagent/subagent/src/continuation.ts @@ -108,17 +108,26 @@ export type ActivationState = 'running' | 'waiting' | 'settled' * children emit the same start/end pair as one-shot runs. */ export interface ActivationObserver { - /** Publish the start edge once the epoch is resident. */ - start(): void /** - * Publish the terminal edge exactly once, pairing this epoch's {@link start}. - * Called only for a resident epoch: a failure before residency publishes no - * edge at all, because inventing one would report a lifecycle the child never - * had. - * @param child - the child agent whose final output the edge reports. + * Publish the start edge once the epoch is resident. + * @param child - the resident child agent, whose log suffix bounds this epoch. + */ + start(child: Agent): void + /** + * Snapshot the child-dependent terminal facts while the child is still + * registered, because handle disposal unregisters it and consumers resolve it + * to read the child's own log and scope. + * @param child - the quiescent child agent about to be released. + */ + capture(child: Agent): void + /** + * Publish the terminal edge exactly once, pairing this epoch's {@link start}, + * after the disposal outcome is known. Called only for a resident epoch: a + * failure before residency publishes no edge, because inventing one would + * report a lifecycle the child never had. * @param failure - the teardown or durability failure, or `undefined` on success. */ - settle(child: Agent, failure: unknown): void + settle(failure: unknown): void } /** Hooks the manager needs from the owning service. */ @@ -585,7 +594,7 @@ export class SubagentContinuationManager { }) // Resident: publish the start edge before any turn can run, so observers // see this epoch before its first request. - observer.start() + observer.start(handle.agent) this.watchSettlement(activation) return activation } @@ -778,12 +787,10 @@ export class SubagentContinuationManager { await activation.handle.agent.whenIdle() const durability = await this.checkpoint(activation) failure ??= durability - // Publish the terminal edge while the child is STILL registered: - // consumers resolve `ctx.agents.get(info.id)` in `subagent/end` to run - // in the child's own cwd and scope, which handle disposal removes. - activation.observer.settle(activation.handle.agent, failure) + // Capture the child-dependent edge data while the child is still live: + // handle disposal unregisters it, and consumers read its log and scope. + activation.observer.capture(activation.handle.agent) } finally { - this.activations.delete(childId) try { await activation.handle.dispose() } catch (error: unknown) { @@ -793,9 +800,16 @@ export class SubagentContinuationManager { { cause: error }, ) } finally { + // Only now is the Activation gone: keeping the entry until disposal + // settles makes a racing delivery wait for release rather than + // cold-resume into the still-registered agent. + this.activations.delete(childId) // Release ownership even on failure: a retained failed child would // pin its ancestors in `waiting` forever. this.releaseOwnership(childId) + // Emit once the disposal outcome is known, so a rejecting scoped + // cleanup cannot be reported as a successful epoch. + activation.observer.settle(failure) } } if (failure !== undefined) throw failure diff --git a/packages/subagent/subagent/src/index.ts b/packages/subagent/subagent/src/index.ts index 03bea866a5..a5e120c860 100644 --- a/packages/subagent/subagent/src/index.ts +++ b/packages/subagent/subagent/src/index.ts @@ -363,22 +363,40 @@ export class SubagentService extends Service { parent: Agent | undefined, ): ActivationObserver { const identity = { runId: SubagentRunId(randomUUID()), provider, id: childId, local: true } + // A cold resume replays earlier turns, so this epoch's telemetry must come + // from the suffix it actually produced — never the whole session, which + // would report a previous epoch's answer when this one opened no turn. + let boundary = 0 + // Assigned by `capture()`, which the disposal path always runs before + // `settle()`; a resident epoch therefore always has its facts by then. + let captured: { stopReason: SubagentResult['stopReason']; output?: ContentBlock[] } = { + stopReason: 'completed', + } let settled = false return { - start: (): void => { + start: (child: Agent): void => { + boundary = child.session.events.length this.emitLifecycle('subagent/start', identity, parent) }, - settle: (child: Agent, failure: unknown): void => { + capture: (child: Agent): void => { + const own = child.session.events.slice(boundary) + const output = lastAssistantOutput(own) + captured = { + stopReason: epochStopReason(own), + ...output === undefined ? {} : { output }, + } + }, + settle: (failure: unknown): void => { // Exactly one terminal edge per epoch: host shutdown, manager unload, // child release, and normal settlement all converge on one disposal. /* v8 ignore next -- the memoized disposal already collapses those callers into a * single settle(); this guard keeps the edge single if that memoization ever changes. */ if (settled) return settled = true - const output = failure === undefined ? lastAssistantOutput(child) : undefined + const output = failure === undefined ? captured.output : undefined this.emitLifecycle('subagent/end', { ...identity, - stopReason: failure === undefined ? childStopReason(child) : 'error', + stopReason: failure === undefined ? captured.stopReason : 'error', ...output === undefined ? {} : { lastAssistantMessage: output }, }, parent) }, @@ -465,11 +483,11 @@ export class SubagentService extends Service { * The child's own `turn/end` is authoritative: teardown succeeding says nothing * about whether the model errored, hit its token ceiling, or was cancelled, so * deriving the reason from disposal would report failed work as completed. - * @param child - the settling child agent whose log is read. + * @param events - this epoch's own event suffix. * @returns its terminal stop reason; `completed` when no ordinary turn closed. */ -function childStopReason(child: Agent): SubagentResult['stopReason'] { - const reason = findLastMessageTurnEnd(child.session.events)?.data.reason +function epochStopReason(events: readonly SessionEvent[]): SubagentResult['stopReason'] { + const reason = findLastMessageTurnEnd(events)?.data.reason // No ordinary turn closed, so nothing failed either. if (reason === undefined) return 'completed' switch (reason.kind) { @@ -494,11 +512,11 @@ function childStopReason(child: Agent): SubagentResult['stopReason'] { /** * The child's last assistant message content, for one Activation's terminal * lifecycle edge. Absent when no assistant message reached the log. - * @param child - the settling child agent whose log is read. + * @param events - this epoch's own event suffix. * @returns its final assistant content, or `undefined` when it produced none. */ -function lastAssistantOutput(child: Agent): ContentBlock[] | undefined { - const message = child.session.events.findLast( +function lastAssistantOutput(events: readonly SessionEvent[]): ContentBlock[] | undefined { + const message = events.findLast( (event): event is SessionEvent<'assistant/message'> => event.type === 'assistant/message', ) return message?.data.message.content diff --git a/packages/subagent/subagent/tests/continuation.spec.ts b/packages/subagent/subagent/tests/continuation.spec.ts index feb18d196a..c20cc175a9 100644 --- a/packages/subagent/subagent/tests/continuation.spec.ts +++ b/packages/subagent/subagent/tests/continuation.spec.ts @@ -685,19 +685,71 @@ describe('continuable review regressions', () => { expect(before).toBeGreaterThan(0) }) - it('publishes the terminal edge while the child agent is still resolvable', async () => { - const { ctx, parent } = await setup([textResponse('answer')]) - const resolvable: boolean[] = [] - // Consumers resolve the child in `subagent/end` to run in its own cwd. - ctx.on('subagent/end', (info) => { - resolvable.push(ctx.agents.get(info.id) !== undefined) - }) + it('reports this epoch\'s own output, captured while the child was still live', async () => { + const { ctx, parent } = await setup([textResponse('first answer'), textResponse('second answer')]) + const ends: SubagentRunEndInfo[] = [] + ctx.on('subagent/end', (info) => { ends.push(info) }) const started = await ctx.subagents.startContinuable(startSpec(parent)) await waitNoActivation(ctx, started.childId) + await vi.waitFor(() => { expect(ends).toHaveLength(1) }) + // Handle disposal unregisters the child, so the edge's content must have + // been captured before that — an after-the-fact lookup would find nothing. + expect(ends[0]!.lastAssistantMessage).toEqual([{ type: 'text', text: 'first answer' }]) - await vi.waitFor(() => { expect(resolvable).toHaveLength(1) }) - expect(resolvable[0]).toBe(true) + // A cold resume is a new epoch: it must report its OWN answer, never the + // previous epoch's, which the replayed transcript still contains. + await followup(ctx, { kind: 'user' }, started.childId, message('again')) + await waitNoActivation(ctx, started.childId) + await vi.waitFor(() => { expect(ends).toHaveLength(2) }) + expect(ends[1]!.lastAssistantMessage).toEqual([{ type: 'text', text: 'second answer' }]) + }) + + it('reports a resumed epoch that opened no turn without the previous answer', async () => { + const { ctx, parent } = await setup([textResponse('first answer')]) + const started = await ctx.subagents.startContinuable(startSpec(parent)) + await waitNoActivation(ctx, started.childId) + + const ends: SubagentRunEndInfo[] = [] + ctx.on('subagent/end', (info) => { ends.push(info) }) + // Block the resumed prompt so this epoch produces nothing of its own. + ctx.on('agent/prompt-submit', async (subject, _content, _source, _signal, next) => { + if (subject === parent) return next() + return { kind: 'block', reason: 'blocked by policy' } + }) + await followup(ctx, { kind: 'user' }, started.childId, message('again')) + await waitNoActivation(ctx, started.childId) + + await vi.waitFor(() => { expect(ends).toHaveLength(1) }) + // Reading the whole session would resurrect 'first answer' here. + expect(ends[0]!.lastAssistantMessage).toBeUndefined() + expect(ends[0]!.stopReason).toBe('completed') + }) + + it('reports handle-disposal failure on the terminal edge', async () => { + const { ctx, parent } = await setup([textResponse('answer')]) + const ends: SubagentRunEndInfo[] = [] + ctx.on('subagent/end', (info) => { ends.push(info) }) + + const started = await ctx.subagents.startContinuable(startSpec(parent)) + const manager = (ctx.subagents as unknown as { + continuations: { activations: Map Promise } }> } + }).continuations + const activation = await vi.waitFor(() => { + const found = manager.activations.get(started.childId) + expect(found).toBeDefined() + return found! + }) + const realDispose = activation.handle.dispose.bind(activation.handle) + activation.handle.dispose = async () => { + await realDispose() + throw new Error('scoped cleanup failed') + } + + await expect(ctx.subagents.drainContinuable()).rejects.toThrow() + await vi.waitFor(() => { expect(ends).toHaveLength(1) }) + // Emitting before disposal would have reported this failed epoch as success. + expect(ends[0]!.stopReason).toBe('error') }) it('cancels a running turn before the final durability checkpoint', async () => {