fix(memory-core): resume validated session transcript appends

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;
 };
 

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"));

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,

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.",
+    ]);
+  });
 });

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] ?? []);

packages/memory-host-sdk/src/engine-sessions.ts

@@ -7,6 +7,7 @@ export {
   listSessionTranscriptCorpusEntriesForAgent,
   loadDreamingNarrativeTranscriptPathSetForAgent,
   loadSessionTranscriptClassificationForAgent,
+  matchesSessionEntryPrefixHash,
   normalizeSessionTranscriptPathForComparison,
   parseCanonicalSessionSyncTargetFromPath,
   resolveSessionIdentityForTranscriptFile,

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);
   });
 });

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<typeof resolveSessionResetRecallCutoff>;
 
 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" },

src/plugin-sdk/memory-core-host-engine-sessions.ts

@@ -9,6 +9,7 @@ export {
   listSessionTranscriptCorpusEntriesForAgent,
   loadDreamingNarrativeTranscriptPathSetForAgent,
   loadSessionTranscriptClassificationForAgent,
+  matchesSessionEntryPrefixHash,
   normalizeSessionTranscriptPathForComparison,
   parseCanonicalSessionSyncTargetFromPath,
   parseSqliteSessionFileMarker,