diff --git a/src/agents/claude-cli-runner.test.ts b/src/agents/claude-cli-runner.test.ts index 7825d00da..a72cbb6e7 100644 --- a/src/agents/claude-cli-runner.test.ts +++ b/src/agents/claude-cli-runner.test.ts @@ -3,6 +3,7 @@ import { beforeEach, describe, expect, it, vi } from "vitest"; import { runClaudeCliAgent } from "./claude-cli-runner.js"; const runCommandWithTimeoutMock = vi.fn(); +const runCliWithStreamingMock = vi.fn(); function createDeferred() { let resolve: (value: T) => void; @@ -30,18 +31,23 @@ vi.mock("../process/exec.js", () => ({ runCommandWithTimeout: (...args: unknown[]) => runCommandWithTimeoutMock(...args), })); +vi.mock("./cli-runner/streaming.js", () => ({ + runCliWithStreaming: (...args: unknown[]) => runCliWithStreamingMock(...args), + mapCliStreamEvent: () => null, +})); + describe("runClaudeCliAgent", () => { beforeEach(() => { runCommandWithTimeoutMock.mockReset(); + runCliWithStreamingMock.mockReset(); }); it("starts a new session with --session-id when none is provided", async () => { - runCommandWithTimeoutMock.mockResolvedValueOnce({ - stdout: JSON.stringify({ message: "ok", session_id: "sid-1" }), - stderr: "", - code: 0, - signal: null, - killed: false, + // Claude CLI now uses streaming by default + runCliWithStreamingMock.mockResolvedValueOnce({ + text: "ok", + sessionId: "sid-1", + events: [], }); await runClaudeCliAgent({ @@ -54,20 +60,17 @@ describe("runClaudeCliAgent", () => { runId: "run-1", }); - expect(runCommandWithTimeoutMock).toHaveBeenCalledTimes(1); - const argv = runCommandWithTimeoutMock.mock.calls[0]?.[0] as string[]; - expect(argv).toContain("claude"); - expect(argv).toContain("--session-id"); - expect(argv).toContain("hi"); + expect(runCliWithStreamingMock).toHaveBeenCalledTimes(1); + const callArgs = runCliWithStreamingMock.mock.calls[0]?.[0] as { args: string[] }; + expect(callArgs.args).toContain("--session-id"); }); it("uses --resume when a claude session id is provided", async () => { - runCommandWithTimeoutMock.mockResolvedValueOnce({ - stdout: JSON.stringify({ message: "ok", session_id: "sid-2" }), - stderr: "", - code: 0, - signal: null, - killed: false, + // Claude CLI now uses streaming by default + runCliWithStreamingMock.mockResolvedValueOnce({ + text: "ok", + sessionId: "sid-2", + events: [], }); await runClaudeCliAgent({ @@ -81,30 +84,18 @@ describe("runClaudeCliAgent", () => { claudeSessionId: "c9d7b831-1c31-4d22-80b9-1e50ca207d4b", }); - expect(runCommandWithTimeoutMock).toHaveBeenCalledTimes(1); - const argv = runCommandWithTimeoutMock.mock.calls[0]?.[0] as string[]; - expect(argv).toContain("--resume"); - expect(argv).toContain("c9d7b831-1c31-4d22-80b9-1e50ca207d4b"); - expect(argv).toContain("hi"); + expect(runCliWithStreamingMock).toHaveBeenCalledTimes(1); + const callArgs = runCliWithStreamingMock.mock.calls[0]?.[0] as { args: string[] }; + expect(callArgs.args).toContain("--resume"); + expect(callArgs.args).toContain("c9d7b831-1c31-4d22-80b9-1e50ca207d4b"); }); it("serializes concurrent claude-cli runs", async () => { - const firstDeferred = createDeferred<{ - stdout: string; - stderr: string; - code: number | null; - signal: NodeJS.Signals | null; - killed: boolean; - }>(); - const secondDeferred = createDeferred<{ - stdout: string; - stderr: string; - code: number | null; - signal: NodeJS.Signals | null; - killed: boolean; - }>(); + type StreamResult = { text: string; sessionId: string; events: unknown[] }; + const firstDeferred = createDeferred(); + const secondDeferred = createDeferred(); - runCommandWithTimeoutMock + runCliWithStreamingMock .mockImplementationOnce(() => firstDeferred.promise) .mockImplementationOnce(() => secondDeferred.promise); @@ -128,25 +119,13 @@ describe("runClaudeCliAgent", () => { runId: "run-2", }); - await waitForCalls(runCommandWithTimeoutMock, 1); + await waitForCalls(runCliWithStreamingMock, 1); - firstDeferred.resolve({ - stdout: JSON.stringify({ message: "ok", session_id: "sid-1" }), - stderr: "", - code: 0, - signal: null, - killed: false, - }); + firstDeferred.resolve({ text: "ok", sessionId: "sid-1", events: [] }); - await waitForCalls(runCommandWithTimeoutMock, 2); + await waitForCalls(runCliWithStreamingMock, 2); - secondDeferred.resolve({ - stdout: JSON.stringify({ message: "ok", session_id: "sid-2" }), - stderr: "", - code: 0, - signal: null, - killed: false, - }); + secondDeferred.resolve({ text: "ok", sessionId: "sid-2", events: [] }); await Promise.all([firstRun, secondRun]); }); diff --git a/src/agents/cli-backends.test.ts b/src/agents/cli-backends.test.ts new file mode 100644 index 000000000..63dfabc59 --- /dev/null +++ b/src/agents/cli-backends.test.ts @@ -0,0 +1,109 @@ +import { describe, expect, it } from "vitest"; + +import { resolveCliBackendConfig } from "./cli-backends.js"; + +describe("resolveCliBackendConfig", () => { + describe("claude-cli defaults", () => { + it("includes --verbose when using stream-json output format", () => { + const resolved = resolveCliBackendConfig("claude-cli"); + expect(resolved).not.toBeNull(); + + const args = resolved!.config.args ?? []; + const resumeArgs = resolved!.config.resumeArgs ?? []; + + // stream-json requires --verbose in print mode + if (args.includes("stream-json")) { + expect(args).toContain("--verbose"); + } + if (resumeArgs.includes("stream-json")) { + expect(resumeArgs).toContain("--verbose"); + } + }); + + it("has streaming enabled with stream-json format", () => { + const resolved = resolveCliBackendConfig("claude-cli"); + expect(resolved).not.toBeNull(); + + expect(resolved!.config.streaming).toBe(true); + expect(resolved!.config.args).toContain("stream-json"); + expect(resolved!.config.output).toBe("jsonl"); + }); + + it("includes required streaming event types", () => { + const resolved = resolveCliBackendConfig("claude-cli"); + expect(resolved).not.toBeNull(); + + const eventTypes = resolved!.config.streamingEventTypes ?? []; + expect(eventTypes).toContain("text"); + expect(eventTypes).toContain("result"); + }); + + it("resume args also use stream-json with --verbose", () => { + const resolved = resolveCliBackendConfig("claude-cli"); + expect(resolved).not.toBeNull(); + + const resumeArgs = resolved!.config.resumeArgs ?? []; + expect(resumeArgs).toContain("stream-json"); + expect(resumeArgs).toContain("--verbose"); + expect(resumeArgs).toContain("--resume"); + expect(resumeArgs).toContain("{sessionId}"); + }); + }); + + describe("codex-cli defaults", () => { + it("has streaming enabled", () => { + const resolved = resolveCliBackendConfig("codex-cli"); + expect(resolved).not.toBeNull(); + + expect(resolved!.config.streaming).toBe(true); + expect(resolved!.config.output).toBe("jsonl"); + }); + + it("includes required streaming event types", () => { + const resolved = resolveCliBackendConfig("codex-cli"); + expect(resolved).not.toBeNull(); + + const eventTypes = resolved!.config.streamingEventTypes ?? []; + expect(eventTypes).toContain("item"); + expect(eventTypes).toContain("turn.completed"); + }); + }); + + describe("config overrides", () => { + it("allows disabling streaming via config override", () => { + const resolved = resolveCliBackendConfig("claude-cli", { + agents: { + defaults: { + cliBackends: { + "claude-cli": { + command: "claude", + streaming: false, + }, + }, + }, + }, + }); + expect(resolved).not.toBeNull(); + expect(resolved!.config.streaming).toBe(false); + }); + + it("preserves base args when override does not specify args", () => { + const resolved = resolveCliBackendConfig("claude-cli", { + agents: { + defaults: { + cliBackends: { + "claude-cli": { + command: "claude", + streaming: false, + }, + }, + }, + }, + }); + expect(resolved).not.toBeNull(); + // Should still have the base args including --verbose + expect(resolved!.config.args).toContain("--verbose"); + expect(resolved!.config.args).toContain("stream-json"); + }); + }); +}); diff --git a/src/agents/cli-backends.ts b/src/agents/cli-backends.ts index 713d01fc5..cc433729f 100644 --- a/src/agents/cli-backends.ts +++ b/src/agents/cli-backends.ts @@ -27,16 +27,17 @@ const CLAUDE_MODEL_ALIASES: Record = { const DEFAULT_CLAUDE_BACKEND: CliBackendConfig = { command: "claude", - args: ["-p", "--output-format", "json", "--dangerously-skip-permissions"], + args: ["-p", "--output-format", "stream-json", "--dangerously-skip-permissions", "--verbose"], resumeArgs: [ "-p", "--output-format", - "json", + "stream-json", "--dangerously-skip-permissions", + "--verbose", "--resume", "{sessionId}", ], - output: "json", + output: "jsonl", input: "arg", modelArg: "--model", modelAliases: CLAUDE_MODEL_ALIASES, @@ -55,6 +56,8 @@ const DEFAULT_CLAUDE_BACKEND: CliBackendConfig = { cacheWrite: ["cache_creation_input_tokens", "cache_write_input_tokens", "cacheWrite"], total: ["total_tokens", "total"], }, + streaming: true, + streamingEventTypes: ["tool_use", "tool_result", "text", "result"], }; const DEFAULT_CODEX_BACKEND: CliBackendConfig = { @@ -84,6 +87,8 @@ const DEFAULT_CODEX_BACKEND: CliBackendConfig = { output: ["completion_tokens", "output_tokens"], total: ["total_tokens"], }, + streaming: true, + streamingEventTypes: ["item", "turn.completed"], }; function normalizeBackendKey(key: string): string { @@ -112,6 +117,8 @@ function mergeBackendConfig(base: CliBackendConfig, override?: CliBackendConfig) sessionIdFields: override.sessionIdFields ?? base.sessionIdFields, sessionArgs: override.sessionArgs ?? base.sessionArgs, resumeArgs: override.resumeArgs ?? base.resumeArgs, + streaming: override.streaming ?? base.streaming, + streamingEventTypes: override.streamingEventTypes ?? base.streamingEventTypes, }; } diff --git a/src/agents/cli-runner.test.ts b/src/agents/cli-runner.test.ts index ca3ea0b90..dccdec185 100644 --- a/src/agents/cli-runner.test.ts +++ b/src/agents/cli-runner.test.ts @@ -6,16 +6,23 @@ import { cleanupSuspendedCliProcesses } from "./cli-runner/helpers.js"; const runCommandWithTimeoutMock = vi.fn(); const runExecMock = vi.fn(); +const runCliWithStreamingMock = vi.fn(); vi.mock("../process/exec.js", () => ({ runCommandWithTimeout: (...args: unknown[]) => runCommandWithTimeoutMock(...args), runExec: (...args: unknown[]) => runExecMock(...args), })); +vi.mock("./cli-runner/streaming.js", () => ({ + runCliWithStreaming: (...args: unknown[]) => runCliWithStreamingMock(...args), + mapCliStreamEvent: () => null, +})); + describe("runCliAgent resume cleanup", () => { beforeEach(() => { runCommandWithTimeoutMock.mockReset(); runExecMock.mockReset(); + runCliWithStreamingMock.mockReset(); }); it("kills stale resume processes for codex sessions", async () => { @@ -25,12 +32,11 @@ describe("runCliAgent resume cleanup", () => { stderr: "", }) // cleanupSuspendedCliProcesses (ps) .mockResolvedValueOnce({ stdout: "", stderr: "" }); // cleanupResumeProcesses (pkill) - runCommandWithTimeoutMock.mockResolvedValueOnce({ - stdout: "ok", - stderr: "", - code: 0, - signal: null, - killed: false, + // Codex CLI now uses streaming by default + runCliWithStreamingMock.mockResolvedValueOnce({ + text: "ok", + sessionId: "thread-123", + events: [], }); await runCliAgent({ diff --git a/src/agents/cli-runner.ts b/src/agents/cli-runner.ts index 01d5432fc..18cd8c775 100644 --- a/src/agents/cli-runner.ts +++ b/src/agents/cli-runner.ts @@ -2,6 +2,7 @@ import type { ImageContent } from "@mariozechner/pi-ai"; import { resolveHeartbeatPrompt } from "../auto-reply/heartbeat.js"; import type { ThinkLevel } from "../auto-reply/thinking.js"; import type { ClawdbotConfig } from "../config/config.js"; +import { emitAgentEvent } from "../infra/agent-events.js"; import { isTruthyEnvValue } from "../infra/env.js"; import { shouldLogVerbose } from "../globals.js"; import { createSubsystemLogger } from "../logging/subsystem.js"; @@ -26,6 +27,7 @@ import { resolveSystemPromptUsage, writeCliImages, } from "./cli-runner/helpers.js"; +import { mapCliStreamEvent, runCliWithStreaming } from "./cli-runner/streaming.js"; import { FailoverError, resolveFailoverStatus } from "./failover-error.js"; import { classifyFailoverReason, isFailoverErrorMessage } from "./pi-embedded-helpers.js"; import type { EmbeddedPiRunResult } from "./pi-embedded-runner.js"; @@ -218,6 +220,66 @@ export async function runCliAgent(params: { await cleanupResumeProcesses(backend, cliSessionIdToSend); } + // Use streaming execution when enabled + const useStreaming = backend.streaming ?? false; + log.info( + `cli runner: useStreaming=${useStreaming} streamingEventTypes=${JSON.stringify(backend.streamingEventTypes)}`, + ); + if (useStreaming) { + try { + const streamResult = await runCliWithStreaming({ + command: backend.command, + args, + cwd: workspaceDir, + env, + input: stdinPayload, + timeoutMs: params.timeoutMs, + eventTypes: backend.streamingEventTypes, + backend, + onEvent: (event) => { + log.info(`cli runner: onEvent received type="${event.type}"`); + const mapped = mapCliStreamEvent(event, backendResolved.id); + if (mapped) { + log.info( + `cli runner: emitting agentEvent stream="${mapped.stream}" runId="${params.runId}"`, + ); + emitAgentEvent({ + runId: params.runId, + stream: mapped.stream, + data: mapped.data, + }); + log.info(`cli runner: emitAgentEvent completed`); + } else { + log.info(`cli runner: mapped is null, not emitting agentEvent`); + } + }, + }); + + if (logOutputText || shouldLogVerbose()) { + log.info( + `cli streaming: text=${streamResult.text.length} chars, events=${streamResult.events.length}`, + ); + } + + return { + text: streamResult.text, + sessionId: streamResult.sessionId, + usage: streamResult.usage, + }; + } catch (streamErr) { + const errMsg = streamErr instanceof Error ? streamErr.message : String(streamErr); + const reason = classifyFailoverReason(errMsg) ?? "unknown"; + const status = resolveFailoverStatus(reason); + throw new FailoverError(errMsg, { + reason, + provider: params.provider, + model: modelId, + status, + }); + } + } + + // Non-streaming path: collect all output then parse const result = await runCommandWithTimeout([backend.command, ...args], { timeoutMs: params.timeoutMs, cwd: workspaceDir, diff --git a/src/agents/cli-runner/streaming.test.ts b/src/agents/cli-runner/streaming.test.ts new file mode 100644 index 000000000..aa46fd2e7 --- /dev/null +++ b/src/agents/cli-runner/streaming.test.ts @@ -0,0 +1,463 @@ +import { describe, expect, it, vi, beforeEach } from "vitest"; +import type { ChildProcessWithoutNullStreams } from "node:child_process"; +import { EventEmitter, Readable, Writable } from "node:stream"; + +import type { CliBackendConfig } from "../../config/types.js"; +import { + runCliWithStreaming, + mapClaudeStreamEvent, + mapCodexStreamEvent, + mapCliStreamEvent, + type CliStreamEvent, +} from "./streaming.js"; + +// Mock child_process.spawn +const spawnMock = vi.fn(); +vi.mock("node:child_process", () => ({ + spawn: (...args: unknown[]) => spawnMock(...args), +})); + +function createMockProcess() { + const stdin = new Writable({ + write(_chunk, _encoding, callback) { + callback(); + }, + }); + const stdout = new Readable({ read() {} }); + const stderr = new Readable({ read() {} }); + const proc = new EventEmitter() as ChildProcessWithoutNullStreams; + proc.stdin = stdin as ChildProcessWithoutNullStreams["stdin"]; + proc.stdout = stdout as ChildProcessWithoutNullStreams["stdout"]; + proc.stderr = stderr as ChildProcessWithoutNullStreams["stderr"]; + proc.killed = false; + proc.kill = vi.fn(() => { + proc.killed = true; + return true; + }); + return proc; +} + +// Helper to wait for readline to process a line +function nextTick(): Promise { + return new Promise((resolve) => process.nextTick(resolve)); +} + +describe("runCliWithStreaming", () => { + const defaultBackend: CliBackendConfig = { + command: "claude", + sessionIdFields: ["session_id"], + usageFields: { + input: ["input_tokens"], + output: ["output_tokens"], + }, + }; + + beforeEach(() => { + spawnMock.mockReset(); + }); + + it("parses NDJSON lines and accumulates text from text events", async () => { + const proc = createMockProcess(); + spawnMock.mockReturnValue(proc); + + const events: CliStreamEvent[] = []; + const promise = runCliWithStreaming({ + command: "claude", + args: ["-p"], + cwd: "/tmp", + env: {}, + timeoutMs: 5000, + backend: defaultBackend, + onEvent: (e) => events.push(e), + }); + + // Give readline time to set up + await nextTick(); + + // Emit streaming events + proc.stdout.push('{"type":"text","text":"Hello "}\n'); + proc.stdout.push('{"type":"text","text":"world!"}\n'); + proc.stdout.push( + '{"type":"result","session_id":"sess-123","usage":{"input_tokens":10,"output_tokens":5}}\n', + ); + proc.stdout.push(null); + + await nextTick(); + proc.emit("close", 0, null); + + const result = await promise; + + expect(result.text).toBe("Hello world!"); + expect(result.sessionId).toBe("sess-123"); + expect(result.usage).toEqual({ input: 10, output: 5 }); + expect(events).toHaveLength(3); + expect(events[0]).toEqual({ type: "text", text: "Hello " }); + expect(events[1]).toEqual({ type: "text", text: "world!" }); + }); + + it("filters events by type when streamingEventTypes is specified", async () => { + const proc = createMockProcess(); + spawnMock.mockReturnValue(proc); + + const events: CliStreamEvent[] = []; + const promise = runCliWithStreaming({ + command: "claude", + args: ["-p"], + cwd: "/tmp", + env: {}, + timeoutMs: 5000, + eventTypes: ["text"], + backend: defaultBackend, + onEvent: (e) => events.push(e), + }); + + await nextTick(); + + proc.stdout.push('{"type":"text","text":"Hello"}\n'); + proc.stdout.push('{"type":"tool_use","name":"bash"}\n'); + proc.stdout.push('{"type":"result"}\n'); + proc.stdout.push(null); + + await nextTick(); + proc.emit("close", 0, null); + + await promise; + + // Only "text" events should be emitted + expect(events).toHaveLength(1); + expect(events[0]?.type).toBe("text"); + }); + + it("supports prefix matching for event types", async () => { + const proc = createMockProcess(); + spawnMock.mockReturnValue(proc); + + const events: CliStreamEvent[] = []; + const promise = runCliWithStreaming({ + command: "codex", + args: ["exec"], + cwd: "/tmp", + env: {}, + timeoutMs: 5000, + eventTypes: ["item"], + backend: { ...defaultBackend, sessionIdFields: ["thread_id"] }, + onEvent: (e) => events.push(e), + }); + + await nextTick(); + + proc.stdout.push('{"type":"item.created","item":{"type":"message"}}\n'); + proc.stdout.push('{"type":"item.completed","item":{"type":"message","text":"Done"}}\n'); + proc.stdout.push('{"type":"turn.completed"}\n'); + proc.stdout.push(null); + + await nextTick(); + proc.emit("close", 0, null); + + await promise; + + // "item" prefix should match "item.created" and "item.completed" + expect(events).toHaveLength(2); + expect(events.map((e) => e.type)).toEqual(["item.created", "item.completed"]); + }); + + it("handles malformed JSON lines gracefully", async () => { + const proc = createMockProcess(); + spawnMock.mockReturnValue(proc); + + const events: CliStreamEvent[] = []; + const promise = runCliWithStreaming({ + command: "claude", + args: ["-p"], + cwd: "/tmp", + env: {}, + timeoutMs: 5000, + backend: defaultBackend, + onEvent: (e) => events.push(e), + }); + + await nextTick(); + + proc.stdout.push('{"type":"text","text":"Valid"}\n'); + proc.stdout.push("not valid json\n"); + proc.stdout.push('{"type":"text","text":" line"}\n'); + proc.stdout.push(null); + + await nextTick(); + proc.emit("close", 0, null); + + const result = await promise; + + expect(result.text).toBe("Valid line"); + expect(events).toHaveLength(2); + }); + + it("rejects on non-zero exit code with stderr", async () => { + const proc = createMockProcess(); + spawnMock.mockReturnValue(proc); + + const promise = runCliWithStreaming({ + command: "claude", + args: ["-p"], + cwd: "/tmp", + env: {}, + timeoutMs: 5000, + backend: defaultBackend, + onEvent: () => {}, + }); + + await nextTick(); + + proc.stderr.push("Error: API key invalid"); + proc.stderr.push(null); + + await nextTick(); + proc.emit("close", 1, null); + + await expect(promise).rejects.toThrow("Error: API key invalid"); + }); + + it("rejects on timeout", async () => { + const proc = createMockProcess(); + spawnMock.mockReturnValue(proc); + + const promise = runCliWithStreaming({ + command: "claude", + args: ["-p"], + cwd: "/tmp", + env: {}, + timeoutMs: 50, // Short timeout for test + backend: defaultBackend, + onEvent: () => {}, + }); + + // Don't close the process - let it timeout + await expect(promise).rejects.toThrow("CLI streaming timeout after 50ms"); + }); + + it("extracts session ID from Claude CLI result event", async () => { + const proc = createMockProcess(); + spawnMock.mockReturnValue(proc); + + const promise = runCliWithStreaming({ + command: "claude", + args: ["-p"], + cwd: "/tmp", + env: {}, + timeoutMs: 5000, + backend: defaultBackend, + onEvent: () => {}, + }); + + await nextTick(); + + proc.stdout.push('{"type":"result","session_id":"my-session-id"}\n'); + proc.stdout.push(null); + + await nextTick(); + proc.emit("close", 0, null); + + const result = await promise; + expect(result.sessionId).toBe("my-session-id"); + }); + + it("extracts thread_id from Codex CLI events", async () => { + const proc = createMockProcess(); + spawnMock.mockReturnValue(proc); + + const promise = runCliWithStreaming({ + command: "codex", + args: ["exec"], + cwd: "/tmp", + env: {}, + timeoutMs: 5000, + backend: { ...defaultBackend, sessionIdFields: ["thread_id"] }, + onEvent: () => {}, + }); + + await nextTick(); + + proc.stdout.push('{"type":"turn.completed","thread_id":"thread-abc"}\n'); + proc.stdout.push(null); + + await nextTick(); + proc.emit("close", 0, null); + + const result = await promise; + expect(result.sessionId).toBe("thread-abc"); + }); +}); + +describe("mapClaudeStreamEvent", () => { + it("maps tool_use event to tool start", () => { + const event: CliStreamEvent = { + type: "tool_use", + id: "tool-1", + name: "bash", + input: { command: "ls" }, + }; + + const result = mapClaudeStreamEvent(event); + + expect(result).toEqual({ + stream: "tool", + data: { + phase: "start", + name: "bash", + id: "tool-1", + input: { command: "ls" }, + }, + }); + }); + + it("maps tool_result event to tool end", () => { + const event: CliStreamEvent = { + type: "tool_result", + tool_use_id: "tool-1", + content: "file1.txt\nfile2.txt", + is_error: false, + }; + + const result = mapClaudeStreamEvent(event); + + expect(result).toEqual({ + stream: "tool", + data: { + phase: "end", + id: "tool-1", + output: "file1.txt\nfile2.txt", + isError: false, + }, + }); + }); + + it("maps text event to assistant delta", () => { + const event: CliStreamEvent = { type: "text", text: "Hello" }; + + const result = mapClaudeStreamEvent(event); + + expect(result).toEqual({ + stream: "assistant", + data: { text: "Hello", delta: true }, + }); + }); + + it("returns null for empty text event", () => { + const event: CliStreamEvent = { type: "text", text: "" }; + expect(mapClaudeStreamEvent(event)).toBeNull(); + }); + + it("returns null for result event", () => { + const event: CliStreamEvent = { type: "result" }; + expect(mapClaudeStreamEvent(event)).toBeNull(); + }); +}); + +describe("mapCodexStreamEvent", () => { + it("maps item.created function_call to tool start", () => { + const event: CliStreamEvent = { + type: "item.created", + item: { + type: "function_call", + id: "call-1", + name: "shell", + arguments: '{"cmd":"ls"}', + }, + }; + + const result = mapCodexStreamEvent(event); + + expect(result).toEqual({ + stream: "tool", + data: { + phase: "start", + name: "shell", + id: "call-1", + input: '{"cmd":"ls"}', + }, + }); + }); + + it("maps item.completed function_call_output to tool end", () => { + const event: CliStreamEvent = { + type: "item.completed", + item: { + type: "function_call_output", + call_id: "call-1", + output: "file1.txt", + }, + }; + + const result = mapCodexStreamEvent(event); + + expect(result).toEqual({ + stream: "tool", + data: { + phase: "end", + id: "call-1", + output: "file1.txt", + }, + }); + }); + + it("maps item.completed message to assistant text", () => { + const event: CliStreamEvent = { + type: "item.completed", + item: { + type: "message", + text: "Task completed", + }, + }; + + const result = mapCodexStreamEvent(event); + + expect(result).toEqual({ + stream: "assistant", + data: { text: "Task completed" }, + }); + }); + + it("returns null for turn.completed", () => { + const event: CliStreamEvent = { type: "turn.completed" }; + expect(mapCodexStreamEvent(event)).toBeNull(); + }); +}); + +describe("mapCliStreamEvent", () => { + it("uses Claude mapper for claude-cli backend", () => { + const event: CliStreamEvent = { type: "text", text: "Hello" }; + const result = mapCliStreamEvent(event, "claude-cli"); + + expect(result).toEqual({ + stream: "assistant", + data: { text: "Hello", delta: true }, + }); + }); + + it("uses Codex mapper for codex-cli backend", () => { + const event: CliStreamEvent = { + type: "item.completed", + item: { type: "message", text: "Done" }, + }; + const result = mapCliStreamEvent(event, "codex-cli"); + + expect(result).toEqual({ + stream: "assistant", + data: { text: "Done" }, + }); + }); + + it("detects Codex format from event type prefix", () => { + const event: CliStreamEvent = { + type: "item.created", + item: { type: "function_call", name: "test", id: "1" }, + }; + // Even with non-codex backend name, detects format from event type + const result = mapCliStreamEvent(event, "custom-backend"); + + expect(result).toEqual({ + stream: "tool", + data: { phase: "start", name: "test", id: "1", input: undefined }, + }); + }); +}); diff --git a/src/agents/cli-runner/streaming.ts b/src/agents/cli-runner/streaming.ts new file mode 100644 index 000000000..d4b168b6d --- /dev/null +++ b/src/agents/cli-runner/streaming.ts @@ -0,0 +1,479 @@ +import { type ChildProcessWithoutNullStreams, spawn } from "node:child_process"; +import { createInterface, type Interface } from "node:readline"; + +import type { CliBackendConfig } from "../../config/types.js"; +import { createSubsystemLogger } from "../../logging/subsystem.js"; + +const log = createSubsystemLogger("agent/cli-streaming"); + +export type CliStreamEvent = { + type: string; + [key: string]: unknown; +}; + +type CliUsage = { + input?: number; + output?: number; + cacheRead?: number; + cacheWrite?: number; + total?: number; +}; + +export type CliStreamResult = { + text: string; + sessionId?: string; + usage?: CliUsage; + events: CliStreamEvent[]; +}; + +export type CliStreamParams = { + command: string; + args: string[]; + cwd: string; + env: NodeJS.ProcessEnv; + input?: string; + timeoutMs: number; + eventTypes?: string[]; + backend: CliBackendConfig; + onEvent: (event: CliStreamEvent) => void; +}; + +function isRecord(value: unknown): value is Record { + return Boolean(value && typeof value === "object" && !Array.isArray(value)); +} + +function pickSessionId( + parsed: Record, + backend: CliBackendConfig, +): string | undefined { + const fields = backend.sessionIdFields ?? [ + "session_id", + "sessionId", + "conversation_id", + "conversationId", + ]; + for (const field of fields) { + const value = parsed[field]; + if (typeof value === "string" && value.trim()) return value.trim(); + } + return undefined; +} + +function toUsage(raw: Record, backend?: CliBackendConfig): CliUsage | undefined { + const pick = (key: string) => + typeof raw[key] === "number" && raw[key] > 0 ? (raw[key] as number) : undefined; + + const pickFirst = (keys: string[] | undefined, fallback: string[]): number | undefined => { + const ordered = keys && keys.length > 0 ? keys : fallback; + for (const key of ordered) { + const value = pick(key); + if (value !== undefined) return value; + } + return undefined; + }; + + const fields = backend?.usageFields; + const input = pickFirst(fields?.input, ["input_tokens", "inputTokens"]); + const output = pickFirst(fields?.output, ["output_tokens", "outputTokens"]); + const cacheRead = pickFirst(fields?.cacheRead, [ + "cache_read_input_tokens", + "cached_input_tokens", + "cacheRead", + ]); + const cacheWrite = pickFirst(fields?.cacheWrite, [ + "cache_creation_input_tokens", + "cache_write_input_tokens", + "cacheWrite", + ]); + const total = pickFirst(fields?.total, ["total_tokens", "total"]); + + if (!input && !output && !cacheRead && !cacheWrite && !total) return undefined; + return { input, output, cacheRead, cacheWrite, total }; +} + +/** Check if an event type matches the filter (supports prefix matching with *). */ +function matchesEventType(eventType: string, filters: string[]): boolean { + for (const filter of filters) { + if (filter === eventType) return true; + // Support prefix matching (e.g., "item" matches "item.created", "item.completed") + if (eventType.startsWith(`${filter}.`)) return true; + } + return false; +} + +/** + * Run a CLI with streaming NDJSON output, parsing events line-by-line. + * Follows the iMessage RPC client pattern for readline-based JSON parsing. + */ +export async function runCliWithStreaming(params: CliStreamParams): Promise { + const { command, args, cwd, env, input, timeoutMs, eventTypes, backend, onEvent } = params; + + return new Promise((resolve, reject) => { + let child: ChildProcessWithoutNullStreams | null = null; + let reader: Interface | null = null; + let timeoutHandle: NodeJS.Timeout | null = null; + let resolved = false; + + const events: CliStreamEvent[] = []; + const textParts: string[] = []; + let sessionId: string | undefined; + let usage: CliUsage | undefined; + let stderrBuffer = ""; + + const cleanup = () => { + if (timeoutHandle) { + clearTimeout(timeoutHandle); + timeoutHandle = null; + } + reader?.close(); + reader = null; + }; + + const fail = (err: Error) => { + if (resolved) return; + resolved = true; + cleanup(); + if (child && !child.killed) { + child.kill("SIGTERM"); + } + reject(err); + }; + + const succeed = () => { + if (resolved) return; + resolved = true; + cleanup(); + resolve({ + text: textParts.join("").trim(), + sessionId, + usage, + events, + }); + }; + + // Timeout handling + if (timeoutMs > 0) { + timeoutHandle = setTimeout(() => { + fail(new Error(`CLI streaming timeout after ${timeoutMs}ms`)); + }, timeoutMs); + } + + try { + child = spawn(command, args, { + stdio: ["pipe", "pipe", "pipe"], + cwd, + env, + }); + } catch (err) { + fail(err instanceof Error ? err : new Error(String(err))); + return; + } + + reader = createInterface({ input: child.stdout }); + + reader.on("line", (line) => { + const trimmed = line.trim(); + if (!trimmed) return; + + // Log raw line (truncated for readability) + log.info( + `cli stream: raw line: ${trimmed.slice(0, 300)}${trimmed.length > 300 ? "..." : ""}`, + ); + + let parsed: Record; + try { + parsed = JSON.parse(trimmed); + } catch { + // Not valid JSON, skip + log.debug(`cli stream: skipping non-JSON line: ${trimmed.slice(0, 100)}`); + return; + } + + if (!isRecord(parsed)) { + log.debug(`cli stream: parsed value is not a record, skipping`); + return; + } + + // Always try to extract session ID and usage from any parsed object + if (!sessionId) { + sessionId = pickSessionId(parsed, backend); + if (sessionId) { + log.info(`cli stream: extracted sessionId=${sessionId}`); + } + } + if (isRecord(parsed.usage)) { + const newUsage = toUsage(parsed.usage, backend); + if (newUsage) { + usage = newUsage; + log.info(`cli stream: extracted usage input=${usage.input} output=${usage.output}`); + } + } + + const eventType = typeof parsed.type === "string" ? parsed.type : ""; + log.info(`cli stream: eventType="${eventType}"`); + + if (!eventType) { + // Non-typed event (e.g., final result object) - already extracted session/usage above + log.debug(`cli stream: no event type, skipping event creation`); + return; + } + + const event: CliStreamEvent = { type: eventType, ...parsed }; + events.push(event); + log.info(`cli stream: created event #${events.length} type="${eventType}"`); + + // Filter events by type if specified + const shouldEmit = + !eventTypes || eventTypes.length === 0 || matchesEventType(eventType, eventTypes); + log.info( + `cli stream: shouldEmit=${shouldEmit} (filters=${JSON.stringify(eventTypes ?? [])})`, + ); + + // Handle Claude CLI event types + if (eventType === "text" || eventType === "content_block_delta") { + // Claude CLI text streaming + const text = typeof parsed.text === "string" ? parsed.text : ""; + if (text) { + textParts.push(text); + log.info( + `cli stream: accumulated text chunk (${text.length} chars): "${text.slice(0, 100)}${text.length > 100 ? "..." : ""}"`, + ); + } + } else if (eventType === "assistant" || eventType === "message") { + // Claude CLI assistant message (may contain full text) + const message = isRecord(parsed.message) ? parsed.message : parsed; + const content = message.content; + if (typeof content === "string") { + textParts.push(content); + log.info(`cli stream: accumulated assistant content (${content.length} chars)`); + } else if (Array.isArray(content)) { + for (const block of content) { + if (isRecord(block) && typeof block.text === "string") { + textParts.push(block.text); + log.info(`cli stream: accumulated content block (${block.text.length} chars)`); + } + } + } + } else if (eventType === "result") { + // Claude CLI final result event - usage already extracted above + const result = isRecord(parsed.result) ? parsed.result : parsed; + if (typeof result.text === "string") { + textParts.push(result.text); + log.info(`cli stream: accumulated result text (${result.text.length} chars)`); + } + } + + // Handle Codex CLI event types + if (eventType.startsWith("item.")) { + const item = isRecord(parsed.item) ? parsed.item : null; + if (item && typeof item.text === "string") { + const itemType = typeof item.type === "string" ? item.type.toLowerCase() : ""; + if (!itemType || itemType.includes("message")) { + textParts.push(item.text); + log.info(`cli stream: accumulated item text (${item.text.length} chars)`); + } + } + } else if (eventType === "turn.completed" || eventType === "thread.completed") { + if (isRecord(parsed.usage)) { + usage = toUsage(parsed.usage, backend) ?? usage; + } + // Extract thread_id as session ID for Codex + if (!sessionId && typeof parsed.thread_id === "string") { + sessionId = parsed.thread_id.trim(); + log.info(`cli stream: extracted thread_id as sessionId=${sessionId}`); + } + } + + // Emit the event if it passes the filter + if (shouldEmit) { + log.info(`cli stream: emitting event type="${eventType}" to onEvent callback`); + try { + onEvent(event); + log.info(`cli stream: onEvent callback completed for type="${eventType}"`); + } catch (err) { + log.info( + `cli stream: onEvent error: ${err instanceof Error ? err.message : String(err)}`, + ); + } + } else { + log.info(`cli stream: skipping emit for type="${eventType}" (filtered out)`); + } + }); + + child.stderr?.on("data", (chunk) => { + stderrBuffer += chunk.toString(); + }); + + child.on("error", (err) => { + fail(err); + }); + + child.on("close", (code, signal) => { + if (code !== 0 && code !== null) { + const reason = signal ? `signal ${signal}` : `code ${code}`; + const errMsg = stderrBuffer.trim() || `CLI exited with ${reason}`; + fail(new Error(errMsg)); + return; + } + succeed(); + }); + + // Write stdin if provided + if (input && child.stdin) { + child.stdin.write(input); + child.stdin.end(); + } else if (child.stdin) { + child.stdin.end(); + } + }); +} + +/** + * Map a Claude CLI stream event to Clawdbot agent event data. + * Returns null for events that should not be emitted. + */ +export function mapClaudeStreamEvent( + event: CliStreamEvent, +): { stream: string; data: Record } | null { + switch (event.type) { + case "tool_use": { + const toolName = typeof event.name === "string" ? event.name : "unknown"; + const toolId = typeof event.id === "string" ? event.id : undefined; + return { + stream: "tool", + data: { + phase: "start", + name: toolName, + id: toolId, + input: event.input, + }, + }; + } + case "tool_result": { + const toolId = typeof event.tool_use_id === "string" ? event.tool_use_id : undefined; + return { + stream: "tool", + data: { + phase: "end", + id: toolId, + output: event.content, + isError: event.is_error === true, + }, + }; + } + case "text": + case "content_block_delta": { + const text = typeof event.text === "string" ? event.text : ""; + if (!text) return null; + return { + stream: "assistant", + data: { text, delta: true }, + }; + } + case "assistant": + case "message": { + // Full assistant message - typically we prefer deltas, but include for completeness + return null; // Text is accumulated separately + } + case "result": { + // Final result - don't emit as event, handled for sessionId/usage extraction + return null; + } + default: + return null; + } +} + +/** + * Map a Codex CLI stream event to Clawdbot agent event data. + * Returns null for events that should not be emitted. + */ +export function mapCodexStreamEvent( + event: CliStreamEvent, +): { stream: string; data: Record } | null { + const eventType = event.type; + + if (eventType === "item.created" || eventType === "item.started") { + const item = isRecord(event.item) ? event.item : null; + if (item && typeof item.type === "string" && item.type === "function_call") { + return { + stream: "tool", + data: { + phase: "start", + name: item.name, + id: item.id, + input: item.arguments, + }, + }; + } + } + + if (eventType === "item.completed") { + const item = isRecord(event.item) ? event.item : null; + if (item) { + if (typeof item.type === "string" && item.type === "function_call_output") { + return { + stream: "tool", + data: { + phase: "end", + id: item.call_id, + output: item.output, + }, + }; + } + if ( + typeof item.type === "string" && + item.type === "message" && + typeof item.text === "string" + ) { + return { + stream: "assistant", + data: { text: item.text }, + }; + } + } + } + + if (eventType === "turn.completed" || eventType === "thread.completed") { + // Lifecycle event, don't emit as assistant event + return null; + } + + return null; +} + +/** + * Map a generic CLI stream event to Clawdbot agent event data. + * Dispatches to the appropriate mapper based on detected CLI format. + */ +export function mapCliStreamEvent( + event: CliStreamEvent, + backendId: string, +): { stream: string; data: Record } | null { + log.info(`cli stream: mapCliStreamEvent called for type="${event.type}" backend="${backendId}"`); + + // Detect format from event types + let result: { stream: string; data: Record } | null; + if ( + backendId.includes("codex") || + event.type.startsWith("item.") || + event.type.startsWith("turn.") || + event.type.startsWith("thread.") + ) { + log.info(`cli stream: using Codex mapper`); + result = mapCodexStreamEvent(event); + } else { + log.info(`cli stream: using Claude mapper`); + result = mapClaudeStreamEvent(event); + } + + if (result) { + log.info( + `cli stream: mapped to stream="${result.stream}" data=${JSON.stringify(result.data).slice(0, 200)}`, + ); + } else { + log.info(`cli stream: mapper returned null for type="${event.type}"`); + } + + return result; +} diff --git a/src/config/types.agent-defaults.ts b/src/config/types.agent-defaults.ts index e55ea2461..f812dc0fe 100644 --- a/src/config/types.agent-defaults.ts +++ b/src/config/types.agent-defaults.ts @@ -97,6 +97,10 @@ export type CliBackendConfig = { cacheWrite?: string[]; total?: string[]; }; + /** Enable streaming output mode (NDJSON line-by-line parsing with real-time events). */ + streaming?: boolean; + /** Event types to emit when streaming. If omitted, all events are emitted. */ + streamingEventTypes?: string[]; }; export type AgentDefaultsConfig = { diff --git a/src/config/zod-schema.core.ts b/src/config/zod-schema.core.ts index 956e44ff3..4888c8f1e 100644 --- a/src/config/zod-schema.core.ts +++ b/src/config/zod-schema.core.ts @@ -275,6 +275,8 @@ export const CliBackendSchema = z total: z.array(z.string()).optional(), }) .optional(), + streaming: z.boolean().optional(), + streamingEventTypes: z.array(z.string()).optional(), }) .strict();