nordisk/nornir
src/server_grpc.rs
//! `nornir-server` — tonic gRPC server. Mirrors the CLI/MCP API one-for-one
//! and is the *intended* single owner of the iceberg warehouse lock. NOTE:
//! today the CLI and `nornir-mcp` still open the warehouse **in-process**
//! (direct `IcebergWarehouse::open`), so they are not yet clients of this
//! process — the tonic-client conversion is still open (see plan.md
//! §"Server / transport").
//!
//! ## Auth (today)
//!
//! Set `NORNIR_SERVER_TOKEN`. Every RPC EXCEPT `Health.Ping` must carry
//! `authorization: Bearer <token>`
//! in its gRPC metadata. Enforced by a tonic interceptor (constant-time token
//! compare); the env var must be ≥ 16 chars or the server refuses to start.
//! `Health.Ping` is deliberately unauthenticated so a liveness probe needs no
//! token — it is registered without the interceptor and matches the proto
//! contract (`proto/nornir.proto`).
//!
//! ## Configuration (flags + env)
//!
//! There is **no boot config file**. Every setting is a `--flag` with the
//! matching env var as fallback — precedence is `--flag` > env > default — so an
//! operator can run the binary by hand while the systemd unit (which sets the
//! envs) keeps working unchanged:
//!
//! - `--token` / `NORNIR_SERVER_TOKEN` (required, ≥16 chars) — bearer secret.
//! If neither is set, falls back to `<home>/.nornir/token`.
//! - `--addr` / `NORNIR_SERVER_ADDR` — bind addr, default `127.0.0.1:7878`.
//!
//! That is the WHOLE server config surface: bind + auth, nothing else. There is no
//! workspace/warehouse/poll boot flag. Workspaces come ONLY from the redb registry
//! at `<home>/.nornir/workspaces`, mutated at RUNTIME via the CLI/gRPC
//! (`nornir workspace register` → `Workspaces.Register`, an idempotent upsert,
//! eagerly served without a restart). The warehouse is `$HOME`-derived per-workspace
//! and the poll cadence is a per-workspace registry field. The registry is
//! opened/created on every boot, and a server with an **empty registry** (zero
//! workspaces) is a valid resting state — it starts cleanly and serves 0.
//!
//! Service coverage: all `proto/nornir.proto` services are implemented
//! (Health, Repos, Guard, Bench, Release, Search, Index, Introspect,
//! Knowledge, Docs, Mimir, Dwarf, Vector, Funnel, Viz). `Docs.Book/Export`
//! need the `docs-export` feature and `Vector.Search` an embedder feature
//! (else UNIMPLEMENTED). Remaining auth work (mTLS + per-user ACL) is Phase 5.
//! ## Reusable server stack
//!
//! This module IS the nornir gRPC server stack (all 20 services + the workspace
//! registry + the monitor poll loop). It was lifted out of `src/bin/nornir-server.rs`
//! so that BOTH the standalone `nornir-server` bin AND the fat `dwarves-full` bin
//! (which stands nornir's `:7878` up side-by-side with dwarves' `:7900` in one
//! process) call the SAME [`run`] entrypoint — no twinned service registration
//! (REUSE law). The bin is now a thin `#[tokio::main]` shell over [`run`].
use std::net::SocketAddr;
use std::path::{Path, PathBuf};
use std::sync::Arc;
use anyhow::{Context, Result, anyhow};
use std::collections::HashMap;
use tokio::sync::Mutex;
use tonic::{Request, Response, Status};
use nornir::config::{self, Loaded};
// The FROZEN `nornir-rpc-<n>` codes every Status below carries (dual channel:
// human string in the message, machine code in the message AND `x-nornir-code`).
use nornir::errcode;
use nornir::warehouse::dep_graph::{query_dep_graph_snapshots, topo_order_from_edges};
use nornir::warehouse::iceberg::IcebergWarehouse;
use nornir::{bench, docs, guard, index, introspect, release};
// The generated `nornir.v1` prost/tonic bindings now live in the `nornir-wire`
// edda leaf; re-exported here as `pb` so every `pb::…_server::…` path below is
// unchanged.
pub use nornir_wire::pb;
use pb::bench_server::{Bench as BenchSvcTrait, BenchServer};
use pb::docs_server::{Docs as DocsSvcTrait, DocsServer};
use pb::dwarf_server::{Dwarf as DwarfSvcTrait, DwarfServer};
use pb::funnel_server::{Funnel as FunnelSvcTrait, FunnelServer};
use pb::guard_server::{Guard as GuardSvcTrait, GuardServer};
use pb::health_server::{Health, HealthServer};
use pb::index_server::{Index as IndexSvcTrait, IndexServer};
use pb::introspect_server::{Introspect as IntrospectSvcTrait, IntrospectServer};
use pb::code_room_server::{CodeRoom as CodeRoomSvcTrait, CodeRoomServer};
use pb::knowledge_server::{Knowledge as KnowledgeSvcTrait, KnowledgeServer};
use pb::mimir_server::{Mimir as MimirSvcTrait, MimirServer};
use pb::release_server::{Release as ReleaseSvcTrait, ReleaseServer};
use pb::repos_server::{Repos as ReposSvcTrait, ReposServer};
use pb::search_server::{Search as SearchSvcTrait, SearchServer};
use pb::telemetry_server::{Telemetry as TelemetrySvcTrait, TelemetryServer};
use pb::vector_server::{Vector as VectorSvcTrait, VectorServer};
// `Viz` is a WIRE service, not a GUI. It used to be `#[cfg(feature = "viz")]` — the
// feature that pulls eframe/egui/facett — on the reasoning that its handlers reuse
// `viz::model`/`viz::knowledge`. Only ONE of its sixteen handlers actually does
// (`Knowledge`; `viz::model` is compiled for `server` too, see `src/viz/mod.rs:16-19`),
// and the old comment here called the fallout "acceptable per the design (§2.5)".
// MEASURED 2026-08-27 on oden, it is not: the forge embeds this lib as
// `nornir { default-features = false, features = ["full-server"] }`, `full-server =
// ["server"]` names no `viz`, so the fat dwarves-server mounted no Viz at all and
//
// nornir jobs --server http://127.0.0.1:7878 --workspace nordisk
// → Error: Viz.Jobs RPC / code: 'Operation is not implemented or not supported'
//
// on today's build (dwarves aa07395, started 11:05) — with the SAME answer one hop out
// at the forge's own :7900, stamped `dwarves-proxy-3:`. That is the whole 🧬 Jera tab
// dark: the proto verb, the thin `nornir-warehouse-embed::viz` client and the forge's
// `VizProxy` all exist and are wired, and the last hop was gated on a window nobody was
// opening. So the SERVICE now rides `server` like every other one, and only the
// `Knowledge` handler keeps the `viz` gate — as a NAMED `Unimplemented`, never a silent
// empty.
use pb::ops_server::{Ops as OpsSvcTrait, OpsServer};
use pb::viz_server::{Viz as VizSvcTrait, VizServer};
use pb::warehouse_server::{Warehouse as WarehouseSvcTrait, WarehouseServer};
use pb::workspaces_server::{Workspaces as WorkspacesSvcTrait, WorkspacesServer};
use nornir::funnel::event::{Event as FunnelEvent, NodeStatus, PlanStatus};
use nornir::funnel::ids::{IdeaId, NodeId, PlanId};
use nornir::funnel::store::Store as FunnelStore;
use nornir::funnel::topo::topo_ready;
use pb::{
BenchHistoryResponse, BinaryOnly, CallQuery, DefinedInRequest, DocExport as PbDocExport,
DocsHistoryRequest, DocsHistoryResponse, DocsResponse, Empty, GateAllResult, GateFailure,
GateResult, GuardPath, GuardReport, IndexBlobFrame, IndexStatsResponse,
KnowledgeCallPathQuery, KnowledgeCallQuery, KnowledgeCalls,
KnowledgeSymbolQuery, KnowledgeSymbols, Kv, Kvf, NameList, OpTarget, OrderRequest,
OrderResponse, PathBetweenRequest, PingResponse, ProgressEvent, RegisterRepoRequest,
RegisterRepoResponse, RegisterWorkspaceRequest, ReloadResponse, RepoList, RepoOnly,
RepoSummary, RunBenchRequest, RunOpResult, RunReleaseRequest, RunTestMatrixRequest,
SearchHit, SearchRequest, SearchResponse, SnapshotRequest, SnapshotResponse,
SubmitBenchRequest, SubmitBenchResponse, Symbol as PbSymbol, SymbolList, SymbolLookupRequest,
UnregisterRepoRequest, UnregisterRepoResponse, WorkspaceFetchReport, WorkspaceFetchRequest,
WorkspaceList, WorkspaceMember, WorkspaceName, WorkspaceRecord,
};
/// Warm process state, opened once at boot.
/// One served workspace: its config, warm Tantivy index, and warehouse handle.
/// The server holds each workspace's redb/iceberg lock for its lifetime.
struct WorkspaceCtx {
/// The served workspace's registry name (e.g. `njord`). Used as the
/// `workspace_name` key when rebuilding the Mímir graph from the warehouse's
/// `dep_graph_edges` (the monitored-workspace fallback).
name: String,
loaded: Mutex<Loaded>,
index: Arc<index::Index>,
warehouse: Arc<IcebergWarehouse>,
/// Lazily-built, cached cross-repo dependency Mímir for this workspace
/// (built from its `nornir-workspace.toml` on first Mimir RPC; the build
/// drives `cargo metadata`, so it's cached for the server's lifetime).
mimir: Mutex<Option<Arc<GraphCtx>>>,
/// Lazily-opened funnel store (idea→plan→node intake DAG), replayed from
/// the workspace's `funnel_events` Iceberg table on first Funnel RPC. Its
/// own warehouse at the home-derived `<home>/.nornir/funnel` (or the config's
/// `[storage].local_path`) — independent of the main warehouse lock,
/// mirroring the MCP.
funnel: Mutex<Option<FunnelStore>>,
/// Unified job ledger (`<warehouse root>/jobs.redb`) — the server is the sole
/// writer (same single-writer discipline as the warehouse). Every server-run
/// op (`Ops.*`, the monitor/sweep) records its lifecycle here; thin clients
/// upsert via `Jobs.Submit`; viz reads via `Viz.Jobs`. See [`nornir::jobs`].
jobs: Arc<nornir::jobs::JobStore>,
/// **Per-workspace work gate (bug #4)** — held across the fetch AND the
/// republish, so `Workspaces.Fetch { force }`, the register-eager populate
/// and the poll sweep cannot run two of either against this workspace's one
/// git tree and one single-writer warehouse (registry lost-update, spurious
/// republish failures, and the false-red clone rows a racing fetch writes).
///
/// The doc that stood here already said "held across the fetch+republish" —
/// but the only acquisition was inside `republish_served`, so the fetch half
/// was never gated at all and the eager-vs-sweep race stayed wide open. The
/// three fetch sites now route through `fetch_served`, which is the one door
/// that takes this gate; nothing else calls `monitor::fetch_workspace*` on
/// the server side.
///
/// Every republish and every fetch runs on a `spawn_blocking` thread, so the
/// gate is never held across an `.await`.
work_gate: nornir::monitor::WorkspaceGate,
/// **The workspace's code room** (Mímir v2) — its own skade lake under
/// `<builds>/mimir_v2/`, opened on first use. Every reader of the code graph
/// goes through [`WorkspaceCtx::reach`] and one of the two dispatch points
/// (`mimir_lens::answer`, `code_room::reader::answer`); no handler opens a
/// lake or names a room query itself.
code_room: Arc<nornir::code_room::CodeRoom>,
/// Where this workspace's consumer gets its dumpers: the nightly driver,
/// probed each time a run is about to start (it can be installed while the
/// server is up). A field so a test can serve a workspace with a scripted
/// dumper — the door, the gate, the jobs and the ledger are then the real ones.
dumpers: DumpersSource,
/// The consumer's BACKGROUND work — backfills and deferred feeds — on jera's
/// chain queues ([`Background`]).
background: Background,
/// **One clone/fetch of `git/` at a time**, apart from the work gate: a
/// member ADDED by a second `register` is cloned at once
/// ([`clone_added_members`]) — a new directory, nothing a running consume
/// reads — while a consume holds the work gate for as long as its
/// `cargo check` takes. Every fetch takes this lock (inside the work gate,
/// never the other way round), so two clones of one member cannot race.
clone_lock: std::sync::Mutex<()>,
}
/// **The background lanes of one served workspace's consumer.**
///
/// Work that must happen and must not happen on the request that asks for it,
/// nor inside a run that holds the workspace's one consumer. It rides jera's
/// chain queue — the queue the lens / feed chain used: idempotent on a key, ONE
/// unit at a time per queue, a `queued` → `running` → `done` / `failed` row in
/// the workspace's job ledger for every unit. Nothing here is a second queue.
///
/// Two lanes (the chain's own rule for work that must not wait for each other):
///
/// * `backfill` — a member's history asked for from a commit
/// (`code_room::backfill`). One at a time per workspace; each unit walks ONE
/// commit per hold of the workspace's gate and lets whoever waits go first.
/// * `feeds` — the deferred feeds (embeddings): long, and holding nothing.
///
/// The queues are in memory and that is deliberate (the chain's design): what
/// is still due is recomputed from the lake on every sweep — an open backfill
/// row, a feed with no ledger row for the member's commit — so a restart
/// resumes from the lake, never from a queue file.
struct Background {
backfill: Arc<nornir::jobs::chain::ChainQueue>,
feeds: Arc<nornir::jobs::chain::ChainQueue>,
/// The `rooms` lane: keep the code room built after a consume
/// (`code_room_warm`) and the dependency ring's embed (`vector_embed`).
rooms: Arc<nornir::jobs::chain::ChainQueue>,
/// The `lenses` lanes: the Mímir lens answers a `Mimir.*` door queued
/// (`mimir_lens_answer`). Their own lanes, so a 40 s `world_impact` never
/// holds a pane's answer on `rooms`; the room builds they need still go
/// one at a time through the room's heavy gate. [`lens_lane`] picks the
/// lane: one lane ran all 21 lenses one after another and the last waited
/// 860 s (live 2026-10-21) behind work that never touched its inputs.
lenses: [Arc<nornir::jobs::chain::ChainQueue>; LENS_LANES],
/// The `panes` lane: the pane answers `CodeRoom.Answer` queued
/// (`code_room_answer`). Its own lane, so a person waiting on a pane (a
/// head's source, a tile of the code map) never waits behind the rooms
/// lane's long work — the dependency ring's embed held every pane answer
/// for minutes (owner 2026-10-17: "it says 'reading source' when I click a
/// header"); the quick questions (a source page, a search, a head's calls)
/// go first on it (`High`).
panes: Arc<nornir::jobs::chain::ChainQueue>,
/// The `lake` lane: the lake's compaction (`lake_compact`, low
/// priority) — uncompressed and small data files rewritten as zstd and
/// merged, one replace snapshot per table. Its own lane: it reads and
/// re-encodes gigabytes and must hold up neither a pane nor the open set.
lake: Arc<nornir::jobs::chain::ChainQueue>,
/// The two workers are spawned once the server's registry is known
/// ([`ensure_background`]).
home: std::sync::OnceLock<(Arc<nornir::registry::Registry>, PathBuf)>,
/// chain key → the job id the queue gave the unit: its turns nest under it
/// and its row carries the progress.
job_ids: std::sync::Mutex<HashMap<String, String>>,
/// Deferred feed work waiting for its unit, by chain key.
deferred: std::sync::Mutex<HashMap<String, nornir::code_room::consumer_loop::Deferred>>,
/// Backfill key → when its last turn failed: resumed after a pause, not on
/// every tick.
failed_at: std::sync::Mutex<HashMap<String, std::time::Instant>>,
/// Pane answers on the `rooms` lane, by chain key: (job id, queued at,
/// started at) — what a repeated ask reports instead of queueing again.
answers: std::sync::Mutex<HashMap<String, (String, std::time::Instant, Option<std::time::Instant>)>>,
/// `groups_backfill` memo: member → (consumed seq, model profile) this
/// process already swept. The sweep builds a room per member, so a consume
/// that moved nothing must not re-pay it; a restart sweeps once more and
/// the content-addressing makes that sweep embed nothing.
groups_done: std::sync::Mutex<HashMap<String, (i64, String)>>,
}
impl Background {
fn new(jobs: &nornir::jobs::JobStore) -> Self {
Self {
backfill: nornir::jobs::chain::ChainQueue::new(jobs.sink()),
feeds: nornir::jobs::chain::ChainQueue::new(jobs.sink()),
rooms: nornir::jobs::chain::ChainQueue::new(jobs.sink()),
lenses: std::array::from_fn(|_| nornir::jobs::chain::ChainQueue::new(jobs.sink())),
panes: nornir::jobs::chain::ChainQueue::new(jobs.sink()),
lake: nornir::jobs::chain::ChainQueue::new(jobs.sink()),
home: std::sync::OnceLock::new(),
job_ids: std::sync::Mutex::new(HashMap::new()),
deferred: std::sync::Mutex::new(HashMap::new()),
failed_at: std::sync::Mutex::new(HashMap::new()),
answers: std::sync::Mutex::new(HashMap::new()),
groups_done: std::sync::Mutex::new(HashMap::new()),
}
}
fn job_id(&self, key: &str) -> Option<String> {
self.job_ids.lock().unwrap_or_else(|e| e.into_inner()).get(key).cloned()
}
fn note_job(&self, key: &str, submitted: &nornir::jobs::chain::Submitted) {
if let Some(id) = submitted.job_id() {
self.job_ids.lock().unwrap_or_else(|e| e.into_inner()).insert(key.to_string(), id.to_string());
}
}
}
/// A fresh [`Dumpers`](nornir::code_room::consumer_loop::Dumpers) for one
/// consumer run, or the typed reason there is none.
type DumpersSource = Arc<
dyn Fn() -> Result<
Box<dyn nornir::code_room::consumer_loop::Dumpers + Send + Sync>,
nornir::code_room::Refusal,
> + Send
+ Sync,
>;
/// The production source: locate the driver from the settings and prove it runs
/// (`code_room::driver`). No driver ⇒ `driver_not_available`, with the step.
fn driver_dumpers() -> DumpersSource {
Arc::new(|| {
let ready = nornir::code_room::driver::Settings::from_env().probe()?;
Ok(Box::new(nornir::code_room::consumer_loop::DriverDumpers(ready)))
})
}
/// Cached dependency graph + the workspace name it was built under (the name
/// keys `release_lineage` for `change::detect`).
struct GraphCtx {
graph: nornir::warehouse::dep_graph::WorkspaceGraph,
workspace_name: String,
}
impl WorkspaceCtx {
/// What this workspace's front doors reach: its code room + its warehouse.
fn reach(&self) -> nornir::code_room::Reach<'_> {
nornir::code_room::Reach::new(&self.code_room, &self.warehouse)
}
/// Get-or-build the cached dependency graph. Resolves the descriptor + runs
/// `cargo metadata` off the `loaded` lock (on a blocking thread).
async fn graph(&self) -> Result<Arc<GraphCtx>, Status> {
if let Some(g) = self.mimir.lock().await.as_ref() {
return Ok(g.clone());
}
let (workspace_root, config_path, members) = {
let l = self.loaded.lock().await;
let members: Vec<String> = l.nornir.repo.keys().cloned().collect();
(l.workspace_root.clone(), l.config_path.clone(), members)
};
let ws_name = self.name.clone();
let wh = self.warehouse.clone();
let (graph, workspace_name) = tokio::task::spawn_blocking(move || {
// For a monitored workspace the synthetic `Loaded.workspace_root` IS
// the `git/` dir the monitor clones each member into as
// `<workspace_root>/<member>` — the on-disk checkout root the
// warehouse fallback needs to read live HEADs for change-detection.
let git_root = workspace_root.clone();
// Preferred: build from a `nornir-workspace.toml` in the checkout
// (drives `cargo metadata` → full graph incl. build order).
match nornir::mimir::build_graph_at(&workspace_root, &config_path) {
Ok(g) => Ok(g),
Err(descriptor_err) => {
// Fallback for monitored workspaces (e.g. njord): the
// checkout has no descriptor, but the monitor has persisted
// the cross-repo edges into `dep_graph_edges`. Rebuild a
// query-only graph from the warehouse the server owns.
match nornir::mimir::build_graph_from_warehouse(
&wh, &ws_name, &members, &git_root,
) {
Ok(Some(g)) => Ok((g, ws_name.clone())),
Ok(None) => Err(anyhow::anyhow!(
"no dependency graph for workspace `{ws_name}`: \
no `nornir-workspace.toml` in the checkout ({descriptor_err}) \
and no `dep_graph_edges` recorded yet (republish to populate it)"
)),
Err(wh_err) => Err(anyhow::anyhow!(
"dep-graph descriptor missing ({descriptor_err}) and \
warehouse fallback failed: {wh_err:#}"
)),
}
}
}
})
.await
.map_err(internal)?
.map_err(internal)?;
let arc = Arc::new(GraphCtx {
graph,
workspace_name,
});
*self.mimir.lock().await = Some(arc.clone());
Ok(arc)
}
/// Funnel store root for this workspace: config `[storage].local_path` else
/// the home-derived `<home>/.nornir/funnel` default. Mirrors the CLI + MCP
/// resolution exactly (no path env-var).
async fn funnel_root(&self) -> PathBuf {
let l = self.loaded.lock().await;
// Single source of truth shared with the CLI + MCP — see
// `FunnelStore::resolve_root`: config `[storage].local_path` >
// home-derived `<home>/.nornir/funnel`.
FunnelStore::resolve_root(&l.workspace_root, &l.nornir.storage.local_path, None)
}
/// Ensure this workspace's funnel store is open, opening it on demand if it
/// is still `None`. After this returns `Ok`, `self.funnel.lock().await` is
/// guaranteed `Some`, so funnel handlers never `.unwrap()`-panic on `None`.
///
/// `read_only` selects the open mode. The funnel root is **process-global**
/// (home-derived), so within one server only the first `WorkspaceCtx`
/// to open it can hold the exclusive `catalog.redb` lock; every other
/// workspace's exclusive open would fail with a lock error. Read-only paths
/// (`Funnel.Show`) therefore pass `read_only = true`, which falls back to a
/// copied-aside read snapshot when the lock is contended — so the Funnel tab
/// loads for every workspace instead of erroring red. Write handlers keep
/// `read_only = false` (a snapshot can't be mutated).
async fn ensure_funnel(&self, read_only: bool) -> Result<(), Status> {
let mut fg = self.funnel.lock().await;
// Re-open when uninitialized, or when a writer needs the store but the
// cached one is a read-only snapshot (a prior `show` opened it under a
// contended lock). The exclusive re-open may still fail with a lock
// error — that's the correct, surfaced outcome for a writer.
let needs_open = match fg.as_ref() {
None => true,
Some(s) => !read_only && s.is_read_only(),
};
if needs_open {
let root = self.funnel_root().await;
let store = if read_only {
FunnelStore::open_read_only_async(root).await
} else {
FunnelStore::open_async(root).await
}
.map_err(internal)?;
*fg = Some(store);
}
Ok(())
}
}
/// Server state: a registry of workspaces keyed by name, with a default.
///
/// Most handlers reach the **default** workspace transparently via `Deref`
/// (single-workspace behavior, unchanged). Per-request routing by the
/// `nornir-workspace` metadata header is opt-in per handler via [`AppState::ws`]
/// (wired handler-by-handler).
struct AppState {
/// Served workspace handles, keyed by name. RUNTIME-MUTABLE: a
/// `Workspaces.Register` (or `Remove`) at runtime inserts (or drops) the served
/// `WorkspaceCtx` here so the workspace is queryable the instant register
/// returns — no server restart. Guarded by a `std::sync::RwLock` (not tokio):
/// readers only ever hold the guard long enough to clone an `Arc` out (never
/// across an `.await`), and the sync poll-loop/republish paths (run on
/// `spawn_blocking`) read it too. INVARIANT: every served handle owns its
/// workspace's redb single-writer lock, so a name appears here at most once.
workspaces: std::sync::RwLock<HashMap<String, Arc<WorkspaceCtx>>>,
/// The server's declarative auth policy (methods + RBAC), shared with the
/// server-wide `PolicyAuthLayer`. PHASE 2c: handlers consult it for per-resource
/// RBAC. A `default()` config (no `methods`, no roles) means both authentication
/// and authorization are OFF — every authz helper below short-circuits to `Ok`,
/// so the whole surface behaves byte-identically to the pre-RBAC server. RBAC
/// only engages when `auth.rbac_enabled()` (roles / repo_roles / default_role
/// configured); the read gate only when `auth.read_gate_enabled()`.
auth: Arc<AuthConfig>,
/// Process-wide embedder for `Vector.Search`, loaded once on first use
/// (model load is ~1s) and shared across workspaces — the embedding model is
/// repo-independent. Only present when built with an embedder backend.
#[cfg(any(feature = "embed-tract", feature = "embed-ort"))]
embedder: Mutex<Option<Arc<dyn nornir::vector::store::Embedder>>>,
}
impl AppState {
/// Look up a served handle by name (cloning the `Arc` out from under the read
/// lock). Used by the sync poll-loop/republish/register paths and by tests.
fn get_ws(&self, name: &str) -> Option<Arc<WorkspaceCtx>> {
self.workspaces
.read()
.expect("workspaces lock poisoned")
.get(name)
.cloned()
}
/// Insert a freshly-opened served handle (the eager-register / boot path). The
/// write lock is held only for the `insert` — never across the long warehouse
/// build, which runs through the served handle afterwards.
fn insert_ws(&self, name: String, ctx: Arc<WorkspaceCtx>) {
self.workspaces
.write()
.expect("workspaces lock poisoned")
.insert(name, ctx);
}
/// Drop a served handle (releasing its redb single-writer lock). Returns the
/// dropped handle so the caller can decide when the underlying redb lock is
/// actually released (when the last `Arc` is dropped).
fn remove_ws(&self, name: &str) -> Option<Arc<WorkspaceCtx>> {
self.workspaces
.write()
.expect("workspaces lock poisoned")
.remove(name)
}
/// Sorted list of served workspace names — used to make "no workspace
/// selected" / "not served" errors actionable (tells the caller what IS
/// available to pick from).
fn served_names(&self) -> Vec<String> {
let mut names: Vec<String> = self
.workspaces
.read()
.expect("workspaces lock poisoned")
.keys()
.cloned()
.collect();
names.sort();
names
}
/// Resolve the workspace for a request from its `nornir-workspace` metadata
/// header. There is NO implicit default route: a request that carries no
/// (or an empty) `nornir-workspace` header is an error — the caller must pick
/// a workspace explicitly. `NotFound` if the name isn't served. Returns an
/// owned `Arc` (cloned from under the read lock) — the map is `RwLock`-guarded,
/// so we can't return a borrow.
fn ws(&self, md: &tonic::metadata::MetadataMap) -> Result<Arc<WorkspaceCtx>, Status> {
self.resolve_ws(md).map(|(_, ctx)| ctx)
}
/// Pure workspace resolver: `(name, ctx)` from the `nornir-workspace` header.
/// NO authorization — the single place the header → served-handle mapping
/// lives. The name is returned so the RBAC doors can key holger's
/// `role_for_repo(<workspace>, subject)` off it.
fn resolve_ws(
&self,
md: &tonic::metadata::MetadataMap,
) -> Result<(String, Arc<WorkspaceCtx>), Status> {
// The per-request workspace-routing extraction is the shared
// `nornir_service::workspace_header` (get → utf-8 → non-empty) — dwarves
// routes the same way under its own `dwarves-workspace` key. The header
// key + the no-workspace policy below stay nornir's.
let header =
nornir_service::workspace_header(md, "nornir-workspace").map(|s| s.to_string());
let name = match header {
Some(n) => n,
None => {
let names = self.served_names();
return Err(not_found(format!(
"no workspace selected — set the `nornir-workspace` header \
(CLI `--workspace <name>`, or pick one in the viz top-bar); served: [{}]",
names.join(", ")
)));
}
};
let ctx = self
.workspaces
.read()
.expect("workspaces lock poisoned")
.get(&name)
.cloned()
.ok_or_else(|| {
let names = self.served_names();
not_found(format!(
"workspace `{name}` is not served (served: [{}])",
names.join(", ")
))
})?;
Ok((name, ctx))
}
// ── PHASE 2c: per-resource RBAC at the workspace choke point ──────────────
// The resource key for nornir is the WORKSPACE: holger's per-repo ACL
// (`role_for_repo(repo, subject)`) is consulted with the `nornir-workspace`
// name as the "repo". `ws`/`ws_write`/`require_write`/`authz_write_named` are the
// ONLY doors that authorize; every one is a strict NO-OP unless RBAC (or the
// opt-in read gate) is configured, so the default (no-policy / static-bearer)
// server is byte-identical to Phase 2b.
/// READ door: resolve the workspace AND authorize the read. The `AuthIdentity`
/// (if any) was stashed in the request extensions by `PolicyAuthLayer`. Read
/// authorization is the ONE `AuthConfig::authorize_read` predicate — a strict
/// `Ok` unless the opt-in read gate (`require_auth_for_reads`) is on, so reads
/// stay open by default exactly as before.
fn ws_read<T>(&self, req: &Request<T>) -> Result<Arc<WorkspaceCtx>, Status> {
let (name, ctx) = self.resolve_ws(req.metadata())?;
self.authz_read(req.extensions().get::<AuthIdentity>(), &name)?;
Ok(ctx)
}
/// WRITE door: resolve the workspace AND require write access. A mutation is
/// also a read, so the read gate is honored first (a writer always `can_read`);
/// then, when RBAC is on, the caller's workspace-scoped role must `can_write`.
fn ws_write<T>(&self, req: &Request<T>) -> Result<Arc<WorkspaceCtx>, Status> {
let (name, ctx) = self.resolve_ws(req.metadata())?;
let id = req.extensions().get::<AuthIdentity>();
self.authz_read(id, &name)?;
self.authz_write(id, &name)?;
Ok(ctx)
}
/// Write gate for a handler that resolves its `WorkspaceCtx` elsewhere (Funnel,
/// via `FunnelSvc::ctx`). A pure `Ok(())` — resolving nothing — when neither RBAC
/// nor the read gate is configured, so the default path is untouched.
fn require_write<T>(&self, req: &Request<T>) -> Result<(), Status> {
if !self.auth.rbac_enabled() && !self.auth.read_gate_enabled() {
return Ok(());
}
let (name, _) = self.resolve_ws(req.metadata())?;
let id = req.extensions().get::<AuthIdentity>();
self.authz_read(id, &name)?;
self.authz_write(id, &name)
}
/// Write gate for a mutation whose resource is named in the request BODY rather
/// than routed by the `nornir-workspace` header (the `Workspaces` lifecycle:
/// Register / Remove / Fetch). `name` is the target workspace.
fn authz_write_named<T>(&self, req: &Request<T>, name: &str) -> Result<(), Status> {
let id = req.extensions().get::<AuthIdentity>();
self.authz_read(id, name)?;
self.authz_write(id, name)
}
/// The read-authorization half: delegate to the single `AuthConfig::authorize_read`
/// predicate (which itself is a no-op unless the read gate is on), mapping its
/// `(u16, msg)` to a gRPC `Status` (401 ⇒ Unauthenticated, else PermissionDenied).
fn authz_read(&self, id: Option<&AuthIdentity>, ws: &str) -> Result<(), Status> {
self.auth.authorize_read(id, ws).map_err(|(code, msg)| {
if code == 401 {
errcode::status(
&errcode::RPC_UNAUTHENTICATED,
tonic::Code::Unauthenticated,
msg,
)
} else {
errcode::status(
&errcode::RPC_PERMISSION_DENIED,
tonic::Code::PermissionDenied,
msg,
)
}
})
}
/// The write-authorization half. NO-OP (byte-identical pre-RBAC write behavior:
/// any authenticated identity may write) unless RBAC is configured. When on: a
/// missing identity ⇒ Unauthenticated; a workspace-scoped role that cannot write
/// ⇒ PermissionDenied.
fn authz_write(&self, id: Option<&AuthIdentity>, ws: &str) -> Result<(), Status> {
if !self.auth.rbac_enabled() {
return Ok(());
}
match id {
None => Err(errcode::status(
&errcode::RPC_UNAUTHENTICATED,
tonic::Code::Unauthenticated,
"authenticated identity required for this operation",
)),
Some(id) => {
if self.auth.role_for_repo(ws, &id.subject).can_write() {
Ok(())
} else {
Err(errcode::status(
&errcode::RPC_PERMISSION_DENIED,
tonic::Code::PermissionDenied,
format!(
"identity `{}` lacks write access to workspace `{ws}`",
id.subject
),
))
}
}
}
}
/// Get-or-load the process-wide embedder (model load is blocking + ~1s; done
/// off the async lock). Cached for the server's lifetime.
#[cfg(any(feature = "embed-tract", feature = "embed-ort"))]
async fn embedder(&self) -> Result<Arc<dyn nornir::vector::store::Embedder>, Status> {
if let Some(e) = self.embedder.lock().await.as_ref() {
return Ok(e.clone());
}
let e: Arc<dyn nornir::vector::store::Embedder> =
tokio::task::spawn_blocking(|| nornir::vector::load_embedder().map(Arc::from))
.await
.map_err(internal)?
.map_err(internal)?;
let mut g = self.embedder.lock().await;
Ok(g.get_or_insert(e).clone())
}
}
type Shared = Arc<AppState>;
/// Resolve a value from a `--flag` (highest precedence) or its env var fallback.
/// Handles both `--flag value` and `--flag=value`. Returns `None` when neither is
/// set, so the caller supplies the default. This is how every server env var gains
/// a matching CLI flag while the existing systemd unit (which uses env) keeps
/// working: flag > env > default.
fn flag_or_env(args: &[String], flag: &str, env: &str) -> Option<String> {
let eq = format!("{flag}=");
let mut it = args.iter();
while let Some(a) = it.next() {
if a == flag {
return it.next().cloned();
}
if let Some(v) = a.strip_prefix(&eq) {
return Some(v.to_string());
}
}
std::env::var(env).ok()
}
// -- Health ------------------------------------------------------------------
// `state` is unused now that `ping` is workspace-INDEPENDENT (liveness must not
// require a selected workspace, and there's no implicit default to look up). Kept
// for construction symmetry with the other services and future health probes.
struct HealthSvc {
#[allow(dead_code)]
state: Shared,
}
#[tonic::async_trait]
impl Health for HealthSvc {
async fn ping(&self, _req: Request<Empty>) -> Result<Response<PingResponse>, Status> {
// Liveness must NOT require a selected workspace — ping answers even on a
// server serving 0 workspaces. `repo_count` is meaningless without a
// selected workspace (there's no implicit default), so it's always 0 here;
// clients query a specific workspace (via the `nornir-workspace` header)
// for real repo counts.
Ok(Response::new(PingResponse {
status: "ok".into(),
// The canonical identity line `v<version> (<git hash>)` — same string
// `nornir --version` prints, so a client's health probe and the CLI
// can never disagree about which build is running.
version: crate::VERSION_LINE.into(),
repo_count: 0,
}))
}
}
// -- Repos -------------------------------------------------------------------
struct ReposSvc {
state: Shared,
}
// The THREE shared Status constructors every handler funnels through. Each stamps
// its frozen `nornir-rpc-<n>` code into BOTH channels — the message text and the
// `x-nornir-code` metadata — so no handler can emit a codeless Status and a thin
// client can branch on the code instead of the prose. See `crate::errcode`.
/// A request field was absent, empty or malformed ⇒ `nornir-rpc-5`.
fn invalid_arg<E: std::fmt::Display>(msg: E) -> Status {
errcode::status(
&errcode::RPC_INVALID_ARGUMENT,
tonic::Code::InvalidArgument,
msg,
)
}
/// The handler failed for an internal reason ⇒ `nornir-rpc-7`.
fn internal<E: std::fmt::Display>(msg: E) -> Status {
errcode::status(&errcode::RPC_INTERNAL, tonic::Code::Internal, msg)
}
/// The addressed resource does not exist ⇒ `nornir-rpc-4`.
fn not_found<E: std::fmt::Display>(msg: E) -> Status {
errcode::status(&errcode::RPC_REPO_UNKNOWN, tonic::Code::NotFound, msg)
}
/// A precondition the caller must satisfy first is unmet ⇒ `nornir-rpc-9`.
///
/// No longer dead: `Workspaces.Repopulate` is the caller — an unserved workspace
/// and a held populate gate are both preconditions, and both refusals go through
/// this ONE door so they carry the same code as every other unmet precondition.
fn failed_precondition<E: std::fmt::Display>(msg: E) -> Status {
errcode::status(
&errcode::RPC_FAILED_PRECONDITION,
tonic::Code::FailedPrecondition,
msg,
)
}
/// The verb is in the proto but not implemented in this build ⇒ `nornir-rpc-8`.
#[allow(dead_code)]
fn unimplemented_verb<E: std::fmt::Display>(msg: E) -> Status {
errcode::status(&errcode::RPC_UNIMPLEMENTED, tonic::Code::Unimplemented, msg)
}
// -- the code-room seam -------------------------------------------------------
//
// The two doors every code-graph handler goes through. A handler builds the
// request JSON (the wire query's fields by their proto names) and decodes the
// answer into its wire type; it never opens the lake and never names a room
// query. Both run on the blocking pool: a room build folds the log.
/// A Mímir lens (`mimir_lens::answer`, lane B) for the served workspace —
/// never computed inside the request when it needs a room: an answer kept for
/// the room version the cache holds now comes back at once; otherwise the lens
/// goes on the workspace's jera `lenses` lane and the answer says `queued` /
/// `running` with the job id (`mimir_lens::queued`). The asker re-asks; one
/// slow lens holds no other.
async fn lens_answer(
wctx: Arc<WorkspaceCtx>,
lens: &'static str,
params: serde_json::Value,
) -> Result<Response<pb::JsonResponse>, Status> {
use nornir::mimir_lens::queued;
let json = tokio::task::spawn_blocking(move || match queued::ready(lens, &wctx.reach(), ¶ms) {
queued::Prepared::Ready(body) => body.to_string(),
queued::Prepared::Due { key } => {
let (job_id, running, ms) = queue_lens_answer(&wctx, lens, ¶ms, &key);
let repo = params.get("repo").and_then(serde_json::Value::as_str).unwrap_or("");
queued::queued_body(lens, repo, &job_id, running, ms).to_string()
}
})
.await
.map_err(internal)?;
Ok(Response::new(pb::JsonResponse { json }))
}
/// Any other reader of the code graph (`code_room::reader::answer`, lane C) for
/// the served workspace. A refusal keeps its state in the status code.
async fn reader_answer(
wctx: Arc<WorkspaceCtx>,
surface: &'static str,
params: serde_json::Value,
) -> Result<serde_json::Value, Status> {
tokio::task::spawn_blocking(move || {
nornir::code_room::reader::answer(surface, &wctx.reach(), ¶ms)
})
.await
.map_err(internal)?
.map_err(|r| refusal_status(surface, &r))
}
/// A code-room [`Refusal`](nornir::code_room::Refusal) as a gRPC status: the
/// verb, then the reason. `not_wired` / `not_linked` are the proto-but-not-here
/// code (`nornir-rpc-8`), a missing precondition is `nornir-rpc-9`.
fn refusal_status(surface: &str, r: &nornir::code_room::Refusal) -> Status {
use nornir::code_room::State;
let msg = format!("{surface}: {r}");
match r.state {
State::NotWired | State::NotLinked => unimplemented_verb(msg),
State::NoRoom
| State::LakeUnavailable
| State::NotConsumed
| State::DriverNotAvailable
| State::NoArtifact
| State::NoEmbedder
| State::VectorsMissing => failed_precondition(msg),
State::InvalidArgument => invalid_arg(msg),
State::Internal => internal(msg),
}
}
/// `{"names": [...]}` → `NameList`.
fn name_list(surface: &str, v: serde_json::Value) -> Result<Response<NameList>, Status> {
#[derive(serde::Deserialize)]
struct Names {
names: Vec<String>,
}
let n: Names = serde_json::from_value(v)
.map_err(|e| internal(format!("{surface}: the reader's answer is not a name list: {e}")))?;
Ok(Response::new(NameList { names: n.names }))
}
/// `{"symbols": [...]}` → `KnowledgeSymbols`.
fn knowledge_symbols(
surface: &str,
v: serde_json::Value,
) -> Result<Response<KnowledgeSymbols>, Status> {
#[derive(serde::Deserialize)]
struct Row {
#[serde(default)]
crate_name: String,
#[serde(default)]
module_path: String,
#[serde(default)]
item_kind: String,
item_name: String,
#[serde(default)]
visibility: String,
#[serde(default)]
file: String,
#[serde(default)]
line: u32,
#[serde(default)]
doc_lines: u32,
#[serde(default)]
signature: String,
}
#[derive(serde::Deserialize)]
struct Rows {
symbols: Vec<Row>,
}
let rows: Rows = serde_json::from_value(v)
.map_err(|e| internal(format!("{surface}: the reader's answer is not symbol rows: {e}")))?;
Ok(Response::new(KnowledgeSymbols {
symbols: rows
.symbols
.into_iter()
.map(|r| pb::KnowledgeSymbol {
crate_name: r.crate_name,
module_path: r.module_path,
item_kind: r.item_kind,
item_name: r.item_name,
visibility: r.visibility,
file: r.file,
line: r.line,
doc_lines: r.doc_lines,
signature: r.signature,
})
.collect(),
}))
}
/// `{"calls": [...]}` → `KnowledgeCalls`.
fn knowledge_calls(
surface: &str,
v: serde_json::Value,
) -> Result<Response<KnowledgeCalls>, Status> {
#[derive(serde::Deserialize)]
struct Row {
#[serde(default)]
crate_name: String,
caller_path: String,
callee_ident: String,
#[serde(default)]
call_kind: String,
#[serde(default)]
file: String,
#[serde(default)]
line: u32,
}
#[derive(serde::Deserialize)]
struct Rows {
calls: Vec<Row>,
}
let rows: Rows = serde_json::from_value(v)
.map_err(|e| internal(format!("{surface}: the reader's answer is not call rows: {e}")))?;
Ok(Response::new(KnowledgeCalls {
calls: rows
.calls
.into_iter()
.map(|r| pb::KnowledgeCall {
crate_name: r.crate_name,
caller_path: r.caller_path,
callee_ident: r.callee_ident,
call_kind: r.call_kind,
file: r.file,
line: r.line,
})
.collect(),
}))
}
#[tonic::async_trait]
impl ReposSvcTrait for ReposSvc {
async fn list(&self, req: Request<Empty>) -> Result<Response<RepoList>, Status> {
let wctx = self.state.ws_read(&req)?;
let l = wctx.loaded.lock().await;
let repos = l
.nornir
.repo
.iter()
.map(|(name, r)| RepoSummary {
name: name.clone(),
remote: r.remote.clone(),
history: r.history.clone(),
readme: r.readme.clone(),
})
.collect();
Ok(Response::new(RepoList { repos }))
}
async fn register(
&self,
req: Request<RegisterRepoRequest>,
) -> Result<Response<RegisterRepoResponse>, Status> {
let wctx = self.state.ws_write(&req)?.clone();
let req = req.into_inner();
if req.name.is_empty() {
return Err(invalid_arg("`name` must not be empty"));
}
if req.name.contains(['.', '/', '\\', '[', ']']) {
return Err(invalid_arg(
"`name` must be a bare identifier (no '.', '/', '\\\\', '[', ']')",
));
}
let mut l = wctx.loaded.lock().await;
let cfg_path = l.config_path.clone();
let existing = std::fs::read_to_string(&cfg_path).ok().unwrap_or_default();
let mut doc: toml_edit::DocumentMut = existing
.parse()
.map_err(|e| internal(format!("parse {}: {e}", cfg_path.display())))?;
let repo_parent = doc
.entry("repo")
.or_insert_with(|| {
let mut t = toml_edit::Table::new();
t.set_implicit(true);
toml_edit::Item::Table(t)
})
.as_table_mut()
.ok_or_else(|| {
errcode::status(
&errcode::RPC_DESCRIPTOR_SHAPE,
tonic::Code::FailedPrecondition,
"`repo` exists but is not a table",
)
})?;
repo_parent.set_implicit(true);
let entry = repo_parent
.entry(&req.name)
.or_insert_with(|| toml_edit::Item::Table(toml_edit::Table::new()));
let tbl = entry.as_table_mut().ok_or_else(|| {
errcode::status(
&errcode::RPC_DESCRIPTOR_SHAPE,
tonic::Code::FailedPrecondition,
format!("`repo.{}` exists but is not a table", req.name),
)
})?;
if !req.remote.is_empty() {
tbl.insert("remote", toml_edit::value(req.remote.as_str()));
}
if !req.history.is_empty() {
tbl.insert("history", toml_edit::value(req.history.as_str()));
}
if !req.readme.is_empty() {
tbl.insert("readme", toml_edit::value(req.readme.as_str()));
}
if !req.path.is_empty() {
tbl.insert("path", toml_edit::value(req.path.as_str()));
}
if let Some(parent) = cfg_path.parent() {
std::fs::create_dir_all(parent).map_err(internal)?;
}
std::fs::write(&cfg_path, doc.to_string()).map_err(|e| {
if e.kind() == std::io::ErrorKind::PermissionDenied {
errcode::status(
&errcode::RPC_FAILED_PRECONDITION,
tonic::Code::FailedPrecondition,
format!(
"{} is read-only (likely guard-locked); call Guard.Release first",
cfg_path.display(),
),
)
} else {
internal(e)
}
})?;
let repo = config::Repo {
// No explicit checkout-path override on a freshly-registered repo:
// empty → the by-name `<workspace_root>/<name>` resolution (#47).
path: String::new(),
remote: req.remote.clone(),
history: req.history.clone(),
readme: req.readme.clone(),
publish_order: Vec::new(),
gates: config::Gates::default(),
bench: config::BenchSpec::default(),
// A repo registered over gRPC defaults to NOT private (public crates.io
// candidate); a private repo is marked in the release config by hand.
private: false,
// No feature overrides for the unit-test aspect on a gRPC-registered repo.
test_features: String::new(),
};
l.nornir.repo.insert(req.name.clone(), repo);
Ok(Response::new(RegisterRepoResponse {
name: req.name,
config_path: cfg_path.display().to_string(),
repo_count: l.nornir.repo.len() as u32,
}))
}
async fn unregister(
&self,
req: Request<UnregisterRepoRequest>,
) -> Result<Response<UnregisterRepoResponse>, Status> {
let wctx = self.state.ws_write(&req)?.clone();
let req = req.into_inner();
let mut l = wctx.loaded.lock().await;
let cfg_path = l.config_path.clone();
let removed_mem = l.nornir.repo.remove(&req.name).is_some();
let removed_disk = if cfg_path.exists() {
let text = std::fs::read_to_string(&cfg_path).map_err(internal)?;
let mut doc: toml_edit::DocumentMut = text
.parse()
.map_err(|e| internal(format!("parse {}: {e}", cfg_path.display())))?;
let removed = doc
.get_mut("repo")
.and_then(|i| i.as_table_mut())
.and_then(|t| t.remove(&req.name))
.is_some();
if removed {
std::fs::write(&cfg_path, doc.to_string()).map_err(internal)?;
}
removed
} else {
false
};
if !removed_mem && !removed_disk {
return Err(not_found(format!("repo `{}` not found", req.name)));
}
Ok(Response::new(UnregisterRepoResponse {
name: req.name,
repo_count: l.nornir.repo.len() as u32,
}))
}
async fn reload(&self, req: Request<Empty>) -> Result<Response<ReloadResponse>, Status> {
let wctx = self.state.ws_write(&req)?;
let mut l = wctx.loaded.lock().await;
let cfg_path = l.config_path.clone();
if !cfg_path.exists() {
return Err(not_found(format!("{} does not exist", cfg_path.display())));
}
let fresh = config::load_explicit(&cfg_path).map_err(internal)?;
// A reload that swapped in a `[repos.x]` descriptor would silently empty
// the repo set of a running server; refuse it the same way the CLI does.
crate::config_spelling::check_config_file(&fresh.config_path).map_err(internal)?;
*l = fresh;
Ok(Response::new(ReloadResponse {
config_path: cfg_path.display().to_string(),
repo_count: l.nornir.repo.len() as u32,
}))
}
}
// -- Guard -------------------------------------------------------------------
struct GuardSvc {
state: Shared,
}
fn to_pb(paths: Vec<guard::PathStatus>) -> GuardReport {
let changed_count = paths.iter().filter(|p| p.changed).count() as u32;
let paths = paths
.into_iter()
.map(|p| GuardPath {
path: p.path.display().to_string(),
exists: p.exists,
writable: p.writable,
changed: p.changed,
})
.collect();
GuardReport {
paths,
changed_count,
}
}
#[tonic::async_trait]
impl GuardSvcTrait for GuardSvc {
async fn status(&self, req: Request<Empty>) -> Result<Response<GuardReport>, Status> {
let wctx = self.state.ws_read(&req)?;
let l = wctx.loaded.lock().await;
let report = guard::status(&l.workspace_root, &l.nornir.guard.forbidden);
Ok(Response::new(to_pb(report)))
}
async fn apply(&self, req: Request<Empty>) -> Result<Response<GuardReport>, Status> {
let wctx = self.state.ws_write(&req)?;
let l = wctx.loaded.lock().await;
let report =
guard::apply(&l.workspace_root, &l.nornir.guard.forbidden).map_err(internal)?;
Ok(Response::new(to_pb(report)))
}
async fn release(&self, req: Request<Empty>) -> Result<Response<GuardReport>, Status> {
let wctx = self.state.ws_write(&req)?;
let l = wctx.loaded.lock().await;
let report =
guard::release(&l.workspace_root, &l.nornir.guard.forbidden).map_err(internal)?;
Ok(Response::new(to_pb(report)))
}
async fn verify(
&self,
req: Request<Empty>,
) -> Result<Response<pb::GuardVerifyResponse>, Status> {
let root = self
.state
.ws_read(&req)?
.loaded
.lock()
.await
.workspace_root
.clone();
let recorded = guard::read_manifest(&root).map_err(internal)?;
let report = guard::verify(&root, &recorded);
let intact = report.iter().all(|v| v.ok());
let paths_json = serde_json::to_string(&report).map_err(internal)?;
Ok(Response::new(pb::GuardVerifyResponse {
intact,
recorded_at: recorded.recorded_at.to_rfc3339(),
paths_json,
}))
}
}
// -- Release ------------------------------------------------------------------
struct ReleaseSvc {
state: Shared,
}
fn last_run_for(loaded: &Loaded, name: &str) -> anyhow::Result<bench::BenchRun> {
let repo = loaded
.nornir
.repo
.get(name)
.ok_or_else(|| anyhow!("repo `{name}` not configured"))?;
let repo_root = config::Nornir::repo_dir(&loaded.workspace_root, name);
let path = repo_root.join(if repo.history.is_empty() {
"bench_history.jsonl"
} else {
&repo.history
});
let runs = bench::history::read_all(&path)?;
runs.into_iter()
.last()
.ok_or_else(|| anyhow!("no bench runs in {}", path.display()))
}
/// All bench runs for `name` from the JSONL receipt, for the `bench_history` doc
/// section. Best-effort (empty when absent); same source as [`last_run_for`].
fn history_for(loaded: &Loaded, name: &str) -> Vec<bench::BenchRun> {
let Some(repo) = loaded.nornir.repo.get(name) else {
return Vec::new();
};
let repo_root = config::Nornir::repo_dir(&loaded.workspace_root, name);
let path = repo_root.join(if repo.history.is_empty() {
"bench_history.jsonl"
} else {
&repo.history
});
bench::history::read_all(&path).unwrap_or_default()
}
fn repo_root_for(loaded: &Loaded, name: &str) -> Result<PathBuf, Status> {
if !loaded.nornir.repo.contains_key(name) {
return Err(not_found(format!("repo `{name}` not configured")));
}
Ok(config::Nornir::repo_dir(&loaded.workspace_root, name))
}
fn gate_pass(name: &str) -> GateResult {
GateResult {
gate: name.into(),
status: "pass".into(),
message: String::new(),
version: String::new(),
max_drop_pct: 0.0,
}
}
fn gate_fail<E: std::fmt::Display>(name: &str, e: E) -> GateResult {
GateResult {
gate: name.into(),
status: "fail".into(),
message: format!("{e:#}"),
version: String::new(),
max_drop_pct: 0.0,
}
}
/// Run a repo's release gate suite — the ONE release-readiness verdict, shared by
/// `Release.GateAll` (🚦 Release-gates pane) and `Ops.RunRelease` (🚀 Release
/// button).
///
/// It exists because those two surfaces disagreed. `gate_all` walked the repo's
/// `[gates]` config; `run_release` loaded the same config, discarded it with a
/// `let _ = &repo_cfg;`, and ran exactly one hardcoded gate — so the two buttons
/// could return different verdicts for the same repo, and the 🚀 one lit up green
/// having ignored every gate the repo declared. Both now walk
/// [`release::gate::selected_gates`], so there is exactly one answer.
///
/// Returns `(passed, failed)` by gate name. Acquires the loaded-workspace lock
/// itself and releases it before the heavy (`coverage`, `tests`) gates, which run
/// off the async executor.
async fn run_gate_suite(
wctx: &WorkspaceCtx,
name: &str,
) -> Result<(Vec<String>, Vec<GateFailure>), Status> {
fn push(
passed: &mut Vec<String>,
failed: &mut Vec<GateFailure>,
n: &str,
r: anyhow::Result<()>,
) {
match r {
Ok(()) => passed.push(n.into()),
Err(e) => failed.push(GateFailure {
name: n.into(),
error: format!("{e:#}"),
}),
}
}
let mut passed: Vec<String> = Vec::new();
let mut failed: Vec<GateFailure> = Vec::new();
// Everything that reads the loaded workspace happens under ONE lock; the
// heavy gates below need it dropped first.
let (root, selected, workspace) = {
let l = wctx.loaded.lock().await;
let root = repo_root_for(&l, name)?;
let repo = l
.nornir
.repo
.get(name)
.ok_or_else(|| not_found(format!("repo `{name}` not configured")))?
.clone();
let selected = release::gate::selected_gates(&repo.gates);
let last = last_run_for(&l, name);
for gate in &selected {
match *gate {
"no_path_patches" => push(
&mut passed,
&mut failed,
gate,
release::gate::no_path_patches(&root),
),
"nexus_floor" => {
let r = last
.as_ref()
.map_err(|e| anyhow!("{e:#}"))
.and_then(release::gate::nexus_floor);
push(&mut passed, &mut failed, gate, r);
}
"no_regression" => {
let pct = if repo.gates.max_regression_pct > 0.0 {
repo.gates.max_regression_pct
} else {
10.0
};
let hp = root.join(if repo.history.is_empty() {
"bench_history.jsonl"
} else {
&repo.history
});
let r = last
.as_ref()
.map_err(|e| anyhow!("{e:#}"))
.and_then(|r| release::gate::no_regression(r, &hp, pct));
push(&mut passed, &mut failed, gate, r);
}
"integration_roundtrip" => {
let kinds: Vec<&str> = repo
.gates
.integration_roundtrip
.iter()
.map(|s| s.as_str())
.collect();
push(
&mut passed,
&mut failed,
gate,
release::gate::integration_roundtrip_via_cargo_test(&root, &kinds),
);
}
"docs_fresh" => {
let ws_root = l.workspace_root.clone();
let r: anyhow::Result<()> = (|| {
let run = last.as_ref().map_err(|e| anyhow!("{e:#}"))?;
let layout = docs::RepoLayout::new(&root);
let history = history_for(&l, name);
let ctx = docs::Ctx::new(&root, &ws_root, Some(run))
.with_history(&history)
.with_path_dep_cloner(&crate::security::docs_path_dep_cloner);
docs::render_check_all(&layout, &ctx).map(|_| ())
})();
push(&mut passed, &mut failed, gate, r);
}
// Handled below, off the lock + off the async executor.
"coverage" | "tests" => {}
other => failed.push(GateFailure {
name: (*other).into(),
error: format!("gate `{other}` is selected but has no server-side runner"),
}),
}
}
(root, selected, l.workspace_name())
};
// AUT5: the autonom completeness gate. LIGHT (reads the warehouse
// surface_coverage map + the repo's allowlist; no suite run) but it does sync
// warehouse I/O → off the async executor. The server has no live clap tree →
// canonical CLI fallback.
if selected.contains(&"coverage") {
let (room, wh) = (wctx.code_room.clone(), wctx.warehouse.clone());
let root_c = root.clone();
let repo_name = name.to_string();
let workspace = workspace.clone();
let r = tokio::task::spawn_blocking(move || -> anyhow::Result<()> {
let run_id = nornir::warehouse::test_results::new_run_id();
// the surface is the code room's; the verdict goes to the warehouse
let state = nornir::autonom::gate_coverage_for_repo_in(
&nornir::code_room::Reach::new(&room, &wh),
&workspace,
&repo_name,
&root_c,
Vec::new(),
&run_id,
)?;
release::gate::coverage_gate(&state.report)
})
.await
.map_err(internal)?;
push(&mut passed, &mut failed, "coverage", r);
}
// Plan #5: the heavy test-matrix gate. It RUNS the matrix over the served
// checkout, folds the fresh rows with the repo's persisted history, and fails
// on red / regression. Release/on-request only, never on workspace load.
if selected.contains(&"tests") {
let wh = wctx.warehouse.clone();
let root_c = root.clone();
let repo_name = name.to_string();
let r = tokio::task::spawn_blocking(move || -> anyhow::Result<()> {
use nornir::warehouse::test_results::{TestSelector, query_test_results};
let history = wh
.block_on(query_test_results(
&wh,
&TestSelector::Repo(repo_name.clone()),
))
.unwrap_or_default();
let aspects = nornir::test_matrix::Aspect::DEFAULT.to_vec();
let fresh = release::gate::run_tests_gate(&root_c, &history, &aspects)?;
// Persist the freshly-run rows so the 🧪 Test pane + history see them.
let _ = nornir::warehouse::test_results::append_test_results_blocking(&wh, &fresh);
Ok(())
})
.await
.map_err(internal)?;
push(&mut passed, &mut failed, "tests", r);
}
Ok((passed, failed))
}
#[tonic::async_trait]
impl ReleaseSvcTrait for ReleaseSvc {
async fn gate_path_patches(
&self,
req: Request<RepoOnly>,
) -> Result<Response<GateResult>, Status> {
let wctx = self.state.ws_read(&req)?.clone();
let name = req.into_inner().repo;
let l = wctx.loaded.lock().await;
let root = repo_root_for(&l, &name)?;
Ok(Response::new(match release::gate::no_path_patches(&root) {
Ok(()) => gate_pass("no_path_patches"),
Err(e) => gate_fail("no_path_patches", e),
}))
}
async fn gate_nexus_floor(
&self,
req: Request<RepoOnly>,
) -> Result<Response<GateResult>, Status> {
let wctx = self.state.ws_read(&req)?.clone();
let name = req.into_inner().repo;
let l = wctx.loaded.lock().await;
let _ = repo_root_for(&l, &name)?;
let run = match last_run_for(&l, &name) {
Ok(r) => r,
Err(e) => return Ok(Response::new(gate_fail("nexus_floor", e))),
};
let mut res = match release::gate::nexus_floor(&run) {
Ok(()) => gate_pass("nexus_floor"),
Err(e) => gate_fail("nexus_floor", e),
};
res.version = run.version;
Ok(Response::new(res))
}
async fn gate_no_regression(
&self,
req: Request<RepoOnly>,
) -> Result<Response<GateResult>, Status> {
let wctx = self.state.ws_read(&req)?.clone();
let name = req.into_inner().repo;
let l = wctx.loaded.lock().await;
let root = repo_root_for(&l, &name)?;
let repo = l.nornir.repo.get(&name).unwrap();
let pct = if repo.gates.max_regression_pct > 0.0 {
repo.gates.max_regression_pct
} else {
10.0
};
let hp = root.join(if repo.history.is_empty() {
"bench_history.jsonl"
} else {
&repo.history
});
let run = match last_run_for(&l, &name) {
Ok(r) => r,
Err(e) => return Ok(Response::new(gate_fail("no_regression", e))),
};
let mut res = match release::gate::no_regression(&run, &hp, pct) {
Ok(()) => gate_pass("no_regression"),
Err(e) => gate_fail("no_regression", e),
};
res.version = run.version;
res.max_drop_pct = pct;
Ok(Response::new(res))
}
async fn gate_docs_fresh(
&self,
req: Request<RepoOnly>,
) -> Result<Response<GateResult>, Status> {
let wctx = self.state.ws_read(&req)?.clone();
let name = req.into_inner().repo;
let l = wctx.loaded.lock().await;
let root = repo_root_for(&l, &name)?;
let run = match last_run_for(&l, &name) {
Ok(r) => r,
Err(e) => return Ok(Response::new(gate_fail("docs_fresh", e))),
};
let layout = docs::RepoLayout::new(&root);
let history = history_for(&l, &name);
let ctx = docs::Ctx::new(&root, &l.workspace_root, Some(&run))
.with_history(&history)
.with_path_dep_cloner(&crate::security::docs_path_dep_cloner);
Ok(Response::new(match docs::render_check_all(&layout, &ctx) {
Ok(_) => gate_pass("docs_fresh"),
Err(e) => gate_fail("docs_fresh", e),
}))
}
async fn gate_all(&self, req: Request<RepoOnly>) -> Result<Response<GateAllResult>, Status> {
let wctx = self.state.ws_read(&req)?.clone();
let actor = rpc_actor(&req);
let name = req.into_inner().repo;
// Job ledger: the release-readiness gate suite is real work (it can run
// tests / coverage / integration-roundtrip subprocesses), so it earns a
// `release_run` job like the other release ops. Best-effort.
let job = stamp_actor(
wctx.jobs.start(
nornir::jobs::kind::RELEASE_RUN,
&name,
&wctx.name,
serde_json::json!({ "op": "gate_all" }),
),
actor,
);
let (passed, failed) = run_gate_suite(&wctx, &name).await?;
job.finish(
serde_json::json!({ "passed": passed.len(), "failed": failed.len() }),
"",
);
Ok(Response::new(GateAllResult {
repo: name,
passed,
failed,
}))
}
async fn order(&self, req: Request<OrderRequest>) -> Result<Response<OrderResponse>, Status> {
let wctx = self.state.ws_read(&req)?.clone();
let mut ws = req.into_inner().workspace;
if ws.is_empty() {
ws = "workspace_holger".into();
}
let (selected, wh_clone) = {
let l = wctx.loaded.lock().await;
(
l.nornir.repo.keys().cloned().collect::<Vec<_>>(),
wctx.warehouse.clone(),
)
};
let ws_for_q = ws.clone();
let snapshots = tokio::task::spawn_blocking(move || {
wh_clone.block_on(query_dep_graph_snapshots(&wh_clone, &ws_for_q, None))
})
.await
.map_err(internal)?
.map_err(internal)?;
let latest = snapshots.into_iter().last();
match latest {
Some(snap) => {
let order = topo_order_from_edges(&selected, &snap.edges);
Ok(Response::new(OrderResponse {
order,
source: "iceberg".into(),
snapshot_id: snap.snapshot_id.to_string(),
}))
}
None => Ok(Response::new(OrderResponse {
order: selected,
source: "fallback".into(),
snapshot_id: String::new(),
})),
}
}
async fn trace(
&self,
req: Request<pb::TraceQuery>,
) -> Result<Response<pb::JsonResponse>, Status> {
let wctx = self.state.ws_read(&req)?.clone();
let q = req.into_inner();
// Best-effort graph for Mímir suspect ranking (trace works without it).
let graph = wctx.graph().await.ok();
let trace = nornir::release::regression::trace_gate_async(
wctx.warehouse.as_ref(),
&q.workspace,
&q.repo,
graph.as_ref().map(|g| &g.graph),
)
.await
.map_err(internal)?;
Ok(Response::new(pb::JsonResponse {
json: serde_json::to_string(&trace).map_err(internal)?,
}))
}
type ProgressStream = std::pin::Pin<
Box<dyn tokio_stream::Stream<Item = Result<ProgressEvent, Status>> + Send + 'static>,
>;
async fn progress(
&self,
req: Request<Empty>,
) -> Result<Response<Self::ProgressStream>, Status> {
// Logs live next to the warehouse (`<workspace>/.nornir/logs`) — derive
// from the configured warehouse root so the stream follows the SELECTED
// workspace (the `nornir-workspace` header), not a hardcoded one. With no
// workspace selected this errors like every other handler.
let log_dir = {
let ctx = self.state.ws_read(&req)?;
let loaded = ctx.loaded.lock().await;
let wh = loaded.warehouse_root();
wh.parent()
.map(|p| p.join("logs"))
.unwrap_or_else(|| wh.join("logs"))
};
let (tx, rx) = tokio::sync::mpsc::channel::<Result<ProgressEvent, Status>>(64);
tokio::spawn(async move {
if let Err(e) = tail_progress(log_dir, tx.clone()).await {
let _ = tx.send(Err(internal(e))).await;
}
});
let stream = tokio_stream::wrappers::ReceiverStream::new(rx);
Ok(Response::new(Box::pin(stream) as Self::ProgressStream))
}
}
/// Find the newest `release-run-*.events.ndjson` file. Polls until one exists
/// or the receiver drops.
async fn latest_log(log_dir: &std::path::Path) -> Option<PathBuf> {
let mut best: Option<(std::time::SystemTime, PathBuf)> = None;
let mut rd = tokio::fs::read_dir(log_dir).await.ok()?;
while let Ok(Some(entry)) = rd.next_entry().await {
let name = entry.file_name();
let n = name.to_string_lossy();
if !(n.starts_with("release-run-") && n.ends_with(".events.ndjson")) {
continue;
}
if let Ok(meta) = entry.metadata().await {
if let Ok(mt) = meta.modified() {
match &best {
Some((t, _)) if *t >= mt => {}
_ => best = Some((mt, entry.path())),
}
}
}
}
best.map(|(_, p)| p)
}
/// Tail the latest release-run ndjson file and forward each line as a
/// ProgressEvent. Closes the stream after RunEnd.
async fn tail_progress(
log_dir: PathBuf,
tx: tokio::sync::mpsc::Sender<Result<ProgressEvent, Status>>,
) -> anyhow::Result<()> {
use tokio::io::{AsyncBufReadExt, AsyncSeekExt};
let path = loop {
if let Some(p) = latest_log(&log_dir).await {
break p;
}
if tx.is_closed() {
return Ok(());
}
tokio::time::sleep(std::time::Duration::from_millis(500)).await;
};
let mut f = tokio::fs::File::open(&path).await?;
f.seek(std::io::SeekFrom::Start(0)).await?;
let mut reader = tokio::io::BufReader::new(f);
let mut buf = String::new();
loop {
buf.clear();
let n = reader.read_line(&mut buf).await?;
if n == 0 {
if tx.is_closed() {
return Ok(());
}
tokio::time::sleep(std::time::Duration::from_millis(250)).await;
continue;
}
let line = buf.trim();
if line.is_empty() {
continue;
}
match serde_json::from_str::<release::progress::ReleaseEvent>(line) {
Ok(ev) => {
let is_end = matches!(ev, release::progress::ReleaseEvent::RunEnd { .. });
let pb_ev = release_event_to_pb(ev);
if tx.send(Ok(pb_ev)).await.is_err() {
return Ok(());
}
if is_end {
return Ok(());
}
}
Err(_) => continue,
}
}
}
fn release_event_to_pb(ev: release::progress::ReleaseEvent) -> ProgressEvent {
use pb::progress_event::Kind as K;
use release::progress::ReleaseEvent as R;
let kind = match ev {
R::RunStart {
run_id, workspace, ..
} => K::RunStart(pb::RunStart { run_id, workspace }),
R::RepoStart { repo, sha, .. } => K::RepoStart(pb::RepoStart { repo, sha }),
R::PhaseStart { repo, phase, .. } => K::PhaseStart(pb::PhaseStart { repo, phase }),
R::PhaseEnd {
repo,
phase,
ok,
duration_ms,
..
} => K::PhaseEnd(pb::PhaseEnd {
repo,
phase,
ok,
duration_ms,
}),
R::BinaryStart { repo, binary, .. } => K::BinaryStart(pb::BinaryStart { repo, binary }),
R::TestPass {
repo, binary, name, ..
} => K::TestPass(pb::TestPass { repo, binary, name }),
R::TestFail {
repo, binary, name, ..
} => K::TestFail(pb::TestFail { repo, binary, name }),
R::BinaryDone {
repo,
binary,
passed,
failed,
..
} => K::BinaryDone(pb::BinaryDone {
repo,
binary,
passed,
failed,
}),
R::RepoEnd { repo, ok, .. } => K::RepoEnd(pb::RepoEnd { repo, ok }),
R::RunEnd { run_id, ok, .. } => K::RunEnd(pb::RunEnd { run_id, ok }),
};
ProgressEvent { kind: Some(kind) }
}
// -- Bench --------------------------------------------------------------------
struct BenchSvc {
state: Shared,
}
fn pb_bench_run_from(run: &bench::BenchRun) -> pb::BenchRun {
let results = run
.results
.iter()
.map(|r| pb::BenchResult {
name: r.name.clone(),
metrics: r
.metrics
.iter()
.filter_map(|(k, v)| {
v.as_f64().map(|f| Kvf {
key: k.clone(),
value: f,
})
})
.collect(),
})
.collect();
let tests = run
.tests
.iter()
.map(|t| pb::TestOutcome {
name: t.name.clone(),
passed: t.passed,
duration_ms: t.duration_ms.unwrap_or(0.0),
has_duration: t.duration_ms.is_some(),
message: t.message.clone().unwrap_or_default(),
})
.collect();
pb::BenchRun {
// The producer's identity travels with the run — that is what makes
// `Bench.Submit` idempotent. A legacy run (nil id) sends the EMPTY string,
// exactly as `nornir.proto:170` specifies, so the store can tell "cannot be
// deduplicated" from a real id rather than inventing one here.
id: if run.has_id() {
run.id.to_string()
} else {
String::new()
},
// What the BOX is, as JSON. Empty when the producer did not look; never an
// empty `Machine`, which would claim facts nobody measured.
machine_facts_json: run
.machine_facts
.as_ref()
.and_then(|m| serde_json::to_string(m).ok())
.unwrap_or_default(),
date: run.date.clone(),
timestamp: run.timestamp.clone().unwrap_or_default(),
version: run.version.clone(),
machine: run.machine.clone(),
cores: run.cores,
results,
tests,
}
}
fn bench_run_from_pb(p: pb::BenchRun) -> bench::BenchRun {
let results = p
.results
.into_iter()
.map(|r| {
let mut map = serde_json::Map::new();
for kv in r.metrics {
if let Some(num) = serde_json::Number::from_f64(kv.value) {
map.insert(kv.key, serde_json::Value::Number(num));
}
}
bench::BenchResult {
name: r.name,
metrics: map,
..Default::default()
}
})
.collect();
let tests = p
.tests
.into_iter()
.map(|t| bench::TestOutcome {
name: t.name,
passed: t.passed,
duration_ms: t.has_duration.then_some(t.duration_ms),
message: if t.message.is_empty() {
None
} else {
Some(t.message)
},
})
.collect();
bench::BenchRun {
// An unparseable or absent id becomes NIL — the documented "legacy, no
// identity" value — rather than a fresh uuid, because minting one here would
// hand the row an identity the producer never had and defeat the dedup it
// exists for.
id: uuid::Uuid::parse_str(&p.id).unwrap_or_else(|_| uuid::Uuid::nil()),
machine_facts: if p.machine_facts_json.is_empty() {
None
} else {
serde_json::from_str(&p.machine_facts_json).ok()
},
date: p.date,
timestamp: if p.timestamp.is_empty() {
None
} else {
Some(p.timestamp)
},
version: p.version,
machine: p.machine,
cores: p.cores,
results,
tests,
}
}
#[tonic::async_trait]
impl BenchSvcTrait for BenchSvc {
async fn history(
&self,
req: Request<RepoOnly>,
) -> Result<Response<BenchHistoryResponse>, Status> {
let wctx = self.state.ws_read(&req)?.clone();
let name = req.into_inner().repo;
{
// Validate the repo is a member, then drop the config lock before the
// warehouse read.
let l = wctx.loaded.lock().await;
if !l.nornir.repo.contains_key(&name) {
return Err(not_found(format!("repo `{name}` not configured")));
}
}
// Source of truth is the iceberg `bench_runs` table (where both
// `Bench.Submit` and `Ops.RunBench` persist) — NOT the legacy on-disk
// `bench_history.jsonl`, which `Ops.RunBench` never wrote (so jsonl reads
// returned []). The JSONL is now a pure receipt; reads go to the warehouse.
let runs = wctx
.warehouse
.query_bench_runs_async(&nornir::warehouse::BenchFilter::for_repo(name.clone()))
.await
.map_err(internal)?;
Ok(Response::new(BenchHistoryResponse {
repo: name,
runs: runs.iter().map(pb_bench_run_from).collect(),
}))
}
async fn submit(
&self,
req: Request<SubmitBenchRequest>,
) -> Result<Response<SubmitBenchResponse>, Status> {
let wctx = self.state.ws_write(&req)?.clone();
let req = req.into_inner();
let pb_run = req.run.ok_or_else(|| invalid_arg("`run` is required"))?;
let run = bench_run_from_pb(pb_run);
if run.machine.is_empty() {
return Err(invalid_arg("BenchRun.machine must not be empty"));
}
let l = wctx.loaded.lock().await;
let repo = l
.nornir
.repo
.get(&req.repo)
.ok_or_else(|| not_found(format!("repo `{}` not configured", req.repo)))?;
let repo_root = config::Nornir::repo_dir(&l.workspace_root, &req.repo);
let path = repo_root.join(if repo.history.is_empty() {
"bench_history.jsonl"
} else {
&repo.history
});
drop(l); // release the config lock before the warehouse write
// Source of truth: persist into the iceberg `bench_runs` table (so
// query_bench_runs / bench_history / gates see it), exactly like the
// local `bench run` path. The JSONL is just the on-disk receipt.
let run_id = wctx
.warehouse
.append_bench_run_async(&req.repo, &run)
.await
.map_err(internal)?;
if let Some(parent) = path.parent() {
std::fs::create_dir_all(parent).map_err(internal)?;
}
bench::history::append(&path, &run).map_err(internal)?;
Ok(Response::new(SubmitBenchResponse {
run_id: run_id.to_string(),
}))
}
}
// -- Telemetry ----------------------------------------------------------------
struct TelemetrySvc {
state: Shared,
}
#[tonic::async_trait]
impl TelemetrySvcTrait for TelemetrySvc {
/// Append a thin client's MCP-call telemetry into the served workspace's
/// `mcp_requests` table. The server owns the warehouse lock, so this is how
/// a client-mode `nornir-mcp` (telemetry-off locally) gets its usage onto
/// the server for the viz 📞 tab.
async fn submit_mcp_calls(
&self,
req: Request<pb::SubmitMcpCallsRequest>,
) -> Result<Response<pb::SubmitMcpCallsResponse>, Status> {
let wctx = self.state.ws_write(&req)?.clone();
let calls: Vec<nornir::warehouse::iceberg::McpCall> = req
.into_inner()
.calls
.into_iter()
.map(|c| nornir::warehouse::iceberg::McpCall {
ts_micros: c.ts_micros,
tool: c.tool,
status: c.status,
latency_ms: c.latency_ms,
})
.collect();
let n = calls.len() as u32;
if !calls.is_empty() {
let wh = wctx.warehouse.clone();
// append_mcp_calls is sync (redb); keep the async reactor free.
tokio::task::spawn_blocking(move || wh.append_mcp_calls(&calls))
.await
.map_err(internal)?
.map_err(internal)?;
}
Ok(Response::new(pb::SubmitMcpCallsResponse { accepted: n }))
}
/// Append client-run bake-off rows into the served workspace's
/// `agent_model_runs` → fills the viz 🏆 Leaderboard on the live server.
async fn submit_bakeoff(
&self,
req: Request<pb::SubmitBakeoffRequest>,
) -> Result<Response<pb::SubmitCountResponse>, Status> {
use nornir::warehouse::agent_model_runs::{AgentModelRunRow, append_agent_model_runs};
let wctx = self.state.ws_write(&req)?.clone();
let rows: Vec<AgentModelRunRow> = req
.into_inner()
.runs
.into_iter()
.map(|r| AgentModelRunRow {
run_id: r.run_id,
ts_micros: r.ts_micros,
agent: r.agent,
model: r.model,
prompt_id: r.prompt_id,
prompt: r.prompt,
output: r.output,
latency_ms: r.latency_ms,
tokens_in: r.tokens_in,
tokens_out: r.tokens_out,
tokens_per_s: r.tokens_per_s,
score: r.score,
ok: r.ok,
error: (!r.error.is_empty()).then_some(r.error),
cost_usd: r.cost_usd,
mcp_tool_calls: r.mcp_tool_calls,
})
.collect();
let n = rows.len() as u32;
if !rows.is_empty() {
append_agent_model_runs(&wctx.warehouse, &rows)
.await
.map_err(internal)?;
}
Ok(Response::new(pb::SubmitCountResponse { accepted: n }))
}
/// Append client-run test rows into the served workspace's `test_results`
/// → fills the viz 🧪 Test pane on the live server.
async fn submit_test_results(
&self,
req: Request<pb::SubmitTestResultsRequest>,
) -> Result<Response<pb::SubmitCountResponse>, Status> {
use nornir::warehouse::test_results::{TestResultRow, append_test_results};
let wctx = self.state.ws_write(&req)?.clone();
let rows: Vec<TestResultRow> = req
.into_inner()
.rows
.into_iter()
.map(|r| TestResultRow {
run_id: r.run_id,
repo: r.repo,
suite: r.suite,
test_name: r.test_name,
status: r.status,
duration_ms: r.duration_ms,
ts_micros: r.ts_micros,
message: r.message,
aspect: r.aspect,
metric: r.metric,
})
.collect();
let n = rows.len() as u32;
if !rows.is_empty() {
append_test_results(&wctx.warehouse, &rows)
.await
.map_err(internal)?;
}
Ok(Response::new(pb::SubmitCountResponse { accepted: n }))
}
/// Append client-run release-op DAG rows into the served workspace's
/// `release_events` → fills the viz 🚀 Release tab on the live server. The
/// fat-CLI `nornir release run` executes on the dev's hardware and mints the
/// events there; this is how they reach the locked server warehouse.
async fn submit_release_events(
&self,
req: Request<pb::SubmitReleaseEventsRequest>,
) -> Result<Response<pb::SubmitCountResponse>, Status> {
use nornir::warehouse::release_events::{ReleaseEventRow, append_release_events};
let wctx = self.state.ws_write(&req)?.clone();
let rows: Vec<ReleaseEventRow> = req
.into_inner()
.rows
.into_iter()
.map(|r| ReleaseEventRow {
run_id: r.run_id,
seq: r.seq,
ts_micros: r.ts_micros,
component: r.component,
repo: r.repo,
op: r.op,
phase: r.phase,
status: r.status,
detail: r.detail,
// `has_depends_on` distinguishes None from an empty list.
depends_on: r.has_depends_on.then_some(r.depends_on),
elapsed_ms: r.has_elapsed_ms.then_some(r.elapsed_ms),
})
.collect();
let n = rows.len() as u32;
if !rows.is_empty() {
append_release_events(wctx.warehouse.as_ref(), &rows)
.await
.map_err(internal)?;
}
Ok(Response::new(pb::SubmitCountResponse { accepted: n }))
}
/// `Telemetry.SubmitJobs` — upsert thin-client job-ledger records into the
/// served workspace's `jobs.redb` (the server is its sole writer). `jobs_json`
/// is a JSON array of `jobs::JobRecord`; mirrors `submit_release_events`.
async fn submit_jobs(
&self,
req: Request<pb::SubmitJobsRequest>,
) -> Result<Response<pb::SubmitCountResponse>, Status> {
let wctx = self.state.ws_write(&req)?.clone();
let jobs_json = req.into_inner().jobs_json;
let store = wctx.jobs.clone();
let state = self.state.clone();
let n = tokio::task::spawn_blocking(move || -> anyhow::Result<u32> {
let recs: Vec<nornir::jobs::JobRecord> = if jobs_json.trim().is_empty() {
Vec::new()
} else {
serde_json::from_str(&jobs_json)?
};
for rec in &recs {
// A QUEUED backfill request is a job the SERVER runs: it goes
// onto the workspace's background lane, and the row the client
// sent becomes its receipt (what was accepted, what was refused).
match backfill_request_of(rec) {
Some((from, member)) => store.submit(&backfill_receipt(&state, &wctx.name, rec, &from, member.as_deref()))?,
None => store.submit(rec)?,
}
}
Ok(recs.len() as u32)
})
.await
.map_err(internal)?
.map_err(internal)?;
Ok(Response::new(pb::SubmitCountResponse { accepted: n }))
}
/// Append client-run `BuildProvenance` records into the served workspace's
/// `build_provenance` ledger → fills the viz warehouse deck on the live
/// server. Each record ships as JSON (like `submit_jobs`) and is reconstructed
/// + appended via the SAME `append_build_provenance` the embedded path calls.
async fn submit_build_provenance(
&self,
req: Request<pb::SubmitBuildProvenanceRequest>,
) -> Result<Response<pb::SubmitCountResponse>, Status> {
use nornir::warehouse::build_provenance::append_build_provenance;
use nornir_build_thing::BuildProvenance;
let wctx = self.state.ws_write(&req)?.clone();
let records_json = req.into_inner().records_json;
let mut n = 0u32;
for json in &records_json {
let prov = BuildProvenance::from_json(json).map_err(|e| {
errcode::status(
&errcode::RPC_INVALID_ARGUMENT,
tonic::Code::InvalidArgument,
format!("bad BuildProvenance json: {e}"),
)
})?;
append_build_provenance(&wctx.warehouse, &prov)
.await
.map_err(internal)?;
// BUILDER 2: a successful run lands in Mímir (its build_info, with
// the lineage rows, re-queues it when it arrives after this)
if prov.outcome == nornir_build_thing::provenance::Outcome::Success {
queue_build_thing_mimir(&wctx, &prov.id);
}
n += 1;
}
Ok(Response::new(pb::SubmitCountResponse { accepted: n }))
}
/// Append client-run §f3 build-info rows (manifest + feed) into the served
/// workspace's `build_info` table → the native landing for a dwarves build.
/// Maps `BuildInfoRowPb` 1:1 back to `BuildInfoRow` (`has_blob` distinguishes
/// a NULL blob from an empty one) and calls the same `append_build_info`.
async fn submit_build_info(
&self,
req: Request<pb::SubmitBuildInfoRequest>,
) -> Result<Response<pb::SubmitCountResponse>, Status> {
use nornir::warehouse::build_info::append_build_info;
use nornir_build_thing::build_info::BuildInfoRow;
let wctx = self.state.ws_write(&req)?.clone();
let rows: Vec<BuildInfoRow> = req
.into_inner()
.rows
.into_iter()
.map(|r| BuildInfoRow {
build_id: r.build_id,
step_id: r.step_id,
ts_micros: r.ts_micros,
kind: r.kind,
level: r.level,
name: r.name,
value: r.value,
content_id: r.content_id,
// `has_blob` distinguishes None (SQL NULL) from an empty payload.
blob: r.has_blob.then_some(r.blob),
})
.collect();
let n = rows.len() as u32;
if !rows.is_empty() {
append_build_info(&wctx.warehouse, &rows)
.await
.map_err(internal)?;
}
// BUILDER 2: a successful Build Thing run's manifest just landed — its
// products go into the workspace's Mímir on the `rooms` lane (jera).
for build_id in nornir::code_room::products::landed_builds(&rows) {
queue_build_thing_mimir(&wctx, &build_id);
}
Ok(Response::new(pb::SubmitCountResponse { accepted: n }))
}
/// File what a forge's Runa step measured (tests / bench / coverage) into the
/// served workspace's warehouse, keyed by (repo, commit) — the facts the
/// `weak_tests`, `bench_gaps` and `change_risk` lenses read. The repo must be
/// a member of the workspace (its rows join the room by that name).
async fn submit_step_feed(
&self,
req: Request<pb::SubmitStepFeedRequest>,
) -> Result<Response<pb::SubmitCountResponse>, Status> {
use nornir::warehouse::iceberg::StepTestOutcome;
use nornir::warehouse::step_feed::{StepFeed, file};
let wctx = self.state.ws_write(&req)?.clone();
let r = req.into_inner();
{
let l = wctx.loaded.lock().await;
if !l.nornir.repo.contains_key(&r.repo) {
return Err(not_found(format!(
"repo `{}` is not a member of this workspace: its {} would join nothing in the code room",
r.repo, r.kind
)));
}
}
let tests = r
.tests
.into_iter()
.map(|t| StepTestOutcome {
name: t.name,
krate: t.krate,
status: t.status,
duration_ms: t.has_duration.then_some(t.duration_ms),
message: (!t.message.is_empty()).then_some(t.message),
})
.collect();
let feed = if matches!(r.kind.as_str(), "trace" | "profile") {
StepFeed::decode_run(r.repo, r.commit, r.run_id, r.ts_micros, &r.kind, &r.trace_json, &r.trace_events, r.source)
} else {
StepFeed::decode(r.repo, r.commit, r.run_id, r.ts_micros, &r.kind, tests, &r.bench_json, &r.coverage_json, &r.selection_json)
}
.map_err(|e| invalid_arg(format!("{e:#}")))?;
// a trace's / profile's functions → the room's (stable ids), on a blocking thread
let fns = feed.filed_fns();
if !fns.is_empty() {
let ids = room_stable_ids(wctx.clone(), fns).await;
let n = nornir::warehouse::step_feed::file_mapped(&wctx.warehouse, &feed, &nornir::warehouse::step_feed::GivenIds(ids))
.await
.map_err(internal)?;
// a filed run does not move the room's version: the replay's kept answers are of the runs before it
#[cfg(feature = "code-room")]
if let Ok(h) = wctx.reach().code_room() {
h.forget_trace_answers();
}
return Ok(Response::new(pb::SubmitCountResponse { accepted: n as u32 }));
}
let n = file(&wctx.warehouse, &feed).await.map_err(internal)?;
Ok(Response::new(pb::SubmitCountResponse { accepted: n as u32 }))
}
}
/// Which room function each traced / profiled function is (its stable id, `""`
/// for none): the workspace's room, read on a blocking thread. No room (a
/// server without the engine, nothing consumed yet) maps nothing — the rows
/// then carry NULL and the replay counts them unplaced.
async fn room_stable_ids(wctx: Arc<WorkspaceCtx>, fns: Vec<nornir_trace::wire::FiledFn>) -> Vec<String> {
#[cfg(feature = "code-room")]
{
use nornir::warehouse::step_feed::RoomMap;
let n = fns.len();
tokio::task::spawn_blocking(move || match nornir::code_room::reader::wired::open(&wctx.reach(), "") {
Ok(room) => nornir::code_room::reader::wired::trace::RoomTraceMap { room }.stable_ids(&fns),
Err(_) => vec![String::new(); fns.len()],
})
.await
.unwrap_or_else(|_| vec![String::new(); n])
}
#[cfg(not(feature = "code-room"))]
{
let _ = wctx;
vec![String::new(); fns.len()]
}
}
// -- Search -------------------------------------------------------------------
struct SearchSvc {
state: Shared,
}
#[tonic::async_trait]
impl SearchSvcTrait for SearchSvc {
async fn query(&self, req: Request<SearchRequest>) -> Result<Response<SearchResponse>, Status> {
let wctx = self.state.ws_read(&req)?.clone();
let req = req.into_inner();
let corpus = if req.corpus.is_empty() {
None
} else {
Some(
index::Corpus::parse(&req.corpus)
.ok_or_else(|| invalid_arg(format!("unknown corpus `{}`", req.corpus)))?,
)
};
let repo_opt = if req.repo.is_empty() {
None
} else {
Some(req.repo.clone())
};
let limit = if req.limit == 0 {
20
} else {
req.limit as usize
};
let idx = wctx.index.clone();
let q = req.query.clone();
let hits =
tokio::task::spawn_blocking(move || idx.search(&q, corpus, repo_opt.as_deref(), limit))
.await
.map_err(internal)?
.map_err(internal)?;
let hits = hits
.into_iter()
.map(|h| SearchHit {
id: String::new(),
corpus: h.corpus,
repo: h.repo,
path: h.path,
score: h.score,
snippet: h.snippet,
title: h.title,
})
.collect();
Ok(Response::new(SearchResponse { hits }))
}
}
// -- Index --------------------------------------------------------------------
struct IndexSvc {
state: Shared,
}
#[tonic::async_trait]
impl IndexSvcTrait for IndexSvc {
async fn stats(&self, req: Request<Empty>) -> Result<Response<IndexStatsResponse>, Status> {
let idx = self.state.ws_read(&req)?.index.clone();
let stats = tokio::task::spawn_blocking(move || idx.stats())
.await
.map_err(internal)?
.map_err(internal)?;
Ok(Response::new(IndexStatsResponse {
total: stats.total,
by_corpus: stats
.by_corpus
.into_iter()
.map(|(k, v)| Kv {
key: k,
value: v.to_string(),
})
.collect(),
}))
}
async fn snapshot(
&self,
req: Request<SnapshotRequest>,
) -> Result<Response<SnapshotResponse>, Status> {
let wctx = self.state.ws_write(&req)?.clone();
let req = req.into_inner();
if req.git_sha.is_empty() {
return Err(invalid_arg("`git_sha` is required"));
}
let workspace = if req.workspace.is_empty() {
"workspace_holger".into()
} else {
req.workspace
};
let repo = if req.repo.is_empty() {
"_workspace".into()
} else {
req.repo
};
let branch = if req.branch.is_empty() {
"unknown".into()
} else {
req.branch
};
let index_dir = wctx.index.index_dir().to_path_buf();
let wh = wctx.warehouse.clone();
// Job ledger: one `snapshot` job for this index→Iceberg landing. Best-effort
// — a missing ledger handle never gates the snapshot.
let job = wctx.jobs.start(
nornir::jobs::kind::SNAPSHOT,
&repo,
&workspace,
serde_json::json!({ "git_sha": req.git_sha, "branch": branch, "source": "index_snapshot" }),
);
let snap = match tokio::task::spawn_blocking(move || {
nornir::index::snapshot::snapshot_to_iceberg(
&wh,
&workspace,
&repo,
&req.git_sha,
&branch,
&index_dir,
)
})
.await
.map_err(internal)?
{
Ok(s) => s,
Err(e) => {
let e = anyhow::anyhow!("{e:#}");
job.fail(&e);
return Err(internal(e));
}
};
job.finish(
serde_json::json!({ "blob_count": snap.blob_count, "total_bytes": snap.total_bytes }),
&format!("index_snapshots:{}", snap.snapshot_id),
);
Ok(Response::new(snapshot_ref_to_pb(snap)))
}
async fn upload_snapshot(
&self,
req: Request<tonic::Streaming<IndexBlobFrame>>,
) -> Result<Response<SnapshotResponse>, Status> {
use pb::index_blob_frame::Body;
let wctx = self.state.ws_write(&req)?.clone();
let mut stream = req.into_inner();
let tmp_root = std::env::temp_dir().join(format!("nornir-upload-{}", uuid::Uuid::new_v4()));
std::fs::create_dir_all(&tmp_root).map_err(internal)?;
// RAII cleanup guard: best-effort rm on drop / function return.
struct TmpDir(PathBuf);
impl Drop for TmpDir {
fn drop(&mut self) {
let _ = std::fs::remove_dir_all(&self.0);
}
}
let _guard = TmpDir(tmp_root.clone());
let root = tmp_root;
let mut meta: Option<pb::IndexBlobMeta> = None;
let mut cur_file: Option<(PathBuf, u64, std::fs::File, u64)> = None; // (path, declared, handle, written)
while let Some(frame) = stream.message().await? {
match frame.body {
Some(Body::Meta(m)) => {
if meta.is_some() {
return Err(invalid_arg("meta frame seen twice"));
}
meta = Some(m);
}
Some(Body::File(f)) => {
if meta.is_none() {
return Err(invalid_arg("first frame must carry `meta`"));
}
if let Some((p, declared, _, written)) = cur_file.take() {
if written != declared {
return Err(invalid_arg(format!(
"file {} declared {} bytes, got {}",
p.display(),
declared,
written
)));
}
}
if f.rel_path.is_empty() || f.rel_path.contains("..") {
return Err(invalid_arg(
"rel_path must be non-empty and not contain `..`",
));
}
let path = root.join(&f.rel_path);
if let Some(parent) = path.parent() {
std::fs::create_dir_all(parent).map_err(internal)?;
}
let fh = std::fs::File::create(&path).map_err(internal)?;
cur_file = Some((path, f.size_bytes, fh, 0));
}
Some(Body::Chunk(bytes)) => {
let Some((_, _, fh, written)) = cur_file.as_mut() else {
return Err(invalid_arg("chunk before file"));
};
use std::io::Write;
fh.write_all(&bytes).map_err(internal)?;
*written += bytes.len() as u64;
}
None => return Err(invalid_arg("empty frame")),
}
}
if let Some((p, declared, _, written)) = cur_file.take() {
if written != declared {
return Err(invalid_arg(format!(
"file {} declared {} bytes, got {}",
p.display(),
declared,
written
)));
}
}
let meta = meta.ok_or_else(|| invalid_arg("stream had no `meta` frame"))?;
if meta.git_sha.is_empty() {
return Err(invalid_arg("meta.git_sha is required"));
}
let workspace = if meta.workspace.is_empty() {
"workspace_holger".into()
} else {
meta.workspace
};
let repo = if meta.repo.is_empty() {
"_workspace".into()
} else {
meta.repo
};
let branch = if meta.branch.is_empty() {
"unknown".into()
} else {
meta.branch
};
let wh = wctx.warehouse.clone();
// Job ledger: a thin client UPLOADING an index snapshot records the same
// `snapshot` op as the server-side `Index.Snapshot`. Best-effort.
let job = wctx.jobs.start(
nornir::jobs::kind::SNAPSHOT,
&repo,
&workspace,
serde_json::json!({ "git_sha": meta.git_sha, "branch": branch, "source": "upload" }),
);
let snap = match tokio::task::spawn_blocking(move || {
nornir::index::snapshot::snapshot_to_iceberg(
&wh,
&workspace,
&repo,
&meta.git_sha,
&branch,
&root,
)
})
.await
.map_err(internal)?
{
Ok(s) => s,
Err(e) => {
let e = anyhow::anyhow!("{e:#}");
job.fail(&e);
return Err(internal(e));
}
};
job.finish(
serde_json::json!({ "blob_count": snap.blob_count, "total_bytes": snap.total_bytes }),
&format!("index_snapshots:{}", snap.snapshot_id),
);
Ok(Response::new(snapshot_ref_to_pb(snap)))
}
}
fn snapshot_ref_to_pb(s: nornir::index::snapshot::SnapshotRef) -> SnapshotResponse {
SnapshotResponse {
snapshot_id: s.snapshot_id.to_string(),
workspace: s.workspace,
repo: s.repo,
git_sha: s.git_sha,
branch: s.branch,
schema_hash: s.schema_hash,
blob_count: s.blob_count as u64,
total_bytes: s.total_bytes as u64,
}
}
// -- Introspect ---------------------------------------------------------------
struct IntrospectSvc {
state: Shared,
}
fn resolve_binary(workspace_root: &std::path::Path, binary: &str) -> PathBuf {
let p = PathBuf::from(binary);
if p.is_absolute() {
p
} else {
workspace_root.join(p)
}
}
fn sym_to_pb(s: &introspect::artifact::Symbol) -> PbSymbol {
PbSymbol {
name: s.name.clone(),
name_demangled: s.name_demangled.clone(),
name_mangled: s.name_mangled.clone(),
file: s.file.clone(),
line: s.line.unwrap_or(0),
size_bytes: s.size_bytes.unwrap_or(0),
krate: s.krate.clone(),
}
}
#[tonic::async_trait]
impl IntrospectSvcTrait for IntrospectSvc {
async fn symbols(&self, req: Request<BinaryOnly>) -> Result<Response<SymbolList>, Status> {
let wctx = self.state.ws_read(&req)?.clone();
let ws = wctx.loaded.lock().await.workspace_root.clone();
let binary = req.into_inner().binary;
let bin = resolve_binary(&ws, &binary);
let ws2 = ws.clone();
// Job ledger: one `symbol_scan` per DWARF/symbol extraction (best-effort).
let job = wctx.jobs.start(
nornir::jobs::kind::SYMBOL_SCAN,
&binary,
&wctx.name,
serde_json::json!({ "op": "symbols" }),
);
let syms = match tokio::task::spawn_blocking(move || {
introspect::artifact::extract_symbols(&bin, &ws2)
})
.await
.map_err(internal)?
{
Ok(s) => s,
Err(e) => {
job.fail(&e);
return Err(internal(e));
}
};
job.finish(serde_json::json!({ "symbols": syms.len() }), "");
Ok(Response::new(SymbolList {
symbols: syms.iter().map(sym_to_pb).collect(),
}))
}
async fn symbol_lookup(
&self,
req: Request<SymbolLookupRequest>,
) -> Result<Response<SymbolList>, Status> {
let wctx = self.state.ws_read(&req)?.clone();
let r = req.into_inner();
let ws = wctx.loaded.lock().await.workspace_root.clone();
let bin = resolve_binary(&ws, &r.binary);
let ws2 = ws.clone();
let limit = if r.limit == 0 { 50 } else { r.limit as usize };
let pat = r.pattern;
let job = wctx.jobs.start(
nornir::jobs::kind::SYMBOL_SCAN,
&r.binary,
&wctx.name,
serde_json::json!({ "op": "symbol_lookup" }),
);
let syms = match tokio::task::spawn_blocking(
move || -> anyhow::Result<Vec<introspect::artifact::Symbol>> {
let all = introspect::artifact::extract_symbols(&bin, &ws2)?;
Ok(introspect::artifact::lookup(&all, &pat)
.into_iter()
.take(limit)
.cloned()
.collect())
},
)
.await
.map_err(internal)?
{
Ok(s) => s,
Err(e) => {
job.fail(&e);
return Err(internal(e));
}
};
job.finish(serde_json::json!({ "symbols": syms.len() }), "");
Ok(Response::new(SymbolList {
symbols: syms.iter().map(sym_to_pb).collect(),
}))
}
async fn defined_in(
&self,
req: Request<DefinedInRequest>,
) -> Result<Response<SymbolList>, Status> {
let wctx = self.state.ws_read(&req)?.clone();
let r = req.into_inner();
let ws = wctx.loaded.lock().await.workspace_root.clone();
let bin = resolve_binary(&ws, &r.binary);
let ws2 = ws.clone();
let suffix = r.suffix;
let job = wctx.jobs.start(
nornir::jobs::kind::SYMBOL_SCAN,
&r.binary,
&wctx.name,
serde_json::json!({ "op": "defined_in" }),
);
let syms = match tokio::task::spawn_blocking(
move || -> anyhow::Result<Vec<introspect::artifact::Symbol>> {
let all = introspect::artifact::extract_symbols(&bin, &ws2)?;
Ok(introspect::artifact::defined_in(&all, &suffix)
.into_iter()
.cloned()
.collect())
},
)
.await
.map_err(internal)?
{
Ok(s) => s,
Err(e) => {
job.fail(&e);
return Err(internal(e));
}
};
job.finish(serde_json::json!({ "symbols": syms.len() }), "");
Ok(Response::new(SymbolList {
symbols: syms.iter().map(sym_to_pb).collect(),
}))
}
// The three DWARF call-graph verbs walked a petgraph built from
// `callgraph_dwarf::extract_callgraph` — a second DWARF call-graph producer and a
// second graph engine, both removed with Mímir v2 (§7.3d). They refuse by name;
// the binary layer comes back as rows on the code graph (lane 9).
async fn callers(&self, req: Request<CallQuery>) -> Result<Response<NameList>, Status> {
let wctx = self.state.ws_read(&req)?;
let q = req.into_inner();
let params = serde_json::json!({ "binary": q.binary, "name": q.name });
name_list("Introspect.Callers", reader_answer(wctx, "Introspect.Callers", params).await?)
}
async fn callees(&self, req: Request<CallQuery>) -> Result<Response<NameList>, Status> {
let wctx = self.state.ws_read(&req)?;
let q = req.into_inner();
let params = serde_json::json!({ "binary": q.binary, "name": q.name });
name_list("Introspect.Callees", reader_answer(wctx, "Introspect.Callees", params).await?)
}
async fn path_between(
&self,
req: Request<PathBetweenRequest>,
) -> Result<Response<NameList>, Status> {
let wctx = self.state.ws_read(&req)?;
let q = req.into_inner();
let params = serde_json::json!({ "binary": q.binary, "from": q.from, "to": q.to });
name_list(
"Introspect.PathBetween",
reader_answer(wctx, "Introspect.PathBetween", params).await?,
)
}
}
// -- Knowledge ----------------------------------------------------------------
struct KnowledgeSvc {
state: Shared,
}
#[tonic::async_trait]
impl KnowledgeSvcTrait for KnowledgeSvc {
// Every Knowledge.* verb read the persisted `symbol_facts` / `call_edges`
// (and the resolved SCIP map), which have no producer since Mímir v2
// (PLAN-MIMIR-V2 §7.3). They keep their wire types and refuse by name rather
// than answer an empty list that reads as "no such symbol / no callers".
async fn symbol_lookup(
&self,
req: Request<KnowledgeSymbolQuery>,
) -> Result<Response<KnowledgeSymbols>, Status> {
let wctx = self.state.ws_read(&req)?;
let q = req.into_inner();
let params = serde_json::json!({ "repo": q.repo, "arg": q.arg, "limit": q.limit });
knowledge_symbols(
"Knowledge.SymbolLookup",
reader_answer(wctx, "Knowledge.SymbolLookup", params).await?,
)
}
async fn defined_in(
&self,
req: Request<KnowledgeSymbolQuery>,
) -> Result<Response<KnowledgeSymbols>, Status> {
let wctx = self.state.ws_read(&req)?;
let q = req.into_inner();
let params = serde_json::json!({ "repo": q.repo, "arg": q.arg, "limit": q.limit });
knowledge_symbols(
"Knowledge.DefinedIn",
reader_answer(wctx, "Knowledge.DefinedIn", params).await?,
)
}
async fn callers(
&self,
req: Request<KnowledgeCallQuery>,
) -> Result<Response<KnowledgeCalls>, Status> {
let wctx = self.state.ws_read(&req)?;
let q = req.into_inner();
let params = serde_json::json!({ "repo": q.repo, "name": q.name, "limit": q.limit });
knowledge_calls(
"Knowledge.Callers",
reader_answer(wctx, "Knowledge.Callers", params).await?,
)
}
async fn callees(
&self,
req: Request<KnowledgeCallQuery>,
) -> Result<Response<KnowledgeCalls>, Status> {
let wctx = self.state.ws_read(&req)?;
let q = req.into_inner();
let params = serde_json::json!({ "repo": q.repo, "name": q.name, "limit": q.limit });
knowledge_calls(
"Knowledge.Callees",
reader_answer(wctx, "Knowledge.Callees", params).await?,
)
}
async fn call_path(
&self,
req: Request<KnowledgeCallPathQuery>,
) -> Result<Response<NameList>, Status> {
let wctx = self.state.ws_read(&req)?;
let q = req.into_inner();
let params = serde_json::json!({ "repo": q.repo, "from": q.from, "to": q.to });
name_list(
"Knowledge.CallPath",
reader_answer(wctx, "Knowledge.CallPath", params).await?,
)
}
}
// -- CodeRoom -----------------------------------------------------------------
/// `CodeRoom.Answer` — the wire door of the code room's readers
/// (`nornir::code_room::reader::door`): a surface by name, its request as
/// JSON, the reader's JSON object back. The body always carries `state`; a
/// refusal (`not_consumed`, `not_wired`, `no_room` …) is a body with
/// `refused: true`, never a gRPC error — the thin client draws the state by
/// name. A request with no served workspace selected answers `no_room`.
struct CodeRoomSvc {
state: Shared,
}
#[tonic::async_trait]
impl CodeRoomSvcTrait for CodeRoomSvc {
async fn answer(
&self,
req: Request<pb::CodeRoomQuestion>,
) -> Result<Response<pb::JsonResponse>, Status> {
use nornir::code_room::{Refusal, State, reader};
let wctx = match self.state.resolve_ws(req.metadata()) {
Ok((name, ctx)) => {
self.state.authz_read(req.extensions().get::<AuthIdentity>(), &name)?;
Some(ctx)
}
// no workspace selected / not served: the door has no room in reach
Err(why) if why.code() == tonic::Code::NotFound => {
let q = req.get_ref();
let body = reader::refusal_body(&q.surface, &Refusal::new(State::NoRoom, why.message().to_string()));
return Ok(Response::new(pb::JsonResponse { json: body.to_string() }));
}
Err(other) => return Err(other),
};
let q = req.into_inner();
let wctx = wctx.expect("resolved above");
// a whole-room pane is never computed here: kept → at once; else it
// goes on the workspace's jera `rooms` lane and the answer says so
// (owner 2026-10-15: "the ui should place stuff on jera queue if its
// not trivial stuff")
let json = tokio::task::spawn_blocking(move || match reader::ready(&q.surface, &wctx.reach(), &q.params_json) {
reader::Prepared::Ready(body) => body.to_string(),
reader::Prepared::Due { key } => {
let (job_id, running, ms) = queue_room_answer(&wctx, &q.surface, &q.params_json, &key);
reader::queued_body(&q.surface, &job_id, running, ms).to_string()
}
})
.await
.map_err(internal)?;
Ok(Response::new(pb::JsonResponse { json }))
}
}
// -- Docs ---------------------------------------------------------------------
/// DR1/DR2: prepare a repo's sibling **path-deps** for a headless docs render.
///
/// Docs-gen runs `cargo metadata` (depgraph section), which — exactly like the
/// `Mimir.SecurityScan` SBOM path — fails on a single-repo server checkout when
/// the repo declares `dep = { path = "../sibling" }` and the sibling's manifest
/// isn't on disk. Clone (feature `net-scan`) + symlink the siblings from the
/// `git/` scan root first (DR1), then if any path-dep is *still* unresolved
/// return a clear, named error (DR2) rather than letting `cargo metadata` fail
/// with the opaque `MetadataCommand::exec` / `Internal`.
///
/// `scan_root` is the workspace's `git/` dir (every member checkout lives there).
fn prepare_docs_path_deps(
repo: &std::path::Path,
scan_root: &std::path::Path,
) -> anyhow::Result<()> {
let (cloned, linked) = nornir::security::prepare_path_deps(repo, scan_root, &|_| None);
if cloned + linked > 0 {
eprintln!(
"nornir-docs: prepared path-deps for {} ({cloned} cloned, {linked} linked)",
repo.display()
);
}
let unresolved = nornir::security::unresolved_path_dep_siblings(repo, scan_root);
if !unresolved.is_empty() {
let names: Vec<String> = unresolved
.iter()
.map(|(dep, path)| format!("{dep} ({path})"))
.collect();
anyhow::bail!(
"docs render for {repo}: unresolved path-dep sibling(s) {sibs} — their \
Cargo.toml is not on disk beside the repo. Check the sibling repo(s) \
out under {root}, or build nornir-server with the `net-scan` feature \
so it clones them automatically (set NORNIR_DEP_CLONE_BASE for a \
non-default git host).",
repo = repo.display(),
sibs = names.join(", "),
root = scan_root.display(),
);
}
Ok(())
}
struct DocsSvc {
state: Shared,
}
#[tonic::async_trait]
impl DocsSvcTrait for DocsSvc {
async fn init(&self, req: Request<RepoOnly>) -> Result<Response<DocsResponse>, Status> {
let wctx = self.state.ws_write(&req)?.clone();
let name = req.into_inner().repo;
let l = wctx.loaded.lock().await;
let root = repo_root_for(&l, &name)?;
let layout = docs::RepoLayout::new(&root);
let srcs = docs::init_repo(&layout).map_err(internal)?;
Ok(Response::new(DocsResponse {
repo: name,
status: "ok".into(),
artifacts: srcs.into_iter().map(|p| p.display().to_string()).collect(),
detail: layout.nornir_dir().display().to_string(),
}))
}
async fn render(&self, req: Request<RepoOnly>) -> Result<Response<DocsResponse>, Status> {
let wctx = self.state.ws_write(&req)?.clone();
let name = req.into_inner().repo;
let l = wctx.loaded.lock().await;
let root = repo_root_for(&l, &name)?;
let layout = docs::RepoLayout::new(&root);
let last = last_run_for(&l, &name).ok();
let history = history_for(&l, &name);
let ctx = docs::Ctx::new(&root, &l.workspace_root, last.as_ref())
.with_history(&history)
.with_path_dep_cloner(&crate::security::docs_path_dep_cloner);
// Job ledger: one `docs_render` job for this trigger (best-effort).
let job = wctx.jobs.start(
nornir::jobs::kind::DOCS_RENDER,
&name,
&wctx.name,
serde_json::Value::Null,
);
// DR1: clone+symlink sibling path-deps (nornir → skade) into the `git/`
// scan root BEFORE docs-gen runs `cargo metadata`, mirroring the
// `Mimir.SecurityScan` SBOM path so docs resolves the same siblings.
// DR2: a still-unresolved sibling surfaces a clear named error.
if let Err(e) = prepare_docs_path_deps(&root, &l.workspace_root) {
job.fail(&e);
return Err(internal(e));
}
let reports = match docs::render_all(&layout, &ctx) {
Ok(r) => r,
Err(e) => {
job.fail(&e);
return Err(internal(e));
}
};
job.finish(serde_json::json!({ "docs": reports.len() }), "");
Ok(Response::new(DocsResponse {
repo: name,
status: "ok".into(),
artifacts: reports
.iter()
.map(|r| r.output.display().to_string())
.collect(),
detail: format!("{} doc(s) rendered", reports.len()),
}))
}
async fn check(&self, req: Request<RepoOnly>) -> Result<Response<DocsResponse>, Status> {
let wctx = self.state.ws_read(&req)?.clone();
let name = req.into_inner().repo;
let l = wctx.loaded.lock().await;
let root = repo_root_for(&l, &name)?;
let layout = docs::RepoLayout::new(&root);
let last = last_run_for(&l, &name).ok();
let history = history_for(&l, &name);
let ctx = docs::Ctx::new(&root, &l.workspace_root, last.as_ref())
.with_history(&history)
.with_path_dep_cloner(&crate::security::docs_path_dep_cloner);
let (status, detail) = match docs::render_check_all(&layout, &ctx) {
Ok(_) => ("ok".to_string(), String::new()),
Err(e) => ("drift".to_string(), format!("{e:#}")),
};
Ok(Response::new(DocsResponse {
repo: name,
status,
artifacts: Vec::new(),
detail,
}))
}
async fn history(
&self,
req: Request<DocsHistoryRequest>,
) -> Result<Response<DocsHistoryResponse>, Status> {
let wctx = self.state.ws_read(&req)?.clone();
let r = req.into_inner();
let repo = r.repo.clone();
let filter = docs::ExportFilter {
doc_name: if r.doc.is_empty() { None } else { Some(r.doc) },
version: if r.version.is_empty() {
None
} else {
Some(r.version)
},
format: if r.format.is_empty() {
None
} else {
Some(r.format)
},
limit: Some(if r.limit == 0 { 50 } else { r.limit as usize }),
};
let wh = wctx.warehouse.clone();
let rows = tokio::task::spawn_blocking(move || docs::list_doc_exports(&wh, &repo, &filter))
.await
.map_err(internal)?
.map_err(internal)?;
let entries = rows
.into_iter()
.map(|e| PbDocExport {
doc: e.doc_name,
version: e.version,
format: e.format,
path: e.export_id,
exported_at: e.generated_at,
size_bytes: e.byte_len.max(0) as u64,
})
.collect();
Ok(Response::new(DocsHistoryResponse { entries }))
}
async fn export(
&self,
req: Request<pb::DocsExportRequest>,
) -> Result<Response<pb::DocsExportResponse>, Status> {
#[cfg(not(feature = "docs-export"))]
{
let _ = req;
Err(unimplemented_verb(
"nornir-server was built without the `docs-export` feature; \
rebuild with `--features server,docs-export` to use Docs.Export",
))
}
#[cfg(feature = "docs-export")]
{
self.render_export(req, "README").await
}
}
async fn book(
&self,
req: Request<pb::DocsExportRequest>,
) -> Result<Response<pb::DocsExportResponse>, Status> {
#[cfg(not(feature = "docs-export"))]
{
let _ = req;
Err(unimplemented_verb(
"nornir-server was built without the `docs-export` feature; \
rebuild with `--features server,docs-export` to use Docs.Book",
))
}
#[cfg(feature = "docs-export")]
{
self.render_export(req, "book").await
}
}
}
#[cfg(feature = "docs-export")]
impl DocsSvc {
/// Shared render+export+historize for both `Docs.Export` ("README", single
/// doc) and `Docs.Book` ("book", whole doc set). The server owns the
/// warehouse handle, so unlike the embedded MCP path it writes through the
/// held `Arc<IcebergWarehouse>` rather than re-opening. Typst render + the
/// iceberg write are blocking (the warehouse drives its own runtime), so the
/// heavy work runs off the async worker thread, with the `loaded` lock dropped.
async fn render_export(
&self,
req: Request<pb::DocsExportRequest>,
doc: &'static str,
) -> Result<Response<pb::DocsExportResponse>, Status> {
let wctx = self.state.ws_read(&req)?.clone();
let r = req.into_inner();
let repo_name = r.repo.clone();
let fmt_str = if r.format.is_empty() {
"pdf".to_string()
} else {
r.format.clone()
};
let (root, workspace_root, last, history) = {
let l = wctx.loaded.lock().await;
let root = repo_root_for(&l, &repo_name)?;
let last = last_run_for(&l, &repo_name).ok();
let history = history_for(&l, &repo_name);
(root, l.workspace_root.clone(), last, history)
};
let wh = wctx.warehouse.clone();
let fmt_for_closure = fmt_str.clone();
// Job ledger: `docs_book` for the whole-doc-set book, else `docs_export`
// for the single README export. Best-effort.
let job_kind = if doc == "book" {
nornir::jobs::kind::DOCS_BOOK
} else {
nornir::jobs::kind::DOCS_EXPORT
};
let job = wctx.jobs.start(
job_kind,
&repo_name,
&wctx.name,
serde_json::json!({ "doc": doc, "format": fmt_str }),
);
let blocking = tokio::task::spawn_blocking(move || -> anyhow::Result<_> {
let layout = docs::RepoLayout::new(&root);
let ctx = docs::Ctx::new(&root, &workspace_root, last.as_ref())
.with_history(&history)
.with_path_dep_cloner(&crate::security::docs_path_dep_cloner);
// DR1: clone+symlink sibling path-deps before docs-gen's
// `cargo metadata`, same as the SBOM scan; DR2: clear named error
// if a sibling stays unresolved (e.g. net-scan off, no checkout).
prepare_docs_path_deps(&root, &workspace_root)?;
// Render managed artifacts first so the export reflects latest source.
docs::render_all(&layout, &ctx)?;
let format = docs::DocFormat::parse(&fmt_for_closure)?;
let ext = format.extension();
// Single doc (README) vs the whole book.
let (bytes, sources): (Vec<u8>, Vec<String>) = if doc == "book" {
let (bytes, srcs) = docs::build_book(&root, &ctx, format)?;
(
bytes,
srcs.iter().map(|p| p.display().to_string()).collect(),
)
} else {
(docs::export_repo(&root, format)?, Vec::new())
};
let version = docs::resolve_version(&root);
let out_path = layout.export_path(doc, ext);
if let Some(parent) = out_path.parent() {
std::fs::create_dir_all(parent)?;
}
std::fs::write(&out_path, &bytes)?;
let workspace = workspace_root
.file_name()
.and_then(|x| x.to_str())
.unwrap_or("_workspace")
.to_string();
let git_sha = nornir::gitio::head_sha(&root).unwrap_or_else(|_| "unknown".to_string());
let record = docs::record_doc_export(
&wh, &workspace, &repo_name, doc, &version, ext, &git_sha, &bytes,
)?;
Ok((out_path, bytes.len(), sources, record))
})
.await
.map_err(internal)?;
let (out_path, nbytes, sources, record) = match blocking {
Ok(v) => v,
Err(e) => {
job.fail(&e);
return Err(internal(e));
}
};
job.finish(
serde_json::json!({ "bytes": nbytes, "format": fmt_str }),
&format!("doc_exports:{}", record.export_id),
);
Ok(Response::new(pb::DocsExportResponse {
repo: r.repo,
format: fmt_str,
bytes: nbytes as u64,
out: out_path.display().to_string(),
sha256: record.sha256,
git_sha: record.git_sha,
export_id: record.export_id,
sources,
}))
}
}
// -- Mimir (cross-repo dependency graph) -------------------------------------
struct MimirSvc {
state: Shared,
}
impl MimirSvc {
/// Resolve the workspace from the request, get-or-build its graph, run the
/// shared `nornir::mimir` query `f`, and wrap the JSON for the wire.
async fn json<F>(
&self,
md: &tonic::metadata::MetadataMap,
f: F,
) -> Result<Response<pb::JsonResponse>, Status>
where
F: FnOnce(
&nornir::warehouse::dep_graph::WorkspaceGraph,
) -> anyhow::Result<serde_json::Value>,
{
let g = self.state.ws(md)?.graph().await?;
let value = f(&g.graph).map_err(internal)?;
let json = serde_json::to_string(&value).map_err(internal)?;
Ok(Response::new(pb::JsonResponse { json }))
}
}
#[tonic::async_trait]
impl MimirSvcTrait for MimirSvc {
async fn deps_of(
&self,
req: Request<pb::DepQuery>,
) -> Result<Response<pb::JsonResponse>, Status> {
let (md, q) = (req.metadata().clone(), req.get_ref().clone());
self.json(&md, |g| nornir::mimir::deps_of(g, &q.repo, q.transitive))
.await
}
async fn dependents_of(
&self,
req: Request<pb::DepQuery>,
) -> Result<Response<pb::JsonResponse>, Status> {
let (md, q) = (req.metadata().clone(), req.get_ref().clone());
self.json(&md, |g| {
nornir::mimir::dependents_of(g, &q.repo, q.transitive)
})
.await
}
async fn affected_by_change(
&self,
req: Request<pb::AffectedQuery>,
) -> Result<Response<pb::JsonResponse>, Status> {
let (md, q) = (req.metadata().clone(), req.get_ref().clone());
self.json(&md, |g| nornir::mimir::affected_by_change(g, &q.repos))
.await
}
async fn build_order(&self, req: Request<Empty>) -> Result<Response<pb::JsonResponse>, Status> {
let md = req.metadata().clone();
self.json(&md, nornir::mimir::build_order).await
}
async fn dep_path(
&self,
req: Request<pb::DepPathQuery>,
) -> Result<Response<pb::JsonResponse>, Status> {
let (md, q) = (req.metadata().clone(), req.get_ref().clone());
self.json(&md, |g| nornir::mimir::dep_path(g, &q.from, &q.to))
.await
}
async fn external_dep_users(
&self,
req: Request<pb::CrateQuery>,
) -> Result<Response<pb::JsonResponse>, Status> {
let (md, q) = (req.metadata().clone(), req.get_ref().clone());
self.json(&md, |g| Ok(nornir::mimir::external_dep_users(g, &q.krate)))
.await
}
async fn svg(&self, req: Request<Empty>) -> Result<Response<pb::JsonResponse>, Status> {
let g = self.state.ws_read(&req)?.graph().await?;
// Body is static SVG text, carried in the `json` string field.
Ok(Response::new(pb::JsonResponse {
json: nornir::mimir::svg(&g.graph),
}))
}
async fn repo_overview(
&self,
req: Request<RepoOnly>,
) -> Result<Response<pb::JsonResponse>, Status> {
let wctx = self.state.ws_read(&req)?.clone();
let repo = req.into_inner().repo;
let g = wctx.graph().await?;
// The knowledge block is the code room's (a room build folds the log) → blocking.
let value = tokio::task::spawn_blocking(move || {
nornir::mimir::repo_overview_in(&g.graph, &wctx.reach(), &repo)
})
.await
.map_err(internal)?
.map_err(internal)?;
Ok(Response::new(pb::JsonResponse {
json: serde_json::to_string(&value).map_err(internal)?,
}))
}
// ── Mimir LENSES (QUERY half lifted from dwarves::mimir, L5a) ────────────
//
// Mímir v2 (PLAN-MIMIR-V2 §7): the jera lens chain, its `lens_<name>` result
// tables and the warehouse read/join halves are gone. Every table-fed lens
// goes through ONE door — `lens_answer` → `mimir_lens::answer(lens, reach,
// params)` — with the wire query's fields as JSON. A lens whose v2 feed is
// not written answers the lens envelope with `lens_meta.state` saying so
// (never an empty `worst`); subclone folds source tokens and stays direct.
async fn generic_cost(
&self,
req: Request<pb::GenericCostQuery>,
) -> Result<Response<pb::JsonResponse>, Status> {
let wctx = self.state.ws_read(&req)?;
let q = req.into_inner();
lens_answer(wctx, "generic_cost", serde_json::json!({ "repo": q.repo, "min_instantiations": q.min_instantiations, "limit": q.limit })).await
}
async fn zombie_features(
&self,
req: Request<pb::LensQuery>,
) -> Result<Response<pb::JsonResponse>, Status> {
let wctx = self.state.ws_read(&req)?;
let q = req.into_inner();
lens_answer(wctx, "zombie_features", serde_json::json!({ "repo": q.repo, "limit": q.limit })).await
}
async fn change_risk(
&self,
req: Request<pb::LensQuery>,
) -> Result<Response<pb::JsonResponse>, Status> {
let wctx = self.state.ws_read(&req)?;
let q = req.into_inner();
lens_answer(wctx, "change_risk", serde_json::json!({ "repo": q.repo, "limit": q.limit })).await
}
async fn similar_code(
&self,
req: Request<pb::LensQuery>,
) -> Result<Response<pb::JsonResponse>, Status> {
let wctx = self.state.ws_read(&req)?;
let q = req.into_inner();
lens_answer(wctx, "similar_code", serde_json::json!({ "repo": q.repo, "limit": q.limit })).await
}
async fn semantic_clones(
&self,
req: Request<pb::LensQuery>,
) -> Result<Response<pb::JsonResponse>, Status> {
let wctx = self.state.ws_read(&req)?;
let q = req.into_inner();
lens_answer(wctx, "semantic_clones", serde_json::json!({ "repo": q.repo, "limit": q.limit })).await
}
async fn exact_clones(
&self,
req: Request<pb::LensQuery>,
) -> Result<Response<pb::JsonResponse>, Status> {
let wctx = self.state.ws_read(&req)?;
let q = req.into_inner();
lens_answer(wctx, "exact_clones", serde_json::json!({ "repo": q.repo, "limit": q.limit })).await
}
async fn near_clones(
&self,
req: Request<pb::LensQuery>,
) -> Result<Response<pb::JsonResponse>, Status> {
let wctx = self.state.ws_read(&req)?;
let q = req.into_inner();
lens_answer(wctx, "near_clones", serde_json::json!({ "repo": q.repo, "limit": q.limit })).await
}
async fn structural_clones(
&self,
req: Request<pb::LensQuery>,
) -> Result<Response<pb::JsonResponse>, Status> {
let wctx = self.state.ws_read(&req)?;
let q = req.into_inner();
lens_answer(wctx, "structural_clones", serde_json::json!({ "repo": q.repo, "limit": q.limit })).await
}
async fn role_neighbours(
&self,
req: Request<pb::LensQuery>,
) -> Result<Response<pb::JsonResponse>, Status> {
let wctx = self.state.ws_read(&req)?;
let q = req.into_inner();
lens_answer(wctx, "role_neighbours", serde_json::json!({ "repo": q.repo, "limit": q.limit })).await
}
async fn semantic_twins(
&self,
req: Request<pb::LensQuery>,
) -> Result<Response<pb::JsonResponse>, Status> {
let wctx = self.state.ws_read(&req)?;
let q = req.into_inner();
lens_answer(wctx, "semantic_twins", serde_json::json!({ "repo": q.repo, "limit": q.limit })).await
}
async fn reimplements_shared(
&self,
req: Request<pb::LensQuery>,
) -> Result<Response<pb::JsonResponse>, Status> {
let wctx = self.state.ws_read(&req)?;
let q = req.into_inner();
lens_answer(wctx, "reimplements_shared", serde_json::json!({ "repo": q.repo, "limit": q.limit })).await
}
async fn same_intent(
&self,
req: Request<pb::LensQuery>,
) -> Result<Response<pb::JsonResponse>, Status> {
let wctx = self.state.ws_read(&req)?;
let q = req.into_inner();
lens_answer(wctx, "same_intent", serde_json::json!({ "repo": q.repo, "limit": q.limit })).await
}
async fn weak_tests(
&self,
req: Request<pb::LensQuery>,
) -> Result<Response<pb::JsonResponse>, Status> {
let wctx = self.state.ws_read(&req)?;
let q = req.into_inner();
lens_answer(wctx, "weak_tests", serde_json::json!({ "repo": q.repo, "limit": q.limit })).await
}
async fn bench_gaps(
&self,
req: Request<pb::LensQuery>,
) -> Result<Response<pb::JsonResponse>, Status> {
let wctx = self.state.ws_read(&req)?;
let q = req.into_inner();
lens_answer(wctx, "bench_gaps", serde_json::json!({ "repo": q.repo, "limit": q.limit })).await
}
async fn villa(
&self,
req: Request<pb::LensQuery>,
) -> Result<Response<pb::JsonResponse>, Status> {
let wctx = self.state.ws_read(&req)?;
let q = req.into_inner();
lens_answer(wctx, "villa", serde_json::json!({ "repo": q.repo, "limit": q.limit })).await
}
async fn subclone(
&self,
req: Request<pb::LensQuery>,
) -> Result<Response<pb::JsonResponse>, Status> {
let wctx = self.state.ws_read(&req)?;
let q = req.into_inner();
// the SOURCE door: repo="" is subclone's structural refusal naming
// the askable repos — cheap, so it rides the same queued path
lens_answer(wctx, "subclone", serde_json::json!({ "repo": q.repo, "limit": q.limit })).await
}
async fn group_twins(
&self,
req: Request<pb::LensQuery>,
) -> Result<Response<pb::JsonResponse>, Status> {
let wctx = self.state.ws_read(&req)?;
let q = req.into_inner();
// the second SOURCE door: repo="" is group_twins' structural refusal
// naming the askable repos; a scan rides the queue like subclone's
lens_answer(wctx, "group_twins", serde_json::json!({ "repo": q.repo, "limit": q.limit })).await
}
// ── the CHANGE / ANOMALY lens family (over the TEMPORAL graph) ───────────
async fn churn(
&self,
req: Request<pb::ChurnQuery>,
) -> Result<Response<pb::JsonResponse>, Status> {
let wctx = self.state.ws_read(&req)?;
let q = req.into_inner();
lens_answer(wctx, "churn", serde_json::json!({ "repo": q.repo, "band": q.band, "cutoff_low": q.cutoff_low, "cutoff_high": q.cutoff_high, "limit": q.limit })).await
}
async fn anomaly(
&self,
req: Request<pb::AnomalyQuery>,
) -> Result<Response<pb::JsonResponse>, Status> {
let wctx = self.state.ws_read(&req)?;
let q = req.into_inner();
lens_answer(wctx, "anomaly", serde_json::json!({ "repo": q.repo, "z_threshold": q.z_threshold, "limit": q.limit })).await
}
async fn world_impact(
&self,
req: Request<pb::LensQuery>,
) -> Result<Response<pb::JsonResponse>, Status> {
let wctx = self.state.ws_read(&req)?;
let q = req.into_inner();
lens_answer(wctx, "world_impact", serde_json::json!({ "repo": q.repo, "limit": q.limit })).await
}
/// `Mimir.BinaryDrift` — source(AST) vs binary(DWARF) diff, the compile-time
/// injection detector: any binary symbol/edge not derivable from the audited
/// source (after the monomorphisation/glue/derive allowlist) was injected by a
/// `build.rs`, a proc-macro, or a compromised toolchain.
async fn binary_drift(
&self,
req: Request<pb::LensQuery>,
) -> Result<Response<pb::JsonResponse>, Status> {
let wctx = self.state.ws_read(&req)?;
let q = req.into_inner();
lens_answer(wctx, "binary_drift", serde_json::json!({ "repo": q.repo, "limit": q.limit })).await
}
/// `Mimir.DataReach` (Gemini #2) — sensitive-data reachability / leak paths:
/// `warehouse_access_edges` × the reversed `call_edges` graph × `symbol_facts`
/// visibility. Honest MISSING-FEED until the access-edges producer persists rows.
async fn data_reach(
&self,
req: Request<pb::LensQuery>,
) -> Result<Response<pb::JsonResponse>, Status> {
let wctx = self.state.ws_read(&req)?;
let q = req.into_inner();
lens_answer(wctx, "data_reach", serde_json::json!({ "repo": q.repo, "limit": q.limit })).await
}
/// `Mimir.Trust` (Gemini 2A) — the provenance-trust semiring over `call_edges`:
/// walk FORWARD from the entry surface under `(⊕=max over paths, ⊗=min
/// weakest-hop)`, dock local trust by shell-out impurity × sensitive-data touch ×
/// test coverage, map inherited trust to act/hold/refuse. The code-provenance
/// decision modgunn's gate wants, made queryable.
async fn trust(
&self,
req: Request<pb::TrustQuery>,
) -> Result<Response<pb::JsonResponse>, Status> {
let wctx = self.state.ws_read(&req)?;
let q = req.into_inner();
lens_answer(wctx, "trust", serde_json::json!({ "repo": q.repo, "act_at": q.act_at, "hold_at": q.hold_at, "limit": q.limit })).await
}
/// `Mimir.NameCollision` (idea #34) — CPG graph-soundness self-audit: leaf fn
/// names defined in more than one place, which the string-name call-graph matcher
/// conflates into one node, and the fraction of `call_edges` pointing at such an
/// ambiguous callee. Measures how much every name-matched reachability join can
/// be trusted.
async fn name_collision(
&self,
req: Request<pb::LensQuery>,
) -> Result<Response<pb::JsonResponse>, Status> {
let wctx = self.state.ws_read(&req)?;
let q = req.into_inner();
lens_answer(wctx, "name_collision", serde_json::json!({ "repo": q.repo, "limit": q.limit })).await
}
/// `Mimir.SymbolIdentity` — the SUBSTRATE invariant gate under every duplicate
/// claim: is `scip_occurrences.symbol` actually an identity (one definition per
/// symbol per snapshot)? Where [`Self::name_collision`] audits the SYN layer's
/// leaf-name matcher, this audits the RESOLVED layer the similarity tier joins
/// against — a detector reasoning over an ambiguous symbol table cannot be
/// trusted about duplicates however good its cosine threshold, because "these
/// two symbols are the same" is exactly the judgement the substrate was already
/// getting wrong. Splits OURS (bare `local <id>`) from UPSTREAM (rust-analyzer).
async fn symbol_identity(
&self,
req: Request<pb::LensQuery>,
) -> Result<Response<pb::JsonResponse>, Status> {
let wctx = self.state.ws_read(&req)?;
let q = req.into_inner();
lens_answer(wctx, "symbol_identity", serde_json::json!({ "repo": q.repo, "limit": q.limit })).await
}
/// `Mimir.ReachableVuln` (baddie-detection b1) — `vuln_findings` ×
/// `sbom_components` × `call_edges`: a CVE present in a repo's resolved components
/// is refuse 🔴 only if the vulnerable crate is actually CALLED (reachability gate),
/// else hold 🟠. Empty until modgunn's `warehouse scan --persist` feeds `vuln_findings`.
async fn reachable_vuln(
&self,
req: Request<pb::LensQuery>,
) -> Result<Response<pb::JsonResponse>, Status> {
let wctx = self.state.ws_read(&req)?;
let q = req.into_inner();
lens_answer(wctx, "reachable_vuln", serde_json::json!({ "repo": q.repo, "limit": q.limit })).await
}
/// `Mimir.UndocApi` (idea #16) — undocumented public API in churny files:
/// `symbol_facts` (pub, doc_lines==0) × `git_heat_facts` (per-file churn) on
/// (repo, file), ranked by recent churn × author breadth.
async fn undoc_api(
&self,
req: Request<pb::UndocApiQuery>,
) -> Result<Response<pb::JsonResponse>, Status> {
let wctx = self.state.ws_read(&req)?;
let q = req.into_inner();
lens_answer(wctx, "undoc_api", serde_json::json!({ "repo": q.repo, "min_commits_90d": q.min_commits_90d, "limit": q.limit })).await
}
/// Deep SBOM + vulnerability + license scan of a workspace repo — server-side
/// (`cargo metadata` over the monitored `git/<repo>` checkout + OSV), so the
/// viz never needs the sources on disk. The workspace's `workspace_root` is
/// the `git/` dir, so `join(repo)` is the member checkout.
async fn security_scan(
&self,
req: Request<RepoOnly>,
) -> Result<Response<pb::JsonResponse>, Status> {
let wctx = self.state.ws_read(&req)?.clone();
let repo = req.into_inner().repo;
let repo_dir = wctx.loaded.lock().await.workspace_root.join(&repo);
// The `git/` dir holding every member checkout — where sibling path-dep
// repos are cloned/linked so a headless `cargo metadata` resolves.
let scan_root = repo_dir.parent().map(|p| p.to_path_buf());
let wh = wctx.warehouse.clone();
// Job ledger: one `security_scan` job per scan trigger (best-effort).
let job = wctx.jobs.start(
nornir::jobs::kind::SECURITY_SCAN,
&repo,
&wctx.name,
serde_json::Value::Null,
);
let value = match tokio::task::spawn_blocking(move || -> anyhow::Result<serde_json::Value> {
use std::collections::HashMap;
// FIX (live "press Security" path): clone+link the repo's path-dep
// siblings (e.g. nornir → skade/skade-katalog) into the scan root
// FIRST, so `cargo metadata` can load the manifest. Without this the
// scan hard-failed on any repo with sibling path-deps. Best-effort:
// a miss degrades to the Cargo.lock fallback inside `components`.
if let Some(root) = &scan_root {
let (cloned, linked) =
nornir::security::prepare_path_deps(&repo_dir, root, &|_| None);
if cloned + linked > 0 {
eprintln!(
"nornir-security: prepared path-deps for scan of {} ({cloned} cloned, {linked} linked)",
repo_dir.display()
);
}
}
// SBOM components from cargo metadata.
let (repo_name, comps) = nornir::security::components(&repo_dir)?;
let now = chrono::Utc::now().timestamp_micros();
// Warehouse-sourced VERDICT + advisory match (real severity/EPSS/KEV):
// scan the components against modgunn's `cve_advisories` ONCE. This is
// the single advisory match when the warehouse is populated — no OSV +
// rustsec double-match. Lock-tolerant read: the server holds the catalog
// write lock, so this reads a copied-aside snapshot (`open_read_only`).
let wh_verdicts = nornir::security::verdict::verdicts_from_warehouse(
wh.root(),
&comps,
nornir::security::Policy::default(),
);
// "Populated" = the load saw a real advisory revision; an empty/unimported
// warehouse must NOT be read as "no vulns" — fall back to query_vulns.
let populated = wh_verdicts
.as_ref()
.map(|vs| vs.iter().any(|v| !v.advisory_db_rev.is_empty()))
.unwrap_or(false);
// Display vulns[]: derive from the warehouse verdicts' findings when the
// warehouse is populated (one in-memory match, zero network); otherwise
// fall back to query_vulns + the vuln_findings cache so an un-imported
// deployment behaves exactly as before.
let (vulns, cache_hits, cache_misses) = match (&wh_verdicts, populated) {
(Ok(verdicts), true) => {
let mut vulns = Vec::new();
for v in verdicts {
if !v.findings.is_empty() {
let ids: Vec<String> = v.findings.iter().map(|f| f.id.clone()).collect();
vulns.push(serde_json::json!({
"crate": v.artifact.name, "version": v.artifact.version, "ids": ids,
}));
}
}
// The warehouse advisory db is a single in-memory source — no
// per-crate network probes, so every component is "served".
(vulns, comps.len(), 0usize)
}
_ => {
// ── Fallback: warehouse not imported — match via query_vulns,
// keyed by the immutable crate@version `vuln_findings` cache ──
let cache = wh.query_vuln_findings().unwrap_or_default();
let key = |c: &nornir::security::Component| (c.name.clone(), c.version.clone());
const TTL_MICROS: i64 = 7 * 24 * 3600 * 1_000_000;
let is_fresh = |c: &nornir::security::Component| {
cache.get(&key(c)).map(|(_, _, ts)| now - *ts < TTL_MICROS).unwrap_or(false)
};
let misses: Vec<nornir::security::Component> =
comps.iter().filter(|c| !is_fresh(c)).cloned().collect();
let fresh = nornir::security::query_vulns(&misses)?;
let fresh_map: HashMap<(String, String), &nornir::security::Vuln> =
fresh.iter().map(|v| ((v.crate_name.clone(), v.version.clone()), v)).collect();
// Persist a row for every miss (clean ones too) so the next scan
// is a pure cache hit (zero network in steady state).
let rows: Vec<nornir::warehouse::iceberg::VulnFinding> = misses
.iter()
.map(|c| {
let (ids, summary) = fresh_map
.get(&key(c))
.map(|v| (v.ids.clone(), v.summary.clone()))
.unwrap_or_default();
nornir::warehouse::iceberg::VulnFinding {
crate_name: c.name.clone(),
version: c.version.clone(),
ids,
summary,
checked_at_micros: now,
}
})
.collect();
let _ = wh.append_vuln_findings(&rows);
let mut vulns = Vec::new();
for c in &comps {
let ids: Vec<String> = if is_fresh(c) {
cache.get(&key(c)).map(|(i, _, _)| i.clone()).unwrap_or_default()
} else {
fresh_map.get(&key(c)).map(|v| v.ids.clone()).unwrap_or_default()
};
if !ids.is_empty() {
vulns.push(serde_json::json!({"crate": c.name, "version": c.version, "ids": ids}));
}
}
let hits = comps.len().saturating_sub(misses.len());
(vulns, hits, misses.len())
}
};
let mut lic: std::collections::BTreeMap<String, usize> = Default::default();
for c in &comps {
*lic.entry(c.license.clone()).or_default() += 1;
}
let mut lic: Vec<_> = lic.into_iter().collect();
lic.sort_by(|a, b| b.1.cmp(&a.1).then(a.0.cmp(&b.0)));
// Fold the Pass/Warn/Block verdict from the same warehouse scan. An
// authoritative `decision` is emitted ONLY when the warehouse is
// populated (real severities); on an un-imported warehouse there is no
// severity to fold, and the display `vulns[]` came from the query_vulns
// fallback — so we omit `decision` rather than assert a misleading PASS
// (consumers then show the raw vuln count, no badge).
let verdict = match &wh_verdicts {
Ok(verdicts) if populated => {
use nornir::security::Decision;
let (mut pass, mut warn, mut block) = (0usize, 0usize, 0usize);
for v in verdicts {
match v.decision {
Decision::Pass => pass += 1,
Decision::Warn => warn += 1,
Decision::Block => block += 1,
}
}
let decision = if block > 0 {
"block"
} else if warn > 0 {
"warn"
} else {
"pass"
};
serde_json::json!({
"decision": decision,
"pass": pass,
"warn": warn,
"block": block,
"source": "warehouse",
"advisory_db_rev": verdicts
.first()
.map(|v| v.advisory_db_rev.clone())
.unwrap_or_default(),
})
}
// Warehouse not imported yet — no authoritative verdict to show.
Ok(_) => serde_json::json!({ "source": "empty" }),
Err(e) => serde_json::json!({ "error": e.to_string() }),
};
Ok(serde_json::json!({
"repo": repo_name,
"components": comps.len(),
"vulns": vulns,
"verdict": verdict,
"licenses": lic.into_iter().take(12).map(|(l, n)| serde_json::json!([l, n])).collect::<Vec<_>>(),
"cache_hits": cache_hits,
"cache_misses": cache_misses,
}))
})
.await
.map_err(internal)?
{
Ok(v) => v,
Err(e) => {
job.fail(&e);
return Err(internal(e));
}
};
job.finish(
serde_json::json!({
"components": value.get("components").cloned().unwrap_or(serde_json::Value::Null),
"vulns": value.get("vulns").and_then(|v| v.as_array()).map(|a| a.len()).unwrap_or(0),
}),
"",
);
Ok(Response::new(pb::JsonResponse {
json: serde_json::to_string(&value).map_err(internal)?,
}))
}
async fn changed_since_last_release(
&self,
req: Request<Empty>,
) -> Result<Response<pb::JsonResponse>, Status> {
let wctx = self.state.ws_read(&req)?.clone();
let g = wctx.graph().await?;
let change = nornir::change::detect(&wctx.warehouse, &g.graph, &g.workspace_name)
.await
.map_err(internal)?;
Ok(Response::new(pb::JsonResponse {
json: serde_json::to_string(&change).map_err(internal)?,
}))
}
/// `Mimir.SourceWindow` — a windowed source read for the thin arch/jump-to-
/// source preview. The server holds the checkouts, so it resolves `repo` to
/// its on-disk clone (`repo_dir` = `workspace_root/<repo>`) and reads `context`
/// lines each side of 1-based `line` from `file` via the SAME
/// [`nornir::arch::source_window`] the fat/local pane uses — so thin == fat
/// bytes. An empty `repo` (the thin arch client resolves a symbol to a bare
/// `src/…` path with no member attached) makes the handler try each configured
/// member checkout and return the FIRST that yields a non-empty window. An
/// unreadable/absent file is `{start_line:0, lines:[]}`, never an error.
async fn source_window(
&self,
req: Request<pb::SourceWindowQuery>,
) -> Result<Response<pb::JsonResponse>, Status> {
let wctx = self.state.ws_read(&req)?.clone();
let q = req.into_inner();
let (root, members) = {
let l = wctx.loaded.lock().await;
(
l.workspace_root.clone(),
l.nornir.repo.keys().cloned().collect::<Vec<_>>(),
)
};
let value = tokio::task::spawn_blocking(move || -> serde_json::Value {
// Candidate checkout roots: the named repo when given, else every
// configured member (first non-empty window wins).
let roots: Vec<std::path::PathBuf> = if q.repo.is_empty() {
members
.iter()
.map(|m| config::Nornir::repo_dir(&root, m))
.collect()
} else {
vec![config::Nornir::repo_dir(&root, &q.repo)]
};
for r in &roots {
let (start, lines) =
nornir::arch::source_window(Some(r), &q.file, q.line, q.context);
if !lines.is_empty() {
return serde_json::json!({ "start_line": start, "lines": lines });
}
}
serde_json::json!({ "start_line": 0u32, "lines": Vec::<String>::new() })
})
.await
.map_err(internal)?;
Ok(Response::new(pb::JsonResponse {
json: serde_json::to_string(&value).map_err(internal)?,
}))
}
}
// -- Dwarf (warehouse-backed persisted DWARF facts) --------------------------
struct DwarfSvc {
state: Shared,
}
impl DwarfSvc {
/// Load the persisted `DwarfFacts` for `(repo, sha)` (latest if `sha` empty)
/// off the async worker, run `f`, and wrap the JSON. Mirrors the embedded
/// MCP `dwarf_load_json`; both call the same `load_dwarf` + DwarfFacts methods.
async fn query<F>(
&self,
md: &tonic::metadata::MetadataMap,
repo: String,
sha: String,
f: F,
) -> Result<Response<pb::JsonResponse>, Status>
where
F: FnOnce(&introspect::persist::DwarfFacts) -> anyhow::Result<String> + Send + 'static,
{
let wctx = self.state.ws(md)?.clone();
let wh = wctx.warehouse.clone();
let warehouse_root = wctx.loaded.lock().await.warehouse_root();
let json = tokio::task::spawn_blocking(move || -> anyhow::Result<String> {
let into = warehouse_root
.parent()
.unwrap_or(warehouse_root.as_path())
.join("cache/dwarf")
.join(&repo);
let sha = if sha.is_empty() {
None
} else {
Some(sha.as_str())
};
let facts = introspect::persist::load_dwarf(&wh, &repo, sha, &into)?;
f(&facts)
})
.await
.map_err(internal)?
.map_err(internal)?;
Ok(Response::new(pb::JsonResponse { json }))
}
}
#[tonic::async_trait]
impl DwarfSvcTrait for DwarfSvc {
async fn symbol_lookup(
&self,
req: Request<pb::DwarfQuery>,
) -> Result<Response<pb::JsonResponse>, Status> {
let (md, q) = (req.metadata().clone(), req.get_ref().clone());
let limit = if q.limit == 0 { 50 } else { q.limit as usize };
self.query(&md, q.repo, q.sha, move |facts| {
let hits: Vec<_> = facts.lookup(&q.arg).into_iter().take(limit).collect();
Ok(serde_json::to_string(&hits)?)
})
.await
}
async fn defined_in(
&self,
req: Request<pb::DwarfQuery>,
) -> Result<Response<pb::JsonResponse>, Status> {
let (md, q) = (req.metadata().clone(), req.get_ref().clone());
let limit = if q.limit == 0 { 100 } else { q.limit as usize };
self.query(&md, q.repo, q.sha, move |facts| {
let hits: Vec<_> = facts.defined_in(&q.arg).into_iter().take(limit).collect();
Ok(serde_json::to_string(&hits)?)
})
.await
}
// The persisted DWARF inline-call edges were walked through the removed
// petgraph `callgraph_dwarf::Callgraph` (Mímir v2 §7.3d) — refused by name.
async fn callers(
&self,
req: Request<pb::DwarfQuery>,
) -> Result<Response<pb::JsonResponse>, Status> {
let wctx = self.state.ws(req.metadata())?;
let q = req.into_inner();
let params = serde_json::json!({
"repo": q.repo, "arg": q.arg, "sha": q.sha, "limit": q.limit,
});
let value = reader_answer(wctx, "Dwarf.Callers", params).await?;
Ok(Response::new(pb::JsonResponse { json: value.to_string() }))
}
async fn callees(
&self,
req: Request<pb::DwarfQuery>,
) -> Result<Response<pb::JsonResponse>, Status> {
let wctx = self.state.ws(req.metadata())?;
let q = req.into_inner();
let params = serde_json::json!({
"repo": q.repo, "arg": q.arg, "sha": q.sha, "limit": q.limit,
});
let value = reader_answer(wctx, "Dwarf.Callees", params).await?;
Ok(Response::new(pb::JsonResponse { json: value.to_string() }))
}
async fn call_path(
&self,
req: Request<pb::DwarfPathQuery>,
) -> Result<Response<pb::JsonResponse>, Status> {
let wctx = self.state.ws(req.metadata())?;
let q = req.into_inner();
let params = serde_json::json!({
"repo": q.repo, "from": q.from, "to": q.to, "sha": q.sha,
});
let value = reader_answer(wctx, "Dwarf.CallPath", params).await?;
Ok(Response::new(pb::JsonResponse { json: value.to_string() }))
}
}
// -- Vector (semantic search over materialized embeddings) -------------------
struct VectorSvc {
// Only read in embedder-enabled builds; the lean build returns UNIMPLEMENTED.
#[cfg_attr(
not(any(feature = "embed-tract", feature = "embed-ort")),
allow(dead_code)
)]
state: Shared,
}
#[tonic::async_trait]
impl VectorSvcTrait for VectorSvc {
async fn search(
&self,
req: Request<pb::VectorSearchRequest>,
) -> Result<Response<pb::JsonResponse>, Status> {
#[cfg(not(any(feature = "embed-tract", feature = "embed-ort")))]
{
let _ = req;
Err(unimplemented_verb(
"nornir-server was built without an embedder; rebuild with \
`--features server,embed-tract` (CPU) or `--features server,embed-ort` (GPU) \
to use Vector.Search",
The file goes on; the console shows all of it.