fix(subagent): address codex review round 2

Round 1 traded one teardown ordering problem for another. The observer now
splits capture from emission, which satisfies both consumers at once:

- Terminal facts are captured while the child is still registered, so consumers
  that resolve it for the child's log and scope still work.
- The edge is emitted only after handle disposal settles, so a rejecting scoped
  cleanup is reported as a failed epoch instead of a successful one.

Also:

- Keep the Activation in the map until disposal settles. Removing it first let a
  racing followup() see no Activation and cold-resume into the still-registered
  agent, and let a concurrent forest drain skip a still-disposing child and
  release its parent first.
- Derive terminal telemetry from this epoch's event suffix rather than the whole
  session, so a cold resume whose prompt is blocked no longer reports the
  previous epoch's answer and turn reason.
- Cancel the ACP bridge's own prompts before awaiting the descendant drain: a
  drain can block on persistence, and the top-level agents must not keep running
  model and tool work for its whole duration.

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) => {
+      await Promise.all(records.map(record => record.dispose()))
-        settlePrompt(record, 'cancelled')
-        await record.dispose()
-      }))
     })()
     return quiescing
   }

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<undefined>()
+    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[] = []

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}.
+   * Publish the start edge once the epoch is resident.
-   * Called only for a resident epoch: a failure before residency publishes no
+   * @param child - the resident child agent, whose log suffix bounds this epoch.
-   * edge at all, because inventing one would report a lifecycle the child never
+   */
-   * had.
+  start(child: Agent): void
-   * @param child - the child agent whose final output the edge reports.
+  /**
+   * 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:
+        // Capture the child-dependent edge data while the child is still live:
-        // consumers resolve `ctx.agents.get(info.id)` in `subagent/end` to run
+        // handle disposal unregisters it, and consumers read its log and scope.
-        // in the child's own cwd and scope, which handle disposal removes.
+        activation.observer.capture(activation.handle.agent)
-        activation.observer.settle(activation.handle.agent, failure)
       } 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

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'] {
+function epochStopReason(events: readonly SessionEvent[]): SubagentResult['stopReason'] {
-  const reason = findLastMessageTurnEnd(child.session.events)?.data.reason
+  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 {
+function lastAssistantOutput(events: readonly SessionEvent[]): ContentBlock[] | undefined {
-  const message = child.session.events.findLast(
+  const message = events.findLast(
     (event): event is SessionEvent<'assistant/message'> => event.type === 'assistant/message',
   )
   return message?.data.message.content

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 () => {
+  it('reports this epoch\'s own output, captured while the child was still live', async () => {
-    const { ctx, parent } = await setup([textResponse('answer')])
+    const { ctx, parent } = await setup([textResponse('first answer'), textResponse('second answer')])
-    const resolvable: boolean[] = []
+    const ends: SubagentRunEndInfo[] = []
-    // Consumers resolve the child in `subagent/end` to run in its own cwd.
+    ctx.on('subagent/end', (info) => { ends.push(info) })
-    ctx.on('subagent/end', (info) => {
-      resolvable.push(ctx.agents.get(info.id) !== undefined)
-    })
 
     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) })
+    // A cold resume is a new epoch: it must report its OWN answer, never the
-    expect(resolvable[0]).toBe(true)
+    // 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<SessionId, { handle: { dispose: () => Promise<void> } }> }
+    }).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 () => {