wulf-pulse/lib/services/analyzer/worker.ts

/**
 * Analyzer job worker.
 *
 * Polls `analyzer_jobs` for queued rows, runs the full pipeline, persists the
 * result, and updates the job's status / result_analysis_id throughout.
 *
 * Concurrency model: a single in-process polling loop. Multiple Next.js
 * workers will all import this module, but `claimQueuedJob()` uses
 * `FOR UPDATE SKIP LOCKED` so they cooperate at the row level — each job is
 * processed exactly once.
 *
 * Auto-start: the worker self-initializes on first server-side import in
 * production. In dev / tests, set ANALYZER_WORKER_AUTOSTART=1 to opt in.
 */

import {
  bulkInsertStageExecutions,
  claimQueuedJob,
  completeJob,
  failJob,
  insertAnalysis,
  insertFailedAnalysis,
  resetStaleJobsToQueued,
  updateAnalysisFingerprint,
  updateJobStatus,
} from './persistence';
import { loadTicketBundle, TicketNotFoundError } from './data-access';
import { runPipeline, type PipelineResult } from './pipeline';
import { runFingerprintStage } from './stages/stage6-fingerprint';
import { HAIKU } from '@/lib/services/llm/models';
import type {
  PreprocessedTicket,
  StageExecutionRecord,
} from '@/lib/types/analyzer';

const POLL_INTERVAL_MS = 2_000;
const STALE_JOB_RESET_MINUTES = 10;

class AnalyzerWorker {
  private timer: NodeJS.Timeout | null = null;
  private running = false;
  private inFlight = false;

  async start(): Promise<void> {
    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);
  }

  async stop(): Promise<void> {
    this.running = false;
    if (this.timer) {
      clearTimeout(this.timer);
      this.timer = null;
    }
  }

  private scheduleNextPoll(delay: number): void {
    if (!this.running) return;
    this.timer = setTimeout(() => {
      void this.poll();
    }, delay);
  }

  private async poll(): Promise<void> {
    if (!this.running) return;
    if (this.inFlight) {
      // Avoid overlap if a previous poll is still running.
      this.scheduleNextPoll(POLL_INTERVAL_MS);
      return;
    }
    this.inFlight = true;
    try {
      const claimed = await claimQueuedJob();
      if (claimed) {
        await this.runJob(claimed.id, claimed.ticket_number, claimed.queued_by_user_id);
      }
    } catch (err) {
      console.error('[ANALYZER-WORKER] poll error:', err);
    } finally {
      this.inFlight = false;
      this.scheduleNextPoll(POLL_INTERVAL_MS);
    }
  }

  /**
   * Process a single job end-to-end. Public so tests + manual triggers can
   * call it directly without going through the poll loop.
   */
  async runJob(
    jobId: string,
    ticketNumber: string,
    triggeredByUserId: string | null
  ): Promise<{ analysis_id: string | null; outcome: PipelineResult['outcome'] | 'failed' }> {
    // Phase 2: collect per-stage records as the pipeline runs, plus the
    // preprocessed bundle, so we can persist a failed analyzer_analyses row
    // (with source_snapshot) when a stage throws.
    const stageRecords: StageExecutionRecord[] = [];
    let capturedPre: PreprocessedTicket | null = null;

    try {
      const bundle = await loadTicketBundle(ticketNumber);

      const result = await runPipeline(
        { bundle, force: false },
        {},
        {
          onStage: (stage) => updateJobStatus(jobId, stage),
          onStageRecord: (rec) => {
            stageRecords.push(rec);
          },
          onPreprocessed: (pre) => {
            capturedPre = pre;
          },
        }
      );

      if (result.outcome === 'idempotent_short_circuit') {
        // Point the job at the existing analysis so the UI can navigate to it.
        await completeJob(jobId, result.existing_analysis_id);
        return {
          analysis_id: result.existing_analysis_id,
          outcome: 'idempotent_short_circuit',
        };
      }

      const inserted = await insertAnalysis({
        ticket_number: result.pre.header.ticket_number,
        autotask_ticket_id: result.pre.header.autotask_ticket_id,
        content_hash: result.pre.content_hash,
        triggered_by_user_id: triggeredByUserId,
        status: 'complete',
        haiku_used: result.meta.haiku_used,
        sonnet_used: result.meta.sonnet_used,
        opus_used: result.meta.opus_used,
        total_input_tokens: result.meta.total_input_tokens,
        total_output_tokens: result.meta.total_output_tokens,
        estimated_cost_usd: result.meta.estimated_cost_usd,
        analysis: result.analysis,
        filtered_noise_count: result.filtered_noise_count,
        model_traces: result.model_traces,
        source_snapshot: result.pre.events,
      });

      // Stage 6 — fingerprint. Failure-tolerant: log and continue.
      const fpStart = new Date();
      let fpInputTokens: number | null = null;
      let fpOutputTokens: number | null = null;
      let fpOutput: unknown = {};
      let fpErr: Error | null = null;
      try {
        const fp = await runFingerprintStage({
          triage: result.triage_response,
          sonnet: result.sonnet_response,
          opus: result.opus_response,
        });
        fpInputTokens = fp.usage.input_tokens;
        fpOutputTokens = fp.usage.output_tokens;
        fpOutput = fp.data;
        await updateAnalysisFingerprint(inserted.id, fp.data);
      } catch (err) {
        fpErr = err instanceof Error ? err : new Error(String(err));
        console.warn(
          `[ANALYZER-WORKER] fingerprint failed for analysis ${inserted.id}: ${fpErr.message}`
        );
      }
      const fpEnd = new Date();
      stageRecords.push({
        stage: 'fingerprint',
        stage_order: 6,
        model_id: HAIKU,
        input_payload: {
          triage_category: result.triage_response.category,
          ticket_number: result.pre.header.ticket_number,
          opus_used: result.opus_response !== null,
        },
        output_payload: fpErr ? {} : fpOutput,
        input_tokens: fpInputTokens,
        output_tokens: fpOutputTokens,
        latency_ms: fpEnd.getTime() - fpStart.getTime(),
        started_at: fpStart,
        completed_at: fpEnd,
        error_message: fpErr ? fpErr.message : null,
      });

      await bulkInsertStageExecutions(inserted.id, stageRecords);

      await completeJob(jobId, inserted.id);
      return { analysis_id: inserted.id, outcome: 'complete' };
    } catch (err) {
      const message = err instanceof Error ? err.message : String(err);
      console.error(
        `[ANALYZER-WORKER] job ${jobId} (${ticketNumber}) failed:`,
        message
      );
      const reason =
        err instanceof TicketNotFoundError
          ? `Ticket ${ticketNumber} not found in local mirror — confirm sync is current.`
          : message;

      // Phase 2: when we have a preprocessed bundle, persist a 'failed'
      // analyzer_analyses row with source_snapshot + accumulated stage rows.
      // Best-effort — if this fails we still fail the job below.
      if (capturedPre !== null) {
        const pre: PreprocessedTicket = capturedPre;
        try {
          const failedAnalysis = await insertFailedAnalysis({
            ticket_number: pre.header.ticket_number,
            autotask_ticket_id: pre.header.autotask_ticket_id,
            content_hash: pre.content_hash,
            triggered_by_user_id: triggeredByUserId,
            source_snapshot: pre.events,
            filtered_noise_count: pre.counts.filtered_noise,
            error_message: reason,
            partial_input_tokens: 0,
            partial_output_tokens: 0,
            partial_cost_usd: 0,
            haiku_used: stageRecords.some((r) => r.stage === 'triage'),
            sonnet_used: stageRecords.some((r) => r.stage === 'analyze'),
            opus_used: stageRecords.some((r) => r.stage === 'deep_review'),
          });
          await bulkInsertStageExecutions(failedAnalysis.id, stageRecords);
        } catch (persistErr) {
          console.error(
            `[ANALYZER-WORKER] failed to persist failed-analysis row for job ${jobId}:`,
            persistErr
          );
        }
      }

      await failJob(jobId, reason);
      return { analysis_id: null, outcome: 'failed' };
    }
  }
}

export const analyzerWorker = new AnalyzerWorker();

/**
 * Should we auto-start the worker on import?
 *   - Skip in browsers (typeof window check).
 *   - Skip during vitest.
 *   - Auto-start in production by default.
 *   - Otherwise opt in via ANALYZER_WORKER_AUTOSTART=1.
 */
function shouldAutoStart(): boolean {
  if (typeof window !== 'undefined') return false;
  if (process.env.VITEST === 'true') return false;
  if (process.env.NODE_ENV === 'production') return true;
  return process.env.ANALYZER_WORKER_AUTOSTART === '1';
}

if (shouldAutoStart()) {
  analyzerWorker.start().catch((err) => {
    console.error('[ANALYZER-WORKER] failed to auto-start:', err);
  });
}

export const _WORKER_INTERNALS = { POLL_INTERVAL_MS };