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 StageExecutionRecord,
  type StageName,
  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 Provider,
  stageModelsFor,
} from '@/lib/services/llm/models';
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;
  /** LLM provider for this run. Defaults to 'anthropic' for back-compat. */
  provider?: Provider;
}

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;
  /** Phase 2: the raw triage / sonnet / opus responses fed to Stage 6 fingerprint. */
  triage_response: TriageResponse;
  sonnet_response: DeepAnalysisResponse;
  opus_response: OpusResponse | null;
  /** LEGACY: 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;
  /**
   * Phase 2 — emitted once per stage attempt, on success OR failure. Worker
   * collects these into an array; on pipeline failure the array still has
   * everything that ran. Pipeline pushes the record before re-throwing.
   */
  onStageRecord?: (record: StageExecutionRecord) => void;
  /**
   * Phase 2 — emitted right after Stage 0 succeeds. Lets the worker capture
   * the preprocessed ticket so it can persist a failed analyzer_analyses row
   * (with source_snapshot + content_hash) when a later stage throws.
   */
  onPreprocessed?: (pre: PreprocessedTicket) => void;
}

/** Internal helper: run a stage, time it, push a record, propagate errors. */
async function recordedStage<T extends { usage?: { input_tokens: number; output_tokens: number } }>(
  meta: {
    stage: StageName;
    stage_order: number;
    model_id: string | null;
    input_payload: unknown;
  },
  fn: () => Promise<T>,
  callbacks: PipelineProgressCallbacks,
  outputSelector: (result: T) => unknown
): Promise<T> {
  const startedAt = new Date();
  try {
    const result = await fn();
    const completedAt = new Date();
    callbacks.onStageRecord?.({
      ...meta,
      output_payload: outputSelector(result),
      input_tokens: result.usage?.input_tokens ?? null,
      output_tokens: result.usage?.output_tokens ?? null,
      latency_ms: completedAt.getTime() - startedAt.getTime(),
      started_at: startedAt,
      completed_at: completedAt,
      error_message: null,
    });
    return result;
  } catch (err) {
    const completedAt = new Date();
    callbacks.onStageRecord?.({
      ...meta,
      output_payload: {},
      input_tokens: null,
      output_tokens: null,
      latency_ms: completedAt.getTime() - startedAt.getTime(),
      started_at: startedAt,
      completed_at: completedAt,
      error_message: err instanceof Error ? err.message : String(err),
    });
    throw err;
  }
}

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

  // ── Stage 0: preprocess ──────────────────────────────────────────────────
  await callbacks.onStage?.('fetching');
  const preStart = new Date();
  const pre = preprocessTicket(input.bundle);
  const preEnd = new Date();
  callbacks.onPreprocessed?.(pre);
  callbacks.onStageRecord?.({
    stage: 'preprocess',
    stage_order: 1,
    model_id: null,
    input_payload: {
      ticket_number: input.bundle.ticket.ticket_number,
      notes_count: input.bundle.notes?.length ?? 0,
      time_entries_count: input.bundle.time_entries?.length ?? 0,
    },
    output_payload: {
      events_count: pre.events.length,
      counts: pre.counts,
      content_hash: pre.content_hash,
    },
    input_tokens: null,
    output_tokens: null,
    latency_ms: preEnd.getTime() - preStart.getTime(),
    started_at: preStart,
    completed_at: preEnd,
    error_message: null,
  });

  // ── 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,
      provider
    );
    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: triage ──────────────────────────────────────────────────────
  await callbacks.onStage?.('triaging');
  const triageModel = stageModels.triage;
  const triageResult = await recordedStage(
    {
      stage: 'triage',
      stage_order: 2,
      model_id: triageModel,
      input_payload: {
        ticket_number: pre.header.ticket_number,
        events_count: pre.events.length,
        filtered_noise_count: pre.counts.filtered_noise,
      },
    },
    () => runTriageStage(pre, anthropic, triageModel),
    callbacks,
    (r) => r.data
  );
  usage = addUsage(usage, triageResult.usage);
  estimatedCostUsd += triageResult.estimated_cost_usd;
  traces.triage = {
    model: triageModel,
    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');
    const itglueStart = new Date();
    let itglueErr: Error | null = null;
    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.
      itglueErr = err instanceof Error ? err : new Error(String(err));
      console.warn(
        `[pipeline] IT Glue search failed for ${pre.header.ticket_number}: ${itglueErr.message}`
      );
    }
    const itglueEnd = new Date();
    callbacks.onStageRecord?.({
      stage: 'itglue',
      stage_order: 3,
      model_id: null,
      input_payload: {
        org_name: pre.header.account_name,
        hints: triageResult.data.itglue_search_hints,
      },
      // Redacted-only payload (itglue-search applies redact() internally).
      output_payload: itglueErr
        ? {}
        : {
            org_id: itglueResult?.org_id ?? null,
            alias_used: itglueResult?.alias_used ?? false,
            docs: itglueDocs,
          },
      input_tokens: null,
      output_tokens: null,
      latency_ms: itglueEnd.getTime() - itglueStart.getTime(),
      started_at: itglueStart,
      completed_at: itglueEnd,
      error_message: itglueErr ? itglueErr.message : null,
    });
    traces.itglue = {
      org_id: itglueResult?.org_id ?? null,
      alias_used: itglueResult?.alias_used ?? false,
      doc_count: itglueDocs.length,
    };
  }

  // ── Stage 3: deep analysis ───────────────────────────────────────────────
  await callbacks.onStage?.('analyzing');
  const deepAnalysisModel = stageModels.deep_analysis;
  const sonnetResult = await recordedStage(
    {
      stage: 'analyze',
      stage_order: 4,
      model_id: deepAnalysisModel,
      input_payload: {
        ticket_number: pre.header.ticket_number,
        events_count: pre.events.length,
        triage: triageResult.data,
        itglue_doc_count: itglueDocs.length,
      },
    },
    () =>
      runDeepAnalysisStage(
        { pre, triage: triageResult.data, itglue_docs: itglueDocs },
        anthropic,
        deepAnalysisModel
      ),
    callbacks,
    (r) => r.data
  );
  usage = addUsage(usage, sonnetResult.usage);
  estimatedCostUsd += sonnetResult.estimated_cost_usd;
  traces.deep_analysis = {
    model: deepAnalysisModel,
    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 opusResponseForResult: OpusResponse | null = null;
  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 deepReasoningModel = stageModels.deep_reasoning;
      const opusResult = await recordedStage(
        {
          stage: 'deep_review',
          stage_order: 5,
          model_id: deepReasoningModel,
          input_payload: {
            ticket_number: pre.header.ticket_number,
            events_count: pre.events.length,
            triage: triageResult.data,
            sonnet_summary: sonnetResult.data.summary,
          },
        },
        () =>
          runDeepReasoningStage(
            { pre, triage: triageResult.data, sonnet: sonnetResult.data },
            anthropic,
            deepReasoningModel
          ),
        callbacks,
        // Per spec: store the FULL Opus response including opus_notes, not
        // just the merged updates that previously overwrote everything.
        (r) => r.data
      );
      usage = addUsage(usage, opusResult.usage);
      estimatedCostUsd += opusResult.estimated_cost_usd;
      opusUsed = true;
      traces.deep_reasoning = {
        model: deepReasoningModel,
        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;
      opusResponseForResult = 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,
    triage_response: triageResult.data,
    sonnet_response: sonnetResult.data,
    opus_response: opusResponseForResult,
    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,
};