From 0e0f3b2f19f9768b2518fca636f882c13bbc60c8 Mon Sep 17 00:00:00 2001 From: Tianyi Cui <53024+tianyicui@users.noreply.github.com> Date: Tue, 7 Jul 2026 21:09:02 +0800 Subject: [PATCH] review: acquire the structured runtime per run, not per backend MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The Codex simplification concern plus the duplication comment on the spawn apply, resolved by deletion: the backend-lifetime holds are gone, so the runtime registers at the first structured run and disposes when the last settles — a deployment that never passes outputSchema carries no always-on global state, and there is no per-backend acquisition block left to extract. The driver spec now drives an INLINE spawn-shaped provider over startInProcessRun, which removes the spawn/fork devDependencies (the test-only workspace cycle); plugin-level structured coverage moves to the backends' own specs (capture through the shipped plugin, mid-run backend unload, seeded fork capture). tools.md, the driver README, and both backend READMEs describe the run-scoped lifetime; the module-graph regenerates without the cycle edges. --- packages/subagent/subagent-fork/src/index.ts | 15 ++--- .../subagent-fork/tests/subagent-fork.spec.ts | 29 +++++++--- .../subagent/subagent-inprocess/README.md | 14 +++-- .../subagent/subagent-inprocess/package.json | 2 - packages/subagent/subagent-spawn/README.md | 2 +- packages/subagent/subagent-spawn/src/index.ts | 20 ++----- .../tests/subagent-spawn.spec.ts | 58 ++++++++++++++++--- pnpm-lock.yaml | 6 -- 8 files changed, 91 insertions(+), 55 deletions(-) diff --git a/packages/subagent/subagent-fork/src/index.ts b/packages/subagent/subagent-fork/src/index.ts index 4ee28001d2..8f91186cf0 100644 --- a/packages/subagent/subagent-fork/src/index.ts +++ b/packages/subagent/subagent-fork/src/index.ts @@ -25,13 +25,13 @@ import z from 'schemastery' import type { SessionEvent } from '@deepseek-ai/dsh-session' import type { Agent } from '@deepseek-ai/dsh-agent' import type { SubagentCapabilities, SubagentProvider, SubagentStartRequest } from '@deepseek-ai/dsh-subagent' -import { acquireStructuredRuntime, startInProcessRun } from '@deepseek-ai/dsh-subagent-inprocess' +import { startInProcessRun } from '@deepseek-ai/dsh-subagent-inprocess' export const name = 'subagent-fork' // `tools` is deliberately NOT injected — same rationale as subagent-spawn: the -// structured runtime gates its capture-tool registration on `tools` itself, so -// this backend's apply timing (and the delegation tool's position in the -// model-visible tool list) is unchanged by structured output. +// per-run structured runtime gates its capture-tool registration on `tools` +// itself, so this backend's apply timing (and the delegation tool's position +// in the model-visible tool list) is unchanged by structured output. export const inject = ['subagents', 'agents'] /** Config: the registry name to register the provider under. */ @@ -84,12 +84,5 @@ class ForkProvider implements SubagentProvider { } export function apply(ctx: Context, config: Config): void { - // Hold the structured runtime for the plugin's lifetime (see the spawn - // backend — same two-level lifetime: backends for availability, runs for - // mid-run survival across a backend unload). - ctx.effect(() => { - const acquisition = acquireStructuredRuntime(ctx) - return () => { acquisition.release() } - }, 'subagent-fork structured runtime') ctx.subagents.registerProvider(new ForkProvider(config.providerName, ctx)) } diff --git a/packages/subagent/subagent-fork/tests/subagent-fork.spec.ts b/packages/subagent/subagent-fork/tests/subagent-fork.spec.ts index 74974942b5..90e2d583d8 100644 --- a/packages/subagent/subagent-fork/tests/subagent-fork.spec.ts +++ b/packages/subagent/subagent-fork/tests/subagent-fork.spec.ts @@ -9,9 +9,10 @@ import AgentRegistry, { AgentId } from '@deepseek-ai/dsh-agent' import AgentLoop from '@deepseek-ai/dsh-agent-loop' import * as Invariants from '@deepseek-ai/dsh-invariants' import SubagentService from '@deepseek-ai/dsh-subagent' -import { MockAdapter, textResponse } from '../../../core/agent-loop/tests/mock-adapter.ts' +import { MockAdapter, textResponse, toolCallResponse } from '../../../core/agent-loop/tests/mock-adapter.ts' import type { StreamChunk } from '@deepseek-ai/dsh-llm' import * as fork from '../src/index.ts' +import { STRUCTURED_OUTPUT_TOOL } from '@deepseek-ai/dsh-subagent-inprocess' import { completedTurnPrefix } from '../src/index.ts' type Script = ConstructorParameters[0] @@ -141,6 +142,26 @@ describe('dsh-subagent-fork', () => { await run.dispose() }) + it('captures structured output through the shipped plugin (seeded child, driver runtime)', async () => { + const { ctx, parent } = await setup([ + textResponse('parent turn'), + toolCallResponse('c1', STRUCTURED_OUTPUT_TOOL, { answer: 9 }), + ]) + parent.send([{ type: 'text', text: 'warm up' }]) + await parent.whenIdle() + const run = ctx.subagents.start('fork', { + prompt: [{ type: 'text', text: 'report structured' }], + parent, + outputSchema: { type: 'object', properties: { answer: { type: 'number' } }, required: ['answer'] }, + }) + const result = await run.result + expect(result.stopReason).toBe('completed') + expect(result.structured).toEqual({ answer: 9 }) + // Run-scoped runtime: nothing stays registered after the settle. + expect(ctx.tools.get(STRUCTURED_OUTPUT_TOOL)).toBeUndefined() + await run.dispose() + }) + it('does NOT return the seeded parent output when the child produces no message of its own', async () => { // Regression: readResult must scope to the child's OWN events (after the // seed). The parent completes a turn with a distinctive assistant message, @@ -170,12 +191,6 @@ describe('dsh-subagent-fork', () => { const ctx = new Context() await ctx.plugin(SubagentService) await ctx.plugin(AgentRegistry) - // The backend does NOT inject 'tools' (the structured runtime gates its - // capture-tool registration on tools availability itself, keeping backend - // apply timing — and the delegation tool's prompt position — unchanged); - // the registries are loaded here so the runtime registers eagerly anyway. - await ctx.plugin(SystemPrompt) - await ctx.plugin(ToolRegistry) const fiber = await ctx.plugin(fork, { providerName: 'fork' }) expect(ctx.subagents.list()).toEqual(['fork']) await fiber.dispose() diff --git a/packages/subagent/subagent-inprocess/README.md b/packages/subagent/subagent-inprocess/README.md index a4fcc51fbc..f6870929a8 100644 --- a/packages/subagent/subagent-inprocess/README.md +++ b/packages/subagent/subagent-inprocess/README.md @@ -8,7 +8,7 @@ The shared **in-process subagent run driver**. A pure library (no provider, no r Runs a child as a child [`Agent`](../../core/agent) on the same cordis context (`ctx.agents`): -1. computes child depth = `depthOf(parent) + 1`; if `request.maxDepth` is set and exceeded, throws `SubagentDepthError` (the `depthLimit` capability); a `request.outputSchema` is asserted against the supported subset (`assertSupportedOutputSchema` from [dsh-tools](../../core/tools/README.md)) before any child exists; +1. computes child depth = `depthOf(parent) + 1`; if `request.maxDepth` is set and exceeded, throws `SubagentDepthError` (the `depthLimit` capability); a `request.outputSchema` is asserted against the supported subset (`assertSupportedOutputSchema` from [dsh-tools](../../core/tools/README.md)) and then snapshotted with `structuredClone` before any child exists — assertion first so a hostile value fails as `OutputSchemaError` (never a raw clone error), the snapshot so a post-`start()` caller mutation cannot drift the enforced schema; 2. creates a child via `ctx.agents.create` with a fresh `AgentId`/`SessionId`, the parent's `cwd` + `parentSession` lineage, the optional `options.seed` (fork's completed-turn prefix; omitted for a fresh child), and `agentOptions` (the child inherits the **parent's model** by default — a child with no model can't run — overridable via `request.agentOptions.model`; the deployment persona needs no inheritance — it is a context-wide prompt section); 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). @@ -19,16 +19,18 @@ Runs a child as a child [`Agent`](../../core/agent) on the same cordis context ( `{ providerName: string; seed?: SessionEvent[] }` — the per-backend inputs: the provider name (for error context) and the optional child-session seed. -### Structured output: `acquireStructuredRuntime(ctx): StructuredAcquisition` +### Structured output (package-internal runtime) -The mechanism behind `outputSchema` for in-process children. One globally registered `structured_output` capture tool (its registered parameters are a placeholder) plus two listeners, registered once per root context and shared by every holder: +The mechanism behind `outputSchema` for in-process children — acquired per structured RUN inside `startInProcessRun` (nothing is registered on a context that never runs a structured child; only the model-facing constants `STRUCTURED_OUTPUT_TOOL`/`STRUCTURED_OUTPUT_INSTRUCTION` are exported). One globally registered `structured_output` capture tool (its registered parameters are a placeholder) plus four listeners: -- an `agent/request` waterfall listener registered `prepend: true` that post-processes `await next()` — **final-request enforcement**: the request that hits the wire never carries `structured_output` for an agent without a structured run, and for one that has it always carries the run's OWN schema (as the tool's `parameters`) plus the calling instruction appended to its `system` text (the demand travels with the tool — `AgentOptions` has no per-agent prompt field to carry it). Per-agent shaping lives here because the tool registry and prompt assembly are context-global while schemas differ per concurrent child; cooperative mutate-then-`next()` would not survive a downstream listener returning a replacement request. +- a `system-prompt/assemble` waterfall listener registered `prepend: true` that post-processes `await next()` — **final-assembly enforcement**: the assembly the loop renders never carries `structured_output` for an agent without a structured run, and for one that has it always carries the run's OWN schema (as the tool's `parameters`) plus the calling instruction as a trailing prompt section (the demand travels with the tool — `AgentOptions` has no per-agent prompt field to carry it). The loop logs the rendered assembly as the step's `request/header`, so the injection is reconstructable log state, never a wire-only mutation. Per-agent shaping lives here because the tool registry and prompt assembly are context-global while schemas differ per concurrent child (FIXME in the module doc: per-agent/per-session scoping would dissolve this); cooperative mutate-then-`next()` would not survive a downstream listener returning a replacement assembly. +- a `tools/post-execute` listener (`prepend: true` = outermost, so `await next()` yields the composed final decision) that COMMITS the capture: the tool body only stages the validated value, and it becomes the run's result only when the final decision accepts the call — a downstream block (a PostToolUse hook) turns the logged result into `isError`, and the run must not report `structured` success for a call the model and session log saw fail. +- a `tools/pre-execute` deny for any call arriving after the agent's capture — terminal means terminal WITHIN the step: a response listing `structured_output` before further tool calls cannot run side effects after the final answer was accepted. - an `agent/turn-continuation` listener (also `prepend: true` — an earlier-registered force-continue listener returning without `next()` must not decide the turn before the veto runs) that stops a child's turn once its output is captured, so a successful capture doesn't buy a wasted extra model step. -The capture tool validates each call against the run's schema (`validateStructuredValue`) — violations become an `INVALID_ARGS` isError result the model retries in-turn; a valid call records the value. +The capture tool validates each call against the run's schema (`validateStructuredValue`) — violations become an `INVALID_ARGS` isError result the model retries in-turn; a valid call stages the value for the post-execute commit. -Lifetime is refcounted with two kinds of holder: each backend acquires for its plugin lifetime (`apply`), and each structured RUN holds its own acquisition from start to settle — so unregistration can never precede a live run's settle, and the runtime disposes only when the last backend AND the last run are gone. `release()` is idempotent per acquisition. +Lifetime is refcounted by live structured runs: each acquires at start and releases at settle, so the registrations exist exactly while at least one structured child is live, a backend hot-reload mid-run cannot unregister the capture tool under a live child, and the last settle disposes everything. `release()` is idempotent per acquisition. ### `depthOf(agent): number` diff --git a/packages/subagent/subagent-inprocess/package.json b/packages/subagent/subagent-inprocess/package.json index ecc177162f..4e6b72533a 100644 --- a/packages/subagent/subagent-inprocess/package.json +++ b/packages/subagent/subagent-inprocess/package.json @@ -37,8 +37,6 @@ "@deepseek-ai/dsh-llm": "workspace:^", "@deepseek-ai/dsh-session": "workspace:^", "@deepseek-ai/dsh-subagent": "workspace:^", - "@deepseek-ai/dsh-subagent-fork": "workspace:^", - "@deepseek-ai/dsh-subagent-spawn": "workspace:^", "@deepseek-ai/dsh-system-prompt": "workspace:^", "@deepseek-ai/dsh-tools": "workspace:^", "cordis": "^4.0.0-rc.6" diff --git a/packages/subagent/subagent-spawn/README.md b/packages/subagent/subagent-spawn/README.md index 696007a693..b976ef5a63 100644 --- a/packages/subagent/subagent-spawn/README.md +++ b/packages/subagent/subagent-spawn/README.md @@ -10,7 +10,7 @@ The run mechanics live in the shared [`@deepseek-ai/dsh-subagent-inprocess`](../ ## Capabilities -`{ outputSchema: true, depthLimit: true, toolFilter: false }`. It constructs the child, so it enforces a recursion cap, and it supports structured output via the driver's shared [structured runtime](../subagent-inprocess/README.md) (the backend acquires it for its plugin lifetime; each structured run holds its own acquisition until it settles). Tool-scoping is deferred (the service rejects a request needing it before `start` runs). +`{ outputSchema: true, depthLimit: true, toolFilter: false }`. It constructs the child, so it enforces a recursion cap, and it supports structured output via the driver's [structured runtime](../subagent-inprocess/README.md) (acquired per structured run inside the driver — this backend registers nothing at apply). Tool-scoping is deferred (the service rejects a request needing it before `start` runs). ## Config diff --git a/packages/subagent/subagent-spawn/src/index.ts b/packages/subagent/subagent-spawn/src/index.ts index 7da954bc44..2d8f118b4e 100644 --- a/packages/subagent/subagent-spawn/src/index.ts +++ b/packages/subagent/subagent-spawn/src/index.ts @@ -22,14 +22,14 @@ import type { Context } from 'cordis' import z from 'schemastery' import type { SubagentCapabilities, SubagentProvider, SubagentStartRequest } from '@deepseek-ai/dsh-subagent' -import { acquireStructuredRuntime, startInProcessRun } from '@deepseek-ai/dsh-subagent-inprocess' +import { startInProcessRun } from '@deepseek-ai/dsh-subagent-inprocess' export const name = 'subagent-spawn' -// `tools` is deliberately NOT injected: the structured runtime gates its own -// capture-tool registration on `tools` availability internally, so this -// backend's apply timing — and with it the provider-mirroring delegation -// tool's position in the model-visible tool list — stays what it was before -// structured output existed. +// `tools` is deliberately NOT injected: the shared driver's structured runtime +// (acquired per structured RUN, not at apply) gates its own capture-tool +// registration on `tools` availability, so this backend's apply timing — and +// with it the provider-mirroring delegation tool's position in the +// model-visible tool list — stays what it was before structured output existed. export const inject = ['subagents', 'agents'] /** Config: the registry name to register the provider under. */ @@ -64,13 +64,5 @@ class SpawnProvider implements SubagentProvider { } export function apply(ctx: Context, config: Config): void { - // Hold the structured runtime for the plugin's lifetime, so the capture tool - // and its request-shaping listeners are registered before the first - // structured run and torn down when the last backend unloads (live runs hold - // their own acquisitions, so an unload mid-run cannot strand a child). - ctx.effect(() => { - const acquisition = acquireStructuredRuntime(ctx) - return () => { acquisition.release() } - }, 'subagent-spawn structured runtime') ctx.subagents.registerProvider(new SpawnProvider(config.providerName, ctx)) } diff --git a/packages/subagent/subagent-spawn/tests/subagent-spawn.spec.ts b/packages/subagent/subagent-spawn/tests/subagent-spawn.spec.ts index 63d26c0531..1dad9748e9 100644 --- a/packages/subagent/subagent-spawn/tests/subagent-spawn.spec.ts +++ b/packages/subagent/subagent-spawn/tests/subagent-spawn.spec.ts @@ -10,9 +10,9 @@ import { SessionId } from '@deepseek-ai/dsh-session' import AgentLoop from '@deepseek-ai/dsh-agent-loop' import * as Invariants from '@deepseek-ai/dsh-invariants' import SubagentService from '@deepseek-ai/dsh-subagent' -import { MockAdapter, maxTokensResponse, textResponse } from '../../../core/agent-loop/tests/mock-adapter.ts' +import { MockAdapter, maxTokensResponse, textResponse, toolCallResponse } from '../../../core/agent-loop/tests/mock-adapter.ts' import * as spawn from '../src/index.ts' -import { depthOf, SubagentDepthError } from '@deepseek-ai/dsh-subagent-inprocess' +import { depthOf, STRUCTURED_OUTPUT_TOOL, SubagentDepthError } from '@deepseek-ai/dsh-subagent-inprocess' type Script = ConstructorParameters[0] @@ -251,18 +251,60 @@ describe('dsh-subagent-spawn', () => { const ctx = new Context() await ctx.plugin(SubagentService) await ctx.plugin(AgentRegistry) - // The backend does NOT inject 'tools' (the structured runtime gates its - // capture-tool registration on tools availability itself, keeping backend - // apply timing — and the delegation tool's prompt position — unchanged); - // the registries are loaded here so the runtime registers eagerly anyway. - await ctx.plugin(SystemPrompt) - await ctx.plugin(ToolRegistry) const fiber = await ctx.plugin(spawn, { providerName: 'spawn' }) expect(ctx.subagents.list()).toEqual(['spawn']) await fiber.dispose() expect(ctx.subagents.list()).toEqual([]) }) + it('captures structured output through the shipped plugin (driver runtime, plugin wiring)', async () => { + const { ctx, parent } = await setup([ + toolCallResponse('c1', STRUCTURED_OUTPUT_TOOL, { answer: 42 }), + ]) + const run = ctx.subagents.start('spawn', { + prompt: [{ type: 'text', text: 'produce the answer' }], + parent, + outputSchema: { type: 'object', properties: { answer: { type: 'number' } }, required: ['answer'] }, + }) + const result = await run.result + expect(result.stopReason).toBe('completed') + expect(result.structured).toEqual({ answer: 42 }) + // Run-scoped runtime: the settle released the last acquisition. + expect(ctx.tools.get(STRUCTURED_OUTPUT_TOOL)).toBeUndefined() + await run.dispose() + }) + + it('a backend unload mid-structured-run settles the run and releases the runtime', async () => { + // Rebuild the stack by hand so we hold the backend's fiber. + const ctx = new Context() + const adapter = new MockAdapter(['hang']) + await ctx.plugin(LlmService) + await ctx.plugin(SessionStore) + await ctx.plugin(SystemPrompt) + await ctx.plugin(ToolRegistry) + await ctx.plugin(AgentRegistry) + await ctx.plugin(Invariants) + await ctx.plugin(AgentLoop, { agents: [] }) + await ctx.plugin(SubagentService) + const fiber = await ctx.plugin(spawn, { providerName: 'spawn' }) + ctx.llm.registerAdapter(['mock'], adapter) + const parent = ctx.agentLoop.create(AgentId('parent'), { model: 'mock' }) + const run = ctx.subagents.start('spawn', { + prompt: [{ type: 'text', text: 'q' }], + parent, + outputSchema: { type: 'object', properties: { a: { type: 'number' } } }, + }) + // Let the child's step start streaming, then unload the backend. The + // backend owns the child agent, so the unload tears the child down and + // the run settles — releasing its own runtime acquisition on the way out. + await new Promise(resolve => setTimeout(resolve, 30)) + await fiber.dispose() + const result = await run.result + expect(result.stopReason).toBe('error') + expect(ctx.tools.get(STRUCTURED_OUTPUT_TOOL)).toBeUndefined() + await run.dispose() + }) + it('has the namespace-plugin export shape (no stray default)', () => { expect('default' in spawn).toBe(false) expect(spawn.name).toBe('subagent-spawn') diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 35d940028b..d5fb68c746 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -655,12 +655,6 @@ importers: '@deepseek-ai/dsh-subagent': specifier: workspace:^ version: link:../subagent - '@deepseek-ai/dsh-subagent-fork': - specifier: workspace:^ - version: link:../subagent-fork - '@deepseek-ai/dsh-subagent-spawn': - specifier: workspace:^ - version: link:../subagent-spawn '@deepseek-ai/dsh-system-prompt': specifier: workspace:^ version: link:../../core/system-prompt