nornir 0.5.0

//! Iceberg writer/reader for the **clone/populate-outcome stream** (`clone_events`,
//! PLAN #6).
//!
//! When the server's poll loop (or a `Workspaces.Fetch` RPC, or the fat-CLI
//! `nornir workspace fetch`) clones+fetches a monitored workspace's git members,
//! per-member failures used to be *log-only* — an `eprintln!("…fetch error…")`
//! that a thin viz/CLI client could never see. It just saw missing data and no
//! reason. This module makes those outcomes **first-class, persisted, and readable
//! remotely**: one row per member attempt (clone/fetch + republish), `ok` or
//! `error`, carrying the error chain and the elapsed time.
//!
//! Write path: [`record_fetch_report`] turns a [`crate::monitor::FetchReport`] into
//! rows and appends them through the warehouse handle the *server already holds*
//! (redb is single-writer — the monitor itself can't open a second handle, so the
//! call sites that own a handle do the write). Best-effort + non-fatal: a logging
//! row failure must never abort a fetch/republish.
//!
//! Read path: [`query_clone_events`] scans the table, scopes by workspace, and
//! returns rows newest-first (Iceberg gives no scan order, so reads sort by
//! `ts_micros`). The `Viz.CloneEvents` RPC + `nornir workspace events` both read
//! through it.
//!
//! Schema: [`super::iceberg_schema::clone_events`].

use std::sync::Arc;

use anyhow::{anyhow, Result};
use arrow::array::{Array, Int64Array, RecordBatch, StringArray, TimestampMicrosecondArray};
use chrono::{DateTime, TimeZone, Utc};
use iceberg::Catalog;
use iceberg::arrow::schema_to_arrow_schema;

use super::iceberg::{IcebergWarehouse, TABLE_CLONE_EVENTS, append_batch};

// Column indices match `iceberg_schema::clone_events` field order.
const COL_TS_MICROS: usize = 0;
const COL_WORKSPACE: usize = 1;
const COL_MEMBER: usize = 2;
const COL_REMOTE: usize = 3;
const COL_OP: usize = 4;
const COL_STATUS: usize = 5;
const COL_DETAIL: usize = 6;
const COL_ELAPSED_MS: usize = 7;

/// Canonical op tags.
pub mod op {
    /// A member clone/fetch attempt (today both go through `clone_or_fetch`).
    pub const CLONE_FETCH: &str = "clone-fetch";
    /// The post-fetch warehouse republish for the whole workspace.
    pub const REPUBLISH: &str = "republish";
}

/// Canonical status tags.
pub mod status {
    pub const OK: &str = "ok";
    pub const ERROR: &str = "error";
}

/// One row of the `clone_events` table — a single member's clone/fetch (or the
/// workspace republish) outcome.
///
/// `serde` derives let the server ship the exact rows to a thin viz client over
/// the `Viz.CloneEvents` RPC (serialize → JSON → deserialize), so the remote 🧬
/// nornir pane + `nornir --server workspace events` render the identical populate
/// status the embedded path reads from the warehouse.
#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub struct CloneEventRow {
    /// Event time, microseconds since the Unix epoch (UTC).
    pub ts_micros: i64,
    /// The workspace this fetch belonged to.
    pub workspace: String,
    /// The member that was fetched (or `*` for a workspace-level `republish`).
    pub member: String,
    /// The git remote that was cloned/fetched (empty for a republish row).
    pub remote: String,
    /// `clone-fetch` | `republish` (see [`op`]).
    pub op: String,
    /// `ok` | `error` (see [`status`]).
    pub status: String,
    /// On error: the error chain (`{e:#}`). On ok: the resolved SHA (clone-fetch)
    /// or the snapshot id (republish). Stored as "" never null so SQL readers see
    /// no surprise nulls.
    pub detail: String,
    /// Wall-clock duration of the op, milliseconds.
    pub elapsed_ms: i64,
}

/// Append a batch of `clone_events` rows (one Iceberg snapshot). Mirrors
/// [`super::release_events::append_release_events`].
pub async fn append_clone_events(wh: &IcebergWarehouse, rows: &[CloneEventRow]) -> Result<()> {
    if rows.is_empty() {
        return Ok(());
    }
    // Blank/torn metadata JSON self-heals (drop+recreate the 0-row table) so the
    // populate-outcome append can't EOF-fail.
    let table = wh
        .load_table_for_append(
            TABLE_CLONE_EVENTS,
            super::iceberg_schema::clone_events()?,
            &[],
        )
        .await?;
    let arrow_schema = Arc::new(schema_to_arrow_schema(table.metadata().current_schema())?);

    let cols: Vec<Arc<dyn Array>> = vec![
        Arc::new(
            TimestampMicrosecondArray::from(rows.iter().map(|r| r.ts_micros).collect::<Vec<_>>())
                .with_timezone("+00:00"),
        ),
        Arc::new(StringArray::from(rows.iter().map(|r| r.workspace.clone()).collect::<Vec<_>>())),
        Arc::new(StringArray::from(rows.iter().map(|r| r.member.clone()).collect::<Vec<_>>())),
        Arc::new(StringArray::from(rows.iter().map(|r| r.remote.clone()).collect::<Vec<_>>())),
        Arc::new(StringArray::from(rows.iter().map(|r| r.op.clone()).collect::<Vec<_>>())),
        Arc::new(StringArray::from(rows.iter().map(|r| r.status.clone()).collect::<Vec<_>>())),
        Arc::new(StringArray::from(rows.iter().map(|r| r.detail.clone()).collect::<Vec<_>>())),
        Arc::new(Int64Array::from(rows.iter().map(|r| r.elapsed_ms).collect::<Vec<_>>())),
    ];
    let batch = RecordBatch::try_new(arrow_schema, cols)?;
    append_batch(wh.catalog(), table, batch).await?;
    Ok(())
}

/// Build `clone_events` rows from a [`crate::monitor::FetchReport`] — one per
/// fetched member outcome (ok or error). Pure (no I/O), so the writer + tests
/// share it. Stamped now (UTC micros).
pub fn rows_from_report(report: &crate::monitor::FetchReport) -> Vec<CloneEventRow> {
    let ts = Utc::now().timestamp_micros();
    report
        .outcomes
        .iter()
        .map(|o| CloneEventRow {
            ts_micros: ts,
            workspace: report.workspace.clone(),
            member: o.member.clone(),
            remote: o.remote.clone(),
            op: o.op.clone(),
            status: o.status.clone(),
            detail: o.detail.clone(),
            elapsed_ms: o.elapsed_ms,
        })
        .collect()
}

/// Persist a fetch report's per-member outcomes to `clone_events`. Best-effort +
/// **non-fatal**: a logging-row failure is logged (stderr) and swallowed so a
/// warehouse hiccup never aborts an otherwise-successful fetch/republish. Called
/// from every path that owns a warehouse handle (server sweep, `Workspaces.Fetch`,
/// the fat-CLI fetch).
pub fn record_fetch_report(wh: &IcebergWarehouse, report: &crate::monitor::FetchReport) {
    let rows = rows_from_report(report);
    if rows.is_empty() {
        return;
    }
    if let Err(e) = wh.block_on(append_clone_events(wh, &rows)) {
        eprintln!(
            "   ⚠ clone_events: dropped {} populate-outcome row(s) for `{}` (non-fatal): {e:#}",
            rows.len(),
            report.workspace
        );
    }
}

/// Keep only the rows worth persisting from a **periodic poll sweep**: errors
/// (so "why didn't this member populate" survives) and members that actually
/// changed this tick. The steady-state "ok, nothing changed" rows are dropped —
/// otherwise the 60s poll loop appends one row per member *every tick forever*,
/// growing the table unboundedly (the cause of the multi-MB `Viz.CloneEvents`
/// payload). User-initiated fetches still log every outcome via
/// [`record_fetch_report`]; only the unattended sweep is thinned.
pub fn rows_from_sweep(report: &crate::monitor::FetchReport) -> Vec<CloneEventRow> {
    rows_from_report(report)
        .into_iter()
        .filter(|r| r.status == status::ERROR || report.changed.iter().any(|c| c == &r.member))
        .collect()
}

/// Like [`record_fetch_report`] but only persists the sweep-worthy rows
/// ([`rows_from_sweep`]). Used by the server poll loop so an idle workspace
/// stops minting a no-op row per member every tick. Non-fatal.
pub fn record_sweep_report(wh: &IcebergWarehouse, report: &crate::monitor::FetchReport) {
    let rows = rows_from_sweep(report);
    if rows.is_empty() {
        return;
    }
    if let Err(e) = wh.block_on(append_clone_events(wh, &rows)) {
        eprintln!(
            "   ⚠ clone_events: dropped {} sweep-outcome row(s) for `{}` (non-fatal): {e:#}",
            rows.len(),
            report.workspace
        );
    }
}

/// Record a single workspace-level `republish` outcome (ok → snapshot id, or the
/// error chain). Non-fatal. `member` is `*` (a republish covers the whole
/// workspace, not one member).
pub fn record_republish(
    wh: &IcebergWarehouse,
    workspace: &str,
    status_: &str,
    detail: &str,
    elapsed_ms: i64,
) {
    let row = CloneEventRow {
        ts_micros: Utc::now().timestamp_micros(),
        workspace: workspace.to_string(),
        member: "*".into(),
        remote: String::new(),
        op: op::REPUBLISH.into(),
        status: status_.to_string(),
        detail: detail.to_string(),
        elapsed_ms,
    };
    if let Err(e) = wh.block_on(append_clone_events(wh, std::slice::from_ref(&row))) {
        eprintln!("   ⚠ clone_events: dropped republish row for `{workspace}` (non-fatal): {e:#}");
    }
}

/// Which clone events to read.
#[derive(Debug, Clone)]
pub enum CloneSelector {
    /// Every event whose `workspace` matches.
    Workspace(String),
    /// Everything in the table.
    All,
}

/// Read clone events, scoped by `sel`, returned **newest-first** (`ts_micros`
/// descending) so the recent populate status is at the top — Iceberg gives no scan
/// order, so the reader sorts.
pub async fn query_clone_events(
    wh: &IcebergWarehouse,
    sel: &CloneSelector,
) -> Result<Vec<CloneEventRow>> {
    // Blank/torn Iceberg metadata JSON → 0 rows, never a JSON-EOF crash (load+read).
    let batches: Vec<RecordBatch> =
        super::iceberg::load_and_read_all(wh, TABLE_CLONE_EVENTS).await?;

    let mut out: Vec<CloneEventRow> = Vec::new();
    for b in &batches {
        let ts = col_ts(b, COL_TS_MICROS)?;
        let workspace = col_str(b, COL_WORKSPACE)?;
        let member = col_str(b, COL_MEMBER)?;
        let remote = col_str(b, COL_REMOTE)?;
        let op_ = col_str(b, COL_OP)?;
        let st = col_str(b, COL_STATUS)?;
        let detail = col_str(b, COL_DETAIL)?;
        let elapsed = col_i64(b, COL_ELAPSED_MS)?;
        for i in 0..b.num_rows() {
            let row = CloneEventRow {
                ts_micros: ts.value(i),
                workspace: workspace.value(i).to_string(),
                member: member.value(i).to_string(),
                remote: remote.value(i).to_string(),
                op: op_.value(i).to_string(),
                status: st.value(i).to_string(),
                detail: detail.value(i).to_string(),
                elapsed_ms: elapsed.value(i),
            };
            let keep = match sel {
                CloneSelector::Workspace(w) => &row.workspace == w,
                CloneSelector::All => true,
            };
            if keep {
                out.push(row);
            }
        }
    }
    out.sort_by(|a, b| b.ts_micros.cmp(&a.ts_micros));
    Ok(out)
}

/// The rolled-up **populate verdict** for ONE workspace (PO1): the latest
/// clone/fetch outcome per member collapsed into a single workspace-level state.
/// `green` ⟺ every member's most-recent op is `ok`; `red` ⟺ at least one member's
/// most-recent op is `error`; `stale` ⟺ no clone_events recorded at all (never
/// populated). The first failing member + its error detail are surfaced so the
/// roster shows *which* member broke and *why* — the same "see what the user sees"
/// data the per-member pane renders, collapsed for the multi-workspace roster.
#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub struct PopulateVerdict {
    /// `"green"` (all latest ok) · `"red"` (≥1 latest error) · `"stale"` (no events).
    pub state: String,
    /// How many distinct members have a latest outcome.
    pub members: usize,
    /// How many members' latest outcome is `error`.
    pub failed: usize,
    /// The first failing member (alphabetical), or empty when none failed.
    pub failing_member: String,
    /// That member's latest error detail (the git error chain), or empty.
    pub last_error: String,
    /// RFC3339 of the newest event across the workspace, or empty when stale.
    pub last_synced: String,
}

/// Verdict states.
pub mod verdict {
    pub const GREEN: &str = "green";
    pub const RED: &str = "red";
    pub const STALE: &str = "stale";
}

/// Roll a workspace's `clone_events` rows (any order) up into a single
/// [`PopulateVerdict`] (PO1). Pure (no I/O) so the CLI roster, the viz roster, and
/// the server roll-up RPC share ONE verdict computation. For each member we keep
/// only its NEWEST event (`republish` rows — member `*` — are ignored: they're a
/// workspace-level op, not a member populate). An empty input is `stale`
/// (RAGNARÖK: never-populated ⇒ stale, not silently green).
pub fn workspace_populate_verdict(rows: &[CloneEventRow]) -> PopulateVerdict {
    use std::collections::BTreeMap;
    // member → its newest row (by ts_micros).
    let mut latest: BTreeMap<&str, &CloneEventRow> = BTreeMap::new();
    let mut newest_ts: i64 = i64::MIN;
    for r in rows {
        if r.member == "*" {
            // workspace-level republish — still track the freshness timestamp.
            newest_ts = newest_ts.max(r.ts_micros);
            continue;
        }
        newest_ts = newest_ts.max(r.ts_micros);
        latest
            .entry(r.member.as_str())
            .and_modify(|cur| {
                if r.ts_micros > cur.ts_micros {
                    *cur = r;
                }
            })
            .or_insert(r);
    }

    if latest.is_empty() && newest_ts == i64::MIN {
        return PopulateVerdict {
            state: verdict::STALE.into(),
            members: 0,
            failed: 0,
            failing_member: String::new(),
            last_error: String::new(),
            last_synced: String::new(),
        };
    }

    let failed: Vec<&&CloneEventRow> =
        latest.values().filter(|r| r.status == status::ERROR).collect();
    // First failing member (BTreeMap iterates members alphabetically).
    let (failing_member, last_error) = latest
        .values()
        .find(|r| r.status == status::ERROR)
        .map(|r| (r.member.clone(), r.detail.clone()))
        .unwrap_or_default();

    PopulateVerdict {
        state: if failed.is_empty() { verdict::GREEN } else { verdict::RED }.into(),
        members: latest.len(),
        failed: failed.len(),
        failing_member,
        last_error,
        last_synced: if newest_ts == i64::MIN { String::new() } else { ts_to_rfc3339(newest_ts) },
    }
}

/// Render a human-readable populate-status view: newest events first, each line
/// `mark op member — detail (elapsed)`. Backs `nornir workspace events`.
pub fn render_events(rows: &[CloneEventRow]) -> String {
    if rows.is_empty() {
        return "(no clone/populate events recorded)\n".to_string();
    }
    let mut out = String::new();
    for r in rows {
        let mark = match r.status.as_str() {
            status::OK => "✓",
            status::ERROR => "✗",
            _ => "·",
        };
        let when = ts_to_rfc3339(r.ts_micros);
        let detail = if r.detail.is_empty() {
            String::new()
        } else {
            format!(" — {}", r.detail)
        };
        out.push_str(&format!(
            "{mark} {when}  {op:<12} {member}{detail} ({elapsed}ms)\n",
            op = r.op,
            member = r.member,
            elapsed = r.elapsed_ms,
        ));
    }
    out
}

// ─── column helpers ─────────────────────────────────────────────────────

fn col_str<'a>(b: &'a RecordBatch, idx: usize) -> Result<&'a StringArray> {
    b.column(idx)
        .as_any()
        .downcast_ref::<StringArray>()
        .ok_or_else(|| anyhow!("clone_events col {idx} is not StringArray"))
}

fn col_i64<'a>(b: &'a RecordBatch, idx: usize) -> Result<&'a Int64Array> {
    b.column(idx)
        .as_any()
        .downcast_ref::<Int64Array>()
        .ok_or_else(|| anyhow!("clone_events col {idx} is not Int64Array"))
}

fn col_ts<'a>(b: &'a RecordBatch, idx: usize) -> Result<&'a TimestampMicrosecondArray> {
    b.column(idx)
        .as_any()
        .downcast_ref::<TimestampMicrosecondArray>()
        .ok_or_else(|| anyhow!("clone_events col {idx} is not TimestampMicrosecondArray"))
}

/// Format `ts_micros` as RFC3339 (UTC). Convenience for renderers/tests.
pub fn ts_to_rfc3339(ts_micros: i64) -> String {
    let dt: DateTime<Utc> = Utc.timestamp_micros(ts_micros).single().unwrap_or_else(Utc::now);
    dt.to_rfc3339()
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::monitor::{FetchReport, MemberOutcome};

    fn outcome(member: &str, status_: &str, detail: &str) -> MemberOutcome {
        MemberOutcome {
            member: member.into(),
            remote: format!("git@example.com:{member}.git"),
            op: op::CLONE_FETCH.into(),
            status: status_.into(),
            detail: detail.into(),
            elapsed_ms: 7,
        }
    }

    /// LAW (inject-assert): a periodic sweep over an idle workspace must NOT mint a
    /// row for steady-state "ok, unchanged" members — only errors and members that
    /// actually changed. This is what stops the unbounded `clone_events` growth that
    /// overflowed the `Viz.CloneEvents` gRPC message-size limit.
    #[test]
    fn sweep_thinning_keeps_only_errors_and_changed() {
        let report = FetchReport {
            workspace: "nordisk".into(),
            fetched: 3,
            changed: vec!["facett".into()],
            errors: vec![("korp".into(), "fatal: could not read".into())],
            outcomes: vec![
                outcome("nornir", status::OK, "abc123"),   // ok + unchanged → DROP
                outcome("facett", status::OK, "def456"),   // ok + changed   → KEEP
                outcome("korp", status::ERROR, "fatal: could not read"), // error → KEEP
            ],
        };

        let kept = rows_from_sweep(&report);
        let members: Vec<&str> = kept.iter().map(|r| r.member.as_str()).collect();
        assert_eq!(members, vec!["facett", "korp"], "only changed + error rows survive a sweep");
        // The no-op heartbeat row for the unchanged-ok member is gone.
        assert!(!members.contains(&"nornir"), "steady-state ok/unchanged row must be dropped");
        // A user-initiated fetch (record_fetch_report path) still logs everything.
        assert_eq!(rows_from_report(&report).len(), 3, "full report keeps all outcomes");
    }

    /// LAW 1 (inject-assert): feed a FetchReport with one OK member and one FAILED
    /// member → assert exactly two rows persist with the right
    /// (member, op, status, detail) and read back newest-first, scoped by
    /// workspace.
    #[test]
    fn fetch_report_failure_persists_and_reads_back() {
        let dir = tempfile::tempdir().unwrap();
        let wh = IcebergWarehouse::open(dir.path()).unwrap();

        let report = FetchReport {
            workspace: "nordisk".into(),
            fetched: 2,
            changed: vec!["facett".into()],
            errors: vec![(
                "korp".into(),
                "clone-fetch https://github.com/nordisk/korp: Couldn't obtain Username".into(),
            )],
            outcomes: vec![
                MemberOutcome {
                    member: "facett".into(),
                    remote: "git@github.com:nordisk/facett.git".into(),
                    op: op::CLONE_FETCH.into(),
                    status: status::OK.into(),
                    detail: "abc123".into(),
                    elapsed_ms: 42,
                },
                MemberOutcome {
                    member: "korp".into(),
                    remote: "https://github.com/nordisk/korp".into(),
                    op: op::CLONE_FETCH.into(),
                    status: status::ERROR.into(),
                    detail: "clone-fetch https://github.com/nordisk/korp: Couldn't obtain Username"
                        .into(),
                    elapsed_ms: 7,
                },
            ],
        };

        // Write through the (non-fatal) recorder — the same call the server uses.
        record_fetch_report(&wh, &report);

        // Read back, scoped to the workspace, newest-first.
        let rows = wh
            .block_on(query_clone_events(&wh, &CloneSelector::Workspace("nordisk".into())))
            .unwrap();
        assert_eq!(rows.len(), 2, "one row per fetched member");

        // The FAILED member is first-class: find it and assert the error detail
        // round-tripped exactly.
        let failed: Vec<_> = rows.iter().filter(|r| r.status == status::ERROR).collect();
        assert_eq!(failed.len(), 1);
        assert_eq!(failed[0].member, "korp");
        assert_eq!(failed[0].op, "clone-fetch");
        assert_eq!(failed[0].remote, "https://github.com/nordisk/korp");
        assert!(
            failed[0].detail.contains("Couldn't obtain Username"),
            "the error chain is readable, got: {}",
            failed[0].detail
        );
        assert_eq!(failed[0].elapsed_ms, 7);

        // The OK member round-trips with its SHA in `detail`.
        let ok: Vec<_> = rows.iter().filter(|r| r.status == status::OK).collect();
        assert_eq!(ok.len(), 1);
        assert_eq!(ok[0].member, "facett");
        assert_eq!(ok[0].detail, "abc123");

        // A different workspace's scope sees nothing.
        let other = wh
            .block_on(query_clone_events(&wh, &CloneSelector::Workspace("holger".into())))
            .unwrap();
        assert!(other.is_empty(), "workspace scope isolates events");

        // The human render names the failed member + its error.
        let txt = render_events(&rows);
        assert!(txt.contains("korp"));
        assert!(txt.contains("Couldn't obtain Username"));
        assert!(txt.contains('✗'), "failure marker present");

        // JSON round-trips for the RPC contract.
        let json = serde_json::to_string(&rows).unwrap();
        let back: Vec<CloneEventRow> = serde_json::from_str(&json).unwrap();
        assert_eq!(back, rows);
    }

    /// PF-cloneevents regression: an UNBOUNDED clone-event ledger (a long-lived
    /// nordisk that never sweep-thinned) overflows the OLD 4 MiB tonic encode
    /// cap — which is why the server now needs BOTH the 2 000-row truncation
    /// (`VIZ_CLONE_EVENTS_CAP`) AND the 64 MiB Viz-service cap (the belt for the
    /// other large, *un-truncated* viz payloads — Architecture, timeline). The
    /// 2 000-row bounded payload must fit 4 MiB (truncation works); the unbounded
    /// ledger must exceed 4 MiB (motivating the cap) yet still fit 64 MiB.
    /// Multi-value across the boundaries: empty / one / at-truncation / unbounded.
    #[test]
    fn clone_events_payload_sizes_motivate_both_truncation_and_64mib_cap() {
        const OLD_CAP: usize = 4 << 20; // tonic default — what the bug overflowed
        const NEW_CAP: usize = 64 << 20; // the Viz-service cap we set on the server
        const ROW_CAP: usize = 2_000; // VIZ_CLONE_EVENTS_CAP in nornir-server

        let mk = |n: usize| -> Vec<CloneEventRow> {
            (0..n)
                .map(|i| CloneEventRow {
                    ts_micros: 1_700_000_000_000_000 + i as i64,
                    workspace: "nordisk".into(),
                    member: format!("member-{i:05}"),
                    remote: format!("git@codeberg.org:nordisk/member-{i:05}.git"),
                    op: "clone-fetch".into(),
                    status: if i % 7 == 0 { "error" } else { "ok" }.into(),
                    // Realistic worst-case detail: a full git error chain / resolved SHA.
                    detail: format!(
                        "fatal: could not read from remote repository for member-{i:05}; \
                         resolved sha 0123456789abcdef0123456789abcdef01234567 after retry {}",
                        i % 5
                    ),
                    elapsed_ms: (i % 9000) as i64,
                })
                .collect()
        };

        // Boundary: empty + one row are trivially small.
        assert!(serde_json::to_string(&mk(0)).unwrap().len() < 16, "empty payload tiny");
        assert!(serde_json::to_string(&mk(1)).unwrap().len() < OLD_CAP, "one row never overflows");

        // At-truncation: the bounded 2 000-row payload fits the OLD 4 MiB cap —
        // proving `VIZ_CLONE_EVENTS_CAP` is a working bound for CloneEvents.
        let bounded = serde_json::to_string(&mk(ROW_CAP)).unwrap();
        assert!(
            bounded.len() < OLD_CAP,
            "the 2 000-row truncation keeps CloneEvents under 4 MiB ({} bytes)",
            bounded.len(),
        );

        // Unbounded: a never-thinned ledger (what nordisk accumulated) blows past
        // 4 MiB — the motivation for the 64 MiB cap, since other viz payloads
        // (Architecture graph, timeline) are NOT row-truncated.
        let unbounded = serde_json::to_string(&mk(20_000)).unwrap();
        assert!(
            unbounded.len() > OLD_CAP,
            "an unbounded ledger ({} bytes) overflows the old 4 MiB cap — the bug",
            unbounded.len(),
        );
        assert!(
            unbounded.len() < NEW_CAP,
            "and it fits the 64 MiB Viz-service cap ({} bytes < {NEW_CAP})",
            unbounded.len(),
        );
    }

    /// PO1 (inject-assert): roll a workspace's clone_events up into ONE verdict.
    /// A workspace with one failing member rolls up RED + names the failing member
    /// and surfaces its error; the LATEST op per member wins (a later ok heals an
    /// earlier error); an all-ok workspace is GREEN; an empty input is STALE.
    #[test]
    fn populate_verdict_rolls_up_per_member_latest() {
        let row = |ts: i64, member: &str, st: &str, detail: &str| CloneEventRow {
            ts_micros: ts,
            workspace: "nordisk".into(),
            member: member.into(),
            remote: format!("git@x:{member}.git"),
            op: op::CLONE_FETCH.into(),
            status: st.into(),
            detail: detail.into(),
            elapsed_ms: 1,
        };

        // facett ok (latest), korp error (latest) → RED, korp named + its error.
        let rows = vec![
            row(1, "facett", status::ERROR, "transient blip"), // older facett error
            row(5, "facett", status::OK, "abc123"),            // newer facett ok → heals
            row(3, "korp", status::ERROR, "Couldn't obtain Username"),
        ];
        let v = workspace_populate_verdict(&rows);
        assert_eq!(v.state, verdict::RED, "one member's latest is an error → red");
        assert_eq!(v.members, 2);
        assert_eq!(v.failed, 1, "only korp's latest is red (facett healed)");
        assert_eq!(v.failing_member, "korp");
        assert!(v.last_error.contains("Couldn't obtain Username"), "error surfaced: {}", v.last_error);
        assert!(v.last_synced.contains('T'), "newest ts rendered RFC3339");

        // All-ok workspace → green, no failing member.
        let green = workspace_populate_verdict(&[
            row(1, "a", status::OK, "sha-a"),
            row(2, "b", status::OK, "sha-b"),
        ]);
        assert_eq!(green.state, verdict::GREEN);
        assert_eq!(green.failed, 0);
        assert!(green.failing_member.is_empty());

        // Never-populated workspace (no events) → stale, not silently green.
        let stale = workspace_populate_verdict(&[]);
        assert_eq!(stale.state, verdict::STALE);
        assert_eq!(stale.members, 0);
        assert!(stale.last_synced.is_empty());
    }

    /// A republish outcome is recordable as its own workspace-level row.
    #[test]
    fn republish_outcome_records() {
        let dir = tempfile::tempdir().unwrap();
        let wh = IcebergWarehouse::open(dir.path()).unwrap();
        record_republish(&wh, "nordisk", status::OK, "snapshot-99", 1234);
        let rows = wh.block_on(query_clone_events(&wh, &CloneSelector::All)).unwrap();
        assert_eq!(rows.len(), 1);
        assert_eq!(rows[0].op, "republish");
        assert_eq!(rows[0].member, "*");
        assert_eq!(rows[0].detail, "snapshot-99");
        assert_eq!(rows[0].elapsed_ms, 1234);
    }
}