cellos-supervisor 0.6.0-pre

//! [`EventSink`] wrapper that signs each emitted CloudEvent (I5).
//!
//! This wraps any `Arc<dyn EventSink>` and, before calling `inner.emit()`,
//! signs the event into a [`cellos_core::SignedEventEnvelopeV1`] (HMAC-SHA256
//! or Ed25519 — operator's choice) then re-wraps it inside a transport
//! CloudEvent of `type` [`SIGNED_ENVELOPE_TRANSPORT_TYPE`] whose `data` field
//! is the JSON-serialized envelope. JetStream / JSONL pipelines downstream
//! see a normal `CloudEventV1` and need no trait change; the projector and
//! other typed consumers detect the wrapper type and unwrap it before
//! projecting the inner event.
//!
//! # Activation
//!
//! Three env vars, parsed in the composition root:
//!
//! - `CELLOS_EVENT_SIGNING={off|ed25519|hmac}` — algorithm selector. Default
//!   `off`. Any other value (including unparseable text) is logged at WARN
//!   and falls back to `off`.
//! - `CELLOS_EVENT_SIGNING_KID=<string>` — producer-asserted signer kid that
//!   downstream verifiers must know about. Required when the algo toggle is
//!   not `off`; missing/empty kid → signing disabled.
//! - `CELLOS_EVENT_SIGNING_KEY_BASE64=<base64url>` — the secret. For
//!   `ed25519` this is the raw 32-byte Ed25519 signing-key seed. For `hmac`
//!   this is the shared symmetric key (any length; HMAC-SHA256 internally
//!   uses RFC 2104's keyed-block padding for keys that are not 64 bytes).
//!   Padded base64url is tolerated. Decode failures → signing disabled.
//!
//! # Doctrine
//!
//! - **D1 — opt-in.** The default is `off`. Producers that don't sign emit
//!   raw `CloudEventV1` envelopes exactly as before; consumers that don't
//!   verify see no change.
//! - **Sign-after-redact.** The composition root wraps:
//!   `SigningEventSink::from_env(RedactingEventSink::from_env(inner))`
//!   so the signed payload reflects the post-redaction text. A consumer
//!   that decodes a `SignedEventEnvelopeV1` and re-runs redaction would
//!   diverge from what the producer signed.
//! - **No transparent fallback to unsigned on emit failure.** If signing is
//!   configured and fails for a single event, that emit returns an error;
//!   the supervisor decides what to do with it (today: surfaces it to the
//!   `emit()` call-site).

use std::sync::Arc;

use async_trait::async_trait;
use base64::engine::general_purpose::URL_SAFE_NO_PAD;
use base64::Engine as _;
use cellos_core::ports::EventSink;
use cellos_core::{
    sign_event_hmac_sha256, sign_event_with, CellosError, CloudEventV1, SignedEventEnvelopeV1,
    Signer, SoftwareSigner,
};
use cellos_sink_redact::{redact_fields_from_env, RedactingEventSink};
use std::collections::HashSet;

/// CloudEvent `type` of the transport wrapper produced by [`SigningEventSink`].
///
/// Downstream consumers (the projector, taudit, etc.) MUST treat any event
/// whose `ty` matches this constant as a wrapper whose `data` JSON
/// deserializes into [`SignedEventEnvelopeV1`].
pub const SIGNED_ENVELOPE_TRANSPORT_TYPE: &str = "dev.cellos.events.signed_envelope.v1";

/// U1-04: a single structured "this signing-config env var was ignored / fell
/// back to the default" record, returned by
/// [`SigningEventSink::from_env_with_warnings`].
///
/// The composition root translates each entry into a
/// `StartupConfigWarnings::record` call so `CELLOS_STRICT_CONFIG=1` escalates
/// signing misconfig to a fatal startup error instead of silently degrading
/// to passthrough. Reviewer wave 2 (bebc77b) flagged that the legacy
/// `from_env` shape only emitted `tracing::warn!` and therefore slipped
/// through strict mode.
#[derive(Debug, Clone)]
pub struct EventSigningConfigWarning {
    /// Env var name that was misconfigured (e.g. `CELLOS_EVENT_SIGNING_KEY_BASE64`).
    pub var: &'static str,
    /// Operator-supplied value (best-effort; empty when only a companion var
    /// was missing). Never contains decoded key material — only the toggle /
    /// kid / opaque "<base64 ...>" placeholders are stored verbatim.
    pub value: String,
    /// Operator-facing explanation of why the value was rejected.
    pub reason: String,
}

/// `source` attribute of the wrapper CloudEvent. Static — the original event's
/// `source` is preserved on the inner event inside the envelope.
const WRAPPER_SOURCE: &str = "/cellos-supervisor/event-signing";

/// Operator's algorithm choice from `CELLOS_EVENT_SIGNING`.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum Algorithm {
    Ed25519,
    HmacSha256,
}

/// Resolved signing configuration: algo + kid + key bytes.
///
/// # Zeroize posture (D7)
///
/// `key_bytes` holds raw signing-key material. Wrapped in
/// [`zeroize::Zeroizing`] so the heap-allocated buffer is wiped on drop;
/// `kid` and `algorithm` are non-secret operator-supplied identifiers and
/// are `#[zeroize(skip)]`. `Clone` is intentionally retained for
/// composition-root plumbing — every clone preserves the same zeroize-on-drop
/// guarantee on its `key_bytes` field.
#[derive(Clone, zeroize::ZeroizeOnDrop)]
struct SigningConfig {
    #[zeroize(skip)]
    algorithm: Algorithm,
    #[zeroize(skip)]
    kid: String,
    /// For Ed25519 this is the 32-byte signing-key seed.
    /// For HMAC-SHA256 this is the raw shared key (any length).
    key_bytes: zeroize::Zeroizing<Vec<u8>>,
}

/// Parse `CELLOS_EVENT_SIGNING` + `CELLOS_EVENT_SIGNING_KID` +
/// `CELLOS_EVENT_SIGNING_KEY_BASE64` into a resolved [`SigningConfig`].
///
/// Returns `(None, warnings)` when signing is `off`/unset/misconfigured (the
/// signing-disabled paths, each carrying the legacy `tracing::warn!` plus a
/// structured [`EventSigningConfigWarning`]); `(Some(cfg), warnings)` when the
/// full triple validates. Shared by [`SigningEventSink::from_env_with_warnings`]
/// (the composed `Sign(Redact(transport))` pipeline) and
/// [`StandaloneEventSigner::from_env`] (the pre-`build_supervisor` admission
/// receipt signer, ADR-0025 §2) so both read the identical config with
/// identical fail-to-passthrough semantics.
fn parse_signing_config_from_env() -> (Option<SigningConfig>, Vec<EventSigningConfigWarning>) {
    let mut warnings: Vec<EventSigningConfigWarning> = Vec::new();

    let toggle = match std::env::var("CELLOS_EVENT_SIGNING") {
        Ok(v) => v,
        Err(_) => return (None, warnings),
    };
    let toggle_norm = toggle.trim().to_ascii_lowercase();
    if toggle_norm.is_empty() || toggle_norm == "off" {
        return (None, warnings);
    }

    let algorithm = match toggle_norm.as_str() {
        "ed25519" => Algorithm::Ed25519,
        "hmac" | "hmac-sha256" => Algorithm::HmacSha256,
        other => {
            tracing::warn!(
                target: "cellos.supervisor.event_signing",
                toggle = %other,
                "CELLOS_EVENT_SIGNING: unknown algorithm — signing disabled (expected off|ed25519|hmac)"
            );
            warnings.push(EventSigningConfigWarning {
                var: "CELLOS_EVENT_SIGNING",
                value: other.to_string(),
                reason: "unknown algorithm (expected off|ed25519|hmac); signing disabled".into(),
            });
            return (None, warnings);
        }
    };

    let kid = match std::env::var("CELLOS_EVENT_SIGNING_KID") {
        Ok(k) => k.trim().to_string(),
        Err(_) => {
            tracing::warn!(
                target: "cellos.supervisor.event_signing",
                "CELLOS_EVENT_SIGNING is set but CELLOS_EVENT_SIGNING_KID is missing — signing disabled"
            );
            warnings.push(EventSigningConfigWarning {
                var: "CELLOS_EVENT_SIGNING_KID",
                value: String::new(),
                reason: format!(
                    "CELLOS_EVENT_SIGNING={toggle_norm} requires a kid but CELLOS_EVENT_SIGNING_KID is unset; signing disabled"
                ),
            });
            return (None, warnings);
        }
    };
    if kid.is_empty() {
        tracing::warn!(
            target: "cellos.supervisor.event_signing",
            "CELLOS_EVENT_SIGNING_KID is empty — signing disabled"
        );
        warnings.push(EventSigningConfigWarning {
            var: "CELLOS_EVENT_SIGNING_KID",
            value: String::new(),
            reason: "kid is empty; signing disabled".into(),
        });
        return (None, warnings);
    }

    let key_b64 = match std::env::var("CELLOS_EVENT_SIGNING_KEY_BASE64") {
        Ok(k) => k,
        Err(_) => {
            tracing::warn!(
                target: "cellos.supervisor.event_signing",
                "CELLOS_EVENT_SIGNING is set but CELLOS_EVENT_SIGNING_KEY_BASE64 is missing — signing disabled"
            );
            warnings.push(EventSigningConfigWarning {
                var: "CELLOS_EVENT_SIGNING_KEY_BASE64",
                value: String::new(),
                reason: format!(
                    "CELLOS_EVENT_SIGNING={toggle_norm} requires a key but CELLOS_EVENT_SIGNING_KEY_BASE64 is unset; signing disabled"
                ),
            });
            return (None, warnings);
        }
    };
    let trimmed = key_b64.trim().trim_end_matches('=');
    // D7: wrap the decoded raw key in `Zeroizing` immediately so that the
    // intermediate buffer is wiped on drop, even on the early-return error
    // paths below.
    let key_bytes: zeroize::Zeroizing<Vec<u8>> = match URL_SAFE_NO_PAD.decode(trimmed) {
        Ok(b) => zeroize::Zeroizing::new(b),
        Err(e) => {
            tracing::warn!(
                target: "cellos.supervisor.event_signing",
                error = %e,
                "CELLOS_EVENT_SIGNING_KEY_BASE64: invalid base64url — signing disabled"
            );
            // Don't echo the operator's raw key bytes back into a structured
            // warning record — record an opaque marker instead, with the
            // base64 decoder's diagnostic in `reason` for triage.
            warnings.push(EventSigningConfigWarning {
                var: "CELLOS_EVENT_SIGNING_KEY_BASE64",
                value: "<base64 decode failed>".into(),
                reason: format!("invalid base64url: {e}; signing disabled"),
            });
            return (None, warnings);
        }
    };

    // For Ed25519 the seed must decode to exactly 32 bytes; HMAC accepts
    // any non-empty length (RFC 2104 § 2 explicitly allows arbitrary
    // key lengths via the keyed-block transformation).
    match algorithm {
        Algorithm::Ed25519 if key_bytes.len() != 32 => {
            tracing::warn!(
                target: "cellos.supervisor.event_signing",
                got_bytes = key_bytes.len(),
                "CELLOS_EVENT_SIGNING=ed25519 requires a 32-byte key — signing disabled"
            );
            warnings.push(EventSigningConfigWarning {
                var: "CELLOS_EVENT_SIGNING_KEY_BASE64",
                value: format!("<{} bytes>", key_bytes.len()),
                reason: format!(
                    "CELLOS_EVENT_SIGNING=ed25519 requires a 32-byte key, got {}; signing disabled",
                    key_bytes.len()
                ),
            });
            return (None, warnings);
        }
        Algorithm::HmacSha256 if key_bytes.is_empty() => {
            tracing::warn!(
                target: "cellos.supervisor.event_signing",
                "CELLOS_EVENT_SIGNING=hmac requires a non-empty key — signing disabled"
            );
            warnings.push(EventSigningConfigWarning {
                var: "CELLOS_EVENT_SIGNING_KEY_BASE64",
                value: "<empty>".into(),
                reason: "CELLOS_EVENT_SIGNING=hmac requires a non-empty key; signing disabled"
                    .into(),
            });
            return (None, warnings);
        }
        _ => {}
    }

    let algo_label = match algorithm {
        Algorithm::Ed25519 => "ed25519",
        Algorithm::HmacSha256 => "hmac-sha256",
    };
    tracing::info!(
        target: "cellos.supervisor.event_signing",
        algorithm = %algo_label,
        kid = %kid,
        "per-event signing enabled"
    );

    (
        Some(SigningConfig {
            algorithm,
            kid,
            key_bytes,
        }),
        warnings,
    )
}

/// S21 (ADR-0029): `true` when the `CELLOS_EVENT_SIGNING` triple resolves to a
/// usable signing config (algorithm set, kid present, key decodes to the right
/// length). The composition root's `airgapped` deployment profile consults this
/// at startup to **fail closed** when signing keys are absent — the airgapped
/// spool's sealed-segment head (S15) must be offline-verifiable, so the
/// ADR-0025 `signed:false` advisory degradation is not acceptable there.
///
/// Reads the exact same env surface as [`SigningEventSink::from_env`] and
/// [`StandaloneEventSigner::from_env`] (no separate operator knob), so a config
/// that signs for the composed sink also satisfies the airgapped gate.
pub fn signing_enabled_from_env() -> bool {
    parse_signing_config_from_env().0.is_some()
}

/// C03: build an Ed25519 [`SoftwareSigner`] from the `CELLOS_EVENT_SIGNING`
/// triple, for callers that hold a raw [`Signer`] rather than an event sink (the
/// startup reconcile receipt). Returns `None` when signing is unconfigured or
/// resolves to HMAC — the reconcile receipt requires an offline-verifiable
/// Ed25519 signature, so HMAC (a symmetric MAC) is deliberately not accepted.
pub(crate) fn ed25519_signer_from_env() -> Option<SoftwareSigner> {
    let (cfg, _warnings) = parse_signing_config_from_env();
    let cfg = cfg?;
    match cfg.algorithm {
        Algorithm::Ed25519 => {
            let seed: [u8; 32] = cfg.key_bytes.as_slice().try_into().ok()?;
            SoftwareSigner::from_seed(&cfg.kid, seed).ok()
        }
        Algorithm::HmacSha256 => None,
    }
}

/// [`EventSink`] wrapper that signs each emitted CloudEvent and forwards a
/// transport-wrapper CloudEvent of type [`SIGNED_ENVELOPE_TRANSPORT_TYPE`].
pub struct SigningEventSink {
    inner: Arc<dyn EventSink>,
    cfg: SigningConfig,
}

impl SigningEventSink {
    /// Construct a signing sink directly. Prefer [`SigningEventSink::from_env`]
    /// in the composition root.
    fn new(inner: Arc<dyn EventSink>, cfg: SigningConfig) -> Self {
        Self { inner, cfg }
    }

    /// Parse `CELLOS_EVENT_SIGNING` + `CELLOS_EVENT_SIGNING_KID` +
    /// `CELLOS_EVENT_SIGNING_KEY_BASE64` and return a wrapped sink if all
    /// three are valid, otherwise return `inner` unwrapped.
    ///
    /// Backwards-compatible thin wrapper over
    /// [`SigningEventSink::from_env_with_warnings`] for callers that don't
    /// need to surface the structured warnings (existing in-module unit
    /// tests, the `sink_composition_chain` integration test). Composition
    /// root MUST use `from_env_with_warnings` so misconfig is observable
    /// under `CELLOS_STRICT_CONFIG=1` (U1-04).
    pub fn from_env(inner: Arc<dyn EventSink>) -> Arc<dyn EventSink> {
        let (sink, _warnings) = Self::from_env_with_warnings(inner);
        sink
    }

    /// Same parsing as [`SigningEventSink::from_env`] but additionally returns
    /// a [`Vec<EventSigningConfigWarning>`] capturing every WARN-and-fall-back
    /// decision. The composition root threads these into the supervisor's
    /// `StartupConfigWarnings` accumulator so a misconfigured `event_signing`
    /// triggers `CELLOS_STRICT_CONFIG=1` startup failure (U1-04 / reviewer
    /// wave 2 finding bebc77b) rather than silently degrading to passthrough.
    ///
    /// Each disable path also emits the legacy `tracing::warn!` so log-based
    /// alerting keeps working unchanged.
    pub fn from_env_with_warnings(
        inner: Arc<dyn EventSink>,
    ) -> (Arc<dyn EventSink>, Vec<EventSigningConfigWarning>) {
        let (cfg, warnings) = parse_signing_config_from_env();
        match cfg {
            Some(cfg) => (Arc::new(Self::new(inner, cfg)), warnings),
            None => (inner, warnings),
        }
    }

    /// Sign `event` and produce the transport-wrapper CloudEvent.
    fn wrap(&self, event: &CloudEventV1) -> Result<CloudEventV1, CellosError> {
        let envelope = sign_event_with_config(event, &self.cfg)?;
        let data = serde_json::to_value(&envelope).map_err(|e| {
            CellosError::InvalidSpec(format!("event signing: serialize envelope: {e}"))
        })?;
        Ok(wrap_signed_envelope(event, data))
    }
}

/// Sign `event` with a resolved [`SigningConfig`], producing the inner
/// [`SignedEventEnvelopeV1`]. Shared by [`SigningEventSink::wrap`] and
/// [`StandaloneEventSigner`] so the two signing paths use byte-identical
/// crypto (same `canonical_event_signing_payload`, same algorithm dispatch).
///
/// S33 (ADR-0031): the Ed25519 path routes through the `cellos_core`
/// [`Signer`] port — an [`Arc<dyn Signer>`] built from a [`SoftwareSigner`]
/// over the operator-supplied `CELLOS_EVENT_SIGNING_KEY_BASE64` seed — rather
/// than calling `ed25519_dalek` (or `sign_event_ed25519`) directly. This makes
/// the private key live behind the same custody abstraction the spool sealer
/// (S15) uses, so an HSM / PKCS#11 signer (S34) or a FIPS-validated provider
/// (ADR-0027) drops in transparently. The result is byte-identical to the
/// previous `sign_event_ed25519` call for the same seed + kid (both route
/// through the active `CryptoProvider` over the same canonical payload). The
/// **operator surface is unchanged** — `CELLOS_EVENT_SIGNING={off|ed25519|hmac}`
/// still selects the algorithm; HMAC-SHA256 (a symmetric MAC, not an
/// `ed25519`-shaped [`Signer`]) stays on the inline keyed-hash path.
fn sign_event_with_config(
    event: &CloudEventV1,
    cfg: &SigningConfig,
) -> Result<SignedEventEnvelopeV1, CellosError> {
    match cfg.algorithm {
        Algorithm::Ed25519 => {
            let seed: [u8; 32] = cfg.key_bytes.as_slice().try_into().map_err(|_| {
                CellosError::InvalidSpec(format!(
                    "event signing: ed25519 key must be 32 bytes, got {}",
                    cfg.key_bytes.len()
                ))
            })?;
            let signer: Arc<dyn Signer> = Arc::new(SoftwareSigner::from_seed(&cfg.kid, seed)?);
            sign_event_with(signer.as_ref(), event)
        }
        Algorithm::HmacSha256 => sign_event_hmac_sha256(event, &cfg.kid, &cfg.key_bytes),
    }
}

/// Build the transport-wrapper CloudEvent carrying a serialized
/// [`SignedEventEnvelopeV1`] in `data`. The wrapper `type` is
/// [`SIGNED_ENVELOPE_TRANSPORT_TYPE`] so downstream consumers detect and unwrap
/// it; the original event id is preserved for correlation.
pub(crate) fn wrap_signed_envelope(
    event: &CloudEventV1,
    envelope_data: serde_json::Value,
) -> CloudEventV1 {
    CloudEventV1 {
        specversion: "1.0".into(),
        // Preserve the original event id so downstream observers can
        // correlate the wrapper with logs that reference the inner id.
        id: event.id.clone(),
        source: WRAPPER_SOURCE.to_string(),
        ty: SIGNED_ENVELOPE_TRANSPORT_TYPE.to_string(),
        datacontenttype: Some("application/json".into()),
        data: Some(envelope_data),
        time: event.time.clone(),
        traceparent: event.traceparent.clone(),
        cex: None,
    }
}

#[async_trait]
impl EventSink for SigningEventSink {
    async fn emit(&self, event: &CloudEventV1) -> Result<(), CellosError> {
        let wrapped = self.wrap(event)?;
        self.inner.emit(&wrapped).await
    }

    async fn flush(&self) -> Result<(), CellosError> {
        self.inner.flush().await
    }
}

/// Top-level marker field injected into the **unsigned** decision record's
/// `data` so an auditor can tell a `signed: false` advisory record from a
/// `SignedEventEnvelopeV1` wrapper without parsing the producer's source
/// (ADR-0025 §Unsigned-mode-honesty).
const SIGNED_MARKER_FIELD: &str = "signed";

/// Standalone Ed25519/HMAC signer for admission-decision receipts emitted
/// **before** `build_supervisor` runs (ADR-0025 §2).
///
/// The ceiling gate (ADR-0024 §3) is hooked at `main.rs` before the
/// `Supervisor` — and therefore before its composed
/// `DlqSink(SigningEventSink(RedactingEventSink))` — exists. A receipt that
/// "reuses `Supervisor::emit`" at that hook is physically unimplementable. This
/// signer resolves that wire-binding contradiction: it is constructed in
/// `main.rs` from the **same** `CELLOS_EVENT_SIGNING` + `KID` + `KEY_BASE64`
/// config and writes to the **same** `RedactingEventSink` transport target the
/// composed sink uses, so allow/reject receipts fire on the live path.
///
/// Redaction runs **before** signing: the event is first passed through
/// `RedactingEventSink::from_env` (the same `CELLOS_REDACT_EVENT_FIELDS` field
/// set), and the post-redaction event is what gets signed — so a consumer who
/// re-runs redaction never diverges from what the producer signed (ADR-0018).
///
/// When signing is unconfigured the signer still emits, degrading to an
/// **unsigned** record carrying an explicit `signed: false` marker (never a
/// *missing* receipt, never silent); an unsigned record is advisory and is not
/// offline-verifiable.
pub struct StandaloneEventSigner {
    /// Redacting wrapper over the primary transport. Used to obtain the
    /// post-redaction event (redact-before-sign) and, in unsigned mode, as the
    /// direct emit target for the advisory `signed: false` record.
    redacting: Arc<dyn EventSink>,
    /// Bare transport target (no redaction wrapper). The signed envelope is
    /// emitted here directly because the envelope already carries the
    /// already-redacted event, and re-redacting the opaque wrapper `data` would
    /// be a no-op at best and corrupt the signed bytes at worst.
    transport: Arc<dyn EventSink>,
    /// `Some` when the signing triple validated; `None` => unsigned advisory mode.
    cfg: Option<SigningConfig>,
    /// Redaction field set resolved **once** at construction from
    /// `CELLOS_REDACT_EVENT_FIELDS`. `None` means redaction is unconfigured.
    /// Snapshotting here (rather than re-reading the env in `redact`) makes the
    /// redact-before-sign step deterministic for the lifetime of the signer,
    /// independent of any later mutation of the process environment.
    redact_fields: Option<HashSet<String>>,
}

impl StandaloneEventSigner {
    /// Construct the signer from the environment over `transport` (the primary
    /// event sink built by the composition root). The signing config is parsed
    /// with the identical semantics as the composed `SigningEventSink`; any
    /// misconfiguration is surfaced as `Vec<EventSigningConfigWarning>` so the
    /// composition root can escalate it under `CELLOS_STRICT_CONFIG=1`, exactly
    /// as the composed sink does.
    pub fn from_env(transport: Arc<dyn EventSink>) -> (Self, Vec<EventSigningConfigWarning>) {
        let (cfg, warnings) = parse_signing_config_from_env();
        let redact_fields = redact_fields_from_env();
        let redacting =
            RedactingEventSink::with_resolved_fields(transport.clone(), redact_fields.as_ref());
        (
            Self {
                redacting,
                transport,
                cfg,
                redact_fields,
            },
            warnings,
        )
    }

    /// `true` when event signing is configured and this signer will produce an
    /// offline-verifiable `SignedEventEnvelopeV1`; `false` when it degrades to
    /// the advisory `signed: false` record.
    pub fn is_signing_enabled(&self) -> bool {
        self.cfg.is_some()
    }

    /// Flush the underlying transport so a just-emitted receipt is guaranteed on
    /// the wire. The ceiling gate (ADR-0024/0025) calls this after emitting a
    /// reject receipt because it then aborts admission immediately; a buffering
    /// transport (NATS) would otherwise drop the unflushed publish on process
    /// exit. Both the signed and unsigned paths write through the same
    /// underlying `transport`, so flushing it covers both.
    pub async fn flush(&self) -> Result<(), CellosError> {
        self.transport.flush().await
    }

    /// Redact `event` (same field set as the composed pipeline) and return the
    /// post-redaction event. Routes through a `RedactingEventSink` built from
    /// the construction-time field snapshot over a one-shot capture sink, so the
    /// standalone path reuses the exact redaction logic rather than
    /// re-implementing it (and never re-reads `CELLOS_REDACT_EVENT_FIELDS`).
    async fn redact(&self, event: &CloudEventV1) -> Result<CloudEventV1, CellosError> {
        let capture = CaptureOnceSink::new();
        self.redacting_for(capture.clone()).emit(event).await?;
        capture.take().ok_or_else(|| {
            CellosError::InvalidSpec(
                "standalone signer: redaction sink dropped the event".to_string(),
            )
        })
    }

    /// Build a redacting wrapper over an arbitrary inner sink using the
    /// field set snapshotted at construction — the same configuration as
    /// [`Self::redacting`] — without re-reading the process environment.
    fn redacting_for(&self, inner: Arc<dyn EventSink>) -> Arc<dyn EventSink> {
        RedactingEventSink::with_resolved_fields(inner, self.redact_fields.as_ref())
    }

    /// Emit a signed (or, when unconfigured, advisory `signed: false`) admission
    /// decision receipt. Redaction always runs before signing.
    ///
    /// On the signed path the returned envelope is `Some` so the caller (the
    /// ceiling gate, ADR-0024) can assert offline verification in tests; on the
    /// advisory path it is `None` (the record is not offline-verifiable).
    pub async fn emit_decision(
        &self,
        event: &CloudEventV1,
    ) -> Result<Option<SignedEventEnvelopeV1>, CellosError> {
        let redacted = self.redact(event).await?;
        match &self.cfg {
            Some(cfg) => {
                let envelope = sign_event_with_config(&redacted, cfg)?;
                let data = serde_json::to_value(&envelope).map_err(|e| {
                    CellosError::InvalidSpec(format!("standalone signer: serialize envelope: {e}"))
                })?;
                let wrapper = wrap_signed_envelope(&redacted, data);
                // Emit the signed wrapper directly to the bare transport: the
                // wrapper's `data` already carries the redacted-and-signed
                // event, so re-redacting here would not improve safety and
                // must not mutate the signed bytes.
                self.transport.emit(&wrapper).await?;
                Ok(Some(envelope))
            }
            None => {
                // Unsigned advisory record: stamp `signed: false` on the
                // post-redaction event and emit it through the redacting
                // transport. Never a missing receipt, never silent.
                let mut advisory = redacted;
                let data = advisory
                    .data
                    .get_or_insert_with(|| serde_json::Value::Object(serde_json::Map::new()));
                if let serde_json::Value::Object(map) = data {
                    map.insert(
                        SIGNED_MARKER_FIELD.to_string(),
                        serde_json::Value::Bool(false),
                    );
                }
                self.redacting.emit(&advisory).await?;
                Ok(None)
            }
        }
    }
}

/// Single-shot capture sink: records the one event emitted to it so the
/// standalone signer can extract the post-redaction event from
/// `RedactingEventSink::emit`. Internal to the redact-before-sign path.
struct CaptureOnceSink(std::sync::Mutex<Option<CloudEventV1>>);

impl CaptureOnceSink {
    fn new() -> Arc<Self> {
        Arc::new(Self(std::sync::Mutex::new(None)))
    }
    fn take(&self) -> Option<CloudEventV1> {
        self.0.lock().expect("capture-once lock poisoned").take()
    }
}

#[async_trait]
impl EventSink for CaptureOnceSink {
    async fn emit(&self, event: &CloudEventV1) -> Result<(), CellosError> {
        *self.0.lock().expect("capture-once lock poisoned") = Some(event.clone());
        Ok(())
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use cellos_core::{verify_signed_event_envelope, CloudEventV1};
    use ed25519_dalek::SigningKey;
    use serde_json::json;
    use std::collections::HashMap;
    use std::sync::Mutex;

    /// Capture sink — records the last event it received.
    struct CaptureSink(Mutex<Option<CloudEventV1>>);

    impl CaptureSink {
        fn new() -> Arc<Self> {
            Arc::new(Self(Mutex::new(None)))
        }
        fn last(&self) -> Option<CloudEventV1> {
            self.0.lock().unwrap().clone()
        }
    }

    #[async_trait]
    impl EventSink for CaptureSink {
        async fn emit(&self, event: &CloudEventV1) -> Result<(), CellosError> {
            *self.0.lock().unwrap() = Some(event.clone());
            Ok(())
        }
    }

    fn sample_event() -> CloudEventV1 {
        CloudEventV1 {
            specversion: "1.0".into(),
            id: "ev-001".into(),
            source: "/cellos-supervisor".into(),
            ty: "dev.cellos.events.cell.lifecycle.v1.started".into(),
            datacontenttype: Some("application/json".into()),
            data: Some(json!({"cellId": "test-cell-1"})),
            time: Some("2026-05-06T12:00:00Z".into()),
            traceparent: None,
            cex: None,
        }
    }

    /// Guards env-var mutation so from_env tests don't race with each other.
    static FROM_ENV_MUTEX: std::sync::Mutex<()> = std::sync::Mutex::new(());

    fn clear_signing_env() {
        std::env::remove_var("CELLOS_EVENT_SIGNING");
        std::env::remove_var("CELLOS_EVENT_SIGNING_KID");
        std::env::remove_var("CELLOS_EVENT_SIGNING_KEY_BASE64");
    }

    #[tokio::test]
    async fn from_env_off_passes_through_unwrapped() {
        let capture = CaptureSink::new();
        let sink = {
            let _g = FROM_ENV_MUTEX.lock().unwrap();
            clear_signing_env();
            std::env::set_var("CELLOS_EVENT_SIGNING", "off");
            SigningEventSink::from_env(capture.clone() as Arc<dyn EventSink>)
        };
        let event = sample_event();
        sink.emit(&event).await.unwrap();
        {
            let _g = FROM_ENV_MUTEX.lock().unwrap();
            clear_signing_env();
        }
        let got = capture.last().unwrap();
        assert_eq!(
            got.ty, "dev.cellos.events.cell.lifecycle.v1.started",
            "off must pass through the original event unchanged"
        );
    }

    #[tokio::test]
    async fn from_env_unknown_toggle_disables_signing() {
        let capture = CaptureSink::new();
        let sink = {
            let _g = FROM_ENV_MUTEX.lock().unwrap();
            clear_signing_env();
            std::env::set_var("CELLOS_EVENT_SIGNING", "rsa-pss-sha512");
            std::env::set_var("CELLOS_EVENT_SIGNING_KID", "ops-event-2026-q2");
            std::env::set_var(
                "CELLOS_EVENT_SIGNING_KEY_BASE64",
                URL_SAFE_NO_PAD.encode([7u8; 32]),
            );
            SigningEventSink::from_env(capture.clone() as Arc<dyn EventSink>)
        };
        let event = sample_event();
        sink.emit(&event).await.unwrap();
        {
            let _g = FROM_ENV_MUTEX.lock().unwrap();
            clear_signing_env();
        }
        let got = capture.last().unwrap();
        assert_eq!(
            got.ty, "dev.cellos.events.cell.lifecycle.v1.started",
            "unknown algorithm must fall back to passthrough"
        );
    }

    #[tokio::test]
    async fn from_env_ed25519_round_trip_via_sink() {
        let capture = CaptureSink::new();
        let signer_seed = [13u8; 32];
        let signer = SigningKey::from_bytes(&signer_seed);
        let sink = {
            let _g = FROM_ENV_MUTEX.lock().unwrap();
            clear_signing_env();
            std::env::set_var("CELLOS_EVENT_SIGNING", "ed25519");
            std::env::set_var("CELLOS_EVENT_SIGNING_KID", "ops-event-2026-q2");
            std::env::set_var(
                "CELLOS_EVENT_SIGNING_KEY_BASE64",
                URL_SAFE_NO_PAD.encode(signer_seed),
            );
            SigningEventSink::from_env(capture.clone() as Arc<dyn EventSink>)
        };
        let event = sample_event();
        sink.emit(&event).await.unwrap();
        {
            let _g = FROM_ENV_MUTEX.lock().unwrap();
            clear_signing_env();
        }

        let got = capture.last().expect("wrapper emitted");
        assert_eq!(got.ty, SIGNED_ENVELOPE_TRANSPORT_TYPE);
        let envelope: SignedEventEnvelopeV1 =
            serde_json::from_value(got.data.expect("wrapper has data")).expect("parse envelope");
        assert_eq!(envelope.algorithm, "ed25519");
        assert_eq!(envelope.signer_kid, "ops-event-2026-q2");

        let mut keys = HashMap::new();
        keys.insert(
            "ops-event-2026-q2".to_string(),
            cellos_core::TrustAnchorPublicKey::from_bytes_unchecked(
                signer.verifying_key().to_bytes(),
            ),
        );
        let hmac_keys: HashMap<String, Vec<u8>> = HashMap::new();
        let inner = verify_signed_event_envelope(&envelope, &keys, &hmac_keys).expect("verify ok");
        assert_eq!(inner.id, event.id, "inner event id round-trips");
        assert_eq!(inner.ty, event.ty);
    }

    #[tokio::test]
    async fn from_env_hmac_round_trip_via_sink() {
        let capture = CaptureSink::new();
        let key = b"super-secret-shared-symmetric-key-bytes-padded";
        let sink = {
            let _g = FROM_ENV_MUTEX.lock().unwrap();
            clear_signing_env();
            std::env::set_var("CELLOS_EVENT_SIGNING", "hmac");
            std::env::set_var("CELLOS_EVENT_SIGNING_KID", "ops-hmac-2026-q2");
            std::env::set_var(
                "CELLOS_EVENT_SIGNING_KEY_BASE64",
                URL_SAFE_NO_PAD.encode(key),
            );
            SigningEventSink::from_env(capture.clone() as Arc<dyn EventSink>)
        };
        let event = sample_event();
        sink.emit(&event).await.unwrap();
        {
            let _g = FROM_ENV_MUTEX.lock().unwrap();
            clear_signing_env();
        }

        let got = capture.last().expect("wrapper emitted");
        assert_eq!(got.ty, SIGNED_ENVELOPE_TRANSPORT_TYPE);
        let envelope: SignedEventEnvelopeV1 =
            serde_json::from_value(got.data.expect("wrapper has data")).expect("parse envelope");
        assert_eq!(envelope.algorithm, "hmac-sha256");

        let verifying_keys: HashMap<String, cellos_core::TrustAnchorPublicKey> = HashMap::new();
        let mut hmac_keys: HashMap<String, Vec<u8>> = HashMap::new();
        hmac_keys.insert("ops-hmac-2026-q2".to_string(), key.to_vec());
        let inner = verify_signed_event_envelope(&envelope, &verifying_keys, &hmac_keys)
            .expect("verify ok");
        assert_eq!(inner.id, event.id);
    }

    #[tokio::test]
    async fn post_sign_event_mutation_fails_verification() {
        let capture = CaptureSink::new();
        let signer_seed = [29u8; 32];
        let signer = SigningKey::from_bytes(&signer_seed);
        let sink = {
            let _g = FROM_ENV_MUTEX.lock().unwrap();
            clear_signing_env();
            std::env::set_var("CELLOS_EVENT_SIGNING", "ed25519");
            std::env::set_var("CELLOS_EVENT_SIGNING_KID", "ops-event-2026-q2");
            std::env::set_var(
                "CELLOS_EVENT_SIGNING_KEY_BASE64",
                URL_SAFE_NO_PAD.encode(signer_seed),
            );
            SigningEventSink::from_env(capture.clone() as Arc<dyn EventSink>)
        };
        let event = sample_event();
        sink.emit(&event).await.unwrap();
        {
            let _g = FROM_ENV_MUTEX.lock().unwrap();
            clear_signing_env();
        }

        let got = capture.last().expect("wrapper emitted");
        let mut envelope: SignedEventEnvelopeV1 =
            serde_json::from_value(got.data.expect("wrapper has data")).expect("parse envelope");
        // Adversary mutates the carried event after signing — different id.
        envelope.event.id = "ev-tampered".into();

        let mut keys = HashMap::new();
        keys.insert(
            "ops-event-2026-q2".to_string(),
            cellos_core::TrustAnchorPublicKey::from_bytes_unchecked(
                signer.verifying_key().to_bytes(),
            ),
        );
        let hmac_keys: HashMap<String, Vec<u8>> = HashMap::new();
        let err = verify_signed_event_envelope(&envelope, &keys, &hmac_keys)
            .expect_err("post-sign mutation must fail verification");
        assert!(format!("{err}").contains("ed25519 verify failed"));
    }

    #[tokio::test]
    async fn from_env_missing_kid_disables_signing() {
        let capture = CaptureSink::new();
        let sink = {
            let _g = FROM_ENV_MUTEX.lock().unwrap();
            clear_signing_env();
            std::env::set_var("CELLOS_EVENT_SIGNING", "ed25519");
            // Deliberately no CELLOS_EVENT_SIGNING_KID set.
            std::env::set_var(
                "CELLOS_EVENT_SIGNING_KEY_BASE64",
                URL_SAFE_NO_PAD.encode([7u8; 32]),
            );
            SigningEventSink::from_env(capture.clone() as Arc<dyn EventSink>)
        };
        let event = sample_event();
        sink.emit(&event).await.unwrap();
        {
            let _g = FROM_ENV_MUTEX.lock().unwrap();
            clear_signing_env();
        }

        let got = capture.last().expect("event emitted");
        assert_eq!(
            got.ty, "dev.cellos.events.cell.lifecycle.v1.started",
            "missing kid must fall back to passthrough"
        );
    }

    // ── StandaloneEventSigner (ADR-0025 §2) ──────────────────────────────

    fn clear_signer_env() {
        clear_signing_env();
        std::env::remove_var("CELLOS_REDACT_EVENT_FIELDS");
    }

    /// A secret-bearing field is redacted BEFORE the signature is computed, so
    /// the verified envelope carries the placeholder, not the secret.
    #[tokio::test]
    async fn standalone_signer_redacts_before_signing() {
        let capture = CaptureSink::new();
        let signer_seed = [41u8; 32];
        let signer = SigningKey::from_bytes(&signer_seed);
        let signer_sink = {
            let _g = FROM_ENV_MUTEX.lock().unwrap();
            clear_signer_env();
            std::env::set_var("CELLOS_EVENT_SIGNING", "ed25519");
            std::env::set_var("CELLOS_EVENT_SIGNING_KID", "org-root-2026");
            std::env::set_var(
                "CELLOS_EVENT_SIGNING_KEY_BASE64",
                URL_SAFE_NO_PAD.encode(signer_seed),
            );
            std::env::set_var("CELLOS_REDACT_EVENT_FIELDS", "reason");
            StandaloneEventSigner::from_env(capture.clone() as Arc<dyn EventSink>).0
        };
        assert!(signer_sink.is_signing_enabled());

        let event = CloudEventV1 {
            data: Some(json!({"reason": "super-secret-db-password", "admitted": false})),
            ..sample_event()
        };
        let returned = signer_sink
            .emit_decision(&event)
            .await
            .expect("emit_decision ok")
            .expect("signed path returns the envelope");
        {
            let _g = FROM_ENV_MUTEX.lock().unwrap();
            clear_signer_env();
        }

        // The returned envelope carries the already-redacted event.
        let inner_data = returned.event.data.as_ref().expect("inner data");
        assert_eq!(
            inner_data.get("reason").and_then(|v| v.as_str()),
            Some("[redacted]"),
            "secret field must be redacted in the signed event"
        );

        // And the offline verification succeeds against ONLY the org-root key.
        let mut keys = HashMap::new();
        keys.insert(
            "org-root-2026".to_string(),
            cellos_core::TrustAnchorPublicKey::from_bytes_unchecked(
                signer.verifying_key().to_bytes(),
            ),
        );
        let hmac_keys: HashMap<String, Vec<u8>> = HashMap::new();
        let verified =
            verify_signed_event_envelope(&returned, &keys, &hmac_keys).expect("verify ok offline");
        assert_eq!(
            verified.data.as_ref().and_then(|d| d.get("reason")),
            inner_data.get("reason"),
            "verified payload is byte-identical to the signed (redacted) event"
        );
    }

    /// The signed wrapper that lands on the transport verifies offline; a single
    /// tampered byte makes verification fail.
    #[tokio::test]
    async fn standalone_signer_emits_verifiable_wrapper() {
        let capture = CaptureSink::new();
        let signer_seed = [53u8; 32];
        let signer = SigningKey::from_bytes(&signer_seed);
        let signer_sink = {
            let _g = FROM_ENV_MUTEX.lock().unwrap();
            clear_signer_env();
            std::env::set_var("CELLOS_EVENT_SIGNING", "ed25519");
            std::env::set_var("CELLOS_EVENT_SIGNING_KID", "org-root-2026");
            std::env::set_var(
                "CELLOS_EVENT_SIGNING_KEY_BASE64",
                URL_SAFE_NO_PAD.encode(signer_seed),
            );
            StandaloneEventSigner::from_env(capture.clone() as Arc<dyn EventSink>).0
        };
        let event = sample_event();
        signer_sink.emit_decision(&event).await.expect("emit ok");
        {
            let _g = FROM_ENV_MUTEX.lock().unwrap();
            clear_signer_env();
        }

        let got = capture.last().expect("wrapper emitted on transport");
        assert_eq!(got.ty, SIGNED_ENVELOPE_TRANSPORT_TYPE);
        let mut envelope: SignedEventEnvelopeV1 =
            serde_json::from_value(got.data.expect("wrapper data")).expect("parse envelope");

        let mut keys = HashMap::new();
        keys.insert(
            "org-root-2026".to_string(),
            cellos_core::TrustAnchorPublicKey::from_bytes_unchecked(
                signer.verifying_key().to_bytes(),
            ),
        );
        let hmac_keys: HashMap<String, Vec<u8>> = HashMap::new();
        verify_signed_event_envelope(&envelope, &keys, &hmac_keys).expect("verify ok");

        // One tampered byte makes verification fail.
        envelope.event.id = "ev-tampered".into();
        let err = verify_signed_event_envelope(&envelope, &keys, &hmac_keys)
            .expect_err("tamper must fail verification");
        assert!(format!("{err}").contains("ed25519 verify failed"));
    }

    /// With signing unconfigured the signer still emits a record carrying an
    /// explicit `signed: false` marker — never a missing receipt.
    #[tokio::test]
    async fn standalone_signer_unconfigured_emits_signed_false() {
        let capture = CaptureSink::new();
        let signer_sink = {
            let _g = FROM_ENV_MUTEX.lock().unwrap();
            clear_signer_env();
            // CELLOS_EVENT_SIGNING deliberately unset => unsigned advisory mode.
            StandaloneEventSigner::from_env(capture.clone() as Arc<dyn EventSink>).0
        };
        assert!(!signer_sink.is_signing_enabled());

        let event = sample_event();
        let returned = signer_sink
            .emit_decision(&event)
            .await
            .expect("emit_decision ok");
        {
            let _g = FROM_ENV_MUTEX.lock().unwrap();
            clear_signer_env();
        }

        assert!(
            returned.is_none(),
            "unsigned advisory mode returns no offline-verifiable envelope"
        );
        let got = capture.last().expect("advisory record emitted");
        assert_ne!(
            got.ty, SIGNED_ENVELOPE_TRANSPORT_TYPE,
            "advisory record is not a signed-envelope wrapper"
        );
        assert_eq!(
            got.data
                .as_ref()
                .and_then(|d| d.get(SIGNED_MARKER_FIELD))
                .and_then(|v| v.as_bool()),
            Some(false),
            "advisory record must carry signed:false"
        );
    }
}