fix: reward pipeline skips abandoned episodes (3 related fixes)

apps/memos-local-plugin/bridge.cts

@@ -353,72 +353,14 @@ async function main(): Promise<void> {
     | ReturnType<NonNullable<typeof bridgeStatus>["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<void> {
     };
     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.

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<EpisodeRow & { meta?: Record<string, unknown> }>,
   ): Promise<void> {
     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<EpisodeId, number>();
     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<string, unknown>).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<string, unknown> })[] {
+    const dirty: (EpisodeRow & { meta?: Record<string, unknown> })[] = [];
+    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<string, unknown> }): 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 &&

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({

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;