cellos_core/

events.rs

//! Versioned CloudEvents `data` payloads (JSON only — no I/O).

use std::fmt;

use serde::{Serialize, Serializer};
use serde_json::{json, Map, Value};

use crate::policy::PolicyViolation;
use crate::{
    CloudEventV1, DnsAuthorityDnssecFailed, DnsAuthorityDrift, DnsAuthorityRebindRejected,
    DnsAuthorityRebindThreshold, DnsQueryEvent, ExecutionCellSpec, ExportReceipt,
    NetworkFlowDecision, WorkloadIdentity,
};

/// Subject URN string used by seam-freeze G3/G4 cross-pointers.
///
/// Outcome field for [`lifecycle_destroyed_data_v1`].
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
#[serde(rename_all = "lowercase")]
pub enum LifecycleDestroyOutcome {
    Succeeded,
    Failed,
}

/// How the supervisor learned the cell terminated, for the
/// `terminalState` field of [`lifecycle_destroyed_data_v1`].
///
/// `outcome` answers "did the run succeed?" (no phase error). This enum
/// answers a different question that audit cares about: "did we observe a
/// real exit, or did we give up and kill it?"
///
/// - `Clean`: an authenticated exit code arrived through the in-VM bridge
///   (Firecracker + cellos-init vsock). Whatever code arrived is the truth.
/// - `Forced`: the supervisor never received an authenticated exit code —
///   the bridge errored out (vsock channel closed before the 4 bytes were
///   delivered) and teardown proceeded via SIGKILL. Any exit code recorded
///   on this path is synthetic (e.g. -1) and must not be trusted.
///
/// Omitted from the event payload (encoded as `None`) on host backends that
/// do not run an in-VM bridge, since the clean/forced distinction has no
/// meaning when the supervisor itself owns the workload process.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
#[serde(rename_all = "lowercase")]
pub enum LifecycleTerminalState {
    Clean,
    Forced,
}

/// FC-50 typed `reason` for `dev.cellos.events.cell.lifecycle.v1.failed` /
/// `.destroyed` payloads.
///
/// Replaces the free-form `Option<&str>` reason used by gap-markers in FC-23,
/// FC-52, FC-59, FC-60, FC-61, FC-63, FC-72 with a constrained set of audit-
/// stable codes. Each variant serializes to its `snake_case` form so the JSON
/// wire format is stable for downstream auditors. `Other(String)` is the
/// escape hatch for operator-supplied free-form reasons that have not yet
/// earned a dedicated variant; it serializes verbatim as the inner string.
///
/// The typed surface is non-exhaustive — adding a new variant is a
/// public-API change that requires schema updates on the
/// `cell.lifecycle.v1.failed` / `.destroyed` contracts. Downstream
/// `match`es outside this crate must include a wildcard arm; `#[non_exhaustive]`
/// is enforced by the compiler so silent breaks at variant-add time aren't
/// possible.
#[derive(Debug, Clone, PartialEq, Eq)]
#[non_exhaustive]
pub enum LifecycleReason {
    /// Workload exceeded its memory limit (cgroup OOM kill or VMM-level OOM).
    Oom,
    /// TTL watchdog fired before the workload completed.
    TtlExceeded,
    /// VMM process exited unexpectedly (e.g. Firecracker crashed).
    VmmCrashed,
    /// Kernel/init failed before reaching `/sbin/init`.
    BootFailed,
    /// Supervisor SIGKILLed the workload after the graceful-shutdown timeout
    /// elapsed.
    SignalKilled,
    /// `cellos-init` segfaulted or aborted inside the guest.
    InitCrashed,
    /// Kernel panicked because it could not mount the rootfs (rootfs
    /// corruption / wrong fs / missing block device).
    KernelCannotMountRoot,
    /// Operator-supplied free-form reason. Serialized verbatim — prefer a
    /// typed variant when one applies.
    Other(String),
}

impl LifecycleReason {
    /// Wire-form string for this reason. Typed variants serialize to
    /// `snake_case`; `Other(s)` returns the inner string verbatim.
    pub fn as_wire_str(&self) -> &str {
        match self {
            LifecycleReason::Oom => "oom",
            LifecycleReason::TtlExceeded => "ttl_exceeded",
            LifecycleReason::VmmCrashed => "vmm_crashed",
            LifecycleReason::BootFailed => "boot_failed",
            LifecycleReason::SignalKilled => "signal_killed",
            LifecycleReason::InitCrashed => "init_crashed",
            LifecycleReason::KernelCannotMountRoot => "kernel_cannot_mount_root",
            LifecycleReason::Other(s) => s.as_str(),
        }
    }
}

impl fmt::Display for LifecycleReason {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.write_str(self.as_wire_str())
    }
}

impl Serialize for LifecycleReason {
    fn serialize<S: Serializer>(&self, serializer: S) -> Result<S::Ok, S::Error> {
        serializer.serialize_str(self.as_wire_str())
    }
}

/// Identity lifecycle step for [`identity_failed_data_v1`].
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
#[serde(rename_all = "lowercase")]
pub enum IdentityFailureOperation {
    Materialize,
    Revoke,
}

/// Tranche-1 seam-freeze G2 provenance pointer.
///
/// Surfaces the parent CloudEvent that *caused* the event carrying this
/// `provenance` block, so downstream auditors (taudit, tencrypt) can walk
/// `artifact → compliance.summary → spec → derivation token` without
/// operator-supplied joins. See `docs/seam-freeze-v1.md` §3 G2.
///
/// Today the supervisor populates this on:
///
/// - `dev.cellos.events.cell.identity.v1.revoked` — revoke is caused by the
///   cell that materialized the identity, so `parent` is the
///   `cell.lifecycle.v1.started` event ID and `parentType` is the started
///   event's CloudEvent type URN.
/// - `dev.cellos.events.cell.export.v2.completed` and
///   `dev.cellos.events.cell.export.v2.failed` — exports are caused by the
///   originating cell run, so `parent` is the started event ID with the
///   matching `parentType`.
///
/// The struct is additive: producers omit it on legacy emissions, and the
/// schemas tolerate its absence so v1/v2 consumers keep parsing legacy
/// events unchanged.
///
/// `parent` is the producing tool's CloudEvent envelope `id` — a stable URN
/// is preferred (`urn:cellos:event:<uuid>`) but a raw UUID is accepted in
/// v1. `parent_type` carries the parent event's CloudEvent `type` so a
/// consumer can route without resolving the parent first.
#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct Provenance {
    /// CloudEvent `id` (or URN) of the parent event that caused this event.
    pub parent: String,
    /// CloudEvent `type` of the parent event (e.g.
    /// `dev.cellos.events.cell.lifecycle.v1.started`). Lets consumers
    /// dispatch on the parent class without dereferencing it.
    pub parent_type: String,
}

/// Tranche-1 seam-freeze G3 subject URN.
///
/// Promotes the CloudEvents `subject` envelope field from a free-form string
/// (`cell:<id>` was the legacy convention) to a typed, validated URN of the
/// form `urn:<tool>:<kind>:<id>`. See `docs/seam-freeze-v1.md` §3 G3 / §4.
///
/// `0ryant-shell` and `tedit` use the URN prefix as a routing key; CellOS
/// emitters use [`cell_subject_urn`] for cell subjects. Other tools mint
/// their own (`urn:tsafe:lease:<id>`, `urn:tencrypt:cert:<id>`, etc.).
///
/// Validation rules (see [`SubjectUrn::parse`]):
///
/// 1. must start with the literal scheme `urn:`;
/// 2. exactly four colon-separated segments — `urn`, `<tool>`, `<kind>`,
///    `<id>` — where `<id>` may itself contain colons;
/// 3. `<tool>`, `<kind>`, `<id>` must each be non-empty;
/// 4. `<tool>` and `<kind>` are restricted to lowercase ASCII alphanumerics
///    and `-` (charset `[a-z0-9-]`);
/// 5. no ASCII control characters and no whitespace anywhere.
///
/// `<id>` is intentionally permissive on charset (so producers can carry
/// existing IDs like ULIDs, UUIDs, or `cell-<host>-<n>`) but still must not
/// contain ASCII control or whitespace.
#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize)]
#[serde(transparent)]
pub struct SubjectUrn(String);

/// Parse-time error returned by [`SubjectUrn::parse`].
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum SubjectUrnError {
    /// Did not start with the literal `urn:` scheme.
    MissingUrnScheme,
    /// Fewer than four colon-separated segments.
    TooFewSegments,
    /// One of `<tool>` / `<kind>` / `<id>` was empty.
    EmptySegment,
    /// `<tool>` or `<kind>` contained a character outside `[a-z0-9-]`.
    InvalidToolOrKindCharset,
    /// Subject contained an ASCII control character or whitespace.
    ControlOrWhitespace,
}

impl std::fmt::Display for SubjectUrnError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            SubjectUrnError::MissingUrnScheme => f.write_str("subject URN must start with `urn:`"),
            SubjectUrnError::TooFewSegments => {
                f.write_str("subject URN must have shape `urn:<tool>:<kind>:<id>`")
            }
            SubjectUrnError::EmptySegment => {
                f.write_str("subject URN tool / kind / id segments must each be non-empty")
            }
            SubjectUrnError::InvalidToolOrKindCharset => {
                f.write_str("subject URN tool and kind must match charset [a-z0-9-]")
            }
            SubjectUrnError::ControlOrWhitespace => {
                f.write_str("subject URN must not contain ASCII control characters or whitespace")
            }
        }
    }
}

impl std::error::Error for SubjectUrnError {}

impl SubjectUrn {
    /// Validate `s` and wrap it as a typed `SubjectUrn`. See the type-level
    /// documentation for the exact rules.
    pub fn parse(s: impl Into<String>) -> Result<Self, SubjectUrnError> {
        let s = s.into();

        // Rule 5: no ASCII control or whitespace anywhere.
        if s.bytes()
            .any(|b| b.is_ascii_control() || (b as char).is_whitespace())
        {
            return Err(SubjectUrnError::ControlOrWhitespace);
        }

        // Rule 1: must start with `urn:`.
        let rest = match s.strip_prefix("urn:") {
            Some(r) => r,
            None => return Err(SubjectUrnError::MissingUrnScheme),
        };

        // Rule 2: split into <tool>:<kind>:<id> with `splitn(3, ':')` so the
        // id portion can itself contain colons (forward-compat for nested
        // identifiers like `urn:cellos:event:<uuid>:<seq>`).
        let mut parts = rest.splitn(3, ':');
        let tool = parts.next().ok_or(SubjectUrnError::TooFewSegments)?;
        let kind = parts.next().ok_or(SubjectUrnError::TooFewSegments)?;
        let id = parts.next().ok_or(SubjectUrnError::TooFewSegments)?;

        // Rule 3: non-empty tool/kind/id.
        if tool.is_empty() || kind.is_empty() || id.is_empty() {
            return Err(SubjectUrnError::EmptySegment);
        }

        // Rule 4: tool/kind charset [a-z0-9-].
        let ok_segment = |seg: &str| {
            seg.bytes()
                .all(|b| matches!(b, b'a'..=b'z' | b'0'..=b'9' | b'-'))
        };
        if !ok_segment(tool) || !ok_segment(kind) {
            return Err(SubjectUrnError::InvalidToolOrKindCharset);
        }

        Ok(SubjectUrn(s))
    }

    /// Borrow the validated URN as a `&str`.
    pub fn as_str(&self) -> &str {
        &self.0
    }

    /// Consume the wrapper and return the owned string.
    pub fn into_inner(self) -> String {
        self.0
    }
}

impl AsRef<str> for SubjectUrn {
    fn as_ref(&self) -> &str {
        &self.0
    }
}

impl std::fmt::Display for SubjectUrn {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.write_str(&self.0)
    }
}

/// Build the canonical CellOS cell subject URN (`urn:cellos:cell:<cell_id>`).
///
/// The single helper means the literal prefix `urn:cellos:cell:` lives in
/// exactly one place; the `scripts/audit/subject-urn-check.sh` CI gate
/// enforces that no other site hand-crafts the same shape. Callers should
/// prefer this helper over `format!("urn:cellos:cell:{cell_id}")`.
///
/// Returns `Err(SubjectUrnError::EmptySegment)` if `cell_id` is empty and
/// surfaces charset / whitespace errors from [`SubjectUrn::parse`] for IDs
/// that contain ASCII control / whitespace characters.
pub fn cell_subject_urn(cell_id: &str) -> Result<SubjectUrn, SubjectUrnError> {
    SubjectUrn::parse(format!("urn:cellos:cell:{cell_id}"))
}

/// `data` for event type `dev.cellos.events.cell.lifecycle.v1.started`.
///
/// Schema: `contracts/schemas/cell-lifecycle-started-data-v1.schema.json`.
///
/// Authority derivation fields (L5-05): when the spec carries an
/// `authority.authorityDerivation` token, the supervisor verifies the proof
/// before this event is emitted and surfaces the outcome here so taudit can
/// answer "prove every prod cell derived authority from role X" without
/// re-running the verification.
///
/// - `derivation_verified`: `Some(true)` when the proof verified, `Some(false)`
///   when verification failed (non-fatal — the run still proceeds), `None`
///   when the spec carried no derivation token at all.
/// - `role_root`: the role root identifier from the derivation token (when
///   the token was present), regardless of whether verification succeeded.
/// - `parent_run_id`: optional lineage attribution from the derivation token.
///
/// All three fields are emitted into the `data` JSON only when `Some`, so
/// specs without a derivation token produce the same payload as before this
/// change (backward-compatible with the v1 schema).
///
/// **A2-03 — per-tenant event stream isolation**: when
/// `spec.correlation.tenantId` is `Some(_)`, the value is mirrored into a
/// top-level `tenantId` field on the event `data` (in addition to staying
/// embedded inside the `correlation` block for joins). The supervisor's
/// JetStream sink reads this field at publish time to substitute
/// `{tenantId}` in the configured subject template. When the tenant is
/// `None`, the top-level `tenantId` is OMITTED entirely so the wire format
/// is byte-identical to pre-A2-03 builds for single-tenant operators.
///
/// **FC-08**: when the host backend has a verified artifact manifest
/// (currently the Firecracker backend), the supervisor surfaces the on-disk
/// SHA256 digests of the boot artifacts on the started event so taudit can
/// answer "which kernel/rootfs/firecracker bytes did this run boot?" without
/// querying any backend-side state. The three digest fields are independently
/// optional — backends without a manifest path (stub, host-cellos) pass
/// `None` for all three; the Firecracker backend passes `Some(hex)` for
/// kernel and rootfs (always verified) and `Some(hex)` for firecracker only
/// when the manifest also declares the firecracker binary digest. Each is
/// emitted into the JSON only when `Some`, preserving the v1 schema's
/// backward-compat semantics.
#[allow(clippy::too_many_arguments)]
pub fn lifecycle_started_data_v1(
    spec: &ExecutionCellSpec,
    cell_id: &str,
    run_id: Option<&str>,
    derivation_verified: Option<bool>,
    role_root: Option<&str>,
    parent_run_id: Option<&str>,
    spec_hash: Option<&str>,
    kernel_digest_sha256: Option<&str>,
    rootfs_digest_sha256: Option<&str>,
    firecracker_digest_sha256: Option<&str>,
) -> Result<Value, serde_json::Error> {
    let mut m = Map::new();
    m.insert("cellId".to_string(), json!(cell_id));
    m.insert("specId".to_string(), json!(&spec.id));
    m.insert("ttlSeconds".to_string(), json!(spec.lifetime.ttl_seconds));
    if let Some(r) = run_id {
        m.insert("runId".to_string(), json!(r));
    }
    if let Some(verified) = derivation_verified {
        m.insert("derivationVerified".to_string(), json!(verified));
    }
    if let Some(role) = role_root {
        m.insert("roleRoot".to_string(), json!(role));
    }
    if let Some(parent) = parent_run_id {
        m.insert("parentRunId".to_string(), json!(parent));
    }
    if let Some(hash) = spec_hash {
        m.insert("specHash".to_string(), json!(hash));
    }
    if let Some(d) = kernel_digest_sha256 {
        m.insert("kernelDigestSha256".to_string(), json!(d));
    }
    if let Some(d) = rootfs_digest_sha256 {
        m.insert("rootfsDigestSha256".to_string(), json!(d));
    }
    if let Some(d) = firecracker_digest_sha256 {
        m.insert("firecrackerDigestSha256".to_string(), json!(d));
    }
    if let Some(placement) = &spec.placement {
        let mut placement_map = Map::new();
        if let Some(pool_id) = &placement.pool_id {
            placement_map.insert("poolId".to_string(), json!(pool_id));
        }
        if let Some(namespace) = &placement.kubernetes_namespace {
            placement_map.insert("kubernetesNamespace".to_string(), json!(namespace));
        }
        if let Some(queue_name) = &placement.queue_name {
            placement_map.insert("queueName".to_string(), json!(queue_name));
        }
        if !placement_map.is_empty() {
            m.insert("placement".to_string(), Value::Object(placement_map));
        }
    }
    if let Some(c) = &spec.correlation {
        if let Some(tid) = &c.tenant_id {
            m.insert("tenantId".to_string(), json!(tid));
        }
        m.insert("correlation".to_string(), serde_json::to_value(c)?);
    }
    Ok(Value::Object(m))
}

/// `data` for event type `dev.cellos.events.cell.lifecycle.v1.destroyed`.
///
/// Schema: `contracts/schemas/cell-lifecycle-destroyed-data-v1.schema.json`.
///
/// `terminal_state` is `None` for backends that do not own workload execution
/// (host-side spawn path) and `Some(...)` for the in-VM bridge path. Auditors
/// keying on this field can distinguish "supervisor read an authenticated exit
/// code" from "supervisor gave up and SIGKILLed the VMM" — see
/// [`LifecycleTerminalState`] for the distinction. The field is emitted into
/// the JSON only when set, preserving backward-compatibility with v1 consumers.
///
/// F5 (D5 destruction-evidence integration):
///
/// - `evidence_bundle_ref`: URN of the `evidence_bundle` artifact (F1b)
///   that aggregates this run's lifecycle, host-side series, and guest
///   declarations. Auditors reading `cell.destroyed` cold MUST be able to
///   walk this URN to the bundle. See `docs/adr/0006-...` and F1.
/// - `residue_class`: final residue classification from
///   `docs/destruction-semantics.md` (`none` / `documented_exception`).
///
/// Both fields are additive: emitted only when `Some(...)` so existing v1
/// consumers parse the legacy payload unchanged. The supervisor passes
/// `None, None` until F1b wires the evidence-bundle aggregator into the
/// teardown path.
///
/// **A2-03**: like `lifecycle_started_data_v1`, this constructor mirrors
/// `spec.correlation.tenantId` into a top-level `tenantId` field on
/// `data` when set, and omits it entirely otherwise.
///
/// **FC-23 invariant**: this constructor intentionally takes no `exit_code`
/// parameter. The destroyed payload never carries an `exitCode` field. The
/// authenticated workload exit code (if any) is reported on the separate
/// `cell.command.v1.completed` event via [`command_completed_data_v1`], which
/// is only emitted on the in-VM-bridge clean-exit path. Forced terminations
/// (SIGKILL fallback, vsock channel closed) therefore cannot surface an
/// `exitCode` on this event by construction — any synthetic code such as
/// `-1` or `137` stays internal to the supervisor and never reaches auditors.
/// See `Plans/firecracker-release-readiness.md` line 70.
#[allow(clippy::too_many_arguments)]
pub fn lifecycle_destroyed_data_v1(
    spec: &ExecutionCellSpec,
    cell_id: &str,
    run_id: Option<&str>,
    outcome: LifecycleDestroyOutcome,
    reason: Option<&str>,
    terminal_state: Option<LifecycleTerminalState>,
    evidence_bundle_ref: Option<&SubjectUrn>,
    residue_class: Option<ResidueClass>,
) -> Result<Value, serde_json::Error> {
    let mut m = Map::new();
    m.insert("cellId".to_string(), json!(cell_id));
    m.insert("specId".to_string(), json!(&spec.id));
    m.insert("ttlSeconds".to_string(), json!(spec.lifetime.ttl_seconds));
    m.insert("outcome".to_string(), serde_json::to_value(outcome)?);
    if let Some(r) = run_id {
        m.insert("runId".to_string(), json!(r));
    }
    if let Some(c) = &spec.correlation {
        if let Some(tid) = &c.tenant_id {
            m.insert("tenantId".to_string(), json!(tid));
        }
        m.insert("correlation".to_string(), serde_json::to_value(c)?);
    }
    if let Some(s) = reason {
        m.insert("reason".to_string(), json!(s));
    }
    if let Some(ts) = terminal_state {
        m.insert("terminalState".to_string(), serde_json::to_value(ts)?);
    }
    if let Some(urn) = evidence_bundle_ref {
        m.insert("evidenceBundleRef".to_string(), json!(urn));
    }
    if let Some(rc) = residue_class {
        m.insert("residueClass".to_string(), serde_json::to_value(rc)?);
    }
    Ok(Value::Object(m))
}

/// CloudEvent `type` URN for [`manifest_failed_data_v1`].
///
/// Emitted when pre-boot artifact digest verification fails — the on-disk
/// artifact does not match the digest declared in the host manifest. The
/// emission point lives in the host backend (see
/// `crates/cellos-host-firecracker/src/lib.rs::verify_artifacts`); this
/// constructor only shapes the `data` payload.
pub const LIFECYCLE_MANIFEST_FAILED_TYPE: &str =
    "dev.cellos.events.cell.lifecycle.v1.manifest-failed";

/// `data` for event type `dev.cellos.events.cell.lifecycle.v1.manifest-failed`.
///
/// Surfaces a manifest verification failure: the artifact bound to `role`
/// (e.g. `"kernel"`, `"rootfs"`, `"firecracker"`) hashed to
/// `actual_sha256` on disk, but the declared digest in `manifest_path` was
/// `expected_sha256`. Both digests are emitted as lowercase hex (no
/// `sha256:` prefix), matching the rest of CellOS's audit surface, so
/// downstream taudit consumers can diff the two strings byte-for-byte.
///
/// FC-51 (`Plans/firecracker-release-readiness.md`): integration test
/// replaces the on-disk kernel with a mismatched file and asserts this
/// event is emitted with the `kernel` role and both digests.
pub fn manifest_failed_data_v1(
    role: &str,
    expected_sha256: &str,
    actual_sha256: &str,
    manifest_path: &str,
) -> Result<Value, serde_json::Error> {
    let mut m = Map::new();
    m.insert("role".to_string(), json!(role));
    m.insert("expectedSha256".to_string(), json!(expected_sha256));
    m.insert("actualSha256".to_string(), json!(actual_sha256));
    m.insert("manifestPath".to_string(), json!(manifest_path));
    Ok(Value::Object(m))
}

/// FC-50 typed-reason variant of [`lifecycle_destroyed_data_v1`].
///
/// Identical wire format to the string-based constructor — `reason` is
/// serialized through [`LifecycleReason::as_wire_str`] and threaded into the
/// existing builder via the `Option<&str>` path, so consumers see the same
/// JSON. The typed surface is the preferred entry point for new callers
/// (FC-23, FC-52, FC-59, FC-60, FC-61, FC-63, FC-72 gap-markers); the
/// string-based constructor remains for back-compat until all call sites
/// migrate.
///
/// Adds two F5 destruction-evidence fields not on the legacy constructor:
///
/// - `evidence_bundle_ref`: pointer (URN or CloudEvent id) to the per-cell
///   `evidence_bundle` produced by Phase F (see ADR-0006). Emitted as
///   `evidenceBundleRef` only when `Some`.
/// - `residue_class`: destruction-semantics residue class string (per
///   `docs/destruction-semantics.md`). Emitted as `residueClass` only when
///   `Some`.
///
/// Both are additive and absent from the legacy v1 payload — consumers that
/// do not understand them must ignore unknown fields per the v1 schema's
/// `additionalProperties` posture.
#[allow(clippy::too_many_arguments)]
pub fn lifecycle_destroyed_data_v1_typed(
    spec: &ExecutionCellSpec,
    cell_id: &str,
    run_id: Option<&str>,
    outcome: LifecycleDestroyOutcome,
    reason: Option<LifecycleReason>,
    terminal_state: Option<LifecycleTerminalState>,
    evidence_bundle_ref: Option<&SubjectUrn>,
    residue_class: Option<ResidueClass>,
) -> Result<Value, serde_json::Error> {
    let reason_str = reason.as_ref().map(|r| r.as_wire_str());
    lifecycle_destroyed_data_v1(
        spec,
        cell_id,
        run_id,
        outcome,
        reason_str,
        terminal_state,
        evidence_bundle_ref,
        residue_class,
    )
}

/// `data` for event type `dev.cellos.events.cell.identity.v1.materialized`.
///
/// Schema: `contracts/schemas/cell-identity-materialized-data-v1.schema.json`.
pub fn identity_materialized_data_v1(
    spec: &ExecutionCellSpec,
    cell_id: &str,
    run_id: Option<&str>,
    identity: &WorkloadIdentity,
) -> Result<Value, serde_json::Error> {
    let mut m = Map::new();
    m.insert("cellId".to_string(), json!(cell_id));
    m.insert("specId".to_string(), json!(&spec.id));
    m.insert("identity".to_string(), serde_json::to_value(identity)?);
    if let Some(r) = run_id {
        m.insert("runId".to_string(), json!(r));
    }
    if let Some(c) = &spec.correlation {
        m.insert("correlation".to_string(), serde_json::to_value(c)?);
    }
    Ok(Value::Object(m))
}

/// Witness fields recording what cellos actually provisioned for a cloud
/// identity (ADR-0026 §Decision 4). These are the *receipt* of a provisioning
/// outcome, not an enforcement claim: cellos requests and records the scoped
/// identity while AWS IAM / Azure RBAC remains the server-side enforcer. The
/// provisioner itself (AWS `AssumeRoleWithWebIdentity` trust setup / Azure RBAC
/// assignment) is an L2 host-backend deliverable; this struct only carries the
/// witness so the provisioning outcome is auditable once that provisioner lands.
///
/// Every field is `Option` and omitted from the wire when `None`, so the
/// witnessed payload is additive over the base `identity_materialized_data_v1`
/// shape — a `FederatedOidc` materialization carries none of these.
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct ProvisioningWitness<'a> {
    /// The AWS role ARN cellos actually assumed for an `AwsAssumeRole` identity.
    /// Emitted as `provisionedRoleArn`.
    pub provisioned_role_arn: Option<&'a str>,
    /// Lowercase-hex digest of the STS session policy bound to the assumed role
    /// (no `sha256:` prefix, matching the rest of CellOS's audit surface).
    /// Emitted as `sessionPolicyDigest`.
    pub session_policy_digest: Option<&'a str>,
    /// The Azure AD object id of the managed identity cellos bound for an
    /// `AzureManagedIdentity` identity. Emitted as `azureObjectId`.
    pub azure_object_id: Option<&'a str>,
    /// JWT id (`jti`) of the provisioned credential, witnessing the exact token
    /// minted so a revocation can be correlated. Emitted as `credentialJti`.
    pub credential_jti: Option<&'a str>,
}

/// ADR-0026 provisioning-receipt variant of [`identity_materialized_data_v1`].
///
/// Identical wire format to the base constructor for the shared fields, plus
/// the [`ProvisioningWitness`] fields recording what cellos actually
/// provisioned for a cloud identity. Each witness field is emitted only when
/// `Some`, so a witness with no populated fields produces a payload
/// byte-identical to the base constructor — the extension is additive and
/// absent from the legacy v1 payload, which consumers ignore per the v1
/// schema's `additionalProperties` posture.
///
/// cellos does **not** enforce the provisioned scope; these fields are the
/// audit witness of a host-backend provisioning outcome (ADR-0026 §Decision 4).
pub fn identity_materialized_data_v1_provisioned(
    spec: &ExecutionCellSpec,
    cell_id: &str,
    run_id: Option<&str>,
    identity: &WorkloadIdentity,
    witness: &ProvisioningWitness<'_>,
) -> Result<Value, serde_json::Error> {
    let mut data = identity_materialized_data_v1(spec, cell_id, run_id, identity)?;
    let m = data
        .as_object_mut()
        .expect("identity_materialized_data_v1 returns a JSON object");
    if let Some(arn) = witness.provisioned_role_arn {
        m.insert("provisionedRoleArn".to_string(), json!(arn));
    }
    if let Some(digest) = witness.session_policy_digest {
        m.insert("sessionPolicyDigest".to_string(), json!(digest));
    }
    if let Some(oid) = witness.azure_object_id {
        m.insert("azureObjectId".to_string(), json!(oid));
    }
    if let Some(jti) = witness.credential_jti {
        m.insert("credentialJti".to_string(), json!(jti));
    }
    Ok(data)
}

/// `data` for event type `dev.cellos.events.cell.identity.v1.revoked`.
///
/// Schema: `contracts/schemas/cell-identity-revoked-data-v1.schema.json`.
///
/// `provenance` (Tranche-1 seam-freeze G2) — when set, points at the parent
/// event that caused this revocation. The supervisor populates it with the
/// `cell.lifecycle.v1.started` envelope so taudit can stitch revoke →
/// lifecycle → spec without operator joins. Omitted from the JSON when
/// `None`, preserving the v1 schema's backward-compat semantics.
pub fn identity_revoked_data_v1(
    spec: &ExecutionCellSpec,
    cell_id: &str,
    run_id: Option<&str>,
    identity: &WorkloadIdentity,
    reason: Option<&str>,
    provenance: Option<&Provenance>,
) -> Result<Value, serde_json::Error> {
    let mut m = Map::new();
    m.insert("cellId".to_string(), json!(cell_id));
    m.insert("specId".to_string(), json!(&spec.id));
    m.insert("identity".to_string(), serde_json::to_value(identity)?);
    if let Some(r) = run_id {
        m.insert("runId".to_string(), json!(r));
    }
    if let Some(c) = &spec.correlation {
        m.insert("correlation".to_string(), serde_json::to_value(c)?);
    }
    if let Some(s) = reason {
        m.insert("reason".to_string(), json!(s));
    }
    if let Some(p) = provenance {
        m.insert("provenance".to_string(), serde_json::to_value(p)?);
    }
    Ok(Value::Object(m))
}

/// `data` for event type `dev.cellos.events.cell.identity.v1.failed`.
///
/// Schema: `contracts/schemas/cell-identity-failed-data-v1.schema.json`.
pub fn identity_failed_data_v1(
    spec: &ExecutionCellSpec,
    cell_id: &str,
    run_id: Option<&str>,
    identity: &WorkloadIdentity,
    operation: IdentityFailureOperation,
    reason: &str,
) -> Result<Value, serde_json::Error> {
    let mut m = Map::new();
    m.insert("cellId".to_string(), json!(cell_id));
    m.insert("specId".to_string(), json!(&spec.id));
    m.insert("identity".to_string(), serde_json::to_value(identity)?);
    m.insert("operation".to_string(), serde_json::to_value(operation)?);
    m.insert("reason".to_string(), json!(reason));
    if let Some(r) = run_id {
        m.insert("runId".to_string(), json!(r));
    }
    if let Some(c) = &spec.correlation {
        m.insert("correlation".to_string(), serde_json::to_value(c)?);
    }
    Ok(Value::Object(m))
}

/// `data` for event type `dev.cellos.events.cell.command.v1.completed`.
///
/// Schema: `contracts/schemas/cell-command-completed-data-v1.schema.json`.
pub fn command_completed_data_v1(
    spec: &ExecutionCellSpec,
    cell_id: &str,
    run_id: Option<&str>,
    argv: &[String],
    exit_code: i32,
    duration_ms: u64,
    spawn_error: Option<&str>,
) -> Result<Value, serde_json::Error> {
    let mut m = Map::new();
    m.insert("cellId".to_string(), json!(cell_id));
    m.insert("specId".to_string(), json!(&spec.id));
    m.insert("exitCode".to_string(), json!(exit_code));
    m.insert("durationMs".to_string(), json!(duration_ms));
    m.insert("argv".to_string(), json!(argv));
    if let Some(r) = run_id {
        m.insert("runId".to_string(), json!(r));
    }
    if let Some(c) = &spec.correlation {
        m.insert("correlation".to_string(), serde_json::to_value(c)?);
    }
    if let Some(s) = spawn_error {
        m.insert("spawnError".to_string(), json!(s));
    }
    Ok(Value::Object(m))
}

/// `dev.cellos.events.cell.observability.v1.network_scope`
///
/// Schema: `contracts/schemas/cell-observability-network-scope-v1.schema.json`.
pub fn observability_network_scope_data_v1(
    spec: &ExecutionCellSpec,
    cell_id: &str,
    run_id: Option<&str>,
    egress_rule_count: usize,
    has_opaque_network_authority: bool,
) -> Result<Value, serde_json::Error> {
    let mut m = Map::new();
    m.insert("cellId".to_string(), json!(cell_id));
    m.insert("specId".to_string(), json!(&spec.id));
    m.insert("egressRuleCount".to_string(), json!(egress_rule_count));
    m.insert(
        "hasOpaqueNetworkAuthority".to_string(),
        json!(has_opaque_network_authority),
    );
    if let Some(r) = run_id {
        m.insert("runId".to_string(), json!(r));
    }
    if let Some(c) = &spec.correlation {
        m.insert("correlation".to_string(), serde_json::to_value(c)?);
    }
    Ok(Value::Object(m))
}

/// `dev.cellos.events.cell.observability.v1.process_spawned`
///
/// Schema: `contracts/schemas/cell-observability-process-spawned-v1.schema.json`.
pub fn observability_process_spawned_data_v1(
    spec: &ExecutionCellSpec,
    cell_id: &str,
    run_id: Option<&str>,
    program: &str,
    argc: usize,
) -> Result<Value, serde_json::Error> {
    let mut m = Map::new();
    m.insert("cellId".to_string(), json!(cell_id));
    m.insert("specId".to_string(), json!(&spec.id));
    m.insert("program".to_string(), json!(program));
    m.insert("argc".to_string(), json!(argc));
    if let Some(r) = run_id {
        m.insert("runId".to_string(), json!(r));
    }
    if let Some(c) = &spec.correlation {
        m.insert("correlation".to_string(), serde_json::to_value(c)?);
    }
    Ok(Value::Object(m))
}

/// `dev.cellos.events.cell.observability.v1.fs_touch`
///
/// Schema: `contracts/schemas/cell-observability-fs-touch-v1.schema.json`.
pub fn observability_fs_touch_export_data_v1(
    spec: &ExecutionCellSpec,
    cell_id: &str,
    run_id: Option<&str>,
    source_path: &str,
    artifact_name: &str,
) -> Result<Value, serde_json::Error> {
    let mut m = Map::new();
    m.insert("cellId".to_string(), json!(cell_id));
    m.insert("specId".to_string(), json!(&spec.id));
    m.insert("purpose".to_string(), json!("export"));
    m.insert("sourcePath".to_string(), json!(source_path));
    m.insert("artifactName".to_string(), json!(artifact_name));
    if let Some(r) = run_id {
        m.insert("runId".to_string(), json!(r));
    }
    if let Some(c) = &spec.correlation {
        m.insert("correlation".to_string(), serde_json::to_value(c)?);
    }
    Ok(Value::Object(m))
}

/// `dev.cellos.events.cell.export.v1.completed`
///
/// Schema: `contracts/schemas/cell-export-completed-v1.schema.json`.
pub fn export_completed_data_v1(
    spec: &ExecutionCellSpec,
    cell_id: &str,
    run_id: Option<&str>,
    artifact_name: &str,
    bytes_written: u64,
    destination_relative: &str,
) -> Result<Value, serde_json::Error> {
    let mut m = Map::new();
    m.insert("cellId".to_string(), json!(cell_id));
    m.insert("specId".to_string(), json!(&spec.id));
    m.insert("artifactName".to_string(), json!(artifact_name));
    m.insert("bytesWritten".to_string(), json!(bytes_written));
    m.insert(
        "destinationRelative".to_string(),
        json!(destination_relative),
    );
    if let Some(r) = run_id {
        m.insert("runId".to_string(), json!(r));
    }
    if let Some(c) = &spec.correlation {
        m.insert("correlation".to_string(), serde_json::to_value(c)?);
    }
    Ok(Value::Object(m))
}

/// `dev.cellos.events.cell.export.v2.completed`
///
/// Schema: `contracts/schemas/cell-export-completed-v2.schema.json`.
///
/// `provenance` (Tranche-1 seam-freeze G2) — when set, points at the parent
/// event that caused this export receipt. The supervisor populates it with
/// the `cell.lifecycle.v1.started` envelope of the originating cell run so
/// downstream consumers (tencrypt envelope promotion, taudit graph build)
/// can walk artifact → lifecycle → spec without operator joins. Omitted
/// from the JSON when `None` to preserve v2 schema backward-compat.
pub fn export_completed_data_v2(
    spec: &ExecutionCellSpec,
    cell_id: &str,
    run_id: Option<&str>,
    artifact_name: &str,
    receipt: &ExportReceipt,
    provenance: Option<&Provenance>,
) -> Result<Value, serde_json::Error> {
    let mut m = Map::new();
    m.insert("cellId".to_string(), json!(cell_id));
    m.insert("specId".to_string(), json!(&spec.id));
    m.insert("artifactName".to_string(), json!(artifact_name));
    m.insert("receipt".to_string(), serde_json::to_value(receipt)?);
    if let Some(r) = run_id {
        m.insert("runId".to_string(), json!(r));
    }
    if let Some(c) = &spec.correlation {
        m.insert("correlation".to_string(), serde_json::to_value(c)?);
    }
    if let Some(p) = provenance {
        m.insert("provenance".to_string(), serde_json::to_value(p)?);
    }
    Ok(Value::Object(m))
}

/// `dev.cellos.events.cell.export.v2.failed`
///
/// Schema: `contracts/schemas/cell-export-failed-v2.schema.json`.
///
/// `provenance` (Tranche-1 seam-freeze G2) — when set, points at the parent
/// event that caused this export attempt. The supervisor populates it with
/// the `cell.lifecycle.v1.started` envelope of the originating cell run so
/// downstream consumers (tencrypt envelope promotion, taudit graph build)
/// can walk failed artifact → lifecycle → spec without operator joins.
/// Omitted from the JSON when `None` to preserve v2 schema backward-compat.
#[allow(clippy::too_many_arguments)] // mirrors CloudEvent payload fields
pub fn export_failed_data_v2(
    spec: &ExecutionCellSpec,
    cell_id: &str,
    run_id: Option<&str>,
    artifact_name: &str,
    target_kind: crate::ExportReceiptTargetKind,
    target_name: Option<&str>,
    destination: Option<&str>,
    reason: &str,
    provenance: Option<&Provenance>,
) -> Result<Value, serde_json::Error> {
    let mut m = Map::new();
    m.insert("cellId".to_string(), json!(cell_id));
    m.insert("specId".to_string(), json!(&spec.id));
    m.insert("artifactName".to_string(), json!(artifact_name));
    m.insert("targetKind".to_string(), serde_json::to_value(target_kind)?);
    m.insert("reason".to_string(), json!(reason));
    if let Some(name) = target_name {
        m.insert("targetName".to_string(), json!(name));
    }
    if let Some(dest) = destination {
        m.insert("destination".to_string(), json!(dest));
    }
    if let Some(r) = run_id {
        m.insert("runId".to_string(), json!(r));
    }
    if let Some(c) = &spec.correlation {
        m.insert("correlation".to_string(), serde_json::to_value(c)?);
    }
    if let Some(p) = provenance {
        m.insert("provenance".to_string(), serde_json::to_value(p)?);
    }
    Ok(Value::Object(m))
}

/// `dev.cellos.events.cell.observability.v1.network_policy`
///
/// Emitted when `CELLOS_SUBPROCESS_UNSHARE` includes the `net` flag, announcing the
/// network isolation mode and declared egress rules for this cell run. The isolation is
/// enforced by the Linux network namespace (no external connectivity without explicit routing).
/// Optional nftables rules may further filter allowed egress within the namespace.
pub fn observability_network_policy_data_v1(
    spec: &ExecutionCellSpec,
    cell_id: &str,
    run_id: Option<&str>,
    isolation_mode: &str,
    egress_rules: &[crate::EgressRule],
) -> Result<Value, serde_json::Error> {
    let mut m = Map::new();
    m.insert("cellId".to_string(), json!(cell_id));
    m.insert("specId".to_string(), json!(&spec.id));
    m.insert("isolationMode".to_string(), json!(isolation_mode));
    m.insert("declaredEgressCount".to_string(), json!(egress_rules.len()));
    m.insert(
        "declaredEgress".to_string(),
        serde_json::to_value(egress_rules)?,
    );
    if let Some(r) = run_id {
        m.insert("runId".to_string(), json!(r));
    }
    if let Some(c) = &spec.correlation {
        m.insert("correlation".to_string(), serde_json::to_value(c)?);
    }
    Ok(Value::Object(m))
}

/// `dev.cellos.events.cell.observability.v1.network_enforcement`
///
/// Emitted after a `spec.run` subprocess exits when the Linux path used `CLONE_NEWNET`, summarizing
/// whether supplementary nftables rules were applied and the command exit code (L2-04).
///
/// Schema: `contracts/schemas/cell-observability-network-enforcement-v1.schema.json`.
#[allow(clippy::too_many_arguments)]
pub fn observability_network_enforcement_data_v1(
    spec: &ExecutionCellSpec,
    cell_id: &str,
    run_id: Option<&str>,
    nft_rules_applied: bool,
    declared_egress_rule_count: usize,
    command_exit_code: i32,
    spawn_error: Option<&str>,
) -> Result<Value, serde_json::Error> {
    let supplementary = nft_rules_applied && declared_egress_rule_count > 0;
    let mut m = Map::new();
    m.insert("cellId".to_string(), json!(cell_id));
    m.insert("specId".to_string(), json!(&spec.id));
    m.insert("isolationMode".to_string(), json!("clone_newnet"));
    m.insert("nftRulesApplied".to_string(), json!(nft_rules_applied));
    m.insert(
        "declaredEgressRuleCount".to_string(),
        json!(declared_egress_rule_count),
    );
    m.insert(
        "supplementaryEgressFilterActive".to_string(),
        json!(supplementary),
    );
    m.insert("commandExitCode".to_string(), json!(command_exit_code));
    if let Some(r) = run_id {
        m.insert("runId".to_string(), json!(r));
    }
    if let Some(s) = spawn_error {
        m.insert("spawnError".to_string(), json!(s));
    }
    if let Some(c) = &spec.correlation {
        m.insert("correlation".to_string(), serde_json::to_value(c)?);
    }
    Ok(Value::Object(m))
}

/// Default `keysetId` when the operator has not published a [`trust-keyset-v1`] document yet.
///
/// [`trust-keyset-v1`]: https://cellos.dev/schemas/trust-keyset-v1.schema.json
pub const TRUST_PLANE_BUILTIN_KEYSET_ID: &str = "cellos:builtin-v0";

/// Default `issuerKid` for resolver-style observability when no published keyset is wired.
pub const TRUST_PLANE_BUILTIN_RESOLVER_KID: &str = "cellos-local-resolve-v0";

/// Default `issuerKid` for L7 gate observability when no published keyset is wired.
pub const TRUST_PLANE_BUILTIN_L7_KID: &str = "cellos-local-l7-v0";

/// Synthetic FQDN for aggregate declared-egress materialization events (`dns_target_set`).
pub const TRUST_PLANE_AGGREGATE_EGRESS_FQDN: &str = "declared-egress.trust.cellos.internal";

/// `dev.cellos.events.cell.observability.v1.dns_resolution`
///
/// Schema: `contracts/schemas/cell-observability-dns-resolution-v1.schema.json`.
#[allow(clippy::too_many_arguments)]
pub fn observability_dns_resolution_data_v1(
    spec: &ExecutionCellSpec,
    cell_id: &str,
    run_id: Option<&str>,
    fqdn: &str,
    resolved_at: &str,
    targets: &[(&str, &str, Option<u16>)],
    ttl_seconds: i64,
    policy_digest: &str,
    keyset_id: &str,
    issuer_kid: &str,
    receipt_id: Option<&str>,
) -> Result<Value, serde_json::Error> {
    let mut rows = Vec::with_capacity(targets.len());
    for (addr, family, port) in targets {
        let mut row = Map::new();
        row.insert("address".to_string(), json!(addr));
        row.insert("family".to_string(), json!(family));
        if let Some(p) = port {
            row.insert("port".to_string(), json!(p));
        }
        rows.push(Value::Object(row));
    }
    let mut m = Map::new();
    m.insert("cellId".to_string(), json!(cell_id));
    m.insert("specId".to_string(), json!(&spec.id));
    if let Some(r) = run_id {
        m.insert("runId".to_string(), json!(r));
    }
    if let Some(rid) = receipt_id {
        m.insert("receiptId".to_string(), json!(rid));
    }
    m.insert("fqdn".to_string(), json!(fqdn));
    m.insert("resolvedAt".to_string(), json!(resolved_at));
    m.insert("targets".to_string(), Value::Array(rows));
    m.insert("ttlSeconds".to_string(), json!(ttl_seconds));
    m.insert("policyDigest".to_string(), json!(policy_digest));
    m.insert("keysetId".to_string(), json!(keyset_id));
    m.insert("issuerKid".to_string(), json!(issuer_kid));
    if let Some(c) = &spec.correlation {
        m.insert("correlation".to_string(), serde_json::to_value(c)?);
    }
    Ok(Value::Object(m))
}

/// `dev.cellos.events.cell.observability.v1.dns_target_set`
///
/// Schema: `contracts/schemas/cell-observability-dns-target-set-v1.schema.json`.
#[allow(clippy::too_many_arguments)]
pub fn observability_dns_target_set_data_v1(
    spec: &ExecutionCellSpec,
    cell_id: &str,
    run_id: Option<&str>,
    fqdn: &str,
    previous_digest: &str,
    current_digest: &str,
    reason: &str,
    updated_at: &str,
    keyset_id: &str,
    issuer_kid: &str,
) -> Result<Value, serde_json::Error> {
    let mut m = Map::new();
    m.insert("cellId".to_string(), json!(cell_id));
    m.insert("specId".to_string(), json!(&spec.id));
    if let Some(r) = run_id {
        m.insert("runId".to_string(), json!(r));
    }
    m.insert("fqdn".to_string(), json!(fqdn));
    m.insert("previousDigest".to_string(), json!(previous_digest));
    m.insert("currentDigest".to_string(), json!(current_digest));
    m.insert("reason".to_string(), json!(reason));
    m.insert("updatedAt".to_string(), json!(updated_at));
    m.insert("keysetId".to_string(), json!(keyset_id));
    m.insert("issuerKid".to_string(), json!(issuer_kid));
    if let Some(c) = &spec.correlation {
        m.insert("correlation".to_string(), serde_json::to_value(c)?);
    }
    Ok(Value::Object(m))
}

/// `dev.cellos.events.cell.observability.v1.dns_authority_drift`
///
/// Schema: `contracts/schemas/cell-observability-dns-authority-drift-v1.schema.json`.
///
/// Emitted by the SEC-21 host-controlled resolver refresh loop when the
/// resolved target set for a declared FQDN differs from the prior observation.
/// All fields mirror [`DnsAuthorityDrift`] one-for-one; this builder exists so
/// supervisor wiring can emit the payload without re-deriving the canonical
/// JSON shape from the struct.
#[allow(clippy::too_many_arguments)]
pub fn dns_authority_drift_data_v1(drift: &DnsAuthorityDrift) -> Result<Value, serde_json::Error> {
    serde_json::to_value(drift)
}

/// CloudEvent envelope constructor for `dns_authority_drift` (SEC-21).
///
/// Mirrors the supervisor-local `cloud_event(...)` helper but lives in
/// `cellos-core` so library callers (resolver refresh module, taudit) can
/// build the envelope without taking a supervisor-internal dependency.
///
/// `source` should identify the publisher (e.g. `"cellos-supervisor"` or
/// `"cellos-resolver-refresh"`). `time` is the RFC3339 emission timestamp;
/// independent of `drift.observed_at`, which records when the *drift* was
/// observed (the two are typically equal but conceptually distinct).
pub fn cloud_event_v1_dns_authority_drift(
    source: &str,
    time: &str,
    drift: &DnsAuthorityDrift,
) -> Result<CloudEventV1, serde_json::Error> {
    Ok(CloudEventV1 {
        specversion: "1.0".into(),
        id: uuid::Uuid::new_v4().to_string(),
        source: source.to_string(),
        ty: "dev.cellos.events.cell.observability.v1.dns_authority_drift".into(),
        datacontenttype: Some("application/json".into()),
        data: Some(dns_authority_drift_data_v1(drift)?),
        time: Some(time.to_string()),
        traceparent: None,
        cex: None,
    })
}

/// `dev.cellos.events.cell.observability.v1.dns_authority_rebind_threshold`
/// (SEC-21 Phase 3e).
///
/// Schema: `contracts/schemas/cell-observability-dns-authority-rebind-threshold-v1.schema.json`.
///
/// Emitted when the SEC-21 host-controlled resolver refresh observes a novel
/// IP for a declared FQDN that pushes the cumulative distinct-IP count past
/// the operator-declared `rebindingPolicy.maxNovelIpsPerHostname` cap. All
/// fields mirror [`DnsAuthorityRebindThreshold`] one-for-one; this builder
/// exists so resolver-refresh wiring can serialize the payload without
/// re-deriving the canonical JSON shape from the struct.
pub fn dns_authority_rebind_threshold_data_v1(
    payload: &DnsAuthorityRebindThreshold,
) -> Result<Value, serde_json::Error> {
    serde_json::to_value(payload)
}

/// CloudEvent envelope constructor for `dns_authority_rebind_threshold`
/// (SEC-21 Phase 3e).
///
/// Mirrors [`cloud_event_v1_dns_authority_drift`] for the rebinding-threshold
/// signal. Lives in `cellos-core` so the resolver-refresh module can build
/// envelopes without taking a supervisor-internal dependency.
///
/// `source` should identify the publisher (e.g. `"cellos-supervisor"` or
/// `"cellos-resolver-refresh"`). `time` is the RFC3339 emission timestamp;
/// independent of `payload.observed_at`, which records when the threshold was
/// observed (the two are typically equal but conceptually distinct).
pub fn cloud_event_v1_dns_authority_rebind_threshold(
    source: &str,
    time: &str,
    payload: &DnsAuthorityRebindThreshold,
) -> Result<CloudEventV1, serde_json::Error> {
    Ok(CloudEventV1 {
        specversion: "1.0".into(),
        id: uuid::Uuid::new_v4().to_string(),
        source: source.to_string(),
        ty: "dev.cellos.events.cell.observability.v1.dns_authority_rebind_threshold".into(),
        datacontenttype: Some("application/json".into()),
        data: Some(dns_authority_rebind_threshold_data_v1(payload)?),
        time: Some(time.to_string()),
        traceparent: None,
        cex: None,
    })
}

/// `dev.cellos.events.cell.observability.v1.dns_authority_rebind_rejected`
/// (SEC-21 Phase 3e).
///
/// Schema: `contracts/schemas/cell-observability-dns-authority-rebind-rejected-v1.schema.json`.
///
/// Emitted when the SEC-21 host-controlled resolver refresh observes an IP
/// for a declared FQDN that is NOT in the operator's
/// `rebindingPolicy.responseIpAllowlist`. Only emitted when the allowlist is
/// non-empty for the hostname; one event per offending IP per tick. Combined
/// with `rejectOnRebind=true`, the IP is also dropped from the resolved
/// target set passed to the workload.
pub fn dns_authority_rebind_rejected_data_v1(
    payload: &DnsAuthorityRebindRejected,
) -> Result<Value, serde_json::Error> {
    serde_json::to_value(payload)
}

/// CloudEvent envelope constructor for `dns_authority_rebind_rejected`
/// (SEC-21 Phase 3e).
///
/// Mirrors [`cloud_event_v1_dns_authority_rebind_threshold`] for the
/// allowlist-violation signal. One event per offending IP per tick (so a
/// single response containing 3 disallowed IPs produces 3 rejected events).
pub fn cloud_event_v1_dns_authority_rebind_rejected(
    source: &str,
    time: &str,
    payload: &DnsAuthorityRebindRejected,
) -> Result<CloudEventV1, serde_json::Error> {
    Ok(CloudEventV1 {
        specversion: "1.0".into(),
        id: uuid::Uuid::new_v4().to_string(),
        source: source.to_string(),
        ty: "dev.cellos.events.cell.observability.v1.dns_authority_rebind_rejected".into(),
        datacontenttype: Some("application/json".into()),
        data: Some(dns_authority_rebind_rejected_data_v1(payload)?),
        time: Some(time.to_string()),
        traceparent: None,
        cex: None,
    })
}

/// `dev.cellos.events.cell.observability.v1.dns_authority_dnssec_failed`
/// (SEC-21 Phase 3h).
///
/// Schema: `contracts/schemas/cell-observability-dns-authority-dnssec-failed-v1.schema.json`.
///
/// Emitted by the SEC-21 host-controlled resolver-refresh loop when an opt-in
/// DNSSEC-validating resolver returns a response that fails the chain-of-trust
/// check (or when the zone is not signed and the operator has opted into
/// DNSSEC). All fields mirror [`DnsAuthorityDnssecFailed`] one-for-one; this
/// builder exists so resolver-refresh wiring can serialize the payload without
/// re-deriving the canonical JSON shape from the struct.
pub fn dns_authority_dnssec_failed_data_v1(
    payload: &DnsAuthorityDnssecFailed,
) -> Result<Value, serde_json::Error> {
    serde_json::to_value(payload)
}

/// CloudEvent envelope constructor for `dns_authority_dnssec_failed`
/// (SEC-21 Phase 3h).
///
/// Mirrors [`cloud_event_v1_dns_authority_drift`] for the DNSSEC-failure
/// signal. Lives in `cellos-core` so the resolver-refresh module can build
/// envelopes without taking a supervisor-internal dependency.
///
/// `source` should identify the publisher (e.g. `"cellos-supervisor"` or
/// `"cellos-resolver-refresh"`). `time` is the RFC3339 emission timestamp;
/// independent of `payload.observed_at`, which records when the failure
/// was observed (the two are typically equal but conceptually distinct).
pub fn cloud_event_v1_dns_authority_dnssec_failed(
    source: &str,
    time: &str,
    payload: &DnsAuthorityDnssecFailed,
) -> Result<CloudEventV1, serde_json::Error> {
    Ok(CloudEventV1 {
        specversion: "1.0".into(),
        id: uuid::Uuid::new_v4().to_string(),
        source: source.to_string(),
        ty: "dev.cellos.events.cell.observability.v1.dns_authority_dnssec_failed".into(),
        datacontenttype: Some("application/json".into()),
        data: Some(dns_authority_dnssec_failed_data_v1(payload)?),
        time: Some(time.to_string()),
        traceparent: None,
        cex: None,
    })
}

/// `dev.cellos.events.cell.observability.v1.dns_query`
///
/// Schema: `contracts/schemas/cell-observability-dns-query-v1.schema.json`.
///
/// Emitted by the SEAM-1 / L2-04 in-netns DNS proxy on every query — allowed,
/// denied, malformed, or upstream-timed-out. All fields mirror [`DnsQueryEvent`]
/// one-for-one; this builder exists so dataplane callers can serialize the
/// payload without re-deriving the canonical JSON shape from the struct.
pub fn dns_query_data_v1(event: &DnsQueryEvent) -> Result<Value, serde_json::Error> {
    serde_json::to_value(event)
}

/// CloudEvent envelope constructor for `dns_query` (SEAM-1 / L2-04).
///
/// Mirrors [`cloud_event_v1_dns_authority_drift`] for the per-query DNS proxy
/// signal. Lives in `cellos-core` so the supervisor's DNS proxy module can
/// build envelopes without taking a supervisor-internal dependency.
///
/// `source` should identify the publisher (e.g. `"cellos-supervisor"` or
/// `"cellos-dns-proxy"`). `time` is the RFC3339 emission timestamp; typically
/// equal to `event.observed_at` but conceptually distinct (the latter is when
/// the proxy *observed* the query).
pub fn cloud_event_v1_dns_query(
    source: &str,
    time: &str,
    event: &DnsQueryEvent,
) -> Result<CloudEventV1, serde_json::Error> {
    Ok(CloudEventV1 {
        specversion: "1.0".into(),
        id: uuid::Uuid::new_v4().to_string(),
        source: source.to_string(),
        ty: "dev.cellos.events.cell.observability.v1.dns_query".into(),
        datacontenttype: Some("application/json".into()),
        data: Some(dns_query_data_v1(event)?),
        time: Some(time.to_string()),
        traceparent: None,
        cex: None,
    })
}

/// SEAM-1 Phase 3 — per-query data payload for
/// `dev.cellos.events.cell.dns.v1.query_permitted`.
///
/// Schema: `contracts/schemas/cell-dns-query-permitted-v1.schema.json`.
///
/// Companion to the existing aggregate `dns_query` event: this is the
/// short-form "the allowlist check accepted this query" signal that fires
/// as soon as the proxy decides to forward, BEFORE the upstream answer
/// arrives. It exists so operators can audit the workload's query
/// intent independently of upstream latency / outcome.
///
/// `qname` is lowercased and trailing-dot-stripped per the upstream
/// parser; `qtype` is the IANA mnemonic (`"A"`, `"AAAA"`, etc); `cell_id`
/// mirrors `lifecycle.started.cellId`; `resolver` mirrors the
/// `dnsAuthority.resolvers[].resolverId` the proxy will forward to.
#[must_use]
pub fn dns_query_permitted_data_v1(
    qname: &str,
    qtype: &str,
    cell_id: &str,
    resolver: &str,
) -> Value {
    json!({
        "schemaVersion": "1.0.0",
        "queryName": qname,
        "queryType": qtype,
        "cellId": cell_id,
        "resolver": resolver,
    })
}

/// CloudEvent envelope constructor for `dns_query_permitted` (SEAM-1 Phase 3).
pub fn cloud_event_v1_dns_query_permitted(
    source: &str,
    time: &str,
    qname: &str,
    qtype: &str,
    cell_id: &str,
    resolver: &str,
) -> CloudEventV1 {
    CloudEventV1 {
        specversion: "1.0".into(),
        id: uuid::Uuid::new_v4().to_string(),
        source: source.to_string(),
        ty: "dev.cellos.events.cell.dns.v1.query_permitted".into(),
        datacontenttype: Some("application/json".into()),
        data: Some(dns_query_permitted_data_v1(qname, qtype, cell_id, resolver)),
        time: Some(time.to_string()),
        traceparent: None,
        cex: None,
    }
}

/// SEAM-1 Phase 3 — per-query data payload for
/// `dev.cellos.events.cell.dns.v1.query_refused`.
///
/// Schema: `contracts/schemas/cell-dns-query-refused-v1.schema.json`.
///
/// Emitted alongside the existing aggregate `dns_query{decision:deny}`
/// event when the allowlist check (or query-type gate) short-circuits a
/// workload query with REFUSED. The short-form payload is intended for
/// SIEM rules that want to fan out on refusal without parsing the
/// richer aggregate event.
///
/// `reason` is one of the proxy's `reasonCode` enum values — typically
/// `"denied_not_in_allowlist"` or `"denied_query_type"`.
#[must_use]
pub fn dns_query_refused_data_v1(qname: &str, qtype: &str, cell_id: &str, reason: &str) -> Value {
    json!({
        "schemaVersion": "1.0.0",
        "queryName": qname,
        "queryType": qtype,
        "cellId": cell_id,
        "reason": reason,
    })
}

/// CloudEvent envelope constructor for `dns_query_refused` (SEAM-1 Phase 3).
pub fn cloud_event_v1_dns_query_refused(
    source: &str,
    time: &str,
    qname: &str,
    qtype: &str,
    cell_id: &str,
    reason: &str,
) -> CloudEventV1 {
    CloudEventV1 {
        specversion: "1.0".into(),
        id: uuid::Uuid::new_v4().to_string(),
        source: source.to_string(),
        ty: "dev.cellos.events.cell.dns.v1.query_refused".into(),
        datacontenttype: Some("application/json".into()),
        data: Some(dns_query_refused_data_v1(qname, qtype, cell_id, reason)),
        time: Some(time.to_string()),
        traceparent: None,
        cex: None,
    }
}

/// `dev.cellos.events.cell.trust.v1.keyset_verified` (SEC-25 Phase 2).
///
/// Schema: `contracts/schemas/cell-trust-keyset-verified-v1.schema.json`.
///
/// Emitted once per supervisor startup when `CELLOS_TRUST_KEYSET_PATH` points
/// at a `signed-trust-keyset-envelope-v1` document AND
/// `verify_signed_trust_keyset_envelope` accepted at least one signature
/// against the operator-loaded keyring (`CELLOS_TRUST_VERIFY_KEYS_PATH`).
///
/// `correlation_id` is optional; the supervisor passes `None` for the default
/// startup-time emission, but other producers may bind a correlation id when
/// the verification is part of a larger orchestration flow.
pub fn keyset_verified_data_v1(
    keyset_id: &str,
    payload_digest: &str,
    verified_signer_kid: &str,
    verified_at: &str,
    correlation_id: Option<&str>,
) -> Result<Value, serde_json::Error> {
    let mut m = Map::new();
    m.insert("schemaVersion".to_string(), json!("1.0.0"));
    m.insert("keysetId".to_string(), json!(keyset_id));
    m.insert("payloadDigest".to_string(), json!(payload_digest));
    m.insert("verifiedSignerKid".to_string(), json!(verified_signer_kid));
    m.insert("verifiedAt".to_string(), json!(verified_at));
    if let Some(cid) = correlation_id {
        m.insert("correlationId".to_string(), json!(cid));
    }
    Ok(Value::Object(m))
}

/// CloudEvent envelope constructor for `keyset_verified` (SEC-25 Phase 2).
///
/// `source` should typically be `"cellos-supervisor"` for the startup-time
/// emission; sibling tooling (`cellos-trustd` etc.) may publish under a
/// different source identifier when verifying envelopes off-host.
pub fn cloud_event_v1_keyset_verified(
    source: &str,
    time: &str,
    keyset_id: &str,
    payload_digest: &str,
    verified_signer_kid: &str,
    verified_at: &str,
    correlation_id: Option<&str>,
) -> Result<CloudEventV1, serde_json::Error> {
    Ok(CloudEventV1 {
        specversion: "1.0".into(),
        id: uuid::Uuid::new_v4().to_string(),
        source: source.to_string(),
        ty: "dev.cellos.events.cell.trust.v1.keyset_verified".into(),
        datacontenttype: Some("application/json".into()),
        data: Some(keyset_verified_data_v1(
            keyset_id,
            payload_digest,
            verified_signer_kid,
            verified_at,
            correlation_id,
        )?),
        time: Some(time.to_string()),
        traceparent: None,
        cex: None,
    })
}

/// `dev.cellos.events.cell.trust.v1.keyset_verification_failed` (SEC-25 Phase 2).
///
/// Schema: `contracts/schemas/cell-trust-keyset-verification-failed-v1.schema.json`.
///
/// Emitted once per supervisor startup when `CELLOS_TRUST_KEYSET_PATH` is set
/// AND verification fails (file open / JSON parse / digest mismatch / no
/// signature verified). Under `CELLOS_REQUIRE_TRUST_VERIFY_KEYS=1` the
/// supervisor instead returns an error from `build_supervisor` and never
/// emits this event — fail-closed startup is the explicit operator opt-in;
/// fail-open is the default and surfaces this event so degraded trust posture
/// is visible in the audit stream.
///
/// `attempted_keyset_path` MUST be the file's basename (the supervisor's
/// emitter strips the directory portion to avoid leaking deployment-side
/// metadata into the event stream — see the schema description).
pub fn keyset_verification_failed_data_v1(
    attempted_keyset_path: &str,
    reason: &str,
    failed_at: &str,
    correlation_id: Option<&str>,
) -> Result<Value, serde_json::Error> {
    let mut m = Map::new();
    m.insert("schemaVersion".to_string(), json!("1.0.0"));
    m.insert(
        "attemptedKeysetPath".to_string(),
        json!(attempted_keyset_path),
    );
    m.insert("reason".to_string(), json!(reason));
    m.insert("failedAt".to_string(), json!(failed_at));
    if let Some(cid) = correlation_id {
        m.insert("correlationId".to_string(), json!(cid));
    }
    Ok(Value::Object(m))
}

/// CloudEvent envelope constructor for `keyset_verification_failed` (SEC-25 Phase 2).
pub fn cloud_event_v1_keyset_verification_failed(
    source: &str,
    time: &str,
    attempted_keyset_path: &str,
    reason: &str,
    failed_at: &str,
    correlation_id: Option<&str>,
) -> Result<CloudEventV1, serde_json::Error> {
    Ok(CloudEventV1 {
        specversion: "1.0".into(),
        id: uuid::Uuid::new_v4().to_string(),
        source: source.to_string(),
        ty: "dev.cellos.events.cell.trust.v1.keyset_verification_failed".into(),
        datacontenttype: Some("application/json".into()),
        data: Some(keyset_verification_failed_data_v1(
            attempted_keyset_path,
            reason,
            failed_at,
            correlation_id,
        )?),
        time: Some(time.to_string()),
        traceparent: None,
        cex: None,
    })
}

/// `dev.cellos.events.cell.observability.v1.network_flow_decision` (FC-38 Phase 1).
///
/// Schema: `contracts/schemas/cell-observability-network-flow-decision-v1.schema.json`.
///
/// Emitted by the supervisor's post-run nft counter scan
/// (`nft list ruleset --json`) when `CELLOS_PER_FLOW_ENFORCEMENT_EVENTS=1`.
/// scope: honest "policy-applied attribution" — one event per matched or
/// applied rule, NOT a real-time per-packet stream. Future eBPF / nflog
/// work targets real-time per-flow events.
///
/// All fields mirror [`NetworkFlowDecision`] one-for-one; this builder exists
/// so supervisor wiring can emit the payload without re-deriving the canonical
/// JSON shape from the struct.
pub fn network_flow_decision_data_v1(
    decision: &NetworkFlowDecision,
) -> Result<Value, serde_json::Error> {
    serde_json::to_value(decision)
}

/// CloudEvent envelope constructor for `network_flow_decision` (FC-38 Phase 1).
///
/// Mirrors [`cloud_event_v1_dns_authority_drift`] for the per-flow signal.
/// Lives in `cellos-core` so the supervisor's post-run counter-scan path can
/// build envelopes without taking a supervisor-internal dependency.
///
/// `source` should identify the publisher (e.g. `"cellos-supervisor"`).
/// `time` is the RFC3339 emission timestamp; typically equal to
/// `decision.observed_at` but conceptually distinct (the latter is when the
/// supervisor *scraped* the counter, the former when the envelope was built).
pub fn cloud_event_v1_network_flow_decision(
    source: &str,
    time: &str,
    decision: &NetworkFlowDecision,
) -> Result<CloudEventV1, serde_json::Error> {
    Ok(CloudEventV1 {
        specversion: "1.0".into(),
        id: uuid::Uuid::new_v4().to_string(),
        source: source.to_string(),
        ty: "dev.cellos.events.cell.observability.v1.network_flow_decision".into(),
        datacontenttype: Some("application/json".into()),
        data: Some(network_flow_decision_data_v1(decision)?),
        time: Some(time.to_string()),
        traceparent: None,
        cex: None,
    })
}

/// `dev.cellos.events.cell.observability.v1.l7_egress_decision`
///
/// Schema: `contracts/schemas/cell-observability-l7-egress-decision-v1.schema.json`.
#[allow(clippy::too_many_arguments)]
pub fn observability_l7_egress_decision_data_v1(
    spec: &ExecutionCellSpec,
    cell_id: &str,
    run_id: Option<&str>,
    decision_id: &str,
    action: &str,
    sni_host: &str,
    policy_digest: &str,
    keyset_id: &str,
    issuer_kid: &str,
    reason_code: &str,
    rule_ref: Option<&str>,
    // scope: per-stream correlation id for h2 paths. `None` for
    // SNI / HTTP/1.x / unknown / peek-timeout paths. Additive — the
    // schema's `streamId` field is optional so the field is omitted
    // from the output when `None`.
    stream_id: Option<u32>,
) -> Result<Value, serde_json::Error> {
    let mut m = Map::new();
    m.insert("cellId".to_string(), json!(cell_id));
    m.insert("specId".to_string(), json!(&spec.id));
    if let Some(r) = run_id {
        m.insert("runId".to_string(), json!(r));
    }
    m.insert("decisionId".to_string(), json!(decision_id));
    m.insert("action".to_string(), json!(action));
    m.insert("sniHost".to_string(), json!(sni_host));
    m.insert("policyDigest".to_string(), json!(policy_digest));
    m.insert("keysetId".to_string(), json!(keyset_id));
    m.insert("issuerKid".to_string(), json!(issuer_kid));
    m.insert("reasonCode".to_string(), json!(reason_code));
    if let Some(rr) = rule_ref {
        m.insert("ruleRef".to_string(), json!(rr));
    }
    if let Some(sid) = stream_id {
        m.insert("streamId".to_string(), json!(sid));
    }
    if let Some(c) = &spec.correlation {
        m.insert("correlation".to_string(), serde_json::to_value(c)?);
    }
    Ok(Value::Object(m))
}

/// `dev.cellos.events.cell.observability.v1.container_security`
///
/// Emitted once per cell run immediately after `lifecycle.started`, recording
/// the Linux capability bitmasks read from `/proc/self/status` at supervisor
/// start time.  This gives the audit trail a concrete answer to "what
/// capabilities did the cell have?" rather than "deepce could not determine
/// capabilities because capsh was not installed."
///
/// Fields:
/// - `capEff`  — effective capability bitmask (hex string, e.g. `"00000000a80425fb"`)
/// - `capPrm`  — permitted capability bitmask
/// - `capBnd`  — bounding set bitmask
/// - `capAmb`  — ambient capability bitmask
/// - `capInh`  — inheritable capability bitmask
/// - `privileged` — `true` when `capEff` equals `0x1ffffffffff` (all 40 caps set)
///
/// All fields are `None`/omitted on non-Linux hosts or when `/proc/self/status`
/// is unreadable.  Consumers should treat absence as "unknown," not "clean."
pub fn observability_container_security_data_v1(
    cell_id: &str,
    run_id: Option<&str>,
    cap_eff: Option<&str>,
    cap_prm: Option<&str>,
    cap_bnd: Option<&str>,
    cap_amb: Option<&str>,
    cap_inh: Option<&str>,
) -> Value {
    let mut m = Map::new();
    m.insert("cellId".to_string(), json!(cell_id));
    if let Some(r) = run_id {
        m.insert("runId".to_string(), json!(r));
    }
    if let (Some(eff), Some(prm), Some(bnd), Some(amb), Some(inh)) =
        (cap_eff, cap_prm, cap_bnd, cap_amb, cap_inh)
    {
        m.insert("capEff".to_string(), json!(eff));
        m.insert("capPrm".to_string(), json!(prm));
        m.insert("capBnd".to_string(), json!(bnd));
        m.insert("capAmb".to_string(), json!(amb));
        m.insert("capInh".to_string(), json!(inh));
        // Full privileged set: all 40 defined capabilities (Linux 5.x bitmask).
        let privileged = eff == "0000001fffffffff";
        m.insert("privileged".to_string(), json!(privileged));
    }
    Value::Object(m)
}

/// `dev.cellos.events.cell.compliance.v1.summary`
///
/// Emitted once per cell run immediately before `lifecycle.destroyed`.
/// Captures the compliance-relevant profile of the run: which policy pack
/// governed it, what security controls were declared, and the final command
/// exit code (when available).
///
/// This event is the canonical compliance receipt — it allows fleet operators
/// and auditors to reconstruct "what policy was in effect and what was
/// observed" for every cell run without replaying the full event stream.
///
/// Schema: `contracts/schemas/cell-compliance-summary-v1.schema.json`.
///
/// Backward-compatible thin delegate to
/// [`compliance_summary_data_v1_with_subjects`] with an empty subject set.
/// Existing callers (supervisor and tests) keep their 4-arg signature; the
/// resulting payload omits `subjectUrns` per the schema's "absent when empty"
/// contract.
pub fn compliance_summary_data_v1(
    spec: &ExecutionCellSpec,
    cell_id: &str,
    run_id: Option<&str>,
    command_exit_code: Option<i32>,
) -> Result<Value, serde_json::Error> {
    compliance_summary_data_v1_with_subjects(spec, cell_id, run_id, command_exit_code, &[])
}

/// Seam-freeze G4 / P0-7 variant of [`compliance_summary_data_v1`] that
/// attaches a typed-URN subject set covered by the compliance receipt.
///
/// `subject_urns` SHOULD be the concrete URNs (cells, leases, exports, …)
/// whose lifecycle this compliance.summary attests to. Callers pass `&[]`
/// for the legacy "no cross-pointer" shape, which is exactly what the
/// thin-delegate [`compliance_summary_data_v1`] does.
///
/// The `subjectUrns` field is **only emitted when non-empty** — schema
/// declares `minItems: 1` and `uniqueItems: true`, so emitting an empty array
/// would fail validation. URN shape validation lives in the schema (regex);
/// this function does not pre-validate to keep the API allocation-free.
///
/// Schema: `contracts/schemas/cell-compliance-summary-v1.schema.json`
/// (`subjectUrns` field).
pub fn compliance_summary_data_v1_with_subjects(
    spec: &ExecutionCellSpec,
    cell_id: &str,
    run_id: Option<&str>,
    command_exit_code: Option<i32>,
    subject_urns: &[SubjectUrn],
) -> Result<Value, serde_json::Error> {
    let egress_rule_count = spec
        .authority
        .egress_rules
        .as_ref()
        .map(|v| v.len())
        .unwrap_or(0);
    let export_target_count = spec
        .export
        .as_ref()
        .and_then(|e| e.targets.as_ref())
        .map(|t| t.len())
        .unwrap_or(0);
    let resource_limits_present = spec.run.as_ref().and_then(|r| r.limits.as_ref()).is_some();
    let secret_delivery_mode = spec
        .run
        .as_ref()
        .map(|r| serde_json::to_value(&r.secret_delivery))
        .transpose()?
        .unwrap_or(serde_json::Value::String("env".into()));

    let mut m = Map::new();
    m.insert("cellId".to_string(), json!(cell_id));
    m.insert("specId".to_string(), json!(&spec.id));
    m.insert(
        "lifetimeTtlSeconds".to_string(),
        json!(spec.lifetime.ttl_seconds),
    );
    m.insert("egressRuleCount".to_string(), json!(egress_rule_count));
    m.insert(
        "resourceLimitsPresent".to_string(),
        json!(resource_limits_present),
    );
    m.insert("secretDeliveryMode".to_string(), secret_delivery_mode);
    m.insert("exportTargetCount".to_string(), json!(export_target_count));

    // Policy attribution — absent when no pack was declared on the spec.
    if let Some(policy) = &spec.policy {
        if let Some(id) = &policy.pack_id {
            m.insert("policyPackId".to_string(), json!(id));
        }
        if let Some(ver) = &policy.pack_version {
            m.insert("policyPackVersion".to_string(), json!(ver));
        }
        if let Some(digest) = &policy.bundle_digest {
            m.insert("policyBundleDigest".to_string(), json!(digest));
        }
    }

    if let Some(placement) = &spec.placement {
        let mut placement_map = Map::new();
        if let Some(pool_id) = &placement.pool_id {
            placement_map.insert("poolId".to_string(), json!(pool_id));
        }
        if let Some(namespace) = &placement.kubernetes_namespace {
            placement_map.insert("kubernetesNamespace".to_string(), json!(namespace));
        }
        if let Some(queue_name) = &placement.queue_name {
            placement_map.insert("queueName".to_string(), json!(queue_name));
        }
        if !placement_map.is_empty() {
            m.insert("placement".to_string(), Value::Object(placement_map));
        }
    }

    if let Some(code) = command_exit_code {
        m.insert("commandExitCode".to_string(), json!(code));
    }
    if let Some(r) = run_id {
        m.insert("runId".to_string(), json!(r));
    }
    if let Some(c) = &spec.correlation {
        m.insert("correlation".to_string(), serde_json::to_value(c)?);
    }

    // Seam-freeze G4 / P0-7 cross-pointer. Emit only when non-empty so the
    // legacy 4-arg path (and any future "no subjects" caller) keeps producing
    // a payload that round-trips against the existing valid example.
    if !subject_urns.is_empty() {
        let urns: Vec<Value> = subject_urns.iter().map(|u| json!(u)).collect();
        m.insert("subjectUrns".to_string(), Value::Array(urns));
    }

    Ok(Value::Object(m))
}

/// `data` for event type `dev.cellos.events.cell.policy.v1.rejected`.
///
/// Emitted when a policy pack's admission gate rejects a cell before the
/// lifecycle starts.  The `violations` array lists every failed rule, giving
/// operators a complete picture of what must change.
///
/// Schema: `contracts/schemas/cell-policy-rejected-v1.schema.json`.
pub fn policy_rejected_data_v1(
    spec: &ExecutionCellSpec,
    violations: &[PolicyViolation],
) -> Result<Value, serde_json::Error> {
    let violation_values: Vec<Value> = violations
        .iter()
        .map(|v| {
            json!({
                "rule": v.rule,
                "message": v.message,
            })
        })
        .collect();

    let mut m = Map::new();
    m.insert("specId".to_string(), json!(&spec.id));
    m.insert("violationCount".to_string(), json!(violations.len()));
    m.insert("violations".to_string(), Value::Array(violation_values));

    // Policy attribution — absent when no pack was declared on the spec.
    if let Some(policy) = &spec.policy {
        if let Some(id) = &policy.pack_id {
            m.insert("policyPackId".to_string(), json!(id));
        }
        if let Some(ver) = &policy.pack_version {
            m.insert("policyPackVersion".to_string(), json!(ver));
        }
    }

    if let Some(c) = &spec.correlation {
        m.insert("correlation".to_string(), serde_json::to_value(c)?);
    }
    Ok(Value::Object(m))
}

/// CloudEvent `type` URN for an admitted cell (ADR-0025 §1).
///
/// Emitted by the org-root ceiling gate (ADR-0024) when a spec's declared
/// authority is a subset of the signed ceiling. Closes the
/// "admission inferred only from `lifecycle.started`" gap: an allow is now a
/// first-class, offline-verifiable record, not an inference.
pub const ADMISSION_ALLOWED_TYPE: &str = "dev.cellos.events.cell.admission.v1.allowed";

/// CloudEvent `type` URN for a rejected cell (ADR-0025 §1).
///
/// Emitted by the org-root ceiling gate (ADR-0024) on the reject early-return
/// path, before the verdict's `Err` is propagated. Sibling to
/// [`ADMISSION_ALLOWED_TYPE`]; both carry an [`admission_decision_data_v1`]
/// payload.
pub const ADMISSION_REJECTED_TYPE: &str = "dev.cellos.events.cell.admission.v1.rejected";

/// CloudEvent `type` of a signed audit-reconcile receipt (S18, ADR-0029):
/// emitted after the spool forwarder (S17) delivers a contiguous range of the
/// BLAKE3-chained spool to the broker. Carries an [`audit_reconciled_data_v1`]
/// payload. Schema: `contracts/schemas/audit-reconciled-v1.schema.json`.
pub const AUDIT_RECONCILED_TYPE: &str = "dev.cellos.events.cell.audit.v1.reconciled";

/// Closed reason discriminant for an admission decision (ADR-0024 §Verdict,
/// ADR-0025 §1).
///
/// The reason travels with the boolean `admitted` outcome in the
/// [`admission_decision_data_v1`] payload so an auditor reads *what was decided
/// and why* from the receipt alone, with no producer source access. The set is
/// closed: it is the exact discriminant set named in ADR-0024, shared by value
/// (the snake_case wire string) with the ADR-0024 ceiling verdict. No
/// environment string (`"prod"`) is ever a member — the predicate is an
/// allowlist comparison, never an environment label.
///
/// Serializes to a stable `snake_case` string (e.g. `authority_exceeds_ceiling`)
/// via serde; two independent producers emit byte-identical `reason` values for
/// the same verdict.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum AdmissionDecisionReason {
    /// Spec authority is not a subset of the signed ceiling — reject.
    AuthorityExceedsCeiling,
    /// Ceiling manifest envelope failed signature/threshold verification.
    CeilingManifestUnverified,
    /// Ceiling manifest inner payload could not be parsed.
    CeilingManifestUnparseable,
    /// Ceiling envelope carried a non-ceiling `payload_type`.
    CeilingManifestWrongPayloadType,
    /// A populated dimension (filesystem/network/dns/cdn) made the subset
    /// comparison undecidable under the hardened profile.
    CeilingDimensionUncomparable,
    /// Ceiling epoch is older than the on-disk freshness anchor (replay).
    CeilingEpochStale,
    /// Ceiling not configured and the profile is not hardened — the single
    /// labeled, degraded fail-open path.
    CeilingDegradedUnconfigured,
    /// A formation edge's child authority is not a subset of its parent.
    FormationEdgeNotNarrowing,
    /// Spec authority is a subset of the signed ceiling — admit.
    AdmittedWithinCeiling,
}

/// Whether an *allow* decision corresponds to a kernel-bound egress allowlist
/// or a declared/audited one (ADR-0025 §Linux-vs-Windows boundary).
///
/// Recorded on the receipt so a verifier on either OS can tell what kind of
/// enforcement backs the admission. The signature is platform-independent; this
/// field makes the receipt's truthfulness about enforcement platform-aware.
///
/// Serializes to `linux-nftables` / `declared-audit`.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
#[serde(rename_all = "kebab-case")]
pub enum EnforcementBinding {
    /// Egress allowlist is bound in-kernel via nftables (Linux).
    LinuxNftables,
    /// Egress allowlist is declared and audited, not kernel-bound
    /// (Windows / portable backends).
    DeclaredAudit,
}

/// `data` for the admission-decision events
/// `dev.cellos.events.cell.admission.v1.{allowed,rejected}` (ADR-0025 §1).
///
/// Sibling of [`policy_rejected_data_v1`]: the org-root ceiling gate
/// (ADR-0024) emits one of these on **every** admission decision — allow and
/// reject alike — so the verdict is a signed, offline-verifiable record rather
/// than an inference from `lifecycle.started`.
///
/// The payload is legible without producer-source access: `admitted` is the
/// verdict as a plain boolean ([`AuthorityCapability::is_superset_of`] result),
/// `reason` is a closed discriminant ([`AdmissionDecisionReason`]), and
/// `predicate` is the human-readable statement of what was checked. `specHash`
/// (`raw_spec_hash`), `runId`, and `enforcementBinding` bind the decision to a
/// concrete spec, run, and enforcement posture. `context` carries
/// non-load-bearing routing labels and is omitted when empty.
///
/// Pure JSON constructor (no I/O). The signing and emission live in the
/// supervisor's standalone main.rs signer (ADR-0025 §2). Determinism: the field
/// order below is stable, so `serde_json::to_vec` of two builds of the same
/// verdict is byte-identical (the canonical signing payload, ADR-0025 §4).
#[allow(clippy::too_many_arguments)]
pub fn admission_decision_data_v1(
    spec: &ExecutionCellSpec,
    admitted: bool,
    reason: AdmissionDecisionReason,
    predicate: &str,
    spec_hash: &str,
    run_id: Option<&str>,
    enforcement_binding: EnforcementBinding,
    context: &[(&str, &str)],
) -> Result<Value, serde_json::Error> {
    let mut m = Map::new();
    m.insert("specId".to_string(), json!(&spec.id));
    m.insert("admitted".to_string(), json!(admitted));
    m.insert("reason".to_string(), serde_json::to_value(reason)?);
    m.insert("predicate".to_string(), json!(predicate));
    m.insert("specHash".to_string(), json!(spec_hash));
    m.insert(
        "enforcementBinding".to_string(),
        serde_json::to_value(enforcement_binding)?,
    );

    // S09 (ADR-0027): stamp the active crypto provider identity so an offline
    // auditor sees WHICH module signed this decision (e.g. dalek vs aws-lc-fips).
    // `fipsMode` / `moduleCert` are a PRODUCER CLAIM, not a validated attestation
    // — SC-13 is met only with an active CMVP cert, never asserted here.
    let pid = crate::crypto::provider_identity();
    m.insert(
        "providerIdentity".to_string(),
        json!({
            "name": pid.name,
            "fipsMode": pid.fips_mode,
            "moduleCert": pid.module_cert,
        }),
    );

    if let Some(r) = run_id {
        m.insert("runId".to_string(), json!(r));
    }

    // Non-load-bearing routing labels. Omitted when empty so a caller passing
    // no context produces a payload that round-trips without a `context` key.
    if !context.is_empty() {
        let mut ctx = Map::new();
        for (k, v) in context {
            ctx.insert((*k).to_string(), json!(v));
        }
        m.insert("context".to_string(), Value::Object(ctx));
    }

    if let Some(c) = &spec.correlation {
        m.insert("correlation".to_string(), serde_json::to_value(c)?);
    }
    Ok(Value::Object(m))
}

/// `data` for [`AUDIT_RECONCILED_TYPE`] (S18, ADR-0029): a signed receipt that a
/// contiguous range `[fromSeq, toSeq]` of the BLAKE3-chained spool was delivered
/// to the broker. Sibling of [`admission_decision_data_v1`]; emitted via the
/// standalone signer after a successful forward (S17), using the S01 canonical
/// encoder so it is offline-verifiable under the org-root key.
/// Schema: `contracts/schemas/audit-reconciled-v1.schema.json`.
#[allow(clippy::too_many_arguments)]
pub fn audit_reconciled_data_v1(
    chain_id: &str,
    from_seq: u64,
    to_seq: u64,
    head_row_hash: &str,
    delivered_count: u64,
    broker_acked_head: Option<&str>,
    integrity: &str,
    signed: bool,
) -> Result<Value, serde_json::Error> {
    let mut m = Map::new();
    m.insert("chainId".to_string(), json!(chain_id));
    m.insert("fromSeq".to_string(), json!(from_seq));
    m.insert("toSeq".to_string(), json!(to_seq));
    m.insert("headRowHash".to_string(), json!(head_row_hash));
    m.insert("deliveredCount".to_string(), json!(delivered_count));
    m.insert("integrity".to_string(), json!(integrity));
    m.insert("signed".to_string(), json!(signed));
    if let Some(h) = broker_acked_head {
        m.insert("brokerAckedHead".to_string(), json!(h));
    }
    Ok(Value::Object(m))
}

/// T12 RBAC — `data` for event type `dev.cellos.events.cell.authz.v1.rejected`.
///
/// Emitted by the supervisor when the operator's `AuthorizationPolicy`
/// (loaded from `CELLOS_AUTHZ_POLICY_PATH`) rejects a cell spec at
/// admission. Sibling event to `cell.policy.v1.rejected`, with a different
/// rule namespace (authz subject/pool/pack vs policy-pack admission). Both
/// gates can fire independently.
///
/// `reason` is one of the strings declared in the schema:
/// `subject_not_authorized`, `pool_not_allowed`, `policy_pack_not_allowed`,
/// `rate_limit_exceeded`. Schema:
/// `contracts/schemas/cell-authz-rejected-v1.schema.json`.
pub fn authz_rejected_data_v1(
    spec: &ExecutionCellSpec,
    reason: &str,
    message: &str,
    denied_pool_id: Option<&str>,
    denied_policy_pack_id: Option<&str>,
) -> Result<Value, serde_json::Error> {
    let mut m = Map::new();
    m.insert("specId".to_string(), json!(&spec.id));
    m.insert("reason".to_string(), json!(reason));
    m.insert("message".to_string(), json!(message));

    let tenant_id = spec
        .correlation
        .as_ref()
        .and_then(|c| c.tenant_id.as_deref());
    if let Some(t) = tenant_id {
        m.insert("tenantId".to_string(), json!(t));
        // 1.0 subject axis: subject == tenantId. Mirrored as a separate field
        // so consumers don't need to reach into correlation, and so future
        // subject axes (oidc, k8s) can replace it without breaking the
        // event shape.
        m.insert("subject".to_string(), json!(t));
    }

    if let Some(p) = denied_pool_id {
        m.insert("deniedPoolId".to_string(), json!(p));
    }
    if let Some(p) = denied_policy_pack_id {
        m.insert("deniedPolicyPackId".to_string(), json!(p));
    }

    if let Some(c) = &spec.correlation {
        m.insert("correlation".to_string(), serde_json::to_value(c)?);
    }
    Ok(Value::Object(m))
}

/// `dev.cellos.events.cell.authority.v1.homeostasis`
///
/// Emitted once per run between `compliance.summary` and `lifecycle.destroyed`.
/// Compares declared vs exercised authority. Aggregation across runs belongs in taudit.
///
/// Schema: `contracts/schemas/cell-authority-homeostasis-v1.schema.json`.
pub fn homeostasis_signal_data_v1(
    spec: &ExecutionCellSpec,
    cell_id: &str,
    run_id: Option<&str>,
    signal: &crate::HomeostasisSignal,
) -> Result<Value, serde_json::Error> {
    let mut m = Map::new();
    m.insert("cellId".to_string(), json!(cell_id));
    m.insert("specId".to_string(), json!(&spec.id));
    m.insert("specHash".to_string(), json!(&signal.spec_hash));
    m.insert(
        "declaredEgressRules".to_string(),
        json!(signal.declared_egress_rules),
    );
    // E2-06: exercisedEgressConnections is nullable while real telemetry (L2-04 / L5-15)
    // is wired. When the count is None, emit JSON `null` AND a sibling
    // `exercisedEgressReason` so consumers can distinguish "telemetry pending" from a
    // true zero.
    m.insert(
        "exercisedEgressConnections".to_string(),
        match signal.exercised_egress_connections {
            Some(n) => json!(n),
            None => Value::Null,
        },
    );
    if let Some(reason) = &signal.exercised_egress_reason {
        m.insert("exercisedEgressReason".to_string(), json!(reason));
    }
    m.insert(
        "declaredMountPaths".to_string(),
        json!(signal.declared_mount_paths),
    );
    m.insert(
        "accessedMountPaths".to_string(),
        json!(signal.accessed_mount_paths),
    );
    m.insert(
        "declaredSecretCount".to_string(),
        json!(signal.declared_secret_count),
    );
    m.insert(
        "authorityEfficiency".to_string(),
        json!(signal.authority_efficiency),
    );
    m.insert(
        "recommendedRemovals".to_string(),
        serde_json::to_value(&signal.recommended_removals)?,
    );
    if let Some(r) = run_id {
        m.insert("runId".to_string(), json!(r));
    }
    if let Some(c) = &spec.correlation {
        m.insert("correlation".to_string(), serde_json::to_value(c)?);
    }
    Ok(Value::Object(m))
}

/// `dev.cellos.events.cell.authority.v1.homeostasis_violation`
///
/// Emitted when exercised egress connections exceed declared authority — a
/// security invariant violation. The workload made connections beyond what
/// it declared it would need, which means either (a) the spec under-declares
/// the workload's real needs, or (b) the nftables/eBPF enforcement layer did
/// not prevent connections that policy rejected.
///
/// Pure JSON constructor (no I/O). Callers compute `overage` outside this
/// helper and pass `exercised`; this builder MUST NOT underflow, so the
/// caller is responsible for guaranteeing `exercised >= declared` before
/// invocation. The supervisor emit site guards on `exercised > declared`.
///
/// Schema: `contracts/schemas/cell-authority-homeostasis-violation-v1.schema.json`.
pub fn homeostasis_violation_data_v1(
    cell_id: &str,
    declared_egress: u64,
    exercised_egress: u64,
    spec_hash: &str,
) -> Value {
    let mut m = Map::new();
    m.insert("cellId".to_string(), json!(cell_id));
    m.insert(
        "declaredEgressRuleCount".to_string(),
        json!(declared_egress),
    );
    m.insert(
        "exercisedEgressConnections".to_string(),
        json!(exercised_egress),
    );
    // Caller guarantees exercised >= declared; saturating_sub protects against
    // accidental misuse without panicking.
    m.insert(
        "overage".to_string(),
        json!(exercised_egress.saturating_sub(declared_egress)),
    );
    m.insert("specHash".to_string(), json!(spec_hash));
    m.insert("severity".to_string(), json!("critical"));
    Value::Object(m)
}

// ---------------------------------------------------------------------------
// F1b — Path B host-side observability primitives + evidence_bundle (ADR-0006)
//
// Builders below construct the JSON `data` payload for the new
// `dev.cellos.events.cell.observability.host.v1.*` event types and the
// per-cell `dev.cellos.events.cell.evidence_bundle.v1.emitted` aggregation
// primitive ("the 1.0 deliverable" — ADR-0006 §4). These are pure JSON
// constructors; sampling and emission live in `cellos-host-telemetry` and
// `cellos-supervisor` (slot F1a).
//
// All five host-probe builders carry a `spec_signature_hash` that the
// supervisor host-stamps before emit (ADR-0006 §6). The evidence_bundle
// builder requires both `spec_signature_hash` and `cell_destroyed_event_ref`
// — the latter enforces D5 destruction-evidence integration: a bundle
// without a `cell.lifecycle.v1.destroyed` reference is structurally invalid
// (see schema gate in `contracts/schemas/evidence-bundle-v1.schema.json`).
// ---------------------------------------------------------------------------

/// `dev.cellos.events.cell.observability.host.v1.fc_metrics`
///
/// Per-cell snapshot of Firecracker metrics endpoint counters (KVM exits,
/// vsock bytes, block ops). Path B (ADR-0006).
///
/// Schema: `contracts/schemas/cell-observability-host-fc-metrics-v1.schema.json`.
#[allow(clippy::too_many_arguments)]
pub fn observability_host_fc_metrics_data_v1(
    spec: &ExecutionCellSpec,
    cell_id: &str,
    run_id: Option<&str>,
    spec_signature_hash: &str,
    sampled_at_unix_ms: u64,
    fc_socket_path: &str,
    vcpu_exits_total: Option<u64>,
    vsock_tx_bytes: Option<u64>,
    vsock_rx_bytes: Option<u64>,
    block_read_ops: Option<u64>,
    block_write_ops: Option<u64>,
    sample_error: Option<&str>,
) -> Result<Value, serde_json::Error> {
    let mut m = Map::new();
    m.insert("cellId".to_string(), json!(cell_id));
    m.insert("specId".to_string(), json!(&spec.id));
    m.insert("specSignatureHash".to_string(), json!(spec_signature_hash));
    m.insert("sampledAtUnixMs".to_string(), json!(sampled_at_unix_ms));
    m.insert("fcSocketPath".to_string(), json!(fc_socket_path));
    if let Some(v) = vcpu_exits_total {
        m.insert("vcpuExitsTotal".to_string(), json!(v));
    }
    if let Some(v) = vsock_tx_bytes {
        m.insert("vsockTxBytes".to_string(), json!(v));
    }
    if let Some(v) = vsock_rx_bytes {
        m.insert("vsockRxBytes".to_string(), json!(v));
    }
    if let Some(v) = block_read_ops {
        m.insert("blockReadOps".to_string(), json!(v));
    }
    if let Some(v) = block_write_ops {
        m.insert("blockWriteOps".to_string(), json!(v));
    }
    if let Some(e) = sample_error {
        m.insert("sampleError".to_string(), json!(e));
    }
    if let Some(r) = run_id {
        m.insert("runId".to_string(), json!(r));
    }
    if let Some(c) = &spec.correlation {
        m.insert("correlation".to_string(), serde_json::to_value(c)?);
    }
    Ok(Value::Object(m))
}

/// One memory.events / cpu.stat / pids.events sample for
/// [`observability_host_cgroup_data_v1`].
#[derive(Debug, Default, Clone)]
pub struct CgroupSample<'a> {
    pub memory_events: Option<&'a [(&'a str, u64)]>,
    pub cpu_stat: Option<&'a [(&'a str, u64)]>,
    pub pids_events: Option<&'a [(&'a str, u64)]>,
}

/// `dev.cellos.events.cell.observability.host.v1.cgroup`
///
/// Per-cell cgroup v2 counter snapshot (memory.events, cpu.stat,
/// pids.events). Path B (ADR-0006).
///
/// `sample` carries the kernel-reported counter pairs the supervisor read
/// from `/sys/fs/cgroup/<cell-leaf>/`. Field names map directly to the
/// schema's nested objects; unknown keys are silently dropped at emit time
/// because the schema's `additionalProperties: false` would reject them.
///
/// Schema: `contracts/schemas/cell-observability-host-cgroup-v1.schema.json`.
fn cgroup_section(keys: &[&str], pairs: Option<&[(&str, u64)]>) -> Option<Value> {
    let pairs = pairs?;
    let mut section = Map::new();
    for (k, v) in pairs {
        if keys.contains(k) {
            section.insert((*k).to_string(), json!(v));
        }
    }
    if section.is_empty() {
        None
    } else {
        Some(Value::Object(section))
    }
}

/// `dev.cellos.events.cell.observability.host.v1.cgroup` (see [`CgroupSample`]).
#[allow(clippy::too_many_arguments)]
pub fn observability_host_cgroup_data_v1(
    spec: &ExecutionCellSpec,
    cell_id: &str,
    run_id: Option<&str>,
    spec_signature_hash: &str,
    sampled_at_unix_ms: u64,
    cgroup_path: &str,
    sample: &CgroupSample<'_>,
    sample_error: Option<&str>,
) -> Result<Value, serde_json::Error> {
    const MEM_KEYS: &[&str] = &["low", "high", "max", "oom", "oomKill"];
    const CPU_KEYS: &[&str] = &[
        "usageUsec",
        "userUsec",
        "systemUsec",
        "nrPeriods",
        "nrThrottled",
        "throttledUsec",
    ];
    const PIDS_KEYS: &[&str] = &["max"];

    let mut m = Map::new();
    m.insert("cellId".to_string(), json!(cell_id));
    m.insert("specId".to_string(), json!(&spec.id));
    m.insert("specSignatureHash".to_string(), json!(spec_signature_hash));
    m.insert("sampledAtUnixMs".to_string(), json!(sampled_at_unix_ms));
    m.insert("cgroupPath".to_string(), json!(cgroup_path));
    if let Some(v) = cgroup_section(MEM_KEYS, sample.memory_events) {
        m.insert("memoryEvents".to_string(), v);
    }
    if let Some(v) = cgroup_section(CPU_KEYS, sample.cpu_stat) {
        m.insert("cpuStat".to_string(), v);
    }
    if let Some(v) = cgroup_section(PIDS_KEYS, sample.pids_events) {
        m.insert("pidsEvents".to_string(), v);
    }
    if let Some(e) = sample_error {
        m.insert("sampleError".to_string(), json!(e));
    }
    if let Some(r) = run_id {
        m.insert("runId".to_string(), json!(r));
    }
    if let Some(c) = &spec.correlation {
        m.insert("correlation".to_string(), serde_json::to_value(c)?);
    }
    Ok(Value::Object(m))
}

/// One nft rule counter row for [`observability_host_nftables_data_v1`].
#[derive(Debug, Clone)]
pub struct NftRuleCounter<'a> {
    pub rule_handle: &'a str,
    pub verdict: Option<&'a str>,
    pub packets: u64,
    pub bytes: u64,
    pub r#match: Option<&'a str>,
}

/// `dev.cellos.events.cell.observability.host.v1.nftables`
///
/// Per-cell nftables counter snapshot scraped from the cell's netns. Path B
/// (ADR-0006).
///
/// Schema: `contracts/schemas/cell-observability-host-nftables-v1.schema.json`.
#[allow(clippy::too_many_arguments)]
pub fn observability_host_nftables_data_v1(
    spec: &ExecutionCellSpec,
    cell_id: &str,
    run_id: Option<&str>,
    spec_signature_hash: &str,
    sampled_at_unix_ms: u64,
    table_name: &str,
    rule_counters: &[NftRuleCounter<'_>],
    sample_error: Option<&str>,
) -> Result<Value, serde_json::Error> {
    let mut m = Map::new();
    m.insert("cellId".to_string(), json!(cell_id));
    m.insert("specId".to_string(), json!(&spec.id));
    m.insert("specSignatureHash".to_string(), json!(spec_signature_hash));
    m.insert("sampledAtUnixMs".to_string(), json!(sampled_at_unix_ms));
    m.insert("tableName".to_string(), json!(table_name));
    let counters: Vec<Value> = rule_counters
        .iter()
        .map(|c| {
            let mut row = Map::new();
            row.insert("ruleHandle".to_string(), json!(c.rule_handle));
            if let Some(v) = c.verdict {
                row.insert("verdict".to_string(), json!(v));
            }
            row.insert("packets".to_string(), json!(c.packets));
            row.insert("bytes".to_string(), json!(c.bytes));
            if let Some(mt) = c.r#match {
                row.insert("match".to_string(), json!(mt));
            }
            Value::Object(row)
        })
        .collect();
    m.insert("ruleCounters".to_string(), Value::Array(counters));
    if let Some(e) = sample_error {
        m.insert("sampleError".to_string(), json!(e));
    }
    if let Some(r) = run_id {
        m.insert("runId".to_string(), json!(r));
    }
    if let Some(c) = &spec.correlation {
        m.insert("correlation".to_string(), serde_json::to_value(c)?);
    }
    Ok(Value::Object(m))
}

/// Per-TAP statistics row for [`observability_host_tap_data_v1`].
#[derive(Debug, Default, Clone, Copy)]
pub struct TapStats {
    pub rx_packets: Option<u64>,
    pub tx_packets: Option<u64>,
    pub rx_bytes: Option<u64>,
    pub tx_bytes: Option<u64>,
    pub rx_errors: Option<u64>,
    pub tx_errors: Option<u64>,
    pub rx_dropped: Option<u64>,
    pub tx_dropped: Option<u64>,
}

/// `dev.cellos.events.cell.observability.host.v1.tap`
///
/// Per-cell TAP-link snapshot for the FC microVM tap device. Path B
/// (ADR-0006).
///
/// Schema: `contracts/schemas/cell-observability-host-tap-v1.schema.json`.
#[allow(clippy::too_many_arguments)]
pub fn observability_host_tap_data_v1(
    spec: &ExecutionCellSpec,
    cell_id: &str,
    run_id: Option<&str>,
    spec_signature_hash: &str,
    sampled_at_unix_ms: u64,
    tap_name: &str,
    link_state: &str,
    stats: &TapStats,
    sample_error: Option<&str>,
) -> Result<Value, serde_json::Error> {
    let mut m = Map::new();
    m.insert("cellId".to_string(), json!(cell_id));
    m.insert("specId".to_string(), json!(&spec.id));
    m.insert("specSignatureHash".to_string(), json!(spec_signature_hash));
    m.insert("sampledAtUnixMs".to_string(), json!(sampled_at_unix_ms));
    m.insert("tapName".to_string(), json!(tap_name));
    m.insert("linkState".to_string(), json!(link_state));
    if let Some(v) = stats.rx_packets {
        m.insert("rxPackets".to_string(), json!(v));
    }
    if let Some(v) = stats.tx_packets {
        m.insert("txPackets".to_string(), json!(v));
    }
    if let Some(v) = stats.rx_bytes {
        m.insert("rxBytes".to_string(), json!(v));
    }
    if let Some(v) = stats.tx_bytes {
        m.insert("txBytes".to_string(), json!(v));
    }
    if let Some(v) = stats.rx_errors {
        m.insert("rxErrors".to_string(), json!(v));
    }
    if let Some(v) = stats.tx_errors {
        m.insert("txErrors".to_string(), json!(v));
    }
    if let Some(v) = stats.rx_dropped {
        m.insert("rxDropped".to_string(), json!(v));
    }
    if let Some(v) = stats.tx_dropped {
        m.insert("txDropped".to_string(), json!(v));
    }
    if let Some(e) = sample_error {
        m.insert("sampleError".to_string(), json!(e));
    }
    if let Some(r) = run_id {
        m.insert("runId".to_string(), json!(r));
    }
    if let Some(c) = &spec.correlation {
        m.insert("correlation".to_string(), serde_json::to_value(c)?);
    }
    Ok(Value::Object(m))
}

/// Final residue class for the cell after teardown
/// (per `docs/destruction-semantics.md`).
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum ResidueClass {
    /// No host- or broker-side residue tracked for the cell after destroy.
    None,
    /// A residue class with a named runbook entry; see
    /// [`EvidenceBundleRefs::residue_exception`] for the runbook id.
    DocumentedException,
}

/// Coarser-grained residue class historically embedded directly in the
/// `lifecycle.v1.destroyed` payload. Retained for back-compat with consumers
/// that read the destruction event directly without joining the
/// `evidence_bundle.v1.emitted` payload. New code should prefer
/// [`ResidueClass`] on the bundle.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
#[serde(rename_all = "camelCase")]
pub enum LifecycleResidueClass {
    None,
    MetadataOnly,
    DocumentedException,
    Unknown,
}

/// References the supervisor aggregates into an `evidence_bundle` payload.
///
/// All identifiers are CloudEvent envelope `id` strings (URN preferred).
/// Required fields enforce ADR-0006 §F1: every emitted bundle ties back to
/// a destruction event (D5) and a host-stamped spec hash (the channel's
/// authenticity primitive). Optional refs are populated only when the
/// supervisor actually emitted the corresponding event for this run —
/// builders never fabricate references.
#[derive(Debug, Default, Clone)]
pub struct EvidenceBundleRefs<'a> {
    /// Required: CloudEvent id of `cell.lifecycle.v1.started`.
    pub started_event_ref: &'a str,
    /// Required: CloudEvent id of `cell.lifecycle.v1.destroyed` (D5).
    pub cell_destroyed_event_ref: &'a str,
    pub command_completed_event_ref: Option<&'a str>,
    pub spawned_event_refs: &'a [&'a str],
    pub fc_metrics_event_refs: &'a [&'a str],
    pub cgroup_event_refs: &'a [&'a str],
    pub nftables_event_refs: &'a [&'a str],
    pub tap_event_refs: &'a [&'a str],
    pub homeostasis_event_ref: Option<&'a str>,
    pub compliance_summary_event_ref: Option<&'a str>,
    /// One row per declared `cell.observability.guest.*` event:
    /// `(eventId, eventType, ruleClass)`.
    pub guest_event_refs: &'a [(&'a str, &'a str, &'a str)],
    /// When `residue_class == DocumentedException`, names the runbook entry.
    pub residue_exception: Option<&'a str>,
}

/// `dev.cellos.events.cell.evidence_bundle.v1.emitted`
///
/// The 1.0 per-cell aggregation primitive (ADR-0006 §4) — what the auditor
/// reads cold and what the CI customer pastes into incident review.
/// Carries the spec hash, lifecycle stream refs, Path B host-side resource
/// series refs, declared guest event refs (with ADG rule-class), and the
/// destruction receipt naming the residue class.
///
/// **D5 gate:** `cell_destroyed_event_ref` is required by the schema and by
/// this builder's signature. A bundle without a destruction event reference
/// is rejected by the schema validator and by `tests/residue.rs`
/// (per ADR-0006 §Compliance "destruction gate").
///
/// Schema: `contracts/schemas/evidence-bundle-v1.schema.json`.
pub fn evidence_bundle_emitted_data_v1(
    spec: &ExecutionCellSpec,
    cell_id: &str,
    run_id: Option<&str>,
    spec_signature_hash: &str,
    emitted_at_unix_ms: u64,
    residue_class: ResidueClass,
    refs: &EvidenceBundleRefs<'_>,
) -> Result<Value, serde_json::Error> {
    let mut m = Map::new();
    m.insert("cellId".to_string(), json!(cell_id));
    m.insert("specId".to_string(), json!(&spec.id));
    m.insert("specSignatureHash".to_string(), json!(spec_signature_hash));
    m.insert("emittedAtUnixMs".to_string(), json!(emitted_at_unix_ms));

    let mut lifecycle = Map::new();
    lifecycle.insert("started".to_string(), json!(refs.started_event_ref));
    lifecycle.insert(
        "destroyed".to_string(),
        json!(refs.cell_destroyed_event_ref),
    );
    if let Some(cc) = refs.command_completed_event_ref {
        lifecycle.insert("commandCompleted".to_string(), json!(cc));
    }
    if !refs.spawned_event_refs.is_empty() {
        lifecycle.insert("spawned".to_string(), json!(refs.spawned_event_refs));
    }
    m.insert("lifecycleEventRefs".to_string(), Value::Object(lifecycle));

    m.insert(
        "cellDestroyedEventRef".to_string(),
        json!(refs.cell_destroyed_event_ref),
    );
    m.insert(
        "residueClass".to_string(),
        serde_json::to_value(residue_class)?,
    );
    if let Some(rx) = refs.residue_exception {
        m.insert("residueException".to_string(), json!(rx));
    }

    let any_host = !refs.fc_metrics_event_refs.is_empty()
        || !refs.cgroup_event_refs.is_empty()
        || !refs.nftables_event_refs.is_empty()
        || !refs.tap_event_refs.is_empty();
    if any_host {
        let mut host = Map::new();
        if !refs.fc_metrics_event_refs.is_empty() {
            host.insert("fcMetrics".to_string(), json!(refs.fc_metrics_event_refs));
        }
        if !refs.cgroup_event_refs.is_empty() {
            host.insert("cgroup".to_string(), json!(refs.cgroup_event_refs));
        }
        if !refs.nftables_event_refs.is_empty() {
            host.insert("nftables".to_string(), json!(refs.nftables_event_refs));
        }
        if !refs.tap_event_refs.is_empty() {
            host.insert("tap".to_string(), json!(refs.tap_event_refs));
        }
        m.insert("hostProbeEventRefs".to_string(), Value::Object(host));
    }

    if !refs.guest_event_refs.is_empty() {
        let rows: Vec<Value> = refs
            .guest_event_refs
            .iter()
            .map(|(id, ty, rc)| {
                let mut row = Map::new();
                row.insert("eventId".to_string(), json!(id));
                row.insert("eventType".to_string(), json!(ty));
                row.insert("ruleClass".to_string(), json!(rc));
                Value::Object(row)
            })
            .collect();
        m.insert("guestEventRefs".to_string(), Value::Array(rows));
    }

    if let Some(h) = refs.homeostasis_event_ref {
        m.insert("homeostasisEventRef".to_string(), json!(h));
    }
    if let Some(c) = refs.compliance_summary_event_ref {
        m.insert("complianceSummaryEventRef".to_string(), json!(c));
    }
    if let Some(r) = run_id {
        m.insert("runId".to_string(), json!(r));
    }
    if let Some(c) = &spec.correlation {
        m.insert("correlation".to_string(), serde_json::to_value(c)?);
    }
    Ok(Value::Object(m))
}

/// `data` for event type `dev.cellos.events.cell.cortex.v1.dispatched`.
///
/// Emitted by `CellosLedgerEmitter` when it processes a Cortex
/// ContextPack dispatch onto a cell. Records the pack id, the cell id,
/// the doctrine refs cited by the pack (subject to ADR-0009 policy), and
/// the Cortex↔CellOS bridge version for downstream auditors.
///
/// Schema: `contracts/schemas/cell-cortex-dispatched-v1.schema.json`.
pub fn cortex_dispatched_data_v1(pack_id: &str, cell_id: &str, doctrine_refs: &[String]) -> Value {
    let mut m = Map::new();
    m.insert("packId".to_string(), json!(pack_id));
    m.insert("cellId".to_string(), json!(cell_id));
    m.insert("doctrineRefs".to_string(), json!(doctrine_refs));
    m.insert("bridgeVersion".to_string(), json!("1.0"));
    Value::Object(m)
}

/// `data` for event type `dev.cellos.events.cell.firecracker.v1.pool_checkout`.
///
/// Emitted by `FirecrackerCellBackend` when it satisfies a cell start by
/// either consuming a slot from the warm pool (`poolHit = true`) or
/// falling through to a cold boot (`poolHit = false`). `slotCount` is the
/// number of warm slots available *before* this checkout — useful for
/// detecting pool-exhaustion patterns in audit.
///
/// Schema: `contracts/schemas/cell-firecracker-pool-checkout-v1.schema.json`.
pub fn firecracker_pool_event_data_v1(cell_id: &str, pool_hit: bool, slot_count: usize) -> Value {
    let mut m = Map::new();
    m.insert("cellId".to_string(), json!(cell_id));
    m.insert("poolHit".to_string(), json!(pool_hit));
    m.insert("slotCount".to_string(), json!(slot_count));
    Value::Object(m)
}

/// CloudEvent envelope constructor for `dev.cellos.events.cell.cortex.v1.dispatched`.
///
/// Pairs with [`cortex_dispatched_data_v1`]. `source` is typically the Cortex
/// bridge identifier (e.g. `"cellos-cortex"`); `time` is RFC 3339.
///
/// Schema: `contracts/schemas/cell-cortex-dispatched-v1.schema.json`.
pub fn cloud_event_v1_cortex_dispatched(
    source: &str,
    time: &str,
    pack_id: &str,
    cell_id: &str,
    doctrine_refs: &[String],
) -> CloudEventV1 {
    CloudEventV1 {
        specversion: "1.0".into(),
        id: uuid::Uuid::new_v4().to_string(),
        source: source.to_string(),
        ty: "dev.cellos.events.cell.cortex.v1.dispatched".into(),
        datacontenttype: Some("application/json".into()),
        data: Some(cortex_dispatched_data_v1(pack_id, cell_id, doctrine_refs)),
        time: Some(time.to_string()),
        traceparent: None,
        cex: None,
    }
}

/// CloudEvent envelope constructor for `dev.cellos.events.cell.firecracker.v1.pool_checkout`.
///
/// Pairs with [`firecracker_pool_event_data_v1`]. `source` is typically
/// `"cellos-host-firecracker"`; `time` is RFC 3339. `slot_count` is the
/// pool's `Available` count observed *before* the checkout attempt — see
/// the data builder for audit semantics.
///
/// Schema: `contracts/schemas/cell-firecracker-pool-checkout-v1.schema.json`.
pub fn cloud_event_v1_firecracker_pool_checkout(
    source: &str,
    time: &str,
    cell_id: &str,
    pool_hit: bool,
    slot_count: usize,
) -> CloudEventV1 {
    CloudEventV1 {
        specversion: "1.0".into(),
        id: uuid::Uuid::new_v4().to_string(),
        source: source.to_string(),
        ty: "dev.cellos.events.cell.firecracker.v1.pool_checkout".into(),
        datacontenttype: Some("application/json".into()),
        data: Some(firecracker_pool_event_data_v1(
            cell_id, pool_hit, slot_count,
        )),
        time: Some(time.to_string()),
        traceparent: None,
        cex: None,
    }
}

// ─────────────────────────────────────────────────────────────────────────────
// Formation-level lifecycle events (Session 15+16 — Formation model).
//
// The Formation state machine governs a group of related cells managed as a
// single unit. Its transitions are:
//
//   PENDING → LAUNCHING → RUNNING → DEGRADED → COMPLETED / FAILED
//
// Each transition emits a CloudEvent so the formation lifecycle is auditable
// the same way cell lifecycle is auditable. The URN namespace is
// `dev.cellos.events.cell.formation.v1.<phase>`, sharing the `cell.` prefix
// so the FC-74 audit consumer's coverage rules apply uniformly. The shared
// data shape is described by [`FormationEventData`] below.
// ─────────────────────────────────────────────────────────────────────────────

/// CloudEvent `type` URN for [`cloud_event_v1_formation_created`].
pub const FORMATION_CREATED_TYPE: &str = "dev.cellos.events.cell.formation.v1.created";

/// CloudEvent `type` URN for [`cloud_event_v1_formation_launching`].
pub const FORMATION_LAUNCHING_TYPE: &str = "dev.cellos.events.cell.formation.v1.launching";

/// CloudEvent `type` URN for [`cloud_event_v1_formation_running`].
pub const FORMATION_RUNNING_TYPE: &str = "dev.cellos.events.cell.formation.v1.running";

/// CloudEvent `type` URN for [`cloud_event_v1_formation_degraded`].
pub const FORMATION_DEGRADED_TYPE: &str = "dev.cellos.events.cell.formation.v1.degraded";

/// CloudEvent `type` URN for [`cloud_event_v1_formation_completed`].
pub const FORMATION_COMPLETED_TYPE: &str = "dev.cellos.events.cell.formation.v1.completed";

/// CloudEvent `type` URN for [`cloud_event_v1_formation_failed`].
pub const FORMATION_FAILED_TYPE: &str = "dev.cellos.events.cell.formation.v1.failed";

/// Build the canonical `data` payload for every
/// `dev.cellos.events.cell.formation.v1.*` CloudEvent.
///
/// All formation lifecycle events share a single shape so downstream auditors
/// (the FC-74 consumer, `cellos-audit-justification`, SIEM ingest) can join
/// on `formationId` across phases without dispatching per-URN parsers.
///
/// Fields:
/// - `formationId` — stable identifier of the Formation document.
/// - `formationName` — operator-facing label (mirrors `metadata.name` on the
///   FormationDocument).
/// - `cellCount` — total number of cells in the formation at the time of
///   transition. Always present so audit can detect mid-run resize.
/// - `failedCellIds` — concrete cell IDs that contributed to the transition
///   (populated for `degraded` / `failed`, empty for the happy path).
/// - `reason` — short, free-form explanation for `degraded` / `failed`
///   transitions. Omitted on the happy path.
fn formation_data_v1(
    formation_id: &str,
    formation_name: &str,
    cell_count: u32,
    failed_cell_ids: &[String],
    reason: Option<&str>,
) -> Value {
    let mut m = Map::new();
    m.insert("formationId".to_string(), json!(formation_id));
    m.insert("formationName".to_string(), json!(formation_name));
    m.insert("cellCount".to_string(), json!(cell_count));
    m.insert("failedCellIds".to_string(), json!(failed_cell_ids));
    if let Some(r) = reason {
        m.insert("reason".to_string(), json!(r));
    }
    Value::Object(m)
}

/// CloudEvent envelope constructor for
/// `dev.cellos.events.cell.formation.v1.created` (PENDING phase entry).
///
/// Emitted by the supervisor immediately after a FormationDocument is
/// admitted but before any cell launch has begun. `failed_cell_ids` is
/// expected to be empty on this transition.
pub fn cloud_event_v1_formation_created(
    source: &str,
    time: &str,
    formation_id: &str,
    formation_name: &str,
    cell_count: u32,
    failed_cell_ids: &[String],
    reason: Option<&str>,
) -> CloudEventV1 {
    CloudEventV1 {
        specversion: "1.0".into(),
        id: uuid::Uuid::new_v4().to_string(),
        source: source.to_string(),
        ty: FORMATION_CREATED_TYPE.to_string(),
        datacontenttype: Some("application/json".into()),
        data: Some(formation_data_v1(
            formation_id,
            formation_name,
            cell_count,
            failed_cell_ids,
            reason,
        )),
        time: Some(time.to_string()),
        traceparent: None,
        cex: None,
    }
}

/// CloudEvent envelope constructor for
/// `dev.cellos.events.cell.formation.v1.launching` (LAUNCHING phase entry).
///
/// Emitted once the supervisor has started attempting to bring up the
/// formation's cells. `failed_cell_ids` is expected to be empty on this
/// transition; transient launch errors that don't fail the whole formation
/// belong on the subsequent `degraded` or `failed` events.
pub fn cloud_event_v1_formation_launching(
    source: &str,
    time: &str,
    formation_id: &str,
    formation_name: &str,
    cell_count: u32,
    failed_cell_ids: &[String],
    reason: Option<&str>,
) -> CloudEventV1 {
    CloudEventV1 {
        specversion: "1.0".into(),
        id: uuid::Uuid::new_v4().to_string(),
        source: source.to_string(),
        ty: FORMATION_LAUNCHING_TYPE.to_string(),
        datacontenttype: Some("application/json".into()),
        data: Some(formation_data_v1(
            formation_id,
            formation_name,
            cell_count,
            failed_cell_ids,
            reason,
        )),
        time: Some(time.to_string()),
        traceparent: None,
        cex: None,
    }
}

/// CloudEvent envelope constructor for
/// `dev.cellos.events.cell.formation.v1.running` (RUNNING phase entry).
///
/// Emitted once all of the formation's cells reach a healthy steady state.
/// `failed_cell_ids` is expected to be empty on this transition.
pub fn cloud_event_v1_formation_running(
    source: &str,
    time: &str,
    formation_id: &str,
    formation_name: &str,
    cell_count: u32,
    failed_cell_ids: &[String],
    reason: Option<&str>,
) -> CloudEventV1 {
    CloudEventV1 {
        specversion: "1.0".into(),
        id: uuid::Uuid::new_v4().to_string(),
        source: source.to_string(),
        ty: FORMATION_RUNNING_TYPE.to_string(),
        datacontenttype: Some("application/json".into()),
        data: Some(formation_data_v1(
            formation_id,
            formation_name,
            cell_count,
            failed_cell_ids,
            reason,
        )),
        time: Some(time.to_string()),
        traceparent: None,
        cex: None,
    }
}

/// CloudEvent envelope constructor for
/// `dev.cellos.events.cell.formation.v1.degraded` (DEGRADED phase entry).
///
/// Emitted when at least one but not all of the formation's cells have
/// failed and the formation is continuing to run in a degraded state.
/// `failed_cell_ids` MUST list the affected cells; `reason` SHOULD be set.
pub fn cloud_event_v1_formation_degraded(
    source: &str,
    time: &str,
    formation_id: &str,
    formation_name: &str,
    cell_count: u32,
    failed_cell_ids: &[String],
    reason: Option<&str>,
) -> CloudEventV1 {
    CloudEventV1 {
        specversion: "1.0".into(),
        id: uuid::Uuid::new_v4().to_string(),
        source: source.to_string(),
        ty: FORMATION_DEGRADED_TYPE.to_string(),
        datacontenttype: Some("application/json".into()),
        data: Some(formation_data_v1(
            formation_id,
            formation_name,
            cell_count,
            failed_cell_ids,
            reason,
        )),
        time: Some(time.to_string()),
        traceparent: None,
        cex: None,
    }
}

/// CloudEvent envelope constructor for
/// `dev.cellos.events.cell.formation.v1.completed` (COMPLETED terminal).
///
/// Emitted when the formation has finished its work successfully and all
/// cells have been torn down cleanly. `failed_cell_ids` is expected to be
/// empty; `reason` MAY carry an operator-facing completion note.
pub fn cloud_event_v1_formation_completed(
    source: &str,
    time: &str,
    formation_id: &str,
    formation_name: &str,
    cell_count: u32,
    failed_cell_ids: &[String],
    reason: Option<&str>,
) -> CloudEventV1 {
    CloudEventV1 {
        specversion: "1.0".into(),
        id: uuid::Uuid::new_v4().to_string(),
        source: source.to_string(),
        ty: FORMATION_COMPLETED_TYPE.to_string(),
        datacontenttype: Some("application/json".into()),
        data: Some(formation_data_v1(
            formation_id,
            formation_name,
            cell_count,
            failed_cell_ids,
            reason,
        )),
        time: Some(time.to_string()),
        traceparent: None,
        cex: None,
    }
}

/// CloudEvent envelope constructor for
/// `dev.cellos.events.cell.formation.v1.failed` (FAILED terminal).
///
/// Emitted when the formation has terminated unsuccessfully. `failed_cell_ids`
/// MUST list the cells that drove the failure (callers MAY include the full
/// cell set when the failure is formation-wide); `reason` SHOULD be set.
pub fn cloud_event_v1_formation_failed(
    source: &str,
    time: &str,
    formation_id: &str,
    formation_name: &str,
    cell_count: u32,
    failed_cell_ids: &[String],
    reason: Option<&str>,
) -> CloudEventV1 {
    CloudEventV1 {
        specversion: "1.0".into(),
        id: uuid::Uuid::new_v4().to_string(),
        source: source.to_string(),
        ty: FORMATION_FAILED_TYPE.to_string(),
        datacontenttype: Some("application/json".into()),
        data: Some(formation_data_v1(
            formation_id,
            formation_name,
            cell_count,
            failed_cell_ids,
            reason,
        )),
        time: Some(time.to_string()),
        traceparent: None,
        cex: None,
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::{
        Correlation, ExecutionCellDocument, ExportReceipt, ExportReceiptTargetKind, Lifetime,
        WorkloadIdentity, WorkloadIdentityKind,
    };

    #[test]
    fn admission_receipt_stamps_provider_identity() {
        // S09: every admission decision records the active crypto provider so an
        // auditor sees which module signed it. fipsMode is a producer claim.
        let spec = crate::ExecutionCellSpec::default();
        let data = admission_decision_data_v1(
            &spec,
            true,
            AdmissionDecisionReason::AdmittedWithinCeiling,
            "ceiling.is_superset_of(spec.authority)",
            "sha256:spechash",
            Some("run-1"),
            EnforcementBinding::DeclaredAudit,
            &[],
        )
        .unwrap();
        let pid = data
            .get("providerIdentity")
            .expect("admission receipt must stamp providerIdentity");
        assert!(
            pid.get("name").and_then(|v| v.as_str()).is_some(),
            "providerIdentity carries a name"
        );
        assert!(
            pid.get("fipsMode").and_then(|v| v.as_bool()).is_some(),
            "providerIdentity carries fipsMode (producer claim)"
        );
    }

    #[test]
    fn audit_reconciled_data_v1_is_deterministic_and_verifies_offline() {
        // S18: the reconcile receipt builder is deterministic and the wrapped
        // CloudEvent is offline-verifiable under the org-root key.
        use crate::{
            sign_event_with, verify_signed_event_envelope, CloudEventV1, Signer, SoftwareSigner,
        };
        use std::collections::HashMap;

        let a = audit_reconciled_data_v1(
            "chain-A",
            0,
            41,
            "abc123",
            42,
            Some("abc123"),
            "blake3-chain+ed25519-head",
            true,
        )
        .unwrap();
        let b = audit_reconciled_data_v1(
            "chain-A",
            0,
            41,
            "abc123",
            42,
            Some("abc123"),
            "blake3-chain+ed25519-head",
            true,
        )
        .unwrap();
        assert_eq!(
            serde_json::to_vec(&a).unwrap(),
            serde_json::to_vec(&b).unwrap(),
            "builder must be deterministic"
        );
        assert_eq!(a["toSeq"], serde_json::json!(41));
        assert_eq!(a["signed"], serde_json::json!(true));

        let event = CloudEventV1 {
            specversion: "1.0".into(),
            id: "rec-1".into(),
            source: "cellos-sink-spool".into(),
            ty: AUDIT_RECONCILED_TYPE.into(),
            datacontenttype: Some("application/json".into()),
            data: Some(a),
            time: None,
            traceparent: None,
            cex: None,
        };
        let signer = SoftwareSigner::from_seed("org-root-2026", [5u8; 32]).unwrap();
        let envelope = sign_event_with(&signer, &event).unwrap();
        let mut keys = HashMap::new();
        keys.insert("org-root-2026".to_string(), signer.verifying_key());
        let hmac: HashMap<String, Vec<u8>> = HashMap::new();
        let verified = verify_signed_event_envelope(&envelope, &keys, &hmac).unwrap();
        assert_eq!(verified.ty, AUDIT_RECONCILED_TYPE);
    }

    #[test]
    fn lifecycle_started_matches_example_shape() {
        let raw =
            include_str!("../../../contracts/examples/execution-cell-ci-correlation.valid.json");
        let doc: ExecutionCellDocument = serde_json::from_str(raw).unwrap();
        let expected: Value = serde_json::from_str(include_str!(
            "../../../contracts/examples/cell-lifecycle-started-data.valid.json"
        ))
        .unwrap();
        let data = lifecycle_started_data_v1(
            &doc.spec,
            "host-cell-abc123",
            Some("run-2026-04-06-001"),
            None,
            None,
            None,
            None,
            None,
            None,
            None,
        )
        .unwrap();
        assert_eq!(data, expected);
    }

    #[test]
    fn lifecycle_started_without_correlation() {
        let spec = ExecutionCellSpec {
            id: "s1".into(),
            correlation: None,
            ingress: None,
            environment: None,
            placement: None,
            policy: None,
            identity: None,
            run: None,
            authority: Default::default(),
            lifetime: Lifetime { ttl_seconds: 60 },
            export: None,
            telemetry: None,
        };
        let data =
            lifecycle_started_data_v1(&spec, "c1", None, None, None, None, None, None, None, None)
                .unwrap();
        assert!(!data.as_object().unwrap().contains_key("correlation"));
        assert!(!data.as_object().unwrap().contains_key("runId"));
        assert!(!data.as_object().unwrap().contains_key("derivationVerified"));
        assert!(!data.as_object().unwrap().contains_key("roleRoot"));
        assert!(!data.as_object().unwrap().contains_key("parentRunId"));
        assert!(!data.as_object().unwrap().contains_key("kernelDigestSha256"));
        assert!(!data.as_object().unwrap().contains_key("rootfsDigestSha256"));
        assert!(!data
            .as_object()
            .unwrap()
            .contains_key("firecrackerDigestSha256"));
    }

    #[test]
    fn lifecycle_started_partial_correlation_serializes() {
        let spec = ExecutionCellSpec {
            id: "s2".into(),
            correlation: Some(Correlation {
                platform: Some("custom".into()),
                external_run_id: None,
                external_job_id: None,
                tenant_id: None,
                labels: None,
                correlation_id: None,
            }),
            ingress: None,
            environment: None,
            placement: None,
            policy: None,
            identity: None,
            run: None,
            authority: Default::default(),
            lifetime: Lifetime { ttl_seconds: 1 },
            export: None,
            telemetry: None,
        };
        let data =
            lifecycle_started_data_v1(&spec, "c2", None, None, None, None, None, None, None, None)
                .unwrap();
        assert_eq!(data["correlation"]["platform"], "custom");
    }

    #[test]
    fn lifecycle_started_with_derivation_fields_emits_them() {
        let spec = ExecutionCellSpec {
            id: "deriv-1".into(),
            correlation: None,
            ingress: None,
            environment: None,
            placement: None,
            policy: None,
            identity: None,
            run: None,
            authority: Default::default(),
            lifetime: Lifetime { ttl_seconds: 60 },
            export: None,
            telemetry: None,
        };
        let data = lifecycle_started_data_v1(
            &spec,
            "cell-deriv",
            Some("run-deriv-1"),
            Some(false),
            Some("role-prod-ci"),
            Some("run-parent-001"),
            Some("abc123def456"),
            None,
            None,
            None,
        )
        .unwrap();
        assert_eq!(data["derivationVerified"], false);
        assert_eq!(data["roleRoot"], "role-prod-ci");
        assert_eq!(data["parentRunId"], "run-parent-001");
    }

    #[test]
    fn lifecycle_destroyed_succeeded_shape() {
        let raw =
            include_str!("../../../contracts/examples/execution-cell-ci-correlation.valid.json");
        let doc: ExecutionCellDocument = serde_json::from_str(raw).unwrap();
        let data = lifecycle_destroyed_data_v1(
            &doc.spec,
            "host-xyz",
            Some("run-test"),
            LifecycleDestroyOutcome::Succeeded,
            None,
            None,
            None,
            None,
        )
        .unwrap();
        assert_eq!(data["outcome"], "succeeded");
        assert!(!data.as_object().unwrap().contains_key("reason"));
        assert!(
            !data.as_object().unwrap().contains_key("terminalState"),
            "terminalState must be omitted when None for backward-compat"
        );
        assert!(
            !data.as_object().unwrap().contains_key("evidenceBundleRef"),
            "evidenceBundleRef must be omitted when None for backward-compat"
        );
        assert!(
            !data.as_object().unwrap().contains_key("residueClass"),
            "residueClass must be omitted when None for backward-compat"
        );
        assert_eq!(data["ttlSeconds"], 3600);
    }

    #[test]
    fn lifecycle_destroyed_failed_includes_reason() {
        let spec = ExecutionCellSpec {
            id: "s1".into(),
            correlation: None,
            ingress: None,
            environment: None,
            placement: None,
            policy: None,
            identity: None,
            run: None,
            authority: Default::default(),
            lifetime: Lifetime { ttl_seconds: 60 },
            export: None,
            telemetry: None,
        };
        let data = lifecycle_destroyed_data_v1(
            &spec,
            "c1",
            None,
            LifecycleDestroyOutcome::Failed,
            Some("secret resolve: denied"),
            None,
            None,
            None,
        )
        .unwrap();
        assert_eq!(data["outcome"], "failed");
        assert_eq!(data["reason"], "secret resolve: denied");
    }

    #[test]
    fn admission_decision_reject_shape_and_determinism() {
        let raw =
            include_str!("../../../contracts/examples/execution-cell-ci-correlation.valid.json");
        let doc: ExecutionCellDocument = serde_json::from_str(raw).unwrap();

        let build = || {
            admission_decision_data_v1(
                &doc.spec,
                false,
                AdmissionDecisionReason::AuthorityExceedsCeiling,
                "spec.authority ⊆ ceiling.authority",
                "sha256:abc",
                Some("run-001"),
                EnforcementBinding::LinuxNftables,
                &[("environment", "ci")],
            )
            .unwrap()
        };

        let data = build();
        assert_eq!(data["admitted"], false);
        assert_eq!(data["reason"], "authority_exceeds_ceiling");
        assert_eq!(data["predicate"], "spec.authority ⊆ ceiling.authority");
        assert_eq!(data["specHash"], "sha256:abc");
        assert_eq!(data["runId"], "run-001");
        assert_eq!(data["enforcementBinding"], "linux-nftables");
        assert_eq!(data["context"]["environment"], "ci");

        // reason is exactly one of the closed snake_case discriminants.
        let closed = [
            "authority_exceeds_ceiling",
            "ceiling_manifest_unverified",
            "ceiling_manifest_unparseable",
            "ceiling_manifest_wrong_payload_type",
            "ceiling_dimension_uncomparable",
            "ceiling_epoch_stale",
            "ceiling_degraded_unconfigured",
            "formation_edge_not_narrowing",
            "admitted_within_ceiling",
        ];
        assert!(closed.contains(&data["reason"].as_str().unwrap()));

        // Two builds of the same verdict are byte-identical (canonical payload).
        let a = serde_json::to_vec(&build()).unwrap();
        let b = serde_json::to_vec(&build()).unwrap();
        assert_eq!(a, b);
    }

    #[test]
    fn admission_decision_allow_omits_optional_fields() {
        let spec = ExecutionCellSpec {
            id: "s1".into(),
            correlation: None,
            ingress: None,
            environment: None,
            placement: None,
            policy: None,
            identity: None,
            run: None,
            authority: Default::default(),
            lifetime: Lifetime { ttl_seconds: 60 },
            export: None,
            telemetry: None,
        };
        let data = admission_decision_data_v1(
            &spec,
            true,
            AdmissionDecisionReason::AdmittedWithinCeiling,
            "spec.authority ⊆ ceiling.authority",
            "sha256:def",
            None,
            EnforcementBinding::DeclaredAudit,
            &[],
        )
        .unwrap();
        assert_eq!(data["admitted"], true);
        assert_eq!(data["reason"], "admitted_within_ceiling");
        assert_eq!(data["enforcementBinding"], "declared-audit");
        let obj = data.as_object().unwrap();
        assert!(!obj.contains_key("runId"), "runId omitted when None");
        assert!(!obj.contains_key("context"), "context omitted when empty");
        assert!(
            !obj.contains_key("correlation"),
            "correlation omitted when None"
        );
    }

    #[test]
    fn lifecycle_destroyed_terminal_state_clean_serializes() {
        let spec = ExecutionCellSpec {
            id: "term-clean".into(),
            correlation: None,
            ingress: None,
            environment: None,
            placement: None,
            policy: None,
            identity: None,
            run: None,
            authority: Default::default(),
            lifetime: Lifetime { ttl_seconds: 60 },
            export: None,
            telemetry: None,
        };
        let data = lifecycle_destroyed_data_v1(
            &spec,
            "c-clean",
            None,
            LifecycleDestroyOutcome::Succeeded,
            None,
            Some(LifecycleTerminalState::Clean),
            None,
            None,
        )
        .unwrap();
        assert_eq!(data["terminalState"], "clean");
    }

    #[test]
    fn lifecycle_destroyed_terminal_state_forced_serializes() {
        let spec = ExecutionCellSpec {
            id: "term-forced".into(),
            correlation: None,
            ingress: None,
            environment: None,
            placement: None,
            policy: None,
            identity: None,
            run: None,
            authority: Default::default(),
            lifetime: Lifetime { ttl_seconds: 60 },
            export: None,
            telemetry: None,
        };
        let data = lifecycle_destroyed_data_v1(
            &spec,
            "c-forced",
            None,
            LifecycleDestroyOutcome::Failed,
            Some("in-VM exit bridge: vsock closed"),
            Some(LifecycleTerminalState::Forced),
            None,
            None,
        )
        .unwrap();
        assert_eq!(data["terminalState"], "forced");
        assert_eq!(data["outcome"], "failed");
    }

    #[test]
    fn lifecycle_destroyed_evidence_bundle_and_residue_class_serialize_when_populated() {
        let spec = ExecutionCellSpec {
            id: "f5-populated".into(),
            correlation: None,
            ingress: None,
            environment: None,
            placement: None,
            policy: None,
            identity: None,
            run: None,
            authority: Default::default(),
            lifetime: Lifetime { ttl_seconds: 60 },
            export: None,
            telemetry: None,
        };
        let bundle = SubjectUrn::parse("urn:cellos:evidence-bundle:run-1").unwrap();
        let data = lifecycle_destroyed_data_v1(
            &spec,
            "c-f5",
            Some("run-1"),
            LifecycleDestroyOutcome::Succeeded,
            None,
            None,
            Some(&bundle),
            Some(ResidueClass::DocumentedException),
        )
        .unwrap();
        assert_eq!(
            data["evidenceBundleRef"],
            "urn:cellos:evidence-bundle:run-1"
        );
        assert_eq!(data["residueClass"], "documented_exception");
    }

    #[test]
    fn lifecycle_destroyed_evidence_bundle_and_residue_class_omitted_when_none() {
        let spec = ExecutionCellSpec {
            id: "f5-omitted".into(),
            correlation: None,
            ingress: None,
            environment: None,
            placement: None,
            policy: None,
            identity: None,
            run: None,
            authority: Default::default(),
            lifetime: Lifetime { ttl_seconds: 60 },
            export: None,
            telemetry: None,
        };
        let data = lifecycle_destroyed_data_v1(
            &spec,
            "c-f5-omit",
            None,
            LifecycleDestroyOutcome::Succeeded,
            None,
            None,
            None,
            None,
        )
        .unwrap();
        let obj = data.as_object().unwrap();
        assert!(
            !obj.contains_key("evidenceBundleRef"),
            "evidenceBundleRef must be omitted when None"
        );
        assert!(
            !obj.contains_key("residueClass"),
            "residueClass must be omitted when None"
        );
    }

    #[test]
    fn identity_materialized_matches_example_shape() {
        let raw =
            include_str!("../../../contracts/examples/execution-cell-github-oidc-s3.valid.json");
        let doc: ExecutionCellDocument = serde_json::from_str(raw).unwrap();
        let identity = doc.spec.identity.as_ref().expect("identity");
        let data = identity_materialized_data_v1(&doc.spec, "host-xyz", Some("run-test"), identity)
            .unwrap();
        assert_eq!(data["identity"]["kind"], "federatedOidc");
        assert_eq!(data["identity"]["provider"], "github-actions");
        assert_eq!(data["identity"]["secretRef"], "AWS_WEB_IDENTITY");
        assert_eq!(data["runId"], "run-test");
    }

    #[test]
    fn identity_materialized_provisioned_witnesses_aws_role_arn() {
        let spec = ExecutionCellSpec {
            id: "s-aws".into(),
            correlation: None,
            ingress: None,
            environment: None,
            placement: None,
            policy: None,
            identity: Some(WorkloadIdentity {
                kind: WorkloadIdentityKind::AwsAssumeRole {
                    role_arn: "arn:aws:iam::123456789012:role/cell".into(),
                    sts_session_policy: None,
                },
                provider: String::new(),
                audience: String::new(),
                subject: None,
                ttl_seconds: Some(900),
                secret_ref: "AWS_WEB_IDENTITY".into(),
            }),
            run: None,
            authority: Default::default(),
            lifetime: Lifetime { ttl_seconds: 3600 },
            export: None,
            telemetry: None,
        };
        let identity = spec.identity.as_ref().unwrap();
        let witness = ProvisioningWitness {
            provisioned_role_arn: Some("arn:aws:iam::123456789012:role/cell"),
            session_policy_digest: Some(
                "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855",
            ),
            azure_object_id: None,
            credential_jti: Some("jti-abc"),
        };
        let data = identity_materialized_data_v1_provisioned(
            &spec,
            "host-aws",
            Some("run-aws"),
            identity,
            &witness,
        )
        .unwrap();
        assert_eq!(
            data["provisionedRoleArn"],
            "arn:aws:iam::123456789012:role/cell"
        );
        assert_eq!(
            data["sessionPolicyDigest"],
            "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855"
        );
        assert_eq!(data["credentialJti"], "jti-abc");
        let obj = data.as_object().unwrap();
        assert!(
            !obj.contains_key("azureObjectId"),
            "azureObjectId must be omitted when None"
        );
    }

    #[test]
    fn identity_materialized_provisioned_empty_witness_matches_base() {
        let raw =
            include_str!("../../../contracts/examples/execution-cell-github-oidc-s3.valid.json");
        let doc: ExecutionCellDocument = serde_json::from_str(raw).unwrap();
        let identity = doc.spec.identity.as_ref().expect("identity");
        let base = identity_materialized_data_v1(&doc.spec, "host-xyz", Some("run-test"), identity)
            .unwrap();
        let provisioned = identity_materialized_data_v1_provisioned(
            &doc.spec,
            "host-xyz",
            Some("run-test"),
            identity,
            &ProvisioningWitness::default(),
        )
        .unwrap();
        assert_eq!(base, provisioned);
    }

    #[test]
    fn identity_revoked_includes_reason() {
        let spec = ExecutionCellSpec {
            id: "s3".into(),
            correlation: None,
            ingress: None,
            environment: None,
            placement: None,
            policy: None,
            identity: Some(WorkloadIdentity {
                kind: WorkloadIdentityKind::FederatedOidc,
                provider: "github-actions".into(),
                audience: "sts.amazonaws.com".into(),
                subject: None,
                ttl_seconds: Some(900),
                secret_ref: "AWS_WEB_IDENTITY".into(),
            }),
            run: None,
            authority: Default::default(),
            lifetime: Lifetime { ttl_seconds: 3600 },
            export: None,
            telemetry: None,
        };
        let identity = spec.identity.as_ref().unwrap();
        let data =
            identity_revoked_data_v1(&spec, "c3", None, identity, Some("teardown"), None).unwrap();
        assert_eq!(data["identity"]["audience"], "sts.amazonaws.com");
        assert_eq!(data["reason"], "teardown");
    }

    #[test]
    fn identity_failed_matches_example_shape() {
        let raw =
            include_str!("../../../contracts/examples/execution-cell-github-oidc-s3.valid.json");
        let doc: ExecutionCellDocument = serde_json::from_str(raw).unwrap();
        let expected: Value = serde_json::from_str(include_str!(
            "../../../contracts/examples/cell-identity-failed-data.valid.json"
        ))
        .unwrap();
        let identity = doc.spec.identity.as_ref().expect("identity");
        let data = identity_failed_data_v1(
            &doc.spec,
            "host-cell-demo",
            Some("run-001"),
            identity,
            IdentityFailureOperation::Materialize,
            "oidc exchange denied by upstream federation policy",
        )
        .unwrap();
        assert_eq!(data, expected);
    }

    #[test]
    fn export_completed_v2_matches_example_shape() {
        let raw =
            include_str!("../../../contracts/examples/execution-cell-github-oidc-s3.valid.json");
        let doc: ExecutionCellDocument = serde_json::from_str(raw).unwrap();
        let receipt = ExportReceipt {
            target_kind: ExportReceiptTargetKind::S3,
            target_name: Some("artifact-bucket".into()),
            destination: "s3://acme-cellos-artifacts/github/acme/widget/123456789/test-results"
                .into(),
            bytes_written: 1024,
        };
        let data = export_completed_data_v2(
            &doc.spec,
            "host-xyz",
            Some("run-test"),
            "test-results",
            &receipt,
            None,
        )
        .unwrap();
        assert_eq!(data["receipt"]["targetKind"], "s3");
        assert_eq!(data["receipt"]["targetName"], "artifact-bucket");
        assert_eq!(data["receipt"]["bytesWritten"], 1024);
    }

    #[test]
    fn export_completed_v2_http_matches_example_shape() {
        let raw = include_str!(
            "../../../contracts/examples/execution-cell-github-oidc-multi-export.valid.json"
        );
        let doc: ExecutionCellDocument = serde_json::from_str(raw).unwrap();
        let expected: Value = serde_json::from_str(include_str!(
            "../../../contracts/examples/cell-export-v2-completed-data-http.valid.json"
        ))
        .unwrap();
        let receipt = ExportReceipt {
            target_kind: ExportReceiptTargetKind::Http,
            target_name: Some("artifact-api".into()),
            destination: "https://artifacts.acme.internal/upload/host-cell-demo/coverage-summary"
                .into(),
            bytes_written: 512,
        };
        let data = export_completed_data_v2(
            &doc.spec,
            "host-cell-demo",
            Some("run-002"),
            "coverage-summary",
            &receipt,
            None,
        )
        .unwrap();
        assert_eq!(data, expected);
    }

    #[test]
    fn export_failed_v2_http_matches_example_shape() {
        let raw = include_str!(
            "../../../contracts/examples/execution-cell-github-oidc-multi-export.valid.json"
        );
        let doc: ExecutionCellDocument = serde_json::from_str(raw).unwrap();
        let expected: Value = serde_json::from_str(include_str!(
            "../../../contracts/examples/cell-export-v2-failed-data.valid.json"
        ))
        .unwrap();
        let data = export_failed_data_v2(
            &doc.spec,
            "host-cell-demo",
            Some("run-002"),
            "coverage-summary",
            ExportReceiptTargetKind::Http,
            Some("artifact-api"),
            Some("https://artifacts.acme.internal/upload/host-cell-demo/coverage-summary"),
            "http put returned 403 Forbidden",
            None,
        )
        .unwrap();
        assert_eq!(data, expected);
    }

    #[test]
    fn compliance_summary_matches_example_shape() {
        let raw =
            include_str!("../../../contracts/examples/execution-cell-ci-correlation.valid.json");
        let doc: ExecutionCellDocument = serde_json::from_str(raw).unwrap();
        let expected: Value = serde_json::from_str(include_str!(
            "../../../contracts/examples/cell-compliance-summary-data.valid.json"
        ))
        .unwrap();
        let data =
            compliance_summary_data_v1(&doc.spec, "host-cell-demo", Some("run-003"), Some(0))
                .unwrap();
        assert_eq!(data, expected);
    }

    #[test]
    fn compliance_summary_omits_placement_when_absent() {
        let spec = ExecutionCellSpec {
            id: "compliance-no-placement".into(),
            correlation: None,
            ingress: None,
            environment: None,
            placement: None,
            policy: None,
            identity: None,
            run: None,
            authority: Default::default(),
            lifetime: Lifetime { ttl_seconds: 60 },
            export: None,
            telemetry: None,
        };
        let data = compliance_summary_data_v1(&spec, "cell-001", None, None).unwrap();
        assert!(!data.as_object().unwrap().contains_key("placement"));
    }

    /// P0-7 / G4: empty `subject_urns` slice MUST keep the payload byte-shape
    /// identical to the legacy 4-arg delegate — i.e. `subjectUrns` is omitted.
    /// This is the contract that lets supervisor.rs keep its current call
    /// site unchanged.
    #[test]
    fn compliance_summary_with_empty_subjects_omits_field() {
        let raw =
            include_str!("../../../contracts/examples/execution-cell-ci-correlation.valid.json");
        let doc: ExecutionCellDocument = serde_json::from_str(raw).unwrap();
        let legacy =
            compliance_summary_data_v1(&doc.spec, "host-cell-demo", Some("run-003"), Some(0))
                .unwrap();
        let with_empty = compliance_summary_data_v1_with_subjects(
            &doc.spec,
            "host-cell-demo",
            Some("run-003"),
            Some(0),
            &[],
        )
        .unwrap();
        assert_eq!(legacy, with_empty);
        assert!(!with_empty.as_object().unwrap().contains_key("subjectUrns"));
    }

    /// P0-7 / G4: non-empty subject set must round-trip against the
    /// `cell-compliance-summary-data-with-subjects.valid.json` golden fixture.
    #[test]
    fn compliance_summary_with_subjects_matches_example_shape() {
        let raw =
            include_str!("../../../contracts/examples/execution-cell-ci-correlation.valid.json");
        let doc: ExecutionCellDocument = serde_json::from_str(raw).unwrap();
        let expected: Value = serde_json::from_str(include_str!(
            "../../../contracts/examples/cell-compliance-summary-data-with-subjects.valid.json"
        ))
        .unwrap();
        let subjects: Vec<SubjectUrn> = vec![
            SubjectUrn::parse("urn:cellos:cell:host-cell-demo").unwrap(),
            SubjectUrn::parse("urn:tsafe:lease:lease-42").unwrap(),
            SubjectUrn::parse("urn:cellos:export:run-003%2Fartifact-1").unwrap(),
        ];
        let data = compliance_summary_data_v1_with_subjects(
            &doc.spec,
            "host-cell-demo",
            Some("run-003"),
            Some(0),
            &subjects,
        )
        .unwrap();
        assert_eq!(data, expected);
        let urns = data["subjectUrns"].as_array().unwrap();
        assert_eq!(urns.len(), 3);
        assert_eq!(urns[0], "urn:cellos:cell:host-cell-demo");
    }

    /// P0-7 / G4 negative fixture: `cell-compliance-summary-data.invalid.json`
    /// is intentionally malformed (URNs that violate the schema regex). The
    /// `scripts/validate_contracts.py` walker only globs `*.valid.json`, so we
    /// load the negative fixture explicitly here and hand-check each entry
    /// against the schema regex shape. If this test ever passes the regex,
    /// the fixture has drifted away from "negative" and must be regenerated.
    #[test]
    fn compliance_summary_invalid_subject_urns_fixture_is_malformed() {
        let raw =
            include_str!("../../../contracts/examples/cell-compliance-summary-data.invalid.json");
        let v: Value = serde_json::from_str(raw).unwrap();
        let urns = v["subjectUrns"]
            .as_array()
            .expect("invalid fixture must carry subjectUrns array");
        assert!(!urns.is_empty(), "negative fixture must have entries");

        // Schema regex from cell-compliance-summary-v1.schema.json. We don't
        // pull in a regex crate just for this — instead we apply a minimal
        // structural decomposition that the schema regex enforces:
        //   urn:<tool>:<kind>:<id>
        //   tool/kind = [a-z0-9][a-z0-9-]*  (lowercase, must start alnum)
        //   id        = non-empty, [A-Za-z0-9._:%-]+
        fn matches_schema_shape(s: &str) -> bool {
            let parts: Vec<&str> = s.splitn(4, ':').collect();
            if parts.len() != 4 {
                return false;
            }
            if parts[0] != "urn" {
                return false;
            }
            let segment_ok = |seg: &str| {
                let mut it = seg.chars();
                match it.next() {
                    Some(c) if c.is_ascii_lowercase() || c.is_ascii_digit() => {}
                    _ => return false,
                }
                it.all(|c| c.is_ascii_lowercase() || c.is_ascii_digit() || c == '-')
            };
            if !segment_ok(parts[1]) || !segment_ok(parts[2]) {
                return false;
            }
            if parts[3].is_empty() {
                return false;
            }
            parts[3]
                .chars()
                .all(|c| c.is_ascii_alphanumeric() || matches!(c, '.' | '_' | ':' | '%' | '-'))
        }

        for (i, urn) in urns.iter().enumerate() {
            let s = urn.as_str().unwrap_or("");
            assert!(
                !matches_schema_shape(s),
                "invalid fixture entry [{i}] {s:?} unexpectedly matches the schema URN regex; \
                 fixture must remain a negative case"
            );
        }
    }

    #[test]
    fn network_enforcement_matches_example_shape() {
        let raw = include_str!(
            "../../../contracts/examples/cell-observability-network-enforcement-data.valid.json"
        );
        let expected: Value = serde_json::from_str(raw).unwrap();
        let spec = ExecutionCellSpec {
            id: "net-enforcement-demo".into(),
            correlation: None,
            ingress: None,
            environment: None,
            placement: None,
            policy: None,
            identity: None,
            run: None,
            authority: Default::default(),
            lifetime: Lifetime { ttl_seconds: 60 },
            export: None,
            telemetry: None,
        };
        let data = observability_network_enforcement_data_v1(
            &spec,
            "net-enforcement-demo",
            Some("run-local-001"),
            true,
            1,
            1,
            None,
        )
        .unwrap();
        assert_eq!(data, expected);
    }

    #[test]
    fn dns_resolution_matches_example_shape() {
        let raw = include_str!(
            "../../../contracts/examples/cell-observability-dns-resolution-data.valid.json"
        );
        let expected: Value = serde_json::from_str(raw).unwrap();
        let spec = ExecutionCellSpec {
            id: "demo-cell-dns".into(),
            correlation: None,
            ingress: None,
            environment: None,
            placement: None,
            policy: None,
            identity: None,
            run: None,
            authority: Default::default(),
            lifetime: Lifetime { ttl_seconds: 60 },
            export: None,
            telemetry: None,
        };
        let targets: &[(&str, &str, Option<u16>)] = &[
            ("203.0.113.10", "inet", Some(443)),
            ("2001:db8::1", "inet6", Some(443)),
        ];
        let data = observability_dns_resolution_data_v1(
            &spec,
            "demo-cell-dns",
            Some("run-001"),
            "api.example.com",
            "2026-04-30T12:00:00Z",
            targets,
            300,
            "sha256:e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855",
            "keyset-demo-001",
            "kid-resolver-01",
            Some("rcpt-demo-0001"),
        )
        .unwrap();
        assert_eq!(data, expected);
    }

    #[test]
    fn dns_target_set_matches_example_shape() {
        let raw = include_str!(
            "../../../contracts/examples/cell-observability-dns-target-set-data.valid.json"
        );
        let expected: Value = serde_json::from_str(raw).unwrap();
        let spec = ExecutionCellSpec {
            id: "demo-cell-dns".into(),
            correlation: None,
            ingress: None,
            environment: None,
            placement: None,
            policy: None,
            identity: None,
            run: None,
            authority: Default::default(),
            lifetime: Lifetime { ttl_seconds: 60 },
            export: None,
            telemetry: None,
        };
        let data = observability_dns_target_set_data_v1(
            &spec,
            "demo-cell-dns",
            Some("run-001"),
            "cdn.example.com",
            "empty",
            "sha256:e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855",
            "refresh",
            "2026-04-30T12:05:00Z",
            "keyset-demo-001",
            "kid-resolver-01",
        )
        .unwrap();
        assert_eq!(data, expected);
    }

    #[test]
    fn dns_query_data_v1_serializes_allow_path() {
        use crate::{DnsQueryDecision, DnsQueryEvent, DnsQueryReasonCode, DnsQueryType};
        let ev = DnsQueryEvent {
            schema_version: "1.0.0".into(),
            cell_id: "demo-cell-dns".into(),
            run_id: "run-2026-05-01-001".into(),
            query_id: "q-3b58b2a4-e4bb-4f89-9c4f-2a0a2c8b6f01".into(),
            query_name: "api.example.com".into(),
            query_type: DnsQueryType::A,
            decision: DnsQueryDecision::Allow,
            reason_code: DnsQueryReasonCode::AllowedByAllowlist,
            response_rcode: Some(0),
            upstream_resolver_id: Some("resolver-do53-internal".into()),
            upstream_latency_ms: Some(4),
            response_target_count: Some(2),
            keyset_id: Some("keyset-demo-001".into()),
            issuer_kid: Some("kid-resolver-01".into()),
            policy_digest: Some(
                "sha256:e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855".into(),
            ),
            correlation_id: Some("corr-demo-0001".into()),
            observed_at: "2026-05-01T12:34:56Z".into(),
        };
        let v = dns_query_data_v1(&ev).unwrap();
        assert_eq!(v["schemaVersion"], "1.0.0");
        assert_eq!(v["queryName"], "api.example.com");
        assert_eq!(v["queryType"], "A");
        assert_eq!(v["decision"], "allow");
        assert_eq!(v["reasonCode"], "allowed_by_allowlist");
        assert_eq!(v["upstreamResolverId"], "resolver-do53-internal");
        assert_eq!(v["responseTargetCount"], 2);
    }

    #[test]
    fn dns_query_data_v1_omits_optionals_on_deny_path() {
        use crate::{DnsQueryDecision, DnsQueryEvent, DnsQueryReasonCode, DnsQueryType};
        let ev = DnsQueryEvent {
            schema_version: "1.0.0".into(),
            cell_id: "demo-cell-dns".into(),
            run_id: "run-2026-05-01-001".into(),
            query_id: "q-deny-001".into(),
            query_name: "blocked.example.com".into(),
            query_type: DnsQueryType::AAAA,
            decision: DnsQueryDecision::Deny,
            reason_code: DnsQueryReasonCode::DeniedNotInAllowlist,
            response_rcode: Some(5),
            upstream_resolver_id: None,
            upstream_latency_ms: None,
            response_target_count: Some(0),
            keyset_id: None,
            issuer_kid: None,
            policy_digest: None,
            correlation_id: None,
            observed_at: "2026-05-01T12:35:00Z".into(),
        };
        let v = dns_query_data_v1(&ev).unwrap();
        let obj = v.as_object().unwrap();
        assert!(!obj.contains_key("upstreamResolverId"));
        assert!(!obj.contains_key("upstreamLatencyMs"));
        assert!(!obj.contains_key("keysetId"));
        assert!(!obj.contains_key("issuerKid"));
        assert!(!obj.contains_key("policyDigest"));
        assert!(!obj.contains_key("correlationId"));
        assert_eq!(v["decision"], "deny");
        assert_eq!(v["reasonCode"], "denied_not_in_allowlist");
        assert_eq!(v["responseRcode"], 5);
    }

    #[test]
    fn cloud_event_v1_dns_query_envelope() {
        use crate::{DnsQueryDecision, DnsQueryEvent, DnsQueryReasonCode, DnsQueryType};
        let ev = DnsQueryEvent {
            schema_version: "1.0.0".into(),
            cell_id: "c1".into(),
            run_id: "r1".into(),
            query_id: "q1".into(),
            query_name: "api.example.com".into(),
            query_type: DnsQueryType::A,
            decision: DnsQueryDecision::Allow,
            reason_code: DnsQueryReasonCode::AllowedByAllowlist,
            response_rcode: Some(0),
            upstream_resolver_id: Some("r-001".into()),
            upstream_latency_ms: Some(3),
            response_target_count: Some(1),
            keyset_id: None,
            issuer_kid: None,
            policy_digest: None,
            correlation_id: None,
            observed_at: "2026-05-01T12:34:56Z".into(),
        };
        let env =
            cloud_event_v1_dns_query("cellos-dns-proxy", "2026-05-01T12:34:56Z", &ev).unwrap();
        assert_eq!(env.specversion, "1.0");
        assert_eq!(env.ty, "dev.cellos.events.cell.observability.v1.dns_query");
        assert_eq!(env.source, "cellos-dns-proxy");
        assert_eq!(env.datacontenttype.as_deref(), Some("application/json"));
        assert!(env.data.is_some());
    }

    #[test]
    fn qtype_mapping_covers_phase1_set() {
        use crate::{qtype_to_dns_query_type, DnsQueryType};
        assert_eq!(qtype_to_dns_query_type(1), Some(DnsQueryType::A));
        assert_eq!(qtype_to_dns_query_type(2), Some(DnsQueryType::NS));
        assert_eq!(qtype_to_dns_query_type(5), Some(DnsQueryType::CNAME));
        assert_eq!(qtype_to_dns_query_type(12), Some(DnsQueryType::PTR));
        assert_eq!(qtype_to_dns_query_type(15), Some(DnsQueryType::MX));
        assert_eq!(qtype_to_dns_query_type(16), Some(DnsQueryType::TXT));
        assert_eq!(qtype_to_dns_query_type(28), Some(DnsQueryType::AAAA));
        assert_eq!(qtype_to_dns_query_type(33), Some(DnsQueryType::SRV));
        assert_eq!(qtype_to_dns_query_type(64), Some(DnsQueryType::SVCB));
        assert_eq!(qtype_to_dns_query_type(65), Some(DnsQueryType::HTTPS));
        // Unmapped types fall outside the allowlist set.
        assert_eq!(qtype_to_dns_query_type(0), None);
        assert_eq!(qtype_to_dns_query_type(99), None);
        assert_eq!(qtype_to_dns_query_type(255), None); // ANY
    }

    #[test]
    fn l7_egress_decision_matches_example_shape() {
        let raw = include_str!(
            "../../../contracts/examples/cell-observability-l7-egress-decision-data.valid.json"
        );
        let expected: Value = serde_json::from_str(raw).unwrap();
        let spec = ExecutionCellSpec {
            id: "demo-cell-dns".into(),
            correlation: None,
            ingress: None,
            environment: None,
            placement: None,
            policy: None,
            identity: None,
            run: None,
            authority: Default::default(),
            lifetime: Lifetime { ttl_seconds: 60 },
            export: None,
            telemetry: None,
        };
        let data = observability_l7_egress_decision_data_v1(
            &spec,
            "demo-cell-dns",
            Some("run-001"),
            "l7-demo-0002",
            "deny",
            "blocked.example.com",
            "sha256:e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855",
            "keyset-demo-001",
            "kid-l7-01",
            "deny_default",
            Some("authority.egressRules.default"),
            None, // streamId omitted on the SNI/HTTP path; Phase 3g.1 only populates for h2.
        )
        .unwrap();
        assert_eq!(data, expected);
    }

    #[test]
    fn l7_egress_decision_with_stream_id_emits_field() {
        let spec = ExecutionCellSpec {
            id: "demo-cell-dns".into(),
            correlation: None,
            ingress: None,
            environment: None,
            placement: None,
            policy: None,
            identity: None,
            run: None,
            authority: Default::default(),
            lifetime: Lifetime { ttl_seconds: 60 },
            export: None,
            telemetry: None,
        };
        let data = observability_l7_egress_decision_data_v1(
            &spec,
            "demo-cell-dns",
            Some("run-001"),
            "l7-demo-0003",
            "deny",
            "evil.example.com",
            "sha256:e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855",
            "keyset-demo-001",
            "kid-l7-01",
            "l7_h2_authority_allowlist_miss",
            None,
            Some(3), // h2 stream id
        )
        .unwrap();
        assert_eq!(data["streamId"], serde_json::json!(3));
    }

    // ── Tranche-1 seam-freeze G1 + G2 provenance/correlation tests ───────────

    fn _seam_g1_g2_minimal_spec(id: &str) -> ExecutionCellSpec {
        ExecutionCellSpec {
            id: id.into(),
            correlation: None,
            ingress: None,
            environment: None,
            placement: None,
            policy: None,
            identity: None,
            run: None,
            authority: Default::default(),
            lifetime: Lifetime { ttl_seconds: 60 },
            export: None,
            telemetry: None,
        }
    }

    #[test]
    fn seam_g2_identity_revoked_includes_provenance_when_set() {
        // G2 — provenance.parent on revoke event.
        let mut spec = _seam_g1_g2_minimal_spec("seam-g2-revoke");
        spec.identity = Some(WorkloadIdentity {
            kind: WorkloadIdentityKind::FederatedOidc,
            provider: "github-actions".into(),
            audience: "sts.amazonaws.com".into(),
            subject: None,
            ttl_seconds: Some(900),
            secret_ref: "AWS_WEB_IDENTITY".into(),
        });
        let identity = spec.identity.as_ref().unwrap();
        let prov = Provenance {
            parent: "urn:cellos:event:00000000-0000-0000-0000-00000000abcd".into(),
            parent_type: "dev.cellos.events.cell.lifecycle.v1.started".into(),
        };
        let data = identity_revoked_data_v1(
            &spec,
            "cell-seam-g2",
            Some("run-seam-g2"),
            identity,
            Some("teardown"),
            Some(&prov),
        )
        .unwrap();
        assert_eq!(
            data["provenance"]["parent"],
            "urn:cellos:event:00000000-0000-0000-0000-00000000abcd"
        );
        assert_eq!(
            data["provenance"]["parentType"],
            "dev.cellos.events.cell.lifecycle.v1.started"
        );
    }

    #[test]
    fn seam_g2_identity_revoked_omits_provenance_when_none() {
        // Backward-compat: producers without a parent reference must not
        // emit a provenance key at all.
        let mut spec = _seam_g1_g2_minimal_spec("seam-g2-revoke-no-prov");
        spec.identity = Some(WorkloadIdentity {
            kind: WorkloadIdentityKind::FederatedOidc,
            provider: "github-actions".into(),
            audience: "sts.amazonaws.com".into(),
            subject: None,
            ttl_seconds: Some(900),
            secret_ref: "AWS_WEB_IDENTITY".into(),
        });
        let identity = spec.identity.as_ref().unwrap();
        let data =
            identity_revoked_data_v1(&spec, "cell-x", None, identity, Some("teardown"), None)
                .unwrap();
        assert!(!data.as_object().unwrap().contains_key("provenance"));
    }

    #[test]
    fn seam_g2_export_completed_v2_includes_provenance_when_set() {
        // G2 — provenance.parent on export receipt completed event.
        let spec = _seam_g1_g2_minimal_spec("seam-g2-export");
        let receipt = ExportReceipt {
            target_kind: ExportReceiptTargetKind::Local,
            target_name: None,
            destination: "/tmp/out/run-1/artifact.json".into(),
            bytes_written: 42,
        };
        let prov = Provenance {
            parent: "urn:cellos:event:11111111-1111-1111-1111-111111111111".into(),
            parent_type: "dev.cellos.events.cell.lifecycle.v1.started".into(),
        };
        let data = export_completed_data_v2(
            &spec,
            "cell-export",
            Some("run-export"),
            "artifact",
            &receipt,
            Some(&prov),
        )
        .unwrap();
        assert_eq!(
            data["provenance"]["parent"],
            "urn:cellos:event:11111111-1111-1111-1111-111111111111"
        );
        assert_eq!(
            data["provenance"]["parentType"],
            "dev.cellos.events.cell.lifecycle.v1.started"
        );
    }

    #[test]
    fn seam_g2_export_failed_v2_includes_provenance_when_set() {
        // G2 — provenance.parent on export receipt failed event.
        let spec = _seam_g1_g2_minimal_spec("seam-g2-export-failed");
        let prov = Provenance {
            parent: "urn:cellos:event:22222222-2222-2222-2222-222222222222".into(),
            parent_type: "dev.cellos.events.cell.lifecycle.v1.started".into(),
        };
        let data = export_failed_data_v2(
            &spec,
            "cell-fail",
            Some("run-fail"),
            "artifact",
            ExportReceiptTargetKind::S3,
            Some("bucket"),
            Some("s3://bucket/artifact"),
            "denied by policy",
            Some(&prov),
        )
        .unwrap();
        assert_eq!(
            data["provenance"]["parent"],
            "urn:cellos:event:22222222-2222-2222-2222-222222222222"
        );
        assert_eq!(data["reason"], "denied by policy");
    }

    #[test]
    fn seam_g1_correlation_id_propagates_when_present_in_spec() {
        // G1 — when the broker (or operator) supplies correlation_id,
        // every event that mirrors `Correlation` carries it through into
        // its `data.correlation.correlationId` payload field. We exercise
        // the property on a representative subset of events that share the
        // same `if let Some(c) = &spec.correlation { ... }` pattern.
        let mut spec = _seam_g1_g2_minimal_spec("seam-g1-corr");
        spec.correlation = Some(Correlation {
            platform: None,
            external_run_id: None,
            external_job_id: None,
            tenant_id: None,
            labels: None,
            correlation_id: Some("urn:tsafe:corr:01J".into()),
        });
        spec.identity = Some(WorkloadIdentity {
            kind: WorkloadIdentityKind::FederatedOidc,
            provider: "github-actions".into(),
            audience: "sts.amazonaws.com".into(),
            subject: None,
            ttl_seconds: Some(900),
            secret_ref: "AWS_WEB_IDENTITY".into(),
        });
        let identity = spec.identity.as_ref().unwrap();

        // identity_revoked
        let data =
            identity_revoked_data_v1(&spec, "cell-1", Some("r"), identity, Some("teardown"), None)
                .unwrap();
        assert_eq!(
            data["correlation"]["correlationId"], "urn:tsafe:corr:01J",
            "identity.revoked must mirror correlationId from spec"
        );

        // export_completed v2
        let receipt = ExportReceipt {
            target_kind: ExportReceiptTargetKind::Local,
            target_name: None,
            destination: "/tmp/x".into(),
            bytes_written: 1,
        };
        let data = export_completed_data_v2(&spec, "c", Some("r"), "art", &receipt, None).unwrap();
        assert_eq!(
            data["correlation"]["correlationId"], "urn:tsafe:corr:01J",
            "export.v2.completed must mirror correlationId from spec"
        );

        // export_failed v2
        let data = export_failed_data_v2(
            &spec,
            "c",
            Some("r"),
            "art",
            ExportReceiptTargetKind::Local,
            None,
            None,
            "boom",
            None,
        )
        .unwrap();
        assert_eq!(
            data["correlation"]["correlationId"], "urn:tsafe:corr:01J",
            "export.v2.failed must mirror correlationId from spec"
        );

        // command_completed
        let data =
            command_completed_data_v1(&spec, "c", Some("r"), &["echo".to_string()], 0, 5, None)
                .unwrap();
        assert_eq!(
            data["correlation"]["correlationId"], "urn:tsafe:corr:01J",
            "command.completed must mirror correlationId from spec"
        );

        // compliance_summary
        let data = compliance_summary_data_v1(&spec, "c", Some("r"), Some(0)).unwrap();
        assert_eq!(
            data["correlation"]["correlationId"], "urn:tsafe:corr:01J",
            "compliance.summary must mirror correlationId from spec"
        );
    }

    // ---------------------------------------------------------------------
    // Seam-freeze G3 (P0-6): SubjectUrn validation tests.
    //
    // Every literal `urn:...` / `cell:...` string in this section is for a
    // negative or positive test case and is intentionally allow-listed by
    // `scripts/audit/subject-urn-check.sh` (which excludes `events.rs`).
    // ---------------------------------------------------------------------

    #[test]
    fn subject_urn_accepts_canonical_cell_form() {
        // Positive case: the canonical CellOS cell subject shape.
        let urn = SubjectUrn::parse("urn:cellos:cell:abc-123").expect("must parse");
        assert_eq!(urn.as_str(), "urn:cellos:cell:abc-123");
    }

    #[test]
    fn subject_urn_accepts_id_with_internal_colons() {
        // Forward-compat: <id> may carry colons (e.g. nested identifiers).
        // Validation only constrains tool/kind charset, not <id>.
        let urn = SubjectUrn::parse("urn:cellos:event:abc:01j").expect("must parse");
        assert_eq!(urn.as_str(), "urn:cellos:event:abc:01j");
    }

    #[test]
    fn subject_urn_rejects_when_no_urn_scheme() {
        // Negative #1: legacy `cell:<id>` shape — no `urn:` scheme.
        let err = SubjectUrn::parse("cell:abc-123").unwrap_err();
        assert_eq!(err, SubjectUrnError::MissingUrnScheme);
    }

    #[test]
    fn subject_urn_rejects_three_segment_form() {
        // Negative #2: only three colon-separated segments.
        let err = SubjectUrn::parse("urn:cellos:cell").unwrap_err();
        assert_eq!(err, SubjectUrnError::TooFewSegments);
    }

    #[test]
    fn subject_urn_rejects_empty_id() {
        // Negative #3: trailing empty `<id>`.
        let err = SubjectUrn::parse("urn:cellos:cell:").unwrap_err();
        assert_eq!(err, SubjectUrnError::EmptySegment);
    }

    #[test]
    fn subject_urn_rejects_uppercase_tool_or_kind() {
        // Negative #4: charset [a-z0-9-] excludes uppercase.
        let err = SubjectUrn::parse("urn:CellOS:cell:abc-123").unwrap_err();
        assert_eq!(err, SubjectUrnError::InvalidToolOrKindCharset);
    }

    #[test]
    fn subject_urn_rejects_embedded_whitespace() {
        // Negative #5: any whitespace (including in <id>) is rejected.
        let err = SubjectUrn::parse("urn:cellos:cell:abc 123").unwrap_err();
        assert_eq!(err, SubjectUrnError::ControlOrWhitespace);
    }

    #[test]
    fn subject_urn_rejects_empty_tool_segment() {
        // Defense: `urn::kind:id` (empty <tool>) is empty, not charset.
        let err = SubjectUrn::parse("urn::cell:abc-123").unwrap_err();
        assert_eq!(err, SubjectUrnError::EmptySegment);
    }

    #[test]
    fn cell_subject_urn_helper_round_trips() {
        // Helper must produce a SubjectUrn that parses identically.
        let urn = cell_subject_urn("cell-host-7").expect("helper must accept ASCII id");
        assert_eq!(urn.as_str(), "urn:cellos:cell:cell-host-7");
        // And re-parsing the same string must succeed without normalization.
        let reparsed = SubjectUrn::parse(urn.as_str()).expect("must reparse");
        assert_eq!(reparsed, urn);
    }

    #[test]
    fn cell_subject_urn_helper_rejects_empty_id() {
        let err = cell_subject_urn("").unwrap_err();
        assert_eq!(err, SubjectUrnError::EmptySegment);
    }

    // ── Formation lifecycle events ────────────────────────────────────────

    #[test]
    fn formation_data_v1_shape_happy_path() {
        let data = formation_data_v1("f-123", "demo-formation", 3, &[], None);
        assert_eq!(data["formationId"], json!("f-123"));
        assert_eq!(data["formationName"], json!("demo-formation"));
        assert_eq!(data["cellCount"], json!(3));
        assert_eq!(data["failedCellIds"], json!([] as [String; 0]));
        // `reason` is omitted (not null) when None — auditors check key presence.
        let obj = data.as_object().unwrap();
        assert!(!obj.contains_key("reason"));
    }

    #[test]
    fn formation_data_v1_shape_degraded_path_includes_failed_cells_and_reason() {
        let failed = vec!["cell-a".to_string(), "cell-b".to_string()];
        let data = formation_data_v1(
            "f-123",
            "demo-formation",
            5,
            &failed,
            Some("2/5 cells exited non-zero"),
        );
        assert_eq!(data["failedCellIds"], json!(failed));
        assert_eq!(data["reason"], json!("2/5 cells exited non-zero"));
    }

    #[test]
    fn formation_created_envelope_carries_correct_urn() {
        let ev = cloud_event_v1_formation_created(
            "cellos-supervisor",
            "2026-05-16T00:00:00Z",
            "f-1",
            "demo",
            2,
            &[],
            None,
        );
        assert_eq!(ev.ty, "dev.cellos.events.cell.formation.v1.created");
        assert_eq!(ev.specversion, "1.0");
        assert_eq!(ev.source, "cellos-supervisor");
        assert!(ev.data.is_some());
    }

    #[test]
    fn formation_launching_envelope_carries_correct_urn() {
        let ev = cloud_event_v1_formation_launching(
            "cellos-supervisor",
            "2026-05-16T00:00:00Z",
            "f-1",
            "demo",
            2,
            &[],
            None,
        );
        assert_eq!(ev.ty, "dev.cellos.events.cell.formation.v1.launching");
    }

    #[test]
    fn formation_running_envelope_carries_correct_urn() {
        let ev = cloud_event_v1_formation_running(
            "cellos-supervisor",
            "2026-05-16T00:00:00Z",
            "f-1",
            "demo",
            2,
            &[],
            None,
        );
        assert_eq!(ev.ty, "dev.cellos.events.cell.formation.v1.running");
    }

    #[test]
    fn formation_degraded_envelope_carries_correct_urn_and_failed_cells() {
        let failed = vec!["cell-a".to_string()];
        let ev = cloud_event_v1_formation_degraded(
            "cellos-supervisor",
            "2026-05-16T00:00:00Z",
            "f-1",
            "demo",
            3,
            &failed,
            Some("one cell exited 1"),
        );
        assert_eq!(ev.ty, "dev.cellos.events.cell.formation.v1.degraded");
        let data = ev.data.unwrap();
        assert_eq!(data["failedCellIds"], json!(failed));
        assert_eq!(data["reason"], json!("one cell exited 1"));
    }

    #[test]
    fn formation_completed_envelope_carries_correct_urn() {
        let ev = cloud_event_v1_formation_completed(
            "cellos-supervisor",
            "2026-05-16T00:00:00Z",
            "f-1",
            "demo",
            2,
            &[],
            None,
        );
        assert_eq!(ev.ty, "dev.cellos.events.cell.formation.v1.completed");
    }

    #[test]
    fn formation_failed_envelope_carries_correct_urn_and_reason() {
        let failed = vec!["cell-a".to_string(), "cell-b".to_string()];
        let ev = cloud_event_v1_formation_failed(
            "cellos-supervisor",
            "2026-05-16T00:00:00Z",
            "f-1",
            "demo",
            2,
            &failed,
            Some("all cells exited non-zero"),
        );
        assert_eq!(ev.ty, "dev.cellos.events.cell.formation.v1.failed");
        let data = ev.data.unwrap();
        assert_eq!(data["failedCellIds"], json!(failed));
        assert_eq!(data["reason"], json!("all cells exited non-zero"));
    }

    #[test]
    fn formation_type_constants_match_envelope_urns() {
        assert_eq!(
            FORMATION_CREATED_TYPE,
            "dev.cellos.events.cell.formation.v1.created"
        );
        assert_eq!(
            FORMATION_LAUNCHING_TYPE,
            "dev.cellos.events.cell.formation.v1.launching"
        );
        assert_eq!(
            FORMATION_RUNNING_TYPE,
            "dev.cellos.events.cell.formation.v1.running"
        );
        assert_eq!(
            FORMATION_DEGRADED_TYPE,
            "dev.cellos.events.cell.formation.v1.degraded"
        );
        assert_eq!(
            FORMATION_COMPLETED_TYPE,
            "dev.cellos.events.cell.formation.v1.completed"
        );
        assert_eq!(
            FORMATION_FAILED_TYPE,
            "dev.cellos.events.cell.formation.v1.failed"
        );
    }
}