diff --git a/extensions/memory-core/src/memory/manager-session-sync-ops.ts b/extensions/memory-core/src/memory/manager-session-sync-ops.ts index 58f66e42487f..aa1c7b4abd52 100644 --- a/extensions/memory-core/src/memory/manager-session-sync-ops.ts +++ b/extensions/memory-core/src/memory/manager-session-sync-ops.ts @@ -230,7 +230,13 @@ export abstract class MemoryManagerSessionSyncOps extends MemoryManagerWatchOps } if (pending.length > 0) { this.sessionsDirty = true; - void this.sync({ reason: "session-delta" }).catch((err: unknown) => { + // Keep both identity and file keys so every transcript backend enters the + // targeted queue instead of letting an active sync clear this newer event. + void this.sync({ + reason: "session-delta", + sessions: pendingTargets, + archiveFiles: pending, + }).catch((err: unknown) => { log.warn(`memory sync failed (session update): ${String(err)}`); }); } diff --git a/extensions/memory-core/src/memory/manager-session-update-race.test.ts b/extensions/memory-core/src/memory/manager-session-update-race.test.ts new file mode 100644 index 000000000000..fff736b24ab1 --- /dev/null +++ b/extensions/memory-core/src/memory/manager-session-update-race.test.ts @@ -0,0 +1,99 @@ +import path from "node:path"; +import { DatabaseSync } from "node:sqlite"; +import type { MemorySessionSyncTarget } from "openclaw/plugin-sdk/memory-core-host-engine-storage"; +import { resolveOpenClawAgentSqlitePath } from "openclaw/plugin-sdk/sqlite-runtime"; +import { describe, expect, it, vi } from "vitest"; +import { createManagerIndexFixture } from "./manager-index.test-support.js"; + +const { closeAllMemorySearchManagers, getMemorySearchManager } = await import("./index.js"); + +describe("memory session update sync", () => { + const fixture = createManagerIndexFixture({ + getMemorySearchManager, + closeAllMemorySearchManagers, + }); + const { createConfig, getFreshManager, seedSessionTranscript } = fixture; + + it("indexes an update that arrives before an active sync clears dirty state", async () => { + fixture.setStateDir(path.join(fixture.paths.workspace, ".state-session-update-during-sync")); + const sessionId = "session-update-during-sync"; + const sessionKey = `agent:main:proof:${sessionId}`; + const updatedMarker = "UPDATE DURING ACTIVE SYNC 811"; + const manager = await getFreshManager( + createConfig({ provider: "none", sources: ["sessions"], sessionMemory: true }), + "cli", + ); + const owner = manager as unknown as { + queuedSessionSync: Promise | null; + sessionPendingTargets: Map; + sessionsDirty: boolean; + sessionsReconcileDirty: boolean; + syncArchiveFiles: (params: unknown) => Promise; + processSessionUpdateBatch: () => Promise; + }; + let releaseActiveSync = () => {}; + const activeSyncGate = new Promise((resolve) => { + releaseActiveSync = resolve; + }); + let markActiveSyncIndexed = () => {}; + const activeSyncIndexed = new Promise((resolve) => { + markActiveSyncIndexed = resolve; + }); + let syncArchiveFilesSpy: { mockRestore: () => void } | undefined; + try { + await seedSessionTranscript({ + sessionId, + sessionKey, + messages: [{ role: "user", timestamp: Date.now(), content: "initial transcript" }], + }); + await manager.sync({ reason: "test-baseline", force: true }); + + owner.sessionsDirty = true; + owner.sessionsReconcileDirty = true; + const syncArchiveFiles = owner.syncArchiveFiles.bind(manager); + syncArchiveFilesSpy = vi + .spyOn(owner, "syncArchiveFiles") + .mockImplementationOnce(async (params) => { + await syncArchiveFiles(params); + markActiveSyncIndexed(); + await activeSyncGate; + }); + const activeSync = manager.sync({ reason: "test-active" }); + await activeSyncIndexed; + + await seedSessionTranscript({ + sessionId, + sessionKey, + messages: [{ role: "assistant", timestamp: Date.now(), content: updatedMarker }], + }); + owner.sessionPendingTargets.set(sessionKey, { agentId: "main", sessionId, sessionKey }); + await owner.processSessionUpdateBatch(); + const queuedSessionSync = owner.queuedSessionSync; + expect(queuedSessionSync).not.toBeNull(); + + releaseActiveSync(); + await activeSync; + await queuedSessionSync; + + const observer = new DatabaseSync(resolveOpenClawAgentSqlitePath({ agentId: "main" }), { + readOnly: true, + }); + try { + const row = observer + .prepare( + "SELECT COUNT(*) AS count FROM memory_index_chunks WHERE source = 'sessions' AND text LIKE ?", + ) + .get(`%${updatedMarker}%`) as { count: number }; + expect(row.count).toBeGreaterThan(0); + } finally { + observer.close(); + } + expect(manager.status().dirty).toBe(false); + } finally { + syncArchiveFilesSpy?.mockRestore(); + releaseActiveSync(); + await manager.close?.(); + fixture.restoreStateDir(); + } + }); +}); diff --git a/extensions/memory-core/src/memory/manager-sync-ops.startup-catchup.test.ts b/extensions/memory-core/src/memory/manager-sync-ops.startup-catchup.test.ts index 3b4bbb853f07..6fa4607b57a4 100644 --- a/extensions/memory-core/src/memory/manager-sync-ops.startup-catchup.test.ts +++ b/extensions/memory-core/src/memory/manager-sync-ops.startup-catchup.test.ts @@ -880,7 +880,8 @@ describe("session startup catch-up", () => { await Promise.resolve(); expect(harness.getDirtyArchiveFiles()).toEqual([session.sessionKey]); - expect(harness.syncCalls).toEqual([{ reason: "session-delta" }]); + expect(harness.syncCalls[0]?.archiveFiles).toEqual([session.sessionKey]); + expect(harness.syncCalls[0]?.sessions).toHaveLength(1); }); it("keeps targeted indexing on the SQLite store resolved by its corpus snapshot", async () => { @@ -1063,7 +1064,7 @@ describe("session startup catch-up", () => { await harness.waitForSessionSync(); expect(harness.getDirtyArchiveFiles()).toEqual([session.filePath]); - expect(harness.syncCalls).toEqual([{ reason: "session-delta" }]); + expect(harness.syncCalls[0]?.archiveFiles).toEqual([session.filePath]); expect(harness.indexedPaths).toEqual([ `sessions/main/thread.jsonl.${reason}.2026-06-23T10-00-00.000Z`, ]);