diff --git a/apps/memos-local-plugin/bridge.cts b/apps/memos-local-plugin/bridge.cts index 449c4a707..eaf7a8b9d 100644 --- a/apps/memos-local-plugin/bridge.cts +++ b/apps/memos-local-plugin/bridge.cts @@ -353,72 +353,14 @@ async function main(): Promise { | ReturnType["startHeartbeat"]> | undefined; - // ─── Startup ordering invariant (issue #1747 + host LLM fallback) ─── - // - // `startStdioServer({ core })` MUST run before `await core.init()`. - // Two independent failure modes if this ordering is reversed: - // - // 1. Host LLM fallback (original motivation for this ordering): - // `core.init()` may recover dirty episodes and run - // reflection/reward/L2/skill work; if that work hits a broken - // primary skill-evolver model, the LLM facade can fall back to - // host before init returns. Starting stdio first gives the - // fallback a transport instead of tripping the lazy bridge guard. - // - // 2. Python adapter `session.open` timeout (issue #1747): - // `core.init()` synchronously scans `episodes WHERE status='open'` - // and recovers stale rows via `recoverOpenEpisodesAsSessionEnd` - // + `recoverDirtyClosedEpisodes` — both of which call the LLM - // and routinely take 10-60+ seconds when a previous chat left - // orphan episodes behind. The Hermes Python adapter's - // `_open_session()` default timeout is 30 s. If stdio starts - // after init, the parent writes `session.open` into the bridge's - // stdin and the Python side gets `asyncio.TimeoutError` before - // the read loop is attached. By starting stdio first, the read - // loop is alive immediately — `core.openSession()` is safe to - // serve pre-init because it depends only on the SQLite handle - // and event bus that `bootstrapMemoryCoreFull()` already - // provisioned. (`ensureLive()` only blocks on `shutDown`, not - // on `initialized`.) - // - // The invariant is pinned by - // `tests/unit/bridge/bridge-startup-ordering.test.ts`. - if (!args.daemon) { - stdio = startStdioServer({ core }); - bridgeStatus?.markConnected(); - bridgeHeartbeat = bridgeStatus?.startHeartbeat(); - void stdio.done.then(() => { - bridgeHeartbeat?.stop(); - bridgeStatus?.markDisconnected("Hermes chat disconnected"); - }); - } - - try { - await core.init(); - } catch (err) { - bridgeHeartbeat?.stop(); - if (stdio) { - try { - await stdio.close(); - } catch { - /* best-effort */ - } - } - throw err; - } - // ─── Daemon mode ────────────────────────────────────────────── - // When started with `--daemon`, skip stdio and run as a pure HTTP - // viewer daemon. Used by install.sh (post-install) and admin/restart - // (self-restart) to keep the Memory Viewer always available. + // When started with `--daemon`, bind the viewer port BEFORE running + // core.init() so that ensure_viewer_daemon()'s 15-second health probe + // succeeds immediately. Without this, a lengthy dirty-episode rescore + // inside core.init() keeps the port unbound past the probe deadline, + // causing ensure_viewer_daemon() to give up and spawn a replacement + // daemon — which kills this one and restarts the cycle. if (args.daemon) { - // Daemon mode is the target of `POST /api/v1/admin/restart`, - // which re-spawns the bridge after a short sleep. On busy - // machines the previous bridge's listening socket can take a - // moment longer than expected to release, so we retry the bind - // a few times before giving up. Without this the user sees - // "重启超时" in the viewer because the new daemon raced its - // predecessor and lost. let viewer: import("./server/types.js").ServerHandle | null = null; const maxBindAttempts = 10; for (let attempt = 1; attempt <= maxBindAttempts; attempt++) { @@ -475,10 +417,53 @@ async function main(): Promise { }; process.on("SIGINT", () => void shutdownDaemon("SIGINT")); process.on("SIGTERM", () => void shutdownDaemon("SIGTERM")); + + // Run core.init() in the background. A dirty-episode rescore can + // take minutes; keeping it async lets the HTTP server stay responsive + // to health probes throughout and prevents ensure_viewer_daemon() + // from timing out and spawning a replacement daemon. + void core.init().catch((err) => { + process.stderr.write( + `bridge: daemon core.init error: ${err instanceof Error ? err.stack ?? err.message : String(err)}\n`, + ); + void shutdownDaemon("init.error"); + }); + // Process stays alive via the HTTP server's ref'd socket. return; } + // ─── Non-daemon setup ───────────────────────────────────────── + // In stdio mode the host fallback path is a reverse JSON-RPC request + // over the same pipe as normal bridge traffic. `core.init()` may + // recover dirty episodes and run reflection/reward/L2/skill work; if + // that work hits a broken primary skill-evolver model, the LLM facade + // can fall back to host before init returns. Start stdio first so that + // fallback has a transport instead of tripping the lazy bridge guard. + if (!args.daemon) { + stdio = startStdioServer({ core }); + bridgeStatus?.markConnected(); + bridgeHeartbeat = bridgeStatus?.startHeartbeat(); + void stdio.done.then(() => { + bridgeHeartbeat?.stop(); + bridgeStatus?.markDisconnected("Hermes chat disconnected"); + }); + + try { + await core.init(); + } catch (err) { + bridgeHeartbeat?.stop(); + if (stdio) { + try { + await stdio.close(); + } catch { + /* best-effort */ + } + } + throw err; + } + } + // ─── Normal (stdio) mode ────────────────────────────────────── // The stdio handle was started before `core.init()` above so host // fallback is available during startup recovery. diff --git a/apps/memos-local-plugin/core/pipeline/memory-core.ts b/apps/memos-local-plugin/core/pipeline/memory-core.ts index 2ae14f053..c66aa2e77 100644 --- a/apps/memos-local-plugin/core/pipeline/memory-core.ts +++ b/apps/memos-local-plugin/core/pipeline/memory-core.ts @@ -628,12 +628,9 @@ export function createMemoryCore( if (nowMs - lastDirtyClosedScan < 30_000) return; lastDirtyClosedScan = nowMs; try { - const allDirty = handle.repos.episodes - .list({ status: "closed", limit: 500 }) - .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. - const dirtyClosed = allDirty.filter((ep) => dirtyEpisodeBackoffElapsed(ep, nowMs)); + const dirtyClosed = collectDirtyClosedEpisodes(nowMs); if (dirtyClosed.length > 0) { await recoverDirtyClosedEpisodes(dirtyClosed); } @@ -954,24 +951,7 @@ export function createMemoryCore( staleForBackground = stale; } const nowForDirty = Date.now(); - const allDirty = handle.repos.episodes - .list({ status: "closed", limit: 500 }) - .filter((ep) => !isLightweightEpisode(ep) && episodeRewardIsDirty(ep)); - const dirtyClosed: typeof allDirty = []; - for (const ep of allDirty) { - if (dirtyEpisodeBackoffElapsed(ep, nowForDirty)) { - dirtyClosed.push(ep); - } else { - const dirtyMeta = (ep.meta?.rewardDirty as - | { failedAttempts?: number; lastFailureAt?: number } - | undefined) ?? {}; - log.debug("init.dirty_closed_episodes.skip_backoff", { - episodeId: ep.id, - failedAttempts: dirtyMeta.failedAttempts ?? 0, - lastFailureAt: dirtyMeta.lastFailureAt ?? 0, - }); - } - } + const dirtyClosed = collectDirtyClosedEpisodes(nowForDirty); dirtyClosedForBackground = dirtyClosed; } catch (err) { log.debug("init.orphan_scan.failed", { @@ -1341,11 +1321,12 @@ export function createMemoryCore( episodes: Array }>, ): Promise { log.info("init.dirty_closed_episodes.rescore", { count: episodes.length }); + const rescored: EpisodeId[] = []; // Snapshot the prior failure counters so we can increment them later // (after the bus chain settles) without an extra DB read. const priorFailedAttempts = new Map(); for (const ep of episodes) { - if (isLightweightEpisode(ep)) continue; + if (isLightweightEpisode(ep) && handle.algorithm.lightweightMemory.enabled) continue; const episodeId = ep.id as EpisodeId; const endedAt = ep.endedAt ?? Date.now(); const prevDirty = (ep.meta?.rewardDirty as @@ -1364,11 +1345,28 @@ export function createMemoryCore( const snapshot = snapshotFromRecoveredEpisode(ep, endedAt, { recoveryReason: "dirty_reward_rescore", }); + // If the episode was tagged lightweight during a prior session but + // lightweight mode is now off, clear the flag so the capture subscriber + // doesn't skip it. + if (snapshot.meta?.lightweightMemory === true && !handle.algorithm.lightweightMemory.enabled) { + delete (snapshot.meta as Record).lightweightMemory; + } handle.buses.session.emit({ kind: "episode.finalized", episode: snapshot, closedBy: "finalized", }); + rescored.push(episodeId); + } + // Drain the capture pass (patches reflections + α onto existing traces). + await handle.flush(); + // In lightweight mode flush() returns before draining the reward + // subscriber. Explicitly run reward for any episode whose trace count + // still mismatches — mirrors the pattern in recoverOpenEpisodesAsSessionEnd. + for (const episodeId of rescored) { + if (episodeRewardIsDirty(handle.repos.episodes.getById(episodeId) ?? {} as never)) { + await handle.rewardRunner.run({ episodeId, feedback: [], trigger: "manual" }); + } } await handle.flush(); // After the reward / reflect chain has finished, account for the @@ -1398,14 +1396,53 @@ export function createMemoryCore( } } + function collectDirtyClosedEpisodes( + nowMs?: number, + ): (EpisodeRow & { meta?: Record })[] { + const dirty: (EpisodeRow & { meta?: Record })[] = []; + let offset = 0; + const pageSize = 500; + while (true) { + const page = handle.repos.episodes.list({ status: "closed", limit: pageSize, offset }); + for (const ep of page) { + if (isLightweightEpisode(ep)) continue; + if (!episodeRewardIsDirty(ep)) continue; + if (nowMs === undefined || dirtyEpisodeBackoffElapsed(ep, nowMs)) { + dirty.push(ep); + } else { + const dirtyMeta = (ep.meta?.rewardDirty as + | { failedAttempts?: number; lastFailureAt?: number } + | undefined) ?? {}; + log.debug("init.dirty_closed_episodes.skip_backoff", { + episodeId: ep.id, + failedAttempts: dirtyMeta.failedAttempts ?? 0, + lastFailureAt: dirtyMeta.lastFailureAt ?? 0, + }); + } + } + if (page.length < pageSize) break; + offset += pageSize; + } + return dirty; + } + function episodeRewardIsDirty(ep: EpisodeRow & { meta?: Record }): boolean { const meta = ep.meta ?? {}; - if (meta.lightweightMemory === true) return false; + if (meta.lightweightMemory === true && handle.algorithm.lightweightMemory.enabled) return false; if (meta.rewardDirty && typeof meta.rewardDirty === "object") return true; const reward = meta.reward; if (reward && typeof reward === "object" && (reward as { skipped?: unknown }).skipped === true) { - return false; + // Abandoned episodes with no prior recovery attempt get one retry: the + // skip decision may have been made with incomplete data before the session + // ended. recoverDirtyClosedEpisodes() patches closeReason → "finalized" + // after processing, so if reward skips again the next check sees + // closeReason !== "abandoned" and stops retrying (no loop). + const isAbandonedNoRecovery = + meta.closeReason === "abandoned" && meta.recoveryReason == null; + if (!isAbandonedNoRecovery) { + return false; + } } if ( ep.rTask == null && diff --git a/apps/memos-local-plugin/core/reward/reward.ts b/apps/memos-local-plugin/core/reward/reward.ts index 1b6d3354c..027689cfd 100644 --- a/apps/memos-local-plugin/core/reward/reward.ts +++ b/apps/memos-local-plugin/core/reward/reward.ts @@ -119,6 +119,7 @@ export function createRewardRunner(deps: RewardDeps): RewardRunner { try { const existingMeta = episode.meta ?? {}; const wasFinalized = existingMeta.closeReason === "finalized"; + deps.episodesRepo.setRTask(input.episodeId, 0); deps.episodesRepo.updateMeta(input.episodeId, { ...(wasFinalized ? {} : { closeReason: "abandoned", abandonReason: skipReason }), reward: { @@ -127,7 +128,9 @@ export function createRewardRunner(deps: RewardDeps): RewardRunner { scoredAt: startedAt, trigger: input.trigger, skipped: true, + traceCount: traceIds.length, }, + rewardDirty: undefined, }); } catch (err) { warnings.push({ diff --git a/apps/memos-local-plugin/core/storage/migrations/013-fix-traces-fts-triggers.sql b/apps/memos-local-plugin/core/storage/migrations/013-fix-traces-fts-triggers.sql new file mode 100644 index 000000000..1f898b975 --- /dev/null +++ b/apps/memos-local-plugin/core/storage/migrations/013-fix-traces-fts-triggers.sql @@ -0,0 +1,54 @@ +-- Rebuild traces_fts and its triggers after the a054c9b8 dedup inadvertently +-- introduced two bugs: +-- +-- 1. The FTS column was renamed from `trace_id` to `id`, breaking the +-- `JOIN traces t ON t.id = f.trace_id` in repos/traces.ts. +-- 2. The UPDATE trigger used `INSERT INTO traces_fts(traces_fts, ...) VALUES('delete', ...)` +-- — the FTS5 special 'delete' command — which only works for external-content +-- or contentless tables. On a regular FTS5 table it throws "SQL logic error", +-- causing every traces UPDATE (including score writes) to fail silently. +-- +-- Fix: drop and rebuild the FTS table with the canonical `trace_id` column +-- and correct direct-DELETE trigger syntax, matching the original 001-initial.sql +-- intent and the TS query in core/storage/repos/traces.ts. + +DROP TRIGGER IF EXISTS traces_fts_ai; +DROP TRIGGER IF EXISTS traces_fts_ad; +DROP TRIGGER IF EXISTS traces_fts_au; +DROP TABLE IF EXISTS traces_fts; + +CREATE VIRTUAL TABLE traces_fts USING fts5( + trace_id UNINDEXED, + user_text, + agent_text, + summary, + reflection, + tags, + tokenize = 'trigram' +); + +INSERT INTO traces_fts(rowid, trace_id, user_text, agent_text, summary, reflection, tags) +SELECT rowid, id, + COALESCE(user_text, ''), + COALESCE(agent_text, ''), + COALESCE(summary, ''), + COALESCE(reflection, ''), + COALESCE(tags_json, '') +FROM traces; + +CREATE TRIGGER traces_fts_ai AFTER INSERT ON traces BEGIN + INSERT INTO traces_fts(rowid, trace_id, user_text, agent_text, summary, reflection, tags) + VALUES (new.rowid, new.id, new.user_text, new.agent_text, + COALESCE(new.summary,''), COALESCE(new.reflection,''), new.tags_json); +END; + +CREATE TRIGGER traces_fts_ad AFTER DELETE ON traces BEGIN + DELETE FROM traces_fts WHERE trace_id = old.id; +END; + +CREATE TRIGGER traces_fts_au AFTER UPDATE ON traces BEGIN + DELETE FROM traces_fts WHERE trace_id = old.id; + INSERT INTO traces_fts(rowid, trace_id, user_text, agent_text, summary, reflection, tags) + VALUES (new.rowid, new.id, new.user_text, new.agent_text, + COALESCE(new.summary,''), COALESCE(new.reflection,''), new.tags_json); +END; 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 7f9d9a0fa..3c4bfaafd 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 @@ -1518,6 +1518,171 @@ algorithm: expect(meta.reward?.traceIds).toEqual(["tr_missing_reward"]); }); + it("dirty-reward recovery does not insert orphan traces (regression: rescore loop guard)", async () => { + home = await makeTmpHome({ + agent: "openclaw", + configYaml: FULL_MEMORY_CONFIG_YAML, + }); + + const seeder = await bootstrapMemoryCore({ + agent: "openclaw", + home: home.home, + config: home.config, + pkgVersion: "rescore-loop-seed", + }); + await seeder.init(); + await seeder.shutdown(); + + const Sqlite = (await import("better-sqlite3")).default; + const writeDb = new Sqlite(home.home.dbFile); + const base = Date.now() - 5_000; + + writeDb + .prepare( + `INSERT INTO sessions (id, agent, started_at, last_seen_at, meta_json) VALUES (?, ?, ?, ?, ?)`, + ) + .run("se_loop", "openclaw", base, base, "{}"); + + // Episode is dirty: traceCount=1 but trace_ids_json has 2 IDs. + writeDb + .prepare( + `INSERT INTO episodes (id, session_id, started_at, ended_at, trace_ids_json, r_task, status, meta_json) VALUES (?, ?, ?, ?, ?, ?, 'closed', ?)`, + ) + .run( + "ep_loop", + "se_loop", + base, + base + 1, + JSON.stringify(["tr_loop_a", "tr_loop_b"]), + 0.5, + JSON.stringify({ + closeReason: "finalized", + reward: { rHuman: 0.5, scoredAt: base - 1_000, traceCount: 1 }, + }), + ); + + 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_loop_a", + "ep_loop", + "se_loop", + base, + "帮我分析一下这段Python代码的性能瓶颈，并给出优化建议。", + "这段代码的主要性能问题在于嵌套循环，时间复杂度是O(n²)，可以用哈希表将其优化到O(n)。", + "Python代码性能分析", + "[]", + null, + null, + 0, + 0, + null, + 0.5, + "[]", + "[]", + base, + 1, + ); + + // The tool call ends at a timestamp that does not match any existing + // trace row, which used to create a synthetic orphan trace. + const toolCallWithDifferentTs = JSON.stringify([ + { + name: "bash", + input: { command: "python -c 'import cProfile; cProfile.run(\"main()\")'" }, + output: "ncalls tottime ... main 1 0.003", + endedAt: base + 300, + }, + ]); + 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_loop_b", + "ep_loop", + "se_loop", + base + 100, + "请用cProfile验证一下", + "运行结果确认了瓶颈在内层循环，优化后耗时减少了约80%。", + "cProfile性能验证", + toolCallWithDifferentTs, + null, + null, + 0, + 0, + null, + 0.5, + "[]", + "[]", + base + 100, + 1, + ); + writeDb.close(); + + core = await bootstrapMemoryCore({ + agent: "openclaw", + home: home.home, + config: home.config, + pkgVersion: "rescore-loop-recover-1", + }); + await core.init(); + await core.waitForStartupRecovery?.(); + await core.shutdown(); + core = null; + + const readDb1 = new Sqlite(home.home.dbFile, { readonly: true }); + const ep1 = readDb1 + .prepare("SELECT trace_ids_json, meta_json FROM episodes WHERE id = ?") + .get("ep_loop") as { trace_ids_json: string; meta_json: string } | undefined; + readDb1.close(); + + expect(ep1).toBeDefined(); + const ids1 = JSON.parse(ep1!.trace_ids_json) as string[]; + expect(ids1.length).toBe(2); + const meta1 = JSON.parse(ep1!.meta_json) as { + recoveryReason?: string; + reward?: { traceCount?: number }; + }; + expect(meta1.recoveryReason).toBe("dirty_reward_rescore"); + expect(meta1.reward?.traceCount).toBe(2); + + core = await bootstrapMemoryCore({ + agent: "openclaw", + home: home.home, + config: home.config, + pkgVersion: "rescore-loop-recover-2", + }); + await core.init(); + await core.waitForStartupRecovery?.(); + + const readDb2 = new Sqlite(home.home.dbFile, { readonly: true }); + const ep2 = readDb2 + .prepare("SELECT trace_ids_json, meta_json FROM episodes WHERE id = ?") + .get("ep_loop") as { trace_ids_json: string; meta_json: string } | undefined; + readDb2.close(); + + expect(ep2).toBeDefined(); + const ids2 = JSON.parse(ep2!.trace_ids_json) as string[]; + expect(ids2.length).toBe(2); + const meta2 = JSON.parse(ep2!.meta_json) as { + reward?: { traceCount?: number }; + }; + expect(meta2.reward?.traceCount).toBe(2); + }); + 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