diff --git a/extensions/memory-core/src/session-backfill-lifecycle.ts b/extensions/memory-core/src/session-backfill-lifecycle.ts index 7f928d819526..4cce4bdd8211 100644 --- 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 { + await writeMemoryCoreWorkspaceEntry({ + 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 { - const entries = await readMemoryCoreWorkspaceEntries({ +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>(); const rewindLineByStateKey = new Map(); 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 { + 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 { + 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), + ), + ), }); } diff --git a/extensions/memory-core/src/session-backfill.test.ts b/extensions/memory-core/src/session-backfill.test.ts index 90cd0d62bdef..a094bc3153b7 100644 --- 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, + }); + }); }); diff --git a/extensions/memory-core/src/session-backfill.ts b/extensions/memory-core/src/session-backfill.ts index bc7104a9d49b..947e29bd1f83 100644 --- 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,