anomaly.ts

// Per-service, per-signal anomaly detection over fixed buckets.
//
// A static threshold ("alert if p95 > 500ms") is wrong for every service but
// one, and a plain z-score breaks in four predictable ways. Each one has a
// specific guard here:
//
//  1. Baseline poisoning - during a long incident the EWMA learns the broken
//     state as "normal" and the alert clears itself. -> the baseline is frozen
//     while a signal is breaching or anomalous.
//  2. Zero variance - synthetic or very stable traffic has std ~ 0, so a 2ms
//     wobble becomes z = 40. -> std is floored relative to the mean.
//  3. Small samples - 1 error in 3 requests is a 33% error rate. -> error rate
//     is scored by its Wilson lower bound, which shrinks toward 0 when n is
//     small, and buckets below a minimum request count are skipped.
//  4. Flapping - one noisy bucket opens and closes an incident. -> k-of-n
//     debounce to open, a longer quiet period to close (hysteresis).
//  Plus a practical-significance gate: 0.1% -> 0.4% errors may be 6 sigma
//  but nobody should be paged for it.

import type { AnomalyState, Signal } from '../types.ts';

export interface DetectorConfig {
  warmupBuckets: number;
  alpha: number; // EWMA smoothing
  zThreshold: number;
  openAfter: number; // consecutive breaching buckets to open
  closeAfter: number; // consecutive normal buckets to close
  minDelta: number; // absolute change that matters (ms, or rate)
  relFloor: number; // std floor as a fraction of the mean
  absFloor: number;
  hardLimit?: number; // breach regardless of baseline (an SLO)
}

export const SIGNAL_CONFIG: Record<Exclude<Signal, 'availability'>, DetectorConfig> = {
  error_rate: { warmupBuckets: 5, alpha: 0.15, zThreshold: 4, openAfter: 2, closeAfter: 3, minDelta: 0.03, relFloor: 0.5, absFloor: 0.01, hardLimit: 0.15 },
  latency_p95: { warmupBuckets: 5, alpha: 0.15, zThreshold: 4, openAfter: 2, closeAfter: 3, minDelta: 120, relFloor: 0.3, absFloor: 15 },
};

/** Lower bound of the Wilson score interval (95%). */
export function wilsonLower(errors: number, n: number, z = 1.96) {
  if (n === 0) return 0;
  const p = errors / n;
  const denom = 1 + (z * z) / n;
  const centre = p + (z * z) / (2 * n);
  const margin = z * Math.sqrt((p * (1 - p)) / n + (z * z) / (4 * n * n));
  return Math.max(0, (centre - margin) / denom);
}

export class Detector {
  readonly cfg: DetectorConfig;
  mean = 0;
  variance = 0;
  n = 0;
  active = false;
  onsetAt?: number;
  firstBreachAt?: number;
  private high = 0;
  private low = 0;
  lastValue = 0;
  lastScore = 0;

  constructor(cfg: DetectorConfig) {
    this.cfg = cfg;
  }

  /** Feed one bucket. Returns 'opened' | 'closed' | undefined on state change. */
  update(value: number, ts: number): 'opened' | 'closed' | undefined {
    const c = this.cfg;
    this.lastValue = value;
    if (this.n < c.warmupBuckets) {
      this.learn(value);
      return undefined;
    }
    const std = Math.max(Math.sqrt(this.variance), c.relFloor * this.mean, c.absFloor);
    const score = (value - this.mean) / std;
    this.lastScore = score;
    const breach = (score > c.zThreshold && value - this.mean > c.minDelta) || (c.hardLimit !== undefined && value > c.hardLimit);

    if (breach) {
      if (this.high === 0) this.firstBreachAt = ts;
      this.high++;
      this.low = 0;
    } else {
      this.low++;
      this.high = 0;
      if (!this.active) this.learn(value); // only learn from healthy buckets
    }

    if (!this.active && this.high >= c.openAfter) {
      this.active = true;
      this.onsetAt = this.firstBreachAt;
      return 'opened';
    }
    if (this.active && this.low >= c.closeAfter) {
      this.active = false;
      this.onsetAt = undefined;
      return 'closed';
    }
    return undefined;
  }

  private learn(x: number) {
    if (this.n === 0) {
      this.mean = x;
      this.variance = 0;
    } else {
      // West's incremental EW mean/variance
      const a = this.cfg.alpha;
      const diff = x - this.mean;
      const incr = a * diff;
      this.mean += incr;
      this.variance = (1 - a) * (this.variance + diff * incr);
    }
    this.n++;
  }
}

const MIN_REQUESTS = 8;

/** All detectors for all services. Availability is a heartbeat check, not a statistic. */
export class AnomalyEngine {
  private detectors = new Map<string, Detector>();
  private down = new Map<string, { since: number; buckets: number; active: boolean }>();

  private get(service: string, signal: Exclude<Signal, 'availability'>) {
    const key = `${service}:${signal}`;
    let d = this.detectors.get(key);
    if (!d) this.detectors.set(key, (d = new Detector(SIGNAL_CONFIG[signal])));
    return d;
  }

  evaluate(service: string, snap: { requests: number; errors: number; p95: number; up: boolean }, ts: number, everSeen = true) {
    const changes: { signal: Signal; change: 'opened' | 'closed' }[] = [];

    // availability: two buckets without a heartbeat. A service we have never
    // heard from is "not onboarded yet", not "down" - otherwise every boot is an outage.
    const dn = this.down.get(service) ?? { since: ts, buckets: 0, active: false };
    if (!snap.up && everSeen) {
      if (dn.buckets === 0) dn.since = ts;
      dn.buckets++;
      if (!dn.active && dn.buckets >= 2) {
        dn.active = true;
        changes.push({ signal: 'availability', change: 'opened' });
      }
    } else {
      if (dn.active) changes.push({ signal: 'availability', change: 'closed' });
      dn.buckets = 0;
      dn.active = false;
    }
    this.down.set(service, dn);

    if (snap.requests >= MIN_REQUESTS) {
      const er = this.get(service, 'error_rate').update(wilsonLower(snap.errors, snap.requests), ts);
      if (er) changes.push({ signal: 'error_rate', change: er });
      const lat = this.get(service, 'latency_p95').update(snap.p95, ts);
      if (lat) changes.push({ signal: 'latency_p95', change: lat });
    }
    return changes;
  }

  state(service: string): AnomalyState[] {
    const out: AnomalyState[] = [];
    for (const signal of ['error_rate', 'latency_p95'] as const) {
      const d = this.detectors.get(`${service}:${signal}`);
      if (!d) continue;
      out.push({ service, signal, active: d.active, onsetAt: d.onsetAt, value: d.lastValue, baseline: d.mean, score: d.lastScore });
    }
    const dn = this.down.get(service);
    out.push({ service, signal: 'availability', active: !!dn?.active, onsetAt: dn?.active ? dn.since : undefined, value: dn?.active ? 0 : 1, baseline: 1, score: dn?.active ? 99 : 0 });
    return out;
  }

  active(service: string) {
    return this.state(service).filter((s) => s.active);
  }

  baseline(service: string, signal: 'error_rate' | 'latency_p95') {
    return this.detectors.get(`${service}:${signal}`)?.mean ?? 0;
  }

  warmedUp(service: string) {
    const d = this.detectors.get(`${service}:latency_p95`);
    return !!d && d.n >= d.cfg.warmupBuckets;
  }
}