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

/**
 * AI Ticket Analyzer pipeline.
 *
 *   Stage 0  preprocess (filter + tag + content hash)
 *   ── idempotency short-circuit if force=false
 *   Stage 1  Haiku triage
 *   Stage 2  (conditional) IT Glue retrieval — redacted
 *   Stage 3  Sonnet deep analysis
 *   Stage 4  (conditional) Opus deep reasoning — apply updates
 *   Stage 5  persist
 *
 * The pipeline is split from the worker so that:
 *   - tests can drive it directly with injected LLM clients
 *   - API routes (re-analyze on demand) can call it synchronously if desired
 *   - the worker provides only queue semantics around it
 */

import {
  type DeepAnalysisResponse,
  type OpusResponse,
  type PreprocessedTicket,
  type TriageResponse,
} from '@/lib/types/analyzer';
import { preprocessTicket, type RawTicketBundle } from './preprocessor';
import { runTriageStage } from './stages/stage1-triage';
import {
  runDeepAnalysisStage,
} from './stages/stage3-deep-analysis';
import {
  applyOpusUpdates,
  runDeepReasoningStage,
  shouldRunDeepReasoning,
} from './stages/stage4-deep-reasoning';
import {
  itglueSearch,
  type ITGlueSearchResult,
  type RedactedDoc,
} from './itglue-search';
import {
  findExistingAnalysisByContentHash,
} from './persistence';
import type { TokenUsage } from '@/lib/services/llm/pricing';
import type Anthropic from '@anthropic-ai/sdk';

/**
 * If the running cost would exceed this before the (expensive) Stage 4 call,
 * we skip Opus and flag the analysis for human review with reason
 * "cost ceiling reached".
 */
const COST_CEILING_USD = 2.0;

export interface PipelineDeps {
  /** Inject for tests. Defaults to live IT Glue search. */
  itglueSearch?: typeof itglueSearch;
  /** Inject for tests. Single Anthropic client used for all stages. */
  anthropic?: Anthropic;
}

export interface PipelineInput {
  bundle: RawTicketBundle;
  /** When false (default), check for an existing analysis with the same content hash and short-circuit if present. */
  force?: boolean;
  /** Override Stage 4 — useful for tests + cost-conscious operators. */
  forceSkipOpus?: boolean;
}

export interface PipelineRunMeta {
  haiku_used: boolean;
  sonnet_used: boolean;
  opus_used: boolean;
  total_input_tokens: number;
  total_output_tokens: number;
  total_cache_creation_tokens: number;
  total_cache_read_tokens: number;
  estimated_cost_usd: number;
  cost_circuit_breaker_tripped: boolean;
}

/** Per-stage trace entry — written to analyzer_analyses.model_traces for debugging. */
interface StageTrace {
  model: string;
  attempts: 1 | 2;
  input_tokens: number;
  output_tokens: number;
  cache_creation_input_tokens?: number;
  cache_read_input_tokens?: number;
  estimated_cost_usd: number;
  events_dropped?: number;
}

export interface PipelineSuccess {
  outcome: 'complete';
  analysis: DeepAnalysisResponse;
  pre: PreprocessedTicket;
  meta: PipelineRunMeta;
  filtered_noise_count: number;
  itglue_search_used: boolean;
  itglue_org_resolved: boolean;
  /** Per-stage debug payload — goes into analyzer_analyses.model_traces. */
  model_traces: {
    triage?: StageTrace;
    deep_analysis?: StageTrace;
    deep_reasoning?: StageTrace;
    triage_response?: TriageResponse;
    sonnet_response?: DeepAnalysisResponse;
    opus_response?: OpusResponse;
    itglue?: { org_id: string | null; alias_used: boolean; doc_count: number };
  };
}

export interface PipelineShortCircuit {
  outcome: 'idempotent_short_circuit';
  existing_analysis_id: string;
  existing_analysis_version: number;
  pre: PreprocessedTicket;
}

export type PipelineResult = PipelineSuccess | PipelineShortCircuit;

function addUsage(running: TokenUsage, add: TokenUsage): TokenUsage {
  return {
    input_tokens: running.input_tokens + add.input_tokens,
    output_tokens: running.output_tokens + add.output_tokens,
    cache_creation_input_tokens:
      (running.cache_creation_input_tokens ?? 0) +
      (add.cache_creation_input_tokens ?? 0),
    cache_read_input_tokens:
      (running.cache_read_input_tokens ?? 0) + (add.cache_read_input_tokens ?? 0),
  };
}

/** Stage progression callbacks — used by the worker to update analyzer_jobs.status. */
export interface PipelineProgressCallbacks {
  onStage?: (
    stage:
      | 'fetching'
      | 'triaging'
      | 'itglue'
      | 'analyzing'
      | 'deep_review'
  ) => Promise<void> | void;
}

export async function runPipeline(
  input: PipelineInput,
  deps: PipelineDeps = {},
  callbacks: PipelineProgressCallbacks = {}
): Promise<PipelineResult> {
  const itglueSearchFn = deps.itglueSearch ?? itglueSearch;
  const anthropic = deps.anthropic;

  // ── Stage 0: preprocess ──────────────────────────────────────────────────
  await callbacks.onStage?.('fetching');
  const pre = preprocessTicket(input.bundle);

  // ── Idempotency: short-circuit if force=false and we have a complete row ─
  if (!input.force) {
    const existing = await findExistingAnalysisByContentHash(
      pre.header.ticket_number,
      pre.content_hash
    );
    if (existing) {
      return {
        outcome: 'idempotent_short_circuit',
        existing_analysis_id: existing.id,
        existing_analysis_version: existing.analysis_version,
        pre,
      };
    }
  }

  // Running cost + usage trackers.
  let usage: TokenUsage = {
    input_tokens: 0,
    output_tokens: 0,
    cache_creation_input_tokens: 0,
    cache_read_input_tokens: 0,
  };
  let estimatedCostUsd = 0;
  const traces: PipelineSuccess['model_traces'] = {};

  // ── Stage 1: Haiku triage ────────────────────────────────────────────────
  await callbacks.onStage?.('triaging');
  const triageResult = await runTriageStage(pre, anthropic);
  usage = addUsage(usage, triageResult.usage);
  estimatedCostUsd += triageResult.estimated_cost_usd;
  traces.triage = {
    model: 'claude-haiku-4-5',
    attempts: triageResult.attempts,
    input_tokens: triageResult.usage.input_tokens,
    output_tokens: triageResult.usage.output_tokens,
    cache_creation_input_tokens: triageResult.usage.cache_creation_input_tokens,
    cache_read_input_tokens: triageResult.usage.cache_read_input_tokens,
    estimated_cost_usd: triageResult.estimated_cost_usd,
    events_dropped: triageResult.events_dropped,
  };
  traces.triage_response = triageResult.data;

  // ── Stage 2 (conditional): IT Glue retrieval ─────────────────────────────
  let itglueDocs: RedactedDoc[] = [];
  let itglueResult: ITGlueSearchResult | null = null;
  if (
    triageResult.data.itglue_lookup_needed &&
    pre.header.account_name
  ) {
    await callbacks.onStage?.('itglue');
    try {
      itglueResult = await itglueSearchFn({
        org_name: pre.header.account_name,
        hints: triageResult.data.itglue_search_hints,
      });
      itglueDocs = itglueResult.docs;
    } catch (err) {
      // Tolerate IT Glue failures — analysis continues without context.
      console.warn(
        `[pipeline] IT Glue search failed for ${pre.header.ticket_number}: ${err instanceof Error ? err.message : String(err)}`
      );
    }
    traces.itglue = {
      org_id: itglueResult?.org_id ?? null,
      alias_used: itglueResult?.alias_used ?? false,
      doc_count: itglueDocs.length,
    };
  }

  // ── Stage 3: Sonnet deep analysis ────────────────────────────────────────
  await callbacks.onStage?.('analyzing');
  const sonnetResult = await runDeepAnalysisStage(
    {
      pre,
      triage: triageResult.data,
      itglue_docs: itglueDocs,
    },
    anthropic
  );
  usage = addUsage(usage, sonnetResult.usage);
  estimatedCostUsd += sonnetResult.estimated_cost_usd;
  traces.deep_analysis = {
    model: 'claude-sonnet-4-6',
    attempts: sonnetResult.attempts,
    input_tokens: sonnetResult.usage.input_tokens,
    output_tokens: sonnetResult.usage.output_tokens,
    cache_creation_input_tokens: sonnetResult.usage.cache_creation_input_tokens,
    cache_read_input_tokens: sonnetResult.usage.cache_read_input_tokens,
    estimated_cost_usd: sonnetResult.estimated_cost_usd,
    events_dropped: sonnetResult.events_dropped,
  };
  traces.sonnet_response = sonnetResult.data;

  let analysis: DeepAnalysisResponse = sonnetResult.data;
  let opusUsed = false;
  let costCircuitBreakerTripped = false;

  // ── Stage 4 (conditional): Opus deep reasoning ───────────────────────────
  const opusTrigger = shouldRunDeepReasoning({
    triage: triageResult.data,
    sonnet: sonnetResult.data,
  });

  if (opusTrigger && !input.forceSkipOpus) {
    if (estimatedCostUsd >= COST_CEILING_USD) {
      // Skip Opus; flag for review; record the reason so the UI can surface it.
      costCircuitBreakerTripped = true;
      analysis = {
        ...analysis,
        needs_human_review: true,
        human_review_reasons: [
          ...analysis.human_review_reasons,
          `cost ceiling reached ($${estimatedCostUsd.toFixed(4)} ≥ $${COST_CEILING_USD.toFixed(2)} before Opus)`,
        ],
      };
    } else {
      await callbacks.onStage?.('deep_review');
      const opusResult = await runDeepReasoningStage(
        { pre, triage: triageResult.data, sonnet: sonnetResult.data },
        anthropic
      );
      usage = addUsage(usage, opusResult.usage);
      estimatedCostUsd += opusResult.estimated_cost_usd;
      opusUsed = true;
      traces.deep_reasoning = {
        model: 'claude-opus-4-7',
        attempts: opusResult.attempts,
        input_tokens: opusResult.usage.input_tokens,
        output_tokens: opusResult.usage.output_tokens,
        cache_creation_input_tokens: opusResult.usage.cache_creation_input_tokens,
        cache_read_input_tokens: opusResult.usage.cache_read_input_tokens,
        estimated_cost_usd: opusResult.estimated_cost_usd,
        events_dropped: opusResult.events_dropped,
      };
      traces.opus_response = opusResult.data;
      analysis = applyOpusUpdates(analysis, opusResult.data.updates);
    }
  }

  return {
    outcome: 'complete',
    analysis,
    pre,
    filtered_noise_count: pre.counts.filtered_noise,
    itglue_search_used: itglueResult !== null,
    itglue_org_resolved: !!itglueResult?.org_id,
    meta: {
      haiku_used: true,
      sonnet_used: true,
      opus_used: opusUsed,
      total_input_tokens: usage.input_tokens,
      total_output_tokens: usage.output_tokens,
      total_cache_creation_tokens: usage.cache_creation_input_tokens ?? 0,
      total_cache_read_tokens: usage.cache_read_input_tokens ?? 0,
      estimated_cost_usd: Math.round(estimatedCostUsd * 10_000) / 10_000,
      cost_circuit_breaker_tripped: costCircuitBreakerTripped,
    },
    model_traces: traces,
  };
}

export const _PIPELINE_INTERNALS = {
  COST_CEILING_USD,
};