tatara-export-worker 0.2.627

//! Pure decision logic for `tatara-export-worker` — the binary that
//! ships one declared `ExportSpec` from an ephemeral Process to its
//! Vector-native channel.
//!
//! The compounding move: every function in this module is pure (no
//! HTTP, no NATS, no kube client, no clock), takes its inputs by
//! reference, and returns a typed value the I/O layer in `main.rs`
//! then consumes. That means the whole worker is unit-testable
//! without standing up infrastructure — and any new artifact source
//! or channel can be added by extending this module first, then the
//! I/O glue mechanically follows.
//!
//! Three substrate primitives live here:
//!
//! 1. [`prepare_event_payload`] — given an [`ArtifactVariant`] + raw
//!    artifact bytes + run id + signal_type, produces the JSON event
//!    the channel will ship. Encoded once, shared by all channels.
//!
//! 2. [`resolve_run_id`] / [`resolve_subject`] — string-template
//!    substitution for `{{run_id}}` in NATS subjects + event labels.
//!    Single source of truth so the worker, the reconciler, and any
//!    downstream cohort-correlation logic agree on what the run id
//!    means.
//!
//! 3. [`compose_export_receipt`] — builds a typed `ReceiptEnvelope`
//!    of the export action itself, with the three BLAKE3 pillars
//!    derived from the ExportSpec (intent), the shipped payload
//!    bytes (artifact), and the outcome (control). The receipt
//!    chains into the Process's attestation tree, so the act of
//!    exporting is itself attested.
//!
//! The I/O glue in `main.rs` is thin — argv → ExportSpec → call
//! these functions → ship to channel → write receipt.

use std::collections::BTreeMap;

use chrono::{DateTime, Utc};
use serde::{Deserialize, Serialize};

use tatara_process::export::{
    ArtifactVariant, ExportSpec, NatsSubjectChannel, ReportFormat, ReportPayloadShape,
    RunMarkerSource,
};
use tatara_process::json_object::JsonMapStrExt;
use tatara_process::receipt::{ReceiptEnvelope, ReceiptKind};
use tatara_process::string_map::BTreeMapStrExt;

// ─── Run id resolution ─────────────────────────────────────────────

/// Resolve the run id used in event labels + subject templates.
///
/// Precedence:
/// 1. `spec.experiment_id_override` when set
/// 2. `{process_namespace}/{process_name}` otherwise
///
/// Single source of truth so every channel (HTTP, NATS, stdout) and
/// every downstream consumer (shinryu cohort math, Vector
/// transforms) agree on what "run id" means for a given export.
///
/// The fallback branch composes the `<ns>/<name>` shape through the
/// substrate primitive [`tatara_process::prelude::qualified_process_ref`],
/// matching what every reconciler-side annotation seed / claim key /
/// label selector composes — so a Process's run-id in the
/// no-override branch is byte-identical to the value the same
/// Process's `tatara.pleme.io/process` annotation carries and the
/// value a `PROCESS=<ref>` label-selector filters on.
pub fn resolve_run_id(spec: &ExportSpec, namespace: &str, name: &str) -> String {
    if let Some(o) = &spec.experiment_id_override {
        if !o.is_empty() {
            return o.clone();
        }
    }
    tatara_process::prelude::qualified_process_ref(namespace, name)
}

/// Substitute `{{run_id}}` placeholders in a NATS subject template.
///
/// The chart's subject template (e.g.
/// `pleme.pleme-dev.ephemeral.{{run_id}}.receipt`) gets expanded
/// once, here, before the NATS publish call.
pub fn resolve_subject(channel: &NatsSubjectChannel, run_id: &str) -> String {
    channel.subject.replace("{{run_id}}", run_id)
}

// ─── Event payload preparation ─────────────────────────────────────

/// JSON event shape shipped through every `VectorChannel`. Stable
/// schema — shinryu's analytical SQL plane reads from it directly.
#[derive(Clone, Debug, Serialize, Deserialize)]
pub struct ExportEvent {
    /// One of "receipt" / "test-report" / "process-snapshot" /
    /// "run-marker". Carried both in the body (for downstream SQL)
    /// and in the channel metadata (HttpEventChannel.signal_type or
    /// NATS subject path).
    pub signal_type: String,

    /// Resolved run id — `{process_namespace}/{process_name}` by
    /// default, or `spec.experiment_id_override` when set.
    pub run_id: String,

    /// Timestamp the export was prepared (RFC 3339 UTC).
    pub timestamp: DateTime<Utc>,

    /// Artifact-source-specific labels — empty for Receipts /
    /// ProcessSnapshot, ConfigMap reference for TestReport, free-form
    /// for RunMarker.
    pub labels: BTreeMap<String, String>,

    /// The shipped artifact bytes, embedded as the `payload` field.
    /// For JSON sources (receipts, snapshots) this is a JSON Value;
    /// for opaque bytes (TestReport with format=Raw) it's a base64
    /// string under the `raw` key.
    pub payload: serde_json::Value,

    /// Format hint copied from `TestReportSource.format` when
    /// applicable. Lets downstream parsers branch by JUnit / TAP /
    /// NDJSON / Raw without inspecting the bytes.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub format: Option<ReportFormat>,
}

/// Build the JSON event the channel ships.
///
/// `artifact_bytes` is the raw artifact (ConfigMap value, snapshot
/// JSON, or empty for run-markers). The function dispatches on
/// `source` to embed the bytes the right way:
///
/// - **Receipts** — `artifact_bytes` is a JSON array of receipt
///   envelopes; embedded under `payload.receipts`.
/// - **TestReport** — `artifact_bytes` is the report file's bytes;
///   embedded as either a parsed JSON value (when `format=NdJson`)
///   or a base64 string (every other format).
/// - **ProcessSnapshot** — `artifact_bytes` is the Process JSON;
///   embedded under `payload.snapshot`.
/// - **RunMarker** — `artifact_bytes` is ignored; labels come from
///   `RunMarkerSource.labels`.
pub fn prepare_event_payload(
    source: ArtifactVariant<'_>,
    artifact_bytes: &[u8],
    run_id: &str,
    signal_type: &str,
    now: DateTime<Utc>,
) -> ExportEvent {
    let mut labels = BTreeMap::new();
    // The three `<BTreeMap<String, String>>.insert(<key>.into(),
    // <val>.<coerce>)` label writes route through the ONE substrate
    // primitive `tatara_process::string_map::BTreeMapStrExt::insert_str`
    // (opened at 565c5f2, which already routed the receipt-CM `.data`
    // seed at `tatara-export-worker::main::write_receipt` onto the same
    // owner). Pre-lift this file restated the SAME `.insert(<k>.into(),
    // <v>.<coerce>)` shape at THREE production sites past the ★★
    // PRIME-DIRECTIVE ≥ 2 duplication threshold — the `run_id` label
    // seed here + the two `configmap` / `key` labels stamped inside the
    // `ArtifactVariant::TestReport` arm below. Post-lift every label
    // write in `prepare_event_payload` rides through the SAME substrate
    // owner the pool-reconciler + write_receipt callers already thread.
    labels.insert_str("run_id", run_id);

    let (payload, format) = match source {
        ArtifactVariant::Receipts(_) => {
            // Receipts are JSON; the worker pre-parses them into an array.
            let parsed: serde_json::Value =
                serde_json::from_slice(artifact_bytes).unwrap_or(serde_json::Value::Array(vec![]));
            (serde_json::json!({ "receipts": parsed }), None)
        }
        ArtifactVariant::TestReport(tr) => {
            labels.insert_str("configmap", tr.configmap.as_str());
            labels.insert_str("key", tr.key.as_str());
            // Closed-set dispatch via `ReportFormat::payload_shape` — the
            // 2-arm match over `ReportPayloadShape` is exhaustive, so
            // adding a future `ReportFormat` variant lands at one
            // `payload_shape` arm in tatara-process and never touches
            // the worker. Replaces the prior `_ => base64` silent
            // default that quietly swallowed new variants.
            let p = match tr.format.payload_shape() {
                ReportPayloadShape::NdJsonLines => {
                    let lines: Vec<serde_json::Value> = artifact_bytes
                        .split(|b| *b == b'\n')
                        .filter(|l| !l.is_empty())
                        .filter_map(|l| serde_json::from_slice(l).ok())
                        .collect();
                    serde_json::json!({ "ndjson": lines })
                }
                ReportPayloadShape::OpaqueBytes => {
                    use base64_inline as base64;
                    serde_json::json!({ "raw_b64": base64::encode(artifact_bytes) })
                }
            };
            (p, Some(tr.format))
        }
        ArtifactVariant::ProcessSnapshot(_) => {
            let parsed: serde_json::Value =
                serde_json::from_slice(artifact_bytes).unwrap_or(serde_json::Value::Null);
            (serde_json::json!({ "snapshot": parsed }), None)
        }
        ArtifactVariant::RunMarker(rm) => {
            merge_labels(&mut labels, &rm.labels);
            (serde_json::Value::Null, None)
        }
    };

    ExportEvent {
        signal_type: signal_type.to_string(),
        run_id: run_id.to_string(),
        timestamp: now,
        labels,
        payload,
        format,
    }
}

fn merge_labels(into: &mut BTreeMap<String, String>, from: &BTreeMap<String, String>) {
    for (k, v) in from {
        into.insert(k.clone(), v.clone());
    }
}

/// Convenience for the worker — calls the right run marker
/// preparation when no artifact bytes exist (e.g. start/end markers
/// the worker synthesizes itself).
pub fn run_marker_event(
    rm: &RunMarkerSource,
    run_id: &str,
    signal_type: &str,
    now: DateTime<Utc>,
) -> ExportEvent {
    prepare_event_payload(
        ArtifactVariant::RunMarker(rm),
        &[],
        run_id,
        signal_type,
        now,
    )
}

// ─── Outcome + receipt composition ─────────────────────────────────

/// Final state of the export action — feeds the `control_hash`
/// pillar of the typed receipt.
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub enum ExportOutcome {
    /// The shipment succeeded — the destination acknowledged
    /// (HTTP 2xx, NATS publish ack, stdout written).
    Shipped,
    /// The destination explicitly rejected the shipment (HTTP 4xx,
    /// NATS no-stream-match). Worker emits a receipt of the failure;
    /// the Process advances to Zombie via Releasing.
    Rejected(String),
    /// The shipment timed out / connection refused / network error.
    /// Same Zombie path; the receipt records the error type.
    Failed(String),
}

impl ExportOutcome {
    /// One short token used in the receipt's `kind` field.
    pub fn kind(&self) -> &'static str {
        match self {
            Self::Shipped => "Shipped",
            Self::Rejected(_) => "Rejected",
            Self::Failed(_) => "Failed",
        }
    }

    /// True iff the outcome is a successful shipment.
    pub fn is_shipped(&self) -> bool {
        matches!(self, Self::Shipped)
    }
}

/// Build a typed `ReceiptEnvelope` of the export action via the
/// existing `ReceiptEnvelope::build()` constructor (single source of
/// truth for the three-pillar composition + BLAKE3 root).
///
/// Three BLAKE3 pillars (each fed to `build()` as a hex string):
/// - **intent_hash**  ← canonical JSON of the `ExportSpec`
/// - **artifact_hash** ← the shipped event bytes (post-`prepare_event_payload`)
/// - **control_hash**  ← canonical JSON of the `ExportOutcome`
///
/// Returned envelope has `kind = ReceiptKind::Export.as_str()`
/// (wire-form `"tatara.export"`), `process_ref` stamped as
/// `{namespace}/{name}`, and structured `evidence` carrying the run
/// id, outcome kind, and any error string. The composed root +
/// version + generated_at are set by `build()`.
///
/// tatara-reconciler's `JobAttested` evaluator reads this envelope
/// from the worker's ConfigMap and verifies the root before
/// advancing the Process out of `Releasing`.
pub fn compose_export_receipt(
    spec: &ExportSpec,
    shipped_event_bytes: &[u8],
    outcome: &ExportOutcome,
    previous_root: Option<&str>,
    run_id: &str,
    process_ref: Option<&str>,
) -> anyhow::Result<ReceiptEnvelope> {
    use tatara_process::hash::hex_blake3;
    use tatara_process::three_pillar::canonical_bytes;
    // Intent + control pillar bytes route through the ONE substrate
    // primitive `tatara_process::three_pillar::canonical_bytes` — the
    // strict, error-propagating peer of `pillar_bytes` that owns the
    // 2-link `serde_json::to_value → to_vec` canonicalization chain.
    // Pre-lift this site read through a module-private `canonical_json`
    // helper (removed) that restated the same 2-link chain byte-for-byte
    // alongside the peer at `tatara_process::hostname::canonical_json`
    // — two hand-authored sites past the ★★ PRIME-DIRECTIVE ≥ 2
    // duplication threshold. Post-lift both consumers name the payload
    // ONCE and route through the ONE substrate owner; the concrete
    // `serde_json::Error` composes into `anyhow::Error` via `?` at this
    // callsite (matching the pre-lift error-forwarding shape). The
    // artifact-pillar `shipped_event_bytes` slot is already bytes and
    // rides through `hex_blake3` directly — the canonicalize-then-hash
    // shape only applies to the two typed-input pillars.
    let intent_hash = hex_blake3(&canonical_bytes(spec)?);
    let artifact_hash = hex_blake3(shipped_event_bytes);
    let control_hash = hex_blake3(&canonical_bytes(outcome)?);

    // The `kind` slot routes through the typed
    // `tatara_process::receipt::ReceiptKind::Export` variant — pre-lift
    // this was a bare `"tatara.export"` `&'static str` argument passed
    // to `ReceiptEnvelope::build`, one hand-authored production
    // restatement of a wire-form literal `ReceiptKind::ClosedLoopAuth`
    // and every other substrate-emitted kind (`DbMigration`,
    // `TestSuite`, `NixBuild`) already route through the typed variant
    // at their author sites. Post-lift the export worker joins the same
    // typed dispatch shape every peer receipt author uses — a rename of
    // the wire form (`"tatara.export"` → `"export"`, or a normalization
    // pass to kebab-case) lands at the ONE `ReceiptKind::as_str` arm in
    // `tatara-process::receipt` and this callsite inherits the upgrade
    // mechanically, and every downstream consumer that iterates
    // `ReceiptKind::ALL` (a future kind-keyed verifier registry, a
    // dashboard completion list, `tatara-check`'s receipt-kind
    // enumeration) sees the Export variant without a grep.
    let mut env = ReceiptEnvelope::build(
        ReceiptKind::Export,
        intent_hash,
        artifact_hash,
        control_hash,
        previous_root,
    );
    env.process_ref = process_ref.map(String::from);

    let mut evidence = serde_json::Map::new();
    // The three `Value::String`-slot evidence writes route through the
    // ONE substrate primitive `tatara_process::json_object::
    // JsonMapStrExt::insert_str` — receiver-shape peer of the
    // `BTreeMapStrExt::insert_str` labels routing above on the same
    // "insert a string at a string key" write axis, split by CARRIER
    // TYPE (JSON-shaped `serde_json::Map<String, Value>` here vs the
    // K8s-canonical `BTreeMap<String, String>` labels map above). Pre-
    // lift THREE production sites in this function restated the SAME
    // `.insert(<k>.into(), Value::String(<v>.<coerce>))` shape past the
    // ★★ PRIME-DIRECTIVE ≥ 2 duplication threshold: the `run_id` +
    // `outcome` seeds unconditionally, and the conditional `error` seed
    // inside the `Rejected | Failed` arm. The `shipped_bytes_len` seed
    // below stays hand-authored because its value slot is a
    // `Value::Number`, not a `Value::String` — outside this trait's axis.
    evidence.insert_str("run_id", run_id);
    evidence.insert_str("outcome", outcome.kind());
    if let ExportOutcome::Rejected(m) | ExportOutcome::Failed(m) = outcome {
        evidence.insert_str("error", m.as_str());
    }
    evidence.insert(
        "shipped_bytes_len".into(),
        serde_json::Value::Number(shipped_event_bytes.len().into()),
    );
    env.evidence = serde_json::Value::Object(evidence);

    Ok(env)
}

// ─── Minimal inline base64 (no extra dep) ──────────────────────────

mod base64_inline {
    const ALPHA: &[u8] = b"ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/";
    pub fn encode(input: &[u8]) -> String {
        let mut out = String::with_capacity((input.len() + 2) / 3 * 4);
        let mut chunks = input.chunks_exact(3);
        for chunk in chunks.by_ref() {
            let n = ((chunk[0] as u32) << 16) | ((chunk[1] as u32) << 8) | (chunk[2] as u32);
            out.push(ALPHA[((n >> 18) & 0x3F) as usize] as char);
            out.push(ALPHA[((n >> 12) & 0x3F) as usize] as char);
            out.push(ALPHA[((n >> 6) & 0x3F) as usize] as char);
            out.push(ALPHA[(n & 0x3F) as usize] as char);
        }
        let rem = chunks.remainder();
        match rem.len() {
            1 => {
                let n = (rem[0] as u32) << 16;
                out.push(ALPHA[((n >> 18) & 0x3F) as usize] as char);
                out.push(ALPHA[((n >> 12) & 0x3F) as usize] as char);
                out.push('=');
                out.push('=');
            }
            2 => {
                let n = ((rem[0] as u32) << 16) | ((rem[1] as u32) << 8);
                out.push(ALPHA[((n >> 18) & 0x3F) as usize] as char);
                out.push(ALPHA[((n >> 12) & 0x3F) as usize] as char);
                out.push(ALPHA[((n >> 6) & 0x3F) as usize] as char);
                out.push('=');
            }
            _ => {}
        }
        out
    }
}

// ─── Tests ─────────────────────────────────────────────────────────

#[cfg(test)]
mod tests {
    use super::*;
    use tatara_process::export::{
        ArtifactSource, HttpEventChannel, ProcessSnapshotSource, ReceiptsSource, RunMarkerSource,
        TestReportSource, VectorChannel,
    };

    fn http_spec(signal_type: &str) -> ExportSpec {
        ExportSpec {
            source: ArtifactSource {
                run_marker: Some(RunMarkerSource::default()),
                ..ArtifactSource::default()
            },
            channel: VectorChannel {
                http_event: Some(HttpEventChannel::signal(signal_type)),
                ..VectorChannel::default()
            },
            when: Default::default(),
            experiment_id_override: None,
        }
    }

    #[test]
    fn run_id_falls_back_to_ns_slash_name() {
        let s = http_spec("x");
        assert_eq!(resolve_run_id(&s, "demo-test", "r1"), "demo-test/r1");
    }

    #[test]
    fn run_id_uses_override_when_set() {
        let mut s = http_spec("x");
        s.experiment_id_override = Some("demo-run-2026-05-20".into());
        assert_eq!(resolve_run_id(&s, "ns", "n"), "demo-run-2026-05-20");
    }

    #[test]
    fn run_id_ignores_empty_override() {
        let mut s = http_spec("x");
        s.experiment_id_override = Some(String::new());
        assert_eq!(resolve_run_id(&s, "ns", "n"), "ns/n");
    }

    #[test]
    fn subject_substitutes_run_id_template() {
        let ch = NatsSubjectChannel::publish(
            "pleme.pleme-dev.ephemeral.{{run_id}}.receipt",
            "EPHEMERAL_RECEIPTS",
        );
        assert_eq!(
            resolve_subject(&ch, "ns/n"),
            "pleme.pleme-dev.ephemeral.ns/n.receipt"
        );
    }

    #[test]
    fn subject_passthrough_when_no_template() {
        let ch = NatsSubjectChannel::publish("pleme.fixed.subject", "S");
        assert_eq!(resolve_subject(&ch, "ignored"), "pleme.fixed.subject");
    }

    #[test]
    fn run_marker_event_has_labels_and_run_id() {
        let mut labels = BTreeMap::new();
        labels.insert("phase".into(), "end".into());
        let rm = RunMarkerSource { labels };
        let now = chrono::Utc::now();
        let ev = run_marker_event(&rm, "ns/n", "ephemeral-marker", now);
        assert_eq!(ev.signal_type, "ephemeral-marker");
        assert_eq!(ev.run_id, "ns/n");
        assert_eq!(ev.labels["phase"], "end");
        assert_eq!(ev.labels["run_id"], "ns/n");
        assert_eq!(ev.payload, serde_json::Value::Null);
    }

    #[test]
    fn test_report_ndjson_parses_into_array() {
        let tr = TestReportSource {
            configmap: "cm".into(),
            key: "out.ndjson".into(),
            format: ReportFormat::NdJson,
            namespace: None,
        };
        let bytes = b"{\"a\":1}\n{\"b\":2}\n\n{\"c\":3}\n";
        let now = chrono::Utc::now();
        let ev = prepare_event_payload(
            ArtifactVariant::TestReport(&tr),
            bytes,
            "ns/n",
            "test-report",
            now,
        );
        let arr = ev.payload["ndjson"].as_array().unwrap();
        assert_eq!(arr.len(), 3);
        assert_eq!(arr[0]["a"], 1);
        assert_eq!(arr[2]["c"], 3);
        assert_eq!(ev.labels["configmap"], "cm");
        assert_eq!(ev.format, Some(ReportFormat::NdJson));
    }

    #[test]
    fn test_report_labels_ride_through_btreemap_str_ext_bytewise_across_all_three_slots() {
        // Slot-by-slot byte-identical parity witness — the post-lift
        // `ev.labels` BTreeMap from `prepare_event_payload(
        // ArtifactVariant::TestReport ...)` MUST match a hand-authored
        // pre-lift `.insert(<k>.into(), <v>.<coerce>)` chain byte-for-
        // byte across ALL THREE label slots (`run_id`, `configmap`,
        // `key`). Fail-before-pass-after granularity: the `key` label
        // slot at pre-lift line 154 was previously asserted by NO
        // upstream test — `test_report_ndjson_parses_into_array` above
        // checks only `configmap`, and `run_marker_event_has_labels_
        // and_run_id` checks only `run_id` on the RunMarker arm — so
        // this pin closes the last untested label slot on the same
        // `BTreeMapStrExt::insert_str` axis the two adjacent sibling
        // slots ride through. A regression that dropped the `key`
        // insert altogether, swapped either slot's `Into<String>` arm,
        // or reshaped the map's key-ordering would surface HERE rather
        // than as silent operator-facing drift at the downstream
        // Vector-native channel that consumes the shipped `labels` map.
        let tr = TestReportSource {
            configmap: "cm".into(),
            key: "out.ndjson".into(),
            format: ReportFormat::NdJson,
            namespace: None,
        };
        let now = chrono::Utc::now();
        let ev = prepare_event_payload(
            ArtifactVariant::TestReport(&tr),
            b"",
            "ns/n",
            "test-report",
            now,
        );

        let mut expected: BTreeMap<String, String> = BTreeMap::new();
        expected.insert("run_id".to_string(), "ns/n".to_string());
        expected.insert("configmap".to_string(), "cm".to_string());
        expected.insert("key".to_string(), "out.ndjson".to_string());

        assert_eq!(
            ev.labels, expected,
            "BTreeMapStrExt::insert_str lift must produce a labels map byte-identical to the pre-lift .insert(<k>.into(), <v>.<coerce>) chain across every TestReport label slot",
        );
    }

    #[test]
    fn export_receipt_evidence_slots_ride_through_json_map_str_ext_bytewise() {
        // Slot-by-slot byte-identical parity witness on the evidence-
        // side JsonMapStrExt lift — the post-lift `r.evidence` object
        // from `compose_export_receipt(..., Rejected(msg), ...)` MUST
        // carry every `Value::String`-slot at the same key with the
        // same underlying string bytes a hand-authored pre-lift
        // `.insert(<k>.into(), Value::String(<v>.<coerce>))` chain
        // would have written. Fail-before-pass-after granularity: the
        // three-slot cross-product (`run_id` + `outcome` + `error`)
        // was previously asserted at TWO separate tests
        // (`export_receipt_chains_three_pillars` covered `run_id` +
        // `outcome`; `export_receipt_failure_carries_error_text`
        // covered `error` + `outcome`), never as ONE atomic parity
        // witness on the same envelope — so a regression that drifted
        // the `Value::String` wrap at ONE slot while leaving the
        // others intact would pass every upstream pin and surface HERE.
        let spec = http_spec("test-report");
        let outcome = ExportOutcome::Rejected("connection refused".into());
        let r = compose_export_receipt(
            &spec,
            b"payload",
            &outcome,
            None,
            "ns/rejected",
            Some("demo-test/r1"),
        )
        .expect("receipt");
        assert_eq!(r.evidence["run_id"], "ns/rejected");
        assert_eq!(r.evidence["outcome"], "Rejected");
        assert_eq!(r.evidence["error"], "connection refused");
        // Every Value::String-typed evidence slot MUST remain the
        // `Value::String` variant post-lift — a regression that dropped
        // the `Value::String` wrap on the JsonMapStrExt path would
        // stamp the string bytes into a different variant and the
        // `.as_str()` projection below would return `None` on the
        // corner.
        assert!(r.evidence["run_id"].is_string());
        assert!(r.evidence["outcome"].is_string());
        assert!(r.evidence["error"].is_string());
    }

    #[test]
    fn test_report_raw_format_base64_encodes() {
        let tr = TestReportSource {
            configmap: "cm".into(),
            key: "report.bin".into(),
            format: ReportFormat::Raw,
            namespace: None,
        };
        let bytes = b"<<binary>>";
        let now = chrono::Utc::now();
        let ev = prepare_event_payload(
            ArtifactVariant::TestReport(&tr),
            bytes,
            "ns/n",
            "test-report",
            now,
        );
        let b64 = ev.payload["raw_b64"].as_str().unwrap();
        // sanity — base64 length is ceil(N/3)*4
        assert_eq!(b64.len(), ((bytes.len() + 2) / 3) * 4);
        assert_eq!(ev.format, Some(ReportFormat::Raw));
    }

    #[test]
    fn receipts_source_embeds_parsed_json() {
        let r = ReceiptsSource::default();
        let raw = serde_json::to_vec(&serde_json::json!([
            { "kind": "tatara.processed.run", "composed_root": "abc" },
            { "kind": "tatara.processed.run", "composed_root": "def" },
        ]))
        .unwrap();
        let now = chrono::Utc::now();
        let ev = prepare_event_payload(ArtifactVariant::Receipts(&r), &raw, "ns/n", "receipt", now);
        let arr = ev.payload["receipts"].as_array().unwrap();
        assert_eq!(arr.len(), 2);
        assert_eq!(arr[1]["composed_root"], "def");
    }

    #[test]
    fn process_snapshot_embeds_parsed_json() {
        let p = ProcessSnapshotSource::default();
        let raw = serde_json::to_vec(&serde_json::json!({ "phase": "Attested" })).unwrap();
        let now = chrono::Utc::now();
        let ev = prepare_event_payload(
            ArtifactVariant::ProcessSnapshot(&p),
            &raw,
            "ns/n",
            "process-snapshot",
            now,
        );
        assert_eq!(ev.payload["snapshot"]["phase"], "Attested");
    }

    // ─── Receipt composition ───────────────────────────────────────

    #[test]
    fn outcome_kind_is_stable() {
        assert_eq!(ExportOutcome::Shipped.kind(), "Shipped");
        assert_eq!(ExportOutcome::Rejected("x".into()).kind(), "Rejected");
        assert_eq!(ExportOutcome::Failed("y".into()).kind(), "Failed");
    }

    #[test]
    fn export_receipt_chains_three_pillars() {
        use tatara_process::receipt::RECEIPT_VERSION;
        let s = http_spec("test-report");
        let event_bytes = b"{\"signalType\":\"test-report\"}";
        let r = compose_export_receipt(
            &s,
            event_bytes,
            &ExportOutcome::Shipped,
            None,
            "ns/n",
            Some("demo-test/r1"),
        )
        .expect("receipt");
        assert_eq!(r.version, RECEIPT_VERSION);
        // Kind wire-form routes through the ONE typed
        // `tatara_process::receipt::ReceiptKind::Export` arm — a
        // regression that reinlined `"tatara.export"` here or drifted
        // the wire form on the substrate side (a `"export"` rename, a
        // kebab-case normalization) would surface HERE at this pin
        // rather than as silent operator-visible skew between the
        // envelope this test asserts and the envelope
        // `compose_export_receipt` actually builds.
        assert_eq!(r.kind, ReceiptKind::Export.as_str());
        assert_eq!(r.kind, "tatara.export");
        // Each pillar is a 64-char BLAKE3 hex digest.
        assert_eq!(r.intent_hash.len(), 64);
        assert_eq!(r.artifact_hash.len(), 64);
        assert_eq!(r.control_hash.len(), 64);
        assert_eq!(r.composed_root.len(), 64);
        // Process ref + evidence stamped through.
        assert_eq!(r.process_ref.as_deref(), Some("demo-test/r1"));
        assert_eq!(r.evidence["run_id"], "ns/n");
        assert_eq!(r.evidence["outcome"], "Shipped");
        // verify_root() agrees the composed_root was built correctly
        // — same guarantee tatara-reconciler's evaluator checks.
        assert!(r.verify_root(None));
    }

    #[test]
    fn export_receipt_kind_slot_routes_through_typed_receipt_kind_export_arm() {
        // Fail-before-pass-after routing pin: the receipt-`kind` field
        // this crate stamps at every `compose_export_receipt` call
        // MUST route through the ONE typed
        // `tatara_process::receipt::ReceiptKind::Export` variant, not
        // a bare `"tatara.export"` `&'static str` literal (which was
        // the pre-lift shape).
        //
        // A regression that reinlined the wire literal at the
        // `ReceiptEnvelope::build(...)` kind slot — silently reopening
        // the bypass this commit closed — would fail HERE at the
        // routing pin rather than as post-`ReceiptKind::Export::as_str`-
        // rename operator-facing skew between the wire form the
        // reconciler `JobAttested` verifier gates on and the wire form
        // the export worker actually writes into its receipt CM.
        //
        // Also binds the shape of the typed dispatch: because
        // `From<ReceiptKind> for String` composes `as_str().to_owned()`,
        // an export-worker-built envelope's `kind` field is byte-
        // identical to `ReceiptKind::Export.as_str()`. A future rename
        // (`"tatara.export"` → `"export"`, a kebab-case normalization)
        // lands at the ONE `as_str` arm in the substrate; both the
        // wire form THIS crate stamps and every peer receipt author's
        // wire form advance in lockstep.
        let s = http_spec("test-report");
        let r = compose_export_receipt(&s, b"x", &ExportOutcome::Shipped, None, "r", None)
            .expect("receipt");
        assert_eq!(
            r.kind,
            ReceiptKind::Export.as_str(),
            "kind slot must route through the ReceiptKind::Export typed variant",
        );
        assert_eq!(
            ReceiptKind::Export.as_str(),
            "tatara.export",
            "wire-form pin — a bump on the substrate side surfaces here at the export worker",
        );
        // Cross-check via the substrate's own decoder: the built
        // envelope's kind decodes back through `known_kind()` to
        // `Some(ReceiptKind::Export)`. Pre-lift the export kind was
        // an OPEN wire literal (`"tatara.export"`) that
        // `known_kind()` returned `None` for — every peer receipt
        // author (`ClosedLoopAuth`, `DbMigration`, `TestSuite`,
        // `NixBuild`) already round-tripped through this decoder, and
        // the export worker was the ONE substrate-emitted kind that
        // did not. Post-lift the closed-set view is complete for
        // every substrate-emitted receipt.
        assert_eq!(r.known_kind(), Some(ReceiptKind::Export));
    }

    #[test]
    fn export_receipt_chains_prev_root() {
        let s = http_spec("test-report");
        let ev = b"x";
        let r1 = compose_export_receipt(&s, ev, &ExportOutcome::Shipped, None, "r", None).unwrap();
        let r2 = compose_export_receipt(
            &s,
            ev,
            &ExportOutcome::Shipped,
            Some(&r1.composed_root),
            "r",
            None,
        )
        .unwrap();
        // Same inputs but chained prev_root → different composed_root.
        assert_ne!(r1.composed_root, r2.composed_root);
        // verify_root checks the chain.
        assert!(r2.verify_root(Some(&r1.composed_root)));
    }

    #[test]
    fn export_receipt_failure_carries_error_text() {
        let s = http_spec("x");
        let r = compose_export_receipt(
            &s,
            b"",
            &ExportOutcome::Failed("connection refused".into()),
            None,
            "ns/n",
            None,
        )
        .unwrap();
        assert_eq!(r.evidence["error"], "connection refused");
        assert_eq!(r.evidence["outcome"], "Failed");
    }

    #[test]
    fn export_receipt_intent_hash_changes_with_spec() {
        let s1 = http_spec("a");
        let s2 = http_spec("b"); // different signal_type → different intent
        let r1 =
            compose_export_receipt(&s1, b"", &ExportOutcome::Shipped, None, "ns/n", None).unwrap();
        let r2 =
            compose_export_receipt(&s2, b"", &ExportOutcome::Shipped, None, "ns/n", None).unwrap();
        assert_ne!(r1.intent_hash, r2.intent_hash);
    }

    #[test]
    fn export_receipt_artifact_hash_changes_with_payload() {
        let s = http_spec("x");
        let r1 = compose_export_receipt(
            &s,
            b"payload-1",
            &ExportOutcome::Shipped,
            None,
            "ns/n",
            None,
        )
        .unwrap();
        let r2 = compose_export_receipt(
            &s,
            b"payload-2",
            &ExportOutcome::Shipped,
            None,
            "ns/n",
            None,
        )
        .unwrap();
        assert_ne!(r1.artifact_hash, r2.artifact_hash);
    }
}