fix(plugin): paginate dirty-closed reward recovery

[chiefmojo]

Description

Resubmits only the paginated dirty-closed recovery portion of #1784 against current main.

episodes.list({ status: "closed", limit: 500 }) is hard-capped at 500 rows, so closed episodes beyond the newest 500 never reach the startup or periodic dirty-reward scan. This change adds a stable (started_at, id) keyset page, processes one 500-row page per scan, and persists the page cursor in the existing kv store so restarts continue toward older episodes rather than repeatedly rescanning the newest page.

No dependency or schema changes are required.

Related Issue: #1782

Type of change

• Bug fix (non-breaking change which fixes an issue)

How Has This Been Tested?

• Unit Test
  • npx vitest run tests/unit/pipeline/memory-core.test.ts -t 'continues a bounded dirty-closed scan past 500 rows across restart'
  • The regression seeds 500 newer closed rows, inserts another newer row between restarts, and verifies that the dirty row beyond the first page is still recovered on the next scan.
• Test Script Or Test Steps
  • Synthetic 30,000-row SQLite measurement: one bounded page returned 500 rows in 0.648 ms; the former 60-page OFFSET traversal returned 30,000 rows in 64.488 ms (raw row retrieval on this machine).
• Pipeline Automated API Test (not applicable; storage/pipeline-only change)

Additional checks:

• npm run lint
• npx vitest run tests/unit/pipeline/memory-core.test.ts (34 passed)
• make format (Ruff checks passed; 601 files unchanged)

npm run test:unit currently has two unrelated baseline failures outside this diff: the startup-recovery source-shape assertion expects a removed scheduleStartupRecovery(...) helper, and the namespace-visibility migrator fixture inserts traces.role, which is absent from the current schema.

Checklist

• I have performed a self-review of my own code | 我已自行检查了自己的代码
• I have commented my code in hard-to-understand areas | 我已在难以理解的地方对代码进行了注释
• I have added tests that prove my fix is effective or that my feature works | 我已添加测试以证明我的修复有效或功能正常
• I have created related documentation issue/PR in MemOS-Docs (if applicable) | 不适用
• I have linked the issue to this PR (if applicable) | 我已将 issue 链接到此 PR（如果适用）
• I have mentioned the person who will review this PR | @shinetata

Reviewer Checklist

• closes #xxxx (Replace xxxx with the GitHub issue number)
• Made sure Checks passed
• Tests have been provided

[Memtensor-AI]

🤖 Open Code Review

Target: PR #2120
Task: 9044bbf30d61be85
Base: main
Head: fix/dirty-closed-keyset-scan

🔍 OpenCodeReview found 3 issue(s) in this PR.

---

1. apps/memos-local-plugin/core/storage/repos/episodes.ts (L189-L206)

Performance: db.prepare() is called on every invocation of listClosedPage().

Unlike the static prepared statements defined at the top of makeEpisodesRepo() (e.g. insert, replace, selectById), this method dynamically calls db.prepare(...) on each invocation. Since there are only two distinct SQL variants (with and without the cursor before clause), you can pre-prepare both statements once and cache them, avoiding repeated parsing and compilation overhead — especially since this function is called frequently from background scan loops.

Suggested approach: prepare two statements at initialization time, e.g.:

const listClosedAll = db.prepare<{ status: string }, RawEpisodeRow>(
  `SELECT ${COLUMNS.join(", ")} FROM episodes WHERE status = @status ORDER BY started_at DESC, id DESC LIMIT 500`
);
const listClosedBefore = db.prepare<{ status: string; before_started_at: number; before_id: string }, RawEpisodeRow>(
  `SELECT ${COLUMNS.join(", ")} FROM episodes WHERE status = @status AND (started_at < @before_started_at OR (started_at = @before_started_at AND id < @before_id)) ORDER BY started_at DESC, id DESC LIMIT 500`
);

Then select between them in listClosedPage based on whether opts.before is provided.

---

2. apps/memos-local-plugin/core/pipeline/memory-core.ts (L1532-L1557)

The single shared KV cursor is advanced by both init() and the periodic autoRescoreDirtyClosedEpisodes(), causing them to interfere with each other.

Scenario: On startup, init() calls collectDirtyClosedEpisodes() and, if there are exactly 500 closed episodes, saves a cursor pointing past page 1. The next call from autoRescoreDirtyClosedEpisodes() (~30 s later) will read that cursor and start from page 2 — skipping all episodes on page 1 entirely. Dirty episodes that were on the first page are silently missed by the periodic rescan.

Conversely, if init() fully exhausts all pages and clears the cursor, the periodic scan always restarts from the very beginning, never progressing through the tail on large databases — defeating the stated purpose of the cursor.

Because both callers write to the same DIRTY_CLOSED_SCAN_CURSOR_KEY, their states bleed into each other. Consider one of:

• Use separate cursor keys for init and the periodic scan (e.g. pipeline.dirty_closed_scan_cursor.init.v1 vs pipeline.dirty_closed_scan_cursor.periodic.v1).
• Or maintain the cursor only in the periodic scan path (keeping init() doing a full scan as before), since init() is a one-shot startup operation where completeness matters more than pagination.

💡 Suggested Change

Before:

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

After:

  const DIRTY_CLOSED_SCAN_CURSOR_KEY_INIT     = "pipeline.dirty_closed_scan_cursor.init.v1";
  const DIRTY_CLOSED_SCAN_CURSOR_KEY_PERIODIC = "pipeline.dirty_closed_scan_cursor.periodic.v1";

  // Pass the correct key depending on the call site:
  function collectDirtyClosedEpisodes(
    cursorKey: string = DIRTY_CLOSED_SCAN_CURSOR_KEY_PERIODIC,
  ): Array<EpisodeRow & { meta?: Record<string, unknown> }> {
    const storedCursor = handle.repos.kv.get<unknown>(cursorKey, null);
    const cursor = isDirtyClosedScanCursor(storedCursor) ? storedCursor : null;
    if (storedCursor !== null && !cursor) {
      handle.repos.kv.del(cursorKey);
    }
    const page = handle.repos.episodes.listClosedPage({
      limit: DIRTY_CLOSED_SCAN_PAGE_SIZE,
      before: cursor,
    });
    if (page.length === 0) {
      handle.repos.kv.del(cursorKey);
      return [];
    }
    const last = page[page.length - 1]!;
    if (page.length < DIRTY_CLOSED_SCAN_PAGE_SIZE) {
      handle.repos.kv.del(cursorKey);
    } else {
      handle.repos.kv.set(cursorKey, { startedAt: last.startedAt, id: last.id });
    }
    return page;
  }

---

3. apps/memos-local-plugin/core/pipeline/memory-core.ts (L1542-L1545)

When page.length === 0 and cursor is already null (no stored cursor / fresh start), kv.del is called on a key that doesn't exist. This is likely a no-op, but more importantly: an empty result here means all pages have been consumed and the scan should wrap around. There is no logic to reset the cursor for the next full-scan cycle — after the last page returns empty the cursor is deleted, so the next call correctly starts from the top. This part is fine on its own, but it is worth a comment to make the wrap-around intent explicit.

Generated by cloud-assistant via Open Code Review.

[chiefmojo]

Follow-up commit 26aa9f24 now imports and uses the storage-layer ClosedEpisodeCursor type for the persisted dirty-scan cursor guard.

I also verified the other two automated findings against the call flow: the cursor advances one bounded page per startup/periodic scan and is intentionally cleared only after a complete pass; the page size is exactly the repository's enforced 500-row cap.

Validation: npm run lint, npx vitest run tests/unit/pipeline/memory-core.test.ts (34 passed), and make format.

[Memtensor-AI]

✅ Automated Test Results: PASSED

All tests passed (44/44 executed). memos_local_plugin/unit: 44/44. Duration: 11s [advisory, non-gating] AI-generated tests on branch test/auto-gen-8c86a9867e66394a-20260718224733: 60/60 passed — these do NOT affect the PR verdict; review the branch manually.

Branch: fix/dirty-closed-keyset-scan

[Memtensor-AI]

✅ Automated Test Results: PASSED

All tests passed (44/44 executed). memos_local_plugin/unit: 44/44. Duration: 11s

Branch: fix/dirty-closed-keyset-scan