diff --git a/src/agent/claude-code/cli-events.ts b/src/agent/claude-code/cli-events.ts new file mode 100644 index 0000000..f532946 --- /dev/null +++ b/src/agent/claude-code/cli-events.ts @@ -0,0 +1,181 @@ +import type { Usage } from "@earendil-works/pi-ai"; + +export type CliUsage = Partial>; + +export type CliBlock = { type: "text" } | { type: "thinking" } | { type: "toolUse"; id: string; name: string }; + +export type CliEvent = + | { type: "messageStart"; usage: CliUsage } + | { type: "blockStart"; index: number; block: CliBlock } + | { type: "textDelta"; index: number; text: string } + | { type: "thinkingDelta"; index: number; thinking: string } + | { type: "toolInputDelta"; index: number; partialJson: string } + | { type: "blockStop"; index: number } + | { type: "messageDelta"; stopReason: string | null; usage: CliUsage } + | { type: "messageStop" } + | { type: "assistantError"; error: string; text: string } + | { type: "resultError"; text: string }; + +type RawUsage = { + input_tokens?: number | null; + output_tokens?: number | null; + cache_read_input_tokens?: number | null; + cache_creation_input_tokens?: number | null; +}; + +type RawContentBlock = { type: string; id?: string; name?: string }; + +type RawDelta = { + type?: string; + text?: string; + thinking?: string; + partial_json?: string; + stop_reason?: string | null; +}; + +type RawStreamEvent = { + type: string; + index?: number; + message?: { usage?: RawUsage }; + content_block?: RawContentBlock; + delta?: RawDelta; + usage?: RawUsage; +}; + +type RawCliLine = { + type?: string; + event?: RawStreamEvent; + error?: string | null; + message?: { content?: { type: string; text?: string }[] }; + is_error?: boolean; + result?: string; + subtype?: string; + errors?: string[]; +}; + +function toUsage(raw: RawUsage): CliUsage { + const usage: CliUsage = {}; + + if (typeof raw.input_tokens === "number") usage.input = raw.input_tokens; + if (typeof raw.output_tokens === "number") usage.output = raw.output_tokens; + if (typeof raw.cache_read_input_tokens === "number") usage.cacheRead = raw.cache_read_input_tokens; + if (typeof raw.cache_creation_input_tokens === "number") usage.cacheWrite = raw.cache_creation_input_tokens; + + return usage; +} + +function toBlock(raw: RawContentBlock): CliBlock | null { + switch (raw.type) { + case "text": + return { type: "text" }; + + case "thinking": + return { type: "thinking" }; + + case "tool_use": + if (typeof raw.id !== "string" || typeof raw.name !== "string") return null; + return { type: "toolUse", id: raw.id, name: raw.name }; + + default: + return null; + } +} + +function parseDelta(index: number, delta: RawDelta): CliEvent | null { + switch (delta.type) { + case "text_delta": + if (typeof delta.text !== "string") return null; + return { type: "textDelta", index, text: delta.text }; + + case "thinking_delta": + if (typeof delta.thinking !== "string") return null; + return { type: "thinkingDelta", index, thinking: delta.thinking }; + + case "input_json_delta": + if (typeof delta.partial_json !== "string") return null; + return { type: "toolInputDelta", index, partialJson: delta.partial_json }; + + default: + return null; + } +} + +function parseBlockEvent(index: number, event: RawStreamEvent): CliEvent | null { + switch (event.type) { + case "content_block_start": { + if (!event.content_block) return null; + + const block = toBlock(event.content_block); + return block ? { type: "blockStart", index, block } : null; + } + + case "content_block_delta": + return event.delta ? parseDelta(index, event.delta) : null; + + case "content_block_stop": + return { type: "blockStop", index }; + + default: + return null; + } +} + +function parseStreamEvent(event: RawStreamEvent): CliEvent | null { + switch (event.type) { + case "message_start": + return { type: "messageStart", usage: toUsage(event.message?.usage ?? {}) }; + + case "message_delta": { + const stopReason = event.delta?.stop_reason ?? null; + return { type: "messageDelta", stopReason, usage: toUsage(event.usage ?? {}) }; + } + + case "message_stop": + return { type: "messageStop" }; + + default: + return event.index === undefined ? null : parseBlockEvent(event.index, event); + } +} + +function getText(line: RawCliLine): string { + const blocks = line.message?.content ?? []; + const texts = blocks.flatMap((block) => (block.type === "text" && block.text ? [block.text] : [])); + + return texts.join("\n"); +} + +function getResultErrorText(line: RawCliLine): string { + if (line.result) return line.result; + if (line.errors?.length) return line.errors.join("\n"); + return `claude CLI failed: ${line.subtype ?? "unknown error"}`; +} + +export function parseCliLine(text: string): CliEvent | null { + let parsed: unknown; + try { + parsed = JSON.parse(text); + } catch { + return null; + } + + if (typeof parsed !== "object" || parsed === null) return null; + + const line = parsed as RawCliLine; + + switch (line.type) { + case "stream_event": + return line.event ? parseStreamEvent(line.event) : null; + + case "assistant": + if (typeof line.error !== "string") return null; + return { type: "assistantError", error: line.error, text: getText(line) }; + + case "result": + if (line.is_error !== true) return null; + return { type: "resultError", text: getResultErrorText(line) }; + + default: + return null; + } +} diff --git a/src/agent/claude-code/live-processes.ts b/src/agent/claude-code/live-processes.ts new file mode 100644 index 0000000..f5e24b3 --- /dev/null +++ b/src/agent/claude-code/live-processes.ts @@ -0,0 +1,41 @@ +export type LiveProcess = { toolCallIds: string[]; kill(): void }; + +// pi runs a message's tools (each capped at 30s by tool-execute) and calls back well within this; +// it stays under MCP_TOOL_TIMEOUT so an orphaned process is killed before the CLI gives up on its own. +export const LIVE_PROCESS_IDLE_TIMEOUT_MS = 10 * 60_000; + +/** Test seam (runner.ts's pdfModelOverride idiom). */ +export const liveProcessIdleTimeoutMs = { current: LIVE_PROCESS_IDLE_TIMEOUT_MS }; + +// Each live process mapped to the timer that kills it if pi never calls back. +const liveProcesses = new Map(); + +export function addLiveProcess(process: LiveProcess): void { + const timer = setTimeout(() => { + removeLiveProcess(process); + process.kill(); + }, liveProcessIdleTimeoutMs.current); + + timer.unref(); + + liveProcesses.set(process, timer); +} + +function hasSameIds(ids: string[], otherIds: string[]): boolean { + return ids.length === otherIds.length && ids.every((id) => otherIds.includes(id)); +} + +export function getLiveProcess(toolCallIds: string[]): LiveProcess | null { + if (toolCallIds.length === 0) return null; + + for (const process of liveProcesses.keys()) { + if (hasSameIds(process.toolCallIds, toolCallIds)) return process; + } + + return null; +} + +export function removeLiveProcess(process: LiveProcess): void { + clearTimeout(liveProcesses.get(process)); + liveProcesses.delete(process); +} diff --git a/src/agent/claude-code/stream.ts b/src/agent/claude-code/stream.ts new file mode 100644 index 0000000..919561a --- /dev/null +++ b/src/agent/claude-code/stream.ts @@ -0,0 +1,602 @@ +import { spawn, type ChildProcessWithoutNullStreams } from "node:child_process"; +import { mkdtemp, rm, writeFile } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { createInterface } from "node:readline"; +import { isDeepStrictEqual } from "node:util"; + +import { + createAssistantMessageEventStream, + type Api, + type AssistantMessage, + type AssistantMessageEventStream, + type Message, + type Model, + type SimpleStreamOptions, + type StopReason, + type StreamFunction, + type ThinkingLevel, + type Tool, + type ToolCall, + type ToolResultMessage, + type TranscriptContext, + validateToolCall, +} from "@earendil-works/pi-ai"; +import type { AnthropicEffort } from "@earendil-works/pi-ai/api/anthropic-messages"; +import { parseStreamingJson, parseToolCallArguments } from "@earendil-works/pi-ai/utils/json-parse"; + +import { getClaudeBinary } from "./binary.ts"; +import { parseCliLine, type CliBlock, type CliEvent, type CliUsage } from "./cli-events.ts"; +import { ClaudeCodeError } from "./constants.ts"; +import { addLiveProcess, getLiveProcess, removeLiveProcess, type LiveProcess } from "./live-processes.ts"; +import { createToolBridge, toolNameFromMcp, type PendingToolCall, type ToolBridge } from "./tool-bridge.ts"; +import { toClaudeInput, toStreamJsonLine } from "./transcript.ts"; + +type RunClaudeOptions = { cwd: string; env: NodeJS.ProcessEnv }; + +export type RunClaude = (args: string[], options: RunClaudeOptions) => ChildProcessWithoutNullStreams | null; + +// A held MCP call lasts until pi has run every tool of the message (each capped at 30s by +// tool-execute) and called back; this must outlast that so the CLI never times out first. +const MCP_TOOL_TIMEOUT_MS = 30 * 60_000; + +const PRIVATE_FILE_MODE = 0o600; + +const EFFORTS: Record = { + minimal: "low", + low: "low", + medium: "medium", + high: "high", + xhigh: "xhigh", + max: "max", +}; + +const STOP_REASONS: Record = { + end_turn: "stop", + pause_turn: "stop", + stop_sequence: "stop", + max_tokens: "length", + tool_use: "toolUse", + refusal: "error", +}; + +const REFUSAL_MESSAGE = "The model refused to complete the request"; +const ABORT_MESSAGE = "Request was aborted"; +const START_FAILED_MESSAGE = "claude CLI failed to start"; +const NO_STOP_REASON_MESSAGE = "claude CLI ended the message without a stop reason"; +const NOT_SIGNED_IN_ERROR = "authentication_failed"; + +export function toEffort(level: ThinkingLevel | undefined): AnthropicEffort { + return level ? EFFORTS[level] : "medium"; +} + +type ClaudeArgsInput = { + model: string; + effort: AnthropicEffort; + systemPromptFile: string; + mcpConfigFile: string; + allowedTools: string[]; +}; + +export function toClaudeArgs(input: ClaudeArgsInput): string[] { + return [ + "-p", + "--input-format", + "stream-json", + "--output-format", + "stream-json", + "--verbose", + "--include-partial-messages", + "--model", + input.model, + "--effort", + input.effort, + "--system-prompt-file", + input.systemPromptFile, + "--tools", + "", + "--strict-mcp-config", + "--mcp-config", + input.mcpConfigFile, + "--setting-sources", + "", + "--disable-slash-commands", + "--no-session-persistence", + "--permission-prompts", + "none", + ...(input.allowedTools.length > 0 ? ["--allowedTools", input.allowedTools.join(",")] : []), + ]; +} + +function createOutput(model: Model): AssistantMessage { + return { + role: "assistant", + content: [], + api: model.api, + provider: model.provider, + model: model.id, + usage: { + input: 0, + output: 0, + cacheRead: 0, + cacheWrite: 0, + totalTokens: 0, + cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 }, + }, + stopReason: "pending", + timestamp: Date.now(), + }; +} + +function getAssistantErrorMessage(error: string, text: string): string { + return error === NOT_SIGNED_IN_ERROR ? ClaudeCodeError.NotSignedIn : text || error; +} + +class ClaudeTurn { + private readonly output: AssistantMessage; + private readonly contentIndexes = new Map(); + private readonly partialJson = new Map(); + started = false; + finished = false; + + constructor( + private readonly stream: AssistantMessageEventStream, + model: Model, + ) { + this.output = createOutput(model); + } + + handle(event: CliEvent): void { + if (this.finished) return; + + switch (event.type) { + case "messageStart": + this.updateUsage(event.usage); + this.start(); + return; + + case "blockStart": + this.startBlock(event.index, event.block); + return; + + case "textDelta": { + const found = this.getBlock(event.index); + if (!found) return; + + const { contentIndex, block } = found; + if (block.type !== "text") return; + + block.text += event.text; + this.stream.push({ type: "text_delta", contentIndex, delta: event.text, partial: this.output }); + return; + } + + case "thinkingDelta": { + const found = this.getBlock(event.index); + if (!found) return; + + const { contentIndex, block } = found; + if (block.type !== "thinking") return; + + block.thinking += event.thinking; + this.stream.push({ type: "thinking_delta", contentIndex, delta: event.thinking, partial: this.output }); + return; + } + + case "toolInputDelta": { + const found = this.getBlock(event.index); + if (!found) return; + + const { contentIndex, block } = found; + if (block.type !== "toolCall") return; + + const partialJson = (this.partialJson.get(event.index) ?? "") + event.partialJson; + this.partialJson.set(event.index, partialJson); + block.arguments = parseStreamingJson(partialJson); + this.stream.push({ type: "toolcall_delta", contentIndex, delta: event.partialJson, partial: this.output }); + return; + } + + case "blockStop": + this.stopBlock(event.index); + return; + + case "messageDelta": + this.updateUsage(event.usage); + if (event.stopReason) this.setStopReason(event.stopReason); + return; + + case "messageStop": + this.finish(); + return; + + case "assistantError": + this.fail(getAssistantErrorMessage(event.error, event.text), "error"); + return; + + case "resultError": + this.fail(event.text, "error"); + return; + } + } + + getToolCalls(): ToolCall[] { + if (!this.finished || this.output.stopReason !== "toolUse") return []; + + return this.output.content.filter((block) => block.type === "toolCall"); + } + + fail(message: string, reason: "aborted" | "error"): void { + if (this.finished) return; + + this.finished = true; + this.output.stopReason = reason; + this.output.errorMessage = message; + this.stream.push({ type: "error", reason, error: this.output }); + this.stream.end(); + } + + private start(): void { + if (this.started) return; + + this.started = true; + this.stream.push({ type: "start", partial: this.output }); + } + + private startBlock(index: number, block: CliBlock): void { + const contentIndex = this.output.content.length; + this.contentIndexes.set(index, contentIndex); + + switch (block.type) { + case "text": + this.output.content.push({ type: "text", text: "" }); + this.stream.push({ type: "text_start", contentIndex, partial: this.output }); + return; + + case "thinking": + this.output.content.push({ type: "thinking", thinking: "" }); + this.stream.push({ type: "thinking_start", contentIndex, partial: this.output }); + return; + + case "toolUse": + this.output.content.push({ type: "toolCall", id: block.id, name: toolNameFromMcp(block.name), arguments: {} }); + this.stream.push({ type: "toolcall_start", contentIndex, partial: this.output }); + return; + } + } + + private getBlock(index: number): { contentIndex: number; block: AssistantMessage["content"][number] } | null { + const contentIndex = this.contentIndexes.get(index); + if (contentIndex === undefined) return null; + + const block = this.output.content[contentIndex]; + return block ? { contentIndex, block } : null; + } + + private stopBlock(index: number): void { + const found = this.getBlock(index); + if (!found) return; + + const { contentIndex, block } = found; + this.contentIndexes.delete(index); + + switch (block.type) { + case "text": + this.stream.push({ type: "text_end", contentIndex, content: block.text, partial: this.output }); + return; + + case "thinking": + this.stream.push({ type: "thinking_end", contentIndex, content: block.thinking, partial: this.output }); + return; + + case "toolCall": + Object.assign(block, parseToolCallArguments(this.partialJson.get(index) ?? "")); + this.partialJson.delete(index); + this.stream.push({ type: "toolcall_end", contentIndex, toolCall: block satisfies ToolCall, partial: this.output }); + return; + } + } + + private updateUsage(usage: CliUsage): void { + Object.assign(this.output.usage, usage); + + const { input, output, cacheRead, cacheWrite } = this.output.usage; + this.output.usage.totalTokens = input + output + cacheRead + cacheWrite; + } + + private setStopReason(stopReason: string): void { + const mapped = STOP_REASONS[stopReason]; + + if (!mapped) { + this.output.stopReason = "error"; + this.output.errorMessage = `Unhandled stop reason: ${stopReason}`; + return; + } + + this.output.stopReason = mapped; + } + + private finish(): void { + const { stopReason } = this.output; + + if (stopReason === "error") { + this.fail(this.output.errorMessage ?? REFUSAL_MESSAGE, "error"); + return; + } + + if (stopReason === "pending" || stopReason === "aborted") { + this.fail(this.output.errorMessage ?? NO_STOP_REASON_MESSAGE, "error"); + return; + } + + this.finished = true; + this.stream.push({ type: "done", reason: stopReason, message: this.output }); + this.stream.end(); + } +} + +type TurnFiles = { cwd: string; privateDir: string; systemPromptFile: string; mcpConfigFile: string }; + +async function createTurnFiles(systemPrompt: string, mcpConfig: string): Promise { + const cwd = await mkdtemp(join(tmpdir(), "claude-code-cwd-")); + const privateDir = await mkdtemp(join(tmpdir(), "claude-code-private-")); + + const systemPromptFile = join(privateDir, "system-prompt.md"); + const mcpConfigFile = join(privateDir, "mcp.json"); + + await writeFile(systemPromptFile, systemPrompt, { mode: PRIVATE_FILE_MODE }); + await writeFile(mcpConfigFile, mcpConfig, { mode: PRIVATE_FILE_MODE }); + + return { cwd, privateDir, systemPromptFile, mcpConfigFile }; +} + +async function removeTurnResources(bridge: ToolBridge, files: TurnFiles): Promise { + await bridge.close(); + await rm(files.cwd, { recursive: true, force: true }); + await rm(files.privateDir, { recursive: true, force: true }); +} + +type UnansweredToolCall = { toolCall: ToolCall; result: ToolResultMessage }; + +// The CLI answers calls to undeclared tools or with invalid arguments itself, without reaching the bridge. +function isValidToolCall(tools: Tool[], toolCall: ToolCall): boolean { + try { + validateToolCall(tools, toolCall); + return true; + } catch { + return false; + } +} + +function isSameToolCall(toolCall: ToolCall, call: PendingToolCall): boolean { + return toolCall.name === call.name && isDeepStrictEqual(toolCall.arguments, call.arguments); +} + +class ClaudeProcess implements LiveProcess { + toolCallIds: string[] = []; + private stderr = ""; + private closed = false; + private signal: AbortSignal | undefined; + + constructor( + private readonly child: ChildProcessWithoutNullStreams, + private readonly bridge: ToolBridge, + private readonly files: TurnFiles, + private readonly tools: Tool[], + private turn: ClaudeTurn, + ) {} + + start(stdinLine: string, signal: AbortSignal | undefined): void { + this.child.once("close", (code: number | null) => this.handleClose(`claude CLI exited with code ${code}`)); + this.child.once("error", (error) => { + console.error("claude CLI process failed:", error); + this.turn.fail(START_FAILED_MESSAGE, "error"); + this.handleClose(START_FAILED_MESSAGE); + }); + + this.child.stderr.setEncoding("utf8"); + this.child.stderr.on("data", (chunk: string) => (this.stderr += chunk)); + + this.child.stdin.on("error", (error) => console.error("claude CLI stdin closed early:", error)); + this.child.stdin.end(`${stdinLine}\n`); + + createInterface({ input: this.child.stdout }).on("line", (line) => this.handleLine(line)); + + this.setSignal(signal); + } + + continueTurn(turn: ClaudeTurn, toolResults: ToolResultMessage[], signal: AbortSignal | undefined): void { + removeLiveProcess(this); + + const unanswered = this.turn.getToolCalls().flatMap((toolCall) => { + const result = toolResults.find((toolResult) => toolResult.toolCallId === toolCall.id); + return result ? [{ toolCall, result }] : []; + }); + + this.turn = turn; + this.setSignal(signal); + + this.answerToolCalls(turn, unanswered).catch((error: unknown) => { + console.error("claude CLI tool results failed:", error); + turn.fail(START_FAILED_MESSAGE, "error"); + this.kill(); + }); + } + + kill(): void { + removeLiveProcess(this); + this.child.kill(); + } + + private readonly abort = () => { + this.turn.fail(ABORT_MESSAGE, "aborted"); + this.kill(); + }; + + private setSignal(signal: AbortSignal | undefined): void { + this.signal?.removeEventListener("abort", this.abort); + this.signal = signal; + signal?.addEventListener("abort", this.abort, { once: true }); + + if (signal?.aborted) this.abort(); + } + + private handleLine(line: string): void { + const event = parseCliLine(line); + if (!event) return; + + // Output after a finished message means the CLI moved on without waiting for our tool results. + if (this.turn.finished) { + this.kill(); + return; + } + + this.turn.handle(event); + if (!this.turn.finished) return; + + const toolCalls = this.turn.getToolCalls(); + this.toolCallIds = toolCalls.map((toolCall) => toolCall.id); + + const reachesBridge = toolCalls.length > 0 && toolCalls.every((toolCall) => isValidToolCall(this.tools, toolCall)); + if (!reachesBridge) { + this.kill(); + return; + } + + addLiveProcess(this); + } + + private async answerToolCalls(turn: ClaudeTurn, unanswered: UnansweredToolCall[]): Promise { + while (unanswered.length > 0 && !turn.started) { + const pending = this.bridge.getPendingToolCalls(); + let answered = false; + + for (const call of pending) { + const index = unanswered.findIndex(({ toolCall }) => isSameToolCall(toolCall, call)); + const match = unanswered[index]; + if (!match) continue; + + this.bridge.resolveToolCall(call, match.result); + unanswered.splice(index, 1); + answered = true; + } + + if (answered) continue; + + // Rejects only once the bridge has closed, which means the process is already gone. + const next = await this.bridge.waitForToolCalls(pending.length + 1).catch(() => null); + if (!next) return; + } + } + + private handleClose(message: string): void { + if (this.closed) return; + + this.closed = true; + removeLiveProcess(this); + this.signal?.removeEventListener("abort", this.abort); + + if (this.stderr.trim()) console.error("claude CLI stderr:", this.stderr.trim()); + + this.turn.fail(message, "error"); + + removeTurnResources(this.bridge, this.files).catch((error: unknown) => { + console.error("claude CLI cleanup failed:", error); + }); + } +} + +function continueLiveProcess(turn: ClaudeTurn, messages: Message[], signal: AbortSignal | undefined): boolean { + const assistantIndex = messages.findLastIndex((message) => message.role === "assistant"); + const assistant = messages[assistantIndex]; + if (assistant?.role !== "assistant") return false; + + const toolCallIds = assistant.content.flatMap((block) => (block.type === "toolCall" ? [block.id] : [])); + const process = getLiveProcess(toolCallIds); + if (!(process instanceof ClaudeProcess)) return false; + + const trailing = messages.slice(assistantIndex + 1); + const toolResults = trailing.filter((message) => message.role === "toolResult"); + + const onlyToolResults = toolResults.length === trailing.length; + const answersEveryCall = toolCallIds.every((id) => toolResults.some((toolResult) => toolResult.toolCallId === id)); + + if (!onlyToolResults || !answersEveryCall) { + process.kill(); + return false; + } + + process.continueTurn(turn, toolResults, signal); + return true; +} + +// Either one takes precedence over the logged-in subscription, so a key saved on the Keys page +// would bill turns to the API account. +const API_CREDENTIAL_ENV_NAMES = ["ANTHROPIC_API_KEY", "ANTHROPIC_AUTH_TOKEN"]; + +function toClaudeEnv(env: NodeJS.ProcessEnv): NodeJS.ProcessEnv { + const claudeEnv: NodeJS.ProcessEnv = { ...env, MCP_TOOL_TIMEOUT: String(MCP_TOOL_TIMEOUT_MS) }; + + for (const name of API_CREDENTIAL_ENV_NAMES) delete claudeEnv[name]; + + return claudeEnv; +} + +async function startProcess( + runClaude: RunClaude, + turn: ClaudeTurn, + model: Model, + context: TranscriptContext, + options: SimpleStreamOptions | undefined, +): Promise { + const input = toClaudeInput(context.messages); + const bridge = await createToolBridge(input.tools); + const files = await createTurnFiles(input.systemPrompt, bridge.mcpConfig); + + let child: ChildProcessWithoutNullStreams | null; + try { + const args = toClaudeArgs({ + model: model.id, + effort: toEffort(options?.reasoning), + systemPromptFile: files.systemPromptFile, + mcpConfigFile: files.mcpConfigFile, + allowedTools: bridge.allowedTools, + }); + child = runClaude(args, { cwd: files.cwd, env: toClaudeEnv(process.env) }); + } catch (error) { + await removeTurnResources(bridge, files); + throw error; + } + + if (!child) { + turn.fail(ClaudeCodeError.CliNotFound, "error"); + await removeTurnResources(bridge, files); + return; + } + + new ClaudeProcess(child, bridge, files, input.tools, turn).start(toStreamJsonLine(input.userContent), options?.signal); +} + +export function createClaudeCodeStreamFunction(runClaude: RunClaude): StreamFunction { + return (model, context, options) => { + const stream = createAssistantMessageEventStream(); + const turn = new ClaudeTurn(stream, model); + + if (continueLiveProcess(turn, context.messages, options?.signal)) return stream; + + startProcess(runClaude, turn, model, context, options).catch((error: unknown) => { + console.error("claude CLI turn failed:", error); + turn.fail(START_FAILED_MESSAGE, "error"); + }); + + return stream; + }; +} + +function spawnClaude(args: string[], options: RunClaudeOptions): ChildProcessWithoutNullStreams | null { + const binary = getClaudeBinary(); + + return binary ? spawn(binary, args, { ...options, stdio: "pipe" }) : null; +} + +export const streamClaudeCode = createClaudeCodeStreamFunction(spawnClaude); diff --git a/tests/claude-code-stream.test.ts b/tests/claude-code-stream.test.ts new file mode 100644 index 0000000..96acc52 --- /dev/null +++ b/tests/claude-code-stream.test.ts @@ -0,0 +1,781 @@ +/** + * Claude Code stream function (src/agent/claude-code/stream.ts) and CLI line + * parser (src/agent/claude-code/cli-events.ts). + * + * The runClaude seam spawns tests/fixtures/claude-code/fake-claude.mjs, which + * replays recorded or synthetic stream-json lines, so the real process + * plumbing (argv, stdin, stdout lines, stderr, exit, kill) runs end to end. + */ +import assert from "node:assert/strict"; +import { spawn, type ChildProcessWithoutNullStreams } from "node:child_process"; +import { once } from "node:events"; +import { existsSync, mkdtempSync, readdirSync, readFileSync, rmSync, statSync, writeFileSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { dirname, join, sep } from "node:path"; +import { setTimeout as sleep } from "node:timers/promises"; +import { fileURLToPath } from "node:url"; +import test, { type TestContext } from "node:test"; + +import type { + AssistantMessage, + AssistantMessageEvent, + AssistantMessageEventStream, + JsonObject, + Message, + SimpleStreamOptions, + StreamFunction, + Api, + Tool, + ToolResultMessage, +} from "@earendil-works/pi-ai"; +import { normalizeContext } from "@earendil-works/pi-ai/utils/transcript"; + +import { parseCliLine } from "../src/agent/claude-code/cli-events.ts"; +import { ClaudeCodeError } from "../src/agent/claude-code/constants.ts"; +import { LIVE_PROCESS_IDLE_TIMEOUT_MS, liveProcessIdleTimeoutMs } from "../src/agent/claude-code/live-processes.ts"; +import { CLAUDE_CODE_MODELS } from "../src/agent/claude-code/provider.ts"; +import { createClaudeCodeStreamFunction, toClaudeArgs, toEffort, type RunClaude } from "../src/agent/claude-code/stream.ts"; +import { toStreamJsonLine } from "../src/agent/claude-code/transcript.ts"; + +const FAKE_CLAUDE = fileURLToPath(new URL("./fixtures/claude-code/fake-claude.mjs", import.meta.url)); +const FIXTURES = fileURLToPath(new URL("./fixtures/claude-code/", import.meta.url)); +const HOLD = JSON.stringify({ fake: "hold" }); +const REPLY = JSON.stringify({ fake: "reply" }); +const CLEANUP_TIMEOUT_MS = 2000; + +const READ_TOOL: Tool = { name: "read", description: "Read a file", parameters: { type: "object", properties: {} } }; +const HI: Message = { role: "user", content: "hi", timestamp: 0 }; + +type ToolUse = { id: string; name: string; arguments: JsonObject }; + +type RunCall = { + args: string[]; + cwd: string; + cwdEntries: string[]; + env: NodeJS.ProcessEnv; + systemPrompt: string; + mcpConfig: string; + mcpConfigMode: number; + child: ChildProcessWithoutNullStreams; +}; + +type ScenarioRun = { events: AssistantMessageEvent[]; calls: RunCall[] }; + +function readFixtureLines(name: string): string[] { + return readFileSync(join(FIXTURES, name), "utf8").split("\n").filter(Boolean); +} + +function getFlagValue(args: string[], flag: string): string { + const value = args[args.indexOf(flag) + 1]; + assert.ok(args.includes(flag) && value !== undefined, `missing ${flag}`); + + return value; +} + +function streamEvent(event: object): string { + return JSON.stringify({ type: "stream_event", event }); +} + +function exitWith(code: number, stderr: string): string { + return JSON.stringify({ fake: "exit", code, stderr }); +} + +function getToolScenario(): string[] { + const lines = readFixtureLines("tool.jsonl"); + const firstMessageStop = lines.findIndex((line) => line.includes('"type":"message_stop"')); + return [...lines.slice(0, firstMessageStop + 1).filter((line) => !line.startsWith('{"type":"user"')), HOLD]; +} + +function createFakeRun( + t: TestContext, + scenario: string[], + extraEnv: Record = {}, +): { runClaude: RunClaude; calls: RunCall[] } { + const dir = mkdtempSync(join(tmpdir(), "claude-code-test-")); + t.after(() => rmSync(dir, { recursive: true, force: true })); + + const scenarioFile = join(dir, "scenario.jsonl"); + writeFileSync(scenarioFile, `${scenario.join("\n")}\n`); + + const calls: RunCall[] = []; + + const runClaude: RunClaude = (args, options) => { + const mcpConfigFile = getFlagValue(args, "--mcp-config"); + const child = spawn(process.execPath, [FAKE_CLAUDE, ...args], { + cwd: options.cwd, + env: { ...options.env, FAKE_CLAUDE_SCENARIO: scenarioFile, ...extraEnv }, + stdio: "pipe", + }); + t.after(() => child.kill()); + calls.push({ + args, + cwd: options.cwd, + cwdEntries: readdirSync(options.cwd), + env: options.env, + systemPrompt: readFileSync(getFlagValue(args, "--system-prompt-file"), "utf8"), + mcpConfig: readFileSync(mcpConfigFile, "utf8"), + mcpConfigMode: statSync(mcpConfigFile).mode & 0o777, + child, + }); + return child; + }; + + return { runClaude, calls }; +} + +async function collectEvents( + stream: AssistantMessageEventStream, + onEvent: (event: AssistantMessageEvent) => void = () => {}, +): Promise { + const events: AssistantMessageEvent[] = []; + + for await (const event of stream) { + events.push(event); + onEvent(event); + } + + return events; +} + +async function runScenario( + t: TestContext, + scenario: string[], + options?: SimpleStreamOptions, + extraEnv?: Record, +): Promise { + const { runClaude, calls } = createFakeRun(t, scenario, extraEnv); + + const model = CLAUDE_CODE_MODELS[0]; + assert.ok(model); + + const context = normalizeContext({ + systemPrompt: "You are a tutor.", + messages: [{ role: "user", content: "hi", timestamp: 0 }], + tools: [READ_TOOL], + }); + const events = await collectEvents(createClaudeCodeStreamFunction(runClaude)(model, context, options)); + + return { events, calls }; +} + +function toToolUseMessage(toolUses: ToolUse[]): string[] { + return [ + streamEvent({ type: "message_start", message: { usage: {} } }), + ...toolUses.flatMap((toolUse, index) => [ + streamEvent({ + type: "content_block_start", + index, + content_block: { type: "tool_use", id: toolUse.id, name: `mcp__learnharness__${toolUse.name}`, input: {} }, + }), + streamEvent({ + type: "content_block_delta", + index, + delta: { type: "input_json_delta", partial_json: JSON.stringify(toolUse.arguments) }, + }), + streamEvent({ type: "content_block_stop", index }), + ]), + streamEvent({ type: "message_delta", delta: { stop_reason: "tool_use" }, usage: {} }), + streamEvent({ type: "message_stop" }), + ]; +} + +function callTool(toolUse: ToolUse): string { + return JSON.stringify({ fake: "call", name: toolUse.name, arguments: toolUse.arguments }); +} + +function toToolResult(toolUse: ToolUse, text: string): ToolResultMessage { + return { + role: "toolResult", + toolCallId: toolUse.id, + toolName: toolUse.name, + content: [{ type: "text", text }], + isError: false, + timestamp: 0, + }; +} + +function getModel() { + const model = CLAUDE_CODE_MODELS[0]; + assert.ok(model); + return model; +} + +async function runTurn( + streamFunction: StreamFunction, + messages: Message[], + options?: SimpleStreamOptions, +): Promise { + const context = normalizeContext({ systemPrompt: "You are a tutor.", messages, tools: [READ_TOOL] }); + + const done = getLastEvent(await collectEvents(streamFunction(getModel(), context, options))); + + assert.ok(done.type === "done", `turn ended with ${done.type}`); + + return done.message; +} + +function getText(message: AssistantMessage): string { + return message.content.flatMap((block) => (block.type === "text" ? [block.text] : [])).join(""); +} + +function getMcpUrl(call: RunCall): string { + const mcpConfig = JSON.parse(call.mcpConfig) as { mcpServers: Record }; + const url = mcpConfig.mcpServers.learnharness?.url; + assert.ok(url); + return url; +} + +function getOnlyCall(calls: RunCall[]): RunCall { + assert.equal(calls.length, 1); + const [call] = calls; + assert.ok(call); + return call; +} + +function getLastEvent(events: AssistantMessageEvent[]): AssistantMessageEvent { + const event = events.at(-1); + assert.ok(event); + return event; +} + +function getErrorMessage(events: AssistantMessageEvent[]): string | undefined { + const event = getLastEvent(events); + assert.equal(event.type, "error"); + return event.type === "error" ? event.error.errorMessage : undefined; +} + +async function waitForExit(child: ChildProcessWithoutNullStreams): Promise { + if (child.exitCode !== null || child.signalCode !== null) return; + + await assert.doesNotReject(once(child, "exit", { signal: AbortSignal.timeout(CLEANUP_TIMEOUT_MS) }), "claude CLI process was not killed"); +} + +async function waitForRemoval(path: string): Promise { + const deadline = Date.now() + CLEANUP_TIMEOUT_MS; + + while (existsSync(path)) { + assert.ok(Date.now() < deadline, `${path} was not removed`); + await sleep(10); + } +} + +async function assertCleanedUp(call: RunCall): Promise { + await waitForExit(call.child); + await waitForRemoval(call.cwd); + await waitForRemoval(dirname(getFlagValue(call.args, "--system-prompt-file"))); +} + +test("toClaudeArgs includes every isolation flag from Global Constraints", () => { + const input = { + model: "opus", + effort: "high", + systemPromptFile: "/private/system-prompt.md", + mcpConfigFile: "/private/mcp.json", + allowedTools: ["mcp__learnharness__read", "mcp__learnharness__write"], + } as const; + + assert.deepEqual(toClaudeArgs({ ...input, allowedTools: [...input.allowedTools] }), [ + "-p", + "--input-format", + "stream-json", + "--output-format", + "stream-json", + "--verbose", + "--include-partial-messages", + "--model", + "opus", + "--effort", + "high", + "--system-prompt-file", + "/private/system-prompt.md", + "--tools", + "", + "--strict-mcp-config", + "--mcp-config", + "/private/mcp.json", + "--setting-sources", + "", + "--disable-slash-commands", + "--no-session-persistence", + "--permission-prompts", + "none", + "--allowedTools", + "mcp__learnharness__read,mcp__learnharness__write", + ]); + + assert.equal(toClaudeArgs({ ...input, allowedTools: [] }).includes("--allowedTools"), false); +}); + +test('toEffort maps low..max verbatim, minimal to "low", undefined to "medium"', () => { + assert.deepEqual( + (["low", "medium", "high", "xhigh", "max"] as const).map((level) => toEffort(level)), + ["low", "medium", "high", "xhigh", "max"], + ); + + assert.equal(toEffort("minimal"), "low"); + assert.equal(toEffort(undefined), "medium"); +}); + +test("text scenario streams start, text_start, text_delta…, text_end, done(stop)", async (t) => { + const { events, calls } = await runScenario(t, readFixtureLines("text.jsonl")); + + assert.deepEqual( + events.map((event) => event.type), + ["start", "text_start", "text_delta", "text_end", "done"], + ); + + const done = getLastEvent(events); + assert.ok(done.type === "done"); + assert.equal(done.reason, "stop"); + assert.equal(done.message.stopReason, "stop"); + assert.deepEqual(done.message.content, [{ type: "text", text: "hello there" }]); + assert.equal(done.message.provider, "claude-code"); + assert.equal(done.message.api, "claude-code-cli"); + assert.equal(done.message.model, "opus"); + + await assertCleanedUp(getOnlyCall(calls)); +}); + +test("thinking deltas map to thinking_*", async (t) => { + const { events } = await runScenario(t, [ + streamEvent({ type: "message_start", message: { usage: { input_tokens: 3, output_tokens: 1 } } }), + streamEvent({ type: "content_block_start", index: 0, content_block: { type: "thinking", thinking: "" } }), + streamEvent({ type: "content_block_delta", index: 0, delta: { type: "thinking_delta", thinking: "Sockets are " } }), + streamEvent({ type: "content_block_delta", index: 0, delta: { type: "thinking_delta", thinking: "endpoints." } }), + streamEvent({ type: "content_block_stop", index: 0 }), + streamEvent({ type: "content_block_start", index: 1, content_block: { type: "text", text: "" } }), + streamEvent({ type: "content_block_delta", index: 1, delta: { type: "text_delta", text: "An endpoint." } }), + streamEvent({ type: "content_block_stop", index: 1 }), + streamEvent({ type: "message_delta", delta: { stop_reason: "end_turn" }, usage: { output_tokens: 9 } }), + streamEvent({ type: "message_stop" }), + ]); + + assert.deepEqual( + events.map((event) => event.type), + ["start", "thinking_start", "thinking_delta", "thinking_delta", "thinking_end", "text_start", "text_delta", "text_end", "done"], + ); + + const done = getLastEvent(events); + assert.ok(done.type === "done"); + assert.deepEqual(done.message.content, [ + { type: "thinking", thinking: "Sockets are endpoints." }, + { type: "text", text: "An endpoint." }, + ]); +}); + +test("tool scenario ends with toolcall_end (Claude's tool_use id, unprefixed name) and done(toolUse)", async (t) => { + const { events } = await runScenario(t, getToolScenario()); + + assert.deepEqual( + events.map((event) => event.type), + ["start", "toolcall_start", "toolcall_delta", "toolcall_delta", "toolcall_delta", "toolcall_end", "done"], + ); + + const toolCallEnd = events.find((event) => event.type === "toolcall_end"); + assert.ok(toolCallEnd?.type === "toolcall_end"); + assert.deepEqual(toolCallEnd.toolCall, { + type: "toolCall", + id: "toolu_01JBB4cJsPa585FWb1XrfbNX", + name: "get_weather", + arguments: { city: "Paris" }, + }); + + const done = getLastEvent(events); + assert.ok(done.type === "done"); + assert.equal(done.reason, "toolUse"); + assert.deepEqual(done.message.content, [toolCallEnd.toolCall]); +}); + +test("unknown tool name and bad JSON args still surface as tool calls", async (t) => { + const { events } = await runScenario(t, [ + streamEvent({ type: "message_start", message: { usage: {} } }), + streamEvent({ + type: "content_block_start", + index: 0, + content_block: { type: "tool_use", id: "toolu_a", name: "mcp__learnharness__nope", input: {} }, + }), + streamEvent({ type: "content_block_delta", index: 0, delta: { type: "input_json_delta", partial_json: '{"path": ' } }), + streamEvent({ type: "content_block_stop", index: 0 }), + streamEvent({ type: "content_block_start", index: 1, content_block: { type: "tool_use", id: "toolu_b", name: "Bash", input: {} } }), + streamEvent({ type: "content_block_delta", index: 1, delta: { type: "input_json_delta", partial_json: '{"command":"ls"}' } }), + streamEvent({ type: "content_block_stop", index: 1 }), + streamEvent({ type: "message_delta", delta: { stop_reason: "tool_use" }, usage: {} }), + streamEvent({ type: "message_stop" }), + HOLD, + ]); + + const toolCalls = events.flatMap((event) => (event.type === "toolcall_end" ? [event.toolCall] : [])); + assert.deepEqual( + toolCalls.map((call) => ({ id: call.id, name: call.name, arguments: call.arguments })), + [ + { id: "toolu_a", name: "nope", arguments: {} }, + { id: "toolu_b", name: "Bash", arguments: { command: "ls" } }, + ], + ); + + assert.match(toolCalls[0]?.argumentsError ?? "", /\{"path": /); + assert.equal(toolCalls[1]?.argumentsError, undefined); + + const done = getLastEvent(events); + assert.ok(done.type === "done"); + assert.equal(done.reason, "toolUse"); +}); + +test("usage maps input/output/cacheRead/cacheWrite with zero cost, message_delta overriding message_start", async (t) => { + const { events } = await runScenario(t, getToolScenario()); + const done = getLastEvent(events); + assert.ok(done.type === "done"); + assert.deepEqual(done.message.usage, { + input: 2, + output: 59, + cacheRead: 0, + cacheWrite: 1102, + totalTokens: 1163, + cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 }, + }); +}); + +test('missing binary → error "claude CLI not found"', async () => { + const model = CLAUDE_CODE_MODELS[0]; + assert.ok(model); + + let cwd = ""; + const runClaude: RunClaude = (_args, options) => { + cwd = options.cwd; + return null; + }; + + const context = normalizeContext({ systemPrompt: "p", messages: [{ role: "user", content: "hi", timestamp: 0 }] }); + const events = await collectEvents(createClaudeCodeStreamFunction(runClaude)(model, context)); + + assert.deepEqual( + events.map((event) => event.type), + ["error"], + ); + + assert.equal(getErrorMessage(events), ClaudeCodeError.CliNotFound); + await waitForRemoval(cwd); +}); + +test('signed-out fixture → error "not signed in"', async (t) => { + const { events, calls } = await runScenario(t, [...readFixtureLines("signed-out.jsonl"), exitWith(1, "")]); + assert.equal(getErrorMessage(events), ClaudeCodeError.NotSignedIn); + await assertCleanedUp(getOnlyCall(calls)); +}); + +test("other assistant errors surface their text", async (t) => { + const assistantError = JSON.stringify({ + type: "assistant", + error: "rate_limit", + message: { content: [{ type: "text", text: "API Error: Rate limit reached" }] }, + }); + const { events } = await runScenario(t, [assistantError, exitWith(1, "")]); + assert.equal(getErrorMessage(events), "API Error: Rate limit reached"); +}); + +test("result error before message_stop surfaces the result text", async (t) => { + const resultError = JSON.stringify({ + type: "result", + subtype: "error_during_execution", + is_error: true, + result: "Credit balance is too low", + }); + const { events } = await runScenario(t, [resultError, HOLD]); + assert.equal(getErrorMessage(events), "Credit balance is too low"); +}); + +test("result error without result text surfaces its errors", async (t) => { + const resultError = JSON.stringify({ + type: "result", + subtype: "error_max_turns", + is_error: true, + errors: ["Reached maximum number of turns"], + }); + const { events } = await runScenario(t, [resultError, HOLD]); + assert.equal(getErrorMessage(events), "Reached maximum number of turns"); +}); + +test("result error without result text or errors surfaces its subtype", async (t) => { + const resultError = JSON.stringify({ type: "result", subtype: "error_during_execution", is_error: true }); + const { events } = await runScenario(t, [resultError, HOLD]); + assert.equal(getErrorMessage(events), "claude CLI failed: error_during_execution"); +}); + +test("refusal stop reason → error", async (t) => { + const { events } = await runScenario(t, [ + streamEvent({ type: "message_start", message: { usage: {} } }), + streamEvent({ type: "message_delta", delta: { stop_reason: "refusal" }, usage: {} }), + streamEvent({ type: "message_stop" }), + ]); + assert.equal(getErrorMessage(events), "The model refused to complete the request"); +}); + +test("non-zero exit mid-stream → error naming the exit code, stderr logged not shown, process and temp dirs gone", async (t) => { + const consoleError = t.mock.method(console, "error", () => {}); + const midStream = readFixtureLines("text.jsonl").slice(0, 5); + const { events, calls } = await runScenario(t, [...midStream, exitWith(1, "Error: ENOENT /home/someone/.claude/x\n at main\n")]); + + assert.deepEqual( + events.map((event) => event.type), + ["start", "text_start", "text_delta", "error"], + ); + + assert.equal(getErrorMessage(events), "claude CLI exited with code 1"); + assert.ok( + consoleError.mock.calls.some((call) => call.arguments.some((argument) => String(argument).includes("ENOENT /home/someone/.claude/x"))), + ); + + const error = getLastEvent(events); + assert.ok(error.type === "error"); + assert.equal(error.reason, "error"); + assert.deepEqual(error.error.content, [{ type: "text", text: "hello there" }]); + + await assertCleanedUp(getOnlyCall(calls)); +}); + +test("exit without stderr names the exit code", async (t) => { + const { events } = await runScenario(t, [exitWith(3, "")]); + assert.equal(getErrorMessage(events), "claude CLI exited with code 3"); +}); + +test("abort kills the process and ends with error(aborted)", async (t) => { + const { runClaude, calls } = createFakeRun(t, [streamEvent({ type: "message_start", message: { usage: {} } }), HOLD]); + const model = CLAUDE_CODE_MODELS[0]; + assert.ok(model); + const controller = new AbortController(); + const context = normalizeContext({ systemPrompt: "p", messages: [{ role: "user", content: "hi", timestamp: 0 }] }); + const stream = createClaudeCodeStreamFunction(runClaude)(model, context, { signal: controller.signal }); + + const events = await collectEvents(stream, (event) => { + if (event.type === "start") controller.abort(); + }); + + assert.deepEqual( + events.map((event) => event.type), + ["start", "error"], + ); + + const error = getLastEvent(events); + assert.ok(error.type === "error"); + assert.equal(error.reason, "aborted"); + + await assertCleanedUp(getOnlyCall(calls)); +}); + +test("cwd is an empty temp dir; prompt and mcp config live in a separate private dir", async (t) => { + const { calls } = await runScenario(t, readFixtureLines("text.jsonl")); + const call = getOnlyCall(calls); + + assert.deepEqual(call.cwdEntries, []); + assert.ok(call.cwd.startsWith(tmpdir())); + + const privateDir = dirname(getFlagValue(call.args, "--system-prompt-file")); + assert.notEqual(privateDir, call.cwd); + assert.ok(!privateDir.startsWith(call.cwd + sep)); + assert.equal(dirname(getFlagValue(call.args, "--mcp-config")), privateDir); + assert.equal(call.systemPrompt, "You are a tutor."); + assert.equal(call.mcpConfigMode, 0o600); + + const mcpConfig = JSON.parse(call.mcpConfig) as { mcpServers: Record }; + assert.match(mcpConfig.mcpServers.learnharness?.url ?? "", /^http:\/\/127\.0\.0\.1:\d+\/mcp$/); +}); + +test("spawns with the model alias, effort, bridge tools, MCP_TOOL_TIMEOUT and the caller's HOME", async (t) => { + const { calls } = await runScenario(t, readFixtureLines("text.jsonl"), { reasoning: "xhigh" }); + const call = getOnlyCall(calls); + assert.equal(getFlagValue(call.args, "--model"), "opus"); + assert.equal(getFlagValue(call.args, "--effort"), "xhigh"); + assert.equal(getFlagValue(call.args, "--allowedTools"), "mcp__learnharness__read"); + assert.equal(call.env.MCP_TOOL_TIMEOUT, "1800000"); + assert.equal(call.env.HOME, process.env.HOME); + assert.equal(call.env.CLAUDE_CONFIG_DIR, process.env.CLAUDE_CONFIG_DIR); +}); + +test("spawns without Anthropic API credentials so turns stay on the subscription", async (t) => { + const saved = { ANTHROPIC_API_KEY: process.env.ANTHROPIC_API_KEY, ANTHROPIC_AUTH_TOKEN: process.env.ANTHROPIC_AUTH_TOKEN }; + t.after(() => { + for (const [name, value] of Object.entries(saved)) { + if (value === undefined) delete process.env[name]; + else process.env[name] = value; + } + }); + + process.env.ANTHROPIC_API_KEY = "sk-ant-test"; + process.env.ANTHROPIC_AUTH_TOKEN = "token-test"; + + const { calls } = await runScenario(t, readFixtureLines("text.jsonl")); + const call = getOnlyCall(calls); + + assert.equal(call.env.ANTHROPIC_API_KEY, undefined); + assert.equal(call.env.ANTHROPIC_AUTH_TOKEN, undefined); +}); + +test("writes the transcript as one stream-json user line on stdin", async (t) => { + const stdinFile = join(mkdtempSync(join(tmpdir(), "claude-code-stdin-")), "stdin.jsonl"); + t.after(() => rmSync(dirname(stdinFile), { recursive: true, force: true })); + + await runScenario(t, readFixtureLines("text.jsonl"), undefined, { FAKE_CLAUDE_STDIN: stdinFile }); + + assert.equal(readFileSync(stdinFile, "utf8"), `${toStreamJsonLine([{ type: "text", text: "hi" }])}\n`); +}); + +test("parseCliLine ignores non-JSON, system, rate-limit and successful result lines", () => { + assert.equal(parseCliLine("not json"), null); + assert.equal(parseCliLine("null"), null); + assert.equal(parseCliLine('{"type":"system","subtype":"init"}'), null); + assert.equal(parseCliLine('{"type":"rate_limit_event"}'), null); + assert.equal(parseCliLine('{"type":"result","is_error":false,"result":"hello there"}'), null); + assert.equal(parseCliLine('{"type":"assistant","error":null,"message":{"content":[{"type":"text","text":"hello there"}]}}'), null); +}); + +test("trailing toolResults matching a live process continue it: held calls answered, new assistant message streamed, no new spawn", async (t) => { + const readA: ToolUse = { id: "toolu_continue_a", name: "read", arguments: { path: "a" } }; + const { runClaude, calls } = createFakeRun(t, [...toToolUseMessage([readA]), callTool(readA), REPLY]); + const streamFunction = createClaudeCodeStreamFunction(runClaude); + const toolMessage = await runTurn(streamFunction, [HI]); + + const reply = await runTurn(streamFunction, [HI, toolMessage, toToolResult(readA, "contents of a")]); + + assert.equal(toolMessage.stopReason, "toolUse"); + assert.equal(getText(reply), "contents of a"); + await assertCleanedUp(getOnlyCall(calls)); +}); + +test("multiple tool calls in one message are all answered by id", async (t) => { + const readA: ToolUse = { id: "toolu_many_a", name: "read", arguments: { path: "a" } }; + const readB: ToolUse = { id: "toolu_many_b", name: "read", arguments: { path: "b" } }; + const { runClaude, calls } = createFakeRun(t, [...toToolUseMessage([readA, readB]), callTool(readB), callTool(readA), REPLY]); + const streamFunction = createClaudeCodeStreamFunction(runClaude); + const toolMessage = await runTurn(streamFunction, [HI]); + + const reply = await runTurn(streamFunction, [HI, toolMessage, toToolResult(readA, "A"), toToolResult(readB, "B")]); + + assert.equal(getText(reply), "B | A"); + assert.equal(calls.length, 1); +}); + +test("partial id match starts fresh and kills the old process", async (t) => { + const readA: ToolUse = { id: "toolu_partial_a", name: "read", arguments: { path: "a" } }; + const readB: ToolUse = { id: "toolu_partial_b", name: "read", arguments: { path: "b" } }; + const { runClaude, calls } = createFakeRun(t, [...toToolUseMessage([readA, readB]), callTool(readA), callTool(readB), REPLY]); + const streamFunction = createClaudeCodeStreamFunction(runClaude); + const toolMessage = await runTurn(streamFunction, [HI]); + + await runTurn(streamFunction, [HI, toolMessage, toToolResult(readA, "A")]); + + assert.equal(calls.length, 2); + const [oldCall] = calls; + assert.ok(oldCall); + await assertCleanedUp(oldCall); +}); + +test("user message after tool results starts fresh with unfinished calls in the replay", async (t) => { + const stdinFile = join(mkdtempSync(join(tmpdir(), "claude-code-stdin-")), "stdin.jsonl"); + t.after(() => rmSync(dirname(stdinFile), { recursive: true, force: true })); + + const readA: ToolUse = { id: "toolu_steer_a", name: "read", arguments: { path: "a" } }; + const { runClaude, calls } = createFakeRun(t, [...toToolUseMessage([readA]), callTool(readA), REPLY], { FAKE_CLAUDE_STDIN: stdinFile }); + const streamFunction = createClaudeCodeStreamFunction(runClaude); + const toolMessage = await runTurn(streamFunction, [HI]); + const steer: Message = { role: "user", content: "wait, explain sockets first", timestamp: 0 }; + + await runTurn(streamFunction, [HI, toolMessage, toToolResult(readA, "A"), steer]); + + assert.equal(calls.length, 2); + + const [oldCall] = calls; + assert.ok(oldCall); + await assertCleanedUp(oldCall); + + const stdin = readFileSync(stdinFile, "utf8"); + assert.ok(stdin.includes('[tool call read] {\\"path\\":\\"a\\"}'), stdin); + assert.ok(stdin.includes("wait, explain sockets first"), stdin); +}); + +test("output while waiting for tool results kills the process", async (t) => { + const readA: ToolUse = { id: "toolu_moved_on_a", name: "read", arguments: { path: "a" } }; + const { runClaude, calls } = createFakeRun(t, [...toToolUseMessage([readA]), REPLY, HOLD]); + + await runTurn(createClaudeCodeStreamFunction(runClaude), [HI]); + + await assertCleanedUp(getOnlyCall(calls)); +}); + +test("an undeclared tool call is not kept waiting, since the CLI answers it itself", async (t) => { + const readA: ToolUse = { id: "toolu_undeclared_a", name: "read", arguments: { path: "a" } }; + const nope: ToolUse = { id: "toolu_undeclared_nope", name: "nope", arguments: {} }; + const { runClaude, calls } = createFakeRun(t, [...toToolUseMessage([readA, nope]), HOLD]); + + const toolMessage = await runTurn(createClaudeCodeStreamFunction(runClaude), [HI]); + + assert.equal(toolMessage.stopReason, "toolUse"); + await assertCleanedUp(getOnlyCall(calls)); +}); + +test("a tool call with invalid JSON arguments is not kept waiting, since the CLI answers it itself", async (t) => { + const { runClaude, calls } = createFakeRun(t, [ + streamEvent({ type: "message_start", message: { usage: {} } }), + streamEvent({ + type: "content_block_start", + index: 0, + content_block: { type: "tool_use", id: "toolu_bad_json", name: "mcp__learnharness__read", input: {} }, + }), + streamEvent({ type: "content_block_delta", index: 0, delta: { type: "input_json_delta", partial_json: '{"path": ' } }), + streamEvent({ type: "content_block_stop", index: 0 }), + streamEvent({ type: "message_delta", delta: { stop_reason: "tool_use" }, usage: {} }), + streamEvent({ type: "message_stop" }), + HOLD, + ]); + + const toolMessage = await runTurn(createClaudeCodeStreamFunction(runClaude), [HI]); + + assert.equal(toolMessage.stopReason, "toolUse"); + await assertCleanedUp(getOnlyCall(calls)); +}); + +test("abort while waiting kills the process and closes the bridge", async (t) => { + const readA: ToolUse = { id: "toolu_abort_a", name: "read", arguments: { path: "a" } }; + const { runClaude, calls } = createFakeRun(t, [...toToolUseMessage([readA]), callTool(readA), REPLY]); + const controller = new AbortController(); + + await runTurn(createClaudeCodeStreamFunction(runClaude), [HI], { signal: controller.signal }); + + controller.abort(); + + const call = getOnlyCall(calls); + await assertCleanedUp(call); + await assert.rejects(fetch(getMcpUrl(call))); +}); + +test("idle timeout kills an orphaned waiting process", async (t) => { + liveProcessIdleTimeoutMs.current = 50; + t.after(() => (liveProcessIdleTimeoutMs.current = LIVE_PROCESS_IDLE_TIMEOUT_MS)); + + const readA: ToolUse = { id: "toolu_idle_a", name: "read", arguments: { path: "a" } }; + const { runClaude, calls } = createFakeRun(t, [...toToolUseMessage([readA]), callTool(readA), REPLY]); + + await runTurn(createClaudeCodeStreamFunction(runClaude), [HI]); + + await assertCleanedUp(getOnlyCall(calls)); +}); + +test("two concurrent streams keep separate processes and results never cross", async (t) => { + const readOne: ToolUse = { id: "toolu_concurrent_one", name: "read", arguments: { path: "one" } }; + const readTwo: ToolUse = { id: "toolu_concurrent_two", name: "read", arguments: { path: "two" } }; + const one = createFakeRun(t, [...toToolUseMessage([readOne]), callTool(readOne), REPLY]); + const two = createFakeRun(t, [...toToolUseMessage([readTwo]), callTool(readTwo), REPLY]); + + const streamOne = createClaudeCodeStreamFunction(one.runClaude); + const streamTwo = createClaudeCodeStreamFunction(two.runClaude); + const [toolMessageOne, toolMessageTwo] = await Promise.all([runTurn(streamOne, [HI]), runTurn(streamTwo, [HI])]); + + const [replyOne, replyTwo] = await Promise.all([ + runTurn(streamOne, [HI, toolMessageOne, toToolResult(readOne, "from one")]), + runTurn(streamTwo, [HI, toolMessageTwo, toToolResult(readTwo, "from two")]), + ]); + + assert.equal(getText(replyOne), "from one"); + assert.equal(getText(replyTwo), "from two"); + assert.equal(one.calls.length, 1); + assert.equal(two.calls.length, 1); +}); diff --git a/tests/fixtures/claude-code/fake-claude.mjs b/tests/fixtures/claude-code/fake-claude.mjs new file mode 100644 index 0000000..cb64d65 --- /dev/null +++ b/tests/fixtures/claude-code/fake-claude.mjs @@ -0,0 +1,79 @@ +/** + * Stand-in for the claude CLI in tests. Reads the stream-json input from + * stdin, then replays the scenario file named by FAKE_CLAUDE_SCENARIO line by + * line. Lines with a "fake" key are directives instead of output: + * {"fake":"hold"} stay alive until killed, like the CLI waiting on a held MCP call + * {"fake":"exit","code":1,"stderr":"..."} write stderr and exit with that code + * {"fake":"call","name":"read","arguments":{...}} call that tool on the MCP bridge and wait for its result + * {"fake":"reply"} stream one assistant message whose text is every result so far, joined by " | " + * FAKE_CLAUDE_STDIN, when set, receives the stdin it read. + */ +import { readFileSync, writeFileSync } from "node:fs"; + +import { Client } from "@modelcontextprotocol/sdk/client/index.js"; +import { StreamableHTTPClientTransport } from "@modelcontextprotocol/sdk/client/streamableHttp.js"; + +const HOLD_INTERVAL_MS = 60_000; +const BRIDGE_SERVER_NAME = "learnharness"; + +function writeLine(line) { + return new Promise((resolve) => process.stdout.write(`${line}\n`, resolve)); +} + +function streamEvent(event) { + return JSON.stringify({ type: "stream_event", event }); +} + +async function connectBridge(mcpConfig) { + const { url, headers } = JSON.parse(mcpConfig).mcpServers[BRIDGE_SERVER_NAME]; + const client = new Client({ name: "fake-claude", version: "1.0.0" }); + + await client.connect(new StreamableHTTPClientTransport(new URL(url), { requestInit: { headers } })); + return client; +} + +async function writeReply(text) { + await writeLine(streamEvent({ type: "message_start", message: { usage: {} } })); + await writeLine(streamEvent({ type: "content_block_start", index: 0, content_block: { type: "text", text: "" } })); + await writeLine(streamEvent({ type: "content_block_delta", index: 0, delta: { type: "text_delta", text } })); + await writeLine(streamEvent({ type: "content_block_stop", index: 0 })); + await writeLine(streamEvent({ type: "message_delta", delta: { stop_reason: "end_turn" }, usage: {} })); + await writeLine(streamEvent({ type: "message_stop" })); +} + +const mcpConfig = readFileSync(process.argv[process.argv.indexOf("--mcp-config") + 1], "utf8"); + +let input = ""; +for await (const chunk of process.stdin) input += chunk; +if (process.env.FAKE_CLAUDE_STDIN) writeFileSync(process.env.FAKE_CLAUDE_STDIN, input); + +let bridge = null; +const results = []; +const lines = readFileSync(process.env.FAKE_CLAUDE_SCENARIO, "utf8").split("\n").filter(Boolean); +for (const line of lines) { + const directive = JSON.parse(line); + + if (directive.fake === "hold") { + setInterval(() => {}, HOLD_INTERVAL_MS); + break; + } + + if (directive.fake === "exit") { + process.stderr.write(directive.stderr ?? ""); + process.exit(directive.code); + } + + if (directive.fake === "call") { + bridge ??= await connectBridge(mcpConfig); + const result = await bridge.callTool({ name: directive.name, arguments: directive.arguments }); + results.push(result.content.map((block) => block.text).join("")); + continue; + + } + if (directive.fake === "reply") { + await writeReply(results.join(" | ")); + continue; + } + + await writeLine(line); +}