diff --git a/docs/gateway/opentelemetry.md b/docs/gateway/opentelemetry.md index a952e1566b12..6266fd8a59e9 100644 --- a/docs/gateway/opentelemetry.md +++ b/docs/gateway/opentelemetry.md @@ -238,6 +238,15 @@ OpenClaw's CLI boundary through - `paired-node-cli` - one-shot Claude Code execution delegated to a paired node. +Claude CLI diagnostics are instantiated only while the process diagnostic +dispatcher is enabled and an internal or trusted event listener is attached. +With no observability plugin or other listener active, Claude CLI turns skip +the synthetic trace hierarchy, content buffers, and diagnostic stream-byte +accounting. When content capture is enabled, prompt and system-prompt fields +are capped at 128 KiB each; assistant output is capped at 128 KiB across at +most 200 envelopes, with 16 KiB and one item reserved for a final visible +fallback response. A marker records truncation when the limit is reached. + OpenClaw gives Claude CLI turns the same ownership hierarchy used by other agent runtimes: `openclaw.harness.run` (`openclaw.harness.id = claude-cli`) contains `openclaw.run`, which contains the Claude `openclaw.model.call` diff --git a/src/agents/cli-runner.before-agent-reply-cron.test.ts b/src/agents/cli-runner.before-agent-reply-cron.test.ts index db3cbd093260..6c4e36f1a3ab 100644 --- a/src/agents/cli-runner.before-agent-reply-cron.test.ts +++ b/src/agents/cli-runner.before-agent-reply-cron.test.ts @@ -200,6 +200,21 @@ describe("runCliAgent before_agent_reply seam", () => { expect(result?.diagnosticTrace).toEqual(harnessStarted?.trace); }); + it("bypasses Claude CLI diagnostics when no event listener is active", async () => { + executePreparedCliRunMock.mockResolvedValue({ text: "real Claude reply" }); + + const result = await runCliAgent({ + ...baseRunParams, + provider: "claude-cli", + modelProvider: "anthropic", + model: "claude-opus-4-7", + runId: "claude-no-diagnostics-listener", + }); + + expect(result.diagnosticTrace).toBeUndefined(); + expect(executePreparedCliRunMock).toHaveBeenCalledOnce(); + }); + it("preserves the send phase when execution fails before successful cleanup", async () => { executePreparedCliRunMock.mockRejectedValueOnce(new Error("CLI process failed")); diff --git a/src/agents/cli-runner.ts b/src/agents/cli-runner.ts index 67dfc9979f77..40c8cfd61013 100644 --- a/src/agents/cli-runner.ts +++ b/src/agents/cli-runner.ts @@ -10,6 +10,8 @@ import { captureAgentRunLifecycleGeneration, withAgentRunLifecycleGeneration, } from "../infra/agent-events.js"; +import { hasInternalDiagnosticEventListeners } from "../infra/diagnostic-event-listener-presence.js"; +import { areDiagnosticsEnabledForProcess } from "../infra/diagnostic-events.js"; import { formatErrorMessage } from "../infra/errors.js"; import { createSubsystemLogger } from "../logging/subsystem.js"; import { @@ -468,8 +470,12 @@ export function runCliAgent(paramsInput: RunCliAgentParams): Promise - isClaudeCliProvider(params.provider) + isClaudeCliProvider(params.provider) && + areDiagnosticsEnabledForProcess() && + hasInternalDiagnosticEventListeners() ? runClaudeCliAgentTurnWithDiagnostics(params, (diagnosticLifecycle) => runCliAgentInternal(params, diagnosticLifecycle), ) diff --git a/src/agents/cli-runner/execute.ts b/src/agents/cli-runner/execute.ts index a6d0daf1245e..3d87efc17058 100644 --- a/src/agents/cli-runner/execute.ts +++ b/src/agents/cli-runner/execute.ts @@ -1644,8 +1644,9 @@ export async function executePreparedCliRun( const stderrHash = crypto.createHash("sha256"); let stderrParseExceeded = false; const consumeStdout = (chunk: string) => { - claudeModelCallDiagnostics?.observeCliOutput(chunk, "stdout"); - stdoutBytes += Buffer.byteLength(chunk); + const chunkBytes = Buffer.byteLength(chunk); + claudeModelCallDiagnostics?.observeCliOutput(chunk, "stdout", chunkBytes); + stdoutBytes += chunkBytes; stdoutHash.update(chunk); stdoutTail = appendCliOutputTail(stdoutTail, chunk); if (!stdoutParseExceeded) { diff --git a/src/agents/cli-runner/model-call-diagnostics.test.ts b/src/agents/cli-runner/model-call-diagnostics.test.ts new file mode 100644 index 000000000000..6a0ddb65c50c --- /dev/null +++ b/src/agents/cli-runner/model-call-diagnostics.test.ts @@ -0,0 +1,355 @@ +// Verifies Claude CLI model diagnostics stay listener-gated and memory-bounded. +import { expectDefined } from "@openclaw/normalization-core"; +import { afterEach, describe, expect, it } from "vitest"; +import { + onTrustedInternalDiagnosticEvent, + resetDiagnosticEventsForTest, + waitForDiagnosticEventsDrained, + type DiagnosticEventPrivateData, +} from "../../infra/diagnostic-events.js"; +import type { CliOutput } from "../cli-output.js"; +import { createClaudeCliModelCallDiagnostics } from "./model-call-diagnostics.js"; +import type { PreparedCliRunContext } from "./types.js"; + +const CONTENT_LIMIT_BYTES = 128 * 1024; + +function createContext(): PreparedCliRunContext { + return { + backendResolved: { id: "claude-cli", modelProvider: "anthropic" }, + contextWindowInfo: undefined, + normalizedModel: "claude-test", + params: { + config: { + diagnostics: { + enabled: true, + otel: { + enabled: true, + traces: true, + captureContent: { + enabled: true, + inputMessages: true, + outputMessages: true, + systemPrompt: true, + }, + }, + }, + }, + runId: "claude-diagnostics-test", + sessionId: "session-test", + }, + } as PreparedCliRunContext; +} + +describe("Claude CLI model-call diagnostics", () => { + afterEach(() => { + resetDiagnosticEventsForTest(); + }); + + it("does not create diagnostics without an active event listener", () => { + expect( + createClaudeCliModelCallDiagnostics({ + context: createContext(), + prompt: "hello", + transport: "stdio", + }), + ).toBeUndefined(); + }); + + it("bounds large prompt and assistant content during a burst", async () => { + let completedPrivateData: DiagnosticEventPrivateData | undefined; + const stop = onTrustedInternalDiagnosticEvent((event, _metadata, privateData) => { + if (event.type === "model.call.completed") { + completedPrivateData = privateData; + } + }); + const largeText = "€".repeat(256 * 1024); + const diagnostics = expectDefined( + createClaudeCliModelCallDiagnostics({ + context: createContext(), + prompt: largeText, + systemPrompt: largeText, + transport: "stdio", + }), + "Claude CLI diagnostics", + ); + + diagnostics.emitStarted(); + for (let index = 0; index < 250; index += 1) { + diagnostics.observeAssistantMessage({ + content: [{ type: "text", text: `${index}:${largeText}` }], + }); + } + diagnostics.emitCompleted({ text: "done" } as CliOutput); + await waitForDiagnosticEventsDrained(); + stop(); + + const modelContent = expectDefined(completedPrivateData?.modelContent, "model content"); + const inputJson = JSON.stringify(modelContent.inputMessages); + const outputJson = JSON.stringify(modelContent.outputMessages); + expect(Buffer.byteLength(inputJson, "utf8")).toBeLessThanOrEqual(CONTENT_LIMIT_BYTES + 256); + expect(Buffer.byteLength(modelContent.systemPrompt ?? "", "utf8")).toBeLessThanOrEqual( + CONTENT_LIMIT_BYTES, + ); + expect(Buffer.byteLength(outputJson, "utf8")).toBeLessThanOrEqual(CONTENT_LIMIT_BYTES); + expect(outputJson).toContain("...(truncated)"); + }); + + it("bounds serialized escaped content across many envelopes", async () => { + let outputMessages: unknown; + const stop = onTrustedInternalDiagnosticEvent((event, _metadata, privateData) => { + if (event.type === "model.call.completed") { + outputMessages = privateData.modelContent?.outputMessages; + } + }); + const diagnostics = expectDefined( + createClaudeCliModelCallDiagnostics({ + context: createContext(), + prompt: "hello", + transport: "stdio", + }), + "Claude CLI diagnostics", + ); + + diagnostics.emitStarted(); + for (let index = 0; index < 199; index += 1) { + diagnostics.observeAssistantMessage({ content: `part-${index}:\u0000`.repeat(100) }); + } + diagnostics.emitCompleted({ text: "done" } as CliOutput); + await waitForDiagnosticEventsDrained(); + stop(); + + expect(Buffer.byteLength(JSON.stringify(outputMessages), "utf8")).toBeLessThanOrEqual( + CONTENT_LIMIT_BYTES, + ); + }); + + it("caps assistant envelope count and records truncation", async () => { + let outputMessages: unknown; + const stop = onTrustedInternalDiagnosticEvent((event, _metadata, privateData) => { + if (event.type === "model.call.completed") { + outputMessages = privateData.modelContent?.outputMessages; + } + }); + const diagnostics = expectDefined( + createClaudeCliModelCallDiagnostics({ + context: createContext(), + prompt: "hello", + transport: "stdio", + }), + "Claude CLI diagnostics", + ); + + diagnostics.emitStarted(); + for (let index = 0; index < 250; index += 1) { + diagnostics.observeAssistantMessage({ content: `message-${index}` }); + } + diagnostics.emitCompleted({ text: "done" } as CliOutput); + await waitForDiagnosticEventsDrained(); + stop(); + + expect(outputMessages).toHaveLength(200); + expect(JSON.stringify(outputMessages)).toContain("...(truncated)"); + }); + + it("records truncation when the item budget ends inside an envelope", async () => { + let outputMessages: unknown; + const stop = onTrustedInternalDiagnosticEvent((event, _metadata, privateData) => { + if (event.type === "model.call.completed") { + outputMessages = privateData.modelContent?.outputMessages; + } + }); + const diagnostics = expectDefined( + createClaudeCliModelCallDiagnostics({ + context: createContext(), + prompt: "hello", + transport: "stdio", + }), + "Claude CLI diagnostics", + ); + + diagnostics.emitStarted(); + for (let index = 0; index < 197; index += 1) { + diagnostics.observeAssistantMessage({ content: `message-${index}` }); + } + diagnostics.observeAssistantMessage({ + content: [ + { type: "text", text: "captured-a" }, + { type: "text", text: "captured-b" }, + { type: "text", text: "dropped-c" }, + ], + }); + diagnostics.emitCompleted({ text: "done" } as CliOutput); + await waitForDiagnosticEventsDrained(); + stop(); + + expect(outputMessages).toHaveLength(199); + expect(JSON.stringify(outputMessages)).toContain("captured-b"); + expect(JSON.stringify(outputMessages)).not.toContain("dropped-c"); + expect(JSON.stringify(outputMessages)).toContain("...(truncated)"); + }); + + it("keeps fallback response text when prior non-text envelopes fill the limit", async () => { + let outputMessages: unknown; + const stop = onTrustedInternalDiagnosticEvent((event, _metadata, privateData) => { + if (event.type === "model.call.completed") { + outputMessages = privateData.modelContent?.outputMessages; + } + }); + const diagnostics = expectDefined( + createClaudeCliModelCallDiagnostics({ + context: createContext(), + prompt: "hello", + transport: "stdio", + }), + "Claude CLI diagnostics", + ); + + diagnostics.emitStarted(); + for (let index = 0; index < 200; index += 1) { + diagnostics.observeAssistantMessage({ + content: [{ type: "thinking", thinking: `reason-${index}` }], + }); + } + diagnostics.emitCompleted({ text: "final visible answer" } as CliOutput); + await waitForDiagnosticEventsDrained(); + stop(); + + expect(outputMessages).toHaveLength(200); + expect(JSON.stringify(outputMessages)).toContain("final visible answer"); + expect(JSON.stringify(outputMessages)).toContain("...(truncated)"); + }); + + it("keeps fallback response text when non-text content fills the byte budget", async () => { + let outputMessages: unknown; + const stop = onTrustedInternalDiagnosticEvent((event, _metadata, privateData) => { + if (event.type === "model.call.completed") { + outputMessages = privateData.modelContent?.outputMessages; + } + }); + const diagnostics = expectDefined( + createClaudeCliModelCallDiagnostics({ + context: createContext(), + prompt: "hello", + transport: "stdio", + }), + "Claude CLI diagnostics", + ); + + diagnostics.emitStarted(); + diagnostics.observeAssistantMessage({ + content: [{ type: "thinking", thinking: "€".repeat(256 * 1024) }], + }); + diagnostics.emitCompleted({ text: "final visible answer" } as CliOutput); + await waitForDiagnosticEventsDrained(); + stop(); + + const outputJson = JSON.stringify(outputMessages); + expect(outputJson).toContain("final visible answer"); + expect(outputJson).toContain("...(truncated)"); + expect(Buffer.byteLength(outputJson, "utf8")).toBeLessThanOrEqual(CONTENT_LIMIT_BYTES + 512); + }); + + it("keeps the fallback reserve after empty text content", async () => { + let outputMessages: unknown; + const stop = onTrustedInternalDiagnosticEvent((event, _metadata, privateData) => { + if (event.type === "model.call.completed") { + outputMessages = privateData.modelContent?.outputMessages; + } + }); + const diagnostics = expectDefined( + createClaudeCliModelCallDiagnostics({ + context: createContext(), + prompt: "hello", + transport: "stdio", + }), + "Claude CLI diagnostics", + ); + + diagnostics.emitStarted(); + diagnostics.observeAssistantMessage({ content: "" }); + diagnostics.observeAssistantMessage({ + content: Array.from({ length: 200 }, (_, index) => ({ + type: "thinking", + thinking: `reason-${index}`, + })), + }); + diagnostics.emitCompleted({ text: "final visible answer" } as CliOutput); + await waitForDiagnosticEventsDrained(); + stop(); + + const outputJson = JSON.stringify(outputMessages); + expect(outputJson).toContain("final visible answer"); + expect(outputJson).toContain("...(truncated)"); + }); + + it("bounds text probing after the capture budget is exhausted", async () => { + let outputMessages: unknown; + const stop = onTrustedInternalDiagnosticEvent((event, _metadata, privateData) => { + if (event.type === "model.call.completed") { + outputMessages = privateData.modelContent?.outputMessages; + } + }); + const diagnostics = expectDefined( + createClaudeCliModelCallDiagnostics({ + context: createContext(), + prompt: "hello", + transport: "stdio", + }), + "Claude CLI diagnostics", + ); + + diagnostics.emitStarted(); + diagnostics.observeAssistantMessage({ + content: Array.from({ length: 198 }, (_, index) => ({ + type: "thinking", + thinking: `reason-${index}`, + })), + }); + diagnostics.observeAssistantMessage({ + content: [ + ...Array.from({ length: 10_000 }, () => ({ type: "thinking", thinking: "later" })), + { type: "text", text: "outside-probe-window" }, + ], + }); + diagnostics.emitCompleted({ text: "final visible answer" } as CliOutput); + await waitForDiagnosticEventsDrained(); + stop(); + + const outputJson = JSON.stringify(outputMessages); + expect(outputJson).toContain("final visible answer"); + expect(outputJson).not.toContain("outside-probe-window"); + }); + + it("counts the truncation marker within the output item limit", async () => { + let outputMessages: unknown; + const stop = onTrustedInternalDiagnosticEvent((event, _metadata, privateData) => { + if (event.type === "model.call.completed") { + outputMessages = privateData.modelContent?.outputMessages; + } + }); + const diagnostics = expectDefined( + createClaudeCliModelCallDiagnostics({ + context: createContext(), + prompt: "hello", + transport: "stdio", + }), + "Claude CLI diagnostics", + ); + + diagnostics.emitStarted(); + diagnostics.observeAssistantMessage({ + content: Array.from({ length: 201 }, (_, index) => ({ + type: "text", + text: `part-${index}`, + })), + }); + diagnostics.emitCompleted({ text: "done" } as CliOutput); + await waitForDiagnosticEventsDrained(); + stop(); + + const messages = outputMessages as Array<{ content?: unknown[] }>; + const itemCount = messages.reduce((sum, message) => sum + (message.content?.length ?? 0), 0); + expect(itemCount).toBe(200); + expect(JSON.stringify(messages)).toContain("...(truncated)"); + }); +}); diff --git a/src/agents/cli-runner/model-call-diagnostics.ts b/src/agents/cli-runner/model-call-diagnostics.ts index 49f3485c2026..5c668c4156f0 100644 --- a/src/agents/cli-runner/model-call-diagnostics.ts +++ b/src/agents/cli-runner/model-call-diagnostics.ts @@ -1,12 +1,15 @@ /** Trusted turn-level model-call diagnostics for the Claude Code CLI runtime. */ import crypto from "node:crypto"; import { isRecord } from "@openclaw/normalization-core/record-coerce"; +import { truncateUtf16Safe } from "@openclaw/normalization-core/utf16-slice"; import { diagnosticErrorCategory, diagnosticErrorFailureKind, diagnosticErrorMessage, } from "../../infra/diagnostic-error-metadata.js"; +import { hasInternalDiagnosticEventListeners } from "../../infra/diagnostic-event-listener-presence.js"; import { + areDiagnosticsEnabledForProcess, emitTrustedDiagnosticEventWithPrivateData, type DiagnosticEventPrivateData, type DiagnosticModelCallContent, @@ -29,15 +32,114 @@ type ModelCallFailureKind = Extract< { type: "model.call.error" } >["failureKind"]; -function assistantContentBlock(block: unknown): Record | undefined { +const MAX_CAPTURED_CONTENT_BYTES = 128 * 1024; +const FALLBACK_RESPONSE_RESERVE_BYTES = 16 * 1024; +const MAX_CAPTURED_OUTPUT_MESSAGES = 200; +const MAX_CAPTURED_OUTPUT_BLOCKS = 200; +const TRUNCATED_CONTENT_SUFFIX = "...(truncated)"; +// One maximal tool-call block per envelope plus stopReason is the largest +// structure possible under the shared 200-envelope/item caps. +const MAX_CAPTURED_OUTPUT_STRUCTURE_BYTES = Buffer.byteLength( + JSON.stringify( + Array.from({ length: MAX_CAPTURED_OUTPUT_MESSAGES }, () => ({ + role: "assistant", + content: [{ type: "tool_call", name: "", id: "" }], + stopReason: "", + })), + ), + "utf8", +); + +type DiagnosticContentBudget = { + remainingBytes: number; + remainingItems: number; + fallbackReserveBytes: number; + fallbackReserveItems: number; + truncated: boolean; +}; + +function serializedStringContentBytes(value: string): number { + return Buffer.byteLength(JSON.stringify(value), "utf8") - 2; +} + +const TRUNCATED_CONTENT_SUFFIX_BYTES = serializedStringContentBytes(TRUNCATED_CONTENT_SUFFIX); + +function truncateSerializedStringSafe(value: string, maxBytes: number): string { + if (maxBytes <= 0) { + return ""; + } + if (serializedStringContentBytes(value) <= maxBytes) { + return value; + } + let low = 0; + let high = Math.min(value.length, maxBytes); + let captured = ""; + while (low <= high) { + const middle = Math.floor((low + high) / 2); + const candidate = truncateUtf16Safe(value, middle); + if (serializedStringContentBytes(candidate) <= maxBytes) { + captured = candidate; + low = middle + 1; + } else { + high = middle - 1; + } + } + return captured; +} + +function releaseFallbackReserve(budget: DiagnosticContentBudget): void { + budget.remainingBytes += budget.fallbackReserveBytes; + budget.remainingItems += budget.fallbackReserveItems; + budget.fallbackReserveBytes = 0; + budget.fallbackReserveItems = 0; +} + +function captureTextWithinBudget( + value: string, + budget: DiagnosticContentBudget, +): string | undefined { + if (budget.remainingBytes <= 0) { + budget.truncated = true; + return undefined; + } + const valueBytes = serializedStringContentBytes(value); + if (valueBytes <= budget.remainingBytes) { + budget.remainingBytes -= valueBytes; + return value; + } + const suffix = truncateSerializedStringSafe(TRUNCATED_CONTENT_SUFFIX, budget.remainingBytes); + const prefixBudget = Math.max(0, budget.remainingBytes - serializedStringContentBytes(suffix)); + const captured = `${truncateSerializedStringSafe(value, prefixBudget)}${suffix}`; + budget.remainingBytes -= serializedStringContentBytes(captured); + budget.truncated = true; + return captured; +} + +function captureBoundedText(value: string): string { + const budget = { + remainingBytes: MAX_CAPTURED_CONTENT_BYTES, + remainingItems: 1, + fallbackReserveBytes: 0, + fallbackReserveItems: 0, + truncated: false, + }; + return captureTextWithinBudget(value, budget) ?? ""; +} + +function assistantContentBlock( + block: unknown, + budget: DiagnosticContentBudget, +): Record | undefined { if (!isRecord(block)) { return undefined; } - if (block.type === "text" && typeof block.text === "string") { - return { type: "text", text: block.text }; + if (block.type === "text" && typeof block.text === "string" && block.text.length > 0) { + const text = captureTextWithinBudget(block.text, budget); + return text === undefined ? undefined : { type: "text", text }; } if (block.type === "thinking" && typeof block.thinking === "string") { - return { type: "thinking", thinking: block.thinking }; + const thinking = captureTextWithinBudget(block.thinking, budget); + return thinking === undefined ? undefined : { type: "thinking", thinking }; } if ( (block.type === "tool_use" || @@ -45,37 +147,122 @@ function assistantContentBlock(block: unknown): Record | undefi block.type === "mcp_tool_use") && typeof block.name === "string" ) { + const name = captureTextWithinBudget(block.name, budget); + if (name === undefined) { + return undefined; + } + const id = typeof block.id === "string" ? captureTextWithinBudget(block.id, budget) : undefined; return { type: "tool_call", - name: block.name, - ...(typeof block.id === "string" ? { id: block.id } : {}), + name, + ...(id !== undefined ? { id } : {}), }; } return undefined; } +function isCapturableAssistantContentBlock(block: unknown): boolean { + if (!isRecord(block)) { + return false; + } + return ( + (block.type === "text" && typeof block.text === "string") || + (block.type === "thinking" && typeof block.thinking === "string") || + ((block.type === "tool_use" || + block.type === "server_tool_use" || + block.type === "mcp_tool_use") && + typeof block.name === "string") + ); +} + +function isTextAssistantContentBlock(block: unknown): boolean { + return ( + isRecord(block) && + block.type === "text" && + typeof block.text === "string" && + block.text.length > 0 + ); +} + +function assistantMessageHasText(message: unknown): boolean { + if (!isRecord(message)) { + return false; + } + if (typeof message.content === "string") { + return message.content.length > 0; + } + if (!Array.isArray(message.content)) { + return false; + } + const limit = Math.min(message.content.length, MAX_CAPTURED_OUTPUT_BLOCKS); + for (let index = 0; index < limit; index += 1) { + if (isTextAssistantContentBlock(message.content[index])) { + return true; + } + } + return false; +} + // Claude's assistant envelopes can contain native tool arguments and opaque // thinking signatures. Keep only the visible response blocks OpenClaw can // represent accurately; external harness tool spans stay metadata-only. -function normalizeClaudeAssistantMessage(message: unknown): Record | undefined { +function normalizeClaudeAssistantMessage( + message: unknown, + budget: DiagnosticContentBudget, +): Record | undefined { if (!isRecord(message)) { return undefined; } - const content = - typeof message.content === "string" - ? [{ type: "text", text: message.content }] - : Array.isArray(message.content) - ? message.content - .map(assistantContentBlock) - .filter((block): block is Record => Boolean(block)) - : []; + const content: Record[] = []; + if (typeof message.content === "string") { + if (message.content.length === 0) { + return undefined; + } + releaseFallbackReserve(budget); + const text = captureTextWithinBudget(message.content, budget); + if (text !== undefined && budget.remainingItems > 0) { + content.push({ type: "text", text }); + budget.remainingItems -= 1; + } else if (text !== undefined) { + budget.truncated = true; + } + } else if (Array.isArray(message.content)) { + const sourceBlocks = message.content.slice(0, MAX_CAPTURED_OUTPUT_BLOCKS); + if (sourceBlocks.length < message.content.length) { + budget.truncated = true; + } + for (const [index, sourceBlock] of sourceBlocks.entries()) { + if (isTextAssistantContentBlock(sourceBlock)) { + releaseFallbackReserve(budget); + } + const block = assistantContentBlock(sourceBlock, budget); + if (block) { + if (budget.remainingItems > 0) { + content.push(block); + budget.remainingItems -= 1; + } else { + budget.truncated = true; + } + } + if (budget.remainingBytes <= 0 || budget.remainingItems <= 0) { + if (sourceBlocks.slice(index + 1).some(isCapturableAssistantContentBlock)) { + budget.truncated = true; + } + break; + } + } + } if (content.length === 0) { return undefined; } + const stopReason = + typeof message.stop_reason === "string" + ? captureTextWithinBudget(message.stop_reason, budget) + : undefined; return { role: "assistant", content, - ...(typeof message.stop_reason === "string" ? { stopReason: message.stop_reason } : {}), + ...(stopReason !== undefined ? { stopReason } : {}), }; } @@ -84,11 +271,29 @@ function hasTextContent(messages: readonly Record[]): boolean { (message) => Array.isArray(message.content) && message.content.some( - (block) => isRecord(block) && block.type === "text" && typeof block.text === "string", + (block) => + isRecord(block) && + block.type === "text" && + typeof block.text === "string" && + block.text.length > 0, ), ); } +function appendOutputTruncationMarker(messages: Record[]): void { + const marker = { type: "text", text: TRUNCATED_CONTENT_SUFFIX }; + if (messages.length < MAX_CAPTURED_OUTPUT_MESSAGES) { + messages.push({ role: "assistant", content: [marker] }); + return; + } + const lastIndex = messages.length - 1; + const lastMessage = messages[lastIndex]; + messages[lastIndex] = { + ...lastMessage, + content: [...(Array.isArray(lastMessage?.content) ? lastMessage.content : []), marker], + }; +} + function privateData(params: { modelContent?: DiagnosticModelCallContent; errorMessage?: string; @@ -128,7 +333,13 @@ export function createClaudeCliModelCallDiagnostics(params: { transport: "paired-node-cli" | "stdio" | "stdio-live"; now?: () => number; }) { - if (params.context.backendResolved.id !== "claude-cli") { + // Listener registration is process-stable after plugin startup. This attempt-local + // snapshot avoids trace ids, capture buffers, and byte accounting when nobody consumes them. + if ( + params.context.backendResolved.id !== "claude-cli" || + !areDiagnosticsEnabledForProcess() || + !hasInternalDiagnosticEventListeners() + ) { return undefined; } @@ -169,6 +380,19 @@ export function createClaudeCliModelCallDiagnostics(params: { trace, }; const capturedAssistantMessages: Record[] = []; + const outputContentBudget: DiagnosticContentBudget = { + remainingBytes: + MAX_CAPTURED_CONTENT_BYTES - + MAX_CAPTURED_OUTPUT_STRUCTURE_BYTES - + TRUNCATED_CONTENT_SUFFIX_BYTES - + FALLBACK_RESPONSE_RESERVE_BYTES, + // One item stays available for the truncation marker; one more is held + // separately for the final visible fallback until normal text arrives. + remainingItems: MAX_CAPTURED_OUTPUT_BLOCKS - 2, + fallbackReserveBytes: FALLBACK_RESPONSE_RESERVE_BYTES, + fallbackReserveItems: 1, + truncated: false, + }; let started = false; let terminalEmitted = false; let startedAt = 0; @@ -186,19 +410,37 @@ export function createClaudeCliModelCallDiagnostics(params: { ...(capture.inputMessages ? { inputMessages: cloneDiagnosticContentValue([ - { role: "user", content: [{ type: "text", text: params.prompt }] }, + { + role: "user", + content: [{ type: "text", text: captureBoundedText(params.prompt) }], + }, ]), } : {}), - ...(capture.systemPrompt && params.systemPrompt ? { systemPrompt: params.systemPrompt } : {}), + ...(capture.systemPrompt && params.systemPrompt + ? { systemPrompt: captureBoundedText(params.systemPrompt) } + : {}), }; return Object.keys(content).length > 0 ? content : undefined; }; const outputMessages = (output?: CliOutput): unknown => { const messages = capturedAssistantMessages.slice(); const responseText = output?.rawText ?? output?.text; - if (!hasTextContent(messages) && responseText) { - messages.push({ role: "assistant", content: [{ type: "text", text: responseText }] }); + if ( + !hasTextContent(messages) && + responseText && + messages.length < MAX_CAPTURED_OUTPUT_MESSAGES + ) { + const fallback = normalizeClaudeAssistantMessage( + { content: responseText }, + outputContentBudget, + ); + if (fallback) { + messages.push(fallback); + } + } + if (outputContentBudget.truncated) { + appendOutputTruncationMarker(messages); } return cloneDiagnosticContentValue(messages); }; @@ -238,20 +480,32 @@ export function createClaudeCliModelCallDiagnostics(params: { observeRequestPayload: (payload: string): void => { requestPayloadBytes = Buffer.byteLength(payload, "utf8"); }, - observeCliOutput: (chunk: string, stream: "stderr" | "stdout"): void => { + observeCliOutput: ( + chunk: string, + stream: "stderr" | "stdout", + knownByteLength?: number, + ): void => { if (!chunk) { return; } firstCliOutputAt ??= now(); if (stream === "stdout") { - responseStreamBytes += Buffer.byteLength(chunk, "utf8"); + responseStreamBytes += knownByteLength ?? Buffer.byteLength(chunk, "utf8"); } }, observeAssistantMessage: (message: unknown): void => { - if (!capture.outputMessages) { + if ( + !capture.outputMessages || + ((outputContentBudget.remainingBytes <= 0 || outputContentBudget.remainingItems <= 0) && + !(outputContentBudget.fallbackReserveItems > 0 && assistantMessageHasText(message))) || + capturedAssistantMessages.length >= MAX_CAPTURED_OUTPUT_MESSAGES - 1 + ) { + if (capture.outputMessages) { + outputContentBudget.truncated = true; + } return; } - const normalized = normalizeClaudeAssistantMessage(message); + const normalized = normalizeClaudeAssistantMessage(message, outputContentBudget); if (normalized) { capturedAssistantMessages.push(normalized); } diff --git a/src/infra/diagnostic-event-listener-presence.ts b/src/infra/diagnostic-event-listener-presence.ts new file mode 100644 index 000000000000..778bf3904017 --- /dev/null +++ b/src/infra/diagnostic-event-listener-presence.ts @@ -0,0 +1,50 @@ +/** Process-wide listener counts used to avoid telemetry work without consumers. */ + +const DIAGNOSTIC_EVENT_LISTENER_PRESENCE_KEY = Symbol.for( + "openclaw.diagnosticEventListenerPresence.v1", +); + +type DiagnosticEventListenerPresence = { + marker: symbol; + internalCount: number; + trustedCount: number; +}; + +function getDiagnosticEventListenerPresence(): DiagnosticEventListenerPresence { + const globalRecord = globalThis as Record; + const existing = globalRecord[DIAGNOSTIC_EVENT_LISTENER_PRESENCE_KEY]; + if ( + existing && + typeof existing === "object" && + (existing as Partial).marker === + DIAGNOSTIC_EVENT_LISTENER_PRESENCE_KEY + ) { + return existing as DiagnosticEventListenerPresence; + } + const state: DiagnosticEventListenerPresence = { + marker: DIAGNOSTIC_EVENT_LISTENER_PRESENCE_KEY, + internalCount: 0, + trustedCount: 0, + }; + Object.defineProperty(globalThis, DIAGNOSTIC_EVENT_LISTENER_PRESENCE_KEY, { + configurable: true, + enumerable: false, + value: state, + writable: false, + }); + return state; +} + +export function setInternalDiagnosticEventListenerCounts( + internalCount: number, + trustedCount: number, +): void { + const state = getDiagnosticEventListenerPresence(); + state.internalCount = internalCount; + state.trustedCount = trustedCount; +} + +export function hasInternalDiagnosticEventListeners(): boolean { + const state = getDiagnosticEventListenerPresence(); + return state.internalCount > 0 || state.trustedCount > 0; +} diff --git a/src/infra/diagnostic-events.test.ts b/src/infra/diagnostic-events.test.ts index 4132bd991c18..9f30bc270514 100644 --- a/src/infra/diagnostic-events.test.ts +++ b/src/infra/diagnostic-events.test.ts @@ -1,6 +1,8 @@ // Covers diagnostic event emission and metadata handling. import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; +import { hasInternalDiagnosticEventListeners } from "./diagnostic-event-listener-presence.js"; import { + areDiagnosticsEnabledForProcess, emitDiagnosticEvent, emitInternalDiagnosticEvent, emitTrustedDiagnosticEvent, @@ -46,6 +48,25 @@ describe("diagnostic-events", () => { expect((message as string).startsWith(prefix)).toBe(true); } + it("reports active internal diagnostic listeners only while dispatch is enabled", () => { + const hasActiveListeners = () => + areDiagnosticsEnabledForProcess() && hasInternalDiagnosticEventListeners(); + expect(hasActiveListeners()).toBe(false); + + const stopInternal = onInternalDiagnosticEvent(() => undefined); + expect(hasActiveListeners()).toBe(true); + setDiagnosticsEnabledForProcess(false); + expect(hasActiveListeners()).toBe(false); + setDiagnosticsEnabledForProcess(true); + stopInternal(); + expect(hasActiveListeners()).toBe(false); + + const stopTrusted = onTrustedInternalDiagnosticEvent(() => undefined); + expect(hasActiveListeners()).toBe(true); + stopTrusted(); + expect(hasActiveListeners()).toBe(false); + }); + it("emits monotonic seq and timestamps to subscribers", () => { vi.spyOn(Date, "now").mockReturnValueOnce(111).mockReturnValueOnce(222); const events: Array<{ seq: number; ts: number; type: string }> = []; diff --git a/src/infra/diagnostic-events.ts b/src/infra/diagnostic-events.ts index 0efa29b6c1b4..ce716231e574 100644 --- a/src/infra/diagnostic-events.ts +++ b/src/infra/diagnostic-events.ts @@ -2,6 +2,7 @@ import { randomUUID } from "node:crypto"; import type { OpenClawConfig } from "../config/types.openclaw.js"; import type { TalkBrain, TalkEventType, TalkMode, TalkTransport } from "../talk/talk-events.js"; +import { setInternalDiagnosticEventListenerCounts } from "./diagnostic-event-listener-presence.js"; import { formatDiagnosticTraceparent, getActiveDiagnosticTraceContext, @@ -1387,8 +1388,10 @@ export function emitFailoverEvent(event: Omit void { const state = getDiagnosticEventsState(); state.listeners.add(listener); + setInternalDiagnosticEventListenerCounts(state.listeners.size, state.trustedListeners.size); return () => { state.listeners.delete(listener); + setInternalDiagnosticEventListenerCounts(state.listeners.size, state.trustedListeners.size); }; } @@ -1398,8 +1401,10 @@ export function onTrustedInternalDiagnosticEvent( ): () => void { const state = getDiagnosticEventsState(); state.trustedListeners.add(listener); + setInternalDiagnosticEventListenerCounts(state.listeners.size, state.trustedListeners.size); return () => { state.trustedListeners.delete(listener); + setInternalDiagnosticEventListenerCounts(state.listeners.size, state.trustedListeners.size); }; } @@ -1466,6 +1471,7 @@ export function resetDiagnosticEventsForTest(): void { state.seq = 0; state.listeners.clear(); state.trustedListeners.clear(); + setInternalDiagnosticEventListenerCounts(0, 0); state.toolExecutionListeners.clear(); state.toolExecutionSeq = 0; state.dispatchDepth = 0;