From f67e55ac002ebe596bc2c545475ca38fdf9b54da Mon Sep 17 00:00:00 2001 From: Peter Steinberger Date: Sun, 19 Jul 2026 23:08:55 -0700 Subject: [PATCH] refactor: unify the embedded-agent fallback run-entry across all four pipelines (#111608) * refactor(agents): unify fallback run-entry across channel, command, memory, and follow-up paths * refactor(agents): keep run-entry contracts module-local * test(reply): derive run-entry params type locally --- src/agents/command/lifecycle.ts | 58 ++-- src/agents/command/post-run.ts | 5 +- src/agents/command/run-embedded-attempt.ts | 161 +++++---- .../embedded-agent-runner/run-entry.test.ts | 262 ++++++++++++++ src/agents/embedded-agent-runner/run-entry.ts | 319 ++++++++++++++++++ src/agents/runtime-plan/build.ts | 2 +- .../agent-runner-execution.test-support.ts | 21 ++ .../reply/agent-runner-fallback-candidate.ts | 78 +++-- .../reply/agent-runner-fallback-settlement.ts | 8 +- .../reply/agent-runner-memory.test.ts | 77 ++++- src/auto-reply/reply/agent-runner-memory.ts | 81 ++--- src/auto-reply/reply/followup-runner.ts | 115 +++---- 12 files changed, 912 insertions(+), 275 deletions(-) create mode 100644 src/agents/embedded-agent-runner/run-entry.test.ts create mode 100644 src/agents/embedded-agent-runner/run-entry.ts diff --git a/src/agents/command/lifecycle.ts b/src/agents/command/lifecycle.ts index ae9b9bb6845c..9a3fff79b7c6 100644 --- a/src/agents/command/lifecycle.ts +++ b/src/agents/command/lifecycle.ts @@ -1,6 +1,10 @@ import { emitAgentEvent } from "../../infra/agent-events.js"; import { createSubsystemLogger } from "../../logging/subsystem.js"; -import { buildAgentRunTerminalOutcome } from "../agent-run-terminal-outcome.js"; +import { + buildAgentRunTerminalOutcome, + type AgentRunTerminalOutcome, +} from "../agent-run-terminal-outcome.js"; +import type { EmbeddedAgentRunEntryTerminal } from "../embedded-agent-runner/run-entry.js"; import { resolveAgentRunAbortLifecycleFields, resolveAgentRunErrorLifecycleFields, @@ -10,6 +14,18 @@ import type { AgentAttemptResult } from "./runtime-loaders.js"; const log = createSubsystemLogger("agents/agent-command"); +function resolveTerminalLogLevel( + outcome: AgentRunTerminalOutcome, +): "info" | "warn" | "error" | undefined { + if (!outcome.stopReason || outcome.stopReason === "end_turn") { + return undefined; + } + if (outcome.reason === "completed") { + return "info"; + } + return outcome.status === "timeout" ? "warn" : "error"; +} + export function resolveAgentRunLifecycleEndLogLevel(meta: { aborted?: unknown; error?: unknown; @@ -32,13 +48,7 @@ export function resolveAgentRunLifecycleEndLogLevel(meta: { timeoutPhase: meta.timeoutPhase, providerStarted: meta.providerStarted, }); - if (!outcome.stopReason || outcome.stopReason === "end_turn") { - return undefined; - } - if (outcome.reason === "completed") { - return "info"; - } - return outcome.status === "timeout" ? "warn" : "error"; + return resolveTerminalLogLevel(outcome); } export function applyAgentRunAbortMetadata( @@ -76,7 +86,7 @@ export function createAgentCommandLifecycle(params: { (runResult.meta.error ? "Agent run failed" : undefined); return { - emitFinishing(runResult: AgentAttemptResult) { + emitFinishing(terminal: EmbeddedAgentRunEntryTerminal) { if ( params.state.lifecycleEnded || params.state.lifecycleFinishing || @@ -94,19 +104,19 @@ export function createAgentCommandLifecycle(params: { phase: "finishing", startedAt: params.startedAt, endedAt: Date.now(), - aborted: runResult.meta.aborted ?? false, - stopReason: runResult.meta.stopReason, + aborted: terminal.metadata.aborted ?? false, + stopReason: terminal.outcome.stopReason, ...resolveAgentRunAbortLifecycleFields(params.abortSignal), }, }); }, - emitEnd(runResult: AgentAttemptResult) { + emitEnd(terminal: EmbeddedAgentRunEntryTerminal) { if (params.state.lifecycleEnded) { return; } params.state.lifecycleEnded = true; - const stopReason = runResult.meta.stopReason; - const logLevel = resolveAgentRunLifecycleEndLogLevel(runResult.meta); + const stopReason = terminal.outcome.stopReason; + const logLevel = resolveTerminalLogLevel(terminal.outcome); if (logLevel) { log[logLevel](`[agent] run ${params.runId} ended with stopReason=${stopReason}`); } @@ -118,14 +128,18 @@ export function createAgentCommandLifecycle(params: { phase: "end", startedAt: params.startedAt, endedAt: Date.now(), - aborted: runResult.meta.aborted ?? false, + aborted: terminal.metadata.aborted ?? false, stopReason, ...resolveAgentRunAbortLifecycleFields(params.abortSignal), }, }); }, resolveResultError, - emitResultError(runResult: AgentAttemptResult, fallbackExhausted: boolean) { + emitResultError( + runResult: AgentAttemptResult, + fallbackExhausted: boolean, + terminal: EmbeddedAgentRunEntryTerminal, + ) { if (params.state.lifecycleEnded) { return; } @@ -142,17 +156,7 @@ export function createAgentCommandLifecycle(params: { startedAt: params.startedAt, endedAt: Date.now(), error, - ...(runResult.meta.stopReason ? { stopReason: runResult.meta.stopReason } : {}), - ...(runResult.meta.livenessState ? { livenessState: runResult.meta.livenessState } : {}), - ...(runResult.meta.timeoutPhase ? { timeoutPhase: runResult.meta.timeoutPhase } : {}), - ...(typeof runResult.meta.providerStarted === "boolean" - ? { providerStarted: runResult.meta.providerStarted } - : {}), - ...(typeof runResult.meta.aborted === "boolean" - ? { aborted: runResult.meta.aborted } - : {}), - ...(runResult.meta.replayInvalid === true ? { replayInvalid: true } : {}), - ...(runResult.meta.yielded === true ? { yielded: true } : {}), + ...terminal.metadata, ...(fallbackExhausted ? { fallbackExhaustedFailure: true } : {}), }, }); diff --git a/src/agents/command/post-run.ts b/src/agents/command/post-run.ts index 5d58eaa2b620..bac9a9a8cd4e 100644 --- a/src/agents/command/post-run.ts +++ b/src/agents/command/post-run.ts @@ -82,6 +82,7 @@ export async function finalizeEmbeddedAgentCommand(params: { userTurnTranscriptRecorder, fallbackTrajectoryRecorder, lifecycle, + terminal, lifecycleGeneration, } = params.attempt; const { skillsSnapshot, runContext } = params.embeddedSessionState; @@ -345,9 +346,9 @@ export async function finalizeEmbeddedAgentCommand(params: { } if (fallbackExhausted || lifecycle.resolveResultError(result, false)) { - lifecycle.emitResultError(result, fallbackExhausted); + lifecycle.emitResultError(result, fallbackExhausted, terminal); } else { - lifecycle.emitEnd(result); + lifecycle.emitEnd(terminal); } return { deliveryResult, diff --git a/src/agents/command/run-embedded-attempt.ts b/src/agents/command/run-embedded-attempt.ts index 77d7facad91c..5fca98acaad0 100644 --- a/src/agents/command/run-embedded-attempt.ts +++ b/src/agents/command/run-embedded-attempt.ts @@ -21,15 +21,13 @@ import { resolveEffectiveModelFallbacks, } from "../agent-scope.js"; import { - classifyEmbeddedAgentRunResultForModelFallback, - mergeEmbeddedAgentRunResultForModelFallbackExhaustion, -} from "../embedded-agent-runner/result-fallback-classifier.js"; + runEmbeddedAgentEntry, + type EmbeddedAgentRunEntryTerminal, +} from "../embedded-agent-runner/run-entry.js"; import { resolveFastModeState } from "../fast-mode.js"; import { runAgentHarnessBeforeMessageWriteHook } from "../harness/hook-helpers.js"; -import { ensureSelectedAgentHarnessPlugin } from "../harness/runtime-plugin.js"; import { prepareInternalSessionEffectsSession } from "../internal-session-effects.js"; import { LiveSessionModelSwitchError } from "../live-model-switch.js"; -import { runWithModelFallback } from "../model-fallback.js"; import { modelKey, resolveThinkingDefault } from "../model-selection.js"; import type { AgentRunSessionTarget } from "../run-session-target.js"; import { @@ -46,7 +44,7 @@ import { createAgentAttemptLifecycleCallbacks, type AgentAttemptLifecycleState, } from "./attempt-callbacks.js"; -import { applyAgentRunAbortMetadata, createAgentCommandLifecycle } from "./lifecycle.js"; +import { createAgentCommandLifecycle } from "./lifecycle.js"; import { normalizeAgentCommandModelRef } from "./model-ref.js"; import type { EmbeddedModelSelection } from "./model-selection.js"; import type { PreparedAgentCommandExecution } from "./prepare.js"; @@ -190,6 +188,7 @@ export async function runEmbeddedAgentAttempt(params: { let fallbackProvider = provider; let fallbackModel = model; let fallbackExhausted = false; + let terminal: EmbeddedAgentRunEntryTerminal; let liveSwitchRetries = 0; let autoFallbackPrimaryProbeInterruptedByLiveSwitch = false; const fastModeStartedAtMs = Date.now(); @@ -224,7 +223,6 @@ export async function runEmbeddedAgentAttempt(params: { : hasStoredAutoFallbackProvenance, }); - let fallbackAttemptIndex = 0; const fallbackRuntimeState: { originRuntime?: "cli" | "embedded" } = {}; attemptLifecycleState.currentTurnUserMessagePersisted = false; let attemptMediaTaskIds = liveSwitchMediaTaskIds; @@ -232,53 +230,75 @@ export async function runEmbeddedAgentAttempt(params: { Boolean( sessionKey && hasNewGeneratedMediaTaskForSessionKey(sessionKey, attemptMediaTaskIds), ); - const fallbackResult = await runWithModelFallback({ - cfg, - provider, - model, - ...modelManifestContext, - runId, - agentDir, - agentId: sessionAgentId, - sessionId, - sessionKey: sessionKey ?? sessionId, - resolveAgentHarnessRuntimeOverride: (candidateProvider) => - resolveSessionRuntimeOverrideForProvider({ - provider: candidateProvider, - entry: sessionEntryForAttempt, - cfg, - }), - prepareAgentHarnessRuntime: async ({ - provider: providerValue, - model: modelValue, - agentHarnessRuntimeOverride, - }) => { - await ensureSelectedAgentHarnessPlugin({ - config: cfg, - provider: providerValue, - modelId: modelValue, - agentId: sessionAgentId, - sessionKey, - agentHarnessRuntimeOverride, - workspaceDir, - }); + const fallbackResult = await runEmbeddedAgentEntry({ + selection: { + cfg, + provider, + model, + agentDir, + fallbacksOverride: effectiveFallbacksOverride, + ...modelManifestContext, }, - fallbacksOverride: effectiveFallbacksOverride, + identity: { + runId, + agentId: sessionAgentId, + sessionId, + sessionKey: sessionKey ?? sessionId, + }, + harness: { + workspaceDir, + sessionKey, + preparation: { kind: "direct" }, + resolveRuntimeOverride: (candidateProvider) => + resolveSessionRuntimeOverrideForProvider({ + provider: candidateProvider, + entry: sessionEntryForAttempt, + cfg, + }), + }, + behavior: { + kind: "command-rpc", + hasCommittedSideEffect: currentAttemptCommittedCronMedia, + }, + sessionOverride: { + kind: "reconcile-completed", + reconcile: async ({ provider: winnerProvider, model: winnerModel }) => { + if ( + !autoFallbackPrimaryProbe || + autoFallbackPrimaryProbeInterruptedByLiveSwitch || + !sessionEntry || + !sessionStore || + !sessionKey || + isModelSelectionLocked(sessionEntry) || + params.suppressVisibleSessionEffects || + params.preserveUserFacingSessionModelState || + !entryMatchesAutoFallbackPrimaryProbe(sessionEntry, autoFallbackPrimaryProbe) || + winnerProvider !== autoFallbackPrimaryProbe.provider || + winnerModel !== autoFallbackPrimaryProbe.model + ) { + return; + } + const nextSessionEntry = { ...sessionEntry }; + clearAutoFallbackPrimaryProbeSelection(nextSessionEntry); + sessionEntry = await persistSessionEntry({ + sessionStore, + sessionKey, + storePath, + initialEntry: sessionEntry, + entry: nextSessionEntry, + shouldPersist: (current) => + Boolean( + current && + entryMatchesAutoFallbackPrimaryProbe(current, autoFallbackPrimaryProbe), + ), + }); + }, + }, + abortSignal: params.opts.abortSignal, onFallbackStep: (step) => { fallbackTrajectoryRecorder?.recordEvent("model.fallback_step", step); }, - classifyResult: ({ provider: providerLocal, model: modelLocal, result: resultLocal }) => { - const classification = classifyEmbeddedAgentRunResultForModelFallback({ - provider: providerLocal, - model: modelLocal, - result: resultLocal, - }); - return classification && currentAttemptCommittedCronMedia() ? undefined : classification; - }, - canFallbackAfterError: () => !currentAttemptCommittedCronMedia(), - mergeExhaustedResult: mergeEmbeddedAgentRunResultForModelFallbackExhaustion, - abortSignal: params.opts.abortSignal, - run: async (providerOverride, modelOverride, runOptions) => { + runCandidate: async (providerOverride, modelOverride, runOptions) => { attemptMediaTaskIds = sessionKey ? getGeneratedMediaTaskIdsForSessionKey(sessionKey) : new Set(); @@ -298,8 +318,6 @@ export async function runEmbeddedAgentAttempt(params: { if (isAutoFallbackPrimaryProbeCandidate) { markAutoFallbackPrimaryProbe({ probe: autoFallbackPrimaryProbe, sessionKey }); } - const isFallbackRetry = fallbackAttemptIndex > 0; - fallbackAttemptIndex += 1; await params.opts.onActiveModelSelected?.({ provider: providerOverride, model: modelOverride, @@ -369,7 +387,7 @@ export async function runEmbeddedAgentAttempt(params: { cwd, body, transcriptBody, - isFallbackRetry, + isFallbackRetry: runOptions.isFallbackRetry, resolvedThinkLevel: candidateThinkLevel, fastMode, fastModeStartedAtMs, @@ -403,7 +421,7 @@ export async function runEmbeddedAgentAttempt(params: { suppressUserTurnPersistence || userTurnTranscriptRecorder.hasPersisted() || userTurnTranscriptRecorder.isBlocked() || - (isFallbackRetry && attemptLifecycleState.currentTurnUserMessagePersisted), + (runOptions.isFallbackRetry && attemptLifecycleState.currentTurnUserMessagePersisted), userTurnTranscriptRecorder, onUserMessagePersisted: attemptLifecycleCallbacks.onUserMessagePersisted, onLifecycleGenerationChanged: (nextLifecycleGeneration) => { @@ -416,41 +434,15 @@ export async function runEmbeddedAgentAttempt(params: { }); }, }); - result = applyAgentRunAbortMetadata(fallbackResult.result, params.opts.abortSignal); + result = fallbackResult.result; + terminal = fallbackResult.terminal; if (isAgentRunRestartAbortReason(params.opts.abortSignal?.reason)) { throw params.opts.abortSignal?.reason; } fallbackProvider = fallbackResult.provider; fallbackModel = fallbackResult.model; fallbackExhausted = fallbackResult.outcome === "exhausted"; - if ( - !fallbackExhausted && - autoFallbackPrimaryProbe && - !autoFallbackPrimaryProbeInterruptedByLiveSwitch && - sessionEntry && - sessionStore && - sessionKey && - !isModelSelectionLocked(sessionEntry) && - !params.suppressVisibleSessionEffects && - !params.preserveUserFacingSessionModelState && - entryMatchesAutoFallbackPrimaryProbe(sessionEntry, autoFallbackPrimaryProbe) && - fallbackProvider === autoFallbackPrimaryProbe.provider && - fallbackModel === autoFallbackPrimaryProbe.model - ) { - const nextSessionEntry = { ...sessionEntry }; - clearAutoFallbackPrimaryProbeSelection(nextSessionEntry); - sessionEntry = await persistSessionEntry({ - sessionStore, - sessionKey, - storePath, - initialEntry: sessionEntry, - entry: nextSessionEntry, - shouldPersist: (current) => - Boolean( - current && entryMatchesAutoFallbackPrimaryProbe(current, autoFallbackPrimaryProbe), - ), - }); - } + await fallbackResult.settleSessionOverride(); if (fallbackResult.attempts.length > 0 && result.meta.agentMeta) { result = { ...result, @@ -464,7 +456,7 @@ export async function runEmbeddedAgentAttempt(params: { }; } if (!fallbackExhausted) { - lifecycle.emitFinishing(result); + lifecycle.emitFinishing(terminal); } break; } catch (err) { @@ -625,6 +617,7 @@ export async function runEmbeddedAgentAttempt(params: { userTurnTranscriptRecorder, fallbackTrajectoryRecorder, lifecycle, + terminal, }; } diff --git a/src/agents/embedded-agent-runner/run-entry.test.ts b/src/agents/embedded-agent-runner/run-entry.test.ts new file mode 100644 index 000000000000..8451e8df86c1 --- /dev/null +++ b/src/agents/embedded-agent-runner/run-entry.test.ts @@ -0,0 +1,262 @@ +import { beforeEach, describe, expect, it, vi } from "vitest"; +import type { OpenClawConfig } from "../../config/types.openclaw.js"; +import type { EmbeddedAgentRunResult } from "./types.js"; + +type CandidateOptions = { + allowTransientCooldownProbe?: boolean; + isFinalFallbackAttempt?: boolean; +}; + +type FallbackRunnerParams = { + provider: string; + model: string; + prepareAgentHarnessRuntime?: (params: { + provider: string; + model: string; + agentHarnessRuntimeOverride?: string; + }) => Promise | void; + classifyResult?: (params: { + result: EmbeddedAgentRunResult; + provider: string; + model: string; + attempt: number; + total: number; + }) => unknown; + mergeExhaustedResult?: (params: { + latestResult: EmbeddedAgentRunResult; + preferredResult: EmbeddedAgentRunResult; + }) => EmbeddedAgentRunResult; + run: ( + provider: string, + model: string, + options?: CandidateOptions, + ) => Promise; +}; + +const state = vi.hoisted(() => ({ + runWithModelFallback: vi.fn(), + ensureSelectedAgentHarnessPlugin: vi.fn(async (_params: unknown) => undefined), +})); + +vi.mock("../model-fallback.js", () => ({ + runWithModelFallback: (params: FallbackRunnerParams) => state.runWithModelFallback(params), +})); + +vi.mock("../harness/runtime-plugin.js", () => ({ + ensureSelectedAgentHarnessPlugin: (params: unknown) => + state.ensureSelectedAgentHarnessPlugin(params), +})); + +function makeResult(params: { + provider: string; + model: string; + classification?: "empty"; +}): EmbeddedAgentRunResult { + return { + payloads: params.classification ? [] : [{ text: "recovered" }], + meta: { + durationMs: 10, + aborted: false, + yielded: true, + providerStarted: true, + stopReason: "end_turn", + agentHarnessResultClassification: params.classification, + agentMeta: { + sessionId: "session-1", + provider: params.provider, + model: params.model, + }, + }, + }; +} + +describe("runEmbeddedAgentEntry", () => { + beforeEach(() => { + state.ensureSelectedAgentHarnessPlugin.mockClear(); + state.runWithModelFallback + .mockReset() + .mockImplementation(async (params: FallbackRunnerParams) => { + await params.prepareAgentHarnessRuntime?.({ + provider: params.provider, + model: params.model, + }); + const primaryResult = await params.run(params.provider, params.model, { + allowTransientCooldownProbe: true, + }); + const classification = await params.classifyResult?.({ + result: primaryResult, + provider: params.provider, + model: params.model, + attempt: 1, + total: 2, + }); + expect(classification).toBeTruthy(); + const fallbackProvider = "fallback-provider"; + const fallbackModel = "fallback-model"; + await params.prepareAgentHarnessRuntime?.({ + provider: fallbackProvider, + model: fallbackModel, + }); + const result = await params.run(fallbackProvider, fallbackModel, { + isFinalFallbackAttempt: true, + }); + return { + outcome: "completed" as const, + result, + provider: fallbackProvider, + model: fallbackModel, + attempts: [ + { + provider: params.provider, + model: params.model, + error: "empty result", + reason: "format" as const, + }, + ], + }; + }); + }); + + it("keeps shared fallback and terminal behavior aligned across entry modes", async () => { + const { runEmbeddedAgentEntry } = await import("./run-entry.js"); + const cfg: OpenClawConfig = {}; + const runMode = async (behavior: "channel-delivery" | "command-rpc") => { + const candidateCalls: Array<{ + provider: string; + model: string; + isFallbackRetry: boolean; + }> = []; + const reconciled: Array<{ provider: string; model: string }> = []; + const result = await runEmbeddedAgentEntry({ + selection: { cfg, provider: "primary-provider", model: "primary-model" }, + identity: { + runId: `run-${behavior}`, + agentId: "main", + sessionId: "session-1", + }, + harness: { + workspaceDir: "/tmp/workspace", + preparation: { kind: "direct" }, + resolveRuntimeOverride: () => undefined, + }, + behavior: + behavior === "channel-delivery" + ? { + kind: "channel-delivery" as const, + readDeliveryEvidence: () => ({ + hasDirectlySentBlockReply: false, + hasBlockReplyPipelineOutput: false, + }), + } + : { + kind: "command-rpc" as const, + hasCommittedSideEffect: () => false, + }, + sessionOverride: { + kind: "reconcile-completed", + reconcile: async (candidate) => { + reconciled.push(candidate); + }, + }, + runCandidate: async (provider, model, options) => { + candidateCalls.push({ provider, model, isFallbackRetry: options.isFallbackRetry }); + return makeResult({ + provider, + model, + classification: options.isFallbackRetry ? undefined : "empty", + }); + }, + }); + await result.settleSessionOverride(); + await result.settleSessionOverride(); + return { result, candidateCalls, reconciled }; + }; + + const channel = await runMode("channel-delivery"); + const command = await runMode("command-rpc"); + + expect(channel.candidateCalls).toEqual(command.candidateCalls); + expect(channel.result.outcome).toBe("completed"); + expect(channel.result.provider).toBe("fallback-provider"); + expect(channel.result.model).toBe("fallback-model"); + expect(channel.result.attempts).toEqual(command.result.attempts); + expect(channel.result.terminal).toEqual(command.result.terminal); + expect(channel.reconciled).toEqual(command.reconciled); + expect(channel.reconciled).toEqual([ + { provider: "fallback-provider", model: "fallback-model" }, + ]); + }); + + it("leaves maintenance fallback classification to thrown candidate errors", async () => { + state.runWithModelFallback.mockImplementationOnce(async (params: FallbackRunnerParams) => { + expect(params.classifyResult).toBeUndefined(); + expect(params.mergeExhaustedResult).toBeUndefined(); + const result = await params.run(params.provider, params.model); + return { + outcome: "completed" as const, + result, + provider: params.provider, + model: params.model, + attempts: [], + }; + }); + const { runEmbeddedAgentEntry } = await import("./run-entry.js"); + const result = await runEmbeddedAgentEntry({ + selection: { cfg: {}, provider: "primary-provider", model: "primary-model" }, + identity: { runId: "maintenance", agentId: "main", sessionId: "session-1" }, + harness: { + workspaceDir: "/tmp/workspace", + preparation: { kind: "direct" }, + resolveRuntimeOverride: () => undefined, + }, + behavior: { kind: "maintenance" }, + sessionOverride: { kind: "preserve" }, + runCandidate: async (provider, model) => makeResult({ provider, model }), + }); + + expect(result.result.payloads).toEqual([{ text: "recovered" }]); + }); + + it("retains non-visible follow-up results for terminal delivery", async () => { + state.runWithModelFallback.mockImplementationOnce(async (params: FallbackRunnerParams) => { + const result = await params.run(params.provider, params.model); + expect( + params.classifyResult?.({ + result, + provider: params.provider, + model: params.model, + attempt: 1, + total: 1, + }), + ).toMatchObject({ + code: "empty_result", + preserveResultOnExhaustion: true, + preserveResultPriority: -1, + }); + return { + outcome: "exhausted" as const, + result, + provider: params.provider, + model: params.model, + attempts: [], + }; + }); + const { runEmbeddedAgentEntry } = await import("./run-entry.js"); + const result = await runEmbeddedAgentEntry({ + selection: { cfg: {}, provider: "primary-provider", model: "primary-model" }, + identity: { runId: "followup", agentId: "main", sessionId: "session-1" }, + harness: { + workspaceDir: "/tmp/workspace", + preparation: { kind: "direct" }, + resolveRuntimeOverride: () => undefined, + }, + behavior: { kind: "followup-delivery" }, + sessionOverride: { kind: "preserve" }, + runCandidate: async (provider, model) => + makeResult({ provider, model, classification: "empty" }), + }); + + expect(result.outcome).toBe("exhausted"); + expect(result.result.payloads).toEqual([]); + }); +}); diff --git a/src/agents/embedded-agent-runner/run-entry.ts b/src/agents/embedded-agent-runner/run-entry.ts new file mode 100644 index 000000000000..f9239ddc760b --- /dev/null +++ b/src/agents/embedded-agent-runner/run-entry.ts @@ -0,0 +1,319 @@ +import type { OpenClawConfig } from "../../config/types.openclaw.js"; +import { buildAgentRunTerminalOutcome } from "../agent-run-terminal-outcome.js"; +import { ensureSelectedAgentHarnessPlugin } from "../harness/runtime-plugin.js"; +import type { ModelFallbackStepFields } from "../model-fallback-observation.js"; +import { runWithModelFallback, type ModelFallbackResultClassification } from "../model-fallback.js"; +import type { FallbackAttempt } from "../model-fallback.types.js"; +import type { ModelManifestNormalizationContext } from "../model-selection-normalize.js"; +import { resolveAgentRunAbortLifecycleFields } from "../run-termination.js"; +import { + classifyEmbeddedAgentRunResultForModelFallback, + mergeEmbeddedAgentRunResultForModelFallbackExhaustion, +} from "./result-fallback-classifier.js"; +import type { EmbeddedAgentRunResult } from "./types.js"; + +type RunEntryCandidateOptions = { + allowTransientCooldownProbe?: boolean; + isFinalFallbackAttempt?: boolean; + isFallbackRetry: boolean; +}; + +type RunEntryHarnessPreparation = + | { kind: "direct" } + | { + kind: "measured"; + run: (prepare: () => Promise) => Promise; + }; + +type DeliveryEvidence = { + hasDirectlySentBlockReply: boolean; + hasBlockReplyPipelineOutput: boolean; +}; + +type RunEntryBehavior = + | { + kind: "channel-delivery"; + readDeliveryEvidence: () => DeliveryEvidence; + } + | { kind: "followup-delivery" } + | { + kind: "command-rpc"; + hasCommittedSideEffect: () => boolean; + } + | { kind: "maintenance" }; + +type RunEntrySessionOverride = + | { kind: "preserve" } + | { + kind: "reconcile-completed"; + reconcile: (candidate: { provider: string; model: string }) => Promise; + }; + +export type EmbeddedAgentRunEntryTerminal = { + outcome: ReturnType; + metadata: Record; +}; + +type EmbeddedAgentRunEntryResult = { + outcome: "completed" | "exhausted"; + result: T; + provider: string; + model: string; + attempts: FallbackAttempt[]; + terminal: EmbeddedAgentRunEntryTerminal; + settleSessionOverride: () => Promise; +}; + +type EmbeddedAgentRunEntryParams = { + selection: { + cfg: OpenClawConfig; + provider: string; + model: string; + fallbacksOverride?: string[]; + agentDir?: string; + } & ModelManifestNormalizationContext; + identity: { + runId: string; + agentId: string; + sessionId: string; + sessionKey?: string; + lane?: string; + }; + harness: { + workspaceDir: string; + sessionKey?: string; + preparation: RunEntryHarnessPreparation; + resolveRuntimeOverride: (provider: string, model: string) => string | undefined; + }; + behavior: RunEntryBehavior; + sessionOverride: RunEntrySessionOverride; + abortSignal?: AbortSignal; + onFallbackStep?: (step: ModelFallbackStepFields) => void | Promise; + runCandidate: (provider: string, model: string, options: RunEntryCandidateOptions) => Promise; +}; + +const PRESERVED_FOLLOWUP_RESULT_CODES = new Set([ + "empty_result", + "reasoning_only_result", + "planning_only_result", +]); + +function preserveFollowupResultForDelivery( + classification: ModelFallbackResultClassification, +): ModelFallbackResultClassification { + if ( + !classification || + !("code" in classification) || + !classification.code || + !PRESERVED_FOLLOWUP_RESULT_CODES.has(classification.code) + ) { + return classification; + } + // Follow-up delivery owns its terminal fallback, so retain the classified + // result for that layer instead of replacing it with a summary error. + return { + ...classification, + preserveResultOnExhaustion: true, + preserveResultPriority: -1, + }; +} + +function resolveTerminalStatus(params: { + result: EmbeddedAgentRunResult; + fallbackExhausted: boolean; +}): "ok" | "error" | "timeout" { + const meta = params.result.meta; + if (meta.stopReason === "timeout" || meta.timeoutPhase) { + return "timeout"; + } + if ( + params.fallbackExhausted || + meta.aborted === true || + meta.error || + meta.stopReason === "error" + ) { + return "error"; + } + return "ok"; +} + +function buildTerminal(params: { + result: EmbeddedAgentRunResult; + fallbackExhausted: boolean; + behavior: RunEntryBehavior; +}): EmbeddedAgentRunEntryTerminal { + const meta = params.result.meta; + const outcome = buildAgentRunTerminalOutcome({ + status: resolveTerminalStatus(params), + error: meta.error?.message, + stopReason: meta.stopReason, + livenessState: meta.livenessState, + timeoutPhase: meta.timeoutPhase, + providerStarted: meta.providerStarted, + }); + const metadata: Record = {}; + if (params.behavior.kind === "channel-delivery" || params.behavior.kind === "followup-delivery") { + for (const key of [ + "stopReason", + "yielded", + "timeoutPhase", + "providerStarted", + "aborted", + "livenessState", + "replayInvalid", + ] as const) { + if (!Object.hasOwn(meta, key)) { + continue; + } + metadata[key] = key in outcome ? outcome[key as keyof typeof outcome] : meta[key]; + } + } else { + for (const key of ["stopReason", "livenessState", "timeoutPhase", "providerStarted"] as const) { + if (outcome[key] !== undefined) { + metadata[key] = outcome[key]; + } + } + if (typeof meta.aborted === "boolean") { + metadata.aborted = meta.aborted; + } + if (meta.replayInvalid === true) { + metadata.replayInvalid = true; + } + if (meta.yielded === true) { + metadata.yielded = true; + } + } + return { outcome, metadata }; +} + +/** Runs a fallback candidate chain and prepares its shared terminal settlement state. */ +export async function runEmbeddedAgentEntry( + params: EmbeddedAgentRunEntryParams, +): Promise> { + let candidateIndex = 0; + const committedSideEffect = + params.behavior.kind === "command-rpc" ? params.behavior.hasCommittedSideEffect : undefined; + const fallbackResult = await runWithModelFallback({ + ...params.selection, + ...params.identity, + abortSignal: params.abortSignal, + resolveAgentHarnessRuntimeOverride: params.harness.resolveRuntimeOverride, + prepareAgentHarnessRuntime: async ({ provider, model, agentHarnessRuntimeOverride }) => { + const prepare = () => + ensureSelectedAgentHarnessPlugin({ + config: params.selection.cfg, + provider, + modelId: model, + agentId: params.identity.agentId, + sessionKey: params.harness.sessionKey, + agentHarnessId: agentHarnessRuntimeOverride, + agentHarnessRuntimeOverride, + workspaceDir: params.harness.workspaceDir, + }); + if (params.harness.preparation.kind === "measured") { + await params.harness.preparation.run(prepare); + } else { + await prepare(); + } + }, + onFallbackStep: params.onFallbackStep, + ...(params.behavior.kind === "maintenance" + ? {} + : { + classifyResult: ({ + result, + provider, + model, + }: { + result: T; + provider: string; + model: string; + }) => { + const deliveryEvidence = + params.behavior.kind === "channel-delivery" + ? params.behavior.readDeliveryEvidence() + : undefined; + const classification = classifyEmbeddedAgentRunResultForModelFallback({ + result, + provider, + model, + ...deliveryEvidence, + }); + const effectiveClassification = + params.behavior.kind === "followup-delivery" + ? preserveFollowupResultForDelivery(classification) + : classification; + return effectiveClassification && committedSideEffect?.() + ? undefined + : effectiveClassification; + }, + }), + ...(committedSideEffect ? { canFallbackAfterError: () => !committedSideEffect() } : {}), + ...(params.behavior.kind === "maintenance" + ? {} + : { + mergeExhaustedResult: ({ + latestResult, + preferredResult, + }: { + latestResult: T; + preferredResult: T; + }) => + mergeEmbeddedAgentRunResultForModelFallbackExhaustion({ + latestResult, + preferredResult, + }) as T, + }), + run: async (provider, model, options) => { + const isFallbackRetry = candidateIndex > 0; + candidateIndex += 1; + return params.runCandidate(provider, model, { + allowTransientCooldownProbe: options?.allowTransientCooldownProbe, + isFinalFallbackAttempt: options?.isFinalFallbackAttempt, + isFallbackRetry, + }); + }, + }); + const abortFields = + params.behavior.kind === "command-rpc" + ? resolveAgentRunAbortLifecycleFields(params.abortSignal) + : {}; + const result = + abortFields.aborted === true + ? ({ + ...fallbackResult.result, + meta: { + ...fallbackResult.result.meta, + ...abortFields, + }, + } as T) + : fallbackResult.result; + const settledResult = { + ...fallbackResult, + outcome: + fallbackResult.outcome === "exhausted" ? ("exhausted" as const) : ("completed" as const), + result, + }; + const terminal = buildTerminal({ + result, + fallbackExhausted: settledResult.outcome === "exhausted", + behavior: params.behavior, + }); + let sessionOverrideSettled = false; + const settleSessionOverride = async () => { + if (sessionOverrideSettled) { + return; + } + sessionOverrideSettled = true; + if ( + settledResult.outcome === "completed" && + params.sessionOverride.kind === "reconcile-completed" + ) { + await params.sessionOverride.reconcile({ + provider: settledResult.provider, + model: settledResult.model, + }); + } + }; + return { ...settledResult, terminal, settleSessionOverride }; +} diff --git a/src/agents/runtime-plan/build.ts b/src/agents/runtime-plan/build.ts index db7c382c8eca..deb8df0346cf 100644 --- a/src/agents/runtime-plan/build.ts +++ b/src/agents/runtime-plan/build.ts @@ -133,7 +133,7 @@ export function buildAgentRuntimeDeliveryPlan( } /** Build run-outcome classification hooks for model fallback decisions. */ -export function buildAgentRuntimeOutcomePlan(): AgentRuntimeOutcomePlan { +function buildAgentRuntimeOutcomePlan(): AgentRuntimeOutcomePlan { return { classifyRunResult: classifyEmbeddedAgentRunResultForModelFallback, }; diff --git a/src/auto-reply/reply/agent-runner-execution.test-support.ts b/src/auto-reply/reply/agent-runner-execution.test-support.ts index 1edf464d66b3..70f345427f50 100644 --- a/src/auto-reply/reply/agent-runner-execution.test-support.ts +++ b/src/auto-reply/reply/agent-runner-execution.test-support.ts @@ -2,6 +2,8 @@ import { afterEach, beforeEach, expect, vi } from "vitest"; import { testing as cliBackendsTesting } from "../../agents/cli-backends.test-support.js"; import { AUTH_INVALID_TOKEN_USER_TEXT } from "../../agents/embedded-agent-helpers/errors.js"; +import type { runEmbeddedAgentEntry } from "../../agents/embedded-agent-runner/run-entry.js"; +import type { EmbeddedAgentRunResult } from "../../agents/embedded-agent-runner/types.js"; import type { ModelDefinitionConfig } from "../../config/types.models.js"; import type { ReplyOptionsWithHeartbeatRunScope } from "../../infra/heartbeat-run-scope.js"; import { @@ -15,6 +17,10 @@ import type { FollowupRun } from "./queue.js"; import type { ReplyOperation } from "./reply-run-registry.js"; import type { TypingSignaler } from "./typing-mode.js"; +type RunEntryParams = Parameters>[0]; +type RunEntryResult = Awaited>>; +type RunEntryDelegate = (params: RunEntryParams) => Promise; + export const PROVIDER_AUTHENTICATION_ERROR_USER_MESSAGE = `⚠️ ${AUTH_INVALID_TOKEN_USER_TEXT}`; export const PROVIDER_RATE_LIMIT_OR_QUOTA_ERROR_USER_MESSAGE = "⚠️ The model provider returned HTTP 429 before replying. This can mean rate limiting, exhausted quota, or an account balance/billing issue. Check the selected provider/model, API key, and provider billing/quota dashboard, then try again."; @@ -23,6 +29,7 @@ export const PROVIDER_INTERNAL_ERROR_USER_MESSAGE = const state = vi.hoisted(() => ({ runEmbeddedAgentMock: vi.fn(), + runEmbeddedAgentEntryMock: vi.fn(), runCliAgentMock: vi.fn(), runWithModelFallbackMock: vi.fn(), isCliProviderMock: vi.fn((_: unknown) => false), @@ -55,6 +62,17 @@ vi.mock("../../agents/embedded-agent.js", () => ({ runEmbeddedAgent: (params: unknown) => state.runEmbeddedAgentMock(params), })); +vi.mock("../../agents/embedded-agent-runner/run-entry.js", async () => { + const actual = await vi.importActual< + typeof import("../../agents/embedded-agent-runner/run-entry.js") + >("../../agents/embedded-agent-runner/run-entry.js"); + return { + ...actual, + runEmbeddedAgentEntry: (params: RunEntryParams) => + state.runEmbeddedAgentEntryMock(params, actual.runEmbeddedAgentEntry as RunEntryDelegate), + }; +}); + vi.mock("../../agents/agent-bundle-mcp-manager-api.js", () => ({ peekSessionMcpRuntime: (params: unknown) => state.peekSessionMcpRuntimeMock(params), })); @@ -553,6 +571,9 @@ export function setupAgentRunnerExecutionTestState() { beforeEach(() => { vi.useRealTimers(); state.runEmbeddedAgentMock.mockReset(); + state.runEmbeddedAgentEntryMock + .mockReset() + .mockImplementation((params: RunEntryParams, delegate: RunEntryDelegate) => delegate(params)); state.runCliAgentMock.mockReset(); state.runWithModelFallbackMock.mockReset(); state.isCliProviderMock.mockReset(); diff --git a/src/auto-reply/reply/agent-runner-fallback-candidate.ts b/src/auto-reply/reply/agent-runner-fallback-candidate.ts index 80dd4858316e..9f1bf9fa47d9 100644 --- a/src/auto-reply/reply/agent-runner-fallback-candidate.ts +++ b/src/auto-reply/reply/agent-runner-fallback-candidate.ts @@ -1,12 +1,9 @@ import { normalizeLowercaseStringOrEmpty } from "@openclaw/normalization-core/string-coerce"; import { markAutoFallbackPrimaryProbe } from "../../agents/agent-scope.js"; -import { mergeEmbeddedAgentRunResultForModelFallbackExhaustion } from "../../agents/embedded-agent-runner/result-fallback-classifier.js"; +import { runEmbeddedAgentEntry } from "../../agents/embedded-agent-runner/run-entry.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 { buildAgentRuntimeOutcomePlan } from "../../agents/runtime-plan/build.js"; import { resolveSessionRuntimeOverrideForProvider } from "../../agents/session-runtime-compat.js"; import { resolveCandidateThinkingLevel } from "../../agents/thinking-runtime.js"; import { resolveHeartbeatRunScope } from "../../infra/heartbeat-run-scope.js"; @@ -30,7 +27,6 @@ export async function runAgentFallbackCandidates(params: AgentFallbackCycleParam 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; @@ -58,47 +54,55 @@ export async function runAgentFallbackCandidates(params: AgentFallbackCycleParam sessionKey: turn.sessionKey, milestone: "before_model_fallback", }); + const selection = resolveModelFallbackOptions(params.effectiveRun, params.runtimeConfig); return params.timing.measure("model_fallback", () => - runWithModelFallback({ - ...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, + runEmbeddedAgentEntry({ + selection: { + cfg: selection.cfg, + provider: selection.provider, + model: selection.model, + agentDir: selection.agentDir, + fallbacksOverride: selection.fallbacksOverride, + }, + identity: { + runId: params.runId, + agentId: turn.followupRun.run.agentId, + sessionId: turn.followupRun.run.sessionId, + sessionKey: selection.sessionKey, + lane: runLane, + }, + harness: { + workspaceDir: turn.followupRun.run.workspaceDir, + sessionKey: turn.followupRun.run.runtimePolicySessionKey ?? turn.sessionKey, + preparation: { + kind: "measured", + run: (prepare) => params.timing.measure("fallback_prepare_harness", prepare), + }, + resolveRuntimeOverride: (provider) => + resolveSessionRuntimeOverrideForProvider({ provider, - modelId: model, - agentId: turn.followupRun.run.agentId, - sessionKey: turn.followupRun.run.runtimePolicySessionKey ?? turn.sessionKey, - agentHarnessRuntimeOverride, - workspaceDir: turn.followupRun.run.workspaceDir, + entry: params.liveModelSwitchRuntimeEntry ?? turn.getActiveSessionEntry(), + cfg: params.runtimeConfig, }), - ); }, - onFallbackStep: (step) => { - emitModelFallbackStepLifecycle({ runId: params.runId, sessionKey: turn.sessionKey, step }); - }, - classifyResult: ({ result, provider, model }) => - outcomePlan.classifyRunResult({ - result, - provider, - model, + behavior: { + kind: "channel-delivery", + readDeliveryEvidence: () => ({ hasDirectlySentBlockReply: params.directlySentBlockKeys.size > 0, hasBlockReplyPipelineOutput: Boolean( turn.blockReplyPipeline?.hasBuffered() || turn.blockReplyPipeline?.didStream(), ), }), - mergeExhaustedResult: mergeEmbeddedAgentRunResultForModelFallbackExhaustion, - run: async (provider, model, runOptions) => { + }, + sessionOverride: { + kind: "reconcile-completed", + reconcile: params.clearRecoveredAutoFallbackPrimaryProbe, + }, + abortSignal: params.runAbortSignal, + onFallbackStep: (step) => { + emitModelFallbackStepLifecycle({ runId: params.runId, sessionKey: turn.sessionKey, step }); + }, + runCandidate: async (provider, model, runOptions) => { params.state.attemptedRuntimeProvider = provider; params.state.attemptedRuntimeModel = model; const candidateRun = resolveFallbackCandidateRun(params.effectiveRun, provider, model); diff --git a/src/auto-reply/reply/agent-runner-fallback-settlement.ts b/src/auto-reply/reply/agent-runner-fallback-settlement.ts index 5db6e5a99ff0..4db11c605079 100644 --- a/src/auto-reply/reply/agent-runner-fallback-settlement.ts +++ b/src/auto-reply/reply/agent-runner-fallback-settlement.ts @@ -8,7 +8,6 @@ import { emitAgentEvent } from "../../infra/agent-events.js"; import { formatErrorMessage } from "../../infra/errors.js"; import { defaultRuntime } from "../../runtime.js"; import { SILENT_REPLY_TOKEN } from "../tokens.js"; -import { resolveAgentLifecycleTerminalMetadata } from "./agent-lifecycle-terminal.js"; import { buildContextOverflowRecoveryText } from "./agent-runner-context-recovery.js"; import { markAgentRunFailureReplyPayload } from "./agent-runner-failure-reply.js"; import type { AgentFallbackCandidatesResult } from "./agent-runner-fallback-candidate.js"; @@ -66,10 +65,7 @@ export async function settleAgentFallbackCycle(params: { })) : []; if (!fallbackExhausted) { - await cycle.clearRecoveredAutoFallbackPrimaryProbe({ - provider: fallbackProvider, - model: fallbackModel, - }); + await fallbackResult.settleSessionOverride(); } const embeddedError = runResult.meta?.error; const deferredLifecycleError = settledLifecycleTerminal?.getDeferredError(); @@ -129,7 +125,7 @@ export async function settleAgentFallbackCycle(params: { }), }; } - const terminalMetadata = resolveAgentLifecycleTerminalMetadata(runResult.meta); + const terminalMetadata = fallbackResult.terminal.metadata; let terminalRunFailed = false; if (fallbackExhausted) { const exhaustionError = new Error( diff --git a/src/auto-reply/reply/agent-runner-memory.test.ts b/src/auto-reply/reply/agent-runner-memory.test.ts index eba2165331c1..39df150c997a 100644 --- a/src/auto-reply/reply/agent-runner-memory.test.ts +++ b/src/auto-reply/reply/agent-runner-memory.test.ts @@ -5,6 +5,8 @@ import os from "node:os"; import path from "node:path"; import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; import { testing as cliBackendsTesting } from "../../agents/cli-backends.test-support.js"; +import type { runEmbeddedAgentEntry } from "../../agents/embedded-agent-runner/run-entry.js"; +import type { EmbeddedAgentRunResult } from "../../agents/embedded-agent-runner/types.js"; import type { SessionEntry } from "../../config/sessions.js"; import { loadSessionEntry, @@ -27,6 +29,7 @@ import type { ReplyOperation } from "./reply-run-registry.js"; const compactEmbeddedAgentSessionMock = vi.fn(); const runWithModelFallbackMock = vi.fn(); +const runEmbeddedAgentEntryMock = vi.fn(); const runEmbeddedAgentMock = vi.fn(); const refreshQueuedFollowupSessionMock = vi.fn(); const incrementCompactionCountMock = vi.fn(); @@ -110,6 +113,14 @@ type ModelFallbackParams = { model: string; agentHarnessRuntimeOverride?: string; }) => Promise | void; + run: ( + provider: string, + model: string, + options?: { + allowTransientCooldownProbe?: boolean; + isFinalFallbackAttempt?: boolean; + }, + ) => Promise; }; type EmbeddedAgentParams = { @@ -206,6 +217,69 @@ describe("runMemoryFlushIfNeeded", () => { model, attempts: [], })); + runEmbeddedAgentEntryMock + .mockReset() + .mockImplementation( + async (params: Parameters>[0]) => { + const fallbackResult = (await runWithModelFallbackMock({ + ...params.selection, + ...params.identity, + abortSignal: params.abortSignal, + resolveAgentHarnessRuntimeOverride: params.harness.resolveRuntimeOverride, + prepareAgentHarnessRuntime: async ({ + provider, + model, + agentHarnessRuntimeOverride, + }: { + provider: string; + model: string; + agentHarnessRuntimeOverride?: string; + }) => { + await ensureSelectedAgentHarnessPluginMock({ + config: params.selection.cfg, + provider, + modelId: model, + agentId: params.identity.agentId, + sessionKey: params.harness.sessionKey, + agentHarnessId: agentHarnessRuntimeOverride, + agentHarnessRuntimeOverride, + workspaceDir: params.harness.workspaceDir, + }); + }, + run: ( + provider: string, + model: string, + options?: ModelFallbackParams["run"] extends ( + provider: string, + model: string, + options?: infer TOptions, + ) => Promise + ? TOptions + : never, + ) => + params.runCandidate(provider, model, { + allowTransientCooldownProbe: options?.allowTransientCooldownProbe, + isFinalFallbackAttempt: options?.isFinalFallbackAttempt, + isFallbackRetry: false, + }), + })) as { + outcome?: "completed" | "exhausted"; + result: EmbeddedAgentRunResult; + provider: string; + model: string; + attempts: []; + }; + return { + ...fallbackResult, + outcome: fallbackResult.outcome ?? ("completed" as const), + terminal: { + outcome: { reason: "completed" as const, status: "ok" as const }, + metadata: {}, + }, + settleSessionOverride: async () => undefined, + }; + }, + ); compactEmbeddedAgentSessionMock.mockReset().mockResolvedValue({ ok: true, compacted: true, @@ -246,12 +320,11 @@ describe("runMemoryFlushIfNeeded", () => { }); setAgentRunnerMemoryTestDeps({ compactEmbeddedAgentSession: compactEmbeddedAgentSessionMock as never, - runWithModelFallback: runWithModelFallbackMock as never, + runEmbeddedAgentEntry: runEmbeddedAgentEntryMock as never, runEmbeddedAgent: runEmbeddedAgentMock as never, ensureMemoryFlushTargetFile: ensureMemoryFlushTargetFileMock as never, refreshQueuedFollowupSession: refreshQueuedFollowupSessionMock as never, incrementCompactionCount: incrementCompactionCountMock as never, - ensureSelectedAgentHarnessPlugin: ensureSelectedAgentHarnessPluginMock as never, registerAgentRunContext: vi.fn() as never, emitAgentEvent: emitAgentEventMock as never, randomUUID: () => "00000000-0000-0000-0000-000000000001", diff --git a/src/auto-reply/reply/agent-runner-memory.ts b/src/auto-reply/reply/agent-runner-memory.ts index a8334e2b82a8..00ace254eca9 100644 --- a/src/auto-reply/reply/agent-runner-memory.ts +++ b/src/auto-reply/reply/agent-runner-memory.ts @@ -10,8 +10,7 @@ import { truncateUtf16Safe } from "@openclaw/normalization-core/utf16-slice"; import { resolveBootstrapWarningSignaturesSeen } from "../../agents/bootstrap-budget.js"; import { estimateMessagesTokens } from "../../agents/compaction.js"; import { classifyCompactionReason } from "../../agents/embedded-agent-runner/compact-reasons.js"; -import { ensureSelectedAgentHarnessPlugin } from "../../agents/harness/runtime-plugin.js"; -import { runWithModelFallback } from "../../agents/model-fallback.js"; +import { runEmbeddedAgentEntry } from "../../agents/embedded-agent-runner/run-entry.js"; import { isCliRuntimeAliasForProvider } from "../../agents/model-runtime-aliases.js"; import { isCliProvider } from "../../agents/model-selection.js"; import { resolveContextConfigProviderForRuntime } from "../../agents/openai-routing.js"; @@ -158,8 +157,7 @@ async function ensureMemoryFlushTargetFile(params: { const memoryDeps = { compactEmbeddedAgentSession: compactEmbeddedAgentSessionDefault, - runWithModelFallback, - ensureSelectedAgentHarnessPlugin, + runEmbeddedAgentEntry, runEmbeddedAgent: runEmbeddedAgentDefault, ensureMemoryFlushTargetFile, registerAgentRunContext, @@ -174,8 +172,7 @@ const memoryDeps = { /** Overrides memory helper dependencies for tests. */ function setAgentRunnerMemoryTestDeps(overrides?: Partial): void { Object.assign(memoryDeps, { - runWithModelFallback, - ensureSelectedAgentHarnessPlugin, + runEmbeddedAgentEntry, compactEmbeddedAgentSession: compactEmbeddedAgentSessionDefault, runEmbeddedAgent: runEmbeddedAgentDefault, ensureMemoryFlushTargetFile, @@ -1341,38 +1338,44 @@ export async function runMemoryFlushIfNeeded(params: { let postCompactionSessionId: string | undefined; let postCompactionSessionFile: string | undefined; try { - await memoryDeps.runWithModelFallback({ - ...resolveMemoryFlushModelFallbackOptions( - params.followupRun.run, - activeMemoryFlushPlan.model, - params.cfg, - ), - runId: flushRunId, - sessionId: activeSessionEntry?.sessionId ?? params.followupRun.run.sessionId, - lane: CommandLane.Main, - abortSignal: params.replyOperation.abortSignal, - resolveAgentHarnessRuntimeOverride: (provider) => - resolveSessionRuntimeOverrideForProvider({ - provider, - entry: activeSessionEntry, - cfg: params.cfg, - }), - prepareAgentHarnessRuntime: async ({ provider, model, agentHarnessRuntimeOverride }) => { - await memoryDeps.ensureSelectedAgentHarnessPlugin({ - config: params.cfg, - provider, - modelId: model, - agentId: params.followupRun.run.agentId, - sessionKey: - params.runtimePolicySessionKey ?? - params.followupRun.run.runtimePolicySessionKey ?? - params.sessionKey, - agentHarnessId: agentHarnessRuntimeOverride, - agentHarnessRuntimeOverride, - workspaceDir: params.followupRun.run.workspaceDir, - }); + const selection = resolveMemoryFlushModelFallbackOptions( + params.followupRun.run, + activeMemoryFlushPlan.model, + params.cfg, + ); + await memoryDeps.runEmbeddedAgentEntry({ + selection: { + cfg: selection.cfg, + provider: selection.provider, + model: selection.model, + agentDir: selection.agentDir, + fallbacksOverride: selection.fallbacksOverride, }, - run: async (provider, model, runOptions) => { + identity: { + runId: flushRunId, + agentId: params.followupRun.run.agentId, + sessionId: activeSessionEntry?.sessionId ?? params.followupRun.run.sessionId, + sessionKey: selection.sessionKey, + lane: CommandLane.Main, + }, + harness: { + workspaceDir: params.followupRun.run.workspaceDir, + sessionKey: + params.runtimePolicySessionKey ?? + params.followupRun.run.runtimePolicySessionKey ?? + params.sessionKey, + preparation: { kind: "direct" }, + resolveRuntimeOverride: (provider) => + resolveSessionRuntimeOverrideForProvider({ + provider, + entry: activeSessionEntry, + cfg: params.cfg, + }), + }, + behavior: { kind: "maintenance" }, + sessionOverride: { kind: "preserve" }, + abortSignal: params.replyOperation.abortSignal, + runCandidate: async (provider, model, runOptions) => { const sessionRuntimeOverride = resolveSessionRuntimeOverrideForProvider({ provider, entry: activeSessionEntry, @@ -1399,7 +1402,7 @@ export async function runMemoryFlushIfNeeded(params: { provider, model, runId: flushRunId, - allowTransientCooldownProbe: runOptions?.allowTransientCooldownProbe, + allowTransientCooldownProbe: runOptions.allowTransientCooldownProbe, }); const result = await memoryDeps.runEmbeddedAgent({ ...embeddedContext, @@ -1415,7 +1418,7 @@ export async function runMemoryFlushIfNeeded(params: { prompt: activeMemoryFlushPlan.prompt, transcriptPrompt: "", extraSystemPrompt: flushSystemPrompt, - isFinalFallbackAttempt: runOptions?.isFinalFallbackAttempt, + isFinalFallbackAttempt: runOptions.isFinalFallbackAttempt, bootstrapPromptWarningSignaturesSeen, bootstrapPromptWarningSignature: bootstrapPromptWarningSignaturesSeen[bootstrapPromptWarningSignaturesSeen.length - 1], diff --git a/src/auto-reply/reply/followup-runner.ts b/src/auto-reply/reply/followup-runner.ts index e8570219026c..be7233a6cca9 100644 --- a/src/auto-reply/reply/followup-runner.ts +++ b/src/auto-reply/reply/followup-runner.ts @@ -18,28 +18,18 @@ import { hasCommittedSourceReplyDeliveryEvidence, hasVisibleOutboundDeliveryEvidence, } from "../../agents/embedded-agent-runner/delivery-evidence.js"; -import { - hasDeliberateSilentTerminalReply, - mergeEmbeddedAgentRunResultForModelFallbackExhaustion, -} from "../../agents/embedded-agent-runner/result-fallback-classifier.js"; +import { hasDeliberateSilentTerminalReply } from "../../agents/embedded-agent-runner/result-fallback-classifier.js"; +import { runEmbeddedAgentEntry } from "../../agents/embedded-agent-runner/run-entry.js"; import { runEmbeddedAgent } from "../../agents/embedded-agent.js"; import type { FastModeAutoProgressState } from "../../agents/fast-mode.js"; -import { ensureSelectedAgentHarnessPlugin } from "../../agents/harness/runtime-plugin.js"; -import { - isFallbackSummaryError, - runWithModelFallback, - type ModelFallbackResultClassification, -} from "../../agents/model-fallback.js"; +import { isFallbackSummaryError } from "../../agents/model-fallback.js"; import { resolveCliRuntimeExecutionProvider } from "../../agents/model-runtime-aliases.js"; import { isCliProvider } from "../../agents/model-selection-cli.js"; import { isAgentRunRestartAbortReason, resolveAgentRunErrorLifecycleFields, } from "../../agents/run-termination.js"; -import { - buildAgentRuntimeDeliveryPlan, - buildAgentRuntimeOutcomePlan, -} from "../../agents/runtime-plan/build.js"; +import { buildAgentRuntimeDeliveryPlan } from "../../agents/runtime-plan/build.js"; import { withLocalSessionPlacementTurnAdmission } from "../../agents/session-placement-admission.js"; import { resolveSessionRuntimeOverrideForProvider } from "../../agents/session-runtime-compat.js"; import { resolveCandidateThinkingLevel } from "../../agents/thinking-runtime.js"; @@ -68,7 +58,6 @@ import { import type { GetReplyOptions, ReplyPayload } from "../types.js"; import { createAgentLifecycleTerminalBackstop, - resolveAgentLifecycleTerminalMetadata, type AgentLifecycleTerminalBackstop, } from "./agent-lifecycle-terminal.js"; import { resolveRunAfterAutoFallbackPrimaryProbeRecheck } from "./agent-runner-auto-fallback.js"; @@ -138,33 +127,6 @@ import type { TypingController } from "./typing.js"; type EmbeddedAgentRunResult = Awaited>; -const PRESERVED_FOLLOWUP_RESULT_CODES = new Set([ - "empty_result", - "reasoning_only_result", - "planning_only_result", -]); - -function preserveNonVisibleFollowupResult( - classification: ModelFallbackResultClassification, -): ModelFallbackResultClassification { - if ( - !classification || - !("code" in classification) || - !classification.code || - !PRESERVED_FOLLOWUP_RESULT_CODES.has(classification.code) - ) { - return classification; - } - // Follow-up delivery owns its terminal fallback. Preserve the classified result - // so that owner can route a visible failure instead of losing it to a thrown summary. - return { - ...classification, - preserveResultOnExhaustion: true, - // Prefer any earlier result that carries a user-facing terminal presentation. - preserveResultPriority: -1, - }; -} - type FollowupAgentEvent = { stream: string; data: Record }; function isStrandedReplyRetryFollowup(queued: FollowupRun): boolean { @@ -1014,37 +976,39 @@ export function createFollowupRunner(params: { resetAnnounced: false, }; try { - const outcomePlan = buildAgentRuntimeOutcomePlan(); - const fallbackResult = await runWithModelFallback({ - ...resolveModelFallbackOptions(run, runtimeConfig), - cfg: runtimeConfig, - runId, - sessionId: run.sessionId, - abortSignal: runAbortSignal, - resolveAgentHarnessRuntimeOverride: (provider) => - resolveSessionRuntimeOverrideForProvider({ - provider, - entry: activeSessionEntry, - cfg: runtimeConfig, - }), - prepareAgentHarnessRuntime: async ({ provider, model, agentHarnessRuntimeOverride }) => { - await ensureSelectedAgentHarnessPlugin({ - config: runtimeConfig, - provider, - modelId: model, - agentId: run.agentId, - sessionKey: run.runtimePolicySessionKey ?? replySessionKey, - agentHarnessId: agentHarnessRuntimeOverride, - agentHarnessRuntimeOverride, - workspaceDir: run.workspaceDir, - }); + const selection = resolveModelFallbackOptions(run, runtimeConfig); + const fallbackResult = await runEmbeddedAgentEntry({ + selection: { + cfg: selection.cfg, + provider: selection.provider, + model: selection.model, + agentDir: selection.agentDir, + fallbacksOverride: selection.fallbacksOverride, }, - classifyResult: ({ result, provider, model }) => - preserveNonVisibleFollowupResult( - outcomePlan.classifyRunResult({ result, provider, model }), - ), - mergeExhaustedResult: mergeEmbeddedAgentRunResultForModelFallbackExhaustion, - run: async (provider, model, runOptions) => { + identity: { + runId, + agentId: run.agentId, + sessionId: run.sessionId, + sessionKey: selection.sessionKey, + }, + harness: { + workspaceDir: run.workspaceDir, + sessionKey: run.runtimePolicySessionKey ?? replySessionKey, + preparation: { kind: "direct" }, + resolveRuntimeOverride: (provider) => + resolveSessionRuntimeOverrideForProvider({ + provider, + entry: activeSessionEntry, + cfg: runtimeConfig, + }), + }, + behavior: { kind: "followup-delivery" }, + sessionOverride: { + kind: "reconcile-completed", + reconcile: clearRecoveredAutoFallbackPrimaryProbe, + }, + abortSignal: runAbortSignal, + runCandidate: async (provider, model, runOptions) => { const suppressQueuedUserPersistenceForCandidate = (run.suppressNextUserMessagePersistence ?? false) || queuedUserMessagePersistedAcrossFallback; @@ -1582,7 +1546,7 @@ export function createFollowupRunner(params: { deferredLifecycleError ?? userFacingErrorPayload ?? (runResult.meta?.error ? "Agent run failed" : undefined); - const terminalMetadata = resolveAgentLifecycleTerminalMetadata(runResult.meta); + const terminalMetadata = fallbackResult.terminal.metadata; if (fallbackExhausted) { const exhaustionError = new Error( terminalErrorMessage ?? "All model fallback candidates failed", @@ -1602,10 +1566,7 @@ export function createFollowupRunner(params: { settledLifecycleTerminal?.emit("end", runResult); } if (!fallbackExhausted) { - await clearRecoveredAutoFallbackPrimaryProbe({ - provider: fallbackProvider, - model: fallbackModel, - }); + await fallbackResult.settleSessionOverride(); } } catch (err) { if (