|
|
@@ -7,7 +7,7 @@ import type { Agent } from '@deepseek-ai/dsh-agent'
|
|
|
import { assembleContextFor } from '@deepseek-ai/dsh-agent'
|
|
|
import AgentLoop from '@deepseek-ai/dsh-agent-loop'
|
|
|
import { mountAgentLoopTestDependencies } from '@deepseek-ai/dsh-agent-loop-testkit'
|
|
|
-import { CallId, LlmAdapter } from '@deepseek-ai/dsh-llm'
|
|
|
+import { CallId, LlmAdapter, createUserMessage } from '@deepseek-ai/dsh-llm'
|
|
|
import type { GenerateOptions, StreamChunk } from '@deepseek-ai/dsh-llm'
|
|
|
import { SessionId } from '@deepseek-ai/dsh-session'
|
|
|
import type { SessionEvent } from '@deepseek-ai/dsh-session'
|
|
|
@@ -20,22 +20,39 @@ import * as tool from '../src/index.ts'
|
|
|
|
|
|
const testSignal = new AbortController().signal
|
|
|
|
|
|
-/** Adapter that keeps child Activations resident until released. */
|
|
|
+/** Adapter that keeps selected Agent requests open until released. */
|
|
|
class HeldAdapter extends LlmAdapter {
|
|
|
readonly requests: GenerateOptions[] = []
|
|
|
- private readonly gate = Promise.withResolvers<undefined>()
|
|
|
+ private readonly gates = new Map<GenerateOptions['sessionId'], PromiseWithResolvers<undefined>>()
|
|
|
+ private readonly releasedSessions = new Set<GenerateOptions['sessionId']>()
|
|
|
+ private released = false
|
|
|
|
|
|
async * stream(options: GenerateOptions): AsyncIterable<StreamChunk> {
|
|
|
this.requests.push(options)
|
|
|
- await this.gate.promise
|
|
|
+ if (!this.released && !this.releasedSessions.has(options.sessionId)) {
|
|
|
+ let gate = this.gates.get(options.sessionId)
|
|
|
+ if (gate === undefined) {
|
|
|
+ gate = Promise.withResolvers<undefined>()
|
|
|
+ this.gates.set(options.sessionId, gate)
|
|
|
+ }
|
|
|
+ await gate.promise
|
|
|
+ }
|
|
|
for (const chunk of textResponse('held answer')) {
|
|
|
if (options.signal?.aborted) throw new Error('aborted')
|
|
|
yield chunk
|
|
|
}
|
|
|
}
|
|
|
|
|
|
- release(): void {
|
|
|
- this.gate.resolve(undefined)
|
|
|
+ release(sessionId?: SessionId): void {
|
|
|
+ if (sessionId !== undefined) {
|
|
|
+ this.releasedSessions.add(sessionId)
|
|
|
+ this.gates.get(sessionId)?.resolve(undefined)
|
|
|
+ this.gates.delete(sessionId)
|
|
|
+ return
|
|
|
+ }
|
|
|
+ this.released = true
|
|
|
+ for (const gate of this.gates.values()) gate.resolve(undefined)
|
|
|
+ this.gates.clear()
|
|
|
}
|
|
|
}
|
|
|
|
|
|
@@ -86,6 +103,17 @@ async function startChild(ctx: Context, parent: Agent, prompt = 'child task') {
|
|
|
return { started, child }
|
|
|
}
|
|
|
|
|
|
+/** Start one parent request that remains open in the held adapter. */
|
|
|
+async function startHeldParentTurn(parent: Agent, adapter: HeldAdapter): Promise<void> {
|
|
|
+ parent.followup(createUserMessage({
|
|
|
+ content: [{ type: 'text', text: 'parent work' }],
|
|
|
+ source: { kind: 'user' },
|
|
|
+ }))
|
|
|
+ await vi.waitFor(() => {
|
|
|
+ expect(adapter.requests.some(request => request.sessionId === parent.id)).toBe(true)
|
|
|
+ })
|
|
|
+}
|
|
|
+
|
|
|
let calls = 0
|
|
|
function callReport(ctx: Context, child: Agent, output: string, signal = testSignal) {
|
|
|
return ctx.tools.execute({
|
|
|
@@ -204,8 +232,8 @@ describe('dsh-tool-subagent-report', () => {
|
|
|
expect(adapter.requests.filter(request => request.sessionId === parent.id)).toHaveLength(parentRequests)
|
|
|
})
|
|
|
|
|
|
- it('queues wakeup reports as one later parent turn', async () => {
|
|
|
- const { ctx, parent, adapter } = await setup({ config: { reportDelivery: 'wakeup' } })
|
|
|
+ it('delivers next-step reports through waking steering', async () => {
|
|
|
+ const { ctx, parent, adapter } = await setup({ config: { reportDelivery: 'next-step' } })
|
|
|
const { child } = await startChild(ctx, parent)
|
|
|
const enqueues: string[] = []
|
|
|
ctx.on('agent/inbox/inserted', ({ agent, message }) => {
|
|
|
@@ -216,19 +244,37 @@ describe('dsh-tool-subagent-report', () => {
|
|
|
|
|
|
const result = await callReport(ctx, child, 'WAKE_UP')
|
|
|
expect(result.isError).toBe(false)
|
|
|
- expect(enqueues).toEqual(['queued'])
|
|
|
+ expect(enqueues).toEqual(['steering'])
|
|
|
await vi.waitFor(() => {
|
|
|
expect(adapter.requests.some(request => request.sessionId === parent.id)).toBe(true)
|
|
|
})
|
|
|
})
|
|
|
|
|
|
- it('preserves accepted order across repeated reports', async () => {
|
|
|
- const { ctx, parent } = await setup()
|
|
|
+ it('batches repeated next-step reports in accepted order', async () => {
|
|
|
+ const { ctx, parent, adapter } = await setup({ config: { reportDelivery: 'next-step' } })
|
|
|
+ await startHeldParentTurn(parent, adapter)
|
|
|
const { child } = await startChild(ctx, parent)
|
|
|
|
|
|
expect((await callReport(ctx, child, 'FIRST')).isError).toBe(false)
|
|
|
expect((await callReport(ctx, child, 'SECOND')).isError).toBe(false)
|
|
|
expect(reports(parent).map(report => report.text.split('\n').at(-1))).toEqual(['FIRST', 'SECOND'])
|
|
|
+ expect(parent.inbox.nextStep).toHaveLength(2)
|
|
|
+ })
|
|
|
+
|
|
|
+ it('keeps a report before the child settlement in one busy-parent batch', async () => {
|
|
|
+ const { ctx, parent, adapter } = await setup({ config: { reportDelivery: 'next-step' } })
|
|
|
+ await startHeldParentTurn(parent, adapter)
|
|
|
+ const { started, child } = await startChild(ctx, parent)
|
|
|
+
|
|
|
+ expect((await callReport(ctx, child, 'ORDERED_REPORT')).isError).toBe(false)
|
|
|
+ adapter.release(started.childId)
|
|
|
+ await vi.waitFor(() => { expect(ctx.agents.get(started.childId)).toBeUndefined() })
|
|
|
+
|
|
|
+ expect(parent.inbox.nextStep.map(message => message.source.kind)).toEqual([
|
|
|
+ 'subagent-report',
|
|
|
+ 'subagent-settled',
|
|
|
+ ])
|
|
|
+ expect(parent.inbox.nextTurn).toHaveLength(0)
|
|
|
})
|
|
|
|
|
|
it('keeps an accepted report after the child settles', async () => {
|
|
|
@@ -261,8 +307,8 @@ describe('dsh-tool-subagent-report', () => {
|
|
|
expect(reports(child)[0]?.text).toContain('FROM_GRANDCHILD')
|
|
|
})
|
|
|
|
|
|
- it('accounts wakeup reports delivered to a resident continuable parent', async () => {
|
|
|
- const { ctx, parent, adapter } = await setup({ config: { reportDelivery: 'wakeup' } })
|
|
|
+ it('accounts next-step reports delivered to a resident continuable parent', async () => {
|
|
|
+ const { ctx, parent, adapter } = await setup({ config: { reportDelivery: 'next-step' } })
|
|
|
const { child } = await startChild(ctx, parent, 'outer task')
|
|
|
const { started: grandchildStart, child: grandchild } = await startChild(ctx, child, 'inner task')
|
|
|
|
|
|
@@ -509,9 +555,10 @@ describe('dsh-tool-subagent-report', () => {
|
|
|
expect('default' in tool).toBe(false)
|
|
|
expect(tool.name).toBe('tool-subagent-report')
|
|
|
expect(tool.inject).toEqual(['subagents', 'tools', 'systemPrompt'])
|
|
|
- // Waking is the default because a report that never wakes its parent
|
|
|
- // cannot deliver a result to an agent that already parked.
|
|
|
- expect(tool.Config({}).reportDelivery).toBe('wakeup')
|
|
|
+ // Next-step delivery wakes a parked parent and lets a running parent act at
|
|
|
+ // its nearest safe boundary.
|
|
|
+ expect(tool.Config({}).reportDelivery).toBe('next-step')
|
|
|
+ expect(() => tool.Config({ reportDelivery: 'wakeup' } as never)).toThrow()
|
|
|
expect(() => tool.Config({ reportDelivery: 'shout' } as never)).toThrow()
|
|
|
})
|
|
|
|
|
|
@@ -526,7 +573,7 @@ describe('dsh-tool-subagent-report', () => {
|
|
|
})
|
|
|
|
|
|
expect((await callReport(ctx, child, 'DEFAULT_WAKES')).isError).toBe(false)
|
|
|
- expect(enqueues).toEqual(['queued'])
|
|
|
+ expect(enqueues).toEqual(['steering'])
|
|
|
await vi.waitFor(() => {
|
|
|
expect(adapter.requests.some(request => request.sessionId === parent.id)).toBe(true)
|
|
|
})
|