fix(subagent): close final continuation races

packages/subagent/subagent/README.i18n.yaml

@@ -2,5 +2,5 @@
 # side as of the last confirmed-consistent state. Both languages carry equal authority;
 # after editing either side, bring the other along and re-record with:
 #   pnpm run verify-translation-pairing --write packages/subagent/subagent/README.md
-README.md: 0e59a1ad5f256de4d6505d3d00d3790d7738a457
-README.zh.md: 073b4903520544e1b5b9209f792aa5e05d9334b0
+README.md: 06047ac87e84d50d8dc1a965c7d2499cbe58076d
+README.zh.md: 206a7d6e95f61ab152829e614cfaf1d15c5bec33

packages/subagent/subagent/README.md

@@ -31,11 +31,11 @@ Multiple providers may coexist under different names. This lets a deployment exp
 | `start(name, request)` | Validate an ordinary caller request, then await the provider until a real one-shot child is ready. Fulfillment returns a holder-owned `SubagentRun`; rejection means the provider has already cleaned every partial startup resource. Continuable children never enter through this operation. |
 | `startContinuable(spec)` | Establish one durable continuable child and deliver its initial prompt. Resolves with `{ childId, messageId }` when the child's inbox accepts that prompt, without waiting for the turn to start or for the message to reach the Session log; any earlier failure rejects with no ids and rolls the child back entirely. Requires `ctx.agents`, session persistence, and a provider with the `prepareContinuable` capability. |
 | `followup(parent, childId, content, { source, signal })` | Deliver one later message from the exact live direct parent as the child's next FIFO turn, matching `Agent.followup()` terminology, and return the accepted `MessageId`. A resident child's inbox accepts it directly (waking a waiting Activation); an absent one cold-resumes from its persisted Session. Requires `ctx.agents`; cold resume also requires session persistence. |
-| `drainContinuable()` | Close continuable admission synchronously, then dispose every live Activation forest child-first. A host calls this before disposing top-level agents so no descendant outlives the runtime that owns its teardown. An aggregate error surfaces after every branch settles when any failed. |
+| `drainContinuable()` | Close continuable admission synchronously, await every materialization that already passed admission through publication or rollback, then dispose the stable live Activation forest child-first. A host calls this before disposing top-level agents so no descendant outlives the runtime that owns its teardown. An aggregate error surfaces after every branch settles when any failed. |
 
 `SubagentStartRequest.signal` is required and is the canonical cancellation channel for a one-shot `start`. An abort before publication makes `start()` reject after rollback; an abort after publication cancels the live child. The request may also select a model, require structured output, cap delegation depth, restrict child tools, or set a child persona. For a continuable start or follow-up, the caller signal owns lookup, materialization, and admission only until inbox acceptance; afterward the manager owns the Activation independently, so later caller cancellation neither cancels the accepted turn nor disposes the child.
 
-Follow-up authority comes from the exact live direct parent recorded in the child's durable header. The `source` on a follow-up is durable provenance retained on the delivered message and grants no authority.
+Follow-up authority comes from the exact live direct parent recorded in the child's durable header. Cold resume checks that authority before reconstruction and again in the final no-await inbox-admission span, so a parent unregistered or replaced during materialization cannot authorize delivery. The `source` on a follow-up is durable provenance retained on the delivered message and grants no authority.
 
 Same-process requests, descriptors, results, and event payloads are trusted typed values borrowed as immutable. The service does not clone or freeze them; serialization and hostile-input validation belong at actual process, worker, persistence, and model boundaries.
 

packages/subagent/subagent/README.zh.md

@@ -31,11 +31,11 @@ subagent seam 允许一个 agent（智能体）通过具名提供方把工作委
 | `start(name, request)` | 校验普通调用方请求，然后等待提供方，直到真实的一次性子 agent 就绪。兑现时返回由持有方拥有的 `SubagentRun`；拒绝表示提供方已清理所有局部启动资源。可继续子 agent 绝不通过此操作进入。 |
 | `startContinuable(spec)` | 建立一个持久化可继续子 agent，并投递其初始提示词。子 agent 的 inbox 接受该提示词时，兑现为 `{ childId, messageId }`，无需等待轮次开始或消息写入 Session 日志；此前任何失败都会以无 id 拒绝，并完全回滚该子 agent。要求 `ctx.agents`、会话持久化以及具备 `prepareContinuable` 能力的提供方。 |
 | `followup(parent, childId, content, { source, signal })` | 将来自确切在线直接父级的一条后续消息作为子 agent 的下一个 FIFO 轮次投递，术语与 `Agent.followup()` 一致，并返回被接受的 `MessageId`。驻留中的子 agent 由其 inbox 直接接受（唤醒处于 waiting 的 Activation）；不驻留的则从其持久化 Session 冷恢复。要求 `ctx.agents`；冷恢复还要求会话持久化。 |
-| `drainContinuable()` | 同步关闭可继续准入，然后以子先于父的顺序 dispose 每一个实时 Activation 森林。host 会在 dispose 顶层 agent 之前调用它，使任何后代都不会比拥有其拆卸职责的运行时存活更久。任一分支失败时，会在所有分支结算后抛出聚合错误。 |
+| `drainContinuable()` | 同步关闭可继续准入，等待每个已经通过准入的物化过程完成发布或回滚，然后按 child-first 顺序 dispose 稳定的在线 Activation 森林。host 会在 dispose 顶层 agent 之前调用它，使任何后代都不会比拥有其拆卸职责的运行时存活更久。任一分支失败时，会在所有分支结算后抛出聚合错误。 |
 
 `SubagentStartRequest.signal` 是必填项，也是一次性 `start` 的规范取消通道。发布前中止会使 `start()` 在回滚后拒绝；发布后中止会取消实时子 agent。请求还可以选择模型、要求结构化输出、限制委派深度、约束子 agent 工具或设置子 agent persona。对于可继续启动或后续操作，调用方信号只在 inbox 接受之前掌管查找、物化和准入；此后由管理器独立拥有 Activation，因此调用方后续取消既不会取消已接受的轮次，也不会 dispose 子 agent。
 
-后续操作的权限来自子 agent 持久化 header 中记录的确切在线直接父级。后续操作上的 `source` 是保留在所投递消息上的持久化来源，不授予任何权限。
+后续操作的权限来自子 agent 持久化 header 中记录的确切在线直接父级。冷恢复会在重建前检查该权限，并在最终无 await 的 inbox 准入区间再次检查，因此在物化期间被注销或替换的 parent 无法授权投递。后续操作上的 `source` 是保留在所投递消息上的持久化来源，不授予任何权限。
 
 同进程请求、描述符、结果和事件 payload 都是以不可变方式借用的可信类型值。服务不会克隆或冻结它们；序列化和不可信输入校验属于真实的进程、worker、持久化和模型边界。
 

packages/subagent/subagent/src/continuation.ts

@@ -156,11 +156,22 @@ interface Activation {
   poke: PromiseWithResolvers<void>
 }
 
+/** Inputs shared by fresh and resumed Activation materialization. */
+interface MaterializeInputs {
+  childId: SessionId
+  provider: string
+  parent: Agent
+  /** Creation inputs; absent for a cold resume, which loads the persisted session. */
+  create?: { seed: readonly SessionEvent[]; meta: NonNullable<CreateAgentOptions['meta']> }
+  agentOptions: AgentOptions
+  composition: { persona?: string | undefined; toolFilter?: ToolRestriction | undefined }
+  signal: AbortSignal
+}
+
 /**
  * Read one Activation's current disposal transaction. This indirection exists
- * because a mutable field read inside a long-lived closure narrows to its
- * last-seen value, which would flatten these genuine runtime checks to
- * constants.
+ * because TypeScript would otherwise narrow repeated reads of the mutable field
+ * inside a long-lived closure to constants instead of re-reading runtime state.
  * @param activation - the Activation to inspect.
  * @returns the in-flight or settled disposal, or `undefined` while resident.
  */
@@ -206,6 +217,8 @@ class ChildLock {
 export class SubagentContinuationManager {
   /** Child session id → its live Activation. Process-local, never durable. */
   private activations = new Map<SessionId, Activation>()
+  /** Materializations admitted before drain, tracked through publication or rollback. */
+  private readonly materializations = new Set<Promise<void>>()
   private readonly locks = new ChildLock()
   /** Structural Cordis owner of every Activation handle. */
   private readonly ownerCtx: Context
@@ -332,7 +345,6 @@ export class SubagentContinuationManager {
         if (activation.disposal !== undefined) {
           return activation.disposal.then(() => undefined, () => undefined)
         }
-        this.authorizeLive(parent, activation)
         return this.submitAdmitted(activation, content, options.source, parent, options.signal)
       })
       /* v8 ignore start -- only the lost-cutoff arm above returns undefined, so only that
@@ -345,17 +357,20 @@ export class SubagentContinuationManager {
   }
 
   /**
-   * Dispose every live Activation forest child-first and await all handles.
-   * Sibling branches drain independently: one failure is recorded but never
-   * prevents the remaining handles from being attempted, and the aggregate
-   * rejects only after every branch settles.
-   * @returns once every snapshotted Activation released its handle.
+   * Close admission, await every already-admitted materialization through
+   * publication or rollback, then dispose the stable live Activation forest
+   * child-first. Sibling branches drain independently: one failure is recorded
+   * but never prevents the remaining handles from being attempted, and the
+   * aggregate rejects only after every branch settles.
+   * @returns once materialization is quiescent and every live Activation released its handle.
    * @throws an aggregate error when any branch failed to release.
    */
   async drain(): Promise<void> {
-    // Close admission synchronously before the first await, so no new creation,
-    // cold resume, or delivery can race the snapshot below.
+    // Close admission synchronously before the first await. Materializations
+    // already past that cutoff remain tracked until their handle is installed
+    // or rollback completes, producing a stable forest for the later snapshot.
     this.draining = true
+    await Promise.all([...this.materializations])
     // Snapshot roots after closing admission: a root is an Activation no live
     // Activation owns, so disposing roots recurses child-first into the forest.
     const owned = new Set<SessionId>()
@@ -489,16 +504,22 @@ export class SubagentContinuationManager {
    * a continuation-managed parent. Rejection leaves no Activation, no handle,
    * and no ownership membership.
    */
-  private async materialize(inputs: {
-    childId: SessionId
-    provider: string
-    parent: Agent
-    /** Creation inputs; absent for a cold resume, which loads the persisted session. */
-    create?: { seed: readonly SessionEvent[]; meta: NonNullable<CreateAgentOptions['meta']> }
-    agentOptions: AgentOptions
-    composition: { persona?: string | undefined; toolFilter?: ToolRestriction | undefined }
-    signal: AbortSignal
-  }): Promise<Activation> {
+  private materialize(inputs: MaterializeInputs): Promise<Activation> {
+    this.assertAdmitting()
+    const settled = Promise.withResolvers<void>()
+    this.materializations.add(settled.promise)
+    return this.materializeTracked(inputs).finally(() => {
+      this.materializations.delete(settled.promise)
+      settled.resolve()
+    })
+  }
+
+  /**
+   * Perform one tracked materialization. The caller keeps the drain barrier
+   * registered until this either returns a resident Activation or finishes
+   * rollback.
+   */
+  private async materializeTracked(inputs: MaterializeInputs): Promise<Activation> {
     const { childId, provider, parent } = inputs
     // No id pre-check here: the child lock serializes each durable child, both
     // callers reach this only after confirming no Activation exists, and
@@ -673,19 +694,12 @@ export class SubagentContinuationManager {
         'ACTIVATION_CLOSING',
       )
     }
-    return this.submit(activation, content, source, parent)
-  }
-
-  /**
-   * Authorize delivery to a live Activation. A parent must be the exact live
-   * direct parent recorded in the child's durable header.
-   */
-  private authorizeLive(parent: Agent, activation: Activation): void {
     this.authorizeLineage(
       parent,
       activation.childId,
       activation.handle.agent.session.header.parentSession,
     )
+    return this.submit(activation, content, source, parent)
   }
 
   /**

packages/subagent/subagent/tests/continuation.spec.ts

@@ -691,6 +691,29 @@ describe('continuable durability and teardown', () => {
     expect(ctx.agents.list()).toEqual([parent])
   })
 
+  it('waits for a published materialization to finish rollback before drain resolves', async () => {
+    const { ctx, parent } = await setup([])
+    const order: string[] = []
+    const drains: Promise<void>[] = []
+    ctx.on('agent/created', (child) => {
+      if (child === parent) return
+      const draining = ctx.subagents.drainContinuable().then(() => { order.push('drain') })
+      drains.push(draining)
+    })
+    ctx.on('agent/disposed', (child) => {
+      if (child !== parent) order.push('disposed')
+    })
+
+    // `agent/created` runs after registry publication but before materialize()
+    // receives the handle and installs the Activation.
+    await expect(ctx.subagents.startContinuable(startSpec(parent)))
+      .rejects.toMatchObject({ code: 'DRAINING' })
+    await Promise.all(drains)
+
+    expect(order).toEqual(['disposed', 'drain'])
+    expect(ctx.agents.list()).toEqual([parent])
+  })
+
   it('admits a live follow-up before a later drain can begin disposal', async () => {
     const hold = Promise.withResolvers<undefined>()
     const adapter = new GatedAdapter([{ chunks: textResponse('working'), gate: hold.promise }])
@@ -739,6 +762,80 @@ describe('continuable durability and teardown', () => {
 })
 
 describe('continuable review regressions', () => {
+  it('rechecks exact parent liveness after cold-resume materialization', async () => {
+    const { ctx } = await setup([textResponse('first')])
+    const parentId = SessionId('replaceable-parent')
+    const originalParent = await ctx.agents.create({
+      sessionId: parentId,
+      agentOptions: { provider: 'mock', model: 'mock' },
+    })
+    const started = await ctx.subagents.startContinuable(startSpec(originalParent.agent))
+    await waitNoActivation(ctx, started.childId)
+
+    const manager = (ctx.subagents as unknown as {
+      continuations: { ownerCtx: Context }
+    }).continuations
+    const ownerAgents = manager.ownerCtx.agents
+    const originalResume = ownerAgents.resume.bind(ownerAgents)
+    const resumed = Promise.withResolvers<undefined>()
+    const releaseResume = Promise.withResolvers<undefined>()
+    const resumeSpy = vi.spyOn(ownerAgents, 'resume').mockImplementation(async (options) => {
+      const handle = await originalResume(options)
+      resumed.resolve(undefined)
+      await releaseResume.promise
+      return handle
+    })
+
+    const delivery = followup(
+      ctx,
+      originalParent.agent,
+      started.childId,
+      message('must not cross parent replacement'),
+    )
+    await resumed.promise
+    await originalParent.dispose()
+    const replacement = await ctx.agents.create({
+      sessionId: parentId,
+      agentOptions: { provider: 'mock', model: 'mock' },
+    })
+    releaseResume.resolve(undefined)
+
+    await expect(delivery).rejects.toMatchObject({ code: 'UNAUTHORIZED' })
+    resumeSpy.mockRestore()
+    await waitNoActivation(ctx, started.childId)
+    const loaded = await ctx.sessionPersistence.load(started.childId)
+    expect(hasUserText(loaded.events, 'must not cross parent replacement')).toBe(false)
+    await replacement.dispose()
+  })
+
+  it('clears the accepted reservation when Agent.followup throws', async () => {
+    const hold = Promise.withResolvers<undefined>()
+    const adapter = new GatedAdapter([{ chunks: textResponse('working'), gate: hold.promise }])
+    const { ctx, parent } = await setupWith(adapter)
+    const started = await ctx.subagents.startContinuable(startSpec(parent))
+    await vi.waitFor(() => { expect(adapter.requests).toHaveLength(1) })
+    const child = ctx.agents.get(started.childId)!
+    const manager = (ctx.subagents as unknown as {
+      continuations: {
+        activations: Map<SessionId, { accepted: Set<MessageId> }>
+      }
+    }).continuations
+    const activation = manager.activations.get(started.childId)!
+    const realFollowup = child.followup.bind(child)
+    child.followup = () => {
+      throw new Error('synthetic inbox failure')
+    }
+
+    await expect(followup(ctx, parent, started.childId, message('throws')))
+      .rejects.toThrow(/synthetic inbox failure/)
+    expect(activation.accepted.size).toBe(0)
+
+    child.followup = realFollowup
+    const drained = ctx.subagents.drainContinuable()
+    hold.resolve(undefined)
+    await drained
+  })
+
   it('reports the child\'s own terminal reason, not teardown success', async () => {
     // The child hits its token ceiling; teardown still succeeds.
     const { ctx, parent } = await setupWith(new MockAdapter([