diff --git a/server/__tests__/workflow-ingest.test.js b/server/__tests__/workflow-ingest.test.js index e49f34cc..9982ecb0 100644 --- a/server/__tests__/workflow-ingest.test.js +++ b/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)", () => { diff --git a/server/index.js b/server/index.js index 9cfeb089..1f148e0f 100644 --- a/server/index.js +++ b/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