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

fix(subagent): identify lifecycle runs explicitly

Browse Source
This commit is contained in:
Tianyi Cui
2026-07-18 14:26:14 +08:00
parent 5af8b38b36
commit 0f85ae0a9d
26 changed files with 253 additions and 93 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -1028,7 +1028,7 @@ export const TYPE_API: readonly TypeApiEntry[] = [
},
{
name: 'SubagentRun',
declaration: 'export interface SubagentRun {\n readonly id: SessionId;\n readonly result: Promise<SubagentResult>;\n dispose(): Promise<void>;\n sendMessage?(content: ContentBlock[]): void;\n resume?(content: ContentBlock[]): Promise<SubagentRun>;\n}',
declaration: 'export interface SubagentRun {\n readonly id: SessionId;\n readonly localAgent: Agent | undefined;\n readonly result: Promise<SubagentResult>;\n dispose(): Promise<void>;\n sendMessage?(content: ContentBlock[]): void;\n resume?(content: ContentBlock[]): Promise<SubagentRun>;\n}',
},
{
name: 'SubagentStartRequest',

7
packages/hooks/hooks-claude/tests/bridge.spec.ts
View File

@@ -9,6 +9,7 @@ import { defineTool } from '@deepseek-ai/dsh-tools'
import AgentLoop, { type ReactLoopAgent } from '@deepseek-ai/dsh-agent-loop'
import { mountAgentLoopTestDependencies } from '@deepseek-ai/dsh-agent-loop-testkit'
import { LocalBashExecutor } from '@deepseek-ai/dsh-bash-local'
import { SubagentRunId } from '@deepseek-ai/dsh-subagent'
import * as HooksClaude from '@deepseek-ai/dsh-hooks-claude'
import { MockAdapter, textResponse, toolCallResponse } from '../../../core/agent-loop/tests/mock-adapter.ts'
@@ -288,8 +289,8 @@ describe('hooks-claude bridge — SubagentStart / SubagentStop (observe)', () =>
// Drive the observe-only lifecycle events directly (no real child needed — the
// bridge just listens). No child agent is registered, so SubagentStart's
// child lookup yields undefined and it simply runs the hook.
ctx.emit('subagent/start', { provider: 'inproc', id: SessionId('child-1') })
ctx.emit('subagent/end', { provider: 'inproc', id: SessionId('child-1'), stopReason: 'completed', lastAssistantMessage: [{ type: 'text', text: 'done' }] })
ctx.emit('subagent/start', { runId: SubagentRunId('run-1'), provider: 'inproc', id: SessionId('child-1'), local: false })
ctx.emit('subagent/end', { runId: SubagentRunId('run-1'), provider: 'inproc', id: SessionId('child-1'), local: false, stopReason: 'completed', lastAssistantMessage: [{ type: 'text', text: 'done' }] })
// Both hooks run async (detached .then); poll for their marker files rather
// than a fixed sleep that flakes under load.
@@ -324,7 +325,7 @@ describe('hooks-claude bridge — SubagentStart / SubagentStop (observe)', () =>
const { ctx, hooks } = await harnessWithFiber(dir, new MockAdapter([]))
const warn = vi.fn()
ctx.logger.warn = warn as never
ctx.emit('subagent/start', { provider: 'inproc', id: SessionId('child-1') })
ctx.emit('subagent/start', { runId: SubagentRunId('run-1'), provider: 'inproc', id: SessionId('child-1'), local: false })
await waitFor(() => existsSync(marker))
const pid = Number(readFileSync(pidFile, 'utf8').trim())
await hooks.dispose()

9
packages/hooks/hooks-claude/tests/coverage-cases.ts
View File

@@ -9,6 +9,7 @@ import { defineTool } from '@deepseek-ai/dsh-tools'
import AgentLoop, { type ReactLoopAgent } from '@deepseek-ai/dsh-agent-loop'
import { mountAgentLoopTestDependencies } from '@deepseek-ai/dsh-agent-loop-testkit'
import { LocalBashExecutor } from '@deepseek-ai/dsh-bash-local'
import { SubagentRunId } from '@deepseek-ai/dsh-subagent'
import * as HooksClaude from '@deepseek-ai/dsh-hooks-claude'
import { MockAdapter, textResponse, toolCallResponse } from '../../../core/agent-loop/tests/mock-adapter.ts'
@@ -231,7 +232,7 @@ export function defineCoverageCases(group: CoverageGroup): void {
const injected: string[] = []
const child = { id: SessionId('child-x'), inject: (content: { type: string; text?: string }[]) => { injected.push(content.map(b => b.text ?? '').join('')) }, session: { id: SessionId('child-x'), header: { id: 'child-x' } } } as unknown as Parameters<typeof ctx.agents.register>[0]
ctx.agents.register(child)
ctx.emit('subagent/start', { provider: 'p', id: SessionId('child-x') })
ctx.emit('subagent/start', { runId: SubagentRunId('run-x'), provider: 'p', id: SessionId('child-x'), local: true })
await waitFor(() => injected.includes('child guidance'))
expect(injected).toContain('child guidance')
})
@@ -247,7 +248,7 @@ export function defineCoverageCases(group: CoverageGroup): void {
const warn = vi.fn(); ctx.logger.warn = warn as never
const child = { id: SessionId('child-y'), inject: () => { throw new Error('inject boom') }, session: { id: SessionId('child-y'), header: { id: 'child-y' } } } as unknown as Parameters<typeof ctx.agents.register>[0]
ctx.agents.register(child)
ctx.emit('subagent/start', { provider: 'p', id: SessionId('child-y') })
ctx.emit('subagent/start', { runId: SubagentRunId('run-y'), provider: 'p', id: SessionId('child-y'), local: true })
await waitFor(() => warn.mock.calls.some(c => String(c[0]).includes('SubagentStart hook failed')))
expect(warn).toHaveBeenCalledWith(expect.stringContaining('SubagentStart hook failed'))
})
@@ -291,7 +292,7 @@ export function defineCoverageCases(group: CoverageGroup): void {
const s = sh(d, 'stop.sh', `#!/usr/bin/env bash\ntouch "${marker}"\n`)
const path = hooks(d, { SubagentStop: [{ hooks: [{ type: 'command', command: s }] }] })
const ctx = await harness(path, new MockAdapter([]))
ctx.emit('subagent/end', { provider: 'p', id: SessionId('child-z'), stopReason: 'completed' })
ctx.emit('subagent/end', { runId: SubagentRunId('run-z'), provider: 'p', id: SessionId('child-z'), local: false, stopReason: 'completed' })
await waitFor(() => existsSync(marker))
expect(existsSync(marker)).toBe(true)
})
@@ -691,7 +692,7 @@ export function defineCoverageCases(group: CoverageGroup): void {
// Register a live child on its own session cwd; emit subagent/end with its id.
const { SessionId } = await import('@deepseek-ai/dsh-session')
const childHandle = await ctx.agents.create({ sessionId: SessionId('child-stop-session'), meta: { cwd: childDir }, agentOptions: { provider: 'mock', model: 'mock' } })
ctx.emit('subagent/end', { provider: 'inproc', id: childHandle.agent.id, stopReason: 'completed' })
ctx.emit('subagent/end', { runId: SubagentRunId('run-stop'), provider: 'inproc', id: childHandle.agent.id, local: true, stopReason: 'completed' })
await waitFor(() => existsSync(marker))
expect(existsSync(marker)).toBe(true) // the marker landed in the CHILD dir

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

@@ -294,6 +294,7 @@ export async function startAcpRun(request: SubagentStartRequest, spec: AcpRunSpe
let disposal: Promise<void> | undefined
return {
id,
localAgent: undefined,
result,
dispose(): Promise<void> {
if (disposal !== undefined) return disposal

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

@@ -175,6 +175,7 @@ export async function startInProcessRun(
return {
id: childId,
localAgent: child,
result,
dispose(): Promise<void> {
request.signal.removeEventListener('abort', onAbort)

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

@@ -50,9 +50,9 @@ Runtime features are optional methods on `SubagentRun`: `sendMessage?` steers a
`SubagentRun.result` resolves to `{ output, structured?, stopReason }`. Child-level failures resolve with a non-`completed` reason; only an infrastructure fault that the seam cannot represent may reject. `dispose()` is idempotent, cancels remaining work, and waits for the child resources to quiesce.
A local run publishes an ordinary child agent/session before `start()` fulfills, returns that shared session id as `SubagentRun.id`, and records `request.parent.session.id` in the child's `parentSession` header. The child may be owned by the parent scope or by a provider/root scope; durable lineage is the transport-neutral local-child relation. Remote providers instead mint a parent-scoped lifecycle id without publishing a local child.
A local run publishes an ordinary child agent/session before `start()` fulfills, returns that shared session id as `SubagentRun.id`, exposes the exact child as `SubagentRun.localAgent`, and records `request.parent.session.id` in the child's `parentSession` header. Remote providers instead mint a parent-scoped lifecycle id and return `localAgent: undefined`.
The service emits `subagent/start` only after `start()` has fulfilled. It attaches the result observer before that synchronous notification, so even an already-settled child still produces `subagent/start` before `subagent/end`. In-process start observers can resolve the published child through `ctx.agents.get(info.id)`; remote providers need not publish a local agent.
The service emits `subagent/start` only after `start()` has fulfilled. It attaches the result observer before that synchronous notification, so even an already-settled child still produces `subagent/start` before `subagent/end`. The pair shares a service-minted `runId`; its `local` flag is snapshotted from the provider's exact `localAgent`, so observers never infer run identity or locality from reusable provider/session names.
Run events are scoped to the delegating parent. Every listener is independently contained: a synchronous throw or rejected returned promise is logged without starving peer listeners or changing the run.

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

@@ -23,6 +23,7 @@
"license": "BSD-3-Clause",
"peerDependencies": {
"@deepseek-ai/dsh-agent": "^0.0.1",
"@deepseek-ai/dsh-brand": "^0.0.1",
"@deepseek-ai/dsh-llm": "^0.0.1",
"@deepseek-ai/dsh-scope": "^0.0.1",
"@deepseek-ai/dsh-session": "^0.0.1",
@@ -31,6 +32,7 @@
},
"devDependencies": {
"@deepseek-ai/dsh-agent": "workspace:^",
"@deepseek-ai/dsh-brand": "workspace:^",
"@deepseek-ai/dsh-llm": "workspace:^",
"@deepseek-ai/dsh-scope": "workspace:^",
"@deepseek-ai/dsh-session": "workspace:^",

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

@@ -28,6 +28,7 @@
* @module @deepseek-ai/dsh-subagent
*/
import { randomUUID } from 'node:crypto'
import { Context, Service } from 'cordis'
import { scopeTarget } from '@deepseek-ai/dsh-scope'
import type { Scoped } from '@deepseek-ai/dsh-scope'
@@ -43,7 +44,9 @@ import type {
SubagentRun,
SubagentStartRequest,
} from './types.ts'
import { SubagentRunId } from './types.ts'
export { SubagentRunId } from './types.ts'
export type {
SubagentCapabilities,
SubagentProvider,
@@ -112,18 +115,26 @@ declare module 'cordis' {
/** Observe-only identifying detail for a ready subagent run. */
export interface SubagentRunInfo {
/** Unique identity shared with the paired terminal event. */
readonly runId: SubagentRunId
/** The provider that established the run. */
readonly provider: string
/** The child agent's id. */
readonly id: SessionId
/** Whether the provider exposed an exact published in-process child. */
readonly local: boolean
}
/** Observe-only outcome detail for a settled subagent run. */
export interface SubagentRunEndInfo {
/** Unique identity shared with the paired start event. */
readonly runId: SubagentRunId
/** The provider that ran it. */
readonly provider: string
/** The child agent's id. */
readonly id: SessionId
/** Whether the provider exposed an exact published in-process child. */
readonly local: boolean
/** The terminal stop reason. */
readonly stopReason: SubagentResult['stopReason']
/** The child's final assistant output, absent on infrastructure rejection. */
@@ -208,22 +219,28 @@ export class SubagentService extends Service {
const parent = request.parent
const run = await provider.start(request)
const runId = SubagentRunId(randomUUID())
const lifecycleIdentity = {
runId,
provider: name,
id: run.id,
local: run.localAgent !== undefined,
}
// Attach the terminal observer before dispatching start. Promise reactions
// still run after this synchronous start emission, preserving start → end.
void run.result.then(
(result) => {
this.emitLifecycle('subagent/end', {
provider: name,
id: run.id,
...lifecycleIdentity,
stopReason: result.stopReason,
lastAssistantMessage: result.output,
}, parent)
},
() => {
this.emitLifecycle('subagent/end', { provider: name, id: run.id, stopReason: 'error' }, parent)
this.emitLifecycle('subagent/end', { ...lifecycleIdentity, stopReason: 'error' }, parent)
},
)
this.emitLifecycle('subagent/start', { provider: name, id: run.id }, parent)
this.emitLifecycle('subagent/start', lifecycleIdentity, parent)
return run
}

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

@@ -7,10 +7,23 @@
*/
import type { Agent, AgentOptions } from '@deepseek-ai/dsh-agent'
import type { Branded } from '@deepseek-ai/dsh-brand'
import type { ContentBlock } from '@deepseek-ai/dsh-llm'
import type { SessionId } from '@deepseek-ai/dsh-session'
import type { StructuredOutputSchema, ToolRestriction } from '@deepseek-ai/dsh-tools'
/** Identifies one accepted subagent run across its lifecycle event pair. */
export type SubagentRunId = Branded<'SubagentRunId'>
/**
* Brand a string as a {@link SubagentRunId}.
* @param id - the raw id string (the service mints UUIDs; tests may pass fixtures).
* @returns the same string, branded.
*/
export function SubagentRunId(id: string): SubagentRunId {
return id as SubagentRunId
}
/**
* Which START-TIME features a provider supports. Checked by the service before delegating to
* {@link SubagentProvider.start}: a request that needs a capability the chosen provider lacks
@@ -139,6 +152,12 @@ export interface SubagentRun {
* remote provider mints an id unique in the parent namespace.
*/
readonly id: SessionId
/**
* The exact published in-process child, or `undefined` for a remote run.
* When present, its id is {@link id}; the provider retains no ownership
* implication beyond the run's ordinary {@link dispose} contract.
*/
readonly localAgent: Agent | undefined
/**
* Resolves with the child's terminal {@link SubagentResult} when the run
* settles. Does NOT reject on a child-level failure — a model/transport

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

@@ -48,6 +48,7 @@ class StubProvider implements SubagentProvider {
this.startCount += 1
return {
id: SessionId(`child:${this.name}:${request.parent.id}`),
localAgent: undefined,
result: Promise.resolve(this.outcome),
async dispose() {},
}
@@ -137,13 +138,14 @@ describe('SubagentService', () => {
const parent = fakeParent('delegator')
const events: string[] = []
const keys: unknown[] = []
ctx.on('subagent/start', function () { events.push('start'); keys.push(carrierKeyOf(this)) })
ctx.on('subagent/end', function () { events.push('end'); keys.push(carrierKeyOf(this)) })
const runIds: string[] = []
ctx.on('subagent/start', function (info) { events.push('start'); keys.push(carrierKeyOf(this)); runIds.push(info.runId) })
ctx.on('subagent/end', function (info) { events.push('end'); keys.push(carrierKeyOf(this)); runIds.push(info.runId) })
const starting = subagents.start('deferred', baseRequest({ parent }))
await Promise.resolve()
expect(events).toEqual([])
ready.resolve({ id: SessionId('child'), result: result.promise, async dispose() {} })
ready.resolve({ id: SessionId('child'), localAgent: undefined, result: result.promise, async dispose() {} })
const run = await starting
expect(events).toEqual(['start'])
result.resolve({ output: [{ type: 'text', text: 'answer' }], stopReason: 'completed' })
@@ -151,6 +153,21 @@ describe('SubagentService', () => {
await Promise.resolve()
expect(events).toEqual(['start', 'end'])
expect(keys).toEqual([parent, parent])
expect(runIds[0]).toBe(runIds[1])
})
it('mints distinct lifecycle identities when provider and child ids repeat', async () => {
const { ctx, subagents } = await service()
subagents.registerProvider(new StubProvider('reused'))
const runIds: string[] = []
ctx.on('subagent/start', info => void runIds.push(info.runId))
const first = await subagents.start('reused', baseRequest())
const second = await subagents.start('reused', baseRequest())
await Promise.all([first.result, second.result])
expect(runIds).toHaveLength(2)
expect(new Set(runIds).size).toBe(2)
})
it('emits no run lifecycle when provider startup rejects', async () => {
@@ -192,7 +209,7 @@ describe('SubagentService', () => {
capabilities: NO_CAPS,
inheritsParentContext: false,
async start() {
return { id: SessionId('infra-child'), result: failure.promise, async dispose() {} }
return { id: SessionId('infra-child'), localAgent: undefined, result: failure.promise, async dispose() {} }
},
})
const failedRun = await subagents.start('infra', baseRequest())

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

@@ -144,6 +144,7 @@ describe('dsh-tool-subagent', () => {
inheritsParentContext: false,
start: async () => ({
id: SessionId('weird-child'),
localAgent: undefined,
result: Promise.resolve({ output: [{ type: 'text', text: 'partial' }], stopReason: 'frobnicated' as never }),
dispose: async () => {},
}),
@@ -171,6 +172,7 @@ describe('dsh-tool-subagent', () => {
seen = request
return {
id: SessionId('capture-child'),
localAgent: undefined,
result: Promise.resolve({ output: [{ type: 'text', text: 'ok' }], stopReason: 'completed' as const }),
dispose: async () => {},
}
@@ -200,6 +202,7 @@ describe('dsh-tool-subagent', () => {
seen = request
return {
id: SessionId('bare-child'),
localAgent: undefined,
result: Promise.resolve({ output: [{ type: 'text', text: 'ok' }], stopReason: 'completed' as const }),
dispose: async () => {},
}
@@ -327,6 +330,7 @@ describe('dsh-tool-subagent', () => {
inheritsParentContext: false,
start: async () => ({
id: SessionId('spy-child'),
localAgent: undefined,
result: Promise.resolve({ output: [{ type: 'text', text: 'ok' }], stopReason: 'completed' as const }),
dispose: async () => void disposed(),
}),
@@ -349,6 +353,7 @@ describe('dsh-tool-subagent', () => {
inheritsParentContext: false,
start: async () => ({
id: SessionId('spy-child'),
localAgent: undefined,
result: Promise.resolve({ output: [], stopReason: 'error' as const }),
dispose: async () => void disposed(),
}),
@@ -380,6 +385,7 @@ describe('dsh-tool-subagent', () => {
}, { once: true })
return {
id: SessionId('spy-child'),
localAgent: undefined,
result,
dispose: async () => {},
}
@@ -472,6 +478,7 @@ describe('dsh-tool-subagent', () => {
seen = request
return {
id: SessionId('capture2-child'),
localAgent: undefined,
result: Promise.resolve({ output: [{ type: 'text', text: 'ok' }], stopReason: 'completed' as const }),
dispose: async () => {},
}
@@ -529,6 +536,7 @@ describe('dsh-tool-subagent', () => {
seen = request
return {
id: SessionId('capture3-child'),
localAgent: undefined,
result: Promise.resolve({ output: [{ type: 'text', text: 'ok' }], stopReason: 'completed' as const }),
dispose: async () => {},
}
@@ -558,6 +566,7 @@ describe('dsh-tool-subagent', () => {
seen = request
return {
id: SessionId('capture4-child'),
localAgent: undefined,
result: Promise.resolve({ output: [{ type: 'text', text: 'ok' }], stopReason: 'completed' as const }),
dispose: async () => {},
}
@@ -732,6 +741,7 @@ describe('dsh-tool-subagent background mode', () => {
}, { once: true })
return {
id,
localAgent: undefined,
result,
dispose: () => Promise.resolve(),
}
@@ -771,6 +781,7 @@ describe('dsh-tool-subagent background mode', () => {
const order: string[] = []
const completed = await settleRun({
id: SessionId('child-1'),
localAgent: undefined,
result: Promise.resolve({ output: [{ type: 'text' as const, text: 'ok' }], stopReason: 'completed' as const }),
dispose() { order.push('dispose'); return Promise.resolve() },
})
@@ -782,6 +793,7 @@ describe('dsh-tool-subagent background mode', () => {
let disposed = false
const failed = await settleRun({
id: SessionId('child-2'),
localAgent: undefined,
result: Promise.reject(new Error('transport gone')),
dispose() { disposed = true; return Promise.resolve() },
})
@@ -790,6 +802,7 @@ describe('dsh-tool-subagent background mode', () => {
const disposeFailed = await settleRun({
id: SessionId('child-3'),
localAgent: undefined,
result: Promise.resolve({ output: [], stopReason: 'completed' }),
dispose: () => Promise.reject(new Error('reap failed')),
})
@@ -797,6 +810,7 @@ describe('dsh-tool-subagent background mode', () => {
const bothFailed = await settleRun({
id: SessionId('child-4'),
localAgent: undefined,
result: Promise.reject(new Error('result failed')),
dispose: () => Promise.reject(new Error('reap failed')),
})
@@ -832,6 +846,7 @@ describe('background preflight failure (no orphaned child, by construction)', ()
starts += 1
return {
id: SessionId('probe-child'),
localAgent: undefined,
result: Promise.resolve({ output: [], stopReason: 'completed' as const }),
dispose: () => Promise.resolve(),
}

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

@@ -69,6 +69,7 @@ class MockSubagentProvider implements SubagentProvider {
})
return {
id,
localAgent: undefined,
result,
dispose(): Promise<void> {
flags.cancelled = true

2
packages/ui/jsonrpc/README.md
View File

@@ -4,7 +4,7 @@ The **SDK server plugin** (`jsonrpc`): mounting it serves a stdio JSON-RPC serve
## Wiring
`inject: ['agents']`. The server gets or creates one agent per `sessionId` from the `initialize.provider`/`initialize.model` pair. It classifies subagent completions through live parent ownership or durable parent lineage and retains parent-scoped provider/id counts after child disposal. A registered owner for the provider route wins; an unowned `deepseek` route mounts `dsh-llm-deepseek` using `$DEEPSEEK_API_KEY` and `$DEEPSEEK_BASE_URL`, while any other unowned provider fails initialization. Persistence, tools, and other adapters come from the surrounding `cordis.yml`.
`inject: ['agents']`. The server gets or creates one agent per `sessionId` from the `initialize.provider`/`initialize.model` pair. It forwards subagent completions only when the lifecycle payload's `local` flag was snapshotted from the provider's exact in-process child; reusable provider names, child ids, and durable lineage never establish locality. A registered owner for the provider route wins; an unowned `deepseek` route mounts `dsh-llm-deepseek` using `$DEEPSEEK_API_KEY` and `$DEEPSEEK_BASE_URL`, while any other unowned provider fails initialization. Persistence, tools, and other adapters come from the surrounding `cordis.yml`.
## Config

45
packages/ui/jsonrpc/src/server.ts
View File

@@ -19,7 +19,7 @@ import type { Agent, AgentHandle } from '@deepseek-ai/dsh-agent'
import { carrierKeyOf, type Scoped } from '@deepseek-ai/dsh-scope'
import { SessionId, type TurnEndReason } from '@deepseek-ai/dsh-session'
import type SubagentService from '@deepseek-ai/dsh-subagent'
import type { SubagentRunEndInfo, SubagentRunInfo } from '@deepseek-ai/dsh-subagent'
import type { SubagentRunEndInfo } from '@deepseek-ai/dsh-subagent'
import * as LlmDeepSeek from '@deepseek-ai/dsh-llm-deepseek'
import type { JsonRpcTransportPeer } from './transport.ts'
@@ -68,15 +68,6 @@ function subagentParentOf(carrier: Scoped<SubagentService>): Agent {
return carrierKeyOf(carrier) as Agent
}
/** Whether the live id names a local child related to this exact delegating parent. */
function isLocalChild(ctx: Context, id: SessionId, parent: Agent): boolean {
const child = ctx.agents.get(id)
return child !== undefined && (
ctx.agents.isOwnedBy(id, parent)
|| child.session.header.parentSession === parent.session.id
)
}
/**
* The SDK server over a booted harness context. Constructing it subscribes to
* session and subagent lifecycle events, forwarding durable session
@@ -92,7 +83,6 @@ export class HarnessSdkServer {
private llmFiber: { dispose(): Promise<void> } | undefined
private readonly sessions = new Map<string, SessionRecord>()
private readonly sessionCreations = new Map<string, Promise<SessionRecord>>()
private readonly localRuns = new Map<string, Map<SessionId, Map<Agent, number>>>()
private readonly disposers: (() => void)[] = []
private shutdownTask: Promise<Record<string, never>> | undefined
private shuttingDown = false
@@ -116,36 +106,12 @@ export class HarnessSdkServer {
childSessionId: String(session.id),
})
}))
// In-process providers publish the child before start. Count starts related
// by exact runtime ownership or durable parent lineage so provider-owned
// roots remain local, completions survive child disposal, and reused ids
// need no settlement-order assumption.
const localRuns = this.localRuns
this.disposers.push(ctx.on('subagent/start', function (this: Scoped<SubagentService>, info: SubagentRunInfo) {
const parent = subagentParentOf(this)
if (!isLocalChild(ctx, info.id, parent)) return
const providerRuns = localRuns.get(info.provider) ?? new Map<SessionId, Map<Agent, number>>()
const parentRuns = providerRuns.get(info.id) ?? new Map<Agent, number>()
parentRuns.set(parent, (parentRuns.get(parent) ?? 0) + 1)
providerRuns.set(info.id, parentRuns)
localRuns.set(info.provider, providerRuns)
}))
this.disposers.push(ctx.on('subagent/end', function (this: Scoped<SubagentService>, info: SubagentRunEndInfo) {
const parent = subagentParentOf(this)
const providerRuns = localRuns.get(info.provider)
const parentRuns = providerRuns?.get(info.id)
const pendingCount = parentRuns?.get(parent)
if (pendingCount !== undefined) {
if (pendingCount === 1) parentRuns?.delete(parent)
else parentRuns?.set(parent, pendingCount - 1)
if (parentRuns?.size === 0) providerRuns?.delete(info.id)
if (providerRuns?.size === 0) localRuns.delete(info.provider)
}
// This protocol reports LOCAL child sessions. A lineage-bearing child
// has the session/created-driven start notification above. A remote run
// has neither a cached local start nor a live child related to this
// parent; an unrelated local agent with the same id never makes it local.
if (pendingCount === undefined && !isLocalChild(ctx, info.id, parent)) return
// This protocol reports only in-process child sessions. The service
// snapshots the provider's exact run provenance through child disposal;
// matching ids or parent lineage alone never establishes locality.
if (!info.local) return
transport.notify('subagent.finished', {
provider: info.provider,
agentId: String(info.id),
@@ -223,7 +189,6 @@ export class HarnessSdkServer {
this.sessionCreations.clear()
const records = [...this.sessions.values()]
this.sessions.clear()
this.localRuns.clear()
const failures: unknown[] = []
while (this.disposers.length > 0) {
try {

1
packages/ui/jsonrpc/tests/built-scope-carrier.e2e.ts
View File

@@ -71,6 +71,7 @@ try {
start() {
return Promise.resolve({
id: child.agent.id,
localAgent: child.agent,
result: result.promise,
dispose() { return Promise.resolve(); },
});

130
packages/ui/jsonrpc/tests/server.spec.ts
View File

@@ -70,7 +70,7 @@ async function makeHarness(storageDir: string) {
async function settleSubagent(
ctx: Context,
parent: Agent,
info: SubagentRunEndInfo,
info: Omit<SubagentRunEndInfo, 'runId' | 'local'> & { localAgent: Agent | undefined },
beforeSettle?: () => Promise<void>,
): Promise<void> {
const result = Promise.withResolvers<SubagentResult>()
@@ -81,6 +81,7 @@ async function settleSubagent(
async start() {
return {
id: info.id,
localAgent: info.localAgent,
result: result.promise,
dispose: () => Promise.resolve(),
}
@@ -294,12 +295,14 @@ describe('HarnessSdkServer', () => {
await settleSubagent(ctx, parentHandle.agent, {
provider: 'spawn',
id: SessionId('child-session'),
localAgent: handle.agent,
stopReason: 'completed',
lastAssistantMessage: [{ type: 'text', text: 'child done' }],
}, () => handle.dispose())
await settleSubagent(ctx, parentHandle.agent, {
provider: 'spawn',
id: SessionId('parentless-child-session'),
localAgent: parentlessHandle.agent,
stopReason: 'error',
}, () => parentlessHandle.dispose())
@@ -335,7 +338,7 @@ describe('HarnessSdkServer', () => {
}
})
it('ignores a remote run id that collides with an unrelated local agent', async () => {
it('ignores a remote run id that collides with a local child of the same parent', async () => {
const storageDir = await mkdtemp(join(tmpdir(), 'dsh-jsonrpc-subagent-remote-collision-'))
const ctx = await makeHarness(storageDir)
try {
@@ -346,24 +349,26 @@ describe('HarnessSdkServer', () => {
meta: { cwd: storageDir },
agentOptions: { model: 'deepseek' },
})
const unrelatedHandle = await ctx.agents.create({
const collidingChild = await parentHandle.agent.ctx.agents.create({
sessionId: SessionId('remote-run-id'),
meta: { cwd: storageDir },
meta: { cwd: storageDir, parentSession: SessionId('collision-parent') },
agentOptions: { model: 'deepseek' },
})
await settleSubagent(ctx, parentHandle.agent, {
provider: 'remote',
id: SessionId('remote-run-id'),
localAgent: undefined,
stopReason: 'completed',
lastAssistantMessage: [],
}, () => unrelatedHandle.dispose())
})
expect(transport.notifications.some(notification =>
notification.method === 'subagent.finished'
&& notification.params?.agentId === 'remote-run-id',
)).toBe(false)
await collidingChild.dispose()
await parentHandle.dispose()
await server.shutdown()
} finally {
@@ -392,12 +397,14 @@ describe('HarnessSdkServer', () => {
await settleSubagent(ctx, parentHandle.agent, {
provider: 'continuation',
id: SessionId('continuation-child'),
localAgent: childHandle.agent,
stopReason: 'completed',
lastAssistantMessage: [{ type: 'text', text: 'first' }],
})
await settleSubagent(ctx, parentHandle.agent, {
provider: 'continuation',
id: SessionId('continuation-child'),
localAgent: childHandle.agent,
stopReason: 'completed',
lastAssistantMessage: [{ type: 'text', text: 'second' }],
}, () => childHandle.dispose())
@@ -436,6 +443,7 @@ describe('HarnessSdkServer', () => {
const replacement = Promise.withResolvers<SubagentResult>()
const results = [first.promise, sameLifetime.promise, replacement.promise]
let starts = 0
let currentLocalAgent = oldChild.agent
const disposeProvider = ctx.subagents.registerProvider({
name: 'reused',
capabilities: { outputSchema: false, depthLimit: false, toolFilter: false, persona: false },
@@ -444,7 +452,7 @@ describe('HarnessSdkServer', () => {
const result = results[starts]
starts += 1
if (result === undefined) throw new Error('unexpected fourth reused-id run')
return Promise.resolve({ id: SessionId('reused-child'), result, dispose: () => Promise.resolve() })
return Promise.resolve({ id: SessionId('reused-child'), localAgent: currentLocalAgent, result, dispose: () => Promise.resolve() })
},
})
@@ -471,6 +479,7 @@ describe('HarnessSdkServer', () => {
meta: { cwd: storageDir, parentSession: SessionId('new-parent') },
agentOptions: { model: 'deepseek' },
})
currentLocalAgent = newChild.agent
const secondRun = await ctx.subagents.start('reused', {
parent: newParent.agent,
prompt: [],
@@ -512,7 +521,99 @@ describe('HarnessSdkServer', () => {
}
})
it('falls back to live lineage and ignores runs without a local child session', async () => {
it('keeps locality bound to the accepted run across provider re-registration', async () => {
const storageDir = await mkdtemp(join(tmpdir(), 'dsh-jsonrpc-subagent-provider-reuse-'))
const ctx = await makeHarness(storageDir)
try {
const transport = new FakeTransport()
const server = new HarnessSdkServer(ctx, transport)
const parent = await ctx.agents.create({
sessionId: SessionId('provider-reuse-parent'),
meta: { cwd: storageDir },
agentOptions: { model: 'deepseek' },
})
const child = await parent.agent.ctx.agents.create({
sessionId: SessionId('provider-reuse-child'),
meta: { cwd: storageDir, parentSession: SessionId('provider-reuse-parent') },
agentOptions: { model: 'deepseek' },
})
const localResult = Promise.withResolvers<SubagentResult>()
const remoteResult = Promise.withResolvers<SubagentResult>()
const unregisterLocal = ctx.subagents.registerProvider({
name: 'reused-provider',
capabilities: { outputSchema: false, depthLimit: false, toolFilter: false, persona: false },
inheritsParentContext: false,
start: () => Promise.resolve({
id: SessionId('provider-reuse-child'),
localAgent: child.agent,
result: localResult.promise,
dispose: () => Promise.resolve(),
}),
})
const localRun = await ctx.subagents.start('reused-provider', {
parent: parent.agent,
prompt: [],
signal: new AbortController().signal,
})
unregisterLocal()
const unregisterRemote = ctx.subagents.registerProvider({
name: 'reused-provider',
capabilities: { outputSchema: false, depthLimit: false, toolFilter: false, persona: false },
inheritsParentContext: false,
start: () => Promise.resolve({
id: SessionId('provider-reuse-child'),
localAgent: undefined,
result: remoteResult.promise,
dispose: () => Promise.resolve(),
}),
})
const remoteRun = await ctx.subagents.start('reused-provider', {
parent: parent.agent,
prompt: [],
signal: new AbortController().signal,
})
remoteResult.resolve({ output: [{ type: 'text', text: 'remote' }], stopReason: 'completed' })
await remoteRun.result
await Promise.resolve()
expect(transport.notifications.some(notification =>
notification.method === 'subagent.finished'
&& notification.params?.lastAssistantMessage !== undefined,
)).toBe(false)
await child.dispose()
localResult.resolve({ output: [{ type: 'text', text: 'local' }], stopReason: 'completed' })
await localRun.result
await Promise.resolve()
expect(transport.notifications.filter(notification =>
notification.method === 'subagent.finished'
&& notification.params?.childSessionId === 'provider-reuse-child',
)).toEqual([{
method: 'subagent.finished',
params: {
provider: 'reused-provider',
agentId: 'provider-reuse-child',
parentSessionId: 'provider-reuse-parent',
childSessionId: 'provider-reuse-child',
status: 'ok',
stopReason: 'completed',
lastAssistantMessage: [{ type: 'text', text: 'local' }],
},
}])
await localRun.dispose()
await remoteRun.dispose()
unregisterRemote()
await parent.dispose()
await server.shutdown()
} finally {
await ctx.fiber.dispose()
await rm(storageDir, { recursive: true, force: true })
}
})
it('uses explicit local provenance when start was missed and ignores remote runs', async () => {
const storageDir = await mkdtemp(join(tmpdir(), 'dsh-jsonrpc-subagent-fallback-'))
const ctx = await makeHarness(storageDir)
let parentHandle: AgentHandle | undefined
@@ -529,6 +630,7 @@ describe('HarnessSdkServer', () => {
meta: { cwd: storageDir, parentSession: SessionId('fallback-parent') },
agentOptions: { provider: 'deepseek', model: 'deepseek' },
})
const fallbackChild = handle.agent
failedHandle = await parentHandle.agent.ctx.agents.create({
sessionId: SessionId('failed-child-session'),
meta: { cwd: storageDir },
@@ -541,12 +643,13 @@ describe('HarnessSdkServer', () => {
inheritsParentContext: true,
start: () => Promise.resolve({
id: SessionId('fallback-child-session'),
localAgent: fallbackChild,
result: missedStartResult.promise,
dispose: () => Promise.resolve(),
}),
})
// Start before the server subscribes, so the terminal fallback must use
// the still-live registry entry rather than a cached start record.
// Start before the server subscribes. The terminal payload still carries
// this run's exact local child without reconstructing it from ids.
const missedStartRun = await ctx.subagents.start('fork', {
parent: parentHandle.agent,
prompt: [],
@@ -560,22 +663,25 @@ describe('HarnessSdkServer', () => {
await Promise.resolve()
await missedStartRun.dispose()
disposeMissedStartProvider()
// The server also missed this agent's creation, but observes the start;
// recover its lineage from the still-live registry entry.
// The server also missed this agent's creation but sees the exact child
// on the run lifecycle payload.
await settleSubagent(ctx, parentHandle.agent, {
provider: 'fork-live-fallback',
id: SessionId('fallback-child-session'),
localAgent: fallbackChild,
stopReason: 'completed',
lastAssistantMessage: [],
})
await settleSubagent(ctx, parentHandle.agent, {
provider: 'fork',
id: SessionId('failed-child-session'),
localAgent: failedHandle.agent,
stopReason: 'error',
})
await settleSubagent(ctx, parentHandle.agent, {
provider: 'fork',
id: SessionId('missing-child-agent'),
localAgent: undefined,
stopReason: 'error',
})
@@ -800,6 +906,6 @@ describe('HarnessSdkServer', () => {
const server = new HarnessSdkServer(ctx, new FakeTransport())
await expect(server.shutdown()).rejects.toBe(listenerFailure)
expect(on).toHaveBeenCalledTimes(4)
expect(on).toHaveBeenCalledTimes(3)
})
})

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

@@ -105,6 +105,7 @@ class StubProvider implements SubagentProvider {
if (request.signal.aborted) throw new Error('child start aborted before publication')
return {
id: SessionId(`stub-child-${index}`),
localAgent: undefined,
result: terminal.promise,
dispose: () => {
controlled.disposeCalls += 1
@@ -362,6 +363,7 @@ describe('dsh-workflow-workerthread', () => {
inheritsParentContext: false,
start: async () => ({
id: SessionId('reject-child'),
localAgent: undefined,
result: Promise.reject(new Error('backend exploded')),
dispose: () => Promise.resolve(),
}),
@@ -396,6 +398,7 @@ describe('dsh-workflow-workerthread', () => {
} as unknown as SubagentResult
const start = vi.spyOn(ctx.subagents, 'start').mockResolvedValue({
id: SessionId('raw-invalid-child'),
localAgent: undefined,
result: Promise.resolve(invalid),
dispose: () => Promise.resolve(),
})
@@ -419,6 +422,7 @@ describe('dsh-workflow-workerthread', () => {
inheritsParentContext: false,
start: async () => ({
id: SessionId('bad-dispose-child'),
localAgent: undefined,
result: Promise.resolve({ output: [{ type: 'text', text: 'fine' }], stopReason: 'completed' }),
cancel: () => { /* settled already */ },
dispose: () => { throw new Error('dispose exploded') },
@@ -440,6 +444,7 @@ describe('dsh-workflow-workerthread', () => {
inheritsParentContext: false,
start: async () => ({
id: SessionId('trap-child'),
localAgent: undefined,
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
@@ -767,6 +772,7 @@ describe('dsh-workflow-workerthread', () => {
}, { once: true })
return {
id: SessionId('signal-only-child'),
localAgent: undefined,
result,
dispose: () => Promise.resolve(),
}
@@ -1088,6 +1094,7 @@ describe('dsh-workflow-workerthread', () => {
ready.resolve({
id: SessionId('late-ready-child'),
localAgent: undefined,
result: Promise.resolve({ output: [], stopReason: 'aborted' }),
dispose: () => {
disposeCalls += 1
@@ -1124,6 +1131,7 @@ describe('dsh-workflow-workerthread', () => {
}, { once: true })
return {
id: SessionId('doomed-child'),
localAgent: undefined,
result: new Promise(() => { /* never settles; the reap is the teardown */ }),
dispose: () => Promise.reject(new Error('dispose exploded during reap')),
}