Merge pull request #203 from Doccy008/fix/gate-maintenance-sweep-workflow-ingest

fix(sweep): gate maintenance-sweep workflow ingest by mtime fingerprint

Author: hoangsonww

server/__tests__/workflow-ingest.test.js

@@ -340,6 +340,58 @@ describe("workflowsMaxMtime", () => {
       "0 when there are no workflow artifacts"
     );
   });
+
+  // Regression guard for the maintenance sweep (server/index.js step 3) and
+  // startWorkflowPoll: both skip a session whose workflow artifacts are
+  // unchanged and re-ingest only once the fingerprint advances. Before the
+  // gate, the 5-min sweep full-re-parsed every workflow journal and every
+  // inner agent-*.jsonl for every active session every cycle; on a large
+  // corpus each sweep outran the interval, sweeps overlapped, and the event
+  // loop pegged (dashboard stopped responding). This asserts the exact
+  // skip/re-ingest decision that gate relies on. Isolated root — no shared
+  // fixture state so later suites are unaffected.
+  it("gates skip vs re-ingest: stable fingerprint when unchanged, higher after a new artifact", () => {
+    const root = fs.mkdtempSync(path.join(os.tmpdir(), "wf-gate-"));
+    try {
+      const sid = "sess-gate";
+      const tp = path.join(root, `${sid}.jsonl`);
+      fs.writeFileSync(tp, "");
+      const wdir = path.join(root, sid, "workflows");
+      writeJson(path.join(wdir, "wf_a.json"), {
+        runId: "wf_a",
+        status: "completed",
+        startTime: 1700000000000,
+        workflowProgress: [],
+      });
+
+      const seen = new Map(); // mirrors sweepWorkflowSeen / lastSeen
+
+      const m1 = workflowsMaxMtime(tp);
+      assert.ok(m1 > 0, "fingerprint > 0 with a journal");
+      assert.equal(m1 === 0 || seen.get(sid) === m1, false, "first sight ingests");
+      seen.set(sid, m1);
+
+      const m2 = workflowsMaxMtime(tp);
+      assert.equal(m2, m1, "fingerprint stable when nothing changes");
+      assert.equal(m2 === 0 || seen.get(sid) === m2, true, "unchanged session is skipped");
+
+      const later = path.join(wdir, "wf_b.json");
+      writeJson(later, {
+        runId: "wf_b",
+        status: "completed",
+        startTime: 1700000001000,
+        workflowProgress: [],
+      });
+      const bump = m1 / 1000 + 5; // seconds, safely newer than m1
+      fs.utimesSync(later, bump, bump);
+
+      const m3 = workflowsMaxMtime(tp);
+      assert.ok(m3 > m1, "fingerprint advances when a new artifact appears");
+      assert.equal(m3 === 0 || seen.get(sid) === m3, false, "changed session is re-ingested");
+    } finally {
+      fs.rmSync(root, { recursive: true, force: true });
+    }
+  });
 });
 
 describe("live running workflow (no terminal journal)", () => {

server/index.js

@@ -733,6 +733,11 @@ if (require.main === module) {
   const { broadcast } = require("./websocket");
   const { importCompactions } = require("../scripts/import-history");
   const { transcriptCache } = require("./routes/hooks");
+  // Per-session newest workflow-artifact mtime already ingested by this sweep,
+  // so step 3 below skips sessions whose workflow files are unchanged (the same
+  // cheap fingerprint startWorkflowPoll uses). Declared once so it persists
+  // across sweep ticks.
+  const sweepWorkflowSeen = new Map();
   setInterval(() => {
     // 1. Stale session cleanup — batch agent updates to avoid N+1 queries
     const stale = cleanupDb.stmts.findStaleSessions.all("__periodic__", STALE_MINUTES);
@@ -809,9 +814,29 @@ if (require.main === module) {
     // 3. Scan active sessions for Workflow-tool run journals (issue #167).
     // Catches workflows that complete without a subsequent hook and flips
     // launch-detected "running" rows to "completed" once their journal lands.
-    const { ingestWorkflowsForSession } = require("./lib/workflow-ingest");
+    const { ingestWorkflowsForSession, workflowsMaxMtime } = require("./lib/workflow-ingest");
+    // Forget fingerprints for sessions that are no longer active so the map
+    // can't grow without bound over the process lifetime.
+    const activeIds = new Set(active.map((r) => r.session_id));
+    for (const id of sweepWorkflowSeen.keys()) {
+      if (!activeIds.has(id)) sweepWorkflowSeen.delete(id);
+    }
     for (const row of active) {
       if (!row.tp) continue;
+      // Skip sessions whose workflow artifacts are unchanged since the last
+      // ingest — the same cheap mtime fingerprint startWorkflowPoll uses.
+      // Without this the sweep full-re-parses every workflow journal and every
+      // inner agent-*.jsonl for every active session every cycle; on a large
+      // corpus that re-parse exceeds the sweep interval, sweeps overlap, and
+      // the event loop pegs (dashboard stops responding — white page).
+      let mtime = 0;
+      try {
+        mtime = workflowsMaxMtime(row.tp);
+      } catch {
+        mtime = 0;
+      }
+      if (mtime === 0 || sweepWorkflowSeen.get(row.session_id) === mtime) continue;
+      sweepWorkflowSeen.set(row.session_id, mtime);
       ingestWorkflowsForSession(cleanupDb, { id: row.session_id, transcript_path: row.tp })
         .then((changed) => {
           if (!changed || changed.length === 0) return;
@@ -820,6 +845,9 @@ if (require.main === module) {
           if (sess) broadcast("session_updated", sess);
         })
         .catch((err) => {
+          // Forget the fingerprint so the next sweep retries this session
+          // instead of skipping it until its artifacts change again.
+          sweepWorkflowSeen.delete(row.session_id);
           console.warn(
             `[SWEEP] Workflow scan failed for session ${row.session_id}:`,
             err?.message || err