fix(acp): scope connection-owned continuation drain
This commit is contained in:
Dudu-0223
Tianyi Cui
+2
-2
@@ -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 .agents/notes/implemented/feature/2026-07-28-continuable-subagent-conversations.md
|
||||
2026-07-28-continuable-subagent-conversations.md: ec194314d88958becde4672a08570fd1facacb3c
|
||||
2026-07-28-continuable-subagent-conversations.zh.md: 7c776a4e34e16c1cfd56c8964f25b4f4176001dd
|
||||
2026-07-28-continuable-subagent-conversations.md: 8e867254a726c57200936cff6c83831abb4c66ad
|
||||
2026-07-28-continuable-subagent-conversations.zh.md: 3383a0b8fd7b17ad40c3a03f32a78df058afb120
|
||||
@@ -107,9 +107,9 @@ Child release occurs only after the child Agent is quiescent, every child of tha
|
||||
|
||||
Ownership is retained until the child Activation is disposed. A later refinement may release a request-scoped lease earlier, but it would require an exact turn-completion correlation that this Task-free proposal deliberately does not add.
|
||||
|
||||
Top-level teardown is host-owned rather than represented as another Activation. The host first asks the manager to enter draining synchronously, which rejects new creation, resume, and delivery admission. The manager then awaits every materialization that already passed admission until it either installs a resident Activation or completes rollback, snapshots the stable live forest, disposes it child-first, and awaits all `AgentHandle.dispose()` calls. Every materialized start and live delivery rechecks caller cancellation, draining, Activation disposal, and exact parent authority in the same synchronous span as inbox submission, so teardown or parent replacement that wins before acceptance prevents delivery to the closing handle. Only after that drain settles may the host dispose top-level Agents and the manager scope. Manager unload uses the same drain.
|
||||
Top-level teardown is host-owned rather than represented as another Activation. Manager unload uses `drainContinuable()` to close manager-wide admission synchronously, await every admitted materialization through publication or rollback, stop the stable live forest, and release it child-first. A host that owns selected top-level Agents instead uses `drainContinuableDescendants(parents)`: exact Agent identities close admission only below those roots until each leaves the registry, while unrelated forests and manager-wide admission remain live; the manager stops their visible descendants before its first await, waits only materializations admitted below those roots, and releases only the selected branches. Every materialized start and live delivery rechecks caller cancellation, the applicable draining scope, Activation disposal, and exact parent authority in the same synchronous span as inbox submission, so teardown or parent replacement that wins before acceptance prevents delivery to the closing handle. Only after the applicable drain settles may the host dispose its top-level Agents; only the global drain precedes manager-scope disposal.
|
||||
|
||||
The activation-owner scope exists because ordinary Cordis owner effects unwind in reverse registration order, which cannot express the dynamic child graph. Manager initialization registers the private scope's structural disposer first and its drain disposer afterward, so reverse unwind invokes the drain before releasing that scope; merely registering a cleanup effect on the same scope as later Agent handles would allow structural handle disposal to bypass child-first ordering. The manager snapshots the live roots after closing admission, stops its outward lifecycle notifications before cancellation, and retains its internal ownership bookkeeping until every handle settles. Each Activation has one memoized disposal promise so host shutdown, manager unload, child release, and normal settlement can converge without double release. Sibling branches drain independently; one disposal failure is recorded but does not prevent the manager from attempting the remaining handles, and the aggregate drain reports failure after all branches settle. Durable child Sessions survive this process-local teardown.
|
||||
The activation-owner scope exists because ordinary Cordis owner effects unwind in reverse registration order, which cannot express the dynamic child graph. Manager initialization registers the private scope's structural disposer first and its drain disposer afterward, so reverse unwind invokes the drain before releasing that scope; merely registering a cleanup effect on the same scope as later Agent handles would allow structural handle disposal to bypass child-first ordering. Each materialization registers its barrier participant and snapshots its exact live ancestry before starting the inner transaction, then remains tracked until it installs an Activation or fully rolls back. The Activation retains weak membership of that ancestry, so an intermediate Agent may leave the registry without hiding a still-live descendant from its host root. Each Activation installs one memoized disposal promise before cancellation or recursive callbacks, allowing scoped host shutdown, global manager unload, child release, and normal settlement to converge without double release. Cancellation propagates top-down before slow descendant cleanup; handle release remains child-first. Sibling branches drain independently; one disposal failure is recorded but does not prevent the manager from attempting the remaining selected handles, and the aggregate drain reports failure after all selected branches settle. Durable child Sessions survive this process-local teardown.
|
||||
|
||||
### Deferred report delivery
|
||||
|
||||
@@ -135,7 +135,7 @@ Parent-originated delivery requires the parent to be live when admitted and keep
|
||||
|
||||
Without Tasks there is no `task_output`, `task_kill`, Task status, per-message result promise, or public subagent cancellation operation. The caller signal can abort start or follow-up only before inbox acceptance. After acceptance, the parent cannot cancel the message, turn, or Activation through `ctx.subagents`; `Agent.cancel()` remains a lower-level Agent capability that this version does not expose through the subagent service.
|
||||
|
||||
Host and manager teardown remains the lifecycle-wide stop path. It closes admission, disposes every live Activation forest child-first, and preserves the durable Sessions.
|
||||
Host and manager teardown remains the lifecycle stop path. Manager unload applies it globally; a host applies it only below the exact top-level Agents it owns. Each form closes the applicable admission scope, stops the selected visible Activations, awaits admitted materializations in that scope, releases child-first, and preserves the durable Sessions.
|
||||
|
||||
Each turn requests the Session durability checkpoint, and final Activation settlement requires the manager to inspect `ctx.sessions.flush()` rather than ignore its boolean result. `true` confirms that at least one durability listener participated and every listener settled successfully. `false` or rejection reports `DURABILITY_FAILED`; normal background settlement logs the lifecycle failure, while an explicit host or manager drain includes it in the aggregate rejection after all branches settle. Either way, the manager still disposes the handle and releases ownership, and the persisted child state may be missing or stale on a later resume.
|
||||
|
||||
@@ -188,17 +188,17 @@ The implementation pins these behaviors:
|
||||
- `followup()` accepts only the exact live direct parent and rechecks that identity at the final no-await inbox-admission boundary after any materialization; durable message provenance cannot authorize delivery.
|
||||
- Continuation messages always use `Agent.followup()` and share its inbox FIFO, including when the child already has an open turn.
|
||||
- `ctx.subagents.followup()` and its `send_message` adapter return only the accepted `MessageId`; the continuation layer accepts no delivery target and defines no subagent-specific route result.
|
||||
- This version exposes no public subagent cancellation operation; caller signals stop start and follow-up only before inbox acceptance, while host and manager teardown retains child-first global cleanup.
|
||||
- This version exposes no public subagent cancellation operation; caller signals stop start and follow-up only before inbox acceptance, while host-scoped and manager-global teardown retain child-first cleanup.
|
||||
- This version exposes no subagent steering operation or current-turn controller state.
|
||||
- An idle Agent with live owned children yields a `waiting` Activation whose `AgentHandle` remains retained.
|
||||
- A `next-turn` delivered to `waiting` wakes the same Activation; delivery after completed disposal cold-resumes a new Activation.
|
||||
- Every continuation-managed parent Activation disposes only after all directly owned child Activations complete `AgentHandle` disposal; top-level Agents do not join the waiting graph.
|
||||
- Final Activation settlement treats only `ctx.sessions.flush(child.session) === true` as durability confirmation; `false` and rejection report `DURABILITY_FAILED`, still dispose the child handle, and still release parent ownership so durability failure cannot leak a `waiting` Activation.
|
||||
- Host and manager teardown synchronously enter draining, reject new materialization and delivery, await every admitted materialization through publication or rollback, stop manager-owned outward notifications, dispose the stable live Activation forest child-first, await every branch despite individual failures, and only then dispose top-level Agents and the manager scope; a private activation-owner scope preserves this order against Cordis effect unwinding, and one memoized disposal promise per Activation makes concurrent normal settlement idempotent.
|
||||
- Manager teardown closes admission globally; a host owning selected top-level Agents instead closes admission only below their exact identities until those roots leave the registry. Both track admitted materializations by exact ancestry, install one memoized disposal cutoff per selected visible Activation, propagate cancellation top-down, release handles child-first, await every selected branch despite individual failures, and only then dispose the corresponding top-level Agents or manager scope.
|
||||
- This version exposes no `report` tool, child-to-parent content delivery, or automatic parent wakeup.
|
||||
- Session logs reconstruct only messages that were actually written, with their admitted provenance; inbox-accepted but unlogged messages have no restart guarantee.
|
||||
- No continuable-subagent path creates or depends on a Task, `TaskId`, Task completion notice, Task cancellation, or intermediate result-bearing execution wrapper.
|
||||
- Unit coverage pins the `startContinuable()` inbox-acceptance return boundary, complete rollback for each pre-acceptance and lifecycle-publication failure, drain quiescence for a materialization caught between Agent publication and Activation registration, provider-independent cold resume, final exact-parent reauthorization after cold-resume materialization, caller-signal and teardown ownership on both sides of acceptance, and the absence of automatic replay for accepted-but-unlogged messages.
|
||||
- Unit coverage pins the `startContinuable()` inbox-acceptance return boundary, complete rollback for each pre-acceptance and lifecycle-publication failure, global and parent-scoped drain quiescence for materialization caught between Agent publication and Activation registration, sibling-forest isolation, exact ancestry after an intermediate Agent leaves the registry, provider-independent cold resume, final exact-parent reauthorization after cold-resume materialization, caller-signal and teardown ownership on both sides of acceptance, and the absence of automatic replay for accepted-but-unlogged messages.
|
||||
- Unit coverage pins the residency-only routing table, single-inbox ordering, `MessageId` correlation through inbox events, follow-up during an open turn, waiting wakeup, cold resume, ownership registration and release, child-first disposal, send-versus-dispose races, both `false` and rejection from the final durability checkpoint without ownership leaks, and the absence of public subagent cancellation, steering, and report tools.
|
||||
- A keyless assembled-app snapshot covers parent delegation and follow-up queueing, the absence of subagent steering, report delivery, and automatic parent wakeup, retained waiting `AgentHandle`, and child-first disposal.
|
||||
|
||||
|
||||
+6
-6
@@ -107,9 +107,9 @@ Agent inbox 是唯一队列。每条继续执行消息都使用 `Agent.followup(
|
||||
|
||||
系统会一直保留所有权，直至 child 激活完成 dispose。后续改进可以更早释放限定到请求的 lease，但这需要精确关联轮次完成，而本 Task-free 提案特意不增加该机制。
|
||||
|
||||
顶层拆卸由宿主负责，而不表示为另一次激活。宿主首先要求管理器同步进入 draining，拒绝新的创建、恢复和投递准入。管理器随后等待每个已经通过准入的物化过程，直至它安装驻留激活或完成回滚，再对稳定的在线森林创建快照，按 child-first 顺序 dispose，并等待全部 `AgentHandle.dispose()` 调用。每个已物化的 start 和在线投递都会在与 inbox 提交相同的同步区间内重新检查调用方取消、draining、激活 dispose 和确切的 parent 权限，因此只要拆卸或 parent 替换先于接受发生，就会阻止向正在关闭的 handle 投递。只有该 drain 结算后，宿主才能 dispose 顶层 Agent 和管理器作用域。管理器卸载使用相同的 drain。
|
||||
顶层拆卸由宿主负责，而不表示为另一次激活。管理器卸载使用 `drainContinuable()` 同步关闭管理器全局准入，等待每个已获准的物化过程完成发布或回滚，停止稳定的在线森林，并按 child-first 顺序释放。拥有选定顶层 Agent 的宿主则使用 `drainContinuableDescendants(parents)`：确切的 Agent 身份只关闭这些根之下的准入，直到每个身份离开注册表，而无关森林和管理器全局准入保持在线；管理器会在第一次 await 之前停止其可见后代，只等待这些根之下已获准的物化过程，并且只释放选定分支。每个已物化的 start 和在线投递都会在与 inbox 提交相同的同步区间内重新检查调用方取消、适用的 draining 作用域、Activation dispose 和确切的 parent 权限，因此只要拆卸或 parent 替换先于接受发生，就会阻止向正在关闭的 handle 投递。只有适用的 drain 结算后，宿主才能 dispose 自己的顶层 Agent；只有全局 drain 会先于管理器作用域 dispose。
|
||||
|
||||
activation-owner 作用域之所以存在，是因为普通 Cordis owner effect 按注册逆序撤销，无法表达动态 child 图。管理器初始化时先注册私有作用域的结构化 disposer，再注册自身的 drain disposer，使逆序撤销先执行 drain、再释放该作用域；如果只在与后续 Agent handle 相同的作用域上注册 cleanup effect，结构化 handle dispose 就可能绕过 child-first 顺序。管理器在关闭准入后对在线根节点创建快照，在取消前停止自身的对外生命周期通知，并保留内部所有权簿记，直至每个 handle 都结算。每次激活有一个记忆化的 dispose promise，使宿主关闭、管理器卸载、child 释放和正常结算能够汇合，而不会重复释放。同级分支独立 drain；系统会记录单次 dispose 失败，但仍会尝试其余 handle，聚合 drain 则在所有分支结算后报告失败。这次进程内拆卸不会销毁持久化 child 会话。
|
||||
activation-owner 作用域之所以存在，是因为普通 Cordis owner effect 按注册逆序撤销，无法表达动态 child 图。管理器初始化时先注册私有作用域的结构化 disposer，再注册自身的 drain disposer，使逆序撤销先执行 drain、再释放该作用域；如果只在与后续 Agent handle 相同的作用域上注册 cleanup effect，结构化 handle dispose 就可能绕过 child-first 顺序。每个物化过程都会在启动内部事务前注册其屏障参与项，并对其确切的在线祖先建立快照，然后保持跟踪，直到安装 Activation 或完全回滚。Activation 会以弱引用方式记录其属于这组祖先，因此中间 Agent 即使离开注册表，也不会让仍在线的后代脱离宿主根节点的可见范围。每个 Activation 都会在取消或递归回调前安装一个记忆化的 dispose promise，使限定作用域的宿主关闭、全局管理器卸载、child 释放和正常结算能够汇合，而不会重复释放。取消会在等待缓慢的后代清理之前自顶向下传播；handle 释放仍是 child-first。同级分支独立 drain；系统会记录单次 dispose 失败，但仍会尝试其余选中 handle，聚合 drain 则在所有选中分支结算后报告失败。这次进程内拆卸不会销毁持久化 child 会话。
|
||||
|
||||
### 延后的报告投递
|
||||
|
||||
@@ -135,7 +135,7 @@ activation-owner 作用域之所以存在，是因为普通 Cordis owner effect
|
||||
|
||||
没有 Task 后，系统不再提供 `task_output`、`task_kill`、Task 状态、逐消息结果 promise 或公开 subagent 取消操作。调用方 signal 只能在 inbox 接受消息前中止 start 或 follow-up。消息被接受后，parent 不能通过 `ctx.subagents` 取消该消息、轮次或激活；`Agent.cancel()` 仍是底层 Agent 能力，但本版本不通过 subagent 服务暴露它。
|
||||
|
||||
宿主和管理器拆卸仍是覆盖整个生命周期的停止路径。它会关闭准入，按 child-first 顺序 dispose 每个在线激活森林，并保留持久化会话。
|
||||
宿主和管理器拆卸仍是生命周期停止路径。管理器卸载会全局应用它；宿主只会在自己确切拥有的顶层 Agent 之下应用它。两种形式都会关闭适用的准入作用域，停止选中的可见 Activation，等待该作用域中已获准的物化过程，按 child-first 顺序释放，并保留持久化 Session。
|
||||
|
||||
每个轮次都会请求执行会话持久性检查点，激活最终结算时，管理器必须检查 `ctx.sessions.flush()`，而不能忽略其布尔结果。`true` 确认至少有一个持久性 listener 参与，且所有 listener 都成功结算。`false` 或 rejection 会报告 `DURABILITY_FAILED`；普通后台结算会记录该生命周期失败，显式的宿主或管理器 drain 则会在所有分支结算后，将其纳入聚合 rejection。无论结果如何，管理器仍会 dispose handle 并释放所有权，后续恢复时持久化 child 状态可能缺失或陈旧。
|
||||
|
||||
@@ -188,17 +188,17 @@ activation-owner 作用域之所以存在，是因为普通 Cordis owner effect
|
||||
- `followup()` 只接受确切的在线直接 parent，并在任何物化之后的最终无 await 的 inbox 准入边界再次检查该身份；持久化消息来源信息不能授权投递。
|
||||
- 继续执行消息始终使用 `Agent.followup()` 并共享其 inbox FIFO，包括 child 已有开放轮次的情况。
|
||||
- `ctx.subagents.followup()` 及其 `send_message` 适配器只返回已接受的 `MessageId`；继续执行层不接受投递 target，也不定义 subagent 专属路由结果。
|
||||
- 本版本不暴露公开 subagent 取消操作；调用方 signal 只能在 inbox 接受消息前停止 start 和 follow-up，宿主和管理器拆卸则保留 child-first 全局清理。
|
||||
- 本版本不暴露公开 subagent 取消操作；调用方 signal 只能在 inbox 接受消息前停止 start 和 follow-up，限定到宿主的拆卸与管理器全局拆卸则保留 child-first 清理。
|
||||
- 本版本不暴露 subagent steering 操作或当前轮次控制方状态。
|
||||
- 带有在线所持 child 的空闲 Agent 会产生 `waiting` 激活，其 `AgentHandle` 继续保留。
|
||||
- 向 `waiting` 投递 `next-turn` 会唤醒同一个激活；完成 dispose 后投递消息会冷恢复新激活。
|
||||
- 每个由继续执行管理器管理的 parent 激活只会在直接持有的所有 child 激活完成 `AgentHandle` dispose 后进行 dispose；顶层 Agent 不加入等待图。
|
||||
- 激活最终结算时，只有 `ctx.sessions.flush(child.session) === true` 才确认持久性；`false` 和 rejection 会报告 `DURABILITY_FAILED`，但仍会 dispose child handle 并释放 parent 所有权，使持久性失败不会泄漏 `waiting` 激活。
|
||||
- 宿主和管理器拆卸会同步进入 draining，拒绝新的物化和投递，等待每个已获准的物化过程完成发布或回滚，停止由管理器负责的对外通知，按 child-first 顺序 dispose 稳定的在线激活森林，即使个别分支失败也会等待所有分支，之后才 dispose 顶层 Agent 和管理器作用域；私有 activation-owner 作用域会确保 Cordis effect 的逆序撤销不破坏该顺序，每次激活使用一个记忆化的 dispose promise，使并发的正常结算保持幂等。
|
||||
- 管理器拆卸会全局关闭准入；拥有选定顶层 Agent 的宿主则只关闭这些确切身份之下的准入，直到这些根离开注册表。两者都会按确切祖先关系跟踪已获准的物化过程，为每个选中的可见 Activation 安装一个记忆化 dispose 截止点，自顶向下传播取消，按 child-first 顺序释放 handle，即使个别分支失败也会等待所有选中分支，之后才 dispose 对应的顶层 Agent 或管理器作用域。
|
||||
- 本版本不暴露 `report` 工具，不提供从 child 到 parent 的内容投递，也不自动唤醒 parent。
|
||||
- 会话日志只能根据准入来源重建实际写入的消息；已被 inbox 接受但未写入日志的消息没有重启保证。
|
||||
- 可继续 subagent 路径不创建或依赖 Task、`TaskId`、Task 完成通知、Task 取消或中间的带结果执行包装层。
|
||||
- 单元覆盖固定 `startContinuable()` 在 inbox 接受消息时的返回边界、每条接受前和生命周期发布失败路径的完整回滚、drain 会等待夹在 Agent 发布与 Activation 注册之间的物化过程完全停稳、不依赖提供方的冷恢复、冷恢复物化后的最终确切 parent 再授权、接受前后两个阶段的调用方 signal 与拆卸所有权，以及已接受但未写入日志的消息不会自动回放。
|
||||
- 单元覆盖固定 `startContinuable()` 在 inbox 接受消息时的返回边界、每条接受前和生命周期发布失败路径的完整回滚、全局和限定到 parent 作用域的 drain 都会等待夹在 Agent 发布与 Activation 注册之间的物化过程完全停稳、同级森林隔离、中间 Agent 离开注册表后的确切祖先关系、不依赖提供方的冷恢复、冷恢复物化后的最终确切 parent 再授权、接受前后两个阶段的调用方 signal 与拆卸所有权，以及已接受但未写入日志的消息不会自动回放。
|
||||
- 单元覆盖固定仅由驻留状态决定的路由表、单 inbox 顺序、通过 inbox 事件关联 `MessageId`、在开放轮次期间 follow-up、等待唤醒、冷恢复、所有权注册与释放、child-first dispose、发送与 dispose 的竞争、最终持久性检查点返回 `false` 和 rejection 时都不泄漏所有权，以及不存在公开 subagent 取消、steering 和报告工具这一事实。
|
||||
- 一项无密钥整套应用快照覆盖 parent 委派和 follow-up 排队、不存在 subagent steering、报告投递和自动唤醒 parent、保留等待中的 `AgentHandle` 以及 child-first dispose。
|
||||
|
||||
|
||||
@@ -27,7 +27,7 @@ export interface AcpConfig {
|
||||
|
||||
Depends on: `Stream` (`@agentclientprotocol/sdk`)
|
||||
|
||||
Source: [`packages/acp/acp/src/index.ts:67`](../packages/acp/acp/src/index.ts)
|
||||
Source: [`packages/acp/acp/src/index.ts:70`](../packages/acp/acp/src/index.ts)
|
||||
|
||||
## `@deepseek-ai/dsh-acp-demo`
|
||||
|
||||
|
||||
@@ -1989,6 +1989,18 @@ async followup( parent: Agent, childId: SessionId, content: ContentBlock[], opti
|
||||
*/
|
||||
async drainContinuable(): Promise<void>
|
||||
|
||||
/**
|
||||
* Close continuable admission below exact live parent Agents, stop only their
|
||||
* visible descendant Activations synchronously, then await admitted scoped
|
||||
* materializations and release those forests child-first. The scoped cutoff
|
||||
* lasts until each exact parent leaves the registry; unrelated parent trees
|
||||
* remain live.
|
||||
* @param parents - exact host-owned parent Agents entering teardown.
|
||||
* @returns once every retained descendant Activation released its `AgentHandle`.
|
||||
* @throws an aggregate error after all scoped branches settle when any failed.
|
||||
*/
|
||||
async drainContinuableDescendants(parents: readonly Agent[]): Promise<void>
|
||||
|
||||
/**
|
||||
* Register a provider under its name. Registration is effect-scoped and HMR
|
||||
* safe; removing a provider blocks new starts but does not revoke runs that
|
||||
|
||||
@@ -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 docs/core-data-structures/subagent.md
|
||||
subagent.md: e160c596acb55f0e94cba84b8c79355c966eb51a
|
||||
subagent.zh.md: 6b934a523fa0ea5d53ea9a670e56b72b7f785593
|
||||
subagent.md: eff78bae6fcf7440cce66f122771ff623a5eb3c7
|
||||
subagent.zh.md: a898f438c257db05a3d84e6318ce23e1f366e193
|
||||
@@ -129,7 +129,7 @@ For both operations the caller signal owns lookup, materialization, and admissio
|
||||
|
||||
Every Activation owns its `AgentHandle` and an `ownedChildren: Set<SessionId>`; because one Session has at most one live Activation, the child Session id identifies the live child without another runtime-incarnation reference. Starting a child or submitting parent-originated work registers the child in a continuation-managed parent's set before the child can run, and that parent cannot settle while the set is non-empty. A top-level or other non-continuation Agent has no Activation and stays outside the waiting graph. Child release happens only after the child Agent is quiescent, every child of that child is disposed, the final durability checkpoint settles, and the child's `AgentHandle` completes disposal.
|
||||
|
||||
Only `ctx.sessions.flush(session) === true` confirms durability; `false` or rejection reports `DURABILITY_FAILED`. Either way the manager still disposes the handle and releases ownership, because retaining a failed child would permanently pin its ancestors in `waiting` — the persisted child state may then be missing or stale on a later resume. `drainContinuable()` is the lifecycle-wide stop path: it closes admission synchronously, then disposes every live Activation forest child-first, awaiting every branch despite individual failures. Durable child Sessions survive that process-local teardown.
|
||||
Only `ctx.sessions.flush(session) === true` confirms durability; `false` or rejection reports `DURABILITY_FAILED`. Either way the manager still disposes the handle and releases ownership, because retaining a failed child would permanently pin its ancestors in `waiting` — the persisted child state may then be missing or stale on a later resume. `drainContinuable()` closes manager-wide admission and disposes every live forest; `drainContinuableDescendants(parents)` closes admission only below exact live host-owned Agents and disposes their continuable descendants while unrelated forests remain live. Both await already-admitted materializations in their scope, propagate cancellation top-down, release handles child-first, and await every selected branch despite individual failures. Durable child Sessions survive that process-local teardown.
|
||||
|
||||
```ts type-equiv
|
||||
/** Attribution for a model coordinator's follow-up to one of its children. */
|
||||
|
||||
@@ -129,7 +129,7 @@ Agent 收件箱是唯一的队列。每条继续执行消息都会成为一个 `
|
||||
|
||||
每个 Activation 都拥有自己的 `AgentHandle` 和一个 `ownedChildren: Set<SessionId>`；由于一份会话至多有一个存活 Activation，子会话 id 无需另一个运行时化身引用即可标识存活的子 agent。启动子 agent 或提交源自 parent 的工作，会在子 agent 能够运行之前将其注册到受继续执行管理的父级集合中；只要该集合非空，该父级就无法 settle。顶层或其他非继续执行的 Agent 没有 Activation，处于 waiting 图之外。只有当子 Agent 已停稳、该子 agent 的每个子级都已 dispose、最终的持久性检查点结算完毕，且子 agent 的 `AgentHandle` 完成 dispose 之后，才会释放子 agent。
|
||||
|
||||
只有 `ctx.sessions.flush(session) === true` 才确认持久性；`false` 或 rejection 会报告 `DURABILITY_FAILED`。无论哪种情况，管理器仍会 dispose 该 handle 并释放所有权，因为保留一个失败的子 agent 会将其祖先永久钉在 `waiting`——此后持久化的子 agent 状态在后续恢复时可能缺失或陈旧。`drainContinuable()` 是覆盖整个生命周期的停止路径：它同步关闭准入，随后以子级优先的方式 dispose 每一片存活的 Activation 森林，尽管个别分支失败仍会等待每个分支。持久化子会话不受该进程内拆卸的影响。
|
||||
只有 `ctx.sessions.flush(session) === true` 才确认持久性；`false` 或 rejection 会报告 `DURABILITY_FAILED`。无论哪种情况，管理器仍会 dispose 该 handle 并释放所有权，因为保留一个失败的子 agent 会将其祖先永久钉在 `waiting`——此后持久化的子 agent 状态在后续恢复时可能缺失或陈旧。`drainContinuable()` 会关闭管理器全局准入并 dispose 每片在线森林；`drainContinuableDescendants(parents)` 只关闭由 host 确切拥有的在线 Agent 之下的准入，并 dispose 其可继续后代，而无关森林保持在线。两者都会等待各自作用域内已获准的物化过程，自顶向下传播取消，按 child-first 顺序释放 handle，并且即使个别分支失败也会等待所有选中分支。持久化子会话不受该进程内拆卸的影响。
|
||||
|
||||
```ts type-equiv
|
||||
/** Attribution for a model coordinator's follow-up to one of its children. */
|
||||
|
||||
@@ -10,7 +10,7 @@ This matrix shows which packages dispatch each harness-owned event and which pac
|
||||
| `agent-loop/config-start-failed` | `emit` | [`packages/core/agent-loop/src/index.ts:157`](../packages/core/agent-loop/src/index.ts) | [`agent-loop`](../packages/core/agent-loop) (`events.dispatch`) | [`tui`](../packages/ui/tui) |
|
||||
| `agent/cancel-requested` | `emit` | [`packages/core/agent/src/types.ts:349`](../packages/core/agent/src/types.ts) | [`agent-loop`](../packages/core/agent-loop) (`emitAgentEvent`) | [`goal-session`](../packages/goal/goal-session) |
|
||||
| `agent/created` | `emit` | [`packages/core/agent/src/types.ts:280`](../packages/core/agent/src/types.ts) | [`agent`](../packages/core/agent) (`events.dispatch`) | [`goal-session`](../packages/goal/goal-session), [`tui`](../packages/ui/tui) |
|
||||
| `agent/disposed` | `emit` | [`packages/core/agent/src/types.ts:289`](../packages/core/agent/src/types.ts) | [`agent`](../packages/core/agent) (`events.dispatch`) | [`agent-loop`](../packages/core/agent-loop), [`goal-session`](../packages/goal/goal-session), [`tui`](../packages/ui/tui) |
|
||||
| `agent/disposed` | `emit` | [`packages/core/agent/src/types.ts:289`](../packages/core/agent/src/types.ts) | [`agent`](../packages/core/agent) (`events.dispatch`) | [`agent-loop`](../packages/core/agent-loop), [`goal-session`](../packages/goal/goal-session), [`subagent`](../packages/subagent/subagent), [`tui`](../packages/ui/tui) |
|
||||
| `agent/error` | `emit` | [`packages/core/agent/src/types.ts:463`](../packages/core/agent/src/types.ts) | [`agent-loop`](../packages/core/agent-loop) (`emitAgentEvent`) | `apiproxy`, [`goal-session`](../packages/goal/goal-session), [`session-telemetry`](../packages/telemetry/session-telemetry), [`tui`](../packages/ui/tui) |
|
||||
| `agent/inbox/dequeue` | `emit` | [`packages/core/agent/src/types.ts:327`](../packages/core/agent/src/types.ts) | [`agent-loop`](../packages/core/agent-loop) (`emitAgentEvent`) | [`agent`](../packages/core/agent), `apiproxy`, [`subagent`](../packages/subagent/subagent), [`tui`](../packages/ui/tui) |
|
||||
| `agent/inbox/discard` | `emit` | [`packages/core/agent/src/types.ts:339`](../packages/core/agent/src/types.ts) | [`agent-loop`](../packages/core/agent-loop) (`emitAgentEvent`) | [`agent`](../packages/core/agent), `apiproxy`, [`subagent`](../packages/subagent/subagent), [`tui`](../packages/ui/tui) |
|
||||
|
||||
@@ -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/acp/acp/README.md
|
||||
README.md: 1b188b994d17ce56e8d5df019ddef755338fcc88
|
||||
README.zh.md: c1e7d045b55119b62ad44d81071188e1ed6110d5
|
||||
README.md: 9a48fdec3330cd364c1ab6de4c117b20af0f443f
|
||||
README.zh.md: 65732f41277a8760bfd2824aea12b0f240ae8025
|
||||
@@ -35,7 +35,7 @@ Committed-message output intentionally trades token-by-token latency for a clean
|
||||
|
||||
## Lifecycle
|
||||
|
||||
Client disconnect and Cordis disposal share one memoized teardown. The bridge first rejects new sessions and prompts, settles pending prompts, then disposes all owned agent handles in parallel and awaits their loop/session cleanup. An ACP-only plugin reload therefore leaves no orphan agent.
|
||||
Client disconnect and Cordis disposal share one memoized teardown. The bridge first rejects new sessions and prompts, settles pending prompts, then drains continuable descendants only below this connection's exact owned Agents before disposing those handles in parallel and awaiting their loop/session cleanup. Other frontends sharing the Context retain their continuable forests and admission. An ACP-only plugin reload therefore leaves no orphan agent.
|
||||
|
||||
## Running
|
||||
|
||||
|
||||
@@ -35,7 +35,7 @@
|
||||
|
||||
## 生命周期
|
||||
|
||||
客户端断开连接与 Cordis 的 dispose（资源释放）共用同一个记忆化清理流程。桥接层先拒绝新会话和提示词，结算待处理提示词，然后并行对其拥有的全部 agent 句柄执行 dispose，并等待它们的循环／会话清理完成。因此，单独重载 ACP 插件不会遗留孤儿 agent。
|
||||
客户端断开与 Cordis 释放共用同一个记忆化清理流程。桥接层先拒绝新会话和提示词，结算待处理提示词，然后只 drain 此连接确切拥有的 Agent 之下的可继续后代，再并行释放这些 handle，并等待它们的循环／会话清理完成。其他共享该上下文的前端会保留其可继续森林和准入。因此，仅 ACP 的插件重载不会遗留 agent。
|
||||
|
||||
## 运行
|
||||
|
||||
|
||||
@@ -49,8 +49,11 @@ export const inject = ['agents']
|
||||
* shutdown hook; an absent service means nothing continuable was materialized.
|
||||
*/
|
||||
interface ContinuableDrain {
|
||||
/** Close continuable admission, then dispose every live Activation child-first. */
|
||||
drainContinuable(): Promise<void>
|
||||
/**
|
||||
* Close admission below exact host-owned parents, then dispose only their
|
||||
* continuable descendants child-first.
|
||||
*/
|
||||
drainContinuableDescendants(parents: readonly Agent[]): Promise<void>
|
||||
}
|
||||
|
||||
/** Preserve invalid-parameter detail in the SDK wire error message. */
|
||||
@@ -345,15 +348,16 @@ export function apply(ctx: Context, config: AcpConfig): void {
|
||||
}
|
||||
quiescing = (async () => {
|
||||
// Continuable subagents outlive the turn that started them, and their
|
||||
// Activations own descendant teardown. Drain that forest child-first
|
||||
// BEFORE disposing the top-level agents, so no descendant is left holding
|
||||
// a runtime its owner already released.
|
||||
// Activations own descendant teardown. Drain only these sessions' forests
|
||||
// child-first BEFORE disposing the top-level agents, so no descendant is
|
||||
// left holding a runtime its owner already released and another frontend
|
||||
// sharing this Context remains live.
|
||||
// Read the one teardown method structurally: the bridge needs no other
|
||||
// part of the subagent seam, so it does not depend on that package.
|
||||
const subagents = ctx.get('subagents') as ContinuableDrain | undefined
|
||||
if (subagents !== undefined) {
|
||||
try {
|
||||
await subagents.drainContinuable()
|
||||
await subagents.drainContinuableDescendants(records.map(record => record.agent))
|
||||
} catch (error: unknown) {
|
||||
logger.warn(`acp: continuable subagent teardown failed: ${String(error)}`)
|
||||
}
|
||||
|
||||
@@ -1,5 +1,6 @@
|
||||
import { afterEach, describe, expect, it, vi } from 'vitest'
|
||||
import { PROTOCOL_VERSION } from '@agentclientprotocol/sdk'
|
||||
import type { Agent } from '@deepseek-ai/dsh-agent'
|
||||
import { SessionId } from '@deepseek-ai/dsh-session'
|
||||
import { makeBridgeHarness, type BridgeHarness } from './harness.ts'
|
||||
|
||||
@@ -28,21 +29,25 @@ describe('ACP connection ownership', () => {
|
||||
it('drains continuable subagents before disposing its own sessions', async () => {
|
||||
harness = await makeBridgeHarness()
|
||||
const order: string[] = []
|
||||
let drainedParents: readonly Agent[] = []
|
||||
// A continuable Activation outlives the turn that started it, so the bridge
|
||||
// must release that forest before the agents whose runtime it depends on.
|
||||
harness.ctx.provide('subagents', {
|
||||
drainContinuable: () => {
|
||||
drainContinuableDescendants: (parents: readonly Agent[]) => {
|
||||
drainedParents = parents
|
||||
order.push('drained')
|
||||
return Promise.resolve()
|
||||
},
|
||||
} 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))!
|
||||
harness.ctx.on('agent/disposed', () => { order.push('agent disposed') })
|
||||
|
||||
await harness.acpFiber.dispose()
|
||||
|
||||
expect(order).toEqual(['drained', 'agent disposed'])
|
||||
expect(drainedParents).toEqual([agent])
|
||||
expect(harness.ctx.agents.get(SessionId(sessionId))).toBeUndefined()
|
||||
})
|
||||
|
||||
@@ -51,7 +56,7 @@ describe('ACP connection ownership', () => {
|
||||
const order: string[] = []
|
||||
const release = Promise.withResolvers<undefined>()
|
||||
harness.ctx.provide('subagents', {
|
||||
drainContinuable: async () => {
|
||||
drainContinuableDescendants: async () => {
|
||||
order.push('drain started')
|
||||
await release.promise
|
||||
order.push('drain finished')
|
||||
@@ -79,7 +84,7 @@ describe('ACP connection ownership', () => {
|
||||
const warnings: string[] = []
|
||||
harness.ctx.logger.warn = (message: string) => { warnings.push(message) }
|
||||
harness.ctx.provide('subagents', {
|
||||
drainContinuable: () => Promise.reject(new Error('activation teardown failed')),
|
||||
drainContinuableDescendants: () => Promise.reject(new Error('activation teardown failed')),
|
||||
} as never)
|
||||
await harness.client.initialize({ protocolVersion: PROTOCOL_VERSION, clientCapabilities: {} })
|
||||
const { sessionId } = await harness.client.newSession({ cwd: process.cwd(), mcpServers: [] })
|
||||
|
||||
@@ -896,6 +896,10 @@ export const SERVICE_API: readonly ServiceApiEntry[] = [
|
||||
signature: 'async drainContinuable(): Promise<void>',
|
||||
jsDoc: '/**\n * Close continuable admission synchronously, then dispose every live\n * Activation forest child-first. A host calls this before disposing top-level\n * agents so no descendant outlives the runtime that owns its teardown.\n * @returns once every live Activation released its `AgentHandle`.\n * @throws an aggregate error after all branches settle when any failed.\n */',
|
||||
},
|
||||
{
|
||||
signature: 'async drainContinuableDescendants(parents: readonly Agent[]): Promise<void>',
|
||||
jsDoc: '/**\n * Close continuable admission below exact live parent Agents, stop only their\n * visible descendant Activations synchronously, then await admitted scoped\n * materializations and release those forests child-first. The scoped cutoff\n * lasts until each exact parent leaves the registry; unrelated parent trees\n * remain live.\n * @param parents - exact host-owned parent Agents entering teardown.\n * @returns once every retained descendant Activation released its `AgentHandle`.\n * @throws an aggregate error after all scoped branches settle when any failed.\n */',
|
||||
},
|
||||
{
|
||||
signature: 'registerProvider(provider: SubagentProvider): () => void',
|
||||
jsDoc: '/**\n * Register a provider under its name. Registration is effect-scoped and HMR\n * safe; removing a provider blocks new starts but does not revoke runs that\n * were already returned to their holders.\n * @param provider - the trusted provider implementation.\n * @returns the exact Cordis effect disposer.\n */',
|
||||
|
||||
@@ -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: 06047ac87e84d50d8dc1a965c7d2499cbe58076d
|
||||
README.zh.md: 206a7d6e95f61ab152829e614cfaf1d15c5bec33
|
||||
README.md: 6fab6859e2c15fdb1ded023642cbc593e0457384
|
||||
README.zh.md: 1f59807a545dcb1fafbac3f301746c7217d15f3a
|
||||
@@ -32,6 +32,7 @@ Multiple providers may coexist under different names. This lets a deployment exp
|
||||
| `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, 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. |
|
||||
| `drainContinuableDescendants(parents)` | Close admission below exact live host-owned parent Agents, stop only their visible continuable descendants, await materializations admitted below those roots through publication or rollback, then release the selected forests child-first. The cutoff lasts until each exact parent leaves the registry; unrelated parent forests and manager-wide admission remain live. |
|
||||
|
||||
`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.
|
||||
|
||||
@@ -76,7 +77,7 @@ The manager derives three internal residency conditions from Agent quiescence an
|
||||
|
||||
The manager reserves the child identity, resolves the durable descriptor, calls `ctx.agents.create()` (or `ctx.agents.resume()` for cold resume) through a private activation-owner scope, installs the returned `AgentHandle` in the Activation, establishes any continuable-parent ownership, and then submits the prompt. Cold resume never dispatches through a provider because the persisted Session already holds the initial prefix and the folded descriptor is the whole reconstruction input.
|
||||
|
||||
A continuation-managed parent Activation records each child Session id in an `ownedChildren` set before the child can run and disposes only after every owned child Activation completes `AgentHandle` disposal (child-first). Top-level and other non-continuation Agents have no Activation and stay outside this waiting graph. Final settlement treats only `ctx.sessions.flush(child.session) === true` as durability confirmation; `false` or rejection reports `DURABILITY_FAILED` and still disposes the handle and releases ownership, because retaining a failed child would permanently pin its ancestors in `waiting`.
|
||||
A continuation-managed parent Activation records each child Session id in an `ownedChildren` set before the child can run and disposes only after every owned child Activation completes `AgentHandle` disposal (child-first). Teardown propagates Agent cancellation top-down before awaiting slow descendants, while handle release remains child-first. Top-level and other non-continuation Agents have no Activation and stay outside this waiting graph. Final settlement treats only `ctx.sessions.flush(child.session) === true` as durability confirmation; `false` or rejection reports `DURABILITY_FAILED` and still disposes the handle and releases ownership, because retaining a failed child would permanently pin its ancestors in `waiting`.
|
||||
|
||||
## Lifecycle events
|
||||
|
||||
|
||||
@@ -32,6 +32,7 @@ subagent seam 允许一个 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()` | 同步关闭可继续准入，等待每个已经通过准入的物化过程完成发布或回滚，然后按 child-first 顺序 dispose 稳定的在线 Activation 森林。host 会在 dispose 顶层 agent 之前调用它，使任何后代都不会比拥有其拆卸职责的运行时存活更久。任一分支失败时，会在所有分支结算后抛出聚合错误。 |
|
||||
| `drainContinuableDescendants(parents)` | 在由 host 确切拥有的在线 parent Agent 之下关闭准入，只停止其可见的可继续后代，等待在这些根之下已获准的物化过程完成发布或回滚，再按 child-first 顺序释放所选森林。该截止状态会持续到每个确切 parent 离开注册表；无关的 parent 森林和管理器全局准入保持在线。 |
|
||||
|
||||
`SubagentStartRequest.signal` 是必填项，也是一次性 `start` 的规范取消通道。发布前中止会使 `start()` 在回滚后拒绝；发布后中止会取消实时子 agent。请求还可以选择模型、要求结构化输出、限制委派深度、约束子 agent 工具或设置子 agent persona。对于可继续启动或后续操作，调用方信号只在 inbox 接受之前掌管查找、物化和准入；此后由管理器独立拥有 Activation，因此调用方后续取消既不会取消已接受的轮次，也不会 dispose 子 agent。
|
||||
|
||||
@@ -76,7 +77,7 @@ subagent seam 允许一个 agent（智能体）通过具名提供方把工作委
|
||||
|
||||
管理器预留子 agent 身份、解析持久化描述符，通过私有的 activation-owner 作用域调用 `ctx.agents.create()`（冷恢复时为 `ctx.agents.resume()`），把返回的 `AgentHandle` 安装到 Activation 中，建立任何可继续父级所有权，然后提交提示词。冷恢复绝不通过提供方分发，因为持久化 Session 已持有初始前缀，折叠后的描述符即是全部重建输入。
|
||||
|
||||
受继续执行管理的父级 Activation 会在子 agent 能够运行之前，把每个子 agent 的 Session id 记录到 `ownedChildren` 集合中，并且只有在每个所拥有的子 agent Activation 完成 `AgentHandle` dispose 之后才会 dispose（子先于父）。顶层及其他非继续执行的 Agent 没有 Activation，处于该等待图之外。最终结算只把 `ctx.sessions.flush(child.session) === true` 视为持久性确认；`false` 或拒绝会报告 `DURABILITY_FAILED`，但仍会 dispose 句柄并释放所有权，因为保留失败的子 agent 会使其祖先永久停留在 `waiting`。
|
||||
受继续执行管理的父级 Activation 会在子 agent 能够运行之前，把每个子 agent 的 Session id 记录到 `ownedChildren` 集合中，并且只有在每个所拥有的子 agent Activation 完成 `AgentHandle` dispose 之后才会 dispose（子先于父）。拆卸会先自顶向下传播 Agent 取消，再等待缓慢的后代，而 handle 释放仍保持 child-first。顶层及其他非继续执行的 Agent 没有 Activation，处于该等待图之外。最终结算只把 `ctx.sessions.flush(child.session) === true` 视为持久性确认；`false` 或拒绝会报告 `DURABILITY_FAILED`，但仍会 dispose 句柄并释放所有权，因为保留失败的子 agent 会使其祖先永久停留在 `waiting`。
|
||||
|
||||
## 生命周期事件
|
||||
|
||||
|
||||
@@ -131,6 +131,12 @@ interface Activation {
|
||||
readonly provider: string
|
||||
/** The retained live Agent handle, disposed exactly once at settlement. */
|
||||
readonly handle: AgentHandle
|
||||
/**
|
||||
* Exact live Agent ancestry observed when this Activation materialized.
|
||||
* Weak membership preserves host-scope identity across an intermediate
|
||||
* ancestor leaving the registry without retaining that ancestor's runtime.
|
||||
*/
|
||||
readonly ancestry: WeakSet<Agent>
|
||||
/**
|
||||
* Session ids of the child Activations this one owns. Because one Session has
|
||||
* at most one live Activation, the id identifies the live child without
|
||||
@@ -168,6 +174,16 @@ interface MaterializeInputs {
|
||||
signal: AbortSignal
|
||||
}
|
||||
|
||||
/**
|
||||
* One admitted materialization and the exact live ancestry observed at its
|
||||
* synchronous admission boundary. Retaining identities lets a scoped teardown
|
||||
* keep waiting even if an intermediate Agent leaves the registry meanwhile.
|
||||
*/
|
||||
interface Materialization {
|
||||
readonly lineage: readonly Agent[]
|
||||
readonly settled: Promise<void>
|
||||
}
|
||||
|
||||
/**
|
||||
* Read one Activation's current disposal transaction. This indirection exists
|
||||
* because TypeScript would otherwise narrow repeated reads of the mutable field
|
||||
@@ -218,10 +234,17 @@ 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 materializations = new Set<Materialization>()
|
||||
private readonly locks = new ChildLock()
|
||||
/** Structural Cordis owner of every Activation handle. */
|
||||
private readonly ownerCtx: Context
|
||||
/**
|
||||
* Exact roots whose host teardown has begun, with the live lineage members
|
||||
* observed under each root. Entries remain until that exact root leaves the
|
||||
* Agent registry, closing admission throughout its host's teardown without
|
||||
* poisoning a later same-id replacement.
|
||||
*/
|
||||
private readonly closingScopes = new Map<Agent, Set<Agent>>()
|
||||
private draining = false
|
||||
|
||||
constructor(
|
||||
@@ -236,6 +259,9 @@ export class SubagentContinuationManager {
|
||||
// child-first ordering.
|
||||
const scope = ctx.plugin(function activationOwner() {})
|
||||
this.ownerCtx = scope.ctx
|
||||
ctx.on('agent/disposed', (agent) => {
|
||||
this.closingScopes.delete(agent)
|
||||
})
|
||||
ctx.effect(function* (this: SubagentContinuationManager) {
|
||||
yield scope.dispose
|
||||
yield () => this.drain()
|
||||
@@ -258,10 +284,10 @@ export class SubagentContinuationManager {
|
||||
* @returns the durable child id and the accepted initial prompt's message id.
|
||||
*/
|
||||
async startContinuable(spec: ContinuableStartSpec): Promise<ContinuableStart> {
|
||||
this.assertAdmitting()
|
||||
this.requirePersistence()
|
||||
const request = spec.request
|
||||
const parent = request.parent
|
||||
this.assertAdmitting(parent)
|
||||
this.requirePersistence()
|
||||
assertSubagentMaxDepth(request.maxDepth)
|
||||
const childId = SessionId(randomUUID())
|
||||
const childDepth = resolveChildDepth(parent, request.maxDepth)
|
||||
@@ -283,7 +309,7 @@ export class SubagentContinuationManager {
|
||||
signal: spec.signal,
|
||||
})
|
||||
spec.signal.throwIfAborted()
|
||||
this.assertAdmitting()
|
||||
this.assertAdmitting(parent)
|
||||
|
||||
const lineageSeedLength = prepared.seed?.length ?? 0
|
||||
const seed = seedDescriptorTurn(childId, prepared.seed, descriptor)
|
||||
@@ -331,7 +357,7 @@ export class SubagentContinuationManager {
|
||||
content: ContentBlock[],
|
||||
options: SubagentFollowupOptions,
|
||||
): Promise<MessageId> {
|
||||
this.assertAdmitting()
|
||||
this.assertAdmitting(parent)
|
||||
while (true) {
|
||||
const live = await this.locks.run(childId, async () => {
|
||||
const activation = this.activations.get(childId)
|
||||
@@ -350,7 +376,7 @@ export class SubagentContinuationManager {
|
||||
/* v8 ignore start -- only the lost-cutoff arm above returns undefined, so only that
|
||||
* race reaches the retry below, which then cold-resumes a new Activation. */
|
||||
if (live !== undefined) return live
|
||||
this.assertAdmitting()
|
||||
this.assertAdmitting(parent)
|
||||
options.signal.throwIfAborted()
|
||||
/* v8 ignore stop */
|
||||
}
|
||||
@@ -370,7 +396,7 @@ export class SubagentContinuationManager {
|
||||
// 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])
|
||||
await Promise.all([...this.materializations].map(materialization => materialization.settled))
|
||||
// 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>()
|
||||
@@ -396,14 +422,128 @@ export class SubagentContinuationManager {
|
||||
}
|
||||
}
|
||||
|
||||
/** Reject new admission once the host or manager began draining. */
|
||||
private assertAdmitting(): void {
|
||||
/**
|
||||
* Stop only the continuable descendants of exact live host-owned parents.
|
||||
* Admission stays closed for those parent trees until each exact parent
|
||||
* leaves the Agent registry; unrelated trees and manager-wide admission stay
|
||||
* live.
|
||||
* @param parents - exact live roots whose continuable descendants must stop.
|
||||
* @returns once every retained descendant Activation released its handle.
|
||||
* @throws an aggregate error after all scoped branches settle when any failed.
|
||||
*/
|
||||
async drainDescendants(parents: readonly Agent[]): Promise<void> {
|
||||
const roots = new Set(parents.filter(parent => this.ctx.agents.get(parent.id) === parent))
|
||||
if (roots.size === 0) return
|
||||
|
||||
// Publish the scoped admission cutoff before the first await. Merge with an
|
||||
// earlier call for the same exact root so a converging drain cannot forget
|
||||
// descendants whose release is already in flight.
|
||||
for (const root of roots) {
|
||||
this.closingMembers(root).add(root)
|
||||
}
|
||||
|
||||
const targets: Activation[] = []
|
||||
for (const activation of this.activations.values()) {
|
||||
const lineage = this.liveLineage(activation.handle.agent)
|
||||
// Strict descendants only: a continuable Agent may itself be a
|
||||
// host-owned root, and its host remains responsible for that root handle.
|
||||
const owners = [...roots].filter(root => activation.handle.agent !== root
|
||||
&& activation.ancestry.has(root))
|
||||
if (owners.length === 0) continue
|
||||
targets.push(activation)
|
||||
for (const owner of owners) {
|
||||
const members = this.closingMembers(owner)
|
||||
members.add(activation.handle.agent)
|
||||
for (const agent of lineage) members.add(agent)
|
||||
}
|
||||
}
|
||||
const materializations = [...this.materializations].filter((materialization) => {
|
||||
const owners = [...roots].filter(root => materialization.lineage.includes(root))
|
||||
for (const owner of owners) {
|
||||
const members = this.closingMembers(owner)
|
||||
for (const agent of materialization.lineage) members.add(agent)
|
||||
}
|
||||
return owners.length > 0
|
||||
})
|
||||
|
||||
const ownedTargets = new Set<SessionId>()
|
||||
for (const activation of targets) {
|
||||
for (const child of activation.ownedChildren) ownedTargets.add(child)
|
||||
}
|
||||
const targetRoots = targets.filter(activation => !ownedTargets.has(activation.childId))
|
||||
|
||||
// Open every selected transaction before the materialization barrier.
|
||||
// Disposal propagates cancellation top-down in the same synchronous span;
|
||||
// handle release remains child-first.
|
||||
for (const activation of targets) {
|
||||
const disposal = this.dispose(activation)
|
||||
void disposal.catch(() => undefined)
|
||||
}
|
||||
|
||||
await Promise.all(materializations.map(materialization => materialization.settled))
|
||||
const failures = await Promise.all(targetRoots.map(async (activation) => {
|
||||
try {
|
||||
await this.dispose(activation)
|
||||
return undefined
|
||||
} catch (error: unknown) {
|
||||
return error
|
||||
}
|
||||
}))
|
||||
const reasons = failures.filter(failure => failure !== undefined)
|
||||
if (reasons.length > 0) {
|
||||
throw new SubagentError(
|
||||
`continuable subagent teardown failed for ${reasons.length} scoped activation(s): `
|
||||
+ reasons.map(reason => errorChain(reason)).join('; '),
|
||||
'ACTIVATION_TEARDOWN_FAILED',
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
/** Return the retained member set for one exact scoped-teardown root. */
|
||||
private closingMembers(root: Agent): Set<Agent> {
|
||||
const existing = this.closingScopes.get(root)
|
||||
if (existing !== undefined) return existing
|
||||
const members = new Set<Agent>()
|
||||
this.closingScopes.set(root, members)
|
||||
return members
|
||||
}
|
||||
|
||||
/**
|
||||
* Return the exact currently resolvable ancestry from `agent` upward. The
|
||||
* first element is always the supplied identity, even when it is already
|
||||
* stale; each ancestor after it must be the registry's current exact entry.
|
||||
*/
|
||||
private liveLineage(agent: Agent): Agent[] {
|
||||
const lineage = [agent]
|
||||
const seen = new Set<SessionId>([agent.id])
|
||||
let parentSession = agent.session.header.parentSession
|
||||
while (parentSession !== undefined) {
|
||||
const parent = this.ctx.agents.get(parentSession)
|
||||
if (parent === undefined || seen.has(parent.id)) break
|
||||
lineage.push(parent)
|
||||
seen.add(parent.id)
|
||||
parentSession = parent.session.header.parentSession
|
||||
}
|
||||
return lineage
|
||||
}
|
||||
|
||||
/** Reject new admission once the manager or this exact parent tree began draining. */
|
||||
private assertAdmitting(agent: Agent): void {
|
||||
if (this.draining) {
|
||||
throw new SubagentError(
|
||||
'continuable subagents are draining; the operation was not admitted',
|
||||
'DRAINING',
|
||||
)
|
||||
}
|
||||
const lineage = this.liveLineage(agent)
|
||||
for (const [root, members] of this.closingScopes) {
|
||||
if (members.has(agent) || lineage.includes(root)) {
|
||||
throw new SubagentError(
|
||||
`continuable subagents below parent "${root.id}" are draining; the operation was not admitted`,
|
||||
'DRAINING',
|
||||
)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -443,7 +583,7 @@ export class SubagentContinuationManager {
|
||||
}
|
||||
// The persistence seam takes no signal; recheck before any child work.
|
||||
options.signal.throwIfAborted()
|
||||
this.assertAdmitting()
|
||||
this.assertAdmitting(parent)
|
||||
// Authorize the persisted header before folding: only the durable child's
|
||||
// exact live direct parent may continue it.
|
||||
this.authorizeLineage(parent, childId, loaded.meta.parentSession)
|
||||
@@ -505,11 +645,16 @@ export class SubagentContinuationManager {
|
||||
* and no ownership membership.
|
||||
*/
|
||||
private materialize(inputs: MaterializeInputs): Promise<Activation> {
|
||||
this.assertAdmitting()
|
||||
this.assertAdmitting(inputs.parent)
|
||||
const settled = Promise.withResolvers<void>()
|
||||
this.materializations.add(settled.promise)
|
||||
return this.materializeTracked(inputs).finally(() => {
|
||||
this.materializations.delete(settled.promise)
|
||||
const lineage = this.liveLineage(inputs.parent)
|
||||
const materialization: Materialization = {
|
||||
lineage,
|
||||
settled: settled.promise,
|
||||
}
|
||||
this.materializations.add(materialization)
|
||||
return this.materializeTracked(inputs, lineage).finally(() => {
|
||||
this.materializations.delete(materialization)
|
||||
settled.resolve()
|
||||
})
|
||||
}
|
||||
@@ -519,7 +664,10 @@ export class SubagentContinuationManager {
|
||||
* registered until this either returns a resident Activation or finishes
|
||||
* rollback.
|
||||
*/
|
||||
private async materializeTracked(inputs: MaterializeInputs): Promise<Activation> {
|
||||
private async materializeTracked(
|
||||
inputs: MaterializeInputs,
|
||||
parentLineage: readonly Agent[],
|
||||
): 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
|
||||
@@ -551,6 +699,7 @@ export class SubagentContinuationManager {
|
||||
childId,
|
||||
provider,
|
||||
handle,
|
||||
ancestry: new WeakSet([handle.agent, ...parentLineage]),
|
||||
ownedChildren: new Set(),
|
||||
observer,
|
||||
disposal: undefined,
|
||||
@@ -562,7 +711,7 @@ export class SubagentContinuationManager {
|
||||
this.activations.set(childId, activation)
|
||||
try {
|
||||
inputs.signal.throwIfAborted()
|
||||
this.assertAdmitting()
|
||||
this.assertAdmitting(parent)
|
||||
this.acquireOwnership(parent, childId)
|
||||
// Every accepted id leaves the inbox exactly once, through dequeue or
|
||||
// discard. Clearing it there is what lets `stateOf()` distinguish a truly
|
||||
@@ -685,7 +834,7 @@ export class SubagentContinuationManager {
|
||||
signal: AbortSignal,
|
||||
): MessageId {
|
||||
signal.throwIfAborted()
|
||||
this.assertAdmitting()
|
||||
this.assertAdmitting(parent)
|
||||
/* v8 ignore next 6 -- only a synchronous re-entrant disposer can change
|
||||
* this field between the caller's live check and this no-await boundary. */
|
||||
if (disposalOf(activation) !== undefined) {
|
||||
@@ -767,83 +916,100 @@ export class SubagentContinuationManager {
|
||||
}
|
||||
|
||||
/**
|
||||
* Release one Activation child-first: dispose owned children, checkpoint
|
||||
* durability, dispose the handle, and release parent ownership. Memoized, so
|
||||
* host shutdown, manager unload, child release, and normal settlement
|
||||
* converge on one teardown.
|
||||
* Stop one Activation immediately, then release it child-first. The memoized
|
||||
* transaction is installed before cancellation or recursive callbacks, so
|
||||
* admission and reentrant teardown converge on the same owner.
|
||||
*
|
||||
* A failed final checkpoint is reported but never prevents handle disposal or
|
||||
* ownership release, because retaining a failed child would permanently pin
|
||||
* its ancestors in `waiting`.
|
||||
* @param activation - the residency epoch to stop and release.
|
||||
* @returns the one disposal transaction owned by this Activation.
|
||||
*/
|
||||
private dispose(activation: Activation): Promise<void> {
|
||||
return (activation.disposal ??= (async () => {
|
||||
// The memoized assignment above already closed admission for this child:
|
||||
// no caller may send to a handle after its disposal transaction begins.
|
||||
this.wake(activation)
|
||||
const { childId } = activation
|
||||
let failure: Error | undefined
|
||||
try {
|
||||
// Child-first: every owned child must complete disposal before this
|
||||
// handle is released.
|
||||
const children = [...activation.ownedChildren]
|
||||
.map(child => this.activations.get(child))
|
||||
.filter((child): child is Activation => child !== undefined)
|
||||
const childFailures = await Promise.all(children.map(async (child) => {
|
||||
try {
|
||||
await this.dispose(child)
|
||||
return undefined
|
||||
} catch (error: unknown) {
|
||||
return error
|
||||
}
|
||||
}))
|
||||
const reasons = childFailures.filter(reason => reason !== undefined)
|
||||
if (reasons.length > 0) {
|
||||
failure = new SubagentError(
|
||||
`subagent "${childId}" child teardown failed: ${reasons.map(reason => errorChain(reason)).join('; ')}`,
|
||||
'ACTIVATION_TEARDOWN_FAILED',
|
||||
)
|
||||
const existing = activation.disposal
|
||||
if (existing !== undefined) return existing
|
||||
const completion = Promise.withResolvers<void>()
|
||||
// Presence is the admission cutoff. Assign it before the async helper starts
|
||||
// because that helper cancels Agents and may synchronously re-enter callers.
|
||||
activation.disposal = completion.promise
|
||||
void this.finishDisposal(activation).then(completion.resolve, completion.reject)
|
||||
return completion.promise
|
||||
}
|
||||
|
||||
/**
|
||||
* Propagate stop synchronously, then finish the child-first release.
|
||||
* @param activation - the Activation whose disposal transaction is installed.
|
||||
* @returns once the handle and ownership edge are released.
|
||||
*/
|
||||
private async finishDisposal(activation: Activation): Promise<void> {
|
||||
this.wake(activation)
|
||||
const { childId } = activation
|
||||
// Stop top-down before the first await. Slow descendant cleanup may delay
|
||||
// release, but it cannot let this ancestor continue model or tool work.
|
||||
activation.handle.agent.cancel({ kind: 'parent' })
|
||||
const idle = activation.handle.agent.whenIdle()
|
||||
const children = [...activation.ownedChildren]
|
||||
.map(child => this.activations.get(child))
|
||||
.filter((child): child is Activation => child !== undefined)
|
||||
const childDisposals = children.map(child => this.dispose(child))
|
||||
|
||||
let failure: Error | undefined
|
||||
try {
|
||||
// Release remains child-first even though cancellation propagated
|
||||
// top-down: every owned child completes before this handle is removed.
|
||||
const childFailures = await Promise.all(childDisposals.map(async (disposal) => {
|
||||
try {
|
||||
await disposal
|
||||
return undefined
|
||||
} catch (error: unknown) {
|
||||
return error
|
||||
}
|
||||
// Quiesce before the checkpoint: a turn still running would keep
|
||||
// appending events the flush cannot cover, and a slow flush would let
|
||||
// model and tool work continue for the whole shutdown.
|
||||
activation.handle.agent.cancel({ kind: 'parent' })
|
||||
await activation.handle.agent.whenIdle()
|
||||
const durability = await this.checkpoint(activation)
|
||||
failure ??= durability
|
||||
// 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)
|
||||
}))
|
||||
const reasons = childFailures.filter(reason => reason !== undefined)
|
||||
if (reasons.length > 0) {
|
||||
failure = new SubagentError(
|
||||
`subagent "${childId}" child teardown failed: ${reasons.map(reason => errorChain(reason)).join('; ')}`,
|
||||
'ACTIVATION_TEARDOWN_FAILED',
|
||||
)
|
||||
}
|
||||
// Quiesce before the checkpoint: a turn still running would keep
|
||||
// appending events the flush cannot cover.
|
||||
await idle
|
||||
const durability = await this.checkpoint(activation)
|
||||
failure ??= durability
|
||||
// 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)
|
||||
} catch (error: unknown) {
|
||||
failure ??= new SubagentError(
|
||||
`subagent "${childId}" activation teardown failed: ${errorChain(error)}`,
|
||||
'ACTIVATION_TEARDOWN_FAILED',
|
||||
{ cause: error },
|
||||
)
|
||||
} finally {
|
||||
try {
|
||||
await activation.handle.dispose()
|
||||
} catch (error: unknown) {
|
||||
failure ??= new SubagentError(
|
||||
`subagent "${childId}" activation teardown failed: ${errorChain(error)}`,
|
||||
`subagent "${childId}" activation handle disposal failed: ${errorChain(error)}`,
|
||||
'ACTIVATION_TEARDOWN_FAILED',
|
||||
{ cause: error },
|
||||
)
|
||||
} finally {
|
||||
try {
|
||||
await activation.handle.dispose()
|
||||
} catch (error: unknown) {
|
||||
failure ??= new SubagentError(
|
||||
`subagent "${childId}" activation handle disposal failed: ${errorChain(error)}`,
|
||||
'ACTIVATION_TEARDOWN_FAILED',
|
||||
{ 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)
|
||||
}
|
||||
// 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
|
||||
})())
|
||||
}
|
||||
if (failure !== undefined) throw failure
|
||||
}
|
||||
|
||||
/**
|
||||
|
||||
@@ -215,6 +215,23 @@ export class SubagentService extends Service {
|
||||
await manager.drain()
|
||||
}
|
||||
|
||||
/**
|
||||
* Close continuable admission below exact live parent Agents, stop only their
|
||||
* visible descendant Activations synchronously, then await admitted scoped
|
||||
* materializations and release those forests child-first. The scoped cutoff
|
||||
* lasts until each exact parent leaves the registry; unrelated parent trees
|
||||
* remain live.
|
||||
* @param parents - exact host-owned parent Agents entering teardown.
|
||||
* @returns once every retained descendant Activation released its `AgentHandle`.
|
||||
* @throws an aggregate error after all scoped branches settle when any failed.
|
||||
*/
|
||||
async drainContinuableDescendants(parents: readonly Agent[]): Promise<void> {
|
||||
const manager = this.continuations
|
||||
// Absent continuation services means nothing was ever materialized.
|
||||
if (manager === undefined) return
|
||||
await manager.drainDescendants(parents)
|
||||
}
|
||||
|
||||
/**
|
||||
* Register a provider under its name. Registration is effect-scoped and HMR
|
||||
* safe; removing a provider blocks new starts but does not revoke runs that
|
||||
|
||||
@@ -663,6 +663,199 @@ describe('continuable durability and teardown', () => {
|
||||
expect(loaded.meta.id).toBe(started.childId)
|
||||
})
|
||||
|
||||
it('drains one parent forest without disabling a sibling parent forest', async () => {
|
||||
const releaseTarget = Promise.withResolvers<undefined>()
|
||||
const releaseGrandchild = Promise.withResolvers<undefined>()
|
||||
const releaseSibling = Promise.withResolvers<undefined>()
|
||||
const adapter = new GatedAdapter([
|
||||
{ chunks: textResponse('target child'), gate: releaseTarget.promise },
|
||||
{ chunks: textResponse('sibling child'), gate: releaseSibling.promise },
|
||||
{ chunks: textResponse('target grandchild'), gate: releaseGrandchild.promise },
|
||||
{ chunks: textResponse('sibling follow-up') },
|
||||
])
|
||||
const { ctx, parent } = await setupWith(adapter)
|
||||
const siblingParent = ctx.agentLoop.create(
|
||||
SessionId('sibling-parent'),
|
||||
{ provider: 'mock', model: 'mock' },
|
||||
)
|
||||
const target = await ctx.subagents.startContinuable(startSpec(parent))
|
||||
const sibling = await ctx.subagents.startContinuable(startSpec(siblingParent))
|
||||
await vi.waitFor(() => { expect(adapter.requests).toHaveLength(2) })
|
||||
const targetChild = ctx.agents.get(target.childId)!
|
||||
const siblingChild = ctx.agents.get(sibling.childId)!
|
||||
const grandchild = await ctx.subagents.startContinuable(startSpec(targetChild))
|
||||
await vi.waitFor(() => { expect(adapter.requests).toHaveLength(3) })
|
||||
const cancellations: SessionId[] = []
|
||||
ctx.on('agent/cancel-requested', (agent) => { cancellations.push(agent.id) })
|
||||
|
||||
const drained = ctx.subagents.drainContinuableDescendants([parent])
|
||||
const convergedDrain = ctx.subagents.drainContinuableDescendants([parent])
|
||||
|
||||
// The scoped cutoff stops only the selected forest. The sibling child stays
|
||||
// resident and can accept later work while target cleanup is still blocked.
|
||||
expect(cancellations).toEqual([target.childId, grandchild.childId])
|
||||
expect(ctx.agents.get(target.childId)).toBe(targetChild)
|
||||
expect(ctx.agents.get(grandchild.childId)).toBeDefined()
|
||||
expect(ctx.agents.get(sibling.childId)).toBe(siblingChild)
|
||||
await expect(followup(ctx, siblingParent, sibling.childId, message('still live')))
|
||||
.resolves.toBeTypeOf('string')
|
||||
await expect(ctx.subagents.startContinuable(startSpec(parent)))
|
||||
.rejects.toMatchObject({ code: 'DRAINING' })
|
||||
await expect(followup(ctx, parent, target.childId, message('too late')))
|
||||
.rejects.toMatchObject({ code: 'DRAINING' })
|
||||
|
||||
releaseTarget.resolve(undefined)
|
||||
releaseGrandchild.resolve(undefined)
|
||||
await Promise.all([drained, convergedDrain])
|
||||
expect(ctx.agents.get(target.childId)).toBeUndefined()
|
||||
expect(ctx.agents.get(grandchild.childId)).toBeUndefined()
|
||||
expect(ctx.agents.get(sibling.childId)).toBe(siblingChild)
|
||||
// The exact root remains closed until its host disposes it, even after all
|
||||
// current descendants are gone.
|
||||
await expect(ctx.subagents.startContinuable(startSpec(parent)))
|
||||
.rejects.toMatchObject({ code: 'DRAINING' })
|
||||
|
||||
releaseSibling.resolve(undefined)
|
||||
await waitNoActivation(ctx, sibling.childId)
|
||||
})
|
||||
|
||||
it('retains a continuable root while draining only its descendants', async () => {
|
||||
const releaseChild = Promise.withResolvers<undefined>()
|
||||
const releaseGrandchild = Promise.withResolvers<undefined>()
|
||||
const adapter = new GatedAdapter([
|
||||
{ chunks: textResponse('child'), gate: releaseChild.promise },
|
||||
{ chunks: textResponse('grandchild'), gate: releaseGrandchild.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 grandchild = await ctx.subagents.startContinuable(startSpec(child))
|
||||
await vi.waitFor(() => { expect(adapter.requests).toHaveLength(2) })
|
||||
const cancellations: SessionId[] = []
|
||||
ctx.on('agent/cancel-requested', (agent) => { cancellations.push(agent.id) })
|
||||
|
||||
const drained = ctx.subagents.drainContinuableDescendants([child])
|
||||
|
||||
expect(cancellations).toEqual([grandchild.childId])
|
||||
expect(ctx.agents.get(started.childId)).toBe(child)
|
||||
releaseGrandchild.resolve(undefined)
|
||||
await drained
|
||||
expect(ctx.agents.get(grandchild.childId)).toBeUndefined()
|
||||
expect(ctx.agents.get(started.childId)).toBe(child)
|
||||
await expect(ctx.subagents.startContinuable(startSpec(child)))
|
||||
.rejects.toMatchObject({ code: 'DRAINING' })
|
||||
|
||||
releaseChild.resolve(undefined)
|
||||
await waitNoActivation(ctx, started.childId)
|
||||
})
|
||||
|
||||
it('finds scoped descendants after an intermediate one-shot Agent leaves the registry', async () => {
|
||||
const releaseIntermediate = Promise.withResolvers<undefined>()
|
||||
const releaseDescendant = Promise.withResolvers<undefined>()
|
||||
const adapter = new GatedAdapter([
|
||||
{ chunks: textResponse('one-shot'), gate: releaseIntermediate.promise },
|
||||
{ chunks: textResponse('continuable descendant'), gate: releaseDescendant.promise },
|
||||
])
|
||||
const { ctx, parent } = await setupWith(adapter)
|
||||
const run = await ctx.subagents.start('spawn', {
|
||||
prompt: message('one-shot task'),
|
||||
parent,
|
||||
signal: testSignal,
|
||||
})
|
||||
const intermediate = run.localAgent
|
||||
expect(intermediate).toBeDefined()
|
||||
if (intermediate === undefined) throw new Error('spawn must publish a local Agent')
|
||||
const descendant = await ctx.subagents.startContinuable(startSpec(intermediate))
|
||||
await vi.waitFor(() => { expect(adapter.requests).toHaveLength(2) })
|
||||
|
||||
const intermediateId = intermediate.id
|
||||
const disposingIntermediate = run.dispose()
|
||||
releaseIntermediate.resolve(undefined)
|
||||
await disposingIntermediate
|
||||
expect(ctx.agents.get(intermediateId)).toBeUndefined()
|
||||
expect(ctx.agents.get(descendant.childId)).toBeDefined()
|
||||
const cancellations: SessionId[] = []
|
||||
ctx.on('agent/cancel-requested', (agent) => { cancellations.push(agent.id) })
|
||||
|
||||
const drained = ctx.subagents.drainContinuableDescendants([parent])
|
||||
|
||||
expect(cancellations).toEqual([descendant.childId])
|
||||
releaseDescendant.resolve(undefined)
|
||||
await drained
|
||||
expect(ctx.agents.get(descendant.childId)).toBeUndefined()
|
||||
})
|
||||
|
||||
it('awaits and rolls back an admitted materialization below a scoped root', async () => {
|
||||
const { ctx, parent } = await setup([])
|
||||
const manager = (ctx.subagents as unknown as {
|
||||
continuations: { ownerCtx: Context }
|
||||
}).continuations
|
||||
const agents = manager.ownerCtx.agents
|
||||
const create = agents.create.bind(agents)
|
||||
const published = Promise.withResolvers<SessionId>()
|
||||
const releaseMaterialization = Promise.withResolvers<undefined>()
|
||||
const createSpy = vi.spyOn(agents, 'create').mockImplementation(async (options) => {
|
||||
const handle = await create(options)
|
||||
published.resolve(handle.agent.id)
|
||||
await releaseMaterialization.promise
|
||||
return handle
|
||||
})
|
||||
|
||||
try {
|
||||
const starting = ctx.subagents.startContinuable(startSpec(parent))
|
||||
const childId = await published.promise
|
||||
let drainResolved = false
|
||||
const drained = ctx.subagents.drainContinuableDescendants([parent]).then(() => {
|
||||
drainResolved = true
|
||||
})
|
||||
await Promise.resolve()
|
||||
expect(drainResolved).toBe(false)
|
||||
|
||||
releaseMaterialization.resolve(undefined)
|
||||
await expect(starting).rejects.toMatchObject({ code: 'DRAINING' })
|
||||
await drained
|
||||
expect(ctx.agents.get(childId)).toBeUndefined()
|
||||
} finally {
|
||||
createSpy.mockRestore()
|
||||
}
|
||||
})
|
||||
|
||||
it('ignores a stale scoped root without disabling its live same-id Agent', async () => {
|
||||
const { ctx, parent } = await setup([textResponse('done')])
|
||||
const stale = { ...parent, id: parent.id } as unknown as Agent
|
||||
|
||||
await ctx.subagents.drainContinuableDescendants([stale])
|
||||
const started = await ctx.subagents.startContinuable(startSpec(parent))
|
||||
|
||||
await waitNoActivation(ctx, started.childId)
|
||||
})
|
||||
|
||||
it('reports a scoped teardown failure after releasing the selected branch', async () => {
|
||||
const hold = Promise.withResolvers<undefined>()
|
||||
const adapter = new GatedAdapter([
|
||||
{ chunks: textResponse('target child'), 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 manager = (ctx.subagents as unknown as {
|
||||
continuations: { activations: Map<SessionId, { handle: { dispose: () => Promise<void> } }> }
|
||||
}).continuations
|
||||
const activation = manager.activations.get(started.childId)!
|
||||
const realDispose = activation.handle.dispose.bind(activation.handle)
|
||||
activation.handle.dispose = async () => {
|
||||
await realDispose()
|
||||
throw new Error('scoped child reap failed')
|
||||
}
|
||||
|
||||
const drained = ctx.subagents.drainContinuableDescendants([parent])
|
||||
hold.resolve(undefined)
|
||||
|
||||
await expect(drained).rejects.toMatchObject({ code: 'ACTIVATION_TEARDOWN_FAILED' })
|
||||
expect(ctx.agents.get(started.childId)).toBeUndefined()
|
||||
})
|
||||
|
||||
it('rejects new materialization and delivery once draining begins', async () => {
|
||||
const { ctx, parent } = await setup([textResponse('done')])
|
||||
const started = await ctx.subagents.startContinuable(startSpec(parent))
|
||||
|
||||
@@ -119,10 +119,11 @@ describe('SubagentService', () => {
|
||||
expect('resume' in provider).toBe(false)
|
||||
})
|
||||
|
||||
it('drains continuable activations as a no-op when no manager was bound', async () => {
|
||||
it('treats global and scoped drains as no-ops when no manager was bound', async () => {
|
||||
const { subagents } = await service()
|
||||
// Without `ctx.agents` no manager exists, so nothing was ever materialized.
|
||||
await expect(subagents.drainContinuable()).resolves.toBeUndefined()
|
||||
await expect(subagents.drainContinuableDescendants([])).resolves.toBeUndefined()
|
||||
})
|
||||
|
||||
it('rejects continuable operations when their runtime services are absent', async () => {
|
||||
|
||||
Reference in New Issue
Block a user