mirror of
https://github.com/openclaw/openclaw.git
synced 2026-08-02 11:31:34 +00:00
refactor(agents): split CLI runner execution (#113716)
* refactor(agents): split CLI runner execution * refactor(agents): keep CLI log helper private
This commit is contained in:
Peter Steinberger
committed by
GitHub
GitHub
parent
f1538c6d6c
commit
22226bf2f6
@@ -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
|
||||
|
||||
21
src/agents/cli-runner/execute-deps.ts
Normal file
21
src/agents/cli-runner/execute-deps.ts
Normal file
@@ -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;
|
||||
300
src/agents/cli-runner/execute-events.ts
Normal file
300
src/agents/cli-runner/execute-events.ts
Normal file
@@ -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<typeof createCliEventHandlers>;
|
||||
210
src/agents/cli-runner/execute-logging.ts
Normal file
210
src/agents/cli-runner/execute-logging.ts
Normal file
@@ -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, `<systemPrompt:${systemPromptValue.length} chars>`);
|
||||
i += 1;
|
||||
continue;
|
||||
}
|
||||
if (arg === params.modelArg) {
|
||||
logArgs.push(arg, params.args[i + 1] ?? "");
|
||||
i += 1;
|
||||
continue;
|
||||
}
|
||||
if (arg === params.imageArg) {
|
||||
logArgs.push(arg, "<image>");
|
||||
i += 1;
|
||||
continue;
|
||||
}
|
||||
logArgs.push(arg);
|
||||
}
|
||||
if (params.argsPrompt) {
|
||||
const promptIndex = logArgs.indexOf(params.argsPrompt);
|
||||
if (promptIndex >= 0) {
|
||||
logArgs[promptIndex] = `<prompt:${params.argsPrompt.length} chars>`;
|
||||
}
|
||||
}
|
||||
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<string, string> {
|
||||
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<string> {
|
||||
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<string, string | undefined>,
|
||||
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, string>): 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, string>): 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<string, string>;
|
||||
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)}`);
|
||||
}
|
||||
}
|
||||
537
src/agents/cli-runner/execute-process.ts
Normal file
537
src/agents/cli-runner/execute-process.ts
Normal file
@@ -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<typeof createClaudeCliModelCallDiagnostics>;
|
||||
nodePlacement: ReturnType<typeof resolveNodeClaudePlacement>;
|
||||
nodeSystemPrompt?: string;
|
||||
nodeEnv?: Record<string, string>;
|
||||
nodeClearEnv?: string[];
|
||||
useManagedClaudeLiveSession: boolean;
|
||||
useResume: boolean;
|
||||
cliSessionIdToUse?: string;
|
||||
resolvedSessionId?: string;
|
||||
executionCommand: string;
|
||||
executionLeadingArgv: readonly string[];
|
||||
executionArgs: string[];
|
||||
env: Record<string, string>;
|
||||
prompt: string;
|
||||
argsPrompt?: string;
|
||||
stdin?: string;
|
||||
noOutputTimeoutMs: number;
|
||||
outputMode: CliBackendConfig["output"];
|
||||
logOutputText: boolean;
|
||||
cliTurnStartedAt: number;
|
||||
fallbackCleanup?: () => Promise<void>;
|
||||
claimFallbackCleanup: () => void;
|
||||
observeForkSuccessor: (sessionId: string) => void;
|
||||
options?: ExecuteCliProcessOptions;
|
||||
}): Promise<CliOutput> {
|
||||
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),
|
||||
};
|
||||
}
|
||||
622
src/agents/cli-runner/execute-tool-tracking.ts
Normal file
622
src/agents/cli-runner/execute-tool-tracking.ts
Normal file
@@ -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<CliLoopbackCall>;
|
||||
activeToolCallIds: Set<string>;
|
||||
};
|
||||
type ActiveCliTool = {
|
||||
toolName: string;
|
||||
args: Record<string, unknown>;
|
||||
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<McpLoopbackToolCallStart>();
|
||||
const pendingMessagingCalls = new Map<
|
||||
string,
|
||||
{ toolName: string; args: Record<string, unknown>; target?: MessagingToolSend }
|
||||
>();
|
||||
const cliLoopbackCalls: CliLoopbackCall[] = [];
|
||||
const activeCliTools = new Map<string, ActiveCliTool>();
|
||||
let cliLoopbackCorrelationOverflowed = false;
|
||||
const messagingToolSentTexts: string[] = [];
|
||||
const messagingToolSentTextKeys = new Set<string>();
|
||||
const messagingToolSentMediaUrls: string[] = [];
|
||||
const messagingToolSentMediaUrlKeys = new Set<string>();
|
||||
const messagingToolSentTargets: MessagingToolSend[] = [];
|
||||
const messagingToolSentTargetKeys = new Set<string>();
|
||||
const messagingToolSourceReplyPayloads: MessagingToolSourceReplyPayload[] = [];
|
||||
const matchesCliLoopbackCall = (
|
||||
toolName: string,
|
||||
toolArgs: Record<string, unknown>,
|
||||
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<CliLoopbackAmbiguityGroup>();
|
||||
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<CliLoopbackCall>(),
|
||||
activeToolCallIds: new Set<string>(),
|
||||
};
|
||||
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<string, unknown>;
|
||||
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<string, unknown>) =>
|
||||
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<typeof createCliToolTracking>;
|
||||
File diff suppressed because it is too large
Load Diff
Reference in New Issue
Block a user