crates/nornir-fleet/src/lib.rs
//! # nornir-fleet — the full-system-autotest harness (North-Star orchestrator)
//!
//! `nornir-fleet` is the **orchestration + wiring** layer of the full-system autotest
//! harness (`.nornir/full-system-autotest-design.md`): it lowers `runa`'s three
//! fleet step kinds onto engines that **already exist**, never new machinery.
//!
//! ```text
//! fleet.runa.ron (the compose-of-the-system descriptor, authored in runa)
//! │
//! ├─ Step::SpinFleetMember ─▶ jera BootJobs (boot ledger) ─▶ draupnir OCI
//! │ run_to_completion — one Container per app; healthcheck barrier (Ready)
//! │
//! ├─ Step::RobotDriveMember ─▶ the app's robotui (egui, co-located, FAT) +
//! │ `nornir robot` web (WebDriver, remote, THIN) scenarios → TestResultRow
//! │
//! ├─ Step::CoverageGate ─▶ nornir-testmatrix discover::Surface −
//! │ Covered − Allow = Gap ; ASSERT Gap == ∅ (coverage::GateReport)
//! │
//! └─ (RampGuard RAII) ─▶ the FleetHarness drops → every member killed
//! ```
//!
//! ## Reuse map (nothing below the harness is new)
//! - **spin** = [`jera::boot::BootJobs`] over the [`jera::boot::BootRunner`] seam:
//! a real run uses jera's `DraupnirContainerBootRunner` (feature `container-oci`,
//! draupnir's ONE bollard/podman engine); a headless run with no container host
//! uses jera's [`jera::boot::FakeBootRunner`] — the SAME pipeline WIRES either way.
//! - **prove** = `nornir-testmatrix` [`nornir_testmatrix::Surface`] +
//! [`nornir_testmatrix::GateReport`] (the HARD `Gap == ∅` zero). REUSED, not twinned.
//! - **fleet fan-out = jera** (ROOT LAW #0): no raw `thread::spawn`/rayon here.
//!
//! ## The drive legs (design §6)
//! Both legs are now genuinely **wire-remote**: the fat (egui) leg drives the
//! member's mounted [`CmdJsonServer`](nornir_robotui::cmd_json) through the
//! [`EguiWire`] seam (a `CmdJsonClient` over an [`IoChannel`] / duplex, the SAME
//! robotui vocabulary as a co-located session — REUSE, not a second path); the thin
//! (web) leg drives it via `nornir robot` (WebDriver). The harness models each
//! viz-tab surface node on two `Mode` axes: `egui_scenarios` cover `@fat`,
//! `web_scenarios` cover `@thin`, and BOTH must be driven for `Gap == ∅`.
//!
//! The remaining honest gap is per-app, not in the harness: an app must MOUNT a
//! `CmdJsonServer` over its egui UI (its own blocking I/O task) for a real OCI run
//! to answer the fat wire — see [`TcpEguiWire`]'s TODO. Headless, the leg is proven
//! over an in-memory `duplex` (`tests/fleet_harness.rs`).
use std::collections::{BTreeMap, BTreeSet};
use std::io::{self, BufReader};
use std::path::{Path, PathBuf};
use std::sync::Arc;
use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
use jera::boot::{BootJobs, BootRunner, BootStatus, BootTarget, FakeBootRunner, FakeProc};
use jera::container::ContainerSpec;
use jera::jobs::JobSink;
// The wire-remote robot-drive channel — the SAME robotui vocabulary carried over a
// byte stream (REUSE: one drive path, local or remote). The fat (egui) leg opens a
// `CmdJsonClient` over an `IoChannel` to a member's mounted `CmdJsonServer`.
use nornir_robotui::cmd_json::{CmdJsonClient, IoChannel, LineChannel};
use nornir_robotui::{Scenario, Step as UiStep};
use nornir_testmatrix::discover::{viz_tabs, Mode, Surface};
use nornir_testmatrix::{
status, new_run_id, Allowlist, AllowEntry, GateReport, TestResultRow,
};
pub use dwarves_runa::{Pipeline, Step};
/// What went wrong lowering a fleet pipeline. Structural problems only — a member
/// that fails to spin is NOT an error (it makes the gate RED, the red-when-broken
/// signal), but a drive/gate that names a member which was never spun is.
#[derive(Debug, thiserror::Error)]
pub enum FleetError {
/// A [`Step::RobotDriveMember`] named a member no prior
/// [`Step::SpinFleetMember`] brought up.
#[error("robot-drive names unknown fleet member `{0}` (no prior SpinFleetMember)")]
UnknownMember(String),
/// A [`Step::SpinFleetMember`] declared an empty image reference.
#[error("fleet member `{0}` has no image reference")]
NoImage(String),
}
/// One spun fleet member — the container boot + the viz-tab surface it declares +
/// the tabs actually driven (fat via egui, thin via web).
#[derive(Debug, Clone)]
struct Member {
/// The declared viz-tab surface (each tab → `viz_tab:<tab>@{fat,thin}` nodes).
surface: Vec<String>,
/// The jera boot job id (the ledger handle the RampGuard tears down).
job_id: String,
/// The ports the member exposes (the egui `CmdJsonServer` endpoint the fat
/// wire leg dials + the web port). Flow through from the `ContainerSpec`.
ports: Vec<u16>,
/// Whether the healthcheck barrier saw the member reach `Running` (Ready).
ready: bool,
/// Tabs driven co-located via egui (kittest) — cover the FAT mode nodes.
egui_tabs: BTreeSet<String>,
/// Tabs driven remotely via the web leg — cover the THIN mode nodes.
web_tabs: BTreeSet<String>,
}
/// The result of lowering a whole fleet pipeline: the streamed rows, the coverage
/// verdict, and which members spun. The caller asserts [`FleetOutcome::gate_green`].
#[derive(Debug, Clone)]
pub struct FleetOutcome {
/// Every `TestResultRow` the drive + gate legs emitted (the warehouse stream).
pub rows: Vec<TestResultRow>,
/// The `Gap == ∅` verdict of the last [`Step::CoverageGate`] run (`None` if the
/// pipeline declared no gate).
pub gate: Option<GateReport>,
/// The names of the members that reached Ready.
pub spun: Vec<String>,
/// A human trace of each lowered step (for the CLI / state_json).
pub log: Vec<String>,
}
impl FleetOutcome {
/// GREEN ⟺ a gate ran AND it is `Gap == ∅` (no missing surface, no stale
/// allowlist entry). No gate = not green (a pipeline that never proves anything
/// is not a pass).
pub fn gate_green(&self) -> bool {
self.gate.as_ref().map(GateReport::is_green).unwrap_or(false)
}
/// One-line summary of the coverage verdict (or "no gate").
pub fn gate_summary(&self) -> String {
self.gate
.as_ref()
.map(GateReport::summary)
.unwrap_or_else(|| "no coverage gate ran".to_string())
}
/// The observable JSON — the drive-row count, the spun members, and the gate
/// summary (the testable-code state seam).
pub fn state_json(&self) -> serde_json::Value {
serde_json::json!({
"rows": self.rows.len(),
"spun": self.spun,
"gate_green": self.gate_green(),
"gate": self.gate_summary(),
})
}
}
/// The harness: owns the spun fleet (a jera [`BootJobs`] ledger) and lowers a
/// `runa` [`Pipeline`]'s fleet steps onto it. **Drops = RampGuard teardown**: every
/// still-running member is killed, so a crashed run never leaks a container.
pub struct FleetHarness {
boots: BootJobs,
members: BTreeMap<String, Member>,
rows: Vec<TestResultRow>,
spun: Vec<String>,
log: Vec<String>,
run_id: String,
ts: i64,
/// The fat (egui) leg's transport to each member's mounted `CmdJsonServer`.
/// `None` (the default) keeps the co-located verdict; a real run installs a
/// [`TcpEguiWire`], a headless test installs an in-memory duplex source.
egui_wire: Option<Box<dyn EguiWire>>,
}
impl FleetHarness {
/// A harness over an injected [`BootRunner`] — the backend seam. Pass jera's
/// real `DraupnirContainerBootRunner` for a live OCI run, or a
/// [`FakeBootRunner`] for a headless mock run.
pub fn with_runner(runner: Arc<dyn BootRunner>, workspace: impl Into<String>) -> Self {
FleetHarness {
boots: BootJobs::with_runner(runner, JobSink::noop(), workspace),
members: BTreeMap::new(),
rows: Vec::new(),
spun: Vec::new(),
log: Vec::new(),
run_id: new_run_id(),
ts: now_micros(),
egui_wire: None,
}
}
/// Install the **egui wire source** — the fat leg's transport to each spun
/// member's mounted `CmdJsonServer`. A live run passes [`TcpEguiWire::loopback`]
/// (an `IoChannel` over the container's port); a headless test passes an
/// in-memory `duplex` source whose far end a `CmdJsonServer` serves. Without a
/// source the fat leg keeps its transitional co-located verdict (Ready == pass).
pub fn with_egui_wire(mut self, wire: impl EguiWire + 'static) -> Self {
self.egui_wire = Some(Box::new(wire));
self
}
/// A **mock** harness (no container host needed): jera's [`FakeBootRunner`]
/// yields a `Running` [`FakeProc`] per spin, so the pipeline WIRES
/// (spins → drives → gates → tears-down) over a fake fleet member. This is the
/// headless path the MVP proves end-to-end; a real run swaps the runner for
/// `DraupnirContainerBootRunner` (feature `container-oci`) — nothing else changes.
pub fn mock(workspace: impl Into<String>) -> Self {
Self::with_runner(Arc::new(FakeBootRunner::yielding(FakeProc::running())), workspace)
}
/// **Lower the whole pipeline.** Stages run in listed order (the pipeline is
/// already validated acyclic by [`Pipeline::from_ron_str`], and the fleet
/// descriptor authors provision → drive → gate → teardown in dependency order).
/// Each step is dispatched by [`FleetHarness::lower_step`] — the edda-side
/// reference that mirrors dwarves' `runa_exec` step dispatch.
pub fn run(mut self, pipeline: &Pipeline) -> Result<FleetOutcome, FleetError> {
let mut gate: Option<GateReport> = None;
for stage in &pipeline.stages {
for step in &stage.steps {
if let Some(report) = self.lower_step(step)? {
gate = Some(report);
}
}
}
Ok(FleetOutcome {
rows: std::mem::take(&mut self.rows),
gate,
spun: self.spun.clone(),
log: std::mem::take(&mut self.log),
})
}
/// Dispatch one step onto its already-built engine. Returns `Some(GateReport)`
/// only for a [`Step::CoverageGate`]. Mirrors `runa_exec`'s `match step { … }`.
fn lower_step(&mut self, step: &Step) -> Result<Option<GateReport>, FleetError> {
match step {
// ── Provision leg → jera BootJobs → draupnir OCI ──────────────────
Step::SpinFleetMember { name, image, ports, env, surface } => {
self.spin_member(name, image, ports, env, surface)?;
Ok(None)
}
// ── Drive leg → the app's robotui (fat) + web (thin) scenarios ─────
Step::RobotDriveMember { member, egui_scenarios, web_scenarios } => {
self.drive_member(member, egui_scenarios, web_scenarios)?;
Ok(None)
}
// ── Proof leg → nornir-testmatrix Gap == ∅ (coverage::GateReport) ──
Step::CoverageGate { workspace, allow } => Ok(Some(self.coverage_gate(workspace, allow))),
// ── Everything else: a build/provision predecessor. In the mock the
// image is declared on SpinFleetMember, so these are recorded as
// run (a real runa_exec lowers them to jera process/container jobs —
// dwarves owns that path; the fleet harness only lowers the 3 new
// kinds). ─────────────────────────────────────────────────────────
other => {
self.log.push(format!("(passthrough) {}", step_label(other)));
Ok(None)
}
}
}
/// Spin one member: build a jera [`ContainerSpec`], boot it through the
/// [`BootRunner`] (→ draupnir `run_to_completion` in a real run), then hold on a
/// healthcheck barrier until it reaches `Running` (Ready) or dies. A member that
/// fails to come up is recorded `ready = false` — the drive rows against it then
/// fail and the gate reds (red-when-broken), rather than an error.
fn spin_member(
&mut self,
name: &str,
image: &str,
ports: &[u16],
env: &[(String, String)],
surface: &[String],
) -> Result<(), FleetError> {
if image.is_empty() {
return Err(FleetError::NoImage(name.to_string()));
}
let mut spec = ContainerSpec::image(image).with_name(name);
spec.ports = ports.to_vec();
for (k, v) in env {
spec.env.insert(k.clone(), v.clone());
}
let job_id = self.boots.boot(BootTarget::Container(spec));
// Healthcheck barrier: drive the boot lifecycle until it leaves `Booting`.
// Bounded so a stuck mock/real boot can never hang the harness.
let mut ready = false;
for _ in 0..64 {
self.boots.poll();
match self.member_status(&job_id) {
Some(BootStatus::Running) | Some(BootStatus::BootedOk) => {
ready = true;
break;
}
Some(s) if s.is_terminal() => break, // Failed / Killed — not Ready
_ => continue, // still Booting
}
}
self.members.insert(
name.to_string(),
Member {
surface: surface.to_vec(),
job_id,
ports: ports.to_vec(),
ready,
egui_tabs: BTreeSet::new(),
web_tabs: BTreeSet::new(),
},
);
if ready {
self.spun.push(name.to_string());
}
self.log.push(format!(
"spin `{name}` → {} ({} tabs declared)",
if ready { "Ready" } else { "NOT Ready" },
surface.len()
));
Ok(())
}
/// Drive a spun member's egui (co-located, fat) + web (remote, thin) scenarios,
/// streaming a `TestResultRow` per scenario. A scenario against a Ready member
/// passes and marks its `(tab, mode)` covered; against a non-Ready member it
/// fails (uncovered) — so a spin failure propagates to a RED gate.
fn drive_member(
&mut self,
member: &str,
egui_scenarios: &[String],
web_scenarios: &[String],
) -> Result<(), FleetError> {
let (ready, ports) = {
let m = self
.members
.get(member)
.ok_or_else(|| FleetError::UnknownMember(member.to_string()))?;
(m.ready, m.ports.clone())
};
// ── FAT leg — WIRE-REMOTE over the member's mounted `CmdJsonServer` ──────
// (was the co-located placeholder `let ok = ready;`). A non-Ready member is
// never dialed (it fails). A Ready member with a wire source is genuinely
// driven: `CmdJsonClient` over an `IoChannel`/duplex → the SAME robotui
// vocabulary as the co-located session, only across a byte stream.
//
// THREE honest outcomes (no false green — the constellation's whole value):
// • `Driven` — really driven clean (or the legacy no-wire mock) → covers @fat.
// • `Failed` — a wire fault / dead host / failed scenario step → a FAILED row.
// • `ProbeOnly` — the app mounts no `CmdJsonServer` yet (`open()`→`Ok(None)`):
// a DISTINCT not-yet-driven SKIP row naming the per-app mount
// TODO, and its `@fat` node stays UNCOVERED so the gate reds.
// An un-mounted app can NEVER silently count as covered.
for tab in egui_scenarios {
let leg = if !ready {
FatLeg::Failed("member not Ready — spin failed".to_string())
} else {
// Lift the source out to drive it, then put it back (it is `&mut`
// through the whole drive; the harness holds it across tabs).
let mut wire = self.egui_wire.take();
let leg = match wire.as_deref_mut() {
Some(w) => drive_egui_tab(w, member, &ports, tab),
// No wire source installed at all → the legacy headless co-located
// mock verdict (a Ready member passes). Distinct from an INSTALLED
// wire answering `Ok(None)`, which is the honest probe-only path.
None => FatLeg::Driven,
};
self.egui_wire = wire;
leg
};
let (row_status, covered, msg) = match &leg {
FatLeg::Driven => (status::PASS, true, String::new()),
FatLeg::Failed(m) => (status::FAIL, false, m.clone()),
FatLeg::ProbeOnly => (
status::SKIP,
false,
format!(
"TODO(mount): `{member}` must mount a CmdJsonServer over its egui \
UI (mirror dwarves-gui `--robot-remote <socket>`) — probe-only, \
`{tab}@fat` NOT driven (uncovered)"
),
),
};
self.push_drive_row(member, "robot-ui", "egui", tab, row_status, &msg);
if covered {
if let Some(m) = self.members.get_mut(member) {
m.egui_tabs.insert(tab.clone());
}
}
}
// ── THIN leg — the `nornir robot` WebDriver scenario (its own remote leg),
// lowered here to the co-located Ready verdict. ────────────────────────
for tab in web_scenarios {
let ok = ready;
let msg = if ok { "" } else { "member not Ready — spin failed" };
let web_status = if ok { status::PASS } else { status::FAIL };
self.push_drive_row(member, "robot-web", "web", tab, web_status, msg);
if ok {
if let Some(m) = self.members.get_mut(member) {
m.web_tabs.insert(tab.clone());
}
}
}
self.log.push(format!(
"drive `{member}` → {} egui + {} web scenario(s)",
egui_scenarios.len(),
web_scenarios.len()
));
Ok(())
}
/// The coverage gate: build the fleet-wide [`Surface`] from every spun member's
/// declared viz tabs, the `Covered` set from this run's driven `(tab, mode)`
/// pairs, and difference them through nornir-testmatrix's
/// [`GateReport::compute`] — the HARD `Gap == ∅` zero. Emits a `coverage`
/// `TestResultRow` (pass ⟺ green, else fail naming the missing count).
fn coverage_gate(&mut self, workspace: &str, allow: &[String]) -> GateReport {
let mut surface = Surface::new();
let mut covered: BTreeSet<String> = BTreeSet::new();
// Aggregate coverage across ALL members into ONE constellation-wide surface.
// Each node's id is NAMESPACED by member (`<member>:<tab>`), so the same tab
// name on two apps (nornir + dwarves both have a `Warehouse` tab; korp +
// holger both have `graph`) stays TWO DISTINCT surfaces — one app covering it
// can never falsely cover another's — and a missing node names `app:tab@leg`
// (`viz_tab:<app>:<tab>@<fat|thin>`). A missing surface on ANY member reds the
// whole gate.
for (name, member) in self.members.iter() {
for mut node in viz_tabs(member.surface.iter().cloned()) {
// Driven check uses the BARE tab id (egui_tabs/web_tabs store bare
// tabs); namespace only AFTER, for the surface/covered keys.
let driven = match node.mode {
Mode::Fat => member.egui_tabs.contains(&node.id),
Mode::Thin => member.web_tabs.contains(&node.id),
Mode::NA => false,
};
node.id = format!("{name}:{}", node.id);
if driven {
covered.insert(node.key_str());
}
surface.extend([node]);
}
}
let allowlist = Allowlist {
entries: allow
.iter()
.map(|k| AllowEntry {
key: k.clone(),
reason: "fleet.runa.ron allow entry".to_string(),
})
.collect(),
};
let report = GateReport::compute(&self.run_id, workspace, &surface, &covered, &allowlist);
let green = report.is_green();
let mut row = TestResultRow::unit(
&self.run_id,
workspace,
"coverage",
"fleet_coverage_gate",
if green { status::PASS } else { status::FAIL },
0.0,
self.ts,
report.summary(),
);
row.aspect = "coverage".to_string();
row.metric = report.gap.missing.len() as f64;
self.rows.push(row);
self.log.push(format!("gate `{workspace}` → {}", report.summary()));
report
}
/// Emit one drive `TestResultRow` (pass/fail) for a scenario leg, carrying the
/// leg's own failure `message` (a wire fault / failed scenario step for the fat
/// leg, the not-Ready reason for a dead member).
fn push_drive_row(&mut self, member: &str, suite: &str, leg: &str, tab: &str, row_status: &str, message: &str) {
let mut row = TestResultRow::unit(
&self.run_id,
member,
suite,
format!("{member}::{leg}::{tab}"),
row_status,
0.0,
self.ts,
message,
);
row.aspect = "robot-ui".to_string();
self.rows.push(row);
}
/// The current status of a spun member's boot job (by job id).
fn member_status(&self, job_id: &str) -> Option<BootStatus> {
self.boots
.jobs()
.iter()
.find(|j| j.job_id == job_id)
.map(|j| j.status.clone())
}
}
impl Drop for FleetHarness {
/// **RampGuard-style RAII teardown** — the fleet handle drops → every member
/// still running is killed through jera's boot ledger, so a crashed / panicked
/// run never leaks a container (`draupnir down`, the design's atomic teardown).
fn drop(&mut self) {
let ids: Vec<String> = self.members.values().map(|m| m.job_id.clone()).collect();
for id in ids {
let _ = self.boots.kill(&id);
}
}
}
// ─────────────────────────────────────────────────────────────────────────────
// The egui wire seam — the fat leg's transport to a member's CmdJsonServer
// ─────────────────────────────────────────────────────────────────────────────
/// Where the **fat (egui) drive leg** gets a wire channel to a member's mounted
/// [`CmdJsonServer`](nornir_robotui::cmd_json::CmdJsonServer). This is the seam that
/// turns the egui leg from co-located into genuinely wire-remote: a real run returns
/// an [`IoChannel`] over the container's socket / TCP port (see [`TcpEguiWire`]); a
/// headless test returns an in-memory `duplex` whose far end a `CmdJsonServer`
/// serves. The protocol above never names a transport — this seam picks one.
///
/// `open` returning `Ok(None)` means the member mounts NO egui `CmdJsonServer` yet,
/// so the fat leg records it as **probe-only / not-yet-driven** ([`FatLeg::ProbeOnly`]):
/// its `@fat` node stays UNCOVERED (the gate reds), with a distinct SKIP row naming
/// the per-app mount TODO. This is deliberately NOT a co-located green — an un-mounted
/// app must never silently count as covered (see [`MountRouter`], the constellation
/// router that answers `Ok(None)` for every app that has not grown a mount yet).
pub trait EguiWire {
/// Open a fresh wire channel to `member`'s egui `CmdJsonServer` (dialing one of
/// its `ports`), or `Ok(None)` when the member mounts no egui endpoint yet
/// (→ probe-only, honestly uncovered).
fn open(&mut self, member: &str, ports: &[u16]) -> io::Result<Option<Box<dyn LineChannel>>>;
/// The robot [`UiStep`]s to drive for one viz `tab`. The default is empty — the
/// fleet then drives a liveness+state probe (`ping` + `read_state`) over the
/// wire. A real per-app source returns the tab's recorded robot steps (the
/// per-app scenario library is the follow-up the `TODO` names).
fn scenario(&self, _member: &str, _tab: &str) -> Vec<UiStep> {
Vec::new()
}
}
/// A `Box<dyn LineChannel>` re-exposed as a [`LineChannel`] so the fleet can build a
/// [`CmdJsonClient`] over a dynamically-chosen transport (a socket in a live run, a
/// duplex in a test). `fleet` owns this newtype, so the impl is orphan-rule clean —
/// no second command path, just a boxed carrier for the one robotui vocabulary.
struct BoxedLineChannel(Box<dyn LineChannel>);
impl LineChannel for BoxedLineChannel {
fn write_line(&mut self, line: &str) -> io::Result<()> {
self.0.write_line(line)
}
fn read_line(&mut self) -> io::Result<Option<String>> {
self.0.read_line()
}
}
/// The **real-socket** egui wire source: connects an [`IoChannel`] over TCP to a
/// spun member's mounted `CmdJsonServer` (its first declared port, on `host`).
/// Wired + compiled here so the live path is real; a headless run has no server to
/// reach, so tests exercise the drive over an in-memory `duplex` instead (see
/// `tests/fleet_harness.rs`).
///
/// TODO(per-app CmdJsonServer mount + real OCI run): the live path is not yet
/// exercised end-to-end because no app serves this port. Two follow-ups, each a
/// SEPARATE change (not made here — do NOT touch nornir-server from this crate):
/// 1. Every app (nornir-server, korp, …) must MOUNT a `CmdJsonServer` over its
/// egui UI on this port (its own blocking I/O task calling `serve`), so a
/// containerized member actually answers the wire.
/// 2. A Unix-socket variant now exists ([`UnixEguiWire`], below) for a member that
/// mounts its `CmdJsonServer` on a local socket rather than a TCP port — the
/// first real mount (dwarves-gui's `--robot-remote <socket>`) uses it. The TCP
/// path stays for a container that publishes a port.
/// Once every app serves, the co-located `Ok(None)` fallback becomes a hard wire
/// requirement (an unmounted member's fat leg then legitimately reds).
pub struct TcpEguiWire {
host: String,
}
impl TcpEguiWire {
/// Dial the member on loopback (`127.0.0.1`) — the port comes from its spec.
pub fn loopback() -> Self {
Self { host: "127.0.0.1".to_string() }
}
/// Dial the member on an explicit host.
pub fn on_host(host: impl Into<String>) -> Self {
Self { host: host.into() }
}
}
impl EguiWire for TcpEguiWire {
fn open(&mut self, _member: &str, ports: &[u16]) -> io::Result<Option<Box<dyn LineChannel>>> {
let Some(&port) = ports.first() else {
return Ok(None); // no port declared → no egui endpoint
};
let stream = std::net::TcpStream::connect((self.host.as_str(), port))?;
let reader = BufReader::new(stream.try_clone()?);
Ok(Some(Box::new(IoChannel::new(reader, stream))))
}
}
/// The **Unix-socket** egui wire source: connects an [`IoChannel`] over a local
/// `UnixStream` to a member's mounted `CmdJsonServer` (the follow-up #2 the
/// [`TcpEguiWire`] TODO named). Each member serves on a socket file
/// `<dir>/<member>.sock`; the member's declared `ports` are irrelevant here (a Unix
/// mount is addressed by path, not port). This is the transport dwarves-gui's
/// `--robot-remote <socket>` mount answers — the FIRST real app to serve the fat
/// wire — so the co-located `Ok(None)` fallback never fires for a member routed
/// through this source: a missing / dead socket reds (the red-when-broken signal),
/// it does not silently pass.
///
/// REUSE: this is the SAME `IoChannel`-over-a-byte-stream the TCP source builds and
/// the SAME robotui `CmdJsonClient` vocabulary the fleet drives — only the carrier
/// (a `UnixStream`) differs. Unix-only (`cfg(unix)`), like the socket it dials.
#[cfg(unix)]
pub struct UnixEguiWire {
dir: PathBuf,
}
#[cfg(unix)]
impl UnixEguiWire {
/// Route every member to `<dir>/<member>.sock` — the directory a fleet run
/// mounts each member's `CmdJsonServer` socket under.
pub fn in_dir(dir: impl Into<PathBuf>) -> Self {
Self { dir: dir.into() }
}
/// The socket path a `member` is expected to serve on (`<dir>/<member>.sock`).
/// The mounting app binds this exact path (see dwarves-gui `serve_robot_remote`).
pub fn socket_path(&self, member: &str) -> PathBuf {
self.dir.join(format!("{member}.sock"))
}
}
#[cfg(unix)]
impl EguiWire for UnixEguiWire {
fn open(&mut self, member: &str, _ports: &[u16]) -> io::Result<Option<Box<dyn LineChannel>>> {
let path = self.socket_path(member);
let stream = std::os::unix::net::UnixStream::connect(&path)?;
let reader = BufReader::new(stream.try_clone()?);
Ok(Some(Box::new(IoChannel::new(reader, stream))))
}
}
/// The **constellation egui-wire router** — the fleet's HONEST per-app fat-leg
/// source. A member in the `mounted` set is driven for real over a [`UnixEguiWire`]
/// (its `CmdJsonServer` socket under `dir`); every OTHER member is honestly
/// **probe-only** — `open()` returns `Ok(None)`, which the fat leg records as
/// [`FatLeg::ProbeOnly`] (a distinct not-yet-driven SKIP, its `@fat` node uncovered →
/// the gate reds) rather than a false co-located green.
///
/// This is exactly the fleet-of-five reality today: **only `dwarves` mounts** (via
/// dwarves-gui's `--robot-remote <socket>`), so `MountRouter::new(dir, ["dwarves"])`
/// drives dwarves genuinely and leaves nornir / holger / korp / ordning honestly
/// un-driven. As each app grows its own `CmdJsonServer` mount (mirroring dwarves'
/// `--robot-remote`), add its name to `mounted` and it flips from probe-only to
/// really-driven — no other change. A member whose socket is DEAD is NOT probe-only:
/// the real [`UnixEguiWire::open`] errors → [`FatLeg::Failed`] (a broken mount reds
/// visibly, distinct from the honest not-yet-mounted skip).
#[cfg(unix)]
pub struct MountRouter {
inner: UnixEguiWire,
mounted: BTreeSet<String>,
}
#[cfg(unix)]
impl MountRouter {
/// Route members in `mounted` to a real `UnixEguiWire` under `dir`
/// (`<dir>/<member>.sock`); every other member is probe-only (`open`→`Ok(None)`).
pub fn new(dir: impl Into<PathBuf>, mounted: impl IntoIterator<Item = impl Into<String>>) -> Self {
Self {
inner: UnixEguiWire::in_dir(dir),
mounted: mounted.into_iter().map(Into::into).collect(),
}
}
/// Whether `member` is declared as really-mounted (routed to the live socket).
pub fn is_mounted(&self, member: &str) -> bool {
self.mounted.contains(member)
}
}
#[cfg(unix)]
impl EguiWire for MountRouter {
fn open(&mut self, member: &str, ports: &[u16]) -> io::Result<Option<Box<dyn LineChannel>>> {
if self.mounted.contains(member) {
// Really mounted → dial the live CmdJsonServer socket (a dead socket
// errors → FatLeg::Failed, the broken-mount red).
self.inner.open(member, ports)
} else {
// No mount yet → probe-only. Honest not-yet-driven, distinct from a dead
// socket: the `@fat` node stays uncovered (the gate reds), never a false green.
Ok(None)
}
}
}
// ─────────────────────────────────────────────────────────────────────────────
// LocalProcessSpinner — spin fleet members as LOCAL PROCESSES (no container host)
// ─────────────────────────────────────────────────────────────────────────────
/// The candidate **mounted set** for a local run — every app that mounts (or is
/// landing a mount for) a `CmdJsonServer` over its egui UI (`--robot-remote
/// <socket>`), so the fat leg can drive it wire-remote over a real Unix socket.
///
/// This is the RUN-PATH config (item 3 of the local-fleet build): the honesty
/// tests keep their OWN `MountedSetWire` expectations — this list is consumed only
/// by [`LocalProcessSpinner`] / the `run_local_fleet` example. `nornir` is on the
/// list because its mount is landing in parallel; a run that does not spin it leaves
/// it **probe-only** (the [`MountRouter`] mounted set the run installs is the set
/// actually spun Ready, so an un-spun candidate is honestly not-yet-driven, NOT a
/// dead-socket `Failed`).
#[cfg(unix)]
pub const CANONICAL_MOUNTED_MEMBERS: &[&str] = &["dwarves", "holger", "korp", "ordning", "nornir"];
/// The placeholder token a [`LaunchSpec`]'s args carry where the member's socket
/// path is substituted in at spin time (`--robot-remote {socket}` → the real
/// `<dir>/<member>.sock`). Keeps a spec pure DATA — the socket dir is chosen by the
/// run, not baked into the spec.
#[cfg(unix)]
pub const SOCKET_PLACEHOLDER: &str = "{socket}";
/// How one fleet member is launched as a LOCAL PROCESS — pure DATA (a program +
/// args, an optional working directory, a readiness timeout). The socket the app
/// binds is `<dir>/<member>.sock`, injected by the spinner wherever an arg holds the
/// [`SOCKET_PLACEHOLDER`]. A run config is a map of member name → `LaunchSpec`, so
/// which apps spin (and how) is declared, not hard-coded in the spinner.
#[cfg(unix)]
#[derive(Debug, Clone)]
pub struct LaunchSpec {
/// The program to exec (e.g. `cargo`, or a built binary path).
pub program: String,
/// Its args; any arg containing [`SOCKET_PLACEHOLDER`] gets the socket path
/// substituted (e.g. `["run", "-p", "dwarves-gui", "--", "--robot-remote", "{socket}"]`).
pub args: Vec<String>,
/// The working directory to launch in (a sibling app repo for `cargo run`), or
/// `None` to inherit the spinner's cwd.
pub cwd: Option<PathBuf>,
/// How long to poll for the socket to appear + answer a `Ping` before giving up
/// (the member is then recorded NOT Ready — the gate reds, no false green).
pub ready_timeout: Duration,
}
#[cfg(unix)]
impl LaunchSpec {
/// A spec for `program` with `args` (default 60s readiness timeout, inherited cwd).
pub fn new(program: impl Into<String>, args: impl IntoIterator<Item = impl Into<String>>) -> Self {
Self {
program: program.into(),
args: args.into_iter().map(Into::into).collect(),
cwd: None,
ready_timeout: Duration::from_secs(60),
}
}
/// Launch in `cwd` (builder) — a sibling app repo for the `cargo run` specs.
pub fn in_dir(mut self, cwd: impl Into<PathBuf>) -> Self {
self.cwd = Some(cwd.into());
self
}
/// Override the readiness poll timeout (builder).
pub fn with_timeout(mut self, timeout: Duration) -> Self {
self.ready_timeout = timeout;
self
}
/// The args with [`SOCKET_PLACEHOLDER`] resolved to `socket`.
fn resolved_args(&self, socket: &Path) -> Vec<String> {
let socket = socket.to_string_lossy();
self.args
.iter()
.map(|a| a.replace(SOCKET_PLACEHOLDER, &socket))
.collect()
}
}
/// Spin fleet members as **local child processes** and hold each on a bounded
/// readiness poll until its `CmdJsonServer` socket appears and answers a `Ping` —
/// the headless, no-container spin backend for a real local fleet run. It is the
/// process/socket twin of jera's boot ledger: a member that never comes Ready is
/// recorded so (the gate then reds — red-when-broken, no false green), and
/// **Drop = RampGuard teardown**: every spawned child is killed + reaped and every
/// socket removed, so a crashed run leaks neither a process nor a socket file.
///
/// ## Concurrency (ROOT LAW #0)
/// NO `thread::spawn` / rayon: spinning is the `std::process` API (spawn a child)
/// plus a **bounded poll loop on the calling thread** (`try_wait` + a socket `Ping`,
/// `sleep`ing between polls). Process spawning + blocking I/O is not CPU fan-out.
///
/// ## Composition with the harness
/// A run spins the apps here FIRST (they bind their sockets), then lowers the
/// `fleet.runa.ron` pipeline through [`FleetHarness::mock`] with a [`MountRouter`]
/// (wrapped in a [`ScenarioWire`]) whose mounted set is exactly the members this
/// spinner brought Ready — so the fat leg dials the LIVE sockets these children
/// serve, and un-spun candidates stay honestly probe-only. The spinner must OUTLIVE
/// the harness run (drop it after, to tear the fleet down).
#[cfg(unix)]
pub struct LocalProcessSpinner {
dir: PathBuf,
children: BTreeMap<String, std::process::Child>,
ready: BTreeSet<String>,
poll_interval: Duration,
}
#[cfg(unix)]
impl LocalProcessSpinner {
/// A spinner that binds each member's socket under `dir` (`<dir>/<member>.sock`).
pub fn new(dir: impl Into<PathBuf>) -> Self {
Self {
dir: dir.into(),
children: BTreeMap::new(),
ready: BTreeSet::new(),
poll_interval: Duration::from_millis(100),
}
}
/// Override the readiness poll interval (default 100ms) — builder.
pub fn with_poll_interval(mut self, interval: Duration) -> Self {
self.poll_interval = interval;
self
}
/// The socket path `member` is expected to serve on (`<dir>/<member>.sock`) — the
/// SAME path a [`UnixEguiWire`] / [`MountRouter`] under the same `dir` dials.
pub fn socket_path(&self, member: &str) -> PathBuf {
self.dir.join(format!("{member}.sock"))
}
/// Spawn `member`'s app per `spec`, then bounded-poll (`try_wait` + a socket
/// `Ping`) until it is Ready, its child exits early, or the timeout lapses.
/// Returns whether it reached Ready. The child is tracked either way (a NOT-Ready
/// child is still killed on teardown). A `spawn` failure (bad program) is the one
/// hard `Err` — nothing was launched to track.
pub fn spin(&mut self, member: &str, spec: &LaunchSpec) -> io::Result<bool> {
let socket = self.socket_path(member);
let _ = std::fs::remove_file(&socket); // clear a stale socket so the app's re-bind is clean
if let Some(parent) = socket.parent() {
let _ = std::fs::create_dir_all(parent);
}
let mut cmd = std::process::Command::new(&spec.program);
cmd.args(spec.resolved_args(&socket));
if let Some(cwd) = &spec.cwd {
cmd.current_dir(cwd);
}
let mut child = cmd.spawn()?;
// Bounded readiness barrier on THIS thread (ROOT LAW #0 — no spawn): poll the
// child liveness + the socket until the app serves a Ping, dies, or times out.
let deadline = Instant::now() + spec.ready_timeout;
let mut ready = false;
while Instant::now() < deadline {
if child.try_wait()?.is_some() {
break; // the child exited before serving — NOT Ready
}
if socket.exists() && socket_answers_ping(&socket) {
ready = true;
break;
}
std::thread::sleep(self.poll_interval);
}
self.children.insert(member.to_string(), child);
if ready {
self.ready.insert(member.to_string());
}
Ok(ready)
}
/// Whether `member` was spun and reached Ready (its socket answered a `Ping`).
pub fn is_ready(&self, member: &str) -> bool {
self.ready.contains(member)
}
/// The members that reached Ready — the exact set a run installs as the
/// [`MountRouter`] mounted set (so un-spun candidates stay probe-only).
pub fn ready_members(&self) -> Vec<String> {
self.ready.iter().cloned().collect()
}
/// The OS pid of a spun member's child (for a teardown assertion), if tracked.
pub fn pid(&self, member: &str) -> Option<u32> {
self.children.get(member).map(|c| c.id())
}
/// Kill + reap every tracked child and remove every socket file — the RampGuard
/// teardown, run on Drop and callable directly. Idempotent (drains the map).
pub fn teardown(&mut self) {
for (member, mut child) in std::mem::take(&mut self.children) {
let _ = child.kill();
let _ = child.wait(); // reap the zombie so the pid is fully gone
let _ = std::fs::remove_file(self.socket_path(&member));
}
self.ready.clear();
}
}
#[cfg(unix)]
impl Drop for LocalProcessSpinner {
/// **RampGuard-style RAII teardown** — the spinner drops → every child is killed
/// + reaped and every socket removed, so a crashed / panicked local run leaks
/// neither a process nor a stale socket file (the local twin of the harness's own
/// boot-ledger teardown).
fn drop(&mut self) {
self.teardown();
}
}
/// Whether a `CmdJsonServer` is live on `path`: connect a `UnixStream`, wrap it in
/// the SAME [`IoChannel`] + [`CmdJsonClient`] the fat leg uses (REUSE — one wire
/// vocabulary), and roundtrip a `Ping`. A short read timeout keeps a bound-but-not-
/// yet-serving socket from hanging the readiness poll. Any failure = not-yet-live.
#[cfg(unix)]
fn socket_answers_ping(path: &Path) -> bool {
let Ok(stream) = std::os::unix::net::UnixStream::connect(path) else {
return false;
};
let _ = stream.set_read_timeout(Some(Duration::from_secs(2)));
let Ok(reader) = stream.try_clone() else {
return false;
};
let mut client = CmdJsonClient::new(IoChannel::new(BufReader::new(reader), stream));
client.ping().is_ok()
}
// ─────────────────────────────────────────────────────────────────────────────
// The per-member scenario source — LOAD an app's recorded per-tab scenarios (the
// wire DATA it emits) and feed them into the fat leg's `EguiWire::scenario` seam,
// so `RobotDriveMember` drives the app's REAL recorded steps instead of the bare
// liveness probe. No build-dep on the app binary: the scenarios cross as serde JSON
// (the SAME `wire` form dwarves-gui's `scenarios --emit` prints — REUSE, no second
// vocabulary), keyed by member name.
// ─────────────────────────────────────────────────────────────────────────────
/// The default env var naming the directory a fleet run reads each member's emitted
/// scenarios from (`<dir>/<member>.scenarios.json`).
pub const FLEET_SCENARIOS_DIR_ENV: &str = "FLEET_SCENARIOS_DIR";
/// A map **member name → the app's recorded [`Scenario`]s** (each `suite` a viz-tab
/// name), loaded from the wire JSON the app emits (dwarves-gui: `scenarios --emit`).
///
/// This is the per-member scenario source the fleet's fat leg drives: for a member
/// with loaded scenarios, [`steps_for`](Self::steps_for) returns the recorded steps
/// of the tab's scenario (matched by `Scenario::suite == tab`); a member with NO
/// scenarios returns an empty step list, so the fat leg keeps its existing
/// select+`read_state` probe fallback (unchanged). The fleet never build-deps the
/// app binary — the scenarios cross as serde JSON keyed by member name.
#[derive(Debug, Clone, Default)]
pub struct ScenarioSource {
by_member: BTreeMap<String, Vec<Scenario>>,
}
impl ScenarioSource {
/// An empty source (every member falls back to the select+probe drive).
pub fn new() -> Self {
Self::default()
}
/// Register a member's scenarios directly (the setter a test/harness populates —
/// e.g. from `dwarves_tab_scenarios()` decoded in-process). Builder style.
pub fn with_scenarios(mut self, member: impl Into<String>, scenarios: Vec<Scenario>) -> Self {
self.by_member.insert(member.into(), scenarios);
self
}
/// Register a member's scenarios directly (in-place; the non-builder twin of
/// [`with_scenarios`](Self::with_scenarios)).
pub fn insert(&mut self, member: impl Into<String>, scenarios: Vec<Scenario>) {
self.by_member.insert(member.into(), scenarios);
}
/// Decode a member's scenarios from the emitted wire JSON (`serde_json` of a
/// `Vec<Scenario>` — exactly what `dwarves-gui scenarios --emit` prints). A parse
/// error is surfaced (a malformed emit reds at load, never a silent empty drive).
pub fn load_member_json(&mut self, member: impl Into<String>, json: &str) -> io::Result<()> {
let scenarios: Vec<Scenario> = serde_json::from_str(json)
.map_err(|e| io::Error::new(io::ErrorKind::InvalidData, e))?;
self.by_member.insert(member.into(), scenarios);
Ok(())
}
/// Load a member's scenarios from `<dir>/<member>.scenarios.json`. Returns
/// `Ok(false)` when the file is ABSENT (the member simply has no recorded
/// scenarios → probe fallback, not an error); `Ok(true)` when it loaded; `Err`
/// only when a PRESENT file fails to read/parse (red-when-broken at load).
pub fn load_member_from_dir(&mut self, dir: impl AsRef<Path>, member: &str) -> io::Result<bool> {
let path = dir.as_ref().join(format!("{member}.scenarios.json"));
match std::fs::read_to_string(&path) {
Ok(json) => {
self.load_member_json(member, &json)?;
Ok(true)
}
Err(e) if e.kind() == io::ErrorKind::NotFound => Ok(false),
Err(e) => Err(e),
}
}
/// Load every named member's scenarios from `dir` (`<dir>/<member>.scenarios.json`),
/// skipping members whose file is absent (they keep the probe fallback). Propagates
/// a read/parse error for a present-but-broken file.
pub fn from_dir<'a>(
dir: impl AsRef<Path>,
members: impl IntoIterator<Item = &'a str>,
) -> io::Result<Self> {
let mut src = Self::new();
for member in members {
src.load_member_from_dir(dir.as_ref(), member)?;
}
Ok(src)
}
/// Load from the directory named by [`FLEET_SCENARIOS_DIR_ENV`]. When the env var
/// is unset/empty, returns an EMPTY source (every member falls back to the probe —
/// so a run with no emitted scenarios behaves exactly as before).
pub fn from_env<'a>(members: impl IntoIterator<Item = &'a str>) -> io::Result<Self> {
match std::env::var(FLEET_SCENARIOS_DIR_ENV) {
Ok(dir) if !dir.trim().is_empty() => Self::from_dir(dir, members),
_ => Ok(Self::new()),
}
}
/// Whether `member` has any loaded scenarios (else it drives the probe fallback).
pub fn has(&self, member: &str) -> bool {
self.by_member.get(member).is_some_and(|v| !v.is_empty())
}
/// The recorded steps for `member`'s `tab` — the scenario whose `suite == tab`.
/// Empty when the member has no scenarios or none matches (→ probe fallback).
pub fn steps_for(&self, member: &str, tab: &str) -> Vec<UiStep> {
self.by_member
.get(member)
.and_then(|scenarios| scenarios.iter().find(|s| s.suite == tab))
.map(|s| s.steps.clone())
.unwrap_or_default()
}
}
/// Wrap ANY [`EguiWire`] transport with a [`ScenarioSource`]: `open()` delegates to
/// the inner wire (the real socket / duplex / router — membership untouched), while
/// [`scenario`](EguiWire::scenario) is served from the LOADED per-member scenarios so
/// the fat leg drives the app's REAL recorded steps. A member/tab with no loaded
/// scenario falls through to the inner wire's own `scenario` (default: empty → the
/// select+`read_state` probe), so un-emitted members keep the existing fallback.
///
/// This is the production seam: `ScenarioWire::new(MountRouter::new(dir, ["dwarves"]),
/// ScenarioSource::from_env(["dwarves", …])?)` drives dwarves' emitted per-tab
/// scenarios over its real mount, and leaves every un-emitted / un-mounted member on
/// its honest fallback — additive to `MountRouter`, which it does not modify.
pub struct ScenarioWire<W> {
inner: W,
source: ScenarioSource,
}
impl<W> ScenarioWire<W> {
/// Drive `inner`'s transport, serving recorded steps from `source`.
pub fn new(inner: W, source: ScenarioSource) -> Self {
Self { inner, source }
}
/// Borrow the loaded scenario source (e.g. to assert what was loaded).
pub fn source(&self) -> &ScenarioSource {
&self.source
}
}
impl<W: EguiWire> EguiWire for ScenarioWire<W> {
fn open(&mut self, member: &str, ports: &[u16]) -> io::Result<Option<Box<dyn LineChannel>>> {
self.inner.open(member, ports)
}
fn scenario(&self, member: &str, tab: &str) -> Vec<UiStep> {
let steps = self.source.steps_for(member, tab);
if steps.is_empty() {
// No recorded scenario for this member/tab → the inner wire's own fallback
// (default empty → the fat leg's select+read_state liveness probe).
self.inner.scenario(member, tab)
} else {
steps
}
}
}
/// The honest outcome of the fat (egui) drive leg for ONE tab. Kept a three-way
/// verdict (not a `bool`) so an un-mounted app reads as a **visibly distinct**
/// probe-only skip — its `@fat` node stays uncovered → the gate reds — rather than a
/// false co-located green. This is the constellation's whole value: no false green.
enum FatLeg {
/// Really driven clean over the wire (or the legacy no-wire mock) → covers `@fat`.
Driven,
/// The app mounts no `CmdJsonServer` yet: `open()` answered `Ok(None)`. NOT
/// covered — the fleet records a distinct not-yet-driven SKIP row naming the
/// per-app mount TODO (mirror dwarves-gui's `--robot-remote`).
ProbeOnly,
/// A wire fault / dead host / failed scenario step → NOT covered, a FAILED row
/// carrying the reason (the red-when-broken signal).
Failed(String),
}
/// Drive ONE egui tab **wire-remote**: open a channel to the member's
/// `CmdJsonServer`, wrap it in a [`CmdJsonClient`], `ping`, then drive the tab's
/// scenario (or a liveness+state probe when the source declares no steps). Returns
/// the [`FatLeg`] verdict the fleet turns into a `TestResultRow` + coverage bit. The
/// whole point: a wire fault, a dead host, or a failed scenario step each surface
/// here as [`FatLeg::Failed`] — the red-when-broken signal — and an app with no mount
/// yet surfaces as [`FatLeg::ProbeOnly`] (honest not-driven), never a swallowed green.
fn drive_egui_tab(wire: &mut dyn EguiWire, member: &str, ports: &[u16], tab: &str) -> FatLeg {
let chan = match wire.open(member, ports) {
Ok(Some(chan)) => chan,
// An INSTALLED wire that declares no endpoint for this member → PROBE-ONLY.
// NOT a co-located green: the app has not mounted a CmdJsonServer, so its
// `@fat` node is honestly not-yet-driven (uncovered → the gate reds).
Ok(None) => return FatLeg::ProbeOnly,
Err(e) => return FatLeg::Failed(format!("egui wire open failed: {e}")),
};
let mut client = CmdJsonClient::new(BoxedLineChannel(chan));
// Handshake travels the wire — a dead / unmounted host reds here.
if let Err(e) = client.ping() {
return FatLeg::Failed(format!("egui host did not answer ping: {e}"));
}
let steps = wire.scenario(member, tab);
if steps.is_empty() {
// No recorded per-app steps yet (the scenario library is a follow-up):
// prove the remote app is live and renders an observable state object.
return match client.read_state() {
Ok(serde_json::Value::Object(_)) => FatLeg::Driven,
Ok(other) => FatLeg::Failed(format!("remote state_json is not an object: {other}")),
Err(e) => FatLeg::Failed(format!("read_state failed over the wire: {e}")),
};
}
// Drive each step remotely; the first failed cell reds the tab.
for step in steps {
match client.drive_step(step) {
Ok((true, _)) => {}
Ok((false, why)) => return FatLeg::Failed(why),
Err(e) => return FatLeg::Failed(format!("drive_step wire error: {e}")),
}
}
FatLeg::Driven
}
/// Microseconds since the Unix epoch (UTC) — the `TestResultRow::ts_micros` clock.
fn now_micros() -> i64 {
SystemTime::now()
.duration_since(UNIX_EPOCH)
.map(|d| d.as_micros() as i64)
.unwrap_or(0)
}
/// The stable label of a passthrough step (build/provision predecessors the fleet
/// harness records but does not itself lower — dwarves' `runa_exec` owns those).
fn step_label(step: &Step) -> &'static str {
match step {
Step::CargoBuild { .. } => "CargoBuild",
Step::Nextest { .. } => "Nextest",
Step::CargoTest { .. } => "CargoTest",
Step::Bench { .. } => "Bench",
Step::Coverage { .. } => "Coverage",
Step::Trace { .. } => "Trace",
Step::Profile { .. } => "Profile",
Step::LiveTrace { .. } => "LiveTrace",
Step::Lint { .. } => "Lint",
Step::Build { .. } => "Build",
Step::BuildWorkspace { .. } => "BuildWorkspace",
Step::ContainerBuild { .. } => "ContainerBuild",
Step::VmBoot => "VmBoot",
Step::Sbom { .. } => "Sbom",
Step::HolgerPackage { .. } => "HolgerPackage",
Step::ReleaseDoctor => "ReleaseDoctor",
Step::PublishCrates => "PublishCrates",
Step::DeployPages => "DeployPages",
Step::SecurityScan { .. } => "SecurityScan",
Step::PushImage { .. } => "PushImage",
Step::Deploy { .. } => "Deploy",
Step::ApprovalGate { .. } => "ApprovalGate",
Step::Notify { .. } => "Notify",
Step::ProvisionTestEnv { .. } => "ProvisionTestEnv",
Step::Xtask { .. } => "Xtask",
Step::SpinFleetMember { .. } => "SpinFleetMember",
Step::RobotDriveMember { .. } => "RobotDriveMember",
Step::CoverageGate { .. } => "CoverageGate",
Step::Sandbox { .. } => "Sandbox",
Step::Forge { .. } => "Forge",
}
}