From f00810b0ea4ff3e33eeae5f47eefd14dc6ba5e8e Mon Sep 17 00:00:00 2001 From: Josh Lehman Date: Sun, 14 Jun 2026 06:18:47 -0700 Subject: [PATCH] refactor: add session accessor seam with gateway consumer (#90463) Merged via squash. Prepared head SHA: 58aa59eaf8059c31a2a70f859eeb96ee35e5beb2 Co-authored-by: jalehman <550978+jalehman@users.noreply.github.com> Co-authored-by: jalehman <550978+jalehman@users.noreply.github.com> Reviewed-by: @jalehman (cherry picked from commit ef47dd610c8706823d120c1738c647a2204e7942) --- .github/workflows/ci.yml | 5 + package.json | 1 + scripts/check-session-accessor-boundary.mjs | 146 +++++ src/config/sessions/combined-store-gateway.ts | 15 +- src/config/sessions/session-accessor.test.ts | 587 ++++++++++++++++++ src/config/sessions/session-accessor.ts | 480 ++++++++++++++ src/config/sessions/store.runtime.ts | 5 + src/config/sessions/store.ts | 255 +++++++- src/config/sessions/transcript-append.ts | 38 ++ src/gateway/server-methods/sessions.ts | 27 +- src/gateway/session-utils.ts | 23 +- src/gateway/sessions-resolve.test.ts | 63 +- src/gateway/sessions-resolve.ts | 31 +- .../check-session-accessor-boundary.test.ts | 75 +++ test/scripts/ci-workflow-guards.test.ts | 16 + .../run-additional-boundary-checks.test.ts | 2 +- 16 files changed, 1703 insertions(+), 66 deletions(-) create mode 100644 scripts/check-session-accessor-boundary.mjs create mode 100644 src/config/sessions/session-accessor.test.ts create mode 100644 src/config/sessions/session-accessor.ts create mode 100644 test/scripts/check-session-accessor-boundary.test.ts diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 6e59702c47d2..bab3985d8cbf 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -1358,6 +1358,8 @@ jobs: - check_name: check-additional-boundaries-bcd group: boundaries boundary_shard: 2/4,3/4,4/4 + - check_name: check-session-accessor-boundary + group: session-accessor-boundary - check_name: check-additional-extension-channels group: extension-channels - check_name: check-additional-extension-bundled @@ -1504,6 +1506,9 @@ jobs: boundaries) node scripts/run-additional-boundary-checks.mjs ;; + session-accessor-boundary) + run_check "lint:tmp:session-accessor-boundary" pnpm run lint:tmp:session-accessor-boundary + ;; extension-channels) run_check "lint:extensions:channels" pnpm run lint:extensions:channels ;; diff --git a/package.json b/package.json index 78b250226a2c..05db0cf82918 100644 --- a/package.json +++ b/package.json @@ -1596,6 +1596,7 @@ "lint:tmp:no-random-messaging": "node scripts/check-no-random-messaging-tmp.mjs", "lint:tmp:no-raw-channel-fetch": "node scripts/check-no-raw-channel-fetch.mjs", "lint:tmp:no-raw-http2-imports": "node scripts/check-no-raw-http2-imports.mjs", + "lint:tmp:session-accessor-boundary": "node scripts/check-session-accessor-boundary.mjs", "lint:tmp:tsgo-core-boundary": "node scripts/check-tsgo-core-boundary.mjs", "lint:ui:no-raw-window-open": "node scripts/check-no-raw-window-open.mjs", "lint:web-fetch-provider-boundaries": "node scripts/check-web-fetch-provider-boundaries.mjs", diff --git a/scripts/check-session-accessor-boundary.mjs b/scripts/check-session-accessor-boundary.mjs new file mode 100644 index 000000000000..e124a6c0e70c --- /dev/null +++ b/scripts/check-session-accessor-boundary.mjs @@ -0,0 +1,146 @@ +#!/usr/bin/env node + +import path from "node:path"; +import ts from "typescript"; +import { + collectFileViolations, + resolveRepoRoot, + resolveSourceRoots, + runAsScript, + toLine, + unwrapExpression, +} from "./lib/ts-guard-utils.mjs"; + +const legacyReaderNames = new Set(["loadSessionStore", "readSessionEntries"]); + +export const migratedSessionAccessorFiles = new Set([ + "src/config/sessions/combined-store-gateway.ts", + "src/gateway/session-utils.ts", + "src/gateway/sessions-resolve.ts", + "src/gateway/server-methods/sessions.ts", +]); + +function normalizeRelativePath(filePath) { + return filePath.replaceAll(path.sep, "/"); +} + +function propertyAccessName(expression) { + const unwrapped = unwrapExpression(expression); + if (ts.isIdentifier(unwrapped)) { + return unwrapped.text; + } + if (ts.isPropertyAccessExpression(unwrapped)) { + return unwrapped.name.text; + } + if (ts.isElementAccessExpression(unwrapped) && ts.isStringLiteral(unwrapped.argumentExpression)) { + return unwrapped.argumentExpression.text; + } + return null; +} + +function bindingName(node) { + if (node.propertyName && ts.isIdentifier(node.propertyName)) { + return node.propertyName.text; + } + if (ts.isIdentifier(node.name)) { + return node.name.text; + } + return null; +} + +export function findSessionAccessorBoundaryViolations(content, fileName = "source.ts") { + const sourceFile = ts.createSourceFile(fileName, content, ts.ScriptTarget.Latest, true); + const violations = []; + + const visit = (node) => { + if (ts.isImportDeclaration(node)) { + const namedBindings = node.importClause?.namedBindings; + if (namedBindings && ts.isNamedImports(namedBindings)) { + for (const specifier of namedBindings.elements) { + const importedName = specifier.propertyName?.text ?? specifier.name.text; + if (legacyReaderNames.has(importedName)) { + violations.push({ + line: toLine(sourceFile, specifier), + reason: `imports legacy session store reader "${importedName}"`, + }); + } + } + } + } + + if (ts.isBindingElement(node)) { + const name = bindingName(node); + if (name && legacyReaderNames.has(name)) { + violations.push({ + line: toLine(sourceFile, node), + reason: `aliases legacy session store reader "${name}"`, + }); + } + } + + if (ts.isPropertyAccessExpression(node) && legacyReaderNames.has(node.name.text)) { + violations.push({ + line: toLine(sourceFile, node.name), + reason: `references legacy session store reader "${node.name.text}"`, + }); + } + + if ( + ts.isElementAccessExpression(node) && + ts.isStringLiteral(node.argumentExpression) && + legacyReaderNames.has(node.argumentExpression.text) + ) { + violations.push({ + line: toLine(sourceFile, node.argumentExpression), + reason: `references legacy session store reader "${node.argumentExpression.text}"`, + }); + } + + if (ts.isCallExpression(node)) { + const calleeName = propertyAccessName(node.expression); + if ( + calleeName && + legacyReaderNames.has(calleeName) && + ts.isIdentifier(unwrapExpression(node.expression)) + ) { + violations.push({ + line: toLine(sourceFile, node.expression), + reason: `calls legacy session store reader "${calleeName}"`, + }); + } + } + + ts.forEachChild(node, visit); + }; + + visit(sourceFile); + return violations; +} + +export async function main() { + const repoRoot = resolveRepoRoot(import.meta.url); + const sourceRoots = resolveSourceRoots(repoRoot, ["src/config/sessions", "src/gateway"]); + const violations = await collectFileViolations({ + repoRoot, + sourceRoots, + skipFile: (filePath) => + !migratedSessionAccessorFiles.has(normalizeRelativePath(path.relative(repoRoot, filePath))), + findViolations: findSessionAccessorBoundaryViolations, + }); + + if (violations.length === 0) { + console.log("session accessor boundary guard passed."); + return; + } + + console.error("Found legacy session store reader usage in session-accessor migrated files:"); + for (const violation of violations) { + console.error(`- ${violation.path}:${violation.line}: ${violation.reason}`); + } + console.error( + "Use src/config/sessions/session-accessor.ts helpers for migrated read/projection paths. Expand this ratchet only after a slice migrates more files.", + ); + process.exit(1); +} + +runAsScript(import.meta.url, main); diff --git a/src/config/sessions/combined-store-gateway.ts b/src/config/sessions/combined-store-gateway.ts index d07bbde42cbc..2f2ca678fb15 100644 --- a/src/config/sessions/combined-store-gateway.ts +++ b/src/config/sessions/combined-store-gateway.ts @@ -9,7 +9,7 @@ import { import { normalizeAgentId } from "../../routing/session-key.js"; import type { OpenClawConfig } from "../types.openclaw.js"; import { resolveStorePath } from "./paths.js"; -import { loadSessionStore } from "./store-load.js"; +import { listSessionEntries } from "./session-accessor.js"; import { resolveAgentSessionStoreTargetsSync, resolveAllAgentSessionStoreTargetsSync, @@ -22,6 +22,15 @@ function isStorePathTemplate(store?: string): boolean { return typeof store === "string" && store.includes("{agentId}"); } +function loadGatewayStoreEntries(storePath: string): Record { + return Object.fromEntries( + listSessionEntries({ clone: false, storePath }).map(({ sessionKey, entry }) => [ + sessionKey, + entry, + ]), + ); +} + function mergeSessionEntryIntoCombined(params: { cfg: OpenClawConfig; combined: Record; @@ -76,7 +85,7 @@ export function loadCombinedSessionStoreForGateway( // A single shared store still needs keys canonicalized as if owned by the default agent. const storePath = resolveStorePath(storeConfig); const defaultAgentId = normalizeAgentId(resolveDefaultAgentId(cfg)); - const store = loadSessionStore(storePath, { clone: false }); + const store = loadGatewayStoreEntries(storePath); const combined: Record = {}; for (const [key, entry] of Object.entries(store)) { const canonicalKey = resolveStoredSessionKeyForAgentStore({ @@ -108,7 +117,7 @@ export function loadCombinedSessionStoreForGateway( for (const target of targets) { const agentId = target.agentId; const storePath = target.storePath; - const store = loadSessionStore(storePath, { clone: false }); + const store = loadGatewayStoreEntries(storePath); for (const [key, entry] of Object.entries(store)) { const canonicalKey = resolveStoredSessionKeyForAgentStore({ cfg, diff --git a/src/config/sessions/session-accessor.test.ts b/src/config/sessions/session-accessor.test.ts new file mode 100644 index 000000000000..2e05db12496a --- /dev/null +++ b/src/config/sessions/session-accessor.test.ts @@ -0,0 +1,587 @@ +import fs from "node:fs"; +import os from "node:os"; +import path from "node:path"; +import { afterEach, beforeEach, describe, expect, it } from "vitest"; +import { onSessionTranscriptUpdate } from "../../sessions/transcript-events.js"; +import { + appendTranscriptMessage, + appendTranscriptEvent, + cleanupSessionLifecycleArtifacts, + listSessionEntries, + loadExactSessionEntry, + loadSessionEntry, + loadTranscriptEvents, + patchSessionEntry, + publishTranscriptUpdate, + readSessionUpdatedAt, + replaceSessionEntry, + updateSessionEntry, + upsertSessionEntry, +} from "./session-accessor.js"; +import { loadSessionStore } from "./store.js"; +import type { SessionEntry } from "./types.js"; + +describe("session accessor file-backed seam", () => { + let tempDir: string; + let storePath: string; + let transcriptPath: string; + + beforeEach(() => { + tempDir = fs.mkdtempSync(path.join(os.tmpdir(), "openclaw-session-accessor-")); + storePath = path.join(tempDir, "sessions.json"); + transcriptPath = path.join(tempDir, "session.jsonl"); + }); + + afterEach(() => { + fs.rmSync(tempDir, { recursive: true, force: true }); + }); + + it("loads, lists, and patches session entries without exposing the file store shape", async () => { + const scope = { + sessionKey: "agent:main:main", + storePath, + }; + + await upsertSessionEntry(scope, { + model: "gpt-5.5", + sessionId: "session-1", + updatedAt: 10, + }); + + expect(loadSessionEntry(scope)).toMatchObject({ + model: "gpt-5.5", + sessionId: "session-1", + updatedAt: expect.any(Number), + }); + expect(readSessionUpdatedAt(scope)).toEqual(expect.any(Number)); + expect(listSessionEntries({ storePath })).toEqual([ + { + sessionKey: "agent:main:main", + entry: expect.objectContaining({ + model: "gpt-5.5", + sessionId: "session-1", + updatedAt: expect.any(Number), + }), + }, + ]); + + await upsertSessionEntry(scope, { model: "sonnet-4.6", updatedAt: 20 }); + + expect(loadSessionEntry(scope)).toMatchObject({ + model: "sonnet-4.6", + sessionId: "session-1", + updatedAt: expect.any(Number), + }); + }); + + it("creates durable session ids for metadata-only inserts", async () => { + const scope = { + sessionKey: "agent:main:main", + storePath, + }; + + const inserted = await upsertSessionEntry(scope, { model: "gpt-5.5" }); + + expect(inserted?.sessionId).toMatch( + /^[0-9a-f]{8}-[0-9a-f]{4}-[1-5][0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$/i, + ); + expect(inserted?.sessionId).not.toBe(scope.sessionKey); + expect(loadSessionEntry(scope)?.sessionId).toBe(inserted?.sessionId); + }); + + it("can borrow cached entry objects for read-only hot paths", async () => { + const scope = { + clone: false, + sessionKey: "agent:main:main", + storePath, + }; + + await upsertSessionEntry(scope, { + sessionId: "session-1", + updatedAt: 10, + }); + const cachedStore = loadSessionStore(storePath, { clone: false }); + + expect(loadSessionEntry(scope)).toBe(cachedStore["agent:main:main"]); + expect(listSessionEntries({ clone: false, storePath })[0]?.entry).toBe( + cachedStore["agent:main:main"], + ); + }); + + it("keeps exact persisted-key lookup separate from canonical entry reads", async () => { + fs.writeFileSync( + storePath, + JSON.stringify({ + "agent:main:main": { + sessionId: "session-1", + updatedAt: 10, + model: "gpt-5.5", + }, + }), + "utf8", + ); + + const mixedCaseScope = { + sessionKey: "AGENT:MAIN:MAIN", + storePath, + }; + + expect(loadSessionEntry(mixedCaseScope)?.sessionId).toBe("session-1"); + expect(loadExactSessionEntry(mixedCaseScope)).toBeUndefined(); + expect(loadExactSessionEntry({ sessionKey: "agent:main:main", storePath })).toEqual({ + sessionKey: "agent:main:main", + entry: expect.objectContaining({ + sessionId: "session-1", + model: "gpt-5.5", + }), + }); + }); + + it("updates existing entries without creating missing sessions", async () => { + const scope = { + sessionKey: "agent:main:main", + storePath, + }; + + await expect(updateSessionEntry(scope, () => ({ model: "gpt-5.5" }))).resolves.toBeNull(); + expect(listSessionEntries({ storePath })).toEqual([]); + + await upsertSessionEntry(scope, { + sessionId: "session-1", + updatedAt: 10, + }); + const beforeNullUpdate = loadSessionEntry(scope); + await expect(updateSessionEntry(scope, () => null)).resolves.toEqual(beforeNullUpdate); + expect(loadSessionEntry(scope)).toMatchObject({ + sessionId: "session-1", + updatedAt: beforeNullUpdate?.updatedAt, + }); + await expect( + updateSessionEntry(scope, () => ({ model: "gpt-5.5", updatedAt: 20 })), + ).resolves.toMatchObject({ + model: "gpt-5.5", + sessionId: "session-1", + updatedAt: expect.any(Number), + }); + }); + + it("replaces entries so deleted fields stay removed", async () => { + const scope = { + sessionKey: "agent:main:main", + storePath, + }; + + await upsertSessionEntry(scope, { + model: "gpt-5.5", + providerOverride: "openai", + sessionId: "session-1", + updatedAt: 10, + }); + + await replaceSessionEntry(scope, { + sessionId: "session-1", + updatedAt: 20, + }); + + expect(loadSessionEntry(scope)).toMatchObject({ + sessionId: "session-1", + updatedAt: expect.any(Number), + }); + expect(loadSessionEntry(scope)?.model).toBeUndefined(); + expect(loadSessionEntry(scope)?.providerOverride).toBeUndefined(); + }); + + it("patches entries atomically with a fallback entry", async () => { + const scope = { + sessionKey: "agent:main:main", + storePath, + }; + let missingContextEntry: SessionEntry | undefined; + let existingContextEntry: SessionEntry | undefined; + + await patchSessionEntry( + scope, + (entry, context) => { + missingContextEntry = context.existingEntry; + return { + ...entry, + model: "gpt-5.5", + }; + }, + { + fallbackEntry: { + sessionId: "session-1", + updatedAt: 10, + }, + replaceEntry: true, + }, + ); + + await patchSessionEntry( + scope, + (entry, context) => { + existingContextEntry = context.existingEntry; + return { + ...entry, + model: undefined, + providerOverride: "openai", + }; + }, + { replaceEntry: true }, + ); + + expect(missingContextEntry).toBeUndefined(); + expect(existingContextEntry).toMatchObject({ model: "gpt-5.5" }); + expect(loadSessionEntry(scope)).toMatchObject({ + providerOverride: "openai", + sessionId: "session-1", + }); + expect(loadSessionEntry(scope)?.model).toBeUndefined(); + }); + + it("can patch metadata without refreshing session activity", async () => { + const scope = { + sessionKey: "agent:main:main", + storePath, + }; + + await upsertSessionEntry(scope, { + sessionId: "session-1", + updatedAt: 10, + }); + const beforePatch = loadSessionEntry(scope); + + await patchSessionEntry( + scope, + () => ({ + model: "gpt-5.5", + updatedAt: 20, + }), + { preserveActivity: true }, + ); + + expect(loadSessionEntry(scope)).toMatchObject({ + model: "gpt-5.5", + sessionId: "session-1", + updatedAt: beforePatch?.updatedAt, + }); + }); + + it("cleans scoped lifecycle entries and unreferenced transcript artifacts", async () => { + const nowMs = Date.now(); + const oldDate = new Date(nowMs - 600_000); + const lifecycleSessionsDir = path.join(tempDir, "state", "agents", "main", "sessions"); + const lifecycleStorePath = path.join(lifecycleSessionsDir, "sessions.json"); + const removedTranscriptPath = path.join(lifecycleSessionsDir, "removed-lifecycle.jsonl"); + const customTranscriptPath = path.join(lifecycleSessionsDir, "custom-lifecycle-old.jsonl"); + const freshDefaultTranscriptPath = path.join(lifecycleSessionsDir, "custom-lifecycle.jsonl"); + const freshTranscriptPath = path.join(lifecycleSessionsDir, "fresh-lifecycle.jsonl"); + const referencedTranscriptPath = path.join(lifecycleSessionsDir, "referenced.jsonl"); + const orphanTranscriptPath = path.join(lifecycleSessionsDir, "orphan-lifecycle.jsonl"); + const siblingDir = path.join(tempDir, "state", "agents", "sibling", "sessions"); + const siblingTranscriptPath = path.join(siblingDir, "sibling-lifecycle.jsonl"); + fs.mkdirSync(lifecycleSessionsDir, { recursive: true }); + fs.mkdirSync(siblingDir, { recursive: true }); + + fs.writeFileSync( + lifecycleStorePath, + JSON.stringify({ + "agent:main:lifecycle-cleanup-removed": { + sessionId: "removed-lifecycle", + }, + "agent:main:lifecycle-cleanup-fresh": { + sessionId: "fresh-lifecycle", + }, + "agent:main:lifecycle-cleanup-custom": { + sessionFile: "custom-lifecycle-old.jsonl", + sessionId: "custom-lifecycle", + }, + "agent:main:lifecycle-cleanup-sibling": { + sessionFile: siblingTranscriptPath, + sessionId: "sibling-lifecycle", + }, + "agent:main:telegram:group:lifecycle-cleanup-room": { + sessionId: "kept-by-segment", + }, + "agent:main:regular": { + sessionId: "referenced", + }, + }), + "utf-8", + ); + fs.writeFileSync(removedTranscriptPath, '{"runId":"lifecycle-marker-removed"}\n', "utf-8"); + fs.writeFileSync(customTranscriptPath, '{"runId":"lifecycle-marker-custom"}\n', "utf-8"); + fs.writeFileSync(freshDefaultTranscriptPath, '{"runId":"lifecycle-marker-default"}\n', "utf-8"); + fs.writeFileSync(freshTranscriptPath, '{"runId":"lifecycle-marker-fresh"}\n', "utf-8"); + fs.writeFileSync(siblingTranscriptPath, '{"runId":"lifecycle-marker-sibling"}\n', "utf-8"); + fs.writeFileSync( + referencedTranscriptPath, + '{"runId":"lifecycle-marker-referenced"}\n', + "utf-8", + ); + fs.writeFileSync(orphanTranscriptPath, '{"runId":"lifecycle-marker-orphan"}\n', "utf-8"); + fs.utimesSync(removedTranscriptPath, oldDate, oldDate); + fs.utimesSync(customTranscriptPath, oldDate, oldDate); + fs.utimesSync(siblingTranscriptPath, oldDate, oldDate); + fs.utimesSync(referencedTranscriptPath, oldDate, oldDate); + fs.utimesSync(orphanTranscriptPath, oldDate, oldDate); + + const result = await cleanupSessionLifecycleArtifacts({ + storePath: lifecycleStorePath, + sessionKeySegmentPrefix: "lifecycle-cleanup-", + transcriptContentMarker: "lifecycle-marker-", + orphanTranscriptMinAgeMs: 300_000, + nowMs, + }); + + expect(result).toEqual({ removedEntries: 3, archivedTranscriptArtifacts: 3 }); + const loaded = loadSessionStore(lifecycleStorePath, { skipCache: true }); + expect(loaded).not.toHaveProperty("agent:main:lifecycle-cleanup-removed"); + expect(loaded).not.toHaveProperty("agent:main:lifecycle-cleanup-custom"); + expect(loaded).not.toHaveProperty("agent:main:lifecycle-cleanup-sibling"); + expect(loaded).toHaveProperty("agent:main:lifecycle-cleanup-fresh"); + expect(loaded).toHaveProperty("agent:main:telegram:group:lifecycle-cleanup-room"); + expect(loaded).toHaveProperty("agent:main:regular"); + const files = fs.readdirSync(lifecycleSessionsDir); + expect( + files.filter((file) => file.startsWith("removed-lifecycle.jsonl.deleted.")), + ).toHaveLength(1); + expect(files.filter((file) => file.startsWith("orphan-lifecycle.jsonl.deleted."))).toHaveLength( + 1, + ); + expect( + files.filter((file) => file.startsWith("custom-lifecycle-old.jsonl.deleted.")), + ).toHaveLength(1); + expect(files).toContain("custom-lifecycle.jsonl"); + expect(files).toContain("fresh-lifecycle.jsonl"); + expect(files).toContain("referenced.jsonl"); + expect(fs.existsSync(siblingTranscriptPath)).toBe(true); + expect(fs.readdirSync(siblingDir)).toEqual(["sibling-lifecycle.jsonl"]); + }); + + it("loads and appends transcript events through a session scope", async () => { + const scope = { + sessionFile: transcriptPath, + sessionId: "session-1", + sessionKey: "agent:main:main", + storePath, + }; + const event = { + payload: { value: "hello" }, + type: "metadata", + }; + + await appendTranscriptEvent(scope, { type: "session", sessionId: "session-1" }); + await appendTranscriptEvent(scope, event); + + await expect(loadTranscriptEvents(scope)).resolves.toEqual([ + { type: "session", sessionId: "session-1" }, + event, + ]); + expect(fs.statSync(transcriptPath).mode & 0o777).toBe(0o600); + }); + + it("appends to an explicit transcript artifact without a session key", async () => { + const scope = { + sessionFile: transcriptPath, + sessionId: "session-1", + storePath, + }; + const event = { + payload: { value: "keyless" }, + type: "metadata", + }; + + await appendTranscriptEvent(scope, event); + + await expect(loadTranscriptEvents(scope)).resolves.toEqual([event]); + // Explicit-artifact writes never touch entry metadata: no entry appears. + expect(listSessionEntries({ storePath })).toEqual([]); + }); + + it("rejects transcript writes without a session key or explicit file", async () => { + await expect( + appendTranscriptEvent({ sessionId: "session-1", storePath }, { type: "metadata" }), + ).rejects.toThrow(/session key or explicit session file/); + }); + + it("rejects raw message transcript events", async () => { + const scope = { + sessionFile: transcriptPath, + sessionId: "session-1", + sessionKey: "agent:main:main", + storePath, + }; + + await expect( + appendTranscriptEvent(scope, { + id: "msg-1", + message: { role: "user", content: "hello" }, + parentId: null, + type: "message", + }), + ).rejects.toThrow(/appendTranscriptMessage/); + expect(fs.existsSync(transcriptPath)).toBe(false); + }); + + it("loads transcript events without a session key when the read target is explicit", async () => { + const scope = { + sessionFile: transcriptPath, + sessionId: "session-1", + }; + const event = { + payload: { value: "hello" }, + type: "metadata", + }; + + await appendTranscriptEvent( + { + ...scope, + sessionKey: "agent:main:main", + storePath, + }, + event, + ); + + await expect(loadTranscriptEvents(scope)).resolves.toEqual([event]); + }); + + it("loads transcript events from a generated read target without a session key", async () => { + const event = { + payload: { value: "hello" }, + type: "metadata", + }; + + fs.writeFileSync(path.join(tempDir, "session-1.jsonl"), `${JSON.stringify(event)}\n`, "utf-8"); + + await expect( + loadTranscriptEvents({ + sessionId: "session-1", + storePath, + }), + ).resolves.toEqual([event]); + }); + + it("appends messages and publishes updates through a session scope", async () => { + const scope = { + agentId: "main", + sessionFile: transcriptPath, + sessionId: "session-1", + sessionKey: "agent:main:main", + storePath, + }; + const updates: unknown[] = []; + const unsubscribe = onSessionTranscriptUpdate((update) => { + updates.push(update); + }); + + const appended = await appendTranscriptMessage(scope, { + cwd: tempDir, + idempotencyLookup: "scan", + message: { + role: "assistant", + content: "hello", + idempotencyKey: "assistant-once", + }, + }); + const replayed = await appendTranscriptMessage(scope, { + cwd: tempDir, + idempotencyLookup: "scan", + message: { + role: "assistant", + content: "hello again", + idempotencyKey: "assistant-once", + }, + }); + await publishTranscriptUpdate(scope, { + agentId: "main", + message: appended.message, + messageId: appended.messageId, + sessionKey: scope.sessionKey, + }); + unsubscribe(); + + expect(replayed).toMatchObject({ + appended: false, + messageId: appended.messageId, + message: expect.objectContaining({ + content: "hello", + idempotencyKey: "assistant-once", + }), + }); + await expect(loadTranscriptEvents(scope)).resolves.toEqual([ + expect.objectContaining({ type: "session" }), + expect.objectContaining({ + id: appended.messageId, + message: expect.objectContaining({ + content: "hello", + idempotencyKey: "assistant-once", + }), + type: "message", + }), + ]); + expect(updates).toEqual([ + { + agentId: "main", + message: appended.message, + messageId: appended.messageId, + sessionFile: transcriptPath, + sessionKey: scope.sessionKey, + }, + ]); + }); + + it("honors thread fallback paths when resolving transcript scope from the store", async () => { + const scope = { + agentId: "main", + sessionId: "session-1", + sessionKey: "agent:main:demo-channel:1234:thread:456", + storePath, + }; + const event = { + payload: { value: "hello" }, + type: "metadata", + }; + + await upsertSessionEntry(scope, { + sessionId: scope.sessionId, + updatedAt: 10, + }); + await appendTranscriptEvent(scope, event); + + const expectedTranscriptPath = path.join(tempDir, "session-1-topic-456.jsonl"); + expect(fs.existsSync(expectedTranscriptPath)).toBe(true); + expect(fs.existsSync(path.join(tempDir, "session-1.jsonl"))).toBe(false); + expect(fs.realpathSync(loadSessionEntry(scope)?.sessionFile ?? "")).toBe( + fs.realpathSync(expectedTranscriptPath), + ); + await expect(loadTranscriptEvents(scope)).resolves.toEqual([event]); + }); + + it("persists transcript metadata under the normalized session key", async () => { + const canonicalScope = { + sessionId: "session-1", + sessionKey: "agent:main:main", + storePath, + }; + + await upsertSessionEntry(canonicalScope, { + sessionId: canonicalScope.sessionId, + updatedAt: 10, + }); + await appendTranscriptEvent( + { + agentId: "main", + sessionId: canonicalScope.sessionId, + sessionKey: "AGENT:MAIN:MAIN", + storePath, + }, + { id: "event-1", type: "metadata" }, + ); + + expect(listSessionEntries({ storePath }).map((entry) => entry.sessionKey)).toEqual([ + canonicalScope.sessionKey, + ]); + expect(loadSessionEntry(canonicalScope)?.sessionFile).toBeTruthy(); + }); +}); diff --git a/src/config/sessions/session-accessor.ts b/src/config/sessions/session-accessor.ts new file mode 100644 index 000000000000..e8f830fc1a02 --- /dev/null +++ b/src/config/sessions/session-accessor.ts @@ -0,0 +1,480 @@ +import { randomUUID } from "node:crypto"; +import path from "node:path"; +import { resolveAgentIdFromSessionKey } from "../../routing/session-key.js"; +import { emitSessionTranscriptUpdate } from "../../sessions/transcript-events.js"; +import type { SessionTranscriptUpdate } from "../../sessions/transcript-events.js"; +import { getRuntimeConfig } from "../io.js"; +import type { OpenClawConfig } from "../types.openclaw.js"; +import { + resolveSessionTranscriptPath, + resolveSessionTranscriptPathInDir, + resolveStorePath, +} from "./paths.js"; +import { resolveAndPersistSessionFile } from "./session-file.js"; +import { + getSessionEntry, + cleanupSessionLifecycleArtifacts as cleanupFileSessionLifecycleArtifacts, + listSessionEntries as listFileSessionEntries, + loadSessionStore, + patchSessionEntry as patchFileSessionEntry, + readSessionUpdatedAt as readFileSessionUpdatedAt, + resolveSessionStoreEntry, + updateSessionStoreEntry as updateFileSessionStoreEntry, + type SessionLifecycleArtifactCleanupParams, + type SessionLifecycleArtifactCleanupResult, +} from "./store.js"; +import { parseSessionThreadInfo } from "./thread-info.js"; +import { + appendSessionTranscriptEvent, + appendSessionTranscriptMessage, +} from "./transcript-append.js"; +import { streamSessionTranscriptLines } from "./transcript-stream.js"; +import { resolveSessionTranscriptFile } from "./transcript.js"; +import type { SessionEntry } from "./types.js"; + +/** + * Session access API for callers that need entries or transcripts without + * depending on the persisted store layout. Callers provide stable session + * identity, and this module resolves the current entry/transcript target while + * preserving canonical-key, transcript-linking, and update-notification rules. + * + * Ownership contract (#88838): this accessor is the permanent storage-neutral + * domain boundary for session/transcript runtime access; the SQLite storage + * flip implements this interface. The entry workflow helpers in store.ts are + * the file-backend implementation it delegates to plus the plugin-SDK + * deprecation-window surface (RFC 0007); they become internal as direct + * callers migrate here. New runtime callers use this module, not store.ts. + */ +export type SessionAccessScope = { + /** Agent owner used when the session key does not already encode one. */ + agentId?: string; + /** + * Set false only for internal read-only hot paths that will not retain or + * mutate the returned entry. + */ + clone?: boolean; + /** Environment override used when resolving agent-scoped store paths in tests/tools. */ + env?: NodeJS.ProcessEnv; + /** Set false for metadata-only reads that do not need hydrated prompt refs. */ + hydrateSkillPromptRefs?: boolean; + /** Canonical or alias session key for the entry being read or written. */ + sessionKey: string; + /** Explicit store path for callers that already resolved the owning store. */ + storePath?: string; +}; + +export type SessionTranscriptReadScope = Omit & { + /** Explicit transcript file path; bypasses store lookup when already known. */ + sessionFile?: string; + /** Runtime session id used to derive a transcript file when no explicit file is provided. */ + sessionId: string; + /** Optional key for read callers that can resolve via the session entry. */ + sessionKey?: string; + /** Channel thread suffix used when deriving topic transcript paths. */ + threadId?: string | number; +}; + +export type SessionTranscriptAccessScope = SessionTranscriptReadScope & { + /** + * Identifies the owning entry when the transcript target must be resolved + * (and possibly persisted) through the session store. May be omitted only + * when an explicit sessionFile binds the operation to a concrete artifact; + * such writes never read or update entry metadata. + */ + sessionKey?: string; +}; + +export type SessionTranscriptWriteScope = Omit & { + /** Optional for appenders that can operate on an existing explicit transcript target. */ + sessionId?: string; +}; + +export type SessionEntrySummary = { + /** Persisted key for the entry. */ + sessionKey: string; + /** Entry value cloned from the backing store unless the caller requested borrowed reads. */ + entry: SessionEntry; +}; + +/** Session entry read by the exact persisted session key, without alias resolution. */ +export type ExactSessionEntry = { + sessionKey: string; + entry: SessionEntry; +}; + +/** Raw transcript record for non-message events; message records use appendTranscriptMessage. */ +export type TranscriptEvent = unknown; + +export type TranscriptMessageAppendOptions = { + /** Runtime config used for message redaction and transcript header metadata. */ + config?: OpenClawConfig; + /** Working directory recorded in a newly created transcript header. */ + cwd?: string; + /** How duplicate message idempotency keys are detected before append. */ + idempotencyLookup?: "scan" | "caller-checked"; + /** Provider/channel message payload to persist. */ + message: TMessage; + /** Testable timestamp override for the generated transcript entry. */ + now?: number; + /** Optional finalizer that runs after duplicate detection but before persistence. */ + prepareMessageAfterIdempotencyCheck?: (message: TMessage) => TMessage | undefined; + /** Allow append without parent-link migration for large legacy linear transcripts. */ + useRawWhenLinear?: boolean; +}; + +export type TranscriptMessageAppendResult = { + /** False when idempotency lookup found an existing transcript message. */ + appended: boolean; + /** Redacted message payload as persisted or replayed from the transcript. */ + message: TMessage; + /** Existing or newly generated transcript message id. */ + messageId: string; +}; + +/** Transcript update fields supplied by callers; sessionFile is resolved here. */ +export type TranscriptUpdatePayload = Omit; + +export type SessionEntryUpdateOptions = { + /** Skip prune/cap/rotation maintenance for specialized internal updates. */ + skipMaintenance?: boolean; + /** Let the writer cache retain the updated object without cloning. */ + takeCacheOwnership?: boolean; +}; + +export type SessionEntryPatchOptions = { + /** Entry to synthesize when a patch operation is allowed to create. */ + fallbackEntry?: SessionEntry; + /** Keep the previous updatedAt value when the patch should not count as activity. */ + preserveActivity?: boolean; + /** Replace the whole entry instead of merging the returned patch. */ + replaceEntry?: boolean; +}; + +export type SessionEntryPatchContext = { + /** Present when the patched entry already existed before fallback synthesis. */ + existingEntry?: SessionEntry; +}; + +export type { SessionLifecycleArtifactCleanupParams, SessionLifecycleArtifactCleanupResult }; + +/** Returns the entry for a canonical or alias session key, if one exists. */ +export function loadSessionEntry(scope: SessionAccessScope): SessionEntry | undefined { + if (scope.clone === false) { + const store = loadSessionStore(resolveAccessStorePath(scope), { + clone: false, + ...(scope.hydrateSkillPromptRefs === false ? { hydrateSkillPromptRefs: false } : {}), + }); + return resolveSessionStoreEntry({ store, sessionKey: scope.sessionKey }).existing; + } + return getSessionEntry(scope); +} + +/** + * Returns only the row persisted under the exact key provided. + * Use this for authorization-sensitive routing where alias canonicalization + * could cross an account or agent boundary. + */ +export function loadExactSessionEntry(scope: SessionAccessScope): ExactSessionEntry | undefined { + const sessionKey = scope.sessionKey.trim(); + if (!sessionKey) { + return undefined; + } + const store = loadSessionStore(resolveAccessStorePath(scope), { + ...(scope.clone === false ? { clone: false } : {}), + ...(scope.hydrateSkillPromptRefs === false ? { hydrateSkillPromptRefs: false } : {}), + }); + const entry = Object.hasOwn(store, sessionKey) ? store[sessionKey] : undefined; + return entry ? { sessionKey, entry } : undefined; +} + +/** Lists entries from the resolved store, preserving the persisted key for each row. */ +export function listSessionEntries( + scope: Partial> = {}, +): SessionEntrySummary[] { + if (scope.clone === false) { + return Object.entries( + loadSessionStore(resolveAccessStorePath({ ...scope, sessionKey: "" }), { + clone: false, + ...(scope.hydrateSkillPromptRefs === false ? { hydrateSkillPromptRefs: false } : {}), + }), + ).map(([sessionKey, entry]) => ({ sessionKey, entry })); + } + return listFileSessionEntries(scope); +} + +/** Reads the last activity timestamp for one session entry, or undefined when absent. */ +export function readSessionUpdatedAt(scope: SessionAccessScope): number | undefined { + if (scope.storePath) { + return readFileSessionUpdatedAt({ + storePath: scope.storePath, + sessionKey: scope.sessionKey, + }); + } + return loadSessionEntry(scope)?.updatedAt; +} + +/** Creates or updates one entry from a partial patch and returns the persisted entry. */ +export async function upsertSessionEntry( + scope: SessionAccessScope, + patch: Partial, +): Promise { + return await patchFileSessionEntry({ + ...scope, + fallbackEntry: createFallbackSessionEntry(patch), + update: () => patch, + }); +} + +/** Replaces one entry with the supplied value and returns the persisted entry. */ +export async function replaceSessionEntry( + scope: SessionAccessScope, + entry: SessionEntry, +): Promise { + return await patchFileSessionEntry({ + ...scope, + fallbackEntry: entry, + replaceEntry: true, + update: () => entry, + }); +} + +/** + * Applies an atomic patch to one entry. + * The updater sees the current entry plus whether it was synthesized from a + * fallback; returning null skips persistence. + */ +export async function patchSessionEntry( + scope: SessionAccessScope, + update: ( + entry: SessionEntry, + context: SessionEntryPatchContext, + ) => Promise | null> | Partial | null, + options: SessionEntryPatchOptions = {}, +): Promise { + return await patchFileSessionEntry({ + ...scope, + fallbackEntry: options.fallbackEntry, + preserveActivity: options.preserveActivity, + replaceEntry: options.replaceEntry, + update, + }); +} + +/** Updates an existing entry only; returns null when the session is absent. */ +export async function updateSessionEntry( + scope: SessionAccessScope, + update: ( + entry: SessionEntry, + ) => Promise | null> | Partial | null, + options: SessionEntryUpdateOptions = {}, +): Promise { + return await updateFileSessionStoreEntry({ + storePath: resolveAccessStorePath(scope), + sessionKey: scope.sessionKey, + skipMaintenance: options.skipMaintenance, + takeCacheOwnership: options.takeCacheOwnership, + update, + }); +} + +/** Removes entries and orphan transcript artifacts owned by a named session lifecycle. */ +export async function cleanupSessionLifecycleArtifacts( + params: SessionLifecycleArtifactCleanupParams, +): Promise { + return await cleanupFileSessionLifecycleArtifacts(params); +} + +/** Reads parsed transcript records from an explicit or derived transcript target. */ +export async function loadTranscriptEvents( + scope: SessionTranscriptReadScope, +): Promise { + const transcript = await resolveTranscriptReadAccess(scope); + const events: TranscriptEvent[] = []; + for await (const line of streamSessionTranscriptLines(transcript.sessionFile)) { + events.push(JSON.parse(line) as TranscriptEvent); + } + return events; +} + +/** + * Appends a non-message transcript record such as session or metadata events. + * Message records must use appendTranscriptMessage so parent links, idempotency, + * and redaction are preserved. + */ +export async function appendTranscriptEvent( + scope: SessionTranscriptAccessScope, + event: TranscriptEvent, +): Promise { + assertNonMessageTranscriptEvent(event); + const transcript = await resolveTranscriptAccess(scope); + await appendSessionTranscriptEvent({ + event, + transcriptPath: transcript.sessionFile, + }); +} + +function assertNonMessageTranscriptEvent(event: TranscriptEvent): void { + if (!event || typeof event !== "object" || Array.isArray(event)) { + return; + } + // Message records require parent-link, idempotency, and redaction handling + // from appendTranscriptMessage; raw event writes would bypass those invariants. + if ((event as { type?: unknown }).type === "message") { + throw new Error( + "appendTranscriptEvent cannot write message transcript records; use appendTranscriptMessage instead.", + ); + } +} + +/** + * Appends one transcript message with message-id generation and optional + * idempotency lookup. The returned message is the redacted persisted value. + */ +export async function appendTranscriptMessage( + scope: SessionTranscriptWriteScope, + options: TranscriptMessageAppendOptions & { + prepareMessageAfterIdempotencyCheck: (message: TMessage) => TMessage | undefined; + }, +): Promise | undefined>; +export async function appendTranscriptMessage( + scope: SessionTranscriptWriteScope, + options: TranscriptMessageAppendOptions, +): Promise>; +export async function appendTranscriptMessage( + scope: SessionTranscriptWriteScope, + options: TranscriptMessageAppendOptions, +): Promise | undefined> { + const transcript = await resolveTranscriptAccess(scope); + return await appendSessionTranscriptMessage({ + transcriptPath: transcript.sessionFile, + message: options.message, + ...(scope.sessionId ? { sessionId: scope.sessionId } : {}), + ...(options.cwd ? { cwd: options.cwd } : {}), + ...(options.config ? { config: options.config } : {}), + ...(options.idempotencyLookup ? { idempotencyLookup: options.idempotencyLookup } : {}), + ...(options.now !== undefined ? { now: options.now } : {}), + ...(options.prepareMessageAfterIdempotencyCheck + ? { prepareMessageAfterIdempotencyCheck: options.prepareMessageAfterIdempotencyCheck } + : {}), + ...(options.useRawWhenLinear !== undefined + ? { useRawWhenLinear: options.useRawWhenLinear } + : {}), + }); +} + +/** Emits a transcript update after resolving the current transcript target. */ +export async function publishTranscriptUpdate( + scope: SessionTranscriptWriteScope, + update: TranscriptUpdatePayload = {}, +): Promise { + const transcript = await resolveTranscriptAccess(scope); + emitSessionTranscriptUpdate({ + ...update, + sessionFile: transcript.sessionFile, + }); +} + +function createFallbackSessionEntry(patch: Partial): SessionEntry { + const now = Date.now(); + return { + sessionId: patch.sessionId ?? randomUUID(), + updatedAt: patch.updatedAt ?? now, + ...patch, + }; +} + +function resolveAccessStorePath(scope: SessionAccessScope): string { + if (scope.storePath) { + return scope.storePath; + } + const agentId = scope.agentId ?? resolveAgentIdFromSessionKey(scope.sessionKey); + return resolveStorePath(getRuntimeConfig().session?.store, { + agentId, + env: scope.env, + }); +} + +async function resolveTranscriptReadAccess(scope: SessionTranscriptReadScope): Promise<{ + sessionFile: string; +}> { + if (scope.sessionFile?.trim()) { + return { sessionFile: scope.sessionFile }; + } + if (scope.sessionKey) { + return await resolveTranscriptAccess({ ...scope, sessionKey: scope.sessionKey }); + } + if (scope.storePath) { + return { + sessionFile: resolveSessionTranscriptPathInDir( + scope.sessionId, + path.dirname(path.resolve(scope.storePath)), + scope.threadId, + ), + }; + } + if (scope.agentId) { + return { + sessionFile: resolveSessionTranscriptPath(scope.sessionId, scope.agentId, scope.threadId), + }; + } + throw new Error(`Cannot resolve transcript read scope without a session target`); +} + +async function resolveTranscriptAccess(scope: SessionTranscriptWriteScope): Promise<{ + sessionFile: string; +}> { + if (scope.sessionFile?.trim()) { + return { sessionFile: scope.sessionFile }; + } + // Past this point resolution goes through the session entry, so the owning + // key is mandatory; explicit-artifact writes returned above never need it. + const scopeSessionKey = scope.sessionKey?.trim(); + if (!scopeSessionKey) { + throw new Error( + "Cannot resolve a transcript write scope without a session key or explicit session file", + ); + } + if (!scope.sessionId) { + throw new Error(`Cannot resolve transcript scope without a session id: ${scopeSessionKey}`); + } + const agentId = scope.agentId ?? resolveAgentIdFromSessionKey(scopeSessionKey); + if (!agentId) { + throw new Error(`Cannot resolve transcript scope without an agent id: ${scopeSessionKey}`); + } + const sessionStore = scope.storePath + ? loadSessionStore(scope.storePath, { skipCache: true }) + : undefined; + const resolvedStoreEntry = sessionStore + ? resolveSessionStoreEntry({ store: sessionStore, sessionKey: scopeSessionKey }) + : undefined; + const sessionEntry = + resolvedStoreEntry?.existing ?? loadSessionEntry({ ...scope, sessionKey: scopeSessionKey }); + const sessionKey = resolvedStoreEntry?.normalizedKey ?? scopeSessionKey; + if (sessionStore && scope.storePath) { + const sessionsDir = path.dirname(path.resolve(scope.storePath)); + const threadId = scope.threadId ?? parseSessionThreadInfo(scopeSessionKey).threadId; + const fallbackSessionFile = + !sessionEntry?.sessionFile && threadId !== undefined + ? resolveSessionTranscriptPathInDir(scope.sessionId, sessionsDir, threadId) + : undefined; + return await resolveAndPersistSessionFile({ + agentId, + fallbackSessionFile, + sessionEntry, + sessionId: scope.sessionId, + sessionKey, + sessionStore, + sessionsDir, + storePath: scope.storePath, + }); + } + return await resolveSessionTranscriptFile({ + agentId, + sessionEntry, + sessionId: scope.sessionId, + sessionKey: scopeSessionKey, + ...(sessionStore ? { sessionStore } : {}), + ...(scope.storePath ? { storePath: scope.storePath } : {}), + ...(scope.threadId !== undefined ? { threadId: scope.threadId } : {}), + }); +} diff --git a/src/config/sessions/store.runtime.ts b/src/config/sessions/store.runtime.ts index 174a2b8a5191..53e6249ecea5 100644 --- a/src/config/sessions/store.runtime.ts +++ b/src/config/sessions/store.runtime.ts @@ -1,6 +1,11 @@ // Runtime facade for session store mutation helpers. export { applySessionStoreEntryPatch, + cleanupSessionLifecycleArtifacts, updateSessionStore, updateSessionStoreEntry, } from "./store.js"; +export type { + SessionLifecycleArtifactCleanupParams, + SessionLifecycleArtifactCleanupResult, +} from "./store.js"; diff --git a/src/config/sessions/store.ts b/src/config/sessions/store.ts index b8ada40afa8c..96fffe3fc8ee 100644 --- a/src/config/sessions/store.ts +++ b/src/config/sessions/store.ts @@ -5,6 +5,7 @@ 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 { emitSessionTranscriptUpdate } from "../../sessions/transcript-events.js"; import { deliveryContextFromChannelRoute, deliveryContextFromSession, @@ -15,9 +16,10 @@ import { import type { DeliveryContext } from "../../utils/delivery-context.types.js"; import { getFileStatSnapshot } from "../cache-utils.js"; import { getRuntimeConfig } from "../io.js"; +import { formatSessionArchiveTimestamp } from "./artifacts.js"; import { enforceSessionDiskBudget, type SessionDiskBudgetSweepResult } from "./disk-budget.js"; import { deriveSessionMetaPatch } from "./metadata.js"; -import { resolveStorePath } from "./paths.js"; +import { resolveSessionFilePath, resolveStorePath } from "./paths.js"; import { ensureSessionStorePromptBlobsForPersistence, isSessionSkillPromptBlobReadable, @@ -186,6 +188,10 @@ type SingleEntryPersistencePatch = { entry: SessionEntry; }; +// The entry workflow helpers below are the file-backend implementation behind +// the session-accessor domain boundary and the plugin-SDK compatibility +// surface (RFC 0007). Internal runtime callers use session-accessor.ts; these +// become internal as direct callers migrate (#88838). type SessionEntryWorkflowOptions = { agentId?: string; env?: NodeJS.ProcessEnv; @@ -193,6 +199,24 @@ type SessionEntryWorkflowOptions = { storePath?: string; }; +export type SessionLifecycleArtifactCleanupParams = { + /** Session store to clean. */ + storePath: string; + /** Matches the persisted session-key segment after `agent::`. */ + sessionKeySegmentPrefix: string; + /** Marker that identifies transcript artifacts owned by this lifecycle. */ + transcriptContentMarker: string; + /** Minimum age before a present transcript can be reclaimed or archived. */ + orphanTranscriptMinAgeMs: number; + /** Testable clock override. */ + nowMs?: number; +}; + +export type SessionLifecycleArtifactCleanupResult = { + removedEntries: number; + archivedTranscriptArtifacts: number; +}; + function cloneSessionEntry(entry: SessionEntry): SessionEntry { return cloneSessionStoreRecord({ entry }).entry; } @@ -505,6 +529,75 @@ function sessionEntriesHaveSameSerializedForm( return previous !== undefined && JSON.stringify(previous) === JSON.stringify(next); } +function normalizePathForLifecycleComparison(filePath: string): string { + try { + return path.normalize(fs.realpathSync(filePath)); + } catch { + return path.normalize(path.resolve(filePath)); + } +} + +function sessionKeySegmentStartsWith(sessionKey: string, prefix: string): boolean { + const firstSeparator = sessionKey.indexOf(":"); + if (firstSeparator < 0) { + return sessionKey.startsWith(prefix); + } + const secondSeparator = sessionKey.indexOf(":", firstSeparator + 1); + const sessionSegment = secondSeparator < 0 ? sessionKey : sessionKey.slice(secondSeparator + 1); + return sessionSegment.startsWith(prefix); +} + +function resolveLifecycleTranscriptPath(params: { + entry: SessionEntry | undefined; + sessionsDir: string; +}): string | null { + const sessionId = params.entry?.sessionId?.trim(); + if (!sessionId) { + return null; + } + try { + return resolveSessionFilePath(sessionId, params.entry, { sessionsDir: params.sessionsDir }); + } catch { + return null; + } +} + +function lifecycleTranscriptIsReclaimable(params: { + transcriptPath: string | null; + nowMs: number; + orphanTranscriptMinAgeMs: number; +}): boolean { + if (!params.transcriptPath || !fs.existsSync(params.transcriptPath)) { + return true; + } + try { + const stat = fs.statSync(params.transcriptPath); + return params.nowMs - stat.mtimeMs >= params.orphanTranscriptMinAgeMs; + } catch { + return true; + } +} + +function archiveExactLifecycleTranscriptPath(params: { + sessionsDir: string; + transcriptPath: string; +}): number { + const resolvedSessionsDir = normalizePathForLifecycleComparison(params.sessionsDir); + const resolvedTranscriptPath = normalizePathForLifecycleComparison(params.transcriptPath); + const relative = path.relative(resolvedSessionsDir, resolvedTranscriptPath); + if (!relative || relative.startsWith("..") || path.isAbsolute(relative)) { + return 0; + } + const archivedPath = `${resolvedTranscriptPath}.deleted.${formatSessionArchiveTimestamp()}`; + try { + fs.renameSync(resolvedTranscriptPath, archivedPath); + emitSessionTranscriptUpdate({ sessionFile: archivedPath }); + return 1; + } catch { + return 0; + } +} + async function saveSessionStoreUnlocked( storePath: string, store: Record, @@ -831,6 +924,161 @@ export async function updateSessionStore( }); } +async function archiveUnreferencedLifecycleTranscriptArtifacts(params: { + storePath: string; + transcriptContentMarker: string; + orphanTranscriptMinAgeMs: number; + nowMs: number; +}): Promise { + const sessionsDir = path.dirname(path.resolve(params.storePath)); + return await runExclusiveSessionStoreWrite(params.storePath, async () => { + const store = loadMutableSessionStoreForWriter(params.storePath); + const referencedTranscriptPaths = new Set(); + for (const entry of Object.values(store)) { + const transcriptPath = resolveLifecycleTranscriptPath({ entry, sessionsDir }); + if (transcriptPath) { + referencedTranscriptPaths.add(normalizePathForLifecycleComparison(transcriptPath)); + } + } + restoreUnchangedSessionStoreCache(params.storePath, store); + + let entries: fs.Dirent[]; + try { + entries = await fs.promises.readdir(sessionsDir, { withFileTypes: true }); + } catch { + return 0; + } + + const { archiveSessionTranscripts } = await loadSessionArchiveRuntime(); + let archived = 0; + // Only archive primary transcripts that are no longer referenced by the + // current store and still carry the lifecycle marker supplied by the caller. + for (const entry of entries) { + if (!entry.isFile() || !entry.name.endsWith(".jsonl")) { + continue; + } + const transcriptPath = path.join(sessionsDir, entry.name); + if (referencedTranscriptPaths.has(normalizePathForLifecycleComparison(transcriptPath))) { + continue; + } + let stat: fs.Stats; + try { + stat = await fs.promises.stat(transcriptPath); + } catch { + continue; + } + if (params.nowMs - stat.mtimeMs < params.orphanTranscriptMinAgeMs) { + continue; + } + let content: string; + try { + content = await fs.promises.readFile(transcriptPath, "utf-8"); + } catch { + continue; + } + if (!content.includes(params.transcriptContentMarker)) { + continue; + } + const sessionId = entry.name.slice(0, -".jsonl".length); + archived += archiveSessionTranscripts({ + sessionId, + storePath: params.storePath, + sessionFile: transcriptPath, + reason: "deleted", + restrictToStoreDir: true, + }).length; + } + return archived; + }); +} + +/** Cleans scoped session lifecycle entries and their unreferenced transcript artifacts. */ +export async function cleanupSessionLifecycleArtifacts( + params: SessionLifecycleArtifactCleanupParams, +): Promise { + const sessionKeySegmentPrefix = params.sessionKeySegmentPrefix.trim(); + const transcriptContentMarker = params.transcriptContentMarker; + if (!sessionKeySegmentPrefix || !transcriptContentMarker) { + return { removedEntries: 0, archivedTranscriptArtifacts: 0 }; + } + + const nowMs = params.nowMs ?? Date.now(); + const storePath = path.resolve(params.storePath); + const sessionsDir = path.dirname(storePath); + const removedSessionFiles = new Map(); + const removedTranscriptPaths: Array<{ sessionId: string; transcriptPath: string }> = []; + let removedEntries = 0; + let archivedTranscriptArtifacts = 0; + + await runExclusiveSessionStoreWrite(storePath, async () => { + const store = loadMutableSessionStoreForWriter(storePath); + // Delete only rows owned by the named lifecycle. Orphan transcript cleanup + // reacquires this writer lock later so its reference set cannot go stale. + for (const [sessionKey, entry] of Object.entries(store)) { + const transcriptPath = resolveLifecycleTranscriptPath({ entry, sessionsDir }); + const matchesLifecycle = sessionKeySegmentStartsWith(sessionKey, sessionKeySegmentPrefix); + if ( + matchesLifecycle && + lifecycleTranscriptIsReclaimable({ + transcriptPath, + nowMs, + orphanTranscriptMinAgeMs: params.orphanTranscriptMinAgeMs, + }) + ) { + rememberRemovedSessionFile(removedSessionFiles, entry); + if (entry.sessionId && transcriptPath && fs.existsSync(transcriptPath)) { + removedTranscriptPaths.push({ sessionId: entry.sessionId, transcriptPath }); + } + delete store[sessionKey]; + removedEntries += 1; + continue; + } + } + + if (removedEntries === 0) { + restoreUnchangedSessionStoreCache(storePath, store); + return; + } + + const referencedSessionIds = new Set( + Object.values(store) + .map((entry) => entry?.sessionId) + .filter((sessionId): sessionId is string => Boolean(sessionId)), + ); + // Archive only the exact transcript path that passed the age/missing guard. + // Broader session-id candidate scans can include fresh sibling transcripts. + for (const { sessionId: removedSessionId, transcriptPath } of removedTranscriptPaths) { + if (referencedSessionIds.has(removedSessionId)) { + continue; + } + archivedTranscriptArtifacts += archiveExactLifecycleTranscriptPath({ + sessionsDir, + transcriptPath, + }); + } + const { removeRemovedSessionTrajectoryArtifacts } = await loadTrajectoryCleanupRuntime(); + await removeRemovedSessionTrajectoryArtifacts({ + removedSessionFiles, + referencedSessionIds, + storePath, + restrictToStoreDir: true, + }); + await saveSessionStoreUnlocked(storePath, store, { skipMaintenance: true }); + }); + + return { + removedEntries, + archivedTranscriptArtifacts: + archivedTranscriptArtifacts + + (await archiveUnreferencedLifecycleTranscriptArtifacts({ + storePath, + transcriptContentMarker, + orphanTranscriptMinAgeMs: params.orphanTranscriptMinAgeMs, + nowMs, + })), + }; +} + export async function runQuotaSuspensionMaintenance(params: { storePath: string; now?: number; @@ -1033,6 +1281,7 @@ export async function patchSessionEntry( replaceEntry?: boolean; update: ( entry: SessionEntry, + context: { existingEntry?: SessionEntry }, ) => Promise | null> | Partial | null; }, ): Promise { @@ -1044,7 +1293,9 @@ export async function patchSessionEntry( if (!existing) { return null; } - const patch = await params.update(cloneSessionEntry(existing)); + const patch = await params.update(cloneSessionEntry(existing), { + existingEntry: resolved.existing ? cloneSessionEntry(resolved.existing) : undefined, + }); if (!patch) { return existing; } diff --git a/src/config/sessions/transcript-append.ts b/src/config/sessions/transcript-append.ts index 119349f2326a..8fce9f2f4fce 100644 --- a/src/config/sessions/transcript-append.ts +++ b/src/config/sessions/transcript-append.ts @@ -15,6 +15,7 @@ import { redactSecrets } from "../../logging/redact.js"; import { createSessionTranscriptHeader } from "./transcript-header.js"; import { appendJsonlEntry, + serializeJsonlEntry, serializeJsonlLine, writeJsonlEntry, writeJsonlLines, @@ -289,6 +290,31 @@ export async function appendSessionTranscriptMessage( ); } +export type AppendSessionTranscriptEventParams = { + config?: OpenClawConfig; + event: unknown; + transcriptPath: string; +}; + +/** Appends a raw transcript event using the same write lock and FIFO as message appends. */ +export async function appendSessionTranscriptEvent( + params: AppendSessionTranscriptEventParams, +): Promise { + const activeLockRunner = resolveOwnedSessionTranscriptWriteLockRunner({ + sessionFile: params.transcriptPath, + }); + if (activeLockRunner) { + return await activeLockRunner(() => + withTranscriptAppendQueue(params.transcriptPath, () => + appendSessionTranscriptEventLocked(params), + ), + ); + } + return await withTranscriptAppendQueue(params.transcriptPath, () => + withSessionTranscriptWriteLock(params, () => appendSessionTranscriptEventLocked(params)), + ); +} + async function withSessionTranscriptWriteLock( params: Pick, run: () => Promise | T, @@ -305,6 +331,18 @@ async function withSessionTranscriptWriteLock( } } +async function appendSessionTranscriptEventLocked( + params: AppendSessionTranscriptEventParams, +): Promise { + await fs.mkdir(path.dirname(params.transcriptPath), { recursive: true }); + const handle = await fs.open(params.transcriptPath, "a", 0o600); + try { + await handle.appendFile(serializeJsonlEntry(params.event), "utf-8"); + } finally { + await handle.close(); + } +} + async function appendSessionTranscriptMessageLocked( params: AppendSessionTranscriptMessageParams, ): Promise | undefined> { diff --git a/src/gateway/server-methods/sessions.ts b/src/gateway/server-methods/sessions.ts index 977c3cab1d3a..4ab786fbd66e 100644 --- a/src/gateway/server-methods/sessions.ts +++ b/src/gateway/server-methods/sessions.ts @@ -49,7 +49,6 @@ import { compactEmbeddedAgentSession } from "../../agents/embedded-agent.js"; import { clearSessionQueues } from "../../auto-reply/reply/queue/cleanup.js"; import { normalizeReasoningLevel, normalizeThinkLevel } from "../../auto-reply/thinking.js"; import { - loadSessionStore, runSessionsCleanup, serializeSessionCleanupResult, resolveMainSessionKey, @@ -261,6 +260,22 @@ function resolveGatewaySessionTargetFromKey( return { cfg, target, storePath: target.storePath }; } +function loadSessionEntriesForTarget(params: { + key: string; + cfg: OpenClawConfig; + agentId?: string; +}) { + const target = resolveGatewaySessionStoreTargetWithStore({ + cfg: params.cfg, + key: params.key, + clone: false, + ...(params.agentId ? { agentId: params.agentId } : {}), + }); + const store = target.store; + const entry = resolveFreshestSessionEntryFromStoreKeys(store, target.storeKeys); + return { target, storePath: target.storePath, store, entry }; +} + function resolveOptionalInitialSessionMessage(params: { task?: unknown; message?: unknown; @@ -1215,9 +1230,7 @@ export const sessionsHandlers: GatewayRequestHandlers = { return; } const cfg = context.getRuntimeConfig(); - const { target, storePath } = resolveGatewaySessionTargetFromKey(key, cfg); - const store = loadSessionStore(storePath); - const entry = resolveFreshestSessionEntryFromStoreKeys(store, target.storeKeys); + const { target, storePath, store, entry } = loadSessionEntriesForTarget({ key, cfg }); if (!entry) { respond(true, { session: null }, undefined); return; @@ -2420,11 +2433,11 @@ export const sessionsHandlers: GatewayRequestHandlers = { respond(false, undefined, requestedAgent.error); return; } - const { target, storePath } = resolveGatewaySessionTargetFromKey(key, cfg, { + const { storePath, entry } = loadSessionEntriesForTarget({ + key, + cfg, agentId: requestedAgent.agentId, }); - const store = loadSessionStore(storePath); - const entry = resolveFreshestSessionEntryFromStoreKeys(store, target.storeKeys); if (!entry?.sessionId) { respond(true, { messages: [] }, undefined); return; diff --git a/src/gateway/session-utils.ts b/src/gateway/session-utils.ts index 7aa8e54d8281..0517744e6f84 100644 --- a/src/gateway/session-utils.ts +++ b/src/gateway/session-utils.ts @@ -61,7 +61,6 @@ import { resolveStateDir } from "../config/paths.js"; import { buildGroupDisplayName, getSessionStoreCacheVersion, - loadSessionStore, resolveAllAgentSessionStoreTargetsSync, resolveAgentMainSessionKey, resolveFreshSessionTotalTokens, @@ -71,6 +70,7 @@ import { type SessionStoreTarget, type SessionScope, } from "../config/sessions.js"; +import { listSessionEntries as listAccessorSessionEntries } from "../config/sessions/session-accessor.js"; import type { OpenClawConfig } from "../config/types.openclaw.js"; import { openRootFileSync } from "../infra/boundary-file-read.js"; import { projectPluginSessionExtensionsSync } from "../plugins/host-hook-state.js"; @@ -1412,6 +1412,18 @@ function resolveGatewaySessionStoreCandidates( return [...targets.values()]; } +function loadGatewaySessionLookupStore( + storePath: string, + clone: boolean | undefined, +): Record { + return Object.fromEntries( + listAccessorSessionEntries({ + ...(clone === false ? { clone: false } : {}), + storePath, + }).map(({ sessionKey, entry }) => [sessionKey, entry]), + ); +} + function resolveGatewaySessionStoreLookup(params: { cfg: OpenClawConfig; key: string; @@ -1430,9 +1442,9 @@ function resolveGatewaySessionStoreLookup(params: { agentId: params.agentId, storePath: resolveStorePath(params.cfg.session?.store, { agentId: params.agentId }), }; - const loadOptions = params.clone === false ? { clone: false } : undefined; + const loadStore = (storePath: string) => loadGatewaySessionLookupStore(storePath, params.clone); let selectedStorePath = fallback.storePath; - let selectedStore = params.initialStore ?? loadSessionStore(fallback.storePath, loadOptions); + let selectedStore = params.initialStore ?? loadStore(fallback.storePath); let selectedMatch = findFreshestStoreMatch(selectedStore, ...scanTargets); let selectedUpdatedAt = selectedMatch?.entry.updatedAt ?? Number.NEGATIVE_INFINITY; @@ -1441,7 +1453,7 @@ function resolveGatewaySessionStoreLookup(params: { if (!candidate) { continue; } - const store = loadSessionStore(candidate.storePath, loadOptions); + const store = loadStore(candidate.storePath); const match = findFreshestStoreMatch(store, ...scanTargets); if (!match) { continue; @@ -1499,12 +1511,11 @@ function resolveExplicitDeletedLegacyMainStoreTarget(params: { match: { entry: SessionEntry; key: string }; } | undefined; - const loadOptions = params.clone === false ? { clone: false } : undefined; for (const target of resolveAllAgentSessionStoreTargetsSync(params.cfg)) { if (target.agentId !== legacyAgentId) { continue; } - const store = loadSessionStore(target.storePath, loadOptions); + const store = loadGatewaySessionLookupStore(target.storePath, params.clone); const match = findFreshestStoreMatch(store, ...lookupSeeds); if (!match) { continue; diff --git a/src/gateway/sessions-resolve.test.ts b/src/gateway/sessions-resolve.test.ts index ec9d29108cc3..050965b9bf1c 100644 --- a/src/gateway/sessions-resolve.test.ts +++ b/src/gateway/sessions-resolve.test.ts @@ -5,11 +5,10 @@ import { ErrorCodes } from "../../packages/gateway-protocol/src/index.js"; import type { SessionEntry } from "../config/sessions/types.js"; const hoisted = vi.hoisted(() => ({ - loadSessionStoreMock: vi.fn(), updateSessionStoreMock: vi.fn(), listSessionsFromStoreMock: vi.fn(), migrateAndPruneGatewaySessionStoreKeyMock: vi.fn(), - resolveGatewaySessionStoreTargetMock: vi.fn(), + resolveGatewaySessionStoreTargetWithStoreMock: vi.fn(), loadCombinedSessionStoreForGatewayMock: vi.fn(), listAgentIdsMock: vi.fn(), })); @@ -29,7 +28,6 @@ vi.mock("../config/sessions.js", async () => { await vi.importActual("../config/sessions.js"); return { ...actual, - loadSessionStore: hoisted.loadSessionStoreMock, updateSessionStore: hoisted.updateSessionStoreMock, }; }); @@ -40,7 +38,8 @@ vi.mock("./session-utils.js", async () => { ...actual, listSessionsFromStore: hoisted.listSessionsFromStoreMock, migrateAndPruneGatewaySessionStoreKey: hoisted.migrateAndPruneGatewaySessionStoreKeyMock, - resolveGatewaySessionStoreTarget: hoisted.resolveGatewaySessionStoreTargetMock, + resolveGatewaySessionStoreTargetWithStore: + hoisted.resolveGatewaySessionStoreTargetWithStoreMock, loadCombinedSessionStoreForGateway: hoisted.loadCombinedSessionStoreForGatewayMock, }; }); @@ -51,6 +50,7 @@ describe("resolveSessionKeyFromResolveParams", () => { const canonicalKey = "agent:main:canon"; const legacyKey = "agent:main:legacy"; const storePath = "/tmp/sessions.json"; + let targetStore: Record; const expectResolveToCanonicalKey = async ( p: Parameters[0]["p"], @@ -68,37 +68,33 @@ describe("resolveSessionKeyFromResolveParams", () => { }; beforeEach(() => { - hoisted.loadSessionStoreMock.mockReset(); hoisted.updateSessionStoreMock.mockReset(); hoisted.listSessionsFromStoreMock.mockReset(); hoisted.migrateAndPruneGatewaySessionStoreKeyMock.mockReset(); - hoisted.resolveGatewaySessionStoreTargetMock.mockReset(); + hoisted.resolveGatewaySessionStoreTargetWithStoreMock.mockReset(); hoisted.loadCombinedSessionStoreForGatewayMock.mockReset(); hoisted.listAgentIdsMock.mockReset(); + targetStore = {}; // Default: all agents are known (main is always present). hoisted.listAgentIdsMock.mockReturnValue(["main"]); - hoisted.resolveGatewaySessionStoreTargetMock.mockReturnValue({ + hoisted.resolveGatewaySessionStoreTargetWithStoreMock.mockImplementation(() => ({ canonicalKey, storeKeys: [canonicalKey, legacyKey], storePath, - }); + store: targetStore, + })); hoisted.migrateAndPruneGatewaySessionStoreKeyMock.mockReturnValue({ primaryKey: canonicalKey }); hoisted.updateSessionStoreMock.mockImplementation( async (_path: string, updater: (store: Record) => void) => { - const store = hoisted.loadSessionStoreMock.mock.results[0]?.value as - | Record - | undefined; - if (store) { - updater(store); - } + updater(targetStore); }, ); }); it("hides canonical keys that fail the spawnedBy visibility filter", async () => { - hoisted.loadSessionStoreMock.mockReturnValue({ + targetStore = { [canonicalKey]: { sessionId: "sess-1", updatedAt: 1 }, - }); + }; hoisted.listSessionsFromStoreMock.mockReturnValue({ sessions: [] }); await expect( @@ -131,7 +127,7 @@ describe("resolveSessionKeyFromResolveParams", () => { updatedAt: now - i, }; } - hoisted.loadSessionStoreMock.mockReturnValue(store); + targetStore = store; await expectResolveToCanonicalKey({ key: canonicalKey, spawnedBy: "controller-1" }); }); @@ -140,7 +136,7 @@ describe("resolveSessionKeyFromResolveParams", () => { const store = { [legacyKey]: { sessionId: "sess-legacy", spawnedBy: "controller-1", updatedAt: Date.now() }, } satisfies Record; - hoisted.loadSessionStoreMock.mockImplementation(() => store); + targetStore = store; await expectResolveToCanonicalKey({ key: canonicalKey, spawnedBy: "controller-1" }); @@ -152,13 +148,14 @@ describe("resolveSessionKeyFromResolveParams", () => { it("rejects sessions belonging to a deleted agent (key-based lookup)", async () => { const deletedAgentKey = "agent:deleted-agent:main"; - hoisted.resolveGatewaySessionStoreTargetMock.mockReturnValue({ + targetStore = { + [deletedAgentKey]: { sessionId: "sess-orphan", updatedAt: 1 }, + }; + hoisted.resolveGatewaySessionStoreTargetWithStoreMock.mockReturnValue({ canonicalKey: deletedAgentKey, storeKeys: [deletedAgentKey], storePath, - }); - hoisted.loadSessionStoreMock.mockReturnValue({ - [deletedAgentKey]: { sessionId: "sess-orphan", updatedAt: 1 }, + store: targetStore, }); // "deleted-agent" is not in the known agents list. hoisted.listAgentIdsMock.mockReturnValue(["main"]); @@ -179,12 +176,7 @@ describe("resolveSessionKeyFromResolveParams", () => { it("resolves ACP harness session keys even when harness id is not in agents.list", async () => { const acpKey = "agent:claude:acp:11111111-1111-4111-8111-111111111111"; - hoisted.resolveGatewaySessionStoreTargetMock.mockReturnValue({ - canonicalKey: acpKey, - storeKeys: [acpKey], - storePath, - }); - hoisted.loadSessionStoreMock.mockReturnValue({ + targetStore = { [acpKey]: { sessionId: "sess-acp", updatedAt: 1, @@ -198,6 +190,12 @@ describe("resolveSessionKeyFromResolveParams", () => { lastActivityAt: 1, }, }, + }; + hoisted.resolveGatewaySessionStoreTargetWithStoreMock.mockReturnValue({ + canonicalKey: acpKey, + storeKeys: [acpKey], + storePath, + store: targetStore, }); hoisted.listAgentIdsMock.mockReturnValue(["main"]); @@ -214,13 +212,14 @@ describe("resolveSessionKeyFromResolveParams", () => { it("rejects non-alias agent:main sessions when main is no longer configured", async () => { const staleMainKey = "agent:main:guildchat:direct:u1"; - hoisted.resolveGatewaySessionStoreTargetMock.mockReturnValue({ + targetStore = { + [staleMainKey]: { sessionId: "sess-stale-main", updatedAt: 1 }, + }; + hoisted.resolveGatewaySessionStoreTargetWithStoreMock.mockReturnValue({ canonicalKey: staleMainKey, storeKeys: [staleMainKey], storePath, - }); - hoisted.loadSessionStoreMock.mockReturnValue({ - [staleMainKey]: { sessionId: "sess-stale-main", updatedAt: 1 }, + store: targetStore, }); hoisted.listAgentIdsMock.mockReturnValue(["ops"]); diff --git a/src/gateway/sessions-resolve.ts b/src/gateway/sessions-resolve.ts index f7e8741501e3..c2d2150fc49d 100644 --- a/src/gateway/sessions-resolve.ts +++ b/src/gateway/sessions-resolve.ts @@ -7,7 +7,7 @@ import { errorShape, type SessionsResolveParams, } from "../../packages/gateway-protocol/src/index.js"; -import { loadSessionStore, updateSessionStore, type SessionEntry } from "../config/sessions.js"; +import { updateSessionStore, type SessionEntry } from "../config/sessions.js"; import type { OpenClawConfig } from "../config/types.openclaw.js"; import { resolveSessionIdMatchSelection } from "../sessions/session-id-resolution.js"; import { parseSessionLabel } from "../sessions/session-label.js"; @@ -17,7 +17,7 @@ import { loadCombinedSessionStoreForGateway, migrateAndPruneGatewaySessionStoreKey, resolveDeletedAgentIdFromSessionKey, - resolveGatewaySessionStoreTarget, + resolveGatewaySessionStoreTargetWithStore, } from "./session-utils.js"; export type SessionsResolveResult = { ok: true; key: string } | { ok: false; error: ErrorShape }; @@ -61,8 +61,7 @@ function validateSessionAgentExists( function isResolvedSessionKeyVisible(params: { cfg: OpenClawConfig; p: SessionsResolveParams; - storePath: string; - store: ReturnType; + store: Record; key: string; }) { if (typeof params.p.spawnedBy !== "string" || params.p.spawnedBy.trim().length === 0) { @@ -125,14 +124,13 @@ export async function resolveSessionKeyFromResolveParams(params: { if (hasKey) { // Key lookups may hit legacy store aliases. Migrate/prune before returning // the canonical key so later calls operate on one store identity. - const target = resolveGatewaySessionStoreTarget({ cfg, key }); - const store = loadSessionStore(target.storePath); + const target = resolveGatewaySessionStoreTargetWithStore({ cfg, key, clone: false }); + const store = target.store; if (store[target.canonicalKey]) { if ( !isResolvedSessionKeyVisible({ cfg, p, - storePath: target.storePath, store, key: target.canonicalKey, }) @@ -160,28 +158,31 @@ export async function resolveSessionKeyFromResolveParams(params: { s[primaryKey] = s[legacyKey]; } }); - const migratedStore = loadSessionStore(target.storePath); + const refreshedTarget = resolveGatewaySessionStoreTargetWithStore({ + cfg, + key: target.canonicalKey, + clone: false, + }); if ( !isResolvedSessionKeyVisible({ cfg, p, - storePath: target.storePath, - store: migratedStore, - key: target.canonicalKey, + store: refreshedTarget.store, + key: refreshedTarget.canonicalKey, }) ) { return noSessionFoundResult(key); } const agentCheckLegacy = validateSessionAgentExists( cfg, - target.canonicalKey, - migratedStore[target.canonicalKey], - { acpMetadataSessionKey: target.canonicalKey }, + refreshedTarget.canonicalKey, + refreshedTarget.store[refreshedTarget.canonicalKey], + { acpMetadataSessionKey: refreshedTarget.canonicalKey }, ); if (agentCheckLegacy) { return agentCheckLegacy; } - return { ok: true, key: target.canonicalKey }; + return { ok: true, key: refreshedTarget.canonicalKey }; } if (hasSessionId) { diff --git a/test/scripts/check-session-accessor-boundary.test.ts b/test/scripts/check-session-accessor-boundary.test.ts new file mode 100644 index 000000000000..9a4b8d36a771 --- /dev/null +++ b/test/scripts/check-session-accessor-boundary.test.ts @@ -0,0 +1,75 @@ +import { describe, expect, it } from "vitest"; +import { + findSessionAccessorBoundaryViolations, + migratedSessionAccessorFiles, +} from "../../scripts/check-session-accessor-boundary.mjs"; + +describe("session accessor boundary guard", () => { + it("ratchets only the files migrated by the session accessor gateway slice", () => { + expect(migratedSessionAccessorFiles).toEqual( + new Set([ + "src/config/sessions/combined-store-gateway.ts", + "src/gateway/session-utils.ts", + "src/gateway/sessions-resolve.ts", + "src/gateway/server-methods/sessions.ts", + ]), + ); + }); + + it("flags legacy reader imports", () => { + expect( + findSessionAccessorBoundaryViolations(` + import { loadSessionStore, readSessionEntries as readEntries } from "../config/sessions.js"; + `), + ).toEqual([ + { line: 2, reason: 'imports legacy session store reader "loadSessionStore"' }, + { line: 2, reason: 'imports legacy session store reader "readSessionEntries"' }, + ]); + }); + + it("flags direct and namespace legacy reader calls", () => { + expect( + findSessionAccessorBoundaryViolations(` + loadSessionStore(storePath); + sessions.readSessionEntries(storePath); + sessions["loadSessionStore"](storePath); + `), + ).toEqual([ + { line: 2, reason: 'calls legacy session store reader "loadSessionStore"' }, + { line: 3, reason: 'references legacy session store reader "readSessionEntries"' }, + { line: 4, reason: 'references legacy session store reader "loadSessionStore"' }, + ]); + }); + + it("flags aliased namespace reader references", () => { + expect( + findSessionAccessorBoundaryViolations(` + const load = sessions.loadSessionStore; + const { readSessionEntries: readEntries } = sessions; + const { loadSessionStore } = sessions; + `), + ).toEqual([ + { line: 2, reason: 'references legacy session store reader "loadSessionStore"' }, + { line: 3, reason: 'aliases legacy session store reader "readSessionEntries"' }, + { line: 4, reason: 'aliases legacy session store reader "loadSessionStore"' }, + ]); + }); + + it("allows migrated accessor reads", () => { + expect( + findSessionAccessorBoundaryViolations(` + import { listSessionEntries } from "../config/sessions/session-accessor.js"; + listSessionEntries({ storePath }); + `), + ).toEqual([]); + }); + + it("ignores comments and strings that describe legacy readers", () => { + expect( + findSessionAccessorBoundaryViolations(` + // loadSessionStore and readSessionEntries used to be called here. + const description = "loadSessionStore"; + `), + ).toEqual([]); + }); +}); diff --git a/test/scripts/ci-workflow-guards.test.ts b/test/scripts/ci-workflow-guards.test.ts index 9ee9270a3274..a641ae31863c 100644 --- a/test/scripts/ci-workflow-guards.test.ts +++ b/test/scripts/ci-workflow-guards.test.ts @@ -16,6 +16,22 @@ function readCriticalQualityWorkflow() { } describe("ci workflow guards", () => { + it("runs the session accessor ratchet as a visible additional check", () => { + const workflow = readCiWorkflow(); + const additionalJob = workflow.jobs["check-additional-shard"]; + const matrixRows = additionalJob.strategy.matrix.include; + expect(matrixRows).toContainEqual({ + check_name: "check-session-accessor-boundary", + group: "session-accessor-boundary", + }); + + const runStep = additionalJob.steps.find((step) => step.name === "Run additional check shard"); + expect(runStep.run).toContain("session-accessor-boundary)"); + expect(runStep.run).toContain( + 'run_check "lint:tmp:session-accessor-boundary" pnpm run lint:tmp:session-accessor-boundary', + ); + }); + it("kills timed manual checkout fetches after the grace period", () => { const workflowPaths = [ ".github/workflows/ci.yml", diff --git a/test/scripts/run-additional-boundary-checks.test.ts b/test/scripts/run-additional-boundary-checks.test.ts index 3ba83369183e..9aa3abf0b89a 100644 --- a/test/scripts/run-additional-boundary-checks.test.ts +++ b/test/scripts/run-additional-boundary-checks.test.ts @@ -94,7 +94,7 @@ describe("run-additional-boundary-checks", () => { }); it("keeps the raw HTTP/2 import guard in source boundary checks", () => { - expect(BOUNDARY_CHECKS[6]).toEqual({ + expect(BOUNDARY_CHECKS).toContainEqual({ label: "lint:tmp:no-raw-http2-imports", command: "pnpm", args: ["run", "lint:tmp:no-raw-http2-imports"],