dwarves-gui/src/dataview.rs
//! **The 🏢 Data view tab — the warehouse DASHBOARD, back under Mímir.**
//!
//! There used to be a facett component that showed all of nornir/Mímir's data at once
//! in lots of graphs. It is [`facett_warehousedeck::WarehouseDeck`] — *a multi-pane
//! dashboard over a lakehouse warehouse: a FacetDeck WALL of N panes, each binding one
//! table to a view-kind (3D graph / grid / chart)*. This tab mounts it over the SELECTED
//! workspace's Mímir/warehouse tables, so "all of Mímir's data in lots of graphs" is
//! back in dwarves.
//!
//! ## The deck (reuse, don't rebuild)
//! The deck is **engine-agnostic** — it takes Arrow `RecordBatch`es, not a catalog. This
//! tab reads nornir's warehouse over the SAME thin `Warehouse.Scan` gRPC read seam the
//! Binary / Coverage / Lenses tabs use (`nornir_warehouse_embed::scan_warehouse_table`),
//! turns each table's stringified reply into an all-Utf8 `RecordBatch`, and binds it to a
//! view-kind through [`facett_warehousedeck::WarehouseRegistry::default_nordisk`] (the
//! table → view-kind map the deck already ships): edge tables (`call_edges`,
//! `dep_graph_edges`) become **3D graphs**, time-series (`git_heat_facts`,
//! `release_events`) become **charts**, everything else a **grid**. It never re-implements
//! a viewer and never links `nornir` (no cycle) — a clean one-way edge, exactly like the
//! other warehouse-fed tabs.
//!
//! ## The code graph's two panes are the room's, not tables
//! The symbol grid and the call graph used to be the `symbol_facts` / `call_edges`
//! tables. Since Mímir v2 they are nornir's code room, asked through the forge
//! ([`crate::room`]): `Knowledge.Symbols` binds to a grid (`code room: symbols`) and
//! `Viz.CallGraph` — the heaviest call edges between the workspace's own fns — to a 3D
//! graph (`code room: call graph`) whose nodes carry the room's ids, so a click asks
//! the 📄 source pane's question by identity. Each answer's state, evidence and the room
//! it was read from are printed above the wall; an answer that is not a measurement
//! gets no pane.
//!
//! ## The DataFusion SQL prompt
//! A text box runs SQL over the warehouse tables. **Where it runs:** honestly,
//! server-side is the ideal (nornir owns the one live catalog and already has the
//! engine — `nornir warehouse query`, `warehouse::sql::query`), but there is no
//! `Warehouse.Sql` gRPC verb yet and adding one is a nornir-repo change. So this runs
//! **client-side over the frames fetched thin** — the fallback the brief allows. It uses
//! the SAME DataFusion engine skade standardizes on and re-exports (`skade::datafusion`,
//! v54 / arrow-58): the thin-fetched tables are registered as in-memory `MemTable`s in a
//! [`SessionContext`], and `SessionContext::sql` runs over them. We do NOT open nornir's
//! Iceberg catalog (the CATALOG DOCTRINE forbids a second redb lock), so skade's
//! server-side `IcebergStaticTableProvider` is not reachable here — the frames are the
//! table. Behind the `dataview-sql` feature (native, off by default like `dupes-fat`):
//! OFF ⇒ the prompt says the engine is not built in and names the feature; ON ⇒ real SQL.
//!
//! Because the thin scan is **stringified**, the registered columns are all `Utf8`; a
//! query that aggregates numerically must `CAST(... AS BIGINT)`. `SELECT repo, count(*)
//! FROM symbol_facts GROUP BY repo` works as-is. A bad query surfaces DataFusion's error
//! verbatim — the prompt shows it, it never crashes.
//!
//! The result is rendered two ways: as a browsable grid (the reused
//! `facett_icebergview::IcebergView`) AND fed back into the deck as a new pane, so a
//! query's output joins the wall.
use std::collections::BTreeSet;
use std::sync::Arc;
use arrow_array::{ArrayRef, RecordBatch, StringArray};
use arrow_schema::{DataType, Field, Schema};
use facett_core::Facet;
use facett_icebergview::{IcebergView, Snapshot, SnapshotMeta};
use facett_warehousedeck::{
hash_color, ChartSpec, GraphSpec, PaneData, ViewKind, WarehouseDeck, WarehouseRegistry,
};
use nornir_warehouse_embed::{scan_warehouse_table, WarehouseScan};
use crate::graph_source::{NodeKind, NodeRef};
use crate::room::{ForgeRoom, RoomAnswer, RoomSource};
/// Row cap per table on the thin read for the deck panes — enough to visualize each
/// table, thin enough not to haul a constellation-scale table over a stringified scan.
const DECK_SCAN_LIMIT: u32 = 2_000;
/// Points reserved below the wall for the SQL prompt + its status line, so the deck/source
/// split never eats the frame the prompt lives in.
const SQL_PROMPT_HEIGHT: f32 = 220.0;
/// Floor for the wall+source split's height — below this the 3D graphs are unreadable and
/// the source pane cannot show a function, so the tab scrolls instead of shrinking further.
const MIN_WALL_HEIGHT: f32 = 320.0;
/// Seconds between warehouse-generation probes — see `crate::generation`.
const GENERATION_PROBE_S: f64 = 4.0;
/// The Mímir/warehouse tables the Data view binds into deck panes, in wall order. A
/// curated slice of the ~58-table warehouse — the ones that answer "what is in here":
/// git heat, coverage, benches, tests, embeddings count, the dependency graph, and the
/// release timeline. (The symbol grid and the call graph are nornir's code room, not
/// tables: [`ROOM_SYMBOLS`] / [`ROOM_CALLS`].) Each is bound to a view-kind via
/// [`WarehouseRegistry::default_nordisk`]; a table the server does not serve is recorded
/// as a note and skipped (never a fabricated empty pane).
pub const KEY_TABLES: &[&str] = &[
"git_heat_facts",
"coverage",
"coverage_fn",
"bench_results",
"test_outcomes",
"dep_graph_edges",
"release_events",
"embeddings",
];
/// The surface behind the `code room: symbols` grid.
pub const S_SYMBOLS: &str = "Knowledge.Symbols";
/// The surface behind the `code room: call graph` pane.
pub const S_CALLS: &str = "Viz.CallGraph";
/// The wall title (and SQL table name `room_symbols`) of the symbol grid.
pub const ROOM_SYMBOLS: &str = "code room: symbols";
/// The wall title of the call-graph pane (its edge rows are the SQL table `room_calls`).
pub const ROOM_CALLS: &str = "code room: call graph";
/// Call edges asked for: the heaviest ones, as many as a 3D pane stays readable with.
const CALL_GRAPH_EDGES: u32 = 400;
/// The SQL prompt's starting query — over a table that is still written.
pub const DEFAULT_SQL: &str = "SELECT repo, count(*) FROM git_heat_facts GROUP BY repo";
/// Edge-table → `(source_column, target_column)` for the 3D-graph panes. Only a table
/// listed here (with both columns present in the scan) becomes a graph; anything else
/// falls back to a grid rather than inventing edges.
fn graph_cols(table: &str) -> Option<(&'static str, &'static str)> {
match table {
// MEASURED 2026-08-25 against `nornir/src/warehouse/iceberg_schema.rs:717`:
// `dep_graph_edges` has `from_repo`/`to_repo`. The `from_crate`/`to_crate` this
// map used to name are columns that do not exist, so `scan_to_graph` never found
// them and the dependency pane silently fell back to a GRID — the honest fallback
// doing its job over a typo. There is no `scip_call_edges` TABLE at all (no schema
// function declares one; it is a DERIVED join in `nornir::knowledge::scip`), so the
// entry for it was likewise unreachable and is gone rather than left to look wired.
"dep_graph_edges" => Some(("from_repo", "to_repo")),
_ => None,
}
}
/// Time-series table → the numeric `y` column for the chart panes. Only a mapped column
/// that actually parses as a number for enough rows becomes a chart; otherwise the table
/// falls back to a grid.
fn chart_col(table: &str) -> Option<&'static str> {
match table {
"git_heat_facts" => Some("commits_90d"),
"release_events" => Some("ts_micros"),
"bench_results" => Some("value"),
_ => None,
}
}
/// Where the Data view's warehouse rows come from. Injected so the robot test drives the
/// tab over a deterministic fixture with no server (the `LensSource` pattern).
///
/// `Send + Sync` because the pane runs the scans on a WORKER thread — eight gRPC round
/// trips must never happen on the frame that is trying to draw. The `Arc` the pane
/// holds is what the worker clones.
pub trait DataViewSource: Send + Sync {
/// Scan one table — the stringified `columns` + `rows` of `Warehouse.Scan`. `Err` is
/// an honest typed failure (unreachable / auth / not-served), never a silent empty.
fn scan(&self, table: &str) -> Result<WarehouseScan, String>;
/// A human description of the source (endpoint + workspace, or "fixture").
fn describe(&self) -> String;
}
/// The shipped source: nornir's warehouse over the thin `Warehouse.Scan` gRPC read seam.
pub struct WarehouseDataSource {
pub endpoint: String,
pub token: String,
pub workspace: String,
}
impl WarehouseDataSource {
pub fn new(
endpoint: impl Into<String>,
token: impl Into<String>,
workspace: impl Into<String>,
) -> Self {
Self {
endpoint: endpoint.into(),
token: token.into(),
workspace: workspace.into(),
}
}
}
impl DataViewSource for WarehouseDataSource {
fn scan(&self, table: &str) -> Result<WarehouseScan, String> {
scan_warehouse_table(
&self.endpoint,
&self.token,
&self.workspace,
table,
DECK_SCAN_LIMIT,
)
.map_err(|e| e.to_string())
}
fn describe(&self) -> String {
format!("nornir warehouse @ {} / {}", self.endpoint, self.workspace)
}
}
/// Build an all-`Utf8` Arrow `RecordBatch` from a stringified `Warehouse.Scan`. The thin
/// seam renders every cell to a display string, so every column is `Utf8`; numeric SQL
/// over the result must `CAST`. An empty scan yields a schema-only (0-row) batch.
fn scan_to_batch(scan: &WarehouseScan) -> RecordBatch {
let cols: Vec<String> = if scan.columns.is_empty() {
vec!["value".to_string()]
} else {
scan.columns.clone()
};
let columns: Vec<ArrayRef> = (0..cols.len())
.map(|c| {
let cells = scan
.rows
.iter()
.map(|r| r.get(c).map(String::as_str).unwrap_or(""));
Arc::new(StringArray::from_iter_values(cells)) as ArrayRef
})
.collect();
let schema = Arc::new(Schema::new(
cols.iter()
.map(|c| Field::new(c, DataType::Utf8, false))
.collect::<Vec<_>>(),
));
RecordBatch::try_new(schema.clone(), columns).unwrap_or_else(|_| RecordBatch::new_empty(schema))
}
/// The distinct values of column `col` in scan order → a stable node index, plus the
/// per-row `(src_idx, dst_idx)` edges. Caps the node/edge counts so a huge table does not
/// spawn an unusable 3D graph.
fn scan_to_graph(scan: &WarehouseScan, src: &str, dst: &str) -> Option<GraphSpec> {
let si = scan.columns.iter().position(|c| c == src)?;
let di = scan.columns.iter().position(|c| c == dst)?;
const MAX_NODES: usize = 160;
const MAX_EDGES: usize = 400;
let mut labels: Vec<String> = Vec::new();
let mut seen: BTreeSet<String> = BTreeSet::new();
let idx = |labels: &mut Vec<String>, seen: &mut BTreeSet<String>, v: &str| -> Option<usize> {
if let Some(p) = labels.iter().position(|l| l == v) {
return Some(p);
}
if labels.len() >= MAX_NODES {
return None;
}
seen.insert(v.to_string());
labels.push(v.to_string());
Some(labels.len() - 1)
};
let mut edges: Vec<(usize, usize)> = Vec::new();
for r in &scan.rows {
if edges.len() >= MAX_EDGES {
break;
}
let (Some(a), Some(b)) = (r.get(si), r.get(di)) else {
continue;
};
if a.is_empty() || b.is_empty() {
continue;
}
let (Some(ai), Some(bi)) = (
idx(&mut labels, &mut seen, a),
idx(&mut labels, &mut seen, b),
) else {
continue;
};
edges.push((ai, bi));
}
if edges.is_empty() {
return None;
}
let nodes = labels.iter().map(|l| (l.clone(), hash_color(l))).collect();
Some(GraphSpec::new(nodes, edges))
}
/// A single line series `(row_index, value)` from the numeric `y` column. `None` if the
/// column is absent or fewer than two rows parse as numbers (a chart of one point is a
/// dot — a grid is more honest).
fn scan_to_chart(scan: &WarehouseScan, y: &str) -> Option<ChartSpec> {
let yi = scan.columns.iter().position(|c| c == y)?;
let pts: Vec<(f64, f64)> = scan
.rows
.iter()
.enumerate()
.filter_map(|(i, r)| {
r.get(yi)
.and_then(|s| s.parse::<f64>().ok())
.map(|v| (i as f64, v))
})
.collect();
if pts.len() < 2 {
return None;
}
Some(ChartSpec::line(vec![(y.to_string(), pts)]))
}
/// The reader's rows under `key` as a stringified scan with the named columns — the
/// shape the grid and the SQL prompt take. Every cell is the reader's value.
fn rows_to_scan(rows: &[serde_json::Value], cols: &[&str]) -> WarehouseScan {
let cell = |v: &serde_json::Value| match v {
serde_json::Value::Null => String::new(),
serde_json::Value::String(s) => s.clone(),
other => other.to_string(),
};
WarehouseScan {
columns: cols.iter().map(|c| c.to_string()).collect(),
rows: rows
.iter()
.map(|r| cols.iter().map(|c| cell(&r[*c])).collect())
.collect(),
}
}
/// The symbol rows of a `Knowledge.Symbols` answer as a grid scan.
fn symbols_scan(a: &RoomAnswer) -> WarehouseScan {
rows_to_scan(
a.rows("symbols"),
&[
"repo",
"crate_name",
"module_path",
"item_kind",
"item_name",
"visibility",
"file",
"line",
"end_line",
"path",
"moniker",
],
)
}
/// [`call_graph_of`]'s answer: the graph, each node's label and room id, and the
/// edge rows as a scan.
type CallGraph = (GraphSpec, Vec<String>, Vec<Option<String>>, WarehouseScan);
/// A `Viz.CallGraph` answer as a 3D graph: the reader's nodes in ITS order (an edge's
/// `from` / `to` are indexes into them), coloured by crate, plus the label and the
/// room's id of each node — and the edge rows as a scan for the SQL prompt. `None`
/// when the answer names no edge.
fn call_graph_of(a: &RoomAnswer) -> Option<CallGraph> {
let nodes = a.rows("nodes");
let labels: Vec<String> = nodes
.iter()
.map(|n| n["id"].as_str().unwrap_or("?").to_string())
.collect();
let monikers: Vec<Option<String>> = nodes
.iter()
.map(|n| n["moniker"].as_str().map(str::to_string))
.collect();
let mut edges: Vec<(usize, usize)> = Vec::new();
let mut rows: Vec<Vec<String>> = Vec::new();
for e in a.rows("edges") {
let (Some(f), Some(t)) = (
e["from"].as_u64().map(|x| x as usize),
e["to"].as_u64().map(|x| x as usize),
) else {
continue;
};
let (Some(from), Some(to)) = (labels.get(f), labels.get(t)) else {
continue;
};
edges.push((f, t));
rows.push(vec![
from.clone(),
to.clone(),
e["sites"].as_u64().unwrap_or(0).to_string(),
]);
}
if edges.is_empty() {
return None;
}
let spec = GraphSpec::new(
nodes
.iter()
.zip(&labels)
.map(|(n, l)| (l.clone(), hash_color(n["crate"].as_str().unwrap_or(""))))
.collect(),
edges,
);
let scan = WarehouseScan {
columns: vec!["caller".into(), "callee".into(), "sites".into()],
rows,
};
Some((spec, labels, monikers, scan))
}
/// One read of everything the wall binds.
pub struct WallRead {
scans: Vec<(&'static str, Result<WarehouseScan, String>)>,
/// The room's answers (`None` = this pane was given no room to ask).
symbols: Option<RoomAnswer>,
calls: Option<RoomAnswer>,
}
/// Milliseconds since the epoch (the snapshot timestamp the result grid stamps).
fn now_ms() -> i64 {
std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map(|d| d.as_millis() as i64)
.unwrap_or(0)
}
/// **Run SQL over the thin-fetched frames with DataFusion** (feature `dataview-sql`).
///
/// Registers each `(name, batch)` as an in-memory `MemTable` in a fresh
/// `SessionContext` and runs `ctx.sql(sql)`, returning the result `RecordBatch`es. This
/// is the SAME DataFusion engine skade re-exports (`skade::datafusion`, v54); the thin
/// client feeds it the frames rather than nornir's Iceberg catalog (the catalog
/// doctrine). A parse/plan/exec error comes back as its `Err(String)` — the prompt shows
/// it, never a crash.
#[cfg(feature = "dataview-sql")]
pub fn run_sql(frames: &[(String, RecordBatch)], sql: &str) -> Result<Vec<RecordBatch>, String> {
use datafusion::datasource::MemTable;
use datafusion::prelude::SessionContext;
let rt = tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()
.map_err(|e| format!("build tokio runtime for DataFusion: {e}"))?;
rt.block_on(async move {
let ctx = SessionContext::new();
for (name, batch) in frames {
let mem = MemTable::try_new(batch.schema(), vec![vec![batch.clone()]])
.map_err(|e| format!("register `{name}`: {e}"))?;
ctx.register_table(name.as_str(), Arc::new(mem))
.map_err(|e| format!("register `{name}`: {e}"))?;
}
let df = ctx.sql(sql).await.map_err(|e| e.to_string())?;
df.collect().await.map_err(|e| e.to_string())
})
}
/// The `dataview-sql`-OFF stub: the engine is not built in. Honest — it names the feature
/// to rebuild with rather than silently returning no rows.
#[cfg(not(feature = "dataview-sql"))]
pub fn run_sql(_frames: &[(String, RecordBatch)], _sql: &str) -> Result<Vec<RecordBatch>, String> {
Err(
"SQL engine not built in — rebuild with `--features dataview-sql` (DataFusion, \
the engine skade re-exports). The prompt then runs SQL over the thin-fetched \
warehouse frames."
.to_string(),
)
}
/// True when this build has the DataFusion SQL engine compiled in.
pub const fn sql_enabled() -> bool {
cfg!(feature = "dataview-sql")
}
/// The **🏢 Data view** pane: the warehouse deck wall + the DataFusion SQL prompt.
pub struct DataViewPane {
title: String,
/// The read seam. `Arc` so the off-thread worker clones it — the SAME shape the
/// Coverage / Dupes / Lenses panes hold theirs in.
source: Arc<dyn DataViewSource>,
/// A read in flight (worker receiver + when it started). The frame polls it; the
/// robot seam blocks on it. The eight `KEY_TABLES` scans run OFF the egui thread.
pending: Option<(std::sync::mpsc::Receiver<WallRead>, std::time::Instant)>,
/// Where the code graph's two panes ask (`None` ⇒ an injected table source with no
/// room: the wall then has no code-room pane and says nothing about one).
room: Option<Arc<dyn RoomSource>>,
/// The room's last answers, drawn as callouts above the wall.
room_symbols: Option<RoomAnswer>,
room_calls: Option<RoomAnswer>,
/// The composed deck, built on Scan (`None` until then).
deck: Option<WarehouseDeck>,
/// The frames the deck bound, `(table, batch)` — the tables the SQL prompt registers.
frames: Vec<(String, RecordBatch)>,
/// Per-table scan notes ("table: <error>") — the evidence for a table that did not
/// bind, so a missing pane reads as a named absence, not a silent drop.
notes: Vec<String>,
/// A fatal error (every table failed — the server is unreachable, not empty).
error: Option<String>,
/// The SQL text the user typed.
sql_input: String,
/// The last SQL run's status line.
sql_status: String,
/// The last SQL error (parse/plan/exec/engine-off), if any.
sql_error: Option<String>,
/// The last SQL result grid (reused `facett_icebergview::IcebergView`).
sql_result: Option<IcebergView>,
/// The last SQL result's column names + row count (the robot oracle).
sql_columns: Vec<String>,
sql_rows: usize,
/// The GENERATION this pane last read at, and the background probe that moves it.
/// The deck rebinds when the warehouse's snapshot id changes — never on a press.
/// `gen` is `None` for a fixture/injected source (no warehouse to probe): one cold
/// read and then silence (LAW 5 / `crate::generation`).
watch: crate::generation::Watch,
gen: Option<crate::generation::WarehouseGeneration>,
/// The 📄 right-hand SOURCE pane — the consumer of a graph-node click.
source_pane: crate::graph_source::GraphSourcePane,
/// Every graph pane on the wall: `(pane title, what its labels MEAN, the labels in
/// `Graph3D` node order, the selection last seen)`. The label vector is the deck's own
/// node order because `WarehouseDeck::add` hands `GraphSpec::nodes` straight to
/// `Graph3D::new`, so index `i` here IS node `i` there.
graph_panes: Vec<GraphPaneLabels>,
}
/// One graph pane's click seam — see [`DataViewPane::graph_panes`].
struct GraphPaneLabels {
title: String,
kind: NodeKind,
labels: Vec<String>,
/// The room's id of each node, in node order (empty for a table's graph).
monikers: Vec<Option<String>>,
/// The selection this pane reported on the previous frame. A CHANGE is the click;
/// re-reading the same index every frame is not, and two graph panes each holding a
/// selection must not fight over the source pane once per frame.
last: Option<usize>,
}
impl DataViewPane {
/// The live tab, bound to the selected workspace over the thin gRPC read seam.
pub fn live(
endpoint: impl Into<String>,
token: impl Into<String>,
workspace: impl Into<String>,
) -> Self {
let (endpoint, token, workspace) = (endpoint.into(), token.into(), workspace.into());
let mut me = Self::with_source(Box::new(WarehouseDataSource::new(
endpoint.clone(),
token.clone(),
workspace.clone(),
)));
me = me.with_room(ForgeRoom::shared(&endpoint, &token, &workspace));
me.gen = Some(crate::generation::WarehouseGeneration::new(
endpoint,
token,
workspace,
GENERATION_PROBE_S,
));
me
}
/// Give the pane a code room to ask — the wall then carries the room's symbol grid
/// and call graph, and the 📄 source pane asks it for a clicked item.
pub fn with_room(mut self, room: Arc<dyn RoomSource>) -> Self {
self.source_pane = crate::graph_source::GraphSourcePane::new().with_room(Arc::clone(&room));
self.room = Some(room);
self
}
/// The pane over an injected source (the robot-test seam). Taken as a `Box` (the
/// caller's spelling is unchanged) and held as an `Arc`, which is what the read
/// worker clones.
pub fn with_source(source: Box<dyn DataViewSource>) -> Self {
Self {
title: "Data view".to_string(),
source: source.into(),
pending: None,
room: None,
room_symbols: None,
room_calls: None,
deck: None,
frames: Vec::new(),
notes: Vec::new(),
error: None,
sql_input: DEFAULT_SQL.to_string(),
sql_status: String::new(),
sql_error: None,
sql_result: None,
sql_columns: Vec::new(),
sql_rows: 0,
watch: crate::generation::Watch::new(0.0),
gen: None,
source_pane: crate::graph_source::GraphSourcePane::new(),
graph_panes: Vec::new(),
}
}
/// **Scan the warehouse and build the deck wall.** Binds each [`KEY_TABLES`] table to
/// its registered view-kind; a table the server does not serve is noted and skipped.
/// If EVERY table failed, the first error is fatal (unreachable, not an empty
/// warehouse) — the two never draw the same.
///
/// **Off the egui thread.** The worker owns a clone of the source and does the ten
/// `KEY_TABLES` round trips; the frame keeps painting and [`Self::poll`] lands the
/// result. A scan started while one is in flight is not a second round trip.
///
/// MEASURED before this was off-thread (2026-08-25, this box, 60 ms per table
/// standing in for the real gRPC scans): the frame carrying the Scan click held for
/// **739 ms** and the window could not paint for the whole read.
///
/// NOTE this is a DIFFERENT defect from the `rt.block_on` in [`run_sql`], which runs
/// DataFusion over frames the client already holds — local compute, no RPC.
pub fn scan(&mut self) {
if self.pending.is_some() {
return;
}
let source = Arc::clone(&self.source);
let room = self.room.clone();
let (tx, rx) = std::sync::mpsc::channel();
std::thread::Builder::new()
.name("dataview-scan".into())
.spawn(move || {
// Only the READ happens here. Binding the scans to view-kinds builds
// `WarehouseDeck`, an egui-side widget tree, so that stays on the frame
// — it is local compute over rows already in memory.
let scans: Vec<(&'static str, Result<WarehouseScan, String>)> =
KEY_TABLES.iter().map(|&t| (t, source.scan(t))).collect();
let ask = |surface: &str, params: serde_json::Value| {
room.as_ref().map(|r| r.ask(surface, ¶ms))
};
let symbols = ask(S_SYMBOLS, serde_json::json!({ "limit": DECK_SCAN_LIMIT }));
let calls = ask(S_CALLS, serde_json::json!({ "limit": CALL_GRAPH_EDGES }));
let _ = tx.send(WallRead {
scans,
symbols,
calls,
});
})
.expect("spawn dataview scan worker");
self.pending = Some((rx, std::time::Instant::now()));
}
/// Non-blocking: land the in-flight read if the worker finished. Called once/frame.
pub fn poll(&mut self) {
let Some((rx, _)) = &self.pending else { return };
match rx.try_recv() {
Ok(res) => {
self.pending = None;
self.land_scan(res);
}
Err(std::sync::mpsc::TryRecvError::Empty) => {}
Err(std::sync::mpsc::TryRecvError::Disconnected) => {
self.pending = None;
// A dead worker is a named failure that keeps the wall already on
// screen — never a silent blank that reads as "the warehouse is empty".
self.error = Some("the data-view read worker died before it answered".into());
}
}
}
/// Blocking: wait for the in-flight read. The ROBOT seam (`update_json`) uses this so
/// a scripted scan is complete when the call returns; the window never blocks.
pub fn wait(&mut self) {
let Some((rx, _)) = self.pending.take() else {
return;
};
match rx.recv() {
Ok(res) => self.land_scan(res),
Err(_) => self.error = Some("the data-view read worker died before it answered".into()),
}
// a new wall asks the source on screen again: that answer is part of this read
self.source_pane.wait();
}
/// Whether a read is in flight.
pub fn is_pending(&self) -> bool {
self.pending.is_some()
}
/// Bind a landed read to the deck wall — the local half of the old `scan`.
fn land_scan(&mut self, read: WallRead) {
let WallRead {
scans,
symbols,
calls,
} = read;
let registry = WarehouseRegistry::default_nordisk();
let mut deck = WarehouseDeck::new("🏢 Mímir warehouse");
let mut frames: Vec<(String, RecordBatch)> = Vec::new();
let mut notes: Vec<String> = Vec::new();
let mut first_err: Option<String> = None;
let mut ok = 0usize;
let mut graph_panes: Vec<GraphPaneLabels> = Vec::new();
// ── the code room's two panes: a measured answer binds, anything else is a
// NOTE carrying the state and the reader's sentence — never an empty pane.
if let Some(a) = &symbols {
if a.is_measured() {
let batch = scan_to_batch(&symbols_scan(a));
frames.push(("room_symbols".to_string(), batch.clone()));
deck.add_for(ROOM_SYMBOLS, "room_symbols", PaneData::Grid(vec![batch]));
ok += 1;
} else {
notes.push(format!(
"{ROOM_SYMBOLS} [{}]: {}",
a.state,
a.evidence_text()
));
}
}
if let Some(a) = &calls {
match (a.is_measured(), call_graph_of(a)) {
(true, Some((spec, labels, monikers, edge_scan))) => {
frames.push(("room_calls".to_string(), scan_to_batch(&edge_scan)));
graph_panes.push(GraphPaneLabels { title: ROOM_CALLS.to_string(), kind: NodeKind::Symbol, labels, monikers, last: None });
deck.add_for(ROOM_CALLS, "room_calls", PaneData::Graph(spec));
ok += 1;
}
(true, None) => notes.push(format!("{ROOM_CALLS} [{}]: the room holds no call edge between the workspace's own fns — {}", a.state, a.evidence_text())),
(false, _) => notes.push(format!("{ROOM_CALLS} [{}]: {}", a.state, a.evidence_text())),
}
}
for (table, res) in scans {
let scan = match res {
Ok(s) => s,
Err(e) => {
first_err.get_or_insert_with(|| e.clone());
notes.push(format!("{table}: {e}"));
continue;
}
};
ok += 1;
let batch = scan_to_batch(&scan);
frames.push((table.to_string(), batch));
// The registered view-kind, with an honest fallback to Grid when the shape a
// graph/chart needs is not present in the scan (never a fabricated graph).
let kind = registry.kind_for(table).unwrap_or(ViewKind::Grid);
let data = match kind {
ViewKind::Graph3D => graph_cols(table)
.and_then(|(s, d)| scan_to_graph(&scan, s, d))
.map(PaneData::Graph)
.unwrap_or_else(|| PaneData::Grid(vec![scan_to_batch(&scan)])),
ViewKind::Chart => chart_col(table)
.and_then(|y| scan_to_chart(&scan, y))
.map(PaneData::Chart)
.unwrap_or_else(|| PaneData::Grid(vec![scan_to_batch(&scan)])),
ViewKind::Grid | ViewKind::BlobViewer => PaneData::Grid(vec![scan_to_batch(&scan)]),
};
// The click seam: remember this graph pane's node labels IN NODE ORDER before
// the spec moves into the deck. Only a table whose labels have an honest
// reading gets one — a graph whose node ids mean something this pane cannot
// name resolves to nothing rather than to a guess.
if let (PaneData::Graph(spec), Some(kind)) =
(&data, crate::graph_source::node_kind_for(table))
{
graph_panes.push(GraphPaneLabels {
title: table.to_string(),
kind,
labels: spec.nodes.iter().map(|(l, _)| l.clone()).collect(),
monikers: Vec::new(),
last: None,
});
}
deck.add_for(table, table, data);
}
if ok == 0 {
// Nothing bound — the server answered nothing, which is UNREACHABLE, not an
// empty warehouse. Keep the note list and surface the reason.
self.error = first_err.or(Some("no warehouse tables available".to_string()));
self.deck = None;
self.frames = Vec::new();
self.notes = notes;
self.graph_panes = Vec::new();
self.room_symbols = symbols;
self.room_calls = calls;
self.source_pane.clear();
return;
}
self.error = None;
self.deck = Some(deck);
self.frames = frames;
self.notes = notes;
self.graph_panes = graph_panes;
self.room_symbols = symbols;
self.room_calls = calls;
// the wall is a new reading: the source on screen is asked again
self.source_pane.reread();
}
/// **Read back which node the user clicked**, for every graph pane on the wall, and
/// feed the label to the 📄 source pane.
///
/// `Graph3D` has always resolved the press and lit the neighbour subgraph; what it
/// could not do was tell the HOST, because a deck pane is a `Box<dyn Facet>`. It now
/// opts into `Facet::as_any_mut`, so `FacetDeck::facet_mut` hands back the concrete
/// pane and its `state().selected` index — which this maps through the pane's own
/// label vector. Only a CHANGE counts as a click, so two panes each holding a
/// selection do not overwrite each other once per frame.
fn poll_graph_clicks(&mut self) {
let Some(deck) = self.deck.as_mut() else {
return;
};
let inner = deck.deck_mut();
for pane in &mut self.graph_panes {
let Some(g) = inner.facet_mut::<facett_graph3d::Graph3D>(&pane.title) else {
continue;
};
let now = g.state().selected;
if now == pane.last {
continue;
}
pane.last = now;
match now.and_then(|i| pane.labels.get(i).map(|l| (i, l))) {
Some((i, label)) => self.source_pane.select(NodeRef {
label: label.clone(),
moniker: pane.monikers.get(i).cloned().flatten(),
kind: pane.kind,
}),
// A click on empty canvas clears — the honest "nothing selected" state,
// not a stale slice from the last node.
None => self.source_pane.clear(),
}
}
}
/// **Run the SQL prompt** over the fetched frames. Requires a prior [`scan`](Self::scan)
/// (the frames are the tables). On success: builds the result grid AND feeds the
/// result back into the deck as a new pane. On failure: records the honest error.
pub fn run_sql(&mut self) {
self.sql_error = None;
if self.frames.is_empty() {
self.sql_status = "no tables loaded — press Scan first".to_string();
self.sql_error = Some(self.sql_status.clone());
self.sql_result = None;
self.sql_columns.clear();
self.sql_rows = 0;
return;
}
let sql = self.sql_input.trim().to_string();
if sql.is_empty() {
self.sql_status = "empty query".to_string();
self.sql_error = Some(self.sql_status.clone());
return;
}
match run_sql(&self.frames, &sql) {
Ok(batches) => {
let rows: usize = batches.iter().map(|b| b.num_rows()).sum();
let columns: Vec<String> = batches
.first()
.map(|b| {
b.schema()
.fields()
.iter()
.map(|f| f.name().clone())
.collect()
})
.unwrap_or_default();
self.sql_rows = rows;
self.sql_columns = columns;
self.sql_status =
format!("ok — {rows} row(s), {} column(s)", self.sql_columns.len());
// The result grid (reused icebergview) …
let snap =
Snapshot::new(SnapshotMeta::new(0, now_ms(), rows as u64), batches.clone());
self.sql_result = Some(IcebergView::new("SQL result", "sql", vec![snap]));
// … and the result feeds the deck as a new pane (a query joins the wall).
if let Some(deck) = &mut self.deck {
let title = format!("SQL: {}", short_sql(&sql));
deck.add_for(title, "sql", PaneData::Grid(batches));
}
}
Err(e) => {
// Honest: show the engine's error, drop no crumbs, never crash.
self.sql_status = format!("SQL error: {e}");
self.sql_error = Some(e);
self.sql_result = None;
self.sql_columns.clear();
self.sql_rows = 0;
}
}
}
fn scanned(&self) -> bool {
self.deck.is_some()
}
/// Deck introspection for the status line + oracle.
fn pane_count(&self) -> usize {
self.deck
.as_ref()
.map(WarehouseDeck::pane_count)
.unwrap_or(0)
}
}
/// A short, single-line form of a SQL string for a pane title.
fn short_sql(sql: &str) -> String {
let one: String = sql.split_whitespace().collect::<Vec<_>>().join(" ");
if one.len() > 48 {
format!("{}…", &one[..48])
} else {
one
}
}
impl Facet for DataViewPane {
fn title(&self) -> &str {
&self.title
}
fn ui(&mut self, ui: &mut egui::Ui) {
// Land an in-flight read, if the worker finished, and keep asking for frames
// while one is running so the "reading" line stays live.
self.poll();
if self.pending.is_some() {
ui.ctx().request_repaint();
}
// GENERATION, not a press (owner 2026-08-25: SCAN is not a user action).
// The deck binds itself on the first frame and rebinds when the workspace's
// snapshot id moves — i.e. when a populate has actually committed new rows.
let now_s = ui.input(|i| i.time);
let g = match self.gen.as_mut() {
Some(p) => p.poll(now_s),
None => Some("fixture".to_string()),
};
if let Some(g) = g {
if self.watch.saw(&g) {
self.scan();
}
}
ui.horizontal(|ui| {
ui.heading("🏢 Data view");
});
ui.label(egui::RichText::new(format!("source: {}", self.source.describe())).small());
// A read IN FLIGHT is named on the surface. The wall already bound (if any)
// stays painted below — a refresh never blanks a good reading.
if let Some((_, started)) = &self.pending {
ui.horizontal(|ui| {
ui.spinner();
ui.label(
egui::RichText::new(format!(
"reading the warehouse — {} table(s), {:.1}s (the scan runs off the \
UI thread; the wall below, if any, is the previous reading)",
KEY_TABLES.len(),
started.elapsed().as_secs_f32()
))
.small(),
);
});
}
if let Some(e) = &self.error {
ui.colored_label(
egui::Color32::from_rgb(0xc6, 0x28, 0x28),
format!("UNAVAILABLE — {e}"),
);
}
// What the code room's two panes rest on: state, evidence, the room read.
if let Some(a) = &self.room_symbols {
crate::room::callout(ui, ROOM_SYMBOLS, a);
}
if let Some(a) = &self.room_calls {
crate::room::callout(ui, ROOM_CALLS, a);
}
if !self.notes.is_empty() {
ui.collapsing(format!("{} table(s) not bound", self.notes.len()), |ui| {
for n in &self.notes {
ui.label(egui::RichText::new(n).small());
}
});
}
if !self.scanned() {
ui.label(
egui::RichText::new(
"no reading yet — Scan reads nornir's warehouse tables and binds each to \
a view-kind (graph / grid / chart) in a wall of panes.",
)
.small(),
);
return;
}
// ── the deck WALL, with the 📄 SOURCE of the clicked node beside it ──
// The wall is on the left, the source pane is a RESIZABLE right panel, and the
// divider between them is a real drag handle (`egui::Panel::right(..)
// .resizable(true)` — `.show`, not the deprecated `.show_inside`). The split is
// allocated a bounded height so the SQL prompt below still has a frame to live in.
ui.separator();
self.poll_graph_clicks();
let (source_pane, deck) = (&mut self.source_pane, self.deck.as_mut());
if let Some(deck) = deck {
let h = (ui.available_height() - SQL_PROMPT_HEIGHT).max(MIN_WALL_HEIGHT);
let size = egui::vec2(ui.available_width(), h);
ui.allocate_ui(size, |ui| {
source_pane.show_beside(ui, |ui| Facet::ui(deck, ui));
});
}
// ── the DataFusion SQL prompt ──
ui.separator();
ui.horizontal(|ui| {
ui.strong("SQL");
if !sql_enabled() {
ui.label(
egui::RichText::new("(engine off — build with --features dataview-sql)")
.small()
.italics(),
);
}
});
ui.add(
egui::TextEdit::multiline(&mut self.sql_input)
.desired_rows(2)
.desired_width(f32::INFINITY)
.font(egui::TextStyle::Monospace)
.hint_text(DEFAULT_SQL),
);
ui.horizontal(|ui| {
if ui.button("Run SQL").clicked() {
self.run_sql();
}
ui.label(
egui::RichText::new(
"runs DataFusion over the fetched frames (columns are Utf8 — CAST for numeric aggregates)",
)
.small(),
);
});
if let Some(e) = &self.sql_error {
ui.colored_label(egui::Color32::from_rgb(0xc6, 0x28, 0x28), e);
} else if !self.sql_status.is_empty() {
ui.label(egui::RichText::new(&self.sql_status).monospace().small());
}
if let Some(grid) = &mut self.sql_result {
ui.separator();
Facet::ui(grid, ui);
}
}
fn state_json(&self) -> serde_json::Value {
let deck_json = self
.deck
.as_ref()
.map(Facet::state_json)
.unwrap_or(serde_json::Value::Null);
serde_json::json!({
"kind": "dataview",
"title": self.title,
"source": self.source.describe(),
"scanned": self.scanned(),
// The in-flight oracle a robot reads to tell "still reading" from "read nothing".
"pending": self.pending.is_some(),
"error": self.error,
"pane_count": self.pane_count(),
"tables_bound": self.frames.iter().map(|(t, _)| t.clone()).collect::<Vec<_>>(),
"notes": self.notes,
// The full deck oracle: pane_count, per-kind counts, panes[] (table/kind/rows).
"deck": deck_json,
// The 📄 source pane: which node is selected, which of the four states it
// resolved to, and the row evidence behind a miss.
"graph_source": self.source_pane.state_json(),
"graph_panes": self.graph_panes.iter()
.map(|p| serde_json::json!({ "title": p.title, "nodes": p.labels.len(), "labels": p.labels }))
.collect::<Vec<_>>(),
// The code room's two panes: each answer's state, evidence and room, and the
// reader's own totals beside what the wall bound.
"room": {
"symbols": self.room_symbols.as_ref().map(RoomAnswer::state_json),
"calls": self.room_calls.as_ref().map(RoomAnswer::state_json),
"symbols_shown": self.room_symbols.as_ref().filter(|a| a.is_measured()).map(|a| a.rows("symbols").len()),
"symbols_total": self.room_symbols.as_ref().filter(|a| a.is_measured()).and_then(|a| a.body["total"].as_u64()),
"call_nodes": self.room_calls.as_ref().filter(|a| a.is_measured()).map(|a| a.rows("nodes").len()),
"call_edges_shown": self.room_calls.as_ref().filter(|a| a.is_measured()).map(|a| a.rows("edges").len()),
"call_edges_total": self.room_calls.as_ref().filter(|a| a.is_measured()).and_then(|a| a.body["total_edges"].as_u64()),
},
// The SQL prompt oracle.
"sql_enabled": sql_enabled(),
"sql_input": self.sql_input,
"sql_status": self.sql_status,
"sql_error": self.sql_error,
"sql_ok": self.sql_error.is_none() && !self.sql_status.is_empty(),
"sql_rows": self.sql_rows,
"sql_columns": self.sql_columns,
})
}
fn update_json(&mut self, msg_json: &str) {
if msg_json.trim() == "\"scan\"" {
self.scan();
self.wait();
return;
}
let Ok(v) = serde_json::from_str::<serde_json::Value>(msg_json) else {
return;
};
if v.get("scan").and_then(|b| b.as_bool()).unwrap_or(false) {
self.scan();
self.wait();
}
// `{"select_node":"<label>"}` — the headless twin of a click on a node label of
// the code room's call graph: asks the 📄 source pane's question and waits.
if let Some(label) = v.get("select_node").and_then(|s| s.as_str()) {
let hit = self.graph_panes.iter().find_map(|p| {
p.labels.iter().position(|l| l == label).map(|i| NodeRef {
label: label.to_string(),
moniker: p.monikers.get(i).cloned().flatten(),
kind: p.kind,
})
});
match hit {
Some(node) => {
self.source_pane.select(node);
self.source_pane.wait();
}
None => self.source_pane.clear(),
}
}
// Set the query text (optionally) …
if let Some(s) = v.get("sql_input").and_then(|s| s.as_str()) {
self.sql_input = s.to_string();
}
// … `{"sql":"…"}` sets the text AND runs it (the robot's one-shot drive) …
if let Some(s) = v.get("sql").and_then(|s| s.as_str()) {
self.sql_input = s.to_string();
self.run_sql();
}
// … `{"run_sql":true}` runs whatever is in the box.
if v.get("run_sql").and_then(|b| b.as_bool()).unwrap_or(false) {
self.run_sql();
}
}
}
/// The shipped Data view tab, bound to the SELECTED workspace over the thin gRPC seam.
pub fn dataview_pane(endpoint: &str, token: &str, workspace: &str) -> DataViewPane {
DataViewPane::live(endpoint, token, workspace)
}
#[cfg(test)]
mod tests {
use super::*;
/// **UNIT — `short_sql` collapses whitespace and truncates at 48 chars.** (autonom gap.)
/// The tab renders a run's SQL as a compact one-liner; the two behaviours that matter
/// are whitespace collapse (a multi-line query becomes one line) and truncation with an
/// ellipsis past 48 chars — both pinned, plus the boundary where truncation begins.
#[test]
fn short_sql_collapses_whitespace_and_truncates_with_an_ellipsis() {
// Whitespace (newlines, runs of spaces, tabs) collapses to single spaces.
assert_eq!(
short_sql("SELECT a,\n b\tFROM t"),
"SELECT a, b FROM t",
"runs of whitespace collapse to single spaces"
);
// Under the cap: returned whole, no ellipsis.
let short = "SELECT 1";
assert_eq!(short_sql(short), "SELECT 1");
assert!(!short_sql(short).contains('…'));
// Exactly 48 chars after collapse: still whole (the cap is `> 48`).
let s48 = "a".repeat(48);
assert_eq!(
short_sql(&s48),
s48,
"48 chars is the last un-truncated length"
);
// Over 48: truncated to 48 chars plus an ellipsis.
let s60 = "b".repeat(60);
let out = short_sql(&s60);
assert_eq!(
out.chars().filter(|c| *c == 'b').count(),
48,
"kept exactly the first 48"
);
assert!(out.ends_with('…'), "and marked as truncated: {out}");
}
/// A never-blank fixture warehouse of tables that are still written: git heat across
/// two repos (a chart, and the sample query groups it into two rows), the dependency
/// graph (a 3D graph), coverage and test outcomes (grids).
pub struct FixtureDataSource;
impl FixtureDataSource {
fn scan_of(cols: &[&str], rows: &[&[&str]]) -> WarehouseScan {
WarehouseScan {
columns: cols.iter().map(|s| s.to_string()).collect(),
rows: rows
.iter()
.map(|r| r.iter().map(|s| s.to_string()).collect())
.collect(),
}
}
}
impl DataViewSource for FixtureDataSource {
fn scan(&self, table: &str) -> Result<WarehouseScan, String> {
match table {
"test_outcomes" => Ok(Self::scan_of(
&["repo", "test_name", "outcome"],
&[
&["znippy", "parse_stream_round_trips", "ok"],
&["znippy", "write_frame_is_stable", "ok"],
&["korp", "spark_ingest_smoke", "failed"],
],
)),
"dep_graph_edges" => Ok(Self::scan_of(
&["from_repo", "to_repo"],
&[
&["korp", "znippy"],
&["korp", "skade"],
&["skade", "znippy"],
],
)),
"git_heat_facts" => Ok(Self::scan_of(
&["repo", "file", "commits_90d"],
&[
&["znippy", "src/hot.rs", "37"],
&["znippy", "src/lib.rs", "12"],
&["korp", "src/spark.rs", "5"],
],
)),
"coverage" => Ok(Self::scan_of(
&["repo", "pct"],
&[&["znippy", "84"], &["korp", "61"]],
)),
// The rest of KEY_TABLES are "not served" in the fixture — recorded as
// notes, proving a missing table reads as a named absence, not a drop.
other => Err(format!("nornir warehouse not_served: no table `{other}`")),
}
}
fn describe(&self) -> String {
"fixture (demo) — two repos, a dependency graph, git heat, coverage, tests".to_string()
}
}
fn fixture() -> DataViewPane {
DataViewPane::with_source(Box::new(FixtureDataSource))
}
#[test]
fn opens_unscanned_then_binds_the_deck_over_key_tables() {
let mut p = fixture();
assert_eq!(
p.state_json()["scanned"],
false,
"opens honestly un-scanned"
);
p.update_json("\"scan\"");
let s = p.state_json();
assert_eq!(s["scanned"], true);
// Four fixture tables bound → four panes on the wall.
assert_eq!(s["pane_count"], 4, "one pane per served key table");
let bound: Vec<String> = serde_json::from_value(s["tables_bound"].clone()).unwrap();
assert!(bound.contains(&"git_heat_facts".to_string()));
assert!(bound.contains(&"dep_graph_edges".to_string()));
// dep_graph_edges bound to a 3D graph, git_heat_facts to a chart, the rest to grids.
assert!(
s["deck"]["graph3d_count"].as_u64().unwrap() >= 1,
"dep_graph_edges → 3D graph"
);
assert!(
s["deck"]["chart_count"].as_u64().unwrap() >= 1,
"git_heat_facts → chart"
);
assert!(
s["deck"]["grid_count"].as_u64().unwrap() >= 1,
"coverage/test_outcomes → grids"
);
// The un-served KEY_TABLES are named, not silently dropped.
let notes: Vec<String> = serde_json::from_value(s["notes"].clone()).unwrap();
assert!(
notes.iter().any(|n| n.starts_with("embeddings:")),
"missing table is named: {notes:?}"
);
// No room was given to this pane: the wall has no code-room pane and claims none.
assert!(!bound.iter().any(|t| t.starts_with("room_")), "{bound:?}");
assert!(s["room"]["symbols"].is_null() && s["room"]["calls"].is_null());
}
/// A call graph and a symbol listing as the reader answers them.
fn room_answers() -> std::sync::Arc<crate::room::testing::CannedRoom> {
use crate::room::testing::{room, CannedRoom};
let sym = |name: &str, kind: &str, file: &str, line: u32, end: u32| {
serde_json::json!({
"repo": "shop", "crate_name": "shop", "module_path": "shop::pay", "item_kind": kind, "item_name": name, "visibility": "pub",
"file": file, "line": line, "end_line": end, "path": format!("shop::pay::{name}"), "moniker": format!("rust-analyzer cargo shop 0.1.0 pay/{name}()."),
})
};
let node = |id: &str, m: &str| serde_json::json!({ "id": id, "moniker": m, "crate": "shop", "kind": "fn", "provenance": "source_only", "file": "src/pay.rs", "line": 7 });
CannedRoom::new(&[
(
S_SYMBOLS,
serde_json::json!({
"state": "complete", "total": 41, "by_kind": { "fn": 30 },
"symbols": [sym("charge", "fn", "src/pay.rs", 7, 22), sym("audit", "fn", "src/pay.rs", 24, 30)],
"evidence": { "missing": [], "notes": [], "capped": { "shown": 2, "total": 41 } },
"room": room("w", "shop", 2, 31, 60),
}),
),
(
S_CALLS,
serde_json::json!({
"state": "complete", "total_edges": 7,
"nodes": [node("shop::checkout", "m:checkout"), node("shop::pay::charge", "m:charge"), node("shop::pay::audit", "m:audit")],
"edges": [{ "from": 0, "to": 1, "sites": 3 }, { "from": 1, "to": 2, "sites": 2 }],
"evidence": { "missing": [], "notes": ["a lower bound"] },
"room": room("w", "shop", 2, 31, 60),
}),
),
(
crate::graph_source::SURFACE,
serde_json::json!({
"state": "complete",
"node": { "path": "shop::pay::charge" }, "candidates": [{}],
"repo": "shop", "file": "src/pay.rs", "start_line": 7, "end_line": 9, "sha": "0123456789abcdef0123",
"lines": ["pub fn charge(cart: &Cart) -> Receipt {", " audit(cart);", "}"],
"truncated": false, "evidence": { "missing": [], "notes": [] }, "room": room("w", "shop", 2, 31, 60),
}),
),
])
}
/// **The code room's two panes are the reader's rows**: the symbol grid holds the
/// reader's symbols, the call graph the reader's nodes and edges in the reader's
/// order, and both say which room they were read from.
#[test]
fn the_wall_binds_the_rooms_symbols_and_call_graph() {
let src = room_answers();
let mut p = fixture().with_room(src.clone());
p.update_json("\"scan\"");
let s = p.state_json();
assert_eq!(s["pane_count"], 6, "four tables + the room's two panes");
let bound: Vec<String> = serde_json::from_value(s["tables_bound"].clone()).unwrap();
assert!(
bound.contains(&"room_symbols".to_string())
&& bound.contains(&"room_calls".to_string()),
"{bound:?}"
);
// the reader's own figures, beside what was bound
assert_eq!(
(
s["room"]["symbols_shown"].as_u64(),
s["room"]["symbols_total"].as_u64()
),
(Some(2), Some(41))
);
assert_eq!(
(
s["room"]["call_nodes"].as_u64(),
s["room"]["call_edges_shown"].as_u64(),
s["room"]["call_edges_total"].as_u64()
),
(Some(3), Some(2), Some(7))
);
assert_eq!(s["room"]["symbols"]["evidence"], "capped: 2 of 41 shown");
assert_eq!(
s["room"]["calls"]["room"],
"workspace w · shop@2 (0123456789) · 31 nodes · 60 edges"
);
let graph = s["graph_panes"]
.as_array()
.unwrap()
.iter()
.find(|g| g["title"] == ROOM_CALLS)
.expect("the call graph is a graph pane");
assert_eq!(
graph["labels"],
serde_json::json!(["shop::checkout", "shop::pay::charge", "shop::pay::audit"])
);
let panes = s["deck"]["panes"].as_array().unwrap();
let call = panes.iter().find(|x| x["title"] == ROOM_CALLS).unwrap();
assert_eq!(
(call["nodes"].as_u64(), call["edges"].as_u64()),
(Some(3), Some(2)),
"{call}"
);
assert_eq!(
panes.iter().find(|x| x["title"] == ROOM_SYMBOLS).unwrap()["rows"],
2
);
// what was asked
let asked = src.asked.lock().unwrap().clone();
assert!(
asked.contains(&(
S_SYMBOLS.to_string(),
serde_json::json!({ "limit": DECK_SCAN_LIMIT })
)),
"{asked:?}"
);
assert!(
asked.contains(&(
S_CALLS.to_string(),
serde_json::json!({ "limit": CALL_GRAPH_EDGES })
)),
"{asked:?}"
);
}
/// **A click on a call-graph node asks the room for THAT item** (by its moniker) and
/// the source pane shows the server's span and text.
#[test]
fn a_clicked_call_graph_node_shows_the_rooms_source() {
let src = room_answers();
let mut p = fixture().with_room(src.clone());
p.update_json("\"scan\"");
p.update_json(r#"{"select_node":"shop::pay::charge"}"#);
let g = &p.state_json()["graph_source"];
assert_eq!(
(g["state"].as_str(), g["moniker"].as_str()),
(Some("source"), Some("m:charge"))
);
assert_eq!(g["heading"], "shop/src/pay.rs:7-9 @ 0123456789");
assert_eq!(
(g["lines"].as_u64(), g["first_line"].as_str()),
(Some(3), Some("pub fn charge(cart: &Cart) -> Receipt {"))
);
assert!(src
.asked
.lock()
.unwrap()
.iter()
.any(|a| a.0 == crate::graph_source::SURFACE && a.1["moniker"] == "m:charge"));
// a repo node of the dependency graph is not an item
p.update_json(r#"{"select_node":"korp"}"#);
assert_eq!(p.state_json()["graph_source"]["state"], "repo_node");
}
/// **A room that refuses gets NO pane**: the wall names the state and the reader's
/// sentence for each, and the tables that are tables still bind.
#[test]
fn a_refusing_room_gets_no_pane_and_is_named() {
use crate::room::testing::{refusal, CannedRoom};
let why = "code room of `w`: no repo has a consumer offset yet";
let src = CannedRoom::new(&[
(S_SYMBOLS, refusal("not_consumed", why)),
(S_CALLS, refusal("not_consumed", why)),
]);
let mut p = fixture().with_room(src);
p.update_json("\"scan\"");
let s = p.state_json();
assert_eq!(s["pane_count"], 4, "only the four tables");
let notes: Vec<String> = serde_json::from_value(s["notes"].clone()).unwrap();
assert!(
notes.contains(&format!("{ROOM_SYMBOLS} [not_consumed]: {why}")),
"{notes:?}"
);
assert!(
notes.contains(&format!("{ROOM_CALLS} [not_consumed]: {why}")),
"{notes:?}"
);
assert_eq!(s["room"]["symbols"]["state"], "not_consumed");
assert!(
s["room"]["symbols_shown"].is_null()
&& s["room"]["symbols_total"].is_null()
&& s["room"]["call_nodes"].is_null(),
"nothing is counted of a refusal: {}",
s["room"]
);
assert!(s["graph_panes"]
.as_array()
.unwrap()
.iter()
.all(|g| g["title"] != ROOM_CALLS));
}
#[cfg(feature = "dataview-sql")]
#[test]
fn sql_prompt_runs_a_known_query_and_a_bad_query_errors_honestly() {
let mut p = fixture();
p.update_json("\"scan\"");
let before = p.pane_count();
// A known GROUP BY → two repos (znippy, korp).
p.update_json(r#"{"sql":"SELECT repo, count(*) AS n FROM git_heat_facts GROUP BY repo"}"#);
let s = p.state_json();
assert_eq!(
s["sql_ok"], true,
"the known query ran: {}",
s["sql_status"]
);
assert_eq!(s["sql_rows"], 2, "two repos group into two rows");
let cols: Vec<String> = serde_json::from_value(s["sql_columns"].clone()).unwrap();
assert!(
cols.contains(&"repo".to_string()) && cols.contains(&"n".to_string()),
"{cols:?}"
);
// The result fed the deck as a new pane (a query joins the wall).
assert_eq!(
s["pane_count"].as_u64().unwrap() as usize,
before + 1,
"SQL result feeds a deck pane"
);
// A bad query surfaces DataFusion's error verbatim — no crash, no empty-as-ok.
p.update_json(r#"{"sql":"SELECT * FROM no_such_table"}"#);
let s = p.state_json();
assert_eq!(s["sql_ok"], false, "a bad query is not ok");
assert!(s["sql_error"].is_string(), "the honest error is shown");
assert!(
s["sql_status"].as_str().unwrap().starts_with("SQL error:"),
"status names it a SQL error: {}",
s["sql_status"]
);
}
#[cfg(not(feature = "dataview-sql"))]
#[test]
fn sql_prompt_says_engine_off_when_not_built_in() {
let mut p = fixture();
p.update_json("\"scan\"");
p.update_json(r#"{"run_sql":true}"#);
let s = p.state_json();
assert_eq!(s["sql_enabled"], false);
assert_eq!(s["sql_ok"], false);
assert!(
s["sql_error"].as_str().unwrap().contains("dataview-sql"),
"the off-stub names the feature to rebuild with: {}",
s["sql_error"]
);
}
/// **A source that answers correctly, just not instantly.**
///
/// The stand-in for what the SHIPPED source really does: `WarehouseDataSource::scan`
/// is one `nornir_warehouse_embed::scan_warehouse_table` gRPC round trip, and
/// [`DataViewPane::scan`] runs one PER TABLE in [`KEY_TABLES`] — eight of them, in
/// series. The per-call sleep models that: what the timing test measures is WHERE the
/// wait happens, not whether the read works.
struct SlowSource(std::time::Duration);
impl DataViewSource for SlowSource {
fn scan(&self, table: &str) -> Result<WarehouseScan, String> {
std::thread::sleep(self.0);
FixtureDataSource.scan(table)
}
fn describe(&self) -> String {
"slow warehouse (test) — answers, but not this frame".to_string()
}
}
/// Per-table hold. Eight `KEY_TABLES` in series ⇒ ~480 ms of read, comfortably longer
/// than a 60 fps frame budget, so "the frame waited" is not a timing coincidence.
const PER_TABLE: std::time::Duration = std::time::Duration::from_millis(60);
/// **RED-when-broken: pressing Scan must give the frame back before the warehouse
/// answers.**
///
/// Driven through the REAL widget tree of the shipped pane (egui_kittest) against a
/// source that holds each table for [`PER_TABLE`]. What is asserted is the wall-clock
/// the frame carrying the click took, plus the state the pane publishes to a robot
/// (`pending`) — not the presence of a channel.
///
/// This is a DIFFERENT defect from the `rt.block_on` in [`run_sql`], which runs
/// DataFusion over frames the client already holds (local compute, no RPC). The one
/// under test here is the ten-round-trip warehouse read.
#[test]
fn the_deck_binds_itself_on_open_without_holding_the_frame() {
use egui_kittest::kittest::{By, Queryable};
use egui_kittest::Harness;
let whole = PER_TABLE * KEY_TABLES.len() as u32;
let pane =
std::cell::RefCell::new(DataViewPane::with_source(Box::new(SlowSource(PER_TABLE))));
let mut h = Harness::builder()
.with_size(egui::vec2(1200.0, 800.0))
.build_ui(|ui| {
pane.borrow_mut().ui(ui);
});
// ── THE OPENING FRAME, timed. There is no Scan button any more (the owner
// 2026-08-25: SCAN is not a user action); the GENERATION drive starts the
// read on the first frame, so a synchronous read would hang the window on
// OPEN with nobody having pressed anything.
let t0 = std::time::Instant::now();
h.step();
let held = t0.elapsed();
assert!(
held < whole,
"the opening frame held for {held:?} — the source needs {whole:?} to answer \
all {} tables, so the warehouse read ran ON the egui thread and the window \
could not paint while it ran",
KEY_TABLES.len()
);
assert_eq!(
pane.borrow().state_json()["pending"],
serde_json::json!(true),
"the opening frame must START the read and hand the frame back — state {}",
pane.borrow().state_json()
);
// Role-filtered: the pane's prose may legitimately say "scan"; it is the
// CONTROL that must be gone.
assert!(
h.query_all(
By::new()
.role(accesskit::Role::Button)
.label_contains("Scan")
)
.next()
.is_none(),
"the deck must carry no Scan control — binding is the generation's job"
);
assert!(
h.query_all_by_label_contains("reading the warehouse")
.next()
.is_some(),
"a pane with a read in flight must SAY SO on the rendered surface"
);
// Keep painting; the worker lands the deck. `step` in a pump, never `run()`,
// which panics on a frame that keeps requesting repaints.
let mut landed = false;
for _ in 0..600 {
h.step();
if pane.borrow().state_json()["pending"] == serde_json::json!(false) {
landed = true;
break;
}
std::thread::sleep(std::time::Duration::from_millis(10));
}
assert!(landed, "the read must land within 6 s");
let js = pane.borrow().state_json();
assert_eq!(
js["scanned"],
serde_json::json!(true),
"…and the deck is the worker's"
);
assert!(
js["pane_count"].as_u64().unwrap() >= 4,
"the SAME wall the inline read built"
);
}
/// **A refresh in flight never blanks the wall already on screen.** The deck the
/// operator is reading stays painted for every frame the new read is running.
#[test]
fn a_refresh_in_flight_never_blanks_the_deck_already_on_screen() {
let mut p = fixture();
p.update_json("\"scan\"");
let panes = p.state_json()["pane_count"].clone();
assert!(panes.as_u64().unwrap_or(0) >= 4, "a wall to protect");
p.source = std::sync::Arc::new(SlowSource(PER_TABLE));
p.scan();
assert!(p.is_pending(), "the refresh is off-thread");
for _ in 0..10 {
p.poll();
assert_eq!(
p.state_json()["pane_count"],
panes,
"a refresh in flight must never blank the wall already on screen"
);
std::thread::sleep(std::time::Duration::from_millis(10));
}
p.wait();
assert_eq!(p.state_json()["pending"], serde_json::json!(false));
}
#[test]
fn a_dead_server_says_unavailable_not_empty() {
struct Dead;
impl DataViewSource for Dead {
fn scan(&self, _t: &str) -> Result<WarehouseScan, String> {
Err("nornir-server unreachable: connect to http://127.0.0.1:0".into())
}
fn describe(&self) -> String {
"nornir warehouse @ http://127.0.0.1:0".into()
}
}
let mut p = DataViewPane::with_source(Box::new(Dead));
p.update_json("\"scan\"");
let s = p.state_json();
assert_eq!(s["scanned"], false, "a dead server produced no deck");
assert_eq!(s["pane_count"], 0);
assert!(s["error"].as_str().unwrap().contains("unreachable"));
}
}