aprender-contracts 0.69.2

//! ONT-001 §3.7 — `extract:release-evidence` (aprender#3715): the release's receipts become a graph in which
//! every obligation is a NODE, so absence is a missing edge that `minCount 1` rejects — never a row nobody read.
//!
//! Operator, 2026-09-21: *"you need to improve our dogfood process and pv SHACL so leaks don't exist pre-release."*
//! Every 0.69.0 leak was absence read as conformance (a rung marked optional, a GPU error read as a skip, an
//! inherited dogfood, models on a host that were not on the ladder). So the universe is DERIVED here, from what
//! was measured, and each (thing × host × verb × context) the release must prove gets a node of its own:
//!
//! | node | IRI | from |
//! |---|---|---|
//! | `release:Release` | `release-subject/<v>` | the CLI's `--release-*` flags (never inferred) |
//! | `release:DogfoodReceipt` | `release-dogfood/<v>` | `--dogfood-receipt`: the file R5 judged |
//! | `release:Host` | `release-host/<v>/<host>` | every `ladder.hosts[]` entry with `required: true` |
//! | `release:HostReceipt` | `release-receipt/<v>/<host>/<file>` | the host's `apr-model-ladder-receipt` (v2, #3712) |
//! | `release:Model` | `model/<sha256>` (shared with `extract:gguf`) | the host's measured `inventory[]` ∪ the ladder's cuda rungs listed for it |
//! | `release:Verb` | `release-verb/<verb>` | run, chat, serve, code |
//! | `release:ContextRung` | `release-context/<id>` | `evidence/release/context-rungs.json`; `consumer-max` is COMPUTED from the consumer records, and `declared` is each model's own GGUF `context_length` |
//! | `release:Cell` | `release-cell/<v>/<host>/<file>/<verb>/think-<on\|off>/<rung>` | model × verb × thinking × rung, per host — the focus of `release-readiness-v1` |
//! | `release:CellReceipt` | `release-row/<v>/<host>/<file>/<i>` | each `cells[]` row that keys onto a cell |
//! | `release:RefusalCell` | as `release:Cell` | a cell that does not FIT the host by the declared memory arithmetic: owed as an honest pre-load refusal (#3710 rule) |
//! | `release:RungCoverage` | `release-coverage/<v>/<file>/<rung>` | every (model, rung) — owed on ≥ 1 required host |
//! | `release:KernelCell` | `release-kernel/<v>/<host>/<kernel>/<quant>` | every (kernel, quant) on any required host's dispatch path × every required host |
//! | `release:KernelDiffReceipt` | `release-krow/<v>/<host>/<file>/<i>` | each kernel-diff row |
//! | `release:TokenizerCell` | `release-tokenizer/<v>/<file>` | every distinct model in any required host's universe (#3726) |
//! | `release:TokenizerParityReceipt` | `release-trow/<v>/<file>/<i>` | each tokenizer-parity row keyed onto a model |
//!
//! The cell universe per model (#3710, operator: "MORE than enough tokens … chat with and without thinking, ditto
//! run, ditto code"): every verb × thinking ∈ {on, off} (only `off` when the host MEASURED no thinking mode) ×
//! every declared rung up to the model's own `context_length`, plus `declared` itself. A model whose length was
//! not measured owes every rung, and the `.model` shape names the missing length.
//!
//! Rust computes only what SHACL-Core cannot say (the universe, the join, `fresh`, `contextMet`, `thinkOk`,
//! `answered`, `withinBound`);
//! the shapes in `contracts/release-readiness-v1.yaml` say what a pass IS, and the fixtures under
//! `tests/fixtures/ont/release-*` falsify both. Nothing here runs unless a release subject is given: an ordinary
//! PR has no release, and inventing one would grade a tree against a commit it is not.

use std::collections::{BTreeMap, BTreeSet};
use std::path::Path;

use crate::ontology::extract::gguf::{self, Rung};
use crate::ontology::extract::pv_contract::{self, scalar};
use crate::ontology::extract::release_inputs::{
    self as inputs, Consumer, ContextRung, Dogfood, KernelReceipt, ReleaseError, Subject,
    TokReceipt,
};
use crate::ontology::rdf::{iri, iri_path, Graph, Term, ONT_BASE, RDF_TYPE};
use crate::ontology::receipts::{self, CellRow, Receipt};

/// The verbs every model must pass on every required host (#3710: "run, chat, serve and code").
pub const VERBS: [&str; 4] = ["run", "chat", "serve", "code"];
/// Both thinking modes (#3710 bar raise: "chat with and without thinking, ditto run, ditto code").
pub const THINKING: [&str; 2] = ["on", "off"];
/// The per-model rung: the model's own declared context length, read from its GGUF on the host.
pub use crate::ontology::extract::release_inputs::DECLARED;

/// The vocabulary root: `https://ont.paiml.dev/v1alpha1/release/<name>` (`release:` in a shape).
#[must_use]
pub fn rel(name: &str) -> String {
    format!("{ONT_BASE}release/{name}")
}

/// What one run derived, for the gate's report.
#[derive(Debug, Clone, Default, PartialEq, Eq, serde::Serialize)]
pub struct ReleaseStats {
    pub required_hosts: usize,
    pub model_receipts: usize,
    pub kernel_receipts: usize,
    /// Distinct (host, model) pairs in the universe.
    pub models: usize,
    pub cells: usize,
    /// Cells with at least one keyed row (a row is not a pass — the shape decides that).
    pub cells_with_row: usize,
    pub kernel_cells: usize,
    /// One per distinct model (sha256) in any required host's universe (#3726).
    pub tokenizer_cells: usize,
    /// Cells owed as an honest pre-load refusal because they do not fit the host (counted inside `cells`).
    pub refusal_cells: usize,
    /// (model, rung) pairs that must be owed on ≥ 1 required host (#3710 rule).
    pub rung_coverage: usize,
    pub context_rungs: usize,
    /// Rows that keyed onto no cell (a model outside the universe, an unknown verb or rung): counted, not graded.
    pub orphan_rows: usize,
    /// Every cell, named `<host>/<file>/<verb>/think-<mode>/<rung>` — so a Pass reads "N/N cells", each named
    /// (#3715 done_when 5), and a reader can see which cells a verdict covered, not only how many.
    pub cell_names: Vec<String>,
}

/// A `ladder.hosts[]` entry with `required: true`.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct HostDecl {
    pub id: String,
    pub cc: String,
}

/// `ladder.cells.long_rungs_for` (#3712 amendment 2): the families that owe every long rung, and one
/// representative file per other architecture. Absent from the ladder → every model owes the long rungs.
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct LongRungsFor {
    pub families: BTreeSet<String>,
    /// `general.architecture` → the one file basename that owes the long rungs for that arch.
    pub representatives: BTreeMap<String, String>,
}

impl LongRungsFor {
    fn owes(&self, arch: Option<&str>, file: &str) -> bool {
        arch.is_some_and(|a| {
            self.families.contains(a) || self.representatives.get(a).is_some_and(|f| f == file)
        })
    }
}

/// What the ladder contract(s) declare for the release.
#[derive(Debug, Clone, Default)]
pub struct Ladder {
    pub hosts: Vec<HostDecl>,
    pub rungs: Vec<Rung>,
    pub long_rungs_for: Option<LongRungsFor>,
}

/// One model the release must prove on one host.
#[derive(Debug, Clone, PartialEq, Eq)]
struct ModelInfo {
    file: String,
    arch: Option<String>,
    quant: Option<String>,
    context_length: Option<u64>,
    thinking_modes: Option<Vec<String>>,
    thinking_markers: Option<Vec<String>>,
    /// The producer's echo, compared with `owes_long`.
    owes_long_echo: Option<bool>,
    /// Derived here, from the ladder contract.
    owes_long: bool,
    /// (weights, KV bytes per token, workspace) — the declared memory arithmetic, when all three were measured.
    mem: Option<(u64, u64, u64)>,
    kv_dtype: Option<String>,
}

impl ModelInfo {
    /// Does a `tokens`-long context fit on a host with `gpu` bytes (#3710 rule, operator "a")? `None` when any
    /// term is unmeasured — and an unknown fit leaves the cell OWED, the stricter reading.
    fn fits(&self, gpu: Option<u64>, tokens: Option<u64>) -> Option<bool> {
        Some(self.required_bytes(tokens?)? <= gpu?)
    }

    fn required_bytes(&self, tokens: u64) -> Option<u64> {
        let (w, kv, ws) = self.mem?;
        Some(
            w.saturating_add(kv.saturating_mul(tokens))
                .saturating_add(ws),
        )
    }

    /// `off` always; `on` unless the host MEASURED that the model has no thinking mode.
    /// The modes the model owes: its measured `thinking_modes` (#3723); BOTH when unmeasured — and both when the
    /// measured list names no valid mode, so a malformed list cannot make a model vanish (the `.model` shape names
    /// it as a contradiction instead).
    fn modes(&self) -> Vec<&'static str> {
        let owed: Vec<&'static str> = match &self.thinking_modes {
            Some(ms) => THINKING
                .iter()
                .copied()
                .filter(|t| ms.iter().any(|m| m == t))
                .collect(),
            None => THINKING.to_vec(),
        };
        if owed.is_empty() {
            THINKING.to_vec()
        } else {
            owed
        }
    }

    /// The rungs this model owes: every declared rung whose size fits its declared length (all of them while the
    /// length is unmeasured, or the rung's own size is), then `declared`, whose target IS the length.
    fn rungs<'r>(&self, rungs: &'r [ContextRung]) -> Vec<(&'r str, Option<u64>)> {
        let fits = |r: &&ContextRung| match (r.tokens, self.context_length) {
            (Some(t), Some(l)) => t <= l,
            _ => true,
        };
        let mut out: Vec<(&str, Option<u64>)> = rungs
            .iter()
            .filter(|r| self.owes_long || !r.long)
            .filter(fits)
            .map(|r| (r.id.as_str(), r.tokens))
            .collect();
        if self.owes_long {
            out.push((DECLARED, self.context_length));
        }
        out
    }

    /// `thinking_modes` must be what its template evidence can prove (#3723's derivation): `enable_thinking`
    /// among the markers ⇔ both modes; no marker at all ⇒ `["off"]`; a list naming no valid mode is itself a
    /// contradiction. (`["on"]` — a prompt that always opens `<think>` — is the producer's to prove; pv checks only
    /// what the markers decide.) `None` when both are unmeasured or they agree.
    fn thinking_contradiction(&self) -> Option<String> {
        let modes = self.thinking_modes.as_ref()?;
        let valid: Vec<&String> = modes
            .iter()
            .filter(|m| THINKING.contains(&m.as_str()))
            .collect();
        let both = THINKING.iter().all(|t| modes.iter().any(|m| m == t));
        let markers = self.thinking_markers.as_deref().unwrap_or_default();
        let switch = markers.iter().any(|m| m == "enable_thinking");
        let bad = valid.is_empty()
            || valid.len() != modes.len()
            || switch != both
            || (self.thinking_markers.is_some() && markers.is_empty() && modes != &["off"]);
        bad.then(|| {
            format!(
                "thinking_modes = [{}] but thinking_markers = [{}]",
                modes.join(", "),
                markers.join(", ")
            )
        })
    }
}

/// One required host, with everything derived for it.
struct HostView<'a> {
    decl: &'a HostDecl,
    node: String,
    receipts: Vec<&'a Receipt>,
    kernels: Vec<&'a KernelReceipt>,
    /// sha256 → model, the host's universe.
    models: BTreeMap<String, ModelInfo>,
    /// The host's measured GPU memory (the fit arithmetic's right-hand side).
    gpu_mem: Option<u64>,
}

/// The required hosts, the rungs and `cells.long_rungs_for`, from every contract that carries a `ladder:` block.
#[must_use]
pub fn ladder_of(contract_dir: &Path) -> Ladder {
    let mut out = Ladder::default();
    for (stem, _rel, doc) in pv_contract::documents(contract_dir) {
        let Some(l) = doc.get("ladder") else {
            continue;
        };
        let decls = l.get("hosts").and_then(serde_yaml::Value::as_sequence);
        for h in decls.into_iter().flatten() {
            if h.get("required").and_then(serde_yaml::Value::as_bool) == Some(true) {
                out.hosts.push(HostDecl {
                    id: scalar(h.get("id")).unwrap_or_default(),
                    cc: scalar(h.get("cc")).unwrap_or_default(),
                });
            }
        }
        out.rungs.extend(gguf::rungs_of(&stem, &doc));
        if let Some(lr) = l.get("cells").and_then(|c| c.get("long_rungs_for")) {
            out.long_rungs_for = Some(long_rungs_for(lr));
        }
    }
    out.hosts.sort_by(|a, b| a.id.cmp(&b.id));
    out.hosts.dedup_by(|a, b| a.id == b.id);
    out
}

fn long_rungs_for(v: &serde_yaml::Value) -> LongRungsFor {
    let families = v
        .get("families")
        .and_then(serde_yaml::Value::as_sequence)
        .into_iter()
        .flatten()
        .filter_map(|f| scalar(Some(f)))
        .collect();
    let representatives = v
        .get("representatives")
        .and_then(serde_yaml::Value::as_mapping)
        .into_iter()
        .flatten()
        .filter_map(|(k, f)| Some((scalar(Some(k))?, scalar(Some(f))?)))
        .collect();
    LongRungsFor {
        families,
        representatives,
    }
}

/// Read the release's evidence and write its graph into `g`. `Err` is a declaration's fault (exit 3).
pub fn extract(
    g: &mut Graph,
    contract_dir: &Path,
    subject: &Subject,
) -> Result<ReleaseStats, ReleaseError> {
    let root = super::repo_root(contract_dir);
    let ladder = ladder_of(contract_dir);
    let (ctx_rungs, consumers) = inputs::read_context_rungs(&root)?.unwrap_or_default();
    let model_receipts =
        receipts::read_dir(&subject.model_dir(&root), &root).map_err(ReleaseError::Receipt)?;
    let kernel_receipts = inputs::read_kernel_receipts(&subject.kernel_dir(&root), &root)?;
    let tok_receipts = inputs::read_tokenizer_receipts(&subject.tokenizer_dir(&root), &root)?;
    let dogfood = subject
        .dogfood_receipt
        .as_deref()
        .map(inputs::read_dogfood)
        .transpose()?;
    Ok(build(
        g,
        subject,
        &Inputs {
            ladder: &ladder,
            ctx_rungs: &ctx_rungs,
            consumers: &consumers,
            model_receipts: &model_receipts,
            kernel_receipts: &kernel_receipts,
            tok_receipts: &tok_receipts,
            dogfood: dogfood.as_ref(),
        },
    ))
}

/// Everything [`build`] reads, already parsed — so the positive control drives the same code in memory.
pub struct Inputs<'a> {
    pub ladder: &'a Ladder,
    pub ctx_rungs: &'a [ContextRung],
    pub consumers: &'a [Consumer],
    pub model_receipts: &'a [Receipt],
    pub kernel_receipts: &'a [KernelReceipt],
    pub tok_receipts: &'a [TokReceipt],
    pub dogfood: Option<&'a Dogfood>,
}

/// The release graph from parsed inputs: pure, no filesystem (R-15).
pub fn build(g: &mut Graph, subject: &Subject, i: &Inputs<'_>) -> ReleaseStats {
    let hosts = &i.ladder.hosts;
    let mut stats = ReleaseStats {
        required_hosts: hosts.len(),
        model_receipts: i.model_receipts.len(),
        kernel_receipts: i.kernel_receipts.len(),
        context_rungs: i.ctx_rungs.len(),
        ..ReleaseStats::default()
    };
    emit_release(g, subject, hosts, i.ctx_rungs, i.dogfood);
    emit_vocabulary_nodes(g, i.ctx_rungs, i.consumers);
    let views: Vec<HostView<'_>> = hosts
        .iter()
        .map(|h| host_view(h, subject, i.model_receipts, i.kernel_receipts, i.ladder))
        .collect();
    let mut coverage = Coverage::new();
    for v in &views {
        emit_host(g, subject, v, &mut stats);
        emit_cells(g, subject, v, i.ctx_rungs, &mut coverage, &mut stats);
    }
    emit_coverage(g, subject, &coverage, &mut stats);
    emit_kernels(g, subject, &views, &mut stats);
    emit_tokenizer(g, subject, &views, i.tok_receipts, &mut stats);
    stats
}

/// The positive control (R-3, PMAT-3704): drawn on EVERY gate run, with or without a release subject. One
/// required host holds one model that owes one cell; the SAMPLE receipt carries that cell's fresh Pass row, the
/// PLANTED one omits it. Fires iff the sample's cell has exactly one fresh row AND the planted cell still exists
/// with ZERO rows — i.e. the extractor still materializes absence as a node for `minCount 1` to reject, which is
/// the whole reason this shape exists. An extractor that emitted only cells it had rows for would turn absence
/// back into silence, and this control would go `not-fired`.
#[must_use]
pub fn positive_control() -> bool {
    let Ok(subject) = Subject::new("0.0.0-pc", PC_COMMIT) else {
        return false;
    };
    let (Ok(sample), Ok(planted)) = (
        receipts::parse("__pc_sample__.json", &pc_receipt(PC_ROW)),
        receipts::parse("__pc_planted__.json", &pc_receipt("")),
    ) else {
        return false;
    };
    let cell = iri_path(
        "release-cell",
        &["0.0.0-pc", "pc-host", "pc.gguf", "run", "think-off", "pc"],
    );
    let rows_of = |r: &Receipt| {
        let mut g = Graph::new();
        let ladder = Ladder {
            hosts: vec![HostDecl {
                id: "pc-host".into(),
                cc: "sm_0".into(),
            }],
            ..Ladder::default()
        };
        let rung = ContextRung {
            id: "pc".into(),
            tokens: Some(1),
            long: false,
            derived_from: vec!["pc".into()],
            unmeasured_consumers: Vec::new(),
        };
        build(
            &mut g,
            &subject,
            &Inputs {
                ladder: &ladder,
                ctx_rungs: std::slice::from_ref(&rung),
                consumers: &[],
                model_receipts: std::slice::from_ref(r),
                kernel_receipts: &[],
                tok_receipts: &[],
                dogfood: None,
            },
        );
        let is_cell = g
            .objects(&cell, RDF_TYPE)
            .iter()
            .any(|t| t.as_iri() == Some(rel("Cell").as_str()));
        let fresh: Vec<bool> = g
            .objects(&cell, &rel("row"))
            .iter()
            .filter_map(|t| t.as_iri())
            .map(|row| {
                g.objects(row, &rel("fresh"))
                    .iter()
                    .any(|t| t.as_literal().map(|(v, _)| v) == Some("true"))
            })
            .collect();
        (is_cell, fresh)
    };
    let (sample_cell, sample_rows) = rows_of(&sample);
    let (planted_cell, planted_rows) = rows_of(&planted);
    sample_cell && sample_rows == [true] && planted_cell && planted_rows.is_empty()
}

const PC_COMMIT: &str = "1111111111111111111111111111111111111111";
const PC_ROW: &str = r#"{"sha256":"cccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccc","file":"pc.gguf","verb":"run","thinking":"off","context":"pc","prompt_tokens":2,"max_tokens":8,"answer_chars":1,"verdict":"pass","backend":"cuda","fallback":false,"rc":0}"#;

fn pc_receipt(cells: &str) -> String {
    format!(
        r#"{{"schema":"apr-model-ladder-receipt/v2","host":"pc-host","version":"0.0.0-pc","sha":"1","apr_sha":"{PC_COMMIT}","inventory":[{{"file":"pc.gguf","sha256":"cccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccc","arch":"pc","context_length":10,"thinking_modes":["off"]}}],"cells":[{cells}],"rungs":[]}}"#
    )
}

fn emit_release(
    g: &mut Graph,
    subject: &Subject,
    hosts: &[HostDecl],
    rungs: &[ContextRung],
    dogfood: Option<&Dogfood>,
) {
    let v = subject.version.as_str();
    let n = iri_path("release-subject", &[v]);
    g.insert(n.clone(), RDF_TYPE, Term::iri(rel("Release")));
    g.insert(n.clone(), rel("version"), Term::string(v));
    g.insert(n.clone(), rel("commit"), Term::string(&subject.commit));
    g.insert(
        n.clone(),
        rel("measuredCommit"),
        Term::string(subject.measured_commit()),
    );
    for h in hosts {
        g.insert(
            n.clone(),
            rel("requiredHost"),
            Term::iri(iri_path("release-host", &[v, &h.id])),
        );
    }
    for r in rungs {
        g.insert(
            n.clone(),
            rel("contextRung"),
            Term::iri(iri("release-context", &r.id)),
        );
    }
    if let Some(d) = dogfood {
        let dn = iri_path("release-dogfood", &[v]);
        g.insert(dn.clone(), RDF_TYPE, Term::iri(rel("DogfoodReceipt")));
        g.insert(dn.clone(), rel("file"), Term::string(&d.file));
        g.insert(dn.clone(), rel("verdict"), Term::string(&d.verdict));
        g.insert(dn.clone(), rel("commit"), Term::string(&d.commit));
        g.insert(dn.clone(), rel("version"), Term::string(&d.version));
        let fresh = d.commit == subject.commit && d.version == subject.version;
        g.insert(dn.clone(), rel("fresh"), Term::boolean(fresh));
        g.insert(n, rel("dogfoodReceipt"), Term::iri(dn));
    }
}

/// The verbs, the context rungs and the consumers they were derived from.
fn emit_vocabulary_nodes(g: &mut Graph, rungs: &[ContextRung], consumers: &[Consumer]) {
    for verb in VERBS {
        let n = iri("release-verb", verb);
        g.insert(n.clone(), RDF_TYPE, Term::iri(rel("Verb")));
        g.insert(n, rel("name"), Term::string(verb));
    }
    // `declared` has no one size — each model's own length is its target — so it is not a `release:ContextRung`
    // (whose shape requires a `tokens`); the model's `contextLength` carries it, and `.model` requires that.
    let declared = iri("release-context", DECLARED);
    g.insert(declared.clone(), RDF_TYPE, Term::iri(rel("DeclaredRung")));
    g.insert(declared.clone(), rel("rungId"), Term::string(DECLARED));
    g.insert(
        declared,
        rel("derivedFrom"),
        Term::string("each model's own GGUF context_length, measured on the host (#3710)"),
    );
    for r in rungs {
        let n = iri("release-context", &r.id);
        g.insert(n.clone(), RDF_TYPE, Term::iri(rel("ContextRung")));
        g.insert(n.clone(), rel("rungId"), Term::string(&r.id));
        if let Some(t) = r.tokens {
            g.insert(n.clone(), rel("tokens"), Term::integer(t));
        }
        for d in &r.derived_from {
            g.insert(n.clone(), rel("derivedFrom"), Term::string(d));
        }
        for c in &r.unmeasured_consumers {
            g.insert(n.clone(), rel("unmeasuredConsumer"), Term::string(c));
        }
    }
    for c in consumers {
        let n = iri("release-consumer", &c.name);
        g.insert(n.clone(), RDF_TYPE, Term::iri(rel("Consumer")));
        if let Some(t) = c.max_prompt_tokens {
            g.insert(n.clone(), rel("maxPromptTokens"), Term::integer(t));
        }
        g.insert(n.clone(), rel("basis"), Term::string(&c.basis));
        g.insert(n, rel("source"), Term::string(&c.source));
    }
}

/// The host's receipts and its universe: every inventory model with a measured hash, plus every ladder rung that
/// claims cuda and lists this host (or lists none). Keyed by sha256, so a copy under a second name is one model.
fn host_view<'a>(
    decl: &'a HostDecl,
    subject: &Subject,
    model_receipts: &'a [Receipt],
    kernel_receipts: &'a [KernelReceipt],
    ladder: &Ladder,
) -> HostView<'a> {
    let receipts: Vec<&Receipt> = model_receipts
        .iter()
        .filter(|r| r.host == decl.id)
        .collect();
    let kernels: Vec<&KernelReceipt> = kernel_receipts
        .iter()
        .filter(|r| r.host == decl.id)
        .collect();
    let mut models: BTreeMap<String, ModelInfo> = BTreeMap::new();
    for item in receipts.iter().flat_map(|r| &r.inventory) {
        if let Some(sha) = &item.sha256 {
            models.entry(sha.clone()).or_insert_with(|| ModelInfo {
                file: item.file.clone(),
                arch: item.arch.clone(),
                quant: item.quant.clone(),
                context_length: item.context_length,
                thinking_modes: item.thinking_modes.clone(),
                thinking_markers: item.thinking_markers.clone(),
                owes_long_echo: item.owes_long_rungs,
                owes_long: true,
                mem: item
                    .weights_bytes
                    .zip(item.kv_bytes_per_token)
                    .zip(item.workspace_bytes)
                    .map(|((w, kv), ws)| (w, kv, ws)),
                kv_dtype: item.kv_dtype.clone(),
            });
        }
    }
    let listed = |r: &&Rung| r.hosts.is_empty() || r.hosts.iter().any(|h| h == &decl.id);
    for r in ladder
        .rungs
        .iter()
        .filter(listed)
        .filter(|r| r.backends.iter().any(|b| b == "cuda"))
    {
        models
            .entry(r.sha256.to_ascii_lowercase())
            .or_insert_with(|| ModelInfo {
                file: r.gguf.clone(),
                arch: Some(r.arch.clone()),
                quant: None,
                context_length: None,
                thinking_modes: None,
                thinking_markers: None,
                owes_long_echo: None,
                owes_long: true,
                mem: None,
                kv_dtype: None,
            });
    }
    if let Some(lr) = &ladder.long_rungs_for {
        for m in models.values_mut() {
            m.owes_long = lr.owes(m.arch.as_deref(), &m.file);
        }
    }
    let gpu_mem = receipts.iter().find_map(|r| r.gpu_mem_total_bytes);
    HostView {
        decl,
        node: iri_path("release-host", &[&subject.version, &decl.id]),
        receipts,
        kernels,
        models,
        gpu_mem,
    }
}

/// Is this receipt of THIS release — its version, and the sha its `apr` was built from?
fn fresh(version: &str, apr_sha: Option<&str>, subject: &Subject) -> bool {
    version == subject.version && apr_sha == Some(subject.measured_commit())
}

fn emit_host(g: &mut Graph, subject: &Subject, v: &HostView<'_>, stats: &mut ReleaseStats) {
    let h = &v.node;
    g.insert(h.clone(), RDF_TYPE, Term::iri(rel("Host")));
    g.insert(h.clone(), rel("hostId"), Term::string(&v.decl.id));
    g.insert(h.clone(), rel("cc"), Term::string(&v.decl.cc));
    for r in &v.receipts {
        let n = iri_path("release-receipt", &[&subject.version, &v.decl.id, &r.file]);
        g.insert(n.clone(), RDF_TYPE, Term::iri(rel("HostReceipt")));
        g.insert(n.clone(), rel("file"), Term::string(&r.file));
        g.insert(n.clone(), rel("schema"), Term::string(&r.schema));
        g.insert(n.clone(), rel("version"), Term::string(&r.version));
        if let Some(s) = &r.apr_sha {
            g.insert(n.clone(), rel("aprSha"), Term::string(s));
        }
        let ok = fresh(&r.version, r.apr_sha.as_deref(), subject);
        g.insert(n.clone(), rel("fresh"), Term::boolean(ok));
        g.insert(h.clone(), rel("hostReceipt"), Term::iri(n));
        for item in r.inventory.iter().filter(|i| i.sha256.is_none()) {
            g.insert(
                h.clone(),
                rel("unmeasuredModel"),
                Term::string(format!("{}: {}", r.file, item.file)),
            );
        }
        for c in r.cells.iter().filter(|c| c.sha256.is_none()) {
            g.insert(
                h.clone(),
                rel("unkeyedRow"),
                Term::string(format!(
                    "{}: {} {} think-{} {} verdict={}",
                    r.file, c.file, c.verb, c.thinking, c.context, c.verdict
                )),
            );
        }
    }
    for (sha, m) in &v.models {
        stats.models += 1;
        let n = iri("model", sha);
        g.insert(n.clone(), RDF_TYPE, Term::iri(rel("Model")));
        g.insert(n.clone(), rel("sha256"), Term::string(sha));
        g.insert(n.clone(), rel("file"), Term::string(&m.file));
        if let Some(a) = &m.arch {
            g.insert(n.clone(), rel("arch"), Term::string(a));
        }
        if let Some(q) = &m.quant {
            g.insert(n.clone(), rel("quant"), Term::string(q));
        }
        if let Some(l) = m.context_length {
            g.insert(n.clone(), rel("contextLength"), Term::integer(l));
        }
        emit_model_evidence(g, &n, m);
        g.insert(h.clone(), rel("holds"), Term::iri(n));
    }
}

/// A model's thinking evidence and long-rung obligation, and every disagreement between what the producer
/// claimed and what pv derives (#3712 amendments 1, 2).
fn emit_model_evidence(g: &mut Graph, n: &str, m: &ModelInfo) {
    let n = n.to_string();
    for t in m.thinking_modes.iter().flatten() {
        g.insert(n.clone(), rel("thinkingMode"), Term::string(t));
    }
    for mk in m.thinking_markers.iter().flatten() {
        g.insert(n.clone(), rel("thinkingMarker"), Term::string(mk));
    }
    if let Some(c) = m.thinking_contradiction() {
        g.insert(n.clone(), rel("thinkingContradiction"), Term::string(c));
    }
    g.insert(n.clone(), rel("owesLongRungs"), Term::boolean(m.owes_long));
    if let Some(k) = &m.kv_dtype {
        g.insert(n.clone(), rel("kvDtype"), Term::string(k));
    }
    if let Some(echo) = m.owes_long_echo.filter(|e| *e != m.owes_long) {
        g.insert(
            n,
            rel("longRungsMismatch"),
            Term::string(format!(
                "{}: receipt says owes_long_rungs={echo}, the ladder contract derives {}",
                m.file, m.owes_long
            )),
        );
    }
}

/// One cell's key: (verb, thinking mode, rung id, the rung's target size for this model).
type CellKey<'r> = (&'static str, &'static str, &'r str, Option<u64>);

/// One `release:Cell` per (model, verb, thinking, rung) on this host, and an edge to every row that keys onto it.
/// (model sha256, model file, rung id) → the hosts on which that rung is OWED (it fits, or its fit is unknown).
type Coverage = BTreeMap<(String, String, String), BTreeSet<String>>;

fn emit_cells(
    g: &mut Graph,
    subject: &Subject,
    v: &HostView<'_>,
    rungs: &[ContextRung],
    coverage: &mut Coverage,
    stats: &mut ReleaseStats,
) {
    let mut keyed: BTreeSet<(String, usize)> = BTreeSet::new();
    for (sha, m) in &v.models {
        let owed = m.rungs(rungs);
        for (id, t) in &owed {
            let hosts = coverage
                .entry((sha.clone(), m.file.clone(), (*id).to_string()))
                .or_default();
            if m.fits(v.gpu_mem, *t) != Some(false) {
                hosts.insert(v.node.clone());
            }
        }
        let modes = m.modes();
        let keys: Vec<CellKey<'_>> = VERBS
            .iter()
            .flat_map(|verb| modes.iter().map(move |mode| (*verb, *mode)))
            .flat_map(|(verb, mode)| owed.iter().map(move |(id, t)| (verb, mode, *id, *t)))
            .collect();
        for key in keys {
            let cell = emit_cell_node(g, subject, v, sha, m, key);
            stats.cells += 1;
            if m.fits(v.gpu_mem, key.3) == Some(false) {
                stats.refusal_cells += 1;
            }
            let (verb, mode, rung, _) = key;
            stats.cell_names.push(format!(
                "{}/{}/{verb}/think-{mode}/{rung}",
                v.decl.id, m.file
            ));
            let rows = rows_for(g, subject, v, sha, key, &mut keyed);
            if !rows.is_empty() {
                stats.cells_with_row += 1;
            }
            for r in rows {
                g.insert(cell.clone(), rel("row"), Term::iri(r));
            }
        }
    }
    let all: usize = v.receipts.iter().map(|r| r.cells.len()).sum();
    stats.orphan_rows += all - keyed.len();
}

fn emit_cell_node(
    g: &mut Graph,
    subject: &Subject,
    v: &HostView<'_>,
    sha: &str,
    m: &ModelInfo,
    (verb, mode, rung, target): CellKey<'_>,
) -> String {
    let think = format!("think-{mode}");
    let cell = iri_path(
        "release-cell",
        &[&subject.version, &v.decl.id, &m.file, verb, &think, rung],
    );
    // a cell that does not FIT this host by the declared arithmetic is owed as an honest pre-load refusal
    // instead (#3710 rule): a different class, a different shape — never an OOM mid-run
    let class = if m.fits(v.gpu_mem, target) == Some(false) {
        "RefusalCell"
    } else {
        "Cell"
    };
    g.insert(cell.clone(), RDF_TYPE, Term::iri(rel(class)));
    if let Some(req) = target.and_then(|t| m.required_bytes(t)) {
        g.insert(cell.clone(), rel("requiredBytes"), Term::integer(req));
    }
    g.insert(cell.clone(), rel("host"), Term::iri(v.node.clone()));
    g.insert(cell.clone(), rel("model"), Term::iri(iri("model", sha)));
    g.insert(
        cell.clone(),
        rel("verb"),
        Term::iri(iri("release-verb", verb)),
    );
    g.insert(cell.clone(), rel("thinking"), Term::string(mode));
    g.insert(
        cell.clone(),
        rel("context"),
        Term::iri(iri("release-context", rung)),
    );
    cell
}

/// Emit every row of this host's receipts that keys onto (sha, verb, thinking, rung); return their IRIs.
fn rows_for(
    g: &mut Graph,
    subject: &Subject,
    v: &HostView<'_>,
    sha: &str,
    key: CellKey<'_>,
    keyed: &mut BTreeSet<(String, usize)>,
) -> Vec<String> {
    let (verb, mode, rung, _) = key;
    let mut out = Vec::new();
    for r in &v.receipts {
        for (i, c) in r.cells.iter().enumerate() {
            let hit = c.sha256.as_deref() == Some(sha)
                && c.verb == verb
                && c.thinking == mode
                && c.context == rung;
            if hit {
                keyed.insert((r.file.clone(), i));
                out.push(emit_row(g, subject, r, i, c, key, v.gpu_mem));
            }
        }
    }
    out
}

/// Did the row fill its rung? A fixed rung: the prompt alone reaches it. `declared`: prompt plus the output
/// budget fill the model's own length. An unknown on either side is `false` — an unmeasured size met nothing.
fn context_met(c: &CellRow, (_, _, rung, target): CellKey<'_>) -> bool {
    let Some(t) = target else {
        return false;
    };
    match (rung == DECLARED, c.prompt_tokens, c.max_tokens) {
        (false, Some(p), _) => p >= t,
        (true, Some(p), Some(o)) => p.saturating_add(o) >= t,
        _ => false,
    }
}

fn emit_row(
    g: &mut Graph,
    subject: &Subject,
    r: &Receipt,
    i: usize,
    c: &CellRow,
    key: CellKey<'_>,
    gpu_total: Option<u64>,
) -> String {
    let n = iri_path(
        "release-row",
        &[&subject.version, &r.host, &r.file, &i.to_string()],
    );
    g.insert(n.clone(), RDF_TYPE, Term::iri(rel("CellReceipt")));
    g.insert(
        n.clone(),
        rel("verdict"),
        Term::string(c.verdict.to_ascii_lowercase()),
    );
    g.insert(n.clone(), rel("backend"), Term::string(&c.backend));
    if let Some(fb) = c.fallback {
        g.insert(n.clone(), rel("fallback"), Term::boolean(fb));
    }
    let ok = fresh(&r.version, r.apr_sha.as_deref(), subject);
    g.insert(n.clone(), rel("fresh"), Term::boolean(ok));
    g.insert(
        n.clone(),
        rel("contextMet"),
        Term::boolean(context_met(c, key)),
    );
    // thinking ON must CLOSE its block (#3720: an unclosed block is reported, never an empty answer)
    let think_ok = c.thinking == "off" || c.think_closed == Some(true);
    g.insert(n.clone(), rel("thinkOk"), Term::boolean(think_ok));
    let answered = c.answer_chars.is_some_and(|a| a > 0);
    // a refusal NAMES its arithmetic: what it needs is more than what the host has
    // honest only against the TOTAL: a refusal because a co-tenant holds memory (free < required ≤ total) is
    // not the arithmetic, it is a busy GPU silently shrinking what the release proves (#3712, 62)
    let named = matches!((c.required_bytes, gpu_total), (Some(r), Some(t)) if r > t);
    g.insert(n.clone(), rel("arithmeticNamed"), Term::boolean(named));
    g.insert(n.clone(), rel("answered"), Term::boolean(answered));
    emit_row_detail(g, &n, r, c);
    n
}

/// What the row measured, carried for the reader: never graded here.
fn emit_row_detail(g: &mut Graph, n: &str, r: &Receipt, c: &CellRow) {
    let n = n.to_string();
    if let Some(p) = c.prompt_tokens {
        g.insert(n.clone(), rel("promptTokens"), Term::integer(p));
    }
    if let Some(o) = c.max_tokens {
        g.insert(n.clone(), rel("maxTokens"), Term::integer(o));
    }
    if let Some(t) = c.ttft_ms {
        g.insert(n.clone(), rel("ttftMs"), Term::double(t));
    }
    if let Some(rc) = c.rc {
        g.insert(n.clone(), rel("rc"), Term::signed(rc));
    }
    if !c.reason.is_empty() {
        g.insert(n.clone(), rel("reason"), Term::string(&c.reason));
    }
    if let Some(s) = &r.apr_sha {
        g.insert(n.clone(), rel("aprSha"), Term::string(s));
    }
    g.insert(n, rel("receiptFile"), Term::string(&r.file));
}

/// Every rung a model declares must be owed — and so Pass — on at least one required host (#3710 rule): the
/// 27B's 262,144 cells refuse on the 24 GB 4090 and are owed on gx10. A rung that fits nowhere is named here.
fn emit_coverage(g: &mut Graph, subject: &Subject, coverage: &Coverage, stats: &mut ReleaseStats) {
    for ((sha, file, rung), hosts) in coverage {
        stats.rung_coverage += 1;
        let n = iri_path("release-coverage", &[&subject.version, file, rung]);
        g.insert(n.clone(), RDF_TYPE, Term::iri(rel("RungCoverage")));
        g.insert(n.clone(), rel("model"), Term::iri(iri("model", sha)));
        g.insert(
            n.clone(),
            rel("context"),
            Term::iri(iri("release-context", rung)),
        );
        for h in hosts {
            g.insert(n.clone(), rel("owedOn"), Term::iri(h.clone()));
        }
    }
}

/// The kernel differential (#3715 addendum): the universe K is every kernel on ANY required host's dispatch
/// path, and each (kernel, host) — each GPU arch — is a `release:KernelCell`. A kernel measured on one arch only
/// is therefore a missing row on the other, and one that passes on sm_121 and fails on sm_89 is a failing row on
/// the sm_89 node: the class that made qwen2.5-coder red on lambda alone.
fn emit_kernels(
    g: &mut Graph,
    subject: &Subject,
    views: &[HostView<'_>],
    stats: &mut ReleaseStats,
) {
    let universe: BTreeSet<(&str, &str)> = views
        .iter()
        .flat_map(|v| &v.kernels)
        .flat_map(|r| &r.dispatch)
        .flat_map(|d| &d.kernels)
        .map(|(k, q)| (k.as_str(), q.as_str()))
        .collect();
    for v in views {
        emit_kernel_host(g, subject, v);
        for key in &universe {
            stats.kernel_cells += 1;
            emit_kernel_cell(g, subject, v, *key);
        }
    }
}

/// The host's kernel receipts, its dispatch kernels, and every held model the dispatch lists do not cover.
fn emit_kernel_host(g: &mut Graph, subject: &Subject, v: &HostView<'_>) {
    let h = &v.node;
    let mut covered: BTreeSet<&str> = BTreeSet::new();
    for r in &v.kernels {
        let n = iri_path("release-kreceipt", &[&subject.version, &v.decl.id, &r.file]);
        g.insert(n.clone(), RDF_TYPE, Term::iri(rel("KernelReceiptFile")));
        g.insert(n.clone(), rel("file"), Term::string(&r.file));
        g.insert(n.clone(), rel("sm"), Term::string(&r.sm));
        let ok = fresh(&r.version, r.apr_sha.as_deref(), subject);
        g.insert(n.clone(), rel("fresh"), Term::boolean(ok));
        g.insert(h.clone(), rel("kernelReceipt"), Term::iri(n));
        for d in &r.dispatch {
            if let Some(s) = &d.sha256 {
                covered.insert(s);
            }
            for (k, q) in &d.kernels {
                g.insert(
                    h.clone(),
                    rel("dispatchKernel"),
                    Term::string(kernel_label(k, q)),
                );
            }
        }
    }
    for (sha, m) in &v.models {
        if !covered.contains(sha.as_str()) {
            g.insert(
                h.clone(),
                rel("modelWithoutDispatch"),
                Term::string(&m.file),
            );
        }
    }
}

/// `q4k_gemv@q4_k`, or the bare kernel when it carries no quant (an f32 elementwise kernel).
fn kernel_label(kernel: &str, quant: &str) -> String {
    if quant.is_empty() {
        kernel.to_string()
    } else {
        format!("{kernel}@{quant}")
    }
}

fn emit_kernel_cell(
    g: &mut Graph,
    subject: &Subject,
    v: &HostView<'_>,
    (kernel, quant): (&str, &str),
) {
    // kernel and quant as their own segments: `release-kernel/<v>/<host>/<kernel>/<quant>`, `-` for none
    let q = if quant.is_empty() { "-" } else { quant };
    let cell = iri_path("release-kernel", &[&subject.version, &v.decl.id, kernel, q]);
    g.insert(cell.clone(), RDF_TYPE, Term::iri(rel("KernelCell")));
    g.insert(cell.clone(), rel("kernel"), Term::string(kernel));
    g.insert(cell.clone(), rel("quant"), Term::string(quant));
    g.insert(cell.clone(), rel("host"), Term::iri(v.node.clone()));
    g.insert(cell.clone(), rel("cc"), Term::string(&v.decl.cc));
    for r in &v.kernels {
        let rows = r.rows.iter().enumerate();
        for (i, row) in rows.filter(|(_, x)| x.kernel == kernel && x.quant == quant) {
            let n = emit_krow(g, subject, r, i, row);
            g.insert(cell.clone(), rel("row"), Term::iri(n));
        }
    }
}

fn emit_krow(
    g: &mut Graph,
    subject: &Subject,
    r: &KernelReceipt,
    i: usize,
    row: &inputs::KernelRow,
) -> String {
    let n = iri_path(
        "release-krow",
        &[&subject.version, &r.host, &r.file, &i.to_string()],
    );
    g.insert(n.clone(), RDF_TYPE, Term::iri(rel("KernelDiffReceipt")));
    g.insert(n.clone(), rel("verdict"), Term::string(&row.verdict));
    g.insert(n.clone(), rel("reference"), Term::string(&row.reference));
    if let Some(e) = row.max_err {
        g.insert(n.clone(), rel("maxErr"), Term::double(e));
    }
    if let Some(b) = row.bound {
        g.insert(n.clone(), rel("bound"), Term::double(b));
    }
    if let Some(m) = row.index_mismatch {
        g.insert(n.clone(), rel("indexMismatch"), Term::integer(m));
    }
    if let Some(t) = row.tie_margin {
        g.insert(n.clone(), rel("tieMargin"), Term::double(t));
    }
    g.insert(
        n.clone(),
        rel("withinBound"),
        Term::boolean(row.within_bound()),
    );
    if !row.bound_source.is_empty() {
        g.insert(
            n.clone(),
            rel("boundSource"),
            Term::string(&row.bound_source),
        );
    }
    let ok = fresh(&r.version, r.apr_sha.as_deref(), subject);
    g.insert(n.clone(), rel("fresh"), Term::boolean(ok));
    if !row.reason.is_empty() {
        g.insert(n.clone(), rel("reason"), Term::string(&row.reason));
    }
    n
}

/// Tokenizer parity (aprender#3726): apr's token ids against the pinned llama.cpp, one cell per distinct model
/// in any required host's universe — host-independent, because the tokenizer is. It is the cell that sees what
/// every apr-vs-apr gate cannot: CPU and GPU share the tokenizer, so `apr parity` passes a tokenizer that maps
/// every non-ASCII byte to id 0.
fn emit_tokenizer(
    g: &mut Graph,
    subject: &Subject,
    views: &[HostView<'_>],
    receipts: &[TokReceipt],
    stats: &mut ReleaseStats,
) {
    let models: BTreeMap<&str, &ModelInfo> = views
        .iter()
        .flat_map(|v| v.models.iter().map(|(sha, m)| (sha.as_str(), m)))
        .collect();
    for (sha, m) in models {
        stats.tokenizer_cells += 1;
        let cell = iri_path("release-tokenizer", &[&subject.version, &m.file]);
        g.insert(cell.clone(), RDF_TYPE, Term::iri(rel("TokenizerCell")));
        g.insert(cell.clone(), rel("model"), Term::iri(iri("model", sha)));
        if let Some(a) = &m.arch {
            g.insert(cell.clone(), rel("family"), Term::string(a));
        }
        for r in receipts {
            let rows = r.rows.iter().enumerate();
            for (i, row) in rows.filter(|(_, x)| x.sha256.as_deref() == Some(sha)) {
                let n = emit_trow(g, subject, r, i, row);
                g.insert(cell.clone(), rel("row"), Term::iri(n));
            }
        }
    }
}

fn emit_trow(
    g: &mut Graph,
    subject: &Subject,
    r: &TokReceipt,
    i: usize,
    row: &inputs::TokRow,
) -> String {
    let n = iri_path("release-trow", &[&subject.version, &r.file, &i.to_string()]);
    g.insert(
        n.clone(),
        RDF_TYPE,
        Term::iri(rel("TokenizerParityReceipt")),
    );
    g.insert(n.clone(), rel("verdict"), Term::string(&row.verdict));
    g.insert(n.clone(), rel("parityExact"), Term::boolean(row.exact()));
    g.insert(
        n.clone(),
        rel("roundTrip"),
        Term::boolean(row.roundtrip_ok == Some(true)),
    );
    let opt = |g: &mut Graph, p: &str, v: &str| {
        if !v.is_empty() {
            g.insert(n.clone(), rel(p), Term::string(v));
        }
    };
    opt(g, "corpusSha", &row.corpus_sha256);
    opt(g, "comparator", &row.comparator);
    opt(g, "comparatorSha", &row.comparator_sha);
    opt(g, "reason", &row.reason);
    if let Some(m) = row.mismatches {
        g.insert(n.clone(), rel("mismatches"), Term::integer(m));
    }
    if let Some(t) = row.tokens_compared {
        g.insert(n.clone(), rel("tokensCompared"), Term::integer(t));
    }
    let ok = fresh(&r.version, r.apr_sha.as_deref(), subject);
    g.insert(n.clone(), rel("fresh"), Term::boolean(ok));
    n
}

#[cfg(test)]
#[path = "release_evidence_tests.rs"]
mod tests;