src/viz/remote.rs
//! Thin-client timeline loader: fetch a [`Timeline`] from a running
//! `nornir-server` over the `Viz.Timeline` gRPC instead of opening a local
//! Iceberg warehouse. The server builds the timeline from the warehouse it
//! owns (it holds the redb lock) and returns it as JSON; we deserialize into
//! the same [`Timeline`] the embedded path produces, so the egui app is
//! source-agnostic.
use anyhow::{Context, Result};
use super::model::{LiveEvent, Timeline};
use crate::warehouse::iceberg::TablePreview;
// The generated `nornir.v1` prost/tonic client stubs now come from the shared
// `nornir-wire` edda leaf (`nornir_wire::pb`) instead of a private copy.
use nornir_wire::pb;
/// Build an `http://…` endpoint url + a `Bearer <token>` metadata value — the
/// shared connect/auth shape for the viz gRPC clients.
fn endpoint_and_bearer(
endpoint: &str,
token: &str,
) -> Result<(
String,
Option<tonic::metadata::MetadataValue<tonic::metadata::Ascii>>,
)> {
let endpoint = if endpoint.starts_with("http") {
endpoint.to_string()
} else {
format!("http://{endpoint}")
};
// The ONE shared client bearer-builder (Law #1) — no hand-rolled twin.
// `None` ⇒ open (empty token); the interceptors stamp only when `Some`.
let bearer = nornir_service::client::bearer(token).context("parse bearer token")?;
Ok((endpoint, bearer))
}
/// The `nornir-workspace` metadata value for `workspace` (empty ⇒ none) — selects
/// which served workspace the gRPC calls target. Driven by the app's currently
/// selected workspace (the in-UI picker), not an env var, so the viz can switch
/// workspaces live.
fn ws_header(workspace: &str) -> Option<tonic::metadata::MetadataValue<tonic::metadata::Ascii>> {
(!workspace.is_empty())
.then(|| workspace.parse().ok())
.flatten()
}
/// List the workspaces the server has registered (`Workspaces.List` RPC) — the
/// names that populate the viz's workspace picker.
pub fn list_workspaces(endpoint: &str, token: &str) -> Result<Vec<String>> {
let rt = tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()
.context("build tokio runtime for viz client")?;
rt.block_on(async {
let (endpoint, bearer) = endpoint_and_bearer(endpoint, token)?;
let channel = tonic::transport::Channel::from_shared(endpoint.clone())
.with_context(|| format!("invalid server url `{endpoint}`"))?
.connect()
.await
.with_context(|| format!("connect to nornir-server at {endpoint}"))?;
let mut client = pb::workspaces_client::WorkspacesClient::with_interceptor(
channel,
move |mut req: tonic::Request<()>| {
if let Some(b) = &bearer {
req.metadata_mut().insert("authorization", b.clone());
}
Ok(req)
},
);
let resp = client
.list(pb::Empty {})
.await
.context("Workspaces.List RPC")?
.into_inner();
Ok(resp.workspaces.into_iter().map(|w| w.name).collect())
})
}
/// `Workspaces.PopulateStatus` (PO1) — the per-workspace populate ROLL-UP for the
/// whole registry, decoded into the same [`RosterRow`](crate::registry::RosterRow)
/// the embedded path builds, so the Nornir-tab roster renders identically in thin
/// mode. ONE round-trip backs the whole roster (a compact summary, not the full
/// clone_events ledger — so it never overflows the gRPC cap on big workspaces).
pub fn fetch_populate_status(
endpoint: &str,
token: &str,
) -> Result<Vec<nornir_wire::view::rows::RosterRow>> {
let rt = tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()
.context("build tokio runtime for viz client")?;
rt.block_on(async {
let (endpoint, bearer) = endpoint_and_bearer(endpoint, token)?;
let channel = tonic::transport::Channel::from_shared(endpoint.clone())
.with_context(|| format!("invalid server url `{endpoint}`"))?
.connect()
.await
.with_context(|| format!("connect to nornir-server at {endpoint}"))?;
let mut client = pb::workspaces_client::WorkspacesClient::with_interceptor(
channel,
move |mut req: tonic::Request<()>| {
if let Some(b) = &bearer {
req.metadata_mut().insert("authorization", b.clone());
}
Ok(req)
},
)
.max_decoding_message_size(64 << 20);
let resp = client
.populate_status(pb::Empty {})
.await
.context("Workspaces.PopulateStatus RPC")?
.into_inner();
let rows =
serde_json::from_str(&resp.json).context("decode populate_status json from server")?;
Ok(rows)
})
}
/// Fetch the timeline for `workspace` from `endpoint` (e.g.
/// `http://127.0.0.1:7878`), authenticating with the bearer `token`. Runs the
/// async tonic call on a private current-thread runtime so it's safe to call
/// from the synchronous egui update loop.
pub fn fetch_timeline(endpoint: &str, token: &str, workspace: &str) -> Result<Timeline> {
let rt = tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()
.context("build tokio runtime for viz client")?;
rt.block_on(async {
let endpoint = if endpoint.starts_with("http") {
endpoint.to_string()
} else {
format!("http://{endpoint}")
};
// The ONE shared client bearer-builder (Law #1) — no hand-rolled twin.
let bearer = nornir_service::client::bearer(token).context("parse bearer token")?;
let ws_md = ws_header(workspace);
let channel = tonic::transport::Channel::from_shared(endpoint.clone())
.with_context(|| format!("invalid server url `{endpoint}`"))?
.connect()
.await
.with_context(|| format!("connect to nornir-server at {endpoint}"))?;
let mut client = pb::viz_client::VizClient::with_interceptor(
channel,
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)
},
)
.max_decoding_message_size(64 << 20);
let resp = client
.timeline(pb::VizTimelineRequest {
workspace: workspace.to_string(),
})
.await
.context("Viz.Timeline RPC")?
.into_inner();
let timeline: Timeline =
serde_json::from_str(&resp.json).context("decode timeline json from server")?;
Ok(timeline)
})
}
/// `Viz.ReleaseEvents` — every `release_events` row for `workspace`, decoded
/// into the same [`ReleaseEventRow`] the embedded warehouse read returns, so the
/// 🚀 Release tab + 📡 Live Run hydrate render identically in thin mode. The
/// server holds the redb lock, so the rows come over this RPC instead of a local
/// warehouse open.
pub fn fetch_release_events(
endpoint: &str,
token: &str,
workspace: &str,
) -> Result<Vec<crate::warehouse::release_events::ReleaseEventRow>> {
call(endpoint, token, workspace, |channel, auth| async move {
let mut c = pb::viz_client::VizClient::with_interceptor(channel, auth)
.max_decoding_message_size(64 << 20);
let r = c
.release_events(pb::VizTimelineRequest {
workspace: String::new(),
})
.await
.context("Viz.ReleaseEvents RPC")?
.into_inner();
let rows =
serde_json::from_str(&r.json).context("decode release_events json from server")?;
Ok(rows)
})
}
/// `Viz.Coverage` — the latest source-coverage run for `repo` in `workspace`,
/// decoded into the same `(Vec<CoverageRow>, Vec<CoverageFnRow>, Vec<f64>)` the
/// embedded warehouse read produces, so the 📐 Coverage panel renders identically
/// in thin mode. The repo rides in the request's `workspace` field (the served
/// workspace itself comes from the `nornir-workspace` metadata header). The
/// server holds the redb lock, so the rows come over this RPC instead of a local
/// warehouse open.
#[allow(clippy::type_complexity)]
pub fn fetch_coverage(
endpoint: &str,
token: &str,
workspace: &str,
repo: &str,
) -> Result<(
Vec<crate::warehouse::coverage::CoverageRow>,
Vec<crate::warehouse::coverage::CoverageFnRow>,
Vec<f64>,
)> {
let repo = repo.to_string();
call(endpoint, token, workspace, |channel, auth| async move {
let mut c = pb::viz_client::VizClient::with_interceptor(channel, auth)
.max_decoding_message_size(64 << 20);
let r = c
// The repo to read rides in `workspace` (the served workspace comes
// from the metadata header `call()` already set).
.coverage(pb::VizTimelineRequest { workspace: repo })
.await
.context("Viz.Coverage RPC")?
.into_inner();
#[derive(serde::Deserialize)]
struct Payload {
rows: Vec<crate::warehouse::coverage::CoverageRow>,
fn_rows: Vec<crate::warehouse::coverage::CoverageFnRow>,
history: Vec<f64>,
}
let p: Payload =
serde_json::from_str(&r.json).context("decode coverage json from server")?;
Ok((p.rows, p.fn_rows, p.history))
})
}
/// `Viz.HolgerRegistry` — the embedded (married) holger dev-pair registry the
/// server serves, decoded into the same `Vec<RepoSummary>` the embedded
/// `holger_embed::read_registry` returns, so the 📦 Holger pane renders
/// identically in thin mode. The thin viz has no local holger to read, so the
/// rows come over this RPC instead of an in-process registry open.
#[cfg(feature = "embed-holger")]
pub fn fetch_holger_registry(
endpoint: &str,
token: &str,
workspace: &str,
) -> Result<Vec<crate::holger_embed::RepoSummary>> {
call(endpoint, token, workspace, |channel, auth| async move {
let mut c = pb::viz_client::VizClient::with_interceptor(channel, auth)
.max_decoding_message_size(64 << 20);
let r = c
.holger_registry(pb::VizTimelineRequest {
workspace: String::new(),
})
.await
.context("Viz.HolgerRegistry RPC")?
.into_inner();
let repos =
serde_json::from_str(&r.json).context("decode holger registry json from server")?;
Ok(repos)
})
}
/// `Viz.TestResults` — every `test_results` row for `workspace`, decoded into
/// the same [`TestResultRow`] the embedded warehouse read returns, so the 🧪
/// Test tab renders identically in thin mode.
pub fn fetch_test_results(
endpoint: &str,
token: &str,
workspace: &str,
) -> Result<Vec<crate::warehouse::test_results::TestResultRow>> {
call(endpoint, token, workspace, |channel, auth| async move {
let mut c = pb::viz_client::VizClient::with_interceptor(channel, auth)
.max_decoding_message_size(64 << 20);
let r = c
.test_results(pb::VizTimelineRequest {
workspace: String::new(),
})
.await
.context("Viz.TestResults RPC")?
.into_inner();
let rows = serde_json::from_str(&r.json).context("decode test_results json from server")?;
Ok(rows)
})
}
/// `Viz.BenchTelemetry` — the `bench_telemetry` + `bench_runs` rows for
/// `workspace`, decoded into the same `(Vec<BenchTelemetryRow>, Vec<BenchRun>)`
/// the embedded warehouse read returns, so the 📡 Bench LIVE panel folds and
/// renders identically in thin mode.
pub fn fetch_bench_live(
endpoint: &str,
token: &str,
workspace: &str,
) -> Result<(
Vec<crate::warehouse::iceberg::BenchTelemetryRow>,
Vec<crate::bench::BenchRun>,
)> {
call(endpoint, token, workspace, |channel, auth| async move {
let mut c = pb::viz_client::VizClient::with_interceptor(channel, auth)
.max_decoding_message_size(64 << 20);
let r = c
.bench_telemetry(pb::VizTimelineRequest {
workspace: String::new(),
})
.await
.context("Viz.BenchTelemetry RPC")?
.into_inner();
let v: serde_json::Value =
serde_json::from_str(&r.json).context("decode bench json from server")?;
let telemetry = serde_json::from_value(v.get("telemetry").cloned().unwrap_or_default())
.context("decode bench_telemetry rows")?;
let runs = serde_json::from_value(v.get("runs").cloned().unwrap_or_default())
.context("decode bench_runs rows")?;
Ok((telemetry, runs))
})
}
/// `Viz.BakeoffResults` — every `agent_model_runs` row for `workspace`, decoded
/// into the same [`AgentModelRunRow`] the embedded warehouse read returns, so the
/// 🏆 Leaderboard tab renders identically in thin mode.
pub fn fetch_bakeoff_results(
endpoint: &str,
token: &str,
workspace: &str,
) -> Result<Vec<crate::warehouse::agent_model_runs::AgentModelRunRow>> {
call(endpoint, token, workspace, |channel, auth| async move {
let mut c = pb::viz_client::VizClient::with_interceptor(channel, auth)
.max_decoding_message_size(64 << 20);
let r = c
.bakeoff_results(pb::VizTimelineRequest {
workspace: String::new(),
})
.await
.context("Viz.BakeoffResults RPC")?
.into_inner();
let rows =
serde_json::from_str(&r.json).context("decode agent_model_runs json from server")?;
Ok(rows)
})
}
/// `Viz.CloneEvents` — the recent `clone_events` populate outcomes for
/// `workspace`, decoded into the same [`CloneEventRow`] the embedded warehouse
/// read returns, so the 🧬 nornir root pane renders per-member populate status (ok
/// / failed + error + when) identically in thin mode. The server holds the redb
/// lock, so the rows come over this RPC instead of a local warehouse open. (PLAN #6)
pub fn fetch_clone_events(
endpoint: &str,
token: &str,
workspace: &str,
) -> Result<Vec<nornir_wire::view::rows::CloneEventRow>> {
call(endpoint, token, workspace, |channel, auth| async move {
let mut c = pb::viz_client::VizClient::with_interceptor(channel, auth)
.max_decoding_message_size(64 << 20);
let r = c
.clone_events(pb::VizTimelineRequest {
workspace: String::new(),
})
.await
.context("Viz.CloneEvents RPC")?
.into_inner();
let rows = serde_json::from_str(&r.json).context("decode clone_events json from server")?;
Ok(rows)
})
}
/// `Viz.Architecture` — the merged EPIC ARCH wiring board for `workspace`,
/// decoded into the same [`ArchPayload`](super::arch_tab::ArchPayload) the
/// embedded `arch::warehouse::merged_arch_from_warehouse` merge produces, so the
/// 🏛 Architecture tab renders identically in thin mode. The server holds the redb
/// lock, so the merge runs server-side and ships over this RPC. Empty/missing
/// board → an empty graph (never an error).
// VIZ-ONLY: returns an egui tab's view model (see the note on `funnel_show`).
#[cfg(feature = "viz")]
pub fn fetch_architecture(
endpoint: &str,
token: &str,
workspace: &str,
) -> Result<super::arch_tab::ArchPayload> {
call(endpoint, token, workspace, |channel, auth| async move {
let mut c = pb::viz_client::VizClient::with_interceptor(channel, auth)
.max_decoding_message_size(64 << 20);
let r = c
.architecture(pb::VizTimelineRequest {
workspace: String::new(),
})
.await
.context("Viz.Architecture RPC")?
.into_inner();
let payload =
serde_json::from_str(&r.json).context("decode architecture json from server")?;
Ok(payload)
})
}
/// `Viz.Knowledge` — the knowledge map of `workspace` from the SERVER's code
/// room, as the JSON its reader answers (`{repos, crates, crate_edges, state,
/// evidence}`). A workspace nothing was consumed into is the RPC's error (the
/// reader's refusal, by state) — never an empty map.
pub fn fetch_knowledge(endpoint: &str, token: &str, workspace: &str) -> Result<serde_json::Value> {
call(endpoint, token, workspace, |channel, auth| async move {
let mut c = pb::viz_client::VizClient::with_interceptor(channel, auth)
.max_decoding_message_size(64 << 20);
let r = c
.knowledge(pb::VizTimelineRequest {
workspace: String::new(),
})
.await
.context("Viz.Knowledge RPC")?
.into_inner();
serde_json::from_str(&r.json).context("decode knowledge json from server")
})
}
/// `Viz.TestMatrix` — `{inventory, inventory_state, results}` for `workspace`
/// from the server: the test nodes of its code room beside its `test_results`.
pub fn fetch_test_matrix(endpoint: &str, token: &str, workspace: &str) -> Result<serde_json::Value> {
call(endpoint, token, workspace, |channel, auth| async move {
let mut c = pb::viz_client::VizClient::with_interceptor(channel, auth)
.max_decoding_message_size(64 << 20);
let r = c
.test_matrix(pb::VizTimelineRequest {
workspace: String::new(),
})
.await
.context("Viz.TestMatrix RPC")?
.into_inner();
serde_json::from_str(&r.json).context("decode test matrix json from server")
})
}
/// A code-room surface answered IN PROCESS — what a FAT pane reads where a
/// thin one calls the RPC. The same reader, so the same JSON: the workspace's
/// own code room under `warehouse_root` (ONE handle per lake per process,
/// [`crate::code_room::reader::local`]). A refusal is the error, with its
/// state in brackets.
pub fn local_surface(
warehouse_root: &std::path::Path,
workspace: &str,
surface: &str,
params: &serde_json::Value,
) -> Result<serde_json::Value> {
let handle = crate::code_room::reader::local(workspace, warehouse_root);
crate::code_room::reader::answer(surface, &crate::code_room::Reach::room_only(&handle), params)
.map_err(|r| anyhow::anyhow!("{r} [{}]", r.state))
}
/// `Telemetry.SubmitBakeoff` — push client-run `agent_model_runs` rows into the
/// served `workspace`'s warehouse (the server owns the lock). Returns the
/// accepted count. The thin counterpart to the embedded `append_agent_model_runs`
/// — fills the live 🏆 Leaderboard. Mirrors the `nornir bakeoff` CLI submit.
pub fn submit_bakeoff(
endpoint: &str,
token: &str,
workspace: &str,
rows: &[crate::warehouse::agent_model_runs::AgentModelRunRow],
) -> Result<u32> {
let runs: Vec<pb::AgentModelRunPb> = rows
.iter()
.map(|r| pb::AgentModelRunPb {
run_id: r.run_id.clone(),
ts_micros: r.ts_micros,
agent: r.agent.clone(),
model: r.model.clone(),
prompt_id: r.prompt_id.clone(),
prompt: r.prompt.clone(),
output: r.output.clone(),
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.clone().unwrap_or_default(),
cost_usd: r.cost_usd,
mcp_tool_calls: r.mcp_tool_calls,
})
.collect();
call(endpoint, token, workspace, |channel, auth| async move {
let mut c = pb::telemetry_client::TelemetryClient::with_interceptor(channel, auth);
let r = c
.submit_bakeoff(pb::SubmitBakeoffRequest { runs })
.await
.context("Telemetry.SubmitBakeoff RPC")?
.into_inner();
Ok(r.accepted)
})
}
/// `Telemetry.SubmitTestResults` — push client-run `test_results` rows into the
/// served `workspace`'s warehouse. Returns the accepted count. Fills the live 🧪
/// Test pane. Mirrors the `nornir test` CLI submit.
pub fn submit_test_results(
endpoint: &str,
token: &str,
workspace: &str,
rows: &[crate::warehouse::test_results::TestResultRow],
) -> Result<u32> {
let pb_rows: Vec<pb::TestResultPb> = rows
.iter()
.map(|r| pb::TestResultPb {
run_id: r.run_id.clone(),
repo: r.repo.clone(),
suite: r.suite.clone(),
test_name: r.test_name.clone(),
status: r.status.clone(),
duration_ms: r.duration_ms,
ts_micros: r.ts_micros,
message: r.message.clone(),
aspect: r.aspect.clone(),
metric: r.metric,
})
.collect();
call(endpoint, token, workspace, |channel, auth| async move {
let mut c = pb::telemetry_client::TelemetryClient::with_interceptor(channel, auth);
let r = c
.submit_test_results(pb::SubmitTestResultsRequest { rows: pb_rows })
.await
.context("Telemetry.SubmitTestResults RPC")?
.into_inner();
Ok(r.accepted)
})
}
/// `Telemetry.SubmitReleaseEvents` — push client-run `release_events` rows into
/// the served `workspace`'s warehouse. Returns the accepted count. Fills the live
/// 🚀 Release tab. Mirrors the `nornir release run` CLI submit. The `Option`
/// fields (`depends_on`, `elapsed_ms`) are carried with a presence flag so None
/// round-trips distinct from an empty list.
pub fn submit_release_events(
endpoint: &str,
token: &str,
workspace: &str,
rows: &[crate::warehouse::release_events::ReleaseEventRow],
) -> Result<u32> {
let pb_rows: Vec<pb::ReleaseEventPb> = rows
.iter()
.map(|r| pb::ReleaseEventPb {
run_id: r.run_id.clone(),
seq: r.seq,
ts_micros: r.ts_micros,
component: r.component.clone(),
repo: r.repo.clone(),
op: r.op.clone(),
phase: r.phase.clone(),
status: r.status.clone(),
detail: r.detail.clone(),
has_depends_on: r.depends_on.is_some(),
depends_on: r.depends_on.clone().unwrap_or_default(),
has_elapsed_ms: r.elapsed_ms.is_some(),
elapsed_ms: r.elapsed_ms.unwrap_or_default(),
})
.collect();
call(endpoint, token, workspace, |channel, auth| async move {
let mut c = pb::telemetry_client::TelemetryClient::with_interceptor(channel, auth);
let r = c
.submit_release_events(pb::SubmitReleaseEventsRequest { rows: pb_rows })
.await
.context("Telemetry.SubmitReleaseEvents RPC")?
.into_inner();
Ok(r.accepted)
})
}
/// `Telemetry.SubmitBuildProvenance` — push client-run `BuildProvenance` records
/// into the served `workspace`'s `build_provenance` ledger. Each ships as
/// `BuildProvenance::to_json()` (so the record can gain fields without a proto
/// bump). Returns the accepted count. Mirrors [`submit_release_events`].
pub fn submit_build_provenance(
endpoint: &str,
token: &str,
workspace: &str,
records: &[nornir_build_thing::BuildProvenance],
) -> Result<u32> {
let records_json: Vec<String> = records
.iter()
.map(|p| p.to_json())
.collect::<Result<_>>()
.context("serialize BuildProvenance records")?;
call(endpoint, token, workspace, |channel, auth| async move {
let mut c = pb::telemetry_client::TelemetryClient::with_interceptor(channel, auth);
let r = c
.submit_build_provenance(pb::SubmitBuildProvenanceRequest { records_json })
.await
.context("Telemetry.SubmitBuildProvenance RPC")?
.into_inner();
Ok(r.accepted)
})
}
/// `Telemetry.SubmitBuildInfo` — push client-run §f3 build-info rows (manifest +
/// feed) into the served `workspace`'s `build_info` table. `blob` carries a
/// presence flag so None round-trips distinct from an empty payload. Returns the
/// accepted count. This is the NATIVE landing for a dwarves build's build-info.
pub fn submit_build_info(
endpoint: &str,
token: &str,
workspace: &str,
rows: &[nornir_build_thing::build_info::BuildInfoRow],
) -> Result<u32> {
let pb_rows: Vec<pb::BuildInfoRowPb> = rows
.iter()
.map(|r| pb::BuildInfoRowPb {
build_id: r.build_id.clone(),
step_id: r.step_id.clone(),
ts_micros: r.ts_micros,
kind: r.kind.clone(),
level: r.level.clone(),
name: r.name.clone(),
value: r.value.clone(),
content_id: r.content_id.clone(),
has_blob: r.blob.is_some(),
blob: r.blob.clone().unwrap_or_default(),
})
.collect();
call(endpoint, token, workspace, |channel, auth| async move {
let mut c = pb::telemetry_client::TelemetryClient::with_interceptor(channel, auth);
let r = c
.submit_build_info(pb::SubmitBuildInfoRequest { rows: pb_rows })
.await
.context("Telemetry.SubmitBuildInfo RPC")?
.into_inner();
Ok(r.accepted)
})
}
/// `CodeRoom.Answer` — ask the served `workspace`'s code room one reader
/// surface (`surface`, its request as JSON). Returns the answer's JSON text; a
/// refusal is a body (`refused` / `state`), not an error.
pub fn code_room_answer(endpoint: &str, token: &str, workspace: &str, surface: &str, params_json: &str) -> Result<String> {
let (surface, params_json) = (surface.to_string(), params_json.to_string());
call(endpoint, token, workspace, |channel, auth| async move {
// a pane answer can be large (the forge proxies at 64 MiB too)
let mut c = pb::code_room_client::CodeRoomClient::with_interceptor(channel, auth).max_decoding_message_size(64 << 20);
let r = c
.answer(pb::CodeRoomQuestion { surface, params_json })
.await
.context("CodeRoom.Answer RPC")?
.into_inner();
Ok(r.json)
})
}
/// `Telemetry.SubmitJobs` — push client-run job-ledger records into the served
/// `workspace`'s `jobs.redb` (the thin path: the client has no local FS access to
/// the server's ledger). `recs` ships as a JSON array of [`crate::jobs::JobRecord`].
/// Returns the accepted count. Mirrors [`submit_release_events`].
pub fn submit_jobs(
endpoint: &str,
token: &str,
workspace: &str,
recs: &[crate::jobs::JobRecord],
) -> Result<u32> {
let jobs_json = serde_json::to_string(recs).context("encode job records")?;
call(endpoint, token, workspace, |channel, auth| async move {
let mut c = pb::telemetry_client::TelemetryClient::with_interceptor(channel, auth);
let r = c
.submit_jobs(pb::SubmitJobsRequest { jobs_json })
.await
.context("Telemetry.SubmitJobs RPC")?
.into_inner();
Ok(r.accepted)
})
}
/// **The `limit` a `nornir jobs` invocation must put ON THE WIRE** — the one writer of
/// that rule, shared by the CLI reader and anything else that asks `Viz.Jobs` on an
/// operator's behalf.
///
/// The wire is not free to carry the flag verbatim, and that is the whole reason this
/// exists rather than a `as u32` at the call site:
///
/// * **`0` is a sentinel, not a number.** `Viz.Jobs` reads `limit == 0` as *"use my own
/// default"*, which is 200 rows. So `0` can express neither "no cap" nor any limit
/// above 200, and a call site that passes a literal `0` has silently dropped whatever
/// the operator typed. That is exactly what `run_jobs` did: `--limit 500` printed 200
/// rows and `--limit 0` — documented on the flag as *no cap* — printed 200 rows.
/// * **`--running` filters on the CLIENT, after the fetch.** Capping the wire under it
/// would hand back the newest N rows and then throw away the terminal ones, so an
/// operator with 12 running jobs behind 50 finished ones would be told there are none.
/// Anything the client still has to filter must fetch uncapped.
///
/// "No cap" is therefore spelled `u32::MAX`: the server's `rows.truncate(limit as usize)`
/// over a ledger that will never hold four billion rows is a no-op, and it needs no
/// change to the wire contract other lanes own.
#[must_use]
pub fn jobs_wire_limit(running: bool, limit: usize) -> u32 {
if running || limit == 0 {
u32::MAX
} else {
u32::try_from(limit).unwrap_or(u32::MAX)
}
}
/// `Viz.Jobs` — the unified job ledger for `workspace`, newest-first, capped to
/// `limit` (0 = server default). Returns decoded [`crate::jobs::JobRecord`]s for
/// the 🧬 nornir tab's job list in remote/thin mode.
pub fn fetch_jobs(
endpoint: &str,
token: &str,
workspace: &str,
kind: &str,
limit: u32,
) -> Result<Vec<crate::jobs::JobRecord>> {
let kind = kind.to_string();
let ws = workspace.to_string();
call(endpoint, token, workspace, |channel, auth| async move {
let mut c = pb::viz_client::VizClient::with_interceptor(channel, auth)
.max_decoding_message_size(64 << 20);
let r = c
.jobs(pb::JobsRequest {
workspace: ws,
kind,
limit,
})
.await
.context("Viz.Jobs RPC")?
.into_inner();
let rows = serde_json::from_str(&r.json).context("decode jobs json from server")?;
Ok(rows)
})
}
/// Open the server's `Release.Progress` server-stream and invoke `on_event`
/// for each converted [`LiveEvent`]. Blocks until the stream closes (the server
/// ends it after `RunEnd`) or errors — call it from a dedicated `std::thread`
/// (see [`super::live`]). Runs its own current-thread tokio runtime so it never
/// touches the egui loop. This is what makes a remote viz animate a release run
/// in real time over Tailscale: same events the local file tail would produce.
pub fn stream_progress(
endpoint: &str,
token: &str,
workspace: &str,
mut on_event: impl FnMut(LiveEvent),
) -> Result<()> {
let rt = tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()
.context("build tokio runtime for viz live client")?;
rt.block_on(async {
let (endpoint, bearer) = endpoint_and_bearer(endpoint, token)?;
let channel = tonic::transport::Channel::from_shared(endpoint.clone())
.with_context(|| format!("invalid server url `{endpoint}`"))?
.connect()
.await
.with_context(|| format!("connect to nornir-server at {endpoint}"))?;
let ws_md = ws_header(workspace);
let mut client = pb::release_client::ReleaseClient::with_interceptor(
channel,
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 stream = client
.progress(pb::Empty {})
.await
.context("Release.Progress RPC")?
.into_inner();
while let Some(ev) = stream.message().await.context("progress stream")? {
if let Some(live) = to_live(ev) {
on_event(live);
}
}
Ok(())
})
}
/// List every warehouse table the server owns (the `Warehouse.Tables` RPC) —
/// the remote counterpart to `IcebergWarehouse::table_names`.
pub fn fetch_tables(endpoint: &str, token: &str, workspace: &str) -> Result<Vec<String>> {
let rt = tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()
.context("build tokio runtime for viz client")?;
rt.block_on(async {
let (endpoint, bearer) = endpoint_and_bearer(endpoint, token)?;
let channel = tonic::transport::Channel::from_shared(endpoint.clone())
.with_context(|| format!("invalid server url `{endpoint}`"))?
.connect()
.await
.with_context(|| format!("connect to nornir-server at {endpoint}"))?;
let ws_md = ws_header(workspace);
let mut client = pb::warehouse_client::WarehouseClient::with_interceptor(
channel,
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 resp = client
.tables(pb::Empty {})
.await
.context("Warehouse.Tables RPC")?
.into_inner();
Ok(resp.names)
})
}
/// Scan one warehouse table for display (the `Warehouse.Scan` RPC) — the remote
/// counterpart to `IcebergWarehouse::scan_preview`. `limit` 0 = server default.
pub fn scan_table(
endpoint: &str,
token: &str,
table: &str,
limit: u32,
workspace: &str,
) -> Result<TablePreview> {
let rt = tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()
.context("build tokio runtime for viz client")?;
rt.block_on(async {
let (endpoint, bearer) = endpoint_and_bearer(endpoint, token)?;
let channel = tonic::transport::Channel::from_shared(endpoint.clone())
.with_context(|| format!("invalid server url `{endpoint}`"))?
.connect()
.await
.with_context(|| format!("connect to nornir-server at {endpoint}"))?;
let ws_md = ws_header(workspace);
let mut client = pb::warehouse_client::WarehouseClient::with_interceptor(
channel,
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 resp = client
.scan(pb::WarehouseScanRequest {
table: table.to_string(),
limit,
})
.await
.context("Warehouse.Scan RPC")?
.into_inner();
Ok(TablePreview {
columns: resp.columns,
rows: resp.rows.into_iter().map(|r| r.cells).collect(),
})
})
}
// ── shared connect helper + reusable interceptor ────────────────────────────
/// Bearer + optional `nornir-workspace` header, reusable across every generated
/// client (so the new RPC wrappers below don't each re-inline the closure).
#[derive(Clone)]
pub(crate) struct Auth {
bearer: Option<tonic::metadata::MetadataValue<tonic::metadata::Ascii>>,
ws: Option<tonic::metadata::MetadataValue<tonic::metadata::Ascii>>,
}
impl tonic::service::Interceptor for Auth {
fn call(&mut self, mut req: tonic::Request<()>) -> Result<tonic::Request<()>, tonic::Status> {
if let Some(b) = &self.bearer {
req.metadata_mut().insert("authorization", b.clone());
}
if let Some(ws) = &self.ws {
req.metadata_mut().insert("nornir-workspace", ws.clone());
}
Ok(req)
}
}
/// Run one blocking gRPC call on a private current-thread runtime: connect to
/// `endpoint`, build the [`Auth`] interceptor (bearer + `workspace` header), and
/// hand `(channel, auth)` to `f`. Keeps the egui loop synchronous.
fn call<T, F, Fut>(endpoint: &str, token: &str, workspace: &str, f: F) -> Result<T>
where
F: FnOnce(tonic::transport::Channel, Auth) -> Fut,
Fut: std::future::Future<Output = Result<T>>,
{
let rt = tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()
.context("build tokio runtime for viz client")?;
rt.block_on(async {
let (endpoint, bearer) = endpoint_and_bearer(endpoint, token)?;
let channel = tonic::transport::Channel::from_shared(endpoint.clone())
.with_context(|| format!("invalid server url `{endpoint}`"))?
.connect()
.await
.with_context(|| format!("connect to nornir-server at {endpoint}"))?;
let auth = Auth {
bearer,
ws: ws_header(workspace),
};
f(channel, auth).await
})
}
// ── plain view structs (UI consumes these, not the pb types) ─────────────────
/// Workspace info for the picker's side panel (`Workspaces.Get`).
#[derive(Clone, Debug, Default)]
pub struct WorkspaceInfo {
pub name: String,
pub mode: String,
pub poll: String,
pub current_snapshot: String,
pub updated_at: String,
/// `(member, source/sha summary)` pairs.
pub members: Vec<(String, String)>,
/// AUT6 working-tree freshness per member: `(member, dirty, digest)`. The
/// SHA in the summary above is committed state only; this reports whether the
/// server's clone has uncommitted changes. Exposed in `state_json`.
pub freshness: Vec<(String, bool, String)>,
}
/// One search hit (`Search.Query`).
#[derive(Clone, Debug)]
pub struct Hit {
pub corpus: String,
pub repo: String,
pub path: String,
pub score: f32,
pub title: String,
pub snippet: String,
}
/// One symbol (`Knowledge.SymbolLookup`).
#[derive(Clone, Debug)]
pub struct KnownSym {
pub crate_name: String,
pub item_kind: String,
pub item_name: String,
pub visibility: String,
pub file: String,
pub line: u32,
pub signature: String,
}
/// Release-gate outcome for a repo (`Release.GateAll`).
#[derive(Clone, Debug, Default)]
pub struct GateReport {
pub repo: String,
pub passed: Vec<String>,
pub failed: Vec<(String, String)>,
}
/// One bench metric series point for charting (`Bench.History`).
#[derive(Clone, Debug)]
pub struct BenchPoint {
pub date: String,
pub version: String,
pub metric: String,
pub value: f64,
}
/// Server identity returned by `Health.Ping` — surfaced in the About popup so a
/// remote viz shows the *server's* version next to the client's.
#[derive(Clone, Debug, Default)]
pub struct ServerInfo {
pub status: String,
pub version: String,
pub repo_count: u32,
}
/// `Health.Ping` — the server's version + status + loaded-repo count. Cheap, no
/// workspace scope. Powers the About popup's "server vX" line.
pub fn ping(endpoint: &str, token: &str) -> Result<ServerInfo> {
call(endpoint, token, "", |channel, auth| async move {
let mut c = pb::health_client::HealthClient::with_interceptor(channel, auth);
let r = c
.ping(pb::Empty {})
.await
.context("Health.Ping RPC")?
.into_inner();
Ok(ServerInfo {
status: r.status,
version: r.version,
repo_count: r.repo_count,
})
})
}
// ── new clickable-surface RPC wrappers ───────────────────────────────────────
/// `Workspaces.Get` — the info-panel backing call for the picker selection.
pub fn get_workspace(endpoint: &str, token: &str, name: &str) -> Result<WorkspaceInfo> {
call(endpoint, token, "", |channel, auth| async move {
let mut c = pb::workspaces_client::WorkspacesClient::with_interceptor(channel, auth);
let r = c
.get(pb::WorkspaceName {
name: name.to_string(),
purge: false,
})
.await
.context("Workspaces.Get RPC")?
.into_inner();
Ok(WorkspaceInfo {
name: r.name,
mode: r.mode,
poll: r.poll,
current_snapshot: r.current_snapshot,
updated_at: r.updated_at,
freshness: r
.members
.iter()
.map(|m| (m.name.clone(), m.worktree_dirty, m.worktree_digest.clone()))
.collect(),
members: r
.members
.into_iter()
.map(|m| {
// Char-boundary safe slice: a non-ASCII / placeholder sha
// would panic on a raw `&..[..8]` byte index.
let sha = match m.last_seen_sha.char_indices().nth(8) {
Some((byte_idx, _)) => &m.last_seen_sha[..byte_idx],
None => &m.last_seen_sha,
};
// AUT6 staleness marker: a dirty server clone is advisory —
// the warehouse was indexed from committed state only.
let fresh = if m.worktree_dirty {
" ⚠ uncommitted"
} else {
""
};
(
m.name,
format!("{} @ {sha} [{}]{fresh}", m.remote, m.sync_state),
)
})
.collect(),
})
})
}
/// `Workspaces.Fetch` — the "⟳ Sync now" button. Returns `(fetched, changed, errors)`.
/// `Workspaces.Fetch` — poll the remote(s) now. `force` republishes the
/// warehouse even when no git member changed (e.g. it was cleared). Returns
/// (fetched, changed, errors, snapshot).
pub fn fetch_workspace(
endpoint: &str,
token: &str,
name: &str,
force: bool,
) -> Result<(u32, Vec<String>, Vec<String>, String)> {
call(endpoint, token, "", |channel, auth| async move {
let mut c = pb::workspaces_client::WorkspacesClient::with_interceptor(channel, auth);
let r = c
.fetch(pb::WorkspaceFetchRequest {
name: name.to_string(),
force,
background: false,
})
.await
.context("Workspaces.Fetch RPC")?
.into_inner();
Ok((r.fetched, r.changed, r.errors, r.snapshot))
})
}
/// `Workspaces.Register` — add (and EAGER-populate) a workspace. `mode` is one of
/// `monitored` | `pushed` | `external` (default pushed); `descriptor` is the
/// server-readable `nornir-workspace.toml` path/URL for a monitored workspace.
/// Returns the registered workspace's `(name, mode, member_count)`. The viz "Add
/// workspace" button calls this; CLI parity = `nornir workspace add`.
pub fn register_workspace(
endpoint: &str,
token: &str,
name: &str,
descriptor: &str,
mode: &str,
poll: &str,
) -> Result<(String, String, usize)> {
call(endpoint, token, "", |channel, auth| async move {
let mut c = pb::workspaces_client::WorkspacesClient::with_interceptor(channel, auth);
let r = c
.register(pb::RegisterWorkspaceRequest {
name: name.to_string(),
descriptor: descriptor.to_string(),
mode: mode.to_string(),
poll: poll.to_string(),
lazy: false,
// The viz UI supplies a server-readable path/URL; pass through.
descriptor_content: String::new(),
})
.await
.context("Workspaces.Register RPC")?
.into_inner();
Ok((r.name, r.mode, r.members.len()))
})
}
/// `Workspaces.Remove` — kill (de-register) a workspace. The viz "Kill workspace"
/// button (behind a confirm) calls this; CLI parity = `nornir workspace rm`. When
/// `purge`, the server also wipes the workspace's on-disk `builds/` data (warehouse
/// + catalog.redb + jobs.redb) — same shared core as `nornir workspace rm --purge`.
pub fn remove_workspace(endpoint: &str, token: &str, name: &str, purge: bool) -> Result<()> {
call(endpoint, token, "", |channel, auth| async move {
let mut c = pb::workspaces_client::WorkspacesClient::with_interceptor(channel, auth);
c.remove(pb::WorkspaceName {
name: name.to_string(),
purge,
})
.await
.context("Workspaces.Remove RPC")?;
Ok(())
})
}
/// The common outcome of an `Ops.*` heavy run: overall verdict + the CLI-parity
/// summary line + per-target rows + the warehouse run id the rows landed under.
#[derive(Clone, Debug, Default)]
pub struct OpRunResult {
pub ok: bool,
pub summary: String,
/// `(target, status, message)` per repo/target.
pub targets: Vec<(String, String, String)>,
pub run_id: String,
}
fn op_result_from(r: pb::RunOpResult) -> OpRunResult {
OpRunResult {
ok: r.ok,
summary: r.summary,
targets: r
.targets
.into_iter()
.map(|t| (t.name, t.status, t.message))
.collect(),
run_id: r.run_id,
}
}
/// `Ops.RunTestMatrix` — run the test matrix server-side over the workspace's
/// checkouts and persist `test_results`. `repo` empty = the FULL matrix (heavy);
/// `aspects` empty = the default set. The 🧪 Test pane's "Run matrix" button calls
/// this; CLI parity = `nornir test all` (client mode) / `nornir test run <repo>`.
pub fn run_test_matrix(
endpoint: &str,
token: &str,
repo: &str,
aspects: &str,
workspace: &str,
) -> Result<OpRunResult> {
call(endpoint, token, workspace, |channel, auth| async move {
let mut c = pb::ops_client::OpsClient::with_interceptor(channel, auth);
let r = c
.run_test_matrix(pb::RunTestMatrixRequest {
repo: repo.to_string(),
aspects: aspects.to_string(),
// The viz button uses the server's own NORNIR_TEST_FEATURES env.
features: std::env::var("NORNIR_TEST_FEATURES").unwrap_or_default(),
})
.await
.context("Ops.RunTestMatrix RPC")?
.into_inner();
Ok(op_result_from(r))
})
}
/// `Ops.RunBench` — run the benches server-side for `repo` (empty = primary) and
/// persist a `BenchRun`. The 📈 Bench pane's "Run bencher" button calls this; CLI
/// parity = `nornir bench run <repo>`.
pub fn run_bench(endpoint: &str, token: &str, repo: &str, workspace: &str) -> Result<OpRunResult> {
call(endpoint, token, workspace, |channel, auth| async move {
let mut c = pb::ops_client::OpsClient::with_interceptor(channel, auth);
let r = c
.run_bench(pb::RunBenchRequest {
repo: repo.to_string(),
})
.await
.context("Ops.RunBench RPC")?
.into_inner();
Ok(op_result_from(r))
})
}
/// `Ops.RunRelease` — run the release-grade heavy gate across the build order and
/// persist `release_events`. The 🚀 Release pane's "Release (heavy gate)" button
/// calls this. Server-side it runs each repo's CONFIGURED `[gates]` suite — the
/// same one `Release.GateAll` runs. No CLI twin (release orchestration moved to
/// `dwarves release`).
pub fn run_release(
endpoint: &str,
token: &str,
repo: &str,
workspace: &str,
) -> Result<OpRunResult> {
call(endpoint, token, workspace, |channel, auth| async move {
let mut c = pb::ops_client::OpsClient::with_interceptor(channel, auth);
let r = c
.run_release(pb::RunReleaseRequest {
repo: repo.to_string(),
})
.await
.context("Ops.RunRelease RPC")?
.into_inner();
Ok(op_result_from(r))
})
}
/// `Workspaces.Fetch{background=true}` — the **Populate** path: clone members
/// now and return right after the (fast) clone, letting the server's poll loop
/// build the warehouse async (single-writer safe). Returns
/// `(fetched, changed, errors, snapshot)`; snapshot is empty (build is deferred).
/// CLI parity = `nornir workspace populate`.
pub fn populate_workspace(
endpoint: &str,
token: &str,
name: &str,
) -> Result<(u32, Vec<String>, Vec<String>, String)> {
call(endpoint, token, "", |channel, auth| async move {
let mut c = pb::workspaces_client::WorkspacesClient::with_interceptor(channel, auth);
let r = c
.fetch(pb::WorkspaceFetchRequest {
name: name.to_string(),
force: false,
background: true,
})
.await
.context("Workspaces.Fetch(background) RPC")?
.into_inner();
Ok((r.fetched, r.changed, r.errors, r.snapshot))
})
}
/// `Search.Query` — BM25 over the Tantivy corpora. `corpus`/`repo` empty = all.
pub fn search(
endpoint: &str,
token: &str,
query: &str,
corpus: &str,
repo: &str,
limit: u32,
workspace: &str,
) -> Result<Vec<Hit>> {
call(endpoint, token, workspace, |channel, auth| async move {
let mut c = pb::search_client::SearchClient::with_interceptor(channel, auth);
let r = c
.query(pb::SearchRequest {
query: query.to_string(),
corpus: corpus.to_string(),
repo: repo.to_string(),
limit,
})
.await
.context("Search.Query RPC")?
.into_inner();
Ok(r.hits
.into_iter()
.map(|h| Hit {
corpus: h.corpus,
repo: h.repo,
path: h.path,
score: h.score,
title: h.title,
snippet: h.snippet,
})
.collect())
})
}
/// `Knowledge.SymbolLookup` — item-name substring search over `symbol_facts`.
pub fn knowledge_lookup(
endpoint: &str,
token: &str,
repo: &str,
arg: &str,
limit: u32,
workspace: &str,
) -> Result<Vec<KnownSym>> {
call(endpoint, token, workspace, |channel, auth| async move {
let mut c = pb::knowledge_client::KnowledgeClient::with_interceptor(channel, auth);
let r = c
.symbol_lookup(pb::KnowledgeSymbolQuery {
repo: repo.to_string(),
arg: arg.to_string(),
limit,
})
.await
.context("Knowledge.SymbolLookup RPC")?
.into_inner();
Ok(r.symbols
.into_iter()
.map(|s| KnownSym {
crate_name: s.crate_name,
item_kind: s.item_kind,
item_name: s.item_name,
visibility: s.visibility,
file: s.file,
line: s.line,
signature: s.signature,
})
.collect())
})
}
/// `Release.GateAll` — run every release gate for `repo`, pass/fail split.
pub fn gate_all(endpoint: &str, token: &str, repo: &str, workspace: &str) -> Result<GateReport> {
call(endpoint, token, workspace, |channel, auth| async move {
let mut c = pb::release_client::ReleaseClient::with_interceptor(channel, auth);
let r = c
.gate_all(pb::RepoOnly {
repo: repo.to_string(),
})
.await
.context("Release.GateAll RPC")?
.into_inner();
Ok(GateReport {
repo: r.repo,
passed: r.passed,
failed: r.failed.into_iter().map(|f| (f.name, f.error)).collect(),
})
})
}
/// `Docs.Render` — server-side render of `repo`'s managed doc artifacts (runs
/// `cargo metadata` + the doc templates against the monitored checkout). Returns
/// `(status, detail)`. The server records a `docs_render` job for the run.
pub fn docs_render(
endpoint: &str,
token: &str,
repo: &str,
workspace: &str,
) -> Result<(String, String)> {
call(endpoint, token, workspace, |channel, auth| async move {
let mut c = pb::docs_client::DocsClient::with_interceptor(channel, auth);
let r = c
.render(pb::RepoOnly {
repo: repo.to_string(),
})
.await
.context("Docs.Render RPC")?
.into_inner();
Ok((r.status, r.detail))
})
}
/// `Mimir.SecurityScan` — server-side SBOM + vuln + license scan of `repo`
/// (scans the monitored `git/<repo>` checkout). Returns the raw JSON; the
/// Security tab parses it. No sources needed on the viz host.
pub fn security_scan(endpoint: &str, token: &str, repo: &str, workspace: &str) -> Result<String> {
call(endpoint, token, workspace, |channel, auth| async move {
let mut c = pb::mimir_client::MimirClient::with_interceptor(channel, auth);
let r = c
.security_scan(pb::RepoOnly {
repo: repo.to_string(),
})
.await
.context("Mimir.SecurityScan RPC")?
.into_inner();
Ok(r.json)
})
}
/// `Mimir.DepsOf` — repos `repo` depends on, as raw JSON. Server resolves the
/// graph from the checkout descriptor OR (monitored workspaces like njord) the
/// warehouse's `dep_graph_edges`. Lets a test assert njord's dep-graph resolves.
pub fn mimir_deps_of(
endpoint: &str,
token: &str,
repo: &str,
workspace: &str,
transitive: bool,
) -> Result<String> {
call(endpoint, token, workspace, |channel, auth| async move {
let mut c = pb::mimir_client::MimirClient::with_interceptor(channel, auth);
let r = c
.deps_of(pb::DepQuery {
repo: repo.to_string(),
transitive,
})
.await
.context("Mimir.DepsOf RPC")?
.into_inner();
Ok(r.json)
})
}
/// `Mimir.SourceWindow` — a windowed source read for the thin arch jump-to-source
/// preview: `context` lines each side of 1-based `line` from `file` inside the
/// server's checkout of `repo` (empty `repo` ⇒ the server tries each workspace
/// member and returns the first that contains `file`). Decodes the
/// `{start_line, lines}` JSON into the SAME `(start_line, lines)` tuple
/// [`crate::arch::source_window`] returns locally, so the remote and local arms of
/// the arch preview are interchangeable. An unreadable/absent file is
/// `(0, vec![])` — never an error (matching the local reader).
pub fn fetch_source_window(
endpoint: &str,
token: &str,
workspace: &str,
repo: &str,
file: &str,
line: u32,
context: u32,
) -> Result<(u32, Vec<String>)> {
let (repo, file) = (repo.to_string(), file.to_string());
call(endpoint, token, workspace, |channel, auth| async move {
let mut c = pb::mimir_client::MimirClient::with_interceptor(channel, auth);
let r = c
.source_window(pb::SourceWindowQuery {
repo,
file,
line,
context,
})
.await
.context("Mimir.SourceWindow RPC")?
.into_inner();
#[derive(serde::Deserialize)]
struct Win {
start_line: u32,
lines: Vec<String>,
}
let w: Win =
serde_json::from_str(&r.json).context("decode source window json from server")?;
Ok((w.start_line, w.lines))
})
}
/// `Release.Trace` — regression time-bisect JSON for `repo` (raw, UI pretty-prints).
pub fn trace(endpoint: &str, token: &str, repo: &str, workspace: &str) -> Result<String> {
let ws = workspace.to_string();
call(endpoint, token, workspace, |channel, auth| async move {
let mut c = pb::release_client::ReleaseClient::with_interceptor(channel, auth);
let r = c
.trace(pb::TraceQuery {
repo: repo.to_string(),
workspace: ws,
})
.await
.context("Release.Trace RPC")?
.into_inner();
Ok(r.json)
})
}
/// `Bench.History` — flatten every run's metrics into chartable points.
pub fn bench_history(
endpoint: &str,
token: &str,
repo: &str,
workspace: &str,
) -> Result<Vec<BenchPoint>> {
call(endpoint, token, workspace, |channel, auth| async move {
let mut c = pb::bench_client::BenchClient::with_interceptor(channel, auth);
let r = c
.history(pb::RepoOnly {
repo: repo.to_string(),
})
.await
.context("Bench.History RPC")?
.into_inner();
let mut pts = Vec::new();
for run in r.runs {
for res in run.results {
for kvf in res.metrics {
pts.push(BenchPoint {
date: run.date.clone(),
version: run.version.clone(),
metric: format!("{}::{}", res.name, kvf.key),
value: kvf.value,
});
}
}
}
Ok(pts)
})
}
/// One semantic-search hit (`Vector.Search`).
#[derive(Clone, Debug)]
pub struct VecHit {
pub score: f64,
pub file: String,
pub start_line: u64,
pub end_line: u64,
}
/// `Vector.Search` — GPU/CPU semantic search over a repo's embedded snapshot.
/// `repo` required; `sha` empty = latest. Returns hits or a clear error (the
/// server replies UNIMPLEMENTED when built without an embedder).
pub fn vector_search(
endpoint: &str,
token: &str,
repo: &str,
query: &str,
limit: u32,
workspace: &str,
) -> Result<Vec<VecHit>> {
call(endpoint, token, workspace, |channel, auth| async move {
let mut c = pb::vector_client::VectorClient::with_interceptor(channel, auth);
let r = c
.search(pb::VectorSearchRequest {
repo: repo.to_string(),
query: query.to_string(),
sha: String::new(),
limit,
})
.await
.context("Vector.Search RPC")?
.into_inner();
let v: serde_json::Value =
serde_json::from_str(&r.json).context("decode vector hits json")?;
let hits = v
.get("hits")
.and_then(|h| h.as_array())
.map(|arr| {
arr.iter()
.map(|h| VecHit {
score: h.get("score").and_then(|x| x.as_f64()).unwrap_or(0.0),
file: h
.get("file")
.and_then(|x| x.as_str())
.unwrap_or("")
.to_string(),
start_line: h.get("start_line").and_then(|x| x.as_u64()).unwrap_or(0),
end_line: h.get("end_line").and_then(|x| x.as_u64()).unwrap_or(0),
})
.collect()
})
.unwrap_or_default();
Ok(hits)
})
}
/// `Index.Stats` — `(total_docs, per-corpus counts)` for the search status panel.
pub fn index_stats(
endpoint: &str,
token: &str,
workspace: &str,
) -> Result<(u64, Vec<(String, String)>)> {
call(endpoint, token, workspace, |channel, auth| async move {
let mut c = pb::index_client::IndexClient::with_interceptor(channel, auth);
let r = c
.stats(pb::Empty {})
.await
.context("Index.Stats RPC")?
.into_inner();
Ok((
r.total,
r.by_corpus
.into_iter()
.map(|kv| (kv.key, kv.value))
.collect(),
))
})
}
/// `Knowledge.Callers` / `Callees` — who-calls / who-is-called-by `name`.
/// `callers = true` → Callers, else Callees. Returns `(caller_path, callee_ident,
/// file, line)` rows.
pub fn knowledge_calls(
endpoint: &str,
token: &str,
repo: &str,
name: &str,
callers: bool,
limit: u32,
workspace: &str,
) -> Result<Vec<(String, String, String, u32)>> {
call(endpoint, token, workspace, |channel, auth| async move {
let mut c = pb::knowledge_client::KnowledgeClient::with_interceptor(channel, auth);
let q = pb::KnowledgeCallQuery {
repo: repo.to_string(),
name: name.to_string(),
limit,
};
let calls = if callers {
c.callers(q).await.context("Knowledge.Callers RPC")?
} else {
c.callees(q).await.context("Knowledge.Callees RPC")?
}
.into_inner()
.calls;
Ok(calls
.into_iter()
.map(|k| (k.caller_path, k.callee_ident, k.file, k.line))
.collect())
})
}
/// `Knowledge.CallPath` — a call path from `from` to `to` (empty = none found).
pub fn knowledge_call_path(
endpoint: &str,
token: &str,
repo: &str,
from: &str,
to: &str,
workspace: &str,
) -> Result<Vec<String>> {
call(endpoint, token, workspace, |channel, auth| async move {
let mut c = pb::knowledge_client::KnowledgeClient::with_interceptor(channel, auth);
let r = c
.call_path(pb::KnowledgeCallPathQuery {
repo: repo.to_string(),
from: from.to_string(),
to: to.to_string(),
})
.await
.context("Knowledge.CallPath RPC")?
.into_inner();
Ok(r.names)
})
}
/// `Funnel.Show` — the whole idea→plan funnel, flattened into a render-ready
/// [`FunnelView`]. The `FunnelDump` already carries per-node `deps`, so the
/// DAG is fully reconstructable without any extra RPC.
// VIZ-ONLY: these two build `funnel_view` RENDER types, which are egui view
// models — the rest of this module is a plain gRPC client the CLI uses, and it
// must build for a server that links no window toolkit.
#[cfg(feature = "viz")]
pub fn funnel_show(
endpoint: &str,
token: &str,
workspace: &str,
) -> Result<super::funnel_view::FunnelView> {
use super::funnel_view::{FunnelView, NodeStat, NodeView, PlanView};
call(endpoint, token, workspace, |channel, auth| async move {
let mut c = pb::funnel_client::FunnelClient::with_interceptor(channel, auth);
let dump = c
.show(pb::Empty {})
.await
.context("Funnel.Show RPC")?
.into_inner();
let mut plans = Vec::new();
for idea in dump.ideas {
for plan in idea.plans {
let nodes = plan
.nodes
.into_iter()
.map(|n| NodeView {
id: n.id,
kind: n.kind,
title: n.title,
status: NodeStat::parse(&n.status),
targets: Vec::new(), // FunnelDumpNode carries no targets
deps: n.deps,
})
.collect();
plans.push(PlanView {
id: plan.id,
summary: plan.summary,
status: plan.status.to_ascii_lowercase(),
idea_text: idea.text.clone(),
nodes,
});
}
}
plans.sort_by(|a, b| a.id.cmp(&b.id));
Ok(FunnelView { plans })
})
}
/// Lenient wire→[`funnel_core::event::NodeStatus`] parse. Accepts the server's
/// snake_case serde token, the `Debug` repr (lower-cased), and the CLI synonyms
/// (`active` ⇒ in-progress, `abandoned` ⇒ failed). Unknown / `pending` ⇒
/// `Pending` (the safe default — an unrecognised status never panics).
fn node_status_from_wire(s: &str) -> funnel_core::event::NodeStatus {
use funnel_core::event::NodeStatus::*;
match s.trim().to_ascii_lowercase().as_str() {
"ready" => Ready,
"inprogress" | "in_progress" | "active" => InProgress,
"done" => Done,
"blocked" => Blocked,
"failed" | "abandoned" => Failed,
_ => Pending,
}
}
/// Lenient wire→[`funnel_core::event::PlanStatus`] parse (the dump carries the
/// `Debug` repr, which `funnel_show` lower-cases). Unknown ⇒ `Draft`.
fn plan_status_from_wire(s: &str) -> funnel_core::event::PlanStatus {
use funnel_core::event::PlanStatus::*;
match s.trim().to_ascii_lowercase().as_str() {
"active" => Active,
"done" => Done,
"abandoned" => Abandoned,
_ => Draft,
}
}
/// **Reconstruct a [`funnel_core::state::Funnel`] from a `Funnel.Show` dump.**
///
/// Pure (no network) so it is unit-testable: it maps the render-ready
/// `FunnelDump` (ideas → plans → nodes, each node carrying its `deps`) back onto
/// the in-memory `Funnel` the pure DAG projections
/// ([`funnel_core::plan_dag::PlanDag::from_funnel`], the 3-D cloud, the card
/// board) consume. This is what lets the **thin/remote** viz path render the
/// same metric-bearing Plan-DAG the local path builds from the on-disk `Store`
/// — the dump already carries everything the graph needs, so no extra RPC or
/// proto change is required (the `Funnel.Show` doc-comment's long-standing
/// promise, finally wired).
///
/// Fidelity notes (additive, never a failure): the dump carries no timestamps,
/// so every card reconstructs with `now` (staleness collapses to a uniform 0 —
/// the topology, statuses, and edges are exact, only the age-heat is flat);
/// `refs`/assignments/lanes are absent from the wire and default empty.
fn funnel_from_dump(dump: &pb::FunnelDump) -> funnel_core::state::Funnel {
use funnel_core::event::ItemKind;
use funnel_core::ids::{IdeaId, NodeId, PlanId};
use funnel_core::state::{Funnel, Idea, Plan, PlanNode};
use std::collections::{BTreeMap, BTreeSet};
let now = chrono::Utc::now();
let mut f = Funnel::default();
for idea in &dump.ideas {
let iid = IdeaId::new(&idea.id);
f.ideas.insert(
iid.clone(),
Idea {
id: iid.clone(),
source: idea.source.clone(),
text: idea.text.clone(),
refs: Vec::new(),
item_kind: ItemKind::parse(&idea.item_kind),
submitted_at: now,
triage: None,
mode: None,
},
);
for plan in &idea.plans {
let pid = PlanId::new(&plan.id);
// The plan's idea_id: prefer the dump's explicit field, fall back to
// the enclosing idea (they agree in a well-formed dump).
let plan_idea = if plan.idea_id.is_empty() {
iid.clone()
} else {
IdeaId::new(&plan.idea_id)
};
let mut nodes: BTreeMap<NodeId, PlanNode> = BTreeMap::new();
for n in &plan.nodes {
let nid = NodeId::new(&n.id);
let mut params = serde_json::Map::new();
if !n.title.is_empty() {
params.insert("title".into(), serde_json::json!(n.title));
}
nodes.insert(
nid.clone(),
PlanNode {
id: nid.clone(),
kind: n.kind.clone(),
params,
targets: Vec::new(),
prompt_excerpt: (!n.title.is_empty()).then(|| n.title.clone()),
created_at: now,
status: node_status_from_wire(&n.status),
last_status_change: now,
},
);
}
// Edges: each node lists its upstream `deps` (dep → node). Skip any
// edge whose endpoints aren't both present (defensive — the pure
// projections tolerate a partial dump, never panic).
let mut edges: BTreeSet<(NodeId, NodeId)> = BTreeSet::new();
let present: BTreeSet<&str> = plan.nodes.iter().map(|n| n.id.as_str()).collect();
for n in &plan.nodes {
let to = NodeId::new(&n.id);
for d in &n.deps {
if present.contains(d.as_str()) && d != &n.id {
edges.insert((NodeId::new(d), to.clone()));
}
}
}
f.plans.insert(
pid.clone(),
Plan {
id: pid,
idea_id: plan_idea,
summary: plan.summary.clone(),
planner: "server".into(),
created_at: now,
status: plan_status_from_wire(&plan.status),
nodes,
edges,
},
);
}
}
f
}
/// `Funnel.Show` → a reconstructed [`funnel_core::state::Funnel`] (the thin-path
/// twin of opening the on-disk funnel `Store`). Fetches the dump once and hands
/// it to [`funnel_from_dump`]; callers then run any pure projection (Plan-DAG,
/// 3-D cloud, cards) exactly as the local path does. Readiness is derived
/// (`promote_ready`) so the remote DAG shows the same Pending→Ready frontier.
pub fn funnel_reconstruct(
endpoint: &str,
token: &str,
workspace: &str,
) -> Result<funnel_core::state::Funnel> {
call(endpoint, token, workspace, |channel, auth| async move {
let mut c = pb::funnel_client::FunnelClient::with_interceptor(channel, auth);
let dump = c
.show(pb::Empty {})
.await
.context("Funnel.Show RPC")?
.into_inner();
let mut f = funnel_from_dump(&dump);
f.promote_ready();
Ok(f)
})
}
/// FI1 — paste an idea/error into the server's funnel (`Funnel.SubmitIdea`).
/// `item_kind` is `"idea"` or `"error"` (empty ⇒ idea). Returns the new item id.
pub fn funnel_submit(
endpoint: &str,
token: &str,
workspace: &str,
text: &str,
item_kind: &str,
source: &str,
) -> Result<String> {
call(endpoint, token, workspace, |channel, auth| async move {
let mut c = pb::funnel_client::FunnelClient::with_interceptor(channel, auth);
let r = c
.submit_idea(pb::SubmitIdeaRequest {
text: text.to_string(),
source: source.to_string(),
item_kind: item_kind.to_string(),
})
.await
.context("Funnel.SubmitIdea RPC")?
.into_inner();
Ok(r.id)
})
}
/// FI2 — fetch the server-side intake history (`Funnel.History`), newest-first,
/// optionally filtered by kind/status (empty = no filter).
// VIZ-ONLY: these two build `funnel_view` RENDER types, which are egui view
// models — the rest of this module is a plain gRPC client the CLI uses, and it
// must build for a server that links no window toolkit.
#[cfg(feature = "viz")]
pub fn funnel_history(
endpoint: &str,
token: &str,
workspace: &str,
kind: &str,
status: &str,
) -> Result<Vec<super::funnel_view::HistoryItemView>> {
use super::funnel_view::HistoryItemView;
call(endpoint, token, workspace, |channel, auth| async move {
let mut c = pb::funnel_client::FunnelClient::with_interceptor(channel, auth);
let resp = c
.history(pb::FunnelHistoryRequest {
kind: kind.to_string(),
status: status.to_string(),
limit: 0,
})
.await
.context("Funnel.History RPC")?
.into_inner();
Ok(resp
.items
.into_iter()
.map(|it| HistoryItemView {
id: it.id,
item_kind: it.item_kind,
text: it.text,
source: it.source,
submitted_at: it.submitted_at,
status: it.status,
plan_ids: it.plan_ids,
})
.collect())
})
}
#[cfg(test)]
mod tests {
use super::*;
/// PURE (no network): a `Funnel.Show` dump reconstructs into a
/// `funnel_core::Funnel` faithful enough that the metric-bearing Plan-DAG
/// projects the identical topology the local path would. This is the crux of
/// the thin/remote Plan-DAG wiring — real server data → real graph, not a
/// placeholder. Builds an idea → plan with a 3-node dependency chain
/// (n-001 → n-002 → n-003) and asserts the reconstructed edges/statuses.
#[test]
fn dump_reconstructs_into_a_projectable_funnel() {
let dump = pb::FunnelDump {
ideas: vec![pb::FunnelDumpIdea {
id: "i-001".into(),
text: "build the planner".into(),
source: "human:alice".into(),
item_kind: "idea".into(),
plans: vec![pb::FunnelDumpPlan {
id: "p-001".into(),
idea_id: "i-001".into(),
summary: "decompose".into(),
status: "Active".into(), // server ships the Debug repr
nodes: vec![
pb::FunnelDumpNode {
id: "n-001".into(),
kind: "code:write".into(),
title: "implement foo".into(),
status: "done".into(),
deps: vec![],
},
pb::FunnelDumpNode {
id: "n-002".into(),
kind: "code:fix".into(),
title: "implement bar".into(),
status: "in_progress".into(),
deps: vec!["n-001".into()],
},
pb::FunnelDumpNode {
id: "n-003".into(),
kind: "test:write".into(),
title: "test baz".into(),
status: "pending".into(),
deps: vec!["n-002".into(), "n-001".into()],
},
],
}],
}],
};
let f = funnel_from_dump(&dump);
assert_eq!(f.ideas.len(), 1);
assert_eq!(f.plans.len(), 1);
let plan = f.plans.values().next().unwrap();
assert_eq!(plan.status, funnel_core::event::PlanStatus::Active);
assert_eq!(plan.nodes.len(), 3);
// n-001 done, n-002 in-progress, n-003 pending — statuses survive the wire.
use funnel_core::ids::NodeId;
assert_eq!(
plan.nodes[&NodeId::new("n-001")].status,
funnel_core::event::NodeStatus::Done
);
assert_eq!(
plan.nodes[&NodeId::new("n-002")].status,
funnel_core::event::NodeStatus::InProgress
);
// 3 dep edges: n-001→n-002, n-002→n-003, n-001→n-003.
assert_eq!(plan.edges.len(), 3);
// And the pure Plan-DAG projects it: mother idea + 3 task nodes, with
// the 3 mother edges (idea→each node) plus the 3 dependency edges.
let dag = funnel_core::plan_dag::PlanDag::from_funnel(&f, None);
assert_eq!(dag.node_count(), 4, "mother + 3 task cards");
assert_eq!(dag.dep_edges(), 3, "the reconstructed dependency edges");
assert_eq!(
dag.mother_edges(),
3,
"one decomposition edge per task node"
);
// Acyclic ⇒ a topo order exists with deps before dependents.
let order = dag.topo_order().expect("acyclic");
let pos = |id: &str| order.iter().position(|x| x == id).unwrap();
assert!(pos("n-001") < pos("n-002"));
assert!(pos("n-002") < pos("n-003"));
}
/// A partial/edge-to-missing-node dump never panics — a defensive
/// reconstruction drops the dangling edge and keeps the valid nodes.
#[test]
fn dump_with_dangling_dep_is_tolerated() {
let dump = pb::FunnelDump {
ideas: vec![pb::FunnelDumpIdea {
id: "i-002".into(),
text: "x".into(),
source: "".into(),
item_kind: "".into(), // empty ⇒ idea
plans: vec![pb::FunnelDumpPlan {
id: "p-002".into(),
idea_id: "".into(), // empty ⇒ falls back to the enclosing idea
summary: "s".into(),
status: "weird-unknown".into(), // ⇒ Draft
nodes: vec![pb::FunnelDumpNode {
id: "n-010".into(),
kind: "code:write".into(),
title: "".into(),
status: "".into(), // ⇒ Pending
deps: vec!["n-999".into()], // missing endpoint
}],
}],
}],
};
let f = funnel_from_dump(&dump);
let plan = f.plans.values().next().unwrap();
assert_eq!(plan.idea_id.as_str(), "i-002", "empty idea_id falls back");
assert_eq!(plan.status, funnel_core::event::PlanStatus::Draft);
assert!(plan.edges.is_empty(), "the dangling n-999 edge is dropped");
assert_eq!(plan.nodes.len(), 1);
}
/// Live smoke against `$NORNIR_SERVER` (+ `$NORNIR_SERVER_TOKEN`). Ignored by
/// default (needs a running server). Run with:
/// NORNIR_SERVER=http://oden:7878 NORNIR_SERVER_TOKEN=… \
/// cargo test --features viz --lib viz::remote::tests -- --ignored --nocapture
#[test]
#[ignore = "live: needs NORNIR_SERVER + NORNIR_SERVER_TOKEN pointing at a running server"]
fn live_list_workspaces() {
let ep = std::env::var("NORNIR_SERVER").expect("set NORNIR_SERVER");
let tok = std::env::var("NORNIR_SERVER_TOKEN").expect("set NORNIR_SERVER_TOKEN");
let ws = list_workspaces(&ep, &tok).expect("list_workspaces should succeed");
eprintln!("live workspaces: {ws:?}");
assert!(!ws.is_empty(), "server returned no workspaces");
}
/// Wire-contract test for `Viz.ReleaseEvents`: the server serializes
/// `Vec<ReleaseEventRow>` exactly as `nornir-server`'s `release_events`
/// handler does (`serde_json::to_string(&rows)`); the client's
/// `fetch_release_events` decodes the response JSON with
/// `serde_json::from_str`. This asserts that round-trip preserves EVERY field
/// (so the remote Release tab renders the identical rows) — LAW: real data in,
/// real rows out, not just "didn't panic".
#[test]
fn release_events_wire_roundtrip() {
use crate::warehouse::release_events::{ReleaseEventRow, status};
let rows = vec![
ReleaseEventRow {
run_id: "run-42".into(),
seq: 0,
ts_micros: 1_000,
component: "znippy".into(),
repo: "znippy".into(),
op: "test".into(),
phase: "start".into(),
status: status::RUNNING.into(),
detail: String::new(),
depends_on: Some(vec![]),
elapsed_ms: None,
},
ReleaseEventRow {
run_id: "run-42".into(),
seq: 1,
ts_micros: 2_000,
component: "holger".into(),
repo: "holger".into(),
op: "gate".into(),
phase: "end".into(),
status: status::OK.into(),
detail: "all gates green".into(),
depends_on: Some(vec!["znippy".into()]),
elapsed_ms: Some(1234),
},
];
// Server side: exactly what the handler emits.
let json = serde_json::to_string(&rows).unwrap();
// Client side: exactly what fetch_release_events decodes.
let back: Vec<ReleaseEventRow> = serde_json::from_str(&json).unwrap();
assert_eq!(
back, rows,
"release_events wire round-trip must be lossless"
);
// And the non-empty content is preserved (real values asserted).
assert_eq!(
back[1].depends_on.as_deref(),
Some(&["znippy".to_string()][..])
);
assert_eq!(back[1].elapsed_ms, Some(1234));
assert_eq!(back[1].detail, "all gates green");
}
/// Wire-contract test for `Viz.CloneEvents` (PLAN #6): the server serializes
/// `Vec<CloneEventRow>` exactly as the `clone_events` handler does
/// (`serde_json::to_string(&rows)`); `fetch_clone_events` decodes with
/// `serde_json::from_str`. Asserts the round-trip preserves the failed member +
/// its error detail so the remote 🧬 nornir pane shows *why* a member didn't
/// populate — real data in, real rows out.
#[test]
fn clone_events_wire_roundtrip() {
use crate::warehouse::clone_events::{CloneEventRow, op, status};
let rows = vec![
CloneEventRow {
ts_micros: 2_000,
workspace: "nordisk".into(),
member: "korp".into(),
remote: "https://github.com/nordisk/korp".into(),
op: op::CLONE_FETCH.into(),
status: status::ERROR.into(),
detail: "clone-fetch https://github.com/nordisk/korp: Couldn't obtain Username"
.into(),
elapsed_ms: 7,
},
CloneEventRow {
ts_micros: 1_000,
workspace: "nordisk".into(),
member: "facett".into(),
remote: "git@github.com:nordisk/facett.git".into(),
op: op::CLONE_FETCH.into(),
status: status::OK.into(),
detail: "abc123".into(),
elapsed_ms: 42,
},
];
// Server side: exactly what the handler emits.
let json = serde_json::to_string(&rows).unwrap();
// Client side: exactly what fetch_clone_events decodes.
let back: Vec<CloneEventRow> = serde_json::from_str(&json).unwrap();
assert_eq!(back, rows, "clone_events wire round-trip must be lossless");
// The failed member + its error chain survive the round-trip (real values).
assert_eq!(back[0].member, "korp");
assert_eq!(back[0].status, "error");
assert!(back[0].detail.contains("Couldn't obtain Username"));
assert_eq!(back[1].member, "facett");
assert_eq!(back[1].detail, "abc123");
}
/// Wire-contract test for `Viz.TestResults`: the server serializes
/// `Vec<TestResultRow>` exactly as `nornir-server`'s `test_results` handler
/// does (`serde_json::to_string(&rows)`); the client's `fetch_test_results`
/// decodes the response JSON with `serde_json::from_str`. This asserts that
/// the round-trip preserves EVERY field so the remote 🧪 Test tab renders
/// the identical matrix — LAW: real data in, real rows out, not just "didn't panic".
#[test]
fn test_results_wire_roundtrip() {
use crate::warehouse::test_results::{TestResultRow, status};
let rows = vec![
TestResultRow {
run_id: "run-t1".into(),
ts_micros: 1_000,
repo: "znippy".into(),
suite: "unit".into(),
test_name: "compress_roundtrip".into(),
status: status::PASS.into(),
duration_ms: 12.0,
message: String::new(),
aspect: "unit".into(),
metric: 0.0,
},
TestResultRow {
run_id: "run-t1".into(),
ts_micros: 2_000,
repo: "znippy".into(),
suite: "clippy".into(),
test_name: "clippy::all".into(),
status: status::FAIL.into(),
duration_ms: 88.0,
message: "error[E0277]: ...".into(),
aspect: "clippy".into(),
metric: 3.0,
},
];
// Server side: exactly what the handler emits.
let json = serde_json::to_string(&rows).unwrap();
// Client side: exactly what fetch_test_results decodes.
let back: Vec<TestResultRow> = serde_json::from_str(&json).unwrap();
assert_eq!(back, rows, "test_results wire round-trip must be lossless");
assert_eq!(back[0].status, status::PASS);
assert_eq!(back[0].duration_ms, 12.0);
assert_eq!(back[1].status, status::FAIL);
assert_eq!(back[1].message, "error[E0277]: ...");
assert_eq!(back[1].metric, 3.0);
}
/// Wire-contract test for `Viz.BenchTelemetry`: the server serializes the two
/// tables as `{telemetry:[BenchTelemetryRow], runs:[BenchRun]}` exactly as
/// `nornir-server`'s `bench_telemetry` handler does; the client's
/// `fetch_bench_live` decodes both arrays back out. This asserts the round-trip
/// preserves EVERY field of BOTH tables so the remote 📡 Bench LIVE panel folds
/// identically to embedded — LAW: real data in, real rows out.
#[test]
fn bench_telemetry_wire_roundtrip() {
use crate::bench::{BenchRun, TestOutcome};
use crate::warehouse::iceberg::BenchTelemetryRow;
let telemetry = vec![BenchTelemetryRow {
run_id: "run-b1".into(),
repo: "znippy".into(),
bench: "znippy.compress".into(),
n_cores: 8,
cpu_pct_avg: 51.5,
cpu_pct_max: 92.0,
cores_busy_avg: 6.5,
cores_busy_max: 8,
mem_peak_mb: 256.0,
mem_pct_max: 30.0,
elapsed_ms: 1234.0,
proc_cpu_ms: Some(7900.0),
child_cpu_ms: Some(50.0),
proc_cores_avg: Some(6.44),
proc_cores_max: Some(7.10),
}];
// A row from before the process-scoped columns existed: it must survive
// the same hop with `None`s intact, or a viewer/gate on the far side
// sees a fabricated 0.0 where nothing was measured.
let legacy = vec![BenchTelemetryRow {
proc_cpu_ms: None,
child_cpu_ms: None,
proc_cores_avg: None,
proc_cores_max: None,
..telemetry[0].clone()
}];
let runs = vec![BenchRun {
id: BenchRun::new_id(),
machine_facts: None,
date: "2026-06-14".into(),
timestamp: None,
version: "1.0".into(),
machine: "host".into(),
cores: 8,
results: Vec::new(),
tests: vec![TestOutcome {
name: "compress".into(),
passed: false,
duration_ms: Some(12.0),
message: None,
}],
}];
// Server side: exactly what the handler emits.
let json = serde_json::json!({ "telemetry": telemetry, "runs": runs }).to_string();
// Client side: exactly what fetch_bench_live decodes.
let v: serde_json::Value = serde_json::from_str(&json).unwrap();
let back_telem: Vec<BenchTelemetryRow> =
serde_json::from_value(v.get("telemetry").cloned().unwrap()).unwrap();
let back_runs: Vec<BenchRun> =
serde_json::from_value(v.get("runs").cloned().unwrap()).unwrap();
assert_eq!(
back_telem, telemetry,
"bench_telemetry wire round-trip must be lossless"
);
// Real field values from the telemetry table.
assert_eq!(back_telem[0].bench, "znippy.compress");
assert_eq!(back_telem[0].n_cores, 8);
assert_eq!(back_telem[0].cores_busy_max, 8);
assert!((back_telem[0].cores_busy_avg - 6.5).abs() < 1e-9);
assert!((back_telem[0].mem_peak_mb - 256.0).abs() < 1e-9);
// The process-scoped columns — the ones a saturation verdict is made
// from — must cross the wire too, in BOTH states.
assert!((back_telem[0].proc_cores_max.unwrap() - 7.10).abs() < 1e-9);
assert!((back_telem[0].proc_cores_avg.unwrap() - 6.44).abs() < 1e-9);
let legacy_json = serde_json::json!({ "telemetry": legacy }).to_string();
let lv: serde_json::Value = serde_json::from_str(&legacy_json).unwrap();
let back_legacy: Vec<BenchTelemetryRow> =
serde_json::from_value(lv.get("telemetry").cloned().unwrap()).unwrap();
assert_eq!(
back_legacy, legacy,
"legacy row must round-trip losslessly too"
);
assert_eq!(
back_legacy[0].proc_cores_max, None,
"absent must not become 0.0"
);
// And a payload from an OLD server, which omits the keys entirely,
// still decodes (serde default) rather than failing the whole fetch.
let old_server = r#"[{"run_id":"r","repo":"znippy","bench":"b","n_cores":8,
"cpu_pct_avg":1.0,"cpu_pct_max":2.0,"cores_busy_avg":3.0,"cores_busy_max":8,
"mem_peak_mb":4.0,"mem_pct_max":5.0,"elapsed_ms":6.0}]"#;
let back_old: Vec<BenchTelemetryRow> = serde_json::from_str(old_server)
.expect("a payload without the new keys must still decode");
assert_eq!(back_old[0].proc_cores_max, None);
// Real field values from the runs table (BenchRun has no PartialEq).
assert_eq!(back_runs.len(), 1);
assert_eq!(back_runs[0].date, "2026-06-14");
assert_eq!(back_runs[0].cores, 8);
assert_eq!(back_runs[0].tests.len(), 1);
assert_eq!(back_runs[0].tests[0].name, "compress");
assert!(
!back_runs[0].tests[0].passed,
"failed test survives the wire"
);
}
/// Wire-contract test for `Viz.BakeoffResults`: same round-trip discipline for
/// `Vec<AgentModelRunRow>` — every metric field must survive so the remote
/// Leaderboard ranks identically to embedded.
#[test]
fn bakeoff_results_wire_roundtrip() {
use crate::warehouse::agent_model_runs::AgentModelRunRow;
let rows = vec![AgentModelRunRow {
run_id: "bake-1".into(),
ts_micros: 9_000,
agent: "claude".into(),
model: "mistral".into(),
prompt_id: "p-sum".into(),
prompt: "summarize".into(),
output: "ok".into(),
latency_ms: 120.5,
tokens_in: 10,
tokens_out: 42,
tokens_per_s: 88.3,
score: 0.91,
ok: true,
error: None,
cost_usd: 0.0123,
mcp_tool_calls: 5,
}];
let json = serde_json::to_string(&rows).unwrap();
let back: Vec<AgentModelRunRow> = serde_json::from_str(&json).unwrap();
assert_eq!(back, rows, "bakeoff wire round-trip must be lossless");
assert_eq!(back[0].tokens_per_s, 88.3);
assert_eq!(back[0].score, 0.91);
// The matrix axis + economics survive the wire (so the remote grid renders).
assert_eq!(back[0].agent, "claude");
assert_eq!(back[0].cost_usd, 0.0123);
assert_eq!(back[0].mcp_tool_calls, 5);
// A pre-matrix server's JSON (no agent/cost/mcp fields) still decodes,
// defaulting the agent to AGENT_UNSET (serde defaults).
let legacy = r#"[{"run_id":"r","ts_micros":1,"model":"m","prompt_id":"p","prompt":"q","output":"o","latency_ms":1.0,"tokens_in":1,"tokens_out":1,"tokens_per_s":1.0,"score":0.5,"ok":true,"error":null}]"#;
let decoded: Vec<AgentModelRunRow> = serde_json::from_str(legacy).unwrap();
assert_eq!(
decoded[0].agent, "-",
"legacy row defaults agent to AGENT_UNSET"
);
assert_eq!(decoded[0].cost_usd, 0.0);
assert_eq!(decoded[0].mcp_tool_calls, 0);
}
/// Wire-contract test for `Viz.Knowledge`: the server builds the summary with
/// `scan_summary_json`; the client decodes it into a `KnowledgeSummary`. Assert
/// the bubbles + per-repo rollups survive (so the remote map renders crates>0).
#[test]
fn knowledge_summary_wire_roundtrip() {
use super::super::knowledge::{Bubble, KnowledgeSummary, RepoScanSummary};
let summary = KnowledgeSummary {
repos: vec![RepoScanSummary {
repo: "znippy".into(),
ok: true,
symbols: 120,
calls: 88,
features: 3,
git_files: 40,
error: None,
}],
crates: vec![Bubble {
repo: "znippy".into(),
krate: "znippy".into(),
symbols: 120,
calls: 88,
files: 40,
gates: 3,
heat_30d: 7,
}],
..Default::default()
};
let json = serde_json::to_string(&summary).unwrap();
let back: KnowledgeSummary = serde_json::from_str(&json).unwrap();
assert_eq!(back.crates.len(), 1);
assert_eq!(back.crates[0].symbols, 120);
assert_eq!(back.repos[0].calls, 88);
assert!(back.repos[0].ok);
}
/// Live `Viz.Architecture` smoke (ignored) — the server-side merged board.
#[test]
#[ignore = "live: needs NORNIR_SERVER + NORNIR_SERVER_TOKEN pointing at a running server"]
fn live_architecture() {
let ep = std::env::var("NORNIR_SERVER").expect("set NORNIR_SERVER");
let tok = std::env::var("NORNIR_SERVER_TOKEN").expect("set NORNIR_SERVER_TOKEN");
let ws = std::env::var("NORNIR_WS").unwrap_or_else(|_| "holger".into());
let p = fetch_architecture(&ep, &tok, &ws).expect("fetch_architecture");
eprintln!(
"live architecture ws={ws}: {} nodes / {} edges, members={:?}",
p.graph.nodes.len(),
p.graph.edges.len(),
p.members_with_data
);
}
/// Live `Viz.ReleaseEvents` smoke for a workspace (default `holger`). Ignored
/// (needs a running server). Asserts the remote Release tab would have rows.
#[test]
#[ignore = "live: needs NORNIR_SERVER + NORNIR_SERVER_TOKEN pointing at a running server"]
fn live_release_events() {
let ep = std::env::var("NORNIR_SERVER").expect("set NORNIR_SERVER");
let tok = std::env::var("NORNIR_SERVER_TOKEN").expect("set NORNIR_SERVER_TOKEN");
let ws = std::env::var("NORNIR_WS").unwrap_or_else(|_| "holger".into());
let rows = fetch_release_events(&ep, &tok, &ws).expect("fetch_release_events");
eprintln!("live release_events ws={ws}: {} rows", rows.len());
}
/// Live `Viz.BakeoffResults` smoke (ignored).
#[test]
#[ignore = "live: needs NORNIR_SERVER + NORNIR_SERVER_TOKEN pointing at a running server"]
fn live_bakeoff_results() {
let ep = std::env::var("NORNIR_SERVER").expect("set NORNIR_SERVER");
let tok = std::env::var("NORNIR_SERVER_TOKEN").expect("set NORNIR_SERVER_TOKEN");
let ws = std::env::var("NORNIR_WS").unwrap_or_else(|_| "holger".into());
let rows = fetch_bakeoff_results(&ep, &tok, &ws).expect("fetch_bakeoff_results");
eprintln!("live agent_model_runs ws={ws}: {} rows", rows.len());
}
/// Live FI1+FI2 round-trip: submit an `error` item then read it back via
/// `Funnel.History`. Exercises the server's new History RPC + the
/// on-write `funnel_events` schema evolution against a real server.
#[test]
#[ignore = "live: needs NORNIR_SERVER + NORNIR_SERVER_TOKEN pointing at a running server"]
fn live_funnel_intake_and_history() {
let ep = std::env::var("NORNIR_SERVER").expect("set NORNIR_SERVER");
let tok = std::env::var("NORNIR_SERVER_TOKEN").expect("set NORNIR_SERVER_TOKEN");
let ws = std::env::var("NORNIR_WS").unwrap_or_else(|_| "znippy".into());
let marker = format!(
"funnel-intake live test {}",
chrono::Utc::now().to_rfc3339()
);
let id = funnel_submit(&ep, &tok, &ws, &marker, "error", "live-test")
.expect("Funnel.SubmitIdea(error) must succeed");
eprintln!("submitted error id={id} ws={ws}");
let errs =
funnel_history(&ep, &tok, &ws, "error", "").expect("Funnel.History must succeed");
let mine = errs
.iter()
.find(|it| it.id == id)
.expect("submitted error in history");
assert_eq!(mine.item_kind, "error");
assert_eq!(mine.text, marker);
eprintln!(
"history error_count={} (found mine, kind={})",
errs.len(),
mine.item_kind
);
}
/// Live `Funnel.Show` smoke for a workspace (default `holger`). Captures the
/// actual RPC outcome — must NOT panic, returns an empty-or-populated view.
#[test]
#[ignore = "live: needs NORNIR_SERVER + NORNIR_SERVER_TOKEN pointing at a running server"]
fn live_funnel_show() {
let ep = std::env::var("NORNIR_SERVER").expect("set NORNIR_SERVER");
let tok = std::env::var("NORNIR_SERVER_TOKEN").expect("set NORNIR_SERVER_TOKEN");
let ws = std::env::var("NORNIR_WS").unwrap_or_else(|_| "holger".into());
match funnel_show(&ep, &tok, &ws) {
Ok(v) => eprintln!("funnel_show ws={ws} OK: {} plans", v.plans.len()),
Err(e) => {
eprintln!("funnel_show ws={ws} ERR: {e:#}");
panic!("funnel_show failed: {e:#}");
}
}
}
}
/// Convert a proto `ProgressEvent` (the oneof) into the viz [`LiveEvent`] —
/// the same enum the local NDJSON tail deserializes, so the pane is wire-source
/// agnostic. Unknown/empty kinds are dropped (forward-compat).
fn to_live(ev: pb::ProgressEvent) -> Option<LiveEvent> {
use pb::progress_event::Kind;
Some(match ev.kind? {
Kind::RunStart(x) => LiveEvent::RunStart {
run_id: x.run_id,
workspace: x.workspace,
},
Kind::RepoStart(x) => LiveEvent::RepoStart {
repo: x.repo,
sha: x.sha,
},
Kind::PhaseStart(x) => LiveEvent::PhaseStart {
repo: x.repo,
phase: x.phase,
},
Kind::PhaseEnd(x) => LiveEvent::PhaseEnd {
repo: x.repo,
phase: x.phase,
ok: x.ok,
duration_ms: x.duration_ms,
},
Kind::BinaryStart(x) => LiveEvent::BinaryStart {
repo: x.repo,
binary: x.binary,
},
Kind::TestPass(x) => LiveEvent::TestPass {
repo: x.repo,
binary: x.binary,
name: x.name,
},
Kind::TestFail(x) => LiveEvent::TestFail {
repo: x.repo,
binary: x.binary,
name: x.name,
},
Kind::BinaryDone(x) => LiveEvent::BinaryDone {
repo: x.repo,
binary: x.binary,
passed: x.passed,
failed: x.failed,
},
Kind::RepoEnd(x) => LiveEvent::RepoEnd {
repo: x.repo,
ok: x.ok,
},
Kind::RunEnd(x) => LiveEvent::RunEnd {
run_id: x.run_id,
ok: x.ok,
},
})
}