diff --git a/src/agents/embedded-agent-runner/run/attempt-bundle-tools.ts b/src/agents/embedded-agent-runner/run/attempt-bundle-tools.ts index f12a98cc5cf8..3bc6bbf56bde 100644 --- a/src/agents/embedded-agent-runner/run/attempt-bundle-tools.ts +++ b/src/agents/embedded-agent-runner/run/attempt-bundle-tools.ts @@ -14,12 +14,12 @@ import { captureFinalEffectiveCronCreatorToolAllowlist } from "../../tools/cron- import { applyFinalEffectiveToolPolicy } from "../effective-tool-policy.js"; import { log } from "../logger.js"; import type { prepareEmbeddedAttemptSetup } from "./attempt-setup.js"; -import type { prepareEmbeddedAttemptToolBase } from "./attempt-tool-base-prepare.js"; import { applyEmbeddedAttemptToolsAllow, shouldCreateBundleLspRuntimeForAttempt, shouldCreateBundleMcpRuntimeForAttempt, } from "./attempt-tool-construction-plan.js"; +import type { prepareEmbeddedAttemptToolBase } from "./attempt-tool-prepare.js"; import type { EmbeddedRunAttemptParams } from "./types.js"; type AttemptSetup = Awaited>; diff --git a/src/agents/embedded-agent-runner/run/attempt-execution-types.ts b/src/agents/embedded-agent-runner/run/attempt-execution-types.ts index 989d9b1504ca..2bfbb9b66501 100644 --- a/src/agents/embedded-agent-runner/run/attempt-execution-types.ts +++ b/src/agents/embedded-agent-runner/run/attempt-execution-types.ts @@ -10,8 +10,8 @@ import type { prepareEmbeddedAttemptSessionRuntime } from "./attempt-session-run import type { prepareEmbeddedAttemptSetup } from "./attempt-setup.js"; import type { prepareEmbeddedAttemptStreamRuntime } from "./attempt-stream-runtime-prepare.js"; import type { prepareEmbeddedAttemptSystemPrompt } from "./attempt-system-prompt-prepare.js"; -import type { prepareEmbeddedAttemptToolBase } from "./attempt-tool-base-prepare.js"; import type { prepareEmbeddedAttemptToolCatalog } from "./attempt-tool-catalog.js"; +import type { prepareEmbeddedAttemptToolBase } from "./attempt-tool-prepare.js"; import type { prepareEmbeddedAttemptTranscriptLifecycle } from "./attempt-transcript-lifecycle-prepare.js"; import type { EmbeddedRunAttemptParams } from "./types.js"; diff --git a/src/agents/embedded-agent-runner/run/attempt-session-boundary.test.ts b/src/agents/embedded-agent-runner/run/attempt-session-boundary.test.ts index 22a41cc76988..703480ee2b74 100644 --- a/src/agents/embedded-agent-runner/run/attempt-session-boundary.test.ts +++ b/src/agents/embedded-agent-runner/run/attempt-session-boundary.test.ts @@ -3,7 +3,7 @@ import { buildTimestampPrefix } from "../../../gateway/server-methods/agent-time import type { AgentMessage } from "../../runtime/index.js"; import type { guardSessionManager } from "../../session-tool-result-guard-wrapper.js"; import type { AgentSession } from "../../sessions/index.js"; -import { prepareEmbeddedAttemptSessionBoundary } from "./attempt-session-boundary.js"; +import { prepareEmbeddedAttemptSessionBoundary } from "./attempt-session-prepare.js"; function createActiveSession(messages: AgentMessage[] = []) { const reset = vi.fn(); diff --git a/src/agents/embedded-agent-runner/run/attempt-session-boundary.ts b/src/agents/embedded-agent-runner/run/attempt-session-boundary.ts deleted file mode 100644 index f1277c01feb0..000000000000 --- a/src/agents/embedded-agent-runner/run/attempt-session-boundary.ts +++ /dev/null @@ -1,127 +0,0 @@ -/** Prepares the restored transcript at the LLM boundary for one attempt. */ -import { resolveUserTimezone } from "../../date-time.js"; -import { relocateCurrentRuntimeContextCarrierToTail } from "../../internal-runtime-context.js"; -import type { AgentMessage } from "../../runtime/index.js"; -import type { guardSessionManager } from "../../session-tool-result-guard-wrapper.js"; -import type { AgentSession } from "../../sessions/index.js"; -import { - replayTrailingEntriesForOrphanRepair, - resolveOrphanRepairPlan, -} from "./attempt-orphan-repair.js"; -import { normalizeMessagesForLlmBoundary } from "./attempt.llm-boundary.js"; -import { reconcilePrePersistedCurrentUserTurn } from "./pre-persisted-user-turn.js"; -import type { EmbeddedRunAttemptParams } from "./types.js"; - -type SessionBoundaryAttempt = Pick< - EmbeddedRunAttemptParams, - | "config" - | "onUserMessagePersistenceInvalidated" - | "operation" - | "prompt" - | "suppressNextUserMessagePersistence" - | "trigger" - | "userTurnTranscriptRecorder" ->; - -type LlmBoundaryOptions = NonNullable[1]>; - -type CurrentUserTimestampOverride = NonNullable; - -export function prepareEmbeddedAttemptSessionBoundary(input: { - activeSession: Pick; - attempt: SessionBoundaryAttempt; - getUserTranscriptContexts: () => LlmBoundaryOptions["userTranscriptContexts"]; - isRawModelRun: boolean; - preparedUserTurnMessage: AgentMessage | undefined; - sessionManager: ReturnType; - setActiveSessionSystemPrompt: (systemPrompt: string) => void; -}): { - boundaryTimezone: string | undefined; - includeBoundaryTimestamp: boolean; - orphanRepair: ReturnType; - setCurrentUserTimestampOverride: (override: CurrentUserTimestampOverride | undefined) => void; -} { - const { activeSession, attempt, isRawModelRun, sessionManager } = input; - const preserveExactPrompt = isRawModelRun || attempt.operation === "settled-tool-finalization"; - if (isRawModelRun) { - // Raw probes measure only the requested provider prompt. Restored history, - // queued work, and the normal system prompt would contaminate it. - activeSession.agent.reset(); - input.setActiveSessionSystemPrompt(""); - } - - const orphanRepairCandidate = preserveExactPrompt - ? undefined - : resolveOrphanRepairPlan({ - sessionManager, - prompt: attempt.prompt, - trigger: attempt.trigger, - }); - // Admission can persist the turn before prompt preparation intentionally omits it. - // Prefer the recorder-owned row so orphan repair cannot detach the canonical leaf. - const currentUserTurnMessage = - attempt.userTurnTranscriptRecorder?.getPersistedMessage?.() ?? input.preparedUserTurnMessage; - const reconciledCurrentUser = - !preserveExactPrompt && - reconcilePrePersistedCurrentUserTurn({ - activeSession, - currentUserTurnMessage, - durableUserTurnMessage: orphanRepairCandidate?.messageEntry.message, - userTurnAlreadyPersisted: attempt.userTurnTranscriptRecorder?.hasPersisted() === true, - }); - const orphanRepair = reconciledCurrentUser ? undefined : orphanRepairCandidate; - if (orphanRepair?.removeLeaf) { - if (orphanRepair.messageEntry.parentId) { - sessionManager.branch(orphanRepair.messageEntry.parentId); - } else { - sessionManager.resetLeaf(); - } - replayTrailingEntriesForOrphanRepair(sessionManager, orphanRepair.trailingEntries); - // The old canonical user turn is gone. Its persistence suppression must not - // discard the merged replacement prompt. - sessionManager.clearNextUserMessagePersistenceSuppression?.(); - attempt.onUserMessagePersistenceInvalidated?.(); - activeSession.agent.state.messages = sessionManager.buildSessionContext().messages; - } - - // This is the single timestamping source for user messages sent to the LLM. - // Raw probes retain exact prompt bytes. - const boundaryTimezone = preserveExactPrompt - ? undefined - : resolveUserTimezone(attempt.config?.agents?.defaults?.userTimezone); - const includeBoundaryTimestamp = !preserveExactPrompt; - let currentUserTimestampOverride: CurrentUserTimestampOverride | undefined; - const buildBoundaryOptions = (): LlmBoundaryOptions => { - if (preserveExactPrompt) { - return { projectPersistedSenderContext: false }; - } - const userTranscriptContexts = input.getUserTranscriptContexts(); - return { - ...(boundaryTimezone ? { timezone: boundaryTimezone } : {}), - ...(includeBoundaryTimestamp ? {} : { includeTimestamp: false }), - ...(userTranscriptContexts?.length ? { userTranscriptContexts } : {}), - ...(currentUserTimestampOverride ? { currentUserTimestampOverride } : {}), - }; - }; - - if (typeof activeSession.agent.convertToLlm === "function") { - const baseConvertToLlm = activeSession.agent.convertToLlm.bind(activeSession.agent); - activeSession.agent.convertToLlm = async (messages) => - await baseConvertToLlm( - // Wire-only relocation keeps the request append-only through the active - // user turn without changing position-sensitive precheck normalization. - relocateCurrentRuntimeContextCarrierToTail( - normalizeMessagesForLlmBoundary(messages, buildBoundaryOptions()), - ), - ); - } - - return { - boundaryTimezone, - includeBoundaryTimestamp, - orphanRepair, - setCurrentUserTimestampOverride: (override) => { - currentUserTimestampOverride = override; - }, - }; -} diff --git a/src/agents/embedded-agent-runner/run/attempt-session-cleanup.test.ts b/src/agents/embedded-agent-runner/run/attempt-session-cleanup.test.ts index d394e7e43a4c..39b2a83392d5 100644 --- a/src/agents/embedded-agent-runner/run/attempt-session-cleanup.test.ts +++ b/src/agents/embedded-agent-runner/run/attempt-session-cleanup.test.ts @@ -20,7 +20,7 @@ vi.mock("./attempt.subscription-cleanup.js", () => ({ cleanupEmbeddedAttemptResources: hoisted.cleanupEmbeddedAttemptResources, })); -import { cleanupEmbeddedAttemptSessionPhase } from "./attempt-session-cleanup.js"; +import { cleanupEmbeddedAttemptSessionPhase } from "./attempt-session-settle.js"; const attempt = { runId: "run-1", diff --git a/src/agents/embedded-agent-runner/run/attempt-session-cleanup.ts b/src/agents/embedded-agent-runner/run/attempt-session-cleanup.ts deleted file mode 100644 index 54ea96d7e986..000000000000 --- a/src/agents/embedded-agent-runner/run/attempt-session-cleanup.ts +++ /dev/null @@ -1,153 +0,0 @@ -/** - * Finalizes trajectory and session-owned resources for one embedded attempt. - */ -import { formatErrorMessage, toErrorObject } from "../../../infra/errors.js"; -import type { createTrajectoryRuntimeRecorder } from "../../../trajectory/runtime.js"; -import type { guardSessionManager } from "../../session-tool-result-guard-wrapper.js"; -import type { AgentSession } from "../../sessions/index.js"; -import { clearToolSearchCatalog, type ToolSearchCatalogRef } from "../../tool-search.js"; -import { log } from "../logger.js"; -import { flushPendingToolResultsAfterIdle } from "../wait-for-idle-before-flush.js"; -import type { EmitDiagnosticRunCompleted } from "./attempt-startup.js"; -import { flushEmbeddedAttemptTrajectoryRecorder } from "./attempt-trajectory-flush-cleanup.js"; -import type { createEmbeddedAttemptTranscriptLifecycle } from "./attempt-transcript-lifecycle.js"; -import { cleanupEmbeddedAttemptResources } from "./attempt.subscription-cleanup.js"; -import type { EmbeddedRunAttemptParams } from "./types.js"; - -type AttemptTranscriptLifecycle = ReturnType; -type TrajectoryRecorder = ReturnType; -type DisposableRuntime = { dispose(): Promise | void }; - -type CleanupEmbeddedAttemptSessionInput = { - attempt: EmbeddedRunAttemptParams; - session?: AgentSession; - sessionManager?: ReturnType; - transcriptLifecycle: AttemptTranscriptLifecycle; - bundleMcpRuntime?: DisposableRuntime; - bundleLspRuntime?: DisposableRuntime; - removeToolResultContextGuard?: () => void; - toolSearchCatalogRef?: ToolSearchCatalogRef; - sandboxSessionKey?: string; - sessionAgentId: string; - buildAbortSettlePromise: () => Promise | null; - trajectoryRecorder: TrajectoryRecorder | null; - trajectoryEndRecorded: boolean; - cleanupYieldAborted: boolean; - emitDiagnosticRunCompleted?: EmitDiagnosticRunCompleted; - readState: () => { - aborted: boolean; - externalAbort: boolean; - timedOut: boolean; - idleTimedOut: boolean; - timedOutDuringCompaction: boolean; - timedOutDuringToolExecution: boolean; - timedOutByRunBudget: boolean; - promptError: unknown; - beforeAgentRunBlocked: boolean; - beforeAgentRunBlockedBy?: string; - }; -}; - -export async function cleanupEmbeddedAttemptSessionPhase( - input: CleanupEmbeddedAttemptSessionInput, -): Promise { - const { attempt } = input; - const initialState = input.readState(); - if (input.trajectoryRecorder && !input.trajectoryEndRecorded) { - input.trajectoryRecorder.recordEvent("session.ended", { - status: initialState.promptError - ? "error" - : initialState.aborted || initialState.timedOut - ? "interrupted" - : "cleanup", - aborted: initialState.aborted, - externalAbort: initialState.externalAbort, - timedOut: initialState.timedOut, - idleTimedOut: initialState.idleTimedOut, - timedOutDuringCompaction: initialState.timedOutDuringCompaction, - timedOutDuringToolExecution: initialState.timedOutDuringToolExecution, - timedOutByRunBudget: initialState.timedOutByRunBudget, - promptError: initialState.promptError - ? formatErrorMessage(initialState.promptError) - : undefined, - }); - } - await flushEmbeddedAttemptTrajectoryRecorder({ - runId: attempt.runId, - sessionId: attempt.sessionId, - log, - trajectoryRecorder: input.trajectoryRecorder, - }); - - // Agent retries can report idle before retried tools finish; waiting before - // the flush prevents synthetic missing-tool results (#8643). Teardown keeps - // lock release ahead of runtime disposal so the next attempt can recover. - let cleanupError: unknown; - try { - clearToolSearchCatalog({ - sessionId: attempt.sessionId, - sessionKey: input.sandboxSessionKey, - agentId: input.sessionAgentId, - runId: attempt.runId, - catalogRef: input.toolSearchCatalogRef, - }); - // Abort handling remains armed during cleanup, so reread after trajectory - // flushing instead of using the state captured at helper entry. - const cleanupState = input.readState(); - const cleanupAborted = - Boolean(attempt.abortSignal?.aborted) || - cleanupState.aborted || - cleanupState.timedOut || - cleanupState.idleTimedOut || - cleanupState.timedOutDuringCompaction; - const cleanupAbortLike = cleanupAborted || input.cleanupYieldAborted; - await input.transcriptLifecycle.beginCleanup(); - await cleanupEmbeddedAttemptResources({ - removeToolResultContextGuard: input.removeToolResultContextGuard, - flushPendingToolResultsAfterIdle, - session: input.session, - sessionManager: input.sessionManager, - bundleMcpRuntime: input.bundleMcpRuntime, - bundleLspRuntime: input.bundleLspRuntime, - // Aborted runs skip the idle wait so teardown cannot strand the lock. - aborted: cleanupAbortLike, - abortSettlePromise: cleanupAborted ? input.buildAbortSettlePromise() : null, - runId: attempt.runId, - sessionId: attempt.sessionId, - }); - } catch (err) { - cleanupError = err; - } finally { - try { - await input.transcriptLifecycle.dispose(); - } catch (err) { - cleanupError ??= err; - } - } - - const finalState = input.readState(); - const cleanupFailure = cleanupError; - input.emitDiagnosticRunCompleted?.( - cleanupFailure - ? "error" - : finalState.beforeAgentRunBlocked - ? "blocked" - : finalState.promptError - ? "error" - : finalState.aborted || - finalState.timedOut || - finalState.idleTimedOut || - finalState.timedOutDuringCompaction - ? "aborted" - : "completed", - cleanupFailure ?? finalState.promptError, - finalState.beforeAgentRunBlocked - ? { blockedBy: finalState.beforeAgentRunBlockedBy ?? "before_agent_run" } - : undefined, - ); - - if (!cleanupFailure) { - return; - } - await Promise.reject(toErrorObject(cleanupFailure, "Non-Error rejection")); -} diff --git a/src/agents/embedded-agent-runner/run/attempt-session-manager-prepare.ts b/src/agents/embedded-agent-runner/run/attempt-session-manager-prepare.ts deleted file mode 100644 index 37acebec1dad..000000000000 --- a/src/agents/embedded-agent-runner/run/attempt-session-manager-prepare.ts +++ /dev/null @@ -1,198 +0,0 @@ -import type { SessionTranscriptRuntimeTarget } from "../../../config/sessions/session-accessor.types.js"; -/** - * Prepares the durable session manager before embedded-agent session creation. - */ -import { OPENCLAW_EMBEDDED_CONTEXT_ENGINE_HOST } from "../../../context-engine/host-compat.js"; -import type { AgentMessage } from "../../runtime/index.js"; -import { guardSessionManager } from "../../session-tool-result-guard-wrapper.js"; -import { SessionManager } from "../../sessions/index.js"; -import { runContextEngineMaintenance } from "../context-engine-maintenance.js"; -import { log } from "../logger.js"; -import { resolveExistingAttemptTranscriptState } from "./attempt-transcript-helpers.js"; -import type { EmbeddedAttemptTranscriptLifecycle } from "./attempt-transcript-lifecycle.js"; -import { - runAttemptContextEngineBootstrap, - type AttemptContextEngine, -} from "./attempt.context-engine-helpers.js"; -import { buildAfterTurnRuntimeContext } from "./attempt.prompt-helpers.js"; -import { resolveAttemptTranscriptPolicy } from "./attempt.transcript-policy.js"; -import { createUserTranscriptContextRegistry } from "./attempt.user-transcript-context-registry.js"; -import { resolveSessionBoundaryPromptCacheKey } from "./session-boundary-prompt-cache-key.js"; -import type { EmbeddedRunAttemptParams } from "./types.js"; - -type AttemptSessionManager = ReturnType; -type WithOwnedTranscriptWrite = (operation: () => Promise | T) => Promise; -export async function prepareEmbeddedAttemptSessionManager(input: { - attempt: EmbeddedRunAttemptParams; - activeContextEngine?: AttemptContextEngine; - agentDir: string; - effectiveCwd: string; - effectiveWorkspace: string; - onSessionManagerCreated: (sessionManager: AttemptSessionManager) => void; - replayAllowedToolNames: ReadonlySet; - resolveActiveContextEnginePluginId: () => string | undefined; - sessionAgentId: string; - transcriptLifecycle: EmbeddedAttemptTranscriptLifecycle; - withOwnedTranscriptWrite: WithOwnedTranscriptWrite; -}) { - const { attempt } = input; - const transcriptState = await resolveExistingAttemptTranscriptState({ - agentId: input.sessionAgentId, - config: attempt.config, - sessionFile: attempt.sessionFile, - sessionId: attempt.sessionId, - sessionKey: attempt.sessionKey, - sessionTarget: attempt.sessionTarget, - }); - const transcriptPolicy = resolveAttemptTranscriptPolicy({ - runtimePlan: attempt.runtimePlan, - runtimePlanModelContext: { - workspaceDir: input.effectiveWorkspace, - modelApi: attempt.model.api, - model: attempt.model, - }, - provider: attempt.provider, - modelId: attempt.modelId, - config: attempt.config, - env: process.env, - }); - const isOpenAIResponsesApi = - attempt.model.api === "openai-responses" || - attempt.model.api === "azure-openai-responses" || - attempt.model.api === "openai-chatgpt-responses"; - - const preparedUserTurnMessage = attempt.skipPreparedUserTurnMessage - ? undefined - : await attempt.userTurnTranscriptRecorder?.resolveMessage(); - let latestPersistedUserMessage: AgentMessage | undefined; - let latestRuntimeUserMessage: AgentMessage | undefined; - let latestUserTurnTranscriptRecorder = attempt.userTurnTranscriptRecorder; - const userTranscriptContextRegistry = createUserTranscriptContextRegistry(); - const sessionManager = guardSessionManager( - attempt.sessionManager ?? - (attempt.sessionTarget - ? SessionManager.open( - attempt.sessionTarget as SessionTranscriptRuntimeTarget, - input.effectiveCwd, - ) - : SessionManager.inMemory(input.effectiveCwd)), - { - agentId: input.sessionAgentId, - sessionKey: attempt.sessionKey, - config: attempt.config, - contextWindowTokens: attempt.contextTokenBudget, - inputProvenance: attempt.inputProvenance, - preparedUserTurnMessage, - preparedUserTurnTranscriptRecorder: preparedUserTurnMessage - ? attempt.userTurnTranscriptRecorder - : undefined, - allowSyntheticToolResults: transcriptPolicy.allowSyntheticToolResults, - missingToolResultText: isOpenAIResponsesApi ? "aborted" : undefined, - allowedToolNames: input.replayAllowedToolNames, - trigger: attempt.trigger, - suppressNextUserMessagePersistence: attempt.suppressNextUserMessagePersistence, - suppressTranscriptOnlyAssistantPersistence: - attempt.suppressTranscriptOnlyAssistantPersistence, - suppressAssistantErrorPersistence: attempt.suppressAssistantErrorPersistence, - skipBeforeMessageWriteHooks: attempt.operation === "settled-tool-finalization", - onUserMessagePreparingForPersistence: (_message, recorder) => { - latestPersistedUserMessage = undefined; - latestUserTurnTranscriptRecorder = recorder; - }, - onUserMessagePersisted: (message, runtimeMessage) => { - latestPersistedUserMessage = message; - latestRuntimeUserMessage = runtimeMessage; - if (runtimeMessage) { - userTranscriptContextRegistry.record(runtimeMessage, message); - } - attempt.onUserMessagePersisted?.(message); - }, - onUserMessagePersistenceSuppressed: (_message, runtimeMessage) => { - latestRuntimeUserMessage = runtimeMessage; - }, - onUserMessageBlocked: () => { - attempt.userTurnTranscriptRecorder?.markBlocked(); - }, - onAssistantErrorMessagePersisted: (message) => { - attempt.onAssistantErrorMessagePersisted?.(message); - }, - }, - ); - attempt.promptCacheKey = resolveSessionBoundaryPromptCacheKey({ - api: attempt.model.api, - boundaryCount: sessionManager.getBoundaryCount(), - promptCacheKey: attempt.promptCacheKey, - sessionId: attempt.sessionId, - }); - // Publish ownership before async bootstrap. Outer cleanup must close this manager - // even when a context-engine or transcript preparation step fails. - input.onSessionManagerCreated(sessionManager); - - await input.withOwnedTranscriptWrite(async () => { - await runAttemptContextEngineBootstrap({ - hadSessionFile: transcriptState.hasBootstrapTranscriptState, - contextEngine: input.activeContextEngine, - sessionId: attempt.sessionId, - sessionKey: attempt.sessionKey, - sessionTarget: attempt.sessionTarget, - sessionFile: attempt.sessionFile, - sessionManager, - runtimeContext: buildAfterTurnRuntimeContext({ - attempt, - workspaceDir: input.effectiveWorkspace, - cwd: input.effectiveCwd, - agentDir: input.agentDir, - tokenBudget: attempt.contextTokenBudget, - activeAgentId: input.sessionAgentId, - contextEnginePluginId: input.resolveActiveContextEnginePluginId(), - }), - contextEngineHostSupport: OPENCLAW_EMBEDDED_CONTEXT_ENGINE_HOST, - providerId: attempt.provider, - requestedModelId: attempt.requestedModelId, - modelId: attempt.modelId, - fallbackReason: attempt.fallbackReason, - degradedReason: attempt.degradedReason, - runMaintenance: async (contextParams) => - await runContextEngineMaintenance({ - contextEngine: contextParams.contextEngine as never, - sessionId: contextParams.sessionId, - sessionKey: contextParams.sessionKey, - sessionTarget: contextParams.sessionTarget, - sessionFile: contextParams.sessionFile, - reason: contextParams.reason, - sessionManager: contextParams.sessionManager as never, - runtimeContext: contextParams.runtimeContext, - runtimeSettings: contextParams.runtimeSettings, - config: attempt.config, - agentId: input.sessionAgentId, - }), - warn: (message) => log.warn(message), - }); - }); - // Bootstrap may repair or migrate transcript rows. Only user writes after - // preparation can be the active prompt source at the provider boundary. - latestPersistedUserMessage = undefined; - latestRuntimeUserMessage = undefined; - userTranscriptContextRegistry.clear(); - - return { - userMessageBoundary: { - getUserTranscriptContexts: () => { - const transcriptMessage = - latestPersistedUserMessage ?? latestUserTurnTranscriptRecorder?.getPersistedMessage?.(); - // A suppressed retry reuses the canonical persisted row, while the SDK - // may rebuild its runtime object. Match against that row as the stable - // fallback after preferring the exact suppressed runtime correlation. - const runtimeMessage = - latestRuntimeUserMessage ?? - (attempt.suppressNextUserMessagePersistence ? transcriptMessage : undefined); - return userTranscriptContextRegistry.list(runtimeMessage, transcriptMessage); - }, - preparedUserTurnMessage, - }, - isOpenAIResponsesApi, - preparedUserTurnMessage, - sessionManager, - transcriptPolicy, - }; -} diff --git a/src/agents/embedded-agent-runner/run/attempt-session-prepare.ts b/src/agents/embedded-agent-runner/run/attempt-session-prepare.ts new file mode 100644 index 000000000000..af5bf0f40326 --- /dev/null +++ b/src/agents/embedded-agent-runner/run/attempt-session-prepare.ts @@ -0,0 +1,541 @@ +/** + * Prepares transcript boundaries, session management, and active resources. + * It may assume attempt configuration and tool inputs are ready. + */ +import type { SessionTranscriptRuntimeTarget } from "../../../config/sessions/session-accessor.types.js"; +import { OPENCLAW_EMBEDDED_CONTEXT_ENGINE_HOST } from "../../../context-engine/host-compat.js"; +import { getGlobalHookRunner } from "../../../plugins/hook-runner-global.js"; +import type { PluginMetadataSnapshot } from "../../../plugins/plugin-metadata-snapshot.types.js"; +import { createPreparedEmbeddedAgentSettingsManager } from "../../agent-project-settings.js"; +import { + applyAgentAutoCompactionGuard, + applyAgentCompactionSettingsFromConfig, + isSilentOverflowProneModel, + resolveEffectiveCompactionMode, +} from "../../agent-settings.js"; +import { toToolDefinitions } from "../../agent-tool-definition-adapter.js"; +import { resolveUserTimezone } from "../../date-time.js"; +import { relocateCurrentRuntimeContextCarrierToTail } from "../../internal-runtime-context.js"; +import type { AgentMessage } from "../../runtime/index.js"; +import { guardSessionManager } from "../../session-tool-result-guard-wrapper.js"; +import { + type AgentSession, + type CreateAgentSessionOptions, + SessionManager, +} from "../../sessions/index.js"; +import { createAgentSessionForEmbeddedRunner } from "../../sessions/sdk.js"; +import { wrapToolDefinition } from "../../sessions/tools/tool-definition-wrapper.js"; +import { resolveToolSearchCatalogTool } from "../../tool-search.js"; +import { runContextEngineMaintenance } from "../context-engine-maintenance.js"; +import { buildEmbeddedExtensionFactories } from "../extensions.js"; +import { log } from "../logger.js"; +import { createEmbeddedAgentResourceLoader } from "../resource-loader.js"; +import { applySystemPromptToSession } from "../system-prompt.js"; +import { prepareEmbeddedAttemptClientTools } from "./attempt-client-tools.js"; +import { + replayTrailingEntriesForOrphanRepair, + resolveOrphanRepairPlan, +} from "./attempt-orphan-repair.js"; +import { resolveExistingAttemptTranscriptState } from "./attempt-transcript-helpers.js"; +import type { EmbeddedAttemptTranscriptLifecycle } from "./attempt-transcript-lifecycle.js"; +import { + type AttemptContextEngine, + runAttemptContextEngineBootstrap, +} from "./attempt.context-engine-helpers.js"; +import { normalizeMessagesForLlmBoundary } from "./attempt.llm-boundary.js"; +import { buildAfterTurnRuntimeContext } from "./attempt.prompt-helpers.js"; +import { resolveAttemptTranscriptPolicy } from "./attempt.transcript-policy.js"; +import { createUserTranscriptContextRegistry } from "./attempt.user-transcript-context-registry.js"; +import { installCodeModeRepairHook } from "./code-mode-repair.js"; +import { installMessageToolOnlyTerminalHook } from "./message-tool-terminal.js"; +import { reconcilePrePersistedCurrentUserTurn } from "./pre-persisted-user-turn.js"; +import { resolveSessionBoundaryPromptCacheKey } from "./session-boundary-prompt-cache-key.js"; +import { notifyToolActivity } from "./tool-activity-heartbeat.js"; +import { + createToolLoopBatchAdmission, + installToolLoopRecoveryCleanup, +} from "./tool-loop-recovery.js"; +import type { EmbeddedRunAttemptParams } from "./types.js"; + +/** + * Prepares embedded-agent resources, tools, and active sessions. + */ + +type ClientToolPreparation = Omit< + Parameters[0], + "attempt" +>; + +type AttemptSessionManager = ReturnType; + +/** Prepares resource loading, client tools, and the active agent session. */ +export async function prepareEmbeddedAttemptAgentSession(input: { + attempt: EmbeddedRunAttemptParams; + activeContextEngineInfo?: AttemptContextEngine["info"]; + agentCoreThinkingLevel: CreateAgentSessionOptions["thinkingLevel"]; + agentDir: string; + clientToolPreparation: ClientToolPreparation; + effectiveCwd: string; + getCurrentAttemptPluginMetadataSnapshot: () => PluginMetadataSnapshot | undefined; + initialSystemPrompt: string; + markStage: (stage: string) => void; + onSessionCreated: (session: AgentSession) => void; + onSystemPromptChanged: (systemPrompt: string) => void; + runAbortSignal: AbortSignal; + sessionAgentId: string; + transcriptLifecycle: EmbeddedAttemptTranscriptLifecycle; + sessionManager: AttemptSessionManager; +}) { + const { attempt } = input; + const settingsManager = createPreparedEmbeddedAgentSettingsManager({ + cwd: input.effectiveCwd, + agentDir: input.agentDir, + cfg: attempt.config, + pluginMetadataSnapshot: input.getCurrentAttemptPluginMetadataSnapshot(), + contextTokenBudget: attempt.contextTokenBudget, + }); + const autoCompactionGuardArgs = { + settingsManager, + contextEngineInfo: input.activeContextEngineInfo, + compactionMode: resolveEffectiveCompactionMode(attempt.config), + silentOverflowProneProvider: isSilentOverflowProneModel({ + provider: attempt.provider, + modelId: attempt.modelId, + baseUrl: attempt.model.baseUrl ?? undefined, + }), + }; + applyAgentAutoCompactionGuard(autoCompactionGuardArgs); + + // These factories carry compaction/pruning runtime state into the resource loader. + const extensionFactories = buildEmbeddedExtensionFactories({ + cfg: attempt.config, + sessionManager: input.sessionManager, + provider: attempt.provider, + modelId: attempt.modelId, + model: attempt.model, + contextTokenBudget: attempt.contextTokenBudget, + agentId: input.sessionAgentId, + sessionId: attempt.sessionId, + sessionKey: attempt.sessionKey ?? attempt.sandboxSessionKey, + runId: attempt.runId, + }); + const resourceLoader = createEmbeddedAgentResourceLoader({ + cwd: input.effectiveCwd, + agentDir: input.agentDir, + settingsManager, + extensionFactories, + }); + await resourceLoader.reload(); + // reload() rehydrates disk settings. Reapply OpenClaw's context budget and + // auto-compaction guards before the session can submit a prompt (#75799). + applyAgentCompactionSettingsFromConfig({ + settingsManager, + cfg: attempt.config, + contextTokenBudget: attempt.contextTokenBudget, + }); + applyAgentAutoCompactionGuard(autoCompactionGuardArgs); + input.markStage("session-resource-loader"); + + // Tool creation needs the same runner later used by lifecycle hooks. + const hookRunner = getGlobalHookRunner(); + const preparedClientTools = prepareEmbeddedAttemptClientTools({ + attempt, + ...input.clientToolPreparation, + }); + const { allCustomTools, sessionToolAllowlist, ...clientToolRuntime } = preparedClientTools; + + const sessionOptions: CreateAgentSessionOptions = { + cwd: input.effectiveCwd, + agentDir: input.agentDir, + authStorage: attempt.authStorage, + modelRegistry: attempt.modelRegistry, + model: attempt.model, + thinkingLevel: input.agentCoreThinkingLevel, + tools: sessionToolAllowlist, + customTools: allCustomTools, + sessionManager: input.sessionManager, + settingsManager, + resourceLoader, + resolveDeferredTool: input.clientToolPreparation.deferredDirectoryToolsCallable + ? ({ toolCall }) => { + const tool = resolveToolSearchCatalogTool( + { + config: attempt.config, + runtimeConfig: attempt.config, + agentId: input.sessionAgentId, + sessionKey: input.clientToolPreparation.sandboxSessionKey, + sessionId: attempt.sessionId, + runId: attempt.runId, + catalogRef: input.clientToolPreparation.toolSearchCatalogRef, + abortSignal: input.runAbortSignal, + }, + toolCall.name, + ); + // Catalog entries already own before_tool_call wrapping. + const definition = tool + ? toToolDefinitions([tool], input.clientToolPreparation.catalogToolHookContext)[0] + : undefined; + const hydratedTool = definition ? wrapToolDefinition(definition) : undefined; + if (hydratedTool) { + log.info(`tool-search: hydrated deferred directory tool ${toolCall.name}`); + const originalExecute = hydratedTool.execute; + hydratedTool.execute = (async (...args: Parameters) => { + const interval = setInterval(() => notifyToolActivity(attempt.runId), 60_000); + interval.unref?.(); + try { + notifyToolActivity(attempt.runId); + return await originalExecute(...args); + } finally { + clearInterval(interval); + notifyToolActivity(attempt.runId); + } + }) as typeof originalExecute; + } + return hydratedTool; + } + : undefined, + withSessionWriteSettlement: (operation) => + input.transcriptLifecycle.withTranscriptWrite(operation), + }; + const createdSession = await createAgentSessionForEmbeddedRunner(sessionOptions, { + // Without a resolved model budget, the outer loop cannot own bounded recovery. + contextOverflowRecoveryOwner: attempt.contextTokenBudget === undefined ? "session" : "caller", + beforeToolBatch: input.clientToolPreparation.catalogToolHookContext + ? createToolLoopBatchAdmission(input.clientToolPreparation.catalogToolHookContext) + : undefined, + }); + const activeSession = createdSession.session; + if (!activeSession) { + throw new Error("Embedded agent session missing"); + } + // Publish ownership before post-construction hooks. Outer cleanup must dispose + // the session if tool activation or terminal-hook installation fails. + input.onSessionCreated(activeSession); + installToolLoopRecoveryCleanup({ agent: activeSession.agent, runId: attempt.runId }); + activeSession.setActiveToolsByName(sessionToolAllowlist); + const setActiveSessionSystemPrompt = (nextSystemPrompt: string) => { + input.onSystemPromptChanged(nextSystemPrompt); + applySystemPromptToSession(activeSession, nextSystemPrompt); + }; + setActiveSessionSystemPrompt(input.initialSystemPrompt); + let didDeliverSourceReplyViaMessageTool = false; + const markSourceReplyDelivered = () => { + didDeliverSourceReplyViaMessageTool = true; + }; + installMessageToolOnlyTerminalHook({ + agent: activeSession.agent, + sourceReplyDeliveryMode: attempt.sourceReplyDeliveryMode, + onDeliveredSourceReply: markSourceReplyDelivered, + }); + if (input.clientToolPreparation.codeModeControlsEnabledForRun) { + installCodeModeRepairHook({ agent: activeSession.agent }); + } + input.markStage("agent-session"); + + return { + activeSession, + allCustomTools, + ...clientToolRuntime, + hasDeliveredSourceReply: () => didDeliverSourceReplyViaMessageTool, + hookRunner, + markSourceReplyDelivered, + setActiveSessionSystemPrompt, + settingsManager, + }; +} + +/** Prepares the restored transcript at the LLM boundary for one attempt. */ + +type SessionBoundaryAttempt = Pick< + EmbeddedRunAttemptParams, + | "config" + | "onUserMessagePersistenceInvalidated" + | "operation" + | "prompt" + | "suppressNextUserMessagePersistence" + | "trigger" + | "userTurnTranscriptRecorder" +>; + +type LlmBoundaryOptions = NonNullable[1]>; + +type CurrentUserTimestampOverride = NonNullable; + +export function prepareEmbeddedAttemptSessionBoundary(input: { + activeSession: Pick; + attempt: SessionBoundaryAttempt; + getUserTranscriptContexts: () => LlmBoundaryOptions["userTranscriptContexts"]; + isRawModelRun: boolean; + preparedUserTurnMessage: AgentMessage | undefined; + sessionManager: ReturnType; + setActiveSessionSystemPrompt: (systemPrompt: string) => void; +}): { + boundaryTimezone: string | undefined; + includeBoundaryTimestamp: boolean; + orphanRepair: ReturnType; + setCurrentUserTimestampOverride: (override: CurrentUserTimestampOverride | undefined) => void; +} { + const { activeSession, attempt, isRawModelRun, sessionManager } = input; + const preserveExactPrompt = isRawModelRun || attempt.operation === "settled-tool-finalization"; + if (isRawModelRun) { + // Raw probes measure only the requested provider prompt. Restored history, + // queued work, and the normal system prompt would contaminate it. + activeSession.agent.reset(); + input.setActiveSessionSystemPrompt(""); + } + + const orphanRepairCandidate = preserveExactPrompt + ? undefined + : resolveOrphanRepairPlan({ + sessionManager, + prompt: attempt.prompt, + trigger: attempt.trigger, + }); + // Admission can persist the turn before prompt preparation intentionally omits it. + // Prefer the recorder-owned row so orphan repair cannot detach the canonical leaf. + const currentUserTurnMessage = + attempt.userTurnTranscriptRecorder?.getPersistedMessage?.() ?? input.preparedUserTurnMessage; + const reconciledCurrentUser = + !preserveExactPrompt && + reconcilePrePersistedCurrentUserTurn({ + activeSession, + currentUserTurnMessage, + durableUserTurnMessage: orphanRepairCandidate?.messageEntry.message, + userTurnAlreadyPersisted: attempt.userTurnTranscriptRecorder?.hasPersisted() === true, + }); + const orphanRepair = reconciledCurrentUser ? undefined : orphanRepairCandidate; + if (orphanRepair?.removeLeaf) { + if (orphanRepair.messageEntry.parentId) { + sessionManager.branch(orphanRepair.messageEntry.parentId); + } else { + sessionManager.resetLeaf(); + } + replayTrailingEntriesForOrphanRepair(sessionManager, orphanRepair.trailingEntries); + // The old canonical user turn is gone. Its persistence suppression must not + // discard the merged replacement prompt. + sessionManager.clearNextUserMessagePersistenceSuppression?.(); + attempt.onUserMessagePersistenceInvalidated?.(); + activeSession.agent.state.messages = sessionManager.buildSessionContext().messages; + } + + // This is the single timestamping source for user messages sent to the LLM. + // Raw probes retain exact prompt bytes. + const boundaryTimezone = preserveExactPrompt + ? undefined + : resolveUserTimezone(attempt.config?.agents?.defaults?.userTimezone); + const includeBoundaryTimestamp = !preserveExactPrompt; + let currentUserTimestampOverride: CurrentUserTimestampOverride | undefined; + const buildBoundaryOptions = (): LlmBoundaryOptions => { + if (preserveExactPrompt) { + return { projectPersistedSenderContext: false }; + } + const userTranscriptContexts = input.getUserTranscriptContexts(); + return { + ...(boundaryTimezone ? { timezone: boundaryTimezone } : {}), + ...(includeBoundaryTimestamp ? {} : { includeTimestamp: false }), + ...(userTranscriptContexts?.length ? { userTranscriptContexts } : {}), + ...(currentUserTimestampOverride ? { currentUserTimestampOverride } : {}), + }; + }; + + if (typeof activeSession.agent.convertToLlm === "function") { + const baseConvertToLlm = activeSession.agent.convertToLlm.bind(activeSession.agent); + activeSession.agent.convertToLlm = async (messages) => + await baseConvertToLlm( + // Wire-only relocation keeps the request append-only through the active + // user turn without changing position-sensitive precheck normalization. + relocateCurrentRuntimeContextCarrierToTail( + normalizeMessagesForLlmBoundary(messages, buildBoundaryOptions()), + ), + ); + } + + return { + boundaryTimezone, + includeBoundaryTimestamp, + orphanRepair, + setCurrentUserTimestampOverride: (override) => { + currentUserTimestampOverride = override; + }, + }; +} + +/** + * Prepares the durable session manager before embedded-agent session creation. + */ + +type WithOwnedTranscriptWrite = (operation: () => Promise | T) => Promise; +export async function prepareEmbeddedAttemptSessionManager(input: { + attempt: EmbeddedRunAttemptParams; + activeContextEngine?: AttemptContextEngine; + agentDir: string; + effectiveCwd: string; + effectiveWorkspace: string; + onSessionManagerCreated: (sessionManager: AttemptSessionManager) => void; + replayAllowedToolNames: ReadonlySet; + resolveActiveContextEnginePluginId: () => string | undefined; + sessionAgentId: string; + transcriptLifecycle: EmbeddedAttemptTranscriptLifecycle; + withOwnedTranscriptWrite: WithOwnedTranscriptWrite; +}) { + const { attempt } = input; + const transcriptState = await resolveExistingAttemptTranscriptState({ + agentId: input.sessionAgentId, + config: attempt.config, + sessionFile: attempt.sessionFile, + sessionId: attempt.sessionId, + sessionKey: attempt.sessionKey, + sessionTarget: attempt.sessionTarget, + }); + const transcriptPolicy = resolveAttemptTranscriptPolicy({ + runtimePlan: attempt.runtimePlan, + runtimePlanModelContext: { + workspaceDir: input.effectiveWorkspace, + modelApi: attempt.model.api, + model: attempt.model, + }, + provider: attempt.provider, + modelId: attempt.modelId, + config: attempt.config, + env: process.env, + }); + const isOpenAIResponsesApi = + attempt.model.api === "openai-responses" || + attempt.model.api === "azure-openai-responses" || + attempt.model.api === "openai-chatgpt-responses"; + + const preparedUserTurnMessage = attempt.skipPreparedUserTurnMessage + ? undefined + : await attempt.userTurnTranscriptRecorder?.resolveMessage(); + let latestPersistedUserMessage: AgentMessage | undefined; + let latestRuntimeUserMessage: AgentMessage | undefined; + let latestUserTurnTranscriptRecorder = attempt.userTurnTranscriptRecorder; + const userTranscriptContextRegistry = createUserTranscriptContextRegistry(); + const sessionManager = guardSessionManager( + attempt.sessionManager ?? + (attempt.sessionTarget + ? SessionManager.open( + attempt.sessionTarget as SessionTranscriptRuntimeTarget, + input.effectiveCwd, + ) + : SessionManager.inMemory(input.effectiveCwd)), + { + agentId: input.sessionAgentId, + sessionKey: attempt.sessionKey, + config: attempt.config, + contextWindowTokens: attempt.contextTokenBudget, + inputProvenance: attempt.inputProvenance, + preparedUserTurnMessage, + preparedUserTurnTranscriptRecorder: preparedUserTurnMessage + ? attempt.userTurnTranscriptRecorder + : undefined, + allowSyntheticToolResults: transcriptPolicy.allowSyntheticToolResults, + missingToolResultText: isOpenAIResponsesApi ? "aborted" : undefined, + allowedToolNames: input.replayAllowedToolNames, + trigger: attempt.trigger, + suppressNextUserMessagePersistence: attempt.suppressNextUserMessagePersistence, + suppressTranscriptOnlyAssistantPersistence: + attempt.suppressTranscriptOnlyAssistantPersistence, + suppressAssistantErrorPersistence: attempt.suppressAssistantErrorPersistence, + skipBeforeMessageWriteHooks: attempt.operation === "settled-tool-finalization", + onUserMessagePreparingForPersistence: (_message, recorder) => { + latestPersistedUserMessage = undefined; + latestUserTurnTranscriptRecorder = recorder; + }, + onUserMessagePersisted: (message, runtimeMessage) => { + latestPersistedUserMessage = message; + latestRuntimeUserMessage = runtimeMessage; + if (runtimeMessage) { + userTranscriptContextRegistry.record(runtimeMessage, message); + } + attempt.onUserMessagePersisted?.(message); + }, + onUserMessagePersistenceSuppressed: (_message, runtimeMessage) => { + latestRuntimeUserMessage = runtimeMessage; + }, + onUserMessageBlocked: () => { + attempt.userTurnTranscriptRecorder?.markBlocked(); + }, + onAssistantErrorMessagePersisted: (message) => { + attempt.onAssistantErrorMessagePersisted?.(message); + }, + }, + ); + attempt.promptCacheKey = resolveSessionBoundaryPromptCacheKey({ + api: attempt.model.api, + boundaryCount: sessionManager.getBoundaryCount(), + promptCacheKey: attempt.promptCacheKey, + sessionId: attempt.sessionId, + }); + // Publish ownership before async bootstrap. Outer cleanup must close this manager + // even when a context-engine or transcript preparation step fails. + input.onSessionManagerCreated(sessionManager); + + await input.withOwnedTranscriptWrite(async () => { + await runAttemptContextEngineBootstrap({ + hadSessionFile: transcriptState.hasBootstrapTranscriptState, + contextEngine: input.activeContextEngine, + sessionId: attempt.sessionId, + sessionKey: attempt.sessionKey, + sessionTarget: attempt.sessionTarget, + sessionFile: attempt.sessionFile, + sessionManager, + runtimeContext: buildAfterTurnRuntimeContext({ + attempt, + workspaceDir: input.effectiveWorkspace, + cwd: input.effectiveCwd, + agentDir: input.agentDir, + tokenBudget: attempt.contextTokenBudget, + activeAgentId: input.sessionAgentId, + contextEnginePluginId: input.resolveActiveContextEnginePluginId(), + }), + contextEngineHostSupport: OPENCLAW_EMBEDDED_CONTEXT_ENGINE_HOST, + providerId: attempt.provider, + requestedModelId: attempt.requestedModelId, + modelId: attempt.modelId, + fallbackReason: attempt.fallbackReason, + degradedReason: attempt.degradedReason, + runMaintenance: async (contextParams) => + await runContextEngineMaintenance({ + contextEngine: contextParams.contextEngine as never, + sessionId: contextParams.sessionId, + sessionKey: contextParams.sessionKey, + sessionTarget: contextParams.sessionTarget, + sessionFile: contextParams.sessionFile, + reason: contextParams.reason, + sessionManager: contextParams.sessionManager as never, + runtimeContext: contextParams.runtimeContext, + runtimeSettings: contextParams.runtimeSettings, + config: attempt.config, + agentId: input.sessionAgentId, + }), + warn: (message) => log.warn(message), + }); + }); + // Bootstrap may repair or migrate transcript rows. Only user writes after + // preparation can be the active prompt source at the provider boundary. + latestPersistedUserMessage = undefined; + latestRuntimeUserMessage = undefined; + userTranscriptContextRegistry.clear(); + + return { + userMessageBoundary: { + getUserTranscriptContexts: () => { + const transcriptMessage = + latestPersistedUserMessage ?? latestUserTurnTranscriptRecorder?.getPersistedMessage?.(); + // A suppressed retry reuses the canonical persisted row, while the SDK + // may rebuild its runtime object. Match against that row as the stable + // fallback after preferring the exact suppressed runtime correlation. + const runtimeMessage = + latestRuntimeUserMessage ?? + (attempt.suppressNextUserMessagePersistence ? transcriptMessage : undefined); + return userTranscriptContextRegistry.list(runtimeMessage, transcriptMessage); + }, + preparedUserTurnMessage, + }, + isOpenAIResponsesApi, + preparedUserTurnMessage, + sessionManager, + transcriptPolicy, + }; +} diff --git a/src/agents/embedded-agent-runner/run/attempt-session-runtime-prepare.test.ts b/src/agents/embedded-agent-runner/run/attempt-session-runtime-prepare.test.ts index ad1bbba47fae..5abb2f58bd78 100644 --- a/src/agents/embedded-agent-runner/run/attempt-session-runtime-prepare.test.ts +++ b/src/agents/embedded-agent-runner/run/attempt-session-runtime-prepare.test.ts @@ -23,18 +23,14 @@ vi.mock("../session-prompt-state.js", () => ({ vi.mock("./attempt-context-guards.js", () => ({ installEmbeddedAttemptContextGuards: mocks.installContextGuards, })); -vi.mock("./attempt-session-boundary.js", () => ({ +vi.mock("./attempt-session-prepare.js", () => ({ + prepareEmbeddedAttemptAgentSession: mocks.prepareAgentSession, prepareEmbeddedAttemptSessionBoundary: mocks.prepareSessionBoundary, -})); -vi.mock("./attempt-session-manager-prepare.js", () => ({ prepareEmbeddedAttemptSessionManager: mocks.prepareSessionManager, })); vi.mock("./attempt-session-settle.js", () => ({ createEmbeddedAttemptSessionSettleTracker: mocks.createSessionSettleTracker, })); -vi.mock("./attempt-session.js", () => ({ - prepareEmbeddedAttemptAgentSession: mocks.prepareAgentSession, -})); vi.mock("./attempt-stream-settle.js", () => ({ prepareEmbeddedAttemptTransport: mocks.prepareTransport, })); diff --git a/src/agents/embedded-agent-runner/run/attempt-session-runtime-prepare.ts b/src/agents/embedded-agent-runner/run/attempt-session-runtime-prepare.ts index b757602f90d6..183471bbf559 100644 --- a/src/agents/embedded-agent-runner/run/attempt-session-runtime-prepare.ts +++ b/src/agents/embedded-agent-runner/run/attempt-session-runtime-prepare.ts @@ -7,10 +7,12 @@ import { getProviderPromptState } from "../provider-prompt-state.js"; import { getEmbeddedSessionPromptState } from "../session-prompt-state.js"; import type { createEmbeddedAttemptExternalAbortController } from "./attempt-abort.js"; import { installEmbeddedAttemptContextGuards } from "./attempt-context-guards.js"; -import { prepareEmbeddedAttemptSessionBoundary } from "./attempt-session-boundary.js"; -import { prepareEmbeddedAttemptSessionManager } from "./attempt-session-manager-prepare.js"; +import { + prepareEmbeddedAttemptAgentSession, + prepareEmbeddedAttemptSessionBoundary, + prepareEmbeddedAttemptSessionManager, +} from "./attempt-session-prepare.js"; import { createEmbeddedAttemptSessionSettleTracker } from "./attempt-session-settle.js"; -import { prepareEmbeddedAttemptAgentSession } from "./attempt-session.js"; import { prepareEmbeddedAttemptTransport } from "./attempt-stream-settle.js"; import { prepareEmbeddedAttemptTrajectory } from "./attempt-trajectory.js"; import type { EmbeddedRunAttemptParams, EmbeddedRunAttemptResult } from "./types.js"; diff --git a/src/agents/embedded-agent-runner/run/attempt-session-settle.ts b/src/agents/embedded-agent-runner/run/attempt-session-settle.ts index d47534a078d1..8ba5149eb0cb 100644 --- a/src/agents/embedded-agent-runner/run/attempt-session-settle.ts +++ b/src/agents/embedded-agent-runner/run/attempt-session-settle.ts @@ -1,5 +1,21 @@ -/** Tracks native prompt and abort settlement through attempt cleanup. */ +/** + * Tracks prompt and abort settlement, then finalizes session-owned resources. + * It may assume the active session and transcript lifecycle are established. + */ +import { formatErrorMessage, toErrorObject } from "../../../infra/errors.js"; +import type { createTrajectoryRuntimeRecorder } from "../../../trajectory/runtime.js"; +import type { guardSessionManager } from "../../session-tool-result-guard-wrapper.js"; import type { AgentSession } from "../../sessions/index.js"; +import { clearToolSearchCatalog, type ToolSearchCatalogRef } from "../../tool-search.js"; +import { log } from "../logger.js"; +import { flushPendingToolResultsAfterIdle } from "../wait-for-idle-before-flush.js"; +import type { EmitDiagnosticRunCompleted } from "./attempt-startup.js"; +import { flushEmbeddedAttemptTrajectoryRecorder } from "./attempt-trajectory-flush-cleanup.js"; +import type { createEmbeddedAttemptTranscriptLifecycle } from "./attempt-transcript-lifecycle.js"; +import { cleanupEmbeddedAttemptResources } from "./attempt.subscription-cleanup.js"; +import type { EmbeddedRunAttemptParams } from "./types.js"; + +/** Tracks native prompt and abort settlement through attempt cleanup. */ export function createEmbeddedAttemptSessionSettleTracker( activeSession: Pick, @@ -41,3 +57,145 @@ export function createEmbeddedAttemptSessionSettleTracker( trackPromptSettlePromise, }; } + +/** + * Finalizes trajectory and session-owned resources for one embedded attempt. + */ + +type AttemptTranscriptLifecycle = ReturnType; +type TrajectoryRecorder = ReturnType; +type DisposableRuntime = { dispose(): Promise | void }; + +type CleanupEmbeddedAttemptSessionInput = { + attempt: EmbeddedRunAttemptParams; + session?: AgentSession; + sessionManager?: ReturnType; + transcriptLifecycle: AttemptTranscriptLifecycle; + bundleMcpRuntime?: DisposableRuntime; + bundleLspRuntime?: DisposableRuntime; + removeToolResultContextGuard?: () => void; + toolSearchCatalogRef?: ToolSearchCatalogRef; + sandboxSessionKey?: string; + sessionAgentId: string; + buildAbortSettlePromise: () => Promise | null; + trajectoryRecorder: TrajectoryRecorder | null; + trajectoryEndRecorded: boolean; + cleanupYieldAborted: boolean; + emitDiagnosticRunCompleted?: EmitDiagnosticRunCompleted; + readState: () => { + aborted: boolean; + externalAbort: boolean; + timedOut: boolean; + idleTimedOut: boolean; + timedOutDuringCompaction: boolean; + timedOutDuringToolExecution: boolean; + timedOutByRunBudget: boolean; + promptError: unknown; + beforeAgentRunBlocked: boolean; + beforeAgentRunBlockedBy?: string; + }; +}; + +export async function cleanupEmbeddedAttemptSessionPhase( + input: CleanupEmbeddedAttemptSessionInput, +): Promise { + const { attempt } = input; + const initialState = input.readState(); + if (input.trajectoryRecorder && !input.trajectoryEndRecorded) { + input.trajectoryRecorder.recordEvent("session.ended", { + status: initialState.promptError + ? "error" + : initialState.aborted || initialState.timedOut + ? "interrupted" + : "cleanup", + aborted: initialState.aborted, + externalAbort: initialState.externalAbort, + timedOut: initialState.timedOut, + idleTimedOut: initialState.idleTimedOut, + timedOutDuringCompaction: initialState.timedOutDuringCompaction, + timedOutDuringToolExecution: initialState.timedOutDuringToolExecution, + timedOutByRunBudget: initialState.timedOutByRunBudget, + promptError: initialState.promptError + ? formatErrorMessage(initialState.promptError) + : undefined, + }); + } + await flushEmbeddedAttemptTrajectoryRecorder({ + runId: attempt.runId, + sessionId: attempt.sessionId, + log, + trajectoryRecorder: input.trajectoryRecorder, + }); + + // Agent retries can report idle before retried tools finish; waiting before + // the flush prevents synthetic missing-tool results (#8643). Teardown keeps + // lock release ahead of runtime disposal so the next attempt can recover. + let cleanupError: unknown; + try { + clearToolSearchCatalog({ + sessionId: attempt.sessionId, + sessionKey: input.sandboxSessionKey, + agentId: input.sessionAgentId, + runId: attempt.runId, + catalogRef: input.toolSearchCatalogRef, + }); + // Abort handling remains armed during cleanup, so reread after trajectory + // flushing instead of using the state captured at helper entry. + const cleanupState = input.readState(); + const cleanupAborted = + Boolean(attempt.abortSignal?.aborted) || + cleanupState.aborted || + cleanupState.timedOut || + cleanupState.idleTimedOut || + cleanupState.timedOutDuringCompaction; + const cleanupAbortLike = cleanupAborted || input.cleanupYieldAborted; + await input.transcriptLifecycle.beginCleanup(); + await cleanupEmbeddedAttemptResources({ + removeToolResultContextGuard: input.removeToolResultContextGuard, + flushPendingToolResultsAfterIdle, + session: input.session, + sessionManager: input.sessionManager, + bundleMcpRuntime: input.bundleMcpRuntime, + bundleLspRuntime: input.bundleLspRuntime, + // Aborted runs skip the idle wait so teardown cannot strand the lock. + aborted: cleanupAbortLike, + abortSettlePromise: cleanupAborted ? input.buildAbortSettlePromise() : null, + runId: attempt.runId, + sessionId: attempt.sessionId, + }); + } catch (err) { + cleanupError = err; + } finally { + try { + await input.transcriptLifecycle.dispose(); + } catch (err) { + cleanupError ??= err; + } + } + + const finalState = input.readState(); + const cleanupFailure = cleanupError; + input.emitDiagnosticRunCompleted?.( + cleanupFailure + ? "error" + : finalState.beforeAgentRunBlocked + ? "blocked" + : finalState.promptError + ? "error" + : finalState.aborted || + finalState.timedOut || + finalState.idleTimedOut || + finalState.timedOutDuringCompaction + ? "aborted" + : "completed", + cleanupFailure ?? finalState.promptError, + finalState.beforeAgentRunBlocked + ? { blockedBy: finalState.beforeAgentRunBlockedBy ?? "before_agent_run" } + : undefined, + ); + + if (!cleanupFailure) { + return; + } + await Promise.reject(toErrorObject(cleanupFailure, "Non-Error rejection")); +} diff --git a/src/agents/embedded-agent-runner/run/attempt-session.test.ts b/src/agents/embedded-agent-runner/run/attempt-session.test.ts index b78356fc3fbe..4bd35d650e36 100644 --- a/src/agents/embedded-agent-runner/run/attempt-session.test.ts +++ b/src/agents/embedded-agent-runner/run/attempt-session.test.ts @@ -69,7 +69,7 @@ vi.mock("./tool-activity-heartbeat.js", () => ({ notifyToolActivity: hoisted.notifyToolActivity, })); -import { prepareEmbeddedAttemptAgentSession } from "./attempt-session.js"; +import { prepareEmbeddedAttemptAgentSession } from "./attempt-session-prepare.js"; const attempt = { authStorage: { id: "auth" }, diff --git a/src/agents/embedded-agent-runner/run/attempt-session.ts b/src/agents/embedded-agent-runner/run/attempt-session.ts deleted file mode 100644 index dcb07d078006..000000000000 --- a/src/agents/embedded-agent-runner/run/attempt-session.ts +++ /dev/null @@ -1,216 +0,0 @@ -/** - * Prepares embedded-agent resources, tools, and active sessions. - */ -import { getGlobalHookRunner } from "../../../plugins/hook-runner-global.js"; -import type { PluginMetadataSnapshot } from "../../../plugins/plugin-metadata-snapshot.types.js"; -import { createPreparedEmbeddedAgentSettingsManager } from "../../agent-project-settings.js"; -import { - applyAgentAutoCompactionGuard, - applyAgentCompactionSettingsFromConfig, - isSilentOverflowProneModel, - resolveEffectiveCompactionMode, -} from "../../agent-settings.js"; -import { toToolDefinitions } from "../../agent-tool-definition-adapter.js"; -import type { guardSessionManager } from "../../session-tool-result-guard-wrapper.js"; -import type { AgentSession, CreateAgentSessionOptions } from "../../sessions/index.js"; -import { createAgentSessionForEmbeddedRunner } from "../../sessions/sdk.js"; -import { wrapToolDefinition } from "../../sessions/tools/tool-definition-wrapper.js"; -import { resolveToolSearchCatalogTool } from "../../tool-search.js"; -import { buildEmbeddedExtensionFactories } from "../extensions.js"; -import { log } from "../logger.js"; -import { createEmbeddedAgentResourceLoader } from "../resource-loader.js"; -import { applySystemPromptToSession } from "../system-prompt.js"; -import { prepareEmbeddedAttemptClientTools } from "./attempt-client-tools.js"; -import type { EmbeddedAttemptTranscriptLifecycle } from "./attempt-transcript-lifecycle.js"; -import type { AttemptContextEngine } from "./attempt.context-engine-helpers.js"; -import { installCodeModeRepairHook } from "./code-mode-repair.js"; -import { installMessageToolOnlyTerminalHook } from "./message-tool-terminal.js"; -import { notifyToolActivity } from "./tool-activity-heartbeat.js"; -import { - createToolLoopBatchAdmission, - installToolLoopRecoveryCleanup, -} from "./tool-loop-recovery.js"; -import type { EmbeddedRunAttemptParams } from "./types.js"; - -type ClientToolPreparation = Omit< - Parameters[0], - "attempt" ->; - -type AttemptSessionManager = ReturnType; - -/** Prepares resource loading, client tools, and the active agent session. */ -export async function prepareEmbeddedAttemptAgentSession(input: { - attempt: EmbeddedRunAttemptParams; - activeContextEngineInfo?: AttemptContextEngine["info"]; - agentCoreThinkingLevel: CreateAgentSessionOptions["thinkingLevel"]; - agentDir: string; - clientToolPreparation: ClientToolPreparation; - effectiveCwd: string; - getCurrentAttemptPluginMetadataSnapshot: () => PluginMetadataSnapshot | undefined; - initialSystemPrompt: string; - markStage: (stage: string) => void; - onSessionCreated: (session: AgentSession) => void; - onSystemPromptChanged: (systemPrompt: string) => void; - runAbortSignal: AbortSignal; - sessionAgentId: string; - transcriptLifecycle: EmbeddedAttemptTranscriptLifecycle; - sessionManager: AttemptSessionManager; -}) { - const { attempt } = input; - const settingsManager = createPreparedEmbeddedAgentSettingsManager({ - cwd: input.effectiveCwd, - agentDir: input.agentDir, - cfg: attempt.config, - pluginMetadataSnapshot: input.getCurrentAttemptPluginMetadataSnapshot(), - contextTokenBudget: attempt.contextTokenBudget, - }); - const autoCompactionGuardArgs = { - settingsManager, - contextEngineInfo: input.activeContextEngineInfo, - compactionMode: resolveEffectiveCompactionMode(attempt.config), - silentOverflowProneProvider: isSilentOverflowProneModel({ - provider: attempt.provider, - modelId: attempt.modelId, - baseUrl: attempt.model.baseUrl ?? undefined, - }), - }; - applyAgentAutoCompactionGuard(autoCompactionGuardArgs); - - // These factories carry compaction/pruning runtime state into the resource loader. - const extensionFactories = buildEmbeddedExtensionFactories({ - cfg: attempt.config, - sessionManager: input.sessionManager, - provider: attempt.provider, - modelId: attempt.modelId, - model: attempt.model, - contextTokenBudget: attempt.contextTokenBudget, - agentId: input.sessionAgentId, - sessionId: attempt.sessionId, - sessionKey: attempt.sessionKey ?? attempt.sandboxSessionKey, - runId: attempt.runId, - }); - const resourceLoader = createEmbeddedAgentResourceLoader({ - cwd: input.effectiveCwd, - agentDir: input.agentDir, - settingsManager, - extensionFactories, - }); - await resourceLoader.reload(); - // reload() rehydrates disk settings. Reapply OpenClaw's context budget and - // auto-compaction guards before the session can submit a prompt (#75799). - applyAgentCompactionSettingsFromConfig({ - settingsManager, - cfg: attempt.config, - contextTokenBudget: attempt.contextTokenBudget, - }); - applyAgentAutoCompactionGuard(autoCompactionGuardArgs); - input.markStage("session-resource-loader"); - - // Tool creation needs the same runner later used by lifecycle hooks. - const hookRunner = getGlobalHookRunner(); - const preparedClientTools = prepareEmbeddedAttemptClientTools({ - attempt, - ...input.clientToolPreparation, - }); - const { allCustomTools, sessionToolAllowlist, ...clientToolRuntime } = preparedClientTools; - - const sessionOptions: CreateAgentSessionOptions = { - cwd: input.effectiveCwd, - agentDir: input.agentDir, - authStorage: attempt.authStorage, - modelRegistry: attempt.modelRegistry, - model: attempt.model, - thinkingLevel: input.agentCoreThinkingLevel, - tools: sessionToolAllowlist, - customTools: allCustomTools, - sessionManager: input.sessionManager, - settingsManager, - resourceLoader, - resolveDeferredTool: input.clientToolPreparation.deferredDirectoryToolsCallable - ? ({ toolCall }) => { - const tool = resolveToolSearchCatalogTool( - { - config: attempt.config, - runtimeConfig: attempt.config, - agentId: input.sessionAgentId, - sessionKey: input.clientToolPreparation.sandboxSessionKey, - sessionId: attempt.sessionId, - runId: attempt.runId, - catalogRef: input.clientToolPreparation.toolSearchCatalogRef, - abortSignal: input.runAbortSignal, - }, - toolCall.name, - ); - // Catalog entries already own before_tool_call wrapping. - const definition = tool - ? toToolDefinitions([tool], input.clientToolPreparation.catalogToolHookContext)[0] - : undefined; - const hydratedTool = definition ? wrapToolDefinition(definition) : undefined; - if (hydratedTool) { - log.info(`tool-search: hydrated deferred directory tool ${toolCall.name}`); - const originalExecute = hydratedTool.execute; - hydratedTool.execute = (async (...args: Parameters) => { - const interval = setInterval(() => notifyToolActivity(attempt.runId), 60_000); - interval.unref?.(); - try { - notifyToolActivity(attempt.runId); - return await originalExecute(...args); - } finally { - clearInterval(interval); - notifyToolActivity(attempt.runId); - } - }) as typeof originalExecute; - } - return hydratedTool; - } - : undefined, - withSessionWriteSettlement: (operation) => - input.transcriptLifecycle.withTranscriptWrite(operation), - }; - const createdSession = await createAgentSessionForEmbeddedRunner(sessionOptions, { - // Without a resolved model budget, the outer loop cannot own bounded recovery. - contextOverflowRecoveryOwner: attempt.contextTokenBudget === undefined ? "session" : "caller", - beforeToolBatch: input.clientToolPreparation.catalogToolHookContext - ? createToolLoopBatchAdmission(input.clientToolPreparation.catalogToolHookContext) - : undefined, - }); - const activeSession = createdSession.session; - if (!activeSession) { - throw new Error("Embedded agent session missing"); - } - // Publish ownership before post-construction hooks. Outer cleanup must dispose - // the session if tool activation or terminal-hook installation fails. - input.onSessionCreated(activeSession); - installToolLoopRecoveryCleanup({ agent: activeSession.agent, runId: attempt.runId }); - activeSession.setActiveToolsByName(sessionToolAllowlist); - const setActiveSessionSystemPrompt = (nextSystemPrompt: string) => { - input.onSystemPromptChanged(nextSystemPrompt); - applySystemPromptToSession(activeSession, nextSystemPrompt); - }; - setActiveSessionSystemPrompt(input.initialSystemPrompt); - let didDeliverSourceReplyViaMessageTool = false; - const markSourceReplyDelivered = () => { - didDeliverSourceReplyViaMessageTool = true; - }; - installMessageToolOnlyTerminalHook({ - agent: activeSession.agent, - sourceReplyDeliveryMode: attempt.sourceReplyDeliveryMode, - onDeliveredSourceReply: markSourceReplyDelivered, - }); - if (input.clientToolPreparation.codeModeControlsEnabledForRun) { - installCodeModeRepairHook({ agent: activeSession.agent }); - } - input.markStage("agent-session"); - - return { - activeSession, - allCustomTools, - ...clientToolRuntime, - hasDeliveredSourceReply: () => didDeliverSourceReplyViaMessageTool, - hookRunner, - markSourceReplyDelivered, - setActiveSessionSystemPrompt, - settingsManager, - }; -} diff --git a/src/agents/embedded-agent-runner/run/attempt-tool-catalog.ts b/src/agents/embedded-agent-runner/run/attempt-tool-catalog.ts index a8a588514577..6ceed8b24f71 100644 --- a/src/agents/embedded-agent-runner/run/attempt-tool-catalog.ts +++ b/src/agents/embedded-agent-runner/run/attempt-tool-catalog.ts @@ -23,7 +23,7 @@ import { applyAgentToolSurfaceCatalog } from "../../tool-surface-plan.js"; import { log } from "../logger.js"; import type { prepareEmbeddedAttemptBundleTools } from "./attempt-bundle-tools.js"; import { collectAttemptExplicitToolAllowlistSources } from "./attempt-tool-allowlist.js"; -import type { prepareEmbeddedAttemptToolBase } from "./attempt-tool-base-prepare.js"; +import type { prepareEmbeddedAttemptToolBase } from "./attempt-tool-prepare.js"; import { buildToolSearchRunPlan } from "./attempt.tool-search-run-plan.js"; import { wrapEmbeddedAttemptToolWithActivity } from "./tool-activity-heartbeat.js"; import type { EmbeddedRunAttemptParams } from "./types.js"; diff --git a/src/agents/embedded-agent-runner/run/attempt-tool-base-prepare.ts b/src/agents/embedded-agent-runner/run/attempt-tool-prepare.ts similarity index 99% rename from src/agents/embedded-agent-runner/run/attempt-tool-base-prepare.ts rename to src/agents/embedded-agent-runner/run/attempt-tool-prepare.ts index 85732dec047e..cedad025c299 100644 --- a/src/agents/embedded-agent-runner/run/attempt-tool-base-prepare.ts +++ b/src/agents/embedded-agent-runner/run/attempt-tool-prepare.ts @@ -1,3 +1,7 @@ +/** + * Prepares the core tool surface for one embedded attempt. + * It may assume workspace, model, and runtime policy inputs are resolved. + */ import { messageToolOwnsVisibleReply } from "../../../auto-reply/source-reply-delivery-mode.js"; import type { DiagnosticTraceContext } from "../../../infra/diagnostic-trace-context.js"; import { extractModelCompat } from "../../../plugins/provider-model-compat.js"; diff --git a/src/agents/embedded-agent-runner/run/attempt.ts b/src/agents/embedded-agent-runner/run/attempt.ts index 08b2fc3c8e1e..75a4602e07ec 100644 --- a/src/agents/embedded-agent-runner/run/attempt.ts +++ b/src/agents/embedded-agent-runner/run/attempt.ts @@ -30,8 +30,8 @@ import { prepareEmbeddedAttemptBootstrap } from "./attempt-bootstrap-prepare.js" import { prepareEmbeddedAttemptBundleTools } from "./attempt-bundle-tools.js"; import { runEmbeddedAttemptExecutionPhase } from "./attempt-execution-phase.js"; import type { EmbeddedAttemptExecutionState } from "./attempt-execution-types.js"; -import { cleanupEmbeddedAttemptSessionPhase } from "./attempt-session-cleanup.js"; import { prepareEmbeddedAttemptSessionRuntime } from "./attempt-session-runtime-prepare.js"; +import { cleanupEmbeddedAttemptSessionPhase } from "./attempt-session-settle.js"; import { prepareEmbeddedAttemptSetup } from "./attempt-setup.js"; import { createEmbeddedRunStageTracker } from "./attempt-stage-timing.js"; import { @@ -40,8 +40,8 @@ import { type EmitDiagnosticRunCompleted, } from "./attempt-startup.js"; import { prepareEmbeddedAttemptSystemPrompt } from "./attempt-system-prompt-prepare.js"; -import { prepareEmbeddedAttemptToolBase } from "./attempt-tool-base-prepare.js"; import { prepareEmbeddedAttemptToolCatalog } from "./attempt-tool-catalog.js"; +import { prepareEmbeddedAttemptToolBase } from "./attempt-tool-prepare.js"; import { prepareEmbeddedAttemptTranscriptLifecycle } from "./attempt-transcript-lifecycle-prepare.js"; import { queueSessionsYieldInterruptMessage,