feat: stream turns through the claude CLI
Each turn runs claude -p with stream-json in and out and maps its events onto pi's assistant message stream. A process that stops on tool calls is kept alive and continued when pi sends the results back, instead of replaying the whole transcript. API credentials are removed from the CLI environment so turns always bill to the signed-in subscription.
This commit is contained in:
Ivanko1608
@@ -0,0 +1,181 @@
|
||||
import type { Usage } from "@earendil-works/pi-ai";
|
||||
|
||||
export type CliUsage = Partial<Pick<Usage, "input" | "output" | "cacheRead" | "cacheWrite">>;
|
||||
|
||||
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;
|
||||
}
|
||||
}
|
||||
@@ -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<LiveProcess, NodeJS.Timeout>();
|
||||
|
||||
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);
|
||||
}
|
||||
@@ -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<ThinkingLevel, AnthropicEffort> = {
|
||||
minimal: "low",
|
||||
low: "low",
|
||||
medium: "medium",
|
||||
high: "high",
|
||||
xhigh: "xhigh",
|
||||
max: "max",
|
||||
};
|
||||
|
||||
const STOP_REASONS: Record<string, StopReason> = {
|
||||
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<Api>): 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<number, number>();
|
||||
private readonly partialJson = new Map<number, string>();
|
||||
started = false;
|
||||
finished = false;
|
||||
|
||||
constructor(
|
||||
private readonly stream: AssistantMessageEventStream,
|
||||
model: Model<Api>,
|
||||
) {
|
||||
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<TurnFiles> {
|
||||
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<void> {
|
||||
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<void> {
|
||||
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<Api>,
|
||||
context: TranscriptContext,
|
||||
options: SimpleStreamOptions | undefined,
|
||||
): Promise<void> {
|
||||
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<Api, SimpleStreamOptions> {
|
||||
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);
|
||||
@@ -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<string, string> = {},
|
||||
): { 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<AssistantMessageEvent[]> {
|
||||
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<string, string>,
|
||||
): Promise<ScenarioRun> {
|
||||
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<Api, SimpleStreamOptions>,
|
||||
messages: Message[],
|
||||
options?: SimpleStreamOptions,
|
||||
): Promise<AssistantMessage> {
|
||||
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<string, { url: string }> };
|
||||
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<void> {
|
||||
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<void> {
|
||||
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<void> {
|
||||
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<string, { url: string }> };
|
||||
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);
|
||||
});
|
||||
+79
@@ -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);
|
||||
}
|
||||
Reference in New Issue
Block a user