cellos-core 0.7.3

//! Pure validation for parsed [`ExecutionCellDocument`](crate::ExecutionCellDocument).

use std::collections::HashSet;

use crate::error::CellosError;
use crate::ExecutionCellDocument;
use url::Url;

/// Returns true when `value` matches `sha256:<64 lowercase hex digits>`.
fn is_sha256_digest(value: &str) -> bool {
    let Some(hex) = value.strip_prefix("sha256:") else {
        return false;
    };
    hex.len() == 64
        && hex
            .chars()
            .all(|c| c.is_ascii_hexdigit() && !c.is_ascii_uppercase())
}

pub(crate) fn is_portable_identifier(value: &str) -> bool {
    let mut chars = value.chars();
    let Some(first) = chars.next() else {
        return false;
    };
    if !first.is_ascii_alphanumeric() {
        return false;
    }
    if value.len() > 128 || value.contains("..") {
        return false;
    }
    chars.all(|c| c.is_ascii_alphanumeric() || matches!(c, '.' | '_' | '-'))
}

fn ensure_portable_identifier(value: &str, field: &str) -> Result<(), CellosError> {
    if is_portable_identifier(value) {
        Ok(())
    } else {
        Err(CellosError::InvalidSpec(format!(
            "{field} must match [A-Za-z0-9][A-Za-z0-9._-]{{0,127}} and must not contain '..'"
        )))
    }
}

fn is_kubernetes_namespace(value: &str) -> bool {
    if value.is_empty() || value.len() > 63 || value.contains("..") {
        return false;
    }

    let mut chars = value.chars();
    let Some(first) = chars.next() else {
        return false;
    };
    if !first.is_ascii_lowercase() && !first.is_ascii_digit() {
        return false;
    }

    let Some(last) = value.chars().last() else {
        return false;
    };
    if !last.is_ascii_lowercase() && !last.is_ascii_digit() {
        return false;
    }

    value
        .chars()
        .all(|c| c.is_ascii_lowercase() || c.is_ascii_digit() || c == '-')
}

fn ensure_kubernetes_namespace(value: &str, field: &str) -> Result<(), CellosError> {
    if is_kubernetes_namespace(value) {
        Ok(())
    } else {
        Err(CellosError::InvalidSpec(format!(
            "{field} must match Kubernetes DNS label rules: lowercase alphanumeric plus '-', <=63 chars"
        )))
    }
}

/// Returns true when `value` is a single DNS label per RFC 1035 (LDH; 1-63 chars;
/// must start and end with an ASCII alphanumeric; underscores rejected).
fn is_dns_label(value: &str) -> bool {
    if value.is_empty() || value.len() > 63 {
        return false;
    }
    let bytes = value.as_bytes();
    let first_ok = bytes
        .first()
        .copied()
        .is_some_and(|b| b.is_ascii_alphanumeric());
    let last_ok = bytes
        .last()
        .copied()
        .is_some_and(|b| b.is_ascii_alphanumeric());
    if !first_ok || !last_ok {
        return false;
    }
    bytes
        .iter()
        .all(|b| b.is_ascii_alphanumeric() || *b == b'-')
}

/// Returns true when `value` is a sane FQDN — optionally with a single leading
/// `*.` wildcard. Total length capped at 253 (RFC 1035). IPv4/IPv6 literals are
/// rejected so callers can't sneak an IP through a hostname slot.
///
/// Used by [`validate_execution_cell_document`] for `dnsAuthority.hostnameAllowlist`,
/// `cdnAuthority.providers[].hostnamePattern`, and any future hostname-shaped
/// fields under T13.
pub(crate) fn is_fqdn_or_wildcard(value: &str) -> bool {
    if value.is_empty() || value.len() > 253 {
        return false;
    }
    // Reject IPv4-like dotted-quads (e.g. "1.2.3.4") — they parse as 4 labels of
    // pure digits, which we don't want under "hostname". An IPv6 literal contains
    // ':' which is rejected by is_dns_label, so no separate guard is needed.
    if value
        .split('.')
        .all(|segment| !segment.is_empty() && segment.chars().all(|c| c.is_ascii_digit()))
        && value.contains('.')
    {
        return false;
    }

    let labels: Vec<&str> = value.split('.').collect();
    if labels.len() < 2 {
        // Require at least one dot — "example" alone is not a usable FQDN here.
        return false;
    }

    // Optional single leading wildcard label.
    let (first, rest) = labels.split_first().expect("non-empty above");
    if *first == "*" {
        if rest.is_empty() {
            return false;
        }
        return rest.iter().all(|label| is_dns_label(label));
    }

    labels.iter().all(|label| is_dns_label(label))
}

fn parse_http_base_url(value: &str, field: &str) -> Result<Url, CellosError> {
    let parsed = Url::parse(value).map_err(|_| {
        CellosError::InvalidSpec(format!(
            "{field} must be an absolute http(s) base URL without query or fragment"
        ))
    })?;
    let scheme = parsed.scheme();
    if scheme != "http" && scheme != "https" {
        return Err(CellosError::InvalidSpec(format!(
            "{field} must use http or https"
        )));
    }
    if parsed.host_str().is_none() || parsed.query().is_some() || parsed.fragment().is_some() {
        return Err(CellosError::InvalidSpec(format!(
            "{field} must be an absolute http(s) base URL without query or fragment"
        )));
    }
    Ok(parsed)
}

/// Validate a policy pack's declared `spec.version` against the runtime's
/// compiled-in supported floor. P4-04.
///
/// Thin façade over [`crate::check_policy_pack_version_compatibility`]
/// exposed from the `spec_validation` namespace. `allow_downgrade` lets the
/// caller (the supervisor, via the `CELLOS_POLICY_ALLOW_DOWNGRADE` operator
/// override) opt out of the strict floor; cellos-core itself does not read
/// process env vars (D11).
pub fn check_policy_pack_version(
    declared: Option<&str>,
    allow_downgrade: bool,
) -> Result<(), CellosError> {
    crate::policy::check_policy_pack_version_compatibility(declared, allow_downgrade)
}

/// Reject a `tenant_id` that contains any NATS subject-token reserved char.
///
/// NATS subjects are dot-delimited tokens; the wildcards `*` and `>` and any
/// whitespace would either fan out the subject across tenants or produce an
/// unroutable wire string. This guard runs at admission so a malformed
/// `correlation.tenantId` cannot bleed into another tenant's subject when
/// substituted into a `{tenantId}` template (see
/// `cellos-supervisor::spec_input::resolve_event_subject`).
///
/// Pure: no env access, no I/O. Empty `tenant_id` is rejected — callers that
/// want "absent tenant" semantics should pass `Option::None` in the field
/// rather than the empty string.
pub fn validate_tenant_id_for_subject_token(tenant_id: &str) -> Result<(), CellosError> {
    if tenant_id.is_empty() {
        return Err(CellosError::InvalidSpec(
            "spec.correlation.tenantId must be non-empty when present".into(),
        ));
    }
    for ch in tenant_id.chars() {
        let bad = ch == '.' || ch == '*' || ch == '>' || ch.is_whitespace();
        if bad {
            return Err(CellosError::InvalidSpec(format!(
                "spec.correlation.tenantId contains NATS-reserved char: {ch:?} (in {tenant_id:?})"
            )));
        }
    }
    Ok(())
}

/// Reject specs that violate MVP invariants (stricter than JSON Schema alone).
pub fn validate_execution_cell_document(doc: &ExecutionCellDocument) -> Result<(), CellosError> {
    if doc.api_version != "cellos.io/v1" {
        return Err(CellosError::InvalidSpec(format!(
            "unsupported apiVersion: {}",
            doc.api_version
        )));
    }
    if doc.kind != "ExecutionCell" {
        return Err(CellosError::InvalidSpec(format!(
            "unsupported kind: {}",
            doc.kind
        )));
    }
    ensure_portable_identifier(&doc.spec.id, "spec.id")?;
    if let Some(correlation) = &doc.spec.correlation {
        if let Some(tenant_id) = correlation.tenant_id.as_deref() {
            validate_tenant_id_for_subject_token(tenant_id)?;
        }
    }
    let authority_secret_refs = doc
        .spec
        .authority
        .secret_refs
        .as_ref()
        .map(|refs| refs.iter().map(String::as_str).collect::<HashSet<_>>())
        .unwrap_or_default();
    for secret_ref in &authority_secret_refs {
        ensure_portable_identifier(secret_ref, "authority.secretRefs[]")?;
    }
    if let Some(identity) = &doc.spec.identity {
        ensure_portable_identifier(&identity.secret_ref, "spec.identity.secretRef")?;
        if let Some(ttl_seconds) = identity.ttl_seconds {
            if ttl_seconds > doc.spec.lifetime.ttl_seconds {
                return Err(CellosError::InvalidSpec(
                    "spec.identity.ttlSeconds must be <= spec.lifetime.ttlSeconds".into(),
                ));
            }
        }
        if !authority_secret_refs.contains(identity.secret_ref.as_str()) {
            return Err(CellosError::InvalidSpec(format!(
                "spec.identity.secretRef {:?} must also appear in authority.secretRefs",
                identity.secret_ref
            )));
        }
    }
    if let Some(env) = &doc.spec.environment {
        if env.image_reference.is_empty() {
            return Err(CellosError::InvalidSpec(
                "spec.environment.imageReference must be non-empty".into(),
            ));
        }
        if let Some(digest) = &env.image_digest {
            if !is_sha256_digest(digest) {
                return Err(CellosError::InvalidSpec(
                    "spec.environment.imageDigest must be a sha256:<hex64> digest when present"
                        .into(),
                ));
            }
        }
        if let Some(template_id) = &env.template_id {
            ensure_portable_identifier(template_id, "spec.environment.templateId")?;
        }
    }
    if let Some(placement) = &doc.spec.placement {
        if placement.pool_id.is_none()
            && placement.kubernetes_namespace.is_none()
            && placement.queue_name.is_none()
        {
            return Err(CellosError::InvalidSpec(
                "spec.placement must set at least one placement hint".into(),
            ));
        }
        if let Some(pool_id) = &placement.pool_id {
            ensure_portable_identifier(pool_id, "spec.placement.poolId")?;
        }
        if let Some(namespace) = &placement.kubernetes_namespace {
            ensure_kubernetes_namespace(namespace, "spec.placement.kubernetesNamespace")?;
        }
        if let Some(queue_name) = &placement.queue_name {
            ensure_portable_identifier(queue_name, "spec.placement.queueName")?;
        }
    }
    if let Some(ingress) = &doc.spec.ingress {
        if let Some(git) = &ingress.git {
            if let Some(secret_ref) = &git.secret_ref {
                ensure_portable_identifier(secret_ref, "spec.ingress.git.secretRef")?;
            }
        }
        if let Some(image) = &ingress.oci_image {
            if let Some(secret_ref) = &image.secret_ref {
                ensure_portable_identifier(secret_ref, "spec.ingress.ociImage.secretRef")?;
            }
        }
    }
    if let Some(run) = &doc.spec.run {
        // FC-65: argv invariants enforced at admission, before any VM is spawned.
        // Non-UTF-8 argv impossible: serde_json deserializes argv as Vec<String>,
        // which is UTF-8 by construction (rejected at the JSON deserialize boundary).
        if run.argv.is_empty() || run.argv.iter().any(|s| s.is_empty()) {
            return Err(CellosError::InvalidSpec(
                "spec.run.argv must be non-empty with no empty strings".into(),
            ));
        }
        // Reject embedded NUL bytes: execve takes NUL-terminated C strings, so an
        // embedded \0 would silently truncate the argument inside the guest.
        if let Some(idx) = run.argv.iter().position(|s| s.as_bytes().contains(&0)) {
            return Err(CellosError::InvalidSpec(format!(
                "spec.run.argv[{}] contains NUL byte (would be silently truncated by execve)",
                idx
            )));
        }
        check_argv_size_within_kernel_cmdline_limit(&run.argv)?;
        if let Some(timeout_ms) = run.timeout_ms {
            let ttl_ms = doc.spec.lifetime.ttl_seconds.saturating_mul(1000);
            if timeout_ms > ttl_ms {
                return Err(CellosError::InvalidSpec(
                    "spec.run.timeoutMs must be <= spec.lifetime.ttlSeconds * 1000".into(),
                ));
            }
        }
        if let Some(limits) = &run.limits {
            if limits.memory_max_bytes == Some(0) {
                return Err(CellosError::InvalidSpec(
                    "spec.run.limits.memoryMaxBytes must be > 0".into(),
                ));
            }
            if let Some(cpu_max) = &limits.cpu_max {
                if cpu_max.quota_micros == 0 {
                    return Err(CellosError::InvalidSpec(
                        "spec.run.limits.cpuMax.quotaMicros must be > 0".into(),
                    ));
                }
                if cpu_max.period_micros == Some(0) {
                    return Err(CellosError::InvalidSpec(
                        "spec.run.limits.cpuMax.periodMicros must be > 0".into(),
                    ));
                }
            }
        }
    }
    if let Some(rules) = &doc.spec.authority.egress_rules {
        for r in rules {
            if r.host.is_empty() {
                return Err(CellosError::InvalidSpec(
                    "authority.egressRules[].host must be non-empty".into(),
                ));
            }
        }
    }
    if let Some(dns_authority) = &doc.spec.authority.dns_authority {
        validate_dns_authority(dns_authority)?;
    }
    if let Some(cdn_authority) = &doc.spec.authority.cdn_authority {
        validate_cdn_authority(cdn_authority)?;
    }
    // If a derivation token is present, validate structural constraints (subset only).
    // Signature verification is performed by the supervisor with role keys loaded
    // from `CELLOS_AUTHORITY_KEYS_PATH` — see `verify_authority_derivation`.
    if let Some(ref token) = doc.spec.authority.authority_derivation {
        verify_authority_derivation_structural(&doc.spec, token)?;
    }
    if let Some(export) = &doc.spec.export {
        let targets = export.targets.as_deref().unwrap_or(&[]);
        let mut target_names = HashSet::new();
        for target in targets {
            ensure_portable_identifier(target.name(), "spec.export.targets[].name")?;
            if !target_names.insert(target.name().to_string()) {
                return Err(CellosError::InvalidSpec(format!(
                    "duplicate export target name {:?}",
                    target.name()
                )));
            }
            if let Some(secret_ref) = target.secret_ref() {
                if !authority_secret_refs.contains(secret_ref) {
                    return Err(CellosError::InvalidSpec(format!(
                        "export target {:?} secretRef {:?} must appear in authority.secretRefs",
                        target.name(),
                        secret_ref
                    )));
                }
            }
            if let crate::ExportTarget::Http(target) = target {
                let parsed =
                    parse_http_base_url(&target.base_url, "spec.export.targets[].baseUrl")?;
                if let Some(rules) = &doc.spec.authority.egress_rules {
                    if !rules.is_empty() {
                        let host = parsed.host_str().expect("checked above");
                        let port = parsed
                            .port_or_known_default()
                            .expect("http/https always has a known default port");
                        let allowed = rules
                            .iter()
                            .any(|rule| rule.port == port && rule.host.eq_ignore_ascii_case(host));
                        if !allowed {
                            return Err(CellosError::InvalidSpec(format!(
                                "http export target {:?} host {}:{} must appear in authority.egressRules",
                                target.name, host, port
                            )));
                        }
                    }
                }
            }
        }
        if let Some(artifacts) = &export.artifacts {
            for artifact in artifacts {
                ensure_portable_identifier(&artifact.name, "spec.export.artifacts[].name")?;
                match artifact.target.as_deref() {
                    Some(target_name) => {
                        ensure_portable_identifier(target_name, "spec.export.artifacts[].target")?;
                        if !target_names.contains(target_name) {
                            return Err(CellosError::InvalidSpec(format!(
                                "export artifact {:?} references unknown target {:?}",
                                artifact.name, target_name
                            )));
                        }
                    }
                    None if targets.len() > 1 => {
                        return Err(CellosError::InvalidSpec(format!(
                            "export artifact {:?} must set target when multiple export targets exist",
                            artifact.name
                        )));
                    }
                    None => {}
                }
            }
        }
    }
    if let Some(telemetry) = &doc.spec.telemetry {
        validate_telemetry_block(telemetry, doc.spec.authority.egress_rules.as_deref())?;
    }
    Ok(())
}

/// F4a — admission validation for [`crate::TelemetrySpec`].
///
/// Checks (in order):
/// 1. `events` is non-empty.
/// 2. `channel == VsockCbor` (the only supported channel today — guards
///    against forward-compat envelopes accidentally bypassing host wiring).
/// 3. `agentVersion` matches a permissive semver shape
///    (`MAJOR.MINOR.PATCH` with optional `-prerelease` suffix).
/// 4. For every entry in `events` that begins with `net.` (case-insensitive),
///    the spec's `authority.egressRules` MUST be non-empty. A `net.*` event
///    declared without any declared egress is an unbacked telemetry/authority
///    claim — the SIEM would receive network observations from a cell that
///    declared no network authority. Returns
///    `CellosError::InvalidSpec("telemetry_without_egress: ...")`.
///
/// Pure structural; no env access. Mirrored in
/// `contracts/schemas/execution-cell-v1.schema.json` under the `Telemetry`
/// definition.
fn validate_telemetry_block(
    telemetry: &crate::TelemetrySpec,
    egress_rules: Option<&[crate::EgressRule]>,
) -> Result<(), CellosError> {
    if telemetry.events.is_empty() {
        return Err(CellosError::InvalidSpec(
            "spec.telemetry.events must be non-empty".into(),
        ));
    }
    match telemetry.channel {
        crate::TelemetryChannel::VsockCbor => {}
    }
    if !is_semver_shape(&telemetry.agent_version) {
        return Err(CellosError::InvalidSpec(
            "spec.telemetry.agentVersion must match MAJOR.MINOR.PATCH semver shape (optional -prerelease suffix)".into()
        ));
    }
    let has_egress = egress_rules.map(|r| !r.is_empty()).unwrap_or(false);
    for event in &telemetry.events {
        let trimmed = event.trim();
        if trimmed.is_empty() {
            return Err(CellosError::InvalidSpec(
                "spec.telemetry.events[] entries must be non-empty".into(),
            ));
        }
        if trimmed.to_ascii_lowercase().starts_with("net.") && !has_egress {
            return Err(CellosError::InvalidSpec(format!(
                "telemetry_without_egress: spec.telemetry.events[] entry {trimmed:?} requires \
                 at least one authority.egressRules entry — net.* telemetry without declared \
                 egress is an unbacked observation claim"
            )));
        }
    }
    Ok(())
}

/// Permissive semver-shape check for `spec.telemetry.agentVersion`.
///
/// Matches `MAJOR.MINOR.PATCH` (each numeric, no leading zeros except "0")
/// with an optional `-<prerelease>` suffix. Build metadata (`+...`) is not
/// accepted for now — keep the shape narrow until operators ask for it.
fn is_semver_shape(value: &str) -> bool {
    let core = match value.split_once('-') {
        Some((core, pre)) => {
            if pre.is_empty()
                || !pre
                    .chars()
                    .all(|c| c.is_ascii_alphanumeric() || matches!(c, '.' | '-'))
            {
                return false;
            }
            core
        }
        None => value,
    };
    let parts: Vec<&str> = core.split('.').collect();
    if parts.len() != 3 {
        return false;
    }
    parts.iter().all(|p| {
        !p.is_empty()
            && p.chars().all(|c| c.is_ascii_digit())
            && !(p.len() > 1 && p.starts_with('0'))
    })
}

/// Structural subset check for an `AuthorityDerivationToken` — does NOT verify
/// signatures. Used by `validate_execution_cell_document`, which has no access
/// to role keys. The supervisor performs full signature verification separately
/// via `verify_authority_derivation` once keys are loaded from
/// `CELLOS_AUTHORITY_KEYS_PATH`.
fn verify_authority_derivation_structural(
    spec: &crate::ExecutionCellSpec,
    token: &crate::AuthorityDerivationToken,
) -> Result<(), crate::error::CellosError> {
    let spec_capability = crate::AuthorityCapability {
        egress_rules: spec.authority.egress_rules.clone().unwrap_or_default(),
        secret_refs: spec.authority.secret_refs.clone().unwrap_or_default(),
    };
    if !spec_capability.is_superset_of(&token.leaf_capability) {
        return Err(crate::error::CellosError::InvalidSpec(
            "spec.authorityDerivation.leafCapability exceeds spec.authority — child authority must be ⊆ declared authority".into()
        ));
    }
    Ok(())
}

/// Validate `authority.dnsAuthority` (T13 / SEC-20).
///
/// Pure structural / domain checks — does NOT contact any resolver and does NOT
/// require trust-keyset state. The supervisor performs full trust-keyset binding
/// at runtime in SEC-21/SEC-22.
fn validate_dns_authority(dns: &crate::DnsAuthority) -> Result<(), CellosError> {
    let mut resolver_ids: HashSet<&str> = HashSet::new();
    for resolver in &dns.resolvers {
        ensure_portable_identifier(
            &resolver.resolver_id,
            "authority.dnsAuthority.resolvers[].resolverId",
        )?;
        if !resolver_ids.insert(resolver.resolver_id.as_str()) {
            return Err(CellosError::InvalidSpec(format!(
                "authority.dnsAuthority.resolvers[].resolverId duplicates value {:?}",
                resolver.resolver_id
            )));
        }
        if resolver.endpoint.is_empty() {
            return Err(CellosError::InvalidSpec(
                "authority.dnsAuthority.resolvers[].endpoint must be non-empty".into(),
            ));
        }
        if let Some(kid) = &resolver.trust_kid {
            ensure_portable_identifier(kid, "authority.dnsAuthority.resolvers[].trustKid")?;
        }
    }

    for hostname in &dns.hostname_allowlist {
        if !is_fqdn_or_wildcard(hostname) {
            return Err(CellosError::InvalidSpec(format!(
                "authority.dnsAuthority.hostnameAllowlist entry {hostname:?} is not a valid FQDN \
                 (single leading '*.' wildcard allowed; IPs are rejected)"
            )));
        }
    }

    if let Some(refresh) = &dns.refresh_policy {
        if let (Some(min_ttl), Some(max_stale)) =
            (refresh.min_ttl_seconds, refresh.max_stale_seconds)
        {
            if min_ttl > max_stale {
                return Err(CellosError::InvalidSpec(format!(
                    "authority.dnsAuthority.refreshPolicy.minTtlSeconds ({min_ttl}) must be <= maxStaleSeconds ({max_stale})"
                )));
            }
        }
    }

    Ok(())
}

/// Validate `authority.cdnAuthority` (T13 / SEC-20).
fn validate_cdn_authority(cdn: &crate::CdnAuthority) -> Result<(), CellosError> {
    let mut provider_ids: HashSet<&str> = HashSet::new();
    for provider in &cdn.providers {
        ensure_portable_identifier(
            &provider.provider_id,
            "authority.cdnAuthority.providers[].providerId",
        )?;
        if !provider_ids.insert(provider.provider_id.as_str()) {
            return Err(CellosError::InvalidSpec(format!(
                "authority.cdnAuthority.providers[].providerId duplicates value {:?}",
                provider.provider_id
            )));
        }
        if !is_fqdn_or_wildcard(&provider.hostname_pattern) {
            return Err(CellosError::InvalidSpec(format!(
                "authority.cdnAuthority.providers[].hostnamePattern {:?} is not a valid FQDN \
                 (single leading '*.' wildcard allowed; IPs are rejected)",
                provider.hostname_pattern
            )));
        }
    }
    Ok(())
}

/// FC-17 / FC-66 — admission cap on the size of `spec.run.argv` once the
/// Firecracker host base64-encodes it onto the kernel boot cmdline.
///
/// The Linux kernel cmdline has a 4 KiB hard limit. The Firecracker host
/// (`crates/cellos-host-firecracker/src/lib.rs::build_boot_args`) writes
/// `cellos.argv=<base64(json_array)>` plus a small fixed prefix (`console=`,
/// `root=`, `cellos.cell_id=`, `cellos.vsock_port=` …). We cap the encoded
/// argv payload at 3 KiB so the rest of the cmdline keeps ~1 KiB of headroom
/// against future cmdline additions.
///
/// `MAX_ARGV_ENCODED_BYTES` is exposed in the typed
/// [`CellosError::ArgvTooLarge::limit_bytes`] field so operators do not have
/// to dig into core to see the budget.
const MAX_ARGV_ENCODED_BYTES: usize = 3072;

/// FC-17 — admission check that `spec.run.argv` will fit inside the Linux
/// kernel boot-cmdline 4 KiB hard limit once Firecracker assembles it.
///
/// FC-66 — surfaces the rejection as the typed
/// [`CellosError::ArgvTooLarge`] variant (carrying the actual encoded byte
/// count and the static limit) rather than as a string-payload `InvalidSpec`,
/// so callers can pattern-match on it without parsing a message.
///
/// We reproduce the host's exact encoding (`base64(serde_json::to_string(argv))`)
/// and compare its length against [`MAX_ARGV_ENCODED_BYTES`] (3072 bytes).
pub(crate) fn check_argv_size_within_kernel_cmdline_limit(
    argv: &[String],
) -> Result<(), CellosError> {
    use base64::engine::general_purpose::STANDARD;
    use base64::Engine as _;

    // serde_json::to_string on Vec<String> cannot fail (no non-string keys, no
    // floats), but the API is fallible — fall back to a conservative upper
    // bound (raw concatenated bytes + JSON framing) on the unreachable Err
    // path so admission never panics on user input.
    let encoded_len = match serde_json::to_string(argv) {
        Ok(json) => STANDARD.encode(json.as_bytes()).len(),
        Err(_) => {
            // ceil(n/3)*4 of a generous JSON framing estimate:
            // 2 brackets + per-arg 3 bytes (quotes + comma) + raw arg bytes,
            // each byte possibly doubled by JSON string-escaping.
            let json_upper = 2 + argv
                .iter()
                .map(|s| s.len().saturating_mul(2).saturating_add(3))
                .sum::<usize>();
            json_upper.div_ceil(3).saturating_mul(4)
        }
    };

    if encoded_len > MAX_ARGV_ENCODED_BYTES {
        return Err(CellosError::ArgvTooLarge {
            encoded_bytes: encoded_len,
            limit_bytes: MAX_ARGV_ENCODED_BYTES,
        });
    }
    Ok(())
}

/// Build the canonical JSON payload that the grantor signs and the supervisor verifies.
///
/// Field order is significant — both signer and verifier MUST agree on the encoding.
/// The `serde_json::json!` macro preserves insertion order for object literals,
/// so the bytes produced here are deterministic given the same inputs.
///
/// Layout:
/// ```json
/// { "roleRoot": "<RoleId>", "leafCapability": <AuthorityCapability>, "parentRunId": <string|null> }
/// ```
pub fn authority_derivation_signing_payload(
    token: &crate::AuthorityDerivationToken,
) -> Result<Vec<u8>, crate::error::CellosError> {
    let value = serde_json::json!({
        "roleRoot": token.role_root.to_string(),
        "leafCapability": &token.leaf_capability,
        "parentRunId": token.parent_run_id,
    });
    serde_json::to_vec(&value).map_err(|e| {
        crate::error::CellosError::InvalidSpec(format!(
            "authority derivation signing payload encode failed: {e}"
        ))
    })
}

/// Verify an `AuthorityDerivationToken` against the declared spec authority.
///
/// Checks (in order):
/// 1. `token.leaf_capability` is a subset of `spec.authority` (egress + secret dims).
/// 2. `role_keys[token.role_root]` exists.
/// 3. The base64-encoded verifying key parses as a valid 32-byte ED25519 key.
/// 4. `token.grantor_signature.bytes` parses as a valid 64-byte ED25519 signature.
/// 5. The signature verifies (`verify_strict`) over the canonical signing payload.
///
/// `role_keys`: map from RoleId string → base64-encoded ED25519 verifying key
/// (raw 32-byte form, base64-STANDARD encoded).
///
/// Returns `Ok(())` on success, `Err(CellosError::InvalidSpec(...))` on any failure.
pub fn verify_authority_derivation(
    spec: &crate::ExecutionCellSpec,
    token: &crate::AuthorityDerivationToken,
    role_keys: &std::collections::HashMap<String, String>,
) -> Result<(), crate::error::CellosError> {
    use base64::engine::general_purpose::STANDARD;
    use base64::Engine as _;
    use ed25519_dalek::{Signature, VerifyingKey};

    // Step 1: structural subset check.
    verify_authority_derivation_structural(spec, token)?;

    // Step 2: look up the verifying key for this role.
    let role_id = token.role_root.to_string();
    let verifying_key_b64 = role_keys.get(&role_id).ok_or_else(|| {
        crate::error::CellosError::InvalidSpec(format!("unknown role: {role_id}"))
    })?;

    // Step 3: decode + parse the verifying key.
    let verifying_key_bytes = STANDARD.decode(verifying_key_b64.as_bytes()).map_err(|e| {
        crate::error::CellosError::InvalidSpec(format!(
            "authority derivation verifying key for role {role_id} is not valid base64: {e}"
        ))
    })?;
    let verifying_key_array: [u8; 32] =
        verifying_key_bytes.as_slice().try_into().map_err(|_| {
            crate::error::CellosError::InvalidSpec(format!(
                "authority derivation verifying key for role {role_id} must be 32 bytes (got {})",
                verifying_key_bytes.len()
            ))
        })?;
    let verifying_key = VerifyingKey::from_bytes(&verifying_key_array).map_err(|e| {
        crate::error::CellosError::InvalidSpec(format!(
            "authority derivation verifying key for role {role_id} is not a valid ED25519 point: {e}"
        ))
    })?;

    // Step 4: decode + parse the signature.
    let sig_bytes = STANDARD
        .decode(token.grantor_signature.bytes.as_bytes())
        .map_err(|e| {
            crate::error::CellosError::InvalidSpec(format!(
                "authority derivation grantor signature is not valid base64: {e}"
            ))
        })?;
    let sig_array: [u8; 64] = sig_bytes.as_slice().try_into().map_err(|_| {
        crate::error::CellosError::InvalidSpec(format!(
            "authority derivation grantor signature must be 64 bytes (got {})",
            sig_bytes.len()
        ))
    })?;
    let signature = Signature::from_bytes(&sig_array);

    // Step 5: build canonical payload + verify.
    let payload = authority_derivation_signing_payload(token)?;
    verifying_key
        .verify_strict(&payload, &signature)
        .map_err(|_| {
            crate::error::CellosError::InvalidSpec("authority derivation signature invalid".into())
        })?;

    tracing::debug!(
        role_root = %token.role_root,
        "authority derivation token verified (structural + signature)"
    );

    Ok(())
}

/// Enforce the derivation-token scope policy after signature verification (L5-16 / I6 / O6).
///
/// `parent_run_id` is grantor-asserted in the signed payload. A token signed with
/// `parentRunId: null` verifies cryptographically against any compatible run —
/// making it a replayable universal delegation token. This function closes that
/// replay window by rejecting universal tokens **by default** (1.0 strict-by-default
/// posture) and only accepting them when an operator EXPLICITLY opts out into
/// permissive mode.
///
/// Behaviour:
/// - When `allow_universal` is `false` — the **default** posture, including
///   when `CELLOS_REQUIRE_SCOPED_DERIVATION_TOKENS` is unset, empty, or set to
///   `1`/`true`/`yes`/`on` — a token with `parent_run_id == None` is rejected
///   as `CellosError::InvalidSpec`.
/// - When `allow_universal` is `true` — only when the operator EXPLICITLY
///   opted out by setting `CELLOS_REQUIRE_SCOPED_DERIVATION_TOKENS` to one of
///   `0`/`false`/`no`/`off` — a token with `parent_run_id == None` is
///   accepted. The supervisor additionally emits a structured CloudEvent
///   (`dev.cellos.events.cell.identity.v1.universal_token_accepted_in_permissive_mode`)
///   recording the exposure so the audit trail captures it. A `WARN` is also
///   logged on the `cellos.supervisor.authority` target.
///
/// This is additive: tokens whose `parent_run_id` is `Some(_)` always pass.
pub fn enforce_derivation_scope_policy(
    token: &crate::AuthorityDerivationToken,
    allow_universal: bool,
) -> Result<(), crate::error::CellosError> {
    if token.parent_run_id.is_some() {
        return Ok(());
    }

    if allow_universal {
        tracing::warn!(
            target: "cellos.supervisor.authority",
            role_root = %token.role_root,
            "authority derivation token has parentRunId: null — universal token accepted in permissive mode (unset CELLOS_REQUIRE_SCOPED_DERIVATION_TOKENS or set it to 1/true/yes/on to restore the strict-by-default posture)"
        );
        return Ok(());
    }

    Err(crate::error::CellosError::InvalidSpec(
        "authority derivation token has parentRunId: null — universal tokens are rejected by the strict-by-default policy (set CELLOS_REQUIRE_SCOPED_DERIVATION_TOKENS to 0/false/no/off to opt into permissive mode, which emits an audit warning event instead)".into(),
    ))
}

/// Verify a SEC-25 signed trust-keyset envelope and return its raw payload bytes.
///
/// `verifying_keys` is a map from `signerKid` to a parsed `ed25519_dalek::VerifyingKey`.
/// The caller is responsible for sourcing this map (e.g. from
/// `CELLOS_TRUST_VERIFY_KEYS_PATH`).
///
/// Verification steps (in order):
/// 1. Decode `payload` as base64url. Compute `sha256:<hex>` over the raw bytes
///    and compare to `payloadDigest`. Mismatch → `CellosError::InvalidSpec`.
/// 2. For each entry in `signatures`:
///    - Look up `signerKid` in `verifying_keys`. Missing kid → that signature is
///      rejected (do not fail the whole envelope yet).
///    - If `notBefore`/`notAfter` are set, parse them as RFC3339 and check
///      `now ∈ [notBefore, notAfter]`. Out-of-window → that signature is rejected.
///    - Decode `signature` as base64url; require exactly 64 bytes; call
///      `verify_strict` over the raw payload bytes.
///    - On success, record the signer's `signerKid` in a deduplicating set.
/// 3. **Phase 3 N-of-M threshold:** compare the count of DISTINCT verified
///    signer kids against `envelope.required_signer_count.unwrap_or(1)`
///    (clamped to a minimum of 1). The Phase 1 default of 1 preserves
///    backward-compat with the original "at least one signature must verify"
///    policy — single-signer envelopes accepted, missing-`required_signer_count`
///    envelopes accepted. Operator deployments raise the threshold (e.g. 2)
///    so a single compromised signer key cannot alone authorize a keyset.
/// 4. If `verified_distinct_signers >= required` → return the raw payload
///    bytes. Otherwise → `CellosError::InvalidSpec` with a message that
///    distinguishes the legacy single-signer failure (`"no signature
///    verified"`) from a threshold shortfall (`"only N distinct signers
///    verified, need M"`).
///
/// **Distinct-kid counting.** Two signature entries with the same `signerKid`
/// — even with different signature bytes — count as one verifier. The
/// schema's `signatures` list permits duplicates (a re-sign by the same
/// operator is structurally valid) but the threshold policy treats them as
/// one operator's vote. This is the intended N-of-M semantic.
///
/// This function does NOT deserialize the inner trust-keyset payload. The
/// schema is the contract of record; callers may parse the bytes into any
/// shape (`serde_json::Value` or a future `TrustKeysetV1` Rust mirror).
pub fn verify_signed_trust_keyset_envelope(
    envelope: &crate::types::SignedTrustKeysetEnvelope,
    verifying_keys: &std::collections::HashMap<String, ed25519_dalek::VerifyingKey>,
    now: std::time::SystemTime,
) -> Result<Vec<u8>, crate::error::CellosError> {
    use base64::engine::general_purpose::URL_SAFE_NO_PAD;
    use base64::Engine as _;
    use ed25519_dalek::Signature;
    use std::collections::HashSet;

    // scope: lock payloadType + algorithms to declared SEC-25 shape.
    if envelope.payload_type != "application/vnd.cellos.trust-keyset-v1+json" {
        return Err(crate::error::CellosError::InvalidSpec(format!(
            "signed trust keyset envelope payloadType must be application/vnd.cellos.trust-keyset-v1+json, got '{}'",
            envelope.payload_type
        )));
    }

    // Step 1: decode payload + verify digest.
    // Accept either base64url-with-padding or base64url-no-padding by stripping '=' first.
    let payload_b64 = envelope.payload.trim_end_matches('=');
    let payload_bytes = URL_SAFE_NO_PAD.decode(payload_b64).map_err(|e| {
        crate::error::CellosError::InvalidSpec(format!(
            "signed trust keyset envelope payload is not valid base64url: {e}"
        ))
    })?;
    let computed_digest = sha256_hex_prefixed(&payload_bytes);
    if computed_digest != envelope.payload_digest {
        return Err(crate::error::CellosError::InvalidSpec(format!(
            "signed trust keyset envelope payload digest mismatch: declared={}, computed={}",
            envelope.payload_digest, computed_digest
        )));
    }

    if envelope.signatures.is_empty() {
        // Schema enforces minItems: 1, but defend against hand-built structs.
        return Err(crate::error::CellosError::InvalidSpec(
            "signed trust keyset envelope has no signatures".into(),
        ));
    }

    // scope: N-of-M threshold. Default + zero clamp to 1 (single-signature
    // semantics). Schema's `minimum: 1` blocks zero at the contract layer;
    // the Rust clamp is a defense-in-depth against hand-built envelopes.
    let required = envelope.required_signer_count.unwrap_or(1).max(1);

    // Step 2: walk signatures; collect distinct signerKids whose signature
    // verified. We collect ALL of them (no early-break) so the threshold
    // check has the true distinct-verifier count.
    let mut verified_signers: HashSet<&str> = HashSet::new();
    for sig_entry in &envelope.signatures {
        if sig_entry.algorithm != "ed25519" {
            // Unknown algorithm: skip (a future-version envelope might have
            // multi-algo signatures alongside our supported one).
            continue;
        }
        // If this kid already verified via an earlier signature, skip the
        // expensive crypto check — a duplicate kid only ever counts once.
        if verified_signers.contains(sig_entry.signer_kid.as_str()) {
            continue;
        }
        let Some(verifying_key) = verifying_keys.get(&sig_entry.signer_kid) else {
            // Unknown kid for the verifier's keyring → skip this signature.
            continue;
        };

        // notBefore / notAfter window check (each bound is optional).
        if !signature_window_contains(
            now,
            sig_entry.not_before.as_deref(),
            sig_entry.not_after.as_deref(),
        )? {
            continue;
        }

        // Decode signature: base64url (with or without padding) → exactly 64 bytes.
        let sig_b64 = sig_entry.signature.trim_end_matches('=');
        let Ok(sig_bytes) = URL_SAFE_NO_PAD.decode(sig_b64) else {
            continue;
        };
        let Ok(sig_array) = <[u8; 64]>::try_from(sig_bytes.as_slice()) else {
            continue;
        };
        let signature = Signature::from_bytes(&sig_array);

        if verifying_key
            .verify_strict(&payload_bytes, &signature)
            .is_ok()
        {
            verified_signers.insert(sig_entry.signer_kid.as_str());
        }
    }

    if verified_signers.len() >= required as usize {
        return Ok(payload_bytes);
    }

    // Preserve the legacy "no signature verified" error message when the
    // threshold is the Phase 1 default (1) and zero signatures verified —
    // existing operator triage runbooks and the SEC-25 Phase 2 supervisor
    // tests grep for that exact substring. The new threshold-shortfall
    // message is reserved for Phase 3 N-of-M (>= 2) failures.
    if required == 1 {
        Err(crate::error::CellosError::InvalidSpec(
            "signed trust keyset envelope: no signature verified".into(),
        ))
    } else {
        Err(crate::error::CellosError::InvalidSpec(format!(
            "signed trust keyset envelope: only {} distinct signers verified, need {}",
            verified_signers.len(),
            required
        )))
    }
}

/// Verify a chain of signed trust-keyset envelopes for replay-safety (SEC-25 Phase 3).
///
/// The chain is ordered **oldest-first**; the HEAD (current) envelope is the
/// LAST entry. Each non-genesis envelope MUST carry a
/// `replacesEnvelopeDigest` equal to `sha256:<hex>` of the immediately prior
/// envelope's raw decoded payload bytes. The first envelope's
/// `replacesEnvelopeDigest` MAY be absent (genesis) or present (chain root
/// reference) — the chain verifier does not check the genesis link.
///
/// Per-envelope verification (signature, digest, threshold, validity window)
/// is delegated to [`verify_signed_trust_keyset_envelope`]; chain integrity
/// is layered on top:
///
/// 1. Empty chain → `CellosError::InvalidSpec`.
/// 2. For each envelope in order, call `verify_signed_trust_keyset_envelope`.
///    Per-envelope verification failures (including N-of-M threshold
///    shortfall) propagate immediately.
/// 3. For each adjacent pair `(prev, next)`, require
///    `next.replaces_envelope_digest == Some("sha256:<hex>(prev_payload_bytes)")`.
///    Mismatch → `CellosError::InvalidSpec` naming the chain index.
///
/// Returns the verified raw payload bytes of the HEAD envelope on success.
///
/// **Replay defense.** Verifiers that cache keyset state SHOULD reject any
/// new envelope whose `replacesEnvelopeDigest` does not match the cached
/// HEAD's payloadDigest — that is either a replay of an old envelope or a
/// chain fork. This function does not maintain that cache; it only verifies
/// the integrity of a chain handed to it.
pub fn verify_signed_trust_keyset_chain(
    chain: &[crate::types::SignedTrustKeysetEnvelope],
    verifying_keys: &std::collections::HashMap<String, ed25519_dalek::VerifyingKey>,
    now: std::time::SystemTime,
) -> Result<Vec<u8>, crate::error::CellosError> {
    if chain.is_empty() {
        return Err(crate::error::CellosError::InvalidSpec(
            "signed trust keyset chain: empty chain".into(),
        ));
    }

    let mut prev_payload_bytes: Option<Vec<u8>> = None;
    for (idx, envelope) in chain.iter().enumerate() {
        let payload_bytes = verify_signed_trust_keyset_envelope(envelope, verifying_keys, now)
            .map_err(|e| {
                crate::error::CellosError::InvalidSpec(format!(
                    "signed trust keyset chain: envelope at index {idx} failed verification: {e}"
                ))
            })?;

        if let Some(prev_bytes) = prev_payload_bytes.as_deref() {
            // Non-genesis: replacesEnvelopeDigest MUST equal sha256(prev.payload_bytes).
            let expected = sha256_hex_prefixed(prev_bytes);
            match envelope.replaces_envelope_digest.as_deref() {
                Some(actual) if actual == expected => {}
                Some(actual) => {
                    return Err(crate::error::CellosError::InvalidSpec(format!(
                        "signed trust keyset chain: envelope at index {idx} replacesEnvelopeDigest mismatch: declared={actual}, expected={expected}"
                    )));
                }
                None => {
                    return Err(crate::error::CellosError::InvalidSpec(format!(
                        "signed trust keyset chain: envelope at index {idx} missing replacesEnvelopeDigest (only the genesis envelope at index 0 may omit it)"
                    )));
                }
            }
        }
        // The genesis envelope (index 0) is allowed to set or omit
        // replacesEnvelopeDigest — we do not check the link before it.

        prev_payload_bytes = Some(payload_bytes);
    }

    // Unwrap is safe: chain non-empty above + prev_payload_bytes is set on
    // every loop iteration.
    Ok(prev_payload_bytes.expect("chain non-empty checked above"))
}

/// `sha256:<hex>` over `bytes`, matching the prefix convention used across
/// CellOS trust-plane schemas (`payloadDigest`, `policyDigest`, etc.).
fn sha256_hex_prefixed(bytes: &[u8]) -> String {
    use sha2::{Digest, Sha256};
    use std::fmt::Write as _;
    let out = Sha256::new().chain_update(bytes).finalize();
    let mut hex = String::with_capacity(7 + 64);
    hex.push_str("sha256:");
    for b in out.iter() {
        let _ = write!(hex, "{b:02x}");
    }
    hex
}

/// Return `Ok(true)` iff `now` is within `[not_before, not_after]` (inclusive).
/// Either bound may be `None` (meaning unbounded on that side). RFC3339 parse
/// failure → `Err(CellosError::InvalidSpec)`.
fn signature_window_contains(
    now: std::time::SystemTime,
    not_before: Option<&str>,
    not_after: Option<&str>,
) -> Result<bool, crate::error::CellosError> {
    use chrono::{DateTime, Utc};
    let now_chrono: DateTime<Utc> = now.into();
    if let Some(nb) = not_before {
        let parsed = DateTime::parse_from_rfc3339(nb).map_err(|e| {
            crate::error::CellosError::InvalidSpec(format!(
                "signed trust keyset envelope notBefore '{nb}' is not RFC3339: {e}"
            ))
        })?;
        if now_chrono < parsed.with_timezone(&Utc) {
            return Ok(false);
        }
    }
    if let Some(na) = not_after {
        let parsed = DateTime::parse_from_rfc3339(na).map_err(|e| {
            crate::error::CellosError::InvalidSpec(format!(
                "signed trust keyset envelope notAfter '{na}' is not RFC3339: {e}"
            ))
        })?;
        if now_chrono > parsed.with_timezone(&Utc) {
            return Ok(false);
        }
    }
    Ok(true)
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::{
        AuthorityBundle, EnvironmentSpec, ExecutionCellSpec, ExportArtifact, ExportChannels,
        ExportTarget, HttpExportTarget, Lifetime, RunCpuMax, RunLimits, RunSpec, S3ExportTarget,
        SecretDeliveryMode, WorkloadIdentity, WorkloadIdentityKind,
    };

    #[test]
    fn rejects_empty_argv_token() {
        let doc = ExecutionCellDocument {
            api_version: "cellos.io/v1".into(),
            kind: "ExecutionCell".into(),
            spec: ExecutionCellSpec {
                id: "x".into(),
                correlation: None,
                ingress: None,
                environment: None,
                placement: None,
                policy: None,
                identity: None,
                run: Some(RunSpec {
                    argv: vec!["sh".into(), "".into()],
                    working_directory: None,
                    timeout_ms: None,
                    limits: None,
                    secret_delivery: SecretDeliveryMode::Env,
                }),
                authority: AuthorityBundle::default(),
                lifetime: Lifetime { ttl_seconds: 1 },
                export: None,
                telemetry: None,
            },
        };
        assert!(validate_execution_cell_document(&doc).is_err());
    }

    /// FC-17 — argv totalling ~1 KiB of payload bytes encodes well under the
    /// 3 KiB cap and must pass admission.
    #[test]
    fn admits_argv_under_kernel_cmdline_limit() {
        // 1 KiB of payload across two args; base64(json) ~= 1.4 KiB, < 3072.
        let payload = "a".repeat(1024);
        let doc = ExecutionCellDocument {
            api_version: "cellos.io/v1".into(),
            kind: "ExecutionCell".into(),
            spec: ExecutionCellSpec {
                id: "argv-fits".into(),
                correlation: None,
                ingress: None,
                environment: None,
                placement: None,
                policy: None,
                identity: None,
                run: Some(RunSpec {
                    argv: vec!["sh".into(), payload],
                    working_directory: None,
                    timeout_ms: None,
                    limits: None,
                    secret_delivery: SecretDeliveryMode::Env,
                }),
                authority: AuthorityBundle::default(),
                lifetime: Lifetime { ttl_seconds: 1 },
                export: None,
                telemetry: None,
            },
        };
        validate_execution_cell_document(&doc).expect("1 KiB argv must pass admission");
    }

    /// FC-17 / FC-66 — argv whose base64(json) encoding exceeds 3 KiB must be
    /// rejected at admission with the typed [`CellosError::ArgvTooLarge`]
    /// variant (not a string-payload `InvalidSpec`) so callers can
    /// pattern-match without parsing a message.
    #[test]
    fn rejects_argv_exceeding_kernel_cmdline_limit() {
        // 5 KiB of raw payload → JSON ~5 KiB → base64 ~6.7 KiB; well over 3072.
        let payload = "a".repeat(5 * 1024);
        let doc = ExecutionCellDocument {
            api_version: "cellos.io/v1".into(),
            kind: "ExecutionCell".into(),
            spec: ExecutionCellSpec {
                id: "argv-too-big".into(),
                correlation: None,
                ingress: None,
                environment: None,
                placement: None,
                policy: None,
                identity: None,
                run: Some(RunSpec {
                    argv: vec!["sh".into(), payload],
                    working_directory: None,
                    timeout_ms: None,
                    limits: None,
                    secret_delivery: SecretDeliveryMode::Env,
                }),
                authority: AuthorityBundle::default(),
                lifetime: Lifetime { ttl_seconds: 1 },
                export: None,
                telemetry: None,
            },
        };
        let err = validate_execution_cell_document(&doc).expect_err("5 KiB argv must reject");
        match err {
            CellosError::ArgvTooLarge {
                encoded_bytes,
                limit_bytes,
            } => {
                assert!(
                    encoded_bytes > limit_bytes,
                    "encoded_bytes ({encoded_bytes}) must exceed limit_bytes ({limit_bytes})"
                );
                assert_eq!(limit_bytes, 3072, "FC-17 budget is 3 KiB");
            }
            other => panic!("expected CellosError::ArgvTooLarge, got: {other:?}"),
        }
    }

    #[test]
    fn rejects_identity_ttl_longer_than_cell_ttl() {
        let doc = ExecutionCellDocument {
            api_version: "cellos.io/v1".into(),
            kind: "ExecutionCell".into(),
            spec: ExecutionCellSpec {
                id: "x".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(120),
                    secret_ref: "AWS_WEB_IDENTITY".into(),
                }),
                run: None,
                authority: AuthorityBundle {
                    filesystem: None,
                    network: None,
                    egress_rules: None,
                    secret_refs: Some(vec!["AWS_WEB_IDENTITY".into()]),
                    authority_derivation: None,
                    dns_authority: None,
                    cdn_authority: None,
                },
                lifetime: Lifetime { ttl_seconds: 60 },
                export: None,
                telemetry: None,
            },
        };
        assert!(validate_execution_cell_document(&doc).is_err());
    }

    #[test]
    fn rejects_unknown_export_target_reference() {
        let doc = ExecutionCellDocument {
            api_version: "cellos.io/v1".into(),
            kind: "ExecutionCell".into(),
            spec: ExecutionCellSpec {
                id: "x".into(),
                correlation: None,
                ingress: None,
                environment: None,
                placement: None,
                policy: None,
                identity: None,
                run: None,
                authority: AuthorityBundle {
                    filesystem: None,
                    network: None,
                    egress_rules: None,
                    secret_refs: Some(vec!["AWS_WEB_IDENTITY".into()]),
                    authority_derivation: None,
                    dns_authority: None,
                    cdn_authority: None,
                },
                lifetime: Lifetime { ttl_seconds: 60 },
                export: Some(ExportChannels {
                    artifacts: Some(vec![ExportArtifact {
                        name: "junit".into(),
                        path: "/tmp/junit.xml".into(),
                        target: Some("missing".into()),
                        content_type: None,
                    }]),
                    targets: Some(vec![ExportTarget::S3(S3ExportTarget {
                        name: "artifacts".into(),
                        bucket: "cellos-artifacts".into(),
                        key_prefix: None,
                        region: None,
                        secret_ref: Some("AWS_WEB_IDENTITY".into()),
                    })]),
                }),
                telemetry: None,
            },
        };
        assert!(validate_execution_cell_document(&doc).is_err());
    }

    #[test]
    fn rejects_spec_id_with_path_traversal() {
        let doc = ExecutionCellDocument {
            api_version: "cellos.io/v1".into(),
            kind: "ExecutionCell".into(),
            spec: ExecutionCellSpec {
                id: "../escape".into(),
                correlation: None,
                ingress: None,
                environment: None,
                placement: None,
                policy: None,
                identity: None,
                run: None,
                authority: AuthorityBundle::default(),
                lifetime: Lifetime { ttl_seconds: 60 },
                export: None,
                telemetry: None,
            },
        };
        assert!(validate_execution_cell_document(&doc).is_err());
    }

    #[test]
    fn rejects_export_artifact_name_with_dotdot() {
        let doc = ExecutionCellDocument {
            api_version: "cellos.io/v1".into(),
            kind: "ExecutionCell".into(),
            spec: ExecutionCellSpec {
                id: "safe-cell".into(),
                correlation: None,
                ingress: None,
                environment: None,
                placement: None,
                policy: None,
                identity: None,
                run: None,
                authority: AuthorityBundle::default(),
                lifetime: Lifetime { ttl_seconds: 60 },
                export: Some(ExportChannels {
                    artifacts: Some(vec![ExportArtifact {
                        name: "bad..name".into(),
                        path: "/tmp/junit.xml".into(),
                        target: None,
                        content_type: None,
                    }]),
                    targets: None,
                }),
                telemetry: None,
            },
        };
        assert!(validate_execution_cell_document(&doc).is_err());
    }

    #[test]
    fn rejects_secret_ref_with_separator() {
        let doc = ExecutionCellDocument {
            api_version: "cellos.io/v1".into(),
            kind: "ExecutionCell".into(),
            spec: ExecutionCellSpec {
                id: "safe-cell".into(),
                correlation: None,
                ingress: None,
                environment: None,
                placement: None,
                policy: None,
                identity: None,
                run: None,
                authority: AuthorityBundle {
                    filesystem: None,
                    network: None,
                    egress_rules: None,
                    secret_refs: Some(vec!["bad/ref".into()]),
                    authority_derivation: None,
                    dns_authority: None,
                    cdn_authority: None,
                },
                lifetime: Lifetime { ttl_seconds: 60 },
                export: None,
                telemetry: None,
            },
        };
        assert!(validate_execution_cell_document(&doc).is_err());
    }

    #[test]
    fn rejects_http_export_target_with_non_http_base_url() {
        let doc = ExecutionCellDocument {
            api_version: "cellos.io/v1".into(),
            kind: "ExecutionCell".into(),
            spec: ExecutionCellSpec {
                id: "safe-cell".into(),
                correlation: None,
                ingress: None,
                environment: None,
                placement: None,
                policy: None,
                identity: None,
                run: None,
                authority: AuthorityBundle {
                    filesystem: None,
                    network: None,
                    egress_rules: None,
                    secret_refs: Some(vec!["ARTIFACT_API_TOKEN".into()]),
                    authority_derivation: None,
                    dns_authority: None,
                    cdn_authority: None,
                },
                lifetime: Lifetime { ttl_seconds: 60 },
                export: Some(ExportChannels {
                    artifacts: Some(vec![ExportArtifact {
                        name: "coverage-summary".into(),
                        path: "/tmp/coverage.txt".into(),
                        target: Some("artifact-api".into()),
                        content_type: Some("text/plain".into()),
                    }]),
                    targets: Some(vec![ExportTarget::Http(HttpExportTarget {
                        name: "artifact-api".into(),
                        base_url: "ftp://artifacts.example.invalid/upload".into(),
                        secret_ref: Some("ARTIFACT_API_TOKEN".into()),
                    })]),
                }),
                telemetry: None,
            },
        };
        assert!(validate_execution_cell_document(&doc).is_err());
    }

    // ── Property-based tests (SEC-11) ─────────────────────────────────────
    // These cover the identifier validation rules systematically so that
    // hand-crafted inputs can't be the only signal of correctness.

    use super::is_portable_identifier;
    use proptest::prelude::*;

    fn minimal_doc_with_placement(placement: crate::PlacementSpec) -> ExecutionCellDocument {
        ExecutionCellDocument {
            api_version: "cellos.io/v1".into(),
            kind: "ExecutionCell".into(),
            spec: ExecutionCellSpec {
                id: "placement-test-cell".into(),
                correlation: None,
                ingress: None,
                environment: None,
                placement: Some(placement),
                policy: None,
                identity: None,
                run: None,
                authority: AuthorityBundle::default(),
                lifetime: Lifetime { ttl_seconds: 60 },
                export: None,
                telemetry: None,
            },
        }
    }

    #[test]
    fn rejects_placement_pool_id_with_separator() {
        let doc = minimal_doc_with_placement(crate::PlacementSpec {
            pool_id: Some("pool/main".into()),
            kubernetes_namespace: None,
            queue_name: None,
        });
        let err = validate_execution_cell_document(&doc).unwrap_err();
        assert!(err.to_string().contains("spec.placement.poolId"));
    }

    #[test]
    fn rejects_empty_placement_hints() {
        let doc = minimal_doc_with_placement(crate::PlacementSpec::default());
        let err = validate_execution_cell_document(&doc).unwrap_err();
        assert!(err
            .to_string()
            .contains("spec.placement must set at least one placement hint"));
    }

    #[test]
    fn rejects_placement_queue_name_with_dotdot() {
        let doc = minimal_doc_with_placement(crate::PlacementSpec {
            pool_id: None,
            kubernetes_namespace: None,
            queue_name: Some("ci..high".into()),
        });
        let err = validate_execution_cell_document(&doc).unwrap_err();
        assert!(err.to_string().contains("spec.placement.queueName"));
    }

    #[test]
    fn rejects_placement_namespace_with_uppercase() {
        let doc = minimal_doc_with_placement(crate::PlacementSpec {
            pool_id: None,
            kubernetes_namespace: Some("CellOS-Prod".into()),
            queue_name: None,
        });
        let err = validate_execution_cell_document(&doc).unwrap_err();
        assert!(err
            .to_string()
            .contains("spec.placement.kubernetesNamespace"));
    }

    #[test]
    fn accepts_valid_placement_hints() {
        let doc = minimal_doc_with_placement(crate::PlacementSpec {
            pool_id: Some("runner-pool-amd64".into()),
            kubernetes_namespace: Some("cellos-prod".into()),
            queue_name: Some("ci-high".into()),
        });
        assert!(validate_execution_cell_document(&doc).is_ok());
    }

    proptest! {
        /// Any identifier containing a forward slash is rejected.
        #[test]
        fn prop_rejects_slash_in_identifier(
            prefix in "[A-Za-z0-9._-]{0,20}",
            suffix in "[A-Za-z0-9._-]{0,20}",
        ) {
            let with_slash = format!("{prefix}/{suffix}");
            prop_assert!(!is_portable_identifier(&with_slash), "slash in {with_slash:?}");
        }

        /// Any identifier containing a backslash is rejected.
        #[test]
        fn prop_rejects_backslash_in_identifier(
            prefix in "[A-Za-z0-9._-]{0,20}",
            suffix in "[A-Za-z0-9._-]{0,20}",
        ) {
            let with_bs = format!("{prefix}\\{suffix}");
            prop_assert!(!is_portable_identifier(&with_bs), "backslash in {with_bs:?}");
        }

        /// Any identifier containing `..` is rejected regardless of surrounding chars.
        #[test]
        fn prop_rejects_dotdot_in_identifier(
            prefix in "[A-Za-z0-9._-]{0,20}",
            suffix in "[A-Za-z0-9._-]{0,20}",
        ) {
            let with_dotdot = format!("{prefix}..{suffix}");
            prop_assert!(!is_portable_identifier(&with_dotdot), ".. in {with_dotdot:?}");
        }

        /// Any identifier starting with a non-alphanumeric ASCII char is rejected.
        #[test]
        fn prop_rejects_non_alphanum_first_char(
            first in "[._\\-]",
            rest in "[A-Za-z0-9._-]{0,20}",
        ) {
            let s = format!("{first}{rest}");
            prop_assert!(!is_portable_identifier(&s), "starts with non-alphanum: {s:?}");
        }

        /// Valid identifiers (alphanumeric start, safe body, ≤128 chars, no ..) are accepted.
        #[test]
        fn prop_accepts_valid_identifiers(
            first in "[A-Za-z0-9]",
            rest in "[A-Za-z0-9._-]{0,60}",
        ) {
            let s = format!("{first}{rest}");
            // Exclude generated strings that happen to contain ".."
            prop_assume!(!s.contains(".."));
            prop_assert!(is_portable_identifier(&s), "should accept {s:?}");
        }

        /// Validate that any spec.id failing is_portable_identifier causes validate() to error.
        #[test]
        fn prop_invalid_spec_id_rejected_by_validate(
            prefix in "[A-Za-z0-9._-]{0,10}",
            suffix in "[A-Za-z0-9._-]{0,10}",
        ) {
            // inject a forward slash so it definitely fails the portable identifier check
            let bad_id = format!("{prefix}/{suffix}");
            let doc = ExecutionCellDocument {
                api_version: "cellos.io/v1".into(),
                kind: "ExecutionCell".into(),
                spec: ExecutionCellSpec {
                    id: bad_id.clone(),
                    correlation: None,
                    ingress: None,
                environment: None,
                    placement: None,
                    policy: None,
                    identity: None,
                    run: None,
                    authority: AuthorityBundle::default(),
                    lifetime: Lifetime { ttl_seconds: 60 },
                    export: None,
                    telemetry: None,
                },
            };
            prop_assert!(
                validate_execution_cell_document(&doc).is_err(),
                "expected error for id {bad_id:?}"
            );
        }

        /// Validate that spec.run.argv with any empty token causes validate() to error.
        #[test]
        fn prop_empty_argv_token_rejected(
            prefix_args in prop::collection::vec("[A-Za-z0-9/_-]{1,20}", 0..5),
            suffix_args in prop::collection::vec("[A-Za-z0-9/_-]{1,20}", 0..5),
        ) {
            // build argv with an empty string inserted in the middle
            let mut argv: Vec<String> = prefix_args;
            argv.push(String::new()); // the bad token
            argv.extend(suffix_args);

            let doc = ExecutionCellDocument {
                api_version: "cellos.io/v1".into(),
                kind: "ExecutionCell".into(),
                spec: ExecutionCellSpec {
                    id: "valid-cell".into(),
                    correlation: None,
                    ingress: None,
                environment: None,
                    placement: None,
                    policy: None,
                    identity: None,
                    run: Some(RunSpec {
                        argv,
                        working_directory: None,
                        timeout_ms: None,
                        limits: None,
                        secret_delivery: SecretDeliveryMode::Env,
                    }),
                    authority: AuthorityBundle::default(),
                    lifetime: Lifetime { ttl_seconds: 60 },
                    export: None,
                    telemetry: None,
                },
            };
            prop_assert!(
                validate_execution_cell_document(&doc).is_err(),
                "empty argv token should be rejected"
            );
        }
    }

    #[test]
    fn rejects_http_export_target_missing_egress_rule() {
        let doc = ExecutionCellDocument {
            api_version: "cellos.io/v1".into(),
            kind: "ExecutionCell".into(),
            spec: ExecutionCellSpec {
                id: "safe-cell".into(),
                correlation: None,
                ingress: None,
                environment: None,
                placement: None,
                policy: None,
                identity: None,
                run: None,
                authority: AuthorityBundle {
                    filesystem: None,
                    network: None,
                    egress_rules: Some(vec![crate::EgressRule {
                        host: "api.github.com".into(),
                        port: 443,
                        protocol: Some("tls".into()),
                        dns_egress_justification: None,
                    }]),
                    secret_refs: Some(vec!["ARTIFACT_API_TOKEN".into()]),
                    authority_derivation: None,
                    dns_authority: None,
                    cdn_authority: None,
                },
                lifetime: Lifetime { ttl_seconds: 60 },
                export: Some(ExportChannels {
                    artifacts: Some(vec![ExportArtifact {
                        name: "coverage-summary".into(),
                        path: "/tmp/coverage.txt".into(),
                        target: Some("artifact-api".into()),
                        content_type: Some("text/plain".into()),
                    }]),
                    targets: Some(vec![ExportTarget::Http(HttpExportTarget {
                        name: "artifact-api".into(),
                        base_url: "https://artifacts.acme.internal/upload".into(),
                        secret_ref: Some("ARTIFACT_API_TOKEN".into()),
                    })]),
                }),
                telemetry: None,
            },
        };
        assert!(validate_execution_cell_document(&doc).is_err());
    }

    #[test]
    fn rejects_run_timeout_longer_than_lifetime() {
        let doc = ExecutionCellDocument {
            api_version: "cellos.io/v1".into(),
            kind: "ExecutionCell".into(),
            spec: ExecutionCellSpec {
                id: "safe-cell".into(),
                correlation: None,
                ingress: None,
                environment: None,
                placement: None,
                policy: None,
                identity: None,
                run: Some(RunSpec {
                    argv: vec!["/usr/bin/true".into()],
                    working_directory: None,
                    timeout_ms: Some(61_000),
                    limits: None,
                    secret_delivery: SecretDeliveryMode::Env,
                }),
                authority: AuthorityBundle::default(),
                lifetime: Lifetime { ttl_seconds: 60 },
                export: None,
                telemetry: None,
            },
        };
        assert!(validate_execution_cell_document(&doc).is_err());
    }

    /// Boundary case for review-2026-04-08 gap #2: `timeout_ms == ttl * 1000`
    /// must be accepted (the validator uses `>`, not `>=`) so an operator
    /// can declare an honest "use the full TTL as the soft cap" spec.
    #[test]
    fn accepts_run_timeout_equal_to_lifetime() {
        let doc = ExecutionCellDocument {
            api_version: "cellos.io/v1".into(),
            kind: "ExecutionCell".into(),
            spec: ExecutionCellSpec {
                id: "safe-cell-boundary".into(),
                correlation: None,
                ingress: None,
                environment: None,
                placement: None,
                policy: None,
                identity: None,
                run: Some(RunSpec {
                    argv: vec!["/usr/bin/true".into()],
                    working_directory: None,
                    timeout_ms: Some(60_000),
                    limits: None,
                    secret_delivery: SecretDeliveryMode::Env,
                }),
                authority: AuthorityBundle::default(),
                lifetime: Lifetime { ttl_seconds: 60 },
                export: None,
                telemetry: None,
            },
        };
        assert!(
            validate_execution_cell_document(&doc).is_ok(),
            "timeout_ms == ttl_seconds * 1000 must be accepted (boundary)"
        );
    }

    /// Just-over-the-boundary case for gap #2: `timeout_ms == ttl*1000 + 1`
    /// must be rejected. Pins the precision of the comparison so a
    /// future refactor that switches to seconds (lossy) cannot pass.
    #[test]
    fn rejects_run_timeout_one_ms_over_lifetime() {
        let doc = ExecutionCellDocument {
            api_version: "cellos.io/v1".into(),
            kind: "ExecutionCell".into(),
            spec: ExecutionCellSpec {
                id: "safe-cell-overshoot".into(),
                correlation: None,
                ingress: None,
                environment: None,
                placement: None,
                policy: None,
                identity: None,
                run: Some(RunSpec {
                    argv: vec!["/usr/bin/true".into()],
                    working_directory: None,
                    timeout_ms: Some(60_001),
                    limits: None,
                    secret_delivery: SecretDeliveryMode::Env,
                }),
                authority: AuthorityBundle::default(),
                lifetime: Lifetime { ttl_seconds: 60 },
                export: None,
                telemetry: None,
            },
        };
        let err = validate_execution_cell_document(&doc).expect_err("must reject");
        let msg = format!("{err}");
        assert!(
            msg.contains("timeoutMs"),
            "error must mention timeoutMs; got {msg:?}"
        );
        assert!(
            msg.contains("ttlSeconds"),
            "error must mention ttlSeconds; got {msg:?}"
        );
    }

    #[test]
    fn accepts_run_limits_and_timeout_within_lifetime() {
        let doc = ExecutionCellDocument {
            api_version: "cellos.io/v1".into(),
            kind: "ExecutionCell".into(),
            spec: ExecutionCellSpec {
                id: "safe-cell".into(),
                correlation: None,
                ingress: None,
                environment: None,
                placement: None,
                policy: None,
                identity: None,
                run: Some(RunSpec {
                    argv: vec!["/usr/bin/true".into()],
                    working_directory: None,
                    timeout_ms: Some(5_000),
                    limits: Some(RunLimits {
                        memory_max_bytes: Some(268_435_456),
                        cpu_max: Some(RunCpuMax {
                            quota_micros: 50_000,
                            period_micros: Some(100_000),
                        }),
                        graceful_shutdown_seconds: None,
                    }),
                    secret_delivery: SecretDeliveryMode::Env,
                }),
                authority: AuthorityBundle::default(),
                lifetime: Lifetime { ttl_seconds: 60 },
                export: None,
                telemetry: None,
            },
        };
        assert!(validate_execution_cell_document(&doc).is_ok());
    }

    // ── spec.environment validation ───────────────────────────────────────────

    fn minimal_doc_with_env(env: EnvironmentSpec) -> ExecutionCellDocument {
        ExecutionCellDocument {
            api_version: "cellos.io/v1".into(),
            kind: "ExecutionCell".into(),
            spec: ExecutionCellSpec {
                id: "env-test-cell".into(),
                correlation: None,
                ingress: None,
                environment: Some(env),
                placement: None,
                policy: None,
                identity: None,
                run: None,
                authority: AuthorityBundle::default(),
                lifetime: Lifetime { ttl_seconds: 60 },
                export: None,
                telemetry: None,
            },
        }
    }

    #[test]
    fn accepts_environment_with_reference_only() {
        let doc = minimal_doc_with_env(EnvironmentSpec {
            image_reference: "ubuntu:24.04".into(),
            image_digest: None,
            template_id: None,
        });
        assert!(validate_execution_cell_document(&doc).is_ok());
    }

    #[test]
    fn accepts_environment_with_digest_and_template() {
        let doc = minimal_doc_with_env(EnvironmentSpec {
            image_reference: "ubuntu:24.04".into(),
            image_digest: Some(
                "sha256:a1b2c3d4e5f6a1b2c3d4e5f6a1b2c3d4e5f6a1b2c3d4e5f6a1b2c3d4e5f6a1b2".into(),
            ),
            template_id: Some("ubuntu-24-04-build".into()),
        });
        assert!(validate_execution_cell_document(&doc).is_ok());
    }

    #[test]
    fn rejects_environment_with_empty_image_reference() {
        let doc = minimal_doc_with_env(EnvironmentSpec {
            image_reference: "".into(),
            image_digest: None,
            template_id: None,
        });
        let err = validate_execution_cell_document(&doc).unwrap_err();
        assert!(
            err.to_string().contains("imageReference"),
            "expected imageReference in error, got: {err}"
        );
    }

    #[test]
    fn rejects_environment_with_bad_digest_missing_prefix() {
        let doc = minimal_doc_with_env(EnvironmentSpec {
            image_reference: "ubuntu:24.04".into(),
            image_digest: Some(
                "a1b2c3d4e5f6a1b2c3d4e5f6a1b2c3d4e5f6a1b2c3d4e5f6a1b2c3d4e5f6a1b2".into(),
            ),
            template_id: None,
        });
        let err = validate_execution_cell_document(&doc).unwrap_err();
        assert!(err.to_string().contains("imageDigest"), "{err}");
    }

    #[test]
    fn rejects_environment_with_digest_wrong_length() {
        let doc = minimal_doc_with_env(EnvironmentSpec {
            image_reference: "ubuntu:24.04".into(),
            image_digest: Some("sha256:deadbeef".into()),
            template_id: None,
        });
        let err = validate_execution_cell_document(&doc).unwrap_err();
        assert!(err.to_string().contains("imageDigest"), "{err}");
    }

    #[test]
    fn rejects_environment_with_uppercase_digest() {
        let doc = minimal_doc_with_env(EnvironmentSpec {
            image_reference: "ubuntu:24.04".into(),
            image_digest: Some(
                "sha256:A1B2C3D4E5F6A1B2C3D4E5F6A1B2C3D4E5F6A1B2C3D4E5F6A1B2C3D4E5F6A1B2".into(),
            ),
            template_id: None,
        });
        let err = validate_execution_cell_document(&doc).unwrap_err();
        assert!(err.to_string().contains("imageDigest"), "{err}");
    }

    #[test]
    fn rejects_environment_with_invalid_template_id() {
        let doc = minimal_doc_with_env(EnvironmentSpec {
            image_reference: "ubuntu:24.04".into(),
            image_digest: None,
            template_id: Some("../escape".into()),
        });
        let err = validate_execution_cell_document(&doc).unwrap_err();
        assert!(err.to_string().contains("templateId"), "{err}");
    }

    #[test]
    fn is_sha256_digest_accepts_valid() {
        assert!(is_sha256_digest(
            "sha256:a1b2c3d4e5f6a1b2c3d4e5f6a1b2c3d4e5f6a1b2c3d4e5f6a1b2c3d4e5f6a1b2"
        ));
    }

    #[test]
    fn is_sha256_digest_rejects_short() {
        assert!(!is_sha256_digest("sha256:deadbeef"));
    }

    #[test]
    fn is_sha256_digest_rejects_no_prefix() {
        assert!(!is_sha256_digest(
            "a1b2c3d4e5f6a1b2c3d4e5f6a1b2c3d4e5f6a1b2c3d4e5f6a1b2c3d4e5f6a1b2"
        ));
    }

    #[test]
    fn is_sha256_digest_rejects_uppercase() {
        assert!(!is_sha256_digest(
            "sha256:A1B2C3D4E5F6A1B2C3D4E5F6A1B2C3D4E5F6A1B2C3D4E5F6A1B2C3D4E5F6A1B2"
        ));
    }

    fn derivation_spec_with_authority(
        egress: Vec<crate::EgressRule>,
        secret_refs: Vec<String>,
    ) -> ExecutionCellSpec {
        ExecutionCellSpec {
            id: "deriv-test".into(),
            correlation: None,
            ingress: None,
            environment: None,
            placement: None,
            policy: None,
            identity: None,
            run: None,
            authority: AuthorityBundle {
                filesystem: None,
                network: None,
                egress_rules: Some(egress),
                secret_refs: Some(secret_refs),
                authority_derivation: None,
                dns_authority: None,
                cdn_authority: None,
            },
            lifetime: Lifetime { ttl_seconds: 60 },
            export: None,
            telemetry: None,
        }
    }

    fn token_with_leaf(leaf: crate::AuthorityCapability) -> crate::AuthorityDerivationToken {
        crate::AuthorityDerivationToken {
            role_root: crate::RoleId("role-test".into()),
            parent_run_id: None,
            derivation_steps: vec![],
            leaf_capability: leaf,
            grantor_signature: crate::AuthoritySignature {
                algorithm: "ed25519".into(),
                bytes: "AA==".into(),
            },
        }
    }

    /// Sign `token` with `signing_key`, returning a token whose `grantor_signature.bytes`
    /// is the base64-STANDARD encoding of the resulting ED25519 signature over the
    /// canonical signing payload.
    fn sign_token(
        signing_key: &ed25519_dalek::SigningKey,
        mut token: crate::AuthorityDerivationToken,
    ) -> crate::AuthorityDerivationToken {
        use base64::engine::general_purpose::STANDARD;
        use base64::Engine as _;
        use ed25519_dalek::Signer as _;

        let payload = authority_derivation_signing_payload(&token).expect("payload encode");
        let signature = signing_key.sign(&payload);
        token.grantor_signature.bytes = STANDARD.encode(signature.to_bytes());
        token
    }

    /// Build a deterministic ED25519 signing key from a fixed seed — avoids pulling
    /// `rand_core` and `OsRng` into dev-deps just for tests.
    fn test_signing_key(seed: u8) -> ed25519_dalek::SigningKey {
        ed25519_dalek::SigningKey::from_bytes(&[seed; 32])
    }

    fn verifying_key_b64(signing_key: &ed25519_dalek::SigningKey) -> String {
        use base64::engine::general_purpose::STANDARD;
        use base64::Engine as _;
        STANDARD.encode(signing_key.verifying_key().to_bytes())
    }

    #[test]
    fn verify_authority_derivation_child_within_parent_passes() {
        // Structural pass — signature still required for the new full verify.
        let spec = derivation_spec_with_authority(
            vec![crate::EgressRule {
                host: "api.example.com".into(),
                port: 443,
                protocol: Some("https".into()),
                dns_egress_justification: None,
            }],
            vec!["api-key".into()],
        );
        let signing_key = test_signing_key(0x42);
        let token = sign_token(
            &signing_key,
            token_with_leaf(crate::AuthorityCapability {
                egress_rules: vec![crate::EgressRule {
                    host: "api.example.com".into(),
                    port: 443,
                    protocol: Some("https".into()),
                    dns_egress_justification: None,
                }],
                secret_refs: vec!["api-key".into()],
            }),
        );
        let mut keys = std::collections::HashMap::new();
        keys.insert("role-test".to_string(), verifying_key_b64(&signing_key));
        assert!(verify_authority_derivation(&spec, &token, &keys).is_ok());
    }

    #[test]
    fn verify_authority_derivation_child_exceeding_parent_fails_egress() {
        let spec = derivation_spec_with_authority(
            vec![crate::EgressRule {
                host: "api.example.com".into(),
                port: 443,
                protocol: Some("https".into()),
                dns_egress_justification: None,
            }],
            vec![],
        );
        let token = token_with_leaf(crate::AuthorityCapability {
            egress_rules: vec![crate::EgressRule {
                host: "evil.example.com".into(),
                port: 443,
                protocol: Some("https".into()),
                dns_egress_justification: None,
            }],
            secret_refs: vec![],
        });
        let keys = std::collections::HashMap::new();
        let err = verify_authority_derivation(&spec, &token, &keys).unwrap_err();
        assert!(err.to_string().contains("leafCapability"), "{err}");
    }

    #[test]
    fn verify_authority_derivation_child_exceeding_parent_fails_secrets() {
        let spec = derivation_spec_with_authority(vec![], vec!["api-key".into()]);
        let token = token_with_leaf(crate::AuthorityCapability {
            egress_rules: vec![],
            secret_refs: vec!["root-token".into()],
        });
        let keys = std::collections::HashMap::new();
        let err = verify_authority_derivation(&spec, &token, &keys).unwrap_err();
        assert!(err.to_string().contains("leafCapability"), "{err}");
    }

    #[test]
    fn validate_doc_rejects_spec_with_exceeding_derivation_token() {
        let mut spec = derivation_spec_with_authority(
            vec![crate::EgressRule {
                host: "api.example.com".into(),
                port: 443,
                protocol: Some("https".into()),
                dns_egress_justification: None,
            }],
            vec![],
        );
        spec.authority.authority_derivation = Some(token_with_leaf(crate::AuthorityCapability {
            egress_rules: vec![crate::EgressRule {
                host: "evil.example.com".into(),
                port: 443,
                protocol: Some("https".into()),
                dns_egress_justification: None,
            }],
            secret_refs: vec![],
        }));
        let doc = ExecutionCellDocument {
            api_version: "cellos.io/v1".into(),
            kind: "ExecutionCell".into(),
            spec,
        };
        let err = validate_execution_cell_document(&doc).unwrap_err();
        assert!(err.to_string().contains("leafCapability"), "{err}");
    }

    #[test]
    fn validate_doc_accepts_spec_with_valid_derivation_token() {
        let mut spec = derivation_spec_with_authority(
            vec![crate::EgressRule {
                host: "api.example.com".into(),
                port: 443,
                protocol: Some("https".into()),
                dns_egress_justification: None,
            }],
            vec!["api-key".into()],
        );
        spec.authority.authority_derivation = Some(token_with_leaf(crate::AuthorityCapability {
            egress_rules: vec![crate::EgressRule {
                host: "api.example.com".into(),
                port: 443,
                protocol: Some("https".into()),
                dns_egress_justification: None,
            }],
            secret_refs: vec!["api-key".into()],
        }));
        let doc = ExecutionCellDocument {
            api_version: "cellos.io/v1".into(),
            kind: "ExecutionCell".into(),
            spec,
        };
        assert!(validate_execution_cell_document(&doc).is_ok());
    }

    // ── ED25519 signature verification (L5-14) ─────────────────────────────

    fn signed_spec_and_token() -> (
        crate::ExecutionCellSpec,
        crate::AuthorityDerivationToken,
        ed25519_dalek::SigningKey,
    ) {
        let spec = derivation_spec_with_authority(
            vec![crate::EgressRule {
                host: "api.example.com".into(),
                port: 443,
                protocol: Some("https".into()),
                dns_egress_justification: None,
            }],
            vec!["api-key".into()],
        );
        let signing_key = test_signing_key(0x11);
        let token = sign_token(
            &signing_key,
            token_with_leaf(crate::AuthorityCapability {
                egress_rules: vec![crate::EgressRule {
                    host: "api.example.com".into(),
                    port: 443,
                    protocol: Some("https".into()),
                    dns_egress_justification: None,
                }],
                secret_refs: vec!["api-key".into()],
            }),
        );
        (spec, token, signing_key)
    }

    #[test]
    fn test_good_signature_passes() {
        let (spec, token, signing_key) = signed_spec_and_token();
        let mut keys = std::collections::HashMap::new();
        keys.insert("role-test".to_string(), verifying_key_b64(&signing_key));
        assert!(verify_authority_derivation(&spec, &token, &keys).is_ok());
    }

    #[test]
    fn test_bad_signature_rejected() {
        let (spec, mut token, signing_key) = signed_spec_and_token();
        // Tamper: flip the last byte of the base64-decoded signature, then re-encode.
        use base64::engine::general_purpose::STANDARD;
        use base64::Engine as _;
        let mut sig_bytes = STANDARD
            .decode(token.grantor_signature.bytes.as_bytes())
            .expect("decode original signature");
        let last = sig_bytes.len() - 1;
        sig_bytes[last] ^= 0x01;
        token.grantor_signature.bytes = STANDARD.encode(&sig_bytes);

        let mut keys = std::collections::HashMap::new();
        keys.insert("role-test".to_string(), verifying_key_b64(&signing_key));
        let err = verify_authority_derivation(&spec, &token, &keys).unwrap_err();
        match err {
            crate::error::CellosError::InvalidSpec(msg) => {
                assert!(
                    msg.contains("authority derivation signature invalid"),
                    "unexpected error message: {msg}"
                );
            }
            other => panic!("expected InvalidSpec, got {other:?}"),
        }
    }

    #[test]
    fn test_unknown_role_rejected() {
        let (spec, token, _signing_key) = signed_spec_and_token();
        let other_key = test_signing_key(0x22);
        let mut keys = std::collections::HashMap::new();
        // Map only contains a different role — token's role-test is unknown.
        keys.insert("role-other".to_string(), verifying_key_b64(&other_key));
        let err = verify_authority_derivation(&spec, &token, &keys).unwrap_err();
        match err {
            crate::error::CellosError::InvalidSpec(msg) => {
                assert!(msg.contains("unknown role"), "unexpected message: {msg}");
                assert!(
                    msg.contains("role-test"),
                    "expected role id in message: {msg}"
                );
            }
            other => panic!("expected InvalidSpec, got {other:?}"),
        }
    }

    #[test]
    fn test_no_token_passes() {
        // Build a doc whose spec has NO authorityDerivation token — it must
        // pass validation unchanged regardless of role_keys availability.
        let spec = derivation_spec_with_authority(
            vec![crate::EgressRule {
                host: "api.example.com".into(),
                port: 443,
                protocol: Some("https".into()),
                dns_egress_justification: None,
            }],
            vec!["api-key".into()],
        );
        assert!(spec.authority.authority_derivation.is_none());
        let doc = ExecutionCellDocument {
            api_version: "cellos.io/v1".into(),
            kind: "ExecutionCell".into(),
            spec,
        };
        assert!(validate_execution_cell_document(&doc).is_ok());
    }

    #[test]
    fn test_empty_keys_with_token_fails() {
        let (spec, token, _signing_key) = signed_spec_and_token();
        let keys: std::collections::HashMap<String, String> = std::collections::HashMap::new();
        let err = verify_authority_derivation(&spec, &token, &keys).unwrap_err();
        match err {
            crate::error::CellosError::InvalidSpec(msg) => {
                assert!(msg.contains("unknown role"), "unexpected message: {msg}");
            }
            other => panic!("expected InvalidSpec, got {other:?}"),
        }
    }

    // ── parentRunId scope policy (L5-16) ──────────────────────────────────

    #[test]
    fn enforce_scope_policy_universal_allowed_when_permissive() {
        let (_spec, token, _signing_key) = signed_spec_and_token();
        assert!(token.parent_run_id.is_none());
        // Permissive mode: WARN-and-pass.
        assert!(enforce_derivation_scope_policy(&token, true).is_ok());
    }

    #[test]
    fn enforce_scope_policy_universal_rejected_when_strict() {
        let (_spec, token, _signing_key) = signed_spec_and_token();
        assert!(token.parent_run_id.is_none());
        let err = enforce_derivation_scope_policy(&token, false).unwrap_err();
        match err {
            crate::error::CellosError::InvalidSpec(msg) => {
                assert!(
                    msg.contains("parentRunId: null")
                        && msg.contains("CELLOS_REQUIRE_SCOPED_DERIVATION_TOKENS"),
                    "unexpected message: {msg}"
                );
            }
            other => panic!("expected InvalidSpec, got {other:?}"),
        }
    }

    #[test]
    fn enforce_scope_policy_scoped_token_passes_in_either_mode() {
        let (_spec, mut token, _signing_key) = signed_spec_and_token();
        token.parent_run_id = Some("run-2026-04-25-abc".into());
        assert!(enforce_derivation_scope_policy(&token, true).is_ok());
        assert!(enforce_derivation_scope_policy(&token, false).is_ok());
    }

    // ── Fixture generator (documentation, not a CI gate) ──────────────────
    //
    // Regenerates the signature embedded in
    // `contracts/examples/execution-cell-ci-runner-signed.valid.json` and the
    // verifying key in `contracts/examples/authority-keys.example.json`.
    //
    // Run with:
    //   cargo test -p cellos-core --lib gen_signed_ci_runner_fixture \
    //     -- --ignored --nocapture
    //
    // Seed: [0x42; 32]  ·  Role: "role-ci-runner"  ·  parentRunId: "run-demo-ref-001"
    //
    // The leaf capability constructed here MUST stay byte-identical with the
    // `leafCapability` JSON object in the fixture file (same fields, same
    // values, same serde-camelCase rendering) — otherwise the canonical
    // signing payload diverges and the signature stops verifying.
    #[test]
    #[ignore]
    fn gen_signed_ci_runner_fixture() {
        let signing_key = test_signing_key(0x42);
        let leaf = crate::AuthorityCapability {
            egress_rules: vec![
                crate::EgressRule {
                    host: "api.github.com".into(),
                    port: 443,
                    protocol: Some("tls".into()),
                    dns_egress_justification: None,
                },
                crate::EgressRule {
                    host: "ghcr.io".into(),
                    port: 443,
                    protocol: Some("tls".into()),
                    dns_egress_justification: None,
                },
                crate::EgressRule {
                    host: "artifacts.internal".into(),
                    port: 443,
                    protocol: Some("tls".into()),
                    dns_egress_justification: None,
                },
                crate::EgressRule {
                    host: "dns.internal".into(),
                    port: 53,
                    protocol: Some("dns-acknowledged".into()),
                    dns_egress_justification: Some(
                        "Internal resolver at dns.internal required for artifacts.internal hostname resolution; nameserver is operator-controlled and air-gapped from public internet.".into(),
                    ),
                },
            ],
            secret_refs: vec!["NPM_TOKEN".into(), "GITHUB_TOKEN".into()],
        };
        let token = crate::AuthorityDerivationToken {
            role_root: crate::RoleId("role-ci-runner".into()),
            parent_run_id: Some("run-demo-ref-001".into()),
            derivation_steps: vec![],
            leaf_capability: leaf,
            grantor_signature: crate::AuthoritySignature {
                algorithm: "ed25519".into(),
                bytes: "AA==".into(),
            },
        };
        let signed = sign_token(&signing_key, token);
        eprintln!("FIXTURE seed=0x42 role=role-ci-runner parentRunId=run-demo-ref-001");
        eprintln!(
            "FIXTURE verifyingKey(b64): {}",
            verifying_key_b64(&signing_key)
        );
        eprintln!(
            "FIXTURE grantorSignature(b64): {}",
            signed.grantor_signature.bytes
        );
    }

    // ── T13 dnsAuthority / cdnAuthority validation (SEC-20) ───────────────

    fn doc_with_authority(authority: AuthorityBundle) -> ExecutionCellDocument {
        ExecutionCellDocument {
            api_version: "cellos.io/v1".into(),
            kind: "ExecutionCell".into(),
            spec: ExecutionCellSpec {
                id: "dns-cdn-test-cell".into(),
                correlation: None,
                ingress: None,
                environment: None,
                placement: None,
                policy: None,
                identity: None,
                run: None,
                authority,
                lifetime: Lifetime { ttl_seconds: 60 },
                export: None,
                telemetry: None,
            },
        }
    }

    fn full_dns_authority() -> crate::DnsAuthority {
        crate::DnsAuthority {
            resolvers: vec![crate::DnsResolver {
                resolver_id: "internal-doh".into(),
                endpoint: "https://1.1.1.1/dns-query".into(),
                protocol: crate::DnsResolverProtocol::Doh,
                trust_kid: Some("resolver-kid-2026q2".into()),
                dnssec: None,
            }],
            allowed_query_types: vec![crate::DnsQueryType::A, crate::DnsQueryType::AAAA],
            hostname_allowlist: vec!["api.example.com".into(), "*.cdn.example.com".into()],
            refresh_policy: Some(crate::DnsRefreshPolicy {
                min_ttl_seconds: Some(30),
                max_stale_seconds: Some(300),
                strategy: Some(crate::DnsRefreshStrategy::TtlHonor),
            }),
            // SEC-21 Phase 3e — keep the test fixture's `dnsAuthority`
            // backward-compat shape (rebinding policy unset).
            rebinding_policy: None,
            block_direct_workload_dns: true,
            // SEC-22 Phase 3d — UDP-side kernel block opt-ins default off
            // for the spec-validation fixture so existing tests keep their
            // pre-3d shape; the nft generator tests below exercise the
            // true case.
            block_udp_doq: false,
            block_udp_http3: false,
        }
    }

    #[test]
    fn t13_accepts_full_dns_and_cdn_authority() {
        let authority = AuthorityBundle {
            filesystem: None,
            network: None,
            egress_rules: None,
            secret_refs: None,
            authority_derivation: None,
            dns_authority: Some(full_dns_authority()),
            cdn_authority: Some(crate::CdnAuthority {
                providers: vec![crate::CdnProvider {
                    provider_id: "cloudfront".into(),
                    hostname_pattern: "*.cdn.example.com".into(),
                    accept_fronting: false,
                }],
            }),
        };
        let doc = doc_with_authority(authority);
        validate_execution_cell_document(&doc).expect("full T13 authority should validate");
    }

    #[test]
    fn t13_rejects_dns_hostname_allowlist_with_ip_literal() {
        let mut dns = full_dns_authority();
        dns.hostname_allowlist = vec!["10.0.0.1".into()];
        let authority = AuthorityBundle {
            dns_authority: Some(dns),
            ..AuthorityBundle::default()
        };
        let err = validate_execution_cell_document(&doc_with_authority(authority)).unwrap_err();
        assert!(
            err.to_string()
                .contains("authority.dnsAuthority.hostnameAllowlist"),
            "{err}"
        );
    }

    #[test]
    fn t13_rejects_dns_hostname_allowlist_with_internal_wildcard() {
        let mut dns = full_dns_authority();
        // Wildcard must be the leading label only — "api.*.example.com" is not allowed.
        dns.hostname_allowlist = vec!["api.*.example.com".into()];
        let authority = AuthorityBundle {
            dns_authority: Some(dns),
            ..AuthorityBundle::default()
        };
        let err = validate_execution_cell_document(&doc_with_authority(authority)).unwrap_err();
        assert!(
            err.to_string()
                .contains("authority.dnsAuthority.hostnameAllowlist"),
            "{err}"
        );
    }

    #[test]
    fn t13_rejects_dns_resolver_with_invalid_resolver_id() {
        let mut dns = full_dns_authority();
        dns.resolvers[0].resolver_id = "../escape".into();
        let authority = AuthorityBundle {
            dns_authority: Some(dns),
            ..AuthorityBundle::default()
        };
        let err = validate_execution_cell_document(&doc_with_authority(authority)).unwrap_err();
        assert!(
            err.to_string()
                .contains("authority.dnsAuthority.resolvers[].resolverId"),
            "{err}"
        );
    }

    #[test]
    fn t13_rejects_duplicate_dns_resolver_ids() {
        let mut dns = full_dns_authority();
        dns.resolvers.push(crate::DnsResolver {
            resolver_id: "internal-doh".into(),
            endpoint: "https://1.0.0.1/dns-query".into(),
            protocol: crate::DnsResolverProtocol::Doh,
            trust_kid: None,
            dnssec: None,
        });
        let authority = AuthorityBundle {
            dns_authority: Some(dns),
            ..AuthorityBundle::default()
        };
        let err = validate_execution_cell_document(&doc_with_authority(authority)).unwrap_err();
        assert!(
            err.to_string().contains("duplicates value"),
            "expected duplicate resolverId rejection, got: {err}"
        );
    }

    #[test]
    fn t13_rejects_refresh_policy_min_ttl_greater_than_max_stale() {
        let mut dns = full_dns_authority();
        dns.refresh_policy = Some(crate::DnsRefreshPolicy {
            min_ttl_seconds: Some(600),
            max_stale_seconds: Some(60),
            strategy: Some(crate::DnsRefreshStrategy::TtlHonor),
        });
        let authority = AuthorityBundle {
            dns_authority: Some(dns),
            ..AuthorityBundle::default()
        };
        let err = validate_execution_cell_document(&doc_with_authority(authority)).unwrap_err();
        assert!(
            err.to_string().contains("minTtlSeconds")
                && err.to_string().contains("maxStaleSeconds"),
            "{err}"
        );
    }

    #[test]
    fn t13_rejects_cdn_provider_with_ip_literal_pattern() {
        let authority = AuthorityBundle {
            cdn_authority: Some(crate::CdnAuthority {
                providers: vec![crate::CdnProvider {
                    provider_id: "fastly".into(),
                    hostname_pattern: "192.168.1.1".into(),
                    accept_fronting: false,
                }],
            }),
            ..AuthorityBundle::default()
        };
        let err = validate_execution_cell_document(&doc_with_authority(authority)).unwrap_err();
        assert!(
            err.to_string()
                .contains("authority.cdnAuthority.providers[].hostnamePattern"),
            "{err}"
        );
    }

    #[test]
    fn t13_rejects_duplicate_cdn_provider_ids() {
        let authority = AuthorityBundle {
            cdn_authority: Some(crate::CdnAuthority {
                providers: vec![
                    crate::CdnProvider {
                        provider_id: "cloudfront".into(),
                        hostname_pattern: "a.cdn.example.com".into(),
                        accept_fronting: false,
                    },
                    crate::CdnProvider {
                        provider_id: "cloudfront".into(),
                        hostname_pattern: "b.cdn.example.com".into(),
                        accept_fronting: true,
                    },
                ],
            }),
            ..AuthorityBundle::default()
        };
        let err = validate_execution_cell_document(&doc_with_authority(authority)).unwrap_err();
        assert!(err.to_string().contains("duplicates value"), "{err}");
    }

    #[test]
    fn t13_dns_resolver_protocol_serialises_kebab_case() {
        let resolver = crate::DnsResolver {
            resolver_id: "p1".into(),
            endpoint: "1.1.1.1:53".into(),
            protocol: crate::DnsResolverProtocol::Do53Udp,
            trust_kid: None,
            dnssec: None,
        };
        let v = serde_json::to_value(&resolver).unwrap();
        assert_eq!(v.get("protocol"), Some(&serde_json::json!("do53-udp")));
    }

    #[test]
    fn t13_dns_authority_roundtrips_through_full_document() {
        let raw = r#"{
            "apiVersion": "cellos.io/v1",
            "kind": "ExecutionCell",
            "spec": {
                "id": "t13-roundtrip",
                "authority": {
                    "dnsAuthority": {
                        "resolvers": [{
                            "resolverId": "internal-doh",
                            "endpoint": "https://1.1.1.1/dns-query",
                            "protocol": "doh",
                            "trustKid": "resolver-kid-2026q2"
                        }],
                        "allowedQueryTypes": ["A", "AAAA", "HTTPS"],
                        "hostnameAllowlist": ["api.example.com", "*.cdn.example.com"],
                        "refreshPolicy": {
                            "minTtlSeconds": 30,
                            "maxStaleSeconds": 300,
                            "strategy": "ttl-honor"
                        },
                        "blockDirectWorkloadDns": true
                    },
                    "cdnAuthority": {
                        "providers": [{
                            "providerId": "cloudfront",
                            "hostnamePattern": "*.cdn.example.com",
                            "acceptFronting": false
                        }]
                    }
                },
                "lifetime": { "ttlSeconds": 60 }
            }
        }"#;
        let doc: ExecutionCellDocument =
            serde_json::from_str(raw).expect("parse T13 example document");
        validate_execution_cell_document(&doc).expect("validation must pass");

        let dns = doc.spec.authority.dns_authority.as_ref().unwrap();
        assert_eq!(dns.resolvers.len(), 1);
        assert!(matches!(
            dns.resolvers[0].protocol,
            crate::DnsResolverProtocol::Doh
        ));
        assert!(dns.block_direct_workload_dns);

        let cdn = doc.spec.authority.cdn_authority.as_ref().unwrap();
        assert_eq!(cdn.providers.len(), 1);
        assert!(!cdn.providers[0].accept_fronting);

        let serialised = serde_json::to_string(&doc).expect("re-serialise");
        let doc2: ExecutionCellDocument =
            serde_json::from_str(&serialised).expect("re-parse roundtrip");
        assert_eq!(
            doc2.spec.authority.dns_authority,
            doc.spec.authority.dns_authority
        );
        assert_eq!(
            doc2.spec.authority.cdn_authority,
            doc.spec.authority.cdn_authority
        );
    }

    #[test]
    fn t13_is_fqdn_or_wildcard_unit_cases() {
        assert!(super::is_fqdn_or_wildcard("api.example.com"));
        assert!(super::is_fqdn_or_wildcard("*.example.com"));
        assert!(super::is_fqdn_or_wildcard("a.b.c.d.example.com"));
        assert!(!super::is_fqdn_or_wildcard(""));
        assert!(!super::is_fqdn_or_wildcard("example")); // no dot
        assert!(!super::is_fqdn_or_wildcard("10.0.0.1")); // IPv4 literal
        assert!(!super::is_fqdn_or_wildcard("2001:db8::1")); // IPv6 literal
        assert!(!super::is_fqdn_or_wildcard("api.*.example.com")); // mid wildcard
        assert!(!super::is_fqdn_or_wildcard("-bad.example.com")); // label starts with '-'
        assert!(!super::is_fqdn_or_wildcard("under_score.example.com")); // underscore
    }

    // ── F4a telemetry admission tests ──────────────────────────────────────

    fn telemetry_doc(
        events: Vec<String>,
        agent_version: &str,
        egress: Vec<crate::EgressRule>,
    ) -> ExecutionCellDocument {
        let authority = AuthorityBundle {
            egress_rules: if egress.is_empty() {
                None
            } else {
                Some(egress)
            },
            ..AuthorityBundle::default()
        };
        ExecutionCellDocument {
            api_version: "cellos.io/v1".into(),
            kind: "ExecutionCell".into(),
            spec: ExecutionCellSpec {
                id: "telemetry-test-cell".into(),
                correlation: None,
                ingress: None,
                environment: None,
                placement: None,
                policy: None,
                identity: None,
                run: None,
                authority,
                lifetime: Lifetime { ttl_seconds: 60 },
                export: None,
                telemetry: Some(crate::TelemetrySpec {
                    channel: crate::TelemetryChannel::VsockCbor,
                    events,
                    rate_limits: None,
                    host_vs_guest_fields: None,
                    agent_version: agent_version.into(),
                }),
            },
        }
    }

    #[test]
    fn telemetry_rejects_empty_events() {
        let doc = telemetry_doc(vec![], "1.0.0", vec![]);
        let err = validate_execution_cell_document(&doc).unwrap_err();
        assert!(
            err.to_string().contains("spec.telemetry.events"),
            "got: {err}"
        );
    }

    #[test]
    fn telemetry_rejects_bad_semver() {
        let doc = telemetry_doc(vec!["process.spawn".into()], "v1", vec![]);
        let err = validate_execution_cell_document(&doc).unwrap_err();
        assert!(err.to_string().contains("agentVersion"), "got: {err}");
    }

    #[test]
    fn telemetry_rejects_net_event_without_egress() {
        let doc = telemetry_doc(vec!["net.connect.attempt".into()], "1.0.0", vec![]);
        let err = validate_execution_cell_document(&doc).unwrap_err();
        let msg = err.to_string();
        assert!(msg.contains("telemetry_without_egress"), "got: {msg}");
    }

    #[test]
    fn telemetry_accepts_net_event_when_egress_declared() {
        let doc = telemetry_doc(
            vec!["net.connect.attempt".into()],
            "1.0.0",
            vec![crate::EgressRule {
                host: "api.example.com".into(),
                port: 443,
                protocol: Some("https".into()),
                dns_egress_justification: None,
            }],
        );
        validate_execution_cell_document(&doc).expect("valid telemetry+egress spec");
    }

    #[test]
    fn telemetry_accepts_non_net_event_without_egress() {
        let doc = telemetry_doc(vec!["process.spawn".into()], "1.0.0", vec![]);
        validate_execution_cell_document(&doc).expect("non-net.* event must not require egress");
    }

    #[test]
    fn telemetry_accepts_prerelease_semver() {
        let doc = telemetry_doc(vec!["process.spawn".into()], "1.2.3-rc.1", vec![]);
        validate_execution_cell_document(&doc).expect("prerelease semver accepted");
    }
}

#[cfg(test)]
mod sec25_envelope_tests {
    //! SEC-25 signed trust-keyset envelope verifier tests.
    //!
    //! All synthetic: deterministic Ed25519 keys, no network, no real keyset
    //! material. The inner payload is a minimal byte string — these tests
    //! exercise the envelope/digest/signature path, not the inner
    //! trust-keyset-v1 schema.
    use super::{
        sha256_hex_prefixed, verify_signed_trust_keyset_chain, verify_signed_trust_keyset_envelope,
    };
    use crate::types::{SignedTrustKeysetEnvelope, TrustKeysetSignature};
    use base64::engine::general_purpose::URL_SAFE_NO_PAD;
    use base64::Engine as _;
    use ed25519_dalek::{Signer as _, SigningKey, VerifyingKey};
    use std::collections::HashMap;
    use std::time::{Duration, SystemTime, UNIX_EPOCH};

    fn signing_key(seed: u8) -> SigningKey {
        SigningKey::from_bytes(&[seed; 32])
    }

    /// Build a well-formed envelope with one valid signature. Returns the
    /// envelope and the signer's verifying key map (a single kid).
    fn make_envelope(
        payload_bytes: &[u8],
        signer: &SigningKey,
        signer_kid: &str,
        not_before: Option<String>,
        not_after: Option<String>,
    ) -> (SignedTrustKeysetEnvelope, HashMap<String, VerifyingKey>) {
        let signature = signer.sign(payload_bytes);
        let envelope = SignedTrustKeysetEnvelope {
            schema_version: "1.0.0".into(),
            payload_type: "application/vnd.cellos.trust-keyset-v1+json".into(),
            payload: URL_SAFE_NO_PAD.encode(payload_bytes),
            signatures: vec![TrustKeysetSignature {
                signer_kid: signer_kid.into(),
                algorithm: "ed25519".into(),
                signature: URL_SAFE_NO_PAD.encode(signature.to_bytes()),
                not_before,
                not_after,
            }],
            payload_digest: sha256_hex_prefixed(payload_bytes),
            produced_at: "2026-05-01T00:00:00Z".into(),
            replaces_envelope_digest: None,
            required_signer_count: None,
        };
        let mut keys = HashMap::new();
        keys.insert(signer_kid.to_string(), signer.verifying_key());
        (envelope, keys)
    }

    #[test]
    fn verifies_well_formed_envelope_with_synthetic_key() {
        let signer = signing_key(7);
        let payload = br#"{"schemaVersion":"1.0.0","keysetId":"ks-7","keys":[]}"#;
        let (env, keys) = make_envelope(payload, &signer, "kid-active-7", None, None);
        let returned = verify_signed_trust_keyset_envelope(&env, &keys, SystemTime::now())
            .expect("envelope should verify");
        assert_eq!(returned, payload, "verifier must return raw payload bytes");
    }

    #[test]
    fn rejects_payload_digest_mismatch() {
        let signer = signing_key(11);
        let payload = b"hello-payload";
        let (mut env, keys) = make_envelope(payload, &signer, "kid-active-11", None, None);
        // Flip a single hex char in the digest — keep prefix valid.
        let last_idx = env.payload_digest.len() - 1;
        let last_char = env.payload_digest.chars().last().unwrap();
        let new_char = if last_char == '0' { '1' } else { '0' };
        env.payload_digest
            .replace_range(last_idx.., &new_char.to_string());

        let err = verify_signed_trust_keyset_envelope(&env, &keys, SystemTime::now())
            .expect_err("digest mismatch must fail");
        assert!(
            format!("{err}").contains("payload digest mismatch"),
            "expected digest-mismatch error, got: {err}"
        );
    }

    #[test]
    fn rejects_unknown_signer_kid() {
        let signer = signing_key(13);
        let payload = b"unknown-kid-payload";
        let (env, _keys) = make_envelope(payload, &signer, "kid-active-13", None, None);
        let empty_keys: HashMap<String, VerifyingKey> = HashMap::new();
        let err = verify_signed_trust_keyset_envelope(&env, &empty_keys, SystemTime::now())
            .expect_err("unknown kid must fail");
        assert!(
            format!("{err}").contains("no signature verified"),
            "expected no-signature-verified error, got: {err}"
        );
    }

    #[test]
    fn rejects_signature_outside_not_after_window() {
        let signer = signing_key(17);
        let payload = b"window-payload";
        // notAfter set well in the past → window check fails.
        let (env, keys) = make_envelope(
            payload,
            &signer,
            "kid-active-17",
            None,
            Some("2000-01-01T00:00:00Z".into()),
        );
        let err = verify_signed_trust_keyset_envelope(&env, &keys, SystemTime::now())
            .expect_err("expired window must fail");
        assert!(
            format!("{err}").contains("no signature verified"),
            "expected no-signature-verified error, got: {err}"
        );
    }

    #[test]
    fn rejects_tampered_payload() {
        let signer = signing_key(19);
        let original = b"original-trust-keyset-payload-bytes";
        let (mut env, keys) = make_envelope(original, &signer, "kid-active-19", None, None);

        // Tamper: flip one byte in the decoded payload, re-encode, and
        // recompute the digest so we exercise the signature check (not the
        // digest check) failing.
        let mut tampered = original.to_vec();
        tampered[0] ^= 0x01;
        env.payload = URL_SAFE_NO_PAD.encode(&tampered);
        env.payload_digest = sha256_hex_prefixed(&tampered);

        let err = verify_signed_trust_keyset_envelope(&env, &keys, SystemTime::now())
            .expect_err("tampered payload must fail signature check");
        assert!(
            format!("{err}").contains("no signature verified"),
            "expected signature failure, got: {err}"
        );
    }

    #[test]
    fn accepts_when_at_least_one_signature_verifies() {
        let signer_good = signing_key(23);
        let signer_other = signing_key(29);
        let payload = b"multi-sig-payload";

        // First signature: bogus kid (verifier's keyring won't know it).
        let bad_sig = signer_other.sign(payload);
        // Second signature: good kid + good signature.
        let good_sig = signer_good.sign(payload);

        let env = SignedTrustKeysetEnvelope {
            schema_version: "1.0.0".into(),
            payload_type: "application/vnd.cellos.trust-keyset-v1+json".into(),
            payload: URL_SAFE_NO_PAD.encode(payload),
            signatures: vec![
                TrustKeysetSignature {
                    signer_kid: "kid-unknown-29".into(),
                    algorithm: "ed25519".into(),
                    signature: URL_SAFE_NO_PAD.encode(bad_sig.to_bytes()),
                    not_before: None,
                    not_after: None,
                },
                TrustKeysetSignature {
                    signer_kid: "kid-active-23".into(),
                    algorithm: "ed25519".into(),
                    signature: URL_SAFE_NO_PAD.encode(good_sig.to_bytes()),
                    not_before: None,
                    not_after: None,
                },
            ],
            payload_digest: sha256_hex_prefixed(payload),
            produced_at: "2026-05-01T00:00:00Z".into(),
            replaces_envelope_digest: None,
            required_signer_count: None,
        };

        let mut keys = HashMap::new();
        // Only the good signer is in the verifier's keyring.
        keys.insert("kid-active-23".to_string(), signer_good.verifying_key());

        let returned = verify_signed_trust_keyset_envelope(
            &env,
            &keys,
            UNIX_EPOCH + Duration::from_secs(1_800_000_000),
        )
        .expect("at least one signature should verify");
        assert_eq!(returned, payload);
    }

    // ── Phase 3: multi-signer threshold + envelope chain ────────────────────

    /// Build an envelope signed by N distinct signers, with an explicit
    /// `required_signer_count`. Returns the envelope and a keyring containing
    /// every signer.
    fn make_multisig_envelope(
        payload_bytes: &[u8],
        signers: &[(&str, &SigningKey)],
        required_signer_count: Option<u32>,
    ) -> (SignedTrustKeysetEnvelope, HashMap<String, VerifyingKey>) {
        let signatures = signers
            .iter()
            .map(|(kid, signer)| {
                let sig = signer.sign(payload_bytes);
                TrustKeysetSignature {
                    signer_kid: (*kid).to_string(),
                    algorithm: "ed25519".into(),
                    signature: URL_SAFE_NO_PAD.encode(sig.to_bytes()),
                    not_before: None,
                    not_after: None,
                }
            })
            .collect();

        let envelope = SignedTrustKeysetEnvelope {
            schema_version: "1.0.0".into(),
            payload_type: "application/vnd.cellos.trust-keyset-v1+json".into(),
            payload: URL_SAFE_NO_PAD.encode(payload_bytes),
            signatures,
            payload_digest: sha256_hex_prefixed(payload_bytes),
            produced_at: "2026-05-01T00:00:00Z".into(),
            replaces_envelope_digest: None,
            required_signer_count,
        };
        let mut keys = HashMap::new();
        for (kid, signer) in signers {
            keys.insert((*kid).to_string(), signer.verifying_key());
        }
        (envelope, keys)
    }

    #[test]
    fn verifies_with_two_distinct_signers_when_threshold_2() {
        let signer_a = signing_key(41);
        let signer_b = signing_key(43);
        let payload = b"phase3-multisig-payload-2of2";

        let (env, keys) = make_multisig_envelope(
            payload,
            &[("kid-ops-a", &signer_a), ("kid-ops-b", &signer_b)],
            Some(2),
        );

        let returned = verify_signed_trust_keyset_envelope(&env, &keys, SystemTime::now())
            .expect("two distinct signers + threshold 2 should verify");
        assert_eq!(returned, payload);
    }

    #[test]
    fn rejects_when_threshold_2_but_only_one_verifies() {
        let signer_a = signing_key(47);
        let signer_b = signing_key(53);
        let payload = b"phase3-only-one-verifies";

        // Build a 2-signer envelope but supply ONLY one signer's public key
        // in the verifier's keyring. The other signature has no key to
        // verify against, so the threshold of 2 is unreachable.
        let (env, _full_keys) = make_multisig_envelope(
            payload,
            &[("kid-ops-a", &signer_a), ("kid-ops-b", &signer_b)],
            Some(2),
        );
        let mut sparse_keys: HashMap<String, VerifyingKey> = HashMap::new();
        sparse_keys.insert("kid-ops-a".into(), signer_a.verifying_key());

        let err = verify_signed_trust_keyset_envelope(&env, &sparse_keys, SystemTime::now())
            .expect_err("threshold 2 with one verifier present must fail");
        let msg = format!("{err}");
        assert!(
            msg.contains("only 1 distinct signers verified, need 2"),
            "expected threshold-shortfall error, got: {msg}"
        );
    }

    #[test]
    fn counts_distinct_kids_only_duplicate_kid_does_not_inflate() {
        // Two signature entries from the same kid count as 1 verifier — they
        // do not satisfy a threshold of 2 even when both individually verify.
        let signer = signing_key(59);
        let payload = b"phase3-duplicate-kid-payload";
        let sig_a = signer.sign(payload);
        let sig_b = signer.sign(payload);

        let env = SignedTrustKeysetEnvelope {
            schema_version: "1.0.0".into(),
            payload_type: "application/vnd.cellos.trust-keyset-v1+json".into(),
            payload: URL_SAFE_NO_PAD.encode(payload),
            signatures: vec![
                TrustKeysetSignature {
                    signer_kid: "kid-ops-only".into(),
                    algorithm: "ed25519".into(),
                    signature: URL_SAFE_NO_PAD.encode(sig_a.to_bytes()),
                    not_before: None,
                    not_after: None,
                },
                TrustKeysetSignature {
                    signer_kid: "kid-ops-only".into(),
                    algorithm: "ed25519".into(),
                    signature: URL_SAFE_NO_PAD.encode(sig_b.to_bytes()),
                    not_before: None,
                    not_after: None,
                },
            ],
            payload_digest: sha256_hex_prefixed(payload),
            produced_at: "2026-05-01T00:00:00Z".into(),
            replaces_envelope_digest: None,
            required_signer_count: Some(2),
        };
        let mut keys = HashMap::new();
        keys.insert("kid-ops-only".into(), signer.verifying_key());

        let err = verify_signed_trust_keyset_envelope(&env, &keys, SystemTime::now())
            .expect_err("duplicate kid must not satisfy threshold 2");
        let msg = format!("{err}");
        assert!(
            msg.contains("only 1 distinct signers verified, need 2"),
            "expected duplicate-kid threshold failure, got: {msg}"
        );
    }

    #[test]
    fn default_threshold_1_preserves_phase1_behavior() {
        // An envelope with required_signer_count = None (and one with Some(1))
        // both behave identically to Phase 1: at-least-one-signature-verifies.
        let signer = signing_key(61);
        let payload = b"phase1-default-threshold";

        for explicit in [None, Some(1)] {
            let (env, keys) =
                make_multisig_envelope(payload, &[("kid-default-61", &signer)], explicit);
            let returned = verify_signed_trust_keyset_envelope(&env, &keys, SystemTime::now())
                .expect("default / explicit-1 threshold should verify");
            assert_eq!(
                returned, payload,
                "raw payload bytes returned (explicit={explicit:?})"
            );
        }
    }

    #[test]
    fn rejects_when_required_zero_treated_as_one() {
        // The schema's `minimum: 1` blocks required_signer_count: 0, but a
        // hand-built envelope could still set Some(0). The Rust verifier
        // clamps to 1 — so an envelope with zero verifying signatures still
        // surfaces the legacy "no signature verified" error rather than
        // passing on a zero threshold.
        let signer = signing_key(67);
        let payload = b"phase3-zero-clamp-payload";
        let (env, _keys) = make_multisig_envelope(
            payload,
            &[("kid-active-67", &signer)],
            Some(0), // hand-built envelope tries to bypass threshold
        );
        // Empty keyring → zero signatures verify; clamped threshold = 1
        // surfaces the legacy single-signer error.
        let empty: HashMap<String, VerifyingKey> = HashMap::new();
        let err = verify_signed_trust_keyset_envelope(&env, &empty, SystemTime::now())
            .expect_err("zero threshold must clamp to 1");
        let msg = format!("{err}");
        assert!(
            msg.contains("no signature verified"),
            "expected legacy single-signer error after clamping to 1, got: {msg}"
        );
    }

    /// Build a chain envelope with an explicit `replacesEnvelopeDigest` value.
    fn make_chain_envelope(
        payload_bytes: &[u8],
        signer: &SigningKey,
        signer_kid: &str,
        replaces: Option<String>,
    ) -> SignedTrustKeysetEnvelope {
        let signature = signer.sign(payload_bytes);
        SignedTrustKeysetEnvelope {
            schema_version: "1.0.0".into(),
            payload_type: "application/vnd.cellos.trust-keyset-v1+json".into(),
            payload: URL_SAFE_NO_PAD.encode(payload_bytes),
            signatures: vec![TrustKeysetSignature {
                signer_kid: signer_kid.into(),
                algorithm: "ed25519".into(),
                signature: URL_SAFE_NO_PAD.encode(signature.to_bytes()),
                not_before: None,
                not_after: None,
            }],
            payload_digest: sha256_hex_prefixed(payload_bytes),
            produced_at: "2026-05-01T00:00:00Z".into(),
            replaces_envelope_digest: replaces,
            required_signer_count: None,
        }
    }

    #[test]
    fn chain_with_correct_replaces_envelope_digest_succeeds() {
        let signer = signing_key(71);
        let payload_old = b"phase3-chain-old-payload";
        let payload_new = b"phase3-chain-new-payload";

        let prev = make_chain_envelope(payload_old, &signer, "kid-active-71", None);
        let next = make_chain_envelope(
            payload_new,
            &signer,
            "kid-active-71",
            Some(sha256_hex_prefixed(payload_old)),
        );

        let mut keys = HashMap::new();
        keys.insert("kid-active-71".into(), signer.verifying_key());

        let head_payload =
            verify_signed_trust_keyset_chain(&[prev, next], &keys, SystemTime::now())
                .expect("correctly-chained envelopes should verify");
        assert_eq!(head_payload, payload_new);
    }

    #[test]
    fn chain_with_mismatched_replaces_envelope_digest_rejected() {
        let signer = signing_key(73);
        let payload_old = b"phase3-chain-mismatch-old";
        let payload_new = b"phase3-chain-mismatch-new";

        let prev = make_chain_envelope(payload_old, &signer, "kid-active-73", None);
        // Tamper: point at a DIFFERENT envelope's digest. This is the replay
        // / fork attempt the chain verifier defends against.
        let bogus_digest = sha256_hex_prefixed(b"this-is-not-the-prior-payload");
        let next = make_chain_envelope(payload_new, &signer, "kid-active-73", Some(bogus_digest));

        let mut keys = HashMap::new();
        keys.insert("kid-active-73".into(), signer.verifying_key());

        let err = verify_signed_trust_keyset_chain(&[prev, next], &keys, SystemTime::now())
            .expect_err("mismatched replacesEnvelopeDigest must fail");
        let msg = format!("{err}");
        assert!(
            msg.contains("replacesEnvelopeDigest mismatch"),
            "expected chain link mismatch error, got: {msg}"
        );
        assert!(
            msg.contains("index 1"),
            "expected index 1 (the bad link) in error, got: {msg}"
        );
    }

    #[test]
    fn chain_first_envelope_without_replaces_accepted_as_genesis() {
        // Genesis envelope (index 0) MAY omit replacesEnvelopeDigest. A
        // single-element chain is the trivial case — only the genesis is
        // present, so no link checks fire.
        let signer = signing_key(79);
        let payload = b"phase3-chain-genesis-only";

        let genesis = make_chain_envelope(payload, &signer, "kid-active-79", None);
        let mut keys = HashMap::new();
        keys.insert("kid-active-79".into(), signer.verifying_key());

        let head_payload = verify_signed_trust_keyset_chain(&[genesis], &keys, SystemTime::now())
            .expect("genesis-only chain should verify");
        assert_eq!(head_payload, payload);
    }

    #[test]
    fn chain_returns_head_payload_bytes() {
        // A 3-envelope chain returns ONLY the HEAD's raw payload bytes — not
        // any intermediate's payload. This is the core contract.
        let signer = signing_key(83);
        let p0 = b"phase3-head-bytes-genesis";
        let p1 = b"phase3-head-bytes-middle";
        let p2 = b"phase3-head-bytes-HEAD-distinct";

        let env0 = make_chain_envelope(p0, &signer, "kid-active-83", None);
        let env1 = make_chain_envelope(p1, &signer, "kid-active-83", Some(sha256_hex_prefixed(p0)));
        let env2 = make_chain_envelope(p2, &signer, "kid-active-83", Some(sha256_hex_prefixed(p1)));

        let mut keys = HashMap::new();
        keys.insert("kid-active-83".into(), signer.verifying_key());

        let head_payload =
            verify_signed_trust_keyset_chain(&[env0, env1, env2], &keys, SystemTime::now())
                .expect("3-envelope chain should verify");
        assert_eq!(
            head_payload, p2,
            "verifier must return HEAD payload, not earlier links"
        );
    }

    #[test]
    fn chain_empty_rejected() {
        let keys: HashMap<String, VerifyingKey> = HashMap::new();
        let err = verify_signed_trust_keyset_chain(&[], &keys, SystemTime::now())
            .expect_err("empty chain must be rejected");
        let msg = format!("{err}");
        assert!(
            msg.contains("empty chain"),
            "expected empty-chain error, got: {msg}"
        );
    }

    #[test]
    fn chain_propagates_threshold_failure_per_envelope() {
        // A chain of two envelopes where the SECOND envelope sets
        // requiredSignerCount=2 but only carries one valid signature should
        // fail at the per-envelope threshold step — the chain verifier MUST
        // surface that as an indexed envelope failure (not as a digest-link
        // failure).
        let signer_a = signing_key(89);
        let signer_b = signing_key(97);
        let p0 = b"phase3-threshold-prop-genesis";
        let p1 = b"phase3-threshold-prop-head";

        let env0 = make_chain_envelope(p0, &signer_a, "kid-ops-a", None);
        // env1 declares required=2 but only signer_a signs — only one
        // distinct signer can verify.
        let mut env1 =
            make_chain_envelope(p1, &signer_a, "kid-ops-a", Some(sha256_hex_prefixed(p0)));
        env1.required_signer_count = Some(2);

        let mut keys = HashMap::new();
        keys.insert("kid-ops-a".into(), signer_a.verifying_key());
        keys.insert("kid-ops-b".into(), signer_b.verifying_key());

        let err = verify_signed_trust_keyset_chain(&[env0, env1], &keys, SystemTime::now())
            .expect_err("env1 threshold-2 with one signer must fail");
        let msg = format!("{err}");
        assert!(
            msg.contains("envelope at index 1 failed verification"),
            "expected indexed envelope failure, got: {msg}"
        );
        assert!(
            msg.contains("only 1 distinct signers verified, need 2"),
            "expected per-envelope threshold message in chain error, got: {msg}"
        );
    }
}