tests/viz_workspace_matrix.rs
//! The **workspace × function matrix** — the owner's law: the viz test matrix
//! must prove that EVERY workspace's EVERY data-bearing function retrieves that
//! workspace's OWN functional data over the live thin-mode RPCs, and that
//! switching the workspace picker re-scopes every function with **no stale leak**
//! from the previously selected workspace (the bug that shipped in the Security
//! tab — fixed in 0.4.15 — because nothing tested the switch transition).
//!
//! `viz_md_matrix.rs` crosses tab × mode × data for ONE workspace. This file adds
//! the missing axis: **workspace**. It spins ONE ephemeral `nornir-server`
//! monitoring THREE workspaces, each `builds/` warehouse pre-seeded with a
//! DISTINCT amount of data (2 / 3 / 4 release_events rows) and a DISTINCT repo
//! set, so:
//!
//! 1. **per-workspace functional data** — a thin app scoped to each workspace
//! reads its OWN counts back over `Viz.ReleaseEvents` / `Viz.BakeoffResults`
//! / `Viz.Timeline` / `Warehouse.Scan`, and the Security tab lists that
//! workspace's repos. Every cell asserts NON-EMPTY + the EXACT distinct
//! count, so a cross-workspace leak fails the test.
//! 2. **switch isolation** — one app switches alpha→beta→gamma→alpha through
//! `switch_workspace_for_test` (the picker's code path) and after each
//! switch every function reports the CURRENT workspace's numbers, never the
//! previous one's.
//!
//! Build / run:
//! cargo test --features viz,server --test viz_workspace_matrix -- --nocapture
#![cfg(all(feature = "viz", feature = "server"))]
use std::io::Write as _;
use std::net::TcpListener;
use std::path::{Path, PathBuf};
use std::process::{Child, Command};
use std::time::{Duration, Instant};
use nornir::release::pipeline::{RepoGitState, RepoReleaseRecord, persist_lineage};
use nornir::viz::UrdrThreadsApp;
use nornir::warehouse::agent_model_runs::{AgentModelRunRow, append_agent_model_runs};
use nornir::warehouse::dep_graph::{CrossRepoEdge, WorkspaceGraph, record_dep_graph};
use nornir::warehouse::iceberg::IcebergWarehouse;
use nornir::warehouse::release_events::{ReleaseEventRow, append_release_events, phase, status};
use serde_json::Value;
use uuid::Uuid;
/// One seeded workspace: its name, its distinct release_events row count, and its
/// distinct repo members. The counts/members differ per workspace so a leak from
/// another workspace is detectable.
struct Ws {
name: &'static str,
/// release_events rows → the Release tab's row count + Warehouse data rows.
events: usize,
/// repo members (declared in the descriptor) → the Security tab's buttons.
repos: &'static [&'static str],
}
const WORKSPACES: [Ws; 3] = [
Ws {
name: "alpha",
events: 2,
repos: &["alpha-core"],
},
Ws {
name: "beta",
events: 3,
repos: &["beta-core", "beta-cli"],
},
Ws {
name: "gamma",
events: 4,
repos: &["gamma-a", "gamma-b", "gamma-c"],
},
];
// ── per-workspace seed ──────────────────────────────────────────────────────
fn release_rows(run: &str, n: usize) -> Vec<ReleaseEventRow> {
(0..n)
.map(|i| ReleaseEventRow {
run_id: run.into(),
seq: i as i64,
ts_micros: 1_000_000 + i as i64 * 1_000,
component: format!("c{i}"),
repo: format!("c{i}"),
op: "test".into(),
phase: if i % 2 == 0 { phase::START } else { phase::END }.into(),
status: if i % 2 == 0 {
status::RUNNING
} else {
status::OK
}
.into(),
detail: format!("event {i}"),
depends_on: None,
elapsed_ms: (i % 2 == 1).then_some(100 + i as i64),
})
.collect()
}
fn bakeoff_rows(run: &str) -> Vec<AgentModelRunRow> {
vec![AgentModelRunRow {
run_id: run.into(),
ts_micros: 200,
agent: "local-llm".into(),
model: "opus".into(),
prompt_id: "p1".into(),
prompt: "Q?".into(),
output: "A".into(),
latency_ms: 100.0,
tokens_in: 12,
tokens_out: 18,
tokens_per_s: 300.0,
score: 0.9,
ok: true,
error: None,
cost_usd: 0.0,
mcp_tool_calls: 0,
}]
}
/// Seed `ws`'s `builds/` warehouse with all the tables the thin viz reads.
fn seed_ws(builds: &Path, ws: &Ws) {
std::fs::create_dir_all(builds).unwrap();
let wh = IcebergWarehouse::open(builds).expect("open warehouse");
let run = format!("run-{}", ws.name);
// dep_graph + release_lineage → the Viz.Timeline RPC builds the timeline.
let edges = vec![CrossRepoEdge {
from: ws.repos[0].into(),
to: ws.repos.last().unwrap().to_string(),
via: [format!("{}-link", ws.name)].into_iter().collect(),
kind: Default::default(),
}];
let graph = WorkspaceGraph::from_query_parts(std::collections::BTreeMap::new(), edges);
let snapshot_id = wh
.block_on(record_dep_graph(&wh, ws.name, &graph))
.expect("dep_graph");
let release_id =
Uuid::from_u128(0x4e6f_726e_6972_0000_0000_0000_0000_0001u128 + ws.events as u128);
let lineage: Vec<RepoReleaseRecord> = ws
.repos
.iter()
.enumerate()
.map(|(i, r)| RepoReleaseRecord {
repo: (*r).to_string(),
build_order_idx: i,
git: RepoGitState {
sha: format!("{r}sha"),
branch: "main".into(),
dirty: false,
},
gate_status: "succeeded".into(),
tests_passed: 1,
tests_failed: 0,
published_versions: vec![((*r).to_string(), "0.1.0".into())],
tantivy_snapshot_id: None,
dwarf_snapshot_id: None,
})
.collect();
wh.block_on(persist_lineage(
&wh,
release_id,
ws.name,
&snapshot_id,
&lineage,
false,
))
.expect("persist_lineage");
// release_events (distinct count) + agent_model_runs (leaderboard).
wh.block_on(append_release_events(&wh, &release_rows(&run, ws.events)))
.expect("release_events");
wh.block_on(append_agent_model_runs(&wh, &bakeoff_rows(&run)))
.expect("agent_model_runs");
drop(wh);
}
// ── ephemeral multi-workspace server ────────────────────────────────────────
fn tmp(tag: &str) -> PathBuf {
// One dir per call: the tests run in parallel in one process, and each one's
// server holds its warehouses' locks while the next one seeds its own.
static SEQ: std::sync::atomic::AtomicU32 = std::sync::atomic::AtomicU32::new(0);
let seq = SEQ.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
let base = std::env::temp_dir().join(format!("nornir-wsmatrix-{tag}-{}-{seq}", std::process::id()));
std::fs::create_dir_all(&base).unwrap();
base
}
fn free_port() -> u16 {
TcpListener::bind("127.0.0.1:0")
.unwrap()
.local_addr()
.unwrap()
.port()
}
struct Server {
child: Child,
root: PathBuf,
endpoint: String,
token: String,
}
impl Drop for Server {
fn drop(&mut self) {
let _ = self.child.kill();
let _ = self.child.wait();
let _ = std::fs::remove_dir_all(&self.root);
}
}
/// Spawn one server monitoring all three seeded workspaces.
fn spawn_server() -> Server {
let token = "viz-wsmatrix-token-0123456789";
let port = free_port();
let endpoint = format!("http://127.0.0.1:{port}");
let root = tmp("srv");
let server_home = root.join("home");
let registry_root = server_home.join(".nornir").join("workspaces");
std::fs::create_dir_all(®istry_root).unwrap();
let mut descs: Vec<(&str, PathBuf)> = Vec::new();
for ws in &WORKSPACES {
// Pre-seed the warehouse BEFORE the server opens (and locks) it.
seed_ws(®istry_root.join(ws.name).join("builds"), ws);
// A path-member descriptor: the fetch loop skips path members (no clone),
// so it registers instantly with the declared repos as members.
let desc = root.join(format!("{}.toml", ws.name));
let mut f = std::fs::File::create(&desc).unwrap();
writeln!(f, "[workspace]\nname = \"{}\"\n", ws.name).unwrap();
for r in ws.repos {
// Quote the key so the hyphenated member name is preserved verbatim.
// Each member points at this crate's dir (a real path so it's skipped
// cleanly); only the member NAME matters for the Security/repos test.
writeln!(
f,
"[repos.\"{}\"]\npath = \"{}\"\n",
r,
env!("CARGO_MANIFEST_DIR")
)
.unwrap();
}
descs.push((ws.name, desc));
}
let child = Command::new(env!("CARGO_BIN_EXE_nornir-server"))
.env("NORNIR_SERVER_ADDR", format!("127.0.0.1:{port}"))
.env("NORNIR_SERVER_TOKEN", token)
.env("HOME", &server_home)
.env("NORNIR_POLL", "3600s")
.stdout(std::process::Stdio::null())
.stderr(std::process::Stdio::null())
.spawn()
.expect("spawn nornir-server");
let srv = Server {
child,
root,
endpoint: endpoint.clone(),
token: token.to_string(),
};
register_when_up(&endpoint, token, &descs.iter().map(|(n, d)| (*n, d.as_path())).collect::<Vec<_>>());
// Readiness: wait until ALL three workspaces are served.
let deadline = Instant::now() + Duration::from_secs(45);
loop {
if let Ok(ws) = nornir::viz::remote::list_workspaces(&endpoint, token) {
if WORKSPACES.iter().all(|w| ws.iter().any(|s| s == w.name)) {
break;
}
}
assert!(
Instant::now() < deadline,
"server never served all 3 workspaces"
);
std::thread::sleep(Duration::from_millis(300));
}
srv
}
// ── reading one workspace's functions over the thin RPCs ────────────────────
/// Build a thin app scoped to `ws`, render every tab a couple of frames so the
/// remote loaders settle, and return its `state_json` plus the directly-scanned
/// warehouse release_events row count (the Warehouse data axis).
fn thin_state(srv: &Server, ws: &Ws) -> Value {
let mut app = UrdrThreadsApp::with_remote(
srv.endpoint.clone(),
srv.token.clone(),
ws.name.to_string(),
PathBuf::new(),
ws.repos.iter().map(|s| s.to_string()).collect(),
);
app.render_all_tabs_headless();
app.render_all_tabs_headless();
let mut s = app.state_json();
let wh_rows =
nornir::viz::remote::scan_table(&srv.endpoint, &srv.token, "release_events", 100, ws.name)
.map(|p| p.rows.len() as u64)
.unwrap_or(0);
s["_wh_data_rows"] = serde_json::json!(wh_rows);
s
}
fn release_rows_of(s: &Value) -> u64 {
s["release"]["rows"]
.as_array()
.map(|a| a.len())
.unwrap_or(0) as u64
}
fn security_repos_of(s: &Value) -> Vec<String> {
s["security"]["repos"]
.as_array()
.map(|a| {
a.iter()
.filter_map(|v| v.as_str().map(String::from))
.collect()
})
.unwrap_or_default()
}
// ════════════════════════════════════════════════════════════════════════════
// 1. per-workspace functional data — every function reads ITS OWN data
// ════════════════════════════════════════════════════════════════════════════
#[test]
fn every_workspace_every_function_retrieves_its_own_data() {
let srv = spawn_server();
let mut grid = String::from("\nworkspace × function — thin RPC functional data\n");
for ws in &WORKSPACES {
let s = thin_state(&srv, ws);
// Release — Viz.ReleaseEvents must return EXACTLY this workspace's count.
let rel = release_rows_of(&s);
assert_eq!(
rel, ws.events as u64,
"{} Release: expected {} events over Viz.ReleaseEvents, got {rel} (leak/empty?)",
ws.name, ws.events
);
// Warehouse — Warehouse.Scan of release_events, scoped to this workspace.
let wh = s["_wh_data_rows"].as_u64().unwrap_or(0);
assert_eq!(
wh, ws.events as u64,
"{} Warehouse release_events rows = {wh}, expected {}",
ws.name, ws.events
);
// Timeline — Viz.Timeline built from this workspace's lineage+dep_graph.
let tl = s["timeline"]["releases"].as_u64().unwrap_or(0);
assert!(
tl > 0,
"{} Timeline: expected ≥1 release over Viz.Timeline, got 0",
ws.name
);
assert!(
s["timeline"]["error"].is_null(),
"{} Timeline not clean",
ws.name
);
// Security — the repo buttons must be EXACTLY this workspace's members.
let mut got = security_repos_of(&s);
got.sort();
let mut want: Vec<String> = ws.repos.iter().map(|s| s.to_string()).collect();
want.sort();
assert_eq!(
got, want,
"{} Security repos {:?} != workspace members {:?}",
ws.name, got, want
);
grid.push_str(&format!(
" {:>6} · Release={rel} Warehouse={wh} Timeline={tl} Security={:?} ✅\n",
ws.name, got
));
}
println!("{grid}");
}
// ════════════════════════════════════════════════════════════════════════════
// 2. switch isolation — the picker re-scopes EVERY function, no stale leak
// ════════════════════════════════════════════════════════════════════════════
#[test]
fn switching_workspace_rescopes_every_function_no_leak() {
let srv = spawn_server();
// One app, started on alpha; drive the picker through every workspace (and
// back to alpha) exactly as a click does, asserting each function follows.
let first = &WORKSPACES[0];
let mut app = UrdrThreadsApp::with_remote(
srv.endpoint.clone(),
srv.token.clone(),
first.name.to_string(),
PathBuf::new(),
first.repos.iter().map(|s| s.to_string()).collect(),
);
let visit_order = [0usize, 1, 2, 0, 2, 1]; // include repeats + backtracking
let mut prev: Option<&Ws> = None;
let mut log = String::from("\nswitch isolation — each switch re-scopes, no leak\n");
for &i in &visit_order {
let ws = &WORKSPACES[i];
app.switch_workspace_for_test(ws.name.to_string());
app.render_all_tabs_headless();
app.render_all_tabs_headless();
let s = app.state_json();
// Release rows must equal the CURRENT workspace's count — never the
// previously-selected workspace's (the stale-leak bug class).
let rel = release_rows_of(&s);
assert_eq!(
rel,
ws.events as u64,
"after switch → {}: Release={rel}, expected {} (leaked from {:?}?)",
ws.name,
ws.events,
prev.map(|p| p.name)
);
// Security repos must be the current workspace's members.
let mut got = security_repos_of(&s);
got.sort();
let mut want: Vec<String> = ws.repos.iter().map(|s| s.to_string()).collect();
want.sort();
assert_eq!(
got, want,
"after switch → {}: Security repos {:?} != {:?}",
ws.name, got, want
);
// And explicitly: none of the previous workspace's repos linger.
if let Some(p) = prev {
for r in p.repos {
if !ws.repos.contains(r) {
assert!(
!got.contains(&r.to_string()),
"{} leaked repo `{r}` from {}",
ws.name,
p.name
);
}
}
}
log.push_str(&format!(
" → {:>6}: Release={rel} Security={:?} ✅\n",
ws.name, got
));
prev = Some(ws);
}
println!("{log}");
}
/// The server reads no `NORNIR_MONITOR` any more: once it answers, register each
/// workspace through the real `Workspaces.Register` RPC so it is served.
fn register_when_up(endpoint: &str, token: &str, descs: &[(&str, &std::path::Path)]) {
let deadline = Instant::now() + Duration::from_secs(45);
while nornir::viz::remote::list_workspaces(endpoint, token).is_err() {
assert!(Instant::now() < deadline, "empty server never came up");
std::thread::sleep(Duration::from_millis(200));
}
for (name, desc) in descs {
nornir::viz::remote::register_workspace(endpoint, token, name, &desc.to_string_lossy(), "monitored", "3600s")
.unwrap_or_else(|e| panic!("register {name}: {e:#}"));
}
}