From 0dfa22c6e0fc7edfc1effe9190ccade84a8996cd Mon Sep 17 00:00:00 2001 From: Josh Lehman Date: Tue, 23 Jun 2026 10:08:29 -0700 Subject: [PATCH] refactor: add embedded run session target seam (#90439) --- .../embedded-agent-runner/compact.queued.ts | 12 +- .../compact.runtime.types.ts | 4 +- src/agents/embedded-agent-runner/compact.ts | 33 +- .../embedded-agent-runner/compact.types.ts | 11 + src/agents/embedded-agent-runner/run.ts | 31 +- .../embedded-agent-runner/run/params.ts | 6 +- src/agents/embedded-agent-runner/run/types.ts | 3 + ...s.subagents.sessions-spawn.test-harness.ts | 27 +- src/agents/run-session-target.test.ts | 59 ++++ src/agents/run-session-target.ts | 79 +++++ src/agents/subagent-spawn.context.test.ts | 78 ++++- src/agents/subagent-spawn.runtime.ts | 1 + src/agents/subagent-spawn.test-helpers.ts | 31 ++ src/agents/subagent-spawn.ts | 92 +++--- src/auto-reply/reply/session-fork.test.ts | 87 +++++ src/auto-reply/reply/session-fork.ts | 247 +++++++++++++- src/auto-reply/reply/session.test.ts | 91 ++++++ src/auto-reply/reply/session.ts | 56 ++-- src/gateway/server-methods/sessions.ts | 188 ++++------- .../session-compaction-checkpoints.test.ts | 79 ++++- src/gateway/session-compaction-checkpoints.ts | 302 +++++++++++++++++- src/talk/agent-consult-runtime.test.ts | 141 +++++++- src/talk/agent-consult-runtime.ts | 89 +++--- 23 files changed, 1409 insertions(+), 338 deletions(-) create mode 100644 src/agents/run-session-target.test.ts create mode 100644 src/agents/run-session-target.ts create mode 100644 src/auto-reply/reply/session-fork.test.ts diff --git a/src/agents/embedded-agent-runner/compact.queued.ts b/src/agents/embedded-agent-runner/compact.queued.ts index 79e2b5c45a38..3393fb8228e0 100644 --- a/src/agents/embedded-agent-runner/compact.queued.ts +++ b/src/agents/embedded-agent-runner/compact.queued.ts @@ -14,9 +14,7 @@ import type { ContextEngineRuntimeSettings, } from "../../context-engine/types.js"; import { - captureCompactionCheckpointSnapshotAsync, - cleanupCompactionCheckpointSnapshot, - persistSessionCompactionCheckpoint, + createFileBackedCompactionCheckpointStore, readSessionLeafStateFromTranscriptAsync, resolveCompactionCheckpointTranscriptPosition, resolveSessionCompactionCheckpointReason, @@ -63,6 +61,8 @@ import { resolveModelAsync } from "./model.js"; import type { EmbeddedAgentCompactResult } from "./types.js"; import { normalizeContextTokenBudget } from "./utils.js"; +const compactionCheckpointStore = createFileBackedCompactionCheckpointStore(); + function shouldFallbackAfterHarnessCompaction( result: EmbeddedAgentCompactResult | undefined, ): boolean { @@ -352,7 +352,7 @@ export async function compactEmbeddedAgentSession( // are notified regardless of which engine is active. const engineOwnsCompaction = contextEngine.info.ownsCompaction === true; checkpointSnapshot = engineOwnsCompaction - ? await captureCompactionCheckpointSnapshotAsync({ + ? await compactionCheckpointStore.captureSnapshot({ sessionFile: params.sessionFile, }) : null; @@ -478,7 +478,7 @@ export async function compactEmbeddedAgentSession( preferredLeafId: postCompactionLeafId, transcriptState, }); - const storedCheckpoint = await persistSessionCompactionCheckpoint({ + const storedCheckpoint = await compactionCheckpointStore.persistCheckpoint({ cfg: params.config, sessionKey: params.sessionKey, sessionId: postCompactionSessionId, @@ -620,7 +620,7 @@ export async function compactEmbeddedAgentSession( }; } finally { if (!checkpointSnapshotRetained) { - await cleanupCompactionCheckpointSnapshot(checkpointSnapshot); + await compactionCheckpointStore.cleanupSnapshot(checkpointSnapshot); } await contextEngine.dispose?.(); } diff --git a/src/agents/embedded-agent-runner/compact.runtime.types.ts b/src/agents/embedded-agent-runner/compact.runtime.types.ts index 525728c7e3d8..d214eea9b9f5 100644 --- a/src/agents/embedded-agent-runner/compact.runtime.types.ts +++ b/src/agents/embedded-agent-runner/compact.runtime.types.ts @@ -1,12 +1,12 @@ /** * Types for the lazy embedded-agent compaction runtime boundary. */ -import type { CompactEmbeddedAgentSessionParams } from "./compact.types.js"; +import type { CompactEmbeddedAgentSessionRuntimeParams } from "./compact.types.js"; import type { EmbeddedAgentCompactResult } from "./types.js"; /** * Lazy-runtime signature for direct embedded session compaction. */ export type CompactEmbeddedAgentSessionDirect = ( - params: CompactEmbeddedAgentSessionParams, + params: CompactEmbeddedAgentSessionRuntimeParams, ) => Promise; diff --git a/src/agents/embedded-agent-runner/compact.ts b/src/agents/embedded-agent-runner/compact.ts index f96346facbcc..0ef6b1d73eec 100644 --- a/src/agents/embedded-agent-runner/compact.ts +++ b/src/agents/embedded-agent-runner/compact.ts @@ -8,9 +8,7 @@ import type { ThinkLevel } from "../../auto-reply/thinking.js"; import { resolveAgentModelFallbackValues } from "../../config/model-input.js"; import type { OpenClawConfig } from "../../config/types.openclaw.js"; import { - captureCompactionCheckpointSnapshotAsync, - cleanupCompactionCheckpointSnapshot, - persistSessionCompactionCheckpoint, + createFileBackedCompactionCheckpointStore, readSessionLeafStateFromTranscriptAsync, resolveCompactionCheckpointTranscriptPosition, resolveSessionCompactionCheckpointReason, @@ -107,6 +105,10 @@ import { wrapStreamFnTextTransforms } from "../plugin-text-transforms.js"; import { resolveAgentPromptSurfaceForSessionKey } from "../prompt-surface.js"; import { applyPreparedRuntimeAuthToModel } from "../provider-request-config.js"; import { registerProviderStreamForModel } from "../provider-stream.js"; +import { + applyAgentRunSessionTargetIdentity, + resolveAgentRunSessionTarget, +} from "../run-session-target.js"; import { collectRuntimeChannelCapabilities } from "../runtime-capabilities.js"; import { buildAgentRuntimePlan } from "../runtime-plan/build.js"; import type { AgentRuntimePlan } from "../runtime-plan/types.js"; @@ -135,6 +137,7 @@ import { } from "./compact-reasons.js"; import type { CompactEmbeddedAgentSessionParams, + CompactEmbeddedAgentSessionRuntimeParams, CompactionMessageMetrics, } from "./compact.types.js"; import { dedupeDuplicateUserMessagesForCompaction } from "./compaction-duplicate-user-messages.js"; @@ -192,6 +195,11 @@ import { mapThinkingLevel, normalizeContextTokenBudget } from "./utils.js"; import { flushPendingToolResultsAfterIdle } from "./wait-for-idle-before-flush.js"; export type { CompactEmbeddedAgentSessionParams } from "./compact.types.js"; +const compactionCheckpointStore = createFileBackedCompactionCheckpointStore(); +type CompactEmbeddedAgentSessionParamsWithSessionFile = CompactEmbeddedAgentSessionRuntimeParams & { + sessionFile: string; +}; + function hasRealConversationContent( msg: AgentMessage, messages: AgentMessage[], @@ -464,8 +472,17 @@ function fallbackFailureToCompactionResult(err: unknown): EmbeddedAgentCompactRe * Use this when already inside a session/global lane to avoid deadlocks. */ export async function compactEmbeddedAgentSessionDirect( - params: CompactEmbeddedAgentSessionParams, + paramsInput: CompactEmbeddedAgentSessionRuntimeParams, ): Promise { + const paramsBase = applyAgentRunSessionTargetIdentity(paramsInput); + const runSessionTarget = await resolveAgentRunSessionTarget(paramsBase); + const params: CompactEmbeddedAgentSessionParamsWithSessionFile = { + ...paramsBase, + agentId: paramsBase.agentId ?? runSessionTarget.agentId, + sessionId: runSessionTarget.sessionId, + sessionKey: paramsBase.sessionKey ?? runSessionTarget.sessionKey, + sessionFile: runSessionTarget.sessionFile, + }; if (hasExplicitCompactionModel(params) || !hasCompactionModelFallbackCandidates(params)) { return await compactEmbeddedAgentSessionDirectOnce(params); } @@ -530,7 +547,7 @@ export async function compactEmbeddedAgentSessionDirect( } async function compactEmbeddedAgentSessionDirectOnce( - params: CompactEmbeddedAgentSessionParams, + params: CompactEmbeddedAgentSessionParamsWithSessionFile, ): Promise { const startedAt = Date.now(); const diagId = params.diagId?.trim() || createCompactionDiagId(); @@ -1190,7 +1207,7 @@ async function compactEmbeddedAgentSessionDirectOnce( : undefined, allowedToolNames, }); - checkpointSnapshot = await captureCompactionCheckpointSnapshotAsync({ + checkpointSnapshot = await compactionCheckpointStore.captureSnapshot({ sessionManager, sessionFile: params.sessionFile, }); @@ -1546,7 +1563,7 @@ async function compactEmbeddedAgentSessionDirectOnce( preferredLeafId: activePostLeafId, transcriptState, }); - const storedCheckpoint = await persistSessionCompactionCheckpoint({ + const storedCheckpoint = await compactionCheckpointStore.persistCheckpoint({ cfg: params.config, sessionKey: params.sessionKey, sessionId: activeSessionId, @@ -1670,7 +1687,7 @@ async function compactEmbeddedAgentSessionDirectOnce( return fail(reason, err); } finally { if (!checkpointSnapshotRetained) { - await cleanupCompactionCheckpointSnapshot(checkpointSnapshot); + await compactionCheckpointStore.cleanupSnapshot(checkpointSnapshot); } restoreSkillEnv?.(); } diff --git a/src/agents/embedded-agent-runner/compact.types.ts b/src/agents/embedded-agent-runner/compact.types.ts index ce2a98955256..4805ed2e659c 100644 --- a/src/agents/embedded-agent-runner/compact.types.ts +++ b/src/agents/embedded-agent-runner/compact.types.ts @@ -9,12 +9,15 @@ import type { ContextEngine, ContextEngineRuntimeContext } from "../../context-e import type { CommandQueueEnqueueFn } from "../../process/command-queue.types.js"; import type { SkillSnapshot } from "../../skills/types.js"; import type { ExecElevatedDefaults, ExecToolDefaults } from "../bash-tools.exec-types.js"; +import type { AgentRunSessionTarget } from "../run-session-target.js"; import type { AgentRuntimePlan } from "../runtime-plan/types.js"; export type CompactEmbeddedAgentSessionParams = { sessionId: string; runId?: string; sessionKey?: string; + /** Storage-neutral transcript/session target. Defaults to sessionId/sessionKey/agentId. */ + sessionTarget?: AgentRunSessionTarget; /** Caller-resolved owner agent for global session aliases. */ agentId?: string; /** Session key used only for runtime policy/sandbox resolution. Defaults to sessionKey. */ @@ -106,6 +109,14 @@ export type CompactEmbeddedAgentSessionParams = { oneShotCliRun?: boolean; }; +export type CompactEmbeddedAgentSessionRuntimeParams = Omit< + CompactEmbeddedAgentSessionParams, + "sessionFile" +> & { + /** Deprecated file-backed artifact target. Prefer sessionTarget for new callers. */ + sessionFile?: string; +}; + export type CompactionMessageMetrics = { messages: number; historyTextChars: number; diff --git a/src/agents/embedded-agent-runner/run.ts b/src/agents/embedded-agent-runner/run.ts index 03842c7b58a4..1f78ebf7a5df 100644 --- a/src/agents/embedded-agent-runner/run.ts +++ b/src/agents/embedded-agent-runner/run.ts @@ -125,6 +125,10 @@ import { import { resolveProviderIdForAuth } from "../provider-auth-aliases.js"; import { hasOnlyAssistantReasoningContent } from "../replay-turn-classification.js"; import { runAgentCleanupStep } from "../run-cleanup-timeout.js"; +import { + applyAgentRunSessionTargetIdentity, + resolveAgentRunSessionTarget, +} from "../run-session-target.js"; import { buildAgentRuntimeAuthPlan } from "../runtime-plan/auth.js"; import { buildAgentRuntimePlan } from "../runtime-plan/build.js"; import { ensureRuntimePluginsLoaded } from "../runtime-plugins.js"; @@ -255,6 +259,7 @@ const BEFORE_AGENT_FINALIZE_RETRY_PROMPT_PREFIX = "Before accepting the previous final answer, apply this revision request and produce the revised final answer. Do not repeat completed work or rerun tools unless the request explicitly requires it."; const MAX_BEFORE_AGENT_FINALIZE_REVISIONS = 3; type EmbeddedRunAttemptForRunner = Awaited>; +type RunEmbeddedAgentParamsWithSessionFile = RunEmbeddedAgentParams & { sessionFile: string }; function isNoRealConversationCompactionNoop(params: { ok?: boolean; @@ -617,20 +622,28 @@ export function runEmbeddedAgent( async function runEmbeddedAgentInternal( paramsInput: RunEmbeddedAgentParams, ): Promise { - let params = paramsInput; - let lifecycleGeneration = params.lifecycleGeneration!; + const paramsBase = applyAgentRunSessionTargetIdentity(paramsInput); + let lifecycleGeneration = paramsBase.lifecycleGeneration!; const queuedLifecycleGeneration = getAgentEventLifecycleGeneration(); // Resolve sessionKey early so all downstream consumers (hooks, LCM, compaction) // receive a non-null key even when callers omit it. See #60552. const effectiveSessionKey = backfillSessionKey({ - config: params.config, - sessionId: params.sessionId, - sessionKey: params.sessionKey, - agentId: params.agentId, + config: paramsBase.config, + sessionId: paramsBase.sessionId, + sessionKey: paramsBase.sessionKey, + agentId: paramsBase.agentId, }); - if (effectiveSessionKey !== params.sessionKey) { - params = { ...params, sessionKey: effectiveSessionKey }; - } + const runSessionTarget = await resolveAgentRunSessionTarget({ + ...paramsBase, + sessionKey: effectiveSessionKey, + }); + let params: RunEmbeddedAgentParamsWithSessionFile = { + ...paramsBase, + agentId: paramsBase.agentId ?? runSessionTarget.agentId, + sessionId: runSessionTarget.sessionId, + sessionKey: effectiveSessionKey ?? runSessionTarget.sessionKey, + sessionFile: runSessionTarget.sessionFile, + }; const sessionLane = resolveSessionLane(params.sessionKey?.trim() || params.sessionId); const globalLane = resolveGlobalLane(params.lane); // Outer fallback attempts defer session suspension only while another diff --git a/src/agents/embedded-agent-runner/run/params.ts b/src/agents/embedded-agent-runner/run/params.ts index dc812261dd72..a4ed7e8e5751 100644 --- a/src/agents/embedded-agent-runner/run/params.ts +++ b/src/agents/embedded-agent-runner/run/params.ts @@ -29,6 +29,7 @@ import type { } from "../../embedded-agent-subscribe.shared-types.js"; import type { FastModeAutoProgressState } from "../../fast-mode.js"; import type { AgentInternalEvent } from "../../internal-events.js"; +import type { AgentRunSessionTarget } from "../../run-session-target.js"; import type { AgentMessage } from "../../runtime/index.js"; import type { SilentReplyPromptMode } from "../../system-prompt.types.js"; import type { PromptMode } from "../../system-prompt.types.js"; @@ -47,6 +48,8 @@ export type CurrentInboundPromptContext = { export type RunEmbeddedAgentParams = { sessionId: string; sessionKey?: string; + /** Storage-neutral transcript/session target. Defaults to sessionId/sessionKey/agentId. */ + sessionTarget?: AgentRunSessionTarget; /** Immutable gateway lifecycle ownership captured when this execution was admitted. */ lifecycleGeneration?: string; /** Provider prompt-cache affinity key; distinct from transcript/session identity. */ @@ -122,7 +125,8 @@ export type RunEmbeddedAgentParams = { forceHeartbeatTool?: boolean; /** Allow runtime plugins for this run to late-bind the gateway subagent. */ allowGatewaySubagentBinding?: boolean; - sessionFile: string; + /** @deprecated Use sessionTarget plus sessionId/sessionKey/agentId for runtime identity. */ + sessionFile?: string; workspaceDir: string; /** Task working directory for tool/runtime execution. Defaults to workspaceDir. */ cwd?: string; diff --git a/src/agents/embedded-agent-runner/run/types.ts b/src/agents/embedded-agent-runner/run/types.ts index 727bdf8ec045..a33af7e20408 100644 --- a/src/agents/embedded-agent-runner/run/types.ts +++ b/src/agents/embedded-agent-runner/run/types.ts @@ -40,6 +40,7 @@ type EmbeddedRunAttemptBase = Omit< | "fastMode" | "lane" | "enqueue" + | "sessionFile" >; export type EmbeddedRunContextWindowInfo = { @@ -51,6 +52,8 @@ export type EmbeddedRunContextWindowInfo = { export type EmbeddedRunFastModeParam = boolean | (() => boolean | undefined); export type EmbeddedRunAttemptParams = EmbeddedRunAttemptBase & { + /** Active file-backed artifact target resolved by the run/session target seam. */ + sessionFile: string; initialReplayState?: EmbeddedRunReplayState; /** Pluggable context engine for ingest/assemble/compact lifecycle. */ contextEngine?: ContextEngine; diff --git a/src/agents/openclaw-tools.subagents.sessions-spawn.test-harness.ts b/src/agents/openclaw-tools.subagents.sessions-spawn.test-harness.ts index 2087db03de1e..53b24a4862b9 100644 --- a/src/agents/openclaw-tools.subagents.sessions-spawn.test-harness.ts +++ b/src/agents/openclaw-tools.subagents.sessions-spawn.test-harness.ts @@ -206,13 +206,26 @@ export async function getSessionsSpawnTool(opts: CreateOpenClawToolsOpts) { compact: async () => ({ ok: true, compacted: false }), ingest: async () => ({ ingested: false }), }), - resolveParentForkDecision: async () => ({ - status: "fork", - maxTokens: 100_000, - }), - forkSessionFromParent: async () => ({ - sessionId: "forked-session-id", - sessionFile: "/tmp/forked-session.jsonl", + forkSessionEntryFromParent: async () => ({ + status: "forked", + fork: { + sessionId: "forked-session-id", + sessionFile: "/tmp/forked-session.jsonl", + }, + parentEntry: { + sessionId: "parent-session-id", + updatedAt: Date.now(), + }, + sessionEntry: { + sessionId: "forked-session-id", + sessionFile: "/tmp/forked-session.jsonl", + forkedFromParent: true, + updatedAt: Date.now(), + }, + decision: { + status: "fork", + maxTokens: 100_000, + }, }), updateSessionStore: async (_storePath, mutator) => mutator({}), }); diff --git a/src/agents/run-session-target.test.ts b/src/agents/run-session-target.test.ts new file mode 100644 index 000000000000..1aafee898034 --- /dev/null +++ b/src/agents/run-session-target.test.ts @@ -0,0 +1,59 @@ +import fs from "node:fs"; +import os from "node:os"; +import path from "node:path"; +import { afterEach, beforeEach, describe, expect, it } from "vitest"; +import { loadSessionStore } from "../config/sessions/store.js"; +import type { OpenClawConfig } from "../config/types.openclaw.js"; +import { resolveAgentRunSessionTarget } from "./run-session-target.js"; + +describe("agent run session target", () => { + let tempDir: string; + + beforeEach(() => { + tempDir = fs.mkdtempSync(path.join(os.tmpdir(), "openclaw-run-session-target-")); + }); + + afterEach(() => { + fs.rmSync(tempDir, { recursive: true, force: true }); + }); + + it("resolves runtime identity through the run config store", async () => { + const storePath = path.join(tempDir, "custom-sessions", "sessions.json"); + const sessionKey = "agent:helper:commitments:test-run"; + + const target = await resolveAgentRunSessionTarget({ + agentId: "helper", + config: { session: { store: storePath } } as OpenClawConfig, + sessionId: "test-run", + sessionKey, + }); + + expect(target).toMatchObject({ + agentId: "helper", + sessionId: "test-run", + sessionKey, + }); + expect(path.dirname(target.sessionFile)).toBe(path.dirname(storePath)); + expect(loadSessionStore(storePath, { skipCache: true })[sessionKey]?.sessionFile).toBe( + target.sessionFile, + ); + }); + + it("uses the agent from an agent-scoped session key when agentId is omitted", async () => { + const storeRoot = path.join(tempDir, "agents", "{agentId}", "sessions.json"); + const sessionKey = "agent:helper:main"; + + const target = await resolveAgentRunSessionTarget({ + config: { session: { store: storeRoot } } as OpenClawConfig, + sessionId: "helper-session", + sessionKey, + }); + + const helperStorePath = path.join(tempDir, "agents", "helper", "sessions.json"); + expect(target.agentId).toBe("helper"); + expect(path.dirname(target.sessionFile)).toBe(path.dirname(helperStorePath)); + expect(loadSessionStore(helperStorePath, { skipCache: true })[sessionKey]?.sessionFile).toBe( + target.sessionFile, + ); + }); +}); diff --git a/src/agents/run-session-target.ts b/src/agents/run-session-target.ts new file mode 100644 index 000000000000..dbc287851e05 --- /dev/null +++ b/src/agents/run-session-target.ts @@ -0,0 +1,79 @@ +import { normalizeOptionalString } from "@openclaw/normalization-core/string-coerce"; +import { resolveStorePath } from "../config/sessions/paths.js"; +import { + resolveSessionTranscriptRuntimeTarget, + type SessionTranscriptRuntimeTarget, +} from "../config/sessions/session-accessor.js"; +import type { OpenClawConfig } from "../config/types.openclaw.js"; +import { resolveAgentIdFromSessionKey } from "../routing/session-key.js"; + +/** Identifies a run transcript target without naming the current storage artifact. */ +export type AgentRunSessionTarget = { + agentId?: string; + sessionId?: string; + sessionKey?: string; + storePath?: string; + threadId?: string | number; +}; + +/** File-backed target resolved from the storage-neutral run identity. */ +export type ResolvedAgentRunSessionTarget = SessionTranscriptRuntimeTarget; + +/** Resolves the active file-backed target used by current run/session internals. */ +export async function resolveAgentRunSessionTarget(params: { + agentId?: string; + config?: OpenClawConfig; + sessionFile?: string; + sessionId: string; + sessionKey?: string; + sessionTarget?: AgentRunSessionTarget; +}): Promise { + const sessionTarget = params.sessionTarget; + const agentId = normalizeOptionalString(sessionTarget?.agentId) ?? params.agentId; + const sessionId = normalizeOptionalString(sessionTarget?.sessionId) ?? params.sessionId; + const sessionKey = normalizeOptionalString(sessionTarget?.sessionKey) ?? params.sessionKey; + const effectiveAgentId = agentId ?? resolveAgentIdFromSessionKey(sessionKey); + const sessionFile = normalizeOptionalString(params.sessionFile); + if (sessionFile) { + return { + agentId: effectiveAgentId ?? "", + sessionFile, + sessionId, + sessionKey: sessionKey ?? "", + }; + } + if (!sessionKey) { + throw new Error(`Cannot resolve run session target without a session key: ${sessionId}`); + } + const storePath = + normalizeOptionalString(sessionTarget?.storePath) ?? + resolveStorePath(params.config?.session?.store, { agentId: effectiveAgentId }); + return await resolveSessionTranscriptRuntimeTarget({ + ...(effectiveAgentId ? { agentId: effectiveAgentId } : {}), + sessionId, + sessionKey, + storePath, + ...(sessionTarget?.threadId !== undefined ? { threadId: sessionTarget.threadId } : {}), + }); +} + +/** Applies identity fields from the explicit target before legacy backfills run. */ +export function applyAgentRunSessionTargetIdentity< + T extends { + agentId?: string; + sessionId: string; + sessionKey?: string; + sessionTarget?: AgentRunSessionTarget; + }, +>(params: T): T { + const target = params.sessionTarget; + if (!target) { + return params; + } + return { + ...params, + agentId: normalizeOptionalString(target.agentId) ?? params.agentId, + sessionId: normalizeOptionalString(target.sessionId) ?? params.sessionId, + sessionKey: normalizeOptionalString(target.sessionKey) ?? params.sessionKey, + }; +} diff --git a/src/agents/subagent-spawn.context.test.ts b/src/agents/subagent-spawn.context.test.ts index 5ec8d20bc5fe..02c6e99ef01e 100644 --- a/src/agents/subagent-spawn.context.test.ts +++ b/src/agents/subagent-spawn.context.test.ts @@ -1,5 +1,3 @@ -// Subagent spawn context tests cover isolated, forked, lightweight, and -// thread-bound bootstrap context preparation for child sessions. import path from "node:path"; import { MAX_TIMER_TIMEOUT_MS } from "@openclaw/normalization-core/number-coercion"; import { beforeAll, beforeEach, describe, expect, it, vi } from "vitest"; @@ -14,7 +12,9 @@ type GatewayRequest = { method?: string; params?: Record }; describe("sessions_spawn context modes", () => { const storePath = "/tmp/subagent-context-session-store.json"; const callGatewayMock = vi.fn(); + const loadSessionStoreMock = vi.fn(); const updateSessionStoreMock = vi.fn(); + const forkSessionEntryFromParentMock = vi.fn(); const forkSessionFromParentMock = vi.fn(); const ensureContextEnginesInitializedMock = vi.fn(); const resolveContextEngineMock = vi.fn(); @@ -25,7 +25,9 @@ describe("sessions_spawn context modes", () => { beforeAll(async () => { ({ spawnSubagentDirect } = await loadSubagentSpawnModuleForTest({ callGatewayMock, + loadSessionStoreMock, updateSessionStoreMock, + forkSessionEntryFromParentMock, forkSessionFromParentMock, ensureContextEnginesInitializedMock, resolveContextEngineMock, @@ -35,7 +37,9 @@ describe("sessions_spawn context modes", () => { beforeEach(() => { callGatewayMock.mockReset(); + loadSessionStoreMock.mockReset(); updateSessionStoreMock.mockReset(); + forkSessionEntryFromParentMock.mockReset(); forkSessionFromParentMock.mockReset(); ensureContextEnginesInitializedMock.mockReset(); resolveContextEngineMock.mockReset(); @@ -44,14 +48,78 @@ describe("sessions_spawn context modes", () => { }); function usePersistentStoreMock(store: SessionStore) { - // The spawn path mutates the session store in-place; this mock keeps that - // contract visible without touching disk. + loadSessionStoreMock.mockReturnValue(store); updateSessionStoreMock.mockImplementation(async (_storePath: unknown, mutator: unknown) => { if (typeof mutator !== "function") { throw new Error("missing session store mutator"); } return await mutator(store); }); + forkSessionEntryFromParentMock.mockImplementation( + async (params: { + agentId: string; + fallbackEntry?: Record; + parentStoreKeys?: string[]; + sessionKey: string; + sessionsDir?: string; + }) => { + const parentEntry = params.parentStoreKeys + ?.map((key) => store[key]) + .find((entry): entry is Record => Boolean(entry)); + const maxTokens = 100_000; + const parentTokens = parentEntry?.totalTokens; + if ( + typeof parentTokens === "number" && + Number.isFinite(parentTokens) && + parentTokens > maxTokens + ) { + const sessionEntry = { + ...params.fallbackEntry, + ...store[params.sessionKey], + }; + return { + status: "skipped", + reason: "decision-skip", + parentEntry, + sessionEntry, + decision: { + status: "skip", + reason: "parent-too-large", + maxTokens, + parentTokens, + message: `Parent context is too large to fork (${parentTokens}/${maxTokens} tokens); starting with isolated context instead.`, + }, + }; + } + const fork = await forkSessionFromParentMock({ + parentEntry, + agentId: params.agentId, + sessionsDir: params.sessionsDir, + }); + if (!fork) { + return { status: "failed" }; + } + const sessionEntry = { + ...params.fallbackEntry, + ...store[params.sessionKey], + sessionId: fork.sessionId, + sessionFile: fork.sessionFile, + forkedFromParent: true, + }; + store[params.sessionKey] = sessionEntry; + return { + status: "forked", + fork, + parentEntry, + sessionEntry, + decision: { + status: "fork", + maxTokens, + ...(typeof parentTokens === "number" ? { parentTokens } : {}), + }, + }; + }, + ); } function requireAcceptedResult(result: Awaited>) { @@ -202,8 +270,6 @@ describe("sessions_spawn context modes", () => { }); it("falls back to isolated context when requested fork is too large", async () => { - // Forking very large transcripts would create expensive child context, so - // the accepted run records the downgrade in its note. const store: SessionStore = { main: { sessionId: "parent-session-id", diff --git a/src/agents/subagent-spawn.runtime.ts b/src/agents/subagent-spawn.runtime.ts index 40cf714a7a1c..be302a85fa7b 100644 --- a/src/agents/subagent-spawn.runtime.ts +++ b/src/agents/subagent-spawn.runtime.ts @@ -10,6 +10,7 @@ export { export { getRuntimeConfig } from "../config/config.js"; export { loadSessionStore, mergeSessionEntry, updateSessionStore } from "../config/sessions.js"; export { + forkSessionEntryFromParent, forkSessionFromParent, resolveParentForkDecision, type ParentForkDecision, diff --git a/src/agents/subagent-spawn.test-helpers.ts b/src/agents/subagent-spawn.test-helpers.ts index b988a7d326cd..4a0c74dcab92 100644 --- a/src/agents/subagent-spawn.test-helpers.ts +++ b/src/agents/subagent-spawn.test-helpers.ts @@ -134,6 +134,7 @@ export async function loadSubagentSpawnModuleForTest(params: { loadSessionStoreMock?: MockFn; ensureContextEnginesInitializedMock?: MockFn; updateSessionStoreMock?: MockFn; + forkSessionEntryFromParentMock?: MockFn; forkSessionFromParentMock?: MockFn; resolveContextEngineMock?: MockFn; resolveParentForkDecisionMock?: MockFn; @@ -215,6 +216,36 @@ export async function loadSubagentSpawnModuleForTest(params: { params.dispatchGatewayMethodInProcessMock?.(...args), hasInProcessGatewayContext: () => Boolean(params.hasInProcessGatewayContextMock?.()), buildSubagentSystemPrompt: () => "system-prompt", + forkSessionEntryFromParent: + params.forkSessionEntryFromParentMock ?? + (async () => { + const fork = ( + params.forkSessionFromParentMock + ? await params.forkSessionFromParentMock() + : { sessionId: "forked-session-id", sessionFile: "/tmp/forked-session.jsonl" } + ) as { sessionId: string; sessionFile: string } | null; + if (!fork) { + return { status: "failed" }; + } + return { + status: "forked", + fork, + parentEntry: { + sessionId: "parent-session-id", + sessionFile: "/tmp/parent-session.jsonl", + updatedAt: Date.now(), + }, + sessionEntry: { + sessionId: fork.sessionId, + sessionFile: fork.sessionFile, + forkedFromParent: true, + }, + decision: { + status: "fork", + maxTokens: 100_000, + }, + }; + }), forkSessionFromParent: params.forkSessionFromParentMock ?? (async () => ({ sessionId: "forked-session-id", sessionFile: "/tmp/forked-session.jsonl" })), diff --git a/src/agents/subagent-spawn.ts b/src/agents/subagent-spawn.ts index 132e6f9f7cbe..7b211df54dfd 100644 --- a/src/agents/subagent-spawn.ts +++ b/src/agents/subagent-spawn.ts @@ -88,7 +88,7 @@ import { callGateway, dispatchGatewayMethodInProcess, emitSessionLifecycleEvent, - forkSessionFromParent, + forkSessionEntryFromParent, getGlobalHookRunner, getSessionBindingService, getRuntimeConfig, @@ -99,7 +99,6 @@ import { normalizeDeliveryContext, pruneLegacyStoreKeys, ensureContextEnginesInitialized, - resolveParentForkDecision, resolveAgentConfig, resolveContextEngine, resolveGatewaySessionStoreTarget, @@ -133,26 +132,24 @@ function resolveConfiguredAgentIds(cfg: OpenClawConfig): string[] { type SubagentSpawnDeps = { callGateway: typeof callGateway; dispatchGatewayMethodInProcess: typeof dispatchGatewayMethodInProcess; - forkSessionFromParent: typeof forkSessionFromParent; + forkSessionEntryFromParent: typeof forkSessionEntryFromParent; getGlobalHookRunner: () => SubagentLifecycleHookRunner | null; getRuntimeConfig: typeof getRuntimeConfig; hasInProcessGatewayContext: typeof hasInProcessGatewayContext; ensureContextEnginesInitialized: typeof ensureContextEnginesInitialized; resolveContextEngine: typeof resolveContextEngine; - resolveParentForkDecision: typeof resolveParentForkDecision; updateSessionStore: typeof updateSessionStore; }; const defaultSubagentSpawnDeps: SubagentSpawnDeps = { callGateway, dispatchGatewayMethodInProcess, - forkSessionFromParent, + forkSessionEntryFromParent, getGlobalHookRunner, getRuntimeConfig, hasInProcessGatewayContext, ensureContextEnginesInitialized, resolveContextEngine, - resolveParentForkDecision, updateSessionStore, }; @@ -510,52 +507,45 @@ async function prepareSubagentSessionContext(params: { const sessionsDir = path.dirname(parentTarget.storePath); try { - const forked = (await updateSubagentSessionStore(childTarget.storePath, async (store) => { - parentEntry = resolveStoreEntryByKeys(store, parentTarget.storeKeys); - childEntry = resolveStoreEntryByKeys(store, childTarget.storeKeys); + if (params.targetAgentId !== params.requesterAgentId) { + throw new Error( + 'context="fork" currently requires the same target agent as the requester; use context="isolated" for cross-agent spawns.', + ); + } - if (params.targetAgentId !== params.requesterAgentId) { - throw new Error( - 'context="fork" currently requires the same target agent as the requester; use context="isolated" for cross-agent spawns.', - ); - } - if (!parentEntry?.sessionId) { - throw new Error( - 'context="fork" requested but the requester session transcript is not available.', - ); - } - const forkDecision = await subagentSpawnDeps.resolveParentForkDecision({ - parentEntry, - storePath: parentTarget.storePath, - }); - if (forkDecision.status === "skip") { - forkFallbackNote = forkDecision.message; - return null; - } - - const fork = await subagentSpawnDeps.forkSessionFromParent({ - parentEntry, - agentId: params.requesterAgentId, - sessionsDir, - }); - if (!fork) { - throw new Error( - 'context="fork" requested but OpenClaw could not fork the requester transcript.', - ); - } - pruneLegacyStoreKeys({ - store, - canonicalKey: childTarget.canonicalKey, - candidates: childTarget.storeKeys, - }); - store[childTarget.canonicalKey] = mergeSessionEntry(store[childTarget.canonicalKey], { - sessionId: fork.sessionId, - sessionFile: fork.sessionFile, - forkedFromParent: true, - }); - childEntry = store[childTarget.canonicalKey]; - return fork; - })) as { sessionId: string; sessionFile: string } | null; + const forkedResult = await subagentSpawnDeps.forkSessionEntryFromParent({ + storePath: childTarget.storePath, + parentSessionKey: parentTarget.canonicalKey, + parentStoreKeys: parentTarget.storeKeys, + sessionKey: childTarget.canonicalKey, + sessionStoreKeys: childTarget.storeKeys, + fallbackEntry: { sessionId: "", updatedAt: Date.now() }, + agentId: params.requesterAgentId, + sessionsDir, + }); + if (forkedResult.status === "missing-parent") { + throw new Error( + 'context="fork" requested but the requester session transcript is not available.', + ); + } + if (forkedResult.status === "failed" || forkedResult.status === "missing-entry") { + throw new Error( + 'context="fork" requested but OpenClaw could not fork the requester transcript.', + ); + } + parentEntry = forkedResult.parentEntry; + childEntry = forkedResult.sessionEntry; + if (forkedResult.status === "skipped") { + forkFallbackNote = + forkedResult.decision?.status === "skip" ? forkedResult.decision.message : undefined; + } + const forked = + forkedResult.status === "forked" + ? { + sessionId: forkedResult.fork.sessionId, + sessionFile: forkedResult.fork.sessionFile, + } + : null; if (params.contextMode === "fork") { if (!parentEntry || !forked) { diff --git a/src/auto-reply/reply/session-fork.test.ts b/src/auto-reply/reply/session-fork.test.ts new file mode 100644 index 000000000000..00c05e4aa959 --- /dev/null +++ b/src/auto-reply/reply/session-fork.test.ts @@ -0,0 +1,87 @@ +// Tests parent-session fork facade storage-boundary behavior. +import fs from "node:fs/promises"; +import os from "node:os"; +import path from "node:path"; +import { afterEach, describe, expect, it, vi } from "vitest"; +import type { OpenClawConfig } from "../../config/types.openclaw.js"; +import { forkSessionEntryFromParent } from "./session-fork.js"; + +const runtimeMocks = vi.hoisted(() => ({ + forkSessionFromParentRuntime: vi.fn(), + resolveParentForkTokenCountRuntime: vi.fn(), +})); + +vi.mock("./session-fork.runtime.js", () => runtimeMocks); + +const roots: string[] = []; + +async function makeRoot(prefix: string): Promise { + const root = await fs.mkdtemp(path.join(os.tmpdir(), prefix)); + roots.push(root); + return root; +} + +afterEach(async () => { + runtimeMocks.forkSessionFromParentRuntime.mockReset(); + runtimeMocks.resolveParentForkTokenCountRuntime.mockReset(); + await Promise.all(roots.splice(0).map((root) => fs.rm(root, { recursive: true, force: true }))); +}); + +describe("forkSessionEntryFromParent", () => { + it("forks transcripts in the directory for the store being mutated", async () => { + const root = await makeRoot("openclaw-session-fork-boundary-"); + const activeStoreDir = path.join(root, "active-store"); + const configStoreDir = path.join(root, "config-store"); + await fs.mkdir(activeStoreDir, { recursive: true }); + await fs.mkdir(configStoreDir, { recursive: true }); + const storePath = path.join(activeStoreDir, "sessions.json"); + const configStorePath = path.join(configStoreDir, "sessions.json"); + const parentSessionKey = "agent:main:main"; + const sessionKey = "agent:main:subagent:child"; + await fs.writeFile( + storePath, + JSON.stringify( + { + [parentSessionKey]: { + sessionId: "parent-session", + sessionFile: path.join(activeStoreDir, "parent.jsonl"), + updatedAt: 1, + }, + [sessionKey]: { sessionId: "", updatedAt: 2 }, + }, + null, + 2, + ), + "utf-8", + ); + + runtimeMocks.resolveParentForkTokenCountRuntime.mockResolvedValue(10); + runtimeMocks.forkSessionFromParentRuntime.mockImplementation( + async ({ sessionsDir }: { sessionsDir: string }) => ({ + sessionId: "forked-session", + sessionFile: path.join(sessionsDir, "forked-session.jsonl"), + }), + ); + + const result = await forkSessionEntryFromParent({ + agentId: "main", + config: { session: { store: configStorePath } } as OpenClawConfig, + fallbackEntry: { sessionId: "", updatedAt: 2 }, + parentSessionKey, + sessionKey, + storePath, + }); + + expect(result.status).toBe("forked"); + expect(runtimeMocks.forkSessionFromParentRuntime).toHaveBeenCalledWith( + expect.objectContaining({ + sessionsDir: activeStoreDir, + }), + ); + const stored = JSON.parse(await fs.readFile(storePath, "utf-8")) as Record< + string, + { sessionFile?: string } + >; + expect(stored[sessionKey]?.sessionFile).toBe(path.join(activeStoreDir, "forked-session.jsonl")); + }); +}); diff --git a/src/auto-reply/reply/session-fork.ts b/src/auto-reply/reply/session-fork.ts index 228bc447a2ee..0e8d145a762a 100644 --- a/src/auto-reply/reply/session-fork.ts +++ b/src/auto-reply/reply/session-fork.ts @@ -1,5 +1,8 @@ -/** Public session-fork facade with parent-size admission checks. */ -import type { SessionEntry } from "../../config/sessions/types.js"; +import path from "node:path"; +import { resolveStorePath } from "../../config/sessions/paths.js"; +import { updateSessionStore } from "../../config/sessions/store.js"; +import { mergeSessionEntry, type SessionEntry } from "../../config/sessions/types.js"; +import type { OpenClawConfig } from "../../config/types.openclaw.js"; import { createLazyImportLoader } from "../../shared/lazy-promise.js"; /** @@ -10,7 +13,6 @@ import { createLazyImportLoader } from "../../shared/lazy-promise.js"; const DEFAULT_PARENT_FORK_MAX_TOKENS = 100_000; const sessionForkRuntimeLoader = createLazyImportLoader(() => import("./session-fork.runtime.js")); -/** Decision for whether a child session should fork parent context or start isolated. */ export type ParentForkDecision = | { status: "fork"; @@ -25,6 +27,66 @@ export type ParentForkDecision = message: string; }; +type ParentForkDecisionParams = { + parentEntry: SessionEntry; + agentId?: string; + config?: OpenClawConfig; + storePath?: string; +}; + +type ForkSessionFromParentParams = { + parentEntry: SessionEntry; + agentId: string; + config?: OpenClawConfig; + sessionsDir?: string; +}; + +export type ForkedParentSessionEntry = { + sessionId: string; + sessionFile: string; +}; + +export type ForkSessionEntryFromParentResult = + | { + status: "forked"; + fork: ForkedParentSessionEntry; + parentEntry: SessionEntry; + sessionEntry: SessionEntry; + decision: Extract; + } + | { + status: "skipped"; + reason: "existing-entry" | "decision-skip"; + parentEntry?: SessionEntry; + sessionEntry: SessionEntry; + decision?: ParentForkDecision; + } + | { status: "missing-entry" } + | { status: "missing-parent" } + | { status: "failed" }; + +export type ForkSessionEntryFromParentParams = Omit & { + parentSessionKey: string; + parentStoreKeys?: readonly string[]; + sessionKey: string; + sessionStoreKeys?: readonly string[]; + storePath?: string; + fallbackEntry?: SessionEntry; + patch?: (params: { + entry: SessionEntry; + parentEntry: SessionEntry; + fork: ForkedParentSessionEntry; + decision: Extract; + }) => Partial; + skipForkWhen?: (entry: SessionEntry) => boolean; + skipPatch?: (entry: SessionEntry) => Partial | null; + decisionSkipPatch?: (params: { + decision: Extract; + entry: SessionEntry; + parentEntry: SessionEntry; + }) => Partial | null; +}; + function loadSessionForkRuntime(): Promise { return sessionForkRuntimeLoader.load(); } @@ -39,15 +101,31 @@ function formatParentForkTooLargeMessage(params: { ); } -/** Decides whether parent context is small enough to fork into a child session. */ -export async function resolveParentForkDecision(params: { - parentEntry: SessionEntry; - storePath: string; -}): Promise { +function resolveParentForkStorePath(params: { + agentId?: string; + config?: OpenClawConfig; + storePath?: string; +}): string { + return ( + params.storePath ?? resolveStorePath(params.config?.session?.store, { agentId: params.agentId }) + ); +} + +function resolveParentForkSessionsDir(params: { + agentId: string; + config?: OpenClawConfig; + sessionsDir?: string; +}): string { + return params.sessionsDir ?? path.dirname(resolveParentForkStorePath(params)); +} + +export async function resolveParentForkDecision( + params: ParentForkDecisionParams, +): Promise { const maxTokens = DEFAULT_PARENT_FORK_MAX_TOKENS; const parentTokens = await resolveParentForkTokenCount({ parentEntry: params.parentEntry, - storePath: params.storePath, + storePath: resolveParentForkStorePath(params), }); if (typeof parentTokens === "number" && parentTokens > maxTokens) { return { @@ -65,14 +143,151 @@ export async function resolveParentForkDecision(params: { }; } -/** Forks a new session transcript from a parent session. */ -export async function forkSessionFromParent(params: { - parentEntry: SessionEntry; - agentId: string; - sessionsDir: string; -}): Promise<{ sessionId: string; sessionFile: string } | null> { +export async function forkSessionFromParent( + params: ForkSessionFromParentParams, +): Promise<{ sessionId: string; sessionFile: string } | null> { const runtime = await loadSessionForkRuntime(); - return runtime.forkSessionFromParentRuntime(params); + return runtime.forkSessionFromParentRuntime({ + ...params, + sessionsDir: resolveParentForkSessionsDir(params), + }); +} + +function resolveEntryFromStoreKeys(params: { + store: Record; + keys: readonly string[]; +}): SessionEntry | undefined { + for (const key of params.keys) { + const entry = params.store[key]; + if (entry) { + return entry; + } + } + return undefined; +} + +function persistForkedSessionEntry(params: { + store: Record; + sessionKey: string; + sessionStoreKeys?: readonly string[]; + existing: SessionEntry; + patch: Partial; +}): SessionEntry { + const next = mergeSessionEntry(params.existing, params.patch); + params.store[params.sessionKey] = next; + for (const key of params.sessionStoreKeys ?? []) { + if (key !== params.sessionKey) { + delete params.store[key]; + } + } + return next; +} + +/** + * Forks the parent transcript and persists the child session entry through one + * storage boundary operation. + */ +export async function forkSessionEntryFromParent( + params: ForkSessionEntryFromParentParams, +): Promise { + const storePath = resolveParentForkStorePath(params); + return await updateSessionStore( + storePath, + async (store) => { + const parentEntry = resolveEntryFromStoreKeys({ + store, + keys: params.parentStoreKeys ?? [params.parentSessionKey], + }); + if (!parentEntry?.sessionId) { + return { status: "missing-parent" }; + } + + const entry = + resolveEntryFromStoreKeys({ + store, + keys: params.sessionStoreKeys ?? [params.sessionKey], + }) ?? params.fallbackEntry; + if (!entry) { + return { status: "missing-entry" }; + } + + if (params.skipForkWhen?.(entry)) { + const patch = params.skipPatch?.(entry); + const sessionEntry = patch + ? persistForkedSessionEntry({ + store, + sessionKey: params.sessionKey, + sessionStoreKeys: params.sessionStoreKeys, + existing: entry, + patch, + }) + : entry; + return { status: "skipped", reason: "existing-entry", parentEntry, sessionEntry }; + } + + const decision = await resolveParentForkDecision({ + parentEntry, + agentId: params.agentId, + config: params.config, + storePath, + }); + if (decision.status === "skip") { + const patch = params.decisionSkipPatch?.({ decision, entry, parentEntry }); + const sessionEntry = patch + ? persistForkedSessionEntry({ + store, + sessionKey: params.sessionKey, + sessionStoreKeys: params.sessionStoreKeys, + existing: entry, + patch, + }) + : entry; + return { + status: "skipped", + reason: "decision-skip", + parentEntry, + sessionEntry, + decision, + }; + } + + const fork = await forkSessionFromParent({ + parentEntry, + agentId: params.agentId, + config: params.config, + sessionsDir: params.sessionsDir ?? path.dirname(storePath), + }); + if (!fork) { + return { status: "failed" }; + } + const sessionEntry = persistForkedSessionEntry({ + store, + sessionKey: params.sessionKey, + sessionStoreKeys: params.sessionStoreKeys, + existing: entry, + patch: { + ...params.patch?.({ entry, parentEntry, fork, decision }), + sessionId: fork.sessionId, + sessionFile: fork.sessionFile, + forkedFromParent: true, + }, + }); + return { + status: "forked", + fork, + parentEntry, + sessionEntry, + decision, + }; + }, + { + skipSaveWhenResult: (result) => + result.status === "missing-entry" || + result.status === "missing-parent" || + result.status === "failed" || + (result.status === "skipped" && result.sessionEntry === params.fallbackEntry), + }, + ); } async function resolveParentForkTokenCount(params: { diff --git a/src/auto-reply/reply/session.test.ts b/src/auto-reply/reply/session.test.ts index e9c582726217..0e5824c052ea 100644 --- a/src/auto-reply/reply/session.test.ts +++ b/src/auto-reply/reply/session.test.ts @@ -50,6 +50,97 @@ type ForkSessionParamsForTest = { }; vi.mock("./session-fork.js", () => ({ + forkSessionEntryFromParent: async (params: { + fallbackEntry?: SessionEntry; + parentSessionKey: string; + storePath: string; + patch?: (patchParams: { + entry: SessionEntry; + parentEntry: SessionEntry; + fork: { sessionId: string; sessionFile: string }; + decision: { status: "fork"; maxTokens: number; parentTokens?: number }; + }) => Partial; + decisionSkipPatch?: (patchParams: { + decision: { + status: "skip"; + reason: "parent-too-large"; + maxTokens: number; + parentTokens: number; + message: string; + }; + entry: SessionEntry; + parentEntry: SessionEntry; + }) => Partial; + sessionsDir: string; + }) => { + const store = JSON.parse(await fs.readFile(params.storePath, "utf-8")) as Record< + string, + SessionEntry + >; + const parentEntry = store[params.parentSessionKey]; + if (!parentEntry?.sessionId) { + return { status: "missing-parent" }; + } + const maxTokens = 100_000; + const parentTokens = await sessionForkMocks.resolveParentForkTokenCount({ + parentEntry, + storePath: params.storePath, + }); + if (typeof parentTokens === "number" && parentTokens > maxTokens) { + const entry = params.fallbackEntry ?? { sessionId: "", updatedAt: Date.now() }; + const decision = { + status: "skip" as const, + reason: "parent-too-large" as const, + maxTokens, + parentTokens, + message: `Parent context is too large to fork (${parentTokens}/${maxTokens} tokens); starting with isolated context instead.`, + }; + return { + status: "skipped", + reason: "decision-skip", + parentEntry, + sessionEntry: { + ...entry, + ...params.decisionSkipPatch?.({ decision, entry, parentEntry }), + }, + decision, + }; + } + const fork = await sessionForkMocks.forkSessionFromParent({ + parentEntry, + sessionsDir: params.sessionsDir, + }); + if (!fork) { + return { status: "failed" }; + } + const entry = params.fallbackEntry ?? { sessionId: "", updatedAt: Date.now() }; + return { + status: "forked", + fork, + parentEntry, + sessionEntry: { + ...entry, + ...params.patch?.({ + entry, + parentEntry, + fork, + decision: { + status: "fork", + maxTokens, + ...(typeof parentTokens === "number" ? { parentTokens } : {}), + }, + }), + sessionId: fork.sessionId, + sessionFile: fork.sessionFile, + forkedFromParent: true, + }, + decision: { + status: "fork", + maxTokens, + ...(typeof parentTokens === "number" ? { parentTokens } : {}), + }, + }; + }, forkSessionFromParent: (...args: [ForkSessionParamsForTest]) => sessionForkMocks.forkSessionFromParent(...args), resolveParentForkTokenCount: (...args: [{ parentEntry: SessionEntry; storePath: string }]) => diff --git a/src/auto-reply/reply/session.ts b/src/auto-reply/reply/session.ts index 9507ab0e1f01..abf2278c8c12 100644 --- a/src/auto-reply/reply/session.ts +++ b/src/auto-reply/reply/session.ts @@ -73,7 +73,7 @@ import { resolveLastChannelRaw, resolveLastToRaw, } from "./session-delivery.js"; -import { forkSessionFromParent, resolveParentForkDecision } from "./session-fork.js"; +import { forkSessionEntryFromParent } from "./session-fork.js"; import { buildSessionEndHookPayload, buildSessionStartHookPayload } from "./session-hooks.js"; import { clearSessionResetRuntimeState } from "./session-reset-cleanup.js"; @@ -753,47 +753,39 @@ export async function initSessionState(params: { const parentSessionKey = normalizeOptionalString(ctx.ParentSessionKey); const alreadyForked = sessionEntry.forkedFromParent === true; let inheritedParentContext = false; - if ( - parentSessionKey && - parentSessionKey !== sessionKey && - sessionStore[parentSessionKey] && - !alreadyForked - ) { - const parentEntry = sessionStore[parentSessionKey]; - const forkDecision = await resolveParentForkDecision({ - parentEntry, + if (parentSessionKey && parentSessionKey !== sessionKey && !alreadyForked) { + const forked = await forkSessionEntryFromParent({ + parentSessionKey, + sessionKey, storePath, + fallbackEntry: sessionEntry, + agentId, + sessionsDir: path.dirname(storePath), + decisionSkipPatch: () => ({ ...sessionEntry, forkedFromParent: true }), + patch: () => ({ + ...sessionEntry, + totalTokens: undefined, + totalTokensFresh: false, + }), }); - if (forkDecision.status === "skip") { + if (forked.status === "skipped" && forked.decision?.status === "skip") { // The parent branch is too large to inherit usefully. Start fresh and // mark as handled so the thread does not retry this decision every turn. log.warn( `skipping parent fork (parent too large): parentKey=${parentSessionKey} → sessionKey=${sessionKey} ` + - `parentTokens=${forkDecision.parentTokens} maxTokens=${forkDecision.maxTokens}`, + `parentTokens=${forked.decision.parentTokens} maxTokens=${forked.decision.maxTokens}`, ); - sessionEntry.forkedFromParent = true; - } else { + sessionEntry = forked.sessionEntry; + } else if (forked.status === "forked") { log.warn( `forking from parent session: parentKey=${parentSessionKey} → sessionKey=${sessionKey} ` + - `parentTokens=${forkDecision.parentTokens ?? "unknown"}`, + `parentTokens=${forked.decision.parentTokens ?? "unknown"}`, ); - const forked = await forkSessionFromParent({ - parentEntry, - agentId, - sessionsDir: path.dirname(storePath), - }); - if (forked) { - sessionId = forked.sessionId; - sessionEntry.sessionId = forked.sessionId; - sessionEntry.sessionFile = forked.sessionFile; - sessionEntry.forkedFromParent = true; - // The fork replaces the target transcript with inherited parent - // history, so any prior target-session token snapshot is stale. - sessionEntry.totalTokens = undefined; - sessionEntry.totalTokensFresh = false; - inheritedParentContext = true; - log.warn(`forked session created: file=${forked.sessionFile}`); - } + sessionId = forked.fork.sessionId; + sessionEntry = forked.sessionEntry; + sessionEntry.forkedFromParent = true; + inheritedParentContext = true; + log.warn(`forked session created: file=${forked.fork.sessionFile}`); } } const threadIdFromSessionKey = parseSessionThreadInfoFast( diff --git a/src/gateway/server-methods/sessions.ts b/src/gateway/server-methods/sessions.ts index 5f6ef66f9acf..700bc067ed95 100644 --- a/src/gateway/server-methods/sessions.ts +++ b/src/gateway/server-methods/sessions.ts @@ -2,7 +2,6 @@ // restore/preview/send flows over session stores, transcripts, and active runs. import { randomUUID } from "node:crypto"; import fs from "node:fs"; -import path from "node:path"; import { normalizeOptionalLowercaseString, normalizeOptionalString, @@ -86,7 +85,7 @@ import { import { ADMIN_SCOPE } from "../operator-scopes.js"; import { resolveSessionKeyForRun } from "../server-session-key.js"; import { - forkCompactionCheckpointTranscriptAsync, + createFileBackedCompactionCheckpointStore, getSessionCompactionCheckpoint, listSessionCompactionCheckpoints, } from "../session-compaction-checkpoints.js"; @@ -136,6 +135,8 @@ import type { } from "./types.js"; import { assertValidParams } from "./validation.js"; +const compactionCheckpointStore = createFileBackedCompactionCheckpointStore(); + function filterSessionStoreToConfiguredAgents( cfg: OpenClawConfig, store: Record, @@ -350,74 +351,6 @@ function buildDashboardSessionKey(agentId: string): string { return `agent:${agentId}:dashboard:${randomUUID()}`; } -function cloneCheckpointSessionEntry(params: { - currentEntry: SessionEntry; - nextSessionId: string; - nextSessionFile: string; - label?: string; - parentSessionKey?: string; - totalTokens?: number; - preserveCompactionCheckpoints?: boolean; -}): SessionEntry { - return { - ...params.currentEntry, - sessionId: params.nextSessionId, - sessionFile: params.nextSessionFile, - updatedAt: Date.now(), - systemSent: false, - abortedLastRun: false, - startedAt: undefined, - endedAt: undefined, - runtimeMs: undefined, - status: undefined, - inputTokens: undefined, - outputTokens: undefined, - cacheRead: undefined, - cacheWrite: undefined, - estimatedCostUsd: undefined, - totalTokens: - typeof params.totalTokens === "number" && Number.isFinite(params.totalTokens) - ? params.totalTokens - : undefined, - totalTokensFresh: - typeof params.totalTokens === "number" && Number.isFinite(params.totalTokens) - ? true - : undefined, - label: params.label ?? params.currentEntry.label, - parentSessionKey: params.parentSessionKey ?? params.currentEntry.parentSessionKey, - compactionCheckpoints: params.preserveCompactionCheckpoints - ? params.currentEntry.compactionCheckpoints - : undefined, - }; -} - -function resolveCheckpointForkSource( - checkpoint: NonNullable>, -): { sourceFile: string; sourceLeafId?: string; totalTokens?: number } | null { - const preCompactionFile = checkpoint.preCompaction.sessionFile?.trim(); - if (preCompactionFile) { - return { - sourceFile: preCompactionFile, - sourceLeafId: checkpoint.preCompaction.entryId ?? checkpoint.preCompaction.leafId, - totalTokens: checkpoint.tokensBefore, - }; - } - const postCompactionFile = checkpoint.postCompaction.sessionFile?.trim(); - if (!postCompactionFile) { - return null; - } - const postCompactionLeafId = - checkpoint.postCompaction.entryId ?? checkpoint.postCompaction.leafId; - if (!postCompactionLeafId) { - return null; - } - return { - sourceFile: postCompactionFile, - sourceLeafId: postCompactionLeafId, - totalTokens: checkpoint.tokensAfter, - }; -} - function isAgentMainSessionKey(cfg: OpenClawConfig, sessionKey: string): boolean { const parsed = parseAgentSessionKey(sessionKey); if (!parsed) { @@ -1663,7 +1596,7 @@ export const sessionsHandlers: GatewayRequestHandlers = { return; } const loaded = loadSessionEntry(key, { agentId: requestedAgent.agentId }); - const { cfg: loadedCfg, entry, canonicalKey } = loaded; + const { cfg: loadedCfg, entry, canonicalKey, legacyKey } = loaded; const target = resolveGatewaySessionStoreTarget({ cfg: loadedCfg, key: canonicalKey, @@ -1678,8 +1611,7 @@ export const sessionsHandlers: GatewayRequestHandlers = { return; } const checkpoint = getSessionCompactionCheckpoint({ entry, checkpointId }); - const forkSource = checkpoint ? resolveCheckpointForkSource(checkpoint) : null; - if (!checkpoint || !forkSource) { + if (!checkpoint) { respond( false, undefined, @@ -1687,12 +1619,34 @@ export const sessionsHandlers: GatewayRequestHandlers = { ); return; } - const branchedSession = await forkCompactionCheckpointTranscriptAsync({ - sourceFile: forkSource.sourceFile, - sourceLeafId: forkSource.sourceLeafId, - sessionDir: path.dirname(forkSource.sourceFile), + const nextKey = buildDashboardSessionKey(target.agentId); + const branchedSession = await compactionCheckpointStore.branchCheckpointSession({ + storePath: target.storePath, + sourceKey: canonicalKey, + sourceStoreKey: legacyKey, + nextKey, + checkpointId, }); - if (!branchedSession?.sessionFile) { + if ( + branchedSession.status === "missing-checkpoint" || + branchedSession.status === "missing-boundary" + ) { + respond( + false, + undefined, + errorShape(ErrorCodes.INVALID_REQUEST, `checkpoint not found: ${checkpointId}`), + ); + return; + } + if (branchedSession.status === "missing-session") { + respond( + false, + undefined, + errorShape(ErrorCodes.INVALID_REQUEST, `session not found: ${key}`), + ); + return; + } + if (branchedSession.status === "failed") { respond( false, undefined, @@ -1700,30 +1654,16 @@ export const sessionsHandlers: GatewayRequestHandlers = { ); return; } - const nextKey = buildDashboardSessionKey(target.agentId); - const label = entry.label?.trim() ? `${entry.label.trim()} (checkpoint)` : "Checkpoint branch"; - const nextEntry = cloneCheckpointSessionEntry({ - currentEntry: entry, - nextSessionId: branchedSession.sessionId, - nextSessionFile: branchedSession.sessionFile, - label, - parentSessionKey: canonicalKey, - totalTokens: forkSource.totalTokens, - }); - - await updateSessionStore(target.storePath, (store) => { - store[nextKey] = nextEntry; - }); respond( true, { ok: true, sourceKey: canonicalKey, - key: nextKey, - sessionId: nextEntry.sessionId, - checkpoint, - entry: nextEntry, + key: branchedSession.key, + sessionId: branchedSession.entry.sessionId, + checkpoint: branchedSession.checkpoint, + entry: branchedSession.entry, }, undefined, ); @@ -1735,7 +1675,7 @@ export const sessionsHandlers: GatewayRequestHandlers = { reason: "checkpoint-branch", }); emitSessionsChanged(context, { - sessionKey: nextKey, + sessionKey: branchedSession.key, reason: "checkpoint-branch", }); }, @@ -1778,7 +1718,7 @@ export const sessionsHandlers: GatewayRequestHandlers = { return; } const loaded = loadSessionEntry(key, { agentId: requestedAgent.agentId }); - const { entry, canonicalKey, storePath } = loaded; + const { entry, canonicalKey, legacyKey, storePath } = loaded; if (!entry?.sessionId) { respond( false, @@ -1788,8 +1728,7 @@ export const sessionsHandlers: GatewayRequestHandlers = { return; } const checkpoint = getSessionCompactionCheckpoint({ entry, checkpointId }); - const forkSource = checkpoint ? resolveCheckpointForkSource(checkpoint) : null; - if (!checkpoint || !forkSource) { + if (!checkpoint) { respond( false, undefined, @@ -1812,12 +1751,32 @@ export const sessionsHandlers: GatewayRequestHandlers = { return; } - const restoredSession = await forkCompactionCheckpointTranscriptAsync({ - sourceFile: forkSource.sourceFile, - sourceLeafId: forkSource.sourceLeafId, - sessionDir: path.dirname(forkSource.sourceFile), + const restoredSession = await compactionCheckpointStore.restoreCheckpointSession({ + storePath, + sessionKey: canonicalKey, + sessionStoreKey: legacyKey, + checkpointId, }); - if (!restoredSession?.sessionFile) { + if ( + restoredSession.status === "missing-checkpoint" || + restoredSession.status === "missing-boundary" + ) { + respond( + false, + undefined, + errorShape(ErrorCodes.INVALID_REQUEST, `checkpoint not found: ${checkpointId}`), + ); + return; + } + if (restoredSession.status === "missing-session") { + respond( + false, + undefined, + errorShape(ErrorCodes.INVALID_REQUEST, `session not found: ${key}`), + ); + return; + } + if (restoredSession.status === "failed") { respond( false, undefined, @@ -1825,26 +1784,15 @@ export const sessionsHandlers: GatewayRequestHandlers = { ); return; } - const nextEntry = cloneCheckpointSessionEntry({ - currentEntry: entry, - nextSessionId: restoredSession.sessionId, - nextSessionFile: restoredSession.sessionFile, - totalTokens: forkSource.totalTokens, - preserveCompactionCheckpoints: true, - }); - - await updateSessionStore(storePath, (store) => { - store[canonicalKey] = nextEntry; - }); respond( true, { ok: true, - key: canonicalKey, - sessionId: nextEntry.sessionId, - checkpoint, - entry: nextEntry, + key: restoredSession.key, + sessionId: restoredSession.entry.sessionId, + checkpoint: restoredSession.checkpoint, + entry: restoredSession.entry, }, undefined, ); diff --git a/src/gateway/session-compaction-checkpoints.test.ts b/src/gateway/session-compaction-checkpoints.test.ts index 72499402f3a7..1c7cfb44525f 100644 --- a/src/gateway/session-compaction-checkpoints.test.ts +++ b/src/gateway/session-compaction-checkpoints.test.ts @@ -12,6 +12,7 @@ import type { OpenClawConfig } from "../config/types.openclaw.js"; import { captureCompactionCheckpointSnapshotAsync, cleanupCompactionCheckpointSnapshot, + createFileBackedCompactionCheckpointStore, forkCompactionCheckpointTranscriptAsync, MAX_COMPACTION_CHECKPOINT_LEAF_SCAN_BYTES, MAX_COMPACTION_CHECKPOINT_RETAINED_BYTES_PER_SESSION, @@ -91,7 +92,10 @@ function checkpointConfig(storePath: string): OpenClawConfig { async function writeSessionStore( storePath: string, sessionKey: string, - entry: { sessionId: string; updatedAt: number; compactionCheckpoints?: unknown[] }, + entry: { sessionId: string; updatedAt: number; compactionCheckpoints?: unknown[] } & Record< + string, + unknown + >, ): Promise { await fs.writeFile(storePath, JSON.stringify({ [sessionKey]: entry }, null, 2), "utf-8"); } @@ -629,6 +633,79 @@ describe("session-compaction-checkpoints", () => { ]); }); + test("file-backed checkpoint store restores from the stored transcript boundary", async () => { + const dir = await fs.mkdtemp(path.join(os.tmpdir(), "openclaw-checkpoint-store-")); + tempDirs.push(dir); + + const session = SessionManager.create(dir, dir); + session.appendMessage({ + role: "user", + content: "checkpoint source", + timestamp: Date.now(), + }); + const checkpointLeafId = requireNonEmptyString( + session.getLeafId(), + "checkpoint leaf id missing", + ); + session.appendMessage({ + role: "assistant", + content: "future turn", + api: "responses", + provider: "openai", + model: "gpt-test", + timestamp: Date.now(), + } as unknown as AssistantMessage); + + const sessionFile = requireNonEmptyString(session.getSessionFile(), "session file missing"); + const storePath = path.join(dir, "sessions.json"); + await writeSessionStore(storePath, MAIN_SESSION_KEY, { + sessionId: "current-session", + sessionFile, + updatedAt: Date.now() - 1, + totalTokens: 200, + compactionCheckpoints: [ + { + checkpointId: "checkpoint-1", + sessionKey: MAIN_SESSION_KEY, + sessionId: "stored-session", + createdAt: Date.now(), + reason: "manual", + tokensAfter: 45, + preCompaction: { sessionId: "pre-session", leafId: "pre-leaf" }, + postCompaction: { + sessionId: "post-session", + sessionFile, + leafId: checkpointLeafId, + }, + }, + ], + }); + const store = createFileBackedCompactionCheckpointStore(); + const restored = await store.restoreCheckpointSession({ + storePath, + sessionKey: MAIN_SESSION_KEY, + checkpointId: "checkpoint-1", + }); + + if (restored.status !== "created") { + throw new Error("expected restored checkpoint transcript"); + } + expect(restored.entry.totalTokens).toBe(45); + const restoredSessionFile = requireNonEmptyString( + restored.entry.sessionFile, + "restored session file missing", + ); + const messages = SessionManager.open(restoredSessionFile, dir).buildSessionContext().messages; + expect(messages.map((message) => (message as { content?: unknown }).content)).toEqual([ + "checkpoint source", + ]); + const nextStore = await readSessionStore<{ sessionFile?: string; totalTokens?: number }>( + storePath, + ); + expect(nextStore[MAIN_SESSION_KEY]?.sessionFile).toBe(restored.entry.sessionFile); + expect(nextStore[MAIN_SESSION_KEY]?.totalTokens).toBe(45); + }); + test("async fork migrates legacy checkpoint snapshots before writing a current header", async () => { const dir = await fs.mkdtemp(path.join(os.tmpdir(), "openclaw-checkpoint-legacy-fork-")); tempDirs.push(dir); diff --git a/src/gateway/session-compaction-checkpoints.ts b/src/gateway/session-compaction-checkpoints.ts index 60195c162611..b24d2a269791 100644 --- a/src/gateway/session-compaction-checkpoints.ts +++ b/src/gateway/session-compaction-checkpoints.ts @@ -57,6 +57,87 @@ type ForkedCompactionCheckpointTranscript = { sessionFile: string; }; +export type CompactionCheckpointForkedTranscript = ForkedCompactionCheckpointTranscript & { + totalTokens?: number; +}; + +export type CompactionCheckpointTranscriptForkResult = + | { status: "created"; transcript: CompactionCheckpointForkedTranscript } + | { status: "missing-boundary" } + | { status: "failed" }; + +export type CompactionCheckpointSessionMutationResult = + | { + status: "created"; + key: string; + checkpoint: SessionCompactionCheckpoint; + entry: SessionEntry; + } + | { status: "missing-session" } + | { status: "missing-checkpoint" } + | { status: "missing-boundary" } + | { status: "failed" }; + +export type BranchCheckpointSessionParams = { + storePath: string; + sourceKey: string; + sourceStoreKey?: string; + nextKey: string; + checkpointId: string; +}; + +export type RestoreCheckpointSessionParams = { + storePath: string; + sessionKey: string; + sessionStoreKey?: string; + checkpointId: string; +}; + +export type PersistSessionCompactionCheckpointParams = { + cfg: OpenClawConfig; + sessionKey: string; + sessionId: string; + reason: SessionCompactionCheckpointReason; + snapshot: CapturedCompactionCheckpointSnapshot; + summary?: string; + firstKeptEntryId?: string; + tokensBefore?: number; + tokensAfter?: number; + postSessionFile?: string; + postLeafId?: string; + postEntryId?: string; + createdAt?: number; +}; + +/** + * Storage boundary for compaction checkpoint capture, persistence, branch, + * restore, and cleanup operations. + */ +export type CompactionCheckpointStore = { + /** Captures the pre-compaction transcript identity without copying rows/files. */ + captureSnapshot: typeof captureCompactionCheckpointSnapshotAsync; + /** Persists checkpoint metadata and prunes checkpoint artifacts owned by this store. */ + persistCheckpoint: ( + params: PersistSessionCompactionCheckpointParams, + ) => Promise; + /** Cleans unpersisted legacy snapshot artifacts after failed persistence. */ + cleanupSnapshot: typeof cleanupCompactionCheckpointSnapshot; + /** + * Creates a checkpoint branch and records its session entry in one logical + * store mutation. + */ + branchCheckpointSession: ( + params: BranchCheckpointSessionParams, + ) => Promise; + /** + * Restores a checkpoint and replaces the current session entry in one logical + * store mutation. + */ + restoreCheckpointSession: ( + params: RestoreCheckpointSessionParams, + ) => Promise; +}; + function checkpointSnapshotPath(checkpoint: SessionCompactionCheckpoint): string | undefined { return checkpoint.preCompaction.sessionFile?.trim() || undefined; } @@ -401,6 +482,209 @@ export async function forkCompactionCheckpointTranscriptAsync(params: { } } +function resolveCheckpointTranscriptForkSource( + checkpoint: SessionCompactionCheckpoint, +): { sourceFile: string; sourceLeafId?: string; totalTokens?: number } | null { + const preCompactionFile = checkpoint.preCompaction.sessionFile?.trim(); + if (preCompactionFile) { + return { + sourceFile: preCompactionFile, + sourceLeafId: checkpoint.preCompaction.entryId ?? checkpoint.preCompaction.leafId, + totalTokens: checkpoint.tokensBefore, + }; + } + + const postCompactionFile = checkpoint.postCompaction.sessionFile?.trim(); + if (!postCompactionFile) { + return null; + } + const postCompactionLeafId = + checkpoint.postCompaction.entryId ?? checkpoint.postCompaction.leafId; + if (!postCompactionLeafId) { + return null; + } + return { + sourceFile: postCompactionFile, + sourceLeafId: postCompactionLeafId, + totalTokens: checkpoint.tokensAfter, + }; +} + +async function forkCheckpointTranscriptFromStoredBoundary(params: { + checkpoint: SessionCompactionCheckpoint; + sessionDir?: string; + targetCwd?: string; +}): Promise { + const forkSource = resolveCheckpointTranscriptForkSource(params.checkpoint); + if (!forkSource) { + return { status: "missing-boundary" }; + } + const forked = await forkCompactionCheckpointTranscriptAsync({ + sourceFile: forkSource.sourceFile, + sourceLeafId: forkSource.sourceLeafId, + sessionDir: params.sessionDir ?? path.dirname(forkSource.sourceFile), + ...(params.targetCwd ? { targetCwd: params.targetCwd } : {}), + }); + if (!forked) { + return { status: "failed" }; + } + return { + status: "created", + transcript: { + ...forked, + ...(typeof forkSource.totalTokens === "number" + ? { totalTokens: forkSource.totalTokens } + : {}), + }, + }; +} + +function cloneCheckpointSessionEntry(params: { + currentEntry: SessionEntry; + nextSessionId: string; + nextSessionFile: string; + label?: string; + parentSessionKey?: string; + totalTokens?: number; + preserveCompactionCheckpoints?: boolean; +}): SessionEntry { + return { + ...params.currentEntry, + sessionId: params.nextSessionId, + sessionFile: params.nextSessionFile, + updatedAt: Date.now(), + systemSent: false, + abortedLastRun: false, + startedAt: undefined, + endedAt: undefined, + runtimeMs: undefined, + status: undefined, + inputTokens: undefined, + outputTokens: undefined, + cacheRead: undefined, + cacheWrite: undefined, + estimatedCostUsd: undefined, + totalTokens: + typeof params.totalTokens === "number" && Number.isFinite(params.totalTokens) + ? params.totalTokens + : undefined, + totalTokensFresh: + typeof params.totalTokens === "number" && Number.isFinite(params.totalTokens) + ? true + : undefined, + label: params.label ?? params.currentEntry.label, + parentSessionKey: params.parentSessionKey ?? params.currentEntry.parentSessionKey, + compactionCheckpoints: params.preserveCompactionCheckpoints + ? params.currentEntry.compactionCheckpoints + : undefined, + }; +} + +async function branchCheckpointSessionFromStoredBoundary( + params: BranchCheckpointSessionParams, +): Promise { + return await updateSessionStore( + params.storePath, + async (store) => { + const currentEntry = store[params.sourceStoreKey ?? params.sourceKey]; + if (!currentEntry?.sessionId) { + return { status: "missing-session" }; + } + const checkpoint = getSessionCompactionCheckpoint({ + entry: currentEntry, + checkpointId: params.checkpointId, + }); + if (!checkpoint) { + return { status: "missing-checkpoint" }; + } + const forkedSession = await forkCheckpointTranscriptFromStoredBoundary({ checkpoint }); + if (forkedSession.status !== "created") { + return forkedSession; + } + + const forkedTranscript = forkedSession.transcript; + const label = currentEntry.label?.trim() + ? `${currentEntry.label.trim()} (checkpoint)` + : "Checkpoint branch"; + const nextEntry = cloneCheckpointSessionEntry({ + currentEntry, + nextSessionId: forkedTranscript.sessionId, + nextSessionFile: forkedTranscript.sessionFile, + label, + parentSessionKey: params.sourceKey, + totalTokens: forkedTranscript.totalTokens, + }); + store[params.nextKey] = nextEntry; + return { + status: "created", + key: params.nextKey, + checkpoint, + entry: nextEntry, + }; + }, + { skipSaveWhenResult: (result) => result.status !== "created" }, + ); +} + +async function restoreCheckpointSessionFromStoredBoundary( + params: RestoreCheckpointSessionParams, +): Promise { + return await updateSessionStore( + params.storePath, + async (store) => { + const currentEntry = store[params.sessionStoreKey ?? params.sessionKey]; + if (!currentEntry?.sessionId) { + return { status: "missing-session" }; + } + const checkpoint = getSessionCompactionCheckpoint({ + entry: currentEntry, + checkpointId: params.checkpointId, + }); + if (!checkpoint) { + return { status: "missing-checkpoint" }; + } + const restoredSession = await forkCheckpointTranscriptFromStoredBoundary({ checkpoint }); + if (restoredSession.status !== "created") { + return restoredSession; + } + + const restoredTranscript = restoredSession.transcript; + const nextEntry = cloneCheckpointSessionEntry({ + currentEntry, + nextSessionId: restoredTranscript.sessionId, + nextSessionFile: restoredTranscript.sessionFile, + totalTokens: restoredTranscript.totalTokens, + preserveCompactionCheckpoints: true, + }); + store[params.sessionKey] = nextEntry; + return { + status: "created", + key: params.sessionKey, + checkpoint, + entry: nextEntry, + }; + }, + { skipSaveWhenResult: (result) => result.status !== "created" }, + ); +} + +/** + * Creates the current file-backed compaction checkpoint domain store. + * + * The branch/restore operations own the transcript fork plus session entry + * update so a SQLite implementation can copy transcript rows and update + * `session_entries.entry_json` inside one write transaction. + */ +export function createFileBackedCompactionCheckpointStore(): CompactionCheckpointStore { + return { + captureSnapshot: captureCompactionCheckpointSnapshotAsync, + persistCheckpoint: persistSessionCompactionCheckpoint, + cleanupSnapshot: cleanupCompactionCheckpointSnapshot, + branchCheckpointSession: branchCheckpointSessionFromStoredBoundary, + restoreCheckpointSession: restoreCheckpointSessionFromStoredBoundary, + }; +} + /** * Capture the stable pre-compaction identity without duplicating the transcript. * Branch/restore uses the compacted successor transcript, while legacy @@ -488,21 +772,9 @@ async function cleanupTrimmedCompactionCheckpointFiles(params: { } } -export async function persistSessionCompactionCheckpoint(params: { - cfg: OpenClawConfig; - sessionKey: string; - sessionId: string; - reason: SessionCompactionCheckpointReason; - snapshot: CapturedCompactionCheckpointSnapshot; - summary?: string; - firstKeptEntryId?: string; - tokensBefore?: number; - tokensAfter?: number; - postSessionFile?: string; - postLeafId?: string; - postEntryId?: string; - createdAt?: number; -}): Promise { +export async function persistSessionCompactionCheckpoint( + params: PersistSessionCompactionCheckpointParams, +): Promise { const snapshotSessionFile = params.snapshot.sessionFile?.trim(); const postSessionFile = params.postSessionFile?.trim(); const postSourceLeafId = params.postEntryId?.trim() || params.postLeafId?.trim(); diff --git a/src/talk/agent-consult-runtime.test.ts b/src/talk/agent-consult-runtime.test.ts index 6dcdaeed4835..5a4f3e87e9c1 100644 --- a/src/talk/agent-consult-runtime.test.ts +++ b/src/talk/agent-consult-runtime.test.ts @@ -1,6 +1,10 @@ -// Agent consult runtime tests cover consult session creation and runtime handoff. import { afterEach, describe, expect, it, vi } from "vitest"; import type { RunEmbeddedAgentParams } from "../agents/embedded-agent-runner/run/params.js"; +import type { + ForkSessionEntryFromParentParams, + ForkSessionEntryFromParentResult, +} from "../auto-reply/reply/session-fork.js"; +import type { SessionEntry } from "../config/sessions/types.js"; import { setRealtimeVoiceAgentConsultDepsForTest, consultRealtimeVoiceAgent, @@ -267,13 +271,47 @@ describe("realtime voice agent consult runtime", () => { maxTokens: 100_000, parentTokens: 100, })); - const forkSessionFromParent = vi.fn(async () => ({ - sessionId: "forked-session", - sessionFile: "/tmp/forked.jsonl", - })); + const forkSessionEntryFromParent = vi.fn( + async ( + params: ForkSessionEntryFromParentParams, + ): Promise => { + const fork = { + sessionId: "forked-session", + sessionFile: "/tmp/forked.jsonl", + }; + const parentEntry = sessionStore["agent:main:main"]; + if (!parentEntry?.sessionId) { + return { status: "missing-parent" }; + } + const typedParentEntry: SessionEntry = { + ...parentEntry, + sessionId: parentEntry.sessionId, + updatedAt: parentEntry.updatedAt ?? Date.now(), + }; + const decision = { + status: "fork" as const, + maxTokens: 100_000, + }; + const entry = params.fallbackEntry ?? { sessionId: "", updatedAt: Date.now() }; + const sessionEntry: SessionEntry = { + ...entry, + ...params.patch?.({ entry, parentEntry: typedParentEntry, fork, decision }), + sessionId: fork.sessionId, + sessionFile: fork.sessionFile, + forkedFromParent: true, + }; + sessionStore[params.sessionKey] = sessionEntry; + return { + status: "forked" as const, + fork, + parentEntry: typedParentEntry, + sessionEntry, + decision, + }; + }, + ); setRealtimeVoiceAgentConsultDepsForTest({ - resolveParentForkDecision, - forkSessionFromParent, + forkSessionEntryFromParent, }); await consultRealtimeVoiceAgent({ @@ -293,15 +331,16 @@ describe("realtime voice agent consult runtime", () => { userLabel: "Participant", }); - expect(resolveParentForkDecision).toHaveBeenCalledWith({ - parentEntry: sessionStore["agent:main:main"], - storePath: "/tmp/sessions.json", - }); - expect(forkSessionFromParent).toHaveBeenCalledWith({ - parentEntry: sessionStore["agent:main:main"], - agentId: "main", - sessionsDir: "/tmp", - }); + expect(resolveParentForkDecision).not.toHaveBeenCalled(); + expect(forkSessionEntryFromParent).toHaveBeenCalledWith( + expect.objectContaining({ + parentSessionKey: "agent:main:main", + agentId: "main", + config: {}, + sessionKey: "agent:main:subagent:google-meet:meet-1", + }), + ); + expect(runtime.session.patchSessionEntry).not.toHaveBeenCalled(); const forkedEntry = sessionStore["agent:main:subagent:google-meet:meet-1"]; if (!forkedEntry) { throw new Error("Expected forked consult session entry"); @@ -316,7 +355,75 @@ describe("realtime voice agent consult runtime", () => { expectPositiveTimestamp(forkedEntry.updatedAt); const call = requireEmbeddedAgentCall(runEmbeddedAgent); expect(call.sessionId).toBe("forked-session"); - expect(call.sessionFile).toBe("/tmp/forked.jsonl"); + expect(call.sessionFile).toBeUndefined(); + expect(call.sessionTarget).toMatchObject({ + agentId: "main", + sessionId: "forked-session", + sessionKey: "agent:main:subagent:google-meet:meet-1", + storePath: "/tmp/sessions.json", + }); + expect(call.spawnedBy).toBe("agent:main:main"); + }); + + it("falls back to a fresh isolated consult session when requester context is too large", async () => { + const { runtime, runEmbeddedAgent } = createAgentRuntime(); + const warn = vi.fn(); + const forkSessionEntryFromParent = vi.fn( + async ( + params: ForkSessionEntryFromParentParams, + ): Promise => ({ + status: "skipped", + reason: "decision-skip", + sessionEntry: { + ...(params.fallbackEntry ?? { sessionId: "", updatedAt: Date.now() }), + sessionId: "", + updatedAt: Date.now(), + }, + decision: { + status: "skip", + reason: "parent-too-large", + maxTokens: 100_000, + parentTokens: 150_000, + message: + "Parent context is too large to fork (150000/100000 tokens); starting with isolated context instead.", + }, + }), + ); + setRealtimeVoiceAgentConsultDepsForTest({ + forkSessionEntryFromParent, + randomUUID: () => "00000000-0000-4000-8000-000000000000", + }); + + await consultRealtimeVoiceAgent({ + cfg: {} as never, + agentRuntime: runtime as never, + logger: { warn }, + agentId: "main", + sessionKey: "agent:main:subagent:google-meet:meet-1", + spawnedBy: "agent:main:main", + contextMode: "fork", + messageProvider: "google-meet", + lane: "google-meet", + runIdPrefix: "google-meet:meet-1", + args: { question: "What should I say?" }, + transcript: [], + surface: "a private Google Meet", + userLabel: "Participant", + }); + + expect(warn).toHaveBeenCalledWith( + "[talk] Parent context is too large to fork (150000/100000 tokens); starting with isolated context instead.", + ); + expect(runtime.session.patchSessionEntry).toHaveBeenCalled(); + const call = requireEmbeddedAgentCall(runEmbeddedAgent); + expect(call.sessionId).toBe("00000000-0000-4000-8000-000000000000"); + expect(call.sessionFile).toBeUndefined(); + expect(call.sessionTarget).toMatchObject({ + agentId: "main", + sessionId: "00000000-0000-4000-8000-000000000000", + sessionKey: "agent:main:subagent:google-meet:meet-1", + storePath: "/tmp/sessions.json", + }); expect(call.spawnedBy).toBe("agent:main:main"); }); diff --git a/src/talk/agent-consult-runtime.ts b/src/talk/agent-consult-runtime.ts index d5b5809fcbdd..a7619bfbd434 100644 --- a/src/talk/agent-consult-runtime.ts +++ b/src/talk/agent-consult-runtime.ts @@ -1,11 +1,7 @@ // Agent consult runtime starts agent consultation flows from talk sessions. import { randomUUID } from "node:crypto"; -import path from "node:path"; import type { RunEmbeddedAgentParams } from "../agents/embedded-agent-runner/run/params.js"; -import { - forkSessionFromParent, - resolveParentForkDecision, -} from "../auto-reply/reply/session-fork.js"; +import { forkSessionEntryFromParent } from "../auto-reply/reply/session-fork.js"; import { parseSessionThreadInfoFast } from "../config/sessions/thread-info.js"; import type { SessionEntry } from "../config/sessions/types.js"; import type { OpenClawConfig } from "../config/types.openclaw.js"; @@ -44,14 +40,12 @@ export { type RealtimeVoiceAgentConsultDeps = { randomUUID: typeof randomUUID; - resolveParentForkDecision: typeof resolveParentForkDecision; - forkSessionFromParent: typeof forkSessionFromParent; + forkSessionEntryFromParent: typeof forkSessionEntryFromParent; }; const defaultRealtimeVoiceAgentConsultDeps: RealtimeVoiceAgentConsultDeps = { randomUUID, - resolveParentForkDecision, - forkSessionFromParent, + forkSessionEntryFromParent, }; let realtimeVoiceAgentConsultDeps = defaultRealtimeVoiceAgentConsultDeps; @@ -133,6 +127,7 @@ function resolveRealtimeVoiceAgentDeliveryContext(params: { async function resolveRealtimeVoiceAgentConsultSessionEntry(params: { agentId: string; + cfg: OpenClawConfig; sessionKey: string; spawnedBy?: string | null; contextMode?: RealtimeVoiceAgentConsultContextMode; @@ -151,7 +146,37 @@ async function resolveRealtimeVoiceAgentConsultSessionEntry(params: { (!requesterAgentId || requesterAgentId === params.agentId); let forkDecisionWarning: string | undefined; - const patched = await params.agentRuntime.session.patchSessionEntry({ + let patched: SessionEntry | null = null; + if (shouldFork) { + const forked = await realtimeVoiceAgentConsultDeps.forkSessionEntryFromParent({ + storePath: params.storePath, + parentSessionKey: requesterSessionKey, + agentId: params.agentId, + config: params.cfg, + sessionKey: params.sessionKey, + fallbackEntry: { + sessionId: "", + updatedAt: now, + }, + skipForkWhen: (entry) => Boolean(entry.sessionId?.trim()), + skipPatch: () => ({ ...deliveryFields, updatedAt: now }), + patch: () => ({ + ...deliveryFields, + spawnedBy: requesterSessionKey, + updatedAt: now, + }), + }); + if (forked.status === "forked" || forked.status === "skipped") { + if (forked.status === "skipped" && forked.decision?.status === "skip") { + forkDecisionWarning = forked.decision.message; + } + if (forked.sessionEntry.sessionId?.trim()) { + patched = forked.sessionEntry; + } + } + } + + patched ??= await params.agentRuntime.session.patchSessionEntry({ storePath: params.storePath, sessionKey: params.sessionKey, fallbackEntry: { @@ -162,39 +187,6 @@ async function resolveRealtimeVoiceAgentConsultSessionEntry(params: { if (entry.sessionId?.trim()) { return { ...deliveryFields, updatedAt: now }; } - // Fork only from same-agent requester sessions. Cross-agent parent sessions may carry - // incompatible provider state, so they get a fresh consult session with spawnedBy linkage. - if (shouldFork) { - const parentEntry = params.agentRuntime.session.getSessionEntry({ - storePath: params.storePath, - sessionKey: requesterSessionKey, - }); - if (parentEntry?.sessionId?.trim()) { - const decision = await realtimeVoiceAgentConsultDeps.resolveParentForkDecision({ - parentEntry, - storePath: params.storePath, - }); - if (decision.status === "fork") { - const fork = await realtimeVoiceAgentConsultDeps.forkSessionFromParent({ - parentEntry, - agentId: params.agentId, - sessionsDir: path.dirname(params.storePath), - }); - if (fork) { - return { - ...deliveryFields, - sessionId: fork.sessionId, - sessionFile: fork.sessionFile, - spawnedBy: requesterSessionKey, - forkedFromParent: true, - updatedAt: now, - }; - } - } else { - forkDecisionWarning = decision.message; - } - } - } return { ...deliveryFields, sessionId: realtimeVoiceAgentConsultDeps.randomUUID(), @@ -259,6 +251,7 @@ export async function consultRealtimeVoiceAgent(params: { }); const sessionEntry = await resolveRealtimeVoiceAgentConsultSessionEntry({ agentId, + cfg: params.cfg, sessionKey: params.sessionKey, spawnedBy: params.spawnedBy, contextMode: params.contextMode, @@ -271,14 +264,17 @@ export async function consultRealtimeVoiceAgent(params: { resolvedDeliveryContext ?? deliveryContextFromSession(sessionEntry); const sessionId = sessionEntry.sessionId; - const sessionFile = params.agentRuntime.session.resolveSessionFilePath(sessionId, sessionEntry, { - agentId, - }); // Voice consults suppress verbose/reasoning output because the bridge needs a short, // speakable answer, not agent-run diagnostics or hidden reasoning artifacts. const result = await params.agentRuntime.runEmbeddedAgent({ sessionId, sessionKey: params.sessionKey, + sessionTarget: { + agentId, + sessionId, + sessionKey: params.sessionKey, + storePath, + }, sandboxSessionKey: resolveRealtimeVoiceAgentSandboxSessionKey(agentId, params.sessionKey), agentId, spawnedBy: params.spawnedBy, @@ -291,7 +287,6 @@ export async function consultRealtimeVoiceAgent(params: { consultDeliveryContext?.threadId != null ? String(consultDeliveryContext.threadId) : undefined, - sessionFile, workspaceDir, config: params.cfg, prompt: buildRealtimeVoiceAgentConsultPrompt({