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

fix(subagent): make strict steering atomic

Browse Source
This commit is contained in:
Dudu-0223
2026-07-24 11:27:57 +08:00
committed by imccyu
parent 9e5ae0d12e
commit bb8ea2be51
24 changed files with 436 additions and 92 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
packages/subagent/subagent-control/src/index.ts
View File

@@ -181,7 +181,12 @@ export class SubagentControlService extends Service {
activation.controller.abort('subagent control service disposed')
activation.terminal.resolve()
}
await Promise.allSettled(active.map(activation => activation.done ?? Promise.resolve()))
await Promise.allSettled(active.map((activation) => {
/* v8 ignore next 2 -- TaskService invokes `run` synchronously before `start` returns;
* every retained activation has `done`, while registration failure removes it. */
if (activation.done === undefined) return Promise.resolve()
return activation.done
}))
}, 'subagentControl.activations()')
}
@@ -354,7 +359,8 @@ export class SubagentControlService extends Service {
const descriptor = foldSubagentDescriptor(loaded.events.slice(loaded.meta.seedLength ?? 0))
if (descriptor === undefined) {
throw new SubagentControlError(
`subagent "${childId}" has no supported continuation descriptor`,
`subagent "${childId}" has no supported continuation state and cannot be resumed; `
+ 'do not retry send_message with this id',
'NOT_RESUMABLE',
)
}

129
packages/subagent/subagent-control/tests/subagent-control.spec.ts
View File

@@ -89,6 +89,20 @@ async function waitTerminal(ctx: Context, taskId: TaskId, parent: Agent) {
return ctx.tasks.wait(taskId, 5_000, parent)
}
async function waitPublishedRun(ctx: Context, childId: SessionId): Promise<void> {
const control = ctx.subagentControl as unknown as {
activations: Map<SessionId, { run: unknown }>
}
await new Promise<void>((resolve) => {
const timer = setInterval(() => {
if (control.activations.get(childId)?.run !== undefined) {
clearInterval(timer)
resolve()
}
}, 5)
})
}
function message(text: string) {
return [{ type: 'text' as const, text }]
}
@@ -145,6 +159,20 @@ describe('SubagentControlService.startContinuable', () => {
expect(ctx.tasks.list(parent)).toEqual([])
})
it('rolls back the activation when Task preflight throws', async () => {
const { ctx, parent } = await setup([textResponse('unused')])
const realStart = ctx.tasks.start.bind(ctx.tasks)
ctx.tasks.start = () => { throw new Error('task preflight failed') }
try {
expect(() => ctx.subagentControl.startContinuable(startSpec(parent)))
.toThrow('task preflight failed')
} finally {
ctx.tasks.start = realStart
}
const control = ctx.subagentControl as unknown as { activations: Map<SessionId, unknown> }
expect(control.activations.size).toBe(0)
})
it('rejects a non-JSON descriptor input synchronously with no Task', async () => {
const { ctx, parent } = await setup([textResponse('unused')])
const spec = startSpec(parent)
@@ -195,6 +223,85 @@ describe('SubagentControlService.startContinuable', () => {
})
describe('SubagentControlService.sendMessage', () => {
it('omits undeclared model selectors and rejects a provider without live delivery', async () => {
const { ctx } = await setup([])
const result = Promise.withResolvers<{
output: { type: 'text'; text: string }[]
stopReason: 'completed'
}>()
let descriptor: SessionEvent<'subagent/descriptor'>['data'] | undefined
ctx.subagents.registerProvider({
name: 'no-steer',
capabilities: { outputSchema: false, depthLimit: false, toolFilter: false, persona: false },
inheritsParentContext: false,
start: async (request) => {
descriptor = request.continuation?.descriptor
return {
id: request.continuation!.sessionId,
localAgent: undefined,
result: result.promise,
async dispose() {},
}
},
resume: async () => { throw new Error('not used') },
})
const parent = ctx.agentLoop.create(SessionId('bare-parent'), {})
const started = ctx.subagentControl.startContinuable(startSpec(parent, 'no-steer'))
await waitPublishedRun(ctx, started.childId)
expect(descriptor).toEqual({ version: SUBAGENT_DESCRIPTOR_VERSION, provider: 'no-steer' })
expect(() => ctx.subagentControl.sendMessage(parent, started.childId, message('join')))
.toThrow(/provider does not accept live delivery/)
let terminalDeliveryError: unknown
ctx.tasks.onTaskDone((snapshot) => {
if (snapshot.id !== started.taskId) return
try {
ctx.subagentControl.sendMessage(parent, started.childId, message('after terminal'))
} catch (error: unknown) {
terminalDeliveryError = error
}
})
result.resolve({ output: [{ type: 'text', text: 'done' }], stopReason: 'completed' })
await waitTerminal(ctx, started.taskId, parent)
expect(String(terminalDeliveryError)).toContain('is completed')
})
it('rejects a registry agent different from the associated run agent', async () => {
const { ctx, parent } = await setup([])
const result = Promise.withResolvers<{
output: { type: 'text'; text: string }[]
stopReason: 'completed'
}>()
ctx.subagents.registerProvider({
name: 'mismatched-local',
capabilities: { outputSchema: false, depthLimit: false, toolFilter: false, persona: false },
inheritsParentContext: false,
start: async (request) => {
const childId = request.continuation!.sessionId
const handle = await ctx.agents.create({
sessionId: childId,
meta: { parentSession: request.parent.id },
agentOptions: { provider: 'mock', model: 'mock' },
})
return {
id: childId,
localAgent: {} as Agent,
result: result.promise,
dispose: () => handle.dispose(),
}
},
resume: async () => { throw new Error('not used') },
})
const started = ctx.subagentControl.startContinuable(startSpec(parent, 'mismatched-local'))
await waitPublishedRun(ctx, started.childId)
expect(() => ctx.subagentControl.sendMessage(parent, started.childId, message('join')))
.toThrow(/registry agent is not the associated activation's agent/)
result.resolve({ output: [{ type: 'text', text: 'done' }], stopReason: 'completed' })
await waitTerminal(ctx, started.taskId, parent)
})
it('steers a running activation into the existing Task without creating a second Task', async () => {
// Hold the child's first model call open so the child is observably
// running when the message arrives; the steered content then drives a
@@ -358,7 +465,23 @@ describe('SubagentControlService.sendMessage', () => {
const attempt = ctx.subagentControl.sendMessage(parent, SessionId('plain-child'), message('continue?'))
const snapshot = await waitTerminal(ctx, attempt.taskId, parent)
expect(snapshot.status).toBe('failed')
expect(snapshot.detail).toContain('continuation descriptor')
expect(snapshot.detail).toContain(
'has no supported continuation state and cannot be resumed; do not retry send_message with this id',
)
})
it('derives fallback and bounded labels for resumed activations', async () => {
const { ctx, parent } = await setup([])
const blank = ctx.subagentControl.sendMessage(parent, SessionId('blank-child'), message(' '))
const longText = 'x'.repeat(100)
const long = ctx.subagentControl.sendMessage(parent, SessionId('long-child'), message(longText))
expect(ctx.tasks.get(blank.taskId, parent).label).toBe('subagent follow-up')
expect(ctx.tasks.get(long.taskId, parent).label).toBe(`${'x'.repeat(79)}…`)
await Promise.all([
waitTerminal(ctx, blank.taskId, parent),
waitTerminal(ctx, long.taskId, parent),
])
})
it('rejects delivery to a live agent outside control-service ownership', async () => {
@@ -491,7 +614,7 @@ describe('service disposal with live activations', () => {
await ctx.plugin(JsonlSessionPersistence, { root })
await ctx.plugin(AgentLoop, { agents: [] })
await ctx.plugin(SubagentService)
await ctx.plugin(TaskService)
await ctx.plugin(LocalTaskService)
await ctx.plugin(ToolTasks, {})
// A provider that stays pending until its signal aborts, so the activation
// is observably mid-start when the control service is disposed.
@@ -518,7 +641,7 @@ describe('service disposal with live activations', () => {
label: 'will be interrupted',
request: { prompt: message('go'), parent },
})
// TaskService keeps the producer Task; the disposing control service must
// LocalTaskService keeps the producer Task; the disposing control service must
// cancel its activation and await settlement rather than strand it.
await controlFiber.dispose()
expect(sawAbort).toBe(true)

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

@@ -276,10 +276,10 @@ function driveTurn(
if (lastBoundary?.type !== 'turn/start') {
throw new Error(`subagent child "${childId}" turn has already closed; the message was not delivered`)
}
// Terminal turn-stops only run between steps: with no step open, the
// loop may be awaiting its continuation/turn-stop checkpoints, where
// pending steering was already folded and a terminal decision discards
// a later arrival. A message accepted during an OPEN step is instead
// Turn settlement only runs between steps: with no step open, the loop
// may be awaiting its continuation/turn-stopping checkpoint, where
// pending steering was already folded and a later arrival would miss
// this turn. A message accepted during an OPEN step is instead
// drained and recorded at that step's settlement checkpoint before any
// terminal decision (cancellation remains the documented shared-outcome
// race).
@@ -289,14 +289,20 @@ function driveTurn(
if (lastStep?.type !== 'step/start') {
throw new Error(`subagent child "${childId}" is between steps; the message was not delivered`)
}
// A committed structured capture makes the pending `agent/turn-stop`
// checkpoint terminal, and the loop then discards late steering. The
// capture is synchronously observable, so reject rather than
// acknowledge a message the run is about to drop.
// A committed structured capture makes the pending step conclusion
// terminal. The capture is synchronously observable, so reject rather
// than acknowledge a message the run is about to drop.
if (structured?.captured() !== undefined) {
throw new Error(`subagent child "${childId}" already reported its structured result; the message was not delivered`)
}
child.steer(createUserMessage({ content, source: { kind: 'user' } }))
// The atomic Agent operation closes before the final drain, so this
// cannot acknowledge content that the current step will not record.
if (child.trySteer === undefined) {
throw new Error(`subagent child "${childId}" agent does not support strict steering; the message was not delivered`)
}
if (!child.trySteer(createUserMessage({ content, source: { kind: 'user' } }))) {
throw new Error(`subagent child "${childId}" passed its steering checkpoint; the message was not delivered`)
}
},
}
}

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

@@ -121,28 +121,25 @@ describe('in-process structured output', () => {
})
it('strict steer rejects delivery once the structured result is captured', async () => {
// Hold the capture's tool result open so the child is observably running
// with a committed capture: the pending agent/turn-stop checkpoint is
// terminal, and the loop would DISCARD a steering message, so an
// acknowledged delivery here would be a lie.
let releaseResult: (() => void) | undefined
const { ctx, parent } = await setup([
toolCallResponse('c1', STRUCTURED_OUTPUT_TOOL, { answer: 7 }),
])
ctx.on('agent/post-step', (agent) => {
if (agent.session.header.parentSession === undefined || releaseResult !== undefined) return
return new Promise<void>((resolve) => { releaseResult = resolve })
let run: Awaited<ReturnType<typeof ctx.subagents.start>> | undefined
let rejected: unknown
ctx.on('session/event', (session, event) => {
if (session.header.parentSession === undefined || run === undefined
|| event.type !== 'tool/result' || rejected !== undefined) return
try {
run.steer?.([{ type: 'text', text: 'one more thing' }])
} catch (error: unknown) {
rejected = error
}
})
const run = await ctx.subagents.start('spawn', structuredRequest(parent))
await new Promise<void>((resolve) => {
const timer = setInterval(() => {
if (releaseResult !== undefined) { clearInterval(timer); resolve() }
}, 5)
})
expect(() => { run.steer!([{ type: 'text', text: 'one more thing' }]) })
.toThrow(/already reported its structured result; the message was not delivered/)
releaseResult!()
run = await ctx.subagents.start('spawn', structuredRequest(parent))
const result = await run.result
expect(rejected).toBeInstanceOf(Error)
expect((rejected as Error).message)
.toMatch(/already reported its structured result; the message was not delivered/)
expect(result.structured).toEqual({ answer: 7 })
await run.dispose()
})

152
packages/subagent/subagent-inprocess/tests/subagent-inprocess.spec.ts
View File

@@ -2,16 +2,16 @@ import { createUserMessage } from '@deepseek-ai/dsh-llm'
import { describe, expect, it } from 'vitest'
import { Context } from 'cordis'
import { type Agent, type AgentOptions } from '@deepseek-ai/dsh-agent'
import { SessionId } from '@deepseek-ai/dsh-session'
import { Session, SessionId } from '@deepseek-ai/dsh-session'
import AgentLoop from '@deepseek-ai/dsh-agent-loop'
import { mountAgentLoopTestDependencies } from '@deepseek-ai/dsh-agent-loop-testkit'
import InvariantService from '@deepseek-ai/dsh-invariants'
import * as SessionInvariant from '@deepseek-ai/dsh-session/invariant'
import * as AgentInvariant from '@deepseek-ai/dsh-agent/invariant'
import * as AgentLoopInvariant from '@deepseek-ai/dsh-agent-loop/invariant'
import SubagentService from '@deepseek-ai/dsh-subagent'
import SubagentService, { SUBAGENT_DESCRIPTOR_VERSION } from '@deepseek-ai/dsh-subagent'
import { maxTokensResponse, MockAdapter, textResponse } from '../../../core/agent-loop/tests/mock-adapter.ts'
import { startInProcessRun } from '../src/index.ts'
import { resumeInProcessRun, startInProcessRun } from '../src/index.ts'
type Script = ConstructorParameters<typeof MockAdapter>[0]
@@ -186,6 +186,61 @@ describe('startInProcessRun', () => {
expect(ctx.sessions.list()).toHaveLength(beforeSessions)
})
it('rejects an already-aborted resume before publication', async () => {
const { parent } = await setup([])
const controller = new AbortController()
controller.abort('too late')
await expect(resumeInProcessRun({
sessionId: SessionId('resumed-child'),
prompt: [{ type: 'text', text: 'continue' }],
parent,
signal: controller.signal,
descriptor: { version: SUBAGENT_DESCRIPTOR_VERSION, provider: 'spawn' },
})).rejects.toThrow('aborted before child publication')
})
it('resumes without inventing undeclared agent model options', async () => {
const childId = SessionId('resumed-child')
const child = {
id: childId,
options: {},
session: new Session(childId),
status: 'idle',
acceptsNextStep: false,
ctx: new Context(),
send(): void {},
reserveTurnAdmission: () => undefined,
updateInbox: () => 'not-found',
followup(): void {},
steer(): void {},
inject(): void {},
cancel(): void {},
whenIdle: () => Promise.resolve(),
} as Agent
let resumedOptions: unknown
const parent = {
ctx: {
agents: {
resume: (options: { agentOptions: unknown }) => {
resumedOptions = options.agentOptions
return Promise.resolve({ agent: child, dispose: () => Promise.resolve() })
},
},
},
} as unknown as Agent
const run = await resumeInProcessRun({
sessionId: childId,
prompt: [{ type: 'text', text: 'continue' }],
parent,
signal: new AbortController().signal,
descriptor: { version: SUBAGENT_DESCRIPTOR_VERSION, provider: 'spawn' },
})
expect(resumedOptions).toEqual({})
await expect(run.result).resolves.toMatchObject({ stopReason: 'error' })
await run.dispose()
})
it('uses the request signal after publication and dispose as cancellation paths', async () => {
const { parent, adapter } = await setup(['hang', 'hang'])
const controller = new AbortController()
@@ -258,14 +313,12 @@ describe('startInProcessRun', () => {
await run.dispose()
})
it('strict steer rejects the between-steps window where a terminal turn-stop discards steering', async () => {
// Hold `agent/turn-stop` open: the step has closed, pending steering was
// already folded into the continuation decision, and a terminal stop
// would discard a message arriving now — the exact window an
// acknowledged delivery would be a lie.
it('strict steer rejects the between-steps turn-stopping window', async () => {
// Hold `agent/turn-stopping` open after the step closed and pending
// steering was folded into the continuation decision.
const { ctx, parent } = await setup([textResponse('quick')])
let releaseStop: (() => void) | undefined
ctx.on('agent/turn-stop', (agent) => {
ctx.on('agent/turn-stopping', (agent) => {
if (agent.session.header.parentSession === undefined || releaseStop !== undefined) return undefined
return new Promise((resolve) => {
releaseStop = () => { resolve(undefined) }
@@ -287,6 +340,87 @@ describe('startInProcessRun', () => {
await run.dispose()
})
it('strict steer rejects reentrant delivery after the final drain begins', async () => {
const { ctx, parent } = await setup([textResponse('quick')])
let run: Awaited<ReturnType<typeof startInProcessRun>> | undefined
let seeded = false
let rejected: unknown
ctx.on('session/event', (session, event) => {
if (session.header.parentSession === undefined || run === undefined) return
if (event.type === 'assistant/chunk' && !seeded) {
seeded = true
run.steer?.([{ type: 'text', text: 'accepted before the drain' }])
} else if (event.type === 'steering/message' && rejected === undefined) {
try {
run.steer?.([{ type: 'text', text: 'after the drain began' }])
} catch (error: unknown) {
rejected = error
}
}
})
run = await startInProcessRun(request(parent), {})
const child = ctx.agents.get(run.id)!
await run.result
expect(seeded).toBe(true)
expect(rejected).toBeInstanceOf(Error)
expect((rejected as Error).message)
.toMatch(/passed its steering checkpoint; the message was not delivered/)
expect(child.session.events.filter(event => event.type === 'steering/message')).toHaveLength(1)
await run.dispose()
})
it('strict steer rejects an Agent implementation without atomic steering', async () => {
const childId = SessionId('custom-loop-child')
const childSession = new Session(childId)
childSession.append('turn/start', {
turn: 1,
trigger: { kind: 'message', source: { kind: 'user' } },
})
childSession.append('step/start', { turn: 1, step: 1 })
const idle = Promise.withResolvers<undefined>()
const child = {
id: childId,
options: {},
session: childSession,
status: 'running',
acceptsNextStep: false,
ctx: new Context(),
send(): void {},
reserveTurnAdmission: () => undefined,
updateInbox: () => 'not-found',
followup(): void {},
steer(): void {},
inject(): void {},
cancel(): void {},
whenIdle: () => idle.promise,
} as Agent
const parentId = SessionId('custom-loop-parent')
const parent = {
id: parentId,
options: {},
session: new Session(parentId),
ctx: {
get: () => undefined,
agents: {
create: () => Promise.resolve({
agent: child,
dispose: () => {
idle.resolve(undefined)
return Promise.resolve()
},
}),
},
},
} as unknown as Agent
const run = await startInProcessRun(request(parent), {})
expect(() => { run.steer!([{ type: 'text', text: 'unsupported strict delivery' }]) })
.toThrow(/does not support strict steering; the message was not delivered/)
await run.dispose()
await run.result
})
it('strict steer rejects the closed-turn flush window where the loop discards steering', async () => {
// Hold the turn-end durability flush open: the turn has closed in the log
// and status is still `running`, exactly the window where the loop would

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

@@ -5,6 +5,9 @@ import { type Agent } from '@deepseek-ai/dsh-agent'
import { HarnessError } from '@deepseek-ai/dsh-llm'
import { carrierKeyOf } from '@deepseek-ai/dsh-scope'
import SubagentService, {
foldSubagentDescriptor,
snapshotSubagentDescriptor,
SUBAGENT_DESCRIPTOR_VERSION,
SubagentError,
assertSubagentMaxDepth,
type SubagentCapabilities,
@@ -13,7 +16,7 @@ import SubagentService, {
type SubagentRun,
type SubagentStartRequest,
} from '@deepseek-ai/dsh-subagent'
import { SessionId } from '@deepseek-ai/dsh-session'
import { SessionId, type SessionEvent } from '@deepseek-ai/dsh-session'
function fakeParent(id = 'parent-1'): Agent {
return { id: SessionId(id) } as unknown as Agent
@@ -99,6 +102,28 @@ describe('SubagentService', () => {
.rejects.toMatchObject({ code: 'NO_PROVIDER' })
})
it('rejects continuable start and resume when the provider has no resume capability', async () => {
const { subagents } = await service()
subagents.registerProvider(new StubProvider('one-shot'))
const descriptor = snapshotSubagentDescriptor({ provider: 'one-shot' })
const sessionId = SessionId('continuable-child')
const parent = fakeParent()
const signal = new AbortController().signal
await expect(subagents.start('one-shot', baseRequest({
parent,
signal,
continuation: { sessionId, descriptor },
}))).rejects.toMatchObject({ code: 'UNSUPPORTED_CAPABILITY' })
await expect(subagents.resume('one-shot', {
sessionId,
prompt: [{ type: 'text', text: 'continue' }],
parent,
signal,
descriptor,
})).rejects.toMatchObject({ code: 'UNSUPPORTED_CAPABILITY' })
})
it.each([
['outputSchema', { outputSchema: { type: 'object', properties: {} } }],
['depthLimit', { maxDepth: 1 }],
@@ -247,3 +272,17 @@ describe('SubagentService', () => {
expect(error.code).toBe('NO_PROVIDER')
})
})
describe('subagent descriptors', () => {
it('omits absent model selectors and rejects unsupported versions', () => {
expect(snapshotSubagentDescriptor({ provider: 'spawn' })).toEqual({
version: SUBAGENT_DESCRIPTOR_VERSION,
provider: 'spawn',
})
const unsupported = {
type: 'subagent/descriptor',
data: { version: SUBAGENT_DESCRIPTOR_VERSION + 1, provider: 'spawn' },
} as unknown as SessionEvent<'subagent/descriptor'>
expect(foldSubagentDescriptor([unsupported])).toBeUndefined()
})
})