nornir 0.5.4

//! 📡 **Build-info LIVE-TAIL** — the slice-5 capstone (design
//! `.nornir/build-info-design.md` §f5, the "Deferred → tail" note).
//!
//! Slices 1–5 built the whole build-info arc: the §f3 Arrow schema + emitter
//! ([`nornir_build_thing::build_info`]), the native warehouse landing + at-rest
//! seal, the jera pipeline-DAG sub-jobs, and the **Flight-backed sink + consumer**
//! ([`crate::build::flight_sink`]). What was left open was the *viz tail*: a pane
//! that subscribes to the build-info Flight stream **per sub-job `step_id`** and
//! shows the rows arriving — each step's **status transitions** (from the `step`
//! rows) and its **log/metric tail** (§f5: "each step's log/metric stream is the
//! f3 feed filtered by `step_id`").
//!
//! # The model is pure, the seam is [`crate::build::flight_sink::subscribe`]
//!
//! [`BuildInfoTail`] is a **pure fold** over [`BuildInfoRow`]s — it never knows
//! *how* the rows arrived. That keeps it:
//!
//! * **robot-testable** — a test feeds rows in batches ([`BuildInfoTail::ingest`])
//!   and reads the growing tail + per-step status back out of [`state_json`],
//!   exactly the inject-and-assert shape the rest of the viz uses (no live server
//!   needed to prove the pane folds the feed).
//! * **transport-agnostic** — the live path ([`BuildInfoTail::pull`], gated behind
//!   `build-info-flight`) calls the existing `subscribe(endpoint, token, key)`
//!   consumer seam by ticket (`key` = [`route_key`](crate::build::flight_sink::route_key)
//!   per sub-job `step_id`) and ingests the decoded rows through the **same**
//!   `ingest`. Live wire and injected rows fold identically.
//!
//! # Reuse (LAW #5)
//!
//! The pane paints with the shared facett palette ([`Theme`]) and the vendored
//! facett widgets ([`facett_ui::status_chip`], zebra rows) — the same themeable,
//! robot-testable material every other viz pane uses. Nothing here is re-rolled.
//!
//! # Continuous auto-refresh (the timer, closing the deferral)
//!
//! The pane now **auto-refreshes** while it is open, mirroring
//! [`super::bench_live`]'s periodic warehouse re-read. A background **poller**
//! ([`BuildInfoTail::start_poller`], gated `build-info-flight`) re-[`pull`]s every
//! tracked sub-job ticket (`route_key(build_id, step_id)`) on a fixed
//! [`POLL_EVERY`] cadence on its own OS-thread runtime (the same pattern
//! [`crate::build::flight_sink::FlightBuildInfoSink`] uses, so a `pull` never
//! needs an ambient runtime and never blocks the egui frame). Decoded rows are
//! handed back over a plain `std` channel (the **inbox**); [`draw`](BuildInfoTail::draw)
//! drains the inbox through the SAME synchronous [`ingest`](BuildInfoTail::ingest)
//! every frame and calls `request_repaint_after` so rows that land mid-build appear
//! without any user input. The poller forwards only rows *past* the per-key
//! high-water mark, so re-pulling the full stream never double-folds (L2: the
//! append-only `ingest` is unchanged). The set of live tickets is tracked
//! ([`subscribe_ticket`](BuildInfoTail::subscribe_ticket)) and **auto-grown** as
//! the fold discovers each sub-job — both are exposed in [`state_json`] so the
//! matrix asserts the ticks advance.

use std::collections::BTreeSet;
use std::sync::mpsc::{Receiver, Sender};
use std::sync::{Arc, Mutex};
use std::time::Duration;

use eframe::egui::{self, RichText, ScrollArea};

use nornir_build_thing::build_info::BuildInfoRow;

use super::facett_theme::{Theme, AMBER, GREEN, RED};
use super::facett_ui;

/// How often (while the pane is open) the background poller re-`pull`s each live
/// sub-job ticket so build-info rows that land mid-build show up. Mirrors
/// [`super::bench_live`]'s warehouse re-read cadence.
pub const POLL_EVERY: Duration = Duration::from_millis(1500);

/// One tailed feed line — a `log` / `metric` / `step` row rendered in arrival
/// order. Exposed verbatim in [`BuildInfoTail::state_json`] so the headless
/// matrix asserts the tail content, not pixels.
#[derive(Debug, Clone, PartialEq)]
pub struct TailLine {
    /// The §f3 `kind`: `"log"` | `"metric"` | `"step"`.
    pub kind: String,
    /// The §f3 `level` (`info` / `error` for logs; the status for `step` rows).
    pub level: String,
    /// The §f3 `name` column (log source / metric name / step label).
    pub name: String,
    /// The §f3 `value` column (the log message / metric value / step variant).
    pub value: String,
    /// The §f3 `ts_micros` timestamp.
    pub ts_micros: i64,
}

impl TailLine {
    fn from_row(r: &BuildInfoRow) -> Self {
        Self {
            kind: r.kind.clone(),
            level: r.level.clone(),
            name: r.name.clone(),
            value: r.value.clone(),
            ts_micros: r.ts_micros,
        }
    }
}

/// The per-sub-job (`step_id`) state folded from the build-info feed: the latest
/// **status** (the most recent `step` row's status, carried in the §f3 `level`
/// column — `queued` / `running` / `done` / `failed`), the ordered list of status
/// **transitions**, and the ordered **log/metric/step tail**.
#[derive(Debug, Clone, PartialEq)]
pub struct StepTail {
    /// The §f3 `step_id` — the sub-job ticket this feed is filtered by.
    pub step_id: String,
    /// The current status (last `step` row's `level`); empty until the first
    /// `step` row arrives.
    pub status: String,
    /// The ordered status transitions seen so far (each distinct new status
    /// appended once), e.g. `["running", "done"]`.
    pub status_history: Vec<String>,
    /// The ordered `log` / `metric` / `step` tail for this sub-job — grows as
    /// rows arrive.
    pub tail: Vec<TailLine>,
}

impl StepTail {
    fn new(step_id: &str) -> Self {
        Self {
            step_id: step_id.to_string(),
            status: String::new(),
            status_history: Vec::new(),
            tail: Vec::new(),
        }
    }
}

/// The **build-info live-tail** model + pane. Folds the §f3 build-info feed into
/// per-sub-job status + tail (`ingest`), renders it themed (`draw`), and exposes
/// everything as data (`state_json`). The live source is the Flight `subscribe`
/// consumer seam (`pull`, behind `build-info-flight`).
pub struct BuildInfoTail {
    /// The root `build_id` (from the first row carrying one); the §f3
    /// build/pipeline this tail is for.
    build_id: Option<String>,
    /// Per-sub-job tails, in first-seen `step_id` order.
    steps: Vec<StepTail>,
    /// Count of manifest rows folded (`provenance` / `dep` / `artifact`, empty
    /// `step_id`) — reported but not tailed (they are the build's static record,
    /// not the live feed).
    manifest_rows: usize,
    /// The Flight endpoint the live path subscribes to (display + `state_json`).
    endpoint: Option<String>,
    /// Last error from a `pull` (never fatal — the pane keeps its folded state).
    error: Option<String>,
    /// The active facett palette (C8) — every colour routes through it so a
    /// palette switch re-skins the tail with the rest of the viz.
    theme: Theme,
    /// **Inbox** — decoded [`BuildInfoRow`] batches delivered by the background
    /// poller (or a test's simulated interval pull), drained into [`ingest`](Self::ingest)
    /// each [`draw`](Self::draw). A plain `std` channel, so the synchronous
    /// injectable fold path is unchanged (L2).
    inbox_tx: Sender<Vec<BuildInfoRow>>,
    inbox_rx: Receiver<Vec<BuildInfoRow>>,
    /// The set of live sub-job tickets `(build_id, step_id)` the poller re-`pull`s.
    /// Auto-grown as the fold discovers sub-jobs (and drivable via
    /// [`subscribe_ticket`](Self::subscribe_ticket)); shared with the poller thread.
    tickets: Arc<Mutex<BTreeSet<(String, String)>>>,
    /// Count of poll batches drained + folded — a "tick" per delivered batch, the
    /// number that ADVANCES as the auto-refresh timer re-pulls. Exposed in
    /// [`state_json`](Self::state_json).
    poll_count: usize,
    /// The running poller (its stop flag), if the auto-refresh timer is live.
    #[cfg(feature = "build-info-flight")]
    poller: Option<Poller>,
}

/// The background auto-refresh poller's handle: a stop flag the [`Drop`] sets so
/// the thread winds down its runtime on its own thread (never in the caller's).
#[cfg(feature = "build-info-flight")]
struct Poller {
    stop: Arc<std::sync::atomic::AtomicBool>,
}

impl Default for BuildInfoTail {
    fn default() -> Self {
        let (inbox_tx, inbox_rx) = std::sync::mpsc::channel();
        Self {
            build_id: None,
            steps: Vec::new(),
            manifest_rows: 0,
            endpoint: None,
            error: None,
            theme: Theme::default(),
            inbox_tx,
            inbox_rx,
            tickets: Arc::new(Mutex::new(BTreeSet::new())),
            poll_count: 0,
            #[cfg(feature = "build-info-flight")]
            poller: None,
        }
    }
}

#[cfg(feature = "build-info-flight")]
impl Drop for BuildInfoTail {
    fn drop(&mut self) {
        if let Some(p) = &self.poller {
            p.stop.store(true, std::sync::atomic::Ordering::Relaxed);
        }
    }
}

impl BuildInfoTail {
    /// A fresh, empty tail.
    pub fn new() -> Self {
        Self::default()
    }

    /// Point the live path at a Flight `endpoint` (recorded for display /
    /// `state_json`; the actual `subscribe` is [`pull`](Self::pull)).
    pub fn set_endpoint(&mut self, endpoint: impl Into<String>) {
        self.endpoint = Some(endpoint.into());
    }

    /// Re-skin this pane with a facett palette (broadcast from the picker / owning
    /// tab's `set_palette`).
    pub fn set_palette(&mut self, t: Theme) {
        self.theme = t;
    }

    /// Fold a batch of §f3 [`BuildInfoRow`]s into the per-sub-job tails. The
    /// heart of the pane: idempotent-append semantics — call it repeatedly as
    /// batches arrive (from `subscribe`, or injected in a test) and the tail
    /// **grows** while each step's **status** tracks its latest `step` row.
    ///
    /// * `kind == "step"` → the row's `level` is the sub-job status; a *new*
    ///   status is recorded as a transition. The step row is also appended to the
    ///   tail so a status change is visible in the stream.
    /// * `kind == "log" | "metric"` → appended to that sub-job's tail.
    /// * manifest rows (`provenance` / `dep` / `artifact`, empty `step_id`) → not
    ///   tailed (they are the build's static record); only counted.
    pub fn ingest(&mut self, rows: &[BuildInfoRow]) {
        for r in rows {
            if self.build_id.is_none() && !r.build_id.is_empty() {
                self.build_id = Some(r.build_id.clone());
            }
            // Auto-grow the live subscription set: each row's `(build_id, step_id)`
            // is a ticket the poller re-`pull`s (manifest rows → the root ticket,
            // empty `step_id`; feed rows → the per-sub-job ticket).
            self.register_ticket(&r.build_id, &r.step_id);
            match r.kind.as_str() {
                "step" => {
                    let step = self.step_mut(&r.step_id);
                    // The §f3 `step` row carries the sub-job status in `level`.
                    if !r.level.is_empty() && step.status != r.level {
                        step.status = r.level.clone();
                        step.status_history.push(r.level.clone());
                    }
                    step.tail.push(TailLine::from_row(r));
                }
                "log" | "metric" => {
                    let step = self.step_mut(&r.step_id);
                    step.tail.push(TailLine::from_row(r));
                }
                // provenance / dep / artifact — the static manifest, not the feed.
                _ => self.manifest_rows += 1,
            }
        }
        self.emit_trace();
    }

    /// **The live seam** (§f5): subscribe to one sub-job's build-info stream by
    /// ticket `key` (from [`route_key`](crate::build::flight_sink::route_key) —
    /// `build_id/step_id`) over the Flight `do_get` consumer, then fold the
    /// decoded rows through [`ingest`](Self::ingest). Returns the number of rows
    /// tailed. Best-effort: a transport/auth error is recorded (never a panic),
    /// and the already-folded state is kept.
    ///
    /// Gated behind `build-info-flight` (the feature that links the Flight wire);
    /// the default viz build keeps the injectable [`ingest`](Self::ingest) path
    /// and pays no Flight/second-arrow tax.
    #[cfg(feature = "build-info-flight")]
    pub async fn pull(
        &mut self,
        endpoint: &str,
        token: Option<&str>,
        key: &str,
    ) -> usize {
        self.endpoint = Some(endpoint.to_string());
        match crate::build::flight_sink::subscribe(endpoint, token, key).await {
            Ok(rows) => {
                let n = rows.len();
                self.error = None;
                self.ingest(&rows);
                n
            }
            Err(e) => {
                self.error = Some(format!("{e:#}"));
                0
            }
        }
    }

    /// Test-only alias for [`ingest`](Self::ingest) — mirrors the other panes'
    /// `inject_for_test` so the inject-and-assert harness reads the tail back out
    /// of `state_json` (the subscribe seam is itself already tested in
    /// `flight_sink`, so this drives the fold on top of it).
    #[doc(hidden)]
    pub fn inject_for_test(&mut self, rows: Vec<BuildInfoRow>) {
        self.ingest(&rows);
    }

    /// Register one sub-job ticket `(build_id, step_id)` to be (re-)subscribed by
    /// the auto-refresh poller. Idempotent (a `BTreeSet`). Called automatically as
    /// the fold discovers sub-jobs; also `pub` so the owner can seed a ticket
    /// before any row has arrived.
    pub fn subscribe_ticket(&mut self, build_id: &str, step_id: &str) {
        self.register_ticket(build_id, step_id);
    }

    fn register_ticket(&self, build_id: &str, step_id: &str) {
        if build_id.is_empty() {
            return;
        }
        if let Ok(mut t) = self.tickets.lock() {
            t.insert((build_id.to_string(), step_id.to_string()));
        }
    }

    /// The live tickets currently tracked, as `(build_id, step_id)` — the set the
    /// poller re-`pull`s. Ordered (a `BTreeSet`), so `state_json` is stable.
    fn tickets_snapshot(&self) -> Vec<(String, String)> {
        self.tickets
            .lock()
            .map(|t| t.iter().cloned().collect())
            .unwrap_or_default()
    }

    /// Drain every batch the poller (or a test) has delivered to the inbox and
    /// fold it through the SAME [`ingest`](Self::ingest). Each non-empty batch
    /// ADVANCES [`poll_count`](Self::poll_count) — the auto-refresh "tick". Called
    /// every [`draw`](Self::draw); cheap when the inbox is empty.
    fn drain_inbox(&mut self) {
        while let Ok(rows) = self.inbox_rx.try_recv() {
            if !rows.is_empty() {
                self.ingest(&rows);
                self.poll_count += 1;
            }
        }
    }

    /// Test seam: the inbox sender, so a test can simulate the poller delivering a
    /// batch on an interval (each `send` == one interval's freshly-pulled rows)
    /// WITHOUT a live Flight server, then [`pump`](Self::pump) / draw to fold it.
    #[doc(hidden)]
    pub fn poll_sender_for_test(&self) -> Sender<Vec<BuildInfoRow>> {
        self.inbox_tx.clone()
    }

    /// Drain the inbox once (what [`draw`](Self::draw) does each frame), exposed so
    /// a headless test can advance the fold after simulating an interval pull.
    #[doc(hidden)]
    pub fn pump(&mut self) {
        self.drain_inbox();
    }

    /// **Start the auto-refresh timer** (gated `build-info-flight`): spawn a
    /// background poller that, on its own OS-thread tokio runtime, re-`pull`s every
    /// tracked sub-job ticket every `interval` and hands the freshly-arrived rows
    /// back to the inbox (drained by [`draw`](Self::draw)). Idempotent — a second
    /// call is a no-op while a poller is live. The runtime lives + dies on the
    /// poller thread (mirrors [`crate::build::flight_sink::FlightBuildInfoSink`]),
    /// so `pull` never blocks the egui frame nor needs an ambient runtime.
    #[cfg(feature = "build-info-flight")]
    pub fn start_poller(
        &mut self,
        endpoint: impl Into<String>,
        token: Option<String>,
        interval: Duration,
    ) {
        use std::sync::atomic::{AtomicBool, Ordering};
        let endpoint = endpoint.into();
        self.endpoint = Some(endpoint.clone());
        if self.poller.is_some() {
            return;
        }
        let tx = self.inbox_tx.clone();
        let tickets = self.tickets.clone();
        let stop = Arc::new(AtomicBool::new(false));
        let stop_thread = stop.clone();
        let _ = std::thread::Builder::new()
            .name("build-info-live-poll".into())
            .spawn(move || {
                let rt = match tokio::runtime::Builder::new_current_thread()
                    .enable_all()
                    .build()
                {
                    Ok(rt) => rt,
                    Err(_) => return,
                };
                rt.block_on(async move {
                    // Per-key high-water mark: `subscribe` returns the FULL stream
                    // for a key each pull, so forward only rows past what we've
                    // already sent — the shared append-only `ingest` never
                    // double-folds a re-pull (L2).
                    let mut seen: std::collections::HashMap<String, usize> =
                        std::collections::HashMap::new();
                    while !stop_thread.load(Ordering::Relaxed) {
                        let keys = tickets
                            .lock()
                            .map(|t| t.iter().cloned().collect::<Vec<(String, String)>>())
                            .unwrap_or_default();
                        for (build_id, step_id) in keys {
                            let key = crate::build::flight_sink::route_key(&build_id, &step_id);
                            if let Ok(rows) = crate::build::flight_sink::subscribe(
                                &endpoint,
                                token.as_deref(),
                                &key,
                            )
                            .await
                            {
                                let already = seen.get(&key).copied().unwrap_or(0);
                                if rows.len() > already {
                                    let fresh: Vec<BuildInfoRow> = rows[already..].to_vec();
                                    seen.insert(key.clone(), rows.len());
                                    if tx.send(fresh).is_err() {
                                        return; // pane dropped — end the poller.
                                    }
                                }
                            }
                        }
                        tokio::time::sleep(interval).await;
                    }
                });
            });
        self.poller = Some(Poller { stop });
    }

    /// Get-or-create the tail bucket for `step_id`, preserving first-seen order.
    fn step_mut(&mut self, step_id: &str) -> &mut StepTail {
        if let Some(idx) = self.steps.iter().position(|s| s.step_id == step_id) {
            &mut self.steps[idx]
        } else {
            self.steps.push(StepTail::new(step_id));
            self.steps.last_mut().expect("just pushed")
        }
    }

    /// Total tail lines across all sub-jobs — the number that must GROW as
    /// batches arrive (the "tail growing" the robot test asserts).
    fn tail_total(&self) -> usize {
        self.steps.iter().map(|s| s.tail.len()).sum()
    }

    fn emit_trace(&self) {
        super::trace::emit_end(
            "build.info.live",
            &serde_json::json!({
                "build_id": self.build_id,
                "steps": self.steps.len(),
                "tail_total": self.tail_total(),
                "manifest_rows": self.manifest_rows,
            }),
        );
        #[cfg(feature = "testmatrix")]
        nornir_testmatrix::functional_status(
            "nornir/viz/build-info-live",
            "ingest_feed",
            true,
            &format!(
                "build_id={:?} steps={} tail={}",
                self.build_id,
                self.steps.len(),
                self.tail_total()
            ),
        );
    }

    /// The pane rendered as data (LAW #6) — folded into the app's
    /// `state_json["build_info_live"]`. The proof the pane subscribes the feed +
    /// folds it per sub-job, not pixels.
    pub fn state_json(&self) -> serde_json::Value {
        #[cfg(feature = "build-info-flight")]
        let polling = self.poller.is_some();
        #[cfg(not(feature = "build-info-flight"))]
        let polling = false;
        serde_json::json!({
            "palette": self.theme.name,
            "endpoint": self.endpoint,
            "error": self.error,
            "build_id": self.build_id,
            "step_count": self.steps.len(),
            "manifest_rows": self.manifest_rows,
            "tail_total": self.tail_total(),
            // Auto-refresh observability: whether the poller is live, how many
            // interval-pull batches have been folded (the "tick" that ADVANCES),
            // and the live sub-job tickets the poller re-subscribes.
            "polling": polling,
            "poll_count": self.poll_count,
            "live_tickets": self.tickets_snapshot().iter().map(|(b, s)| serde_json::json!({
                "build_id": b,
                "step_id": s,
            })).collect::<Vec<_>>(),
            "steps": self.steps.iter().map(|s| serde_json::json!({
                "step_id": s.step_id,
                "status": s.status,
                "status_history": s.status_history,
                "tail_len": s.tail.len(),
                "tail": s.tail.iter().map(|l| serde_json::json!({
                    "kind": l.kind,
                    "level": l.level,
                    "name": l.name,
                    "value": l.value,
                    "ts_micros": l.ts_micros,
                })).collect::<Vec<_>>(),
            })).collect::<Vec<_>>(),
        })
    }

    /// Paint the live-tail pane, themed. One collapsing section per sub-job with a
    /// status chip + the ordered tail; the whole thing scrolls.
    pub fn draw(&mut self, ui: &mut egui::Ui) {
        // Auto-refresh: fold whatever the background poller delivered since the
        // last frame, and keep the egui frame ticking so new rows appear without
        // any user input (mirrors `super::bench_live`'s `request_repaint_after`).
        self.drain_inbox();
        ui.ctx().request_repaint_after(POLL_EVERY);

        let theme = self.theme;
        ui.heading(RichText::new("📡 Build-info — live tail").color(theme.text));
        ui.horizontal(|ui| {
            if let Some(b) = &self.build_id {
                ui.label(RichText::new(format!("build {b}")).color(theme.text_dim));
            }
            if let Some(ep) = &self.endpoint {
                ui.label(RichText::new(format!("· {ep}")).color(theme.text_dim));
            }
            ui.label(
                RichText::new(format!(
                    "· {} sub-job(s) · {} tail line(s)",
                    self.steps.len(),
                    self.tail_total()
                ))
                .color(theme.text_dim),
            );
        });

        if let Some(err) = self.error.clone() {
            ui.colored_label(RED, format!("subscribe failed: {err}"));
        }

        if self.steps.is_empty() {
            ui.label(
                RichText::new("no build-info feed yet — waiting for a sub-job to emit")
                    .color(theme.text_dim),
            );
            return;
        }

        ui.separator();
        ScrollArea::vertical().auto_shrink([false, false]).show(ui, |ui| {
            for step in &self.steps {
                let (chip, color) = facett_ui::status_chip(&theme, &step.status);
                let title = if step.step_id.is_empty() {
                    "(root)".to_string()
                } else {
                    step.step_id.clone()
                };
                egui::CollapsingHeader::new(
                    RichText::new(format!("{chip}  {title}")).color(color),
                )
                .default_open(true)
                .show(ui, |ui| {
                    if !step.status_history.is_empty() {
                        ui.label(
                            RichText::new(format!(
                                "status: {}",
                                step.status_history.join(" → ")
                            ))
                            .color(theme.text_dim),
                        );
                    }
                    for line in &step.tail {
                        let lc = match line.level.as_str() {
                            "error" | "failed" => RED,
                            "running" | "warn" => AMBER,
                            "done" | "info" => GREEN,
                            _ => theme.text,
                        };
                        ui.label(
                            RichText::new(format!(
                                "[{}] {} = {}",
                                line.kind, line.name, line.value
                            ))
                            .color(lc)
                            .monospace(),
                        );
                    }
                });
            }
        });
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use nornir_build_thing::build_info::{BuildInfoKind, BuildInfoRow};

    /// A `step` row for `step_id` carrying `status` in the §f3 `level` column.
    fn step_row(build: &str, step: &str, ts: i64, status: &str) -> BuildInfoRow {
        BuildInfoRow::new(build, ts, BuildInfoKind::Step)
            .step(step)
            .level(status)
            .kv("cargo build", "full")
    }

    /// A `log` row for `step_id`.
    fn log_row(build: &str, step: &str, ts: i64, msg: &str) -> BuildInfoRow {
        BuildInfoRow::new(build, ts, BuildInfoKind::Log)
            .step(step)
            .level("info")
            .kv("cargo", msg)
    }

    /// RED-when-broken: feeding the feed in batches GROWS the tail and tracks
    /// each sub-job's status transitions — read back out of `state_json`.
    #[test]
    fn ingest_grows_tail_and_tracks_status() {
        let build = "b-live";
        let mut tail = BuildInfoTail::new();

        // Batch 1: the "compile" sub-job starts running and logs a line.
        tail.ingest(&[
            step_row(build, "compile", 1, "running"),
            log_row(build, "compile", 2, "Compiling holger v0.1"),
        ]);
        let v1 = tail.state_json();
        assert_eq!(v1["build_id"], "b-live");
        assert_eq!(v1["step_count"], 1);
        assert_eq!(v1["tail_total"], 2, "step + log tailed");
        let compile1 = &v1["steps"][0];
        assert_eq!(compile1["step_id"], "compile");
        assert_eq!(compile1["status"], "running", "first status");
        assert_eq!(compile1["tail_len"], 2);

        // Batch 2: compile finishes (a transition), a metric lands, and a second
        // sub-job "test" appears — the tail must GROW, not replace.
        tail.ingest(&[
            log_row(build, "compile", 3, "Finished release"),
            step_row(build, "compile", 4, "done"),
            step_row(build, "test", 5, "running"),
        ]);
        let v2 = tail.state_json();
        assert_eq!(v2["step_count"], 2, "second sub-job appeared");
        assert_eq!(
            v2["tail_total"], 5,
            "tail GREW to 5 across the two batches, not reset"
        );

        // compile: status transitioned running → done, both recorded in order.
        let compile2 = v2["steps"]
            .as_array()
            .unwrap()
            .iter()
            .find(|s| s["step_id"] == "compile")
            .unwrap();
        assert_eq!(compile2["status"], "done", "latest status");
        let hist: Vec<&str> = compile2["status_history"]
            .as_array()
            .unwrap()
            .iter()
            .map(|s| s.as_str().unwrap())
            .collect();
        assert_eq!(hist, vec!["running", "done"], "transitions in order");
        assert_eq!(compile2["tail_len"], 4, "compile tail grew to 4");

        // test: newly running.
        let test = v2["steps"]
            .as_array()
            .unwrap()
            .iter()
            .find(|s| s["step_id"] == "test")
            .unwrap();
        assert_eq!(test["status"], "running");
        assert_eq!(test["tail_len"], 1);
    }

    /// RED-when-broken (auto-refresh): a batch delivered to the inbox (what the
    /// background poller does on each interval `pull`) is folded through the SAME
    /// `ingest` when the pane drains — so `state_json` ADVANCES across simulated
    /// interval pulls (tail grows, `poll_count` ticks) without a re-inject, and
    /// the live-ticket set auto-grows to cover each discovered sub-job.
    #[test]
    fn auto_refresh_drains_inbox_and_advances_ticks() {
        let build = "b-auto";
        let mut tail = BuildInfoTail::new();
        let tx = tail.poll_sender_for_test();

        // Nothing pulled yet.
        let v0 = tail.state_json();
        assert_eq!(v0["tail_total"], 0);
        assert_eq!(v0["poll_count"], 0);
        assert_eq!(v0["live_tickets"].as_array().unwrap().len(), 0);

        // Interval-pull #1 lands a batch; drain (what `draw` does) folds it.
        tx.send(vec![
            step_row(build, "compile", 1, "running"),
            log_row(build, "compile", 2, "Compiling holger v0.1"),
        ])
        .unwrap();
        tail.pump();
        let v1 = tail.state_json();
        assert_eq!(v1["tail_total"], 2, "first interval pull folded");
        assert_eq!(v1["poll_count"], 1, "one tick");

        // Interval-pull #2 ADVANCES the tail (must GROW, not reset) + ticks again.
        tx.send(vec![
            step_row(build, "compile", 3, "done"),
            step_row(build, "test", 4, "running"),
        ])
        .unwrap();
        tail.pump();
        let v2 = tail.state_json();
        assert_eq!(v2["tail_total"], 4, "tail GREW after the second interval pull");
        assert_eq!(v2["poll_count"], 2, "ticks advanced per interval pull");

        // The live-ticket set auto-grew to cover both discovered sub-jobs.
        let tickets = v2["live_tickets"].as_array().unwrap();
        assert!(tickets.iter().any(|t| t["step_id"] == "compile"), "compile ticket");
        assert!(tickets.iter().any(|t| t["step_id"] == "test"), "test ticket");
        // compile transitioned running → done across the two pulls.
        let compile = v2["steps"]
            .as_array()
            .unwrap()
            .iter()
            .find(|s| s["step_id"] == "compile")
            .unwrap();
        assert_eq!(compile["status"], "done", "status advanced with the pulls");
    }

    /// An empty batch is not a tick — the poller only forwards freshly-pulled rows
    /// (rows past the per-key high-water mark), so a no-op interval must not
    /// advance `poll_count` nor the tail.
    #[test]
    fn empty_interval_pull_is_not_a_tick() {
        let mut tail = BuildInfoTail::new();
        let tx = tail.poll_sender_for_test();
        tx.send(vec![]).unwrap();
        tail.pump();
        let v = tail.state_json();
        assert_eq!(v["poll_count"], 0, "an empty pull is not a tick");
        assert_eq!(v["tail_total"], 0);
    }

    /// Manifest rows (empty step_id) are counted, not tailed — the feed pane is
    /// the live log/metric/step stream, not the static build record.
    #[test]
    fn manifest_rows_counted_not_tailed() {
        let mut tail = BuildInfoTail::new();
        tail.ingest(&[
            BuildInfoRow::new("b", 1, BuildInfoKind::Provenance).kv("repo", "holger"),
            BuildInfoRow::new("b", 2, BuildInfoKind::Dep).kv("feature", "full"),
            BuildInfoRow::new("b", 3, BuildInfoKind::Artifact).kv("target/x", "10"),
            log_row("b", "compile", 4, "line"),
        ]);
        let v = tail.state_json();
        assert_eq!(v["manifest_rows"], 3, "3 manifest rows counted");
        assert_eq!(v["step_count"], 1, "only the feed row made a sub-job");
        assert_eq!(v["tail_total"], 1, "manifest rows not in the tail");
    }
}