From 26a117f842c9cc87987963a2cd127b2c980e283c Mon Sep 17 00:00:00 2001 From: Dudu-0223 Date: Thu, 30 Jul 2026 13:26:13 +0800 Subject: [PATCH] feat(subagent): activation-based continuable subagents (source) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Replace the Task-backed continuation manager with one durable Session plus at most one process-local Activation — a residency epoch for a reconstructed child Agent, not a request, result, cancellation, or Task boundary. The manager owns activation admission, authority, the live ownership graph, cold resume, and child-first disposal; the Agent inbox is the only turn FIFO. - startContinuable() is async and returns { childId, messageId } at inbox acceptance; followup() takes a SubagentAuthority and returns AgentMessageId. - SubagentProvider.resume?(), SubagentProviderResumeRequest, SubagentRun.steer?(), SubagentProviderStartRequest and SubagentContinuation are deleted; prepareContinuable?() is the continuable-creation capability. - Cold resume calls ctx.agents.resume() from the manager through a private activation-owner scope, never dispatching through a provider. - Extract shared child composition, descriptor seeding, depth accounting, and one-shot run settlement so the manager and one-shot driver keep one home per fact. Tests and docs follow in subsequent commits. --- ...ntinuable-subagent-conversations.i18n.yaml | 6 + ...7-28-continuable-subagent-conversations.md | 216 ++++ ...8-continuable-subagent-conversations.zh.md | 216 ++++ packages/subagent/subagent-fork/src/index.ts | 20 +- .../subagent/subagent-inprocess/src/index.ts | 220 +--- packages/subagent/subagent-spawn/src/index.ts | 16 +- packages/subagent/subagent/src/child-agent.ts | 128 ++ .../subagent/subagent/src/continuation.ts | 1114 ++++++++++------- packages/subagent/subagent/src/depth.ts | 51 + .../subagent/subagent/src/descriptor-seed.ts | 31 + packages/subagent/subagent/src/index.ts | 254 ++-- .../subagent/subagent/src/run-settlement.ts | 71 ++ packages/subagent/subagent/src/types.ts | 162 +-- .../tool-subagent-control/src/index.ts | 37 +- packages/subagent/tool-subagent/src/index.ts | 47 +- 15 files changed, 1721 insertions(+), 868 deletions(-) create mode 100644 .agents/notes/proposed/feature/2026-07-28-continuable-subagent-conversations.i18n.yaml create mode 100644 .agents/notes/proposed/feature/2026-07-28-continuable-subagent-conversations.md create mode 100644 .agents/notes/proposed/feature/2026-07-28-continuable-subagent-conversations.zh.md create mode 100644 packages/subagent/subagent/src/child-agent.ts create mode 100644 packages/subagent/subagent/src/depth.ts create mode 100644 packages/subagent/subagent/src/descriptor-seed.ts create mode 100644 packages/subagent/subagent/src/run-settlement.ts diff --git a/.agents/notes/proposed/feature/2026-07-28-continuable-subagent-conversations.i18n.yaml b/.agents/notes/proposed/feature/2026-07-28-continuable-subagent-conversations.i18n.yaml new file mode 100644 index 0000000000..4ef20ef978 --- /dev/null +++ b/.agents/notes/proposed/feature/2026-07-28-continuable-subagent-conversations.i18n.yaml @@ -0,0 +1,6 @@ +# Bilingual-pair consistency record (docs/i18n/README.md): the git blob hash of each +# 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/proposed/feature/2026-07-28-continuable-subagent-conversations.md +2026-07-28-continuable-subagent-conversations.md: 3902fbc33004219f98d070d4b898de6b2c19d40d +2026-07-28-continuable-subagent-conversations.zh.md: 11f59d8f1a57e2d1bf375a3e1c1cd46043c60a3f diff --git a/.agents/notes/proposed/feature/2026-07-28-continuable-subagent-conversations.md b/.agents/notes/proposed/feature/2026-07-28-continuable-subagent-conversations.md new file mode 100644 index 0000000000..3902fbc330 --- /dev/null +++ b/.agents/notes/proposed/feature/2026-07-28-continuable-subagent-conversations.md @@ -0,0 +1,216 @@ +# Agent Note: Continuable subagents + +Status: proposed + +English | [中文](2026-07-28-continuable-subagent-conversations.zh.md) + +This proposal would replace the Task-backed continuation manager from [Continuable background subagents](../../implemented/feature/2026-07-21-continuable-background-subagents.md). It retains the single `ctx.subagents` service from [Merge subagent control into the subagent service](../../implemented/simplification/2026-07-26-merge-subagent-control-service.md) and the intent-named `followup` operation from [Intent-named subagent continuation operations](../../implemented/simplification/2026-07-27-intent-named-subagent-continuation-operations.md). + +## Problem + +The continuation manager currently makes one Task, one provider execution, and one result boundary the same object lifetime. Task settlement disposes the child Agent, Task completion injects the completion notice, and later input reconstructs another Agent. This couples a generic background-work abstraction to conversation delivery even though a continuable subagent already has a Session and an Agent inbox. + +Giving queued parent requests to the continuation manager and user messages to the Agent creates two FIFOs with no single ordering authority. Giving both to Tasks instead duplicates the Agent loop's admission, cancellation, and quiescence machinery. `Agent.whenIdle()` cannot recover a per-request Task result because one running interval may drain multiple queued turns, and broad `Agent.cancel()` cannot remove one queued request exactly. + +The runtime lifetime is also wider than one turn. A subagent can finish its own turn while a child it created is still running. Disposing the parent runtime at that point removes the Agent that still owns descendant teardown. Keeping every historical subagent resident instead would make memory use unbounded. + +Users and parent Agents also need to send later work to the same live child without changing its current turn. Queueing every continuation message as a follow-up preserves one ordering rule for both senders. + +## Proposal + +A continuable subagent has one durable Session and at most one process-local Activation: + +```text +persisted Session + -> optional live Activation + -> one retained AgentHandle + -> Agent inbox as the only turn FIFO + -> zero or more owned child Activations +``` + +An Activation is one residency epoch for a reconstructed child Agent. It may execute multiple FIFO turns and remain resident while waiting for descendants. It is not a request, result, cancellation, or Task boundary. + +The continuation manager owns activation admission, authority checks, the live ownership graph, cold resume, and child-first disposal. The Agent loop owns all turn ordering and execution. The proposal creates no Task for a continuable subagent, no Activation FIFO, and no queued Activation state. + +### Materialization and public operations + +The named subagent provider participates only in preparing the initial creation spec, where `spawn` and `fork` differ. Its optional `prepareContinuable(request): Promise` method is the continuable-creation capability. The returned spec contains only detached provider-specific creation inputs such as the optional parent-history seed; it contains no Agent, `AgentHandle`, prompt delivery, result, disposal, or resume operation. The manager reserves the child identity, resolves the durable descriptor and common Agent setup, calls `ctx.agents.create()` through a private activation-owner scope, installs the returned `AgentHandle` into the Activation, establishes any continuable-parent ownership, and then calls `Agent.followup(initialPrompt)`. Inbox acceptance yields an `AgentMessageId`; at that boundary `ctx.subagents.startContinuable()` returns `{ childId, messageId }` without waiting for the turn to start or for the message to enter the Session log. + +Any failure before inbox acceptance rejects without returning either id. Agent creation provides rollback before handle transfer; after transfer, the manager disposes the created handle, removes the Activation, and rolls back any parent `ownedChildren` membership before rejecting. + +`backgroundMode: 'one-shot' | 'continuable'` remains deployment policy. Configured continuable mode requires `prepareContinuable`; method presence replaces `SubagentProvider.resume?()` as the capability check, while a capable provider may still run one-shot work. + +Cold resume does not dispatch through a subagent provider. The continuation manager folds the generic in-process descriptor, calls `ctx.agents.resume()` through the same activation-owner scope, installs the returned `AgentHandle`, and submits the waiting `next-turn`. `SubagentProvider.resume?()` and `SubagentProviderResumeRequest` are absent, and the initial provider name is not a recovery capability; remote providers require a separate design. + +`SubagentProvider.start()` and `SubagentRun` remain exclusively on the unchanged one-shot path. A continuable Activation directly owns its `AgentHandle` and never creates, wraps, or retains a `SubagentRun`; `SubagentRun.steer?()` is therefore absent. + +`ctx.subagents.followup(authority, childId, content, { source, signal })` remains the sole continuation-message operation. `authority` is either `{ kind: 'parent', agent }` or `{ kind: 'user' }`; the parent variant is admitted only from an exact live Agent tool context, while only a trusted host adapter can supply user authority. `source` remains durable provenance and grants no authority. The model-facing `send_message` tool keeps only its stable `subagent_id` and `message` fields and always submits a follow-up turn. Both start and follow-up return the accepted `AgentMessageId`, and neither reports how the manager materialized the Activation. + +For start and follow-up, the caller signal owns lookup, materialization, and admission only until inbox acceptance. After the operation returns its `AgentMessageId`, the manager owns the Activation independently; later caller cancellation does not cancel the accepted turn or dispose the child. + +### Durable Session and live Activation + +The Session owns the stable child identity, transcript, direct-parent lineage, delegation depth, and versioned continuation descriptor. `SessionHeader.parentSession` is durable provenance and an authorization input; it is not a live routing capability and does not imply that the historical parent is resident. + +An idle historical Session has no `AgentHandle`. The first authorized `next-turn` delivery resumes an Activation from the persisted Session and submits the message to its inbox. A user-authorized cold resume does not load the historical parent Agent. A parent-originated resume uses the exact live parent Agent for authorization and, when that parent has an Activation, ownership; it never uses the parent for reconstruction. + +The Activation directly owns the published `AgentHandle` until it settles, while the manager's private activation-owner scope is its structural Cordis owner. The continuable path creates no intermediate result-bearing execution wrapper, including `SubagentRun`; one-shot delegation remains unchanged and outside this lifecycle. Remote providers are outside the MVP and require a separate Activation ownership contract when introduced. Historical Sessions consume no runtime memory after their Activation is disposed. + +### Activation lifecycle + +The public lifecycle has three states and no `queued` state: + +```text +running + | Agent quiescent with live children + v +waiting + | next-turn + +--------------------------> running + +running or waiting + | Agent quiescent and no live children + v +settled + | AgentHandle.dispose completes + v +no Activation +``` + +`running` means the Agent has an active admission or turn, or its inbox contains waking work. `waiting` means the Agent is quiescent but the Activation still owns at least one child Activation that has not completed disposal. `settled` means the Agent is quiescent and every owned child is disposed; the manager then disposes the `AgentHandle` and removes the Activation. + +The manager derives these states from Agent quiescence and the owned-child set rather than maintaining a second execution state machine. A `next-turn` delivered while `running` joins the Agent inbox. A `next-turn` delivered while `waiting` wakes the same Agent and returns the Activation to `running`. Delivery after disposal cold-resumes a new Activation. + +The manager linearizes delivery, child release, and disposal for each durable child. If a delivery races with final disposal, exactly one side wins the admission cutoff: delivery either enters the still-live Agent inbox, or waits for disposal and cold-resumes a new Activation. No caller can send to a handle after its disposal transaction begins. + +### One inbox and follow-up delivery + +The Agent inbox is the only queue. Every continuation message uses `Agent.followup()` and becomes one FIFO turn; neither the continuation manager nor the host maintains another message queue. Every accepted waking item keeps the current Activation live until `Agent.whenIdle()` observes the complete waking suffix. + +Routing depends only on Activation residency: + +| Activation state | Sender | `followup` | +|---|---|---| +| `running` | parent or user | enqueue in the same Activation | +| `waiting` | parent or user | wake the same Activation | +| no Activation | parent or user | cold-resume a new Activation | + +The continuation layer defines no separate delivery-route result. Successful `ctx.subagents.followup()` and `send_message` delivery returns the accepted `AgentMessageId`, while delivery failure throws. Existing `agent/inbox/enqueue`, `agent/inbox/dequeue`, and `agent/inbox/discard` events remain the message-lifecycle observations; adapters may render a generic acceptance but do not expose `started`, `queued`, `resumed`, or another subagent-specific route vocabulary. + +### Child ownership + +Every Activation owns its `AgentHandle` and an `ownedChildren: Set`. Because one Session has at most one live Activation, the child Session id identifies the live child without another runtime-incarnation reference. `SessionHeader.parentSession` records the durable direct-parent identity, while membership in `ownedChildren` records the process-local ownership relationship. + +When the authenticated parent is itself a continuation-managed Activation, starting a child or submitting parent-originated work adds the child Session id to that parent's `ownedChildren` before the child can run or the message can enter its inbox. That parent cannot settle or dispose while this set is non-empty. A top-level or other non-continuation Agent has no Activation and does not join this waiting graph. + +Child release occurs 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. The manager calls `ctx.sessions.flush(child.session)`: `true` confirms durability, while `false` or rejection is normalized to `DURABILITY_FAILED`. A failed checkpoint is reported but does not prevent handle disposal or ownership release, because retaining a failed child would permanently pin its ancestors in `waiting`. If the child is owned, the manager then resolves the live parent through `SessionHeader.parentSession` and removes the child Session id from its `ownedChildren`; a user-resumed child with no live owner has nothing to release. Manager teardown uses the same child-first order. + +A user cold-resume creates an Activation without adding it to the historical parent's `ownedChildren`. If the direct parent later submits work to that live Activation and is itself continuation-managed, admission establishes ownership before enqueueing the message; a non-continuation parent remains outside the waiting graph. + +The MVP retains ownership 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, then disposes every live Activation forest in child-first order and awaits all `AgentHandle.dispose()` calls. Only after that drain settles may the host dispose top-level Agents and the manager scope. Manager unload uses the same drain and includes user-resumed Activations without live owners. + +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. + +### Deferred report delivery + +The MVP exposes no `report` tool and provides no child-to-parent content delivery or automatic parent wakeup. The durable child Session remains the source of the child's detailed output. + +A later proposal may add an ordinary model-facing `report(output)` tool that can be called zero or multiple times in one turn. Its delivery policy may distinguish quiet parent injection from waking the parent; recipient selection, acknowledgement, durability, and retry semantics are deferred with that tool. Adding report delivery does not require another Activation state or execution queue. + +### Deferred steering + +The MVP exposes no subagent steering operation. Parent and user continuation messages always open later FIFO turns, so the continuation layer stores no current-turn controller and adds no controller-aware Agent admission seam. + +A later host UI may expose separate **Steer** and **Follow up** actions. User steering would be strict and live-only: it may call the existing Agent steering path only while the Activation accepts a next step, must reject otherwise, and must never fall back to queueing or cold resume. Exposing parent steering to a model-facing tool remains a separate design because distinct tool names express intent but do not establish whether the parent may modify a user-controlled turn. + +### Authority and provenance + +Authority is supplied by a trusted host interaction or an exact live Agent tool context. `MessageSource` and `senderSessionId` are durable provenance after admission, not caller-controlled authority. + +The MVP authorizes the host user and the durable child's direct parent. Parent authorization checks `SessionHeader.parentSession` against the authenticated parent Agent before registering the child in that parent's `ownedChildren`. Other Agents, ancestors, teams, and workflows remain rejected until an explicit authority protocol exists. + +User authority may cold-resume a child without its parent. Parent-originated delivery requires the parent to be live when admitted and keeps it live through the ownership relationship. + +### Durability, disposal, and recovery + +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, neither parent nor user can cancel the message, turn, or Activation through `ctx.subagents`; `Agent.cancel()` remains a lower-level Agent capability that this MVP 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. + +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. + +Only messages written to the child Session log are reconstructable with their admitted provenance; inbox acceptance alone provides no restart guarantee. + +Session and descriptor persistence survive restart. Activation state, Agent inbox contents, and the ownership graph are process-local. A process crash may lose an accepted initial prompt or follow-up that remained in the inbox without reaching the Session log. The Session and descriptor may survive so a later authorized message can cold-resume the child, but the lost message is not replayed automatically. Recovering accepted unfinished or unlogged messages requires a durable inbox protocol and is not implied here. + +### Scope + +The MVP covers continuable in-process children and leaves one-shot delegation unchanged. Remote providers require a separate Activation handle with equivalent authenticated control and child-first quiescence contracts before they can support the same behavior. + +The MVP adds no subagent steering operation, report tool, child-to-parent content delivery, automatic parent wakeup, durable mailbox, cross-process lease, automatic replay of interrupted inbox work, team authority, workflow authority, public subagent cancellation operation, new live-Activation or descendant limit, or runtime cache. Existing delegation-depth policy remains unchanged. + +## Alternatives considered + +**Keep Task-backed Activations.** Tasks provide generic status, result collection, and cancellation, but using them for conversation delivery creates a second queue and duplicates turn ownership. The proposal gives up those generic Task controls so the Agent inbox remains the only execution order. + +**Create one Activation per `next-turn`.** This restores independent result and cancellation boundaries, but it requires a manager FIFO beside the Agent inbox and makes a retained Agent cross artificial Activation boundaries. One Activation per residency epoch is smaller and follows the `AgentHandle` lifetime directly. + +**Dispose the Agent while waiting.** Reconstructing a parent while its child still belongs to the previous process-local ownership graph would require a durable ownership and teardown protocol. Retaining the `AgentHandle` only for the unfinished graph preserves child-first teardown without keeping settled history resident. + +**Let the provider create, resume, or deliver through an Agent handle.** Initial providers own only `prepareContinuable()` and its detached creation-spec distinction: whether a child begins fresh or with a parent prefix. The manager must call `ctx.agents.create()` through its private activation-owner scope so that scope is a structural owner of every handle. A persisted in-process Session already contains the initial prefix and generic reconstruction descriptor, while delivery belongs to the Agent inbox. Giving providers any later handle, `SubagentRun`, or message ownership would preserve a seam with no MVP behavior to own and would complicate user cold resume with an unnecessary live-parent input. + +**Add report delivery to the MVP.** A repeatable model-facing tool is compatible with this lifecycle, but quiet versus waking delivery, recipient selection, acknowledgement, durability, and retry behavior are independent product choices. Deferring the tool keeps the first version focused on conversation admission and residency without constraining that later policy. + +**Treat `SessionHeader.parentSession` as live ownership.** Durable lineage does not prove that the historical parent currently owns the child. Membership in the live parent's `ownedChildren` records the process-local relationship without changing durable provenance. + +**Retain the exact parent Agent in a separate link.** The parent Activation already owns its `AgentHandle`, and `ownedChildren` prevents that Activation from disposing while the child remains live. Resolving the parent by Session id is therefore sufficient and avoids a redundant runtime reference. + +**Maintain a separate queue for parent messages.** A second FIFO creates ambiguous ordering against user messages already accepted by the Agent. A single Agent inbox gives both origins one observable order. + +**Expose subagent steering in the MVP.** User steering can be a strict live-only host action, but parent steering needs current-turn controller state to protect a user-controlled turn. Queueing every first-version continuation avoids that state and its admission race. A later UI can add a distinct user-only action without changing follow-up ordering. + +**Return a subagent-specific delivery route.** Labels such as `started`, `queued`, and `resumed` duplicate Activation and inbox state without giving the caller an independent result. Reusing `AgentMessageId` and the existing inbox events keeps delivery correlation on the Agent contract that owns it. + +**Use a child reference count.** A count cannot identify which child still owns teardown work and permits duplicate decrement errors. An identity set retains cancellation and disposal obligations explicitly. + +## Acceptance criteria + +- A continuable child has at most one live Activation and one Agent inbox; the continuation manager has no Activation FIFO or queued Activation state. +- `SubagentProvider.prepareContinuable?()` returns only a detached `ContinuableCreateSpec`; configured continuable mode requires that capability, while `backgroundMode` remains an independent policy choice. +- The manager calls `ctx.agents.create()` through its private activation-owner scope, installs the returned `AgentHandle` and parent ownership, calls `Agent.followup(initialPrompt)`, and returns `{ childId, messageId }` when inbox acceptance yields the `AgentMessageId`, without waiting for turn start or a Session-log write. +- Every failure before initial-prompt inbox acceptance rejects without ids and rolls back any created handle, Activation, and parent `ownedChildren` membership. +- Cold resume calls `ctx.agents.resume()` from the continuation manager and never dispatches through the initial subagent provider; `SubagentProvider.resume?()` and `SubagentProviderResumeRequest` are absent. +- A continuable Activation directly owns `AgentHandle` and never creates, wraps, or retains `SubagentRun`; `SubagentProvider.start()` and `SubagentRun` remain one-shot-only, without `SubagentRun.steer?()`. +- A user can cold-resume a persisted child without loading its historical parent. +- `followup()` accepts only trusted parent or user authority; durable message provenance cannot authorize delivery. +- Parent and user continuation messages always use `Agent.followup()` and share its inbox FIFO, including when one origin queues behind the other or the child already has an open turn. +- `ctx.subagents.followup()` and its `send_message` adapter return only the accepted `AgentMessageId`; the continuation layer accepts no delivery target and defines no subagent-specific route result. +- The MVP 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. +- The MVP 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, stop manager-owned outward notifications, dispose every snapshotted 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. +- The MVP 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 failure, caller-signal 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, `AgentMessageId` 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, mixed parent/user follow-up queueing, the absence of subagent steering, report delivery, and automatic parent wakeup, retained waiting `AgentHandle`, and child-first disposal. + +## Risks + +Removing Tasks gives up generic background-work inspection, result collection, and exact Task cancellation. If those product features become requirements, they need a request ticket or inbox capability that does not reintroduce a second execution queue. + +Retaining an Activation while descendants run consumes Agent resources proportional to the unfinished ownership graph. The existing delegation-depth policy still bounds nesting, but the MVP adds no live-Activation or total-descendant limit; settled historical Sessions retain no `AgentHandle`. + +The process-local inbox and ownership graph do not coordinate two harness processes. Deployments allowing concurrent access to one persistence store still require a durable lease and mailbox protocol. + +Without report delivery, completing a child turn neither sends its content to nor wakes the historical parent. The output remains in the durable child Session until a caller inspects that transcript or submits another authorized turn. A later report tool may add quiet or waking delivery without changing the Activation lifecycle. + +Queueing every continuation message means a parent cannot correct an in-progress child turn immediately; the correction runs as the next turn. A later user-only UI steering action may reduce that latency without introducing parent-versus-user controller policy into the MVP. + +A failed final durability checkpoint allows the runtime ownership graph to drain but leaves the persisted child state missing or stale. The failure is observable as `DURABILITY_FAILED`; retry and repair require a separate recovery design. diff --git a/.agents/notes/proposed/feature/2026-07-28-continuable-subagent-conversations.zh.md b/.agents/notes/proposed/feature/2026-07-28-continuable-subagent-conversations.zh.md new file mode 100644 index 0000000000..11f59d8f1a --- /dev/null +++ b/.agents/notes/proposed/feature/2026-07-28-continuable-subagent-conversations.zh.md @@ -0,0 +1,216 @@ +# Agent Note（agent 决策记录）：可继续的 subagent + +Status: proposed + +[English](2026-07-28-continuable-subagent-conversations.md) | 中文 + +本提案将取代[可继续的后台 subagent](../../implemented/feature/2026-07-21-continuable-background-subagents.md)中由 Task 支撑的继续执行管理器。提案保留[将 subagent 控制合并到 subagent 服务](../../implemented/simplification/2026-07-26-merge-subagent-control-service.md)确立的单一 `ctx.subagents` 服务，以及[以意图命名的 subagent 继续执行操作](../../implemented/simplification/2026-07-27-intent-named-subagent-continuation-operations.md)确立的 `followup` 操作。 + +## 问题 + +继续执行管理器目前让一个 Task、一次提供方执行和一个结果边界共享同一生命周期。Task 结算会 dispose（资源释放）child Agent，Task 完成会注入完成通知，后续输入则重建另一个 Agent。这使通用后台工作抽象与会话投递耦合，而可继续 subagent 已经具备会话和 Agent inbox。 + +如果继续执行管理器为 parent 请求排队，而 Agent 接收用户消息，系统就会出现两个 FIFO，且没有唯一的顺序权威。如果两种消息都交给 Task，系统又会重复 agent loop（智能体循环）已有的准入、取消和完全停稳机制。`Agent.whenIdle()` 无法恢复单项请求的 Task 结果，因为一个运行区间可能清空多个排队轮次；宽泛的 `Agent.cancel()` 也不能精确移除一项排队请求。 + +运行时生命周期也比单个轮次更长。subagent 可能已经结束自身轮次，但它创建的 child 仍在运行。此时 dispose parent 运行时，会移除仍负责后代拆卸的 Agent。反之，如果让所有历史 subagent 始终驻留，内存使用就会失去上界。 + +用户和 parent Agent 还需要在不改变当前轮次的前提下，向同一个在线 child 发送后续工作。将每条继续执行消息作为 follow-up 排队，可以让两类发送方遵循同一项排序规则。 + +## 提案 + +一个可继续 subagent 拥有一个持久化会话，并且至多拥有一个进程内激活： + +```text +persisted Session + -> optional live Activation + -> one retained AgentHandle + -> Agent inbox as the only turn FIFO + -> zero or more owned child Activations +``` + +激活是重建 child Agent 的一次驻留周期。它可以执行多个 FIFO 轮次，并在等待后代时保持驻留。它不是请求、结果、取消或 Task 边界。 + +继续执行管理器负责激活准入、权限检查、在线所有权图、冷恢复和 child-first dispose。Agent loop 负责全部轮次排序与执行。本提案不会为可继续 subagent 创建 Task、激活 FIFO 或 queued 激活状态。 + +### 物化与公开操作 + +具名 subagent 提供方只参与准备初始创建规格，此时 `spawn` 与 `fork` 有所区别。其可选的 `prepareContinuable(request): Promise` 方法就是可继续创建能力。返回的规格只包含与 Agent 实例分离且由提供方决定的创建输入，例如可选的 parent 历史种子；它不包含 Agent、`AgentHandle`、提示词投递、结果、dispose 或恢复操作。管理器会预留 child 身份，解析持久化描述符和通用 Agent 配置，通过私有 activation-owner 作用域调用 `ctx.agents.create()`，将返回的 `AgentHandle` 安装到激活中，建立适用的可继续 parent 所有权，然后调用 `Agent.followup(initialPrompt)`。inbox 接受消息后会产生一个 `AgentMessageId`；`ctx.subagents.startContinuable()` 在此边界返回 `{ childId, messageId }`，不等待轮次开始，也不等待消息写入会话日志。 + +inbox 接受消息前发生任何失败，操作都会在不返回任何 id 的情况下被拒绝。Agent 创建流程负责 handle 移交前的回滚；移交后，管理器会先 dispose 已创建的 handle、移除激活并回滚 parent `ownedChildren` 中的任何成员关系，再拒绝操作。 + +`backgroundMode: 'one-shot' | 'continuable'` 仍是部署策略。配置为 continuable 时要求存在 `prepareContinuable`；该方法是否存在会取代 `SubagentProvider.resume?()` 成为能力检查，而具备该能力的提供方仍可运行 one-shot 工作。 + +冷恢复不会通过 subagent 提供方分发。继续执行管理器会归并通用的进程内描述符，通过同一个 activation-owner 作用域调用 `ctx.agents.resume()`，安装返回的 `AgentHandle`，并提交等待中的 `next-turn`。`SubagentProvider.resume?()` 和 `SubagentProviderResumeRequest` 均不存在，初始提供方名称也不是恢复能力；远程提供方需要单独设计。 + +`SubagentProvider.start()` 和 `SubagentRun` 只保留在不变的 one-shot 路径上。可继续激活直接持有自身的 `AgentHandle`，绝不创建、包装或保留 `SubagentRun`；因此，`SubagentRun.steer?()` 不存在。 + +`ctx.subagents.followup(authority, childId, content, { source, signal })` 仍是唯一的继续执行消息操作。`authority` 可以是 `{ kind: 'parent', agent }` 或 `{ kind: 'user' }`；parent 变体仅能从确切的在线 Agent 工具上下文通过准入，只有可信宿主适配器才能提供用户权限。`source` 仍是持久化来源信息，不赋予任何权限。面向模型的 `send_message` 工具只保留稳定的 `subagent_id` 和 `message` 字段，并始终提交一个 follow-up 轮次。start 和 follow-up 都返回已接受的 `AgentMessageId`，两者都不报告管理器如何物化激活。 + +对于 start 和 follow-up，调用方 signal 只在 inbox 接受消息前持有查找、物化和准入。操作返回 `AgentMessageId` 后，管理器会独立持有该激活；调用方之后的取消不会取消已接受的轮次，也不会 dispose child。 + +### 持久化会话与在线激活 + +会话持有稳定的 child 身份、transcript（文本记录）、直接 parent 谱系、委派深度和带版本的继续执行描述符。`SessionHeader.parentSession` 是持久化来源信息和鉴权输入；它不是在线路由能力，也不表示历史 parent 仍然驻留。 + +空闲的历史会话没有 `AgentHandle`。第一条通过鉴权的 `next-turn` 投递会根据持久化会话恢复激活，并将消息提交到其 inbox。经用户授权的冷恢复不会加载历史 parent Agent。parent 发起的恢复使用经过身份认证的确切在线 parent Agent 执行鉴权；当该 parent 有激活时，还使用它建立所有权，但绝不使用 parent 执行重建。 + +激活作为消费方会直接持有已发布的 `AgentHandle` 直至结算，而管理器的私有 activation-owner 作用域则是其 Cordis 结构化所有者。可继续 subagent 路径不创建任何中间的带结果执行包装层，包括 `SubagentRun`；一次性委派保持不变，且不属于该生命周期。远程提供方不在 MVP 范围内，引入时需要单独的激活所有权契约。激活 dispose 后，历史会话不消耗运行时内存。 + +### 激活生命周期 + +公开生命周期只有 3 个状态，没有 `queued` 状态： + +```text +running + | Agent quiescent with live children + v +waiting + | next-turn + +--------------------------> running + +running or waiting + | Agent quiescent and no live children + v +settled + | AgentHandle.dispose completes + v +no Activation +``` + +`running` 表示 Agent 正在执行准入或轮次，或者 inbox 中存在会唤醒 Agent 的工作。`waiting` 表示 Agent 已经完全停稳，但激活仍持有至少一个尚未完成 dispose 的 child 激活。`settled` 表示 Agent 已经完全停稳且所有持有的 child 都已 dispose；随后管理器会 dispose `AgentHandle` 并移除激活。 + +管理器根据 Agent 是否完全停稳以及所持 child 集合派生这些状态，而不是维护第二套执行状态机。在 `running` 时投递的 `next-turn` 会进入 Agent inbox。在 `waiting` 时投递的 `next-turn` 会唤醒同一个 Agent，并使激活回到 `running`。在 dispose 完成后投递消息则会冷恢复新激活。 + +管理器会针对每个持久化 child，将投递、child 释放和 dispose 线性化。如果投递与最终 dispose 发生竞争，只有一方能越过准入截止点：投递要么进入仍在线的 Agent inbox，要么等待 dispose 完成后冷恢复新激活。任何调用方都不能向已经开始 dispose 事务的 handle 发送消息。 + +### 一个 inbox 与 follow-up 投递 + +Agent inbox 是唯一队列。每条继续执行消息都使用 `Agent.followup()`，并成为一个 FIFO 轮次；继续执行管理器和宿主都不维护另一条消息队列。每个已接受且会唤醒 Agent 的条目都会让当前激活保持在线，直至 `Agent.whenIdle()` 观察到完整的唤醒工作后缀已经结束。 + +路由只取决于激活的驻留状态： + +| 激活状态 | 发送方 | `followup` | +|---|---|---| +| `running` | parent 或 user | 在同一激活中排队 | +| `waiting` | parent 或 user | 唤醒同一激活 | +| 无激活 | parent 或 user | 冷恢复新激活 | + +继续执行层不定义单独的投递路由结果。成功投递 `ctx.subagents.followup()` 或 `send_message` 时会返回已接受的 `AgentMessageId`，投递失败则会抛出异常。现有的 `agent/inbox/enqueue`、`agent/inbox/dequeue` 和 `agent/inbox/discard` 事件仍用于观测消息生命周期；适配器可以呈现通用的接受确认，但不暴露 `started`、`queued`、`resumed` 或其他 subagent 专属路由词汇。 + +### child 所有权 + +每次激活都持有自身的 `AgentHandle` 和一个 `ownedChildren: Set`。由于一个会话至多有一次在线激活，child 会话 id 足以标识在线 child，无需另一个运行时 incarnation 引用。`SessionHeader.parentSession` 记录持久化的直接 parent 身份，`ownedChildren` 中的成员关系则记录进程内所有权关系。 + +当经过身份认证的 parent 自身是由继续执行管理器管理的激活时，启动 child 或提交由 parent 发起的工作，会在 child 可以运行或消息可以进入其 inbox 前，将 child 会话 id 加入该 parent 的 `ownedChildren`。该集合非空时，这个 parent 不能结算或 dispose。顶层 Agent 或其他非继续执行 Agent 没有激活，也不会加入该等待图。 + +只有在 child Agent 完全停稳、该 child 持有的每个 child 都已 dispose、最终持久性检查点结算且 child 的 `AgentHandle` 完成 dispose 后，系统才释放 child。管理器会调用 `ctx.sessions.flush(child.session)`：只有 `true` 确认持久性，`false` 或 rejection 则统一报告为 `DURABILITY_FAILED`。检查点失败会被报告，但不会阻止 handle dispose 或释放所有权，因为保留失败的 child 会让其祖先永久固定在 `waiting`。如果 child 归 parent 所有，管理器随后会通过 `SessionHeader.parentSession` 解析在线 parent，并从其 `ownedChildren` 中移除 child 会话 id；由用户恢复且没有在线 owner 的 child 则没有需要释放的所有权记录。管理器拆卸使用相同的 child-first 顺序。 + +用户冷恢复会创建一次激活，但不会将其加入历史 parent 的 `ownedChildren`。如果直接 parent 随后向这个在线激活提交工作，且该 parent 自身由继续执行管理器管理，准入过程会在消息入队前建立所有权；非继续执行 parent 仍位于等待图之外。 + +MVP 会一直保留所有权，直至 child 激活完成 dispose。后续改进可以更早释放限定到请求的 lease，但这需要精确关联轮次完成，而本 Task-free 提案特意不增加该机制。 + +顶层拆卸由宿主负责，而不表示为另一次激活。宿主首先要求管理器同步进入 draining，拒绝新的创建、恢复和投递准入；然后按 child-first 顺序 dispose 整个在线激活森林，并等待全部 `AgentHandle.dispose()` 调用。只有该 drain 结算后，宿主才能 dispose 顶层 Agent 和管理器作用域。管理器卸载使用相同的 drain，并涵盖由用户恢复且没有在线 owner 的激活。 + +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 会话。 + +### 延后的报告投递 + +MVP 不暴露 `report` 工具，也不提供从 child 到 parent 的内容投递或自动唤醒 parent。持久化 child 会话仍是 child 详细输出的来源。 + +后续提案可以增加一个普通的面向模型 `report(output)` 工具；模型在一个轮次中可以调用它零次或多次。其投递策略可以区分静默注入 parent 与唤醒 parent；接收方选择、确认、持久性和重试语义均与该工具一并延后决定。增加报告投递无需引入另一个激活状态或执行队列。 + +### 延后的 steering（中途引导） + +MVP 不暴露 subagent steering 操作。parent 和用户的继续执行消息始终开启后续 FIFO 轮次，因此继续执行层不存储当前轮次控制方，也不新增能够感知控制方的 Agent 准入 seam。 + +后续宿主 UI 可以分别暴露 **Steer** 和 **Follow up** 操作。用户 steering 必须严格且仅限在线使用：只有当激活接受下一步骤时，它才能调用现有的 Agent steering 路径；其他情况必须拒绝，而且绝不能转为排队或冷恢复。是否通过面向模型的工具暴露 parent steering 仍需单独设计，因为不同的工具名称可以表达意图，却不能确定 parent 是否可以修改由用户控制的轮次。 + +### 权限与来源 + +权限来自可信宿主交互或确切的在线 Agent 工具上下文。`MessageSource` 和 `senderSessionId` 是准入后的持久化来源信息，不是由调用方控制的权限。 + +MVP 授权宿主用户和持久化 child 的直接 parent。系统会根据经过身份认证的 parent Agent 检查 `SessionHeader.parentSession`，然后才将 child 注册到该 parent 的 `ownedChildren`。其他 Agent、祖先、团队和工作流仍被拒绝，直至系统具备显式权限协议。 + +用户权限可以在 parent 不在线时冷恢复 child。由 parent 发起的投递要求 parent 在准入时在线，并通过所有权关系使其继续在线。 + +### 持久性、dispose 与恢复 + +没有 Task 后，系统不再提供 `task_output`、`task_kill`、Task 状态、逐消息结果 promise 或公开 subagent 取消操作。调用方 signal 只能在 inbox 接受消息前中止 start 或 follow-up。消息被接受后，parent 和用户都不能通过 `ctx.subagents` 取消该消息、轮次或激活；`Agent.cancel()` 仍是底层 Agent 能力，但本 MVP 不通过 subagent 服务暴露它。 + +宿主和管理器拆卸仍是覆盖整个生命周期的停止路径。它会关闭准入，按 child-first 顺序 dispose 每个在线激活森林，并保留持久化会话。 + +每个轮次都会请求执行会话持久性检查点，激活最终结算时，管理器必须检查 `ctx.sessions.flush()`，而不能忽略其布尔结果。`true` 确认至少有一个持久性 listener 参与，且所有 listener 都成功结算。`false` 或 rejection 会报告 `DURABILITY_FAILED`；普通后台结算会记录该生命周期失败，显式的宿主或管理器 drain 则会在所有分支结算后，将其纳入聚合 rejection。无论结果如何，管理器仍会 dispose handle 并释放所有权，后续恢复时持久化 child 状态可能缺失或陈旧。 + +只有实际写入 child 会话日志的消息，才能根据其准入来源重建；仅被 inbox 接受并不提供重启保证。 + +会话和描述符的持久化状态可在重启后保留。激活状态、Agent inbox 内容和所有权图都是进程内状态。进程崩溃可能丢失已被接受但仍留在 inbox、尚未写入会话日志的初始提示词或 follow-up。会话和描述符可能保留，因此后续获得授权的消息仍可冷恢复 child，但丢失的消息不会自动回放。恢复已接受但未完成或未写入日志的消息需要持久化 inbox 协议，本提案不隐含该能力。 + +### 范围 + +MVP 覆盖可继续的进程内 child，一次性委派保持不变。远程提供方必须具备单独的激活 handle，以及等价的认证控制与 child-first 完全停稳契约，才能支持同样的行为。 + +MVP 不新增 subagent steering 操作、报告工具、从 child 到 parent 的内容投递、自动唤醒 parent、持久化邮箱、跨进程 lease、中断 inbox 工作的自动回放、团队权限、工作流权限、公开 subagent 取消操作、新的在线激活数量或后代总数限制，以及运行时缓存。现有委派深度策略保持不变。 + +## 曾考虑的替代方案 + +**保留由 Task 支撑的激活。** Task 可以提供通用状态、结果收集和取消，但使用 Task 投递会话会产生第二条队列，并重复轮次所有权。本提案放弃这些通用 Task 控制，让 Agent inbox 成为唯一执行顺序。 + +**每个 `next-turn` 创建一次激活。** 这会恢复独立的结果与取消边界，但需要在 Agent inbox 旁维护管理器 FIFO，还会使所保留的 Agent 跨越人为划分的激活边界。每个驻留周期对应一次激活更小，也直接跟随 `AgentHandle` 生命周期。 + +**等待期间 dispose Agent。** child 仍属于上一个进程内所有权图时重建 parent，需要持久化所有权与拆卸协议。只为尚未完成的所有权图保留 `AgentHandle`，可以在不让已结算历史驻留的前提下，保留 child-first 拆卸。 + +**让提供方通过 Agent handle 创建、恢复 child 或投递消息。** 初始提供方只持有 `prepareContinuable()` 及其分离式创建规格这一项差异：child 是全新启动，还是带有 parent 前缀。管理器必须通过私有 activation-owner 作用域自行调用 `ctx.agents.create()`，使该作用域成为每个 handle 的结构化所有者。持久化的进程内会话已经包含初始前缀及通用重建描述符，消息投递则属于 Agent inbox。让提供方持有任何后续 handle、`SubagentRun` 或消息所有权，会保留一条没有 MVP 行为可承载的 seam，还会因不必要的在线 parent 输入使用户冷恢复更加复杂。 + +**在 MVP 中增加报告投递。** 可重复调用的面向模型工具与该生命周期兼容，但静默投递还是唤醒投递、接收方选择、确认、持久性和重试行为都是独立的产品决策。延后该工具，可以让首个版本专注于会话准入与驻留，又不限制后续策略。 + +**将 `SessionHeader.parentSession` 视为在线所有权。** 持久化谱系不能证明历史 parent 当前持有 child。在线 parent 的 `ownedChildren` 成员关系会记录进程内关系，而不改变持久化来源。 + +**在单独的 link 中保留确切的 parent Agent。** parent 激活已经持有自身 `AgentHandle`，而且 `ownedChildren` 会在 child 仍然在线时阻止该激活 dispose。因此，通过会话 id 解析 parent 已经足够，也可以避免冗余的运行时引用。 + +**为 parent 消息维护单独队列。** 第二个 FIFO 会让它和 Agent 已接受的用户消息之间顺序不明确。单个 Agent inbox 为两种来源提供唯一且可观察的顺序。 + +**在 MVP 中暴露 subagent steering。** 用户 steering 可以是严格且仅限在线使用的宿主操作，但 parent steering 需要当前轮次控制方状态，以保护由用户控制的轮次。首个版本将每条继续执行消息都排队，可以避免引入该状态及其准入竞争。后续 UI 可以新增一项仅限用户的独立操作，而不改变 follow-up 排序。 + +**返回 subagent 专属的投递路由。** `started`、`queued` 和 `resumed` 等标签重复了激活与 inbox 状态，却没有给调用方提供独立结果。复用 `AgentMessageId` 和现有 inbox 事件，可以让投递关联继续由其所属的 Agent 契约承载。 + +**使用 child 引用计数。** 计数无法识别哪个 child 仍持有拆卸工作，也允许重复递减错误。身份集合会显式保留取消和 dispose 义务。 + +## 验收标准 + +- 可继续 child 至多拥有一个在线激活和一个 Agent inbox；继续执行管理器没有激活 FIFO 或 queued 激活状态。 +- `SubagentProvider.prepareContinuable?()` 只返回分离式 `ContinuableCreateSpec`；配置为 continuable 时要求具备该能力，而 `backgroundMode` 仍是独立的策略选择。 +- 管理器通过私有 activation-owner 作用域调用 `ctx.agents.create()`，安装返回的 `AgentHandle` 并建立 parent 所有权，调用 `Agent.followup(initialPrompt)`，然后在 inbox 接受消息并产生 `AgentMessageId` 时返回 `{ childId, messageId }`，而不等待轮次开始或消息写入会话日志。 +- 初始提示词被 inbox 接受前的每条失败路径都会导致操作被拒绝且不返回 id，并回滚已创建的任何 handle、激活和 parent `ownedChildren` 成员关系。 +- 冷恢复由继续执行管理器调用 `ctx.agents.resume()`，绝不通过初始 subagent 提供方分发；`SubagentProvider.resume?()` 和 `SubagentProviderResumeRequest` 均不存在。 +- 可继续激活直接持有 `AgentHandle`，绝不创建、包装或保留 `SubagentRun`；`SubagentProvider.start()` 和 `SubagentRun` 只用于 one-shot，且没有 `SubagentRun.steer?()`。 +- 用户可以在不加载历史 parent 的前提下冷恢复持久化 child。 +- `followup()` 只接受可信 parent 或用户权限；持久化消息来源信息不能授权投递。 +- Parent 和用户的继续执行消息始终使用 `Agent.followup()` 并共享其 inbox FIFO，包括一种来源排在另一种来源之后，以及 child 已有开放轮次的情况。 +- `ctx.subagents.followup()` 及其 `send_message` 适配器只返回已接受的 `AgentMessageId`；继续执行层不接受投递 target，也不定义 subagent 专属路由结果。 +- MVP 不暴露公开 subagent 取消操作；调用方 signal 只能在 inbox 接受消息前停止 start 和 follow-up，宿主和管理器拆卸则保留 child-first 全局清理。 +- MVP 不暴露 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，使并发的正常结算保持幂等。 +- MVP 不暴露 `report` 工具，不提供从 child 到 parent 的内容投递，也不自动唤醒 parent。 +- 会话日志只能根据准入来源重建实际写入的消息；已被 inbox 接受但未写入日志的消息没有重启保证。 +- 可继续 subagent 路径不创建或依赖 Task、`TaskId`、Task 完成通知、Task 取消或中间的带结果执行包装层。 +- 单元覆盖固定 `startContinuable()` 在 inbox 接受消息时的返回边界、每条接受前失败路径的完整回滚、接受前后两个阶段的调用方 signal 所有权，以及已接受但未写入日志的消息不会自动回放。 +- 单元覆盖固定仅由驻留状态决定的路由表、单 inbox 顺序、通过 inbox 事件关联 `AgentMessageId`、在开放轮次期间 follow-up、等待唤醒、冷恢复、所有权注册与释放、child-first dispose、发送与 dispose 的竞争、最终持久性检查点返回 `false` 和 rejection 时都不泄漏所有权，以及不存在公开 subagent 取消、steering 和报告工具这一事实。 +- 一项无密钥整套应用快照覆盖 parent 委派、parent 与用户混合的 follow-up 排队、不存在 subagent steering、报告投递和自动唤醒 parent、保留等待中的 `AgentHandle` 以及 child-first dispose。 + +## 风险 + +移除 Task 会放弃通用后台工作检查、结果收集和精确 Task 取消。如果这些产品功能成为需求，就需要不会重新引入第二条执行队列的请求 ticket 或 inbox 能力。 + +在后代运行期间保留激活，会按尚未完成所有权图的规模消耗 Agent 资源。现有委派深度策略仍会限制嵌套层级，但 MVP 不新增在线激活数量或后代总数限制；已结算的历史会话不保留 `AgentHandle`。 + +进程内 inbox 和所有权图无法协调两个 harness 进程。允许多个进程并发访问同一持久化存储的部署，仍需要持久化 lease 和邮箱协议。 + +没有报告投递时，完成 child 轮次既不会把内容发送给历史 parent，也不会唤醒它。输出会保留在持久化 child 会话中，直至调用方检查该 transcript 或提交另一个经过授权的轮次。后续报告工具可以增加静默投递或唤醒投递，而无需改变激活生命周期。 + +将每条继续执行消息排队，意味着 parent 无法立即纠正正在进行的 child 轮次；纠正操作会在下一个轮次执行。后续仅限用户的 UI steering 操作可以缩短该延迟，而无需在 MVP 中引入 parent 与用户之间的控制方策略。 + +最终持久性检查点失败时，运行时所有权图仍可完成 drain，但持久化 child 状态会缺失或陈旧。该失败会以 `DURABILITY_FAILED` 的形式被观测到；重试与修复需要单独的恢复设计。 diff --git a/packages/subagent/subagent-fork/src/index.ts b/packages/subagent/subagent-fork/src/index.ts index 8461795120..f4708dc14f 100644 --- a/packages/subagent/subagent-fork/src/index.ts +++ b/packages/subagent/subagent-fork/src/index.ts @@ -12,12 +12,13 @@ import z from 'schemastery' import type { SessionEvent } from '@deepseek-ai/dsh-session' import type { Agent } from '@deepseek-ai/dsh-agent' import type { + ContinuableCreateRequest, + ContinuableCreateSpec, SubagentCapabilities, SubagentProvider, - SubagentProviderResumeRequest, - SubagentProviderStartRequest, + SubagentStartRequest, } from '@deepseek-ai/dsh-subagent' -import { resumeInProcessRun, startInProcessRun } from '@deepseek-ai/dsh-subagent-inprocess' +import { startInProcessRun } from '@deepseek-ai/dsh-subagent-inprocess' export const name = 'subagent-fork' // `tools` is deliberately NOT injected — same rationale as subagent-spawn: the @@ -64,7 +65,7 @@ class ForkProvider implements SubagentProvider { constructor(readonly name: string) {} - start(request: SubagentProviderStartRequest) { + start(request: SubagentStartRequest) { const seed = completedTurnPrefix(request.parent) return startInProcessRun(request, { // Only pass a seed when there's a completed turn to inherit; an empty seed @@ -73,11 +74,12 @@ class ForkProvider implements SubagentProvider { }) } - resume(request: SubagentProviderResumeRequest) { - // Cold resume loads the child's OWN persisted transcript, which already - // contains the completed-turn prefix captured at initial creation; it - // never forks the parent's newer history again. - return resumeInProcessRun(request) + prepareContinuable(request: ContinuableCreateRequest): Promise { + // The fork prefix is captured ONCE, at creation: it becomes part of the + // child's own durable transcript, so a later cold resume replays that + // prefix instead of re-forking the parent's newer history. + const seed = completedTurnPrefix(request.parent) + return Promise.resolve(seed.length > 0 ? { seed } : {}) } } diff --git a/packages/subagent/subagent-inprocess/src/index.ts b/packages/subagent/subagent-inprocess/src/index.ts index 38fd418f39..ddbcf8753e 100644 --- a/packages/subagent/subagent-inprocess/src/index.ts +++ b/packages/subagent/subagent-inprocess/src/index.ts @@ -1,24 +1,32 @@ /** - * Shared driver for in-process subagent providers. The agent factory's + * Shared driver for in-process ONE-SHOT subagent providers. The agent factory's * creation transaction owns unpublished setup and rollback; after publication * the returned AgentHandle is the one quiescent lifecycle owner held by the * provider's caller. * + * Continuable children never come through here: the continuation manager + * composes and drives them directly, so this driver owns exactly one turn with + * one result. + * * @module @deepseek-ai/dsh-subagent-inprocess */ import { randomUUID } from 'node:crypto' import type { Context } from 'cordis' -import type { Agent, AgentHandle, AgentOptions } from '@deepseek-ai/dsh-agent' +import type { Agent, AgentHandle } from '@deepseek-ai/dsh-agent' import { findLastMessageTurnEnd, SessionId, type SessionEvent, type TurnEndReason } from '@deepseek-ai/dsh-session' -import { createUserMessage, errorChain, type ContentBlock, type MessageSource } from '@deepseek-ai/dsh-llm' -import { assertSubagentMaxDepth, delegationDepthOf, SubagentError } from '@deepseek-ai/dsh-subagent' +import { createUserMessage, type ContentBlock } from '@deepseek-ai/dsh-llm' +import { + applyChildComposition, + assertSubagentMaxDepth, + childSessionMeta, + resolveChildAgentOptions, + resolveChildDepth, +} from '@deepseek-ai/dsh-subagent' import type { - SubagentDescriptorData, - SubagentProviderResumeRequest, - SubagentProviderStartRequest, SubagentResult, SubagentRun, + SubagentStartRequest, SubagentStopReason, } from '@deepseek-ai/dsh-subagent' // Type-only: make `ctx.get('sandboxPolicy')` / `ctx.get('approval')` resolve @@ -36,14 +44,6 @@ export { STRUCTURED_OUTPUT_INSTRUCTION, } from './structured.ts' -/** Thrown when starting a child would exceed the requested depth cap. */ -class SubagentDepthError extends Error { - constructor(public readonly attemptedDepth: number, public readonly maxDepth: number) { - super(`subagent depth ${attemptedDepth} exceeds maxDepth ${maxDepth}`) - this.name = 'SubagentDepthError' - } -} - /** Map a session turn outcome to the subagent seam's terminal vocabulary. */ function toStopReason(reason: TurnEndReason | undefined): SubagentStopReason { switch (reason?.kind) { @@ -67,76 +67,31 @@ export interface InProcessRunOptions { readonly seed?: SessionEvent[] } -/** Whether one activation must prove its final state durable before success. */ -type Durability = 'best-effort' | 'required' - -/** Activation-specific inputs to the shared in-process driver. */ -interface DriveTurnOptions { - readonly durability: Durability - /** Attribution for a resumed activation's follow-up prompt. */ - readonly source?: MessageSource - readonly structured?: StructuredAttachment -} - /** Error used when cancellation wins before the child publication boundary. */ function prePublicationAbort(): Error { return new Error('subagent request was aborted before child publication') } /** - * Register the one-shot child-scoped contribution that appends the durable - * `subagent/descriptor` event. The prepended `agent/prompt-submit` wrapper - * appends before downstream admission can block or throw. Allowed admission - * opens the initial turn afterward; the final required checkpoint also - * persists the descriptor when no turn opens. - */ -function attachDescriptorAppend(childCtx: Context, descriptor: SubagentDescriptorData): void { - childCtx.once('agent/prompt-submit', (agent, _message, _signal, next) => { - agent.session.append('subagent/descriptor', descriptor) - return next() - }, { prepend: true }) -} - -/** - * Establish and drive one in-process child. Fulfillment means the agent is - * already published in the registry; rejection means the agent factory's + * Establish and drive one in-process one-shot child. Fulfillment means the agent + * is already published in the registry; rejection means the agent factory's * creation transaction and any partially-created child have reached quiescence. - * A `request.continuation` publishes exactly its stable child id and appends - * its descriptor before the child's initial prompt admission. * @param request - the trusted typed start request, including its required signal. * @param options - the optional fork seed. * @returns a ready holder-owned run. */ export async function startInProcessRun( - request: SubagentProviderStartRequest, + request: SubagentStartRequest, options: InProcessRunOptions, ): Promise { assertSubagentMaxDepth(request.maxDepth) if (request.signal.aborted) throw prePublicationAbort() const parent = request.parent - const childDepth = delegationDepthOf(parent) + 1 - if (!Number.isSafeInteger(childDepth)) { - throw new RangeError('subagent child depth exceeds the safe-integer range') - } - if (request.maxDepth !== undefined && childDepth > request.maxDepth) { - throw new SubagentDepthError(childDepth, request.maxDepth) - } + const childDepth = resolveChildDepth(parent, request.maxDepth) - // A continuable delegation names the durable conversation up front; the - // provider publishes exactly that id instead of allocating one internally. - const childId = request.continuation?.sessionId ?? SessionId(randomUUID()) - const seedLength = options.seed?.length ?? 0 - const parentHeader = parent.session.header - const parentProvider = parent.options.provider - const parentModel = parent.options.model - const parentMaxTokens = parent.options.maxTokens - const agentOptions: AgentOptions = { - ...parentProvider !== undefined ? { provider: parentProvider } : {}, - ...parentModel !== undefined ? { model: parentModel } : {}, - ...parentMaxTokens !== undefined ? { maxTokens: parentMaxTokens } : {}, - ...request.agentOptions, - subagentDepth: childDepth, - } + const childId = SessionId(randomUUID()) + const seed = options.seed + const activationBoundary = seed?.length ?? 0 // Capture before the first await: a later parent switch belongs to the // parent's future. @@ -145,6 +100,8 @@ export async function startInProcessRun( let structured: StructuredAttachment | undefined const setup = (childCtx: Context): void => { + // Inherited overrides land on the child's own log, so its effective policy + // is reconstructable from that log alone. const childSession = (childCtx.agent as Agent).session if (inheritedMode !== undefined) { childSession.append('sandbox/mode', { mode: inheritedMode, source: 'delegation' }) @@ -152,29 +109,20 @@ export async function startInProcessRun( if (inheritedPolicy !== undefined) { childSession.append('approval/policy', { policy: inheritedPolicy, source: 'delegation' }) } - if (request.persona !== undefined) { - childCtx.systemPrompt.section({ name: 'deployment:persona', order: 0, text: request.persona }) - } - if (request.toolFilter !== undefined) childCtx.tools.restrict(request.toolFilter) + applyChildComposition(childCtx, { + persona: request.persona, + toolFilter: request.toolFilter, + }) if (request.outputSchema !== undefined) { structured = attachStructuredRuntime(childCtx, request.outputSchema) } - if (request.continuation !== undefined) { - attachDescriptorAppend(childCtx, request.continuation.descriptor) - } } const handle = await parent.ctx.agents.create({ sessionId: childId, - meta: { - ...parentHeader.cwd !== undefined ? { cwd: parentHeader.cwd } : {}, - parentSession: parentHeader.id, - // Durable: the recursion budget must survive persistence and resume. - delegationDepth: childDepth, - ...seedLength > 0 ? { seedLength } : {}, - }, - ...options.seed === undefined ? {} : { seed: options.seed }, - agentOptions, + meta: childSessionMeta(parent, childDepth, activationBoundary), + ...seed !== undefined ? { seed } : {}, + agentOptions: resolveChildAgentOptions(parent, request.agentOptions, childDepth), signal: request.signal, setup, }) @@ -183,62 +131,15 @@ export async function startInProcessRun( request.signal, request.prompt, childId, - seedLength, - { - durability: request.continuation === undefined ? 'best-effort' : 'required', - ...structured === undefined ? {} : { structured }, - }, + activationBoundary, + structured, ) } /** - * Reconstruct a persisted continuable child under the live parent's scope and - * drive one follow-up turn. The resumed session's own transcript is the seed - * (loaded through the parent's persistence-backed registry `resume`), so a - * fork child never re-forks current parent history; the persisted header - * remains authoritative for lineage and the delegation-depth floor. - * @param request - the fully resolved resume request from the continuation manager. - * @returns a fresh ready holder-owned run for this activation. - */ -export async function resumeInProcessRun(request: SubagentProviderResumeRequest): Promise { - if (request.signal.aborted) throw prePublicationAbort() - const descriptor = request.descriptor - const agentOptions: AgentOptions = { - ...descriptor.agentProvider !== undefined ? { provider: descriptor.agentProvider } : {}, - ...descriptor.agentModel !== undefined ? { model: descriptor.agentModel } : {}, - } - const setup = (childCtx: Context): void => { - if (descriptor.persona !== undefined) { - childCtx.systemPrompt.section({ name: 'deployment:persona', order: 0, text: descriptor.persona }) - } - if (descriptor.toolFilter !== undefined) childCtx.tools.restrict(descriptor.toolFilter) - } - - const handle = await request.parent.ctx.agents.resume({ - resumeSessionId: request.sessionId, - agentOptions, - signal: request.signal, - setup, - }) - // The result boundary is this activation's own work: everything already in - // the resumed transcript belongs to earlier turns. - const resumePoint = handle.agent.session.events.length - return driveTurn( - handle, - request.signal, - request.prompt, - request.sessionId, - resumePoint, - { durability: 'required', source: request.source }, - ) -} - -/** - * Drive one activation turn on a published child and wrap it as a run. The - * caller has already created or resumed the agent; this owns the - * signal-handoff race, the live abort listener, result collection past - * `boundary`, the continuable-run durability confirmation, confirmed - * steering, and disposal. + * Drive one turn on a published child and wrap it as a run. The caller has + * already created the agent; this owns the signal-handoff race, the live abort + * listener, result collection past `boundary`, and disposal. */ function driveTurn( handle: AgentHandle, @@ -246,10 +147,9 @@ function driveTurn( prompt: ContentBlock[], childId: SessionId, boundary: number, - options: DriveTurnOptions, + structured: StructuredAttachment | undefined, ): SubagentRun | Promise { const child = handle.agent - const { durability, source, structured } = options // Agent creation detaches its creation-only abort listener before returning. // Close the narrow handoff race before installing the live-run listener. if (signal.aborted) { @@ -265,30 +165,13 @@ function driveTurn( const result: Promise = (async () => { try { - child.followup(createUserMessage({ content: prompt, source: source ?? { kind: 'user' } })) + child.followup(createUserMessage({ content: prompt, source: { kind: 'user' } })) await child.whenIdle() - if (durability === 'required') { - try { - const participated = await child.ctx.sessions.flush(child.session) - if (!participated) { - throw new Error(`session "${child.id}" required durability checkpoint has no registered listener`) - } - } catch (error: unknown) { - if (!signal.aborted) { - throw new SubagentError( - `subagent "${childId}" durability checkpoint failed; the latest child state was not confirmed persisted and may be unavailable or stale on resume: ${errorChain(error)}`, - 'DURABILITY_FAILED', - { cause: error }, - ) - } - } - } return readResult( child, boundary, flags.cancelled, structured ? { captured: structured.captured() } : undefined, - durability === 'required' && signal.aborted, ) } finally { signal.removeEventListener('abort', onAbort) @@ -304,23 +187,6 @@ function driveTurn( flags.cancelled = true return handle.dispose() }, - async steer(content: ContentBlock[], steeringSource: MessageSource): Promise { - // The status check and submission share one synchronous frame. An idle - // Agent.steer() would queue an untracked turn after this run's result. - if (child.status !== 'running') { - throw new Error(`subagent child "${childId}" is not running; the message was not delivered`) - } - // Avoid waiting for the structured terminal checkpoint when its outcome - // is already authoritative and synchronously visible. - if (structured?.captured() !== undefined) { - throw new Error(`subagent child "${childId}" already reported its structured result; the message was not delivered`) - } - const receipt = child.steer(createUserMessage({ content, source: steeringSource })) - const outcome = await receipt.outcome - if (outcome.status === 'rejected') { - throw new Error(`subagent child "${childId}" stopped before steering admission; the message was not delivered`) - } - }, } } @@ -330,7 +196,6 @@ function readResult( boundary: number, cancelled: boolean, structured?: { captured?: { value: unknown } | undefined }, - cancellationOwnsCompleted = false, ): SubagentResult { const own = child.session.events.slice(boundary) const lastMessage = own.findLast((event): event is SessionEvent<'assistant/message'> => event.type === 'assistant/message') @@ -338,13 +203,8 @@ function readResult( const output: ContentBlock[] = lastMessage?.data.message.content ?? [] const recorded = toStopReason(lastEnd?.data.reason) // Disposal can tear the owner down before the loop records its ordinary - // `aborted` end, yielding `disposed` instead. Activation cancellation during - // its final durability checkpoint also owns a recorded completed turn because - // the provider has not published that result yet. - const stopReason: SubagentStopReason = cancelled - && (recorded !== 'completed' || cancellationOwnsCompleted) - ? 'aborted' - : recorded + // `aborted` end, yielding `disposed` instead. + const stopReason: SubagentStopReason = cancelled && recorded !== 'completed' ? 'aborted' : recorded if (structured !== undefined) { if (structured.captured !== undefined) { return { output, structured: structured.captured.value, stopReason } diff --git a/packages/subagent/subagent-spawn/src/index.ts b/packages/subagent/subagent-spawn/src/index.ts index 0080c31521..7dceeac2ae 100644 --- a/packages/subagent/subagent-spawn/src/index.ts +++ b/packages/subagent/subagent-spawn/src/index.ts @@ -9,12 +9,12 @@ import type { Context } from 'cordis' import z from 'schemastery' import type { + ContinuableCreateSpec, SubagentCapabilities, SubagentProvider, - SubagentProviderResumeRequest, - SubagentProviderStartRequest, + SubagentStartRequest, } from '@deepseek-ai/dsh-subagent' -import { resumeInProcessRun, startInProcessRun } from '@deepseek-ai/dsh-subagent-inprocess' +import { startInProcessRun } from '@deepseek-ai/dsh-subagent-inprocess' export const name = 'subagent-spawn' // `tools` is deliberately not injected: the child factory already provides it during setup, @@ -45,17 +45,17 @@ class SpawnProvider implements SubagentProvider { constructor(readonly name: string) {} - start(request: SubagentProviderStartRequest) { + start(request: SubagentStartRequest) { // Fresh child: no seed. The shared driver mints ids, stamps cwd/lineage/ // depth, drives the one-shot (including the structured capture when the // request carries an outputSchema), and maps the result. return startInProcessRun(request, {}) } - resume(request: SubagentProviderResumeRequest) { - // Cold resume reconstructs the persisted child from its own transcript - // under the live parent scope; the shared driver drives the follow-up turn. - return resumeInProcessRun(request) + prepareContinuable(): Promise { + // A spawned child starts fresh, so it contributes no seed; the continuation + // manager owns every later operation on it. + return Promise.resolve({}) } } diff --git a/packages/subagent/subagent/src/child-agent.ts b/packages/subagent/subagent/src/child-agent.ts new file mode 100644 index 0000000000..f93e5d5dff --- /dev/null +++ b/packages/subagent/subagent/src/child-agent.ts @@ -0,0 +1,128 @@ +/** + * Shared in-process child composition: the delegation-depth budget, the + * durable session metadata, the resolved child `AgentOptions`, and the scoped + * setup a child agent needs. Both the one-shot provider driver and the + * continuation manager compose children this way, so depth accounting and + * lineage stamping have one home. + * + * @module @deepseek-ai/dsh-subagent/child-agent + */ + +import type { Context } from 'cordis' +import type { Agent, AgentOptions, CreateAgentOptions } from '@deepseek-ai/dsh-agent' +import type { SessionId } from '@deepseek-ai/dsh-session' +import type { ToolRestriction } from '@deepseek-ai/dsh-tools' +import { delegationDepthOf } from './depth.ts' + +/** Thrown when starting a child would exceed the requested depth cap. */ +export class SubagentDepthError extends Error { + constructor(public readonly attemptedDepth: number, public readonly maxDepth: number) { + super(`subagent depth ${attemptedDepth} exceeds maxDepth ${maxDepth}`) + this.name = 'SubagentDepthError' + } +} + +/** + * Resolve the child's delegation depth from its parent and enforce an optional + * cap. The persisted parent header is the monotone floor, so a resumed parent + * cannot delegate as if it were top-level. + * @param parent - the delegating parent agent. + * @param maxDepth - optional absolute cap the resolved depth must not exceed. + * @returns the child's non-negative safe-integer depth. + * @throws {SubagentDepthError} when the resolved depth exceeds `maxDepth`. + * @throws {RangeError} when the resolved depth leaves the safe-integer range. + */ +export function resolveChildDepth(parent: Agent, maxDepth: number | undefined): number { + const childDepth = delegationDepthOf(parent) + 1 + if (!Number.isSafeInteger(childDepth)) { + throw new RangeError('subagent child depth exceeds the safe-integer range') + } + if (maxDepth !== undefined && childDepth > maxDepth) { + throw new SubagentDepthError(childDepth, maxDepth) + } + return childDepth +} + +/** + * Resolve the child's `AgentOptions`: the parent's provider/model/maxTokens + * route unless the request overrides it, stamped with the child's own + * delegation depth. + * @param parent - the delegating parent whose route the child inherits. + * @param requested - per-child overrides, if any. + * @param childDepth - the resolved delegation depth to stamp. + * @returns the resolved options for `ctx.agents.create()`. + */ +export function resolveChildAgentOptions( + parent: Agent, + requested: AgentOptions | undefined, + childDepth: number, +): AgentOptions { + const parentProvider = parent.options.provider + const parentModel = parent.options.model + const parentMaxTokens = parent.options.maxTokens + return { + ...parentProvider !== undefined ? { provider: parentProvider } : {}, + ...parentModel !== undefined ? { model: parentModel } : {}, + ...parentMaxTokens !== undefined ? { maxTokens: parentMaxTokens } : {}, + ...requested, + subagentDepth: childDepth, + } +} + +/** + * Build the child session's durable creation metadata: the parent's workspace, + * its direct lineage, the recursion budget that must survive persistence, and + * the seed boundary that separates inherited parent history from child work. + * @param parent - the delegating parent agent. + * @param childDepth - the resolved delegation depth to persist. + * @param lineageSeedLength - how many leading events came from the parent's log. + * @returns the `meta` for `ctx.agents.create()`. + */ +export function childSessionMeta( + parent: Agent, + childDepth: number, + lineageSeedLength: number, +): NonNullable { + const parentHeader = parent.session.header + return { + ...parentHeader.cwd !== undefined ? { cwd: parentHeader.cwd } : {}, + parentSession: parentHeader.id, + // Durable: the recursion budget must survive persistence and resume. + delegationDepth: childDepth, + ...lineageSeedLength > 0 ? { seedLength: lineageSeedLength } : {}, + } +} + +/** The scoped composition a child agent's creation window applies. */ +export interface ChildComposition { + /** Per-child persona shadowing the deployment persona. */ + readonly persona?: string | undefined + /** Per-child tool scoping. */ + readonly toolFilter?: ToolRestriction | undefined +} + +/** + * Apply one child's scoped composition inside its creation window: a shadowing + * persona section and a tool restriction, both owned by the child's scope and + * therefore invisible to its parent and siblings. + * @param childCtx - the child agent's scoped creation context. + * @param composition - the persona and tool filter to install. + */ +export function applyChildComposition(childCtx: Context, composition: ChildComposition): void { + if (composition.persona !== undefined) { + childCtx.systemPrompt.section({ name: 'deployment:persona', order: 0, text: composition.persona }) + } + if (composition.toolFilter !== undefined) childCtx.tools.restrict(composition.toolFilter) +} + +/** Identity and lineage inputs shared by every in-process child creation. */ +export interface ChildCreateInputs { + /** The child's reserved session id. */ + readonly sessionId: SessionId + /** The delegating parent agent. */ + readonly parent: Agent + /** The resolved delegation depth. */ + readonly childDepth: number + /** How many leading seed events came from the parent's log. */ + readonly lineageSeedLength: number +} diff --git a/packages/subagent/subagent/src/continuation.ts b/packages/subagent/subagent/src/continuation.ts index 1f4aa54a0c..80021ce205 100644 --- a/packages/subagent/subagent/src/continuation.ts +++ b/packages/subagent/subagent/src/continuation.ts @@ -1,34 +1,43 @@ /** * Internal continuable-subagent manager: stable child ids, descriptor - * persistence and lookup by known child id, Task-backed activation, and - * steer-or-resume message routing behind `ctx.subagents`. + * persistence, activation admission, the live ownership graph, cold resume, + * and child-first disposal behind `ctx.subagents`. * - * Every continuable activation — initial or resumed, parent- or human-started - * — has exactly one Task and one result. Task settlement awaits the child - * result, disposes the run, and only then records the outcome, so a terminal - * Task leaves the durable child session but no live child Agent. Cancellation - * targets the whole activation: parent and human messages that joined one - * turn share its result and its `killed` outcome. + * A continuable child has one durable Session and at most one process-local + * {@link Activation} — one residency epoch for a reconstructed child Agent. An + * Activation is not a request, result, cancellation, or Task boundary: it may + * execute many FIFO turns and stays resident while descendants it created are + * still running. The Agent inbox is the only turn queue, so this manager owns + * residency while the Agent loop owns all turn ordering and execution. No + * continuable path creates a Task or an intermediate result-bearing wrapper. * * @module @deepseek-ai/dsh-subagent */ import { randomUUID } from 'node:crypto' import type { Context } from 'cordis' -import type { Agent } from '@deepseek-ai/dsh-agent' -import { HarnessError } from '@deepseek-ai/dsh-llm' -import type { ContentBlock, MessageSource } from '@deepseek-ai/dsh-llm' -import { SessionId } from '@deepseek-ai/dsh-session' -import type { SessionPersistence } from '@deepseek-ai/dsh-session-persistence' -import { foldSubagentDescriptor, snapshotSubagentDescriptor } from './descriptor.ts' import type { - SubagentProviderResumeRequest, - SubagentProviderStartRequest, - SubagentResult, - SubagentRun, - SubagentStartRequest, -} from './types.ts' -import type { TaskHooks, TaskId, TaskOutcome } from '@deepseek-ai/dsh-tasks' + Agent, + AgentHandle, + AgentOptions, + CreateAgentOptions, +} from '@deepseek-ai/dsh-agent' +import { createUserMessage, errorChain } from '@deepseek-ai/dsh-llm' +import type { ContentBlock, MessageId, MessageSource } from '@deepseek-ai/dsh-llm' +import { SessionId } from '@deepseek-ai/dsh-session' +import type { SessionEvent } from '@deepseek-ai/dsh-session' +import type { SessionPersistence } from '@deepseek-ai/dsh-session-persistence' +import type { ToolRestriction } from '@deepseek-ai/dsh-tools' +import { foldSubagentDescriptor, snapshotSubagentDescriptor } from './descriptor.ts' +import type { SubagentDescriptorData } from './descriptor.ts' +import { + applyChildComposition, + childSessionMeta, + resolveChildAgentOptions, + resolveChildDepth, +} from './child-agent.ts' +import { seedDescriptorTurn } from './descriptor-seed.ts' +import type { ContinuableCreateRequest, ContinuableCreateSpec, SubagentStartRequest } from './types.ts' import { SubagentError } from './error.ts' /** Attribution for a model coordinator's follow-up to one of its children. */ @@ -44,197 +53,251 @@ declare module '@deepseek-ai/dsh-llm' { } } +/** + * Who authorizes one continuable-subagent operation. Authority comes from a + * trusted host interaction or an exact live Agent tool context; durable + * {@link MessageSource} provenance never authorizes delivery. + */ +export type SubagentAuthority = + /** The exact live parent Agent whose tool context is making the call. */ + | { readonly kind: 'parent'; readonly agent: Agent } + /** A trusted host adapter acting for the human user. */ + | { readonly kind: 'user' } + /** What a caller asks for when starting a continuable background child. */ export interface ContinuableStartSpec { - /** The `ctx.subagents` provider to establish the child on. */ + /** The `ctx.subagents` provider whose continuable-creation capability establishes the child. */ readonly provider: string - /** One-line model-facing Task label (the delegation description). */ - readonly label: string /** - * The delegation request. The service resolves the stable child id and the - * durable descriptor, then supplies the Task-owned cancellation signal and - * `continuation` itself. + * The delegation request. The manager reserves the stable child id, resolves + * the durable descriptor, and composes the child itself. */ - readonly request: Omit + readonly request: Omit + /** Caller cancellation, owning the operation only until inbox acceptance. */ + readonly signal: AbortSignal } -/** Identities returned by a continuable start. */ +/** Identities returned once a continuable child accepted its initial prompt. */ export interface ContinuableStart { /** The durable child session id, stable across activations. */ readonly childId: SessionId - /** The initial activation's Task id. */ - readonly taskId: TaskId + /** The accepted initial prompt's inbox message id. */ + readonly messageId: MessageId } -/** - * Options for following up with one continuable child. - */ +/** Options for following up with one continuable child. */ export interface SubagentFollowupOptions { - /** Durable attribution retained on either live or resumed delivery. */ + /** Durable attribution retained on the delivered message; it grants no authority. */ readonly source: MessageSource - /** Caller cancellation for a live-delivery admission wait. */ + /** Caller cancellation, owning the operation only until inbox acceptance. */ readonly signal: AbortSignal } /** - * How a continuable follow-up was routed: - * `steered` joined the running activation's existing Task without creating a - * Task of its own; `started` created a fresh Task that cold-resumes the - * durable child with the content. Failure is an exception, never a result — - * undelivered content throws. + * The public residency state of one continuable child, derived from Agent + * quiescence and the owned-child set rather than a second state machine: + * `running` — the Agent has an active admission or turn, or waking inbox work; + * `waiting` — the Agent is quiescent but still owns undisposed children; + * `settled` — quiescent with every owned child disposed, so the manager + * disposes the `AgentHandle` and removes the Activation. */ -export type SubagentFollowupResult = - | { readonly route: 'steered'; readonly taskId: TaskId } - | { readonly route: 'started'; readonly taskId: TaskId } - -type StartProvider = (name: string, request: SubagentProviderStartRequest) => Promise -type ResumeProvider = (request: SubagentProviderResumeRequest) => Promise +export type ActivationState = 'running' | 'waiting' | 'settled' /** - * One child's current process-local activation: its Task and, after provider - * publication, its run. Installed before any provider or persistence await - * and removed only after run disposal and Task terminal publication. This - * exists solely so parent and human senders can find the same activation — it - * is not a durable catalog, admission reservation, or run-state machine. + * Lifecycle observer for one Activation's residency epoch, so continuable + * children emit the same start/end pair as one-shot runs. */ -interface ActiveActivation { - /** Assigned in the same synchronous frame as the install, when the Task registers. */ - taskId: TaskId | undefined - /** Filled when the provider publishes; `undefined` while starting or resuming. */ - run: SubagentRun | undefined - /** The activation-owned cancellation authority, created before any await. */ - readonly controller: AbortController - /** The producer's settlement (run disposed, outcome produced); assigned when the Task registers. */ - done: Promise | undefined - /** Resolved by the completion listener when the Task's terminal snapshot is recorded. */ - readonly terminal: PromiseWithResolvers +export interface ActivationObserver { + /** Publish the start edge once the epoch is resident. */ + start(): void + /** + * Publish the terminal edge exactly once. An epoch that never became resident + * emits nothing, because it has no start edge to pair. + * @param child - the child agent whose final output the edge reports, if any. + * @param failure - the teardown or durability failure, or `undefined` on success. + */ + settle(child: Agent | undefined, failure: unknown): void +} + +/** Hooks the manager needs from the owning service. */ +export interface ContinuationHost { + /** + * Resolve one provider's continuable-creation contribution, or reject when + * the provider is unknown or lacks the capability. + * @param name - the configured provider name. + * @param request - the reserved identity, delegating parent, and cancellation. + * @returns the provider's detached creation spec. + */ + prepareContinuable(name: string, request: ContinuableCreateRequest): Promise + /** + * Build the lifecycle observer for one Activation's residency epoch. + * @param provider - the provider name recorded in the durable descriptor. + * @param childId - the durable child session id. + * @param parent - the delegating parent for scoped dispatch, if any. + * @returns the observer whose edges this epoch publishes. + */ + observeActivation(provider: string, childId: SessionId, parent: Agent | undefined): ActivationObserver } /** - * Map a child result to the task outcome: completed carries final text, - * aborted is killed, and every other reason is failed without partial output. - * @param result - child terminal result. - * @returns outcome for the `ctx.tasks` registration. + * One residency epoch for a reconstructed continuable child Agent. It directly + * owns the published `AgentHandle`; the manager's private activation-owner + * scope is its structural Cordis owner. */ -function runOutcome(result: SubagentResult): TaskOutcome { - switch (result.stopReason) { - case 'completed': - return { status: 'completed', output: finalText(result.output) } - case 'aborted': - return { status: 'killed' } - case 'error': - case 'max-tokens': - case 'refusal': - return { status: 'failed', detail: result.stopReason } - // Merge-extensible reasons remain failures with their raw detail. - default: - return { status: 'failed', detail: String(result.stopReason) } +interface Activation { + /** The durable child this Activation is an epoch of. */ + readonly childId: SessionId + /** The provider name recorded in the durable descriptor. */ + readonly provider: string + /** The retained live Agent handle, disposed exactly once at settlement. */ + readonly handle: AgentHandle + /** + * 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 + * another runtime-incarnation reference. Non-empty blocks settlement. + */ + readonly ownedChildren: Set + /** The lifecycle observer that emits this epoch's start and terminal edges. */ + readonly observer: ActivationObserver + /** + * The memoized disposal transaction. Presence IS the admission cutoff: it is + * assigned synchronously when disposal begins, so no delivery can join a + * handle being torn down, and a racing delivery awaits it before cold-resuming + * a new Activation. Every converging releaser shares this one teardown. + */ + disposal: Promise | undefined + /** Renewed whenever a settlement watcher must re-observe quiescence. */ + poke: PromiseWithResolvers +} + +/** + * 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. + * @param activation - the Activation to inspect. + * @returns the in-flight or settled disposal, or `undefined` while resident. + */ +function disposalOf(activation: Activation): Promise | undefined { + return activation.disposal +} + +/** Whether one settlement attempt opened the disposal transaction. */ +type SettlementAttempt = + | { readonly settling: false } + | { readonly settling: true; readonly done: Promise } + +/** Serialize each durable child's delivery, release, and disposal. */ +class ChildLock { + private tails = new Map>() + + /** + * Run `operation` after every previously queued operation for `childId`. + * @param childId - the durable child whose operations are linearized. + * @param operation - the critical section to run in order. + * @returns the operation's own settlement. + */ + run(childId: SessionId, operation: () => Promise): Promise { + const previous = this.tails.get(childId) ?? Promise.resolve() + const result = previous.then(operation, operation) + // Absorb rejections in the chaining tail so one failed critical section + // cannot reject an unrelated later caller. + const tail = result.then(() => undefined, () => undefined) + this.tails.set(childId, tail) + void tail.then(() => { + if (this.tails.get(childId) === tail) this.tails.delete(childId) + }) + return result } } -/** Render infrastructure failure detail without hiding a durability diagnosis. */ -function runFailureDetail(error: unknown): string { - return error instanceof HarnessError && error.code === 'DURABILITY_FAILED' - ? error.message - : String(error) -} - /** - * Await the child result, dispose the run, then return its task outcome. Result - * and disposal failures become `failed`; when both fail, both details survive. - * @param run - live run to settle and release. - * @returns outcome after child resources are released. - */ -export async function settleRun(run: SubagentRun): Promise { - let outcome: TaskOutcome - try { - outcome = runOutcome(await run.result) - } catch (error: unknown) { - outcome = { status: 'failed', detail: runFailureDetail(error) } - } - try { - await run.dispose() - } catch (error: unknown) { - const prefix = outcome.detail === undefined ? '' : `${outcome.detail}; ` - return { status: 'failed', detail: `${prefix}dispose failed: ${String(error)}` } - } - return outcome -} - -/** Flatten a child's final output blocks to the task's final text. */ -function finalText(blocks: ContentBlock[]): string { - return blocks - .filter((block): block is Extract => block.type === 'text') - .map(block => block.text) - .join('') -} - -/** - * The continuable-subagent orchestration service. Tool schema and UI adapters - * are consumers of this one contract: parent and human messages route through - * {@link followup} and share one activation result and cancellation - * boundary, while foreground one-shot delegation keeps calling - * `ctx.subagents.start()` directly. + * The continuable-subagent orchestration service behind `ctx.subagents`. Tool + * schema and host adapters are consumers of this one contract; foreground + * one-shot delegation keeps calling `ctx.subagents.start()` and never enters + * this lifecycle. */ export class SubagentContinuationManager { - /** Child session id → its current activation. Process-local, never durable. */ - private activations = new Map() + /** Child session id → its live Activation. Process-local, never durable. */ + private activations = new Map() + private readonly locks = new ChildLock() + /** Structural Cordis owner of every Activation handle. */ + private readonly ownerCtx: Context + private draining = false constructor( private readonly ctx: Context, - private readonly startProvider: StartProvider, - private readonly resumeProvider: ResumeProvider, + private readonly host: ContinuationHost, ) { - // Terminal publication is one of the two removal conditions. The exact - // Task id pins the resolution to this activation, never a later same-child one. - ctx.tasks.onTaskDone((snapshot) => { - for (const activation of this.activations.values()) { - if (activation.taskId === snapshot.id) activation.terminal.resolve() - } - }) - // TaskService deliberately keeps producer Tasks alive across a - // follow-up-tool or producer reload, so this manager's disposal must not - // strand the activations it can no longer route to: cancel each one and - // await producer settlement (run disposal) before releasing the map. The - // effect-scoped onTaskDone listener above is already gone by then, so - // terminal publication is resolved here instead of waiting forever. - ctx.effect(() => async () => { - const active = [...this.activations.values()] - this.activations.clear() - for (const activation of active) { - activation.controller.abort('subagent continuation manager disposed') - activation.terminal.resolve() - } - await Promise.allSettled(active.map((activation) => { - /* v8 ignore next 2 -- TaskService invokes `run` synchronously before `start` returns; - * every retained activation has `done`, while registration failure removes it. */ - if (activation.done === undefined) return Promise.resolve() - return activation.done - })) - }, 'subagents.continuations()') + // Ordinary Cordis owner effects unwind in reverse registration order, which + // cannot express the dynamic child graph. Register the private scope's + // structural disposer FIRST and the drain SECOND, so reverse unwind invokes + // the drain before releasing the scope; a cleanup effect on the same scope + // as the Agent handles would let structural handle disposal bypass + // child-first ordering. + const scope = ctx.plugin(function activationOwner() {}) + this.ownerCtx = scope.ctx + ctx.effect(function* (this: SubagentContinuationManager) { + yield scope.dispose + yield () => this.drain() + }.bind(this), 'subagents.continuations()') } /** - * Start a continuable background child: allocate its stable session id, - * snapshot its durable descriptor, and register the initial activation's - * Task. A synchronous validation failure (a non-JSON descriptor input, - * missing persistence, Task preflight) throws without creating a Task; the - * method otherwise returns both identities immediately, without waiting for - * child publication or descriptor durability. Asynchronous startup failure - * settles the returned Task as `failed` (or `killed` when cancelled) after - * any published run is disposed, which can leave an unmaterialized child id - * that later by-id operations report as unavailable. - * @param spec - provider, Task label, and the delegation request. - * @returns the stable child id and the initial activation's Task id. + * Whether this manager still admits new materialization and delivery. Host + * teardown closes admission synchronously through {@link enterDraining}. + * @returns true once draining began. */ - startContinuable(spec: ContinuableStartSpec): ContinuableStart { + get isDraining(): boolean { + return this.draining + } + + /** + * Close admission synchronously: reject new creation, cold resume, and + * delivery so a host can drain the live Activation forest without racing new + * work. Idempotent. + */ + enterDraining(): void { + this.draining = true + } + + /** + * Read one durable child's live residency state. + * @param childId - the durable child session id. + * @returns its Activation state, or `undefined` when no Activation is live. + */ + activationState(childId: SessionId): ActivationState | undefined { + const activation = this.activations.get(childId) + if (activation === undefined) return undefined + return this.stateOf(activation) + } + + /** + * Start one continuable background child: reserve its durable identity, + * resolve the provider's detached creation spec, create the child Agent + * through the private activation-owner scope, establish any continuable-parent + * ownership, and submit the initial prompt. Resolves when inbox acceptance + * yields the message id — without waiting for the turn to start or for the + * message to reach the Session log. + * + * Every failure before that acceptance rejects without either id, disposing + * any created handle and rolling back the Activation and parent ownership. + * The caller signal owns lookup, materialization, and admission only until + * acceptance; afterwards the manager owns the Activation independently. + * @param spec - provider, delegation request, and caller cancellation. + * @returns the durable child id and the accepted initial prompt's message id. + */ + async startContinuable(spec: ContinuableStartSpec): Promise { + this.assertAdmitting() this.requirePersistence() - const childId = SessionId(randomUUID()) const request = spec.request - // Snapshot before Task creation: invalid descriptor JSON rejects the call - // with no Task, and the detached value is what reaches the child log. - const agentProvider = request.agentOptions?.provider ?? request.parent.options.provider - const agentModel = request.agentOptions?.model ?? request.parent.options.model + const parent = request.parent + const childId = SessionId(randomUUID()) + const childDepth = resolveChildDepth(parent, request.maxDepth) + // Snapshot before any await: invalid descriptor JSON rejects the call + // before a child exists, and the detached value is what reaches the log. + const agentProvider = request.agentOptions?.provider ?? parent.options.provider + const agentModel = request.agentOptions?.model ?? parent.options.model const descriptor = snapshotSubagentDescriptor({ provider: spec.provider, ...agentProvider !== undefined ? { agentProvider } : {}, @@ -242,293 +305,498 @@ export class SubagentContinuationManager { ...request.persona !== undefined ? { persona: request.persona } : {}, ...request.toolFilter !== undefined ? { toolFilter: request.toolFilter } : {}, }) - const taskId = this.startActivation(childId, spec.label, request.parent, signal => - this.startProvider(spec.provider, { - ...request, - signal, - continuation: { sessionId: childId, descriptor }, - })) - return { childId, taskId } + + const prepared = await this.host.prepareContinuable(spec.provider, { + sessionId: childId, + parent, + signal: spec.signal, + }) + spec.signal.throwIfAborted() + this.assertAdmitting() + + const lineageSeedLength = prepared.seed?.length ?? 0 + const seed = seedDescriptorTurn(childId, prepared.seed, descriptor) + const messageId = await this.locks.run(childId, async () => { + const activation = await this.materialize({ + childId, + provider: spec.provider, + parent, + seed, + meta: childSessionMeta(parent, childDepth, lineageSeedLength), + agentOptions: resolveChildAgentOptions(parent, request.agentOptions, childDepth), + composition: { persona: request.persona, toolFilter: request.toolFilter }, + signal: spec.signal, + }) + return this.submit(activation, request.prompt, { kind: 'user' }, { kind: 'parent', agent: parent }) + }) + return { childId, messageId } } /** - * Follow up with a known continuable child: steer its running - * activation, or cold-resume the durable session into a fresh Task-backed - * activation. The two routes are reported distinctly so timing-dependent - * routing is observable. Rejection means the message was NOT delivered — in - * particular, losing a race with Task settlement does not fall through to - * cold resume within the same call; a later retry after Task terminal may - * start the next activation. The started Task owns descriptor lookup and - * direct-parent authorization (its AbortSignal exists before that lookup), - * so an unknown, foreign, or descriptor-less child settles the started Task - * as `failed` with a detail reporting the id as unavailable. - * @param parent - the live parent agent sending the message (model tool or - * human adapter); Task access is authorized by its session id. - * @param childId - the stable child session id. + * Deliver one later message to a known continuable child as its next FIFO + * turn. Routing depends only on Activation residency: a `running` Activation + * enqueues, a `waiting` one wakes the same Agent, and an absent one + * cold-resumes a new Activation from the persisted Session. The Agent inbox + * is the only queue, so parent and user messages share one observable order. + * + * The caller signal owns lookup, materialization, and admission only until + * inbox acceptance; afterwards the accepted turn cannot be cancelled through + * this service. + * @param authority - trusted parent or user authority for this delivery. + * @param childId - the durable child session id. * @param content - the user-role content to deliver. - * @param options - caller attribution and cancellation. During live delivery, - * abort cancels the shared activation and rejects only after quiescence. - * @returns whether the content `steered` the existing Task or `started` a new one. + * @param options - durable provenance and caller cancellation. + * @returns the accepted message's inbox id. + * @throws when authority, availability, or admission rejects the delivery. */ async followup( - parent: Agent, + authority: SubagentAuthority, childId: SessionId, content: ContentBlock[], options: SubagentFollowupOptions, - ): Promise { - this.assertOwnership(childId) - const activation = this.activations.get(childId) - if (activation !== undefined) { - return { - route: 'steered', - taskId: await this.steerActivation( - activation, - parent, - childId, - content, - options.source, - options.signal, - ), - } - } - return { - route: 'started', - taskId: this.resumeActivation(parent, childId, content, options.source), + ): Promise { + this.assertAdmitting() + while (true) { + const live = await this.locks.run(childId, async () => { + const activation = this.activations.get(childId) + if (activation === undefined) return this.coldResume(authority, childId, content, options) + // A delivery that arrives after the disposal transaction began must not + // reach a handle being torn down; wait for release, then cold-resume. + if (activation.disposal !== undefined) { + return activation.disposal.then(() => undefined, () => undefined) + } + await this.authorizeLive(authority, activation) + return this.submit(activation, content, options.source, authority) + }) + if (live !== undefined) return live + // The racing disposal completed; retry admission, which now cold-resumes. + this.assertAdmitting() + options.signal.throwIfAborted() } } /** - * Synchronous ownership compare before any by-id routing: a live registry - * Agent outside the association — or different from the associated run's - * agent — was started by something else. Fail instead of adopting an idle - * Agent or attaching an untracked turn. + * 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. + * @throws an aggregate error when any branch failed to release. */ - private assertOwnership(childId: SessionId): void { - const live = this.ctx.agents.get(childId) - if (live === undefined) return - const activation = this.activations.get(childId) - if (activation === undefined) { + async drain(): Promise { + this.enterDraining() + // 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() + for (const activation of this.activations.values()) { + for (const child of activation.ownedChildren) owned.add(child) + } + const roots = [...this.activations.values()].filter(activation => !owned.has(activation.childId)) + const failures = await Promise.all(roots.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} activation(s): ` + + reasons.map(reason => errorChain(reason)).join('; '), + 'ACTIVATION_TEARDOWN_FAILED', + ) + } + } + + /** Reject new admission once the host or manager began draining. */ + private assertAdmitting(): void { + if (this.draining) { + throw new SubagentError( + 'continuable subagents are draining; the operation was not admitted', + 'DRAINING', + ) + } + } + + /** + * Derive residency from Agent quiescence and the owned-child set. `running` + * covers an active admission, an open turn, or waking inbox work. + */ + private stateOf(activation: Activation): ActivationState { + if (activation.handle.agent.status === 'running') return 'running' + if (activation.ownedChildren.size > 0) return 'waiting' + return 'settled' + } + + /** + * Cold-resume a persisted child: load and authorize its Session, fold the + * generic descriptor, create the Activation through `ctx.agents.resume()`, + * and submit the waiting turn. This never dispatches through a subagent + * provider — the persisted Session already holds the initial prefix and the + * descriptor is the whole reconstruction input. + */ + private async coldResume( + authority: SubagentAuthority, + childId: SessionId, + content: ContentBlock[], + options: SubagentFollowupOptions, + ): Promise { + const persistence = this.requirePersistence() + let loaded: Awaited> + try { + loaded = await persistence.load(childId) + } catch (error: unknown) { + throw new SubagentError(`subagent "${childId}" is unavailable`, 'NOT_RESUMABLE', { cause: error }) + } + // The persistence seam takes no signal; recheck before any child work. + options.signal.throwIfAborted() + this.assertAdmitting() + // Authorize the persisted header before folding: only the durable child's + // direct parent — or the host user — may continue it. + this.authorizeLineage(authority, childId, loaded.meta.parentSession) + // Fold only the child's own suffix: a fork seed replays the parent's log, + // which may carry an ANCESTOR's descriptor when the parent is itself a + // continuable child. + const descriptor = foldSubagentDescriptor(loaded.events.slice(loaded.meta.seedLength ?? 0)) + if (descriptor === undefined) { + throw new SubagentError( + `subagent "${childId}" has no supported continuation state and cannot be resumed; ` + + 'do not retry send_message with this id', + 'NOT_RESUMABLE', + ) + } + const activation = await this.materialize({ + childId, + provider: descriptor.provider, + parent: authority.kind === 'parent' ? authority.agent : undefined, + resume: true, + agentOptions: { + ...descriptor.agentProvider !== undefined ? { provider: descriptor.agentProvider } : {}, + ...descriptor.agentModel !== undefined ? { model: descriptor.agentModel } : {}, + }, + composition: { persona: descriptor.persona, toolFilter: descriptor.toolFilter }, + signal: options.signal, + }) + return this.submit(activation, content, options.source, authority) + } + + /** + * Create or resume the child Agent through the private activation-owner + * scope, install the handle in a fresh Activation, and register ownership on + * a continuation-managed parent. Rejection leaves no Activation, no handle, + * and no ownership membership. + */ + private async materialize(inputs: { + childId: SessionId + provider: string + parent: Agent | undefined + resume?: boolean + seed?: readonly SessionEvent[] + meta?: NonNullable + agentOptions: AgentOptions + composition: { persona?: string | undefined; toolFilter?: ToolRestriction | undefined } + signal: AbortSignal + }): Promise { + const { childId, provider, parent } = inputs + if (this.activations.has(childId)) { + throw new SubagentError( + `subagent "${childId}" already has a live activation; the message was not delivered`, + 'ACTIVATION_CONFLICT', + ) + } + if (this.ctx.agents.get(childId) !== undefined) { throw new SubagentError( `subagent "${childId}" has a live agent outside continuation ownership; the message was not delivered`, 'OWNERSHIP_CONFLICT', ) } - if (activation.run !== undefined && activation.run.localAgent !== live) { - throw new SubagentError( - `subagent "${childId}" registry agent is not the associated activation's agent; the message was not delivered`, - 'OWNERSHIP_CONFLICT', - ) - } - } - - /** Deliver to the running activation's Task through confirmed live steering. */ - private async steerActivation( - activation: ActiveActivation, - parent: Agent, - childId: SessionId, - message: ContentBlock[], - source: MessageSource, - signal: AbortSignal, - ): Promise { - const taskId = activation.taskId - /* v8 ignore next 3 -- the install and Task registration share one synchronous frame, so an observed activation carries its Task id. */ - if (taskId === undefined) { - throw new SubagentError(`subagent "${childId}" activation is starting; the message was not delivered`, 'NOT_DELIVERED') - } - // Owner-session authorization plus the live status for admission. - const snapshot = this.ctx.tasks.get(taskId, parent) - if (snapshot.status !== 'running') { - throw new SubagentError( - `subagent "${childId}" task ${taskId} is ${snapshot.status}; the message was not delivered ` - + '— retry after it settles to start the next activation', - 'NOT_DELIVERED', - ) - } - const run = activation.run - if (run === undefined) { - throw new SubagentError(`subagent "${childId}" activation is starting; the message was not delivered`, 'NOT_DELIVERED') - } - if (run.steer === undefined) { - throw new SubagentError( - `subagent "${childId}" provider does not accept live delivery; the message was not delivered`, - 'NOT_DELIVERED', - ) - } - const cancelActivation = (): void => { - activation.controller.abort(signal.reason) - } - signal.addEventListener('abort', cancelActivation, { once: true }) - if (signal.aborted) { - cancelActivation() - signal.removeEventListener('abort', cancelActivation) - return await this.cancelledLiveDelivery(activation, childId) - } + inputs.signal.throwIfAborted() + const setup = (childCtx: Context): void => { applyChildComposition(childCtx, inputs.composition) } + const observer = this.host.observeActivation(provider, childId, parent) + let handle: AgentHandle try { - await run.steer(message, source) + handle = inputs.resume === true + ? await this.ownerCtx.agents.resume({ + resumeSessionId: childId, + agentOptions: inputs.agentOptions, + signal: inputs.signal, + setup, + }) + : await this.ownerCtx.agents.create({ + sessionId: childId, + ...inputs.meta !== undefined ? { meta: inputs.meta } : {}, + ...inputs.seed !== undefined ? { seed: inputs.seed } : {}, + agentOptions: inputs.agentOptions, + signal: inputs.signal, + setup, + }) } catch (error: unknown) { - try { - signal.throwIfAborted() - } catch { - return await this.cancelledLiveDelivery(activation, childId, error) - } - // Confirmed steering lost the race with request admission. Deliberately no - // cold-resume fallback here: that would attach the message to a turn the - // caller did not observe. - throw new SubagentError( - `subagent "${childId}" stopped before delivery; the message was not delivered`, - 'NOT_DELIVERED', - { cause: error }, - ) - } finally { - signal.removeEventListener('abort', cancelActivation) + // Agent creation provides rollback before handle transfer, so nothing + // outlives this rejection; report the epoch that never became resident. + observer.settle(undefined, error) + throw error } - return taskId + + const activation: Activation = { + childId, + provider, + handle, + ownedChildren: new Set(), + observer, + disposal: undefined, + poke: Promise.withResolvers(), + } + // After transfer, any failure must dispose the created handle, remove the + // Activation, and roll back parent ownership before rejecting. + this.activations.set(childId, activation) + try { + inputs.signal.throwIfAborted() + this.assertAdmitting() + this.acquireOwnership(parent, childId) + } catch (error: unknown) { + // Roll the transfer back completely: the Activation leaves the map, the + // parent's ownership membership is released, and the created handle is + // disposed before this rejection surfaces. + this.activations.delete(childId) + this.releaseOwnership(childId) + activation.disposal = (async () => { + try { + await handle.dispose() + } finally { + observer.settle(handle.agent, error) + } + })() + await activation.disposal.catch(() => undefined) + throw error + } + // Resident: publish the start edge before any turn can run, so observers + // see this epoch before its first request. + observer.start() + this.watchSettlement(activation) + return activation } - /** Reject a cancelled live delivery only after its shared activation is quiescent. */ - private async cancelledLiveDelivery( - activation: ActiveActivation, - childId: SessionId, - cause?: unknown, - ): Promise { - /* v8 ignore if -- a published run implies the producer assigned `done` before its provider await resolved. */ - if (activation.done === undefined) { - throw new Error('published subagent activation has no settlement promise') + /** + * Register the child in a continuation-managed parent's owned set before the + * child can run, so that parent cannot settle while the child is live. A + * top-level or other non-continuation Agent has no Activation and stays + * outside the waiting graph. + */ + private acquireOwnership(parent: Agent | undefined, childId: SessionId): void { + if (parent === undefined) return + const parentActivation = this.activations.get(parent.id) + if (parentActivation === undefined) return + if (parentActivation.disposal !== undefined) { + throw new SubagentError( + `subagent parent "${parent.id}" is being disposed; the child was not established`, + 'ACTIVATION_CLOSING', + ) } - await activation.done - throw new SubagentError( - `subagent "${childId}" live delivery was cancelled; the message was not delivered`, - 'CANCELLED', - cause === undefined ? undefined : { cause }, + parentActivation.ownedChildren.add(childId) + } + + /** Remove one child from its live owner's set and let that owner re-check settlement. */ + private releaseOwnership(childId: SessionId): void { + for (const candidate of this.activations.values()) { + if (candidate.ownedChildren.delete(childId)) this.wake(candidate) + } + } + + /** Let a settlement watcher re-observe quiescence after ownership or inbox changes. */ + private wake(activation: Activation): void { + activation.poke.resolve() + activation.poke = Promise.withResolvers() + } + + /** + * Submit one message as the child's next FIFO turn and return its accepted + * inbox id. Acceptance is the operation's success boundary; the manager owns + * the Activation independently afterwards. + */ + private submit( + activation: Activation, + content: ContentBlock[], + source: MessageSource, + authority: SubagentAuthority, + ): MessageId { + // Parent-originated delivery keeps the parent live through ownership, so + // establish it before the message can enter the child's inbox. + if (authority.kind === 'parent') this.acquireOwnership(authority.agent, activation.childId) + const message = createUserMessage({ content, source }) + activation.handle.agent.followup(message) + // Accepted waking work keeps this Activation live until whenIdle() observes + // the complete waking suffix. + this.wake(activation) + return message.id + } + + /** + * Authorize delivery to a live Activation. A parent must be the exact live + * direct parent recorded in the child's durable header. + */ + private async authorizeLive(authority: SubagentAuthority, activation: Activation): Promise { + await Promise.resolve() + this.authorizeLineage( + authority, + activation.childId, + activation.handle.agent.session.header.parentSession, ) } /** - * Cold-resume a persisted child into a fresh Task-backed activation. The - * Task owns its `AbortController` before descriptor lookup: the load, - * direct-parent authorization, and descriptor fold run inside the - * activation, with cancellation rechecked after the un-signalled - * persistence await so an early `task_kill` prevents any later child work. + * Authorize one operation against the durable direct-parent lineage. User + * authority may continue any child without loading its parent; parent + * authority requires the exact live direct parent. Other agents, ancestors, + * teams, and workflows remain rejected until an explicit authority protocol + * exists. */ - private resumeActivation( - parent: Agent, + private authorizeLineage( + authority: SubagentAuthority, childId: SessionId, - message: ContentBlock[], - source: MessageSource, - ): TaskId { - const persistence = this.requirePersistence() - return this.startActivation(childId, resumeLabel(message), parent, async (signal) => { - let loaded: Awaited> - try { - loaded = await persistence.load(childId) - } catch (error: unknown) { - throw new SubagentError( - `subagent "${childId}" is unavailable`, - 'NOT_RESUMABLE', - { cause: error }, - ) - } - // The persistence seam takes no signal; recheck before any child work. - if (signal.aborted) throw new SubagentError('subagent resume was cancelled during lookup', 'CANCELLED') - // Authorize the persisted header before folding: only the direct parent - // recorded at creation may continue this child. - if (loaded.meta.parentSession !== parent.id) { - throw new SubagentError( - `subagent "${childId}" belongs to another parent session`, - 'UNAUTHORIZED', - ) - } - // Fold only the child's own suffix: a fork seed replays the parent's - // log, which may carry an ANCESTOR's descriptor when the parent is - // itself a continuable child. - const descriptor = foldSubagentDescriptor(loaded.events.slice(loaded.meta.seedLength ?? 0)) - if (descriptor === undefined) { - throw new SubagentError( - `subagent "${childId}" has no supported continuation state and cannot be resumed; ` - + 'do not retry send_message with this id', - 'NOT_RESUMABLE', - ) - } - return this.resumeProvider({ - sessionId: childId, - prompt: message, - source, - parent, - signal, - descriptor, - }) - }) + parentSession: SessionId | undefined, + ): void { + if (authority.kind === 'user') return + const parent = authority.agent + if (this.ctx.agents.get(parent.id) !== parent) { + throw new SubagentError( + `subagent "${childId}" delivery requires the exact live parent agent`, + 'UNAUTHORIZED', + ) + } + if (parentSession !== parent.id) { + throw new SubagentError(`subagent "${childId}" belongs to another parent session`, 'UNAUTHORIZED') + } } /** - * Install the activation association, register its Task, and bind the two - * removal conditions. The association is installed before any persistence - * or provider await — the producer body runs synchronously up to its first - * await — and removed only after run disposal (the producer settled) and - * Task terminal publication. This synchronous install admits one activation - * per child in this process; a competing untracked publication still loses - * at the Agent registry collision boundary inside the provider. + * Follow one Activation to settlement: wait for Agent quiescence, then for + * every owned child to complete disposal, and dispose the handle once both + * hold. A `next-turn` delivered while `waiting` wakes the same Agent and + * returns it to `running`, so this re-observes rather than settling early. */ - private startActivation( - childId: SessionId, - label: string, - owner: Agent, - begin: (signal: AbortSignal) => Promise, - ): TaskId { - const activation: ActiveActivation = { - taskId: undefined, - run: undefined, - controller: new AbortController(), - done: undefined, - terminal: Promise.withResolvers(), - } - this.activations.set(childId, activation) - let taskId: TaskId - try { - taskId = this.ctx.tasks.start({ - kind: 'subagent', - label, - owner, - run: (): TaskHooks => { - const done = (async (): Promise => { - try { - const run = await begin(activation.controller.signal) - activation.run = run - return await settleRun(run) - } catch (error: unknown) { - // A pre-publication abort rejects only after the provider's - // creation transaction rolled back to quiescence, so recording - // `killed` here honors the settlement-after-rollback contract. - return activation.controller.signal.aborted - ? { status: 'killed' } - : { status: 'failed', detail: String(error) } - } - })() - activation.done = done - void Promise.allSettled([done, activation.terminal.promise]).then(() => { - /* v8 ignore else -- service teardown clears the map while a producer is still settling. */ - if (this.activations.get(childId) === activation) this.activations.delete(childId) - }) - return { - cancel: (reason?: string) => { - // Cancellation targets the whole activation: every message that - // joined this turn shares the `killed` outcome. - activation.controller.abort(reason ?? 'subagent activation killed') - }, - done, - // No readOutput: the child session owns intermediate detail. + private watchSettlement(activation: Activation): void { + void (async () => { + while (disposalOf(activation) === undefined) { + const poked = activation.poke.promise + await Promise.race([activation.handle.agent.whenIdle(), poked]) + if (disposalOf(activation) !== undefined) return + // Re-check settlement INSIDE the child lock and begin disposal in the + // same critical section, so a concurrent delivery either wins admission + // before the transaction opens or waits for release and cold-resumes. + // Deciding outside the lock would let a delivery observe a not-yet + // resident handle that this watcher is already about to tear down. + const settling = await this.locks.run(activation.childId, () => { + if (disposalOf(activation) !== undefined || this.stateOf(activation) !== 'settled') { + return Promise.resolve({ settling: false }) } - }, - }) + // `dispose()` assigns its memoized transaction synchronously, so + // admission is closed before this critical section releases. + return Promise.resolve({ settling: true, done: this.dispose(activation) }) + }) + if (!settling.settling) { + // Still running, or waiting on descendants: re-observe after the next + // accepted message or ownership release. + if (activation.handle.agent.status !== 'running') await poked + continue + } + try { + await settling.done + } catch (error: unknown) { + this.ctx.logger.warn( + `subagent "${activation.childId}" activation teardown failed: ${errorChain(error)}`, + ) + } + return + } + })() + } + + /** + * 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. + * + * 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`. + */ + private dispose(activation: Activation): Promise { + 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 durability = await this.checkpoint(activation) + failure ??= durability + } finally { + this.activations.delete(childId) + 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 { + // Release ownership even on failure: a retained failed child would + // pin its ancestors in `waiting` forever. + this.releaseOwnership(childId) + activation.observer.settle(activation.handle.agent, failure) + } + } + if (failure !== undefined) throw failure + })()) + } + + /** + * Request the final durability checkpoint. Only `true` confirms durability; + * `false` and rejection both report `DURABILITY_FAILED` so the persisted + * child state is known to be possibly missing or stale on a later resume. + */ + private async checkpoint(activation: Activation): Promise { + const child = activation.handle.agent + try { + const participated = await child.ctx.sessions.flush(child.session) + if (participated) return undefined + return new SubagentError( + `subagent "${activation.childId}" required durability checkpoint has no registered listener; ` + + 'the latest child state was not confirmed persisted and may be unavailable or stale on resume', + 'DURABILITY_FAILED', + ) } catch (error: unknown) { - // Task preflight failed; nothing started, so the install rolls back. - this.activations.delete(childId) - throw error + return new SubagentError( + `subagent "${activation.childId}" durability checkpoint failed; the latest child state was not ` + + `confirmed persisted and may be unavailable or stale on resume: ${errorChain(error)}`, + 'DURABILITY_FAILED', + { cause: error }, + ) } - // Same synchronous frame as the install: an observer that can run at all - // runs after this assignment. - activation.taskId = taskId - return taskId } /** Resolve the persistence service continuable children require, or fail loud. */ @@ -544,11 +812,5 @@ export class SubagentContinuationManager { } } -/** Derive a resumed activation's Task label from its message. */ -function resumeLabel(message: ContentBlock[]): string { - const text = finalText(message).trim().replace(/\s+/g, ' ') - if (text.length === 0) return 'subagent follow-up' - return text.length > 80 ? `${text.slice(0, 79)}…` : text -} - +export type { SubagentDescriptorData } export default SubagentContinuationManager diff --git a/packages/subagent/subagent/src/depth.ts b/packages/subagent/subagent/src/depth.ts new file mode 100644 index 0000000000..d9fabab860 --- /dev/null +++ b/packages/subagent/subagent/src/depth.ts @@ -0,0 +1,51 @@ +/** + * Delegation-depth accounting: the recursion budget a parent passes to its + * children. Kept apart from the service so composition helpers can read it + * without importing the registry. + * + * @module @deepseek-ai/dsh-subagent/depth + */ + +import type { Agent } from '@deepseek-ai/dsh-agent' + +declare module '@deepseek-ai/dsh-agent' { + interface AgentOptions { + /** Delegation depth: zero for a top-level agent and parent depth + 1 for a child. */ + subagentDepth?: number + } +} + +/** + * Read an agent's delegation depth, treating absence as top-level depth zero. + * The persisted session header is authoritative and monotone: runtime + * `AgentOptions.subagentDepth` may DEEPEN the count but can never lower it — + * a resumed child arrives with fresh options, and counting it from zero would + * let it delegate as if it were top-level. + * @param agent - the agent whose header and options carry the depth. + * @returns its non-negative safe-integer depth. + * @throws if the runtime `AgentOptions.subagentDepth` is not a non-negative safe integer. + */ +export function delegationDepthOf(agent: Agent): number { + const runtime = agent.options.subagentDepth + if (runtime !== undefined && (!Number.isSafeInteger(runtime) || runtime < 0 || Object.is(runtime, -0))) { + throw new TypeError('agent subagentDepth must be a non-negative safe integer') + } + // The header value was validated at the session boundary (creation and + // persistence load both construct through the store). + return Math.max(agent.session.header.delegationDepth ?? 0, runtime ?? 0) +} + +/** + * Reject a recursion cap that cannot represent an exact delegation depth. + * @param maxDepth - the optional runtime value to validate. + */ +export function assertSubagentMaxDepth(maxDepth: unknown): void { + if (maxDepth !== undefined && ( + typeof maxDepth !== 'number' + || !Number.isSafeInteger(maxDepth) + || maxDepth < 0 + || Object.is(maxDepth, -0) + )) { + throw new TypeError('subagent maxDepth must be a non-negative safe integer') + } +} diff --git a/packages/subagent/subagent/src/descriptor-seed.ts b/packages/subagent/subagent/src/descriptor-seed.ts new file mode 100644 index 0000000000..836b40009d --- /dev/null +++ b/packages/subagent/subagent/src/descriptor-seed.ts @@ -0,0 +1,31 @@ +/** + * Seeding of a continuable child's durable descriptor event: the model-hidden + * record of the child's declared composition before its first request, so a + * later cold resume can reconstruct it from its own log. + * + * @module @deepseek-ai/dsh-subagent/descriptor-seed + */ + +import { Session } from '@deepseek-ai/dsh-session' +import type { SessionEvent, SessionId } from '@deepseek-ai/dsh-session' +import type { SubagentDescriptorData } from './descriptor.ts' + +/** + * Build the child's creation seed: any inherited parent-history prefix followed + * by one model-hidden, between-turn `descriptor` event. Staging through a + * `Session` assigns the sequence number and enforces the same lossless-JSON + * rules the durable log does. + * @param childId - the reserved child session id the staged log belongs to. + * @param seed - the inherited completed-turn prefix, or `undefined` for a fresh child. + * @param descriptor - the snapshotted composition record to persist. + * @returns the complete seed events, contiguous from sequence zero. + */ +export function seedDescriptorTurn( + childId: SessionId, + seed: readonly SessionEvent[] | undefined, + descriptor: SubagentDescriptorData, +): SessionEvent[] { + const staged = new Session(childId, seed) + staged.append('subagent/descriptor', descriptor) + return [...staged.events] +} diff --git a/packages/subagent/subagent/src/index.ts b/packages/subagent/subagent/src/index.ts index 023ac1fe26..0bfebc8cf5 100644 --- a/packages/subagent/subagent/src/index.ts +++ b/packages/subagent/subagent/src/index.ts @@ -13,11 +13,13 @@ * (`@deepseek-ai/dsh-subagent-spawn`, `-fork`, `-acp`) and the model-facing * consumer (`@deepseek-ai/dsh-tool-subagent`) are separate packages. * - * Public operations express caller intent: `start` returns one ready owned run, - * `startContinuable` starts a Task-backed durable child, and `followup` routes - * later content without exposing whether the child is live. Provider resume - * dispatch stays private because only the continuation manager holds the - * resolved descriptor and authorization facts. + * Public operations express caller intent: `start` returns one ready owned + * one-shot run, `startContinuable` establishes a durable continuable child, and + * `followup` delivers later content without exposing whether the child is + * resident. Continuable children never become a {@link SubagentRun}: the + * continuation manager holds their `AgentHandle` directly and orders every turn + * through the child's own inbox, so providers contribute only the detached + * creation spec and see no handle, turn, or teardown. * * Same-process providers are trusted typed collaborators. Requests, provider * descriptors, results, and lifecycle payloads are borrowed immutable values; @@ -32,36 +34,38 @@ import { Context, Service } from 'cordis' import { scopeTarget } from '@deepseek-ai/dsh-scope' import type { Scoped } from '@deepseek-ai/dsh-scope' import { assertObjectJsonSchema } from '@deepseek-ai/dsh-tools' -import type { ContentBlock } from '@deepseek-ai/dsh-llm' +import type { ContentBlock, MessageId } from '@deepseek-ai/dsh-llm' import type { Agent } from '@deepseek-ai/dsh-agent' -import type { SessionId } from '@deepseek-ai/dsh-session' +import type { SessionEvent, SessionId } from '@deepseek-ai/dsh-session' import type { + ContinuableCreateRequest, + ContinuableCreateSpec, SubagentCapabilities, SubagentProvider, - SubagentProviderResumeRequest, - SubagentProviderStartRequest, SubagentResult, SubagentRun, SubagentStartRequest, } from './types.ts' import { SubagentRunId } from './types.ts' import { SubagentError } from './error.ts' +import { assertSubagentMaxDepth } from './depth.ts' import SubagentContinuationManager from './continuation.ts' import type { + ActivationObserver, + ActivationState, ContinuableStart, ContinuableStartSpec, + SubagentAuthority, SubagentFollowupOptions, - SubagentFollowupResult, } from './continuation.ts' export * from './out-of-process.ts' export { SubagentRunId } from './types.ts' export type { + ContinuableCreateRequest, + ContinuableCreateSpec, SubagentCapabilities, - SubagentContinuation, SubagentProvider, - SubagentProviderResumeRequest, - SubagentProviderStartRequest, SubagentResult, SubagentRun, SubagentStartRequest, @@ -74,58 +78,28 @@ export { SUBAGENT_DESCRIPTOR_VERSION, } from './descriptor.ts' export type { SubagentDescriptorData, SubagentDescriptorInput } from './descriptor.ts' +export { seedDescriptorTurn } from './descriptor-seed.ts' export { SubagentError } from './error.ts' -export { settleRun } from './continuation.ts' +export { settleRun } from './run-settlement.ts' +export { assertSubagentMaxDepth, delegationDepthOf } from './depth.ts' +export { + applyChildComposition, + childSessionMeta, + resolveChildAgentOptions, + resolveChildDepth, + SubagentDepthError, +} from './child-agent.ts' +export type { ChildComposition } from './child-agent.ts' export type { + ActivationObserver, + ActivationState, ContinuableStart, ContinuableStartSpec, CoordinatorMessageSource, + SubagentAuthority, SubagentFollowupOptions, - SubagentFollowupResult, } from './continuation.ts' -declare module '@deepseek-ai/dsh-agent' { - interface AgentOptions { - /** Delegation depth: zero for a top-level agent and parent depth + 1 for a child. */ - subagentDepth?: number - } -} - -/** - * Read an agent's delegation depth, treating absence as top-level depth zero. - * The persisted session header is authoritative and monotone: runtime - * `AgentOptions.subagentDepth` may DEEPEN the count but can never lower it — - * a resumed child arrives with fresh options, and counting it from zero would - * let it delegate as if it were top-level. - * @param agent - the agent whose header and options carry the depth. - * @returns its non-negative safe-integer depth. - * @throws if the runtime `AgentOptions.subagentDepth` is not a non-negative safe integer. - */ -export function delegationDepthOf(agent: Agent): number { - const runtime = agent.options.subagentDepth - if (runtime !== undefined && (!Number.isSafeInteger(runtime) || runtime < 0 || Object.is(runtime, -0))) { - throw new TypeError('agent subagentDepth must be a non-negative safe integer') - } - // The header value was validated at the session boundary (creation and - // persistence load both construct through the store). - return Math.max(agent.session.header.delegationDepth ?? 0, runtime ?? 0) -} - -/** - * Reject a recursion cap that cannot represent an exact delegation depth. - * @param maxDepth - the optional runtime value to validate. - */ -export function assertSubagentMaxDepth(maxDepth: unknown): void { - if (maxDepth !== undefined && ( - typeof maxDepth !== 'number' - || !Number.isSafeInteger(maxDepth) - || maxDepth < 0 - || Object.is(maxDepth, -0) - )) { - throw new TypeError('subagent maxDepth must be a non-negative safe integer') - } -} - declare module 'cordis' { interface Context { subagents: SubagentService @@ -195,19 +169,18 @@ export interface SubagentRunEndInfo { readonly lastAssistantMessage?: ContentBlock[] } -/** Named provider registry with raw and Task-backed continuation operations. */ +/** Named provider registry with one-shot runs and continuable-child operations. */ export class SubagentService extends Service { private providers = new Map() private continuations: SubagentContinuationManager | undefined constructor(ctx: Context) { super(ctx, 'subagents') - ctx.inject(['tasks', 'agents'], (childCtx: Context) => { - const manager = new SubagentContinuationManager( - childCtx, - (name, request) => this.startProvider(name, request), - request => this.resumeProvider(request), - ) + ctx.inject(['agents'], (childCtx: Context) => { + const manager = new SubagentContinuationManager(childCtx, { + prepareContinuable: (name, request) => this.prepareContinuable(name, request), + observeActivation: (provider, childId, parent) => this.observeActivation(provider, childId, parent), + }) this.continuations = manager childCtx.effect(() => () => { /* v8 ignore else -- one injected binding owns the slot until its fiber disposes. */ @@ -217,34 +190,64 @@ export class SubagentService extends Service { } /** - * Start one durable continuable child through a Task-backed initial - * activation. - * @param spec - provider, Task label, and delegation request. - * @returns the stable child id and initial activation Task id. + * Establish one durable continuable child and deliver its initial prompt. + * Resolves 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 back the child entirely. + * @param spec - provider, delegation request, and caller cancellation. + * @returns the durable child id and the accepted prompt's message id. + * @throws when continuation services are unavailable or materialization fails. */ - startContinuable(spec: ContinuableStartSpec): ContinuableStart { + startContinuable(spec: ContinuableStartSpec): Promise { return this.requireContinuations().startContinuable(spec) } /** - * Follow up with a continuable child. A live child is steered and fulfillment - * confirms request admission; an idle child immediately returns a fresh Task - * whose descriptor lookup, authorization, and cold resume may later fail. - * @param parent - live direct parent authorizing the operation. + * Deliver one later message to a continuable child as its next FIFO turn. A + * resident child's Agent inbox accepts it directly (waking a `waiting` + * Activation), while an absent one is cold-resumed from its persisted + * Session. The Agent inbox is the only queue, so parent and user messages + * share one observable order. + * @param authority - trusted parent or user authority for this delivery. * @param childId - durable child session id. * @param content - user-role content to deliver. - * @param options - durable attribution and caller cancellation; aborting a - * live-delivery wait cancels the shared activation and awaits quiescence. - * @returns the existing steered Task or newly started Task. - * @throws when continuation services are unavailable or live delivery is not admitted. + * @param options - durable provenance and caller cancellation, which stops the + * operation only before inbox acceptance. + * @returns the accepted message's inbox id. + * @throws when continuation services are unavailable, authority is rejected, + * or the message was not admitted. */ followup( - parent: Agent, + authority: SubagentAuthority, childId: SessionId, content: ContentBlock[], options: SubagentFollowupOptions, - ): Promise { - return this.requireContinuations().followup(parent, childId, content, options) + ): Promise { + return this.requireContinuations().followup(authority, childId, content, options) + } + + /** + * Read one durable child's live residency state. + * @param childId - durable child session id. + * @returns its Activation state, or `undefined` when no Activation is live. + * @throws when continuation services are unavailable. + */ + activationState(childId: SessionId): ActivationState | undefined { + return this.requireContinuations().activationState(childId) + } + + /** + * 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. + * @returns once every live Activation released its `AgentHandle`. + * @throws an aggregate error after all branches settle when any failed. + */ + async drainContinuable(): Promise { + const manager = this.continuations + // Absent continuation services means nothing was ever materialized. + if (manager === undefined) return + await manager.drain() } /** @@ -298,40 +301,32 @@ export class SubagentService extends Service { * @param request - child prompt, parent, signal, and optional capabilities. * @returns the ready holder-owned run. */ - async start(name: string, request: SubagentStartRequest & { readonly continuation?: never }): Promise { - return this.startProvider(name, request) - } - - /** Validate and dispatch one ordinary or service-resolved provider start. */ - private async startProvider( - name: string, - request: SubagentProviderStartRequest, - ): Promise { + async start(name: string, request: SubagentStartRequest): Promise { const provider = this.expectProvider(name) this.assertCapabilities(provider, request) assertSubagentMaxDepth(request.maxDepth) if (request.outputSchema !== undefined) assertObjectJsonSchema(request.outputSchema) - if (request.continuation !== undefined && provider.resume === undefined) { - throw new SubagentError( - `subagent provider "${provider.name}" does not support continuable children (no resume capability)`, - 'UNSUPPORTED_CAPABILITY', - ) - } - return this.observeRun(name, request.parent, await provider.start(request)) } - /** Dispatch one authorized provider resume and observe its run lifecycle. */ - private async resumeProvider(request: SubagentProviderResumeRequest): Promise { - const name = request.descriptor.provider + /** + * Resolve one provider's detached continuable-creation contribution. Method + * presence on the provider IS the capability, so a provider without it is + * rejected before the manager reserves any child resources. + */ + private async prepareContinuable( + name: string, + request: ContinuableCreateRequest, + ): Promise { const provider = this.expectProvider(name) - if (provider.resume === undefined) { + if (provider.prepareContinuable === undefined) { throw new SubagentError( - `subagent provider "${provider.name}" does not support resuming persisted children (no resume capability)`, + `subagent provider "${provider.name}" does not support continuable children ` + + '(no prepareContinuable capability)', 'UNSUPPORTED_CAPABILITY', ) } - return this.observeRun(name, request.parent, await provider.resume(request)) + return provider.prepareContinuable(request) } /** Look up a provider for dispatch or fail loud. */ @@ -354,6 +349,41 @@ export class SubagentService extends Service { return this.continuations } + /** + * Emit the start/end lifecycle pair for one continuable Activation's + * residency epoch. Observers see the same vocabulary as a one-shot run, so a + * child's start and settlement remain observable without exposing whether the + * manager materialized, woke, or cold-resumed it. Creation failure before + * residency reports only the terminal edge. + */ + private observeActivation( + provider: string, + childId: SessionId, + parent: Agent | undefined, + ): ActivationObserver { + const identity = { runId: SubagentRunId(randomUUID()), provider, id: childId, local: true } + let started = false + let settled = false + return { + start: (): void => { + started = true + this.emitLifecycle('subagent/start', identity, parent) + }, + settle: (child: Agent | undefined, failure: unknown): void => { + // A failure before residency has no start edge to pair, and inventing + // one would report a lifecycle the child never had. + if (settled || !started) return + settled = true + const output = failure === undefined ? lastAssistantOutput(child) : undefined + this.emitLifecycle('subagent/end', { + ...identity, + stopReason: failure === undefined ? 'completed' : 'error', + ...output === undefined ? {} : { lastAssistantMessage: output }, + }, parent) + }, + } + } + /** Emit the start/end lifecycle pair for one accepted run and return it. */ private observeRun(name: string, parent: Agent, run: SubagentRun): SubagentRun { const runId = SubagentRunId(randomUUID()) @@ -385,14 +415,16 @@ export class SubagentService extends Service { * Emit lifecycle events with per-listener synchronous and asynchronous * exception containment. Payloads are borrowed immutable values. */ - private emitLifecycle(name: 'subagent/start', info: SubagentRunInfo, parent: Agent): void - private emitLifecycle(name: 'subagent/end', info: SubagentRunEndInfo, parent: Agent): void + private emitLifecycle(name: 'subagent/start', info: SubagentRunInfo, parent: Agent | undefined): void + private emitLifecycle(name: 'subagent/end', info: SubagentRunEndInfo, parent: Agent | undefined): void private emitLifecycle(name: 'subagent/provider-removed', info: string): void private emitLifecycle( name: 'subagent/start' | 'subagent/end' | 'subagent/provider-removed', info: SubagentRunInfo | SubagentRunEndInfo | string, - parent?: Agent, + parent?: Agent , ): void { + // A user-resumed continuable child has no delegating parent to key the + // carrier by, so its lifecycle reaches unscoped listeners globally. const dispatchArgs: unknown[] = parent === undefined ? [name, info] : [scopeTarget(this, parent), name, info] @@ -427,6 +459,18 @@ export class SubagentService extends Service { } } +/** + * The child's last assistant message content, for one Activation's terminal + * lifecycle edge. Absent when no assistant message reached the log. + */ +function lastAssistantOutput(child: Agent | undefined): ContentBlock[] | undefined { + if (child === undefined) return undefined + const message = child.session.events.findLast( + (event): event is SessionEvent<'assistant/message'> => event.type === 'assistant/message', + ) + return message?.data.message.content +} + /** Render any listener-thrown value without letting coercion escape containment. */ function renderThrown(value: unknown): string { try { diff --git a/packages/subagent/subagent/src/run-settlement.ts b/packages/subagent/subagent/src/run-settlement.ts new file mode 100644 index 0000000000..92d0986bcd --- /dev/null +++ b/packages/subagent/subagent/src/run-settlement.ts @@ -0,0 +1,71 @@ +/** + * Settlement of one ONE-SHOT subagent run into a background-Task outcome. Only + * the one-shot background path uses Tasks; continuable children have no Task, + * no per-message result, and no Task cancellation. + * + * @module @deepseek-ai/dsh-subagent/run-settlement + */ + +import { HarnessError } from '@deepseek-ai/dsh-llm' +import type { ContentBlock } from '@deepseek-ai/dsh-llm' +import type { TaskOutcome } from '@deepseek-ai/dsh-tasks' +import type { SubagentResult, SubagentRun } from './types.ts' + +/** Flatten a child's final output blocks to the task's final text. */ +function finalText(blocks: ContentBlock[]): string { + return blocks + .filter((block): block is Extract => block.type === 'text') + .map(block => block.text) + .join('') +} + +/** + * Map a child result to the task outcome: completed carries final text, + * aborted is killed, and every other reason is failed without partial output. + * @param result - child terminal result. + * @returns outcome for the `ctx.tasks` registration. + */ +function runOutcome(result: SubagentResult): TaskOutcome { + switch (result.stopReason) { + case 'completed': + return { status: 'completed', output: finalText(result.output) } + case 'aborted': + return { status: 'killed' } + case 'error': + case 'max-tokens': + case 'refusal': + return { status: 'failed', detail: result.stopReason } + // Merge-extensible reasons remain failures with their raw detail. + default: + return { status: 'failed', detail: String(result.stopReason) } + } +} + +/** Render infrastructure failure detail without hiding a durability diagnosis. */ +function runFailureDetail(error: unknown): string { + return error instanceof HarnessError && error.code === 'DURABILITY_FAILED' + ? error.message + : String(error) +} + +/** + * Await the child result, dispose the run, then return its task outcome. Result + * and disposal failures become `failed`; when both fail, both details survive. + * @param run - live run to settle and release. + * @returns outcome after child resources are released. + */ +export async function settleRun(run: SubagentRun): Promise { + let outcome: TaskOutcome + try { + outcome = runOutcome(await run.result) + } catch (error: unknown) { + outcome = { status: 'failed', detail: runFailureDetail(error) } + } + try { + await run.dispose() + } catch (error: unknown) { + const prefix = outcome.detail === undefined ? '' : `${outcome.detail}; ` + return { status: 'failed', detail: `${prefix}dispose failed: ${String(error)}` } + } + return outcome +} diff --git a/packages/subagent/subagent/src/types.ts b/packages/subagent/subagent/src/types.ts index da75ae67cc..3ff7368b37 100644 --- a/packages/subagent/subagent/src/types.ts +++ b/packages/subagent/subagent/src/types.ts @@ -6,10 +6,9 @@ import type { Agent, AgentOptions } from '@deepseek-ai/dsh-agent' import type { Branded } from '@deepseek-ai/dsh-brand' -import type { ContentBlock, MessageSource } from '@deepseek-ai/dsh-llm' -import type { SessionId } from '@deepseek-ai/dsh-session' +import type { ContentBlock } from '@deepseek-ai/dsh-llm' +import type { SessionEvent, SessionId } from '@deepseek-ai/dsh-session' import type { ObjectJsonSchema, ToolRestriction } from '@deepseek-ai/dsh-tools' -import type { SubagentDescriptorData } from './descriptor.ts' /** Identifies one accepted subagent run across its lifecycle event pair. */ export type SubagentRunId = Branded<'SubagentRunId'> @@ -27,11 +26,12 @@ export function SubagentRunId(id: string): SubagentRunId { * Which START-TIME features a provider supports. Checked by the service before delegating to * {@link SubagentProvider.start}: a request that needs a capability the chosen provider lacks * is rejected with a typed error rather than accepted-then-ignored (the "fail loud, no silent - * degradation" rule). These static flags cover features needed before a run exists; runtime - * capabilities are optional methods whose presence is the capability — confirmed live steering - * is {@link SubagentRun.steer} and persisted cold resume is {@link SubagentProvider.resume}. Each - * flag corresponds one-to-one to a {@link SubagentStartRequest} option: `depthLimit` to - * `maxDepth`; the other names match. + * degradation" rule). These flags describe the ONE-SHOT + * {@link SubagentProvider.start} path, where the provider composes the child; + * continuable children are composed by the continuation manager itself and are + * gated by {@link SubagentProvider.prepareContinuable} instead. Each flag + * corresponds one-to-one to a {@link SubagentStartRequest} option: `depthLimit` + * to `maxDepth`; the other names match. */ export interface SubagentCapabilities { readonly outputSchema: boolean @@ -41,10 +41,10 @@ export interface SubagentCapabilities { } /** - * What a caller asks for when starting a subagent. The tool layer builds this - * from the model's `{ description, prompt }` plus its own config; the service - * validates {@link SubagentCapabilities} against the named provider and - * resolves a {@link SubagentProviderStartRequest} for dispatch. + * What a caller asks for when starting a ONE-SHOT subagent. The tool layer + * builds this from the model's `{ description, prompt }` plus its own config; + * the service validates {@link SubagentCapabilities} against the named provider + * before dispatching to {@link SubagentProvider.start}. */ export interface SubagentStartRequest { /** Content delivered as the child's user message. */ @@ -96,63 +96,37 @@ export interface SubagentStartRequest { } /** - * Provider-facing start request after the service resolves optional - * continuation state. Ordinary callers use {@link SubagentStartRequest}; only - * the Task-backed continuation path can attach a stable child identity and - * durable descriptor. + * What the continuation manager asks a provider for while materializing one + * continuable child's FIRST activation. The manager has already reserved the + * durable child identity and owns every later operation, so this request + * carries only what distinguishes a fresh child from one seeded with parent + * history. */ -export interface SubagentProviderStartRequest extends SubagentStartRequest { - /** - * Continuable-child state resolved by `ctx.subagents` before provider dispatch. - * The provider MUST publish exactly `sessionId` as the child identity - * instead of allocating one internally, and MUST append the snapshotted, - * model-hidden `subagent/descriptor` before the initial prompt is admitted. - * Requires {@link SubagentProvider.resume} (the - * continuation capability); the service rejects the request otherwise. - */ - readonly continuation?: SubagentContinuation | undefined -} - -/** - * The resolved continuable-child identity and durable composition record the - * service attaches before provider dispatch. - */ -export interface SubagentContinuation { - /** Service-allocated stable child session id, published verbatim. */ +export interface ContinuableCreateRequest { + /** The reserved durable child session id, for provider diagnostics. */ readonly sessionId: SessionId - /** Snapshotted descriptor persisted in the child log for cold resume. */ - readonly descriptor: SubagentDescriptorData -} - -/** - * Provider-facing request for reconstructing a persisted continuable child. - * The continuation manager loads the child log, folds and authorizes its - * descriptor, then privately dispatches this resolved request to - * {@link SubagentProvider.resume}. The provider reconstructs the declared - * composition under the live parent's scope and drives one turn with `prompt`. - */ -export interface SubagentProviderResumeRequest { - /** The persisted child session id to resume. */ - readonly sessionId: SessionId - /** The follow-up message that starts the resumed activation's turn. */ - readonly prompt: ContentBlock[] - /** Attribution retained when the follow-up becomes the resumed turn's user-role message. */ - readonly source: MessageSource - /** - * The live parent agent — the direct parent recorded in the persisted child - * header. In-process backends reconstruct the child under this agent's - * currently loaded scope. - */ + /** The delegating parent agent whose history a seeding provider reads. */ readonly parent: Agent /** - * Activation-owned cancellation signal, created before descriptor lookup. - * Same pre/post-publication contract as {@link SubagentStartRequest.signal}: - * an abort before publication rejects after rollback quiescence, and an - * abort afterward cancels the published child turn. + * Caller cancellation, which owns preparation only until the manager accepts + * the initial prompt into the child's inbox. */ readonly signal: AbortSignal - /** The folded durable descriptor whose composition the provider reconstructs. */ - readonly descriptor: SubagentDescriptorData +} + +/** + * A provider's detached contribution to one continuable child's creation. This + * is DATA, never a capability: it carries no Agent, `AgentHandle`, prompt + * delivery, result, disposal, or resume operation, because the continuation + * manager owns the child's whole lifecycle after preparation. + */ +export interface ContinuableCreateSpec { + /** + * Completed-turn prefix of the parent's log to seed the child session with, + * or absent for a fresh child. Same durable contract as + * `CreateAgentOptions.seed`: contiguous from seq 0, lossless JSON, balanced. + */ + readonly seed?: readonly SessionEvent[] } /** @@ -196,9 +170,12 @@ export interface SubagentResult { } /** - * Child handle returned only after readiness. Consumers await {@link result} and must always - * {@link dispose} to cancel remaining work and reach quiescence. Optional methods are runtime - * capability discovery; narrow their presence before calling. + * ONE-SHOT child handle returned only after readiness. Consumers await + * {@link result} and must always {@link dispose} to cancel remaining work and + * reach quiescence. A run is one disposable foreground delegation with one + * result; continuable conversations have no run — the continuation manager + * holds their `AgentHandle` directly and orders every turn through the child's + * own inbox. */ export interface SubagentRun { /** @@ -217,10 +194,8 @@ export interface SubagentRun { * Resolves with the child's terminal {@link SubagentResult} when the run * settles. Does NOT reject on a child-level failure — a model/transport * failure resolves with `stopReason: 'error'` so the consumer maps it to an - * `isError` tool result. For a continuable activation, a completed result - * also means the provider confirmed the activation's final state durable. - * Rejects on an infrastructure fault the seam cannot represent as a stop - * reason, including a failed required durability checkpoint. + * `isError` tool result. Rejects on an infrastructure fault the seam cannot + * represent as a stop reason. */ readonly result: Promise /** @@ -228,17 +203,6 @@ export interface SubagentRun { * Idempotent. */ dispose(): Promise - /** - * OPTIONAL (confirmed live-steering capability): submit additional content - * to the active child and fulfill only after a committed request snapshot - * admits it. Rejects when terminal policy, cancellation, disposal, or a lost - * settlement race prevents admission; it never falls through to a queued - * untracked turn or cold resume. A run represents one disposable activation, - * so resuming a settled child goes through {@link SubagentProvider.resume}. - * `source` is retained on the admitted steering message without changing its - * user role in model history. - */ - steer?(content: ContentBlock[], source: MessageSource): Promise } /** @@ -258,23 +222,27 @@ export interface SubagentProvider { */ readonly inheritsParentContext: boolean /** - * Establish a child and return its handle only after publication. The - * service has already validated that every requested start-time capability - * is supported, so an implementation may assume e.g. `request.maxDepth` is - * honorable when present. If setup fails or `request.signal` aborts before - * fulfillment, the provider owns and cleans all partial resources before this - * promise rejects. Ownership transfers to the caller only on fulfillment. + * Establish a ONE-SHOT child and return its handle only after publication. + * The service has already validated that every requested start-time + * capability is supported, so an implementation may assume e.g. + * `request.maxDepth` is honorable when present. If setup fails or + * `request.signal` aborts before fulfillment, the provider owns and cleans + * all partial resources before this promise rejects. Ownership transfers to + * the caller only on fulfillment. */ - start(request: SubagentProviderStartRequest): Promise + start(request: SubagentStartRequest): Promise /** - * OPTIONAL (continuation capability): reconstruct a persisted continuable - * child from its own transcript and declared descriptor, drive one - * follow-up turn, and return a fresh run. Method presence is the capability - * — the service rejects continuable starts and cold-resume dispatch on - * providers without it. Same publication contract as {@link start}: if - * reconstruction fails or `request.signal` aborts before fulfillment, the - * provider rolls its creation transaction back to quiescence before - * rejecting; after fulfillment the same signal cancels the published run. + * OPTIONAL (continuable-creation capability): contribute the detached + * creation inputs that distinguish this provider's continuable children — + * today only whether the child session is seeded with parent history. Method + * presence IS the capability: the service rejects continuable starts on + * providers without it, while a provider that has it may still serve + * ordinary one-shot delegations. + * + * This is the provider's ONLY participation in a continuable child. The + * continuation manager owns identity reservation, composition, Agent + * creation, prompt delivery, cold resume, ownership, and disposal, so a + * provider never sees the child's Agent, handle, turns, or teardown. */ - resume?(request: SubagentProviderResumeRequest): Promise + prepareContinuable?(request: ContinuableCreateRequest): Promise } diff --git a/packages/subagent/tool-subagent-control/src/index.ts b/packages/subagent/tool-subagent-control/src/index.ts index af85457262..1fbbc3d3fb 100644 --- a/packages/subagent/tool-subagent-control/src/index.ts +++ b/packages/subagent/tool-subagent-control/src/index.ts @@ -1,9 +1,9 @@ /** * The globally named `send_message` tool: a thin model-facing adapter over - * `ctx.subagents.followup()`. It performs no lifecycle routing of its - * own — steer-or-resume orchestration belongs to the subagent service — and it - * lives apart from the provider-bound `@deepseek-ai/dsh-tool-subagent` - * instances so multiple delegation tools share one control tool. + * `ctx.subagents.followup()`. It performs no lifecycle routing of its own — + * residency and cold resume belong to the subagent service — and it lives apart + * from the provider-bound `@deepseek-ai/dsh-tool-subagent` instances so multiple + * delegation tools share one control tool. * @module @deepseek-ai/dsh-tool-subagent-control */ @@ -24,10 +24,10 @@ export function apply(ctx: Context): void { ctx.tools.register(defineTool({ name: 'send_message', description: - 'Send a follow-up message to a background subagent by its subagent id. If it is still working, the ' - + 'message joins its current task; if it has finished, this starts a new task that continues the same ' - + 'subagent conversation. Either way the response arrives through the returned task id — collect it ' - + 'with `task_output`. A failure means the message was NOT delivered.', + 'Send a message to a background subagent by its subagent id, continuing the same conversation. It ' + + 'becomes the subagent\'s next turn: if it is still working, the message waits until its current turn ' + + 'finishes, so it cannot redirect work already underway. The subagent does not reply to you — read its ' + + 'transcript by its id to see what it did. A failure means the message was NOT delivered.', parameters: { subagent_id: { type: 'string', @@ -45,30 +45,23 @@ export function apply(ctx: Context): void { type: 'object', additionalProperties: false, properties: { - route: { - type: 'string', - required: true, - enum: ['steered', 'started'], - }, - taskId: { type: 'string', required: true }, + messageId: { type: 'string', required: true }, }, }, - render: (args, value) => [{ + render: (args, _value) => [{ type: 'text', - text: value.route === 'steered' - ? `message delivered to running task ${value.taskId}` - : `message started task ${value.taskId} continuing subagent ${args.subagent_id}`, + text: `message queued as the next turn for subagent ${args.subagent_id}`, }], }, async execute(args, exec) { const parent = exec.agent if (!parent) { - // Non-agent callers have no session to authorize Task access with. + // Parent authority requires an exact live calling agent. throw new Error('send_message requires a calling agent (exec.agent was undefined)') } const message: ContentBlock[] = [{ type: 'text', text: args.message }] - const result = await ctx.subagents.followup( - parent, + const messageId = await ctx.subagents.followup( + { kind: 'parent', agent: parent }, SessionId(args.subagent_id), message, { @@ -76,7 +69,7 @@ export function apply(ctx: Context): void { signal: exec.signal, }, ) - return result + return { messageId } }, })) } diff --git a/packages/subagent/tool-subagent/src/index.ts b/packages/subagent/tool-subagent/src/index.ts index 6033bea875..62fef8c055 100644 --- a/packages/subagent/tool-subagent/src/index.ts +++ b/packages/subagent/tool-subagent/src/index.ts @@ -36,9 +36,9 @@ export interface Config { */ enableRunInBackground?: boolean /** - * Background execution policy (default `one-shot`). `continuable` requires - * a provider with persisted resume support and returns both child and Task - * ids; follow-up adapters remain independently optional. + * Background execution policy (default `one-shot`). `continuable` requires a + * provider with the `prepareContinuable` capability and returns the durable + * child id; follow-up adapters remain independently optional. */ backgroundMode?: 'one-shot' | 'continuable' /** @@ -197,7 +197,7 @@ export function apply(ctx: Context, config: Config): void { const wording = providerWording(provider.inheritsParentContext) const backgroundEnabled = config.enableRunInBackground !== false const continuable = (config.backgroundMode ?? 'one-shot') === 'continuable' - if (continuable && provider.resume === undefined) { + if (continuable && provider.prepareContinuable === undefined) { throw new Error( `tool-subagent: provider "${provider.name}" does not support \`backgroundMode: continuable\``, ) @@ -206,9 +206,9 @@ export function apply(ctx: Context, config: Config): void { name: config.toolName ?? 'subagent', description: wording.description + (backgroundEnabled ? continuable - ? ' Set `run_in_background: true` to start a continuable background subagent: you receive its' - + ' stable subagent id and current task id; collect the result with `task_output` and stop it with' - + ' `task_kill`.' + ? ' Set `run_in_background: true` to start a background subagent that keeps its conversation:' + + ' you receive its subagent id and it works on its own. It does not report back to you, so read' + + ' its transcript by that id, or send it more work with `send_message`.' : ' Set `run_in_background: true` to return a task id; collect with `task_output` and stop with `task_kill`.' : ''), parameters: { @@ -226,8 +226,8 @@ export function apply(ctx: Context, config: Config): void { run_in_background: { type: 'boolean' as const, description: continuable - ? 'Run as a continuable background subagent and return its subagent and task ids; ' - + 'collect with task_output or stop with task_kill.' + ? 'Run as a background subagent that keeps its conversation and return its subagent id; ' + + 'send it more work with send_message.' : 'Run as a background task and return its id; collect with task_output or stop with task_kill.', }, } : {}, @@ -241,7 +241,14 @@ export function apply(ctx: Context, config: Config): void { properties: { kind: { type: 'string', required: true, const: 'background' }, taskId: { type: 'string', required: true }, - subagentId: { type: 'string' }, + }, + }, + { + type: 'object', + additionalProperties: false, + properties: { + kind: { type: 'string', required: true, const: 'continuable' }, + subagentId: { type: 'string', required: true }, }, }, { @@ -258,10 +265,10 @@ export function apply(ctx: Context, config: Config): void { render: (_args, value) => [{ type: 'text', text: value.kind === 'background' - ? value.subagentId === undefined - ? `started background subagent task ${value.taskId}` - : `started subagent ${value.subagentId} as task ${value.taskId}` - : outputValueText(value.output), + ? `started background subagent task ${value.taskId}` + : value.kind === 'continuable' + ? `started subagent ${value.subagentId}` + : outputValueText(value.output), }], }, async execute(args, exec) { @@ -288,16 +295,14 @@ export function apply(ctx: Context, config: Config): void { throw new Error('run_in_background is disabled for this tool instance (enableRunInBackground: false)') } if (continuable) { - const started = ctx.subagents.startContinuable({ + // Resolves at inbox acceptance: the child owns its own turns from + // there, so this call neither waits for nor collects a result. + const started = await ctx.subagents.startContinuable({ provider: config.provider, - label: args.description, request, + signal: exec.signal, }) - return { - kind: 'background' as const, - taskId: started.taskId, - subagentId: started.childId, - } + return { kind: 'continuable' as const, subagentId: started.childId } } const tasks = ctx.get('tasks') if (tasks === undefined) {