mirror of
https://github.com/deepseek-ai/deepseek-harness
synced 2026-08-15 21:04:50 +00:00
306dd2b1fe
Contain arbitrary plugin and runtime failures even when a thrown Proxy traps instanceof checks or its string coercion throws. Fall back to a stable diagnostic instead of letting executeCli reject outside its exit-code contract. Route abort reasons through the same total renderer so cancellation cannot escape containment through an exotic reason value. Add a focused regression that exercises both hostile inspection paths and verifies stdout remains empty, stderr remains labelled, and the CLI resolves with exit code 1.
475 lines
20 KiB
TypeScript
475 lines
20 KiB
TypeScript
import { readdir, mkdtemp } from 'node:fs/promises'
|
|
import { tmpdir } from 'node:os'
|
|
import { join, resolve } from 'node:path'
|
|
import { Context } from 'cordis'
|
|
import type { Agent } from '@deepseek-ai/dsh-agent'
|
|
import { CallId, LlmAdapter, type GenerateOptions, type StreamChunk, type TokenUsage } from '@deepseek-ai/dsh-llm'
|
|
import { SessionId, type SessionEvent, type TurnEndReason } from '@deepseek-ai/dsh-session'
|
|
import { afterEach, describe, expect, it } from 'vitest'
|
|
import * as cliDemo from '../src/index.ts'
|
|
import {
|
|
executeCli,
|
|
formatTurnFailure,
|
|
parseCliArgs,
|
|
runOneShot,
|
|
type CliResult,
|
|
} from '../src/cli.ts'
|
|
|
|
type ScriptEntry = readonly StreamChunk[] | 'hang'
|
|
|
|
class ScriptedAdapter extends LlmAdapter {
|
|
readonly requests: GenerateOptions[] = []
|
|
private cursor = 0
|
|
|
|
constructor(private readonly script: readonly ScriptEntry[]) {
|
|
super()
|
|
}
|
|
|
|
async * stream(options: GenerateOptions): AsyncIterable<StreamChunk> {
|
|
this.requests.push(options)
|
|
const entry = this.script[this.cursor++]
|
|
if (entry === undefined) throw new Error('script exhausted')
|
|
if (entry === 'hang') {
|
|
yield { type: 'block-start', index: 0, blockType: 'text' }
|
|
yield { type: 'text-delta', index: 0, text: 'partial' }
|
|
await new Promise<void>((_resolve, reject) => {
|
|
if (options.signal?.aborted === true) {
|
|
reject(new Error('aborted'))
|
|
return
|
|
}
|
|
options.signal?.addEventListener('abort', () => { reject(new Error('aborted')) }, { once: true })
|
|
})
|
|
return
|
|
}
|
|
for (const chunk of entry) yield chunk
|
|
}
|
|
}
|
|
|
|
function textResponse(text: string, usage?: TokenUsage, finish: 'stop' | 'max-tokens' = 'stop'): StreamChunk[] {
|
|
return [
|
|
{ type: 'block-start', index: 0, blockType: 'text' },
|
|
{ type: 'text-delta', index: 0, text },
|
|
{ type: 'block-end', index: 0, block: { type: 'text', text } },
|
|
...usage === undefined ? [] : [{ type: 'usage', usage } as const],
|
|
{ type: 'finish', reason: { kind: finish } },
|
|
]
|
|
}
|
|
|
|
function toolResponse(usage: TokenUsage): StreamChunk[] {
|
|
const id = CallId('cli-call')
|
|
const args = JSON.stringify({ text: 'round trip' })
|
|
return [
|
|
{ type: 'block-start', index: 0, blockType: 'text' },
|
|
{ type: 'text-delta', index: 0, text: 'working' },
|
|
{ type: 'block-end', index: 0, block: { type: 'text', text: 'working' } },
|
|
{ type: 'block-start', index: 1, blockType: 'tool-call' },
|
|
{ type: 'tool-call-delta', index: 1, id, name: 'echo', argumentsDelta: args },
|
|
{ type: 'block-end', index: 1, block: { type: 'tool-call', id, name: 'echo', arguments: args } },
|
|
{ type: 'usage', usage },
|
|
{ type: 'finish', reason: { kind: 'tool-calls' } },
|
|
]
|
|
}
|
|
|
|
function reasoningResponse(text: string): StreamChunk[] {
|
|
return [
|
|
{ type: 'block-start', index: 0, blockType: 'reasoning' },
|
|
{ type: 'reasoning-delta', index: 0, text },
|
|
{ type: 'block-end', index: 0, block: { type: 'reasoning', text } },
|
|
{ type: 'finish', reason: { kind: 'stop' } },
|
|
]
|
|
}
|
|
|
|
interface Harness {
|
|
readonly ctx: Context
|
|
readonly agent: Agent
|
|
readonly persistenceRoot: string
|
|
}
|
|
|
|
const liveContexts: Context[] = []
|
|
|
|
async function harness(script: readonly ScriptEntry[]): Promise<Harness> {
|
|
const root = await mkdtemp(join(tmpdir(), 'dsh-cli-runner-'))
|
|
const skillHome = await mkdtemp(join(tmpdir(), 'dsh-cli-runner-skills-'))
|
|
const ctx = new Context()
|
|
liveContexts.push(ctx)
|
|
await ctx.plugin(cliDemo, {
|
|
provider: 'mock',
|
|
model: 'mock',
|
|
persistenceRoot: root,
|
|
skills: { local: { dshHome: join(skillHome, '.dsh'), agentsHome: join(skillHome, '.agents') } },
|
|
workspaceContext: false,
|
|
})
|
|
await new Promise(resolve => setTimeout(resolve, 80))
|
|
ctx.llm.registerAdapter(['mock'], new ScriptedAdapter(script))
|
|
ctx.tools.register({
|
|
name: 'echo',
|
|
description: 'Echo text.',
|
|
parameters: { text: { type: 'string', required: true } },
|
|
execute: async args => [{ type: 'text', text: `ECHO: ${(args as { text: string }).text}` }],
|
|
})
|
|
const [agent] = ctx.agents.roots()
|
|
if (agent === undefined) throw new Error('test main agent missing')
|
|
return { ctx, agent, persistenceRoot: root }
|
|
}
|
|
|
|
async function invoke(
|
|
ctx: Context,
|
|
args: readonly string[],
|
|
options: { signal?: AbortSignal; failStdout?: boolean; failDispose?: boolean } = {},
|
|
): Promise<{ code: number; stdout: string; stderr: string }> {
|
|
let stdout = ''
|
|
let stderr = ''
|
|
const code = await executeCli(args, {
|
|
cwd: '/tmp/cli-cwd',
|
|
...options.signal === undefined ? {} : { signal: options.signal },
|
|
boot: async () => ctx,
|
|
loadEnv: () => {},
|
|
writeStdout: (chunk) => {
|
|
if (options.failStdout === true) throw new Error('stdout closed')
|
|
stdout += chunk
|
|
},
|
|
writeStderr: (chunk) => { stderr += chunk },
|
|
...options.failDispose === true
|
|
? { dispose: async (target: Context) => {
|
|
await target.fiber.dispose()
|
|
throw new Error('dispose exploded')
|
|
} }
|
|
: {},
|
|
})
|
|
return { code, stdout, stderr }
|
|
}
|
|
|
|
afterEach(async () => {
|
|
await Promise.all(liveContexts.splice(0).map(ctx => ctx.fiber.dispose()))
|
|
})
|
|
|
|
describe('parseCliArgs', () => {
|
|
it('parses defaults, explicit options, spaces, and an option-like task after --', () => {
|
|
expect(parseCliArgs(['task with spaces'])).toEqual({
|
|
kind: 'run', configPath: './cordis.yml', outputFormat: 'text', task: 'task with spaces',
|
|
})
|
|
expect(parseCliArgs(['--config', 'custom.yml', '--output-format', 'stream-json', 'do it'])).toEqual({
|
|
kind: 'run', configPath: 'custom.yml', outputFormat: 'stream-json', task: 'do it',
|
|
})
|
|
expect(parseCliArgs(['--', '-task'])).toMatchObject({ task: '-task' })
|
|
expect(parseCliArgs(['--help', 'ignored'])).toEqual({ kind: 'help' })
|
|
})
|
|
|
|
it('rejects missing, blank, extra, invalid-format, and unsupported flags', () => {
|
|
expect(() => parseCliArgs([])).toThrow('received 0')
|
|
expect(() => parseCliArgs([' '])).toThrow('must not be blank')
|
|
expect(() => parseCliArgs(['one', 'two'])).toThrow('received 2')
|
|
expect(() => parseCliArgs(['--output-format', 'xml', 'task'])).toThrow('unsupported output format')
|
|
expect(() => parseCliArgs(['-p', 'task'])).toThrow('Unknown option')
|
|
})
|
|
})
|
|
|
|
describe('runOneShot and executeCli', () => {
|
|
it('prints help and argument diagnostics without booting or contaminating stdout', async () => {
|
|
let booted = false
|
|
let stdout = ''
|
|
let stderr = ''
|
|
const runtime = {
|
|
boot: async (): Promise<Context> => { booted = true; throw new Error('unexpected') },
|
|
writeStdout: (chunk: string): void => { stdout += chunk },
|
|
writeStderr: (chunk: string): void => { stderr += chunk },
|
|
}
|
|
expect(await executeCli(['--help'], runtime)).toBe(0)
|
|
expect(stdout).toContain('Usage: dsh-cli-demo')
|
|
stdout = ''
|
|
expect(await executeCli([], runtime)).toBe(1)
|
|
expect(stdout).toBe('')
|
|
expect(stderr).toContain('received 0')
|
|
expect(booted).toBe(false)
|
|
})
|
|
|
|
it('leaves stdout empty for environment and boot failures and resolves the default config', async () => {
|
|
let bootPath = ''
|
|
let stderr = ''
|
|
const code = await executeCli(['task'], {
|
|
cwd: '/tmp/cli-work',
|
|
loadEnv: (_name, _dir, warn) => { warn('env warning\n') },
|
|
boot: async (_name, path) => { bootPath = path; throw 'boot exploded' },
|
|
writeStdout: () => { throw new Error('stdout must stay empty') },
|
|
writeStderr: (chunk) => { stderr += chunk },
|
|
})
|
|
expect(code).toBe(1)
|
|
expect(bootPath).toBe(resolve('/tmp/cli-work/cordis.yml'))
|
|
expect(stderr).toContain('env warning')
|
|
expect(stderr).toContain('boot exploded')
|
|
})
|
|
|
|
it('contains a thrown value whose inspection and coercion both fail', async () => {
|
|
const hostile = new Proxy({}, {
|
|
getPrototypeOf: () => { throw new Error('prototype trap escaped') },
|
|
get: (target, key, receiver) => {
|
|
if (key === Symbol.toPrimitive) throw new Error('coercion escaped')
|
|
return Reflect.get(target, key, receiver) as unknown
|
|
},
|
|
})
|
|
let stdout = ''
|
|
let stderr = ''
|
|
const code = await executeCli(['task'], {
|
|
boot: async () => { throw hostile },
|
|
loadEnv: () => {},
|
|
writeStdout: (chunk) => { stdout += chunk },
|
|
writeStderr: (chunk) => { stderr += chunk },
|
|
})
|
|
expect(code).toBe(1)
|
|
expect(stdout).toBe('')
|
|
expect(stderr).toBe('dsh-cli-demo: [unrenderable thrown value]\n')
|
|
})
|
|
|
|
it('interrupts Loader boot and contains every late boot outcome', async () => {
|
|
const abort = new AbortController()
|
|
const lateContext = new Context()
|
|
liveContexts.push(lateContext)
|
|
const boot = Promise.withResolvers<Context>()
|
|
const disposed = Promise.withResolvers<undefined>()
|
|
let disposeCalls = 0
|
|
let stderr = ''
|
|
const running = executeCli(['task'], {
|
|
signal: abort.signal,
|
|
boot: () => boot.promise,
|
|
loadEnv: () => {},
|
|
writeStdout: () => {},
|
|
writeStderr: (chunk) => { stderr += chunk },
|
|
dispose: async (ctx) => {
|
|
disposeCalls += 1
|
|
await ctx.fiber.dispose()
|
|
disposed.resolve(undefined)
|
|
},
|
|
})
|
|
abort.abort('received SIGTERM')
|
|
await expect(running).resolves.toBe(1)
|
|
expect(stderr).toContain('received SIGTERM')
|
|
expect(disposeCalls).toBe(0)
|
|
boot.resolve(lateContext)
|
|
await disposed.promise
|
|
expect(disposeCalls).toBe(1)
|
|
|
|
const rejectedBoot = Promise.withResolvers<Context>()
|
|
const rejectedAbort = new AbortController()
|
|
const rejected = executeCli(['task'], {
|
|
signal: rejectedAbort.signal,
|
|
boot: () => rejectedBoot.promise,
|
|
loadEnv: () => {},
|
|
writeStdout: () => {},
|
|
writeStderr: () => {},
|
|
})
|
|
rejectedAbort.abort('stop rejected boot')
|
|
await expect(rejected).resolves.toBe(1)
|
|
rejectedBoot.reject(new Error('late boot rejection'))
|
|
await Promise.resolve()
|
|
|
|
let ordinaryBootStderr = ''
|
|
const ordinaryBootFailure = await executeCli(['task'], {
|
|
signal: new AbortController().signal,
|
|
boot: async () => { throw new Error('ordinary boot failure') },
|
|
loadEnv: () => {},
|
|
writeStdout: () => {},
|
|
writeStderr: (chunk) => { ordinaryBootStderr += chunk },
|
|
})
|
|
expect(ordinaryBootFailure).toBe(1)
|
|
expect(ordinaryBootStderr).toContain('ordinary boot failure')
|
|
|
|
const failedCleanupBoot = Promise.withResolvers<Context>()
|
|
const failedCleanupAbort = new AbortController()
|
|
const cleanupFailure = Promise.withResolvers<undefined>()
|
|
const failedCleanupContext = new Context()
|
|
liveContexts.push(failedCleanupContext)
|
|
const failedCleanup = executeCli(['task'], {
|
|
signal: failedCleanupAbort.signal,
|
|
boot: () => failedCleanupBoot.promise,
|
|
loadEnv: () => {},
|
|
writeStdout: () => {},
|
|
writeStderr: (chunk) => {
|
|
if (chunk.includes('dispose after interrupted boot failed: late cleanup')) cleanupFailure.resolve(undefined)
|
|
},
|
|
dispose: async (ctx) => {
|
|
await ctx.fiber.dispose()
|
|
throw new Error('late cleanup')
|
|
},
|
|
})
|
|
failedCleanupAbort.abort('stop failed cleanup boot')
|
|
await expect(failedCleanup).resolves.toBe(1)
|
|
failedCleanupBoot.resolve(failedCleanupContext)
|
|
await cleanupFailure.promise
|
|
})
|
|
|
|
it('renders text, flushes a persisted fresh session, and disposes the context', async () => {
|
|
const { ctx, agent, persistenceRoot } = await harness([textResponse('final answer')])
|
|
const output = await invoke(ctx, ['task'])
|
|
expect(output).toEqual({ code: 0, stdout: 'final answer\n', stderr: '' })
|
|
expect(agent.status).toBe('disposed')
|
|
const files = await readdir(persistenceRoot, { recursive: true })
|
|
expect(files.some(file => file.endsWith('.jsonl'))).toBe(true)
|
|
})
|
|
|
|
it('sums usage across tool steps and selects the last text-bearing assistant message', async () => {
|
|
const first = { inputTokens: 10, outputTokens: 3, cacheReadTokens: 2, cacheWriteTokens: 1 }
|
|
const second = { inputTokens: 7, outputTokens: 5, cacheReadTokens: 4, reasoningTokens: 6 }
|
|
const { ctx } = await harness([toolResponse(first), textResponse('done', second)])
|
|
const output = await invoke(ctx, ['--output-format', 'json', 'task'])
|
|
const result = JSON.parse(output.stdout) as CliResult
|
|
expect(output.code).toBe(0)
|
|
expect(result).toMatchObject({ type: 'result', success: true, turn: 1, result: 'done', reason: { kind: 'completed' } })
|
|
expect(result.usage).toEqual({
|
|
inputTokens: 17,
|
|
outputTokens: 8,
|
|
cacheReadTokens: 6,
|
|
cacheWriteTokens: 1,
|
|
reasoningTokens: 6,
|
|
})
|
|
})
|
|
|
|
it('keeps the prior text when a later assistant message has no text blocks', async () => {
|
|
const { ctx } = await harness([
|
|
toolResponse({ inputTokens: 1, outputTokens: 1 }),
|
|
reasoningResponse('reasoning only'),
|
|
])
|
|
const result = await runOneShot(ctx, { task: 'task' })
|
|
expect(result.result).toBe('working')
|
|
})
|
|
|
|
it('streams only the correlated main message turn and then the result envelope', async () => {
|
|
const { ctx, agent } = await harness([textResponse('streamed')])
|
|
const other = ctx.sessions.create(SessionId('unrelated'))
|
|
let injected = false
|
|
ctx.on('agent/queued', (subject) => {
|
|
if (subject !== agent || injected) return
|
|
injected = true
|
|
agent.inject([{ type: 'text', text: 'startup injection' }], { source: { kind: 'plugin', plugin: 'test' } })
|
|
other.append('turn/start', { turn: 1, trigger: { kind: 'injection', source: { kind: 'plugin', plugin: 'test' } } })
|
|
other.append('turn/end', { turn: 1, reason: { kind: 'completed' } })
|
|
})
|
|
const output = await invoke(ctx, ['--output-format', 'stream-json', 'task'])
|
|
const lines = output.stdout.trimEnd().split('\n').map(line => JSON.parse(line) as Record<string, unknown>)
|
|
const events = lines.slice(0, -1).map(line => line['event'] as SessionEvent)
|
|
expect(lines.at(-1)).toMatchObject({ type: 'result', success: true, turn: 2, result: 'streamed' })
|
|
expect(events[0]).toMatchObject({ type: 'turn/start', data: { turn: 2, trigger: { kind: 'message' } } })
|
|
expect(events.at(-1)).toMatchObject({ type: 'turn/end', data: { turn: 2 } })
|
|
expect(lines.slice(0, -1).every(line => line['sessionId'] === agent.session.id)).toBe(true)
|
|
expect(events.some(event => event.type === 'context/message')).toBe(false)
|
|
})
|
|
|
|
it('emits partial data and a diagnostic for non-completed turns', async () => {
|
|
const { ctx } = await harness([textResponse('partial', { inputTokens: 2, outputTokens: 3 }, 'max-tokens')])
|
|
const output = await invoke(ctx, ['--output-format', 'json', 'task'])
|
|
expect(JSON.parse(output.stdout)).toMatchObject({ success: false, result: 'partial', reason: { kind: 'max-tokens' } })
|
|
expect(output.code).toBe(1)
|
|
expect(output.stderr).toContain('output-token limit')
|
|
})
|
|
|
|
it('cancels an active turn, emits its durable aborted result, and disposes', async () => {
|
|
const { ctx, agent } = await harness(['hang'])
|
|
const abort = new AbortController()
|
|
let started!: () => void
|
|
const running = new Promise<void>((resolveStarted) => { started = resolveStarted })
|
|
ctx.on('session/event', (session, event) => {
|
|
if (session === agent.session && event.type === 'assistant/chunk') started()
|
|
})
|
|
const outcome = invoke(ctx, ['--output-format', 'json', 'task'], { signal: abort.signal })
|
|
await running
|
|
abort.abort('received SIGINT')
|
|
const output = await outcome
|
|
expect(JSON.parse(output.stdout)).toMatchObject({ success: false, reason: { kind: 'aborted', reason: 'received SIGINT' } })
|
|
expect(output.code).toBe(1)
|
|
expect(output.stderr).toContain('was aborted: received SIGINT')
|
|
expect(agent.status).toBe('disposed')
|
|
})
|
|
|
|
it('contains stream-writer failures, cancels, flushes, and returns the output error', async () => {
|
|
const { ctx, agent } = await harness(['hang'])
|
|
await expect(runOneShot(ctx, {
|
|
task: 'task',
|
|
onEvent: () => { throw new Error('stream sink failed') },
|
|
})).rejects.toThrow('stream sink failed')
|
|
expect(agent.status).toBe('idle')
|
|
})
|
|
|
|
it('handles cancellation before submission, a missing main agent, and final-output failure', async () => {
|
|
const early = await harness([textResponse('unused')])
|
|
const fakeSignal = {
|
|
aborted: true,
|
|
reason: undefined,
|
|
} as unknown as AbortSignal
|
|
await expect(runOneShot(early.ctx, { task: 'task', signal: fakeSignal })).rejects.toThrow('interrupted')
|
|
|
|
const preBootAbort = new AbortController()
|
|
preBootAbort.abort('before boot completed')
|
|
const preBoot = await invoke(early.ctx, ['task'], { signal: preBootAbort.signal })
|
|
expect(preBoot).toMatchObject({ code: 1, stdout: '' })
|
|
expect(preBoot.stderr).toContain('before boot completed')
|
|
|
|
const empty = new Context()
|
|
liveContexts.push(empty)
|
|
await expect(runOneShot(empty, { task: 'task' })).rejects.toThrow('exactly one top-level agent')
|
|
|
|
const final = await harness([textResponse('answer')])
|
|
const output = await invoke(final.ctx, ['task'], { failStdout: true })
|
|
expect(output.code).toBe(1)
|
|
expect(output.stdout).toBe('')
|
|
expect(output.stderr).toContain('stdout closed')
|
|
expect(final.agent.status).toBe('disposed')
|
|
|
|
const disposal = await harness([textResponse('answer')])
|
|
const disposalOutput = await invoke(disposal.ctx, ['task'], { failDispose: true })
|
|
expect(disposalOutput).toMatchObject({ code: 1, stdout: 'answer\n' })
|
|
expect(disposalOutput.stderr).toContain('dispose exploded')
|
|
})
|
|
|
|
it('reports disposal failure alongside an earlier run failure', async () => {
|
|
const ctx = new Context()
|
|
liveContexts.push(ctx)
|
|
const output = await invoke(ctx, ['task'], { failDispose: true })
|
|
expect(output).toEqual({
|
|
code: 1,
|
|
stdout: '',
|
|
stderr: 'dsh-cli-demo: config must create exactly one top-level agent, found 0\n'
|
|
+ 'dsh-cli-demo: dispose failed: dispose exploded\n',
|
|
})
|
|
})
|
|
|
|
it('cancels startup work and queued work before the correlated turn begins', async () => {
|
|
const startup = await harness(['hang'])
|
|
let started!: () => void
|
|
const running = new Promise<void>((resolveStarted) => { started = resolveStarted })
|
|
startup.ctx.on('session/event', (session, event) => {
|
|
if (session === startup.agent.session && event.type === 'assistant/chunk') started()
|
|
})
|
|
startup.agent.send([{ type: 'text', text: 'first' }])
|
|
await running
|
|
const startupAbort = new AbortController()
|
|
const waiting = runOneShot(startup.ctx, { task: 'second', signal: startupAbort.signal })
|
|
startupAbort.abort('cancel startup')
|
|
await expect(waiting).rejects.toThrow('cancel startup')
|
|
await startup.agent.whenIdle()
|
|
|
|
const queued = await harness([textResponse('unused')])
|
|
const queuedAbort = new AbortController()
|
|
queued.ctx.on('agent/queued', (agent) => {
|
|
if (agent === queued.agent) queuedAbort.abort('cancel queued')
|
|
})
|
|
await expect(runOneShot(queued.ctx, { task: 'task', signal: queuedAbort.signal })).rejects.toThrow('cancel queued')
|
|
await queued.agent.whenIdle()
|
|
})
|
|
})
|
|
|
|
describe('formatTurnFailure', () => {
|
|
it('diagnoses every durable reason and preserves merge-extensible unknowns', () => {
|
|
const cases: [TurnEndReason, string][] = [
|
|
[{ kind: 'completed' }, 'completed'],
|
|
[{ kind: 'aborted' }, 'was aborted'],
|
|
[{ kind: 'aborted', reason: 'stop' }, 'was aborted: stop'],
|
|
[{ kind: 'error', step: 2, message: 'bad' }, 'failed at step 2: bad'],
|
|
[{ kind: 'disposed' }, 'was disposed'],
|
|
[{ kind: 'max-tokens' }, 'output-token limit'],
|
|
[{ kind: 'rejected', reason: 'policy' }, 'was rejected: policy'],
|
|
[{ kind: 'interrupted' }, 'persistence recovery'],
|
|
]
|
|
for (const [reason, expected] of cases) expect(formatTurnFailure(reason)).toContain(expected)
|
|
expect(formatTurnFailure({ kind: 'extension' } as unknown as TurnEndReason)).toContain('extension')
|
|
})
|
|
})
|