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

feat(subagent): let ancestors interrupt descendants

Browse Source 
interrupt_agent(agent_id) passes the calling agent as the ancestor
authority for ctx.subagents.interrupt(); the service verifies live
registry identity and recorded lineage, so a direct child or deeper
descendant stops with the same generic parameter while send_message keeps
its exact-direct-parent authority.

Discovery: list_agents gains an optional scope. descendants walks the new
SubagentService.listDescendants() — one lineage trace flattened in stable
pre-order across ordinary and one-shot intermediates, each entry carrying
its verified parentId and depth — and every status now comes from the
live Agent registry (running/idle/complete).

Refs #1535
This commit is contained in:
Hypatia May
2026-08-06 13:46:55 +08:00
committed by Tianyi Cui
parent 4b29f9ca7a
commit d769d3cbb7
41 changed files with 1325 additions and 162 deletions
Download Patch File Download Diff File Expand all files Collapse all files

165
packages/subagent/tool-subagent-control/tests/list-agents.spec.ts
View File

@@ -12,9 +12,37 @@ import SessionProjectionRegistry from '@deepseek-ai/dsh-session-projection'
import SubagentService from '@deepseek-ai/dsh-subagent'
import type { SubagentListEntry } from '@deepseek-ai/dsh-subagent'
import * as SubagentSpawn from '@deepseek-ai/dsh-subagent-spawn'
import type { GenerateOptions, StreamChunk } from '@deepseek-ai/dsh-llm'
import { LlmAdapter } from '@deepseek-ai/dsh-llm'
import { MockAdapter, textResponse } from '../../../core/agent-loop/tests/mock-adapter.ts'
import * as tool from '../src/list-agents.ts'
/** One scripted response that may wait on a caller-released gate before streaming. */
interface GatedEntry {
chunks: StreamChunk[]
gate?: Promise<undefined>
}
/** Adapter whose entries can hold a model call open until the test releases it. */
class GatedAdapter extends LlmAdapter {
readonly requests: GenerateOptions[] = []
constructor(private script: GatedEntry[]) {
super()
}
async * stream(options: GenerateOptions): AsyncIterable<StreamChunk> {
this.requests.push(options)
const entry = this.script.shift()
if (!entry) throw new Error('GatedAdapter: script exhausted')
if (entry.gate) await entry.gate
for (const chunk of entry.chunks) {
if (options.signal?.aborted) throw new Error('aborted')
yield chunk
}
}
}
const testToolSignal = new AbortController().signal
const roots: string[] = []
@@ -22,7 +50,7 @@ afterEach(() => {
for (const root of roots.splice(0)) rmSync(root, { recursive: true, force: true })
})
async function setup(script: ConstructorParameters<typeof MockAdapter>[0]) {
async function setupWith(adapter: MockAdapter | GatedAdapter) {
const ctx = new Context()
await mountAgentLoopTestDependencies(ctx)
const root = mkdtempSync(join(tmpdir(), 'dsh-tool-list-agents-'))
@@ -33,9 +61,13 @@ async function setup(script: ConstructorParameters<typeof MockAdapter>[0]) {
await ctx.plugin(SubagentService)
await ctx.plugin(SubagentSpawn, { providerName: 'spawn' })
await ctx.plugin(tool)
ctx.llm.registerAdapter(['mock'], new MockAdapter(script))
ctx.llm.registerAdapter(['mock'], adapter)
const parent = ctx.agentLoop.create(SessionId('parent'), { provider: 'mock', model: 'mock' })
return { ctx, parent }
return { ctx, parent, adapter }
}
async function setup(script: ConstructorParameters<typeof MockAdapter>[0]) {
return setupWith(new MockAdapter(script))
}
function text(result: { content: { type: string; text?: string }[] }): string {
@@ -67,13 +99,19 @@ async function waitNoActivation(ctx: Context, childId: SessionId): Promise<void>
}
describe('dsh-tool-subagent-control/list-agents', () => {
it('registers list_agents once, globally, with no parameters', async () => {
it('registers list_agents once, globally, with only the optional scope parameter', async () => {
const { ctx } = await setup([])
const schemas = ctx.tools.schemas().filter(schema => schema.name === 'list_agents')
expect(schemas).toHaveLength(1)
const props = (schemas[0]!.parameters as { properties?: Record<string, unknown> }).properties ?? {}
expect(Object.keys(props)).toEqual([])
const parameters = schemas[0]!.parameters as {
properties?: Record<string, { enum?: string[] }>
required?: string[]
}
expect(Object.keys(parameters.properties ?? {})).toEqual(['scope'])
expect(parameters.properties?.scope?.enum).toEqual(['children', 'descendants'])
expect(parameters.required ?? []).toEqual([])
expect(schemas[0]!.description).toContain('send_message')
expect(schemas[0]!.description).toContain('interrupt_agent')
})
it('renders the empty result as (no subagents)', async () => {
@@ -84,7 +122,7 @@ describe('dsh-tool-subagent-control/list-agents', () => {
expect(text(result)).toBe('(no subagents)')
})
it('renders children and diagnostics in array order with the fixed text forms', async () => {
it('renders children and diagnostics in array order with registry-derived statuses', async () => {
const { ctx, parent } = await setup([textResponse('done')])
const started = await ctx.subagents.startContinuable({
provider: 'spawn',
@@ -94,7 +132,8 @@ describe('dsh-tool-subagent-control/list-agents', () => {
})
await waitNoActivation(ctx, started.childId)
// Pin the render deterministically past the service: the tool is a thin
// adapter, so its fixed text forms are what this test pins.
// adapter, so its fixed text forms are what this test pins. Status comes
// from the live Agent registry, stubbed per candidate id.
const entries: SubagentListEntry[] = [
{
kind: 'child',
@@ -120,14 +159,28 @@ describe('dsh-tool-subagent-control/list-agents', () => {
activity: 'running',
hasChildren: true,
},
{
kind: 'child',
id: SessionId('waiting-child'),
label: 'waiting on descendants',
mode: 'continuable',
activity: 'running',
hasChildren: true,
},
{ kind: 'diagnostic', id: SessionId('broken-child'), reason: 'corrupt' },
]
ctx.subagents.listChildren = () => Promise.resolve(entries)
const agents = new Map<string, { status: 'running' | 'idle' }>([
['running-child', { status: 'running' }],
['waiting-child', { status: 'idle' }],
])
vi.spyOn(ctx.agents, 'get').mockImplementation(id => agents.get(id) as never)
const result = await callTool(ctx, 'list_agents', {}, parent)
expect(result.isError).toBe(false)
expect(text(result)).toBe(
`${started.childId} [complete] — real child\n`
+ 'running-child [running] — still working\n'
+ 'waiting-child [idle] — waiting on descendants\n'
+ 'broken-child [diagnostic: corrupt]',
)
})
@@ -186,7 +239,101 @@ describe('dsh-tool-subagent-control/list-agents', () => {
it('has the namespace-plugin export shape', () => {
expect('default' in tool).toBe(false)
expect(tool.name).toBe('tool-subagent-list-agents')
expect(tool.inject).toEqual(['tools', 'subagents'])
expect(tool.inject).toEqual(['tools', 'subagents', 'agents'])
expect(typeof tool.apply).toBe('function')
})
it('walks the complete descendant tree in pre-order with parent and depth annotations', async () => {
const releaseChild = Promise.withResolvers<undefined>()
const releaseGrandchild = Promise.withResolvers<undefined>()
const adapter = new GatedAdapter([
{ chunks: textResponse('child'), gate: releaseChild.promise },
{ chunks: textResponse('grandchild'), gate: releaseGrandchild.promise },
])
const { ctx, parent } = await setupWith(adapter)
const started = await ctx.subagents.startContinuable({
provider: 'spawn',
label: 'waiting branch',
request: { prompt: [{ type: 'text', text: 'branch work' }], parent },
signal: testToolSignal,
})
await vi.waitFor(() => { expect(adapter.requests).toHaveLength(1) })
const child = ctx.agents.get(started.childId)!
const grandchild = await ctx.subagents.startContinuable({
provider: 'spawn',
label: 'nested leaf',
request: { prompt: [{ type: 'text', text: 'leaf work' }], parent: child },
signal: testToolSignal,
})
await vi.waitFor(() => { expect(adapter.requests).toHaveLength(2) })
// The branch finishes its own turn but stays resident waiting on the
// grandchild it owns: the live-registry `idle` status.
releaseChild.resolve(undefined)
await vi.waitFor(() => {
expect(ctx.agents.get(started.childId)?.status).toBe('idle')
}, { timeout: 5_000 })
const result = await callTool(ctx, 'list_agents', { scope: 'descendants' }, parent)
expect(result.isError).toBe(false)
expect(text(result)).toBe(
`${started.childId} [idle] parent=${parent.id} depth=1 — waiting branch\n`
+ `${grandchild.childId} [running] parent=${started.childId} depth=2 — nested leaf`,
)
releaseGrandchild.resolve(undefined)
await waitNoActivation(ctx, grandchild.childId)
await waitNoActivation(ctx, started.childId)
})
it('omits one-shot intermediates from descendants output while surfacing what they own', async () => {
const { ctx, parent } = await setup([])
// Deterministic service rows: a one-shot intermediate owning a continuable
// leaf, plus a positioned diagnostic. The tool filters only the one-shot.
ctx.subagents.listDescendants = () => Promise.resolve([
{
kind: 'child',
id: SessionId('one-shot-mid'),
label: 'one-shot intermediate',
mode: 'one-shot',
activity: 'inactive',
hasChildren: true,
parentId: parent.id,
depth: 1,
},
{
kind: 'child',
id: SessionId('deep-leaf'),
label: 'deep leaf',
mode: 'continuable',
activity: 'inactive',
hasChildren: false,
parentId: SessionId('one-shot-mid'),
depth: 2,
},
{
kind: 'diagnostic',
id: SessionId('broken-node'),
reason: 'unavailable',
parentId: parent.id,
depth: 1,
},
])
const result = await callTool(ctx, 'list_agents', { scope: 'descendants' }, parent)
expect(result.isError).toBe(false)
expect(text(result)).toBe(
'deep-leaf [complete] parent=one-shot-mid depth=2 — deep leaf\n'
+ `broken-node [diagnostic: unavailable] parent=${parent.id} depth=1`,
)
})
it('forwards the tool cancellation signal to descendant enumeration', async () => {
const { ctx, parent } = await setup([])
const signal = new AbortController().signal
const listDescendants = vi.spyOn(ctx.subagents, 'listDescendants').mockResolvedValue([])
const result = await callTool(ctx, 'list_agents', { scope: 'descendants' }, parent, signal)
expect(result.isError).toBe(false)
expect(listDescendants).toHaveBeenCalledWith(parent.id, signal)
})
})

202
packages/subagent/tool-subagent-control/tests/tool-subagent-control.spec.ts
View File

@@ -11,9 +11,37 @@ import JsonlSessionPersistence from '@deepseek-ai/dsh-session-persistence-jsonl'
import SessionProjectionRegistry from '@deepseek-ai/dsh-session-projection'
import SubagentService from '@deepseek-ai/dsh-subagent'
import * as SubagentSpawn from '@deepseek-ai/dsh-subagent-spawn'
import type { GenerateOptions, StreamChunk } from '@deepseek-ai/dsh-llm'
import { LlmAdapter } from '@deepseek-ai/dsh-llm'
import { MockAdapter, textResponse } from '../../../core/agent-loop/tests/mock-adapter.ts'
import * as tool from '../src/index.ts'
/** One scripted response that may wait on a caller-released gate before streaming. */
interface GatedEntry {
chunks: StreamChunk[]
gate?: Promise<undefined>
}
/** Adapter whose entries can hold a model call open until the test releases it. */
class GatedAdapter extends LlmAdapter {
readonly requests: GenerateOptions[] = []
constructor(private script: GatedEntry[]) {
super()
}
async * stream(options: GenerateOptions): AsyncIterable<StreamChunk> {
this.requests.push(options)
const entry = this.script.shift()
if (!entry) throw new Error('GatedAdapter: script exhausted')
if (entry.gate) await entry.gate
for (const chunk of entry.chunks) {
if (options.signal?.aborted) throw new Error('aborted')
yield chunk
}
}
}
const testToolSignal = new AbortController().signal
const roots: string[] = []
@@ -21,7 +49,7 @@ afterEach(() => {
for (const root of roots.splice(0)) rmSync(root, { recursive: true, force: true })
})
async function setup(script: ConstructorParameters<typeof MockAdapter>[0]) {
async function setupWith(adapter: MockAdapter | GatedAdapter) {
const ctx = new Context()
await mountAgentLoopTestDependencies(ctx)
const root = mkdtempSync(join(tmpdir(), 'dsh-tool-subagent-control-'))
@@ -32,12 +60,15 @@ async function setup(script: ConstructorParameters<typeof MockAdapter>[0]) {
await ctx.plugin(SubagentService)
await ctx.plugin(SubagentSpawn, { providerName: 'spawn' })
await ctx.plugin(tool)
const adapter = new MockAdapter(script)
ctx.llm.registerAdapter(['mock'], adapter)
const parent = ctx.agentLoop.create(SessionId('parent'), { provider: 'mock', model: 'mock' })
return { ctx, parent, adapter }
}
async function setup(script: ConstructorParameters<typeof MockAdapter>[0]) {
return setupWith(new MockAdapter(script))
}
function text(result: { content: { type: string; text?: string }[] }): string {
return result.content.filter(block => block.type === 'text').map(block => block.text).join('')
}
@@ -177,8 +208,10 @@ describe('dsh-tool-subagent-control', () => {
await ctx.plugin(SubagentService)
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 === 'interrupt_agent')).toBe(true)
await fiber.dispose()
expect(ctx.tools.schemas().some(schema => schema.name === 'send_message')).toBe(false)
expect(ctx.tools.schemas().some(schema => schema.name === 'interrupt_agent')).toBe(false)
})
it('has the namespace-plugin export shape (no stray default)', () => {
@@ -188,3 +221,168 @@ describe('dsh-tool-subagent-control', () => {
expect(typeof tool.apply).toBe('function')
})
})
describe('dsh-tool-subagent-control interrupt_agent', () => {
it('registers interrupt_agent with the single agent_id parameter and current-turn wording', async () => {
const { ctx } = await setup([])
const schemas = ctx.tools.schemas().filter(schema => schema.name === 'interrupt_agent')
expect(schemas).toHaveLength(1)
const props = (schemas[0]!.parameters as { properties?: Record<string, unknown> }).properties ?? {}
expect(Object.keys(props)).toEqual(['agent_id'])
expect(schemas[0]!.description).toContain('current turn')
expect(schemas[0]!.description).toContain('send_message')
})
it('interrupts a running direct child with the parent cause, parking its queue', async () => {
const releaseFirst = Promise.withResolvers<undefined>()
const adapter = new GatedAdapter([
{ chunks: textResponse('held'), gate: releaseFirst.promise },
{ chunks: textResponse('parked answer') },
{ chunks: textResponse('waking answer') },
])
const { ctx, parent } = await setupWith(adapter)
const started = await ctx.subagents.startContinuable({
provider: 'spawn',
label: 'long work',
request: { prompt: [{ type: 'text', text: 'long work' }], parent },
signal: testToolSignal,
})
await vi.waitFor(() => { expect(adapter.requests).toHaveLength(1) })
const child = ctx.agents.get(started.childId)!
const queued = await callTool(ctx, 'send_message', {
subagent_id: started.childId,
message: 'parked follow-up',
}, parent)
expect(queued.isError).toBe(false)
const cancelSpy = vi.spyOn(child, 'cancel')
const result = await callTool(ctx, 'interrupt_agent', { agent_id: started.childId }, parent)
expect(result.isError).toBe(false)
expect(text(result)).toBe(`interrupt requested for agent ${started.childId}`)
expect(cancelSpy).toHaveBeenCalledExactlyOnceWith({ kind: 'parent' }, { keepInbox: true })
releaseFirst.resolve(undefined)
await child.whenIdle()
// Parked, not resumed: the queued follow-up waits for a waking send.
expect(adapter.requests).toHaveLength(1)
expect(child.inbox.nextTurn).toHaveLength(1)
const waking = await callTool(ctx, 'send_message', {
subagent_id: started.childId,
message: 'wake up',
}, parent)
expect(waking.isError).toBe(false)
await waitNoActivation(ctx, 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] : [])
: [])
expect(prompts).toEqual(['long work', 'parked follow-up', 'wake up'])
})
it('lets a deep live ancestor interrupt a descendant it did not directly create', async () => {
const releaseChild = Promise.withResolvers<undefined>()
const releaseGrandchild = Promise.withResolvers<undefined>()
const adapter = new GatedAdapter([
{ chunks: textResponse('child'), gate: releaseChild.promise },
{ chunks: textResponse('grandchild'), gate: releaseGrandchild.promise },
])
const { ctx, parent } = await setupWith(adapter)
const started = await ctx.subagents.startContinuable({
provider: 'spawn',
label: 'child',
request: { prompt: [{ type: 'text', text: 'child work' }], parent },
signal: testToolSignal,
})
await vi.waitFor(() => { expect(adapter.requests).toHaveLength(1) })
const child = ctx.agents.get(started.childId)!
const grandchild = await ctx.subagents.startContinuable({
provider: 'spawn',
label: 'grandchild',
request: { prompt: [{ type: 'text', text: 'grandchild work' }], parent: child },
signal: testToolSignal,
})
await vi.waitFor(() => { expect(adapter.requests).toHaveLength(2) })
const grandchildAgent = ctx.agents.get(grandchild.childId)!
const cancelSpy = vi.spyOn(grandchildAgent, 'cancel')
const result = await callTool(ctx, 'interrupt_agent', { agent_id: grandchild.childId }, parent)
expect(result.isError).toBe(false)
expect(cancelSpy).toHaveBeenCalledExactlyOnceWith({ kind: 'parent' }, { keepInbox: true })
releaseChild.resolve(undefined)
releaseGrandchild.resolve(undefined)
await waitNoActivation(ctx, grandchild.childId)
await waitNoActivation(ctx, started.childId)
})
it('rejects self, sibling, and unrelated callers without touching the target', async () => {
const releaseA = Promise.withResolvers<undefined>()
const releaseB = Promise.withResolvers<undefined>()
const adapter = new GatedAdapter([
{ chunks: textResponse('a'), gate: releaseA.promise },
{ chunks: textResponse('b'), gate: releaseB.promise },
])
const { ctx, parent } = await setupWith(adapter)
const target = await ctx.subagents.startContinuable({
provider: 'spawn',
label: 'target',
request: { prompt: [{ type: 'text', text: 'a' }], parent },
signal: testToolSignal,
})
await vi.waitFor(() => { expect(adapter.requests).toHaveLength(1) })
const sibling = await ctx.subagents.startContinuable({
provider: 'spawn',
label: 'sibling',
request: { prompt: [{ type: 'text', text: 'b' }], parent },
signal: testToolSignal,
})
await vi.waitFor(() => { expect(adapter.requests).toHaveLength(2) })
const targetAgent = ctx.agents.get(target.childId)!
const siblingAgent = ctx.agents.get(sibling.childId)!
const stranger = ctx.agentLoop.create(SessionId('stranger'), { provider: 'mock', model: 'mock' })
const cancelSpy = vi.spyOn(targetAgent, 'cancel')
const self = await callTool(ctx, 'interrupt_agent', { agent_id: target.childId }, targetAgent)
expect(self.isError).toBe(true)
expect(text(self)).toContain('cannot interrupt itself')
const fromSibling = await callTool(ctx, 'interrupt_agent', { agent_id: target.childId }, siblingAgent)
expect(fromSibling.isError).toBe(true)
expect(text(fromSibling)).toContain('not a live descendant')
const fromStranger = await callTool(ctx, 'interrupt_agent', { agent_id: target.childId }, stranger)
expect(fromStranger.isError).toBe(true)
expect(text(fromStranger)).toContain('not a live descendant')
expect(cancelSpy).not.toHaveBeenCalled()
releaseA.resolve(undefined)
releaseB.resolve(undefined)
await waitNoActivation(ctx, target.childId)
await waitNoActivation(ctx, sibling.childId)
})
it('accepts an absent target as a no-op without cold-resuming it', async () => {
const { ctx, parent } = await setup([textResponse('done')])
const started = await ctx.subagents.startContinuable({
provider: 'spawn',
label: 'settled child',
request: { prompt: [{ type: 'text', text: 'child work' }], parent },
signal: testToolSignal,
})
await waitNoActivation(ctx, started.childId)
const settled = await callTool(ctx, 'interrupt_agent', { agent_id: started.childId }, parent)
expect(settled.isError).toBe(false)
expect(text(settled)).toBe(`interrupt requested for agent ${started.childId}`)
const unknown = await callTool(ctx, 'interrupt_agent', { agent_id: 'no-such-agent' }, parent)
expect(unknown.isError).toBe(false)
// No cold resume: the settled target never rematerialized.
expect(ctx.agents.get(started.childId)).toBeUndefined()
})
it('fails loud when invoked without a calling agent', async () => {
const { ctx } = await setup([])
const result = await callTool(ctx, 'interrupt_agent', { agent_id: 'x' })
expect(result.isError).toBe(true)
expect(text(result)).toContain('requires a calling agent')
})
})