Skip to content

Commit 8102434

Browse files
giodl73-repoCopilot
giodl73-repo
and
Copilot
committed
fix(memory): catch up stale sessions on startup
Add a startup catch-up scan for memory session source files so clean gateway restarts compare on-disk transcripts against persisted index file state and mark only missing/newer/resized session files dirty for a normal incremental sync. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
1 parent 2285876 commit 8102434

7 files changed

Lines changed: 327 additions & 7 deletions

‎extensions/memory-core/src/memory/manager-session-sync-state.test.ts‎

Lines changed: 46 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,8 @@
11
import { describe, expect, it } from "vitest";
2-
import { resolveMemorySessionSyncPlan } from "./manager-session-sync-state.js";
2+
import {
3+
resolveMemorySessionStartupDirtyFiles,
4+
resolveMemorySessionSyncPlan,
5+
} from "./manager-session-sync-state.js";
36

47
describe("memory session sync state", () => {
58
it("tracks active paths and bulk hashes for full scans", () => {
@@ -61,4 +64,46 @@ describe("memory session sync state", () => {
6164
expect(plan.indexAll).toBe(false);
6265
expect(plan.activePaths).toEqual(new Set(["sessions/incremental.jsonl"]));
6366
});
67+
68+
it("marks missing and changed startup session files dirty", () => {
69+
const dirtyFiles = resolveMemorySessionStartupDirtyFiles({
70+
files: [
71+
{
72+
absPath: "/tmp/sessions/unchanged.jsonl",
73+
path: "sessions/unchanged.jsonl",
74+
mtimeMs: 100,
75+
size: 10,
76+
},
77+
{
78+
absPath: "/tmp/sessions/newer.jsonl",
79+
path: "sessions/newer.jsonl",
80+
mtimeMs: 250,
81+
size: 20,
82+
},
83+
{
84+
absPath: "/tmp/sessions/resized.jsonl",
85+
path: "sessions/resized.jsonl",
86+
mtimeMs: 300,
87+
size: 31,
88+
},
89+
{
90+
absPath: "/tmp/sessions/missing.jsonl",
91+
path: "sessions/missing.jsonl",
92+
mtimeMs: 400,
93+
size: 40,
94+
},
95+
],
96+
existingRows: [
97+
{ path: "sessions/unchanged.jsonl", hash: "hash-unchanged", mtime: 100, size: 10 },
98+
{ path: "sessions/newer.jsonl", hash: "hash-newer", mtime: 200, size: 20 },
99+
{ path: "sessions/resized.jsonl", hash: "hash-resized", mtime: 300, size: 30 },
100+
],
101+
});
102+
103+
expect(dirtyFiles).toEqual([
104+
"/tmp/sessions/newer.jsonl",
105+
"/tmp/sessions/resized.jsonl",
106+
"/tmp/sessions/missing.jsonl",
107+
]);
108+
});
64109
});

‎extensions/memory-core/src/memory/manager-session-sync-state.ts‎

Lines changed: 32 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,37 @@
11
import { type MemorySourceFileStateRow } from "./manager-source-state.js";
22

3+
export type MemorySessionStartupFileState = {
4+
absPath: string;
5+
path: string;
6+
mtimeMs: number;
7+
size: number;
8+
};
9+
10+
export function resolveMemorySessionStartupDirtyFiles(params: {
11+
files: MemorySessionStartupFileState[];
12+
existingRows?: MemorySourceFileStateRow[] | null;
13+
}): string[] {
14+
const indexedRows = new Map((params.existingRows ?? []).map((row) => [row.path, row]));
15+
const dirtyFiles: string[] = [];
16+
for (const file of params.files) {
17+
const existing = indexedRows.get(file.path);
18+
if (!existing) {
19+
dirtyFiles.push(file.absPath);
20+
continue;
21+
}
22+
const indexedMtimeMs = Number(existing.mtime);
23+
const indexedSize = Number(existing.size);
24+
if (!Number.isFinite(indexedMtimeMs) || !Number.isFinite(indexedSize)) {
25+
dirtyFiles.push(file.absPath);
26+
continue;
27+
}
28+
if (file.size !== indexedSize || file.mtimeMs > indexedMtimeMs) {
29+
dirtyFiles.push(file.absPath);
30+
}
31+
}
32+
return dirtyFiles;
33+
}
34+
335
export function resolveMemorySessionSyncPlan(params: {
436
needsFullReindex: boolean;
537
files: string[];

‎extensions/memory-core/src/memory/manager-source-state.test.ts‎

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -15,8 +15,8 @@ describe("memory source state", () => {
1515
all: (...args) => {
1616
calls.push({ sql, args });
1717
return [
18-
{ path: "memory/one.md", hash: "hash-1" },
19-
{ path: "memory/two.md", hash: "hash-2" },
18+
{ path: "memory/one.md", hash: "hash-1", mtime: 100, size: 10 },
19+
{ path: "memory/two.md", hash: "hash-2", mtime: 200, size: 20 },
2020
];
2121
},
2222
get: () => undefined,
@@ -27,8 +27,8 @@ describe("memory source state", () => {
2727

2828
expect(calls).toEqual([{ sql: MEMORY_SOURCE_FILE_STATE_SQL, args: ["memory"] }]);
2929
expect(state.rows).toEqual([
30-
{ path: "memory/one.md", hash: "hash-1" },
31-
{ path: "memory/two.md", hash: "hash-2" },
30+
{ path: "memory/one.md", hash: "hash-1", mtime: 100, size: 10 },
31+
{ path: "memory/two.md", hash: "hash-2", mtime: 200, size: 20 },
3232
]);
3333
expect(state.hashes).toEqual(
3434
new Map([

‎extensions/memory-core/src/memory/manager-source-state.ts‎

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -4,6 +4,8 @@ import type { MemorySource } from "openclaw/plugin-sdk/memory-core-host-engine-s
44
export type MemorySourceFileStateRow = {
55
path: string;
66
hash: string;
7+
mtime?: number;
8+
size?: number;
79
};
810

911
type MemorySourceStateDb = {
@@ -13,7 +15,7 @@ type MemorySourceStateDb = {
1315
};
1416
};
1517

16-
export const MEMORY_SOURCE_FILE_STATE_SQL = `SELECT path, hash FROM files WHERE source = ?`;
18+
export const MEMORY_SOURCE_FILE_STATE_SQL = `SELECT path, hash, mtime, size FROM files WHERE source = ?`;
1719
export const MEMORY_SOURCE_FILE_HASH_SQL = `SELECT hash FROM files WHERE path = ? AND source = ?`;
1820

1921
export function loadMemorySourceFileState(params: {

‎extensions/memory-core/src/memory/manager-sync-ops.startup-catchup.test.ts‎

Lines changed: 175 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,175 @@
1+
import fs from "node:fs/promises";
2+
import os from "node:os";
3+
import path from "node:path";
4+
import type { DatabaseSync } from "node:sqlite";
5+
import {
6+
resolveSessionTranscriptsDirForAgent,
7+
type OpenClawConfig,
8+
type ResolvedMemorySearchConfig,
9+
} from "openclaw/plugin-sdk/memory-core-host-engine-foundation";
10+
import type {
11+
MemorySource,
12+
MemorySyncProgressUpdate,
13+
} from "openclaw/plugin-sdk/memory-core-host-engine-storage";
14+
import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
15+
import { MemoryManagerSyncOps } from "./manager-sync-ops.js";
16+
17+
type MemoryIndexEntry = {
18+
path: string;
19+
absPath: string;
20+
mtimeMs: number;
21+
size: number;
22+
hash: string;
23+
content?: string;
24+
};
25+
26+
type SyncParams = {
27+
reason?: string;
28+
force?: boolean;
29+
sessionFiles?: string[];
30+
progress?: (update: MemorySyncProgressUpdate) => void;
31+
};
32+
33+
type SourceStateRow = { path: string; hash: string; mtime: number; size: number };
34+
35+
class SessionStartupCatchupHarness extends MemoryManagerSyncOps {
36+
protected readonly cfg = {} as OpenClawConfig;
37+
protected readonly agentId = "main";
38+
protected readonly workspaceDir = "/tmp/openclaw-test-workspace";
39+
protected readonly settings = {
40+
sync: {
41+
sessions: {
42+
deltaBytes: 100_000,
43+
deltaMessages: 50,
44+
postCompactionForce: true,
45+
},
46+
},
47+
} as ResolvedMemorySearchConfig;
48+
protected readonly batch = {
49+
enabled: false,
50+
wait: false,
51+
concurrency: 1,
52+
pollIntervalMs: 0,
53+
timeoutMs: 0,
54+
};
55+
protected readonly vector = { enabled: false, available: false };
56+
protected readonly cache = { enabled: false };
57+
protected db: DatabaseSync;
58+
59+
readonly syncCalls: SyncParams[] = [];
60+
61+
constructor(sourceRows: SourceStateRow[]) {
62+
super();
63+
this.sources.add("sessions");
64+
this.db = {
65+
prepare: () => ({
66+
all: () => sourceRows,
67+
get: () => undefined,
68+
run: () => undefined,
69+
}),
70+
} as unknown as DatabaseSync;
71+
}
72+
73+
async catchUp(): Promise<string[]> {
74+
return await this.runSessionStartupCatchup();
75+
}
76+
77+
getDirtySessionFiles(): string[] {
78+
return Array.from(this.sessionsDirtyFiles);
79+
}
80+
81+
isSessionsDirty(): boolean {
82+
return this.sessionsDirty;
83+
}
84+
85+
protected computeProviderKey(): string {
86+
return "test";
87+
}
88+
89+
protected async sync(params?: SyncParams): Promise<void> {
90+
this.syncCalls.push(params ?? {});
91+
}
92+
93+
protected async withTimeout<T>(
94+
promise: Promise<T>,
95+
_timeoutMs: number,
96+
_message: string,
97+
): Promise<T> {
98+
return await promise;
99+
}
100+
101+
protected getIndexConcurrency(): number {
102+
return 1;
103+
}
104+
105+
protected pruneEmbeddingCacheIfNeeded(): void {}
106+
107+
protected async indexFile(
108+
_entry: MemoryIndexEntry,
109+
_options: { source: MemorySource; content?: string },
110+
): Promise<void> {}
111+
}
112+
113+
describe("session startup catch-up", () => {
114+
let stateDir = "";
115+
116+
beforeEach(async () => {
117+
stateDir = await fs.mkdtemp(path.join(os.tmpdir(), "openclaw-session-startup-"));
118+
vi.stubEnv("OPENCLAW_STATE_DIR", stateDir);
119+
});
120+
121+
afterEach(async () => {
122+
vi.unstubAllEnvs();
123+
await fs.rm(stateDir, { recursive: true, force: true });
124+
});
125+
126+
async function writeSessionFile(
127+
name: string,
128+
): Promise<{ filePath: string; size: number; mtimeMs: number }> {
129+
const sessionsDir = resolveSessionTranscriptsDirForAgent("main");
130+
await fs.mkdir(sessionsDir, { recursive: true });
131+
const filePath = path.join(sessionsDir, name);
132+
await fs.writeFile(
133+
filePath,
134+
JSON.stringify({ type: "message", message: { role: "user", content: "startup catchup" } }) +
135+
"\n",
136+
"utf-8",
137+
);
138+
const stat = await fs.stat(filePath);
139+
return { filePath, size: stat.size, mtimeMs: stat.mtimeMs };
140+
}
141+
142+
it("marks stale indexed session files dirty and schedules catch-up sync", async () => {
143+
const session = await writeSessionFile("thread.jsonl");
144+
const harness = new SessionStartupCatchupHarness([
145+
{
146+
path: "sessions/main/thread.jsonl",
147+
hash: "old-hash",
148+
mtime: session.mtimeMs - 1000,
149+
size: session.size,
150+
},
151+
]);
152+
153+
await expect(harness.catchUp()).resolves.toEqual([session.filePath]);
154+
expect(harness.getDirtySessionFiles()).toEqual([session.filePath]);
155+
expect(harness.isSessionsDirty()).toBe(true);
156+
expect(harness.syncCalls).toEqual([{ reason: "session-startup-catchup" }]);
157+
});
158+
159+
it("leaves unchanged indexed session files clean", async () => {
160+
const session = await writeSessionFile("thread.jsonl");
161+
const harness = new SessionStartupCatchupHarness([
162+
{
163+
path: "sessions/main/thread.jsonl",
164+
hash: "current-hash",
165+
mtime: session.mtimeMs,
166+
size: session.size,
167+
},
168+
]);
169+
170+
await expect(harness.catchUp()).resolves.toEqual([]);
171+
expect(harness.getDirtySessionFiles()).toEqual([]);
172+
expect(harness.isSessionsDirty()).toBe(false);
173+
expect(harness.syncCalls).toEqual([]);
174+
});
175+
});

‎extensions/memory-core/src/memory/manager-sync-ops.ts‎

Lines changed: 64 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -53,7 +53,11 @@ import {
5353
type MemoryIndexMeta,
5454
} from "./manager-reindex-state.js";
5555
import { shouldSyncSessionsForReindex } from "./manager-session-reindex.js";
56-
import { resolveMemorySessionSyncPlan } from "./manager-session-sync-state.js";
56+
import {
57+
resolveMemorySessionStartupDirtyFiles,
58+
resolveMemorySessionSyncPlan,
59+
type MemorySessionStartupFileState,
60+
} from "./manager-session-sync-state.js";
5761
import {
5862
loadMemorySourceFileState,
5963
resolveMemorySourceExistingHash,
@@ -491,6 +495,65 @@ export abstract class MemoryManagerSyncOps {
491495
});
492496
}
493497

498+
protected ensureSessionStartupCatchup(): void {
499+
if (!this.sources.has("sessions")) {
500+
return;
501+
}
502+
void this.runSessionStartupCatchup().catch((err) => {
503+
log.warn("memory session startup catch-up failed: " + String(err));
504+
});
505+
}
506+
507+
protected async runSessionStartupCatchup(): Promise<string[]> {
508+
if (!this.sources.has("sessions") || this.closed) {
509+
return [];
510+
}
511+
const files = await listSessionFilesForAgent(this.agentId);
512+
if (files.length === 0 || this.closed) {
513+
return [];
514+
}
515+
const existingRows = loadMemorySourceFileState({
516+
db: this.db,
517+
source: "sessions",
518+
}).rows;
519+
const fileStates = (
520+
await runWithConcurrency(
521+
files.map((file) => async (): Promise<MemorySessionStartupFileState | null> => {
522+
try {
523+
const stat = await fs.stat(file);
524+
if (!stat.isFile()) {
525+
return null;
526+
}
527+
return {
528+
absPath: file,
529+
path: sessionPathForFile(file),
530+
mtimeMs: stat.mtimeMs,
531+
size: stat.size,
532+
};
533+
} catch (err) {
534+
if (isFileMissingError(err)) {
535+
return null;
536+
}
537+
throw err;
538+
}
539+
}),
540+
this.getIndexConcurrency(),
541+
)
542+
).filter((file): file is MemorySessionStartupFileState => file !== null);
543+
const dirtyFiles = resolveMemorySessionStartupDirtyFiles({ files: fileStates, existingRows });
544+
if (dirtyFiles.length === 0 || this.closed) {
545+
return dirtyFiles;
546+
}
547+
for (const file of dirtyFiles) {
548+
this.sessionsDirtyFiles.add(file);
549+
}
550+
this.sessionsDirty = true;
551+
void this.sync({ reason: "session-startup-catchup" }).catch((err) => {
552+
log.warn("memory sync failed (session-startup-catchup): " + String(err));
553+
});
554+
return dirtyFiles;
555+
}
556+
494557
private scheduleSessionDirty(sessionFile: string) {
495558
this.sessionPendingFiles.add(sessionFile);
496559
if (this.sessionWatchTimer) {

‎extensions/memory-core/src/memory/manager.ts‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -246,6 +246,9 @@ export class MemoryIndexManager extends MemoryManagerEmbeddingOps implements Mem
246246
hasIndexedMeta: Boolean(meta),
247247
});
248248
this.batch = this.resolveBatchConfig();
249+
if (!transient) {
250+
this.ensureSessionStartupCatchup();
251+
}
249252
}
250253

251254
private applyProviderResult(providerResult: EmbeddingProviderResult): void {

0 commit comments

Comments
 (0)