fix(plugin): paginate dirty-closed reward recovery

apps/memos-local-plugin/core/pipeline/memory-core.ts

@@ -86,6 +86,7 @@ import {
   FLOAT32_BYTES,
 } from "../storage/repos/index.js";
 import type { EmbeddingCountsBucket } from "../storage/repos/index.js";
+import type { ClosedEpisodeCursor } from "../storage/repos/episodes.js";
 import { createEmbedder } from "../embedding/embedder.js";
 import { createLlmClient } from "../llm/client.js";
 import {
@@ -621,6 +622,8 @@ export function createMemoryCore(
   const MAX_DIRTY_REWARD_ATTEMPTS = 3;
   const DIRTY_REWARD_BACKOFF_BASE_MS = 60 * 60 * 1000; // 1h
   const DIRTY_REWARD_BACKOFF_MAX_MS = 24 * 60 * 60 * 1000; // 24h
+  const DIRTY_CLOSED_SCAN_PAGE_SIZE = 500;
+  const DIRTY_CLOSED_SCAN_CURSOR_KEY = "pipeline.dirty_closed_scan_cursor.v1";
 
   // ─── Startup recovery background promise (issue #1776 + #1808) ──
   // `init()` used to `await` the entire reflect → reward → L2 chain for
@@ -704,8 +707,7 @@ export function createMemoryCore(
     // must be a no-op.
     if (handle.algorithm.lightweightMemory.enabled) return;
     try {
-      const allDirty = handle.repos.episodes
-        .list({ status: "closed", limit: 500 })
+      const allDirty = collectDirtyClosedEpisodes()
         .filter((ep) => !isLightweightEpisode(ep) && episodeRewardIsDirty(ep));
       // Apply the same backoff filter as init() so the 10-min periodic
       // scan does not hammer episodes whose LLM call keeps failing.
@@ -1083,8 +1085,7 @@ export function createMemoryCore(
         dirtyClosedForBackground = [];
       } else {
         const nowForDirty = Date.now();
-        const allDirty = handle.repos.episodes
-          .list({ status: "closed", limit: 500 })
+        const allDirty = collectDirtyClosedEpisodes()
           .filter((ep) => !isLightweightEpisode(ep) && episodeRewardIsDirty(ep));
         const dirtyClosed: typeof allDirty = [];
         for (const ep of allDirty) {
@@ -1528,6 +1529,42 @@ export function createMemoryCore(
     }
   }
 
+  function collectDirtyClosedEpisodes(): Array<EpisodeRow & { meta?: Record<string, unknown> }> {
+    const storedCursor = handle.repos.kv.get<unknown>(DIRTY_CLOSED_SCAN_CURSOR_KEY, null);
+    const cursor = isDirtyClosedScanCursor(storedCursor) ? storedCursor : null;
+    if (storedCursor !== null && !cursor) {
+      handle.repos.kv.del(DIRTY_CLOSED_SCAN_CURSOR_KEY);
+    }
+    const page = handle.repos.episodes.listClosedPage({
+      limit: DIRTY_CLOSED_SCAN_PAGE_SIZE,
+      before: cursor,
+    });
+    if (page.length === 0) {
+      handle.repos.kv.del(DIRTY_CLOSED_SCAN_CURSOR_KEY);
+      return [];
+    }
+
+    const last = page[page.length - 1]!;
+    if (page.length < DIRTY_CLOSED_SCAN_PAGE_SIZE) {
+      handle.repos.kv.del(DIRTY_CLOSED_SCAN_CURSOR_KEY);
+    } else {
+      handle.repos.kv.set(DIRTY_CLOSED_SCAN_CURSOR_KEY, {
+        startedAt: last.startedAt,
+        id: last.id,
+      });
+    }
+    return page;
+  }
+
+  function isDirtyClosedScanCursor(value: unknown): value is ClosedEpisodeCursor {
+    return Boolean(
+      value &&
+      typeof value === "object" &&
+      typeof (value as { startedAt?: unknown }).startedAt === "number" &&
+      typeof (value as { id?: unknown }).id === "string",
+    );
+  }
+
   function episodeRewardIsDirty(ep: EpisodeRow & { meta?: Record<string, unknown> }): boolean {
     const meta = ep.meta ?? {};
     if (meta.lightweightMemory === true) return false;

apps/memos-local-plugin/core/storage/repos/episodes.ts

@@ -3,6 +3,7 @@ import type { EpisodeListFilter, StorageDb } from "../types.js";
 import { buildInsert, buildUpdate } from "../tx.js";
 import {
   buildPageClauses,
+  clampLimit,
   fromJsonText,
   joinWhere,
   normalizeShareForStorage,
@@ -31,6 +32,11 @@ export interface EpisodeMetaRow {
   meta?: Record<string, unknown>;
 }
 
+export interface ClosedEpisodeCursor {
+  startedAt: number;
+  id: string;
+}
+
 export function makeEpisodesRepo(db: StorageDb) {
   const insert = db.prepare(buildInsert({ table: "episodes", columns: COLUMNS }));
   const replace = db.prepare(
@@ -176,6 +182,29 @@ export function makeEpisodesRepo(db: StorageDb) {
       return db.prepare<typeof params, RawEpisodeRow>(sql).all(params).map(mapRow);
     },
 
+    /**
+     * Return one stable, newest-first page of closed episodes. A cursor keeps
+     * scans progressing when new rows are inserted ahead of an earlier page.
+     */
+    listClosedPage(opts: {
+      limit?: number;
+      before?: ClosedEpisodeCursor | null;
+    } = {}): Array<EpisodeRow & EpisodeMetaRow> {
+      const limit = clampLimit(opts.limit ?? 500);
+      const params: Record<string, unknown> = { status: "closed" };
+      const fragments = ["status = @status"];
+      if (opts.before) {
+        fragments.push(
+          "(started_at < @before_started_at OR (started_at = @before_started_at AND id < @before_id))",
+        );
+        params.before_started_at = opts.before.startedAt;
+        params.before_id = opts.before.id;
+      }
+      const where = joinWhere(fragments);
+      const sql = `SELECT ${COLUMNS.join(", ")} FROM episodes ${where} ORDER BY started_at DESC, id DESC LIMIT ${limit}`;
+      return db.prepare<typeof params, RawEpisodeRow>(sql).all(params).map(mapRow);
+    },
+
     count(filter: Omit<EpisodeListFilter, "limit" | "offset"> = {}): number {
       const tr = timeRangeWhere(filter, "started_at");
       const fragments: string[] = [];

apps/memos-local-plugin/tests/unit/pipeline/memory-core.test.ts

@@ -1519,6 +1519,129 @@ algorithm:
     expect(meta.reward?.traceIds).toEqual(["tr_missing_reward"]);
   });
 
+  it("continues a bounded dirty-closed scan past 500 rows across restart", async () => {
+    home = await makeTmpHome({
+      agent: "openclaw",
+      configYaml: FULL_MEMORY_CONFIG_YAML,
+    });
+
+    const seeder = await bootstrapMemoryCore({
+      agent: "openclaw",
+      home: home.home,
+      config: home.config,
+      pkgVersion: "dirty-page-cursor-seed",
+    });
+    await seeder.init();
+    await seeder.shutdown();
+
+    const Sqlite = (await import("better-sqlite3")).default;
+    const writeDb = new Sqlite(home.home.dbFile);
+    const ts = Date.now() - 10_000;
+    writeDb
+      .prepare(
+        `INSERT INTO sessions (id, agent, started_at, last_seen_at, meta_json) VALUES (?, ?, ?, ?, ?)`,
+      )
+      .run("se_dirty_page_cursor", "openclaw", ts, ts, "{}");
+    const insertEpisode = writeDb.prepare(
+      `INSERT INTO episodes (id, session_id, started_at, ended_at, trace_ids_json, r_task, status, meta_json) VALUES (?, ?, ?, ?, ?, ?, 'closed', ?)`,
+    );
+    for (let i = 0; i < 500; i++) {
+      insertEpisode.run(
+        `ep_clean_${i}`,
+        "se_dirty_page_cursor",
+        ts + i,
+        ts + i + 1,
+        "[]",
+        0,
+        JSON.stringify({ closeReason: "finalized" }),
+      );
+    }
+    insertEpisode.run(
+      "ep_dirty_past_first_page",
+      "se_dirty_page_cursor",
+      ts - 1,
+      ts,
+      JSON.stringify(["tr_dirty_past_first_page"]),
+      null,
+      JSON.stringify({ closeReason: "finalized" }),
+    );
+    writeDb
+      .prepare(
+        `INSERT INTO traces (
+          id, episode_id, session_id, ts, user_text, agent_text, summary,
+          tool_calls_json, reflection, agent_thinking, value, alpha, r_human,
+          priority, tags_json, error_signatures_json, vec_summary, vec_action,
+          share_scope, share_target, shared_at, turn_id, schema_version
+        ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, NULL, NULL, NULL, NULL, NULL, ?, ?)`,
+      )
+      .run(
+        "tr_dirty_past_first_page",
+        "ep_dirty_past_first_page",
+        "se_dirty_page_cursor",
+        ts,
+        "Please recover the closed episode beyond the initial page.",
+        "The recovery should eventually score this episode.",
+        "Recovery pagination regression",
+        "[]",
+        null,
+        null,
+        0,
+        0,
+        null,
+        0.5,
+        "[]",
+        "[]",
+        ts,
+        1,
+      );
+    writeDb.close();
+
+    const first = await bootstrapMemoryCore({
+      agent: "openclaw",
+      home: home.home,
+      config: home.config,
+      pkgVersion: "dirty-page-cursor-first-scan",
+    });
+    await first.init();
+    await first.waitForStartupRecovery?.();
+    await first.shutdown();
+
+    const betweenScansDb = new Sqlite(home.home.dbFile);
+    betweenScansDb.prepare(
+      `INSERT INTO episodes (id, session_id, started_at, ended_at, trace_ids_json, r_task, status, meta_json) VALUES (?, ?, ?, ?, ?, ?, 'closed', ?)`,
+    ).run(
+      "ep_inserted_between_scans",
+      "se_dirty_page_cursor",
+      ts + 10_000,
+      ts + 10_001,
+      "[]",
+      0,
+      JSON.stringify({ closeReason: "finalized" }),
+    );
+    betweenScansDb.close();
+
+    core = await bootstrapMemoryCore({
+      agent: "openclaw",
+      home: home.home,
+      config: home.config,
+      pkgVersion: "dirty-page-cursor-second-scan",
+    });
+    await core.init();
+    await core.waitForStartupRecovery?.();
+
+    const readDb = new Sqlite(home.home.dbFile, { readonly: true });
+    const episode = readDb
+      .prepare("SELECT r_task, meta_json FROM episodes WHERE id = ?")
+      .get("ep_dirty_past_first_page") as { r_task: number | null; meta_json: string } | undefined;
+    readDb.close();
+
+    expect(episode).toBeDefined();
+    expect(episode!.r_task).toBe(0);
+    expect(JSON.parse(episode!.meta_json)).toMatchObject({
+      recoveryReason: RECOVERY_REASONS.DIRTY_REWARD_RESCORE,
+    });
+  });
+
   it("init() returns immediately even when a stale orphan's recovery chain stalls (issue #1808)", async () => {
     // Issue #1808: on databases with 30k+ traces, the dreaming chain
     // synchronous-await inside `init()` blocked the OpenClaw Gateway's