nornir 0.5.0

//! Unified **job ledger** — the one durable list of every long-running nornir
//! operation (~1s–minutes): release runs, workspace fetch/republish, docs
//! render/export/book, knowledge scan, deepscan, bench, test matrix, arch
//! generate, index build, vector embed, bakeoff, airgap stages, …
//!
//! Today each op writes its *result* to a domain Iceberg table at the end
//! (`bench_runs`, `test_results`, `architecture_wiring`, `doc_exports`,
//! `clone_events`, …). What was missing is one place that answers **"what's
//! running / did it finish / how long did it take"** — the op *lifecycle*,
//! orthogonal to the result. That is this module.
//!
//! **Why redb, not Iceberg.** A job record is *mutable* (`running → done|failed`)
//! and short-lived; redb overwrites the key in place — no append-only
//! "two-rows-then-coalesce-on-read" dance, no per-event Iceberg snapshot
//! (a snapshot+manifest per `start`/`finish` is the small-files antipattern).
//! Iceberg stays the home for what an op *produced*; redb is the ledger that an
//! op *ran*. `result_ref` (e.g. `bench_runs:<uuid>`, `release_events:<run_id>`)
//! links a job back to its Iceberg result row.
//!
//! **Storage.** Its own redb instance at `<warehouse root>/jobs.redb` — distinct
//! from skade-katalog's `catalog.redb` and the server's `registry.redb`. Access
//! mirrors the warehouse exactly (see [`crate::warehouse`]):
//!
//! - **fat client** (local FS via library code, no gRPC): the [`JobHandle`]
//!   writes the local `jobs.redb` directly through a redb [`JobSink`].
//! - **thin client** (no local FS): the [`JobHandle`]'s sink submits each record
//!   over gRPC (`Jobs.Submit`); the server applies it to *its* `jobs.redb`.
//!
//! Single-writer discipline matches `registry.redb`: the owning process is the
//! sole writer; viz reads via [`JobStore::open_read_only`], which copies the
//! file aside when the server holds the lock (same trick as
//! [`crate::warehouse::IcebergWarehouse::open_read_only`]).

use std::path::Path;
use std::sync::Arc;

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

/// `job_id → serde_json(JobRecord)`. Bytes in, bytes out (like the registry).
const TABLE: TableDefinition<&str, &[u8]> = TableDefinition::new("jobs");

/// Retention applied opportunistically on terminal writes: drop terminal jobs
/// older than this many days, and cap the total row count (oldest terminal
/// pruned first). Keeps `jobs.redb` bounded on a long-running server.
pub const KEEP_DAYS: i64 = 30;
/// Hard cap on retained rows — see [`KEEP_DAYS`].
pub const CAP: usize = 2000;

/// Lifecycle status of a job. Mutable: the same key is overwritten in place as a
/// job advances. `queued` and `running` are the two non-terminal states.
///
/// A `queued` job has been admitted to the ledger but is **waiting on the serial
/// scheduler** (see [`is_exclusive`]) — another exclusive job (e.g. a `bench_run`)
/// is running and this one waits behind it. It transitions `queued → running`
/// when it acquires the exclusive gate, then `running → done|failed` as usual.
/// Non-exclusive jobs never pass through `queued`; they start `running` directly.
pub mod status {
    /// Admitted but waiting on the serial scheduler (an exclusive job is running
    /// ahead of this one). Non-terminal; visible in `nornir jobs` / the viz list.
    pub const QUEUED: &str = "queued";
    pub const RUNNING: &str = "running";
    pub const DONE: &str = "done";
    pub const FAILED: &str = "failed";
    /// Whether `s` is a terminal status (job will not change further).
    pub fn is_terminal(s: &str) -> bool {
        s == DONE || s == FAILED
    }
}

/// Canonical job kinds — one per op family. Strings (not an enum) so a new op
/// needs no schema change; these are the conventional tags readers group/filter on.
pub mod kind {
    pub const RELEASE_RUN: &str = "release_run";
    /// `nornir release undo` (task #32): a reversal run, tracked like a release.
    pub const RELEASE_UNDO: &str = "release_undo";
    pub const WORKSPACE_FETCH: &str = "workspace_fetch";
    pub const WORKSPACE_REPUBLISH: &str = "workspace_republish";
    /// The PARENT job of a workspace **populate** (register-eager): the umbrella op
    /// that clones every member then builds the warehouse. Its children stream in as
    /// [`WORKSPACE_CLONE`] (one per member, as each lands) plus a final
    /// [`WORKSPACE_REPUBLISH`] for the build. Readers fold the children under it.
    pub const WORKSPACE_POPULATE: &str = "workspace_populate";
    /// The clone/fetch of ONE member during a populate — a CHILD of a
    /// [`WORKSPACE_POPULATE`] parent (`parent_id` set). Streamed: emitted+finished as
    /// each member's clone completes, so the Jobs panel shows members appear
    /// running→done one at a time. Distinct from [`WORKSPACE_FETCH`] (the on-demand
    /// `Workspaces.Fetch` RPC over the whole workspace).
    pub const WORKSPACE_CLONE: &str = "workspace_clone";
    pub const DOCS_RENDER: &str = "docs_render";
    pub const DOCS_EXPORT: &str = "docs_export";
    pub const DOCS_BOOK: &str = "docs_book";
    pub const DOCS_BOOK_SVG: &str = "docs_book_svg";
    pub const KNOWLEDGE_SCAN: &str = "knowledge_scan";
    pub const SCIP_INGEST: &str = "scip_ingest";
    pub const DEEPSCAN: &str = "deepscan";
    /// An **index snapshot** of one repo's RA/SCIP index INTO the warehouse —
    /// `Index.Snapshot` / `Index.UploadSnapshot` (`snapshot_to_iceberg`). Distinct
    /// from [`INDEX_BUILD`] (building the on-disk index): this is the
    /// build-the-index → land-it-in-Iceberg op the snapshot RPCs run.
    pub const SNAPSHOT: &str = "snapshot";
    /// A **DWARF/symbol scan** of one built binary — `Introspect.Symbols` /
    /// `Introspect.SymbolLookup` / `Introspect.DefinedIn` (`extract_symbols`).
    /// Real work (reads + parses the artifact's debug info), tracked so the
    /// "symbol-scan" op family shows in the Jobs panel like every other scan.
    pub const SYMBOL_SCAN: &str = "symbol_scan";
    /// A **security / SBOM + vulnerability scan** of one workspace repo —
    /// `Mimir.SecurityScan` (`cargo metadata` SBOM + OSV query, persisting
    /// `vuln_findings`). Heavy (subprocess + network), so it earns a job.
    pub const SECURITY_SCAN: &str = "security_scan";
    pub const BENCH_RUN: &str = "bench_run";
    pub const TEST_MATRIX: &str = "test_matrix";
    pub const ARCH_GENERATE: &str = "arch_generate";
    pub const INDEX_BUILD: &str = "index_build";
    pub const VECTOR_EMBED: &str = "vector_embed";
    pub const BAKEOFF: &str = "bakeoff";
    pub const AIRGAP: &str = "airgap";
    /// `nornir coverage` — source line/region coverage measurement.
    pub const COVERAGE: &str = "coverage";
    /// A container run launched through the engine (`podman run …`), tracked like
    /// any other job: spawn → running, exit → done|failed, logs captured into
    /// `detail_json`. See [`super::run_container`].
    pub const CONTAINER: &str = "container";
    /// `nornir warehouse compact` — small-file COMPACTION: rewrite a table's
    /// (partition's) many small data files into a few large SORTED Parquet files
    /// and commit a replace snapshot. Real IO + a catalog commit, so it earns a
    /// job row in the Jobs panel.
    pub const WAREHOUSE_COMPACT: &str = "warehouse_compact";
    /// `nornir warehouse expire` — snapshot EXPIRY + orphan-file GC: drop old
    /// snapshots from the catalog and delete their now-unreferenced data/manifest
    /// files. Reclaims disk; tracked like every other server op.
    pub const WAREHOUSE_EXPIRE: &str = "warehouse_expire";
}

// ---------------------------------------------------------------------------
// Concurrency policy — how aggressively a kind may run alongside its peers.
// ---------------------------------------------------------------------------

/// Concurrency class of a job [`kind`] — the scheduler's policy for it.
///
/// The server is one long-lived process, so "how many of these may run at once"
/// is a process-wide question answered here. Today there are two classes; more
/// (e.g. a per-resource pool) can be added without touching call sites.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum JobClass {
    /// **Heavy / serial.** CPU- or IO-saturating work whose results are only
    /// meaningful in isolation — a benchmark sharing the box with another
    /// benchmark measures noise. At most **one** exclusive job runs process-wide
    /// at a time; the rest wait in `queued` (see [`status::QUEUED`]).
    Exclusive,
    /// **Parallel.** The default — runs immediately, never gated, unaffected by
    /// (and not affecting) the exclusive gate. Most ops (docs render, fetch,
    /// arch generate, container runs, …) are here.
    Parallel,
}

/// The [`JobClass`] for a job [`kind`]. The default is [`JobClass::Parallel`];
/// only genuinely box-saturating kinds whose measurement demands isolation are
/// [`JobClass::Exclusive`]. Today that is the benchmark runner ([`kind::BENCH_RUN`]):
/// a queued bench waits for the running one to finish so neither perturbs the
/// other's numbers. Add a kind here when it must not run concurrently with its
/// peers.
pub fn class_of(kind: &str) -> JobClass {
    match kind {
        kind::BENCH_RUN => JobClass::Exclusive,
        _ => JobClass::Parallel,
    }
}

/// `true` when `kind` is [`JobClass::Exclusive`] — i.e. the serial scheduler must
/// admit at most one such job at a time process-wide. Convenience over
/// [`class_of`].
pub fn is_exclusive(kind: &str) -> bool {
    matches!(class_of(kind), JobClass::Exclusive)
}

/// The process-global serial gate for [`JobClass::Exclusive`] jobs — a
/// `Semaphore` with a single permit. Held from the moment an exclusive job leaves
/// `queued` (acquires the permit, flips to `running`) until it reaches a terminal
/// state, so only one exclusive job runs at a time across the whole process.
/// `OnceLock` because the server is the long-lived owner and the gate must be the
/// same instance for every call site.
fn exclusive_gate() -> &'static Arc<tokio::sync::Semaphore> {
    static GATE: std::sync::OnceLock<Arc<tokio::sync::Semaphore>> = std::sync::OnceLock::new();
    GATE.get_or_init(|| Arc::new(tokio::sync::Semaphore::new(1)))
}

/// One job's lifecycle record. The redb value (serde_json). `detail_json` and
/// `result_ref` are free-form (`""` when absent) so the record never needs a
/// schema migration as new ops adopt the ledger.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct JobRecord {
    pub job_id: String,
    /// One of [`kind`].
    pub kind: String,
    /// Repo / workspace / artifact the job acts on (human-readable).
    pub target: String,
    pub workspace: String,
    /// One of [`status`]; overwritten in place as the job advances.
    pub status: String,
    pub ts_start_micros: i64,
    /// Set on the terminal write only.
    pub ts_end_micros: Option<i64>,
    /// `(ts_end - ts_start) / 1000`, set on the terminal write.
    pub elapsed_ms: Option<i64>,
    /// Free-form JSON detail (args / counts / error chain), or `""`.
    pub detail_json: String,
    /// Pointer to the domain result row this job produced, e.g.
    /// `"bench_runs:<uuid>"`, `"release_events:<run_id>"`, or `""`.
    pub result_ref: String,
    /// **Hierarchy** — the `job_id` of this job's PARENT, when it is a *sub-job*
    /// of a larger op (`None` = a top-level job). A multi-phase / multi-member op
    /// emits one parent job plus a child per phase/member: e.g. a workspace
    /// *populate* is a [`kind::WORKSPACE_POPULATE`] parent with a
    /// [`kind::WORKSPACE_CLONE`] child per member (streamed as each clone lands)
    /// and a [`kind::WORKSPACE_REPUBLISH`] child for the build. Readers (the Jobs
    /// panel / `Viz.Jobs` / `nornir jobs`) render children folded under their
    /// parent. `#[serde(default)]` so old rows (written before this field) decode
    /// as top-level — no migration.
    #[serde(default)]
    pub parent_id: Option<String>,
}

impl JobRecord {
    /// `true` when the job is in a terminal state.
    pub fn is_terminal(&self) -> bool {
        status::is_terminal(&self.status)
    }
}

/// Which jobs a read returns.
#[derive(Debug, Clone)]
pub enum JobSelector {
    /// Every job in the ledger.
    All,
    /// Only jobs for this workspace.
    Workspace(String),
    /// Only jobs of this [`kind`].
    Kind(String),
}

impl JobSelector {
    fn matches(&self, r: &JobRecord) -> bool {
        match self {
            JobSelector::All => true,
            JobSelector::Workspace(w) => &r.workspace == w,
            JobSelector::Kind(k) => &r.kind == k,
        }
    }
}

fn now_micros() -> i64 {
    chrono::Utc::now().timestamp_micros()
}

// ---------------------------------------------------------------------------
// JobSink — where a JobHandle's records go. Fat = redb; thin = a submit closure
// (the gRPC client, supplied by the binary so the library stays tonic-free).
// ---------------------------------------------------------------------------

/// The write side of a [`JobHandle`]. Cheap to clone (`Arc`). A fat client makes
/// one over the local `jobs.redb` ([`JobStore::sink`]); a thin client makes one
/// from a closure that submits the record over gRPC ([`JobSink::remote`]).
#[derive(Clone)]
pub struct JobSink(Arc<dyn Fn(&JobRecord) + Send + Sync>);

impl JobSink {
    /// A redb-backed sink: upserts each record into `db`, opportunistically
    /// pruning on terminal writes. Best-effort — a write failure is logged and
    /// swallowed (a ledger hiccup must never abort the op it tracks).
    pub fn redb(db: Arc<Database>) -> Self {
        JobSink(Arc::new(move |rec: &JobRecord| {
            if let Err(e) = redb_upsert(&db, rec) {
                eprintln!("   ⚠ jobs: dropped {} row for {}/{} (non-fatal): {e:#}", rec.status, rec.kind, rec.target);
                return;
            }
            if rec.is_terminal() {
                if let Err(e) = redb_prune(&db, KEEP_DAYS, CAP) {
                    eprintln!("   ⚠ jobs: prune failed (non-fatal): {e:#}");
                }
            }
        }))
    }

    /// A sink that hands each record to `f` — the thin-client path, where `f`
    /// performs the `Jobs.Submit` RPC. The library never links tonic; the binary
    /// supplies the closure.
    pub fn remote(f: impl Fn(&JobRecord) + Send + Sync + 'static) -> Self {
        JobSink(Arc::new(f))
    }

    /// A sink that drops every record — for ops that have no ledger configured
    /// (keeps the `start`/`finish` call sites unconditional).
    pub fn noop() -> Self {
        JobSink(Arc::new(|_| {}))
    }

    fn upsert(&self, rec: &JobRecord) {
        (self.0)(rec)
    }
}

// ---------------------------------------------------------------------------
// JobHandle — wraps one op's lifecycle. start → (finish | fail | Drop=failed).
// ---------------------------------------------------------------------------

/// Tracks a single op: writes a `running` record on [`start`](JobHandle::start),
/// a terminal record on [`finish`](JobHandle::finish)/[`fail`](JobHandle::fail),
/// and — via [`Drop`] — a `failed` record if neither was called (so a panic or
/// early `?` return never leaves a zombie `running` row).
pub struct JobHandle {
    sink: JobSink,
    rec: JobRecord,
    finished: bool,
    /// The exclusive-gate permit, held for an [`JobClass::Exclusive`] job from the
    /// moment it leaves `queued` until it terminates (then dropped, admitting the
    /// next queued exclusive job). `None` for parallel jobs and for jobs not run
    /// through the scheduler.
    _gate_permit: Option<tokio::sync::OwnedSemaphorePermit>,
}

impl JobHandle {
    /// Begin a job: build the `running` record and write it through `sink`.
    /// `detail` is any JSON (args / counts); pass [`serde_json::Value::Null`]
    /// for none.
    pub fn start(
        sink: JobSink,
        kind: &str,
        target: &str,
        workspace: &str,
        detail: serde_json::Value,
    ) -> Self {
        let rec = Self::new_record(kind, target, workspace, detail, status::RUNNING, None);
        sink.upsert(&rec);
        JobHandle { sink, rec, finished: false, _gate_permit: None }
    }

    /// Begin a **child** job under `parent_id` — a sub-job of a larger op (e.g. one
    /// member's clone under a workspace-populate parent). Identical to
    /// [`start`](Self::start) but stamps `parent_id` so readers fold it under its
    /// parent. Children are emitted+finished one at a time as the op streams, so the
    /// Jobs panel shows them appear progressively.
    pub fn start_child(
        sink: JobSink,
        parent_id: &str,
        kind: &str,
        target: &str,
        workspace: &str,
        detail: serde_json::Value,
    ) -> Self {
        let rec = Self::new_record(
            kind,
            target,
            workspace,
            detail,
            status::RUNNING,
            Some(parent_id.to_string()),
        );
        sink.upsert(&rec);
        JobHandle { sink, rec, finished: false, _gate_permit: None }
    }

    /// **Serial scheduler entry.** Begin a job that respects its [`class_of`]
    /// concurrency policy:
    ///
    /// - [`JobClass::Parallel`] kinds behave exactly like [`start`](Self::start):
    ///   the record is written `running` and the handle returns immediately.
    /// - [`JobClass::Exclusive`] kinds (e.g. [`kind::BENCH_RUN`]) are admitted to
    ///   the ledger as [`status::QUEUED`] first, then this future **waits** on the
    ///   process-global [`exclusive_gate`]. If another exclusive job is running,
    ///   the row stays `queued` (visible in `nornir jobs` / the viz list) until
    ///   that job finishes; only then does it flip `queued → running` and the
    ///   work proceeds. The returned handle holds the gate permit until it reaches
    ///   a terminal state, so the next queued exclusive job waits behind it.
    ///
    /// Always resolves to a `running` [`JobHandle`]; the caller does its work and
    /// calls [`finish`](Self::finish)/[`fail`](Self::fail) as usual.
    pub async fn start_scheduled(
        sink: JobSink,
        kind: &str,
        target: &str,
        workspace: &str,
        detail: serde_json::Value,
    ) -> Self {
        if !is_exclusive(kind) {
            return Self::start(sink, kind, target, workspace, detail);
        }
        // Exclusive: publish a `queued` row so the wait is visible, then block on
        // the single-permit gate. The row sits `queued` for the whole wait.
        let mut rec = Self::new_record(kind, target, workspace, detail, status::QUEUED, None);
        sink.upsert(&rec);
        let permit = exclusive_gate()
            .clone()
            .acquire_owned()
            .await
            .expect("exclusive job gate is never closed");
        // Acquired — promote queued → running in place (same job_id key), and
        // re-stamp the start time so `elapsed_ms` measures actual work, not the
        // queue wait.
        rec.status = status::RUNNING.to_string();
        rec.ts_start_micros = now_micros();
        sink.upsert(&rec);
        JobHandle { sink, rec, finished: false, _gate_permit: Some(permit) }
    }

    /// Build a fresh `JobRecord` in `initial_status` (no sink write).
    fn new_record(
        kind: &str,
        target: &str,
        workspace: &str,
        detail: serde_json::Value,
        initial_status: &str,
        parent_id: Option<String>,
    ) -> JobRecord {
        JobRecord {
            job_id: uuid::Uuid::new_v4().to_string(),
            kind: kind.to_string(),
            target: target.to_string(),
            workspace: workspace.to_string(),
            status: initial_status.to_string(),
            ts_start_micros: now_micros(),
            ts_end_micros: None,
            elapsed_ms: None,
            detail_json: detail_to_string(detail),
            result_ref: String::new(),
            parent_id,
        }
    }

    /// The job's id — stamp it into a domain `result_ref` for a bidirectional
    /// link if desired.
    pub fn job_id(&self) -> &str {
        &self.rec.job_id
    }

    /// Terminal success. `result_ref` points at the domain row produced (or `""`).
    pub fn finish(mut self, detail: serde_json::Value, result_ref: &str) {
        self.terminate(status::DONE, detail_to_string(detail), result_ref.to_string());
    }

    /// Terminal failure, carrying the error chain into `detail_json`.
    pub fn fail(mut self, err: &anyhow::Error) {
        let detail = serde_json::json!({ "error": format!("{err:#}") }).to_string();
        self.terminate(status::FAILED, detail, String::new());
    }

    /// Terminal failure carrying a caller-supplied `detail` JSON instead of an
    /// error chain — for ops that "ran but the result is a failure" (e.g. a
    /// container that exited non-zero: the spawn/wait succeeded, so there is no
    /// `anyhow::Error`, but the captured logs/exit_code belong on the row).
    pub fn fail_with_detail(mut self, detail: serde_json::Value) {
        self.terminate(status::FAILED, detail_to_string(detail), String::new());
    }

    fn terminate(&mut self, st: &str, detail_json: String, result_ref: String) {
        let end = now_micros();
        self.rec.status = st.to_string();
        self.rec.ts_end_micros = Some(end);
        self.rec.elapsed_ms = Some((end - self.rec.ts_start_micros) / 1000);
        self.rec.detail_json = detail_json;
        self.rec.result_ref = result_ref;
        self.sink.upsert(&self.rec);
        self.finished = true;
    }
}

impl Drop for JobHandle {
    fn drop(&mut self) {
        if !self.finished {
            let detail =
                serde_json::json!({ "error": "job dropped without finish/fail (panic or early return)" })
                    .to_string();
            self.terminate(status::FAILED, detail, String::new());
        }
    }
}

fn detail_to_string(detail: serde_json::Value) -> String {
    if detail.is_null() {
        String::new()
    } else {
        detail.to_string()
    }
}

// ---------------------------------------------------------------------------
// JobStore — the redb file. Fat writer / lock-tolerant reader.
// ---------------------------------------------------------------------------

/// The redb-backed job ledger at `<root>/jobs.redb`. Open once; the owning
/// process is the sole writer (single-writer discipline, like `registry.redb`).
pub struct JobStore {
    db: Arc<Database>,
    /// Holds a copied-aside snapshot's temp dir alive (set by
    /// [`open_read_only`](Self::open_read_only) when the live file was locked).
    _snapshot: Option<tempfile::TempDir>,
}

impl JobStore {
    /// Open (creating if absent) the ledger at `<root>/jobs.redb` for read+write.
    pub fn open(root: &Path) -> Result<Self> {
        std::fs::create_dir_all(root)
            .with_context(|| format!("create jobs root {}", root.display()))?;
        let path = root.join("jobs.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: Arc::new(db), _snapshot: None })
    }

    /// Lock-tolerant read open. Like [`open`](Self::open), but if `jobs.redb` is
    /// already locked exclusively by another process (the live `nornir-server` —
    /// redb is single-writer), it copies the file aside into a temp dir and opens
    /// *that*, yielding a point-in-time read-only snapshot. Mirrors
    /// [`crate::warehouse::IcebergWarehouse::open_read_only`]; the viz local-mode
    /// feed uses this so it coexists with a running server instead of deadlocking.
    pub fn open_read_only(root: &Path) -> Result<Self> {
        match Self::open(root) {
            Ok(s) => Ok(s),
            Err(e) if is_lock_error(&e) => {
                let live = root.join("jobs.redb");
                if !live.exists() {
                    return Err(e);
                }
                let tmp = tempfile::Builder::new()
                    .prefix("nornir-jobs-snapshot-")
                    .tempdir()
                    .context("create temp dir for jobs snapshot")?;
                let snap = tmp.path().join("jobs.redb");
                copy_redb_consistent(&live, &snap).with_context(|| {
                    format!("copy locked jobs {} -> snapshot {}", live.display(), snap.display())
                })?;
                let db = Database::create(&snap)
                    .with_context(|| format!("open jobs snapshot {}", snap.display()))?;
                Ok(Self { db: Arc::new(db), _snapshot: Some(tmp) })
            }
            Err(e) => Err(e),
        }
    }

    /// A redb sink over this store — the fat-client write path.
    pub fn sink(&self) -> JobSink {
        JobSink::redb(self.db.clone())
    }

    /// Begin a job whose lifecycle writes land in this store (fat path).
    pub fn start(
        &self,
        kind: &str,
        target: &str,
        workspace: &str,
        detail: serde_json::Value,
    ) -> JobHandle {
        JobHandle::start(self.sink(), kind, target, workspace, detail)
    }

    /// Begin a **child** job under `parent_id` in this store — see
    /// [`JobHandle::start_child`]. Used to stream the per-member / per-phase
    /// sub-jobs of an umbrella op (a workspace populate) under its parent.
    pub fn start_child(
        &self,
        parent_id: &str,
        kind: &str,
        target: &str,
        workspace: &str,
        detail: serde_json::Value,
    ) -> JobHandle {
        JobHandle::start_child(self.sink(), parent_id, kind, target, workspace, detail)
    }

    /// Begin a job through the **serial scheduler** (fat path) — see
    /// [`JobHandle::start_scheduled`]. Exclusive kinds (e.g. [`kind::BENCH_RUN`])
    /// are admitted `queued` and wait on the process-global gate; parallel kinds
    /// start `running` immediately. `.await` resolves to a `running` handle.
    pub async fn start_scheduled(
        &self,
        kind: &str,
        target: &str,
        workspace: &str,
        detail: serde_json::Value,
    ) -> JobHandle {
        JobHandle::start_scheduled(self.sink(), kind, target, workspace, detail).await
    }

    /// Apply a record submitted by a thin client (`Jobs.Submit`): upsert it and
    /// prune on terminal writes. The server's write seam.
    pub fn submit(&self, rec: &JobRecord) -> Result<()> {
        redb_upsert(&self.db, rec)?;
        if rec.is_terminal() {
            redb_prune(&self.db, KEEP_DAYS, CAP)?;
        }
        Ok(())
    }

    /// Read jobs matching `sel`, newest-first by start time.
    pub fn list(&self, sel: &JobSelector) -> Result<Vec<JobRecord>> {
        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?;
            let rec: JobRecord =
                serde_json::from_slice(v.value()).context("decode job record")?;
            if sel.matches(&rec) {
                out.push(rec);
            }
        }
        out.sort_by(|a, b| b.ts_start_micros.cmp(&a.ts_start_micros));
        Ok(out)
    }

    /// Retention: drop terminal jobs older than `keep_days`, then — if more than
    /// `cap` rows remain — drop the oldest terminal jobs down to `cap`. Running
    /// jobs are never pruned. Returns how many were removed.
    pub fn prune(&self, keep_days: i64, cap: usize) -> Result<usize> {
        redb_prune(&self.db, keep_days, cap)
    }
}

/// Upsert one record (key = `job_id`), overwriting any prior state for that job.
fn redb_upsert(db: &Database, rec: &JobRecord) -> Result<()> {
    let bytes = serde_json::to_vec(rec).context("encode job record")?;
    let w = db.begin_write()?;
    {
        let mut t = w.open_table(TABLE)?;
        t.insert(rec.job_id.as_str(), bytes.as_slice())?;
    }
    w.commit()?;
    Ok(())
}

/// See [`JobStore::prune`].
fn redb_prune(db: &Database, keep_days: i64, cap: usize) -> Result<usize> {
    // Snapshot all rows first (read txn), decide what to drop, then remove in one
    // write txn — keeps the borrow of the read table from overlapping the write.
    let all: Vec<JobRecord> = {
        let r = db.begin_read()?;
        let t = r.open_table(TABLE)?;
        let mut v = Vec::new();
        for row in t.iter()? {
            let (_k, val) = row?;
            v.push(serde_json::from_slice::<JobRecord>(val.value()).context("decode job record")?);
        }
        v
    };

    let cutoff = now_micros() - keep_days * 86_400 * 1_000_000;
    let mut victims: Vec<String> = Vec::new();

    // 1) Aged-out terminal jobs.
    for rec in &all {
        if rec.is_terminal() && rec.ts_end_micros.unwrap_or(rec.ts_start_micros) < cutoff {
            victims.push(rec.job_id.clone());
        }
    }

    // 2) Over-cap: drop the oldest terminal jobs (by start time) not already a
    //    victim, until the surviving count is within `cap`. Running jobs stay.
    let surviving = all.len().saturating_sub(victims.len());
    if surviving > cap {
        let mut terminal_left: Vec<&JobRecord> = all
            .iter()
            .filter(|r| r.is_terminal() && !victims.contains(&r.job_id))
            .collect();
        terminal_left.sort_by(|a, b| a.ts_start_micros.cmp(&b.ts_start_micros)); // oldest first
        let need = surviving - cap;
        for rec in terminal_left.into_iter().take(need) {
            victims.push(rec.job_id.clone());
        }
    }

    if victims.is_empty() {
        return Ok(0);
    }
    let w = db.begin_write()?;
    {
        let mut t = w.open_table(TABLE)?;
        for id in &victims {
            t.remove(id.as_str())?;
        }
    }
    w.commit()?;
    Ok(victims.len())
}

/// True when `err` (anywhere in its chain) is redb's exclusive-lock rejection —
/// another process holds the single-writer lock on `jobs.redb`. Matches on the
/// message (redb's `DatabaseAlreadyOpen` / lock wording), like
/// [`crate::warehouse::is_catalog_lock_error`].
fn is_lock_error(err: &anyhow::Error) -> bool {
    err.chain().any(|e| {
        let m = e.to_string();
        m.contains("Database already open") || m.contains("Cannot acquire lock")
    })
}

/// Copy a locked `live` redb file → `dst` across a **size-stable window**, so the
/// copy is a complete, openable image even while the owner is committing. redb
/// grows the file in regions (header advertises the larger layout before the file
/// is extended); a blind copy can capture a file shorter than its own header
/// claims and trip redb's length invariant on open. Same rationale and fix as
/// `warehouse::copy_catalog_consistent` — copy, re-stat, retry until the source
/// length is unchanged across the copy and the copy is not short.
fn copy_redb_consistent(live: &Path, dst: &Path) -> Result<()> {
    const MAX_ATTEMPTS: usize = 64;
    let len_of = |p: &Path| -> Result<u64> {
        Ok(std::fs::metadata(p).with_context(|| format!("stat {}", p.display()))?.len())
    };
    for attempt in 0..MAX_ATTEMPTS {
        let before = len_of(live)?;
        std::fs::copy(live, dst)
            .with_context(|| format!("copy {} -> {}", live.display(), dst.display()))?;
        let after = len_of(live)?;
        if before == after && len_of(dst)? >= after {
            return Ok(());
        }
        std::thread::sleep(std::time::Duration::from_millis(2 + attempt as u64 / 4));
    }
    anyhow::bail!(
        "snapshot of locked jobs db {} never reached a size-stable window after {MAX_ATTEMPTS} attempts",
        live.display()
    )
}

// ---------------------------------------------------------------------------
// Container jobs — run a container (podman/docker) AS a tracked nornir job.
// ---------------------------------------------------------------------------

/// Spec for a [`run_container`] job: the engine binary (`podman`/`docker`), the
/// image, and the argv passed *after* the image. Everything is data so a caller
/// (CLI, server RPC) can build it without linking any container library — the
/// engine is invoked as a subprocess, exactly like `cargo`/`git` elsewhere.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct ContainerSpec {
    /// The container engine binary, e.g. `"podman"` or `"docker"`.
    pub engine: String,
    /// The image reference, e.g. `"docker.io/library/alpine:3"`.
    pub image: String,
    /// Command + args run inside the container (after the image), e.g.
    /// `["echo", "hi"]`. Empty = the image's default entrypoint.
    pub cmd: Vec<String>,
    /// Extra flags inserted between `run` and the image, e.g. `["--rm"]`,
    /// `["-e", "FOO=bar"]`. `--rm` is recommended so nothing leaks.
    pub run_flags: Vec<String>,
}

impl ContainerSpec {
    /// `podman run --rm <image> <cmd…>`.
    pub fn podman(image: &str, cmd: &[&str]) -> Self {
        ContainerSpec {
            engine: "podman".into(),
            image: image.into(),
            cmd: cmd.iter().map(|s| s.to_string()).collect(),
            run_flags: vec!["--rm".into()],
        }
    }

    /// The full argv this spec runs (`<engine> run <flags…> <image> <cmd…>`),
    /// for display/inspection.
    pub fn argv(&self) -> Vec<String> {
        let mut v = vec![self.engine.clone(), "run".into()];
        v.extend(self.run_flags.iter().cloned());
        v.push(self.image.clone());
        v.extend(self.cmd.iter().cloned());
        v
    }
}

/// Outcome of a container job: exit code (None = killed by signal) plus the
/// captured stdout/stderr (truncated to keep `detail_json` bounded).
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct ContainerOutcome {
    pub exit_code: Option<i32>,
    pub stdout: String,
    pub stderr: String,
}

/// stdout/stderr captured per stream is truncated to this many bytes so a chatty
/// container can't bloat the redb value. The head is kept (where errors usually are).
const LOG_CAP_BYTES: usize = 16 * 1024;

fn truncate_log(mut s: String) -> String {
    if s.len() > LOG_CAP_BYTES {
        s.truncate(LOG_CAP_BYTES);
        s.push_str("\n…[truncated]");
    }
    s
}

/// Run a container **as a tracked nornir job**: open a `running` row, spawn
/// `<engine> run …`, wait for it, capture logs + exit status, then write the
/// terminal row (`done` on exit 0, else `failed`). Returns the [`ContainerOutcome`]
/// even on a non-zero exit (the *job* failed but the call succeeds — the caller
/// inspects `exit_code`); only a spawn/wait error returns `Err` (and the
/// [`JobHandle`] [`Drop`] guard then records `failed`).
///
/// `target` is a human label for the job row (e.g. the image name); `workspace`
/// scopes it like every other job. The pid is stamped into the `running` row's
/// `detail_json` for liveness, then the logs replace it on the terminal write.
///
/// Synchronous and blocking by design — it mirrors how nornir already shells out
/// to `cargo`/`git`. A caller wanting it off the request thread spawns a thread
/// (the server does this for bench/test today).
pub fn run_container(
    sink: JobSink,
    spec: &ContainerSpec,
    target: &str,
    workspace: &str,
) -> Result<ContainerOutcome> {
    let argv = spec.argv();
    let job = JobHandle::start(
        sink,
        kind::CONTAINER,
        target,
        workspace,
        serde_json::json!({ "argv": argv, "engine": spec.engine, "image": spec.image }),
    );

    // Spawn `<engine> run <flags> <image> <cmd…>`, capturing both streams.
    let mut command = std::process::Command::new(&spec.engine);
    command.arg("run");
    command.args(&spec.run_flags);
    command.arg(&spec.image);
    command.args(&spec.cmd);
    command.stdout(std::process::Stdio::piped());
    command.stderr(std::process::Stdio::piped());

    let child = match command.spawn() {
        Ok(c) => c,
        Err(e) => {
            // Spawn failed (engine missing?). Record the job failed, surface Err.
            let err = anyhow::anyhow!(e).context(format!(
                "spawn container engine `{}` (is it installed and on PATH?)",
                spec.engine
            ));
            job.fail(&err);
            return Err(err);
        }
    };
    let pid = child.id();

    let output = match child.wait_with_output() {
        Ok(o) => o,
        Err(e) => {
            let err = anyhow::anyhow!(e).context("wait for container");
            job.fail(&err);
            return Err(err);
        }
    };

    let exit_code = output.status.code();
    let outcome = ContainerOutcome {
        exit_code,
        stdout: truncate_log(String::from_utf8_lossy(&output.stdout).into_owned()),
        stderr: truncate_log(String::from_utf8_lossy(&output.stderr).into_owned()),
    };

    let detail = serde_json::json!({
        "argv": argv,
        "pid": pid,
        "exit_code": exit_code,
        "stdout": outcome.stdout,
        "stderr": outcome.stderr,
    });

    if output.status.success() {
        job.finish(detail, "");
    } else {
        // Non-zero exit / signal: terminal `failed`, but the call itself succeeded
        // (the caller reads `outcome.exit_code`). Carry the captured logs.
        job.fail_with_detail(detail);
    }
    Ok(outcome)
}

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

    fn tmpdir(tag: &str) -> std::path::PathBuf {
        std::env::temp_dir().join(format!("nornir-jobs-{tag}-{}", std::process::id()))
    }

    #[test]
    fn start_finish_fail_and_drop_zombie_guard() {
        let dir = tmpdir("life");
        std::fs::remove_dir_all(&dir).ok();
        let store = JobStore::open(&dir).unwrap();

        // 1) finish → done, with a result_ref.
        let h = store.start(kind::BENCH_RUN, "facett", "nordisk", serde_json::json!({"n": 1}));
        let id_done = h.job_id().to_string();
        h.finish(serde_json::json!({"rows": 12}), "bench_runs:abc");

        // 2) fail → failed, with the error chain.
        let h = store.start(kind::TEST_MATRIX, "holger", "nordisk", serde_json::Value::Null);
        h.fail(&anyhow::anyhow!("boom"));

        // 3) dropped without terminal → failed (zombie guard).
        let id_drop = {
            let h = store.start(kind::DOCS_BOOK_SVG, "nornir", "nordisk", serde_json::Value::Null);
            let id = h.job_id().to_string();
            drop(h);
            id
        };

        let all = store.list(&JobSelector::All).unwrap();
        assert_eq!(all.len(), 3, "three jobs recorded");
        // Newest-first by start; statuses all terminal.
        assert!(all.iter().all(|r| r.is_terminal()), "no zombie running rows");

        let done = all.iter().find(|r| r.job_id == id_done).unwrap();
        assert_eq!(done.status, status::DONE);
        assert_eq!(done.result_ref, "bench_runs:abc");
        assert!(done.elapsed_ms.is_some(), "finished job has elapsed");
        assert!(done.ts_end_micros.is_some());

        let dropped = all.iter().find(|r| r.job_id == id_drop).unwrap();
        assert_eq!(dropped.status, status::FAILED, "dropped handle records failed");
        assert!(dropped.detail_json.contains("dropped"), "drop reason carried");

        // Selector scoping.
        assert_eq!(store.list(&JobSelector::Kind(kind::BENCH_RUN.into())).unwrap().len(), 1);
        assert_eq!(store.list(&JobSelector::Workspace("nordisk".into())).unwrap().len(), 3);
        assert_eq!(store.list(&JobSelector::Workspace("other".into())).unwrap().len(), 0);

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

    #[test]
    fn running_then_done_overwrites_in_place_no_duplicate() {
        let dir = tmpdir("overwrite");
        std::fs::remove_dir_all(&dir).ok();
        let store = JobStore::open(&dir).unwrap();

        let h = store.start(kind::RELEASE_RUN, "ws", "ws", serde_json::Value::Null);
        let id = h.job_id().to_string();
        // While running, exactly one row, status running.
        let running = store.list(&JobSelector::All).unwrap();
        assert_eq!(running.len(), 1);
        assert_eq!(running[0].status, status::RUNNING);
        assert!(running[0].elapsed_ms.is_none());

        h.finish(serde_json::Value::Null, "release_events:r1");
        // Still one row (same job_id key), now done.
        let done = store.list(&JobSelector::All).unwrap();
        assert_eq!(done.len(), 1, "finish overwrote the running row, not appended");
        assert_eq!(done[0].job_id, id);
        assert_eq!(done[0].status, status::DONE);
        assert_eq!(done[0].result_ref, "release_events:r1");

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

    #[test]
    fn prune_drops_aged_terminal_and_caps_count_but_keeps_running() {
        let dir = tmpdir("prune");
        std::fs::remove_dir_all(&dir).ok();
        let store = JobStore::open(&dir).unwrap();

        // Insert directly via redb_upsert (NOT submit, which auto-prunes on
        // terminal writes) so we exercise `prune` itself.
        // An aged-out terminal job (ended well before the cutoff).
        let old_end = now_micros() - (KEEP_DAYS + 5) * 86_400 * 1_000_000;
        redb_upsert(
            &store.db,
            &JobRecord {
                job_id: "old".into(),
                kind: kind::DOCS_BOOK.into(),
                target: "t".into(),
                workspace: "w".into(),
                status: status::DONE.into(),
                ts_start_micros: old_end - 1000,
                ts_end_micros: Some(old_end),
                elapsed_ms: Some(1),
                detail_json: String::new(),
                result_ref: String::new(),
                parent_id: None,
            },
        )
        .unwrap();
        // A fresh running job (must survive pruning).
        redb_upsert(
            &store.db,
            &JobRecord {
                job_id: "live".into(),
                kind: kind::BENCH_RUN.into(),
                target: "t".into(),
                workspace: "w".into(),
                status: status::RUNNING.into(),
                ts_start_micros: now_micros(),
                ts_end_micros: None,
                elapsed_ms: None,
                detail_json: String::new(),
                result_ref: String::new(),
                parent_id: None,
            },
        )
        .unwrap();

        let removed = store.prune(KEEP_DAYS, CAP).unwrap();
        assert_eq!(removed, 1, "only the aged terminal job is pruned");
        let left = store.list(&JobSelector::All).unwrap();
        assert_eq!(left.len(), 1);
        assert_eq!(left[0].job_id, "live", "the running job survives retention");

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

    #[test]
    fn container_spec_argv_is_engine_run_flags_image_cmd() {
        let spec = ContainerSpec::podman("alpine:3", &["echo", "hi"]);
        assert_eq!(
            spec.argv(),
            vec!["podman", "run", "--rm", "alpine:3", "echo", "hi"]
        );
        // Round-trips as JSON (it rides in detail_json / RPC payloads).
        let s = serde_json::to_string(&spec).unwrap();
        assert_eq!(serde_json::from_str::<ContainerSpec>(&s).unwrap(), spec);
    }

    #[test]
    fn container_job_missing_engine_records_failed_and_errs() {
        // No real container engine needed: a bogus engine name → spawn fails →
        // the job row is `failed` and the call returns Err (inject-assert: we
        // feed a known-bad engine and assert both the Err and the ledger row).
        let dir = tmpdir("container-missing");
        std::fs::remove_dir_all(&dir).ok();
        let store = JobStore::open(&dir).unwrap();

        let spec = ContainerSpec {
            engine: "nornir-no-such-container-engine".into(),
            image: "alpine".into(),
            cmd: vec!["true".into()],
            run_flags: vec!["--rm".into()],
        };
        let res = run_container(store.sink(), &spec, "alpine", "ws");
        assert!(res.is_err(), "missing engine surfaces as Err");

        let jobs = store.list(&JobSelector::Kind(kind::CONTAINER.into())).unwrap();
        assert_eq!(jobs.len(), 1, "one container job recorded");
        assert_eq!(jobs[0].status, status::FAILED, "spawn failure → failed row");
        assert!(jobs[0].is_terminal());
        assert!(jobs[0].detail_json.contains("error"), "error carried in detail");
        std::fs::remove_dir_all(&dir).ok();
    }

    /// Real container run, gated on podman being present (this box has it per
    /// [[this-box-is-oden]]). Skips silently in CI without a container engine.
    #[test]
    fn container_job_real_podman_echo_tracks_done() {
        if std::process::Command::new("podman")
            .arg("--version")
            .stdout(std::process::Stdio::null())
            .stderr(std::process::Stdio::null())
            .status()
            .map(|s| !s.success())
            .unwrap_or(true)
        {
            eprintln!("skip: podman not available");
            return;
        }
        let dir = tmpdir("container-podman");
        std::fs::remove_dir_all(&dir).ok();
        let store = JobStore::open(&dir).unwrap();

        // `podman run --rm alpine echo nornir-jobs-ok` — short, deterministic.
        let spec = ContainerSpec::podman(
            "docker.io/library/alpine:3",
            &["echo", "nornir-jobs-ok"],
        );
        let outcome = run_container(store.sink(), &spec, "alpine:3", "ws")
            .expect("container ran");
        assert_eq!(outcome.exit_code, Some(0), "echo exits 0");
        assert!(
            outcome.stdout.contains("nornir-jobs-ok"),
            "container stdout captured: {:?}",
            outcome.stdout
        );

        let jobs = store.list(&JobSelector::Kind(kind::CONTAINER.into())).unwrap();
        assert_eq!(jobs.len(), 1);
        assert_eq!(jobs[0].status, status::DONE, "exit 0 → done");
        assert!(jobs[0].elapsed_ms.is_some());
        assert!(jobs[0].detail_json.contains("nornir-jobs-ok"), "logs on row");
        std::fs::remove_dir_all(&dir).ok();
    }

    #[test]
    fn truncate_log_caps_and_marks() {
        let big = "x".repeat(LOG_CAP_BYTES + 5000);
        let out = truncate_log(big);
        assert!(out.len() <= LOG_CAP_BYTES + 32, "capped near LOG_CAP_BYTES");
        assert!(out.ends_with("…[truncated]"), "truncation is marked");
        // A short log is returned verbatim.
        assert_eq!(truncate_log("ok".to_string()), "ok");
    }

    /// Real container that exits non-zero → `fail_with_detail` path: the call
    /// SUCCEEDS (we get the outcome) but the ledger row is `failed`, carrying the
    /// exit code + logs (no anyhow error). Gated on podman.
    #[test]
    fn container_job_nonzero_exit_records_failed_keeps_outcome() {
        if std::process::Command::new("podman")
            .arg("--version")
            .stdout(std::process::Stdio::null())
            .stderr(std::process::Stdio::null())
            .status()
            .map(|s| !s.success())
            .unwrap_or(true)
        {
            eprintln!("skip: podman not available");
            return;
        }
        let dir = tmpdir("container-nonzero");
        std::fs::remove_dir_all(&dir).ok();
        let store = JobStore::open(&dir).unwrap();

        // `sh -c 'exit 7'` → exit code 7, job failed but the call returns Ok.
        let spec = ContainerSpec::podman(
            "docker.io/library/alpine:3",
            &["sh", "-c", "exit 7"],
        );
        let outcome = run_container(store.sink(), &spec, "alpine:3", "ws")
            .expect("spawn/wait succeeded even though the container failed");
        assert_eq!(outcome.exit_code, Some(7), "non-zero exit captured");

        let jobs = store.list(&JobSelector::Kind(kind::CONTAINER.into())).unwrap();
        assert_eq!(jobs.len(), 1);
        assert_eq!(jobs[0].status, status::FAILED, "non-zero exit → failed row");
        assert!(jobs[0].detail_json.contains("\"exit_code\":7"), "exit code on row: {}", jobs[0].detail_json);
        std::fs::remove_dir_all(&dir).ok();
    }

    #[test]
    fn classify_bench_exclusive_others_parallel() {
        assert!(is_exclusive(kind::BENCH_RUN), "bench is the heavy/serial kind");
        assert_eq!(class_of(kind::BENCH_RUN), JobClass::Exclusive);
        // Everything else stays parallel (the default) for now.
        for k in [
            kind::RELEASE_RUN,
            kind::TEST_MATRIX,
            kind::ARCH_GENERATE,
            kind::COVERAGE,
            kind::CONTAINER,
            kind::DOCS_BOOK,
            kind::WORKSPACE_FETCH,
        ] {
            assert!(!is_exclusive(k), "{k} should be parallel");
            assert_eq!(class_of(k), JobClass::Parallel, "{k}");
        }
    }

    /// The scheduler's headline guarantee: two `BENCH_RUN` jobs **serialize**. The
    /// second is admitted `queued` while the first is `running`, and only flips to
    /// `running` once the first reaches a terminal state. We drive it explicitly:
    /// hold the first handle (so it never releases the gate), assert the second is
    /// stuck `queued`, then finish the first and observe the second run.
    /// The exclusive gate is process-global, so the two scheduler tests below
    /// must not overlap (each fully drives the single permit through acquire →
    /// release). This std `Mutex` serializes them within the test binary.
    static SCHED_TEST_LOCK: std::sync::Mutex<()> = std::sync::Mutex::new(());

    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn exclusive_bench_jobs_serialize_second_observes_queued() {
        let _guard = SCHED_TEST_LOCK.lock().unwrap_or_else(|e| e.into_inner());
        let dir = tmpdir("sched-serialize");
        std::fs::remove_dir_all(&dir).ok();
        let store = Arc::new(JobStore::open(&dir).unwrap());

        // First bench acquires the gate and is `running`.
        let first = store
            .start_scheduled(kind::BENCH_RUN, "alpha", "ws", serde_json::Value::Null)
            .await;
        let first_id = first.job_id().to_string();
        assert_eq!(
            store
                .list(&JobSelector::Kind(kind::BENCH_RUN.into()))
                .unwrap()
                .iter()
                .find(|r| r.job_id == first_id)
                .unwrap()
                .status,
            status::RUNNING,
            "first bench runs immediately"
        );

        // Submit a second bench on a task — it must block on the gate, sitting
        // `queued` (NOT running, NOT invisible) while `first` holds the permit.
        let store2 = store.clone();
        let second = tokio::spawn(async move {
            let h = store2
                .start_scheduled(kind::BENCH_RUN, "beta", "ws", serde_json::Value::Null)
                .await;
            let id = h.job_id().to_string();
            h.finish(serde_json::Value::Null, "");
            id
        });

        // Give the spawned task time to write its `queued` row and block.
        let second_id = poll_until(&store, |recs| {
            recs.iter().find(|r| r.target == "beta").map(|r| {
                assert_eq!(r.status, status::QUEUED, "second bench is queued behind the first");
                r.job_id.clone()
            })
        })
        .await;

        // The second is definitely NOT running while the first holds the gate.
        let recs = store.list(&JobSelector::Kind(kind::BENCH_RUN.into())).unwrap();
        let running: Vec<_> = recs.iter().filter(|r| r.status == status::RUNNING).collect();
        assert_eq!(running.len(), 1, "exactly one bench runs at a time");
        assert_eq!(running[0].job_id, first_id);

        // Release the gate: finish the first. The second must now run and finish.
        first.finish(serde_json::Value::Null, "");
        let got_id = second.await.unwrap();
        assert_eq!(got_id, second_id, "the queued bench is the one that ran");

        let final_recs = store.list(&JobSelector::Kind(kind::BENCH_RUN.into())).unwrap();
        for r in &final_recs {
            assert_eq!(r.status, status::DONE, "both benches finished: {r:?}");
        }
        std::fs::remove_dir_all(&dir).ok();
    }

    /// A non-exclusive (parallel) job is **not** blocked by a running exclusive
    /// job: with a bench holding the gate, a `release_run` still starts `running`
    /// immediately (never `queued`).
    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn parallel_job_not_blocked_by_running_exclusive() {
        let _guard = SCHED_TEST_LOCK.lock().unwrap_or_else(|e| e.into_inner());
        let dir = tmpdir("sched-parallel");
        std::fs::remove_dir_all(&dir).ok();
        let store = JobStore::open(&dir).unwrap();

        // Hold an exclusive bench (occupies the single gate permit).
        let bench = store
            .start_scheduled(kind::BENCH_RUN, "alpha", "ws", serde_json::Value::Null)
            .await;

        // A parallel job goes straight to `running` despite the bench.
        let other = store
            .start_scheduled(kind::RELEASE_RUN, "rel", "ws", serde_json::Value::Null)
            .await;
        let other_rec = store
            .list(&JobSelector::Kind(kind::RELEASE_RUN.into()))
            .unwrap()
            .into_iter()
            .next()
            .unwrap();
        assert_eq!(other_rec.status, status::RUNNING, "parallel job is never gated");

        other.finish(serde_json::Value::Null, "");
        bench.finish(serde_json::Value::Null, "");
        std::fs::remove_dir_all(&dir).ok();
    }

    /// Poll the ledger until `pick` returns `Some`, briefly yielding between reads.
    /// Bounded so a logic bug fails the test instead of hanging it.
    async fn poll_until<T>(
        store: &JobStore,
        pick: impl Fn(&[JobRecord]) -> Option<T>,
    ) -> T {
        for _ in 0..500 {
            let recs = store.list(&JobSelector::All).unwrap();
            if let Some(v) = pick(&recs) {
                return v;
            }
            tokio::time::sleep(std::time::Duration::from_millis(5)).await;
        }
        panic!("condition not reached within timeout");
    }

    #[test]
    fn job_record_vec_json_round_trips() {
        // The RPC contract (Viz.Jobs / Jobs.Submit) ships JobRecord as JSON.
        let recs = vec![JobRecord {
            job_id: "j".into(),
            kind: kind::ARCH_GENERATE.into(),
            target: "skade".into(),
            workspace: "nordisk".into(),
            status: status::DONE.into(),
            ts_start_micros: 1,
            ts_end_micros: Some(2),
            elapsed_ms: Some(0),
            detail_json: "{\"nodes\":310}".into(),
            result_ref: "architecture_wiring:xyz".into(),
            parent_id: None,
        }];
        let s = serde_json::to_string(&recs).unwrap();
        let back: Vec<JobRecord> = serde_json::from_str(&s).unwrap();
        assert_eq!(recs, back);
    }

    /// Parent/child hierarchy: a populate-style parent with two streamed children.
    /// Each child carries `parent_id`, and the ledger stores+returns all three.
    #[test]
    fn child_jobs_carry_parent_id_and_persist() {
        let dir = tmpdir("hierarchy");
        std::fs::remove_dir_all(&dir).ok();
        let store = JobStore::open(&dir).unwrap();

        // Parent: the umbrella populate job (stays running while children stream).
        let parent = store.start(
            kind::WORKSPACE_POPULATE,
            "nordisk",
            "nordisk",
            serde_json::json!({ "phase": "populate" }),
        );
        let pid = parent.job_id().to_string();

        // Child 1 streams in and finishes (one member's clone).
        let c1 = store.start_child(&pid, kind::WORKSPACE_CLONE, "alpha", "nordisk", serde_json::Value::Null);
        let c1_id = c1.job_id().to_string();
        c1.finish(serde_json::json!({ "sha": "abc" }), "");

        // Child 2 streams in and finishes (second member's clone).
        let c2 = store.start_child(&pid, kind::WORKSPACE_CLONE, "beta", "nordisk", serde_json::Value::Null);
        c2.finish(serde_json::json!({ "sha": "def" }), "");

        // Build child under the same parent.
        let build = store.start_child(&pid, kind::WORKSPACE_REPUBLISH, "nordisk", "nordisk", serde_json::Value::Null);
        build.finish(serde_json::json!({ "snapshot": "snap-1" }), "");

        parent.finish(serde_json::json!({ "members": 2 }), "");

        let all = store.list(&JobSelector::Workspace("nordisk".into())).unwrap();
        assert_eq!(all.len(), 4, "parent + 2 clone children + 1 build child");

        // The parent is top-level.
        let p = all.iter().find(|r| r.job_id == pid).unwrap();
        assert_eq!(p.parent_id, None, "the populate parent has no parent");
        assert_eq!(p.kind, kind::WORKSPACE_POPULATE);

        // Every non-parent row is a child of the parent.
        let children: Vec<&JobRecord> = all.iter().filter(|r| r.parent_id.as_deref() == Some(pid.as_str())).collect();
        assert_eq!(children.len(), 3, "two clones + one build are children of the populate");
        assert!(children.iter().any(|r| r.job_id == c1_id && r.kind == kind::WORKSPACE_CLONE));
        assert!(children.iter().all(|r| r.is_terminal()), "all children terminal");

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