diff --git a/src/agent/claude-code/binary.ts b/src/agent/claude-code/binary.ts index 81b4a55..13e86e6 100644 --- a/src/agent/claude-code/binary.ts +++ b/src/agent/claude-code/binary.ts @@ -21,7 +21,9 @@ function getWindowsSearchSpec(env: BinaryLookup["env"], homeDir: string): Search return { paths, - names: ["claude.exe", "claude.cmd"], + // Direct spawn deliberately uses no shell. A .cmd-only npm install + // is not runnable and must not count as an available provider. + names: ["claude.exe"], knownDirs: [ paths.join(homeDir, ".local", "bin"), paths.join(homeDir, ".claude", "local"), diff --git a/src/agent/claude-code/cli-events.ts b/src/agent/claude-code/cli-events.ts index f532946..f3dddca 100644 --- a/src/agent/claude-code/cli-events.ts +++ b/src/agent/claude-code/cli-events.ts @@ -14,7 +14,8 @@ export type CliEvent = | { type: "messageDelta"; stopReason: string | null; usage: CliUsage } | { type: "messageStop" } | { type: "assistantError"; error: string; text: string } - | { type: "resultError"; text: string }; + | { type: "resultError"; text: string } + | { type: "toolResults"; ids: string[] }; type RawUsage = { input_tokens?: number | null; @@ -46,7 +47,7 @@ type RawCliLine = { type?: string; event?: RawStreamEvent; error?: string | null; - message?: { content?: { type: string; text?: string }[] }; + message?: { content?: { type: string; text?: string; tool_use_id?: string }[] }; is_error?: boolean; result?: string; subtype?: string; @@ -139,7 +140,7 @@ function parseStreamEvent(event: RawStreamEvent): CliEvent | null { } function getText(line: RawCliLine): string { - const blocks = line.message?.content ?? []; + const blocks = Array.isArray(line.message?.content) ? line.message.content : []; const texts = blocks.flatMap((block) => (block.type === "text" && block.text ? [block.text] : [])); return texts.join("\n"); @@ -171,6 +172,12 @@ export function parseCliLine(text: string): CliEvent | null { if (typeof line.error !== "string") return null; return { type: "assistantError", error: line.error, text: getText(line) }; + case "user": { + const blocks = Array.isArray(line.message?.content) ? line.message.content : []; + const ids = blocks.flatMap((block) => block.type === "tool_result" && typeof block.tool_use_id === "string" ? [block.tool_use_id] : []); + return ids.length ? { type: "toolResults", ids } : null; + } + case "result": if (line.is_error !== true) return null; return { type: "resultError", text: getResultErrorText(line) }; diff --git a/src/agent/claude-code/live-processes.ts b/src/agent/claude-code/live-processes.ts index 61f036a..1a68672 100644 --- a/src/agent/claude-code/live-processes.ts +++ b/src/agent/claude-code/live-processes.ts @@ -1,4 +1,4 @@ -export type LiveProcess = { toolCallIds: string[]; kill(): void }; +export type LiveProcess = { readonly owner: symbol; toolCallIds: string[]; kill(): void }; // Tools usually call back within seconds, but a question tool holds its call until the learner // answers. Past this, the process is killed and the answer starts a fresh one with a full replay. @@ -12,6 +12,7 @@ export const liveProcessIdleTimeoutMs = { current: LIVE_PROCESS_IDLE_TIMEOUT_MS const liveProcesses = new Map(); export function addLiveProcess(process: LiveProcess): void { + removeLiveProcess(process); const timer = setTimeout(() => { removeLiveProcess(process); process.kill(); @@ -23,14 +24,14 @@ export function addLiveProcess(process: LiveProcess): void { } function hasSameIds(ids: string[], otherIds: string[]): boolean { - return ids.length === otherIds.length && ids.every((id) => otherIds.includes(id)); + return ids.length === otherIds.length && new Set(otherIds).size === otherIds.length && ids.every((id) => otherIds.includes(id)); } -export function getLiveProcess(toolCallIds: string[]): LiveProcess | null { +export function getLiveProcess(toolCallIds: string[], owner: symbol): LiveProcess | null { if (toolCallIds.length === 0) return null; for (const process of liveProcesses.keys()) { - if (hasSameIds(process.toolCallIds, toolCallIds)) return process; + if (process.owner === owner && hasSameIds(process.toolCallIds, toolCallIds)) return process; } return null; diff --git a/src/agent/claude-code/reuse-context.ts b/src/agent/claude-code/reuse-context.ts new file mode 100644 index 0000000..63cbd7d --- /dev/null +++ b/src/agent/claude-code/reuse-context.ts @@ -0,0 +1,27 @@ +import type { Api, Message, Model, TranscriptContext } from "@earendil-works/pi-ai"; +import { getCurrentSystemPrompt, getCurrentTools } from "@earendil-works/pi-ai/utils/transcript"; + +/** Compare model-visible history, not app IDs, timestamps, usage, or Pi's added thinkingLevel. */ +export function conversationForReuse(messages: Message[]): unknown[] { + return messages.flatMap((message): unknown[] => { + switch (message.role) { + case "system": return []; + case "user": return [{ role: message.role, content: message.content }]; + case "assistant": return [{ role: message.role, content: message.content }]; + case "toolResult": return [{ + role: message.role, + toolCallId: message.toolCallId, + toolName: message.toolName, + content: message.content, + isError: message.isError, + }]; + } + }); +} + +export function configurationForReuse(model: Model, context: TranscriptContext, effort: string): unknown { + const systemPrompt = getCurrentSystemPrompt(context.messages); + const tools = getCurrentTools(context.messages); + // The wire is JSON: TypeBox's schema symbols are not CLI configuration. + return JSON.parse(JSON.stringify({ model, effort, systemPrompt, tools })) as unknown; +} diff --git a/src/agent/claude-code/stream.ts b/src/agent/claude-code/stream.ts index 0f9dd4d..cd27167 100644 --- a/src/agent/claude-code/stream.ts +++ b/src/agent/claude-code/stream.ts @@ -16,11 +16,9 @@ import { 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"; @@ -29,7 +27,8 @@ 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 { configurationForReuse, conversationForReuse } from "./reuse-context.ts"; +import { BRIDGE_SERVER_NAME, createToolBridge, toolNameFromMcp, type PendingToolCall, type ToolBridge } from "./tool-bridge.ts"; import { toClaudeInput, toStreamJsonLine } from "./transcript.ts"; type RunClaudeOptions = { cwd: string; env: NodeJS.ProcessEnv }; @@ -137,6 +136,8 @@ class ClaudeTurn { private readonly output: AssistantMessage; private readonly contentIndexes = new Map(); private readonly partialJson = new Map(); + private readonly toolBlocks = new Map(); + private readonly intents: ToolCall[] = []; started = false; finished = false; @@ -188,13 +189,12 @@ class ClaudeTurn { const found = this.getBlock(event.index); if (!found) return; - const { contentIndex, block } = found; + const { 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; } @@ -208,7 +208,9 @@ class ClaudeTurn { return; case "messageStop": - this.finish(); + // Tool intent is not execution admission. The process publishes only + // calls that actually reach the authenticated MCP bridge. + if (this.output.stopReason !== "toolUse") this.finish(); return; case "assistantError": @@ -221,10 +223,22 @@ class ClaudeTurn { } } - getToolCalls(): ToolCall[] { - if (!this.finished || this.output.stopReason !== "toolUse") return []; + get waitingForAdmission(): boolean { return !this.finished && this.output.stopReason === "toolUse"; } - return this.output.content.filter((block) => block.type === "toolCall"); + getIntents(): ToolCall[] { return this.intents; } + + getMessage(): AssistantMessage { return this.output; } + + admit(calls: ToolCall[]): void { + this.start(); + for (const call of calls) { + const contentIndex = this.output.content.length; + this.output.content.push(call); + this.stream.push({ type: "toolcall_start", contentIndex, partial: this.output }); + this.stream.push({ type: "toolcall_end", contentIndex, toolCall: call, partial: this.output }); + } + this.output.stopReason = "toolUse"; + this.finish(); } fail(message: string, reason: "aborted" | "error"): void { @@ -260,13 +274,18 @@ class ClaudeTurn { 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 }); + if (!block.name.startsWith(`mcp__${BRIDGE_SERVER_NAME}__`)) { + this.fail("claude CLI emitted an unsupported native tool call", "error"); + return; + } + this.toolBlocks.set(index, { type: "toolCall", id: block.id, name: toolNameFromMcp(block.name), arguments: {} }); return; } } private getBlock(index: number): { contentIndex: number; block: AssistantMessage["content"][number] } | null { + const tool = this.toolBlocks.get(index); + if (tool) return { contentIndex: -1, block: tool }; const contentIndex = this.contentIndexes.get(index); if (contentIndex === undefined) return null; @@ -293,7 +312,8 @@ class ClaudeTurn { 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 }); + this.toolBlocks.delete(index); + this.intents.push(block); return; } } @@ -340,51 +360,48 @@ type TurnFiles = { cwd: string; privateDir: string; systemPromptFile: string; mc 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 { + let privateDir: string | undefined; try { - validateToolCall(tools, toolCall); - return true; - } catch { - return false; + 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 }; + } catch (error) { + await Promise.all([rm(cwd, { recursive: true, force: true }), ...(privateDir ? [rm(privateDir, { recursive: true, force: true })] : [])]); + throw error; } } -function isSameToolCall(toolCall: ToolCall, call: PendingToolCall): boolean { - return toolCall.name === call.name && isDeepStrictEqual(toolCall.arguments, call.arguments); +async function removeTurnResources(bridge: ToolBridge, files: TurnFiles): Promise { + await Promise.all([ + bridge.close(), + rm(files.cwd, { recursive: true, force: true }), + rm(files.privateDir, { recursive: true, force: true }), + ]); } class ClaudeProcess implements LiveProcess { toolCallIds: string[] = []; private stderr = ""; private closed = false; + private stopping = false; + private killTimer: NodeJS.Timeout | undefined; private signal: AbortSignal | undefined; + private sourceCalls: ToolCall[] = []; + private sourceComplete = false; + private admitted: PendingToolCall[] = []; + private pendingError: string | undefined; constructor( private readonly child: ChildProcessWithoutNullStreams, private readonly bridge: ToolBridge, private readonly files: TurnFiles, - private readonly tools: Tool[], private turn: ClaudeTurn, + readonly owner: symbol, + private readonly configuration: unknown, + private conversation: unknown[], ) {} start(stdinLine: string, signal: AbortSignal | undefined): void { @@ -394,39 +411,92 @@ class ClaudeProcess implements LiveProcess { 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.stderr.on("data", (chunk: string) => { this.stderr = (this.stderr + chunk).slice(-65_536); }); 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)); - + createInterface({ input: this.child.stdout }).on("line", (line) => { + try { this.handleLine(line); } + catch { this.fail("claude CLI emitted malformed stream output"); } + }); this.setSignal(signal); } - continueTurn(turn: ClaudeTurn, toolResults: ToolResultMessage[], signal: AbortSignal | undefined): void { + canContinue(owner: symbol, configuration: unknown, messages: Message[], results: ToolResultMessage[]): boolean { + const conversation = conversationForReuse(messages); + return !this.stopping && owner === this.owner + && isDeepStrictEqual(configuration, this.configuration) + && isDeepStrictEqual(conversation.slice(0, this.conversation.length), this.conversation) + && conversation.length === this.conversation.length + this.admitted.length + && results.length === this.admitted.length + && new Set(results.map((result) => result.toolCallId)).size === results.length + && this.admitted.every((call) => results.some((result) => result.toolCallId === call.id && result.toolName === call.name)); + } + + continueTurn(turn: ClaudeTurn, messages: Message[], results: 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 }] : []; - }); - + const admitted = this.admitted; + this.admitted = []; + this.conversation = structuredClone(conversationForReuse(messages)); this.turn = turn; + this.sourceComplete = this.sourceCalls.length > 0; 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(); - }); + if (this.stopping) return; + for (const call of admitted) { + const result = results.find((candidate) => candidate.toolCallId === call.id)!; + this.bridge.resolveToolCall(call, result); + } + this.flushAdmissions(); } kill(): void { removeLiveProcess(this); + if (this.stopping || this.closed) return; + this.stopping = true; this.child.kill(); + this.killTimer = setTimeout(() => this.child.kill("SIGKILL"), 1000); + this.killTimer.unref(); + } + + fail(message: string): void { + // A parallel request can fail after the admitted slice was handed to + // Pi. Retain that failure for its continuation instead of silently + // replaying and hiding the compatibility error. + if (this.turn.finished && this.toolCallIds.length > 0) this.pendingError = message; + this.turn.fail(message, "error"); + this.kill(); + if (this.pendingError) addLiveProcess(this); + } + + consumeFailure(owner: symbol, turn: ClaudeTurn): boolean { + if (owner !== this.owner || !this.pendingError) return false; + const message = this.pendingError; + this.pendingError = undefined; + this.kill(); + turn.fail(message, "error"); + return true; + } + + /** Only authenticated MCP requests, never stdout intent alone, admit side effects. */ + flushAdmissions(): void { + if (this.stopping || this.turn.finished || !this.sourceComplete) return; + const pending = this.bridge.getPendingToolCalls(); + if (pending.length === 0) return; + const calls: ToolCall[] = []; + for (const call of pending) { + const intent = this.sourceCalls.find((candidate) => candidate.id === call.id); + if (!intent || intent.name !== call.name || !isDeepStrictEqual(intent.arguments, call.arguments)) { + this.fail("claude CLI compatibility error: MCP call does not match its streamed tool intent"); + return; + } + calls.push(intent); + } + this.sourceCalls = this.sourceCalls.filter((call) => !pending.some((admitted) => admitted.id === call.id)); + this.admitted = pending; + this.toolCallIds = pending.map((call) => call.id); + this.turn.admit(calls); + this.conversation.push(...structuredClone(conversationForReuse([this.turn.getMessage()]))); + addLiveProcess(this); } private readonly abort = () => { @@ -438,96 +508,72 @@ class ClaudeProcess implements LiveProcess { 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 { + if (this.stopping) return; 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 (event.type === "toolResults") { + if (event.ids.some((id) => this.sourceCalls.some((call) => call.id === id))) { + this.fail("claude CLI refused a harness tool call before MCP admission"); } - - 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; + return; + } + // Moving on before held calls were answered means the CLI did not + // honor this tool wait. Never execute further stdout-only intentions. + if (this.turn.finished) { + this.fail("claude CLI moved on while harness tool calls were waiting for results"); + return; + } + if (event.type === "messageStart" && this.sourceComplete && this.sourceCalls.length > 0) { + this.fail("claude CLI moved on without admitting its harness tool calls"); + return; + } + this.turn.handle(event); + if (this.turn.finished) { this.kill(); return; } + if (event.type === "messageStop" && this.turn.waitingForAdmission) { + this.sourceCalls = this.turn.getIntents(); + if (this.sourceCalls.length === 0 || new Set(this.sourceCalls.map((call) => call.id)).size !== this.sourceCalls.length) { + this.fail("claude CLI emitted an invalid tool batch"); + return; + } + this.sourceComplete = true; + this.flushAdmissions(); } } private handleClose(message: string): void { if (this.closed) return; - this.closed = true; + clearTimeout(this.killTimer); removeLiveProcess(this); + if (this.pendingError) addLiveProcess(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 { +function continueLiveProcess(turn: ClaudeTurn, owner: symbol, configuration: unknown, 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); + const ids = assistant.content.flatMap((block) => block.type === "toolCall" ? [block.id] : []); + const process = getLiveProcess(ids, owner); if (!(process instanceof ClaudeProcess)) return false; - + if (process.consumeFailure(owner, turn)) return true; 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) { + const results = trailing.filter((message) => message.role === "toolResult"); + if (trailing.length !== results.length || !process.canContinue(owner, configuration, messages, results)) { process.kill(); return false; } - - process.continueTurn(turn, toolResults, signal); + process.continueTurn(turn, messages, results, signal); return true; } @@ -549,10 +595,28 @@ async function startProcess( model: Model, context: TranscriptContext, options: SimpleStreamOptions | undefined, + owner: symbol, + configuration: unknown, ): Promise { + if (options?.signal?.aborted) { turn.fail(ABORT_MESSAGE, "aborted"); return; } const input = toClaudeInput(context.messages); - const bridge = await createToolBridge(input.tools); - const files = await createTurnFiles(input.systemPrompt, bridge.mcpConfig); + let cliProcess: ClaudeProcess | undefined; + const bridge = await createToolBridge(input.tools, { + onCall: () => cliProcess?.flushAdmissions(), + onError: (error) => cliProcess?.fail(error.message), + }); + let files: TurnFiles; + try { + files = await createTurnFiles(input.systemPrompt, bridge.mcpConfig); + } catch (error) { + await bridge.close(); + throw error; + } + if (options?.signal?.aborted) { + turn.fail(ABORT_MESSAGE, "aborted"); + await removeTurnResources(bridge, files); + return; + } let child: ChildProcessWithoutNullStreams | null; try { @@ -575,17 +639,20 @@ async function startProcess( return; } - new ClaudeProcess(child, bridge, files, input.tools, turn).start(toStreamJsonLine(input.userContent), options?.signal); + cliProcess = new ClaudeProcess(child, bridge, files, turn, owner, configuration, structuredClone(conversationForReuse(context.messages))); + cliProcess.start(toStreamJsonLine(input.userContent), options?.signal); } export function createClaudeCodeStreamFunction(runClaude: RunClaude): StreamFunction { + const owner = Symbol("claude-code stream owner"); return (model, context, options) => { const stream = createAssistantMessageEventStream(); const turn = new ClaudeTurn(stream, model); - if (continueLiveProcess(turn, context.messages, options?.signal)) return stream; + const configuration = configurationForReuse(model, context, toEffort(options?.reasoning)); + if (continueLiveProcess(turn, owner, configuration, context.messages, options?.signal)) return stream; - startProcess(runClaude, turn, model, context, options).catch((error: unknown) => { + startProcess(runClaude, turn, model, context, options, owner, configuration).catch((error: unknown) => { console.error("claude CLI turn failed:", error); turn.fail(START_FAILED_MESSAGE, "error"); }); diff --git a/src/agent/claude-code/tool-bridge.ts b/src/agent/claude-code/tool-bridge.ts index 92e2aad..db9f4da 100644 --- a/src/agent/claude-code/tool-bridge.ts +++ b/src/agent/claude-code/tool-bridge.ts @@ -7,7 +7,9 @@ import { Server } from "@modelcontextprotocol/sdk/server/index.js"; import { StreamableHTTPServerTransport } from "@modelcontextprotocol/sdk/server/streamableHttp.js"; import { CallToolRequestSchema, + ErrorCode, ListToolsRequestSchema, + McpError, type CallToolResult, type Tool as McpTool, } from "@modelcontextprotocol/sdk/types.js"; @@ -18,7 +20,11 @@ const MCP_TOOL_PREFIX = `mcp__${BRIDGE_SERVER_NAME}__`; const LOOPBACK_HOST = "127.0.0.1"; const TOKEN_BYTES = 32; -export type PendingToolCall = { name: string; arguments: JsonObject }; +export type PendingToolCall = { id: string; name: string; arguments: JsonObject }; + +type BridgeCallbacks = { onCall(): void; onError(error: Error): void }; + +export const MCP_IDENTITY_ERROR = "claude CLI compatibility error: MCP calls must include a unique claudecode/toolUseId"; export type ToolBridge = { mcpConfig: string; @@ -104,14 +110,28 @@ class ToolCallQueue { // Low-level Server on purpose: McpServer.registerTool only takes zod schemas, and pi tools carry // JSON Schema that must reach the CLI unchanged. -function createMcpServer(tools: Tool[], queue: ToolCallQueue): Server { +function createMcpServer(tools: Tool[], queue: ToolCallQueue, callbacks?: BridgeCallbacks): Server { const server = new Server({ name: BRIDGE_SERVER_NAME, version: "1.0.0" }, { capabilities: { tools: {} } }); server.setRequestHandler(ListToolsRequestSchema, () => ({ tools: tools.map(toMcpTool) })); + const seenIds = new Set(); server.setRequestHandler(CallToolRequestSchema, (request) => { - const call = { name: request.params.name, arguments: (request.params.arguments ?? {}) as JsonObject }; - return queue.add(call); + const id = request.params._meta?.["claudecode/toolUseId"]; + if (typeof id !== "string" || !id.trim() || seenIds.has(id)) { + const error = new McpError(ErrorCode.InvalidParams, MCP_IDENTITY_ERROR); + callbacks?.onError(error); + throw error; + } + if (!tools.some((tool) => tool.name === request.params.name)) { + const error = new McpError(ErrorCode.InvalidParams, "claude CLI requested an undeclared harness tool"); + callbacks?.onError(error); + throw error; + } + seenIds.add(id); + const result = queue.add({ id, name: request.params.name, arguments: (request.params.arguments ?? {}) as JsonObject }); + callbacks?.onCall(); + return result; }); return server; @@ -167,16 +187,22 @@ function toMcpConfig(port: number, authorization: string): string { return JSON.stringify({ mcpServers: { [BRIDGE_SERVER_NAME]: server } }); } -export async function createToolBridge(tools: Tool[]): Promise { +export async function createToolBridge(tools: Tool[], callbacks?: BridgeCallbacks): Promise { const authorization = `Bearer ${randomBytes(TOKEN_BYTES).toString("hex")}`; const queue = new ToolCallQueue(); - const mcpServer = createMcpServer(tools, queue); + const mcpServer = createMcpServer(tools, queue, callbacks); const transport = new StreamableHTTPServerTransport({ sessionIdGenerator: () => randomUUID() }); await mcpServer.connect(transport); const httpServer = createHttpServer(transport, authorization); - const { port } = await listen(httpServer); + let port: number; + try { + ({ port } = await listen(httpServer)); + } catch (error) { + await mcpServer.close(); + throw error; + } return { mcpConfig: toMcpConfig(port, authorization), diff --git a/src/agent/teacher-agent.ts b/src/agent/teacher-agent.ts index 6fe6ac1..cfdab68 100644 --- a/src/agent/teacher-agent.ts +++ b/src/agent/teacher-agent.ts @@ -17,6 +17,7 @@ import { import { practiceTools, type PracticeToolContext } from "./tools/practice.ts"; import { healToolRecord, withArgumentHeal } from "./tools/heal.ts"; import { sendToUserTool } from "./tools/send-to-user.ts"; +import { CLAUDE_CODE_PROVIDER } from "./claude-code/constants.ts"; import { quizInputObject, quizInputProblems } from "../core/question-core.ts"; import { QUESTION_TOOLS, type EnabledTools } from "../core/pending.ts"; import { toLlmMessages, nativeInputCause, PRACTICE_PROBLEM_TOOL } from "../core/native-messages.ts"; @@ -202,7 +203,7 @@ function baseAgentOptions(kind: AgentKind, opts: AgentBuildOptions, tools: Agent // transport. Explicit test registries remain caller-owned. tools: [ ...tools, - ...(opts.tools === undefined && engine.model.id.toLowerCase().includes("claude") + ...(opts.tools === undefined && (engine.model.provider === CLAUDE_CODE_PROVIDER || engine.model.id.toLowerCase().includes("claude")) ? [withArgumentHeal(sendToUserTool)] : []), ], diff --git a/tests/claude-code-binary.test.ts b/tests/claude-code-binary.test.ts index 66a716a..599f5d9 100644 --- a/tests/claude-code-binary.test.ts +++ b/tests/claude-code-binary.test.ts @@ -39,15 +39,15 @@ test("npm global bin follows npm_config_prefix when set", () => { assert.equal(binary, "/opt/npm/bin/claude"); }); -test("windows looks for claude.cmd and claude.exe", () => { +test("windows finds native claude.exe and does not advertise unsupported .cmd-only installs", () => { const winHome = "C:\\Users\\u"; const env = { PATH: "C:\\tools;C:\\bin", APPDATA: `${winHome}\\AppData\\Roaming` }; - assert.equal(resolveWith(["C:\\bin\\claude.cmd"], env, "win32", winHome), "C:\\bin\\claude.cmd"); + assert.equal(resolveWith(["C:\\bin\\claude.cmd"], env, "win32", winHome), null); assert.equal(resolveWith(["C:\\bin\\claude.exe"], env, "win32", winHome), "C:\\bin\\claude.exe"); assert.equal(resolveWith([`${winHome}\\.local\\bin\\claude.exe`], env, "win32", winHome), `${winHome}\\.local\\bin\\claude.exe`); assert.equal( resolveWith([`${winHome}\\AppData\\Roaming\\npm\\claude.cmd`], env, "win32", winHome), - `${winHome}\\AppData\\Roaming\\npm\\claude.cmd`, + null, ); assert.equal(resolveWith(["C:\\bin\\claude"], env, "win32", winHome), null); }); diff --git a/tests/claude-code-harness.test.ts b/tests/claude-code-harness.test.ts index 1dcd8c2..d300b35 100644 --- a/tests/claude-code-harness.test.ts +++ b/tests/claude-code-harness.test.ts @@ -55,7 +55,7 @@ function toAskScenario(): string[] { streamEvent({ type: "content_block_stop", index: 0 }), streamEvent({ type: "message_delta", delta: { stop_reason: "tool_use" }, usage: {} }), streamEvent({ type: "message_stop" }), - JSON.stringify({ fake: "call", name: "ask_user_question", arguments: ASK_INPUT }), + JSON.stringify({ fake: "call", id: ASK_ID, name: "ask_user_question", arguments: ASK_INPUT }), JSON.stringify({ fake: "reply" }), ]; } @@ -168,6 +168,34 @@ test("a question answered before the idle timeout continues the waiting CLI proc await waitForExit(waiting); }); +test("a question pauses later calls from the same CLI response until the learner answers", async (t) => { + const secondId = "toolu_second_question"; + const secondQuestion = "What do you want to learn next?"; + const scenario = toAskScenario(); + scenario.splice(4, 0, + streamEvent({ type: "content_block_start", index: 1, content_block: { type: "tool_use", id: secondId, name: "mcp__learnharness__ask_user_question", input: {} } }), + streamEvent({ type: "content_block_delta", index: 1, delta: { type: "input_json_delta", partial_json: JSON.stringify({ question: secondQuestion }) } }), + streamEvent({ type: "content_block_stop", index: 1 }), + ); + scenario.splice(scenario.length - 1, 0, JSON.stringify({ fake: "call", id: secondId, name: "ask_user_question", arguments: { question: secondQuestion } })); + const fakeRun = createFakeRun(t, [scenario]); + const factory = createAgentFactory(fakeRun); + const chatId = createClaudeCodeChat(t); + await runChatTurn(chatId, { type: "text", text: "teach me sockets" }, factory); + assert.equal(danglingQuestionOf(chatStore.getChat(chatId)?.messages ?? [])?.toolCallId, ASK_ID); + assert.equal(JSON.stringify(chatStore.getChat(chatId)?.messages).includes(secondId), false, "unadmitted second question must not be posed or persisted"); + await runChatTurn(chatId, toAnswerInput(), factory); + assert.equal(danglingQuestionOf(chatStore.getChat(chatId)?.messages ?? [])?.toolCallId, secondId); + await runChatTurn(chatId, { + type: "answer", toolCallId: secondId, + answer: { kind: "ask", questionText: secondQuestion, dontKnow: false, answers: [], text: "HTTP next", answeredAt: new Date().toISOString() }, + }, factory); + assert.equal(fakeRun.children.length, 1); + assert.ok(getLastAssistantText(chatId).includes(ANSWER_TEXT)); + assert.ok(getLastAssistantText(chatId).includes("HTTP next")); + await waitForExit(fakeRun.children[0]!); +}); + test("a question answered after the idle timeout starts a fresh CLI process that replays the question and answer", async (t) => { liveProcessIdleTimeoutMs.current = 300; t.after(() => (liveProcessIdleTimeoutMs.current = LIVE_PROCESS_IDLE_TIMEOUT_MS)); diff --git a/tests/claude-code-stream.test.ts b/tests/claude-code-stream.test.ts index d5e48b3..efdd322 100644 --- a/tests/claude-code-stream.test.ts +++ b/tests/claude-code-stream.test.ts @@ -29,6 +29,8 @@ import type { ToolResultMessage, } from "@earendil-works/pi-ai"; import { normalizeContext } from "@earendil-works/pi-ai/utils/transcript"; +import { Agent } from "@earendil-works/pi-agent-core"; +import { Type } from "typebox"; import { parseCliLine } from "../src/agent/claude-code/cli-events.ts"; import { ClaudeCodeError } from "../src/agent/claude-code/constants.ts"; @@ -83,7 +85,8 @@ function exitWith(code: number, stderr: string): string { 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]; + return [...lines.slice(0, firstMessageStop + 1).filter((line) => !line.startsWith('{"type":"user"')), + callTool({ id: "toolu_01JBB4cJsPa585FWb1XrfbNX", name: "get_weather", arguments: { city: "Paris" } }), HOLD]; } function createFakeRun( @@ -104,9 +107,9 @@ function createFakeRun( 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()); + stdio: ["pipe", "pipe", "pipe", "ipc"], + }) as ChildProcessWithoutNullStreams; + t.after(() => child.kill("SIGKILL")); calls.push({ args, cwd: options.cwd, @@ -151,7 +154,7 @@ async function runScenario( const context = normalizeContext({ systemPrompt: "You are a tutor.", messages: [{ role: "user", content: "hi", timestamp: 0 }], - tools: [READ_TOOL], + tools: [READ_TOOL, { name: "get_weather", description: "Weather", parameters: { type: "object", properties: {} } }], }); const events = await collectEvents(createClaudeCodeStreamFunction(runClaude)(model, context, options)); @@ -180,7 +183,7 @@ function toToolUseMessage(toolUses: ToolUse[]): string[] { } function callTool(toolUse: ToolUse): string { - return JSON.stringify({ fake: "call", name: toolUse.name, arguments: toolUse.arguments }); + return JSON.stringify({ fake: "call", id: toolUse.id, name: toolUse.name, arguments: toolUse.arguments }); } function toToolResult(toolUse: ToolUse, text: string): ToolResultMessage { @@ -368,7 +371,7 @@ test("tool scenario ends with toolcall_end (Claude's tool_use id, unprefixed nam assert.deepEqual( events.map((event) => event.type), - ["start", "toolcall_start", "toolcall_delta", "toolcall_delta", "toolcall_delta", "toolcall_end", "done"], + ["start", "toolcall_start", "toolcall_end", "done"], ); const toolCallEnd = events.find((event) => event.type === "toolcall_end"); @@ -386,7 +389,7 @@ test("tool scenario ends with toolcall_end (Claude's tool_use id, unprefixed nam assert.deepEqual(done.message.content, [toolCallEnd.toolCall]); }); -test("unknown tool name and bad JSON args still surface as tool calls", async (t) => { +test("unsupported native tool intent fails without exposing executable calls", async (t) => { const { events } = await runScenario(t, [ streamEvent({ type: "message_start", message: { usage: {} } }), streamEvent({ @@ -404,21 +407,8 @@ test("unknown tool name and bad JSON args still surface as tool calls", async (t 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"); + assert.equal(events.some((event) => event.type === "toolcall_end"), false); + assert.match(getErrorMessage(events) ?? "", /unsupported native tool/); }); test("usage maps input/output/cacheRead/cacheWrite with zero cost, message_delta overriding message_start", async (t) => { @@ -585,7 +575,7 @@ test("spawns with the model alias, effort, bridge tools, MCP_TOOL_TIMEOUT and th 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(getFlagValue(call.args, "--allowedTools"), "mcp__learnharness__read,mcp__learnharness__get_weather"); 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); @@ -641,27 +631,35 @@ test("trailing toolResults matching a live process continue it: held calls answe await assertCleanedUp(getOnlyCall(calls)); }); -test("multiple tool calls in one message are all answered by id", async (t) => { +test("sequential identical calls split one response and correlate results by id, not arguments", 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 readB: ToolUse = { id: "toolu_many_b", name: "read", arguments: { path: "a" } }; + const source = toToolUseMessage([readA, readB]); + source[0] = streamEvent({ type: "message_start", message: { usage: { input_tokens: 7, output_tokens: 2 } } }); + const { runClaude, calls } = createFakeRun(t, [...source, 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.deepEqual(toolMessage.content.filter((block) => block.type === "toolCall").map((call) => call.id), [readB.id]); + const history = [HI, toolMessage, toToolResult(readB, "B")]; + const second = await runTurn(streamFunction, history); + assert.deepEqual(second.content.filter((block) => block.type === "toolCall").map((call) => call.id), [readA.id]); + const reply = await runTurn(streamFunction, [...history, second, toToolResult(readA, "A")]); assert.equal(getText(reply), "B | A"); + assert.equal(toolMessage.usage.totalTokens, 9); + assert.equal(second.usage.totalTokens, 0, "splitting one response must not charge its tokens twice"); assert.equal(calls.length, 1); }); -test("partial id match starts fresh and kills the old process", async (t) => { +test("mismatched result id 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")]); + await runTurn(streamFunction, [HI, toolMessage, toToolResult(readB, "B")]); assert.equal(calls.length, 2); const [oldCall] = calls; @@ -695,24 +693,23 @@ test("user message after tool results starts fresh with unfinished calls in the 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]); - + const events = await collectEvents(createClaudeCodeStreamFunction(runClaude)(getModel(), normalizeContext({ messages: [HI], tools: [READ_TOOL] }))); + assert.match(getErrorMessage(events) ?? "", /moved on without admitting/); + assert.equal(events.some((event) => event.type === "toolcall_end"), false); await assertCleanedUp(getOnlyCall(calls)); }); -test("an undeclared tool call is not kept waiting, since the CLI answers it itself", async (t) => { +test("unadmitted tool intent never becomes an executable call", 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"); + const { runClaude, calls } = createFakeRun(t, [...toToolUseMessage([readA, nope]), exitWith(1, "tool denied")]); + const events = await collectEvents(createClaudeCodeStreamFunction(runClaude)(getModel(), normalizeContext({ messages: [HI], tools: [READ_TOOL] }))); + assert.equal(events.some((event) => event.type === "toolcall_end"), false); + assert.match(getErrorMessage(events) ?? "", /exited with code 1/); await assertCleanedUp(getOnlyCall(calls)); }); -test("a tool call with invalid JSON arguments is not kept waiting, since the CLI answers it itself", async (t) => { +test("a CLI-rejected malformed call is not exposed for execution", async (t) => { const { runClaude, calls } = createFakeRun(t, [ streamEvent({ type: "message_start", message: { usage: {} } }), streamEvent({ @@ -724,12 +721,13 @@ test("a tool call with invalid JSON arguments is not kept waiting, since the CLI streamEvent({ type: "content_block_stop", index: 0 }), streamEvent({ type: "message_delta", delta: { stop_reason: "tool_use" }, usage: {} }), streamEvent({ type: "message_stop" }), + JSON.stringify({ type: "user", message: { content: [{ type: "tool_result", tool_use_id: "toolu_bad_json", is_error: true, content: "invalid arguments" }] } }), HOLD, ]); - const toolMessage = await runTurn(createClaudeCodeStreamFunction(runClaude), [HI]); - - assert.equal(toolMessage.stopReason, "toolUse"); + const events = await collectEvents(createClaudeCodeStreamFunction(runClaude)(getModel(), normalizeContext({ messages: [HI], tools: [READ_TOOL] }))); + assert.equal(events.some((event) => event.type === "toolcall_end"), false); + assert.match(getErrorMessage(events) ?? "", /refused.*before MCP admission/); await assertCleanedUp(getOnlyCall(calls)); }); @@ -805,9 +803,140 @@ test("a tool result arriving late but before the idle timeout still continues th await assertCleanedUp(getOnlyCall(calls)); }); -test("two concurrent streams keep separate processes and results never cross", async (t) => { +test("CLI denial cannot execute a tool in the native Pi loop", async (t) => { + const read: ToolUse = { id: "toolu_denied", name: "read", arguments: { path: "a" } }; + const { runClaude, calls } = createFakeRun(t, [...toToolUseMessage([read]), JSON.stringify({ type: "user", message: { content: [ + { type: "tool_result", tool_use_id: read.id, is_error: true, content: "Denied by managed policy" }, + ] } }), HOLD]); + let executed = 0; + const agent = new Agent({ initialState: { model: getModel(), tools: [{ + name: "read", label: "Read", description: "Read", parameters: Type.Object({ path: Type.String() }), + execute: async () => { executed++; return { content: [{ type: "text", text: "must not run" }], details: null }; }, + }] }, streamFn: createClaudeCodeStreamFunction(runClaude) }); + await agent.prompt("hi"); + assert.equal(executed, 0); + const last = agent.state.messages.at(-1); + assert.ok(last?.role === "assistant"); + assert.equal(last.stopReason, "error"); + assert.match(last.errorMessage ?? "", /refused.*before MCP admission/); + await assertCleanedUp(getOnlyCall(calls)); +}); + +test("pre-aborted requests do not spawn a process", async () => { + const controller = new AbortController(); + controller.abort(); + let spawned = false; + const stream = createClaudeCodeStreamFunction(() => { spawned = true; return null; }); + const events = await collectEvents(stream(getModel(), normalizeContext({ messages: [HI] }), { signal: controller.signal })); + assert.equal(spawned, false); + assert.match(getErrorMessage(events) ?? "", /aborted/); +}); + +test("missing MCP identity fails closed without emitting executable calls", async (t) => { + const read: ToolUse = { id: "toolu_no_meta", name: "read", arguments: { path: "a" } }; + const { runClaude, calls } = createFakeRun(t, [...toToolUseMessage([read]), JSON.stringify({ fake: "call", name: read.name, arguments: read.arguments })]); + const events = await collectEvents(createClaudeCodeStreamFunction(runClaude)(getModel(), normalizeContext({ messages: [HI], tools: [READ_TOOL] }))); + assert.equal(events.some((event) => event.type === "toolcall_end"), false); + assert.match(getErrorMessage(events) ?? "", /compatibility error.*claudecode\/toolUseId/); + await assertCleanedUp(getOnlyCall(calls)); +}); + +test("a parallel identity failure after an admitted slice is reported on continuation, not hidden by replay", async (t) => { + const first: ToolUse = { id: "toolu_good_parallel", name: "read", arguments: { path: "a" } }; + const second: ToolUse = { id: "toolu_bad_parallel", name: "read", arguments: { path: "b" } }; + const { runClaude, calls } = createFakeRun(t, [...toToolUseMessage([first, second]), + JSON.stringify({ fake: "start-call", id: first.id, name: first.name, arguments: first.arguments }), + JSON.stringify({ fake: "wait-for-message" }), + JSON.stringify({ fake: "call", name: second.name, arguments: second.arguments }), HOLD]); + const stream = createClaudeCodeStreamFunction(runClaude); + const message = await runTurn(stream, [HI]); + getOnlyCall(calls).child.send("continue"); + await waitForExit(getOnlyCall(calls).child); + const events = await collectEvents(stream(getModel(), normalizeContext({ systemPrompt: "You are a tutor.", messages: [HI, message, toToolResult(first, "A")], tools: [READ_TOOL] }))); + assert.match(getErrorMessage(events) ?? "", /compatibility error.*claudecode\/toolUseId/); + assert.equal(calls.length, 1, "must not hide the compatibility failure by spawning again"); + await assertCleanedUp(getOnlyCall(calls)); +}); + +test("output after MCP admission fails continuation instead of silently replaying", async (t) => { + const call: ToolUse = { id: "toolu_moved_on", name: "read", arguments: { path: "a" } }; + const { runClaude, calls } = createFakeRun(t, [...toToolUseMessage([call]), + JSON.stringify({ fake: "start-call", id: call.id, name: call.name, arguments: call.arguments }), + JSON.stringify({ fake: "wait-for-message" }), ...toToolUseMessage([call]), HOLD]); + const stream = createClaudeCodeStreamFunction(runClaude); + const message = await runTurn(stream, [HI]); + getOnlyCall(calls).child.send("continue"); + await waitForExit(getOnlyCall(calls).child); + const events = await collectEvents(stream(getModel(), normalizeContext({ systemPrompt: "You are a tutor.", messages: [HI, message, toToolResult(call, "A")], tools: [READ_TOOL] }))); + assert.match(getErrorMessage(events) ?? "", /moved on while.*waiting for results/); + assert.equal(calls.length, 1); + await assertCleanedUp(getOnlyCall(calls)); +}); + +test("normal completion escalates termination when the child ignores SIGTERM", async (t) => { + const { events, calls } = await runScenario(t, [JSON.stringify({ fake: "ignore-term" }), ...readFixtureLines("text.jsonl"), HOLD]); + assert.equal(getLastEvent(events).type, "done"); + await assertCleanedUp(getOnlyCall(calls)); + assert.equal(getOnlyCall(calls).child.signalCode, "SIGKILL"); +}); + +test("a synchronous spawn failure removes both temp directories and the bridge", async () => { + let resources: { cwd: string; privateDir: string; url: string } | undefined; + const stream = createClaudeCodeStreamFunction((args, options) => { + const config = JSON.parse(readFileSync(getFlagValue(args, "--mcp-config"), "utf8")); + resources = { cwd: options.cwd, privateDir: dirname(getFlagValue(args, "--mcp-config")), url: config.mcpServers.learnharness.url }; + throw new Error("synthetic spawn failure"); + }); + const events = await collectEvents(stream(getModel(), normalizeContext({ messages: [HI] }))); + assert.match(getErrorMessage(events) ?? "", /failed to start/); + assert.ok(resources); + assert.equal(existsSync(resources.cwd), false); + assert.equal(existsSync(resources.privateDir), false); + await assert.rejects(fetch(resources.url)); +}); + +test("a malformed JSON event shape fails the turn rather than escaping the readline handler", async (t) => { + const { events, calls } = await runScenario(t, [JSON.stringify({ type: "assistant", error: "bad", message: { content: [null] } }), HOLD]); + assert.match(getErrorMessage(events) ?? "", /malformed stream output/); + await assertCleanedUp(getOnlyCall(calls)); +}); + +test("a mismatched MCP identity cannot admit a streamed call", async (t) => { + const read: ToolUse = { id: "toolu_intent", name: "read", arguments: { path: "a" } }; + const { runClaude, calls } = createFakeRun(t, [...toToolUseMessage([read]), callTool({ ...read, id: "toolu_someone_else" })]); + const events = await collectEvents(createClaudeCodeStreamFunction(runClaude)(getModel(), normalizeContext({ messages: [HI], tools: [READ_TOOL] }))); + assert.equal(events.some((event) => event.type === "toolcall_end"), false); + assert.match(getErrorMessage(events) ?? "", /does not match its streamed tool intent/); + await assertCleanedUp(getOnlyCall(calls)); +}); + +for (const change of ["model", "effort", "system prompt", "tool schema", "tool description", "history"] as const) { + test(`a changed ${change} prevents reuse of a held process`, async (t) => { + const read: ToolUse = { id: `toolu_reuse_${change}`, name: "read", arguments: { path: "a" } }; + const waiting = createFakeRun(t, [...toToolUseMessage([read]), callTool(read), REPLY]); + const fresh = createFakeRun(t, readFixtureLines("text.jsonl")); + const stream = createClaudeCodeStreamFunction((args, options) => (waiting.calls.length === 0 ? waiting : fresh).runClaude(args, options)); + const message = await runTurn(stream, [HI]); + const messages = [change === "history" ? { ...HI, content: "replacement history after compaction" } : HI, message, toToolResult(read, "A")]; + const context = normalizeContext({ + systemPrompt: change === "system prompt" ? "Changed tutor instructions" : "You are a tutor.", + messages, + tools: [{ ...READ_TOOL, + ...(change === "tool schema" ? { parameters: { type: "object", properties: { path: { type: "string" } } } } : {}), + ...(change === "tool description" ? { description: "Changed description" } : {}), + }], + }); + const model = change === "model" ? CLAUDE_CODE_MODELS[1]! : getModel(); + const events = await collectEvents(stream(model, context, change === "effort" ? { reasoning: "high" } : undefined)); + assert.equal(getLastEvent(events).type, "done"); + assert.equal(fresh.calls.length, 1, "must start a fresh CLI rather than silently use stale configuration"); + await assertCleanedUp(getOnlyCall(waiting.calls)); + }); +} + +for (const sharedIdentity of [false, true]) test(`two concurrent stream owners keep separate processes${sharedIdentity ? " even with identical call IDs and arguments" : ""}`, 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 readTwo: ToolUse = sharedIdentity ? { ...readOne } : { 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]); diff --git a/tests/claude-code-tool-bridge.test.ts b/tests/claude-code-tool-bridge.test.ts index 3d18881..9b44dfb 100644 --- a/tests/claude-code-tool-bridge.test.ts +++ b/tests/claude-code-tool-bridge.test.ts @@ -107,13 +107,13 @@ test("tools/call stays pending until resolved and returns its text and image con t.after(() => client.close()); let settled = false; - const pending = client.callTool({ name: "read", arguments: { path: "notes.md" } }).finally(() => { + const pending = client.callTool({ name: "read", arguments: { path: "notes.md" }, _meta: { "claudecode/toolUseId": "call-1" } }).finally(() => { settled = true; }); const [call] = await bridge.waitForToolCalls(1); assert.ok(call); - assert.deepEqual(call, { name: "read", arguments: { path: "notes.md" } }); + assert.deepEqual(call, { id: "call-1", name: "read", arguments: { path: "notes.md" } }); assert.deepEqual(bridge.getPendingToolCalls(), [call]); await new Promise((resolve) => setTimeout(resolve, 20)); @@ -147,15 +147,15 @@ test("waitForToolCalls resolves once that many calls are pending, in arrival ord waited = true; }); - void client.callTool({ name: "read", arguments: { path: "a.md" } }).catch(() => {}); + void client.callTool({ name: "read", arguments: { path: "a.md" }, _meta: { "claudecode/toolUseId": "call-1" } }).catch(() => {}); await new Promise((resolve) => setTimeout(resolve, 20)); assert.equal(waited, false); - void client.callTool({ name: "screenshot", arguments: {} }).catch(() => {}); + void client.callTool({ name: "screenshot", arguments: {}, _meta: { "claudecode/toolUseId": "call-2" } }).catch(() => {}); assert.deepEqual(await waiting, [ - { name: "read", arguments: { path: "a.md" } }, - { name: "screenshot", arguments: {} }, + { id: "call-1", name: "read", arguments: { path: "a.md" } }, + { id: "call-2", name: "screenshot", arguments: {} }, ]); }); @@ -165,7 +165,7 @@ test("isError result becomes MCP isError true", async (t) => { const client = await connectClient(bridge); t.after(() => client.close()); - const pending = client.callTool({ name: "read", arguments: { path: "missing.md" } }); + const pending = client.callTool({ name: "read", arguments: { path: "missing.md" }, _meta: { "claudecode/toolUseId": "call-1" } }); const [call] = await bridge.waitForToolCalls(1); assert.ok(call); @@ -180,7 +180,7 @@ test("close fails pending calls and frees the port", async () => { const bridge = await createToolBridge([READ_TOOL]); const port = Number(new URL(getServerConfig(bridge).url).port); const client = await connectClient(bridge); - const pending = client.callTool({ name: "read", arguments: { path: "notes.md" } }); + const pending = client.callTool({ name: "read", arguments: { path: "notes.md" }, _meta: { "claudecode/toolUseId": "call-1" } }); await bridge.waitForToolCalls(1); await bridge.close(); @@ -219,6 +219,25 @@ test('mcpConfig is {"mcpServers":{"learnharness":{"type":"http","url":...,"heade assert.match(server.headers.Authorization, /^Bearer [0-9a-f]{64}$/); }); +test("missing and reused Claude tool IDs are rejected rather than matched by arguments", async (t) => { + const failures: string[] = []; + const bridge = await createToolBridge([READ_TOOL], { onCall: () => {}, onError: (error) => failures.push(error.message) }); + t.after(() => bridge.close()); + const client = await connectClient(bridge); + t.after(() => client.close()); + await assert.rejects(client.callTool({ name: "read", arguments: { path: "a" } }), /claudecode\/toolUseId/); + assert.deepEqual(bridge.getPendingToolCalls(), []); + const params = { name: "read", arguments: { path: "a" }, _meta: { "claudecode/toolUseId": "same-id" } }; + const first = client.callTool(params); + const [call] = await bridge.waitForToolCalls(1); + assert.ok(call); + bridge.resolveToolCall(call, toolResult([{ type: "text", text: "first" }])); + await first; + await assert.rejects(client.callTool(params), /claudecode\/toolUseId/); + assert.deepEqual(bridge.getPendingToolCalls(), []); + assert.equal(failures.length, 2); +}); + test("toolNameFromMcp strips the bridge prefix", () => { assert.equal(toolNameFromMcp("mcp__learnharness__read"), "read"); assert.equal(toolNameFromMcp("read"), "read"); diff --git a/tests/fixtures/claude-code/fake-claude.mjs b/tests/fixtures/claude-code/fake-claude.mjs index cb64d65..d1fef86 100644 --- a/tests/fixtures/claude-code/fake-claude.mjs +++ b/tests/fixtures/claude-code/fake-claude.mjs @@ -4,11 +4,15 @@ * 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":"call","id":"toolu_...","name":"read","arguments":{...}} call the MCP bridge and wait for its result + * {"fake":"start-call",...} start the same request without waiting for its result + * {"fake":"wait-for-message"} pause until the test releases the child over IPC + * {"fake":"ignore-term"} ignore SIGTERM to exercise kill escalation * {"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 { once } from "node:events"; import { Client } from "@modelcontextprotocol/sdk/client/index.js"; import { StreamableHTTPClientTransport } from "@modelcontextprotocol/sdk/client/streamableHttp.js"; @@ -49,10 +53,15 @@ if (process.env.FAKE_CLAUDE_STDIN) writeFileSync(process.env.FAKE_CLAUDE_STDIN, let bridge = null; const results = []; +const pendingResults = []; 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 === "wait-for-message") { + await once(process, "message"); + continue; + } if (directive.fake === "hold") { setInterval(() => {}, HOLD_INTERVAL_MS); break; @@ -63,14 +72,28 @@ for (const line of lines) { process.exit(directive.code); } - if (directive.fake === "call") { + if (directive.fake === "ignore-term") { + process.on("SIGTERM", () => {}); + continue; + } + if (directive.fake === "call" || directive.fake === "start-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("")); + const request = bridge.callTool({ + name: directive.name, + arguments: directive.arguments, + ...(directive.id !== undefined ? { _meta: { "claudecode/toolUseId": directive.id } } : {}), + }); + if (directive.fake === "start-call") { + pendingResults.push(request); + void request.catch(() => {}); + } else { + const result = await request; + results.push(result.content.map((block) => block.text).join("")); + } continue; - } if (directive.fake === "reply") { + for (const result of await Promise.all(pendingResults.splice(0))) results.push(result.content.map((block) => block.text).join("")); await writeReply(results.join(" | ")); continue; } diff --git a/tests/send-to-user.test.ts b/tests/send-to-user.test.ts index beaeae3..80109d4 100644 --- a/tests/send-to-user.test.ts +++ b/tests/send-to-user.test.ts @@ -88,6 +88,9 @@ describe("send_to_user model gate", () => { ["openrouter", "anthropic/claude-sonnet-5.5", true], ["amazon-bedrock", "us.anthropic.claude-opus-5-5-v1:0", true], ["fixture", "CLAUDE-fixture", true], + ["claude-code", "opus", true], + ["claude-code", "sonnet", true], + ["claude-code", "haiku", true], ["openrouter", "z-ai/glm-5.3-flash", false], ["openai", "gpt-4o-mini", false], ["anthropic", "opaque-alias", false],