feat(subagent): sibling delegations are concurrency-safe
One assistant message carrying several subagent calls now overlaps them under the rolling pool (maxParallelToolCalls) instead of serializing each foreground delegation behind an exclusive barrier; results still commit in model order. tool-subagent declares isConcurrencySafe: () => true for every call form: children work in their own sessions, a run never mutates the parent session, and sibling workspace coordination already belongs to the model for background, continuable, and workflow children. The former serial-execution pin flips to a parallel pin plus a gated genuine-overlap test, and the authored subagent-parallel snapshot pins the assembled transcript (tool/call, tool/call, tool/result, tool/result) with interchangeable twin children, which replay race-free under the existing first-call binding and createdAt harvest ordering. The provider seam now requires concurrent starts and continuable preparations for distinct children to isolate operation-local state, cancellation, settlement, and cleanup; a continuable gate test proves a cancelled preparation leaves no Agent or durable Session while its sibling persists independently. Closes deepseek-harness/deepseek-harness#1489
This commit is contained in:
@@ -33,6 +33,8 @@ export interface Config {
|
||||
inheritsParentContext?: boolean
|
||||
/** Structured value returned when the request asks for one. */
|
||||
structured?: unknown
|
||||
/** Observes each start; the child's result additionally waits for the returned promise. */
|
||||
onStart?: (request: SubagentStartRequest) => Promise<void> | void
|
||||
}
|
||||
|
||||
/** Scripted provider whose result aborts if its signal or disposer wins first. */
|
||||
@@ -68,9 +70,10 @@ class ScriptedSubagentProvider implements SubagentProvider {
|
||||
...wantsStructured ? { structured: this.config.structured ?? { reply } } : {},
|
||||
stopReason: state.cancelled ? 'aborted' : stopReason,
|
||||
})
|
||||
const result = new Promise<SubagentResult>((resolve) => {
|
||||
const gate = Promise.resolve(this.config.onStart?.(request))
|
||||
const result = gate.then(() => new Promise<SubagentResult>((resolve) => {
|
||||
setTimeout(() => { resolve(resultFor()) }, 0)
|
||||
}).finally(() => {
|
||||
})).finally(() => {
|
||||
request.signal.removeEventListener('abort', onAbort)
|
||||
})
|
||||
|
||||
|
||||
@@ -128,20 +128,42 @@ describe('dsh-tool-subagent', () => {
|
||||
expect(foreground.isError).toBe(false)
|
||||
})
|
||||
|
||||
it('keeps foreground and background calls exclusive', async () => {
|
||||
it('classifies foreground and background calls concurrency-safe (sibling delegations overlap)', async () => {
|
||||
const ctx = await setup({ provider: 'mock' })
|
||||
expect(ctx.tools.executionMode({
|
||||
signal: testToolSignal,
|
||||
callId: CallId('subagent-foreground'),
|
||||
name: 'subagent',
|
||||
arguments: { description: 'do work', prompt: 'Reply OK' },
|
||||
})).toEqual({ kind: 'exclusive' })
|
||||
})).toEqual({ kind: 'parallel' })
|
||||
expect(ctx.tools.executionMode({
|
||||
signal: testToolSignal,
|
||||
callId: CallId('subagent-background'),
|
||||
name: 'subagent',
|
||||
arguments: { description: 'do work', prompt: 'Reply OK', run_in_background: true },
|
||||
})).toEqual({ kind: 'exclusive' })
|
||||
})).toEqual({ kind: 'parallel' })
|
||||
})
|
||||
|
||||
it('overlaps sibling foreground delegations dispatched concurrently', async () => {
|
||||
// Two children each block until both have started: hidden serialization
|
||||
// in the tool body, registry pipeline, or provider start path would
|
||||
// deadlock here instead of passing silently.
|
||||
const started: string[] = []
|
||||
let releaseBoth!: () => void
|
||||
const bothStarted = new Promise<void>((resolve) => { releaseBoth = resolve })
|
||||
const ctx = await setup({ provider: 'mock', enableRunInBackground: false }, {
|
||||
onStart: (request: SubagentStartRequest) => {
|
||||
started.push(request.label ?? '(unlabeled)')
|
||||
if (started.length === 2) releaseBoth()
|
||||
return bothStarted
|
||||
},
|
||||
})
|
||||
const results = await Promise.all([
|
||||
callSubagent(ctx, { description: 'first', prompt: 'p1' }),
|
||||
callSubagent(ctx, { description: 'second', prompt: 'p2' }),
|
||||
])
|
||||
expect(started.sort()).toEqual(['first', 'second'])
|
||||
for (const result of results) expect(result.isError).toBe(false)
|
||||
})
|
||||
|
||||
it.each([
|
||||
@@ -957,6 +979,16 @@ describe('dsh-tool-subagent continuable background mode', () => {
|
||||
return { ctx, parent }
|
||||
}
|
||||
|
||||
it('classifies continuable background calls concurrency-safe', async () => {
|
||||
const { ctx } = await continuableSetup()
|
||||
expect(ctx.tools.executionMode({
|
||||
signal: testToolSignal,
|
||||
callId: CallId('subagent-continuable'),
|
||||
name: 'subagent',
|
||||
arguments: { description: 'do work', prompt: 'Reply OK', run_in_background: true },
|
||||
})).toEqual({ kind: 'parallel' })
|
||||
})
|
||||
|
||||
it('starts a continuable child and returns only its durable id, creating no Task', async () => {
|
||||
const { ctx, parent } = await continuableSetup()
|
||||
const schema = ctx.tools.schemas().find(s => s.name === 'subagent')!
|
||||
@@ -986,6 +1018,69 @@ describe('dsh-tool-subagent continuable background mode', () => {
|
||||
expect(loaded.events.some(event => event.type === 'assistant/message')).toBe(true)
|
||||
})
|
||||
|
||||
it('isolates a cancelled continuable preparation from a concurrent sibling', async () => {
|
||||
const { ctx, parent } = await continuableSetup()
|
||||
const bothPreparing = Promise.withResolvers<undefined>()
|
||||
const releasePreparations = Promise.withResolvers<undefined>()
|
||||
const cancelled = new AbortController()
|
||||
let preparationCount = 0
|
||||
let cancelledChildId: ReturnType<typeof SessionId> | undefined
|
||||
let survivingChildId: ReturnType<typeof SessionId> | undefined
|
||||
ctx.subagents.registerProvider({
|
||||
name: 'gated',
|
||||
capabilities: { outputSchema: true, depthLimit: true, toolFilter: true, persona: true },
|
||||
inheritsParentContext: false,
|
||||
start: async () => { throw new Error('continuable policy must not start a one-shot child') },
|
||||
prepareContinuable: async (request) => {
|
||||
preparationCount += 1
|
||||
if (request.signal === cancelled.signal) cancelledChildId = request.sessionId
|
||||
else survivingChildId = request.sessionId
|
||||
if (preparationCount === 2) bothPreparing.resolve(undefined)
|
||||
await releasePreparations.promise
|
||||
return {}
|
||||
},
|
||||
})
|
||||
tool.apply(ctx, {
|
||||
provider: 'gated',
|
||||
toolName: 'subagent_gated',
|
||||
backgroundMode: 'continuable',
|
||||
maxDepth: 3,
|
||||
})
|
||||
|
||||
const execute = (callId: string, description: string, signal: AbortSignal) => ctx.tools.execute({
|
||||
signal,
|
||||
callId: CallId(callId),
|
||||
name: 'subagent_gated',
|
||||
arguments: { description, prompt: 'work', run_in_background: true },
|
||||
agent: parent,
|
||||
})
|
||||
const cancelledResult = execute('continuable-cancelled', 'cancelled sibling', cancelled.signal)
|
||||
const survivingResult = execute('continuable-surviving', 'surviving sibling', testToolSignal)
|
||||
await bothPreparing.promise
|
||||
cancelled.abort()
|
||||
releasePreparations.resolve(undefined)
|
||||
|
||||
const [failed, succeeded] = await Promise.all([cancelledResult, survivingResult])
|
||||
expect(preparationCount).toBe(2)
|
||||
expect(failed.isError).toBe(true)
|
||||
expect(succeeded.isError).toBe(false)
|
||||
expect(cancelledChildId).toBeDefined()
|
||||
expect(survivingChildId).toBeDefined()
|
||||
expect(ctx.agents.get(cancelledChildId!)).toBeUndefined()
|
||||
await expect(ctx.sessionPersistence.load(cancelledChildId!)).rejects.toThrow(/not found/)
|
||||
|
||||
expect(succeeded.isError ? undefined : succeeded.value).toEqual({
|
||||
kind: 'continuable',
|
||||
subagentId: survivingChildId,
|
||||
})
|
||||
await vi.waitFor(() => {
|
||||
expect(ctx.agents.get(survivingChildId!)).toBeUndefined()
|
||||
}, { timeout: 5_000 })
|
||||
const loaded = await ctx.sessionPersistence.load(survivingChildId!)
|
||||
expect(loaded.events.some(event => event.type === 'subagent/descriptor')).toBe(true)
|
||||
expect(loaded.events.some(event => event.type === 'assistant/message')).toBe(true)
|
||||
})
|
||||
|
||||
})
|
||||
|
||||
describe('background preflight failure (no orphaned child, by construction)', () => {
|
||||
|
||||
Reference in New Issue
Block a user