From 378e68ad8a3e127084ee2c457e2f598550c93e43 Mon Sep 17 00:00:00 2001
From: lorentz <lorentz@wulfconsulting.com>
Date: Wed, 29 Apr 2026 14:56:00 -0400
Subject: [PATCH] fix(analyzer): reset stale in-flight jobs on worker boot
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

A container restart leaves analyzer_jobs rows stuck in
fetching/triaging/itglue/analyzing/deep_review forever — the worker's
claimQueuedJob only picks up status='queued', so a job mid-pipeline
when the process died gets orphaned.

resetStaleJobsToQueued() reverts any active-state row whose started_at
is older than 10 min back to 'queued' with started_at=NULL. The worker
calls it once on start() before scheduling the first poll. 10 min is
3x the realistic pipeline ceiling — well past Sonnet+Opus combined.

Logs the count when nonzero so restarts that recover work are visible.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
---
 lib/services/analyzer/persistence.ts | 22 ++++++++++++++++++++++
 lib/services/analyzer/worker.ts      | 17 +++++++++++++++++
 2 files changed, 39 insertions(+)

diff --git a/lib/services/analyzer/persistence.ts b/lib/services/analyzer/persistence.ts
index 2265b5e..a87f7a2 100644
--- a/lib/services/analyzer/persistence.ts
+++ b/lib/services/analyzer/persistence.ts
@@ -329,6 +329,28 @@ export async function completeJob(
   );
 }
 
+/**
+ * Reclaim orphaned in-flight jobs whose started_at is older than the threshold.
+ * Called once on worker boot — any job in an active state (fetching/triaging/
+ * itglue/analyzing/deep_review) that's been "running" longer than the expected
+ * pipeline ceiling is almost certainly orphaned by a container restart and
+ * needs to be re-queued. Returns the number of rows reset.
+ */
+export async function resetStaleJobsToQueued(
+  thresholdMinutes = 10
+): Promise<number> {
+  const res = await postgresClient.query(
+    `UPDATE analyzer_jobs
+        SET status = 'queued',
+            started_at = NULL
+      WHERE status IN ('fetching','triaging','itglue','analyzing','deep_review')
+        AND started_at IS NOT NULL
+        AND started_at < NOW() - ($1::int || ' minutes')::interval`,
+    [thresholdMinutes]
+  );
+  return res.rowCount ?? 0;
+}
+
 export async function failJob(jobId: string, errorMessage: string): Promise<void> {
   await postgresClient.query(
     `UPDATE analyzer_jobs
diff --git a/lib/services/analyzer/worker.ts b/lib/services/analyzer/worker.ts
index 20a42a6..213f88b 100644
--- a/lib/services/analyzer/worker.ts
+++ b/lib/services/analyzer/worker.ts
@@ -20,6 +20,7 @@ import {
   failJob,
   insertAnalysis,
   insertFailedAnalysis,
+  resetStaleJobsToQueued,
   updateAnalysisFingerprint,
   updateJobStatus,
 } from './persistence';
@@ -33,6 +34,7 @@ import type {
 } from '@/lib/types/analyzer';
 
 const POLL_INTERVAL_MS = 2_000;
+const STALE_JOB_RESET_MINUTES = 10;
 
 class AnalyzerWorker {
   private timer: NodeJS.Timeout | null = null;
@@ -43,6 +45,21 @@ class AnalyzerWorker {
     if (this.running) return;
     this.running = true;
     console.log('[ANALYZER-WORKER] starting; polling every 2s');
+
+    // Reclaim jobs orphaned by a previous restart. Any active-state job
+    // whose started_at is older than the pipeline ceiling is presumed
+    // orphaned and gets reset to 'queued' so this worker can re-claim it.
+    try {
+      const reset = await resetStaleJobsToQueued(STALE_JOB_RESET_MINUTES);
+      if (reset > 0) {
+        console.log(
+          `[ANALYZER-WORKER] reset ${reset} stale in-flight job(s) to queued (older than ${STALE_JOB_RESET_MINUTES}min)`
+        );
+      }
+    } catch (err) {
+      console.error('[ANALYZER-WORKER] stale-job reset failed:', err);
+    }
+
     this.scheduleNextPoll(0);
   }