crates/nornir-warehouse-embed/src/room.rs
//! **Code-room READ client — the THIN gRPC leg for `CodeRoom.Answer`.**
//!
//! Since Mímir v2 the code graph is nornir's CODE ROOM, asked through one door: a
//! reader surface by name (`Viz.CrateGraph`, `Knowledge.TestReach`,
//! `Coderoom.SnapshotDiff` …) and its request as JSON. This module speaks that door
//! and turns EVERY outcome into a [`RoomAnswer`] — an answer with a named state:
//!
//! | state | who says it | means |
//! |---|---|---|
//! | `complete` · `partial` · `empty_fact` | the reader | computed; `partial` = an input is missing, a lower bound |
//! | `empty_gap` | the reader | empty BECAUSE an input is missing — not a finding |
//! | `not_wired` · `not_consumed` · `no_room` · `lake_unavailable` · `driver_not_available` · `not_linked` · `invalid_argument` · `internal` | the reader | a refusal, with its sentence |
//! | `unreachable` · `auth_failed` · `not_served` · `door_not_served` · `rpc_failed` | THIS client | the question never reached a reader |
//! | `queued` · `running` | the server | a whole-room pane is computed on the server's jera queue: the body names the job; [`ask`] asks again until the answer is there |
//! | `unstated` | this client | the server answered without a state this client knows |
//!
//! There is no `Result` on [`ask`] on purpose: a pane draws the state, and a failure
//! that is an `Err` in one caller and an empty list in the next is how "the server is
//! down" used to draw as "no findings". Only the first three states are a measurement
//! ([`RoomAnswer::is_measured`]); nothing else may be drawn as a number.
//!
//! The answer also says WHICH room it was read from ([`RoomAnswer::room_line`]):
//! workspace, repo@seq with the commit, node and edge counts — the line a pane prints
//! so a person can ask the server the same question and compare.
use crate::{norm_endpoint, pb, status_to_read_error, WarehouseReadError};
/// What a code-room answer is, by its state.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum RoomState {
/// Computed, every input present.
Complete,
/// Computed with an input missing: every figure is a lower bound.
Partial,
/// Computed, every input present, nothing found. A fact.
EmptyFact,
/// Empty BECAUSE an input is missing. Not a finding.
EmptyGap,
/// The reader refused (its own state word is in [`RoomAnswer::state`]).
Refused,
/// The question never reached a reader (this client's word is in `state`).
NotAsked,
/// The server put the answer on its jera queue (`queued` / `running`); the
/// note names the job. Not a measurement: ask again.
Queued,
/// The server answered without a state this client knows.
Unstated,
}
/// The reader's refusal states (nornir `code_room::State`).
/// Every word of nornir's `code_room::State::ALL` — a word missing here was read as
/// `unstated` ("a state this client knows not"): `no_artifact` was, on dwarf
/// 2026-10-16, for every `Viz.Binary` answer.
pub const REFUSAL_STATES: [&str; 11] = [
"not_wired",
"not_linked",
"no_room",
"lake_unavailable",
"not_consumed",
"driver_not_available",
"no_artifact",
"invalid_argument",
"internal",
"no_embedder",
"vectors_missing",
];
/// One answer of the code room, with where it was asked.
#[derive(Debug, Clone, PartialEq)]
pub struct RoomAnswer {
/// The surface asked (`Viz.CrateGraph`).
pub surface: String,
/// The state word — the reader's own, or this client's for a question that never
/// reached a reader.
pub state: String,
pub kind: RoomState,
/// The reader's JSON object (`Null` when no reader answered).
pub body: serde_json::Value,
/// A refusal's sentence, or what failed on the way. Empty for a computed answer.
pub note: String,
/// Where it was asked: `CodeRoom.Answer(<surface>) @ <endpoint> / <workspace>`.
pub via: String,
}
impl RoomAnswer {
/// The answer a reader's body is.
pub fn from_body(surface: &str, via: &str, body: serde_json::Value) -> Self {
let word = body
.get("state")
.and_then(|s| s.as_str())
.unwrap_or("")
.to_string();
let kind = match word.as_str() {
"complete" => RoomState::Complete,
"partial" => RoomState::Partial,
"empty_fact" => RoomState::EmptyFact,
"empty_gap" => RoomState::EmptyGap,
w if REFUSAL_STATES.contains(&w) => RoomState::Refused,
"queued" | "running" => RoomState::Queued,
_ => RoomState::Unstated,
};
let note = match kind {
RoomState::Refused | RoomState::Queued => body.get("note").and_then(|n| n.as_str()).unwrap_or("").to_string(),
RoomState::Unstated => format!(
"the server answered `{surface}` without a state this client knows (`{word}`): nothing in it is read as a measurement"
),
_ => String::new(),
};
let state = if kind == RoomState::Unstated {
"unstated".to_string()
} else {
word
};
Self {
surface: surface.to_string(),
state,
kind,
body,
note,
via: via.to_string(),
}
}
/// The answer a question that never reached a reader is.
pub fn not_asked(surface: &str, via: &str, err: &WarehouseReadError) -> Self {
let (state, note) = match err {
WarehouseReadError::Unreachable(m) => ("unreachable", format!("the server did not answer: {m}")),
WarehouseReadError::AuthFailed(m) => ("auth_failed", format!("the server refused the credentials: {m}")),
WarehouseReadError::NotServed(m) => ("not_served", format!("the server does not serve this: {m}")),
WarehouseReadError::Unsupported(m) => (
"door_not_served",
format!("the server has no `CodeRoom.Answer` verb (a nornir or a forge older than the code-room door): {m}"),
),
WarehouseReadError::Rpc(m) => ("rpc_failed", format!("the call failed: {m}")),
};
Self {
surface: surface.to_string(),
state: state.to_string(),
kind: RoomState::NotAsked,
body: serde_json::Value::Null,
note,
via: via.to_string(),
}
}
/// An answer that was never asked for (a pane before its first read).
pub fn pending(surface: &str, via: &str) -> Self {
Self {
surface: surface.to_string(),
state: "not_asked".to_string(),
kind: RoomState::NotAsked,
body: serde_json::Value::Null,
note: "not read yet".to_string(),
via: via.to_string(),
}
}
/// Did the reader COMPUTE — is what the body holds (also an empty list) a
/// measurement? Nothing else may be drawn as a number.
pub fn is_measured(&self) -> bool {
matches!(
self.kind,
RoomState::Complete | RoomState::Partial | RoomState::EmptyFact
)
}
/// Is the answer on the server's jera queue (ask again)?
pub fn is_queued(&self) -> bool {
self.kind == RoomState::Queued
}
/// The server's job id of a queued answer.
pub fn job_id(&self) -> Option<&str> {
self.body.get("job_id").and_then(|j| j.as_str())
}
/// The reader's rows under `key` when the answer is a measurement — an empty
/// slice otherwise (which the caller must not draw as "none": see the state).
pub fn rows(&self, key: &str) -> &[serde_json::Value] {
match (
self.is_measured(),
self.body.get(key).and_then(|v| v.as_array()),
) {
(true, Some(a)) => a,
_ => &[],
}
}
/// What the answer rests on, in the reader's own words: a refusal's sentence,
/// or the evidence's MISSING inputs and notes.
pub fn evidence_text(&self) -> String {
if !self.note.is_empty() {
return self.note.clone();
}
let ev = &self.body["evidence"];
let mut parts: Vec<String> = Vec::new();
for m in ev["missing"].as_array().map(Vec::as_slice).unwrap_or(&[]) {
let input = m["input"].as_str().unwrap_or("?");
let why = m["why"].as_str().unwrap_or("");
parts.push(match m["repo"].as_str() {
Some(r) => format!("MISSING {input} [{r}]: {why}"),
None => format!("MISSING {input}: {why}"),
});
}
if let (Some(shown), Some(total)) = (
ev["capped"]["shown"].as_u64(),
ev["capped"]["total"].as_u64(),
) {
parts.push(format!("capped: {shown} of {total} shown"));
}
for n in ev["notes"].as_array().map(Vec::as_slice).unwrap_or(&[]) {
if let Some(n) = n.as_str() {
parts.push(n.to_string());
}
}
parts.join(" · ")
}
/// The room the answer was read from, on one line — `workspace w · repo@seq
/// (sha) · N nodes · M edges` — or empty when the answer names none (a
/// refusal, a question that was never answered).
pub fn room_line(&self) -> String {
room_line_of(&self.body["room"])
}
/// The observable form a pane folds into its own `state_json`.
pub fn state_json(&self) -> serde_json::Value {
serde_json::json!({
"surface": self.surface,
"state": self.state,
"measured": self.is_measured(),
"evidence": self.evidence_text(),
"room": self.room_line(),
"via": self.via,
})
}
}
/// One `room` identity block (`RoomAnswer::body["room"]`, or a snapshot diff's
/// `room_from` / `room_to`) on one line.
pub fn room_line_of(room: &serde_json::Value) -> String {
let Some(ws) = room["workspace"].as_str() else {
return String::new();
};
let repos: Vec<String> = room["repos"]
.as_array()
.map(Vec::as_slice)
.unwrap_or(&[])
.iter()
.map(|r| {
let (repo, seq) = (
r["repo"].as_str().unwrap_or("?"),
r["seq"].as_i64().unwrap_or(0),
);
match r["sha"].as_str() {
Some(sha) => format!("{repo}@{seq} ({})", &sha[..sha.len().min(10)]),
None => format!("{repo}@{seq}"),
}
})
.collect();
format!(
"workspace {ws} · {} · {} nodes · {} edges{}",
if repos.is_empty() {
"no repo".to_string()
} else {
repos.join(", ")
},
room["nodes"].as_u64().unwrap_or(0),
room["edges_total"].as_u64().unwrap_or(0),
if room["universe"].as_bool() == Some(true) {
" · with the dependency ring"
} else {
""
},
)
}
/// The label a pane prints for where an answer was asked.
pub fn via(endpoint: &str, workspace: &str, surface: &str) -> String {
format!(
"CodeRoom.Answer({surface}) @ {} / {}",
norm_endpoint(endpoint),
if workspace.is_empty() {
"(no workspace)"
} else {
workspace
}
)
}
/// The intercepted `CodeRoom` client type (bearer + `nornir-workspace` header).
type RoomClient = pb::code_room_client::CodeRoomClient<
tonic::service::interceptor::InterceptedService<
tonic::transport::Channel,
crate::NornirInterceptor,
>,
>;
/// `CodeRoom.Answer` as the wire gives it: the reader's JSON object, or the typed
/// reason the question never reached a reader. Runs on a DEDICATED OS thread (a host
/// may call this from inside its own tokio runtime), 3 s connect timeout.
pub fn ask_raw(
endpoint: &str,
token: &str,
workspace: &str,
surface: &str,
params: &serde_json::Value,
) -> Result<serde_json::Value, WarehouseReadError> {
let endpoint = norm_endpoint(endpoint);
let (token, workspace) = (token.to_string(), workspace.to_string());
let question = pb::CodeRoomQuestion {
surface: surface.to_string(),
params_json: params.to_string(),
};
std::thread::scope(|s| {
s.spawn(|| {
let rt = tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()
.map_err(|e| {
WarehouseReadError::Unreachable(format!("build tokio runtime: {e}"))
})?;
rt.block_on(async move {
// The ONE shared client bearer-builder (Law #1). `None` ⇒ open.
let bearer = nornir_service::client::bearer(&token).map_err(|e| {
WarehouseReadError::AuthFailed(format!("bad bearer token: {e}"))
})?;
let ws_md: Option<tonic::metadata::MetadataValue<tonic::metadata::Ascii>> =
(!workspace.is_empty())
.then(|| workspace.parse().ok())
.flatten();
let channel = tonic::transport::Channel::from_shared(endpoint.clone())
.map_err(|e| {
WarehouseReadError::Unreachable(format!("invalid url `{endpoint}`: {e}"))
})?
.connect_timeout(std::time::Duration::from_secs(3))
// a room is folded from the log on its first ask: minutes on a big one
.timeout(crate::SCAN_TIMEOUT)
.connect()
.await
.map_err(|e| {
WarehouseReadError::Unreachable(format!("connect to {endpoint}: {e}"))
})?;
let interceptor: crate::NornirInterceptor =
Box::new(move |mut req: tonic::Request<()>| {
if let Some(b) = &bearer {
req.metadata_mut().insert("authorization", b.clone());
}
if let Some(ws) = &ws_md {
req.metadata_mut().insert("nornir-workspace", ws.clone());
}
Ok(req)
});
let mut client: RoomClient =
pb::code_room_client::CodeRoomClient::with_interceptor(channel, interceptor)
.max_decoding_message_size(64 << 20);
let resp = client
.answer(question)
.await
.map_err(status_to_read_error)?
.into_inner();
let body: serde_json::Value = serde_json::from_str(&resp.json).map_err(|e| {
WarehouseReadError::Rpc(format!("CodeRoom.Answer body is not JSON: {e}"))
})?;
if !body.is_object() {
return Err(WarehouseReadError::Rpc(
"CodeRoom.Answer answered Ok with a body that is not a JSON object".into(),
));
}
Ok(body)
})
})
.join()
.map_err(|_| WarehouseReadError::Rpc("code-room read thread panicked".into()))?
})
}
/// How long [`ask`] waits between two asks of an answer that is on the server's
/// jera queue. Every ask is ONE short call; nothing waits on a call for the work.
pub const REASK: std::time::Duration = std::time::Duration::from_millis(1500);
/// How long [`ask`] keeps asking again before it hands the `queued` / `running`
/// answer back as it is (a pane then shows the job and may ask later).
pub const QUEUED_PATIENCE: std::time::Duration = std::time::Duration::from_secs(30 * 60);
/// The answers this process is waiting for on a server's jera queue:
/// `<workspace> · <surface>` → the server's note ("running on jera — job … (running 12s)").
static WAITING: std::sync::Mutex<std::collections::BTreeMap<String, String>> =
std::sync::Mutex::new(std::collections::BTreeMap::new());
/// What this process waits for on the server's jera queue right now, one line
/// per answer (`Viz.CrateGraph: running on jera — job 5f… (running 12s)`), for a
/// pane's "asking…" line.
pub fn waiting() -> Vec<String> {
WAITING
.lock()
.unwrap_or_else(|e| e.into_inner())
.iter()
.map(|(k, v)| format!("{}: {v}", k.rsplit(" · ").next().unwrap_or(k)))
.collect()
}
/// **Ask the code room ONCE.** Total: every outcome is a [`RoomAnswer`] with a
/// state — also `queued` / `running` ([`RoomState::Queued`]).
pub fn ask_once(
endpoint: &str,
token: &str,
workspace: &str,
surface: &str,
params: &serde_json::Value,
) -> RoomAnswer {
let via = via(endpoint, workspace, surface);
match ask_raw(endpoint, token, workspace, surface, params) {
Ok(body) => RoomAnswer::from_body(surface, &via, body),
Err(e) => RoomAnswer::not_asked(surface, &via, &e),
}
}
/// **Ask the code room.** Total: every outcome is a [`RoomAnswer`] with a state.
/// An answer the server put on its jera queue is asked for again every [`REASK`]
/// (each a short call; [`waiting`] tells a pane which job it waits for) until it
/// is there, or [`QUEUED_PATIENCE`] has passed. Call it off the UI thread.
pub fn ask(
endpoint: &str,
token: &str,
workspace: &str,
surface: &str,
params: &serde_json::Value,
) -> RoomAnswer {
let slot = format!("{workspace} · {surface}");
let t0 = std::time::Instant::now();
loop {
let a = ask_once(endpoint, token, workspace, surface, params);
if !a.is_queued() || t0.elapsed() >= QUEUED_PATIENCE {
WAITING
.lock()
.unwrap_or_else(|e| e.into_inner())
.remove(&slot);
return a;
}
WAITING
.lock()
.unwrap_or_else(|e| e.into_inner())
.insert(slot.clone(), a.note.clone());
std::thread::sleep(REASK);
}
}
#[cfg(test)]
mod tests {
use super::*;
use serde_json::json;
fn body(state: &str) -> serde_json::Value {
json!({
"state": state,
"crates": [{ "id": "a@1" }],
"evidence": {
"missing": [{ "input": "binary stream", "repo": "a", "why": "consumed without the binary layer" }],
"notes": ["a lower bound"],
"capped": { "shown": 2, "total": 7 },
},
"room": { "workspace": "w", "repos": [{ "repo": "a", "seq": 3, "sha": "0123456789abcdef" }], "nodes": 41, "edges_total": 97, "universe": false },
})
}
/// Only the three computed states are a measurement; a refusal and an answer
/// with no state hand out NO rows, whatever the body holds.
#[test]
fn only_a_computed_answer_hands_out_rows() {
for (word, measured) in [
("complete", true),
("partial", true),
("empty_fact", true),
("empty_gap", false),
("not_wired", false),
("not_consumed", false),
("computed", false),
("", false),
] {
let a = RoomAnswer::from_body("Viz.CrateGraph", "via", body(word));
assert_eq!(a.is_measured(), measured, "`{word}`");
assert_eq!(
a.rows("crates").len(),
measured as usize,
"`{word}`: rows only from a measurement"
);
}
let unknown = RoomAnswer::from_body("Viz.CrateGraph", "via", body("computed"));
assert_eq!(
(unknown.kind, unknown.state.as_str()),
(RoomState::Unstated, "unstated")
);
assert!(unknown.note.contains("`computed`"), "{}", unknown.note);
}
/// Every refusal nornir's reader can send is a refusal here, with its sentence —
/// never `unstated`. (`no_artifact` reached dwarf's Binary tab as "a state this
/// client knows not".) The list is nornir's `code_room::State::ALL`, by word.
#[test]
fn every_refusal_of_the_reader_is_known() {
for word in [
"not_wired",
"not_linked",
"no_room",
"lake_unavailable",
"not_consumed",
"driver_not_available",
"no_artifact",
"invalid_argument",
"internal",
"no_embedder",
"vectors_missing",
] {
let a = RoomAnswer::from_body(
"Viz.Binary",
"via",
json!({ "state": word, "refused": true, "note": "no linked artifact of this commit — looked in /x" }),
);
assert_eq!(
(a.kind, a.state.as_str()),
(RoomState::Refused, word),
"`{word}`"
);
assert_eq!(
a.evidence_text(),
"no linked artifact of this commit — looked in /x",
"`{word}`"
);
}
}
/// A refusal's sentence is the evidence; a computed answer's evidence is what
/// is MISSING, the cap and the notes — in the reader's words.
#[test]
fn the_evidence_text_is_the_readers_own_words() {
let partial = RoomAnswer::from_body("Viz.CrateGraph", "via", body("partial"));
assert_eq!(partial.evidence_text(), "MISSING binary stream [a]: consumed without the binary layer · capped: 2 of 7 shown · a lower bound");
assert_eq!(
partial.room_line(),
"workspace w · a@3 (0123456789) · 41 nodes · 97 edges"
);
let refused = RoomAnswer::from_body(
"Coderoom.Compat",
"via",
json!({ "state": "not_wired", "refused": true, "note": "the world ring has no producer" }),
);
assert_eq!(
(
refused.kind,
refused.evidence_text().as_str(),
refused.room_line().as_str()
),
(RoomState::Refused, "the world ring has no producer", "")
);
let s = refused.state_json();
assert_eq!(
(s["state"].as_str(), s["measured"].as_bool()),
(Some("not_wired"), Some(false))
);
}
/// An answer on the server's jera queue is its own state: not a
/// measurement, no rows, the server's note (naming the job) as evidence.
#[test]
fn a_queued_answer_is_no_measurement_and_names_its_job() {
for word in ["queued", "running"] {
let mut b = body(word);
b["job_id"] = json!("j-7");
b["note"] = json!("running on jera — job j-7 (running 3s); ask again for the answer");
let a = RoomAnswer::from_body("Viz.CrateGraph", "via", b);
assert_eq!(
(a.kind, a.is_queued(), a.is_measured(), a.job_id()),
(RoomState::Queued, true, false, Some("j-7")),
"`{word}`"
);
assert!(a.rows("crates").is_empty());
assert!(
a.evidence_text().contains("job j-7"),
"{}",
a.evidence_text()
);
}
}
/// A question that never reached a reader has a state of its own — never an
/// empty answer, and never one of the reader's words.
#[test]
fn a_question_that_never_reached_a_reader_is_named() {
let dead = ask("http://127.0.0.1:1", "", "w", "Viz.CrateGraph", &json!({}));
assert_eq!(
(dead.kind, dead.state.as_str()),
(RoomState::NotAsked, "unreachable")
);
assert!(!dead.is_measured() && dead.rows("crates").is_empty());
assert!(
dead.evidence_text().contains("did not answer"),
"{}",
dead.evidence_text()
);
assert!(
dead.via.contains("Viz.CrateGraph")
&& dead.via.contains("127.0.0.1:1")
&& dead.via.contains("/ w"),
"{}",
dead.via
);
let old = RoomAnswer::not_asked(
"Viz.CrateGraph",
"via",
&WarehouseReadError::Unsupported("Unimplemented".into()),
);
assert_eq!(old.state, "door_not_served");
assert!(!REFUSAL_STATES.contains(&old.state.as_str()));
}
}