nornir 0.5.4

//! nornir's OWN linked bencher registrations (Task #39 — bench-move).
//!
//! The `register_bench!` calls MUST compile INTO the `nornir` library so that
//! `inventory` links them into every binary/test that uses the lib — crucially
//! the release completeness-gate binary. The bencher-discovery seam the coverage
//! gate sweeps, [`nornir_bench::api::registered_bench_ids`], reports exactly the
//! benchers linked into the CURRENT unit; when nornir's benchers lived only in
//! `examples/nornir-bench.rs` (an example is NOT linked into the lib) that seam
//! saw ZERO benchers at gate time and every hot-path fell back to a seeded
//! allowlist excuse — the gate had nothing real to check.
//!
//! With the registrations here, a gate run sees `nornir.dep_graph_build` /
//! `nornir.dep_graph_build_ext`, so the `nornir.dep_graph` hot-path is verified
//! for real (its allowlist entry is deleted). `examples/nornir-bench.rs` stays
//! the RUNNER: it links this lib, so `run_main_json()` iterates these same
//! `inventory` registrations — no bencher is defined in the example any more.
//!
//! ## Benches
//!
//! - `nornir.dep_graph_build` / `nornir.dep_graph_build_ext` — the B-bucket
//!   dependency-Mímir workload, cloned from `tests/scenarios.rs::l3/l4`.
//!   Generates a layered-DAG synthetic workspace of 50 real cargo crates, times
//!   [`dep_graph::build`](crate::warehouse::dep_graph::build)
//!   (which runs `cargo metadata` per repo), then times the warm
//!   transitive-closure queries the Mímir answers (`dependents_transitive` /
//!   `deps_transitive`) over every repo. The `_ext` variant additionally pins 300
//!   locked "external" crates so the build also resolves external-dependency edges.
//!
//! The synthetic-workspace generator is the same one the scenario tests use,
//! `#[path]`-included exactly like `scenarios.rs` does. It is kept private here so
//! moving it into the lib adds NO public API surface — only the `inventory`
//! registrations become visible, via the discovery seam.
#![allow(dead_code)]

use anyhow::Result;

use crate::bench::api::Bencher;
use crate::bench::BenchResult;
use crate::workspace::WorkspaceDescriptor;

#[path = "../tests/support/genworkspace.rs"]
mod genworkspace;
use genworkspace::{gen_workspace, GenSpec, Shape};

/// B-bucket: build a `WorkspaceGraph` over a synthetic layered DAG, then run
/// warm transitive-closure queries over every repo — the dependency-Mímir's
/// hot path.
///
/// This is a **latency / IO microbench, NOT a parallel-compute arm** (bencher
/// O1): the build phase is one `cargo metadata` subprocess per repo (I/O +
/// process bound) and the query phase is a warm in-memory closure battery. The
/// **authoritative metric is `queries_per_sec`** (~600K, run-to-run flat);
/// `build_ms` is host-contention wall-time that swings ~10× and must NOT be read
/// as a compute regression or the arm treated as multi-threaded. The
/// core-saturation LAW therefore exempts it (see
/// [`crate::release::gate::is_latency_microbench`]); the genuinely parallel
/// nornir headline arms are `knowledge_scan` / `warehouse_query`.
struct DepGraphBuild {
    id: &'static str,
    repos: usize,
    ext_crates: usize,
}

impl Bencher for DepGraphBuild {
    fn id(&self) -> &'static str {
        self.id
    }

    fn run(&self) -> Result<BenchResult> {
        // Build the synthetic workspace on disk (real cargo crates + descriptor).
        let mut spec = GenSpec::new(Shape::LayeredDag, self.repos).seed(42);
        if self.ext_crates > 0 {
            spec = spec.ext_crates(self.ext_crates);
        }
        let ws = gen_workspace(&spec);

        // ── Time the graph build (one `cargo metadata` per repo). ──
        let desc = WorkspaceDescriptor::load(&ws.descriptor_path)?;
        let t0 = std::time::Instant::now();
        let g = crate::warehouse::dep_graph::build(&desc)?;
        let build = t0.elapsed();

        // ── Time the warm closure queries (the Mímir's hot answers). ──
        // Two closures per repo, exactly the L3/L4 query battery.
        let t1 = std::time::Instant::now();
        let mut answered = 0u64;
        for name in &ws.repo_names {
            let _ = g.dependents_transitive(name);
            let _ = g.deps_transitive(name);
            answered += 2;
        }
        let queries = t1.elapsed();

        let crates = ws.repo_names.len() as u64;
        let queries_per_sec = if queries.as_secs_f64() > 0.0 {
            answered as f64 / queries.as_secs_f64()
        } else {
            0.0
        };

        let mut metrics = serde_json::Map::new();
        metrics.insert("build_ms".into(), serde_json::json!(build.as_secs_f64() * 1000.0));
        metrics.insert("crates".into(), serde_json::json!(crates));
        metrics.insert("queries".into(), serde_json::json!(answered));
        metrics.insert("queries_per_sec".into(), serde_json::json!(queries_per_sec));
        metrics.insert("ext_crates".into(), serde_json::json!(self.ext_crates));
        metrics.insert(
            "elapsed_ms".into(),
            serde_json::json!((build + queries).as_secs_f64() * 1000.0),
        );

        Ok(BenchResult { name: self.id.into(), metrics , ..Default::default() })
    }
}

// `nornir.dep_graph_build`     — 50 repos, no externals (mirrors L3).
crate::register_bench!(DepGraphBuild {
    id: "nornir.dep_graph_build",
    repos: 50,
    ext_crates: 0,
});

/// The **introspection DELTA** bench (design doc `P0.1`): in an N-file crate,
/// change ONE file, then measure the time to an updated warehouse via the
/// file-level delta rescan vs a full from-scratch rebuild. This is the number
/// that has to beat JetBrains on re-index latency — the whole point of
/// [`crate::knowledge::delta`].
struct DeltaRescan {
    id: &'static str,
    files: usize,
}

impl Bencher for DeltaRescan {
    fn id(&self) -> &'static str {
        self.id
    }

    fn run(&self) -> Result<BenchResult> {
        use crate::knowledge::delta::SymbolDelta;

        // A synthetic single-crate workspace of `files` one-fn modules on disk.
        let dir = tempfile::tempdir()?;
        std::fs::write(
            dir.path().join("Cargo.toml"),
            "[package]\nname = \"deltabench\"\nversion = \"0.0.0\"\nedition = \"2021\"\n",
        )?;
        let src = dir.path().join("src");
        std::fs::create_dir_all(&src)?;
        let mut lib = String::new();
        for i in 0..self.files {
            std::fs::write(
                src.join(format!("m{i}.rs")),
                format!("pub fn f{i}() {{ let _ = {i}; }}\npub struct S{i};\n"),
            )?;
            lib.push_str(&format!("pub mod m{i};\n"));
        }
        std::fs::write(src.join("lib.rs"), lib)?;

        let now = chrono::Utc::now();

        // ── Cold: one full build (parse EVERY file) + its manifest. ──
        let t0 = std::time::Instant::now();
        let mut wh = SymbolDelta::full_build(dir.path(), "deltabench", uuid::Uuid::new_v4(), now)?;
        let full_build = t0.elapsed();
        let baseline_syms = wh.scan.symbols.len();

        // Edit ONE file (mid-tree). Bump mtime so the cheap gate can't miss it.
        let target = src.join(format!("m{}.rs", self.files / 2));
        std::thread::sleep(std::time::Duration::from_millis(12));
        std::fs::write(&target, "pub fn edited() {}\npub struct Edited;\npub struct Extra;\n")?;

        // ── Delta: re-parse ONLY the changed file, merge. ──
        let t1 = std::time::Instant::now();
        let plan = wh.rescan(dir.path(), uuid::Uuid::new_v4(), now)?;
        let delta = t1.elapsed();

        // ── Reference: a full from-scratch rebuild of the mutated tree. ──
        let t2 = std::time::Instant::now();
        let _fresh =
            SymbolDelta::full_build(dir.path(), "deltabench", uuid::Uuid::new_v4(), now)?;
        let full_rebuild = t2.elapsed();

        let speedup = if delta.as_secs_f64() > 0.0 {
            full_rebuild.as_secs_f64() / delta.as_secs_f64()
        } else {
            0.0
        };

        let mut metrics = serde_json::Map::new();
        metrics.insert("files".into(), serde_json::json!(self.files));
        metrics.insert("full_build_ms".into(), serde_json::json!(full_build.as_secs_f64() * 1000.0));
        metrics.insert("delta_rescan_ms".into(), serde_json::json!(delta.as_secs_f64() * 1000.0));
        metrics.insert("full_rebuild_ms".into(), serde_json::json!(full_rebuild.as_secs_f64() * 1000.0));
        metrics.insert("speedup_vs_full".into(), serde_json::json!(speedup));
        metrics.insert("files_touched".into(), serde_json::json!(plan.touched()));
        metrics.insert("baseline_symbols".into(), serde_json::json!(baseline_syms));
        metrics.insert("elapsed_ms".into(), serde_json::json!((full_build + delta + full_rebuild).as_secs_f64() * 1000.0));

        Ok(BenchResult { name: self.id.into(), metrics, ..Default::default() })
    }
}

// `nornir.delta_rescan` — 1-file edit in a 400-file crate → updated warehouse.
crate::register_bench!(DeltaRescan {
    id: "nornir.delta_rescan",
    files: 400,
});

// `nornir.dep_graph_build_ext` — 50 repos + 300 locked externals (mirrors L4).
crate::register_bench!(DepGraphBuild {
    id: "nornir.dep_graph_build_ext",
    repos: 50,
    ext_crates: 300,
});

/// The **knowledge-scan** hot path (`BENCH_HOT_PATHS[nornir.knowledge_scan]`): a
/// from-scratch symbol + call-edge scan of a whole crate. Builds a synthetic
/// N-file crate on disk (each module a fn that CALLS its sibling, so the scan
/// resolves real call-edges too), then times
/// [`crate::knowledge::symbols::scan_repo`] — the parallel `syn` parse that
/// produces the `symbol_facts` / `call_edges` index the introspection warehouse
/// is built from. This is the raw index-build number the delta rescan is
/// measured against.
struct KnowledgeScan {
    id: &'static str,
    files: usize,
}

impl Bencher for KnowledgeScan {
    fn id(&self) -> &'static str {
        self.id
    }

    fn run(&self) -> Result<BenchResult> {
        use crate::knowledge::symbols::scan_repo;

        // A synthetic single-crate workspace of `files` modules, each a fn that
        // calls the next module's fn (so the scan produces real call-edges) plus
        // a struct (so there are type symbols to collect).
        let dir = tempfile::tempdir()?;
        std::fs::write(
            dir.path().join("Cargo.toml"),
            "[package]\nname = \"scanbench\"\nversion = \"0.0.0\"\nedition = \"2021\"\n",
        )?;
        let src = dir.path().join("src");
        std::fs::create_dir_all(&src)?;
        let mut lib = String::new();
        for i in 0..self.files {
            let next = (i + 1) % self.files;
            std::fs::write(
                src.join(format!("m{i}.rs")),
                format!(
                    "pub fn f{i}() {{ let _ = crate::m{next}::g{next}(); }}\n\
                     pub fn g{i}() -> u32 {{ {i} }}\n\
                     pub struct S{i} {{ pub v: u32 }}\n"
                ),
            )?;
            lib.push_str(&format!("pub mod m{i};\n"));
        }
        std::fs::write(src.join("lib.rs"), lib)?;

        let now = chrono::Utc::now();

        // ── Time the full from-scratch scan (parse EVERY file across all cores). ──
        let t0 = std::time::Instant::now();
        let scan = scan_repo(dir.path(), "scanbench", uuid::Uuid::new_v4(), now)?;
        let elapsed = t0.elapsed();

        let symbols = scan.symbols.len() as u64;
        let calls = scan.calls.len() as u64;
        let symbols_per_sec = if elapsed.as_secs_f64() > 0.0 {
            symbols as f64 / elapsed.as_secs_f64()
        } else {
            0.0
        };

        let mut metrics = serde_json::Map::new();
        metrics.insert("files".into(), serde_json::json!(self.files));
        metrics.insert("symbols".into(), serde_json::json!(symbols));
        metrics.insert("call_edges".into(), serde_json::json!(calls));
        metrics.insert("symbols_per_sec".into(), serde_json::json!(symbols_per_sec));
        metrics.insert(
            "elapsed_ms".into(),
            serde_json::json!(elapsed.as_secs_f64() * 1000.0),
        );

        Ok(BenchResult { name: self.id.into(), metrics, ..Default::default() })
    }
}

// `nornir.knowledge_scan` — from-scratch symbol + call-edge scan of a 300-file crate.
crate::register_bench!(KnowledgeScan {
    id: "nornir.knowledge_scan",
    files: 300,
});

/// The **vector-search** hot path (`BENCH_HOT_PATHS[nornir.vector_search]`): the
/// embedding similarity query the semantic-search index answers. Fills a real
/// [`crate::vector::VectorIndex`] with `rows` deterministic `dim`-D vectors
/// (over [`crate::vector::I8_SCAN_THRESHOLD`], so `search_auto` dispatches the
/// same int8/VNNI kernel `store::search` uses in production), then times a
/// battery of top-k nearest-neighbour queries — the exact cosine-similarity
/// scan the `vector_search` MCP tool / CLI runs per query.
#[cfg(feature = "vector")]
struct VectorSearch {
    id: &'static str,
    rows: usize,
    dim: usize,
    queries: usize,
    k: usize,
}

#[cfg(feature = "vector")]
impl Bencher for VectorSearch {
    fn id(&self) -> &'static str {
        self.id
    }

    fn run(&self) -> Result<BenchResult> {
        use crate::vector::VectorIndex;

        // Deterministic pseudo-embedding: a smooth per-(row,dim) function so every
        // vector is distinct and non-degenerate (no external model needed — we are
        // benching the SEARCH kernel, not the embedder).
        let mk = |seed: u64, dim: usize| -> Vec<f32> {
            (0..dim)
                .map(|d| {
                    let x = (seed.wrapping_mul(2_654_435_761).wrapping_add(d as u64)) as f32;
                    (x * 0.000_1).sin() + 0.001 * (d as f32 + 1.0)
                })
                .collect()
        };

        // ── Build the index (excluded from the query timing). ──
        let mut idx = VectorIndex::new(self.dim)?;
        let mut flat = Vec::with_capacity(self.rows * self.dim);
        let mut ids = Vec::with_capacity(self.rows);
        for r in 0..self.rows as u64 {
            flat.extend(mk(r, self.dim));
            ids.push(r * 7 + 3);
        }
        idx.add(&flat, &ids)?;

        // ── Time the top-k query battery (the real similarity search). ──
        let t0 = std::time::Instant::now();
        let mut hits = 0u64;
        for q in 0..self.queries as u64 {
            let query = mk(q.wrapping_mul(31).wrapping_add(1_000), self.dim);
            hits += idx.search_auto(&query, self.k).len() as u64;
        }
        let elapsed = t0.elapsed();

        let queries_per_sec = if elapsed.as_secs_f64() > 0.0 {
            self.queries as f64 / elapsed.as_secs_f64()
        } else {
            0.0
        };

        let mut metrics = serde_json::Map::new();
        metrics.insert("rows".into(), serde_json::json!(self.rows));
        metrics.insert("dim".into(), serde_json::json!(self.dim));
        metrics.insert("queries".into(), serde_json::json!(self.queries));
        metrics.insert("k".into(), serde_json::json!(self.k));
        metrics.insert("hits".into(), serde_json::json!(hits));
        metrics.insert("queries_per_sec".into(), serde_json::json!(queries_per_sec));
        metrics.insert(
            "elapsed_ms".into(),
            serde_json::json!(elapsed.as_secs_f64() * 1000.0),
        );

        Ok(BenchResult { name: self.id.into(), metrics, ..Default::default() })
    }
}

// `nornir.vector_search` — 8192×128-D index, 256 top-10 similarity queries.
#[cfg(feature = "vector")]
crate::register_bench!(VectorSearch {
    id: "nornir.vector_search",
    rows: 8192,
    dim: 128,
    queries: 256,
    k: 10,
});

/// The **warehouse-query** hot path (`BENCH_HOT_PATHS[nornir.warehouse_query]`):
/// a representative columnar read out of the local Iceberg warehouse. Writes one
/// coverage snapshot of `files` files (via the real `rows_for` + `append_report`
/// writer) into a temp-dir warehouse, then times
/// [`crate::warehouse::coverage::query_coverage`] — the Arrow `RecordBatch` scan
/// + decode + per-repo filter + sort that the Test-pane coverage panel and the
/// docs `nornir:gen:coverage` table read through. Fully offline (no network).
struct WarehouseQuery {
    id: &'static str,
    files: usize,
}

impl Bencher for WarehouseQuery {
    fn id(&self) -> &'static str {
        self.id
    }

    fn run(&self) -> Result<BenchResult> {
        use crate::coverage::{summarise, Boundary, FileCoverage, FnCoverage};
        use crate::warehouse::coverage::{append_report, query_coverage, rows_for};
        use crate::warehouse::iceberg::IcebergWarehouse;

        // A synthetic coverage report over `files` files (each with one fn), rolled
        // up by the real `summarise`, so the row shapes match a live `nornir
        // coverage` run exactly.
        let file_cov: Vec<FileCoverage> = (0..self.files)
            .map(|i| {
                let file = format!("src/gen/m{i}.rs");
                FileCoverage {
                    file: file.clone(),
                    krate: "nornir".into(),
                    lines: 20,
                    lines_covered: (i as u64 % 21),
                    regions: 12,
                    regions_covered: (i as u64 % 13),
                    functions: vec![FnCoverage {
                        name: format!("nornir::gen::m{i}::f{i}"),
                        file,
                        lines: 20,
                        lines_covered: (i as u64 % 21),
                        regions: 12,
                        regions_covered: (i as u64 % 13),
                        boundary: Boundary::Core,
                    }],
                }
            })
            .collect();
        let report = summarise("nornir", file_cov)?;
        let (rows, fn_rows) = rows_for(&report, "bench-run", "nornir", 1_000);

        let dir = tempfile::tempdir()?;
        let wh = IcebergWarehouse::open(dir.path())?;

        // ── Write one snapshot (excluded from the read timing). ──
        wh.block_on(append_report(&wh, &rows, &fn_rows))?;

        // ── Time the representative read query. ──
        let t0 = std::time::Instant::now();
        let back = wh.block_on(query_coverage(&wh, "nornir"))?;
        let elapsed = t0.elapsed();

        let read = back.len() as u64;
        let rows_per_sec = if elapsed.as_secs_f64() > 0.0 {
            read as f64 / elapsed.as_secs_f64()
        } else {
            0.0
        };

        let mut metrics = serde_json::Map::new();
        metrics.insert("files".into(), serde_json::json!(self.files));
        metrics.insert("rows_written".into(), serde_json::json!(rows.len()));
        metrics.insert("rows_read".into(), serde_json::json!(read));
        metrics.insert("rows_per_sec".into(), serde_json::json!(rows_per_sec));
        metrics.insert(
            "elapsed_ms".into(),
            serde_json::json!(elapsed.as_secs_f64() * 1000.0),
        );

        Ok(BenchResult { name: self.id.into(), metrics, ..Default::default() })
    }
}

// `nornir.warehouse_query` — read back a 500-file coverage snapshot from Iceberg.
crate::register_bench!(WarehouseQuery {
    id: "nornir.warehouse_query",
    files: 500,
});

/// The **knowledge-graph query** hot path (R0.3 / R2.1): the call-graph questions
/// the MCP `callers_of` / `callees_of` / `call_path` tools and the viz code-graph
/// panel answer over a loaded [`crate::knowledge::query::KnowledgeView`]. Builds a
/// synthetic `nodes`-deep call chain (`f0 → f1 → … → fN`) with a shared fan-in
/// callee (`util::helper`, invoked by every node), warms the memoized inverted
/// index + adjacency ONCE, then times a battery of:
///   * `call_path(f0, fN)` — a full-depth BFS over the whole chain (now a dense
///     CSR integer BFS: nodes interned to `u32` ids once at index time, so the
///     traversal walks flat `Vec` arrays with zero per-node string hashing), and
///   * `callers_of("helper")` — an O(hits) posting-list lookup returning all N
///     fan-in edges.
/// Pure/offline (no warehouse, no network) — a clean scaling number for the
/// query-layer index, complementing `nornir.knowledge_scan` (the build side).
struct KnowledgeQuery {
    id: &'static str,
    nodes: usize,
    queries: usize,
}

impl Bencher for KnowledgeQuery {
    fn id(&self) -> &'static str {
        self.id
    }

    fn run(&self) -> Result<BenchResult> {
        use crate::knowledge::query::KnowledgeView;
        use crate::knowledge::symbols::CallEdgeRow;

        // A `nodes`-deep chain plus a shared fan-in callee, so both the deep
        // `call_path` BFS and the wide `callers_of` posting lookup have real work.
        let mut calls = Vec::with_capacity(self.nodes * 2);
        for i in 0..self.nodes {
            let caller = format!("app::m{i}::f{i}");
            if i + 1 < self.nodes {
                calls.push(CallEdgeRow {
                    crate_name: "app".into(),
                    caller_path: caller.clone(),
                    callee_ident: format!("app::m{}::f{}", i + 1, i + 1),
                    call_kind: "call".into(),
                    file: format!("src/m{i}.rs"),
                    line: 1,
                });
            }
            calls.push(CallEdgeRow {
                crate_name: "app".into(),
                caller_path: caller,
                callee_ident: "util::helper".into(),
                call_kind: "call".into(),
                file: format!("src/m{i}.rs"),
                line: 2,
            });
        }

        let view = KnowledgeView::new(vec![], calls);
        let from = "f0".to_string();
        let to = format!("f{}", self.nodes - 1);

        // Warm the memoized index + adjacency OUTSIDE the timed region (its build
        // is a one-time per-snapshot cost, not the per-query hot path).
        let _ = view.call_path(&from, &to);
        let _ = view.callers_of("helper");

        // ── Time the query battery. ──
        let t0 = std::time::Instant::now();
        let mut path_hops = 0u64;
        let mut fan_in = 0u64;
        for _ in 0..self.queries {
            if let Some(p) = view.call_path(&from, &to) {
                path_hops += p.len() as u64;
            }
            fan_in += view.callers_of("helper").len() as u64;
        }
        let elapsed = t0.elapsed();

        // Two queries per iteration (one deep BFS + one fan-in lookup).
        let answered = self.queries as u64 * 2;
        let queries_per_sec = if elapsed.as_secs_f64() > 0.0 {
            answered as f64 / elapsed.as_secs_f64()
        } else {
            0.0
        };

        // ── R2.1 batched "Ragnar" stree leg ──────────────────────────────
        // The battery above only touches the single-key HashMap path + the CSR
        // BFS; it never routes a BATCH through the sorted-key Eytzinger S+tree
        // (`callers_of_batch`) — Ragnar's pipelined SIMD + software-prefetch walk
        // that overlaps cache-miss latency across the whole batch. Exercise it
        // here so the heavy sweep captures ONE tree walk over the batch vs. N
        // independent HashMap probes. New metrics only; nothing above changes.
        //
        // A mixed batch: every chain leaf `f{i}` (each a 1-hit or 0-hit lookup)
        // plus the shared `helper` (an N-hit fan-in) — real routing work.
        let batch_names: Vec<String> = (0..self.nodes)
            .map(|i| format!("f{i}"))
            .chain(std::iter::once("helper".to_string()))
            .collect();
        let name_refs: Vec<&str> = batch_names.iter().map(|s| s.as_str()).collect();

        // Byte-identical guard (ONCE, outside timing): the batched stree path must
        // return exactly what the per-name HashMap `callers_of` yields, positionally
        // and in the same edge order — the stree only changes HOW a key is located.
        let batched = view.callers_of_batch(&name_refs);
        let per_name: Vec<Vec<u32>> = name_refs
            .iter()
            .map(|n| view.callers_of(n).iter().map(|e| e.line).collect())
            .collect();
        let batch_matches = batched.len() == per_name.len()
            && batched.iter().zip(&per_name).all(|(b, exp)| {
                b.len() == exp.len() && b.iter().zip(exp).all(|(e, &l)| e.line == l)
            });
        debug_assert!(batch_matches, "callers_of_batch must match per-name callers_of");
        let batch_hits: u64 = per_name.iter().map(|v| v.len() as u64).sum();

        // Time the batched path over the same `queries` budget.
        let tb = std::time::Instant::now();
        let mut batch_lookups = 0u64;
        for _ in 0..self.queries {
            for res in view.callers_of_batch(&name_refs) {
                batch_lookups += res.len() as u64;
            }
        }
        let batch_elapsed = tb.elapsed();
        let names_routed = self.queries as u64 * name_refs.len() as u64;
        let batch_names_per_sec = if batch_elapsed.as_secs_f64() > 0.0 {
            names_routed as f64 / batch_elapsed.as_secs_f64()
        } else {
            0.0
        };

        // ── A/B baseline: the SAME batch resolved the OLD way ─────────────
        // The Ragnar batch leg above was timed in isolation, so nothing here
        // proved the stree actually BEATS the plain per-name HashMap loop —
        // only the BENCH RULE's byte-identical half was guarded. Resolve the
        // identical `name_refs` as N sequential single-key `callers_of` probes
        // over the same `queries` budget: that is exactly what a caller pays
        // WITHOUT the batch API, and what the pipelined SIMD + software-prefetch
        // walk must out-run to earn its place. `batch_speedup = seq/batch` is
        // the number the heavy sweep reads to keep-or-drop the R2.1 path.
        let ts = std::time::Instant::now();
        let mut seq_lookups = 0u64;
        for _ in 0..self.queries {
            for n in &name_refs {
                seq_lookups += view.callers_of(n).len() as u64;
            }
        }
        let seq_elapsed = ts.elapsed();
        // Same total work both ways (byte-identical ⇒ identical hit counts): a
        // second, independent cross-check of the equality guard above.
        debug_assert_eq!(
            seq_lookups, batch_lookups,
            "sequential and batched callers_of must resolve the same total edges"
        );
        let batch_speedup = if batch_elapsed.as_secs_f64() > 0.0 {
            seq_elapsed.as_secs_f64() / batch_elapsed.as_secs_f64()
        } else {
            0.0
        };

        let mut metrics = serde_json::Map::new();
        metrics.insert("nodes".into(), serde_json::json!(self.nodes));
        metrics.insert("queries".into(), serde_json::json!(answered));
        metrics.insert("edges".into(), serde_json::json!(view.calls.len()));
        metrics.insert("path_hops".into(), serde_json::json!(path_hops));
        metrics.insert("fan_in_hits".into(), serde_json::json!(fan_in));
        metrics.insert("queries_per_sec".into(), serde_json::json!(queries_per_sec));
        // R2.1 batched stree leg.
        metrics.insert("batch_names".into(), serde_json::json!(name_refs.len()));
        metrics.insert("batch_hits".into(), serde_json::json!(batch_hits));
        metrics.insert("batch_lookups".into(), serde_json::json!(batch_lookups));
        metrics.insert("batch_matches".into(), serde_json::json!(batch_matches));
        metrics.insert("batch_names_per_sec".into(), serde_json::json!(batch_names_per_sec));
        metrics.insert(
            "batch_elapsed_ms".into(),
            serde_json::json!(batch_elapsed.as_secs_f64() * 1000.0),
        );
        // A/B baseline (per-name HashMap loop) + the derived keep-or-drop ratio.
        metrics.insert("seq_lookups".into(), serde_json::json!(seq_lookups));
        metrics.insert(
            "seq_elapsed_ms".into(),
            serde_json::json!(seq_elapsed.as_secs_f64() * 1000.0),
        );
        metrics.insert("batch_speedup".into(), serde_json::json!(batch_speedup));
        metrics.insert(
            "elapsed_ms".into(),
            serde_json::json!(elapsed.as_secs_f64() * 1000.0),
        );

        Ok(BenchResult { name: self.id.into(), metrics, ..Default::default() })
    }
}

// `nornir.knowledge_query` — 2000-node chain, 200 deep-BFS + fan-in query pairs.
crate::register_bench!(KnowledgeQuery {
    id: "nornir.knowledge_query",
    nodes: 2000,
    queries: 200,
});

/// The **funnel Plan-DAG assembly** hot path: build a synthetic funnel whose one
/// idea decomposes into an `n`-node dependency chain, then time
/// [`funnel_core::plan_dag::PlanDag::from_funnel`] — the projection + edge
/// assembly the `nornir viz` Plan-DAG panel runs whenever the plan selection
/// changes. Pure/offline (no warehouse), so it's a clean scaling number for the
/// DAG-over-N-nodes path.
struct PlanDagAssemble {
    id: &'static str,
    nodes: usize,
}

impl Bencher for PlanDagAssemble {
    fn id(&self) -> &'static str {
        self.id
    }

    fn run(&self) -> Result<BenchResult> {
        use funnel_core::event::{Event, ItemKind, PlanStatus};
        use funnel_core::ids::{IdeaId, NodeId, PlanId};
        use funnel_core::plan_dag::PlanDag;
        use funnel_core::state::Funnel;

        // Build the funnel: one idea → one Active plan → an n-node chain a→b→c→…
        let mut f = Funnel::default();
        let now = chrono::Utc::now();
        let idea = IdeaId::seq(1);
        let plan = PlanId::seq(1);
        f.apply(&Event::IdeaSubmitted {
            id: idea.clone(),
            source: "bench".into(),
            text: "assemble".into(),
            refs: vec![],
            item_kind: ItemKind::Idea,
            ts: now,
        })?;
        f.apply(&Event::PlanCreated {
            id: plan.clone(),
            idea_id: idea,
            summary: "chain".into(),
            planner: "bench".into(),
            ts: now,
        })?;
        f.apply(&Event::PlanStatusChanged { plan_id: plan.clone(), status: PlanStatus::Active, why: None, ts: now })?;
        for i in 1..=self.nodes as u64 {
            f.apply(&Event::NodeAdded {
                plan_id: plan.clone(),
                node_id: NodeId::seq(i),
                kind: "code:write".into(),
                params: Default::default(),
                targets: vec![],
                prompt_excerpt: Some(format!("task {i}")),
                ts: now,
            })?;
            if i > 1 {
                f.apply(&Event::EdgeAdded {
                    plan_id: plan.clone(),
                    from_node: NodeId::seq(i - 1),
                    to_node: NodeId::seq(i),
                    ts: now,
                })?;
            }
        }

        // ── Time the assembly (the hot path). ──
        let iters = 200u64;
        let t0 = std::time::Instant::now();
        let mut last_nodes = 0usize;
        for _ in 0..iters {
            let dag = PlanDag::from_funnel(&f, None);
            last_nodes = dag.node_count();
        }
        let elapsed = t0.elapsed();
        let per = elapsed.as_secs_f64() * 1000.0 / iters as f64;

        let mut metrics = serde_json::Map::new();
        metrics.insert("nodes".into(), serde_json::json!(last_nodes));
        metrics.insert("assemble_ms".into(), serde_json::json!(per));
        metrics.insert("elapsed_ms".into(), serde_json::json!(elapsed.as_secs_f64() * 1000.0));
        metrics.insert(
            "assemblies_per_sec".into(),
            serde_json::json!(if elapsed.as_secs_f64() > 0.0 { iters as f64 / elapsed.as_secs_f64() } else { 0.0 }),
        );

        Ok(BenchResult { name: self.id.into(), metrics, ..Default::default() })
    }
}

// `nornir.plan_dag_assemble` — assemble the funnel Plan-DAG over a 200-node chain.
crate::register_bench!(PlanDagAssemble {
    id: "nornir.plan_dag_assemble",
    nodes: 200,
});

/// **Warehouse WRITE hot path** (previously un-benched): append N event-sourced
/// funnel events to the `funnel_events` Iceberg table — **each append is its own
/// snapshot** (the event-sourcing cost). `PlanDagAssemble` above covers the
/// in-memory DAG projection; nothing measured the *write* leg until now.
/// Authoritative metric: `events_per_sec`. Latency/IO microbench (one warehouse,
/// serial appends — not a parallel-compute arm).
///
/// Zero-copy / allocation candidate (see `.nornir/warehouse-write-bench.md`):
/// each `append_event` builds a fresh single-row Arrow `RecordBatch` (per-append
/// column builders) AND re-derives the `TableIdent` (`name.to_string()`) + runs
/// `ensure_table_schema` every call. **Batching many events into one append**
/// (one snapshot, builders sized once) is the win to prove here.
struct FunnelAppend {
    id: &'static str,
    events: usize,
    /// `false` = one Iceberg snapshot per event (the fsync-bound baseline);
    /// `true` = all N events in ONE snapshot via `append_events` (the win).
    batch: bool,
}

impl Bencher for FunnelAppend {
    fn id(&self) -> &'static str {
        self.id
    }

    fn run(&self) -> Result<BenchResult> {
        use crate::funnel::event::{Event, ItemKind};
        use crate::funnel::ids::IdeaId;
        use crate::warehouse::funnel::{append_event, append_events};
        use crate::warehouse::iceberg::IcebergWarehouse;

        let dir = tempfile::tempdir()?;
        let wh = IcebergWarehouse::open(dir.path())?;
        let now = chrono::Utc::now();
        let evs: Vec<Event> = (0..self.events)
            .map(|i| Event::IdeaSubmitted {
                id: IdeaId::seq((i as u64) + 1),
                source: "bench".into(),
                text: format!("bench idea {i}"),
                refs: vec![],
                item_kind: ItemKind::Idea,
                ts: now,
            })
            .collect();

        let batch = self.batch;
        let t0 = std::time::Instant::now();
        wh.block_on(async {
            if batch {
                // One snapshot for all N events (fsync amortized).
                append_events(&wh, &evs).await?;
            } else {
                for ev in &evs {
                    append_event(&wh, ev).await?;
                }
            }
            anyhow::Ok(())
        })?;
        let elapsed = t0.elapsed();

        let secs = elapsed.as_secs_f64();
        let per_sec = if secs > 0.0 { self.events as f64 / secs } else { 0.0 };
        let per_append_us =
            if self.events > 0 { elapsed.as_micros() as f64 / self.events as f64 } else { 0.0 };

        let mut metrics = serde_json::Map::new();
        metrics.insert("events".into(), serde_json::json!(self.events));
        metrics.insert("append_ms".into(), serde_json::json!(secs * 1000.0));
        metrics.insert("events_per_sec".into(), serde_json::json!(per_sec));
        metrics.insert("per_append_us".into(), serde_json::json!(per_append_us));
        metrics.insert("elapsed_ms".into(), serde_json::json!(secs * 1000.0));
        metrics.insert("mode".into(), serde_json::json!(if self.batch { "batch" } else { "per_event" }));
        metrics.insert("snapshots".into(), serde_json::json!(if self.batch { 1 } else { self.events }));
        Ok(BenchResult { name: self.id.into(), metrics, ..Default::default() })
    }
}

// `nornir.funnel_append` — 64 event-sourced appends, one snapshot each (the
// fsync-bound baseline). `nornir.funnel_append_batch` — the SAME 64 events in ONE
// snapshot via `append_events` (the fsync-amortizing win; this is the path
// `store_subtasks` now takes). Both light by design; the heavy batched-vs-serial
// sweep runs on the quiet bench box (Loki/Odin), never here.
crate::register_bench!(FunnelAppend {
    id: "nornir.funnel_append",
    events: 64,
    batch: false,
});
crate::register_bench!(FunnelAppend {
    id: "nornir.funnel_append_batch",
    events: 64,
    batch: true,
});

/// **Introspection-warehouse WRITE hot path** (previously un-benched): persist one
/// full [`SymbolScan`](crate::knowledge::symbols::SymbolScan) —
/// `symbol_facts` + `call_edges` + `feature_gate_facts` + `test_inventory` — into a
/// temp-dir Iceberg warehouse via the real
/// [`append_symbol_scan_async`](crate::warehouse::iceberg::IcebergWarehouse::append_symbol_scan_async).
/// This is the write leg every `nornir scan` takes; `nornir.knowledge_scan` measures
/// the syn PARSE that produces the scan and `nornir.warehouse_query` measures the
/// coverage READ back, but nothing measured the introspection *persist* until now
/// (the funnel write path is covered by `nornir.funnel_append`).
///
/// Already row-batched (one multi-row Arrow `RecordBatch` per table), so — unlike
/// the funnel per-event baseline — this is a one-snapshot-per-table cost: `symbols`
/// symbols + `symbols` call-edges + `symbols/10` feature-gates + `symbols/10` tests
/// commit as FOUR Iceberg snapshots. Authoritative metric: `rows_per_sec`. Latency /
/// IO microbench (one warehouse, serial commits — not a parallel-compute arm). Light
/// by design; the heavy scaling sweep runs on the quiet bench box (Loki/Odin).
struct SymbolAppend {
    id: &'static str,
    symbols: usize,
}

impl Bencher for SymbolAppend {
    fn id(&self) -> &'static str {
        self.id
    }

    fn run(&self) -> Result<BenchResult> {
        use crate::knowledge::symbols::{
            CallEdgeRow, FeatureGateRow, SymbolRow, SymbolScan, TestDefRow,
        };
        use crate::warehouse::iceberg::IcebergWarehouse;

        let n = self.symbols;
        let symbols: Vec<SymbolRow> = (0..n)
            .map(|i| SymbolRow {
                crate_name: "nornir".into(),
                module_path: format!("nornir::gen::m{}", i % 64),
                item_kind: "fn".into(),
                item_name: format!("f{i}"),
                visibility: "pub".into(),
                file: format!("src/gen/m{}.rs", i % 64),
                line: (i as u32 % 400) + 1,
                doc_lines: (i as u32 % 5),
                signature: Some(format!("fn f{i}(x: usize) -> usize")),
            })
            .collect();
        let calls: Vec<CallEdgeRow> = (0..n)
            .map(|i| CallEdgeRow {
                crate_name: "nornir".into(),
                caller_path: format!("nornir::gen::m{}::f{i}", i % 64),
                callee_ident: format!("f{}", (i + 1) % n.max(1)),
                call_kind: "call".into(),
                file: format!("src/gen/m{}.rs", i % 64),
                line: (i as u32 % 400) + 1,
            })
            .collect();
        let features: Vec<FeatureGateRow> = (0..n / 10)
            .map(|i| FeatureGateRow {
                crate_name: "nornir".into(),
                module_path: format!("nornir::gen::m{}", i % 64),
                item_name: format!("gated{i}"),
                cfg_expr: "feature = \"scip\"".into(),
                file: format!("src/gen/m{}.rs", i % 64),
                line: (i as u32 % 400) + 1,
            })
            .collect();
        let tests: Vec<TestDefRow> = (0..n / 10)
            .map(|i| TestDefRow {
                crate_name: "nornir".into(),
                module_path: format!("nornir::gen::m{}::tests", i % 64),
                test_name: format!("t{i}"),
                file: format!("src/gen/m{}.rs", i % 64),
                line: (i as u32 % 400) + 1,
                is_heavy: i % 7 == 0,
                is_async: i % 3 == 0,
            })
            .collect();
        let n_features = features.len();
        let n_tests = tests.len();
        let scan = SymbolScan {
            snapshot_id: uuid::Uuid::new_v4(),
            ts: chrono::Utc::now(),
            repo: "nornir".into(),
            symbols,
            calls,
            features,
            tests,
        };

        let dir = tempfile::tempdir()?;
        let wh = IcebergWarehouse::open(dir.path())?;

        let t0 = std::time::Instant::now();
        wh.block_on(wh.append_symbol_scan_async(&scan))?;
        let elapsed = t0.elapsed();

        let rows_total = (n + n + n_features + n_tests) as u64;
        let secs = elapsed.as_secs_f64();
        let rows_per_sec = if secs > 0.0 { rows_total as f64 / secs } else { 0.0 };
        let per_row_us =
            if rows_total > 0 { elapsed.as_micros() as f64 / rows_total as f64 } else { 0.0 };

        let mut metrics = serde_json::Map::new();
        metrics.insert("symbols".into(), serde_json::json!(n));
        metrics.insert("calls".into(), serde_json::json!(n));
        metrics.insert("features".into(), serde_json::json!(n_features));
        metrics.insert("tests".into(), serde_json::json!(n_tests));
        metrics.insert("rows_total".into(), serde_json::json!(rows_total));
        metrics.insert("append_ms".into(), serde_json::json!(secs * 1000.0));
        metrics.insert("rows_per_sec".into(), serde_json::json!(rows_per_sec));
        metrics.insert("per_row_us".into(), serde_json::json!(per_row_us));
        metrics.insert("elapsed_ms".into(), serde_json::json!(secs * 1000.0));
        // One Iceberg snapshot per non-empty table (symbols, calls, features, tests).
        metrics.insert("snapshots".into(), serde_json::json!(4));
        Ok(BenchResult { name: self.id.into(), metrics, ..Default::default() })
    }
}

// `nornir.symbol_append` — persist a 500-symbol introspection scan (500 symbols +
// 500 call-edges + 50 feature-gates + 50 tests) as four Iceberg snapshots. The core
// `nornir scan` WRITE leg, complementing `nornir.knowledge_scan` (parse) and
// `nornir.warehouse_query` (read). Light by design; the heavy scaling sweep runs on
// the quiet bench box (Loki/Odin), never here.
crate::register_bench!(SymbolAppend {
    id: "nornir.symbol_append",
    symbols: 500,
});

#[cfg(test)]
mod tests {
    /// The whole point of moving the registrations into the lib (Task #39): the
    /// discovery seam MUST be non-empty in a build that links this crate, and MUST
    /// carry the `nornir.dep_graph` benchers — so the completeness gate verifies
    /// that hot-path for real instead of excusing it via an allowlist entry.
    /// Light smoke of the warehouse-WRITE bench (8 events, not the registered 64):
    /// proves `FunnelAppend::run` opens a warehouse, appends event-sourced rows,
    /// and reports a positive `events_per_sec`. The heavy sweep runs on the bench box.
    #[test]
    fn funnel_append_bench_runs_light() {
        use crate::bench::api::Bencher;
        // Per-event baseline (8 events, 8 snapshots).
        let single = super::FunnelAppend { id: "nornir.funnel_append", events: 8, batch: false }
            .run()
            .expect("funnel_append bench runs");
        let eps = single.metrics.get("events_per_sec").and_then(|v| v.as_f64()).unwrap_or(0.0);
        let us = single.metrics.get("per_append_us").and_then(|v| v.as_f64()).unwrap_or(0.0);
        assert!(eps > 0.0, "events_per_sec must be positive: {:?}", single.metrics);
        // Batched (8 events, 1 snapshot) — the amortized path.
        let batched = super::FunnelAppend { id: "nornir.funnel_append_batch", events: 8, batch: true }
            .run()
            .expect("funnel_append_batch bench runs");
        let bus = batched.metrics.get("per_append_us").and_then(|v| v.as_f64()).unwrap_or(0.0);
        assert!(
            batched.metrics.get("events_per_sec").and_then(|v| v.as_f64()).unwrap_or(0.0) > 0.0,
            "batch events_per_sec must be positive: {:?}", batched.metrics
        );
        assert_eq!(batched.metrics.get("snapshots").and_then(|v| v.as_u64()), Some(1), "batch = 1 snapshot");
        eprintln!("nornir.funnel_append light: per_event {us:.0} us/append (8 snap) vs batch {bus:.0} us/append (1 snap)");
    }

    /// Light smoke of the introspection-warehouse WRITE bench (40 symbols, not the
    /// registered 500): proves `SymbolAppend::run` opens a warehouse, persists a full
    /// SymbolScan (symbols + calls + features + tests) across the four introspection
    /// tables, and reports a positive `rows_per_sec`. The heavy scaling sweep runs on
    /// the bench box (Loki/Odin).
    #[test]
    fn symbol_append_bench_runs_light() {
        use crate::bench::api::Bencher;
        let out = super::SymbolAppend { id: "nornir.symbol_append", symbols: 40 }
            .run()
            .expect("symbol_append bench runs");
        let rps = out.metrics.get("rows_per_sec").and_then(|v| v.as_f64()).unwrap_or(0.0);
        let per = out.metrics.get("per_row_us").and_then(|v| v.as_f64()).unwrap_or(0.0);
        assert!(rps > 0.0, "rows_per_sec must be positive: {:?}", out.metrics);
        // 40 symbols + 40 call-edges + 4 features + 4 tests = 88 rows across 4 snapshots.
        assert_eq!(out.metrics.get("rows_total").and_then(|v| v.as_u64()), Some(88), "row accounting: {:?}", out.metrics);
        assert_eq!(out.metrics.get("snapshots").and_then(|v| v.as_u64()), Some(4), "four introspection tables = four snapshots");
        eprintln!("nornir.symbol_append light: {per:.0} us/row, {rps:.0} rows/s (88 rows, 4 snap)");
    }

    /// Light smoke of the knowledge-graph query bench (200-node chain, 20 query
    /// pairs — far below the registered 2000/200): proves it builds a view,
    /// warms the index, answers a full-depth `call_path` and a fan-in
    /// `callers_of`, and reports a positive `queries_per_sec`. The heavy scaling
    /// sweep runs on the bench box (Loki/Odin).
    #[test]
    fn knowledge_query_bench_runs_light() {
        use crate::bench::api::Bencher;
        let out = super::KnowledgeQuery { id: "nornir.knowledge_query", nodes: 200, queries: 20 }
            .run()
            .expect("knowledge_query bench runs");
        let qps = out.metrics.get("queries_per_sec").and_then(|v| v.as_f64()).unwrap_or(0.0);
        assert!(qps > 0.0, "queries_per_sec must be positive: {:?}", out.metrics);
        // A 200-node chain yields a 200-hop path each iteration and 200 fan-in hits.
        assert_eq!(out.metrics.get("path_hops").and_then(|v| v.as_u64()), Some(200 * 20));
        assert_eq!(out.metrics.get("fan_in_hits").and_then(|v| v.as_u64()), Some(200 * 20));
        // R2.1 batched stree leg: 200 chain leaves + `helper` = 201 routed names;
        // the batched path must byte-match the per-name path and report throughput.
        assert_eq!(out.metrics.get("batch_names").and_then(|v| v.as_u64()), Some(201));
        assert_eq!(out.metrics.get("batch_matches").and_then(|v| v.as_bool()), Some(true));
        // f1..f199 (1 hit each) + helper (200) + f0 (0) = 399 hits per pass.
        assert_eq!(out.metrics.get("batch_hits").and_then(|v| v.as_u64()), Some(399));
        assert_eq!(out.metrics.get("batch_lookups").and_then(|v| v.as_u64()), Some(399 * 20));
        assert!(
            out.metrics.get("batch_names_per_sec").and_then(|v| v.as_f64()).unwrap_or(0.0) > 0.0,
            "batch_names_per_sec must be positive: {:?}",
            out.metrics
        );
        // A/B baseline leg: the sequential per-name loop resolves the SAME total
        // edges as the batch (independent cross-check of byte-identity) and the
        // keep-or-drop speedup ratio is present + positive.
        assert_eq!(
            out.metrics.get("seq_lookups").and_then(|v| v.as_u64()),
            out.metrics.get("batch_lookups").and_then(|v| v.as_u64()),
            "seq and batch must resolve identical edge counts: {:?}",
            out.metrics
        );
        assert!(
            out.metrics.get("batch_speedup").and_then(|v| v.as_f64()).unwrap_or(0.0) > 0.0,
            "batch_speedup must be present and positive: {:?}",
            out.metrics
        );
    }

    #[test]
    fn dep_graph_benchers_are_linked_and_discoverable() {
        let ids = crate::bench::api::registered_bench_ids();
        assert!(!ids.is_empty(), "linked build must register at least one bencher: {ids:?}");
        assert!(
            ids.iter().any(|id| id == "nornir.dep_graph_build"),
            "nornir.dep_graph_build must be linked + discoverable: {ids:?}"
        );
        assert!(
            ids.iter().any(|id| id == "nornir.dep_graph_build_ext"),
            "nornir.dep_graph_build_ext must be linked + discoverable: {ids:?}"
        );
    }

    /// The bencher ids satisfy the `nornir.dep_graph` hot-path per the gate's
    /// matching rule (`id == key` or `<key>_`/`<key>.` prefix), so the gate row
    /// resolves covered from the LIVE registry — no allowlist excuse needed.
    #[test]
    fn dep_graph_hot_path_is_covered_by_registry() {
        let registered = crate::bench::api::registered_bench_ids();
        let rows = crate::autonom::bench_rows(&["nornir.dep_graph"], &registered);
        let dg = rows.iter().find(|r| r.module == "nornir.dep_graph").unwrap();
        assert!(dg.has_bencher, "dep_graph must be covered by the linked registry: {registered:?}");
    }

    /// The three formerly-sentinelled hot-paths — `knowledge_scan`,
    /// `warehouse_query`, and (under `vector`) `vector_search` — are now benched
    /// by LINKED registrations whose ids equal the hot-path key, so the
    /// completeness gate resolves them covered from the live registry and their
    /// `autonom-allow.toml` sentinels are deleted (a stale excuse would red the
    /// gate). Mirrors `dep_graph_hot_path_is_covered_by_registry`.
    #[test]
    fn wired_hot_paths_are_covered_by_registry() {
        let registered = crate::bench::api::registered_bench_ids();

        let mut keys = vec!["nornir.knowledge_scan", "nornir.warehouse_query"];
        #[cfg(feature = "vector")]
        keys.push("nornir.vector_search");

        let rows = crate::autonom::bench_rows(&keys, &registered);
        for key in &keys {
            let row = rows.iter().find(|r| &r.module == key).unwrap();
            assert!(
                row.has_bencher,
                "{key} must be covered by the linked registry: {registered:?}"
            );
        }
    }
}