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

Merge branch 'structured-output-subagent-seam' into worktree-dynamic-workflows

Browse Source
This commit is contained in:
Tianyi Cui
2026-07-07 21:23:20 +08:00
parent afab88b139 23fb1febcd
commit f8517654a8
15 changed files with 399 additions and 142 deletions
Download Patch File Download Diff File Expand all files Collapse all files

35
packages/core/tools/src/json-schema.ts
View File

@@ -91,9 +91,20 @@ const ANNOTATION_KEYWORDS = new Set(['description', 'title', 'default', 'example
const SCHEMA_TYPES: readonly StructuredSchemaType[] = ['object', 'array', 'string', 'number', 'integer', 'boolean', 'null']
/** Whether a value is a non-null, non-array object (structural, realm-agnostic). */
/**
* Whether a value is a PLAIN JSON object — non-null, non-array, and with a
* prototype chain of at most one link (`null`-proto, or any realm's
* `Object.prototype`, whose own prototype is `null`). Realm-agnostic on
* purpose: a schema materialized in another realm carries THAT realm's
* `Object.prototype`, which an identity check would wrongly reject. Exotic
* hosts (`Date`, `Map`, class instances) have longer chains and are rejected —
* they would serialize lossily (`Date` → string, `Map` → `{}`) instead of
* failing loud.
*/
function isObjectLike(value: unknown): value is Record<string, unknown> {
return typeof value === 'object' && value !== null && !Array.isArray(value)
if (typeof value !== 'object' || value === null || Array.isArray(value)) return false
const proto: unknown = Object.getPrototypeOf(value)
return proto === null || Object.getPrototypeOf(proto) === null
}
/** Whether a value is a supported scalar (`enum`/`const` member): string, finite number, boolean, or null. */
@@ -116,6 +127,9 @@ function isJsonData(value: unknown, seen: Set<object>): boolean {
seen.add(value)
try {
if (Array.isArray(value)) return value.every(entry => isJsonData(entry, seen))
// A non-plain object (Date, Map, class instance) is NOT JSON data even when
// it has no enumerable values — it would serialize lossily, not loudly.
if (!isObjectLike(value)) return false
return Object.values(value).every(entry => isJsonData(entry, seen))
} finally {
seen.delete(value)
@@ -194,8 +208,11 @@ function checkSchemaNode(node: unknown, path: string, violations: string[], seen
violations.push(`${path}.required must be an array of strings`)
} else {
const declared = isObjectLike(properties) ? properties : {}
for (const key of required) {
if (!(key in declared)) violations.push(`${path}.required names "${key}" which is not in properties`)
// The guard above proved every entry is a string.
for (const key of required as string[]) {
// Own-property check: `in` would let inherited names (`toString`)
// satisfy the declared-in-properties contract via the prototype.
if (!Object.hasOwn(declared, key)) violations.push(`${path}.required names "${key}" which is not in properties`)
}
}
}
@@ -256,16 +273,20 @@ function checkValue(node: StructuredSchemaNode, value: unknown, path: string): s
if (!isObjectLike(value)) return [`"${path}" must be an object`]
const violations: string[] = []
const properties = node.properties ?? {}
// Own-property discipline throughout: JSON carries own enumerable
// properties only, so an inherited `toString` must not satisfy
// `required`, dodge `additionalProperties: false`, or be validated as if
// the value carried it.
for (const key of node.required ?? []) {
if (value[key] === undefined) violations.push(`missing required property "${path}.${key}"`)
if (!Object.hasOwn(value, key) || value[key] === undefined) violations.push(`missing required property "${path}.${key}"`)
}
for (const [key, child] of Object.entries(properties)) {
if (value[key] === undefined) continue
if (!Object.hasOwn(value, key) || value[key] === undefined) continue
violations.push(...checkValue(child, value[key], `${path}.${key}`))
}
if (node.additionalProperties === false) {
for (const key of Object.keys(value)) {
if (!(key in properties)) violations.push(`"${path}.${key}" is not a declared property (additionalProperties: false)`)
if (!Object.hasOwn(properties, key)) violations.push(`"${path}.${key}" is not a declared property (additionalProperties: false)`)
}
}
return violations

50
packages/core/tools/tests/json-schema.spec.ts
View File

@@ -154,6 +154,30 @@ describe('assertSupportedOutputSchema', () => {
const leaf = { type: 'string' }
asserted({ type: 'object', properties: { a: leaf, b: leaf } })
})
it('required cannot be satisfied by INHERITED names — `toString` is not a declared property', () => {
// `'toString' in {}` is true via Object.prototype; the declared-property
// contract must be an own-property check.
expect(violationsOf({ type: 'object', properties: {}, required: ['toString'] }))
.toEqual(['schema.required names "toString" which is not in properties'])
})
it('rejects exotic host objects where the subset expects plain JSON structure', () => {
// A Map as `properties` has no own enumerable entries: structurally it
// would read as "no properties" and serialize to {} — lossy, not loud.
expect(violationsOf({ type: 'object', properties: new Map() }))
.toEqual(['schema.properties must be an object of schemas'])
// A Date node is not a schema object even though Object.values(date) is [].
expect(violationsOf({ type: 'object', properties: { at: new Date(0) } }))
.toEqual(['schema.properties.at must be a schema object'])
})
it('rejects exotic annotation payloads that would serialize lossily', () => {
expect(violationsOf({ type: 'object', default: new Date(0) }))
.toEqual(['schema.default annotation must be JSON data'])
expect(violationsOf({ type: 'object', examples: [new Map()] }))
.toEqual(['schema.examples annotation must be JSON data'])
})
})
describe('validateStructuredValue', () => {
@@ -223,6 +247,32 @@ describe('validateStructuredValue', () => {
expect(validateStructuredValue(schema, { file: undefined })).toEqual(['missing required property "value.file"'])
})
it('inherited properties satisfy nothing: required, additionalProperties, and recursion are own-property only', () => {
// required: ['toString'] must NOT be satisfied by Object.prototype.toString.
expect(validateStructuredValue(
asserted({ type: 'object', properties: { toString: { type: 'string' } }, required: ['toString'] }),
{},
)).toEqual(['missing required property "value.toString"'])
// additionalProperties: false must flag an OWN `toString` key even though
// `'toString' in properties` is true via the prototype.
expect(validateStructuredValue(
asserted({ type: 'object', additionalProperties: false }),
{ toString: 1 },
)).toEqual(['"value.toString" is not a declared property (additionalProperties: false)'])
// A declared property the value does NOT carry must not be validated
// against the value's INHERITED member (constructor is a function on
// every plain object's prototype, not a carried property).
expect(validateStructuredValue(
asserted({ type: 'object', properties: { constructor: { type: 'string' } } }),
{},
)).toEqual([])
})
it('a non-plain object value is not an object in the JSON sense', () => {
expect(validateStructuredValue(asserted({ type: 'object' }), new Date(0)))
.toEqual(['"value" must be an object'])
})
it('collects multiple violations across branches in one pass', () => {
expect(validateStructuredValue(schema, { line: 'x', severity: 'mid' })).toEqual([
'missing required property "value.file"',

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

@@ -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))
}

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

@@ -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<typeof MockAdapter>[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()

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

@@ -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`

2
packages/subagent/subagent-inprocess/package.json
View File

@@ -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"

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

@@ -25,11 +25,12 @@ import {
type StructuredAcquisition,
} from './structured.ts'
// The runtime itself (acquire/attach/release) is package-internal: runs
// acquire it inside startInProcessRun, and no other package drives it. Only
// the model-facing vocabulary is public.
export {
acquireStructuredRuntime,
STRUCTURED_OUTPUT_TOOL,
STRUCTURED_OUTPUT_INSTRUCTION,
type StructuredAcquisition,
} from './structured.ts'
declare module '@deepseek-ai/dsh-agent' {
@@ -110,15 +111,18 @@ export function startInProcessRun(
if (request.maxDepth !== undefined && childDepth > request.maxDepth) {
throw new SubagentDepthError(childDepth, request.maxDepth)
}
// Snapshot, then assert, the schema subset BEFORE any child exists (the
// Assert, then snapshot, the schema subset BEFORE any child exists (the
// service has already capability-gated; this rejects a schema outside the
// enforced subset loud). The snapshot is load-bearing: the caller keeps its
// reference, so validating and attaching the ORIGINAL would let a
// post-start() mutation drift the enforced schema away from the asserted
// one — the clone pins assertion, the model-visible parameters, and
// validateStructuredValue to the same isolation-immutable value.
// enforced subset loud). Assertion comes FIRST so a hostile value fails as
// OutputSchemaError, never as structuredClone's raw DataCloneError — the
// asserted subset is plain JSON data, which always clones. The snapshot is
// load-bearing: the caller keeps its reference, so attaching the ORIGINAL
// would let a post-start() mutation drift the enforced schema away from the
// asserted one — the clone (taken synchronously with the assertion, no
// interleaving possible) pins assertion, the model-visible parameters, and
// validateStructuredValue to one isolation-immutable value.
if (request.outputSchema !== undefined) assertSupportedOutputSchema(request.outputSchema)
const schema = request.outputSchema === undefined ? undefined : structuredClone(request.outputSchema)
if (schema !== undefined) assertSupportedOutputSchema(schema)
const childId = AgentId(randomUUID())
// The child's OWN events begin after the seed (fork seeds the parent's

64
packages/subagent/subagent-inprocess/src/structured.ts
View File

@@ -36,14 +36,20 @@
* closes the within-step window the continuation veto cannot: a
* `tools/pre-execute` deny for any call arriving after the agent's capture, so
* a response that lists `structured_output` before further tool calls cannot
* run side effects after the final answer was accepted.
* run side effects after the final answer was accepted. A fourth,
* `tools/post-execute`, is the capture COMMIT: the tool body only stages the
* validated value, and it becomes the run's captured result only when the
* final post-execute decision accepts the call — a blocking hook downstream
* yields `isError` in the log, and the run must not report success for it.
*
* Lifetime is refcounted with two kinds of holder: each backend acquires for
* its plugin lifetime (so the tool exists before any run), and each structured
* RUN acquires from start to settle (so a backend hot-reload mid-run cannot
* unregister the capture tool out from under a live child). Registrations are
* effects on the ROOT context — their natural upper bound is app teardown — and
* the refcount disposes them when the last holder releases.
* Lifetime is refcounted by structured RUNS: each acquires from start to
* settle, so the registrations exist exactly while at least one structured
* child is live — a plain deployment that never passes `outputSchema` carries
* no always-on global state, and a backend hot-reload mid-run cannot
* unregister the capture tool out from under a live child (the run holds its
* own acquisition). Registrations land on the ROOT context and the refcount
* disposes them when the last run settles; the next structured run
* re-registers them.
*
* @module @deepseek-ai/dsh-subagent-inprocess/structured
*/
@@ -53,7 +59,7 @@ import type { Agent } from '@deepseek-ai/dsh-agent'
import type { ContentBlock, ToolSchema } from '@deepseek-ai/dsh-llm'
import type { ContinuationDecision } from '@deepseek-ai/dsh-agent'
import type { AssembleContext, PromptAssembly } from '@deepseek-ai/dsh-system-prompt'
import type { PreToolDecision, ToolExecution } from '@deepseek-ai/dsh-tools'
import type { PostToolDecision, PreToolDecision, ToolExecution, ToolExecutionResult } from '@deepseek-ai/dsh-tools'
import { ToolArgsError, validateStructuredValue, type StructuredOutputSchema } from '@deepseek-ai/dsh-tools'
/** The model-facing tool name a structured child must call to finish. */
@@ -75,6 +81,15 @@ export const STRUCTURED_OUTPUT_INSTRUCTION
/** One structured run's state: the schema to enforce and the captured value, once recorded. */
interface RunState {
readonly schema: StructuredOutputSchema
/**
* A validated value awaiting the post-execute verdict on ITS OWN call. Set
* by the capture tool's body, promoted to {@link RunState.captured} only
* when the final `tools/post-execute` decision accepts the call — a
* downstream block turns the logged result into `isError`, and a value
* committed at body time would let the run report success for a call the
* model saw fail.
*/
pending?: { value: unknown }
captured?: { value: unknown }
}
@@ -106,7 +121,7 @@ export interface StructuredAcquisition {
/**
* Acquire the per-root-context structured runtime, registering the capture tool
* and the two waterfall listeners on the FIRST acquisition. See the module doc
* and the runtime's listeners on the FIRST acquisition. See the module doc
* for the enforcement and lifetime design.
* @param ctx - any context of the app; the runtime keys off `ctx.root`.
* @returns this holder's handle (attach/captured/detach + idempotent release).
@@ -179,7 +194,9 @@ function registerRuntime(root: Context, runtime: StructuredRuntime): void {
// ToolArgsError → isError result with INVALID_ARGS: the model retries
// within the same turn, exactly like a schema-validated defineTool call.
if (violations.length > 0) throw new ToolArgsError(violations)
state.captured = { value: args }
// Two-phase commit: the body only STAGES the value; the post-execute
// listener below promotes it once the final decision accepts the call.
state.pending = { value: args }
return Promise.resolve([{ type: 'text', text: 'Structured output recorded.' }])
},
})
@@ -243,6 +260,33 @@ function registerRuntime(root: Context, runtime: StructuredRuntime): void {
return next()
}, { prepend: true }))
// The capture COMMIT: promote the staged value only when the final
// post-execute decision accepts the call. The capture tool's body cannot
// decide — `tools/post-execute` runs after it, and a blocking listener (a
// PostToolUse hook) turns the logged result into `isError` feedback; a value
// committed at body time would make readResult report `structured` success
// for a call whose result the model and session log saw fail. `prepend:
// true` = outermost at registration time, so `await next()` returns the
// COMPOSED downstream decision — the same final verdict the registry maps
// onto the result. (A later-registered outer listener that blocks without
// delegating skips this commit entirely: the staged value is dropped and the
// run errors — failure-safe in the same direction.) The staging slot clears
// on every path, including a rejecting downstream listener.
runtime.disposers.push(root.on('tools/post-execute', async function (
this: unknown, exec: ToolExecution, _result: ToolExecutionResult, next: () => Promise<PostToolDecision>,
): Promise<PostToolDecision> {
const state = exec.agent ? runtime.states.get(exec.agent) : undefined
if (!state || exec.name !== STRUCTURED_OUTPUT_TOOL || state.pending === undefined) return next()
const pending = state.pending
try {
const decision = await next()
if (decision.kind === 'accept') state.captured = pending
return decision
} finally {
delete state.pending
}
}, { prepend: true }))
// Terminal means terminal WITHIN the step, not only at its end: the
// turn-continuation veto above runs after every call in the current model
// response has executed, so a response that puts `structured_output` before

186
packages/subagent/subagent-inprocess/tests/structured.spec.ts
View File

@@ -11,8 +11,7 @@ import * as Invariants from '@deepseek-ai/dsh-invariants'
import SubagentService, { type SubagentStartRequest } from '@deepseek-ai/dsh-subagent'
import type { StructuredOutputSchema } from '@deepseek-ai/dsh-tools'
import { MockAdapter, textResponse, toolCallResponse } from '../../../core/agent-loop/tests/mock-adapter.ts'
import * as spawn from '@deepseek-ai/dsh-subagent-spawn'
import * as fork from '@deepseek-ai/dsh-subagent-fork'
import { startInProcessRun } from '../src/index.ts'
import {
acquireStructuredRuntime,
STRUCTURED_OUTPUT_INSTRUCTION,
@@ -28,11 +27,14 @@ const SCHEMA: StructuredOutputSchema = {
}
/**
* Real loop + scripted mock model + the REAL spawn backend (which acquires the
* structured runtime at apply, exactly as shipped). The mock model script
* drives the child's structured_output calls.
* Real loop + scripted mock model + an INLINE spawn-shaped provider over the
* shared driver. The concrete backend plugins are deliberately NOT loaded —
* they would devDep-cycle this package (spawn/fork already depend on the
* driver), and the runtime under test is the driver's; plugin-level structured
* coverage lives in the spawn/fork specs. The mock model script drives the
* child's structured_output calls.
*/
async function setup(script: Script, options?: { withFork?: boolean }) {
async function setup(script: Script) {
const ctx = new Context()
const adapter = new MockAdapter(script)
await ctx.plugin(LlmService)
@@ -43,13 +45,15 @@ async function setup(script: Script, options?: { withFork?: boolean }) {
await ctx.plugin(Invariants)
await ctx.plugin(AgentLoop, { agents: [] })
await ctx.plugin(SubagentService)
const fiber = await ctx.plugin(spawn, { providerName: 'spawn' })
const forkFiber = options?.withFork
? await ctx.plugin(fork, { providerName: 'fork' })
: undefined
const disposeProvider = ctx.subagents.registerProvider({
name: 'spawn',
capabilities: { outputSchema: true, depthLimit: true, toolFilter: false },
inheritsParentContext: false,
start: (request: SubagentStartRequest) => startInProcessRun(ctx, request, { providerName: 'spawn' }),
})
ctx.llm.registerAdapter(['mock'], adapter)
const parent = ctx.agentLoop.create(AgentId('parent'), { model: 'mock' })
return { ctx, parent, adapter, fiber, forkFiber }
return { ctx, parent, adapter, disposeProvider }
}
function structuredRequest(parent: SubagentStartRequest['parent'], extra?: Partial<SubagentStartRequest>): SubagentStartRequest {
@@ -263,6 +267,62 @@ describe('in-process structured output', () => {
expect(ctx.agents.get(AgentId('parent'))).toBeDefined()
})
it('a schema carrying non-JSON values fails as OutputSchemaError, never as a raw clone error', async () => {
const { ctx, parent } = await setup([])
// Assertion runs BEFORE the defensive structuredClone: a function-valued
// annotation must surface as the subset violation it is, not escape as
// structuredClone's DataCloneError.
expect(() => ctx.subagents.start('spawn', structuredRequest(parent, {
outputSchema: { type: 'object', default: () => {} } as unknown as StructuredOutputSchema,
}))).toThrow(/unsupported output schema.*annotation must be JSON data/)
})
it('a post-execute BLOCK on the capture call denies the capture: log and result agree on failure', async () => {
const { ctx, parent, adapter } = await setup([
toolCallResponse('c1', STRUCTURED_OUTPUT_TOOL, { answer: 7 }),
textResponse('continues after the blocked capture'),
])
// A PostToolUse-style hook, registered AFTER the runtime (so the runtime's
// prepend commit listener stays outermost and composes this verdict).
ctx.on('tools/post-execute', (exec, _result, next) => {
if (exec.name === STRUCTURED_OUTPUT_TOOL) {
return Promise.resolve({ kind: 'block' as const, feedback: [{ type: 'text' as const, text: 'capture rejected by hook' }] })
}
return next()
})
const run = ctx.subagents.start('spawn', structuredRequest(parent))
const result = await run.result
// No capture was committed: the run reports the schema shortfall...
expect(result.structured).toBeUndefined()
expect(result.stopReason).toBe('error')
// ...the logged tool result is the blocked isError with the feedback...
const child = ctx.agents.get(run.id)!
const results = child.session.events.filter(e => e.type === 'tool/result')
expect((results[0]!.data as { isError?: boolean }).isError).toBe(true)
expect(JSON.stringify((results[0]!.data as { content: unknown }).content)).toContain('capture rejected by hook')
// ...and the turn CONTINUED past the blocked call (no captured veto):
// the model got to react to the failure with a second step.
expect(adapter.requests.length).toBe(2)
await run.dispose()
})
it('a post-execute accept-with-replacement still commits the capture', async () => {
const { ctx, parent } = await setup([
toolCallResponse('c1', STRUCTURED_OUTPUT_TOOL, { answer: 8 }),
])
ctx.on('tools/post-execute', (exec, _result, next) => {
if (exec.name === STRUCTURED_OUTPUT_TOOL) {
return Promise.resolve({ kind: 'accept' as const, content: [{ type: 'text' as const, text: 'recorded (rewritten)' }] })
}
return next()
})
const run = ctx.subagents.start('spawn', structuredRequest(parent))
const result = await run.result
expect(result.stopReason).toBe('completed')
expect(result.structured).toEqual({ answer: 8 })
await run.dispose()
})
it('appends the structured instruction to the child REQUEST\'s system text (base prompt preserved)', async () => {
const { ctx, parent, adapter } = await setup([toolCallResponse('c1', STRUCTURED_OUTPUT_TOOL, { answer: 1 })])
// A context-wide section stands in for the deployment persona: the
@@ -298,6 +358,21 @@ describe('in-process structured output', () => {
})
describe('final-request enforcement (the prepend agent/request listener)', () => {
it('a plain agent assembling while the runtime is LIVE gets the placeholder stripped', async () => {
// Run-scoped acquisition means a plain deployment never registers the
// tool at all; the strip branch exists for the CONCURRENT case — a plain
// agent taking a turn while some structured child holds the runtime open.
const { ctx, parent, adapter } = await setup([textResponse('parent answer')])
const hold = acquireStructuredRuntime(ctx)
parent.send([{ type: 'text', text: 'hello' }])
await parent.whenIdle()
// The placeholder IS in the registry during this turn; the assembly the
// loop rendered must not carry it for an agent without a structured run.
expect(ctx.tools.get(STRUCTURED_OUTPUT_TOOL)).toBeDefined()
expect(toolNames(adapter.requests[0]!)).not.toContain(STRUCTURED_OUTPUT_TOOL)
hold.release()
})
it('a structured child sees structured_output with ITS schema; a plain agent never sees the tool', async () => {
const { ctx, parent, adapter } = await setup([
// Parent turn (a plain agent): must NOT see the tool.
@@ -396,10 +471,13 @@ describe('in-process structured output', () => {
// and shape a structured agent's assembly on the same path the loop
// renders and logs as the request header.
const { ctx, parent } = await setup([])
const acquisition = acquireStructuredRuntime(ctx)
// Bare assemble WHILE the runtime is live: the no-agent branch must
// strip the registered placeholder (before the acquisition there is
// nothing to strip — run-scoped registration).
const bare = await ctx.systemPrompt.assemble({})
expect(bare.tools.map(tool => tool.name)).not.toContain(STRUCTURED_OUTPUT_TOOL)
const acquisition = acquireStructuredRuntime(ctx)
acquisition.attach(parent, SCHEMA)
const shaped = await ctx.systemPrompt.assemble({ agent: parent })
expect(shaped.tools.map(tool => tool.name)).toContain(STRUCTURED_OUTPUT_TOOL)
@@ -412,57 +490,37 @@ describe('in-process structured output', () => {
})
})
describe('runtime lifetime (refcount: backends + live runs)', () => {
it('registers the capture tool while a backend is loaded and unregisters when the last unloads', async () => {
const { ctx, fiber, forkFiber } = await setup([], { withFork: true })
expect(ctx.tools.get(STRUCTURED_OUTPUT_TOOL)).toBeDefined()
await fiber.dispose()
// fork still holds a reference.
expect(ctx.tools.get(STRUCTURED_OUTPUT_TOOL)).toBeDefined()
await forkFiber!.dispose()
expect(ctx.tools.get(STRUCTURED_OUTPUT_TOOL)).toBeUndefined()
})
it('a live run-level acquisition keeps the runtime registered after EVERY backend unloads', async () => {
// Simulates the run-holder half of the two-level lifetime: a structured
// run acquires at start and releases at settle, so registration ordering
// is settle-then-unregister even if all backends unload first. (A real
// in-process child dies WITH its backend's fiber — the acquisition's
// observable job is this ordering, which a manual holder pins directly.)
const { ctx, fiber, forkFiber } = await setup([], { withFork: true })
const runHolder = acquireStructuredRuntime(ctx)
await fiber.dispose()
await forkFiber!.dispose()
expect(ctx.tools.get(STRUCTURED_OUTPUT_TOOL)).toBeDefined()
runHolder.release()
expect(ctx.tools.get(STRUCTURED_OUTPUT_TOOL)).toBeUndefined()
})
it('a structured run releases its acquisition when it settles (backend unload mid-run)', async () => {
const { ctx, parent, fiber } = await setup(['hang'])
const run = ctx.subagents.start('spawn', structuredRequest(parent))
// 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 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')
// Both holders (backend + run) released — nothing keeps the runtime now.
expect(ctx.tools.get(STRUCTURED_OUTPUT_TOOL)).toBeUndefined()
await run.dispose()
})
it('fork children capture structured output through the same runtime', async () => {
describe('runtime lifetime (refcount: live structured runs)', () => {
it('the runtime exists exactly while structured runs are live: nothing before, nothing after', async () => {
const { ctx, parent } = await setup([
toolCallResponse('c1', STRUCTURED_OUTPUT_TOOL, { answer: 9 }),
], { withFork: true })
const run = ctx.subagents.start('fork', structuredRequest(parent))
toolCallResponse('c1', STRUCTURED_OUTPUT_TOOL, { answer: 4 }),
])
// No always-on global state: a context that has run no structured child
// carries no capture tool.
expect(ctx.tools.get(STRUCTURED_OUTPUT_TOOL)).toBeUndefined()
const run = ctx.subagents.start('spawn', structuredRequest(parent))
const result = await run.result
expect(result.structured).toEqual({ answer: 9 })
// The capture succeeded — the registrations existed while the run lived.
expect(result.structured).toEqual({ answer: 4 })
// The run's settle released the last acquisition.
expect(ctx.tools.get(STRUCTURED_OUTPUT_TOOL)).toBeUndefined()
await run.dispose()
})
it('concurrent structured runs share one runtime; the last settle disposes it', async () => {
const { ctx, parent } = await setup([
toolCallResponse('c1', STRUCTURED_OUTPUT_TOOL, { answer: 1 }),
toolCallResponse('c2', STRUCTURED_OUTPUT_TOOL, { answer: 2 }),
])
const first = ctx.subagents.start('spawn', structuredRequest(parent))
const second = ctx.subagents.start('spawn', structuredRequest(parent))
const [a, b] = await Promise.all([first.result, second.result])
expect([a.structured, b.structured].sort()).toEqual([{ answer: 1 }, { answer: 2 }].sort())
expect(ctx.tools.get(STRUCTURED_OUTPUT_TOOL)).toBeUndefined()
await first.dispose()
await second.dispose()
})
it('acquisition release is idempotent (double release cannot underflow the refcount)', async () => {
const ctx = new Context()
await ctx.plugin(SystemPrompt)
@@ -513,13 +571,16 @@ describe('in-process structured output', () => {
acquisition.detach(parent)
acquisition.detach(parent)
acquisition.release()
// The backend still holds its own reference from setup().
expect(ctx.tools.get(STRUCTURED_OUTPUT_TOOL)).toBeDefined()
// That manual acquisition was the ONLY holder - release disposes.
expect(ctx.tools.get(STRUCTURED_OUTPUT_TOOL)).toBeUndefined()
})
})
it('a direct structured_output call from an agent WITHOUT a structured run is an isError', async () => {
const { ctx, parent } = await setup([])
// Hold the runtime open (run-scoped: nothing is registered otherwise) so
// the call reaches the capture tool's own fail-loud guard, not UNKNOWN_TOOL.
const hold = acquireStructuredRuntime(ctx)
const result = await ctx.tools.execute({
callId: 'x' as never,
name: STRUCTURED_OUTPUT_TOOL,
@@ -527,16 +588,19 @@ describe('in-process structured output', () => {
agent: parent,
})
expect(result.isError).toBe(true)
expect(result.content[0]).toMatchObject({ type: 'text' })
expect(JSON.stringify(result.content)).toContain('only available to subagents')
hold.release()
})
it('a structured_output call with NO calling agent at all is an isError', async () => {
const { ctx } = await setup([])
const hold = acquireStructuredRuntime(ctx)
const result = await ctx.tools.execute({
callId: 'x' as never,
name: STRUCTURED_OUTPUT_TOOL,
arguments: { answer: 1 },
})
expect(result.isError).toBe(true)
hold.release()
})
})

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

@@ -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

20
packages/subagent/subagent-spawn/src/index.ts
View File

@@ -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))
}

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

@@ -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<typeof MockAdapter>[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')