mirror of
https://github.com/openclaw/openclaw.git
synced 2026-08-23 02:45:38 -06:00
refactor(agents): consolidate session and tool attempt steps (#121344)
* refactor(agents): consolidate session-phase attempt steps * refactor(agents): consolidate tool-phase attempt steps * refactor(agents): repoint consolidated tool imports * refactor(agents): remove absorbed session and tool steps * test(agents): preserve merged session mock exports * refactor(agents): preserve attempt phase leaf contracts
This commit is contained in:
Peter Steinberger
committed by
GitHub
GitHub
parent
8430fc0e3b
commit
79eb5bde43
@@ -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<ReturnType<typeof prepareEmbeddedAttemptSetup>>;
|
||||
|
||||
@@ -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";
|
||||
|
||||
|
||||
@@ -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();
|
||||
|
||||
@@ -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<Parameters<typeof normalizeMessagesForLlmBoundary>[1]>;
|
||||
|
||||
type CurrentUserTimestampOverride = NonNullable<LlmBoundaryOptions["currentUserTimestampOverride"]>;
|
||||
|
||||
export function prepareEmbeddedAttemptSessionBoundary(input: {
|
||||
activeSession: Pick<AgentSession, "agent">;
|
||||
attempt: SessionBoundaryAttempt;
|
||||
getUserTranscriptContexts: () => LlmBoundaryOptions["userTranscriptContexts"];
|
||||
isRawModelRun: boolean;
|
||||
preparedUserTurnMessage: AgentMessage | undefined;
|
||||
sessionManager: ReturnType<typeof guardSessionManager>;
|
||||
setActiveSessionSystemPrompt: (systemPrompt: string) => void;
|
||||
}): {
|
||||
boundaryTimezone: string | undefined;
|
||||
includeBoundaryTimestamp: boolean;
|
||||
orphanRepair: ReturnType<typeof resolveOrphanRepairPlan>;
|
||||
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;
|
||||
},
|
||||
};
|
||||
}
|
||||
@@ -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",
|
||||
|
||||
@@ -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<typeof createEmbeddedAttemptTranscriptLifecycle>;
|
||||
type TrajectoryRecorder = ReturnType<typeof createTrajectoryRuntimeRecorder>;
|
||||
type DisposableRuntime = { dispose(): Promise<void> | void };
|
||||
|
||||
type CleanupEmbeddedAttemptSessionInput = {
|
||||
attempt: EmbeddedRunAttemptParams;
|
||||
session?: AgentSession;
|
||||
sessionManager?: ReturnType<typeof guardSessionManager>;
|
||||
transcriptLifecycle: AttemptTranscriptLifecycle;
|
||||
bundleMcpRuntime?: DisposableRuntime;
|
||||
bundleLspRuntime?: DisposableRuntime;
|
||||
removeToolResultContextGuard?: () => void;
|
||||
toolSearchCatalogRef?: ToolSearchCatalogRef;
|
||||
sandboxSessionKey?: string;
|
||||
sessionAgentId: string;
|
||||
buildAbortSettlePromise: () => Promise<void> | 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<void> {
|
||||
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"));
|
||||
}
|
||||
@@ -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<typeof guardSessionManager>;
|
||||
type WithOwnedTranscriptWrite = <T>(operation: () => Promise<T> | T) => Promise<T>;
|
||||
export async function prepareEmbeddedAttemptSessionManager(input: {
|
||||
attempt: EmbeddedRunAttemptParams;
|
||||
activeContextEngine?: AttemptContextEngine;
|
||||
agentDir: string;
|
||||
effectiveCwd: string;
|
||||
effectiveWorkspace: string;
|
||||
onSessionManagerCreated: (sessionManager: AttemptSessionManager) => void;
|
||||
replayAllowedToolNames: ReadonlySet<string>;
|
||||
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,
|
||||
};
|
||||
}
|
||||
@@ -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<typeof prepareEmbeddedAttemptClientTools>[0],
|
||||
"attempt"
|
||||
>;
|
||||
|
||||
type AttemptSessionManager = ReturnType<typeof guardSessionManager>;
|
||||
|
||||
/** 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<typeof originalExecute>) => {
|
||||
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<Parameters<typeof normalizeMessagesForLlmBoundary>[1]>;
|
||||
|
||||
type CurrentUserTimestampOverride = NonNullable<LlmBoundaryOptions["currentUserTimestampOverride"]>;
|
||||
|
||||
export function prepareEmbeddedAttemptSessionBoundary(input: {
|
||||
activeSession: Pick<AgentSession, "agent">;
|
||||
attempt: SessionBoundaryAttempt;
|
||||
getUserTranscriptContexts: () => LlmBoundaryOptions["userTranscriptContexts"];
|
||||
isRawModelRun: boolean;
|
||||
preparedUserTurnMessage: AgentMessage | undefined;
|
||||
sessionManager: ReturnType<typeof guardSessionManager>;
|
||||
setActiveSessionSystemPrompt: (systemPrompt: string) => void;
|
||||
}): {
|
||||
boundaryTimezone: string | undefined;
|
||||
includeBoundaryTimestamp: boolean;
|
||||
orphanRepair: ReturnType<typeof resolveOrphanRepairPlan>;
|
||||
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 = <T>(operation: () => Promise<T> | T) => Promise<T>;
|
||||
export async function prepareEmbeddedAttemptSessionManager(input: {
|
||||
attempt: EmbeddedRunAttemptParams;
|
||||
activeContextEngine?: AttemptContextEngine;
|
||||
agentDir: string;
|
||||
effectiveCwd: string;
|
||||
effectiveWorkspace: string;
|
||||
onSessionManagerCreated: (sessionManager: AttemptSessionManager) => void;
|
||||
replayAllowedToolNames: ReadonlySet<string>;
|
||||
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,
|
||||
};
|
||||
}
|
||||
@@ -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,
|
||||
}));
|
||||
|
||||
@@ -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";
|
||||
|
||||
@@ -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<AgentSession, "abort">,
|
||||
@@ -41,3 +57,145 @@ export function createEmbeddedAttemptSessionSettleTracker(
|
||||
trackPromptSettlePromise,
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* Finalizes trajectory and session-owned resources for one embedded attempt.
|
||||
*/
|
||||
|
||||
type AttemptTranscriptLifecycle = ReturnType<typeof createEmbeddedAttemptTranscriptLifecycle>;
|
||||
type TrajectoryRecorder = ReturnType<typeof createTrajectoryRuntimeRecorder>;
|
||||
type DisposableRuntime = { dispose(): Promise<void> | void };
|
||||
|
||||
type CleanupEmbeddedAttemptSessionInput = {
|
||||
attempt: EmbeddedRunAttemptParams;
|
||||
session?: AgentSession;
|
||||
sessionManager?: ReturnType<typeof guardSessionManager>;
|
||||
transcriptLifecycle: AttemptTranscriptLifecycle;
|
||||
bundleMcpRuntime?: DisposableRuntime;
|
||||
bundleLspRuntime?: DisposableRuntime;
|
||||
removeToolResultContextGuard?: () => void;
|
||||
toolSearchCatalogRef?: ToolSearchCatalogRef;
|
||||
sandboxSessionKey?: string;
|
||||
sessionAgentId: string;
|
||||
buildAbortSettlePromise: () => Promise<void> | 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<void> {
|
||||
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"));
|
||||
}
|
||||
|
||||
@@ -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" },
|
||||
|
||||
@@ -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<typeof prepareEmbeddedAttemptClientTools>[0],
|
||||
"attempt"
|
||||
>;
|
||||
|
||||
type AttemptSessionManager = ReturnType<typeof guardSessionManager>;
|
||||
|
||||
/** 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<typeof originalExecute>) => {
|
||||
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,
|
||||
};
|
||||
}
|
||||
@@ -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";
|
||||
|
||||
+4
@@ -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";
|
||||
@@ -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,
|
||||
|
||||
Reference in New Issue
Block a user