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

fix(subagent): preserve follow-up provenance

Browse Source
This commit is contained in:
Dudu-0223
2026-07-24 13:18:35 +08:00
committed by imccyu
parent 1ab3cbf673
commit 189502e4ac
20 changed files with 202 additions and 87 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
packages/subagent/subagent-control/README.md
View File

@@ -6,7 +6,7 @@ The continuable-subagent control service (`ctx.subagentControl`): the one orches
A continuable background subagent is a durable child session with a series of Task-backed activations. `startContinuable()` allocates the stable child session id before Task creation, snapshots the descriptor inputs (a non-JSON input throws with no Task), and registers the initial activation's Task; the provider publishes exactly that child id and appends the versioned `subagent/descriptor` event inside the child's first turn. Every activation — initial or resumed — creates a fresh Task whose settlement awaits the provider's durability-confirmed child result, disposes the run, and only then records the `TaskOutcome`: a terminal Task leaves the durable child session but no live child Agent. A provider rejection with `DURABILITY_FAILED` settles the Task as `failed` and copies the error message into `detail`, so `task_output` reports the failed checkpoint and resumability risk without exposing unconfirmed output.
`sendMessage(parent, childId, message)` owns steer-or-resume routing. A running activation receives live delivery through the run's strict `steer` capability and returns the existing Task id (`steered`); an absent activation starts a fresh Task that loads the persisted child, authorizes the recorded `parentSession` as the direct parent, folds the descriptor, and dispatches `SubagentService.resume()` (`started`). Failure throws and means the message was not delivered: losing a strict-steering race with Task settlement never falls through to cold resume within the same call, and a live registry Agent outside the activation association is an ownership conflict rather than an adoption target.
`sendMessage(parent, childId, message, source)` owns steer-or-resume routing and requires the caller's `MessageSource`. A running activation preserves it through the run's strict `steer` capability and returns the existing Task id (`steered`); an absent activation starts a fresh Task that loads the persisted child, authorizes the recorded `parentSession` as the direct parent, folds the descriptor, and dispatches `SubagentService.resume()` with the same source (`started`). Either route projects the content to the model as a user-role message while retaining its source in the child log. Failure throws and means the message was not delivered: losing a strict-steering race with Task settlement never falls through to cold resume within the same call, and a live registry Agent outside the activation association is an ownership conflict rather than an adoption target.
Cancellation targets the whole activation. `task_kill` or owner disposal aborts the Task-owned signal; before publication the provider rejects only after its creation transaction rolled back to quiescence, afterwards the signal cancels the published run, and settlement records `killed` only once the activation is quiescent. Human input shares this path: an adapter submits child input through `sendMessage()` under the loaded parent, so parent and human messages that joined one turn share its result and cancellation outcome, and `TaskService.start()`'s control-surface requirement applies (load `@deepseek-ai/dsh-tool-tasks` or attach a surface).

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

@@ -20,7 +20,7 @@ import { randomUUID } from 'node:crypto'
import { Context, Service } from 'cordis'
import type { Agent } from '@deepseek-ai/dsh-agent'
import { HarnessError } from '@deepseek-ai/dsh-llm'
import type { ContentBlock } from '@deepseek-ai/dsh-llm'
import type { ContentBlock, MessageSource } from '@deepseek-ai/dsh-llm'
import { SessionId } from '@deepseek-ai/dsh-session'
import type { SessionPersistence } from '@deepseek-ai/dsh-session-persistence'
import { foldSubagentDescriptor, snapshotSubagentDescriptor } from '@deepseek-ai/dsh-subagent'
@@ -33,6 +33,19 @@ declare module 'cordis' {
}
}
/** Attribution for a model coordinator's follow-up to one of its children. */
export interface CoordinatorMessageSource {
readonly kind: 'coordinator'
/** Session id of the agent whose tool call produced the follow-up. */
readonly senderSessionId: SessionId
}
declare module '@deepseek-ai/dsh-llm' {
interface MessageSourceMap {
coordinator: CoordinatorMessageSource
}
}
/** Typed error for control-service routing, authorization, and delivery failures. */
export class SubagentControlError extends HarnessError {
constructor(message: string, code: string, options?: ErrorOptions) {
@@ -248,16 +261,20 @@ export class SubagentControlService extends Service {
* @param parent - the live parent agent sending the message (model tool or
* human adapter); Task access is authorized by its session id.
* @param childId - the stable child session id.
* @param message - the content to deliver.
* @param message - the user-role content to deliver.
* @param source - caller-supplied attribution retained across either route.
* @returns whether the message `steered` the existing Task or `started` a new one.
*/
sendMessage(parent: Agent, childId: SessionId, message: ContentBlock[]): SendMessageResult {
sendMessage(parent: Agent, childId: SessionId, message: ContentBlock[], source: MessageSource): SendMessageResult {
this.assertOwnership(childId)
const activation = this.activations.get(childId)
if (activation !== undefined) {
return { route: 'steered', taskId: this.steerActivation(activation, parent, childId, message) }
return {
route: 'steered',
taskId: this.steerActivation(activation, parent, childId, message, source),
}
}
return { route: 'started', taskId: this.resumeActivation(parent, childId, message) }
return { route: 'started', taskId: this.resumeActivation(parent, childId, message, source) }
}
/**
@@ -290,6 +307,7 @@ export class SubagentControlService extends Service {
parent: Agent,
childId: SessionId,
message: ContentBlock[],
source: MessageSource,
): TaskId {
const taskId = activation.taskId
/* v8 ignore next 3 -- the install and Task registration share one synchronous frame, so an observed activation carries its Task id. */
@@ -316,7 +334,7 @@ export class SubagentControlService extends Service {
)
}
try {
run.steer(message)
run.steer(message, source)
} catch (error: unknown) {
// Strict steering lost the race with turn settlement. Deliberately no
// cold-resume fallback here: that would attach the message to a turn the
@@ -337,7 +355,12 @@ export class SubagentControlService extends Service {
* activation, with cancellation rechecked after the un-signalled
* persistence await so an early `task_kill` prevents any later child work.
*/
private resumeActivation(parent: Agent, childId: SessionId, message: ContentBlock[]): TaskId {
private resumeActivation(
parent: Agent,
childId: SessionId,
message: ContentBlock[],
source: MessageSource,
): TaskId {
const persistence = this.requirePersistence()
return this.startActivation(childId, resumeLabel(message), parent, async (signal) => {
let loaded: Awaited<ReturnType<typeof persistence.load>>
@@ -374,6 +397,7 @@ export class SubagentControlService extends Service {
return this.ctx.subagents.resume(descriptor.provider, {
sessionId: childId,
prompt: message,
source,
parent,
signal,
descriptor,

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

@@ -107,6 +107,20 @@ function message(text: string) {
return [{ type: 'text' as const, text }]
}
const coordinatorSource = {
kind: 'coordinator',
senderSessionId: SessionId('parent'),
} as const
function sendMessage(
ctx: Context,
parent: Agent,
childId: SessionId,
content: ReturnType<typeof message>,
) {
return ctx.subagentControl.sendMessage(parent, childId, content, { kind: 'user' })
}
describe('SubagentControlService.startContinuable', () => {
it('returns both identities immediately; the Task settles with the child result after disposal', async () => {
const { ctx, parent } = await setup([textResponse('first answer')])
@@ -202,7 +216,7 @@ describe('SubagentControlService.startContinuable', () => {
expect(snapshot.status).toBe('failed')
expect(snapshot.detail).toContain('maxDepth')
// The unmaterialized child id is reported unavailable on later use.
const followUp = ctx.subagentControl.sendMessage(parent, started.childId, message('hello?'))
const followUp = sendMessage(ctx, parent, started.childId, message('hello?'))
expect(followUp.route).toBe('started')
const failed = await waitTerminal(ctx, followUp.taskId, parent)
expect(failed.status).toBe('failed')
@@ -250,14 +264,14 @@ describe('SubagentControlService.sendMessage', () => {
await waitPublishedRun(ctx, started.childId)
expect(descriptor).toEqual({ version: SUBAGENT_DESCRIPTOR_VERSION, provider: 'no-steer' })
expect(() => ctx.subagentControl.sendMessage(parent, started.childId, message('join')))
expect(() => sendMessage(ctx, 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'))
sendMessage(ctx, parent, started.childId, message('after terminal'))
} catch (error: unknown) {
terminalDeliveryError = error
}
@@ -296,7 +310,7 @@ describe('SubagentControlService.sendMessage', () => {
const started = ctx.subagentControl.startContinuable(startSpec(parent, 'mismatched-local'))
await waitPublishedRun(ctx, started.childId)
expect(() => ctx.subagentControl.sendMessage(parent, started.childId, message('join')))
expect(() => sendMessage(ctx, 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)
@@ -324,7 +338,12 @@ describe('SubagentControlService.sendMessage', () => {
}, 5)
})
const delivered = ctx.subagentControl.sendMessage(parent, started.childId, message('also consider Y'))
const delivered = ctx.subagentControl.sendMessage(
parent,
started.childId,
message('also consider Y'),
coordinatorSource,
)
expect(delivered).toEqual({ route: 'steered', taskId: started.taskId })
releaseFirst()
const snapshot = await waitTerminal(ctx, started.taskId, parent)
@@ -334,6 +353,11 @@ describe('SubagentControlService.sendMessage', () => {
// The steered content joined the SAME child turn and drove another step.
const output = ctx.tasks.read(started.taskId, parent)
expect(output.text).toBe('steered turn answer')
const loaded = await ctx.sessionPersistence.load(started.childId)
const steering = loaded.events.find(
(event): event is SessionEvent<'steering/message'> => event.type === 'steering/message',
)
expect(steering?.data.message.source).toEqual(coordinatorSource)
})
it('cold-resumes a settled child into a fresh Task and reports `started`', async () => {
@@ -342,7 +366,12 @@ describe('SubagentControlService.sendMessage', () => {
await waitTerminal(ctx, started.taskId, parent)
expect(ctx.agents.get(started.childId)).toBeUndefined()
const followUp = ctx.subagentControl.sendMessage(parent, started.childId, message('and then?'))
const followUp = ctx.subagentControl.sendMessage(
parent,
started.childId,
message('and then?'),
coordinatorSource,
)
expect(followUp.route).toBe('started')
expect(followUp.taskId).not.toBe(started.taskId)
const snapshot = await waitTerminal(ctx, followUp.taskId, parent)
@@ -356,6 +385,8 @@ describe('SubagentControlService.sendMessage', () => {
const userMessages = loaded.events.filter((event): event is SessionEvent<'user/message'> => event.type === 'user/message')
expect(userMessages.map(event => (event.data.content[0] as { text: string }).text))
.toEqual(['child task', 'and then?'])
expect(userMessages.map(event => event.data.source))
.toEqual([{ kind: 'user' }, coordinatorSource])
})
it('reconstructs the declared composition on cold resume', async () => {
@@ -378,7 +409,7 @@ describe('SubagentControlService.sendMessage', () => {
expect(descriptor?.data.persona).toBe('You are the resumable child.')
expect(descriptor?.data.toolFilter).toEqual({ deny: [] })
const followUp = ctx.subagentControl.sendMessage(parent, started.childId, message('continue'))
const followUp = sendMessage(ctx, parent, started.childId, message('continue'))
const snapshot = await waitTerminal(ctx, followUp.taskId, parent)
expect(snapshot.status).toBe('completed')
// The resumed child's system prompt carried the persona back.
@@ -407,7 +438,7 @@ describe('SubagentControlService.sendMessage', () => {
parent.followup(createUserMessage({ content: message('parent question two'), source: { kind: 'user' } }))
await parent.whenIdle()
const followUp = ctx.subagentControl.sendMessage(parent, started.childId, message('follow up'))
const followUp = sendMessage(ctx, parent, started.childId, message('follow up'))
await waitTerminal(ctx, followUp.taskId, parent)
const resumed = await ctx.sessionPersistence.load(started.childId)
// The persisted seed boundary is unchanged and parent turn two is absent.
@@ -423,7 +454,7 @@ describe('SubagentControlService.sendMessage', () => {
const { ctx, parent } = await setup([textResponse('first'), textResponse('second')])
const started = ctx.subagentControl.startContinuable(startSpec(parent))
await waitTerminal(ctx, started.taskId, parent)
const followUp = ctx.subagentControl.sendMessage(parent, started.childId, message('go on'))
const followUp = sendMessage(ctx, parent, started.childId, message('go on'))
const childAgents: Agent[] = []
const stop = ctx.on('agent/created', (agent: Agent) => {
@@ -443,7 +474,7 @@ describe('SubagentControlService.sendMessage', () => {
const started = ctx.subagentControl.startContinuable(startSpec(otherParent))
await waitTerminal(ctx, started.taskId, otherParent)
const attempt = ctx.subagentControl.sendMessage(parent, started.childId, message('mine now'))
const attempt = sendMessage(ctx, parent, started.childId, message('mine now'))
expect(attempt.route).toBe('started')
const snapshot = await waitTerminal(ctx, attempt.taskId, parent)
expect(snapshot.status).toBe('failed')
@@ -462,7 +493,7 @@ describe('SubagentControlService.sendMessage', () => {
await handle.agent.whenIdle()
await handle.dispose()
const attempt = ctx.subagentControl.sendMessage(parent, SessionId('plain-child'), message('continue?'))
const attempt = sendMessage(ctx, parent, SessionId('plain-child'), message('continue?'))
const snapshot = await waitTerminal(ctx, attempt.taskId, parent)
expect(snapshot.status).toBe('failed')
expect(snapshot.detail).toContain(
@@ -472,9 +503,9 @@ describe('SubagentControlService.sendMessage', () => {
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 blank = sendMessage(ctx, parent, SessionId('blank-child'), message(' '))
const longText = 'x'.repeat(100)
const long = ctx.subagentControl.sendMessage(parent, SessionId('long-child'), message(longText))
const long = sendMessage(ctx, 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)}…`)
@@ -492,9 +523,9 @@ describe('SubagentControlService.sendMessage', () => {
meta: { parentSession: parent.id },
agentOptions: { provider: 'mock', model: 'mock' },
})
expect(() => ctx.subagentControl.sendMessage(parent, SessionId('rogue-child'), message('hello')))
expect(() => sendMessage(ctx, parent, SessionId('rogue-child'), message('hello')))
.toThrow(SubagentControlError)
expect(() => ctx.subagentControl.sendMessage(parent, SessionId('rogue-child'), message('hello')))
expect(() => sendMessage(ctx, parent, SessionId('rogue-child'), message('hello')))
.toThrow(/outside control-service ownership.*not delivered/)
await handle.dispose()
})
@@ -535,13 +566,13 @@ describe('SubagentControlService.sendMessage', () => {
// Strict steering finds the settled child, fails loud, and does NOT start
// a cold resume within this call.
expect(() => ctx.subagentControl.sendMessage(parent, started.childId, message('too late?')))
expect(() => sendMessage(ctx, parent, started.childId, message('too late?')))
.toThrow(/not delivered/)
expect(ctx.tasks.list(parent).map(task => task.id)).toEqual([started.taskId])
releaseDispose()
await waitTerminal(ctx, started.taskId, parent)
// AFTER the Task settles, retry legitimately starts the next activation.
const retry = ctx.subagentControl.sendMessage(parent, started.childId, message('retry'))
const retry = sendMessage(ctx, parent, started.childId, message('retry'))
expect(retry.route).toBe('started')
await waitTerminal(ctx, retry.taskId, parent)
})
@@ -550,7 +581,7 @@ describe('SubagentControlService.sendMessage', () => {
const { ctx, parent } = await setup([textResponse('first'), textResponse('second')])
const started = ctx.subagentControl.startContinuable(startSpec(parent))
await waitTerminal(ctx, started.taskId, parent)
const followUp = ctx.subagentControl.sendMessage(parent, started.childId, message('more'))
const followUp = sendMessage(ctx, parent, started.childId, message('more'))
const other = ctx.agentLoop.create(SessionId('intruder'), { provider: 'mock', model: 'mock' })
expect(() => ctx.tasks.get(followUp.taskId, other)).toThrow(/belongs to another session/)
})
@@ -569,7 +600,7 @@ describe('SubagentControlService.sendMessage', () => {
return realLoad(id)
}
const followUp = ctx.subagentControl.sendMessage(parent, started.childId, message('follow up'))
const followUp = sendMessage(ctx, parent, started.childId, message('follow up'))
expect(ctx.tasks.kill(followUp.taskId, parent)).toBe('requested')
releaseLoad()
const snapshot = await waitTerminal(ctx, followUp.taskId, parent)
@@ -591,11 +622,11 @@ describe('SubagentControlService.sendMessage', () => {
return realLoad(id)
}
const first = ctx.subagentControl.sendMessage(parent, started.childId, message('first follow-up'))
const first = sendMessage(ctx, parent, started.childId, message('first follow-up'))
expect(first.route).toBe('started')
// The association is installed synchronously, so the competing caller
// observes the pending activation instead of starting a duplicate resume.
expect(() => ctx.subagentControl.sendMessage(parent, started.childId, message('second follow-up')))
expect(() => sendMessage(ctx, parent, started.childId, message('second follow-up')))
.toThrow(/not delivered/)
releaseLoad()
const snapshot = await waitTerminal(ctx, first.taskId, parent)