crates/nornir-warehouse-embed/src/mimir.rs
//! **Mimir LENS read client** — the THIN gRPC leg for the six `Mimir.*` lens RPCs.
//!
//! The QUERY half of every dwarves lens (generic bloat, zombie features, change
//! risk, similar code, weak tests, bench gaps) lives in **nornir** as a `Mimir.*`
//! RPC (ARCHITECTURE L5a) — nornir owns the warehouse, so a query only dwarves can
//! reach is a query korp's UI and nornir's own viz cannot use. This module is the
//! reusable READ client for those RPCs: the SAME `<guest>-embed` / no-`nornir`-dep
//! shape as the warehouse/workspaces legs beside it, so a THIN dwarves host (t14s
//! against oden) asks the server — which HAS the warehouse and the fresh data — for
//! the lens results instead of trying to fold locally over a warehouse it does not
//! own.
//!
//! Each RPC returns the lens report's `state_json()` (ranked findings + an
//! `evidence_note` that names missing feeds by table) as a JSON string; these
//! functions parse it to a [`serde_json::Value`] and hand it back verbatim. The
//! honest typed [`WarehouseReadError`] is reused, so a thin caller can tell an
//! unreachable server from an unauthorized one from an unserved workspace — never a
//! silent empty (a real finding-set and an outage must never draw the same).
use crate::{norm_endpoint, pb, status_to_read_error, WarehouseReadError};
use anyhow::Context;
/// The intercepted `Mimir` client type (bearer + `nornir-workspace` header) — the
/// exact interceptor shape the warehouse read leg uses.
type MimirClient = pb::mimir_client::MimirClient<
tonic::service::interceptor::InterceptedService<
tonic::transport::Channel,
crate::NornirInterceptor,
>,
>;
/// Run a blocking `Mimir` gRPC call on a DEDICATED OS thread (never a nested-runtime
/// panic — a thin host may drive this from inside its own tokio runtime), with the
/// bearer + `nornir-workspace` header + a 3s connect timeout. A connect failure maps
/// to [`WarehouseReadError::Unreachable`]; the closure maps its RPC `Status` via
/// [`status_to_read_error`]. Mirrors `with_warehouse_client` — the one read-client
/// spawn body per served nornir service (the crate already carries one per client
/// type; this is the `Mimir` one, not a twin of the `Warehouse` one).
fn with_mimir_client<T, F>(
endpoint: &str,
token: &str,
workspace: &str,
f: F,
) -> Result<T, WarehouseReadError>
where
T: Send + 'static,
F: for<'a> FnOnce(
&'a mut MimirClient,
) -> std::pin::Pin<
Box<dyn std::future::Future<Output = Result<T, WarehouseReadError>> + 'a>,
> + Send,
{
let endpoint = norm_endpoint(endpoint);
let token = token.to_string();
let workspace = workspace.to_string();
std::thread::scope(|s| {
s.spawn(|| {
let rt = tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()
.map_err(|e| {
WarehouseReadError::Unreachable(format!("build tokio runtime: {e}"))
})?;
rt.block_on(async move {
// The ONE shared client bearer-builder (Law #1) — no per-site
// `format!("Bearer …").parse()` twin. `None` ⇒ open (empty token).
let bearer = nornir_service::client::bearer(&token).map_err(|e| {
WarehouseReadError::AuthFailed(format!("bad bearer token: {e}"))
})?;
let ws_md: Option<tonic::metadata::MetadataValue<tonic::metadata::Ascii>> =
(!workspace.is_empty())
.then(|| workspace.parse().ok())
.flatten();
let channel = tonic::transport::Channel::from_shared(endpoint.clone())
.map_err(|e| {
WarehouseReadError::Unreachable(format!("invalid url `{endpoint}`: {e}"))
})?
.connect_timeout(std::time::Duration::from_secs(3))
.timeout(crate::READ_TIMEOUT)
.connect()
.await
.map_err(|e| {
WarehouseReadError::Unreachable(format!("connect to {endpoint}: {e}"))
})?;
let interceptor: crate::NornirInterceptor =
Box::new(move |mut req: tonic::Request<()>| {
if let Some(b) = &bearer {
req.metadata_mut().insert("authorization", b.clone());
}
if let Some(ws) = &ws_md {
req.metadata_mut().insert("nornir-workspace", ws.clone());
}
Ok(req)
});
let mut client =
pb::mimir_client::MimirClient::with_interceptor(channel, interceptor)
.max_decoding_message_size(64 << 20);
f(&mut client).await
})
})
.join()
.map_err(|_| WarehouseReadError::Rpc("mimir-read thread panicked".into()))?
})
}
/// Parse a `JsonResponse.json` body into a [`serde_json::Value`]. A malformed body
/// is an RPC-level failure (the server said Ok but sent junk), not a silent empty.
fn parse_json(json: &str, lens: &str) -> Result<serde_json::Value, WarehouseReadError> {
serde_json::from_str(json)
.with_context(|| format!("parse {lens} JsonResponse body"))
.map_err(|e| WarehouseReadError::Rpc(format!("{e:#}")))
}
/// **`Mimir.GenericCost`** — per-generic monomorphisation cost in shipped bytes
/// (`symbol_sizes × binary_facts`). `min_instantiations` 0 = server default (2);
/// `limit` 0 = per-lens default. `repo` "" = all repos.
pub fn generic_cost(
endpoint: &str,
token: &str,
workspace: &str,
repo: &str,
min_instantiations: u32,
limit: u32,
) -> Result<serde_json::Value, WarehouseReadError> {
let repo = repo.to_string();
with_mimir_client(endpoint, token, workspace, move |client| {
Box::pin(async move {
let resp = client
.generic_cost(pb::GenericCostQuery {
repo,
min_instantiations,
limit,
})
.await
.map_err(status_to_read_error)?
.into_inner();
parse_json(&resp.json, "GenericCost")
})
})
}
/// **`Mimir.ZombieFeatures`** — cargo features whose gated code has no reachable
/// caller (`feature_gate_facts × call_edges × symbol_sizes`).
pub fn zombie_features(
endpoint: &str,
token: &str,
workspace: &str,
repo: &str,
limit: u32,
) -> Result<serde_json::Value, WarehouseReadError> {
let repo = repo.to_string();
with_mimir_client(endpoint, token, workspace, move |client| {
Box::pin(async move {
let resp = client
.zombie_features(pb::LensQuery { repo, limit })
.await
.map_err(status_to_read_error)?
.into_inner();
parse_json(&resp.json, "ZombieFeatures")
})
})
}
/// **`Mimir.ChangeRisk`** — Tornhill's behavioural risk over `git_heat_facts ×
/// call_edges × coverage`, the three inputs carried verbatim beside the score.
pub fn change_risk(
endpoint: &str,
token: &str,
workspace: &str,
repo: &str,
limit: u32,
) -> Result<serde_json::Value, WarehouseReadError> {
let repo = repo.to_string();
with_mimir_client(endpoint, token, workspace, move |client| {
Box::pin(async move {
let resp = client
.change_risk(pb::LensQuery { repo, limit })
.await
.map_err(status_to_read_error)?
.into_inner();
parse_json(&resp.json, "ChangeRisk")
})
})
}
/// **`Mimir.SimilarCode`** — the relocated `dwarves::dupes` query (`embeddings ×
/// symbol_facts × scip_occurrences`).
pub fn similar_code(
endpoint: &str,
token: &str,
workspace: &str,
repo: &str,
limit: u32,
) -> Result<serde_json::Value, WarehouseReadError> {
let repo = repo.to_string();
with_mimir_client(endpoint, token, workspace, move |client| {
Box::pin(async move {
let resp = client
.similar_code(pb::LensQuery { repo, limit })
.await
.map_err(status_to_read_error)?
.into_inner();
parse_json(&resp.json, "SimilarCode")
})
})
}
/// **`Mimir.SemanticClones`** — SimilarCode's AND-gate INVERTED: independent
/// re-implementations (`embeddings × symbol_facts × call_edges`), high cosine and
/// LOW structural agreement, same-crate pairs allowed. The two lenses PARTITION the
/// high-cosine pair space rather than overlapping it, so a pair is one lens's
/// finding or the other's, never both.
///
/// The server verb has existed since the lens did (`nornir.proto:791`,
/// `server_grpc.rs:2987`); this wrapper did not, which is why the pane had no view
/// and the Dupes pane had to fold locally to show anything.
pub fn semantic_clones(
endpoint: &str,
token: &str,
workspace: &str,
repo: &str,
limit: u32,
) -> Result<serde_json::Value, WarehouseReadError> {
let repo = repo.to_string();
with_mimir_client(endpoint, token, workspace, move |client| {
Box::pin(async move {
let resp = client
.semantic_clones(pb::LensQuery { repo, limit })
.await
.map_err(status_to_read_error)?
.into_inner();
parse_json(&resp.json, "SemanticClones")
})
})
}
/// **`Mimir.SemanticTwins`** — v1's semantic twins on v2: Type-3 copies (cosine
/// ≥ 0.92 AND structural ≥ 0.50) and Type-4 twins (cosine ≥ 0.882 AND structural
/// ≤ 0.35, mutual kNN k = 8); each finding carries its plain `why`.
pub fn semantic_twins(
endpoint: &str,
token: &str,
workspace: &str,
repo: &str,
limit: u32,
) -> Result<serde_json::Value, WarehouseReadError> {
let repo = repo.to_string();
with_mimir_client(endpoint, token, workspace, move |client| {
Box::pin(async move {
let resp = client
.semantic_twins(pb::LensQuery { repo, limit })
.await
.map_err(status_to_read_error)?
.into_inner();
parse_json(&resp.json, "SemanticTwins")
})
})
}
/// **`Mimir.WeakTests`** — tests that cannot go red (`test_outcomes × git_heat_facts
/// × coverage_fn`).
pub fn weak_tests(
endpoint: &str,
token: &str,
workspace: &str,
repo: &str,
limit: u32,
) -> Result<serde_json::Value, WarehouseReadError> {
let repo = repo.to_string();
with_mimir_client(endpoint, token, workspace, move |client| {
Box::pin(async move {
let resp = client
.weak_tests(pb::LensQuery { repo, limit })
.await
.map_err(status_to_read_error)?
.into_inner();
parse_json(&resp.json, "WeakTests")
})
})
}
/// **`Mimir.BenchGaps`** — hot paths with no bencher (`git_heat_facts × call_edges ×
/// coverage_fn × bench_results`).
pub fn bench_gaps(
endpoint: &str,
token: &str,
workspace: &str,
repo: &str,
limit: u32,
) -> Result<serde_json::Value, WarehouseReadError> {
let repo = repo.to_string();
with_mimir_client(endpoint, token, workspace, move |client| {
Box::pin(async move {
let resp = client
.bench_gaps(pb::LensQuery { repo, limit })
.await
.map_err(status_to_read_error)?
.into_inner();
parse_json(&resp.json, "BenchGaps")
})
})
}
/// **`Mimir.Villa`** — IMPURE / shell-out heresy over the CPG (`call_edges ×
/// symbol_facts`). Every `std::process::Command`, raw `system`/`exec*` escape and
/// `process::exit` lights up as a finding; a site in a known-offender repo names
/// the external tool (git/xorriso/osmium/podman/cc) being replaced with pure Rust.
pub fn villa(
endpoint: &str,
token: &str,
workspace: &str,
repo: &str,
limit: u32,
) -> Result<serde_json::Value, WarehouseReadError> {
let repo = repo.to_string();
with_mimir_client(endpoint, token, workspace, move |client| {
Box::pin(async move {
let resp = client
.villa(pb::LensQuery { repo, limit })
.await
.map_err(status_to_read_error)?
.into_inner();
parse_json(&resp.json, "Villa")
})
})
}
/// **`Mimir.Churn`** — the FREQUENCY FILTER over the per-node change-series
/// (`git_heat_facts × symbol_facts`). `band` is "high"/"low"/"band";
/// `cutoff_low`/`cutoff_high` are change-frequency (commits_90d) thresholds.
#[allow(clippy::too_many_arguments)]
pub fn churn(
endpoint: &str,
token: &str,
workspace: &str,
repo: &str,
band: &str,
cutoff_low: i64,
cutoff_high: i64,
limit: u32,
) -> Result<serde_json::Value, WarehouseReadError> {
let repo = repo.to_string();
let band = band.to_string();
with_mimir_client(endpoint, token, workspace, move |client| {
Box::pin(async move {
let resp = client
.churn(pb::ChurnQuery {
repo,
band,
cutoff_low,
cutoff_high,
limit,
})
.await
.map_err(status_to_read_error)?
.into_inner();
parse_json(&resp.json, "Churn")
})
})
}
/// **`Mimir.Anomaly`** — statistical OUTLIERS on the change-series
/// (`git_heat_facts`, all snapshots). `z_threshold` 0 = the 3.0 default.
pub fn anomaly(
endpoint: &str,
token: &str,
workspace: &str,
repo: &str,
z_threshold: f64,
limit: u32,
) -> Result<serde_json::Value, WarehouseReadError> {
let repo = repo.to_string();
with_mimir_client(endpoint, token, workspace, move |client| {
Box::pin(async move {
let resp = client
.anomaly(pb::AnomalyQuery {
repo,
z_threshold,
limit,
})
.await
.map_err(status_to_read_error)?
.into_inner();
parse_json(&resp.json, "Anomaly")
})
})
}
/// **`Mimir.WorldImpact`** — changes ranked by world-wide + constellation downstream
/// (`semver_diffs × call_edges × dep_graph_edges`).
pub fn world_impact(
endpoint: &str,
token: &str,
workspace: &str,
repo: &str,
limit: u32,
) -> Result<serde_json::Value, WarehouseReadError> {
let repo = repo.to_string();
with_mimir_client(endpoint, token, workspace, move |client| {
Box::pin(async move {
let resp = client
.world_impact(pb::LensQuery { repo, limit })
.await
.map_err(status_to_read_error)?
.into_inner();
parse_json(&resp.json, "WorldImpact")
})
})
}
/// **`Mimir.BinaryDrift`** — successive Build Thing artifacts of one variant
/// (builder 2's filings, `symbol_sizes` by sha256) compared symbol by symbol:
/// each change with what explains it (`source` · `toolchain` · `nothing`), and
/// the "explained by its provenance: yes/no" attestation.
pub fn binary_drift(
endpoint: &str,
token: &str,
workspace: &str,
repo: &str,
limit: u32,
) -> Result<serde_json::Value, WarehouseReadError> {
let repo = repo.to_string();
with_mimir_client(endpoint, token, workspace, move |client| {
Box::pin(async move {
let resp = client
.binary_drift(pb::LensQuery { repo, limit })
.await
.map_err(status_to_read_error)?
.into_inner();
parse_json(&resp.json, "BinaryDrift")
})
})
}
/// One `LensQuery` lens verb: the six that joined the trim on 2026-10-16 share
/// this body (DataReach, NameCollision, SymbolIdentity, ReachableVuln).
macro_rules! lens_query_fn {
($(#[$doc:meta])* $name:ident, $verb:literal) => {
$(#[$doc])*
pub fn $name(
endpoint: &str,
token: &str,
workspace: &str,
repo: &str,
limit: u32,
) -> Result<serde_json::Value, WarehouseReadError> {
let repo = repo.to_string();
with_mimir_client(endpoint, token, workspace, move |client| {
Box::pin(async move {
let resp = client
.$name(pb::LensQuery { repo, limit })
.await
.map_err(status_to_read_error)?
.into_inner();
parse_json(&resp.json, $verb)
})
})
}
};
}
lens_query_fn!(
/// **`Mimir.DataReach`** — which code touches which warehouse table, and who
/// can reach it (`TOUCHES_TABLE` × reversed `CALLS`).
data_reach,
"DataReach"
);
lens_query_fn!(
/// **`Mimir.NameCollision`** — `CALLS` landing on merged ids vs the leaf-name
/// baseline (call-graph soundness).
name_collision,
"NameCollision"
);
lens_query_fn!(
/// **`Mimir.SymbolIdentity`** — own definitions whose id the fold merged, by cause.
symbol_identity,
"SymbolIdentity"
);
lens_query_fn!(
/// **`Mimir.ReachableVuln`** — advisories × sbom; `refuse` only when own code
/// CALLS the vulnerable crate, `hold` when it is present but unreached.
reachable_vuln,
"ReachableVuln"
);
lens_query_fn!(
/// **`Mimir.Subclone`** — SUB-ITEM clone regions of ONE repo's source (nornir's
/// SOURCE-door lens: source text × the room's item spans). PER-REPO by
/// construction: `repo = ""` returns the lens's own REFUSAL naming the askable
/// repos (`repos_available`), never a constellation scan. The verb is in dwarves'
/// trimmed proto only for now — a live nornir answers `Unimplemented` until it
/// lands there, and the caller renders that as this one lens's error.
subclone,
"Subclone"
);
lens_query_fn!(
/// **`Mimir.GroupTwins`** — the same semantic GROUP of statements in two
/// functions of ONE repo's source, found across interleaved noise (nornir's
/// second SOURCE-door lens: source text × the room's item spans ×
/// `embeddings_groups`). PER-REPO like [`subclone`]: `repo = ""` returns the
/// lens's REFUSAL naming the askable repos; a scan without group vectors still
/// answers its exact-key pairs and says what is missing.
group_twins,
"GroupTwins"
);
lens_query_fn!(
/// **`Mimir.ReimplementsShared`** — a crate that DEPENDS ON crate B and holds a
/// copy or twin of one of B's functions: call the library's instead. nornir
/// joins its copy / twin pairs (`SemanticTwins`, `SimilarCode`) with the
/// `DEPENDS_ON` edges; each finding names the function to replace, the one to
/// call, `drop_in` / `adapter` and the gain.
reimplements_shared,
"ReimplementsShared"
);
/// **`Mimir.Trust`** — bottleneck trust from the entry surface (act / hold /
/// refuse). `act_at` / `hold_at` 0 = the server's bounds.
pub fn trust(
endpoint: &str,
token: &str,
workspace: &str,
repo: &str,
act_at: f64,
hold_at: f64,
limit: u32,
) -> Result<serde_json::Value, WarehouseReadError> {
let repo = repo.to_string();
with_mimir_client(endpoint, token, workspace, move |client| {
Box::pin(async move {
let resp = client
.trust(pb::TrustQuery {
repo,
act_at,
hold_at,
limit,
})
.await
.map_err(status_to_read_error)?
.into_inner();
parse_json(&resp.json, "Trust")
})
})
}
/// **`Mimir.UndocApi`** — undocumented public API ranked by its file's churn.
/// `min_commits_90d` 0 = the server's floor.
pub fn undoc_api(
endpoint: &str,
token: &str,
workspace: &str,
repo: &str,
min_commits_90d: i64,
limit: u32,
) -> Result<serde_json::Value, WarehouseReadError> {
let repo = repo.to_string();
with_mimir_client(endpoint, token, workspace, move |client| {
Box::pin(async move {
let resp = client
.undoc_api(pb::UndocApiQuery {
repo,
min_commits_90d,
limit,
})
.await
.map_err(status_to_read_error)?
.into_inner();
parse_json(&resp.json, "UndocApi")
})
})
}
/// Is this lens answer one the server has not computed yet — `queued` or
/// `running` on its jera `lenses` lane (nornir answers a lens over a room that
/// is not built that way, at once, and keeps the answer for the next ask)?
pub fn is_waiting(answer: &serde_json::Value) -> bool {
matches!(
answer.pointer("/lens_meta/state").and_then(|s| s.as_str()),
Some("queued" | "running")
)
}
/// **Ask until answered**: call `ask` again while the server says the lens is
/// `queued` / `running`, pausing between asks (0.3 s growing to 3 s), for at
/// most `bound`. The last answer comes back either way — one still queued
/// after `bound` says so itself (its `lens_meta.state` and note name the job).
/// For a caller that wants ONE final answer (the CLI, a jera job); the Lenses
/// tab instead shows each lens's queued state while it waits.
pub fn until_answered(
bound: std::time::Duration,
mut ask: impl FnMut() -> Result<serde_json::Value, WarehouseReadError>,
) -> Result<serde_json::Value, WarehouseReadError> {
let t0 = std::time::Instant::now();
let mut pause = std::time::Duration::from_millis(300);
loop {
let answer = ask()?;
if !is_waiting(&answer) || t0.elapsed() >= bound {
return Ok(answer);
}
std::thread::sleep(pause.min(bound.saturating_sub(t0.elapsed())));
pause = (pause * 3 / 2).min(std::time::Duration::from_secs(3));
}
}