crates/dwarves-embed/src/jobsfleet.rs
//! **The Live Jobs + Agent-fleet pane** — the CI forge cockpit's "what is running
//! where" (separation-spec item **B3**).
//!
//! Two live legs:
//!
//! 1. **Jobs** — fed LIVE from the shared **jera JobRecord ledger** (edda's single
//! source-of-truth for "what ran"). We read it with [`jera::jobs::JobStore::open_read_only`]
//! (the lock-tolerant snapshot read nornir's viz uses, so it coexists with a live
//! writer) and map each [`JobRecord`] onto a [`facett_jobview::JobEntry`] — REUSING
//! the shared [`JobList`] widget (L5, no hand-rolled jobs list). Crucially we surface
//! each job's **`node`** (the fleet machine that ran it — Loki/Odin/NUC/Njord), the
//! multi-machine placement the jera scheduler stamps.
//! 2. **Agent-fleet** — the worker roster + heartbeat/online state over the
//! **`AgentPool` RPC** (thin gRPC to a running `dwarves-server`, mirroring
//! [`crate::ForgePanel`]'s `History`/`GetRunReport` fetch). Heavy work places on
//! Loki/Odin (never the NUC) per the jera placement rules; the roster shows each
//! agent's group/caps/cores and its idle/busy liveness.
//!
//! Auto-polls on an interval like the [`ForgePanel`](crate::ForgePanel), and drives
//! headless over the `state_json`/`update_json` seam (the same channel a
//! `nornir-robotui` scenario and a thin gRPC client use) — so the pane is robot-tested
//! without pixels.
//!
//! ## Both fixtures are gone (2026-09-10 — the ninth pane)
//! An empty/unreadable ledger used to mount `fixture_records()` and a **failed**
//! `AgentPool` RPC used to mount `fixture_agents()`, each marked `source: "fixture"`.
//!
//! The fleet half was the worst instance in the whole family. The `Err` arm installed a
//! healthy three-machine roster — Loki idle, Odin busy, the NUC idle — and tucked the
//! transport error into a side field behind a `⚠` hover. So the pane painted a working
//! fleet at precisely the moment the fleet could not be reached: **it lied exactly when
//! something was broken, which is the one moment it exists for.** A roster with green
//! online dots IS a claim that those machines answered, whatever the caption says.
//!
//! Now: rows or a NAMED refusal ([`facett_core::empty::EmptyState`]), never a fixture —
//! and the empties are kept apart, because `unreachable` (we could not ask) and
//! `no-rows` (we asked and the answer was none) are different facts. On the jobs leg
//! that distinction is [`LedgerRead`]'s three-way `Absent`/`Empty`/`Rows`, which this
//! pane now carries all the way into the render instead of flattening to a `Vec`.
use std::path::{Path, PathBuf};
use std::time::Duration;
use anyhow::{Context, Result};
use facett_core::empty::EmptyState;
use facett_core::Facet;
use facett_core::Panel;
use facett_jobview::{JobEntry, JobList, JobStatus};
use jera::jobs::{JobRecord, JobSelector, JobStore};
use crate::pb;
/// Default seconds between live auto-poll re-reads of the ledger + fleet.
pub const DEFAULT_JOBS_POLL_INTERVAL_S: f64 = 2.0;
/// Default cap on how many most-recent ledger jobs the pane reads.
pub const DEFAULT_LEDGER_LIMIT: usize = 100;
/// The node label shown for a job the ledger records with `node = None` — a job that
/// ran on the local control-plane (not dispatched across the fleet).
pub const LOCAL_NODE: &str = "local";
// ── ledger read + pure mapping (jera JobRecord → facett JobEntry) ─────────────
/// Resolve the jera ledger root the pane reads: `$DWARVES_JOBS_DIR`, else
/// `<warehouse>/jobs` where the warehouse mirrors `dwarves-server`'s
/// `default_warehouse` (`$DWARVES_WAREHOUSE`, else `~/.dwarves/warehouse`). This is the
/// same `jobs.redb` a `dwarves run` / `dwarves-server` writes.
pub fn default_ledger_root() -> PathBuf {
if let Some(d) = std::env::var_os("DWARVES_JOBS_DIR") {
if !d.is_empty() {
return PathBuf::from(d);
}
}
match std::env::var("DWARVES_WAREHOUSE") {
Ok(w) if !w.is_empty() => PathBuf::from(w),
_ => match std::env::var("HOME") {
Ok(h) if !h.is_empty() => PathBuf::from(h).join(".dwarves").join("warehouse"),
_ => PathBuf::from(".dwarves").join("warehouse"),
},
}
}
/// **What a ledger read actually found** — three outcomes that must never render alike.
///
/// `read_ledger_jobs` used to return `Ok(Vec::new())` for an absent `jobs.redb` and for
/// an empty one, with a comment calling that deliberate. It is the defect the owner named
/// on 2026-08-25: **nuke the workspace and its jobs are nuked with it**, so a ledger that
/// is GONE is the normal end state of a removed workspace — and collapsing it into "no
/// jobs" makes a deleted workspace indistinguishable from a quiet forge. In `jera.rs` it
/// was worse than indistinguishable: an empty read fell through to the demo fixture, so a
/// purged workspace rendered **fabricated jobs** as though they were real.
///
/// Measured on this box 2026-08-25: an `--uninstall`/`--install` cycle wiped
/// `~/.nornir/warehouse/`, `~/.nornir/workspaces/default/builds/` and `.../nordisk/builds/`
/// while printing `1 item(s) wiped, 0 workspace(s) deregistered`. Three trees went; the
/// receipt counted one. A reader that cannot tell *gone* from *empty* turns that
/// miscount into a UI that looks fine.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum LedgerRead {
/// No `jobs.redb` at this root at all. The workspace was never populated, or it was
/// removed and took its ledger with it. **Never render this as "no jobs".**
Absent,
/// The ledger is there and readable and holds no rows: a real, live, idle forge.
Empty,
/// The newest rows, newest-first, capped at the caller's limit.
Rows(Vec<JobRecord>),
}
impl LedgerRead {
/// The rows, if any — `Absent` and `Empty` both yield an empty slice. Use this ONLY
/// where the three states are already distinguished elsewhere; reaching for it to
/// "simplify" is how the collapse got reintroduced.
pub fn rows(&self) -> &[JobRecord] {
match self {
LedgerRead::Rows(r) => r,
_ => &[],
}
}
/// True when there is no ledger file — the workspace is gone (or never was).
pub fn is_absent(&self) -> bool {
matches!(self, LedgerRead::Absent)
}
/// A short, honest word for the state, for a header or a robot oracle.
pub fn key(&self) -> &'static str {
match self {
LedgerRead::Absent => "absent",
LedgerRead::Empty => "empty",
LedgerRead::Rows(_) => "rows",
}
}
}
/// Read the most-recent ledger jobs (newest-first, capped at `limit`) from the jera
/// `JobStore` at `root`, as the three-way [`LedgerRead`].
///
/// An absent `jobs.redb` is [`LedgerRead::Absent`] — **not** an error and **not** an
/// empty list. Uses the lock-tolerant [`JobStore::open_read_only`] so it coexists with a
/// live writer; a genuine open/list failure is still `Err`, because a broken read is a
/// fourth thing again and must not read as any of the three.
///
/// # "Lock-tolerant" is true, and it is not enough — MEASURED, t14s 2026-08-25
///
/// The standing diagnosis was that this call *is not* lock-tolerant and that the word
/// above is a false claim. It is not false. Three probes against a live writer:
///
/// | probe | result |
/// |---|---|
/// | a second plain `JobStore::open` | `Database already open. Cannot acquire lock.` |
/// | `JobStore::open_read_only`, same live writer | **Ok**, rows returned — it copies the file aside and opens the copy |
/// | `open_read_only` on a `0o444` ledger | `I/O error: Permission denied (os error 13)` |
///
/// So the lock is handled. What is NOT handled is **ownership**: `open` routes through
/// `redb::Database::create`, so a ledger this process may read but not WRITE fails before
/// jera's `is_lock_error` is ever consulted and the snapshot fallback never runs. The
/// installed forge runs as user `dwarves` and writes `-rw-r--r--`; a UI started by a human
/// does not run as `dwarves`. And the snapshot that makes it lock-tolerant is a whole-file
/// byte copy — paid on **every** read, at the Jera tab's 2 s cadence.
///
/// That is why the shipped Jera tab reads `Viz.Jobs` over gRPC
/// (`dwarves_gui::jera::WireJobs`) and this function is the FILE road, for a host that
/// genuinely owns the ledger it is pointing at. Keeping it is not a twin of the wire leg:
/// they answer different questions (*this path on this disk* vs *the served workspace*),
/// and they share the one `JobRecord → JobEntry` mapper below.
/// # `kind` — the SELECT, and why it has to happen before the truncate
///
/// `kind` is a [`jera::jobs::kind`] constant, or `""` for every class. It is not a
/// convenience: it is the whole difference between the file road and a road that cannot
/// see a class.
///
/// MEASURED on the live forge 2026-08-27, the server's own 200-row window against a
/// kind-scoped read of the SAME ledger: `snapshot_member` 57, `scip_member` 47,
/// `knowledge_member` 38 = **142 of 200 (71 %)**, `feed` **6 of 84 (7.1 %)**,
/// `workspace_populate` **0 of 1 — invisible**. Those rows were never truncated *away* by
/// a filter; they were never READ, because the window closed before anything asked what
/// class it wanted. Filtering the result afterwards cannot recover a row the read did not
/// return, and widening the window (this function's callers used to ask for 20x the limit
/// and then `retain`) only moves the odds: a class holding 0.5 % of the ledger needs
/// roughly a 200x window to be sure of one row.
///
/// So the select is done here, on the whole scan, and the caller's `limit` is applied to
/// the rows that survived it. The predicate is [`JobSelector::matches`] — jera's own, not
/// a `r.kind == kind` twin that drifts when a selector variant is added (LAW 5).
///
/// **[`LedgerRead::Empty`] stays a question about the LEDGER.** It is decided from the
/// unscoped scan, BEFORE the class window: a ledger holding 800 rows and none of the
/// asked-for kind is `Rows(vec![])` ("0 jobs of this class beside a live source"), never
/// `Empty` ("the forge is idle"). Collapsing those two is the same defect as collapsing
/// `Absent` and `Empty`, one axis over.
pub fn read_ledger_jobs(root: &Path, limit: usize, kind: &str) -> Result<LedgerRead> {
if !root.join("jobs.redb").exists() {
return Ok(LedgerRead::Absent);
}
let store = JobStore::open_read_only(root)
.with_context(|| format!("open jera ledger at {}", root.display()))?;
// ONE table scan answers both facts. `list` already sorts newest-first.
let all = store.list(&JobSelector::All).context("read jera ledger")?;
if all.is_empty() {
return Ok(LedgerRead::Empty);
}
let mut jobs = if kind.is_empty() {
all
} else {
let sel = JobSelector::Kind(kind.to_string());
all.into_iter().filter(|r| sel.matches(r)).collect()
};
jobs.truncate(limit);
Ok(LedgerRead::Rows(jobs))
}
/// The node label a ledger record surfaces — its stamped fleet machine, or
/// [`LOCAL_NODE`] when it ran on the local control-plane (`node = None`).
fn node_label(r: &JobRecord) -> String {
r.node
.clone()
.filter(|n| !n.is_empty())
.unwrap_or_else(|| LOCAL_NODE.to_string())
}
/// A short one-line detail for a ledger job, prefixed with its **node** so the
/// placement is visible in the row itself (the shared `JobList` renders `detail` but
/// not `meta`). Falls back to the job kind when no richer detail is recorded.
fn record_detail(r: &JobRecord) -> String {
let node = node_label(r);
// Try to pull a compact summary from the free-form detail_json; else use the kind.
let body = serde_json::from_str::<serde_json::Value>(&r.detail_json)
.ok()
.and_then(|v| {
v.get("pipeline")
.and_then(|p| p.as_str())
.map(|p| p.to_string())
.or_else(|| {
v.get("summary")
.and_then(|s| s.as_str())
.map(|s| s.to_string())
})
})
.filter(|s| !s.is_empty())
.unwrap_or_else(|| r.kind.clone());
// WHO asked, next to WHERE it ran. Without this the panel showed a release and
// a bench with no way to tell an operator's own run from one an external client
// started. `actor_label()` says `unknown` for rows that predate the field or
// that nobody stamped — it never invents an initiator.
format!("[{node}] {body} · by {}", r.actor_label())
}
/// **Pure** map one jera [`JobRecord`] onto a shared [`JobEntry`], SURFACING the node
/// placement (into both the visible `detail` and `meta`). jera's status strings
/// (`queued`/`running`/`done`/`failed`) map 1:1 onto [`JobStatus`]. Unit-testable with
/// no ledger/network.
pub fn entry_from_record(r: &JobRecord) -> JobEntry {
let node = node_label(r);
let mut meta = vec![
("node".to_string(), node),
("by".to_string(), r.actor_label()),
];
if !r.workspace.is_empty() {
meta.push(("workspace".to_string(), r.workspace.clone()));
}
if !r.result_ref.is_empty() {
meta.push(("result".to_string(), r.result_ref.clone()));
}
let entry = JobEntry {
target: r.target.clone(),
started_micros: r.ts_start_micros,
elapsed_ms: r.elapsed_ms,
detail: record_detail(r),
meta,
parent_id: r.parent_id.clone(),
..JobEntry::new(
r.job_id.clone(),
r.kind.clone(),
JobStatus::parse(&r.status),
)
};
// **PHASE — the one field a running job changes while you watch it, and the one this
// mapper used to drop on the floor.**
//
// `JobRecord::phase` is written durably by jera's `daemon_report_sink` when a
// `Reporter::phase` fires, and jera's `upsert` even grew a preserve limb so a
// `finish()` cannot erase it (that bug was fixed the same day). ALL of that
// persistence work reaches a pane through THIS function, and it did not carry the
// field: the reporter wrote `cutting`, redb stored `cutting`, `open_read_only` handed
// it back — and the panel rendered a row that said only `running`. A durable value
// nothing surfaces is indistinguishable from one nothing writes.
//
// It goes in `meta` rather than a new `JobEntry` field on purpose, and the reason is
// facett-jobview's own (`with_progress`'s doc): `meta` is the declared channel for
// "arbitrary extra key/values the host wants to surface", and adding a non-defaulted
// struct field breaks every full-struct-literal construction site in every host
// (korp, njord, nornir). Stamped through `JobEntry::with_phase` and NOT by a
// `meta.push(("phase", …))` here: `nornir::viz::nornir_tab::job_entry` drops the same
// field and has to be fixed too, and two hosts pushing the same key by hand is the
// twin LAW 5 forbids. One writer owns the spelling AND the empty-value rule.
//
// **NOT surfaced (named, not hidden):** `DaemonPhase::healthy`. The row already
// carries a status colour and there is no honest second channel for an amber
// "running but unhealthy" in `JobStatus` today; inventing one here would be a UI
// decision made inside a mapper. `fraction()` IS carried — it is the durable mirror
// of the progress tick and `with_progress` is the existing writer for exactly it.
let Some(dp) = r.daemon_phase() else {
return entry;
};
let entry = entry.with_phase(Some(dp.name.as_str()));
match dp.fraction() {
Some(f) => entry.with_progress(f),
None => entry,
}
}
/// Pure map a whole ledger snapshot onto the shared widget's entries.
pub fn entries_from_records(recs: &[JobRecord]) -> Vec<JobEntry> {
recs.iter().map(entry_from_record).collect()
}
/// **The WIRE leg's decode** — `Viz.Jobs`' JSON rows onto the SAME [`JobEntry`]s the
/// file read produces, through the SAME [`entries_from_records`] mapper.
///
/// `Viz.Jobs` serializes `nornir::jobs::JobRecord`, and `nornir-jobs` is a thin façade
/// that re-exports `jera::jobs` verbatim — so the type on the wire is not a structural
/// twin of [`JobRecord`], it IS [`JobRecord`]. That is what lets the remote and local
/// roads share one mapper instead of growing a second one that could drift (L5).
///
/// A row that will not deserialize is an `Err`, never a skipped row: a ledger that
/// half-decodes must not render as a shorter ledger. The caller turns that into a NAMED
/// failure, and a named failure keeps the previous reading on screen.
pub fn entries_from_records_json(rows: &[serde_json::Value]) -> Result<Vec<JobEntry>> {
let recs = rows
.iter()
.enumerate()
.map(|(i, v)| {
serde_json::from_value::<JobRecord>(v.clone())
.with_context(|| format!("decode Viz.Jobs row {i} as a jera JobRecord"))
})
.collect::<Result<Vec<_>>>()?;
Ok(entries_from_records(&recs))
}
/// The distinct fleet nodes present across a ledger snapshot (order of first
/// appearance) — the "which machines ran work" roll-up the pane surfaces.
fn distinct_nodes(recs: &[JobRecord]) -> Vec<String> {
let mut seen: Vec<String> = Vec::new();
for r in recs {
let n = node_label(r);
if !seen.contains(&n) {
seen.push(n);
}
}
seen
}
// ── the agent fleet (AgentPool RPC) ───────────────────────────────────────────
/// One worker-fleet row the pane renders — the domain projection of a wire
/// [`pb::AgentRow`], with a derived `online` flag from the liveness `state`.
#[derive(Clone, Debug, PartialEq, Eq, serde::Serialize)]
pub struct AgentRow {
pub id: String,
pub group: String,
pub caps: Vec<String>,
pub cores: u32,
pub mem_mb: u64,
/// jera liveness (`idle`/`busy`/…).
pub state: String,
/// Derived: the agent is reachable/online (any non-empty, non-`offline` state).
pub online: bool,
}
/// Pure map the `AgentPool` reply rows onto the fleet roster (derives `online` from
/// the liveness state). Order preserved. Unit-testable with stub rows (no server).
pub fn agents_from_wire(rows: &[pb::AgentRow]) -> Vec<AgentRow> {
rows.iter()
.map(|a| {
let state = a.state.clone();
let online = !state.is_empty()
&& !state.eq_ignore_ascii_case("offline")
&& !state.eq_ignore_ascii_case("down");
AgentRow {
id: a.id.clone(),
group: a.group.clone(),
caps: a.caps.clone(),
cores: a.cores,
mem_mb: a.mem_mb,
state,
online,
}
})
.collect()
}
/// 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}")
}
}
/// Fetch the worker fleet from `dwarves-server` at `endpoint` over the thin `AgentPool`
/// gRPC (Bearer `token`, empty ⇒ no real auth). Runs on a private current-thread
/// runtime so it is safe from the synchronous egui loop; a 3s connect timeout keeps a
/// dead server from wedging the UI thread. An unreachable server / errored RPC returns
/// `Err` (the pane then shows the fixture roster) — never a panic.
pub fn fetch_agents(endpoint: &str, token: &str) -> Result<Vec<pb::AgentRow>> {
let rt = tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()
.context("build tokio runtime for fleet client")?;
let endpoint = norm_endpoint(endpoint);
let token = token.to_string();
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)
},
);
let agents = client
.agent_pool(pb::Empty {})
.await
.context("Dwarf.AgentPool RPC")?
.into_inner()
.agents;
Ok(agents)
})
}
// ── the pane ──────────────────────────────────────────────────────────────────
/// The jobs feed, named the way an operator would go look it up.
const JOBS_FEED: &str = "jera JobRecord ledger (jobs.redb)";
/// The fleet feed.
const FLEET_FEED: &str = "dwarves-server AgentPool RPC";
/// Where the rendered jobs came from. `Fixture` is gone (2026-09-10): the only two
/// answers are the live ledger and an honest nothing, and the nothing carries a
/// [`EmptyState`] saying WHICH of the three ledger states produced it.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
enum JobsSource {
Ledger,
Empty,
}
impl JobsSource {
fn key(self) -> &'static str {
match self {
JobsSource::Ledger => "ledger",
JobsSource::Empty => "empty",
}
}
}
/// Where the rendered fleet came from — the live `AgentPool` RPC, or nothing.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
enum FleetSource {
Pool,
Empty,
}
impl FleetSource {
fn key(self) -> &'static str {
match self {
FleetSource::Pool => "pool",
FleetSource::Empty => "empty",
}
}
}
/// One refresh's raw inputs — split so the pane's state transition is drivable headless
/// in tests (the same `state_json` seam a robot reads) without a redb ledger or a live
/// server.
///
/// `jobs` carries the **whole** [`LedgerRead`], not its rows. It used to be a
/// `Vec<JobRecord>`, and flattening it here is what let `apply` mount a fixture: once
/// `Absent`, `Empty` and a read error all arrive as an empty vec, the pane cannot tell
/// *the workspace is gone* from *the forge is idle*, and it picked the cheerful reading.
/// `LedgerRead`'s own doc warns about exactly this — "reaching for [`rows`] to
/// 'simplify' is how the collapse got reintroduced" — and `refresh` was reaching for it.
pub(crate) struct FleetSnapshot {
pub jobs: Result<LedgerRead>,
pub agents: Result<Vec<pb::AgentRow>>,
}
/// The Live Jobs + Agent-fleet cockpit pane. A [`Facet`] (hence a [`Panel`]).
pub struct JobsFleetPanel {
endpoint: String,
token: String,
ledger_root: PathBuf,
limit: usize,
/// The shared jobs widget, fed live (or from the fixture).
inner: JobList,
jobs_source: JobsSource,
fleet_source: FleetSource,
/// Set when there are no jobs to draw — the NAMED empty state, distinguishing the
/// gone workspace (`not-configured`) from the idle forge (`no-rows`) from the
/// unreadable ledger (`unreachable`). This is what replaced `fixture_records()`.
jobs_empty: Option<EmptyState>,
/// Set when there is no fleet to draw. `unreachable` when the `AgentPool` RPC
/// failed — the case that used to install `fixture_agents()`.
fleet_empty: Option<EmptyState>,
/// The rendered ledger job count + node roll-up (surfaced in `state_json`).
job_count: usize,
nodes: Vec<String>,
/// The rendered fleet roster.
agents: Vec<AgentRow>,
/// The last errors (folded into `state_json` for the hover tooltip + tests).
ledger_error: Option<String>,
fleet_error: Option<String>,
auto_poll: bool,
poll_interval_s: f64,
last_poll_s: f64,
}
impl JobsFleetPanel {
/// Build the pane for a `dwarves-server` `endpoint` + a jera `ledger_root`, and
/// eagerly do the first refresh (read the ledger + fetch the fleet). Offline / empty
/// ⇒ the fixture fallbacks, never a panic.
pub fn new(
endpoint: impl Into<String>,
token: impl Into<String>,
ledger_root: impl Into<PathBuf>,
) -> Self {
let mut p = Self {
endpoint: endpoint.into(),
token: token.into(),
ledger_root: ledger_root.into(),
limit: DEFAULT_LEDGER_LIMIT,
inner: JobList::new("dwarves — jobs (jera ledger · agent-aware)"),
jobs_source: JobsSource::Empty,
fleet_source: FleetSource::Empty,
jobs_empty: None,
fleet_empty: None,
job_count: 0,
nodes: Vec::new(),
agents: Vec::new(),
ledger_error: None,
fleet_error: None,
auto_poll: true,
poll_interval_s: DEFAULT_JOBS_POLL_INTERVAL_S,
last_poll_s: 0.0,
};
p.refresh();
p
}
/// True when auto-poll is on AND at least one interval has elapsed. 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-)read the ledger + (re-)fetch the fleet and apply the outcome.
pub fn refresh(&mut self) {
// The WHOLE three-way read goes through — `apply` needs to tell `Absent` from
// `Empty`, and `.rows()` here would throw that away (see `FleetSnapshot`).
let jobs = read_ledger_jobs(&self.ledger_root, self.limit, "");
let agents = fetch_agents(&self.endpoint, &self.token);
self.apply(FleetSnapshot { jobs, agents });
}
/// Fold a refresh snapshot into the pane's rendered state. Split from [`refresh`] so
/// the transitions drive headless in tests.
///
/// **Rows or a named empty, never a fixture** (2026-09-10). Each leg has exactly one
/// live outcome and a set of empty ones, and the empties are kept APART because they
/// are different facts about the world:
///
/// | leg | outcome | rendered |
/// |---|---|---|
/// | jobs | `Rows` | the ledger's jobs, `source: "ledger"` |
/// | jobs | `Empty` | nothing, `no-rows` — a real, live, idle forge |
/// | jobs | `Absent` | nothing, `not-configured` — the workspace is GONE, which is not an idle forge |
/// | jobs | `Err` | nothing, `unreachable` — the ledger is there and would not open |
/// | fleet | `Ok(rows)` | the roster, `source: "pool"` |
/// | fleet | `Ok([])` | nothing, `no-rows` — the pool answered, it has no agents |
/// | fleet | `Err` | nothing, `unreachable` — **this is the one that used to draw a healthy fleet** |
pub(crate) fn apply(&mut self, snap: FleetSnapshot) {
// ── jobs leg ──
let (rows, jobs_empty, ledger_error) = match snap.jobs {
Ok(LedgerRead::Rows(r)) if !r.is_empty() => (r, None, None),
// `Rows(vec![])` cannot happen from `read_ledger_jobs`, but a caller could
// build one; treat it as the idle forge it describes rather than panicking.
Ok(LedgerRead::Rows(_)) | Ok(LedgerRead::Empty) => (
Vec::new(),
Some(
EmptyState::no_rows(
JOBS_FEED,
format!(
"{} is readable and holds no jobs",
self.ledger_root.display()
),
)
.with_hint("run something (`dwarves run …`)"),
),
None,
),
Ok(LedgerRead::Absent) => {
let msg = format!(
"no jera ledger at {} — the workspace is gone (or was never \
populated). This is NOT an idle forge.",
self.ledger_root.display()
);
(
Vec::new(),
Some(
EmptyState::not_configured(JOBS_FEED, msg.clone())
.with_hint("populate the workspace, or point $DWARVES_JOBS_DIR at it"),
),
Some(msg),
)
}
Err(e) => {
let msg = format!("{e:#}");
(
Vec::new(),
Some(EmptyState::unreachable(JOBS_FEED, msg.clone())),
Some(msg),
)
}
};
self.job_count = rows.len();
self.nodes = distinct_nodes(&rows);
self.inner.set_jobs(entries_from_records(&rows));
self.jobs_source = if jobs_empty.is_none() {
JobsSource::Ledger
} else {
JobsSource::Empty
};
self.jobs_empty = jobs_empty;
self.ledger_error = ledger_error;
// ── fleet leg ──
match snap.agents {
Ok(rows) if !rows.is_empty() => {
self.agents = agents_from_wire(&rows);
self.fleet_source = FleetSource::Pool;
self.fleet_empty = None;
self.fleet_error = None;
}
Ok(_) => {
// The server answered and has no agents. That is a live, true fact about
// the fleet — distinct from not having reached it.
self.agents = Vec::new();
self.fleet_source = FleetSource::Pool;
self.fleet_empty = Some(
EmptyState::no_rows(FLEET_FEED, "the pool answered with no agents")
.with_hint("start an agent, or check the forge's agent registration"),
);
self.fleet_error = None;
}
Err(e) => {
// THE LIE THAT WAS HERE: `agents_from_wire(&fixture_agents())`, so an
// unreachable pool rendered three healthy machines. An `AgentPool` that
// will not answer says NOTHING about fleet health; drawing a fleet is
// the fabrication.
let msg = format!("{e:#}");
self.agents = Vec::new();
self.fleet_source = FleetSource::Empty;
self.fleet_empty = Some(EmptyState::unreachable(FLEET_FEED, msg.clone()));
self.fleet_error = Some(msg);
}
}
}
fn online_count(&self) -> usize {
self.agents.iter().filter(|a| a.online).count()
}
}
impl Facet for JobsFleetPanel {
fn title(&self) -> &str {
"Jobs (live)"
}
fn ui(&mut self, ui: &mut egui::Ui) {
// LIVE auto-poll: re-read the ledger + re-fetch the fleet on the interval.
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(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("re-read the jera ledger + re-fetch the agent fleet")
.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;
}
ui.separator();
// Jobs source + node roll-up (PLAIN NUMBERS — a nordisk law).
let src = if self.jobs_source == JobsSource::Ledger {
"jera ledger"
} else {
"—"
};
ui.label(format!("{} jobs · {}", self.job_count, src));
if !self.nodes.is_empty() {
ui.weak(format!("nodes: {}", self.nodes.join(" · ")));
}
});
// The agent-fleet roster strip (id · group · caps · cores · liveness).
ui.separator();
let fleet_src = if self.fleet_source == FleetSource::Pool {
"AgentPool"
} else {
"—"
};
ui.horizontal(|ui| {
ui.strong("fleet");
ui.weak(format!(
"{} agents · {} online · {}",
self.agents.len(),
self.online_count(),
fleet_src
));
});
if let Some(e) = &self.fleet_empty {
// No roster ⇒ no machine rows, no ● online dots. An unreachable pool must
// not be painted as a fleet.
e.ui(ui);
} else {
egui::Grid::new("dwarves-embed-fleet")
.num_columns(5)
.striped(true)
.show(ui, |ui| {
for a in &self.agents {
let dot = if a.online { "●" } else { "○" };
ui.label(format!("{dot} {}", a.id));
ui.weak(&a.group);
ui.weak(a.caps.join(","));
ui.weak(format!("{}c", a.cores));
ui.weak(&a.state);
ui.end_row();
}
});
}
if do_refresh {
self.refresh();
self.last_poll_s = now_s;
}
// The shared jobs widget below (the live task-manager tree) — or, with no jobs,
// the NAMED reason there are none. `not-configured` (the workspace is gone) reads
// very differently from `no-rows` (an idle forge), and that is the whole point.
ui.separator();
if let Some(e) = &self.jobs_empty {
e.ui(ui);
} else {
Facet::ui(&mut self.inner, ui);
}
}
fn state_json(&self) -> serde_json::Value {
serde_json::json!({
"kind": "jobs_fleet",
"endpoint": self.endpoint,
"ledger_root": self.ledger_root.display().to_string(),
// Jobs leg — the live ledger and its node placement roll-up. `"fixture"` is
// gone from `source`: the only values are `ledger` and `empty`, and `empty`
// carries its own named block naming which of the three no-jobs states it is.
"source": self.jobs_source.key(),
"jobs_feed": JOBS_FEED,
"jobs_empty": self.jobs_empty.as_ref().map(|e| e.state_json()),
"job_count": self.job_count,
"nodes": self.nodes,
"node_count": self.nodes.len(),
"ledger_error": self.ledger_error,
// Agent-fleet leg — the roster + online liveness.
"fleet_source": self.fleet_source.key(),
"fleet_feed": FLEET_FEED,
"fleet_empty": self.fleet_empty.as_ref().map(|e| e.state_json()),
"agent_count": self.agents.len(),
"online_agents": self.online_count(),
"agents": self.agents,
"fleet_error": self.fleet_error,
// Poll state.
"auto_poll": self.auto_poll,
"poll_interval_s": self.poll_interval_s,
// The shared JobList's own rich (fold-aware) state — the rendered tree.
"jobs": Facet::state_json(&self.inner),
})
}
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} → re-read the ledger + re-fetch the fleet
// {"auto_poll": bool} → toggle live polling
// Anything else is forwarded to the inner JobList (its sort/fold/select/filter
// JSON→Msg seam is host-supplied, so we route via `Msg` for the common controls).
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 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;
}
// Route the shared jobs-list controls onto its typed Msg (the widget takes
// the no-op update_json default, so the host supplies the JSON→Msg seam).
use facett_jobview::{Msg, SortKey};
if let Some(s) = v.get("sort").and_then(|s| s.as_str()) {
let k = match s {
"started" => Some(SortKey::Started),
"kind" => Some(SortKey::Kind),
"target" => Some(SortKey::Target),
"status" => Some(SortKey::Status),
"elapsed" => Some(SortKey::Elapsed),
_ => None,
};
if let Some(k) = k {
self.inner.update(Msg::SetSort(k));
}
return;
}
if let Some(id) = v.get("fold").and_then(|s| s.as_str()) {
self.inner.update(Msg::ToggleFold(id.to_string()));
return;
}
if let Some(sel) = v.get("select") {
match sel {
serde_json::Value::String(id) => {
self.inner.update(Msg::Select(id.clone()));
}
serde_json::Value::Null => {
self.inner.update(Msg::ClearSelection);
}
_ => {}
}
return;
}
if let Some(kf) = v.get("kind_filter") {
match kf {
serde_json::Value::String(k) => {
self.inner.update(Msg::SetKindFilter(Some(k.clone())));
}
serde_json::Value::Null => {
self.inner.update(Msg::SetKindFilter(None));
}
_ => {}
}
}
}
}
}
// ── the seam ──────────────────────────────────────────────────────────────────
/// **The Live Jobs + Agent-fleet pane as a `Box<dyn Panel>`.** `endpoint` is the
/// `dwarves-server` url (fleet); the ledger root is resolved from the environment
/// ([`default_ledger_root`]). The bearer comes from the dwarves client chain
/// ([`crate::ClientCredential`]: `$DWARVES_SERVER_TOKEN`, then `~/.dwarves/token`).
/// The first refresh happens eagerly; empty/offline ⇒ the fixture fallbacks, never a
/// panic.
pub fn jobs_fleet_panel(endpoint: impl Into<String>) -> Box<dyn Panel> {
let token = crate::ClientCredential::resolve().token;
Box::new(JobsFleetPanel::new(
endpoint.into(),
token,
default_ledger_root(),
))
}
#[cfg(test)]
mod tests {
use super::*;
/// Recent epoch micros so a terminal record isn't pruned as "older than 30 days".
fn now_micros() -> i64 {
std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.unwrap()
.as_micros() as i64
}
fn rec(
job_id: &str,
kind: &str,
status: &str,
start: i64,
parent: Option<&str>,
node: Option<&str>,
) -> JobRecord {
JobRecord {
job_id: job_id.into(),
kind: kind.into(),
target: "dwarves".into(),
workspace: "nordisk".into(),
status: status.into(),
// `start` is a small relative offset; anchor it to NOW so terminal rows
// survive the ledger's 30-day terminal-prune.
ts_start_micros: now_micros() + start,
ts_end_micros: None,
elapsed_ms: if status == "running" {
None
} else {
Some(1000)
},
detail_json: String::new(),
result_ref: String::new(),
parent_id: parent.map(String::from),
node: node.map(String::from),
phase: None,
actor: None,
actor_method: None,
}
}
/// The panel must say WHO started a job, and must NOT guess when nobody stamped
/// it. The three rows are chosen so no single wrong implementation passes:
/// an unstamped row (must read `unknown`), a local operator run, and a REMOTE
/// caller using the operator's own name (must still read as remote — the auth
/// method decides, never the name).
#[test]
fn a_job_entry_says_who_initiated_it_and_never_guesses() {
let by = |e: &JobEntry| {
e.meta
.iter()
.find(|(k, _)| k == "by")
.map(|(_, v)| v.clone())
.expect("the entry carries a `by` meta")
};
let anon = entry_from_record(&rec("#1", "release", "done", 0, None, None));
assert_eq!(
by(&anon),
"unknown",
"an unstamped job must not acquire an initiator"
);
assert!(
anon.detail.contains("by unknown"),
"and it is visible in the row: {}",
anon.detail
);
let mut r = rec("#2", "release", "done", 0, None, None);
r.actor = Some("alice".into());
r.actor_method = Some("local".into());
let local = entry_from_record(&r);
assert_eq!(by(&local), "alice (local)");
assert!(
local.detail.contains("by alice (local)"),
"detail: {}",
local.detail
);
let mut r = rec("#3", "release", "done", 0, None, None);
r.actor = Some("alice".into());
r.actor_method = Some("bearer".into());
let remote = entry_from_record(&r);
assert_eq!(
by(&remote),
"alice (bearer)",
"same NAME as the local run — the panel must still show the method, or an \
external caller holding the operator's token reads as the operator"
);
assert_ne!(
by(&remote),
by(&local),
"the two must be distinguishable in the panel"
);
}
/// PURE mapping: a jera JobRecord maps onto a JobEntry SURFACING its node (in both
/// `meta` and the visible `detail`), and the status crosses 1:1. RED-when-broken: a
/// dropped node mapping leaves the placement invisible.
#[test]
fn a_record_maps_to_a_job_entry_surfacing_its_node() {
let r = rec("#7", "cargo-build", "running", 42, Some("#1"), Some("odin"));
let e = entry_from_record(&r);
assert_eq!(e.id, "#7");
assert_eq!(e.status, JobStatus::Running);
assert_eq!(e.parent_id.as_deref(), Some("#1"));
// The node placement is surfaced BOTH ways.
assert!(
e.meta.iter().any(|(k, v)| k == "node" && v == "odin"),
"node in meta"
);
assert!(
e.detail.contains("odin"),
"node prefixed into the visible detail"
);
// A control-plane job (node=None) reads as `local`, never blank.
let local = entry_from_record(&rec("#8", "release", "done", 1, None, None));
assert!(local
.meta
.iter()
.any(|(k, v)| k == "node" && v == LOCAL_NODE));
assert!(local.detail.contains(LOCAL_NODE));
}
/// PURE mapping: an AgentPool reply maps onto the fleet roster with a derived
/// `online` flag (idle/busy = online; offline = not).
#[test]
fn agent_pool_rows_map_to_the_fleet_roster_with_online() {
let rows = vec![
pb::AgentRow {
id: "loki".into(),
group: "linux-x64".into(),
caps: vec!["cargo".into(), "heavy".into()],
cores: 12,
mem_mb: 65536,
state: "idle".into(),
},
pb::AgentRow {
id: "gone".into(),
group: "linux-x64".into(),
caps: vec![],
cores: 0,
mem_mb: 0,
state: "offline".into(),
},
];
let fleet = agents_from_wire(&rows);
assert_eq!(fleet.len(), 2);
assert!(fleet[0].online, "idle ⇒ online");
assert_eq!(
fleet[0].caps,
vec!["cargo".to_string(), "heavy".to_string()]
);
assert!(!fleet[1].online, "offline ⇒ not online");
}
/// **The live-ledger read + RED-when-broken.** Writing JobRecords (with nodes) into
/// a real jera `JobStore` and pointing the pane at it makes the pane render them
/// LIVE from the ledger (`source: "ledger"`), surfacing the node placement — while a
/// root with no ledger renders NOTHING and names why. RED-when-broken: neutralize
/// the ledger read (empty root) and the live-ledger assertions (`source ==
/// "ledger"`, the ledger's `odin` node, the live count) all fail.
///
/// The empty half asserts emptiness deliberately, and that is sound HERE because the
/// test CONSTRUCTS the empty world (a fresh temp root, a port nothing listens on).
/// The trap to avoid is asserting emptiness of the ambient box — green only where
/// nothing has ever run, red on a real forge. Emits a testmatrix cell (reuse edda).
#[test]
fn the_pane_reads_the_jera_ledger_live_and_surfaces_node_placement() {
let dir = tempdir_here();
// Write a real ledger: a build parent + two dispatched steps on loki/odin.
let store = JobStore::open(&dir).unwrap();
store
.submit(&rec("#100", "ci_build", "done", 10_000, None, None))
.unwrap();
store
.submit(&rec(
"#100.1",
"cargo-build",
"done",
10_100,
Some("#100"),
Some("loki"),
))
.unwrap();
store
.submit(&rec(
"#100.2",
"unit",
"running",
10_200,
Some("#100"),
Some("odin"),
))
.unwrap();
// Offline server (nothing listens) ⇒ the fleet is UNREACHABLE and draws nothing,
// but the JOBS come LIVE from the ledger.
let p = JobsFleetPanel::new("http://127.0.0.1:0", "", dir.clone());
let st = Facet::state_json(&p);
let live_ok = st["source"] == "ledger"
&& st["job_count"] == 3
&& st["nodes"].as_array().unwrap().iter().any(|n| n == "odin")
&& st["nodes"].as_array().unwrap().iter().any(|n| n == "loki");
assert!(
live_ok,
"jobs come LIVE from the jera ledger, node placement surfaced: {st}"
);
// The inner shared JobList holds the same 3 rows (one foldable parent).
assert_eq!(st["jobs"]["count"], 3);
assert_eq!(st["jobs"]["parents"], 1, "the build parent is foldable");
assert_eq!(st["jobs"]["running"], 1, "the unit step is in flight");
// THE FLEET IS UNREACHABLE (nothing listens on :0) — and that draws NOTHING.
// This assertion is the whole point of the 2026-09-10 fix: it used to read
// `fleet_source == "fixture"` with `agent_count >= 1`, i.e. three healthy
// machines invented because the RPC failed.
assert_eq!(
st["fleet_source"], "empty",
"an unreachable pool is not a fleet: {st}"
);
assert_eq!(
st["agent_count"], 0,
"no roster is drawn for a pool we could not ask"
);
assert_eq!(
st["online_agents"], 0,
"…and nothing is claimed to be online"
);
assert_eq!(
st["fleet_empty"]["cause"], "unreachable",
"kept apart from no-rows: {st}"
);
assert_eq!(st["fleet_empty"]["feed"], FLEET_FEED);
// None of the deleted roster's machines appears anywhere.
let whole = st.to_string();
for gone in ["loki\",\"group", "nuc"] {
assert!(
!whole.contains(gone),
"the deleted fixture roster's `{gone}` is gone: {whole}"
);
}
// RED-when-broken: a root with NO ledger cannot show the ledger jobs. It is
// `Absent` — the workspace is gone — which is NOT the idle forge that `Empty`
// means, and the pane must keep those apart.
let empty = tempdir_here();
let p2 = JobsFleetPanel::new("http://127.0.0.1:0", "", empty);
let st2 = Facet::state_json(&p2);
assert_eq!(
st2["source"], "empty",
"no ledger ⇒ no jobs (the RED-when-broken twin)"
);
assert_eq!(st2["job_count"], 0, "…and none invented: {st2}");
assert_eq!(
st2["jobs_empty"]["cause"], "not-configured",
"an ABSENT ledger is the gone workspace, not an idle forge: {st2}"
);
let red_when_broken = st2["source"] == "empty" && st2["job_count"] == 0 && live_ok;
facett_core::testmatrix::emit(
"dwarves-embed",
"jobs_fleet_reads_ledger_live",
red_when_broken,
"the Live Jobs pane reads the jera JobRecord ledger (JobStore) and surfaces each job's node placement; a root with no ledger draws ZERO jobs and names the absence (not-configured), and an unreachable AgentPool draws ZERO agents (unreachable) rather than the deleted fixture roster (RED-when-broken)",
);
}
/// ROBOT SCENARIO — the pane driven purely over the `state_json`/`update_json` seam
/// (the channel a `nornir-robotui` scenario + a thin gRPC client use): a live-ledger
/// snapshot + a stub `AgentPool` reply → assert the ledger jobs (count + node) AND
/// the fleet roster, then fold/sort over the seam. RED-when-broken: an empty snapshot
/// yields the fixture, not the ledger's nodes.
#[test]
fn robot_jobs_fleet_pane_drives_over_the_seam() {
let mut p = JobsFleetPanel::new("http://127.0.0.1:0", "", tempdir_here());
// Drive a live snapshot in over the seam: 3 ledger jobs (loki/odin) + a live
// 2-agent fleet from a stub AgentPool reply.
p.apply(FleetSnapshot {
jobs: Ok(LedgerRead::Rows(vec![
rec("#200", "ci_build", "done", 20_000, None, None),
rec(
"#200.1",
"cargo-build",
"done",
20_100,
Some("#200"),
Some("loki"),
),
rec(
"#200.2",
"bench_run",
"running",
20_200,
Some("#200"),
Some("odin"),
),
])),
agents: Ok(vec![
pb::AgentRow {
id: "loki".into(),
group: "linux-x64".into(),
caps: vec!["cargo".into(), "heavy".into()],
cores: 12,
mem_mb: 65536,
state: "busy".into(),
},
pb::AgentRow {
id: "odin".into(),
group: "linux-x64".into(),
caps: vec!["cargo".into(), "heavy".into()],
cores: 32,
mem_mb: 131072,
state: "idle".into(),
},
]),
});
let st = Facet::state_json(&p);
let jobs_ok = st["source"] == "ledger"
&& st["job_count"] == 3
&& st["nodes"].as_array().unwrap().iter().any(|n| n == "odin");
let fleet_ok = st["fleet_source"] == "pool"
&& st["agent_count"] == 2
&& st["online_agents"] == 2
&& st["agents"][0]["id"] == "loki";
assert!(jobs_ok, "ledger jobs + node reflected: {st}");
assert!(fleet_ok, "the live AgentPool roster is reflected: {st}");
// Drive the shared jobs widget over the seam: fold the parent, sort by elapsed.
let before = Facet::state_json(&p)["jobs"]["jobs"]
.as_array()
.unwrap()
.len();
Facet::update_json(&mut p, r##"{"fold":"#200"}"##);
let folded = Facet::state_json(&p)["jobs"]["jobs"]
.as_array()
.unwrap()
.len();
let fold_ok = before == 3 && folded == 1;
Facet::update_json(&mut p, r#"{"sort":"elapsed"}"#);
let sort_ok = Facet::state_json(&p)["jobs"]["sort"] == "Elapsed";
Facet::update_json(&mut p, r#"{"auto_poll":false}"#);
let poll_ok = Facet::state_json(&p)["auto_poll"] == serde_json::json!(false);
// RED-when-broken: an idle ledger + an offline pool cannot surface the ledger
// nodes, and — the 2026-09-10 fix — cannot surface anything ELSE either. Both
// legs go to a NAMED empty, and the two causes stay apart: the ledger answered
// "no jobs" (`no-rows`), the pool never answered at all (`unreachable`).
p.apply(FleetSnapshot {
jobs: Ok(LedgerRead::Empty),
agents: Err(anyhow::anyhow!("offline")),
});
let st2 = Facet::state_json(&p);
let red = st2["source"] == "empty"
&& st2["job_count"] == 0
&& st2["jobs_empty"]["cause"] == "no-rows"
&& st2["fleet_source"] == "empty"
&& st2["agent_count"] == 0
&& st2["fleet_empty"]["cause"] == "unreachable";
let drivable = jobs_ok && fleet_ok && fold_ok && sort_ok && poll_ok && red;
assert!(
drivable,
"the Jobs+fleet pane drives live-ledger + fleet + fold/sort over the seam"
);
facett_core::testmatrix::emit(
"dwarves-embed",
"jobs_fleet_pane_drivable",
drivable,
"the Live Jobs + Agent-fleet pane drives over state_json/update_json: live ledger jobs (count + node) + the AgentPool roster + fold/sort; an idle ledger and an offline pool each draw ZERO rows behind their own NAMED empty state, no-rows and unreachable kept apart (RED-when-broken)",
);
}
/// **BOTH FIXTURES ARE GONE** (2026-09-10, the ninth pane). Pins the four no-data
/// states, and above all that a FAILED `AgentPool` no longer draws a healthy fleet:
///
/// (a) `Err` on the fleet ⇒ zero agents, zero online, cause `unreachable`;
/// (b) `Ok([])` ⇒ zero agents but cause `no-rows` — the pool ANSWERED, and that is
/// a different fact from never reaching it. The two must not collapse;
/// (c) `Absent` ledger ⇒ `not-configured`, `Empty` ledger ⇒ `no-rows` — the
/// three-way `LedgerRead` survives all the way into the render;
/// (d) none of either fixture's invented ids appears anywhere in the state.
///
/// RED-when-broken: restore either fallback and (a)/(d) fail.
#[test]
fn no_feed_names_its_refusal_and_never_draws_a_fixture() {
let mut p = JobsFleetPanel::new("http://127.0.0.1:0", "", tempdir_here());
// (a) the headline: the RPC failed, so there is NO fleet to draw.
p.apply(FleetSnapshot {
jobs: Ok(LedgerRead::Absent),
agents: Err(anyhow::anyhow!("connect to dwarves-server: refused")),
});
let st = Facet::state_json(&p);
assert_eq!(st["fleet_source"], "empty");
assert_eq!(
st["agent_count"], 0,
"a failed AgentPool draws no machines: {st}"
);
assert_eq!(
st["online_agents"], 0,
"…and claims nothing is online: {st}"
);
assert_eq!(st["fleet_empty"]["cause"], "unreachable");
assert_eq!(st["fleet_empty"]["feed"], FLEET_FEED);
assert!(st["fleet_empty"]["detail"]
.as_str()
.unwrap()
.contains("refused"));
// (c) the ABSENT ledger is the gone workspace, not the idle forge.
assert_eq!(st["source"], "empty");
assert_eq!(st["job_count"], 0);
assert_eq!(st["jobs_empty"]["cause"], "not-configured");
assert_eq!(st["jobs_empty"]["feed"], JOBS_FEED);
// (d) neither fixture's invented rows survive.
let whole = st.to_string();
for gone in ["#100", "#200", "ci_build", "nuc", "arm64"] {
assert!(
!whole.contains(gone),
"a deleted fixture's `{gone}` is gone: {whole}"
);
}
// (b) a pool that ANSWERED with no agents is `no-rows`, NOT `unreachable`.
p.apply(FleetSnapshot {
jobs: Ok(LedgerRead::Empty),
agents: Ok(vec![]),
});
let st = Facet::state_json(&p);
assert_eq!(
st["fleet_source"], "pool",
"answering IS being live, even with 0 agents"
);
assert_eq!(st["agent_count"], 0);
assert_eq!(
st["fleet_empty"]["cause"], "no-rows",
"an empty pool is not an unreachable one: {st}"
);
// …and the idle ledger is `no-rows` too, distinct from the Absent above.
assert_eq!(
st["jobs_empty"]["cause"], "no-rows",
"an idle forge is not a gone one: {st}"
);
// Rows on both legs ⇒ NO empty block on either (never neither, never both).
p.apply(FleetSnapshot {
jobs: Ok(LedgerRead::Rows(vec![rec(
"#1",
"ci_build",
"done",
1,
None,
Some("loki"),
)])),
agents: Ok(vec![pb::AgentRow {
id: "loki".into(),
group: "linux-x64".into(),
caps: vec![],
cores: 1,
mem_mb: 1,
state: "idle".into(),
}]),
});
let st = Facet::state_json(&p);
assert_eq!(st["source"], "ledger");
assert!(
st["jobs_empty"].is_null(),
"rows and an empty block together: {st}"
);
assert_eq!(st["fleet_source"], "pool");
assert!(
st["fleet_empty"].is_null(),
"roster and an empty block together: {st}"
);
}
/// **THE OTHER WORLD.** Every test above constructs its own emptiness, which can
/// only ever prove the pane is honest on a box where nothing runs. The rule is
/// *rows or a named empty, in BOTH worlds* — and the second world needs a real
/// forge, so this one drives the pane against a live `dwarves-server`.
///
/// It asserts the DISJUNCTION, never emptiness: a forge with agents must show them
/// with no empty block, a forge without must show none with one. Asserting either
/// side alone would be a fabricated verdict — green only where the world happens to
/// match, red where it does not.
///
/// Run: `DWARVES_TEST_ENDPOINT=http://oden:7900 cargo test -p dwarves-embed
/// --offline -- --ignored --nocapture`
#[test]
#[ignore = "requires: a reachable dwarves-server; set DWARVES_TEST_ENDPOINT (e.g. http://oden:7900), optionally DWARVES_TEST_TOKEN"]
fn against_a_live_forge_the_pane_is_honest_too() {
let Ok(endpoint) = std::env::var("DWARVES_TEST_ENDPOINT") else {
panic!("set DWARVES_TEST_ENDPOINT to a reachable dwarves-server");
};
let token = std::env::var("DWARVES_TEST_TOKEN").unwrap_or_default();
let p = JobsFleetPanel::new(&endpoint, &token, tempdir_here());
let st = Facet::state_json(&p);
eprintln!(
"live forge {endpoint} → {}",
serde_json::to_string_pretty(&st).unwrap()
);
// The fleet leg, both ways round. `source` is never `"fixture"` — that value no
// longer exists — and the roster and the empty block are mutually exclusive.
let agents = st["agent_count"].as_u64().expect("agent_count");
let named = st["fleet_empty"]["empty"] == serde_json::Value::Bool(true)
&& !st["fleet_empty"]["feed"].as_str().unwrap_or("").is_empty();
if agents > 0 {
assert_eq!(
st["fleet_source"], "pool",
"agents shown ⇒ they came from the pool: {st}"
);
assert!(
st["fleet_empty"].is_null(),
"a roster AND an empty block: {st}"
);
assert!(
st["online_agents"].as_u64().unwrap() <= agents,
"more online than exist: {st}"
);
} else {
assert!(
named,
"no agents ⇒ the pane must NAME why (feed + cause): {st}"
);
let cause = st["fleet_empty"]["cause"].as_str().unwrap_or("");
assert!(
cause == "unreachable" || cause == "no-rows",
"an unreachable pool and an empty one must stay apart, got {cause:?}: {st}"
);
}
// The jobs leg is pointed at a fresh temp root on purpose, so it is the ABSENT
// case even against a live server — the gone workspace, never an idle forge.
assert_eq!(st["source"], "empty");
assert_eq!(st["jobs_empty"]["cause"], "not-configured", "{st}");
}
#[test]
fn poll_due_respects_interval_and_toggle() {
let p = JobsFleetPanel::new("http://127.0.0.1:0", "", tempdir_here());
assert!(p.auto_poll);
assert!(!p.poll_due(0.5), "not due before one interval");
assert!(
p.poll_due(DEFAULT_JOBS_POLL_INTERVAL_S),
"due at one interval"
);
let mut p = p;
p.set_auto_poll(false);
assert!(!p.poll_due(99.0), "never due while paused");
}
/// A unique empty temp dir under the OS temp root (no external crate needed).
fn tempdir_here() -> PathBuf {
use std::sync::atomic::{AtomicU64, Ordering};
static SEQ: AtomicU64 = AtomicU64::new(0);
let n = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.unwrap()
.as_nanos();
let seq = SEQ.fetch_add(1, Ordering::Relaxed);
let mut d = std::env::temp_dir();
d.push(format!(
"dwarves-embed-jobsfleet-test-{}-{n}-{seq}",
std::process::id()
));
std::fs::create_dir_all(&d).unwrap();
d
}
}
#[cfg(test)]
mod ledger_lifecycle_tests {
use super::*;
/// **RED-when-broken: nuke the workspace, and its jobs are nuked with it — and the
/// reader can TELL.**
///
/// The owner, 2026-08-25. A jera job is owned by its workspace: when the workspace
/// goes, `nornir::registry::remove` unlinks `<root>/<name>/builds/`, which takes
/// `jobs.redb` with it. So "the ledger file is gone" is the NORMAL end state of a
/// removed workspace, and the reader that cannot distinguish it from "the forge is
/// idle" turns a deletion into a UI that looks fine.
///
/// Asserted against the **ledger on disk**, never a counter: a real `JobStore`, a
/// real row written through the real sink, and a real `remove_file` standing in for
/// the purge. The three legs are three DIFFERENT return values, and the middle one
/// is what the old `Ok(Vec::new())` could not say.
#[test]
fn a_removed_ledger_reads_as_absent_and_an_idle_one_reads_as_empty() {
let dir = tempfile::tempdir().unwrap();
let root = dir.path();
// (1) NOTHING THERE — never populated, or removed. Not an empty read.
assert_eq!(
read_ledger_jobs(root, 50, "").unwrap(),
LedgerRead::Absent,
"a root with no jobs.redb must read as ABSENT"
);
assert!(
!root.join("jobs.redb").exists(),
"…and the file really is not there"
);
// (2) A REAL, EMPTY ledger — a live forge that has not run anything. The file
// exists; `open` created it.
let store = jera::jobs::JobStore::open(root).expect("open jera ledger");
assert!(
root.join("jobs.redb").exists(),
"opening the store creates the ledger"
);
assert_eq!(
read_ledger_jobs(root, 50, "").unwrap(),
LedgerRead::Empty,
"a ledger that exists and holds no rows is EMPTY, not ABSENT — an idle forge \
is not a deleted workspace"
);
// (3) A REAL ROW, written through the real sink, for a named workspace.
let job = store.start(
"lens",
"change_risk @nordisk",
"nordisk",
serde_json::json!({ "lens": "change_risk" }),
);
job.finish(serde_json::json!({ "rows": 1 }), "");
drop(store);
let read = read_ledger_jobs(root, 50, "").unwrap();
let rows = match &read {
LedgerRead::Rows(r) => r.clone(),
other => panic!("a ledger with a row must read as Rows, got {other:?}"),
};
// Assert the ROW's own content — the applied output, not "len() == 1".
assert_eq!(rows.len(), 1, "one row was written, one row comes back");
assert_eq!(
rows[0].workspace, "nordisk",
"the row is stamped with its workspace"
);
assert_eq!(rows[0].kind, "lens");
assert_eq!(rows[0].target, "change_risk @nordisk");
// (4) THE PURGE. `Registry::remove(name, purge=true)` unlinks the whole
// `builds/` tree; here the file itself stands in for it.
std::fs::remove_file(root.join("jobs.redb")).unwrap();
let after = read_ledger_jobs(root, 50, "").unwrap();
assert_eq!(
after,
LedgerRead::Absent,
"a ledger removed with its workspace must read as ABSENT — NOT as Empty and \
NOT as an empty Rows. This is the assertion the old `Ok(Vec::new())` made \
impossible: it returned the same value here as it did for an idle forge."
);
assert!(after.rows().is_empty(), "and it carries no rows");
assert!(after.is_absent());
assert_eq!(after.key(), "absent");
// The three states really are three different values — the property the whole
// change exists for. Without this line a future `impl PartialEq` that collapsed
// Absent and Empty would pass every assertion above.
assert_ne!(LedgerRead::Absent, LedgerRead::Empty);
assert_ne!(LedgerRead::Empty, LedgerRead::Rows(rows));
}
/// **RED-when-broken: the FILE road must SELECT the class, not truncate and hope.**
///
/// The shape of the live defect (`jerakind-1`), reproduced on a real `jobs.redb`: one
/// rare class buried under a crowd of a common one. `workspace_populate` measured
/// **0 of 200** rows on the forge — one populate job in the whole ledger, and the
/// server's window closed before anything asked for it.
///
/// The fixture is deliberately harsher than the census: 60 `lens` rows written AFTER
/// the single `workspace_populate` row, so the populate is the OLDEST row in the
/// ledger and `list`'s newest-first order puts it dead last. A read that truncates
/// before it selects cannot return it at ANY limit below 61 — and the pane's real ask
/// is 100 with a 20x widen, which is why the old client-side `retain` appeared to work
/// on small ledgers and failed on the real one.
///
/// Asserted on the returned ROW (`target`, `workspace`, `phase`), never on a count:
/// `len() == 1` is also what a broken read that returned the wrong single row gives.
#[test]
fn a_kind_scoped_file_read_finds_a_row_the_window_would_have_truncated() {
let dir = tempfile::tempdir().unwrap();
let root = dir.path();
let store = jera::jobs::JobStore::open(root).expect("open jera ledger");
// The needle FIRST, so newest-first sorting buries it under everything after it.
let _populate = store.start(
jera::jobs::kind::WORKSPACE_POPULATE,
"nordisk",
"nordisk",
serde_json::json!({ "summary": "34 member(s)" }),
);
// 60 rows of the crowding class, all newer.
for i in 0..60 {
let j = store.start(
"lens",
&format!("change_risk #{i}"),
"nordisk",
serde_json::json!({}),
);
j.finish(serde_json::json!({ "rows": 1 }), "");
}
drop(store);
// The RED anchor, measured rather than asserted-around: at the limit the pane
// actually asks for, an UNSCOPED read does not contain the populate row at all.
let unscoped = read_ledger_jobs(root, 5, "").unwrap();
let unscoped_rows = unscoped.rows();
assert_eq!(
unscoped_rows.len(),
5,
"the unscoped read is capped at the limit"
);
assert!(
!unscoped_rows
.iter()
.any(|r| r.kind == jera::jobs::kind::WORKSPACE_POPULATE),
"PREMISE: the unscoped window must NOT contain the populate row, or this test \
proves nothing about scoping. Got: {:?}",
unscoped_rows
.iter()
.map(|r| r.kind.as_str())
.collect::<Vec<_>>()
);
// The SCOPED read at the SAME limit must find it — the select happened first.
let scoped = read_ledger_jobs(root, 5, jera::jobs::kind::WORKSPACE_POPULATE).unwrap();
let rows = match &scoped {
LedgerRead::Rows(r) => r.clone(),
other => panic!("a ledger holding a populate row must read as Rows, got {other:?}"),
};
assert_eq!(
rows.len(),
1,
"exactly the one populate row — not the 60 lens rows, not zero. A read that \
truncated to 5 before selecting returns 0 here."
);
assert_eq!(rows[0].kind, jera::jobs::kind::WORKSPACE_POPULATE);
assert_eq!(
rows[0].target, "nordisk",
"the ROW that came back is the right one"
);
assert_eq!(rows[0].workspace, "nordisk");
// A class with NO rows in a ledger that HAS rows is `Rows(vec![])` — "0 jobs of
// this class beside a live source" — and NEVER `Empty`, which renders as "the
// forge is idle". Collapsing those is the `Absent`/`Empty` defect one axis over.
let none = read_ledger_jobs(root, 5, jera::jobs::kind::BENCH_RUN).unwrap();
assert_eq!(
none,
LedgerRead::Rows(Vec::new()),
"a kind holding no rows in a NON-empty ledger must be Rows(empty), not Empty"
);
assert_ne!(
none,
LedgerRead::Empty,
"…and the two really are different values"
);
assert_eq!(none.key(), "rows");
}
/// **RED-when-broken: the phase reaches the rendered row.**
///
/// `phase` is the one field a running job changes while an operator watches it, it is
/// written durably (jera's `upsert` grew a preserve limb so `finish()` cannot erase
/// it), and BOTH host mappers dropped it — so every one of those fixes landed behind
/// a mapper that threw the value away. Asserted through
/// [`facett_jobview::JobEntry::phase`], the read side the pane uses, not by grepping
/// `meta` for a string literal: a test that reads the key by hand passes even if the
/// writer and the renderer disagree about its spelling.
///
/// The phase is emitted through the REAL producer — `start_daemon` → `h.reporter()` →
/// `Reporter::phase`, the daemon-wired [`jera::jobs::JobStore::daemon_report_sink`]
/// path that is the only thing in the constellation that persists a phase — and read
/// back out of the file by [`read_ledger_jobs`], not injected into a hand-built
/// record. The writer store keeps the redb lock for the whole test, so the read is
/// the out-of-process `open_read_only` snapshot a viewer really gets, whose in-memory
/// report registry is empty and cannot answer for the row.
#[test]
fn the_mapper_carries_the_phase_a_running_job_reports() {
let dir = tempfile::tempdir().unwrap();
let root = dir.path();
let store = jera::jobs::JobStore::open(root).expect("open jera ledger");
let h = store.start_daemon("osm-tiles", "korp", serde_json::json!({ "enabled": true }));
let reporter = h.reporter();
reporter.progress(30, Some(100)); // in-memory only — no redb write of its own
reporter.phase("cutting", true); // THIS is what persists onto the row
let read = read_ledger_jobs(root, 10, "").unwrap();
let recs = read.rows();
assert_eq!(recs.len(), 1, "one row on disk");
let dp = recs[0].daemon_phase().unwrap_or_else(|| {
panic!(
"PREMISE: the phase must be on DISK. If this fails the defect is in \
jera's writer, not in this mapper. Row: {:?}",
recs[0]
)
});
assert_eq!(dp.name, "cutting");
assert_eq!(
dp.fraction(),
Some(0.30),
"with the progress snapshot taken at phase() time"
);
let entries = entries_from_records(recs);
assert_eq!(
entries[0].phase(),
Some("cutting"),
"the mapped row must SURFACE the phase the ledger holds — this is the \
assertion that fails when `entry_from_record` drops the field, which is \
exactly what it did while the pane rendered a bare `running`"
);
assert_eq!(
entries[0].status,
JobStatus::Running,
"…and it is still running"
);
assert_eq!(
entries[0].progress_meter(),
facett_jobview::ProgressMeter::Determinate(0.30),
"the durable progress mirror rides along — a determinate meter, not a sweep"
);
// APPLIED OUTPUT: the pixels the paint path fills, not the enum it fills them
// from. On a 200 px track a working 30 % mirror is 60 px; a dropped one is 0.
// Tolerance, not equality: the fraction round-trips through `with_progress`'s
// `{:.4}` meta encoding and comes back 60.000004. A sub-pixel epsilon still
// separates 60 from the 0.0 a dropped mirror gives — the distinction the
// assertion is for.
let fill = facett_jobview::meter_fill_width(entries[0].progress_meter(), 200.0);
assert!(
(fill - 60.0).abs() < 0.01,
"the bar must be DRAWN to 30 % of its 200 px track; got {fill} px (0.0 means \
the durable progress mirror was dropped)"
);
// A record with NO phase must not stamp a blank key: an empty cell reads as a
// phase, and "no phase reported" is a different fact. The identity-value trap.
let mut bare = recs[0].clone();
bare.phase = None;
assert_eq!(
entry_from_record(&bare).phase(),
None,
"a row with no phase must surface as NO phase, not as a blank one"
);
assert!(
!entry_from_record(&bare)
.meta
.iter()
.any(|(k, _)| k == facett_jobview::PHASE_META_KEY),
"…and the key is absent from meta entirely, not present-and-empty"
);
// The whitespace rule lives in the ONE writer, so assert it there rather than
// faking a whitespace phase jera would never persist.
assert_eq!(
JobEntry::new("x", "k", JobStatus::Running)
.with_phase(Some(" "))
.phase(),
None,
"the shared writer refuses a whitespace-only phase"
);
h.stop(serde_json::Value::Null);
}
}