fix(memory): make legacy session backfill rollback re-applicable (#116004)

* fix(memory): reset legacy backfill cursors on rollback * fix(memory): track complete backfill rewind baselines
2026-08-02 12:21:36 +00:00 · 2026-07-29 13:08:25 -04:00
parent 4823d7fe7b
commit 841bfcdcb8
3 changed files with 281 additions and 18 deletions
--- a/extensions/memory-core/src/session-backfill-lifecycle.ts
+++ b/extensions/memory-core/src/session-backfill-lifecycle.ts
@@ -1,7 +1,7 @@
 import { createHash } from "node:crypto";
 import { readSessionIngestionState, writeSessionIngestionState } from "./dreaming-phases.js";
 import {
-  clearMemoryCoreWorkspaceNamespace,
+  deleteMemoryCoreWorkspaceEntry,
  readMemoryCoreWorkspaceEntries,
  SESSION_BACKFILL_REWIND_NAMESPACE,
  writeMemoryCoreWorkspaceEntry,
@@ -13,6 +13,8 @@ import type {

 const DEFAULT_SESSION_BACKFILL_LIMIT_DAYS = 92;
 const MEMORY_DAY_RE = /^\d{4}-\d{2}-\d{2}$/;
+// Batch keys are SHA-256 hex digests, so this colon-delimited marker cannot collide.
+const SESSION_BACKFILL_BASELINE_KEY_PREFIX = "complete-baseline:";

 type SessionBackfillRewindCandidate = {
  contentIndex: number;
@@ -26,6 +28,12 @@ type SessionBackfillRewindBatch = {
  candidates: SessionBackfillRewindCandidate[];
 };

+type SessionBackfillBaseline = {
+  version: 1;
+  complete: true;
+  agentId: string;
+};
+
 function normalizeMemoryDay(value: string | undefined, flag: string): string | undefined {
  if (value === undefined) {
    return undefined;
@@ -81,6 +89,18 @@ export async function recordSessionBackfillRewindBatch(params: {
  });
 }

+export async function markSessionBackfillRewindBaseline(params: {
+  workspaceDir: string;
+  agentId: string;
+}): Promise<void> {
+  await writeMemoryCoreWorkspaceEntry<SessionBackfillBaseline>({
+    namespace: SESSION_BACKFILL_REWIND_NAMESPACE,
+    workspaceDir: params.workspaceDir,
+    key: `${SESSION_BACKFILL_BASELINE_KEY_PREFIX}${params.agentId}`,
+    value: { version: 1, complete: true, agentId: params.agentId },
+  });
+}
+
 function isSessionBackfillRewindCandidate(value: unknown): value is SessionBackfillRewindCandidate {
  if (value === null || typeof value !== "object" || Array.isArray(value)) {
    return false;
@@ -98,25 +118,43 @@ function isSessionBackfillRewindCandidate(value: unknown): value is SessionBackf
  );
 }

-export async function rewindSessionBackfillIngestionState(workspaceDir: string): Promise<void> {
-  const entries = await readMemoryCoreWorkspaceEntries<SessionBackfillRewindBatch>({
+function isSessionBackfillRewindBatch(
+  value: SessionBackfillRewindBatch | SessionBackfillBaseline,
+): value is SessionBackfillRewindBatch {
+  return value.version === 1 && "candidates" in value && Array.isArray(value.candidates);
+}
+
+export async function rewindSessionBackfillIngestionState(params: {
+  workspaceDir: string;
+  agentId: string;
+}): Promise<{
+  completeCoverage: boolean;
+  rewoundCandidates: number;
+}> {
+  const entries = await readMemoryCoreWorkspaceEntries<
+    SessionBackfillRewindBatch | SessionBackfillBaseline
+  >({
    namespace: SESSION_BACKFILL_REWIND_NAMESPACE,
-    workspaceDir,
+    workspaceDir: params.workspaceDir,
  });
-  const candidates = entries.flatMap((entry) =>
-    entry.value?.version === 1 && Array.isArray(entry.value.candidates)
+  const completeCoverage = entries.some(
+    (entry) =>
+      entry.key === `${SESSION_BACKFILL_BASELINE_KEY_PREFIX}${params.agentId}` &&
+      "complete" in entry.value &&
+      entry.value.agentId === params.agentId,
+  );
+  const batchEntries = entries.filter((entry) => isSessionBackfillRewindBatch(entry.value));
+  const candidates = batchEntries.flatMap((entry) =>
+    isSessionBackfillRewindBatch(entry.value)
      ? entry.value.candidates.filter(isSessionBackfillRewindCandidate)
      : [],
  );
  if (candidates.length === 0) {
-    await clearMemoryCoreWorkspaceNamespace({
-      namespace: SESSION_BACKFILL_REWIND_NAMESPACE,
-      workspaceDir,
-    });
-    return;
+    await deleteSessionBackfillRewindBatches(params.workspaceDir, batchEntries);
+    return { completeCoverage, rewoundCandidates: 0 };
  }

-  const state = await readSessionIngestionState(workspaceDir);
+  const state = await readSessionIngestionState(params.workspaceDir);
  const removedHashesByScope = new Map<string, Set<string>>();
  const rewindLineByStateKey = new Map<string, number>();
  for (const candidate of candidates) {
@@ -153,10 +191,57 @@ export async function rewindSessionBackfillIngestionState(workspaceDir: string):
      };
    }
  }
-  await writeSessionIngestionState(workspaceDir, { ...state, files, seenMessages });
-  await clearMemoryCoreWorkspaceNamespace({
-    namespace: SESSION_BACKFILL_REWIND_NAMESPACE,
-    workspaceDir,
+  await writeSessionIngestionState(params.workspaceDir, { ...state, files, seenMessages });
+  await deleteSessionBackfillRewindBatches(params.workspaceDir, batchEntries);
+  return { completeCoverage, rewoundCandidates: candidates.length };
+}
+
+async function deleteSessionBackfillRewindBatches(
+  workspaceDir: string,
+  entries: Array<{ key: string }>,
+): Promise<void> {
+  await Promise.all(
+    entries.map((entry) =>
+      deleteMemoryCoreWorkspaceEntry({
+        namespace: SESSION_BACKFILL_REWIND_NAMESPACE,
+        workspaceDir,
+        key: entry.key,
+      }),
+    ),
+  );
+}
+
+function belongsToAgentFileState(key: string, agentId: string): boolean {
+  return key.startsWith(`${agentId}:`);
+}
+
+function belongsToAgentSeenState(key: string, agentId: string): boolean {
+  const archivePrefix = "archive:";
+  if (!key.startsWith(archivePrefix)) {
+    return key.startsWith(`${agentId}:`);
+  }
+  const archiveAgentEnd = key.indexOf(":", archivePrefix.length);
+  if (archiveAgentEnd === -1) {
+    return agentId === "archive";
+  }
+  return key.slice(archivePrefix.length, archiveAgentEnd) === agentId;
+}
+
+export async function resetSessionBackfillIngestionState(params: {
+  workspaceDir: string;
+  agentId: string;
+}): Promise<void> {
+  const state = await readSessionIngestionState(params.workspaceDir);
+  await writeSessionIngestionState(params.workspaceDir, {
+    ...state,
+    files: Object.fromEntries(
+      Object.entries(state.files).filter(([key]) => !belongsToAgentFileState(key, params.agentId)),
+    ),
+    seenMessages: Object.fromEntries(
+      Object.entries(state.seenMessages).filter(
+        ([key]) => !belongsToAgentSeenState(key, params.agentId),
+      ),
+    ),
  });
 }

--- a/extensions/memory-core/src/session-backfill.test.ts
+++ b/extensions/memory-core/src/session-backfill.test.ts
@@ -10,13 +10,23 @@ import { upsertSessionEntry } from "openclaw/plugin-sdk/session-store-runtime";
 import { appendSessionTranscriptMessageByIdentity } from "openclaw/plugin-sdk/session-transcript-runtime";
 import { afterEach, describe, expect, it, vi } from "vitest";
 import { writeBackfillDiaryEntries } from "./dreaming-narrative.js";
+import { writeSessionIngestionState } from "./dreaming-phases.js";
+import {
+  clearMemoryCoreWorkspaceNamespace,
+  SESSION_BACKFILL_REWIND_NAMESPACE,
+} from "./dreaming-state.js";
+import {
+  markSessionBackfillRewindBaseline,
+  resetSessionBackfillIngestionState,
+  rewindSessionBackfillIngestionState,
+} from "./session-backfill-lifecycle.js";
 import {
  executeSessionBackfill,
  executeSessionBackfillBatch,
  runSessionBackfill,
 } from "./session-backfill.js";
 import { readShortTermRecallEntries } from "./short-term-promotion.js";
-import { createMemoryCoreTestHarness } from "./test-helpers.js";
+import { createMemoryCoreTestHarness, dreamingTestState } from "./test-helpers.js";

 const harness = createMemoryCoreTestHarness();

@@ -553,4 +563,161 @@ describe("runSessionBackfill", () => {
    expect(reapplied.candidateCount).toBe(2);
    expect(hashStagedContent(afterReapply)).toBe(firstContentHash);
  });
+
+  it("resets agent ingestion state when legacy staged entries have no rewind journal", async () => {
+    const workspaceDir = await createIsolatedWorkspace("legacy-rollback-");
+    await seedCanonicalTranscript("legacy", [
+      {
+        role: "user",
+        content: "The preferred terminal is Ghostty",
+        timestamp: "2026-04-01T10:00:00.000Z",
+        owner: true,
+      },
+    ]);
+    const applyParams = {
+      agentId: "main",
+      workspaceDir,
+      apply: true,
+      nowMs: Date.parse("2026-04-02T12:00:00.000Z"),
+      timezone: "UTC",
+    } as const;
+
+    await runSessionBackfill(applyParams);
+    const firstContentHash = hashStagedContent(await readShortTermRecallEntries({ workspaceDir }));
+    await clearMemoryCoreWorkspaceNamespace({
+      namespace: SESSION_BACKFILL_REWIND_NAMESPACE,
+      workspaceDir,
+    });
+
+    const rollback = await runSessionBackfill({
+      agentId: "main",
+      workspaceDir,
+      rollback: true,
+    });
+    const preview = await runSessionBackfill({
+      agentId: "main",
+      workspaceDir,
+      timezone: "UTC",
+    });
+    const reapplied = await runSessionBackfill(applyParams);
+    const afterReapply = await readShortTermRecallEntries({ workspaceDir });
+
+    expect(rollback.rollback).toEqual({
+      removedDiaryEntries: 1,
+      removedStagedEntries: 1,
+    });
+    expect(preview.candidateCount).toBe(1);
+    expect(reapplied.candidateCount).toBe(1);
+    expect(hashStagedContent(afterReapply)).toBe(firstContentHash);
+
+    const stateAfterReapply = await dreamingTestState.readSessionIngestionState(workspaceDir);
+    const scope = Object.keys(stateAfterReapply.seenMessages)[0];
+    if (!scope) {
+      throw new Error("Expected re-applied session ingestion scope");
+    }
+    await writeSessionIngestionState(workspaceDir, {
+      ...stateAfterReapply,
+      seenMessages: {
+        ...stateAfterReapply.seenMessages,
+        [scope]: [...(stateAfterReapply.seenMessages[scope] ?? []), "later-live"],
+      },
+    });
+    await runSessionBackfill({ agentId: "main", workspaceDir, rollback: true });
+    expect(
+      (await dreamingTestState.readSessionIngestionState(workspaceDir)).seenMessages[scope],
+    ).toEqual(["later-live"]);
+  });
+
+  it("resets mixed legacy state when later journal rows do not prove complete coverage", async () => {
+    const workspaceDir = await createIsolatedWorkspace("mixed-legacy-rollback-");
+    const applyParams = {
+      agentId: "main",
+      workspaceDir,
+      apply: true,
+      nowMs: Date.parse("2026-05-03T12:00:00.000Z"),
+      timezone: "UTC",
+    } as const;
+    await seedCanonicalTranscript("legacy-mixed", [
+      {
+        role: "user",
+        content: "The preferred shell is zsh",
+        timestamp: "2026-05-01T10:00:00.000Z",
+        owner: true,
+      },
+    ]);
+    await runSessionBackfill(applyParams);
+    await clearMemoryCoreWorkspaceNamespace({
+      namespace: SESSION_BACKFILL_REWIND_NAMESPACE,
+      workspaceDir,
+    });
+    await seedCanonicalTranscript("journaled-mixed", [
+      {
+        role: "user",
+        content: "The preferred pager is less",
+        timestamp: "2026-05-02T10:00:00.000Z",
+        owner: true,
+      },
+    ]);
+    await runSessionBackfill(applyParams);
+    const firstContentHash = hashStagedContent(await readShortTermRecallEntries({ workspaceDir }));
+
+    await runSessionBackfill({ agentId: "main", workspaceDir, rollback: true });
+    const preview = await runSessionBackfill({ agentId: "main", workspaceDir, timezone: "UTC" });
+    const reapplied = await runSessionBackfill(applyParams);
+
+    expect(preview.candidateCount).toBe(2);
+    expect(reapplied.candidateCount).toBe(2);
+    expect(hashStagedContent(await readShortTermRecallEntries({ workspaceDir }))).toBe(
+      firstContentHash,
+    );
+  });
+
+  it("keeps other agents' archived scopes when resetting an agent named archive", async () => {
+    const workspaceDir = await createIsolatedWorkspace("archive-agent-reset-");
+    const fileState = {
+      mtimeMs: 1,
+      size: 1,
+      contentHash: "hash",
+      lineCount: 1,
+      lastContentLine: 1,
+    };
+    await writeSessionIngestionState(workspaceDir, {
+      version: 3,
+      files: {
+        "archive:sessions/archive/own": fileState,
+        "main:sessions/main/other": fileState,
+      },
+      seenMessages: {
+        "archive:sessions/archive/own": ["own-live"],
+        "archive:archive:/tmp/own.jsonl": ["own-archive"],
+        "archive:main:/tmp/other.jsonl": ["other-archive"],
+        "main:sessions/main/other": ["other-live"],
+      },
+    });
+
+    await resetSessionBackfillIngestionState({ workspaceDir, agentId: "archive" });
+
+    expect(await dreamingTestState.readSessionIngestionState(workspaceDir)).toEqual({
+      version: 3,
+      files: { "main:sessions/main/other": fileState },
+      seenMessages: {
+        "archive:main:/tmp/other.jsonl": ["other-archive"],
+        "main:sessions/main/other": ["other-live"],
+      },
+    });
+  });
+
+  it("does not share a clean rewind baseline across agents", async () => {
+    const workspaceDir = await createIsolatedWorkspace("agent-baseline-");
+    await markSessionBackfillRewindBaseline({ workspaceDir, agentId: "main" });
+
+    expect(await rewindSessionBackfillIngestionState({ workspaceDir, agentId: "other" })).toEqual({
+      completeCoverage: false,
+      rewoundCandidates: 0,
+    });
+    expect(await rewindSessionBackfillIngestionState({ workspaceDir, agentId: "main" })).toEqual({
+      completeCoverage: true,
+      rewoundCandidates: 0,
+    });
+  });
 });
--- a/extensions/memory-core/src/session-backfill.ts
+++ b/extensions/memory-core/src/session-backfill.ts
@@ -38,8 +38,10 @@ import type {
 } from "./session-backfill-contract.js";
 import {
  drainSessionBackfill,
+  markSessionBackfillRewindBaseline,
  normalizeSessionBackfillSelection,
  recordSessionBackfillRewindBatch,
+  resetSessionBackfillIngestionState,
  rewindSessionBackfillIngestionState,
 } from "./session-backfill-lifecycle.js";
 import {
@@ -537,7 +539,16 @@ async function executeSessionBackfillCore(
      removeBackfillDiaryEntries({ workspaceDir }),
      removeGroundedShortTermCandidates({ workspaceDir }),
    ]);
-    await rewindSessionBackfillIngestionState(workspaceDir);
+    const rewind = await rewindSessionBackfillIngestionState({
+      workspaceDir,
+      agentId: params.agentId,
+    });
+    if (!rewind.completeCoverage && (diary.removed > 0 || staged.removed > 0)) {
+      // Applies from before the rewind journal shipped have no owned offsets to restore.
+      // Without this agent-scoped reset, rollback deletes artifacts but re-apply finds nothing.
+      await resetSessionBackfillIngestionState({ workspaceDir, agentId: params.agentId });
+    }
+    await markSessionBackfillRewindBaseline({ workspaceDir, agentId: params.agentId });
    return {
      result: {
        agentId: params.agentId,