ggen-engine 26.10.11

//! The five-stage sync pipeline: Resolve → Enrich → Extract → Render → Write.
//!
//! [`sync`] loads `ggen.toml` and the ontology (Resolve), runs every
//! template `construct:` query once and inserts the produced triples
//! (Enrich — **single pass**: closure is not iterated to a fixed point yet;
//! constructs that depend on other constructs' output require a second
//! `sync` run), evaluates `when:` ASK guards and named `sparql:` SELECTs
//! (Extract), renders each template body via Tera (Render), and applies the
//! Hygen write semantics from [`crate::write`] (Write).
//!
//! After a non-dry-run sync a praxis-core [`ReceiptRecord`] is chained over
//! the payload `{ graph_hash, outputs: { path → BLAKE3 } }` and written to
//! `<root>/.ggen-v2/receipt.json`.
//!
//! Receipt determinism note: `ts_ns` is fixed to `0`. praxis-core's live
//! `LawObject::receipt` path falls back to the system wall clock when
//! `ReceiptMeta::ts_ns` is `None`; this crate forbids wall clocks, so the
//! record is built directly and chained through
//! [`ReceiptRecord::recompute_chain_hash`], which uses the exact same
//! `build_admission_frame`/`chain_from_frame` construction as the live
//! emission path.

use std::{
    collections::{BTreeMap, BTreeSet},
    path::{Path, PathBuf},
    sync::Arc,
    time::Instant,
};

use crate::receipt_chain_seam::{
    epoch::{
        read_receipt_epoch, AdmissionDecision, AdmissionItem, AdmissionLedger, AndonLevel,
        CeilingLevel, ComponentLevels, EquivalenceMap, EquivalenceStatus, ObservedOutcome,
        ReceiptEpochV2, ReceiptEpochV2Builder, SCHEMA_V2,
    },
    Andon, ChainStanding, ChainVerification, ReceiptRecord, CHAIN_RULE_V2_FOLD,
    RECEIPT_RECORD_VERSION,
};
use serde::Serialize;
use tera::Value;

use crate::{
    config::GgenConfig,
    error::{AppError, Result},
    generation_rules::duration_ms,
    graph::{
        DeterministicGraph, EngineQueryResults, EngineValue, GraphEngine, GraphLawStore,
        TurtleDocument,
    },
    template::{build_tera_with_packs, sparql_to_value, Frontmatter, MatchSpec, Template},
    write::{
        plan_write, preflight_checksum_slot, preflight_structured_matchers, validate_match_specs,
        WriteOutcome, MAX_OUTPUT_BYTES,
    },
};

/// Which [`GraphEngine`] a sync runs on.
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
pub enum EngineKind {
    /// praxis-graphlaw (the roxi fork) — the default: law-state engine with
    /// N3/Datalog materialization, SHACL/ShEx gates, and denial checks.
    #[default]
    GraphLaw,
    /// Plain oxigraph — no law-state support (typed `[FM-LAW-*]` refusal on
    /// any law operation). Kept for A/B determinism testing.
    Oxigraph,
}

/// Construct a fresh, empty graph engine for `kind`.
///
/// Shared by [`sync`]'s Stage 1 and `crate::generation_rules::run`'s
/// Resolve stage, which previously each matched on [`EngineKind`]
/// independently (byte-identical duplication) — this is the one
/// authoritative construction point.
///
/// # Errors
/// Propagates the chosen engine's own construction failure.
pub fn new_graph_engine(kind: EngineKind) -> Result<Arc<dyn GraphEngine>> {
    Ok(match kind {
        EngineKind::GraphLaw => Arc::new(GraphLawStore::new()?),
        EngineKind::Oxigraph => Arc::new(DeterministicGraph::new()?),
    })
}

/// Read one ontology Turtle file relative to `root`, returning
/// `(label, content)` where `label` is `path`'s root-relative display form
/// (the same label [`sync`]'s and `crate::generation_rules::run`'s input
/// closures key ontology entries by).
///
/// Shared by both `ggen.toml` schemas' ontology-source and
/// ontology-import reads — each previously duplicated this exact
/// `read_to_string` + `[FM-CONFIG-003]` error-wrap independently.
///
/// # Errors
/// Fails closed with `[FM-CONFIG-003]` if `path` is unreadable.
pub fn read_ontology_file(root: &Path, path: &Path) -> Result<(String, String)> {
    let content = std::fs::read_to_string(path).map_err(|e| {
        AppError::fm_config(
            3,
            format!(
                "ontology `{}` unreadable: {e}. Remediation: fix [ontology].source or \
                 [ontology].imports.",
                path.display()
            ),
        )
    })?;
    Ok((rel_display(root, path), content))
}

/// Absorb the declarative manifest's optional `[rules]` section
/// (v26.10.10 §4.1: `rules.n3` / `rules.datalog`) into the declarative-rules
/// pipeline's existing law stage. Runs at Stage 0, before
/// `crate::generation_rules::run` — the one shared dispatch point every
/// `GgenManifest` passes through.
///
/// Semantics:
/// - Absent `[rules]`, or both lists empty: a strict no-op — the manifest is
///   byte-for-byte unchanged, so pre-`[rules]` projects drift by zero.
/// - `[rules].datalog`: every entry is routed through
///   [`crate::generation_rules::resolve_rule_sources`], which refuses with
///   the typed UNSUPPORTED text (ggen-engine has no Datalog text parser this
///   release) — surfaced here as a typed `[FM-LAW-019]` sync refusal, never
///   a silent skip.
/// - `[rules].n3`: every entry is resolved (readability + extension checked
///   through the same seam) and then appended, in declaration order, to
///   `manifest.law.rules`. The declarative path's law stage — the exact
///   `GraphEngine::{load_rules, materialize, check_denials}` stage
///   `sync`'s frontmatter path already runs — then forward-chains the rules
///   over the loaded ontology facts and folds the derived triples back into
///   the graph *before* any `[[generation.rules]]` SPARQL query, so
///   inference is visible to template rendering. A fired `=> false` denial
///   fuse surfaces from that same stage as the existing `[FM-LAW-016]`
///   refusal naming the `DENIED: … => false.` fuse line — a fired fuse is a
///   refusal, never a warning.
/// - Rule files land in the sync receipt's input closure (the law stage
///   hashes every loaded rule path), so editing a declared rule changes the
///   receipt even when outputs are unchanged.
///
/// # Errors
/// Fails closed with `[FM-LAW-019]` on an unreadable `[rules]` path or any
/// `[rules].datalog` entry.
fn absorb_manifest_rules(
    root: &Path, manifest: &mut ggen_config::manifest::GgenManifest,
) -> Result<()> {
    let Some(rules) = manifest.rules.take() else {
        return Ok(());
    };
    if rules.n3.is_empty() && rules.datalog.is_empty() {
        return Ok(());
    }
    let as_strings = |paths: &[std::path::PathBuf]| -> Vec<String> {
        paths
            .iter()
            .map(|p| p.to_string_lossy().into_owned())
            .collect()
    };
    // Datalog first or second would both be correct; n3 is resolved first so
    // an unreadable n3 path is not masked by (or masked as) a datalog
    // refusal when both sections carry problems.
    if let Err(e) = crate::generation_rules::resolve_rule_sources(root, &as_strings(&rules.n3)) {
        return Err(rule_source_refusal(&e));
    }
    if let Err(e) = crate::generation_rules::resolve_rule_sources(root, &as_strings(&rules.datalog))
    {
        return Err(rule_source_refusal(&e));
    }
    manifest.law.rules.extend(rules.n3);
    Ok(())
}

/// Map a [`crate::generation_rules::RuleSourceError`] onto sync's typed
/// `[FM-LAW-019]` refusal shape, preserving the UNSUPPORTED wording for the
/// Datalog case.
fn rule_source_refusal(e: &crate::generation_rules::RuleSourceError) -> AppError {
    match e {
        crate::generation_rules::RuleSourceError::DatalogUnsupported { path } => AppError::fm_law(
            19,
            format!(
                "UNSUPPORTED: [rules].datalog entry `{}` refused — ggen-engine has no \
                 Datalog text parser this release (graphlaw exposes N3 only). \
                 Remediation: translate the rules to N3 and declare them under \
                 [rules].n3.",
                path.display()
            ),
        ),
        crate::generation_rules::RuleSourceError::Unreadable { path, reason } => AppError::fm_law(
            19,
            format!(
                "rule file `{}` unreadable: {reason}. Remediation: fix [rules].n3 in \
                 ggen.toml.",
                path.display()
            ),
        ),
    }
}

/// Options controlling a [`sync`] run.
#[derive(Debug, Clone, Copy, Default)]
pub struct SyncOptions {
    /// Compute outcomes without writing any file (and without a receipt).
    pub dry_run: bool,
    /// Which graph engine executes the run (default: `GraphLaw`).
    pub engine: EngineKind,
    /// Gall CP37-38: provenance tag threaded onto the resulting receipt's
    /// `ReceiptRecord::origin` (descriptive only, never part of the chain
    /// hash — see `ReceiptRecord::origin`'s own (graphlaw-backed via `crate::receipt_chain_seam`)
    /// doc comment). `None` (the default) for the ordinary human/LLM-
    /// reviewed path; `Some("unattended-dispatch")` set only by
    /// `ggen-mcp`'s bounded unattended-write dispatcher
    /// (`crates/ggen-mcp/src/tools/unattended_dispatch.rs`). A `&'static
    /// str` rather than `String` so `SyncOptions` stays `Copy`.
    pub receipt_origin: Option<&'static str>,
    /// Consumer mode (OS-13): suppress installer emission of outputs
    /// derived from `fixtureOnly`-marked spec individuals. Opt-in via
    /// `ggen sync run --consumer-mode`; also settable permanently per
    /// consumer via `[templates] consumer_mode = true` in `ggen.toml`
    /// (either source enables it). Default `false`: existing behavior
    /// byte-identical.
    pub consumer_mode: bool,
}

/// The outcome of one [`sync`] run.
#[derive(Debug, Clone, Serialize)]
pub struct SyncReport {
    /// Files created, overwritten, or injected into (project-root relative).
    pub written: Vec<PathBuf>,
    /// Files not written, with the reason.
    pub skipped: Vec<(PathBuf, String)>,
    /// BLAKE3 hex of the post-Enrich canonical graph state.
    pub graph_hash_hex: String,
    /// Root-relative output path → write decision ("written", "injected",
    /// `"skipped: <reason>"`).
    pub decisions: BTreeMap<String, String>,
    /// Pack name → BLAKE3 hex of the pack's content hash.
    pub packs: BTreeMap<String, String>,
    /// Input-closure binding: root-relative input path (ontology, pack
    /// ontologies, template files, `from:` bodies) → BLAKE3 hex of the file
    /// bytes, plus an `actuator` entry naming the generator version. A
    /// declared input that cannot be read at binding time is recorded as
    /// `MISSING`, never dropped.
    pub closure: BTreeMap<String, String>,
}

/// Relative path of the sync receipt under the project root.
pub const RECEIPT_REL_PATH: &str = ".ggen-v2/receipt.json";

/// Maximum rows admitted from any one named SPARQL result. This bounds
/// semantic fan-out before Tera rendering or shell/filesystem actuation.
pub const MAX_QUERY_RESULT_ROWS: usize = 4_096;

/// Maximum materialized outputs admitted in one sync, including the
/// engine-owned module aggregator.
pub const MAX_PENDING_OUTPUTS: usize = 8_192;

/// Maximum combined rendered body bytes admitted across one sync.
pub const MAX_TOTAL_RENDERED_BYTES: usize = 64 * 1024 * 1024;

/// One fully-rendered template awaiting `apply` — the boundary between the
/// "may fail" render pass and the write pass, which by construction only
/// begins once every template in the run has rendered successfully.
struct PendingWrite {
    to: String,
    body: String,
    frontmatter: Frontmatter,
}

#[derive(Debug, Clone, Copy)]
enum ProjectionMode {
    Whole,
    Row(usize),
    Aggregate,
}

/// Relative path of the append-only receipt history log (one JSON line per
/// non-dry-run sync) under the project root.
pub const RECEIPT_LOG_REL_PATH: &str = ".ggen-v2/receipt-log.jsonl";

/// On-disk shape of `<root>/.ggen-v2/receipt.json`.
#[derive(Debug, Clone, Serialize, serde::Deserialize)]
pub struct SyncReceipt {
    /// The praxis-core chain record over `payload`.
    pub record: ReceiptRecord,
    /// The hashed payload: graph hash plus per-output BLAKE3 hashes.
    pub payload: ReceiptPayload,
}

/// The payload a sync receipt is chained over.
///
/// `deny_unknown_fields`: the payload's wire shape is a closed contract —
/// `#[serde(default)]` fields below are the only sanctioned way to add a
/// backward-compatible field. An attacker (or a corrupted receipt) that
/// injects an extra, unauthenticated field into stored payload JSON must
/// fail to deserialize, never be silently accepted and ignored.
#[derive(Debug, Clone, Serialize, serde::Deserialize)]
#[serde(deny_unknown_fields)]
pub struct ReceiptPayload {
    /// BLAKE3 hex of the post-Enrich canonical graph state.
    pub graph_hash: String,
    /// Root-relative output path → BLAKE3 hex of the file bytes on disk.
    pub outputs: BTreeMap<String, String>,
    /// Pack name → BLAKE3 hex of the pack's content hash.
    #[serde(default)]
    pub packs: BTreeMap<String, String>,
    /// Root-relative output path → why the file landed (or did not).
    #[serde(default)]
    pub decisions: BTreeMap<String, String>,
    /// Input-closure binding (see [`SyncReport::closure`]): editing a
    /// template, ontology, pack ontology, or `from:` body changes the
    /// receipt even when the rendered outputs happen to be byte-identical.
    #[serde(default)]
    pub closure: BTreeMap<String, String>,
    /// Build provenance BLAKE3 hex of the `ggen-engine` binary that
    /// produced this receipt (see [`crate::build_provenance`]). Additive;
    /// absent (serialized as nothing) only when deserialized from a
    /// pre-provenance receipt.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub build_provenance: Option<String>,
}

/// Run the five-stage pipeline rooted at `root` (the directory containing
/// `ggen.toml`).
///
/// # Errors
/// Fails closed on any resolve/parse/query/render/write failure; see the
/// FM-* codes on [`crate::config`], [`crate::graph`], [`crate::template`],
/// and [`crate::write`].
// One sequential five-stage pipeline orchestrator (load/extract/validate/
// generate/emit), each stage carrying its own OTEL span open/close pair;
// splitting it into sub-functions would scatter that span bracketing across
// call boundaries rather than shrink real complexity.
#[allow(clippy::too_many_lines)]
pub fn sync(root: &Path, opts: SyncOptions) -> Result<SyncReport> {
    // ── Stage 0: Schema dispatch ─────────────────────────────────────────
    //
    // A project's `ggen.toml` is either a frontmatter project (this
    // function's own `GgenConfig` schema below, `deny_unknown_fields`) or a
    // declarative-rules project (`ggen_config::manifest::GgenManifest`,
    // `[[generation.rules]]`) — never both. `crate::schema_dispatch::load`
    // is the ONE shared dispatch point (backed by
    // `ggen_config::classify_ggen_toml`, the shared structural classifier)
    // every ggen.toml-reading call site in this crate goes through —
    // `verbs::handlers::handle_doctor`/`handle_graph_validate`/
    // `build_law_engine` call the exact same function, not a copy of this
    // stage's logic. See that module's own doc comment for the full design
    // and error-behavior contract (missing file falls through to
    // `GgenConfig::load`'s canonical `[FM-CONFIG-001]`; a structurally
    // ambiguous/unsupported/malformed document fails closed with a typed
    // `CONFIG_SCHEMA_*` code instead of a generic TOML deserialization
    // error).
    let parsed = crate::schema_dispatch::load(root)?;
    let config = match parsed {
        crate::schema_dispatch::ParsedGgenToml::DeclarativeRules(mut manifest) => {
            absorb_manifest_rules(root, &mut manifest)?;
            return crate::generation_rules::run(root, &manifest, opts);
        }
        crate::schema_dispatch::ParsedGgenToml::Frontmatter(config) => *config,
    };

    // ── Stage 1: Resolve ────────────────────────────────────────────────
    let load_start = Instant::now();
    let load_span = tracing::info_span!(
        "pipeline.load",
        "operation.name" = "pipeline.load",
        "operation.type" = "pipeline",
        "pipeline.stage" = "load",
        "pipeline.duration_ms" = tracing::field::Empty,
    );
    let load_guard = load_span.enter();

    let ontology_path = root.join(&config.ontology.source);
    let (ontology_label, ttl) = read_ontology_file(root, &ontology_path)?;
    let graph: Arc<dyn GraphEngine> = new_graph_engine(opts.engine)?;
    // Resolve every ontology document before mutating the graph. Each Turtle
    // parser retains document-local prefix/base/blank-node scope; the admitted
    // union is committed through one bounded store mutation.
    //
    // `dry_run` gates *which* resolver runs, not merely whether the later
    // write stage fires: `crate::pack::resolve` permits a `PackRef::Git`
    // pack's cache to be wiped, re-cloned, and re-pinned as needed, which is
    // real network I/O and filesystem mutation performed unconditionally --
    // "unconditionally" being exactly the red-team-confirmed gap (F1,
    // contract-drift) this dry_run branch closes. `opts.dry_run == true`
    // (the only way `ggen-mcp`'s `ggen_sync_dry_run`, documented and
    // annotated `readOnlyHint: true`, ever calls `sync`) instead uses
    // `crate::pack::resolve_read_only`, which resolves a git pack only from
    // an already-correctly-pinned cache and fails closed with
    // `[FM-PACK-012]` on any cache miss/mismatch rather than mutating
    // anything -- see that function's doc comment.
    let packs = if opts.dry_run {
        crate::pack::resolve_read_only(&config, root)?
    } else {
        crate::pack::resolve(&config, root)?
    };
    let lock_entries = crate::pack::lock_entries(&config, &packs)?;
    crate::pack::check_lock(root, &lock_entries)?;
    let mut ontology_sources = Vec::with_capacity(
        2 + packs
            .iter()
            .map(|pack| 1 + pack.extra_ontology_paths.len())
            .sum::<usize>(),
    );
    ontology_sources.push((ontology_label, ttl));
    for import in &config.ontology.imports {
        ontology_sources.push(read_ontology_file(root, &root.join(import))?);
    }
    for pack in &packs {
        let pack_ttl = std::fs::read_to_string(&pack.ontology_path).map_err(|e| {
            AppError::fm_pack(
                4,
                format!(
                    "pack `{}`: ontology `{}` unreadable: {e}",
                    pack.name,
                    pack.ontology_path.display()
                ),
            )
        })?;
        ontology_sources.push((rel_display(root, &pack.ontology_path), pack_ttl));
        for (declared, extra_path) in &pack.extra_ontology_paths {
            let extra_ttl = std::fs::read_to_string(extra_path).map_err(|e| {
                AppError::fm_pack(
                    4,
                    format!(
                        "pack `{}`: extra ontology `{declared}` unreadable at `{}`: {e}",
                        pack.name,
                        extra_path.display()
                    ),
                )
            })?;
            ontology_sources.push((declared.clone(), extra_ttl));
        }
    }
    // Compile declared dependency/type/requires/provides facts into the same
    // canonical graph consumed by gates and templates. The projection uses
    // public Schema.org/Dublin Core predicates and is empty for legacy packs,
    // preserving their graph identity.
    let pack_topology_ttl = crate::pack_scope::topology_turtle(&packs);
    if !pack_topology_ttl.is_empty() {
        ontology_sources.push(("urn:ggen:pack-topology".to_string(), pack_topology_ttl));
    }
    let ontology_documents: Vec<TurtleDocument<'_>> = ontology_sources
        .iter()
        .map(|(label, content)| TurtleDocument::new(label, content))
        .collect();
    let ontology_receipt = graph.insert_turtle_documents(&ontology_documents)?;
    tracing::debug!(
        ontology.documents = ontology_receipt.documents,
        ontology.parsed_quads = ontology_receipt.parsed_quads,
        ontology.inserted_quads = ontology_receipt.inserted_quads,
        "ontology batch admitted"
    );

    let templates = discover_templates(root, &config, &packs)?;

    // Bind the input closure: everything that determines the outputs beyond
    // the graph itself. Hashed from disk at binding time; a declared input
    // that has become unreadable is recorded as MISSING so a verifier sees
    // the gap instead of a silently narrower closure.
    let mut closure: BTreeMap<String, String> = BTreeMap::new();
    closure.insert(
        "actuator".to_string(),
        concat!("ggen@", env!("CARGO_PKG_VERSION")).to_string(),
    );
    closure.insert(
        "policy:frontmatter-projection-limits".to_string(),
        format!(
            "max_query_rows={MAX_QUERY_RESULT_ROWS};max_outputs={MAX_PENDING_OUTPUTS};             max_total_rendered_bytes={MAX_TOTAL_RENDERED_BYTES};             max_output_bytes={MAX_OUTPUT_BYTES}"
        ),
    );
    // Project identity: binds `[project].name` into the receipt so a receipt
    // cannot be silently relabeled as belonging to a different project.
    closure.insert("project".to_string(), config.project.name.clone());
    // The manifest itself (`ggen.toml`) determines the whole pipeline
    // (ontology source, templates dir, law rules/shapes, packs) and must be
    // part of the closure like every other governing input -- previously
    // omitted, which meant editing ggen.toml left no trace in the receipt.
    let manifest_path = root.join("ggen.toml");
    closure.insert(
        rel_display(root, &manifest_path),
        hash_file_or_missing(&manifest_path),
    );
    closure.insert(
        rel_display(root, &ontology_path),
        hash_file_or_missing(&ontology_path),
    );
    for pack in &packs {
        closure.insert(
            rel_display(root, &pack.ontology_path),
            hash_file_or_missing(&pack.ontology_path),
        );
        for (declared, extra_path) in &pack.extra_ontology_paths {
            closure.insert(declared.clone(), hash_file_or_missing(extra_path));
        }
    }
    for (tpl_path, tpl) in &templates {
        closure.insert(rel_display(root, tpl_path), hash_file_or_missing(tpl_path));
        let template_dir = tpl_path.parent().unwrap_or_else(|| Path::new("."));
        if let Some(from) = tpl.frontmatter.from.as_deref() {
            if let Ok(from_path) = crate::write::resolve_target(template_dir, from) {
                closure.insert(
                    rel_display(root, &from_path),
                    hash_file_or_missing(&from_path),
                );
            }
        }
        for rdf_file in &tpl.frontmatter.rdf {
            if let Ok(rdf_path) = crate::write::resolve_target(template_dir, rdf_file) {
                closure.insert(
                    rel_display(root, &rdf_path),
                    hash_file_or_missing(&rdf_path),
                );
            }
        }
    }

    // P2: reflexive receipts (opt-in via `[law].reflexive`). Parses this
    // project's own `<root>/.ggen-v2/receipt-log.jsonl` (if it exists) and
    // inserts one fixed-shape `ggenr:Sync` fact cluster per
    // successfully-parsed line, BEFORE any template renders — so a
    // template's SPARQL SELECTs can see the project's own sync history
    // (syncs 0..N-1) as ordinary graph facts. A line this sync itself is
    // about to append is not yet in the log at this point in the pipeline
    // (the log is only appended in Stage 5, non-dry-run), so a first sync
    // with `reflexive = true` sees zero prior `ggenr:Sync` individuals.
    //
    // Reflexive off (the default) or on-but-unqueried leaves output
    // byte-identical to a pre-reflexive project: the extra `ggenr:` facts
    // only change `graph_hash` (and hence anything that actually SELECTs
    // them), never the write path itself.
    if config.law.reflexive {
        let log_path = root.join(RECEIPT_LOG_REL_PATH);
        if let Ok(raw) = std::fs::read_to_string(&log_path) {
            let mut malformed_count: usize = 0;
            for (idx, line) in raw.lines().enumerate() {
                if line.trim().is_empty() {
                    continue;
                }
                if let Ok(receipt) = serde_json::from_str::<SyncReceipt>(line) {
                    let andon_label = format!("{:?}", receipt.record.andon);
                    let ttl = format!(
                        "@prefix ggenr: <http://seanchatmangpt.github.io/ggen/receipt#> .\n\
                         ggenr:sync-{chain} a ggenr:Sync ;\n\
                         \x20   ggenr:index {idx} ;\n\
                         \x20   ggenr:chainHash \"{chain_esc}\" ;\n\
                         \x20   ggenr:prevChainHash \"{prev_esc}\" ;\n\
                         \x20   ggenr:graphHash \"{graph_esc}\" ;\n\
                         \x20   ggenr:andon \"{andon_esc}\" ;\n\
                         \x20   ggenr:outputCount {outputs} .\n",
                        chain = receipt.record.chain_hash_hex,
                        idx = idx,
                        chain_esc = turtle_escape(&receipt.record.chain_hash_hex),
                        prev_esc = turtle_escape(&receipt.record.prev_chain_hash_hex),
                        graph_esc = turtle_escape(&receipt.payload.graph_hash),
                        andon_esc = turtle_escape(&andon_label),
                        outputs = receipt.payload.outputs.len(),
                    );
                    graph.insert_turtle(&ttl)?;
                } else {
                    // THE ONLY place in this pipeline allowed to
                    // silently skip bad input: everywhere else fails
                    // closed on a malformed governing input, but the
                    // receipt log is history this sync's author does
                    // not directly control byte-for-byte (an
                    // older-tool-version or hand-edited line must not
                    // brick every future sync of the project) — the
                    // skip is recorded here, never silently hidden.
                    closure.insert(
                        format!("receipt-log:MALFORMED-{malformed_count}"),
                        "MALFORMED".to_string(),
                    );
                    malformed_count += 1;
                }
            }
        }
    }

    drop(load_guard);
    load_span.record("pipeline.duration_ms", duration_ms(load_start.elapsed()));

    // ── Stage 2: Enrich (single pass; see module docs) ──────────────────
    //
    // A template declaring `rdf:`/`rdf_inline:` builds its own per-template
    // overlay graph in Stage 3/4 below (see `build_rdf_overlay`), and its
    // `construct:` runs there, against the overlay — not here, against the
    // shared graph. Running it here would leak that template's rdf-sourced
    // triples into the shared graph and hence into every other template,
    // breaking overlay isolation (see `frontmatter_rdf_e2e.rs`).
    let extract_start = Instant::now();
    let extract_span = tracing::info_span!(
        "pipeline.extract",
        "operation.name" = "pipeline.extract",
        "operation.type" = "pipeline",
        "pipeline.stage" = "extract",
        "pipeline.duration_ms" = tracing::field::Empty,
    );
    let extract_guard = extract_span.enter();

    for (_, tpl) in &templates {
        if declares_rdf_overlay(&tpl.frontmatter) {
            continue;
        }
        if let Some(construct) = tpl.frontmatter.construct.as_deref() {
            insert_construct(graph.as_ref(), construct)?;
        }
    }

    drop(extract_guard);
    extract_span.record("pipeline.duration_ms", duration_ms(extract_start.elapsed()));

    // ── Stage 2b: Law — materialize rules, then gate (denials, SPARQL gates)
    //
    // All optional-when-unconfigured: an absent `[law]` table runs no law
    // stage at all, so pre-law projects sync unchanged. With the oxigraph
    // engine any configured law input is a typed `[FM-LAW-*]` refusal.
    let validate_start = Instant::now();
    let validate_span = tracing::info_span!(
        "pipeline.validate",
        "operation.name" = "pipeline.validate",
        "operation.type" = "pipeline",
        "pipeline.stage" = "validate",
        "pipeline.duration_ms" = tracing::field::Empty,
    );
    let validate_guard = validate_span.enter();

    // [law].rules: load every declared N3/Datalog rule file. Always
    // attempted (an empty `rules` list is simply a no-op loop) — splitting
    // this out from the materialize/check_denials gate below lets a
    // hook-only project (no [law].rules at all) still get its hook packs
    // materialized.
    for rel in &config.law.rules {
        let rule_path = root.join(rel);
        let src = std::fs::read_to_string(&rule_path).map_err(|e| {
            AppError::fm_law(
                10,
                format!(
                    "rule file `{}` unreadable: {e}. Remediation: fix [law].rules.",
                    rule_path.display()
                ),
            )
        })?;
        closure.insert(
            rel_display(root, &rule_path),
            hash_file_or_missing(&rule_path),
        );
        graph.load_rules(&src)?;
    }

    // Pack-shipped `kh:` Knowledge Hooks (P1, live pack hooks): any resolved
    // pack may carry a `hook.ttl` next to its `ontology.ttl`/`shapes.ttl`.
    // Loaded (and, below, materialized) BEFORE the pack SHACL-shapes loop
    // runs, so hook-derived facts are visible to that gate too — a hook that
    // derives a fact violating a pack's own shapes.ttl (or another pack's)
    // must be catchable, not silently invisible to shape validation. The
    // hook document joins the receipt closure like every other governing
    // input.
    let mut any_hook_loaded = false;
    for pack in &packs {
        let hook_path = pack.root.join("hook.ttl");
        if !hook_path.is_file() {
            continue;
        }
        let hook_ttl = std::fs::read_to_string(&hook_path).map_err(|e| {
            AppError::fm_pack(
                14,
                format!(
                    "pack `{}`: hook.ttl at `{}` exists but is unreadable: {e}. \
                     Remediation: fix the file's permissions/encoding or remove it.",
                    pack.name,
                    hook_path.display()
                ),
            )
        })?;
        closure.insert(
            rel_display(root, &hook_path),
            hash_file_or_missing(&hook_path),
        );
        graph.load_hook_pack(&hook_ttl)?;
        any_hook_loaded = true;
    }

    // Materialize once whenever there is anything to materialize: either
    // [law].rules is non-empty, or at least one pack contributed a hook.ttl
    // this run. A project with neither runs no law stage at all (existing
    // behavior for pre-law projects is unchanged); a project with only
    // [law].rules or only hook packs still gets a real materialize +
    // check_denials pass (previously gated on `[law].rules` alone, which
    // would have silently skipped materialization for a hook-only project).
    if !config.law.rules.is_empty() || any_hook_loaded {
        graph.materialize()?;
        let denials = graph.check_denials()?;
        if !denials.is_empty() {
            return Err(AppError::fm_law(
                11,
                format!(
                    "{} denial rule(s) violated after materialization: {}. \
                     Remediation: fix the facts or the denial rules in [law].rules \
                     (or a pack's hook.ttl).",
                    denials.len(),
                    denials.join("; ")
                ),
            ));
        }
    }
    // `[law].shapes` (SHACL) was replaced by `[law].gates` (SPARQL gate
    // queries, below). The field stays deserializable so a legacy config
    // gets THIS clear, typed refusal instead of a serde unknown-field error
    // — a file that used to be law must never be silently ignored.
    if !config.law.shapes.is_empty() {
        return Err(AppError::fm_law(
            12,
            format!(
                "[law].shapes ({} SHACL shapes file(s) declared) is no longer \
                 supported at sync time; SHACL shape gates were replaced by \
                 engine-independent SPARQL gate queries. Remediation: migrate to \
                 [law].gates = [\"…/*.rq\"] — each file holds one ASK (true = \
                 violation) or SELECT (any row = violation) query, optionally \
                 preceded by `# MESSAGE: <text>` comment lines.",
                config.law.shapes.len()
            ),
        ));
    }

    // `[law].gates`: SPARQL gate queries over the post-materialization
    // union graph. Engine-independent by construction — `GraphEngine::query`
    // answers ASK/SELECT on both backends (the GraphLaw store delegates to
    // its oxigraph mirror, which holds every materialized derived fact), so
    // the same gate refuses the same facts under either engine.
    //
    // Every gate identity is recorded into `gates_attempted` as it is about
    // to be evaluated, and the list flows into the portable receipt envelope
    // (RFC-GPACK-001 §15: a gate result binds gate identity + attempt
    // observed; §55 `admission.gates_attempted`). A refusal emits the
    // envelope with the typed refusal before the error propagates (§57).
    let mut gates_attempted: Vec<String> = Vec::new();
    // Gate-path glob expansion (fix-forward of the v26.10.6 candidate): a
    // `[law].gates` entry containing `*` or `?` expands — sorted, so the
    // documented gate-precedes-writes evaluation order stays deterministic
    // — against paths relative to the manifest dir; a matching-nothing glob
    // is a fail-closed `[FM-CONFIG-014]` refusal. Literal entries unchanged.
    let resolved_gates = resolve_gate_entries(root, &config.law.gates)?;
    for (rel_disp, gate_path) in &resolved_gates {
        let src = std::fs::read_to_string(gate_path).map_err(|e| {
            AppError::fm_law(
                12,
                format!(
                    "SPARQL gate file `{}` unreadable: {e}. Remediation: fix [law].gates.",
                    gate_path.display()
                ),
            )
        })?;
        closure.insert(rel_disp.clone(), hash_file_or_missing(gate_path));
        gates_attempted.push(rel_disp.clone());
        let gate = parse_gate_source(&src);
        match evaluate_gate(graph.as_ref(), &gate.query)? {
            GateOutcome::Pass => {}
            GateOutcome::NotAGate => {
                return Err(refuse_with_portable_envelope(
                    root,
                    &packs,
                    graph.as_ref(),
                    &gates_attempted,
                    "GATE_INVALID",
                    opts.dry_run,
                    AppError::fm_law(
                        12,
                        format!(
                            "SPARQL gate `{rel_disp}` is not a gate query: it must be an ASK \
                             (true = violation) or a SELECT (any row = violation), not \
                             a CONSTRUCT/DESCRIBE. Remediation: fix [law].gates."
                        ),
                    ),
                ));
            }
            GateOutcome::Violation(detail) => {
                return Err(refuse_with_portable_envelope(
                    root,
                    &packs,
                    graph.as_ref(),
                    &gates_attempted,
                    "GATE_VIOLATION",
                    opts.dry_run,
                    AppError::fm_law(
                        13,
                        format!(
                            "SPARQL gate `{}` refused the sync: {}{detail}. \
                             Remediation: fix the offending facts or the gate query.",
                            rel_disp,
                            gate.message_prefix(),
                        ),
                    ),
                ));
            }
        }
    }

    // Pack-shipped SPARQL gates: any resolved pack may carry a `gates/`
    // directory of `*.rq` files next to its `ontology.ttl`. Each one is
    // evaluated (in sorted filename order) against the same union graph the
    // templates will render from, and any violation refuses the sync
    // outright — this is what makes cross-pack data drift (e.g. two packs
    // mirroring the same subject IRI with diverged property values, which
    // previously surfaced only as a downstream rustc duplicate-variant
    // error, or not at all) refusable at sync time instead of detectable
    // only by whichever consumer happens to compile the union. Packs
    // without a `gates/` directory are unaffected. Every gate file joins
    // the receipt closure like every other governing input. This replaces
    // the previous pack `shapes.ttl` SHACL gate: gates are plain ASK/SELECT
    // queries, so they behave identically under both graph engines.
    for pack in &packs {
        // A pack still shipping the legacy SHACL gate file gets a loud,
        // typed refusal — a file that used to be law must never be silently
        // ignored.
        let legacy_shapes = pack.root.join("shapes.ttl");
        if legacy_shapes.exists() {
            return Err(AppError::fm_pack(
                12,
                format!(
                    "pack `{}`: shapes.ttl at `{}` is no longer supported; pack \
                     SHACL shape gates were replaced by engine-independent SPARQL \
                     gate queries. Remediation: migrate to gates/*.rq — each file \
                     holds one ASK (true = violation) or SELECT (any row = \
                     violation) query, optionally preceded by `# MESSAGE: <text>` \
                     comment lines — then delete shapes.ttl.",
                    pack.name,
                    legacy_shapes.display()
                ),
            ));
        }
        for gate_path in list_gate_files(&pack.root.join("gates"), &pack.name)? {
            let src = std::fs::read_to_string(&gate_path).map_err(|e| {
                AppError::fm_pack(
                    12,
                    format!(
                        "pack `{}`: gate `{}` exists but is unreadable: {e}. \
                         Remediation: fix the file's permissions/encoding or remove it.",
                        pack.name,
                        gate_path.display()
                    ),
                )
            })?;
            closure.insert(
                rel_display(root, &gate_path),
                hash_file_or_missing(&gate_path),
            );
            gates_attempted.push(rel_display(root, &gate_path));
            let gate = parse_gate_source(&src);
            match evaluate_gate(graph.as_ref(), &gate.query)? {
                GateOutcome::Pass => {}
                GateOutcome::NotAGate => {
                    return Err(refuse_with_portable_envelope(
                        root,
                        &packs,
                        graph.as_ref(),
                        &gates_attempted,
                        "GATE_INVALID",
                        opts.dry_run,
                        AppError::fm_pack(
                            12,
                            format!(
                                "pack `{}`: gate `{}` is not a gate query: it must be an \
                                 ASK (true = violation) or a SELECT (any row = \
                                 violation), not a CONSTRUCT/DESCRIBE. Remediation: fix \
                                 or remove the gate file.",
                                pack.name,
                                gate_path.display()
                            ),
                        ),
                    ));
                }
                GateOutcome::Violation(detail) => {
                    return Err(refuse_with_portable_envelope(
                        root,
                        &packs,
                        graph.as_ref(),
                        &gates_attempted,
                        "GATE_VIOLATION",
                        opts.dry_run,
                        AppError::fm_pack(
                            13,
                            format!(
                                "pack `{}` gate `{}` refused the sync against the union \
                                 graph: {}{detail}. Remediation: fix the offending facts \
                                 (in this pack, another pack, or the project ontology — \
                                 gates run against the UNION, so a violation may come \
                                 from any graph source), or fix the gate query.",
                                pack.name,
                                gate_path
                                    .file_name()
                                    .unwrap_or(gate_path.as_os_str())
                                    .to_string_lossy(),
                                gate.message_prefix(),
                            ),
                        ),
                    ));
                }
            }
        }
    }

    drop(validate_guard);
    validate_span.record(
        "pipeline.duration_ms",
        duration_ms(validate_start.elapsed()),
    );

    // ── Stage 2c: EA Graph (ABB/SBB) & Depgraph Admission ──────────────
    //
    // If an EA graph (strategy/capabilities/abbs/contracts/sbbs/qualifications)
    // is declared in the project, admit every candidate SBB against the
    // architecture contract before proceeding to template rendering.
    // In addition, if packs define [graph] dependencies or ports, resolve
    // cross-pack DAG, cycles, and port completeness.
    {
        // 1. Pack-level dependency/port resolution
        let mut pack_manifests = Vec::new();
        for pack in &packs {
            let manifest_path = pack.root.join("pack.toml");
            if manifest_path.is_file() {
                if let Ok(content) = std::fs::read_to_string(&manifest_path) {
                    if let Ok(pm) = ggen_abb_sbb::depgraph::extract_pack_manifest(&content) {
                        pack_manifests.push(pm);
                    }
                }
            }
        }
        if !pack_manifests.is_empty() {
            let consumer_edges = if manifest_path.is_file() {
                std::fs::read_to_string(&manifest_path).ok().and_then(|c| {
                    ggen_abb_sbb::depgraph::extract_consumer_edges(&c)
                        .ok()
                        .flatten()
                })
            } else {
                None
            };
            ggen_abb_sbb::depgraph::resolve_sync_order(&pack_manifests, consumer_edges.as_ref())
                .map_err(|refusal| {
                    AppError::fm_pack(
                        15,
                        format!("Cross-pack dependency graph admission refused: {refusal}"),
                    )
                })?;
        }

        // 2. EA graph admission: if ea.graph.json exists or is embedded
        let ea_graph_path = root.join("ea.graph.json");
        if ea_graph_path.is_file() {
            let content = std::fs::read_to_string(&ea_graph_path).map_err(|e| {
                AppError::fm_law(
                    15,
                    format!(
                        "ea.graph.json at `{}` unreadable: {e}",
                        ea_graph_path.display()
                    ),
                )
            })?;
            let ea_graph = ggen_abb_sbb::parse_graph(&content).map_err(|refusal| {
                AppError::fm_law(
                    15,
                    format!(
                        "EA graph at `{}` malformed: {refusal}",
                        ea_graph_path.display()
                    ),
                )
            })?;
            // Validate graph integrity
            ea_graph.validate().map_err(|refusal| {
                AppError::fm_law(
                    15,
                    format!(
                        "EA graph at `{}` failed validation: {refusal}",
                        ea_graph_path.display()
                    ),
                )
            })?;
            // Admit every candidate SBB defined in the graph
            for sbb in &ea_graph.candidate_sbbs {
                let req = ggen_abb_sbb::Request {
                    abb: sbb.abb.clone(),
                    sbb: sbb.id.clone(),
                    requested_authority: ggen_abb_sbb::Authority::Construct,
                    expected_graph_digest: None,
                };
                ggen_abb_sbb::admit(&ea_graph, &req).map_err(|refusal| {
                    AppError::fm_law(
                        15,
                        format!("EA SBB admission refused for `{}`: {refusal}", sbb.id),
                    )
                })?;
            }
        }
    }

    let graph_hash_hex = hex32(&graph.state_hash()?);

    // ── Stages 3–4: Extract, Render every template into memory ──────────
    //
    // All rendering happens before any write. A render failure on template N
    // (a bad SPARQL query, a Tera error, a determinism mismatch) must leave
    // NO partial result on disk from templates 1..N-1 — so writes are
    // deferred to a second pass (below) that only starts once every
    // template has rendered successfully.
    let generate_start = Instant::now();
    let generate_span = tracing::info_span!(
        "pipeline.generate",
        "operation.name" = "pipeline.generate",
        "operation.type" = "pipeline",
        "pipeline.stage" = "generate",
        "pipeline.duration_ms" = tracing::field::Empty,
    );
    let generate_guard = generate_span.enter();

    let mut tera = build_tera_with_packs(Arc::clone(&graph), &packs)?;

    // ── Consumer-mode emission filter (OS-13, WP-5 consumer half) ───────
    //
    // Opt-in via `ggen sync run --consumer-mode` (or the permanent
    // `[templates] consumer_mode = true` config key): sync must never emit
    // installer artifacts derived from `fixtureOnly`-marked spec
    // individuals into a consumer tree. The marked subjects are collected
    // once from the post-enrich union graph (the same graph the pack
    // gates just evaluated), then every candidate write is checked for
    // provenance before joining `pending` — so suppressed outputs never
    // reach `apply`, never run `sh_before`/`sh_after`, never join the
    // `aggregate_modules` aggregator, and never enter the receipt's
    // `outputs` map. Each suppression is recorded as a typed skip in
    // `decisions`/`skipped`, never silently dropped.
    let consumer_fixture_subjects: Option<std::collections::BTreeSet<String>> =
        if opts.consumer_mode || config.templates.consumer_mode {
            Some(fixture_only_subjects(graph.as_ref())?)
        } else {
            None
        };

    let mut skipped: Vec<(PathBuf, String)> = Vec::new();
    let mut decisions: BTreeMap<String, String> = BTreeMap::new();
    // Every pending write owns the output-phase frontmatter rendered from
    // the same semantic context as its path and body.
    let mut pending: Vec<PendingWrite> = Vec::new();
    // `determinism: true`'s recheck must be a genuinely independent second
    // load (new `GraphEngine` instance, ontology/pack Turtle re-read and
    // re-parsed from disk), never a second query against the SAME
    // already-loaded `graph` — see [`build_independent_recheck_base_graph`].
    // Built lazily, at most once per sync, the first time any template
    // declares `determinism: true` (every template in one sync shares the
    // same base graph state, so one independent build suffices for all of
    // them).
    let mut independent_recheck_base: Option<Arc<dyn GraphEngine>> = None;

    for (tpl_path, tpl) in &templates {
        // Gap 1: per-template RDF overlay (`rdf:`/`rdf_inline:`) — a brand
        // new store layered over the base graph's current triples, built
        // fresh for every template that declares either field so it is
        // deterministic and side-effect-free across templates processed in
        // this same run (the shared `graph`/`tera` are never touched).
        // Templates declaring neither field keep using the shared
        // graph+tera exactly as before this gap existed.
        let overlay = build_rdf_overlay(&graph, tpl_path, &tpl.frontmatter)?;
        if let Some(overlay_graph) = &overlay {
            if let Some(construct) = tpl.frontmatter.construct.as_deref() {
                insert_construct(overlay_graph.as_ref(), construct)?;
            }
        }
        let mut overlay_tera_slot;
        let (active_graph, active_tera): (&Arc<dyn GraphEngine>, &mut tera::Tera) = match &overlay {
            Some(og) => {
                overlay_tera_slot = build_tera_with_packs(Arc::clone(og), &packs)?;
                (og, &mut overlay_tera_slot)
            }
            None => (&graph, &mut tera),
        };

        // `determinism: true` — run the SAME extraction a SECOND,
        // independent time, once per template (not once per row), and BEFORE
        // the primary extraction's own `when:` guard is evaluated below. This
        // ordering matters: it lets the primary's `when:` guard skip branch
        // (the very next block) consult the recheck too, instead of the
        // recheck being computed only when the primary happened to produce
        // output. `check_determinism` (called after a successful primary
        // render) and `check_determinism_guard_agrees_false`/
        // `check_determinism_rows_agree_empty` (called from the three skip
        // branches below) all reuse this one recheck instead of re-running
        // the query per row. The common case (`determinism` unset/false)
        // never calls `extract_query_results` a second time: zero extra
        // query cost, zero behavior change.
        //
        // The recheck runs against a genuinely independent second load of
        // the ontology (a fresh `GraphEngine` instance, Turtle re-read and
        // re-parsed from disk — see `build_independent_recheck_base_graph`),
        // NOT against `active_graph`/`graph`. Reusing the already-loaded
        // store here would make this recheck structurally incapable of
        // catching load-time nondeterminism: oxigraph's
        // `RdfParser::rename_blank_nodes()` (used by
        // `graph::ontology_batch::insert_documents`) assigns every blank
        // node a fresh `rand::random()` id per parse, so a query that binds
        // a blank node genuinely differs across two independent loads of
        // byte-identical Turtle — but is invisible to a recheck that only
        // re-queries the one store already sitting in memory (proven by
        // `tests/determinism_independent_reload_e2e.rs`).
        let determinism_recheck: Option<ExtractedRows> = if tpl.frontmatter.determinism
            == Some(true)
        {
            let fresh_base = if let Some(existing) = &independent_recheck_base {
                Arc::clone(existing)
            } else {
                let built =
                    build_independent_recheck_base_graph(root, &config, opts, &packs, &templates)?;
                independent_recheck_base = Some(Arc::clone(&built));
                built
            };
            let recheck_graph: Arc<dyn GraphEngine> =
                match build_rdf_overlay(&fresh_base, tpl_path, &tpl.frontmatter)? {
                    Some(fresh_overlay) => fresh_overlay,
                    None => fresh_base,
                };
            Some(extract_query_results(
                recheck_graph.as_ref(),
                tpl_path,
                &tpl.frontmatter,
                "determinism_recheck",
            )?)
        } else {
            None
        };

        // Extract: `when:` ASK guard + named `sparql:` SELECTs → rows.
        let Some((named, results)) =
            extract_query_results(active_graph.as_ref(), tpl_path, &tpl.frontmatter, "primary")?
        else {
            // The independent recheck was already computed above (if
            // `determinism: true`) for exactly this situation. A primary
            // `when:` guard evaluating false must not silently bypass
            // verification just because there is nothing to render: if the
            // independently-reloaded second graph's guard evaluated true
            // instead, that asymmetry is itself the load-time nondeterminism
            // `determinism: true` exists to catch (see
            // [FM-TPL-009] `check_determinism_guard_agrees_false`).
            check_determinism_guard_agrees_false(tpl_path, determinism_recheck.as_ref())?;
            let reason = format!("when guard false ({})", tpl_path.display());
            let evidence_key =
                if tpl.frontmatter.to.contains("{{") || tpl.frontmatter.to.contains("{%") {
                    format!("@template/{}", rel_display(root, tpl_path))
                } else {
                    crate::write::resolve_target(root, &tpl.frontmatter.to)?;
                    tpl.frontmatter.to.clone()
                };
            decisions.insert(evidence_key.clone(), format!("skipped: {reason}"));
            skipped.push((PathBuf::from(evidence_key), reason));
            continue;
        };

        let fan_out = tpl.frontmatter.to.contains("{{");
        let explicit_driver = tpl.frontmatter.for_each.is_some();
        if explicit_driver && !fan_out {
            let Some((body, frontmatter)) = render_aggregate_projection(
                active_tera,
                &tpl.body,
                tpl_path,
                &tpl.frontmatter,
                &named,
                &results,
            )?
            else {
                // The named `for_each:` driver legitimately returned zero
                // rows -- a lawful answer, not a failure (the same
                // "honestly reported zero" distinction `ggen-mcp`'s
                // `honest_zero_rows_is_lawful` referee invariant makes) --
                // but previously silent: no `decisions`/`skipped` entry
                // recorded it, so a real empty result was indistinguishable
                // from a template dropped for any other reason. Recorded
                // the same observable way as a `when:` guard false, above.
                //
                // The independent recheck was already computed above (if
                // `determinism: true`) for exactly this situation -- a
                // zero-row primary result must not silently skip
                // verification just because there is nothing to render.
                check_determinism_rows_agree_empty(
                    tpl_path,
                    determinism_recheck.as_ref(),
                    "named `for_each:` driver",
                )?;
                let driver_name = tpl.frontmatter.for_each.as_deref().unwrap_or("?");
                let reason = format!(
                    "for_each `{driver_name}` produced 0 rows ({})",
                    tpl_path.display()
                );
                crate::write::resolve_target(root, &tpl.frontmatter.to)?;
                let evidence_key = tpl.frontmatter.to.clone();
                decisions.insert(evidence_key.clone(), format!("skipped: {reason}"));
                skipped.push((PathBuf::from(evidence_key), reason));
                continue;
            };
            admit_shape_files(
                root,
                tpl_path,
                &frontmatter.shape,
                active_graph.as_ref(),
                &mut closure,
            )?;
            check_determinism(
                active_tera,
                &tpl.body,
                tpl_path,
                &tpl.frontmatter,
                determinism_recheck.as_ref(),
                ProjectionMode::Aggregate,
                &body,
                &frontmatter,
            )?;
            // Consumer-mode (OS-13): an aggregate projection over any
            // fixtureOnly-marked subject is suppressed whole.
            if let Some(subjects) = &consumer_fixture_subjects {
                if let Some(iri) = named
                    .values()
                    .find_map(|v| fixture_reference_in(v, subjects))
                {
                    let reason = format!(
                        "consumer-mode: aggregate projection references fixtureOnly-marked \
                         spec <{iri}> — installer emission suppressed"
                    );
                    decisions.insert(tpl.frontmatter.to.clone(), format!("skipped: {reason}"));
                    skipped.push((PathBuf::from(tpl.frontmatter.to.clone()), reason));
                    continue;
                }
            }
            pending.push(PendingWrite {
                to: tpl.frontmatter.to.clone(),
                body,
                frontmatter,
            });
        } else if fan_out {
            if results.is_empty() {
                // The implicit per-row fan-out driver (the primary
                // `sparql:` result `to:` interpolates over, with no named
                // `for_each:` field at all) legitimately returned zero rows.
                // Previously this loop just ran zero times with no trace
                // anywhere in the report -- recorded the same observable
                // way as the named-driver zero-row case just above.
                //
                // The independent recheck was already computed above (if
                // `determinism: true`) for exactly this situation -- a
                // zero-row primary result must not silently skip
                // verification just because there is nothing to render.
                check_determinism_rows_agree_empty(
                    tpl_path,
                    determinism_recheck.as_ref(),
                    "implicit row fan-out",
                )?;
                let reason = format!(
                    "for_each (implicit row fan-out) produced 0 rows ({})",
                    tpl_path.display()
                );
                let evidence_key = format!("@template/{}", rel_display(root, tpl_path));
                decisions.insert(evidence_key.clone(), format!("skipped: {reason}"));
                skipped.push((PathBuf::from(evidence_key), reason));
            }
            for (row_index, row) in results.iter().enumerate() {
                let ctx = row_context(&named, &results, row);
                let to = render_str(active_tera, &tpl.frontmatter.to, &ctx, tpl_path)?;
                let body = render_str(active_tera, &tpl.body, &ctx, tpl_path)?;
                let frontmatter =
                    render_output_frontmatter(active_tera, &tpl.frontmatter, &ctx, tpl_path, &to)?;
                admit_shape_files(
                    root,
                    tpl_path,
                    &frontmatter.shape,
                    active_graph.as_ref(),
                    &mut closure,
                )?;
                check_determinism(
                    active_tera,
                    &tpl.body,
                    tpl_path,
                    &tpl.frontmatter,
                    determinism_recheck.as_ref(),
                    ProjectionMode::Row(row_index),
                    &body,
                    &frontmatter,
                )?;
                // Consumer-mode (OS-13): a row bound to a fixtureOnly-marked
                // spec individual is suppressed — its installer emission is
                // recorded as a typed skip keyed by the rendered output path.
                if let Some(subjects) = &consumer_fixture_subjects {
                    if let Some(iri) = fixture_reference_in(&results[row_index], subjects) {
                        let reason = format!(
                            "consumer-mode: row references fixtureOnly-marked spec \
                             <{iri}> — installer emission suppressed"
                        );
                        decisions.insert(to.clone(), format!("skipped: {reason}"));
                        skipped.push((PathBuf::from(to.clone()), reason));
                        continue;
                    }
                }
                pending.push(PendingWrite {
                    to,
                    body,
                    frontmatter,
                });
            }
        } else {
            let ctx = base_context(&named, &results);
            let to = tpl.frontmatter.to.clone();
            let body = render_str(active_tera, &tpl.body, &ctx, tpl_path)?;
            let frontmatter =
                render_output_frontmatter(active_tera, &tpl.frontmatter, &ctx, tpl_path, &to)?;
            admit_shape_files(
                root,
                tpl_path,
                &frontmatter.shape,
                active_graph.as_ref(),
                &mut closure,
            )?;
            check_determinism(
                active_tera,
                &tpl.body,
                tpl_path,
                &tpl.frontmatter,
                determinism_recheck.as_ref(),
                ProjectionMode::Whole,
                &body,
                &frontmatter,
            )?;
            // Consumer-mode (OS-13): a whole-file projection whose query
            // results name a fixtureOnly-marked subject is suppressed.
            if let Some(subjects) = &consumer_fixture_subjects {
                if let Some(iri) = named
                    .values()
                    .find_map(|v| fixture_reference_in(v, subjects))
                {
                    let reason = format!(
                        "consumer-mode: projection references fixtureOnly-marked spec \
                         <{iri}> — installer emission suppressed"
                    );
                    decisions.insert(to.clone(), format!("skipped: {reason}"));
                    skipped.push((PathBuf::from(to.clone()), reason));
                    continue;
                }
            }
            pending.push(PendingWrite {
                to,
                body,
                frontmatter,
            });
        }
    }

    // `[templates] aggregate_modules = true`: emit one engine-owned
    // aggregator (`src/ggen_pack_mods.rs`) mounting every generated
    // `src/*.rs` output, so the consumer's own lib.rs needs exactly one
    // permanent `include!("ggen_pack_mods.rs");` line rather than one
    // hand-written mount per pack. Exactly one writer — this stage — so two
    // packs can never collide on it the way per-pack `to: src/lib.rs`
    // templates did (the duplicate-target refusal below still guards the
    // aggregator path itself against a template claiming it directly).
    // Deterministic: entries sorted by target path; mod names derived from
    // the file stem. Skipped entirely (no file, not an empty file) when no
    // `src/*.rs` outputs exist this run.
    if config.templates.aggregate_modules {
        const AGGREGATOR_REL_PATH: &str = "src/ggen_pack_mods.rs";
        let mut mounts: Vec<(String, String)> = pending
            .iter()
            .filter(|pw| {
                pw.to.starts_with("src/")
                    && Path::new(&pw.to)
                        .extension()
                        .is_some_and(|ext| ext.eq_ignore_ascii_case("rs"))
                    && pw.to != AGGREGATOR_REL_PATH
            })
            .map(|pw| {
                let rel_to_src = pw.to.trim_start_matches("src/").to_string();
                let mod_name = rel_to_src
                    .trim_end_matches(".rs")
                    .chars()
                    .map(|c| if c.is_ascii_alphanumeric() { c } else { '_' })
                    .collect::<String>();
                (mod_name, rel_to_src)
            })
            .collect();
        mounts.sort();
        mounts.dedup();
        if !mounts.is_empty() {
            // Plain `//` comments, NOT `//!` inner docs: the file is meant
            // to be consumed via `include!(...)` from the consumer's lib.rs,
            // where pasted-in inner doc comments are a hard E0753 compile
            // error (found by actually building a real consumer, not
            // assumed).
            let mut body = String::from(
                "// GENERATED by `ggen sync` (`[templates] aggregate_modules = true`).\n\
                 // One mount per generated `src/*.rs` output this run produced. Do not\n\
                 // edit by hand — regenerated every sync. Consumer wiring: exactly one\n\
                 // permanent line in your own lib.rs:\n\
                 //     include!(\"ggen_pack_mods.rs\");\n\n",
            );
            for (mod_name, rel_to_src) in &mounts {
                use std::fmt::Write as _;
                let _ = writeln!(body, "#[path = \"{rel_to_src}\"]\npub mod {mod_name};");
            }
            let frontmatter = Frontmatter {
                to: AGGREGATOR_REL_PATH.to_string(),
                sparql: BTreeMap::new(),
                for_each: None,
                construct: None,
                inject: false,
                before: None,
                after: None,
                at_line: None,
                skip_if: None,
                unless_exists: false,
                unattended_write_eligible: false,
                force: true,
                when: None,
                skip_empty: false,
                from: None,
                sh_before: None,
                sh_after: None,
                backup: false,
                shape: Vec::new(),
                determinism: None,
                freeze_policy: None,
                freeze_slots_dir: None,
                rdf: Vec::new(),
                rdf_inline: Vec::new(),
                prefixes: BTreeMap::new(),
                base: None,
            };
            pending.push(PendingWrite {
                to: AGGREGATOR_REL_PATH.to_string(),
                body,
                frontmatter,
            });
        }
    }

    // Admit the complete prospective write set before any hook or write.
    // This is the Fortune 5 boundary: bounded count/bytes, safe canonical
    // targets, and no aliasing of one filesystem object through two spellings.
    if pending.len() > MAX_PENDING_OUTPUTS {
        return Err(AppError::fm_tpl(
            22,
            format!(
                "sync materialized {} outputs, over the Fortune 5 limit of                  {MAX_PENDING_OUTPUTS}. Remediation: partition the ontology view or                  aggregate rows into bounded artifacts.",
                pending.len()
            ),
        ));
    }
    let total_rendered_bytes = pending
        .iter()
        .fold(0usize, |total, pw| total.saturating_add(pw.body.len()));
    if total_rendered_bytes > MAX_TOTAL_RENDERED_BYTES {
        return Err(AppError::fm_tpl(
            22,
            format!(
                "sync rendered {total_rendered_bytes} body bytes, over the Fortune 5                  limit of {MAX_TOTAL_RENDERED_BYTES}. Remediation: reduce selected                  knowledge, split the run, or aggregate more compactly."
            ),
        ));
    }
    {
        let mut seen: BTreeMap<PathBuf, (String, usize)> = BTreeMap::new();
        for pw in &pending {
            let admitted = crate::write::resolve_target(root, &pw.to)?;
            let entry = seen.entry(admitted).or_insert_with(|| (pw.to.clone(), 0));
            entry.1 += 1;
        }
        if let Some((target, (declared, n))) = seen.into_iter().find(|(_, (_, n))| *n > 1) {
            return Err(AppError::fm_write(
                8,
                format!(
                    "{n} rendered templates resolve to the same admitted output `{}`                      (one declaration was `{declared}`). Remediation: make `to:` unique                      after path normalization; aliases such as `x` and `./x` are the                      same target.",
                    target.display()
                ),
            ));
        }
    }

    drop(generate_guard);
    generate_span.record(
        "pipeline.duration_ms",
        duration_ms(generate_start.elapsed()),
    );

    // ── Stage 5: Write every already-rendered template ───────────────────
    let emit_start = Instant::now();
    let emit_span = tracing::info_span!(
        "pipeline.emit",
        "operation.name" = "pipeline.emit",
        "operation.type" = "pipeline",
        "pipeline.stage" = "emit",
        "pipeline.duration_ms" = tracing::field::Empty,
        "pipeline.files_generated" = tracing::field::Empty,
    );
    let emit_guard = emit_span.enter();

    // Not transactional: each `apply()` call performs a real filesystem write
    // immediately. A mid-loop I/O failure (disk full, permission change,
    // ...) must not leave the files it *did* manage to write unrecorded by
    // the receipt -- that is a contract-drift hole (red-team finding F1):
    // `written`/`skipped`/`decisions` are populated incrementally by each
    // successful `apply()` call before this loop ever reaches a failing one,
    // so whatever they hold at the moment of failure is already an accurate
    // description of on-disk reality. Capture the failure instead of
    // propagating it immediately so a receipt can still be chained over that
    // accurate partial state below, before the error is surfaced to the
    // caller.
    let mut written: Vec<PathBuf> = Vec::new();
    let mut emit_err: Option<AppError> = None;
    for pw in &pending {
        if let Err(e) = apply(
            root,
            &pw.to,
            &pw.body,
            &pw.frontmatter,
            opts,
            &mut written,
            &mut skipped,
            &mut decisions,
        ) {
            emit_err = Some(e);
            break;
        }
    }

    drop(emit_guard);
    emit_span.record("pipeline.duration_ms", duration_ms(emit_start.elapsed()));
    emit_span.record("pipeline.files_generated", written.len() as u64);

    let pack_hashes: BTreeMap<String, String> = lock_entries
        .iter()
        .map(|e| {
            (
                e.name.clone(),
                e.content_hash.trim_start_matches("blake3:").to_string(),
            )
        })
        .collect();

    let report = SyncReport {
        written,
        skipped,
        graph_hash_hex,
        decisions,
        packs: pack_hashes,
        closure,
    };

    if !opts.dry_run {
        // Always chain a receipt over whatever was actually written, even
        // when the loop above stopped early: `write_receipt` re-hashes every
        // `report.decisions` target straight off disk (see its doc comment),
        // so a receipt written here after a partial failure still correctly
        // binds the files that *did* land, closing the drift window instead
        // of leaving the previous (now-stale) receipt as the last word on
        // them.
        if let Err(receipt_err) = write_receipt(root, &report, graph.as_ref(), opts.receipt_origin)
        {
            return Err(match emit_err {
                // The write-stage failure is the operative incident; a
                // receipt-write failure on top of it must be surfaced too
                // (never silently dropped), but must not replace it.
                Some(e) => AppError::fm_write(
                    12,
                    format!(
                        "write stage failed ({e}); additionally could not persist a \
                         receipt reflecting the outputs written before that failure \
                         ({receipt_err}). Remediation: inspect on-disk outputs manually \
                         (they are not yet bound by any receipt) before re-running sync."
                    ),
                ),
                None => receipt_err,
            });
        }
        // Portable receipt envelope (RFC-GPACK-001 §54/§55, §85 step 12):
        // additive alongside the BLAKE3 chain receipt (§56 — separate file,
        // never claimed byte-equivalent). REPORTS standing (§57): ALIVE when
        // the run completed, PARTIAL_ALIVE when the write stage stopped
        // early but a partial legacy receipt was still chained above.
        // `admission.gates_attempted` carries every gate actually evaluated
        // this run (§15); `replay` stays UNKNOWN (no replay court yet).
        let portable_standing = if emit_err.is_none() {
            crate::portable_receipt::PortableStanding::Alive
        } else {
            crate::portable_receipt::PortableStanding::PartialAlive
        };
        if let Err(env_err) = crate::portable_receipt::write_portable_envelope(
            root,
            &packs,
            &report.graph_hash_hex,
            &gates_attempted,
            &[],
            &portable_standing,
            &report.decisions,
        ) {
            return Err(match emit_err {
                // The write-stage failure is the operative incident; an
                // envelope-write failure on top of it must be surfaced too
                // (never silently dropped), but must not replace it.
                Some(e) => AppError::fm_write(
                    12,
                    format!(
                        "write stage failed ({e}); additionally could not persist \
                         the portable receipt envelope ({env_err}). Remediation: \
                         inspect on-disk outputs manually before re-running sync."
                    ),
                ),
                None => env_err,
            });
        }
        // The pack lockfile describes a fully-materialized run; skip it when
        // the write stage itself did not complete, rather than lock onto a
        // set of pack versions that only partially produced their outputs.
        if emit_err.is_none() && !lock_entries.is_empty() {
            crate::pack::write_lock(root, &lock_entries)?;
        }
    }
    if let Some(e) = emit_err {
        return Err(e);
    }
    Ok(report)
}

// ---------------------------------------------------------------------------
// SPARQL gate queries — the engine-independent sync gate convention
// ---------------------------------------------------------------------------

/// A parsed SPARQL gate file: optional leading `# MESSAGE: <text>` comment
/// line(s) (the operator-facing violation text), then one ASK or SELECT
/// query. ASK → `true` is a violation; SELECT → any returned row is a
/// violation. Used by `[law].gates`, pack `gates/*.rq`, and the
/// declarative-rules `[validation].gates` (see `crate::generation_rules`).
pub(crate) struct GateFile {
    /// Joined text of the leading `# MESSAGE:` lines, if any.
    pub message: Option<String>,
    /// The query text (everything after the leading MESSAGE block —
    /// further `#` lines are ordinary SPARQL comments and stay in place).
    pub query: String,
}

impl GateFile {
    /// `"<message>: "` when a MESSAGE was declared, `""` otherwise —
    /// prefix form for violation error text.
    pub(crate) fn message_prefix(&self) -> String {
        self.message
            .as_deref()
            .map(|m| format!("{m}: "))
            .unwrap_or_default()
    }
}

/// Split a gate file's source into its optional leading `# MESSAGE:` block
/// and the query text. Only *leading* `# MESSAGE:` lines are consumed;
/// everything from the first non-MESSAGE line onward (including plain `#`
/// comments, which are valid SPARQL) is the query.
pub(crate) fn parse_gate_source(src: &str) -> GateFile {
    let mut message_lines: Vec<String> = Vec::new();
    let mut query_lines: Vec<&str> = Vec::new();
    let mut in_header = true;
    for line in src.lines() {
        if in_header {
            if let Some(rest) = line.trim_start().strip_prefix("# MESSAGE:") {
                message_lines.push(rest.trim().to_string());
                continue;
            }
            in_header = false;
        }
        query_lines.push(line);
    }
    GateFile {
        message: if message_lines.is_empty() {
            None
        } else {
            Some(message_lines.join(" "))
        },
        query: query_lines.join("\n"),
    }
}

/// Outcome of evaluating one gate query.
pub(crate) enum GateOutcome {
    /// ASK returned `false`, or SELECT returned zero rows.
    Pass,
    /// ASK returned `true`, or SELECT returned ≥1 row — the detail string
    /// carries the row count and the first row's bindings.
    Violation(String),
    /// The query produced CONSTRUCT/DESCRIBE results — not a gate; the
    /// caller refuses with its own typed "invalid gate file" code.
    NotAGate,
}

/// Evaluate one gate query against `graph` via [`GraphEngine::query`] —
/// plain SPARQL, so the identical gate refuses the identical facts under
/// both the `GraphLaw` and the Oxigraph engine.
///
/// # Errors
/// Propagates query parse/evaluation failures (fail closed).
pub(crate) fn evaluate_gate(graph: &dyn GraphEngine, query: &str) -> Result<GateOutcome> {
    match graph.query(query)? {
        EngineQueryResults::Boolean(false) => Ok(GateOutcome::Pass),
        EngineQueryResults::Boolean(true) => {
            Ok(GateOutcome::Violation("ASK returned true".to_string()))
        }
        EngineQueryResults::Solutions(rows) => match rows.first() {
            None => Ok(GateOutcome::Pass),
            Some(first) => {
                let bindings: Vec<String> = first
                    .iter()
                    .map(|(var, value)| format!("?{var} = {}", engine_value_display(value)))
                    .collect();
                Ok(GateOutcome::Violation(format!(
                    "SELECT returned {} row(s); first row: {{ {} }}",
                    rows.len(),
                    bindings.join(", ")
                )))
            }
        },
        EngineQueryResults::Graph(_) => Ok(GateOutcome::NotAGate),
    }
}

/// Plain display form of an [`EngineValue`] for gate violation messages.
fn engine_value_display(value: &crate::graph::EngineValue) -> String {
    match value {
        crate::graph::EngineValue::Bool(b) => b.to_string(),
        crate::graph::EngineValue::Int(n) => n.to_string(),
        crate::graph::EngineValue::Float(f) => f.to_string(),
        crate::graph::EngineValue::String(s) => s.clone(),
    }
}

/// Consumer-mode (OS-13): collect every subject IRI carrying a
/// `fixtureOnly`-marker predicate (any predicate whose IRI contains
/// `fixtureOnly` — covers `aex:fixtureOnly` and the mirrored
/// `pr:priorArtFixtureOnly` idiom) with the literal `true` from the
/// post-enrich union graph.
///
/// # Errors
/// Propagates any [`crate::error::AppError`] from the graph query.
fn fixture_only_subjects(
    graph: &dyn crate::graph::GraphEngine,
) -> Result<std::collections::BTreeSet<String>> {
    const QUERY: &str =
        "SELECT ?s WHERE { ?s ?p true . FILTER(CONTAINS(STR(?p), \"fixtureOnly\")) }";
    match graph.query(QUERY)? {
        EngineQueryResults::Solutions(rows) => Ok(rows
            .into_iter()
            .filter_map(|row| match row.get("s") {
                Some(EngineValue::String(iri)) => Some(iri.clone()),
                _ => None,
            })
            .collect()),
        // A marker query is a SELECT by construction; the other result
        // shapes cannot arise from this query text.
        EngineQueryResults::Graph(_) | EngineQueryResults::Boolean(_) => {
            Ok(std::collections::BTreeSet::default())
        }
    }
}

/// Consumer-mode (OS-13): does this rendered row/named-result JSON name a
/// `fixtureOnly`-marked subject? Returns the first matched subject IRI (in
/// `BTreeSet` order, so the skip reason is deterministic) for the typed skip
/// log.
fn fixture_reference_in(
    value: &serde_json::Value, subjects: &std::collections::BTreeSet<String>,
) -> Option<String> {
    use serde_json::Value as J;
    match value {
        J::String(s) => subjects
            .iter()
            .find(|iri| s.contains(iri.as_str()))
            .cloned(),
        J::Array(items) => items.iter().find_map(|i| fixture_reference_in(i, subjects)),
        J::Object(map) => map.values().find_map(|v| fixture_reference_in(v, subjects)),
        J::Null | J::Bool(_) | J::Number(_) => None,
    }
}

/// The sorted `*.rq` files under a pack's `gates/` directory. A missing
/// directory is simply "no gates" (empty vec); a directory that exists but
/// cannot be enumerated is a typed `[FM-PACK-012]` refusal — an existing
/// gate directory whose contents cannot be seen must never be silently
/// treated as gateless (fail-open).
fn list_gate_files(gates_dir: &Path, pack_name: &str) -> Result<Vec<PathBuf>> {
    if !gates_dir.is_dir() {
        return Ok(Vec::new());
    }
    let entries = std::fs::read_dir(gates_dir).map_err(|e| {
        AppError::fm_pack(
            12,
            format!(
                "pack `{pack_name}`: gates directory `{}` exists but is unreadable: {e}. \
                 Remediation: fix the directory's permissions or remove it.",
                gates_dir.display()
            ),
        )
    })?;
    let mut paths: Vec<PathBuf> = Vec::new();
    for entry in entries {
        let path = entry
            .map_err(|e| {
                AppError::fm_pack(
                    12,
                    format!(
                        "pack `{pack_name}`: gates directory `{}` entry unreadable: {e}.",
                        gates_dir.display()
                    ),
                )
            })?
            .path();
        if path.is_file() && path.extension().is_some_and(|e| e == "rq") {
            paths.push(path);
        }
    }
    paths.sort();
    Ok(paths)
}

/// Expand `[law].gates` entries into concrete gate files, in evaluation
/// order. A literal entry (no `*`/`?`) is joined to the manifest root
/// unchanged. An entry containing `*` or `?` is a glob pattern expanded
/// against paths relative to the manifest dir; matches are sorted so the
/// documented gate-precedes-writes order stays deterministic. A glob that
/// matches nothing is a typed, fail-closed `[FM-CONFIG-014]` refusal — a
/// typo'd pattern must never silently pass zero gates.
fn resolve_gate_entries(root: &Path, gates: &[PathBuf]) -> Result<Vec<(String, PathBuf)>> {
    let mut resolved: Vec<(String, PathBuf)> = Vec::new();
    for rel in gates {
        let pattern = rel.to_string_lossy();
        if !pattern.contains('*') && !pattern.contains('?') {
            let path = root.join(rel);
            resolved.push((rel_display(root, &path), path));
            continue;
        }
        let matches = expand_gate_glob(root, &pattern)?;
        if matches.is_empty() {
            return Err(AppError::fm_config(
                14,
                format!(
                    "GLOB_NO_MATCHES: `{pattern}` matched no files under {}. \
                     Remediation: fix the [law].gates pattern, or create the gate \
                     files it names.",
                    root.display()
                ),
            ));
        }
        for matched in matches {
            let path = root.join(&matched);
            resolved.push((rel_display(root, &path), path));
        }
    }
    Ok(resolved)
}

/// Sorted root-relative file paths under `root` matching a `*`/`?` glob.
/// `*` matches any run of characters within one path component (it never
/// crosses `/`); `?` matches exactly one character. The whole tree under
/// `root` is walked (gate trees are small); an unreadable directory is a
/// typed refusal, not silently-no-matches (fail-closed, same discipline as
/// [`list_gate_files`]).
fn expand_gate_glob(root: &Path, pattern: &str) -> Result<Vec<String>> {
    fn walk(dir: &Path, rel: &str, out: &mut Vec<String>) -> std::io::Result<()> {
        let mut entries = std::fs::read_dir(dir)?;
        let mut names: Vec<(String, std::path::PathBuf)> = Vec::new();
        for entry in &mut entries {
            let entry = entry?;
            names.push((
                entry.file_name().to_string_lossy().into_owned(),
                entry.path(),
            ));
        }
        // Enumerate deterministically so `sort` below is the only order that
        // matters, not the filesystem's.
        names.sort();
        for (name, path) in names {
            let child_rel = if rel.is_empty() {
                name
            } else {
                format!("{rel}/{name}")
            };
            if path.is_dir() {
                walk(&path, &child_rel, out)?;
            } else {
                out.push(child_rel);
            }
        }
        Ok(())
    }

    fn match_component(pat: &[char], text: &[char]) -> bool {
        match (pat.first(), text.first()) {
            (None, None) => true,
            (Some('*'), _) => {
                // Greedy-with-backtrack: '*' absorbs 0..=n chars of this component.
                (0..=text.len()).any(|skip| match_component(&pat[1..], &text[skip..]))
            }
            (Some('?'), Some(_)) => match_component(&pat[1..], &text[1..]),
            (Some(p), Some(t)) if p == t => match_component(&pat[1..], &text[1..]),
            _ => false,
        }
    }

    fn match_components(pat: &[&str], text: &[&str]) -> bool {
        match (pat.first(), text.first()) {
            (None, None) => true,
            (None, Some(_)) | (Some(_), None) => false,
            (Some(p), Some(t)) => {
                let pc: Vec<char> = p.chars().collect();
                let tc: Vec<char> = t.chars().collect();
                match_component(&pc, &tc) && match_components(&pat[1..], &text[1..])
            }
        }
    }

    let pat_parts: Vec<&str> = pattern
        .split('/')
        .filter(|p| !p.is_empty() && *p != ".")
        .collect();
    let mut files: Vec<String> = Vec::new();
    walk(root, "", &mut files).map_err(|e| {
        AppError::fm_law(
            12,
            format!(
                "[law].gates glob `{pattern}` could not enumerate files under {}: {e}. \
                 Remediation: fix the directory's permissions or the pattern.",
                root.display()
            ),
        )
    })?;
    files.retain(|f| match_components(&pat_parts, f.split('/').collect::<Vec<_>>().as_slice()));
    files.sort();
    Ok(files)
}

/// Emit the portable refusal envelope (RFC-GPACK-001 §54/§55) for a
/// gate-stage refusal, then return `original` unchanged — the typed
/// `[FM-LAW-*]`/`[FM-PACK-*]` refusal stays the operative incident.
///
/// §57: the envelope existing is not standing — it REPORTS standing, so a
/// refused sync's envelope carries the typed `REFUSED:<code>` identity in
/// `admission.refusals` and `REFUSED:<code>` as `standing`, with empty
/// `consequences` (gates precede every write). `graph.canonical_digest` is
/// computed from the live graph (`"UNKNOWN"` if that itself fails — §54:
/// present, never invented). A dry run writes no envelope at all (zero
/// side effects, same contract as the legacy receipt).
///
/// If the envelope cannot be persisted, the original refusal is returned
/// wrapped with the additional failure (never silently dropped, never
/// replacing the original).
fn refuse_with_portable_envelope(
    root: &Path, packs: &[crate::pack::Pack], graph: &dyn GraphEngine, gates_attempted: &[String],
    refusal_code: &str, dry_run: bool, original: AppError,
) -> AppError {
    if dry_run {
        return original;
    }
    let graph_hash = graph
        .state_hash()
        .map_or_else(|_| "UNKNOWN".to_string(), |h| hex32(&h));
    let standing = crate::portable_receipt::PortableStanding::Refused(refusal_code.to_string());
    match crate::portable_receipt::write_portable_envelope(
        root,
        packs,
        &graph_hash,
        gates_attempted,
        &[format!("REFUSED:{refusal_code}")],
        &standing,
        &BTreeMap::new(),
    ) {
        Ok(()) => original,
        Err(env_err) => AppError::Validation(format!(
            "{original}; additionally, the portable refusal envelope could not be \
             written ({env_err})"
        )),
    }
}

/// Root-relative display form of a closure input path (falls back to the
/// full path for inputs outside the project root, e.g. absolute pack dirs).
///
/// `pub(crate)`: also used by `crate::generation_rules` to bind its own
/// input closure entries (ontology imports, query/template files) the exact
/// same way.
pub(crate) fn rel_display(root: &Path, path: &Path) -> String {
    path.strip_prefix(root)
        .unwrap_or(path)
        .display()
        .to_string()
}

/// BLAKE3 hex of a closure input's bytes, or the literal `MISSING` marker
/// when the declared input cannot be read — recorded, never dropped.
///
/// `pub(crate)`: also used by `crate::generation_rules` (see
/// [`rel_display`]'s doc comment).
pub(crate) fn hash_file_or_missing(path: &Path) -> String {
    match std::fs::read(path) {
        Ok(bytes) => blake3::hash(&bytes).to_hex().to_string(),
        Err(_) => "MISSING".to_string(),
    }
}

/// The outcome of [`extract_query_results`] when the `when:` guard (if any)
/// evaluated `true`: the named `sparql:` results plus the driving row array.
/// `None` (the outer [`Option`] returned by [`extract_query_results`] itself)
/// means the `when:` guard evaluated `false` — the template must be skipped.
type ExtractedRows = Option<(BTreeMap<String, Value>, Vec<Value>)>;

/// Run this template's `when:` ASK guard and named `sparql:` SELECTs against
/// `active_graph`. Returns `Ok(None)` when a `when:` guard is present and
/// evaluates `false` (the template must be skipped this run); otherwise
/// `Ok(Some((named, results)))`.
///
/// Shared by the per-template loop's primary extraction (against `graph`)
/// and [`check_determinism`]'s second, independent recheck (against a
/// separate, independently-loaded graph built by
/// [`build_independent_recheck_base_graph`]) — the exact same code path
/// runs both times, but against two genuinely distinct `GraphEngine`
/// instances, so `determinism: true` catches both query-execution
/// nondeterminism (differing bindings/ordering across two calls) AND
/// load-time nondeterminism (e.g. oxigraph's `rename_blank_nodes()`
/// assigning a fresh random id per parse — see
/// `build_independent_recheck_base_graph`'s doc comment), not merely
/// Tera-render nondeterminism against a single, already-fixed context.
///
/// `phase` (`"primary"` / `"determinism_recheck"`) is a `tracing` tag only —
/// no behavior depends on it. It lets a test observe, via a
/// `tracing::Subscriber`, that a determinism recheck genuinely re-executes
/// every declared query rather than re-rendering cached values.
///
/// # Errors
/// `[FM-TPL-004]` if `when:` is present and is not an ASK query.
/// Propagates `[FM-GRAPH-*]` from the underlying `GraphEngine::query` calls.
fn extract_query_results(
    active_graph: &dyn GraphEngine, tpl_path: &Path, frontmatter: &Frontmatter, phase: &'static str,
) -> Result<ExtractedRows> {
    if let Some(ask) = frontmatter.when.as_deref() {
        tracing::debug!(
            target: "ggen_engine::sync::query_extraction",
            phase,
            tpl = %tpl_path.display(),
            "executing `when:` ASK guard"
        );
        match active_graph.query(ask)? {
            EngineQueryResults::Boolean(true) => {}
            EngineQueryResults::Boolean(false) => return Ok(None),
            _ => {
                return Err(AppError::fm_tpl(
                    16,
                    format!(
                        "`when:` in {} must be an ASK query. \
                         Remediation: use ASK {{ … }}.",
                        tpl_path.display()
                    ),
                ));
            }
        }
    }
    let mut named: BTreeMap<String, Value> = BTreeMap::new();
    for (key, query) in &frontmatter.sparql {
        tracing::debug!(
            target: "ggen_engine::sync::query_extraction",
            phase,
            tpl = %tpl_path.display(),
            sparql_key = %key,
            "executing named sparql query"
        );
        let value = sparql_to_value(active_graph, query)?;
        if let Some(rows) = value.as_array() {
            if rows.len() > MAX_QUERY_RESULT_ROWS {
                return Err(AppError::fm_tpl(
                    21,
                    format!(
                        "{}: named `sparql:` result `{key}` produced {} row(s), over the                          Fortune 5 limit of {MAX_QUERY_RESULT_ROWS}. Remediation: constrain                          the query with a selective WHERE clause, grouping, or an explicit LIMIT.",
                        tpl_path.display(),
                        rows.len()
                    ),
                ));
            }
        }
        named.insert(key.clone(), value);
    }
    let results: Vec<Value> = if let Some(driver) = frontmatter.for_each.as_deref() {
        let value = named.get(driver).ok_or_else(|| {
            AppError::fm_tpl(
                19,
                format!(
                    "{}: `for_each: {driver}` does not name a declared `sparql:` result. \
                     Remediation: add that named query or correct `for_each`.",
                    tpl_path.display()
                ),
            )
        })?;
        value.as_array().cloned().ok_or_else(|| {
            AppError::fm_tpl(
                19,
                format!(
                    "{}: `for_each: {driver}` must name an array-valued SELECT/graph result, \
                     but that query returned {}. Remediation: point `for_each` at a SELECT \
                     result that yields the intended projection rows.",
                    tpl_path.display(),
                    value_type_name(value)
                ),
            )
        })?
    } else {
        // Historical compatibility: the first named query in BTreeMap key
        // order with an array result remains the implicit driving row set.
        named
            .values()
            .find_map(|v| v.as_array().cloned())
            .unwrap_or_default()
    };
    Ok(Some((named, results)))
}

/// Build the shared Tera context: `results` plus every named sparql key.
fn base_context(named: &BTreeMap<String, Value>, results: &[Value]) -> tera::Context {
    let mut ctx = tera::Context::new();
    ctx.insert("results", results);
    for (k, v) in named {
        ctx.insert(k, v);
    }
    ctx
}

fn row_context(named: &BTreeMap<String, Value>, results: &[Value], row: &Value) -> tera::Context {
    let mut ctx = base_context(named, results);
    ctx.insert("row", row);
    if let Value::Object(map) = row {
        for (k, v) in map {
            ctx.insert(k, v);
        }
    }
    ctx
}

/// Render a Tera string, mapping errors to `[FM-TPL-017]`.
///
/// Uses `template::tera_error_full_chain`, not bare `{e}` Display: Tera's
/// top-level `Display` is frequently just "Failed to render
/// '`__tera_one_off`'" with the actual cause (unknown filter, missing
/// variable, wrong argument type) only reachable via `Error::source()`
/// chaining — the same gap `generation_rules.rs`'s `[FM-GEN-008]` path
/// already closed with this same helper; this call site had not been
/// updated to match, so every `[FM-TPL-017]` render failure printed a
/// content-free message.
fn render_str(
    tera: &mut tera::Tera, template: &str, ctx: &tera::Context, tpl_path: &Path,
) -> Result<String> {
    tera.render_str(template, ctx).map_err(|e| {
        AppError::fm_tpl(
            17,
            format!(
                "render failed for {}: {}. Available top-level context keys: {}.",
                tpl_path.display(),
                crate::template::tera_error_full_chain(&e),
                context_key_summary(ctx)
            ),
        )
    })
}

/// Render one optional output-phase string through the same Tera context
/// as the output path and body.
fn render_optional_output_field(
    tera: &mut tera::Tera, value: Option<&str>, ctx: &tera::Context, tpl_path: &Path,
) -> Result<Option<String>> {
    value
        .map(|template| render_str(tera, template, ctx, tpl_path))
        .transpose()
}

/// Render only the pattern inside a literal or structured match declaration,
/// preserving its matcher/scope/occurrence defaults and explicit overrides.
fn render_optional_match_field(
    tera: &mut tera::Tera, value: Option<&MatchSpec>, ctx: &tera::Context, tpl_path: &Path,
) -> Result<Option<MatchSpec>> {
    value
        .map(|spec| {
            render_str(tera, spec.pattern(), ctx, tpl_path)
                .map(|pattern| spec.with_pattern(pattern))
        })
        .transpose()
}

/// Compile the output-phase frontmatter for one materialized projection.
///
/// Resolve/enrich/extract fields remain static because they determine the
/// semantic context itself. Structural markers, idempotence needles,
/// hooks, shape paths, and checksum-slot paths are rendered after query
/// extraction so one ontology row can lawfully specialize the complete
/// write lifecycle rather than only `to:` and the body.
fn render_output_frontmatter(
    tera: &mut tera::Tera, source: &Frontmatter, ctx: &tera::Context, tpl_path: &Path,
    rendered_to: &str,
) -> Result<Frontmatter> {
    let mut rendered = source.clone();
    rendered.to = rendered_to.to_string();
    rendered.before = render_optional_match_field(tera, source.before.as_ref(), ctx, tpl_path)?;
    rendered.after = render_optional_match_field(tera, source.after.as_ref(), ctx, tpl_path)?;
    rendered.skip_if = render_optional_match_field(tera, source.skip_if.as_ref(), ctx, tpl_path)?;
    rendered.sh_before =
        render_optional_output_field(tera, source.sh_before.as_deref(), ctx, tpl_path)?;
    rendered.sh_after =
        render_optional_output_field(tera, source.sh_after.as_deref(), ctx, tpl_path)?;
    rendered.freeze_slots_dir =
        render_optional_output_field(tera, source.freeze_slots_dir.as_deref(), ctx, tpl_path)?;
    rendered.shape = source
        .shape
        .iter()
        .map(|shape| render_str(tera, shape, ctx, tpl_path))
        .collect::<Result<Vec<_>>>()?;
    Ok(rendered)
}

fn render_aggregate_projection(
    tera: &mut tera::Tera, body_template: &str, tpl_path: &Path, frontmatter: &Frontmatter,
    named: &BTreeMap<String, Value>, results: &[Value],
) -> Result<Option<(String, Frontmatter)>> {
    if results.is_empty() {
        return Ok(None);
    }
    let mut body = String::new();
    let mut canonical_frontmatter: Option<Frontmatter> = None;
    for (row_index, row) in results.iter().enumerate() {
        let ctx = row_context(named, results, row);
        body.push_str(&render_str(tera, body_template, &ctx, tpl_path)?);
        let candidate =
            render_output_frontmatter(tera, frontmatter, &ctx, tpl_path, &frontmatter.to)?;
        if let Some(expected) = &canonical_frontmatter {
            if expected != &candidate {
                return Err(AppError::fm_tpl(
                    20,
                    format!(
                        "{}: static `to:` with `for_each` produced row-varying output-phase \
                         frontmatter at row {row_index}. One target must have one composition, \
                         hook, shape, and ownership law. Remediation: make those properties \
                         invariant across rows or include a row binding in `to:` to fan out.",
                        tpl_path.display()
                    ),
                ));
            }
        } else {
            canonical_frontmatter = Some(candidate);
        }
    }
    Ok(canonical_frontmatter.map(|frontmatter| (body, frontmatter)))
}

/// Summarize a Tera context's top-level keys (and, for arrays, their
/// length) for a render-failure error message — enough to spot a typo'd
/// variable name (`row.nuon` vs `row.noun`) without dumping the full,
/// potentially large, context payload.
fn context_key_summary(ctx: &tera::Context) -> String {
    let Value::Object(map) = ctx.clone().into_json() else {
        return "(none)".to_string();
    };
    if map.is_empty() {
        return "(none)".to_string();
    }
    map.iter()
        .map(|(k, v)| match v {
            Value::Array(items) => format!("{k} (array, {} items)", items.len()),
            Value::Object(fields) => format!("{k} (object, {} fields)", fields.len()),
            other => format!("{k} ({})", value_type_name(other)),
        })
        .collect::<Vec<_>>()
        .join(", ")
}

fn value_type_name(v: &Value) -> &'static str {
    match v {
        Value::Null => "null",
        Value::Bool(_) => "bool",
        Value::Number(_) => "number",
        Value::String(_) => "string",
        Value::Array(_) => "array",
        Value::Object(_) => "object",
    }
}

/// Apply (or dry-run) one write and record the outcome and its decision.
#[allow(clippy::too_many_arguments)]
fn apply(
    root: &Path, rel_to: &str, body: &str, frontmatter: &Frontmatter, opts: SyncOptions,
    written: &mut Vec<PathBuf>, skipped: &mut Vec<(PathBuf, String)>,
    decisions: &mut BTreeMap<String, String>,
) -> Result<()> {
    let target = crate::write::resolve_target(root, rel_to)?;
    if frontmatter.skip_empty && body.trim().is_empty() {
        let reason = "skip_empty: rendered body empty".to_string();
        decisions.insert(rel_to.to_string(), format!("skipped: {reason}"));
        skipped.push((PathBuf::from(rel_to), reason));
        return Ok(());
    }
    // Matcher syntax/configuration is admitted before any shell hook or
    // filesystem actuation. Structured cardinality is then observed against
    // the actual host file.
    validate_match_specs(frontmatter)?;
    preflight_checksum_slot(root, rel_to, frontmatter)?;
    let mut match_evidence = preflight_structured_matchers(root, rel_to, frontmatter)?;

    if opts.dry_run {
        // Dry run: classify without touching the filesystem or running
        // sh_before/sh_after (a dry run must have zero side effects).
        // Identical existing content reports Skipped(unchanged); everything
        // else is reported as a planned write.
        match std::fs::read_to_string(&target) {
            Ok(existing) if existing == body => {
                let reason = "unchanged: content identical".to_string();
                decisions.insert(rel_to.to_string(), format!("skipped: {reason}"));
                skipped.push((PathBuf::from(rel_to), reason));
            }
            Ok(_) => {
                let suffix = match_evidence_suffix(&match_evidence);
                decisions.insert(
                    rel_to.to_string(),
                    format!("planned: write (dry-run){suffix}"),
                );
                written.push(PathBuf::from(rel_to));
            }
            Err(e) if e.kind() == std::io::ErrorKind::NotFound => {
                let suffix = match_evidence_suffix(&match_evidence);
                decisions.insert(
                    rel_to.to_string(),
                    format!("planned: write (dry-run){suffix}"),
                );
                written.push(PathBuf::from(rel_to));
            }
            Err(e) => {
                // An existing target that cannot be read as UTF-8 (or at
                // all) is not a plain "planned: write" — the real run would
                // fail closed here, and a dry run must not classify it
                // more optimistically than the run it predicts.
                return Err(AppError::fm_write(
                    9,
                    format!(
                        "dry-run: target `{}` exists but is unreadable as UTF-8: {e}. \
                         A non-dry-run sync would refuse here too. \
                         Remediation: remove or fix the target file.",
                        target.display()
                    ),
                ));
            }
        }
        return Ok(());
    }

    if let Some(cmd) = frontmatter.sh_before.as_deref() {
        run_shell_hook(root, cmd, "sh_before")?;
        // Bind the post-hook host state used by the write decision. A hook
        // that changes selector cardinality cannot silently reuse stale
        // preflight evidence.
        match_evidence = preflight_structured_matchers(root, rel_to, frontmatter)?;
    }

    let evidence_suffix = match_evidence_suffix(&match_evidence);
    match plan_write(root, rel_to, body, frontmatter)? {
        WriteOutcome::Written => {
            decisions.insert(rel_to.to_string(), format!("written{evidence_suffix}"));
            written.push(PathBuf::from(rel_to));
            if let Some(cmd) = frontmatter.sh_after.as_deref() {
                run_shell_hook(root, cmd, "sh_after")?;
            }
        }
        WriteOutcome::Injected => {
            decisions.insert(rel_to.to_string(), format!("injected{evidence_suffix}"));
            written.push(PathBuf::from(rel_to));
            if let Some(cmd) = frontmatter.sh_after.as_deref() {
                run_shell_hook(root, cmd, "sh_after")?;
            }
        }
        WriteOutcome::Skipped(reason) => {
            decisions.insert(
                rel_to.to_string(),
                format!("skipped: {reason}{evidence_suffix}"),
            );
            skipped.push((PathBuf::from(rel_to), reason));
        }
    }
    Ok(())
}

fn match_evidence_suffix(evidence: &[String]) -> String {
    if evidence.is_empty() {
        String::new()
    } else {
        format!("; match={}", evidence.join(" | "))
    }
}

/// Refuse `cmd` against the shell-command denylist, then run it with `root`
/// as its working directory. Non-zero exit is a hard error.
fn run_shell_hook(root: &Path, cmd: &str, which: &str) -> Result<()> {
    crate::shell_safety::check_shell_command_safe(cmd)?;
    let status = std::process::Command::new("sh")
        .arg("-c")
        .arg(cmd)
        .current_dir(root)
        .status()
        .map_err(|e| AppError::fm_shell(2, format!("{which} failed to launch: {e}")))?;
    if !status.success() {
        return Err(AppError::fm_shell(
            3,
            format!("{which} command `{cmd}` exited with {status}. Remediation: fix the command or remove it from frontmatter."),
        ));
    }
    Ok(())
}

/// Discover and parse every template a sync would process: project
/// templates (recursive under `[templates].dir`, sorted) followed by each
/// resolved pack's templates in pack order. Shared by [`sync`] and
/// `ggen graph validate` so both surfaces see the identical template set.
///
/// # Errors
/// Unreadable templates dir or any template parse failure (fail closed).
pub fn discover_templates(
    root: &Path, config: &GgenConfig, packs: &[crate::pack::Pack],
) -> Result<Vec<(PathBuf, Template)>> {
    let mut templates = load_templates(&root.join(&config.templates.dir))?;
    for pack in packs {
        for path in &pack.template_paths {
            let content = std::fs::read_to_string(path)?;
            if !content
                .strip_prefix('\u{feff}')
                .unwrap_or(&content)
                .starts_with("---")
            {
                // Rules-consumed template (see load_templates): the consumer's
                // own [[generation.rules]] renders it; self-discovery skips it.
                continue;
            }
            templates.push((path.clone(), parse_template_file(path)?));
        }
    }
    Ok(templates)
}

/// Load and parse every `*.tmpl` under `dir` (recursive), in sorted path
/// order for deterministic processing.
///
/// Files that do not begin with a `---` frontmatter block are skipped, not
/// refused: they are `[[generation.rules]]`-consumed templates (the repo's
/// documented convention — see packs/github-actions-pack/templates/
/// workflow.yml.tmpl's header comment and .specify/templates/*.tera), whose
/// rendering is driven entirely by a consumer's own generation rule and
/// which by design carry no frontmatter. Attempting to frontmatter-parse
/// them here made every consumer of such a pack refuse with FM-TPL-006
/// even when the consumer never self-discovers those templates.
fn load_templates(dir: &Path) -> Result<Vec<(PathBuf, Template)>> {
    let mut paths: Vec<PathBuf> = Vec::new();
    collect_tmpl_paths(dir, &mut paths)?;
    paths.sort();
    let mut out = Vec::with_capacity(paths.len());
    for path in paths {
        let content = std::fs::read_to_string(&path)?;
        if !content
            .strip_prefix('\u{feff}')
            .unwrap_or(&content)
            .starts_with("---")
        {
            continue;
        }
        let tpl = parse_template_file(&path)?;
        out.push((path, tpl));
    }
    Ok(out)
}

/// Read and parse one `*.tmpl` file (identical frontmatter contract for
/// project and pack templates — no special casing). If the frontmatter sets
/// `from:`, the Tera body is replaced with the content of that path
/// (resolved relative to this template file's own directory); frontmatter
/// fields still come from `path` itself.
///
/// # Errors
/// `[FM-TPL-008]` when `from:` escapes the template's own directory —
/// distinct from [`check_shape_files_exist`]'s `[FM-TPL-014]` (a missing
/// `shape:` file); the two used to share this numeric code despite being
/// unrelated causes.
fn parse_template_file(path: &Path) -> Result<Template> {
    let content = std::fs::read_to_string(path)?;
    let mut tpl = Template::parse(&content)
        .map_err(|e| AppError::fm_tpl(6, format!("{}: {e}", path.display())))?;
    if let Some(from) = tpl.frontmatter.from.clone() {
        let base = path.parent().unwrap_or_else(|| Path::new("."));
        let from_path = crate::write::resolve_target(base, &from).map_err(|e| {
            AppError::fm_tpl(
                8,
                format!(
                    "{}: `from: {from}` is not a safe path (must stay inside the \
                     template's own directory): {e}",
                    path.display()
                ),
            )
        })?;
        tpl.body = std::fs::read_to_string(&from_path).map_err(|e| {
            AppError::fm_tpl(
                7,
                format!(
                    "{}: `from: {from}` unreadable at `{}`: {e}. \
                     Remediation: fix the `from:` path (relative to the template's own directory).",
                    path.display(),
                    from_path.display()
                ),
            )
        })?;
    }
    Ok(tpl)
}

/// Admit every rendered `shape:` path as a safe, regular, readable,
/// receipt-bound governing input, AND validate `active_graph` (the
/// per-template RDF overlay when `rdf:`/`rdf_inline:` declared one, else the
/// shared project graph -- exactly the graph this template's own `sparql:`
/// queries run against) against the union of all declared shape documents,
/// via the same [`crate::graph::GraphEngine::validate_shacl`] the
/// `ggen graph validate`/`ggen law validate` CLI verbs already use.
///
/// LEAD 1 of the 2026-08-03 unverified-leads audit, confirmed and closed
/// here: "`shape:` is hashed but never SHACL-validated while a working
/// engine sits unused." Before this, a `shape:` declaration was
/// existence-checked only -- its bytes joined the receipt's input closure
/// (a "governing" input, tracked for drift across syncs) as if it were
/// enforced, but the real `praxis-graphlaw` SHACL engine was never invoked
/// against it. Confirmed non-hypothetical: `templates/cli/subcommand/
/// rust.tmpl` and `templates/api/endpoint/rust.tmpl` both declare real
/// `shape:` constraints (`graphs/shapes/{cli,api}.shacl.ttl`) that were
/// admitted but never checked. A `shape:` declaration now fails the sync
/// closed (typed `FM-TPL-025`) on non-conformance, or `FM-TPL-026` on an
/// engine with no SHACL support (e.g. `--engine oxigraph`), matching the
/// combination pattern `verbs::handlers::validate_files` already uses for
/// `ggen graph validate --shapes`.
fn admit_shape_files(
    root: &Path, tpl_path: &Path, shapes: &[String], active_graph: &dyn GraphEngine,
    closure: &mut BTreeMap<String, String>,
) -> Result<()> {
    if shapes.is_empty() {
        return Ok(());
    }
    let mut combined_shapes_ttl = String::new();
    for shape in shapes {
        let shape_path = crate::write::resolve_target(root, shape).map_err(|error| {
            AppError::fm_tpl(
                14,
                format!(
                    "{}: `shape:` entry `{shape}` is not an admitted project path:                      {error}. Remediation: use a safe relative path inside the project.",
                    tpl_path.display()
                ),
            )
        })?;
        let metadata = std::fs::metadata(&shape_path).map_err(|error| {
            AppError::fm_tpl(
                14,
                format!(
                    "{}: `shape:` entry `{shape}` is unreadable at `{}`: {error}.                      Remediation: provide a readable regular file.",
                    tpl_path.display(),
                    shape_path.display()
                ),
            )
        })?;
        if !metadata.is_file() {
            return Err(AppError::fm_tpl(
                14,
                format!(
                    "{}: `shape:` entry `{shape}` resolves to `{}`, which is not a                      regular file. Remediation: point to a readable shape document.",
                    tpl_path.display(),
                    shape_path.display()
                ),
            ));
        }
        let bytes = std::fs::read(&shape_path).map_err(|error| {
            AppError::fm_tpl(
                14,
                format!(
                    "{}: `shape:` entry `{shape}` could not be read at `{}`: {error}",
                    tpl_path.display(),
                    shape_path.display()
                ),
            )
        })?;
        // resolve_target canonicalizes (e.g. /tmp -> /private/tmp on macOS),
        // so strip against the canonicalized root or the key leaks the
        // absolute path and the closure map stops being project-relative.
        let closure_root = root.canonicalize().unwrap_or_else(|_| root.to_path_buf());
        closure.insert(
            rel_display(&closure_root, &shape_path),
            blake3::hash(&bytes).to_hex().to_string(),
        );
        let text = String::from_utf8(bytes).map_err(|error| {
            AppError::fm_tpl(
                24,
                format!(
                    "{}: `shape:` entry `{shape}` at `{}` is not valid UTF-8 Turtle: {error}",
                    tpl_path.display(),
                    shape_path.display()
                ),
            )
        })?;
        combined_shapes_ttl.push_str(&text);
        combined_shapes_ttl.push('\n');
    }
    let outcome = active_graph
        .validate_shacl(&combined_shapes_ttl)
        .map_err(|error| {
            AppError::fm_tpl(
                26,
                format!(
                    "{}: `shape:` validation could not run: {error}. Remediation: use the \
                 default GraphLaw engine (no `--engine oxigraph`), which is the only \
                 engine with SHACL support.",
                    tpl_path.display()
                ),
            )
        })?;
    if !outcome.conforms {
        return Err(AppError::fm_tpl(
            25,
            format!(
                "{}: `shape:` validation failed, {} violation(s): {}",
                tpl_path.display(),
                outcome.violations.len(),
                outcome.violations.join("; ")
            ),
        ));
    }
    Ok(())
}

/// When `frontmatter.determinism == Some(true)`, independently re-run
/// extraction and reproduce the same whole, row-fanout, or static aggregate
/// projection. Query order, output bytes, paths, and lifecycle law are all
/// part of the assertion.
#[allow(clippy::too_many_arguments)]
fn check_determinism(
    tera: &mut tera::Tera, body_template: &str, tpl_path: &Path,
    frontmatter: &crate::template::Frontmatter, determinism_recheck: Option<&ExtractedRows>,
    mode: ProjectionMode, first_render: &str, first_frontmatter: &Frontmatter,
) -> Result<()> {
    let Some(recheck) = determinism_recheck else {
        return Ok(());
    };
    let Some((named2, results2)) = recheck else {
        return Err(AppError::fm_tpl(
            9,
            format!(
                "{}: `determinism: true` violated — re-running the `when:` guard a second, \
                 independent time evaluated false after the first run evaluated true. \
                 Remediation: remove non-deterministic terms from `when:`.",
                tpl_path.display()
            ),
        ));
    };

    let (second_render, second_frontmatter) = match mode {
        ProjectionMode::Whole => {
            let ctx = base_context(named2, results2);
            let second_to = render_str(tera, &frontmatter.to, &ctx, tpl_path)?;
            let second_frontmatter =
                render_output_frontmatter(tera, frontmatter, &ctx, tpl_path, &second_to)?;
            let second_render = render_str(tera, body_template, &ctx, tpl_path)?;
            (second_render, second_frontmatter)
        }
        ProjectionMode::Row(idx) => {
            let row2 = results2.get(idx).ok_or_else(|| {
                AppError::fm_tpl(
                    9,
                    format!(
                        "{}: `determinism: true` violated — the second independent query \
                         execution produced {} row(s), which does not cover row index {idx}. \
                         Remediation: add a complete ORDER BY with a stable tie-break.",
                        tpl_path.display(),
                        results2.len()
                    ),
                )
            })?;
            let ctx = row_context(named2, results2, row2);
            let second_to = render_str(tera, &frontmatter.to, &ctx, tpl_path)?;
            let second_frontmatter =
                render_output_frontmatter(tera, frontmatter, &ctx, tpl_path, &second_to)?;
            let second_render = render_str(tera, body_template, &ctx, tpl_path)?;
            (second_render, second_frontmatter)
        }
        ProjectionMode::Aggregate => render_aggregate_projection(
            tera,
            body_template,
            tpl_path,
            frontmatter,
            named2,
            results2,
        )?
        .ok_or_else(|| {
            AppError::fm_tpl(
                9,
                format!(
                    "{}: `determinism: true` violated — the second independent `for_each` \
                     execution produced zero rows after the first produced an aggregate output.",
                    tpl_path.display()
                ),
            )
        })?,
    };

    if &second_frontmatter != first_frontmatter {
        return Err(AppError::fm_tpl(
            9,
            format!(
                "{}: `determinism: true` violated — the second independent execution \
                 produced different path, cardinality, composition, hook, shape, or \
                 ownership semantics.",
                tpl_path.display()
            ),
        ));
    }
    if second_render != first_render {
        return Err(AppError::fm_tpl(
            9,
            format!(
                "{}: `determinism: true` violated — the second independent execution \
                 produced different output bytes.",
                tpl_path.display()
            ),
        ));
    }
    Ok(())
}

/// `determinism: true`: verify the independently-reloaded second graph
/// agrees that the primary's `when:` guard evaluated false. Called from the
/// primary extraction's own zero-output skip branch (`sync`'s per-template
/// loop) — the mirror image of `check_determinism`'s own guard check, which
/// only runs when the PRIMARY guard evaluated true. If the second load's
/// guard evaluates true instead (`determinism_recheck` is `Some(Some(_))`),
/// that asymmetry is itself the load-time nondeterminism `determinism:
/// true` exists to catch (e.g. a `when:` ASK that binds a blank node whose
/// id oxigraph's `rename_blank_nodes()` reassigns per parse), and must
/// surface as an error rather than a silent "skipped: when guard false".
fn check_determinism_guard_agrees_false(
    tpl_path: &Path, determinism_recheck: Option<&ExtractedRows>,
) -> Result<()> {
    let Some(recheck) = determinism_recheck else {
        return Ok(());
    };
    if recheck.is_some() {
        return Err(AppError::fm_tpl(
            9,
            format!(
                "{}: `determinism: true` violated — the independent second load's \
                 `when:` guard evaluated true while the first evaluated false. \
                 Remediation: remove non-deterministic terms from `when:`.",
                tpl_path.display()
            ),
        ));
    }
    Ok(())
}

/// `determinism: true`: verify the independently-reloaded second graph
/// agrees that the primary's driving row set (a named `for_each:` or the
/// implicit per-row fan-out) was genuinely empty. Called from `sync`'s
/// per-template loop's zero-row skip branches — the mirror image of
/// `check_determinism`'s `ProjectionMode::Aggregate`/`Row` checks, which
/// only run once the primary has already produced at least one row. A
/// zero-row primary result is a lawful answer, not a failure, but it must
/// not silently bypass verification: the (expensive) independent recheck
/// was already computed for exactly this purpose, and a second load that
/// disagrees — either its `when:` guard evaluates true where the first was
/// consumed as false, or it binds `>= 1` row where the first bound none —
/// is exactly the class of divergence `determinism: true` exists to catch.
fn check_determinism_rows_agree_empty(
    tpl_path: &Path, determinism_recheck: Option<&ExtractedRows>, driver_desc: &str,
) -> Result<()> {
    let Some(recheck) = determinism_recheck else {
        return Ok(());
    };
    let Some((_, results2)) = recheck else {
        return Err(AppError::fm_tpl(
            9,
            format!(
                "{}: `determinism: true` violated — re-running the `when:` guard a second, \
                 independent time evaluated false after the first run evaluated true. \
                 Remediation: remove non-deterministic terms from `when:`.",
                tpl_path.display()
            ),
        ));
    };
    if !results2.is_empty() {
        return Err(AppError::fm_tpl(
            9,
            format!(
                "{}: `determinism: true` violated — the independent second load's \
                 {driver_desc} produced {} row(s) after the first produced zero. \
                 Remediation: add a complete ORDER BY with a stable tie-break, or \
                 remove non-deterministic terms from the query.",
                tpl_path.display(),
                results2.len()
            ),
        ));
    }
    Ok(())
}

/// Recursively collect `*.tmpl` files. A missing templates dir fails closed.
fn collect_tmpl_paths(dir: &Path, out: &mut Vec<PathBuf>) -> Result<()> {
    let entries = std::fs::read_dir(dir).map_err(|e| {
        AppError::fm_config(
            4,
            format!(
                "templates dir `{}` unreadable: {e}. Remediation: fix [templates].dir.",
                dir.display()
            ),
        )
    })?;
    for entry in entries {
        let entry = entry?;
        let path = entry.path();
        if path.is_dir() {
            collect_tmpl_paths(&path, out)?;
        } else if path.extension().is_some_and(|e| e == "tmpl") {
            out.push(path);
        }
    }
    Ok(())
}

/// Build a genuinely independent second copy of the fully-prepared base
/// graph, used ONLY by `determinism: true`'s recheck
/// (`sync`'s per-template loop, `determinism_recheck`).
///
/// A brand-new [`GraphEngine`] instance (`new_graph_engine`, never
/// `graph`/`active_graph`), with the project ontology and every pack
/// ontology re-read from disk and re-parsed from scratch — a real second
/// [`GraphEngine::insert_turtle_documents`] admission, not a re-query of
/// already-parsed quads — then every non-overlay template's `construct:`
/// and any `[law].rules`/pack `hook.ttl` materialization replayed against
/// it, so its derived facts match what the primary `graph` has by the time
/// templates are extracted.
///
/// This closes the gap where the recheck reused the SAME already-loaded
/// `active_graph` for both the primary extraction and the recheck: reusing
/// one store can prove query-execution nondeterminism (differing
/// bindings/order across two calls against one unchanged store) but is
/// structurally incapable of catching load-time nondeterminism — e.g.
/// oxigraph's `RdfParser::rename_blank_nodes()` (see
/// `graph::ontology_batch::insert_documents`) assigns every blank node a
/// fresh `rand::random()` id per parse, so two independent loads of
/// byte-identical Turtle containing a blank node produce genuinely
/// different bindings. See `tests/determinism_independent_reload_e2e.rs`
/// for a real, non-mocked demonstration.
///
/// Read-only law operations (`check_denials`, `[law].gates`,
/// pack-shipped gates) are deliberately NOT replayed here: they never
/// mutate graph state, so they cannot change what a `sparql:`/`when:`
/// query observes, and this sync already ran them once (Stage 2b) before
/// reaching the per-template loop that calls this function.
///
/// # Errors
/// Propagates the same `[FM-CONFIG-*]`/`[FM-PACK-*]`/`[FM-GRAPH-*]`/
/// `[FM-LAW-*]` errors the primary Stage 1/2/2b construction already
/// succeeded at once this sync — a failure here means the filesystem
/// changed mid-sync (TOCTOU), not a pre-existing project defect.
fn build_independent_recheck_base_graph(
    root: &Path, config: &GgenConfig, opts: SyncOptions, packs: &[crate::pack::Pack],
    templates: &[(PathBuf, Template)],
) -> Result<Arc<dyn GraphEngine>> {
    let fresh: Arc<dyn GraphEngine> = new_graph_engine(opts.engine)?;

    // Independent Stage-1 equivalent: re-read every ontology document from
    // disk into fresh owned strings and re-parse them through a brand-new
    // `insert_turtle_documents` admission (a second, independent call to
    // the exact code path that assigns blank-node ids on the primary
    // graph, at line ~300 above) — never touching `graph`'s already-parsed
    // quads.
    let ontology_path = root.join(&config.ontology.source);
    let (ontology_label, ttl) = read_ontology_file(root, &ontology_path)?;
    let mut ontology_sources = Vec::with_capacity(
        1 + packs
            .iter()
            .map(|pack| 1 + pack.extra_ontology_paths.len())
            .sum::<usize>(),
    );
    ontology_sources.push((ontology_label, ttl));
    for import in &config.ontology.imports {
        ontology_sources.push(read_ontology_file(root, &root.join(import))?);
    }
    for pack in packs {
        let pack_ttl = std::fs::read_to_string(&pack.ontology_path).map_err(|e| {
            AppError::fm_pack(
                4,
                format!(
                    "pack `{}`: ontology `{}` unreadable: {e}",
                    pack.name,
                    pack.ontology_path.display()
                ),
            )
        })?;
        ontology_sources.push((rel_display(root, &pack.ontology_path), pack_ttl));
        for (declared, extra_path) in &pack.extra_ontology_paths {
            let extra_ttl = std::fs::read_to_string(extra_path).map_err(|e| {
                AppError::fm_pack(
                    4,
                    format!(
                        "pack `{}`: extra ontology `{declared}` unreadable at `{}`: {e}",
                        pack.name,
                        extra_path.display()
                    ),
                )
            })?;
            ontology_sources.push((declared.clone(), extra_ttl));
        }
    }
    let ontology_documents: Vec<TurtleDocument<'_>> = ontology_sources
        .iter()
        .map(|(label, content)| TurtleDocument::new(label, content))
        .collect();
    let recheck_ontology_receipt = fresh.insert_turtle_documents(&ontology_documents)?;
    tracing::debug!(
        ontology.documents = recheck_ontology_receipt.documents,
        ontology.parsed_quads = recheck_ontology_receipt.parsed_quads,
        ontology.inserted_quads = recheck_ontology_receipt.inserted_quads,
        "determinism-recheck ontology batch admitted (independent second load)"
    );

    // Independent Stage-2 equivalent: replay every non-overlay template's
    // `construct:` (overlay templates build their own independent overlay
    // separately — see the `build_rdf_overlay` call at this function's one
    // call site).
    for (_, tpl) in templates {
        if declares_rdf_overlay(&tpl.frontmatter) {
            continue;
        }
        if let Some(construct) = tpl.frontmatter.construct.as_deref() {
            insert_construct(fresh.as_ref(), construct)?;
        }
    }

    // Independent Stage-2b equivalent: `[law].rules` + pack `hook.ttl` +
    // materialize, so any queried derived facts match the primary graph's.
    for rel in &config.law.rules {
        let rule_path = root.join(rel);
        let src = std::fs::read_to_string(&rule_path).map_err(|e| {
            AppError::fm_law(
                10,
                format!(
                    "rule file `{}` unreadable: {e}. Remediation: fix [law].rules.",
                    rule_path.display()
                ),
            )
        })?;
        fresh.load_rules(&src)?;
    }
    let mut any_hook_loaded = false;
    for pack in packs {
        let hook_path = pack.root.join("hook.ttl");
        if !hook_path.is_file() {
            continue;
        }
        let hook_ttl = std::fs::read_to_string(&hook_path).map_err(|e| {
            AppError::fm_pack(
                14,
                format!(
                    "pack `{}`: hook.ttl at `{}` exists but is unreadable: {e}. \
                     Remediation: fix the file's permissions/encoding or remove it.",
                    pack.name,
                    hook_path.display()
                ),
            )
        })?;
        fresh.load_hook_pack(&hook_ttl)?;
        any_hook_loaded = true;
    }
    if !config.law.rules.is_empty() || any_hook_loaded {
        fresh.materialize()?;
    }

    Ok(fresh)
}

/// `true` when a template's frontmatter declares `rdf:` and/or
/// `rdf_inline:` — such a template gets its own per-template overlay graph
/// (see [`build_rdf_overlay`]) instead of using the shared graph directly.
fn declares_rdf_overlay(fm: &Frontmatter) -> bool {
    !fm.rdf.is_empty() || !fm.rdf_inline.is_empty()
}

/// Turtle prolog (`@base`/`@prefix` directives, literal Turtle syntax —
/// *not* the SPARQL-style `BASE`/`PREFIX` prolog ggen-core's `build_prolog`
/// emits for SPARQL query text) prepended to `rdf:`/`rdf_inline:` content
/// before parsing, so relative/prefixed IRIs in that content resolve as the
/// frontmatter's `base:`/`prefixes:` declare. `@base` first (when present),
/// then `@prefix` lines in deterministic (`BTreeMap`) key order.
fn build_turtle_prolog(prefixes: &BTreeMap<String, String>, base: Option<&str>) -> String {
    let mut out = String::new();
    if let Some(b) = base {
        out.push_str("@base <");
        out.push_str(b);
        out.push_str("> .\n");
    }
    for (name, iri) in prefixes {
        out.push_str("@prefix ");
        out.push_str(name);
        out.push_str(": <");
        out.push_str(iri);
        out.push_str("> .\n");
    }
    out
}

/// Build a per-template overlay graph for a template declaring
/// `rdf:`/`rdf_inline:`: a brand-new store containing the base graph's
/// current triples (re-serialized as N-Triples — a syntactic subset of
/// Turtle, same trick [`insert_construct`] relies on) plus this template's
/// own RDF/Turtle content. `base` is never mutated — this is a fresh store,
/// so two templates in the same run with different `rdf:`/`rdf_inline:`
/// never see each other's extra triples, and templates that declare
/// neither field are entirely unaffected.
///
/// Returns `None` when the template declares neither field, meaning callers
/// should keep using `base` (and the shared `tera`) directly.
///
/// # Errors
/// - `[FM-TPL-010]` an `rdf:` path escapes the template's own directory —
///   the identical traversal-safety check `from:` uses
///   ([`crate::write::resolve_target`]), applied relative to the template
///   file's own directory (never the project root).
/// - `[FM-TPL-011]` an `rdf:` path is unreadable.
/// - Propagates `[FM-GRAPH-*]` on any Turtle parse failure (the base
///   graph's re-serialized triples, or `rdf:`/`rdf_inline:` content).
fn build_rdf_overlay(
    base: &Arc<dyn GraphEngine>, tpl_path: &Path, fm: &Frontmatter,
) -> Result<Option<Arc<dyn GraphEngine>>> {
    use std::fmt::Write as _;

    if !declares_rdf_overlay(fm) {
        return Ok(None);
    }

    let overlay = DeterministicGraph::new()?;
    let base_ntriples: String =
        base.canonical_quads()?
            .into_iter()
            .fold(String::new(), |mut doc, line| {
                let _ = writeln!(doc, "{line} .");
                doc
            });
    if !base_ntriples.is_empty() {
        overlay.insert_turtle(&base_ntriples)?;
    }

    let prolog = build_turtle_prolog(&fm.prefixes, fm.base.as_deref());
    let template_dir = tpl_path.parent().unwrap_or_else(|| Path::new("."));

    for rdf_file in &fm.rdf {
        let rdf_path = crate::write::resolve_target(template_dir, rdf_file).map_err(|e| {
            AppError::fm_tpl(
                18,
                format!(
                    "{}: `rdf: {rdf_file}` is not a safe path (must stay inside the \
                     template's own directory): {e}",
                    tpl_path.display()
                ),
            )
        })?;
        let content = std::fs::read_to_string(&rdf_path).map_err(|e| {
            AppError::fm_tpl(
                11,
                format!(
                    "{}: `rdf: {rdf_file}` unreadable at `{}`: {e}. \
                     Remediation: fix the path (relative to the template's own directory).",
                    tpl_path.display(),
                    rdf_path.display()
                ),
            )
        })?;
        overlay.insert_turtle(&format!("{prolog}{content}"))?;
    }

    for inline in &fm.rdf_inline {
        overlay.insert_turtle(&format!("{prolog}{inline}"))?;
    }

    Ok(Some(Arc::new(overlay)))
}

/// Run a CONSTRUCT query and insert its triples back into the graph.
fn insert_construct(graph: &dyn GraphEngine, construct: &str) -> Result<()> {
    use std::fmt::Write as _;
    let EngineQueryResults::Graph(triples) = graph.query(construct)? else {
        return Err(AppError::fm_graph(
            7,
            "`construct:` frontmatter must be a CONSTRUCT/DESCRIBE query. \
             Remediation: use CONSTRUCT { … } WHERE { … }.",
        ));
    };
    let mut doc = String::new();
    for triple in triples {
        let _ = writeln!(doc, "{} .", triple.ntriples);
    }
    if !doc.is_empty() {
        // N-Triples is a syntactic subset of Turtle.
        graph.insert_turtle(&doc)?;
    }
    Ok(())
}

/// Read the previous chain head, if any. The receipt log's tail is the
/// source of truth (it is appended before the head pointer is rewritten, so
/// a partially-failed sync leaves a re-runnable state); `receipt.json` is a
/// fallback for pre-log projects.
fn read_prev_head(receipt_path: &Path, log_path: &Path) -> Result<Option<SyncReceipt>> {
    match std::fs::read_to_string(log_path) {
        Ok(raw) => {
            if let Some(line) = raw.lines().rev().find(|l| !l.trim().is_empty()) {
                let prev: SyncReceipt = serde_json::from_str(line).map_err(|e| {
                    AppError::fm_chain(
                        3,
                        format!(
                            "receipt log `{}` tail malformed: {e}. \
                             Remediation: run `ggen receipt history` and repair the log.",
                            log_path.display()
                        ),
                    )
                })?;
                return Ok(Some(prev));
            }
        }
        Err(e) if e.kind() == std::io::ErrorKind::NotFound => {}
        Err(e) => {
            return Err(AppError::fm_chain(
                3,
                format!("receipt log `{}` unreadable: {e}", log_path.display()),
            ))
        }
    }
    match std::fs::read_to_string(receipt_path) {
        Ok(raw) => {
            let prev: SyncReceipt = serde_json::from_str(&raw).map_err(|e| {
                AppError::fm_chain(
                    3,
                    format!(
                        "previous receipt `{}` malformed: {e}. \
                         Remediation: verify or remove the stale receipt.",
                        receipt_path.display()
                    ),
                )
            })?;
            Ok(Some(prev))
        }
        Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(None),
        Err(e) => Err(AppError::fm_chain(
            3,
            format!(
                "previous receipt `{}` unreadable: {e}",
                receipt_path.display()
            ),
        )),
    }
}

/// Chain-rule downgrade guard for a head that verifies only as
/// [`ChainStanding::LegacyV2Unbound`] (FM-CHAIN-009): a legacy base-rule
/// record is lawful only in the pre-F1 prefix of the ledger, so if any
/// record in the receipt log declares a chain rule, a legacy head is a
/// downgrade (e.g. a fold-sealed head re-sealed under the base rule with
/// its declaration stripped). Same law as
/// `ChainRuleMonotonicity` (graphlaw-backed via `crate::receipt_chain_seam`), applied at the
/// head-only call sites (`ggen sync`'s write path and `ggen receipt
/// verify`) so they fail closed exactly where `ggen receipt history` does.
///
/// Returns the index of the first log record that declares a chain rule,
/// or `None` when the log is absent or carries no declared record. `code`
/// is the FM-CHAIN code the caller reports read failures under.
///
/// # Errors
/// The log exists but is unreadable, or one of its lines is malformed
/// (fail closed: an unparseable log cannot prove the head is lawful).
pub(crate) fn first_declared_chain_rule_in_log(
    log_path: &Path, code: u16,
) -> Result<Option<usize>> {
    let raw = match std::fs::read_to_string(log_path) {
        Ok(raw) => raw,
        Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(None),
        Err(e) => {
            return Err(AppError::fm_chain(
                code,
                format!("receipt log `{}` unreadable: {e}", log_path.display()),
            ))
        }
    };
    for (idx, line) in raw.lines().filter(|l| !l.trim().is_empty()).enumerate() {
        let receipt: SyncReceipt = serde_json::from_str(line).map_err(|e| {
            AppError::fm_chain(
                code,
                format!(
                    "receipt log `{}` entry {idx} malformed: {e}. \
                     Remediation: run `ggen receipt history` and repair the log.",
                    log_path.display()
                ),
            )
        })?;
        if receipt.record.chain_rule.is_some() {
            return Ok(Some(idx));
        }
    }
    Ok(None)
}

/// Chain a praxis-core [`ReceiptRecord`] over `{ graph_hash, outputs }` and
/// write it to [`RECEIPT_REL_PATH`]. `ts_ns` is fixed to 0 (no wall clock;
/// see module docs).
///
/// Both files are written compact (not pretty-printed) so the payload's
/// on-disk bytes are exactly the bytes `payload_hash_hex` was computed
/// over — the verifier hashes the raw stored payload rather than
/// re-serializing it, which keeps old receipts verifiable across payload
/// schema additions.
///
/// `pub(crate)`: also called by `crate::generation_rules::run` so the
/// declarative `[[generation.rules]]` sync path chains onto the exact same
/// receipt history/log as the frontmatter path (one receipt chain per
/// project, regardless of which schema `ggen.toml` uses).
///
/// This is one of two deliberately coexisting receipt mechanisms: the
/// chained BLAKE3 ledger written here is a continuous chain-of-custody
/// record across syncs. `ggen_config::receipt::Receipt` is the other —
/// Ed25519-signed over SHA-256, emitted once per pack install to
/// `.ggen/receipts/pack-*.json`. Different lifecycles, different hash
/// algorithms; neither replaces the other.
/// Which equivalence class (of [`EquivalenceMap`]'s 8) a closure key belongs
/// to, for the two classes `ggen sync` can honestly evaluate from its own
/// input-closure hashes. `None` means the key participates in neither
/// evaluated class (e.g. `receipt-log:MALFORMED-*` markers, which are
/// evidence-adjacent history, not a governing source/config input).
///
/// - `config`: the inputs that select/parameterize the pipeline rather than
///   feed content into it — `ggen.toml` (the manifest), `project` (the bound
///   project identity), and `actuator` (the ggen version that ran).
/// - `source`: every other closure entry — ontologies (project + pack +
///   extra), templates, and template-declared `from`/`rdf` inputs.
fn closure_equivalence_class(key: &str) -> Option<&'static str> {
    if key.starts_with("receipt-log:") {
        return None;
    }
    match key {
        "ggen.toml" | "project" | "actuator" => Some("config"),
        _ => Some("source"),
    }
}

/// Byte-hash equality over one equivalence class's slice of the input
/// closure: this run's `{key -> blake3}` submap vs the previous receipt's.
/// Divergence records WHICH keys differ (added/removed/changed), bounded to
/// the first few so the reason stays a reason, not a dump.
fn compare_closure_class(
    class: &'static str, current: &BTreeMap<String, String>, previous: &BTreeMap<String, String>,
) -> EquivalenceStatus {
    let cur: BTreeMap<&String, &String> = current
        .iter()
        .filter(|(k, _)| closure_equivalence_class(k) == Some(class))
        .collect();
    let prev: BTreeMap<&String, &String> = previous
        .iter()
        .filter(|(k, _)| closure_equivalence_class(k) == Some(class))
        .collect();
    if cur == prev {
        return EquivalenceStatus::Equivalent;
    }
    let mut diffs: Vec<String> = Vec::new();
    for (k, v) in &cur {
        match prev.get(*k) {
            None => diffs.push(format!("added `{k}`")),
            Some(pv) if pv != v => diffs.push(format!("changed `{k}`")),
            Some(_) => {}
        }
    }
    for k in prev.keys() {
        if !cur.contains_key(*k) {
            diffs.push(format!("removed `{k}`"));
        }
    }
    let shown = diffs.iter().take(4).cloned().collect::<Vec<_>>().join(", ");
    let suffix = if diffs.len() > 4 {
        format!(" (+{} more)", diffs.len() - 4)
    } else {
        String::new()
    };
    EquivalenceStatus::Divergent(format!(
        "closure {class} hashes differ from previous receipt: {shown}{suffix}"
    ))
}

/// Which [`ComponentLevels`] axis an admitted external `ver:Check` fact's
/// stable name (`ver:name`, e.g. `test-workspace`, `clippy-reference-gate`,
/// `fmt-check`, `build`) feeds. Substring-based on purpose: the verify-pack
/// vocabulary deliberately leaves `ver:name` open (consumers add their own
/// required checks), so an unrecognized name lands on the `gate` aggregate
/// axis -- exactly the "everything else passed/failed" bucket
/// `ComponentLevels::gate`'s own doc comment reserves for it -- rather than
/// being dropped.
fn verify_check_axis(name: &str) -> &'static str {
    if name.contains("test") {
        "test"
    } else if name.contains("clippy") || name.contains("lint") {
        "lint"
    } else if name.contains("fmt") {
        "fmt"
    } else {
        "gate"
    }
}

/// The `{check name -> observed outcome}` submap of one receipt's admission
/// ledger restricted to external `ver:` evidence items on either the `test`
/// axis (`test_axis == true`, feeding the `tests` equivalence class) or the
/// three non-test axes (`test_axis == false`, feeding `gates`).
fn verify_outcome_map(
    items: &[AdmissionItem], test_axis: bool,
) -> BTreeMap<String, ObservedOutcome> {
    items
        .iter()
        .filter_map(|item| {
            let name = item.evidence_id.strip_prefix("ver:")?;
            if (verify_check_axis(name) == "test") == test_axis {
                Some((name.to_string(), item.observed_outcome.clone()))
            } else {
                None
            }
        })
        .collect()
}

/// Evaluate the `tests`/`gates` equivalence class from admitted external
/// `ver:Check` evidence: this run's per-check outcomes vs the previous
/// receipt's recorded `ver:` admission items. `Unknown` stays the honest
/// answer whenever EITHER side has no evidence for the class -- absence of
/// evidence is never promoted to `Equivalent`.
fn compare_verify_class(
    class: &'static str, test_axis: bool, current_items: &[AdmissionItem],
    prev_epoch: &ReceiptEpochV2,
) -> EquivalenceStatus {
    let prev_items: &[AdmissionItem] = match &prev_epoch.admission {
        AdmissionLedger::Recorded(items) => items,
        AdmissionLedger::LegacyUnrecorded => &[],
    };
    let cur = verify_outcome_map(current_items, test_axis);
    let prev = verify_outcome_map(prev_items, test_axis);
    if cur.is_empty() || prev.is_empty() {
        return EquivalenceStatus::Unknown;
    }
    if cur == prev {
        return EquivalenceStatus::Equivalent;
    }
    let mut diffs: Vec<String> = Vec::new();
    for (name, outcome) in &cur {
        match prev.get(name) {
            None => diffs.push(format!("added `{name}`")),
            Some(prev_outcome) if prev_outcome != outcome => {
                diffs.push(format!("changed `{name}`"));
            }
            Some(_) => {}
        }
    }
    for name in prev.keys() {
        if !cur.contains_key(name) {
            diffs.push(format!("removed `{name}`"));
        }
    }
    let shown = diffs.iter().take(4).cloned().collect::<Vec<_>>().join(", ");
    let suffix = if diffs.len() > 4 {
        format!(" (+{} more)", diffs.len() - 4)
    } else {
        String::new()
    };
    EquivalenceStatus::Divergent(format!(
        "external ver:Check {class} outcomes differ from previous receipt: {shown}{suffix}"
    ))
}

/// One class-producer fact read from the admitted union graph: the recorded
/// exit code of the externally-run producer plus the identity string the
/// class's equivalence comparison keys on (docs manifest digest, receipt
/// chain head, required-set digest, binary sha256).
struct ClassProducerFact {
    exit: i64,
    identity: String,
}

/// The identity string a previous receipt recorded for one producer-fed
/// equivalence class, recovered from its admission ledger's
/// `class:<class>:<identity>` evidence id. `None` when the previous
/// generation had no producer fact for the class (Unknown floor, never
/// rounded up).
fn prev_class_identity(prev_epoch: &ReceiptEpochV2, class: &str) -> Option<String> {
    let AdmissionLedger::Recorded(items) = &prev_epoch.admission else {
        return None;
    };
    let prefix = format!("class:{class}:");
    items
        .iter()
        .find_map(|i| i.evidence_id.strip_prefix(&prefix).map(str::to_string))
}

/// Generic producer-fed class evaluation (docs / evidence / compiled-binary
/// share it; receipts has its own chain-linkage rule at the call site):
/// no current fact -> `Unknown`; red current fact -> `Divergent` (never
/// rounded up); green fact with no previous identity (or a previous
/// explicit `skipped` marker) -> `Unknown`; identical identities ->
/// `Equivalent`; differing -> `Divergent` naming both.
fn compare_producer_class(
    class: &'static str, current: Option<&ClassProducerFact>, prev_epoch: &ReceiptEpochV2,
) -> EquivalenceStatus {
    let Some(fact) = current else {
        return EquivalenceStatus::Unknown;
    };
    if fact.exit != 0 {
        return EquivalenceStatus::Divergent(format!(
            "{class} producer recorded red evidence (exit {})",
            fact.exit
        ));
    }
    match prev_class_identity(prev_epoch, class) {
        None => EquivalenceStatus::Unknown,
        Some(prev) if prev == "skipped" => EquivalenceStatus::Unknown,
        Some(prev) if prev == fact.identity => EquivalenceStatus::Equivalent,
        Some(prev) => EquivalenceStatus::Divergent(format!(
            "{class} identity changed from previous receipt: {prev} -> {}",
            fact.identity
        )),
    }
}

///
/// `graph`: the already-loaded project+pack union graph for this run,
/// queried here for `ccn:Law` individuals (see the admission-items block
/// below). Passing the live handle rather than re-loading anything keeps
/// this a read against the exact same state the rest of the pipeline saw.
// One sequential receipt-assembly function (admission ledger, equivalence
// classification, chain hashing, signing, then the two on-disk writes);
// splitting it would scatter the single linear assembly across call
// boundaries rather than shrink real complexity.
#[allow(clippy::too_many_lines)]
pub(crate) fn write_receipt(
    root: &Path, report: &SyncReport, graph: &dyn GraphEngine, receipt_origin: Option<&'static str>,
) -> Result<()> {
    use std::io::Write as _;

    // Bind every decision target that exists on disk (written this run or
    // skipped-unchanged), so a no-op re-sync produces the identical payload.
    // A target that is absent is fine (e.g. a false `when` guard whose file
    // was never generated); a target that exists but cannot be read must
    // fail closed — silently omitting it would unbind the file from the
    // receipt and from every later doctor/staleness check.
    let mut outputs = BTreeMap::new();
    for rel in report.decisions.keys() {
        let target = root.join(rel);
        match std::fs::read(&target) {
            Ok(bytes) => {
                outputs.insert(rel.clone(), blake3::hash(&bytes).to_hex().to_string());
            }
            Err(e) if e.kind() == std::io::ErrorKind::NotFound => {}
            Err(e) => {
                return Err(AppError::fm_chain(
                    8,
                    format!(
                        "output `{}` cannot be read for receipt binding: {e}. \
                         Remediation: fix the file's permissions.",
                        target.display()
                    ),
                ))
            }
        }
    }
    let payload = ReceiptPayload {
        graph_hash: report.graph_hash_hex.clone(),
        outputs,
        packs: report.packs.clone(),
        decisions: report.decisions.clone(),
        closure: report.closure.clone(),
        build_provenance: Some(crate::build_provenance::BUILD_PROVENANCE_HEX.to_string()),
    };
    let payload_bytes = serde_json::to_vec(&payload)?;
    let payload_hash_hex = blake3::hash(&payload_bytes).to_hex().to_string();

    // Chain onto the previous sync's receipt when one exists; a genesis
    // receipt chains from all-zeros. Never extend an unverified head: a
    // tampered previous receipt must be refused here, not discovered later
    // by `receipt history`.
    let receipt_path = root.join(RECEIPT_REL_PATH);
    let log_path = root.join(RECEIPT_LOG_REL_PATH);
    let prev_head = read_prev_head(&receipt_path, &log_path)?;
    // Rule-aware verification (FM-CHAIN-009): the previous head is checked
    // under the chain rule that produced it (`ReceiptRecord::verify_chain`).
    // A pre-F1 head that verifies only under the base rule is extended, but
    // its standing is capped: its `v2` payload was never bound into its
    // chain hash, so it is read as the legacy-bounded epoch below, never as
    // the stored (unprovable) payload.
    let prev_chain_standing = match &prev_head {
        Some(prev) => match prev.record.verify_chain().map_err(|e| {
            AppError::fm_chain(9, format!("previous receipt chain recompute failed: {e}"))
        })? {
            ChainVerification::Verified(ChainStanding::LegacyV2Unbound) => {
                // Never extend a downgraded head: a legacy head after a
                // declared record is exactly what `receipt history` refuses.
                if let Some(first) = first_declared_chain_rule_in_log(&log_path, 9)? {
                    return Err(AppError::fm_chain(
                        9,
                        format!(
                            "previous receipt head is invalid: chain-rule downgrade -- the head \
                             verifies only as `{}` but receipt log record {first} declared a \
                             chain rule. Refusing to extend a downgraded chain. Remediation: run \
                             `ggen receipt history` and restore the receipts.",
                            ChainStanding::LegacyV2Unbound.as_str()
                        ),
                    ));
                }
                Some(ChainStanding::LegacyV2Unbound)
            }
            ChainVerification::Verified(standing) => Some(standing),
            ChainVerification::Mismatch { rule, recomputed } => {
                return Err(AppError::fm_chain(
                    9,
                    format!(
                        "previous receipt head is invalid: stored chain hash {} does not \
                         match recompute {} (chain rule `{}`). Refusing to extend a tampered \
                         chain. Remediation: run `ggen receipt history` and restore the \
                         receipts.",
                        prev.record.chain_hash_hex,
                        hex32(&recomputed),
                        rule.as_str()
                    ),
                ));
            }
        },
        None => None,
    };
    let prev_chain_hash_hex = match &prev_head {
        Some(prev) => prev.record.chain_hash_hex.clone(),
        None => "0".repeat(64),
    };

    // ── Receipt epoch v2 ────────────────────────────────────────────────
    //
    // Every output-decision entry becomes one itemized admission item.
    // `ggen sync` today only reaches this point after every stage in the
    // pipeline has already returned `Ok` -- there is no per-file
    // quarantine/refusal concept yet -- so every item is honestly
    // `Admitted`; the precedence machinery (`derive_andon`) is real and
    // wired for the day a future stage does quarantine/refuse a specific
    // output, not a decorative pass-through. The equivalence map's `source`
    // and `config` classes are the two this call site can honestly evaluate
    // (byte-hash equality of the input closure vs the previous receipt's,
    // see `compare_closure_class`); the other six classes stay explicitly
    // `Unknown` because this call site does not evaluate them, and claiming
    // otherwise would overclaim checks that were never run (see
    // `.claude/rules/no-overclaiming-rust.md`).
    // A `freeze_policy: always` output whose write.rs skip reason names
    // real drift (candidate content would differ from the frozen file on
    // disk) is quarantined rather than silently Admitted-Green: the file is
    // never overwritten (freeze semantics unchanged), but the fact that
    // generation now wants to produce different content than what's frozen
    // is real, load-bearing information this generation observed and must
    // not drop. `Fail`+`Quarantined` (not `Refused`) mirrors the
    // `ccn:mechanized=false` precedent immediately below: an honestly-
    // recorded open obligation, not a fresh hard refusal.
    let mut admission_items: Vec<AdmissionItem> = report
        .decisions
        .iter()
        .map(|(path, reason)| {
            if reason.contains("DRIFT:") {
                AdmissionItem {
                    evidence_id: path.clone(),
                    observed_outcome: ObservedOutcome::Fail,
                    decision: AdmissionDecision::Quarantined,
                    reason: reason.clone(),
                    obligations_discharged: Vec::new(),
                    obligations_created: vec![format!("FROZEN-DRIFT:{path}")],
                }
            } else {
                AdmissionItem {
                    evidence_id: path.clone(),
                    observed_outcome: ObservedOutcome::Pass,
                    decision: AdmissionDecision::Admitted,
                    reason: reason.clone(),
                    obligations_discharged: Vec::new(),
                    obligations_created: Vec::new(),
                }
            }
        })
        .collect();

    // L5 condition 13 (義務数が憲法判定後の未解決義務から計算される): one
    // AdmissionItem per `ccn:Law` individual found in the already-loaded
    // union graph -- `ccn:mechanized true` discharges an obligation named
    // after the law, `ccn:mechanized false` creates one (an open
    // obligation this generation did not resolve). Both branches use
    // `AdmissionDecision::Admitted` deliberately: `ccn:mechanized` is a
    // STATIC, HUMAN-AUTHORED assertion in the ontology, not something this
    // pipeline independently re-verifies at sync time, so an unmechanized
    // law is an honestly-recorded open obligation, not a fresh refusal this
    // sync detected -- it must not flip the derived Andon level (that would
    // overclaim a live verification that doesn't exist; see
    // `packs/ggen-constitution-pack/ontology.ttl`'s own doc comment on
    // `ccn:mechanized`). A project with no `ccn:Law` individuals loaded (no
    // constitution pack) simply gets zero rows back -- this SELECT is
    // always safe to run and does not require any `[law]`/`[packs]`
    // configuration.
    let constitution_law_query: &str = "PREFIX ccn: <http://seanchatmangpt.github.io/packs/ggen-constitution#>\nSELECT ?law ?mech WHERE { ?law a ccn:Law ; ccn:mechanized ?mech . } ORDER BY ?law";
    if let EngineQueryResults::Solutions(rows) = graph.query(constitution_law_query)? {
        for row in rows {
            let Some(EngineValue::String(law_iri)) = row.get("law") else {
                continue;
            };
            let law_id = law_iri
                .rsplit(['#', '/'])
                .next()
                .unwrap_or(law_iri.as_str())
                .to_string();
            let mechanized = matches!(row.get("mech"), Some(EngineValue::Bool(true)));
            admission_items.push(if mechanized {
                AdmissionItem {
                    evidence_id: format!("constitution:{law_id}"),
                    observed_outcome: ObservedOutcome::Pass,
                    decision: AdmissionDecision::Admitted,
                    reason: "ccn:mechanized=true (static human-authored assertion in \
                             packs/ggen-constitution-pack/ontology.ttl, not independently \
                             re-verified by this pipeline)"
                        .to_string(),
                    obligations_discharged: vec![law_id],
                    obligations_created: Vec::new(),
                }
            } else {
                AdmissionItem {
                    evidence_id: format!("constitution:{law_id}"),
                    observed_outcome: ObservedOutcome::Fail,
                    decision: AdmissionDecision::Admitted,
                    reason: "ccn:mechanized=false: no enforcing mechanism found for this \
                             constitutional principle yet -- an open obligation, not a \
                             refusal this sync detected"
                        .to_string(),
                    obligations_discharged: Vec::new(),
                    obligations_created: vec![law_id],
                }
            });
        }
    }

    // ── External verification evidence (the ggen-verify-pack lane) ─────
    //
    // When the already-loaded union graph carries admitted `ver:Check`
    // facts (emitted by a consumer's verify script per
    // `packs/ggen-verify-pack/ontology.ttl` -- real exit codes of
    // externally-run commands, never hand-authored), each one becomes an
    // itemized admission entry AND live per-axis component evidence. A red
    // check whose name is an allowlisted `ver:KnownDivergence` is recorded
    // as `Fail` honestly but does not lower the axis -- the same accepted-
    // divergence semantics the pack's own `020_evidence_green.rq` gate
    // enforces. A graph with no `ver:Check` facts gets zero rows back and
    // every axis stays at the uniform floor below.
    let verify_check_query: &str = "PREFIX ver: <http://seanchatmangpt.github.io/packs/ggen-verify#>\nSELECT ?name ?exit WHERE { ?c a ver:Check ; ver:name ?name ; ver:exitCode ?exit . } ORDER BY ?name";
    let known_divergence_query: &str = "PREFIX ver: <http://seanchatmangpt.github.io/packs/ggen-verify#>\nSELECT ?name WHERE { ?kd a ver:KnownDivergence ; ver:name ?name . } ORDER BY ?name";
    let mut allowlisted_divergences: BTreeSet<String> = BTreeSet::new();
    if let EngineQueryResults::Solutions(rows) = graph.query(known_divergence_query)? {
        for row in rows {
            if let Some(EngineValue::String(name)) = row.get("name") {
                allowlisted_divergences.insert(name.clone());
            }
        }
    }
    let mut ver_checks: BTreeMap<String, i64> = BTreeMap::new();
    if let EngineQueryResults::Solutions(rows) = graph.query(verify_check_query)? {
        for row in rows {
            let Some(EngineValue::String(name)) = row.get("name") else {
                continue;
            };
            let Some(EngineValue::Int(exit)) = row.get("exit") else {
                continue;
            };
            ver_checks.insert(name.clone(), *exit);
        }
    }
    for (name, exit) in &ver_checks {
        let pass = *exit == 0;
        let allowlisted = allowlisted_divergences.contains(name);
        admission_items.push(AdmissionItem {
            evidence_id: format!("ver:{name}"),
            observed_outcome: if pass {
                ObservedOutcome::Pass
            } else {
                ObservedOutcome::Fail
            },
            decision: AdmissionDecision::Admitted,
            reason: if pass {
                "ver:Check exitCode=0 (externally-run evidence admitted via the graph)".to_string()
            } else if allowlisted {
                format!(
                    "ver:Check exitCode={exit}, allowlisted as ver:KnownDivergence -- \
                     recorded as Fail, does not lower the component axis"
                )
            } else {
                format!(
                    "ver:Check exitCode={exit} -- red external evidence; lowers the \
                     `{}` component axis by the meet law",
                    verify_check_axis(name)
                )
            },
            obligations_discharged: Vec::new(),
            obligations_created: Vec::new(),
        });
    }

    // ── Class-producer evidence (the four previously-Unknown classes) ───
    //
    // Each of `docs`/`receipts`/`evidence`/`compiled_binary` has a dedicated
    // generated producer (packs/ggen-verify-pack/templates/produce_*.tmpl)
    // that runs its class's contract externally and emits ver:-lane facts.
    // Sync only CONSUMES those admitted facts -- it never infers a class
    // verdict itself. A class with no producer fact stays at the Unknown
    // floor; a producer that ran with `--skip` emits an explicit skipped
    // marker so absence and skip are distinguished honestly.
    let docs_manifest_query: &str = "PREFIX ver: <http://seanchatmangpt.github.io/packs/ggen-verify#>\nSELECT ?path ?sha WHERE { ?e a ver:DocsManifestEntry ; ver:path ?path ; ver:sha256 ?sha . } ORDER BY ?path";
    let receipts_check_query: &str = "PREFIX ver: <http://seanchatmangpt.github.io/packs/ggen-verify#>\nSELECT ?exit ?head WHERE { ?c a ver:ReceiptsCheck ; ver:exitCode ?exit ; ver:chainHead ?head . } ORDER BY ?head";
    let evidence_check_query: &str = "PREFIX ver: <http://seanchatmangpt.github.io/packs/ggen-verify#>\nSELECT ?exit ?dig WHERE { ?c a ver:EvidenceCheck ; ver:exitCode ?exit ; ver:requiredSetDigest ?dig . } ORDER BY ?dig";
    let binary_check_query: &str = "PREFIX ver: <http://seanchatmangpt.github.io/packs/ggen-verify#>\nSELECT ?exit ?ha WHERE { ?c a ver:BinaryCheck ; ver:exitCode ?exit ; ver:hashA ?ha ; ver:hashB ?hb . } ORDER BY ?ha";
    let binary_skip_query: &str = "PREFIX ver: <http://seanchatmangpt.github.io/packs/ggen-verify#>\nSELECT ?s WHERE { ?c a ver:BinaryCheck ; ver:skipped ?s . }";

    // docs: the producer emits one ver:DocsManifestEntry per generated doc
    // (path + sha256); the class identity is a blake3 digest over the
    // path-sorted entry list, so any tampered doc changes the identity.
    let docs_fact: Option<ClassProducerFact> = {
        let mut entries: Vec<(String, String)> = Vec::new();
        if let EngineQueryResults::Solutions(rows) = graph.query(docs_manifest_query)? {
            for row in rows {
                if let (Some(EngineValue::String(path)), Some(EngineValue::String(sha))) =
                    (row.get("path"), row.get("sha"))
                {
                    entries.push((path.clone(), sha.clone()));
                }
            }
        }
        if entries.is_empty() {
            None
        } else {
            entries.sort();
            let mut hasher = blake3::Hasher::new();
            for (path, sha) in &entries {
                hasher.update(path.as_bytes());
                hasher.update(b"\n");
                hasher.update(sha.as_bytes());
                hasher.update(b"\n");
            }
            Some(ClassProducerFact {
                exit: 0,
                identity: hasher.finalize().to_hex().to_string(),
            })
        }
    };

    let read_exit_ident = |query: &str, ident_var: &str| -> Result<Option<ClassProducerFact>> {
        if let EngineQueryResults::Solutions(rows) = graph.query(query)? {
            for row in rows {
                if let (Some(EngineValue::Int(exit)), Some(EngineValue::String(ident))) =
                    (row.get("exit"), row.get(ident_var))
                {
                    return Ok(Some(ClassProducerFact {
                        exit: *exit,
                        identity: ident.clone(),
                    }));
                }
            }
        }
        Ok(None)
    };
    let receipts_fact = read_exit_ident(receipts_check_query, "head")?;
    let evidence_fact = read_exit_ident(evidence_check_query, "dig")?;
    let binary_fact = read_exit_ident(binary_check_query, "ha")?;
    let mut binary_skipped = false;
    if let EngineQueryResults::Solutions(rows) = graph.query(binary_skip_query)? {
        for row in rows {
            binary_skipped |= matches!(row.get("s"), Some(EngineValue::Bool(true)))
                || matches!(row.get("s"), Some(EngineValue::String(s)) if s == "true");
        }
    }

    // Each present producer fact becomes one recorded admission item whose
    // evidence id carries the class identity (`class:<class>:<identity>`),
    // which is exactly what the NEXT sync's equivalence comparison reads
    // back from this receipt. A red producer is recorded `Fail` and lowers
    // the `gate` component axis by the meet law below.
    let mut producer_red = false;
    let push_producer_item =
        |items: &mut Vec<AdmissionItem>, class: &str, fact: &ClassProducerFact, what: &str| {
            let pass = fact.exit == 0;
            items.push(AdmissionItem {
                evidence_id: format!("class:{class}:{}", fact.identity),
                observed_outcome: if pass {
                    ObservedOutcome::Pass
                } else {
                    ObservedOutcome::Fail
                },
                decision: AdmissionDecision::Admitted,
                reason: if pass {
                    format!("{what} producer evidence admitted (exit 0)")
                } else {
                    format!(
                        "{what} producer recorded red evidence (exit {}) -- lowers the \
                         `gate` component axis by the meet law",
                        fact.exit
                    )
                },
                obligations_discharged: Vec::new(),
                obligations_created: Vec::new(),
            });
        };
    if let Some(fact) = &docs_fact {
        push_producer_item(&mut admission_items, "docs", fact, "docs-manifest");
    }
    if let Some(fact) = &receipts_fact {
        producer_red |= fact.exit != 0;
        push_producer_item(&mut admission_items, "receipts", fact, "receipts-check");
    }
    if let Some(fact) = &evidence_fact {
        producer_red |= fact.exit != 0;
        push_producer_item(&mut admission_items, "evidence", fact, "evidence-check");
    }
    if binary_skipped {
        // Explicit Unknown marker: the producer RAN and chose --skip.
        // Distinguished from absence (no marker at all) honestly.
        admission_items.push(AdmissionItem {
            evidence_id: "class:compiled_binary:skipped".to_string(),
            observed_outcome: ObservedOutcome::Unknown,
            decision: AdmissionDecision::Admitted,
            reason: "binary-check producer ran with --skip: explicit Unknown marker \
                     (skipped, not absent); the compiled_binary class stays Unknown"
                .to_string(),
            obligations_discharged: Vec::new(),
            obligations_created: Vec::new(),
        });
    } else if let Some(fact) = &binary_fact {
        producer_red |= fact.exit != 0;
        push_producer_item(
            &mut admission_items,
            "compiled_binary",
            fact,
            "binary-check",
        );
    }

    let prev_epoch = match (&prev_head, prev_chain_standing) {
        // A legacy (pre-F1, base-rule) head's `v2` payload is not covered by
        // its chain hash: read it as the fixed legacy-bounded sentinel
        // (ceiling `LegacyObserved`, unrecorded ledger), exactly like a v1
        // record, so an unprovable stored ceiling/ledger never feeds the
        // ratchet or the equivalence map.
        (Some(_), Some(ChainStanding::LegacyV2Unbound)) => Some(ReceiptEpochV2::legacy_bounded()),
        (Some(prev), _) => Some(read_receipt_epoch(&prev.record).map_err(|e| {
            AppError::fm_chain(10, format!("previous receipt epoch unreadable: {e}"))
        })?),
        (None, _) => None,
    };
    let prev_ceiling = match &prev_epoch {
        Some(epoch) => epoch.standing_ceiling,
        // Genesis: no prior evidence to constrain the ceiling yet -- Green
        // is the lattice's top/identity element (see `CeilingLevel::Green`).
        None => CeilingLevel::Green,
    };

    // The admission-derived level is this generation's uniform floor: every
    // axis starts there (see `ComponentLevels::uniform`'s doc comment --
    // sync-time has no independent lint/test/fmt observations of its own).
    // Admitted external `ver:Check` evidence then refines individual axes
    // strictly by the meet law: a red, non-allowlisted check lowers its axis
    // to Red; a green (or allowlisted-red) check leaves the axis at the
    // floor -- evidence can never raise an axis above the floor, and an axis
    // with NO evidence stays exactly at the floor, never rounded up.
    let provisional_level = if admission_items
        .iter()
        .any(|i| i.decision == AdmissionDecision::Refused)
    {
        AndonLevel::Red
    } else if admission_items
        .iter()
        .any(|i| i.decision == AdmissionDecision::Quarantined)
    {
        AndonLevel::Yellow
    } else {
        AndonLevel::Green
    };
    let mut components = ComponentLevels::uniform(provisional_level);
    for (name, exit) in &ver_checks {
        let evidence_level = if *exit == 0 || allowlisted_divergences.contains(name) {
            AndonLevel::Green
        } else {
            AndonLevel::Red
        };
        let axis = match verify_check_axis(name) {
            "test" => &mut components.test,
            "lint" => &mut components.lint,
            "fmt" => &mut components.fmt,
            _ => &mut components.gate,
        };
        *axis = (*axis).min(evidence_level);
    }
    // A red class-producer fact (receipts/evidence/binary) lowers the `gate`
    // aggregate axis by the same meet law -- evidence can lower, never raise.
    if producer_red {
        components.gate = components.gate.min(AndonLevel::Red);
    }

    // Equivalence: `source`/`config` are evaluated against the previous
    // receipt's input closure by byte-hash equality; `tests`/`gates` are
    // evaluated from admitted external `ver:Check` outcomes vs the previous
    // receipt's recorded `ver:` admission items (see
    // `compare_verify_class` -- `Unknown` whenever either side lacks
    // evidence for the class). Genesis (no previous receipt) has nothing to
    // compare against, so all four stay `Unknown` honestly rather than
    // defaulting to `Equivalent`-from-absence. The remaining four classes
    // stay explicitly `Unknown` because this call site does not evaluate
    // them.
    let equivalence = match (&prev_head, &prev_epoch) {
        (Some(prev), Some(prev_epoch)) => {
            let mut map = EquivalenceMap::all_unknown();
            map.source = compare_closure_class("source", &report.closure, &prev.payload.closure);
            map.config = compare_closure_class("config", &report.closure, &prev.payload.closure);
            map.tests = compare_verify_class("tests", true, &admission_items, prev_epoch);
            map.gates = compare_verify_class("gates", false, &admission_items, prev_epoch);
            // The four producer-fed classes: evaluated ONLY from admitted
            // producer facts (never inferred at sync). `docs`/`evidence`/
            // `compiled_binary` compare this run's identity against the
            // previous receipt's recorded one; `receipts` has a stronger
            // rule -- the producer's replayed chain head must equal the
            // VERIFIED previous head this very receipt chains from.
            map.docs = compare_producer_class("docs", docs_fact.as_ref(), prev_epoch);
            map.evidence = compare_producer_class("evidence", evidence_fact.as_ref(), prev_epoch);
            map.compiled_binary = if binary_skipped {
                EquivalenceStatus::Unknown
            } else {
                compare_producer_class("compiled_binary", binary_fact.as_ref(), prev_epoch)
            };
            map.receipts = match &receipts_fact {
                None => EquivalenceStatus::Unknown,
                Some(fact) if fact.exit != 0 => EquivalenceStatus::Divergent(format!(
                    "receipts producer recorded red evidence (exit {}): the receipt log \
                     failed schema validation or chain replay",
                    fact.exit
                )),
                Some(fact) if fact.identity == prev.record.chain_hash_hex => {
                    EquivalenceStatus::Equivalent
                }
                Some(fact) => EquivalenceStatus::Divergent(format!(
                    "receipts producer replayed chain head {} does not match the \
                     verified previous head {}",
                    fact.identity, prev.record.chain_hash_hex
                )),
            };
            map
        }
        _ => EquivalenceMap::all_unknown(),
    };

    let mut epoch_builder =
        ReceiptEpochV2Builder::new(prev_ceiling, components).equivalence(equivalence);
    for item in admission_items {
        epoch_builder = epoch_builder.admission_item(item);
    }
    let epoch = epoch_builder
        .build()
        .map_err(|e| AppError::fm_chain(11, format!("receipt epoch construction failed: {e}")))?;

    // The legacy top-level `Andon` field is derived from the same v2
    // precedence, not hardcoded -- this is the hardcoded-`Green` literal
    // this migration replaces.
    let legacy_andon = match epoch.andon {
        AndonLevel::Green => Andon::Green,
        AndonLevel::Yellow => Andon::Overridden {
            by: "ggen.sync.v2".to_string(),
            reason: "quarantined admission item(s); see receipt v2.admission".to_string(),
            at: 0,
        },
        AndonLevel::Red => Andon::Halted {
            unmet: Vec::new(),
            refusals: Vec::new(),
            at: 0,
        },
    };
    let legacy_obligation_count = epoch.obligation_count.remaining().unwrap_or(0);

    let mut record = ReceiptRecord {
        version: RECEIPT_RECORD_VERSION,
        instruction_id: 0,
        activity_idx: 0,
        activity: Some("ggen.sync".to_string()),
        node_kind: 0,
        ts_ns: 0,
        duration_ms: None,
        origin: receipt_origin.map(str::to_string),
        object_ids: vec![format!("law:{}", &payload_hash_hex[..16])],
        payload_hash_hex,
        prev_chain_hash_hex,
        chain_hash_hex: String::new(),
        andon: legacy_andon,
        obligation_count: legacy_obligation_count,
        signature_hex: None,
        schema: SCHEMA_V2.to_string(),
        v2: Some(epoch),
        // Every newly written v2 record declares the rule that seals it, so
        // verifiers never have to infer it (see `ReceiptRecord::chain_rule`).
        chain_rule: Some(CHAIN_RULE_V2_FOLD.to_string()),
    };
    let chain = record
        .recompute_chain_hash()
        .map_err(|e| AppError::fm_chain(2, format!("receipt chain computation failed: {e}")))?;
    record.chain_hash_hex = hex32(&chain);

    // Sign the chain hash (T063): resolve the signing key per the documented
    // policy (`GGEN_SIGNING_KEY` env var, else `.ggen/keys/signing.key`, else
    // generate+persist a fresh keypair), sign `chain_hash_hex`'s UTF-8 bytes,
    // and hex-encode the raw ed25519 signature into `signature_hex`. Fails
    // closed: a malformed env var/key file is a hard error, never a silently
    // unsigned receipt.
    {
        use ed25519_dalek::Signer as _;
        let signing_key = crate::keys::resolve_signing_key(root)?;
        let signature = signing_key.sign(record.chain_hash_hex.as_bytes());
        record.signature_hex = Some(hex::encode(signature.to_bytes()));
    }

    let receipt = SyncReceipt { record, payload };
    if let Some(parent) = receipt_path.parent() {
        std::fs::create_dir_all(parent)?;
    }

    // Append-only history first (it is the chain's source of truth for the
    // next sync), head pointer second: if the head write fails, a re-run
    // chains from the log tail and the history stays linear.
    let mut line = serde_json::to_vec(&receipt)?;
    line.push(b'\n');
    std::fs::OpenOptions::new()
        .create(true)
        .append(true)
        .open(&log_path)
        .and_then(|mut f| f.write_all(&line))
        .map_err(|e| {
            AppError::fm_chain(
                4,
                format!("receipt log `{}` append failed: {e}", log_path.display()),
            )
        })?;
    std::fs::write(&receipt_path, serde_json::to_vec(&receipt)?)?;
    Ok(())
}

/// Escape a plain string for embedding in a Turtle `"..."` literal
/// (backslash and double-quote only — the values this is used for, hex
/// hashes and a `{:?}`-formatted [`Andon`] (graphlaw-backed via
/// `crate::receipt_chain_seam`), never contain
/// literal newlines in the sync-writer's own output, but escaping is
/// unconditional rather than assumed).
fn turtle_escape(s: &str) -> String {
    s.replace('\\', "\\\\").replace('"', "\\\"")
}

/// Lowercase hex of a 32-byte hash.
pub(crate) fn hex32(bytes: &[u8; 32]) -> String {
    use std::fmt::Write as _;
    let mut s = String::with_capacity(64);
    for b in bytes {
        let _ = write!(s, "{b:02x}");
    }
    s
}

#[cfg(test)]
#[allow(clippy::unwrap_used, clippy::expect_used)]
mod tests {
    use std::{collections::BTreeMap, path::Path};

    use super::{
        check_determinism_guard_agrees_false, check_determinism_rows_agree_empty,
        hash_file_or_missing, resolve_gate_entries, ExtractedRows, SyncReceipt,
        RECEIPT_LOG_REL_PATH, RECEIPT_REL_PATH,
    };

    /// Gate-path glob expansion (fix-forward): a `gates/*.rq` entry over a
    /// 3-file gates dir expands to exactly those 3 files, in sorted order;
    /// a non-matching glob is a typed, fail-closed `[FM-CONFIG-014]`
    /// refusal; literal entries pass through unchanged.
    #[test]
    fn gate_glob_expansion_is_sorted_and_fail_closed() {
        let dir = tempfile::tempdir().expect("tempdir");
        let root = dir.path();
        std::fs::create_dir_all(root.join("gates")).expect("mkdir gates");
        // Create deliberately out of lexical order; expansion must sort.
        for name in ["c_gate.rq", "a_gate.rq", "b_gate.rq"] {
            std::fs::write(root.join("gates").join(name), "ASK { FILTER(false) }")
                .expect("write gate file");
        }

        // 1. `gates/*.rq` -> all three, sorted.
        let resolved = resolve_gate_entries(root, &[std::path::PathBuf::from("gates/*.rq")])
            .expect("glob resolves");
        let rels: Vec<&str> = resolved.iter().map(|(rel, _)| rel.as_str()).collect();
        assert_eq!(
            rels,
            vec!["gates/a_gate.rq", "gates/b_gate.rq", "gates/c_gate.rq"]
        );

        // 2. `gates/*.norq` matches nothing -> typed fail-closed refusal.
        let err = resolve_gate_entries(root, &[std::path::PathBuf::from("gates/*.norq")])
            .expect_err("non-matching glob refuses");
        let msg = err.to_string();
        assert!(msg.contains("FM-CONFIG-014"), "{msg}");
        assert!(msg.contains("GLOB_NO_MATCHES"), "{msg}");
        assert!(msg.contains("gates/*.norq"), "{msg}");

        // 3. Literal entries unchanged (no `*`/`?` -> single path as given).
        let literal = resolve_gate_entries(root, &[std::path::PathBuf::from("gates/a_gate.rq")])
            .expect("literal resolves");
        assert_eq!(literal.len(), 1);
        assert_eq!(literal[0].0, "gates/a_gate.rq");

        // 4. `?` matches exactly one character, per component.
        let qmark = resolve_gate_entries(root, &[std::path::PathBuf::from("gates/?_gate.rq")])
            .expect("?-glob resolves");
        assert_eq!(qmark.len(), 3, "? matches exactly one char: {qmark:?}");
    }

    /// One-time, explicit migration: re-seals `.ggen-v2/receipt-log.jsonl`'s
    /// `chain_hash_hex`/`prev_chain_hash_hex`/`signature_hex` for every
    /// record after commit 59df7d812 ("F1 fix": `recompute_chain_hash`
    /// started folding a record's `v2` epoch payload into the chain hash,
    /// closing a real gap where `standing_ceiling`/`admission`/
    /// `equivalence`/`promotion_eligible` could be edited in place without
    /// invalidating `chain_hash_hex`). That fix did not migrate the
    /// receipts already on disk that were sealed under the pre-fix
    /// (non-folding) formula, so `ggen sync run`/`receipt verify` correctly
    /// refuse to extend the chain against them (`[FM-CHAIN-007/009/014]`).
    ///
    /// `#[ignore]`d — a deliberate, run-once, disclosed migration, not a
    /// normal test — and destructive (rewrites `.ggen-v2/receipt-log.jsonl`
    /// and `.ggen-v2/receipt.json` in place): run explicitly via
    /// `cargo test -p ggen-engine --lib -- --ignored --nocapture reseal_receipt_log_under_post_f1_chain_hash_formula`.
    ///
    /// What this does NOT claim: the pre-fix formula never covered the `v2`
    /// payload, so there is no way to cryptographically prove whether any
    /// individual legacy record's `v2` payload was edited before this
    /// migration runs. This migration trusts the `v2` content currently on
    /// disk as ground truth (the repository owner's own explicit decision,
    /// made in response to being asked, not assumed) and re-seals from that
    /// point forward under the current, post-F1 formula — closing the gap
    /// for every record from here on, not retroactively proving history
    /// that cannot be proven.
    #[test]
    #[ignore = "one-time destructive receipt migration; bind GGEN_RECEIPT_MIGRATION_ROOT for a project-scoped legacy chain"]
    fn reseal_receipt_log_under_post_f1_chain_hash_formula() {
        use ed25519_dalek::Signer as _;

        // Default to the historical workspace-root migration target, but allow an
        // operator to bind the destructive migration to an exact admitted project root.
        // This preserves the existing F1 reseal algorithm while making it reusable for
        // project-scoped legacy chains such as examples/interview-sandbox.
        let root = std::env::var_os("GGEN_RECEIPT_MIGRATION_ROOT").map_or_else(
            || {
                std::path::PathBuf::from(env!("CARGO_MANIFEST_DIR"))
                    .parent()
                    .and_then(std::path::Path::parent)
                    .expect("workspace root")
                    .to_path_buf()
            },
            std::path::PathBuf::from,
        );
        let log_path = root.join(RECEIPT_LOG_REL_PATH);
        let head_path = root.join(RECEIPT_REL_PATH);

        let content = std::fs::read_to_string(&log_path).expect("read receipt-log.jsonl");
        let mut receipts: Vec<SyncReceipt> = content
            .lines()
            .filter(|l| !l.trim().is_empty())
            .map(|l| serde_json::from_str(l).expect("parse SyncReceipt line"))
            .collect();
        assert!(
            !receipts.is_empty(),
            "receipt log is empty — nothing to migrate"
        );

        // Resolve the signing key exactly the way this file's own
        // `write_receipt` does at real sync-write time (`GGEN_SIGNING_KEY`
        // env var, else `.ggen/keys/signing.key`, else generate+persist) —
        // never a bespoke migration-only key.
        let signing_key = crate::keys::resolve_signing_key(&root).expect("resolve signing key");

        let mut prev_chain_hash_hex = "0".repeat(64);
        let mut resealed = 0usize;
        let mut unchanged = 0usize;

        for receipt in &mut receipts {
            receipt.record.prev_chain_hash_hex = prev_chain_hash_hex.clone();

            let new_chain = receipt
                .record
                .recompute_chain_hash()
                .expect("recompute_chain_hash under current (post-F1) formula");
            let new_chain_hex = hex::encode(new_chain);

            if new_chain_hex == receipt.record.chain_hash_hex {
                unchanged += 1;
            } else {
                receipt.record.chain_hash_hex = new_chain_hex.clone();
                let signature = signing_key.sign(receipt.record.chain_hash_hex.as_bytes());
                receipt.record.signature_hex = Some(hex::encode(signature.to_bytes()));
                resealed += 1;
            }

            prev_chain_hash_hex = receipt.record.chain_hash_hex.clone();
        }

        // Verify every record self-verifies under the current formula
        // before writing anything back — never persist a migration that
        // would itself fail `receipt verify`.
        let mut running_prev = "0".repeat(64);
        for (i, receipt) in receipts.iter().enumerate() {
            assert_eq!(
                receipt.record.prev_chain_hash_hex, running_prev,
                "record {i}: chain linkage broken after migration"
            );
            let recomputed = receipt
                .record
                .recompute_chain_hash()
                .unwrap_or_else(|e| panic!("record {i}: recompute failed post-migration: {e}"));
            assert_eq!(
                hex::encode(recomputed),
                receipt.record.chain_hash_hex,
                "record {i}: chain_hash_hex does not self-verify post-migration"
            );
            running_prev = receipt.record.chain_hash_hex.clone();
        }

        let mut out = String::new();
        for receipt in &receipts {
            out.push_str(&serde_json::to_string(receipt).expect("serialize SyncReceipt"));
            out.push('\n');
        }
        std::fs::write(&log_path, out).expect("write resealed receipt-log.jsonl");

        let head = receipts.last().expect("non-empty receipts");
        std::fs::write(
            &head_path,
            serde_json::to_vec(head).expect("serialize head SyncReceipt"),
        )
        .expect("write resealed receipt.json head");

        eprintln!(
            "migration complete: {resealed} records re-sealed, {unchanged} already matched current formula (total {})",
            receipts.len()
        );
    }

    /// A declared closure input that cannot be read binds as the literal
    /// `MISSING` marker — recorded in the receipt, never dropped.
    #[test]
    fn unreadable_closure_input_binds_as_missing() {
        let dir = tempfile::tempdir().expect("tempdir");
        let gone = dir.path().join("deleted.tmpl");
        assert_eq!(hash_file_or_missing(&gone), "MISSING");

        let real = dir.path().join("real.tmpl");
        std::fs::write(&real, b"body").expect("write");
        assert_eq!(
            hash_file_or_missing(&real),
            blake3::hash(b"body").to_hex().to_string()
        );
    }

    fn tpl_path() -> &'static Path {
        Path::new("templates/example.tmpl")
    }

    // ── F1 regression: `determinism: true`'s independent recheck must not
    // be silently discarded whenever the PRIMARY extraction produced
    // nothing to render (`when:` guard false, or a zero-row `for_each`/
    // implicit fan-out driver). These exercise the real, non-mocked pure
    // comparison functions the fix added, called from `sync`'s per-template
    // loop skip branches — the exact call sites F1 named. ──

    #[test]
    fn guard_agrees_false_is_noop_when_determinism_not_requested() {
        // determinism_recheck is None whenever `determinism: true` was
        // never set — must never error regardless of what an (unused)
        // recheck would have looked like.
        assert!(check_determinism_guard_agrees_false(tpl_path(), None).is_ok());
    }

    #[test]
    fn guard_agrees_false_ok_when_recheck_guard_also_false() {
        // Both the primary and the independent recheck evaluated the
        // `when:` guard to false — fully consistent, no divergence.
        let recheck: ExtractedRows = None;
        assert!(check_determinism_guard_agrees_false(tpl_path(), Some(&recheck)).is_ok());
    }

    #[test]
    fn guard_agrees_false_errors_when_recheck_guard_evaluated_true() {
        // The exact bug F1 describes: the PRIMARY `when:` guard evaluated
        // false (nothing to render), but the independently reloaded second
        // graph's guard evaluated true. That asymmetry must surface as
        // [FM-TPL-009], not vanish into a silent "skipped: when guard
        // false" with the already-computed recheck never consulted.
        let recheck: ExtractedRows = Some((BTreeMap::new(), Vec::new()));
        let err = check_determinism_guard_agrees_false(tpl_path(), Some(&recheck))
            .expect_err("must reject a guard-true recheck after a guard-false primary");
        let msg = err.to_string();
        assert!(msg.contains("FM-TPL-009"), "{msg}");
        assert!(
            msg.contains("evaluated true while the first evaluated false"),
            "{msg}"
        );
    }

    #[test]
    fn rows_agree_empty_is_noop_when_determinism_not_requested() {
        assert!(
            check_determinism_rows_agree_empty(tpl_path(), None, "named `for_each:` driver")
                .is_ok()
        );
    }

    #[test]
    fn rows_agree_empty_ok_when_recheck_also_zero_rows() {
        let recheck: ExtractedRows = Some((BTreeMap::new(), Vec::new()));
        assert!(check_determinism_rows_agree_empty(
            tpl_path(),
            Some(&recheck),
            "named `for_each:` driver"
        )
        .is_ok());
    }

    #[test]
    fn rows_agree_empty_errors_when_recheck_guard_evaluated_false() {
        // The primary reached the for_each/fan-out row check at all only
        // because its own `when:` guard (if any) evaluated true; a recheck
        // whose guard evaluates false is the same class of divergence
        // `check_determinism`'s success path already rejects.
        let recheck: ExtractedRows = None;
        let err = check_determinism_rows_agree_empty(
            tpl_path(),
            Some(&recheck),
            "named `for_each:` driver",
        )
        .expect_err("must reject a guard-false recheck after a zero-row primary");
        assert!(err.to_string().contains("FM-TPL-009"));
    }

    #[test]
    fn rows_agree_empty_errors_when_recheck_produced_rows_after_primary_produced_zero() {
        // The exact bug F1 describes for the for_each/fan-out skip points:
        // the PRIMARY driving row set was genuinely empty (a lawful zero),
        // but the independently reloaded second graph bound >= 1 row for
        // the identical query. Previously this divergence was silently
        // discarded at the `continue` in sync.rs's per-template loop; the
        // (expensive) recheck had already been computed but was never
        // consulted.
        let recheck: ExtractedRows =
            Some((BTreeMap::new(), vec![serde_json::json!({"name": "alice"})]));
        let err =
            check_determinism_rows_agree_empty(tpl_path(), Some(&recheck), "implicit row fan-out")
                .expect_err("must reject a non-empty recheck after a zero-row primary");
        let msg = err.to_string();
        assert!(msg.contains("FM-TPL-009"), "{msg}");
        assert!(msg.contains("implicit row fan-out"), "{msg}");
        assert!(msg.contains("produced 1 row"), "{msg}");
    }
}