From 22226bf2f67037da76ef75ca4e2417fe42fcb655 Mon Sep 17 00:00:00 2001 From: Peter Steinberger Date: Sat, 25 Jul 2026 08:07:41 -0700 Subject: [PATCH] refactor(agents): split CLI runner execution (#113716) * refactor(agents): split CLI runner execution * refactor(agents): keep CLI log helper private --- config/max-lines-baseline.txt | 1 - src/agents/cli-runner/execute-deps.ts | 21 + src/agents/cli-runner/execute-events.ts | 300 +++ src/agents/cli-runner/execute-logging.ts | 210 ++ src/agents/cli-runner/execute-process.ts | 537 ++++ .../cli-runner/execute-tool-tracking.ts | 622 +++++ src/agents/cli-runner/execute.ts | 2238 ++--------------- 7 files changed, 1964 insertions(+), 1965 deletions(-) create mode 100644 src/agents/cli-runner/execute-deps.ts create mode 100644 src/agents/cli-runner/execute-events.ts create mode 100644 src/agents/cli-runner/execute-logging.ts create mode 100644 src/agents/cli-runner/execute-process.ts create mode 100644 src/agents/cli-runner/execute-tool-tracking.ts diff --git a/config/max-lines-baseline.txt b/config/max-lines-baseline.txt index b50351f44d9d..6af70a4a7ff5 100644 --- a/config/max-lines-baseline.txt +++ b/config/max-lines-baseline.txt @@ -394,7 +394,6 @@ src/agents/cli-runner.spawn.test.ts src/agents/cli-runner.ts src/agents/cli-runner/claude-live-session.ts src/agents/cli-runner/execute.supervisor-capture.test.ts -src/agents/cli-runner/execute.ts src/agents/cli-runner/prepare.test.ts src/agents/cli-runner/prepare.ts src/agents/code-mode-namespaces.ts diff --git a/src/agents/cli-runner/execute-deps.ts b/src/agents/cli-runner/execute-deps.ts new file mode 100644 index 000000000000..9fce1fefd558 --- /dev/null +++ b/src/agents/cli-runner/execute-deps.ts @@ -0,0 +1,21 @@ +import { invokeNodeClaudeCliRun } from "../../gateway/node-agent-cli-runtime.js"; +import { requestHeartbeat as requestHeartbeatImpl } from "../../infra/heartbeat-wake.js"; +import { enqueueSystemEvent as enqueueSystemEventImpl } from "../../infra/system-events.js"; +import { getProcessSupervisor as getProcessSupervisorImpl } from "../../process/supervisor/index.js"; +import { + registerExecApprovalRequestForHostOrThrow, + resolveRegisteredExecApprovalDecision, +} from "../bash-tools.exec-approval-request.js"; +import { writeCliSystemPromptFile } from "./helpers.js"; + +export const executeDeps = { + getProcessSupervisor: getProcessSupervisorImpl, + enqueueSystemEvent: enqueueSystemEventImpl, + requestHeartbeat: requestHeartbeatImpl, + writeCliSystemPromptFile, + invokeNodeClaudeCliRun, + registerExecApprovalRequestForHostOrThrow, + resolveRegisteredExecApprovalDecision, +}; + +export type CliExecuteDeps = typeof executeDeps; diff --git a/src/agents/cli-runner/execute-events.ts b/src/agents/cli-runner/execute-events.ts new file mode 100644 index 000000000000..1cfebf5f2575 --- /dev/null +++ b/src/agents/cli-runner/execute-events.ts @@ -0,0 +1,300 @@ +import { emitAgentEvent } from "../../infra/agent-events.js"; +import { emitTrustedDiagnosticEvent } from "../../infra/diagnostic-events.js"; +import type { + CliPlanUpdate, + CliStreamingDelta, + CliThinkingDelta, + CliThinkingProgress, + CliToolUseStartDelta, +} from "../cli-output.js"; +import { sanitizeToolArgs, sanitizeToolResult } from "../embedded-agent-subscribe.tools.js"; +import { applyPluginTextReplacements } from "../plugin-text-transforms.js"; +import { resolveCliToolTerminalReason } from "../run-termination.js"; +import type { CliToolTracking } from "./execute-tool-tracking.js"; +import type { PreparedCliRunContext } from "./types.js"; + +type CliToolResult = { + toolCallId: string; + name: string; + isError: boolean; + result?: unknown; +}; + +export function createCliEventHandlers(params: { + context: PreparedCliRunContext; + toolTracking: CliToolTracking; + getRunState: () => { failed: boolean; error: unknown }; +}) { + const context = params.context; + const runParams = context.params; + const emitLiveEvents = runParams.executionMode !== "side-question"; + let observedCliActivity = false; + let signaledToolExecutionStarted = false; + let signaledAssistantOutputStarted = false; + let commentaryCounter = 0; + const activeParsedTools = new Map< + string, + { startedAt: number; toolName: string; kind: CliToolUseStartDelta["kind"] } + >(); + const emitCliToolUseStart = (event: CliToolUseStartDelta) => { + observedCliActivity = true; + if (!signaledToolExecutionStarted) { + signaledToolExecutionStarted = true; + runParams.onExecutionPhase?.({ + phase: "tool_execution_started", + provider: runParams.provider, + model: context.modelId, + backend: context.backendResolved.id, + }); + } + params.toolTracking.handleCliToolUseStart(event); + if (emitLiveEvents) { + emitAgentEvent({ + runId: runParams.runId, + stream: "tool", + data: { + phase: "start", + name: event.name, + toolCallId: event.toolCallId, + args: sanitizeToolArgs(event.args), + }, + }); + } + }; + const emitCliToolResult = (event: CliToolResult) => { + observedCliActivity = true; + params.toolTracking.handleCliToolResult(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, { + startedAt, + toolName: event.name, + kind: event.kind, + }); + emitTrustedDiagnosticEvent({ + type: "tool.execution.started", + runId: runParams.runId, + sessionId: runParams.sessionId, + ...(runParams.sessionKey ? { sessionKey: runParams.sessionKey } : {}), + ...(runParams.agentId ? { agentId: runParams.agentId } : {}), + toolName: event.name, + toolSource: event.name.startsWith("mcp__") ? "mcp" : "core", + toolOwner: "cli-runner", + toolCallId: event.toolCallId, + }); + emitCliToolUseStart(event); + }; + const emitParsedToolTerminal = (event: { + toolCallId: string; + name: string; + isError: boolean; + incomplete?: boolean; + }) => { + const activeTool = activeParsedTools.get(event.toolCallId); + activeParsedTools.delete(event.toolCallId); + const trustedOutcome = params.toolTracking.resolveCliLoopbackTerminalOutcome(event.toolCallId); + const toolName = activeTool?.toolName ?? event.name; + const now = Date.now(); + const trustedTerminalReason = + trustedOutcome && + trustedOutcome.outcome !== "blocked" && + trustedOutcome.outcome !== "completed" && + trustedOutcome.outcome !== "unknown" + ? trustedOutcome.outcome + : undefined; + const runState = params.getRunState(); + const terminalReason = + trustedTerminalReason ?? + resolveCliToolTerminalReason({ + error: event.incomplete ? runState.error : undefined, + abortSignal: runParams.abortSignal, + }); + // Incomplete client/MCP tools inherit the enclosing failed run even when + // the loopback disconnect is ambiguous. Server-native tools do not. + const useEnclosingTerminalReason = + event.incomplete && + runState.failed && + activeTool !== undefined && + activeTool.kind !== "server_tool_use"; + const diagnosticBase = { + runId: runParams.runId, + sessionId: runParams.sessionId, + ...(runParams.sessionKey ? { sessionKey: runParams.sessionKey } : {}), + ...(runParams.agentId ? { agentId: runParams.agentId } : {}), + toolName, + toolSource: toolName.startsWith("mcp__") ? ("mcp" as const) : ("core" as const), + toolOwner: "cli-runner", + toolCallId: event.toolCallId, + durationMs: Math.max(0, now - (activeTool?.startedAt ?? now)), + }; + if (trustedOutcome?.outcome === "unknown" && !useEnclosingTerminalReason) { + emitTrustedDiagnosticEvent({ + type: "tool.execution.error", + ...diagnosticBase, + errorCategory: "cli_tool_ambiguous", + errorCode: "tool_outcome_unknown", + }); + return; + } + if (event.incomplete && activeTool?.kind === "server_tool_use" && !trustedOutcome) { + emitTrustedDiagnosticEvent({ + type: "tool.execution.error", + ...diagnosticBase, + errorCategory: "cli_tool_ambiguous", + errorCode: "tool_outcome_unknown", + }); + return; + } + const trustedFailure = trustedOutcome !== undefined && trustedOutcome.outcome !== "completed"; + emitTrustedDiagnosticEvent( + trustedOutcome?.outcome === "blocked" + ? { + type: "tool.execution.blocked", + ...diagnosticBase, + deniedReason: trustedOutcome.deniedReason, + reason: "blocked by before-tool policy", + } + : trustedFailure || (!trustedOutcome && event.isError) + ? { + type: "tool.execution.error", + ...diagnosticBase, + errorCategory: + terminalReason === "cancelled" + ? "aborted" + : event.incomplete && (!trustedOutcome || useEnclosingTerminalReason) + ? "cli_tool_incomplete" + : "cli_tool", + terminalReason, + } + : { type: "tool.execution.completed", ...diagnosticBase }, + ); + }; + const emitParsedToolResult = (event: CliToolResult) => { + emitParsedToolTerminal(event); + emitCliToolResult(event); + }; + const finalizeParsedTools = () => { + for (const [toolCallId, activeTool] of Array.from(activeParsedTools)) { + emitParsedToolTerminal({ + toolCallId, + name: activeTool.toolName, + isError: true, + incomplete: true, + }); + } + }; + const emitCliCommentaryText = (text: string) => { + if (!emitLiveEvents) { + return; + } + commentaryCounter += 1; + emitAgentEvent({ + runId: runParams.runId, + stream: "item", + data: { + kind: "preamble", + itemId: `commentary-${runParams.runId}-${commentaryCounter}`, + phase: "update", + title: "commentary", + status: "running", + progressText: applyPluginTextReplacements( + text, + context.backendResolved.textTransforms?.output, + ), + }, + }); + }; + const emitCliAssistantDelta = ({ text, delta }: CliStreamingDelta) => { + if (text || delta) { + observedCliActivity = true; + if (!signaledAssistantOutputStarted) { + signaledAssistantOutputStarted = true; + runParams.onExecutionPhase?.({ + phase: "assistant_output_started", + provider: runParams.provider, + model: context.modelId, + backend: context.backendResolved.id, + }); + } + } + if (emitLiveEvents) { + emitAgentEvent({ + runId: runParams.runId, + stream: "assistant", + data: { + text: applyPluginTextReplacements(text, context.backendResolved.textTransforms?.output), + delta: applyPluginTextReplacements(delta, context.backendResolved.textTransforms?.output), + }, + }); + } + }; + + // Emit-always: thinking reaches the event bus and session archive like the + // embedded reasoning stream; /reasoning and /verbose gate presentation only. + const emitCliThinkingDelta = ({ text, delta, isReasoningSnapshot }: CliThinkingDelta) => { + if (text || delta) { + observedCliActivity = true; + } + if (emitLiveEvents) { + emitAgentEvent({ + runId: runParams.runId, + stream: "thinking", + data: { text, delta, ...(isReasoningSnapshot ? { isReasoningSnapshot } : {}) }, + }); + } + }; + + const emitCliThinkingProgress = ({ progressTokens }: CliThinkingProgress) => { + observedCliActivity = true; + if (emitLiveEvents) { + emitAgentEvent({ + runId: runParams.runId, + stream: "thinking", + data: { progressTokens }, + }); + } + }; + + const emitCliPlanUpdate = ({ steps }: CliPlanUpdate) => { + observedCliActivity = true; + if (emitLiveEvents) { + emitAgentEvent({ + runId: runParams.runId, + stream: "plan", + data: { phase: "update", title: "Plan updated", source: "codex-exec", steps }, + }); + } + }; + + return { + emitLiveEvents, + emitCliToolUseStart, + emitCliToolResult, + emitParsedToolUseStart, + emitParsedToolResult, + finalizeParsedTools, + emitCliCommentaryText, + emitCliAssistantDelta, + emitCliThinkingDelta, + emitCliThinkingProgress, + emitCliPlanUpdate, + hasObservedCliActivity: () => observedCliActivity, + activeParsedToolCount: () => activeParsedTools.size, + }; +} + +export type CliEventHandlers = ReturnType; diff --git a/src/agents/cli-runner/execute-logging.ts b/src/agents/cli-runner/execute-logging.ts new file mode 100644 index 000000000000..98ce3cbea2f9 --- /dev/null +++ b/src/agents/cli-runner/execute-logging.ts @@ -0,0 +1,210 @@ +import crypto from "node:crypto"; +import type { CliReusableSession, PreparedCliRunContext } from "./types.js"; + +function buildCliLogArgs(params: { + args: string[]; + systemPromptArg?: string; + modelArg?: string; + imageArg?: string; + argsPrompt?: string; +}): string[] { + const logArgs: string[] = []; + for (let i = 0; i < params.args.length; i += 1) { + const arg = params.args[i] ?? ""; + if (arg === params.systemPromptArg) { + const systemPromptValue = params.args[i + 1] ?? ""; + logArgs.push(arg, ``); + i += 1; + continue; + } + if (arg === params.modelArg) { + logArgs.push(arg, params.args[i + 1] ?? ""); + i += 1; + continue; + } + if (arg === params.imageArg) { + logArgs.push(arg, ""); + i += 1; + continue; + } + logArgs.push(arg); + } + if (params.argsPrompt) { + const promptIndex = logArgs.indexOf(params.argsPrompt); + if (promptIndex >= 0) { + logArgs[promptIndex] = ``; + } + } + return logArgs; +} +const CLI_ENV_AUTH_LOG_KEYS = [ + "AI_GATEWAY_API_KEY", + "ANTHROPIC_API_KEY", + "ANTHROPIC_API_KEY_OLD", + "ANTHROPIC_API_TOKEN", + "ANTHROPIC_AUTH_TOKEN", + "ANTHROPIC_BASE_URL", + "ANTHROPIC_CUSTOM_HEADERS", + "ANTHROPIC_OAUTH_TOKEN", + "ANTHROPIC_UNIX_SOCKET", + "AZURE_OPENAI_API_KEY", + "CLAUDE_CODE_OAUTH_TOKEN", + "CLAUDE_CODE_PROVIDER_MANAGED_BY_HOST", + "OPENAI_API_KEY", + "OPENAI_STEIPETE_API_KEY", + "OPENROUTER_API_KEY", +] as const; +const CLI_ENV_RUNTIME_LOG_KEYS = ["GEMINI_CLI_HOME", "GEMINI_CLI_SYSTEM_SETTINGS_PATH"] as const; +export const CLAUDE_SELECTED_AUTH_ENV_KEYS = new Set([ + "ANTHROPIC_API_KEY", + "CLAUDE_CODE_OAUTH_TOKEN", +]); +export const NODE_CLAUDE_FORWARD_ENV_KEYS = new Set(["CLAUDE_CODE_AUTO_COMPACT_WINDOW"]); +export function resolveNodeClaudeAuthEnv(context: PreparedCliRunContext): Record { + const secretInput = context.preparedBackend.secretInput; + if (!secretInput) { + return {}; + } + const descriptorEnv = context.preparedBackend.env ?? {}; + const requestEnv = Object.hasOwn(descriptorEnv, "CLAUDE_CODE_OAUTH_TOKEN_FILE_DESCRIPTOR") + ? "CLAUDE_CODE_OAUTH_TOKEN" + : Object.hasOwn(descriptorEnv, "CLAUDE_CODE_API_KEY_FILE_DESCRIPTOR") + ? "ANTHROPIC_API_KEY" + : undefined; + if (!requestEnv) { + return {}; + } + const data = secretInput.createData(); + try { + return { [requestEnv]: data.toString("utf8") }; + } finally { + data.fill(0); + } +} +export const CLI_BACKEND_PRESERVE_ENV = "OPENCLAW_LIVE_CLI_BACKEND_PRESERVE_ENV"; +export function parseCliBackendPreserveEnv(raw: string | undefined): Set { + const trimmed = raw?.trim(); + if (!trimmed) { + return new Set(); + } + if (trimmed.startsWith("[")) { + try { + const parsed = JSON.parse(trimmed) as unknown; + return new Set( + Array.isArray(parsed) + ? parsed.filter((entry): entry is string => typeof entry === "string") + : [], + ); + } catch { + return new Set(); + } + } + return new Set( + trimmed + .split(/[,\s]+/) + .map((entry) => entry.trim()) + .filter((entry) => entry.length > 0), + ); +} +function listPresentCliEnvKeys( + env: Record, + keys: readonly string[], +): string[] { + return keys.filter((key) => { + const value = env[key]; + return typeof value === "string" && value.length > 0; + }); +} +function formatCliEnvKeyList(keys: readonly string[]): string { + return keys.length > 0 ? keys.join(",") : "none"; +} +function buildCliEnvMcpLog(childEnv: Record): string { + return [ + `token=${childEnv.OPENCLAW_MCP_TOKEN ? "set" : "missing"}`, + `capture=${childEnv.OPENCLAW_MCP_CLI_CAPTURE_KEY ? "set" : "missing"}`, + ].join(" "); +} +function fingerprintCliSessionId(sessionId?: string): string { + const trimmed = sessionId?.trim(); + if (!trimmed) { + return "none"; + } + return crypto.createHash("sha256").update(trimmed).digest("hex").slice(0, 12); +} +function formatCliSessionReuseLogState(reusableSession: CliReusableSession): string { + switch (reusableSession.mode) { + case "reuse": + return "reusable"; + case "reuse-with-drift": + return `reusable-drift:${reusableSession.drift.reasons.join(",")}`; + case "invalidate": + return `invalidated:${reusableSession.invalidatedReason}`; + case "none": + return "none"; + } + const exhaustive: never = reusableSession; + return exhaustive; +} + +/** Builds the compact execution summary logged before a CLI backend run. */ +export function buildCliExecLogLine(params: { + provider: string; + model: string; + promptChars: number; + trigger?: string; + useResume: boolean; + cliSessionId?: string; + resolvedSessionId?: string; + reusableSession: CliReusableSession; + hasHistoryPrompt: boolean; +}): string { + return [ + `cli exec: provider=${params.provider}`, + `model=${params.model}`, + `promptChars=${params.promptChars}`, + `trigger=${params.trigger ?? "unknown"}`, + `useResume=${params.useResume ? "true" : "false"}`, + `session=${params.cliSessionId ? "present" : "none"}`, + `resumeSession=${params.useResume ? fingerprintCliSessionId(params.resolvedSessionId) : "none"}`, + `reuse=${formatCliSessionReuseLogState(params.reusableSession)}`, + `historyPrompt=${params.hasHistoryPrompt ? "present" : "none"}`, + ].join(" "); +} + +/** Summarizes auth-related env keys preserved or cleared for a CLI child process. */ +export function buildCliEnvAuthLog(childEnv: Record): string { + const hostKeys = listPresentCliEnvKeys(process.env, CLI_ENV_AUTH_LOG_KEYS); + const childKeys = listPresentCliEnvKeys(childEnv, CLI_ENV_AUTH_LOG_KEYS); + const childKeySet = new Set(childKeys); + const clearedKeys = hostKeys.filter((key) => !childKeySet.has(key)); + const runtimeHostKeys = listPresentCliEnvKeys(process.env, CLI_ENV_RUNTIME_LOG_KEYS); + const runtimeChildKeys = listPresentCliEnvKeys(childEnv, CLI_ENV_RUNTIME_LOG_KEYS); + const runtimeChildKeySet = new Set(runtimeChildKeys); + const runtimeClearedKeys = runtimeHostKeys.filter((key) => !runtimeChildKeySet.has(key)); + return [ + `host=${formatCliEnvKeyList(hostKeys)}`, + `child=${formatCliEnvKeyList(childKeys)}`, + `cleared=${formatCliEnvKeyList(clearedKeys)}`, + `runtimeHost=${formatCliEnvKeyList(runtimeHostKeys)}`, + `runtimeChild=${formatCliEnvKeyList(runtimeChildKeys)}`, + `runtimeCleared=${formatCliEnvKeyList(runtimeClearedKeys)}`, + ].join(" "); +} + +export function logCliInvocation(params: { + args: string[]; + command: string; + env: Record; + systemPromptArg?: string; + modelArg?: string; + imageArg?: string; + argsPrompt?: string; + log: (message: string) => void; +}): void { + const logArgs = buildCliLogArgs(params); + params.log(`cli argv: ${params.command} ${logArgs.join(" ")}`); + params.log(`cli env auth: ${buildCliEnvAuthLog(params.env)}`); + if (params.env.OPENCLAW_MCP_TOKEN) { + params.log(`cli env mcp: ${buildCliEnvMcpLog(params.env)}`); + } +} diff --git a/src/agents/cli-runner/execute-process.ts b/src/agents/cli-runner/execute-process.ts new file mode 100644 index 000000000000..5a639364e5d6 --- /dev/null +++ b/src/agents/cli-runner/execute-process.ts @@ -0,0 +1,537 @@ +import crypto from "node:crypto"; +import { shouldLogVerbose } from "../../globals.js"; +import { + resolveEventSessionKeyForPolicy, + resolveEventSessionRoutingPolicy, + scopedHeartbeatWakeOptionsForPolicy, +} from "../../infra/event-session-routing.js"; +import type { CliBackendConfig } from "../../plugins/cli-backend.types.js"; +import type { RunExit } from "../../process/supervisor/types.js"; +import { + createCliJsonlStreamingParser, + extractCliErrorMessage, + parseCliOutput, + type CliOutput, +} from "../cli-output.js"; +import { classifyFailoverReason } from "../embedded-agent-helpers.js"; +import { FailoverError, resolveFailoverStatus } from "../failover-error.js"; +import { applyPluginTextReplacements } from "../plugin-text-transforms.js"; +import { runClaudeLiveSessionTurn } from "./claude-live-session.js"; +import type { CliExecuteDeps } from "./execute-deps.js"; +import type { CliEventHandlers } from "./execute-events.js"; +import { + createCliAbortError, + executeNodeClaudeRun, + type resolveNodeClaudePlacement, +} from "./execute-node-claude.js"; +import { appendCliOutputTail } from "./execute-output-buffer.js"; +import type { CliToolTracking } from "./execute-tool-tracking.js"; +import { buildCliSupervisorScopeKey } from "./helpers.js"; +import { cliBackendLog, formatCliBackendOutputDigest } from "./log.js"; +import type { createClaudeCliModelCallDiagnostics } from "./model-call-diagnostics.js"; +import { createCliOutputFailoverError } from "./output-error.js"; +import type { PreparedCliRunContext } from "./types.js"; + +const CLI_RUNNER_OUTPUT_PARSE_BYTES = 1024 * 1024; + +function appendCliOutputParseBuffer(buffer: Buffer, chunk: string) { + if (!chunk) { + return { buffer, exceeded: false }; + } + const chunkBuffer = Buffer.from(chunk); + if (buffer.byteLength + chunkBuffer.byteLength <= CLI_RUNNER_OUTPUT_PARSE_BYTES) { + return { + buffer: Buffer.concat([buffer, chunkBuffer], buffer.byteLength + chunkBuffer.byteLength), + exceeded: false, + }; + } + const remainingBytes = CLI_RUNNER_OUTPUT_PARSE_BYTES - buffer.byteLength; + return { + buffer: + remainingBytes <= 0 + ? buffer + : Buffer.concat( + [buffer, chunkBuffer.subarray(0, remainingBytes)], + CLI_RUNNER_OUTPUT_PARSE_BYTES, + ), + exceeded: true, + }; +} + +type ExecuteCliProcessOptions = { + onPhase?: (phase: "send" | "resolve" | "cleanup") => void; +}; + +export async function executeCliProcess(params: { + context: PreparedCliRunContext; + backend: CliBackendConfig; + deps: CliExecuteDeps; + events: CliEventHandlers; + toolTracking: CliToolTracking; + diagnostics: ReturnType; + nodePlacement: ReturnType; + nodeSystemPrompt?: string; + nodeEnv?: Record; + nodeClearEnv?: string[]; + useManagedClaudeLiveSession: boolean; + useResume: boolean; + cliSessionIdToUse?: string; + resolvedSessionId?: string; + executionCommand: string; + executionLeadingArgv: readonly string[]; + executionArgs: string[]; + env: Record; + prompt: string; + argsPrompt?: string; + stdin?: string; + noOutputTimeoutMs: number; + outputMode: CliBackendConfig["output"]; + logOutputText: boolean; + cliTurnStartedAt: number; + fallbackCleanup?: () => Promise; + claimFallbackCleanup: () => void; + observeForkSuccessor: (sessionId: string) => void; + options?: ExecuteCliProcessOptions; +}): Promise { + const context = params.context; + const runParams = context.params; + const failoverContext = { + provider: runParams.provider, + model: context.modelId, + sessionId: runParams.sessionId, + lane: runParams.lane, + }; + const outputErrorContext = { ...failoverContext, runId: runParams.runId }; + const hasJsonlOutput = params.outputMode === "jsonl"; + if (params.useManagedClaudeLiveSession) { + if (!hasJsonlOutput) { + throw new Error("Claude live session requires JSONL streaming parser"); + } + runParams.onExecutionPhase?.({ + phase: "process_spawned", + provider: runParams.provider, + model: context.modelId, + backend: context.backendResolved.id, + }); + params.claimFallbackCleanup(); + const liveResult = await runClaudeLiveSessionTurn({ + context, + args: params.executionArgs, + executableCommand: params.executionCommand, + executableLeadingArgv: params.executionLeadingArgv, + env: params.env, + prompt: params.prompt, + useResume: params.useResume, + forceNewSession: + params.cliSessionIdToUse === undefined && context.openClawHistoryPrompt !== undefined, + requiredSessionGeneration: params.cliSessionIdToUse + ? context.requiredClaudeLiveSessionGeneration + : undefined, + noOutputTimeoutMs: params.noOutputTimeoutMs, + getProcessSupervisor: params.deps.getProcessSupervisor, + onAssistantDelta: params.events.emitCliAssistantDelta, + onThinkingDelta: params.events.emitCliThinkingDelta, + onThinkingProgress: params.events.emitCliThinkingProgress, + onToolUseStart: params.events.emitCliToolUseStart, + onToolResult: params.events.emitCliToolResult, + resolveToolResultTerminalOutcome: (event) => { + const outcome = params.toolTracking.resolveCliLoopbackTerminalOutcome(event.toolCallId); + return outcome?.outcome === "completed" ? undefined : outcome; + }, + onCommentaryText: + params.events.emitLiveEvents && runParams.emitCommentaryText + ? params.events.emitCliCommentaryText + : undefined, + onMcpCaptureReady: params.toolTracking.beginGatewayCapture, + cleanup: async () => { + await params.fallbackCleanup?.(); + }, + onSessionId: params.observeForkSuccessor, + onAssistantMessage: params.diagnostics?.observeAssistantMessage, + onUsage: params.diagnostics?.observeUsage, + onCliOutput: params.diagnostics?.observeCliOutput, + onRequestPayload: params.diagnostics?.observeRequestPayload, + onPhase: params.options?.onPhase, + }); + params.options?.onPhase?.("resolve"); + const rawText = liveResult.output.text; + return { + ...liveResult.output, + rawText, + finalPromptText: params.prompt, + text: applyPluginTextReplacements(rawText, context.backendResolved.textTransforms?.output), + }; + } + + const streamingParser = hasJsonlOutput + ? createCliJsonlStreamingParser({ + backend: params.backend, + providerId: context.backendResolved.id, + onAssistantDelta: params.events.emitCliAssistantDelta, + onThinkingDelta: params.events.emitCliThinkingDelta, + onThinkingProgress: params.events.emitCliThinkingProgress, + onPlanUpdate: params.events.emitCliPlanUpdate, + onToolUseStart: params.events.emitParsedToolUseStart, + onToolResult: params.events.emitParsedToolResult, + onCommentaryText: + params.events.emitLiveEvents && runParams.emitCommentaryText + ? params.events.emitCliCommentaryText + : undefined, + onSessionId: params.observeForkSuccessor, + onAssistantMessage: params.diagnostics?.observeAssistantMessage, + onUsage: params.diagnostics?.observeUsage, + }) + : null; + let stdoutTail = ""; + let stdoutParseBuffer: Buffer = Buffer.alloc(0); + let stdoutBytes = 0; + const stdoutHash = crypto.createHash("sha256"); + let stdoutParseExceeded = false; + let stderrTail = ""; + let stderrParseBuffer: Buffer = Buffer.alloc(0); + let stderrBytes = 0; + const stderrHash = crypto.createHash("sha256"); + let stderrParseExceeded = false; + const consumeStdout = (chunk: string) => { + const chunkBytes = Buffer.byteLength(chunk); + params.diagnostics?.observeCliOutput(chunk, "stdout", chunkBytes); + stdoutBytes += chunkBytes; + stdoutHash.update(chunk); + stdoutTail = appendCliOutputTail(stdoutTail, chunk); + if (!stdoutParseExceeded) { + const next = appendCliOutputParseBuffer(stdoutParseBuffer, chunk); + stdoutParseBuffer = next.buffer; + stdoutParseExceeded = next.exceeded; + } + streamingParser?.push(chunk); + }; + const consumeStderr = (chunk: string) => { + params.diagnostics?.observeCliOutput(chunk, "stderr"); + stderrBytes += Buffer.byteLength(chunk); + stderrHash.update(chunk); + stderrTail = appendCliOutputTail(stderrTail, chunk); + if (!stderrParseExceeded) { + const next = appendCliOutputParseBuffer(stderrParseBuffer, chunk); + stderrParseBuffer = next.buffer; + stderrParseExceeded = next.exceeded; + } + }; + + runParams.onExecutionPhase?.({ + phase: "process_spawned", + provider: runParams.provider, + model: context.modelId, + backend: context.backendResolved.id, + }); + let managedRunPid: number | undefined; + let nodeRunAbortSignal: AbortSignal | undefined; + let nodeRunTruncated = false; + let result: RunExit; + params.diagnostics?.observeRequestPayload(params.stdin ?? params.argsPrompt ?? ""); + if (params.nodePlacement) { + const nodeRun = await executeNodeClaudeRun({ + context, + nodePlacement: params.nodePlacement, + executionArgs: params.executionArgs, + stdinPayload: params.stdin ?? "", + ...(params.nodeSystemPrompt !== undefined + ? { nodeSystemPrompt: params.nodeSystemPrompt } + : {}), + ...(params.nodeEnv ? { nodeEnv: params.nodeEnv } : {}), + ...(params.nodeClearEnv ? { nodeClearEnv: params.nodeClearEnv } : {}), + noOutputTimeoutMs: params.noOutputTimeoutMs, + consumeStdout, + consumeStderr, + deps: params.deps, + }); + result = nodeRun.result; + nodeRunAbortSignal = nodeRun.nodeRunAbortSignal; + nodeRunTruncated = nodeRun.nodeRunTruncated; + } else { + const supervisor = params.deps.getProcessSupervisor(); + const scopeKey = buildCliSupervisorScopeKey({ + backend: params.backend, + backendId: context.backendResolved.id, + cliSessionId: params.useResume ? params.resolvedSessionId : undefined, + }); + const managedRun = await supervisor.spawn({ + sessionId: runParams.sessionId, + backendId: context.backendResolved.id, + scopeKey, + replaceExistingScope: Boolean(params.useResume && scopeKey), + mode: "child", + argv: [params.executionCommand, ...params.executionLeadingArgv, ...params.executionArgs], + timeoutMs: runParams.timeoutMs, + noOutputTimeoutMs: params.noOutputTimeoutMs, + cwd: context.cwd ?? context.workspaceDir, + env: params.env, + input: params.stdin ?? "", + secretInput: context.preparedBackend.secretInput, + captureOutput: false, + onStdout: consumeStdout, + onStderr: consumeStderr, + }); + managedRunPid = managedRun.pid; + let replyBackendCompleted = false; + const replyBackendHandle = runParams.replyOperation + ? { + kind: "cli" as const, + cancel: () => managedRun.cancel("manual-cancel"), + isStreaming: () => !replyBackendCompleted, + } + : undefined; + if (replyBackendHandle) { + runParams.replyOperation?.attachBackend(replyBackendHandle); + } + const abortManagedRun = () => managedRun.cancel("manual-cancel"); + runParams.abortSignal?.addEventListener("abort", abortManagedRun, { once: true }); + if (runParams.abortSignal?.aborted) { + abortManagedRun(); + } + try { + result = await managedRun.wait(); + } finally { + replyBackendCompleted = true; + if (replyBackendHandle) { + runParams.replyOperation?.detachBackend(replyBackendHandle); + } + runParams.abortSignal?.removeEventListener("abort", abortManagedRun); + } + } + if ( + (runParams.abortSignal?.aborted || nodeRunAbortSignal?.aborted) && + result.reason === "manual-cancel" + ) { + throw createCliAbortError(); + } + params.options?.onPhase?.("resolve"); + streamingParser?.finish(); + const streamingParserErrorText = + params.outputMode === "jsonl" ? (streamingParser?.getErrorText() ?? null) : null; + if (streamingParserErrorText) { + throw new FailoverError(streamingParserErrorText, { + reason: "format", + ...failoverContext, + status: resolveFailoverStatus("format"), + }); + } + // The node re-injects the terminal result after truncation. If even that is + // missing, the turn outcome is unknowable and cannot pass as a clean exit. + if ( + nodeRunTruncated && + result.exitCode === 0 && + !result.timedOut && + !streamingParser?.getOutput() + ) { + throw new FailoverError( + "paired node truncated the Claude CLI stream before the terminal result; refusing to accept partial output.", + { + reason: "format", + ...failoverContext, + status: resolveFailoverStatus("format"), + }, + ); + } + + const stdout = stdoutParseBuffer.toString("utf8").trim(); + const stdoutDiagnostic = stdoutTail.trim(); + const stderr = stderrParseBuffer.toString("utf8").trim(); + const stderrDiagnostic = stderrTail.trim(); + const processDiagnostics = { + backendId: context.backendResolved.id, + processReason: result.reason, + exitCode: result.exitCode, + exitSignal: result.exitSignal, + durationMs: result.durationMs, + stdoutBytes, + stdoutHash: stdoutHash.digest("hex").slice(0, 12), + stderrBytes, + stderrHash: stderrHash.digest("hex").slice(0, 12), + useResume: params.useResume, + }; + if (params.logOutputText) { + if (stdoutDiagnostic) { + cliBackendLog.info(`cli stdout:\n${stdoutDiagnostic}`); + } + if (stderrDiagnostic) { + cliBackendLog.info(`cli stderr:\n${stderrDiagnostic}`); + } + } + if (shouldLogVerbose()) { + if (stdoutDiagnostic) { + cliBackendLog.debug(`cli stdout:\n${stdoutDiagnostic}`); + } + if (stderrDiagnostic) { + cliBackendLog.debug(`cli stderr:\n${stderrDiagnostic}`); + } + } + + const streamedJsonlOutput = + params.outputMode === "jsonl" ? (streamingParser?.getOutput() ?? null) : null; + const parsedStructuredOutput = + streamedJsonlOutput ?? + (params.outputMode === "json" && !stdoutParseExceeded + ? parseCliOutput({ + raw: stdout, + backend: params.backend, + providerId: context.backendResolved.id, + outputMode: params.outputMode, + fallbackSessionId: params.resolvedSessionId, + }) + : null); + // A completed terminal record is authoritative even if the CLI hangs + // afterward. Reclassifying it as a timeout could replay completed tools. + if (parsedStructuredOutput?.terminalFailure) { + const terminalError = createCliOutputFailoverError({ + output: parsedStructuredOutput, + ...outputErrorContext, + }); + if (terminalError) { + throw terminalError; + } + } + + if (result.exitCode !== 0 || result.reason !== "exit") { + params.options?.onPhase?.("send"); + if (result.reason === "no-output-timeout" || result.noOutputTimedOut) { + const timeoutSeconds = Math.round(params.noOutputTimeoutMs / 1000); + cliBackendLog.warn( + `cli watchdog timeout: provider=${runParams.provider} model=${context.modelId} session=${params.resolvedSessionId ?? runParams.sessionId} noOutputTimeoutMs=${params.noOutputTimeoutMs} pid=${managedRunPid ?? "node"}`, + ); + const retryable = + !params.events.hasObservedCliActivity() && !stdoutDiagnostic && !stderrDiagnostic; + const deferNotice = + retryable && + Boolean(params.cliSessionIdToUse) && + Boolean(params.resolvedSessionId) && + Boolean(context.openClawHistoryPrompt) && + Boolean(runParams.sessionKey) && + runParams.timeoutMs - (Date.now() - context.started) > 0; + if (runParams.sessionKey && params.events.emitLiveEvents && !deferNotice) { + const stallNotice = [ + `CLI agent (${runParams.provider}) produced no output for ${timeoutSeconds}s and was terminated.`, + "It may have been waiting for interactive input or an approval prompt.", + ...(params.nodePlacement + ? ["Check the node's Claude permission settings for pending prompts."] + : ["For Claude Code, prefer --permission-mode bypassPermissions --print."]), + ].join(" "); + const routing = resolveEventSessionRoutingPolicy({ + cfg: runParams.config, + sessionKey: runParams.sessionKey, + channel: runParams.messageProvider, + accountId: runParams.agentAccountId, + }); + params.deps.enqueueSystemEvent(stallNotice, { + sessionKey: resolveEventSessionKeyForPolicy(runParams.sessionKey, routing), + }); + params.deps.requestHeartbeat( + scopedHeartbeatWakeOptionsForPolicy( + runParams.sessionKey, + { source: "cli-watchdog", intent: "event", reason: "cli:watchdog:stall" }, + routing, + ), + ); + } + throw new FailoverError(`CLI produced no output for ${timeoutSeconds}s and was terminated.`, { + reason: "timeout", + ...failoverContext, + status: resolveFailoverStatus("timeout"), + code: retryable ? "cli_no_output_timeout" : undefined, + cliTimeout: { + mode: "no-output", + timeoutSeconds, + observedActivity: params.events.hasObservedCliActivity(), + activeToolCount: params.events.activeParsedToolCount(), + backgroundTaskCount: 0, + }, + }); + } + if (result.reason === "overall-timeout") { + const timeoutSeconds = Math.round(runParams.timeoutMs / 1000); + throw new FailoverError(`CLI exceeded timeout (${timeoutSeconds}s) and was terminated.`, { + reason: "timeout", + ...failoverContext, + status: resolveFailoverStatus("timeout"), + code: "cli_overall_timeout", + cliTimeout: { + mode: "overall", + timeoutSeconds, + observedActivity: params.events.hasObservedCliActivity(), + activeToolCount: params.events.activeParsedToolCount(), + backgroundTaskCount: 0, + }, + }); + } + const errorCandidates = [stderr, stdout, stderrDiagnostic, stdoutDiagnostic].filter(Boolean); + const structuredError = + errorCandidates.map((candidate) => extractCliErrorMessage(candidate)).find(Boolean) ?? null; + let classifiedErrorText = structuredError; + let reason = structuredError + ? classifyFailoverReason(structuredError, { provider: runParams.provider }) + : null; + if (!reason) { + for (const candidate of errorCandidates) { + reason = classifyFailoverReason(candidate, { provider: runParams.provider }); + if (reason) { + classifiedErrorText = candidate; + break; + } + } + } + const errorText = structuredError || classifiedErrorText || errorCandidates[0] || "CLI failed."; + reason ??= "unknown"; + const retryCode = + reason === "context_overflow" + ? "cli_context_overflow" + : reason === "unknown" && + result.reason === "exit" && + errorCandidates.length === 0 && + !params.events.hasObservedCliActivity() + ? "cli_unknown_empty_failure" + : undefined; + throw new FailoverError(errorText, { + reason, + ...failoverContext, + status: resolveFailoverStatus(reason), + code: retryCode, + }); + } + + if (stdoutParseExceeded && !streamedJsonlOutput) { + throw new FailoverError( + `CLI stdout exceeded ${CLI_RUNNER_OUTPUT_PARSE_BYTES} bytes; refusing to parse truncated output.`, + { + reason: "format", + ...failoverContext, + status: resolveFailoverStatus("format"), + }, + ); + } + const parsed = + parsedStructuredOutput ?? + parseCliOutput({ + raw: stdout, + backend: params.backend, + providerId: context.backendResolved.id, + outputMode: params.outputMode, + fallbackSessionId: params.resolvedSessionId, + }); + const parsedError = createCliOutputFailoverError({ + output: parsed, + ...outputErrorContext, + }); + if (parsedError) { + throw parsedError; + } + const rawText = parsed.text; + cliBackendLog.info( + `cli turn: provider=${runParams.provider} model=${context.modelId} durationMs=${Date.now() - params.cliTurnStartedAt} ${formatCliBackendOutputDigest(rawText)}`, + ); + return { + ...parsed, + diagnostics: { ...parsed.diagnostics, process: processDiagnostics }, + rawText, + finalPromptText: params.prompt, + text: applyPluginTextReplacements(rawText, context.backendResolved.textTransforms?.output), + }; +} diff --git a/src/agents/cli-runner/execute-tool-tracking.ts b/src/agents/cli-runner/execute-tool-tracking.ts new file mode 100644 index 000000000000..7ddc83202899 --- /dev/null +++ b/src/agents/cli-runner/execute-tool-tracking.ts @@ -0,0 +1,622 @@ +import { isDeepStrictEqual } from "node:util"; +import { + beginMcpLoopbackToolCallCapture, + clearMcpLoopbackToolCallCapture, + type McpLoopbackToolCallStart, + type McpLoopbackToolCallTerminalOutcome, + waitForMcpLoopbackToolCallCaptureIdle, +} from "../../gateway/mcp-http.loopback-runtime.js"; +import { shouldUseInternalSourceReplySink } from "../../infra/outbound/internal-source-reply.js"; +import type { CliOutput, CliToolUseStartDelta } from "../cli-output.js"; +import { + isDeliveredMessageToolOnlySourceReplyResult, + isDeliveredMessagingToolResult, +} from "../embedded-agent-message-tool-source-reply.js"; +import { + isMessagingTool, + isMessagingToolDeliveryAction, + isMessagingToolSendAction, +} from "../embedded-agent-messaging.js"; +import type { + MessagingToolSend, + MessagingToolSourceReplyPayload, +} from "../embedded-agent-messaging.types.js"; +import { + extractMessagingToolSendResult, + extractMessagingToolSourceReplyPayload, +} from "../embedded-agent-subscribe.tools.js"; +import { rotateClaudeLiveMcpCaptureKeyForContext } from "./claude-live-session.js"; +import { attachCliMessagingDeliveryEvidence } from "./delivery-evidence.js"; +import { + appendUniqueCliMessagingEvidence, + buildMessagingToolSendEvidenceKey, + CLI_MESSAGING_EVIDENCE_MAX_CALLS, + extractCliMessagingContent, + extractCliMessagingTarget, + normalizeCliMessagingToolName, +} from "./execute-messaging.js"; +import type { PreparedCliRunContext } from "./types.js"; + +const CLI_LOOPBACK_CORRELATION_MAX_CALLS = 64; +const CLI_MCP_DELIVERY_DRAIN_GRACE_MS = 5_000; +const CLI_MCP_REQUEST_ADMISSION_GRACE_MS = 250; + +type CliToolTerminalOutcome = McpLoopbackToolCallTerminalOutcome | { outcome: "completed" }; +type CliLoopbackCall = { + admitted: McpLoopbackToolCallStart; + current: McpLoopbackToolCallStart; + boundToolCallId?: string; + outcome?: CliToolTerminalOutcome; + ambiguous: boolean; + ambiguityGroup?: CliLoopbackAmbiguityGroup; +}; +type CliLoopbackAmbiguityGroup = { + calls: Set; + activeToolCallIds: Set; +}; +type ActiveCliTool = { + toolName: string; + args: Record; + loopbackCall?: CliLoopbackCall; + loopbackAmbiguous: boolean; + ambiguityGroup?: CliLoopbackAmbiguityGroup; +}; + +export function createCliToolTracking(context: PreparedCliRunContext) { + let gatewayCaptureKey: string | undefined; + let yielded = false; + let didSendViaMessagingTool = false; + let didDeliverSourceReplyViaMessageTool = false; + let inFlightUnclassifiedMcpRequests = 0; + let inFlightMessagingToolCalls = 0; + const inFlightPreparedMessagingCalls = new Set(); + const pendingMessagingCalls = new Map< + string, + { toolName: string; args: Record; target?: MessagingToolSend } + >(); + const cliLoopbackCalls: CliLoopbackCall[] = []; + const activeCliTools = new Map(); + let cliLoopbackCorrelationOverflowed = false; + const messagingToolSentTexts: string[] = []; + const messagingToolSentTextKeys = new Set(); + const messagingToolSentMediaUrls: string[] = []; + const messagingToolSentMediaUrlKeys = new Set(); + const messagingToolSentTargets: MessagingToolSend[] = []; + const messagingToolSentTargetKeys = new Set(); + const messagingToolSourceReplyPayloads: MessagingToolSourceReplyPayload[] = []; + const matchesCliLoopbackCall = ( + toolName: string, + toolArgs: Record, + call: McpLoopbackToolCallStart, + ) => + normalizeCliMessagingToolName(toolName) === call.toolName && + isDeepStrictEqual(toolArgs, call.args); + const markCliLoopbackCallsAmbiguous = ( + calls: CliLoopbackCall[], + activeEntries = Array.from(activeCliTools.entries()).filter( + ([, activeTool]) => + activeTool.loopbackCall !== undefined && calls.includes(activeTool.loopbackCall), + ), + ) => { + const groups = new Set(); + for (const call of calls) { + if (call.ambiguityGroup) { + groups.add(call.ambiguityGroup); + } + } + for (const [, activeTool] of activeEntries) { + if (activeTool.ambiguityGroup) { + groups.add(activeTool.ambiguityGroup); + } + } + const group = groups.values().next().value ?? { + calls: new Set(), + activeToolCallIds: new Set(), + }; + for (const existing of groups) { + if (existing === group) { + continue; + } + for (const call of existing.calls) { + call.ambiguityGroup = group; + group.calls.add(call); + } + for (const toolCallId of existing.activeToolCallIds) { + const activeTool = activeCliTools.get(toolCallId); + if (activeTool) { + activeTool.ambiguityGroup = group; + group.activeToolCallIds.add(toolCallId); + } + } + existing.calls.clear(); + existing.activeToolCallIds.clear(); + } + for (const call of calls) { + call.ambiguous = true; + call.ambiguityGroup = group; + group.calls.add(call); + } + for (const [toolCallId, activeTool] of activeEntries) { + activeTool.loopbackAmbiguous = true; + activeTool.ambiguityGroup = group; + group.activeToolCallIds.add(toolCallId); + } + }; + const matchingActiveCliTools = (call: McpLoopbackToolCallStart): Array<[string, ActiveCliTool]> => + Array.from(activeCliTools.entries()).filter(([, activeTool]) => + matchesCliLoopbackCall(activeTool.toolName, activeTool.args, call), + ); + const markCliLoopbackSignatureAmbiguous = (call: McpLoopbackToolCallStart) => { + const calls = cliLoopbackCalls.filter((candidate) => + matchesCliLoopbackCall(call.toolName, call.args, candidate.admitted), + ); + markCliLoopbackCallsAmbiguous(calls, matchingActiveCliTools(call)); + }; + const retainCliLoopbackCall = (call: McpLoopbackToolCallStart) => { + if (cliLoopbackCalls.length >= CLI_LOOPBACK_CORRELATION_MAX_CALLS) { + cliLoopbackCorrelationOverflowed = true; + for (const activeTool of activeCliTools.values()) { + if (activeTool.loopbackCall || activeTool.toolName.startsWith("mcp__")) { + activeTool.loopbackAmbiguous = true; + } + } + cliLoopbackCalls.length = 0; + return undefined; + } + const retained: CliLoopbackCall = { admitted: call, current: call, ambiguous: false }; + cliLoopbackCalls.push(retained); + return retained; + }; + const bindCliLoopbackCall = ( + call: CliLoopbackCall, + toolCallId: string, + activeTool: ActiveCliTool, + ) => { + call.boundToolCallId = toolCallId; + activeTool.loopbackCall = call; + activeTool.loopbackAmbiguous ||= call.ambiguous; + if (call.ambiguityGroup) { + activeTool.ambiguityGroup = call.ambiguityGroup; + call.ambiguityGroup.activeToolCallIds.add(toolCallId); + } + }; + const removeCliLoopbackCall = (call: CliLoopbackCall | undefined) => { + if (!call) { + return; + } + const index = cliLoopbackCalls.indexOf(call); + if (index >= 0) { + cliLoopbackCalls.splice(index, 1); + } + }; + const retireCliLoopbackCorrelation = ( + toolCallId: string, + activeTool: ActiveCliTool | undefined, + ) => { + removeCliLoopbackCall(activeTool?.loopbackCall); + const group = activeTool?.ambiguityGroup; + if (!group) { + return; + } + group.activeToolCallIds.delete(toolCallId); + const hasUnboundCall = Array.from(group.calls).some( + (call) => call.boundToolCallId === undefined && cliLoopbackCalls.includes(call), + ); + if (group.activeToolCallIds.size > 0 || hasUnboundCall) { + return; + } + // An ambiguous group owns unbound captures too. Retire the whole group + // once its parsed tools finish so stale calls cannot poison later tools. + for (const call of group.calls) { + removeCliLoopbackCall(call); + } + group.calls.clear(); + }; + const commitMessagingToolResult = (params: { + toolName: string; + target?: MessagingToolSend; + args?: Record; + result?: unknown; + isError?: boolean; + }) => { + if (!isDeliveredMessagingToolResult(params)) { + return; + } + didSendViaMessagingTool = true; + const toolArgs = params.args ?? {}; + const isMessagingSend = isMessagingToolSendAction(params.toolName, toolArgs); + const content = isMessagingSend ? extractCliMessagingContent(toolArgs, params.result) : {}; + if (isMessagingSend) { + appendUniqueCliMessagingEvidence( + messagingToolSentTexts, + messagingToolSentTextKeys, + content.text ? [content.text] : [], + ); + appendUniqueCliMessagingEvidence( + messagingToolSentMediaUrls, + messagingToolSentMediaUrlKeys, + content.mediaUrls ?? [], + ); + if ( + isDeliveredMessageToolOnlySourceReplyResult({ + sourceReplyDeliveryMode: context.params.sourceReplyDeliveryMode, + toolName: params.toolName, + args: params.args, + result: params.result, + isError: params.isError, + }) + ) { + didDeliverSourceReplyViaMessageTool = true; + const payload = extractMessagingToolSourceReplyPayload(params.result); + if (payload) { + if (messagingToolSourceReplyPayloads.length >= CLI_MESSAGING_EVIDENCE_MAX_CALLS) { + messagingToolSourceReplyPayloads.shift(); + } + // Each internal source-reply send is a distinct delivery, even when + // two intentional sends have identical text or media. + messagingToolSourceReplyPayloads.push(payload); + } + } + } + if (!params.target) { + return; + } + const targetWithContent = { + ...extractMessagingToolSendResult(params.target, params.result), + ...content, + }; + const evidenceKey = buildMessagingToolSendEvidenceKey(targetWithContent); + if (messagingToolSentTargetKeys.has(evidenceKey)) { + return; + } + if (messagingToolSentTargets.length >= CLI_MESSAGING_EVIDENCE_MAX_CALLS) { + const removed = messagingToolSentTargets.shift(); + if (removed) { + messagingToolSentTargetKeys.delete(buildMessagingToolSendEvidenceKey(removed)); + } + } + messagingToolSentTargets.push(targetWithContent); + messagingToolSentTargetKeys.add(evidenceKey); + }; + const isPreparedInternalSourceReply = async (call: McpLoopbackToolCallStart) => { + if ( + context.params.sourceReplyDeliveryMode !== "message_tool_only" || + normalizeCliMessagingToolName(call.toolName) !== "message" || + call.args.action !== "send" || + !context.params.config + ) { + return false; + } + return await shouldUseInternalSourceReplySink( + { + cfg: context.params.config, + action: "send", + sessionKey: context.params.sessionKey, + sourceReplyDeliveryMode: context.params.sourceReplyDeliveryMode, + toolContext: { + currentChannelProvider: context.params.messageChannel ?? context.params.messageProvider, + currentChannelId: context.params.currentChannelId, + currentThreadTs: context.params.currentThreadTs, + currentMessageId: context.params.currentMessageId, + }, + }, + call.args, + ); + }; + const beginGatewayCapture = (captureKey: string | undefined) => { + if (!captureKey || gatewayCaptureKey === captureKey) { + return; + } + if (gatewayCaptureKey) { + throw new Error("CLI MCP capture key changed during an active attempt"); + } + context.preparedBackend.mcpClientGrantCapture?.activate(captureKey); + gatewayCaptureKey = captureKey; + const isPotentialDelivery = (toolName: string) => + isMessagingTool(normalizeCliMessagingToolName(toolName)); + const isPreparedDelivery = (toolName: string, toolArgs: Record) => + toolArgs.dryRun !== true && + isMessagingToolDeliveryAction(normalizeCliMessagingToolName(toolName), toolArgs); + beginMcpLoopbackToolCallCapture({ + captureKey, + onYield: () => { + yielded = true; + }, + onRequestStart: () => { + inFlightUnclassifiedMcpRequests += 1; + }, + onRequestClassified: () => { + inFlightUnclassifiedMcpRequests = Math.max(0, inFlightUnclassifiedMcpRequests - 1); + }, + onToolCallStart: (call) => { + const retained = retainCliLoopbackCall(call); + const candidates = matchingActiveCliTools(call); + // Parallel same-name calls can reach the loopback out of stream order. + // Bind only a unique name+arguments match; ambiguity is safer than a wrong outcome. + let matched = + retained && + candidates.length === 1 && + !candidates[0]?.[1].loopbackCall && + !candidates[0]?.[1].loopbackAmbiguous + ? candidates[0] + : undefined; + if (retained && matched) { + bindCliLoopbackCall(retained, matched[0], matched[1]); + } else if (retained && candidates.length > 0) { + markCliLoopbackSignatureAmbiguous(call); + matched = candidates.find(([, activeTool]) => !activeTool.loopbackCall); + if (matched) { + bindCliLoopbackCall(retained, matched[0], matched[1]); + } + } + if (isPotentialDelivery(call.toolName)) { + inFlightMessagingToolCalls += 1; + } + return matched?.[0]; + }, + onToolCallUpdate: ({ previous, current }) => { + const candidates = cliLoopbackCalls.filter((candidate) => + matchesCliLoopbackCall(previous.toolName, previous.args, candidate.current), + ); + const candidate = candidates.at(0); + if (candidates.length === 1 && candidate && !candidate.ambiguous) { + candidate.current = current; + } else if (candidates.length > 0) { + markCliLoopbackCallsAmbiguous(candidates); + } + inFlightPreparedMessagingCalls.delete(previous); + const wasDelivery = isPotentialDelivery(previous.toolName); + const isDelivery = isPreparedDelivery(current.toolName, current.args); + if (wasDelivery !== isDelivery) { + inFlightMessagingToolCalls = Math.max( + 0, + inFlightMessagingToolCalls + (isDelivery ? 1 : -1), + ); + } + if (isDelivery) { + inFlightPreparedMessagingCalls.add(current); + } + }, + onToolCallFinish: (call, { prepared }) => { + const isDelivery = prepared + ? isPreparedDelivery(call.toolName, call.args) + : isPotentialDelivery(call.toolName); + if (isDelivery) { + inFlightMessagingToolCalls = Math.max(0, inFlightMessagingToolCalls - 1); + } + inFlightPreparedMessagingCalls.delete(call); + }, + onToolCallResult: (call) => { + const terminalOutcome: CliToolTerminalOutcome = + call.outcome === "blocked" + ? { outcome: call.outcome, deniedReason: call.deniedReason } + : { outcome: call.outcome }; + const correlated = call.correlationId + ? cliLoopbackCalls.find((candidate) => candidate.boundToolCallId === call.correlationId) + : undefined; + const candidates = correlated + ? [correlated] + : cliLoopbackCalls.filter((candidate) => + matchesCliLoopbackCall(call.toolName, call.args, candidate.current), + ); + if (candidates.length === 1 && candidates[0]) { + candidates[0].outcome = terminalOutcome; + } else if (candidates.length > 1) { + markCliLoopbackCallsAmbiguous(candidates); + } + const toolName = normalizeCliMessagingToolName(call.toolName); + if (isMessagingToolDeliveryAction(toolName, call.args)) { + commitMessagingToolResult({ + toolName, + target: extractCliMessagingTarget(context, toolName, call.args), + args: call.args, + result: "result" in call ? call.result : undefined, + isError: call.outcome !== "completed", + }); + } + }, + }); + }; + const handleCliToolUseStart = (event: CliToolUseStartDelta) => { + if (event.kind !== "server_tool_use") { + const activeTool: ActiveCliTool = { + toolName: event.name, + args: event.args, + loopbackAmbiguous: cliLoopbackCorrelationOverflowed && event.name.startsWith("mcp__"), + }; + activeCliTools.set(event.toolCallId, activeTool); + const admittedCall = { + toolName: normalizeCliMessagingToolName(event.name), + args: event.args, + }; + const pendingCandidates = cliLoopbackCalls.filter( + (candidate) => + candidate.boundToolCallId === undefined && + matchesCliLoopbackCall(event.name, event.args, candidate.admitted), + ); + const hasAssociatedPeer = matchingActiveCliTools(admittedCall).some( + ([toolCallId, peer]) => + toolCallId !== event.toolCallId && + (peer.loopbackCall !== undefined || peer.loopbackAmbiguous), + ); + const pending = pendingCandidates[0]; + if (hasAssociatedPeer || pendingCandidates.length > 1 || pending?.ambiguous) { + markCliLoopbackSignatureAmbiguous(admittedCall); + if (pending) { + bindCliLoopbackCall(pending, event.toolCallId, activeTool); + } + } else if (pendingCandidates.length === 1 && pending) { + bindCliLoopbackCall(pending, event.toolCallId, activeTool); + } + } + const toolName = normalizeCliMessagingToolName(event.name); + if ( + event.kind === "server_tool_use" || + gatewayCaptureKey || + event.args.dryRun === true || + !isMessagingToolDeliveryAction(toolName, event.args) + ) { + return; + } + if (pendingMessagingCalls.size >= CLI_MESSAGING_EVIDENCE_MAX_CALLS) { + const oldestToolCallId = pendingMessagingCalls.keys().next().value; + if (oldestToolCallId !== undefined) { + pendingMessagingCalls.delete(oldestToolCallId); + // Once an unresolved send is evicted, its later result cannot be correlated. + // Fail closed so a failed turn cannot duplicate it. + didSendViaMessagingTool = true; + } + } + pendingMessagingCalls.set(event.toolCallId, { + toolName, + args: event.args, + target: extractCliMessagingTarget(context, toolName, event.args), + }); + }; + const handleCliToolResult = (event: { + toolCallId: string; + name: string; + isError: boolean; + result?: unknown; + }) => { + const activeTool = activeCliTools.get(event.toolCallId); + activeCliTools.delete(event.toolCallId); + retireCliLoopbackCorrelation(event.toolCallId, activeTool); + const pending = pendingMessagingCalls.get(event.toolCallId); + if (pending) { + pendingMessagingCalls.delete(event.toolCallId); + commitMessagingToolResult({ + toolName: pending.toolName, + target: pending.target, + args: pending.args, + result: event.result, + isError: event.isError, + }); + } + }; + const resolveCliLoopbackTerminalOutcome = (toolCallId: string) => { + const activeTool = activeCliTools.get(toolCallId); + if (activeTool?.loopbackAmbiguous) { + return { outcome: "unknown" } as const; + } + return activeTool?.loopbackCall?.outcome; + }; + + const finishDeliveryTracking = async (params: { + useManagedClaudeLiveSession: boolean; + recordRunError: (error: unknown) => void; + }) => { + try { + if (!gatewayCaptureKey && pendingMessagingCalls.size > 0) { + const calls = Array.from(pendingMessagingCalls.values()); + const internalStates = await Promise.all(calls.map(isPreparedInternalSourceReply)); + if (internalStates.some((internal) => !internal)) { + didSendViaMessagingTool = true; + params.recordRunError( + new Error("CLI JSONL message tool call remained unresolved after exit"), + ); + } else { + params.recordRunError( + new Error("CLI JSONL source reply call remained unresolved after exit"), + ); + } + } + if (!gatewayCaptureKey) { + return; + } + const captureBecameIdle = await waitForMcpLoopbackToolCallCaptureIdle(gatewayCaptureKey, { + timeoutMs: CLI_MCP_DELIVERY_DRAIN_GRACE_MS, + admissionGraceMs: CLI_MCP_REQUEST_ADMISSION_GRACE_MS, + }); + if (captureBecameIdle) { + return; + } + if (params.useManagedClaudeLiveSession) { + await rotateClaudeLiveMcpCaptureKeyForContext(context); + } + const internalStates = await Promise.all( + Array.from(inFlightPreparedMessagingCalls).map(isPreparedInternalSourceReply), + ); + const internalCount = internalStates.filter(Boolean).length; + const hasPotentialVisibleSend = inFlightMessagingToolCalls > internalCount; + if (inFlightUnclassifiedMcpRequests > 0 || hasPotentialVisibleSend) { + didSendViaMessagingTool = true; + params.recordRunError(new Error("CLI message tool call remained in flight after exit")); + } else if (inFlightMessagingToolCalls > 0) { + params.recordRunError(new Error("CLI source reply call remained in flight after exit")); + } + } catch (error) { + if ( + pendingMessagingCalls.size > 0 || + inFlightUnclassifiedMcpRequests > 0 || + inFlightMessagingToolCalls > 0 + ) { + didSendViaMessagingTool = true; + } + params.recordRunError(error); + } + }; + + const finalizeCapture = (finalizeParsedTools: () => void) => { + // Captured MCP calls may settle after the CLI process exits. Drain first so + // finalization can use their trusted terminal outcomes. + try { + finalizeParsedTools(); + } finally { + if (gatewayCaptureKey) { + // Fence this exact grant generation before clearing observers; otherwise + // a late request escapes accounting. + try { + context.preparedBackend.mcpClientGrantCapture?.deactivate(gatewayCaptureKey); + } finally { + clearMcpLoopbackToolCallCapture(gatewayCaptureKey); + } + } + } + }; + + const evidence = () => ({ + didSendViaMessagingTool, + didDeliverSourceReplyViaMessageTool, + messagingToolSentTexts, + messagingToolSentMediaUrls, + messagingToolSentTargets, + messagingToolSourceReplyPayloads, + }); + return { + beginGatewayCapture, + handleCliToolUseStart, + handleCliToolResult, + resolveCliLoopbackTerminalOutcome, + finishDeliveryTracking, + finalizeCapture, + withExecutionEvidence(output: CliOutput): CliOutput { + const current = evidence(); + return { + ...output, + ...(yielded ? { yielded: true as const } : {}), + ...(current.didSendViaMessagingTool ? { didSendViaMessagingTool: true } : {}), + ...(current.didDeliverSourceReplyViaMessageTool + ? { didDeliverSourceReplyViaMessageTool: true } + : {}), + ...(current.messagingToolSentTexts.length > 0 + ? { messagingToolSentTexts: current.messagingToolSentTexts.slice() } + : {}), + ...(current.messagingToolSentMediaUrls.length > 0 + ? { messagingToolSentMediaUrls: current.messagingToolSentMediaUrls.slice() } + : {}), + ...(current.messagingToolSentTargets.length > 0 + ? { messagingToolSentTargets: current.messagingToolSentTargets.slice() } + : {}), + ...(current.messagingToolSourceReplyPayloads.length > 0 + ? { messagingToolSourceReplyPayloads: current.messagingToolSourceReplyPayloads.slice() } + : {}), + }; + }, + attachDeliveryEvidence(error: unknown) { + return attachCliMessagingDeliveryEvidence(error, evidence()); + }, + }; +} + +export type CliToolTracking = ReturnType; diff --git a/src/agents/cli-runner/execute.ts b/src/agents/cli-runner/execute.ts index 8e639c673aa5..0a4d9c981c57 100644 --- a/src/agents/cli-runner/execute.ts +++ b/src/agents/cli-runner/execute.ts @@ -1,184 +1,68 @@ -/** - * Executes prepared CLI backend runs, including env isolation, streaming parse, - * live-session routing, and diagnostics. - */ +/** Executes prepared CLI backend runs and owns their queue and resource lifecycle. */ import crypto from "node:crypto"; -import { isDeepStrictEqual } from "node:util"; -import { - beginMcpLoopbackToolCallCapture, - clearMcpLoopbackToolCallCapture, - type McpLoopbackToolCallStart, - type McpLoopbackToolCallTerminalOutcome, - waitForMcpLoopbackToolCallCaptureIdle, -} from "../../gateway/mcp-http.loopback-runtime.js"; -import { invokeNodeClaudeCliRun } from "../../gateway/node-agent-cli-runtime.js"; -import { shouldLogVerbose } from "../../globals.js"; -import { - assertAgentRunLifecycleGenerationCurrent, - emitAgentEvent, -} from "../../infra/agent-events.js"; -import { emitTrustedDiagnosticEvent } from "../../infra/diagnostic-events.js"; +import { assertAgentRunLifecycleGenerationCurrent } from "../../infra/agent-events.js"; import { isTruthyEnvValue } from "../../infra/env.js"; import { formatErrorMessage, toErrorObject } from "../../infra/errors.js"; -import { - resolveEventSessionKeyForPolicy, - resolveEventSessionRoutingPolicy, - scopedHeartbeatWakeOptionsForPolicy, -} from "../../infra/event-session-routing.js"; -import { requestHeartbeat as requestHeartbeatImpl } from "../../infra/heartbeat-wake.js"; import { sanitizeHostExecEnv } from "../../infra/host-env-security.js"; -import { shouldUseInternalSourceReplySink } from "../../infra/outbound/internal-source-reply.js"; -import { enqueueSystemEvent as enqueueSystemEventImpl } from "../../infra/system-events.js"; import type { CliBackendThinkingLevel } from "../../plugins/cli-backend.types.js"; -import { getProcessSupervisor as getProcessSupervisorImpl } from "../../process/supervisor/index.js"; -import type { RunExit } from "../../process/supervisor/types.js"; import { applySkillEnvOverridesFromSnapshot } from "../../skills/runtime/env-overrides.js"; -import { - registerExecApprovalRequestForHostOrThrow, - resolveRegisteredExecApprovalDecision, -} from "../bash-tools.exec-approval-request.js"; import { appendBootstrapPromptWarning } from "../bootstrap-budget.js"; import { fingerprintCliRuntimeArtifact, resolveCliRuntimeOwnerFingerprint, } from "../cli-auth-epoch.js"; import { resolveCliExecutableIdentity } from "../cli-executable-identity.js"; -import { - createCliJsonlStreamingParser, - extractCliErrorMessage, - parseCliOutput, - type CliOutput, - type CliPlanUpdate, - type CliStreamingDelta, - type CliThinkingDelta, - type CliThinkingProgress, - type CliToolUseStartDelta, -} from "../cli-output.js"; -import { classifyFailoverReason } from "../embedded-agent-helpers.js"; -import { - isDeliveredMessageToolOnlySourceReplyResult, - isDeliveredMessagingToolResult, -} from "../embedded-agent-message-tool-source-reply.js"; -import { - isMessagingTool, - isMessagingToolDeliveryAction, - isMessagingToolSendAction, -} from "../embedded-agent-messaging.js"; -import type { - MessagingToolSend, - MessagingToolSourceReplyPayload, -} from "../embedded-agent-messaging.types.js"; +import type { CliOutput } from "../cli-output.js"; import { detectImageReferences, hasHydratableMediaImages, } from "../embedded-agent-runner/run/images.js"; -import { - extractMessagingToolSendResult, - extractMessagingToolSourceReplyPayload, - sanitizeToolArgs, - sanitizeToolResult, -} from "../embedded-agent-subscribe.tools.js"; -import { FailoverError, resolveFailoverStatus } from "../failover-error.js"; import { applyPluginTextReplacements } from "../plugin-text-transforms.js"; -import { resolveCliToolTerminalReason } from "../run-termination.js"; import { prepareCliBundleMcpCaptureAttempt } from "./bundle-mcp.js"; import { closeClaudeLiveSessionForContext, - rotateClaudeLiveMcpCaptureKeyForContext, - runClaudeLiveSessionTurn, shouldUseClaudeLiveSession, } from "./claude-live-session.js"; import { prepareClaudeCliSkillsPlugin } from "./claude-skills-plugin.js"; -import { attachCliMessagingDeliveryEvidence } from "./delivery-evidence.js"; +import { executeDeps } from "./execute-deps.js"; +import { createCliEventHandlers } from "./execute-events.js"; import { - appendUniqueCliMessagingEvidence, - buildMessagingToolSendEvidenceKey, - CLI_MESSAGING_EVIDENCE_MAX_CALLS, - extractCliMessagingContent, - extractCliMessagingTarget, - normalizeCliMessagingToolName, -} from "./execute-messaging.js"; + buildCliEnvAuthLog, + buildCliExecLogLine, + CLAUDE_SELECTED_AUTH_ENV_KEYS, + CLI_BACKEND_PRESERVE_ENV, + logCliInvocation, + NODE_CLAUDE_FORWARD_ENV_KEYS, + parseCliBackendPreserveEnv, + resolveNodeClaudeAuthEnv, +} from "./execute-logging.js"; import { createCliAbortError, - executeNodeClaudeRun, resolveNodeClaudePlacement, stripGatewayLocalClaudeArgs, } from "./execute-node-claude.js"; -import { appendCliOutputTail } from "./execute-output-buffer.js"; +import { executeCliProcess } from "./execute-process.js"; +import { createCliToolTracking } from "./execute-tool-tracking.js"; import { - buildCliSupervisorScopeKey, buildClaudeOwnerKey, buildCliArgs, - resolveCliRunQueueKey, enqueueCliRun, prepareCliPromptImagePayload, resolveCliNoOutputTimeoutMs, + resolveCliRunQueueKey, resolveCliRunTimeoutOverrideMs, resolvePromptInput, resolveSessionIdToSend, resolveSystemPromptUsage, - writeCliSystemPromptFile, } from "./helpers.js"; import { cliBackendLog, CLI_BACKEND_LOG_OUTPUT_ENV, - formatCliBackendOutputDigest, LEGACY_CLAUDE_CLI_LOG_OUTPUT_ENV, } from "./log.js"; import { createClaudeCliModelCallDiagnostics } from "./model-call-diagnostics.js"; -import { createCliOutputFailoverError } from "./output-error.js"; import { buildCliBackendToolAvailability } from "./tool-policy.js"; -import type { CliReusableSession, PreparedCliRunContext } from "./types.js"; - -const CLI_RUNNER_OUTPUT_PARSE_BYTES = 1024 * 1024; - -function appendCliOutputParseBuffer( - buffer: Buffer, - chunk: string, -): { buffer: Buffer; exceeded: boolean } { - if (!chunk) { - return { buffer, exceeded: false }; - } - const chunkBuffer = Buffer.from(chunk); - if (buffer.byteLength + chunkBuffer.byteLength > CLI_RUNNER_OUTPUT_PARSE_BYTES) { - const remainingBytes = CLI_RUNNER_OUTPUT_PARSE_BYTES - buffer.byteLength; - if (remainingBytes <= 0) { - return { buffer, exceeded: true }; - } - return { - buffer: Buffer.concat( - [buffer, chunkBuffer.subarray(0, remainingBytes)], - CLI_RUNNER_OUTPUT_PARSE_BYTES, - ), - exceeded: true, - }; - } - return { - buffer: Buffer.concat([buffer, chunkBuffer], buffer.byteLength + chunkBuffer.byteLength), - exceeded: false, - }; -} - -const executeDeps = { - getProcessSupervisor: getProcessSupervisorImpl, - - enqueueSystemEvent: enqueueSystemEventImpl, - - requestHeartbeat: requestHeartbeatImpl, - - writeCliSystemPromptFile, - - invokeNodeClaudeCliRun, - - registerExecApprovalRequestForHostOrThrow, - resolveRegisteredExecApprovalDecision, -}; - -const CLI_LOOPBACK_CORRELATION_MAX_CALLS = 64; - -const CLI_MCP_DELIVERY_DRAIN_GRACE_MS = 5_000; - -const CLI_MCP_REQUEST_ADMISSION_GRACE_MS = 250; +import type { PreparedCliRunContext } from "./types.js"; function normalizeCliBackendThinkingLevel( level: PreparedCliRunContext["params"]["thinkLevel"], @@ -186,225 +70,12 @@ function normalizeCliBackendThinkingLevel( return level === "ultra" ? "max" : level; } -function buildCliMcpCaptureKey(context: PreparedCliRunContext): string | undefined { - if (!context.mcpDeliveryCapture) { - return undefined; - } - return crypto.randomUUID(); -} - -/** Overrides process/event dependencies for CLI runner execution tests. */ -function setCliRunnerExecuteTestDeps(overrides: Partial): void { - Object.assign(executeDeps, overrides); -} - -function buildCliLogArgs(params: { - args: string[]; - systemPromptArg?: string; - modelArg?: string; - imageArg?: string; - argsPrompt?: string; -}): string[] { - const logArgs: string[] = []; - for (let i = 0; i < params.args.length; i += 1) { - const arg = params.args[i] ?? ""; - if (arg === params.systemPromptArg) { - const systemPromptValue = params.args[i + 1] ?? ""; - logArgs.push(arg, ``); - i += 1; - continue; - } - if (arg === params.modelArg) { - logArgs.push(arg, params.args[i + 1] ?? ""); - i += 1; - continue; - } - if (arg === params.imageArg) { - logArgs.push(arg, ""); - i += 1; - continue; - } - logArgs.push(arg); - } - if (params.argsPrompt) { - const promptIndex = logArgs.indexOf(params.argsPrompt); - if (promptIndex >= 0) { - logArgs[promptIndex] = ``; - } - } - return logArgs; -} - -const CLI_ENV_AUTH_LOG_KEYS = [ - "AI_GATEWAY_API_KEY", - "ANTHROPIC_API_KEY", - "ANTHROPIC_API_KEY_OLD", - "ANTHROPIC_API_TOKEN", - "ANTHROPIC_AUTH_TOKEN", - "ANTHROPIC_BASE_URL", - "ANTHROPIC_CUSTOM_HEADERS", - "ANTHROPIC_OAUTH_TOKEN", - "ANTHROPIC_UNIX_SOCKET", - "AZURE_OPENAI_API_KEY", - "CLAUDE_CODE_OAUTH_TOKEN", - "CLAUDE_CODE_PROVIDER_MANAGED_BY_HOST", - "OPENAI_API_KEY", - "OPENAI_STEIPETE_API_KEY", - "OPENROUTER_API_KEY", -] as const; - -const CLI_ENV_RUNTIME_LOG_KEYS = ["GEMINI_CLI_HOME", "GEMINI_CLI_SYSTEM_SETTINGS_PATH"] as const; -const CLAUDE_SELECTED_AUTH_ENV_KEYS = new Set(["ANTHROPIC_API_KEY", "CLAUDE_CODE_OAUTH_TOKEN"]); -const NODE_CLAUDE_FORWARD_ENV_KEYS = new Set(["CLAUDE_CODE_AUTO_COMPACT_WINDOW"]); - -function resolveNodeClaudeAuthEnv(context: PreparedCliRunContext): Record { - const secretInput = context.preparedBackend.secretInput; - if (!secretInput) { - return {}; - } - const descriptorEnv = context.preparedBackend.env ?? {}; - const requestEnv = Object.hasOwn(descriptorEnv, "CLAUDE_CODE_OAUTH_TOKEN_FILE_DESCRIPTOR") - ? "CLAUDE_CODE_OAUTH_TOKEN" - : Object.hasOwn(descriptorEnv, "CLAUDE_CODE_API_KEY_FILE_DESCRIPTOR") - ? "ANTHROPIC_API_KEY" - : undefined; - if (!requestEnv) { - return {}; - } - const data = secretInput.createData(); - try { - return { [requestEnv]: data.toString("utf8") }; - } finally { - data.fill(0); - } -} - -const CLI_BACKEND_PRESERVE_ENV = "OPENCLAW_LIVE_CLI_BACKEND_PRESERVE_ENV"; - -function parseCliBackendPreserveEnv(raw: string | undefined): Set { - const trimmed = raw?.trim(); - if (!trimmed) { - return new Set(); - } - if (trimmed.startsWith("[")) { - try { - const parsed = JSON.parse(trimmed) as unknown; - return new Set( - Array.isArray(parsed) - ? parsed.filter((entry): entry is string => typeof entry === "string") - : [], - ); - } catch { - return new Set(); - } - } - return new Set( - trimmed - .split(/[,\s]+/) - .map((entry) => entry.trim()) - .filter((entry) => entry.length > 0), - ); -} - -function listPresentCliAuthEnvKeys(env: Record): string[] { - return CLI_ENV_AUTH_LOG_KEYS.filter((key) => { - const value = env[key]; - return typeof value === "string" && value.length > 0; - }); -} - -function listPresentCliRuntimeEnvKeys(env: Record): string[] { - return CLI_ENV_RUNTIME_LOG_KEYS.filter((key) => { - const value = env[key]; - return typeof value === "string" && value.length > 0; - }); -} - -function formatCliEnvKeyList(keys: readonly string[]): string { - return keys.length > 0 ? keys.join(",") : "none"; -} - -function buildCliEnvMcpLog(childEnv: Record): string { - return [ - `token=${childEnv.OPENCLAW_MCP_TOKEN ? "set" : "missing"}`, - `capture=${childEnv.OPENCLAW_MCP_CLI_CAPTURE_KEY ? "set" : "missing"}`, - ].join(" "); -} - -function fingerprintCliSessionId(sessionId?: string): string { - const trimmed = sessionId?.trim(); - if (!trimmed) { - return "none"; - } - return crypto.createHash("sha256").update(trimmed).digest("hex").slice(0, 12); -} - -function formatCliSessionReuseLogState(reusableSession: CliReusableSession): string { - switch (reusableSession.mode) { - case "reuse": - return "reusable"; - case "reuse-with-drift": - return `reusable-drift:${reusableSession.drift.reasons.join(",")}`; - case "invalidate": - return `invalidated:${reusableSession.invalidatedReason}`; - case "none": - return "none"; - } - const exhaustive: never = reusableSession; - return exhaustive; -} - -/** Builds the compact execution summary logged before a CLI backend run. */ -function buildCliExecLogLine(params: { - provider: string; - model: string; - promptChars: number; - trigger?: string; - useResume: boolean; - cliSessionId?: string; - resolvedSessionId?: string; - reusableSession: CliReusableSession; - hasHistoryPrompt: boolean; -}): string { - return [ - `cli exec: provider=${params.provider}`, - `model=${params.model}`, - `promptChars=${params.promptChars}`, - `trigger=${params.trigger ?? "unknown"}`, - `useResume=${params.useResume ? "true" : "false"}`, - `session=${params.cliSessionId ? "present" : "none"}`, - `resumeSession=${params.useResume ? fingerprintCliSessionId(params.resolvedSessionId) : "none"}`, - `reuse=${formatCliSessionReuseLogState(params.reusableSession)}`, - `historyPrompt=${params.hasHistoryPrompt ? "present" : "none"}`, - ].join(" "); -} - -/** Summarizes auth-related env keys preserved or cleared for a CLI child process. */ -function buildCliEnvAuthLog(childEnv: Record): string { - const hostKeys = listPresentCliAuthEnvKeys(process.env); - const childKeys = listPresentCliAuthEnvKeys(childEnv); - const childKeySet = new Set(childKeys); - const clearedKeys = hostKeys.filter((key) => !childKeySet.has(key)); - const runtimeHostKeys = listPresentCliRuntimeEnvKeys(process.env); - const runtimeChildKeys = listPresentCliRuntimeEnvKeys(childEnv); - const runtimeChildKeySet = new Set(runtimeChildKeys); - const runtimeClearedKeys = runtimeHostKeys.filter((key) => !runtimeChildKeySet.has(key)); - return [ - `host=${formatCliEnvKeyList(hostKeys)}`, - `child=${formatCliEnvKeyList(childKeys)}`, - `cleared=${formatCliEnvKeyList(clearedKeys)}`, - `runtimeHost=${formatCliEnvKeyList(runtimeHostKeys)}`, - `runtimeChild=${formatCliEnvKeyList(runtimeChildKeys)}`, - `runtimeCleared=${formatCliEnvKeyList(runtimeClearedKeys)}`, - ].join(" "); -} - if (process.env.VITEST || process.env.NODE_ENV === "test") { (globalThis as Record)[Symbol.for("openclaw.cliRunnerExecuteTestApi")] = { buildCliEnvAuthLog, buildCliExecLogLine, setCliRunnerExecuteTestDeps: (overrides: Record) => { - setCliRunnerExecuteTestDeps(overrides as Partial); + Object.assign(executeDeps, overrides as Partial); }, }; } @@ -438,21 +109,14 @@ export async function executePreparedCliRun( isNewSession: isNew || resendSystemPromptForSoftResume, systemPrompt: context.systemPrompt, }); + const shouldSendSystemPrompt = + systemPromptArg && + (!useResume || backend.systemPromptWhen === "always" || resendSystemPromptForSoftResume); const systemPromptFile = - !nodePlacement && - systemPromptArg && - (!useResume || backend.systemPromptWhen === "always" || resendSystemPromptForSoftResume) - ? await executeDeps.writeCliSystemPromptFile({ - backend, - systemPrompt: systemPromptArg, - }) - : undefined; - const nodeSystemPrompt = - nodePlacement && - systemPromptArg && - (!useResume || backend.systemPromptWhen === "always" || resendSystemPromptForSoftResume) - ? systemPromptArg + !nodePlacement && shouldSendSystemPrompt + ? await executeDeps.writeCliSystemPromptFile({ backend, systemPrompt: systemPromptArg }) : undefined; + const nodeSystemPrompt = nodePlacement && shouldSendSystemPrompt ? systemPromptArg : undefined; const basePrompt = cliSessionIdToUse ? params.prompt @@ -471,11 +135,7 @@ export async function executePreparedCliRun( ) { throw new Error("paired-node Claude CLI sessions do not support attachments or images"); } - const { - prompt: promptWithImages, - imagePaths, - cleanupImages, - } = nodePlacement + const imagePayload = nodePlacement ? { prompt, imagePaths: [] as string[], cleanupImages: async () => {} } : await prepareCliPromptImagePayload({ backend, @@ -486,13 +146,8 @@ export async function executePreparedCliRun( imageOrder: params.imageOrder, media: params.media, }); - prompt = promptWithImages; - - const { argsPrompt, stdin } = resolvePromptInput({ - backend, - prompt, - }); - const stdinPayload = stdin ?? ""; + prompt = imagePayload.prompt; + const { argsPrompt, stdin } = resolvePromptInput({ backend, prompt }); const baseArgs = useResume ? (backend.resumeArgs ?? backend.args ?? []) : (backend.args ?? []); const resolvedArgs = useResume ? baseArgs.map((entry) => entry.replaceAll("{sessionId}", resolvedSessionId ?? "")) @@ -519,9 +174,7 @@ export async function executePreparedCliRun( thinkingLevel: normalizeCliBackendThinkingLevel(params.thinkLevel), executionMode: params.executionMode ?? "agent", // Node runs project the native subset only: gateway-loopback MCP tools do - // not exist on the node, and --allowedTools auto-approval must never cross - // the node boundary. Dropping availability entirely would let a restricted - // session run with the node's full native toolset. + // not exist on the node, and auto-approval must not cross that boundary. toolAvailability: params.cliToolAvailability ? buildCliBackendToolAvailability( nodePlacement @@ -544,21 +197,16 @@ export async function executePreparedCliRun( const executionBaseArgs = nodePlacement ? stripGatewayLocalClaudeArgs(resolvedExecutionArgs ?? baseArgsWithSkills) : (resolvedExecutionArgs ?? baseArgsWithSkills); - const argsBackend = nodePlacement - ? { - ...backend, - systemPromptArg: undefined, - systemPromptFileArg: undefined, - } - : backend; const args = buildCliArgs({ - backend: argsBackend, + backend: nodePlacement + ? { ...backend, systemPromptArg: undefined, systemPromptFileArg: undefined } + : backend, baseArgs: Array.from(executionBaseArgs), modelId: context.normalizedModel, sessionId: resolvedSessionId, systemPrompt: nodePlacement ? undefined : systemPromptArg, systemPromptFilePath: systemPromptFile?.filePath, - imagePaths, + imagePaths: imagePayload.imagePaths, promptArg: argsPrompt, useResume, forkResume: params.forkCliSessionOnResume, @@ -581,15 +229,11 @@ export async function executePreparedCliRun( cliSessionId: useResume ? resolvedSessionId : undefined, ownerKey: claudeOwnerKey, }); - - let completedOutput: CliOutput | undefined; - let executionError: unknown; - let outerCleanupError: Error | undefined; const useManagedClaudeLiveSession = shouldUseClaudeLiveSession(context) && !params.onSuccessfulAuthBinding; // Fresh-session retries invoke this function again. Keep one helper per // observable CLI attempt so every started call retains its own terminal event. - const claudeModelCallDiagnostics = createClaudeCliModelCallDiagnostics({ + const diagnostics = createClaudeCliModelCallDiagnostics({ context, prompt, systemPrompt: systemPromptArg ?? undefined, @@ -599,6 +243,9 @@ export async function executePreparedCliRun( ? "stdio-live" : "stdio", }); + let completedOutput: CliOutput | undefined; + let executionError: unknown; + let outerCleanupError: Error | undefined; let forkResumeClaimed = false; let forkSuccessorObserved = false; let forkSuccessorPersistence: Promise | undefined; @@ -627,7 +274,7 @@ export async function executePreparedCliRun( try { await cleanup?.(); } catch (error) { - if (completedOutput?.didSendViaMessagingTool === true) { + if (completedOutput?.didSendViaMessagingTool) { cliBackendLog.warn( `CLI outer resource cleanup failed after confirmed message delivery: ${formatErrorMessage(error)}`, ); @@ -642,12 +289,236 @@ export async function executePreparedCliRun( throw error; } }; + const executeAttempt = async (): Promise => { + await context.preparedBackend.beforeExecution?.(); + const cliTurnStartedAt = Date.now(); + const restoreSkillEnv = params.skillsSnapshot + ? applySkillEnvOverridesFromSnapshot({ + snapshot: params.skillsSnapshot, + config: params.config, + }) + : undefined; + let cleanupMcpCaptureAttempt: (() => Promise) | undefined; + let runOutput: CliOutput | undefined; + let runError: unknown; + let runFailed = false; + const recordRunError = (error: unknown) => { + if (!runFailed) { + runFailed = true; + runError = error; + } + }; + const toolTracking = createCliToolTracking(context); + const events = createCliEventHandlers({ + context, + toolTracking, + getRunState: () => ({ failed: runFailed, error: runError }), + }); + try { + cliBackendLog.info( + buildCliExecLogLine({ + provider: params.provider, + model: context.normalizedModel, + promptChars: basePrompt.length, + trigger: params.trigger, + useResume, + cliSessionId: cliSessionIdToUse, + resolvedSessionId, + reusableSession: context.reusableCliSession, + hasHistoryPrompt: Boolean(context.openClawHistoryPrompt), + }), + ); + const logOutputText = + isTruthyEnvValue(process.env[CLI_BACKEND_LOG_OUTPUT_ENV]) || + isTruthyEnvValue(process.env[LEGACY_CLAUDE_CLI_LOG_OUTPUT_ENV]); + const outputMode = useResume ? (backend.resumeOutput ?? backend.output) : backend.output; + const initialGatewayCaptureKey = + useManagedClaudeLiveSession || nodePlacement || !context.mcpDeliveryCapture + ? undefined + : crypto.randomUUID(); + const mcpCaptureAttempt = nodePlacement + ? { env: {}, cleanup: undefined } + : await prepareCliBundleMcpCaptureAttempt({ + mode: context.backendResolved.bundleMcpMode, + backend, + env: context.preparedBackend.env, + captureKey: initialGatewayCaptureKey, + }); + cleanupMcpCaptureAttempt = mcpCaptureAttempt.cleanup; + const preparedBackendEnv = context.preparedBackend.env ?? {}; + const hasSelectedClaudeAuth = + Boolean(context.preparedBackend.secretInput) || + [...CLAUDE_SELECTED_AUTH_ENV_KEYS].some((key) => Object.hasOwn(preparedBackendEnv, key)); + const selectedClaudeClearEnv = hasSelectedClaudeAuth + ? new Set(backend.clearEnv ?? []) + : undefined; + const configuredBackendEnv = Object.fromEntries( + Object.entries(backend.env ?? {}).filter(([key]) => !selectedClaudeClearEnv?.has(key)), + ); + const backendEnv = { ...configuredBackendEnv, ...preparedBackendEnv }; + const nodeEnvEntries = Object.entries(preparedBackendEnv).filter(([key]) => + NODE_CLAUDE_FORWARD_ENV_KEYS.has(key), + ); + const nodeEnv = nodePlacement + ? { + ...Object.fromEntries(nodeEnvEntries), + ...resolveNodeClaudeAuthEnv(context), + } + : undefined; + const env = sanitizeHostExecEnv({ baseEnv: process.env, blockPathOverrides: true }); + const preservedEnv = parseCliBackendPreserveEnv(process.env[CLI_BACKEND_PRESERVE_ENV]); + for (const key of backend.clearEnv ?? []) { + if (!preservedEnv.has(key) || selectedClaudeClearEnv?.has(key)) { + delete env[key]; + } + } + if (Object.keys(backendEnv).length > 0) { + Object.assign( + env, + sanitizeHostExecEnv({ + baseEnv: {}, + overrides: backendEnv, + blockPathOverrides: true, + }), + ); + } + Object.assign(env, mcpCaptureAttempt.env); + // Never mark Claude CLI as host-managed. That marker routes runs into + // Anthropic's separate host-managed usage tier instead of normal CLI use. + delete env.CLAUDE_CODE_PROVIDER_MANAGED_BY_HOST; + + let executionCommand = backend.command; + let executionLeadingArgv: readonly string[] = []; + context.runtimeOwnerFingerprint = undefined; + context.runtimeArtifactFingerprint = undefined; + if (params.onSuccessfulAuthBinding && !nodePlacement) { + const executableIdentity = await resolveCliExecutableIdentity({ + command: backend.command, + cwd: context.cwd ?? context.workspaceDir, + env, + ...(context.backendResolved.runtimeArtifact + ? { runtimeArtifact: context.backendResolved.runtimeArtifact } + : {}), + }); + if (!executableIdentity) { + throw new Error( + `CLI backend ${context.backendResolved.id} executable cannot be bound to one durable absolute owner`, + ); + } + executionCommand = executableIdentity.invocation.command; + executionLeadingArgv = executableIdentity.invocation.leadingArgv; + context.runtimeArtifactFingerprint = fingerprintCliRuntimeArtifact({ + provider: params.provider, + backendId: context.backendResolved.id, + executableIdentity, + }); + if (!context.authBindingFingerprint) { + context.runtimeOwnerFingerprint = await resolveCliRuntimeOwnerFingerprint({ + provider: params.provider, + config: params.config ?? context.contextEngineConfig, + ...(context.agentDir ? { agentDir: context.agentDir } : {}), + agentId: params.agentId, + runtimeOwnerId: context.backendResolved.id, + ...(context.effectiveAuthProfileId + ? { authProfileId: context.effectiveAuthProfileId } + : {}), + ...(context.authBindingSkipsLocalCredential ? { skipLocalCredential: true } : {}), + runtimeArtifactFingerprint: context.runtimeArtifactFingerprint, + }); + } + } + if (logOutputText) { + logCliInvocation({ + args, + command: executionCommand, + env, + systemPromptArg: backend.systemPromptArg, + modelArg: backend.modelArg, + imageArg: backend.imageArg, + argsPrompt, + log: (message) => cliBackendLog.info(message), + }); + } + const runTimeoutOverrideMs = resolveCliRunTimeoutOverrideMs({ + config: params.config, + lane: params.lane, + timeoutMs: params.timeoutMs, + runTimeoutOverrideMs: params.runTimeoutOverrideMs, + }); + const noOutputTimeoutMs = resolveCliNoOutputTimeoutMs({ + backend, + timeoutMs: params.timeoutMs, + runTimeoutOverrideMs, + useResume, + trigger: params.trigger, + }); + toolTracking.beginGatewayCapture(initialGatewayCaptureKey); + runOutput = await executeCliProcess({ + context, + backend, + deps: executeDeps, + events, + toolTracking, + diagnostics, + nodePlacement, + nodeSystemPrompt, + nodeEnv: nodeEnv && Object.keys(nodeEnv).length > 0 ? nodeEnv : undefined, + nodeClearEnv: selectedClaudeClearEnv ? [...selectedClaudeClearEnv] : undefined, + useManagedClaudeLiveSession, + useResume, + cliSessionIdToUse, + resolvedSessionId, + executionCommand, + executionLeadingArgv, + executionArgs: args, + env, + prompt, + argsPrompt, + stdin, + noOutputTimeoutMs, + outputMode, + logOutputText, + cliTurnStartedAt, + fallbackCleanup: fallbackClaudeSkillsPlugin?.cleanup, + claimFallbackCleanup: () => { + fallbackClaudeSkillsPluginCleanupOwned = fallbackClaudeSkillsPlugin !== undefined; + }, + observeForkSuccessor, + options, + }); + } catch (error) { + recordRunError(error); + } finally { + await toolTracking.finishDeliveryTracking({ + useManagedClaudeLiveSession, + recordRunError, + }); + toolTracking.finalizeCapture(events.finalizeParsedTools); + try { + await cleanupMcpCaptureAttempt?.(); + } catch (error) { + recordRunError(error); + } + try { + restoreSkillEnv?.(); + } catch (error) { + recordRunError(error); + } + } + if (runFailed) { + throw toolTracking.attachDeliveryEvidence(runError); + } + if (!runOutput) { + throw new Error("CLI run completed without output"); + } + return toolTracking.withExecutionEvidence(runOutput); + }; try { completedOutput = await enqueueCliRun(queueKey, async () => { if (params.lifecycleGeneration) { assertAgentRunLifecycleGenerationCurrent(params.lifecycleGeneration); } - claudeModelCallDiagnostics?.emitStarted(); + diagnostics?.emitStarted(); if (params.forkCliSessionOnResume && useResume) { if (!params.persistCliSessionForkSuccessor) { throw new Error("CLI session fork successor persistence is unavailable"); @@ -657,1566 +528,10 @@ export async function executePreparedCliRun( throw new Error("CLI session fork marker is no longer available"); } // The fork argument only applies at process startup; a cached warm child - // would run this turn inside the source session. Force a fresh spawn. + // would run inside the source session. Force a fresh spawn. await closeClaudeLiveSessionForContext(context); } - await context.preparedBackend.beforeExecution?.(); - const cliTurnStartedAt = Date.now(); - const restoreSkillEnv = params.skillsSnapshot - ? applySkillEnvOverridesFromSnapshot({ - snapshot: params.skillsSnapshot, - config: params.config, - }) - : undefined; - let gatewayCaptureKey: string | undefined; - let cleanupMcpCaptureAttempt: (() => Promise) | undefined; - let yielded = false; - let didSendViaMessagingTool = false; - let didDeliverSourceReplyViaMessageTool = false; - let inFlightUnclassifiedMcpRequests = 0; - let inFlightMessagingToolCalls = 0; - const inFlightPreparedMessagingCalls = new Set(); - const pendingMessagingCalls = new Map< - string, - { toolName: string; args: Record; target?: MessagingToolSend } - >(); - type CliToolTerminalOutcome = McpLoopbackToolCallTerminalOutcome | { outcome: "completed" }; - type CliLoopbackAmbiguityGroup = { - calls: Set; - activeToolCallIds: Set; - }; - type CliLoopbackCall = { - admitted: McpLoopbackToolCallStart; - current: McpLoopbackToolCallStart; - boundToolCallId?: string; - outcome?: CliToolTerminalOutcome; - ambiguous: boolean; - ambiguityGroup?: CliLoopbackAmbiguityGroup; - }; - type ActiveCliTool = { - toolName: string; - args: Record; - loopbackCall?: CliLoopbackCall; - loopbackAmbiguous: boolean; - ambiguityGroup?: CliLoopbackAmbiguityGroup; - }; - const cliLoopbackCalls: CliLoopbackCall[] = []; - const activeCliTools = new Map(); - let cliLoopbackCorrelationOverflowed = false; - const matchesCliLoopbackCall = ( - toolName: string, - toolArgs: Record, - call: McpLoopbackToolCallStart, - ) => - normalizeCliMessagingToolName(toolName) === call.toolName && - isDeepStrictEqual(toolArgs, call.args); - const markCliLoopbackCallsAmbiguous = ( - calls: CliLoopbackCall[], - activeEntries = Array.from(activeCliTools.entries()).filter( - ([, activeTool]) => - activeTool.loopbackCall !== undefined && calls.includes(activeTool.loopbackCall), - ), - ) => { - const groups = new Set(); - for (const call of calls) { - if (call.ambiguityGroup) { - groups.add(call.ambiguityGroup); - } - } - for (const [, activeTool] of activeEntries) { - if (activeTool.ambiguityGroup) { - groups.add(activeTool.ambiguityGroup); - } - } - const group = groups.values().next().value ?? { - calls: new Set(), - activeToolCallIds: new Set(), - }; - for (const existing of groups) { - if (existing === group) { - continue; - } - for (const call of existing.calls) { - call.ambiguityGroup = group; - group.calls.add(call); - } - for (const toolCallId of existing.activeToolCallIds) { - const activeTool = activeCliTools.get(toolCallId); - if (activeTool) { - activeTool.ambiguityGroup = group; - group.activeToolCallIds.add(toolCallId); - } - } - existing.calls.clear(); - existing.activeToolCallIds.clear(); - } - for (const call of calls) { - call.ambiguous = true; - call.ambiguityGroup = group; - group.calls.add(call); - } - for (const [toolCallId, activeTool] of activeEntries) { - activeTool.loopbackAmbiguous = true; - activeTool.ambiguityGroup = group; - group.activeToolCallIds.add(toolCallId); - } - }; - const markCliLoopbackSignatureAmbiguous = (call: McpLoopbackToolCallStart) => { - const calls = cliLoopbackCalls.filter((candidate) => - matchesCliLoopbackCall(call.toolName, call.args, candidate.admitted), - ); - const activeEntries = Array.from(activeCliTools.entries()).filter(([, activeTool]) => - matchesCliLoopbackCall(activeTool.toolName, activeTool.args, call), - ); - markCliLoopbackCallsAmbiguous(calls, activeEntries); - }; - const retainCliLoopbackCall = (call: McpLoopbackToolCallStart) => { - if (cliLoopbackCalls.length >= CLI_LOOPBACK_CORRELATION_MAX_CALLS) { - cliLoopbackCorrelationOverflowed = true; - for (const activeTool of activeCliTools.values()) { - if (activeTool.loopbackCall || activeTool.toolName.startsWith("mcp__")) { - activeTool.loopbackAmbiguous = true; - } - } - cliLoopbackCalls.length = 0; - return undefined; - } - const retained: CliLoopbackCall = { - admitted: call, - current: call, - ambiguous: false, - }; - cliLoopbackCalls.push(retained); - return retained; - }; - const bindCliLoopbackCall = ( - call: CliLoopbackCall, - toolCallId: string, - activeTool: ActiveCliTool, - ) => { - call.boundToolCallId = toolCallId; - activeTool.loopbackCall = call; - activeTool.loopbackAmbiguous ||= call.ambiguous; - if (call.ambiguityGroup) { - activeTool.ambiguityGroup = call.ambiguityGroup; - call.ambiguityGroup.activeToolCallIds.add(toolCallId); - } - }; - const removeCliLoopbackCall = (call: CliLoopbackCall | undefined) => { - if (!call) { - return; - } - const index = cliLoopbackCalls.indexOf(call); - if (index >= 0) { - cliLoopbackCalls.splice(index, 1); - } - }; - const retireCliLoopbackCorrelation = ( - toolCallId: string, - activeTool: ActiveCliTool | undefined, - ) => { - removeCliLoopbackCall(activeTool?.loopbackCall); - const group = activeTool?.ambiguityGroup; - if (!group) { - return; - } - group.activeToolCallIds.delete(toolCallId); - const hasUnboundCall = Array.from(group.calls).some( - (call) => call.boundToolCallId === undefined && cliLoopbackCalls.includes(call), - ); - if (group.activeToolCallIds.size > 0 || hasUnboundCall) { - return; - } - // An ambiguous group owns unbound captures too. Retire the whole group - // once its parsed tools finish so stale calls cannot poison later tools. - for (const call of group.calls) { - removeCliLoopbackCall(call); - } - group.calls.clear(); - }; - const resolveCliLoopbackTerminalOutcome = (toolCallId: string) => { - const activeTool = activeCliTools.get(toolCallId); - if (activeTool?.loopbackAmbiguous) { - return { outcome: "unknown" } as const; - } - return activeTool?.loopbackCall?.outcome; - }; - const matchingActiveCliTools = ( - call: McpLoopbackToolCallStart, - ): Array<[string, ActiveCliTool]> => - Array.from(activeCliTools.entries()).filter(([, activeTool]) => - matchesCliLoopbackCall(activeTool.toolName, activeTool.args, call), - ); - const messagingToolSentTexts: string[] = []; - const messagingToolSentTextKeys = new Set(); - const messagingToolSentMediaUrls: string[] = []; - const messagingToolSentMediaUrlKeys = new Set(); - const messagingToolSentTargets: MessagingToolSend[] = []; - const messagingToolSentTargetKeys = new Set(); - const messagingToolSourceReplyPayloads: MessagingToolSourceReplyPayload[] = []; - const isPreparedInternalSourceReply = async (call: McpLoopbackToolCallStart) => { - if ( - context.params.sourceReplyDeliveryMode !== "message_tool_only" || - normalizeCliMessagingToolName(call.toolName) !== "message" || - call.args.action !== "send" || - !context.params.config - ) { - return false; - } - return await shouldUseInternalSourceReplySink( - { - cfg: context.params.config, - action: "send", - sessionKey: context.params.sessionKey, - sourceReplyDeliveryMode: context.params.sourceReplyDeliveryMode, - toolContext: { - currentChannelProvider: - context.params.messageChannel ?? context.params.messageProvider, - currentChannelId: context.params.currentChannelId, - currentThreadTs: context.params.currentThreadTs, - currentMessageId: context.params.currentMessageId, - }, - }, - call.args, - ); - }; - let runOutput: CliOutput | undefined; - let runError: unknown; - let runFailed = false; - const recordRunError = (error: unknown) => { - if (runFailed) { - return; - } - runFailed = true; - runError = error; - }; - const withExecutionEvidence = (output: CliOutput): CliOutput => { - return { - ...output, - ...(yielded ? { yielded: true as const } : {}), - ...(didSendViaMessagingTool ? { didSendViaMessagingTool: true } : {}), - ...(didDeliverSourceReplyViaMessageTool - ? { didDeliverSourceReplyViaMessageTool: true } - : {}), - ...(messagingToolSentTexts.length > 0 - ? { messagingToolSentTexts: messagingToolSentTexts.slice() } - : {}), - ...(messagingToolSentMediaUrls.length > 0 - ? { messagingToolSentMediaUrls: messagingToolSentMediaUrls.slice() } - : {}), - ...(messagingToolSentTargets.length > 0 - ? { messagingToolSentTargets: messagingToolSentTargets.slice() } - : {}), - ...(messagingToolSourceReplyPayloads.length > 0 - ? { messagingToolSourceReplyPayloads: messagingToolSourceReplyPayloads.slice() } - : {}), - }; - }; - let finalizeParsedTools = () => {}; - try { - cliBackendLog.info( - buildCliExecLogLine({ - provider: params.provider, - model: context.normalizedModel, - promptChars: basePrompt.length, - trigger: params.trigger, - useResume, - cliSessionId: cliSessionIdToUse, - resolvedSessionId, - reusableSession: context.reusableCliSession, - hasHistoryPrompt: Boolean(context.openClawHistoryPrompt), - }), - ); - const logOutputText = - isTruthyEnvValue(process.env[CLI_BACKEND_LOG_OUTPUT_ENV]) || - isTruthyEnvValue(process.env[LEGACY_CLAUDE_CLI_LOG_OUTPUT_ENV]); - const outputMode = useResume ? (backend.resumeOutput ?? backend.output) : backend.output; - const hasJsonlOutput = outputMode === "jsonl"; - const initialGatewayCaptureKey = - useManagedClaudeLiveSession || nodePlacement ? undefined : buildCliMcpCaptureKey(context); - const mcpCaptureAttempt = nodePlacement - ? { env: {}, cleanup: undefined } - : await prepareCliBundleMcpCaptureAttempt({ - mode: context.backendResolved.bundleMcpMode, - backend, - env: context.preparedBackend.env, - captureKey: initialGatewayCaptureKey, - }); - cleanupMcpCaptureAttempt = mcpCaptureAttempt.cleanup; - const preparedBackendEnv = context.preparedBackend.env ?? {}; - const hasSelectedClaudeAuth = - Boolean(context.preparedBackend.secretInput) || - [...CLAUDE_SELECTED_AUTH_ENV_KEYS].some((key) => Object.hasOwn(preparedBackendEnv, key)); - const selectedClaudeClearEnv = hasSelectedClaudeAuth - ? new Set(backend.clearEnv ?? []) - : undefined; - const configuredBackendEnv = Object.fromEntries( - Object.entries(backend.env ?? {}).filter(([key]) => !selectedClaudeClearEnv?.has(key)), - ); - const backendEnv = { ...configuredBackendEnv, ...preparedBackendEnv }; - const nodeEnvEntries = Object.entries(preparedBackendEnv).filter(([key]) => - NODE_CLAUDE_FORWARD_ENV_KEYS.has(key), - ); - const nodeEnv = (() => { - if (!nodePlacement) { - return undefined; - } - const forwarded = { - ...Object.fromEntries(nodeEnvEntries), - ...resolveNodeClaudeAuthEnv(context), - }; - return Object.keys(forwarded).length > 0 ? forwarded : undefined; - })(); - const env = (() => { - const next = sanitizeHostExecEnv({ - baseEnv: process.env, - blockPathOverrides: true, - }); - const preservedEnv = parseCliBackendPreserveEnv(process.env[CLI_BACKEND_PRESERVE_ENV]); - for (const key of backend.clearEnv ?? []) { - if (preservedEnv.has(key) && !selectedClaudeClearEnv?.has(key)) { - continue; - } - delete next[key]; - } - if (Object.keys(backendEnv).length > 0) { - Object.assign( - next, - sanitizeHostExecEnv({ - baseEnv: {}, - overrides: backendEnv, - blockPathOverrides: true, - }), - ); - } - Object.assign(next, mcpCaptureAttempt.env); - - // Never mark Claude CLI as host-managed. That marker routes runs into - // Anthropic's separate host-managed usage tier instead of normal CLI - // subscription behavior. - delete next["CLAUDE_CODE_PROVIDER_MANAGED_BY_HOST"]; - - return next; - })(); - let executionCommand = backend.command; - let executionLeadingArgv: readonly string[] = []; - let executionArgs = args; - context.runtimeOwnerFingerprint = undefined; - context.runtimeArtifactFingerprint = undefined; - if (params.onSuccessfulAuthBinding && !nodePlacement) { - const executableIdentity = await resolveCliExecutableIdentity({ - command: backend.command, - cwd: context.cwd ?? context.workspaceDir, - env, - ...(context.backendResolved.runtimeArtifact - ? { runtimeArtifact: context.backendResolved.runtimeArtifact } - : {}), - }); - if (!executableIdentity) { - throw new Error( - `CLI backend ${context.backendResolved.id} executable cannot be bound to one durable absolute owner`, - ); - } - executionCommand = executableIdentity.invocation.command; - executionLeadingArgv = executableIdentity.invocation.leadingArgv; - executionArgs = args; - context.runtimeArtifactFingerprint = fingerprintCliRuntimeArtifact({ - provider: params.provider, - backendId: context.backendResolved.id, - executableIdentity, - }); - if (!context.authBindingFingerprint) { - context.runtimeOwnerFingerprint = await resolveCliRuntimeOwnerFingerprint({ - provider: params.provider, - config: params.config ?? context.contextEngineConfig, - ...(context.agentDir ? { agentDir: context.agentDir } : {}), - agentId: params.agentId, - runtimeOwnerId: context.backendResolved.id, - ...(context.effectiveAuthProfileId - ? { authProfileId: context.effectiveAuthProfileId } - : {}), - ...(context.authBindingSkipsLocalCredential ? { skipLocalCredential: true } : {}), - runtimeArtifactFingerprint: context.runtimeArtifactFingerprint, - }); - } - } - if (logOutputText) { - const logArgs = buildCliLogArgs({ - args: executionArgs, - systemPromptArg: backend.systemPromptArg, - modelArg: backend.modelArg, - imageArg: backend.imageArg, - argsPrompt, - }); - cliBackendLog.info(`cli argv: ${executionCommand} ${logArgs.join(" ")}`); - cliBackendLog.info(`cli env auth: ${buildCliEnvAuthLog(env)}`); - if (env.OPENCLAW_MCP_TOKEN) { - cliBackendLog.info(`cli env mcp: ${buildCliEnvMcpLog(env)}`); - } - } - - const runTimeoutOverrideMs = resolveCliRunTimeoutOverrideMs({ - config: params.config, - lane: params.lane, - timeoutMs: params.timeoutMs, - runTimeoutOverrideMs: params.runTimeoutOverrideMs, - }); - const noOutputTimeoutMs = resolveCliNoOutputTimeoutMs({ - backend, - timeoutMs: params.timeoutMs, - runTimeoutOverrideMs, - useResume, - trigger: params.trigger, - }); - const commitMessagingToolResult = (paramsLocal: { - toolName: string; - target?: MessagingToolSend; - args?: Record; - result?: unknown; - isError?: boolean; - }) => { - if (!isDeliveredMessagingToolResult(paramsLocal)) { - return; - } - didSendViaMessagingTool = true; - const toolArgs = paramsLocal.args ?? {}; - const isMessagingSend = isMessagingToolSendAction(paramsLocal.toolName, toolArgs); - const content = isMessagingSend - ? extractCliMessagingContent(toolArgs, paramsLocal.result) - : {}; - if (isMessagingSend) { - appendUniqueCliMessagingEvidence( - messagingToolSentTexts, - messagingToolSentTextKeys, - content.text ? [content.text] : [], - ); - appendUniqueCliMessagingEvidence( - messagingToolSentMediaUrls, - messagingToolSentMediaUrlKeys, - content.mediaUrls ?? [], - ); - if ( - isDeliveredMessageToolOnlySourceReplyResult({ - sourceReplyDeliveryMode: context.params.sourceReplyDeliveryMode, - toolName: paramsLocal.toolName, - args: paramsLocal.args, - result: paramsLocal.result, - isError: paramsLocal.isError, - }) - ) { - didDeliverSourceReplyViaMessageTool = true; - const sourceReplyPayload = extractMessagingToolSourceReplyPayload(paramsLocal.result); - if (sourceReplyPayload) { - if (messagingToolSourceReplyPayloads.length >= CLI_MESSAGING_EVIDENCE_MAX_CALLS) { - messagingToolSourceReplyPayloads.shift(); - } - // Each internal source-reply send is a distinct delivery, even when - // two intentional sends have identical text or media. - messagingToolSourceReplyPayloads.push(sourceReplyPayload); - } - } - } - if (paramsLocal.target) { - const confirmedTarget = extractMessagingToolSendResult( - paramsLocal.target, - paramsLocal.result, - ); - const targetWithContent = { - ...confirmedTarget, - ...content, - }; - const evidenceKey = buildMessagingToolSendEvidenceKey(targetWithContent); - if (messagingToolSentTargetKeys.has(evidenceKey)) { - return; - } - if (messagingToolSentTargets.length >= CLI_MESSAGING_EVIDENCE_MAX_CALLS) { - const removed = messagingToolSentTargets.shift(); - if (removed) { - messagingToolSentTargetKeys.delete(buildMessagingToolSendEvidenceKey(removed)); - } - } - messagingToolSentTargets.push(targetWithContent); - messagingToolSentTargetKeys.add(evidenceKey); - } - }; - const beginGatewayCapture = (captureKey: string | undefined) => { - if (!captureKey) { - return; - } - if (gatewayCaptureKey === captureKey) { - return; - } - if (gatewayCaptureKey) { - throw new Error("CLI MCP capture key changed during an active attempt"); - } - context.preparedBackend.mcpClientGrantCapture?.activate(captureKey); - gatewayCaptureKey = captureKey; - const isAdmittedPotentialMessagingDelivery = (toolName: string) => { - return isMessagingTool(normalizeCliMessagingToolName(toolName)); - }; - const isPreparedMessagingDelivery = ( - toolName: string, - toolArgs: Record, - ) => { - return ( - toolArgs.dryRun !== true && - isMessagingToolDeliveryAction(normalizeCliMessagingToolName(toolName), toolArgs) - ); - }; - beginMcpLoopbackToolCallCapture({ - captureKey: gatewayCaptureKey, - onYield: () => { - yielded = true; - }, - onRequestStart: () => { - inFlightUnclassifiedMcpRequests += 1; - }, - onRequestClassified: () => { - inFlightUnclassifiedMcpRequests = Math.max(0, inFlightUnclassifiedMcpRequests - 1); - }, - onToolCallStart: (call) => { - const retained = retainCliLoopbackCall(call); - const candidates = matchingActiveCliTools(call); - // Parallel same-name calls can reach the loopback out of stream - // order. Bind only a unique name+arguments match; ambiguity is - // safer than assigning a trusted terminal outcome to the wrong call. - let matched = - retained && - candidates.length === 1 && - !candidates[0]?.[1].loopbackCall && - !candidates[0]?.[1].loopbackAmbiguous - ? candidates[0] - : undefined; - if (retained && matched) { - bindCliLoopbackCall(retained, matched[0], matched[1]); - } else if (retained && candidates.length > 0) { - markCliLoopbackSignatureAmbiguous(call); - // The exact identity is unknowable, but pairing an unmatched - // peer keeps the ambiguity group's lifetime count complete. - matched = candidates.find(([, activeTool]) => !activeTool.loopbackCall); - if (matched) { - bindCliLoopbackCall(retained, matched[0], matched[1]); - } - } - if (isAdmittedPotentialMessagingDelivery(call.toolName)) { - inFlightMessagingToolCalls += 1; - } - return matched?.[0]; - }, - onToolCallUpdate: ({ previous, current }) => { - const candidates = cliLoopbackCalls.filter((candidate) => - matchesCliLoopbackCall(previous.toolName, previous.args, candidate.current), - ); - const candidate = candidates.at(0); - if (candidates.length === 1 && candidate && !candidate.ambiguous) { - candidate.current = current; - } else if (candidates.length > 0) { - markCliLoopbackCallsAmbiguous(candidates); - } - inFlightPreparedMessagingCalls.delete(previous); - const wasMessagingSend = isAdmittedPotentialMessagingDelivery(previous.toolName); - const isMessagingSend = isPreparedMessagingDelivery(current.toolName, current.args); - if (wasMessagingSend !== isMessagingSend) { - inFlightMessagingToolCalls = Math.max( - 0, - inFlightMessagingToolCalls + (isMessagingSend ? 1 : -1), - ); - } - if (isMessagingSend) { - inFlightPreparedMessagingCalls.add(current); - } - }, - onToolCallFinish: (call, { prepared }) => { - const isMessagingSend = prepared - ? isPreparedMessagingDelivery(call.toolName, call.args) - : isAdmittedPotentialMessagingDelivery(call.toolName); - if (isMessagingSend) { - inFlightMessagingToolCalls = Math.max(0, inFlightMessagingToolCalls - 1); - } - inFlightPreparedMessagingCalls.delete(call); - }, - onToolCallResult: (call) => { - const terminalOutcome: CliToolTerminalOutcome = - call.outcome === "blocked" - ? { outcome: call.outcome, deniedReason: call.deniedReason } - : { outcome: call.outcome }; - const correlated = call.correlationId - ? cliLoopbackCalls.find( - (candidate) => candidate.boundToolCallId === call.correlationId, - ) - : undefined; - const candidates = correlated - ? [correlated] - : cliLoopbackCalls.filter((candidate) => - matchesCliLoopbackCall(call.toolName, call.args, candidate.current), - ); - if (candidates.length === 1 && candidates[0]) { - candidates[0].outcome = terminalOutcome; - } else if (candidates.length > 1) { - markCliLoopbackCallsAmbiguous(candidates); - } - const normalizedToolName = normalizeCliMessagingToolName(call.toolName); - if (!isMessagingToolDeliveryAction(normalizedToolName, call.args)) { - return; - } - commitMessagingToolResult({ - toolName: normalizedToolName, - target: extractCliMessagingTarget(context, normalizedToolName, call.args), - args: call.args, - result: "result" in call ? call.result : undefined, - isError: call.outcome !== "completed", - }); - }, - }); - }; - beginGatewayCapture(initialGatewayCaptureKey); - let observedCliActivity = false; - let signaledToolExecutionStarted = false; - let signaledAssistantOutputStarted = false; - const emitLiveEvents = params.executionMode !== "side-question"; - const activeParsedTools = new Map< - string, - { startedAt: number; toolName: string; kind: CliToolUseStartDelta["kind"] } - >(); - const emitCliToolUseStart = (event: CliToolUseStartDelta) => { - observedCliActivity = true; - if (!signaledToolExecutionStarted) { - signaledToolExecutionStarted = true; - params.onExecutionPhase?.({ - phase: "tool_execution_started", - provider: params.provider, - model: context.modelId, - backend: context.backendResolved.id, - }); - } - // Server-native calls have their own result stream and must never inherit MCP outcomes. - if (event.kind !== "server_tool_use") { - const activeTool = { - toolName: event.name, - args: event.args, - loopbackAmbiguous: cliLoopbackCorrelationOverflowed && event.name.startsWith("mcp__"), - }; - activeCliTools.set(event.toolCallId, activeTool); - const admittedCall = { - toolName: normalizeCliMessagingToolName(event.name), - args: event.args, - }; - const pendingCandidates = cliLoopbackCalls.filter( - (candidate) => - candidate.boundToolCallId === undefined && - matchesCliLoopbackCall(event.name, event.args, candidate.admitted), - ); - const hasAssociatedPeer = matchingActiveCliTools(admittedCall).some( - ([toolCallId, peer]) => - toolCallId !== event.toolCallId && - (peer.loopbackCall !== undefined || peer.loopbackAmbiguous), - ); - const pending = pendingCandidates[0]; - if (hasAssociatedPeer || pendingCandidates.length > 1 || pending?.ambiguous) { - markCliLoopbackSignatureAmbiguous(admittedCall); - if (pending) { - bindCliLoopbackCall(pending, event.toolCallId, activeTool); - } - } else if (pendingCandidates.length === 1 && pending) { - bindCliLoopbackCall(pending, event.toolCallId, activeTool); - } - } - const toolName = normalizeCliMessagingToolName(event.name); - if ( - event.kind !== "server_tool_use" && - !gatewayCaptureKey && - event.args.dryRun !== true && - isMessagingToolDeliveryAction(toolName, event.args) - ) { - if (pendingMessagingCalls.size >= CLI_MESSAGING_EVIDENCE_MAX_CALLS) { - const oldestToolCallId = pendingMessagingCalls.keys().next().value; - if (oldestToolCallId !== undefined) { - pendingMessagingCalls.delete(oldestToolCallId); - // Once an unresolved send is evicted, its later result cannot be - // correlated. Fail closed so a failed turn cannot duplicate it. - didSendViaMessagingTool = true; - } - } - pendingMessagingCalls.set(event.toolCallId, { - toolName, - args: event.args, - target: extractCliMessagingTarget(context, toolName, event.args), - }); - } - if (!emitLiveEvents) { - return; - } - emitAgentEvent({ - runId: params.runId, - stream: "tool", - data: { - phase: "start", - name: event.name, - toolCallId: event.toolCallId, - args: sanitizeToolArgs(event.args), - }, - }); - }; - const emitCliToolResult = (event: { - toolCallId: string; - name: string; - isError: boolean; - result?: unknown; - }) => { - observedCliActivity = true; - const activeTool = activeCliTools.get(event.toolCallId); - activeCliTools.delete(event.toolCallId); - retireCliLoopbackCorrelation(event.toolCallId, activeTool); - const pending = pendingMessagingCalls.get(event.toolCallId); - if (pending) { - pendingMessagingCalls.delete(event.toolCallId); - commitMessagingToolResult({ - toolName: pending.toolName, - target: pending.target, - args: pending.args, - result: event.result, - isError: event.isError, - }); - } - if (!emitLiveEvents) { - return; - } - emitAgentEvent({ - runId: params.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, { - startedAt, - toolName: event.name, - kind: event.kind, - }); - emitTrustedDiagnosticEvent({ - type: "tool.execution.started", - runId: params.runId, - sessionId: params.sessionId, - ...(params.sessionKey ? { sessionKey: params.sessionKey } : {}), - ...(params.agentId ? { agentId: params.agentId } : {}), - toolName: event.name, - toolSource: event.name.startsWith("mcp__") ? "mcp" : "core", - toolOwner: "cli-runner", - toolCallId: event.toolCallId, - }); - emitCliToolUseStart(event); - }; - const emitParsedToolTerminal = (event: { - toolCallId: string; - name: string; - isError: boolean; - incomplete?: boolean; - }) => { - const activeTool = activeParsedTools.get(event.toolCallId); - activeParsedTools.delete(event.toolCallId); - const trustedOutcome = resolveCliLoopbackTerminalOutcome(event.toolCallId); - const toolName = activeTool?.toolName ?? event.name; - const now = Date.now(); - const trustedTerminalReason = - trustedOutcome && - trustedOutcome.outcome !== "blocked" && - trustedOutcome.outcome !== "completed" && - trustedOutcome.outcome !== "unknown" - ? trustedOutcome.outcome - : undefined; - const terminalReason = - trustedTerminalReason ?? - resolveCliToolTerminalReason({ - error: event.incomplete ? runError : undefined, - abortSignal: params.abortSignal, - }); - // Incomplete client/MCP tools inherit the enclosing failed run even when - // the loopback disconnect is ambiguous. Server-native tools do not. - const useEnclosingTerminalReason = - event.incomplete && - runFailed && - activeTool !== undefined && - activeTool.kind !== "server_tool_use"; - const diagnosticBase = { - runId: params.runId, - sessionId: params.sessionId, - ...(params.sessionKey ? { sessionKey: params.sessionKey } : {}), - ...(params.agentId ? { agentId: params.agentId } : {}), - toolName, - toolSource: toolName.startsWith("mcp__") ? ("mcp" as const) : ("core" as const), - toolOwner: "cli-runner", - toolCallId: event.toolCallId, - durationMs: Math.max(0, now - (activeTool?.startedAt ?? now)), - }; - if (trustedOutcome?.outcome === "unknown" && !useEnclosingTerminalReason) { - emitTrustedDiagnosticEvent({ - type: "tool.execution.error", - ...diagnosticBase, - errorCategory: "cli_tool_ambiguous", - errorCode: "tool_outcome_unknown", - }); - return; - } - if ( - event.incomplete && - activeTool?.kind === "server_tool_use" && - trustedOutcome === undefined - ) { - emitTrustedDiagnosticEvent({ - type: "tool.execution.error", - ...diagnosticBase, - errorCategory: "cli_tool_ambiguous", - errorCode: "tool_outcome_unknown", - }); - return; - } - const trustedFailure = - trustedOutcome !== undefined && trustedOutcome.outcome !== "completed"; - emitTrustedDiagnosticEvent( - trustedOutcome?.outcome === "blocked" - ? { - type: "tool.execution.blocked", - ...diagnosticBase, - deniedReason: trustedOutcome.deniedReason, - reason: "blocked by before-tool policy", - } - : trustedFailure || (trustedOutcome === undefined && event.isError) - ? { - type: "tool.execution.error", - ...diagnosticBase, - errorCategory: - terminalReason === "cancelled" - ? "aborted" - : event.incomplete && (!trustedOutcome || useEnclosingTerminalReason) - ? "cli_tool_incomplete" - : "cli_tool", - terminalReason, - } - : { - type: "tool.execution.completed", - ...diagnosticBase, - }, - ); - }; - const emitParsedToolResult = (event: { - toolCallId: string; - name: string; - isError: boolean; - result?: unknown; - }) => { - emitParsedToolTerminal(event); - emitCliToolResult(event); - }; - finalizeParsedTools = () => { - for (const [toolCallId, activeTool] of Array.from(activeParsedTools)) { - emitParsedToolTerminal({ - toolCallId, - name: activeTool.toolName, - isError: true, - incomplete: true, - }); - } - }; - let commentaryCounter = 0; - const emitCliCommentaryText = (text: string) => { - if (!emitLiveEvents) { - return; - } - commentaryCounter += 1; - const transformedText = applyPluginTextReplacements( - text, - context.backendResolved.textTransforms?.output, - ); - emitAgentEvent({ - runId: params.runId, - stream: "item", - data: { - kind: "preamble", - itemId: `commentary-${params.runId}-${commentaryCounter}`, - phase: "update", - title: "commentary", - status: "running", - progressText: transformedText, - }, - }); - }; - const emitCliAssistantDelta = ({ text, delta }: CliStreamingDelta) => { - if (text || delta) { - observedCliActivity = true; - if (!signaledAssistantOutputStarted) { - signaledAssistantOutputStarted = true; - params.onExecutionPhase?.({ - phase: "assistant_output_started", - provider: params.provider, - model: context.modelId, - backend: context.backendResolved.id, - }); - } - } - if (!emitLiveEvents) { - return; - } - emitAgentEvent({ - runId: params.runId, - stream: "assistant", - data: { - text: applyPluginTextReplacements( - text, - context.backendResolved.textTransforms?.output, - ), - delta: applyPluginTextReplacements( - delta, - context.backendResolved.textTransforms?.output, - ), - }, - }); - }; - // Emit-always: thinking reaches the agent-event bus and session archive - // like the embedded reasoning stream; /reasoning and /verbose gate only - // presentation. Text stays raw here to match the thinking-stream contract - // shared with embedded-agent-subscribe, which archives untransformed - // reasoning regardless of source. - const emitCliThinkingDelta = ({ text, delta, isReasoningSnapshot }: CliThinkingDelta) => { - if (text || delta) { - observedCliActivity = true; - } - if (!emitLiveEvents) { - return; - } - emitAgentEvent({ - runId: params.runId, - stream: "thinking", - data: { text, delta, ...(isReasoningSnapshot ? { isReasoningSnapshot } : {}) }, - }); - }; - const emitCliThinkingProgress = ({ progressTokens }: CliThinkingProgress) => { - observedCliActivity = true; - if (!emitLiveEvents) { - return; - } - emitAgentEvent({ - runId: params.runId, - stream: "thinking", - data: { progressTokens }, - }); - }; - const emitCliPlanUpdate = ({ steps }: CliPlanUpdate) => { - observedCliActivity = true; - if (!emitLiveEvents) { - return; - } - emitAgentEvent({ - runId: params.runId, - stream: "plan", - data: { - phase: "update", - title: "Plan updated", - source: "codex-exec", - steps, - }, - }); - }; - if (useManagedClaudeLiveSession) { - if (!hasJsonlOutput) { - throw new Error("Claude live session requires JSONL streaming parser"); - } - params.onExecutionPhase?.({ - phase: "process_spawned", - provider: params.provider, - model: context.modelId, - backend: context.backendResolved.id, - }); - fallbackClaudeSkillsPluginCleanupOwned = fallbackClaudeSkillsPlugin !== undefined; - const liveResult = await runClaudeLiveSessionTurn({ - context, - args: executionArgs, - executableCommand: executionCommand, - executableLeadingArgv: executionLeadingArgv, - env, - prompt, - useResume, - forceNewSession: - cliSessionIdToUse === undefined && context.openClawHistoryPrompt !== undefined, - requiredSessionGeneration: cliSessionIdToUse - ? context.requiredClaudeLiveSessionGeneration - : undefined, - noOutputTimeoutMs, - getProcessSupervisor: executeDeps.getProcessSupervisor, - onAssistantDelta: emitCliAssistantDelta, - onThinkingDelta: emitCliThinkingDelta, - onThinkingProgress: emitCliThinkingProgress, - onToolUseStart: emitCliToolUseStart, - onToolResult: emitCliToolResult, - resolveToolResultTerminalOutcome: (event) => { - const outcome = resolveCliLoopbackTerminalOutcome(event.toolCallId); - return outcome?.outcome === "completed" ? undefined : outcome; - }, - onCommentaryText: - emitLiveEvents && context.params.emitCommentaryText - ? emitCliCommentaryText - : undefined, - onMcpCaptureReady: beginGatewayCapture, - cleanup: async () => { - await fallbackClaudeSkillsPlugin?.cleanup(); - }, - onSessionId: (sessionId) => { - observeForkSuccessor(sessionId); - }, - onAssistantMessage: claudeModelCallDiagnostics?.observeAssistantMessage, - onUsage: claudeModelCallDiagnostics?.observeUsage, - onCliOutput: claudeModelCallDiagnostics?.observeCliOutput, - onRequestPayload: claudeModelCallDiagnostics?.observeRequestPayload, - onPhase: options?.onPhase, - }); - options?.onPhase?.("resolve"); - const rawText = liveResult.output.text; - runOutput = { - ...liveResult.output, - rawText, - finalPromptText: prompt, - text: applyPluginTextReplacements( - rawText, - context.backendResolved.textTransforms?.output, - ), - }; - } else { - const streamingParser = hasJsonlOutput - ? createCliJsonlStreamingParser({ - backend, - providerId: context.backendResolved.id, - onAssistantDelta: emitCliAssistantDelta, - onThinkingDelta: emitCliThinkingDelta, - onThinkingProgress: emitCliThinkingProgress, - onPlanUpdate: emitCliPlanUpdate, - onToolUseStart: emitParsedToolUseStart, - onToolResult: emitParsedToolResult, - onCommentaryText: - emitLiveEvents && context.params.emitCommentaryText - ? emitCliCommentaryText - : undefined, - onSessionId: (sessionId) => { - observeForkSuccessor(sessionId); - }, - onAssistantMessage: claudeModelCallDiagnostics?.observeAssistantMessage, - onUsage: claudeModelCallDiagnostics?.observeUsage, - }) - : null; - let stdoutTail = ""; - let stdoutParseBuffer: Buffer = Buffer.alloc(0); - let stdoutBytes = 0; - const stdoutHash = crypto.createHash("sha256"); - let stdoutParseExceeded = false; - let stderrTail = ""; - let stderrParseBuffer: Buffer = Buffer.alloc(0); - let stderrBytes = 0; - const stderrHash = crypto.createHash("sha256"); - let stderrParseExceeded = false; - const consumeStdout = (chunk: string) => { - const chunkBytes = Buffer.byteLength(chunk); - claudeModelCallDiagnostics?.observeCliOutput(chunk, "stdout", chunkBytes); - stdoutBytes += chunkBytes; - stdoutHash.update(chunk); - stdoutTail = appendCliOutputTail(stdoutTail, chunk); - if (!stdoutParseExceeded) { - const nextStdoutParse = appendCliOutputParseBuffer(stdoutParseBuffer, chunk); - stdoutParseBuffer = nextStdoutParse.buffer; - stdoutParseExceeded = nextStdoutParse.exceeded; - } - streamingParser?.push(chunk); - }; - const consumeStderr = (chunk: string) => { - claudeModelCallDiagnostics?.observeCliOutput(chunk, "stderr"); - stderrBytes += Buffer.byteLength(chunk); - stderrHash.update(chunk); - stderrTail = appendCliOutputTail(stderrTail, chunk); - if (!stderrParseExceeded) { - const nextStderrParse = appendCliOutputParseBuffer(stderrParseBuffer, chunk); - stderrParseBuffer = nextStderrParse.buffer; - stderrParseExceeded = nextStderrParse.exceeded; - } - }; - - params.onExecutionPhase?.({ - phase: "process_spawned", - provider: params.provider, - model: context.modelId, - backend: context.backendResolved.id, - }); - let managedRunPid: number | undefined; - let nodeRunAbortSignal: AbortSignal | undefined; - let nodeRunTruncated = false; - let result: RunExit; - claudeModelCallDiagnostics?.observeRequestPayload(stdin ?? argsPrompt ?? ""); - if (nodePlacement) { - const nodeRun = await executeNodeClaudeRun({ - context, - nodePlacement, - executionArgs, - stdinPayload, - ...(nodeSystemPrompt !== undefined ? { nodeSystemPrompt } : {}), - ...(nodeEnv ? { nodeEnv } : {}), - ...(selectedClaudeClearEnv ? { nodeClearEnv: [...selectedClaudeClearEnv] } : {}), - noOutputTimeoutMs, - consumeStdout, - consumeStderr, - deps: executeDeps, - }); - result = nodeRun.result; - nodeRunAbortSignal = nodeRun.nodeRunAbortSignal; - nodeRunTruncated = nodeRun.nodeRunTruncated; - } else { - const supervisor = executeDeps.getProcessSupervisor(); - const scopeKey = buildCliSupervisorScopeKey({ - backend, - backendId: context.backendResolved.id, - cliSessionId: useResume ? resolvedSessionId : undefined, - }); - const managedRun = await supervisor.spawn({ - sessionId: params.sessionId, - backendId: context.backendResolved.id, - scopeKey, - replaceExistingScope: Boolean(useResume && scopeKey), - mode: "child", - argv: [executionCommand, ...executionLeadingArgv, ...executionArgs], - timeoutMs: params.timeoutMs, - noOutputTimeoutMs, - cwd: context.cwd ?? context.workspaceDir, - env, - input: stdinPayload, - secretInput: context.preparedBackend.secretInput, - captureOutput: false, - onStdout: consumeStdout, - onStderr: consumeStderr, - }); - managedRunPid = managedRun.pid; - let replyBackendCompleted = false; - const replyBackendHandle = params.replyOperation - ? { - kind: "cli" as const, - cancel: () => { - managedRun.cancel("manual-cancel"); - }, - isStreaming: () => !replyBackendCompleted, - } - : undefined; - if (replyBackendHandle) { - params.replyOperation?.attachBackend(replyBackendHandle); - } - const abortManagedRun = () => { - managedRun.cancel("manual-cancel"); - }; - params.abortSignal?.addEventListener("abort", abortManagedRun, { once: true }); - if (params.abortSignal?.aborted) { - abortManagedRun(); - } - try { - result = await managedRun.wait(); - } finally { - replyBackendCompleted = true; - if (replyBackendHandle) { - params.replyOperation?.detachBackend(replyBackendHandle); - } - params.abortSignal?.removeEventListener("abort", abortManagedRun); - } - } - if ( - (params.abortSignal?.aborted || nodeRunAbortSignal?.aborted) && - result.reason === "manual-cancel" - ) { - throw createCliAbortError(); - } - options?.onPhase?.("resolve"); - streamingParser?.finish(); - const streamingParserErrorText = - outputMode === "jsonl" ? (streamingParser?.getErrorText() ?? null) : null; - if (streamingParserErrorText) { - throw new FailoverError(streamingParserErrorText, { - reason: "format", - provider: params.provider, - model: context.modelId, - sessionId: params.sessionId, - lane: params.lane, - status: resolveFailoverStatus("format"), - }); - } - // The node re-injects the terminal result line when its output cap - // truncates the stream; if even that is missing, the turn outcome is - // unknowable and must not pass as a clean exit. - if ( - nodeRunTruncated && - result.exitCode === 0 && - !result.timedOut && - !streamingParser?.getOutput() - ) { - throw new FailoverError( - "paired node truncated the Claude CLI stream before the terminal result; refusing to accept partial output.", - { - reason: "format", - provider: params.provider, - model: context.modelId, - sessionId: params.sessionId, - lane: params.lane, - status: resolveFailoverStatus("format"), - }, - ); - } - - const stdout = stdoutParseBuffer.toString("utf8").trim(); - const stdoutDiagnostic = stdoutTail.trim(); - const stderr = stderrParseBuffer.toString("utf8").trim(); - const stderrDiagnostic = stderrTail.trim(); - const processDiagnostics = { - backendId: context.backendResolved.id, - processReason: result.reason, - exitCode: result.exitCode, - exitSignal: result.exitSignal, - durationMs: result.durationMs, - stdoutBytes, - stdoutHash: stdoutHash.digest("hex").slice(0, 12), - stderrBytes, - stderrHash: stderrHash.digest("hex").slice(0, 12), - useResume, - }; - if (logOutputText) { - if (stdoutDiagnostic) { - cliBackendLog.info(`cli stdout:\n${stdoutDiagnostic}`); - } - if (stderrDiagnostic) { - cliBackendLog.info(`cli stderr:\n${stderrDiagnostic}`); - } - } - if (shouldLogVerbose()) { - if (stdoutDiagnostic) { - cliBackendLog.debug(`cli stdout:\n${stdoutDiagnostic}`); - } - if (stderrDiagnostic) { - cliBackendLog.debug(`cli stderr:\n${stderrDiagnostic}`); - } - } - - const streamedJsonlOutput = - outputMode === "jsonl" ? (streamingParser?.getOutput() ?? null) : null; - const parsedStructuredOutput = - streamedJsonlOutput ?? - (outputMode === "json" && !stdoutParseExceeded - ? parseCliOutput({ - raw: stdout, - backend, - providerId: context.backendResolved.id, - outputMode, - fallbackSessionId: resolvedSessionId, - }) - : null); - // A completed terminal record is authoritative even if the CLI hangs - // afterward. Reclassifying it as a timeout could replay completed tools. - if (parsedStructuredOutput?.terminalFailure) { - const terminalError = createCliOutputFailoverError({ - output: parsedStructuredOutput, - provider: params.provider, - model: context.modelId, - runId: params.runId, - sessionId: params.sessionId, - lane: params.lane, - }); - if (terminalError) { - throw terminalError; - } - } - - if (result.exitCode !== 0 || result.reason !== "exit") { - options?.onPhase?.("send"); - if (result.reason === "no-output-timeout" || result.noOutputTimedOut) { - const timeoutReason = `CLI produced no output for ${Math.round(noOutputTimeoutMs / 1000)}s and was terminated.`; - cliBackendLog.warn( - `cli watchdog timeout: provider=${params.provider} model=${context.modelId} session=${resolvedSessionId ?? params.sessionId} noOutputTimeoutMs=${noOutputTimeoutMs} pid=${managedRunPid ?? "node"}`, - ); - const retryableNoOutputTimeout = - !observedCliActivity && - stdoutDiagnostic.length === 0 && - stderrDiagnostic.length === 0; - const deferWatchdogNoticeForFreshRetry = - retryableNoOutputTimeout && - Boolean(cliSessionIdToUse) && - Boolean(resolvedSessionId) && - Boolean(context.openClawHistoryPrompt) && - Boolean(params.sessionKey) && - params.timeoutMs - (Date.now() - context.started) > 0; - if (params.sessionKey && emitLiveEvents && !deferWatchdogNoticeForFreshRetry) { - const stallNotice = [ - `CLI agent (${params.provider}) produced no output for ${Math.round(noOutputTimeoutMs / 1000)}s and was terminated.`, - "It may have been waiting for interactive input or an approval prompt.", - // Node runs strip --permission-mode, so the local-flag hint - // would be unfollowable advice there. - ...(nodePlacement - ? ["Check the node's Claude permission settings for pending prompts."] - : ["For Claude Code, prefer --permission-mode bypassPermissions --print."]), - ].join(" "); - const eventRouting = resolveEventSessionRoutingPolicy({ - cfg: params.config, - sessionKey: params.sessionKey, - channel: params.messageProvider, - accountId: params.agentAccountId, - }); - executeDeps.enqueueSystemEvent(stallNotice, { - sessionKey: resolveEventSessionKeyForPolicy(params.sessionKey, eventRouting), - }); - executeDeps.requestHeartbeat( - scopedHeartbeatWakeOptionsForPolicy( - params.sessionKey, - { - source: "cli-watchdog", - intent: "event", - reason: "cli:watchdog:stall", - }, - eventRouting, - ), - ); - } - throw new FailoverError(timeoutReason, { - reason: "timeout", - provider: params.provider, - model: context.modelId, - sessionId: params.sessionId, - lane: params.lane, - status: resolveFailoverStatus("timeout"), - code: retryableNoOutputTimeout ? "cli_no_output_timeout" : undefined, - cliTimeout: { - mode: "no-output", - timeoutSeconds: Math.round(noOutputTimeoutMs / 1000), - observedActivity: observedCliActivity, - activeToolCount: activeParsedTools.size, - backgroundTaskCount: 0, - }, - }); - } - if (result.reason === "overall-timeout") { - const timeoutReason = `CLI exceeded timeout (${Math.round(params.timeoutMs / 1000)}s) and was terminated.`; - throw new FailoverError(timeoutReason, { - reason: "timeout", - provider: params.provider, - model: context.modelId, - sessionId: params.sessionId, - lane: params.lane, - status: resolveFailoverStatus("timeout"), - code: "cli_overall_timeout", - cliTimeout: { - mode: "overall", - timeoutSeconds: Math.round(params.timeoutMs / 1000), - observedActivity: observedCliActivity, - activeToolCount: activeParsedTools.size, - backgroundTaskCount: 0, - }, - }); - } - const errorCandidates = [stderr, stdout, stderrDiagnostic, stdoutDiagnostic].filter( - (candidate) => candidate.length > 0, - ); - const structuredError = - errorCandidates.map((candidate) => extractCliErrorMessage(candidate)).find(Boolean) ?? - null; - let classifiedErrorText = structuredError; - let reason = structuredError - ? classifyFailoverReason(structuredError, { provider: params.provider }) - : null; - if (!reason) { - for (const candidate of errorCandidates) { - reason = classifyFailoverReason(candidate, { provider: params.provider }); - if (reason) { - classifiedErrorText = candidate; - break; - } - } - } - const err = - structuredError || classifiedErrorText || errorCandidates[0] || "CLI failed."; - reason = reason ?? "unknown"; - const status = resolveFailoverStatus(reason); - const retryCode = - reason === "context_overflow" - ? "cli_context_overflow" - : reason === "unknown" && - result.reason === "exit" && - errorCandidates.length === 0 && - !observedCliActivity - ? "cli_unknown_empty_failure" - : undefined; - throw new FailoverError(err, { - reason, - provider: params.provider, - model: context.modelId, - sessionId: params.sessionId, - lane: params.lane, - status, - code: retryCode, - }); - } - - if (stdoutParseExceeded && !streamedJsonlOutput) { - throw new FailoverError( - `CLI stdout exceeded ${CLI_RUNNER_OUTPUT_PARSE_BYTES} bytes; refusing to parse truncated output.`, - { - reason: "format", - provider: params.provider, - model: context.modelId, - sessionId: params.sessionId, - lane: params.lane, - status: resolveFailoverStatus("format"), - }, - ); - } - - const parsed = - parsedStructuredOutput ?? - parseCliOutput({ - raw: stdout, - backend, - providerId: context.backendResolved.id, - outputMode, - fallbackSessionId: resolvedSessionId, - }); - const parsedError = createCliOutputFailoverError({ - output: parsed, - provider: params.provider, - model: context.modelId, - runId: params.runId, - sessionId: params.sessionId, - lane: params.lane, - }); - if (parsedError) { - throw parsedError; - } - const rawText = parsed.text; - cliBackendLog.info( - `cli turn: provider=${params.provider} model=${context.modelId} durationMs=${Date.now() - cliTurnStartedAt} ${formatCliBackendOutputDigest(rawText)}`, - ); - runOutput = { - ...parsed, - diagnostics: { - ...parsed.diagnostics, - process: processDiagnostics, - }, - rawText, - finalPromptText: prompt, - text: applyPluginTextReplacements( - rawText, - context.backendResolved.textTransforms?.output, - ), - }; - } - } catch (error) { - recordRunError(error); - } finally { - try { - if (!gatewayCaptureKey && pendingMessagingCalls.size > 0) { - const unresolvedJsonlMessagingCalls = Array.from(pendingMessagingCalls.values()); - const internalSourceReplyStates = await Promise.all( - unresolvedJsonlMessagingCalls.map(isPreparedInternalSourceReply), - ); - const hasPotentialVisibleSend = internalSourceReplyStates.some( - (isInternalSourceReply) => !isInternalSourceReply, - ); - if (hasPotentialVisibleSend) { - // A JSONL start without a result may have delivered before the CLI exited. - // Fail closed so retry/failover cannot duplicate a late visible send. - didSendViaMessagingTool = true; - recordRunError( - new Error("CLI JSONL message tool call remained unresolved after exit"), - ); - } else { - recordRunError( - new Error("CLI JSONL source reply call remained unresolved after exit"), - ); - } - } - if (gatewayCaptureKey) { - const captureBecameIdle = await waitForMcpLoopbackToolCallCaptureIdle( - gatewayCaptureKey, - { - timeoutMs: CLI_MCP_DELIVERY_DRAIN_GRACE_MS, - admissionGraceMs: CLI_MCP_REQUEST_ADMISSION_GRACE_MS, - }, - ); - if (!captureBecameIdle) { - if (useManagedClaudeLiveSession) { - await rotateClaudeLiveMcpCaptureKeyForContext(context); - } - const unresolvedPreparedMessagingCalls = Array.from(inFlightPreparedMessagingCalls); - const internalSourceReplyStates = await Promise.all( - unresolvedPreparedMessagingCalls.map(isPreparedInternalSourceReply), - ); - const internalSourceReplyCount = internalSourceReplyStates.filter(Boolean).length; - const hasPotentialVisibleSend = inFlightMessagingToolCalls > internalSourceReplyCount; - if (inFlightUnclassifiedMcpRequests > 0 || hasPotentialVisibleSend) { - // An admitted request or send may complete after its CLI process exits. - // Fail closed so retry/failover cannot duplicate a late visible send. - didSendViaMessagingTool = true; - recordRunError(new Error("CLI message tool call remained in flight after exit")); - } else if (inFlightMessagingToolCalls > 0) { - // Internal source replies are only result payloads; they have no external - // side effect, so keep the failed turn retryable instead of dropping them. - recordRunError(new Error("CLI source reply call remained in flight after exit")); - } - } - } - } catch (error) { - if ( - pendingMessagingCalls.size > 0 || - inFlightUnclassifiedMcpRequests > 0 || - inFlightMessagingToolCalls > 0 - ) { - // A failed drain/classification cannot prove an admitted messaging request harmless. - didSendViaMessagingTool = true; - } - recordRunError(error); - } finally { - // Captured MCP calls may settle after the CLI process exits. Drain - // first so finalization can use their trusted terminal outcomes. - try { - finalizeParsedTools(); - } finally { - if (gatewayCaptureKey) { - // Drain accepted work, then fence this exact grant generation before - // clearing observers; otherwise a late request escapes accounting. - try { - context.preparedBackend.mcpClientGrantCapture?.deactivate(gatewayCaptureKey); - } finally { - clearMcpLoopbackToolCallCapture(gatewayCaptureKey); - } - } - } - } - try { - await cleanupMcpCaptureAttempt?.(); - } catch (error) { - recordRunError(error); - } - try { - restoreSkillEnv?.(); - } catch (error) { - recordRunError(error); - } - } - if (runFailed) { - throw attachCliMessagingDeliveryEvidence(runError, { - didSendViaMessagingTool, - didDeliverSourceReplyViaMessageTool, - messagingToolSentTexts, - messagingToolSentMediaUrls, - messagingToolSentTargets, - messagingToolSourceReplyPayloads, - }); - } - if (!runOutput) { - throw new Error("CLI run completed without output"); - } - return withExecutionEvidence(runOutput); + return await executeAttempt(); }); if (completedOutput.sessionId) { observeForkSuccessor(completedOutput.sessionId); @@ -2229,7 +544,7 @@ export async function executePreparedCliRun( } } catch (error) { executionError = error; - claudeModelCallDiagnostics?.emitError(error); + diagnostics?.emitError(error); let failure = error; try { await finishForkSuccessorPersistence(); @@ -2248,29 +563,24 @@ export async function executePreparedCliRun( if (!fallbackClaudeSkillsPluginCleanupOwned) { await cleanupOuterResource(fallbackClaudeSkillsPlugin?.cleanup); } - if (systemPromptFile) { - await cleanupOuterResource(systemPromptFile.cleanup); - } - if (cleanupImages) { - await cleanupOuterResource(cleanupImages); - } + await cleanupOuterResource(systemPromptFile?.cleanup); + await cleanupOuterResource(imagePayload.cleanupImages); } catch (error) { outerCleanupError = toErrorObject(error, "CLI outer resource cleanup failed"); } } - if (outerCleanupError !== undefined) { + if (outerCleanupError) { options?.onPhase?.("cleanup"); - claudeModelCallDiagnostics?.emitError(outerCleanupError); + diagnostics?.emitError(outerCleanupError); throw outerCleanupError; } if (!completedOutput) { const error = new Error("CLI run completed without output"); - claudeModelCallDiagnostics?.emitError(error); + diagnostics?.emitError(error); throw error; } - // Success stays provisional until fallible persistence and cleanup finish; - // otherwise a rejected turn would be exported as completed. - claudeModelCallDiagnostics?.emitCompleted(completedOutput); + // Success stays provisional until persistence and cleanup finish; otherwise + // a rejected turn would be exported as completed. + diagnostics?.emitCompleted(completedOutput); return completedOutput; } -/* oxlint-disable max-lines -- TODO: split this grandfathered oversized file. */