[akushonkamen]

Description

Fixes #2459 — two reliability gaps in the memos-local-plugin dirty-closed rescan, plus one follow-up guard found in review. All changes are in apps/memos-local-plugin/core/pipeline/memory-core.ts with new unit tests; both commits are based on dev-v2.0.36 (8511502).

Root cause

• Gap 1 (cursor starvation): collectDirtyClosedEpisodes() paged only downward via listClosedPage({ before: cursor }) and only reset the cursor on a short page, so an episode that became dirty after the cursor paged past its position was never revisited until the walk reached the end of the closed set and the cursor reset — a full lap of ceil(closed/500) × 10 min (~70 min on the reporter's DB, hours on the 30k+-episode DBs of v2.0.2+ regression: bootstrapMemoryCoreFull() hangs with 100% CPU on databases >500MB #1787).
• Gap 2 (interrupted drain lost the whole batch's phase-2): recoverDirtyClosedEpisodes() returned right after await handle.flush() when startupRecoveryCancelled was set, skipping settlement for every queued episode: successes kept their rewardDirty markers (re-collected and re-run later — duplicate LLM spend) and failures lost their failedAttempts backoff counters.

Fix

• Gap 1: every scan first takes one bounded newest-first head page (before: null) merged into an id-deduplicated collected map, then continues the cursor walk (core/pipeline/memory-core.ts:1788-1831). Cursor advancement still derives from the walk page only (core/pipeline/memory-core.ts:1823-1831), so head-page rows do not disturb cursor semantics; freshly closed / freshly re-marked rows are now seen on the very next tick.
• Gap 2: phase-2 settlement moved into try { flush } finally with one try/catch per episode (core/pipeline/memory-core.ts:1700-1775) — cancel or a flush throw mid-teardown still settles each row (clear rewardDirty on success, bump failedAttempts + lastFailureAt on failure); a per-row error logs init.dirty_closed_episodes.settle_failed and cannot drop the rest of the batch.
• Review follow-up (second commit): recoverDirtyClosedEpisodes records each emitted episode id in a process-local in-flight set before emitting and clears it as soon as that row settles; both auto-collection sites (init classification core/pipeline/memory-core.ts:811-825 and the 10-min periodic rescan core/pipeline/memory-core.ts:1215-1227) filter in-flight ids, so a tick racing an open drain can no longer re-collect and re-emit the whole batch (duplicate LLM spend). Freshly dirtied rows are not blocked — per-episode guard, not a whole-scan mutex.
• init.orphan_scan.failed raised from debug to warn (core/pipeline/memory-core.ts:1219-1226): a skipped startup classification used to be invisible at default log levels.

AI assistance disclosure: this change was produced with AI assistance in an automated contributor workflow; every command and test result below was executed locally as recorded.

Type of change

Please delete options that are not relevant.

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

How Has This Been Tested?

• Unit Test
• Test Script Or Test Steps (please provide)

Environment: macOS (arm64), Node + vitest 2.1.9, run from apps/memos-local-plugin.

Reproduce:

cd apps/memos-local-plugin
# targeted issue-#2459 cases (2 new in commit 1, 1 more in commit 2)
npx vitest run tests/unit/pipeline/memory-core.test.ts --reporter=dot -t "issue #2459"
# regression subset
npx vitest run tests/unit/pipeline/memory-core.test.ts tests/unit/startup-recovery.test.ts tests/unit/reward/ --reporter=dot
# full unit / integration / typecheck
npx vitest run tests/unit --reporter=dot
npx vitest run tests/integration --reporter=dot
npx tsc -p tsconfig.json --noEmit

Results (final verification at this branch head):

• Targeted issue-[Bug] dirty-closed rescan misses freshly-marked episodes for a full cursor lap; interrupted drains lose phase-2 accounting #2459 tests: 3 passed | 48 skipped.
• Before the fix (same targeted command, pre-fix tree): 2 failed | 48 skipped — Gap 1 (recoveryReason undefined on the row dirtied above the cursor) and Gap 2 (rewardDirty markers unsettled after cancelled drain).
• Regression subset (pipeline + startup-recovery + reward): 8 files / 97 passed.
• Full unit suite: 181 files / 1592 passed | 2 skipped (~58s).
• Integration suite: 4 files / 6 passed.
• Repo typecheck/lint (npx tsc -p tsconfig.json --noEmit): exit 0.

Note: the ticket's vitest … -q flag does not exist in vitest 2.1.9 (CACError: Unknown option -q), so --reporter=dot was used instead. CI is expected red on this scope for external PRs due to the known environment issue tracked in #2476 (autotest environment preparation fails); local green evidence is provided above.

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) | 我已在 MemOS-Docs 中创建了相关的文档 issue/PR（如果适用）
• I have linked the issue to this PR (if applicable) | 我已将 issue 链接到此 PR（如果适用）
• I have mentioned the person who will review this PR | 我已提及将审查此 PR 的人

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 #2482
Task: de02b4a723224271
Base: dev-v2.0.36
Head: fix/2459-dirty-closed-rescan

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

---

1. apps/memos-local-plugin/core/pipeline/memory-core.ts (L1818-L1821)

Walk-termination check tests the cursor-page length, not the merged collection size. When the cursor page returns 0 rows but the head-page pass already populated collected with freshly-dirtied entries, this branch deletes the persisted cursor and returns an empty array, silently discarding every head-page episode. Their reward chains are never run, the rewardWasSkipped cleanup is skipped, and the cursor resets so the backlog walk restarts from the newest row — exactly the latency regression this change was written to eliminate.

Fix: replace page.length === 0 with collected.size === 0.

💡 Suggested Change

Before:

    if (page.length === 0) {
      handle.repos.kv.del(DIRTY_CLOSED_SCAN_CURSOR_KEY);
      return [];
    }

After:

    if (collected.size === 0) {
      handle.repos.kv.del(DIRTY_CLOSED_SCAN_CURSOR_KEY);
      return [];
    }

---

2. apps/memos-local-plugin/core/pipeline/memory-core.ts (L1708-L1728)

dirtyRescoreInFlight.add(episodeId) is called inside the emit loop, but the finally block that calls dirtyRescoreInFlight.delete(episodeId) only runs after the try { await handle.flush() } block that follows the loop. Any synchronous throw between the add and the opening try (e.g. handle.repos.episodes.updateMeta or handle.buses.session.emit throwing) exits the loop without ever entering the try/finally, so the delete is never called. The episode ID stays in the Set for the entire lifetime of the process, silently suppressing all future rescoring for that episode with no TTL or sweep to recover it.

Fix: wrap the per-episode registration and emit in its own try/catch so a throw removes the ID immediately, or move the try { await handle.flush() } finally { … } to cover the entire emit loop:

💡 Suggested Change

Before:

      priorFailedAttempts.set(episodeId, prevAttempts);
      // Mark in-flight BEFORE emitting so a periodic tick racing this
      // drain cannot queue a duplicate run for the same row.
      dirtyRescoreInFlight.add(episodeId);
      handle.repos.episodes.updateMeta(episodeId, {
        closeReason: "finalized",
        recoveredAtStartup: endedAt,
        recoveryReason: RECOVERY_REASONS.DIRTY_REWARD_RESCORE,
      });
      const snapshot = snapshotFromRecoveredEpisode(ep, endedAt, {
        recoveryReason: RECOVERY_REASONS.DIRTY_REWARD_RESCORE,
      });
      handle.buses.session.emit({
        kind: "episode.finalized",
        episode: snapshot,
        closedBy: "finalized",
      });
    }
    try {
      await handle.flush();
    } finally {

After:

      priorFailedAttempts.set(episodeId, prevAttempts);
      // Mark in-flight BEFORE emitting so a periodic tick racing this
      // drain cannot queue a duplicate run for the same row.
      dirtyRescoreInFlight.add(episodeId);
      try {
        handle.repos.episodes.updateMeta(episodeId, {
          closeReason: "finalized",
          recoveredAtStartup: endedAt,
          recoveryReason: RECOVERY_REASONS.DIRTY_REWARD_RESCORE,
        });
        const snapshot = snapshotFromRecoveredEpisode(ep, endedAt, {
          recoveryReason: RECOVERY_REASONS.DIRTY_REWARD_RESCORE,
        });
        handle.buses.session.emit({
          kind: "episode.finalized",
          episode: snapshot,
          closedBy: "finalized",
        });
      } catch (err) {
        // Registration failed before the flush — remove immediately so
        // the episode is not permanently excluded from future scans.
        dirtyRescoreInFlight.delete(episodeId);
        priorFailedAttempts.delete(episodeId);
        log.warn("init.dirty_closed_episodes.emit_failed", {
          episodeId,
          err: err instanceof Error ? err.message : String(err),
        });
      }
    }
    try {
      await handle.flush();
    } finally {

---

3. apps/memos-local-plugin/core/pipeline/memory-core.ts (L813-L821)

The three-predicate dirty-collection chain (!isLightweightEpisode && episodeRewardIsDirty && !dirtyRescoreInFlight.has) is copy-pasted verbatim at this call site and at the init-time scan (~line 1217). Any future change to in-flight semantics — adding a TTL check, logging skipped IDs, adding a per-reason exclusion — must be applied to both sites independently and will silently drift otherwise.

Extract a shared helper, for example:

function isCollectableDirtyClosedEpisode(ep: EpisodeRow & { meta?: Record<string, unknown> }): boolean {
  return !isLightweightEpisode(ep)
    && episodeRewardIsDirty(ep)
    && !dirtyRescoreInFlight.has(ep.id as EpisodeId);
}

and call it at both sites.

Generated by cloud-assistant via Open Code Review.

[Memtensor-AI]

⚠️ Automated Test Results: ENV ISSUE

The test environment encountered an issue that requires manual attention.

Details: Environment preparation failed before any gating tests executed. Failed scopes: memos_local_plugin
Branch: fix/2459-dirty-closed-rescan