feat(cli): add streaming NDJSON support for CLI backends
Adds real-time streaming output support to CLI backends, enabling line-by-line parsing of NDJSON output instead of waiting for the full response. This brings CLI backends closer to the embedded/API flow by emitting events as they arrive. Key changes: - New streaming execution module (cli-runner/streaming.ts): - Uses readline to parse NDJSON lines as they arrive - Extracts session IDs, usage stats, and text from stream - Supports event type filtering with prefix matching - Maps CLI-specific events to Clawdbot agent events - Config extension: - Added `streaming?: boolean` to enable streaming mode - Added `streamingEventTypes?: string[]` to filter events - Claude CLI defaults: stream-json format with --verbose flag - Codex CLI defaults: streaming enabled with item/turn events - Event mapping for different CLI formats: - Claude CLI: tool_use, tool_result, text, result events - Codex CLI: item.*, turn.completed, thread.completed events - Debug logging throughout the pipeline: - Logs raw JSON lines, parsed types, session/usage extraction - Logs event emission and mapping decisions - Helps diagnose streaming issues in production The streaming path is enabled by default for Claude CLI and Codex CLI. Users can disable it by setting `streaming: false` in their config. Non-streaming path via runCommandWithTimeout remains available.
This commit is contained in:
Ross Morsali
parent
7fc6476688
commit
c312516392
@ -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<T>() {
|
||||
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<StreamResult>();
|
||||
const secondDeferred = createDeferred<StreamResult>();
|
||||
|
||||
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]);
|
||||
});
|
||||
|
||||
109
src/agents/cli-backends.test.ts
Normal file
109
src/agents/cli-backends.test.ts
Normal file
@ -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");
|
||||
});
|
||||
});
|
||||
});
|
||||
@ -27,16 +27,17 @@ const CLAUDE_MODEL_ALIASES: Record<string, string> = {
|
||||
|
||||
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,
|
||||
};
|
||||
}
|
||||
|
||||
|
||||
@ -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({
|
||||
|
||||
@ -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,
|
||||
|
||||
463
src/agents/cli-runner/streaming.test.ts
Normal file
463
src/agents/cli-runner/streaming.test.ts
Normal file
@ -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<void> {
|
||||
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 },
|
||||
});
|
||||
});
|
||||
});
|
||||
479
src/agents/cli-runner/streaming.ts
Normal file
479
src/agents/cli-runner/streaming.ts
Normal file
@ -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<string, unknown> {
|
||||
return Boolean(value && typeof value === "object" && !Array.isArray(value));
|
||||
}
|
||||
|
||||
function pickSessionId(
|
||||
parsed: Record<string, unknown>,
|
||||
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<string, unknown>, 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<CliStreamResult> {
|
||||
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<string, unknown>;
|
||||
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<string, unknown> } | 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<string, unknown> } | 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<string, unknown> } | null {
|
||||
log.info(`cli stream: mapCliStreamEvent called for type="${event.type}" backend="${backendId}"`);
|
||||
|
||||
// Detect format from event types
|
||||
let result: { stream: string; data: Record<string, unknown> } | 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;
|
||||
}
|
||||
@ -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 = {
|
||||
|
||||
@ -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();
|
||||
|
||||
|
||||
Loading…
Reference in New Issue
Block a user