rca.ts

// Root cause analysis.
//
// The question is never "which service is red" - in a microservice outage
// half the graph is red. The question is "which red service is red for its
// own reasons". Each candidate gets independent, explainable features:
//
//   origin      share of failing traces whose deepest error is in this service
//   latency     share of the excess per-trace self-time spent in this service
//   frontier    anomalous, but none of its own dependencies are
//   onset       how early its anomaly started relative to the others
//   change      a deploy / flag change shortly before onset
//   saturation  it owns a resource (connection pool) that is full
//   leak        it holds most of someone else's saturated resource, for too long
//   newLogs     error log signatures that did not exist before the incident
//   down        no heartbeat at all
//
// Scores are a weighted sum of features; confidence is a softmax over
// candidates, discounted when there is little evidence. Every feature that
// contributes also emits an Evidence item, so the UI (and the LLM narrator)
// can only ever say things that trace back to data.

import type { TelemetryStore } from '../store/store.ts';
import type { AnomalyEngine } from './anomaly.ts';
import type { Candidate, Evidence, HypothesisKind, RcaResult, Span } from '../types.ts';

export const WEIGHTS = { origin: 3, latency: 2.5, frontier: 1, onset: 0.8, change: 2.5, saturation: 1, leak: 3.5, newLogs: 1.2, down: 4 };

export interface RcaInput {
  store: TelemetryStore;
  anomaly: AnomalyEngine;
  services: string[]; // all known services
  onsetAt: number;
  now: number;
}

const pct = (x: number) => `${(x * 100).toFixed(x < 0.01 ? 2 : 1)}%`;
const clock = (t: number) => new Date(t).toLocaleTimeString('en-GB', { hour12: false });
const fmtMs = (x: number) => (x >= 1000 ? `${(x / 1000).toFixed(2)}s` : `${Math.round(x)}ms`);

export function downstreamMap(edges: { source: string; target: string }[]) {
  const m = new Map<string, Set<string>>();
  for (const e of edges) {
    if (!m.has(e.source)) m.set(e.source, new Set());
    m.get(e.source)!.add(e.target);
  }
  return m;
}

function groupTraces(spans: Span[]) {
  const t = new Map<string, Span[]>();
  for (const s of spans) {
    let l = t.get(s.traceId);
    if (!l) t.set(s.traceId, (l = []));
    l.push(s);
  }
  return t;
}

/** Deepest error in a trace. A failing CLIENT span with no failing child means the callee was slow or unreachable - blame the callee. */
export function errorOrigin(trace: Span[]): { service: string; span: Span; via: 'self' | 'callee' } | undefined {
  const errs = trace.filter((s) => s.error);
  if (!errs.length) return undefined;
  const failingChildren = new Set(errs.map((s) => s.parentSpanId).filter(Boolean));
  const leaves = errs.filter((s) => !failingChildren.has(s.spanId));
  // prefer the earliest-finishing leaf: that is where the failure started
  leaves.sort((a, b) => a.endMs - b.endMs);
  const leaf = leaves[0];
  if (leaf.kind === 3 && leaf.peerService) return { service: leaf.peerService, span: leaf, via: 'callee' };
  return { service: leaf.service, span: leaf, via: 'self' };
}

/** Exclusive (self) time per service, averaged per trace. */
export function selfTimePerTrace(traces: Map<string, Span[]>) {
  const total = new Map<string, number>();
  let n = 0;
  for (const spans of traces.values()) {
    if (!spans.some((s) => s.kind === 2 && !s.parentSpanId)) continue; // only complete, rooted traces
    n++;
    const childSum = new Map<string, number>();
    for (const s of spans) if (s.parentSpanId) childSum.set(s.parentSpanId, (childSum.get(s.parentSpanId) ?? 0) + s.durationMs);
    for (const s of spans) {
      const self = Math.max(0, s.durationMs - (childSum.get(s.spanId) ?? 0));
      // waiting on a callee that never answered (no child span) is the callee's time
      const owner = s.kind === 3 && s.peerService && !childSum.has(s.spanId) ? s.peerService : s.service;
      total.set(owner, (total.get(owner) ?? 0) + self);
    }
  }
  const out = new Map<string, number>();
  for (const [k, v] of total) out.set(k, n ? v / n : 0);
  return { perTrace: out, traces: n };
}

export function analyze(input: RcaInput): RcaResult {
  const { store, anomaly, services, onsetAt, now } = input;
  const bucket = store.bucketMs;
  const from = onsetAt - bucket;
  const baseFrom = from - 60_000;
  const baseTo = from - bucket;

  const evidence: Evidence[] = [];
  const add = (e: Omit<Evidence, 'id'>) => {
    const ev = { ...e, id: '' } as Evidence;
    evidence.push(ev);
    return ev;
  };

  const edges = store.edges(from - 30_000, now);
  const down = downstreamMap(edges);
  const anomalous = new Map(services.map((s) => [s, anomaly.active(s)]));
  const isAnom = (s: string) => (anomalous.get(s)?.length ?? 0) > 0;

  const feats = new Map<string, Record<string, number>>();
  const evFor = new Map<string, Evidence[]>();
  const f = (svc: string) => {
    let x = feats.get(svc);
    if (!x) feats.set(svc, (x = {}));
    return x;
  };
  const attach = (svc: string, e: Evidence) => {
    let l = evFor.get(svc);
    if (!l) evFor.set(svc, (l = []));
    l.push(e);
  };

  // --- symptoms: anomalous signals ----------------------------------------
  for (const svc of services) {
    for (const a of anomalous.get(svc) ?? []) {
      if (a.signal === 'availability') continue;
      const series = seriesFor(store, svc, a.signal, now);
      const e = add({
        kind: 'metric', service: svc, weight: 0,
        title: a.signal === 'error_rate' ? `${svc} error rate ${pct(series.at(-1)?.v ?? a.value)} (baseline ${pct(a.baseline)})` : `${svc} p95 latency ${fmtMs(a.value)} (baseline ${fmtMs(a.baseline)})`,
        detail: `Anomalous since ${clock(a.onsetAt ?? now)}, score ${a.score.toFixed(1)}σ.`,
        refs: { series },
      });
      attach(svc, e);
    }
  }

  // --- trace origin attribution -------------------------------------------
  const traces = groupTraces(store.spansBetween(from, now));
  const origins = new Map<string, { n: number; viaCallee: number; messages: Map<string, number>; traceIds: string[] }>();
  let failing = 0;
  for (const [traceId, spans] of traces) {
    const o = errorOrigin(spans);
    if (!o) continue;
    failing++;
    let r = origins.get(o.service);
    if (!r) origins.set(o.service, (r = { n: 0, viaCallee: 0, messages: new Map(), traceIds: [] }));
    r.n++;
    if (o.via === 'callee') r.viaCallee++;
    const msg = o.span.errorMessage ?? o.span.name;
    r.messages.set(msg, (r.messages.get(msg) ?? 0) + 1);
    if (r.traceIds.length < 5) r.traceIds.push(traceId);
  }
  for (const [svc, r] of origins) {
    const share = r.n / failing;
    f(svc).origin = share;
    const top = [...r.messages].sort((a, b) => b[1] - a[1])[0][0];
    attach(svc, add({
      kind: 'trace', service: svc, weight: share * WEIGHTS.origin,
      title: `${r.n}/${failing} failing traces originate in ${svc}${r.viaCallee > r.n / 2 ? ' (callers timed out or could not connect)' : ''}`,
      detail: `Most common error at the origin: "${top}"`,
      refs: { traceIds: r.traceIds },
    }));
  }

  // --- latency attribution (self time vs baseline) -------------------------
  const cur = selfTimePerTrace(traces);
  const base = selfTimePerTrace(groupTraces(store.spansBetween(baseFrom, baseTo)));
  const excess = new Map<string, number>();
  let excessTotal = 0;
  for (const [svc, v] of cur.perTrace) {
    const d = Math.max(0, v - (base.perTrace.get(svc) ?? 0));
    excess.set(svc, d);
    excessTotal += d;
  }
  if (excessTotal > 60 && cur.traces > 5) {
    for (const [svc, d] of excess) {
      const share = d / excessTotal;
      if (share < 0.15) continue;
      f(svc).latency = share;
      attach(svc, add({
        kind: 'trace', service: svc, weight: share * WEIGHTS.latency,
        title: `${svc} accounts for ${pct(share)} of the added latency per request`,
        detail: `Self time per trace ${fmtMs(cur.perTrace.get(svc) ?? 0)} vs ${fmtMs(base.perTrace.get(svc) ?? 0)} before the incident (${cur.traces} traces analysed).`,
      }));
    }
  }

  // --- graph position ------------------------------------------------------
  for (const svc of services) {
    if (!isAnom(svc) && !f(svc).origin) continue;
    const deps = [...(down.get(svc) ?? [])];
    const badDeps = deps.filter((d) => isAnom(d) || (f(d).origin ?? 0) > 0.2);
    if (badDeps.length === 0) {
      f(svc).frontier = 1;
    } else if (isAnom(svc)) {
      attach(svc, add({
        kind: 'topology', service: svc, weight: 0,
        title: `${svc} depends on unhealthy ${badDeps.join(', ')}`,
        detail: `Its symptoms are consistent with propagation from downstream.`,
      }));
    }
  }

  // --- onset ordering ------------------------------------------------------
  const onsets = services.flatMap((s) => (anomalous.get(s) ?? []).map((a) => ({ s, t: a.onsetAt ?? now })));
  const earliest = Math.min(...onsets.map((o) => o.t), now);
  for (const { s, t } of onsets) f(s).onset = Math.max(f(s).onset ?? 0, Math.exp(-(t - earliest) / (2 * bucket)));

  // --- change correlation --------------------------------------------------
  for (const c of store.changes) {
    // only changes that happened before the symptoms, and never our own remediations
    if (c.ts > onsetAt + 2 * bucket || c.ts < onsetAt - 10 * 60_000 || c.actor === 'codepulse') continue;
    if (c.kind !== 'deploy' && c.kind !== 'flag') continue;
    // a deploy that was already superseded (rolled back / redeployed) before onset is not live code
    if (c.kind === 'deploy' && store.changes.some((d) => d.service === c.service && (d.kind === 'deploy' || d.kind === 'rollback') && d.ts > c.ts && d.ts <= onsetAt + bucket)) continue;
    const age = Math.max(0, onsetAt - c.ts);
    const strength = (c.kind === 'deploy' ? 1 : 0.6) * Math.pow(0.5, age / (5 * 60_000));
    if (strength <= (f(c.service).change ?? 0)) continue;
    f(c.service).change = strength;
    attach(c.service, add({
      kind: 'change', service: c.service, weight: strength * WEIGHTS.change,
      title: c.ts >= onsetAt ? `${c.summary} right as the anomaly started` : `${c.summary} ${Math.round((onsetAt - c.ts) / 1000)}s before the anomaly started`,
      detail: `Change by ${c.actor} at ${clock(c.ts)}.`,
    }));
  }

  // --- availability ---------------------------------------------------------
  for (const svc of services) {
    if (!(anomalous.get(svc) ?? []).some((a) => a.signal === 'availability')) continue;
    f(svc).down = 1;
    const seen = store.seen(svc);
    const exit = [...store.changes].reverse().find((c) => c.service === svc && c.kind === 'process_exit' && c.ts > onsetAt - 30_000);
    attach(svc, add({
      kind: 'metric', service: svc, weight: WEIGHTS.down,
      title: `${svc} stopped reporting telemetry ${seen ? Math.round((now - seen.ts) / 1000) + 's ago' : ''}`.trim(),
      detail: exit ? `Orchestrator: ${exit.summary}.` : 'No spans, metrics or logs received; callers report connection failures.',
    }));
  }

  // --- resource saturation & leak attribution -----------------------------
  const leaks: { owner: string; holder: string }[] = [];
  for (const g of store.allGauges()) {
    if (!g.name.endsWith('.pool.max')) continue;
    const prefix = g.name.slice(0, -'.max'.length);
    const inUse = store.gauges(g.service, `${prefix}.in_use`);
    const used = inUse.reduce((s, p) => s + p.value, 0);
    const util = used / g.value;
    if (util < 0.9) continue;
    const owner = g.service;
    f(owner).saturation = util;
    const waiting = store.gauges(owner, `${prefix}.waiting`)[0]?.value ?? 0;
    attach(owner, add({
      kind: 'resource', service: owner, weight: util * WEIGHTS.saturation,
      title: `${owner} connection pool saturated: ${used}/${g.value} in use, ${waiting} waiting`,
      detail: `New requests queue for a connection and time out.`,
      refs: { series: store.gaugeSeries(owner, `${prefix}.in_use`, inUse.sort((a, b) => b.value - a.value)[0]?.attrs ?? {}) },
    }));
    const holder = inUse.sort((a, b) => b.value - a.value)[0];
    const client = holder?.attrs.client;
    if (!client || client === 'none' || holder.value / used < 0.8) continue;
    const age = store.gauges(owner, `${prefix}.lease_age_max_s`).find((p) => p.attrs.client === client)?.value ?? 0;
    if (age < 5) continue;
    f(client).leak = holder.value / used;
    leaks.push({ owner, holder: client });
    attach(client, add({
      kind: 'resource', service: client, weight: WEIGHTS.leak,
      title: `${client} holds ${holder.value}/${used} ${owner} connections; oldest held for ${age.toFixed(0)}s`,
      detail: `Normal queries hold a connection for ~10ms. Connections held this long are not being released: a leak in ${client}, not a capacity problem in ${owner}.`,
      refs: { series: store.gaugeSeries(owner, `${prefix}.lease_age_max_s`, { client }) },
    }));
  }

  // --- new log signatures ---------------------------------------------------
  // Errors get wrapped as they travel up: "orders failed: payments failed:
  // ledger-db failed: pool exhausted". A wrapped message is a propagated
  // symptom, so a new signature only counts for the service whose message is
  // not just someone else's message with a prefix.
  const fresh: { svc: string; id: number; c: number; text: string }[] = [];
  for (const svc of services) {
    let best: { id: number; c: number } | undefined;
    for (const [id, c] of store.templateCounts(svc, from, now)) {
      const t = store.miner.get(id);
      if (!t || t.firstSeen < from - bucket || c < 3) continue;
      if (!best || c > best.c) best = { id, c };
    }
    if (best) fresh.push({ svc, ...best, text: store.miner.text(best.id) });
  }
  const tail = (x: string) => (x.includes('failed: ') ? x.slice(x.indexOf('failed: ') + 8) : x);
  for (const n of fresh) {
    const wraps = fresh.some((o) => o.svc !== n.svc && o.text !== n.text && n.text.endsWith(tail(o.text)));
    if (wraps) continue;
    f(n.svc).newLogs = 1;
    const t = store.miner.get(n.id)!;
    attach(n.svc, add({
      kind: 'log', service: n.svc, weight: WEIGHTS.newLogs,
      title: `New error signature in ${n.svc} (${n.c}x since ${clock(t.firstSeen)})`,
      detail: n.text,
      refs: { logSample: t.sample },
    }));
  }
  const wrapped = fresh.filter((n) => !f(n.svc).newLogs).map((n) => n.svc);
  if (wrapped.length) {
    add({
      kind: 'log', service: wrapped[0], weight: 0,
      title: `Error messages in ${wrapped.join(', ')} wrap a downstream error`,
      detail: `They re-raise a failure from further down the call chain; not counted as independent evidence.`,
    });
  }

  // --- score ----------------------------------------------------------------
  const score = (x: Record<string, number>) => Object.entries(x).reduce((s, [k, v]) => s + (WEIGHTS[k as keyof typeof WEIGHTS] ?? 0) * v, 0);
  const scores = new Map([...feats].map(([s, x]) => [s, score(x)]));

  // Blame transfer: an owner whose resource is exhausted by someone else's leak is a victim.
  for (const { owner, holder } of leaks) {
    const o = scores.get(owner) ?? 0;
    scores.set(holder, (scores.get(holder) ?? 0) + 0.8 * o);
    scores.set(owner, 0.3 * o);
  }

  const ranked = [...scores].filter(([, s]) => s > 0.5).sort((a, b) => b[1] - a[1]);
  const T = 1.5;
  const z = ranked.reduce((s, [, v]) => s + Math.exp(v / T), 0) || 1;
  const coverage = Math.min(1, (failing + (cur.traces > 5 && excessTotal > 60 ? 10 : 0) + (f(ranked[0]?.[0] ?? '').down ? 10 : 0)) / 10);

  // Assign stable ids by weight so E1 is always the strongest evidence.
  evidence.sort((a, b) => b.weight - a.weight);
  evidence.forEach((e, i) => (e.id = `E${i + 1}`));

  const candidates: Candidate[] = ranked.map(([service, s]) => ({
    service,
    score: Number(s.toFixed(2)),
    confidence: Number(Math.min(0.98, (Math.exp(s / T) / z) * (0.6 + 0.4 * coverage)).toFixed(2)),
    features: Object.fromEntries(Object.entries(feats.get(service) ?? {}).map(([k, v]) => [k, Number(v.toFixed(2))])),
    evidence: (evFor.get(service) ?? []).map((e) => e.id),
  }));

  const top = candidates[0];
  const leak = leaks.find((l) => l.holder === top?.service);
  const hypothesis = classify(top, leak?.owner, store, onsetAt);

  const sampleTraceId = origins.get(leak?.owner ?? top?.service ?? '')?.traceIds[0] ?? [...origins.values()][0]?.traceIds[0];
  return { computedAt: now, hypothesis, candidates, evidence, sampleTraceId };
}

function classify(top: Candidate | undefined, leakOwner: string | undefined, store: TelemetryStore, onsetAt: number): RcaResult['hypothesis'] {
  if (!top) return { kind: 'error_spike', service: 'unknown', summary: 'Not enough evidence yet to identify a root cause.', confidence: 0 };
  const x = top.features;
  const s = top.service;
  let kind: HypothesisKind = 'error_spike';
  let summary = `${s} is producing errors that propagate to its callers.`;
  if (x.down) {
    kind = 'service_down';
    summary = `${s} is down (no telemetry, callers cannot connect).`;
  } else if (x.leak && leakOwner) {
    kind = 'connection_pool_leak';
    summary = `${s} is leaking ${leakOwner} connections; the pool is exhausted and every service that needs ${leakOwner} is failing.`;
  } else if ((x.change ?? 0) >= 0.4) {
    const c = [...store.changes].reverse().find((c) => c.service === s && c.kind === 'deploy' && c.ts <= onsetAt + 2 * store.bucketMs);
    kind = 'bad_deploy';
    summary = `Regression introduced by ${c ? c.summary : `a recent change to ${s}`}.`;
  } else if ((x.latency ?? 0) >= 0.4) {
    kind = 'latency_degradation';
    summary = `${s} became slow; callers are timing out waiting on it.`;
  }
  return { kind, service: s, summary, confidence: top.confidence, symptomService: leakOwner };
}

function seriesFor(store: TelemetryStore, svc: string, signal: string, now: number) {
  const out: { t: number; v: number }[] = [];
  const end = store.bucketOf(now);
  for (let b = end - 40 * store.bucketMs; b < end; b += store.bucketMs) {
    const s = store.snapshot(svc, b, now);
    out.push({ t: b, v: signal === 'error_rate' ? s.errorRate : s.p95 });
  }
  return out;
}