| import { normalizeOptionalString } from "@openclaw/normalization-core/string-coerce"; |
| import { resolveBootstrapWarningSignaturesSeen } from "../../agents/bootstrap-budget.js"; |
| import type { |
| CompactionAccountingFact, |
| RunEmbeddedAgentInternalParams, |
| } from "../../agents/embedded-agent-runner/run/internal-params.js"; |
| import { runEmbeddedAgent } from "../../agents/embedded-agent.js"; |
| import { resolveAgentHarnessPolicy } from "../../agents/harness/policy.js"; |
| import { resolveOpenAIRuntimeProvider } from "../../agents/openai-routing.js"; |
| import type { CompactionRequestBudget } from "../../agents/sessions/compaction/request-budget.js"; |
| import { resolveGroupSessionKey } from "../../config/sessions.js"; |
| import { logVerbose } from "../../globals.js"; |
| import { resolveSessionPinnedHarnessId } from "../../sessions/agent-harness-session-key.js"; |
| import { |
| isMarkdownCapableMessageChannel, |
| resolveMessageChannel, |
| } from "../../utils/message-channel.js"; |
| import type { PartialReplyPayload } from "../get-reply-options.types.js"; |
| import type { ReplyPayload } from "../types.js"; |
| import { createAgentLifecycleTerminalBackstop } from "./agent-lifecycle-terminal.js"; |
| import { |
| createAgentRunEventHandler, |
| type MessageToolDeliveryState, |
| } from "./agent-runner-event-handler.js"; |
| import type { CompletedAgentAuthSelection } from "./agent-runner-execution.types.js"; |
| import type { AgentFallbackCandidateCommonParams } from "./agent-runner-fallback-cycle.types.js"; |
| import { buildEmbeddedRunExecutionParams } from "./agent-runner-utils.js"; |
| import { resolveReplyOperationTerminationFields } from "./reply-operation-abort.js"; |
| import { markReplyOperationGlobalLaneWaitProgress } from "./reply-run-registry.js"; |
| import { |
| bindSourceReplyDeliveryRuntime, |
| readSourceReplyDeliveryRuntime, |
| } from "./source-reply-delivery-runtime.js"; |
|
|
| export async function runEmbeddedFallbackCandidate( |
| params: AgentFallbackCandidateCommonParams & { |
| effectiveRun: AgentFallbackCandidateCommonParams["candidateRun"]; |
| sessionRuntimeOverride?: string; |
| getLifecycleGeneration: () => string; |
| onLifecycleGeneration: (generation: string) => void; |
| allowTransientCooldownProbe?: boolean; |
| notifyUserAboutCompaction: boolean; |
| messageToolDeliveryState: MessageToolDeliveryState; |
| githubPublicationAvailable: boolean; |
| onCompactionFacts: (facts: { |
| accounting?: CompactionAccountingFact; |
| postCompactionModelAttempted: boolean; |
| }) => void; |
| }, |
| ): Promise<{ |
| result: Awaited<ReturnType<typeof runEmbeddedAgent>>; |
| maintenanceAuthProfile?: CompletedAgentAuthSelection; |
| compactionRequestBudget?: CompactionRequestBudget; |
| bootstrapPromptWarningSignaturesSeen: string[]; |
| }> { |
| const turn = params.turn; |
| let maintenanceAuthProfile: CompletedAgentAuthSelection | undefined; |
| let compactionRequestBudget: CompactionRequestBudget | undefined; |
| const sourceReplyDeliveryRuntime = readSourceReplyDeliveryRuntime(params.candidateRun); |
| const candidateRun = { |
| ...params.candidateRun, |
| ...params.candidateFastMode, |
| thinkLevel: params.candidateThinkLevel, |
| }; |
| const { embeddedContext, senderContext, runBaseParams } = await buildEmbeddedRunExecutionParams({ |
| run: candidateRun, |
| 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, |
| }); |
| if (sourceReplyDeliveryRuntime) { |
| bindSourceReplyDeliveryRuntime(runBaseParams, sourceReplyDeliveryRuntime); |
| } |
| 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); |
| let attemptCompactionCount = 0; |
| let postCompactionModelAttempted = false; |
| let compactionAccounting: CompactionAccountingFact | undefined; |
| const lifecycleBackstop = createAgentLifecycleTerminalBackstop({ |
| runId: params.runId, |
| sessionKey: turn.sessionKey, |
| getLifecycleGeneration: params.getLifecycleGeneration, |
| resolveTerminationFields: (error) => |
| resolveReplyOperationTerminationFields(error, params.runAbortSignal, turn.replyOperation), |
| }); |
| params.onLifecycleBackstop(lifecycleBackstop); |
| try { |
| |
| params.timing.logMilestoneIfSlow({ |
| runId: params.runId, |
| sessionId: turn.followupRun.run.sessionId, |
| sessionKey: turn.sessionKey, |
| milestone: "before_embedded_run", |
| }); |
| let eventHandler: ReturnType<typeof createAgentRunEventHandler> | undefined; |
| const result = await params.timing.measure("embedded_run", () => { |
| const embeddedRunParams: RunEmbeddedAgentInternalParams = { |
| preparedRunAdmission: params.preparedRunAdmission, |
| githubPublicationAvailable: params.githubPublicationAvailable, |
| ...embeddedContext, |
| messageActionTurnCapability: params.messageActionTurnCapability, |
| lifecycleGeneration: params.getLifecycleGeneration(), |
| allowGatewaySubagentBinding: true, |
| trigger: turn.isHeartbeat ? "heartbeat" : "user", |
| cronCreatorAuthorityCapability: turn.opts?.cronCreatorAuthorityCapability, |
| cronCreatorAuthorityUnavailableReason: |
| turn.opts?.turnAdoptionLifecycle?.cronCreatorAuthorityUnavailable, |
| groupId: resolveGroupSessionKey(turn.sessionCtx)?.id, |
| groupChannel: |
| normalizeOptionalString(turn.sessionCtx.GroupChannel) ?? |
| normalizeOptionalString(turn.sessionCtx.GroupSubject), |
| groupSpace: normalizeOptionalString(turn.sessionCtx.GroupSpace), |
| ...senderContext, |
| ...runBaseParams, |
| contextWindow: turn.getActiveSessionEntry()?.contextWindow, |
| lane: params.runLane, |
| provider: embeddedRunProvider, |
| agentHarnessId: resolveSessionPinnedHarnessId(turn.getActiveSessionEntry()), |
| agentHarnessRuntimeOverride: embeddedRunHarnessOverride, |
| agentHarnessRuntimePreparationHint: |
| agentHarnessPolicy.runtimeSource !== "implicit" ? agentHarnessPolicy.runtime : undefined, |
| fastModeStartedAtMs: params.fastModeStartedAtMs, |
| fastModeAutoProgressState: params.fastModeAutoProgressState, |
| isFinalFallbackAttempt: params.isFinalFallbackAttempt, |
| sandboxSessionKey: turn.runtimePolicySessionKey, |
| prompt: turn.commandBody, |
| transcriptPrompt: turn.transcriptCommandBody, |
| media: turn.followupRun.media, |
| userTurnTranscriptRecorder: params.userTurnTranscriptRecorder, |
| contextEngineLogicalTurnLease: params.contextEngineLogicalTurnLease, |
| onContextEngineTurnCandidate: params.onContextEngineTurnCandidate, |
| currentInboundEventKind: turn.followupRun.currentInboundEventKind, |
| currentInboundContext: turn.followupRun.currentInboundContext, |
| explicitSkillSelections: turn.followupRun.explicitSkillSelections, |
| extraSystemPrompt: turn.followupRun.run.extraSystemPrompt, |
| sourceReplyDeliveryMode: turn.followupRun.run.sourceReplyDeliveryMode, |
| forceMessageTool: turn.followupRun.run.sourceReplyDeliveryMode === "message_tool_only", |
| |
| |
| ...(turn.isHeartbeat ? { requireExplicitMessageTarget: true } : {}), |
| cleanupBundleMcpOnRunEnd: turn.opts?.cleanupBundleMcpOnRunEnd, |
| silentReplyPromptMode: turn.followupRun.run.silentReplyPromptMode, |
| suppressNextUserMessagePersistence: params.suppressQueuedUserPersistenceForCandidate, |
| onUserMessagePersisted: params.notifyUserMessagePersisted, |
| suppressTranscriptOnlyAssistantPersistence: |
| turn.followupRun.run.suppressTranscriptOnlyAssistantPersistence, |
| assistantErrorTranscript: params.assistantErrorTranscript, |
| authProfileFailurePolicy: params.authProfileFailurePolicy, |
| prepareAssistantTranscriptMessage: turn.opts?.prepareAssistantTranscriptMessage, |
| onAutoCompactionSucceeded: (count) => { |
| attemptCompactionCount = Math.max(attemptCompactionCount, count); |
| }, |
| toolResultFormat: (() => { |
| const channel = resolveMessageChannel(turn.sessionCtx.Surface, turn.sessionCtx.Provider); |
| return !channel || isMarkdownCapableMessageChannel(channel) ? "markdown" : "plain"; |
| })(), |
| toolProgressDetail: turn.toolProgressDetail, |
| toolsAllow: turn.opts?.toolsAllow, |
| disableTools: turn.opts?.disableTools, |
| |
| toolAuthorityFingerprint: turn.replyOperation?.toolAuthorityFingerprint, |
| 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, |
| onAttemptStart: lifecycleBackstop.beginAttempt, |
| onCompactionAccounting: (fact) => { |
| compactionAccounting = fact; |
| }, |
| onCompactionRequestBudget: (budget) => { |
| compactionRequestBudget = budget; |
| }, |
| onDeferredLifecycleOwner: params.deferredLifecycle.adopt, |
| onDeferredLifecycleAbort: params.deferredLifecycle.abort, |
| onRetryWait: params.deferredLifecycle.beginRetryWait, |
| onExecutionStarted: (info) => { |
| if (info?.lifecycleGeneration) { |
| params.onLifecycleGeneration(info.lifecycleGeneration); |
| } |
| }, |
| onExecutionPhase: (info) => { |
| if (info.phase === "model_call_started" && attemptCompactionCount > 0) { |
| postCompactionModelAttempted = true; |
| } |
| params.signalExecutionPhaseForTyping(info); |
| }, |
| onLaneWait: ({ waiting }) => { |
| const replyOperation = turn.replyOperation; |
| if (waiting && replyOperation) { |
| markReplyOperationGlobalLaneWaitProgress(replyOperation); |
| } |
| }, |
| blockReplyBreak: turn.resolvedBlockStreamingBreak, |
| blockReplyChunking: turn.blockReplyChunking, |
| |
| onPartialReply: async (payload) => { |
| const classified = params.presentation.classifyStreamingPartial(payload); |
| if (classified.skip || !classified.text) { |
| return false; |
| } |
| const textForTyping = classified.text; |
| let didMaterialize = false; |
| let materializedText: string | undefined; |
| const partialPayload: PartialReplyPayload = { |
| get text() { |
| if (!didMaterialize) { |
| const sanitized = params.presentation.sanitizeStreamingText(textForTyping, false); |
| materializedText = sanitized.skip ? undefined : sanitized.text; |
| didMaterialize = true; |
| } |
| return materializedText; |
| }, |
| mediaUrls: payload.mediaUrls, |
| }; |
| const onPartialReply = turn.opts?.onPartialReply; |
| return await params.presentation.presentWithTyping( |
| turn.typingSignals.signalTextDelta(textForTyping), |
| () => (onPartialReply ? onPartialReply(partialPayload) : false), |
| ); |
| }, |
| onAssistantMessageStart: async () => { |
| await params.presentation.presentWithTyping(turn.typingSignals.signalMessageStart(), () => |
| turn.opts?.onAssistantMessageStart?.(), |
| ); |
| }, |
| onReasoningStream: |
| turn.typingSignals.shouldStartOnReasoning || turn.opts?.onReasoningStream |
| ? async (payload) => { |
| if (turn.followupRun.run.silentExpected) { |
| return; |
| } |
| await params.presentation.presentWithTyping( |
| 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 |
| ? async () => { |
| await turn.opts?.onReasoningEnd?.(); |
| } |
| : undefined, |
| onAgentEvent: (event) => { |
| eventHandler ??= createAgentRunEventHandler({ |
| turn, |
| lifecycleBackstop, |
| notifyAgentRunStart: params.notifyAgentRunStart, |
| sourceRepliesAreToolOnly: |
| (sourceReplyDeliveryRuntime?.currentMode ?? |
| turn.followupRun.run.sourceReplyDeliveryMode) === "message_tool_only", |
| messageToolDeliveryState: params.messageToolDeliveryState, |
| provider: params.provider, |
| model: params.model, |
| runId: params.runId, |
| effectiveSessionId: params.effectiveRun.sessionId, |
| notifyUserAboutCompaction: params.notifyUserAboutCompaction, |
| onCompactionCompleted: () => { |
| attemptCompactionCount += 1; |
| return attemptCompactionCount; |
| }, |
| }); |
| return eventHandler(event); |
| }, |
| |
| 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 |
| ? (() => { |
| |
| let toolResultChain: Promise<void> = Promise.resolve(); |
| return (payload: ReplyPayload) => { |
| const delivery = 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 }); |
| }); |
| |
| toolResultChain = delivery.catch((err: unknown) => { |
| logVerbose(`tool result delivery failed: ${String(err)}`); |
| }); |
| const task = toolResultChain.finally(() => { |
| turn.pendingToolTasks.delete(task); |
| }); |
| turn.pendingToolTasks.add(task); |
| return delivery; |
| }; |
| })() |
| : undefined, |
| }; |
| embeddedRunParams.onSuccessfulAuthProfile = (profileId) => { |
| maintenanceAuthProfile = { |
| authProfileId: profileId, |
| authProfileIdSource: profileId |
| ? profileId === runBaseParams.authProfileId |
| ? runBaseParams.authProfileIdSource |
| : "auto" |
| : undefined, |
| }; |
| }; |
| return runEmbeddedAgent(embeddedRunParams); |
| }); |
| const resultCompactionCount = Math.max(0, result.meta?.agentMeta?.compactionCount ?? 0); |
| attemptCompactionCount = Math.max(attemptCompactionCount, resultCompactionCount); |
| return { |
| result, |
| maintenanceAuthProfile, |
| compactionRequestBudget, |
| bootstrapPromptWarningSignaturesSeen: resolveBootstrapWarningSignaturesSeen( |
| result.meta?.systemPromptReport, |
| ), |
| }; |
| } finally { |
| |
| const accounting: CompactionAccountingFact | undefined = |
| compactionAccounting ?? |
| (attemptCompactionCount > 0 |
| ? { |
| kind: "presentation-only", |
| count: attemptCompactionCount, |
| currentContextSnapshot: { tokens: undefined }, |
| } |
| : undefined); |
| params.onCompactionFacts({ accounting, postCompactionModelAttempted }); |
| } |
| } |
|
|
