From da967db7c3f3b15d7607a2b1fad95924df5abb3d Mon Sep 17 00:00:00 2001 From: Ross Morsali Date: Mon, 26 Jan 2026 19:33:21 +0100 Subject: [PATCH] feat(cli): apply parseReplyDirectives to streaming text events Match embedded flow's text processing: extract media URLs, clean directives, compute delta from cleaned text. --- src/agents/cli-backends.ts | 57 ++++ src/agents/cli-runner/streaming.test.ts | 369 +++++++++++++++++++- src/agents/cli-runner/streaming.ts | 430 ++++++++++++++++++------ src/config/types.agent-defaults.ts | 53 +++ src/config/zod-schema.core.ts | 44 +++ 5 files changed, 837 insertions(+), 116 deletions(-) diff --git a/src/agents/cli-backends.ts b/src/agents/cli-backends.ts index cc433729f..1efc3b0c4 100644 --- a/src/agents/cli-backends.ts +++ b/src/agents/cli-backends.ts @@ -58,6 +58,30 @@ const DEFAULT_CLAUDE_BACKEND: CliBackendConfig = { }, streaming: true, streamingEventTypes: ["tool_use", "tool_result", "text", "result"], + streamingFormat: { + text: { + eventTypes: ["assistant"], + contentPath: "message.content", + matchType: "text", + textField: "text", + }, + toolUse: { + eventTypes: ["assistant"], + contentPath: "message.content", + matchType: "tool_use", + idField: "id", + nameField: "name", + inputField: "input", + }, + toolResult: { + eventTypes: ["user"], + contentPath: "message.content", + matchType: "tool_result", + idField: "tool_use_id", + outputField: "content", + isErrorField: "is_error", + }, + }, }; const DEFAULT_CODEX_BACKEND: CliBackendConfig = { @@ -89,6 +113,29 @@ const DEFAULT_CODEX_BACKEND: CliBackendConfig = { }, streaming: true, streamingEventTypes: ["item", "turn.completed"], + streamingFormat: { + text: { + eventTypes: ["item.completed"], + contentPath: "item", + matchType: "message", + textField: "text", + }, + toolUse: { + eventTypes: ["item.created", "item.started"], + contentPath: "item", + matchType: "function_call", + idField: "id", + nameField: "name", + inputField: "arguments", + }, + toolResult: { + eventTypes: ["item.completed"], + contentPath: "item", + matchType: "function_call_output", + idField: "call_id", + outputField: "output", + }, + }, }; function normalizeBackendKey(key: string): string { @@ -119,6 +166,16 @@ function mergeBackendConfig(base: CliBackendConfig, override?: CliBackendConfig) resumeArgs: override.resumeArgs ?? base.resumeArgs, streaming: override.streaming ?? base.streaming, streamingEventTypes: override.streamingEventTypes ?? base.streamingEventTypes, + streamingFormat: override.streamingFormat + ? { + text: { ...base.streamingFormat?.text, ...override.streamingFormat.text }, + toolUse: { ...base.streamingFormat?.toolUse, ...override.streamingFormat.toolUse }, + toolResult: { + ...base.streamingFormat?.toolResult, + ...override.streamingFormat.toolResult, + }, + } + : base.streamingFormat, }; } diff --git a/src/agents/cli-runner/streaming.test.ts b/src/agents/cli-runner/streaming.test.ts index aa46fd2e7..fa0421787 100644 --- a/src/agents/cli-runner/streaming.test.ts +++ b/src/agents/cli-runner/streaming.test.ts @@ -2,7 +2,7 @@ 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 type { CliBackendConfig, StreamingFormat } from "../../config/types.js"; import { runCliWithStreaming, mapClaudeStreamEvent, @@ -95,6 +95,71 @@ describe("runCliWithStreaming", () => { expect(events[1]).toEqual({ type: "text", text: "world!" }); }); + it("emits cumulative text with delta when using streamingFormat config", async () => { + const proc = createMockProcess(); + spawnMock.mockReturnValue(proc); + + const backendWithFormat: CliBackendConfig = { + ...defaultBackend, + streamingFormat: { + text: { + eventTypes: ["assistant"], + contentPath: "message.content", + matchType: "text", + textField: "text", + }, + }, + }; + + const events: CliStreamEvent[] = []; + const promise = runCliWithStreaming({ + command: "claude", + args: ["-p"], + cwd: "/tmp", + env: {}, + timeoutMs: 5000, + backend: backendWithFormat, + onEvent: (e) => events.push(e), + }); + + await nextTick(); + + // Claude CLI format: assistant events with nested content + proc.stdout.push( + '{"type":"assistant","message":{"content":[{"type":"text","text":"Hello "}]}}\n', + ); + proc.stdout.push( + '{"type":"assistant","message":{"content":[{"type":"text","text":"world!"}]}}\n', + ); + proc.stdout.push(null); + + await nextTick(); + proc.emit("close", 0, null); + + const result = await promise; + + expect(result.text).toBe("Hello world!"); + + // Find text events (onEvent receives extracted events) + const textEvents = events.filter((e) => e.type === "text"); + expect(textEvents).toHaveLength(2); + + // First text event: cumulative = "Hello" (trimmed), delta = "Hello" + // parseReplyDirectives trims trailing whitespace + expect(textEvents[0]).toEqual({ + type: "text", + text: "Hello", + delta: "Hello", + }); + + // Second text event: cumulative = "Hello world!", delta = " world!" (space preserved in delta) + expect(textEvents[1]).toEqual({ + type: "text", + text: "Hello world!", + delta: " world!", + }); + }); + it("filters events by type when streamingEventTypes is specified", async () => { const proc = createMockProcess(); spawnMock.mockReturnValue(proc); @@ -286,6 +351,80 @@ describe("runCliWithStreaming", () => { const result = await promise; expect(result.sessionId).toBe("thread-abc"); }); + + it("extracts tool_use and tool_result with streamingFormat config", async () => { + const proc = createMockProcess(); + spawnMock.mockReturnValue(proc); + + const backendWithFormat: CliBackendConfig = { + ...defaultBackend, + streamingFormat: { + toolUse: { + eventTypes: ["assistant"], + contentPath: "message.content", + matchType: "tool_use", + idField: "id", + nameField: "name", + inputField: "input", + }, + toolResult: { + eventTypes: ["user"], + contentPath: "message.content", + matchType: "tool_result", + idField: "tool_use_id", + outputField: "content", + isErrorField: "is_error", + }, + }, + }; + + const events: CliStreamEvent[] = []; + const promise = runCliWithStreaming({ + command: "claude", + args: ["-p"], + cwd: "/tmp", + env: {}, + timeoutMs: 5000, + backend: backendWithFormat, + onEvent: (e) => events.push(e), + }); + + await nextTick(); + + // Tool use event + proc.stdout.push( + '{"type":"assistant","message":{"content":[{"type":"tool_use","id":"tool-1","name":"bash","input":{"cmd":"ls"}}]}}\n', + ); + // Tool result event + proc.stdout.push( + '{"type":"user","message":{"content":[{"type":"tool_result","tool_use_id":"tool-1","content":"file.txt","is_error":false}]}}\n', + ); + proc.stdout.push(null); + + await nextTick(); + proc.emit("close", 0, null); + + await promise; + + const toolUseEvents = events.filter((e) => e.type === "tool_use"); + const toolResultEvents = events.filter((e) => e.type === "tool_result"); + + expect(toolUseEvents).toHaveLength(1); + expect(toolUseEvents[0]).toMatchObject({ + type: "tool_use", + id: "tool-1", + name: "bash", + input: { cmd: "ls" }, + }); + + expect(toolResultEvents).toHaveLength(1); + expect(toolResultEvents[0]).toMatchObject({ + type: "tool_result", + tool_use_id: "tool-1", + content: "file.txt", + is_error: false, + }); + }); }); describe("mapClaudeStreamEvent", () => { @@ -331,14 +470,25 @@ describe("mapClaudeStreamEvent", () => { }); }); - it("maps text event to assistant delta", () => { + it("maps text event to assistant with cumulative text and delta", () => { + const event: CliStreamEvent = { type: "text", text: "Hello world!", delta: "world!" }; + + const result = mapClaudeStreamEvent(event); + + expect(result).toEqual({ + stream: "assistant", + data: { text: "Hello world!", delta: "world!" }, + }); + }); + + it("maps text event without explicit delta (delta defaults to text)", () => { const event: CliStreamEvent = { type: "text", text: "Hello" }; const result = mapClaudeStreamEvent(event); expect(result).toEqual({ stream: "assistant", - data: { text: "Hello", delta: true }, + data: { text: "Hello", delta: "Hello" }, }); }); @@ -400,7 +550,7 @@ describe("mapCodexStreamEvent", () => { }); }); - it("maps item.completed message to assistant text", () => { + it("maps item.completed message to assistant text with delta", () => { const event: CliStreamEvent = { type: "item.completed", item: { @@ -413,7 +563,7 @@ describe("mapCodexStreamEvent", () => { expect(result).toEqual({ stream: "assistant", - data: { text: "Task completed" }, + data: { text: "Task completed", delta: "Task completed" }, }); }); @@ -424,13 +574,53 @@ describe("mapCodexStreamEvent", () => { }); describe("mapCliStreamEvent", () => { - it("uses Claude mapper for claude-cli backend", () => { - const event: CliStreamEvent = { type: "text", text: "Hello" }; + it("maps text event with cumulative text and delta", () => { + const event: CliStreamEvent = { type: "text", text: "Hello world!", delta: "world!" }; const result = mapCliStreamEvent(event, "claude-cli"); expect(result).toEqual({ stream: "assistant", - data: { text: "Hello", delta: true }, + data: { text: "Hello world!", delta: "world!" }, + }); + }); + + it("maps tool_use event directly", () => { + const event: CliStreamEvent = { + type: "tool_use", + id: "tool-1", + name: "bash", + input: { command: "ls" }, + }; + const result = mapCliStreamEvent(event, "claude-cli"); + + expect(result).toEqual({ + stream: "tool", + data: { + phase: "start", + name: "bash", + id: "tool-1", + input: { command: "ls" }, + }, + }); + }); + + it("maps tool_result event directly", () => { + const event: CliStreamEvent = { + type: "tool_result", + tool_use_id: "tool-1", + content: "output", + is_error: true, + }; + const result = mapCliStreamEvent(event, "claude-cli"); + + expect(result).toEqual({ + stream: "tool", + data: { + phase: "end", + id: "tool-1", + output: "output", + isError: true, + }, }); }); @@ -443,7 +633,7 @@ describe("mapCliStreamEvent", () => { expect(result).toEqual({ stream: "assistant", - data: { text: "Done" }, + data: { text: "Done", delta: "Done" }, }); }); @@ -460,4 +650,165 @@ describe("mapCliStreamEvent", () => { data: { phase: "start", name: "test", id: "1", input: undefined }, }); }); + + it("accepts optional format parameter", () => { + const format: StreamingFormat = { + text: { eventTypes: ["text"], textField: "text" }, + }; + const event: CliStreamEvent = { type: "text", text: "Hello", delta: "Hello" }; + const result = mapCliStreamEvent(event, "claude-cli", format); + + expect(result).toEqual({ + stream: "assistant", + data: { text: "Hello", delta: "Hello" }, + }); + }); + + it("passes through mediaUrls when present", () => { + const event: CliStreamEvent = { + type: "text", + text: "Here is an image", + delta: "image", + mediaUrls: ["https://example.com/img.png"], + }; + const result = mapCliStreamEvent(event, "claude-cli"); + + expect(result).toEqual({ + stream: "assistant", + data: { + text: "Here is an image", + delta: "image", + mediaUrls: ["https://example.com/img.png"], + }, + }); + }); + + it("omits mediaUrls when empty array", () => { + const event: CliStreamEvent = { + type: "text", + text: "Hello", + delta: "Hello", + mediaUrls: [], + }; + const result = mapCliStreamEvent(event, "claude-cli"); + + expect(result).toEqual({ + stream: "assistant", + data: { text: "Hello", delta: "Hello" }, + }); + }); +}); + +describe("runCliWithStreaming - directive parsing", () => { + const defaultBackend: CliBackendConfig = { + command: "claude", + sessionIdFields: ["session_id"], + }; + + beforeEach(() => { + spawnMock.mockReset(); + }); + + it("parses reply directives and extracts mediaUrls from text", async () => { + const proc = createMockProcess(); + spawnMock.mockReturnValue(proc); + + const backendWithFormat: CliBackendConfig = { + ...defaultBackend, + streamingFormat: { + text: { + eventTypes: ["assistant"], + contentPath: "message.content", + matchType: "text", + textField: "text", + }, + }, + }; + + const events: CliStreamEvent[] = []; + const promise = runCliWithStreaming({ + command: "claude", + args: ["-p"], + cwd: "/tmp", + env: {}, + timeoutMs: 5000, + backend: backendWithFormat, + onEvent: (e) => events.push(e), + }); + + await nextTick(); + + // Text with MEDIA token should have media extracted and text cleaned + proc.stdout.push( + '{"type":"assistant","message":{"content":[{"type":"text","text":"Here is the image\\nMEDIA: https://example.com/img.png"}]}}\n', + ); + proc.stdout.push(null); + + await nextTick(); + proc.emit("close", 0, null); + + await promise; + + const textEvents = events.filter((e) => e.type === "text"); + expect(textEvents).toHaveLength(1); + expect(textEvents[0]?.text).toBe("Here is the image"); + expect(textEvents[0]?.mediaUrls).toEqual(["https://example.com/img.png"]); + }); + + it("computes delta correctly after directive parsing", async () => { + const proc = createMockProcess(); + spawnMock.mockReturnValue(proc); + + const backendWithFormat: CliBackendConfig = { + ...defaultBackend, + streamingFormat: { + text: { + eventTypes: ["assistant"], + contentPath: "message.content", + matchType: "text", + textField: "text", + }, + }, + }; + + const events: CliStreamEvent[] = []; + const promise = runCliWithStreaming({ + command: "claude", + args: ["-p"], + cwd: "/tmp", + env: {}, + timeoutMs: 5000, + backend: backendWithFormat, + onEvent: (e) => events.push(e), + }); + + await nextTick(); + + // First chunk + proc.stdout.push( + '{"type":"assistant","message":{"content":[{"type":"text","text":"Hello "}]}}\n', + ); + // Second chunk + proc.stdout.push( + '{"type":"assistant","message":{"content":[{"type":"text","text":"world!"}]}}\n', + ); + proc.stdout.push(null); + + await nextTick(); + proc.emit("close", 0, null); + + await promise; + + const textEvents = events.filter((e) => e.type === "text"); + expect(textEvents).toHaveLength(2); + + // First event: text="Hello" (trimmed), delta="Hello" + // parseReplyDirectives trims trailing whitespace + expect(textEvents[0]?.text).toBe("Hello"); + expect(textEvents[0]?.delta).toBe("Hello"); + + // Second event: text="Hello world!", delta=" world!" (space preserved in delta) + expect(textEvents[1]?.text).toBe("Hello world!"); + expect(textEvents[1]?.delta).toBe(" world!"); + }); }); diff --git a/src/agents/cli-runner/streaming.ts b/src/agents/cli-runner/streaming.ts index d4b168b6d..e06b7c6e6 100644 --- a/src/agents/cli-runner/streaming.ts +++ b/src/agents/cli-runner/streaming.ts @@ -1,13 +1,38 @@ import { type ChildProcessWithoutNullStreams, spawn } from "node:child_process"; import { createInterface, type Interface } from "node:readline"; -import type { CliBackendConfig } from "../../config/types.js"; +import { parseReplyDirectives } from "../../auto-reply/reply/reply-directives.js"; +import type { + CliBackendConfig, + StreamingFormat, + StreamingFormatText, + StreamingFormatToolResult, + StreamingFormatToolUse, +} from "../../config/types.js"; import { createSubsystemLogger } from "../../logging/subsystem.js"; const log = createSubsystemLogger("agent/cli-streaming"); export type CliStreamEvent = { type: string; + /** Cumulative text (for text events). */ + text?: string; + /** Incremental text delta (for text events). */ + delta?: string; + /** Extracted media URLs (for text events). */ + mediaUrls?: string[]; + /** Tool ID. */ + id?: string; + /** Tool name. */ + name?: string; + /** Tool input. */ + input?: unknown; + /** Tool use ID reference (for tool results). */ + tool_use_id?: string; + /** Tool result content. */ + content?: unknown; + /** Whether the tool result is an error. */ + is_error?: boolean; [key: string]: unknown; }; @@ -38,6 +63,11 @@ export type CliStreamParams = { onEvent: (event: CliStreamEvent) => void; }; +export type MappedCliEvent = { + stream: string; + data: Record; +}; + function isRecord(value: unknown): value is Record { return Boolean(value && typeof value === "object" && !Array.isArray(value)); } @@ -101,12 +131,52 @@ function matchesEventType(eventType: string, filters: string[]): boolean { return false; } +/** Navigate to a nested value via dot-notation path (e.g., "message.content"). */ +function getByPath(obj: Record, path: string): unknown { + if (!path) return obj; + const parts = path.split("."); + let current: unknown = obj; + for (const part of parts) { + if (!isRecord(current)) return undefined; + current = current[part]; + } + return current; +} + +/** Extract matching items from parsed event using format config. */ +function extractByFormat( + parsed: Record, + eventType: string, + format: StreamingFormatText | StreamingFormatToolUse | StreamingFormatToolResult | undefined, +): Record[] { + if (!format) return []; + + // Check if event type matches + if (format.eventTypes && format.eventTypes.length > 0) { + if (!matchesEventType(eventType, format.eventTypes)) return []; + } + + // Navigate to content via contentPath + const content = getByPath(parsed, format.contentPath ?? ""); + + // Handle array or single object + const items = Array.isArray(content) ? content : content && isRecord(content) ? [content] : []; + + // Filter by matchType if specified + return items.filter((item) => { + if (!isRecord(item)) return false; + if (!format.matchType) return true; + return item.type === format.matchType; + }) as Record[]; +} + /** * 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; + const format = backend.streamingFormat; return new Promise((resolve, reject) => { let child: ChildProcessWithoutNullStreams | null = null; @@ -116,6 +186,8 @@ export async function runCliWithStreaming(params: CliStreamParams): Promise 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)`); + // Config-driven extraction if format is available + if (format) { + // Extract text blocks + const textBlocks = extractByFormat(parsed, eventType, format.text); + for (const block of textBlocks) { + const textField = format.text?.textField ?? "text"; + const newText = block[textField]; + if (typeof newText === "string" && newText) { + textParts.push(newText); + accumulatedText += newText; + + log.info( + `cli stream: accumulated text chunk (${newText.length} chars): "${newText.slice(0, 100)}${newText.length > 100 ? "..." : ""}"`, + ); + + // Emit text event with cumulative + delta + if (shouldEmit || matchesEventType("text", eventTypes ?? [])) { + // Parse directives to match embedded flow (extracts media, cleans text) + const { text: cleanedAccumulated, mediaUrls } = parseReplyDirectives(accumulatedText); + const { text: cleanedPrevious } = parseReplyDirectives(previousAccumulated); + const cleanedDelta = cleanedAccumulated.startsWith(cleanedPrevious) + ? cleanedAccumulated.slice(cleanedPrevious.length) + : cleanedAccumulated; + previousAccumulated = accumulatedText; + + const textEvent: CliStreamEvent = { + type: "text", + text: cleanedAccumulated, + delta: cleanedDelta, + ...(mediaUrls?.length ? { mediaUrls } : {}), + }; + log.info(`cli stream: emitting text event (cumulative=${cleanedAccumulated.length})`); + try { + onEvent(textEvent); + } catch (err) { + log.info( + `cli stream: onEvent error: ${err instanceof Error ? err.message : String(err)}`, + ); + } } } } - } else if (eventType === "result") { - // Claude CLI final result event - usage already extracted above + + // Extract tool use blocks + const toolUseBlocks = extractByFormat(parsed, eventType, format.toolUse); + for (const block of toolUseBlocks) { + const idField = format.toolUse?.idField ?? "id"; + const nameField = format.toolUse?.nameField ?? "name"; + const inputField = format.toolUse?.inputField ?? "input"; + + const toolEvent: CliStreamEvent = { + type: "tool_use", + id: block[idField] as string | undefined, + name: block[nameField] as string | undefined, + input: block[inputField], + }; + log.info( + `cli stream: found tool_use name="${String(toolEvent.name)}" id="${String(toolEvent.id)}"`, + ); + if (shouldEmit || matchesEventType("tool_use", eventTypes ?? [])) { + log.info(`cli stream: emitting tool_use event`); + try { + onEvent(toolEvent); + } catch (err) { + log.info( + `cli stream: onEvent error: ${err instanceof Error ? err.message : String(err)}`, + ); + } + } + } + + // Extract tool result blocks + const toolResultBlocks = extractByFormat(parsed, eventType, format.toolResult); + for (const block of toolResultBlocks) { + const idField = format.toolResult?.idField ?? "tool_use_id"; + const outputField = format.toolResult?.outputField ?? "content"; + const isErrorField = format.toolResult?.isErrorField; + + const toolResultEvent: CliStreamEvent = { + type: "tool_result", + tool_use_id: block[idField] as string | undefined, + content: block[outputField], + is_error: isErrorField ? (block[isErrorField] as boolean) : undefined, + }; + log.info(`cli stream: found tool_result id="${String(toolResultEvent.tool_use_id)}"`); + if (shouldEmit || matchesEventType("tool_result", eventTypes ?? [])) { + log.info(`cli stream: emitting tool_result event`); + try { + onEvent(toolResultEvent); + } catch (err) { + log.info( + `cli stream: onEvent error: ${err instanceof Error ? err.message : String(err)}`, + ); + } + } + } + } else { + // Fallback: legacy hardcoded text accumulation for backwards compatibility + // Note: legacy path emits raw events, not extracted events + accumulateLegacyText(parsed, eventType, textParts); + + // Emit raw event (original behavior) + if (shouldEmit) { + log.info(`cli stream: emitting raw 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)}`, + ); + } + } + } + + // Handle Codex thread_id extraction (not in standard format config) + if ( + (eventType === "turn.completed" || eventType === "thread.completed") && + !sessionId && + typeof parsed.thread_id === "string" + ) { + sessionId = parsed.thread_id.trim(); + log.info(`cli stream: extracted thread_id as sessionId=${sessionId}`); + } + + // Handle result event final text (Claude CLI) + if (eventType === "result") { const result = isRecord(parsed.result) ? parsed.result : parsed; - if (typeof result.text === "string") { + if (typeof result.text === "string" && result.text) { 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) => { @@ -328,13 +465,128 @@ export async function runCliWithStreaming(params: CliStreamParams): Promise, + eventType: string, + textParts: string[], +): void { + // Handle Claude CLI event types + if (eventType === "text" || eventType === "content_block_delta") { + 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") { + 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)) continue; + const blockType = typeof block.type === "string" ? block.type : ""; + if (blockType === "text" && typeof block.text === "string") { + textParts.push(block.text); + log.info(`cli stream: accumulated text block (${block.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)`); + } + } + } +} + +/** + * Map a CLI stream event to Clawdbot agent event data. + * Uses config-driven extraction when available, falls back to legacy mappers. + */ +export function mapCliStreamEvent( + event: CliStreamEvent, + backendId: string, + _format?: StreamingFormat, +): MappedCliEvent | null { + log.info(`cli stream: mapCliStreamEvent called for type="${event.type}" backend="${backendId}"`); + + // For text events with cumulative text, use the embedded flow pattern + if (event.type === "text") { + const text = typeof event.text === "string" ? event.text : ""; + const delta = typeof event.delta === "string" ? event.delta : text; + if (!text && !delta) return null; + return { + stream: "assistant", + data: { + text, + delta, + mediaUrls: event.mediaUrls?.length ? event.mediaUrls : undefined, + }, + }; + } + + // For tool_use events + if (event.type === "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, + }, + }; + } + + // For tool_result events + if (event.type === "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, + }, + }; + } + + // Legacy fallback for unknown event types - detect format from event type or backend name + if ( + backendId.includes("codex") || + event.type.startsWith("item.") || + event.type.startsWith("turn.") || + event.type.startsWith("thread.") + ) { + log.info(`cli stream: using Codex mapper for legacy event`); + return mapCodexStreamEvent(event); + } + + log.info(`cli stream: using Claude mapper for legacy event`); + return mapClaudeStreamEvent(event); +} + /** * Map a Claude CLI stream event to Clawdbot agent event data. * Returns null for events that should not be emitted. + * @deprecated Use mapCliStreamEvent with streamingFormat config instead. */ -export function mapClaudeStreamEvent( - event: CliStreamEvent, -): { stream: string; data: Record } | null { +export function mapClaudeStreamEvent(event: CliStreamEvent): MappedCliEvent | null { switch (event.type) { case "tool_use": { const toolName = typeof event.name === "string" ? event.name : "unknown"; @@ -364,10 +616,11 @@ export function mapClaudeStreamEvent( case "text": case "content_block_delta": { const text = typeof event.text === "string" ? event.text : ""; - if (!text) return null; + const delta = typeof event.delta === "string" ? event.delta : text; + if (!text && !delta) return null; return { stream: "assistant", - data: { text, delta: true }, + data: { text, delta }, }; } case "assistant": @@ -387,10 +640,9 @@ export function mapClaudeStreamEvent( /** * Map a Codex CLI stream event to Clawdbot agent event data. * Returns null for events that should not be emitted. + * @deprecated Use mapCliStreamEvent with streamingFormat config instead. */ -export function mapCodexStreamEvent( - event: CliStreamEvent, -): { stream: string; data: Record } | null { +export function mapCodexStreamEvent(event: CliStreamEvent): MappedCliEvent | null { const eventType = event.type; if (eventType === "item.created" || eventType === "item.started") { @@ -428,7 +680,7 @@ export function mapCodexStreamEvent( ) { return { stream: "assistant", - data: { text: item.text }, + data: { text: item.text, delta: item.text }, }; } } @@ -441,39 +693,3 @@ export function mapCodexStreamEvent( 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 f812dc0fe..142d632b5 100644 --- a/src/config/types.agent-defaults.ts +++ b/src/config/types.agent-defaults.ts @@ -12,6 +12,57 @@ import type { } from "./types.sandbox.js"; import type { MemorySearchConfig } from "./types.tools.js"; +// Streaming format types for CLI backend configuration +export type StreamingFormatText = { + /** Top-level event types containing text (e.g., ["assistant"], ["item.completed"]). */ + eventTypes?: string[]; + /** Path to content array/object (e.g., "message.content", "item"). */ + contentPath?: string; + /** Type value to match in content blocks (e.g., "text"). */ + matchType?: string; + /** Field containing text (e.g., "text"). */ + textField?: string; +}; + +export type StreamingFormatToolUse = { + /** Top-level event types containing tool use (e.g., ["assistant"], ["item.created"]). */ + eventTypes?: string[]; + /** Path to content array/object (e.g., "message.content", "item"). */ + contentPath?: string; + /** Type value to match (e.g., "tool_use", "function_call"). */ + matchType?: string; + /** Field containing tool ID. */ + idField?: string; + /** Field containing tool name. */ + nameField?: string; + /** Field containing tool input/arguments. */ + inputField?: string; +}; + +export type StreamingFormatToolResult = { + /** Top-level event types containing tool results (e.g., ["user"], ["item.completed"]). */ + eventTypes?: string[]; + /** Path to content array/object (e.g., "message.content", "item"). */ + contentPath?: string; + /** Type value to match (e.g., "tool_result", "function_call_output"). */ + matchType?: string; + /** Field containing tool use ID reference. */ + idField?: string; + /** Field containing result output. */ + outputField?: string; + /** Field indicating if result is an error. */ + isErrorField?: string; +}; + +export type StreamingFormat = { + /** Format for extracting text content from events. */ + text?: StreamingFormatText; + /** Format for extracting tool use from events. */ + toolUse?: StreamingFormatToolUse; + /** Format for extracting tool results from events. */ + toolResult?: StreamingFormatToolResult; +}; + export type AgentModelEntryConfig = { alias?: string; /** Provider-specific API parameters (e.g., GLM-4.7 thinking mode). */ @@ -101,6 +152,8 @@ export type CliBackendConfig = { streaming?: boolean; /** Event types to emit when streaming. If omitted, all events are emitted. */ streamingEventTypes?: string[]; + /** Configurable format for parsing streaming events (text, tool_use, tool_result). */ + streamingFormat?: StreamingFormat; }; export type AgentDefaultsConfig = { diff --git a/src/config/zod-schema.core.ts b/src/config/zod-schema.core.ts index 4888c8f1e..d2b102987 100644 --- a/src/config/zod-schema.core.ts +++ b/src/config/zod-schema.core.ts @@ -239,6 +239,49 @@ export const HumanDelaySchema = z }) .strict(); +export const StreamingFormatTextSchema = z + .object({ + eventTypes: z.array(z.string()).optional(), + contentPath: z.string().optional(), + matchType: z.string().optional(), + textField: z.string().optional(), + }) + .strict() + .optional(); + +export const StreamingFormatToolUseSchema = z + .object({ + eventTypes: z.array(z.string()).optional(), + contentPath: z.string().optional(), + matchType: z.string().optional(), + idField: z.string().optional(), + nameField: z.string().optional(), + inputField: z.string().optional(), + }) + .strict() + .optional(); + +export const StreamingFormatToolResultSchema = z + .object({ + eventTypes: z.array(z.string()).optional(), + contentPath: z.string().optional(), + matchType: z.string().optional(), + idField: z.string().optional(), + outputField: z.string().optional(), + isErrorField: z.string().optional(), + }) + .strict() + .optional(); + +export const StreamingFormatSchema = z + .object({ + text: StreamingFormatTextSchema, + toolUse: StreamingFormatToolUseSchema, + toolResult: StreamingFormatToolResultSchema, + }) + .strict() + .optional(); + export const CliBackendSchema = z .object({ command: z.string(), @@ -277,6 +320,7 @@ export const CliBackendSchema = z .optional(), streaming: z.boolean().optional(), streamingEventTypes: z.array(z.string()).optional(), + streamingFormat: StreamingFormatSchema, }) .strict();