nornir 0.5.1

//! Server-global **workspace registry** — the persistent index of every
//! workspace the `nornir` server tracks.
//!
//! This is the index the self-sync poll loop updates, `nornir workspace ls`
//! reads, and (next phase) the `Workspaces.*` gRPC service / viz picker query.
//! It is distinct from a *warehouse* (one per workspace, under `builds/`) and
//! from a [`crate::workspace::WorkspaceDescriptor`] (the per-workspace
//! `nornir-workspace.toml` that lists members) — the registry just *indexes*
//! those, keyed by workspace name.
//!
//! **Storage:** a single redb file at `<server.root>/registry.redb`. redb
//! stores **bytes**, not structs, so each row is `workspace_name (&str) →`
//! the [`Workspace`] record serialized with serde_json. The server is the sole
//! writer, matching the redb/iceberg single-writer discipline used for the
//! warehouse.

use std::path::{Path, PathBuf};

use anyhow::{Context, Result, bail};
use redb::{Database, ReadableTable, TableDefinition};
use serde::{Deserialize, Serialize};

/// `name → serde_json(Workspace)`. Bytes in, bytes out.
const TABLE: TableDefinition<&str, &[u8]> = TableDefinition::new("workspaces");

/// Validate that a workspace name is a single safe path segment before it is
/// ever joined onto the registry root and handed to `remove_dir_all`/`open`.
///
/// A workspace name keys the registry **and** names an on-disk dir
/// (`<root>/<name>/builds`), so a name like `..`, `/etc`, or `a/b` could make the
/// purge path escape the registry root and `remove_dir_all` an arbitrary
/// directory. Reject empty names, anything containing a path separator (`/` or
/// `\`), the parent ref `..`, the current-dir ref `.`, and absolute / rooted
/// names. This is the data-loss guard the nordisk wipe taught us to add.
pub fn validate_ws_name(name: &str) -> Result<()> {
    if name.is_empty() {
        bail!("workspace name must not be empty");
    }
    if name == "." || name == ".." {
        bail!("workspace name must not be `.` or `..`");
    }
    if name.contains('/') || name.contains('\\') {
        bail!("workspace name `{name}` must be a single path segment (no `/` or `\\`)");
    }
    // Defense in depth: reject anything the OS would treat as more than one
    // normal component (absolute paths, drive prefixes, `..` components).
    let p = Path::new(name);
    let mut comps = p.components();
    match (comps.next(), comps.next()) {
        (Some(std::path::Component::Normal(c)), None) if c == name => Ok(()),
        _ => bail!("workspace name `{name}` is not a single safe path segment"),
    }
}

/// How a workspace's source is provided.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum Mode {
    /// Server polls member git URLs and republishes (server-user `nornir`).
    Monitored,
    /// A thin client pushes its working-tree source in; server computes.
    Pushed,
    /// Sources live outside (fat-style external checkout); only `builds/` is
    /// server-owned and `git/` stays empty.
    External,
}

impl Mode {
    pub fn as_str(&self) -> &'static str {
        match self {
            Mode::Monitored => "monitored",
            Mode::Pushed => "pushed",
            Mode::External => "external",
        }
    }
    /// Parse a mode string; anything unrecognized ⇒ [`Mode::Pushed`].
    pub fn parse(s: &str) -> Self {
        match s {
            "monitored" => Mode::Monitored,
            "external" => Mode::External,
            _ => Mode::Pushed,
        }
    }
}

/// Per-member sync state (meaningful in [`Mode::Monitored`]).
#[derive(Debug, Clone, Default, Serialize, Deserialize)]
pub struct MemberState {
    pub name: String,
    /// Git URL fetched/polled (mirrors `RepoSpec.git`).
    pub remote: String,
    /// Tracked ref/branch; empty ⇒ `main`.
    pub git_ref: String,
    /// Last SHA seen via `git ls-remote`; empty ⇒ never polled.
    pub last_seen_sha: String,
    /// RFC3339 of the last successful fetch; empty ⇒ never.
    pub last_synced: String,
    /// `"ok"` | `"fetching"` | `"error: …"` | "".
    pub sync_state: String,
    /// Working-tree freshness of the server's clone (AUT6). The SHA above can
    /// only see *committed* state; this digest sees uncommitted edits too. A
    /// thin client compares its own local digest against this to flag staleness.
    /// Equals [`crate::gitio::CLEAN_WORKTREE_DIGEST`] when the clone is clean;
    /// empty ⇒ never computed. See [`crate::gitio::worktree_freshness`].
    #[serde(default)]
    pub worktree_digest: String,
    /// `true` ⇒ the server's clone has uncommitted changes (tracked mods, staged,
    /// or untracked). Mirrors `worktree_digest != CLEAN_WORKTREE_DIGEST`.
    #[serde(default)]
    pub worktree_dirty: bool,
}

/// One registry row: a workspace the server tracks.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct Workspace {
    pub name: String,
    pub mode: Mode,
    /// Seed: a local path or git URL to the workspace's `nornir-workspace.toml`.
    pub descriptor: String,
    /// Poll interval for monitored workspaces, e.g. `"60s"`; empty ⇒ default.
    pub poll: String,
    /// Iceberg snapshot id the warehouse currently publishes; empty ⇒ none yet.
    pub current_snapshot: String,
    pub members: Vec<MemberState>,
    /// The descriptor's raw TOML **content**, stored IN the registry so the row is
    /// the self-contained source of truth — the server re-reads `deep_scan`/members
    /// from here, never a sibling `descriptors/*.toml` file. Empty for a git-URL
    /// seed whose content isn't known until first fetch. `#[serde(default)]` so rows
    /// written before this field round-trip cleanly.
    #[serde(default)]
    pub descriptor_content: String,
    pub created_at: String,
    pub updated_at: String,
}

impl Workspace {
    /// Build a record, seeding members from a local `nornir-workspace.toml`
    /// descriptor when it exists (each repo's `git` + `branch` →
    /// `remote`/`git_ref`; `path`-only members get an empty remote). A git-URL
    /// descriptor (not a local path) yields an empty member list until first
    /// fetch. Pass `created_at` to preserve it across a re-register.
    pub fn new(
        name: String,
        descriptor: String,
        mode: Mode,
        poll: String,
        created_at: Option<String>,
    ) -> Self {
        // Read the descriptor FILE when the seed is a local path, capturing its raw
        // content into the row so the registry — not the file — is the source of
        // truth thereafter (a git-URL seed leaves both empty until first fetch).
        let mut members = Vec::new();
        let mut content = String::new();
        let dpath = std::path::Path::new(&descriptor);
        if dpath.exists()
            && let Ok(text) = std::fs::read_to_string(dpath)
        {
            if let Ok(desc) = crate::workspace::WorkspaceDescriptor::from_content(&text) {
                members = Self::members_from(&desc);
            }
            content = text;
        }
        Self::assemble(name, descriptor, content, mode, poll, members, created_at)
    }

    /// Build a record from descriptor **content** (thin-mode: the client shipped the
    /// TOML). Members come from the content and the content itself is stored in the
    /// row — nothing touches disk, so no `descriptors/*.toml` file is needed.
    pub fn from_content(
        name: String,
        descriptor: String,
        content: String,
        mode: Mode,
        poll: String,
        created_at: Option<String>,
    ) -> Self {
        let members = crate::workspace::WorkspaceDescriptor::from_content(&content)
            .map(|d| Self::members_from(&d))
            .unwrap_or_default();
        Self::assemble(name, descriptor, content, mode, poll, members, created_at)
    }

    fn members_from(desc: &crate::workspace::WorkspaceDescriptor) -> Vec<MemberState> {
        desc.repos
            .iter()
            .map(|(mname, spec)| MemberState {
                name: mname.clone(),
                remote: spec.git.clone().unwrap_or_default(),
                git_ref: spec.branch.clone().unwrap_or_default(),
                ..Default::default()
            })
            .collect()
    }

    #[allow(clippy::too_many_arguments)]
    fn assemble(
        name: String,
        descriptor: String,
        descriptor_content: String,
        mode: Mode,
        poll: String,
        members: Vec<MemberState>,
        created_at: Option<String>,
    ) -> Self {
        let now = chrono::Utc::now().to_rfc3339();
        Workspace {
            name,
            mode,
            descriptor,
            poll,
            current_snapshot: String::new(),
            members,
            descriptor_content,
            created_at: created_at.unwrap_or_else(|| now.clone()),
            updated_at: now,
        }
    }
}

/// The redb-backed registry. Open once; cheap to clone-free reuse.
pub struct Registry {
    db: Database,
    /// The server root the registry lives under (`<root>/registry.redb`). Each
    /// workspace's on-disk build data sits beside it at `<root>/<name>/builds/`;
    /// [`Registry::remove`] resolves that from here when purging.
    root: PathBuf,
}

impl Registry {
    /// Open (creating if absent) the registry at `<root>/registry.redb`.
    pub fn open(root: &Path) -> Result<Self> {
        std::fs::create_dir_all(root)
            .with_context(|| format!("create registry root {}", root.display()))?;
        let path = root.join("registry.redb");
        let db = Database::create(&path)
            .with_context(|| format!("open {}", path.display()))?;
        // Materialize the table so first-time reads don't fail.
        let w = db.begin_write()?;
        {
            let _ = w.open_table(TABLE)?;
        }
        w.commit()?;
        Ok(Self { db, root: root.to_path_buf() })
    }

    /// The on-disk `builds/` dir for `name` (`<root>/<name>/builds`): the
    /// per-workspace warehouse + `catalog.redb` + `jobs.redb`. Resolving it here (off
    /// the registry root) keeps the layout in one place for the purge path.
    pub fn builds_dir(&self, name: &str) -> PathBuf {
        self.root.join(name).join("builds")
    }

    /// Registry root (the dir holding every `<name>/` workspace dir). Used by the
    /// purge path to assert a resolved `builds_dir` cannot escape the root.
    pub fn root(&self) -> &Path {
        &self.root
    }

    /// Insert or replace a workspace row.
    pub fn upsert(&self, ws: &Workspace) -> Result<()> {
        let bytes = serde_json::to_vec(ws).context("encode workspace record")?;
        let w = self.db.begin_write()?;
        {
            let mut t = w.open_table(TABLE)?;
            t.insert(ws.name.as_str(), bytes.as_slice())?;
        }
        w.commit()?;
        Ok(())
    }

    /// Fetch one workspace by name.
    pub fn get(&self, name: &str) -> Result<Option<Workspace>> {
        let r = self.db.begin_read()?;
        let t = r.open_table(TABLE)?;
        match t.get(name)? {
            Some(v) => Ok(Some(
                serde_json::from_slice(v.value()).context("decode workspace record")?,
            )),
            None => Ok(None),
        }
    }

    /// All workspaces, in key (name) order.
    pub fn list(&self) -> Result<Vec<Workspace>> {
        let r = self.db.begin_read()?;
        let t = r.open_table(TABLE)?;
        let mut out = Vec::new();
        for row in t.iter()? {
            let (_k, v) = row?;
            out.push(serde_json::from_slice(v.value()).context("decode workspace record")?);
        }
        Ok(out)
    }

    /// Atomically set a workspace's `current_snapshot` (and bump `updated_at`)
    /// inside a SINGLE redb write transaction — read the row, mutate only those
    /// two fields, write it back, all under the one writer lock. This closes the
    /// republish read-modify-write window where a concurrent `fetch_workspace`
    /// member-SHA write could be clobbered by a stale read-then-upsert in
    /// `republish_served`. Returns whether a row existed to update.
    pub fn set_snapshot(&self, name: &str, snapshot: &str) -> Result<bool> {
        let w = self.db.begin_write()?;
        let existed = {
            let mut t = w.open_table(TABLE)?;
            let cur = t.get(name)?.map(|v| v.value().to_vec());
            match cur {
                Some(bytes) => {
                    let mut ws: Workspace =
                        serde_json::from_slice(&bytes).context("decode workspace record")?;
                    ws.current_snapshot = snapshot.to_string();
                    ws.updated_at = chrono::Utc::now().to_rfc3339();
                    let out = serde_json::to_vec(&ws).context("encode workspace record")?;
                    t.insert(name, out.as_slice())?;
                    true
                }
                None => false,
            }
        };
        w.commit()?;
        Ok(existed)
    }

    /// Remove a workspace; returns whether the registry row existed.
    ///
    /// This is the **shared core** both faces land on: the CLI `nornir workspace rm`
    /// (locally, or via the `Workspaces.Remove` RPC) and the viz "🗑 Kill workspace"
    /// confirm (always via the RPC) both call here, so the purge semantics are fixed
    /// once, not per-face.
    ///
    /// - `purge == false` (default, back-compat): drop ONLY the registry row. The
    ///   workspace's `builds/` warehouse stays on disk.
    /// - `purge == true`: after dropping the row, also `remove_dir_all` the
    ///   workspace's [`builds_dir`](Self::builds_dir) (`<root>/<name>/builds/` —
    ///   warehouse + `catalog.redb` + `jobs.redb`), so the workspace is fully gone.
    ///   The on-disk wipe is **independent of whether a registry row existed**: an
    ///   earlier non-purge `rm` (or viz Kill without "delete data") leaves an orphan
    ///   `builds/` behind with no row, and a later `rm <name> --purge` must finish
    ///   that cleanup rather than bail. A missing builds dir is a no-op.
    ///
    /// Returns `true` if **anything was removed** — either the registry row OR (under
    /// `purge`) an orphaned `builds/` dir — so the caller reports success instead of
    /// "no such workspace" when it cleaned up disk-only leftovers.
    pub fn remove(&self, name: &str, purge: bool) -> Result<bool> {
        // Validate BEFORE touching the FS — the purge runs even when no row
        // matches, so a malicious/garbage name must be rejected up front.
        validate_ws_name(name)?;
        let w = self.db.begin_write()?;
        let existed = {
            let mut t = w.open_table(TABLE)?;
            let removed = t.remove(name)?.is_some();
            removed
        };
        w.commit()?;
        let mut purged = false;
        if purge {
            let builds = self.builds_dir(name);
            if builds.exists() {
                // Canonicalize and assert the resolved dir is still under the
                // registry root before the irreversible wipe — a symlink or a
                // name that slipped validation must never let remove_dir_all
                // escape `<root>/`.
                let canon_builds = std::fs::canonicalize(&builds)
                    .with_context(|| format!("canonicalize builds dir {}", builds.display()))?;
                let canon_root = std::fs::canonicalize(&self.root)
                    .with_context(|| format!("canonicalize registry root {}", self.root.display()))?;
                if !canon_builds.starts_with(&canon_root) {
                    bail!(
                        "refusing to purge {}: resolved outside registry root {}",
                        canon_builds.display(),
                        canon_root.display()
                    );
                }
                std::fs::remove_dir_all(&canon_builds)
                    .with_context(|| format!("purge builds dir {}", canon_builds.display()))?;
                purged = true;
            }
        }
        Ok(existed || purged)
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    fn rec(name: &str, mode: Mode) -> Workspace {
        Workspace {
            name: name.into(),
            mode,
            descriptor: "/tmp/ws/nornir-workspace.toml".into(),
            poll: "60s".into(),
            current_snapshot: String::new(),
            members: vec![MemberState {
                name: "holger".into(),
                remote: "git@codeberg.org:nordisk/holger".into(),
                git_ref: "main".into(),
                ..Default::default()
            }],
            descriptor_content: String::new(),
            created_at: "2026-06-08T00:00:00Z".into(),
            updated_at: "2026-06-08T00:00:00Z".into(),
        }
    }

    #[test]
    fn roundtrip_upsert_get_list_remove() {
        let dir = std::env::temp_dir().join(format!("nornir-reg-{}", std::process::id()));
        let reg = Registry::open(&dir).unwrap();

        assert!(reg.get("a").unwrap().is_none());
        reg.upsert(&rec("a", Mode::Monitored)).unwrap();
        reg.upsert(&rec("b", Mode::Pushed)).unwrap();

        let a = reg.get("a").unwrap().unwrap();
        assert_eq!(a.mode, Mode::Monitored);
        assert_eq!(a.members[0].remote, "git@codeberg.org:nordisk/holger");

        let all = reg.list().unwrap();
        assert_eq!(all.len(), 2);
        assert_eq!(all[0].name, "a"); // key order

        assert!(reg.remove("a", false).unwrap());
        assert!(!reg.remove("a", false).unwrap());
        assert_eq!(reg.list().unwrap().len(), 1);

        std::fs::remove_dir_all(&dir).ok();
    }

    /// INJECT-ASSERT: the shared purge core. `remove(.., true)` deletes the
    /// workspace's `<root>/<name>/builds/` dir; `remove(.., false)` leaves it (the
    /// back-compat registry-only path). Resolution is off the registry root.
    #[test]
    fn remove_purge_deletes_builds_dir_non_purge_leaves_it() {
        let dir = std::env::temp_dir().join(format!("nornir-purge-{}", std::process::id()));
        std::fs::remove_dir_all(&dir).ok();
        let reg = Registry::open(&dir).unwrap();

        // Two workspaces, each with an on-disk builds/ tree holding warehouse files.
        for ws in ["keep", "nuke"] {
            reg.upsert(&rec(ws, Mode::Pushed)).unwrap();
            let builds = reg.builds_dir(ws);
            std::fs::create_dir_all(builds.join("warehouse")).unwrap();
            std::fs::write(builds.join("catalog.redb"), b"x").unwrap();
            std::fs::write(builds.join("jobs.redb"), b"y").unwrap();
            assert!(builds.exists());
        }

        // Non-purge: row gone, builds/ stays.
        assert!(reg.remove("keep", false).unwrap());
        assert!(reg.get("keep").unwrap().is_none());
        assert!(reg.builds_dir("keep").exists(), "non-purge must leave builds/");

        // Purge: row gone AND builds/ removed.
        assert!(reg.remove("nuke", true).unwrap());
        assert!(reg.get("nuke").unwrap().is_none());
        assert!(!reg.builds_dir("nuke").exists(), "purge must remove builds/");

        // Purge of a missing builds dir is a no-op (row already gone ⇒ false).
        assert!(!reg.remove("nuke", true).unwrap());

        // Orphaned builds/ with NO registry row (an earlier non-purge `rm` left it):
        // `rm --purge` must still wipe the dir AND report success (true), not bail.
        let orphan = reg.builds_dir("orphan");
        std::fs::create_dir_all(orphan.join("warehouse")).unwrap();
        std::fs::write(orphan.join("catalog.redb"), b"x").unwrap();
        assert!(reg.get("orphan").unwrap().is_none(), "no registry row for orphan");
        assert!(
            reg.remove("orphan", true).unwrap(),
            "purge of orphaned builds/ (no row) must report success"
        );
        assert!(!reg.builds_dir("orphan").exists(), "orphan builds/ must be wiped");
        // Non-purge on a row-less name with no builds/ is still a no-op (false).
        assert!(!reg.remove("orphan", false).unwrap());

        std::fs::remove_dir_all(&dir).ok();
    }

    /// Bug #4: set_snapshot updates only current_snapshot (+ updated_at) in one
    /// txn and preserves every other field (e.g. a concurrently-written member
    /// SHA), and is a no-op false when the row is absent.
    #[test]
    fn set_snapshot_is_atomic_field_update() {
        let dir = std::env::temp_dir().join(format!("nornir-reg-snap-{}", std::process::id()));
        std::fs::remove_dir_all(&dir).ok();
        let reg = Registry::open(&dir).unwrap();

        assert!(!reg.set_snapshot("missing", "s1").unwrap(), "no row → false");

        reg.upsert(&rec("w", Mode::Monitored)).unwrap();
        assert!(reg.set_snapshot("w", "snap-123").unwrap());
        let got = reg.get("w").unwrap().unwrap();
        assert_eq!(got.current_snapshot, "snap-123");
        // Untouched fields survive the targeted update.
        assert_eq!(got.members[0].remote, "git@codeberg.org:nordisk/holger");
        assert_eq!(got.mode, Mode::Monitored);

        std::fs::remove_dir_all(&dir).ok();
    }

    /// Bug #2: an unsafe workspace name must be rejected before the purge path
    /// can `remove_dir_all` anything — empty, separators, `..`, `.`, absolute.
    #[test]
    fn validate_ws_name_rejects_path_escapes() {
        assert!(validate_ws_name("nordisk").is_ok());
        assert!(validate_ws_name("a-b_c.123").is_ok());
        for bad in ["", ".", "..", "/", "/etc", "a/b", "..\\..", "../sibling", "\\", "."] {
            assert!(validate_ws_name(bad).is_err(), "must reject {bad:?}");
        }
        // And `remove` itself refuses an escaping name even with purge=true.
        let dir = std::env::temp_dir().join(format!("nornir-reg-bad-{}", std::process::id()));
        std::fs::remove_dir_all(&dir).ok();
        let reg = Registry::open(&dir).unwrap();
        assert!(reg.remove("../escape", true).is_err());
        std::fs::remove_dir_all(&dir).ok();
    }
}

/// One row of the `workspace ls` roster — the projection BOTH the fat (registry
/// [`Workspace`]) and thin (proto) faces map into, so they render identically
/// (CLI⟺UI parity, the Nornir-tab roster's data). AUT9.
#[derive(Debug, Clone, Default, serde::Serialize, serde::Deserialize)]
pub struct RosterRow {
    pub name: String,
    pub mode: String,
    pub members: usize,
    pub last_synced: String,
    /// PO1 — the rolled-up POPULATE verdict for this workspace (latest clone/fetch
    /// outcome per member → a workspace state). `"green"` (all members' latest ok),
    /// `"red"` (≥1 member's latest is an error), `"stale"` (never populated — no
    /// clone_events), or `""` when the populate roll-up wasn't requested
    /// (`workspace ls` without `--status`). The roster's red/green badge.
    #[serde(default, skip_serializing_if = "String::is_empty")]
    pub populate: String,
    /// The first failing member (when `populate == "red"`), else empty.
    #[serde(default, skip_serializing_if = "String::is_empty")]
    pub failing_member: String,
    /// That member's latest error detail (when `populate == "red"`), else empty.
    #[serde(default, skip_serializing_if = "String::is_empty")]
    pub last_error: String,
}

/// `workspace ls` as a uniform [`crate::cli_outcome::CommandOutcome`] — the CLI twin
/// of the Nornir-tab workspace roster. `rows` are already projected from the
/// canonical source (fat: `Registry::list`; thin: `Workspaces.List` RPC) — this only
/// shapes + renders. `ok ⟺ ≥1 workspace` (an empty registry is a valid resting state
/// for a user, but for the autopilot a seeded-then-empty roster is RED).
pub fn roster_outcome(rows: &[RosterRow]) -> crate::cli_outcome::CommandOutcome {
    use crate::cli_outcome::CommandOutcome;
    if rows.is_empty() {
        return CommandOutcome::fail(
            "workspace ls",
            "no workspaces registered — `nornir workspace register …`",
        );
    }
    // PO1: show the POPULATE column only when at least one row carries a verdict
    // (i.e. `--status` was passed / the viz roster requested the roll-up).
    let has_populate = rows.iter().any(|r| !r.populate.is_empty());
    let mut human = if has_populate {
        format!(
            "{:<20} {:<10} {:>7}  {:<22} {:<8} {}",
            "NAME", "MODE", "MEMBERS", "LAST SYNCED", "POPULATE", "LAST ERROR"
        )
    } else {
        format!("{:<20} {:<10} {:>7}  {}", "NAME", "MODE", "MEMBERS", "LAST SYNCED")
    };
    for r in rows {
        if has_populate {
            let mark = match r.populate.as_str() {
                "green" => "✓ ok",
                "red" => "✗ failed",
                "stale" => "· stale",
                _ => "—",
            };
            let err = if r.failing_member.is_empty() {
                String::new()
            } else {
                format!("{}: {}", r.failing_member, r.last_error)
            };
            human.push_str(&format!(
                "\n{:<20} {:<10} {:>7}  {:<22} {:<8} {}",
                r.name, r.mode, r.members, r.last_synced, mark, err
            ));
        } else {
            human.push_str(&format!(
                "\n{:<20} {:<10} {:>7}  {}",
                r.name, r.mode, r.members, r.last_synced
            ));
        }
    }
    CommandOutcome::ok("workspace ls", serde_json::json!(rows), human)
}

#[cfg(test)]
mod roster_outcome_tests {
    use super::*;

    #[test]
    fn empty_registry_roster_is_red() {
        let o = roster_outcome(&[]);
        assert_eq!(o.command, "workspace ls");
        assert!(!o.is_sannr(), "an empty roster is RED for the autopilot");
    }

    #[test]
    fn populated_roster_is_sannr_with_rows() {
        let rows = vec![
            RosterRow { name: "nordisk".into(), mode: "monitored".into(), members: 8, last_synced: "2026-06-23".into(), ..Default::default() },
            RosterRow { name: "holger".into(), mode: "local".into(), members: 1, last_synced: "never".into(), ..Default::default() },
        ];
        let o = roster_outcome(&rows);
        assert!(o.is_sannr(), "a populated roster is a true (sannr) outcome");
        let arr = o.data.as_array().unwrap();
        assert_eq!(arr.len(), 2);
        assert_eq!(arr[0]["name"], serde_json::json!("nordisk"));
        assert_eq!(arr[0]["members"], serde_json::json!(8));
    }

    /// PO1 (inject-assert): a roster with a RED workspace (a failing member) and a
    /// GREEN workspace renders the verdicts + names the failing member in the
    /// human view, and carries them in the JSON data. The CLI twin of the
    /// Nornir-tab roster's red/green badge.
    #[test]
    fn roster_with_populate_verdicts_shows_red_and_failing_member() {
        let rows = vec![
            RosterRow {
                name: "nordisk".into(),
                mode: "monitored".into(),
                members: 8,
                last_synced: "2026-06-23".into(),
                populate: "red".into(),
                failing_member: "korp".into(),
                last_error: "Couldn't obtain Username".into(),
            },
            RosterRow {
                name: "holger".into(),
                mode: "monitored".into(),
                members: 2,
                last_synced: "2026-06-23".into(),
                populate: "green".into(),
                ..Default::default()
            },
        ];
        let o = roster_outcome(&rows);
        assert!(o.is_sannr());
        // JSON carries the verdict + failing member for the failed workspace.
        let arr = o.data.as_array().unwrap();
        let nordisk = arr.iter().find(|r| r["name"] == "nordisk").unwrap();
        assert_eq!(nordisk["populate"], "red");
        assert_eq!(nordisk["failing_member"], "korp");
        assert!(nordisk["last_error"].as_str().unwrap().contains("Couldn't obtain Username"));
        let holger = arr.iter().find(|r| r["name"] == "holger").unwrap();
        assert_eq!(holger["populate"], "green");
        // A green workspace omits the failing-member/error keys (skip_serializing_if).
        assert!(holger.get("failing_member").is_none());
        // The human render shows the POPULATE column + names the failing member.
        assert!(o.human.contains("POPULATE"), "populate column shown: {}", o.human);
        assert!(o.human.contains("✗ failed"));
        assert!(o.human.contains("korp: Couldn't obtain Username"));
    }
}