mirror of
https://github.com/openclaw/openclaw.git
synced 2026-08-12 21:53:00 -06:00
fix(memory): remove interrupted Dreaming sessions after Gateway restart (#117885)
* fix(memory): clean interrupted Dreaming sessions after Gateway restart * test(sessions): preserve historical timestamps in lifecycle fixtures * test(plugins): remove duplicate registry handle mock from main * style(sessions): format lifecycle conformance assertion
This commit is contained in:
committed by
GitHub
parent
26cf4e6130
commit
f5c985d92c
@@ -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<string, unknown>;
|
||||
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",
|
||||
|
||||
@@ -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 = "<!-- openclaw:dreaming:diary:start -->";
|
||||
const DIARY_END_MARKER = "<!-- openclaw:dreaming:diary:end -->";
|
||||
const BACKFILL_ENTRY_MARKER = "openclaw:dreaming:backfill-entry";
|
||||
@@ -814,47 +810,6 @@ async function appendNarrativeEntry(params: {
|
||||
|
||||
// ── Orchestrator ───────────────────────────────────────────────────────
|
||||
|
||||
async function scrubDreamingNarrativeArtifacts(logger: Logger): Promise<void> {
|
||||
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}`,
|
||||
|
||||
@@ -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<void> {
|
||||
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"}.`,
|
||||
);
|
||||
}
|
||||
}
|
||||
@@ -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> | void;
|
||||
|
||||
let stateDir: string;
|
||||
let stopGateway: (() => Promise<void>) | 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<string, GatewayHook>();
|
||||
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"),
|
||||
);
|
||||
});
|
||||
});
|
||||
@@ -827,8 +827,10 @@ export function registerShortTermPromotionDreaming(api: OpenClawPluginApi): void
|
||||
let lastRuntimeConfigKey: string | null = null;
|
||||
let lastRuntimeCronRef: CronServiceLike | null = null;
|
||||
let startupCronRetryTimer: ReturnType<typeof setTimeout> | null = null;
|
||||
let startupDreamingCleanupTimer: ReturnType<typeof setTimeout> | null = null;
|
||||
let runtimeCronReconcileTimer: ReturnType<typeof setInterval> | 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<void> => {
|
||||
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<void> => {
|
||||
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)}`,
|
||||
);
|
||||
});
|
||||
}
|
||||
});
|
||||
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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;
|
||||
|
||||
@@ -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";
|
||||
|
||||
@@ -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<string>;
|
||||
pluginOwnerId?: string;
|
||||
referencedSessionIds: ReadonlySet<string>;
|
||||
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<string>();
|
||||
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,
|
||||
|
||||
@@ -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<SessionLifecycleArtifactCleanupResult> {
|
||||
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,
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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<ReturnType<typeof connectGatewayClient>>;
|
||||
|
||||
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<string> {
|
||||
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<string[]> {
|
||||
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<GatewaySessionClient> {
|
||||
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);
|
||||
});
|
||||
Reference in New Issue
Block a user