fix(subagent): keep the closing message when another tool ran beside the relay

- subagent: a send_message that shares its step with another tool call no longer omits the closing message
- tests: read settled job results through job_output and give job-owning test agents a mutable inbox
This commit is contained in:
turtle1999 committed 2026-10-12 21:03:04 +08:00
1 parent 70eab5706e
commit 3241fa9923
6 files changed
+81 -14

No files matched your search

@@ -1,5 +1,5 @@
import { describe, expect, it } from 'vitest'
import { provideWorkingDirectoryFixture, unsupportedInbox } from '@deepseek-ai/dsh-agent-loop-testkit'
import { createInboxStub, provideWorkingDirectoryFixture } from '@deepseek-ai/dsh-agent-loop-testkit'
import { Context } from '@deepseek-ai/cordis'
import { ToolCallId } from '@deepseek-ai/dsh-llm'
import { Session, SessionId } from '@deepseek-ai/dsh-session'
@@ -20,7 +20,7 @@ async function fakeAgent(ctx: Context, rawId: string): Promise<Agent> {
const id = SessionId(rawId)
const session = Session.create(id)
const agent: Agent = {
id, options: {}, session, inbox: unsupportedInbox(),
id, options: {}, session, inbox: createInboxStub(),
status: 'idle',
ctx: scope.ctx,
send: () => {},
@@ -82,10 +82,10 @@ async function fakeAgent(ctx: Context, sessionId: string, delivery: FakeDelivery
ctx: scopeFiber.ctx,
inject: delivery.inject ?? (() => {}),
followup: delivery.followup ?? (() => {}),
inbox: { remove: delivery.remove ?? (() => true) },
status: delivery.status ?? 'running',
session: { id, header: { version: 0, id, createdAt: 0 } },
} as unknown as Agent
Object.assign(agent, { inbox: { remove: delivery.remove ?? (() => true) } })
agentRegistryDisposers.set(agent, await ctx.agents.register(agent))
agentScopeFibers.set(agent, scopeFiber)
return agent
+4 -1
View File
@@ -73,11 +73,14 @@ interface ActivationBase {
announced: boolean
/**
* Whether this child's last action before its closing text was a message to
* its direct parent: no input was claimed and no tool call started since.
* its direct parent: the only tool call of its step, with no input claimed
* and no tool call started since.
* Its settlement notice then omits the closing message, which restates that
* message.
*/
relayed: boolean
/** Tool calls started in the child's current step, which decides {@link relayed}. */
stepToolCalls: number
}
/** Resources retained only for the selected execution kind. */
+10 -3
View File
@@ -935,6 +935,7 @@ export class SubagentManager {
observer,
announced: false,
relayed: false,
stepToolCalls: 0,
}
void activation.result.promise.catch(() => undefined)
this.resident.set(childId, activation)
@@ -949,10 +950,16 @@ export class SubagentManager {
activation.handle.agent.ctx.on('agent/inbox/inserted', wakeSettlement)
activation.handle.agent.ctx.on('agent/inbox/claimed', wakeSettlement)
activation.handle.agent.ctx.on('agent/inbox/discarded', wakeSettlement)
// Input or a tool call after the child's last message to its parent
// means the closing text may report work that message did not.
// Input, a later tool call, or another call in the same step as the
// child's message to its parent means the closing text may report work
// that message did not.
activation.handle.agent.ctx.on('agent/inbox/claimed', () => { activation.relayed = false })
activation.handle.agent.ctx.on('agent/pre-step', (_payload, next) => {
activation.stepToolCalls = 0
return next()
})
activation.handle.agent.ctx.on('tools/pre-execute', (_exec, next) => {
activation.stepToolCalls += 1
activation.relayed = false
return next()
})
@@ -1307,7 +1314,7 @@ export class SubagentManager {
}
const message = createAgentMessage(sender, content)
this.sendAgentMessage(parent, message)
activation.relayed = true
activation.relayed = activation.stepToolCalls <= 1
return message.id
}
@@ -3060,6 +3060,49 @@ describe('continuable settlement delivery', () => {
)
})
it('reports the closing message when another tool ran in the same step as the message', async () => {
let childTurns = 0
const respond = (options: GenerateOptions): StreamChunk[] => {
const history = JSON.stringify(options.messages)
if (!history.includes('Your parent agent id is')) return textResponse('parent ack')
childTurns += 1
if (childTurns > 1) return textResponse('the answer')
const parentId = /Your parent agent id is \\"([^\\]+)\\"/.exec(history)![1]!
const calls = [
{ id: ToolCallId('n1'), name: 'noop', arguments: '{}' },
{ id: ToolCallId('s1'), name: 'send_message', arguments: JSON.stringify({ agent_id: parentId, message: 'interim' }) },
]
return [
...calls.flatMap((call, index): StreamChunk[] => [
{ type: 'block-start', index, blockType: 'tool-call' },
{ type: 'tool-call-delta', index, id: call.id, name: call.name, argumentsDelta: call.arguments },
{ type: 'block-end', index, block: { type: 'tool-call', ...call } },
]),
{ type: 'usage', usage: { inputTokens: 10, outputTokens: 5 } },
{ type: 'finish', reason: { kind: 'tool-calls' } },
]
}
const { ctx, parent } = await setup(Array.from({ length: 6 }, () => respond))
await ctx.plugin(SubagentControl)
ctx.tools.register(defineTool({
name: 'noop',
description: 'does nothing',
parameters: {},
output: {
schema: { type: 'object', additionalProperties: false, properties: {} },
render: () => [{ type: 'text', text: 'noop' }],
},
execute: () => Promise.resolve({}),
}))
const started = await ctx.subagents.startActivation({ ...startSpec(parent), delivery: 'parent' })
await waitNoActivation(ctx, started.childId)
await vi.waitFor(() => { expect(settlementNotices(parent)).toHaveLength(1) })
expect(settlementNotices(parent)[0]!.text).toBe(
`Background subagent ${started.childId} finished and will do no further work unless you send it more.\nIts closing message:\nthe answer`,
)
})
it('delivers the terminal reason when the child never had a chance to report', async () => {
const { ctx, parent } = await setup([maxTokensResponse('half an ans'), textResponse('parent ack')])
const started = await ctx.subagents.startActivation({ ...startSpec(parent), delivery: 'parent' })
@@ -6,7 +6,7 @@ import SystemPrompt from '@deepseek-ai/dsh-system-prompt'
import ToolRuntime, { TOOL_ABORTED_BEFORE_DISPATCH } from '@deepseek-ai/dsh-tools'
import type { ToolExecutionResult, ToolExecutionToken } from '@deepseek-ai/dsh-tools'
import type { Agent } from '@deepseek-ai/dsh-agent'
import { unsupportedInbox } from '@deepseek-ai/dsh-agent-loop-testkit'
import { createInboxStub } from '@deepseek-ai/dsh-agent-loop-testkit'
import { WorkflowRunId, WorkflowEngine } from '@deepseek-ai/dsh-workflow'
import type {
WorkflowAgentEndInfo, WorkflowAgentInfo, WorkflowResult, WorkflowRun,
@@ -457,7 +457,7 @@ describe('dsh-tool-workflow', () => {
const engine = ctx.workflowEngine as StubEngine
const session = Session.create(SessionId('caller'))
const parent: Agent = {
id: session.id, options: {}, session, inbox: unsupportedInbox(), status: 'idle', ctx,
id: session.id, options: {}, session, inbox: createInboxStub(), status: 'idle', ctx,
send: () => {}, followup: () => {}, steer: () => {}, inject: () => {}, cancel: () => {},
runMaintenance: task => task(new AbortController().signal), whenIdle: () => Promise.resolve(),
}
@@ -511,9 +511,16 @@ describe('dsh-tool-workflow', () => {
const settled = jobs.get(job.id, parent.id)
expect(settled.detail).toBe('4 agents')
expect(settled.progress).toBeUndefined()
const read = jobs.read(job.id, parent.id)
expect(read.result).toContain('workflow "audit" completed (4 agents)')
expect(read.result).toContain('"findings": 2')
const read = await ctx.tools.execute({
signal: testToolSignal,
callId: ToolCallId('call-settled-read'),
name: 'job_output',
arguments: { job_id: jobId },
agent: parent,
})
const readText = (read.content[0] as { text: string }).text
expect(readText).toContain('workflow "audit" completed (4 agents)')
expect(readText).toContain('"findings": 2')
expect(engine.disposed).toBe(1)
// The durable session record still brackets the background run.
expect(session.snapshotEvents().map(event => event.type)).toEqual([
@@ -676,7 +683,7 @@ describe('dsh-tool-workflow', () => {
await ctx.plugin(toolWorkflow, {})
const session = Session.create(SessionId('caller'))
const parent: Agent = {
id: session.id, options: {}, session, inbox: unsupportedInbox(), status: 'idle', ctx,
id: session.id, options: {}, session, inbox: createInboxStub(), status: 'idle', ctx,
send: () => {}, followup: () => {}, steer: () => {}, inject: () => {}, cancel: () => {},
runMaintenance: task => task(new AbortController().signal), whenIdle: () => Promise.resolve(),
}
@@ -696,7 +703,14 @@ describe('dsh-tool-workflow', () => {
expect(jobs.get(jobId, parent.id).detail).toBe('0 agents')
const ring = jobs.readAt(jobId, 0, parent.id).chunks.map(chunk => chunk.text).join('')
expect(ring).toContain('halfway')
expect(jobs.read(jobId, parent.id).result).toContain('"ok": true')
const read = await ctx.tools.execute({
signal: testToolSignal,
callId: ToolCallId('call-ptc-settled-read'),
name: 'job_output',
arguments: { job_id: jobId },
agent: parent,
})
expect((read.content[0] as { text: string }).text).toContain('"ok": true')
}, 15_000)
})
})