nornir 0.5.4

//! Yggdrasil — the **infra crown** (`infra_nodes` / `infra_edges`) + the
//! **runtime band** (`runtime_spans`) of the one super-graph.
//!
//! `yggdrasil-supergraph.md` describes the warehouse as one tree whose Iceberg
//! tables are bands of nodes/edges at every scale — instruction → symbol → call
//! → crate → repo → dependency → architecture, and, at the CROWN, **where the
//! code actually runs**: the machines / services / VMs / containers of the
//! constellation, plus the general runtime traces threaded through them. That
//! crown + runtime band was the last "MISSING ⛔" scale; this module lands it.
//!
//! Persistence mirrors the dependency-graph snapshot (`warehouse::dep_graph`):
//! each `record_*` call writes ONE snapshot (all rows share a fresh
//! `snapshot_id` + one `ts_micros`), scoped by `workspace_name`; each `query_*`
//! reads them back grouped into whole snapshots. It reuses the EXISTING
//! `IcebergWarehouse` (skade's warehouse) — no schema owned outside nornir, no
//! new skade dependency: the warehouse already stores arbitrary tables.
//!
//! **Real producer (`infra_from_jobs` / `spans_from_jobs`):** the jera
//! `JobRecord` ledger. Every long-running nornir op is a durable ledger row that
//! already carries *which node ran it* (`JobRecord::node`, the fleet machine)
//! and its lifecycle timing. From that we derive:
//!   * a **machine** infra-node per distinct `node`, a **service** infra-node
//!     per distinct job `kind`, and a `runs_on` **infra-edge** for every
//!     `(service kind → machine)` pair that co-occurred;
//!   * a **runtime span** per job (`span_id = job_id`, `parent = parent_id`,
//!     `machine = node`, `dur_us = elapsed_ms * 1000`).
//!
//! So the crown is not dead schema — it is populated straight from the ledger
//! that jera's multi-machine scheduler already stamps.

use std::collections::{BTreeMap, BTreeSet};
use std::sync::Arc;

use anyhow::{anyhow, Result};
use arrow::array::{
    Array, BooleanArray, Int32Array, Int64Array, RecordBatch, StringArray,
    TimestampMicrosecondArray,
};
use chrono::Utc;
use iceberg::arrow::schema_to_arrow_schema;
use iceberg::Catalog;
use uuid::Uuid;

use super::iceberg::{
    append_batch, ensure_table_schema, IcebergWarehouse, TABLE_INFRA_EDGES, TABLE_INFRA_NODES,
    TABLE_RUNTIME_SPANS,
};
use crate::jobs::JobRecord;

/// A node in the infra crown — a machine, service, VM, or container.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct InfraNode {
    /// Stable id (host name for a machine, `service:<kind>` for a service, …).
    pub id: String,
    /// `machine` | `service` | `vm` | `container`.
    pub kind: String,
    /// The host this lives on (a machine hosts itself).
    pub host: String,
    /// Worker-group / cluster this belongs to.
    pub group: String,
    /// CPU cores (0 when unknown — the ledger doesn't carry roster specs).
    pub cores: i32,
    /// Memory in MiB (0 when unknown).
    pub mem_mb: i64,
    /// Free-form role label (`worker` / `control-plane` / the service kind).
    pub role: String,
    /// Whether it was seen live in this snapshot.
    pub online: bool,
}

/// An edge in the infra crown (`runs_on` / `hosts` / `deploys` / `member_of`).
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct InfraEdge {
    pub from: String,
    pub to: String,
    pub rel: String,
}

/// One executed span in the runtime band (a jera job or a server op).
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct RuntimeSpan {
    pub span_id: String,
    /// Parent span id (`""` = a root/top-level span).
    pub parent: String,
    /// The service that ran it (the job `kind`).
    pub service: String,
    /// What it acted on (the job `target`).
    pub op: String,
    pub start_micros: i64,
    /// Duration in microseconds (0 when the op is still running / untimed).
    pub dur_us: i64,
    pub job_id: String,
    /// The fleet machine that executed it (`local` when unstamped).
    pub machine: String,
    pub status: String,
}

/// Machine placeholder for a job whose `node` is unset (ran on the local
/// control-plane, not dispatched across the fleet).
pub const LOCAL_MACHINE: &str = "local";
/// Default worker-group when the ledger carries no cluster label.
pub const CONSTELLATION_GROUP: &str = "constellation";

/// Derive the infra crown (nodes + edges) from a slice of jera job records —
/// the **real producer**. Pure and warehouse-free, so it unit-tests on its own.
///
/// * one `machine` node per distinct `node` (unset → [`LOCAL_MACHINE`]),
/// * one `service` node per distinct job `kind`,
/// * a `runs_on` edge `service → machine` for each `(kind, node)` that ran,
/// * a `member_of` edge `machine → group` per machine.
///
/// Nodes/edges are de-duplicated and returned in a stable (sorted) order so the
/// snapshot is deterministic.
pub fn infra_from_jobs(jobs: &[JobRecord]) -> (Vec<InfraNode>, Vec<InfraEdge>) {
    let mut machines: BTreeSet<String> = BTreeSet::new();
    let mut services: BTreeSet<String> = BTreeSet::new();
    // (service_kind, machine) pairs that co-occurred.
    let mut runs_on: BTreeSet<(String, String)> = BTreeSet::new();

    for j in jobs {
        let machine = j.node.clone().filter(|n| !n.is_empty()).unwrap_or_else(|| LOCAL_MACHINE.to_string());
        machines.insert(machine.clone());
        if !j.kind.is_empty() {
            services.insert(j.kind.clone());
            runs_on.insert((j.kind.clone(), machine));
        }
    }

    let mut nodes: Vec<InfraNode> = Vec::new();
    for m in &machines {
        nodes.push(InfraNode {
            id: m.clone(),
            kind: "machine".to_string(),
            host: m.clone(),
            group: CONSTELLATION_GROUP.to_string(),
            cores: 0,
            mem_mb: 0,
            role: if m == LOCAL_MACHINE { "control-plane".to_string() } else { "worker".to_string() },
            online: true,
        });
    }
    for s in &services {
        nodes.push(InfraNode {
            id: format!("service:{s}"),
            kind: "service".to_string(),
            host: String::new(),
            group: CONSTELLATION_GROUP.to_string(),
            cores: 0,
            mem_mb: 0,
            role: s.clone(),
            online: true,
        });
    }

    let mut edges: Vec<InfraEdge> = Vec::new();
    for (kind, machine) in &runs_on {
        edges.push(InfraEdge {
            from: format!("service:{kind}"),
            to: machine.clone(),
            rel: "runs_on".to_string(),
        });
    }
    for m in &machines {
        edges.push(InfraEdge {
            from: m.clone(),
            to: CONSTELLATION_GROUP.to_string(),
            rel: "member_of".to_string(),
        });
    }
    edges.sort_by(|a, b| (&a.from, &a.to, &a.rel).cmp(&(&b.from, &b.to, &b.rel)));
    (nodes, edges)
}

/// Derive one runtime span per job record — the **real producer** for the
/// runtime band. Pure/warehouse-free. Spans are returned sorted by
/// `(start_micros, span_id)` so the snapshot is deterministic.
pub fn spans_from_jobs(jobs: &[JobRecord]) -> Vec<RuntimeSpan> {
    let mut spans: Vec<RuntimeSpan> = jobs
        .iter()
        .map(|j| RuntimeSpan {
            span_id: j.job_id.clone(),
            parent: j.parent_id.clone().unwrap_or_default(),
            service: j.kind.clone(),
            op: j.target.clone(),
            start_micros: j.ts_start_micros,
            dur_us: j.elapsed_ms.map(|ms| ms.saturating_mul(1000)).unwrap_or(0),
            job_id: j.job_id.clone(),
            machine: j.node.clone().filter(|n| !n.is_empty()).unwrap_or_else(|| LOCAL_MACHINE.to_string()),
            status: j.status.clone(),
        })
        .collect();
    spans.sort_by(|a, b| (a.start_micros, &a.span_id).cmp(&(b.start_micros, &b.span_id)));
    spans
}

// ─── persistence (suite warehouse) ────────────────────────────────────────

/// Append one infra-node snapshot. All rows share the returned `snapshot_id`
/// and a single `ts_micros`. Empty `nodes` writes no snapshot.
pub async fn record_infra_nodes(
    wh: &IcebergWarehouse,
    workspace_name: &str,
    nodes: &[InfraNode],
) -> Result<Uuid> {
    let snapshot_id = Uuid::new_v4();
    if nodes.is_empty() {
        return Ok(snapshot_id);
    }
    let ts = Utc::now().timestamp_micros();
    let sid = snapshot_id.to_string();
    let n = nodes.len();

    let ident = wh.table_ident(TABLE_INFRA_NODES);
    let table = wh.catalog().load_table(&ident).await?;
    let table =
        ensure_table_schema(wh.catalog(), &ident, table, &super::iceberg_schema::infra_nodes()?)
            .await?;
    let arrow_schema = Arc::new(schema_to_arrow_schema(table.metadata().current_schema())?);

    let cols: Vec<Arc<dyn Array>> = vec![
        Arc::new(StringArray::from(vec![sid.as_str(); n])),
        Arc::new(StringArray::from(vec![workspace_name; n])),
        Arc::new(TimestampMicrosecondArray::from(vec![ts; n]).with_timezone("+00:00")),
        Arc::new(StringArray::from(nodes.iter().map(|x| x.id.clone()).collect::<Vec<_>>())),
        Arc::new(StringArray::from(nodes.iter().map(|x| x.kind.clone()).collect::<Vec<_>>())),
        Arc::new(StringArray::from(nodes.iter().map(|x| x.host.clone()).collect::<Vec<_>>())),
        Arc::new(StringArray::from(nodes.iter().map(|x| x.group.clone()).collect::<Vec<_>>())),
        Arc::new(Int32Array::from(nodes.iter().map(|x| x.cores).collect::<Vec<_>>())),
        Arc::new(Int64Array::from(nodes.iter().map(|x| x.mem_mb).collect::<Vec<_>>())),
        Arc::new(StringArray::from(nodes.iter().map(|x| x.role.clone()).collect::<Vec<_>>())),
        Arc::new(BooleanArray::from(nodes.iter().map(|x| x.online).collect::<Vec<_>>())),
    ];
    let batch = RecordBatch::try_new(arrow_schema, cols)?;
    append_batch(wh.catalog(), table, batch).await?;
    Ok(snapshot_id)
}

/// Append one infra-edge snapshot. Empty `edges` writes no snapshot.
pub async fn record_infra_edges(
    wh: &IcebergWarehouse,
    workspace_name: &str,
    edges: &[InfraEdge],
) -> Result<Uuid> {
    let snapshot_id = Uuid::new_v4();
    if edges.is_empty() {
        return Ok(snapshot_id);
    }
    let ts = Utc::now().timestamp_micros();
    let sid = snapshot_id.to_string();
    let n = edges.len();

    let ident = wh.table_ident(TABLE_INFRA_EDGES);
    let table = wh.catalog().load_table(&ident).await?;
    let table =
        ensure_table_schema(wh.catalog(), &ident, table, &super::iceberg_schema::infra_edges()?)
            .await?;
    let arrow_schema = Arc::new(schema_to_arrow_schema(table.metadata().current_schema())?);

    let cols: Vec<Arc<dyn Array>> = vec![
        Arc::new(StringArray::from(vec![sid.as_str(); n])),
        Arc::new(StringArray::from(vec![workspace_name; n])),
        Arc::new(TimestampMicrosecondArray::from(vec![ts; n]).with_timezone("+00:00")),
        Arc::new(StringArray::from(edges.iter().map(|x| x.from.clone()).collect::<Vec<_>>())),
        Arc::new(StringArray::from(edges.iter().map(|x| x.to.clone()).collect::<Vec<_>>())),
        Arc::new(StringArray::from(edges.iter().map(|x| x.rel.clone()).collect::<Vec<_>>())),
    ];
    let batch = RecordBatch::try_new(arrow_schema, cols)?;
    append_batch(wh.catalog(), table, batch).await?;
    Ok(snapshot_id)
}

/// Append one runtime-span snapshot. Empty `spans` writes no snapshot.
pub async fn record_runtime_spans(
    wh: &IcebergWarehouse,
    workspace_name: &str,
    spans: &[RuntimeSpan],
) -> Result<Uuid> {
    let snapshot_id = Uuid::new_v4();
    if spans.is_empty() {
        return Ok(snapshot_id);
    }
    let ts = Utc::now().timestamp_micros();
    let sid = snapshot_id.to_string();
    let n = spans.len();

    let ident = wh.table_ident(TABLE_RUNTIME_SPANS);
    let table = wh.catalog().load_table(&ident).await?;
    let table =
        ensure_table_schema(wh.catalog(), &ident, table, &super::iceberg_schema::runtime_spans()?)
            .await?;
    let arrow_schema = Arc::new(schema_to_arrow_schema(table.metadata().current_schema())?);

    let cols: Vec<Arc<dyn Array>> = vec![
        Arc::new(StringArray::from(vec![sid.as_str(); n])),
        Arc::new(StringArray::from(vec![workspace_name; n])),
        Arc::new(TimestampMicrosecondArray::from(vec![ts; n]).with_timezone("+00:00")),
        Arc::new(StringArray::from(spans.iter().map(|x| x.span_id.clone()).collect::<Vec<_>>())),
        Arc::new(StringArray::from(spans.iter().map(|x| x.parent.clone()).collect::<Vec<_>>())),
        Arc::new(StringArray::from(spans.iter().map(|x| x.service.clone()).collect::<Vec<_>>())),
        Arc::new(StringArray::from(spans.iter().map(|x| x.op.clone()).collect::<Vec<_>>())),
        Arc::new(Int64Array::from(spans.iter().map(|x| x.start_micros).collect::<Vec<_>>())),
        Arc::new(Int64Array::from(spans.iter().map(|x| x.dur_us).collect::<Vec<_>>())),
        Arc::new(StringArray::from(spans.iter().map(|x| x.job_id.clone()).collect::<Vec<_>>())),
        Arc::new(StringArray::from(spans.iter().map(|x| x.machine.clone()).collect::<Vec<_>>())),
        Arc::new(StringArray::from(spans.iter().map(|x| x.status.clone()).collect::<Vec<_>>())),
    ];
    let batch = RecordBatch::try_new(arrow_schema, cols)?;
    append_batch(wh.catalog(), table, batch).await?;
    Ok(snapshot_id)
}

/// Read every infra node for `workspace_name`, sorted by `(kind, id)`.
pub async fn query_infra_nodes(wh: &IcebergWarehouse, workspace_name: &str) -> Result<Vec<InfraNode>> {
    let filter = skade::ScanFilter::eq("workspace_name", workspace_name);
    let cols = [
        "workspace_name", "node_id", "kind", "host", "node_group", "cores", "mem_mb", "role",
        "online",
    ];
    let batches =
        super::iceberg::load_and_read_filtered(wh, TABLE_INFRA_NODES, &filter, &cols).await?;
    let mut out: Vec<InfraNode> = Vec::new();
    for b in &batches {
        let ws = col::<StringArray>(b, "workspace_name")?;
        let id = col::<StringArray>(b, "node_id")?;
        let kind = col::<StringArray>(b, "kind")?;
        let host = col::<StringArray>(b, "host")?;
        let group = col::<StringArray>(b, "node_group")?;
        let cores = col::<Int32Array>(b, "cores")?;
        let mem = col::<Int64Array>(b, "mem_mb")?;
        let role = col::<StringArray>(b, "role")?;
        let online = col::<BooleanArray>(b, "online")?;
        for i in 0..b.num_rows() {
            // Residual guard: pushdown prunes at file granularity, not per row.
            if ws.value(i) != workspace_name {
                continue;
            }
            out.push(InfraNode {
                id: id.value(i).to_string(),
                kind: kind.value(i).to_string(),
                host: host.value(i).to_string(),
                group: group.value(i).to_string(),
                cores: cores.value(i),
                mem_mb: mem.value(i),
                role: role.value(i).to_string(),
                online: online.value(i),
            });
        }
    }
    out.sort_by(|a, b| (&a.kind, &a.id).cmp(&(&b.kind, &b.id)));
    Ok(out)
}

/// Read every infra edge for `workspace_name`, sorted by `(rel, from, to)`.
pub async fn query_infra_edges(wh: &IcebergWarehouse, workspace_name: &str) -> Result<Vec<InfraEdge>> {
    let filter = skade::ScanFilter::eq("workspace_name", workspace_name);
    let cols = ["workspace_name", "from_node", "to_node", "rel"];
    let batches =
        super::iceberg::load_and_read_filtered(wh, TABLE_INFRA_EDGES, &filter, &cols).await?;
    let mut out: Vec<InfraEdge> = Vec::new();
    for b in &batches {
        let ws = col::<StringArray>(b, "workspace_name")?;
        let from = col::<StringArray>(b, "from_node")?;
        let to = col::<StringArray>(b, "to_node")?;
        let rel = col::<StringArray>(b, "rel")?;
        for i in 0..b.num_rows() {
            if ws.value(i) != workspace_name {
                continue;
            }
            out.push(InfraEdge {
                from: from.value(i).to_string(),
                to: to.value(i).to_string(),
                rel: rel.value(i).to_string(),
            });
        }
    }
    out.sort_by(|a, b| (&a.rel, &a.from, &a.to).cmp(&(&b.rel, &b.from, &b.to)));
    Ok(out)
}

/// Read every runtime span for `workspace_name`, sorted by `(start_micros, span_id)`.
pub async fn query_runtime_spans(
    wh: &IcebergWarehouse,
    workspace_name: &str,
) -> Result<Vec<RuntimeSpan>> {
    let filter = skade::ScanFilter::eq("workspace_name", workspace_name);
    let cols = [
        "workspace_name", "span_id", "parent_span", "service", "op", "start_micros", "dur_us",
        "job_id", "machine", "status",
    ];
    let batches =
        super::iceberg::load_and_read_filtered(wh, TABLE_RUNTIME_SPANS, &filter, &cols).await?;
    let mut out: Vec<RuntimeSpan> = Vec::new();
    for b in &batches {
        let ws = col::<StringArray>(b, "workspace_name")?;
        let span_id = col::<StringArray>(b, "span_id")?;
        let parent = col::<StringArray>(b, "parent_span")?;
        let service = col::<StringArray>(b, "service")?;
        let op = col::<StringArray>(b, "op")?;
        let start = col::<Int64Array>(b, "start_micros")?;
        let dur = col::<Int64Array>(b, "dur_us")?;
        let job_id = col::<StringArray>(b, "job_id")?;
        let machine = col::<StringArray>(b, "machine")?;
        let status = col::<StringArray>(b, "status")?;
        for i in 0..b.num_rows() {
            if ws.value(i) != workspace_name {
                continue;
            }
            out.push(RuntimeSpan {
                span_id: span_id.value(i).to_string(),
                parent: parent.value(i).to_string(),
                service: service.value(i).to_string(),
                op: op.value(i).to_string(),
                start_micros: start.value(i),
                dur_us: dur.value(i),
                job_id: job_id.value(i).to_string(),
                machine: machine.value(i).to_string(),
                status: status.value(i).to_string(),
            });
        }
    }
    out.sort_by(|a, b| (a.start_micros, &a.span_id).cmp(&(b.start_micros, &b.span_id)));
    Ok(out)
}

/// One-call **producer + persist**: derive the infra crown + runtime band from
/// the jera job ledger and write all three snapshots. Returns the three
/// `snapshot_id`s `(infra_nodes, infra_edges, runtime_spans)`. This is the seam
/// a monitor / `nornir yggdrasil ingest` calls with `JobStore::list(&All)`.
pub async fn ingest_from_jobs(
    wh: &IcebergWarehouse,
    workspace_name: &str,
    jobs: &[JobRecord],
) -> Result<(Uuid, Uuid, Uuid)> {
    let (nodes, edges) = infra_from_jobs(jobs);
    let spans = spans_from_jobs(jobs);
    let node_snap = record_infra_nodes(wh, workspace_name, &nodes).await?;
    let edge_snap = record_infra_edges(wh, workspace_name, &edges).await?;
    let span_snap = record_runtime_spans(wh, workspace_name, &spans).await?;
    Ok((node_snap, edge_snap, span_snap))
}

/// Downcast a projected column **by name** (a `.select` scan reorders/drops
/// columns, so positional access would read the wrong array). Mirrors
/// `dep_graph::col`.
fn col<'a, T: 'static>(batch: &'a RecordBatch, name: &str) -> Result<&'a T> {
    batch
        .column_by_name(name)
        .ok_or_else(|| anyhow!("projected batch missing column `{name}`"))?
        .as_any()
        .downcast_ref::<T>()
        .ok_or_else(|| anyhow!("column `{name}` has unexpected arrow type"))
}

/// Sync wrapper: derive + persist the Yggdrasil infra/runtime scales from a job
/// ledger on this warehouse's private runtime. Returns the three snapshot ids.
impl IcebergWarehouse {
    pub fn record_yggdrasil_from_jobs(
        &self,
        workspace_name: &str,
        jobs: &[JobRecord],
    ) -> Result<(Uuid, Uuid, Uuid)> {
        self.block_on(ingest_from_jobs(self, workspace_name, jobs))
    }
}

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

    /// A job record with the fields the producer reads set explicitly.
    fn job(job_id: &str, kind: &str, target: &str, node: Option<&str>, parent: Option<&str>, elapsed_ms: Option<i64>, status: &str, start: i64) -> JobRecord {
        JobRecord {
            job_id: job_id.to_string(),
            kind: kind.to_string(),
            target: target.to_string(),
            workspace: "ws".to_string(),
            status: status.to_string(),
            ts_start_micros: start,
            ts_end_micros: elapsed_ms.map(|_| start + 1),
            elapsed_ms,
            detail_json: String::new(),
            result_ref: String::new(),
            parent_id: parent.map(|s| s.to_string()),
            node: node.map(|s| s.to_string()),
        }
    }

    fn sample_jobs() -> Vec<JobRecord> {
        vec![
            job("j1", "release", "nornir", Some("loki"), None, Some(1200), "done", 1000),
            job("j2", "bench", "nornir", Some("odin"), Some("j1"), Some(50), "done", 2000),
            job("j3", "release", "holger", Some("loki"), None, Some(900), "failed", 3000),
            // Unstamped node → LOCAL_MACHINE; still-running → dur 0.
            job("j4", "docs", "skade", None, None, None, "running", 4000),
        ]
    }

    #[test]
    fn producer_derives_machines_services_edges_and_spans() {
        let jobs = sample_jobs();
        let (nodes, edges) = infra_from_jobs(&jobs);

        // Machines: loki, odin, local (j4 unstamped). Services: release, bench, docs.
        let machines: BTreeSet<&str> =
            nodes.iter().filter(|n| n.kind == "machine").map(|n| n.id.as_str()).collect();
        assert_eq!(machines, ["loki", "local", "odin"].into_iter().collect());
        let services: BTreeSet<&str> =
            nodes.iter().filter(|n| n.kind == "service").map(|n| n.id.as_str()).collect();
        assert_eq!(services, ["service:release", "service:bench", "service:docs"].into_iter().collect());

        // `release` ran on BOTH loki and odin? no — release on loki (j1,j3), bench on
        // odin (j2), docs on local (j4). Assert the runs_on edges precisely.
        let runs_on: BTreeSet<(&str, &str)> = edges
            .iter()
            .filter(|e| e.rel == "runs_on")
            .map(|e| (e.from.as_str(), e.to.as_str()))
            .collect();
        assert_eq!(
            runs_on,
            [
                ("service:release", "loki"),
                ("service:bench", "odin"),
                ("service:docs", "local"),
            ]
            .into_iter()
            .collect()
        );
        // Every machine is member_of the constellation group.
        let member_of: BTreeSet<&str> = edges
            .iter()
            .filter(|e| e.rel == "member_of" && e.to == CONSTELLATION_GROUP)
            .map(|e| e.from.as_str())
            .collect();
        assert_eq!(member_of, ["loki", "odin", "local"].into_iter().collect());

        // Spans: one per job, machine + dur derived.
        let spans = spans_from_jobs(&jobs);
        assert_eq!(spans.len(), 4);
        let j2 = spans.iter().find(|s| s.span_id == "j2").unwrap();
        assert_eq!(j2.machine, "odin");
        assert_eq!(j2.parent, "j1");
        assert_eq!(j2.dur_us, 50_000, "elapsed_ms 50 → 50_000 us");
        let j4 = spans.iter().find(|s| s.span_id == "j4").unwrap();
        assert_eq!(j4.machine, "local", "unstamped node → local");
        assert_eq!(j4.dur_us, 0, "running job has no duration yet");
        assert_eq!(j4.parent, "", "no parent → empty");
    }

    /// RED-when-broken warehouse round-trip: record the ledger-derived crown +
    /// runtime band, read all three back, and assert field-identical to what the
    /// pure producer built. A wrong column mapping, a dropped row, or a lost
    /// numeric value flips an `assert_eq!`.
    #[test]
    fn ledger_scales_round_trip_field_identical() {
        let dir = tempfile::tempdir().unwrap();
        let wh = IcebergWarehouse::open(dir.path()).unwrap();
        let jobs = sample_jobs();

        let (mut want_nodes, want_edges) = infra_from_jobs(&jobs);
        let mut want_spans = spans_from_jobs(&jobs);

        wh.record_yggdrasil_from_jobs("wsA", &jobs).unwrap();

        // Nodes round-trip (sorted the same way the query sorts).
        want_nodes.sort_by(|a, b| (&a.kind, &a.id).cmp(&(&b.kind, &b.id)));
        let got_nodes = wh.block_on(query_infra_nodes(&wh, "wsA")).unwrap();
        assert_eq!(got_nodes, want_nodes, "infra_nodes byte/field-identical round-trip");

        // Edges round-trip.
        let mut want_edges = want_edges;
        want_edges.sort_by(|a, b| (&a.rel, &a.from, &a.to).cmp(&(&b.rel, &b.from, &b.to)));
        let got_edges = wh.block_on(query_infra_edges(&wh, "wsA")).unwrap();
        assert_eq!(got_edges, want_edges, "infra_edges byte/field-identical round-trip");

        // Spans round-trip — the numeric dur_us / start_micros survive exactly.
        want_spans.sort_by(|a, b| (a.start_micros, &a.span_id).cmp(&(b.start_micros, &b.span_id)));
        let got_spans = wh.block_on(query_runtime_spans(&wh, "wsA")).unwrap();
        assert_eq!(got_spans, want_spans, "runtime_spans byte/field-identical round-trip");

        // Mutation check: a DIFFERENT workspace sees none of wsA's rows.
        assert!(wh.block_on(query_infra_nodes(&wh, "wsB")).unwrap().is_empty());
        assert!(wh.block_on(query_infra_edges(&wh, "wsB")).unwrap().is_empty());
        assert!(wh.block_on(query_runtime_spans(&wh, "wsB")).unwrap().is_empty());

        // A specific numeric round-trips (guards a Long→Int truncation regression).
        let bench_span = got_spans.iter().find(|s| s.span_id == "j2").unwrap();
        assert_eq!(bench_span.dur_us, 50_000);
        assert_eq!(bench_span.machine, "odin");
    }

    #[test]
    fn empty_ledger_writes_no_rows() {
        let dir = tempfile::tempdir().unwrap();
        let wh = IcebergWarehouse::open(dir.path()).unwrap();
        wh.record_yggdrasil_from_jobs("empty", &[]).unwrap();
        assert!(wh.block_on(query_infra_nodes(&wh, "empty")).unwrap().is_empty());
        assert!(wh.block_on(query_infra_edges(&wh, "empty")).unwrap().is_empty());
        assert!(wh.block_on(query_runtime_spans(&wh, "empty")).unwrap().is_empty());
    }
}