diff --git a/extensions/memory-core/src/dreaming-narrative.test.ts b/extensions/memory-core/src/dreaming-narrative.test.ts index f64ffbd559b1..907a9983d30e 100644 --- a/extensions/memory-core/src/dreaming-narrative.test.ts +++ b/extensions/memory-core/src/dreaming-narrative.test.ts @@ -962,7 +962,7 @@ describe("runDreamNarrative", () => { "agent:main:dreaming-narrative-light-1": { sessionId: "orphan", sessionFile: orphanPath, - updatedAt, + updatedAt: updatedAt - 600_000, }, "agent:main:kept-session": { sessionId: "still-live", @@ -1020,7 +1020,7 @@ describe("runDreamNarrative", () => { const updatedStore = readSessionStoreEntries(storePath) as Record; expect(updatedStore).not.toHaveProperty("agent:main:dreaming-narrative-light-1"); - expect(updatedStore).not.toHaveProperty("agent:main:dreaming-narrative-corrupt-normal"); + expect(updatedStore).toHaveProperty("agent:main:dreaming-narrative-corrupt-normal"); expect(updatedStore).toHaveProperty("agent:main:kept-session"); expect(updatedStore).toHaveProperty("agent:main:telegram:group:dreaming-narrative-room"); expect(loadTranscriptEventsSync({ agentId: "main", sessionId: "orphan", storePath })).toEqual( @@ -1052,7 +1052,7 @@ describe("runDreamNarrative", () => { "agent:main:dreaming-narrative-deep-orphan": { sessionId: "orphan-dreaming", sessionFile: orphanTranscript, - updatedAt, + updatedAt: updatedAt - 600_000, }, "agent:main:dreaming-narrative-deep-live": { sessionId: "live-dreaming", diff --git a/extensions/memory-core/src/dreaming-narrative.ts b/extensions/memory-core/src/dreaming-narrative.ts index a4ebe36918d0..165aa9008954 100644 --- a/extensions/memory-core/src/dreaming-narrative.ts +++ b/extensions/memory-core/src/dreaming-narrative.ts @@ -1,8 +1,5 @@ // Memory Core plugin module implements dreaming narrative behavior. import { createHash } from "node:crypto"; -import type { Dirent } from "node:fs"; -import fs from "node:fs/promises"; -import path from "node:path"; import { createAsyncLock } from "openclaw/plugin-sdk/async-lock-runtime"; import { extractErrorCode, @@ -12,12 +9,14 @@ import { SUBAGENT_RUNTIME_REQUEST_SCOPE_ERROR_CODE, } from "openclaw/plugin-sdk/error-runtime"; import { resolveGlobalMap } from "openclaw/plugin-sdk/global-singleton"; -import { resolveStateDir } from "openclaw/plugin-sdk/memory-core-host-runtime-core"; import { getRuntimeConfig } from "openclaw/plugin-sdk/runtime-config-snapshot"; -import { cleanupSessionLifecycleArtifacts } from "openclaw/plugin-sdk/session-store-runtime"; import { truncateUtf16Safe } from "openclaw/plugin-sdk/text-utility-runtime"; import pLimit from "p-limit"; import { readDreamsFile, resolveDreamsPath, updateDreamsFile } from "./dreaming-dreams-file.js"; +import { + DREAMING_SESSION_KEY_PREFIX, + scrubDreamingNarrativeArtifacts, +} from "./dreaming-session-cleanup.js"; // ── Types ────────────────────────────────────────────────────────────── @@ -102,10 +101,7 @@ const NARRATIVE_MESSAGE_FETCH_LIMIT = 5; // A completed run can reach the session reader before the final assistant text // is visible, so retry briefly before falling back to synthetic diary text. const NARRATIVE_MESSAGE_SETTLE_DELAYS_MS = [50, 150, 300, 750] as const; -const DREAMING_SESSION_KEY_PREFIX = "dreaming-narrative-"; const DREAMING_SESSION_OWNER_KEY = "memory-core-v2"; -const DREAMING_TRANSCRIPT_RUN_MARKER = '"runId":"dreaming-narrative-'; -const DREAMING_ORPHAN_MIN_AGE_MS = 300_000; const DIARY_START_MARKER = ""; const DIARY_END_MARKER = ""; const BACKFILL_ENTRY_MARKER = "openclaw:dreaming:backfill-entry"; @@ -814,47 +810,6 @@ async function appendNarrativeEntry(params: { // ── Orchestrator ─────────────────────────────────────────────────────── -async function scrubDreamingNarrativeArtifacts(logger: Logger): Promise { - const cfg = getRuntimeConfig(); - const agentsDir = path.join(resolveStateDir(), "agents"); - let agentEntries: Dirent[]; - try { - agentEntries = await fs.readdir(agentsDir, { withFileTypes: true }); - } catch { - return; - } - - let prunedEntries = 0; - let archivedOrphans = 0; - - for (const agentEntry of agentEntries) { - if (!agentEntry.isDirectory()) { - continue; - } - - try { - const result = await cleanupSessionLifecycleArtifacts({ - agentId: agentEntry.name, - archiveRemovedEntryTranscripts: false, - sessionStore: cfg.session?.store, - sessionKeySegmentPrefix: DREAMING_SESSION_KEY_PREFIX, - transcriptContentMarker: DREAMING_TRANSCRIPT_RUN_MARKER, - orphanTranscriptMinAgeMs: DREAMING_ORPHAN_MIN_AGE_MS, - }); - prunedEntries += result.removedEntries; - archivedOrphans += result.archivedTranscriptArtifacts; - } catch { - continue; - } - } - - if (prunedEntries > 0 || archivedOrphans > 0) { - logger.info( - `memory-core: dreaming cleanup scrubbed ${prunedEntries} stale session entr${prunedEntries === 1 ? "y" : "ies"} and archived ${archivedOrphans} orphan transcript${archivedOrphans === 1 ? "" : "s"}.`, - ); - } -} - export type DreamNarrativeRequest = { /** Agent that owns this workspace; the narrative session lives in its SQLite store. */ agentId: string; @@ -1050,7 +1005,11 @@ async function generateAndAppendDreamNarrative( } } - await scrubDreamingNarrativeArtifacts(params.logger).catch((scrubErr: unknown) => { + await scrubDreamingNarrativeArtifacts({ + agentId: params.agentId, + config: getRuntimeConfig(), + logger: params.logger, + }).catch((scrubErr: unknown) => { cleanupFailure = formatErrorMessage(scrubErr); params.logger.warn( `memory-core: dreaming cleanup scrub failed for ${params.data.phase} phase: ${cleanupFailure}`, diff --git a/extensions/memory-core/src/dreaming-session-cleanup.ts b/extensions/memory-core/src/dreaming-session-cleanup.ts new file mode 100644 index 000000000000..602ae43e8bf7 --- /dev/null +++ b/extensions/memory-core/src/dreaming-session-cleanup.ts @@ -0,0 +1,31 @@ +import type { OpenClawConfig } from "openclaw/plugin-sdk/config-contracts"; +import { cleanupSessionLifecycleArtifacts } from "openclaw/plugin-sdk/session-store-runtime"; + +export const DREAMING_SESSION_KEY_PREFIX = "dreaming-narrative-"; +export const DREAMING_ORPHAN_MIN_AGE_MS = 300_000; +const DREAMING_TRANSCRIPT_RUN_MARKER = '"runId":"dreaming-narrative-'; + +export async function scrubDreamingNarrativeArtifacts(params: { + agentId: string; + config: OpenClawConfig; + logger: { info: (message: string) => void }; + nowMs?: number; +}): Promise { + const result = await cleanupSessionLifecycleArtifacts({ + agentId: params.agentId, + archiveRemovedEntryTranscripts: false, + orphanTranscriptMinAgeMs: DREAMING_ORPHAN_MIN_AGE_MS, + pluginOwnerId: "memory-core", + sessionStore: params.config.session?.store, + sessionKeySegmentPrefix: DREAMING_SESSION_KEY_PREFIX, + transcriptContentMarker: DREAMING_TRANSCRIPT_RUN_MARKER, + ...(params.nowMs === undefined ? {} : { nowMs: params.nowMs }), + }); + const prunedEntries = result.removedEntries; + const archivedOrphans = result.archivedTranscriptArtifacts; + if (prunedEntries > 0 || archivedOrphans > 0) { + params.logger.info( + `memory-core: dreaming cleanup scrubbed ${prunedEntries} stale session entr${prunedEntries === 1 ? "y" : "ies"} and archived ${archivedOrphans} orphan transcript${archivedOrphans === 1 ? "" : "s"}.`, + ); + } +} diff --git a/extensions/memory-core/src/dreaming-startup-cleanup.test.ts b/extensions/memory-core/src/dreaming-startup-cleanup.test.ts new file mode 100644 index 000000000000..f62a6bc613ff --- /dev/null +++ b/extensions/memory-core/src/dreaming-startup-cleanup.test.ts @@ -0,0 +1,381 @@ +import fs from "node:fs/promises"; +import os from "node:os"; +import path from "node:path"; +import type { OpenClawConfig } from "openclaw/plugin-sdk/config-contracts"; +import type { OpenClawPluginApi } from "openclaw/plugin-sdk/plugin-entry"; +import { getSessionEntry, upsertSessionEntry } from "openclaw/plugin-sdk/session-store-runtime"; +import { + appendSqliteSessionTranscriptEventForTest, + closeOpenClawAgentDatabasesForTest, +} from "openclaw/plugin-sdk/sqlite-runtime-testing"; +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; +import { registerShortTermPromotionDreaming } from "./dreaming.js"; + +const ORPHAN_AGE_MS = 300_000; + +type GatewayHook = (event: unknown, context: unknown) => Promise | void; + +let stateDir: string; +let stopGateway: (() => Promise) | undefined; + +beforeEach(async () => { + stateDir = await fs.mkdtemp(path.join(os.tmpdir(), "openclaw-dreaming-startup-")); + vi.stubEnv("OPENCLAW_STATE_DIR", stateDir); +}); + +afterEach(async () => { + await stopGateway?.(); + stopGateway = undefined; + closeOpenClawAgentDatabasesForTest(); + vi.useRealTimers(); + vi.unstubAllEnvs(); + vi.restoreAllMocks(); + await fs.rm(stateDir, { recursive: true, force: true }); +}); + +function createGateway( + params: { agentIds?: string[]; failCronReconciliation?: boolean; sessionStore?: string } = {}, +) { + const agentIds = params.agentIds ?? ["main"]; + const config = { + agents: { + list: agentIds.map((id, index) => ({ + id, + default: index === 0, + workspace: path.join(stateDir, `workspace-${id}`), + })), + }, + plugins: { + entries: { + "memory-core": { config: { dreaming: { enabled: true, frequency: "0 2 * * *" } } }, + }, + }, + ...(params.sessionStore ? { session: { store: params.sessionStore } } : {}), + } as OpenClawConfig; + const hooks = new Map(); + const logger = { debug: vi.fn(), info: vi.fn(), warn: vi.fn(), error: vi.fn() }; + const cron = { + list: vi.fn(async () => { + if (params.failCronReconciliation) { + throw new Error("cron startup failed"); + } + return []; + }), + add: vi.fn(async () => ({})), + update: vi.fn(async () => ({})), + remove: vi.fn(async () => ({ removed: false })), + removeStaleJobFamily: vi.fn(async () => 0), + }; + const api = { + config, + pluginConfig: {}, + logger, + runtime: { config: { current: () => config } }, + on: (eventName: string, hook: GatewayHook) => hooks.set(eventName, hook), + } as unknown as OpenClawPluginApi; + registerShortTermPromotionDreaming(api); + + const start = async () => { + const hook = hooks.get("gateway_start"); + if (!hook) { + throw new Error("gateway_start hook missing"); + } + await hook({ port: 0 }, { config, getCron: () => cron }); + }; + const stop = async () => { + await hooks.get("gateway_stop")?.({ reason: "test" }, {}); + }; + stopGateway = stop; + return { config, cron, logger, start, stop }; +} + +async function seedSession(params: { + agentId?: string; + suffix: string; + updatedAt: number; + transcriptAt?: number; + pluginOwnerId?: string; + storePath?: string; +}) { + const agentId = params.agentId ?? "main"; + const sessionKey = `agent:${agentId}:${params.suffix}`; + const sessionId = `${agentId}-${params.suffix}`; + await upsertSessionEntry({ + agentId, + sessionKey, + ...(params.storePath ? { storePath: params.storePath } : {}), + entry: { + sessionId, + updatedAt: params.updatedAt, + ...(params.pluginOwnerId ? { pluginOwnerId: params.pluginOwnerId } : {}), + }, + }); + if (params.transcriptAt !== undefined) { + await appendSqliteSessionTranscriptEventForTest({ + agentId, + sessionId, + sessionKey, + ...(params.storePath ? { storePath: params.storePath } : {}), + event: { + runId: `dreaming-narrative-${sessionId}`, + timestamp: params.transcriptAt, + type: "metadata", + }, + }); + } + return { agentId, sessionKey }; +} + +function hasSession(scope: { agentId: string; sessionKey: string }): boolean { + return getSessionEntry(scope) !== undefined; +} + +describe("dreaming gateway restart cleanup", () => { + it("does not create agent storage on a fresh gateway startup", async () => { + const gateway = createGateway({ agentIds: ["main", "researcher"] }); + + await gateway.start(); + + await expect(fs.access(path.join(stateDir, "agents"))).rejects.toMatchObject({ + code: "ENOENT", + }); + }); + + it("reclaims every stale dreaming phase without touching active, unrelated, or other-agent sessions", async () => { + const now = Date.now(); + const stale = await Promise.all( + ["light", "rem", "deep", "consolidation"].map((phase) => + seedSession({ + suffix: `dreaming-narrative-${phase}-interrupted`, + ...(phase === "rem" ? {} : { pluginOwnerId: "memory-core" }), + updatedAt: now - ORPHAN_AGE_MS - 1, + transcriptAt: now - ORPHAN_AGE_MS - 1, + }), + ), + ); + const activeWithTranscript = await seedSession({ + suffix: "dreaming-narrative-light-active", + updatedAt: now, + transcriptAt: now, + }); + const activeBeforeTranscript = await seedSession({ + suffix: "dreaming-narrative-rem-starting", + updatedAt: now, + }); + const userSession = await seedSession({ + suffix: "telegram:group:dreaming-narrative-room", + updatedAt: now - ORPHAN_AGE_MS - 1, + }); + const foreignPluginSession = await seedSession({ + suffix: "dreaming-narrative-foreign", + pluginOwnerId: "other-plugin", + updatedAt: now - ORPHAN_AGE_MS - 1, + transcriptAt: now - ORPHAN_AGE_MS - 1, + }); + const otherAgent = await seedSession({ + agentId: "researcher", + suffix: "dreaming-narrative-light-interrupted", + updatedAt: now - ORPHAN_AGE_MS - 1, + transcriptAt: now - ORPHAN_AGE_MS - 1, + }); + const gateway = createGateway(); + + await gateway.start(); + + expect(stale.map(hasSession)).toEqual([false, false, false, false]); + expect(hasSession(activeWithTranscript)).toBe(true); + expect(hasSession(activeBeforeTranscript)).toBe(true); + expect(hasSession(userSession)).toBe(true); + expect(hasSession(foreignPluginSession)).toBe(true); + expect(hasSession(otherAgent)).toBe(true); + }); + + it("rechecks fresh interrupted sessions once they age without deleting newly started work", async () => { + vi.useFakeTimers({ now: new Date("2026-08-01T12:00:00.000Z") }); + const interrupted = await seedSession({ + suffix: "dreaming-narrative-deep-interrupted", + updatedAt: Date.now() - 1, + transcriptAt: Date.now() - 1, + }); + const gateway = createGateway(); + + await gateway.start(); + expect(hasSession(interrupted)).toBe(true); + + await vi.advanceTimersByTimeAsync(ORPHAN_AGE_MS - 1); + const newlyStarted = await seedSession({ + suffix: "dreaming-narrative-light-just-started", + updatedAt: Date.now(), + }); + expect(hasSession(interrupted)).toBe(true); + + await vi.advanceTimersByTimeAsync(1); + + expect(hasSession(interrupted)).toBe(false); + expect(hasSession(newlyStarted)).toBe(true); + }); + + it("does not reclaim post-startup runs when the deferred cleanup callback runs late", async () => { + vi.useFakeTimers({ now: new Date("2026-08-01T12:00:00.000Z") }); + const interruptedAtStartup = await seedSession({ + suffix: "dreaming-narrative-deep-interrupted", + updatedAt: Date.now() - 1, + transcriptAt: Date.now() - 1, + }); + const gateway = createGateway(); + + await gateway.start(); + const postStartupUpdatedAt = Date.now(); + const activeWithTranscript = await seedSession({ + suffix: "dreaming-narrative-light-still-running", + updatedAt: postStartupUpdatedAt, + transcriptAt: postStartupUpdatedAt, + }); + const activeWithoutTranscript = await seedSession({ + suffix: "dreaming-narrative-rem-still-running", + updatedAt: postStartupUpdatedAt, + }); + + // Wall-clock stalls can outlive agent.wait; that wait never cancels the agent run. + vi.setSystemTime(Date.now() + ORPHAN_AGE_MS + 1); + await vi.advanceTimersByTimeAsync(ORPHAN_AGE_MS); + + expect(Date.now() - postStartupUpdatedAt).toBeGreaterThan(ORPHAN_AGE_MS); + expect(hasSession(interruptedAtStartup)).toBe(false); + expect(hasSession(activeWithTranscript)).toBe(true); + expect(hasSession(activeWithoutTranscript)).toBe(true); + }); + + it("does not reclaim post-startup runs when cron startup reconciliation stalls", async () => { + vi.useFakeTimers({ now: new Date("2026-08-01T12:00:00.000Z") }); + const staleAtStartup = await seedSession({ + suffix: "dreaming-narrative-deep-interrupted", + updatedAt: Date.now() - ORPHAN_AGE_MS - 1, + }); + const postStartupSessions: Array<{ agentId: string; sessionKey: string }> = []; + let postStartupUpdatedAt = 0; + const gateway = createGateway(); + gateway.cron.list.mockImplementationOnce(async () => { + await vi.advanceTimersByTimeAsync(1); + postStartupUpdatedAt = Date.now(); + postStartupSessions.push( + await seedSession({ + suffix: "dreaming-narrative-light-reconciliation-active", + updatedAt: postStartupUpdatedAt, + transcriptAt: postStartupUpdatedAt, + }), + await seedSession({ + suffix: "dreaming-narrative-rem-reconciliation-active", + updatedAt: postStartupUpdatedAt, + }), + ); + vi.setSystemTime(postStartupUpdatedAt + ORPHAN_AGE_MS + 1); + return []; + }); + + await gateway.start(); + + expect(Date.now() - postStartupUpdatedAt).toBeGreaterThan(ORPHAN_AGE_MS); + expect(hasSession(staleAtStartup)).toBe(false); + expect(postStartupSessions.map(hasSession)).toEqual([true, true]); + }); + + it("cleans the independent SQLite stores of every configured dreaming agent", async () => { + const now = Date.now(); + const stale = await Promise.all( + ["main", "researcher"].map((agentId) => + seedSession({ + agentId, + suffix: "dreaming-narrative-light-interrupted", + pluginOwnerId: "memory-core", + updatedAt: now - ORPHAN_AGE_MS - 1, + transcriptAt: now - ORPHAN_AGE_MS - 1, + }), + ), + ); + const gateway = createGateway({ agentIds: ["main", "researcher"] }); + + await gateway.start(); + + expect(stale.map(hasSession)).toEqual([false, false]); + }); + + it("preserves unconfigured agents sharing a configured agent's SQLite store", async () => { + const storePath = path.join(stateDir, "shared.sqlite"); + const staleUpdatedAt = Date.now() - ORPHAN_AGE_MS - 1; + const main = await seedSession({ + agentId: "main", + suffix: "dreaming-narrative-main-interrupted", + updatedAt: staleUpdatedAt, + storePath, + }); + const researcher = await seedSession({ + agentId: "researcher", + suffix: "dreaming-narrative-researcher-interrupted", + updatedAt: staleUpdatedAt, + storePath, + }); + const gateway = createGateway({ sessionStore: storePath }); + + await gateway.start(); + + expect(getSessionEntry({ ...main, storePath })).toBeUndefined(); + expect(getSessionEntry({ ...researcher, storePath })).toMatchObject({ + sessionId: "researcher-dreaming-narrative-researcher-interrupted", + }); + }); + + it("replaces an earlier gateway generation's deferred cleanup timer", async () => { + vi.useFakeTimers({ now: new Date("2026-08-01T12:00:00.000Z") }); + const interrupted = await seedSession({ + suffix: "dreaming-narrative-light-interrupted", + updatedAt: Date.now(), + transcriptAt: Date.now(), + }); + const gateway = createGateway(); + + await gateway.start(); + await vi.advanceTimersByTimeAsync(120_000); + await gateway.start(); + await vi.advanceTimersByTimeAsync(ORPHAN_AGE_MS - 120_000); + + expect(hasSession(interrupted)).toBe(true); + + await vi.advanceTimersByTimeAsync(120_000); + + expect(hasSession(interrupted)).toBe(false); + }); + + it("cancels deferred orphan cleanup when the gateway stops", async () => { + vi.useFakeTimers({ now: new Date("2026-08-01T12:00:00.000Z") }); + const interrupted = await seedSession({ + suffix: "dreaming-narrative-light-interrupted", + updatedAt: Date.now(), + transcriptAt: Date.now(), + }); + const gateway = createGateway(); + + await gateway.start(); + await gateway.stop(); + await vi.advanceTimersByTimeAsync(ORPHAN_AGE_MS); + + expect(hasSession(interrupted)).toBe(true); + }); + + it("still cleans interrupted dreaming sessions when cron startup reconciliation fails", async () => { + const now = Date.now(); + const interrupted = await seedSession({ + suffix: "dreaming-narrative-light-interrupted", + updatedAt: now - ORPHAN_AGE_MS - 1, + transcriptAt: now - ORPHAN_AGE_MS - 1, + }); + const gateway = createGateway({ failCronReconciliation: true }); + + await gateway.start(); + + expect(hasSession(interrupted)).toBe(false); + expect(gateway.logger.error).toHaveBeenCalledWith( + expect.stringContaining("dreaming startup reconciliation failed"), + ); + }); +}); diff --git a/extensions/memory-core/src/dreaming.ts b/extensions/memory-core/src/dreaming.ts index 542286c5a1a0..798cbdbe8666 100644 --- a/extensions/memory-core/src/dreaming.ts +++ b/extensions/memory-core/src/dreaming.ts @@ -827,8 +827,10 @@ export function registerShortTermPromotionDreaming(api: OpenClawPluginApi): void let lastRuntimeConfigKey: string | null = null; let lastRuntimeCronRef: CronServiceLike | null = null; let startupCronRetryTimer: ReturnType | null = null; + let startupDreamingCleanupTimer: ReturnType | null = null; let runtimeCronReconcileTimer: ReturnType | null = null; let startupCronRetryAttempts = 0; + let gatewayLifecycleGeneration = 0; let disposed = false; const resolveCurrentConfig = (): OpenClawConfig => @@ -855,7 +857,12 @@ export function registerShortTermPromotionDreaming(api: OpenClawPluginApi): void const disposeStartupCronRetry = (): void => { disposed = true; + gatewayLifecycleGeneration += 1; clearStartupCronRetry(); + if (startupDreamingCleanupTimer) { + clearTimeout(startupDreamingCleanupTimer); + startupDreamingCleanupTimer = null; + } if (runtimeCronReconcileTimer) { clearInterval(runtimeCronReconcileTimer); runtimeCronReconcileTimer = null; @@ -1014,8 +1021,83 @@ export function registerShortTermPromotionDreaming(api: OpenClawPluginApi): void runtimeCronReconcileTimer.unref?.(); }; + const startDreamingSessionCleanup = async ( + config: OpenClawConfig, + generation: number, + startupStartedAtMs: number, + ): Promise => { + const { DREAMING_ORPHAN_MIN_AGE_MS, scrubDreamingNarrativeArtifacts } = + await import("./dreaming-session-cleanup.js"); + if (disposed || generation !== gatewayLifecycleGeneration) { + return; + } + const scrubConfiguredAgents = async ( + currentConfig: OpenClawConfig, + nowMs?: number, + ): Promise => { + const agentIds = uniqueStrings( + resolveMemoryDreamingWorkspaces(currentConfig).flatMap( + ({ agentIds: workspaceAgentIds }) => workspaceAgentIds, + ), + ); + for (const agentId of agentIds) { + if (disposed || generation !== gatewayLifecycleGeneration) { + return; + } + try { + await scrubDreamingNarrativeArtifacts({ + agentId, + config: currentConfig, + logger: api.logger, + ...(nowMs === undefined ? {} : { nowMs }), + }); + } catch (error) { + api.logger.warn( + `memory-core: dreaming startup cleanup failed for agent ${agentId}: ${formatErrorMessage(error)}`, + ); + } + } + }; + + // Cron reconciliation can itself stall; never classify sessions admitted after startup. + await scrubConfiguredAgents(config, startupStartedAtMs); + if (disposed || generation !== gatewayLifecycleGeneration) { + return; + } + // Interrupted runs are initially indistinguishable from live runs; revisit once their + // persisted activity ages past the same guard used by normal narrative cleanup. + const cleanupTimer = setTimeout(() => { + if ( + disposed || + generation !== gatewayLifecycleGeneration || + startupDreamingCleanupTimer !== cleanupTimer + ) { + return; + } + startupDreamingCleanupTimer = null; + // Keep the cutoff strictly before startup: equal-millisecond sessions may have + // started after the hook and must survive even when this timer runs late. + void scrubConfiguredAgents( + resolveCurrentConfig(), + startupStartedAtMs + DREAMING_ORPHAN_MIN_AGE_MS - 1, + ).catch((error: unknown) => { + api.logger.warn( + `memory-core: deferred dreaming startup cleanup failed: ${formatErrorMessage(error)}`, + ); + }); + }, DREAMING_ORPHAN_MIN_AGE_MS); + startupDreamingCleanupTimer = cleanupTimer; + startupDreamingCleanupTimer.unref?.(); + }; + api.on("gateway_start", async (_event, ctx) => { + const startupStartedAtMs = Date.now(); disposed = false; + if (startupDreamingCleanupTimer) { + clearTimeout(startupDreamingCleanupTimer); + startupDreamingCleanupTimer = null; + } + const generation = ++gatewayLifecycleGeneration; // Store the gateway context for runtime cron resolution retries. gatewayContext = ctx as unknown as { getCron?: () => CronServiceLike | null }; try { @@ -1031,6 +1113,15 @@ export function registerShortTermPromotionDreaming(api: OpenClawPluginApi): void } finally { startRuntimeCronReconcileTimer(); scheduleStartupCronRetry(); + await startDreamingSessionCleanup( + ctx.config ?? api.config, + generation, + startupStartedAtMs, + ).catch((error: unknown) => { + api.logger.warn( + `memory-core: dreaming startup cleanup failed: ${formatErrorMessage(error)}`, + ); + }); } }); diff --git a/src/config/sessions/session-accessor.conformance.test.ts b/src/config/sessions/session-accessor.conformance.test.ts index 633189430a63..9216efdeb0c0 100644 --- a/src/config/sessions/session-accessor.conformance.test.ts +++ b/src/config/sessions/session-accessor.conformance.test.ts @@ -392,26 +392,26 @@ describe.each([publicAccessorAdapter, sqliteAdapter])( } }; - await adapter.upsertSessionEntry(scopedEntry("agent:main:lifecycle-cleanup-missing"), { + await adapter.replaceSessionEntry(scopedEntry("agent:main:lifecycle-cleanup-missing"), { sessionId: "missing-lifecycle", updatedAt: oldTimestamp, }); - await adapter.upsertSessionEntry(scopedEntry("agent:main:lifecycle-cleanup-removed"), { + await adapter.replaceSessionEntry(scopedEntry("agent:main:lifecycle-cleanup-removed"), { sessionId: "removed-lifecycle", updatedAt: oldTimestamp, }); - await adapter.upsertSessionEntry(scopedEntry("agent:main:lifecycle-cleanup-fresh"), { + await adapter.replaceSessionEntry(scopedEntry("agent:main:lifecycle-cleanup-fresh"), { sessionId: "fresh-lifecycle", updatedAt: nowMs, }); - await adapter.upsertSessionEntry( + await adapter.replaceSessionEntry( scopedEntry("agent:main:telegram:group:lifecycle-cleanup-room"), { sessionId: "kept-by-segment", updatedAt: oldTimestamp, }, ); - await adapter.upsertSessionEntry(scopedEntry("agent:main:regular"), { + await adapter.replaceSessionEntry(scopedEntry("agent:main:regular"), { sessionId: "referenced", updatedAt: oldTimestamp, }); @@ -435,6 +435,18 @@ describe.each([publicAccessorAdapter, sqliteAdapter])( old: true, }); + for (const sessionKey of [ + "agent:main:lifecycle-cleanup-missing", + "agent:main:lifecycle-cleanup-removed", + "agent:main:telegram:group:lifecycle-cleanup-room", + "agent:main:regular", + ]) { + expect(adapter.readSessionUpdatedAt(scopedEntry(sessionKey))).toBe(oldTimestamp); + } + expect(adapter.readSessionUpdatedAt(scopedEntry("agent:main:lifecycle-cleanup-fresh"))).toBe( + nowMs, + ); + await expect( adapter.cleanupSessionLifecycleArtifacts({ storePath: cleanupStorePath, diff --git a/src/config/sessions/session-accessor.lifecycle-types.ts b/src/config/sessions/session-accessor.lifecycle-types.ts index 4c043b607d2b..e110a943b0ed 100644 --- a/src/config/sessions/session-accessor.lifecycle-types.ts +++ b/src/config/sessions/session-accessor.lifecycle-types.ts @@ -8,6 +8,8 @@ export type SessionLifecycleArtifactCleanupParams = { agentId?: string; storePath: string; archiveRemovedEntryTranscripts?: boolean; + /** Preserve explicitly foreign plugin-owned state while retaining ownerless legacy rows. */ + pluginOwnerId?: string; sessionKeySegmentPrefix: string; transcriptContentMarker: string; orphanTranscriptMinAgeMs: number; diff --git a/src/config/sessions/session-accessor.sqlite-cleanup-race.test.ts b/src/config/sessions/session-accessor.sqlite-cleanup-race.test.ts index f59453f2aebf..d1fff1431de9 100644 --- a/src/config/sessions/session-accessor.sqlite-cleanup-race.test.ts +++ b/src/config/sessions/session-accessor.sqlite-cleanup-race.test.ts @@ -34,6 +34,298 @@ describe("SQLite lifecycle cleanup races", () => { closeOpenClawAgentDatabasesForTest(); }); + it("ages transcript-free session rows before reclaiming them", async () => { + const now = Date.now(); + const activeSessionKey = "agent:main:cleanup-race-active"; + const orphanSessionKey = "agent:main:cleanup-race-orphan"; + await replaceSessionEntry( + { sessionKey: activeSessionKey, storePath }, + { sessionId: "active-without-transcript", updatedAt: now }, + ); + await replaceSessionEntry( + { sessionKey: orphanSessionKey, storePath }, + { sessionId: "orphan-without-transcript", updatedAt: now - 600_000 }, + ); + + await expect( + cleanupSessionLifecycleArtifacts({ + storePath, + sessionKeySegmentPrefix: "cleanup-race-", + transcriptContentMarker: "cleanup-race-marker", + orphanTranscriptMinAgeMs: 300_000, + nowMs: now, + }), + ).resolves.toEqual({ removedEntries: 1, archivedTranscriptArtifacts: 0 }); + + expect(loadSessionEntry({ sessionKey: activeSessionKey, storePath })).toMatchObject({ + sessionId: "active-without-transcript", + }); + expect(loadSessionEntry({ sessionKey: orphanSessionKey, storePath })).toBeUndefined(); + }); + + it("preserves newly admitted sessions reusing an older transcript", async () => { + const now = Date.now(); + const sessionKey = "agent:main:cleanup-race-reused"; + const sessionId = "reused-active-session"; + await replaceSessionEntry({ sessionKey, storePath }, { sessionId, updatedAt: now }); + await replaceSqliteTranscriptEvents({ sessionKey, sessionId, storePath }, [ + { + runId: "cleanup-race-marker-reused", + timestamp: new Date(now - 600_000).toISOString(), + type: "metadata", + }, + ]); + + await expect( + cleanupSessionLifecycleArtifacts({ + storePath, + sessionKeySegmentPrefix: "cleanup-race-", + transcriptContentMarker: "cleanup-race-marker", + orphanTranscriptMinAgeMs: 300_000, + nowMs: now, + }), + ).resolves.toEqual({ removedEntries: 0, archivedTranscriptArtifacts: 0 }); + + expect(loadSessionEntry({ sessionKey, storePath })).toMatchObject({ sessionId }); + }); + + it("preserves foreign plugin ownership while reclaiming owned and legacy rows", async () => { + const now = Date.now(); + const staleUpdatedAt = now - 600_000; + const ownedKey = "agent:main:cleanup-race-owned"; + const legacyKey = "agent:main:cleanup-race-legacy"; + const foreignKey = "agent:main:cleanup-race-foreign"; + await replaceSessionEntry( + { sessionKey: ownedKey, storePath }, + { sessionId: "owned-session", updatedAt: staleUpdatedAt, pluginOwnerId: "memory-core" }, + ); + await replaceSessionEntry( + { sessionKey: legacyKey, storePath }, + { sessionId: "legacy-session", updatedAt: staleUpdatedAt }, + ); + await replaceSessionEntry( + { sessionKey: foreignKey, storePath }, + { sessionId: "foreign-session", updatedAt: staleUpdatedAt, pluginOwnerId: "other-plugin" }, + ); + + await expect( + cleanupSessionLifecycleArtifacts({ + storePath, + pluginOwnerId: "memory-core", + sessionKeySegmentPrefix: "cleanup-race-", + transcriptContentMarker: "cleanup-race-marker", + orphanTranscriptMinAgeMs: 300_000, + nowMs: now, + }), + ).resolves.toEqual({ removedEntries: 2, archivedTranscriptArtifacts: 0 }); + + expect(loadSessionEntry({ sessionKey: ownedKey, storePath })).toBeUndefined(); + expect(loadSessionEntry({ sessionKey: legacyKey, storePath })).toBeUndefined(); + expect(loadSessionEntry({ sessionKey: foreignKey, storePath })).toMatchObject({ + pluginOwnerId: "other-plugin", + }); + }); + + it("keeps another agent's current and historical sessions in a shared SQLite store", async () => { + const now = Date.now(); + const staleUpdatedAt = now - 600_000; + const sharedStorePath = path.join(tempDir, "shared.sqlite"); + const mainKey = "agent:main:cleanup-race-main"; + const researcherKey = "agent:researcher:cleanup-race-researcher"; + const researcherHistoryKey = "agent:researcher:normal-session"; + const researcherHistoryId = "researcher-orphaned-history"; + + await replaceSessionEntry( + { agentId: "main", sessionKey: mainKey, storePath: sharedStorePath }, + { sessionId: "main-orphan", updatedAt: staleUpdatedAt, pluginOwnerId: "memory-core" }, + ); + await replaceSessionEntry( + { agentId: "researcher", sessionKey: researcherKey, storePath: sharedStorePath }, + { sessionId: "researcher-orphan", updatedAt: staleUpdatedAt, pluginOwnerId: "memory-core" }, + ); + await replaceSessionEntry( + { agentId: "researcher", sessionKey: researcherHistoryKey, storePath: sharedStorePath }, + { sessionId: researcherHistoryId, updatedAt: staleUpdatedAt }, + ); + const researcherHistoryEvent = { + runId: "cleanup-race-marker-researcher-history", + timestamp: new Date(staleUpdatedAt).toISOString(), + type: "metadata", + }; + await replaceSqliteTranscriptEvents( + { + agentId: "researcher", + sessionKey: researcherHistoryKey, + sessionId: researcherHistoryId, + storePath: sharedStorePath, + }, + [researcherHistoryEvent], + ); + await replaceSessionEntry( + { agentId: "researcher", sessionKey: researcherHistoryKey, storePath: sharedStorePath }, + { sessionId: "researcher-current", updatedAt: now }, + ); + + await expect( + cleanupSessionLifecycleArtifacts({ + agentId: "main", + storePath: sharedStorePath, + pluginOwnerId: "memory-core", + sessionKeySegmentPrefix: "cleanup-race-", + transcriptContentMarker: "cleanup-race-marker", + orphanTranscriptMinAgeMs: 300_000, + nowMs: now, + }), + ).resolves.toEqual({ removedEntries: 1, archivedTranscriptArtifacts: 0 }); + + expect( + loadSessionEntry({ agentId: "main", sessionKey: mainKey, storePath: sharedStorePath }), + ).toBeUndefined(); + expect( + loadSessionEntry({ + agentId: "researcher", + sessionKey: researcherKey, + storePath: sharedStorePath, + }), + ).toMatchObject({ sessionId: "researcher-orphan" }); + await expect( + loadTranscriptEvents({ + agentId: "researcher", + sessionKey: researcherHistoryKey, + sessionId: researcherHistoryId, + storePath: sharedStorePath, + }), + ).resolves.toEqual([researcherHistoryEvent]); + }); + + it("does not reclaim orphaned historical windows owned by another plugin", async () => { + const now = Date.now(); + const staleUpdatedAt = now - 600_000; + const foreignKey = "agent:main:foreign-history"; + const foreignHistoryId = "foreign-history-previous"; + await replaceSessionEntry( + { sessionKey: foreignKey, storePath }, + { sessionId: foreignHistoryId, updatedAt: staleUpdatedAt, pluginOwnerId: "other-plugin" }, + ); + const foreignEvent = { + type: "metadata", + timestamp: new Date(staleUpdatedAt).toISOString(), + runId: "cleanup-race-marker-foreign", + }; + await replaceSqliteTranscriptEvents( + { sessionKey: foreignKey, sessionId: foreignHistoryId, storePath }, + [foreignEvent], + ); + await replaceSessionEntry( + { sessionKey: foreignKey, storePath }, + { sessionId: "foreign-history-current", updatedAt: now, pluginOwnerId: "other-plugin" }, + ); + + await expect( + cleanupSessionLifecycleArtifacts({ + storePath, + pluginOwnerId: "memory-core", + sessionKeySegmentPrefix: "cleanup-race-", + transcriptContentMarker: "cleanup-race-marker", + orphanTranscriptMinAgeMs: 300_000, + nowMs: now, + }), + ).resolves.toEqual({ removedEntries: 0, archivedTranscriptArtifacts: 0 }); + + await expect( + loadTranscriptEvents({ sessionKey: foreignKey, sessionId: foreignHistoryId, storePath }), + ).resolves.toEqual([foreignEvent]); + }); + + it("preserves foreign current windows when their session node is a placeholder", async () => { + const now = Date.now(); + const sessionKey = "agent:main:cleanup-race-placeholder"; + const sessionId = "foreign-placeholder-session"; + await replaceSessionEntry( + { sessionKey, storePath }, + { sessionId, updatedAt: now - 600_000, pluginOwnerId: "other-plugin" }, + ); + const event = { + runId: "cleanup-race-marker-foreign-placeholder", + timestamp: new Date(now - 600_000).toISOString(), + type: "metadata", + }; + await replaceSqliteTranscriptEvents({ sessionKey, sessionId, storePath }, [event]); + const databasePath = resolveSqliteTargetFromSessionStorePath(storePath, { + agentId: "main", + }).path; + const database = openOpenClawAgentDatabase({ agentId: "main", path: databasePath }); + database.db + .prepare("UPDATE session_nodes SET entry_json = ?, entry_valid = ? WHERE session_key = ?") + .run("{}", -1, sessionKey); + + await expect( + cleanupSessionLifecycleArtifacts({ + storePath, + pluginOwnerId: "memory-core", + sessionKeySegmentPrefix: "cleanup-race-", + transcriptContentMarker: "cleanup-race-marker", + orphanTranscriptMinAgeMs: 300_000, + nowMs: now, + }), + ).resolves.toEqual({ removedEntries: 0, archivedTranscriptArtifacts: 0 }); + + expect( + database.db + .prepare("SELECT current_session_id FROM session_nodes WHERE session_key = ?") + .get(sessionKey), + ).toEqual({ current_session_id: sessionId }); + await expect(loadTranscriptEvents({ sessionKey, sessionId, storePath })).resolves.toEqual([ + event, + ]); + }); + + it("preserves owned nodes that still reference another plugin's historical generation", async () => { + const now = Date.now(); + const sessionKey = "agent:main:cleanup-race-mixed-generations"; + const foreignSessionId = "mixed-foreign-history"; + await replaceSessionEntry( + { sessionKey, storePath }, + { + sessionId: foreignSessionId, + updatedAt: now - 600_000, + pluginOwnerId: "other-plugin", + }, + ); + await replaceSqliteTranscriptEvents({ sessionKey, sessionId: foreignSessionId, storePath }, [ + { + runId: "cleanup-race-marker-mixed-foreign", + timestamp: new Date(now - 600_000).toISOString(), + type: "metadata", + }, + ]); + await replaceSessionEntry( + { sessionKey, storePath }, + { + previousSessionId: foreignSessionId, + sessionId: "mixed-owned-current", + updatedAt: now - 600_000, + pluginOwnerId: "memory-core", + }, + ); + + await expect( + cleanupSessionLifecycleArtifacts({ + storePath, + pluginOwnerId: "memory-core", + sessionKeySegmentPrefix: "cleanup-race-", + transcriptContentMarker: "cleanup-race-marker", + orphanTranscriptMinAgeMs: 300_000, + nowMs: now, + }), + ).resolves.toEqual({ removedEntries: 0, archivedTranscriptArtifacts: 0 }); + + expect(loadSessionEntry({ sessionKey, storePath })).toMatchObject({ + sessionId: "mixed-owned-current", + previousSessionId: foreignSessionId, + }); + }); + it("revalidates entries before deleting their transcript state", async () => { const sessionKey = "agent:main:cleanup-race"; const sessionId = "cleanup-race-session"; diff --git a/src/config/sessions/session-accessor.sqlite-lifecycle-state.ts b/src/config/sessions/session-accessor.sqlite-lifecycle-state.ts index ddcee4c5274a..8087ed41182b 100644 --- a/src/config/sessions/session-accessor.sqlite-lifecycle-state.ts +++ b/src/config/sessions/session-accessor.sqlite-lifecycle-state.ts @@ -3,6 +3,7 @@ import { executeSqliteQuerySync, executeSqliteQueryTakeFirstSync, } from "../../infra/kysely-sync.js"; +import { normalizeAgentId, parseAgentSessionKey } from "../../routing/session-key.js"; import type { OpenClawAgentDatabase } from "../../state/openclaw-agent-db.js"; import type { MaterializedSqliteSessionStateDeletePlan, @@ -75,6 +76,14 @@ function sessionKeySegmentStartsWith(sessionKey: string, prefix: string): boolea return sessionSegment.startsWith(prefix); } +function sessionKeyBelongsToAgent(sessionKey: string, agentId: string | undefined): boolean { + if (agentId === undefined) { + return true; + } + const parsed = parseAgentSessionKey(sessionKey); + return parsed !== null && normalizeAgentId(parsed.agentId) === normalizeAgentId(agentId); +} + function readSessionTranscriptUpdatedAt( database: OpenClawAgentDatabase, sessionId: string, @@ -95,11 +104,16 @@ function readSessionTranscriptUpdatedAt( function sqliteTranscriptStateIsReclaimable(params: { database: OpenClawAgentDatabase; + sessionUpdatedAt?: number; sessionId: string; nowMs: number; orphanTranscriptMinAgeMs: number; }): boolean { - const updatedAt = readSessionTranscriptUpdatedAt(params.database, params.sessionId); + const transcriptUpdatedAt = readSessionTranscriptUpdatedAt(params.database, params.sessionId); + const updatedAt = + params.sessionUpdatedAt === undefined + ? transcriptUpdatedAt + : Math.max(params.sessionUpdatedAt, transcriptUpdatedAt ?? params.sessionUpdatedAt); return updatedAt === undefined || params.nowMs - updatedAt >= params.orphanTranscriptMinAgeMs; } @@ -487,10 +501,12 @@ function deleteSqliteSessionStateRows(database: OpenClawAgentDatabase, sessionId // Plans orphan cleanup without file writes or row deletion; finalization // handles archive durability before removing rows. function planSqliteOrphanLifecycleTranscriptStateDeletes(params: { + agentId?: string; archiveRemovedEntryTranscripts: boolean; archiveDirectory: string; database: OpenClawAgentDatabase; excludedSessionIds?: ReadonlySet; + pluginOwnerId?: string; referencedSessionIds: ReadonlySet; transcriptContentMarker: string; orphanTranscriptMinAgeMs: number; @@ -499,7 +515,10 @@ function planSqliteOrphanLifecycleTranscriptStateDeletes(params: { const db = getSessionKysely(params.database.db); const rows = executeSqliteQuerySync( params.database.db, - db.selectFrom("session_windows").select("session_id").orderBy("session_id", "asc"), + db + .selectFrom("session_windows") + .select(["session_id", "session_key", "plugin_owner_id"]) + .orderBy("session_id", "asc"), ).rows; const deletePlans: SqliteSessionStateDeletePlan[] = []; @@ -507,8 +526,10 @@ function planSqliteOrphanLifecycleTranscriptStateDeletes(params: { // longer the node's current id. The marker scopes cleanup to this lifecycle. for (const row of rows) { if ( + !sessionKeyBelongsToAgent(row.session_key, params.agentId) || params.referencedSessionIds.has(row.session_id) || - params.excludedSessionIds?.has(row.session_id) + params.excludedSessionIds?.has(row.session_id) || + (params.pluginOwnerId && row.plugin_owner_id && row.plugin_owner_id !== params.pluginOwnerId) ) { continue; } @@ -545,8 +566,10 @@ function planSqliteOrphanLifecycleTranscriptStateDeletes(params: { export function planSqliteSessionLifecycleArtifactCleanup( database: OpenClawAgentDatabase, params: { + agentId?: string; archiveRemovedEntryTranscripts: boolean; archiveDirectory: string; + pluginOwnerId?: string; sessionKeySegmentPrefix: string; transcriptContentMarker: string; orphanTranscriptMinAgeMs: number; @@ -558,20 +581,52 @@ export function planSqliteSessionLifecycleArtifactCleanup( database.db, db .selectFrom("session_nodes") - .select(["entry_json", "session_key", "current_session_id"]) + .select(["entry_json", "session_key", "current_session_id", "updated_at"]) .orderBy("session_key", "asc"), ).rows; const removedSessionIds = new Set(); const entries: SqliteLifecycleArtifactCleanupPlan["entries"] = []; const projectedStore = readSqliteSessionEntryStore(database); + const foreignOwnedSessionIds = params.pluginOwnerId + ? new Set( + executeSqliteQuerySync( + database.db, + db + .selectFrom("session_windows") + .select("session_id") + .where("plugin_owner_id", "is not", null) + .where("plugin_owner_id", "!=", params.pluginOwnerId), + ).rows.map((row) => row.session_id), + ) + : undefined; for (const row of rows) { - if (!sessionKeySegmentStartsWith(row.session_key, params.sessionKeySegmentPrefix)) { + if ( + !sessionKeyBelongsToAgent(row.session_key, params.agentId) || + !sessionKeySegmentStartsWith(row.session_key, params.sessionKeySegmentPrefix) + ) { + continue; + } + const entry = parseSessionEntryRow(row); + const sessionIds = uniqueStrings([ + row.current_session_id, + ...(entry ? collectSqliteSessionStateIdsForEntry(entry) : []), + ]); + // Window ownership survives placeholder nodes and ownerless row projections; preserve + // the entire node when any referenced generation belongs to another plugin. + if ( + (params.pluginOwnerId && + entry?.pluginOwnerId && + entry.pluginOwnerId !== params.pluginOwnerId) || + sessionIds.some((sessionId) => foreignOwnedSessionIds?.has(sessionId)) + ) { continue; } if ( !sqliteTranscriptStateIsReclaimable({ database, + // Admission updates the node even when a run has no event yet or reuses old events. + sessionUpdatedAt: normalizeSqliteNumber(row.updated_at), sessionId: row.current_session_id, nowMs: params.nowMs, orphanTranscriptMinAgeMs: params.orphanTranscriptMinAgeMs, @@ -579,10 +634,7 @@ export function planSqliteSessionLifecycleArtifactCleanup( ) { continue; } - const entry = parseSessionEntryRow(row); - for (const sessionId of entry - ? collectSqliteSessionStateIdsForEntry(entry) - : [row.current_session_id]) { + for (const sessionId of sessionIds) { removedSessionIds.add(sessionId); } entries.push({ @@ -612,10 +664,12 @@ export function planSqliteSessionLifecycleArtifactCleanup( } deletePlans.push( ...planSqliteOrphanLifecycleTranscriptStateDeletes({ + ...(params.agentId ? { agentId: params.agentId } : {}), archiveRemovedEntryTranscripts: params.archiveRemovedEntryTranscripts, archiveDirectory: params.archiveDirectory, database, excludedSessionIds: removedSessionIds, + ...(params.pluginOwnerId ? { pluginOwnerId: params.pluginOwnerId } : {}), referencedSessionIds, transcriptContentMarker: params.transcriptContentMarker, orphanTranscriptMinAgeMs: params.orphanTranscriptMinAgeMs, diff --git a/src/config/sessions/session-accessor.sqlite-lifecycle.ts b/src/config/sessions/session-accessor.sqlite-lifecycle.ts index 12f1f271a17e..153543e3b46c 100644 --- a/src/config/sessions/session-accessor.sqlite-lifecycle.ts +++ b/src/config/sessions/session-accessor.sqlite-lifecycle.ts @@ -8,6 +8,7 @@ import { resolveAgentHarnessSessionStoreEntryError, } from "../../sessions/agent-harness-session-key.js"; import { emitSessionIdentityMutation } from "../../sessions/session-lifecycle-events.js"; +import { withOpenClawAgentDatabaseReadOnly } from "../../state/openclaw-agent-db-readonly.js"; import type { DB as OpenClawAgentKyselyDatabase } from "../../state/openclaw-agent-db.generated.js"; import { openOpenClawAgentDatabase, @@ -109,6 +110,7 @@ export async function cleanupSqliteSessionLifecycleArtifacts( ): Promise { const sessionKeySegmentPrefix = params.sessionKeySegmentPrefix.trim(); const transcriptContentMarker = params.transcriptContentMarker; + const pluginOwnerId = params.pluginOwnerId?.trim(); if (!sessionKeySegmentPrefix || !transcriptContentMarker) { return { removedEntries: 0, archivedTranscriptArtifacts: 0 }; } @@ -117,11 +119,18 @@ export async function cleanupSqliteSessionLifecycleArtifacts( ...(params.agentId ? { agentId: params.agentId } : {}), storePath: params.storePath, }); + const databaseOptions = toDatabaseOptions(resolved); + // Maintenance must not turn a read-only startup probe into a newly materialized agent store. + if (!withOpenClawAgentDatabaseReadOnly(() => true, databaseOptions).found) { + return { removedEntries: 0, archivedTranscriptArtifacts: 0 }; + } return await runExclusiveSqliteSessionWrite(resolved, async () => { - const database = openOpenClawAgentDatabase(toDatabaseOptions(resolved)); + const database = openOpenClawAgentDatabase(databaseOptions); const cleanupPlan = planSqliteSessionLifecycleArtifactCleanup(database, { + ...(params.agentId !== undefined ? { agentId: resolved.agentId } : {}), archiveRemovedEntryTranscripts: params.archiveRemovedEntryTranscripts !== false, archiveDirectory: resolveSqliteTranscriptArchiveDirectory(resolved), + ...(pluginOwnerId ? { pluginOwnerId } : {}), sessionKeySegmentPrefix, transcriptContentMarker, orphanTranscriptMinAgeMs: params.orphanTranscriptMinAgeMs, @@ -142,7 +151,7 @@ export async function cleanupSqliteSessionLifecycleArtifacts( transactionDb, cleanupPlan.entries, ); - }, toDatabaseOptions(resolved)); + }, databaseOptions); emitCommittedSessionEntryRemovals(cleanupPlan.entries); return { removedEntries, diff --git a/src/plugin-sdk/session-store-runtime.ts b/src/plugin-sdk/session-store-runtime.ts index 562a0a952867..32aa3f22ebe8 100644 --- a/src/plugin-sdk/session-store-runtime.ts +++ b/src/plugin-sdk/session-store-runtime.ts @@ -133,6 +133,7 @@ type SessionLifecycleArtifactsCleanupParams = { archiveRemovedEntryTranscripts?: boolean; env?: NodeJS.ProcessEnv; orphanTranscriptMinAgeMs: number; + pluginOwnerId?: string; sessionStore?: string; sessionKeySegmentPrefix: string; storePath?: string; @@ -572,6 +573,7 @@ export async function cleanupSessionLifecycleArtifacts( storePath, ...(params.agentId !== undefined ? { agentId: params.agentId } : {}), archiveRemovedEntryTranscripts: params.archiveRemovedEntryTranscripts, + ...(params.pluginOwnerId !== undefined ? { pluginOwnerId: params.pluginOwnerId } : {}), sessionKeySegmentPrefix: params.sessionKeySegmentPrefix, transcriptContentMarker: params.transcriptContentMarker, orphanTranscriptMinAgeMs: params.orphanTranscriptMinAgeMs, diff --git a/test/dreaming-startup-cleanup.e2e.test.ts b/test/dreaming-startup-cleanup.e2e.test.ts new file mode 100644 index 000000000000..6c0d0a40fab1 --- /dev/null +++ b/test/dreaming-startup-cleanup.e2e.test.ts @@ -0,0 +1,187 @@ +// A real Gateway restart must remove interrupted Dreaming sessions from its public session list. +import { afterEach, describe, expect, it, vi } from "vitest"; +import type { OpenClawConfig } from "../src/config/types.openclaw.js"; +import { connectGatewayClient, disconnectGatewayClient } from "../src/gateway/test-helpers.e2e.js"; +import { getSessionEntry, upsertSessionEntry } from "../src/plugin-sdk/session-store-runtime.js"; +import { + appendSqliteSessionTranscriptEventForTest, + closeOpenClawAgentDatabasesForTest, +} from "../src/plugin-sdk/sqlite-runtime-testing.js"; +import { + createOpenClawTestInstance, + type OpenClawTestInstance, +} from "./helpers/openclaw-test-instance.js"; + +const STALE_AGE_MS = 600_000; +const WAIT_OPTIONS = { interval: 50, timeout: 15_000 } as const; +const instances: OpenClawTestInstance[] = []; + +type GatewaySessionClient = Awaited>; + +afterEach(async () => { + await Promise.all(instances.splice(0).map(async (instance) => await instance.cleanup())); + closeOpenClawAgentDatabasesForTest(); +}); + +async function seedSession(params: { + agentId: string; + suffix: string; + updatedAt: number; + pluginOwnerId?: string; + transcript?: boolean; +}): Promise { + const sessionKey = `agent:${params.agentId}:${params.suffix}`; + const sessionId = `${params.agentId}-${params.suffix}`; + await upsertSessionEntry({ + agentId: params.agentId, + sessionKey, + entry: { + sessionId, + updatedAt: params.updatedAt, + ...(params.pluginOwnerId ? { pluginOwnerId: params.pluginOwnerId } : {}), + }, + }); + if (params.transcript) { + await appendSqliteSessionTranscriptEventForTest({ + agentId: params.agentId, + sessionId, + sessionKey, + event: { + runId: `dreaming-narrative-${sessionId}`, + timestamp: params.updatedAt, + type: "metadata", + }, + }); + } + return sessionKey; +} + +async function listSessionKeys(client: GatewaySessionClient): Promise { + const result = await client.request<{ sessions: Array<{ key: string }> }>("sessions.list", { + includeGlobal: true, + includeUnknown: true, + limit: 100, + }); + return result.sessions.map(({ key }) => key); +} + +async function connect(instance: OpenClawTestInstance): Promise { + return await connectGatewayClient({ + url: instance.url, + token: instance.gatewayToken, + role: "operator", + scopes: ["operator.admin", "operator.read", "operator.write"], + }); +} + +describe("Gateway dreaming session restart cleanup", () => { + it("removes stale child sessions after a real restart even when dreaming and cron are disabled", async () => { + const config = { + agents: { list: [{ id: "main", default: true }, { id: "worker" }] }, + plugins: { + enabled: true, + allow: ["memory-core"], + slots: { memory: "memory-core" }, + entries: { + "memory-core": { + enabled: true, + config: { dreaming: { enabled: false } }, + }, + }, + }, + } satisfies OpenClawConfig; + const instance = await createOpenClawTestInstance({ + name: "dreaming-startup-cleanup", + config, + env: { OPENCLAW_TEST_MINIMAL_GATEWAY: undefined }, + }); + instances.push(instance); + instance.state.applyEnv(); + expect(instance.env.OPENCLAW_SKIP_CRON).toBe("1"); + + const sentinel = await seedSession({ + agentId: "main", + suffix: "dreaming-narrative-startup-sentinel", + updatedAt: Date.now() - STALE_AGE_MS, + pluginOwnerId: "memory-core", + }); + await instance.startGateway(); + let client = await connect(instance); + + try { + // gateway_start fires after the listener opens; observe its first sweep before + // creating interrupted rows that must survive until the second process starts. + await vi.waitFor(() => { + expect(getSessionEntry({ agentId: "main", sessionKey: sentinel }), instance.logs()).toBe( + undefined, + ); + }, WAIT_OPTIONS); + + const now = Date.now(); + const stale = await Promise.all([ + ...["light", "rem", "deep", "consolidation"].map(async (phase) => + seedSession({ + agentId: "main", + suffix: `dreaming-narrative-${phase}-interrupted`, + updatedAt: now - STALE_AGE_MS, + transcript: true, + ...(phase === "rem" ? {} : { pluginOwnerId: "memory-core" }), + }), + ), + seedSession({ + agentId: "worker", + suffix: "dreaming-narrative-worker-interrupted", + updatedAt: now - STALE_AGE_MS, + pluginOwnerId: "memory-core", + }), + ]); + const preserved = await Promise.all([ + seedSession({ + agentId: "main", + suffix: "dreaming-narrative-active-with-transcript", + updatedAt: now, + pluginOwnerId: "memory-core", + transcript: true, + }), + seedSession({ + agentId: "main", + suffix: "dreaming-narrative-active-before-transcript", + updatedAt: now, + pluginOwnerId: "memory-core", + }), + seedSession({ + agentId: "main", + suffix: "dreaming-narrative-foreign", + updatedAt: now - STALE_AGE_MS, + pluginOwnerId: "other-plugin", + transcript: true, + }), + seedSession({ + agentId: "main", + suffix: "telegram:group:dreaming-narrative-conversation", + updatedAt: now - STALE_AGE_MS, + }), + ]); + + await vi.waitFor(async () => { + const keys = await listSessionKeys(client); + expect(keys, instance.logs()).toEqual(expect.arrayContaining([...stale, ...preserved])); + }, WAIT_OPTIONS); + + await disconnectGatewayClient(client); + await instance.stopGateway(); + await instance.startGateway(); + client = await connect(instance); + + await vi.waitFor(async () => { + const keys = await listSessionKeys(client); + expect(keys, instance.logs()).toEqual(expect.arrayContaining(preserved)); + for (const sessionKey of stale) { + expect(keys, instance.logs()).not.toContain(sessionKey); + } + }, WAIT_OPTIONS); + } finally { + await disconnectGatewayClient(client).catch(() => undefined); + } + }, 180_000); +});