From 351dd40171c42db76bbbe44368af7f211c447688 Mon Sep 17 00:00:00 2001 From: Doccy008 Date: Fri, 3 Jul 2026 16:18:02 +0200 Subject: [PATCH 1/2] fix(sweep): gate maintenance-sweep workflow ingest by mtime fingerprint MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The 5-minute maintenance sweep re-ran ingestWorkflowsForSession for every active session every cycle with no change detection — unlike the near-real-time startWorkflowPoll, which skips sessions whose workflow artifacts are unchanged (workflowsMaxMtime + a lastSeen map). ingestWorkflowsForSession fully re-parses every workflow journal and every inner agent-*.jsonl from scratch (parseSubagentFile). On a session that accumulates hundreds of workflow runs, each sweep's full re-parse eventually exceeds SWEEP_INTERVAL_MS; the fire-and-forget sweeps overlap and the event loop pegs at ~100% CPU, so the HTTP server stops accepting connections and the dashboard renders a blank page. DASHBOARD_WORKFLOW_POLL_MS=0 does not help — it only disables the poll, not this sweep path. Apply the same cheap mtime fingerprint the poll already uses: keep a per-sweep sweepWorkflowSeen map and skip a session whose workflowsMaxMtime is unchanged since its last ingest. A completing workflow writes its terminal journal, which advances the fingerprint, so the sweep still catches the running→completed transitions it exists to catch; only genuine no-ops are skipped. The identical gate already runs in production via startWorkflowPoll. Add a regression test asserting the skip/re-ingest decision: the fingerprint is stable when nothing changes (skip) and advances when a new artifact appears (re-ingest). Co-Authored-By: Claude Opus 4.8 (1M context) --- server/__tests__/workflow-ingest.test.js | 52 ++++++++++++++++++++++++ server/index.js | 21 +++++++++- 2 files changed, 72 insertions(+), 1 deletion(-) 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..9eb3b8df 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,23 @@ 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"); 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; From d0b72c42cfba083f9d9bd856e18df709e467b164 Mon Sep 17 00:00:00 2001 From: Doccy008 Date: Fri, 3 Jul 2026 16:25:10 +0200 Subject: [PATCH 2/2] fix(sweep): retry on ingest failure, bound sweepWorkflowSeen to active sessions Address review feedback on the workflow-ingest gate: - A transient ingest failure previously stuck a session: the fingerprint was recorded before the async ingest, so on rejection the session was skipped until its artifacts changed again. Delete the fingerprint in .catch so the next sweep retries it. - sweepWorkflowSeen never evicted entries for sessions that went inactive, so it grew unbounded over the process lifetime. Prune it to the current active set at the start of each sweep. Co-Authored-By: Claude Opus 4.8 (1M context) --- server/index.js | 9 +++++++++ 1 file changed, 9 insertions(+) diff --git a/server/index.js b/server/index.js index 9eb3b8df..1f148e0f 100644 --- a/server/index.js +++ b/server/index.js @@ -815,6 +815,12 @@ if (require.main === module) { // Catches workflows that complete without a subsequent hook and flips // launch-detected "running" rows to "completed" once their journal lands. 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 @@ -839,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