refactor(auto-reply): split agent runner execution (#107985)

* refactor(auto-reply): split agent runner execution

* style(auto-reply): satisfy promise executor lint

* fix(ci): align refactor baselines and test types

* fix(ci): prune stale max-lines baseline
This commit is contained in:
Peter Steinberger
2026-07-14 22:20:25 -07:00
committed by GitHub
parent 9384a40393
commit eebc0dbc33
20 changed files with 3381 additions and 2983 deletions

View File

@@ -608,7 +608,6 @@ src/auto-reply/commands-registry.shared.ts
src/auto-reply/inbound.test.ts
src/auto-reply/reply/abort.test.ts
src/auto-reply/reply/agent-runner-execution.test.ts
src/auto-reply/reply/agent-runner-execution.ts
src/auto-reply/reply/agent-runner-memory.test.ts
src/auto-reply/reply/agent-runner-memory.ts
src/auto-reply/reply/agent-runner-payloads.test.ts

View File

@@ -170,8 +170,7 @@ export const KNIP_UNUSED_EXPORT_BASELINE = [
"src/audit/message-audit-events.ts: resetMessageAuditEventsForTest",
"src/auto-reply/reply/abort.ts: testing",
"src/auto-reply/reply/acp-reset-target.ts: testing",
"src/auto-reply/reply/agent-runner-execution.ts: buildContextOverflowRecoveryText",
"src/auto-reply/reply/agent-runner-execution.ts: computeContextAwareReserveTokensFloor",
"src/auto-reply/reply/agent-runner-context-recovery.ts: computeContextAwareReserveTokensFloor",
"src/auto-reply/reply/agent-runner-memory.ts: setAgentRunnerMemoryTestDeps",
"src/auto-reply/reply/agent-runner-session-reset.ts: setAgentRunnerSessionResetTestDeps",
"src/auto-reply/reply/commands-login.ts: testing",

View File

@@ -6,7 +6,7 @@ import { areRuntimeModelRefsEquivalent } from "../agents/model-runtime-aliases.j
import type { OpenClawConfig } from "../config/types.openclaw.js";
import type { FallbackNoticeState } from "../status/fallback-notice-state.js";
import { formatProviderModelRef } from "./model-runtime.js";
import type { RuntimeFallbackAttempt } from "./reply/agent-runner-execution.js";
import type { RuntimeFallbackAttempt } from "./reply/agent-runner-execution.types.js";
const FALLBACK_REASON_PART_MAX = 80;
const TRANSIENT_FALLBACK_REASONS = new Set([

View File

@@ -0,0 +1,75 @@
import { normalizeOptionalString } from "@openclaw/normalization-core/string-coerce";
import {
hasSessionAutoModelFallbackProvenance,
resolveAutoFallbackPrimaryProbe,
} from "../../agents/agent-scope.js";
import { resolvePersistedOverrideModelRef } from "../../agents/model-selection.js";
import type { SessionEntry } from "../../config/sessions.js";
import type { FollowupRun } from "./queue.js";
/** Decides whether to retry after rechecking auto-fallback primary probe state. */
export function resolveRunAfterAutoFallbackPrimaryProbeRecheck(params: {
run: FollowupRun["run"];
entry?: SessionEntry;
sessionKey?: string;
}): FollowupRun["run"] {
const probe = params.run.autoFallbackPrimaryProbe;
if (!probe || !params.sessionKey || !params.entry) {
return params.run;
}
const resolveEntrySelectionRun = (): FollowupRun["run"] => {
const entryRef = resolvePersistedOverrideModelRef({
defaultProvider: params.run.provider,
overrideProvider: params.entry?.providerOverride,
overrideModel: params.entry?.modelOverride,
});
const hasEntryModelOverride = Boolean(entryRef);
const authProfileId = normalizeOptionalString(params.entry?.authProfileOverride);
const fallbackRun: FollowupRun["run"] = {
...params.run,
provider: entryRef?.provider ?? params.run.provider,
model: entryRef?.model ?? params.run.model,
autoFallbackPrimaryProbe: undefined,
};
if (hasEntryModelOverride) {
fallbackRun.hasSessionModelOverride = true;
fallbackRun.hasAutoFallbackProvenance =
hasSessionAutoModelFallbackProvenance(params.entry) || undefined;
} else {
delete fallbackRun.hasSessionModelOverride;
delete fallbackRun.hasAutoFallbackProvenance;
}
if (hasEntryModelOverride && params.entry?.modelOverrideSource) {
fallbackRun.modelOverrideSource = params.entry.modelOverrideSource;
} else {
delete fallbackRun.modelOverrideSource;
}
if (hasEntryModelOverride && authProfileId) {
fallbackRun.authProfileId = authProfileId;
if (params.entry?.authProfileOverrideSource) {
fallbackRun.authProfileIdSource = params.entry.authProfileOverrideSource;
} else {
delete fallbackRun.authProfileIdSource;
}
} else if (hasEntryModelOverride) {
delete fallbackRun.authProfileId;
delete fallbackRun.authProfileIdSource;
}
return fallbackRun;
};
const refreshedProbe = resolveAutoFallbackPrimaryProbe({
entry: params.entry,
sessionKey: params.sessionKey,
primaryProvider: probe.provider,
primaryModel: probe.model,
});
if (!refreshedProbe) {
return resolveEntrySelectionRun();
}
return {
...params.run,
provider: refreshedProbe.provider,
model: refreshedProbe.model,
autoFallbackPrimaryProbe: refreshedProbe,
};
}

View File

@@ -0,0 +1,334 @@
import { resolveBootstrapWarningSignaturesSeen } from "../../agents/bootstrap-budget.js";
import type { BootstrapContextRunKind } from "../../agents/bootstrap-mode.js";
import type { RunCliAgentParams } from "../../agents/cli-runner/types.js";
import { getCliSessionBinding } from "../../agents/cli-session.js";
import type { RunEmbeddedAgentParams } from "../../agents/embedded-agent-runner/run/params.js";
import type { FastModeAutoProgressState } from "../../agents/fast-mode.js";
import {
AGENT_RUN_RESTART_ABORT_STOP_REASON,
resolveAgentRunErrorLifecycleFields,
} from "../../agents/run-termination.js";
import { withLocalSessionPlacementTurnAdmission } from "../../agents/session-placement-admission.js";
import type { OpenClawConfig } from "../../config/types.openclaw.js";
import type { ThinkLevel } from "../thinking.js";
import type { ReplyPayload } from "../types.js";
import {
createAgentLifecycleTerminalBackstop,
type AgentLifecycleTerminalBackstop,
} from "./agent-lifecycle-terminal.js";
import { resolveRunAuthProfile } from "./agent-runner-auth-profile.js";
import {
clearDroppedCliSessionBinding,
createCliReasoningStreamBridge,
createCliToolSummaryTracker,
keepCliSessionBindingOnlyWhenReused,
runCliAgentWithLifecycle,
} from "./agent-runner-cli-dispatch.js";
import type { AgentTurnParams } from "./agent-runner-execution.types.js";
import type { createAgentTurnPresentation } from "./agent-runner-presentation.js";
import type { AgentTurnTimingTracker } from "./agent-runner-turn-timing.js";
import { shouldBridgeCliPreambleEvents } from "./get-reply.types.js";
import { hasInboundAudio } from "./inbound-media.js";
import { resolveOriginMessageProvider } from "./origin-routing.js";
import type { FollowupRun } from "./queue.js";
import { isReplyOperationRestartAbort } from "./reply-operation-abort.js";
type CliPresentation = Pick<
ReturnType<typeof createAgentTurnPresentation>,
"handlePartialForTyping" | "preparePartialForTyping" | "startPresentationWhileTyping"
>;
export async function runCliFallbackCandidate(params: {
turn: AgentTurnParams;
candidateRun: FollowupRun["run"];
runtimeConfig: OpenClawConfig;
provider: string;
model: string;
cliExecutionProvider: string;
candidateThinkLevel?: ThinkLevel;
candidateFastMode: Pick<RunCliAgentParams, "fastMode" | "fastModeAutoOnSeconds">;
runId: string;
lifecycleGeneration: string;
runAbortSignal?: AbortSignal;
runLane: RunCliAgentParams["lane"];
isFinalFallbackAttempt?: boolean;
suppressQueuedUserPersistenceForCandidate: boolean;
userTurnTranscriptRecorder: RunCliAgentParams["userTurnTranscriptRecorder"];
notifyUserMessagePersisted: () => void;
fastModeStartedAtMs: number;
fastModeAutoProgressState: FastModeAutoProgressState;
bootstrapContextRunKind: BootstrapContextRunKind;
bootstrapPromptWarningSignaturesSeen: string[];
currentTurnImages: Awaited<
ReturnType<typeof import("./current-turn-images.js").resolveCurrentTurnImages>
>;
signalExecutionPhaseForTyping: NonNullable<RunEmbeddedAgentParams["onExecutionPhase"]>;
notifyAgentRunStart: () => void;
preserveProgressCallbackStartOrder: boolean;
presentation: CliPresentation;
timing: AgentTurnTimingTracker;
onLifecycleBackstop: (backstop: AgentLifecycleTerminalBackstop) => void;
}): Promise<{
result: Awaited<ReturnType<typeof runCliAgentWithLifecycle>>;
bootstrapPromptWarningSignaturesSeen: string[];
}> {
const turn = params.turn;
const cliSessionBinding = getCliSessionBinding(
turn.getActiveSessionEntry(),
params.cliExecutionProvider,
);
const cliLifecycleStartedAt = Date.now();
const lifecycleBackstop = createAgentLifecycleTerminalBackstop({
runId: params.runId,
sessionKey: turn.sessionKey,
startedAt: cliLifecycleStartedAt,
getLifecycleGeneration: () => params.lifecycleGeneration,
resolveTerminationFields: (error) => ({
...resolveAgentRunErrorLifecycleFields(error, params.runAbortSignal),
...(isReplyOperationRestartAbort(turn.replyOperation)
? {
aborted: true as const,
stopReason: AGENT_RUN_RESTART_ABORT_STOP_REASON,
}
: {}),
}),
});
params.onLifecycleBackstop(lifecycleBackstop);
const authProfile = resolveRunAuthProfile(params.candidateRun, params.cliExecutionProvider, {
config: params.runtimeConfig,
});
let droppedCliSessionReplacement = false;
const hookMessageProvider = resolveOriginMessageProvider({
originatingChannel: turn.followupRun.originatingChannel,
provider: turn.sessionCtx.Provider,
});
const cliCurrentThreadId =
turn.followupRun.originatingThreadId ?? turn.sessionCtx.MessageThreadId;
const isRestartSentinelContinuation =
turn.sessionCtx.InputProvenance?.kind === "internal_system" &&
turn.sessionCtx.InputProvenance.sourceTool === "restart-sentinel";
const cliCurrentMessageId = isRestartSentinelContinuation
? turn.sessionCtx.ReplyToId
: (turn.sessionCtx.MessageSidFull ?? turn.sessionCtx.MessageSid);
const cliToolSummaryTracker = createCliToolSummaryTracker({
detailMode: turn.toolProgressDetail,
shouldEmitToolResult: turn.shouldEmitToolResult,
shouldEmitToolOutput: turn.shouldEmitToolOutput,
deliver: async (payload) => {
await turn.opts?.onToolResult?.(payload);
},
});
const result = await params.timing.measure("cli_run", () =>
withLocalSessionPlacementTurnAdmission(
{
sessionId: turn.followupRun.run.sessionId,
sessionKey: turn.sessionKey,
agentId: turn.followupRun.run.agentId,
runId: params.runId,
},
() =>
runCliAgentWithLifecycle({
runId: params.runId,
lifecycleGeneration: params.lifecycleGeneration,
provider: params.cliExecutionProvider,
startedAt: cliLifecycleStartedAt,
emitLifecycleTerminal: false,
onAgentRunStart: params.notifyAgentRunStart,
suppressAssistantBridge: turn.followupRun.run.silentExpected,
onActivity: () => turn.replyOperation?.recordActivity(),
preserveProgressCallbackStartOrder: params.preserveProgressCallbackStartOrder,
onAssistantText: async (text) => {
if (!params.preserveProgressCallbackStartOrder) {
const textForTyping = await params.presentation.handlePartialForTyping({
text,
} as ReplyPayload);
if (textForTyping === undefined || !turn.opts?.onPartialReply) {
return;
}
await turn.opts.onPartialReply({ text: textForTyping });
return;
}
const textForTyping = params.presentation.preparePartialForTyping({
text,
} as ReplyPayload);
if (textForTyping === undefined) {
return;
}
// Assistant and tool CLI bridges drain independently. Stage presentation first.
await params.presentation.startPresentationWhileTyping(
turn.typingSignals.signalTextDelta(textForTyping),
() => turn.opts?.onPartialReply?.({ text: textForTyping }),
);
},
onReasoningText: createCliReasoningStreamBridge(turn.opts?.onReasoningStream),
onReasoningProgress: async (payload) => {
await turn.opts?.onReasoningProgress?.(payload);
},
onToolEvent: async (payload) => {
if (!params.preserveProgressCallbackStartOrder) {
await cliToolSummaryTracker.noteToolEvent(payload);
if (payload.phase === "result") {
return;
}
const { name, phase, args } = payload;
await Promise.all([
turn.typingSignals.signalToolStart(),
turn.opts?.onToolStart?.({
name,
phase,
args,
detailMode: turn.toolProgressDetail,
}),
]);
return;
}
const summaryPromise = cliToolSummaryTracker.noteToolEvent(payload);
if (payload.phase === "result") {
await summaryPromise;
return;
}
const { name, phase, args } = payload;
// Tool and assistant bridges drain independently. Preserve source order.
await Promise.all([
summaryPromise,
params.presentation.startPresentationWhileTyping(
turn.typingSignals.signalToolStart(),
() =>
turn.opts?.onToolStart?.({
name,
phase,
args,
detailMode: turn.toolProgressDetail,
}),
),
]);
},
onCommentaryText:
turn.opts?.onItemEvent && shouldBridgeCliPreambleEvents(turn.opts)
? async (payload) => {
await turn.opts?.onItemEvent?.({
itemId: payload.itemId,
kind: "preamble",
progressText: payload.text,
});
}
: undefined,
onFastModeAutoProgress: async (payload) => {
await turn.opts?.onToolResult?.(payload);
},
transformResult:
turn.followupRun.currentInboundEventKind === "room_event"
? (resultLocal) =>
keepCliSessionBindingOnlyWhenReused({
result: resultLocal,
existingSessionId: cliSessionBinding?.sessionId,
onDroppedReplacement: () => {
droppedCliSessionReplacement = true;
},
})
: undefined,
runParams: {
sessionId: turn.followupRun.run.sessionId,
sessionKey: turn.sessionKey,
runtimePolicySessionKey:
turn.followupRun.run.runtimePolicySessionKey ?? turn.runtimePolicySessionKey,
agentId: turn.followupRun.run.agentId,
trigger: turn.isHeartbeat ? "heartbeat" : "user",
sessionFile: turn.followupRun.run.sessionFile,
workspaceDir: turn.followupRun.run.workspaceDir,
cwd: turn.followupRun.run.cwd,
config: params.runtimeConfig,
prompt: turn.commandBody,
transcriptPrompt: turn.transcriptCommandBody,
suppressNextUserMessagePersistence: params.suppressQueuedUserPersistenceForCandidate,
userTurnTranscriptRecorder: params.userTurnTranscriptRecorder,
onUserMessagePersisted: params.notifyUserMessagePersisted,
persistAssistantTranscript:
turn.followupRun.currentInboundEventKind !== "room_event" &&
turn.followupRun.run.suppressTranscriptOnlyAssistantPersistence !== true,
storePath: turn.storePath,
currentInboundEventKind: turn.followupRun.currentInboundEventKind,
currentInboundContext: turn.followupRun.currentInboundContext,
inputProvenance: turn.followupRun.run.inputProvenance,
modelProvider: params.provider,
provider: params.cliExecutionProvider,
execOverrides: turn.followupRun.run.execOverrides,
bashElevated: turn.followupRun.run.bashElevated,
model: params.model,
thinkLevel: params.candidateThinkLevel,
fastMode: params.candidateFastMode.fastMode,
fastModeStartedAtMs: params.fastModeStartedAtMs,
fastModeAutoOnSeconds: params.candidateFastMode.fastModeAutoOnSeconds,
fastModeAutoProgressState: params.fastModeAutoProgressState,
isFinalFallbackAttempt: params.isFinalFallbackAttempt,
timeoutMs: turn.followupRun.run.timeoutMs,
runTimeoutOverrideMs: turn.followupRun.run.runTimeoutOverrideMs,
runId: params.runId,
lane: params.runLane,
extraSystemPrompt: turn.followupRun.run.extraSystemPrompt,
sourceReplyDeliveryMode: turn.followupRun.run.sourceReplyDeliveryMode,
taskSuggestionDeliveryMode: turn.followupRun.run.taskSuggestionDeliveryMode,
silentReplyPromptMode: turn.followupRun.run.silentReplyPromptMode,
allowEmptyAssistantReplyAsSilent: turn.followupRun.run.allowEmptyAssistantReplyAsSilent,
extraSystemPromptStatic: turn.followupRun.run.extraSystemPromptStatic,
cliSessionBindingFacts: turn.followupRun.run.cliSessionBindingFacts,
ownerNumbers: turn.followupRun.run.ownerNumbers,
cliSessionId: cliSessionBinding?.sessionId,
cliSessionBinding,
authProfileId: authProfile.authProfileId,
bootstrapContextMode: turn.opts?.bootstrapContextMode,
bootstrapContextRunKind: params.bootstrapContextRunKind,
bootstrapPromptWarningSignaturesSeen: params.bootstrapPromptWarningSignaturesSeen,
bootstrapPromptWarningSignature:
params.bootstrapPromptWarningSignaturesSeen[
params.bootstrapPromptWarningSignaturesSeen.length - 1
],
images: params.currentTurnImages.images,
imageOrder: params.currentTurnImages.imageOrder,
skillsSnapshot: turn.followupRun.run.skillsSnapshot,
messageChannel: turn.followupRun.originatingChannel ?? undefined,
messageProvider: hookMessageProvider,
clientCaps: turn.followupRun.run.clientCaps,
currentChannelId:
turn.followupRun.originatingTo ?? turn.sessionCtx.OriginatingTo ?? turn.sessionCtx.To,
senderId: turn.followupRun.run.senderId,
senderName: turn.followupRun.run.senderName,
senderUsername: turn.followupRun.run.senderUsername,
senderE164: turn.followupRun.run.senderE164,
groupId: turn.followupRun.run.groupId,
groupChannel: turn.followupRun.run.groupChannel,
groupSpace: turn.followupRun.run.groupSpace,
spawnedBy: turn.followupRun.run.spawnedBy,
chatId: turn.followupRun.originatingChatId,
channelContext: turn.followupRun.run.channelContext,
currentThreadTs: cliCurrentThreadId != null ? String(cliCurrentThreadId) : undefined,
currentMessageId: cliCurrentMessageId,
currentInboundAudio: hasInboundAudio(turn.sessionCtx),
agentAccountId: turn.followupRun.run.agentAccountId,
senderIsOwner: turn.followupRun.run.senderIsOwner,
approvalReviewerDeviceId: turn.followupRun.run.approvalReviewerDeviceId,
toolsAllow: turn.opts?.toolsAllow,
disableTools: turn.opts?.disableTools,
abortSignal: params.runAbortSignal,
onExecutionPhase: params.signalExecutionPhaseForTyping,
replyOperation: turn.replyOperation,
},
}),
),
);
if (droppedCliSessionReplacement) {
await clearDroppedCliSessionBinding({
provider: params.cliExecutionProvider,
sessionKey: turn.sessionKey,
sessionStore: turn.activeSessionStore,
storePath: turn.storePath,
activeSessionEntry: turn.getActiveSessionEntry(),
});
}
return {
result,
bootstrapPromptWarningSignaturesSeen: resolveBootstrapWarningSignaturesSeen(
result.meta?.systemPromptReport,
),
};
}

View File

@@ -0,0 +1,82 @@
import {
normalizeLowercaseStringOrEmpty,
readStringValue,
} from "@openclaw/normalization-core/string-coerce";
import type { GetReplyOptions } from "../types.js";
function readRecordValue(value: unknown): Record<string, unknown> | undefined {
return value && typeof value === "object" && !Array.isArray(value)
? (value as Record<string, unknown>)
: undefined;
}
function readFiniteNumberValue(value: unknown): number | undefined {
return typeof value === "number" && Number.isFinite(value) ? value : undefined;
}
function readNullableNumberValue(value: unknown): number | null | undefined {
if (value === null) {
return null;
}
return readFiniteNumberValue(value);
}
function isCommandToolName(name: string | undefined): boolean {
const normalized = normalizeLowercaseStringOrEmpty(name);
return normalized === "exec" || normalized === "bash" || normalized === "shell";
}
/** Projects a completed command-tool event into the channel command-output contract. */
export function buildCommandOutputFromToolResultEvent(evt: {
stream: string;
data: Record<string, unknown>;
}): Parameters<NonNullable<GetReplyOptions["onCommandOutput"]>>[0] | undefined {
if (evt.stream !== "tool" || readStringValue(evt.data.phase) !== "result") {
return undefined;
}
const name = readStringValue(evt.data.name);
if (!isCommandToolName(name)) {
return undefined;
}
const result = readRecordValue(evt.data.result);
const details = readRecordValue(result?.details);
const output =
readStringValue(evt.data.output) ??
readStringValue(result?.output) ??
readStringValue(details?.output);
const explicitStatus =
readStringValue(evt.data.status) ??
readStringValue(result?.status) ??
readStringValue(details?.status);
const exitCode = readNullableNumberValue(
result?.exitCode ?? details?.exitCode ?? evt.data.exitCode,
);
const durationMs = readFiniteNumberValue(
result?.durationMs ?? details?.durationMs ?? evt.data.durationMs,
);
const cwd = readStringValue(evt.data.cwd);
const hasConcreteCommandResult =
output !== undefined ||
explicitStatus !== undefined ||
exitCode !== undefined ||
durationMs !== undefined ||
cwd !== undefined ||
(result !== undefined && Object.keys(result).length > 0);
if (!hasConcreteCommandResult) {
return undefined;
}
const errorStatus =
evt.data.isError === true ? "failed" : evt.data.isError === false ? "completed" : undefined;
return {
itemId: readStringValue(evt.data.itemId),
phase: "end",
title: readStringValue(evt.data.title),
toolCallId: readStringValue(evt.data.toolCallId),
name,
output,
status: explicitStatus ?? errorStatus,
exitCode,
durationMs,
cwd,
};
}

View File

@@ -0,0 +1,306 @@
import {
normalizeLowercaseStringOrEmpty,
normalizeOptionalString,
} from "@openclaw/normalization-core/string-coerce";
import { resolveContextTokensForModel } from "../../agents/context.js";
import { resolveModelRefFromString } from "../../agents/model-selection.js";
import type { SessionEntry } from "../../config/sessions.js";
import type { FollowupRun } from "./queue.js";
const DEFAULT_RESERVE_TOKENS_FLOOR = 20_000;
/** Computes a reserve-token floor scaled to the selected context window. */
export function computeContextAwareReserveTokensFloor(contextWindow: number | undefined): number {
if (typeof contextWindow !== "number" || contextWindow <= 0) {
return DEFAULT_RESERVE_TOKENS_FLOOR;
}
if (contextWindow >= 1_000_000) {
return 100_000;
}
if (contextWindow >= 200_000) {
return 50_000;
}
if (contextWindow >= 100_000) {
return 35_000;
}
return DEFAULT_RESERVE_TOKENS_FLOOR;
}
function resolveContextWindowForCompactionHint(params: {
cfg: FollowupRun["run"]["config"];
primaryProvider?: string;
primaryModel?: string;
runtimeProvider?: string;
runtimeModel?: string;
agentId?: string;
activeSessionEntry?: SessionEntry;
}): number | undefined {
let modelWindow: number | undefined;
const entryProvider = params.activeSessionEntry?.modelProvider;
const entryModel = params.activeSessionEntry?.model;
const runtimeProvider = params.runtimeProvider ?? entryProvider;
const runtimeModel = params.runtimeModel ?? entryModel;
const hasExplicitRuntimeRef = Boolean(params.runtimeProvider && params.runtimeModel);
if (runtimeProvider && runtimeModel) {
const resolved = resolveContextTokensForModel({
cfg: params.cfg,
provider: runtimeProvider,
model: runtimeModel,
allowAsyncLoad: false,
});
if (typeof resolved === "number" && resolved > 0) {
modelWindow = resolved;
}
}
const sessionWindow = normalizePositiveContextTokens(params.activeSessionEntry?.contextTokens);
const sessionMatchesRuntimeRef = runtimeProvider === entryProvider && runtimeModel === entryModel;
const trustedSessionWindow =
!hasExplicitRuntimeRef || sessionMatchesRuntimeRef ? sessionWindow : undefined;
if (modelWindow === undefined && sessionMatchesRuntimeRef && sessionWindow !== undefined) {
modelWindow = sessionWindow;
}
if (
modelWindow === undefined &&
!hasExplicitRuntimeRef &&
params.primaryProvider &&
params.primaryModel
) {
const resolved = resolveContextTokensForModel({
cfg: params.cfg,
provider: params.primaryProvider,
model: params.primaryModel,
allowAsyncLoad: false,
});
if (typeof resolved === "number" && resolved > 0) {
modelWindow = resolved;
}
}
const contextWindow = modelWindow ?? trustedSessionWindow;
const agentCap = resolveAgentContextTokensForHint({
cfg: params.cfg,
agentId: params.agentId,
});
if (agentCap !== undefined && contextWindow !== undefined) {
return Math.min(agentCap, contextWindow);
}
return agentCap ?? contextWindow;
}
function buildContextOverflowResetHint(contextWindowTokens: number | undefined): string {
const reserveFloor = computeContextAwareReserveTokensFloor(contextWindowTokens);
return (
"\n\nTo prevent this, increase your compaction buffer by setting " +
`\`agents.defaults.compaction.reserveTokensFloor\` to ${reserveFloor} or higher in your config.`
);
}
type ModelRefLike = {
provider: string;
model: string;
};
function resolveAgentHeartbeatModelRaw(params: {
cfg: FollowupRun["run"]["config"];
agentId?: string;
}): string | undefined {
const defaultModel = normalizeOptionalString(params.cfg.agents?.defaults?.heartbeat?.model);
const agentId = normalizeLowercaseStringOrEmpty(params.agentId);
const agentModel = agentId
? normalizeOptionalString(
params.cfg.agents?.list?.find(
(entry) => normalizeLowercaseStringOrEmpty(entry?.id) === agentId,
)?.heartbeat?.model,
)
: undefined;
return agentModel ?? defaultModel;
}
function normalizeModelRefForCompare(ref: ModelRefLike | undefined) {
if (!ref) {
return undefined;
}
const provider = normalizeLowercaseStringOrEmpty(ref.provider);
const model = normalizeLowercaseStringOrEmpty(ref.model);
return provider && model ? { provider, model } : undefined;
}
function modelRefsEqual(left: ModelRefLike | undefined, right: ModelRefLike | undefined) {
const normalizedLeft = normalizeModelRefForCompare(left);
const normalizedRight = normalizeModelRefForCompare(right);
return (
normalizedLeft !== undefined &&
normalizedRight !== undefined &&
normalizedLeft.provider === normalizedRight.provider &&
normalizedLeft.model === normalizedRight.model
);
}
function formatContextWindowLabel(tokens: number): string {
if (tokens >= 1_000_000) {
return `${Math.round((tokens / 1_000_000) * 10) / 10}M`;
}
return `${Math.round(tokens / 1024)}k`;
}
function normalizePositiveContextTokens(value: unknown): number | undefined {
if (typeof value !== "number" || !Number.isFinite(value) || value <= 0) {
return undefined;
}
return Math.floor(value);
}
function resolveAgentContextTokensForHint(params: {
cfg: FollowupRun["run"]["config"];
agentId?: string;
}): number | undefined {
const defaultContextTokens = normalizePositiveContextTokens(
params.cfg.agents?.defaults?.contextTokens,
);
const agentId = normalizeLowercaseStringOrEmpty(params.agentId);
const agentContextTokens = agentId
? normalizePositiveContextTokens(
params.cfg.agents?.list?.find(
(entry) => normalizeLowercaseStringOrEmpty(entry?.id) === agentId,
)?.contextTokens,
)
: undefined;
return agentContextTokens ?? defaultContextTokens;
}
function resolveContextWindowForHint(params: {
cfg: FollowupRun["run"]["config"];
agentId?: string;
ref: ModelRefLike;
activeSessionEntry?: SessionEntry;
}) {
const sessionContextTokens = normalizePositiveContextTokens(
params.activeSessionEntry?.contextTokens,
);
const modelContextTokens = resolveContextTokensForModel({
cfg: params.cfg,
provider: params.ref.provider,
model: params.ref.model,
allowAsyncLoad: false,
});
const contextTokens = modelContextTokens ?? sessionContextTokens;
if (contextTokens === undefined) {
return undefined;
}
const agentContextTokens = resolveAgentContextTokensForHint({
cfg: params.cfg,
agentId: params.agentId,
});
return agentContextTokens !== undefined
? Math.min(agentContextTokens, contextTokens)
: contextTokens;
}
function resolveHeartbeatBleedHint(params: {
cfg: FollowupRun["run"]["config"];
agentId?: string;
primaryProvider?: string;
primaryModel?: string;
activeSessionEntry?: SessionEntry;
}): string | undefined {
const primaryProvider = normalizeOptionalString(params.primaryProvider);
const primaryModel = normalizeOptionalString(params.primaryModel);
const runtimeProvider = normalizeOptionalString(params.activeSessionEntry?.modelProvider);
const runtimeModel = normalizeOptionalString(params.activeSessionEntry?.model);
if (!primaryProvider || !primaryModel || !runtimeProvider || !runtimeModel) {
return undefined;
}
const primaryRef = { provider: primaryProvider, model: primaryModel };
const runtimeRef = { provider: runtimeProvider, model: runtimeModel };
if (modelRefsEqual(primaryRef, runtimeRef)) {
return undefined;
}
const heartbeatModelRaw = resolveAgentHeartbeatModelRaw({
cfg: params.cfg,
agentId: params.agentId,
});
const heartbeatRef = heartbeatModelRaw
? resolveModelRefFromString({
cfg: params.cfg,
raw: heartbeatModelRaw,
defaultProvider: primaryProvider,
})?.ref
: undefined;
if (!modelRefsEqual(runtimeRef, heartbeatRef)) {
return undefined;
}
const runtimeWindow = resolveContextWindowForHint({
cfg: params.cfg,
agentId: params.agentId,
ref: runtimeRef,
activeSessionEntry: params.activeSessionEntry,
});
const primaryWindow = resolveContextWindowForHint({
cfg: params.cfg,
agentId: params.agentId,
ref: primaryRef,
});
if (
typeof runtimeWindow === "number" &&
typeof primaryWindow === "number" &&
runtimeWindow >= primaryWindow
) {
return undefined;
}
const runtimeLabel =
typeof runtimeWindow === "number" && runtimeWindow > 0
? ` (${formatContextWindowLabel(runtimeWindow)} context)`
: "";
return (
`\n\nThe previous heartbeat turn left this session on ${runtimeProvider}/${runtimeModel}` +
`${runtimeLabel} instead of ${primaryProvider}/${primaryModel}. This matches the configured ` +
"`heartbeat.model`, so the overflow is likely heartbeat model bleed rather than a " +
"compaction-buffer problem. Set `heartbeat.isolatedSession: true`, enable " +
"`heartbeat.lightContext: true`, or use a heartbeat model with a larger context window."
);
}
/** Builds recovery instructions for context-overflow failures. */
export function buildContextOverflowRecoveryText(params: {
duringCompaction?: boolean;
preserveSessionMapping?: boolean;
cfg: FollowupRun["run"]["config"];
agentId?: string;
primaryProvider?: string;
primaryModel?: string;
runtimeProvider?: string;
runtimeModel?: string;
activeSessionEntry?: SessionEntry;
}): string {
const prefix = params.preserveSessionMapping
? "⚠️ Auto-compaction could not recover this turn. I kept this conversation mapped to the current session. Please try again, use /compact, or use /new to start a fresh session."
: params.duringCompaction
? "⚠️ Context limit exceeded during compaction. I've reset our conversation to start fresh - please try again."
: "⚠️ Context limit exceeded. I've reset our conversation to start fresh - please try again.";
const primaryContextWindow = resolveContextWindowForCompactionHint({
cfg: params.cfg,
primaryProvider: params.primaryProvider,
primaryModel: params.primaryModel,
runtimeProvider: params.runtimeProvider,
runtimeModel: params.runtimeModel,
agentId: params.agentId,
activeSessionEntry: params.activeSessionEntry,
});
const explicitRuntimeMatchesSession =
!params.runtimeProvider ||
!params.runtimeModel ||
(params.runtimeProvider === params.activeSessionEntry?.modelProvider &&
params.runtimeModel === params.activeSessionEntry?.model);
const heartbeatBleedHint = explicitRuntimeMatchesSession
? resolveHeartbeatBleedHint({
cfg: params.cfg,
agentId: params.agentId,
primaryProvider: params.primaryProvider,
primaryModel: params.primaryModel,
activeSessionEntry: params.activeSessionEntry,
})
: undefined;
return prefix + (heartbeatBleedHint ?? buildContextOverflowResetHint(primaryContextWindow));
}

View File

@@ -0,0 +1,382 @@
import { normalizeOptionalString } from "@openclaw/normalization-core/string-coerce";
import { resolveBootstrapWarningSignaturesSeen } from "../../agents/bootstrap-budget.js";
import type { BootstrapContextRunKind } from "../../agents/bootstrap-mode.js";
import type { RunEmbeddedAgentParams } from "../../agents/embedded-agent-runner/run/params.js";
import { runEmbeddedAgent } from "../../agents/embedded-agent.js";
import type { FastModeAutoProgressState } from "../../agents/fast-mode.js";
import { resolveAgentHarnessPolicy } from "../../agents/harness/policy.js";
import { resolveOpenAIRuntimeProvider } from "../../agents/openai-routing.js";
import {
AGENT_RUN_RESTART_ABORT_STOP_REASON,
resolveAgentRunErrorLifecycleFields,
} from "../../agents/run-termination.js";
import { resolveGroupSessionKey } from "../../config/sessions.js";
import type { OpenClawConfig } from "../../config/types.openclaw.js";
import {
isTrustedMessageActionTurnIngress,
mintMessageActionTurnCapability,
revokeMessageActionTurnCapability,
} from "../../gateway/message-action-turn-capability.js";
import { logVerbose } from "../../globals.js";
import {
isMarkdownCapableMessageChannel,
resolveMessageChannel,
} from "../../utils/message-channel.js";
import type { ThinkLevel } from "../thinking.js";
import type { ReplyPayload } from "../types.js";
import {
createAgentLifecycleTerminalBackstop,
type AgentLifecycleTerminalBackstop,
} from "./agent-lifecycle-terminal.js";
import {
createAgentRunEventHandler,
type MessageToolDeliveryState,
} from "./agent-runner-event-handler.js";
import type { AgentTurnParams } from "./agent-runner-execution.types.js";
import type { createAgentTurnPresentation } from "./agent-runner-presentation.js";
import type { AgentTurnTimingTracker } from "./agent-runner-turn-timing.js";
import { buildEmbeddedRunExecutionParams } from "./agent-runner-utils.js";
import type { FollowupRun } from "./queue.js";
import { isReplyOperationRestartAbort } from "./reply-operation-abort.js";
type EmbeddedPresentation = Pick<
ReturnType<typeof createAgentTurnPresentation>,
| "normalizeStreamingText"
| "preparePartialForTyping"
| "handlePartialForTyping"
| "startPresentationWhileTyping"
| "blockReplyHandler"
>;
export async function runEmbeddedFallbackCandidate(params: {
turn: AgentTurnParams;
effectiveRun: FollowupRun["run"];
candidateRun: FollowupRun["run"];
runtimeConfig: OpenClawConfig;
provider: string;
model: string;
sessionRuntimeOverride?: string;
candidateThinkLevel?: ThinkLevel;
candidateFastMode: Pick<RunEmbeddedAgentParams, "fastMode" | "fastModeAutoOnSeconds">;
runId: string;
getLifecycleGeneration: () => string;
onLifecycleGeneration: (generation: string) => void;
runAbortSignal?: AbortSignal;
allowTransientCooldownProbe?: boolean;
isFinalFallbackAttempt?: boolean;
suppressQueuedUserPersistenceForCandidate: boolean;
suppressAssistantErrorPersistenceForCandidate: boolean;
onAssistantErrorMessagePersisted: () => void;
userTurnTranscriptRecorder: NonNullable<AgentTurnParams["opts"]>["userTurnTranscriptRecorder"];
notifyUserMessagePersisted: () => void;
fastModeStartedAtMs: number;
fastModeAutoProgressState: FastModeAutoProgressState;
bootstrapContextRunKind: BootstrapContextRunKind;
bootstrapPromptWarningSignaturesSeen: string[];
currentTurnImages: Awaited<
ReturnType<typeof import("./current-turn-images.js").resolveCurrentTurnImages>
>;
signalExecutionPhaseForTyping: NonNullable<
Parameters<typeof runEmbeddedAgent>[0]["onExecutionPhase"]
>;
notifyAgentRunStart: () => void;
notifyUserAboutCompaction: boolean;
sourceRepliesAreToolOnly: boolean;
messageToolDeliveryState: MessageToolDeliveryState;
preserveProgressCallbackStartOrder: boolean;
presentation: EmbeddedPresentation;
timing: AgentTurnTimingTracker;
onLifecycleBackstop: (backstop: AgentLifecycleTerminalBackstop) => void;
onCompactionCount: (count: number) => void;
}): Promise<{
result: Awaited<ReturnType<typeof runEmbeddedAgent>>;
bootstrapPromptWarningSignaturesSeen: string[];
}> {
const turn = params.turn;
const { embeddedContext, senderContext, runBaseParams } = buildEmbeddedRunExecutionParams({
run: {
...params.candidateRun,
...params.candidateFastMode,
thinkLevel: params.candidateThinkLevel,
},
replyRoute: turn.followupRun,
sessionCtx: turn.sessionCtx,
hasRepliedRef: turn.opts?.hasRepliedRef,
provider: params.provider,
runId: params.runId,
promptCacheKey: turn.opts?.promptCacheKey,
allowTransientCooldownProbe: params.allowTransientCooldownProbe,
model: params.model,
});
const agentHarnessPolicy = params.sessionRuntimeOverride
? ({ runtime: params.sessionRuntimeOverride, runtimeSource: "model" } as const)
: resolveAgentHarnessPolicy({
provider: params.provider,
modelId: params.model,
config: params.runtimeConfig,
agentId: turn.followupRun.run.agentId,
sessionKey: turn.followupRun.run.runtimePolicySessionKey ?? turn.sessionKey,
});
const embeddedRunProvider = resolveOpenAIRuntimeProvider({
provider: params.provider,
harnessRuntime: agentHarnessPolicy.runtime,
authProfileProvider: runBaseParams.authProfileId?.split(":", 1)[0],
authProfileId: runBaseParams.authProfileId,
config: params.runtimeConfig,
workspaceDir: turn.followupRun.run.workspaceDir,
});
const embeddedRunHarnessOverride =
params.sessionRuntimeOverride ??
(agentHarnessPolicy.runtime === "openclaw" && embeddedRunProvider !== params.provider
? "openclaw"
: undefined);
const messageActionCapabilitySessionKey =
turn.runtimePolicySessionKey ?? embeddedContext.sessionKey;
const messageActionTurnCapability =
isTrustedMessageActionTurnIngress(turn.sessionCtx.Provider) &&
!turn.isHeartbeat &&
embeddedContext.agentId &&
messageActionCapabilitySessionKey &&
embeddedContext.messageProvider &&
embeddedContext.currentChannelId
? mintMessageActionTurnCapability({
agentId: embeddedContext.agentId,
runId: params.runId,
sessionKey: messageActionCapabilitySessionKey,
sessionId: embeddedContext.sessionId,
requesterAccountId: embeddedContext.agentAccountId,
requesterSenderId: senderContext.senderId,
toolContext: {
currentChannelId: embeddedContext.currentChannelId,
currentChatType: embeddedContext.chatType,
currentMessagingTarget: embeddedContext.currentMessagingTarget,
currentGraphChannelId: embeddedContext.currentGraphChannelId,
currentChannelProvider: embeddedContext.currentChannelProvider,
currentThreadTs: embeddedContext.currentThreadTs,
currentMessageId: embeddedContext.currentMessageId,
replyToMode: embeddedContext.replyToMode,
hasRepliedRef: embeddedContext.hasRepliedRef,
sameChannelThreadRequired: embeddedContext.sameChannelThreadRequired,
},
ttlMs: runBaseParams.timeoutMs + 60_000,
})
: undefined;
let attemptCompactionCount = 0;
const lifecycleBackstop = createAgentLifecycleTerminalBackstop({
runId: params.runId,
sessionKey: turn.sessionKey,
getLifecycleGeneration: params.getLifecycleGeneration,
resolveTerminationFields: (error) => ({
...resolveAgentRunErrorLifecycleFields(error, params.runAbortSignal),
...(isReplyOperationRestartAbort(turn.replyOperation)
? {
aborted: true as const,
stopReason: AGENT_RUN_RESTART_ABORT_STOP_REASON,
}
: {}),
}),
});
params.onLifecycleBackstop(lifecycleBackstop);
try {
// Profiler milestone. Exposes pre-dispatch delay without normal-path logging.
params.timing.logMilestoneIfSlow({
runId: params.runId,
sessionId: turn.followupRun.run.sessionId,
sessionKey: turn.sessionKey,
milestone: "before_embedded_run",
});
const result = await params.timing.measure("embedded_run", () =>
runEmbeddedAgent({
...embeddedContext,
messageActionTurnCapability,
lifecycleGeneration: params.getLifecycleGeneration(),
allowGatewaySubagentBinding: true,
trigger: turn.isHeartbeat ? "heartbeat" : "user",
groupId: resolveGroupSessionKey(turn.sessionCtx)?.id,
groupChannel:
normalizeOptionalString(turn.sessionCtx.GroupChannel) ??
normalizeOptionalString(turn.sessionCtx.GroupSubject),
groupSpace: normalizeOptionalString(turn.sessionCtx.GroupSpace),
...senderContext,
...runBaseParams,
provider: embeddedRunProvider,
agentHarnessId: embeddedRunHarnessOverride,
agentHarnessRuntimeOverride: embeddedRunHarnessOverride,
fastModeStartedAtMs: params.fastModeStartedAtMs,
fastModeAutoProgressState: params.fastModeAutoProgressState,
isFinalFallbackAttempt: params.isFinalFallbackAttempt,
sandboxSessionKey: turn.runtimePolicySessionKey,
prompt: turn.commandBody,
transcriptPrompt: turn.transcriptCommandBody,
userTurnTranscriptRecorder: params.userTurnTranscriptRecorder,
currentInboundEventKind: turn.followupRun.currentInboundEventKind,
currentInboundContext: turn.followupRun.currentInboundContext,
extraSystemPrompt: turn.followupRun.run.extraSystemPrompt,
sourceReplyDeliveryMode: turn.followupRun.run.sourceReplyDeliveryMode,
forceMessageTool: turn.followupRun.run.sourceReplyDeliveryMode === "message_tool_only",
silentReplyPromptMode: turn.followupRun.run.silentReplyPromptMode,
suppressNextUserMessagePersistence: params.suppressQueuedUserPersistenceForCandidate,
onUserMessagePersisted: params.notifyUserMessagePersisted,
suppressTranscriptOnlyAssistantPersistence:
turn.followupRun.run.suppressTranscriptOnlyAssistantPersistence,
suppressAssistantErrorPersistence: params.suppressAssistantErrorPersistenceForCandidate,
onAssistantErrorMessagePersisted: params.onAssistantErrorMessagePersisted,
toolResultFormat: (() => {
const channel = resolveMessageChannel(turn.sessionCtx.Surface, turn.sessionCtx.Provider);
return !channel || isMarkdownCapableMessageChannel(channel) ? "markdown" : "plain";
})(),
toolProgressDetail: turn.toolProgressDetail,
suppressToolErrorWarnings:
turn.opts?.shouldSuppressToolErrorWarnings ?? turn.opts?.suppressToolErrorWarnings,
toolsAllow: turn.opts?.toolsAllow,
disableTools: turn.opts?.disableTools,
enableHeartbeatTool: turn.opts?.enableHeartbeatTool,
forceHeartbeatTool: turn.opts?.forceHeartbeatTool,
bootstrapContextMode: turn.opts?.bootstrapContextMode,
bootstrapContextRunKind: params.bootstrapContextRunKind,
images: params.currentTurnImages.images,
imageOrder: params.currentTurnImages.imageOrder,
abortSignal: params.runAbortSignal,
replyOperation: turn.replyOperation,
deferTerminalLifecycle: true,
onExecutionStarted: (info) => {
if (info?.lifecycleGeneration) {
params.onLifecycleGeneration(info.lifecycleGeneration);
}
},
onExecutionPhase: params.signalExecutionPhaseForTyping,
blockReplyBreak: turn.resolvedBlockStreamingBreak,
blockReplyChunking: turn.blockReplyChunking,
// Subscriber callbacks are detached. Stage channel presentation before typing I/O.
onPartialReply: async (payload) => {
if (!params.preserveProgressCallbackStartOrder) {
const textForTyping = await params.presentation.handlePartialForTyping(payload);
if (!turn.opts?.onPartialReply || textForTyping === undefined) {
return;
}
await turn.opts.onPartialReply({ text: textForTyping, mediaUrls: payload.mediaUrls });
return;
}
const textForTyping = params.presentation.preparePartialForTyping(payload);
if (textForTyping === undefined) {
return;
}
await params.presentation.startPresentationWhileTyping(
turn.typingSignals.signalTextDelta(textForTyping),
() =>
turn.opts?.onPartialReply?.({ text: textForTyping, mediaUrls: payload.mediaUrls }),
);
},
onAssistantMessageStart: async () => {
if (!params.preserveProgressCallbackStartOrder) {
await turn.typingSignals.signalMessageStart();
await turn.opts?.onAssistantMessageStart?.();
return;
}
await params.presentation.startPresentationWhileTyping(
turn.typingSignals.signalMessageStart(),
() => turn.opts?.onAssistantMessageStart?.(),
);
},
onReasoningStream:
turn.typingSignals.shouldStartOnReasoning || turn.opts?.onReasoningStream
? async (payload) => {
if (turn.followupRun.run.silentExpected) {
return;
}
if (!params.preserveProgressCallbackStartOrder) {
await turn.typingSignals.signalReasoningDelta();
await turn.opts?.onReasoningStream?.({
text: payload.text,
mediaUrls: payload.mediaUrls,
isReasoningSnapshot: payload.isReasoningSnapshot,
requiresReasoningProgressOptIn: payload.requiresReasoningProgressOptIn,
});
return;
}
await params.presentation.startPresentationWhileTyping(
turn.typingSignals.signalReasoningDelta(),
() =>
turn.opts?.onReasoningStream?.({
text: payload.text,
mediaUrls: payload.mediaUrls,
isReasoningSnapshot: payload.isReasoningSnapshot,
requiresReasoningProgressOptIn: payload.requiresReasoningProgressOptIn,
}),
);
}
: undefined,
streamReasoningInNonStreamModes: turn.opts?.streamReasoningInNonStreamModes,
onReasoningEnd: turn.opts?.onReasoningEnd,
onAgentEvent: createAgentRunEventHandler({
turn,
lifecycleBackstop,
notifyAgentRunStart: params.notifyAgentRunStart,
sourceRepliesAreToolOnly: params.sourceRepliesAreToolOnly,
messageToolDeliveryState: params.messageToolDeliveryState,
provider: params.provider,
model: params.model,
effectiveSessionId: params.effectiveRun.sessionId,
notifyUserAboutCompaction: params.notifyUserAboutCompaction,
onCompactionCompleted: () => {
attemptCompactionCount += 1;
return attemptCompactionCount;
},
}),
// Flush-before-tool requires a handler even when regular block streaming is off.
onBlockReply: params.presentation.blockReplyHandler,
onBlockReplyFlush:
turn.blockStreamingEnabled && turn.blockReplyPipeline
? async () => {
await turn.blockReplyPipeline?.flush({ force: true });
}
: undefined,
shouldEmitToolResult: turn.shouldEmitToolResult,
shouldEmitToolOutput: turn.shouldEmitToolOutput,
bootstrapPromptWarningSignaturesSeen: params.bootstrapPromptWarningSignaturesSeen,
bootstrapPromptWarningSignature:
params.bootstrapPromptWarningSignaturesSeen[
params.bootstrapPromptWarningSignaturesSeen.length - 1
],
onToolResult: turn.opts?.onToolResult
? (() => {
// Serialized delivery preserves tool result order across detached callbacks.
let toolResultChain: Promise<void> = Promise.resolve();
return (payload: ReplyPayload) => {
toolResultChain = toolResultChain
.then(async () => {
turn.replyOperation?.recordActivity();
const { text, skip } = params.presentation.normalizeStreamingText(payload);
if (skip) {
return;
}
if (text !== undefined) {
await turn.typingSignals.signalTextDelta(text);
}
await turn.opts?.onToolResult?.({ ...payload, text });
})
.catch((err: unknown) => {
logVerbose(`tool result delivery failed: ${String(err)}`);
});
const task = toolResultChain.finally(() => {
turn.pendingToolTasks.delete(task);
});
turn.pendingToolTasks.add(task);
};
})()
: undefined,
}),
);
const resultCompactionCount = Math.max(0, result.meta?.agentMeta?.compactionCount ?? 0);
attemptCompactionCount = Math.max(attemptCompactionCount, resultCompactionCount);
return {
result,
bootstrapPromptWarningSignaturesSeen: resolveBootstrapWarningSignaturesSeen(
result.meta?.systemPromptReport,
),
};
} finally {
params.onCompactionCount(attemptCompactionCount);
revokeMessageActionTurnCapability(messageActionTurnCapability);
}
}

View File

@@ -0,0 +1,309 @@
import { sanitizeForLog } from "../../../packages/terminal-core/src/ansi.js";
import {
classifyOAuthRefreshFailure,
classifyOAuthRefreshFailureError,
} from "../../agents/auth-profiles/oauth-refresh-failure.js";
import {
formatRateLimitOrOverloadedErrorCopy,
isBillingErrorMessage,
isCompactionFailureError,
isLikelyContextOverflowError,
isOverloadedErrorMessage,
isRateLimitErrorMessage,
isTransientHttpError,
} from "../../agents/embedded-agent-helpers.js";
import { sanitizeUserFacingText } from "../../agents/embedded-agent-helpers/sanitize-user-facing-text.js";
import { isFailoverError } from "../../agents/failover-error.js";
import { LiveSessionModelSwitchError } from "../../agents/live-model-switch-error.js";
import { isFallbackSummaryError } from "../../agents/model-fallback.js";
import {
AGENT_RUN_RESTART_ABORT_STOP_REASON,
resolveAgentRunErrorLifecycleFields,
} from "../../agents/run-termination.js";
import { emitAgentEvent } from "../../infra/agent-events.js";
import { formatErrorMessage } from "../../infra/errors.js";
import { CommandLaneClearedError, GatewayDrainingError } from "../../process/command-queue.js";
import { defaultRuntime } from "../../runtime.js";
import { SILENT_REPLY_TOKEN } from "../tokens.js";
import { buildContextOverflowRecoveryText } from "./agent-runner-context-recovery.js";
import type { AgentRunLoopResult, AgentTurnParams } from "./agent-runner-execution.types.js";
import {
GENERIC_EXTERNAL_RUN_FAILURE_TEXT,
HEARTBEAT_EXTERNAL_RUN_FAILURE_TEXT,
} from "./agent-runner-failure-copy.js";
import {
buildAuthProfileFailoverFailureText,
buildExternalRunFailureReply,
buildRateLimitCooldownMessage,
hasBillingAttemptSummary,
isNonDirectConversationContext,
isPureTransientRateLimitSummary,
isVerboseFailureDetailEnabled,
markAgentRunFailureReplyPayload,
resolveBillingFailureReplyText,
resolveExternalRunFailureTextForConversation,
} from "./agent-runner-failure-reply.js";
import type { AgentFallbackCycleState } from "./agent-runner-fallback-cycle.js";
import type { AgentTurnTimingTracker } from "./agent-runner-turn-timing.js";
import { classifyProviderRequestError } from "./provider-request-error-classifier.js";
import {
buildRestartLifecycleReplyText,
isReplyOperationRestartAbort,
isReplyOperationUserAbort,
resolveRestartLifecycleError,
} from "./reply-operation-abort.js";
const MAX_LIVE_SWITCH_RETRIES = 2;
const TRANSIENT_HTTP_RETRY_DELAY_MS = 2_500;
type ErrorAction =
| { kind: "retry"; liveModelSwitchError?: LiveSessionModelSwitchError }
| Extract<AgentRunLoopResult, { kind: "final" }>;
export async function handleAgentExecutionError(params: {
turn: AgentTurnParams;
error: unknown;
runtimeConfig: AgentTurnParams["followupRun"]["run"]["config"];
runId: string;
state: AgentFallbackCycleState;
liveModelSwitchRetries: number;
shouldSurfaceToControlUi: boolean;
timing: AgentTurnTimingTracker;
consumeTransientHttpRetry: () => boolean;
modelPatch: { fail: (error: unknown) => Promise<void> };
}): Promise<ErrorAction> {
const turn = params.turn;
const err = params.error;
const takePendingLifecycleTerminal = () => {
const terminal = params.state.pendingLifecycleTerminal?.backstop;
params.state.pendingLifecycleTerminal = undefined;
return terminal;
};
if (err instanceof LiveSessionModelSwitchError) {
if (params.liveModelSwitchRetries <= MAX_LIVE_SWITCH_RETRIES) {
params.state.pendingLifecycleTerminal = undefined;
return { kind: "retry", liveModelSwitchError: err };
}
defaultRuntime.error(
`Live model switch failed after ${MAX_LIVE_SWITCH_RETRIES} retries ` +
`(${sanitizeForLog(err.provider)}/${sanitizeForLog(err.model)}). The requested model may be unavailable.`,
);
takePendingLifecycleTerminal()?.emit("error", err);
const switchErrorText = params.shouldSurfaceToControlUi
? "⚠️ Agent failed before reply: model switch could not be completed. " +
"The requested model may be temporarily unavailable.\n" +
"Logs: openclaw logs --follow"
: isVerboseFailureDetailEnabled(turn.resolvedVerboseLevel)
? "⚠️ Agent failed before reply: model switch could not be completed. " +
"The requested model may be temporarily unavailable. Please try again shortly."
: "⚠️ Model switch could not be completed. The requested model may be temporarily unavailable. Please try again shortly.";
turn.replyOperation?.fail("run_failed", err);
await params.modelPatch.fail(err);
return {
kind: "final",
payload: markAgentRunFailureReplyPayload({
text: resolveExternalRunFailureTextForConversation({
text: switchErrorText,
sessionCtx: turn.sessionCtx,
isGenericRunnerFailure: !params.shouldSurfaceToControlUi,
cfg: turn.followupRun.run.config,
}),
}),
};
}
const message = formatErrorMessage(err);
params.timing.logIfSlow({
runId: params.runId,
sessionId: turn.followupRun.run.sessionId,
sessionKey: turn.sessionKey,
outcome: "error",
error: message,
});
const isFallbackSummary = isFallbackSummaryError(err);
const isBilling = isFallbackSummary
? hasBillingAttemptSummary(err)
: isFailoverError(err)
? err.reason === "billing"
: isBillingErrorMessage(message);
const isContextOverflow =
!isBilling &&
((isFailoverError(err) && err.reason === "context_overflow") ||
isLikelyContextOverflowError(message));
const isCompactionFailure = !isBilling && isCompactionFailureError(message);
const oauthRefreshFailure =
classifyOAuthRefreshFailureError(err) ?? classifyOAuthRefreshFailure(message);
const hasAuthProfileFailoverFailure = buildAuthProfileFailoverFailureText(err) !== null;
const providerRequestError =
!isBilling &&
!oauthRefreshFailure &&
!hasAuthProfileFailoverFailure &&
!params.shouldSurfaceToControlUi
? classifyProviderRequestError(err)
: undefined;
const isTransientHttp =
isTransientHttpError(message) ||
(isFailoverError(err) && (err.reason === "timeout" || err.reason === "server_error"));
if (isReplyOperationRestartAbort(turn.replyOperation)) {
takePendingLifecycleTerminal()?.emit("end", err);
return {
kind: "final",
payload:
turn.isRestartRecoveryArmed?.() === true
? { text: SILENT_REPLY_TOKEN }
: markAgentRunFailureReplyPayload({ text: buildRestartLifecycleReplyText() }),
};
}
if (isReplyOperationUserAbort(turn.replyOperation)) {
takePendingLifecycleTerminal()?.emit("error", err);
return { kind: "final", payload: { text: SILENT_REPLY_TOKEN } };
}
const restartLifecycleError = resolveRestartLifecycleError(err);
if (
restartLifecycleError instanceof GatewayDrainingError ||
restartLifecycleError instanceof CommandLaneClearedError
) {
takePendingLifecycleTerminal()?.emit("error", restartLifecycleError);
turn.replyOperation?.fail(
restartLifecycleError instanceof GatewayDrainingError
? "gateway_draining"
: "command_lane_cleared",
restartLifecycleError,
);
return {
kind: "final",
payload: markAgentRunFailureReplyPayload({ text: buildRestartLifecycleReplyText() }),
};
}
if (isCompactionFailure) {
takePendingLifecycleTerminal()?.emit("error", err);
defaultRuntime.error(
`Auto-compaction failed (${message}). Preserving existing session mapping for ${turn.sessionKey ?? turn.followupRun.run.sessionId}.`,
);
turn.replyOperation?.fail("run_failed", err);
return {
kind: "final",
payload: markAgentRunFailureReplyPayload({
text: buildContextOverflowRecoveryText({
duringCompaction: true,
preserveSessionMapping: true,
cfg: params.runtimeConfig,
agentId: turn.followupRun.run.agentId,
primaryProvider: turn.followupRun.run.provider,
primaryModel: turn.followupRun.run.model,
runtimeProvider: params.state.attemptedRuntimeProvider,
runtimeModel: params.state.attemptedRuntimeModel,
activeSessionEntry: turn.getActiveSessionEntry(),
}),
}),
};
}
if (providerRequestError) {
takePendingLifecycleTerminal()?.emit("error", err);
turn.replyOperation?.fail("run_failed", err);
await params.modelPatch.fail(err);
return {
kind: "final",
payload: markAgentRunFailureReplyPayload({ text: providerRequestError.userMessage }),
};
}
if (isTransientHttp && params.consumeTransientHttpRetry()) {
params.state.pendingLifecycleTerminal = undefined;
defaultRuntime.error(
`Transient HTTP provider error before reply (${message}). Retrying once in ${TRANSIENT_HTTP_RETRY_DELAY_MS}ms.`,
);
await new Promise<void>((resolve) => {
setTimeout(resolve, TRANSIENT_HTTP_RETRY_DELAY_MS);
});
return { kind: "retry" };
}
defaultRuntime.error(`Embedded agent failed before reply: ${message}`);
const isPureTransientSummary = isFallbackSummary ? isPureTransientRateLimitSummary(err) : false;
const failoverReason = !isFallbackSummary && isFailoverError(err) ? err.reason : undefined;
const isOverloaded = failoverReason === "overloaded" || isOverloadedErrorMessage(message);
const isRateLimit = isFallbackSummary
? isPureTransientSummary
: failoverReason
? failoverReason === "rate_limit" || failoverReason === "overloaded"
: isRateLimitErrorMessage(message);
const rateLimitOrOverloadedCopy =
!isFallbackSummary || isPureTransientSummary
? formatRateLimitOrOverloadedErrorCopy(
failoverReason === "overloaded" ? "overloaded" : message,
)
: undefined;
const trimmedMessage = (
isTransientHttp ? sanitizeUserFacingText(message, { errorContext: true }) : message
).replace(/\.\s*$/, "");
const externalRunFailureReply =
!isBilling &&
!(isRateLimit && !isOverloaded) &&
!rateLimitOrOverloadedCopy &&
!isContextOverflow &&
!params.shouldSurfaceToControlUi
? buildExternalRunFailureReply(
{ message, error: err },
{
includeAuthProfileId: !isNonDirectConversationContext(turn.sessionCtx),
includeDetails: isVerboseFailureDetailEnabled(turn.resolvedVerboseLevel),
isHeartbeat: turn.isHeartbeat,
},
)
: undefined;
const fallbackText = isBilling
? resolveBillingFailureReplyText(err)
: isRateLimit && !isOverloaded
? buildRateLimitCooldownMessage(err)
: rateLimitOrOverloadedCopy
? rateLimitOrOverloadedCopy
: isContextOverflow
? "⚠️ Context overflow — prompt too large for this model. Try a shorter message or a larger-context model."
: params.shouldSurfaceToControlUi
? `⚠️ Agent failed before reply: ${trimmedMessage}.\nLogs: openclaw logs --follow`
: (externalRunFailureReply?.text ??
(turn.isHeartbeat
? HEARTBEAT_EXTERNAL_RUN_FAILURE_TEXT
: GENERIC_EXTERNAL_RUN_FAILURE_TEXT));
const userVisibleFallbackText = resolveExternalRunFailureTextForConversation({
text: fallbackText,
sessionCtx: turn.sessionCtx,
isGenericRunnerFailure: externalRunFailureReply?.isGenericRunnerFailure ?? false,
cfg: turn.followupRun.run.config,
});
const abortedSignal =
turn.replyOperation?.abortSignal.aborted === true
? turn.replyOperation.abortSignal
: turn.opts?.abortSignal?.aborted === true
? turn.opts.abortSignal
: undefined;
const abortLifecycleFields = {
...resolveAgentRunErrorLifecycleFields(err, abortedSignal),
...(isReplyOperationRestartAbort(turn.replyOperation)
? { aborted: true as const, stopReason: AGENT_RUN_RESTART_ABORT_STOP_REASON }
: {}),
};
const failedLifecycleTerminal = takePendingLifecycleTerminal();
if (failedLifecycleTerminal) {
failedLifecycleTerminal.emit("error", err, { fallbackExhaustedFailure: true });
} else {
emitAgentEvent({
runId: params.runId,
lifecycleGeneration: params.state.lifecycleGeneration,
...(turn.sessionKey ? { sessionKey: turn.sessionKey } : {}),
stream: "lifecycle",
data: {
phase: "error",
error: message,
endedAt: Date.now(),
...abortLifecycleFields,
fallbackExhaustedFailure: true,
},
});
}
turn.replyOperation?.fail("run_failed", err);
await params.modelPatch.fail(err);
return {
kind: "final",
payload: markAgentRunFailureReplyPayload({ text: userVisibleFallbackText }),
};
}

View File

@@ -0,0 +1,315 @@
import { readStringValue } from "@openclaw/normalization-core/string-coerce";
import { isMessagingToolSendAction } from "../../agents/embedded-agent-messaging.js";
import type { RunEmbeddedAgentParams } from "../../agents/embedded-agent-runner/run/params.js";
import { logVerbose } from "../../globals.js";
import { createSubsystemLogger } from "../../logging/subsystem.js";
import type { ReplyPayload } from "../types.js";
import type { AgentLifecycleTerminalBackstop } from "./agent-lifecycle-terminal.js";
import { buildCommandOutputFromToolResultEvent } from "./agent-runner-command-output.js";
import type { AgentTurnParams } from "./agent-runner-execution.types.js";
import {
createCompactionHookNoticePayload,
createCompactionNoticePayload,
formatCompactionModelRef,
readCompactionHookMessages,
} from "./compaction-notice.js";
const agentCompactionLog = createSubsystemLogger("auto-reply/compaction");
const CODEX_APP_SERVER_COMPACTION_BACKEND = "codex-app-server";
export type MessageToolDeliveryState = {
toolCallIds: Set<string>;
completed: boolean;
};
function readApprovalScopeValue(value: unknown): "turn" | "session" | undefined {
return value === "turn" || value === "session" ? value : undefined;
}
/** Bridges embedded-agent events into channel progress and compaction notices. */
export function createAgentRunEventHandler(params: {
turn: AgentTurnParams;
lifecycleBackstop: AgentLifecycleTerminalBackstop;
notifyAgentRunStart: () => void;
sourceRepliesAreToolOnly: boolean;
provider: string;
model: string;
effectiveSessionId?: string;
notifyUserAboutCompaction: boolean;
onCompactionCompleted: () => number;
messageToolDeliveryState: MessageToolDeliveryState;
}): NonNullable<RunEmbeddedAgentParams["onAgentEvent"]> {
const commentaryTextByItem = new Map<string, string>();
const lastEmittedCommentaryByItem = new Map<string, string>();
const shouldSuppressProgressAfterMessageToolDelivery = () =>
params.sourceRepliesAreToolOnly &&
params.messageToolDeliveryState.completed &&
params.turn.opts?.allowProgressCallbacksWhenSourceDeliverySuppressed !== true;
const currentMessageId =
params.turn.sessionCtx.MessageSidFull ?? params.turn.sessionCtx.MessageSid;
const deliverCompactionNoticePayload = async (noticePayload: ReplyPayload, label: string) => {
const deliver = params.turn.opts?.onBlockReply ?? params.turn.onCompactionNoticePayload;
if (!deliver) {
return;
}
try {
await deliver(noticePayload);
} catch (err) {
logVerbose(`compaction ${label} notice delivery failed (non-fatal): ${String(err)}`);
}
};
const sendCompactionNotice = async (phase: "start" | "end" | "incomplete") => {
await deliverCompactionNoticePayload(
createCompactionNoticePayload({
phase,
currentMessageId,
applyReplyToMode: params.turn.applyReplyToMode,
}),
phase,
);
};
const sendCompactionHookMessages = async (messages: string[]) => {
const noticePayload = createCompactionHookNoticePayload({
messages,
currentMessageId,
applyReplyToMode: params.turn.applyReplyToMode,
});
if (noticePayload) {
await deliverCompactionNoticePayload(noticePayload, "hook");
}
};
return async (evt) => {
params.turn.replyOperation?.recordActivity();
params.lifecycleBackstop.note(evt);
const hasLifecyclePhase = evt.stream === "lifecycle" && typeof evt.data.phase === "string";
if (evt.stream !== "lifecycle" || hasLifecyclePhase) {
params.notifyAgentRunStart();
}
if (evt.stream === "tool" && evt.data.hideFromChannelProgress !== true) {
const phase = readStringValue(evt.data.phase) ?? "";
const name = readStringValue(evt.data.name);
const toolCallId = readStringValue(evt.data.toolCallId) ?? "";
const args =
evt.data.args && typeof evt.data.args === "object"
? (evt.data.args as Record<string, unknown>)
: undefined;
if (
params.sourceRepliesAreToolOnly &&
toolCallId &&
name &&
(phase === "start" || phase === "update") &&
args &&
isMessagingToolSendAction(name, args)
) {
params.messageToolDeliveryState.toolCallIds.add(toolCallId);
}
if (shouldSuppressProgressAfterMessageToolDelivery()) {
return;
}
if (phase === "start" || phase === "update") {
const toolStartProgressPromise = params.turn.opts?.onToolStart?.({
itemId: readStringValue(evt.data.itemId),
toolCallId: readStringValue(evt.data.toolCallId),
name,
phase,
args,
detailMode: params.turn.toolProgressDetail,
});
await Promise.all([params.turn.typingSignals.signalToolStart(), toolStartProgressPromise]);
}
const commandOutput = buildCommandOutputFromToolResultEvent(evt);
if (commandOutput) {
await params.turn.opts?.onCommandOutput?.(commandOutput);
}
}
const suppressItemChannelProgress =
evt.stream === "item" &&
evt.data.suppressChannelProgress === true &&
Boolean(params.turn.opts?.onToolStart);
const hideItemFromChannelProgress =
evt.stream === "item" && evt.data.hideFromChannelProgress === true;
const itemPhase = evt.stream === "item" ? readStringValue(evt.data.phase) : "";
const itemName = evt.stream === "item" ? readStringValue(evt.data.name) : "";
const itemStatus = evt.stream === "item" ? readStringValue(evt.data.status) : "";
const itemToolCallId =
evt.stream === "item" ? (readStringValue(evt.data.toolCallId) ?? "") : "";
const completedMessageToolDelivery =
params.sourceRepliesAreToolOnly &&
itemPhase === "end" &&
itemStatus === "completed" &&
itemToolCallId.length > 0 &&
params.messageToolDeliveryState.toolCallIds.has(itemToolCallId);
const suppressProgressAfterMessageToolDelivery =
shouldSuppressProgressAfterMessageToolDelivery();
if (completedMessageToolDelivery) {
params.messageToolDeliveryState.toolCallIds.delete(itemToolCallId);
params.messageToolDeliveryState.completed = true;
}
if (
evt.stream === "assistant" &&
readStringValue(evt.data.phase) === "commentary" &&
!shouldSuppressProgressAfterMessageToolDelivery()
) {
const commentaryItemId = readStringValue(evt.data.itemId) ?? "";
const snapshotText = readStringValue(evt.data.text);
const deltaText = readStringValue(evt.data.delta);
const accumulated =
evt.data.replace === true && snapshotText
? snapshotText
: deltaText
? `${commentaryTextByItem.get(commentaryItemId) ?? ""}${deltaText}`
: (snapshotText ?? "");
commentaryTextByItem.set(commentaryItemId, accumulated);
const commentaryText = accumulated.replace(/\s+/g, " ").trim();
if (commentaryText && lastEmittedCommentaryByItem.get(commentaryItemId) !== commentaryText) {
lastEmittedCommentaryByItem.set(commentaryItemId, commentaryText);
await params.turn.opts?.onItemEvent?.({
itemId: commentaryItemId || undefined,
kind: "preamble",
title: "Preamble",
phase: "update",
progressText: commentaryText,
});
}
}
if (
evt.stream === "item" &&
!hideItemFromChannelProgress &&
!suppressItemChannelProgress &&
(!suppressProgressAfterMessageToolDelivery || completedMessageToolDelivery)
) {
await params.turn.opts?.onItemEvent?.({
itemId: readStringValue(evt.data.itemId),
toolCallId: readStringValue(evt.data.toolCallId),
kind: readStringValue(evt.data.kind),
title: readStringValue(evt.data.title),
name: itemName,
phase: itemPhase,
status: itemStatus,
summary: readStringValue(evt.data.summary),
progressText: readStringValue(evt.data.progressText),
meta: readStringValue(evt.data.meta),
approvalId: readStringValue(evt.data.approvalId),
approvalSlug: readStringValue(evt.data.approvalSlug),
});
}
if (evt.stream === "plan" && !shouldSuppressProgressAfterMessageToolDelivery()) {
await params.turn.opts?.onPlanUpdate?.({
phase: readStringValue(evt.data.phase),
title: readStringValue(evt.data.title),
explanation: readStringValue(evt.data.explanation),
steps: Array.isArray(evt.data.steps)
? evt.data.steps.filter((step): step is string => typeof step === "string")
: undefined,
source: readStringValue(evt.data.source),
});
}
if (evt.stream === "approval" && !shouldSuppressProgressAfterMessageToolDelivery()) {
await params.turn.opts?.onApprovalEvent?.({
phase: readStringValue(evt.data.phase),
kind: readStringValue(evt.data.kind),
status: readStringValue(evt.data.status),
title: readStringValue(evt.data.title),
itemId: readStringValue(evt.data.itemId),
toolCallId: readStringValue(evt.data.toolCallId),
approvalId: readStringValue(evt.data.approvalId),
approvalSlug: readStringValue(evt.data.approvalSlug),
command: readStringValue(evt.data.command),
host: readStringValue(evt.data.host),
reason: readStringValue(evt.data.reason),
scope: readApprovalScopeValue(evt.data.scope),
message: readStringValue(evt.data.message),
});
}
if (evt.stream === "command_output" && !shouldSuppressProgressAfterMessageToolDelivery()) {
await params.turn.opts?.onCommandOutput?.({
itemId: readStringValue(evt.data.itemId),
phase: readStringValue(evt.data.phase),
title: readStringValue(evt.data.title),
toolCallId: readStringValue(evt.data.toolCallId),
name: readStringValue(evt.data.name),
output: readStringValue(evt.data.output),
status: readStringValue(evt.data.status),
exitCode:
typeof evt.data.exitCode === "number" || evt.data.exitCode === null
? evt.data.exitCode
: undefined,
durationMs: typeof evt.data.durationMs === "number" ? evt.data.durationMs : undefined,
cwd: readStringValue(evt.data.cwd),
});
}
if (evt.stream === "patch" && !shouldSuppressProgressAfterMessageToolDelivery()) {
await params.turn.opts?.onPatchSummary?.({
itemId: readStringValue(evt.data.itemId),
phase: readStringValue(evt.data.phase),
title: readStringValue(evt.data.title),
toolCallId: readStringValue(evt.data.toolCallId),
name: readStringValue(evt.data.name),
added: Array.isArray(evt.data.added)
? evt.data.added.filter((entry): entry is string => typeof entry === "string")
: undefined,
modified: Array.isArray(evt.data.modified)
? evt.data.modified.filter((entry): entry is string => typeof entry === "string")
: undefined,
deleted: Array.isArray(evt.data.deleted)
? evt.data.deleted.filter((entry): entry is string => typeof entry === "string")
: undefined,
summary: readStringValue(evt.data.summary),
});
}
if (evt.stream !== "compaction") {
return;
}
const phase = readStringValue(evt.data.phase) ?? "";
const backend = readStringValue(evt.data.backend);
const hookMessages = readCompactionHookMessages(evt.data.messages);
const sendCompactionUserNotices = async (noticePhase: "start" | "end" | "incomplete") => {
if (hookMessages.length > 0) {
await sendCompactionHookMessages(hookMessages);
}
if (params.notifyUserAboutCompaction) {
await sendCompactionNotice(noticePhase);
}
};
if (phase === "start") {
await params.turn.opts?.onCompactionStart?.();
await sendCompactionUserNotices("start");
return;
}
if (phase !== "end") {
return;
}
if (evt.data.completed !== true) {
await sendCompactionUserNotices("incomplete");
return;
}
const compactionCount = params.onCompactionCompleted();
if (backend === CODEX_APP_SERVER_COMPACTION_BACKEND) {
const modelRef = formatCompactionModelRef(params.provider, params.model);
const consoleMessage =
`codex app-server auto-compaction succeeded for ${modelRef}; ` +
"refreshed session context";
agentCompactionLog.info("codex app-server auto-compaction succeeded", {
event: "codex_app_server_compaction_succeeded",
backend,
provider: params.provider,
model: params.model,
sessionKey: params.turn.sessionKey,
sessionId: params.effectiveSessionId,
threadId: readStringValue(evt.data.threadId),
turnId: readStringValue(evt.data.turnId),
itemId: readStringValue(evt.data.itemId),
compactionCount,
consoleMessage,
});
}
await params.turn.opts?.onCompactionEnd?.();
await sendCompactionUserNotices("end");
};
}

View File

@@ -28,14 +28,16 @@ import type { TemplateContext } from "../templating.js";
import { SILENT_REPLY_TOKEN } from "../tokens.js";
import type { GetReplyOptions, ReplyPayload } from "../types.js";
import { resolveFallbackCandidateRun } from "./agent-runner-auth-profile.js";
import { resolveRunAfterAutoFallbackPrimaryProbeRecheck } from "./agent-runner-auto-fallback.js";
import {
buildContextOverflowRecoveryText,
computeContextAwareReserveTokensFloor,
} from "./agent-runner-context-recovery.js";
import { HEARTBEAT_EXTERNAL_RUN_FAILURE_TEXT } from "./agent-runner-failure-copy.js";
import {
buildEmptyInteractiveReplyPayload,
buildKnownAgentRunFailureReplyPayload,
buildContextOverflowRecoveryText,
computeContextAwareReserveTokensFloor,
resolveRunAfterAutoFallbackPrimaryProbeRecheck,
} from "./agent-runner-execution.js";
import { HEARTBEAT_EXTERNAL_RUN_FAILURE_TEXT } from "./agent-runner-failure-copy.js";
} from "./agent-runner-failure-reply.js";
import type { InternalGetReplyOptions } from "./get-reply.types.js";
import { PROVIDER_CONVERSATION_STATE_ERROR_USER_MESSAGE } from "./provider-request-error-classifier.js";
import type { FollowupRun } from "./queue.js";
@@ -5928,6 +5930,84 @@ describe("runAgentTurnWithFallback", () => {
expect(onCommandOutput).not.toHaveBeenCalled();
});
it("preserves message-tool-only suppression across fallback candidates", async () => {
const onItemEvent = vi.fn();
const onCommandOutput = vi.fn();
state.runEmbeddedAgentMock
.mockImplementationOnce(async (params: EmbeddedAgentParams) => {
await params.onAgentEvent?.({
stream: "tool",
data: {
phase: "start",
name: "message",
toolCallId: "message-1",
args: { action: "send", message: "Visible reply" },
},
});
await params.onAgentEvent?.({
stream: "item",
data: {
itemId: "tool-message-1",
phase: "end",
kind: "tool",
name: "message",
toolCallId: "message-1",
status: "completed",
},
});
return { payloads: [], meta: {} };
})
.mockImplementationOnce(async (params: EmbeddedAgentParams) => {
await params.onAgentEvent?.({
stream: "command_output",
data: {
itemId: "command:exec-1",
phase: "end",
name: "exec",
output: "must stay suppressed",
status: "completed",
exitCode: 0,
},
});
return { payloads: [{ text: "NO_REPLY" }], meta: {} };
});
state.runWithModelFallbackMock.mockImplementationOnce(async (params: FallbackRunnerParams) => {
await params.run("anthropic", "primary");
return {
result: await params.run("openai", "fallback"),
provider: "openai",
model: "fallback",
attempts: [],
};
});
const runAgentTurnWithFallback = await getRunAgentTurnWithFallback();
const followupRun = createFollowupRun();
followupRun.run.sourceReplyDeliveryMode = "message_tool_only";
await runAgentTurnWithFallback({
commandBody: "hello",
followupRun,
sessionCtx: { Provider: "discord", MessageSid: "msg" } as unknown as TemplateContext,
opts: { onItemEvent, onCommandOutput } satisfies GetReplyOptions,
typingSignals: createMockTypingSignaler(),
blockReplyPipeline: null,
blockStreamingEnabled: false,
resolvedBlockStreamingBreak: "message_end",
applyReplyToMode: (payload) => payload,
shouldEmitToolResult: () => true,
shouldEmitToolOutput: () => false,
pendingToolTasks: new Set(),
resetSessionAfterRoleOrderingConflict: async () => false,
isHeartbeat: false,
sessionKey: "main",
getActiveSessionEntry: () => undefined,
resolvedVerboseLevel: "on",
});
expect(onItemEvent).toHaveBeenCalledTimes(1);
expect(onCommandOutput).not.toHaveBeenCalled();
});
it("keeps opted-in progress callbacks active after message-tool-only delivery completes", async () => {
const onToolStart = vi.fn();
const onCommandOutput = vi.fn();

File diff suppressed because it is too large Load Diff

View File

@@ -0,0 +1,81 @@
import type { runEmbeddedAgent } from "../../agents/embedded-agent.js";
import type { SessionEntry } from "../../config/sessions.js";
import type { TemplateContext } from "../templating.js";
import type { VerboseLevel } from "../thinking.js";
import type { ReplyPayload } from "../types.js";
import type { BlockReplyPipeline } from "./block-reply-pipeline.js";
import type { InternalGetReplyOptions } from "./get-reply.types.js";
import type { FollowupRun } from "./queue.js";
import type { ReplyMediaContext } from "./reply-media-paths.js";
import type { ReplyOperation } from "./reply-run-registry.js";
import type { TypingSignaler } from "./typing-mode.js";
/** One attempted runtime fallback candidate and its failure reason. */
export type RuntimeFallbackAttempt = {
provider: string;
model: string;
error: string;
reason?: string;
status?: number;
code?: string;
};
/** Result of running an agent turn through fallback/retry handling. */
export type AgentRunLoopResult =
| {
kind: "success";
runId: string;
runResult: Awaited<ReturnType<typeof runEmbeddedAgent>>;
fallbackProvider?: string;
fallbackModel?: string;
fallbackExhausted?: true;
fallbackAttempts: RuntimeFallbackAttempt[];
didLogHeartbeatStrip: boolean;
autoCompactionCount: number;
/** Payload keys sent directly (not via pipeline) during tool flush. */
directlySentBlockKeys?: Set<string>;
/** Payloads successfully sent directly during tool flush. */
directlySentBlockPayloads?: ReplyPayload[];
/** Prepared terminal failure, appended only after delivery evidence settles. */
terminalFailurePayload?: ReplyPayload;
}
| { kind: "final"; payload: ReplyPayload };
/** Inputs shared by direct and queued agent-turn execution. */
export type AgentTurnParams = {
commandBody: string;
transcriptCommandBody?: string;
followupRun: FollowupRun;
sessionCtx: TemplateContext;
replyThreading?: TemplateContext["ReplyThreading"];
replyOperation?: ReplyOperation;
opts?: InternalGetReplyOptions;
typingSignals: TypingSignaler;
blockReplyPipeline: BlockReplyPipeline | null;
blockStreamingEnabled: boolean;
blockReplyChunking?: {
minChars: number;
maxChars: number;
breakPreference: "paragraph" | "newline" | "sentence";
flushOnParagraph?: boolean;
};
resolvedBlockStreamingBreak: "text_end" | "message_end";
applyReplyToMode: (payload: ReplyPayload) => ReplyPayload;
shouldEmitToolResult: () => boolean;
shouldEmitToolOutput: () => boolean;
pendingToolTasks: Set<Promise<void>>;
resetSessionAfterRoleOrderingConflict: (reason: string) => Promise<boolean>;
isHeartbeat: boolean;
sessionKey?: string;
runtimePolicySessionKey?: string;
getActiveSessionEntry: () => SessionEntry | undefined;
activeSessionStore?: Record<string, SessionEntry>;
storePath?: string;
resolvedVerboseLevel: VerboseLevel;
toolProgressDetail?: "explain" | "raw";
replyMediaContext?: ReplyMediaContext;
onCompactionNoticePayload?: (payload: ReplyPayload) => Promise<void> | void;
isRestartRecoveryArmed?: () => boolean;
};
export type EmbeddedAgentRunResult = Awaited<ReturnType<typeof runEmbeddedAgent>>;

View File

@@ -0,0 +1,566 @@
import { expectDefined } from "@openclaw/normalization-core";
import { normalizeLowercaseStringOrEmpty } from "@openclaw/normalization-core/string-coerce";
import { truncateUtf16Safe } from "@openclaw/normalization-core/utf16-slice";
import { formatAuthProfileFailureMessage } from "../../agents/auth-profiles/failure-copy.js";
import {
buildOAuthRefreshFailureLoginCommand,
classifyOAuthRefreshFailure,
classifyOAuthRefreshFailureError,
formatOAuthRefreshFailureLoginCommandMarkdown,
} from "../../agents/auth-profiles/oauth-refresh-failure.js";
import {
BILLING_ERROR_USER_MESSAGE,
formatBillingErrorMessage,
formatRateLimitOrOverloadedErrorCopy,
isBillingErrorMessage,
isOverloadedErrorMessage,
isRateLimitErrorMessage,
} from "../../agents/embedded-agent-helpers.js";
import { isPeriodicUsageLimitErrorMessage } from "../../agents/embedded-agent-helpers/failover-matches.js";
import { sanitizeUserFacingText } from "../../agents/embedded-agent-helpers/sanitize-user-facing-text.js";
import { findCliMaxTurnsError, isFailoverError } from "../../agents/failover-error.js";
import { isMissingProviderAuthError } from "../../agents/model-auth.js";
import { isFallbackSummaryError } from "../../agents/model-fallback.js";
import { resolveSilentReplyPolicy } from "../../config/silent-reply.js";
import type { OpenClawConfig } from "../../config/types.openclaw.js";
import { formatErrorMessage } from "../../infra/errors.js";
import { markReplyPayloadForSourceSuppressionDelivery } from "../reply-payload.js";
import type { TemplateContext } from "../templating.js";
import type { VerboseLevel } from "../thinking.js";
import { isSilentReplyText, SILENT_REPLY_TOKEN } from "../tokens.js";
import type { ReplyPayload } from "../types.js";
import {
GENERIC_EXTERNAL_RUN_FAILURE_TEXT,
HEARTBEAT_EXTERNAL_RUN_FAILURE_TEXT,
} from "./agent-runner-failure-copy.js";
import { classifyProviderRequestError } from "./provider-request-error-classifier.js";
/** Builds a human-friendly rate-limit message, including a known cooldown. */
export function buildRateLimitCooldownMessage(err: unknown): string {
const codexUsageLimitMessage = extractCodexUsageLimitErrorMessage(err);
if (codexUsageLimitMessage) {
return codexUsageLimitMessage;
}
if (isFallbackSummaryError(err) && hasBillingAttemptSummary(err)) {
return BILLING_ERROR_USER_MESSAGE;
}
const message = formatErrorMessage(err);
if (isBillingErrorMessage(message)) {
return BILLING_ERROR_USER_MESSAGE;
}
if (!isFallbackSummaryError(err)) {
if (isPeriodicUsageLimitErrorMessage(message)) {
const providerMessage = sanitizeUserFacingText(message, { errorContext: true });
return providerMessage.startsWith("⚠️") ? providerMessage : `⚠️ ${providerMessage}`;
}
return "⚠️ All models are temporarily rate-limited. Please try again in a few minutes.";
}
const expiry = err.soonestCooldownExpiry;
const now = Date.now();
if (typeof expiry === "number" && expiry > now) {
const secsLeft = Math.max(1, Math.ceil((expiry - now) / 1000));
if (secsLeft <= 60) {
return `⚠️ Rate-limited — ready in ~${secsLeft}s. Please wait a moment.`;
}
return `⚠️ Rate-limited — ready in ~${Math.ceil(secsLeft / 60)} min. Please try again shortly.`;
}
return "⚠️ All models are temporarily rate-limited. Please try again in a few minutes.";
}
export function resolveBillingFailureReplyText(err: unknown): string {
const billingFailure = isFallbackSummaryError(err)
? err.attempts.find(
(attempt) =>
attempt.reason === "billing" &&
(attempt.authMode === "oauth" || attempt.authMode === "token"),
)
: isFailoverError(err) && err.reason === "billing"
? err
: undefined;
if (
!billingFailure ||
(billingFailure.authMode !== "oauth" && billingFailure.authMode !== "token")
) {
return BILLING_ERROR_USER_MESSAGE;
}
return formatBillingErrorMessage(
billingFailure.provider,
billingFailure.model,
billingFailure.authMode,
);
}
function extractCodexUsageLimitErrorMessage(err: unknown): string | undefined {
if (isFallbackSummaryError(err)) {
for (const attempt of err.attempts) {
const message = extractCodexUsageLimitMessage(attempt.error);
if (message) {
return `⚠️ ${message}`;
}
}
return undefined;
}
const message = extractCodexUsageLimitMessage(formatErrorMessage(err));
return message ? `⚠️ ${message}` : undefined;
}
function extractCodexUsageLimitMessage(text: string): string | undefined {
const markers = [
"You've reached your Codex subscription usage limit.",
"Codex usage limit reached.",
];
let markerIndex: number | undefined;
for (const marker of markers) {
const index = text.indexOf(marker);
if (index >= 0 && (markerIndex === undefined || index < markerIndex)) {
markerIndex = index;
}
}
if (markerIndex === undefined) {
return undefined;
}
const message = sanitizeUserFacingText(text.slice(markerIndex), { errorContext: true })
.split(/\r?\n/u)
.map((line) => line.trim())
.filter(Boolean)
.join(" ")
.trim();
if (!message) {
return undefined;
}
return message.length > 500 ? `${truncateUtf16Safe(message, 497)}...` : message;
}
export function isPureTransientRateLimitSummary(err: unknown): boolean {
return (
isFallbackSummaryError(err) &&
err.attempts.length > 0 &&
err.attempts.every((attempt) => {
const reason = attempt.reason;
return reason === "rate_limit" || reason === "overloaded";
})
);
}
export function hasBillingAttemptSummary(err: unknown): boolean {
return (
isFallbackSummaryError(err) &&
err.attempts.length > 0 &&
err.attempts.some((attempt) => attempt.reason === "billing")
);
}
function collapseRepeatedFailureDetail(message: string): string {
const parts = message
.split(/\s+\|\s+/u)
.map((part) => part.trim())
.filter(Boolean);
if (parts.length >= 2 && parts.every((part) => part === parts[0])) {
return expectDefined(parts[0], "parts entry at 0");
}
return message.trim();
}
const SAFE_MISSING_API_KEY_PROVIDERS = new Set(["anthropic", "google", "openai"]);
const EXTERNAL_RUN_FAILURE_DETAIL_MAX_CHARS = 900;
const AGENT_FAILED_BEFORE_REPLY_TEXT = "Agent failed before reply:";
const PREFLIGHT_COMPACTION_FAILURE_PREFIX = "Preflight compaction required but failed:";
type ExternalRunFailureReply = {
text: string;
isGenericRunnerFailure: boolean;
};
type ExternalRunFailureInput = string | { message: string; error?: unknown };
type ExternalFailureConversationContext = Pick<
TemplateContext,
"ChatType" | "Provider" | "SessionKey" | "Surface"
>;
export function isNonDirectConversationContext(ctx: ExternalFailureConversationContext): boolean {
const chatType = normalizeLowercaseStringOrEmpty(ctx.ChatType);
return chatType === "group" || chatType === "channel";
}
export function isVerboseFailureDetailEnabled(level: VerboseLevel | undefined): boolean {
return level === "on" || level === "full";
}
export function resolveExternalRunFailureTextForConversation(params: {
text: string;
sessionCtx: ExternalFailureConversationContext;
isGenericRunnerFailure: boolean;
cfg?: OpenClawConfig;
}): string {
if (!isNonDirectConversationContext(params.sessionCtx)) {
return params.text;
}
if (!params.isGenericRunnerFailure && !params.text.includes(AGENT_FAILED_BEFORE_REPLY_TEXT)) {
return params.text;
}
const silentPolicy = resolveSilentReplyPolicy({
cfg: params.cfg,
sessionKey: params.sessionCtx.SessionKey,
surface: params.sessionCtx.Surface ?? params.sessionCtx.Provider,
conversationType: "group",
});
return silentPolicy === "disallow" ? params.text : SILENT_REPLY_TOKEN;
}
const CLI_BACKEND_NO_OUTPUT_STALL_RE =
/\bCLI produced no output for\s+(\d+)\s*s\s+and was terminated\b/iu;
const CLI_BACKEND_OVERALL_TIMEOUT_RE =
/\bCLI exceeded timeout\s*\(\s*(\d+)\s*s\s*\)\s+and was terminated\b/iu;
const CLI_BACKEND_ROUTING_REF_BEFORE_ERROR_RE = /\b([\w.-]+\/[A-Za-z][\w.-]*)\s*:\s*CLI\b/iu;
const CODEX_APP_SERVER_CLIENT_CLOSED_BEFORE_REPLY_RE =
/\bcodex app-server client closed before turn completed\b/iu;
const CODEX_APP_SERVER_TURN_COMPLETION_IDLE_TIMEOUT_RE =
/\bcodex app-server turn idle timed out waiting for turn\/completed\b/iu;
function buildCodexAppServerFailureText(message: string): string | null {
const normalizedMessage = collapseRepeatedFailureDetail(message);
if (CODEX_APP_SERVER_CLIENT_CLOSED_BEFORE_REPLY_RE.test(normalizedMessage)) {
return "⚠️ Codex app-server connection closed before this turn finished. OpenClaw retried once when the stdio turn was still replay-safe; please try again if this keeps happening.";
}
if (CODEX_APP_SERVER_TURN_COMPLETION_IDLE_TIMEOUT_RE.test(normalizedMessage)) {
return "⚠️ Codex app-server stopped before confirming turn completion. OpenClaw did not replay the turn automatically because it may still be active; try again, or use /new if the session stays stuck.";
}
return null;
}
/** Formats the reply shown when preflight compaction fails before a run. */
export function buildPreflightCompactionFailureText(
message: string,
options?: { includeDetails?: boolean },
): string | null {
const normalizedMessage = collapseRepeatedFailureDetail(message);
if (!normalizedMessage.startsWith(PREFLIGHT_COMPACTION_FAILURE_PREFIX)) {
return null;
}
const reason = sanitizeUserFacingText(
normalizedMessage.slice(PREFLIGHT_COMPACTION_FAILURE_PREFIX.length),
{ errorContext: true },
)
.trim()
.replace(/\s+/gu, " ");
const reasonSuffix = options?.includeDetails && reason ? ` Reason: ${reason}.` : "";
return (
"⚠️ Context is too large and auto-compaction could not recover this turn." +
`${reasonSuffix} Try again, use /compact, or use /new to start a fresh session.`
);
}
function buildCliBackendTimeoutFailureText(message: string): string | null {
const normalizedMessage = collapseRepeatedFailureDetail(message);
const stall = normalizedMessage.match(CLI_BACKEND_NO_OUTPUT_STALL_RE);
const overall = normalizedMessage.match(CLI_BACKEND_OVERALL_TIMEOUT_RE);
const seconds = (stall ?? overall)?.[1];
if (!seconds) {
return null;
}
const routedModelRef = normalizedMessage.match(CLI_BACKEND_ROUTING_REF_BEFORE_ERROR_RE)?.[1];
const routingSuffix = routedModelRef ? ` (routing ${routedModelRef})` : "";
const modeLabel = stall ? "no-output stall" : "overall CLI turn budget";
return (
`⚠️ CLI subprocess${routingSuffix}: timed out after ${seconds}s (${modeLabel}). The gateway may still be healthy. Try \`/new\`, a lighter model, or raise ` +
"`agents.defaults.timeoutSeconds` and the watchdog `noOutputTimeoutMs` entries under `cliBackends.<your-runtime>`."
);
}
function buildMissingApiKeyFailureText(input: { message: string; error?: unknown }): string | null {
const normalizedMessage = collapseRepeatedFailureDetail(input.message);
const provider = isMissingProviderAuthError(input.error)
? input.error.provider.trim().toLowerCase()
: normalizedMessage
.match(/No API key found for provider "([^"]+)"/u)?.[1]
?.trim()
.toLowerCase();
if (!provider) {
return null;
}
if (provider === "openai" && normalizedMessage.includes("OpenAI Codex OAuth")) {
return "⚠️ Missing API key for OpenAI on the gateway. Use `openai/gpt-5.6-sol` with the OpenAI OAuth profile, or set `OPENAI_API_KEY` for direct OpenAI API-key runs.";
}
if (provider === "openai") {
return '⚠️ Missing API key for provider "openai". Run `openclaw doctor --fix` to repair stale OpenAI model/session routes, restart the gateway if doctor asks, then try again. If doctor has nothing to repair or the error persists, re-auth with `openclaw models auth login --provider openai` or run `openclaw configure`.';
}
if (SAFE_MISSING_API_KEY_PROVIDERS.has(provider)) {
return `⚠️ Missing API key for provider "${provider}". Configure the gateway auth for that provider, then try again.`;
}
return "⚠️ Missing API key for the selected provider on the gateway. Configure provider auth, then try again.";
}
export function buildAuthProfileFailoverFailureText(error: unknown): string | null {
if (!isFailoverError(error) || !error.provider || !error.authProfileFailure) {
return null;
}
return formatAuthProfileFailureMessage({
reason: error.reason,
provider: error.provider,
allInCooldown: error.authProfileFailure.allInCooldown,
cause: error.cause,
});
}
function formatForwardedExternalRunFailureText(message: string): string {
const sanitized = sanitizeUserFacingText(message, { errorContext: true })
.trim()
.replace(/^\s*/u, "")
.replace(/\s+/gu, " ");
if (!sanitized) {
return GENERIC_EXTERNAL_RUN_FAILURE_TEXT;
}
const detail =
sanitized.length > EXTERNAL_RUN_FAILURE_DETAIL_MAX_CHARS
? `${truncateUtf16Safe(sanitized, EXTERNAL_RUN_FAILURE_DETAIL_MAX_CHARS - 1).trimEnd()}`
: sanitized;
return `⚠️ Agent failed before reply: ${detail}${/[.!?]$/u.test(detail) ? "" : "."} Please try again, or use /new to start a fresh session.`;
}
function supportsChannelCodexLogin(provider: string | null | undefined): boolean {
if (!provider) {
return false;
}
const normalizedProvider = provider.trim().toLowerCase().replace(/_/gu, "-");
return (
normalizedProvider === "openai" ||
normalizedProvider === "codex" ||
normalizedProvider === "openai-codex"
);
}
export function buildExternalRunFailureReply(
input: ExternalRunFailureInput,
options?: {
includeAuthProfileId?: boolean;
includeDetails?: boolean;
isHeartbeat?: boolean;
},
): ExternalRunFailureReply {
const message = typeof input === "string" ? input : input.message;
const error = typeof input === "string" ? undefined : input.error;
const normalizedMessage = collapseRepeatedFailureDetail(message);
const oauthRefreshFailure =
classifyOAuthRefreshFailureError(error) ?? classifyOAuthRefreshFailure(normalizedMessage);
if (oauthRefreshFailure) {
const loginCommand = buildOAuthRefreshFailureLoginCommand(oauthRefreshFailure.provider, {
profileId: options?.includeAuthProfileId ? oauthRefreshFailure.profileId : undefined,
});
const loginCommandMarkdown = formatOAuthRefreshFailureLoginCommandMarkdown(loginCommand);
const providerText = oauthRefreshFailure.provider ? ` for ${oauthRefreshFailure.provider}` : "";
const supportsCodexLogin = supportsChannelCodexLogin(oauthRefreshFailure.provider);
const channelLoginHint = supportsCodexLogin
? "Send `/login codex` from a private chat or Web UI session to pair a new Codex login, or re-auth"
: "Re-auth";
const retryLoginHint = supportsCodexLogin
? "send `/login codex` from a private chat or Web UI session to pair a new Codex login, or re-auth"
: "re-auth";
if (oauthRefreshFailure.reason) {
return {
text: `⚠️ Model login expired on the gateway${providerText}. ${channelLoginHint} with ${loginCommandMarkdown} in a terminal, then try again.`,
isGenericRunnerFailure: false,
};
}
return {
text: `⚠️ Model login failed on the gateway${providerText}. Please try again. If this keeps happening, ${retryLoginHint} with ${loginCommandMarkdown} in a terminal.`,
isGenericRunnerFailure: false,
};
}
const authProfileFailoverFailure = buildAuthProfileFailoverFailureText(error);
if (authProfileFailoverFailure) {
return { text: authProfileFailoverFailure, isGenericRunnerFailure: false };
}
const cliMaxTurnsError = findCliMaxTurnsError(error);
if (cliMaxTurnsError) {
return {
text: sanitizeUserFacingText(cliMaxTurnsError.message, { errorContext: true }),
isGenericRunnerFailure: false,
};
}
const providerRequestError = classifyProviderRequestError(error ?? normalizedMessage);
if (providerRequestError) {
return { text: providerRequestError.userMessage, isGenericRunnerFailure: false };
}
const missingApiKeyFailure = buildMissingApiKeyFailureText({
message: normalizedMessage,
error,
});
if (missingApiKeyFailure) {
return { text: missingApiKeyFailure, isGenericRunnerFailure: false };
}
if (options?.isHeartbeat) {
return { text: HEARTBEAT_EXTERNAL_RUN_FAILURE_TEXT, isGenericRunnerFailure: false };
}
const cliBackendTimeoutFailure = buildCliBackendTimeoutFailureText(normalizedMessage);
if (cliBackendTimeoutFailure) {
return { text: cliBackendTimeoutFailure, isGenericRunnerFailure: false };
}
const codexAppServerFailure = buildCodexAppServerFailureText(normalizedMessage);
if (codexAppServerFailure) {
return { text: codexAppServerFailure, isGenericRunnerFailure: false };
}
return {
text: options?.includeDetails
? formatForwardedExternalRunFailureText(normalizedMessage)
: GENERIC_EXTERNAL_RUN_FAILURE_TEXT,
isGenericRunnerFailure: true,
};
}
export function markAgentRunFailureReplyPayload<T extends ReplyPayload>(payload: T): T {
const marked = markReplyPayloadForSourceSuppressionDelivery(payload);
if (!isSilentReplyText(marked.text, SILENT_REPLY_TOKEN)) {
marked.isError = true;
}
return marked;
}
export function buildTerminalAgentRunFailureReplyPayload(params: {
isHeartbeat?: boolean;
sessionCtx: ExternalFailureConversationContext;
cfg?: OpenClawConfig;
}): ReplyPayload {
return markAgentRunFailureReplyPayload({
text: resolveExternalRunFailureTextForConversation({
text: params.isHeartbeat
? HEARTBEAT_EXTERNAL_RUN_FAILURE_TEXT
: GENERIC_EXTERNAL_RUN_FAILURE_TEXT,
sessionCtx: params.sessionCtx,
isGenericRunnerFailure: true,
cfg: params.cfg,
}),
});
}
export function buildEmptyInteractiveReplyPayload(params: {
isInteractive: boolean;
isHeartbeat?: boolean;
silentExpected?: boolean;
allowEmptyAssistantReplyAsSilent?: boolean;
isMessageToolOnly: boolean;
hasPendingContinuation: boolean;
hasExplicitSilentReply: boolean;
hasCommittedDelivery: boolean;
sessionCtx: ExternalFailureConversationContext;
cfg?: OpenClawConfig;
}): ReplyPayload | undefined {
if (
!params.isInteractive ||
params.isHeartbeat === true ||
params.silentExpected === true ||
params.allowEmptyAssistantReplyAsSilent === true ||
params.isMessageToolOnly ||
params.hasPendingContinuation ||
params.hasExplicitSilentReply ||
params.hasCommittedDelivery
) {
return undefined;
}
return markAgentRunFailureReplyPayload({
text: resolveExternalRunFailureTextForConversation({
text: "I finished the turn, but it did not produce a visible reply. Please try again, or start a new session if this keeps happening.",
sessionCtx: params.sessionCtx,
isGenericRunnerFailure: true,
cfg: params.cfg,
}),
});
}
/** Converts known agent-run failures into user-facing reply payloads. */
export function buildKnownAgentRunFailureReplyPayload(params: {
err: unknown;
sessionCtx: TemplateContext;
resolvedVerboseLevel: VerboseLevel | undefined;
cfg?: OpenClawConfig;
}): ReplyPayload | undefined {
const message = formatErrorMessage(params.err);
const isFallbackSummary = isFallbackSummaryError(params.err);
const isBilling = isFallbackSummary
? hasBillingAttemptSummary(params.err)
: isFailoverError(params.err)
? params.err.reason === "billing"
: isBillingErrorMessage(message);
if (isBilling) {
return markAgentRunFailureReplyPayload({
text: resolveExternalRunFailureTextForConversation({
text: resolveBillingFailureReplyText(params.err),
sessionCtx: params.sessionCtx,
isGenericRunnerFailure: false,
cfg: params.cfg,
}),
});
}
const preflightCompactionFailureText = buildPreflightCompactionFailureText(message, {
includeDetails: isVerboseFailureDetailEnabled(params.resolvedVerboseLevel),
});
if (preflightCompactionFailureText) {
return markAgentRunFailureReplyPayload({
text: resolveExternalRunFailureTextForConversation({
text: preflightCompactionFailureText,
sessionCtx: params.sessionCtx,
isGenericRunnerFailure: false,
cfg: params.cfg,
}),
});
}
const isPureTransientSummary = isFallbackSummary
? isPureTransientRateLimitSummary(params.err)
: false;
const failoverReason =
!isFallbackSummary && isFailoverError(params.err) ? params.err.reason : undefined;
const isOverloaded = failoverReason === "overloaded" || isOverloadedErrorMessage(message);
const isRateLimit = isFallbackSummary
? isPureTransientSummary
: failoverReason
? failoverReason === "rate_limit" || failoverReason === "overloaded"
: isRateLimitErrorMessage(message);
const rateLimitOrOverloadedCopy =
!isFallbackSummary || isPureTransientSummary
? formatRateLimitOrOverloadedErrorCopy(
failoverReason === "overloaded" ? "overloaded" : message,
)
: undefined;
if (isRateLimit && !isOverloaded) {
return markAgentRunFailureReplyPayload({
text: resolveExternalRunFailureTextForConversation({
text: buildRateLimitCooldownMessage(params.err),
sessionCtx: params.sessionCtx,
isGenericRunnerFailure: false,
cfg: params.cfg,
}),
});
}
if (rateLimitOrOverloadedCopy) {
return markAgentRunFailureReplyPayload({
text: resolveExternalRunFailureTextForConversation({
text: rateLimitOrOverloadedCopy,
sessionCtx: params.sessionCtx,
isGenericRunnerFailure: false,
cfg: params.cfg,
}),
});
}
const externalRunFailureReply = buildExternalRunFailureReply(
{ message, error: params.err },
{
includeAuthProfileId: !isNonDirectConversationContext(params.sessionCtx),
includeDetails: isVerboseFailureDetailEnabled(params.resolvedVerboseLevel),
},
);
if (externalRunFailureReply.isGenericRunnerFailure) {
return undefined;
}
return markAgentRunFailureReplyPayload({
text: resolveExternalRunFailureTextForConversation({
text: externalRunFailureReply.text,
sessionCtx: params.sessionCtx,
isGenericRunnerFailure: false,
cfg: params.cfg,
}),
});
}

View File

@@ -0,0 +1,460 @@
import { normalizeLowercaseStringOrEmpty } from "@openclaw/normalization-core/string-coerce";
import { markAutoFallbackPrimaryProbe } from "../../agents/agent-scope.js";
import { isContextOverflowError } from "../../agents/embedded-agent-helpers.js";
import { mergeEmbeddedAgentRunResultForModelFallbackExhaustion } from "../../agents/embedded-agent-runner/result-fallback-classifier.js";
import type { RunEmbeddedAgentParams } from "../../agents/embedded-agent-runner/run/params.js";
import type { FastModeAutoProgressState } from "../../agents/fast-mode.js";
import { ensureSelectedAgentHarnessPlugin } from "../../agents/harness/runtime-plugin.js";
import { runWithModelFallback } from "../../agents/model-fallback.js";
import { resolveCliRuntimeExecutionProvider } from "../../agents/model-runtime-aliases.js";
import { isCliProvider } from "../../agents/model-selection.js";
import {
createAgentRunRestartAbortError,
isAgentRunRestartAbortReason,
} from "../../agents/run-termination.js";
import { buildAgentRuntimeOutcomePlan } from "../../agents/runtime-plan/build.js";
import { resolveSessionRuntimeOverrideForProvider } from "../../agents/session-runtime-compat.js";
import { resolveCandidateThinkingLevel } from "../../agents/thinking-runtime.js";
import type { SessionEntry } from "../../config/sessions.js";
import type { OpenClawConfig } from "../../config/types.openclaw.js";
import { logVerbose } from "../../globals.js";
import { emitAgentEvent } from "../../infra/agent-events.js";
import { formatErrorMessage } from "../../infra/errors.js";
import { resolveHeartbeatRunScope } from "../../infra/heartbeat-run-scope.js";
import { CommandLane } from "../../process/lanes.js";
import { defaultRuntime } from "../../runtime.js";
import { SILENT_REPLY_TOKEN } from "../tokens.js";
import {
resolveAgentLifecycleTerminalMetadata,
type AgentLifecycleTerminalBackstop,
} from "./agent-lifecycle-terminal.js";
import { resolveFallbackCandidateRun, resolveRunAuthProfile } from "./agent-runner-auth-profile.js";
import { runCliFallbackCandidate } from "./agent-runner-cli-candidate.js";
import { buildContextOverflowRecoveryText } from "./agent-runner-context-recovery.js";
import { runEmbeddedFallbackCandidate } from "./agent-runner-embedded-candidate.js";
import type { MessageToolDeliveryState } from "./agent-runner-event-handler.js";
import type {
AgentRunLoopResult,
AgentTurnParams,
EmbeddedAgentRunResult,
RuntimeFallbackAttempt,
} from "./agent-runner-execution.types.js";
import { markAgentRunFailureReplyPayload } from "./agent-runner-failure-reply.js";
import { emitModelFallbackStepLifecycle } from "./agent-runner-model-fallback-lifecycle.js";
import type { createAgentTurnPresentation } from "./agent-runner-presentation.js";
import type { AgentTurnTimingTracker } from "./agent-runner-turn-timing.js";
import {
resolveModelFallbackOptions,
resolveRunFastModeForFallbackCandidate,
} from "./agent-runner-utils.js";
import { drainPendingToolTasks } from "./pending-tool-task-drain.js";
import {
classifyProviderRequestError,
PROVIDER_CONVERSATION_STATE_ERROR_USER_MESSAGE,
} from "./provider-request-error-classifier.js";
import type { FollowupRun } from "./queue.js";
import {
isReplyOperationRestartAbort,
isReplyOperationUserAbort,
} from "./reply-operation-abort.js";
type Presentation = ReturnType<typeof createAgentTurnPresentation>;
export type AgentFallbackCycleState = {
lifecycleGeneration: string;
autoCompactionCount: number;
attemptedRuntimeProvider: string;
attemptedRuntimeModel: string;
bootstrapPromptWarningSignaturesSeen: string[];
pendingLifecycleTerminal?: {
provider: string;
model: string;
backstop: AgentLifecycleTerminalBackstop;
};
};
type CompletedFallbackCycle = {
kind: "completed";
runResult: EmbeddedAgentRunResult;
fallbackProvider: string;
fallbackModel: string;
fallbackExhausted: boolean;
fallbackAttempts: RuntimeFallbackAttempt[];
terminalRunFailed: boolean;
};
type ModelPatch = {
captureFallbackFailure: (attempts: RuntimeFallbackAttempt[]) => boolean | undefined;
captureFailure: (error: unknown) => void;
};
export async function executeAgentFallbackCycle(params: {
turn: AgentTurnParams;
effectiveRun: FollowupRun["run"];
runtimeConfig: OpenClawConfig;
liveModelSwitchRuntimeEntry?: Pick<
SessionEntry,
"agentHarnessId" | "agentRuntimeOverride" | "modelSelectionLocked"
>;
runId: string;
runAbortSignal?: AbortSignal;
currentTurnImages: Awaited<
ReturnType<typeof import("./current-turn-images.js").resolveCurrentTurnImages>
>;
state: AgentFallbackCycleState;
presentation: Presentation;
directlySentBlockKeys: Set<string>;
notifyAgentRunStart: () => void;
signalExecutionPhaseForTyping: NonNullable<RunEmbeddedAgentParams["onExecutionPhase"]>;
notifyUserAboutCompaction: boolean;
timing: AgentTurnTimingTracker;
modelPatch: ModelPatch;
shouldSurfaceToControlUi: boolean;
commitTerminalOutcome: () => void;
clearRecoveredAutoFallbackPrimaryProbe: (candidate: {
provider: string;
model: string;
}) => Promise<void>;
}): Promise<CompletedFallbackCycle | Extract<AgentRunLoopResult, { kind: "final" }>> {
const turn = params.turn;
const preserveProgressCallbackStartOrder = turn.opts?.preserveProgressCallbackStartOrder === true;
const sourceRepliesAreToolOnly =
turn.followupRun.run.sourceReplyDeliveryMode === "message_tool_only";
const outcomePlan = buildAgentRuntimeOutcomePlan();
const runLane = CommandLane.Main;
let queuedUserMessagePersistedAcrossFallback = false;
let assistantErrorPersistedAcrossFallback = false;
const messageToolDeliveryState: MessageToolDeliveryState = {
toolCallIds: new Set(),
completed: false,
};
const userTurnTranscriptRecorder =
turn.followupRun.userTurnTranscriptRecorder ?? turn.opts?.userTurnTranscriptRecorder;
const fastModeStartedAtMs = Date.now();
const fastModeAutoProgressState: FastModeAutoProgressState = {
offAnnounced: false,
resetAnnounced: false,
};
const bootstrapContextRunKind =
resolveHeartbeatRunScope(turn.opts) === "commitment-only"
? ("commitment-only" as const)
: turn.opts?.isHeartbeat
? ("heartbeat" as const)
: ("default" as const);
params.timing.logMilestoneIfSlow({
runId: params.runId,
sessionId: turn.followupRun.run.sessionId,
sessionKey: turn.sessionKey,
milestone: "before_model_fallback",
});
const fallbackResult = await params.timing.measure("model_fallback", () =>
runWithModelFallback<EmbeddedAgentRunResult>({
...resolveModelFallbackOptions(params.effectiveRun, params.runtimeConfig),
runId: params.runId,
sessionId: turn.followupRun.run.sessionId,
lane: runLane,
abortSignal: params.runAbortSignal,
resolveAgentHarnessRuntimeOverride: (provider) =>
resolveSessionRuntimeOverrideForProvider({
provider,
entry: params.liveModelSwitchRuntimeEntry ?? turn.getActiveSessionEntry(),
cfg: params.runtimeConfig,
}),
prepareAgentHarnessRuntime: async ({ provider, model, agentHarnessRuntimeOverride }) => {
await params.timing.measure("fallback_prepare_harness", () =>
ensureSelectedAgentHarnessPlugin({
config: params.runtimeConfig,
provider,
modelId: model,
agentId: turn.followupRun.run.agentId,
sessionKey: turn.followupRun.run.runtimePolicySessionKey ?? turn.sessionKey,
agentHarnessRuntimeOverride,
workspaceDir: turn.followupRun.run.workspaceDir,
}),
);
},
onFallbackStep: (step) => {
emitModelFallbackStepLifecycle({ runId: params.runId, sessionKey: turn.sessionKey, step });
},
classifyResult: ({ result, provider, model }) =>
outcomePlan.classifyRunResult({
result,
provider,
model,
hasDirectlySentBlockReply: params.directlySentBlockKeys.size > 0,
hasBlockReplyPipelineOutput: Boolean(
turn.blockReplyPipeline?.hasBuffered() || turn.blockReplyPipeline?.didStream(),
),
}),
mergeExhaustedResult: mergeEmbeddedAgentRunResultForModelFallbackExhaustion,
run: async (provider, model, runOptions) => {
params.state.attemptedRuntimeProvider = provider;
params.state.attemptedRuntimeModel = model;
const candidateRun = resolveFallbackCandidateRun(params.effectiveRun, provider, model);
const candidateThinkLevel = resolveCandidateThinkingLevel({
cfg: params.runtimeConfig,
provider,
modelId: model,
level: turn.followupRun.run.thinkLevel,
agentId: turn.followupRun.run.agentId,
sessionKey: turn.followupRun.run.runtimePolicySessionKey ?? turn.sessionKey,
sessionEntry: turn.getActiveSessionEntry(),
});
const candidateFastMode = resolveRunFastModeForFallbackCandidate({
run: candidateRun,
config: params.runtimeConfig,
provider,
model,
sessionEntry: turn.getActiveSessionEntry(),
});
const activeProbe = params.effectiveRun.autoFallbackPrimaryProbe;
if (activeProbe && provider === activeProbe.provider && model === activeProbe.model) {
markAutoFallbackPrimaryProbe({ probe: activeProbe, sessionKey: turn.sessionKey });
}
turn.opts?.onModelSelected?.({ provider, model, thinkLevel: candidateThinkLevel });
const runtime = params.timing.measureSync("fallback_resolve_runtime", () => {
const activeEntry = params.liveModelSwitchRuntimeEntry ?? turn.getActiveSessionEntry();
const sessionRuntimeOverride = resolveSessionRuntimeOverrideForProvider({
provider,
entry: activeEntry,
cfg: params.runtimeConfig,
});
const locksPersistedHarness =
activeEntry?.modelSelectionLocked === true &&
normalizeLowercaseStringOrEmpty(activeEntry.agentHarnessId) === sessionRuntimeOverride;
const selectedAuthProfile = resolveRunAuthProfile(candidateRun, provider, {
config: params.runtimeConfig,
});
const pinnedCliRuntime =
!locksPersistedHarness &&
sessionRuntimeOverride &&
isCliProvider(sessionRuntimeOverride, params.runtimeConfig)
? sessionRuntimeOverride
: undefined;
const cliExecutionProvider =
pinnedCliRuntime ??
(sessionRuntimeOverride
? provider
: (resolveCliRuntimeExecutionProvider({
provider,
cfg: params.runtimeConfig,
agentId: turn.followupRun.run.agentId,
modelId: model,
authProfileId: selectedAuthProfile.authProfileId,
}) ?? provider));
return {
sessionRuntimeOverride,
cliExecutionProvider,
useCliExecution:
pinnedCliRuntime !== undefined ||
(!sessionRuntimeOverride &&
isCliProvider(cliExecutionProvider, params.runtimeConfig)),
};
});
const common = {
turn,
candidateRun,
runtimeConfig: params.runtimeConfig,
provider,
model,
candidateThinkLevel,
candidateFastMode,
runId: params.runId,
runAbortSignal: params.runAbortSignal,
isFinalFallbackAttempt: runOptions?.isFinalFallbackAttempt,
suppressQueuedUserPersistenceForCandidate:
(turn.followupRun.run.suppressNextUserMessagePersistence ?? false) ||
queuedUserMessagePersistedAcrossFallback,
userTurnTranscriptRecorder,
notifyUserMessagePersisted: () => {
queuedUserMessagePersistedAcrossFallback = true;
},
fastModeStartedAtMs,
fastModeAutoProgressState,
bootstrapContextRunKind,
bootstrapPromptWarningSignaturesSeen: params.state.bootstrapPromptWarningSignaturesSeen,
currentTurnImages: params.currentTurnImages,
signalExecutionPhaseForTyping: params.signalExecutionPhaseForTyping,
notifyAgentRunStart: params.notifyAgentRunStart,
preserveProgressCallbackStartOrder,
presentation: params.presentation,
timing: params.timing,
onLifecycleBackstop: (backstop: AgentLifecycleTerminalBackstop) => {
params.state.pendingLifecycleTerminal = { provider, model, backstop };
},
};
if (runtime.useCliExecution) {
const candidate = await runCliFallbackCandidate({
...common,
cliExecutionProvider: runtime.cliExecutionProvider,
lifecycleGeneration: params.state.lifecycleGeneration,
runLane,
});
params.state.bootstrapPromptWarningSignaturesSeen =
candidate.bootstrapPromptWarningSignaturesSeen;
return candidate.result;
}
const candidate = await runEmbeddedFallbackCandidate({
...common,
effectiveRun: params.effectiveRun,
sessionRuntimeOverride: runtime.sessionRuntimeOverride,
getLifecycleGeneration: () => params.state.lifecycleGeneration,
onLifecycleGeneration: (generation) => {
params.state.lifecycleGeneration = generation;
},
allowTransientCooldownProbe: runOptions?.allowTransientCooldownProbe,
suppressAssistantErrorPersistenceForCandidate: assistantErrorPersistedAcrossFallback,
onAssistantErrorMessagePersisted: () => {
assistantErrorPersistedAcrossFallback = true;
},
notifyUserAboutCompaction: params.notifyUserAboutCompaction,
sourceRepliesAreToolOnly,
messageToolDeliveryState,
onCompactionCount: (count) => {
params.state.autoCompactionCount += count;
},
});
params.state.bootstrapPromptWarningSignaturesSeen =
candidate.bootstrapPromptWarningSignaturesSeen;
return candidate.result;
},
}),
);
params.timing.logIfSlow({
runId: params.runId,
sessionId: turn.followupRun.run.sessionId,
sessionKey: turn.sessionKey,
outcome: "completed",
});
const runResult = fallbackResult.result;
const fallbackProvider = fallbackResult.provider;
const fallbackModel = fallbackResult.model;
const fallbackExhausted = fallbackResult.outcome === "exhausted";
const settledLifecycleTerminal =
params.state.pendingLifecycleTerminal?.provider === fallbackProvider &&
params.state.pendingLifecycleTerminal.model === fallbackModel
? params.state.pendingLifecycleTerminal.backstop
: undefined;
params.state.pendingLifecycleTerminal = undefined;
if (isReplyOperationRestartAbort(turn.replyOperation)) {
settledLifecycleTerminal?.emit("end", runResult);
throw isAgentRunRestartAbortReason(params.runAbortSignal?.reason)
? params.runAbortSignal?.reason
: createAgentRunRestartAbortError();
}
if (isReplyOperationUserAbort(turn.replyOperation)) {
settledLifecycleTerminal?.emit("end", runResult);
await drainPendingToolTasks({ tasks: turn.pendingToolTasks, onTimeout: logVerbose });
return { kind: "final", payload: { text: SILENT_REPLY_TOKEN } };
}
params.commitTerminalOutcome();
const fallbackAttempts = Array.isArray(fallbackResult.attempts)
? fallbackResult.attempts.map((attempt) => ({
provider: attempt.provider,
model: attempt.model,
error: attempt.error,
reason: attempt.reason || undefined,
status: typeof attempt.status === "number" ? attempt.status : undefined,
code: attempt.code || undefined,
}))
: [];
if (!fallbackExhausted) {
await params.clearRecoveredAutoFallbackPrimaryProbe({
provider: fallbackProvider,
model: fallbackModel,
});
}
const embeddedError = runResult.meta?.error;
const deferredLifecycleError = settledLifecycleTerminal?.getDeferredError();
const userFacingErrorPayload = runResult.payloads?.find(
(payload) => payload.isError === true && typeof payload.text === "string",
)?.text;
const terminalErrorMessage =
deferredLifecycleError ??
userFacingErrorPayload ??
(embeddedError ? "Agent run failed" : undefined);
const emitSettledLifecycleError = (error: Error, extraData?: Record<string, unknown>) => {
if (settledLifecycleTerminal) {
settledLifecycleTerminal.emit("error", error, extraData);
return;
}
emitAgentEvent({
runId: params.runId,
lifecycleGeneration: params.state.lifecycleGeneration,
...(turn.sessionKey ? { sessionKey: turn.sessionKey } : {}),
stream: "lifecycle",
data: { phase: "error", error: error.message, endedAt: Date.now(), ...extraData },
});
};
if (embeddedError && isContextOverflowError(embeddedError.message)) {
emitSettledLifecycleError(new Error(terminalErrorMessage ?? "Agent run failed"));
defaultRuntime.error(
`Auto-compaction failed (${embeddedError.message}). Preserving existing session mapping for ${turn.sessionKey ?? turn.followupRun.run.sessionId}.`,
);
turn.replyOperation?.fail("run_failed", embeddedError);
return {
kind: "final",
payload: markAgentRunFailureReplyPayload({
text: buildContextOverflowRecoveryText({
preserveSessionMapping: true,
cfg: params.runtimeConfig,
agentId: turn.followupRun.run.agentId,
primaryProvider: turn.followupRun.run.provider,
primaryModel: turn.followupRun.run.model,
runtimeProvider: params.state.attemptedRuntimeProvider,
runtimeModel: params.state.attemptedRuntimeModel,
activeSessionEntry: turn.getActiveSessionEntry(),
}),
}),
};
}
if (embeddedError?.kind === "role_ordering") {
emitSettledLifecycleError(new Error(terminalErrorMessage ?? "Agent run failed"));
const providerRequestError = classifyProviderRequestError(embeddedError);
turn.replyOperation?.fail("run_failed", embeddedError);
const embeddedErrorText = formatErrorMessage(embeddedError).replace(/\.\s*$/, "");
return {
kind: "final",
payload: markAgentRunFailureReplyPayload({
text: params.shouldSurfaceToControlUi
? `⚠️ Agent failed before reply: ${embeddedErrorText}.\nLogs: openclaw logs --follow`
: (providerRequestError?.userMessage ?? PROVIDER_CONVERSATION_STATE_ERROR_USER_MESSAGE),
}),
};
}
const terminalMetadata = resolveAgentLifecycleTerminalMetadata(runResult.meta);
let terminalRunFailed = false;
if (fallbackExhausted) {
const exhaustionError = new Error(
terminalErrorMessage ?? "All model fallback candidates failed",
);
terminalRunFailed = true;
if (params.modelPatch.captureFallbackFailure(fallbackAttempts) === undefined) {
params.modelPatch.captureFailure(embeddedError ?? exhaustionError);
}
emitSettledLifecycleError(exhaustionError, {
...terminalMetadata,
fallbackExhaustedFailure: true,
});
turn.replyOperation?.retainFailureUntilComplete();
turn.replyOperation?.fail("run_failed", exhaustionError);
} else if (deferredLifecycleError || embeddedError) {
const terminalError = new Error(terminalErrorMessage ?? "Agent run failed");
terminalRunFailed = true;
params.modelPatch.captureFailure(embeddedError ?? terminalError);
emitSettledLifecycleError(terminalError, terminalMetadata);
turn.replyOperation?.retainFailureUntilComplete();
turn.replyOperation?.fail("run_failed", terminalError);
} else {
settledLifecycleTerminal?.emit("end", runResult);
}
return {
kind: "completed",
runResult,
fallbackProvider,
fallbackModel,
fallbackExhausted,
fallbackAttempts,
terminalRunFailed,
};
}

View File

@@ -0,0 +1,135 @@
import { resolveSendableOutboundReplyParts } from "openclaw/plugin-sdk/reply-payload";
import { sanitizeUserFacingText } from "../../agents/embedded-agent-helpers/sanitize-user-facing-text.js";
import { logVerbose } from "../../globals.js";
import { stripHeartbeatToken } from "../heartbeat.js";
import {
HEARTBEAT_TOKEN,
isSilentReplyPrefixText,
isSilentReplyText,
SILENT_REPLY_TOKEN,
startsWithSilentToken,
stripLeadingSilentToken,
} from "../tokens.js";
import type { ReplyPayload } from "../types.js";
import type { AgentTurnParams } from "./agent-runner-execution.types.js";
import { createBlockReplyDeliveryHandler } from "./reply-delivery.js";
import type { ReplyMediaContext } from "./reply-media-paths.js";
type AgentTurnPresentation = {
normalizeStreamingText: (payload: ReplyPayload) => { text?: string; skip: boolean };
preparePartialForTyping: (payload: ReplyPayload) => string | undefined;
handlePartialForTyping: (payload: ReplyPayload) => Promise<string | undefined>;
startPresentationWhileTyping: (
typingPromise: Promise<void>,
startPresentation: () => void | Promise<void>,
) => Promise<void>;
blockReplyHandler: ReturnType<typeof createBlockReplyDeliveryHandler> | undefined;
};
/** Builds the channel-presentation callbacks shared by CLI and embedded runs. */
export function createAgentTurnPresentation(params: {
turn: AgentTurnParams;
replyMediaContext: ReplyMediaContext;
directlySentBlockKeys: Set<string>;
directlySentBlockPayloads: Array<ReplyPayload | undefined>;
heartbeatState: { didLogStrip: boolean };
}): AgentTurnPresentation {
const normalizeStreamingText = (payload: ReplyPayload): { text?: string; skip: boolean } => {
let text = payload.text;
const reply = resolveSendableOutboundReplyParts(payload);
if (params.turn.followupRun.run.silentExpected) {
return { skip: true };
}
if (!params.turn.isHeartbeat && text?.includes("HEARTBEAT_OK")) {
const stripped = stripHeartbeatToken(text, { mode: "message" });
if (stripped.didStrip && !params.heartbeatState.didLogStrip) {
params.heartbeatState.didLogStrip = true;
logVerbose("Stripped stray HEARTBEAT_OK token from reply");
}
if (stripped.shouldSkip && !reply.hasMedia) {
return { skip: true };
}
text = stripped.text;
}
if (isSilentReplyText(text, SILENT_REPLY_TOKEN)) {
return { skip: true };
}
if (
isSilentReplyPrefixText(text, SILENT_REPLY_TOKEN) ||
isSilentReplyPrefixText(text, HEARTBEAT_TOKEN)
) {
return { skip: true };
}
if (text && startsWithSilentToken(text, SILENT_REPLY_TOKEN)) {
text = stripLeadingSilentToken(text, SILENT_REPLY_TOKEN);
}
if (!text) {
return reply.hasMedia ? { text: undefined, skip: false } : { skip: true };
}
const sanitized = sanitizeUserFacingText(text, {
errorContext: Boolean(payload.isError),
});
return sanitized.trim() ? { text: sanitized, skip: false } : { skip: true };
};
const preparePartialForTyping = (payload: ReplyPayload): string | undefined => {
if (isSilentReplyPrefixText(payload.text, SILENT_REPLY_TOKEN)) {
return undefined;
}
const { text, skip } = normalizeStreamingText(payload);
return skip || !text ? undefined : text;
};
const handlePartialForTyping = async (payload: ReplyPayload): Promise<string | undefined> => {
const text = preparePartialForTyping(payload);
if (text === undefined) {
return undefined;
}
await params.turn.typingSignals.signalTextDelta(text);
return text;
};
const startPresentationWhileTyping = async (
typingPromise: Promise<void>,
startPresentation: () => void | Promise<void>,
) => {
let presentationPromise: void | Promise<void>;
try {
presentationPromise = startPresentation();
} catch (err) {
// Typing already started; observe a secondary failure if presentation throws inline.
void typingPromise.catch(() => undefined);
throw err;
}
await Promise.all([typingPromise, presentationPromise]);
};
const blockReplyPipeline = params.turn.blockReplyPipeline;
// One handler owns threading and direct-send dedupe for this fallback cycle.
const blockReplyHandler = params.turn.opts?.onBlockReply
? createBlockReplyDeliveryHandler({
onBlockReply: params.turn.opts.onBlockReply,
currentMessageId:
params.turn.sessionCtx.MessageSidFull ?? params.turn.sessionCtx.MessageSid,
replyThreading: params.turn.replyThreading,
normalizeStreamingText,
applyReplyToMode: params.turn.applyReplyToMode,
normalizeMediaPaths: params.replyMediaContext.normalizePayload,
typingSignals: params.turn.typingSignals,
reasoningPayloadsEnabled: params.turn.opts?.reasoningPayloadsEnabled,
commentaryPayloadsEnabled: params.turn.opts?.commentaryPayloadsEnabled,
blockStreamingEnabled: params.turn.blockStreamingEnabled,
blockReplyPipeline,
directlySentBlockKeys: params.directlySentBlockKeys,
directlySentBlockPayloads: params.directlySentBlockPayloads,
})
: undefined;
return {
normalizeStreamingText,
preparePartialForTyping,
handlePartialForTyping,
startPresentationWhileTyping,
blockReplyHandler,
};
}

View File

@@ -0,0 +1,147 @@
import { createSubsystemLogger } from "../../logging/subsystem.js";
type AgentTurnTimingSpan = {
name: string;
durationMs: number;
elapsedMs: number;
};
type AgentTurnTimingSummary = {
totalMs: number;
spans: AgentTurnTimingSpan[];
};
export type AgentTurnTimingTracker = {
measure: <T>(name: string, run: () => Promise<T> | T) => Promise<T>;
measureSync: <T>(name: string, run: () => T) => T;
logIfSlow: (params: {
runId: string;
sessionId?: string;
sessionKey?: string;
outcome: "completed" | "error";
error?: string;
}) => void;
logMilestoneIfSlow: (params: {
runId: string;
sessionId?: string;
sessionKey?: string;
milestone: string;
}) => void;
};
const agentTurnTimingLog = createSubsystemLogger("auto-reply/agent-turn-timing");
const AGENT_TURN_TIMING_WARN_TOTAL_MS = 1_000;
const AGENT_TURN_TIMING_WARN_STAGE_MS = 500;
/** Creates a no-overhead pass-through unless reply profiling is enabled. */
export function createAgentTurnTimingTracker(
options: {
profilerEnabled?: boolean;
} = {},
): AgentTurnTimingTracker {
if (!options.profilerEnabled) {
// This tracker wraps the agent-turn hot path. Without an explicit profiler
// flag, keep every wrapper pass-through so normal turns avoid Date.now and
// span-array work entirely.
return {
async measure(_name, run) {
return await run();
},
measureSync(_name, run) {
return run();
},
logIfSlow() {},
logMilestoneIfSlow() {},
};
}
const startedAt = Date.now();
let didLog = false;
const spans: AgentTurnTimingSpan[] = [];
const toMs = (value: number) => Math.max(0, Math.round(value));
const record = (name: string, spanStartedAt: number) => {
spans.push({
name,
durationMs: toMs(Date.now() - spanStartedAt),
elapsedMs: toMs(Date.now() - startedAt),
});
};
const snapshot = (): AgentTurnTimingSummary => ({
totalMs: toMs(Date.now() - startedAt),
spans: spans.slice(),
});
const shouldLog = (summary: AgentTurnTimingSummary) =>
summary.totalMs >= AGENT_TURN_TIMING_WARN_TOTAL_MS ||
summary.spans.some((span) => span.durationMs >= AGENT_TURN_TIMING_WARN_STAGE_MS);
const formatSpans = (summary: AgentTurnTimingSummary) =>
summary.spans.length > 0
? summary.spans
.map((span) => `${span.name}:${span.durationMs}ms@${span.elapsedMs}ms`)
.join(",")
: "none";
return {
async measure(name, run) {
const spanStartedAt = Date.now();
try {
return await run();
} finally {
record(name, spanStartedAt);
}
},
measureSync(name, run) {
const spanStartedAt = Date.now();
try {
return run();
} finally {
record(name, spanStartedAt);
}
},
logIfSlow(params) {
if (didLog) {
return;
}
const summary = snapshot();
if (!shouldLog(summary)) {
return;
}
didLog = true;
agentTurnTimingLog.warn(
`agent turn timings runId=${params.runId} sessionId=${
params.sessionId ?? "unknown"
} sessionKey=${params.sessionKey ?? "unknown"} outcome=${params.outcome} totalMs=${
summary.totalMs
} stages=${formatSpans(summary)}${params.error ? ` error="${params.error}"` : ""}`,
{
runId: params.runId,
sessionId: params.sessionId,
sessionKey: params.sessionKey,
outcome: params.outcome,
error: params.error,
totalMs: summary.totalMs,
spans: summary.spans,
},
);
},
logMilestoneIfSlow(params) {
const summary = snapshot();
if (!shouldLog(summary)) {
return;
}
agentTurnTimingLog.warn(
`agent turn milestone runId=${params.runId} sessionId=${
params.sessionId ?? "unknown"
} sessionKey=${params.sessionKey ?? "unknown"} milestone=${params.milestone} totalMs=${
summary.totalMs
} stages=${formatSpans(summary)}`,
{
runId: params.runId,
sessionId: params.sessionId,
sessionKey: params.sessionKey,
milestone: params.milestone,
totalMs: summary.totalMs,
spans: summary.spans,
},
);
},
};
}

View File

@@ -61,8 +61,12 @@ vi.mock("../../media/outbound-attachment.js", () => ({
resolveOutboundAttachmentFromUrlMock(...args),
}));
vi.mock("./agent-runner-execution.js", () => ({
vi.mock("./agent-runner-failure-reply.js", () => ({
buildEmptyInteractiveReplyPayload: vi.fn(() => undefined),
buildKnownAgentRunFailureReplyPayload: vi.fn(() => undefined),
}));
vi.mock("./agent-runner-execution.js", () => ({
runAgentTurnWithFallback: (...args: unknown[]) => runAgentTurnWithFallbackMock(...args),
}));

View File

@@ -78,11 +78,11 @@ import type { OriginatingChannelType, TemplateContext } from "../templating.js";
import type { VerboseLevel } from "../thinking.js";
import { SILENT_REPLY_TOKEN } from "../tokens.js";
import type { GetReplyOptions, ReplyPayload } from "../types.js";
import { runAgentTurnWithFallback } from "./agent-runner-execution.js";
import {
buildEmptyInteractiveReplyPayload,
buildKnownAgentRunFailureReplyPayload,
runAgentTurnWithFallback,
} from "./agent-runner-execution.js";
} from "./agent-runner-failure-reply.js";
import {
createShouldEmitToolOutput,
createShouldEmitToolResult,

View File

@@ -70,6 +70,7 @@ import {
resolveAgentLifecycleTerminalMetadata,
type AgentLifecycleTerminalBackstop,
} from "./agent-lifecycle-terminal.js";
import { resolveRunAfterAutoFallbackPrimaryProbeRecheck } from "./agent-runner-auto-fallback.js";
import {
clearDroppedCliSessionBinding,
createCliReasoningStreamBridge,
@@ -77,13 +78,12 @@ import {
keepCliSessionBindingOnlyWhenReused,
runCliAgentWithLifecycle,
} from "./agent-runner-cli-dispatch.js";
import { buildCommandOutputFromToolResultEvent } from "./agent-runner-command-output.js";
import {
buildEmptyInteractiveReplyPayload,
buildTerminalAgentRunFailureReplyPayload,
buildCommandOutputFromToolResultEvent,
buildPreflightCompactionFailureText,
resolveRunAfterAutoFallbackPrimaryProbeRecheck,
} from "./agent-runner-execution.js";
buildTerminalAgentRunFailureReplyPayload,
} from "./agent-runner-failure-reply.js";
import { runPreflightCompactionIfNeeded } from "./agent-runner-memory.js";
import { appendUsageLine, resolveResponseUsageLine } from "./agent-runner-usage-line.js";
import {