mirror of
https://github.com/openclaw/openclaw.git
synced 2026-08-12 21:53:00 -06:00
refactor(agents): split cli-runner into concept modules (#122442)
* refactor(agents): split CLI runner concepts * chore(lint): ratchet cli-runner max-lines baseline * refactor(agents): remove stale runner import * refactor(agents): preserve cleanup error typing * refactor(agents): disambiguate CLI backend predicate * test(agents): align code-mode pending boundary Fixes red main: both host calls are unsettled when the guest snapshot is first parked; later waits still prove that settled calls are filtered. * test(agents): tolerate code-mode settlement race Fixes red main: the fast host call may settle as the guest snapshot is parked, while the deliberately slow call must remain pending.
This commit is contained in:
Peter Steinberger
committed by
GitHub
GitHub
parent
08142099da
commit
31aa7c7c75
@@ -336,7 +336,6 @@ src/agents/btw.ts
|
||||
src/agents/cli-auth-epoch.test.ts
|
||||
src/agents/cli-runner.reliability.test.ts
|
||||
src/agents/cli-runner.spawn.test.ts
|
||||
src/agents/cli-runner.ts
|
||||
src/agents/cli-runner/execute.supervisor-capture.test.ts
|
||||
src/agents/cli-runner/prepare.test.ts
|
||||
src/agents/cli-runner/prepare.ts
|
||||
|
||||
+97
-1206
File diff suppressed because it is too large
Load Diff
@@ -0,0 +1,207 @@
|
||||
import { formatErrorMessage } from "../../infra/errors.js";
|
||||
import type { EmbeddedAgentRunResult } from "../embedded-agent-runner.js";
|
||||
import { type FailoverError, isFailoverError } from "../failover-error.js";
|
||||
import { createCliFailoverError } from "./exit-error.js";
|
||||
import { cliBackendLog } from "./log.js";
|
||||
import type { CliReusableSession, PreparedCliRunContext } from "./types.js";
|
||||
|
||||
export type CliRecoveryOptions = {
|
||||
timeoutMs?: number;
|
||||
forkCliSessionOnResume?: boolean;
|
||||
resumeAt?: string;
|
||||
onForkSuccessorPersisted?: (sessionId: string) => void;
|
||||
};
|
||||
|
||||
export function resolveCliSessionId(reusableCliSession: CliReusableSession): string | undefined {
|
||||
return reusableCliSession.mode === "reuse" || reusableCliSession.mode === "reuse-with-drift"
|
||||
? reusableCliSession.sessionId
|
||||
: undefined;
|
||||
}
|
||||
|
||||
function shouldRetryFreshCliSessionAfterFailover(params: {
|
||||
error: FailoverError;
|
||||
hasHistoryPrompt: boolean;
|
||||
}): boolean {
|
||||
if (!params.hasHistoryPrompt) {
|
||||
return false;
|
||||
}
|
||||
switch (params.error.reason) {
|
||||
case "session_expired":
|
||||
return true;
|
||||
case "unknown":
|
||||
return params.error.code === "cli_unknown_empty_failure";
|
||||
case "empty_response":
|
||||
return params.error.code === "cli_unknown_empty_failure";
|
||||
case "format":
|
||||
return params.error.code === "cli_synthetic_no_response";
|
||||
case "timeout":
|
||||
return params.error.code === "cli_no_output_timeout";
|
||||
case "context_overflow":
|
||||
return params.error.code === "cli_context_overflow";
|
||||
default:
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
function shouldRetryForkedCliSessionAfterFailover(error: FailoverError): boolean {
|
||||
return error.reason === "timeout" && error.code === "cli_no_output_timeout";
|
||||
}
|
||||
|
||||
function isUnsupportedCliResumeAtError(error: unknown, resumeAtArg: string): boolean {
|
||||
const message = formatErrorMessage(error).toLowerCase();
|
||||
return (
|
||||
message.includes(resumeAtArg.toLowerCase()) &&
|
||||
/\b(?:unknown|unexpected|unrecognized)\b|\bnot\s+recognized\b/.test(message)
|
||||
);
|
||||
}
|
||||
|
||||
export async function runCliRecovery<TAttempt>(params: {
|
||||
context: PreparedCliRunContext;
|
||||
executeAttempt: (cliSessionIdToUse?: string, options?: CliRecoveryOptions) => Promise<TAttempt>;
|
||||
finishAttempt: (
|
||||
attempt: TAttempt,
|
||||
fallbackCliSessionId?: string,
|
||||
) => Promise<EmbeddedAgentRunResult>;
|
||||
finishDeliveredFailure: (error: unknown) => Promise<EmbeddedAgentRunResult | undefined>;
|
||||
onTerminalFailure: (error: unknown) => Promise<void>;
|
||||
}): Promise<EmbeddedAgentRunResult> {
|
||||
const { context } = params;
|
||||
const runParams = context.params;
|
||||
const reusableCliSessionId = resolveCliSessionId(context.reusableCliSession);
|
||||
const resumeCheckpointId = runParams.cliSessionBinding?.resumeCheckpointId;
|
||||
let retryableSessionId = reusableCliSessionId;
|
||||
try {
|
||||
return await params.finishAttempt(
|
||||
await params.executeAttempt(
|
||||
reusableCliSessionId,
|
||||
runParams.forkCliSessionOnResume
|
||||
? {
|
||||
onForkSuccessorPersisted: (sessionId) => {
|
||||
retryableSessionId = sessionId;
|
||||
},
|
||||
}
|
||||
: undefined,
|
||||
),
|
||||
reusableCliSessionId,
|
||||
);
|
||||
} catch (err) {
|
||||
const deliveredFailure = await params.finishDeliveredFailure(err);
|
||||
if (deliveredFailure) {
|
||||
return deliveredFailure;
|
||||
}
|
||||
let recoveryError = err;
|
||||
if (
|
||||
runParams.forkCliSessionOnResume &&
|
||||
resumeCheckpointId &&
|
||||
context.preparedBackend.backend.resumeAtArg &&
|
||||
isUnsupportedCliResumeAtError(err, context.preparedBackend.backend.resumeAtArg)
|
||||
) {
|
||||
recoveryError = createCliFailoverError(
|
||||
"CLI backend cannot resume from the stored checkpoint.",
|
||||
"session_expired",
|
||||
{
|
||||
provider: runParams.provider,
|
||||
model: context.modelId,
|
||||
sessionId: runParams.sessionId,
|
||||
lane: runParams.lane,
|
||||
},
|
||||
{ cause: err },
|
||||
);
|
||||
}
|
||||
if (isFailoverError(recoveryError)) {
|
||||
if (
|
||||
!runParams.forkCliSessionOnResume &&
|
||||
shouldRetryForkedCliSessionAfterFailover(recoveryError) &&
|
||||
retryableSessionId &&
|
||||
resumeCheckpointId &&
|
||||
runParams.sessionKey &&
|
||||
context.preparedBackend.backend.forkArg &&
|
||||
context.preparedBackend.backend.resumeAtArg &&
|
||||
runParams.onBeforeForkedCliSessionRetry
|
||||
) {
|
||||
try {
|
||||
const retryTimeoutMs = runParams.timeoutMs - (Date.now() - context.started);
|
||||
if (retryTimeoutMs <= 0) {
|
||||
throw recoveryError;
|
||||
}
|
||||
const forkPrepared = await runParams.onBeforeForkedCliSessionRetry({
|
||||
provider: runParams.provider,
|
||||
reason: recoveryError.reason,
|
||||
sessionId: retryableSessionId,
|
||||
});
|
||||
if (!forkPrepared) {
|
||||
throw recoveryError;
|
||||
}
|
||||
cliBackendLog.warn(
|
||||
`cli session recovery fork: provider=${runParams.provider} reason=${recoveryError.reason} sessionKey=${runParams.sessionKey}`,
|
||||
);
|
||||
return await params.finishAttempt(
|
||||
await params.executeAttempt(retryableSessionId, {
|
||||
timeoutMs: retryTimeoutMs,
|
||||
forkCliSessionOnResume: true,
|
||||
resumeAt: resumeCheckpointId,
|
||||
onForkSuccessorPersisted: (sessionId) => {
|
||||
retryableSessionId = sessionId;
|
||||
},
|
||||
}),
|
||||
);
|
||||
} catch (forkError) {
|
||||
const deliveredForkFailure = await params.finishDeliveredFailure(forkError);
|
||||
if (deliveredForkFailure) {
|
||||
return deliveredForkFailure;
|
||||
}
|
||||
recoveryError = isUnsupportedCliResumeAtError(
|
||||
forkError,
|
||||
context.preparedBackend.backend.resumeAtArg,
|
||||
)
|
||||
? err
|
||||
: forkError;
|
||||
}
|
||||
}
|
||||
if (
|
||||
isFailoverError(recoveryError) &&
|
||||
shouldRetryFreshCliSessionAfterFailover({
|
||||
error: recoveryError,
|
||||
hasHistoryPrompt: Boolean(context.openClawHistoryPrompt),
|
||||
}) &&
|
||||
retryableSessionId &&
|
||||
runParams.sessionKey
|
||||
) {
|
||||
try {
|
||||
const retryTimeoutMs = runParams.timeoutMs - (Date.now() - context.started);
|
||||
if (retryTimeoutMs <= 0) {
|
||||
throw recoveryError;
|
||||
}
|
||||
if (runParams.onBeforeFreshCliSessionRetry) {
|
||||
const clearedStaleBinding = await runParams.onBeforeFreshCliSessionRetry({
|
||||
provider: runParams.provider,
|
||||
reason: recoveryError.reason,
|
||||
sessionId: retryableSessionId,
|
||||
});
|
||||
if (!clearedStaleBinding) {
|
||||
throw recoveryError;
|
||||
}
|
||||
}
|
||||
cliBackendLog.warn(
|
||||
`cli session recovery retry: provider=${runParams.provider} reason=${recoveryError.reason} sessionKey=${runParams.sessionKey}`,
|
||||
);
|
||||
return await params.finishAttempt(
|
||||
await params.executeAttempt(undefined, {
|
||||
timeoutMs: retryTimeoutMs,
|
||||
forkCliSessionOnResume: false,
|
||||
}),
|
||||
);
|
||||
} catch (retryErr) {
|
||||
const deliveredRetryFailure = await params.finishDeliveredFailure(retryErr);
|
||||
if (deliveredRetryFailure) {
|
||||
return deliveredRetryFailure;
|
||||
}
|
||||
await params.onTerminalFailure(retryErr);
|
||||
throw retryErr;
|
||||
}
|
||||
}
|
||||
}
|
||||
await params.onTerminalFailure(recoveryError);
|
||||
throw recoveryError;
|
||||
}
|
||||
}
|
||||
+1
-1
@@ -4,7 +4,7 @@ import {
|
||||
isCliBindingFlushed,
|
||||
restoreCliRunnerTestDeps,
|
||||
setCliRunnerTestDeps,
|
||||
} from "./cli-runner.js";
|
||||
} from "../cli-runner.js";
|
||||
|
||||
describe("isCliBindingFlushed", () => {
|
||||
const workspaceDir = "/tmp/openclaw-workspace";
|
||||
@@ -0,0 +1,657 @@
|
||||
import { setReplyPayloadMetadata, type ReplyPayload } from "../../auto-reply/reply-payload.js";
|
||||
import { SILENT_REPLY_TOKEN } from "../../auto-reply/tokens.js";
|
||||
import { formatErrorMessage } from "../../infra/errors.js";
|
||||
import { createSubsystemLogger } from "../../logging/subsystem.js";
|
||||
import {
|
||||
externalCliDiscoveryForProviderAuth,
|
||||
loadAuthProfileStoreForRuntime,
|
||||
markAuthProfileFailure,
|
||||
markAuthProfileSuccess,
|
||||
type AuthProfileStore,
|
||||
} from "../auth-profiles.js";
|
||||
import {
|
||||
resolveCliRuntimeArtifactFingerprint,
|
||||
resolveCliRuntimeOwnerFingerprint,
|
||||
} from "../cli-auth-epoch.js";
|
||||
import type { CliOutput } from "../cli-output-contracts.js";
|
||||
import { claudeCliSessionTranscriptHasContent as claudeCliSessionTranscriptHasContentImpl } from "../command/attempt-execution.helpers.js";
|
||||
import type { EmbeddedAgentRunResult } from "../embedded-agent-runner.js";
|
||||
import { resolveExplicitFinalSourceReplyDeliveryEvidence } from "../embedded-agent-runner/delivery-evidence.js";
|
||||
import { resolveAuthProfileFailureReason } from "../embedded-agent-runner/run/auth-profile-failure-policy.js";
|
||||
import { buildEmbeddedRunPayloads } from "../embedded-agent-runner/run/payloads.js";
|
||||
import { coerceToFailoverError, isFailoverError } from "../failover-error.js";
|
||||
import { CliAuthProfilePreparationError } from "./auth-profile-preparation-error.js";
|
||||
import { hashCliReseedPrompt } from "./reseed-envelope.js";
|
||||
import type { ClaudeCliRunDiagnosticLifecycle } from "./run-diagnostics.js";
|
||||
import type { PreparedCliRunContext, RunCliAgentParams } from "./types.js";
|
||||
|
||||
const log = createSubsystemLogger("agents/cli-runner");
|
||||
|
||||
export const cliRunSettlementDeps = {
|
||||
claudeCliSessionTranscriptHasContent: claudeCliSessionTranscriptHasContentImpl,
|
||||
delay: async (delayMs: number) => {
|
||||
await new Promise((resolve) => {
|
||||
setTimeout(resolve, delayMs);
|
||||
});
|
||||
},
|
||||
loadAuthProfileStoreForRuntime,
|
||||
markAuthProfileFailure,
|
||||
markAuthProfileSuccess,
|
||||
};
|
||||
|
||||
async function settleCliAuthProfile(params: {
|
||||
store: AuthProfileStore;
|
||||
profileId: string;
|
||||
provider: string;
|
||||
agentDir?: string;
|
||||
terminal:
|
||||
| { outcome: "success" }
|
||||
| {
|
||||
outcome: "failure";
|
||||
error: unknown;
|
||||
config?: RunCliAgentParams["config"];
|
||||
runId: string;
|
||||
modelId?: string;
|
||||
};
|
||||
}): Promise<void> {
|
||||
try {
|
||||
if (params.terminal.outcome === "success") {
|
||||
await cliRunSettlementDeps.markAuthProfileSuccess({
|
||||
store: params.store,
|
||||
profileId: params.profileId,
|
||||
provider: params.provider,
|
||||
agentDir: params.agentDir,
|
||||
});
|
||||
return;
|
||||
}
|
||||
const error = params.terminal.error;
|
||||
const reason = resolveAuthProfileFailureReason({
|
||||
failoverReason: isFailoverError(error) ? error.reason : null,
|
||||
providerStarted:
|
||||
isFailoverError(error) && error.reason === "timeout"
|
||||
? error.cliTimeout?.observedActivity
|
||||
: undefined,
|
||||
});
|
||||
if (reason) {
|
||||
await cliRunSettlementDeps.markAuthProfileFailure({
|
||||
store: params.store,
|
||||
profileId: params.profileId,
|
||||
reason,
|
||||
cfg: params.terminal.config,
|
||||
agentDir: params.agentDir,
|
||||
runId: params.terminal.runId,
|
||||
modelId: params.terminal.modelId,
|
||||
});
|
||||
}
|
||||
} catch (error) {
|
||||
log.warn(
|
||||
`CLI auth-profile ${params.terminal.outcome} settlement failed: ${formatErrorMessage(error)}`,
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
export function isClaudeCliBackend(provider: string): boolean {
|
||||
return provider.trim().toLowerCase() === "claude-cli";
|
||||
}
|
||||
|
||||
export async function assertCliRuntimeBinding(context: PreparedCliRunContext): Promise<void> {
|
||||
if (!context.runtimeArtifactFingerprint) {
|
||||
return;
|
||||
}
|
||||
const currentArtifact = await resolveCliRuntimeArtifactFingerprint({
|
||||
provider: context.params.provider,
|
||||
config: context.params.config ?? context.contextEngineConfig,
|
||||
agentId: context.params.agentId,
|
||||
runtimeArtifactId: context.backendResolved.id,
|
||||
});
|
||||
if (currentArtifact !== context.runtimeArtifactFingerprint) {
|
||||
throw new Error("CLI executable/package artifact changed during successful inference");
|
||||
}
|
||||
if (!context.runtimeOwnerFingerprint) {
|
||||
return;
|
||||
}
|
||||
const currentOwner = await resolveCliRuntimeOwnerFingerprint({
|
||||
provider: context.params.provider,
|
||||
config: context.params.config ?? context.contextEngineConfig,
|
||||
...(context.agentDir ? { agentDir: context.agentDir } : {}),
|
||||
agentId: context.params.agentId,
|
||||
runtimeOwnerId: context.backendResolved.id,
|
||||
...(context.effectiveAuthProfileId ? { authProfileId: context.effectiveAuthProfileId } : {}),
|
||||
...(context.authBindingSkipsLocalCredential ? { skipLocalCredential: true } : {}),
|
||||
runtimeArtifactFingerprint: currentArtifact,
|
||||
});
|
||||
if (currentOwner !== context.runtimeOwnerFingerprint) {
|
||||
throw new Error("CLI runtime owner changed during successful inference");
|
||||
}
|
||||
}
|
||||
|
||||
export async function settleCliPreparationError(
|
||||
error: unknown,
|
||||
params: RunCliAgentParams,
|
||||
): Promise<void> {
|
||||
if (!(error instanceof CliAuthProfilePreparationError)) {
|
||||
return;
|
||||
}
|
||||
const store = cliRunSettlementDeps.loadAuthProfileStoreForRuntime(error.agentDir, {
|
||||
externalCli: externalCliDiscoveryForProviderAuth({
|
||||
cfg: params.config,
|
||||
provider: error.provider,
|
||||
profileId: error.profileId,
|
||||
}),
|
||||
});
|
||||
await settleCliAuthProfile({
|
||||
store,
|
||||
profileId: error.profileId,
|
||||
provider: error.provider,
|
||||
agentDir: error.agentDir,
|
||||
terminal: {
|
||||
outcome: "failure",
|
||||
error,
|
||||
config: params.config,
|
||||
runId: params.runId,
|
||||
modelId: params.model,
|
||||
},
|
||||
});
|
||||
}
|
||||
|
||||
export async function settlePreparedCliRun(params: {
|
||||
context: PreparedCliRunContext;
|
||||
diagnosticLifecycle?: ClaudeCliRunDiagnosticLifecycle;
|
||||
run: () => Promise<EmbeddedAgentRunResult>;
|
||||
}): Promise<EmbeddedAgentRunResult> {
|
||||
const { context, diagnosticLifecycle, run } = params;
|
||||
const runParams = context.params;
|
||||
let result: EmbeddedAgentRunResult | undefined;
|
||||
let runError: unknown;
|
||||
try {
|
||||
result = await run();
|
||||
} catch (error) {
|
||||
runError = error;
|
||||
}
|
||||
const terminalRunError = runError;
|
||||
let cleanupError: unknown;
|
||||
const recordCleanupError = (error: unknown) => {
|
||||
cleanupError ??= error;
|
||||
};
|
||||
if (runParams.cleanupCliLiveSessionOnRunEnd === true) {
|
||||
try {
|
||||
const { closeClaudeSession } = await import("./claude-live-registry.js");
|
||||
await closeClaudeSession(context, "restart");
|
||||
} catch (error) {
|
||||
recordCleanupError(error);
|
||||
}
|
||||
}
|
||||
if (runParams.cleanupBundleMcpOnRunEnd === true) {
|
||||
// The run's session ID is immutable; its session key can already belong to
|
||||
// a newer run. Never retire the newer runtime or close the shared listener.
|
||||
try {
|
||||
const { retireSessionMcpRuntime } = await import("../agent-bundle-mcp-tools.js");
|
||||
await retireSessionMcpRuntime({
|
||||
sessionId: runParams.sessionId,
|
||||
reason: "cli-run-end",
|
||||
onError: recordCleanupError,
|
||||
});
|
||||
} catch (error) {
|
||||
recordCleanupError(error);
|
||||
}
|
||||
}
|
||||
if (cleanupError) {
|
||||
if (runError || result?.didSendViaMessagingTool === true) {
|
||||
log.warn(`cli run cleanup failed after completion: ${formatErrorMessage(cleanupError)}`);
|
||||
} else {
|
||||
diagnosticLifecycle?.setPhase("cleanup");
|
||||
runError =
|
||||
cleanupError instanceof Error ? cleanupError : new Error(formatErrorMessage(cleanupError));
|
||||
}
|
||||
}
|
||||
// Settle only after backend recovery is exhausted. Recording inside an
|
||||
// attempt would quarantine a healthy profile for a recovered session fault.
|
||||
if (context.effectiveAuthProfileId && context.authProfileStore) {
|
||||
const profileId = context.effectiveAuthProfileId;
|
||||
const authProfileStore = context.authProfileStore;
|
||||
if (terminalRunError) {
|
||||
await settleCliAuthProfile({
|
||||
store: authProfileStore,
|
||||
profileId,
|
||||
provider: authProfileStore.profiles[profileId]?.provider ?? runParams.provider,
|
||||
agentDir: context.agentDir,
|
||||
terminal: {
|
||||
outcome: "failure",
|
||||
error: terminalRunError,
|
||||
config: runParams.config,
|
||||
runId: runParams.runId,
|
||||
modelId: context.modelId,
|
||||
},
|
||||
});
|
||||
} else if (result?.meta.executionTrace?.attempts?.at(-1)?.result === "success") {
|
||||
const provider = authProfileStore.profiles[profileId]?.provider ?? runParams.provider;
|
||||
await settleCliAuthProfile({
|
||||
store: authProfileStore,
|
||||
profileId,
|
||||
provider,
|
||||
agentDir: context.agentDir,
|
||||
terminal: { outcome: "success" },
|
||||
});
|
||||
}
|
||||
}
|
||||
if (runError) {
|
||||
throw runError instanceof Error ? runError : new Error(formatErrorMessage(runError));
|
||||
}
|
||||
return result as EmbeddedAgentRunResult;
|
||||
}
|
||||
|
||||
export function resolveCliSourceReplyMirror(params: {
|
||||
evidence: Pick<
|
||||
CliOutput,
|
||||
| "didSendViaMessagingTool"
|
||||
| "didDeliverSourceReplyViaMessageTool"
|
||||
| "messagingToolSentTargets"
|
||||
| "messagingToolSourceReplyPayloads"
|
||||
>;
|
||||
runParams: RunCliAgentParams;
|
||||
modelId: string;
|
||||
}): { payloads: ReplyPayload[]; delivered: boolean; visibleText?: string } {
|
||||
const { evidence, modelId, runParams } = params;
|
||||
const payloads = buildEmbeddedRunPayloads({
|
||||
assistantTexts: [],
|
||||
lastAssistant: undefined,
|
||||
sessionKey: runParams.sessionKey ?? "",
|
||||
provider: runParams.provider,
|
||||
model: modelId,
|
||||
didSendViaMessagingTool: evidence.didSendViaMessagingTool,
|
||||
didDeliverSourceReplyViaMessageTool: evidence.didDeliverSourceReplyViaMessageTool,
|
||||
messagingToolSentTargets: evidence.messagingToolSentTargets,
|
||||
messagingToolSourceReplyPayloads: evidence.messagingToolSourceReplyPayloads,
|
||||
sourceReplyDeliveryMode: runParams.sourceReplyDeliveryMode,
|
||||
agentId: runParams.agentId,
|
||||
runId: runParams.runId,
|
||||
});
|
||||
const delivered =
|
||||
payloads.length > 0 ||
|
||||
(runParams.sourceReplyDeliveryMode === "message_tool_only" &&
|
||||
evidence.didDeliverSourceReplyViaMessageTool === true);
|
||||
const visibleText =
|
||||
payloads
|
||||
.map((payload) => payload.text?.trim() ?? "")
|
||||
.filter(Boolean)
|
||||
.join("\n\n") || undefined;
|
||||
return { payloads, delivered, visibleText };
|
||||
}
|
||||
|
||||
export function buildBlockedCliRunResult(params: {
|
||||
message: string;
|
||||
context: PreparedCliRunContext;
|
||||
preparedContextAgentMeta: { contextTokens?: number };
|
||||
sessionBindingDisabled: boolean;
|
||||
}): EmbeddedAgentRunResult {
|
||||
const { context, message, preparedContextAgentMeta, sessionBindingDisabled } = params;
|
||||
const runParams = context.params;
|
||||
return {
|
||||
payloads: [{ text: message, isError: true }],
|
||||
meta: {
|
||||
durationMs: Date.now() - context.started,
|
||||
finalAssistantVisibleText: message,
|
||||
finalAssistantRawText: message,
|
||||
livenessState: "blocked",
|
||||
error: {
|
||||
kind: "hook_block",
|
||||
message,
|
||||
},
|
||||
systemPromptReport: context.systemPromptReport,
|
||||
executionTrace: {
|
||||
winnerProvider: runParams.provider,
|
||||
winnerModel: context.modelId,
|
||||
attempts: [
|
||||
{
|
||||
provider: runParams.provider,
|
||||
model: context.modelId,
|
||||
result: "error",
|
||||
reason: "before_agent_run blocked the run",
|
||||
},
|
||||
],
|
||||
fallbackUsed: false,
|
||||
runner: "cli",
|
||||
},
|
||||
requestShaping: {
|
||||
...(runParams.thinkLevel ? { thinking: runParams.thinkLevel } : {}),
|
||||
...(context.effectiveAuthProfileId ? { authMode: "auth-profile" } : {}),
|
||||
},
|
||||
completion: {
|
||||
finishReason: "blocked",
|
||||
stopReason: "blocked",
|
||||
refusal: true,
|
||||
},
|
||||
agentMeta: {
|
||||
sessionId: runParams.sessionId ?? "",
|
||||
provider: runParams.provider,
|
||||
model: context.modelId,
|
||||
...preparedContextAgentMeta,
|
||||
...(sessionBindingDisabled ? { clearCliSessionBinding: true } : {}),
|
||||
},
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
export function buildCliDeliveredFailure(params: {
|
||||
error: unknown;
|
||||
evidence: NonNullable<
|
||||
ReturnType<typeof import("./delivery-evidence.js").getCliMessagingDeliveryEvidence>
|
||||
>;
|
||||
context: PreparedCliRunContext;
|
||||
preparedContextAgentMeta: { contextTokens?: number };
|
||||
sessionBindingDisabled: boolean;
|
||||
reusableCliSessionId?: string;
|
||||
}): EmbeddedAgentRunResult {
|
||||
const {
|
||||
context,
|
||||
error,
|
||||
evidence,
|
||||
preparedContextAgentMeta,
|
||||
reusableCliSessionId,
|
||||
sessionBindingDisabled,
|
||||
} = params;
|
||||
const runParams = context.params;
|
||||
const message = formatErrorMessage(error);
|
||||
const { payloads } = resolveCliSourceReplyMirror({
|
||||
evidence,
|
||||
runParams,
|
||||
modelId: context.modelId,
|
||||
});
|
||||
const visiblePayloads =
|
||||
payloads.length > 0
|
||||
? payloads
|
||||
: resolveExplicitFinalSourceReplyDeliveryEvidence(evidence) === false
|
||||
? [{ text: "The reply stopped after sending progress. Please try again.", isError: true }]
|
||||
: undefined;
|
||||
return {
|
||||
...(visiblePayloads ? { payloads: visiblePayloads } : {}),
|
||||
meta: {
|
||||
durationMs: Date.now() - context.started,
|
||||
systemPromptReport: context.systemPromptReport,
|
||||
stopReason: "error",
|
||||
executionTrace: {
|
||||
winnerProvider: runParams.provider,
|
||||
winnerModel: context.modelId,
|
||||
attempts: [
|
||||
{
|
||||
provider: runParams.provider,
|
||||
model: context.modelId,
|
||||
result: "error",
|
||||
reason: message,
|
||||
},
|
||||
],
|
||||
fallbackUsed: false,
|
||||
runner: "cli",
|
||||
},
|
||||
requestShaping: {
|
||||
...(runParams.thinkLevel ? { thinking: runParams.thinkLevel } : {}),
|
||||
...(context.effectiveAuthProfileId ? { authMode: "auth-profile" } : {}),
|
||||
},
|
||||
completion: {
|
||||
finishReason: "error",
|
||||
stopReason: "error",
|
||||
refusal: false,
|
||||
},
|
||||
agentMeta: {
|
||||
sessionId: "",
|
||||
provider: runParams.provider,
|
||||
model: context.modelId,
|
||||
...preparedContextAgentMeta,
|
||||
...(sessionBindingDisabled || reusableCliSessionId ? { clearCliSessionBinding: true } : {}),
|
||||
},
|
||||
},
|
||||
didSendViaMessagingTool: true,
|
||||
...(evidence.didDeliverSourceReplyViaMessageTool
|
||||
? { didDeliverSourceReplyViaMessageTool: true }
|
||||
: {}),
|
||||
...(evidence.messagingToolSentTexts?.length
|
||||
? { messagingToolSentTexts: evidence.messagingToolSentTexts }
|
||||
: {}),
|
||||
...(evidence.messagingToolSentMediaUrls?.length
|
||||
? { messagingToolSentMediaUrls: evidence.messagingToolSentMediaUrls }
|
||||
: {}),
|
||||
...(evidence.messagingToolSentTargets?.length
|
||||
? { messagingToolSentTargets: evidence.messagingToolSentTargets }
|
||||
: {}),
|
||||
...(evidence.messagingToolSourceReplyPayloads?.length
|
||||
? { messagingToolSourceReplyPayloads: evidence.messagingToolSourceReplyPayloads }
|
||||
: {}),
|
||||
};
|
||||
}
|
||||
|
||||
export function buildCliRunResult(params: {
|
||||
context: PreparedCliRunContext;
|
||||
output: CliOutput;
|
||||
effectiveCliSessionId?: string;
|
||||
bindingFlushOk?: boolean;
|
||||
assistantTranscriptOwned?: boolean;
|
||||
usedHistoryPrompt: boolean;
|
||||
userTurnHandled: boolean;
|
||||
sessionBindingDisabled: boolean;
|
||||
preparedContextAgentMeta: { contextTokens?: number };
|
||||
}): EmbeddedAgentRunResult {
|
||||
const {
|
||||
assistantTranscriptOwned,
|
||||
bindingFlushOk,
|
||||
context,
|
||||
effectiveCliSessionId,
|
||||
output,
|
||||
preparedContextAgentMeta,
|
||||
sessionBindingDisabled,
|
||||
usedHistoryPrompt,
|
||||
userTurnHandled,
|
||||
} = params;
|
||||
const runParams = context.params;
|
||||
const text = output.text?.trim();
|
||||
const rawText = output.rawText?.trim();
|
||||
const sourceReplyMirror = resolveCliSourceReplyMirror({
|
||||
evidence: output,
|
||||
runParams,
|
||||
modelId: context.modelId,
|
||||
});
|
||||
const finalAssistantVisibleText = sourceReplyMirror.delivered
|
||||
? sourceReplyMirror.visibleText
|
||||
: text;
|
||||
const payloads =
|
||||
sourceReplyMirror.payloads.length > 0
|
||||
? sourceReplyMirror.payloads
|
||||
: sourceReplyMirror.delivered
|
||||
? undefined
|
||||
: text
|
||||
? [
|
||||
assistantTranscriptOwned
|
||||
? setReplyPayloadMetadata({ text }, { assistantTranscriptOwned: true })
|
||||
: { text },
|
||||
]
|
||||
: runParams.allowEmptyAssistantReplyAsSilent === true
|
||||
? [{ text: SILENT_REPLY_TOKEN }]
|
||||
: undefined;
|
||||
const unflushedCliSessionId =
|
||||
!sessionBindingDisabled && effectiveCliSessionId && bindingFlushOk === false
|
||||
? effectiveCliSessionId
|
||||
: undefined;
|
||||
const persistedCliSessionId = sessionBindingDisabled
|
||||
? undefined
|
||||
: unflushedCliSessionId
|
||||
? undefined
|
||||
: effectiveCliSessionId;
|
||||
const createdReseedReceipt =
|
||||
persistedCliSessionId &&
|
||||
usedHistoryPrompt &&
|
||||
isClaudeCliBackend(runParams.provider) &&
|
||||
output.finalPromptText !== undefined &&
|
||||
userTurnHandled &&
|
||||
runParams.sessionId
|
||||
? {
|
||||
version: 1 as const,
|
||||
promptHash: hashCliReseedPrompt(output.finalPromptText),
|
||||
localSessionId: runParams.sessionId,
|
||||
userTurnDisposition: runParams.userTurnTranscriptRecorder?.hasPersisted()
|
||||
? ("persisted" as const)
|
||||
: ("omitted" as const),
|
||||
}
|
||||
: undefined;
|
||||
const preservedReseedReceipt =
|
||||
runParams.cliSessionBinding && persistedCliSessionId === runParams.cliSessionBinding.sessionId
|
||||
? runParams.cliSessionBinding.reseedReceipt
|
||||
: undefined;
|
||||
const reseedReceipt = createdReseedReceipt ?? preservedReseedReceipt;
|
||||
const agentSessionId = sessionBindingDisabled
|
||||
? (runParams.sessionId ?? "")
|
||||
: unflushedCliSessionId
|
||||
? ""
|
||||
: (effectiveCliSessionId ?? runParams.sessionId ?? "");
|
||||
const yielded = output.yielded === true;
|
||||
const stopReason = yielded ? "end_turn" : "completed";
|
||||
|
||||
runParams.onSuccessfulAuthBinding?.({
|
||||
...(context.effectiveAuthProfileId ? { authProfileId: context.effectiveAuthProfileId } : {}),
|
||||
...(context.authBindingFingerprint ? { authFingerprint: context.authBindingFingerprint } : {}),
|
||||
...(!context.authBindingFingerprint && context.runtimeOwnerFingerprint
|
||||
? {
|
||||
runtimeOwnerFingerprint: context.runtimeOwnerFingerprint,
|
||||
runtimeOwnerKind: "cli-runtime" as const,
|
||||
runtimeOwnerId: context.backendResolved.id,
|
||||
}
|
||||
: {}),
|
||||
...(context.runtimeArtifactFingerprint
|
||||
? {
|
||||
runtimeArtifactFingerprint: context.runtimeArtifactFingerprint,
|
||||
runtimeArtifactId: context.backendResolved.id,
|
||||
}
|
||||
: {}),
|
||||
...(context.authBindingSkipsLocalCredential ? { skipLocalCredential: true } : {}),
|
||||
});
|
||||
|
||||
return {
|
||||
payloads,
|
||||
meta: {
|
||||
durationMs: Date.now() - context.started,
|
||||
...(output.finalPromptText ? { finalPromptText: output.finalPromptText } : {}),
|
||||
...(finalAssistantVisibleText || rawText
|
||||
? {
|
||||
...(finalAssistantVisibleText ? { finalAssistantVisibleText } : {}),
|
||||
...(rawText ? { finalAssistantRawText: rawText } : {}),
|
||||
}
|
||||
: {}),
|
||||
systemPromptReport: context.systemPromptReport,
|
||||
...(yielded ? { yielded: true, livenessState: "paused" as const, stopReason } : {}),
|
||||
executionTrace: {
|
||||
winnerProvider: runParams.provider,
|
||||
winnerModel: context.modelId,
|
||||
attempts: [{ provider: runParams.provider, model: context.modelId, result: "success" }],
|
||||
fallbackUsed: false,
|
||||
runner: "cli",
|
||||
},
|
||||
requestShaping: {
|
||||
...(runParams.thinkLevel ? { thinking: runParams.thinkLevel } : {}),
|
||||
...(context.effectiveAuthProfileId ? { authMode: "auth-profile" } : {}),
|
||||
},
|
||||
completion: {
|
||||
finishReason: yielded ? "end_turn" : "stop",
|
||||
stopReason,
|
||||
refusal: false,
|
||||
},
|
||||
...(output.toolSummary ? { toolSummary: output.toolSummary } : {}),
|
||||
agentMeta: {
|
||||
sessionId: agentSessionId,
|
||||
provider: runParams.provider,
|
||||
model: context.modelId,
|
||||
...preparedContextAgentMeta,
|
||||
usage: output.usage,
|
||||
...(output.usage ? { lastCallUsage: output.usage } : {}),
|
||||
...(output.diagnosticUsage ? { diagnosticUsage: output.diagnosticUsage } : {}),
|
||||
...(persistedCliSessionId
|
||||
? {
|
||||
cliSessionBinding: {
|
||||
sessionId: persistedCliSessionId,
|
||||
...(context.effectiveAuthProfileId
|
||||
? { authProfileId: context.effectiveAuthProfileId }
|
||||
: {}),
|
||||
...(output.resumeCheckpointId
|
||||
? { resumeCheckpointId: output.resumeCheckpointId }
|
||||
: {}),
|
||||
...(context.authEpoch ? { authEpoch: context.authEpoch } : {}),
|
||||
authEpochVersion: context.authEpochVersion,
|
||||
...(context.extraSystemPromptHash
|
||||
? { extraSystemPromptHash: context.extraSystemPromptHash }
|
||||
: {}),
|
||||
...(context.messageToolPolicyHash
|
||||
? { messageToolPolicyHash: context.messageToolPolicyHash }
|
||||
: {}),
|
||||
...(context.promptToolNamesHash
|
||||
? { promptToolNamesHash: context.promptToolNamesHash }
|
||||
: {}),
|
||||
...(context.cwdHash ? { cwdHash: context.cwdHash } : {}),
|
||||
...(context.preparedBackend.mcpConfigHash
|
||||
? { mcpConfigHash: context.preparedBackend.mcpConfigHash }
|
||||
: {}),
|
||||
...(context.preparedBackend.mcpResumeHash
|
||||
? { mcpResumeHash: context.preparedBackend.mcpResumeHash }
|
||||
: {}),
|
||||
...(reseedReceipt ? { reseedReceipt } : {}),
|
||||
},
|
||||
}
|
||||
: {}),
|
||||
...(sessionBindingDisabled || unflushedCliSessionId
|
||||
? { clearCliSessionBinding: true }
|
||||
: {}),
|
||||
},
|
||||
},
|
||||
...(output.didSendViaMessagingTool ? { didSendViaMessagingTool: true } : {}),
|
||||
...(output.didDeliverSourceReplyViaMessageTool
|
||||
? { didDeliverSourceReplyViaMessageTool: true }
|
||||
: {}),
|
||||
...(output.messagingToolSentTexts?.length
|
||||
? { messagingToolSentTexts: output.messagingToolSentTexts }
|
||||
: {}),
|
||||
...(output.messagingToolSentMediaUrls?.length
|
||||
? { messagingToolSentMediaUrls: output.messagingToolSentMediaUrls }
|
||||
: {}),
|
||||
...(output.messagingToolSentTargets?.length
|
||||
? { messagingToolSentTargets: output.messagingToolSentTargets }
|
||||
: {}),
|
||||
...(output.messagingToolSourceReplyPayloads?.length
|
||||
? { messagingToolSourceReplyPayloads: output.messagingToolSourceReplyPayloads }
|
||||
: {}),
|
||||
};
|
||||
}
|
||||
|
||||
export function settleCliBackendOutcome(params: {
|
||||
runResult: EmbeddedAgentRunResult | undefined;
|
||||
runError: unknown;
|
||||
runFailed: boolean;
|
||||
cleanupError: Error | undefined;
|
||||
deliveredMessagingSideEffect: boolean;
|
||||
diagnosticLifecycle?: ClaudeCliRunDiagnosticLifecycle;
|
||||
failoverContext: { provider: string; model: string; sessionId: string; lane?: string };
|
||||
}): EmbeddedAgentRunResult {
|
||||
const {
|
||||
cleanupError,
|
||||
deliveredMessagingSideEffect,
|
||||
diagnosticLifecycle,
|
||||
failoverContext,
|
||||
runError,
|
||||
runFailed,
|
||||
runResult,
|
||||
} = params;
|
||||
if (cleanupError) {
|
||||
if (!deliveredMessagingSideEffect) {
|
||||
if (runFailed) {
|
||||
log.warn(`CLI run also failed before backend cleanup: ${formatErrorMessage(runError)}`);
|
||||
}
|
||||
diagnosticLifecycle?.setPhase("cleanup");
|
||||
throw cleanupError;
|
||||
}
|
||||
log.warn(
|
||||
`CLI backend cleanup failed after confirmed message delivery: ${formatErrorMessage(cleanupError)}`,
|
||||
);
|
||||
}
|
||||
if (runFailed) {
|
||||
throw coerceToFailoverError(runError, failoverContext) ?? runError;
|
||||
}
|
||||
if (!runResult) {
|
||||
throw new Error("CLI run completed without a result");
|
||||
}
|
||||
return runResult;
|
||||
}
|
||||
@@ -0,0 +1,404 @@
|
||||
import { resolveSessionStorePathCore } from "../../config/sessions/paths.js";
|
||||
import { patchSessionEntryCore } from "../../config/sessions/session-accessor.js";
|
||||
import { appendExactAssistantMessageToSessionTranscript } from "../../config/sessions/transcript.js";
|
||||
import { buildGenericCliContextEngineHostSupport } from "../../context-engine/host-compat.js";
|
||||
import { formatErrorMessage } from "../../infra/errors.js";
|
||||
import { createSubsystemLogger } from "../../logging/subsystem.js";
|
||||
import { resolveAgentIdFromSessionKey } from "../../routing/session-key.js";
|
||||
import { isHeartbeatLifecycleRunKind } from "../bootstrap-mode.js";
|
||||
import type { CliOutput } from "../cli-output-contracts.js";
|
||||
import {
|
||||
awaitAgentEndSideEffects,
|
||||
runAgentEndSideEffects,
|
||||
} from "../harness/agent-end-side-effects.js";
|
||||
import {
|
||||
finalizeHarnessContextEngineTurn,
|
||||
runHarnessContextEngineMaintenance,
|
||||
} from "../harness/context-engine-lifecycle.js";
|
||||
import { runAgentHarnessBeforeMessageWriteHook } from "../harness/hook-helpers.js";
|
||||
import type { AgentMessage } from "../runtime/index.js";
|
||||
import { SessionManager } from "../sessions/session-manager.js";
|
||||
import { buildAssistantMessage, buildUsageWithNoCost } from "../stream-message-shared.js";
|
||||
import type { PreparedCliRunContext, RunCliAgentParams } from "./types.js";
|
||||
|
||||
const log = createSubsystemLogger("agents/cli-runner");
|
||||
|
||||
export function buildCliHookUserMessage(prompt: string): unknown {
|
||||
return {
|
||||
role: "user",
|
||||
content: prompt,
|
||||
timestamp: Date.now(),
|
||||
};
|
||||
}
|
||||
|
||||
export function buildCliHookAssistantMessage(params: {
|
||||
text: string;
|
||||
provider: string;
|
||||
model: string;
|
||||
usage?: {
|
||||
input?: number;
|
||||
output?: number;
|
||||
cacheRead?: number;
|
||||
cacheWrite?: number;
|
||||
total?: number;
|
||||
};
|
||||
}): unknown {
|
||||
return {
|
||||
role: "assistant",
|
||||
content: [{ type: "text", text: params.text }],
|
||||
api: "responses",
|
||||
provider: params.provider,
|
||||
model: params.model,
|
||||
...(params.usage ? { usage: params.usage } : {}),
|
||||
stopReason: "stop",
|
||||
timestamp: Date.now(),
|
||||
};
|
||||
}
|
||||
|
||||
function isAgentMessage(value: unknown): value is AgentMessage {
|
||||
return Boolean(value && typeof value === "object" && "role" in value);
|
||||
}
|
||||
|
||||
function buildCliContextEngineUserMessage(prompt: string): AgentMessage {
|
||||
return {
|
||||
role: "user",
|
||||
content: prompt,
|
||||
timestamp: Date.now(),
|
||||
} as AgentMessage;
|
||||
}
|
||||
|
||||
function buildCliContextEngineAssistantMessage(params: {
|
||||
text: string;
|
||||
provider: string;
|
||||
model: string;
|
||||
usage?: {
|
||||
input?: number;
|
||||
output?: number;
|
||||
cacheRead?: number;
|
||||
cacheWrite?: number;
|
||||
total?: number;
|
||||
};
|
||||
}): AgentMessage {
|
||||
return buildCliHookAssistantMessage(params) as AgentMessage;
|
||||
}
|
||||
|
||||
type CliAgentEndHookParams = Parameters<typeof runAgentEndSideEffects>[0];
|
||||
|
||||
function shouldAwaitCliAgentEndHook(params: RunCliAgentParams): boolean {
|
||||
return !params.messageChannel && !params.messageProvider;
|
||||
}
|
||||
|
||||
export async function runCliAgentEndHook(
|
||||
params: RunCliAgentParams,
|
||||
hookParams: CliAgentEndHookParams,
|
||||
): Promise<void> {
|
||||
if (shouldAwaitCliAgentEndHook(params)) {
|
||||
await awaitAgentEndSideEffects(hookParams);
|
||||
return;
|
||||
}
|
||||
runAgentEndSideEffects(hookParams);
|
||||
}
|
||||
|
||||
export async function persistApprovedCliUserTurnTranscript(
|
||||
params: RunCliAgentParams,
|
||||
): Promise<boolean> {
|
||||
const recorder = params.userTurnTranscriptRecorder;
|
||||
const reusingPersistedTurn = params.suppressNextUserMessagePersistence === true;
|
||||
if (!recorder || (reusingPersistedTurn && !recorder.hasPersisted())) {
|
||||
return recorder?.isBlocked() === true;
|
||||
}
|
||||
|
||||
const persisted = await recorder.persistApproved({
|
||||
cwd: params.cwd ?? params.workspaceDir,
|
||||
});
|
||||
if (!persisted && !recorder.hasPersisted() && (await recorder.resolveMessage())) {
|
||||
// A prepared user row can be rejected by before_message_write. Preserve
|
||||
// that terminal decision so outer transcript mirrors do not retry it.
|
||||
recorder.markBlocked();
|
||||
}
|
||||
if (persisted && !reusingPersistedTurn) {
|
||||
try {
|
||||
const notification = params.onUserMessagePersisted?.(persisted.message);
|
||||
if (notification) {
|
||||
void Promise.resolve(notification).catch((error: unknown) => {
|
||||
log.warn(`CLI user turn persistence notification failed: ${formatErrorMessage(error)}`);
|
||||
});
|
||||
}
|
||||
} catch (error) {
|
||||
log.warn(`CLI user turn persistence notification failed: ${formatErrorMessage(error)}`);
|
||||
}
|
||||
}
|
||||
return persisted !== undefined || recorder.hasPersisted() || recorder.isBlocked();
|
||||
}
|
||||
|
||||
export async function persistCliAssistantTranscript(params: {
|
||||
runParams: RunCliAgentParams;
|
||||
text: string;
|
||||
modelId: string;
|
||||
usage?: {
|
||||
input?: number;
|
||||
output?: number;
|
||||
cacheRead?: number;
|
||||
cacheWrite?: number;
|
||||
total?: number;
|
||||
};
|
||||
}): Promise<{
|
||||
owned: boolean;
|
||||
terminalAnchor?: import("../../config/sessions/session-accessor.js").TranscriptEntryAnchor;
|
||||
}> {
|
||||
const { runParams } = params;
|
||||
if (runParams.currentInboundEventKind === "room_event") {
|
||||
const admission = runParams.userTurnTranscriptRecorder?.getAdmissionReceipt();
|
||||
return {
|
||||
owned: true,
|
||||
...(admission ? { terminalAnchor: admission } : {}),
|
||||
};
|
||||
}
|
||||
if (!params.text) {
|
||||
const admission = runParams.userTurnTranscriptRecorder?.getAdmissionReceipt();
|
||||
return {
|
||||
owned: false,
|
||||
...(admission ? { terminalAnchor: admission } : {}),
|
||||
};
|
||||
}
|
||||
if (!runParams.persistAssistantTranscript || !runParams.sessionKey) {
|
||||
return { owned: false };
|
||||
}
|
||||
try {
|
||||
const result = await appendExactAssistantMessageToSessionTranscript({
|
||||
sessionKey: runParams.sessionKey,
|
||||
agentId: runParams.agentId,
|
||||
expectedSessionId: runParams.sessionId,
|
||||
...(runParams.expectedLifecycleRevision !== undefined
|
||||
? { expectedLifecycleRevision: runParams.expectedLifecycleRevision }
|
||||
: {}),
|
||||
...(runParams.expectedWriterRunId !== undefined
|
||||
? { expectedWriterRunId: runParams.expectedWriterRunId }
|
||||
: {}),
|
||||
storePath: runParams.storePath,
|
||||
idempotencyKey: `cli-assistant:${runParams.runId}`,
|
||||
config: runParams.config,
|
||||
beforeMessageWrite: runAgentHarnessBeforeMessageWriteHook,
|
||||
message: buildAssistantMessage({
|
||||
model: {
|
||||
api: "cli",
|
||||
provider: runParams.provider,
|
||||
id: params.modelId,
|
||||
},
|
||||
content: [{ type: "text", text: params.text }],
|
||||
stopReason: "stop",
|
||||
usage: buildUsageWithNoCost({
|
||||
input: params.usage?.input,
|
||||
output: params.usage?.output,
|
||||
cacheRead: params.usage?.cacheRead,
|
||||
cacheWrite: params.usage?.cacheWrite,
|
||||
totalTokens: params.usage?.total,
|
||||
}),
|
||||
}),
|
||||
});
|
||||
if (!result.ok) {
|
||||
log.warn(`CLI assistant transcript persistence skipped: ${result.reason}`);
|
||||
return { owned: result.code === "blocked" || result.code === "session-rebound" };
|
||||
}
|
||||
return { owned: true, ...(result.anchor ? { terminalAnchor: result.anchor } : {}) };
|
||||
} catch (error) {
|
||||
log.warn(`CLI assistant transcript persistence failed: ${formatErrorMessage(error)}`);
|
||||
return { owned: false };
|
||||
}
|
||||
}
|
||||
|
||||
async function notifyCliUserMessagePersisted(
|
||||
params: RunCliAgentParams,
|
||||
message: Extract<AgentMessage, { role: "user" }>,
|
||||
context: string,
|
||||
): Promise<void> {
|
||||
try {
|
||||
await Promise.resolve(params.onUserMessagePersisted?.(message));
|
||||
} catch (err) {
|
||||
log.warn(`${context} notification failed: ${formatErrorMessage(err)}`);
|
||||
}
|
||||
}
|
||||
|
||||
export async function persistCliRunBlock(
|
||||
params: RunCliAgentParams,
|
||||
block: { message: string; pluginId: string },
|
||||
): Promise<void> {
|
||||
const nowMs = Date.now();
|
||||
const redactedUserMessage = {
|
||||
role: "user" as const,
|
||||
content: [{ type: "text" as const, text: block.message }],
|
||||
timestamp: nowMs,
|
||||
idempotencyKey: `hook-block:before_agent_run:user:${params.runId}`,
|
||||
__openclaw: {
|
||||
beforeAgentRunBlocked: {
|
||||
blockedBy: block.pluginId,
|
||||
blockedAt: nowMs,
|
||||
},
|
||||
},
|
||||
};
|
||||
try {
|
||||
const persisted = await params.userTurnTranscriptRecorder?.persistBlocked(redactedUserMessage);
|
||||
if (persisted) {
|
||||
await notifyCliUserMessagePersisted(
|
||||
params,
|
||||
persisted.message,
|
||||
"before_agent_run block user-turn persistence",
|
||||
);
|
||||
return;
|
||||
}
|
||||
} catch (err) {
|
||||
log.warn(
|
||||
`before_agent_run block: failed to persist canonical CLI user message: ${formatErrorMessage(
|
||||
err,
|
||||
)}`,
|
||||
);
|
||||
}
|
||||
|
||||
try {
|
||||
const sessionKey = params.sessionKey?.trim() || params.sessionId;
|
||||
const agentId = params.agentId ?? resolveAgentIdFromSessionKey(sessionKey);
|
||||
let sessionManager = params.sessionManager;
|
||||
if (!sessionManager) {
|
||||
const sessionTarget = params.sessionTarget ?? {
|
||||
agentId,
|
||||
sessionId: params.sessionId,
|
||||
sessionKey,
|
||||
storePath:
|
||||
params.storePath ??
|
||||
resolveSessionStorePathCore(params.config?.session?.store, {
|
||||
agentId,
|
||||
}),
|
||||
};
|
||||
const persistedEntry = await patchSessionEntryCore(
|
||||
sessionTarget,
|
||||
(entry, patchContext) => {
|
||||
if (patchContext.existingEntry && entry.sessionId !== sessionTarget.sessionId) {
|
||||
return null;
|
||||
}
|
||||
return {
|
||||
sessionId: sessionTarget.sessionId,
|
||||
updatedAt: Date.now(),
|
||||
};
|
||||
},
|
||||
{
|
||||
fallbackEntry: params.sessionEntry
|
||||
? undefined
|
||||
: { sessionId: sessionTarget.sessionId, updatedAt: Date.now() },
|
||||
skipMaintenance: true,
|
||||
},
|
||||
);
|
||||
if (persistedEntry?.sessionId !== sessionTarget.sessionId) {
|
||||
// Skip only this stale blocked-message write; the outer runner still returns blocked.
|
||||
return;
|
||||
}
|
||||
sessionManager = SessionManager.open(sessionTarget);
|
||||
}
|
||||
sessionManager.appendMessage(
|
||||
redactedUserMessage as Parameters<typeof sessionManager.appendMessage>[0],
|
||||
);
|
||||
sessionManager.flushPendingPersistence();
|
||||
} catch (err) {
|
||||
log.warn(
|
||||
`before_agent_run block: failed to persist redacted CLI user message: ${formatErrorMessage(
|
||||
err,
|
||||
)}`,
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
export async function finalizeCliContextEngineTurn(params: {
|
||||
context: PreparedCliRunContext;
|
||||
historyMessages: unknown[];
|
||||
assistantText: string;
|
||||
terminalAnchor?: import("../../config/sessions/session-accessor.js").TranscriptEntryAnchor;
|
||||
output: CliOutput;
|
||||
}): Promise<void> {
|
||||
const { context } = params;
|
||||
if (!context.contextEngine) {
|
||||
return;
|
||||
}
|
||||
|
||||
const { params: runParams } = context;
|
||||
const prePromptMessages = params.historyMessages.filter(isAgentMessage);
|
||||
const turnMessages: AgentMessage[] = [];
|
||||
if (context.contextEngineTurnPrompt) {
|
||||
turnMessages.push(buildCliContextEngineUserMessage(context.contextEngineTurnPrompt));
|
||||
}
|
||||
if (params.assistantText) {
|
||||
turnMessages.push(
|
||||
buildCliContextEngineAssistantMessage({
|
||||
text: params.assistantText,
|
||||
provider: runParams.provider,
|
||||
model: context.modelId,
|
||||
usage: params.output.usage,
|
||||
}),
|
||||
);
|
||||
}
|
||||
|
||||
const contextEngineHostSupport = buildGenericCliContextEngineHostSupport({
|
||||
backendId: context.backendResolved.id,
|
||||
});
|
||||
const finalizeTurn = async (transcript: {
|
||||
messagesSnapshot: AgentMessage[];
|
||||
prePromptMessageCount: number;
|
||||
sessionManager?: SessionManager;
|
||||
withSessionManagerRewriteLock: <T>(operation: () => Promise<T> | T) => Promise<T>;
|
||||
}) => {
|
||||
let deferredTurnMaintenance: Promise<void> | undefined;
|
||||
const result = await finalizeHarnessContextEngineTurn({
|
||||
contextEngine: context.contextEngine,
|
||||
promptError: false,
|
||||
aborted: runParams.abortSignal?.aborted === true,
|
||||
yieldAborted: false,
|
||||
sessionIdUsed: runParams.sessionId,
|
||||
sessionKey: runParams.sessionKey,
|
||||
sessionFile: runParams.sessionFile,
|
||||
isHeartbeat: isHeartbeatLifecycleRunKind(runParams.bootstrapContextRunKind),
|
||||
messagesSnapshot: transcript.messagesSnapshot,
|
||||
prePromptMessageCount: transcript.prePromptMessageCount,
|
||||
sessionManager: transcript.sessionManager,
|
||||
config: context.contextEngineConfig,
|
||||
contextEngineHostSupport,
|
||||
providerId: runParams.provider,
|
||||
modelId: context.modelId,
|
||||
runMaintenance: async (maintenanceParams) =>
|
||||
await runHarnessContextEngineMaintenance({
|
||||
...maintenanceParams,
|
||||
withSessionManagerRewriteLock: transcript.withSessionManagerRewriteLock,
|
||||
onDeferredMaintenance: (promise) => {
|
||||
deferredTurnMaintenance = promise;
|
||||
},
|
||||
}),
|
||||
warn: (message) => log.warn(message),
|
||||
});
|
||||
if (result.postTurnFinalizationSucceeded && deferredTurnMaintenance) {
|
||||
context.contextEngineDeferredTurnMaintenance = deferredTurnMaintenance;
|
||||
}
|
||||
};
|
||||
const admission = runParams.userTurnTranscriptRecorder?.getAdmissionReceipt();
|
||||
if (runParams.onContextEngineTurnCandidate) {
|
||||
if (admission && params.terminalAnchor) {
|
||||
runParams.onContextEngineTurnCandidate({
|
||||
boundary: { admission, terminal: params.terminalAnchor },
|
||||
sessionIdUsed: runParams.sessionId,
|
||||
sessionKey: runParams.sessionKey,
|
||||
sessionTarget: runParams.sessionTarget,
|
||||
sessionFile: runParams.sessionFile,
|
||||
promptError: false,
|
||||
aborted: runParams.abortSignal?.aborted === true,
|
||||
yieldAborted: false,
|
||||
contextEngineHostSupport,
|
||||
providerId: runParams.provider,
|
||||
modelId: context.modelId,
|
||||
config: context.contextEngineConfig,
|
||||
isHeartbeat: isHeartbeatLifecycleRunKind(runParams.bootstrapContextRunKind),
|
||||
});
|
||||
}
|
||||
} else {
|
||||
await finalizeTurn({
|
||||
messagesSnapshot: [...prePromptMessages, ...turnMessages],
|
||||
prePromptMessageCount: prePromptMessages.length,
|
||||
withSessionManagerRewriteLock: async (operation) => await operation(),
|
||||
});
|
||||
}
|
||||
}
|
||||
@@ -650,7 +650,10 @@ describe("Code Mode wait, scope, and suspended runs", () => {
|
||||
);
|
||||
expect(first.status).toBe("waiting");
|
||||
expect(first.output).toEqual([{ type: "text", text: "before timeout" }]);
|
||||
expect(first.pendingToolCalls).toEqual([expect.objectContaining({ method: "callValue" })]);
|
||||
// The fast call may settle as the snapshot is parked, but the slow call must remain pending.
|
||||
expect(first.pendingToolCalls).toContainEqual(
|
||||
expect.objectContaining({ id: "bridge:callValue:2", method: "callValue" }),
|
||||
);
|
||||
const runId = first.runId;
|
||||
expect(typeof runId).toBe("string");
|
||||
if (typeof runId !== "string") {
|
||||
|
||||
Reference in New Issue
Block a user