From 51be45da31e9eac60eea13e7b4feccf0ac76a496 Mon Sep 17 00:00:00 2001 From: clawsweeper <274271284+clawsweeper[bot]@users.noreply.github.com> Date: Tue, 4 Aug 2026 23:00:03 +0000 Subject: [PATCH] fix(memory-core): resume validated session transcript appends --- .../src/dreaming-ingestion-state.ts | 2 + .../memory-core/src/dreaming-phases.test.ts | 153 +++++++++++++++++- .../memory-core/src/session-backfill.ts | 2 + .../memory-core/src/session-ingestion.test.ts | 55 ++++++- .../memory-core/src/session-ingestion.ts | 13 +- .../memory-host-sdk/src/engine-sessions.ts | 1 + .../host/session-files-reset-revision.test.ts | 43 ++++- .../memory-host-sdk/src/host/session-files.ts | 82 ++++++++-- .../memory-core-host-engine-sessions.ts | 1 + 9 files changed, 322 insertions(+), 30 deletions(-) diff --git a/extensions/memory-core/src/dreaming-ingestion-state.ts b/extensions/memory-core/src/dreaming-ingestion-state.ts index 0d2ad9dcce12..67706ebd58d2 100644 --- a/extensions/memory-core/src/dreaming-ingestion-state.ts +++ b/extensions/memory-core/src/dreaming-ingestion-state.ts @@ -22,8 +22,10 @@ export type DailyIngestionState = { export type SessionIngestionFileState = { mtimeMs: number; size: number; + /** Canonical hash of the full exported snapshot described by lineCount. */ contentHash: string; lineCount: number; + /** Consumption cursor within that snapshot; it may trail lineCount. */ lastContentLine: number; }; diff --git a/extensions/memory-core/src/dreaming-phases.test.ts b/extensions/memory-core/src/dreaming-phases.test.ts index ca59c3c55268..10ab3be26a76 100644 --- a/extensions/memory-core/src/dreaming-phases.test.ts +++ b/extensions/memory-core/src/dreaming-phases.test.ts @@ -24,7 +24,14 @@ import { writeMemoryCoreWorkspaceEntry, } from "./dreaming-state.js"; import { previewRemHarness } from "./rem-harness.js"; -import { writeSessionIngestionState } from "./session-ingestion.js"; +import { + appendSessionCorpusLines, + foreignSessionIngestionSource, + mergeTrackedMessageHashes, + readSessionIngestionState, + scanSessionIngestionSource, + writeSessionIngestionState, +} from "./session-ingestion.js"; import { applyShortTermPromotions, rankShortTermPromotionCandidates, @@ -2241,6 +2248,150 @@ describe("memory-core dreaming phases", () => { expect(corpus).toContain("bulk-line-159"); }); + it("does not restage transcript messages after the seen-hash cap rolls over", async () => { + const workspaceDir = await createDreamingWorkspace(); + setDreamingTestEnv(path.join(workspaceDir, ".state")); + const transcriptPath = path.join(workspaceDir, "rollover-session.jsonl"); + const originalCount = 4_097; + try { + const startMs = Date.parse("2026-04-05T09:00:00.000Z"); + const record = (id: string, timestamp: number | string, content: string) => + `${JSON.stringify({ + type: "message", + id, + timestamp, + message: { role: "user", content, timestamp }, + })}\n`; + await fs.writeFile( + transcriptPath, + Array.from({ length: originalCount }, (_, index) => + record(`message-${index}`, startMs + index, `rollover-message-${index % 16}`), + ).join(""), + ); + const source = foreignSessionIngestionSource("main", transcriptPath); + const initialScan = await scanSessionIngestionSource({ + source, + seenMessages: {}, + verifyContent: true, + classifyDay: () => "include", + }); + const initialFileState = expectDefined(initialScan.fileState, "initial ingestion checkpoint"); + expect(initialScan.candidates).toHaveLength(originalCount); + const firstRenderedCorpusLine = expectDefined( + initialScan.candidates[0]?.rendered, + "first rendered rollover corpus line", + ); + const initialResults = await appendSessionCorpusLines({ + workspaceDir, + day: "2026-04-05", + lines: initialScan.candidates, + }); + await recordShortTermRecalls({ + workspaceDir, + query: "__dreaming_sessions__:2026-04-05", + // One old claim is enough to detect accidental reinforcement after the append. + results: initialResults.slice(0, 1), + signalType: "daily", + dedupeByQueryPerDay: true, + dayBucket: "2026-04-05", + nowMs: Date.parse("2026-04-05T10:05:00.000Z"), + }); + await writeSessionIngestionState(workspaceDir, { + version: 3, + files: { [source.stateKey]: initialFileState }, + seenMessages: { + [source.scope]: mergeTrackedMessageHashes( + [], + initialScan.candidates.map((candidate) => candidate.hash), + ), + }, + }); + + await fs.appendFile( + transcriptPath, + record("message-new", "2026-04-06T09:00:00.000Z", "rollover-message-new"), + ); + const persisted = await readSessionIngestionState(workspaceDir); + expect(persisted.seenMessages[source.scope]).toHaveLength(4_096); + expect(persisted.seenMessages[source.scope]).not.toContain(initialScan.candidates[0]?.hash); + const appendedScan = await scanSessionIngestionSource({ + source, + previous: persisted.files[source.stateKey], + seenMessages: persisted.seenMessages, + verifyContent: true, + classifyDay: () => "include", + }); + expect(appendedScan.candidates.map((candidate) => candidate.snippet)).toEqual([ + "User: rollover-message-new", + ]); + const appendedFileState = expectDefined( + appendedScan.fileState, + "appended ingestion checkpoint", + ); + const appendedResults = await appendSessionCorpusLines({ + workspaceDir, + day: "2026-04-06", + lines: appendedScan.candidates, + }); + await recordShortTermRecalls({ + workspaceDir, + query: "__dreaming_sessions__:2026-04-06", + results: appendedResults, + signalType: "daily", + dedupeByQueryPerDay: true, + dayBucket: "2026-04-06", + nowMs: Date.parse("2026-04-06T10:05:00.000Z"), + }); + await writeSessionIngestionState(workspaceDir, { + version: 3, + files: { [source.stateKey]: appendedFileState }, + seenMessages: { + [source.scope]: mergeTrackedMessageHashes( + persisted.seenMessages[source.scope] ?? [], + appendedScan.candidates.map((candidate) => candidate.hash), + ), + }, + }); + + const sessionCorpusDir = path.join(workspaceDir, "memory", ".dreams", "session-corpus"); + const corpusLines = ( + await Promise.all( + ( + await fs.readdir(sessionCorpusDir) + ) + .filter((name) => name.endsWith(".txt")) + .map((name) => fs.readFile(path.join(sessionCorpusDir, name), "utf-8")), + ) + ) + .flatMap((content) => content.split(/\r?\n/u)) + .filter(Boolean); + expect(corpusLines.filter((line) => line === firstRenderedCorpusLine)).toHaveLength(1); + expect( + corpusLines.filter((line) => line.endsWith("User: rollover-message-new")), + ).toHaveLength(1); + + const ranked = await rankShortTermPromotionCandidates({ + workspaceDir, + minScore: 0, + minRecallCount: 0, + minUniqueQueries: 0, + nowMs: Date.parse("2026-04-06T10:05:00.000Z"), + }); + const originalCandidates = ranked.filter( + (candidate) => candidate.snippet === "User: rollover-message-0", + ); + const appendedCandidates = ranked.filter( + (candidate) => candidate.snippet === "User: rollover-message-new", + ); + expect(originalCandidates).toHaveLength(1); + expect(originalCandidates[0]?.dailyCount).toBe(1); + expect(appendedCandidates).toHaveLength(1); + expect(appendedCandidates[0]?.dailyCount).toBe(1); + } finally { + restoreDreamingTestEnv(); + } + }); + it("preserves checkpoints for known sessions beyond a capped sweep", async () => { const workspaceDir = await createDreamingWorkspace(); setDreamingTestEnv(path.join(workspaceDir, ".state")); diff --git a/extensions/memory-core/src/session-backfill.ts b/extensions/memory-core/src/session-backfill.ts index 703934cb2e93..d95b09cf0214 100644 --- a/extensions/memory-core/src/session-backfill.ts +++ b/extensions/memory-core/src/session-backfill.ts @@ -216,6 +216,8 @@ function mergeSessionBackfillFileProgress(params: { ...(firstUnselected ? [firstUnselected.contentIndex] : []), ...(scan.progressBlockIndex !== undefined ? [scan.progressBlockIndex] : []), ]; + // The full snapshot identity stays paired while only its consumption cursor rewinds. + // Session ingestion uses that snapshot as the append-prefix proof on the next scan. files[scan.stateKey] = { mtimeMs: scan.mtimeMs, size: scan.size, diff --git a/extensions/memory-core/src/session-ingestion.test.ts b/extensions/memory-core/src/session-ingestion.test.ts index af63b27cfa07..e8be14476875 100644 --- a/extensions/memory-core/src/session-ingestion.test.ts +++ b/extensions/memory-core/src/session-ingestion.test.ts @@ -2,17 +2,14 @@ import fs from "node:fs/promises"; import os from "node:os"; import path from "node:path"; import { afterEach, describe, expect, it } from "vitest"; +import { useAutoCleanupTempDirTracker } from "../../../test/helpers/temp-dir.js"; import { foreignSessionIngestionSource, scanSessionIngestionSource, sessionIngestionSourceFromCorpus, } from "./session-ingestion.js"; -const tempDirs: string[] = []; - -afterEach(async () => { - await Promise.all(tempDirs.splice(0).map((dir) => fs.rm(dir, { recursive: true, force: true }))); -}); +const tempDirs = useAutoCleanupTempDirTracker(afterEach); describe("session ingestion", () => { it("preserves file-backed scope identity when a session id ends in .jsonl", () => { @@ -28,8 +25,7 @@ describe("session ingestion", () => { }); it("verifies backfill content despite an unchanged size and mtime", async () => { - const dir = await fs.mkdtemp(path.join(os.tmpdir(), "openclaw-session-ingestion-")); - tempDirs.push(dir); + const dir = tempDirs.make("openclaw-session-ingestion-"); const archiveFile = path.join(dir, "archive.jsonl"); const record = (content: string) => `${JSON.stringify({ @@ -73,4 +69,49 @@ describe("session ingestion", () => { "User: Bravo durable note.", ]); }); + + it("resumes after a validated transcript append", async () => { + const dir = tempDirs.make("openclaw-session-ingestion-"); + const archiveFile = path.join(dir, "archive.jsonl"); + const record = (id: string, content: string) => + `${JSON.stringify({ + type: "message", + id, + timestamp: "2026-04-05T18:00:00.000Z", + message: { + role: "user", + content, + timestamp: "2026-04-05T18:00:00.000Z", + }, + })}\n`; + await fs.writeFile( + archiveFile, + record("message-1", "Alpha durable note.") + record("message-2", "Bravo durable note."), + ); + const source = foreignSessionIngestionSource("main", archiveFile); + const first = await scanSessionIngestionSource({ + source, + seenMessages: {}, + verifyContent: true, + maxCandidates: 1, + classifyDay: () => "include", + }); + if (!first.fileState) { + throw new Error("expected initial backfill checkpoint"); + } + + await fs.appendFile(archiveFile, record("message-3", "Charlie durable note.")); + const second = await scanSessionIngestionSource({ + source, + previous: first.fileState, + seenMessages: {}, + verifyContent: true, + classifyDay: () => "include", + }); + + expect(second.candidates.map((candidate) => candidate.snippet)).toEqual([ + "User: Bravo durable note.", + "User: Charlie durable note.", + ]); + }); }); diff --git a/extensions/memory-core/src/session-ingestion.ts b/extensions/memory-core/src/session-ingestion.ts index f0fd1c3ac414..66a4d27036f4 100644 --- a/extensions/memory-core/src/session-ingestion.ts +++ b/extensions/memory-core/src/session-ingestion.ts @@ -4,6 +4,7 @@ import path from "node:path"; import { truncateUtf16Safe } from "openclaw/plugin-sdk/memory-core-host-engine-foundation"; import { buildSessionEntry, + matchesSessionEntryPrefixHash, parseUsageCountedSessionIdFromFileName, sessionPathForFile, statSessionEntrySync, @@ -238,12 +239,12 @@ export async function scanSessionIngestionSource(params: { ) { return emptyScan("unchanged", params.previous); } - const sameContent = - params.previous?.mtimeMs === fileFingerprint.mtimeMs && - params.previous.size === fileFingerprint.size && - params.previous.contentHash === terminalState.contentHash && - params.previous.lineCount === lines.length; - const startIndex = sameContent + const previousSnapshotMatches = + params.previous !== undefined && + params.previous.contentHash.length > 0 && + params.previous.lineCount <= lines.length && + matchesSessionEntryPrefixHash(entry, params.previous.lineCount, params.previous.contentHash); + const startIndex = previousSnapshotMatches ? Math.max(0, Math.min(params.previous?.lastContentLine ?? 0, lines.length)) : 0; const seen = new Set(params.seenMessages[params.source.scope] ?? []); diff --git a/packages/memory-host-sdk/src/engine-sessions.ts b/packages/memory-host-sdk/src/engine-sessions.ts index 935c398408bd..b6c2aa47d4ca 100644 --- a/packages/memory-host-sdk/src/engine-sessions.ts +++ b/packages/memory-host-sdk/src/engine-sessions.ts @@ -7,6 +7,7 @@ export { listSessionTranscriptCorpusEntriesForAgent, loadDreamingNarrativeTranscriptPathSetForAgent, loadSessionTranscriptClassificationForAgent, + matchesSessionEntryPrefixHash, normalizeSessionTranscriptPathForComparison, parseCanonicalSessionSyncTargetFromPath, resolveSessionIdentityForTranscriptFile, diff --git a/packages/memory-host-sdk/src/host/session-files-reset-revision.test.ts b/packages/memory-host-sdk/src/host/session-files-reset-revision.test.ts index dd503b2bd7b8..428d0be42c48 100644 --- a/packages/memory-host-sdk/src/host/session-files-reset-revision.test.ts +++ b/packages/memory-host-sdk/src/host/session-files-reset-revision.test.ts @@ -13,7 +13,12 @@ import { } from "../../../../src/config/sessions/session-accessor.js"; import { closeOpenClawAgentDatabasesForTest } from "../../../../src/state/openclaw-agent-db.js"; import { closeOpenClawStateDatabaseForTest } from "../../../../src/state/openclaw-state-db.js"; -import { buildSessionEntry, type SessionFileEntry } from "./session-files.js"; +import { hashText } from "./hash.js"; +import { + buildSessionEntry, + matchesSessionEntryPrefixHash, + type SessionFileEntry, +} from "./session-files.js"; function requireSessionEntry(entry: SessionFileEntry | null): SessionFileEntry { if (!entry) { @@ -22,6 +27,18 @@ function requireSessionEntry(entry: SessionFileEntry | null): SessionFileEntry { return entry; } +function legacySessionEntryHash(entry: SessionFileEntry): string { + return hashText( + entry.content + + "\n" + + entry.lineMap.join(",") + + "\n" + + entry.messageTimestampsMs.join(",") + + "\n" + + JSON.stringify(entry.lineProvenance), + ); +} + let tmpDir: string; let previousStateDir: string | undefined; let previousConfigPath: string | undefined; @@ -54,7 +71,7 @@ afterEach(() => { }); describe("SQLite session reset content revision", () => { - it("invalidates a session hash when a reset boundary changes its generation", async () => { + it("invalidates current and legacy prefix hashes across a reset generation", async () => { const sessionsDir = path.join(tmpDir, "agents", "main", "sessions"); const storePath = path.join(sessionsDir, "sessions.json"); const sessionKey = "agent:main:chat:reset-revision"; @@ -80,6 +97,22 @@ describe("SQLite session reset content revision", () => { updatedAtMs: 1, }; const before = requireSessionEntry(await buildSessionEntry(sessionKey, buildOptions)); + const beforeLineCount = before.content.split("\n").length; + const legacyBeforeHash = legacySessionEntryHash(before); + + await persistSessionTranscriptTurn( + { agentId: "main", sessionId, sessionKey, storePath }, + { + messages: [{ message: { role: "assistant", content: "ordinary appended text" } }], + touchSessionEntry: true, + updateMode: "none", + }, + ); + const afterAppend = requireSessionEntry(await buildSessionEntry(sessionKey, buildOptions)); + expect(matchesSessionEntryPrefixHash(afterAppend, beforeLineCount, before.hash)).toBe(true); + expect(matchesSessionEntryPrefixHash(afterAppend, beforeLineCount, legacyBeforeHash)).toBe( + true, + ); await resetSessionEntryLifecycle({ agentId: "main", @@ -94,8 +127,8 @@ describe("SQLite session reset content revision", () => { }); const after = requireSessionEntry(await buildSessionEntry(sessionKey, buildOptions)); - expect(after.content).toBe(before.content); - expect(after.lineMap).toEqual(before.lineMap); + expect(after.content).toBe(afterAppend.content); + expect(after.lineMap).toEqual(afterAppend.lineMap); const cutoffSymbol = Symbol.for("openclaw.memory.sessionResetRecallCutoff"); expect(Object.getOwnPropertyDescriptor(after, cutoffSymbol)).toMatchObject({ enumerable: false, @@ -103,5 +136,7 @@ describe("SQLite session reset content revision", () => { }); expect(Object.keys(after)).not.toContain(cutoffSymbol.description); expect(after.hash).not.toBe(before.hash); + expect(matchesSessionEntryPrefixHash(after, beforeLineCount, before.hash)).toBe(false); + expect(matchesSessionEntryPrefixHash(after, beforeLineCount, legacyBeforeHash)).toBe(false); }); }); diff --git a/packages/memory-host-sdk/src/host/session-files.ts b/packages/memory-host-sdk/src/host/session-files.ts index c88dc5275e5e..7edfcaa49085 100644 --- a/packages/memory-host-sdk/src/host/session-files.ts +++ b/packages/memory-host-sdk/src/host/session-files.ts @@ -58,6 +58,8 @@ const SESSION_EXPORT_CONTENT_WRAP_CHARS = 800; const SESSION_ENTRY_PARSE_YIELD_LINES = 250; const MAX_DATE_TIMESTAMP_MS = 8_640_000_000_000_000; const DIRECT_CRON_PROMPT_RE = /^\[cron:[^\]]+\]\s*/; +const SESSION_RESET_RECALL_CUTOFF = Symbol.for("openclaw.memory.sessionResetRecallCutoff"); +type SessionResetRecallCutoff = ReturnType; export type SessionFileEntry = { path: string; @@ -116,6 +118,66 @@ export type ResolvedSessionTranscriptIdentity = { sessionKey?: string; }; +function hashSessionEntrySnapshot(params: { + content: string; + lineMap: readonly number[]; + messageTimestampsMs: readonly number[]; + lineProvenance: readonly MemoryEntryProvenance[]; + resetRecallCutoff?: SessionResetRecallCutoff; +}): string { + const snapshot = + params.content + + "\n" + + params.lineMap.join(",") + + "\n" + + params.messageTimestampsMs.join(",") + + "\n" + + JSON.stringify(params.lineProvenance); + return hashText( + params.resetRecallCutoff + ? `${snapshot}\n${JSON.stringify(params.resetRecallCutoff)}` + : snapshot, + ); +} + +function readSessionEntryResetRecallCutoff(entry: SessionFileEntry): SessionResetRecallCutoff { + const value: unknown = Object.getOwnPropertyDescriptor(entry, SESSION_RESET_RECALL_CUTOFF)?.value; + if (!value || typeof value !== "object" || !("state" in value)) { + return { state: "invalid" }; + } + if (value.state === "absent" || value.state === "invalid") { + return { state: value.state }; + } + if (value.state === "valid" && "cutoffLine" in value && typeof value.cutoffLine === "number") { + return { state: "valid", cutoffLine: value.cutoffLine }; + } + return { state: "invalid" }; +} + +export function matchesSessionEntryPrefixHash( + entry: SessionFileEntry, + lineCount: number, + expectedHash: string, +): boolean { + const lines = entry.content ? entry.content.split("\n") : []; + if (!Number.isInteger(lineCount) || lineCount < 0 || lineCount > lines.length) { + return false; + } + const resetRecallCutoff = readSessionEntryResetRecallCutoff(entry); + const prefix = { + content: lines.slice(0, lineCount).join("\n"), + lineMap: entry.lineMap.slice(0, lineCount), + messageTimestampsMs: entry.messageTimestampsMs.slice(0, lineCount), + lineProvenance: entry.lineProvenance.slice(0, lineCount), + }; + if (hashSessionEntrySnapshot({ ...prefix, resetRecallCutoff }) === expectedHash) { + return true; + } + // Checkpoints written before reset generations carried the same snapshot + // hash without cutoff state. They are safe to advance only before any reset. + return resetRecallCutoff.state === "absent" && hashSessionEntrySnapshot(prefix) === expectedHash; +} + type SessionTranscriptStoreEntry = { sessionFile?: unknown; sessionId?: unknown; @@ -971,17 +1033,13 @@ export async function buildSessionEntry( absPath, mtimeMs, size, - hash: hashText( - content + - "\n" + - lineMap.join(",") + - "\n" + - messageTimestampsMs.join(",") + - "\n" + - JSON.stringify(lineProvenance) + - "\n" + - JSON.stringify(rawSource?.resetRecallCutoff ?? { state: "absent" }), - ), + hash: hashSessionEntrySnapshot({ + content, + lineMap, + messageTimestampsMs, + lineProvenance, + resetRecallCutoff: rawSource?.resetRecallCutoff ?? { state: "absent" }, + }), content, lineMap, messageTimestampsMs, @@ -990,7 +1048,7 @@ export async function buildSessionEntry( ...(generatedByDreamingNarrative ? { generatedByDreamingNarrative: true } : {}), ...(generatedByCronRun ? { generatedByCronRun: true } : {}), }; - Object.defineProperty(entry, Symbol.for("openclaw.memory.sessionResetRecallCutoff"), { + Object.defineProperty(entry, SESSION_RESET_RECALL_CUTOFF, { configurable: false, enumerable: false, value: rawSource?.resetRecallCutoff ?? { state: "absent" }, diff --git a/src/plugin-sdk/memory-core-host-engine-sessions.ts b/src/plugin-sdk/memory-core-host-engine-sessions.ts index 4a4a1cff710a..8d3c40ab199d 100644 --- a/src/plugin-sdk/memory-core-host-engine-sessions.ts +++ b/src/plugin-sdk/memory-core-host-engine-sessions.ts @@ -9,6 +9,7 @@ export { listSessionTranscriptCorpusEntriesForAgent, loadDreamingNarrativeTranscriptPathSetForAgent, loadSessionTranscriptClassificationForAgent, + matchesSessionEntryPrefixHash, normalizeSessionTranscriptPathForComparison, parseCanonicalSessionSyncTargetFromPath, parseSqliteSessionFileMarker,