Skip to content

Commit c24d266

Browse files
jalehmanjalehman
authored
refactor: use accessor-backed transcript corpus for memory (#96162)
* refactor: ratchet memory transcript corpus access * test: use narrow runtime config snapshot import * test: update plugin sdk surface budgets * refactor: split memory transcript corpus module
1 parent 9405b8f commit c24d266

14 files changed

Lines changed: 1370 additions & 56 deletions

‎extensions/memory-core/src/dreaming-phases.ts‎

Lines changed: 5 additions & 16 deletions
Original file line numberDiff line numberDiff line change
@@ -5,9 +5,7 @@ import fs from "node:fs/promises";
55
import path from "node:path";
66
import {
77
buildSessionEntry,
8-
listSessionFilesForAgent,
9-
loadSessionTranscriptClassificationForAgent,
10-
normalizeSessionTranscriptPathForComparison,
8+
listSessionTranscriptCorpusEntriesForAgent,
119
parseUsageCountedSessionIdFromFileName,
1210
sessionPathForFile,
1311
} from "openclaw/plugin-sdk/memory-core-host-engine-qmd";
@@ -848,25 +846,16 @@ async function collectSessionIngestionBatches(params: {
848846
sessionPath: string;
849847
}> = [];
850848
for (const agentId of agentIds) {
851-
const files = await listSessionFilesForAgent(agentId);
852-
const transcriptClassification =
853-
files.length > 0
854-
? loadSessionTranscriptClassificationForAgent(agentId)
855-
: {
856-
dreamingNarrativeTranscriptPaths: new Set<string>(),
857-
cronRunTranscriptPaths: new Set<string>(),
858-
};
859-
for (const absolutePath of files) {
849+
for (const entry of await listSessionTranscriptCorpusEntriesForAgent(agentId)) {
850+
const absolutePath = entry.sessionFile;
860851
if (isCheckpointSessionTranscriptPath(absolutePath)) {
861852
continue;
862853
}
863-
const normalizedPath = normalizeSessionTranscriptPathForComparison(absolutePath);
864854
sessionFiles.push({
865855
agentId,
866856
absolutePath,
867-
generatedByDreamingNarrative:
868-
transcriptClassification.dreamingNarrativeTranscriptPaths.has(normalizedPath),
869-
generatedByCronRun: transcriptClassification.cronRunTranscriptPaths.has(normalizedPath),
857+
generatedByDreamingNarrative: entry.generatedByDreamingNarrative === true,
858+
generatedByCronRun: entry.generatedByCronRun === true,
870859
sessionPath: sessionPathForFile(absolutePath),
871860
});
872861
}

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

Lines changed: 222 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -14,6 +14,10 @@ import type {
1414
MemorySyncParams,
1515
MemorySyncProgressUpdate,
1616
} from "openclaw/plugin-sdk/memory-core-host-engine-storage";
17+
import {
18+
clearConfigCache,
19+
clearRuntimeConfigSnapshot,
20+
} from "openclaw/plugin-sdk/runtime-config-snapshot";
1721
import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
1822
import { MemoryManagerSyncOps } from "./manager-sync-ops.js";
1923

@@ -102,6 +106,7 @@ class SessionStartupCatchupHarness extends MemoryManagerSyncOps {
102106

103107
readonly syncCalls: SyncParams[] = [];
104108
readonly indexedPaths: string[] = [];
109+
readonly indexedContents: string[] = [];
105110

106111
constructor(sourceRows: SourceStateRow[]) {
107112
super();
@@ -139,6 +144,35 @@ class SessionStartupCatchupHarness extends MemoryManagerSyncOps {
139144
return Array.from(this.sessionPendingFiles);
140145
}
141146

147+
addPendingSessionTarget(target: NonNullable<MemorySyncParams["sessions"]>[number]): void {
148+
this.sessionPendingTargets.set(
149+
[target.agentId ?? "", target.sessionId, target.sessionKey ?? ""].join("\0"),
150+
target,
151+
);
152+
}
153+
154+
async processPendingSessionDeltas(): Promise<void> {
155+
await (
156+
this as unknown as {
157+
processSessionDeltaBatch: () => Promise<void>;
158+
}
159+
).processSessionDeltaBatch();
160+
}
161+
162+
async combineTargetSessionFilesForTest(params: {
163+
sessions?: MemorySyncParams["sessions"];
164+
sessionFiles?: string[];
165+
}): Promise<Set<string> | null> {
166+
return await (
167+
this as unknown as {
168+
combineTargetSessionFiles: (params: {
169+
sessions?: MemorySyncParams["sessions"];
170+
sessionFiles?: string[];
171+
}) => Promise<Set<string> | null>;
172+
}
173+
).combineTargetSessionFiles(params);
174+
}
175+
142176
isSessionsDirty(): boolean {
143177
return this.sessionsDirty;
144178
}
@@ -184,9 +218,10 @@ class SessionStartupCatchupHarness extends MemoryManagerSyncOps {
184218

185219
protected async indexFile(
186220
entry: MemoryIndexEntry,
187-
_options: { source: MemorySource; content?: string },
221+
options: { source: MemorySource; content?: string },
188222
): Promise<void> {
189223
this.indexedPaths.push(entry.path);
224+
this.indexedContents.push(options.content ?? "");
190225
}
191226
}
192227

@@ -202,6 +237,8 @@ describe("session startup catch-up", () => {
202237
vi.clearAllTimers();
203238
vi.useRealTimers();
204239
vi.unstubAllEnvs();
240+
clearRuntimeConfigSnapshot();
241+
clearConfigCache();
205242
await fs.rm(stateDir, { recursive: true, force: true });
206243
});
207244

@@ -396,6 +433,147 @@ describe("session startup catch-up", () => {
396433
expect(harness.indexedPaths).toEqual([]);
397434
});
398435

436+
it("resolves identity-targeted delta sync through a custom session store", async () => {
437+
const storeDir = path.join(stateDir, "custom-sessions");
438+
const sessionFile = path.join(storeDir, "custom-thread.jsonl");
439+
const storePath = path.join(storeDir, "sessions.json");
440+
const configPath = path.join(stateDir, "openclaw.json");
441+
await fs.mkdir(storeDir, { recursive: true });
442+
await fs.writeFile(
443+
sessionFile,
444+
JSON.stringify({
445+
type: "message",
446+
message: { role: "user", content: "custom store target" },
447+
}) + "\n",
448+
"utf-8",
449+
);
450+
await fs.writeFile(
451+
storePath,
452+
JSON.stringify({
453+
"agent:main:chat:custom": {
454+
sessionFile: "custom-thread.jsonl",
455+
sessionId: "custom-thread",
456+
},
457+
}),
458+
"utf-8",
459+
);
460+
await fs.writeFile(configPath, JSON.stringify({ session: { store: storePath } }), "utf-8");
461+
vi.stubEnv("OPENCLAW_CONFIG_PATH", configPath);
462+
clearRuntimeConfigSnapshot();
463+
clearConfigCache();
464+
const harness = new SessionStartupCatchupHarness([]);
465+
(harness as unknown as { settings: ResolvedMemorySearchConfig }).settings.sync.sessions = {
466+
deltaBytes: 1,
467+
deltaMessages: 1,
468+
postCompactionForce: true,
469+
};
470+
harness.addPendingSessionTarget({
471+
agentId: "main",
472+
sessionId: "custom-thread",
473+
sessionKey: "agent:main:chat:custom",
474+
});
475+
476+
await harness.processPendingSessionDeltas();
477+
await Promise.resolve();
478+
479+
expect(harness.getDirtySessionFiles()).toEqual([sessionFile]);
480+
expect(harness.syncCalls).toEqual([{ reason: "session-delta" }]);
481+
});
482+
483+
it("keeps explicit custom-store session file targets at the sync gate", async () => {
484+
const storeDir = path.join(stateDir, "custom-sessions");
485+
const sessionFile = path.join(storeDir, "explicit-target.jsonl");
486+
const storePath = path.join(storeDir, "sessions.json");
487+
const configPath = path.join(stateDir, "openclaw.json");
488+
await fs.mkdir(storeDir, { recursive: true });
489+
await fs.writeFile(
490+
sessionFile,
491+
JSON.stringify({
492+
type: "message",
493+
message: { role: "user", content: "explicit target" },
494+
}) + "\n",
495+
"utf-8",
496+
);
497+
await fs.writeFile(
498+
storePath,
499+
JSON.stringify({
500+
"agent:main:chat:explicit-target": {
501+
sessionFile: "explicit-target.jsonl",
502+
sessionId: "explicit-target",
503+
},
504+
}),
505+
"utf-8",
506+
);
507+
await fs.writeFile(configPath, JSON.stringify({ session: { store: storePath } }), "utf-8");
508+
vi.stubEnv("OPENCLAW_CONFIG_PATH", configPath);
509+
clearRuntimeConfigSnapshot();
510+
clearConfigCache();
511+
const harness = new SessionStartupCatchupHarness([]);
512+
513+
await expect(
514+
harness.combineTargetSessionFilesForTest({ sessionFiles: [sessionFile] }),
515+
).resolves.toEqual(new Set([sessionFile]));
516+
});
517+
518+
it("preserves generated-session classification during targeted custom-store indexing", async () => {
519+
const storeDir = path.join(stateDir, "custom-sessions");
520+
const sessionFile = path.join(storeDir, "cron-thread.jsonl");
521+
const otherSessionFile = path.join(storeDir, "other-thread.jsonl");
522+
const storePath = path.join(storeDir, "sessions.json");
523+
const configPath = path.join(stateDir, "openclaw.json");
524+
await fs.mkdir(storeDir, { recursive: true });
525+
await fs.writeFile(
526+
sessionFile,
527+
JSON.stringify({
528+
type: "message",
529+
message: { role: "assistant", content: "Internal cron output that must stay out." },
530+
}) + "\n",
531+
"utf-8",
532+
);
533+
await fs.writeFile(
534+
otherSessionFile,
535+
JSON.stringify({
536+
type: "message",
537+
message: { role: "user", content: "Other custom-store content" },
538+
}) + "\n",
539+
"utf-8",
540+
);
541+
await fs.writeFile(
542+
storePath,
543+
JSON.stringify({
544+
"agent:main:cron:job-1:run:run-1": {
545+
sessionFile: "cron-thread.jsonl",
546+
sessionId: "cron-thread",
547+
},
548+
"agent:main:chat:other": {
549+
sessionFile: "other-thread.jsonl",
550+
sessionId: "other-thread",
551+
},
552+
}),
553+
"utf-8",
554+
);
555+
await fs.writeFile(configPath, JSON.stringify({ session: { store: storePath } }), "utf-8");
556+
vi.stubEnv("OPENCLAW_CONFIG_PATH", configPath);
557+
clearRuntimeConfigSnapshot();
558+
clearConfigCache();
559+
const harness = new SessionStartupCatchupHarness([]);
560+
561+
await (
562+
harness as unknown as {
563+
syncSessionFiles: (params: {
564+
needsFullReindex: boolean;
565+
targetSessionFiles: string[];
566+
}) => Promise<void>;
567+
}
568+
).syncSessionFiles({
569+
needsFullReindex: false,
570+
targetSessionFiles: [sessionFile],
571+
});
572+
573+
expect(harness.indexedPaths).toEqual(["sessions/cron-thread.jsonl"]);
574+
expect(harness.indexedContents).toEqual([""]);
575+
});
576+
399577
it("queues transcript update identity without requiring a session file", async () => {
400578
vi.useFakeTimers();
401579
const harness = new SessionStartupCatchupHarness([]);
@@ -456,6 +634,49 @@ describe("session startup catch-up", () => {
456634
harness.stopTranscriptListener();
457635
});
458636

637+
it("queues file-only transcript updates from a custom session store", async () => {
638+
vi.useFakeTimers();
639+
const storeDir = path.join(stateDir, "custom-sessions");
640+
const sessionFile = path.join(storeDir, "custom-update.jsonl");
641+
const storePath = path.join(storeDir, "sessions.json");
642+
const configPath = path.join(stateDir, "openclaw.json");
643+
await fs.mkdir(storeDir, { recursive: true });
644+
await fs.writeFile(
645+
sessionFile,
646+
JSON.stringify({
647+
type: "message",
648+
message: { role: "user", content: "custom update" },
649+
}) + "\n",
650+
"utf-8",
651+
);
652+
await fs.writeFile(
653+
storePath,
654+
JSON.stringify({
655+
"agent:main:chat:custom-update": {
656+
sessionFile: "custom-update.jsonl",
657+
sessionId: "custom-update",
658+
},
659+
}),
660+
"utf-8",
661+
);
662+
await fs.writeFile(configPath, JSON.stringify({ session: { store: storePath } }), "utf-8");
663+
vi.stubEnv("OPENCLAW_CONFIG_PATH", configPath);
664+
clearRuntimeConfigSnapshot();
665+
clearConfigCache();
666+
const harness = new SessionStartupCatchupHarness([]);
667+
harness.startTranscriptListener();
668+
669+
emitSessionTranscriptUpdate({
670+
sessionFile,
671+
sessionKey: "agent:main:chat:custom-update",
672+
});
673+
await Promise.resolve();
674+
675+
expect(harness.getPendingSessionFiles()).toEqual([sessionFile]);
676+
expect(harness.getPendingSessionTargets()).toEqual([]);
677+
harness.stopTranscriptListener();
678+
});
679+
459680
it("prefers transcript update path compatibility before identity", async () => {
460681
vi.useFakeTimers();
461682
const session = await writeSessionFile("thread.jsonl");

0 commit comments

Comments
 (0)