diff --git a/docs/docs_map.md b/docs/docs_map.md index fecb621adbec..fd9b5766fe9d 100644 --- a/docs/docs_map.md +++ b/docs/docs_map.md @@ -5863,6 +5863,7 @@ Do not edit it by hand; run `pnpm docs:map:gen`. - H2: Minimal backend plugin - H2: Config shape - H2: Advanced backend hooks + - H3: parseJsonlEvent: provider-specific JSONL streams - H3: ownsNativeCompaction: opting out of OpenClaw compaction - H2: MCP tool bridge - H2: Selecting the backend diff --git a/docs/plugins/cli-backend-plugins.md b/docs/plugins/cli-backend-plugins.md index 7d198dcea3c5..72eecd66ee12 100644 --- a/docs/plugins/cli-backend-plugins.md +++ b/docs/plugins/cli-backend-plugins.md @@ -304,6 +304,23 @@ preserves it through the `2026.8.x` line. New and updated plugins should use can `toolAvailabilityEnforcement: "execution-args"` explicitly; the beta compatibility path is scheduled for removal after that window. +### `parseJsonlEvent`: provider-specific JSONL streams + +Set `parseJsonlEvent` when a backend emits line-delimited JSON that does not +match the built-in Claude, Codex, or Gemini dialects. The hook receives one raw +line plus the resolved backend id and config, and returns one normalized event, +multiple events, or `null` to let the built-in parser try the line. + +Supported events are incremental assistant text, incremental thinking, native +tool start/result display, session ids, and terminal results. Terminal results +may include final text, usage, an error, and a successor session id. Session ids +reported by either event shape participate in resumed-session and fork +persistence. + +Tool events describe work the backend already performed. OpenClaw renders and +summarizes them, but does not treat them as host tool execution, trusted +diagnostics, loopback correlation, or message-delivery evidence. + ### `ownsNativeCompaction`: opting out of OpenClaw compaction If your backend runs an agent that compacts its **own** transcript, set diff --git a/src/agents/cli-backends.test.ts b/src/agents/cli-backends.test.ts index 62c8999d6001..a847877c9a28 100644 --- a/src/agents/cli-backends.test.ts +++ b/src/agents/cli-backends.test.ts @@ -122,6 +122,16 @@ describe("resolveCliBackendConfig", () => { }); }); + it("preserves the plugin-owned JSONL parser through runtime resolution", () => { + const parseJsonlEvent = vi.fn(); + cliBackendsTesting.setDepsForTest({ + resolveRuntimeCliBackends: () => [runtimeEntry({ parseJsonlEvent })], + resolvePluginSetupCliBackend: () => undefined, + }); + + expect(requireBackend().parseJsonlEvent).toBe(parseJsonlEvent); + }); + it("normalizes the registered adapter with agent and runtime config context", () => { const normalizeConfig = vi.fn( (config: CliBackendConfig): CliBackendConfig => ({ @@ -167,7 +177,11 @@ describe("resolveCliBackendConfig", () => { }); it("falls back to setup registration before runtime activation", () => { - const entry = setupEntry({ config: { command: "setup-acme", args: ["run"] } }); + const parseJsonlEvent = vi.fn(); + const entry = setupEntry({ + config: { command: "setup-acme", args: ["run"] }, + parseJsonlEvent, + }); cliBackendsTesting.setDepsForTest({ resolveRuntimeCliBackends: () => [], resolvePluginSetupCliBackend: ({ backend }) => (backend === "acme-cli" ? entry : undefined), @@ -178,6 +192,7 @@ describe("resolveCliBackendConfig", () => { expect(resolved.pluginId).toBeUndefined(); expect(resolved.config).toEqual({ command: "setup-acme", args: ["run"] }); expect(resolved.runtimeArtifact).toEqual(runtimeArtifact); + expect(resolved.parseJsonlEvent).toBe(parseJsonlEvent); }); it("returns null when no plugin owns the backend", () => { diff --git a/src/agents/cli-backends.ts b/src/agents/cli-backends.ts index 2059db37ad7b..a4928a601af8 100644 --- a/src/agents/cli-backends.ts +++ b/src/agents/cli-backends.ts @@ -57,6 +57,7 @@ export type ResolvedCliBackend = { ownsNativeCompaction?: boolean; prepareExecution?: CliBackendPlugin["prepareExecution"]; resolveExecutionArgs?: CliBackendPlugin["resolveExecutionArgs"]; + parseJsonlEvent?: CliBackendPlugin["parseJsonlEvent"]; toolAvailabilityEnforcement?: CliBackendToolAvailabilityEnforcement; nativeToolMode?: CliBackendNativeToolMode; sideQuestionToolMode?: CliBackendSideQuestionToolMode; @@ -96,6 +97,7 @@ type FallbackCliBackendPolicy = { ownsNativeCompaction?: boolean; prepareExecution?: CliBackendPlugin["prepareExecution"]; resolveExecutionArgs?: CliBackendPlugin["resolveExecutionArgs"]; + parseJsonlEvent?: CliBackendPlugin["parseJsonlEvent"]; toolAvailabilityEnforcement?: CliBackendToolAvailabilityEnforcement; nativeToolMode?: CliBackendNativeToolMode; sideQuestionToolMode?: CliBackendSideQuestionToolMode; @@ -162,6 +164,7 @@ function resolveSetupCliBackendPolicy(provider: string): FallbackCliBackendPolic ownsNativeCompaction: entry.backend.ownsNativeCompaction, prepareExecution: entry.backend.prepareExecution, resolveExecutionArgs: entry.backend.resolveExecutionArgs, + parseJsonlEvent: entry.backend.parseJsonlEvent, toolAvailabilityEnforcement: entry.backend.toolAvailabilityEnforcement, nativeToolMode: entry.backend.nativeToolMode, sideQuestionToolMode: entry.backend.sideQuestionToolMode, @@ -403,6 +406,7 @@ export function resolveCliBackendConfig( ownsNativeCompaction: registered.ownsNativeCompaction, prepareExecution: registered.prepareExecution, resolveExecutionArgs: registered.resolveExecutionArgs, + parseJsonlEvent: registered.parseJsonlEvent, toolAvailabilityEnforcement: resolveToolAvailabilityEnforcement(registered), nativeToolMode: registered.nativeToolMode, sideQuestionToolMode: registered.sideQuestionToolMode, @@ -436,6 +440,7 @@ export function resolveCliBackendConfig( ownsNativeCompaction: fallbackPolicy.ownsNativeCompaction, prepareExecution: fallbackPolicy.prepareExecution, resolveExecutionArgs: fallbackPolicy.resolveExecutionArgs, + parseJsonlEvent: fallbackPolicy.parseJsonlEvent, toolAvailabilityEnforcement: fallbackPolicy.toolAvailabilityEnforcement, nativeToolMode: fallbackPolicy.nativeToolMode, sideQuestionToolMode: fallbackPolicy.sideQuestionToolMode, diff --git a/src/agents/cli-output.test.ts b/src/agents/cli-output.test.ts index 68cfef49f9e9..319de1326b5f 100644 --- a/src/agents/cli-output.test.ts +++ b/src/agents/cli-output.test.ts @@ -2070,6 +2070,280 @@ describe("createCliJsonlStreamingParser", () => { }); }); + it("streams plugin-owned JSONL events through normalized core projections", () => { + const assistantDeltas: Array<{ text: string; delta: string; sessionId?: string }> = []; + const thinkingDeltas: Array<{ text: string; delta: string }> = []; + const displayStarts: CliToolUseStartDelta[] = []; + const displayResults: CliToolResultDelta[] = []; + const parsedStarts: CliToolUseStartDelta[] = []; + const sessionIds: string[] = []; + const usageEvents: Array<{ usage: unknown; isTerminal: boolean }> = []; + const parser = createCliJsonlStreamingParser({ + backend: { command: "acme", output: "jsonl" }, + providerId: "acme-cli", + parseJsonlEvent: (line) => { + const event = JSON.parse(line) as { + type: string; + text?: string; + session?: string; + id?: string; + name?: string; + result?: unknown; + }; + if (event.type === "session") { + return { kind: "sessionId", sessionId: event.session ?? "" }; + } + if (event.type === "thinking") { + return { kind: "thinking", text: event.text ?? "" }; + } + if (event.type === "text") { + return { kind: "text", text: event.text ?? "" }; + } + if (event.type === "tool-start") { + return { + kind: "toolStart", + toolCallId: event.id ?? "", + name: event.name ?? "", + args: { query: "weather" }, + }; + } + if (event.type === "tool-result") { + return { + kind: "toolResult", + toolCallId: event.id ?? "", + name: event.name, + result: event.result, + }; + } + return { + kind: "result", + text: event.text, + sessionId: event.session, + usage: { input: 3, output: 2, total: 5 }, + }; + }, + onAssistantDelta: (delta) => assistantDeltas.push(delta), + onThinkingDelta: (delta) => thinkingDeltas.push(delta), + onToolUseStart: (delta) => parsedStarts.push(delta), + onDisplayToolUseStart: (delta) => displayStarts.push(delta), + onDisplayToolResult: (delta) => displayResults.push(delta), + onSessionId: (sessionId) => sessionIds.push(sessionId), + onUsage: (usage, isTerminal) => usageEvents.push({ usage, isTerminal }), + }); + + parser.push( + [ + JSON.stringify({ type: "session", session: "custom-session" }), + JSON.stringify({ type: "thinking", text: "Checking " }), + JSON.stringify({ type: "thinking", text: "facts." }), + JSON.stringify({ type: "text", text: "Hello " }), + JSON.stringify({ type: "text", text: "world" }), + JSON.stringify({ type: "tool-start", id: "call-1", name: "search" }), + JSON.stringify({ + type: "tool-result", + id: "call-1", + name: "search", + result: "sunny", + }), + JSON.stringify({ type: "result", text: "Hello world", session: "custom-successor" }), + "", + ].join("\n"), + ); + parser.finish(); + + expect(assistantDeltas).toEqual([ + { text: "Hello ", delta: "Hello ", sessionId: "custom-session", usage: undefined }, + { text: "Hello world", delta: "world", sessionId: "custom-session", usage: undefined }, + ]); + expect(thinkingDeltas).toEqual([ + { text: "Checking ", delta: "Checking ", isReasoningSnapshot: true }, + { text: "Checking facts.", delta: "facts.", isReasoningSnapshot: true }, + ]); + expect(displayStarts).toEqual([ + { + toolCallId: "call-1", + name: "search", + kind: "tool_use", + args: { query: "weather" }, + }, + ]); + expect(displayResults).toEqual([ + { toolCallId: "call-1", name: "search", isError: false, result: "sunny" }, + ]); + expect(parsedStarts).toEqual([]); + expect(sessionIds).toEqual(["custom-session", "custom-successor"]); + expect(usageEvents).toEqual([{ usage: { input: 3, output: 2, total: 5 }, isTerminal: true }]); + expect(parser.getOutput()).toEqual({ + text: "Hello world", + sessionId: "custom-successor", + usage: { input: 3, output: 2, total: 5 }, + }); + }); + + it("turns plugin-owned JSONL parser exceptions into bounded provider errors", () => { + let calls = 0; + const parser = createCliJsonlStreamingParser({ + backend: { command: "acme", output: "jsonl" }, + providerId: "acme-cli", + parseJsonlEvent: (line) => { + calls += 1; + if (line.includes("result")) { + return { kind: "result", text: "must not replace the parser error" }; + } + throw new Error("invalid custom event"); + }, + onAssistantDelta: () => {}, + }); + + parser.push('{"type":"broken"}\n{"type":"result"}\n'); + parser.finish(); + + expect(calls).toBe(1); + expect(parser.getOutput()).toEqual({ + text: "", + sessionId: undefined, + usage: undefined, + errorText: "CLI backend acme-cli JSONL parser failed: invalid custom event", + }); + }); + + it("keeps plugin-owned terminal errors ahead of later result summaries", () => { + const usageEvents: Array<{ usage: unknown; isTerminal: boolean }> = []; + const parser = createCliJsonlStreamingParser({ + backend: { command: "acme", output: "jsonl" }, + providerId: "acme-cli", + parseJsonlEvent: (line) => + line === "failed" + ? { kind: "result", errorText: "provider failed" } + : { + kind: "result", + text: "must not replace the provider error", + sessionId: "late-successor", + usage: { input: 2, output: 1, total: 3 }, + }, + onAssistantDelta: () => {}, + onUsage: (usage, isTerminal) => usageEvents.push({ usage, isTerminal }), + }); + + parser.push("failed\nsummary\n"); + parser.finish(); + + expect(parser.getOutput()).toEqual({ + text: "", + sessionId: "late-successor", + usage: { input: 2, output: 1, total: 3 }, + errorText: "provider failed", + }); + expect(usageEvents).toEqual([{ usage: { input: 2, output: 1, total: 3 }, isTerminal: true }]); + }); + + it("preserves plugin-owned session ids emitted after terminal errors", () => { + const sessionIds: string[] = []; + const parser = createCliJsonlStreamingParser({ + backend: { command: "acme", output: "jsonl" }, + providerId: "acme-cli", + parseJsonlEvent: () => [ + { kind: "result", errorText: "provider failed" }, + { kind: "sessionId", sessionId: "late-successor" }, + ], + onAssistantDelta: () => {}, + onSessionId: (sessionId) => sessionIds.push(sessionId), + }); + + parser.push("terminal\n"); + parser.finish(); + + expect(sessionIds).toEqual(["late-successor"]); + expect(parser.getOutput()).toEqual({ + text: "", + sessionId: "late-successor", + usage: undefined, + errorText: "provider failed", + }); + }); + + it("preserves streamed plugin text when the terminal result text is empty", () => { + const parser = createCliJsonlStreamingParser({ + backend: { command: "acme", output: "jsonl" }, + providerId: "acme-cli", + parseJsonlEvent: (line) => + line === "delta" + ? { kind: "text", text: "streamed answer" } + : { kind: "result", text: " " }, + onAssistantDelta: () => {}, + }); + + parser.push("delta\nresult\n"); + parser.finish(); + + expect(parser.getOutput()).toEqual({ + text: "streamed answer", + sessionId: undefined, + usage: undefined, + }); + }); + + it("preserves earlier plugin result text when a later result only adds metadata", () => { + const parser = createCliJsonlStreamingParser({ + backend: { command: "acme", output: "jsonl" }, + providerId: "acme-cli", + parseJsonlEvent: (line) => + line === "result" + ? { kind: "result", text: "completed answer" } + : { + kind: "result", + sessionId: "summary-session", + usage: { input: 5, output: 3, total: 8 }, + }, + onAssistantDelta: () => {}, + }); + + parser.push("result\nsummary\n"); + parser.finish(); + + expect(parser.getOutput()).toEqual({ + text: "completed answer", + sessionId: "summary-session", + usage: { input: 5, output: 3, total: 8 }, + }); + }); + + it("retains built-in fallback text after a plugin handles other lines", () => { + const parser = createCliJsonlStreamingParser({ + backend: { command: "acme", output: "jsonl" }, + providerId: "acme-cli", + parseJsonlEvent: (line) => { + if (line === "session") { + return { kind: "sessionId", sessionId: "custom-session" }; + } + if (line === "prefix") { + return { kind: "text", text: "streamed prefix" }; + } + return null; + }, + onAssistantDelta: () => {}, + }); + + parser.push( + [ + "session", + "prefix", + JSON.stringify({ + type: "item.completed", + item: { type: "agent_message", text: "delegated answer" }, + }), + "", + ].join("\n"), + ); + parser.finish(); + + expect(parser.getOutput()).toEqual({ + text: "delegated answer", + sessionId: "custom-session", + usage: undefined, + }); + }); + it("streams detailed Gemini error events over generic result errors", () => { const parser = createCliJsonlStreamingParser({ backend: { diff --git a/src/agents/cli-output.ts b/src/agents/cli-output.ts index 0bd33e643324..ef61f224adc0 100644 --- a/src/agents/cli-output.ts +++ b/src/agents/cli-output.ts @@ -7,7 +7,12 @@ import { normalizeLowercaseStringOrEmpty } from "@openclaw/normalization-core/st import { normalizeStringEntries } from "@openclaw/normalization-core/string-normalization"; import { truncateUtf16Safe } from "@openclaw/normalization-core/utf16-slice"; import type { AgentPlanStep } from "../channels/streaming.js"; -import type { CliBackendConfig } from "../plugins/cli-backend.types.js"; +import { formatErrorMessage } from "../infra/errors.js"; +import type { + CliBackendConfig, + CliBackendParseJsonlEvent, + CliBackendParsedJsonlEvent, +} from "../plugins/cli-backend.types.js"; import { extractBalancedJsonFragments } from "../shared/balanced-json.js"; import { isRecord } from "../utils.js"; import type { @@ -1202,12 +1207,15 @@ function readGeminiCliStreamJsonError(parsed: Record): string | export function createCliJsonlStreamingParser(params: { backend: CliBackendConfig; providerId: string; + parseJsonlEvent?: CliBackendParseJsonlEvent; onAssistantDelta: (delta: CliStreamingDelta) => void; onThinkingDelta?: (delta: CliThinkingDelta) => void; onThinkingProgress?: (progress: CliThinkingProgress) => void; onPlanUpdate?: (update: CliPlanUpdate) => void; onToolUseStart?: (delta: CliToolUseStartDelta) => void; onToolResult?: (delta: CliToolResultDelta) => void; + onDisplayToolUseStart?: (delta: CliToolUseStartDelta) => void; + onDisplayToolResult?: (delta: CliToolResultDelta) => void; onCommentaryText?: (text: string) => void; onSessionId?: (sessionId: string) => void; onAssistantMessage?: (message: unknown) => void; @@ -1215,6 +1223,7 @@ export function createCliJsonlStreamingParser(params: { }) { let lineBuffer = ""; let assistantText = ""; + let customThinkingText = ""; let pendingClaudeText = ""; let sessionId: string | undefined; let resumeCheckpointId: string | undefined; @@ -1225,6 +1234,7 @@ export function createCliJsonlStreamingParser(params: { let rawChars = 0; let rawLines = 0; const texts: string[] = []; + let sawCustomJsonlEvent = false; const toolTracker = createToolUseTracker(); const outputLimits = resolveCliStreamJsonOutputLimits(params.backend); // Classification is keyed on consumer presence so reclassified pre-tool text @@ -1259,6 +1269,128 @@ export function createCliJsonlStreamingParser(params: { } }; + const updateSessionId = (nextSessionId: string | undefined) => { + const normalized = nextSessionId?.trim(); + if (!normalized || normalized === sessionId) { + return; + } + sessionId = normalized; + params.onSessionId?.(normalized); + }; + + const handleCustomJsonlEvent = (event: CliBackendParsedJsonlEvent) => { + if (output?.errorText && event.kind !== "sessionId" && event.kind !== "result") { + return; + } + sawCustomJsonlEvent = true; + if (event.kind === "sessionId") { + updateSessionId(event.sessionId); + if (output) { + output = { ...output, sessionId }; + } + return; + } + if (event.kind === "text") { + if (!event.text) { + return; + } + assistantText = `${assistantText}${event.text}`; + params.onAssistantDelta({ + text: assistantText, + delta: event.text, + sessionId, + usage, + }); + return; + } + if (event.kind === "thinking") { + if (!event.text || !params.onThinkingDelta) { + return; + } + customThinkingText = `${customThinkingText}${event.text}`; + params.onThinkingDelta({ + text: customThinkingText, + delta: event.text, + isReasoningSnapshot: true, + }); + return; + } + if (event.kind === "toolStart") { + emitToolStartOnce( + toolTracker, + event.toolCallId, + event.name, + "tool_use", + event.args ?? {}, + params.onDisplayToolUseStart ?? params.onToolUseStart, + ); + return; + } + if (event.kind === "toolResult") { + if (event.name) { + toolTracker.nameById.set(event.toolCallId, event.name); + } + emitToolResultOnce( + toolTracker, + event.toolCallId, + event.isError === true, + event.result, + params.onDisplayToolResult ?? params.onToolResult, + ); + return; + } + updateSessionId(event.sessionId); + if (event.usage) { + usage = event.usage; + params.onUsage?.(event.usage, true); + } + const existingErrorText = output?.errorText; + const eventText = event.text?.trim() ?? ""; + const existingText = output?.text.trim() ?? ""; + const streamedText = assistantText.trim(); + const delegatedText = texts.join("\n").trim(); + const resultText = existingErrorText + ? existingText || delegatedText || streamedText + : eventText || existingText || delegatedText || streamedText; + const errorText = existingErrorText || event.errorText; + output = { + ...output, + text: resultText, + sessionId, + usage, + ...(errorText ? { errorText } : {}), + }; + }; + + const handleCustomJsonlLine = (line: string): boolean => { + if (parseErrorText) { + return true; + } + if (!params.parseJsonlEvent) { + return false; + } + let parsed: ReturnType; + try { + parsed = params.parseJsonlEvent(line, { + backendId: params.providerId, + backend: params.backend, + }); + } catch (error) { + parseErrorText = truncateUtf16Safe( + `CLI backend ${params.providerId} JSONL parser failed: ${formatErrorMessage(error)}`, + 500, + ); + return true; + } + if (parsed == null) { + return false; + } + for (const event of Array.isArray(parsed) ? parsed : [parsed]) { + handleCustomJsonlEvent(event); + } + return true; + }; + const handleParsedRecord = (parsed: Record) => { if (parseErrorText) { return; @@ -1484,6 +1616,9 @@ export function createCliJsonlStreamingParser(params: { lineBuffer = ""; return; } + if (handleCustomJsonlLine(line)) { + continue; + } for (const parsed of parseJsonRecordCandidates(line)) { handleParsedRecord(parsed); } @@ -1496,6 +1631,9 @@ export function createCliJsonlStreamingParser(params: { if (!tail) { return; } + if (handleCustomJsonlLine(tail)) { + return; + } for (const parsed of parseJsonRecordCandidates(tail)) { handleParsedRecord(parsed); } @@ -1545,6 +1683,9 @@ export function createCliJsonlStreamingParser(params: { if (output) { return output; } + if (sawCustomJsonlEvent) { + return { text: texts.join("\n").trim() || assistantText.trim(), sessionId, usage }; + } if (isStreamJsonDialect(params) && assistantText.trim()) { return { text: assistantText.trim(), diff --git a/src/agents/cli-runner/execute-events.ts b/src/agents/cli-runner/execute-events.ts index 325b3f16868d..5c4a50f40f07 100644 --- a/src/agents/cli-runner/execute-events.ts +++ b/src/agents/cli-runner/execute-events.ts @@ -126,6 +126,50 @@ export function createCliEventHandlers(params: { }); } }; + // Plugin-parsed events describe native work already performed by the backend. + // Render and summarize them without host-tool correlation or delivery evidence. + const emitCliDisplayToolUseStart = (event: CliToolUseStartDelta) => { + observedCliActivity = true; + recordToolStart(event); + if (!signaledToolExecutionStarted) { + signaledToolExecutionStarted = true; + runParams.onExecutionPhase?.({ + phase: "tool_execution_started", + provider: runParams.provider, + model: context.modelId, + backend: context.backendResolved.id, + }); + } + if (emitLiveEvents) { + emitAgentEvent({ + runId: runParams.runId, + stream: "tool", + data: { + phase: "start", + name: event.name, + toolCallId: event.toolCallId, + args: sanitizeToolArgs(event.args), + }, + }); + } + }; + const emitCliDisplayToolResult = (event: CliToolResult) => { + observedCliActivity = true; + recordToolResult(event); + if (emitLiveEvents) { + emitAgentEvent({ + runId: runParams.runId, + stream: "tool", + data: { + phase: "result", + name: event.name, + toolCallId: event.toolCallId, + isError: event.isError, + result: sanitizeToolResult(event.result), + }, + }); + } + }; const emitParsedToolUseStart = (event: CliToolUseStartDelta) => { const startedAt = Date.now(); activeParsedTools.set(event.toolCallId, { @@ -332,6 +376,8 @@ export function createCliEventHandlers(params: { emitLiveEvents, emitCliToolUseStart, emitCliToolResult, + emitCliDisplayToolUseStart, + emitCliDisplayToolResult, emitParsedToolUseStart, emitParsedToolResult, finalizeParsedTools, diff --git a/src/agents/cli-runner/execute-process.ts b/src/agents/cli-runner/execute-process.ts index 3c9aa32b70d4..d7ed67e5e6b6 100644 --- a/src/agents/cli-runner/execute-process.ts +++ b/src/agents/cli-runner/execute-process.ts @@ -167,12 +167,15 @@ export async function executeCliProcess(params: { ? createCliJsonlStreamingParser({ backend: params.backend, providerId: context.backendResolved.id, + parseJsonlEvent: context.backendResolved.parseJsonlEvent, onAssistantDelta: params.events.emitCliAssistantDelta, onThinkingDelta: params.events.emitCliThinkingDelta, onThinkingProgress: params.events.emitCliThinkingProgress, onPlanUpdate: params.events.emitCliPlanUpdate, onToolUseStart: params.events.emitParsedToolUseStart, onToolResult: params.events.emitParsedToolResult, + onDisplayToolUseStart: params.events.emitCliDisplayToolUseStart, + onDisplayToolResult: params.events.emitCliDisplayToolResult, onCommentaryText: params.events.emitLiveEvents && runParams.emitCommentaryText ? params.events.emitCliCommentaryText diff --git a/src/agents/cli-runner/execute.supervisor-capture.test.ts b/src/agents/cli-runner/execute.supervisor-capture.test.ts index e8c9558845f5..cfe1d0c035b2 100644 --- a/src/agents/cli-runner/execute.supervisor-capture.test.ts +++ b/src/agents/cli-runner/execute.supervisor-capture.test.ts @@ -15,6 +15,7 @@ import { resetDiagnosticEventsForTest, type TrustedToolExecutionEvent, } from "../../infra/diagnostic-events.js"; +import type { CliBackendParseJsonlEvent } from "../../plugins/cli-backend.types.js"; import type { getProcessSupervisor } from "../../process/supervisor/index.js"; import { findCliMaxTurnsError } from "../failover-error.js"; import { getCliMessagingDeliveryEvidence } from "./delivery-evidence.js"; @@ -92,6 +93,7 @@ function buildPreparedCliRunContext(params: { provider?: string; runId?: string; beforeExecution?: () => Promise; + parseJsonlEvent?: CliBackendParseJsonlEvent; }): PreparedCliRunContext { const provider = params.provider ?? "codex-cli"; const backend = { @@ -121,6 +123,7 @@ function buildPreparedCliRunContext(params: { id: provider, config: backend, bundleMcp: false, + parseJsonlEvent: params.parseJsonlEvent, }, preparedBackend: { backend, @@ -634,6 +637,161 @@ describe("executePreparedCliRun supervisor output capture", () => { expect(restoreCliSessionFork).toHaveBeenCalledTimes(1); }); + it("composes plugin-owned JSONL parsing into the production executor", async () => { + const agentEvents: Array<{ stream: string; phase?: string; text?: string }> = []; + const trustedEvents: TrustedToolExecutionEvent[] = []; + const stopAgentEvents = onAgentEvent((event) => { + agentEvents.push({ + stream: event.stream, + phase: typeof event.data.phase === "string" ? event.data.phase : undefined, + text: typeof event.data.text === "string" ? event.data.text : undefined, + }); + }); + const stopTrustedEvents = onTrustedToolExecutionEvent((event) => trustedEvents.push(event)); + const parseJsonlEvent: CliBackendParseJsonlEvent = (line) => { + const event = JSON.parse(line) as { + type: string; + text?: string; + session?: string; + id?: string; + name?: string; + result?: unknown; + }; + switch (event.type) { + case "session": + return { kind: "sessionId", sessionId: event.session ?? "" }; + case "thinking": + return { kind: "thinking", text: event.text ?? "" }; + case "text": + return { kind: "text", text: event.text ?? "" }; + case "tool-start": + return { + kind: "toolStart", + toolCallId: event.id ?? "", + name: event.name ?? "", + args: { query: "weather" }, + }; + case "tool-result": + return { + kind: "toolResult", + toolCallId: event.id ?? "", + name: event.name, + result: event.result, + }; + default: + return { + kind: "result", + text: event.text, + sessionId: event.session, + usage: { input: 4, output: 2, total: 6 }, + }; + } + }; + const chunks = [ + `${JSON.stringify({ type: "session", session: "custom-session" })}\n`, + `${JSON.stringify({ type: "thinking", text: "Checking facts." })}\n`, + `${JSON.stringify({ type: "text", text: "Hello world" })}\n`, + `${JSON.stringify({ type: "tool-start", id: "call-1", name: "search" })}\n`, + `${JSON.stringify({ + type: "tool-result", + id: "call-1", + name: "search", + result: "sunny", + })}\n`, + `${JSON.stringify({ type: "result", text: "Hello world", session: "custom-successor" })}\n`, + ]; + supervisorSpawnMock.mockImplementationOnce(async (...args: unknown[]) => { + const input = args[0] as SupervisorSpawnInput; + for (const chunk of chunks) { + input.onStdout?.(chunk); + } + return createManagedRun({ + reason: "exit", + exitCode: 0, + exitSignal: null, + durationMs: 50, + stdout: "", + stderr: "", + timedOut: false, + noOutputTimedOut: false, + }); + }); + + try { + const context = buildPreparedCliRunContext({ + output: "jsonl", + provider: "acme-cli", + parseJsonlEvent, + }); + const result = await executePreparedCliRun(context); + + expect(result).toMatchObject({ + text: "Hello world", + sessionId: "custom-successor", + usage: { input: 4, output: 2, total: 6 }, + toolSummary: { calls: 1, tools: ["search"], failures: 0 }, + }); + expect(getCliMessagingDeliveryEvidence(context.params.runId)).toBeUndefined(); + expect(agentEvents).toEqual( + expect.arrayContaining([ + expect.objectContaining({ stream: "thinking", text: "Checking facts." }), + expect.objectContaining({ stream: "assistant", text: "Hello world" }), + expect.objectContaining({ stream: "tool", phase: "start" }), + expect.objectContaining({ stream: "tool", phase: "result" }), + ]), + ); + expect(trustedEvents).toEqual([]); + } finally { + stopAgentEvents(); + stopTrustedEvents(); + } + }); + + it("persists plugin-owned successor session ids for forked resumes", async () => { + const parseJsonlEvent: CliBackendParseJsonlEvent = (line) => { + const event = JSON.parse(line) as { type: string; session?: string; text?: string }; + return event.type === "session" + ? { kind: "sessionId", sessionId: event.session ?? "" } + : { kind: "result", text: event.text }; + }; + const chunks = [ + `${JSON.stringify({ type: "session", session: "fork-successor" })}\n`, + `${JSON.stringify({ type: "result", text: "done" })}\n`, + ]; + supervisorSpawnMock.mockImplementationOnce(async (...args: unknown[]) => { + const input = args[0] as SupervisorSpawnInput; + for (const chunk of chunks) { + input.onStdout?.(chunk); + } + return createManagedRun({ + reason: "exit", + exitCode: 0, + exitSignal: null, + durationMs: 50, + stdout: "", + stderr: "", + timedOut: false, + noOutputTimedOut: false, + }); + }); + const persistCliSessionForkSuccessor = vi.fn().mockResolvedValue(undefined); + const context = buildPreparedCliRunContext({ + output: "jsonl", + provider: "acme-cli", + parseJsonlEvent, + }); + context.preparedBackend.backend.resumeArgs = ["--resume", "{sessionId}"]; + context.preparedBackend.backend.forkArg = "--fork-session"; + context.params.forkCliSessionOnResume = true; + context.params.claimCliSessionFork = vi.fn().mockResolvedValue(true); + context.params.persistCliSessionForkSuccessor = persistCliSessionForkSuccessor; + + const result = await executePreparedCliRun(context, "fork-source"); + + expect(result).toMatchObject({ text: "done", sessionId: "fork-successor" }); + expect(persistCliSessionForkSuccessor).toHaveBeenCalledWith("fork-successor"); + }); + it("still streams every JSONL stdout chunk with supervisor capture disabled", async () => { // Streaming events are emitted from live chunks, not from the final captured // stdout string, so users still see deltas when captureOutput is false. diff --git a/src/gateway/server.sessions.create.test.ts b/src/gateway/server.sessions.create.test.ts index 9721070caf06..2cc1e02386d9 100644 --- a/src/gateway/server.sessions.create.test.ts +++ b/src/gateway/server.sessions.create.test.ts @@ -7,6 +7,7 @@ import path from "node:path"; import { promisify } from "node:util"; import { afterEach, expect, test, vi } from "vitest"; import { useAutoCleanupTempDirTracker } from "../../test/helpers/temp-dir.js"; +import { findGitCheckoutRoot } from "../agents/worktrees/git.js"; import { findLiveRegistryWorktreeByOwner, listRegistryWorktrees, @@ -61,6 +62,21 @@ const { createSessionStoreDir, createSelectedGlobalSessionStore, openClient } = const execFileAsync = promisify(execFile); const tempDirs = useAutoCleanupTempDirTracker(afterEach); +async function makeNonGitTempDir(prefix: string): Promise { + let root = await fs.realpath(os.tmpdir()); + for (;;) { + const checkoutRoot = findGitCheckoutRoot(root); + if (!checkoutRoot) { + return tempDirs.make(prefix, root); + } + const parent = path.dirname(checkoutRoot); + if (parent === checkoutRoot) { + throw new Error("could not find a temp root outside a git checkout"); + } + root = parent; + } +} + test("sessions.create and sessions.delete preserve every concurrent session lifecycle", async () => { const { storePath } = await createSessionStoreDir(); const sessionCount = 24; @@ -1147,7 +1163,7 @@ test("sessions.create persists a Gateway cwd without a managed worktree", async }); test("sessions.create uses a non-git Gateway cwd directly but not as a worktree source", async () => { - const cwd = tempDirs.make("openclaw-session-direct-cwd-", await fs.realpath(os.tmpdir())); + const cwd = await makeNonGitTempDir("openclaw-session-direct-cwd-"); const client = { client: { connect: { scopes: ["operator.admin"] } } as never }; const direct = await directSessionReq("sessions.create", { cwd }, client); expect(direct.ok).toBe(true); @@ -1369,9 +1385,7 @@ test("sessions.create reset-in-place persists the returned worktree cwd", async }); test("sessions.create rejects worktrees for non-git agent workspaces", async () => { - const workspace = await fs.mkdtemp( - path.join(await fs.realpath(os.tmpdir()), "openclaw-session-plain-workspace-"), - ); + const workspace = await makeNonGitTempDir("openclaw-session-plain-workspace-"); testState.agentConfig = { workspace }; await createSessionStoreDir(); try { @@ -1388,7 +1402,6 @@ test("sessions.create rejects worktrees for non-git agent workspaces", async () }); } finally { testState.agentConfig = undefined; - await fs.rm(workspace, { recursive: true, force: true }); } }); diff --git a/src/plugin-sdk/cli-backend.ts b/src/plugin-sdk/cli-backend.ts index 9d075a40e65c..2bca94adb989 100644 --- a/src/plugin-sdk/cli-backend.ts +++ b/src/plugin-sdk/cli-backend.ts @@ -5,8 +5,12 @@ export type { CliBackendAuthEpochMode, CliBackendConfig, CliBackendExecutionMode, + CliBackendJsonlUsage, CliBackendNormalizeConfigContext, CliBackendNativeToolMode, + CliBackendParseJsonlEvent, + CliBackendParseJsonlEventContext, + CliBackendParsedJsonlEvent, CliBackendPlugin, CliBackendPreparedExecution, CliBackendPrepareExecutionContext, diff --git a/src/plugins/cli-backend.types.ts b/src/plugins/cli-backend.types.ts index 52ae42ae3c9f..0fb79e558185 100644 --- a/src/plugins/cli-backend.types.ts +++ b/src/plugins/cli-backend.types.ts @@ -175,6 +175,49 @@ export type CliBackendResolveExecutionArgs = ( ctx: CliBackendResolveExecutionArgsContext, ) => readonly string[] | null | undefined; +export type CliBackendJsonlUsage = { + input?: number; + output?: number; + cacheRead?: number; + cacheWrite?: number; + total?: number; +}; + +export type CliBackendParsedJsonlEvent = + | { kind: "text"; text: string } + | { kind: "thinking"; text: string } + | { + kind: "toolStart"; + toolCallId: string; + name: string; + args?: Record; + } + | { + kind: "toolResult"; + toolCallId: string; + name?: string; + isError?: boolean; + result?: unknown; + } + | { + kind: "result"; + text?: string; + sessionId?: string; + usage?: CliBackendJsonlUsage; + errorText?: string; + } + | { kind: "sessionId"; sessionId: string }; + +export type CliBackendParseJsonlEventContext = { + backendId: string; + backend: Readonly; +}; + +export type CliBackendParseJsonlEvent = ( + line: string, + ctx: CliBackendParseJsonlEventContext, +) => CliBackendParsedJsonlEvent | readonly CliBackendParsedJsonlEvent[] | null | undefined; + export type CliBackendAuthEpochMode = "combined" | "profile-only"; export type CliBackendNativeToolMode = "none" | "always-on" | "selectable"; @@ -337,6 +380,13 @@ export type CliBackendPlugin = { resolveExecutionArgs?: CliBackendResolveExecutionArgs; /** How this backend enforces an exact per-run `toolAvailability` contract. */ toolAvailabilityEnforcement?: CliBackendToolAvailabilityEnforcement; + /** + * Backend-owned JSONL line parser for provider-specific stream formats. + * + * Tool events report execution already performed by the backend. OpenClaw + * renders them but does not treat them as host tool execution or delivery evidence. + */ + parseJsonlEvent?: CliBackendParseJsonlEvent; /** * Whether this CLI backend can expose native tools outside OpenClaw's tool * catalog. Exact restricted runs require `selectable` plus a declared diff --git a/src/plugins/types.ts b/src/plugins/types.ts index 2ec8662ef70d..a840251b7f58 100644 --- a/src/plugins/types.ts +++ b/src/plugins/types.ts @@ -10,8 +10,12 @@ export type { CliBackendAuthEpochMode, CliBackendConfig, CliBackendExecutionMode, + CliBackendJsonlUsage, CliBackendNormalizeConfigContext, CliBackendNativeToolMode, + CliBackendParseJsonlEvent, + CliBackendParseJsonlEventContext, + CliBackendParsedJsonlEvent, CliBackendPlugin, CliBackendPreparedExecution, CliBackendPrepareExecutionContext,