crates/dwarves-embed/src/lib.rs
//! **dwarves-embed** — the LIVE forge run as an embeddable `Panel` (backend (2)).
//!
//! The reusable `<guest>-embed` adapter shape (nornir-funnel-embed is the template):
//! a **thin tonic gRPC client** fetches one run's full report from a running
//! `dwarves-server`'s read-only [`Dwarf.GetRunReport`](pb) RPC, **reconstructs** a
//! [`dwarves::brokkr::engine::RunReport`] + its build [`Dag`] client-side, and renders
//! it on the shared [`facett_pipeline::PipelineView`] via the existing
//! [`ForgePipelineAdapter::from_report`] — the SAME reused DAG viewer the in-process
//! demo Pipeline tab uses. Steps → nodes, `needs` → edges, per-step status → node
//! status, build durations → edge labels.
//!
//! **No `nornir` dependency, no `dwarves` server feature.** nornir already depends on
//! dwarves, so a `dwarves-embed → nornir` edge would be a cycle; and pulling dwarves'
//! `server` feature would build the whole tonic router + skidbladnir lifecycle. Instead
//! this crate compiles ONLY dwarves' `GetRunReport`/`History` proto (a trimmed copy
//! under `proto/`, package `dwarf.v1`, so the gRPC method paths
//! `/dwarf.v1.Dwarf/GetRunReport` + `/dwarf.v1.Dwarf/History` match the live server)
//! and reconstructs against the `dwarves` domain types (pulled lean with only the
//! `facett-adapter` feature) + the generated `pb` types directly.
//!
//! ## The seam
//! [`panel`] returns a `Box<dyn Panel>` a host drops into a tab. It connects to the
//! given endpoint (default `http://127.0.0.1:7900`) and, if the server is unreachable
//! (or has no runs yet), shows a clean *"forge unavailable"* state — it never panics.
//! With no pinned run id it renders the NEWEST run; auto-poll re-fetches so a run in
//! flight fills in its steps live. A **Refresh** action re-fetches.
use std::time::Duration;
use anyhow::{anyhow, Context, Result};
use dwarves::brokkr::dag::{Dag, DagNode};
use dwarves::brokkr::engine::RunReport;
use dwarves::brokkr::facett_adapter::ForgePipelineAdapter;
use dwarves::brokkr::history::{BuildRow, PipelineRunRow, RunStatus};
use facett_core::{Facet, Panel};
use facett_pipeline::PipelineView;
/// The generated thin client + wire types for dwarves' `GetRunReport` + `History` +
/// `AgentPool`.
pub mod pb {
tonic::include_proto!("dwarf.v1");
}
/// The UI's leg of the forge's 🧭 Explorer (`dwarf.v1.Explorer`): the project tree, a
/// node's detail, and the actions — each one short call, answered at once.
pub mod explorer;
/// The UI's leg of the forge's 🚀 Mímir release plan (`dwarf.v1.MimirRelease.Plan`).
pub mod mimir_release;
/// The UI's leg of the forge's 🔬 Profile an app (`dwarf.v1.TuneApp`): Start · Status · Approve.
pub mod tune_app;
mod cicdwizard;
pub use cicdwizard::{cicd_wizard_panel, CicdWizardPanel};
mod jobsfleet;
pub use jobsfleet::{
agents_from_wire, default_ledger_root, entries_from_records, entries_from_records_json,
fetch_agents, jobs_fleet_panel, read_ledger_jobs, JobsFleetPanel, LedgerRead,
DEFAULT_JOBS_POLL_INTERVAL_S, DEFAULT_LEDGER_LIMIT,
};
mod holgerbuild;
pub use holgerbuild::{
fetch_holger_build, holger_build_panel, holger_build_panel_with, tiers_from_wire, BuildErr,
HolgerBuildPanel, TierRow, DEFAULT_BUILD_POLL_INTERVAL_S, HOLGER_ROOT_ENV, MATRIX_COLUMNS,
};
mod releaseview;
// `release_fixture` is NOT re-exported any more (2026-09-10): it became `#[cfg(test)]`,
// because it was what the shipped Release pane drew with no release run on disk —
// four real crates with invented versions, two of them red "needs-release".
pub use releaseview::{
default_convergence_path, load_convergence_at, panel_from_convergence, panel_from_json,
release_panel, release_panel_at, workspace_convergence_path,
};
mod benchesview;
// `bench_fixture` + `testmatrix_fixture` are NOT re-exported any more (2026-09-10):
// they became `#[cfg(test)]`, because they were what the shipped pane drew when the
// workspace's files were missing. A missing file now yields a NAMED empty state.
pub use benchesview::{
append_matrix_rows, benches_panel, benches_panel_at, mimir_series, robot_ui_rows, run_robot_ui,
workspace_bench_history_path, workspace_testmatrix_path, BenchSeries, BenchesView,
DWARVES_ASPECTS, MIMIR_PREFIX,
};
mod sagaview;
// `sample_tape` is NOT re-exported any more (2026-09-10): it became `#[cfg(test)]`,
// because it was what the shipped Saga pane REPLAYED with no imported tape — two
// invented njord sessions, one stuck on `/offert/7` with fabricated validation errors.
pub use sagaview::{
saga_panel, saga_panel_at, saga_panel_from_store, saga_tape_path, workspace_saga_tape_path,
SagaView,
};
mod warehousehealth;
// `health_fixture` is NOT re-exported any more (2026-09-10): it became `#[cfg(test)]`,
// because it was what the shipped Warehouse Health pane drew with no stats file — a
// POPULATED warehouse at plausible byte sizes, from the pane whose whole job is to make
// an empty one impossible to miss. The eighth pane; `bf692d9` covered the other seven.
pub use warehousehealth::{
human_bytes, human_snapshot, warehouse_health_panel, warehouse_health_panel_at,
workspace_health_path, TableHealthRow, WarehouseHealthDoc, WarehouseHealthView,
};
/// Which rung of the credential chain answered — re-exported from the crate that owns
/// the ladder, so a host naming it cannot pick a second copy. See [`ClientCredential`].
pub use nornir_service::TokenSource;
/// The default `dwarves-server` endpoint on this box.
pub const DEFAULT_ENDPOINT: &str = "http://127.0.0.1:7900";
/// **The bound on a READ, which `connect_timeout` is not.** A server that ACCEPTS the
/// socket and then never answers clears the connect bound instantly and then waits
/// forever — the shape that left `dwarves-ui-gpu` with no window at all (measured
/// against the live oden forge, 2026-09-06). Pairing the connect bound with a request
/// bound is what makes an unresponsive host degrade to an honest unavailable state.
pub(crate) const READ_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(20);
/// **The dwarves THIN CLIENT's credential** — one resolver, one place, and it names
/// the rung that answered.
///
/// # The defect this closes
///
/// Every thin-client seam in this crate read its bearer as
/// `std::env::var("DWARVES_SERVER_TOKEN").unwrap_or_default()`, and so did
/// `dwarves-gui`'s `build_native_app`. That is a hand-rolled ONE-RUNG copy of a ladder
/// that already exists: the `dwarves` CLI resolves its bearer through
/// `nornir_service::resolve_token(--token, $DWARVES_SERVER_TOKEN, ~/.dwarves/token)`,
/// the shared ladder korp/nornir/dwarves all consume. The client had the env rung and
/// nothing else.
///
/// The owner, 2026-08-27: *"the token chain is wrong EVERY time server and client are
/// started."* It is the same box, the same operator and the same secret — a client
/// given `--server <url>` was told there was no credential while
/// `~/.dwarves/token` sat there holding it, and the UI's own refusal text asserted
/// *"There is no `~/.nornir/token` fallback"* as though the absence were the rule
/// rather than the bug.
///
/// # Which file, and why not `~/.nornir/token`
///
/// `~/.dwarves/token` — [`dwarves::server::dwarves_token_file`], the SAME 0600 file
/// `dwarves-server` reads when it resolves the secret it will demand. That is the
/// point of the convention: a server and a client on one box share one file and
/// nothing needs to be in the environment at all.
///
/// It is deliberately NOT `~/.nornir/token`, even though on oden the two files are
/// byte-identical (verified 2026-08-27, `cmp` says IDENTICAL) — which is precisely why
/// picking the wrong one would go unnoticed here and break on the first box where they
/// differ. An endpoint given as `--server <dwarves url>` is a DWARVES endpoint; the
/// credential resolves from the dwarves chain. nornir's secret is nornir's.
///
/// # Empty is still legal, and that is not a hedge
///
/// `dwarves-gui`'s own doctrine is *"Empty ⇒ open (the host's dev default)"*, and the
/// panes' gRPC tests drive a mock that wants no credential. So an unresolved credential
/// is NOT refused here; it is resolved to `""` with [`TokenSource::Nothing`] and the
/// CALLER decides. What changes is that the state is now nameable, which is what makes
/// *no credential configured* and *token rejected* different sentences instead of one
/// generic gRPC refusal.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ClientCredential {
/// The bearer to send. `""` ⇒ nothing was configured (open, for a dev host).
pub token: String,
/// Which rung answered — `$DWARVES_SERVER_TOKEN`, the token file, or nothing.
pub whence: nornir_service::TokenSource,
}
/// **The variable the dwarves thin client reads its bearer from.** Named once so the
/// code that reads it and every message that names it cannot drift apart.
pub const TOKEN_ENV: &str = "DWARVES_SERVER_TOKEN";
/// The dwarves user token file: `~/.dwarves/token`. `None` when `$HOME` is unset.
///
/// Spelled through [`dwarves::store::forge_state_in`] — the function that owns the
/// `.dwarves` name — so the client cannot look in a directory the server does not
/// write. (`dwarves::server::dwarves_token_file` is the server-side spelling of the
/// same path, and it is behind the `server` feature, which a thin client must not
/// link.)
pub fn token_file() -> Option<std::path::PathBuf> {
std::env::var_os("HOME")
.filter(|h| !h.is_empty())
.map(|h| dwarves::store::forge_state_in(std::path::Path::new(&h)).join("token"))
}
impl ClientCredential {
/// Walk the chain: `$DWARVES_SERVER_TOKEN` → `~/.dwarves/token` → nothing.
///
/// The SHARED `nornir_service::resolve_token_whence` does the walking (LAW 5 — the
/// ladder is not re-implemented here, and the rung is a FACT from the walk, not an
/// inference from comparing the value back to `$VAR`, which on oden would answer
/// "env" either way because both places hold the same bytes).
pub fn resolve() -> Self {
Self::resolve_with(None)
}
/// [`resolve`](ClientCredential::resolve) with an explicit `--token`-style
/// override at the top of the chain, for a host that parsed one.
pub fn resolve_with(flag: Option<&str>) -> Self {
let file = token_file();
let (token, whence) = nornir_service::resolve_token_whence(
flag.filter(|t| !t.trim().is_empty()),
TOKEN_ENV,
file.as_deref(),
);
ClientCredential { token, whence }
}
/// Is there a credential at all?
pub fn is_configured(&self) -> bool {
!self.token.trim().is_empty()
}
/// **Where this bearer came from, for a pane's provenance line.** Never the secret.
pub fn provenance(&self) -> String {
match &self.whence {
nornir_service::TokenSource::Nothing => format!(
"no credential: ${TOKEN_ENV} is unset or empty and {} holds none",
token_file()
.map(|p| p.display().to_string())
.unwrap_or_else(|| "the token file ($HOME unset)".into())
),
other => format!("{} ({} chars)", other.describe(), self.token.trim().len()),
}
}
/// **Say which KIND of auth failure this was**, once the server has answered.
///
/// The server's `The request does not have valid authentication credentials` reads
/// identically for four different situations, and the client knows something the
/// server does not: what it sent, and from where. Returns `None` for anything that
/// is not an auth refusal — a dead endpoint, a missing member, a clone failure all
/// keep their own words.
///
/// The two answers are the two REPAIRS, and they must never be confusable:
///
/// * **NO CREDENTIAL** — nothing was configured. Fix: set `$DWARVES_SERVER_TOKEN`,
/// or put the secret in `~/.dwarves/token` (the message names the exact path, so
/// the fallback is a rung the operator can USE rather than one the client claims
/// not to have).
/// * **TOKEN REJECTED** — a credential exists, was sent, and the server said no.
/// Fix: a different secret. Nothing about the chain will help, and the message
/// says which rung produced the rejected one, because a stale token file and a
/// stale environment are different mistakes.
pub fn explain_auth_failure(&self, status: &str) -> Option<String> {
let low = status.to_ascii_lowercase();
let is_auth = low.contains("authentication credentials")
|| low.contains("unauthenticated")
|| (low.contains("invalid") && low.contains("token"));
if !is_auth {
return None;
}
if !self.is_configured() {
let file = token_file()
.map(|p| p.display().to_string())
.unwrap_or_else(|| "<no $HOME, so no token file>".into());
return Some(format!(
"error: NO CREDENTIAL CONFIGURED — the server refused because this \
client sent an EMPTY bearer, not because a token was wrong. The chain \
it walks is ${TOKEN_ENV}, then {file}, and NEITHER held a secret. Set \
the env var, or write the host's token into {file} (0600) — that file \
IS a supported rung, and it is the same one dwarves-server reads.\
\n\nserver said: {status}"
));
}
Some(format!(
"error: TOKEN REJECTED — the server refused the credential this client DID \
send ({} chars, from {}). It exists and is wrong, revoked, or meant for a \
different host. This is NOT a missing-token problem and no change to the \
resolution chain will fix it.\n\nserver said: {status}",
self.token.trim().len(),
self.whence.describe(),
))
}
}
/// Default seconds between live auto-poll re-fetches of the run report.
pub const DEFAULT_POLL_INTERVAL_S: f64 = 1.5;
/// Default number of recent runs the run-picker lists (`History(limit)`).
pub const DEFAULT_HISTORY_LIMIT: u32 = 20;
// ── the run-picker (pure wire→pick mapping, unit-testable) ────────────────────
/// One entry in the Forge(live) **run-picker** — enough to label a historized run in
/// the dropdown and pin it (its `run_id`). The pure [`run_picks_from_wire`] mapping
/// makes the picker list unit-testable WITHOUT a live server.
#[derive(Clone, Debug, PartialEq, Eq, serde::Serialize)]
pub struct RunPick {
/// The run's id — what selecting the entry pins (`update_json {"run_id": …}`).
pub run_id: String,
/// The pipeline the run belongs to (the human label in the dropdown).
pub pipeline: String,
/// The run's overall status label (`passed`/`failed`/`running`/…).
pub status: String,
}
/// Map a `History` reply's rows onto the run-picker entries. Pure (no network) so the
/// picker list is unit-testable; preserves the server's order (newest first) verbatim.
pub fn run_picks_from_wire(runs: &[pb::RunRow]) -> Vec<RunPick> {
runs.iter()
.map(|r| RunPick {
run_id: r.run_id.clone(),
pipeline: r.pipeline.clone(),
status: r.status.clone(),
})
.collect()
}
// ── wire → dwarves-domain reconstruction (pure, unit-testable) ────────────────
/// Lenient wire→[`RunStatus`] parse: accepts the server's status label
/// (`passed`/`failed`/`skipped`/`running`) and a few synonyms. Unknown ⇒ `Skipped`
/// (the neutral "never started / no signal" — never a false green, never a panic).
fn run_status_from_wire(s: &str) -> RunStatus {
match s.trim().to_ascii_lowercase().as_str() {
"passed" | "done" | "green" => RunStatus::Passed,
"failed" | "red" => RunStatus::Failed,
"running" | "in_progress" | "inprogress" | "active" => RunStatus::Running,
_ => RunStatus::Skipped,
}
}
/// Map a wire backend string onto the domain's `&'static str` backend tag (the
/// [`DagNode`] field is `&'static`). Unknown ⇒ `"variant"` (the default run kind).
fn backend_from_wire(s: &str) -> &'static str {
match s.trim().to_ascii_lowercase().as_str() {
"container" => "container",
"vm" => "vm",
_ => "variant",
}
}
/// Empty string ⇒ `None` (the wire has no proto optionals; empty is "absent").
fn opt(s: &str) -> Option<String> {
(!s.is_empty()).then(|| s.to_string())
}
/// **Reconstruct the build [`Dag`] + [`RunReport`] from a `GetRunReport` reply.**
///
/// Pure (no network), so it is unit-testable: maps the wire report (parent RunRow +
/// per-step BuildRows carrying their `needs` edges) back onto the in-memory pair the
/// [`ForgePipelineAdapter`] consumes. The DAG is reconstructed from the per-step
/// `needs` alone (no descriptor needed on the client) — a faithful twin of the
/// server's own `run_pipeline` → (`Dag`, `RunReport`) pair.
pub fn report_from_wire(report: &pb::RunReport) -> (Dag, RunReport) {
let run = report.run.clone().unwrap_or_default();
// ONE wire→domain RunRow lowering, shared with `fetch_show` (L5: this used to be a
// second hand-written copy of the same eleven fields).
let parent = run_row_from_wire(&run);
let mut steps = Vec::with_capacity(report.steps.len());
let mut order = Vec::with_capacity(report.steps.len());
for s in &report.steps {
steps.push(BuildRow {
run_id: s.run_id.clone(),
step: s.step.clone(),
stage: s.stage.clone(),
backend: s.backend.clone(),
matrix: s.matrix.clone(),
status: run_status_from_wire(&s.status),
summary: s.summary.clone(),
artifact_id: opt(&s.artifact_id),
job_ref: opt(&s.job_ref),
duration_ms: s.duration_ms,
finished_at: s.finished_at.clone(),
});
order.push(DagNode {
step: s.step.clone(),
stage: s.stage.clone(),
needs: s.needs.clone(),
backend: backend_from_wire(&s.backend),
});
}
(Dag { order }, RunReport { run: parent, steps })
}
/// Build the REUSED [`PipelineView`] from a reconstructed (`Dag`, `RunReport`) pair —
/// report → adapter → model → view, exactly as the in-process demo Pipeline tab does.
fn view_from_report(dag: &Dag, report: &RunReport) -> PipelineView {
let adapter = ForgePipelineAdapter::from_report(dag, report);
PipelineView::from_adapter(&adapter)
}
// ── thin gRPC fetch ───────────────────────────────────────────────────────────
/// Normalize an endpoint: prefix `http://` if the scheme is missing.
fn norm_endpoint(endpoint: &str) -> String {
if endpoint.starts_with("http") {
endpoint.to_string()
} else {
format!("http://{endpoint}")
}
}
/// The outcome of a forge fetch with its **two legs split**: the recent-run LIST
/// (`History`) and the SELECTED run's report (`GetRunReport`). The list is returned
/// whenever `History` succeeded — **even if the pinned run's report then failed** — so
/// the run-picker stays populated and the user can switch away from a broken run
/// instead of the whole pane collapsing to `unavailable`. An outer `Err` from
/// [`fetch_forge_split`] means the leg that feeds the LIST failed (couldn't connect /
/// `History` errored) — that is the genuinely-unavailable case.
pub struct ForgeFetch {
/// The recent runs for the picker (newest first). Empty only when the forge has none.
pub runs: Vec<pb::RunRow>,
/// The selected run's report — `Err` when the pinned run's `GetRunReport` failed
/// even though `History` succeeded (a bad pin, not a dead server).
pub report: Result<pb::RunReport>,
}
/// Fetch BOTH the recent-run list (for the run-picker) AND one run's report from
/// `dwarves-server` at `endpoint`, over the thin `History` + `GetRunReport` gRPC, in a
/// SINGLE connection, with the **two legs split** (see [`ForgeFetch`]). Authenticates
/// with `Bearer <token>` (empty ⇒ no real auth). The `History(limit=history_limit)`
/// reply feeds the picker and — when `run_id` is `None`/empty — selects the newest run
/// to render. If `History` succeeds but the SELECTED run's `GetRunReport` fails (a bad
/// pin / a run that vanished), the returned `ForgeFetch` still carries the fresh run
/// list with `report = Err(…)` — so the picker survives the bad pin. An outer `Err`
/// means the connection or `History` itself failed (genuinely unavailable). Runs on a
/// private current-thread runtime so it is safe to call from the synchronous egui loop;
/// a 3s connect timeout keeps a dead server from wedging the UI thread.
pub fn fetch_forge_split(
endpoint: &str,
token: &str,
run_id: Option<&str>,
history_limit: u32,
) -> Result<ForgeFetch> {
fetch_runs_of(endpoint, token, "", run_id, history_limit)
}
/// **One pipeline's runs** (newest first) and the selected one's report — the
/// [`fetch_forge_split`] the Forge tab reads, narrowed on the forge (`History.pipeline`).
/// `run_id` `None` ⇒ the pipeline's newest run.
pub fn fetch_pipeline_runs(
endpoint: &str,
token: &str,
pipeline: &str,
run_id: Option<&str>,
history_limit: u32,
) -> Result<ForgeFetch> {
fetch_runs_of(endpoint, token, pipeline, run_id, history_limit)
}
fn fetch_runs_of(
endpoint: &str,
token: &str,
pipeline: &str,
run_id: Option<&str>,
history_limit: u32,
) -> Result<ForgeFetch> {
let pipeline = pipeline.to_string();
let rt = tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()
.context("build tokio runtime for forge client")?;
let endpoint = norm_endpoint(endpoint);
let token = token.to_string();
let run_id = run_id.map(|s| s.to_string());
let history_limit = history_limit.max(1);
rt.block_on(async move {
// The ONE shared client bearer-builder (Law #1) — no per-site
// `format!("Bearer …").parse()` twin. `None` ⇒ open (empty token).
let bearer = nornir_service::client::bearer(&token).context("parse bearer token")?;
let channel = tonic::transport::Channel::from_shared(endpoint.clone())
.with_context(|| format!("invalid server url `{endpoint}`"))?
.connect_timeout(Duration::from_secs(3))
.timeout(crate::READ_TIMEOUT)
.connect()
.await
.with_context(|| format!("connect to dwarves-server at {endpoint}"))?;
let mut client = pb::dwarf_client::DwarfClient::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);
// ONE History call serves both the run-picker list AND newest-run selection.
// A failure here means the LIST leg failed ⇒ genuinely unavailable (outer Err).
let runs = client
.history(pb::HistoryRequest {
limit: history_limit,
pipeline,
})
.await
.context("Dwarf.History RPC")?
.into_inner()
.runs;
// Pick the run to show: the pinned id, else the newest historized run.
let id = match run_id {
Some(id) if !id.trim().is_empty() => Some(id),
_ => runs.first().map(|r| r.run_id.clone()),
};
// The report leg is kept SEPARATE: a failure here (bad pin / vanished run /
// empty forge) does NOT discard the freshly-fetched `runs` list.
let report = match id {
Some(id) => client
.get_run_report(pb::RunStatusRequest { run_id: id })
.await
.map(|r| r.into_inner())
.context("Dwarf.GetRunReport RPC"),
None => Err(anyhow!("no runs on the forge yet")),
};
Ok(ForgeFetch { runs, report })
})
}
/// **One step's log** from the forge (`Dwarf.StepLog`): the last `tail` lines of step
/// row `seq` of `run_id` (`0` = the whole log, its end).
pub fn fetch_step_log(
endpoint: &str,
token: &str,
run_id: &str,
seq: u32,
tail: u32,
) -> Result<pb::StepLogReply> {
let rt = tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()
.context("build tokio runtime for forge client")?;
let endpoint = norm_endpoint(endpoint);
let (token, run_id) = (token.to_string(), run_id.to_string());
rt.block_on(async move {
let bearer = nornir_service::client::bearer(&token).context("parse bearer token")?;
let channel = tonic::transport::Channel::from_shared(endpoint.clone())
.with_context(|| format!("invalid server url `{endpoint}`"))?
.connect_timeout(Duration::from_secs(3))
.timeout(crate::READ_TIMEOUT)
.connect()
.await
.with_context(|| format!("connect to dwarves-server at {endpoint}"))?;
let mut client = pb::dwarf_client::DwarfClient::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);
Ok(client
.step_log(pb::StepLogRequest { run_id, seq, tail })
.await
.context("Dwarf.StepLog RPC")?
.into_inner())
})
}
/// The bearer interceptor of [`dwarf_call`].
type Bearer = Box<dyn FnMut(tonic::Request<()>) -> std::result::Result<tonic::Request<()>, tonic::Status> + Send>;
/// The client [`dwarf_call`] hands its call.
type BearerClient = pb::dwarf_client::DwarfClient<tonic::service::interceptor::InterceptedService<tonic::transport::Channel, Bearer>>;
/// One call on the forge's `Dwarf` service, with the bearer.
fn dwarf_call<T>(
endpoint: &str,
token: &str,
f: impl for<'c> FnOnce(&'c mut BearerClient) -> std::pin::Pin<Box<dyn std::future::Future<Output = Result<T>> + 'c>>,
) -> Result<T> {
let rt = tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()
.context("build tokio runtime for forge client")?;
let endpoint = norm_endpoint(endpoint);
let token = token.to_string();
rt.block_on(async move {
let bearer = nornir_service::client::bearer(&token).context("parse bearer token")?;
let channel = tonic::transport::Channel::from_shared(endpoint.clone())
.with_context(|| format!("invalid server url `{endpoint}`"))?
.connect_timeout(Duration::from_secs(3))
.timeout(crate::READ_TIMEOUT)
.connect()
.await
.with_context(|| format!("connect to dwarves-server at {endpoint}"))?;
let icpt: Bearer =
Box::new(move |mut req: tonic::Request<()>| {
if let Some(b) = &bearer {
req.metadata_mut().insert("authorization", b.clone());
}
Ok(req)
});
let mut client = pb::dwarf_client::DwarfClient::with_interceptor(channel, icpt).max_decoding_message_size(64 << 20);
f(&mut client).await
})
}
/// **Queue a release build** (lane RELBUILD-RUNA): `Dwarf.Trigger` of `release` /
/// `release-build` with the request's inputs (TOML, `[inputs]`: the same inputs as
/// `dwarves release-build request`). The forge queues it for the release runner and
/// answers the run id.
pub fn trigger_release(endpoint: &str, token: &str, request: &str) -> Result<pb::TriggerReply> {
let request = request.to_string();
dwarf_call(endpoint, token, move |c| {
Box::pin(async move {
Ok(c.trigger(pb::TriggerRequest {
project: "release".into(),
pipeline: "release-build".into(),
trigger: "dwarves-ui".into(),
request,
})
.await
.context("Dwarf.Trigger RPC")?
.into_inner())
})
})
}
/// **The owner's signed bundle for a release run** (`Dwarf.ReleaseBundle`): the forge
/// verifies it against the release key and the run's statement before it stores it.
pub fn release_bundle(endpoint: &str, token: &str, run_id: &str, bundle: Vec<u8>) -> Result<pb::TriggerReply> {
let run_id = run_id.to_string();
dwarf_call(endpoint, token, move |c| {
Box::pin(async move {
Ok(c.release_bundle(pb::ReleaseBundleRequest { run_id, bundle })
.await
.context("Dwarf.ReleaseBundle RPC")?
.into_inner())
})
})
}
/// Fetch BOTH the recent-run list AND the selected run's report, folding the split back
/// into the strict `(runs, report)` pair — a report error becomes an overall `Err`. The
/// original combined API, kept for callers that want all-or-nothing; the live pane uses
/// [`fetch_forge_split`] so a bad pin keeps the picker list.
pub fn fetch_forge(
endpoint: &str,
token: &str,
run_id: Option<&str>,
history_limit: u32,
) -> Result<(Vec<pb::RunRow>, pb::RunReport)> {
let f = fetch_forge_split(endpoint, token, run_id, history_limit)?;
let report = f.report?;
Ok((f.runs, report))
}
/// **The Status BOARD's feed** — the forge's `Show` dump: the REGISTERED pipeline
/// roster plus the recent runs, in ONE round trip.
///
/// Why `Show` and not `History`: a board draws one chip per *pipeline*, and a pipeline
/// that has never run contributes no history row. Fed by `History` alone, a board can
/// only ever show pipelines that already ran — which is precisely the set that cannot
/// be in trouble. `Show.pipelines` is the registry, so an unrun pipeline lands on the
/// board as the grey "unknown" chip it should be.
///
/// Returns the pair as the domain types the shared `brokkr::badge` selector consumes:
/// `(pipeline names, runs)`. Same private-runtime / 3s-connect-timeout shape as
/// [`fetch_forge_split`], so it is safe to call off the egui thread.
pub fn fetch_show(endpoint: &str, token: &str) -> Result<(Vec<String>, Vec<PipelineRunRow>)> {
let rt = tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()
.context("build tokio runtime for forge client")?;
let endpoint = norm_endpoint(endpoint);
let token = token.to_string();
let reply = rt.block_on(async move {
let bearer = nornir_service::client::bearer(&token).context("parse bearer token")?;
let channel = tonic::transport::Channel::from_shared(endpoint.clone())
.with_context(|| format!("invalid server url `{endpoint}`"))?
.connect_timeout(Duration::from_secs(3))
.timeout(crate::READ_TIMEOUT)
.connect()
.await
.with_context(|| format!("connect to dwarves-server at {endpoint}"))?;
let mut client = pb::dwarf_client::DwarfClient::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);
anyhow::Ok(
client
.show(pb::Empty {})
.await
.context("Dwarf.Show RPC")?
.into_inner(),
)
})?;
// The board's roster: every REGISTERED pipeline, plus any pipeline a recent run
// names that the registry no longer does (a de-registered pipeline whose runs are
// still on the ledger must not vanish from the board — its last status is a fact).
let mut pipelines: Vec<String> = reply.pipelines.iter().map(|p| p.name.clone()).collect();
for r in &reply.recent_runs {
if !r.pipeline.is_empty() && !pipelines.iter().any(|p| p == &r.pipeline) {
pipelines.push(r.pipeline.clone());
}
}
let runs = reply.recent_runs.iter().map(run_row_from_wire).collect();
Ok((pipelines, runs))
}
/// Lower one wire [`pb::RunRow`] into the domain [`PipelineRunRow`] the shared
/// `brokkr::badge`/history selectors read. `sha`/`job_ref` are not on this wire (the
/// server's `RunRow` does not carry them), so they stay `None` rather than invented.
pub fn run_row_from_wire(r: &pb::RunRow) -> PipelineRunRow {
PipelineRunRow {
run_id: r.run_id.clone(),
pipeline: r.pipeline.clone(),
project: r.project.clone(),
trigger: r.trigger.clone(),
sha: None,
status: run_status_from_wire(&r.status),
steps: r.steps as usize,
passed: r.passed as usize,
duration_ms: r.duration_ms,
job_ref: None,
finished_at: r.finished_at.clone(),
}
}
/// Fetch one run's full report — the thin single-report wrapper over [`fetch_forge`]
/// (kept for callers that don't need the run list). Selects the newest run when
/// `run_id` is `None`/empty.
pub fn fetch_run_report(
endpoint: &str,
token: &str,
run_id: Option<&str>,
) -> Result<pb::RunReport> {
fetch_forge(endpoint, token, run_id, DEFAULT_HISTORY_LIMIT).map(|(_runs, report)| report)
}
// ── the embeddable Panel ────────────────────────────────────────────────────
/// Connection outcome of the last fetch (drives the status line + `state_json`).
#[derive(Clone, Debug, PartialEq, Eq)]
enum Conn {
/// Never fetched yet.
Idle,
/// The server returned a run report.
Loaded,
/// `History` succeeded (the run list is fresh) but the SELECTED run's report failed
/// — a recoverable state: the picker stays usable to switch to a working run.
ReportError,
/// **Connected, but the forge has no runs yet.** `History` succeeded (the server
/// is UP and reachable) and returned zero runs. Distinct from [`Conn::Unavailable`]
/// (couldn't connect) so a wired-but-idle forge reads "connected · no runs yet",
/// never the misleading "offline".
Empty,
/// The server was unreachable / `History` errored — a clean, non-panicking state.
Unavailable,
}
/// The LIVE forge-run pane: fetches one run's report from dwarves-server over thin
/// gRPC and renders the reconstructed build pipeline with [`facett_pipeline`]. A
/// [`Facet`] (hence a [`Panel`] via the blanket impl) a host mounts in a tab.
pub struct ForgePanel {
endpoint: String,
token: String,
/// The pinned run id (`None` ⇒ always render the newest run).
run_id: Option<String>,
/// The run id actually rendered by the last successful fetch (for the status line).
shown_run_id: Option<String>,
/// The recent runs the picker offers (newest first), from the last `History` fetch.
/// Left as-is on a transient fetch error (a blip doesn't blank the dropdown).
runs: Vec<RunPick>,
/// How many recent runs the picker requests (`History(limit)`).
history_limit: u32,
view: Option<PipelineView>,
conn: Conn,
node_count: usize,
edge_count: usize,
/// The last error's display text (folded into `state_json` for tests).
error: Option<String>,
/// LIVE auto-poll: re-fetch every `poll_interval_s` so a run in flight fills in
/// its steps on its own. On by default.
auto_poll: bool,
poll_interval_s: f64,
last_poll_s: f64,
/// The shown run's `BuildWorkspace` steps' units (the server's `units_json`).
units: UnitsPane,
}
// ── a BuildWorkspace step's units, in the run view ─────────────────────────────
/// The units of the shown run's `BuildWorkspace` steps: per step the counts, a repo
/// table (a click drills into that repo) and its failed units (a click shows the
/// error) — drawn with the shared `facett-table`.
#[derive(Default)]
pub struct UnitsPane {
steps: Vec<dwarves::run_units::StepUnits>,
/// The repo drilled into (`None` ⇒ every repo's failed units).
repo: Option<String>,
/// `(step index, failed index)` of the failed unit whose error is shown.
picked: Option<(usize, usize)>,
/// Per step: the repo table and the failed-units table as drawn.
tables: Vec<(facett_table::Table, facett_table::Table)>,
}
impl UnitsPane {
/// The units carried by a report's step rows.
pub fn of_report(report: &pb::RunReport) -> Vec<dwarves::run_units::StepUnits> {
report
.steps
.iter()
.filter(|s| !s.units_json.is_empty())
.filter_map(|s| serde_json::from_str(&s.units_json).ok())
.collect()
}
pub fn set(&mut self, steps: Vec<dwarves::run_units::StepUnits>) {
if steps != self.steps {
self.picked = None;
if self
.repo
.as_ref()
.is_some_and(|r| !steps.iter().any(|s| s.repos.iter().any(|x| &x.repo == r)))
{
self.repo = None;
}
}
self.steps = steps;
self.rebuild();
}
/// The failed units of step `i` the drill shows (the repo's, or all).
fn shown(&self, i: usize) -> Vec<&dwarves::run_units::FailedUnit> {
self.steps[i]
.failed
.iter()
.filter(|f| self.repo.as_ref().is_none_or(|r| f.repos.contains(r)))
.collect()
}
fn rebuild(&mut self) {
self.tables = (0..self.steps.len())
.map(|i| {
let st = &self.steps[i];
let cols = |c: &[&str]| c.iter().map(|s| s.to_string()).collect::<Vec<_>>();
let mut repos = facett_table::Table::new(
format!("units-repos-{}", st.seq),
cols(&[
"repo", "result", "compiled", "fresh", "failed", "skipped", "phase",
]),
)
.with_row_ids(st.repos.iter().map(|r| r.repo.clone()).collect());
for r in &st.repos {
let c = &r.counts;
repos.push_row(vec![
r.repo.clone(),
r.result.clone(),
c.compiled.to_string(),
c.fresh.to_string(),
c.failed.to_string(),
c.skipped.to_string(),
r.phase.clone(),
]);
}
if let Some(r) = &self.repo {
let _ = repos.update(facett_table::Msg::ToggleRow(r.clone()));
}
let mut failed = facett_table::Table::new(
format!("units-failed-{}", st.seq),
cols(&["crate", "kind", "target", "features", "repos", "error"]),
);
for f in self.shown(i) {
let first = f
.error
.and_then(|e| st.errors.get(e))
.and_then(|e| {
e.lines()
.find(|l| l.contains("error"))
.or_else(|| e.lines().next())
})
.unwrap_or("(no error text)");
failed.push_row(vec![
f.krate.clone(),
f.kind.clone(),
f.target.clone(),
f.features_hash.clone(),
f.repos.join(","),
first.to_string(),
]);
}
if let Some((_, n)) = self.picked.filter(|p| p.0 == i) {
failed.select_row(n);
}
(repos, failed)
})
.collect();
}
/// Drill into `repo` (empty ⇒ all repos).
pub fn drill(&mut self, repo: &str) {
self.repo = (!repo.is_empty()).then(|| repo.to_string());
self.picked = None;
self.rebuild();
}
/// Show the error of the `n`th shown failed unit of step `i`.
pub fn pick(&mut self, i: usize, n: usize) {
if i < self.steps.len() && n < self.shown(i).len() {
self.picked = Some((i, n));
self.rebuild();
}
}
fn picked_error(&self) -> Option<String> {
let (i, n) = self.picked?;
let f = self.shown(i).get(n).copied()?;
Some(match f.error.and_then(|e| self.steps[i].errors.get(e)) {
Some(e) => e.clone(),
None => format!("{} failed; nornir gave no error text for it", f.krate),
})
}
pub fn ui(&mut self, ui: &mut egui::Ui) {
let mut drill: Option<String> = None;
let mut pick: Option<(usize, usize)> = None;
for i in 0..self.steps.len() {
let st = &self.steps[i];
let title = format!(
"🧱 {} · BuildWorkspace {} — {}",
st.stage,
st.workspace,
st.line()
);
egui::CollapsingHeader::new(title)
.id_salt(("dwarves-embed-units", st.seq))
.default_open(true)
.show(ui, |ui| {
if st.source != "followed" {
ui.weak(st.why.clone().unwrap_or_default());
return;
}
let (repos, failed) = &self.tables[i];
ui.push_id(("units-repos", i), |ui| {
egui::ScrollArea::vertical()
.max_height(180.0)
.show(ui, |ui| {
for m in repos.view(ui) {
if let facett_table::Msg::ToggleRow(r) = m {
drill = Some(if self.repo.as_deref() == Some(r.as_str()) {
String::new()
} else {
r
});
}
}
});
});
let shown = self.shown(i).len();
ui.label(format!(
"failed units{} — {shown} shown of {}{}",
self.repo
.as_ref()
.map(|r| format!(" of {r}"))
.unwrap_or_default(),
st.failed_total,
if st.failed_total as usize > st.failed.len() {
" (the run view carries the first 500)"
} else {
""
}
));
ui.push_id(("units-failed", i), |ui| {
egui::ScrollArea::vertical()
.max_height(160.0)
.show(ui, |ui| {
for m in failed.view(ui) {
if let facett_table::Msg::ToggleRow(r) = m {
if let Ok(n) = r.parse::<usize>() {
pick = Some((i, n));
}
}
}
});
});
for n in &st.notes {
ui.weak(n);
}
});
}
if let Some(e) = self.picked_error() {
egui::ScrollArea::vertical()
.id_salt("dwarves-embed-units-error")
.max_height(200.0)
.show(ui, |ui| {
ui.monospace(e);
});
}
if let Some(r) = drill {
self.drill(&r);
}
if let Some((i, n)) = pick {
self.pick(i, n);
}
}
pub fn state_json(&self) -> serde_json::Value {
let steps: Vec<serde_json::Value> = (0..self.steps.len())
.map(|i| {
let st = &self.steps[i];
serde_json::json!({
"seq": st.seq, "stage": st.stage, "workspace": st.workspace, "source": st.source, "why": st.why,
"line": st.line(), "counts": st.counts, "repos": st.repos, "failed_total": st.failed_total,
"failed_shown": self.shown(i).iter().map(|f| serde_json::json!({
"crate": f.krate, "kind": f.kind, "target": f.target, "features_hash": f.features_hash, "repos": f.repos,
"has_error": f.error.is_some(),
})).collect::<Vec<_>>(),
"drawn": { "repo_rows": self.tables.get(i).map_or(0, |t| t.0.rows.len()), "failed_rows": self.tables.get(i).map_or(0, |t| t.1.rows.len()) },
"notes": st.notes,
})
})
.collect();
serde_json::json!({ "steps": steps, "repo": self.repo, "picked": self.picked, "error": self.picked_error() })
}
}
impl ForgePanel {
/// Build the pane for `endpoint` and eagerly attempt the first fetch. A failed
/// fetch is captured as the `Unavailable` state (never a panic). `run_id` pins a
/// specific run; `None` renders the newest.
pub fn new(
endpoint: impl Into<String>,
token: impl Into<String>,
run_id: Option<String>,
) -> Self {
let mut p = Self {
endpoint: endpoint.into(),
token: token.into(),
run_id: run_id.filter(|s| !s.trim().is_empty()),
shown_run_id: None,
runs: Vec::new(),
history_limit: DEFAULT_HISTORY_LIMIT,
view: None,
conn: Conn::Idle,
node_count: 0,
edge_count: 0,
error: None,
auto_poll: true,
poll_interval_s: DEFAULT_POLL_INTERVAL_S,
last_poll_s: 0.0,
units: UnitsPane::default(),
};
p.refresh();
p
}
/// True when auto-poll is on AND at least one interval has elapsed since the last
/// fetch (`now_s` = egui input time). Pure → unit-testable without a clock.
pub fn poll_due(&self, now_s: f64) -> bool {
self.auto_poll && (now_s - self.last_poll_s) >= self.poll_interval_s
}
/// Turn live auto-poll on/off (the ● live / ⏸ paused toggle).
pub fn set_auto_poll(&mut self, on: bool) {
self.auto_poll = on;
}
/// (Re-)fetch the run list + selected report and apply the outcome. On the
/// list-leg failing, drops into the clean `Unavailable` state; on only the
/// report-leg failing, into the recoverable `ReportError` state (picker preserved).
pub fn refresh(&mut self) {
let outcome = fetch_forge_split(
&self.endpoint,
&self.token,
self.run_id.as_deref(),
self.history_limit,
);
self.apply_fetch(outcome);
}
/// Fold a [`fetch_forge_split`] outcome into the pane's rendered state. Split out
/// from [`refresh`] so the state transitions are drivable headless in tests without
/// a live server (the same `state_json` seam a robot reads). The crux: an `Ok`
/// fetch whose `report` is `Err` KEEPS the fresh run-picker list and lands in the
/// recoverable `ReportError` state — it never throws the list away.
fn apply_fetch(&mut self, outcome: Result<ForgeFetch>) {
match outcome {
Ok(f) => {
// History succeeded ⇒ the picker list is FRESH regardless of the report.
self.runs = run_picks_from_wire(&f.runs);
match f.report {
Ok(report) => {
self.units.set(UnitsPane::of_report(&report));
let (dag, run_report) = report_from_wire(&report);
let view = view_from_report(&dag, &run_report);
self.node_count = view.state().nodes.len();
self.edge_count = view.state().edge_ends.len();
self.shown_run_id = Some(run_report.run.run_id.clone());
self.view = Some(view);
self.conn = Conn::Loaded;
self.error = None;
}
Err(e) => {
// Only the selected run's report failed. Keep the list so the
// picker can switch to a working run. An EMPTY forge (nothing
// to pick) is the plain unavailable/no-runs state as before; a
// bad pin WITH runs to switch to is the recoverable one.
self.view = None;
self.units.set(Vec::new());
self.node_count = 0;
self.edge_count = 0;
self.shown_run_id = None;
// History CONNECTED (outer Ok): an empty forge is "connected
// but no runs yet" (Empty), NOT unavailable; a bad pin WITH
// runs to switch to is the recoverable ReportError.
self.conn = if self.runs.is_empty() {
Conn::Empty
} else {
Conn::ReportError
};
self.error = Some(format!("{e:#}"));
}
}
}
Err(e) => {
// The LIST leg failed (dead server / History errored): genuinely
// unavailable. Leave `self.runs` as-is (a blip doesn't blank the dropdown).
self.view = None;
self.units.set(Vec::new());
self.node_count = 0;
self.edge_count = 0;
self.shown_run_id = None;
self.conn = Conn::Unavailable;
self.error = Some(format!("{e:#}"));
}
}
}
fn conn_key(&self) -> &'static str {
match self.conn {
Conn::Idle => "idle",
Conn::Loaded if self.node_count == 0 => "empty",
Conn::Loaded => "loaded",
Conn::ReportError => "report_error",
Conn::Empty => "no_runs",
Conn::Unavailable => "unavailable",
}
}
}
impl Facet for ForgePanel {
fn title(&self) -> &str {
"Forge"
}
fn ui(&mut self, ui: &mut egui::Ui) {
// LIVE auto-poll: re-fetch on the interval so a run in flight fills in.
let now_s = ui.input(|i| i.time);
if self.poll_due(now_s) {
self.refresh();
self.last_poll_s = now_s;
}
if self.auto_poll {
ui.ctx()
.request_repaint_after(std::time::Duration::from_secs_f32(
self.poll_interval_s as f32,
));
}
let mut do_refresh = false;
ui.horizontal(|ui| {
if ui
.button("⟳ Refresh")
.on_hover_text(format!("re-fetch the run report from {}", self.endpoint))
.clicked()
{
do_refresh = true;
}
let (glyph, hint) = if self.auto_poll {
("● live", "auto-polling — click to pause")
} else {
("⏸ paused", "auto-poll paused — click to resume live")
};
if ui.button(glyph).on_hover_text(hint).clicked() {
self.auto_poll = !self.auto_poll;
self.last_poll_s = now_s;
}
// ── the run-picker: choose which historized run to render ──────────
// Selecting an entry pins its run_id (== the `{"run_id": …}` seam a robot
// drives); "● newest" clears the pin. Snapshot the list so the dropdown
// closure borrows no `self` field mutably.
let runs = self.runs.clone();
let pinned = self.run_id.clone();
let selected_label = match &pinned {
Some(id) => id.clone(),
None => "newest".to_string(),
};
let mut picked: Option<Option<String>> = None;
egui::ComboBox::from_id_salt("dwarves-embed-run-picker")
.selected_text(selected_label)
.show_ui(ui, |ui| {
if ui.selectable_label(pinned.is_none(), "● newest").clicked() {
picked = Some(None);
}
for r in &runs {
let is_sel = pinned.as_deref() == Some(r.run_id.as_str());
let text = format!("{} · {} · {}", r.run_id, r.pipeline, r.status);
if ui.selectable_label(is_sel, text).clicked() {
picked = Some(Some(r.run_id.clone()));
}
}
if runs.is_empty() {
ui.weak("no runs yet");
}
});
if let Some(sel) = picked {
self.run_id = sel.filter(|s| !s.trim().is_empty());
do_refresh = true;
self.last_poll_s = now_s;
}
ui.separator();
match self.conn {
Conn::Idle => {
ui.weak("connecting…");
}
Conn::Loaded if self.node_count == 0 => {
ui.weak("run has no steps yet");
}
Conn::Loaded => {
let run = self.shown_run_id.as_deref().unwrap_or("?");
// PLAIN NUMBERS (no thousands separators) — a nordisk law.
ui.label(format!(
"live run {run} · {} steps / {} edges",
self.node_count, self.edge_count
));
}
Conn::ReportError => {
// History is fresh (the picker above still lists the runs); only the
// selected run's report failed — invite switching to another run.
let run = self.run_id.as_deref().unwrap_or("?");
let resp = ui.colored_label(
egui::Color32::from_rgb(224, 160, 60),
format!("run {run} unavailable — pick another run"),
);
if let Some(e) = &self.error {
resp.on_hover_text(e);
}
}
Conn::Empty => {
// Connected + reachable, but the forge has run nothing yet — an
// honest "wired but idle", visibly different from offline.
ui.colored_label(
egui::Color32::from_rgb(120, 190, 120),
format!("● connected to {} · forge has no runs yet", self.endpoint),
)
.on_hover_text("dwarves-server answered History with zero runs — trigger a pipeline and it will appear here live");
}
Conn::Unavailable => {
let resp = ui.colored_label(
egui::Color32::from_rgb(224, 120, 90),
format!("forge unavailable — cannot reach dwarves-server at {}", self.endpoint),
);
if let Some(e) = &self.error {
resp.on_hover_text(e);
}
}
}
});
if do_refresh {
self.refresh();
self.last_poll_s = now_s;
}
ui.separator();
if !self.units.steps.is_empty() {
self.units.ui(ui);
ui.separator();
}
match &mut self.view {
Some(v) => Facet::ui(v, ui),
None => {
ui.centered_and_justified(|ui| match self.conn {
Conn::Unavailable => {
ui.weak(format!(
"could not reach dwarves-server at {} — press ⟳ Refresh",
self.endpoint
));
}
Conn::Empty => {
ui.weak("connected — the forge has no runs yet; trigger a pipeline to see it here live");
}
Conn::ReportError => {
ui.weak("that run's report could not be loaded — choose another from the picker");
}
_ => {
ui.weak("no run to render");
}
});
}
}
}
fn state_json(&self) -> serde_json::Value {
let pipeline = match &self.view {
Some(v) => Facet::state_json(v),
None => serde_json::json!({ "nodes": [] }),
};
serde_json::json!({
"kind": "forge",
"endpoint": self.endpoint,
"run_id": self.shown_run_id,
// The picker's pin (None ⇒ "newest"); distinct from `run_id` (the run
// actually rendered by the last successful fetch).
"selected_run_id": self.run_id,
"status": self.conn_key(),
"node_count": self.node_count,
"edge_count": self.edge_count,
"error": self.error,
"auto_poll": self.auto_poll,
"poll_interval_s": self.poll_interval_s,
// The run-picker surface: the recent runs offered + how many were asked for.
"runs": self.runs,
"run_count": self.runs.len(),
"history_limit": self.history_limit,
"pipeline": pipeline,
// the shown run's BuildWorkspace steps: units per repo, failed units, errors
"units": self.units.state_json(),
})
}
fn update_json(&mut self, msg_json: &str) {
// The pane's own control surface (drives it headless from a robot test):
// {"refresh": true} / {"poll": true} → force a re-fetch
// {"auto_poll": bool} → toggle live polling
// {"run_id": "…"} → pin a run in the picker (""=newest) + re-fetch
// {"history_limit": N} → resize the run-picker list + re-fetch
// Anything else is forwarded to the inner pipeline facet (its own JSON seam).
if let Ok(v) = serde_json::from_str::<serde_json::Value>(msg_json) {
if let Some(on) = v.get("auto_poll").and_then(|r| r.as_bool()) {
self.auto_poll = on;
return;
}
if let Some(id) = v.get("run_id").and_then(|r| r.as_str()) {
self.run_id = (!id.trim().is_empty()).then(|| id.to_string());
self.refresh();
return;
}
// {"units_repo": "…"} → drill the units into a repo (""=all)
// {"units_failed": [step, n]} → show the error of a shown failed unit
if let Some(r) = v.get("units_repo").and_then(|r| r.as_str()) {
self.units.drill(r);
return;
}
if let Some(a) = v.get("units_failed").and_then(|r| r.as_array()) {
if let (Some(i), Some(n)) = (
a.first().and_then(|x| x.as_u64()),
a.get(1).and_then(|x| x.as_u64()),
) {
self.units.pick(i as usize, n as usize);
}
return;
}
if let Some(n) = v.get("history_limit").and_then(|r| r.as_u64()) {
self.history_limit = n.clamp(1, 500) as u32;
self.refresh();
return;
}
if v.get("refresh").and_then(|r| r.as_bool()) == Some(true)
|| v.get("poll").and_then(|r| r.as_bool()) == Some(true)
{
self.refresh();
return;
}
}
if let Some(v) = &mut self.view {
Facet::update_json(v, msg_json);
}
}
}
/// **The backend-(2) seam.** Build the live forge-run pane as a `Box<dyn Panel>` a
/// host drops into a tab. `endpoint` is the `dwarves-server` url (e.g.
/// `http://127.0.0.1:7900`). The bearer comes from the dwarves client chain
/// ([`ClientCredential`]: `$DWARVES_SERVER_TOKEN`, then `~/.dwarves/token`) and an
/// optional pinned run id from `$DWARVES_RUN_ID` (empty ⇒ newest run). The
/// first fetch happens eagerly; an unreachable server (or a forge with no runs) yields
/// a clean *"forge unavailable"* pane, never a panic.
pub fn panel(endpoint: impl Into<String>) -> Box<dyn Panel> {
let token = ClientCredential::resolve().token;
let run_id = std::env::var("DWARVES_RUN_ID")
.ok()
.filter(|s| !s.trim().is_empty());
Box::new(ForgePanel::new(endpoint.into(), token, run_id))
}
/// Like [`panel`] but with the token + run id passed explicitly (for hosts that
/// manage their own credentials rather than reading the environment).
pub fn panel_with(
endpoint: impl Into<String>,
token: impl Into<String>,
run_id: Option<String>,
) -> Box<dyn Panel> {
Box::new(ForgePanel::new(endpoint.into(), token.into(), run_id))
}
#[cfg(test)]
mod tests {
use super::*;
/// ★ **The client chain names its rungs, and the FILE is one of them.**
///
/// The defect: every thin-client seam here, and `dwarves-gui`'s
/// `build_native_app`, read the bearer as
/// `std::env::var("DWARVES_SERVER_TOKEN").unwrap_or_default()` — the env rung and
/// nothing else — while the `dwarves` CLI has always walked
/// `nornir_service::resolve_token(--token, $DWARVES_SERVER_TOKEN, ~/.dwarves/token)`.
///
/// This is the PURE half: the ladder itself is asserted in nornir-service (where it
/// lives), and what is asserted here is that the two SENTENCES cannot be confused —
/// no credential configured versus token rejected — and that the no-credential one
/// offers the file rung by path instead of denying it.
#[test]
fn no_credential_and_token_rejected_are_different_sentences_with_different_repairs() {
const REFUSAL: &str = "status: Unauthenticated, message: \"The request does not \
have valid authentication credentials\"";
let none = ClientCredential {
token: String::new(),
whence: TokenSource::Nothing,
};
assert!(!none.is_configured());
let a = none
.explain_auth_failure(REFUSAL)
.expect("an auth refusal is classified");
assert!(a.contains("NO CREDENTIAL CONFIGURED"), "{a}");
assert!(a.contains("EMPTY bearer"), "{a}");
assert!(a.contains(TOKEN_ENV), "the env rung must be named: {a}");
assert!(
a.contains(".dwarves/token"),
"and the FILE rung, by path: {a}"
);
assert!(
a.contains("supported rung"),
"the file must be offered as usable — the old message DENIED having one: {a}"
);
assert!(
!a.contains(".nornir/token"),
"nornir's secret is not this client's: {a}"
);
let sent = ClientCredential {
token: "deadbeefdeadbeef".into(),
whence: TokenSource::File("/home/x/.dwarves/token".into()),
};
let b = sent
.explain_auth_failure(REFUSAL)
.expect("still an auth refusal");
assert!(b.contains("TOKEN REJECTED"), "{b}");
assert!(b.contains("16 chars"), "it must say what it sent: {b}");
assert!(
b.contains("/home/x/.dwarves/token"),
"…and from WHICH rung: {b}"
);
assert!(
!b.contains("NO CREDENTIAL"),
"the whole point: a rejected credential must not read as a missing one: {b}"
);
// The same rejected secret from the env rung reads differently — a stale token
// FILE and a stale ENVIRONMENT are different mistakes.
let from_env = ClientCredential {
token: "deadbeefdeadbeef".into(),
whence: TokenSource::Env(TOKEN_ENV.into()),
};
assert_ne!(b, from_env.explain_auth_failure(REFUSAL).unwrap());
// A non-auth failure keeps its own words. Over-matching here would hide every
// other cause behind a token story.
assert_eq!(
sent.explain_auth_failure("error: member clone failed: repository not found"),
None
);
// The provenance line never carries the secret — only the rung and the length.
assert!(
!sent.provenance().contains("deadbeef"),
"{}",
sent.provenance()
);
assert!(
sent.provenance().contains("16 chars"),
"{}",
sent.provenance()
);
assert!(
none.provenance().contains("no credential"),
"{}",
none.provenance()
);
}
/// ★ The token FILE is the dwarves one, spelled by the function that owns the
/// `.dwarves` name — never `.nornir`, and never a literal typed here.
#[test]
fn the_client_token_file_is_the_dwarves_file_under_home() {
let home = std::env::var_os("HOME").expect("oden always has HOME");
let p = token_file().expect("a set HOME yields a token file path");
assert_eq!(
p,
dwarves::store::forge_state_in(std::path::Path::new(&home)).join("token")
);
assert!(p.ends_with(".dwarves/token"), "{}", p.display());
assert!(!p.to_string_lossy().contains(".nornir"), "{}", p.display());
}
/// A `GetRunReport` reply with two steps wired `compile → test` reconstructs into
/// a projectable (`Dag`, `RunReport`) pair: the per-step rows AND the DAG `needs`
/// edges cross, so the reused viewer holds one node per step + one edge per `needs`
/// with the live per-node status. The crux of the thin/remote wiring.
#[test]
fn wire_report_reconstructs_the_dag_and_report() {
let report = pb::RunReport {
run: Some(pb::RunRow {
run_id: "run-1".into(),
pipeline: "diamond-ci".into(),
project: "demo".into(),
trigger: "manual".into(),
status: "passed".into(),
steps: 2,
passed: 2,
duration_ms: 42,
finished_at: "now".into(),
}),
steps: vec![
pb::BuildRow {
run_id: "run-1".into(),
step: "compile".into(),
stage: "build".into(),
backend: "variant".into(),
matrix: String::new(),
status: "passed".into(),
summary: String::new(),
artifact_id: String::new(),
job_ref: String::new(),
duration_ms: 20,
finished_at: "now".into(),
needs: vec![],
units_json: String::new(),
},
pb::BuildRow {
run_id: "run-1".into(),
step: "test".into(),
stage: "verify".into(),
backend: "container".into(),
matrix: String::new(),
status: "passed".into(),
summary: String::new(),
artifact_id: String::new(),
job_ref: String::new(),
duration_ms: 22,
finished_at: "now".into(),
needs: vec!["compile".into()],
units_json: String::new(),
},
],
release_json: Default::default(),
};
let (dag, run_report) = report_from_wire(&report);
assert_eq!(dag.order.len(), 2, "one DAG node per step");
assert_eq!(run_report.steps.len(), 2, "one BuildRow per step");
assert_eq!(run_report.run.run_id, "run-1");
assert_eq!(run_report.run.status, RunStatus::Passed);
let test = dag.order.iter().find(|n| n.step == "test").unwrap();
assert_eq!(
test.needs,
vec!["compile".to_string()],
"the `needs` DAG edge crosses"
);
assert_eq!(
test.backend, "container",
"backend maps to the &'static tag"
);
// The reused viewer holds the 2 build nodes + the 1 `needs` edge, all green.
let view = view_from_report(&dag, &run_report);
let (idle, running, done, failed) = view.counts();
assert_eq!(
(idle, running, done, failed),
(0, 0, 2, 0),
"all-green run → both nodes Done"
);
assert_eq!(view.state().nodes.len(), 2, "one node per step");
assert_eq!(view.state().edge_ends.len(), 1, "one edge per `needs` link");
facett_core::testmatrix::emit(
"dwarves-embed",
"get_run_report_reconstructs_dag",
done == 2 && view.state().edge_ends.len() == 1,
"a GetRunReport reply reconstructs the build Dag + RunReport client-side (per-step rows + `needs` edges) and renders on the reused facett-pipeline viewer",
);
}
/// A failed step keeps its `Failed` status and surfaces its summary as the node
/// error — a red run reconstructs honestly, never a false green.
#[test]
fn failed_step_reconstructs_as_failed_with_error() {
let report = pb::RunReport {
run: Some(pb::RunRow {
run_id: "run-2".into(),
pipeline: "p".into(),
project: "demo".into(),
trigger: "manual".into(),
status: "failed".into(),
steps: 1,
passed: 0,
duration_ms: 5,
finished_at: "now".into(),
}),
steps: vec![pb::BuildRow {
run_id: "run-2".into(),
step: "compile".into(),
stage: "build".into(),
backend: "variant".into(),
matrix: String::new(),
status: "failed".into(),
summary: "compile error E0001".into(),
artifact_id: String::new(),
job_ref: String::new(),
duration_ms: 5,
finished_at: "now".into(),
needs: vec![],
units_json: String::new(),
}],
release_json: Default::default(),
};
let (dag, run_report) = report_from_wire(&report);
assert_eq!(run_report.steps[0].status, RunStatus::Failed);
assert_eq!(run_report.steps[0].summary, "compile error E0001");
let view = view_from_report(&dag, &run_report);
let (_, _, _, failed) = view.counts();
assert_eq!(failed, 1, "the failed step is Failed on the reused viewer");
}
/// An offline endpoint yields the clean `Unavailable` state, never a panic — the
/// behaviour the design mandates. Uses a port nothing listens on.
#[test]
fn offline_server_is_a_clean_unavailable_state() {
let mut p = ForgePanel::new("http://127.0.0.1:0", "", None);
assert_eq!(p.conn, Conn::Unavailable);
let st = Facet::state_json(&p);
assert_eq!(st["status"], "unavailable");
assert_eq!(st["node_count"], 0);
assert!(st["run_id"].is_null(), "no run rendered when offline");
assert!(
st["error"].is_string(),
"the error is captured for the hover tooltip"
);
// Refresh on a still-dead server stays clean.
p.refresh();
assert_eq!(p.conn, Conn::Unavailable);
facett_core::testmatrix::emit(
"dwarves-embed",
"offline_forge_is_clean_unavailable",
p.conn == Conn::Unavailable,
"an unreachable dwarves-server → clean `forge unavailable` state, no panic (the read leg's non-panic discipline)",
);
}
#[test]
fn poll_due_respects_interval_and_toggle() {
let p = ForgePanel::new("http://127.0.0.1:0", "", None);
assert!(p.auto_poll);
assert!(!p.poll_due(0.5), "not due before one interval");
assert!(
p.poll_due(DEFAULT_POLL_INTERVAL_S),
"due at exactly one interval"
);
let mut p = p;
p.set_auto_poll(false);
assert!(!p.poll_due(9.0), "never due while paused");
}
/// The run-picker: a `History` reply's rows become picker entries (pure mapping,
/// order preserved) and selecting one PINS its run over the `{"run_id": …}` seam a
/// robot / the ComboBox click both drive; an empty selection clears back to
/// "newest". RED-when-broken: a broken mapping drops entries; a broken pin leaves
/// `selected_run_id` null. Offline-safe (no server needed for the picker surface).
#[test]
fn run_picker_lists_runs_and_pins_a_selection() {
let rows = vec![
pb::RunRow {
run_id: "run-3".into(),
pipeline: "diamond-ci".into(),
status: "passed".into(),
..Default::default()
},
pb::RunRow {
run_id: "run-2".into(),
pipeline: "diamond-ci".into(),
status: "failed".into(),
..Default::default()
},
pb::RunRow {
run_id: "run-1".into(),
pipeline: "smoke".into(),
status: "running".into(),
..Default::default()
},
];
let picks = run_picks_from_wire(&rows);
assert_eq!(picks.len(), 3, "one picker entry per historized run");
assert_eq!(picks[0].run_id, "run-3", "newest-first order preserved");
assert_eq!(picks[1].status, "failed", "the run's status label crosses");
assert_eq!(picks[2].pipeline, "smoke");
assert!(
run_picks_from_wire(&[]).is_empty(),
"no runs → empty picker"
);
// Inject the list into an offline pane and drive the picker over the seam.
let mut p = ForgePanel::new("http://127.0.0.1:0", "", None);
p.runs = picks.clone();
let st = Facet::state_json(&p);
assert_eq!(
st["run_count"], 3,
"the picker list is exposed for the dropdown"
);
assert_eq!(st["runs"][0]["run_id"], "run-3");
assert_eq!(st["history_limit"], DEFAULT_HISTORY_LIMIT);
assert!(
st["selected_run_id"].is_null(),
"unpinned ⇒ newest (null pin)"
);
// Pick a run (the ComboBox click drives this exact seam).
Facet::update_json(&mut p, r#"{"run_id":"run-2"}"#);
assert_eq!(
Facet::state_json(&p)["selected_run_id"],
"run-2",
"the picker pins the chosen run"
);
// Clear back to newest.
Facet::update_json(&mut p, r#"{"run_id":""}"#);
assert!(
Facet::state_json(&p)["selected_run_id"].is_null(),
"empty selection ⇒ newest"
);
// Resize the list request.
Facet::update_json(&mut p, r#"{"history_limit":5}"#);
assert_eq!(
Facet::state_json(&p)["history_limit"],
5,
"the picker list size is drivable"
);
facett_core::testmatrix::emit(
"dwarves-embed",
"forge_run_picker_pins_selection",
picks.len() == 3 && !run_picks_from_wire(&rows).is_empty(),
"History runs populate the Forge(live) run-picker; selecting one pins it over the update_json seam (empty ⇒ newest), list size drivable",
);
}
/// A pinned run whose `GetRunReport` fails must NOT blank the run-picker. When
/// `History` succeeded (the recent-run list is fresh) but the selected run's report
/// errored, the pane lands in the distinct `report_error` state, KEEPS the picker
/// list populated (so the user can switch to a working run) and surfaces the error —
/// it does not collapse to `unavailable` and throw the list away. Then switching to
/// a good run loads it and clears the error. RED-when-broken: the old all-or-nothing
/// fetch discarded the whole result on a report error, so `run_count` would be 0 and
/// the status `unavailable`. Empty forge (no runs to pick) stays plain `unavailable`.
/// Offline-safe: drives the real pane over the `state_json` seam via `apply_fetch`.
#[test]
fn a_report_error_keeps_the_picker_list_usable() {
let mut p = ForgePanel::new("http://127.0.0.1:0", "", Some("run-2".into()));
// History returned three runs, but the pinned run-2's GetRunReport failed.
let runs = vec![
pb::RunRow {
run_id: "run-3".into(),
pipeline: "diamond-ci".into(),
status: "passed".into(),
..Default::default()
},
pb::RunRow {
run_id: "run-2".into(),
pipeline: "diamond-ci".into(),
status: "failed".into(),
..Default::default()
},
pb::RunRow {
run_id: "run-1".into(),
pipeline: "smoke".into(),
status: "running".into(),
..Default::default()
},
];
p.apply_fetch(Ok(ForgeFetch {
runs,
report: Err(anyhow!("run-2 not found")),
}));
let st = Facet::state_json(&p);
assert_eq!(
st["status"], "report_error",
"a bad pin WITH a live list ⇒ recoverable report_error, not unavailable"
);
assert_eq!(
st["run_count"], 3,
"the fresh History list is KEPT so the picker stays usable"
);
assert_eq!(
st["runs"][0]["run_id"], "run-3",
"the list is the freshly-fetched one"
);
assert_eq!(
st["selected_run_id"], "run-2",
"the failing pin is preserved (user can switch away)"
);
assert!(
st["run_id"].is_null(),
"no run rendered while the report errors"
);
assert_eq!(st["node_count"], 0);
assert!(
st["error"].as_str().unwrap().contains("not found"),
"the report error is surfaced"
);
let recovered = st["status"] == "report_error" && st["run_count"] == 3;
// The user switches to a good run over the SAME seam the ComboBox drives; a good
// fetch loads it and clears the error.
let good = pb::RunReport {
release_json: Default::default(),
run: Some(pb::RunRow {
run_id: "run-3".into(),
pipeline: "diamond-ci".into(),
status: "passed".into(),
steps: 1,
passed: 1,
..Default::default()
}),
steps: vec![pb::BuildRow {
run_id: "run-3".into(),
step: "compile".into(),
stage: "build".into(),
backend: "variant".into(),
status: "passed".into(),
..Default::default()
}],
};
p.apply_fetch(Ok(ForgeFetch {
runs: vec![],
report: Ok(good),
}));
let st = Facet::state_json(&p);
assert_eq!(st["status"], "loaded", "a good report loads");
assert_eq!(st["run_id"], "run-3", "the good run renders");
assert!(st["error"].is_null(), "the error clears on success");
// Empty forge but CONNECTED (History returned zero runs) ⇒ the honest
// "no_runs" (connected, idle), distinct from a truly-offline "unavailable".
let mut e = ForgePanel::new("http://127.0.0.1:0", "", None);
e.apply_fetch(Ok(ForgeFetch {
runs: vec![],
report: Err(anyhow!("no runs on the forge yet")),
}));
assert_eq!(
Facet::state_json(&e)["status"],
"no_runs",
"connected + empty forge ⇒ no_runs, not unavailable"
);
facett_core::testmatrix::emit(
"dwarves-embed",
"report_error_keeps_picker_list",
recovered,
"a pinned run's GetRunReport error keeps the fresh History run-picker list (recoverable `report_error`, switch to a good run) instead of blanking to `unavailable`; empty forge stays `unavailable`",
);
}
/// **RED-WHEN-EMPTY live-data oracle** — the test the old port-0 test SHOULD have
/// been. It drives the Forge pane through the SAME `apply_fetch` path a live
/// `dwarves-server` fetch drives, feeding a real History (2 runs) + a real
/// `GetRunReport` (2 steps wired `compile → test`), then ASSERTS the pane shows
/// LIVE ROWS: `status == "loaded"`, the run-picker is populated (`run_count ≥ 1`),
/// and the DAG rendered real nodes + an edge (`node_count ≥ 1`, `edge_count ≥ 1`).
///
/// This is RED when the pane is empty/disconnected: a blank pane has
/// `node_count == 0` / `run_count == 0` / `status != "loaded"`, so "it drew but
/// drew nothing real" FAILS the test instead of passing (the exact hole in the
/// old assertion, which only checked the widget rendered against a dead port 0).
/// The paired negative leg pins the oracle: the same pane fed an EMPTY connected
/// fetch lands in `no_runs` with zero nodes — the state the assertion rejects.
#[test]
fn robot_forge_shows_live_rows_red_when_empty() {
// A live-ish fetch: History with 2 runs + the newest run's 2-step report.
let runs = vec![
pb::RunRow {
run_id: "run-9".into(),
pipeline: "diamond-ci".into(),
status: "passed".into(),
steps: 2,
passed: 2,
..Default::default()
},
pb::RunRow {
run_id: "run-8".into(),
pipeline: "smoke".into(),
status: "running".into(),
..Default::default()
},
];
let report = pb::RunReport {
release_json: Default::default(),
run: Some(pb::RunRow {
run_id: "run-9".into(),
pipeline: "diamond-ci".into(),
status: "passed".into(),
steps: 2,
passed: 2,
..Default::default()
}),
steps: vec![
pb::BuildRow {
run_id: "run-9".into(),
step: "compile".into(),
stage: "build".into(),
backend: "variant".into(),
status: "passed".into(),
needs: vec![],
..Default::default()
},
pb::BuildRow {
run_id: "run-9".into(),
step: "test".into(),
stage: "verify".into(),
backend: "container".into(),
status: "passed".into(),
needs: vec!["compile".into()],
..Default::default()
},
],
};
let mut p = ForgePanel::new("http://127.0.0.1:0", "", None);
// Feed the LIVE fetch through the real state-application path.
p.apply_fetch(Ok(ForgeFetch {
runs,
report: Ok(report),
}));
let st = Facet::state_json(&p);
// The pane shows LIVE ROWS — the assertion that goes RED on a blank pane.
let shows_live_rows = st["status"] == "loaded"
&& st["run_count"].as_u64().unwrap() >= 1 // run-picker populated
&& st["node_count"].as_u64().unwrap() >= 1 // DAG has real nodes
&& st["edge_count"].as_u64().unwrap() >= 1 // …and a real `needs` edge
&& st["run_id"] == "run-9"; // the newest run rendered
assert!(
shows_live_rows,
"the Forge pane must show LIVE rows from the server, not a blank pane: {st}"
);
// Negative leg — pin the oracle: an EMPTY connected forge is `no_runs` with
// zero nodes, exactly the state `shows_live_rows` rejects (so the test can
// only pass when real data is present).
let mut empty = ForgePanel::new("http://127.0.0.1:0", "", None);
empty.apply_fetch(Ok(ForgeFetch {
runs: vec![],
report: Err(anyhow!("no runs on the forge yet")),
}));
let est = Facet::state_json(&empty);
let empty_is_red =
est["status"] == "no_runs" && est["node_count"] == 0 && est["run_count"] == 0;
assert!(
empty_is_red,
"an empty forge is the no_runs/zero-node state the live oracle rejects: {est}"
);
facett_core::testmatrix::emit(
"dwarves-embed",
"forge_shows_live_rows_red_when_empty",
shows_live_rows && empty_is_red,
"Forge pane driven through apply_fetch shows LIVE rows (loaded + run_count/node_count/edge_count ≥ 1, newest run rendered); RED when the pane is empty/disconnected (no_runs, zero nodes) — closes the port-0 'drew nothing real' hole",
);
}
#[test]
fn update_json_toggles_auto_poll_and_forces_poll() {
let mut p = ForgePanel::new("http://127.0.0.1:0", "", None);
Facet::update_json(&mut p, r#"{"auto_poll": false}"#);
assert_eq!(Facet::state_json(&p)["auto_poll"], serde_json::json!(false));
// A forced poll on the dead server stays clean (never wedges / panics).
Facet::update_json(&mut p, r#"{"poll": true}"#);
assert_eq!(Facet::state_json(&p)["status"], "unavailable");
}
}