src/mimir_lens/room_feed.rs
//! # What the lenses fold, read from where it lives now
//!
//! A lens has two kinds of input (wiring map `.nornir/mimir-v2-land.md` §6.4):
//!
//! * the workspace's CODE ROOM — the graph (`mimir-room`). Typed queries; the
//! `Evidence` every room answer carries becomes rows of the lens's `feeds`
//! and sentences of its `evidence_note` ([`Basis::room`]);
//! * the workspace's WAREHOUSE — the feeds that are not a graph (git heat,
//! bench and test runs, sbom, advisories, the dep graph) ([`Warehouse`]).
//!
//! This module turns both into the typed inputs the folds in the sibling files
//! take, and the answer back into the lens envelope ([`finish`]). The folds
//! themselves (scoring, ranking, report) do not know where a fact came from.
//!
//! ## What every answer says
//!
//! `feeds` lists each input with its row count; an input with no rows is
//! `missing`, and the note says "a missing producer, not an absence of
//! findings". `lens_meta.state` is one of
//!
//! | state | meaning |
//! |---|---|
//! | `complete` | findings, every input was there |
//! | `partial` | findings, but inputs are missing: a lower bound |
//! | `empty_fact` | nothing found and every input was there |
//! | `empty_gap` | nothing found and inputs are missing: the gap may be why |
//! | a refusal ([`crate::code_room::State`]) | the lens did not compute at all |
//!
//! `lens_meta.lower_bounds` names what the graph cannot see even when every
//! input is there (calls through operators / closures / fn pointers, P6).
#[cfg(feature = "code-room")]
pub mod boundary;
#[cfg(all(test, feature = "code-room"))]
mod boundary_tests;
#[cfg(feature = "code-room")]
pub mod built;
#[cfg(all(test, feature = "code-room"))]
mod built_tests;
#[cfg(feature = "code-room")]
pub mod corpus;
#[cfg(all(test, feature = "code-room"))]
pub mod fixture;
#[cfg(feature = "code-room")]
mod graph;
/// Does a `cfg` predicate require `test`? (The readers outside the lenses ask too.)
#[cfg(feature = "code-room")]
pub use graph::cfg_requires_test;
#[cfg(feature = "code-room")]
mod lenses;
#[cfg(all(test, feature = "code-room"))]
mod room_tests;
#[cfg(all(test, feature = "code-room"))]
mod reach_tests;
#[cfg(all(test, feature = "code-room"))]
mod feed_tests;
#[cfg(all(test, feature = "code-room", feature = "vector"))]
mod vector_tests;
#[cfg(all(feature = "code-room", feature = "vector"))]
mod vectors;
#[cfg(all(feature = "code-room", feature = "vector"))]
mod clones;
#[cfg(all(feature = "code-room", feature = "vector"))]
mod graph_first;
#[cfg(all(test, feature = "code-room", feature = "vector"))]
mod clone_tests;
#[cfg(all(test, feature = "code-room", feature = "vector"))]
mod graph_first_tests;
#[cfg(all(feature = "code-room", feature = "vector"))]
mod reuse;
#[cfg(all(test, feature = "code-room", feature = "vector"))]
mod reuse_tests;
use std::time::Instant;
use serde_json::{Value, json};
use super::rows::Table;
use super::sources::Feed;
use crate::code_room::{Reach, Refusal};
use crate::warehouse::iceberg::IcebergWarehouse;
/// Rows a warehouse table read may return (effectively "the whole table").
const SCAN_ROWS: usize = 5_000_000;
/// Decoded bytes a warehouse table read may hold; a cut is said out loud.
const SCAN_BYTES: usize = 1 << 30;
/// P6, said by every lens that walks CALLS.
pub const CALLS_LOWER_BOUND: &str = "calls through operators, closures and fn pointers are not recorded in the graph (producer gap P6): \
every caller / reachability figure is a LOWER bound";
/// The request a front door sent, by the wire query's field names.
pub struct Query<'a> {
pub lens: &'a str,
/// `""` = every repo of the workspace.
pub repo: &'a str,
pub limit: usize,
params: &'a Value,
started: Instant,
}
impl<'a> Query<'a> {
pub fn new(lens: &'a str, params: &'a Value, default_limit: usize) -> Self {
let limit = params.get("limit").and_then(Value::as_u64).filter(|n| *n > 0).map_or(default_limit, |n| n as usize);
Self { lens, repo: params.get("repo").and_then(Value::as_str).unwrap_or(""), limit, params, started: Instant::now() }
}
pub fn f64(&self, key: &str) -> f64 {
self.params.get(key).and_then(Value::as_f64).unwrap_or(0.0)
}
pub fn i64(&self, key: &str) -> i64 {
self.params.get(key).and_then(Value::as_i64).unwrap_or(0)
}
pub fn str(&self, key: &str) -> &'a str {
self.params.get(key).and_then(Value::as_str).unwrap_or("")
}
pub fn flag(&self, key: &str) -> bool {
self.params.get(key).and_then(Value::as_bool).unwrap_or(false)
}
/// Does `repo` pass the request's repo filter?
pub fn wants(&self, repo: &str) -> bool {
self.repo.is_empty() || self.repo == repo
}
}
/// What an answer rests on: the inputs read, what was missing, what the
/// figures therefore mean. Filled while a lens gathers its inputs.
#[derive(Debug, Default)]
pub struct Basis {
pub feeds: Vec<Feed>,
/// Inputs the answer needed and did not get (each one sentence).
pub gaps: Vec<String>,
/// Limits that hold even with every input present.
pub lower_bounds: Vec<String>,
pub notes: Vec<String>,
/// The room's scope (`repo@seq (sha) … · universe off`), when a room was read.
pub scope: Option<String>,
/// The repos the request covers, when a room was read: what the room
/// misses for ANOTHER repo of the workspace is not this answer's gap.
pub repos: Vec<String>,
}
impl Basis {
fn push_unique(list: &mut Vec<String>, s: String) {
if !list.contains(&s) {
list.push(s);
}
}
pub fn gap(&mut self, s: impl Into<String>) {
Self::push_unique(&mut self.gaps, s.into());
}
pub fn lower_bound(&mut self, s: impl Into<String>) {
Self::push_unique(&mut self.lower_bounds, s.into());
}
pub fn note(&mut self, s: impl Into<String>) {
Self::push_unique(&mut self.notes, s.into());
}
/// Record one input by name and row count (`missing` = it carried no row).
pub fn feed(&mut self, table: impl Into<String>, rows: usize, root: impl Into<String>) {
let table = table.into();
if self.feeds.iter().any(|f| f.table == table) {
return;
}
self.feeds.push(Feed { table, rows, root: root.into(), missing: rows == 0 });
}
/// Record an input whose producer is known to have run: no row is then a
/// FACT about the code (a repo with no cargo feature, no test), not a gap.
pub fn fact(&mut self, table: impl Into<String>, rows: usize, root: impl Into<String>) {
let table = table.into();
if !self.feeds.iter().any(|f| f.table == table) {
self.feeds.push(Feed { table, rows, root: root.into(), missing: false });
}
}
/// Any input missing — a feed without rows, or a named gap?
pub fn has_gaps(&self) -> bool {
!self.gaps.is_empty() || self.feeds.iter().any(|f| f.missing)
}
/// `complete` / `partial` / `empty_fact` / `empty_gap`.
pub fn state(&self, empty: bool) -> &'static str {
match (empty, self.has_gaps()) {
(false, false) => "complete",
(false, true) => "partial",
(true, false) => "empty_fact",
(true, true) => "empty_gap",
}
}
/// The sentences that follow the lens's own note.
fn tail(&self) -> String {
let mut out = String::new();
for g in &self.gaps {
out.push_str(&format!(" MISSING: {g}."));
}
for b in &self.lower_bounds {
out.push_str(&format!(" LOWER BOUND: {b}."));
}
for n in &self.notes {
out.push_str(&format!(" {n}."));
}
out
}
}
/// The warehouse a front door reaches — borrowed when the door holds one, else
/// opened read-only beside the code room (`<builds>`, the directory the code
/// room's `mimir_v2/` sits in), else absent with the reason.
pub struct Warehouse<'a> {
handle: Handle<'a>,
}
enum Handle<'a> {
Borrowed(&'a IcebergWarehouse),
Owned(Box<IcebergWarehouse>),
Absent(String),
}
impl<'a> Warehouse<'a> {
pub fn of(reach: &Reach<'a>) -> Self {
if let Some(wh) = reach.warehouse {
return Self { handle: Handle::Borrowed(wh) };
}
let Some(room) = reach.room else {
return Self { handle: Handle::Absent("this front door reaches no warehouse (no workspace selected)".into()) };
};
let Some(builds) = room.layout().base().parent() else {
return Self { handle: Handle::Absent("the code room has no warehouse directory beside it".into()) };
};
match IcebergWarehouse::open_read_only(builds) {
Ok(wh) => Self { handle: Handle::Owned(Box::new(wh)) },
Err(e) => Self { handle: Handle::Absent(format!("no warehouse readable at {}: {}", builds.display(), first_line(&format!("{e:#}")))) },
}
}
/// The warehouse itself, for a reader with typed tables of its own (the
/// function vectors), or why there is none.
pub fn handle(&self) -> Result<&IcebergWarehouse, &str> {
self.get()
}
fn get(&self) -> Result<&IcebergWarehouse, &str> {
match &self.handle {
Handle::Borrowed(wh) => Ok(wh),
Handle::Owned(wh) => Ok(wh),
Handle::Absent(why) => Err(why),
}
}
/// Read `table` whole (every snapshot generation) and record it in `basis`.
/// A table that cannot be read is EMPTY in the fold and MISSING in the
/// answer, with the reason.
pub fn table(&self, basis: &mut Basis, table: &str) -> Table {
let wh = match self.get() {
Ok(wh) => wh,
Err(why) => {
basis.feed(table, 0, "");
basis.gap(format!("`{table}` was not read — {why}"));
return Table::default();
}
};
match wh.scan_preview_bounded(table, SCAN_ROWS, SCAN_BYTES) {
Ok((p, cut)) => {
if cut.is_cut() {
basis.gap(format!("`{table}` was CUT at {} rows ({cut:?}): figures over it are a lower bound", p.rows.len()));
}
basis.feed(table, p.rows.len(), wh.root().display().to_string());
Table::new(p.columns, p.rows)
}
Err(e) => {
basis.feed(table, 0, wh.root().display().to_string());
basis.note(format!("`{table}` could not be scanned ({})", first_line(&format!("{e:#}"))));
Table::default()
}
}
}
/// [`Self::table`], keeping only each repo's newest snapshot — the tables
/// a scan appends one generation to per run.
pub fn latest(&self, basis: &mut Basis, table: &str) -> Table {
let t = self.table(basis, table);
if t.is_empty() { t } else { super::rows::latest_snapshot_rows(&t) }
}
}
/// Warehouse tables as the provenance-carrying set the folds read.
pub fn sources(tables: Vec<(&str, Table)>) -> super::sources::Sources {
let mut s = super::sources::Sources::default();
s.add_root("warehouse", tables.into_iter().map(|(n, t)| (n.to_string(), t)).collect());
s
}
fn first_line(s: &str) -> &str {
s.lines().next().unwrap_or(s)
}
/// The envelope of a lens that did not compute, by state.
pub fn refused(lens: &str, repo: &str, r: &Refusal) -> Value {
json!({
"kind": lens,
"worst": [],
"feeds": [],
"evidence_note": format!("{lens}: {}", r.note),
"lens_meta": {
"state": r.state.as_str(),
"lens": lens,
"repo_scope": repo,
},
})
}
/// The envelope of a lens that stays `not_wired`, naming the producer it waits for.
pub fn not_wired(lens: &str, repo: &str, producer: &str) -> Value {
let mut v = super::not_wired(lens, repo);
v["evidence_note"] = json!(format!("{lens}: {}. Missing producer: {producer}.", super::NOT_WIRED_NOTE));
v["lens_meta"]["missing_producer"] = json!(producer);
v
}
/// Turn a fold's report (`state_json`) into the lens envelope: the dispatch
/// name as `kind`, the basis as `feeds` + the tail of `evidence_note`, and
/// `lens_meta` with the machine state.
pub fn finish(q: &Query<'_>, basis: Basis, mut report: Value) -> Value {
let ranked = ["worst", "findings", "anomalies", "worst_artifacts", "per_repo"];
let empty = !ranked.iter().any(|k| report.get(*k).and_then(Value::as_array).is_some_and(|a| !a.is_empty()));
if report.get("worst").is_none() {
// one key every pane can render a ranked list from
let alias = ranked.iter().find_map(|k| report.get(*k).cloned()).unwrap_or_else(|| json!([]));
report["worst"] = alias;
}
let own = report.get("evidence_note").and_then(Value::as_str).unwrap_or("").to_string();
let own = if own.is_empty() { format!("{}:", q.lens) } else { own };
report["report_kind"] = report.get("kind").cloned().unwrap_or(Value::Null);
report["kind"] = json!(q.lens);
report["evidence_note"] = json!(format!("{own}{}", basis.tail()));
report["lens_meta"] = json!({
"state": basis.state(empty),
"lens": q.lens,
"repo_scope": q.repo,
"room_scope": basis.scope,
"gaps": basis.gaps,
"lower_bounds": basis.lower_bounds,
"ms": q.started.elapsed().as_millis() as u64,
});
report["feeds"] = serde_json::to_value(&basis.feeds).unwrap_or(Value::Null);
report
}
/// `anomaly` — statistical outliers on the change series. Reads
/// `git_heat_facts` only (every snapshot generation: the history IS the signal).
pub fn anomaly(wh: &Warehouse<'_>, q: &Query<'_>) -> Value {
use super::{anomaly, change_series};
let mut basis = Basis::default();
let heat = wh.table(&mut basis, "git_heat_facts");
let mut series = change_series::parse_series(&heat);
heat_scope(&mut basis, &mut series, q);
let mut params = anomaly::Params::default();
if q.f64("z_threshold") > 0.0 {
params.z_threshold = q.f64("z_threshold");
}
let report = anomaly::analyze(&series, params, basis.feeds.clone(), q.limit);
if report.nodes_seen > 0 && report.insufficient_history == report.nodes_seen {
basis.gap(format!(
"not enough history: none of the {} file(s) has the {} scans of `git_heat_facts` a baseline needs, nothing was judged",
report.nodes_seen, params.min_history
));
}
finish(q, basis, report.state_json())
}
/// Apply the repo filter to the change series, and say what an empty series means.
pub fn heat_scope(basis: &mut Basis, series: &mut Vec<super::change_series::NodeSeries>, q: &Query<'_>) {
if series.is_empty() {
basis.note(
"`git_heat_facts` is written by the consumer's `git_heat` feed once a member is caught up (`code_room::feeds`, one scan per consumed sha): \
no row means that feed has not run for this workspace yet",
);
return;
}
if q.repo.is_empty() {
return;
}
let known: std::collections::BTreeSet<String> = series.iter().map(|s| s.repo.clone()).collect();
series.retain(|s| s.repo == q.repo);
if series.is_empty() {
basis.gap(format!("`git_heat_facts` holds no row of repo `{}` (it has: {})", q.repo, known.into_iter().collect::<Vec<_>>().join(", ")));
}
}
/// `subclone` — sub-item clone regions of ONE repo's checkout
/// (`<ws>/git/<repo>`, beside the warehouse root), with the code room's node
/// spans as the item spans. The lens answers without a room too: every file
/// then goes in span-less, counted, and the note says why.
#[cfg(feature = "vector")]
pub fn subclone(reach: &Reach<'_>, q: &Query<'_>) -> Value {
let root = match (reach.warehouse, reach.room) {
(Some(wh), _) => Some(wh.root().to_path_buf()),
(None, Some(room)) => room.layout().base().parent().map(std::path::Path::to_path_buf),
(None, None) => None,
};
let Some(root) = root else {
return refused(q.lens, q.repo, &Refusal::new(crate::code_room::State::NoRoom, "no workspace in reach of this front door: there is no checkout to scan"));
};
#[cfg(feature = "code-room")]
let v = lenses::subclone(reach, q, &root);
#[cfg(not(feature = "code-room"))]
let v = {
let spans = super::subclone::Spans::none("this binary was built without the `code-room` feature: no item spans");
super::subclone::analyze(&root, q.repo, q.limit, &spans)
};
v
}
/// Without the `vector` feature the sub-item clone detector is not built.
#[cfg(not(feature = "vector"))]
pub fn subclone(_: &Reach<'_>, q: &Query<'_>) -> Value {
refused(q.lens, q.repo, &Refusal::new(crate::code_room::State::NotLinked, "this binary was built without the `vector` feature: the sub-item clone detector is not in it"))
}
/// `group_twins` — the same GROUP of statements in two different functions
/// of ONE repo, from the compiled graph first; the embedding index
/// (checkout + `embeddings_groups`) only adds pairs beside it. Without a
/// room the index still runs, and the note says the graph is missing.
#[cfg(feature = "vector")]
pub fn group_twins(reach: &Reach<'_>, q: &Query<'_>) -> Value {
let root = match (reach.warehouse, reach.room) {
(Some(wh), _) => Some(wh.root().to_path_buf()),
(None, Some(room)) => room.layout().base().parent().map(std::path::Path::to_path_buf),
(None, None) => None,
};
let Some(root) = root else {
return refused(q.lens, q.repo, &Refusal::new(crate::code_room::State::NoRoom, "no workspace in reach of this front door: there is no checkout to scan"));
};
let wh = Warehouse::of(reach);
#[cfg(feature = "code-room")]
let v = lenses::group_twins(reach, &wh, q, &root);
#[cfg(not(feature = "code-room"))]
let v = {
let why = "this binary was built without the `code-room` feature: no item spans, no compiled graph";
let spans = super::subclone::Spans::none(why);
super::group_twins::analyze(wh.handle(), &root, q.repo, q.limit, q.str("model_profile"), &spans, &super::group_twins::Compiled::Unread(why.into()))
};
stamp_source_lens(q, v)
}
/// Without the `vector` feature the group-twin detector is not built.
#[cfg(not(feature = "vector"))]
pub fn group_twins(_: &Reach<'_>, q: &Query<'_>) -> Value {
refused(q.lens, q.repo, &Refusal::new(crate::code_room::State::NotLinked, "this binary was built without the `vector` feature: the group-twin detector is not in it"))
}
/// The envelope keys every lens answer carries, on a source lens's own report
/// — the [`lenses::subclone`] stamp, shared with `group_twins` (same honesty
/// states: a span-less or vector-less scan must never read as clean).
#[cfg(feature = "vector")]
pub fn stamp_source_lens(q: &Query<'_>, mut v: Value) -> Value {
let scanned = v["scanned"].as_bool().unwrap_or(false);
let partial = ["files_without_spans", "files_unparsed", "buckets_skipped", "groups_without_vector", "candidates_unjudged", "functions_untokenised", "functions_without_span", "graph_unread", "graph_candidates_unjudged"]
.iter()
.any(|k| v[*k].as_u64().unwrap_or(0) > 0);
let empty = v["worst"].as_array().is_none_or(Vec::is_empty);
let state = match (scanned, empty, partial) {
(false, _, _) if v["scope_required"].as_bool().unwrap_or(false) => crate::code_room::State::InvalidArgument.as_str(),
(false, _, _) => "empty_gap",
(true, false, false) => "complete",
(true, false, true) => "partial",
(true, true, false) => "empty_fact",
(true, true, true) => "empty_gap",
};
v["lens_meta"] = json!({ "state": state, "lens": q.lens, "repo_scope": q.repo });
v
}
/// A lens that reads the code room: gather, fold, envelope — or the refusal,
/// by state (no room in reach, lake held elsewhere, repo not consumed …).
#[cfg(feature = "code-room")]
pub fn from_room(reach: &Reach<'_>, wh: &Warehouse<'_>, q: &Query<'_>) -> Option<Value> {
Some(match lenses::answer(q.lens, reach, wh, q)? {
Ok((basis, report)) => finish(q, basis, report),
Err(r) => refused(q.lens, q.repo, &r),
})
}
/// Without the `code-room` feature the engine is not linked into this binary
/// (a thin atom): every room-fed lens says so, and the server answers it.
#[cfg(not(feature = "code-room"))]
pub fn from_room(_: &Reach<'_>, _: &Warehouse<'_>, q: &Query<'_>) -> Option<Value> {
let r = Refusal::new(crate::code_room::State::NotLinked, "this binary was built without the `code-room` feature: the lens is answered by the server (gRPC `Mimir.*`)");
Some(refused(q.lens, q.repo, &r))
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn the_state_tells_a_gap_from_a_fact() {
let mut b = Basis::default();
b.feed("git_heat_facts", 3, "wh");
assert_eq!((b.state(false), b.state(true)), ("complete", "empty_fact"));
b.feed("bench_runs", 0, "wh");
assert_eq!((b.state(false), b.state(true)), ("partial", "empty_gap"));
// a lower bound is not a gap
let mut c = Basis::default();
c.feed("code_edges:CALLS", 9, "room");
c.lower_bound(CALLS_LOWER_BOUND);
assert_eq!(c.state(false), "complete");
assert!(c.tail().contains("LOWER BOUND"));
}
#[test]
fn a_door_without_a_warehouse_reports_the_table_missing_with_the_reason() {
let wh = Warehouse::of(&Reach::none());
let mut b = Basis::default();
let t = wh.table(&mut b, "git_heat_facts");
assert!(t.is_empty());
assert!(b.feeds[0].missing);
assert!(b.gaps[0].contains("reaches no warehouse"), "{:?}", b.gaps);
}
#[test]
fn anomaly_without_heat_is_an_empty_gap_that_names_the_producer() {
let params = json!({ "repo": "r" });
let q = Query::new("anomaly", ¶ms, 50);
let v = anomaly(&Warehouse::of(&Reach::none()), &q);
assert_eq!(v["kind"], "anomaly");
assert_eq!(v["lens_meta"]["state"], "empty_gap");
let note = v["evidence_note"].as_str().unwrap();
assert!(note.contains("git_heat_facts") && note.contains("missing producer"), "{note}");
assert_eq!(v["feeds"][0]["missing"], true);
assert!(v["worst"].as_array().unwrap().is_empty());
}
}