diff --git a/apps/memos-local-plugin/core/pipeline/memory-core.ts b/apps/memos-local-plugin/core/pipeline/memory-core.ts index 96721cc0b..17c896f3f 100644 --- a/apps/memos-local-plugin/core/pipeline/memory-core.ts +++ b/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 }> { + const storedCursor = handle.repos.kv.get(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 }): boolean { const meta = ep.meta ?? {}; if (meta.lightweightMemory === true) return false; diff --git a/apps/memos-local-plugin/core/storage/repos/episodes.ts b/apps/memos-local-plugin/core/storage/repos/episodes.ts index 1322b2afb..78b49d5fc 100644 --- a/apps/memos-local-plugin/core/storage/repos/episodes.ts +++ b/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; } +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(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 { + const limit = clampLimit(opts.limit ?? 500); + const params: Record = { 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(sql).all(params).map(mapRow); + }, + count(filter: Omit = {}): number { const tr = timeRangeWhere(filter, "started_at"); const fragments: string[] = []; diff --git a/apps/memos-local-plugin/tests/unit/pipeline/memory-core.test.ts b/apps/memos-local-plugin/tests/unit/pipeline/memory-core.test.ts index 62c18fa16..903a53c2e 100644 --- a/apps/memos-local-plugin/tests/unit/pipeline/memory-core.test.ts +++ b/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