From 1c30548068ad21384c65c758f05f51efef6cb9da Mon Sep 17 00:00:00 2001 From: Dudu-0223 Date: Thu, 30 Jul 2026 13:31:02 +0800 Subject: [PATCH] test(subagent): rewrite continuation spec for activation lifecycle Covers the inbox-acceptance return boundary, pre-acceptance rollback, caller-signal ownership on both sides of acceptance, residency-only routing, single-inbox FIFO ordering across parent and user origins, waiting wakeup with a retained handle, cold resume without the historical parent, ownership registration and release, child-first disposal, send-versus-dispose races, durability failure without an ownership leak, per-epoch lifecycle pairing, and the absence of cancellation, steering, and report surfaces. --- .../subagent/tests/continuation.spec.ts | 1261 +++++++---------- 1 file changed, 474 insertions(+), 787 deletions(-) diff --git a/packages/subagent/subagent/tests/continuation.spec.ts b/packages/subagent/subagent/tests/continuation.spec.ts index 24392d847e..8335d665e1 100644 --- a/packages/subagent/subagent/tests/continuation.spec.ts +++ b/packages/subagent/subagent/tests/continuation.spec.ts @@ -11,17 +11,14 @@ import type { SessionEvent } from '@deepseek-ai/dsh-session' import JsonlSessionPersistence from '@deepseek-ai/dsh-session-persistence-jsonl' 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, HarnessError, LlmAdapter } from '@deepseek-ai/dsh-llm' -import { MockAdapter, textResponse, toolCallResponse } from '../../../core/agent-loop/tests/mock-adapter.ts' +import type { GenerateOptions, MessageId, 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 SubagentService, { - settleRun, SubagentError, SUBAGENT_DESCRIPTOR_VERSION, } from '../src/index.ts' +import type { SubagentAuthority, SubagentRunEndInfo, SubagentRunInfo } from '../src/index.ts' type Script = ConstructorParameters[0] @@ -56,13 +53,14 @@ afterEach(() => { for (const root of roots.splice(0)) rmSync(root, { recursive: true, force: true }) }) -/** Boot the full continuable stack: loop, persistence, providers, tasks, and subagents. */ +/** Boot the full continuable stack: loop, persistence, providers, and subagents. */ async function setupWith(adapter: LlmAdapter, options: { persistence?: boolean } = {}) { const ctx = new Context() await mountAgentLoopTestDependencies(ctx) let disposePersistence: (() => Promise) | undefined + let root: string | undefined if (options.persistence !== false) { - const root = mkdtempSync(join(tmpdir(), 'dsh-subagent-continuation-')) + root = mkdtempSync(join(tmpdir(), 'dsh-subagent-continuation-')) roots.push(root) const persistenceFiber = await ctx.plugin(JsonlSessionPersistence, { root }) disposePersistence = () => persistenceFiber.dispose() @@ -71,83 +69,88 @@ async function setupWith(adapter: LlmAdapter, options: { persistence?: boolean } await ctx.plugin(SubagentService) await ctx.plugin(SubagentSpawn, { providerName: 'spawn' }) await ctx.plugin(SubagentFork, { providerName: 'fork' }) - await ctx.plugin(LocalTaskService) - await ctx.plugin(ToolTasks, {}) ctx.llm.registerAdapter(['mock'], adapter) const parent = ctx.agentLoop.create(SessionId('parent'), { provider: 'mock', model: 'mock' }) - return { ctx, parent, disposePersistence } + return { ctx, parent, disposePersistence, root } } async function setup(script: Script, options: { persistence?: boolean } = {}) { const adapter = new MockAdapter(script) - const { ctx, parent } = await setupWith(adapter, options) - return { ctx, parent, adapter } + const booted = await setupWith(adapter, options) + return { ...booted, adapter } } -function startSpec(parent: Agent, provider = 'spawn') { +const testSignal = new AbortController().signal + +function startSpec(parent: Agent, provider = 'spawn', signal: AbortSignal = testSignal) { return { provider, - label: 'delegated work', request: { prompt: [{ type: 'text' as const, text: 'child task' }], parent }, + signal, } } -async function waitTerminal(ctx: Context, taskId: TaskId, parent: Agent) { - return ctx.tasks.wait(taskId, 5_000, parent) -} - -async function waitPublishedRun(ctx: Context, childId: SessionId): Promise { - const continuations = ctx.subagents as unknown as { - continuations: { activations: Map } - } - await new Promise((resolve) => { - const timer = setInterval(() => { - if (continuations.continuations.activations.get(childId)?.run !== undefined) { - clearInterval(timer) - resolve() - } - }, 5) - }) -} - function message(text: string) { return [{ type: 'text' as const, text }] } -const coordinatorSource = { - kind: 'coordinator', - senderSessionId: SessionId('parent'), -} as const -const testSendSignal = new AbortController().signal +function hasUserText(events: readonly SessionEvent[], text: string): boolean { + return events.some(event => event.type === 'user/message' + && event.data.content.some(block => block.type === 'text' && block.text === text)) +} + +/** Every user-role message text in log order, for FIFO assertions. */ +function userTexts(events: readonly SessionEvent[]): string[] { + return events.flatMap(event => event.type === 'user/message' + ? event.data.content.flatMap(block => block.type === 'text' ? [block.text] : []) + : []) +} function followup( ctx: Context, - parent: Agent, + authority: SubagentAuthority, childId: SessionId, content: ReturnType, - signal: AbortSignal = testSendSignal, + signal: AbortSignal = testSignal, ) { - return ctx.subagents.followup(parent, childId, content, { + return ctx.subagents.followup(authority, childId, content, { source: { kind: 'user' }, signal, }) } -describe('SubagentService.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.subagents.startContinuable(startSpec(parent)) - expect(started.childId).toMatch(/[0-9a-f-]{36}/) - expect(started.taskId).toBe('subagent-1') +/** Wait until a child's Activation is gone, i.e. its handle finished disposal. */ +async function waitNoActivation(ctx: Context, childId: SessionId): Promise { + await vi.waitFor(() => { + expect(ctx.subagents.activationState(childId)).toBeUndefined() + expect(ctx.agents.get(childId)).toBeUndefined() + }, { timeout: 5_000 }) +} - 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() +describe('SubagentService.startContinuable', () => { + it('returns both identities at inbox acceptance, without waiting for the turn or the log', async () => { + const { ctx, parent, adapter } = await setup([textResponse('first answer')]) + const enqueued: { id: MessageId; loggedYet: boolean }[] = [] + ctx.on('agent/inbox/enqueue', (agent, accepted) => { + // Acceptance is the boundary `startContinuable` resolves at, so observe + // the log state exactly there rather than after later microtasks. + enqueued.push({ id: accepted.message.id, loggedYet: hasUserText(agent.session.events, 'child task') }) + }) + + const started = await ctx.subagents.startContinuable(startSpec(parent)) + + expect(started.childId).toMatch(/[0-9a-f-]{36}/) + // The returned id is exactly the accepted inbox message's id, and nothing + // was logged or requested to earn it. + expect(enqueued).toEqual([{ id: started.messageId, loggedYet: false }]) + expect(adapter.requests).toEqual([]) + + await waitNoActivation(ctx, started.childId) + const loaded = await ctx.sessionPersistence.load(started.childId) + expect(hasUserText(loaded.events, 'child task')).toBe(true) }) - it('fails a continuable Task before dispatch when its provider has no resume capability', async () => { + it('rejects without ids when the provider has no prepareContinuable capability', async () => { const { ctx, parent } = await setup([]) const start = vi.fn(async () => { throw new Error('must not dispatch') }) ctx.subagents.registerProvider({ @@ -157,48 +160,30 @@ describe('SubagentService.startContinuable', () => { start, }) - const started = ctx.subagents.startContinuable(startSpec(parent, 'one-shot')) - const snapshot = await waitTerminal(ctx, started.taskId, parent) - - expect(snapshot.status).toBe('failed') - expect(snapshot.detail).toContain('does not support continuable children') + await expect(ctx.subagents.startContinuable(startSpec(parent, 'one-shot'))) + .rejects.toThrow(/does not support continuable children/) expect(start).not.toHaveBeenCalled() + // No child Agent and no session were created. + expect(ctx.agents.list().map(agent => agent.id)).toEqual([SessionId('parent')]) }) - it('fails the Task when persistence detaches before the activation completes', async () => { - const releaseResponse = Promise.withResolvers() - const adapter = new GatedAdapter([ - { chunks: textResponse('unconfirmed answer'), gate: releaseResponse.promise }, - ]) - const { ctx, parent, disposePersistence } = await setupWith(adapter) - const started = ctx.subagents.startContinuable(startSpec(parent)) - await vi.waitFor(() => { expect(adapter.requests).toHaveLength(1) }) - - await disposePersistence!() - releaseResponse.resolve(undefined) - - const snapshot = await waitTerminal(ctx, started.taskId, parent) - expect(snapshot.status).toBe('failed') - expect(snapshot.detail).toContain('durability checkpoint failed') - expect(snapshot.detail).toContain('required durability checkpoint has no registered listener') - expect(ctx.agents.get(started.childId)).toBeUndefined() + it('rejects synchronously when persistence is not configured', async () => { + const { ctx, parent } = await setup([textResponse('unused')], { persistence: false }) + await expect(ctx.subagents.startContinuable(startSpec(parent))) + .rejects.toThrow(/require session persistence/) }) - it('publishes the service-allocated child id and appends the pre-turn descriptor', async () => { + it('publishes the reserved child id and appends the pre-turn 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.subagents.startContinuable(startSpec(parent)) - await waitTerminal(ctx, started.taskId, parent) + const started = await ctx.subagents.startContinuable(startSpec(parent)) + await waitNoActivation(ctx, started.childId) - 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') + const loaded = await ctx.sessionPersistence.load(started.childId) + const descriptorIndex = loaded.events.findIndex(event => event.type === 'subagent/descriptor') + const turnStartIndex = loaded.events.findIndex(event => event.type === 'turn/start') + expect(descriptorIndex).toBeGreaterThanOrEqual(0) expect(descriptorIndex).toBeLessThan(turnStartIndex) - expect(descriptorIndex).toBeLessThan(firstAssistant) - const descriptor = seen[descriptorIndex] as SessionEvent<'subagent/descriptor'> + const descriptor = loaded.events[descriptorIndex] as SessionEvent<'subagent/descriptor'> expect(descriptor.data).toEqual({ version: SUBAGENT_DESCRIPTOR_VERSION, provider: 'spawn', @@ -207,732 +192,434 @@ describe('SubagentService.startContinuable', () => { }) // Model-hidden: the descriptor never carries surface metadata. expect('surfaceOp' in descriptor).toBe(false) - - // The durable log kept the exact service-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.each(['block', 'throw'] as const)( - 'persists the descriptor before initial prompt admission can $0', - async (outcome) => { - const { ctx, parent, adapter } = await setup([]) - ctx.on('agent/prompt-submit', async (subject, _message, _signal, next) => { - if (subject === parent) return next() - if (outcome === 'block') return { kind: 'block', reason: 'blocked by policy' } - throw new Error('prompt admission failed') - }) - - const started = ctx.subagents.startContinuable(startSpec(parent)) - const snapshot = await waitTerminal(ctx, started.taskId, parent) - - expect(snapshot.status).toBe('failed') - expect(adapter.requests).toEqual([]) - const loaded = await ctx.sessionPersistence.load(started.childId) - const descriptorIndexes = loaded.events.flatMap((event, index) => - event.type === 'subagent/descriptor' ? [index] : []) - expect(descriptorIndexes).toHaveLength(1) - expect(loaded.events.some(event => event.type === 'turn/start')).toBe(false) - }, - ) - - it('rejects synchronously with no Task when persistence is not configured', async () => { - const { ctx, parent } = await setup([textResponse('unused')], { persistence: false }) - expect(() => ctx.subagents.startContinuable(startSpec(parent))) - .toThrow(/require session persistence/) - expect(ctx.tasks.list(parent)).toEqual([]) - }) - - it('rolls back the activation when Task preflight throws', async () => { + it('rolls the child back completely when the caller signal aborts before acceptance', async () => { const { ctx, parent } = await setup([textResponse('unused')]) - const realStart = ctx.tasks.start.bind(ctx.tasks) - ctx.tasks.start = () => { throw new Error('task preflight failed') } - try { - expect(() => ctx.subagents.startContinuable(startSpec(parent))) - .toThrow('task preflight failed') - } finally { - ctx.tasks.start = realStart - } - const continuations = ctx.subagents as unknown as { - continuations: { activations: Map } - } - expect(continuations.continuations.activations.size).toBe(0) - }) - - 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.subagents.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.subagents.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 = await followup(ctx, 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.subagents.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() - }) - - it('task_kill during the final durability checkpoint settles killed', async () => { - const { ctx, parent } = await setup([textResponse('driver answer')]) - const checkpointStarted = Promise.withResolvers() - const releaseCheckpoint = Promise.withResolvers() - let flushes = 0 - ctx.on('session/flush', async (session) => { - if (session.header.parentSession === undefined) return - flushes++ - if (flushes !== 2) return - checkpointStarted.resolve(undefined) - await releaseCheckpoint.promise + const controller = new AbortController() + // Abort inside the child's creation window: setup runs before publication. + ctx.on('agent/created', (child) => { + if (child !== parent) controller.abort('caller gave up') }) - const started = ctx.subagents.startContinuable(startSpec(parent)) - await checkpointStarted.promise - expect(ctx.tasks.kill(started.taskId, parent, 'no longer needed')).toBe('requested') - releaseCheckpoint.resolve(undefined) + await expect(ctx.subagents.startContinuable(startSpec(parent, 'spawn', controller.signal))) + .rejects.toThrow() + // No Activation, no live child Agent, and no parent ownership remains. + await vi.waitFor(() => { + expect(ctx.agents.list().map(agent => agent.id)).toEqual([SessionId('parent')]) + }) + }) - const snapshot = await waitTerminal(ctx, started.taskId, parent) - expect(snapshot.status).toBe('killed') - expect(ctx.agents.get(started.childId)).toBeUndefined() + it('rejects a continuable child that would exceed the configured depth cap', async () => { + const { ctx, parent } = await setup([]) + await expect(ctx.subagents.startContinuable({ + ...startSpec(parent), + request: { prompt: message('deep'), parent, maxDepth: 0 }, + })).rejects.toThrow(/exceeds maxDepth 0/) + expect(ctx.agents.list().map(agent => agent.id)).toEqual([SessionId('parent')]) + }) + + it('records the declared persona in the descriptor and reapplies it on cold resume', async () => { + const { ctx, parent } = await setup([textResponse('scoped'), textResponse('resumed')]) + const started = await ctx.subagents.startContinuable({ + ...startSpec(parent), + request: { + prompt: message('scoped work'), + parent, + persona: 'You are scoped.', + }, + }) + await waitNoActivation(ctx, started.childId) + + const loaded = await ctx.sessionPersistence.load(started.childId) + const descriptor = loaded.events.find(event => event.type === 'subagent/descriptor') + expect(descriptor?.data).toMatchObject({ persona: 'You are scoped.' }) + + // Cold resume reconstructs the declared composition from that descriptor. + await followup(ctx, { kind: 'user' }, started.childId, message('resume it')) + await waitNoActivation(ctx, started.childId) + const resumed = await ctx.sessionPersistence.load(started.childId) + expect(hasUserText(resumed.events, 'resume it')).toBe(true) }) }) -describe('SubagentService.followup', () => { - it('fails a cold-resume Task when the provider loses its resume capability', async () => { - const { ctx, parent } = await setup([textResponse('first answer')]) - const started = ctx.subagents.startContinuable(startSpec(parent)) - await waitTerminal(ctx, started.taskId, parent) - - const provider = ctx.subagents.getProvider('spawn')! - Object.defineProperty(provider, 'resume', { value: undefined, configurable: true }) - - const next = await followup(ctx, parent, started.childId, message('continue')) - const snapshot = await waitTerminal(ctx, next.taskId, parent) - - expect(snapshot.status).toBe('failed') - expect(snapshot.detail).toContain('does not support resuming persisted children') - }) - - it('omits undeclared model selectors and rejects a provider without live delivery', async () => { - const { ctx } = await setup([]) - const result = Promise.withResolvers<{ - output: { type: 'text'; text: string }[] - stopReason: 'completed' - }>() - let descriptor: SessionEvent<'subagent/descriptor'>['data'] | undefined - ctx.subagents.registerProvider({ - name: 'no-steer', - capabilities: { outputSchema: false, depthLimit: false, toolFilter: false, persona: false }, - inheritsParentContext: false, - start: async (request) => { - descriptor = request.continuation?.descriptor - return { - id: request.continuation!.sessionId, - localAgent: undefined, - result: result.promise, - async dispose() {}, - } - }, - resume: async () => { throw new Error('not used') }, - }) - const parent = ctx.agentLoop.create(SessionId('bare-parent'), {}) - const started = ctx.subagents.startContinuable(startSpec(parent, 'no-steer')) - await waitPublishedRun(ctx, started.childId) - - expect(descriptor).toEqual({ version: SUBAGENT_DESCRIPTOR_VERSION, provider: 'no-steer' }) - await expect(followup(ctx, parent, started.childId, message('join'))) - .rejects.toThrow(/provider does not accept live delivery/) - - let terminalDeliveryError: unknown - let terminalDelivery: Promise | undefined - ctx.tasks.onTaskDone((snapshot) => { - if (snapshot.id !== started.taskId) return - terminalDelivery = followup(ctx, parent, started.childId, message('after terminal')).then( - () => undefined, - (error: unknown) => { - terminalDeliveryError = error - }, - ) - }) - result.resolve({ output: [{ type: 'text', text: 'done' }], stopReason: 'completed' }) - await waitTerminal(ctx, started.taskId, parent) - await terminalDelivery - expect(String(terminalDeliveryError)).toContain('is completed') - }) - - it('rejects a registry agent different from the associated run agent', async () => { - const { ctx, parent } = await setup([]) - const result = Promise.withResolvers<{ - output: { type: 'text'; text: string }[] - stopReason: 'completed' - }>() - ctx.subagents.registerProvider({ - name: 'mismatched-local', - capabilities: { outputSchema: false, depthLimit: false, toolFilter: false, persona: false }, - inheritsParentContext: false, - start: async (request) => { - const childId = request.continuation!.sessionId - const handle = await ctx.agents.create({ - sessionId: childId, - meta: { parentSession: request.parent.id }, - agentOptions: { provider: 'mock', model: 'mock' }, - }) - return { - id: childId, - localAgent: {} as Agent, - result: result.promise, - dispose: () => handle.dispose(), - } - }, - resume: async () => { throw new Error('not used') }, - }) - const started = ctx.subagents.startContinuable(startSpec(parent, 'mismatched-local')) - await waitPublishedRun(ctx, started.childId) - - await expect(followup(ctx, parent, started.childId, message('join'))) - .rejects.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) - }) - - 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((resolve) => { releaseFirst = resolve }) +describe('SubagentService.followup residency routing', () => { + it('enqueues in the same Activation while it is running, preserving one inbox FIFO', async () => { + const releaseFirst = Promise.withResolvers() const adapter = new GatedAdapter([ - { chunks: textResponse('first step answer'), gate }, - { chunks: textResponse('steered turn answer') }, + { chunks: textResponse('first'), gate: releaseFirst.promise }, + { chunks: textResponse('second') }, + { chunks: textResponse('third') }, ]) const { ctx, parent } = await setupWith(adapter) - - const started = ctx.subagents.startContinuable(startSpec(parent)) - // Wait until the first immutable request has crossed the adapter boundary. - await new Promise((resolve) => { - const timer = setInterval(() => { - if (adapter.requests.length === 1) { - clearInterval(timer) - resolve() - } - }, 5) - }) - - const delivery = ctx.subagents.followup( - parent, - started.childId, - message('also consider Y'), - { source: coordinatorSource, signal: testSendSignal }, - ) - releaseFirst() - const delivered = await delivery - expect(delivered).toEqual({ route: 'steered', taskId: started.taskId }) - 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') - 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('cancels the active Task without enqueueing when live delivery is already aborted', async () => { - const { ctx, parent, adapter } = await setup(['hang']) - const started = ctx.subagents.startContinuable(startSpec(parent)) + const started = await ctx.subagents.startContinuable(startSpec(parent)) await vi.waitFor(() => { expect(adapter.requests).toHaveLength(1) }) + const child = ctx.agents.get(started.childId) + expect(ctx.subagents.activationState(started.childId)).toBe('running') + + // Both origins queue behind the open turn, in call order. + const parentMessage = await followup(ctx, { kind: 'parent', agent: parent }, started.childId, message('from parent')) + const userMessage = await followup(ctx, { kind: 'user' }, started.childId, message('from user')) + expect(parentMessage).not.toBe(userMessage) + // Still the same Activation: no second child Agent was created. + expect(ctx.agents.get(started.childId)).toBe(child) + + releaseFirst.resolve() + await waitNoActivation(ctx, started.childId) + const loaded = await ctx.sessionPersistence.load(started.childId) + expect(userTexts(loaded.events)).toEqual(['child task', 'from parent', 'from user']) + }) + + it('cold-resumes a settled child into a new Activation', async () => { + const { ctx, parent } = await setup([textResponse('first'), textResponse('after resume')]) + const started = await ctx.subagents.startContinuable(startSpec(parent)) + await waitNoActivation(ctx, started.childId) + + const messageId = await followup(ctx, { kind: 'user' }, started.childId, message('continue please')) + expect(messageId).toBeTypeOf('string') + await waitNoActivation(ctx, started.childId) + + const loaded = await ctx.sessionPersistence.load(started.childId) + expect(userTexts(loaded.events)).toEqual(['child task', 'continue please']) + // One descriptor only: cold resume never re-seeds it. + expect(loaded.events.filter(event => event.type === 'subagent/descriptor')).toHaveLength(1) + }) + + it('wakes a waiting Activation instead of cold-resuming it', async () => { + const releaseGrandchild = Promise.withResolvers() + const adapter = new GatedAdapter([ + // The child delegates, then finishes its own turn while the grandchild runs. + { chunks: textResponse('child done') }, + { chunks: textResponse('grandchild'), gate: releaseGrandchild.promise }, + { chunks: textResponse('woken') }, + ]) + const { ctx, parent } = await setupWith(adapter) + const started = await ctx.subagents.startContinuable(startSpec(parent)) + const child = await vi.waitFor(() => { + const found = ctx.agents.get(started.childId) + expect(found).toBeDefined() + return found! + }) + // The child starts its own continuable grandchild, then goes quiescent. + const grandchild = await ctx.subagents.startContinuable(startSpec(child)) + await vi.waitFor(() => { expect(adapter.requests.length).toBeGreaterThanOrEqual(2) }) + await vi.waitFor(() => { + expect(ctx.subagents.activationState(started.childId)).toBe('waiting') + }, { timeout: 5_000 }) + // Waiting retains the handle: the same Agent is still live. + expect(ctx.agents.get(started.childId)).toBe(child) + + await followup(ctx, { kind: 'user' }, started.childId, message('while waiting')) + // Woken back to running on the SAME Activation. + expect(ctx.agents.get(started.childId)).toBe(child) + + releaseGrandchild.resolve() + await waitNoActivation(ctx, grandchild.childId) + await waitNoActivation(ctx, started.childId) + const loaded = await ctx.sessionPersistence.load(started.childId) + expect(userTexts(loaded.events)).toEqual(['child task', 'while waiting']) + }) + + it('rejects a parent that is not the durable direct parent', async () => { + const { ctx, parent } = await setup([textResponse('first')]) + const started = await ctx.subagents.startContinuable(startSpec(parent)) + await waitNoActivation(ctx, started.childId) + const stranger = ctx.agentLoop.create(SessionId('stranger'), { provider: 'mock', model: 'mock' }) + + await expect(followup(ctx, { kind: 'parent', agent: stranger }, started.childId, message('mine now'))) + .rejects.toThrow(/belongs to another parent session/) + }) + + it('lets user authority cold-resume a child without loading its historical parent', async () => { + const { ctx, parent, root } = await setup([textResponse('first')]) + const started = await ctx.subagents.startContinuable(startSpec(parent)) + await waitNoActivation(ctx, started.childId) + await ctx.sessionPersistence.load(started.childId) + + // A fresh runtime over the same store has no parent Agent at all. + const fresh = new Context() + await mountAgentLoopTestDependencies(fresh) + await fresh.plugin(JsonlSessionPersistence, { root: root! }) + await fresh.plugin(AgentLoop, { agents: [] }) + await fresh.plugin(SubagentService) + await fresh.plugin(SubagentSpawn, { providerName: 'spawn' }) + fresh.llm.registerAdapter(['mock'], new MockAdapter([textResponse('resumed cold')])) + expect(fresh.agents.get(SessionId('parent'))).toBeUndefined() + + await followup(fresh, { kind: 'user' }, started.childId, message('user continues')) + await waitNoActivation(fresh, started.childId) + + const loaded = await fresh.sessionPersistence.load(started.childId) + expect(hasUserText(loaded.events, 'user continues')).toBe(true) + // The historical parent was never reconstructed. + expect(fresh.agents.get(SessionId('parent'))).toBeUndefined() + }) + + it('reports an unresumable child whose persisted log has no supported descriptor', async () => { + const { ctx, parent } = await setup([textResponse('one shot')]) + // A ONE-SHOT child persists a log but never seeds a descriptor. + const run = await ctx.subagents.start('spawn', { + prompt: message('one-shot work'), + parent, + signal: testSignal, + }) + await run.result + await ctx.sessions.flush(run.localAgent!.session) + const oneShotId = run.id + await run.dispose() + + await expect(followup(ctx, { kind: 'user' }, oneShotId, message('continue'))) + .rejects.toThrow(/no supported continuation state/) + }) + + it('reports an unknown child id as unavailable', async () => { + const { ctx } = await setup([]) + await expect(followup(ctx, { kind: 'user' }, SessionId('missing'), message('hello'))) + .rejects.toMatchObject({ code: 'NOT_RESUMABLE' }) + }) + + it('cold-resumes after losing a race with final disposal', async () => { + const { ctx, parent } = await setup([textResponse('first'), textResponse('after the race')]) + const started = await ctx.subagents.startContinuable(startSpec(parent)) + const child = await vi.waitFor(() => { + const found = ctx.agents.get(started.childId) + expect(found).toBeDefined() + return found! + }) + // Send exactly while the Activation is settling: one side wins the cutoff, + // and a delivery that loses waits for release and cold-resumes. + await child.whenIdle() + const delivery = followup(ctx, { kind: 'user' }, started.childId, message('raced')) + + await expect(delivery).resolves.toBeTypeOf('string') + await waitNoActivation(ctx, started.childId) + const loaded = await ctx.sessionPersistence.load(started.childId) + expect(hasUserText(loaded.events, 'raced')).toBe(true) + }) +}) + +describe('continuable child ownership', () => { + it('keeps a parent Activation waiting until its child completes disposal', async () => { + const releaseGrandchild = Promise.withResolvers() + const adapter = new GatedAdapter([ + { chunks: textResponse('child done') }, + { chunks: textResponse('grandchild'), gate: releaseGrandchild.promise }, + ]) + const { ctx, parent } = await setupWith(adapter) + const started = await ctx.subagents.startContinuable(startSpec(parent)) + const child = await vi.waitFor(() => { + const found = ctx.agents.get(started.childId) + expect(found).toBeDefined() + return found! + }) + const grandchild = await ctx.subagents.startContinuable(startSpec(child)) + + await vi.waitFor(() => { + expect(ctx.subagents.activationState(started.childId)).toBe('waiting') + }, { timeout: 5_000 }) + // Child-first: the parent handle is retained while the grandchild is live. + expect(ctx.agents.get(started.childId)).toBe(child) + expect(ctx.agents.get(grandchild.childId)).toBeDefined() + + releaseGrandchild.resolve() + await waitNoActivation(ctx, grandchild.childId) + await waitNoActivation(ctx, started.childId) + }) + + it('does not add a top-level parent to the waiting graph', async () => { + const { ctx, parent } = await setup([textResponse('done')]) + const started = await ctx.subagents.startContinuable(startSpec(parent)) + await waitNoActivation(ctx, started.childId) + // The top-level parent has no Activation of its own. + expect(ctx.subagents.activationState(parent.id)).toBeUndefined() + expect(ctx.agents.get(parent.id)).toBe(parent) + }) +}) + +describe('continuable durability and teardown', () => { + it('reports DURABILITY_FAILED without leaking a waiting Activation', async () => { + const releaseResponse = Promise.withResolvers() + const adapter = new GatedAdapter([ + { chunks: textResponse('unconfirmed answer'), gate: releaseResponse.promise }, + ]) + const { ctx, parent, disposePersistence } = await setupWith(adapter) + const warnings: string[] = [] + ctx.logger.warn = (message: string) => { warnings.push(message) } + + const started = await ctx.subagents.startContinuable(startSpec(parent)) + await vi.waitFor(() => { expect(adapter.requests).toHaveLength(1) }) + // Remove every durability listener, so the final checkpoint cannot confirm. + await disposePersistence!() + releaseResponse.resolve() + + // The handle is still disposed and ownership released, so nothing is pinned. + await waitNoActivation(ctx, started.childId) + await vi.waitFor(() => { + expect(warnings.some(warning => warning.includes('durability'))).toBe(true) + }) + }) + + it('disposes every live Activation forest child-first on manager teardown', async () => { + const hold = Promise.withResolvers() + const adapter = new GatedAdapter([ + { chunks: textResponse('child done') }, + { chunks: textResponse('grandchild'), gate: hold.promise }, + ]) + const { ctx, parent } = await setupWith(adapter) + const started = await ctx.subagents.startContinuable(startSpec(parent)) + const child = await vi.waitFor(() => { + const found = ctx.agents.get(started.childId) + expect(found).toBeDefined() + return found! + }) + const grandchild = await ctx.subagents.startContinuable(startSpec(child)) + await vi.waitFor(() => { expect(ctx.agents.get(grandchild.childId)).toBeDefined() }) + + const disposals: SessionId[] = [] + ctx.on('agent/disposed', (agent) => { disposals.push(agent.id) }) + const drained = ctx.subagents.drainContinuable() + // Let the held model call observe its cancellation so quiescence can settle. + hold.resolve() + await drained + + // Child-first: the grandchild's disposal precedes its parent's. + expect(disposals.indexOf(grandchild.childId)).toBeGreaterThanOrEqual(0) + expect(disposals.indexOf(grandchild.childId)) + .toBeLessThan(disposals.indexOf(started.childId)) + // Durable sessions survive process-local teardown. + const loaded = await ctx.sessionPersistence.load(started.childId) + expect(loaded.meta.id).toBe(started.childId) + }) + + it('rejects new materialization and delivery once draining begins', async () => { + const { ctx, parent } = await setup([textResponse('done')]) + const started = await ctx.subagents.startContinuable(startSpec(parent)) + await waitNoActivation(ctx, started.childId) + + await ctx.subagents.drainContinuable() + + await expect(ctx.subagents.startContinuable(startSpec(parent))) + .rejects.toMatchObject({ code: 'DRAINING' }) + await expect(followup(ctx, { kind: 'user' }, started.childId, message('too late'))) + .rejects.toMatchObject({ code: 'DRAINING' }) + }) + + it('has no automatic replay for an accepted but unlogged message', async () => { + const hold = Promise.withResolvers() + const adapter = new GatedAdapter([{ chunks: textResponse('first'), gate: hold.promise }]) + const { ctx, parent } = await setupWith(adapter) + const started = await ctx.subagents.startContinuable(startSpec(parent)) + await vi.waitFor(() => { expect(adapter.requests).toHaveLength(1) }) + // Accepted into the inbox, but this queued turn never opens. + await followup(ctx, { kind: 'user' }, started.childId, message('never logged')) + + const drained = ctx.subagents.drainContinuable() + hold.resolve() + await drained + await waitNoActivation(ctx, started.childId) + + const loaded = await ctx.sessionPersistence.load(started.childId) + // Only what actually reached the log is reconstructable. + expect(hasUserText(loaded.events, 'never logged')).toBe(false) + }) +}) + +describe('continuable lifecycle observation', () => { + it('emits one paired start/end per residency epoch', async () => { + const { ctx, parent } = await setup([textResponse('first'), textResponse('second')]) + const starts: SubagentRunInfo[] = [] + const ends: SubagentRunEndInfo[] = [] + ctx.on('subagent/start', info => { starts.push(info) }) + ctx.on('subagent/end', info => { ends.push(info) }) + + const started = await ctx.subagents.startContinuable(startSpec(parent)) + await waitNoActivation(ctx, started.childId) + await vi.waitFor(() => { expect(ends).toHaveLength(1) }) + + // A cold resume is a NEW epoch with its own pair. + await followup(ctx, { kind: 'user' }, started.childId, message('again')) + await waitNoActivation(ctx, started.childId) + await vi.waitFor(() => { expect(ends).toHaveLength(2) }) + + expect(starts).toHaveLength(2) + expect(starts.map(info => info.id)).toEqual([started.childId, started.childId]) + expect(starts.map(info => info.provider)).toEqual(['spawn', 'spawn']) + // Each end pairs its own start's runId. + expect(ends.map(info => info.runId)).toEqual(starts.map(info => info.runId)) + }) +}) + +describe('continuable public surface', () => { + it('exposes no cancellation, steering, or report operation', async () => { + const { ctx } = await setup([]) + const subagents: Record = ctx.subagents as unknown as Record + for (const absent of ['cancel', 'kill', 'steer', 'steerContinuable', 'report', 'resume']) { + expect(subagents[absent]).toBeUndefined() + } + // No steering tool and no report tool are registered by this seam. + const names = ctx.tools.schemas().map(schema => schema.name) + expect(names).not.toContain('report') + expect(names).not.toContain('steer_subagent') + }) + + it('keeps one-shot runs free of a steering capability', async () => { + const { ctx, parent } = await setup([textResponse('one shot')]) + const run = await ctx.subagents.start('spawn', { + prompt: message('one-shot work'), + parent, + signal: testSignal, + }) + expect('steer' in run).toBe(false) + await run.result + await run.dispose() + }) + + it('reports a caller-signal abort before acceptance without delivering', async () => { + const { ctx, parent } = await setup([textResponse('first')]) + const started = await ctx.subagents.startContinuable(startSpec(parent)) + await waitNoActivation(ctx, started.childId) + const controller = new AbortController() - controller.abort('caller already cancelled') + controller.abort('caller gave up') + await expect(followup(ctx, { kind: 'user' }, started.childId, message('aborted'), controller.signal)) + .rejects.toThrow() - await expect(followup( - ctx, - parent, - started.childId, - message('must not enqueue'), - controller.signal, - )).rejects.toMatchObject({ code: 'CANCELLED' }) - expect(ctx.agents.get(started.childId)).toBeUndefined() - const snapshot = await waitTerminal(ctx, started.taskId, parent) - expect(snapshot.status).toBe('killed') const loaded = await ctx.sessionPersistence.load(started.childId) - expect(loaded.events.some(event => event.type === 'steering/message')).toBe(false) + expect(hasUserText(loaded.events, 'aborted')).toBe(false) }) - it('rejects before acknowledgement when terminal policy prevents steering admission', async () => { - const { ctx, parent, adapter } = await setup([ - toolCallResponse('c1', 'structured_output', { answer: 7 }), + it('does not cancel an accepted turn when the caller signal aborts afterwards', async () => { + const releaseFirst = Promise.withResolvers() + const adapter = new GatedAdapter([ + { chunks: textResponse('first'), gate: releaseFirst.promise }, + { chunks: textResponse('second') }, ]) - const startedTool = Promise.withResolvers() - const releaseTool = Promise.withResolvers() - ctx.on('tools/pre-execute', async (exec, next) => { - if (exec.name === 'structured_output') { - startedTool.resolve(undefined) - await releaseTool.promise - } - return next() - }) + const { ctx, parent } = await setupWith(adapter) + const started = await ctx.subagents.startContinuable(startSpec(parent)) + await vi.waitFor(() => { expect(adapter.requests).toHaveLength(1) }) - const base = startSpec(parent) - const started = ctx.subagents.startContinuable({ - ...base, - request: { - ...base.request, - outputSchema: { - type: 'object', - properties: { answer: { type: 'number' } }, - required: ['answer'], - }, - }, - }) - await startedTool.promise + const controller = new AbortController() + await followup(ctx, { kind: 'user' }, started.childId, message('survives'), controller.signal) + // After acceptance the manager owns the Activation independently. + controller.abort('caller gave up') - const delivery = ctx.subagents.followup( - parent, - started.childId, - message('follow-up that terminal policy rejects'), - { source: coordinatorSource, signal: testSendSignal }, - ) - releaseTool.resolve(undefined) - await expect(delivery).rejects.toThrow(/message was not delivered/) - - const snapshot = await waitTerminal(ctx, started.taskId, parent) - expect(snapshot.status).toBe('completed') - expect(adapter.requests).toHaveLength(1) + releaseFirst.resolve() + await waitNoActivation(ctx, started.childId) const loaded = await ctx.sessionPersistence.load(started.childId) - expect(loaded.events.some(event => event.type === 'steering/message')).toBe(false) + expect(hasUserText(loaded.events, 'survives')).toBe(true) }) +}) - 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.subagents.startContinuable(startSpec(parent)) - await waitTerminal(ctx, started.taskId, parent) - expect(ctx.agents.get(started.childId)).toBeUndefined() - - const followUp = await ctx.subagents.followup( - parent, - started.childId, - message('and then?'), - { source: coordinatorSource, signal: testSendSignal }, - ) - 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?']) - expect(userMessages.map(event => event.data.source)) - .toEqual([{ kind: 'user' }, coordinatorSource]) - }) - - 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.subagents.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 = await followup(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. - 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.subagents.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 = await followup(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. - 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.subagents.startContinuable(startSpec(parent)) - await waitTerminal(ctx, started.taskId, parent) - const followUp = await followup(ctx, 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.subagents.startContinuable(startSpec(otherParent)) - await waitTerminal(ctx, started.taskId, otherParent) - - const attempt = await followup(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') - 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 = await followup(ctx, parent, SessionId('plain-child'), message('continue?')) - const snapshot = await waitTerminal(ctx, attempt.taskId, parent) - expect(snapshot.status).toBe('failed') - expect(snapshot.detail).toContain( - 'has no supported continuation state and cannot be resumed; do not retry send_message with this id', - ) - }) - - it('derives fallback and bounded labels for resumed activations', async () => { - const { ctx, parent } = await setup([]) - const blank = await followup(ctx, parent, SessionId('blank-child'), message(' ')) - const longText = 'x'.repeat(100) - const long = await followup(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)}…`) - await Promise.all([ - waitTerminal(ctx, blank.taskId, parent), - waitTerminal(ctx, long.taskId, parent), - ]) - }) - - it('rejects delivery to a live agent outside continuation ownership', async () => { +describe('continuable errors', () => { + it('rejects a second live Activation for the same durable child', async () => { const { ctx, parent } = await setup([textResponse('unused')]) - // A live child created outside continuation orchestration. - const handle = await ctx.agents.create({ - sessionId: SessionId('rogue-child'), - meta: { parentSession: parent.id }, - agentOptions: { provider: 'mock', model: 'mock' }, - }) - await expect(followup(ctx, parent, SessionId('rogue-child'), message('hello'))) + // Occupy the id with an unmanaged live Agent. + const squatter = ctx.agentLoop.create(SessionId('squatted'), { provider: 'mock', model: 'mock' }) + await ctx.sessions.flush(squatter.session) + await expect(followup(ctx, { kind: 'user' }, SessionId('squatted'), message('hello'))) .rejects.toThrow(SubagentError) - await expect(followup(ctx, parent, SessionId('rogue-child'), message('hello'))) - .rejects.toThrow(/outside continuation ownership.*not delivered/) - await handle.dispose() - }) - - it('does not fall through to cold resume when steering loses the admission 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((resolve) => { releaseDispose = resolve }) - const provider = ctx.subagents.getProvider('spawn')! - const realStart = provider.start.bind(provider) - provider.start = async (request) => { - const run = await realStart(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.subagents.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((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) - }) - - // Confirmed steering finds the settled child, fails loud, and does NOT start - // a cold resume within this call. - await expect(followup(ctx, parent, started.childId, message('too late?'))) - .rejects.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 = await followup(ctx, 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.subagents.startContinuable(startSpec(parent)) - await waitTerminal(ctx, started.taskId, parent) - const followUp = await followup(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/) - }) - - 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.subagents.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((resolve) => { releaseLoad = resolve }) - ctx.sessionPersistence.load = async (id) => { - await gate - return realLoad(id) - } - - const followUp = await followup(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) - 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.subagents.startContinuable(startSpec(parent)) - await waitTerminal(ctx, started.taskId, parent) - - const realLoad = ctx.sessionPersistence.load.bind(ctx.sessionPersistence) - let releaseLoad!: () => void - const gate = new Promise((resolve) => { releaseLoad = resolve }) - ctx.sessionPersistence.load = async (id) => { - await gate - return realLoad(id) - } - - const first = await followup(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. - await expect(followup(ctx, parent, started.childId, message('second follow-up'))) - .rejects.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('service disposal with live activations', () => { - it('cancels and settles a starting activation on service disposal instead of stranding it', async () => { - const ctx = new Context() - await mountAgentLoopTestDependencies(ctx) - const root = mkdtempSync(join(tmpdir(), 'dsh-subagent-continuation-hmr-')) - roots.push(root) - await ctx.plugin(JsonlSessionPersistence, { root }) - await ctx.plugin(AgentLoop, { agents: [] }) - const subagentsFiber = await ctx.plugin(SubagentService) - await ctx.plugin(LocalTaskService) - await ctx.plugin(ToolTasks, {}) - // A provider that stays pending until its signal aborts, so the activation - // is observably mid-start when the subagent service is disposed. - let sawAbort = false - ctx.subagents.registerProvider({ - name: 'pending', - capabilities: { outputSchema: false, depthLimit: false, toolFilter: false, persona: false }, - inheritsParentContext: false, - start: request => new Promise((_resolve, reject) => { - request.signal.addEventListener('abort', () => { - sawAbort = true - reject(new Error('startup aborted')) - }, { once: true }) - }), - resume: () => Promise.reject(new Error('unreachable')), - }) - ctx.llm.registerAdapter(['mock'], new MockAdapter([])) - const parent = ctx.agentLoop.create(SessionId('parent'), { provider: 'mock', model: 'mock' }) - - const started = ctx.subagents.startContinuable({ - provider: 'pending', - label: 'will be interrupted', - request: { prompt: message('go'), parent }, - }) - // LocalTaskService keeps the producer Task; the disposing subagent service must - // cancel its activation and await settlement rather than strand it. - await subagentsFiber.dispose() - expect(sawAbort).toBe(true) - const snapshot = await waitTerminal(ctx, started.taskId, parent) - expect(snapshot.status).toBe('killed') - }) -}) - -describe('outcome mapping helpers', () => { - it.each([ - ['completed', { status: 'completed', output: 'partial' }], - ['aborted', { status: 'killed' }], - ['error', { status: 'failed', detail: 'error' }], - ['max-tokens', { status: 'failed', detail: 'max-tokens' }], - ['refusal', { status: 'failed', detail: 'refusal' }], - ['paused', { status: 'failed', detail: 'paused' }], - ] as const)('settleRun maps the %s stop reason onto its Task outcome', async (stopReason, expected) => { - const output = [{ type: 'text' as const, text: 'partial' }] - await expect(settleRun({ - id: SessionId('child'), - localAgent: undefined, - result: Promise.resolve({ output, stopReason: stopReason as never }), - dispose: () => Promise.resolve(), - })).resolves.toEqual(expected) - }) - - 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 durabilityMessage = 'subagent "child-3" durability checkpoint failed; latest state unavailable: disk full' - const durabilityFailed = await settleRun({ - id: SessionId('child-3'), - localAgent: undefined, - result: Promise.reject(new HarnessError( - durabilityMessage, - 'DURABILITY_FAILED', - { cause: new Error('disk full') }, - )), - dispose: () => Promise.resolve(), - }) - expect(durabilityFailed).toEqual({ status: 'failed', detail: durabilityMessage }) - - const disposeFailed = await settleRun({ - id: SessionId('child-4'), - 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-5'), - 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', - }) + void parent }) })