src/viz/live.rs
//! 📡 Live release run pane.
//!
//! **Durable source = the warehouse `release_events` table** (LAW
//! "persist-to-warehouse-stream-not-tmp"). On load/reload the pane *hydrates*
//! from `release_events` — the same table the 🚀 Release tab and the
//! `nornir release events` CLI read — so a run **survives a reboot** and is
//! **visible to other clients** the moment it lands in the warehouse, exactly
//! like every other release surface.
//!
//! On top of that durable baseline, a **fast live-tail** keeps the pane moving
//! at sub-second latency while a run is in flight:
//! * **local** mode tails `<workspace>/.nornir/logs/release-run-*.events.ndjson`
//! (the producer flushes there per line — cheaper than a warehouse append per
//! event), and
//! * **remote** (thin-client) mode streams the run from a `nornir-server` over
//! `Release.Progress`.
//!
//! The tail is a *fallback accelerator*, never the source of truth: if the
//! `/tmp` log is gone (reboot, GC) the warehouse-hydrated events still render,
//! and a reload re-reads the warehouse. eframe's `App::update` runs sync at
//! ~60 Hz so no tokio/SSE client is needed for the local pane.
use std::fs::File;
use std::io::{BufRead, BufReader, Seek, SeekFrom};
use std::path::{Path, PathBuf};
use std::sync::{Arc, Mutex};
use std::thread;
use std::time::Duration;
use eframe::egui::{self, RichText, ScrollArea};
use super::facett_theme::{GREEN, RED, Theme};
// `LiveEvent` lives in `model` (the UNGATED half of this module) because the
// gRPC thin client in `remote` needs it and must build for a server that links
// no egui. See `super::model::LiveEvent`.
use super::model::LiveEvent;
use crate::warehouse::iceberg::IcebergWarehouse;
use crate::warehouse::release_events::{
self, EventSelector, ReleaseEventRow, query_release_events,
};
/// Where the durable hydrate reads `release_events` from (mirrors the other
/// tabs' local/remote split). The fast live-tail (`/tmp` file or `Release.Progress`
/// stream) layers on top of whichever durable baseline this provides.
#[derive(Clone)]
enum Durable {
/// Local warehouse root — hydrate by opening it read-only and querying
/// `release_events` for the newest run.
Local(PathBuf),
/// Remote server — hydrate the durable baseline over `Viz.ReleaseEvents`
/// (the warehouse lives on the server, which holds the redb lock), then the
/// `Release.Progress` stream layers the live tail on top. Carries
/// `(endpoint, token, workspace)`: the workspace scopes the RPC, the endpoint
/// is also the `state_json` label.
Remote {
endpoint: String,
token: String,
workspace: String,
},
/// No durable source (e.g. a test that injects events directly).
None,
}
#[derive(Default)]
struct Shared {
events: Vec<LiveEvent>,
err: Option<String>,
file_path: Option<PathBuf>,
eof_run_ended: bool,
/// True once the warehouse hydrate (the durable read) has populated the
/// baseline — so `state_json` / the pane can show the run survived a reboot
/// even before any `/tmp` tail line arrives.
hydrated_from_warehouse: bool,
/// `run_id` of the run the durable hydrate loaded (the newest in the table).
hydrated_run_id: Option<String>,
}
pub struct LiveRunState {
log_dir: PathBuf,
/// When set (thin-client / remote viz), stream the run from a `nornir-server`
/// over `Release.Progress` instead of tailing a local file. `(endpoint, token)`.
remote: Option<(String, String, String)>,
/// The durable source the pane hydrates from on load/reload (the warehouse).
durable: Durable,
shared: Arc<Mutex<Shared>>,
started: bool,
theme: Theme,
}
impl LiveRunState {
/// `log_dir` is `<workspace_root>/.nornir/logs`; `warehouse_root` is the
/// warehouse the durable `release_events` hydrate reads from. Construction
/// does not hydrate or start the tail — that happens on first `draw()` so a
/// user who never opens the tab pays nothing.
pub fn new(log_dir: PathBuf, warehouse_root: PathBuf) -> Self {
Self {
log_dir,
remote: None,
durable: Durable::Local(warehouse_root),
shared: Arc::new(Mutex::new(Shared::default())),
started: false,
theme: Theme::default(),
}
}
/// Set the facett palette the pane paints with.
pub fn set_palette(&mut self, t: Theme) {
self.theme = t;
}
/// Remote constructor: stream the live run from a running `nornir-server`
/// over the `Release.Progress` gRPC (e.g. a friend's viz watching your
/// server over Tailscale). The server's warehouse is the durable source; the
/// stream is both its hydrate and its live tail.
pub fn new_remote(endpoint: String, token: String, workspace: String) -> Self {
Self {
log_dir: PathBuf::new(),
remote: Some((endpoint.clone(), token.clone(), workspace.clone())),
durable: Durable::Remote {
endpoint,
token,
workspace,
},
shared: Arc::new(Mutex::new(Shared::default())),
started: false,
theme: Theme::default(),
}
}
/// Re-point the thin-mode durable hydrate at a new workspace (the picker
/// switched). No-op when local; the caller follows with `reload()`.
pub fn set_workspace(&mut self, workspace: String) {
if let Durable::Remote { workspace: w, .. } = &mut self.durable {
*w = workspace;
}
}
/// Test-only: inject release-op rows as the durable hydrate (no warehouse on
/// disk), exactly as if `release_events` had been read. Mirrors the Release
/// tab's `inject_for_test` so the inject-and-assert harness can read the
/// hydrated run back out of `state_json`.
#[doc(hidden)]
pub fn inject_for_test(&mut self, rows: Vec<ReleaseEventRow>) {
self.durable = Durable::None;
self.started = true; // don't spawn a tail thread in the headless test
let (events, run_id) = events_from_release_rows(rows);
let mut s = self.shared.lock().unwrap();
s.events = events;
s.hydrated_from_warehouse = true;
s.hydrated_run_id = run_id;
s.err = None;
}
/// Re-scope (workspace switch / reload): re-read the durable warehouse source
/// so the pane re-hydrates from `release_events`. The live tail keeps running.
pub fn reload(&mut self) {
{
let mut s = self.shared.lock().unwrap_or_else(|p| p.into_inner());
s.hydrated_from_warehouse = false;
s.hydrated_run_id = None;
}
self.hydrate_from_warehouse();
}
/// Durable read: open the warehouse read-only and load the **newest run**'s
/// `release_events` rows as the baseline the pane renders. Lock-tolerant +
/// read-only so it coexists with a live server holding the catalog lock, the
/// same read path `nornir release events` / the Release tab use. Best-effort:
/// a read error is recorded into `err` and does not panic the UI.
fn hydrate_from_warehouse(&mut self) {
// Read the durable `release_events` baseline: locally by opening the
// warehouse, remotely over Viz.ReleaseEvents (the server holds the lock).
// The Release.Progress stream then layers the live tail on top — but with
// no run in flight (a historical workspace) the stream is silent, so this
// durable read is what makes the pane non-empty in thin mode.
let res = match &self.durable {
Durable::Local(root) => {
super::trace::emit_in(
"live.hydrate",
&serde_json::json!({ "source": "warehouse.release_events", "root": root.display().to_string() }),
);
IcebergWarehouse::open_read_only(root)
.and_then(|wh| wh.block_on(query_release_events(&wh, &EventSelector::All)))
}
Durable::Remote {
endpoint,
token,
workspace,
} => {
super::trace::emit_in(
"live.hydrate",
&serde_json::json!({ "source": "Viz.ReleaseEvents", "endpoint": endpoint, "workspace": workspace }),
);
super::remote::fetch_release_events(endpoint, token, workspace)
}
Durable::None => return,
};
let mut s = self.shared.lock().unwrap_or_else(|p| p.into_inner());
match res {
Ok(rows) => {
let (events, run_id) = events_from_release_rows(rows);
let n = events.len();
s.events = events;
s.hydrated_from_warehouse = true;
s.hydrated_run_id = run_id.clone();
s.err = None;
nornir_testmatrix::functional_status(
"nornir/viz/live-run",
"hydrate_release_events",
true,
&format!("{n} event(s), run_id={run_id:?}"),
);
super::trace::emit_out(
"live.hydrate",
&serde_json::json!({ "run_id": run_id, "events": n }),
);
}
Err(e) => {
// No table yet / empty warehouse is normal before the first run —
// record it but keep whatever the live tail may have collected.
s.err = Some(format!("{e:#}"));
nornir_testmatrix::functional_status(
"nornir/viz/live-run",
"hydrate_release_events",
false,
&format!("{e:#}"),
);
super::trace::emit_out(
"live.hydrate",
&serde_json::json!({ "error": format!("{e:#}") }),
);
}
}
}
fn ensure_started(&mut self) {
if self.started {
return;
}
self.started = true;
// Durable FIRST: hydrate the baseline from the warehouse so the run is
// visible on load even if no `/tmp` log exists (e.g. after a reboot).
self.hydrate_from_warehouse();
let shared = Arc::clone(&self.shared);
if let Some((endpoint, token, workspace)) = self.remote.clone() {
thread::Builder::new()
.name("nornir-live-remote".into())
.spawn(move || remote_loop(endpoint, token, workspace, shared))
.expect("spawn live remote thread");
} else {
let log_dir = self.log_dir.clone();
thread::Builder::new()
.name("nornir-live-tail".into())
.spawn(move || tail_loop(log_dir, shared))
.expect("spawn tail thread");
}
}
pub fn draw(&mut self, ui: &mut egui::Ui) {
self.ensure_started();
let theme = self.theme;
let remote_ep = self.remote.as_ref().map(|(e, _, _)| e.clone());
// Recover a poisoned lock instead of cascade-panicking the UI thread: if
// a background stream/tail thread panicked mid-write the shared buffer is
// still readable, so degrade to whatever it last held rather than crash.
let shared = self.shared.lock().unwrap_or_else(|p| p.into_inner());
if let Some(err) = &shared.err {
ui.colored_label(RED, err);
}
ui.horizontal(|ui| {
ui.label("durable:");
match &self.durable {
Durable::Local(p) => ui.monospace(format!("warehouse {}", p.display())),
Durable::Remote {
endpoint,
workspace,
..
} => ui.monospace(format!(
"server {endpoint} ws={workspace} (Viz.ReleaseEvents)"
)),
Durable::None => ui.monospace("(injected)"),
};
if shared.hydrated_from_warehouse {
ui.colored_label(GREEN, "✓ hydrated");
} else {
ui.weak("(loading…)");
}
});
ui.horizontal(|ui| {
if let Some(ep) = &remote_ep {
ui.label("live-tail:");
ui.monospace(format!("server {ep}"));
} else {
ui.label("live-tail:");
if let Some(p) = &shared.file_path {
ui.monospace(p.display().to_string());
} else {
ui.weak("(warehouse baseline; waiting for next release-run-*.events.ndjson)");
}
}
});
ui.separator();
let summary = RunSummary::from_events(&shared.events);
ui.horizontal(|ui| {
ui.label(RichText::new("run").strong());
ui.monospace(&summary.run_id);
ui.separator();
ui.label("workspace:");
ui.monospace(&summary.workspace);
ui.separator();
match summary.run_ok {
None => ui.colored_label(theme.accent, "● running"),
Some(true) => ui.colored_label(GREEN, "✓ done"),
Some(false) => ui.colored_label(RED, "✗ failed"),
};
});
ui.horizontal(|ui| {
ui.label(RichText::new("repo").strong());
ui.monospace(&summary.current_repo);
ui.separator();
ui.label("phase:");
ui.monospace(&summary.current_phase);
ui.separator();
ui.label("binary:");
ui.monospace(short_bin(&summary.current_binary));
});
ui.horizontal(|ui| {
ui.colored_label(GREEN, format!("✓ {}", summary.total_pass));
ui.colored_label(RED, format!("✗ {}", summary.total_fail));
});
ui.separator();
// Scrolling log: most recent at top, last 500 lines.
ScrollArea::vertical()
.auto_shrink([false, false])
.stick_to_bottom(true)
.show(ui, |ui| {
for ev in shared.events.iter().rev().take(500).rev() {
render_event_row(ui, ev, &theme);
}
});
// Drive the next frame so live updates appear without user input.
ui.ctx().request_repaint_after(Duration::from_millis(200));
}
/// Test/introspection hook (LAW #6): the live run the pane is rendering —
/// the durable source, whether the warehouse hydrate landed, the run header,
/// and the rendered event lines — folded into the app's `state_json`. This is
/// the proof the pane reads from the warehouse, not just `/tmp`.
pub fn state_json(&self) -> serde_json::Value {
let shared = self.shared.lock().unwrap_or_else(|p| p.into_inner());
let summary = RunSummary::from_events(&shared.events);
let lines: Vec<String> = shared.events.iter().map(event_line).collect();
serde_json::json!({
"durable_source": match &self.durable {
Durable::Local(p) => format!("warehouse {}", p.display()),
Durable::Remote { endpoint, workspace, .. } => {
format!("server {endpoint} ws={workspace} (Viz.ReleaseEvents)")
}
Durable::None => "injected".to_string(),
},
"hydrated_from_warehouse": shared.hydrated_from_warehouse,
"hydrated_run_id": shared.hydrated_run_id,
"live_tail": match &self.remote {
Some((e, _, _)) => format!("server {e} (Release.Progress)"),
None => shared.file_path.as_ref()
.map(|p| format!("file {}", p.display()))
.unwrap_or_else(|| "(none yet)".to_string()),
},
"error": shared.err,
"run_id": summary.run_id,
"workspace": summary.workspace,
"run_ok": summary.run_ok,
"current_repo": summary.current_repo,
"current_phase": summary.current_phase,
"total_pass": summary.total_pass,
"total_fail": summary.total_fail,
"events": shared.events.len(),
"lines": lines,
"palette": self.theme.name,
})
}
}
/// Convert durable `release_events` rows (newest run only) into the `LiveEvent`
/// stream the pane renders, so the warehouse table and the `/tmp`/SSE live tail
/// feed one identical render path. Returns the events + the hydrated `run_id`.
///
/// Mapping (the durable `op × phase × status` boundary → live event):
/// * `op == "run"`, `phase == start` → `RunStart` (carries `run_id`)
/// * `op == "run"`, `phase == end` → `RunEnd` (`ok = status == ok`)
/// * any op, `phase == start` → `PhaseStart` (the `[repo] op` boundary)
/// * any op, `phase == end` → `PhaseEnd` (`ok` + `duration_ms`)
fn events_from_release_rows(rows: Vec<ReleaseEventRow>) -> (Vec<LiveEvent>, Option<String>) {
use release_events::{phase, status};
// Pick the newest run (max ts), so the pane shows the live/most-recent run —
// the same "newest run on top" rule the Release tab uses.
let mut latest: Option<(&str, i64)> = None;
for r in &rows {
let cand = (r.run_id.as_str(), r.ts_micros);
if latest.map_or(true, |(_, ts)| cand.1 >= ts) {
latest = Some(cand);
}
}
let Some((run_id, _)) = latest else {
return (Vec::new(), None);
};
let run_id = run_id.to_string();
let mut run_rows: Vec<&ReleaseEventRow> = rows.iter().filter(|r| r.run_id == run_id).collect();
run_rows.sort_by_key(|r| r.seq);
let mut out: Vec<LiveEvent> = Vec::with_capacity(run_rows.len());
for r in run_rows {
let ev = match (r.op.as_str(), r.phase.as_str()) {
("run", p) if p == phase::START => LiveEvent::RunStart {
run_id: r.run_id.clone(),
// `run/start` detail carries the workspace label when present.
workspace: if r.detail.is_empty() {
r.run_id.clone()
} else {
r.detail.clone()
},
},
("run", p) if p == phase::END => LiveEvent::RunEnd {
run_id: r.run_id.clone(),
ok: r.status == status::OK,
},
(_, p) if p == phase::START => LiveEvent::PhaseStart {
repo: r.component.clone(),
phase: r.op.clone(),
},
(_, p) if p == phase::END => LiveEvent::PhaseEnd {
repo: r.component.clone(),
phase: r.op.clone(),
ok: r.status == status::OK,
duration_ms: r.elapsed_ms.unwrap_or(0).max(0) as u64,
},
_ => continue, // skip phases (skip/etc) the pane doesn't render
};
out.push(ev);
}
(out, Some(run_id))
}
/// The reduced run header the pane shows + `state_json` exposes.
struct RunSummary {
run_id: String,
workspace: String,
current_repo: String,
current_phase: String,
current_binary: String,
total_pass: u32,
total_fail: u32,
run_ok: Option<bool>,
}
impl RunSummary {
fn from_events(events: &[LiveEvent]) -> Self {
let mut s = RunSummary {
run_id: "—".into(),
workspace: "—".into(),
current_repo: "—".into(),
current_phase: "—".into(),
current_binary: "—".into(),
total_pass: 0,
total_fail: 0,
run_ok: None,
};
for ev in events {
match ev {
LiveEvent::RunStart { run_id, workspace } => {
s.run_id = run_id.clone();
s.workspace = workspace.clone();
s.run_ok = None;
s.total_pass = 0;
s.total_fail = 0;
}
LiveEvent::RepoStart { repo, .. } => s.current_repo = repo.clone(),
LiveEvent::PhaseStart { repo, phase } => {
s.current_repo = repo.clone();
s.current_phase = phase.clone();
}
LiveEvent::PhaseEnd { repo, phase, .. } => {
s.current_repo = repo.clone();
s.current_phase = phase.clone();
}
LiveEvent::BinaryStart { binary, .. } => s.current_binary = binary.clone(),
LiveEvent::BinaryDone { passed, failed, .. } => {
s.total_pass += passed;
s.total_fail += failed;
}
LiveEvent::TestPass { .. } => s.total_pass += 1,
LiveEvent::TestFail { .. } => s.total_fail += 1,
LiveEvent::RunEnd { ok, .. } => s.run_ok = Some(*ok),
_ => {}
}
}
s
}
}
/// One stable text line per event — what the pane's log row renders, exposed in
/// `state_json["lines"]` so the headless matrix asserts the rendered content.
fn event_line(ev: &LiveEvent) -> String {
match ev {
LiveEvent::RunStart { run_id, workspace } => format!("▶ run {run_id} ({workspace})"),
LiveEvent::RepoStart { repo, sha } => format!("▶ {repo} sha={}", short_sha(sha)),
LiveEvent::PhaseStart { repo, phase } => format!("… {repo} {phase} start"),
LiveEvent::PhaseEnd {
repo,
phase,
ok,
duration_ms,
} => format!(
"{} {repo} {phase} ({:.1}s)",
if *ok { "✓" } else { "✗" },
*duration_ms as f32 / 1000.0,
),
LiveEvent::BinaryStart { repo, binary } => format!("📂 {repo} {}", short_bin(binary)),
LiveEvent::TestPass { name, .. } => format!("✓ {name}"),
LiveEvent::TestFail { name, .. } => format!("✗ {name}"),
LiveEvent::BinaryDone { passed, failed, .. } => {
format!("≡ {passed} passed, {failed} failed")
}
LiveEvent::RepoEnd { repo, ok } => {
format!("{} {repo} done", if *ok { "✓" } else { "✗" })
}
LiveEvent::RunEnd { ok, .. } => (if *ok {
"✓ run complete"
} else {
"✗ run failed"
})
.to_string(),
}
}
fn render_event_row(ui: &mut egui::Ui, ev: &LiveEvent, theme: &Theme) {
match ev {
LiveEvent::BinaryStart { repo, binary } => {
ui.horizontal(|ui| {
ui.colored_label(theme.accent, "📂");
ui.monospace(format!("{repo} {}", short_bin(binary)));
});
}
LiveEvent::PhaseStart { repo, phase } => {
ui.horizontal(|ui| {
ui.colored_label(theme.accent, "…");
ui.monospace(format!("{repo} {phase} start"));
});
}
LiveEvent::TestPass { name, .. } => {
ui.horizontal(|ui| {
ui.colored_label(GREEN, "✓");
ui.label(name);
});
}
LiveEvent::TestFail { name, .. } => {
ui.horizontal(|ui| {
ui.colored_label(RED, "✗");
ui.colored_label(RED, name);
});
}
LiveEvent::BinaryDone { passed, failed, .. } => {
ui.horizontal(|ui| {
ui.colored_label(theme.text_dim, "≡");
ui.label(format!("{passed} passed, {failed} failed"));
});
}
LiveEvent::PhaseEnd {
repo,
phase,
ok,
duration_ms,
} => {
let col = if *ok { GREEN } else { RED };
ui.colored_label(
col,
format!(
"{} {repo} {phase} ({:.1}s)",
if *ok { "✓" } else { "✗" },
*duration_ms as f32 / 1000.0,
),
);
}
LiveEvent::RepoStart { repo, sha } => {
ui.colored_label(theme.text, format!("▶ {repo} sha={}", short_sha(sha)));
}
LiveEvent::RunStart { run_id, workspace } => {
ui.colored_label(theme.text, format!("▶ run {run_id} ({workspace})"));
}
LiveEvent::RunEnd { ok, .. } => {
let col = if *ok { GREEN } else { RED };
ui.colored_label(
col,
if *ok {
"✓ run complete"
} else {
"✗ run failed"
},
);
}
_ => {}
}
}
fn short_bin(s: &str) -> String {
// cargo prints `Running tests/foo.rs (target/debug/deps/foo-abcd)`
// — keep the unittests / tests/<name> half, drop the deps path.
s.split_whitespace().next().unwrap_or(s).to_string()
}
fn short_sha(s: &str) -> String {
// Char-boundary safe: SHAs are ASCII, but a malformed/placeholder ref could
// carry multibyte UTF-8 and `&s[..12]` would panic mid-codepoint.
if s.chars().count() > 12 {
s.chars().take(12).collect()
} else {
s.to_string()
}
}
/// Remote counterpart to [`tail_loop`]: stream the live run from a
/// `nornir-server` over `Release.Progress`. Each event is pushed into the same
/// `Shared` state the pane renders. The server closes the stream after `RunEnd`;
/// we then reconnect (after a short backoff) so the pane keeps watching for the
/// next release run — agents hammering components show up live as they happen.
fn remote_loop(endpoint: String, token: String, workspace: String, shared: Arc<Mutex<Shared>>) {
loop {
{
let mut s = shared.lock().unwrap();
s.events.clear();
s.err = None;
s.eof_run_ended = false;
s.file_path = None;
}
let sink = Arc::clone(&shared);
let res = super::remote::stream_progress(&endpoint, &token, &workspace, move |ev| {
let mut s = sink.lock().unwrap();
if let LiveEvent::RunEnd { .. } = &ev {
s.eof_run_ended = true;
}
s.events.push(ev);
});
if let Err(e) = res {
shared.lock().unwrap().err = Some(format!("stream from {endpoint}: {e:#}"));
}
// Stream ended (RunEnd) or errored — wait, then reconnect for the next run.
thread::sleep(Duration::from_secs(2));
}
}
fn tail_loop(log_dir: PathBuf, shared: Arc<Mutex<Shared>>) {
let mut current: Option<(PathBuf, BufReader<File>)> = None;
loop {
// Re-scan for the newest events file every cycle so a fresh
// run (new timestamp suffix) supersedes the previous one.
if let Some(newest) = newest_events_file(&log_dir) {
let pick = match ¤t {
Some((p, _)) if *p == newest => None,
_ => Some(newest),
};
if let Some(path) = pick {
match File::open(&path) {
Ok(mut f) => {
let _ = f.seek(SeekFrom::Start(0));
let mut s = shared.lock().unwrap();
// A live `/tmp` run supersedes the warehouse baseline: it
// is the same run at sub-second latency. (The warehouse
// hydrate already proved the run is durable; the tail just
// makes it move faster.)
s.events.clear();
s.err = None;
s.eof_run_ended = false;
s.file_path = Some(path.clone());
drop(s);
current = Some((path, BufReader::new(f)));
}
Err(e) => {
shared.lock().unwrap().err = Some(format!("open failed: {e}"));
}
}
}
}
if let Some((_, reader)) = current.as_mut() {
let mut line = String::new();
loop {
line.clear();
match reader.read_line(&mut line) {
Ok(0) => break, // EOF — wait, then retry / re-scan
Ok(_) => {
let trimmed = line.trim_end_matches(['\r', '\n']);
if trimmed.is_empty() {
continue;
}
match serde_json::from_str::<LiveEvent>(trimmed) {
Ok(ev) => {
let mut s = shared.lock().unwrap();
if let LiveEvent::RunEnd { .. } = &ev {
s.eof_run_ended = true;
}
s.events.push(ev);
}
Err(_) => {
// Forward-compat: ignore unknown variants.
}
}
}
Err(e) => {
shared.lock().unwrap().err = Some(format!("read failed: {e}"));
break;
}
}
}
}
thread::sleep(Duration::from_millis(250));
}
}
fn newest_events_file(log_dir: &Path) -> Option<PathBuf> {
let rd = std::fs::read_dir(log_dir).ok()?;
let mut best: Option<(std::time::SystemTime, PathBuf)> = None;
for entry in rd.flatten() {
let p = entry.path();
let ok_name = p
.file_name()
.and_then(|s| s.to_str())
.map(|s| s.starts_with("release-run-") && s.ends_with(".events.ndjson"))
.unwrap_or(false);
if !ok_name {
continue;
}
if let Ok(m) = entry.metadata() {
if let Ok(t) = m.modified() {
if best.as_ref().map_or(true, |(bt, _)| t > *bt) {
best = Some((t, p));
}
}
}
}
best.map(|(_, p)| p)
}
#[cfg(test)]
mod tests {
use super::*;
use crate::warehouse::release_events::{phase, status};
fn row(
run: &str,
seq: i64,
comp: &str,
op: &str,
ph: &str,
st: &str,
ms: Option<i64>,
detail: &str,
) -> ReleaseEventRow {
ReleaseEventRow {
run_id: run.into(),
seq,
ts_micros: seq * 1_000,
component: comp.into(),
repo: comp.into(),
op: op.into(),
phase: ph.into(),
status: st.into(),
detail: detail.into(),
depends_on: None,
elapsed_ms: ms,
}
}
#[test]
fn release_rows_become_live_events_for_newest_run() {
// Two runs; the newest (higher ts) must be the one hydrated.
let rows = vec![
row(
"old",
0,
"old",
"run",
phase::START,
status::RUNNING,
None,
"wsA",
),
row(
"run-1",
0,
"run-1",
"run",
phase::START,
status::RUNNING,
None,
"holger",
),
row(
"run-1",
2,
"znippy",
"test",
phase::END,
status::OK,
Some(1200),
"3 passed",
),
row(
"run-1",
1,
"znippy",
"test",
phase::START,
status::RUNNING,
None,
"",
),
row("run-1", 3, "run-1", "run", phase::END, status::OK, None, ""),
];
let (events, run_id) = events_from_release_rows(rows);
assert_eq!(run_id.as_deref(), Some("run-1"), "newest run hydrated");
// The run header reduces from the events: run id, workspace, ok.
let s = RunSummary::from_events(&events);
assert_eq!(s.run_id, "run-1");
assert_eq!(s.workspace, "holger", "run/start detail → workspace label");
assert_eq!(s.run_ok, Some(true), "run/end ok → done");
assert_eq!(s.current_repo, "znippy");
// The rendered lines carry the real op boundaries (durable → live).
let lines: Vec<String> = events.iter().map(event_line).collect();
assert!(
lines.iter().any(|l| l == "… znippy test start"),
"got {lines:?}"
);
assert!(
lines.iter().any(|l| l == "✓ znippy test (1.2s)"),
"got {lines:?}"
);
assert!(lines.iter().any(|l| l == "✓ run complete"), "got {lines:?}");
}
#[test]
fn inject_hydrates_state_json_without_warehouse() {
let mut live = LiveRunState::new(PathBuf::new(), PathBuf::new());
live.inject_for_test(vec![
row(
"r",
0,
"r",
"run",
phase::START,
status::RUNNING,
None,
"ws",
),
row(
"r",
1,
"znippy",
"test",
phase::END,
status::OK,
Some(500),
"",
),
]);
let v = live.state_json();
assert_eq!(v["hydrated_from_warehouse"], true);
assert_eq!(v["hydrated_run_id"], "r");
assert_eq!(v["run_id"], "r");
assert_eq!(v["workspace"], "ws");
let lines: Vec<&str> = v["lines"]
.as_array()
.unwrap()
.iter()
.filter_map(|l| l.as_str())
.collect();
assert!(
lines.iter().any(|l| *l == "✓ znippy test (0.5s)"),
"got {lines:?}"
);
}
}