root/fusion
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(FN-2938): merge fusion/fn-2938

Browse Source 
- fix(FN-2938): satisfy lint for control-request debug logging
- test(FN-2938): complete Step 4 — update protocol regression tests
- feat(FN-2938): complete Step 3 — remove stdin control request routing
- feat(FN-2938): complete Step 2 — make control handler pure
- feat(FN-2938): complete Step 1 — fix stdin EOF and spawn flags
- feat(FN-2929): merge fusion/fn-2929

Fusion-Task-Id: FN-2938
This commit is contained in:
Fusion
2026-04-29 02:06:05 -07:00
committed by gsxdsm
parent 10d565e344
commit e4dee503c5
6 changed files with 161 additions and 236 deletions
Download Patch File Download Diff File Expand all files Collapse all files

105
packages/pi-claude-cli/src/__tests__/control-handler.test.ts
View File

@@ -1,5 +1,4 @@
import { describe, it, expect, vi } from "vitest";
import { PassThrough } from "node:stream";
import type { ClaudeControlRequest } from "../types";
import {
handleControlRequest,
@@ -7,13 +6,6 @@ import {
MCP_PREFIX,
} from "../control-handler";
function createMockStdin() {
const stream = new PassThrough();
const chunks: string[] = [];
stream.on("data", (data: Buffer) => chunks.push(data.toString()));
return { stream, chunks };
}
function makeControlRequest(
toolName: string,
requestId = "req-test-001",
@@ -44,147 +36,128 @@ describe("control-handler", () => {
});
describe("denies custom MCP tools (mcp__custom-tools__*)", () => {
it("denies mcp__custom-tools__weather and returns false", () => {
const { stream, chunks } = createMockStdin();
it("denies mcp__custom-tools__weather", () => {
const msg = makeControlRequest("mcp__custom-tools__weather");
const result = handleControlRequest(msg, stream);
const result = handleControlRequest(msg);
expect(result).toBe(false);
const response = JSON.parse(chunks[0].trim());
expect(response.response.response.behavior).toBe("deny");
expect(response.response.response.message).toBe(
expect(result.allowed).toBe(false);
expect(result.response.response.response.behavior).toBe("deny");
expect(result.response.response.response.message).toBe(
TOOL_EXECUTION_DENIED_MESSAGE,
);
});
it("denies mcp__custom-tools__deploy", () => {
const { stream, chunks } = createMockStdin();
const msg = makeControlRequest("mcp__custom-tools__deploy");
const result = handleControlRequest(msg, stream);
const result = handleControlRequest(msg);
expect(result).toBe(false);
const response = JSON.parse(chunks[0].trim());
expect(response.response.response.behavior).toBe("deny");
expect(result.allowed).toBe(false);
expect(result.response.response.response.behavior).toBe("deny");
});
});
describe("allows user MCP tools and other tools", () => {
it("allows user MCP tool mcp__database__query and returns true", () => {
const { stream, chunks } = createMockStdin();
it("allows user MCP tool mcp__database__query", () => {
const msg = makeControlRequest("mcp__database__query");
const result = handleControlRequest(msg, stream);
const result = handleControlRequest(msg);
expect(result).toBe(true);
const response = JSON.parse(chunks[0].trim());
expect(response.response.response.behavior).toBe("allow");
expect(result.allowed).toBe(true);
expect(result.response.response.response.behavior).toBe("allow");
});
it("allows built-in tool Read", () => {
const { stream, chunks } = createMockStdin();
const msg = makeControlRequest("Read");
const result = handleControlRequest(msg, stream);
const result = handleControlRequest(msg);
expect(result).toBe(true);
const response = JSON.parse(chunks[0].trim());
expect(response.response.response.behavior).toBe("allow");
expect(result.allowed).toBe(true);
expect(result.response.response.response.behavior).toBe("allow");
});
it("allows internal tools like ToolSearch", () => {
const { stream, chunks } = createMockStdin();
const msg = makeControlRequest("ToolSearch");
const result = handleControlRequest(msg, stream);
const result = handleControlRequest(msg);
expect(result).toBe(true);
const response = JSON.parse(chunks[0].trim());
expect(response.response.response.behavior).toBe("allow");
expect(result.allowed).toBe(true);
expect(result.response.response.response.behavior).toBe("allow");
});
it("allows unknown tools", () => {
const { stream, chunks } = createMockStdin();
const msg = makeControlRequest("SomeUnknownTool");
const result = handleControlRequest(msg, stream);
const result = handleControlRequest(msg);
expect(result).toBe(true);
const response = JSON.parse(chunks[0].trim());
expect(response.response.response.behavior).toBe("allow");
expect(result.allowed).toBe(true);
expect(result.response.response.response.behavior).toBe("allow");
});
});
describe("response format", () => {
it("includes matching request_id", () => {
const { stream, chunks } = createMockStdin();
const msg = makeControlRequest("Read", "custom-req-id-42");
handleControlRequest(msg, stream);
const result = handleControlRequest(msg);
const response = JSON.parse(chunks[0].trim());
expect(response.request_id).toBe("custom-req-id-42");
expect(result.response.request_id).toBe("custom-req-id-42");
});
it("writes response as NDJSON (JSON + newline)", () => {
const { stream, chunks } = createMockStdin();
it("returns a JSON-serializable response object", () => {
const msg = makeControlRequest("Read");
handleControlRequest(msg, stream);
const result = handleControlRequest(msg);
const serialized = JSON.stringify(result.response);
expect(chunks[0].endsWith("\n")).toBe(true);
expect(() => JSON.parse(chunks[0].trim())).not.toThrow();
expect(() => JSON.parse(serialized)).not.toThrow();
});
it("deny response includes message field", () => {
const { stream, chunks } = createMockStdin();
const msg = makeControlRequest("mcp__custom-tools__foo");
handleControlRequest(msg, stream);
const result = handleControlRequest(msg);
const response = JSON.parse(chunks[0].trim());
expect(response.response.response.message).toBe(
expect(result.response.response.response.message).toBe(
TOOL_EXECUTION_DENIED_MESSAGE,
);
});
it("allow response does not include a message field", () => {
const { stream, chunks } = createMockStdin();
const msg = makeControlRequest("mcp__database__query");
handleControlRequest(msg, stream);
const result = handleControlRequest(msg);
const response = JSON.parse(chunks[0].trim());
expect(response.response.response.message).toBeUndefined();
expect(result.response.response.response.message).toBeUndefined();
});
});
describe("malformed input", () => {
it("returns false for missing request_id", () => {
const { stream } = createMockStdin();
it("returns denied decision object for missing request_id", () => {
const spy = vi.spyOn(console, "error").mockImplementation(() => {});
const msg = {
type: "control_request",
} as unknown as ClaudeControlRequest;
const result = handleControlRequest(msg, stream);
const result = handleControlRequest(msg);
expect(result).toBe(false);
expect(result.allowed).toBe(false);
expect(result.response.response.response.behavior).toBe("deny");
spy.mockRestore();
});
it("returns false for missing request object", () => {
const { stream } = createMockStdin();
it("returns denied decision object for missing request object", () => {
const spy = vi.spyOn(console, "error").mockImplementation(() => {});
const msg = {
type: "control_request",
request_id: "req-001",
} as unknown as ClaudeControlRequest;
const result = handleControlRequest(msg, stream);
const result = handleControlRequest(msg);
expect(result).toBe(false);
expect(result.allowed).toBe(false);
expect(result.response.response.response.behavior).toBe("deny");
spy.mockRestore();
});
});

16
packages/pi-claude-cli/src/__tests__/process-manager.test.ts
View File

@@ -105,8 +105,8 @@ describe("spawnClaude", () => {
expect(args).not.toContain("--no-session-persistence");
expect(args).toContain("--model");
expect(args).toContain("claude-sonnet-4-5-20250929");
expect(args).toContain("--permission-prompt-tool");
expect(args).toContain("stdio");
expect(args).not.toContain("--permission-prompt-tool");
expect(args).not.toContain("stdio");
});
it("passes stream-json for both input-format and output-format", () => {
@@ -255,13 +255,13 @@ describe("writeUserMessage", () => {
expect(written.endsWith("\n")).toBe(true);
});
it("does NOT call stdin.end()", () => {
it("calls stdin.end() after writing user message", () => {
const mockStdin = { write: vi.fn(), end: vi.fn() };
const proc = { stdin: mockStdin } as unknown as ChildProcess;
writeUserMessage(proc, "test");
expect(mockStdin.end).not.toHaveBeenCalled();
expect(mockStdin.end).toHaveBeenCalledTimes(1);
});
it("sends string content in NDJSON when given string", () => {
@@ -420,13 +420,11 @@ describe("CLI flags", () => {
expect(args).not.toContain("dontAsk");
});
it("spawnClaude includes --permission-prompt-tool followed by stdio in args", () => {
it("spawnClaude does NOT include --permission-prompt-tool in args", () => {
spawnClaude("claude-sonnet-4-5-20250929");
const args = (spawn as any).mock.calls[0][1] as string[];
expect(args).toContain("--permission-prompt-tool");
const idx = args.indexOf("--permission-prompt-tool");
expect(args[idx + 1]).toBe("stdio");
expect(args).not.toContain("--permission-prompt-tool");
});
});
@@ -472,7 +470,7 @@ describe("mcp-config flag", () => {
expect(args).toContain("--append-system-prompt");
expect(args).toContain("--effort");
expect(args).not.toContain("--mcp-config");
expect(args).toContain("--permission-prompt-tool");
expect(args).not.toContain("--permission-prompt-tool");
});
});

221
packages/pi-claude-cli/src/__tests__/provider.test.ts
View File

@@ -238,6 +238,79 @@ describe("streamViaCli", () => {
expect(parsed.message.role).toBe("user");
});
describe("stdin close behavior", () => {
it("stdin.end() is called after writeUserMessage", async () => {
const model = mockModels[0] as any;
const context = {
messages: [{ role: "user", content: "Hello" }],
};
streamViaCli(model, context);
await vi.advanceTimersByTimeAsync(0);
const proc = (spawn as any).mock.results[0].value;
expect(proc.stdin.end).toHaveBeenCalledTimes(1);
expect(proc.stdin.write.mock.invocationCallOrder[0]).toBeLessThan(
proc.stdin.end.mock.invocationCallOrder[0],
);
});
it("unexpected control_request on stdout is logged and ignored", async () => {
process.env.PI_CLAUDE_CLI_DEBUG = "1";
const model = mockModels[0] as any;
const context = {
messages: [{ role: "user", content: "Hello" }],
};
const errorSpy = vi.spyOn(console, "error");
streamViaCli(model, context);
await vi.advanceTimersByTimeAsync(0);
const proc = (spawn as any).mock.results[0].value;
const lines = [
JSON.stringify({
type: "control_request",
request_id: "req_123",
request: {
subtype: "can_use_tool",
tool_name: "Read",
input: { file_path: "/foo.ts" },
},
}),
JSON.stringify({
type: "stream_event",
event: {
type: "message_start",
message: { usage: { input_tokens: 10, output_tokens: 0 } },
},
}),
JSON.stringify({
type: "stream_event",
event: { type: "message_stop" },
}),
JSON.stringify({
type: "result",
subtype: "success",
result: "ok",
}),
];
for (const line of lines) {
proc.stdout.write(line + "\n");
}
proc.stdout.end();
await vi.advanceTimersByTimeAsync(100);
const mockStream = MockAssistantMessageEventStream.mock.instances[0];
expect(mockStream._events.some((e: any) => e.type === "done")).toBe(true);
expect(proc.stdin.write).toHaveBeenCalledTimes(1);
expect(errorSpy).toHaveBeenCalledWith(
expect.stringContaining("unexpected control_request received"),
);
});
});
it("handles full text streaming sequence via NDJSON", async () => {
const model = mockModels[0] as any;
const context = {
@@ -406,74 +479,7 @@ describe("streamViaCli", () => {
await vi.advanceTimersByTimeAsync(100);
});
it("routes control_request through handleControlRequest and writes response to stdin", async () => {
const model = mockModels[0] as any;
const context = {
messages: [{ role: "user", content: "Hello" }],
};
streamViaCli(model, context);
await vi.advanceTimersByTimeAsync(0);
const proc = (spawn as any).mock.results[0].value;
// Clear initial stdin.write (user message)
proc.stdin.write.mockClear();
// Simulate a control_request NDJSON line arriving on stdout
const controlRequest = JSON.stringify({
type: "control_request",
request_id: "req_123",
request: {
subtype: "can_use_tool",
tool_name: "Read",
input: { file_path: "/foo.ts" },
},
});
// Then follow with stream events and result so stream completes
const lines = [
controlRequest,
JSON.stringify({
type: "stream_event",
event: {
type: "message_start",
message: { usage: { input_tokens: 10, output_tokens: 0 } },
},
}),
JSON.stringify({
type: "stream_event",
event: { type: "message_stop" },
}),
JSON.stringify({
type: "result",
subtype: "success",
result: "ok",
}),
];
for (const line of lines) {
proc.stdout.write(line + "\n");
}
proc.stdout.end();
await vi.advanceTimersByTimeAsync(100);
// Verify control_response was written to stdin
expect(proc.stdin.write).toHaveBeenCalled();
const stdinCalls = proc.stdin.write.mock.calls;
const controlResponse = stdinCalls.find((call: any[]) => {
try {
const parsed = JSON.parse(call[0]);
return parsed.type === "control_response";
} catch {
return false;
}
});
expect(controlResponse).toBeDefined();
const parsed = JSON.parse(controlResponse[0]);
expect(parsed.request_id).toBe("req_123");
expect(parsed.response.response.behavior).toBe("allow");
});
describe("thinking effort wiring", () => {
it("passes effort to spawnClaude when options.reasoning is provided on non-Opus model", async () => {
@@ -550,88 +556,7 @@ describe("streamViaCli", () => {
});
});
it("stream events continue flowing after control_request handling", async () => {
const model = mockModels[0] as any;
const context = {
messages: [{ role: "user", content: "Hello" }],
};
streamViaCli(model, context);
await vi.advanceTimersByTimeAsync(0);
const proc = (spawn as any).mock.results[0].value;
// control_request followed by normal stream events
const lines = [
JSON.stringify({
type: "control_request",
request_id: "req_456",
request: {
subtype: "can_use_tool",
tool_name: "Bash",
input: { command: "ls" },
},
}),
JSON.stringify({
type: "stream_event",
event: {
type: "message_start",
message: { usage: { input_tokens: 10, output_tokens: 0 } },
},
}),
JSON.stringify({
type: "stream_event",
event: {
type: "content_block_start",
index: 0,
content_block: { type: "text", text: "" },
},
}),
JSON.stringify({
type: "stream_event",
event: {
type: "content_block_delta",
index: 0,
delta: { type: "text_delta", text: "After control" },
},
}),
JSON.stringify({
type: "stream_event",
event: { type: "content_block_stop", index: 0 },
}),
JSON.stringify({
type: "stream_event",
event: {
type: "message_delta",
delta: { stop_reason: "end_turn" },
usage: { output_tokens: 3 },
},
}),
JSON.stringify({
type: "stream_event",
event: { type: "message_stop" },
}),
JSON.stringify({
type: "result",
subtype: "success",
result: "ok",
}),
];
for (const line of lines) {
proc.stdout.write(line + "\n");
}
proc.stdout.end();
await vi.advanceTimersByTimeAsync(100);
// Verify the stream still received text events after the control_request
const mockStream = MockAssistantMessageEventStream.mock.instances[0];
const events = mockStream._events;
const eventTypes = events.map((e: any) => e.type);
expect(eventTypes).toContain("text_start");
expect(eventTypes).toContain("text_delta");
expect(eventTypes).toContain("done");
});
describe("mcpConfigPath passthrough", () => {
it("passes mcpConfigPath to spawnClaude options", async () => {

30
packages/pi-claude-cli/src/control-handler.ts
View File

@@ -1,8 +1,8 @@
/**
* Control protocol handler for Claude CLI stream-json communication.
*
* Processes control_request messages from Claude CLI stdout and writes
* control_response messages to stdin.
* Processes control_request messages from Claude CLI stdout and returns a
* control_response decision object.
*
* - Custom MCP tools (mcp__custom-tools__*): DENIED — pi executes these
* - Everything else (user MCP tools, internal tools): ALLOWED — Claude handles
@@ -35,18 +35,33 @@ interface ControlResponse {
* Denies custom MCP tools (mcp__custom-tools__*) so pi can execute them.
* Allows everything else (user MCP tools, internal Claude tools).
*
* @returns true if the tool was allowed, false if denied
* Pure function: no side effects and no stdin writes.
*
* @returns Decision payload with allow/deny result and serialized response object
*/
export function handleControlRequest(
msg: ClaudeControlRequest,
stdin: NodeJS.WritableStream,
): boolean {
): { allowed: boolean; response: ControlResponse } {
if (!msg.request_id || !msg.request) {
console.error(
"[pi-claude-cli] Malformed control_request: missing request_id or request object",
msg,
);
return false;
return {
allowed: false,
response: {
type: "control_response",
request_id: msg.request_id ?? "",
response: {
subtype: "success",
response: {
behavior: "deny",
message: TOOL_EXECUTION_DENIED_MESSAGE,
},
},
},
};
}
const toolName = msg.request?.tool_name ?? "";
@@ -63,6 +78,5 @@ export function handleControlRequest(
},
};
stdin.write(JSON.stringify(response) + "\n");
return !isCustomTool;
return { allowed: !isCustomTool, response };
}

13
packages/pi-claude-cli/src/process-manager.ts
View File

@@ -11,6 +11,11 @@ import { writeFileSync, unlinkSync } from "node:fs";
import { join } from "node:path";
import { tmpdir } from "node:os";
function debugLog(message: string): void {
if (process.env.PI_CLAUDE_CLI_DEBUG !== "1") return;
console.error(`[pi-claude-cli] ${message}`);
}
/**
* Spawn a Claude CLI subprocess with all required flags for stream-json communication.
*
@@ -39,8 +44,6 @@ export function buildClaudeSpawnArgs(
"--include-partial-messages",
"--model",
modelId,
"--permission-prompt-tool",
"stdio",
];
if (options?.resumeSessionId) {
@@ -97,6 +100,8 @@ export function spawnClaude(
cwd: options?.cwd ?? process.cwd(),
});
debugLog(`spawnClaude: pid=${proc.pid} model=${modelId}`);
return proc as ChildProcess;
}
@@ -114,7 +119,8 @@ export function cleanupSystemPromptFile(): void {
/**
* Write a user message to the subprocess stdin as NDJSON.
* Does NOT call stdin.end() -- stdin stays open for control_response in Phase 2.
* Calls stdin.end() after writing the user message to signal EOF, allowing
* Claude CLI to process the input and start generating.
*
* Accepts both string (text-only prompt) and array (ContentBlock[] with images)
* content. JSON.stringify handles both natively. The stream-json protocol
@@ -135,6 +141,7 @@ export function writeUserMessage(
},
};
proc.stdin!.write(JSON.stringify(message) + "\n");
proc.stdin!.end();
}
/**

12
packages/pi-claude-cli/src/provider.ts
View File

@@ -42,7 +42,6 @@ import {
} from "./process-manager.js";
import { parseLine } from "./stream-parser.js";
import { createEventBridge } from "./event-bridge.js";
import { handleControlRequest } from "./control-handler.js";
import { mapThinkingEffort } from "./thinking-config.js";
import { isPiKnownClaudeTool } from "./tool-mapping.js";
/**
@@ -161,6 +160,7 @@ export function streamViaCli(
// Write user message to subprocess stdin
writeUserMessage(proc, prompt);
debugLog("user message written to stdin, stdin.end() called");
// Create event bridge (before endStreamWithError so bridge is in scope)
const bridge = createEventBridge(stream, model);
@@ -227,6 +227,7 @@ export function streamViaCli(
// Track tool_use blocks for break-early decision at message_stop
let sawBuiltInOrCustomTool = false;
let firstLineReceived = false;
// Guard against buffered readline lines firing after rl.close()
let broken = false;
@@ -247,6 +248,7 @@ export function streamViaCli(
// Handle subprocess close -- surface crashes with stderr and exit code
proc.on("close", (code: number | null, _signal: string | null) => {
clearTimeout(inactivityTimer);
debugLog(`subprocess closed: code=${code} signal=${_signal}`);
if (broken) return; // Break-early kill, expected
const stderr = getStderr().trim();
if (stderr) {
@@ -267,6 +269,10 @@ export function streamViaCli(
// NOTE: Using 'line' event instead of `for await` because the async
// iterator batches lines, breaking real-time streaming to pi.
rl.on("line", (line: string) => {
if (!firstLineReceived) {
firstLineReceived = true;
debugLog("first stdout line received from Claude CLI");
}
if (broken) return; // Guard: ignore buffered lines after break-early
// Reset inactivity timer on each line of output
@@ -319,7 +325,9 @@ export function streamViaCli(
return; // Don't process further -- done event already pushed by event bridge
}
} else if (msg.type === "control_request") {
handleControlRequest(msg, proc!.stdin!);
debugLog(
`unexpected control_request received (stdin already closed): ${msg.request_id}`,
);
} else if (msg.type === "result") {
if (msg.subtype === "error") {
endStreamWithError(msg.error ?? "Unknown error from Claude CLI");