test(subagent): update service, send_message, and delegation specs

The continuable path has no Task, so send_message reports a queued next turn
and continuable delegation returns only the durable child id. Pins that a
follow-up queues behind an open turn rather than steering it, and that a
non-parent caller is rejected.

Also makes startContinuable/followup reject rather than throw synchronously
when continuation services are absent, so callers have one failure mode.
This commit is contained in:
Dudu-0223
2026-07-30 13:33:46 +08:00
committed by Tianyi Cui
parent 1c30548068
commit 357f317b4c
4 changed files with 101 additions and 106 deletions

View File

@@ -198,7 +198,7 @@ export class SubagentService extends Service {
* @returns the durable child id and the accepted prompt's message id. * @returns the durable child id and the accepted prompt's message id.
* @throws when continuation services are unavailable or materialization fails. * @throws when continuation services are unavailable or materialization fails.
*/ */
startContinuable(spec: ContinuableStartSpec): Promise<ContinuableStart> { async startContinuable(spec: ContinuableStartSpec): Promise<ContinuableStart> {
return this.requireContinuations().startContinuable(spec) return this.requireContinuations().startContinuable(spec)
} }
@@ -217,7 +217,7 @@ export class SubagentService extends Service {
* @throws when continuation services are unavailable, authority is rejected, * @throws when continuation services are unavailable, authority is rejected,
* or the message was not admitted. * or the message was not admitted.
*/ */
followup( async followup(
authority: SubagentAuthority, authority: SubagentAuthority,
childId: SessionId, childId: SessionId,
content: ContentBlock[], content: ContentBlock[],
@@ -342,7 +342,7 @@ export class SubagentService extends Service {
private requireContinuations(): SubagentContinuationManager { private requireContinuations(): SubagentContinuationManager {
if (this.continuations === undefined) { if (this.continuations === undefined) {
throw new SubagentError( throw new SubagentError(
'continuable subagents require the tasks and agents services', 'continuable subagents require the agents service',
'CONTINUATION_UNAVAILABLE', 'CONTINUATION_UNAVAILABLE',
) )
} }

View File

@@ -12,7 +12,6 @@ import SubagentService, {
assertSubagentMaxDepth, assertSubagentMaxDepth,
type SubagentCapabilities, type SubagentCapabilities,
type SubagentProvider, type SubagentProvider,
type SubagentProviderStartRequest,
type SubagentResult, type SubagentResult,
type SubagentRun, type SubagentRun,
type SubagentStartRequest, type SubagentStartRequest,
@@ -38,7 +37,7 @@ function baseRequest(overrides: Partial<SubagentStartRequest> = {}): SubagentSta
class StubProvider implements SubagentProvider { class StubProvider implements SubagentProvider {
readonly inheritsParentContext = false readonly inheritsParentContext = false
startCount = 0 startCount = 0
lastRequest: SubagentProviderStartRequest | undefined lastRequest: SubagentStartRequest | undefined
constructor( constructor(
readonly name: string, readonly name: string,
@@ -49,7 +48,7 @@ class StubProvider implements SubagentProvider {
}, },
) {} ) {}
async start(request: SubagentProviderStartRequest): Promise<SubagentRun> { async start(request: SubagentStartRequest): Promise<SubagentRun> {
this.startCount += 1 this.startCount += 1
this.lastRequest = request this.lastRequest = request
return { return {
@@ -112,21 +111,27 @@ describe('SubagentService', () => {
const request = baseRequest() const request = baseRequest()
await subagents.start('one-shot', request) await subagents.start('one-shot', request)
// One-shot start borrows the caller's exact request; the seam has no
// provider-facing resume or steer surface to dispatch through.
expect(provider.lastRequest).toBe(request) expect(provider.lastRequest).toBe(request)
expectTypeOf<SubagentProviderStartRequest>() expectTypeOf<Parameters<SubagentService['start']>[1]>().toExtend<SubagentStartRequest>()
.not.toExtend<Parameters<SubagentService['start']>[1]>()
expect('resume' in subagents).toBe(false) expect('resume' in subagents).toBe(false)
expect('resume' in provider).toBe(false)
}) })
it('rejects Task-backed continuation operations when their runtime services are absent', async () => { it('rejects continuable operations when their runtime services are absent', async () => {
const { subagents } = await service() const { subagents } = await service()
expect(() => { await expect(subagents.startContinuable({
subagents.startContinuable({ provider: 'unused',
provider: 'unused', request: baseRequest(),
label: 'work', signal: new AbortController().signal,
request: baseRequest(), })).rejects.toMatchObject({ code: 'CONTINUATION_UNAVAILABLE' })
}) await expect(subagents.followup(
}).toThrow(expect.objectContaining({ code: 'CONTINUATION_UNAVAILABLE' })) { kind: 'user' },
SessionId('child'),
[{ type: 'text', text: 'hello' }],
{ source: { kind: 'user' }, signal: new AbortController().signal },
)).rejects.toMatchObject({ code: 'CONTINUATION_UNAVAILABLE' })
}) })
it.each([ it.each([

View File

@@ -10,8 +10,6 @@ import { SessionId } from '@deepseek-ai/dsh-session'
import JsonlSessionPersistence from '@deepseek-ai/dsh-session-persistence-jsonl' import JsonlSessionPersistence from '@deepseek-ai/dsh-session-persistence-jsonl'
import SubagentService from '@deepseek-ai/dsh-subagent' import SubagentService from '@deepseek-ai/dsh-subagent'
import * as SubagentSpawn from '@deepseek-ai/dsh-subagent-spawn' import * as SubagentSpawn from '@deepseek-ai/dsh-subagent-spawn'
import LocalTaskService from '@deepseek-ai/dsh-tasks-local'
import * as ToolTasks from '@deepseek-ai/dsh-tool-tasks'
import { MockAdapter, textResponse } from '../../../core/agent-loop/tests/mock-adapter.ts' import { MockAdapter, textResponse } from '../../../core/agent-loop/tests/mock-adapter.ts'
import * as tool from '../src/index.ts' import * as tool from '../src/index.ts'
@@ -31,8 +29,6 @@ async function setup(script: ConstructorParameters<typeof MockAdapter>[0]) {
await ctx.plugin(AgentLoop, { agents: [] }) await ctx.plugin(AgentLoop, { agents: [] })
await ctx.plugin(SubagentService) await ctx.plugin(SubagentService)
await ctx.plugin(SubagentSpawn, { providerName: 'spawn' }) await ctx.plugin(SubagentSpawn, { providerName: 'spawn' })
await ctx.plugin(LocalTaskService)
await ctx.plugin(ToolTasks, {})
await ctx.plugin(tool) await ctx.plugin(tool)
const adapter = new MockAdapter(script) const adapter = new MockAdapter(script)
ctx.llm.registerAdapter(['mock'], adapter) ctx.llm.registerAdapter(['mock'], adapter)
@@ -61,6 +57,13 @@ function callTool(
}) })
} }
/** Wait until a child's Activation released its handle. */
async function waitNoActivation(ctx: Context, childId: SessionId): Promise<void> {
await vi.waitFor(() => {
expect(ctx.agents.get(childId)).toBeUndefined()
}, { timeout: 5_000 })
}
describe('dsh-tool-subagent-control', () => { describe('dsh-tool-subagent-control', () => {
it('registers send_message once, globally, with the two required parameters', async () => { it('registers send_message once, globally, with the two required parameters', async () => {
const { ctx } = await setup([]) const { ctx } = await setup([])
@@ -68,90 +71,62 @@ describe('dsh-tool-subagent-control', () => {
expect(schemas).toHaveLength(1) expect(schemas).toHaveLength(1)
const props = (schemas[0]!.parameters as { properties?: Record<string, unknown> }).properties ?? {} const props = (schemas[0]!.parameters as { properties?: Record<string, unknown> }).properties ?? {}
expect(Object.keys(props).sort()).toEqual(['message', 'subagent_id']) expect(Object.keys(props).sort()).toEqual(['message', 'subagent_id'])
expect(schemas[0]!.description).toContain('task_output') // The continuable path has no Task, so the schema must not promise one.
expect(schemas[0]!.description).not.toContain('task_output')
expect(schemas[0]!.description).not.toContain('task id')
// Follow-up ordering is model-visible: it cannot redirect the open turn.
expect(schemas[0]!.description).toContain('next turn')
}) })
it('cold-resumes a settled child and renders the started route with its task id', async () => { it('cold-resumes a settled child and reports the queued next turn', async () => {
const { ctx, parent } = await setup([textResponse('first answer'), textResponse('second answer')]) const { ctx, parent } = await setup([textResponse('first answer'), textResponse('second answer')])
const started = ctx.subagents.startContinuable({ const started = await ctx.subagents.startContinuable({
provider: 'spawn', provider: 'spawn',
label: 'work',
request: { prompt: [{ type: 'text', text: 'child task' }], parent }, request: { prompt: [{ type: 'text', text: 'child task' }], parent },
signal: testToolSignal,
}) })
await ctx.tasks.wait(started.taskId, 5_000, parent) await waitNoActivation(ctx, started.childId)
const result = await callTool(ctx, 'send_message', { const result = await callTool(ctx, 'send_message', {
subagent_id: started.childId, subagent_id: started.childId,
message: 'and then?', message: 'and then?',
}, parent) }, parent)
expect(result.isError).toBe(false) expect(result.isError).toBe(false)
expect(text(result)).toBe(`message started task subagent-2 continuing subagent ${started.childId}`) expect(text(result)).toBe(`message queued as the next turn for subagent ${started.childId}`)
const collected = await callTool(ctx, 'task_output', { task_id: 'subagent-2', wait: true }, parent) await waitNoActivation(ctx, started.childId)
expect(text(collected)).toBe('second answer\n[status: completed]')
const loaded = await ctx.sessionPersistence.load(started.childId) const loaded = await ctx.sessionPersistence.load(started.childId)
const followUp = loaded.events.findLast(event => const followUp = loaded.events.findLast(event => event.type === 'user/message')
event.type === 'user/message', // Durable provenance records the calling agent without granting authority.
)
expect(followUp?.type === 'user/message' && followUp.data.source).toEqual({ expect(followUp?.type === 'user/message' && followUp.data.source).toEqual({
kind: 'coordinator', kind: 'coordinator',
senderSessionId: parent.id, senderSessionId: parent.id,
}) })
}) })
it('renders the steered route when the child is still running', async () => { it('queues behind an open turn instead of joining it', async () => {
// Script the child's single turn as two steps: the steer joins mid-turn. const { ctx, parent, adapter } = await setup([textResponse('first'), textResponse('second')])
const { ctx, parent } = await setup([]) const started = await ctx.subagents.startContinuable({
let steered: string | undefined provider: 'spawn',
let source: unknown request: { prompt: [{ type: 'text', text: 'long work' }], parent },
// Reach past the tool into the subagent service to fake a running route signal: testToolSignal,
// deterministically: the tool is a thin adapter, so its steered wording is })
// what this test pins. await vi.waitFor(() => { expect(adapter.requests).toHaveLength(1) })
ctx.subagents.followup = async (agent, _childId, message, options) => {
steered = (message[0] as { text: string }).text
source = options.source
return { route: 'steered', taskId: ctx.tasks.list(agent)[0]?.id ?? ('subagent-9' as never) }
}
const result = await callTool(ctx, 'send_message', { const result = await callTool(ctx, 'send_message', {
subagent_id: 'some-child', subagent_id: started.childId,
message: 'also consider Y', message: 'also consider Y',
}, parent) }, parent)
expect(result.isError).toBe(false) expect(result.isError).toBe(false)
expect(steered).toBe('also consider Y')
expect(source).toEqual({ kind: 'coordinator', senderSessionId: parent.id })
expect(text(result)).toBe('message delivered to running task subagent-9')
})
it('cancels a pending live-delivery wait when the tool signal aborts', async () => { await waitNoActivation(ctx, started.childId)
const { ctx, parent, adapter } = await setup(['hang'])
const started = ctx.subagents.startContinuable({
provider: 'spawn',
label: 'hung work',
request: { prompt: [{ type: 'text', text: 'wait' }], parent },
})
await vi.waitFor(() => { expect(adapter.requests).toHaveLength(1) })
const deliveryStarted: PromiseWithResolvers<void> = Promise.withResolvers()
const followup = ctx.subagents.followup.bind(ctx.subagents)
ctx.subagents.followup = (agent, childId, message, options) => {
const delivery = followup(agent, childId, message, options)
deliveryStarted.resolve()
return delivery
}
const controller = new AbortController()
const execution = callTool(ctx, 'send_message', {
subagent_id: started.childId,
message: 'follow up',
}, parent, controller.signal)
await deliveryStarted.promise
controller.abort('parent tool cancelled')
const result = await execution
expect(result.isError).toBe(true)
expect(result.error?.info?.code).toBe('CANCELLED')
expect(ctx.agents.get(started.childId)).toBeUndefined()
const snapshot = await ctx.tasks.wait(started.taskId, 5_000, parent)
expect(snapshot.status).toBe('killed')
const loaded = await ctx.sessionPersistence.load(started.childId) const loaded = await ctx.sessionPersistence.load(started.childId)
const prompts = loaded.events.flatMap(event => event.type === 'user/message'
? event.data.content.flatMap(block => block.type === 'text' ? [block.text] : [])
: [])
// A follow-up is its own later turn, never steering inside the first one.
expect(prompts).toEqual(['long work', 'also consider Y'])
expect(loaded.events.some(event => event.type === 'steering/message')).toBe(false) expect(loaded.events.some(event => event.type === 'steering/message')).toBe(false)
}) })
@@ -161,17 +136,26 @@ describe('dsh-tool-subagent-control', () => {
subagent_id: 'no-such-child', subagent_id: 'no-such-child',
message: 'hello?', message: 'hello?',
}, parent) }, parent)
// Unknown ids start a Task whose failure carries the unavailable detail; expect(result.isError).toBe(true)
// synchronous rejections (ownership conflicts) become isError results. expect(text(result)).toContain('unavailable')
if (result.isError) { })
expect(text(result)).toContain('not delivered')
} else { it('rejects a caller that is not the child\'s durable direct parent', async () => {
const taskId = text(result).match(/task (\S+) /)?.[1] const { ctx, parent } = await setup([textResponse('first')])
expect(taskId).toBeDefined() const started = await ctx.subagents.startContinuable({
const snapshot = await ctx.tasks.wait(taskId as never, 5_000, parent) provider: 'spawn',
expect(snapshot.status).toBe('failed') request: { prompt: [{ type: 'text', text: 'child task' }], parent },
expect(snapshot.detail).toContain('unavailable') signal: testToolSignal,
} })
await waitNoActivation(ctx, started.childId)
const stranger = ctx.agentLoop.create(SessionId('stranger'), { provider: 'mock', model: 'mock' })
const result = await callTool(ctx, 'send_message', {
subagent_id: started.childId,
message: 'mine now',
}, stranger)
expect(result.isError).toBe(true)
expect(text(result)).toContain('another parent session')
}) })
it('fails loud when invoked without a calling agent', async () => { it('fails loud when invoked without a calling agent', async () => {
@@ -186,7 +170,6 @@ describe('dsh-tool-subagent-control', () => {
await mountAgentLoopTestDependencies(ctx) await mountAgentLoopTestDependencies(ctx)
await ctx.plugin(AgentLoop, { agents: [] }) await ctx.plugin(AgentLoop, { agents: [] })
await ctx.plugin(SubagentService) await ctx.plugin(SubagentService)
await ctx.plugin(LocalTaskService)
const fiber = await ctx.plugin(tool) const fiber = await ctx.plugin(tool)
expect(ctx.tools.schemas().some(schema => schema.name === 'send_message')).toBe(true) expect(ctx.tools.schemas().some(schema => schema.name === 'send_message')).toBe(true)
await fiber.dispose() await fiber.dispose()

View File

@@ -68,7 +68,7 @@ function text(result: { content: { type: string; text?: string }[] }): string {
} }
describe('dsh-tool-subagent', () => { describe('dsh-tool-subagent', () => {
it('rejects continuable background policy when the configured provider cannot resume', async () => { it('rejects continuable background policy when the provider cannot prepare continuable children', async () => {
let failure: unknown let failure: unknown
try { try {
await setup({ await setup({
@@ -668,10 +668,10 @@ describe('dsh-tool-subagent background mode', () => {
return ctx return ctx
} }
it('keeps a resumable provider one-shot when backgroundMode selects one-shot', async () => { it('keeps a continuable-capable provider one-shot when backgroundMode selects one-shot', async () => {
const ctx = await backgroundSetup({ provider: 'mock' }) const ctx = await backgroundSetup({ provider: 'mock' })
const parent = ownerAgent(ctx, 'sess-parent') const parent = ownerAgent(ctx, 'sess-parent')
let resumeCalls = 0 let prepareCalls = 0
ctx.subagents.registerProvider({ ctx.subagents.registerProvider({
name: 'resumable', name: 'resumable',
capabilities: { outputSchema: false, depthLimit: false, toolFilter: false, persona: false }, capabilities: { outputSchema: false, depthLimit: false, toolFilter: false, persona: false },
@@ -685,9 +685,9 @@ describe('dsh-tool-subagent background mode', () => {
}), }),
dispose: () => Promise.resolve(), dispose: () => Promise.resolve(),
}), }),
resume: async () => { prepareContinuable: async () => {
resumeCalls += 1 prepareCalls += 1
throw new Error('one-shot policy must not resume') throw new Error('one-shot policy must not prepare a continuable child')
}, },
}) })
tool.apply(ctx, { tool.apply(ctx, {
@@ -706,7 +706,7 @@ describe('dsh-tool-subagent background mode', () => {
}) })
expect(text(started)).toBe('started background subagent task subagent-1') expect(text(started)).toBe('started background subagent task subagent-1')
expect(resumeCalls).toBe(0) expect(prepareCalls).toBe(0)
}) })
it('returns a task id immediately and the answer is collected through task_output', async () => { it('returns a task id immediately and the answer is collected through task_output', async () => {
@@ -899,10 +899,13 @@ describe('dsh-tool-subagent continuable background mode', () => {
return { ctx, parent } return { ctx, parent }
} }
it('starts a continuable child and returns both ids without send_message', async () => { it('starts a continuable child and returns only its durable id, creating no Task', async () => {
const { ctx, parent } = await continuableSetup() const { ctx, parent } = await continuableSetup()
const schema = ctx.tools.schemas().find(s => s.name === 'subagent')! const schema = ctx.tools.schemas().find(s => s.name === 'subagent')!
expect(schema.description).not.toContain('send_message') // Continuable delegation has no Task, so the schema promises no collection.
expect(schema.description).not.toContain('task_output')
expect(schema.description).not.toContain('task_kill')
expect(schema.description).toContain('send_message')
const started = await callSubagent( const started = await callSubagent(
ctx, ctx,
@@ -910,15 +913,19 @@ describe('dsh-tool-subagent continuable background mode', () => {
{ agent: parent }, { agent: parent },
) )
expect(started.isError).toBe(false) expect(started.isError).toBe(false)
const match = /^started subagent (\S+) as task (\S+)$/.exec(text(started)) const match = /^started subagent (\S+)$/.exec(text(started))
expect(match).not.toBeNull() expect(match).not.toBeNull()
const [, childId, taskId] = match! const [, childId] = match!
const snapshot = await ctx.tasks.wait(taskId as never, 5_000, parent) // No Task was created for the continuable child.
expect(snapshot.status).toBe('completed') expect(ctx.tasks.list(parent)).toEqual([])
expect(ctx.tasks.read(taskId as never, parent).text).toBe('continuable answer')
// The child id names a durable session that outlives the settled Task. await vi.waitFor(() => {
expect(ctx.agents.get(SessionId(childId!))).toBeUndefined()
}, { timeout: 5_000 })
// The child id names a durable session carrying its continuation descriptor.
const loaded = await ctx.sessionPersistence.load(SessionId(childId!)) const loaded = await ctx.sessionPersistence.load(SessionId(childId!))
expect(loaded.events.some(event => event.type === 'subagent/descriptor')).toBe(true) expect(loaded.events.some(event => event.type === 'subagent/descriptor')).toBe(true)
expect(loaded.events.some(event => event.type === 'assistant/message')).toBe(true)
}) })
}) })