|
|
@@ -1,170 +1,184 @@
|
|
|
-/**
|
|
|
- * One-shot runner behavior over a scripted in-process API: idle-to-idle
|
|
|
- * aggregation (last text of the whole interval), exit-code mapping by the
|
|
|
- * final turn-end reason, stream-error and RPC-error paths, and the
|
|
|
- * launcher-owned `ctx.headlessIo` requirement.
|
|
|
- */
|
|
|
+/** Direct one-shot Agent driving, durable aggregation, flushing, and exit mapping. */
|
|
|
|
|
|
import { describe, expect, it } from 'vitest'
|
|
|
import { Context } from 'cordis'
|
|
|
-import type { Agent } from '@deepseek-ai/dsh-agent'
|
|
|
+import AgentRegistry, { Inbox } from '@deepseek-ai/dsh-agent'
|
|
|
+import type { Agent, AgentHandle, CreateAgentOptions } from '@deepseek-ai/dsh-agent'
|
|
|
+import AgentDefaultModelService from '@deepseek-ai/dsh-agent-default-model'
|
|
|
+import { createAssistantMessage } from '@deepseek-ai/dsh-llm'
|
|
|
+import SessionStore from '@deepseek-ai/dsh-session'
|
|
|
+import type { Session, UserMessage } from '@deepseek-ai/dsh-session'
|
|
|
import { apply, Config, type HeadlessIo } from '../src/index.ts'
|
|
|
|
|
|
-interface ScriptedEvent { type: string; seq?: number; time?: number; sessionId?: string; data: Record<string, unknown> }
|
|
|
-
|
|
|
-let nextSeq = 0
|
|
|
-/** Stamp the envelope fields the wire schema requires. */
|
|
|
-function stamped(event: ScriptedEvent): ScriptedEvent {
|
|
|
- nextSeq += 1
|
|
|
- return { seq: nextSeq, time: nextSeq, ...event }
|
|
|
+interface Script {
|
|
|
+ before?(session: Session): void
|
|
|
+ afterPrompt(session: Session, message: UserMessage): Promise<void> | void
|
|
|
}
|
|
|
|
|
|
-interface RpcShapedRequest { rpcId: string }
|
|
|
-
|
|
|
-interface ScriptedApiOptions {
|
|
|
- promptFails?: boolean
|
|
|
- framesAfterPrompt?: boolean
|
|
|
- onPrompt?: () => void
|
|
|
+function appendTurn(
|
|
|
+ session: Session,
|
|
|
+ turn: number,
|
|
|
+ message: UserMessage,
|
|
|
+ text: string | undefined,
|
|
|
+ completed: boolean,
|
|
|
+): void {
|
|
|
+ session.append('turn/start', { turn })
|
|
|
+ session.append('step/start', { turn, step: 1 })
|
|
|
+ session.append('user/message', message, { surfaceOp: 'append' })
|
|
|
+ if (text !== undefined) {
|
|
|
+ session.append('assistant/message', {
|
|
|
+ turn,
|
|
|
+ step: 1,
|
|
|
+ message: createAssistantMessage({
|
|
|
+ content: [{ type: 'text', text }],
|
|
|
+ source: { provider: 'test-provider', model: 'test-model' },
|
|
|
+ }),
|
|
|
+ }, { surfaceOp: 'append' })
|
|
|
+ }
|
|
|
+ session.append('step/end', { turn, step: 1 })
|
|
|
+ session.append('turn/end', {
|
|
|
+ turn,
|
|
|
+ reason: completed
|
|
|
+ ? { kind: 'completed' }
|
|
|
+ : { kind: 'aborted', reason: { kind: 'user' } },
|
|
|
+ })
|
|
|
}
|
|
|
|
|
|
-/** Build a fake apiProxy (echoing rpcIds like the real gateway) whose mux stream replays `events` for the created session. */
|
|
|
-function scriptedApi(events: ScriptedEvent[], options: ScriptedApiOptions = {}): unknown {
|
|
|
- let releaseFrames = (): void => {}
|
|
|
- const framesReady = options.framesAfterPrompt === true
|
|
|
- ? new Promise<void>((resolve) => { releaseFrames = resolve })
|
|
|
- : Promise.resolve()
|
|
|
- const prepared = events.map((event) => {
|
|
|
- if (event.type === 'stream/error') return { streamError: true } as const
|
|
|
- const { sessionId = 'S1', ...rest } = event
|
|
|
- return { streamError: false, sessionId, event: stamped(rest) } as const
|
|
|
+/** Mount the real registries around a small scripted Agent factory. */
|
|
|
+async function bench(script: Script): Promise<{
|
|
|
+ ctx: Context
|
|
|
+ run(): Promise<{ code: number; out: string; err: string; order: string[] }>
|
|
|
+}> {
|
|
|
+ const ctx = new Context()
|
|
|
+ await ctx.plugin(SessionStore)
|
|
|
+ await ctx.plugin(AgentRegistry)
|
|
|
+ await ctx.plugin(AgentDefaultModelService, { provider: 'test-provider', model: 'test-model' })
|
|
|
+ ctx.agents.setFactory({
|
|
|
+ async createAgent(ownerCtx: Context, options: CreateAgentOptions): Promise<AgentHandle> {
|
|
|
+ const session = ctx.sessions.create(options.sessionId, {
|
|
|
+ ...options.meta === undefined ? {} : { meta: options.meta },
|
|
|
+ })
|
|
|
+ let idle = Promise.resolve()
|
|
|
+ const agent = {} as Agent
|
|
|
+ const agentCtx = ownerCtx.extend({ agent })
|
|
|
+ Object.assign(agent, {
|
|
|
+ id: session.id,
|
|
|
+ options: options.agentOptions ?? {},
|
|
|
+ session,
|
|
|
+ inbox: new Inbox(session, { inserted: () => {}, discarded: () => {}, claimed: () => {} }),
|
|
|
+ status: 'idle',
|
|
|
+ ctx: agentCtx,
|
|
|
+ cancel: () => {},
|
|
|
+ runMaintenance: () => Promise.reject(new Error('not used')),
|
|
|
+ send: () => {},
|
|
|
+ followup: (message: UserMessage) => {
|
|
|
+ agent.inbox.append('next-turn', message)
|
|
|
+ idle = Promise.resolve().then(() => script.afterPrompt(session, message))
|
|
|
+ },
|
|
|
+ steer: () => {},
|
|
|
+ inject: () => {},
|
|
|
+ whenIdle: () => idle,
|
|
|
+ } satisfies Partial<Agent>)
|
|
|
+ await options.setup?.(agentCtx)
|
|
|
+ script.before?.(session)
|
|
|
+ ctx.agents.register(agent)
|
|
|
+ return { agent, dispose: () => Promise.resolve() }
|
|
|
+ },
|
|
|
+ resume: () => Promise.reject(new Error('not used')),
|
|
|
})
|
|
|
return {
|
|
|
- sessions: {
|
|
|
- create: (request: RpcShapedRequest) =>
|
|
|
- Promise.resolve({ rpcId: request.rpcId, result: { ok: true, value: { sessionId: 'S1' } } }),
|
|
|
- prompt: (request: RpcShapedRequest) => {
|
|
|
- releaseFrames()
|
|
|
- options.onPrompt?.()
|
|
|
- return Promise.resolve(options.promptFails === true
|
|
|
- // A code from the closed wire union: the carrier schema rejects invented codes.
|
|
|
- ? { rpcId: request.rpcId, result: { ok: false, error: { code: 'agent-busy', message: 'agent is busy', details: { reason: 'test' } } } }
|
|
|
- : { rpcId: request.rpcId, result: { ok: true, value: { accepted: true } } })
|
|
|
- },
|
|
|
- },
|
|
|
- events: {
|
|
|
- mux: async function* () {
|
|
|
- await framesReady
|
|
|
- for (const item of prepared) {
|
|
|
- if (item.streamError) {
|
|
|
- yield { rpcId: 'e', payload: { type: 'stream/error', error: { code: 'cancelled', message: 'stream broke', details: {} } } }
|
|
|
- continue
|
|
|
- }
|
|
|
- yield { rpcId: 'e', payload: { type: 'session/event', sessionId: item.sessionId, event: item.event } }
|
|
|
+ ctx,
|
|
|
+ run: async () => {
|
|
|
+ let out = ''
|
|
|
+ let err = ''
|
|
|
+ const order: string[] = []
|
|
|
+ ctx.on('session/flush', () => { order.push('flush') })
|
|
|
+ const exited = new Promise<number>((resolve) => {
|
|
|
+ const io: HeadlessIo = {
|
|
|
+ stdout: { write: (chunk: string) => { out += chunk; return true } },
|
|
|
+ stderr: { write: (chunk: string) => { err += chunk; return true } },
|
|
|
+ exit: (code) => { order.push('exit'); resolve(code) },
|
|
|
}
|
|
|
- },
|
|
|
+ ctx.provide('headlessIo', io)
|
|
|
+ })
|
|
|
+ apply(ctx, { task: 'do the thing' })
|
|
|
+ return { code: await exited, out, err, order }
|
|
|
},
|
|
|
}
|
|
|
}
|
|
|
|
|
|
-/**
|
|
|
- * Mount the runner against a scripted API, emit the idle transition after the
|
|
|
- * scripted frames drain, and wait for its exit request.
|
|
|
- */
|
|
|
-async function run(
|
|
|
- events: ScriptedEvent[],
|
|
|
- options: { promptFails?: boolean; framesAfterPrompt?: boolean; idleInPrompt?: boolean } = {},
|
|
|
-): Promise<{ code: number; out: string; err: string }> {
|
|
|
- const ctx = new Context()
|
|
|
- let out = ''
|
|
|
- let err = ''
|
|
|
- const exited = new Promise<number>((resolve) => {
|
|
|
- const io: HeadlessIo = {
|
|
|
- stdout: { write: (chunk: string) => { out += chunk; return true } },
|
|
|
- stderr: { write: (chunk: string) => { err += chunk; return true } },
|
|
|
- exit: resolve,
|
|
|
- }
|
|
|
- ctx.provide('headlessIo', io)
|
|
|
- })
|
|
|
- const emitIdle = (): void => {
|
|
|
- ctx.emit('agent/status', { agent: { id: 'S1', session: { seq: nextSeq + 1 } } as Agent, status: 'idle' })
|
|
|
- }
|
|
|
- ctx.provide('apiProxy', scriptedApi(events, {
|
|
|
- ...options.promptFails === undefined ? {} : { promptFails: options.promptFails },
|
|
|
- ...options.framesAfterPrompt === undefined ? {} : { framesAfterPrompt: options.framesAfterPrompt },
|
|
|
- ...options.idleInPrompt === true ? { onPrompt: emitIdle } : {},
|
|
|
- }) as never)
|
|
|
- ctx.provide('httpServer', { port: 12345 } as never)
|
|
|
- apply(ctx, { task: 'do the thing' })
|
|
|
- // Quiescence is out of band: give the scripted stream a beat to drain, then
|
|
|
- // flip the agent idle exactly as the loop would. Foreign agents and
|
|
|
- // non-idle transitions must not settle the run.
|
|
|
- if (options.idleInPrompt !== true) {
|
|
|
- await new Promise(resolve => setTimeout(resolve, 10))
|
|
|
- ctx.emit('agent/status', { agent: { id: 'OTHER' } as Agent, status: 'idle' })
|
|
|
- ctx.emit('agent/status', { agent: { id: 'S1' } as Agent, status: 'running' })
|
|
|
- emitIdle()
|
|
|
- }
|
|
|
- const code = await exited
|
|
|
- await ctx.fiber.dispose()
|
|
|
- return { code, out, err }
|
|
|
-}
|
|
|
-
|
|
|
-const startupTurn: ScriptedEvent = { type: 'turn/start', data: { turn: 0, trigger: { kind: 'startup' } } }
|
|
|
-const messageTurn: ScriptedEvent = { type: 'turn/start', data: { turn: 1, trigger: { kind: 'message' } } }
|
|
|
-const text = (turn: number, value: string): ScriptedEvent => ({
|
|
|
- type: 'assistant/message',
|
|
|
- data: { turn, message: { content: [{ type: 'text', text: value }] } },
|
|
|
-})
|
|
|
-const end = (turn: number, reason: string): ScriptedEvent => ({ type: 'turn/end', data: { turn, reason: { kind: reason } } })
|
|
|
-
|
|
|
describe('headless runner', () => {
|
|
|
- it('aggregates to quiescence: last text wins across turns, final turn-end reason maps to exit 0', async () => {
|
|
|
- const { code, out, err } = await run([
|
|
|
- // Frames before the first turn/start are outside the task interval.
|
|
|
- { type: 'assistant/message', data: { turn: 0, message: { content: [{ type: 'text', text: 'pre-task noise' }] } } },
|
|
|
- startupTurn,
|
|
|
- // Off-session, non-text, and text-empty frames never affect the aggregate.
|
|
|
- { type: 'assistant/message', sessionId: 'OTHER', data: { turn: 1, message: { content: [{ type: 'text', text: 'other session' }] } } },
|
|
|
- { type: 'assistant/message', data: { turn: 1, message: { content: [{ type: 'tool_call', text: 'ignored' }] } } },
|
|
|
- text(0, 'draft'),
|
|
|
- end(0, 'completed'),
|
|
|
- messageTurn,
|
|
|
- text(1, 'final answer'),
|
|
|
- end(1, 'completed'),
|
|
|
- ])
|
|
|
- expect(code).toBe(0)
|
|
|
- expect(out).toBe('final answer\n')
|
|
|
- expect(err).toContain('observing at http://127.0.0.1:12345')
|
|
|
+ it('aggregates the final text across the complete idle-to-idle interval and flushes before exit', async () => {
|
|
|
+ const test = await bench({
|
|
|
+ before(session) {
|
|
|
+ const setupMessage = {
|
|
|
+ role: 'user', content: [{ type: 'text', text: 'setup' }], source: { kind: 'user' }, id: 'setup',
|
|
|
+ } as UserMessage
|
|
|
+ appendTurn(session, 0, setupMessage, 'pre-task noise', true)
|
|
|
+ },
|
|
|
+ async afterPrompt(session, message) {
|
|
|
+ await Promise.resolve()
|
|
|
+ appendTurn(session, 1, message, '', true)
|
|
|
+ appendTurn(session, 2, message, 'final answer', true)
|
|
|
+ },
|
|
|
+ })
|
|
|
+ const result = await test.run()
|
|
|
+ expect(result).toEqual({
|
|
|
+ code: 0,
|
|
|
+ out: 'final answer\n',
|
|
|
+ err: '',
|
|
|
+ order: ['flush', 'exit'],
|
|
|
+ })
|
|
|
+ await test.ctx.fiber.dispose()
|
|
|
})
|
|
|
|
|
|
- it('consumes through the idle sequence when queued frames arrive after the status transition', async () => {
|
|
|
- const { code, out } = await run(
|
|
|
- [messageTurn, text(1, 'race-free answer'), end(1, 'completed')],
|
|
|
- { framesAfterPrompt: true, idleInPrompt: true },
|
|
|
- )
|
|
|
- expect(code).toBe(0)
|
|
|
- expect(out).toBe('race-free answer\n')
|
|
|
+ it('waits for asynchronously appended events instead of racing Agent idleness', async () => {
|
|
|
+ const test = await bench({
|
|
|
+ afterPrompt: async (session, message) => {
|
|
|
+ await new Promise(resolve => setTimeout(resolve, 5))
|
|
|
+ appendTurn(session, 1, message, 'race-free answer', true)
|
|
|
+ },
|
|
|
+ })
|
|
|
+ expect(await test.run()).toMatchObject({ code: 0, out: 'race-free answer\n', err: '' })
|
|
|
+ await test.ctx.fiber.dispose()
|
|
|
})
|
|
|
|
|
|
- it('exits 1 when the final turn ends for any other reason', async () => {
|
|
|
- const { code } = await run([messageTurn, end(1, 'aborted')])
|
|
|
- expect(code).toBe(1)
|
|
|
+ it('exits 1 when the final turn does not complete', async () => {
|
|
|
+ const test = await bench({
|
|
|
+ afterPrompt(session, message) { appendTurn(session, 1, message, undefined, false) },
|
|
|
+ })
|
|
|
+ expect(await test.run()).toMatchObject({ code: 1, out: '\n', err: '' })
|
|
|
+ await test.ctx.fiber.dispose()
|
|
|
})
|
|
|
|
|
|
- it('exits 1 when no turn ever starts (idle without work)', async () => {
|
|
|
- const { code, out } = await run([])
|
|
|
- expect(code).toBe(1)
|
|
|
- expect(out).toBe('\n')
|
|
|
+ it('prints the durable model failure when the final turn ends in error', async () => {
|
|
|
+ const test = await bench({
|
|
|
+ afterPrompt(session, message) {
|
|
|
+ session.append('turn/start', { turn: 1 })
|
|
|
+ session.append('step/start', { turn: 1, step: 1 })
|
|
|
+ session.append('user/message', message, { surfaceOp: 'append' })
|
|
|
+ session.append('step/end', { turn: 1, step: 1 })
|
|
|
+ session.append('turn/end', {
|
|
|
+ turn: 1,
|
|
|
+ reason: { kind: 'error', error: { code: 'SERVER', message: 'provider unavailable' } },
|
|
|
+ })
|
|
|
+ },
|
|
|
+ })
|
|
|
+ expect(await test.run()).toMatchObject({
|
|
|
+ code: 1,
|
|
|
+ out: '\n',
|
|
|
+ err: 'dsh: SERVER: provider unavailable\n',
|
|
|
+ })
|
|
|
+ await test.ctx.fiber.dispose()
|
|
|
})
|
|
|
|
|
|
- it('keeps the error outcome after a stream error ends the frame consumer early', async () => {
|
|
|
- const { code } = await run([messageTurn, { type: 'stream/error', data: {} }, end(1, 'completed')])
|
|
|
- // The consumer stopped at the stream error; the completed turn-end after
|
|
|
- // it is never observed, so the reason stays 'error'.
|
|
|
- expect(code).toBe(1)
|
|
|
+ it('exits 1 when the owned interval contains no turn', async () => {
|
|
|
+ const test = await bench({ afterPrompt: () => {} })
|
|
|
+ expect(await test.run()).toMatchObject({ code: 1, out: '\n', err: '' })
|
|
|
+ await test.ctx.fiber.dispose()
|
|
|
})
|
|
|
|
|
|
- it('prints an RPC business error and exits 1 without waiting for idle', async () => {
|
|
|
+ it('reports a direct Agent creation failure', async () => {
|
|
|
const ctx = new Context()
|
|
|
let err = ''
|
|
|
const exited = new Promise<number>((resolve) => {
|
|
|
@@ -174,15 +188,16 @@ describe('headless runner', () => {
|
|
|
exit: resolve,
|
|
|
} satisfies HeadlessIo)
|
|
|
})
|
|
|
- ctx.provide('apiProxy', scriptedApi([messageTurn, end(1, 'completed')], { promptFails: true }) as never)
|
|
|
- ctx.provide('httpServer', { port: 1 } as never)
|
|
|
+ ctx.provide('agentDefaultModel', { currentSelection: () => ({ provider: 'p', model: 'm' }) } as never)
|
|
|
+ ctx.provide('sessions', { flush: () => Promise.resolve(true) } as never)
|
|
|
+ ctx.provide('agents', { create: () => Promise.reject(new Error('factory exploded')) } as never)
|
|
|
apply(ctx, { task: 't' })
|
|
|
expect(await exited).toBe(1)
|
|
|
- expect(err).toContain('agent-busy')
|
|
|
+ expect(err).toBe('dsh: factory exploded\n')
|
|
|
await ctx.fiber.dispose()
|
|
|
})
|
|
|
|
|
|
- it('reports the stream-failed diagnostic when the event channel dies, still settling at idle', async () => {
|
|
|
+ it('stringifies a non-Error Agent creation failure', async () => {
|
|
|
const ctx = new Context()
|
|
|
let err = ''
|
|
|
const exited = new Promise<number>((resolve) => {
|
|
|
@@ -192,67 +207,52 @@ describe('headless runner', () => {
|
|
|
exit: resolve,
|
|
|
} satisfies HeadlessIo)
|
|
|
})
|
|
|
- ctx.provide('apiProxy', {
|
|
|
- sessions: {
|
|
|
- create: (request: RpcShapedRequest) =>
|
|
|
- Promise.resolve({ rpcId: request.rpcId, result: { ok: true, value: { sessionId: 'S1' } } }),
|
|
|
- prompt: (request: RpcShapedRequest) =>
|
|
|
- Promise.resolve({ rpcId: request.rpcId, result: { ok: true, value: { accepted: true } } }),
|
|
|
+ ctx.provide('agentDefaultModel', { currentSelection: () => ({ provider: 'p', model: 'm' }) } as never)
|
|
|
+ ctx.provide('sessions', { flush: () => Promise.resolve(true) } as never)
|
|
|
+ const rejected = {
|
|
|
+ then(_resolve: (value: never) => void, reject: (reason: unknown) => void): void {
|
|
|
+ reject('factory exploded')
|
|
|
},
|
|
|
- events: {
|
|
|
- // Synchronous throw: the SSE response never forms, so the client-side
|
|
|
- // iterable rejects — the runner's own catch path, not a carrier frame.
|
|
|
- mux: () => { throw new Error('channel exploded') },
|
|
|
- },
|
|
|
- } as never)
|
|
|
- ctx.provide('httpServer', { port: 1 } as never)
|
|
|
+ }
|
|
|
+ ctx.provide('agents', { create: () => rejected } as never)
|
|
|
apply(ctx, { task: 't' })
|
|
|
- await new Promise(resolve => setTimeout(resolve, 10))
|
|
|
- ctx.emit('agent/status', { agent: { id: 'S1', session: { seq: nextSeq + 1 } } as Agent, status: 'idle' })
|
|
|
expect(await exited).toBe(1)
|
|
|
- expect(err).toContain('event stream failed')
|
|
|
+ expect(err).toBe('dsh: factory exploded\n')
|
|
|
await ctx.fiber.dispose()
|
|
|
})
|
|
|
|
|
|
- it('waits for Loader settlement and abandons the run when the tree died during it', async () => {
|
|
|
+ it('abandons a run when the tree is disposed during Loader settlement', async () => {
|
|
|
const ctx = new Context()
|
|
|
- let err = ''
|
|
|
let exited = false
|
|
|
ctx.provide('headlessIo', {
|
|
|
stdout: { write: () => true },
|
|
|
- stderr: { write: (chunk: string) => { err += chunk; return true } },
|
|
|
+ stderr: { write: () => true },
|
|
|
exit: () => { exited = true },
|
|
|
} satisfies HeadlessIo)
|
|
|
- ctx.provide('apiProxy', scriptedApi([]) as never)
|
|
|
- // The webserver is provided by a child fiber whose disposal (early
|
|
|
- // SIGTERM during the boot window) removes the service; settlement
|
|
|
- // resolves only afterwards, and the runner must abandon rather than
|
|
|
- // crash on the torn-down port read.
|
|
|
- const webserverFiber = ctx.plugin((childCtx: Context) => {
|
|
|
- childCtx.provide('httpServer', { port: 1 } as never)
|
|
|
+ const services = ctx.plugin((child: Context) => {
|
|
|
+ child.provide('agentDefaultModel', { currentSelection: () => ({ provider: 'p', model: 'm' }) } as never)
|
|
|
+ child.provide('sessions', {} as never)
|
|
|
+ child.provide('agents', {} as never)
|
|
|
})
|
|
|
- await webserverFiber
|
|
|
+ await services
|
|
|
let release: () => void
|
|
|
const settlement = new Promise<void>((resolve) => { release = resolve })
|
|
|
ctx.provide('loader', { await: () => settlement } as never)
|
|
|
apply(ctx, { task: 't' })
|
|
|
- await webserverFiber.dispose()
|
|
|
+ await services.dispose()
|
|
|
release!()
|
|
|
await new Promise(resolve => setTimeout(resolve, 10))
|
|
|
- expect(err).toBe('')
|
|
|
expect(exited).toBe(false)
|
|
|
await ctx.fiber.dispose()
|
|
|
})
|
|
|
|
|
|
it('fails loud without the launcher-owned headlessIo seam', () => {
|
|
|
const ctx = new Context()
|
|
|
- ctx.provide('apiProxy', scriptedApi([]) as never)
|
|
|
- ctx.provide('httpServer', { port: 1 } as never)
|
|
|
expect(() => { apply(ctx, { task: 't' }) }).toThrow('must provide ctx.headlessIo')
|
|
|
})
|
|
|
|
|
|
it('validates config: the task is required', () => {
|
|
|
- expect(() => new Config({ } as never)).toThrow()
|
|
|
+ expect(() => new Config({} as never)).toThrow()
|
|
|
expect(new Config({ task: 'x' })).toEqual({ task: 'x' })
|
|
|
})
|
|
|
})
|