407 lines
15 KiB
TypeScript
407 lines
15 KiB
TypeScript
/**
|
|
* Command parser and one-turn driver for `dsh-cli-demo`. The executable wrapper
|
|
* owns process signals; this module owns output, durability, and cleanup.
|
|
* @module @deepseek-ai/dsh-cli-demo/cli
|
|
*/
|
|
|
|
import { parseArgs } from 'node:util'
|
|
import type { Context } from 'cordis'
|
|
import type { Agent } from '@deepseek-ai/dsh-agent'
|
|
import { createUserMessage, type TokenUsage } from '@deepseek-ai/dsh-llm'
|
|
import type { SessionEvent } from '@deepseek-ai/dsh-session'
|
|
import { boot, loadEnv, resolveConfigPath } from '@deepseek-ai/dsh-app-boot'
|
|
|
|
const CLI_NAME = 'dsh-cli-demo'
|
|
const DEFAULT_CONFIG_PATH = './cordis.yml'
|
|
const OUTPUT_FORMATS = ['text', 'json', 'stream-json'] as const
|
|
const USAGE = `Usage: ${CLI_NAME} [--config path] [--output-format text|json|stream-json] (-p <task> | <task>)\n`
|
|
|
|
/** Supported CLI output encodings. */
|
|
export type OutputFormat = typeof OUTPUT_FORMATS[number]
|
|
|
|
/** Parsed command: help exits before boot; run carries one validated task. */
|
|
export type CliCommand =
|
|
| { readonly kind: 'help' }
|
|
| {
|
|
readonly kind: 'run'
|
|
readonly configPath: string
|
|
readonly outputFormat: OutputFormat
|
|
readonly task: string
|
|
}
|
|
|
|
/** DSH-native final record emitted by JSON modes. */
|
|
export interface CliResult {
|
|
readonly type: 'result'
|
|
readonly sessionId: string
|
|
readonly output: string
|
|
readonly usage?: TokenUsage
|
|
}
|
|
|
|
/** Options for one turn against the configured top-level agent. */
|
|
export interface OneShotOptions {
|
|
/** Exactly one nonblank user task. */
|
|
readonly task: string
|
|
/** Optional signal that cancels the selected agent. */
|
|
readonly signal?: AbortSignal
|
|
/** Synchronous task-turn observer; a throw cancels the agent and fails the run after flush. */
|
|
readonly onEvent?: (sessionId: string, event: SessionEvent) => void
|
|
}
|
|
|
|
/** Injectable process boundaries used by {@link executeCli}. */
|
|
export interface CliRuntime {
|
|
/** Process cwd for config resolution and `.env` loading. */
|
|
readonly cwd?: string
|
|
/** Cancellation signal, normally aborted by SIGINT or SIGTERM. */
|
|
readonly signal?: AbortSignal
|
|
/** Loader boot boundary. */
|
|
readonly boot?: (name: string, absoluteConfigPath: string) => Promise<Context>
|
|
/** Optional `.env` loader boundary. */
|
|
readonly loadEnv?: (name: string, dir: string, warn: (line: string) => void) => void
|
|
/** Stdout sink; throws are treated as output failures. */
|
|
readonly writeStdout?: (chunk: string) => unknown
|
|
/** Stderr diagnostic sink. */
|
|
readonly writeStderr?: (chunk: string) => unknown
|
|
/** Context disposal boundary. */
|
|
readonly dispose?: (ctx: Context) => Promise<void>
|
|
}
|
|
|
|
interface ParsedArguments {
|
|
readonly values: {
|
|
readonly config?: string
|
|
readonly 'output-format'?: string
|
|
readonly help?: boolean
|
|
readonly prompt?: string
|
|
}
|
|
readonly positionals: string[]
|
|
}
|
|
|
|
class CliArgumentError extends Error {
|
|
constructor(message: string) {
|
|
super(message)
|
|
this.name = 'CliArgumentError'
|
|
}
|
|
}
|
|
|
|
class CliInterruptedError extends Error {
|
|
constructor(reason: string) {
|
|
super(reason)
|
|
this.name = 'CliInterruptedError'
|
|
}
|
|
}
|
|
|
|
/** Render an arbitrary value without trusting its type traps or string coercion. */
|
|
function renderUnknown(value: unknown): string {
|
|
try {
|
|
return String(value)
|
|
} catch {
|
|
return '[unrenderable thrown value]'
|
|
}
|
|
}
|
|
|
|
/** Normalize an arbitrary thrown value without letting inspection escape containment. */
|
|
function toError(error: unknown): Error {
|
|
try {
|
|
if (error instanceof Error) return error
|
|
} catch {
|
|
// A hostile proxy may throw during instanceof; use the total renderer below.
|
|
}
|
|
return new Error(renderUnknown(error))
|
|
}
|
|
|
|
function interruptionReason(signal: AbortSignal): string {
|
|
return signal.reason === undefined ? 'interrupted' : renderUnknown(signal.reason)
|
|
}
|
|
|
|
/**
|
|
* Parse the bin arguments and enforce the one-positional-task contract.
|
|
* @param args - arguments after the executable name.
|
|
* @returns a help or run command.
|
|
* @throws {@link CliArgumentError} for unknown flags, invalid formats, or task cardinality.
|
|
*/
|
|
export function parseCliArgs(args: readonly string[]): CliCommand {
|
|
let parsed: ParsedArguments
|
|
try {
|
|
parsed = parseArgs({
|
|
args: [...args],
|
|
options: {
|
|
config: { type: 'string' },
|
|
'output-format': { type: 'string' },
|
|
help: { type: 'boolean' },
|
|
prompt: { type: 'string', short: 'p' },
|
|
},
|
|
allowPositionals: true,
|
|
strict: true,
|
|
})
|
|
} catch (error: unknown) {
|
|
throw new CliArgumentError(toError(error).message)
|
|
}
|
|
|
|
if (parsed.values.help === true) return { kind: 'help' }
|
|
const prompt = parsed.values.prompt
|
|
if (prompt !== undefined && parsed.positionals.length > 0) {
|
|
throw new CliArgumentError('-p/--prompt and a positional task are mutually exclusive')
|
|
}
|
|
if (prompt === undefined && parsed.positionals.length !== 1) {
|
|
throw new CliArgumentError(`expected exactly one positional task or -p, received ${parsed.positionals.length} positional(s)`)
|
|
}
|
|
// Cardinality was checked above, so the fallback index zero exists.
|
|
// oxlint-disable-next-line typescript/no-non-null-assertion
|
|
const task = prompt ?? parsed.positionals[0]!
|
|
if (task.trim().length === 0) throw new CliArgumentError('task must not be blank')
|
|
|
|
const requestedFormat = parsed.values['output-format'] ?? 'text'
|
|
if (!OUTPUT_FORMATS.some(format => format === requestedFormat)) {
|
|
throw new CliArgumentError(`unsupported output format ${JSON.stringify(requestedFormat)}`)
|
|
}
|
|
return {
|
|
kind: 'run',
|
|
configPath: parsed.values.config ?? DEFAULT_CONFIG_PATH,
|
|
outputFormat: requestedFormat as OutputFormat,
|
|
task,
|
|
}
|
|
}
|
|
|
|
function addUsage(total: TokenUsage | undefined, step: TokenUsage): TokenUsage {
|
|
const next: TokenUsage = {
|
|
inputTokens: (total?.inputTokens ?? 0) + step.inputTokens,
|
|
outputTokens: (total?.outputTokens ?? 0) + step.outputTokens,
|
|
}
|
|
for (const key of ['cacheReadTokens', 'cacheWriteTokens', 'reasoningTokens'] as const) {
|
|
if (total?.[key] !== undefined || step[key] !== undefined) next[key] = (total?.[key] ?? 0) + (step[key] ?? 0)
|
|
}
|
|
return next
|
|
}
|
|
|
|
function assistantText(event: Extract<SessionEvent, { type: 'assistant/message' }>): string | undefined {
|
|
const blocks = event.data.message.content.filter(block => block.type === 'text')
|
|
return blocks.length === 0 ? undefined : blocks.map(block => block.text).join('')
|
|
}
|
|
|
|
/** Wait for startup quiescence while making pre-run cancellation terminal. */
|
|
async function waitForStartupIdle(agent: Agent, signal?: AbortSignal): Promise<void> {
|
|
if (signal === undefined) {
|
|
await agent.whenIdle()
|
|
return
|
|
}
|
|
if (signal.aborted) {
|
|
agent.cancel({ kind: 'user' })
|
|
throw new CliInterruptedError(interruptionReason(signal))
|
|
}
|
|
await new Promise<void>((resolve, reject) => {
|
|
const onAbort = (): void => {
|
|
agent.cancel({ kind: 'user' })
|
|
reject(new CliInterruptedError(interruptionReason(signal)))
|
|
}
|
|
signal.addEventListener('abort', onAbort, { once: true })
|
|
void agent.whenIdle().then(resolve, reject).finally(() => {
|
|
signal.removeEventListener('abort', onAbort)
|
|
})
|
|
})
|
|
}
|
|
|
|
/**
|
|
* Run one owned activity interval on the configured top-level agent, from the
|
|
* task's durable enqueue receipt through whole-agent idle.
|
|
* @param ctx - settled Loader root containing one agent plus `ctx.sessions`.
|
|
* @param options - task, optional cancellation, and optional stream observer.
|
|
* @returns the DSH-native result envelope after durable quiescence.
|
|
*/
|
|
export async function runOneShot(ctx: Context, options: OneShotOptions): Promise<CliResult> {
|
|
const agents = ctx.get('agents')?.roots() ?? []
|
|
const [agent] = agents
|
|
if (agent === undefined || agents.length !== 1) {
|
|
throw new Error(`config must create exactly one top-level agent, found ${agents.length}`)
|
|
}
|
|
await waitForStartupIdle(agent, options.signal)
|
|
|
|
const message = createUserMessage({ content: [{ type: 'text', text: options.task }], source: { kind: 'user' } })
|
|
let received = false
|
|
let output = ''
|
|
const usageByStep = new Map<string, TokenUsage>()
|
|
let outputError: Error | undefined
|
|
let interrupted: CliInterruptedError | undefined
|
|
const observe = (sessionId: string, event: SessionEvent): void => {
|
|
if (outputError !== undefined || options.onEvent === undefined) return
|
|
try {
|
|
options.onEvent(sessionId, event)
|
|
} catch (error: unknown) {
|
|
outputError = toError(error)
|
|
queueMicrotask(() => {
|
|
agent.cancel({ kind: 'user' })
|
|
})
|
|
}
|
|
}
|
|
|
|
const disposeListener = ctx.on('session/event', (session, event) => {
|
|
if (session !== agent.session) return
|
|
if (!received) {
|
|
if (event.type !== 'agent/inbox/spliced'
|
|
|| !event.data.inserted.some(inserted => inserted.id === message.id)) return
|
|
received = true
|
|
}
|
|
observe(session.id, event)
|
|
if (event.type === 'assistant/chunk' && event.data.chunk.type === 'usage') {
|
|
usageByStep.set(`${event.data.turn}/${event.data.step}`, event.data.chunk.usage)
|
|
}
|
|
if (event.type === 'assistant/message') {
|
|
output = assistantText(event) ?? output
|
|
if (event.data.usage !== undefined) {
|
|
usageByStep.set(`${event.data.turn}/${event.data.step}`, event.data.usage)
|
|
}
|
|
}
|
|
})
|
|
|
|
const signal = options.signal
|
|
let onAbort: (() => void) | undefined
|
|
if (signal !== undefined) {
|
|
onAbort = (): void => {
|
|
interrupted ??= new CliInterruptedError(interruptionReason(signal))
|
|
agent.cancel({ kind: 'user' })
|
|
}
|
|
signal.addEventListener('abort', onAbort, { once: true })
|
|
/* v8 ignore next -- closes the race between startup-idle completion and listener registration */
|
|
if (signal.aborted) onAbort()
|
|
}
|
|
|
|
try {
|
|
if (interrupted === undefined) agent.followup(message)
|
|
await agent.whenIdle()
|
|
} finally {
|
|
if (onAbort !== undefined) signal?.removeEventListener('abort', onAbort)
|
|
disposeListener()
|
|
}
|
|
|
|
await ctx.sessions.flush(agent.session)
|
|
if (outputError !== undefined) throw outputError
|
|
if (interrupted !== undefined) throw interrupted
|
|
const usage = [...usageByStep.values()].reduce<TokenUsage | undefined>(addUsage, undefined)
|
|
return {
|
|
type: 'result',
|
|
sessionId: agent.session.id,
|
|
output,
|
|
...usage === undefined ? {} : { usage },
|
|
}
|
|
}
|
|
|
|
function renderResult(outputFormat: OutputFormat, result: CliResult): string {
|
|
return outputFormat === 'text' ? `${result.output}\n` : `${JSON.stringify(result)}\n`
|
|
}
|
|
|
|
/**
|
|
* Race Loader boot with cancellation without abandoning a context that becomes
|
|
* available after the caller has been released. Waiting for that late context
|
|
* would recreate the signal hang, so its disposal and diagnostics run detached.
|
|
*/
|
|
async function bootInterruptibly(
|
|
start: () => Promise<Context>,
|
|
signal: AbortSignal | undefined,
|
|
disposeLateContext: (ctx: Context) => Promise<void>,
|
|
reportLateDisposalFailure: (error: unknown) => void,
|
|
): Promise<Context> {
|
|
if (signal === undefined) return await start()
|
|
if (signal.aborted) throw new CliInterruptedError(interruptionReason(signal))
|
|
|
|
let onAbort!: () => void
|
|
const interruptedBoot = new Promise<never>((_resolve, reject) => {
|
|
onAbort = (): void => {
|
|
reject(new CliInterruptedError(interruptionReason(signal)))
|
|
}
|
|
signal.addEventListener('abort', onAbort, { once: true })
|
|
/* v8 ignore next -- closes registration against a non-standard synchronously mutating signal */
|
|
if (signal.aborted) onAbort()
|
|
})
|
|
const booting = Promise.resolve().then(start)
|
|
try {
|
|
return await Promise.race([booting, interruptedBoot])
|
|
} catch (error: unknown) {
|
|
// The awaited race permits the signal to change after the preflight check.
|
|
// oxlint-disable-next-line typescript/no-unnecessary-condition
|
|
if (signal.aborted) {
|
|
void booting.then(
|
|
async (lateContext) => {
|
|
try {
|
|
await disposeLateContext(lateContext)
|
|
} catch (error: unknown) {
|
|
reportLateDisposalFailure(error)
|
|
}
|
|
},
|
|
() => {},
|
|
)
|
|
}
|
|
throw error
|
|
} finally {
|
|
signal.removeEventListener('abort', onAbort)
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Execute one CLI invocation. Argument and boot failures never write stdout;
|
|
* context disposal is awaited before return, and its failure does not replace
|
|
* an earlier diagnostic.
|
|
* @param args - arguments after the executable name.
|
|
* @param runtime - optional injected process boundaries for tests and embedding.
|
|
* @returns the ordinary process exit code; the thin bin overrides it for Unix signals.
|
|
*/
|
|
export async function executeCli(args: readonly string[], runtime: CliRuntime = {}): Promise<number> {
|
|
/* v8 ignore next -- default process sinks are exercised by the built-bin smoke */
|
|
const writeStdout = runtime.writeStdout ?? (chunk => process.stdout.write(chunk))
|
|
/* v8 ignore next -- default process sinks are exercised by the built-bin smoke */
|
|
const writeStderr = runtime.writeStderr ?? (chunk => process.stderr.write(chunk))
|
|
let command: CliCommand
|
|
try {
|
|
command = parseCliArgs(args)
|
|
} catch (error: unknown) {
|
|
writeStderr(`${CLI_NAME}: ${toError(error).message}\n${USAGE}`)
|
|
return 1
|
|
}
|
|
if (command.kind === 'help') {
|
|
writeStdout(USAGE)
|
|
return 0
|
|
}
|
|
|
|
/* v8 ignore next -- default process cwd is exercised by the built-bin smoke */
|
|
const cwd = runtime.cwd ?? process.cwd()
|
|
/* v8 ignore next -- default env/boot boundaries are exercised by the Loader and built-bin smokes */
|
|
const loadEnvironment = runtime.loadEnv ?? loadEnv
|
|
/* v8 ignore next -- default env/boot boundaries are exercised by the Loader and built-bin smokes */
|
|
const bootContext = runtime.boot ?? boot
|
|
/* v8 ignore next -- default disposal is exercised by the built-bin smoke */
|
|
const disposeContext = runtime.dispose ?? (target => target.fiber.dispose())
|
|
let ctx: Context | undefined
|
|
let exitCode = 1
|
|
let diagnostic: string | undefined
|
|
try {
|
|
loadEnvironment(CLI_NAME, cwd, line => writeStderr(line))
|
|
ctx = await bootInterruptibly(
|
|
() => bootContext(CLI_NAME, resolveConfigPath(command.configPath, undefined, cwd)),
|
|
runtime.signal,
|
|
disposeContext,
|
|
error => writeStderr(`${CLI_NAME}: dispose after interrupted boot failed: ${toError(error).message}\n`),
|
|
)
|
|
const result = await runOneShot(ctx, {
|
|
task: command.task,
|
|
...runtime.signal === undefined ? {} : { signal: runtime.signal },
|
|
...command.outputFormat === 'stream-json'
|
|
? { onEvent: (sessionId: string, event: SessionEvent) => {
|
|
writeStdout(`${JSON.stringify({ type: 'session_event', sessionId, event })}\n`)
|
|
} }
|
|
: {},
|
|
})
|
|
writeStdout(renderResult(command.outputFormat, result))
|
|
exitCode = 0
|
|
} catch (error: unknown) {
|
|
diagnostic = `${CLI_NAME}: ${toError(error).message}\n`
|
|
} finally {
|
|
if (ctx !== undefined) {
|
|
try {
|
|
await disposeContext(ctx)
|
|
} catch (error: unknown) {
|
|
diagnostic = `${diagnostic ?? ''}${CLI_NAME}: dispose failed: ${toError(error).message}\n`
|
|
exitCode = 1
|
|
}
|
|
}
|
|
}
|
|
if (diagnostic !== undefined) writeStderr(diagnostic)
|
|
return exitCode
|
|
}
|