nornir 0.4.49

//! 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. `running` is the only non-terminal state.
pub mod status {
    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";
    pub const WORKSPACE_FETCH: &str = "workspace_fetch";
    pub const WORKSPACE_REPUBLISH: &str = "workspace_republish";
    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 DEEPSCAN: &str = "deepscan";
    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";
}

/// 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,
}

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,
}

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 = JobRecord {
            job_id: uuid::Uuid::new_v4().to_string(),
            kind: kind.to_string(),
            target: target.to_string(),
            workspace: workspace.to_string(),
            status: status::RUNNING.to_string(),
            ts_start_micros: now_micros(),
            ts_end_micros: None,
            elapsed_ms: None,
            detail_json: detail_to_string(detail),
            result_ref: String::new(),
        };
        sink.upsert(&rec);
        JobHandle { sink, rec, finished: false }
    }

    /// 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());
    }

    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)
    }

    /// 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()
    )
}

#[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(),
            },
        )
        .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(),
            },
        )
        .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 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(),
        }];
        let s = serde_json::to_string(&recs).unwrap();
        let back: Vec<JobRecord> = serde_json::from_str(&s).unwrap();
        assert_eq!(recs, back);
    }
}