diff --git a/CHANGELOG.md b/CHANGELOG.md index d4c10e5dd004..eafd5b33e287 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,7 @@ Docs: https://docs.openclaw.ai ### Changes - Discord: allow configuring a bounded `agentComponents.ttlMs` callback registry lifetime for long-running component workflows, with per-account overrides and a 24-hour cap. (#84189) Thanks @100menotu001. +- Plugin SDK: add row-level session workflow helpers and deprecate `loadSessionStore` so plugins can read and patch sessions without depending on the legacy whole-store shape. (#84693) Thanks @efpiva. - Gateway/plugins: reuse a compatible Gateway startup plugin registry during dispatch so safe plugin dispatches avoid redundant registry loading. (#84324) Thanks @ai-hpc. - Dependencies: refresh provider, plugin, UI, and tooling packages, update `protobufjs` to 8.4.0 to clear the current npm advisory, and carry the Claude ACP completion patch forward to `@agentclientprotocol/claude-agent-acp` 0.36.1. - Agents/tools: remove the old sender-owner tool gating path so configured tools stay visible for trusted sessions while command and channel-action auth still carry real sender identity. diff --git a/docs/.generated/plugin-sdk-api-baseline.sha256 b/docs/.generated/plugin-sdk-api-baseline.sha256 index b64627548918..6e8efd3dc63c 100644 --- a/docs/.generated/plugin-sdk-api-baseline.sha256 +++ b/docs/.generated/plugin-sdk-api-baseline.sha256 @@ -1,2 +1,2 @@ -68bf012e03431fa1e29416489b829be3d2b6897f732d45de5b20fa30eb754119 plugin-sdk-api-baseline.json -7c6aa6ff2b42935228cc5cba1c2dd963af7534ac37daec3a5a8d0b9b3ba1620d plugin-sdk-api-baseline.jsonl +bb0da3ba4560521d2c9725cd96429f64ce8e6150972ba77be71fdf8ea03e0234 plugin-sdk-api-baseline.json +4d951b989cc00a86f64907bb28d52a950a466116382c9877f24807b0fba3df44 plugin-sdk-api-baseline.jsonl diff --git a/docs/plugins/sdk-runtime.md b/docs/plugins/sdk-runtime.md index fd172d991e4c..19ea06d970cc 100644 --- a/docs/plugins/sdk-runtime.md +++ b/docs/plugins/sdk-runtime.md @@ -153,16 +153,18 @@ two-party event loops that do not go through the shared channel-turn kernel. **Session store helpers** are under `api.runtime.agent.session`: ```typescript - const storePath = api.runtime.agent.session.resolveStorePath(cfg); - const store = api.runtime.agent.session.loadSessionStore(storePath); - await api.runtime.agent.session.updateSessionStore(storePath, (nextStore) => { - // Patch one entry without replacing the whole file from stale state. - nextStore[sessionKey] = { ...nextStore[sessionKey], thinkingLevel: "high" }; + const entry = api.runtime.agent.session.getSessionEntry({ agentId, sessionKey }); + for (const { sessionKey, entry } of api.runtime.agent.session.listSessionEntries({ agentId })) { + // Iterate session rows without depending on the legacy sessions.json shape. + } + await api.runtime.agent.session.patchSessionEntry({ + agentId, + sessionKey, + update: (entry) => ({ thinkingLevel: "high" }), }); - const filePath = api.runtime.agent.session.resolveSessionFilePath(cfg, sessionId); ``` - Prefer `updateSessionStore(...)` or `updateSessionStoreEntry(...)` for runtime writes. They route through the Gateway-owned session-store writer, preserve concurrent updates, and reuse the hot cache. `saveSessionStore(...)` remains available for compatibility and offline maintenance-style rewrites. + Prefer `getSessionEntry(...)`, `listSessionEntries(...)`, `patchSessionEntry(...)`, or `upsertSessionEntry(...)` for session workflows. These helpers address sessions by agent/session identity so plugins do not depend on the legacy `sessions.json` storage shape. Use `preserveActivity: true` for metadata-only patches that should not refresh session activity, and `replaceEntry: true` only when the callback returns a complete entry and deleted fields must stay deleted. `loadSessionStore(...)` remains as a deprecated compatibility escape hatch for callers that intentionally need a mutable whole-store clone. diff --git a/docs/plugins/sdk-subpaths.md b/docs/plugins/sdk-subpaths.md index 138de02a7682..0e14fd80c139 100644 --- a/docs/plugins/sdk-subpaths.md +++ b/docs/plugins/sdk-subpaths.md @@ -248,7 +248,7 @@ focused channel/runtime subpaths, `config-contracts`, `string-coerce-runtime`, | `plugin-sdk/reply-history` | Shared short-window reply-history helpers. New message-turn code should use `createChannelHistoryWindow`; lower-level map helpers remain deprecated compatibility exports only | | `plugin-sdk/reply-reference` | `createReplyReferencePlanner` | | `plugin-sdk/reply-chunking` | Narrow text/markdown chunking helpers | - | `plugin-sdk/session-store-runtime` | Session store path, session-key, updated-at, and store mutation helpers | + | `plugin-sdk/session-store-runtime` | Session workflow helpers (`getSessionEntry`, `listSessionEntries`, `patchSessionEntry`, `upsertSessionEntry`), legacy session store path/session-key helpers, updated-at reads, and deprecated whole-store mutation helpers | | `plugin-sdk/cron-store-runtime` | Cron store path/load/save helpers | | `plugin-sdk/state-paths` | State/OAuth dir path helpers | | `plugin-sdk/routing` | Route/session-key/account binding helpers such as `resolveAgentRoute`, `buildAgentSessionKey`, and `resolveDefaultAgentBoundAccountId` | diff --git a/extensions/active-memory/index.test.ts b/extensions/active-memory/index.test.ts index d413b413b8e1..46065cca1ca0 100644 --- a/extensions/active-memory/index.test.ts +++ b/extensions/active-memory/index.test.ts @@ -30,7 +30,10 @@ const hoisted = vi.hoisted(() => { closeActiveMemorySearchManager: vi.fn(async () => {}), sessionStore, updateSessionStore: vi.fn( - async (_storePath: string, updater: (store: Record) => void) => { + async ( + _storePath: string, + updater: (store: Record>) => void, + ) => { updater(sessionStore); }, ), @@ -113,6 +116,42 @@ describe("active-memory plugin", () => { resolveStorePath: vi.fn(() => "/tmp/openclaw-session-store.json"), loadSessionStore: vi.fn(() => hoisted.sessionStore), saveSessionStore: vi.fn(async () => {}), + getSessionEntry: vi.fn( + (params: { sessionKey: string }) => hoisted.sessionStore[params.sessionKey], + ), + listSessionEntries: vi.fn(() => + Object.entries(hoisted.sessionStore).map(([sessionKey, entry]) => ({ + sessionKey, + entry, + })), + ), + patchSessionEntry: vi.fn( + async (params: { + sessionKey: string; + fallbackEntry?: Record; + update: (entry: Record) => Record | null; + }) => { + let result: Record | null = null; + await hoisted.updateSessionStore( + "/tmp/openclaw-session-store.json", + (store: Record>) => { + const existing = store[params.sessionKey] ?? params.fallbackEntry; + if (!existing) { + return; + } + const patch = params.update({ ...existing }); + if (!patch) { + result = existing; + return; + } + const next = { ...existing, ...patch }; + store[params.sessionKey] = next; + result = next; + }, + ); + return result; + }, + ), }, }, state: { @@ -2961,7 +3000,7 @@ describe("active-memory plugin", () => { it("returns timeout within a hard deadline even when the subagent never checks the abort signal", async () => { const CONFIGURED_TIMEOUT_MS = 25; - const HARD_DEADLINE_MARGIN_MS = 500; + const HARD_DEADLINE_MARGIN_MS = 1_500; testing.setMinimumTimeoutMsForTests(1); testing.setSetupGraceTimeoutMsForTests(0); api.pluginConfig = { @@ -2999,6 +3038,7 @@ describe("active-memory plugin", () => { const CONFIGURED_TIMEOUT_MS = 50; testing.setMinimumTimeoutMsForTests(1); testing.setSetupGraceTimeoutMsForTests(0); + testing.setTimeoutPartialDataGraceMsForTests(50); api.pluginConfig = { agents: ["main"], timeoutMs: CONFIGURED_TIMEOUT_MS, @@ -3142,6 +3182,10 @@ describe("active-memory plugin", () => { timeoutMs: 100, }; plugin.register(api as unknown as OpenClawPluginApi); + hoisted.sessionStore["agent:main:memory-get-miss"] = { + sessionId: "s-memory-get-miss", + updatedAt: 0, + }; runEmbeddedPiAgent.mockImplementationOnce(async (params: { sessionFile: string }) => { await writeTranscriptJsonl(params.sessionFile, [ { diff --git a/extensions/active-memory/index.ts b/extensions/active-memory/index.ts index b342f9983927..7e2334dbd03c 100644 --- a/extensions/active-memory/index.ts +++ b/extensions/active-memory/index.ts @@ -20,10 +20,6 @@ import { import { definePluginEntry, type OpenClawPluginApi } from "openclaw/plugin-sdk/plugin-entry"; import { parseAgentSessionKey, parseThreadSessionSuffix } from "openclaw/plugin-sdk/routing"; import { isPathInside, replaceFileAtomic } from "openclaw/plugin-sdk/security-runtime"; -import { - resolveSessionStoreEntry, - updateSessionStore, -} from "openclaw/plugin-sdk/session-store-runtime"; import { tempWorkspace, resolvePreferredOpenClawTmpDir } from "openclaw/plugin-sdk/temp-path"; const DEFAULT_TIMEOUT_MS = 15_000; @@ -542,20 +538,15 @@ function resolveCanonicalSessionKeyFromSessionId(params: { return undefined; } try { - const storePath = params.api.runtime.agent.session.resolveStorePath( - params.api.config.session?.store, - { - agentId: params.agentId, - }, - ); - const store = params.api.runtime.agent.session.loadSessionStore(storePath, { clone: false }); let bestMatch: | { sessionKey: string; updatedAt: number; } | undefined; - for (const [sessionKey, entry] of Object.entries(store)) { + for (const { sessionKey, entry } of params.api.runtime.agent.session.listSessionEntries({ + agentId: params.agentId, + })) { if (!entry || typeof entry !== "object") { continue; } @@ -673,17 +664,10 @@ function resolveRecallRunChannelContext(params: { } try { - const storePath = params.api.runtime.agent.session.resolveStorePath( - params.api.config.session?.store, - { - agentId: params.agentId, - }, - ); - const store = params.api.runtime.agent.session.loadSessionStore(storePath, { clone: false }); - const sessionEntry = resolveSessionStoreEntry({ - store, + const sessionEntry = params.api.runtime.agent.session.getSessionEntry({ + agentId: params.agentId, sessionKey: resolvedSessionKey, - }).existing; + }); const rawStrongEntryChannel = normalizeOptionalString(sessionEntry?.lastChannel) ?? normalizeOptionalString(sessionEntry?.channel); @@ -1594,13 +1578,11 @@ async function persistPluginStatusLines(params: { return; } try { - const storePath = params.api.runtime.agent.session.resolveStorePath( - params.api.config.session?.store, - agentId ? { agentId } : undefined, - ); if (!params.statusLine && !debugLine) { - const store = params.api.runtime.agent.session.loadSessionStore(storePath, { clone: false }); - const existingEntry = resolveSessionStoreEntry({ store, sessionKey }).existing; + const existingEntry = params.api.runtime.agent.session.getSessionEntry({ + agentId, + sessionKey, + }); const hasActiveMemoryEntry = Array.isArray(existingEntry?.pluginDebugEntries) ? existingEntry.pluginDebugEntries.some((entry) => entry?.pluginId === "active-memory") : false; @@ -1608,39 +1590,38 @@ async function persistPluginStatusLines(params: { return; } } - await updateSessionStore(storePath, (store) => { - const resolved = resolveSessionStoreEntry({ store, sessionKey }); - const existing = resolved.existing; - if (!existing) { - return; - } - const previousEntries = Array.isArray(existing.pluginDebugEntries) - ? existing.pluginDebugEntries - : []; - const nextEntries = previousEntries.filter( - (entry): entry is PluginDebugEntry => - Boolean(entry) && - typeof entry === "object" && - typeof entry.pluginId === "string" && - entry.pluginId !== "active-memory", - ); - const nextLines: string[] = []; - if (params.statusLine) { - nextLines.push(params.statusLine); - } - if (debugLine) { - nextLines.push(debugLine); - } - if (nextLines.length > 0) { - nextEntries.push({ - pluginId: "active-memory", - lines: nextLines, - }); - } - store[resolved.normalizedKey] = { - ...existing, - pluginDebugEntries: nextEntries.length > 0 ? nextEntries : undefined, - }; + await params.api.runtime.agent.session.patchSessionEntry({ + agentId, + sessionKey, + preserveActivity: true, + update: (existing) => { + const previousEntries = Array.isArray(existing.pluginDebugEntries) + ? existing.pluginDebugEntries + : []; + const nextEntries = previousEntries.filter( + (entry): entry is PluginDebugEntry => + Boolean(entry) && + typeof entry === "object" && + typeof entry.pluginId === "string" && + entry.pluginId !== "active-memory", + ); + const nextLines: string[] = []; + if (params.statusLine) { + nextLines.push(params.statusLine); + } + if (debugLine) { + nextLines.push(debugLine); + } + if (nextLines.length > 0) { + nextEntries.push({ + pluginId: "active-memory", + lines: nextLines, + }); + } + return { + pluginDebugEntries: nextEntries.length > 0 ? nextEntries : undefined, + }; + }, }); } catch (error) { params.api.logger.debug?.( diff --git a/extensions/voice-call/src/response-generator.test.ts b/extensions/voice-call/src/response-generator.test.ts index 659dd23830ba..fe08eae04b01 100644 --- a/extensions/voice-call/src/response-generator.test.ts +++ b/extensions/voice-call/src/response-generator.test.ts @@ -9,6 +9,10 @@ type TestSessionEntry = { providerOverride?: string; modelOverride?: string; modelOverrideSource?: string; + model?: string; + modelProvider?: string; + contextTokens?: number; + authProfileOverride?: string; }; type EmbeddedAgentArgs = { @@ -32,6 +36,34 @@ function createAgentRuntime(payloads: Array>) { return await mutator(sessionStore); }, ); + const getSessionEntry = vi.fn( + (params: { sessionKey: string }) => sessionStore[params.sessionKey], + ); + const patchSessionEntry = vi.fn( + async (params: { + sessionKey: string; + fallbackEntry?: TestSessionEntry; + replaceEntry?: boolean; + update: (entry: TestSessionEntry) => Partial | null; + }) => { + const existing = sessionStore[params.sessionKey] ?? params.fallbackEntry; + if (!existing) { + return null; + } + const patch = params.update({ ...existing }); + if (!patch) { + return existing; + } + const next = params.replaceEntry ? (patch as TestSessionEntry) : { ...existing, ...patch }; + sessionStore[params.sessionKey] = next; + return next; + }, + ); + const upsertSessionEntry = vi.fn( + async (params: { sessionKey: string; entry: TestSessionEntry }) => { + sessionStore[params.sessionKey] = { ...params.entry }; + }, + ); const runEmbeddedPiAgent = vi.fn(async () => ({ payloads, meta: { durationMs: 12, aborted: false }, @@ -71,6 +103,9 @@ function createAgentRuntime(payloads: Array>) { loadSessionStore: () => sessionStore, saveSessionStore, updateSessionStore, + getSessionEntry, + patchSessionEntry, + upsertSessionEntry, resolveSessionFilePath, }, } as unknown as CoreAgentDeps; @@ -80,6 +115,7 @@ function createAgentRuntime(payloads: Array>) { runEmbeddedPiAgent, saveSessionStore, updateSessionStore, + patchSessionEntry, sessionStore, resolveAgentDir, resolveAgentWorkspaceDir, @@ -187,9 +223,17 @@ describe("generateVoiceResponse", () => { }); it("pins the voice session to responseModel before running the embedded agent", async () => { - const { runtime, runEmbeddedPiAgent, updateSessionStore, sessionStore } = createAgentRuntime([ + const { runtime, runEmbeddedPiAgent, patchSessionEntry, sessionStore } = createAgentRuntime([ { text: '{"spoken":"Pinned model works."}' }, ]); + sessionStore["voice:15550001111"] = { + sessionId: "existing-session", + updatedAt: 100, + model: "old-model", + modelProvider: "old-provider", + contextTokens: 123, + authProfileOverride: "old-auth-profile", + }; const voiceConfig = VoiceCallConfigSchema.parse({ responseModel: "openai/gpt-4.1-nano", responseTimeoutMs: 5000, @@ -210,12 +254,20 @@ describe("generateVoiceResponse", () => { expect(pinnedSessionEntry?.providerOverride).toBe("openai"); expect(pinnedSessionEntry?.modelOverride).toBe("gpt-4.1-nano"); expect(pinnedSessionEntry?.modelOverrideSource).toBe("auto"); - const updateSessionStoreCall = requireFirstMockCall( - updateSessionStore.mock.calls, - "session store update", + expect(pinnedSessionEntry?.model).toBeUndefined(); + expect(pinnedSessionEntry?.modelProvider).toBeUndefined(); + expect(pinnedSessionEntry?.contextTokens).toBeUndefined(); + expect(pinnedSessionEntry?.authProfileOverride).toBeUndefined(); + const patchSessionEntryCall = requireFirstMockCall( + patchSessionEntry.mock.calls, + "session entry patch", ); - expect(updateSessionStoreCall[0]).toBe("/tmp/openclaw/main/sessions.json"); - expect(updateSessionStoreCall[1]).toBeTypeOf("function"); + expect(patchSessionEntryCall[0]).toMatchObject({ + storePath: "/tmp/openclaw/main/sessions.json", + sessionKey: "voice:15550001111", + replaceEntry: true, + }); + expect((patchSessionEntryCall[0] as { update?: unknown }).update).toBeTypeOf("function"); const args = requireEmbeddedAgentArgs(runEmbeddedPiAgent); expect(args.provider).toBe("openai"); expect(args.model).toBe("gpt-4.1-nano"); diff --git a/extensions/voice-call/src/response-generator.ts b/extensions/voice-call/src/response-generator.ts index b34011508c4a..e32fd2168917 100644 --- a/extensions/voice-call/src/response-generator.ts +++ b/extensions/voice-call/src/response-generator.ts @@ -6,7 +6,6 @@ import crypto from "node:crypto"; import { applyModelOverrideToSessionEntry } from "openclaw/plugin-sdk/model-session-runtime"; import { normalizeLowercaseStringOrEmpty } from "openclaw/plugin-sdk/string-coerce-runtime"; -import type { SessionEntry } from "../api.js"; import { resolveVoiceCallSessionKey, type VoiceCallConfig } from "./config.js"; import type { CoreAgentDeps, CoreConfig } from "./core-bridge.js"; import { resolveVoiceResponseModel } from "./response-model.js"; @@ -251,34 +250,47 @@ export async function generateVoiceResponse( await agentRuntime.ensureAgentWorkspace({ dir: workspaceDir }); // Load or create session entry - const sessionStore = agentRuntime.session.loadSessionStore(storePath); const now = Date.now(); - const existingSessionEntry = sessionStore[resolvedSessionKey] as SessionEntry | undefined; + const existingSessionEntry = agentRuntime.session.getSessionEntry({ + storePath, + sessionKey: resolvedSessionKey, + }); // Resolve model from config const { provider, model } = resolveVoiceResponseModel({ voiceConfig, agentRuntime }); let sessionEntry = existingSessionEntry; if (!sessionEntry?.sessionId || voiceConfig.responseModel) { - sessionEntry = await agentRuntime.session.updateSessionStore(storePath, (store) => { - let entry = store[resolvedSessionKey] as SessionEntry | undefined; - if (!entry?.sessionId) { - entry = { - ...entry, + sessionEntry = + (await agentRuntime.session.patchSessionEntry({ + storePath, + sessionKey: resolvedSessionKey, + replaceEntry: true, + fallbackEntry: sessionEntry ?? { sessionId: crypto.randomUUID(), updatedAt: now, - }; - store[resolvedSessionKey] = entry; - } - if (voiceConfig.responseModel) { - applyModelOverrideToSessionEntry({ - entry, - selection: { provider, model }, - selectionSource: "auto", - }); - } - return entry; - }); + }, + update: (entry) => { + const next = entry.sessionId + ? { ...entry } + : { + ...entry, + sessionId: crypto.randomUUID(), + updatedAt: now, + }; + if (voiceConfig.responseModel) { + applyModelOverrideToSessionEntry({ + entry: next, + selection: { provider, model }, + selectionSource: "auto", + }); + } + return next; + }, + })) ?? undefined; + } + if (!sessionEntry?.sessionId) { + return { text: null, error: "Voice response session could not be initialized" }; } const sessionId = sessionEntry.sessionId; diff --git a/src/agents/agent-command.ts b/src/agents/agent-command.ts index 954abefdc3d5..263a92882df5 100644 --- a/src/agents/agent-command.ts +++ b/src/agents/agent-command.ts @@ -56,6 +56,7 @@ import { import { isStoredCredentialCompatibleWithAuthProvider } from "./auth-profiles/order.js"; import { clearSessionAuthProfileOverride } from "./auth-profiles/session-override.js"; import { ensureAuthProfileStore } from "./auth-profiles/store.js"; +import { createAgentAttemptLifecycleCallbacks } from "./command/attempt-callbacks.js"; import { persistSessionEntry as persistSessionEntryBase, prependInternalEventContext, @@ -1083,7 +1084,11 @@ async function agentCommandInternal( } const startedAt = Date.now(); - let lifecycleEnded = false; + const attemptLifecycleState = { + currentTurnUserMessagePersisted: false, + lifecycleEnded: false, + }; + const attemptLifecycleCallbacks = createAgentAttemptLifecycleCallbacks(attemptLifecycleState); const attemptExecutionRuntime = await loadAttemptExecutionRuntime(); const runContext = resolveAgentRunContext(opts); const messageChannel = resolveMessageChannel( @@ -1123,7 +1128,7 @@ async function agentCommandInternal( }); let fallbackAttemptIndex = 0; - let currentTurnUserMessagePersisted = false; + attemptLifecycleState.currentTurnUserMessagePersisted = false; const fallbackResult = await runWithModelFallback({ cfg, provider, @@ -1216,19 +1221,9 @@ async function agentCommandInternal( !isNewSession || (await attemptExecutionRuntime.sessionFileHasContent(sessionFile)), suppressPromptPersistenceOnRetry: opts.suppressPromptPersistence === true || - (isFallbackRetry && currentTurnUserMessagePersisted), - onUserMessagePersisted: () => { - currentTurnUserMessagePersisted = true; - }, - onAgentEvent: (evt) => { - if ( - evt.stream === "lifecycle" && - typeof evt.data?.phase === "string" && - (evt.data.phase === "end" || evt.data.phase === "error") - ) { - lifecycleEnded = true; - } - }, + (isFallbackRetry && attemptLifecycleState.currentTurnUserMessagePersisted), + onUserMessagePersisted: attemptLifecycleCallbacks.onUserMessagePersisted, + onAgentEvent: attemptLifecycleCallbacks.onAgentEvent, }); }, }); @@ -1290,7 +1285,7 @@ async function agentCommandInternal( }, }; } - if (!lifecycleEnded) { + if (!attemptLifecycleState.lifecycleEnded) { const stopReason = result.meta.stopReason; if (stopReason && stopReason !== "end_turn") { console.error(`[agent] run ${runId} ended with stopReason=${stopReason}`); @@ -1315,7 +1310,7 @@ async function agentCommandInternal( log.error( `Live session model switch in subagent run ${runId}: exceeded maximum retries (${MAX_LIVE_SWITCH_RETRIES})`, ); - if (!lifecycleEnded) { + if (!attemptLifecycleState.lifecycleEnded) { emitAgentEvent({ runId, stream: "lifecycle", @@ -1340,7 +1335,7 @@ async function agentCommandInternal( `Live session model switch in subagent run ${runId}: ` + `rejected ${sanitizeForLog(err.provider)}/${sanitizeForLog(err.model)} (not in allowlist)`, ); - if (!lifecycleEnded) { + if (!attemptLifecycleState.lifecycleEnded) { emitAgentEvent({ runId, stream: "lifecycle", @@ -1384,13 +1379,13 @@ async function agentCommandInternal( storedModelOverride = err.model; storedModelOverrideSource = "user"; } - lifecycleEnded = false; + attemptLifecycleState.lifecycleEnded = false; log.info( `Live session model switch in subagent run ${runId}: switching to ${sanitizeForLog(err.provider)}/${sanitizeForLog(err.model)}`, ); continue; } - if (!lifecycleEnded) { + if (!attemptLifecycleState.lifecycleEnded) { emitAgentEvent({ runId, stream: "lifecycle", diff --git a/src/agents/command/attempt-callbacks.test.ts b/src/agents/command/attempt-callbacks.test.ts new file mode 100644 index 000000000000..eea70aacc8ea --- /dev/null +++ b/src/agents/command/attempt-callbacks.test.ts @@ -0,0 +1,29 @@ +import { describe, expect, it } from "vitest"; +import { createAgentAttemptLifecycleCallbacks } from "./attempt-callbacks.js"; + +describe("createAgentAttemptLifecycleCallbacks", () => { + it("tracks user-message persistence without closing over the agent command scope", () => { + const state = { currentTurnUserMessagePersisted: false, lifecycleEnded: false }; + const callbacks = createAgentAttemptLifecycleCallbacks(state); + + callbacks.onUserMessagePersisted?.({ + role: "user", + content: "hello", + timestamp: Date.now(), + }); + + expect(state.currentTurnUserMessagePersisted).toBe(true); + expect(state.lifecycleEnded).toBe(false); + }); + + it("tracks terminal lifecycle phases", () => { + const state = { currentTurnUserMessagePersisted: false, lifecycleEnded: false }; + const callbacks = createAgentAttemptLifecycleCallbacks(state); + + callbacks.onAgentEvent({ stream: "lifecycle", data: { phase: "start" } }); + expect(state.lifecycleEnded).toBe(false); + + callbacks.onAgentEvent({ stream: "lifecycle", data: { phase: "end" } }); + expect(state.lifecycleEnded).toBe(true); + }); +}); diff --git a/src/agents/command/attempt-callbacks.ts b/src/agents/command/attempt-callbacks.ts new file mode 100644 index 000000000000..e86ee28da658 --- /dev/null +++ b/src/agents/command/attempt-callbacks.ts @@ -0,0 +1,32 @@ +import type { AgentMessage } from "@earendil-works/pi-agent-core"; + +export type AgentAttemptLifecycleState = { + currentTurnUserMessagePersisted: boolean; + lifecycleEnded: boolean; +}; + +export type AgentAttemptLifecycleEvent = { + stream: string; + data?: Record; + sessionKey?: string; +}; + +export function createAgentAttemptLifecycleCallbacks(state: AgentAttemptLifecycleState): { + onUserMessagePersisted: (message: Extract) => void; + onAgentEvent: (evt: AgentAttemptLifecycleEvent) => void; +} { + return { + onUserMessagePersisted: () => { + state.currentTurnUserMessagePersisted = true; + }, + onAgentEvent: (evt) => { + if ( + evt.stream === "lifecycle" && + typeof evt.data?.phase === "string" && + (evt.data.phase === "end" || evt.data.phase === "error") + ) { + state.lifecycleEnded = true; + } + }, + }; +} diff --git a/src/agents/subagent-announce-output.ts b/src/agents/subagent-announce-output.ts index 946443bd5690..1fbadee8cd18 100644 --- a/src/agents/subagent-announce-output.ts +++ b/src/agents/subagent-announce-output.ts @@ -8,7 +8,7 @@ import { import { callGateway, getRuntimeConfig, - loadSessionStore, + readSessionEntry, resolveAgentIdFromSessionKey, resolveStorePath, } from "./subagent-announce.runtime.js"; @@ -573,7 +573,7 @@ export async function buildCompactAnnounceStatsLine(params: { const cfg = subagentAnnounceOutputDeps.getRuntimeConfig(); const agentId = resolveAgentIdFromSessionKey(params.sessionKey); const storePath = resolveStorePath(cfg.session?.store, { agentId }); - let entry = loadSessionStore(storePath)[params.sessionKey]; + let entry = readSessionEntry(storePath, params.sessionKey); const tokenWaitAttempts = isFastTestMode() ? 1 : 3; for (let attempt = 0; attempt < tokenWaitAttempts; attempt += 1) { const hasTokenData = @@ -586,7 +586,7 @@ export async function buildCompactAnnounceStatsLine(params: { if (!isFastTestMode()) { await new Promise((resolve) => setTimeout(resolve, 150)); } - entry = loadSessionStore(storePath)[params.sessionKey]; + entry = readSessionEntry(storePath, params.sessionKey); } const input = typeof entry?.inputTokens === "number" ? entry.inputTokens : 0; diff --git a/src/agents/subagent-announce.runtime.ts b/src/agents/subagent-announce.runtime.ts index 56760afa7048..6c799ac79576 100644 --- a/src/agents/subagent-announce.runtime.ts +++ b/src/agents/subagent-announce.runtime.ts @@ -1,6 +1,7 @@ export { getRuntimeConfig } from "../config/config.js"; export { loadSessionStore, + readSessionEntry, resolveAgentIdFromSessionKey, resolveStorePath, } from "../config/sessions.js"; diff --git a/src/agents/subagent-announce.test.ts b/src/agents/subagent-announce.test.ts index caf61a337f3b..32f242a0ccf1 100644 --- a/src/agents/subagent-announce.test.ts +++ b/src/agents/subagent-announce.test.ts @@ -60,6 +60,8 @@ vi.mock("./subagent-announce.runtime.js", () => ({ isEmbeddedPiRunActive: (sessionId: string) => isEmbeddedPiRunActiveMock(sessionId), getRuntimeConfig: () => mockConfig, loadSessionStore: (storePath: string) => loadSessionStoreMock(storePath), + readSessionEntry: (storePath: string, sessionKey: string) => + (loadSessionStoreMock(storePath) as Record)[sessionKey], resolveAgentIdFromSessionKey: (sessionKey: string) => resolveAgentIdFromSessionKeyMock(sessionKey), resolveMainSessionKey: (cfg: unknown) => resolveMainSessionKeyMock(cfg), diff --git a/src/agents/subagent-announce.timeout.test.ts b/src/agents/subagent-announce.timeout.test.ts index 202846fee3f9..674e80a8a9f3 100644 --- a/src/agents/subagent-announce.timeout.test.ts +++ b/src/agents/subagent-announce.timeout.test.ts @@ -191,6 +191,7 @@ vi.mock("./subagent-announce.runtime.js", () => ({ }, getRuntimeConfig: () => configOverride, loadSessionStore: vi.fn(() => sessionStore), + readSessionEntry: (_storePath: string, sessionKey: string) => sessionStore[sessionKey], resolveAgentIdFromSessionKey: () => "main", resolveStorePath: () => "/tmp/sessions-main.json", resolveMainSessionKey: () => "agent:main:main", diff --git a/src/auto-reply/reply/dispatch-from-config.acp-abort.test.ts b/src/auto-reply/reply/dispatch-from-config.acp-abort.test.ts index 717b48a53e2d..a3bf1787d3d4 100644 --- a/src/auto-reply/reply/dispatch-from-config.acp-abort.test.ts +++ b/src/auto-reply/reply/dispatch-from-config.acp-abort.test.ts @@ -191,6 +191,7 @@ describe("dispatchReplyFromConfig ACP abort", () => { internalHookMocks.triggerInternalHook.mockReset(); sessionStoreMocks.currentEntry = undefined; sessionStoreMocks.loadSessionStore.mockReset().mockReturnValue({}); + sessionStoreMocks.readSessionEntry.mockReset().mockReturnValue(undefined); sessionStoreMocks.resolveStorePath.mockReset().mockReturnValue("/tmp/mock-sessions.json"); sessionStoreMocks.resolveSessionStoreEntry.mockReset().mockReturnValue({ existing: undefined }); acpMocks.listAcpSessionEntries.mockReset().mockResolvedValue([]); diff --git a/src/auto-reply/reply/dispatch-from-config.reply-dispatch.test.ts b/src/auto-reply/reply/dispatch-from-config.reply-dispatch.test.ts index 18c92f0c9a29..08dd1e21e834 100644 --- a/src/auto-reply/reply/dispatch-from-config.reply-dispatch.test.ts +++ b/src/auto-reply/reply/dispatch-from-config.reply-dispatch.test.ts @@ -83,6 +83,7 @@ describe("dispatchReplyFromConfig reply_dispatch hook", () => { sessionBindingMocks.touch.mockReset(); sessionStoreMocks.currentEntry = undefined; sessionStoreMocks.loadSessionStore.mockReset().mockReturnValue({}); + sessionStoreMocks.readSessionEntry.mockReset().mockReturnValue(undefined); sessionStoreMocks.resolveStorePath.mockReset().mockReturnValue("/tmp/mock-sessions.json"); sessionStoreMocks.resolveSessionStoreEntry.mockReset().mockReturnValue({ existing: undefined }); sessionStoreMocks.updateSessionStoreEntry.mockClear(); diff --git a/src/auto-reply/reply/dispatch-from-config.runtime.ts b/src/auto-reply/reply/dispatch-from-config.runtime.ts index 3ce5f38ff465..326f363c01fb 100644 --- a/src/auto-reply/reply/dispatch-from-config.runtime.ts +++ b/src/auto-reply/reply/dispatch-from-config.runtime.ts @@ -1,6 +1,7 @@ export { resolveStorePath } from "../../config/sessions/paths.js"; export { loadSessionStore, + readSessionEntry, resolveSessionStoreEntry, updateSessionStoreEntry, } from "../../config/sessions/store.js"; diff --git a/src/auto-reply/reply/dispatch-from-config.shared.test-harness.ts b/src/auto-reply/reply/dispatch-from-config.shared.test-harness.ts index 6d3395c3421c..59213c782531 100644 --- a/src/auto-reply/reply/dispatch-from-config.shared.test-harness.ts +++ b/src/auto-reply/reply/dispatch-from-config.shared.test-harness.ts @@ -92,6 +92,7 @@ const pluginConversationBindingMocks = vi.hoisted(() => ({ const sessionStoreMocks = vi.hoisted(() => ({ currentEntry: undefined as Record | undefined, loadSessionStore: vi.fn(() => ({})), + readSessionEntry: vi.fn(() => sessionStoreMocks.currentEntry), resolveStorePath: vi.fn(() => "/tmp/mock-sessions.json"), resolveSessionStoreEntry: vi.fn(() => ({ existing: sessionStoreMocks.currentEntry })), updateSessionStoreEntry: vi.fn( @@ -209,6 +210,7 @@ vi.mock("../../config/sessions/thread-info.js", () => ({ vi.mock("./dispatch-from-config.runtime.js", () => ({ createInternalHookEvent: internalHookMocks.createInternalHookEvent, loadSessionStore: sessionStoreMocks.loadSessionStore, + readSessionEntry: sessionStoreMocks.readSessionEntry, resolveSessionStoreEntry: sessionStoreMocks.resolveSessionStoreEntry, resolveStorePath: sessionStoreMocks.resolveStorePath, triggerInternalHook: internalHookMocks.triggerInternalHook, diff --git a/src/auto-reply/reply/dispatch-from-config.test.ts b/src/auto-reply/reply/dispatch-from-config.test.ts index 4ad249bc54cc..69f8080b9e24 100644 --- a/src/auto-reply/reply/dispatch-from-config.test.ts +++ b/src/auto-reply/reply/dispatch-from-config.test.ts @@ -110,6 +110,7 @@ const pluginConversationBindingMocks = vi.hoisted(() => ({ const sessionStoreMocks = vi.hoisted(() => ({ currentEntry: undefined as Record | undefined, loadSessionStore: vi.fn(() => ({})), + readSessionEntry: vi.fn(() => sessionStoreMocks.currentEntry), resolveStorePath: vi.fn(() => "/tmp/mock-sessions.json"), resolveSessionStoreEntry: vi.fn(() => ({ existing: sessionStoreMocks.currentEntry })), updateSessionStoreEntry: vi.fn( @@ -381,6 +382,7 @@ vi.mock("../../config/sessions/thread-info.js", () => ({ vi.mock("./dispatch-from-config.runtime.js", () => ({ createInternalHookEvent: internalHookMocks.createInternalHookEvent, loadSessionStore: sessionStoreMocks.loadSessionStore, + readSessionEntry: sessionStoreMocks.readSessionEntry, resolveSessionStoreEntry: sessionStoreMocks.resolveSessionStoreEntry, resolveStorePath: sessionStoreMocks.resolveStorePath, triggerInternalHook: internalHookMocks.triggerInternalHook, @@ -880,6 +882,8 @@ describe("dispatchReplyFromConfig", () => { sessionBindingMocks.touch.mockReset(); sessionStoreMocks.currentEntry = undefined; sessionStoreMocks.loadSessionStore.mockClear(); + sessionStoreMocks.readSessionEntry.mockReset(); + sessionStoreMocks.readSessionEntry.mockImplementation(() => sessionStoreMocks.currentEntry); sessionStoreMocks.resolveStorePath.mockClear(); sessionStoreMocks.resolveSessionStoreEntry.mockClear(); threadInfoMocks.parseSessionThreadInfo.mockReset(); @@ -1877,6 +1881,55 @@ describe("dispatchReplyFromConfig", () => { expect(dispatcher.sendFinalReply).toHaveBeenCalledWith({ text: "done" }); }); + it("refreshes verbose progress with session entry snapshots", async () => { + setNoAbort(); + sessionStoreMocks.currentEntry = { + verboseLevel: "on", + }; + sessionStoreMocks.readSessionEntry.mockReturnValue({ verboseLevel: "off" }); + const cfg = { + ...emptyConfig, + agents: { + defaults: { + verboseDefault: "on", + }, + }, + } satisfies OpenClawConfig; + const dispatcher = createDispatcher(); + const ctx = buildTestCtx({ + Provider: "telegram", + ChatType: "direct", + SessionKey: "agent:main:main", + }); + + const replyResolver = async ( + _ctx: MsgContext, + opts?: GetReplyOptions, + _cfg?: OpenClawConfig, + ) => { + sessionStoreMocks.loadSessionStore.mockClear(); + sessionStoreMocks.resolveSessionStoreEntry.mockClear(); + sessionStoreMocks.readSessionEntry.mockClear(); + await opts?.onPlanUpdate?.({ + phase: "update", + explanation: "Inspect code, patch it, run tests.", + steps: ["Inspect code", "Patch code", "Run tests"], + }); + return { text: "done" } satisfies ReplyPayload; + }; + + await dispatchReplyFromConfig({ ctx, cfg, dispatcher, replyResolver }); + + expect(sessionStoreMocks.readSessionEntry).toHaveBeenCalledWith( + "/tmp/mock-sessions.json", + "agent:main:main", + ); + expect(sessionStoreMocks.loadSessionStore).not.toHaveBeenCalled(); + expect(sessionStoreMocks.resolveSessionStoreEntry).not.toHaveBeenCalled(); + expect(dispatcher.sendToolResult).not.toHaveBeenCalled(); + expect(dispatcher.sendFinalReply).toHaveBeenCalledWith({ text: "done" }); + }); + it("suppresses text-only tool summaries when preview tool-progress suppression is enabled", async () => { setNoAbort(); const cfg = emptyConfig; diff --git a/src/auto-reply/reply/dispatch-from-config.ts b/src/auto-reply/reply/dispatch-from-config.ts index 4bab0ebe4231..5abc67f69879 100644 --- a/src/auto-reply/reply/dispatch-from-config.ts +++ b/src/auto-reply/reply/dispatch-from-config.ts @@ -107,6 +107,7 @@ import { resolveConversationBindingContextFromMessage } from "./conversation-bin import { createInternalHookEvent, loadSessionStore, + readSessionEntry, resolveSessionStoreEntry, resolveStorePath, triggerInternalHook, @@ -351,8 +352,7 @@ const createShouldEmitVerboseProgress = (params: { const resolveLevel = () => { if (params.sessionKey && params.storePath) { try { - const store = loadSessionStore(params.storePath); - const entry = resolveSessionStoreEntry({ store, sessionKey: params.sessionKey }).existing; + const entry = readSessionEntry(params.storePath, params.sessionKey); const currentLevel = normalizeVerboseLevel(entry?.verboseLevel ?? ""); if (currentLevel) { return currentLevel; diff --git a/src/config/sessions.cache.test.ts b/src/config/sessions.cache.test.ts index 9e92f6459b5a..2c0fc459c2a9 100644 --- a/src/config/sessions.cache.test.ts +++ b/src/config/sessions.cache.test.ts @@ -2,11 +2,23 @@ import fs from "node:fs"; import path from "node:path"; import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest"; import { createSuiteTempRootTracker } from "../test-helpers/temp-dir.js"; -import { readSessionStoreCache, writeSessionStoreCache } from "./sessions/store-cache.js"; +import { + getSerializedSessionStore, + getSerializedSessionStoreCacheStatsForTest, + getSessionStoreStringInternStatsForTest, + readSessionStoreCache, + setSerializedSessionStore, + writeSessionStoreCache, +} from "./sessions/store-cache.js"; import { clearSessionStoreCacheForTest, loadSessionStore, + readSessionEntries, + readSessionEntry, + readSessionStoreSnapshot, + readSessionUpdatedAt, saveSessionStore, + updateSessionStore, } from "./sessions/store.js"; import type { SessionEntry } from "./sessions/types.js"; @@ -53,6 +65,35 @@ describe("Session Store Cache", () => { afterEach(() => { clearSessionStoreCacheForTest(); delete process.env.OPENCLAW_SESSION_CACHE_TTL_MS; + delete process.env.OPENCLAW_SESSION_SERIALIZED_CACHE_MAX_BYTES; + }); + + it("bounds the serialized session store cache by total bytes", () => { + process.env.OPENCLAW_SESSION_SERIALIZED_CACHE_MAX_BYTES = "64"; + clearSessionStoreCacheForTest(); + + setSerializedSessionStore("store:1", "a".repeat(40)); + setSerializedSessionStore("store:2", "b".repeat(40)); + + expect(getSerializedSessionStore("store:1")).toBeUndefined(); + expect(getSerializedSessionStore("store:2")).toBe("b".repeat(40)); + expect(getSerializedSessionStoreCacheStatsForTest().entries).toBe(1); + expect(getSerializedSessionStoreCacheStatsForTest().totalBytes).toBe(40); + }); + + it("bounds the serialized session store cache by path count", () => { + const maxEntries = getSerializedSessionStoreCacheStatsForTest().maxEntries; + + for (let index = 0; index < maxEntries + 2; index += 1) { + setSerializedSessionStore(`store:${index}`, `serialized:${index}`); + } + + expect(getSerializedSessionStore("store:0")).toBeUndefined(); + expect(getSerializedSessionStore("store:1")).toBeUndefined(); + expect(getSerializedSessionStore(`store:${maxEntries + 1}`)).toBe( + `serialized:${maxEntries + 1}`, + ); + expect(getSerializedSessionStoreCacheStatsForTest().entries).toBe(maxEntries); }); it("should load session store from disk on first call", async () => { @@ -247,6 +288,190 @@ describe("Session Store Cache", () => { stringifySpy.mockRestore(); }); + it("interns duplicate large skillsSnapshot prompts across cached loads", async () => { + const largePrompt = "skill prompt ".repeat(200); + const testStore = { + "session:1": createSessionEntry({ + skillsSnapshot: { + prompt: largePrompt, + skills: [{ name: "alpha" }], + }, + }), + "session:2": createSessionEntry({ + sessionId: "id-2", + displayName: "Test Session 2", + skillsSnapshot: { + prompt: largePrompt, + skills: [{ name: "beta" }], + }, + }), + }; + + await saveSessionStore(storePath, testStore); + clearSessionStoreCacheForTest(); + + const loaded1 = loadSessionStore(storePath); + const afterFirstLoad = getSessionStoreStringInternStatsForTest(); + expect(afterFirstLoad.poolSize).toBe(1); + expect(afterFirstLoad.stored).toBe(1); + expect(afterFirstLoad.reused).toBeGreaterThanOrEqual(1); + + if (loaded1["session:1"].skillsSnapshot?.skills?.length) { + loaded1["session:1"].skillsSnapshot.skills[0].name = "mutated"; + } + + const loaded2 = loadSessionStore(storePath); + const afterSecondLoad = getSessionStoreStringInternStatsForTest(); + expect(afterSecondLoad.poolSize).toBe(1); + expect(afterSecondLoad.reused).toBeGreaterThanOrEqual(afterFirstLoad.reused + 2); + expect(loaded2["session:1"].skillsSnapshot?.skills?.[0]?.name).toBe("alpha"); + }); + + it("does not intern short skillsSnapshot prompts", async () => { + const testStore = { + "session:1": createSessionEntry({ + skillsSnapshot: { + prompt: "short prompt", + skills: [{ name: "alpha" }], + }, + }), + "session:2": createSessionEntry({ + sessionId: "id-2", + displayName: "Test Session 2", + skillsSnapshot: { + prompt: "short prompt", + skills: [{ name: "beta" }], + }, + }), + }; + + await saveSessionStore(storePath, testStore); + clearSessionStoreCacheForTest(); + + loadSessionStore(storePath); + + const stats = getSessionStoreStringInternStatsForTest(); + expect(stats.poolSize).toBe(0); + expect(stats.skippedSmall).toBeGreaterThanOrEqual(2); + }); + + it("reads updatedAt from immutable session snapshots without cloning cached stores", async () => { + const updatedAt = Date.now(); + const testStore = createSingleSessionStore( + createSessionEntry({ + updatedAt, + }), + "agent:main:main", + ); + + await saveSessionStore(storePath, testStore); + clearSessionStoreCacheForTest(); + readSessionStoreSnapshot(storePath); + expect(readSessionEntry(storePath, "agent:main:main")?.updatedAt).toBe(updatedAt); + + const parseSpy = vi.spyOn(JSON, "parse"); + + expect(readSessionUpdatedAt({ storePath, sessionKey: "agent:main:main" })).toBe(updatedAt); + expect(parseSpy).not.toHaveBeenCalled(); + + parseSpy.mockRestore(); + }); + + it("serves immutable session snapshots without cloning cache hits", async () => { + const testStore = createSingleSessionStore( + createSessionEntry({ + origin: { provider: "openai" }, + skillsSnapshot: { + prompt: "snapshot skill prompt ".repeat(200), + skills: [{ name: "alpha" }], + }, + }), + ); + + await saveSessionStore(storePath, testStore); + clearSessionStoreCacheForTest(); + + const snapshot1 = readSessionStoreSnapshot(storePath); + const snapshot2 = readSessionStoreSnapshot(storePath); + + expect(snapshot2).toBe(snapshot1); + expect(Object.isFrozen(snapshot1)).toBe(true); + expect(Object.isFrozen(snapshot1["session:1"])).toBe(true); + expect(Object.isFrozen(snapshot1["session:1"].skillsSnapshot?.skills)).toBe(true); + expect(readSessionEntry(storePath, "session:1")?.sessionId).toBe("id-1"); + expect(readSessionEntries(storePath).map(([key]) => key)).toEqual(["session:1"]); + + expect(() => { + (snapshot1 as Record)["session:2"] = createSessionEntry({ + sessionId: "id-2", + }); + }).toThrow(TypeError); + + const mutable = loadSessionStore(storePath); + mutable["session:1"].origin = { provider: "mutated" }; + + expect(readSessionStoreSnapshot(storePath)["session:1"].origin?.provider).toBe("openai"); + }); + + it("does not tag snapshots with stats from writes racing after a disk read", async () => { + await saveSessionStore( + storePath, + createSingleSessionStore(createSessionEntry({ displayName: "Before race" })), + ); + clearSessionStoreCacheForTest(); + + const afterRaceStore = createSingleSessionStore( + createSessionEntry({ displayName: "After cross-process race" }), + ); + const originalReadFileSync = fs.readFileSync.bind(fs); + let wroteAfterRead = false; + const readSpy = vi.spyOn(fs, "readFileSync").mockImplementation((file, ...args) => { + const result = originalReadFileSync( + file, + ...(args as [Parameters[1]]), + ); + if (file === storePath && !wroteAfterRead) { + wroteAfterRead = true; + fs.writeFileSync(storePath, JSON.stringify(afterRaceStore, null, 2)); + const bumped = new Date(Date.now() + 2_000); + fs.utimesSync(storePath, bumped, bumped); + } + return result; + }); + + const first = readSessionStoreSnapshot(storePath); + expect(first["session:1"].displayName).toBe("Before race"); + + readSpy.mockRestore(); + + const second = readSessionStoreSnapshot(storePath); + expect(second["session:1"].displayName).toBe("After cross-process race"); + }); + + it("publishes a new immutable snapshot after session store writes", async () => { + await saveSessionStore(storePath, createSingleSessionStore()); + + const before = readSessionStoreSnapshot(storePath); + + await updateSessionStore( + storePath, + (store) => { + store["session:1"] = { + ...store["session:1"], + displayName: "Updated Session", + updatedAt: Date.now() + 1, + }; + }, + { skipMaintenance: true }, + ); + + const after = readSessionStoreSnapshot(storePath); + + expect(after).not.toBe(before); + expect(before["session:1"].displayName).toBe("Test Session 1"); + expect(after["session:1"].displayName).toBe("Updated Session"); + }); + it("should refresh cache when store file changes on disk", async () => { const testStore = createSingleSessionStore(); diff --git a/src/config/sessions.test.ts b/src/config/sessions.test.ts index 7b5d8c1be1ca..3e3fab2073dc 100644 --- a/src/config/sessions.test.ts +++ b/src/config/sessions.test.ts @@ -8,6 +8,7 @@ import { buildGroupDisplayName, deriveSessionKey, loadSessionStore, + patchSessionEntry, resolveSessionFilePath, resolveSessionFilePathOptions, resolveSessionKey, @@ -16,6 +17,7 @@ import { updateLastRoute, updateSessionStore, updateSessionStoreEntry, + upsertSessionEntry, } from "./sessions.js"; describe("sessions", () => { @@ -506,6 +508,102 @@ describe("sessions", () => { expect(store[sessionKey]?.thinkingLevel).toBe("low"); }); + it("patchSessionEntry can preserve activity for metadata-only updates", async () => { + const sessionKey = "agent:main:main"; + const { storePath } = await createSessionStoreFixture({ + prefix: "patchSessionEntry-preserve-activity", + entries: { + [sessionKey]: { + sessionId: "sess-1", + updatedAt: 100, + pluginDebugEntries: [{ pluginId: "other", lines: ["keep"] }], + }, + }, + }); + + await patchSessionEntry({ + storePath, + sessionKey, + preserveActivity: true, + update: () => ({ + pluginDebugEntries: [{ pluginId: "active-memory", lines: ["status"] }], + }), + }); + + const store = loadSessionStore(storePath); + expect(store[sessionKey]?.updatedAt).toBe(100); + expect(store[sessionKey]?.pluginDebugEntries).toEqual([ + { pluginId: "active-memory", lines: ["status"] }, + ]); + }); + + it("patchSessionEntry can replace an entry so deleted fields stay deleted", async () => { + const sessionKey = "agent:main:main"; + const { storePath } = await createSessionStoreFixture({ + prefix: "patchSessionEntry-replace-entry", + entries: { + [sessionKey]: { + sessionId: "sess-1", + updatedAt: 100, + model: "old-model", + modelProvider: "old-provider", + }, + }, + }); + + await patchSessionEntry({ + storePath, + sessionKey, + replaceEntry: true, + update: (entry) => { + const next = { ...entry, providerOverride: "openai" }; + delete next.model; + delete next.modelProvider; + return next; + }, + }); + + const store = loadSessionStore(storePath); + expect(store[sessionKey]?.providerOverride).toBe("openai"); + expect(store[sessionKey]?.model).toBeUndefined(); + expect(store[sessionKey]?.modelProvider).toBeUndefined(); + }); + + it("upsertSessionEntry preserves existing ACP metadata by default", async () => { + const sessionKey = "agent:main:main"; + const acp = { + backend: "codex", + agent: "main", + runtimeSessionName: "runtime-session", + mode: "persistent" as const, + state: "idle" as const, + lastActivityAt: 100, + }; + const { storePath } = await createSessionStoreFixture({ + prefix: "upsertSessionEntry-acp", + entries: { + [sessionKey]: { + sessionId: "sess-1", + updatedAt: 100, + acp, + }, + }, + }); + + await upsertSessionEntry({ + storePath, + sessionKey, + entry: { + sessionId: "sess-2", + updatedAt: 200, + }, + }); + + const store = loadSessionStore(storePath); + expect(store[sessionKey]?.sessionId).toBe("sess-2"); + expect(store[sessionKey]?.acp).toStrictEqual(acp); + }); + it("updateSessionStore preserves concurrent additions", async () => { const dir = await createCaseDir("updateSessionStore"); const storePath = path.join(dir, "sessions.json"); diff --git a/src/config/sessions/delivery-info.test.ts b/src/config/sessions/delivery-info.test.ts index 1570e13a0e3d..e33b67c612df 100644 --- a/src/config/sessions/delivery-info.test.ts +++ b/src/config/sessions/delivery-info.test.ts @@ -3,10 +3,15 @@ import { setActivePluginRegistry } from "../../plugins/runtime.js"; import { createSessionConversationTestRegistry } from "../../test-utils/session-conversation-registry.js"; import type { SessionEntry } from "./types.js"; -const storeState = vi.hoisted(() => ({ - store: {} as Record, - stores: {} as Record>, -})); +const storeState = vi.hoisted(() => { + const state = { + store: {} as Record, + stores: {} as Record>, + loadSessionStore: vi.fn((storePath: string) => state.stores[storePath] ?? state.store), + readSessionStoreSnapshot: vi.fn((storePath: string) => state.stores[storePath] ?? state.store), + }; + return state; +}); vi.mock("../io.js", () => ({ getRuntimeConfig: () => ({}), @@ -18,7 +23,8 @@ vi.mock("./paths.js", () => ({ })); vi.mock("./store.js", () => ({ - loadSessionStore: (storePath: string) => storeState.stores[storePath] ?? storeState.store, + loadSessionStore: storeState.loadSessionStore, + readSessionStoreSnapshot: storeState.readSessionStoreSnapshot, })); vi.mock("./targets.js", () => ({ @@ -46,6 +52,8 @@ beforeEach(() => { setActivePluginRegistry(createSessionConversationTestRegistry()); storeState.store = {}; storeState.stores = {}; + storeState.loadSessionStore.mockClear(); + storeState.readSessionStoreSnapshot.mockClear(); }); describe("extractDeliveryInfo", () => { @@ -85,6 +93,21 @@ describe("extractDeliveryInfo", () => { }); }); + it("uses session-store snapshots for direct session keys", () => { + const sessionKey = "agent:main:webchat:dm:user-123"; + storeState.store[sessionKey] = buildEntry({ + channel: "webchat", + to: "webchat:user-123", + accountId: "default", + }); + + const result = extractDeliveryInfo(sessionKey); + + expect(result.deliveryContext?.to).toBe("webchat:user-123"); + expect(storeState.readSessionStoreSnapshot).toHaveBeenCalledWith("/tmp/sessions.json"); + expect(storeState.loadSessionStore).not.toHaveBeenCalled(); + }); + it("returns deliveryContext for direct session keys", () => { const sessionKey = "agent:main:webchat:dm:user-123"; storeState.store[sessionKey] = buildEntry({ diff --git a/src/config/sessions/delivery-info.ts b/src/config/sessions/delivery-info.ts index f4462680b4eb..defcde9efd73 100644 --- a/src/config/sessions/delivery-info.ts +++ b/src/config/sessions/delivery-info.ts @@ -8,7 +8,7 @@ import { getRuntimeConfig } from "../io.js"; import type { OpenClawConfig } from "../types.openclaw.js"; import { resolveStorePath } from "./paths.js"; import { normalizeStoreSessionKey } from "./store-entry.js"; -import { loadSessionStore } from "./store.js"; +import { readSessionStoreSnapshot } from "./store.js"; import { resolveAllAgentSessionStoreTargetsSync } from "./targets.js"; import { parseSessionThreadInfo } from "./thread-info.js"; import type { SessionEntry } from "./types.js"; @@ -79,26 +79,31 @@ function resolveDeliveryStorePaths(cfg: OpenClawConfig, agentId: string): string return [...paths]; } +function asSessionEntry(entry: unknown): SessionEntry | undefined { + return entry as SessionEntry | undefined; +} + function findSessionEntryInStore( - store: ReturnType, + store: ReturnType, keys: readonly string[], ) { let normalizedIndex: Map | undefined; let bestEntry: SessionEntry | undefined; let bestUpdatedAt = 0; let bestRoutable = false; - const acceptCandidate = (candidate: SessionEntry | undefined) => { + const acceptCandidate = (candidate: unknown) => { if (!candidate) { return; } - const candidateRoutable = hasRoutableDeliveryContext(deliveryContextFromSession(candidate)); - const candidateUpdatedAt = candidate.updatedAt ?? 0; + const entry = candidate as SessionEntry; + const candidateRoutable = hasRoutableDeliveryContext(deliveryContextFromSession(entry)); + const candidateUpdatedAt = entry.updatedAt ?? 0; if ( !bestEntry || (candidateRoutable && !bestRoutable) || (candidateRoutable === bestRoutable && candidateUpdatedAt > bestUpdatedAt) ) { - bestEntry = candidate; + bestEntry = entry; bestUpdatedAt = candidateUpdatedAt; bestRoutable = candidateRoutable; } @@ -110,7 +115,7 @@ function findSessionEntryInStore( let foundRoutableCandidate = false; if (Object.prototype.hasOwnProperty.call(store, normalized)) { foundRoutableCandidate ||= hasRoutableDeliveryContext( - deliveryContextFromSession(store[normalized]), + deliveryContextFromSession(asSessionEntry(store[normalized])), ); acceptCandidate(store[normalized]); } @@ -119,13 +124,13 @@ function findSessionEntryInStore( Object.prototype.hasOwnProperty.call(store, foldedLegacyKey) ) { foundRoutableCandidate ||= hasRoutableDeliveryContext( - deliveryContextFromSession(store[foldedLegacyKey]), + deliveryContextFromSession(asSessionEntry(store[foldedLegacyKey])), ); acceptCandidate(store[foldedLegacyKey]); } if (trimmed !== normalized && Object.prototype.hasOwnProperty.call(store, trimmed)) { foundRoutableCandidate ||= hasRoutableDeliveryContext( - deliveryContextFromSession(store[trimmed]), + deliveryContextFromSession(asSessionEntry(store[trimmed])), ); acceptCandidate(store[trimmed]); } @@ -142,10 +147,14 @@ function findSessionEntryInStore( } function buildFreshestSessionEntryIndex( - store: Record, + store: Readonly>, ): Map { const index = new Map(); - for (const [key, entry] of Object.entries(store)) { + for (const [key, candidate] of Object.entries(store)) { + const entry = asSessionEntry(candidate); + if (!entry) { + continue; + } const normalized = normalizeStoreSessionKey(key); const existing = index.get(normalized); const entryRoutable = hasRoutableDeliveryContext(deliveryContextFromSession(entry)); @@ -200,7 +209,7 @@ function loadDeliverySessionEntry(params: { } | undefined; for (const storePath of resolveDeliveryStorePaths(params.cfg, agentId)) { - const store = loadSessionStore(storePath); + const store = readSessionStoreSnapshot(storePath); const entry = findSessionEntryInStore(store, sessionKeys); const baseEntry = findSessionEntryInStore(store, baseKeys); if (!entry && !baseEntry) { diff --git a/src/config/sessions/store-cache.ts b/src/config/sessions/store-cache.ts index 6e27b9e38566..1bf43199fa1b 100644 --- a/src/config/sessions/store-cache.ts +++ b/src/config/sessions/store-cache.ts @@ -1,6 +1,22 @@ import { createExpiringMapCache, isCacheEnabled, resolveCacheTtlMs } from "../cache-utils.js"; import type { SessionEntry } from "./types.js"; +export type DeepReadonly = T extends (...args: never[]) => unknown + ? T + : T extends readonly (infer U)[] + ? ReadonlyArray> + : T extends object + ? { readonly [K in keyof T]: DeepReadonly } + : T; + +export type SessionStoreSnapshot = DeepReadonly>; + +export type SessionStoreSnapshotEntry = DeepReadonly; + +export type SessionStoreSnapshotEntries = ReadonlyArray< + readonly [string, SessionStoreSnapshotEntry] +>; + type SessionStoreCacheEntry = { store: Record; mtimeMs?: number; @@ -8,18 +24,190 @@ type SessionStoreCacheEntry = { serialized?: string; }; +type SessionStoreSnapshotCacheEntry = { + snapshot: SessionStoreSnapshot; + mtimeMs?: number; + sizeBytes?: number; + generation: number; + createdAt: number; + entryCount: number; +}; + +type SerializedSessionStoreCacheEntry = { + serialized: string; + sizeBytes: number; +}; + const DEFAULT_SESSION_STORE_TTL_MS = 45_000; // 45 seconds (between 30-60s) +const DEFAULT_SESSION_STORE_SERIALIZED_CACHE_MAX_ENTRIES = 64; +const DEFAULT_SESSION_STORE_SERIALIZED_CACHE_MAX_BYTES = 64 * 1024 * 1024; +const LARGE_SESSION_STORE_STRING_MIN_CHARS = 512; +const LARGE_SESSION_STORE_STRING_MAX_INTERNED = 256; const SESSION_STORE_CACHE = createExpiringMapCache({ ttlMs: getSessionStoreTtl, }); -const SESSION_STORE_SERIALIZED_CACHE = new Map(); +const SESSION_STORE_SNAPSHOT_CACHE = createExpiringMapCache( + { + ttlMs: getSessionStoreTtl, + }, +); +const SESSION_STORE_SERIALIZED_CACHE = new Map(); +const SESSION_STORE_STRING_INTERN_POOL = new Map(); +const SESSION_STORE_STRING_INTERN_STATS = { + stored: 0, + reused: 0, + skippedSmall: 0, + skippedFull: 0, +}; +let sessionStoreSnapshotGeneration = 0; +let sessionStoreSerializedCacheBytes = 0; + +function parseNonNegativeInteger(value: string | undefined): number | null { + const trimmed = value?.trim(); + if (!trimmed) { + return null; + } + const parsed = Number.parseInt(trimmed, 10); + return Number.isFinite(parsed) && parsed >= 0 ? parsed : null; +} + +function getSerializedSessionStoreCacheMaxBytes(): number { + return ( + parseNonNegativeInteger(process.env.OPENCLAW_SESSION_SERIALIZED_CACHE_MAX_BYTES) ?? + DEFAULT_SESSION_STORE_SERIALIZED_CACHE_MAX_BYTES + ); +} + +function getSerializedSessionStoreCacheMaxEntries(): number { + return DEFAULT_SESSION_STORE_SERIALIZED_CACHE_MAX_ENTRIES; +} + +function resetSessionStoreStringInternStats(): void { + SESSION_STORE_STRING_INTERN_STATS.stored = 0; + SESSION_STORE_STRING_INTERN_STATS.reused = 0; + SESSION_STORE_STRING_INTERN_STATS.skippedSmall = 0; + SESSION_STORE_STRING_INTERN_STATS.skippedFull = 0; +} + +function internLargeSessionStoreString(value: string): string { + if (value.length < LARGE_SESSION_STORE_STRING_MIN_CHARS) { + SESSION_STORE_STRING_INTERN_STATS.skippedSmall += 1; + return value; + } + const interned = SESSION_STORE_STRING_INTERN_POOL.get(value); + if (interned !== undefined) { + SESSION_STORE_STRING_INTERN_STATS.reused += 1; + return interned; + } + if (SESSION_STORE_STRING_INTERN_POOL.size >= LARGE_SESSION_STORE_STRING_MAX_INTERNED) { + SESSION_STORE_STRING_INTERN_STATS.skippedFull += 1; + return value; + } + SESSION_STORE_STRING_INTERN_POOL.set(value, value); + SESSION_STORE_STRING_INTERN_STATS.stored += 1; + return value; +} + +export function internSessionEntryLargeStrings(entry: SessionEntry): void { + const snapshot = entry.skillsSnapshot; + if (!snapshot?.prompt) { + return; + } + // The live session store repeatedly clones a small set of large skills prompts. + // Intern only that known high-duplication field so behavior and serialization stay unchanged. + snapshot.prompt = internLargeSessionStoreString(snapshot.prompt); +} + +export function internSessionStoreLargeStrings(store: Record): void { + for (const entry of Object.values(store)) { + internSessionEntryLargeStrings(entry); + } +} + +export function getSessionStoreStringInternStatsForTest(): { + poolSize: number; + stored: number; + reused: number; + skippedSmall: number; + skippedFull: number; + minChars: number; + maxEntries: number; +} { + return { + poolSize: SESSION_STORE_STRING_INTERN_POOL.size, + stored: SESSION_STORE_STRING_INTERN_STATS.stored, + reused: SESSION_STORE_STRING_INTERN_STATS.reused, + skippedSmall: SESSION_STORE_STRING_INTERN_STATS.skippedSmall, + skippedFull: SESSION_STORE_STRING_INTERN_STATS.skippedFull, + minChars: LARGE_SESSION_STORE_STRING_MIN_CHARS, + maxEntries: LARGE_SESSION_STORE_STRING_MAX_INTERNED, + }; +} + +export function getSerializedSessionStoreCacheStatsForTest(): { + entries: number; + totalBytes: number; + maxEntries: number; + maxBytes: number; +} { + pruneSerializedSessionStoreCache(); + return { + entries: SESSION_STORE_SERIALIZED_CACHE.size, + totalBytes: sessionStoreSerializedCacheBytes, + maxEntries: getSerializedSessionStoreCacheMaxEntries(), + maxBytes: getSerializedSessionStoreCacheMaxBytes(), + }; +} + +function deepFreeze(value: T, seen = new WeakSet()): DeepReadonly { + if (!value || typeof value !== "object") { + return value as DeepReadonly; + } + const object = value as object; + if (seen.has(object)) { + return value as DeepReadonly; + } + seen.add(object); + for (const child of Object.values(value as Record)) { + deepFreeze(child, seen); + } + return Object.freeze(value) as DeepReadonly; +} export function cloneSessionStoreRecord( store: Record, serialized?: string, ): Record { - return JSON.parse(serialized ?? JSON.stringify(store)) as Record; + const cloned = JSON.parse(serialized ?? JSON.stringify(store)) as Record; + internSessionStoreLargeStrings(cloned); + return cloned; +} + +function cloneJsonLikeValue(value: T): T { + if (!value || typeof value !== "object") { + return value; + } + if (Array.isArray(value)) { + return value.map((item) => cloneJsonLikeValue(item)) as T; + } + const cloned: Record = {}; + for (const [key, child] of Object.entries(value as Record)) { + cloned[key] = cloneJsonLikeValue(child); + } + return cloned as T; +} + +export function cloneSessionStoreSnapshot( + store: Record, + serialized?: string, +): SessionStoreSnapshot { + const cloned = + serialized === undefined + ? cloneJsonLikeValue(store) + : cloneSessionStoreRecord(store, serialized); + internSessionStoreLargeStrings(cloned); + return deepFreeze(cloned); } export function getSessionStoreTtl(): number { @@ -35,30 +223,108 @@ export function isSessionStoreCacheEnabled(): boolean { export function clearSessionStoreCaches(): void { SESSION_STORE_CACHE.clear(); + SESSION_STORE_SNAPSHOT_CACHE.clear(); SESSION_STORE_SERIALIZED_CACHE.clear(); + sessionStoreSerializedCacheBytes = 0; + SESSION_STORE_STRING_INTERN_POOL.clear(); + resetSessionStoreStringInternStats(); } export function invalidateSessionStoreCache(storePath: string): void { SESSION_STORE_CACHE.delete(storePath); + SESSION_STORE_SNAPSHOT_CACHE.delete(storePath); + deleteSerializedSessionStore(storePath); +} + +function deleteSerializedSessionStore(storePath: string): void { + const cached = SESSION_STORE_SERIALIZED_CACHE.get(storePath); + if (!cached) { + return; + } SESSION_STORE_SERIALIZED_CACHE.delete(storePath); + sessionStoreSerializedCacheBytes -= cached.sizeBytes; +} + +function pruneSerializedSessionStoreCache(): void { + const maxEntries = getSerializedSessionStoreCacheMaxEntries(); + const maxBytes = getSerializedSessionStoreCacheMaxBytes(); + while ( + SESSION_STORE_SERIALIZED_CACHE.size > 0 && + (SESSION_STORE_SERIALIZED_CACHE.size > maxEntries || + sessionStoreSerializedCacheBytes > maxBytes) + ) { + const oldestKey = SESSION_STORE_SERIALIZED_CACHE.keys().next().value; + if (typeof oldestKey !== "string") { + break; + } + deleteSerializedSessionStore(oldestKey); + } } export function getSerializedSessionStore(storePath: string): string | undefined { - return SESSION_STORE_SERIALIZED_CACHE.get(storePath); + pruneSerializedSessionStoreCache(); + return SESSION_STORE_SERIALIZED_CACHE.get(storePath)?.serialized; } export function setSerializedSessionStore(storePath: string, serialized?: string): void { + deleteSerializedSessionStore(storePath); if (serialized === undefined) { - SESSION_STORE_SERIALIZED_CACHE.delete(storePath); return; } - SESSION_STORE_SERIALIZED_CACHE.set(storePath, serialized); + const sizeBytes = Buffer.byteLength(serialized, "utf8"); + const maxEntries = getSerializedSessionStoreCacheMaxEntries(); + const maxBytes = getSerializedSessionStoreCacheMaxBytes(); + if (maxEntries <= 0 || maxBytes <= 0 || sizeBytes > maxBytes) { + return; + } + SESSION_STORE_SERIALIZED_CACHE.set(storePath, { serialized, sizeBytes }); + sessionStoreSerializedCacheBytes += sizeBytes; + pruneSerializedSessionStoreCache(); } export function dropSessionStoreObjectCache(storePath: string): void { SESSION_STORE_CACHE.delete(storePath); } +export function dropSessionStoreSnapshotCache(storePath: string): void { + SESSION_STORE_SNAPSHOT_CACHE.delete(storePath); +} + +export function readSessionStoreSnapshotCache(params: { + storePath: string; + mtimeMs?: number; + sizeBytes?: number; +}): SessionStoreSnapshot | null { + const cached = SESSION_STORE_SNAPSHOT_CACHE.get(params.storePath); + if (!cached) { + return null; + } + if (params.mtimeMs !== cached.mtimeMs || params.sizeBytes !== cached.sizeBytes) { + invalidateSessionStoreCache(params.storePath); + return null; + } + return cached.snapshot; +} + +export function writeSessionStoreSnapshotCache(params: { + storePath: string; + store: Record; + mtimeMs?: number; + sizeBytes?: number; + serialized?: string; +}): SessionStoreSnapshot { + const snapshot = cloneSessionStoreSnapshot(params.store, params.serialized); + SESSION_STORE_SNAPSHOT_CACHE.set(params.storePath, { + snapshot, + mtimeMs: params.mtimeMs, + sizeBytes: params.sizeBytes, + generation: (sessionStoreSnapshotGeneration += 1), + createdAt: Date.now(), + entryCount: Object.keys(snapshot).length, + }); + return snapshot; +} + export function readSessionStoreCache(params: { storePath: string; mtimeMs?: number; @@ -103,13 +369,16 @@ export function writeSessionStoreCache(params: { sizeBytes?: number; serialized?: string; }): void { + const store = + params.serialized === undefined ? cloneSessionStoreRecord(params.store) : params.store; + if (params.serialized !== undefined) { + internSessionStoreLargeStrings(store); + } SESSION_STORE_CACHE.set(params.storePath, { - store: params.serialized === undefined ? cloneSessionStoreRecord(params.store) : params.store, + store, mtimeMs: params.mtimeMs, sizeBytes: params.sizeBytes, serialized: params.serialized, }); - if (params.serialized !== undefined) { - SESSION_STORE_SERIALIZED_CACHE.set(params.storePath, params.serialized); - } + setSerializedSessionStore(params.storePath, params.serialized); } diff --git a/src/config/sessions/store-load.ts b/src/config/sessions/store-load.ts index 1428fdf3819c..6cb71413ac7f 100644 --- a/src/config/sessions/store-load.ts +++ b/src/config/sessions/store-load.ts @@ -11,12 +11,20 @@ import { import { getFileStatSnapshot } from "../cache-utils.js"; import { cloneSessionStoreRecord, + cloneSessionStoreSnapshot, + internSessionEntryLargeStrings, isSessionStoreCacheEnabled, readSessionStoreCache, + readSessionStoreSnapshotCache, setSerializedSessionStore, writeSessionStoreCache, + writeSessionStoreSnapshotCache, + type SessionStoreSnapshot, + type SessionStoreSnapshotEntries, + type SessionStoreSnapshotEntry, } from "./store-cache.js"; import { normalizePersistedSessionEntryShape } from "./store-entry-shape.js"; +import { resolveSessionStoreEntry } from "./store-entry.js"; import { collectSessionMaintenancePreserveKeys } from "./store-maintenance-preserve.js"; import { resolveMaintenanceConfig } from "./store-maintenance-runtime.js"; import { @@ -335,6 +343,7 @@ export function normalizeSessionStore(store: Record): bool ), ), ); + internSessionEntryLargeStrings(normalized); if (normalized !== entry) { store[key] = normalized; changed = true; @@ -380,8 +389,9 @@ export function loadSessionStore( store = parsed; serializedFromDisk = raw; } - fileStat = getFileStatSnapshot(storePath) ?? fileStat; - mtimeMs = fileStat?.mtimeMs; + // Cache with the stat observed before this read. If another process + // writes the file after readFileSync returns, a post-read stat could tag + // stale content as current and make future cache hits return old data. break; } catch { if (attempt < maxReadAttempts - 1) { @@ -446,3 +456,44 @@ export function loadSessionStore( return opts.clone === false ? store : cloneSessionStoreRecord(store, serializedFromDisk); } + +export function readSessionStoreSnapshot(storePath: string): SessionStoreSnapshot { + const currentFileStat = getFileStatSnapshot(storePath); + if (isSessionStoreCacheEnabled()) { + const cached = readSessionStoreSnapshotCache({ + storePath, + mtimeMs: currentFileStat?.mtimeMs, + sizeBytes: currentFileStat?.sizeBytes, + }); + if (cached) { + return cached; + } + } + + const store = loadSessionStore(storePath); + if (!isSessionStoreCacheEnabled()) { + return cloneSessionStoreSnapshot(store); + } + return writeSessionStoreSnapshotCache({ + storePath, + store, + mtimeMs: currentFileStat?.mtimeMs, + sizeBytes: currentFileStat?.sizeBytes, + }); +} + +export function readSessionEntry( + storePath: string, + sessionKey: string, +): SessionStoreSnapshotEntry | undefined { + const snapshot = readSessionStoreSnapshot(storePath); + const resolved = resolveSessionStoreEntry({ + store: snapshot as Record, + sessionKey, + }); + return resolved.existing as SessionStoreSnapshotEntry | undefined; +} + +export function readSessionEntries(storePath: string): SessionStoreSnapshotEntries { + return Object.entries(readSessionStoreSnapshot(storePath)) as SessionStoreSnapshotEntries; +} diff --git a/src/config/sessions/store.ts b/src/config/sessions/store.ts index 08fb79d8c643..83edc8dc99f8 100644 --- a/src/config/sessions/store.ts +++ b/src/config/sessions/store.ts @@ -3,6 +3,7 @@ import path from "node:path"; import type { MsgContext } from "../../auto-reply/templating.js"; import { writeTextAtomic } from "../../infra/json-files.js"; import { createSubsystemLogger } from "../../logging/subsystem.js"; +import { resolveAgentIdFromSessionKey } from "../../routing/session-key.js"; import { deliveryContextFromChannelRoute, deliveryContextFromSession, @@ -12,18 +13,28 @@ import { } from "../../utils/delivery-context.shared.js"; import type { DeliveryContext } from "../../utils/delivery-context.types.js"; import { getFileStatSnapshot } from "../cache-utils.js"; +import { getRuntimeConfig } from "../io.js"; import { enforceSessionDiskBudget, type SessionDiskBudgetSweepResult } from "./disk-budget.js"; import { deriveSessionMetaPatch } from "./metadata.js"; +import { resolveStorePath } from "./paths.js"; import { + cloneSessionStoreRecord, dropSessionStoreObjectCache, + dropSessionStoreSnapshotCache, getSerializedSessionStore, isSessionStoreCacheEnabled, setSerializedSessionStore, takeMutableSessionStoreCache, writeSessionStoreCache, + writeSessionStoreSnapshotCache, } from "./store-cache.js"; import { normalizeStoreSessionKey, resolveSessionStoreEntry } from "./store-entry.js"; -import { loadSessionStore, normalizeSessionStore } from "./store-load.js"; +import { + loadSessionStore, + normalizeSessionStore, + readSessionEntries, + readSessionEntry, +} from "./store-load.js"; import { collectSessionMaintenancePreserveKeys } from "./store-maintenance-preserve.js"; import { resolveMaintenanceConfig } from "./store-maintenance-runtime.js"; import { @@ -49,7 +60,17 @@ export { getSessionStoreWriterQueueSizeForTest, } from "./store-writer-state.js"; export { withSessionStoreWriterForTest } from "./store-writer.js"; -export { loadSessionStore } from "./store-load.js"; +export { + loadSessionStore, + readSessionEntries, + readSessionEntry, + readSessionStoreSnapshot, +} from "./store-load.js"; +export type { + SessionStoreSnapshot, + SessionStoreSnapshotEntries, + SessionStoreSnapshotEntry, +} from "./store-cache.js"; export { normalizeStoreSessionKey, resolveSessionStoreEntry } from "./store-entry.js"; const log = createSubsystemLogger("sessions/store"); @@ -83,9 +104,7 @@ export function readSessionUpdatedAt(params: { sessionKey: string; }): number | undefined { try { - const store = loadSessionStore(params.storePath); - const resolved = resolveSessionStoreEntry({ store, sessionKey: params.sessionKey }); - return resolved.existing?.updatedAt; + return readSessionEntry(params.storePath, params.sessionKey)?.updatedAt; } catch { return undefined; } @@ -133,6 +152,49 @@ type SaveSessionStoreOptions = { maintenanceConfig?: ResolvedSessionMaintenanceConfig; }; +type SessionEntryWorkflowOptions = { + agentId?: string; + env?: NodeJS.ProcessEnv; + storePath?: string; +}; + +function cloneSessionEntry(entry: SessionEntry): SessionEntry { + return cloneSessionStoreRecord({ entry }).entry; +} + +function resolveSessionWorkflowStorePath( + options: SessionEntryWorkflowOptions & { sessionKey?: string }, +): string { + if (options.storePath) { + return options.storePath; + } + const agentId = options.agentId ?? resolveAgentIdFromSessionKey(options.sessionKey); + return resolveStorePath(getRuntimeConfig().session?.store, { + agentId, + env: options.env, + }); +} + +export function getSessionEntry( + options: SessionEntryWorkflowOptions & { sessionKey: string }, +): SessionEntry | undefined { + const entry = readSessionEntry(resolveSessionWorkflowStorePath(options), options.sessionKey) as + | SessionEntry + | undefined; + return entry ? cloneSessionEntry(entry) : undefined; +} + +export function listSessionEntries( + options: SessionEntryWorkflowOptions = {}, +): Array<{ sessionKey: string; entry: SessionEntry }> { + return readSessionEntries(resolveSessionWorkflowStorePath(options)).map( + ([sessionKey, entry]) => ({ + sessionKey, + entry: cloneSessionEntry(entry as SessionEntry), + }), + ); +} + function updateSessionStoreWriteCaches(params: { storePath: string; store: Record; @@ -142,6 +204,7 @@ function updateSessionStoreWriteCaches(params: { setSerializedSessionStore(params.storePath, params.serialized); if (!isSessionStoreCacheEnabled()) { dropSessionStoreObjectCache(params.storePath); + dropSessionStoreSnapshotCache(params.storePath); return; } writeSessionStoreCache({ @@ -151,6 +214,12 @@ function updateSessionStoreWriteCaches(params: { sizeBytes: fileStat?.sizeBytes, serialized: params.serialized, }); + writeSessionStoreSnapshotCache({ + storePath: params.storePath, + store: params.store, + mtimeMs: fileStat?.mtimeMs, + sizeBytes: fileStat?.sizeBytes, + }); } function loadMutableSessionStoreForWriter(storePath: string): Record { @@ -575,6 +644,67 @@ export async function updateSessionStoreEntry(params: { }); } +export async function patchSessionEntry( + params: SessionEntryWorkflowOptions & { + sessionKey: string; + fallbackEntry?: SessionEntry; + preserveActivity?: boolean; + replaceEntry?: boolean; + update: ( + entry: SessionEntry, + ) => Promise | null> | Partial | null; + }, +): Promise { + const storePath = resolveSessionWorkflowStorePath(params); + return await runExclusiveSessionStoreWrite(storePath, async () => { + const store = loadMutableSessionStoreForWriter(storePath); + const resolved = resolveSessionStoreEntry({ store, sessionKey: params.sessionKey }); + const existing = resolved.existing ?? params.fallbackEntry; + if (!existing) { + return null; + } + const patch = await params.update(cloneSessionEntry(existing)); + if (!patch) { + return existing; + } + const next = params.replaceEntry + ? cloneSessionEntry(patch as SessionEntry) + : params.preserveActivity + ? mergeSessionEntryPreserveActivity(existing, patch) + : mergeSessionEntry(existing, patch); + return await persistResolvedSessionEntry({ + storePath, + store, + resolved, + next, + }); + }); +} + +export async function upsertSessionEntry( + params: SessionEntryWorkflowOptions & { + sessionKey: string; + entry: SessionEntry; + allowDropAcpMeta?: boolean; + }, +): Promise { + const storePath = resolveSessionWorkflowStorePath(params); + await runExclusiveSessionStoreWrite(storePath, async () => { + const store = loadMutableSessionStoreForWriter(storePath); + const resolved = resolveSessionStoreEntry({ store, sessionKey: params.sessionKey }); + const next = cloneSessionEntry(params.entry); + if (!params.allowDropAcpMeta && resolved.existing?.acp && !next.acp) { + next.acp = resolved.existing.acp; + } + await persistResolvedSessionEntry({ + storePath, + store, + resolved, + next, + }); + }); +} + export async function recordSessionMetaFromInbound(params: { storePath: string; sessionKey: string; diff --git a/src/cron/isolated-agent/delivery-target.test.ts b/src/cron/isolated-agent/delivery-target.test.ts index 8fd64dd55d97..ec4d7e370e92 100644 --- a/src/cron/isolated-agent/delivery-target.test.ts +++ b/src/cron/isolated-agent/delivery-target.test.ts @@ -27,6 +27,7 @@ vi.mock("../../config/sessions/paths.js", () => ({ vi.mock("../../config/sessions/store-load.js", () => ({ loadSessionStore: vi.fn().mockReturnValue({}), + readSessionEntry: vi.fn(), })); vi.mock("../../infra/outbound/channel-selection.runtime.js", () => ({ @@ -57,7 +58,7 @@ const mockedModuleIds = [ "../../pairing/allow-from-store-read.js", ]; -import { loadSessionStore } from "../../config/sessions/store-load.js"; +import { loadSessionStore, readSessionEntry } from "../../config/sessions/store-load.js"; import { resolveMessageChannelSelection } from "../../infra/outbound/channel-selection.runtime.js"; import { maybeResolveIdLikeTarget } from "../../infra/outbound/target-id-resolution.js"; import { resolveOutboundTarget } from "../../infra/outbound/targets.runtime.js"; @@ -118,6 +119,8 @@ beforeEach(() => { vi.mocked(readChannelAllowFromStoreEntriesSync).mockReset(); vi.mocked(readChannelAllowFromStoreEntriesSync).mockReturnValue([]); vi.mocked(resolveOutboundTarget).mockReset(); + vi.mocked(loadSessionStore).mockReset().mockReturnValue({}); + vi.mocked(readSessionEntry).mockReset().mockReturnValue(undefined); setActivePluginRegistry( createTestRegistry([ { @@ -194,6 +197,7 @@ type SessionStore = ReturnType; function setSessionStore(store: SessionStore) { vi.mocked(loadSessionStore).mockReturnValue(store); + vi.mocked(readSessionEntry).mockImplementation((_storePath, sessionKey) => store[sessionKey]); } function setMainSessionEntry(entry?: SessionStore[string]) { @@ -242,6 +246,21 @@ async function resolveLastTarget(cfg: OpenClawConfig) { } describe("resolveDeliveryTarget", () => { + it("uses session-entry snapshot reads for implicit last delivery lookup", async () => { + setLastSessionEntry({ + sessionId: "sess-w1", + lastChannel: "alpha", + lastTo: "room-allowed", + }); + + const result = await resolveLastTarget(makeCfg({ channels: { alpha: { allowFrom: [] } } })); + + expect(result.channel).toBe("alpha"); + expect(result.to).toBe("room-allowed"); + expect(readSessionEntry).toHaveBeenCalledWith("/tmp/test-store.json", "agent:test:main"); + expect(loadSessionStore).not.toHaveBeenCalled(); + }); + it("reroutes implicit delivery to an authorized allowFrom recipient", async () => { setLastSessionEntry({ sessionId: "sess-w1", diff --git a/src/cron/isolated-agent/delivery-target.ts b/src/cron/isolated-agent/delivery-target.ts index 635ba65abfe1..436853acc49c 100644 --- a/src/cron/isolated-agent/delivery-target.ts +++ b/src/cron/isolated-agent/delivery-target.ts @@ -2,7 +2,7 @@ import { parseExplicitTargetForLoadedChannel } from "../../channels/plugins/targ import type { ChannelId } from "../../channels/plugins/types.public.js"; import { resolveAgentMainSessionKey } from "../../config/sessions/main-session.js"; import { resolveStorePath } from "../../config/sessions/paths.js"; -import { loadSessionStore } from "../../config/sessions/store-load.js"; +import { readSessionEntry } from "../../config/sessions/store-load.js"; import type { SessionEntry } from "../../config/sessions/types.js"; import type { OpenClawConfig } from "../../config/types.openclaw.js"; import { formatErrorMessage } from "../../infra/errors.js"; @@ -132,7 +132,6 @@ export async function resolveDeliveryTarget( const sessionCfg = cfg.session; const mainSessionKey = resolveAgentMainSessionKey({ cfg, agentId }); const storePath = resolveStorePath(sessionCfg?.store, { agentId }); - const store = loadSessionStore(storePath); // Look up thread-specific session first (e.g. agent:main:main:thread:1234), // then fall back to the main session entry. @@ -156,8 +155,11 @@ export async function resolveDeliveryTarget( deliveryContext: storedDeliveryContext, } satisfies SessionEntry) : undefined; - const threadEntry = threadSessionKey ? store[threadSessionKey] : undefined; - const main = storedDeliveryEntry ?? threadEntry ?? store[mainSessionKey]; + const threadEntry = threadSessionKey + ? (readSessionEntry(storePath, threadSessionKey) as SessionEntry | undefined) + : undefined; + const mainEntry = readSessionEntry(storePath, mainSessionKey) as SessionEntry | undefined; + const main = storedDeliveryEntry ?? threadEntry ?? mainEntry; const preliminary = resolveSessionDeliveryTarget({ entry: main, diff --git a/src/cron/service/timer.ts b/src/cron/service/timer.ts index 3efd0210001e..fe7babc32e7a 100644 --- a/src/cron/service/timer.ts +++ b/src/cron/service/timer.ts @@ -1,12 +1,13 @@ import { resolveFailoverReasonFromError } from "../../agents/failover-error.js"; import { formatEmbeddedAgentExecutionPhase } from "../../agents/pi-embedded-runner/execution-phase.js"; +import { readSessionEntry } from "../../config/sessions/store-load.js"; +import type { SessionEntry } from "../../config/sessions/types.js"; import type { CronConfig, CronRetryOn } from "../../config/types.cron.js"; import type { HeartbeatRunResult } from "../../infra/heartbeat-wake.js"; import { HEARTBEAT_SKIP_CRON_IN_PROGRESS, isRetryableHeartbeatBusySkipReason, } from "../../infra/heartbeat-wake.js"; -import { loadSessionStore } from "../../config/sessions/store-load.js"; import { DEFAULT_AGENT_ID, isSubagentSessionKey, @@ -438,9 +439,7 @@ function normalizeCronLaneSegment(value: string | undefined, fallback: string): function resolveMainSessionCronRunSessionKey(job: CronJob, startedAt: number): string { const explicitAgentId = job.agentId?.trim(); - const agentId = normalizeAgentId( - explicitAgentId || resolveAgentIdFromSessionKey(job.sessionKey), - ); + const agentId = normalizeAgentId(explicitAgentId || resolveAgentIdFromSessionKey(job.sessionKey)); const jobSegment = normalizeCronLaneSegment(job.id, "job"); const runSegment = normalizeCronLaneSegment(String(Math.max(0, Math.floor(startedAt))), "run"); return `agent:${agentId}:cron:${jobSegment}:run:${runSegment}`; @@ -463,7 +462,8 @@ function resolveMainSessionCronDeliveryContext( return undefined; } try { - return deliveryContextFromSession(loadSessionStore(storePath)[targetSessionKey]); + const sessionEntry = readSessionEntry(storePath, targetSessionKey) as SessionEntry | undefined; + return deliveryContextFromSession(sessionEntry); } catch { return undefined; } diff --git a/src/plugin-sdk/config-runtime.ts b/src/plugin-sdk/config-runtime.ts index 2d56e1afef6f..08d4d5f54011 100644 --- a/src/plugin-sdk/config-runtime.ts +++ b/src/plugin-sdk/config-runtime.ts @@ -4,6 +4,15 @@ * config-mutation, and runtime-config-snapshot. */ +import { loadSessionStore as loadSessionStoreImpl } from "../config/sessions/store-load.js"; + +/** + * @deprecated Use getSessionEntry/listSessionEntries for reads and + * patchSessionEntry/upsertSessionEntry for writes. loadSessionStore keeps the + * legacy mutable whole-store shape and will remain a compatibility escape hatch. + */ +export const loadSessionStore = loadSessionStoreImpl; + export { resolveDefaultAgentId } from "../agents/agent-scope.js"; export { requireRuntimeConfig, @@ -132,13 +141,16 @@ export type { } from "../config/types.js"; export { clearSessionStoreCacheForTest, - loadSessionStore, + getSessionEntry, + listSessionEntries, + patchSessionEntry, readSessionUpdatedAt, recordSessionMetaFromInbound, saveSessionStore, updateLastRoute, updateSessionStore, updateSessionStoreEntry, + upsertSessionEntry, resolveSessionStoreEntry, } from "../config/sessions/store.js"; export { resolveSessionKey } from "../config/sessions/session-key.js"; diff --git a/src/plugin-sdk/session-store-runtime.ts b/src/plugin-sdk/session-store-runtime.ts index 93de3ecdde17..fb9e2483f1ad 100644 --- a/src/plugin-sdk/session-store-runtime.ts +++ b/src/plugin-sdk/session-store-runtime.ts @@ -1,6 +1,14 @@ // Narrow session-store helpers for channel hot paths. -export { loadSessionStore } from "../config/sessions/store-load.js"; +import { loadSessionStore as loadSessionStoreImpl } from "../config/sessions/store-load.js"; + +/** + * @deprecated Use getSessionEntry/listSessionEntries for reads and + * patchSessionEntry/upsertSessionEntry for writes. loadSessionStore keeps the + * legacy mutable whole-store shape and will remain a compatibility escape hatch. + */ +export const loadSessionStore = loadSessionStoreImpl; + export { resolveSessionStoreEntry } from "../config/sessions/store-entry.js"; export { resolveSessionTranscriptPathInDir, resolveStorePath } from "../config/sessions/paths.js"; export { resolveAndPersistSessionFile } from "../config/sessions/session-file.js"; @@ -10,12 +18,16 @@ export { resolveGroupSessionKey } from "../config/sessions/group.js"; export { canonicalizeMainSessionAlias } from "../config/sessions/main-session.js"; export { clearSessionStoreCacheForTest, + getSessionEntry, + listSessionEntries, + patchSessionEntry, readSessionUpdatedAt, recordSessionMetaFromInbound, saveSessionStore, updateLastRoute, updateSessionStore, updateSessionStoreEntry, + upsertSessionEntry, } from "../config/sessions/store.js"; export { evaluateSessionFreshness, diff --git a/src/plugin-sdk/test-helpers/plugin-runtime-mock.ts b/src/plugin-sdk/test-helpers/plugin-runtime-mock.ts index 5d51449ed879..99490a1df4a7 100644 --- a/src/plugin-sdk/test-helpers/plugin-runtime-mock.ts +++ b/src/plugin-sdk/test-helpers/plugin-runtime-mock.ts @@ -379,6 +379,22 @@ export function createPluginRuntimeMock(overrides: DeepPartial = loadSessionStore: vi.fn( () => ({}), ) as unknown as PluginRuntime["agent"]["session"]["loadSessionStore"], + getSessionEntry: vi.fn( + () => undefined, + ) as unknown as PluginRuntime["agent"]["session"]["getSessionEntry"], + listSessionEntries: vi.fn( + () => [], + ) as unknown as PluginRuntime["agent"]["session"]["listSessionEntries"], + patchSessionEntry: vi + .fn() + .mockResolvedValue( + null, + ) as unknown as PluginRuntime["agent"]["session"]["patchSessionEntry"], + upsertSessionEntry: vi + .fn() + .mockResolvedValue( + undefined, + ) as unknown as PluginRuntime["agent"]["session"]["upsertSessionEntry"], saveSessionStore: vi .fn() .mockResolvedValue( diff --git a/src/plugins/runtime/index.test.ts b/src/plugins/runtime/index.test.ts index 9ab833a374d1..cc6ac9b0f305 100644 --- a/src/plugins/runtime/index.test.ts +++ b/src/plugins/runtime/index.test.ts @@ -311,6 +311,10 @@ describe("plugin runtime command execution", () => { "resolveAgentDir", ]); expectFunctionKeys(runtime.agent.session as Record, [ + "getSessionEntry", + "listSessionEntries", + "patchSessionEntry", + "upsertSessionEntry", "updateSessionStore", "updateSessionStoreEntry", "resolveSessionFilePath", diff --git a/src/plugins/runtime/runtime-agent.ts b/src/plugins/runtime/runtime-agent.ts index fbf097f22623..613c9c88f1af 100644 --- a/src/plugins/runtime/runtime-agent.ts +++ b/src/plugins/runtime/runtime-agent.ts @@ -11,10 +11,14 @@ import { normalizeThinkLevel, resolveThinkingProfile } from "../../auto-reply/th import { getRuntimeConfig } from "../../config/config.js"; import { resolveSessionFilePath, resolveStorePath } from "../../config/sessions/paths.js"; import { + getSessionEntry, + listSessionEntries, loadSessionStore, + patchSessionEntry, saveSessionStore, updateSessionStore, updateSessionStoreEntry, + upsertSessionEntry, } from "../../config/sessions/store.js"; import { createLazyRuntimeMethod, createLazyRuntimeModule } from "../../shared/lazy-runtime.js"; import { defineCachedValue } from "./runtime-cache.js"; @@ -71,6 +75,10 @@ export function createRuntimeAgent(): PluginRuntime["agent"] { ); defineCachedValue(agentRuntime, "session", () => ({ resolveStorePath, + getSessionEntry, + listSessionEntries, + patchSessionEntry, + upsertSessionEntry, loadSessionStore, saveSessionStore, updateSessionStore, diff --git a/src/plugins/runtime/types-core.ts b/src/plugins/runtime/types-core.ts index a82563596b21..65addaed42c8 100644 --- a/src/plugins/runtime/types-core.ts +++ b/src/plugins/runtime/types-core.ts @@ -203,6 +203,15 @@ export type PluginRuntimeCore = { ensureAgentWorkspace: typeof import("../../agents/workspace.js").ensureAgentWorkspace; session: { resolveStorePath: typeof import("../../config/sessions/paths.js").resolveStorePath; + getSessionEntry: typeof import("../../config/sessions/store.js").getSessionEntry; + listSessionEntries: typeof import("../../config/sessions/store.js").listSessionEntries; + patchSessionEntry: typeof import("../../config/sessions/store.js").patchSessionEntry; + upsertSessionEntry: typeof import("../../config/sessions/store.js").upsertSessionEntry; + /** + * @deprecated Use getSessionEntry/listSessionEntries for reads and + * patchSessionEntry/upsertSessionEntry for writes. This keeps the legacy + * mutable whole-store compatibility shape. + */ loadSessionStore: typeof import("../../config/sessions/store-load.js").loadSessionStore; saveSessionStore: import("../../config/sessions/runtime-types.js").SaveSessionStore; updateSessionStore: typeof import("../../config/sessions/store.js").updateSessionStore; diff --git a/src/talk/agent-consult-runtime.test.ts b/src/talk/agent-consult-runtime.test.ts index 93c016206f4f..33473d82eb42 100644 --- a/src/talk/agent-consult-runtime.test.ts +++ b/src/talk/agent-consult-runtime.test.ts @@ -42,6 +42,35 @@ function createAgentRuntime(payloads: unknown[] = [{ text: "Speak this." }]) { return await mutator(sessionStore); }, ); + const getSessionEntry = vi.fn( + (params: { sessionKey: string }) => sessionStore[params.sessionKey], + ); + const patchSessionEntry = vi.fn( + async (params: { + sessionKey: string; + fallbackEntry?: Record; + update: ( + entry: Record, + ) => Promise | null> | Record | null; + }) => { + const existing = sessionStore[params.sessionKey] ?? params.fallbackEntry; + if (!existing) { + return null; + } + const patch = await params.update({ ...existing }); + if (!patch) { + return existing; + } + const next = { ...existing, ...patch }; + sessionStore[params.sessionKey] = next; + return next; + }, + ); + const upsertSessionEntry = vi.fn( + async (params: { sessionKey: string; entry: Record }) => { + sessionStore[params.sessionKey] = { ...params.entry }; + }, + ); return { runtime: { resolveAgentDir: vi.fn(() => "/tmp/agent"), @@ -53,6 +82,9 @@ function createAgentRuntime(payloads: unknown[] = [{ text: "Speak this." }]) { loadSessionStore: vi.fn(() => sessionStore), saveSessionStore: vi.fn(async () => {}), updateSessionStore, + getSessionEntry, + patchSessionEntry, + upsertSessionEntry, resolveSessionFilePath: vi.fn( (_sessionId: string, entry?: { sessionFile?: string }) => entry?.sessionFile ?? "/tmp/session.json", diff --git a/src/talk/agent-consult-runtime.ts b/src/talk/agent-consult-runtime.ts index 4a4c0d0b4e66..24aafb1ba13f 100644 --- a/src/talk/agent-consult-runtime.ts +++ b/src/talk/agent-consult-runtime.ts @@ -87,7 +87,6 @@ function resolveRealtimeVoiceAgentDeliveryContext(params: { }): DeliveryContext | undefined { const requesterSessionKey = params.spawnedBy?.trim(); try { - const store = params.agentRuntime.session.loadSessionStore(params.storePath); const candidates: string[] = []; if (requesterSessionKey) { const { baseSessionKey } = parseSessionThreadInfoFast(requesterSessionKey); @@ -97,7 +96,11 @@ function resolveRealtimeVoiceAgentDeliveryContext(params: { } candidates.push(params.sessionKey); for (const key of candidates) { - const context = deliveryContextFromSession(store[key] as SessionEntry | undefined); + const entry = params.agentRuntime.session.getSessionEntry({ + storePath: params.storePath, + sessionKey: key, + }); + const context = deliveryContextFromSession(entry); if (hasRoutableDeliveryContext(context)) { return context; } @@ -119,64 +122,72 @@ async function resolveRealtimeVoiceAgentConsultSessionEntry(params: { logger: Pick; }): Promise { const now = Date.now(); - return await params.agentRuntime.session.updateSessionStore(params.storePath, async (store) => { - const existing = store[params.sessionKey] as SessionEntry | undefined; - const deliveryFields = resolveDeliverySessionFields(params.deliveryContext); - if (existing?.sessionId?.trim()) { - const next: SessionEntry = { ...existing, ...deliveryFields, updatedAt: now }; - store[params.sessionKey] = next; - return next; - } + const deliveryFields = resolveDeliverySessionFields(params.deliveryContext); + const requesterSessionKey = params.spawnedBy?.trim(); + const requesterAgentId = parseAgentSessionKey(requesterSessionKey)?.agentId; + const shouldFork = + params.contextMode === "fork" && + requesterSessionKey && + (!requesterAgentId || requesterAgentId === params.agentId); + let forkDecisionWarning: string | undefined; - const requesterSessionKey = params.spawnedBy?.trim(); - const requesterAgentId = parseAgentSessionKey(requesterSessionKey)?.agentId; - const shouldFork = - params.contextMode === "fork" && - requesterSessionKey && - (!requesterAgentId || requesterAgentId === params.agentId); - - if (shouldFork) { - const parentEntry = store[requesterSessionKey] as SessionEntry | undefined; - if (parentEntry?.sessionId?.trim()) { - const decision = await realtimeVoiceAgentConsultDeps.resolveParentForkDecision({ - parentEntry, + const patched = await params.agentRuntime.session.patchSessionEntry({ + storePath: params.storePath, + sessionKey: params.sessionKey, + fallbackEntry: { + sessionId: "", + updatedAt: now, + }, + update: async (entry) => { + if (entry.sessionId?.trim()) { + return { ...deliveryFields, updatedAt: now }; + } + if (shouldFork) { + const parentEntry = params.agentRuntime.session.getSessionEntry({ storePath: params.storePath, + sessionKey: requesterSessionKey, }); - if (decision.status === "fork") { - const fork = await realtimeVoiceAgentConsultDeps.forkSessionFromParent({ + if (parentEntry?.sessionId?.trim()) { + const decision = await realtimeVoiceAgentConsultDeps.resolveParentForkDecision({ parentEntry, - agentId: params.agentId, - sessionsDir: path.dirname(params.storePath), + storePath: params.storePath, }); - if (fork) { - const next: SessionEntry = { - ...existing, - ...deliveryFields, - sessionId: fork.sessionId, - sessionFile: fork.sessionFile, - spawnedBy: requesterSessionKey, - forkedFromParent: true, - updatedAt: now, - }; - store[params.sessionKey] = next; - return next; + 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; } - } else { - params.logger.warn(`[talk] ${decision.message}`); } } - } - - const next: SessionEntry = { - ...existing, - ...deliveryFields, - sessionId: realtimeVoiceAgentConsultDeps.randomUUID(), - ...(requesterSessionKey ? { spawnedBy: requesterSessionKey } : {}), - updatedAt: now, - }; - store[params.sessionKey] = next; - return next; + return { + ...deliveryFields, + sessionId: realtimeVoiceAgentConsultDeps.randomUUID(), + ...(requesterSessionKey ? { spawnedBy: requesterSessionKey } : {}), + updatedAt: now, + }; + }, }); + if (forkDecisionWarning) { + params.logger.warn(`[talk] ${forkDecisionWarning}`); + } + if (patched?.sessionId?.trim()) { + return patched; + } + throw new Error("realtime voice agent consult session could not be initialized"); } export async function consultRealtimeVoiceAgent(params: {