Merge remote-tracking branch 'origin/worktree-agent-scope-design' into codex/package-readme-limitations-audit-20260712
# Conflicts: # packages/core/scope/README.md # packages/session-persistence/session-persistence-jsonl/README.md # packages/session-persistence/session-persistence-sqlite/README.md # packages/subagent/subagent-acp/README.md # packages/subagent/subagent-fork/README.md # packages/subagent/subagent-inprocess/README.md # packages/subagent/subagent/README.md # packages/subagent/tool-subagent/README.md # packages/support/invariants/README.md # packages/support/subagent-mock/README.md # packages/workflow/workflow-workerthread/README.md # packages/workflow/workflow/README.md
This commit is contained in:
Tianyi Cui
@@ -1,44 +1,66 @@
|
||||
# @deepseek-ai/dsh-subagent
|
||||
|
||||
The **subagent seam**: an abstract `SubagentService` (`ctx.subagents`) for an agent delegating work to another agent. A *subagent* is a child agent; a `SubagentProvider` is one transport for running it.
|
||||
The subagent seam lets one agent delegate work to a child through a named provider. Callers use one service API (`ctx.subagents`); providers decide whether the child runs in this process, in another process, or through a future transport.
|
||||
|
||||
This package is the interface third of the capability seam, split so each concern evolves (and swaps) independently:
|
||||
## Package roles
|
||||
|
||||
The family separates the stable interface from implementations and model-facing tools:
|
||||
|
||||
| Package | Role |
|
||||
|---|---|
|
||||
| `@deepseek-ai/dsh-subagent` (this) | the interface: registry service + vocabulary types |
|
||||
| `@deepseek-ai/dsh-subagent-spawn` | an implementation: fresh in-process child |
|
||||
| `@deepseek-ai/dsh-subagent-fork` | an implementation: in-process child seeded from the parent's log |
|
||||
| `@deepseek-ai/dsh-subagent-acp` | an implementation: ACP client driving another process |
|
||||
| `@deepseek-ai/dsh-tool-subagent` | the model-facing tool over `ctx.subagents` |
|
||||
| `@deepseek-ai/dsh-subagent` | Provider registry, request/result types, and lifecycle events. |
|
||||
| `@deepseek-ai/dsh-subagent-spawn` | Fresh in-process child. |
|
||||
| `@deepseek-ai/dsh-subagent-fork` | In-process child seeded with completed parent turns. |
|
||||
| `@deepseek-ai/dsh-subagent-acp` | Fresh out-of-process ACP child. |
|
||||
| `@deepseek-ai/dsh-tool-subagent` | Model-facing tool over one configured provider. |
|
||||
|
||||
Unlike the bash seam (one executor per context, second load throws), **multiple providers coexist** here. Each registers under a unique name and a caller picks one by name — the shape mirrors the LLM adapter registry (`LlmService.registerAdapter`), not the single-service bash executor. This is the requirement that rules out the bash shape: an agent may want an in-process child for a cheap subtask and an out-of-process ACP child for an isolated one, in the same runtime.
|
||||
Multiple providers may coexist under different names. This lets a deployment expose, for example, a cheap in-process child and an isolated ACP child without changing the service contract.
|
||||
|
||||
## Service API (`ctx.subagents`)
|
||||
## Service API
|
||||
|
||||
| Member | Semantics |
|
||||
`SubagentService` has four main operations:
|
||||
|
||||
| Member | Meaning |
|
||||
|---|---|
|
||||
| `registerProvider(provider)` | Register a frozen acceptance snapshot under `provider.name`; later caller mutation cannot change registry behavior or HMR cleanup, while `start` stays bound to the original provider receiver. Throws `SubagentError('DUPLICATE_PROVIDER')` on a name clash. Effect-scoped (HMR-safe); returns the disposer. |
|
||||
| `getProvider(name)` | Look up the frozen registry snapshot (`undefined` if absent). |
|
||||
| `list()` | Registered provider names (insertion order). |
|
||||
| `start(name, request)` | Resolve the provider (`NO_PROVIDER` if absent), validate every requested START-TIME capability (`UNSUPPORTED_CAPABILITY` for the first unmet one — before any child is created), then delegate to `provider.start`. Emit `subagent/start` only after `run.started` fulfills and the paired `subagent/end` after that started run settles; a pre-publication readiness rejection emits neither. |
|
||||
| `registerProvider(provider)` | Register one trusted same-process implementation by name. Registration is effect-scoped; removing it prevents new starts but does not revoke runs already returned to callers. Duplicate names fail loud. |
|
||||
| `getProvider(name)` | Return the provider, or `undefined` when absent. |
|
||||
| `list()` | Return provider names in insertion order. |
|
||||
| `start(name, request)` | Validate requested capabilities and semantic values, then await the provider until a real child is ready. Fulfillment returns a holder-owned `SubagentRun`; rejection means the provider has already cleaned every partial startup resource. |
|
||||
|
||||
## Capabilities: two kinds, discovered two ways
|
||||
`SubagentStartRequest.signal` is required and is the canonical cancellation channel. An abort before publication makes `start()` reject after rollback; an abort after publication cancels the live child. The request may also select a model, require structured output, cap delegation depth, restrict child tools, or set a child persona.
|
||||
|
||||
- **Start-time features** (`outputSchema`, `depthLimit`, `toolFilter/persona`) are a static `provider.capabilities` descriptor, checked by the service BEFORE a run exists. A request that needs one the provider lacks is **rejected loud** (`UNSUPPORTED_CAPABILITY`), never accepted-then-ignored.
|
||||
- **Runtime features** (steering, resume) are **optional methods** on `SubagentRun` (`sendMessage?`, `resume?`). The method's presence IS the capability; TS narrowing is the discovery mechanism — a consumer cannot call an absent method without narrowing first, so there is no silent degradation path.
|
||||
Same-process requests, descriptors, results, and event payloads are trusted typed values borrowed as immutable. The service does not clone or freeze them; serialization and hostile-input validation belong at actual process, worker, persistence, and model boundaries.
|
||||
|
||||
Beside `capabilities` sits one DESCRIPTIVE fact, not validated by the service: `provider.inheritsParentContext` — whether a child sees the parent conversation (`fork`: true — seeded with the completed-turn prefix; `spawn`/`acp`: false). The model-facing consumer (`dsh-tool-subagent`) derives truthful tool wording from it.
|
||||
## Capabilities
|
||||
|
||||
## Run lifecycle
|
||||
Start-time features are advertised in `provider.capabilities` because the service must reject an unsupported request before child creation:
|
||||
|
||||
`provider.start(request)` returns a `SubagentRun`: a handle with `started` (the publication/readiness promise), `result` (the terminal outcome), `cancel()`, `dispose()`, and the optional runtime methods. `started` resolves only after the provider has established a real child and rejects if the attempt fails or is cancelled first. `result` resolves with a `SubagentResult` (`output`, optional `structured`, `stopReason`) — it does **not** reject on a child-level failure (a model/transport failure resolves with `stopReason: 'error'`), so the consumer maps a non-`completed` reason to an `isError` tool result. The consumer MUST `dispose()` on every path (success, error, abort) to reach child quiescence and avoid leaking an idle child / session.
|
||||
- `outputSchema` — enforce a structured final result.
|
||||
- `depthLimit` — enforce `maxDepth`.
|
||||
- `toolFilter` — apply the requested child tool restriction.
|
||||
- `persona` — apply a per-child persona.
|
||||
|
||||
The service also announces provider lifecycle: `subagent/provider-added` (the frozen registry snapshot) fires after a registration and `subagent/provider-removed` (the accepted name) after an unregistration, so a consumer deriving state from a named provider (the model-facing tool wording) mirrors registry membership instead of assuming load order — the cordis Loader starts sibling plugins concurrently, so "listed earlier" does not mean "registered earlier". Run lifecycle is gated by provider readiness: `subagent/start` (payload `SubagentRunInfo`) fires only after `run.started` fulfills, and `subagent/end` (payload `SubagentRunEndInfo`) fires only for that announced run; readiness rejection emits neither. For spawn/fork, the start listener can resolve the published child via `ctx.agents.get(info.id)`; a remote provider need not have a local registry entry. Both events are **observe-only** plain emits. The service observes `result` immediately even while readiness is pending, clones its output before the caller can mutate it, and buffers that end payload until start has fired; a rejecting result cannot become an unhandled detached promise, start always precedes end, and a listener cannot corrupt the caller's result. `subagent/end` carries the cloned output as `lastAssistantMessage` on the settle path and omits it on infrastructure rejection. Any run-affecting decision is out of scope for this observe-only surface.
|
||||
Runtime features are optional methods on `SubagentRun`: `sendMessage?` steers a live child, while `resume?` asynchronously creates a continuation run. Method presence is the capability check.
|
||||
|
||||
`inheritsParentContext` is descriptive rather than enforceable. It says only whether the child sees completed parent conversation history (`fork` does; `spawn` and ACP do not), not whether it inherits tools, services, or authority.
|
||||
|
||||
## Ownership and lifecycle
|
||||
|
||||
`provider.start(request): Promise<SubagentRun>` is the ownership-transfer boundary. Before fulfillment, the provider owns setup and must cancel, roll back, and quiesce partial resources on every failure. After fulfillment, the caller owns the run and must call `dispose()` on every path.
|
||||
|
||||
`SubagentRun.result` resolves to `{ output, structured?, stopReason }`. Child-level failures resolve with a non-`completed` reason; only an infrastructure fault that the seam cannot represent may reject. `dispose()` is idempotent, cancels remaining work, and waits for the child resources to quiesce.
|
||||
|
||||
The service emits `subagent/start` only after `start()` has fulfilled. It attaches the result observer before that synchronous notification, so even an already-settled child still produces `subagent/start` before `subagent/end`. In-process start observers can resolve the published child through `ctx.agents.get(info.id)`; remote providers need not publish a local agent.
|
||||
|
||||
Run events are scoped to the delegating parent. Every listener is independently contained: a synchronous throw or rejected returned promise is logged without starving peer listeners or changing the run.
|
||||
|
||||
Provider additions and removals also emit `subagent/provider-added` and `subagent/provider-removed`. Consumers such as the model-facing tool use those events because Cordis may load sibling plugins concurrently; configuration order does not prove registration order.
|
||||
|
||||
## Collection model
|
||||
|
||||
The current model-facing tool collects synchronously: it awaits the child result and disposes the run before returning. Background collection and polling remain outside this seam. See the [capability-seam RFC](../../../docs/rfc/implemented/feature/2026-06-21-subagent-capability-seam.md) and `src/types.ts` for the complete contracts.
|
||||
|
||||
## Known Limitations and Deferred Work
|
||||
|
||||
- **The consumer collects synchronously** — it starts a run and awaits `result`; steering (`sendMessage`) is part of the contract but intentionally unused, and background / poll / spill semantics are deferred to a future redesign unifying long-running-tool handling across subagents and bash ([the seam RFC](../../../docs/rfc/implemented/feature/2026-06-21-subagent-capability-seam.md)).
|
||||
- **The lifecycle events are observe-only** — a run-affecting `subagent/end` (an awaited continuation/decision surface) is deferred until a consumer needs one (`FIXME(subagent-continuation)`).
|
||||
|
||||
See `src/types.ts` for the full contracts.
|
||||
- **The current consumer collects synchronously** — the model-facing tool starts a run and awaits `result`; steering (`sendMessage`) is part of the seam but intentionally unused, and background/poll/spill semantics are deferred to a future long-running-runtime design.
|
||||
- **The lifecycle events are observe-only** — a run-affecting `subagent/end` continuation or decision surface is deferred until a consumer needs one.
|
||||
|
||||
@@ -1,41 +1,21 @@
|
||||
/**
|
||||
* The subagent seam (`ctx.subagents`): a named-provider registry plus a
|
||||
* capability-validating `start` surface. A subagent is an agent delegating
|
||||
* work to another agent; a {@link SubagentProvider} is one transport for
|
||||
* running that child (in-process spawn/fork, ACP to another process, and —
|
||||
* later — A2A, the Codex app-server, the Claude Code Agent SDK).
|
||||
* capability-validating asynchronous start surface. Providers establish a
|
||||
* child before returning its run, so fulfillment is the single publication and
|
||||
* ownership-transfer boundary.
|
||||
*
|
||||
* Unlike the bash seam (one executor per context, second load throws), MULTIPLE
|
||||
* providers coexist here: each registers under a unique name and a caller picks
|
||||
* one by name. The shape mirrors the LLM adapter registry
|
||||
* (`LlmService.registerAdapter`), not the single-service bash executor.
|
||||
*
|
||||
* This package is the INTERFACE third of the capability seam. Implementations
|
||||
* (`@deepseek-ai/dsh-subagent-spawn`, `-fork`, `-acp`) and the model-facing
|
||||
* consumer (`@deepseek-ai/dsh-tool-subagent`) are separate packages.
|
||||
*
|
||||
* Scope (first cut): the consumer collects synchronously — it starts a run and
|
||||
* awaits {@link SubagentRun.result}. Steering ({@link SubagentRun.sendMessage})
|
||||
* is part of the contract but intentionally unused; background / poll / spill
|
||||
* semantics are deferred to a future redesign that unifies long-running-tool
|
||||
* handling across subagents and bash.
|
||||
*
|
||||
* The `subagent/start` / `subagent/end` lifecycle events carry an OBSERVE-ONLY
|
||||
* payload; `subagent/end` additionally carries the child's `lastAssistantMessage`
|
||||
* — see `docs/rfc/implemented/feature/2026-06-30-subagent-observe-enrich.md`.
|
||||
* FIXME(subagent-continuation): a control-flow `subagent/end` (an awaited
|
||||
* waterfall returning a stop/continue decision, like the other interception
|
||||
* seams) would require reshaping this emit into a waterfall, awaiting listeners
|
||||
* before settling, and a `resume` capability on the in-process provider — part
|
||||
* of the deferred background/steering redesign, NOT this observe-only cut.
|
||||
* Same-process providers are trusted typed collaborators. Requests, provider
|
||||
* descriptors, results, and lifecycle payloads are borrowed immutable values;
|
||||
* serialization and hostile-input validation belong at real process, worker,
|
||||
* persistence, and model boundaries.
|
||||
*
|
||||
* @module @deepseek-ai/dsh-subagent
|
||||
*/
|
||||
|
||||
import { Context, Service } from 'cordis'
|
||||
import { scopeTarget } from '@deepseek-ai/dsh-scope'
|
||||
import { assertSupportedOutputSchema } from '@deepseek-ai/dsh-tools'
|
||||
import type { Scoped } from '@deepseek-ai/dsh-scope'
|
||||
import { assertSupportedOutputSchema } from '@deepseek-ai/dsh-tools'
|
||||
import { HarnessError } from '@deepseek-ai/dsh-llm'
|
||||
import type { ContentBlock } from '@deepseek-ai/dsh-llm'
|
||||
import type { Agent, AgentId } from '@deepseek-ai/dsh-agent'
|
||||
@@ -57,6 +37,21 @@ export type {
|
||||
SubagentStopReasonMap,
|
||||
} from './types.ts'
|
||||
|
||||
/**
|
||||
* 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
|
||||
@@ -64,88 +59,59 @@ declare module 'cordis' {
|
||||
|
||||
interface Events {
|
||||
/**
|
||||
* A provider became resolvable in the {@link SubagentService} registry.
|
||||
* Consumers that derive state from a named provider (e.g. the model-facing
|
||||
* tool wording in `dsh-tool-subagent`) react HERE instead of assuming load
|
||||
* order — the cordis Loader starts sibling plugins concurrently, so
|
||||
* "listed earlier in cordis.yml" does not mean "registered earlier".
|
||||
* @param provider - the registry's frozen acceptance snapshot of the provider.
|
||||
* A provider became resolvable in the registry.
|
||||
* @param provider - the registered provider.
|
||||
* @mode emit
|
||||
*/
|
||||
'subagent/provider-added'(provider: SubagentProvider): void
|
||||
/**
|
||||
* A provider left the registry (its plugin's fiber was disposed — an
|
||||
* unload or an HMR reload). Consumers holding provider-derived state drop
|
||||
* it here; a reload re-fires `subagent/provider-added` with the fresh
|
||||
* provider. Delivered with per-listener containment: a throwing
|
||||
* subscriber is logged, never starves later subscribers, and never
|
||||
* disrupts the provider's teardown.
|
||||
* @param name - the registry name that no longer resolves.
|
||||
* A provider left the registry. Accepted runs remain holder-owned.
|
||||
* @param name - the provider name that no longer resolves.
|
||||
* @mode emit
|
||||
*/
|
||||
'subagent/provider-removed'(name: string): void
|
||||
/**
|
||||
* A subagent run started — emitted only after {@link SubagentRun.started}
|
||||
* fulfills, when the provider has established a live child. For an
|
||||
* in-process provider, `ctx.agents.get(info.id)` is therefore guaranteed to
|
||||
* resolve during this notification. A readiness rejection emits neither
|
||||
* lifecycle event; every emitted start is paired with
|
||||
* {@link Events['subagent/end']}.
|
||||
* Scope-filtered dispatch (`@deepseek-ai/dsh-scope`): the carrier is keyed
|
||||
* by the DELEGATING PARENT — a listener registered through the parent's
|
||||
* `agent.ctx` observes only its own delegations; a plain plugin listener
|
||||
* observes every run.
|
||||
* @param info - which provider started which child agent.
|
||||
* A provider established a ready child. For in-process providers,
|
||||
* `ctx.agents.get(info.id)` resolves during this notification.
|
||||
* Scope-filtered dispatch keys the carrier by the delegating parent, so a
|
||||
* parent-scoped listener observes only its own delegations. Paired with
|
||||
* `subagent/end`.
|
||||
* @param info - the provider and ready child identity.
|
||||
* @mode emit
|
||||
*/
|
||||
'subagent/start'(this: Scoped<SubagentService>, info: SubagentRunInfo): void
|
||||
/**
|
||||
* A started subagent run settled — emitted when {@link SubagentRun.result}
|
||||
* resolves (any stop reason) or rejects (reported as `error`). Paired with
|
||||
* {@link Events['subagent/start']}; a run whose readiness rejected emits
|
||||
* neither event.
|
||||
* Scope-filtered dispatch (`@deepseek-ai/dsh-scope`): the carrier is keyed
|
||||
* by the DELEGATING PARENT — a listener registered through the parent's
|
||||
* `agent.ctx` observes only its own delegations; a plain plugin listener
|
||||
* observes every run.
|
||||
* @param info - the run identity plus stop reason and final output.
|
||||
* A ready child settled. Scope-filtered dispatch uses the same delegating
|
||||
* parent carrier as `subagent/start`, so the lifecycle pair reaches the
|
||||
* same scoped audience.
|
||||
* @param info - the run identity and terminal outcome.
|
||||
* @mode emit
|
||||
*/
|
||||
'subagent/end'(this: Scoped<SubagentService>, info: SubagentRunEndInfo): void
|
||||
}
|
||||
}
|
||||
|
||||
/** Identifying detail for a started subagent run (the `subagent/start` payload). */
|
||||
/** Observe-only identifying detail for a ready subagent run. */
|
||||
export interface SubagentRunInfo {
|
||||
/** The provider that started the run. */
|
||||
provider: string
|
||||
/** The provider that established the run. */
|
||||
readonly provider: string
|
||||
/** The child agent's id. */
|
||||
id: AgentId
|
||||
readonly id: AgentId
|
||||
}
|
||||
|
||||
/** Outcome detail for a settled subagent run (the `subagent/end` payload). */
|
||||
/** Observe-only outcome detail for a settled subagent run. */
|
||||
export interface SubagentRunEndInfo {
|
||||
/** The provider that ran it. */
|
||||
provider: string
|
||||
readonly provider: string
|
||||
/** The child agent's id. */
|
||||
id: AgentId
|
||||
readonly id: AgentId
|
||||
/** The terminal stop reason. */
|
||||
stopReason: SubagentResult['stopReason']
|
||||
/**
|
||||
* The child's final assistant output ({@link SubagentResult.output}), carried
|
||||
* onto the end event so an observer sees WHAT the subagent produced without
|
||||
* holding the run. Absent when the run rejected at the infrastructure level
|
||||
* (no {@link SubagentResult} was produced — the seam only knows `stopReason:
|
||||
* 'error'`).
|
||||
*/
|
||||
lastAssistantMessage?: ContentBlock[]
|
||||
readonly stopReason: SubagentResult['stopReason']
|
||||
/** The child's final assistant output, absent on infrastructure rejection. */
|
||||
readonly lastAssistantMessage?: ContentBlock[]
|
||||
}
|
||||
|
||||
/**
|
||||
* Typed error for subagent-seam failures. Extends {@link HarnessError}, so the
|
||||
* `code` string (`DUPLICATE_PROVIDER`, `NO_PROVIDER`, `UNSUPPORTED_CAPABILITY`)
|
||||
* is shared, machine-routable taxonomy.
|
||||
*/
|
||||
/** Typed error for provider lookup, registration, and capability failures. */
|
||||
export class SubagentError extends HarnessError {
|
||||
constructor(message: string, code: string, options?: ErrorOptions) {
|
||||
super(message, code, options)
|
||||
@@ -153,10 +119,7 @@ export class SubagentError extends HarnessError {
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* The `subagents` service: a registry of named {@link SubagentProvider}s and a
|
||||
* capability-checked {@link start} surface.
|
||||
*/
|
||||
/** Named provider registry and capability-checked start surface. */
|
||||
export class SubagentService extends Service {
|
||||
private providers = new Map<string, SubagentProvider>()
|
||||
|
||||
@@ -165,202 +128,88 @@ export class SubagentService extends Service {
|
||||
}
|
||||
|
||||
/**
|
||||
* Register a provider under its `provider.name`. Throws {@link SubagentError}
|
||||
* (`DUPLICATE_PROVIDER`) if the name is already taken. The registry snapshots
|
||||
* the name, static descriptors, and `start` callback identity at acceptance;
|
||||
* later caller mutation cannot change lookup, capability validation, consumer
|
||||
* wording, dispatch, or HMR cleanup. The callback remains bound to the
|
||||
* original provider object, so provider-owned mutable state stays live.
|
||||
* Effect-scoped: disposed with the calling fiber (HMR-safe). Emits
|
||||
* `subagent/provider-added` after the registration and
|
||||
* `subagent/provider-removed` on unregistration, so consumers can mirror
|
||||
* provider lifecycle instead of assuming load order.
|
||||
* @param provider - the provider; its `name` is the registry key.
|
||||
* @returns the disposer that unregisters the provider. The exact
|
||||
* Cordis effect disposer (single-shot): composite (generator) effects may
|
||||
* yield it directly — exact identity nests the teardown in order.
|
||||
* Register a provider under its name. Registration is effect-scoped and HMR
|
||||
* safe; removing a provider blocks new starts but does not revoke runs that
|
||||
* were already returned to their holders.
|
||||
* @param provider - the trusted provider implementation.
|
||||
* @returns the exact Cordis effect disposer.
|
||||
*/
|
||||
registerProvider(provider: SubagentProvider): () => Promise<void> | void {
|
||||
// Snapshot the accepted registration contract before entering the effect.
|
||||
// Cleanup must never re-read caller-owned `provider.name`: an HMR host may
|
||||
// mutate or reuse the provider object before its old fiber unloads. Binding
|
||||
// preserves the provider method's receiver while making replacement of the
|
||||
// public callback field after registration inert.
|
||||
const capabilities: SubagentCapabilities = Object.freeze({
|
||||
outputSchema: provider.capabilities.outputSchema,
|
||||
depthLimit: provider.capabilities.depthLimit,
|
||||
toolFilter: provider.capabilities.toolFilter,
|
||||
persona: provider.capabilities.persona,
|
||||
})
|
||||
const snapshot: SubagentProvider = Object.freeze({
|
||||
name: provider.name,
|
||||
capabilities,
|
||||
inheritsParentContext: provider.inheritsParentContext,
|
||||
start: provider.start.bind(provider),
|
||||
})
|
||||
const dispose = this.ctx.effect(function* (this: SubagentService) {
|
||||
if (this.providers.has(snapshot.name)) {
|
||||
throw new SubagentError(`a subagent provider named "${snapshot.name}" is already registered`, 'DUPLICATE_PROVIDER')
|
||||
const name = provider.name
|
||||
return this.ctx.effect(function* (this: SubagentService) {
|
||||
if (this.providers.has(name)) {
|
||||
throw new SubagentError(`a subagent provider named "${name}" is already registered`, 'DUPLICATE_PROVIDER')
|
||||
}
|
||||
this.providers.set(snapshot.name, snapshot)
|
||||
// Yield the rollback BEFORE emitting `subagent/provider-added`: a
|
||||
// throwing added-listener then unregisters the provider (and announces
|
||||
// the removal) instead of leaking it into the registry. The removal
|
||||
// announcement itself is contained PER LISTENER ({@link emitLifecycle}):
|
||||
// it runs inside this disposer, where a propagating subscriber would
|
||||
// disrupt the backend fiber's teardown and starve later mirrors.
|
||||
this.providers.set(name, provider)
|
||||
yield () => {
|
||||
this.providers.delete(snapshot.name)
|
||||
this.emitLifecycle('subagent/provider-removed', snapshot.name)
|
||||
this.providers.delete(name)
|
||||
this.emitLifecycle('subagent/provider-removed', name)
|
||||
}
|
||||
this.ctx.emit('subagent/provider-added', snapshot)
|
||||
// A throwing added-listener unwinds the yielded rollback, matching the
|
||||
// repository's fail-loud registration semantics.
|
||||
this.ctx.emit('subagent/provider-added', provider)
|
||||
}.bind(this), 'subagents.registerProvider()')
|
||||
// The EXACT cordis effect disposer, not a wrapper: a composite (generator)
|
||||
// effect that owns a teardown ORDER must be able to yield THIS function —
|
||||
// cordis nests a disposer out of the fiber's concurrent sibling list by
|
||||
// exact function identity, so a wrapper would silently break the nesting
|
||||
// (the agents.register() lesson). Fire-and-forget callers may still
|
||||
// discard the (always-resolved) promise.
|
||||
return dispose
|
||||
}
|
||||
|
||||
/**
|
||||
* Look up the registry's frozen provider snapshot by its accepted name
|
||||
* (`undefined` if absent).
|
||||
* @param name - the provider name accepted at registration.
|
||||
* @returns the frozen acceptance snapshot, or undefined when the name is unknown.
|
||||
* Look up a provider by name.
|
||||
* @param name - the provider name.
|
||||
* @returns the provider, or undefined when absent.
|
||||
*/
|
||||
getProvider(name: string): SubagentProvider | undefined {
|
||||
return this.providers.get(name)
|
||||
}
|
||||
|
||||
/**
|
||||
* The names of all registered providers (insertion order).
|
||||
* @returns the registered provider names.
|
||||
* List registered provider names in insertion order.
|
||||
* @returns the registered names.
|
||||
*/
|
||||
list(): string[] {
|
||||
return [...this.providers.keys()]
|
||||
}
|
||||
|
||||
/**
|
||||
* Start a subagent run on the named provider. Resolves the provider (throws
|
||||
* `NO_PROVIDER` if absent), validates every requested START-TIME capability
|
||||
* against {@link SubagentProvider.capabilities} (throws `UNSUPPORTED_CAPABILITY`
|
||||
* for the first unmet one — fail loud, before any child is created), then
|
||||
* delegates to {@link SubagentProvider.start}, then emits `subagent/start` /
|
||||
* `subagent/end` only after the run's readiness boundary fulfills. A provider
|
||||
* that fails before establishing a child emits neither event.
|
||||
* @param name - the provider to run on.
|
||||
* @param request - the child's prompt, capabilities, and options.
|
||||
* @returns the live run (its `result` resolves when the child settles).
|
||||
* Establish a ready child on the named provider. Capability and semantic
|
||||
* checks run before delegation. Provider ownership lasts until its promise
|
||||
* fulfills; a rejection therefore has no run for the caller to dispose and
|
||||
* emits no run lifecycle events.
|
||||
* @param name - the provider to use.
|
||||
* @param request - child prompt, parent, signal, and optional capabilities.
|
||||
* @returns the ready holder-owned run.
|
||||
*/
|
||||
start(name: string, request: SubagentStartRequest): SubagentRun {
|
||||
// Parent is the lifecycle scope identity accepted at start. Never reread it
|
||||
// from the caller-owned request after the provider/result async boundary,
|
||||
// or start/end could be dispatched into different agent scopes.
|
||||
const parent = request.parent
|
||||
async start(name: string, request: SubagentStartRequest): Promise<SubagentRun> {
|
||||
const provider = this.providers.get(name)
|
||||
if (!provider) {
|
||||
if (provider === undefined) {
|
||||
throw new SubagentError(`no subagent provider registered for "${name}"`, 'NO_PROVIDER')
|
||||
}
|
||||
this.assertCapabilities(provider, request)
|
||||
assertSubagentMaxDepth(request.maxDepth)
|
||||
if (request.outputSchema !== undefined) assertSupportedOutputSchema(request.outputSchema)
|
||||
|
||||
// Detach every data field before crossing into a provider. Parent/signal
|
||||
// are live identity capabilities and stay exact; the mutable request record
|
||||
// and its arrays/objects are never retained, so every backend (including an
|
||||
// async out-of-process one) observes the request accepted at start.
|
||||
const accepted: SubagentStartRequest = {
|
||||
prompt: structuredClone(request.prompt),
|
||||
parent,
|
||||
...request.signal !== undefined ? { signal: request.signal } : {},
|
||||
...request.agentOptions !== undefined ? { agentOptions: structuredClone(request.agentOptions) } : {},
|
||||
...request.outputSchema !== undefined ? { outputSchema: structuredClone(request.outputSchema) } : {},
|
||||
...request.maxDepth !== undefined ? { maxDepth: request.maxDepth } : {},
|
||||
...request.toolFilter !== undefined ? { toolFilter: structuredClone(request.toolFilter) } : {},
|
||||
...request.persona !== undefined ? { persona: request.persona } : {},
|
||||
}
|
||||
const run = provider.start(accepted)
|
||||
|
||||
// Observe result settlement IMMEDIATELY, before waiting on readiness. A
|
||||
// provider may fail both promises in the same turn; deferring the rejection
|
||||
// handler until `started` fulfilled would leave `result` transiently
|
||||
// unhandled. The settled event is buffered until start has been announced,
|
||||
// preserving start → end order even for an already-settled scripted run.
|
||||
let readiness: 'pending' | 'started' | 'failed' = 'pending'
|
||||
let pendingEnd: SubagentRunEndInfo | undefined
|
||||
const deliverEnd = (info: SubagentRunEndInfo): void => {
|
||||
if (readiness === 'started') this.emitLifecycle('subagent/end', info, parent)
|
||||
else if (readiness === 'pending') pendingEnd = info
|
||||
// A pre-publication readiness failure has no lifecycle pair; result
|
||||
// remains observable by the run's consumer, but telemetry must not claim
|
||||
// that a child started.
|
||||
}
|
||||
const parent = request.parent
|
||||
const run = await provider.start(request)
|
||||
// Attach the terminal observer before dispatching start. Promise reactions
|
||||
// still run after this synchronous start emission, preserving start → end.
|
||||
void run.result.then(
|
||||
(result) => {
|
||||
// Snapshot before the caller's own `await run.result` continuation. Even
|
||||
// when readiness is still pending, buffering the clone rather than the
|
||||
// caller-owned result keeps the eventual observe-only event immutable
|
||||
// with respect to consumer mutation.
|
||||
let lastAssistantMessage: SubagentResult['output'] | undefined
|
||||
try {
|
||||
lastAssistantMessage = structuredClone(result.output)
|
||||
} catch (error: unknown) {
|
||||
this.ctx.logger.warn(`subagent: could not clone ${name} output for subagent/end: ${String(error)}`)
|
||||
}
|
||||
deliverEnd({
|
||||
this.emitLifecycle('subagent/end', {
|
||||
provider: name,
|
||||
id: run.id,
|
||||
stopReason: result.stopReason,
|
||||
...lastAssistantMessage !== undefined ? { lastAssistantMessage } : {},
|
||||
})
|
||||
},
|
||||
() => { deliverEnd({ provider: name, id: run.id, stopReason: 'error' }) },
|
||||
)
|
||||
|
||||
// Readiness is the publication boundary owned by the provider. For
|
||||
// in-process runs, fulfillment means the agent registry already contains
|
||||
// `run.id`; for ACP it means the remote session exists. Emit start with
|
||||
// per-listener containment, then flush an outcome that settled unusually
|
||||
// early. A readiness rejection is handled here and deliberately emits no
|
||||
// false start/end pair; the result path above remains independently handled.
|
||||
void run.started.then(
|
||||
() => {
|
||||
readiness = 'started'
|
||||
this.emitLifecycle('subagent/start', { provider: name, id: run.id }, parent)
|
||||
if (pendingEnd !== undefined) {
|
||||
const info = pendingEnd
|
||||
pendingEnd = undefined
|
||||
this.emitLifecycle('subagent/end', info, parent)
|
||||
}
|
||||
lastAssistantMessage: result.output,
|
||||
}, parent)
|
||||
},
|
||||
() => {
|
||||
readiness = 'failed'
|
||||
pendingEnd = undefined
|
||||
this.emitLifecycle('subagent/end', { provider: name, id: run.id, stopReason: 'error' }, parent)
|
||||
},
|
||||
)
|
||||
this.emitLifecycle('subagent/start', { provider: name, id: run.id }, parent)
|
||||
return run
|
||||
}
|
||||
|
||||
/**
|
||||
* Emit a `subagent/*` lifecycle event with PER-LISTENER containment: dispatch
|
||||
* each subscriber individually and log (never propagate) a thrown one, so one
|
||||
* bad subscriber can neither strand the already-live run, surface as an
|
||||
* unhandled rejection on the detached settle hook, NOR starve the listeners
|
||||
* registered after it. A single try/catch around `ctx.emit` would not do the
|
||||
* last part — cordis `emit` runs listeners in a `.map(cb => cb())` that halts
|
||||
* on the first throw — so this resolves the listener callbacks via
|
||||
* `ctx.events.dispatch` and contains each call, the same guarantee
|
||||
* `BashExecutor.notifyTaskDone` gives its own listener set.
|
||||
*
|
||||
* `subagent/provider-removed` routes through here too: it fires inside the
|
||||
* provider registration's DISPOSER, where a propagating listener would
|
||||
* disrupt the backend fiber's teardown (dispose must reach quiescence) and a
|
||||
* starved later listener would leave a mirror consumer (`dsh-tool-subagent`)
|
||||
* holding a tool for a provider that no longer exists. `subagent/provider-added`
|
||||
* deliberately does NOT: it fires at registration time, where a throwing
|
||||
* listener unwinds the yielded rollback — the same fail-loud register-time
|
||||
* semantics as the system-prompt registries.
|
||||
* 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
|
||||
@@ -370,27 +219,22 @@ export class SubagentService extends Service {
|
||||
info: SubagentRunInfo | SubagentRunEndInfo | string,
|
||||
parent?: Agent,
|
||||
): void {
|
||||
// Run lifecycle events dispatch in the DELEGATING PARENT's scope (a
|
||||
// parent-scoped listener observes only its own delegations); the
|
||||
// provider-removed registry notification stays unfiltered. The carrier is
|
||||
// args[0] of the dispatch call, exactly as cordis' own emit spells it.
|
||||
const dispatchArgs: unknown[] = parent === undefined
|
||||
? [name, info]
|
||||
: [scopeTarget(this, parent), name, info]
|
||||
for (const callback of this.ctx.events.dispatch('emit', dispatchArgs)) {
|
||||
try {
|
||||
callback(info)
|
||||
const returned: unknown = callback(info)
|
||||
void Promise.resolve(returned).catch((error: unknown) => {
|
||||
this.ctx.logger.warn(`subagent: ${name} listener rejected: ${renderThrown(error)}`)
|
||||
})
|
||||
} catch (error: unknown) {
|
||||
this.ctx.logger.warn(`subagent: ${name} listener threw: ${String(error)}`)
|
||||
this.ctx.logger.warn(`subagent: ${name} listener threw: ${renderThrown(error)}`)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Reject a request that needs a start-time capability the provider lacks.
|
||||
* Each optional request field maps to one {@link SubagentCapabilities} flag;
|
||||
* the first unmet one throws `UNSUPPORTED_CAPABILITY`.
|
||||
*/
|
||||
/** Reject the first requested capability that the provider lacks. */
|
||||
private assertCapabilities(provider: SubagentProvider, request: SubagentStartRequest): void {
|
||||
const needs: { when: boolean; cap: keyof SubagentCapabilities }[] = [
|
||||
{ when: request.outputSchema !== undefined, cap: 'outputSchema' },
|
||||
@@ -409,4 +253,13 @@ export class SubagentService extends Service {
|
||||
}
|
||||
}
|
||||
|
||||
/** Render any listener-thrown value without letting coercion escape containment. */
|
||||
function renderThrown(value: unknown): string {
|
||||
try {
|
||||
return value instanceof Error ? `${value.name}: ${value.message}` : String(value)
|
||||
} catch {
|
||||
return '<unrenderable thrown value>'
|
||||
}
|
||||
}
|
||||
|
||||
export default SubagentService
|
||||
|
||||
@@ -8,7 +8,7 @@
|
||||
|
||||
import type { Agent, AgentId, AgentOptions } from '@deepseek-ai/dsh-agent'
|
||||
import type { ContentBlock } from '@deepseek-ai/dsh-llm'
|
||||
import type { StructuredOutputSchema } from '@deepseek-ai/dsh-tools'
|
||||
import type { StructuredOutputSchema, ToolRestriction } from '@deepseek-ai/dsh-tools'
|
||||
|
||||
/**
|
||||
* Which START-TIME features a provider supports. Checked by the service
|
||||
@@ -24,13 +24,13 @@ import type { StructuredOutputSchema } from '@deepseek-ai/dsh-tools'
|
||||
*/
|
||||
export interface SubagentCapabilities {
|
||||
/** Honor {@link SubagentStartRequest.outputSchema} (structured final output). */
|
||||
outputSchema: boolean
|
||||
readonly outputSchema: boolean
|
||||
/** Enforce {@link SubagentStartRequest.maxDepth} (recursion cap). */
|
||||
depthLimit: boolean
|
||||
readonly depthLimit: boolean
|
||||
/** Enforce {@link SubagentStartRequest.toolFilter} (child tool scoping). */
|
||||
toolFilter: boolean
|
||||
readonly toolFilter: boolean
|
||||
/** Honor {@link SubagentStartRequest.persona} (a per-child persona). */
|
||||
persona: boolean
|
||||
readonly persona: boolean
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -41,22 +41,24 @@ export interface SubagentCapabilities {
|
||||
*/
|
||||
export interface SubagentStartRequest {
|
||||
/** The task/prompt for the child agent (a user message in the child session). */
|
||||
prompt: ContentBlock[]
|
||||
readonly prompt: ContentBlock[]
|
||||
/**
|
||||
* The spawning ("parent") agent — the one whose tool call started this
|
||||
* subagent. REQUIRED: in-process backends read `parent.session.header` for
|
||||
* the working directory, the `parentSession` lineage to stamp on the child,
|
||||
* and the parent's delegation depth. Out-of-process backends (ACP) ignore it.
|
||||
*/
|
||||
parent: Agent
|
||||
readonly parent: Agent
|
||||
/**
|
||||
* Cancellation signal from the spawning context (the tool's `exec.signal`).
|
||||
* A provider that honors it aborts the child when the signal fires; the
|
||||
* consumer also bridges it to {@link SubagentRun.cancel} explicitly.
|
||||
* This is the canonical cancellation channel both before and after startup:
|
||||
* a provider rejects `start()` after cleaning partial resources when it
|
||||
* fires before publication, and cancels a published child when it fires
|
||||
* afterward.
|
||||
*/
|
||||
signal?: AbortSignal
|
||||
readonly signal: AbortSignal
|
||||
/** Per-child agent options (model, system prompt). */
|
||||
agentOptions?: AgentOptions
|
||||
readonly agentOptions?: AgentOptions
|
||||
/**
|
||||
* Optional structured-output schema — an object-rooted JSON Schema within the
|
||||
* enforced subset (see `assertSupportedOutputSchema` in dsh-tools; a schema
|
||||
@@ -67,12 +69,14 @@ export interface SubagentStartRequest {
|
||||
* data — a caller holding foreign-realm data materializes it first.
|
||||
* Requesting it against a provider that lacks the capability is rejected at start.
|
||||
*/
|
||||
outputSchema?: StructuredOutputSchema
|
||||
readonly outputSchema?: StructuredOutputSchema
|
||||
/**
|
||||
* Optional recursion cap (max delegation depth below this child). Requires
|
||||
* {@link SubagentCapabilities.depthLimit}; rejected at start otherwise.
|
||||
* Optional absolute delegation-depth cap for the child being started: its
|
||||
* computed depth must be less than or equal to this non-negative safe
|
||||
* integer. Requires {@link SubagentCapabilities.depthLimit}; rejected at
|
||||
* start otherwise.
|
||||
*/
|
||||
maxDepth?: number
|
||||
readonly maxDepth?: number
|
||||
/**
|
||||
* Optional child tool scoping. Requires {@link SubagentCapabilities.toolFilter};
|
||||
* rejected at start otherwise. In-process backends apply it as a scoped
|
||||
@@ -80,7 +84,7 @@ export interface SubagentStartRequest {
|
||||
* from the child's prompt AND refuse to execute (one visibility), with loud
|
||||
* unknown-name validation.
|
||||
*/
|
||||
toolFilter?: { allow?: string[]; deny?: string[] }
|
||||
readonly toolFilter?: ToolRestriction
|
||||
/**
|
||||
* Optional per-child persona. Requires {@link SubagentCapabilities.persona};
|
||||
* rejected at start otherwise. In-process backends register it as a scoped
|
||||
@@ -88,7 +92,7 @@ export interface SubagentStartRequest {
|
||||
* persona for this child alone — same template semantics as the deployment
|
||||
* persona (strict `{{…}}` interpolation against the registered variables).
|
||||
*/
|
||||
persona?: string
|
||||
readonly persona?: string
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -100,7 +104,7 @@ export interface SubagentStartRequest {
|
||||
export interface SubagentStopReasonMap {
|
||||
/** The child finished its turn normally. */
|
||||
completed: 'completed'
|
||||
/** The run was cancelled (parent signal, explicit `cancel()`, or peer cancel). */
|
||||
/** The run was cancelled by its request signal or by disposal. */
|
||||
aborted: 'aborted'
|
||||
/** The child failed (model error, transport error). */
|
||||
error: 'error'
|
||||
@@ -118,7 +122,7 @@ export type SubagentStopReason = SubagentStopReasonMap[keyof SubagentStopReasonM
|
||||
*/
|
||||
export interface SubagentResult {
|
||||
/** The child's final assistant output (the last assistant message's content). */
|
||||
output: ContentBlock[]
|
||||
readonly output: ContentBlock[]
|
||||
/**
|
||||
* The structured result after a requested `outputSchema` was successfully
|
||||
* satisfied. Requesting a schema does not guarantee presence: a provider can
|
||||
@@ -126,32 +130,24 @@ export interface SubagentResult {
|
||||
* valid capture. Shape is validated against the request schema by the
|
||||
* provider; `unknown` here because the seam is schema-agnostic.
|
||||
*/
|
||||
structured?: unknown
|
||||
readonly structured?: unknown
|
||||
/** Why the run ended. A non-`completed` reason means `output` may be partial. */
|
||||
stopReason: SubagentStopReason
|
||||
readonly stopReason: SubagentStopReason
|
||||
}
|
||||
|
||||
/**
|
||||
* A live subagent run: a handle the consumer holds while a child executes.
|
||||
* Returned by {@link SubagentProvider.start} (via the service). The consumer
|
||||
* awaits {@link result}, may {@link cancel} mid-flight, and MUST {@link dispose}
|
||||
* on every path to reach child quiescence (no leaked idle child / session).
|
||||
* Returned by {@link SubagentProvider.start} (via the service) only after the
|
||||
* child is ready. The consumer awaits {@link result} and MUST {@link dispose}
|
||||
* on every path to cancel any remaining work and reach child quiescence.
|
||||
*
|
||||
* {@link sendMessage} and {@link resume} are OPTIONAL: a provider that supports
|
||||
* the runtime capability defines the method; one that doesn't omits it. The
|
||||
* presence of the method IS the capability — narrow before calling.
|
||||
*/
|
||||
export interface SubagentRun {
|
||||
/** The child agent's id (local in-process runs publish it in `ctx.agents`; remote transports need not). */
|
||||
/** The child agent's id (local in-process runs are already published in `ctx.agents`; remote transports need not publish locally). */
|
||||
readonly id: AgentId
|
||||
/**
|
||||
* The provider's publication/readiness boundary. Resolves only after a real
|
||||
* child is established: an in-process agent is live in `ctx.agents`, or a
|
||||
* remote transport has created its child session. Rejects when the attempt
|
||||
* fails or is cancelled before that boundary. The service emits the paired
|
||||
* `subagent/start`/`subagent/end` lifecycle only after this fulfills.
|
||||
*/
|
||||
readonly started: Promise<void>
|
||||
/**
|
||||
* Resolves with the child's terminal {@link SubagentResult} when the run
|
||||
* settles. Does NOT reject on a child-level failure — a model/transport
|
||||
@@ -160,12 +156,10 @@ export interface SubagentRun {
|
||||
* cannot represent as a stop reason.
|
||||
*/
|
||||
readonly result: Promise<SubagentResult>
|
||||
/** Request cancellation of the in-flight run; {@link result} settles `aborted`. */
|
||||
cancel(reason?: string): void
|
||||
/**
|
||||
* Reach child quiescence and release the run's resources (in-process: dispose
|
||||
* the owned agent handle and remove its session; ACP: kill the subprocess).
|
||||
* Idempotent; awaits the child actually stopping, not merely requesting it.
|
||||
* Cancel remaining work, reach child quiescence, and release the run's
|
||||
* resources (in-process: dispose the owned agent and remove its session;
|
||||
* ACP: kill and reap the subprocess). Idempotent.
|
||||
*/
|
||||
dispose(): Promise<void>
|
||||
/**
|
||||
@@ -177,7 +171,7 @@ export interface SubagentRun {
|
||||
* OPTIONAL (resume capability): send a follow-up task to a settled child,
|
||||
* continuing its session, and return a fresh run for the continuation.
|
||||
*/
|
||||
resume?(content: ContentBlock[]): SubagentRun
|
||||
resume?(content: ContentBlock[]): Promise<SubagentRun>
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -185,8 +179,8 @@ export interface SubagentRun {
|
||||
* spawn/fork, ACP to another process, …). Implementations register under a
|
||||
* unique name via {@link SubagentService.registerProvider}; multiple providers
|
||||
* coexist in one context (unlike the single-implementation bash seam). The
|
||||
* service freezes the public descriptor and callback identity at registration;
|
||||
* the captured `start` remains bound to the original provider receiver.
|
||||
* Providers are trusted same-process implementations; callers treat their
|
||||
* descriptors and returned values as borrowed immutable data.
|
||||
*/
|
||||
export interface SubagentProvider {
|
||||
/** Unique registry name (e.g. `spawn`, `fork`, `acp`). */
|
||||
@@ -194,22 +188,24 @@ export interface SubagentProvider {
|
||||
/** The start-time features this provider supports (see {@link SubagentCapabilities}). */
|
||||
readonly capabilities: SubagentCapabilities
|
||||
/**
|
||||
* The provider's context contract: `true` when a child SEES the parent
|
||||
* The provider's conversation-history descriptor: `true` when a child SEES the parent
|
||||
* conversation (fork — the child is seeded with the parent's completed-turn
|
||||
* prefix), `false` when it starts fresh (spawn, ACP). A DESCRIPTIVE fact,
|
||||
* not a start-time capability: the service validates nothing against it —
|
||||
* the model-facing consumer (`dsh-tool-subagent`) derives truthful tool
|
||||
* wording from it, so a tool bound to a fork provider stops telling the
|
||||
* model the child "does not see this conversation".
|
||||
* model the child "does not see this conversation". This descriptor concerns
|
||||
* conversation history only; it says nothing about tool registrations,
|
||||
* injected services, or authority inheritance.
|
||||
*/
|
||||
readonly inheritsParentContext: boolean
|
||||
/**
|
||||
* Start preparing a child run and return its handle synchronously. The
|
||||
* 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. The returned {@link SubagentRun.started} must mark
|
||||
* the real publication/readiness boundary; the result path must observe that
|
||||
* promise immediately so a pre-start rejection cannot become unhandled.
|
||||
* 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: SubagentStartRequest): SubagentRun
|
||||
start(request: SubagentStartRequest): Promise<SubagentRun>
|
||||
}
|
||||
|
||||
@@ -5,6 +5,7 @@ import { HarnessError } from '@deepseek-ai/dsh-llm'
|
||||
import { carrierKeyOf } from '@deepseek-ai/dsh-scope'
|
||||
import SubagentService, {
|
||||
SubagentError,
|
||||
assertSubagentMaxDepth,
|
||||
type SubagentCapabilities,
|
||||
type SubagentProvider,
|
||||
type SubagentResult,
|
||||
@@ -12,580 +13,218 @@ import SubagentService, {
|
||||
type SubagentStartRequest,
|
||||
} from '@deepseek-ai/dsh-subagent'
|
||||
|
||||
/** A minimal parent Agent stand-in — the service only reads `parent.id`. */
|
||||
function fakeParent(id = 'parent-1'): Agent {
|
||||
return { id: AgentId(id) } as unknown as Agent
|
||||
}
|
||||
|
||||
const ALL_CAPS: SubagentCapabilities = { outputSchema: true, depthLimit: true, toolFilter: true, persona: false }
|
||||
const ALL_CAPS: SubagentCapabilities = { outputSchema: true, depthLimit: true, toolFilter: true, persona: true }
|
||||
const NO_CAPS: SubagentCapabilities = { outputSchema: false, depthLimit: false, toolFilter: false, persona: false }
|
||||
|
||||
/** A scripted provider whose run settles immediately with a fixed result. */
|
||||
function baseRequest(overrides: Partial<SubagentStartRequest> = {}): SubagentStartRequest {
|
||||
return {
|
||||
prompt: [{ type: 'text', text: 'do a thing' }],
|
||||
parent: fakeParent(),
|
||||
signal: new AbortController().signal,
|
||||
...overrides,
|
||||
}
|
||||
}
|
||||
|
||||
class StubProvider implements SubagentProvider {
|
||||
startCount = 0
|
||||
readonly inheritsParentContext = false
|
||||
startCount = 0
|
||||
|
||||
constructor(
|
||||
readonly name: string,
|
||||
readonly capabilities: SubagentCapabilities = ALL_CAPS,
|
||||
private readonly result: SubagentResult = { output: [{ type: 'text', text: 'ok' }], stopReason: 'completed' },
|
||||
private readonly outcome: SubagentResult = {
|
||||
output: [{ type: 'text', text: 'ok' }],
|
||||
stopReason: 'completed',
|
||||
},
|
||||
) {}
|
||||
|
||||
start(request: SubagentStartRequest): SubagentRun {
|
||||
this.startCount++
|
||||
async start(request: SubagentStartRequest): Promise<SubagentRun> {
|
||||
this.startCount += 1
|
||||
return {
|
||||
id: AgentId(`child:${this.name}:${request.parent.id}`),
|
||||
started: Promise.resolve(),
|
||||
result: Promise.resolve(this.result),
|
||||
cancel() {},
|
||||
result: Promise.resolve(this.outcome),
|
||||
async dispose() {},
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
function baseRequest(overrides: Partial<SubagentStartRequest> = {}): SubagentStartRequest {
|
||||
return { prompt: [{ type: 'text', text: 'do a thing' }], parent: fakeParent(), ...overrides }
|
||||
async function service(): Promise<{ ctx: Context; subagents: SubagentService }> {
|
||||
const ctx = new Context()
|
||||
await ctx.plugin(SubagentService)
|
||||
return { ctx, subagents: ctx.subagents }
|
||||
}
|
||||
|
||||
describe('SubagentService', () => {
|
||||
it('announces provider lifecycle: added on register, removed on dispose', async () => {
|
||||
const ctx = new Context()
|
||||
await ctx.plugin(SubagentService)
|
||||
it('registers, lists, looks up, starts, and removes providers', async () => {
|
||||
const { ctx, subagents } = await service()
|
||||
const added: string[] = []
|
||||
const removed: string[] = []
|
||||
ctx.on('subagent/provider-added', provider => void added.push(provider.name))
|
||||
ctx.on('subagent/provider-removed', name => void removed.push(name))
|
||||
|
||||
const dispose = ctx.subagents.registerProvider(new StubProvider('alpha'))
|
||||
expect(added).toEqual(['alpha'])
|
||||
expect(removed).toEqual([])
|
||||
|
||||
await dispose()
|
||||
expect(removed).toEqual(['alpha'])
|
||||
})
|
||||
|
||||
it('rolls back the registration when a provider-added listener throws', async () => {
|
||||
const ctx = new Context()
|
||||
await ctx.plugin(SubagentService)
|
||||
let threw = false
|
||||
const off = ctx.on('subagent/provider-added', () => {
|
||||
if (!threw) { threw = true; throw new Error('boom added listener') }
|
||||
})
|
||||
|
||||
expect(() => ctx.subagents.registerProvider(new StubProvider('alpha'))).toThrow('boom added listener')
|
||||
expect(ctx.subagents.getProvider('alpha')).toBeUndefined() // nothing leaked
|
||||
|
||||
off()
|
||||
ctx.subagents.registerProvider(new StubProvider('alpha'))
|
||||
expect(ctx.subagents.getProvider('alpha')).toBeDefined()
|
||||
})
|
||||
|
||||
it('contains a throwing provider-removed listener: later mirrors still hear it, teardown completes', async () => {
|
||||
// provider-removed fires inside the registration's DISPOSER, so a
|
||||
// propagating listener would disrupt the backend's teardown; and cordis
|
||||
// emit halts on the first throw, so an uncontained one would starve every
|
||||
// mirror registered after it (a stale model-facing tool). Both are
|
||||
// prevented by per-listener containment.
|
||||
const ctx = new Context()
|
||||
await ctx.plugin(SubagentService)
|
||||
const warnings: string[] = []
|
||||
ctx.logger.warn = ((message: unknown) => void warnings.push(String(message))) as typeof ctx.logger.warn
|
||||
ctx.on('subagent/provider-removed', () => { throw new Error('boom removed listener') })
|
||||
const heard: string[] = []
|
||||
ctx.on('subagent/provider-removed', name => void heard.push(name))
|
||||
|
||||
const dispose = ctx.subagents.registerProvider(new StubProvider('alpha'))
|
||||
expect(() => void dispose()).not.toThrow()
|
||||
expect(heard).toEqual(['alpha']) // the listener AFTER the thrower still ran
|
||||
expect(ctx.subagents.getProvider('alpha')).toBeUndefined() // teardown reached quiescence
|
||||
expect(warnings.some(w => w.includes('boom removed listener'))).toBe(true)
|
||||
})
|
||||
|
||||
it('registers a provider and starts a run on it by name', async () => {
|
||||
const ctx = new Context()
|
||||
await ctx.plugin(SubagentService)
|
||||
const provider = new StubProvider('alpha')
|
||||
ctx.subagents.registerProvider(provider)
|
||||
|
||||
expect(ctx.subagents.list()).toEqual(['alpha'])
|
||||
expect(ctx.subagents.getProvider('alpha')).toMatchObject({ name: 'alpha' })
|
||||
|
||||
const run = ctx.subagents.start('alpha', baseRequest())
|
||||
expect(provider.startCount).toBe(1)
|
||||
await expect(run.result).resolves.toMatchObject({ stopReason: 'completed' })
|
||||
})
|
||||
|
||||
it('lets multiple providers coexist (the defining requirement)', async () => {
|
||||
const ctx = new Context()
|
||||
await ctx.plugin(SubagentService)
|
||||
ctx.subagents.registerProvider(new StubProvider('spawn'))
|
||||
ctx.subagents.registerProvider(new StubProvider('acp'))
|
||||
|
||||
expect(ctx.subagents.list()).toEqual(['spawn', 'acp'])
|
||||
expect(ctx.subagents.getProvider('spawn')).toBeDefined()
|
||||
expect(ctx.subagents.getProvider('acp')).toBeDefined()
|
||||
})
|
||||
|
||||
it('throws NO_PROVIDER when starting on an unregistered name', async () => {
|
||||
const ctx = new Context()
|
||||
await ctx.plugin(SubagentService)
|
||||
try {
|
||||
ctx.subagents.start('missing', baseRequest())
|
||||
expect.fail('expected NO_PROVIDER')
|
||||
} catch (error: unknown) {
|
||||
expect(error).toBeInstanceOf(SubagentError)
|
||||
expect((error as SubagentError).code).toBe('NO_PROVIDER')
|
||||
}
|
||||
})
|
||||
|
||||
it('rejects duplicate provider names with DUPLICATE_PROVIDER', async () => {
|
||||
const ctx = new Context()
|
||||
await ctx.plugin(SubagentService)
|
||||
ctx.subagents.registerProvider(new StubProvider('dup'))
|
||||
try {
|
||||
ctx.subagents.registerProvider(new StubProvider('dup'))
|
||||
expect.fail('expected DUPLICATE_PROVIDER')
|
||||
} catch (error: unknown) {
|
||||
expect(error).toBeInstanceOf(SubagentError)
|
||||
expect((error as SubagentError).code).toBe('DUPLICATE_PROVIDER')
|
||||
}
|
||||
})
|
||||
|
||||
it('unregisters a provider when its owning fiber is disposed (HMR safety)', async () => {
|
||||
const ctx = new Context()
|
||||
await ctx.plugin(SubagentService)
|
||||
|
||||
const fiber = await ctx.plugin(Object.assign((inner: Context) => {
|
||||
inner.subagents.registerProvider(new StubProvider('scoped'))
|
||||
}, { inject: ['subagents'] }))
|
||||
expect(ctx.subagents.list()).toEqual(['scoped'])
|
||||
|
||||
await fiber.dispose()
|
||||
expect(ctx.subagents.list()).toEqual([])
|
||||
})
|
||||
|
||||
it('snapshots a provider registration so caller mutation cannot corrupt dispatch or HMR cleanup', async () => {
|
||||
const ctx = new Context()
|
||||
await ctx.plugin(SubagentService)
|
||||
const capabilities: SubagentCapabilities = {
|
||||
outputSchema: true,
|
||||
depthLimit: true,
|
||||
toolFilter: true,
|
||||
persona: true,
|
||||
}
|
||||
const provider = new StubProvider('stable', capabilities)
|
||||
const added: SubagentProvider[] = []
|
||||
const removed: string[] = []
|
||||
ctx.on('subagent/provider-added', registered => void added.push(registered))
|
||||
ctx.on('subagent/provider-removed', name => void removed.push(name))
|
||||
const owner = await ctx.plugin({
|
||||
name: 'mutable-provider-owner',
|
||||
inject: ['subagents'],
|
||||
apply(pluginCtx: Context) {
|
||||
pluginCtx.subagents.registerProvider(provider)
|
||||
},
|
||||
})
|
||||
const accepted = ctx.subagents.getProvider('stable')
|
||||
|
||||
const mutable = provider as unknown as {
|
||||
name: string
|
||||
capabilities: SubagentCapabilities
|
||||
inheritsParentContext: boolean
|
||||
start: SubagentProvider['start']
|
||||
}
|
||||
mutable.name = 'mutated'
|
||||
capabilities.outputSchema = false
|
||||
capabilities.depthLimit = false
|
||||
capabilities.toolFilter = false
|
||||
capabilities.persona = false
|
||||
mutable.capabilities = NO_CAPS
|
||||
mutable.inheritsParentContext = true
|
||||
const replacementStart = vi.fn((_request: SubagentStartRequest): SubagentRun => {
|
||||
throw new Error('replacement start must not run')
|
||||
})
|
||||
mutable.start = replacementStart
|
||||
|
||||
expect(added).toEqual([accepted])
|
||||
expect(accepted).not.toBe(provider)
|
||||
expect(Object.isFrozen(accepted)).toBe(true)
|
||||
expect(Object.isFrozen(accepted?.capabilities)).toBe(true)
|
||||
expect(accepted).toMatchObject({
|
||||
name: 'stable',
|
||||
capabilities: { outputSchema: true, depthLimit: true, toolFilter: true, persona: true },
|
||||
inheritsParentContext: false,
|
||||
})
|
||||
expect(ctx.subagents.list()).toEqual(['stable'])
|
||||
expect(ctx.subagents.getProvider('mutated')).toBeUndefined()
|
||||
|
||||
const run = ctx.subagents.start('stable', baseRequest({
|
||||
outputSchema: { type: 'object', properties: { answer: { type: 'string' } } },
|
||||
maxDepth: 2,
|
||||
toolFilter: { deny: ['bash'] },
|
||||
persona: 'reviewer',
|
||||
}))
|
||||
const dispose = subagents.registerProvider(provider)
|
||||
expect(subagents.list()).toEqual(['alpha'])
|
||||
expect(subagents.getProvider('alpha')).toBe(provider)
|
||||
const run = await subagents.start('alpha', baseRequest())
|
||||
await expect(run.result).resolves.toMatchObject({ stopReason: 'completed' })
|
||||
expect(provider.startCount).toBe(1)
|
||||
expect(replacementStart).not.toHaveBeenCalled()
|
||||
|
||||
await owner.dispose()
|
||||
expect(removed).toEqual(['stable'])
|
||||
expect(ctx.subagents.list()).toEqual([])
|
||||
expect(() => ctx.subagents.registerProvider(new StubProvider('stable'))).not.toThrow()
|
||||
})
|
||||
|
||||
it('re-registers a name after its prior registration is disposed (not wedged)', async () => {
|
||||
const ctx = new Context()
|
||||
await ctx.plugin(SubagentService)
|
||||
|
||||
const dispose = ctx.subagents.registerProvider(new StubProvider('reuse'))
|
||||
expect(ctx.subagents.list()).toEqual(['reuse'])
|
||||
await dispose()
|
||||
expect(ctx.subagents.list()).toEqual([])
|
||||
|
||||
const disposeAgain = ctx.subagents.registerProvider(new StubProvider('reuse'))
|
||||
expect(ctx.subagents.list()).toEqual(['reuse'])
|
||||
await disposeAgain()
|
||||
expect(ctx.subagents.list()).toEqual([])
|
||||
expect(added).toEqual(['alpha'])
|
||||
expect(removed).toEqual(['alpha'])
|
||||
expect(subagents.getProvider('alpha')).toBeUndefined()
|
||||
})
|
||||
|
||||
describe('start-time capability validation (fail loud, before any child)', () => {
|
||||
it.each([
|
||||
{ field: 'outputSchema', request: baseRequest({ outputSchema: { type: 'object', properties: { x: { type: 'string' } } } }) },
|
||||
{ field: 'maxDepth', request: baseRequest({ maxDepth: 2 }) },
|
||||
{ field: 'toolFilter', request: baseRequest({ toolFilter: { deny: ['bash'] } }) },
|
||||
])('rejects $field against a provider that lacks the capability — before start() runs', ({ request }) => {
|
||||
const ctx = new Context()
|
||||
return ctx.plugin(SubagentService).then(() => {
|
||||
const provider = new StubProvider('weak', NO_CAPS)
|
||||
ctx.subagents.registerProvider(provider)
|
||||
try {
|
||||
ctx.subagents.start('weak', request)
|
||||
expect.fail('expected UNSUPPORTED_CAPABILITY')
|
||||
} catch (error: unknown) {
|
||||
expect(error).toBeInstanceOf(SubagentError)
|
||||
expect((error as SubagentError).code).toBe('UNSUPPORTED_CAPABILITY')
|
||||
}
|
||||
// The child was never started — the check is pre-spawn.
|
||||
expect(provider.startCount).toBe(0)
|
||||
})
|
||||
})
|
||||
|
||||
it('allows a capability request when the provider supports it', async () => {
|
||||
const ctx = new Context()
|
||||
await ctx.plugin(SubagentService)
|
||||
const provider = new StubProvider('strong', ALL_CAPS)
|
||||
ctx.subagents.registerProvider(provider)
|
||||
ctx.subagents.start('strong', baseRequest({ outputSchema: { type: 'object', properties: { x: { type: 'string' } } }, maxDepth: 1 }))
|
||||
expect(provider.startCount).toBe(1)
|
||||
})
|
||||
it('rolls registration back when provider-added throws', async () => {
|
||||
const { ctx, subagents } = await service()
|
||||
ctx.on('subagent/provider-added', () => { throw new Error('added boom') })
|
||||
expect(() => { subagents.registerProvider(new StubProvider('alpha')) }).toThrow('added boom')
|
||||
expect(subagents.getProvider('alpha')).toBeUndefined()
|
||||
})
|
||||
|
||||
it('emits subagent/start then subagent/end around a run', async () => {
|
||||
const ctx = new Context()
|
||||
await ctx.plugin(SubagentService)
|
||||
ctx.subagents.registerProvider(new StubProvider('events'))
|
||||
|
||||
const started = vi.fn()
|
||||
const ended = vi.fn()
|
||||
ctx.on('subagent/start', started)
|
||||
ctx.on('subagent/end', ended)
|
||||
|
||||
const run = ctx.subagents.start('events', baseRequest())
|
||||
await run.started
|
||||
expect(started).toHaveBeenCalledWith(expect.objectContaining({ provider: 'events', id: run.id }))
|
||||
|
||||
await run.result
|
||||
// `subagent/end` fires from a `.then` on the result — let the microtask run.
|
||||
await Promise.resolve()
|
||||
expect(ended).toHaveBeenCalledWith(expect.objectContaining({ provider: 'events', id: run.id, stopReason: 'completed' }))
|
||||
it('rejects duplicate and absent provider names with typed errors', async () => {
|
||||
const { subagents } = await service()
|
||||
subagents.registerProvider(new StubProvider('dup'))
|
||||
expect(() => { subagents.registerProvider(new StubProvider('dup')) })
|
||||
.toThrow(expect.objectContaining({ code: 'DUPLICATE_PROVIDER' }))
|
||||
await expect(subagents.start('missing', baseRequest()))
|
||||
.rejects.toMatchObject({ code: 'NO_PROVIDER' })
|
||||
})
|
||||
|
||||
it('waits for provider readiness and observes an early result rejection without reordering lifecycle', async () => {
|
||||
const ctx = new Context()
|
||||
await ctx.plugin(SubagentService)
|
||||
const readiness = Promise.withResolvers<undefined>()
|
||||
ctx.subagents.registerProvider({
|
||||
name: 'delayed-start',
|
||||
capabilities: NO_CAPS,
|
||||
inheritsParentContext: false,
|
||||
start: () => ({
|
||||
id: AgentId('delayed-child'),
|
||||
started: readiness.promise,
|
||||
// Already rejected: SubagentService must attach its result handler in
|
||||
// the same synchronous start() call, before awaiting readiness.
|
||||
result: Promise.reject(new Error('early infrastructure fault')),
|
||||
cancel() {},
|
||||
async dispose() {},
|
||||
}),
|
||||
})
|
||||
const lifecycle: string[] = []
|
||||
ctx.on('subagent/start', () => void lifecycle.push('start'))
|
||||
ctx.on('subagent/end', info => void lifecycle.push(`end:${info.stopReason}`))
|
||||
|
||||
const run = ctx.subagents.start('delayed-start', baseRequest())
|
||||
await expect(run.result).rejects.toThrow('early infrastructure fault')
|
||||
expect(lifecycle).toEqual([])
|
||||
|
||||
readiness.resolve(undefined)
|
||||
await run.started
|
||||
expect(lifecycle).toEqual(['start', 'end:error'])
|
||||
it.each([
|
||||
['outputSchema', { outputSchema: { type: 'object', properties: {} } }],
|
||||
['depthLimit', { maxDepth: 1 }],
|
||||
['toolFilter', { toolFilter: { deny: ['bash'] } }],
|
||||
['persona', { persona: 'reviewer' }],
|
||||
] as const)('rejects unsupported %s before provider startup', async (_capability, override) => {
|
||||
const { subagents } = await service()
|
||||
const provider = new StubProvider('weak', NO_CAPS)
|
||||
subagents.registerProvider(provider)
|
||||
await expect(subagents.start('weak', baseRequest(override)))
|
||||
.rejects.toMatchObject({ code: 'UNSUPPORTED_CAPABILITY' })
|
||||
expect(provider.startCount).toBe(0)
|
||||
})
|
||||
|
||||
it('emits no lifecycle pair when readiness rejects before a child exists', async () => {
|
||||
const ctx = new Context()
|
||||
await ctx.plugin(SubagentService)
|
||||
const readiness = Promise.withResolvers<undefined>()
|
||||
it('validates depth and schema semantics before provider startup', async () => {
|
||||
const { subagents } = await service()
|
||||
const provider = new StubProvider('strong')
|
||||
subagents.registerProvider(provider)
|
||||
await expect(subagents.start('strong', baseRequest({ maxDepth: -1 })))
|
||||
.rejects.toThrow('non-negative safe integer')
|
||||
await expect(subagents.start('strong', baseRequest({ outputSchema: { type: 'string' } as never })))
|
||||
.rejects.toThrow()
|
||||
expect(provider.startCount).toBe(0)
|
||||
expect(() => { assertSubagentMaxDepth(undefined) }).not.toThrow()
|
||||
})
|
||||
|
||||
it('publishes lifecycle only after async provider start and keeps parent scope', async () => {
|
||||
const { ctx, subagents } = await service()
|
||||
const ready = Promise.withResolvers<SubagentRun>()
|
||||
const result = Promise.withResolvers<SubagentResult>()
|
||||
ctx.subagents.registerProvider({
|
||||
name: 'never-started',
|
||||
subagents.registerProvider({
|
||||
name: 'deferred',
|
||||
capabilities: NO_CAPS,
|
||||
inheritsParentContext: false,
|
||||
start: () => ({
|
||||
id: AgentId('never-started-child'),
|
||||
started: readiness.promise,
|
||||
result: result.promise,
|
||||
cancel() {},
|
||||
async dispose() {},
|
||||
}),
|
||||
start: () => ready.promise,
|
||||
})
|
||||
const parent = fakeParent('delegator')
|
||||
const events: string[] = []
|
||||
const keys: unknown[] = []
|
||||
ctx.on('subagent/start', function () { events.push('start'); keys.push(carrierKeyOf(this)) })
|
||||
ctx.on('subagent/end', function () { events.push('end'); keys.push(carrierKeyOf(this)) })
|
||||
|
||||
const starting = subagents.start('deferred', baseRequest({ parent }))
|
||||
await Promise.resolve()
|
||||
expect(events).toEqual([])
|
||||
ready.resolve({ id: AgentId('child'), result: result.promise, async dispose() {} })
|
||||
const run = await starting
|
||||
expect(events).toEqual(['start'])
|
||||
result.resolve({ output: [{ type: 'text', text: 'answer' }], stopReason: 'completed' })
|
||||
await run.result
|
||||
await Promise.resolve()
|
||||
expect(events).toEqual(['start', 'end'])
|
||||
expect(keys).toEqual([parent, parent])
|
||||
})
|
||||
|
||||
it('emits no run lifecycle when provider startup rejects', async () => {
|
||||
const { ctx, subagents } = await service()
|
||||
subagents.registerProvider({
|
||||
name: 'failed',
|
||||
capabilities: NO_CAPS,
|
||||
inheritsParentContext: false,
|
||||
start: async () => { throw new Error('setup rolled back') },
|
||||
})
|
||||
const lifecycle = vi.fn()
|
||||
ctx.on('subagent/start', lifecycle)
|
||||
ctx.on('subagent/end', lifecycle)
|
||||
|
||||
const run = ctx.subagents.start('never-started', baseRequest())
|
||||
readiness.reject(new Error('publication rolled back'))
|
||||
await expect(run.started).rejects.toThrow('publication rolled back')
|
||||
result.resolve({ output: [], stopReason: 'aborted' })
|
||||
await run.result
|
||||
await Promise.resolve()
|
||||
await expect(subagents.start('failed', baseRequest())).rejects.toThrow('setup rolled back')
|
||||
expect(lifecycle).not.toHaveBeenCalled()
|
||||
})
|
||||
|
||||
it('pins start and end to the parent accepted at start despite caller mutation', async () => {
|
||||
const ctx = new Context()
|
||||
await ctx.plugin(SubagentService)
|
||||
const gate = Promise.withResolvers<SubagentResult>()
|
||||
let acceptedRequest: SubagentStartRequest | undefined
|
||||
ctx.subagents.registerProvider({
|
||||
name: 'deferred',
|
||||
capabilities: NO_CAPS,
|
||||
inheritsParentContext: false,
|
||||
start: (accepted) => {
|
||||
acceptedRequest = accepted
|
||||
return {
|
||||
id: AgentId('deferred-child'),
|
||||
started: Promise.resolve(),
|
||||
result: gate.promise,
|
||||
cancel() {},
|
||||
async dispose() {},
|
||||
}
|
||||
},
|
||||
it('emits an enriched end event and maps result rejection to error telemetry', async () => {
|
||||
const { ctx, subagents } = await service()
|
||||
const completed = new StubProvider('completed', NO_CAPS, {
|
||||
output: [{ type: 'text', text: 'answer' }],
|
||||
stopReason: 'completed',
|
||||
})
|
||||
const accepted = fakeParent('accepted-parent')
|
||||
const replacement = fakeParent('replacement-parent')
|
||||
const keys: unknown[] = []
|
||||
ctx.on('subagent/start', function () { keys.push(carrierKeyOf(this)) })
|
||||
ctx.on('subagent/end', function () { keys.push(carrierKeyOf(this)) })
|
||||
const request = baseRequest({ parent: accepted })
|
||||
|
||||
const run = ctx.subagents.start('deferred', request)
|
||||
request.parent = replacement
|
||||
request.prompt[0] = { type: 'text', text: 'mutated prompt' }
|
||||
expect(acceptedRequest?.parent).toBe(accepted)
|
||||
expect(acceptedRequest?.prompt).toEqual([{ type: 'text', text: 'do a thing' }])
|
||||
expect(acceptedRequest?.prompt).not.toBe(request.prompt)
|
||||
gate.resolve({ output: [], stopReason: 'completed' })
|
||||
await run.result
|
||||
await Promise.resolve()
|
||||
|
||||
expect(keys).toEqual([accepted, accepted])
|
||||
})
|
||||
|
||||
it('carries lastAssistantMessage (the child output) onto the end event', async () => {
|
||||
const ctx = new Context()
|
||||
await ctx.plugin(SubagentService)
|
||||
ctx.subagents.registerProvider(new StubProvider(
|
||||
'enriched',
|
||||
ALL_CAPS,
|
||||
{ output: [{ type: 'text', text: 'the child answer' }], stopReason: 'completed' },
|
||||
))
|
||||
|
||||
const started = vi.fn()
|
||||
subagents.registerProvider(completed)
|
||||
const ended = vi.fn()
|
||||
ctx.on('subagent/start', started)
|
||||
ctx.on('subagent/end', ended)
|
||||
|
||||
const run = ctx.subagents.start('enriched', baseRequest())
|
||||
await run.started
|
||||
expect(started).toHaveBeenCalledWith(expect.objectContaining({ provider: 'enriched', id: run.id }))
|
||||
|
||||
const run = await subagents.start('completed', baseRequest())
|
||||
await run.result
|
||||
await Promise.resolve()
|
||||
expect(ended).toHaveBeenCalledWith(expect.objectContaining({
|
||||
provider: 'enriched',
|
||||
id: run.id,
|
||||
provider: 'completed',
|
||||
lastAssistantMessage: [{ type: 'text', text: 'answer' }],
|
||||
stopReason: 'completed',
|
||||
lastAssistantMessage: [{ type: 'text', text: 'the child answer' }],
|
||||
}))
|
||||
})
|
||||
|
||||
it('observe-only: a subagent/end listener mutating lastAssistantMessage cannot corrupt the caller\'s result', async () => {
|
||||
// The subagent/end emit fires from a detached `.then` registered before
|
||||
// start() returns — i.e. BEFORE the caller's own `await run.result`
|
||||
// continuation. If the event shared the result.output reference, a mutating
|
||||
// listener would change the SubagentResult the caller consumes. The service
|
||||
// deep-clones output onto the event, so the listener mutates only its copy.
|
||||
const ctx = new Context()
|
||||
await ctx.plugin(SubagentService)
|
||||
ctx.subagents.registerProvider(new StubProvider(
|
||||
'clone',
|
||||
ALL_CAPS,
|
||||
{ output: [{ type: 'text', text: 'original' }], stopReason: 'completed' },
|
||||
))
|
||||
|
||||
ctx.on('subagent/end', (info) => {
|
||||
// A hostile/buggy listener reaches in and mutates the event's array.
|
||||
const blocks = info.lastAssistantMessage
|
||||
if (blocks?.[0]?.type === 'text') blocks[0].text = 'HIJACKED'
|
||||
blocks?.push({ type: 'text', text: 'injected' })
|
||||
})
|
||||
|
||||
const run = ctx.subagents.start('clone', baseRequest())
|
||||
const result = await run.result
|
||||
await Promise.resolve() // let the detached settle hook (and its listener) run
|
||||
// The caller's result.output is untouched by the listener's mutation.
|
||||
expect(result.output).toEqual([{ type: 'text', text: 'original' }])
|
||||
})
|
||||
|
||||
it('omits lastAssistantMessage on the reject path (no SubagentResult was produced)', async () => {
|
||||
const ctx = new Context()
|
||||
await ctx.plugin(SubagentService)
|
||||
ctx.subagents.registerProvider({
|
||||
name: 'rej',
|
||||
const failure = Promise.withResolvers<SubagentResult>()
|
||||
subagents.registerProvider({
|
||||
name: 'infra',
|
||||
capabilities: NO_CAPS,
|
||||
inheritsParentContext: false,
|
||||
start: () => ({
|
||||
id: AgentId('rej-child'),
|
||||
started: Promise.resolve(),
|
||||
result: Promise.reject(new Error('infra fault')),
|
||||
cancel() {},
|
||||
dispose: async () => {},
|
||||
}),
|
||||
async start() {
|
||||
return { id: AgentId('infra-child'), result: failure.promise, async dispose() {} }
|
||||
},
|
||||
})
|
||||
|
||||
const ended = vi.fn()
|
||||
ctx.on('subagent/end', ended)
|
||||
const run = ctx.subagents.start('rej', baseRequest())
|
||||
await run.result.catch(() => {})
|
||||
const failedRun = await subagents.start('infra', baseRequest())
|
||||
failure.reject(new Error('transport'))
|
||||
await expect(failedRun.result).rejects.toThrow('transport')
|
||||
await Promise.resolve()
|
||||
|
||||
const endInfo = ended.mock.calls[0]![0] as Record<string, unknown>
|
||||
expect(endInfo.stopReason).toBe('error')
|
||||
expect('lastAssistantMessage' in endInfo).toBe(false) // no output exists on reject
|
||||
expect(ended).toHaveBeenCalledWith(expect.objectContaining({ provider: 'infra', stopReason: 'error' }))
|
||||
})
|
||||
|
||||
it('contains a structuredClone failure: emits subagent/end without lastAssistantMessage (no unhandled rejection)', async () => {
|
||||
// The clone runs inside onFulfilled, OUTSIDE emitLifecycle's per-listener
|
||||
// containment. An uncloneable output (here a content block carrying a
|
||||
// function) would otherwise throw and become an unhandled rejection on the
|
||||
// detached `.then`. The handler must instead log and emit the event WITHOUT
|
||||
// lastAssistantMessage, still carrying the real stopReason.
|
||||
const ctx = new Context()
|
||||
await ctx.plugin(SubagentService)
|
||||
const warn = vi.fn(); ctx.logger.warn = warn as never
|
||||
// An output value structuredClone cannot handle (a function is uncloneable).
|
||||
const uncloneable = [{ type: 'text', text: 'x', evil: () => 0 }] as unknown as SubagentResult['output']
|
||||
ctx.subagents.registerProvider({
|
||||
name: 'unclone',
|
||||
capabilities: NO_CAPS,
|
||||
inheritsParentContext: false,
|
||||
start: () => ({
|
||||
id: AgentId('unclone-child'),
|
||||
started: Promise.resolve(),
|
||||
result: Promise.resolve({ output: uncloneable, stopReason: 'completed' } as SubagentResult),
|
||||
cancel() {},
|
||||
dispose: async () => {},
|
||||
}),
|
||||
})
|
||||
it('contains synchronous and asynchronous lifecycle observer failures', async () => {
|
||||
const { ctx, subagents } = await service()
|
||||
const warnings: string[] = []
|
||||
ctx.logger.warn = ((message: unknown) => void warnings.push(String(message))) as typeof ctx.logger.warn
|
||||
const heard: string[] = []
|
||||
ctx.on('subagent/provider-removed', () => { throw new Error('sync boom') })
|
||||
// Runtime listeners may return thenables even though the declaration's observable result is void.
|
||||
// eslint-disable-next-line @typescript-eslint/no-misused-promises -- exercises rejected-listener containment
|
||||
ctx.on('subagent/provider-removed', async () => { throw new Error('async boom') })
|
||||
ctx.on('subagent/provider-removed', () => { throw { toString: () => { throw new Error('coercion') } } })
|
||||
ctx.on('subagent/provider-removed', name => void heard.push(name))
|
||||
const dispose = subagents.registerProvider(new StubProvider('contained'))
|
||||
|
||||
const ended = vi.fn()
|
||||
ctx.on('subagent/end', ended)
|
||||
const run = ctx.subagents.start('unclone', baseRequest())
|
||||
await run.result
|
||||
await dispose()
|
||||
await Promise.resolve()
|
||||
|
||||
const endInfo = ended.mock.calls[0]![0] as Record<string, unknown>
|
||||
expect(endInfo.stopReason).toBe('completed') // the real outcome is preserved
|
||||
expect('lastAssistantMessage' in endInfo).toBe(false) // clone failed → omitted, not crashed
|
||||
expect(warn).toHaveBeenCalledWith(expect.stringContaining('could not clone'))
|
||||
expect(heard).toEqual(['contained'])
|
||||
expect(warnings.some(message => message.includes('sync boom'))).toBe(true)
|
||||
expect(warnings.some(message => message.includes('async boom'))).toBe(true)
|
||||
expect(warnings.some(message => message.includes('<unrenderable thrown value>'))).toBe(true)
|
||||
})
|
||||
|
||||
it('emits subagent/end with stopReason "error" when the run result promise rejects', async () => {
|
||||
const ctx = new Context()
|
||||
await ctx.plugin(SubagentService)
|
||||
// A provider whose run.result REJECTS (an infrastructure fault — the seam
|
||||
// contract says child-level failures resolve with stopReason 'error', but a
|
||||
// rejection is still surfaced as an 'error' telemetry event).
|
||||
ctx.subagents.registerProvider({
|
||||
name: 'rejecter',
|
||||
capabilities: NO_CAPS,
|
||||
inheritsParentContext: false,
|
||||
start: () => ({
|
||||
id: AgentId('rej-child'),
|
||||
started: Promise.resolve(),
|
||||
result: Promise.reject(new Error('infra fault')),
|
||||
cancel() {},
|
||||
dispose: async () => {},
|
||||
}),
|
||||
})
|
||||
|
||||
const ended = vi.fn()
|
||||
ctx.on('subagent/end', ended)
|
||||
const run = ctx.subagents.start('rejecter', baseRequest())
|
||||
// Observe (and swallow) the rejection the consumer would see, then let the
|
||||
// detached `.then` settle the telemetry emit.
|
||||
await run.result.catch(() => {})
|
||||
await Promise.resolve()
|
||||
expect(ended).toHaveBeenCalledWith(expect.objectContaining({ provider: 'rejecter', id: run.id, stopReason: 'error' }))
|
||||
})
|
||||
|
||||
it('contains a throwing subagent/start listener per-listener: a later listener still observes the event and start() returns the run', async () => {
|
||||
const ctx = new Context()
|
||||
await ctx.plugin(SubagentService)
|
||||
ctx.subagents.registerProvider(new StubProvider('contain'))
|
||||
// Two listeners; the FIRST throws. Per-listener containment means the second
|
||||
// must STILL run (a single try/catch around ctx.emit would let the first
|
||||
// throw halt the dispatch and starve the second — the round-2 regression).
|
||||
const second = vi.fn()
|
||||
ctx.on('subagent/start', () => { throw new Error('bad start listener') })
|
||||
ctx.on('subagent/start', second)
|
||||
|
||||
const run = ctx.subagents.start('contain', baseRequest())
|
||||
expect(run.id).toBeDefined()
|
||||
await run.started
|
||||
expect(second).toHaveBeenCalledWith(expect.objectContaining({ provider: 'contain', id: run.id }))
|
||||
await expect(run.result).resolves.toMatchObject({ stopReason: 'completed' })
|
||||
})
|
||||
|
||||
it('contains a throwing subagent/end listener per-listener: a later listener still observes the settle, no unhandled rejection', async () => {
|
||||
const ctx = new Context()
|
||||
await ctx.plugin(SubagentService)
|
||||
ctx.subagents.registerProvider(new StubProvider('contain-end'))
|
||||
const second = vi.fn()
|
||||
ctx.on('subagent/end', () => { throw new Error('bad end listener') })
|
||||
ctx.on('subagent/end', second)
|
||||
|
||||
const run = ctx.subagents.start('contain-end', baseRequest())
|
||||
await run.result
|
||||
// Let the detached `.then` + the contained emit run.
|
||||
await Promise.resolve()
|
||||
await Promise.resolve()
|
||||
expect(second).toHaveBeenCalledWith(expect.objectContaining({ provider: 'contain-end', id: run.id, stopReason: 'completed' }))
|
||||
})
|
||||
|
||||
it('SubagentError extends the shared HarnessError base', () => {
|
||||
const err = new SubagentError('boom', 'NO_PROVIDER')
|
||||
expect(err).toBeInstanceOf(HarnessError)
|
||||
expect(err.name).toBe('SubagentError')
|
||||
expect(err.code).toBe('NO_PROVIDER')
|
||||
it('SubagentError participates in the harness error taxonomy', () => {
|
||||
const error = new SubagentError('boom', 'NO_PROVIDER')
|
||||
expect(error).toBeInstanceOf(HarnessError)
|
||||
expect(error.name).toBe('SubagentError')
|
||||
expect(error.code).toBe('NO_PROVIDER')
|
||||
})
|
||||
})
|
||||
|
||||
Reference in New Issue
Block a user