Coder/deepseek-harness
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(scope): close final ownership races

Browse Source 
Drain idle injection flushes before agent teardown, snapshot approval and subagent provider inputs, and gate subagent lifecycle events on real child readiness. Align the RFCs and generated contracts with the hardened behavior.
This commit is contained in:
Tianyi Cui
2026-07-12 00:31:46 +08:00
parent e6c7522a75
commit 3529b3c166
43 changed files with 825 additions and 261 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
packages/cordis/tool-cordis/src/api-catalog.ts
View File

@@ -381,7 +381,7 @@ export const EVENT_API: readonly EventApiEntry[] = [
name: 'subagent/end',
mode: 'emit',
signature: '\'subagent/end\'(this: Scoped<SubagentService>, info: SubagentRunEndInfo): void',
summary: 'A subagent run settled — emitted when SubagentRun.result resolves (any stop reason).',
summary: 'A started subagent run settled — emitted when SubagentRun.result resolves (any stop reason) or rejects (reported as `error`).',
},
{
name: 'subagent/provider-added',
@@ -399,7 +399,7 @@ export const EVENT_API: readonly EventApiEntry[] = [
name: 'subagent/start',
mode: 'emit',
signature: '\'subagent/start\'(this: Scoped<SubagentService>, info: SubagentRunInfo): void',
summary: 'A subagent run started — emitted after the provider is resolved and its capabilities validated, as the child run begins.',
summary: 'A subagent run started — emitted only after SubagentRun.started fulfills, when the provider has established a live child.',
},
{
name: 'system-prompt/assemble',
@@ -869,7 +869,7 @@ export const TYPE_API: readonly TypeApiEntry[] = [
},
{
name: 'SubagentRun',
declaration: 'export interface SubagentRun {\n readonly id: AgentId;\n readonly result: Promise<SubagentResult>;\n cancel(reason?: string): void;\n dispose(): Promise<void>;\n sendMessage?(content: ContentBlock[]): void;\n resume?(content: ContentBlock[]): SubagentRun;\n}',
declaration: 'export interface SubagentRun {\n readonly id: AgentId;\n readonly started: Promise<void>;\n readonly result: Promise<SubagentResult>;\n cancel(reason?: string): void;\n dispose(): Promise<void>;\n sendMessage?(content: ContentBlock[]): void;\n resume?(content: ContentBlock[]): SubagentRun;\n}',
},
{
name: 'SubagentStartRequest',

2
packages/core/agent-loop/README.md
View File

@@ -8,7 +8,7 @@ This is the only package in the harness that contains concrete loop logic. Every
### Public API
Lifecycle (scoped): programmatic creation and resume snapshot caller-owned identity/configuration data, reserve both IDs, mint `agent.ctx`, and install the ordered teardown skeleton before awaiting optional `setup`. Resume installs an owner-liveness sentinel before persistence load, then hands ownership directly to the full lifecycle. After setup resolves, the factory checks its lifecycle flag, owner-fiber state, and owning agent status around one microtask checkpoint so a same-turn Cordis unload wins before publication. Successful setup inserts both session and agent before announcing either, enables driving immediately before `agent/session-start`, then starts the loop. Setup calls to `send`/`steer`/`inject`/`cancel` reject structurally; load/setup rejection or owner unload publishes nothing. Teardown runs stop/drain → unregister → detach session → unwind scope. All `agent/*` dispatches go through `agentEvents(ctx, agent)`; per-step assembly through `assembleContextFor(agent)`; the turn-end durability checkpoint through `ctx.sessions.flush(session)`.
Lifecycle (scoped): programmatic creation and resume snapshot caller-owned identity/configuration data, reserve both IDs, mint `agent.ctx`, and install the ordered teardown skeleton before awaiting optional `setup`. Resume installs an owner-liveness sentinel before persistence load, then hands ownership directly to the full lifecycle. After setup resolves, the factory checks its lifecycle flag, owner-fiber state, and owning agent status around one microtask checkpoint so a same-turn Cordis unload wins before publication. Successful setup inserts both session and agent before announcing either, enables driving immediately before `agent/session-start`, then starts the loop. Setup calls to `send`/`steer`/`inject`/`cancel` reject structurally; load/setup rejection or owner unload publishes nothing. Teardown runs stop/drain (including outstanding idle-injection flushes) → unregister → detach session → unwind scope. All `agent/*` dispatches go through `agentEvents(ctx, agent)`; per-step assembly through `assembleContextFor(agent)`; the turn-end durability checkpoint through `ctx.sessions.flush(session)`.
- `ctx.agentLoop.create(id: string, options?: AgentOptions, meta?: { cwd?: string }): ReactLoopAgent` — config-driven create: an agent on a fresh per-run session id `${id}-session-<uuid>` with optional session metadata. Used for `cordis.yml`-configured agents. The per-run uuid avoids colliding with the on-disk log a prior run materialized once a durable persistence backend is loaded; each run is a new session (a deliberate demo simplification — a real resume-or-create policy is a TODO). Disposed with the calling fiber.

96
packages/core/agent-loop/src/agent.ts
View File

@@ -31,8 +31,12 @@ export interface PreparedReactLoopAgent {
agent: ReactLoopAgent
/** Open its driving verbs at the rollback-covered publication boundary. */
enableDrive(): void
/** Start its driver after publication and session-start notification. */
startDriver(): () => void
/**
* Start its driver after publication and session-start notification.
* The returned disposer reaches quiescence for both the loop and every
* fire-and-forget idle-injection flush the agent started.
*/
startDriver(): () => Promise<void>
}
/**
@@ -126,6 +130,12 @@ export class ReactLoopAgent implements Agent {
* the `disposed` transition fires and leave the promise hanging.
*/
private idleWaiters: (() => void)[] = []
/**
* Durability checkpoints started by idle {@link inject} calls. `inject()` is
* synchronous, so it cannot await them itself; the driver disposer drains
* this set before the lifecycle unregisters the agent or detaches its session.
*/
private pendingIdleFlushes = new Set<Promise<void>>()
constructor(
private loopCtx: Context,
@@ -249,14 +259,16 @@ export class ReactLoopAgent implements Agent {
// will flush this turn. Fire-and-forget with error containment: inject()
// is synchronous, and a persistence backend failing must not throw into
// the caller (e.g. a tool-bash task-done callback). Disposal still drains
// independently, so a slow flush is safe. A flush failure is reported via
// agent/error (step 0 — the idle-injection convention, there is no real
// step) AND the logger, mirroring the loop's post-turn/end flush path so
// plugins monitoring agent/error see idle-injection persistence failures
// too. A throwing agent/error listener is contained.
// independently, so a slow flush is safe. The task is tracked until it
// settles: driver disposal awaits every pending idle-injection checkpoint
// before unregistering the agent or detaching the session. A flush failure
// is reported via agent/error (step 0 — the idle-injection convention,
// there is no real step) AND the logger, mirroring the loop's post-turn/end
// flush path so plugins monitoring agent/error see idle-injection
// persistence failures too. A throwing agent/error listener is contained.
if (turnRecorded) {
// Through the store's flush (the carrier owner), never a raw parallel.
void this.loopCtx.sessions.flush(this.session).catch((error: unknown) => {
const flush = this.loopCtx.sessions.flush(this.session).catch((error: unknown) => {
const err = error instanceof Error ? error : new Error(String(error))
this.loopCtx.logger.warn(`agent "${this.id}": flush after idle injection failed: ${err.message}`)
try {
@@ -266,6 +278,13 @@ export class ReactLoopAgent implements Agent {
// listener must not escape this fire-and-forget catch.
}
})
this.pendingIdleFlushes.add(flush)
// Attach the same retirement callback to both settlement arms so even a
// logger failure in the catch above cannot become an unhandled rejection.
// Teardown uses allSettled for the same reason: a reporting failure must
// not strand ownership.
const retire = (): void => { this.pendingIdleFlushes.delete(flush) }
void flush.then(retire, retire)
}
}
}
@@ -310,8 +329,8 @@ export class ReactLoopAgent implements Agent {
* fully ended) or chaining {@link done} on `disposed` (wait for the loop to
* actually exit). Implements the {@link Agent.whenIdle} contract: a non-owner
* quiescence-observation hook, distinct from teardown (a lifecycle owner stops
* and unregisters via `AgentHandle.dispose()`, which awaits {@link done}
* directly, not through this).
* and unregisters via `AgentHandle.dispose()`, whose driver boundary awaits
* both {@link done} and outstanding idle-injection flushes, not through this).
*/
whenIdle(): Promise<void> {
if (this._status === 'disposed') return this.done
@@ -334,12 +353,14 @@ export class ReactLoopAgent implements Agent {
* Start the driver loop. Returns a disposer: calling it sets status to
* `disposed`, emits `agent/status('disposed')`, resolves the disposed
* promise (unblocking the idle wait), releases any `whenIdle` waiters, and
* aborts the current request if any. The returned `agent.done` promise
* resolves once the loop exits.
* @returns the disposer — idempotent and infallible (it runs inside the
* fiber's LIFO disposal chain, where a throw would skip later disposers).
* aborts the current request if any. Its returned promise resolves only after
* the loop exits and every idle-injection flush started by this agent settles.
* @returns the disposer — idempotent, synchronously marks the agent disposed,
* and asynchronously reaches loop + flush quiescence without rejecting (it
* runs inside the fiber's LIFO disposal chain, where a rejection would skip
* later disposers).
*/
[startDriver](): () => void {
[startDriver](): () => Promise<void> {
this.done = runLoop(this.loopCtx, this, {
inbox: this.#inbox,
setStatus: (status) => { this.setStatus(status) },
@@ -360,22 +381,35 @@ export class ReactLoopAgent implements Agent {
// The disposer must be infallible: it runs inside the fiber's LIFO
// disposal chain, where a throw would skip later disposers (e.g. the
// registry unregistration) and leave `done` pending forever.
return () => {
if (this._status === 'disposed') return
this._status = 'disposed'
this.resolveDisposed()
// Release whenIdle waiters BEFORE the (guarded) event emit — they are
// internal state that must settle even if a listener throws below. Each
// waiter chains `done`, so it resolves only once the loop actually exits.
this.settleIdleWaiters()
this.currentAbort?.abort('disposed')
// setStatus refuses transitions out of 'disposed', so emit directly —
// 'disposed' is part of the agent/status contract. Guarded: a throwing
// listener must not break the disposal chain.
try {
this.loopCtx.emit(this.carrier, 'agent/status', this, 'disposed')
} catch {
// listener error during disposal — nothing safe left to do with it
return async () => {
if (this._status !== 'disposed') {
this._status = 'disposed'
this.resolveDisposed()
// Release whenIdle waiters BEFORE the (guarded) event emit — they are
// internal state that must settle even if a listener throws below. Each
// waiter chains `done`, so it resolves only once the loop actually exits.
this.settleIdleWaiters()
this.currentAbort?.abort('disposed')
// setStatus refuses transitions out of 'disposed', so emit directly —
// 'disposed' is part of the agent/status contract. Guarded: a throwing
// listener must not break the disposal chain.
try {
this.loopCtx.emit(this.carrier, 'agent/status', this, 'disposed')
} catch {
// listener error during disposal — nothing safe left to do with it
}
}
// An unexpected driver rejection must not skip registry/session/scope
// cleanup. The normal loop contains turn failures itself; allSettled is the
// final lifecycle backstop for anything outside those boundaries.
await Promise.allSettled([this.done])
// No new inject() can start after the synchronous disposed transition.
// Loop because settled tasks retire themselves in promise reactions that
// may run beside this continuation; either the set is empty or this waits
// the exact remaining quiescence boundary. allSettled keeps a failure in
// error reporting from skipping the registry/session/scope disposers.
while (this.pendingIdleFlushes.size > 0) {
await Promise.allSettled([...this.pendingIdleFlushes])
}
}
}

14
packages/core/agent-loop/src/index.ts
View File

@@ -375,7 +375,7 @@ export class AgentLoop extends Service implements AgentFactory {
let active = true
let detachSession: (() => void) | undefined
let detachAgent: (() => void) | undefined
let stop: (() => void) | undefined
let stop: (() => Promise<void>) | undefined
const { promise: deactivated, resolve: markDeactivated } = Promise.withResolvers<void>()
const { promise: torndown, resolve: markTorndown } = Promise.withResolvers<void>()
@@ -401,8 +401,7 @@ export class AgentLoop extends Service implements AgentFactory {
active = false
markDeactivated()
if (stop === undefined) return
stop()
return agent.done
return stop()
}
}, 'agentLoop.lifecycle()')
@@ -461,8 +460,9 @@ export class AgentLoop extends Service implements AgentFactory {
/**
* Build an {@link AgentHandle} for a PREPARED session + a fresh agent. The
* handle's `dispose()` runs the composite effect's disposer (see
* {@link start}) — which stops the loop, awaits its exit (final flush
* captured), unregisters the agent, and detaches the session, in that order.
* {@link start}) — which stops the loop, awaits its exit and outstanding
* idle-injection flushes, unregisters the agent, and detaches the session, in
* that order.
* The same composite effect is what a fiber unload disposes, so both teardown
* triggers honor the ordering identically.
*
@@ -470,8 +470,8 @@ export class AgentLoop extends Service implements AgentFactory {
* single-shot (a second call returns immediately because the effect's epoch is
* already cleared, NOT awaiting the in-flight teardown), so concurrent/repeated
* `dispose()` calls would otherwise resolve before the first call's
* `await agent.done` + final flush completed. Memoizing the promise makes every
* caller observe the SAME quiescence boundary, honoring the
* loop + flush quiescence boundary completed. Memoizing the promise makes
* every caller observe that SAME boundary, honoring the
* `AgentHandle.dispose(): Promise<void>` contract (mirrors the ACP `quiesce()`
* helper).
*/

2
packages/core/agent-loop/src/loop.ts
View File

@@ -489,7 +489,7 @@ async function runTurn(
// against the system prompt (it counts toward the budget). runStep reuses
// this same assembly for the request, so the prompt is assembled once per
// step. renderPrompt IS the full prompt — the persona is the order-0
// section (registered by the AgentLoop plugin) and `{{variable}}`
// section (owned by dsh-system-prompt) and `{{variable}}`
// interpolation happens in the render, so there is no separate join.
const assembly = await ctx.systemPrompt.assemble(assembleContextFor(agent))
const fullSystemPrompt = renderPrompt(assembly)

9
packages/core/agent-loop/tests/agent.spec.ts
View File

@@ -242,11 +242,12 @@ describe('ReactLoopAgent', () => {
const dispose = prepared.startDriver()
// First dispose
dispose()
const firstDisposal = dispose()
expect(agent.status).toBe('disposed')
await firstDisposal
// Second dispose — idempotent, no throw
expect(() => { dispose() }).not.toThrow()
await expect(dispose()).resolves.toBeUndefined()
expect(agent.status).toBe('disposed')
})
@@ -347,10 +348,10 @@ describe('ReactLoopAgent', () => {
expect(agent.status).toBe('running')
const idle = agent.whenIdle() // queues an internal waiter (running)
dispose() // settles the waiter synchronously; whenIdle chains done
const disposal = dispose() // settles the waiter synchronously; whenIdle chains done
await idle
expect(agent.status).toBe('disposed')
await agent.done
await disposal
})
it('whenIdle() subscribed while running survives a FIBER dispose (no hung promise)', async () => {

31
packages/core/agent-loop/tests/scope-lifecycle.spec.ts
View File

@@ -440,4 +440,35 @@ describe('agent scope lifecycle', () => {
expect(ctx.sessions.get(SessionId('h1-s'))).toBeUndefined()
await unload
})
it('handle.dispose() awaits an idle-injection flush before unregistering or detaching', async () => {
const ctx = await harness()
const handle = await ctx.agents.create({
agentId: AgentId('idle-flush'),
sessionId: SessionId('idle-flush-s'),
agentOptions: { model: 'mock' },
})
const gate = Promise.withResolvers<undefined>()
let flushStarted = false
ctx.on('session/flush', (session) => {
if (session !== handle.agent.session) return
flushStarted = true
return gate.promise
})
handle.agent.inject(text('durable idle context'), { source: { kind: 'plugin', plugin: 'test' } })
expect(flushStarted).toBe(true)
let disposed = false
const disposal = handle.dispose().then(() => { disposed = true })
await new Promise(resolve => setTimeout(resolve, 0))
expect(disposed).toBe(false)
expect(ctx.agents.get(AgentId('idle-flush'))).toBe(handle.agent)
expect(ctx.sessions.get(SessionId('idle-flush-s'))).toBe(handle.agent.session)
gate.resolve(undefined)
await disposal
expect(ctx.agents.get(AgentId('idle-flush'))).toBeUndefined()
expect(ctx.sessions.get(SessionId('idle-flush-s'))).toBeUndefined()
})
})

2
packages/core/agent/README.md
View File

@@ -23,7 +23,7 @@ Agent *creation* is provided by the plugin implementing `AgentFactory` (`dsh-age
- `ctx.agents.create(options: CreateAgentOptions): Promise<AgentHandle>` — snapshot caller-owned IDs/options/metadata/seed, construct and await optional setup while unpublished, insert and announce both session and agent, open the `agent/session-start` driving boundary, then start a new loop on the caller-supplied `sessionId`. Agent/session IDs are reserved across setup; setup rejection or owner unload publishes nothing. Publication is rollback-covered: if a creation listener throws, entries and scope unwind but effects of already-delivered notifications remain observable; an agent whose announcement began emits `agent/disposed` during that rollback. Rejects if no factory is registered.
- `ctx.agents.resume(options: ResumeAgentOptions): Promise<AgentHandle>` — snapshot caller-owned IDs/options, load a persisted session ([session persistence](../../../docs/rfc/implemented/architecture/2026-06-14-session-persistence.md)), mint a fresh agent scope, await optional setup while unpublished, then follow the same insert → announce → session-start → loop-start boundary. The IDs are reserved across persistence load and setup; load/setup rejection or owner unload publishes nothing. Rejects if no factory is registered or session persistence is unconfigured.
`AgentHandle = { agent: Agent; dispose(): Promise<void> }`. The disposer is a **capability** — only the holder can tear this agent down. `dispose()` stops the loop, `await`s its exit (quiescence — NOT just the `disposed` status flip), unregisters the agent, removes its session from the store, and finally unwinds its scoped world. This order captures the loop's final `session/flush` before the session is detached and keeps scoped listeners alive through that flush. `ctx.agents.get(id)` still returns a bare `Agent` — the handle is only for the OWNER that created it. The ACP bridge and in-process subagent backends are production consumers; config-created agents are owned by the loop fiber and never need a handle.
`AgentHandle = { agent: Agent; dispose(): Promise<void> }`. The disposer is a **capability** — only the holder can tear this agent down. `dispose()` stops the loop, `await`s its exit plus every outstanding idle-injection flush (quiescence — NOT just the `disposed` status flip), unregisters the agent, removes its session from the store, and finally unwinds its scoped world. This order captures every agent-started `session/flush` before the session is detached and keeps scoped listeners alive through those checkpoints. `ctx.agents.get(id)` still returns a bare `Agent` — the handle is only for the OWNER that created it. The ACP bridge and in-process subagent backends are production consumers; config-created agents are owned by the loop fiber and never need a handle.
### Live events

11
packages/core/agent/src/index.ts
View File

@@ -107,11 +107,12 @@ export interface ResumeAgentOptions {
/**
* An owned agent plus its disposer, returned by {@link AgentRegistry.create} /
* {@link AgentRegistry.resume}. The disposer is a CAPABILITY: only the holder
* can tear this agent down. `dispose()` stops the loop, awaits its exit
* (quiescence — NOT just the `disposed` status flip), unregisters the agent,
* removes its session from the store, and finally unwinds its scoped world.
* This order captures the loop's final `session/flush` before the session is
* detached and keeps scoped listeners alive through that flush.
* can tear this agent down. `dispose()` stops the loop, awaits its exit and
* every outstanding idle-injection flush (quiescence — NOT just the `disposed`
* status flip), unregisters the agent, removes its session from the store, and
* finally unwinds its scoped world. This order captures every agent-started
* `session/flush` before the session is detached and keeps scoped listeners
* alive through those checkpoints.
*
* `ctx.agents.get(id)` still returns a bare {@link Agent} — the handle is only
* for the OWNER that created it. Config-created agents (the loop's own startup)

6
packages/core/agent/src/types.ts
View File

@@ -222,8 +222,10 @@ export interface Agent {
* turn (`turn/start` → `context/message` → `turn/end`) and checkpoints it for
* durability, so every event stays inside a turn and a persistence backend
* never loses a between-turn notice. The idle checkpoint is fire-and-forget
* (inject is synchronous): a failing flush is reported via `agent/error`
* (step `0`) and the logger, never thrown into the caller.
* from this synchronous method, but lifecycle disposal awaits it before
* unregistering the agent or detaching its session. A failing flush is
* reported via `agent/error` (step `0`) and the logger, never thrown into the
* caller.
*
* Live-adapter review has validated the tagged-envelope rendering against
* current DeepSeek behavior; provider-specific mismatches belong in that

2
packages/hooks/hooks-claude/README.md
View File

@@ -39,7 +39,7 @@ The hooks **themselves** run in the agent's session workspace: for the agent-sco
| `PreToolUse` | `tools/pre-execute` (waterfall) | `deny` → `PreToolDecision.deny`; `ask` → `PreToolDecision.ask` |
| `PostToolUse` | `tools/post-execute` (waterfall) | `deny` → `block` with feedback; additionalContext-only → delegate via `next()` then fold context onto the downstream decision (a Code Mode sub-call’s context is dropped by the run_code bridge — see [the pipeline doc](../../../docs/tool-execution-pipeline.md)) |
| `Stop` | `agent/turn-continuation` (waterfall) | a blocking Stop hook forces `continue`, feeding its reason as next-step steering |
| `SubagentStart` | `subagent/start` (emit) | additionalContext → `agent.inject()` into the live child |
| `SubagentStart` | `subagent/start` (emit) | additionalContext → `agent.inject()` into a live in-process child; a remote child has no local injection target |
| `SubagentStop` | `subagent/end` (emit) | observe-only |
The three emit points run detached — no seam awaits a `SessionStart`/`SubagentStart`/`SubagentStop` hook. Each run chain is tracked, and disposing the bridge aborts still-running hook processes, then drains the continuations before the dispose resolves (`createDetachedRuns` in `dsh-hook-protocol`).

4
packages/subagent/subagent-acp/README.md
View File

@@ -6,7 +6,7 @@ It is the direction-inverted twin of the server-side bridge in [`@deepseek-ai/ds
## What it does
`start(request)` spawns the configured command, wraps its stdio in an ACP `ClientSideConnection`, and drives one session: `initialize` → `newSession` → `prompt`. The child's streamed `agent_message_chunk` text becomes the `SubagentResult.output`; the prompt's terminal `StopReason` maps to the stop reason. `dispose()` kills the subprocess and awaits its exit.
`start(request)` spawns the configured command, wraps its stdio in an ACP `ClientSideConnection`, and drives one session: `initialize` → `newSession` → `prompt`. `run.started` resolves after `newSession` publishes the remote session and rejects when initialization fails or cancellation wins first; the service emits no start/end pair for a child that never became live. The child's streamed `agent_message_chunk` text becomes the `SubagentResult.output`; the prompt's terminal `StopReason` maps to the stop reason. `dispose()` kills the subprocess and awaits its exit.
**Fresh process per run.** Each `start` spawns a new child, runs exactly one ACP session, and disposes it. Persistent-process pooling is a future optimization (see the RFC).
@@ -22,7 +22,7 @@ Unlike the in-process backends, the child does NOT share this cordis context —
| `providerName` | string | `acp` | Registry name on `ctx.subagents`. |
| `command` | string | — (required) | The executable to spawn for each run (the child ACP agent). |
| `args` | string[] | `[]` | Arguments passed to `command`. |
| `cwd` | string | parent cwd | Working directory for the child process and its ACP session. |
| `cwd` | string | process cwd | Working directory for the child process and its ACP session. |
| `permission` | `'allow' \| 'reject'` | `reject` | How to auto-answer the child's `session/request_permission` prompts. `reject` declines every prompt (answer `cancelled`); `allow` approves via the first allow-shaped option. The first cut surfaces no prompt to a human. |
| `env` | Record<string,string> | `{}` | Extra env vars for the child (e.g. its own `DEEPSEEK_API_KEY`). Forwarded on top of a credential-scrubbed copy of the parent env, so an explicit key reaches the child while ambient secrets do not leak implicitly. |
| `disposeEofGraceMs` | number | `6000` | Dispose ladder tier 1: how long the child gets to quiesce on its own after stdin EOF (flush persistence, tear down its nested subprocesses) before SIGTERM. |

97
packages/subagent/subagent-acp/src/run.ts
View File

@@ -196,9 +196,15 @@ export function startAcpRun(request: SubagentStartRequest, spec: AcpRunSpec): Su
// return an inert run that settled `aborted`, rather than launching the
// configured binary just to tear it down. `dispose`/`cancel` are no-ops.
if (request.signal?.aborted) {
const started = Promise.reject(new Error('subagent request was aborted before the ACP child started'))
// The result is derived from the same boundary so the readiness rejection
// is observed even when this provider is driven directly rather than
// through SubagentService.
const result: Promise<SubagentResult> = started.catch(() => ({ output: [], stopReason: 'aborted' }))
return {
id,
result: Promise.resolve({ output: [], stopReason: 'aborted' }),
started,
result,
cancel(_reason?: string): void { /* nothing was started */ },
dispose(): Promise<void> { return Promise.resolve() },
}
@@ -289,54 +295,70 @@ export function startAcpRun(request: SubagentStartRequest, spec: AcpRunSpec): Su
const onAbort = (): void => { requestCancel() }
request.signal?.addEventListener('abort', onAbort, { once: true })
// The accumulated child text as harness ContentBlocks (empty array when the
// child streamed nothing). Read at every return so a partial answer survives
// a later cancel/error.
const collectOutput = (): ContentBlock[] => {
const text = output.join('')
return text.length > 0 ? [{ type: 'text', text }] : []
}
// A provider is "started" only once the remote child has completed ACP
// initialization and published a session. SubagentService gates its
// `subagent/start` notification on this boundary, just as the in-process
// provider gates it on local Agent publication. Failure or cancellation
// before this point rejects readiness and therefore produces no paired
// lifecycle events for a child that never became live.
const started: Promise<void> = Promise.race([
(async (): Promise<void> => {
await conn.initialize({
protocolVersion: PROTOCOL_VERSION,
// Advertise NO optional client capabilities (no fs, no terminal): the
// child self-serves in its own process.
clientCapabilities: {},
})
const session = await conn.newSession({ cwd: spec.cwd, mcpServers: [] })
sessionId = session.sessionId
if (flags.cancelled) throw new Error('subagent cancelled before the ACP session started')
})(),
spawnFailed.then((err): never => { throw err }),
cancelSettled.then((): never => { throw new Error('subagent cancelled before the ACP session started') }),
])
const result: Promise<SubagentResult> = (async (): Promise<SubagentResult> => {
// The accumulated child text as harness ContentBlocks (empty array when the
// child streamed nothing). Read at every return so a partial answer survives
// a later cancel/error.
const collectOutput = (): ContentBlock[] => {
const text = output.join('')
return text.length > 0 ? [{ type: 'text', text }] : []
}
try {
// Race three outcomes, first to settle wins:
// - driveAcp: the normal initialize → newSession → prompt path;
// - spawnFailed: a bad command never speaks ACP, so `initialize` would
// hang forever — the spawn `error` event is the only signal, and a
// rejected race settles the run `error` via the catch;
// Readiness is the initialize → newSession phase above. Awaiting the SAME
// promise immediately observes its rejection even without the service,
// and guarantees the prompt phase never starts before the provider can
// truthfully announce a live child.
await started
// Race two post-start outcomes, first to settle wins:
// - prompt: the normal remote turn;
// - cancelSettled: a cancel was requested — settle `aborted` immediately
// rather than waiting on a child that may ignore `session/cancel` or
// wedge the prompt (the `cancel()` contract: `result` settles `aborted`).
const driveAcp = async (): Promise<SubagentResult> => {
await conn.initialize({
protocolVersion: PROTOCOL_VERSION,
// Advertise NO optional client capabilities (no fs, no terminal): the
// child self-serves in its own process.
clientCapabilities: {},
})
const session = await conn.newSession({ cwd: spec.cwd, mcpServers: [] })
sessionId = session.sessionId
// A cancel that raced ahead of `newSession` set `cancelled` but could not
// send `session/cancel` (no session id yet). Honor it here: settle
// `aborted` without ever issuing the prompt, rather than running the child
// to completion and ignoring the cancel.
if (flags.cancelled) return { output: collectOutput(), stopReason: 'aborted' }
const promptResult = await conn.prompt({ sessionId, prompt: toAcpPrompt(request.prompt) })
// A spawn error can only precede readiness and is already one arm of
// `started`; after `newSession` succeeds, transport/process failure rejects
// the in-flight prompt RPC through the connection.
const prompt = async (): Promise<SubagentResult> => {
// `started` cannot fulfill without assigning the session id; the cast
// records that local invariant without an unreachable defensive arm.
const promptResult = await conn.prompt({ sessionId: sessionId as string, prompt: toAcpPrompt(request.prompt) })
return { output: collectOutput(), stopReason: acpStopReason(promptResult.stopReason) }
}
return await Promise.race([
driveAcp(),
spawnFailed.then((err): SubagentResult => { throw err }),
prompt(),
cancelSettled.then((): SubagentResult => ({ output: collectOutput(), stopReason: 'aborted' })),
])
} catch (error: unknown) {
if (flags.cancelled) return { output: collectOutput(), stopReason: 'aborted' }
// The seam contract: result resolves (never rejects) on a child-level
// failure. Cancellation is handled by the `cancelSettled` race arm above
// (it settles `aborted` the instant cancel is requested, beating any
// rejection), so a rejection that reaches HERE is always a genuine
// child-level error — the awaited ACP RPCs or the spawn-failure race
// (initialize/newSession/prompt transport/RPC errors, or ENOENT), not a
// local bug. Flatten to `error` and surface the original via onError so a
// real fault is preserved rather than silently lost.
// failure. A cancellation is recognized by the flag above even when it
// wins during readiness; every other rejection is a genuine child-level
// error — initialize/newSession/prompt transport/RPC failure or ENOENT.
// Flatten to `error` and surface the original via onError so a real fault
// is preserved rather than silently lost.
try {
spec.onError?.(toError(error), 'error')
} catch {
@@ -350,6 +372,7 @@ export function startAcpRun(request: SubagentStartRequest, spec: AcpRunSpec): Su
return {
id,
started,
result,
cancel(_reason?: string): void {
requestCancel()

2
packages/subagent/subagent-fork/README.md
View File

@@ -1,6 +1,6 @@
# @deepseek-ai/dsh-subagent-fork
The in-process **fork** subagent backend: a [`SubagentProvider`](../subagent/README.md) that runs each child as a child [`Agent`](../../core/agent) **seeded with a prefix of the parent's session log** — so the child inherits the parent's conversation context instead of starting fresh. Shares the run driver (`startInProcessRun`) with [`dsh-subagent-spawn`](../subagent-spawn/README.md); the only difference is the seed.
The in-process **fork** subagent backend: a [`SubagentProvider`](../subagent/README.md) that runs each child as a child [`Agent`](../../core/agent) **seeded with a prefix of the parent's session log** — so the child inherits the parent's conversation context instead of starting fresh. Shares the run driver (`startInProcessRun`) with [`dsh-subagent-spawn`](../subagent-spawn/README.md); the only difference is the seed. The shared `run.started` boundary resolves only after the seeded child is published, so `subagent/start` observers see a live registry entry.
## The seed boundary (the crux)

17
packages/subagent/subagent-fork/tests/subagent-fork.spec.ts
View File

@@ -71,6 +71,23 @@ describe('completedTurnPrefix', () => {
})
describe('dsh-subagent-fork', () => {
it('emits subagent/start only after the seeded child is published', async () => {
const { ctx, parent } = await setup([textResponse('child answer')])
let childAtStart: ReturnType<typeof ctx.agents.get>
ctx.on('subagent/start', (info) => {
if (info.provider === 'fork') childAtStart = ctx.agents.get(info.id)
})
const run = ctx.subagents.start('fork', { prompt: [{ type: 'text', text: 'child q' }], parent })
expect(childAtStart).toBeUndefined()
await run.started
expect(childAtStart).toBe(ctx.agents.get(run.id))
expect(childAtStart?.id).toBe(run.id)
await run.result
await run.dispose()
})
it('forks an UNSEEDED (fresh) child when the parent has no completed turn', async () => {
// The parent has never completed a turn → empty prefix → the provider omits
// the seed → the child runs fresh. Exercises the `seed.length > 0` false arm.

4
packages/subagent/subagent-inprocess/README.md
View File

@@ -9,11 +9,11 @@ The shared **in-process subagent run driver**. A library with no provider or imp
Runs a child as a child [`Agent`](../../core/agent) on the same cordis context (`ctx.agents`):
1. snapshots the accepted request before asynchronous owner setup: the parent and signal remain identity capabilities but are never reread from the caller-owned record; tool filter, seed, agent options, output schema, and prompt are detached. It computes child depth = `depthOf(parent) + 1` and rejects `request.maxDepth` overflow with `SubagentDepthError`; `outputSchema` is asserted before cloning so a hostile value fails as `OutputSchemaError`, while the prompt passes the session log's lossless-JSON check before and after cloning;
2. first installs provider ownership, then attaches the request abort listener and creates one run-owner Cordis fiber under `parent.ctx`; an already-unloading provider therefore leaves no child or orphaned listener. Async child creation goes through that fiber's `ctx.agents` service with fresh IDs, lineage/seed, inherited model, and an unpublished setup transaction for persona, tool restriction, and structured output. Parent teardown, provider teardown, and manual `run.dispose()` all dispose this exact node, preventing publication after it becomes inactive and awaiting the same quiescence boundary. `startInProcessRun` still returns its `SubagentRun` immediately, and cancellation during creation is recorded and applied when a child exists;
2. first installs provider ownership, then attaches the request abort listener and creates one run-owner Cordis fiber under `parent.ctx`; an already-unloading provider therefore leaves no child or orphaned listener. Async child creation goes through that fiber's `ctx.agents` service with fresh IDs, lineage/seed, inherited model, and an unpublished setup transaction for persona, tool restriction, and structured output. Parent teardown, provider teardown, and manual `run.dispose()` all dispose this exact node, preventing publication after it becomes inactive and awaiting the same quiescence boundary. `startInProcessRun` still returns its `SubagentRun` immediately: `run.started` resolves only after `ctx.agents.create()` has published the child (and rejects if publication never happens), while cancellation during creation is recorded and applied when a child exists;
3. drives the one-shot: `child.send(prompt)` then `await child.whenIdle()` (ordering matters — `send` enqueues synchronously, so `whenIdle` observes the queued work and resolves on the child's `running → idle` transition, never before the turn starts); there is deliberately NO re-prompt for a structured child that finished cleanly without calling `structured_output` — the shortfall maps to an `error` result for the parent;
4. reads the result, scoped to the child's OWN events (everything at or after `seedLength`, so a seeded child that produced no message of its own never returns the seeded parent's last message): the last `assistant/message` content (deep-cloned — the log is frozen) and the last `turn/end.reason` mapped to a `SubagentStopReason`. A structured run surfaces the captured value as `result.structured`; a structured child that finished cleanly WITHOUT ever capturing settles `error` (a clean finish without the demanded result is a failure, not a success with a missing field).
`dispose()` awaits creation or rollback and then delegates to `AgentHandle.dispose()` (stop and drain → remove agent → detach session → unwind scope); `cancel()` records its request even before publication and cancels the child immediately once available. A cancel landing before any `turn/end` still settles `aborted`, honoring the cancel contract rather than the generic no-turn `error`.
`SubagentService` waits for `run.started` before emitting `subagent/start`, so a synchronous start observer can resolve the published child with `ctx.agents.get(run.id)`; the result driver awaits the same boundary before sending the prompt. An attempt that never publishes rejects readiness and emits no false start/end pair; its result reports a deliberate cancel/dispose as `aborted` and propagates an infrastructure fault. `dispose()` awaits creation or rollback and then delegates to `AgentHandle.dispose()` (stop and drain → remove agent → detach session → unwind scope); `cancel()` records its request even before publication and cancels the child immediately once available. A cancel landing before any `turn/end` still settles `aborted`, honoring the cancel contract rather than the generic no-turn `error`.
### `InProcessRunOptions`

15
packages/subagent/subagent-inprocess/src/index.ts
View File

@@ -289,11 +289,23 @@ export function startInProcessRun(
return created.agent
})()
// Provider readiness is a distinct lifecycle boundary from accepting the
// request. It resolves only after the factory has published the child and
// returned its handle, so SubagentService can emit `subagent/start` while
// `ctx.agents.get(childId)` is guaranteed to resolve. The result path awaits
// THIS SAME promise immediately, which also observes a readiness rejection
// when the driver is invoked directly rather than through SubagentService.
const started: Promise<void> = creation.then(() => undefined)
const result: Promise<SubagentResult> = (async () => {
try {
let liveChild: Agent
try {
liveChild = await creation
await started
// `creation` assigns `child` before it fulfills, and `started` is its
// direct fulfillment projection. The cast records that local invariant
// without manufacturing an unreachable runtime branch.
liveChild = child as Agent
} catch (error: unknown) {
if (isManualDisposeRequested()) return { output: [], stopReason: 'aborted' }
throw error instanceof Error ? error : new Error('subagent child creation failed with a non-Error value', { cause: error })
@@ -313,6 +325,7 @@ export function startInProcessRun(
let disposing: Promise<void> | undefined
return {
id: childId,
started,
result,
cancel(reason?: string): void {
requestCancel(reason ?? 'subagent cancelled')

2
packages/subagent/subagent-spawn/README.md
View File

@@ -6,7 +6,7 @@ The run mechanics live in the shared [`@deepseek-ai/dsh-subagent-inprocess`](../
## What it does
`start(request)` delegates to `startInProcessRun(ctx, request, {})` with no seed: a fresh child agent with the parent's `cwd`/`parentSession` lineage and (by default) the parent's model. The driver creates one run-owner fiber under `parent.ctx`; parent teardown, this provider's teardown, and manual disposal all converge there before child publication. See the [driver README](../subagent-inprocess/README.md) for the full lifecycle (depth check, one-shot drive, result read, dispose).
`start(request)` delegates to `startInProcessRun(ctx, request, {})` with no seed: a fresh child agent with the parent's `cwd`/`parentSession` lineage and (by default) the parent's model. The driver creates one run-owner fiber under `parent.ctx`; parent teardown, this provider's teardown, and manual disposal all converge there before child publication. Its `run.started` boundary resolves only after the fresh child is published, so `subagent/start` observers see a live registry entry. See the [driver README](../subagent-inprocess/README.md) for the full lifecycle (depth check, one-shot drive, result read, dispose).
## Capabilities

19
packages/subagent/subagent-spawn/tests/subagent-spawn.spec.ts
View File

@@ -55,6 +55,25 @@ describe('dsh-subagent-spawn', () => {
await run.dispose()
})
it('emits subagent/start only after the fresh child is published', async () => {
const { ctx, parent } = await setup([textResponse('child answer')])
let childAtStart: ReturnType<typeof ctx.agents.get>
ctx.on('subagent/start', (info) => {
if (info.provider === 'spawn') childAtStart = ctx.agents.get(info.id)
})
const run = ctx.subagents.start('spawn', { prompt: [{ type: 'text', text: 'do X' }], parent })
// Creation is asynchronous; no lifecycle claim is made while the child is
// still inside its unpublished setup transaction.
expect(childAtStart).toBeUndefined()
await run.started
expect(childAtStart).toBe(ctx.agents.get(run.id))
expect(childAtStart?.id).toBe(run.id)
await run.result
await run.dispose()
})
it('gives the child its OWN session (not the parent\'s), with parentSession lineage', async () => {
const { ctx, parent } = await setup([textResponse('hi')])
const run = ctx.subagents.start('spawn', { prompt: [{ type: 'text', text: 'p' }], parent })

10
packages/subagent/subagent/README.md
View File

@@ -18,10 +18,10 @@ Unlike the bash seam (one executor per context, second load throws), **multiple
| Member | Semantics |
|---|---|
| `registerProvider(provider)` | Register under `provider.name`. Throws `SubagentError('DUPLICATE_PROVIDER')` on a name clash. Effect-scoped (HMR-safe); returns the disposer. |
| `getProvider(name)` | Look up a provider (`undefined` if absent). |
| `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` and emit `subagent/start` / `subagent/end` around the run. |
| `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. |
## Capabilities: two kinds, discovered two ways
@@ -32,9 +32,9 @@ Beside `capabilities` sits one DESCRIPTIVE fact, not validated by the service: `
## Run lifecycle
`provider.start(request)` returns a `SubagentRun`: a handle with a `result` promise, `cancel()`, `dispose()`, and the optional runtime methods. `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.
`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.
The service also announces provider lifecycle: `subagent/provider-added` (the live provider) fires after a registration and `subagent/provider-removed` (the 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". The service emits `subagent/start` (payload `SubagentRunInfo`) and `subagent/end` (payload `SubagentRunEndInfo`) around the run — both **observe-only** (plain `emit`s; `subagent/end` fires from a detached `.then` and awaits no listener). `subagent/end` carries `lastAssistantMessage` (a deep clone of the child's final `output`) on the settle path, absent when the run rejected at the infrastructure level. The clone keeps the surface observe-only: the end emit fires from a detached `.then` before the caller's `await run.result` resumes, so a shared reference would let a mutating listener corrupt the caller's result. A `subagent/start` listener can still reach the live child via `ctx.agents.get(info.id)`; a `subagent/end` listener can only observe (the run has settled). Any run-affecting decision (continuation, injection that changes the run) is out of scope for this observe-only surface.
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.
## Scope (first cut)

148
packages/subagent/subagent/src/index.ts
View File

@@ -69,7 +69,7 @@ declare module 'cordis' {
* 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 provider that just registered, live in the registry.
* @param provider - the registry's frozen acceptance snapshot of the provider.
* @mode emit
*/
'subagent/provider-added'(provider: SubagentProvider): void
@@ -85,8 +85,11 @@ declare module 'cordis' {
*/
'subagent/provider-removed'(name: string): void
/**
* A subagent run started — emitted after the provider is resolved and its
* capabilities validated, as the child run begins. Paired with
* 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
@@ -97,8 +100,10 @@ declare module 'cordis' {
*/
'subagent/start'(this: Scoped<SubagentService>, info: SubagentRunInfo): void
/**
* A subagent run settled — emitted when {@link SubagentRun.result}
* resolves (any stop reason). Paired with {@link Events['subagent/start']}.
* 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
@@ -161,21 +166,43 @@ export class SubagentService extends Service {
/**
* Register a provider under its `provider.name`. Throws {@link SubagentError}
* (`DUPLICATE_PROVIDER`) if the name is already taken. 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.
* (`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.
*/
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(provider.name)) {
throw new SubagentError(`a subagent provider named "${provider.name}" is already registered`, 'DUPLICATE_PROVIDER')
if (this.providers.has(snapshot.name)) {
throw new SubagentError(`a subagent provider named "${snapshot.name}" is already registered`, 'DUPLICATE_PROVIDER')
}
this.providers.set(provider.name, 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
@@ -183,10 +210,10 @@ export class SubagentService extends Service {
// it runs inside this disposer, where a propagating subscriber would
// disrupt the backend fiber's teardown and starve later mirrors.
yield () => {
this.providers.delete(provider.name)
this.emitLifecycle('subagent/provider-removed', provider.name)
this.providers.delete(snapshot.name)
this.emitLifecycle('subagent/provider-removed', snapshot.name)
}
this.ctx.emit('subagent/provider-added', provider)
this.ctx.emit('subagent/provider-added', snapshot)
}.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 —
@@ -198,9 +225,10 @@ export class SubagentService extends Service {
}
/**
* Look up a registered provider by name (`undefined` if absent).
* @param name - the provider name as registered.
* @returns the provider, or undefined when the name is unknown.
* 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.
*/
getProvider(name: string): SubagentProvider | undefined {
return this.providers.get(name)
@@ -219,8 +247,9 @@ export class SubagentService extends Service {
* `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} and emits `subagent/start` /
* `subagent/end` around the run.
* 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).
@@ -252,46 +281,63 @@ export class SubagentService extends Service {
...request.persona !== undefined ? { persona: request.persona } : {},
}
const run = provider.start(accepted)
// Emit `subagent/start` with PER-LISTENER containment (see {@link emitLifecycle}):
// the run is already live, so neither a throwing subscriber escaping
// `start()` (the caller would never receive the run to dispose it — a leaked
// child) NOR one bad subscriber starving the listeners after it is
// acceptable. `ctx.emit` halts the dispatch on the first throw, so a single
// surrounding try/catch is not enough — each listener is invoked and
// contained individually.
this.emitLifecycle('subagent/start', { provider: name, id: run.id }, parent)
// Emit `subagent/end` when the run settles. The result promise does not
// reject on a child-level failure (it resolves with stopReason 'error'),
// so a rejection here is an infrastructure fault — surface its stop reason
// as 'error' for the telemetry event without swallowing the rejection
// (the consumer still observes it via `run.result`). On the resolve path the
// child's final output rides on the event (lastAssistantMessage); on the
// reject path there is no SubagentResult, so only the stop reason is known.
// Per-listener containment also keeps a thrown `subagent/end` listener from
// becoming an unhandled rejection on this detached `.then`.
// 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.
}
void run.result.then(
(result) => {
// Deep-clone the output onto the event: this detached `.then` runs BEFORE
// the caller's own `await run.result` continuation, so handing listeners
// the SAME array reference the caller consumes would let a mutating
// `subagent/end` listener corrupt the caller's SubagentResult.output —
// breaking the observe-only contract. A snapshot makes the event a
// read-only view, not a shared handle. The clone is wrapped: it runs
// inside `onFulfilled`, OUTSIDE emitLifecycle's per-listener containment,
// so an uncloneable value (a future non-serializable content-block type,
// or a contract-violating result with no `output`) would otherwise become
// an unhandled rejection on this detached `.then`. On clone failure, log
// and emit the event WITHOUT lastAssistantMessage rather than dropping the
// whole `subagent/end`.
// 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)}`)
}
this.emitLifecycle('subagent/end', { provider: name, id: run.id, stopReason: result.stopReason, ...lastAssistantMessage !== undefined ? { lastAssistantMessage } : {} }, parent)
deliverEnd({
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)
}
},
() => {
readiness = 'failed'
pendingEnd = undefined
},
() => { this.emitLifecycle('subagent/end', { provider: name, id: run.id, stopReason: 'error' }, parent) },
)
return run
}

23
packages/subagent/subagent/src/types.ts
View File

@@ -142,8 +142,16 @@ export interface SubagentResult {
* presence of the method IS the capability — narrow before calling.
*/
export interface SubagentRun {
/** The child agent's id (use `ctx.agents.get(id)` to reach the live child). */
/** The child agent's id (local in-process runs publish it in `ctx.agents`; remote transports need not). */
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
@@ -176,7 +184,9 @@ export interface SubagentRun {
* A subagent backend: one transport for running a child agent (in-process
* 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).
* 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.
*/
export interface SubagentProvider {
/** Unique registry name (e.g. `spawn`, `fork`, `acp`). */
@@ -194,9 +204,12 @@ export interface SubagentProvider {
*/
readonly inheritsParentContext: boolean
/**
* Start a child run. 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.
* Start preparing a child run and return its handle synchronously. 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.
*/
start(request: SubagentStartRequest): SubagentRun
}

140
packages/subagent/subagent/tests/service.spec.ts
View File

@@ -34,6 +34,7 @@ class StubProvider implements SubagentProvider {
this.startCount++
return {
id: AgentId(`child:${this.name}:${request.parent.id}`),
started: Promise.resolve(),
result: Promise.resolve(this.result),
cancel() {},
async dispose() {},
@@ -106,7 +107,7 @@ describe('SubagentService', () => {
ctx.subagents.registerProvider(provider)
expect(ctx.subagents.list()).toEqual(['alpha'])
expect(ctx.subagents.getProvider('alpha')).toBe(provider)
expect(ctx.subagents.getProvider('alpha')).toMatchObject({ name: 'alpha' })
const run = ctx.subagents.start('alpha', baseRequest())
expect(provider.startCount).toBe(1)
@@ -162,6 +163,75 @@ describe('SubagentService', () => {
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',
}))
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)
@@ -220,6 +290,7 @@ describe('SubagentService', () => {
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
@@ -228,6 +299,67 @@ describe('SubagentService', () => {
expect(ended).toHaveBeenCalledWith(expect.objectContaining({ provider: 'events', id: run.id, stopReason: 'completed' }))
})
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('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>()
const result = Promise.withResolvers<SubagentResult>()
ctx.subagents.registerProvider({
name: 'never-started',
capabilities: NO_CAPS,
inheritsParentContext: false,
start: () => ({
id: AgentId('never-started-child'),
started: readiness.promise,
result: result.promise,
cancel() {},
async dispose() {},
}),
})
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()
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)
@@ -241,6 +373,7 @@ describe('SubagentService', () => {
acceptedRequest = accepted
return {
id: AgentId('deferred-child'),
started: Promise.resolve(),
result: gate.promise,
cancel() {},
async dispose() {},
@@ -282,6 +415,7 @@ describe('SubagentService', () => {
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 }))
await run.result
@@ -331,6 +465,7 @@ describe('SubagentService', () => {
inheritsParentContext: false,
start: () => ({
id: AgentId('rej-child'),
started: Promise.resolve(),
result: Promise.reject(new Error('infra fault')),
cancel() {},
dispose: async () => {},
@@ -365,6 +500,7 @@ describe('SubagentService', () => {
inheritsParentContext: false,
start: () => ({
id: AgentId('unclone-child'),
started: Promise.resolve(),
result: Promise.resolve({ output: uncloneable, stopReason: 'completed' } as SubagentResult),
cancel() {},
dispose: async () => {},
@@ -395,6 +531,7 @@ describe('SubagentService', () => {
inheritsParentContext: false,
start: () => ({
id: AgentId('rej-child'),
started: Promise.resolve(),
result: Promise.reject(new Error('infra fault')),
cancel() {},
dispose: async () => {},
@@ -424,6 +561,7 @@ describe('SubagentService', () => {
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' })
})

10
packages/subagent/tool-subagent/tests/tool-subagent.spec.ts
View File

@@ -115,6 +115,7 @@ describe('dsh-tool-subagent', () => {
inheritsParentContext: false,
start: () => ({
id: AgentId('weird-child'),
started: Promise.resolve(),
result: Promise.resolve({ output: [{ type: 'text', text: 'partial' }], stopReason: 'frobnicated' as never }),
cancel() {},
dispose: async () => {},
@@ -143,6 +144,7 @@ describe('dsh-tool-subagent', () => {
seen = request
return {
id: AgentId('capture-child'),
started: Promise.resolve(),
result: Promise.resolve({ output: [{ type: 'text', text: 'ok' }], stopReason: 'completed' as const }),
cancel() {},
dispose: async () => {},
@@ -173,6 +175,7 @@ describe('dsh-tool-subagent', () => {
seen = request
return {
id: AgentId('bare-child'),
started: Promise.resolve(),
result: Promise.resolve({ output: [{ type: 'text', text: 'ok' }], stopReason: 'completed' as const }),
cancel() {},
dispose: async () => {},
@@ -301,6 +304,7 @@ describe('dsh-tool-subagent', () => {
inheritsParentContext: false,
start: () => ({
id: AgentId('spy-child'),
started: Promise.resolve(),
result: Promise.resolve({ output: [{ type: 'text', text: 'ok' }], stopReason: 'completed' as const }),
cancel() {},
dispose: async () => void disposed(),
@@ -324,6 +328,7 @@ describe('dsh-tool-subagent', () => {
inheritsParentContext: false,
start: () => ({
id: AgentId('spy-child'),
started: Promise.resolve(),
result: Promise.resolve({ output: [], stopReason: 'error' as const }),
cancel() {},
dispose: async () => void disposed(),
@@ -351,6 +356,7 @@ describe('dsh-tool-subagent', () => {
const result = new Promise<{ output: never[]; stopReason: 'aborted' }>((res) => { resolveResult = res })
return {
id: AgentId('spy-child'),
started: Promise.resolve(),
result,
cancel: () => {
cancelled()
@@ -398,6 +404,7 @@ describe('dsh-tool-subagent', () => {
const result = new Promise<{ output: never[]; stopReason: 'aborted' }>((res) => { resolveResult = res })
return {
id: AgentId('spy-child'),
started: Promise.resolve(),
result,
cancel: () => {
cancelled()
@@ -466,6 +473,7 @@ describe('dsh-tool-subagent', () => {
seen = request
return {
id: AgentId('capture2-child'),
started: Promise.resolve(),
result: Promise.resolve({ output: [{ type: 'text', text: 'ok' }], stopReason: 'completed' as const }),
cancel() {},
dispose: async () => {},
@@ -499,6 +507,7 @@ describe('dsh-tool-subagent', () => {
seen = request
return {
id: AgentId('capture3-child'),
started: Promise.resolve(),
result: Promise.resolve({ output: [{ type: 'text', text: 'ok' }], stopReason: 'completed' as const }),
cancel() {},
dispose: async () => {},
@@ -529,6 +538,7 @@ describe('dsh-tool-subagent', () => {
seen = request
return {
id: AgentId('capture4-child'),
started: Promise.resolve(),
result: Promise.resolve({ output: [{ type: 'text', text: 'ok' }], stopReason: 'completed' as const }),
cancel() {},
dispose: async () => {},

3
packages/support/subagent-mock/src/index.ts
View File

@@ -66,6 +66,9 @@ class MockSubagentProvider implements SubagentProvider {
return {
id,
// A scripted run has no asynchronous publication phase; it is ready as
// soon as the provider returns the handle.
started: Promise.resolve(),
result: Promise.resolve().then(resultFor),
cancel() {
cancelled = true

2
packages/ui/user-approval/README.md
View File

@@ -2,7 +2,7 @@
User-approval seam. Owns the `ctx.approval` service ([`ApprovalService`](src/index.ts)) and the one-shot permission vocabulary the harness shares: `ApprovalRequest` (agent + tool identity + reason + abort signal), the closed `ApprovalOutcome` union (`allowed-once` / `rejected` / `cancelled` / `unavailable`), the `ApprovalRequestId` brand pairing the two log-only audit events (`approval/asked` / `approval/decided`), and the `approval/request` waterfall the answerers listen on. It lives in the UI group because its purpose is human permission, while remaining channel-neutral: it depends only on Cordis and core vocabulary packages, never on a concrete UI.
The contract in one line: `ctx.approval.request(req)` puts exactly one question — "may this specific action proceed?" — to whatever answerers the deployment composed, and always resolves to an outcome, never rejects: an aborted signal yields `cancelled`, a throwing or missing answerer yields `unavailable`, and `allowed-once` is a grant for the single asked-about action, never a class of future ones. The one precondition: ask from inside an open turn — the audit pair is turn-enclosed by contract (the turn is the durable log's commit/replay boundary; a bare event between turns is crash-tail garbage on reload), so an idle ask throws before appending anything.
The contract in one line: `ctx.approval.request(req)` puts exactly one question — "may this specific action proceed?" — to whatever answerers the deployment composed, and always resolves to an outcome, never rejects: an aborted signal yields `cancelled`, a throwing or missing answerer yields `unavailable`, and `allowed-once` is a grant for the single asked-about action, never a class of future ones. Acceptance is synchronous: the service shallow-freezes a detached request record before dispatch, preserving the exact `agent` and `AbortSignal` identities while making later caller mutation unable to redirect scope, payload, cancellation, or either audit event. Session observers run after an event enters the append-only log; if one throws, the service recognizes that the audit is already authoritative, contains the observer failure, and completes the pair. The one precondition: ask from inside an open turn — the audit pair is turn-enclosed by contract (the turn is the durable log's commit/replay boundary; a bare event between turns is crash-tail garbage on reload), so an idle ask throws before appending anything.
The service is the mechanism, answerers are the policy. Answerers are `approval/request` waterfall listeners occupying a single decision slot: answer for an agent you own by returning an outcome without calling `next()`, or delegate an agent you don't recognize by calling `next()` — the chain's built-in default is `unavailable`, so a deployment with no answerer (headless, CI) fails closed with zero configuration. Dispatch is keyed by `req.agent`: a listener registered through `agent.ctx` receives only that agent's questions, while a plain-context listener receives every agent's. Registration order across sibling plugins is not load-order deterministic; compose one terminal answerer per deployment and use `prepend` listeners only for decide-or-delegate gates.

85
packages/ui/user-approval/src/index.ts
View File

@@ -63,7 +63,10 @@ declare module 'cordis' {
* Scope-filtered dispatch (`@deepseek-ai/dsh-scope`) keys the carrier by `req.agent`: a
* listener registered through `agent.ctx` receives only that agent's
* questions, while a plain-context listener receives every agent's.
* @param req - the pending decision (agent, tool identity, reason, signal).
* `req` is the service's shallow-frozen acceptance snapshot: later caller
* mutation cannot redirect the question, while the `agent` and `signal`
* identity capabilities remain exact.
* @param req - the accepted decision (agent, tool identity, reason, signal).
* @mode waterfall
*/
'approval/request'(this: Scoped<ApprovalService>, req: ApprovalRequest, next: () => Promise<ApprovalOutcome>): Promise<ApprovalOutcome>
@@ -234,7 +237,9 @@ export function setApprovalPolicy(session: Session, policy: ApprovalPolicy): voi
* for an answerer to present it and for the audit events to reconstruct what
* was asked — it deliberately does NOT carry tool arguments: a UI answerer
* attaches the prompt to the already-streamed tool call via `callId` instead
* of re-rendering the call.
* of re-rendering the call. `request()` synchronously copies and shallow-freezes
* this record before crossing an asynchronous boundary. Scalar fields are
* detached; the `agent` and `signal` identity capabilities are preserved.
*/
export interface ApprovalRequest {
/**
@@ -364,14 +369,36 @@ export class ApprovalService extends Service {
* Within that precondition it always resolves to an outcome, never rejects:
* an aborted signal yields `'cancelled'`, a missing or throwing answerer
* yields `'unavailable'` (fail closed), and a rogue non-vocabulary return
* value is normalized to `'unavailable'`. Appends the
* value is normalized to `'unavailable'`. The caller-owned request is
* synchronously snapshotted, so later mutation cannot split routing,
* dispatch payload, cancellation, or the audit pair across agents/sessions.
* Appends the
* `approval/asked`/`approval/decided` audit pair (log-only) around the
* decision regardless of outcome.
* decision regardless of outcome. A synchronous session observer failure
* after an audit event entered the append-only log is contained; the event
* is already authoritative, so the pair still completes and the request
* still resolves.
* @param req - the pending decision (agent, tool identity, reason, signal).
* @returns the closed outcome; `'allowed-once'` is the only grant.
*/
async request(req: ApprovalRequest): Promise<ApprovalOutcome> {
if (!hasOpenTurn(req.agent.session.events)) {
// Accept one immutable request shape before the first async boundary. The
// caller retains its record and may mutate it as soon as this async method
// returns; identity capabilities stay live, but the record is never reread.
const agent = req.agent
const toolName = req.toolName
const callId = req.callId
const reason = req.reason
const signal = req.signal
const accepted: Readonly<ApprovalRequest> = Object.freeze({
agent,
toolName,
...callId !== undefined ? { callId } : {},
...reason !== undefined ? { reason } : {},
...signal !== undefined ? { signal } : {},
})
const session = accepted.agent.session
if (!hasOpenTurn(session.events)) {
throw new Error(
'approval.request() outside an open turn: the approval/asked + approval/decided audit pair '
+ 'must be turn-enclosed (a bare event between turns is crash-tail garbage on reload). '
@@ -379,17 +406,47 @@ export class ApprovalService extends Service {
)
}
const id = ApprovalRequestId(randomUUID())
req.agent.session.append('approval/asked', {
id,
toolName: req.toolName,
...req.callId !== undefined ? { callId: req.callId } : {},
...req.reason !== undefined ? { reason: req.reason } : {},
this.appendAudit(session, 'approval/asked', id, () => {
session.append('approval/asked', {
id,
toolName: accepted.toolName,
...accepted.callId !== undefined ? { callId: accepted.callId } : {},
...accepted.reason !== undefined ? { reason: accepted.reason } : {},
})
})
const outcome = await this.decide(accepted)
this.appendAudit(session, 'approval/decided', id, () => {
session.append('approval/decided', { id, outcome })
})
const outcome = await this.decide(req)
req.agent.session.append('approval/decided', { id, outcome })
return outcome
}
/**
* Append one audit event while distinguishing a post-append observer throw
* from a failure that prevented the event entering the log. `Session.append`
* pushes first and then notifies synchronously, so log growth proves the
* event is already authoritative; that observer failure is reported and
* contained so it cannot reject the approval or suppress its matching event.
* @param session - the captured session receiving both audit events.
* @param type - the audit event currently being appended.
* @param id - the request id, used to identify the contained failure.
* @param append - the single concrete `Session.append` call.
*/
private appendAudit(
session: Session,
type: 'approval/asked' | 'approval/decided',
id: ApprovalRequestId,
append: () => void,
): void {
const length = session.events.length
try {
append()
} catch (error) {
if (session.events.length === length) throw error
this.ctx.logger.warn(`approval request "${id}": ${type} observer threw after the event was appended`)
}
}
/**
* The session's effective policy: its own `approval/policy` fold, else the
* configured default (the schema already defaulted an omitted policy to
@@ -401,8 +458,8 @@ export class ApprovalService extends Service {
return effectiveApprovalPolicy(agent.session.events) ?? this.config.policy ?? 'ask'
}
/** Dispatch the waterfall, contained and raced against `req.signal`. */
private async decide(req: ApprovalRequest): Promise<ApprovalOutcome> {
/** Dispatch the waterfall, contained and raced against the accepted signal. */
private async decide(req: Readonly<ApprovalRequest>): Promise<ApprovalOutcome> {
if (req.signal?.aborted) return 'cancelled'
// The 'never' policy is decided HERE, before any dispatch: a listener
// registered with `prepend: true` after this service mounts would sit

134
packages/ui/user-approval/tests/approval.spec.ts
View File

@@ -3,7 +3,7 @@ import { Context } from 'cordis'
import type { Agent } from '@deepseek-ai/dsh-agent'
import { CallId } from '@deepseek-ai/dsh-llm'
import { carrierKeyOf, scopeHost } from '@deepseek-ai/dsh-scope'
import { Session, SessionId } from '@deepseek-ai/dsh-session'
import SessionStore, { Session, SessionId } from '@deepseek-ai/dsh-session'
import type { SessionEvent } from '@deepseek-ai/dsh-session'
import SystemPrompt from '@deepseek-ai/dsh-system-prompt'
import ApprovalService, { ApprovalOutcome, ApprovalRequest, effectiveApprovalPolicy, setApprovalPolicy } from '@deepseek-ai/dsh-user-approval'
@@ -78,6 +78,138 @@ describe('ApprovalService.request', () => {
expect(Object.keys(appended[0]?.data ?? {}).sort()).toEqual(['id', 'toolName'])
})
it('snapshots request identity, scope, payload, and audit before deferred dispatch', async () => {
const ctx = await mounted()
const { agent: acceptedAgent, appended: acceptedAudit } = fakeAgent()
const { agent: replacementAgent, appended: replacementAudit } = fakeAgent()
const host = await scopeHost(ctx, ['approval'])
const acceptedScope = host.mint(acceptedAgent)
const replacementScope = host.mint(replacementAgent)
const dispatchStarted = Promise.withResolvers<'started'>()
const answer = Promise.withResolvers<ApprovalOutcome>()
const originalSignal = new AbortController().signal
const replacementSignal = new AbortController().signal
let heardBy: 'accepted' | 'replacement' | undefined
let received: ApprovalRequest | undefined
let carrier: unknown
acceptedScope.ctx.on('approval/request', function (req) {
heardBy = 'accepted'
received = req
carrier = carrierKeyOf(this)
dispatchStarted.resolve('started')
return answer.promise
})
replacementScope.ctx.on('approval/request', function (req) {
heardBy = 'replacement'
received = req
carrier = carrierKeyOf(this)
dispatchStarted.resolve('started')
return answer.promise
})
const request = requestOf(acceptedAgent, {
toolName: 'original-tool',
callId: CallId('original-call'),
reason: 'original reason',
signal: originalSignal,
})
const pending = ctx.approval.request(request)
// request() has returned, but the answerer dispatch is deliberately queued
// in a microtask. Mutating the caller-owned record must not redirect it.
request.agent = replacementAgent
request.toolName = 'mutated-before-dispatch'
request.callId = CallId('mutated-call')
request.reason = 'mutated reason'
request.signal = replacementSignal
await dispatchStarted.promise
// Mutation while the answer is pending must not redirect the final audit.
request.toolName = 'mutated-after-dispatch'
request.reason = 'mutated again'
answer.resolve('allowed-once')
await expect(pending).resolves.toBe('allowed-once')
expect(heardBy).toBe('accepted')
expect(carrier).toBe(acceptedAgent)
expect(received).not.toBe(request)
expect(Object.isFrozen(received)).toBe(true)
expect(received).toMatchObject({
agent: acceptedAgent,
toolName: 'original-tool',
callId: 'original-call',
reason: 'original reason',
signal: originalSignal,
})
expect(acceptedAudit).toHaveLength(2)
expect(acceptedAudit[0]?.data).toMatchObject({
toolName: 'original-tool',
callId: 'original-call',
reason: 'original reason',
})
expect(acceptedAudit[1]?.data).toMatchObject({ outcome: 'allowed-once' })
expect(acceptedAudit[1]?.data['id']).toBe(acceptedAudit[0]?.data['id'])
expect(replacementAudit).toEqual([])
await host.dispose()
})
it('contains an approval/asked observer throw after append and still completes the pair', async () => {
const ctx = new Context()
await ctx.plugin(SessionStore)
await ctx.plugin(ApprovalService)
const session = ctx.sessions.create(SessionId('asked-observer-throw'))
session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } })
const agent = { session } as unknown as Agent
const warn = vi.spyOn(ctx.logger, 'warn').mockImplementation(() => {})
ctx.on('session/event', (_session, event) => {
if (event.type === 'approval/asked') throw new Error('observer failed after asked append')
})
ctx.on('approval/request', () => Promise.resolve<ApprovalOutcome>('allowed-once'))
await expect(ctx.approval.request(requestOf(agent))).resolves.toBe('allowed-once')
const audit = session.events.filter(event => event.type.startsWith('approval/'))
const asked = session.events.find((event): event is SessionEvent<'approval/asked'> => event.type === 'approval/asked')
const decided = session.events.find((event): event is SessionEvent<'approval/decided'> => event.type === 'approval/decided')
expect(audit.map(event => event.type)).toEqual(['approval/asked', 'approval/decided'])
expect(decided?.data.id).toBe(asked?.data.id)
expect(warn).toHaveBeenCalledWith(expect.stringContaining('approval/asked observer threw'))
})
it('contains an approval/decided observer throw after append and still resolves', async () => {
const ctx = new Context()
await ctx.plugin(SessionStore)
await ctx.plugin(ApprovalService)
const session = ctx.sessions.create(SessionId('decided-observer-throw'))
session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } })
const agent = { session } as unknown as Agent
const warn = vi.spyOn(ctx.logger, 'warn').mockImplementation(() => {})
ctx.on('session/event', (_session, event) => {
if (event.type === 'approval/decided') throw new Error('observer failed after decided append')
})
ctx.on('approval/request', () => Promise.resolve<ApprovalOutcome>('rejected'))
await expect(ctx.approval.request(requestOf(agent))).resolves.toBe('rejected')
const audit = session.events.filter(event => event.type.startsWith('approval/'))
const asked = session.events.find((event): event is SessionEvent<'approval/asked'> => event.type === 'approval/asked')
const decided = session.events.find((event): event is SessionEvent<'approval/decided'> => event.type === 'approval/decided')
expect(audit.map(event => event.type)).toEqual(['approval/asked', 'approval/decided'])
expect(decided?.data).toMatchObject({ id: asked?.data.id, outcome: 'rejected' })
expect(warn).toHaveBeenCalledWith(expect.stringContaining('approval/decided observer threw'))
})
it('does not misclassify a pre-append failure as an observer failure', async () => {
const ctx = await mounted()
const failure = new Error('append failed before log growth')
const agent = {
session: {
events: [{ type: 'turn/start' }],
append: () => { throw failure },
},
} as unknown as Agent
await expect(ctx.approval.request(requestOf(agent))).rejects.toBe(failure)
})
it('returns the first answering listener outcome (single decision slot)', async () => {
const ctx = await mounted()
const { agent } = fakeAgent()

7
packages/workflow/workflow-workerthread/tests/workflow-workerthread.spec.ts
View File

@@ -83,6 +83,7 @@ class StubProvider implements SubagentProvider {
}
return {
id: AgentId(`stub-child-${index}`),
started: Promise.resolve(),
result,
cancel: (reason?: string) => {
controlled.cancelled = reason ?? 'cancelled'
@@ -222,6 +223,7 @@ describe('dsh-workflow-workerthread', () => {
inheritsParentContext: false,
start: () => ({
id: AgentId('reject-child'),
started: Promise.resolve(),
result: Promise.reject(new Error('backend exploded')),
cancel: () => { /* nothing in flight */ },
dispose: () => Promise.resolve(),
@@ -245,6 +247,7 @@ describe('dsh-workflow-workerthread', () => {
inheritsParentContext: false,
start: () => ({
id: AgentId('bad-dispose-child'),
started: Promise.resolve(),
result: Promise.resolve({ output: [{ type: 'text', text: 'fine' }], stopReason: 'completed' }),
cancel: () => { /* settled already */ },
dispose: () => Promise.reject(new Error('dispose exploded')),
@@ -266,6 +269,7 @@ describe('dsh-workflow-workerthread', () => {
inheritsParentContext: false,
start: () => ({
id: AgentId('trap-child'),
started: Promise.resolve(),
result: Promise.resolve({ output: [{ type: 'text', text: 'fine' }], stopReason: 'completed' }),
cancel: () => { /* settled already */ },
// The rejection VALUE's own coercion throws: a warn built with bare
@@ -530,6 +534,7 @@ describe('dsh-workflow-workerthread', () => {
}, { once: true })
return {
id: AgentId('signal-only-child'),
started: Promise.resolve(),
result,
// The seam leaves a provider free to honor EITHER cancel channel;
// this one deliberately ignores run.cancel() — only the request
@@ -572,6 +577,7 @@ describe('dsh-workflow-workerthread', () => {
starts += 1
return {
id: AgentId('cancel-only-child'),
started: Promise.resolve(),
result: new Promise(() => { /* only cancel() ends this child */ }),
// Deliberately ignores the request signal — the seam leaves a
// provider free to honor ONLY the explicit cancel() channel.
@@ -749,6 +755,7 @@ describe('dsh-workflow-workerthread', () => {
inheritsParentContext: false,
start: () => ({
id: AgentId('doomed-child'),
started: Promise.resolve(),
result: new Promise(() => { /* never settles; the reap is the teardown */ }),
cancel: (reason?: string) => { cancelled.push(reason ?? 'cancelled') },
dispose: () => Promise.reject(new Error('dispose exploded during reap')),