src/mimir_lens/anomaly.rs
//! **Lens — `Mimir.Anomaly`: statistical OUTLIERS on the per-node change-series.**
//!
//! Where [`super::churn_filter`] is a frequency FILTER, this lens is an OUTLIER
//! detector over the same [`super::change_series`] substrate (`git_heat_facts`, all
//! snapshots). It looks for four anomaly shapes, grounded in behavioural code analysis
//! and standard change-frequency anomaly detection:
//!
//! - **(a) stable-node churn** — a node whose history says it is stable (baseline
//! change-rate ≈ 0) that SUDDENLY churns. z-score of the latest per-interval change
//! count against the node's own baseline. *A rarely-touched foundation moving.*
//! - **(b) change-rate spike** — a node already active whose latest interval is a
//! statistical spike vs its baseline (`z ≥ z_threshold`), cross-checked against an
//! EWMA baseline so a slow ramp is not mistaken for a spike.
//! - **(c) new temporal coupling** — files that SUDDENLY start co-changing: two
//! historically-quiet nodes in one repo that both fire in the same latest snapshot.
//! **This is a documented PROXY.** `git_heat_facts` aggregates per file and does not
//! retain the per-commit file-SET, so true same-commit coupling is not recoverable
//! from it; this signal is "both were quiet, both fired in the same interval", which
//! is a heuristic for an emergent coupling edge, not a proof of one. The evidence note
//! says so. A precise same-commit coupling signal needs a per-commit-file-set
//! producer (logged as the upgrade).
//! - **(d) breadth outlier** — a snapshot whose change BREADTH (how many distinct files
//! moved in that interval) is a statistical outlier vs the repo's history. A commit
//! wave touching anomalously many files (a sweeping refactor / a merge storm).
//!
//! **Honest about statistics.** A z-score needs a baseline; with fewer than
//! `min_history` snapshots a node has no baseline and is NOT judged (reported as
//! `insufficient_history`, never fabricated as clean or as anomalous). Every anomaly
//! carries `baseline` vs `observed` so the verdict is traceable, never a lone score.
//!
//! Attribution: Adam Tornhill, *Your Code as a Crime Scene* (2015) / *Software Design
//! X-Rays* (2018) — behavioural code analysis, hotspots, change (temporal) coupling.
//! z-score / EWMA change-point detection is standard SPC.
use serde::{Deserialize, Serialize};
use super::change_series::{NodeSeries, ewma, mean_std};
use super::sources::{Feed, evidence_note};
/// The kind of anomaly.
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
#[serde(rename_all = "kebab-case")]
pub enum Kind {
/// (a) A normally-stable node that suddenly churns.
StableNodeChurn,
/// (b) A change-rate spike on an already-active node.
Spike,
/// (d) A snapshot touching anomalous BREADTH (outlier fan-out).
BreadthOutlier,
/// (c) A NEW temporal-coupling candidate (documented proxy).
NewCoupling,
}
impl Kind {
pub fn label(self) -> &'static str {
match self {
Kind::StableNodeChurn => "stable-node-churn",
Kind::Spike => "spike",
Kind::BreadthOutlier => "breadth-outlier",
Kind::NewCoupling => "new-coupling",
}
}
}
/// One anomaly, with its baseline-vs-observed evidence.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct Anomaly {
pub repo: String,
/// The node (`repo/file`); for a breadth outlier this is the snapshot id, for a
/// coupling candidate the `"a ~ b"` pair.
pub subject: String,
pub kind: Kind,
/// The baseline the observation is judged against (mean per-interval change, or
/// mean breadth). Carried so the score is never a lone verdict.
pub baseline: f64,
/// The observed value (latest interval change count, or latest breadth).
pub observed: f64,
/// The z-score of `observed` against the baseline (`(observed − mean) / std`).
/// A SORT KEY, read beside `baseline`/`observed`.
pub z: f64,
pub evidence_note: String,
}
/// The whole reading.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct AnomalyReport {
pub z_threshold: f64,
pub nodes_seen: usize,
/// Nodes skipped because they had fewer than `min_history` snapshots — no baseline.
pub insufficient_history: usize,
/// The scans a node needed to be judged at all.
pub min_history: usize,
pub anomalies: Vec<Anomaly>,
pub feeds: Vec<Feed>,
pub evidence_note: String,
pub model_attribution: String,
}
pub const ATTRIBUTION: &str = "Behavioural code analysis: Adam Tornhill, \
`Your Code as a Crime Scene` (2015) / `Software Design X-Rays` (2018) — hotspots \
and change (temporal) coupling. Anomaly math: z-score + EWMA change-point \
detection (standard SPC) over the git_heat_facts change-series. The new-coupling \
signal is a documented churn-correlation PROXY, not same-commit coupling.";
impl AnomalyReport {
pub fn state_json(&self) -> serde_json::Value {
serde_json::json!({
"kind": "anomaly",
"z_threshold": self.z_threshold,
"nodes_seen": self.nodes_seen,
"insufficient_history": self.insufficient_history,
"min_history": self.min_history,
"anomalies_found": self.anomalies.len(),
"evidence_note": self.evidence_note,
"model_attribution": self.model_attribution,
"feeds": self.feeds,
"worst": self.anomalies.iter().map(|a| serde_json::json!({
"repo": a.repo,
"subject": a.subject,
"kind": a.kind.label(),
"baseline": a.baseline,
"observed": a.observed,
"z": a.z,
"evidence_note": a.evidence_note,
})).collect::<Vec<_>>(),
})
}
}
/// **Scans of `git_heat_facts` a file needs before it is judged.** Six scans are
/// five intervals; the first is the backfilled history, so the baseline holds at
/// least four observed intervals. Fewer is "not enough history" — measured on
/// loki 2026-10-16 (≈2.25 generations per file), `min_history = 3` turned a
/// one-point baseline and one change into 50 "z 4.0" alarms.
pub const MIN_HISTORY: usize = 6;
/// Tunables for the detector.
#[derive(Debug, Clone, Copy)]
pub struct Params {
/// z above which an observation is an outlier. Default 3.0 (three-sigma).
pub z_threshold: f64,
/// Minimum snapshots a node needs before it can be judged (≥ 3 ⇒ ≥ 2 baseline
/// increments so a std is defined).
pub min_history: usize,
/// A baseline mean per-interval change at or below this counts the node as
/// "stable" — a fire on top of it is a stable-node-churn (kind a), not a spike.
pub stable_baseline: f64,
/// EWMA responsiveness (0,1]. Cross-checks the z-spike against a weighted baseline.
pub ewma_alpha: f64,
}
impl Default for Params {
fn default() -> Self {
Self {
z_threshold: 3.0,
min_history: MIN_HISTORY,
stable_baseline: 0.5,
ewma_alpha: 0.4,
}
}
}
/// **Detect the anomalies.** Pure over the parsed series.
pub fn analyze(series: &[NodeSeries], p: Params, feeds: Vec<Feed>, top_n: usize) -> AnomalyReport {
let mut anomalies: Vec<Anomaly> = Vec::new();
let mut insufficient = 0usize;
// ── (a)+(b): per-node z-score / EWMA on the per-interval change series ───────
for s in series {
let incs = s.increments();
if incs.len() < p.min_history {
insufficient += 1;
continue;
}
// The first increment is the pre-existing history the first scan BACKFILLED,
// not an observed interval change — exclude it so it does not inflate the
// baseline. The change-rate series is the deltas between consecutive scans.
let observed = *incs.last().unwrap() as f64;
let baseline: Vec<f64> = incs[1..incs.len() - 1].iter().map(|&x| x as f64).collect();
let (mean, std) = mean_std(&baseline);
let ewma_base = ewma(&baseline, p.ewma_alpha);
// z against the node's own baseline. std==0 (a perfectly flat baseline): any
// positive change is unbounded-sigma — treat a fire on a flat baseline as a
// strong anomaly with a saturated z so it ranks, but only when it actually
// moved (observed > mean).
let z = if std > 1e-9 {
(observed - mean) / std
} else if observed > mean {
p.z_threshold.max(4.0)
} else {
0.0
};
if z < p.z_threshold || observed <= mean {
continue;
}
let (kind, why) = if mean <= p.stable_baseline {
(
Kind::StableNodeChurn,
format!(
"normally-stable node (baseline {mean:.2} changes/interval, ewma {ewma_base:.2}) \
churned {observed:.0} in the latest interval — z {z:.1}. A rarely-touched \
node moving is a high-risk event."
),
)
} else {
(
Kind::Spike,
format!(
"change-rate spike: baseline {mean:.2}±{std:.2} (ewma {ewma_base:.2}) → \
observed {observed:.0}, z {z:.1}."
),
)
};
anomalies.push(Anomaly {
repo: s.repo.clone(),
subject: format!("{}/{}", s.repo, s.file),
kind,
baseline: mean,
observed,
z,
evidence_note: why,
});
}
// ── (d): per-repo change-BREADTH outlier across snapshots ───────────────────
anomalies.extend(breadth_outliers(series, &p));
// ── (c): NEW temporal-coupling candidates (documented proxy) ────────────────
anomalies.extend(new_coupling(series, &p));
// Most anomalous first (z), then a stable order.
anomalies.sort_by(|a, b| {
b.z.partial_cmp(&a.z)
.unwrap_or(std::cmp::Ordering::Equal)
.then_with(|| a.kind.cmp(&b.kind))
.then_with(|| a.subject.cmp(&b.subject))
});
if top_n > 0 {
anomalies.truncate(top_n);
}
let mut evidence_note = evidence_note(&feeds, "change anomalies");
if insufficient > 0 {
evidence_note.push_str(&format!(
" {insufficient} of {} file(s): not enough history — fewer than {} scans of `git_heat_facts`, not judged (no baseline, no z-score).",
series.len(),
p.min_history
));
}
AnomalyReport {
z_threshold: p.z_threshold,
nodes_seen: series.len(),
insufficient_history: insufficient,
min_history: p.min_history,
anomalies,
feeds,
evidence_note,
model_attribution: ATTRIBUTION.to_string(),
}
}
/// (d) For each repo, the breadth series = number of distinct files that changed
/// (increment > 0) at each snapshot index. If the latest snapshot's breadth is a
/// z-outlier vs the earlier breadths, flag it.
fn breadth_outliers(series: &[NodeSeries], p: &Params) -> Vec<Anomaly> {
use std::collections::BTreeMap;
// repo -> snapshot_id (in order) -> count of files that moved that interval.
let mut per_repo: BTreeMap<String, BTreeMap<String, (i64, i64)>> = BTreeMap::new();
// value = (breadth_count, ts) keyed by snapshot; ts to order.
for s in series {
let incs = s.increments();
for (i, p_) in s.points.iter().enumerate() {
let moved = if incs[i] > 0 { 1 } else { 0 };
let e = per_repo
.entry(s.repo.clone())
.or_default()
.entry(p_.snapshot_id.clone())
.or_insert((0, p_.ts_micros));
e.0 += moved;
}
}
let mut out = Vec::new();
for (repo, snaps) in per_repo {
// order snapshots by ts.
let mut ordered: Vec<(String, i64)> = snaps
.into_iter()
.map(|(sid, (cnt, ts))| (sid, cnt, ts))
.map(|(sid, cnt, ts)| (format!("{ts}\u{1}{sid}"), cnt))
.collect();
ordered.sort_by(|a, b| a.0.cmp(&b.0));
if ordered.len() < p.min_history {
continue;
}
let breadths: Vec<f64> = ordered.iter().map(|(_, c)| *c as f64).collect();
let observed = *breadths.last().unwrap();
let baseline = &breadths[..breadths.len() - 1];
let (mean, std) = mean_std(baseline);
let z = if std > 1e-9 {
(observed - mean) / std
} else if observed > mean {
p.z_threshold.max(4.0)
} else {
0.0
};
if z < p.z_threshold || observed <= mean {
continue;
}
let sid = ordered
.last()
.unwrap()
.0
.split('\u{1}')
.next_back()
.unwrap_or("")
.to_string();
out.push(Anomaly {
repo: repo.clone(),
subject: format!("snapshot {sid}"),
kind: Kind::BreadthOutlier,
baseline: mean,
observed,
z,
evidence_note: format!(
"the latest snapshot of `{repo}` touched {observed:.0} files vs a baseline of \
{mean:.1}±{std:.1} — an anomalously broad change wave (sweeping refactor / merge \
storm), z {z:.1}."
),
});
}
out
}
/// (c) NEW temporal-coupling candidates — PROXY. Within one repo, the files that were
/// historically quiet (baseline mean ≤ `stable_baseline`) and all fired in the SAME
/// latest interval are treated as newly co-changing. Emits their pairs, bounded.
fn new_coupling(series: &[NodeSeries], p: &Params) -> Vec<Anomaly> {
use std::collections::BTreeMap;
let mut fired_by_repo: BTreeMap<String, Vec<String>> = BTreeMap::new();
for s in series {
let incs = s.increments();
if incs.len() < p.min_history {
continue;
}
let observed = *incs.last().unwrap() as f64;
// Exclude the first increment (the backfilled pre-existing history) from the
// "was it historically quiet" baseline.
let baseline: Vec<f64> = incs[1..incs.len() - 1].iter().map(|&x| x as f64).collect();
let (mean, _) = mean_std(&baseline);
// historically quiet AND fired in the latest interval.
if mean <= p.stable_baseline && observed > 0.0 {
fired_by_repo
.entry(s.repo.clone())
.or_default()
.push(s.file.clone());
}
}
let mut out = Vec::new();
for (repo, mut files) in fired_by_repo {
if files.len() < 2 {
continue;
}
files.sort();
// Bounded: cap the fan-out so a big refactor does not emit O(n²) pairs — the
// top few pairs are enough to name the emergent cluster. (A big wave is already
// caught by the breadth outlier.)
let cap = files.len().min(6);
for i in 0..cap {
for j in (i + 1)..cap {
out.push(Anomaly {
repo: repo.clone(),
subject: format!("{} ~ {}", files[i], files[j]),
kind: Kind::NewCoupling,
baseline: 0.0,
observed: 1.0,
// A proxy signal — rank it BELOW the z-scored anomalies.
z: p.z_threshold,
evidence_note: format!(
"PROXY: `{}` and `{}` were both historically quiet and both changed in the \
same latest snapshot of `{repo}` — a candidate NEW coupling edge. \
git_heat_facts has no per-commit file-set, so this is a churn-correlation \
heuristic, not proven same-commit coupling.",
files[i], files[j]
),
});
}
}
}
out
}
#[cfg(test)]
mod tests {
use super::super::change_series::{NodeSeries, parse_series};
use super::super::rows::Table;
use super::*;
fn heat(rows: &[&[&str]]) -> Vec<NodeSeries> {
let cols = [
"repo",
"file",
"snapshot_id",
"ts_micros",
"commits_total",
"commits_30d",
"commits_90d",
"authors_total",
];
parse_series(&Table::new(
cols.iter().map(|s| s.to_string()).collect(),
rows.iter()
.map(|r| r.iter().map(|s| s.to_string()).collect())
.collect(),
))
}
/// A KNOWN stable node that anomalously churns: flat history (0 change per interval)
/// then a burst. The detector must flag it as `stable-node-churn`.
#[test]
fn a_stable_node_that_suddenly_churns_is_flagged() {
// commits_total flat at 5 across six scans, then jumps to 25 — a burst of 20.
let series = heat(&[
&["korp", "src/core.rs", "s1", "100", "5", "0", "0", "1"],
&["korp", "src/core.rs", "s2", "200", "5", "0", "0", "1"],
&["korp", "src/core.rs", "s3", "300", "5", "0", "0", "1"],
&["korp", "src/core.rs", "s4", "400", "5", "0", "0", "1"],
&["korp", "src/core.rs", "s5", "500", "5", "0", "0", "1"],
&["korp", "src/core.rs", "s6", "600", "5", "0", "0", "1"],
&["korp", "src/core.rs", "s7", "700", "25", "20", "20", "2"],
]);
let r = analyze(&series, Params::default(), vec![], 20);
let a = r
.anomalies
.iter()
.find(|a| a.subject == "korp/src/core.rs")
.expect("flagged");
assert_eq!(a.kind, Kind::StableNodeChurn);
assert!(a.observed >= 20.0);
assert!(
a.baseline <= 0.5,
"baseline was ~0 (stable): {}",
a.baseline
);
assert!(a.evidence_note.contains("stable"));
nornir_testmatrix::functional_status(
"nornir/mimir/anomaly",
"stable_node_churn_flagged",
a.kind == Kind::StableNodeChurn && a.observed >= 20.0,
&format!(
"kind={} baseline={:.2} observed={:.0} z={:.1}",
a.kind.label(),
a.baseline,
a.observed,
a.z
),
);
}
/// A node with a normal churn rhythm that spikes is a `spike`, distinct from a
/// stable node firing.
#[test]
fn an_active_node_spike_is_classified_as_spike() {
// increments ~5,5,5,5 then 40 — an active node with a spike.
let series = heat(&[
&["korp", "hot.rs", "s1", "100", "5", "5", "5", "1"],
&["korp", "hot.rs", "s2", "200", "10", "5", "5", "1"],
&["korp", "hot.rs", "s3", "300", "15", "5", "5", "1"],
&["korp", "hot.rs", "s4", "400", "20", "5", "5", "1"],
&["korp", "hot.rs", "s5", "500", "25", "5", "5", "1"],
&["korp", "hot.rs", "s6", "600", "30", "5", "5", "1"],
&["korp", "hot.rs", "s7", "700", "70", "40", "40", "3"],
]);
let r = analyze(&series, Params::default(), vec![], 20);
let a = r
.anomalies
.iter()
.find(|a| a.subject == "korp/hot.rs")
.expect("flagged");
assert_eq!(a.kind, Kind::Spike);
assert!(a.z >= 3.0);
}
/// A node with too little history is NOT judged — reported honestly.
#[test]
fn a_short_history_is_insufficient_not_fabricated() {
let series = heat(&[
&["korp", "new.rs", "s1", "100", "3", "3", "3", "1"],
&["korp", "new.rs", "s2", "200", "9", "6", "6", "1"],
]);
let r = analyze(&series, Params::default(), vec![], 20);
assert_eq!(r.insufficient_history, 1);
assert!(r.anomalies.iter().all(|a| a.subject != "korp/new.rs"));
}
/// **Too few generations are "not enough history", never a z = 4.0 alarm.**
/// Measured on loki 2026-10-16 (≈2.25 `git_heat` generations per file): all 50
/// anomalies were "baseline 0.00, churned 1, z 4.0" — a node with a one- or
/// two-point baseline and one change. Four scans are not a baseline.
#[test]
fn four_scans_are_not_enough_history_for_a_verdict() {
let series = heat(&[
&["korp", "src/core.rs", "s1", "100", "5", "0", "0", "1"],
&["korp", "src/core.rs", "s2", "200", "5", "0", "0", "1"],
&["korp", "src/core.rs", "s3", "300", "5", "0", "0", "1"],
&["korp", "src/core.rs", "s4", "400", "25", "20", "20", "2"],
]);
let r = analyze(&series, Params::default(), vec![], 20);
assert!(r.anomalies.is_empty(), "four scans judged: {:?}", r.anomalies);
assert_eq!(r.insufficient_history, 1);
assert_eq!(Params::default().min_history, MIN_HISTORY);
let v = r.state_json();
assert_eq!(v["min_history"], MIN_HISTORY);
assert!(v["evidence_note"].as_str().unwrap().contains("not enough history"), "{v}");
}
/// A breadth outlier: several snapshots touch ~1 file, then one snapshot touches
/// many. The repo-level breadth anomaly must fire.
#[test]
fn a_broad_change_wave_is_a_breadth_outlier() {
// 6 snapshots. s1..s5 each move file a only; s6 moves 4 files at once.
let mut rows: Vec<Vec<String>> = Vec::new();
let push = |rows: &mut Vec<Vec<String>>, f: &str, sid: &str, ts: &str, tot: &str| {
rows.push(
["korp", f, sid, ts, tot, "0", "0", "1"]
.iter()
.map(|s| s.to_string())
.collect(),
);
};
// file a moves each snapshot; files b,c,d only appear/move at s6.
for i in 1..=6 {
push(&mut rows, "a", &format!("s{i}"), &format!("{i}00"), &i.to_string());
}
for f in ["b", "c", "d"] {
for i in 1..=5 {
push(&mut rows, f, &format!("s{i}"), &format!("{i}00"), "0");
}
push(&mut rows, f, "s6", "600", "5");
}
let cols = [
"repo",
"file",
"snapshot_id",
"ts_micros",
"commits_total",
"commits_30d",
"commits_90d",
"authors_total",
];
let series = parse_series(&Table::new(
cols.iter().map(|s| s.to_string()).collect(),
rows,
));
let r = analyze(&series, Params::default(), vec![], 50);
assert!(
r.anomalies
.iter()
.any(|a| a.kind == Kind::BreadthOutlier && a.repo == "korp"),
"the broad wave at s6 must be a breadth outlier: {:?}",
r.anomalies
);
// …and the quiet files b/c/d that all fired at s3 are new-coupling candidates.
assert!(
r.anomalies.iter().any(|a| a.kind == Kind::NewCoupling),
"historically-quiet files firing together are new-coupling candidates"
);
}
}