From d7a6a0a0b96ffab5bc7a81f722c7d8ffb0998828 Mon Sep 17 00:00:00 2001 From: Peter Steinberger Date: Wed, 18 Feb 2026 19:02:25 +0000 Subject: [PATCH] refactor(reply): share embedded run fallback/context builders --- .../reply/agent-runner-execution.ts | 48 ++---- src/auto-reply/reply/agent-runner-memory.ts | 53 ++---- .../reply/agent-runner-utils.test.ts | 152 ++++++++++++++++++ src/auto-reply/reply/agent-runner-utils.ts | 60 +++++++ 4 files changed, 241 insertions(+), 72 deletions(-) create mode 100644 src/auto-reply/reply/agent-runner-utils.test.ts diff --git a/src/auto-reply/reply/agent-runner-execution.ts b/src/auto-reply/reply/agent-runner-execution.ts index 2620351d39a..e48155aa374 100644 --- a/src/auto-reply/reply/agent-runner-execution.ts +++ b/src/auto-reply/reply/agent-runner-execution.ts @@ -1,6 +1,5 @@ import crypto from "node:crypto"; import fs from "node:fs"; -import { resolveAgentModelFallbacksOverride } from "../../agents/agent-scope.js"; import { runCliAgent } from "../../agents/cli-runner.js"; import { getCliSessionId } from "../../agents/cli-session.js"; import { runWithModelFallback } from "../../agents/model-fallback.js"; @@ -14,7 +13,6 @@ import { } from "../../agents/pi-embedded-helpers.js"; import { runEmbeddedPiAgent } from "../../agents/pi-embedded.js"; import { - resolveAgentIdFromSessionKey, resolveGroupSessionKey, resolveSessionTranscriptPath, type SessionEntry, @@ -33,11 +31,10 @@ import type { VerboseLevel } from "../thinking.js"; import { isSilentReplyText, SILENT_REPLY_TOKEN } from "../tokens.js"; import type { GetReplyOptions, ReplyPayload } from "../types.js"; import { - buildEmbeddedContextFromTemplate, - buildTemplateSenderContext, - resolveRunAuthProfile, + buildEmbeddedRunBaseParams, + buildEmbeddedRunContexts, + resolveModelFallbackOptions, } from "./agent-runner-utils.js"; -import { resolveEnforceFinalTag } from "./agent-runner-utils.js"; import { type BlockReplyPipeline } from "./block-reply-pipeline.js"; import type { FollowupRun } from "./queue.js"; import { createBlockReplyDeliveryHandler } from "./reply-delivery.js"; @@ -157,14 +154,7 @@ export async function runAgentTurnWithFallback(params: { const blockReplyPipeline = params.blockReplyPipeline; const onToolResult = params.opts?.onToolResult; const fallbackResult = await runWithModelFallback({ - cfg: params.followupRun.run.config, - provider: params.followupRun.run.provider, - model: params.followupRun.run.model, - agentDir: params.followupRun.run.agentDir, - fallbacksOverride: resolveAgentModelFallbacksOverride( - params.followupRun.run.config, - resolveAgentIdFromSessionKey(params.followupRun.run.sessionKey), - ), + ...resolveModelFallbackOptions(params.followupRun.run), run: (provider, model) => { // Notify that model selection is complete (including after fallback). // This allows responsePrefix template interpolation with the actual model. @@ -262,13 +252,19 @@ export async function runAgentTurnWithFallback(params: { } })(); } - const authProfile = resolveRunAuthProfile(params.followupRun.run, provider); - const embeddedContext = buildEmbeddedContextFromTemplate({ + const { authProfile, embeddedContext, senderContext } = buildEmbeddedRunContexts({ run: params.followupRun.run, sessionCtx: params.sessionCtx, hasRepliedRef: params.opts?.hasRepliedRef, + provider, + }); + const runBaseParams = buildEmbeddedRunBaseParams({ + run: params.followupRun.run, + provider, + model, + runId, + authProfile, }); - const senderContext = buildTemplateSenderContext(params.sessionCtx); return runEmbeddedPiAgent({ ...embeddedContext, groupId: resolveGroupSessionKey(params.sessionCtx)?.id, @@ -276,22 +272,9 @@ export async function runAgentTurnWithFallback(params: { params.sessionCtx.GroupChannel?.trim() ?? params.sessionCtx.GroupSubject?.trim(), groupSpace: params.sessionCtx.GroupSpace?.trim() ?? undefined, ...senderContext, - sessionFile: params.followupRun.run.sessionFile, - workspaceDir: params.followupRun.run.workspaceDir, - agentDir: params.followupRun.run.agentDir, - config: params.followupRun.run.config, - skillsSnapshot: params.followupRun.run.skillsSnapshot, + ...runBaseParams, prompt: params.commandBody, extraSystemPrompt: params.followupRun.run.extraSystemPrompt, - ownerNumbers: params.followupRun.run.ownerNumbers, - enforceFinalTag: resolveEnforceFinalTag(params.followupRun.run, provider), - provider, - model, - ...authProfile, - thinkLevel: params.followupRun.run.thinkLevel, - verboseLevel: params.followupRun.run.verboseLevel, - reasoningLevel: params.followupRun.run.reasoningLevel, - execOverrides: params.followupRun.run.execOverrides, toolResultFormat: (() => { const channel = resolveMessageChannel( params.sessionCtx.Surface, @@ -303,9 +286,6 @@ export async function runAgentTurnWithFallback(params: { return isMarkdownCapableMessageChannel(channel) ? "markdown" : "plain"; })(), suppressToolErrorWarnings: params.opts?.suppressToolErrorWarnings, - bashElevated: params.followupRun.run.bashElevated, - timeoutMs: params.followupRun.run.timeoutMs, - runId, images: params.opts?.images, abortSignal: params.opts?.abortSignal, blockReplyBreak: params.resolvedBlockStreamingBreak, diff --git a/src/auto-reply/reply/agent-runner-memory.ts b/src/auto-reply/reply/agent-runner-memory.ts index 1b61104d2cd..681fb76a2c9 100644 --- a/src/auto-reply/reply/agent-runner-memory.ts +++ b/src/auto-reply/reply/agent-runner-memory.ts @@ -1,26 +1,20 @@ import crypto from "node:crypto"; -import { resolveAgentModelFallbacksOverride } from "../../agents/agent-scope.js"; import { runWithModelFallback } from "../../agents/model-fallback.js"; import { isCliProvider } from "../../agents/model-selection.js"; import { runEmbeddedPiAgent } from "../../agents/pi-embedded.js"; import { resolveSandboxConfigForAgent, resolveSandboxRuntimeStatus } from "../../agents/sandbox.js"; import type { OpenClawConfig } from "../../config/config.js"; -import { - resolveAgentIdFromSessionKey, - type SessionEntry, - updateSessionStoreEntry, -} from "../../config/sessions.js"; +import { type SessionEntry, updateSessionStoreEntry } from "../../config/sessions.js"; import { logVerbose } from "../../globals.js"; import { registerAgentRunContext } from "../../infra/agent-events.js"; import type { TemplateContext } from "../templating.js"; import type { VerboseLevel } from "../thinking.js"; import type { GetReplyOptions } from "../types.js"; import { - buildEmbeddedContextFromTemplate, - buildTemplateSenderContext, - resolveRunAuthProfile, + buildEmbeddedRunBaseParams, + buildEmbeddedRunContexts, + resolveModelFallbackOptions, } from "./agent-runner-utils.js"; -import { resolveEnforceFinalTag } from "./agent-runner-utils.js"; import { resolveMemoryFlushContextWindowTokens, resolveMemoryFlushPromptForRun, @@ -103,47 +97,30 @@ export async function runMemoryFlushIfNeeded(params: { .join("\n\n"); try { await runWithModelFallback({ - cfg: params.followupRun.run.config, - provider: params.followupRun.run.provider, - model: params.followupRun.run.model, - agentDir: params.followupRun.run.agentDir, - fallbacksOverride: resolveAgentModelFallbacksOverride( - params.followupRun.run.config, - resolveAgentIdFromSessionKey(params.followupRun.run.sessionKey), - ), + ...resolveModelFallbackOptions(params.followupRun.run), run: (provider, model) => { - const authProfile = resolveRunAuthProfile(params.followupRun.run, provider); - const embeddedContext = buildEmbeddedContextFromTemplate({ + const { authProfile, embeddedContext, senderContext } = buildEmbeddedRunContexts({ run: params.followupRun.run, sessionCtx: params.sessionCtx, hasRepliedRef: params.opts?.hasRepliedRef, + provider, + }); + const runBaseParams = buildEmbeddedRunBaseParams({ + run: params.followupRun.run, + provider, + model, + runId: flushRunId, + authProfile, }); - const senderContext = buildTemplateSenderContext(params.sessionCtx); return runEmbeddedPiAgent({ ...embeddedContext, ...senderContext, - sessionFile: params.followupRun.run.sessionFile, - workspaceDir: params.followupRun.run.workspaceDir, - agentDir: params.followupRun.run.agentDir, - config: params.followupRun.run.config, - skillsSnapshot: params.followupRun.run.skillsSnapshot, + ...runBaseParams, prompt: resolveMemoryFlushPromptForRun({ prompt: memoryFlushSettings.prompt, cfg: params.cfg, }), extraSystemPrompt: flushSystemPrompt, - ownerNumbers: params.followupRun.run.ownerNumbers, - enforceFinalTag: resolveEnforceFinalTag(params.followupRun.run, provider), - provider, - model, - ...authProfile, - thinkLevel: params.followupRun.run.thinkLevel, - verboseLevel: params.followupRun.run.verboseLevel, - reasoningLevel: params.followupRun.run.reasoningLevel, - execOverrides: params.followupRun.run.execOverrides, - bashElevated: params.followupRun.run.bashElevated, - timeoutMs: params.followupRun.run.timeoutMs, - runId: flushRunId, onAgentEvent: (evt) => { if (evt.stream === "compaction") { const phase = typeof evt.data.phase === "string" ? evt.data.phase : ""; diff --git a/src/auto-reply/reply/agent-runner-utils.test.ts b/src/auto-reply/reply/agent-runner-utils.test.ts new file mode 100644 index 00000000000..1ccf86a213d --- /dev/null +++ b/src/auto-reply/reply/agent-runner-utils.test.ts @@ -0,0 +1,152 @@ +import { beforeEach, describe, expect, it, vi } from "vitest"; +import type { FollowupRun } from "./queue.js"; + +const hoisted = vi.hoisted(() => { + const resolveAgentModelFallbacksOverrideMock = vi.fn(); + const resolveAgentIdFromSessionKeyMock = vi.fn(); + return { resolveAgentModelFallbacksOverrideMock, resolveAgentIdFromSessionKeyMock }; +}); + +vi.mock("../../agents/agent-scope.js", () => ({ + resolveAgentModelFallbacksOverride: (...args: unknown[]) => + hoisted.resolveAgentModelFallbacksOverrideMock(...args), +})); + +vi.mock("../../config/sessions.js", () => ({ + resolveAgentIdFromSessionKey: (...args: unknown[]) => + hoisted.resolveAgentIdFromSessionKeyMock(...args), +})); + +const { + buildEmbeddedRunBaseParams, + buildEmbeddedRunContexts, + resolveModelFallbackOptions, + resolveProviderScopedAuthProfile, +} = await import("./agent-runner-utils.js"); + +function makeRun(overrides: Partial = {}): FollowupRun["run"] { + return { + sessionId: "session-1", + agentId: "agent-1", + config: { models: { providers: {} } }, + provider: "openai", + model: "gpt-4.1", + agentDir: "/tmp/agent", + sessionKey: "agent:test:session", + sessionFile: "/tmp/session.json", + workspaceDir: "/tmp/workspace", + skillsSnapshot: [], + ownerNumbers: ["+15550001"], + enforceFinalTag: false, + thinkLevel: "medium", + verboseLevel: "off", + reasoningLevel: "none", + execOverrides: {}, + bashElevated: false, + timeoutMs: 60_000, + ...overrides, + } as unknown as FollowupRun["run"]; +} + +describe("agent-runner-utils", () => { + beforeEach(() => { + hoisted.resolveAgentModelFallbacksOverrideMock.mockReset(); + hoisted.resolveAgentIdFromSessionKeyMock.mockReset(); + }); + + it("resolves model fallback options from run context", () => { + hoisted.resolveAgentIdFromSessionKeyMock.mockReturnValue("agent-id"); + hoisted.resolveAgentModelFallbacksOverrideMock.mockReturnValue(["fallback-model"]); + const run = makeRun(); + + const resolved = resolveModelFallbackOptions(run); + + expect(hoisted.resolveAgentIdFromSessionKeyMock).toHaveBeenCalledWith(run.sessionKey); + expect(hoisted.resolveAgentModelFallbacksOverrideMock).toHaveBeenCalledWith( + run.config, + "agent-id", + ); + expect(resolved).toEqual({ + cfg: run.config, + provider: run.provider, + model: run.model, + agentDir: run.agentDir, + fallbacksOverride: ["fallback-model"], + }); + }); + + it("builds embedded run base params with auth profile and run metadata", () => { + const run = makeRun({ enforceFinalTag: true }); + const authProfile = resolveProviderScopedAuthProfile({ + provider: "openai", + primaryProvider: "openai", + authProfileId: "profile-openai", + authProfileIdSource: "user", + }); + + const resolved = buildEmbeddedRunBaseParams({ + run, + provider: "openai", + model: "gpt-4.1-mini", + runId: "run-1", + authProfile, + }); + + expect(resolved).toMatchObject({ + sessionFile: run.sessionFile, + workspaceDir: run.workspaceDir, + agentDir: run.agentDir, + config: run.config, + skillsSnapshot: run.skillsSnapshot, + ownerNumbers: run.ownerNumbers, + enforceFinalTag: true, + provider: "openai", + model: "gpt-4.1-mini", + authProfileId: "profile-openai", + authProfileIdSource: "user", + thinkLevel: run.thinkLevel, + verboseLevel: run.verboseLevel, + reasoningLevel: run.reasoningLevel, + execOverrides: run.execOverrides, + bashElevated: run.bashElevated, + timeoutMs: run.timeoutMs, + runId: "run-1", + }); + }); + + it("builds embedded contexts and scopes auth profile by provider", () => { + const run = makeRun({ + authProfileId: "profile-openai", + authProfileIdSource: "auto", + }); + + const resolved = buildEmbeddedRunContexts({ + run, + sessionCtx: { + Provider: "OpenAI", + To: "channel-1", + SenderId: "sender-1", + }, + hasRepliedRef: undefined, + provider: "anthropic", + }); + + expect(resolved.authProfile).toEqual({ + authProfileId: undefined, + authProfileIdSource: undefined, + }); + expect(resolved.embeddedContext).toMatchObject({ + sessionId: run.sessionId, + sessionKey: run.sessionKey, + agentId: run.agentId, + messageProvider: "openai", + messageTo: "channel-1", + }); + expect(resolved.senderContext).toEqual({ + senderId: "sender-1", + senderName: undefined, + senderUsername: undefined, + senderE164: undefined, + }); + }); +}); diff --git a/src/auto-reply/reply/agent-runner-utils.ts b/src/auto-reply/reply/agent-runner-utils.ts index 7e9a6223587..cafad03ac8e 100644 --- a/src/auto-reply/reply/agent-runner-utils.ts +++ b/src/auto-reply/reply/agent-runner-utils.ts @@ -1,8 +1,10 @@ +import { resolveAgentModelFallbacksOverride } from "../../agents/agent-scope.js"; import type { NormalizedUsage } from "../../agents/usage.js"; import { getChannelDock } from "../../channels/dock.js"; import type { ChannelId, ChannelThreadingToolContext } from "../../channels/plugins/types.js"; import { normalizeAnyChannelId, normalizeChannelId } from "../../channels/registry.js"; import type { OpenClawConfig } from "../../config/config.js"; +import { resolveAgentIdFromSessionKey } from "../../config/sessions.js"; import { isReasoningTagProvider } from "../../utils/provider-utils.js"; import { estimateUsageCost, formatTokenCount, formatUsd } from "../../utils/usage-format.js"; import type { TemplateContext } from "../templating.js"; @@ -135,6 +137,47 @@ export const appendUsageLine = (payloads: ReplyPayload[], line: string): ReplyPa export const resolveEnforceFinalTag = (run: FollowupRun["run"], provider: string) => Boolean(run.enforceFinalTag || isReasoningTagProvider(provider)); +export function resolveModelFallbackOptions(run: FollowupRun["run"]) { + return { + cfg: run.config, + provider: run.provider, + model: run.model, + agentDir: run.agentDir, + fallbacksOverride: resolveAgentModelFallbacksOverride( + run.config, + resolveAgentIdFromSessionKey(run.sessionKey), + ), + }; +} + +export function buildEmbeddedRunBaseParams(params: { + run: FollowupRun["run"]; + provider: string; + model: string; + runId: string; + authProfile: ReturnType; +}) { + return { + sessionFile: params.run.sessionFile, + workspaceDir: params.run.workspaceDir, + agentDir: params.run.agentDir, + config: params.run.config, + skillsSnapshot: params.run.skillsSnapshot, + ownerNumbers: params.run.ownerNumbers, + enforceFinalTag: resolveEnforceFinalTag(params.run, params.provider), + provider: params.provider, + model: params.model, + ...params.authProfile, + thinkLevel: params.run.thinkLevel, + verboseLevel: params.run.verboseLevel, + reasoningLevel: params.run.reasoningLevel, + execOverrides: params.run.execOverrides, + bashElevated: params.run.bashElevated, + timeoutMs: params.run.timeoutMs, + runId: params.runId, + }; +} + export function buildEmbeddedContextFromTemplate(params: { run: FollowupRun["run"]; sessionCtx: TemplateContext; @@ -175,6 +218,23 @@ export function resolveRunAuthProfile(run: FollowupRun["run"], provider: string) }); } +export function buildEmbeddedRunContexts(params: { + run: FollowupRun["run"]; + sessionCtx: TemplateContext; + hasRepliedRef: { value: boolean } | undefined; + provider: string; +}) { + return { + authProfile: resolveRunAuthProfile(params.run, params.provider), + embeddedContext: buildEmbeddedContextFromTemplate({ + run: params.run, + sessionCtx: params.sessionCtx, + hasRepliedRef: params.hasRepliedRef, + }), + senderContext: buildTemplateSenderContext(params.sessionCtx), + }; +} + export function resolveProviderScopedAuthProfile(params: { provider: string; primaryProvider: string;