feat(cli): apply parseReplyDirectives to streaming text events
Match embedded flow's text processing: extract media URLs, clean directives, compute delta from cleaned text.
This commit is contained in:
Ross Morsali
parent
c312516392
commit
da967db7c3
@ -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,
|
||||
};
|
||||
}
|
||||
|
||||
|
||||
@ -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!");
|
||||
});
|
||||
});
|
||||
|
||||
@ -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<string, unknown>;
|
||||
};
|
||||
|
||||
function isRecord(value: unknown): value is Record<string, unknown> {
|
||||
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<string, unknown>, 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<string, unknown>,
|
||||
eventType: string,
|
||||
format: StreamingFormatText | StreamingFormatToolUse | StreamingFormatToolResult | undefined,
|
||||
): Record<string, unknown>[] {
|
||||
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<string, unknown>[];
|
||||
}
|
||||
|
||||
/**
|
||||
* 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;
|
||||
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<CliS
|
||||
|
||||
const events: CliStreamEvent[] = [];
|
||||
const textParts: string[] = [];
|
||||
let accumulatedText = "";
|
||||
let previousAccumulated = "";
|
||||
let sessionId: string | undefined;
|
||||
let usage: CliUsage | undefined;
|
||||
let stderrBuffer = "";
|
||||
@ -218,6 +290,7 @@ export async function runCliWithStreaming(params: CliStreamParams): Promise<CliS
|
||||
return;
|
||||
}
|
||||
|
||||
// Create base event
|
||||
const event: CliStreamEvent = { type: eventType, ...parsed };
|
||||
events.push(event);
|
||||
log.info(`cli stream: created event #${events.length} type="${eventType}"`);
|
||||
@ -229,75 +302,139 @@ export async function runCliWithStreaming(params: CliStreamParams): Promise<CliS
|
||||
`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)`);
|
||||
// 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<CliS
|
||||
});
|
||||
}
|
||||
|
||||
/** Legacy text accumulation for backends without streamingFormat config. */
|
||||
function accumulateLegacyText(
|
||||
parsed: Record<string, unknown>,
|
||||
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<string, unknown> } | 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<string, unknown> } | 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<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;
|
||||
}
|
||||
|
||||
@ -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 = {
|
||||
|
||||
@ -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();
|
||||
|
||||
|
||||
Loading…
Reference in New Issue
Block a user