feat(subagent): continuable background subagents
Implement the continuable background subagents RFC: a durable child session with a series of Task-backed activations, each disposing its run before the Task settles. - dsh-subagent: rename SubagentRun.sendMessage to strict steer, drop run-level resume, add SubagentProvider.resume dispatch via SubagentService.resume, the continuation start field, and the versioned model-hidden subagent/descriptor session event. - dsh-subagent-inprocess/-spawn/-fork: publish the control-allocated child id, append the descriptor inside the initial turn, implement cold resume from the child's own transcript under the live parent scope, and strict running-only steer. - dsh-subagent-control (new): SubagentControlService owning stable child ids, descriptor snapshot/fold/authorization, Task-backed activation with settle-then-dispose ordering, the process-local active-run association, and steer-or-resume sendMessage routing. - dsh-tool-subagent: background route branches on the provider's resume capability (continuable via the control service; one-shot task for ACP), returning both child and task ids. - dsh-tool-subagent-control (new): the globally named send_message tool rendering steered/started routes. Keyless coverage spans Task ownership and disposal ordering, running delivery, cold follow-up, descriptor rejection and rollback, known-id reconstruction, kill during lookup, admission races, and a new subagent-continuable ACP snapshot scenario.
This commit is contained in:
Dudu-0223
@@ -0,0 +1,539 @@
|
||||
import { afterEach, describe, expect, it } from 'vitest'
|
||||
import { mkdtempSync, rmSync } from 'node:fs'
|
||||
import { tmpdir } from 'node:os'
|
||||
import { join } from 'node:path'
|
||||
import { Context } from 'cordis'
|
||||
import type { Agent } from '@deepseek-ai/dsh-agent'
|
||||
import AgentLoop from '@deepseek-ai/dsh-agent-loop'
|
||||
import { mountAgentLoopTestDependencies } from '@deepseek-ai/dsh-agent-loop-testkit'
|
||||
import { SessionId } from '@deepseek-ai/dsh-session'
|
||||
import type { SessionEvent } from '@deepseek-ai/dsh-session'
|
||||
import JsonlSessionPersistence from '@deepseek-ai/dsh-session-persistence-jsonl'
|
||||
import SubagentService, { SUBAGENT_DESCRIPTOR_VERSION } from '@deepseek-ai/dsh-subagent'
|
||||
import * as SubagentSpawn from '@deepseek-ai/dsh-subagent-spawn'
|
||||
import * as SubagentFork from '@deepseek-ai/dsh-subagent-fork'
|
||||
import { TaskId } from '@deepseek-ai/dsh-tasks'
|
||||
import LocalTaskService from '@deepseek-ai/dsh-tasks-local'
|
||||
import * as ToolTasks from '@deepseek-ai/dsh-tool-tasks'
|
||||
import type { GenerateOptions, StreamChunk } from '@deepseek-ai/dsh-llm'
|
||||
import { createUserMessage, LlmAdapter } from '@deepseek-ai/dsh-llm'
|
||||
import { MockAdapter, textResponse } from '../../../core/agent-loop/tests/mock-adapter.ts'
|
||||
import SubagentControlService, { runOutcome, settleRun, SubagentControlError } from '../src/index.ts'
|
||||
|
||||
type Script = ConstructorParameters<typeof MockAdapter>[0]
|
||||
|
||||
/** One scripted response that may wait on a caller-released gate before streaming. */
|
||||
interface GatedEntry {
|
||||
chunks: StreamChunk[]
|
||||
gate?: Promise<void>
|
||||
}
|
||||
|
||||
/** Adapter whose entries can hold a model call open until the test releases it. */
|
||||
class GatedAdapter extends LlmAdapter {
|
||||
constructor(private script: GatedEntry[]) {
|
||||
super()
|
||||
}
|
||||
|
||||
async * stream(options: GenerateOptions): AsyncIterable<StreamChunk> {
|
||||
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 roots: string[] = []
|
||||
afterEach(() => {
|
||||
for (const root of roots.splice(0)) rmSync(root, { recursive: true, force: true })
|
||||
})
|
||||
|
||||
/** Boot the full continuable stack: loop, persistence, providers, tasks, control. */
|
||||
async function setupWith(adapter: LlmAdapter, options: { persistence?: boolean } = {}) {
|
||||
const ctx = new Context()
|
||||
await mountAgentLoopTestDependencies(ctx)
|
||||
if (options.persistence !== false) {
|
||||
const root = mkdtempSync(join(tmpdir(), 'dsh-subagent-control-'))
|
||||
roots.push(root)
|
||||
await ctx.plugin(JsonlSessionPersistence, { root })
|
||||
}
|
||||
await ctx.plugin(AgentLoop, { agents: [] })
|
||||
await ctx.plugin(SubagentService)
|
||||
await ctx.plugin(SubagentSpawn, { providerName: 'spawn' })
|
||||
await ctx.plugin(SubagentFork, { providerName: 'fork' })
|
||||
await ctx.plugin(LocalTaskService)
|
||||
await ctx.plugin(ToolTasks, {})
|
||||
await ctx.plugin(SubagentControlService)
|
||||
ctx.llm.registerAdapter(['mock'], adapter)
|
||||
const parent = ctx.agentLoop.create(SessionId('parent'), { provider: 'mock', model: 'mock' })
|
||||
return { ctx, parent }
|
||||
}
|
||||
|
||||
async function setup(script: Script, options: { persistence?: boolean } = {}) {
|
||||
const adapter = new MockAdapter(script)
|
||||
const { ctx, parent } = await setupWith(adapter, options)
|
||||
return { ctx, parent, adapter }
|
||||
}
|
||||
|
||||
function startSpec(parent: Agent, provider = 'spawn') {
|
||||
return {
|
||||
provider,
|
||||
label: 'delegated work',
|
||||
request: { prompt: [{ type: 'text' as const, text: 'child task' }], parent },
|
||||
}
|
||||
}
|
||||
|
||||
async function waitTerminal(ctx: Context, taskId: TaskId, parent: Agent) {
|
||||
return ctx.tasks.wait(taskId, 5_000, parent)
|
||||
}
|
||||
|
||||
function message(text: string) {
|
||||
return [{ type: 'text' as const, text }]
|
||||
}
|
||||
|
||||
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')])
|
||||
const started = ctx.subagentControl.startContinuable(startSpec(parent))
|
||||
expect(started.childId).toMatch(/[0-9a-f-]{36}/)
|
||||
expect(started.taskId).toBe('subagent-1')
|
||||
|
||||
const snapshot = await waitTerminal(ctx, started.taskId, parent)
|
||||
expect(snapshot.status).toBe('completed')
|
||||
expect(ctx.tasks.read(started.taskId, parent).text).toBe('first answer')
|
||||
// Disposal ordering: the terminal Task leaves no live child Agent.
|
||||
expect(ctx.agents.get(started.childId)).toBeUndefined()
|
||||
})
|
||||
|
||||
it('publishes the control-allocated child id and appends the turn-enclosed descriptor', async () => {
|
||||
const { ctx, parent } = await setup([textResponse('answer')])
|
||||
const seen: SessionEvent[] = []
|
||||
ctx.on('session/event', (session, event) => {
|
||||
if (session.id !== SessionId('parent')) seen.push(event)
|
||||
})
|
||||
const started = ctx.subagentControl.startContinuable(startSpec(parent))
|
||||
await waitTerminal(ctx, started.taskId, parent)
|
||||
|
||||
const descriptorIndex = seen.findIndex(event => event.type === 'subagent/descriptor')
|
||||
const turnStartIndex = seen.findIndex(event => event.type === 'turn/start')
|
||||
const firstAssistant = seen.findIndex(event => event.type === 'assistant/message')
|
||||
expect(descriptorIndex).toBeGreaterThan(turnStartIndex)
|
||||
expect(descriptorIndex).toBeLessThan(firstAssistant)
|
||||
const descriptor = seen[descriptorIndex] as SessionEvent<'subagent/descriptor'>
|
||||
expect(descriptor.data).toEqual({
|
||||
version: SUBAGENT_DESCRIPTOR_VERSION,
|
||||
provider: 'spawn',
|
||||
agentProvider: 'mock',
|
||||
agentModel: 'mock',
|
||||
})
|
||||
// Model-hidden: the descriptor never carries surface metadata.
|
||||
expect('surfaceOp' in descriptor).toBe(false)
|
||||
|
||||
// The durable log kept the exact control-allocated id.
|
||||
const loaded = await ctx.sessionPersistence.load(started.childId)
|
||||
expect(loaded.meta.id).toBe(started.childId)
|
||||
expect(loaded.meta.parentSession).toBe(SessionId('parent'))
|
||||
expect(loaded.events.some(event => event.type === 'subagent/descriptor')).toBe(true)
|
||||
})
|
||||
|
||||
it('rejects synchronously with no Task when persistence is not configured', async () => {
|
||||
const { ctx, parent } = await setup([textResponse('unused')], { persistence: false })
|
||||
expect(() => ctx.subagentControl.startContinuable(startSpec(parent)))
|
||||
.toThrow(/require session persistence/)
|
||||
expect(ctx.tasks.list(parent)).toEqual([])
|
||||
})
|
||||
|
||||
it('rejects a non-JSON descriptor input synchronously with no Task', async () => {
|
||||
const { ctx, parent } = await setup([textResponse('unused')])
|
||||
const spec = startSpec(parent)
|
||||
expect(() => ctx.subagentControl.startContinuable({
|
||||
...spec,
|
||||
// A symbol survives the static ToolRestriction type only through this
|
||||
// cast — exactly the durable-boundary input the snapshot rejects.
|
||||
request: { ...spec.request, toolFilter: { deny: [Symbol('boom') as unknown as string] } },
|
||||
})).toThrow(/not losslessly JSON-serializable/)
|
||||
expect(ctx.tasks.list(parent)).toEqual([])
|
||||
})
|
||||
|
||||
it('settles the Task as failed when provider startup fails after the ids were returned', async () => {
|
||||
const { ctx, parent } = await setup([textResponse('unused')])
|
||||
const spec = {
|
||||
provider: 'spawn',
|
||||
label: 'broken delegation',
|
||||
request: {
|
||||
prompt: [{ type: 'text' as const, text: 'child task' }],
|
||||
parent,
|
||||
// The spawn provider enforces depth: parent depth 0 → child depth 1 > 0.
|
||||
maxDepth: 0,
|
||||
},
|
||||
}
|
||||
const started = ctx.subagentControl.startContinuable(spec)
|
||||
const snapshot = await waitTerminal(ctx, started.taskId, parent)
|
||||
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?'))
|
||||
expect(followUp.route).toBe('started')
|
||||
const failed = await waitTerminal(ctx, followUp.taskId, parent)
|
||||
expect(failed.status).toBe('failed')
|
||||
expect(failed.detail).toContain('unavailable')
|
||||
})
|
||||
|
||||
it('task_kill during the run aborts, disposes, and settles killed after quiescence', async () => {
|
||||
const { ctx, parent } = await setup(['hang'])
|
||||
const started = ctx.subagentControl.startContinuable(startSpec(parent))
|
||||
// Let the child publish and begin its turn.
|
||||
await new Promise(resolve => setTimeout(resolve, 30))
|
||||
expect(ctx.agents.get(started.childId)).toBeDefined()
|
||||
expect(ctx.tasks.kill(started.taskId, parent, 'no longer needed')).toBe('requested')
|
||||
const snapshot = await waitTerminal(ctx, started.taskId, parent)
|
||||
expect(snapshot.status).toBe('killed')
|
||||
expect(ctx.agents.get(started.childId)).toBeUndefined()
|
||||
})
|
||||
})
|
||||
|
||||
describe('SubagentControlService.sendMessage', () => {
|
||||
it('steers a running activation into the existing Task without creating a second Task', async () => {
|
||||
// Hold the child's first model call open so the child is observably
|
||||
// running when the message arrives; the steered content then drives a
|
||||
// second step in the SAME turn.
|
||||
let releaseFirst!: () => void
|
||||
const gate = new Promise<void>((resolve) => { releaseFirst = resolve })
|
||||
const { ctx, parent } = await setupWith(new GatedAdapter([
|
||||
{ chunks: textResponse('first step answer'), gate },
|
||||
{ chunks: textResponse('steered turn answer') },
|
||||
]))
|
||||
|
||||
const started = ctx.subagentControl.startContinuable(startSpec(parent))
|
||||
// Wait for the child agent to publish and enter running.
|
||||
await new Promise<void>((resolve) => {
|
||||
const timer = setInterval(() => {
|
||||
if (ctx.agents.get(started.childId)?.status === 'running') {
|
||||
clearInterval(timer)
|
||||
resolve()
|
||||
}
|
||||
}, 5)
|
||||
})
|
||||
|
||||
const delivered = ctx.subagentControl.sendMessage(parent, started.childId, message('also consider Y'))
|
||||
expect(delivered).toEqual({ route: 'steered', taskId: started.taskId })
|
||||
releaseFirst()
|
||||
const snapshot = await waitTerminal(ctx, started.taskId, parent)
|
||||
expect(snapshot.status).toBe('completed')
|
||||
// Exactly one Task exists: steering created none.
|
||||
expect(ctx.tasks.list(parent).map(task => task.id)).toEqual([started.taskId])
|
||||
// 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')
|
||||
})
|
||||
|
||||
it('cold-resumes a settled child into a fresh Task and reports `started`', async () => {
|
||||
const { ctx, parent } = await setup([textResponse('first answer'), textResponse('second answer')])
|
||||
const started = ctx.subagentControl.startContinuable(startSpec(parent))
|
||||
await waitTerminal(ctx, started.taskId, parent)
|
||||
expect(ctx.agents.get(started.childId)).toBeUndefined()
|
||||
|
||||
const followUp = ctx.subagentControl.sendMessage(parent, started.childId, message('and then?'))
|
||||
expect(followUp.route).toBe('started')
|
||||
expect(followUp.taskId).not.toBe(started.taskId)
|
||||
const snapshot = await waitTerminal(ctx, followUp.taskId, parent)
|
||||
expect(snapshot.status).toBe('completed')
|
||||
expect(ctx.tasks.read(followUp.taskId, parent).text).toBe('second answer')
|
||||
// Fresh activation disposed again: durable child, no live Agent.
|
||||
expect(ctx.agents.get(started.childId)).toBeUndefined()
|
||||
|
||||
// The durable transcript accumulated BOTH activations' turns.
|
||||
const loaded = await ctx.sessionPersistence.load(started.childId)
|
||||
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?'])
|
||||
})
|
||||
|
||||
it('reconstructs the declared composition on cold resume', async () => {
|
||||
const { ctx, parent } = await setup([textResponse('first'), textResponse('second')])
|
||||
const spec = {
|
||||
provider: 'spawn',
|
||||
label: 'scoped delegation',
|
||||
request: {
|
||||
prompt: [{ type: 'text' as const, text: 'child task' }],
|
||||
parent,
|
||||
persona: 'You are the resumable child.',
|
||||
toolFilter: { deny: [] as string[] },
|
||||
},
|
||||
}
|
||||
const started = ctx.subagentControl.startContinuable(spec)
|
||||
await waitTerminal(ctx, started.taskId, parent)
|
||||
|
||||
const loaded = await ctx.sessionPersistence.load(started.childId)
|
||||
const descriptor = loaded.events.find((event): event is SessionEvent<'subagent/descriptor'> => event.type === 'subagent/descriptor')
|
||||
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 snapshot = await waitTerminal(ctx, followUp.taskId, parent)
|
||||
expect(snapshot.status).toBe('completed')
|
||||
// The resumed child's system prompt carried the persona back.
|
||||
const resumed = await ctx.sessionPersistence.load(started.childId)
|
||||
const headers = resumed.events.filter((event): event is SessionEvent<'request/header'> => event.type === 'request/header')
|
||||
expect(headers.at(-1)?.data.header.system).toContain('You are the resumable child.')
|
||||
})
|
||||
|
||||
it('fork children resume from their own transcript without re-forking parent history', async () => {
|
||||
const { ctx, parent } = await setup([
|
||||
textResponse('parent turn one'),
|
||||
textResponse('fork first answer'),
|
||||
textResponse('parent turn two'),
|
||||
textResponse('fork second answer'),
|
||||
])
|
||||
parent.followup(createUserMessage({ content: message('parent question one'), source: { kind: 'user' } }))
|
||||
await parent.whenIdle()
|
||||
|
||||
const started = ctx.subagentControl.startContinuable(startSpec(parent, 'fork'))
|
||||
await waitTerminal(ctx, started.taskId, parent)
|
||||
const firstLoad = await ctx.sessionPersistence.load(started.childId)
|
||||
const seedLength = firstLoad.meta.seedLength ?? 0
|
||||
expect(seedLength).toBeGreaterThan(0)
|
||||
|
||||
// The parent gains NEW history the resume must not re-fork.
|
||||
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'))
|
||||
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.
|
||||
expect(resumed.meta.seedLength).toBe(seedLength)
|
||||
const texts = resumed.events
|
||||
.filter((event): event is SessionEvent<'user/message'> => event.type === 'user/message')
|
||||
.map(event => (event.data.content[0] as { text: string }).text)
|
||||
expect(texts).toContain('parent question one')
|
||||
expect(texts).not.toContain('parent question two')
|
||||
})
|
||||
|
||||
it('a resumed child cannot regain a top-level delegation budget (header floor)', async () => {
|
||||
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 childAgents: Agent[] = []
|
||||
const stop = ctx.on('agent/created', (agent: Agent) => {
|
||||
if (agent.id === started.childId) childAgents.push(agent)
|
||||
})
|
||||
await waitTerminal(ctx, followUp.taskId, parent)
|
||||
stop()
|
||||
// The resumed runtime options carry no depth, so the header keeps the floor.
|
||||
const resumedChild = childAgents.at(-1)
|
||||
expect(resumedChild).toBeDefined()
|
||||
expect(resumedChild!.session.header.delegationDepth).toBe(1)
|
||||
})
|
||||
|
||||
it('rejects a foreign child id: the started Task fails with UNAUTHORIZED and delivers nothing', async () => {
|
||||
const { ctx, parent } = await setup([textResponse('other parent answer'), textResponse('unused')])
|
||||
const otherParent = ctx.agentLoop.create(SessionId('other-parent'), { provider: 'mock', model: 'mock' })
|
||||
const started = ctx.subagentControl.startContinuable(startSpec(otherParent))
|
||||
await waitTerminal(ctx, started.taskId, otherParent)
|
||||
|
||||
const attempt = ctx.subagentControl.sendMessage(parent, started.childId, message('mine now'))
|
||||
expect(attempt.route).toBe('started')
|
||||
const snapshot = await waitTerminal(ctx, attempt.taskId, parent)
|
||||
expect(snapshot.status).toBe('failed')
|
||||
expect(snapshot.detail).toContain('another parent session')
|
||||
})
|
||||
|
||||
it('rejects a persisted child with no descriptor as not resumable', async () => {
|
||||
const { ctx, parent } = await setup([textResponse('plain child')])
|
||||
// A plain (non-continuable) child session persisted under this parent.
|
||||
const handle = await ctx.agents.create({
|
||||
sessionId: SessionId('plain-child'),
|
||||
meta: { parentSession: parent.id, delegationDepth: 1 },
|
||||
agentOptions: { provider: 'mock', model: 'mock' },
|
||||
})
|
||||
handle.agent.followup(createUserMessage({ content: message('do something'), source: { kind: 'user' } }))
|
||||
await handle.agent.whenIdle()
|
||||
await handle.dispose()
|
||||
|
||||
const attempt = ctx.subagentControl.sendMessage(parent, SessionId('plain-child'), message('continue?'))
|
||||
const snapshot = await waitTerminal(ctx, attempt.taskId, parent)
|
||||
expect(snapshot.status).toBe('failed')
|
||||
expect(snapshot.detail).toContain('continuation descriptor')
|
||||
})
|
||||
|
||||
it('rejects delivery to a live agent outside control-service ownership', async () => {
|
||||
const { ctx, parent } = await setup([textResponse('unused')])
|
||||
// A live child created around the control service.
|
||||
const handle = await ctx.agents.create({
|
||||
sessionId: SessionId('rogue-child'),
|
||||
meta: { parentSession: parent.id },
|
||||
agentOptions: { provider: 'mock', model: 'mock' },
|
||||
})
|
||||
expect(() => ctx.subagentControl.sendMessage(parent, SessionId('rogue-child'), message('hello')))
|
||||
.toThrow(SubagentControlError)
|
||||
expect(() => ctx.subagentControl.sendMessage(parent, SessionId('rogue-child'), message('hello')))
|
||||
.toThrow(/outside control-service ownership.*not delivered/)
|
||||
await handle.dispose()
|
||||
})
|
||||
|
||||
it('does not fall through to cold resume when strict steering loses the settlement race', async () => {
|
||||
// Deterministic race: hold run disposal open so the association still
|
||||
// names a run whose child turn has already ended.
|
||||
const { ctx, parent } = await setup([textResponse('quick answer'), textResponse('unused')])
|
||||
let releaseDispose!: () => void
|
||||
const disposeGate = new Promise<void>((resolve) => { releaseDispose = resolve })
|
||||
const realStart = ctx.subagents.start.bind(ctx.subagents)
|
||||
ctx.subagents.start = async (name, request) => {
|
||||
const run = await realStart(name, request)
|
||||
const realDispose = run.dispose.bind(run)
|
||||
return {
|
||||
...run,
|
||||
...run.steer !== undefined ? { steer: run.steer.bind(run) } : {},
|
||||
dispose: async () => {
|
||||
await disposeGate
|
||||
return realDispose()
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
const started = ctx.subagentControl.startContinuable(startSpec(parent))
|
||||
// Wait for the child to finish its turn while the run remains undisposed
|
||||
// and the association therefore still holds.
|
||||
await new Promise<void>((resolve) => {
|
||||
const timer = setInterval(() => {
|
||||
const child = ctx.agents.get(started.childId)
|
||||
if (child !== undefined && child.status === 'idle'
|
||||
&& child.session.events.some(event => event.type === 'turn/end')) {
|
||||
clearInterval(timer)
|
||||
resolve()
|
||||
}
|
||||
}, 5)
|
||||
})
|
||||
|
||||
// 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?')))
|
||||
.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'))
|
||||
expect(retry.route).toBe('started')
|
||||
await waitTerminal(ctx, retry.taskId, parent)
|
||||
})
|
||||
|
||||
it('each follow-up Task result is fenced to the parent session', async () => {
|
||||
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 other = ctx.agentLoop.create(SessionId('intruder'), { provider: 'mock', model: 'mock' })
|
||||
expect(() => ctx.tasks.get(followUp.taskId, other)).toThrow(/belongs to another session/)
|
||||
})
|
||||
|
||||
it('kills a cold-resume activation during descriptor lookup without starting child work', async () => {
|
||||
const { ctx, parent } = await setup([textResponse('first'), textResponse('never used')])
|
||||
const started = ctx.subagentControl.startContinuable(startSpec(parent))
|
||||
await waitTerminal(ctx, started.taskId, parent)
|
||||
|
||||
// Make the persistence load hang until the kill lands.
|
||||
const realLoad = ctx.sessionPersistence.load.bind(ctx.sessionPersistence)
|
||||
let releaseLoad!: () => void
|
||||
const gate = new Promise<void>((resolve) => { releaseLoad = resolve })
|
||||
ctx.sessionPersistence.load = async (id) => {
|
||||
await gate
|
||||
return realLoad(id)
|
||||
}
|
||||
|
||||
const followUp = ctx.subagentControl.sendMessage(parent, started.childId, message('follow up'))
|
||||
expect(ctx.tasks.kill(followUp.taskId, parent)).toBe('requested')
|
||||
releaseLoad()
|
||||
const snapshot = await waitTerminal(ctx, followUp.taskId, parent)
|
||||
expect(snapshot.status).toBe('killed')
|
||||
// Cancellation during lookup prevented any child publication.
|
||||
expect(ctx.agents.get(started.childId)).toBeUndefined()
|
||||
})
|
||||
|
||||
it('admits one process-local activation per child: a second send during resume load steers or fails, never duplicates', async () => {
|
||||
const { ctx, parent } = await setup([textResponse('first'), textResponse('resumed answer')])
|
||||
const started = ctx.subagentControl.startContinuable(startSpec(parent))
|
||||
await waitTerminal(ctx, started.taskId, parent)
|
||||
|
||||
const realLoad = ctx.sessionPersistence.load.bind(ctx.sessionPersistence)
|
||||
let releaseLoad!: () => void
|
||||
const gate = new Promise<void>((resolve) => { releaseLoad = resolve })
|
||||
ctx.sessionPersistence.load = async (id) => {
|
||||
await gate
|
||||
return realLoad(id)
|
||||
}
|
||||
|
||||
const first = ctx.subagentControl.sendMessage(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')))
|
||||
.toThrow(/not delivered/)
|
||||
releaseLoad()
|
||||
const snapshot = await waitTerminal(ctx, first.taskId, parent)
|
||||
expect(snapshot.status).toBe('completed')
|
||||
// Exactly one follow-up Task was created.
|
||||
expect(ctx.tasks.list(parent).map(task => task.id)).toEqual([started.taskId, first.taskId])
|
||||
})
|
||||
})
|
||||
|
||||
describe('outcome mapping helpers', () => {
|
||||
it('runOutcome maps the stop-reason vocabulary onto task outcomes', () => {
|
||||
const output = [{ type: 'text' as const, text: 'partial' }]
|
||||
expect(runOutcome({ output, stopReason: 'completed' })).toEqual({ status: 'completed', output: 'partial' })
|
||||
expect(runOutcome({ output, stopReason: 'aborted' })).toEqual({ status: 'killed' })
|
||||
expect(runOutcome({ output, stopReason: 'error' })).toEqual({ status: 'failed', detail: 'error' })
|
||||
expect(runOutcome({ output, stopReason: 'max-tokens' })).toEqual({ status: 'failed', detail: 'max-tokens' })
|
||||
expect(runOutcome({ output, stopReason: 'refusal' })).toEqual({ status: 'failed', detail: 'refusal' })
|
||||
// Merge-extensible: an unknown reason is failed-with-detail, never success.
|
||||
expect(runOutcome({ output, stopReason: 'paused' as never })).toEqual({ status: 'failed', detail: 'paused' })
|
||||
})
|
||||
|
||||
it('settleRun disposes the run before reporting, on both result paths', async () => {
|
||||
const order: string[] = []
|
||||
const completed = await settleRun({
|
||||
id: SessionId('child-1'),
|
||||
localAgent: undefined,
|
||||
result: Promise.resolve({ output: [{ type: 'text' as const, text: 'ok' }], stopReason: 'completed' as const }),
|
||||
dispose() { order.push('dispose'); return Promise.resolve() },
|
||||
})
|
||||
order.push('reported')
|
||||
expect(completed).toEqual({ status: 'completed', output: 'ok' })
|
||||
expect(order).toEqual(['dispose', 'reported'])
|
||||
|
||||
// An infrastructure rejection still disposes and reports failed.
|
||||
let disposed = false
|
||||
const failed = await settleRun({
|
||||
id: SessionId('child-2'),
|
||||
localAgent: undefined,
|
||||
result: Promise.reject(new Error('transport gone')),
|
||||
dispose() { disposed = true; return Promise.resolve() },
|
||||
})
|
||||
expect(failed).toEqual({ status: 'failed', detail: 'Error: transport gone' })
|
||||
expect(disposed).toBe(true)
|
||||
|
||||
const disposeFailed = await settleRun({
|
||||
id: SessionId('child-3'),
|
||||
localAgent: undefined,
|
||||
result: Promise.resolve({ output: [], stopReason: 'completed' }),
|
||||
dispose: () => Promise.reject(new Error('reap failed')),
|
||||
})
|
||||
expect(disposeFailed).toEqual({ status: 'failed', detail: 'dispose failed: Error: reap failed' })
|
||||
|
||||
const bothFailed = await settleRun({
|
||||
id: SessionId('child-4'),
|
||||
localAgent: undefined,
|
||||
result: Promise.reject(new Error('result failed')),
|
||||
dispose: () => Promise.reject(new Error('reap failed')),
|
||||
})
|
||||
expect(bothFailed).toEqual({
|
||||
status: 'failed',
|
||||
detail: 'Error: result failed; dispose failed: Error: reap failed',
|
||||
})
|
||||
})
|
||||
})
|
||||
Reference in New Issue
Block a user