Skip to content

Commit b2ce1bc

Browse files
jalehmanjalehman
committed
clawdbot-89e: add memory session sync identity api
1 parent 8a87609 commit b2ce1bc

18 files changed

Lines changed: 541 additions & 111 deletions

‎docs/.generated/plugin-sdk-api-baseline.sha256‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,2 +1,2 @@
1-
d522f8860146243ff1e7fd0e4b7b89bce6be0c78ab06c564d25c204bdb93287b plugin-sdk-api-baseline.json
2-
62d3c6a2f7bdc01c196a970cc269bb83afac34db27be3d8951edb1bbbbff8eaf plugin-sdk-api-baseline.jsonl
1+
944ca9fb6d46b8a3fa5582fc276478adfecdb3125d8854523492a7ac155ee318 plugin-sdk-api-baseline.json
2+
4b79e9cdc7feadb8bcaa89c31160e445141894556ec03652232c3e6a1948ce50 plugin-sdk-api-baseline.jsonl

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

Lines changed: 6 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -1,17 +1,18 @@
1+
import type { MemorySyncParams } from "openclaw/plugin-sdk/memory-core-host-engine-storage";
2+
13
export function shouldSyncSessionsForReindex(params: {
24
hasSessionSource: boolean;
35
sessionsDirty: boolean;
46
dirtySessionFileCount: number;
5-
sync?: {
6-
reason?: string;
7-
force?: boolean;
8-
sessionFiles?: string[];
9-
};
7+
sync?: MemorySyncParams;
108
needsFullReindex?: boolean;
119
}): boolean {
1210
if (!params.hasSessionSource) {
1311
return false;
1412
}
13+
if (params.sync?.sessions?.some((session) => session.sessionId.trim().length > 0)) {
14+
return true;
15+
}
1516
if (params.sync?.sessionFiles?.some((sessionFile) => sessionFile.trim().length > 0)) {
1617
return true;
1718
}

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

Lines changed: 13 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -65,6 +65,19 @@ describe("memory session sync state", () => {
6565
expect(plan.activePaths).toEqual(new Set(["sessions/incremental.jsonl"]));
6666
});
6767

68+
it("marks identity-targeted syncs as session work", async () => {
69+
const { shouldSyncSessionsForReindex } = await import("./manager-session-reindex.js");
70+
71+
expect(
72+
shouldSyncSessionsForReindex({
73+
hasSessionSource: true,
74+
sessionsDirty: false,
75+
dirtySessionFileCount: 0,
76+
sync: { sessions: [{ agentId: "main", sessionId: "targeted" }] },
77+
}),
78+
).toBe(true);
79+
});
80+
6881
it("marks missing and changed startup session files dirty", () => {
6982
const dirtyFiles = resolveMemorySessionStartupDirtyFiles({
7083
files: [

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

Lines changed: 47 additions & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -4,7 +4,11 @@ import {
44
createSubsystemLogger,
55
type OpenClawConfig,
66
} from "openclaw/plugin-sdk/memory-core-host-engine-foundation";
7-
import type { MemorySyncProgressUpdate } from "openclaw/plugin-sdk/memory-core-host-engine-storage";
7+
import type {
8+
MemorySessionSyncTarget,
9+
MemorySyncParams,
10+
MemorySyncProgressUpdate,
11+
} from "openclaw/plugin-sdk/memory-core-host-engine-storage";
812

913
const log = createSubsystemLogger("memory");
1014

@@ -21,6 +25,7 @@ export type MemoryReadonlyRecoveryState = {
2125
runSync: (params?: {
2226
reason?: string;
2327
force?: boolean;
28+
sessions?: MemorySessionSyncTarget[];
2429
sessionFiles?: string[];
2530
progress?: (update: MemorySyncProgressUpdate) => void;
2631
}) => Promise<void>;
@@ -82,12 +87,7 @@ export function extractMemoryErrorReason(err: unknown): string {
8287

8388
export async function runMemorySyncWithReadonlyRecovery(
8489
state: MemoryReadonlyRecoveryState,
85-
params?: {
86-
reason?: string;
87-
force?: boolean;
88-
sessionFiles?: string[];
89-
progress?: (update: MemorySyncProgressUpdate) => void;
90-
},
90+
params?: MemorySyncParams,
9191
): Promise<void> {
9292
try {
9393
await state.runSync(params);
@@ -123,37 +123,46 @@ export function enqueueMemoryTargetedSessionSync(
123123
isClosed: () => boolean;
124124
getSyncing: () => Promise<void> | null;
125125
getQueuedSessionFiles: () => Set<string>;
126+
getQueuedSessions: () => Map<string, MemorySessionSyncTarget>;
126127
getQueuedSessionSync: () => Promise<void> | null;
127128
setQueuedSessionSync: (value: Promise<void> | null) => void;
128-
sync: (params?: {
129-
reason?: string;
130-
force?: boolean;
131-
sessionFiles?: string[];
132-
progress?: (update: MemorySyncProgressUpdate) => void;
133-
}) => Promise<void>;
129+
sync: (params?: MemorySyncParams) => Promise<void>;
134130
},
135-
sessionFiles?: string[],
131+
targets?: Pick<MemorySyncParams, "sessions" | "sessionFiles">,
136132
): Promise<void> {
137133
const queuedSessionFiles = state.getQueuedSessionFiles();
138-
for (const sessionFile of sessionFiles ?? []) {
134+
for (const sessionFile of targets?.sessionFiles ?? []) {
139135
const trimmed = sessionFile.trim();
140136
if (trimmed) {
141137
queuedSessionFiles.add(trimmed);
142138
}
143139
}
144-
if (queuedSessionFiles.size === 0) {
140+
const queuedSessions = state.getQueuedSessions();
141+
for (const session of targets?.sessions ?? []) {
142+
const normalized = normalizeQueuedMemorySessionSyncTarget(session);
143+
if (normalized) {
144+
queuedSessions.set(memorySessionSyncTargetKey(normalized), normalized);
145+
}
146+
}
147+
if (queuedSessionFiles.size === 0 && queuedSessions.size === 0) {
145148
return state.getSyncing() ?? Promise.resolve();
146149
}
147150
if (!state.getQueuedSessionSync()) {
148151
state.setQueuedSessionSync(
149152
(async () => {
150153
try {
151154
await state.getSyncing()?.catch(() => undefined);
152-
while (!state.isClosed() && state.getQueuedSessionFiles().size > 0) {
155+
while (
156+
!state.isClosed() &&
157+
(state.getQueuedSessionFiles().size > 0 || state.getQueuedSessions().size > 0)
158+
) {
153159
const pendingSessionFiles = Array.from(state.getQueuedSessionFiles());
160+
const pendingSessions = Array.from(state.getQueuedSessions().values());
154161
state.getQueuedSessionFiles().clear();
162+
state.getQueuedSessions().clear();
155163
await state.sync({
156-
reason: "queued-session-files",
164+
reason: "queued-sessions",
165+
sessions: pendingSessions,
157166
sessionFiles: pendingSessionFiles,
158167
});
159168
}
@@ -166,6 +175,26 @@ export function enqueueMemoryTargetedSessionSync(
166175
return state.getQueuedSessionSync() ?? Promise.resolve();
167176
}
168177

178+
function normalizeQueuedMemorySessionSyncTarget(
179+
target: MemorySessionSyncTarget,
180+
): MemorySessionSyncTarget | null {
181+
const sessionId = target.sessionId.trim();
182+
if (!sessionId) {
183+
return null;
184+
}
185+
const agentId = target.agentId?.trim();
186+
const sessionKey = target.sessionKey?.trim();
187+
return {
188+
...(agentId ? { agentId } : {}),
189+
sessionId,
190+
...(sessionKey ? { sessionKey } : {}),
191+
};
192+
}
193+
194+
function memorySessionSyncTargetKey(target: MemorySessionSyncTarget): string {
195+
return [target.agentId ?? "", target.sessionId, target.sessionKey ?? ""].join("\0");
196+
}
197+
169198
export function createMemorySyncControlConfigForTests(
170199
workspaceDir: string,
171200
indexPath: string,

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

Lines changed: 55 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -9,6 +9,7 @@ import {
99
} from "openclaw/plugin-sdk/memory-core-host-engine-foundation";
1010
import type {
1111
MemorySource,
12+
MemorySyncParams,
1213
MemorySyncProgressUpdate,
1314
} from "openclaw/plugin-sdk/memory-core-host-engine-storage";
1415
import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
@@ -26,6 +27,7 @@ type MemoryIndexEntry = {
2627
type SyncParams = {
2728
reason?: string;
2829
force?: boolean;
30+
sessions?: MemorySyncParams["sessions"];
2931
sessionFiles?: string[];
3032
progress?: (update: MemorySyncProgressUpdate) => void;
3133
};
@@ -37,14 +39,33 @@ class SessionStartupCatchupHarness extends MemoryManagerSyncOps {
3739
protected readonly agentId = "main";
3840
protected readonly workspaceDir = "/tmp/openclaw-test-workspace";
3941
protected readonly settings = {
42+
chunking: {
43+
overlap: 0,
44+
tokens: 256,
45+
},
46+
extraPaths: [],
47+
multimodal: {
48+
enabled: false,
49+
modalities: [],
50+
maxFileBytes: 0,
51+
},
52+
provider: "none",
53+
store: {
54+
fts: {
55+
tokenizer: "unicode61",
56+
},
57+
vector: {
58+
enabled: false,
59+
},
60+
},
4061
sync: {
4162
sessions: {
4263
deltaBytes: 100_000,
4364
deltaMessages: 50,
4465
postCompactionForce: true,
4566
},
4667
},
47-
} as ResolvedMemorySearchConfig;
68+
} as unknown as ResolvedMemorySearchConfig;
4869
protected readonly batch = {
4970
enabled: false,
5071
wait: false,
@@ -59,6 +80,7 @@ class SessionStartupCatchupHarness extends MemoryManagerSyncOps {
5980
protected db: DatabaseSync;
6081

6182
readonly syncCalls: SyncParams[] = [];
83+
readonly indexedPaths: string[] = [];
6284

6385
constructor(sourceRows: SourceStateRow[]) {
6486
super();
@@ -80,6 +102,10 @@ class SessionStartupCatchupHarness extends MemoryManagerSyncOps {
80102
return await this.markSessionStartupCatchupDirtyFiles();
81103
}
82104

105+
async runSyncForTest(params?: MemorySyncParams): Promise<void> {
106+
await this.runSync(params);
107+
}
108+
83109
getDirtySessionFiles(): string[] {
84110
return Array.from(this.sessionsDirtyFiles);
85111
}
@@ -113,9 +139,11 @@ class SessionStartupCatchupHarness extends MemoryManagerSyncOps {
113139
protected resetProviderInitializationForRetry(): void {}
114140

115141
protected async indexFile(
116-
_entry: MemoryIndexEntry,
142+
entry: MemoryIndexEntry,
117143
_options: { source: MemorySource; content?: string },
118-
): Promise<void> {}
144+
): Promise<void> {
145+
this.indexedPaths.push(entry.path);
146+
}
119147
}
120148

121149
describe("session startup catch-up", () => {
@@ -197,4 +225,28 @@ describe("session startup catch-up", () => {
197225
expect(harness.isSessionsDirty()).toBe(false);
198226
expect(harness.syncCalls).toEqual([]);
199227
});
228+
229+
it("does not fall back to full session sync when identity targets normalize away", async () => {
230+
await writeSessionFile("thread.jsonl");
231+
const harness = new SessionStartupCatchupHarness([]);
232+
233+
await harness.runSyncForTest({
234+
reason: "queued-sessions",
235+
sessions: [{ agentId: "other", sessionId: "thread" }],
236+
});
237+
238+
expect(harness.indexedPaths).toEqual([]);
239+
});
240+
241+
it("does not fall back to full session sync for malformed identity session ids", async () => {
242+
await writeSessionFile("thread.jsonl");
243+
const harness = new SessionStartupCatchupHarness([]);
244+
245+
await harness.runSyncForTest({
246+
reason: "queued-sessions",
247+
sessions: [{ agentId: "main", sessionId: "bad/nested" }],
248+
});
249+
250+
expect(harness.indexedPaths).toEqual([]);
251+
});
200252
});

0 commit comments

Comments
 (0)