use serde::{Deserialize, Serialize};
use std::fmt;
use std::path::{Path, PathBuf};
use std::sync::{Arc, OnceLock};
use tokio::sync::{OwnedSemaphorePermit, Semaphore};

pub type Result<T> = std::result::Result<T, AsxError>;

/// Escapes the five canonical XML special characters (`&`, `<`, `>`, `"`, `'`)
/// and strips XML 1.0 §2.2 forbidden control characters (including NUL bytes)
/// to prevent injection into SOAP/SBDH envelopes.
///
/// # ⚠ Forbidden-character stripping
///
/// Bytes in the ranges `U+0000–U+0008`, `U+000B–U+000C`, `U+000E–U+001F`, and
/// `U+007F` are **stripped** (not replaced) from the output.  If the input
/// contains such bytes, the output will differ from the input — which can
/// cause integrity failures that are hard to diagnose.  A `tracing::warn!`
/// event is emitted (when the `trace` feature is enabled) so that callers
/// can detect this condition via their observability pipeline.
///
/// If you need to **reject** inputs containing forbidden characters rather
/// than silently strip them, validate `s` before calling this function.
pub fn escape_xml(s: &str) -> String {
    let mut out = String::new();
    let mut last = 0usize;
    let mut modified = false;
    let mut stripped_count: usize = 0;
    for (i, b) in s.bytes().enumerate() {
        let escaped = match b {
            // Forbidden XML 1.0 §2.2 characters — strip and warn.
            0x00..=0x08 | 0x0B..=0x0C | 0x0E..=0x1F | 0x7F => {
                if !modified {
                    out.reserve(s.len());
                    modified = true;
                }
                out.push_str(&s[last..i]);
                last = i + 1;
                stripped_count += 1;
                continue;
            }
            b'&' => "&amp;",
            b'<' => "&lt;",
            b'>' => "&gt;",
            b'"' => "&quot;",
            b'\'' => "&apos;",
            _ => continue,
        };
        if !modified {
            out.reserve(s.len() + 16);
            modified = true;
        }
        out.push_str(&s[last..i]);
        out.push_str(escaped);
        last = i + 1;
    }
    if !modified {
        return s.to_owned();
    }
    out.push_str(&s[last..]);
    if stripped_count > 0 {
        // Emit a warning observable via the tracing subscriber so that
        // embedders can detect and alert on truncated XML field values.
        tracing::warn!(
            stripped_bytes = stripped_count,
            "escape_xml stripped {} forbidden XML 1.0 control character(s) from input; \
             output differs from input — check the source field for binary/control data",
            stripped_count,
        );
    }
    out
}

fn default_blocking_crypto_concurrency() -> usize {
    // Slightly above core-count keeps throughput stable while bounding queueing.
    std::thread::available_parallelism()
        .map(|n| n.get().saturating_mul(2))
        .unwrap_or(8)
        .clamp(4, 128)
}

const BLOCKING_CRYPTO_CONCURRENCY_ENV: &str = "ASX_BLOCKING_CRYPTO_CONCURRENCY";

fn configured_blocking_crypto_concurrency() -> usize {
    std::env::var(BLOCKING_CRYPTO_CONCURRENCY_ENV)
        .ok()
        .and_then(|raw| raw.trim().parse::<usize>().ok())
        .filter(|value| *value > 0)
        .map(|value| value.clamp(1, 4096))
        .unwrap_or_else(default_blocking_crypto_concurrency)
}

fn blocking_crypto_semaphore() -> Arc<Semaphore> {
    static SEM: OnceLock<Arc<Semaphore>> = OnceLock::new();
    Arc::clone(
        SEM.get_or_init(|| Arc::new(Semaphore::new(configured_blocking_crypto_concurrency()))),
    )
}

// ---------------------------------------------------------------------------
// CryptoAdmissionControl — instance-scoped or process-global semaphore
// ---------------------------------------------------------------------------

/// Default inbound payload ceiling (256 MiB).
///
/// Production EDI batch files routinely exceed 50–100 MB; this default
/// accommodates X12, EDIFACT, and Peppol BIS payloads without operator
/// intervention.  Override via `StreamLimits` for resource-constrained
/// deployments.
pub const DEFAULT_MAX_BODY_BYTES: usize = 256 * 1024 * 1024;

/// Controls how many concurrent CPU-heavy crypto/protocol tasks are admitted.
///
/// The default ([`CryptoAdmissionControl::process_global()`]) shares a
/// process-wide semaphore across all sessions.  For multi-tenant embeddings,
/// create a **per-tenant** instance so that one tenant's burst traffic cannot
/// starve another's:
///
/// ```rust,ignore
/// let control = Arc::new(CryptoAdmissionControl::new(32));
/// // Store in your tenant context and call:
/// // control.acquire(stage, session).await?
/// ```
///
/// ## Choosing a concurrency limit
///
/// A value of `num_cpus * 2` (the default) is appropriate for workloads where
/// crypto dominates.  For mixed workloads, set it to the number of Tokio
/// blocking threads you are willing to dedicate to crypto work.
#[derive(Clone, Debug)]
pub struct CryptoAdmissionControl {
    semaphore: Arc<Semaphore>,
    /// Human-readable label used in error messages.
    label: &'static str,
}

impl CryptoAdmissionControl {
    /// Create a new **instance-scoped** admission controller with `concurrency`
    /// permits.
    ///
    /// Use this for per-tenant or per-connection isolation.
    pub fn new(concurrency: usize) -> Self {
        let cap = concurrency.clamp(1, 4096);
        Self {
            semaphore: Arc::new(Semaphore::new(cap)),
            label: "instance-scoped crypto semaphore",
        }
    }

    /// Return the **process-global** admission controller.
    ///
    /// Concurrency is configured by the `ASX_BLOCKING_CRYPTO_CONCURRENCY` env var
    /// or defaults to `num_cpus × 2`.
    pub fn process_global() -> Self {
        Self {
            semaphore: blocking_crypto_semaphore(),
            label: "process-global crypto semaphore",
        }
    }

    /// Acquire one permit, waiting if all permits are currently held.
    pub async fn acquire(
        &self,
        stage: &'static str,
        session: &SessionContext,
    ) -> Result<OwnedSemaphorePermit> {
        Arc::clone(&self.semaphore)
            .acquire_owned()
            .await
            .map_err(|_| {
                AsxError::new(
                    ErrorCode::TransportFailure,
                    format!("{} is closed", self.label),
                    ErrorContext::for_session(stage, session),
                )
            })
    }
}

/// Interpret a byte slice as UTF-8, returning a contextual error on failure.
/// Avoids repeating the 4-line `from_utf8(...).map_err(|_| AsxError::new(...))` pattern.
#[cfg(feature = "as4")]
pub(crate) fn bytes_to_utf8_str<'a>(
    bytes: &'a [u8],
    stage: &'static str,
    session: &SessionContext,
) -> Result<&'a str> {
    std::str::from_utf8(bytes).map_err(|_| {
        AsxError::new(
            ErrorCode::ParseFailed,
            format!("{stage}: payload is not valid UTF-8"),
            ErrorContext::for_session(stage, session),
        )
    })
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ErrorContext {
    pub stage: &'static str,
    pub message_id: Option<String>,
    pub partner_id: Option<String>,
    pub session_id: Option<String>,
}

impl ErrorContext {
    #[must_use]
    pub fn new(stage: &'static str) -> Self {
        Self {
            stage,
            message_id: None,
            partner_id: None,
            session_id: None,
        }
    }

    /// Create an `ErrorContext` pre-populated with session and partner IDs from a `SessionContext`.
    /// Prefer this over chaining `with_session_id` + `with_partner_id` to avoid redundant clones.
    #[must_use]
    pub fn for_session(stage: &'static str, session: &SessionContext) -> Self {
        Self::new(stage).with_session_and_partner(session.session_id(), session.partner_id())
    }

    /// Create an `ErrorContext` pre-populated with session, partner, and message IDs.
    /// Prefer this over chaining three separate `with_*` calls.
    #[must_use]
    pub fn for_session_with_message(
        stage: &'static str,
        session: &SessionContext,
        message_id: impl Into<String>,
    ) -> Self {
        Self::new(stage)
            .with_session_and_partner(session.session_id(), session.partner_id())
            .with_message_id(message_id)
    }

    #[must_use]
    pub fn with_session_and_partner(
        mut self,
        session_id: impl Into<String>,
        partner_id: impl Into<String>,
    ) -> Self {
        self.session_id = Some(session_id.into());
        self.partner_id = Some(partner_id.into());
        self
    }

    #[must_use]
    pub fn with_message_id(mut self, message_id: impl Into<String>) -> Self {
        self.message_id = Some(message_id.into());
        self
    }

    #[must_use]
    pub fn with_partner_id(mut self, partner_id: impl Into<String>) -> Self {
        self.partner_id = Some(partner_id.into());
        self
    }

    #[must_use]
    pub fn with_session_id(mut self, session_id: impl Into<String>) -> Self {
        self.session_id = Some(session_id.into());
        self
    }
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[non_exhaustive]
pub enum ErrorCode {
    InvalidInput,
    ParseFailed,
    SecurityVerificationFailed,
    DecryptionFailed,
    PolicyViolation,
    TransportFailure,
    InteropViolation,
    ReliabilityFailure,
    /// A requested resource (e.g. SMP endpoint, profile) does not exist.
    NotFound,
    /// A bounded resource (e.g. conversation gate, connection pool) has no
    /// remaining capacity.  Callers should shed load or retry after a delay.
    CapacityExhausted,
    /// The inbound request body exceeds the configured size limit.
    ///
    /// HTTP semantics: respond with 413 Content Too Large.
    PayloadTooLarge,
    /// A storage or infrastructure backend (dedup store, reconciliation queue,
    /// audit sink) failed with an I/O or connectivity error.
    ///
    /// This is distinct from [`ReliabilityFailure`](Self::ReliabilityFailure)
    /// (protocol-level duplicate/ordering failure) and from
    /// [`TransportFailure`](Self::TransportFailure) (network/HTTP failure).
    ///
    /// HTTP semantics: respond with 503 Service Unavailable — the server is
    /// temporarily unable to handle the request due to a backend outage.
    StorageBackendFailure,
}

impl ErrorCode {
    pub fn as_str(self) -> &'static str {
        match self {
            Self::InvalidInput => "invalid_input",
            Self::ParseFailed => "parse_failed",
            Self::SecurityVerificationFailed => "security_verification_failed",
            Self::DecryptionFailed => "decryption_failed",
            Self::PolicyViolation => "policy_violation",
            Self::TransportFailure => "transport_failure",
            Self::InteropViolation => "interop_violation",
            Self::ReliabilityFailure => "reliability_failure",
            Self::NotFound => "not_found",
            Self::CapacityExhausted => "capacity_exhausted",
            Self::PayloadTooLarge => "payload_too_large",
            Self::StorageBackendFailure => "storage_backend_failure",
        }
    }

    /// Returns the canonical HTTP status code for this error.
    ///
    /// Embedders can use this in transport adapters to map `AsxError` to HTTP
    /// responses without writing a custom match.  For errors that represent
    /// an external failure (e.g., `TransportFailure`), the code is 5xx; for
    /// caller-induced errors it is 4xx.
    ///
    /// | Variant                       | HTTP status |
    /// |-------------------------------|-------------|
    /// | `InvalidInput`                | 400         |
    /// | `ParseFailed`                 | 400         |
    /// | `SecurityVerificationFailed`  | 401         |
    /// | `DecryptionFailed`            | 400         |
    /// | `PolicyViolation`             | 422         |
    /// | `TransportFailure`            | 502         |
    /// | `InteropViolation`            | 400         |
    /// | `ReliabilityFailure`          | 503         |
    /// | `NotFound`                    | 404         |
    /// | `CapacityExhausted`           | 429         |
    /// | `PayloadTooLarge`             | 413         |
    /// | `StorageBackendFailure`        | 503         |
    pub fn to_http_status(self) -> u16 {
        match self {
            Self::InvalidInput => 400,
            Self::ParseFailed => 400,
            Self::SecurityVerificationFailed => 401,
            Self::DecryptionFailed => 400,
            Self::PolicyViolation => 422,
            Self::TransportFailure => 502,
            Self::InteropViolation => 400,
            Self::ReliabilityFailure => 503,
            Self::NotFound => 404,
            Self::CapacityExhausted => 429,
            Self::PayloadTooLarge => 413,
            Self::StorageBackendFailure => 503,
        }
    }

    /// A short operator-facing remediation hint for the most common failure
    /// codes, or `None` for failures that require caller-specific diagnosis.
    ///
    /// Intended for structured logging and monitoring dashboards; not a
    /// substitute for full error context.
    pub fn remediation_hint(self) -> Option<&'static str> {
        match self {
            Self::DecryptionFailed => Some(
                "Verify that the recipient certificate PEM and its private key PEM match. \
                 Ensure the sender is encrypting to the correct public certificate. \
                 Re-key the key pair if the certificate has been re-issued.",
            ),
            Self::SecurityVerificationFailed => Some(
                "Confirm the trust anchor PEM includes the full CA chain of the signer. \
                 Check certificate validity period. \
                 Ensure CRL distribution points or OCSP responders are reachable.",
            ),
            Self::TransportFailure => Some(
                "Check network connectivity and DNS resolution for the remote endpoint. \
                 Verify TLS certificate chain and mutual-TLS configuration. \
                 Ensure the spool directory exists and is writable.",
            ),
            Self::ReliabilityFailure => Some(
                "Ensure an EventBus broadcast subscriber is active before message sends. \
                 Check dedup and reconciliation backend availability and capacity.",
            ),
            Self::PolicyViolation => Some(
                "Review the PMode and profile configuration against the partner specification. \
                 Verify the interop mode matches the partner's published requirements.",
            ),
            Self::CapacityExhausted => Some(
                "Shed load or retry after a backoff delay. \
                 Consider increasing channel capacity or conversation gate limits.",
            ),
            Self::StorageBackendFailure => Some(
                "Check dedup/reconciliation/audit backend connectivity and disk space. \
                 Inspect backend logs for I/O errors. \
                 Consider a circuit-breaker or fallback backend for resilience.",
            ),
            _ => None,
        }
    }
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct AsxError {
    pub code: ErrorCode,
    pub message: String,
    pub context: ErrorContext,
}

impl AsxError {
    pub fn new(code: ErrorCode, message: impl Into<String>, context: ErrorContext) -> Self {
        Self {
            code,
            message: message.into(),
            context,
        }
    }

    /// Short operator-facing remediation hint, delegated from [`ErrorCode::remediation_hint`].
    ///
    /// Returns `None` for error codes that do not have a generic hint.
    pub fn remediation_hint(&self) -> Option<&'static str> {
        self.code.remediation_hint()
    }

    /// Enrich the error context with the partner ID for easier correlation in embedder logs.
    ///
    /// Delegates to [`ErrorContext::with_partner_id`]. Designed for use in `map_err` closures
    /// where the full `ErrorContext` is not directly accessible:
    ///
    /// ```ignore
    /// some_op().map_err(|e| e.with_partner_id(session.partner_id()))?;
    /// ```
    #[must_use]
    pub fn with_partner_id(mut self, partner_id: impl Into<String>) -> Self {
        self.context = self.context.with_partner_id(partner_id);
        self
    }

    /// Enrich the error context with the session ID for easier correlation in embedder logs.
    ///
    /// Delegates to [`ErrorContext::with_session_id`].
    #[must_use]
    pub fn with_session_id(mut self, session_id: impl Into<String>) -> Self {
        self.context = self.context.with_session_id(session_id);
        self
    }

    /// Enrich the error context with both session and partner IDs in one call.
    ///
    /// Delegates to [`ErrorContext::with_session_and_partner`].
    #[must_use]
    pub fn with_session_and_partner(
        mut self,
        session_id: impl Into<String>,
        partner_id: impl Into<String>,
    ) -> Self {
        self.context = self
            .context
            .with_session_and_partner(session_id, partner_id);
        self
    }

    /// Enrich the error context with the message ID for easier correlation in embedder logs.
    ///
    /// Delegates to [`ErrorContext::with_message_id`].
    #[must_use]
    pub fn with_message_id(mut self, message_id: impl Into<String>) -> Self {
        self.context = self.context.with_message_id(message_id);
        self
    }

    /// Returns `true` if this error represents a replay-protection rejection
    /// (the message was already seen by the dedup store).
    ///
    /// Equivalent to `err.code == ErrorCode::ReliabilityFailure && err.message contains "replay"`,
    /// but stable across message-text changes.  Use this instead of matching on error text.
    ///
    /// # Example
    /// ```rust,ignore
    /// if let Err(e) = receive_push_with_dedup_async(...).await {
    ///     if e.is_duplicate() {
    ///         // idempotent: send receipt without re-processing
    ///     }
    /// }
    /// ```
    ///
    /// **Prefer [`asx_rs::as4::As4ReceiveOutcome`] over this method** — the discriminated
    /// outcome enum is the primary API for duplicate detection on the receive path.
    /// This method is provided for error-path scenarios (e.g., when a storage backend
    /// fails-closed and propagates a duplicate as an error rather than an outcome).
    #[inline]
    pub fn is_duplicate(&self) -> bool {
        self.code == ErrorCode::ReliabilityFailure && self.message.contains("replay")
    }

    /// Returns `true` if this error represents a transient storage or infrastructure failure.
    #[inline]
    pub fn is_storage_failure(&self) -> bool {
        self.code == ErrorCode::StorageBackendFailure
    }
}

impl fmt::Display for AsxError {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        write!(
            f,
            "{} [{}|stage={}]",
            self.message,
            self.code.as_str(),
            self.context.stage
        )?;
        if let Some(pid) = &self.context.partner_id {
            write!(f, "[partner={pid}]")?;
        }
        if let Some(mid) = &self.context.message_id {
            write!(f, "[msg={mid}]")?;
        }
        if let Some(sid) = &self.context.session_id {
            write!(f, "[session={sid}]")?;
        }
        Ok(())
    }
}

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

#[derive(Debug, Clone, Copy, PartialEq, Eq, Default, Serialize, Deserialize)]
#[non_exhaustive]
pub enum InteropMode {
    #[default]
    Strict,
    /// Relaxed interoperability mode — permits deviations from strict AS2/AS4
    /// profile requirements to accommodate legacy or non-conformant partners.
    ///
    /// # Feature gate
    ///
    /// This variant is only available when the **`interop-relaxed`** Cargo
    /// feature is enabled:
    ///
    /// ```toml
    /// [dependencies]
    /// asx = { version = "0.2", features = ["interop-relaxed"] }
    /// ```
    #[cfg(feature = "interop-relaxed")]
    Relaxed,
}

/// Per-partner session context — identity, trust configuration, and correlation
/// metadata required by every send/receive operation in `asx`.
///
/// # Cardinality and Lifetime
///
/// `SessionContext` is scoped to **one trading-partner relationship** and
/// represents a reusable, long-lived object — not a per-message allocation.
/// Typical lifecycle patterns:
///
/// | Deployment style | Recommended granularity |
/// |---|---|
/// | Single fixed partner (e.g. one supplier) | One `SessionContext` per process; share via `Arc`. |
/// | Multiple partners (hub/spoke) | One `SessionContext` per partner; keyed by `partner_id`. |
/// | Short-lived CLI / batch | One `SessionContext` per batch run; `Clone` is O(1). |
///
/// Using a *new* `SessionContext` for every outbound message is valid but
/// wasteful — it pays session-ID generation and cert-validation costs on every
/// call.  Prefer reusing and, when certificates rotate, call
/// [`rotate_cert_handle`] in place rather than rebuilding.
///
/// # Observability Impact
///
/// The `session_id` is emitted on every span, metric label, and error context
/// produced during message processing.  Using inconsistent or randomly
/// generated `session_id` values per message will fragment observability data
/// in your monitoring backend and make per-partner dashboards unusable.
/// Choose a stable, human-readable ID such as `"partner-acme-prod"`.
///
/// # Cloning
///
/// `Clone` is O(1) — all heavy data (certificates, trust anchors) is behind
/// `Arc` and is not deep-copied.  Clones share the same `CertHandle` lineage
/// until [`rotate_cert_handle`] is called on one of them.
///
/// # Construction
///
/// Prefer [`SessionContext::builder`] for incremental construction, or
/// [`SessionContext::new`] for the minimal three-field shorthand.
///
/// [`rotate_cert_handle`]: SessionContext::rotate_cert_handle
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct SessionContext {
    session_id: String,
    partner_id: String,
    profile_name: String,
    metadata: SessionMetadata,
    /// The certificate/trust configuration for this session.
    ///
    /// Stored behind an `Arc` so that:
    /// - `SessionContext::clone()` is O(1) — increments a refcount rather than
    ///   deep-copying all PEM strings and DER blobs.
    /// - In-flight verifications continue using the cert snapshot they started
    ///   with even after a rotation (Arc keeps the old handle alive).
    /// - [`rotate_cert_handle`] can swap the handle without rebuilding the session,
    ///   preserving `session_id`, `partner_id`, reliability queues, and event
    ///   subscriptions.
    ///
    /// Access via [`SessionContext::cert_handle`].
    ///
    /// [`rotate_cert_handle`]: SessionContext::rotate_cert_handle
    cert_handle: Arc<CertHandle>,
    correlation_scope: CorrelationScope,
}

#[derive(Debug, Clone, PartialEq, Eq, Default)]
pub struct SessionMetadata {
    pub effective_policy_snapshot_json: Option<String>,
    pub strict_runtime_bootstrap_validated: bool,
}

/// Builder for ergonomic incremental construction of [`SessionContext`].
#[derive(Debug, Clone)]
pub struct SessionContextBuilder {
    session_id: String,
    partner_id: String,
    profile_name: String,
    cert_handle: Option<CertHandle>,
    effective_policy_snapshot_json: Option<String>,
    correlation_scope: Option<CorrelationScope>,
}

impl SessionContextBuilder {
    /// Create a new builder with required identity fields.
    ///
    /// `profile_name` defaults to `"strict"` and can be overridden with
    /// [`profile_name`][Self::profile_name].
    pub fn new(session_id: impl Into<String>, partner_id: impl Into<String>) -> Self {
        Self {
            session_id: session_id.into(),
            partner_id: partner_id.into(),
            profile_name: "strict".to_string(),
            cert_handle: None,
            effective_policy_snapshot_json: None,
            correlation_scope: None,
        }
    }

    /// Override the profile name (defaults to `"strict"`).
    pub fn profile_name(mut self, profile_name: impl Into<String>) -> Self {
        self.profile_name = profile_name.into();
        self
    }

    /// Set explicit certificate/trust material for the session.
    pub fn cert_handle(mut self, cert_handle: CertHandle) -> Self {
        self.cert_handle = Some(cert_handle);
        self
    }

    /// Append a single PEM-encoded trust anchor to the session's certificate store.
    ///
    /// This is a convenience alternative to constructing a [`CertHandle`] manually.
    /// Can be called multiple times to build up a set of trusted root/intermediate CAs.
    ///
    /// # Example
    /// ```rust,ignore
    /// let session = SessionContextBuilder::new("s1", "partner-a")
    ///     .with_trust_anchor_pem(root_ca_pem)
    ///     .build()?;
    /// ```
    pub fn with_trust_anchor_pem(mut self, pem: impl Into<String>) -> Self {
        let cert_handle = self.cert_handle.get_or_insert_with(|| CertHandle::new(""));
        cert_handle.trust_anchor_pems.push(pem.into());
        self
    }

    /// Append multiple PEM-encoded trust anchors to the session's certificate store.
    ///
    /// Equivalent to calling [`with_trust_anchor_pem`][Self::with_trust_anchor_pem]
    /// in a loop.
    pub fn with_trust_anchor_pems(
        mut self,
        pems: impl IntoIterator<Item = impl Into<String>>,
    ) -> Self {
        let cert_handle = self.cert_handle.get_or_insert_with(|| CertHandle::new(""));
        cert_handle
            .trust_anchor_pems
            .extend(pems.into_iter().map(|p| p.into()));
        self
    }

    /// Set the OCSP revocation-check mode for the session.
    ///
    /// Defaults to [`OcspMode::default()`] when not specified.
    pub fn with_ocsp_mode(mut self, mode: OcspMode) -> Self {
        let cert_handle = self.cert_handle.get_or_insert_with(|| CertHandle::new(""));
        cert_handle.ocsp_mode = mode;
        self
    }

    /// Attach a serialized effective policy snapshot JSON string.
    pub fn effective_policy_snapshot_json(mut self, snapshot_json: impl Into<String>) -> Self {
        self.effective_policy_snapshot_json = Some(snapshot_json.into());
        self
    }

    /// Override the default correlation scope.
    pub fn correlation_scope(
        mut self,
        root_id: impl Into<String>,
        parent_message_id: Option<String>,
    ) -> Self {
        self.correlation_scope = Some(CorrelationScope {
            root_id: root_id.into(),
            parent_message_id,
            traceparent: None,
        });
        self
    }

    /// Build a validated [`SessionContext`].
    pub fn build(self) -> Result<SessionContext> {
        let mut session = SessionContext::new(self.session_id, self.partner_id, self.profile_name)?;

        if let Some(cert_handle) = self.cert_handle {
            session = session.with_cert_handle(cert_handle)?;
        }

        if let Some(correlation_scope) = self.correlation_scope {
            if correlation_scope.root_id.trim().is_empty() {
                return Err(AsxError::new(
                    ErrorCode::InvalidInput,
                    "correlation root_id must not be empty",
                    ErrorContext::for_session("session_context_builder", &session),
                ));
            }
            session.correlation_scope = correlation_scope;
        }

        if let Some(snapshot_json) = self.effective_policy_snapshot_json {
            session = session.with_effective_policy_snapshot_json(snapshot_json)?;
        }

        Ok(session)
    }
}

/// Lazy cache of trust-anchor X.509 certificates parsed from
/// [`CertHandle::trust_anchor_pems`].  Shared across clones of the same
/// [`CertHandle`] via an `Arc` so that clones see the same populated cache
/// rather than re-parsing independently.
///
/// Equality is always `true` — the cache is an implementation detail derived
/// from the authoritative `trust_anchor_pems` field; it does not contribute to
/// the identity of a [`CertHandle`].
#[derive(Debug, Default, Clone)]
#[cfg(any(feature = "as2", feature = "as4"))]
pub(crate) struct TrustAnchorCache(Arc<OnceLock<Vec<openssl::x509::X509>>>);

#[cfg(any(feature = "as2", feature = "as4"))]
impl PartialEq for TrustAnchorCache {
    fn eq(&self, _: &Self) -> bool {
        true // cache state is not part of CertHandle identity
    }
}
#[cfg(any(feature = "as2", feature = "as4"))]
impl Eq for TrustAnchorCache {}

/// Lazy cache of the `X509Store` built from [`CertHandle::trust_anchor_pems`].
///
/// Caching avoids rebuilding an `X509Store` (O(n_anchors) OpenSSL allocations)
/// on every inbound message verification.  Shared across clones via `Arc`;
/// the store is built at most once per `CertHandle` lineage regardless of how
/// many clones exist.  Equality is always `true` (derived from `trust_anchor_pems`).
#[derive(Default, Clone)]
#[cfg(any(feature = "as2", feature = "as4"))]
pub(crate) struct X509StoreCache(Arc<OnceLock<Arc<openssl::x509::store::X509Store>>>);

#[cfg(any(feature = "as2", feature = "as4"))]
impl std::fmt::Debug for X509StoreCache {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_tuple("X509StoreCache")
            .field(&self.0.get().is_some())
            .finish()
    }
}

#[cfg(any(feature = "as2", feature = "as4"))]
impl PartialEq for X509StoreCache {
    fn eq(&self, _: &Self) -> bool {
        true
    }
}
#[cfg(any(feature = "as2", feature = "as4"))]
impl Eq for X509StoreCache {}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct CertHandle {
    pub key_id: String,
    pub fingerprint_sha256: String,
    pub trust_anchor_pems: Vec<String>,
    /// Intermediate CA certificates (PEM) used for chain building during
    /// signature verification.  Supply any CAs that issued the partner's
    /// signing certificate when those intermediates are not embedded in
    /// the signed payload and are not already in `trust_anchor_pems`.
    pub intermediate_ca_pems: Vec<String>,
    pub revocation_crl_pems: Vec<String>,
    pub ocsp_mode: OcspMode,
    pub ocsp_failure_mode: OcspFailureMode,
    pub stapled_ocsp_responses_der: Vec<Vec<u8>>,
    pub responder_ocsp_responses_der: Vec<Vec<u8>>,
    /// Lazy-parsed trust anchor X.509 certificates — shared across clones.
    /// Populated on first call to [`CertHandle::trust_anchors_x509`].
    #[cfg(any(feature = "as2", feature = "as4"))]
    pub(crate) trust_anchors_cache: TrustAnchorCache,
    /// Lazy-built X.509 trust store derived from `trust_anchor_pems`.
    /// Shared across clones; built at most once per `CertHandle` lineage.
    /// Populated on first call to [`CertHandle::trust_anchor_x509_store`].
    #[cfg(any(feature = "as2", feature = "as4"))]
    pub(crate) x509_store_cache: X509StoreCache,
}

impl CertHandle {
    /// Construct a `CertHandle` with sensible defaults for the given key ID.
    ///
    /// All PEM/DER fields default to empty (`Vec::new()`), with OCSP mode
    /// `ResponderOnly` and `HardFail`.  Override any field with struct update
    /// syntax before passing to [`SessionContext::with_cert_handle`] or
    /// [`SessionContext::rotate_cert_handle`]:
    ///
    /// ```rust,ignore
    /// let handle = CertHandle {
    ///     trust_anchor_pems: vec![root_pem],
    ///     ocsp_mode: OcspMode::Disabled,
    ///     ..CertHandle::new("partner-cert")
    /// };
    /// ```
    pub fn new(key_id: impl Into<String>) -> Self {
        Self {
            key_id: key_id.into(),
            fingerprint_sha256: String::new(),
            trust_anchor_pems: Vec::new(),
            intermediate_ca_pems: Vec::new(),
            revocation_crl_pems: Vec::new(),
            ocsp_mode: OcspMode::default(),
            ocsp_failure_mode: OcspFailureMode::HardFail,
            stapled_ocsp_responses_der: Vec::new(),
            responder_ocsp_responses_der: Vec::new(),
            #[cfg(any(feature = "as2", feature = "as4"))]
            trust_anchors_cache: TrustAnchorCache::default(),
            #[cfg(any(feature = "as2", feature = "as4"))]
            x509_store_cache: X509StoreCache::default(),
        }
    }

    /// Return the parsed trust-anchor X.509 certificates, parsing from
    /// [`Self::trust_anchor_pems`] on the first call.  Subsequent calls
    /// (including from clones of this handle) return the cached result at
    /// zero parse cost.
    ///
    /// Each returned `X509` is backed by OpenSSL's internal reference counting,
    /// so cloning the returned `Vec` is O(n) in the number of anchors but O(1)
    /// per certificate.
    #[cfg(any(feature = "as2", feature = "as4"))]
    pub(crate) fn trust_anchors_x509(&self) -> Result<Vec<openssl::x509::X509>> {
        // Fast path: already initialised by a previous call on any clone.
        if let Some(anchors) = self.trust_anchors_cache.0.get() {
            return Ok(anchors.clone());
        }
        // Slow path: parse once and populate the lock.  If two threads race
        // here both compute the same result (deterministic from the PEM input)
        // and the loser's `set` is silently dropped — correctness is preserved.
        let mut anchors = Vec::new();
        for pem in &self.trust_anchor_pems {
            let certs = openssl::x509::X509::stack_from_pem(pem.as_bytes()).map_err(|e| {
                AsxError::new(
                    ErrorCode::InvalidInput,
                    format!("invalid trust-anchor PEM in CertHandle: {e}"),
                    ErrorContext::new("cert_handle_parse_anchors"),
                )
            })?;
            anchors.extend(certs);
        }
        // Ignore Err — means another thread already set the same value.
        let _ = self.trust_anchors_cache.0.set(anchors.clone());
        Ok(anchors)
    }

    /// Return an `Arc`-wrapped `X509Store` built from the trust-anchor PEMs,
    /// constructing and caching it on the first call.  Subsequent calls
    /// (including from clones of this handle) return the same `Arc` at zero cost.
    ///
    /// The store is built from `trust_anchor_pems` only; CRL/OCSP revocation
    /// checks are handled separately by the verification pipeline.
    #[cfg(any(feature = "as2", feature = "as4"))]
    pub(crate) fn trust_anchor_x509_store(&self) -> Result<Arc<openssl::x509::store::X509Store>> {
        if let Some(store) = self.x509_store_cache.0.get() {
            return Ok(Arc::clone(store));
        }
        // Build from cached parsed certs (avoids re-parsing PEM on slow path).
        let anchors = self.trust_anchors_x509()?;
        let mut builder = openssl::x509::store::X509StoreBuilder::new().map_err(|e| {
            AsxError::new(
                ErrorCode::InvalidInput,
                format!("failed to build X.509 trust store: {e}"),
                ErrorContext::new("cert_handle_build_x509_store"),
            )
        })?;
        for cert in &anchors {
            builder.add_cert(cert.clone()).map_err(|e| {
                AsxError::new(
                    ErrorCode::InvalidInput,
                    format!("failed to add trust anchor to X.509 store: {e}"),
                    ErrorContext::new("cert_handle_build_x509_store"),
                )
            })?;
        }
        let store = Arc::new(builder.build());
        let _ = self.x509_store_cache.0.set(Arc::clone(&store));
        Ok(store)
    }
}

#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
#[non_exhaustive]
pub enum OcspMode {
    Disabled,
    StapledOnly,
    #[default]
    ResponderOnly,
    StapledThenResponder,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
#[non_exhaustive]
pub enum OcspFailureMode {
    #[default]
    HardFail,
    SoftFail,
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct CorrelationScope {
    pub root_id: String,
    pub parent_message_id: Option<String>,
    /// Inbound W3C Trace Context `traceparent` header value, if present and valid.
    ///
    /// Populated by the transport ingress layer for inbound messages and
    /// propagated into every [`ScopedAsxEvent`] emitted during the
    /// corresponding processing session, enabling distributed-trace correlation
    /// with upstream callers.
    ///
    /// [`ScopedAsxEvent`]: crate::observability::ScopedAsxEvent
    pub traceparent: Option<Arc<str>>,
}

/// Payload bytes supplied to AS2/AS4 send/receive operations.
///
/// Consolidates the previously separate `As2PayloadInput` and `PushPayloadInput` types.
/// `Shared` avoids a copy when the caller already holds an `Arc<[u8]>`.
#[non_exhaustive]
pub enum PayloadInput<'a> {
    Owned(Vec<u8>),
    Shared(Arc<[u8]>),
    Borrowed(&'a [u8]),
}

impl<'a> PayloadInput<'a> {
    pub fn as_slice(&self) -> &[u8] {
        match self {
            Self::Owned(payload) => payload,
            Self::Shared(payload) => payload,
            Self::Borrowed(payload) => payload,
        }
    }

    pub fn into_arc(self) -> Arc<[u8]> {
        match self {
            Self::Owned(payload) => Arc::from(payload),
            Self::Shared(payload) => payload,
            Self::Borrowed(payload) => Arc::from(payload),
        }
    }
}

/// Receive-body abstraction used by verifier contracts.
///
/// In-memory handles avoid extra copies, while spooled handles keep large
/// messages out of RSS and make materialization an explicit decision point.
#[derive(Debug, Clone, PartialEq, Eq)]
#[non_exhaustive]
pub enum SpoolEncryption {
    Plaintext,
    Aes256Gcm { key: Arc<[u8; 32]> },
}

pub(crate) const SPOOLED_AES256_GCM_MAGIC: [u8; 8] = *b"ASXSPG01";
pub(crate) const SPOOLED_AES256_GCM_NONCE_LEN: usize = 12;
pub(crate) const SPOOLED_AES256_GCM_TAG_LEN: usize = 16;

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct SpoolLifecyclePolicy {
    pub delete_on_materialize: bool,
    pub secure_delete_on_materialize: bool,
}

impl Default for SpoolLifecyclePolicy {
    fn default() -> Self {
        Self {
            delete_on_materialize: true,
            secure_delete_on_materialize: false,
        }
    }
}

#[derive(Debug, Clone, PartialEq, Eq)]
#[non_exhaustive]
pub enum ReceivedBodyHandle {
    InMemory(Arc<[u8]>),
    Spooled {
        path: PathBuf,
        encryption: SpoolEncryption,
        lifecycle: SpoolLifecyclePolicy,
    },
}

impl ReceivedBodyHandle {
    #[must_use]
    pub fn from_payload_input(input: PayloadInput<'_>) -> Self {
        Self::InMemory(input.into_arc())
    }

    pub fn payload_len(&self, stage: &'static str, session: &SessionContext) -> Result<usize> {
        match self {
            Self::InMemory(bytes) => Ok(bytes.len()),
            Self::Spooled { path, .. } => {
                let metadata = std::fs::metadata(path).map_err(|err| {
                    AsxError::new(
                        ErrorCode::TransportFailure,
                        format!("failed to stat spooled body {}: {err}", path.display()),
                        ErrorContext::for_session(stage, session),
                    )
                })?;
                usize::try_from(metadata.len()).map_err(|_| {
                    AsxError::new(
                        ErrorCode::PolicyViolation,
                        format!(
                            "spooled body {} exceeds platform addressable size",
                            path.display()
                        ),
                        ErrorContext::for_session(stage, session),
                    )
                })
            }
        }
    }

    pub fn materialize_contiguous(
        &self,
        stage: &'static str,
        session: &SessionContext,
    ) -> Result<Arc<[u8]>> {
        match self {
            Self::InMemory(bytes) => Ok(Arc::clone(bytes)),
            Self::Spooled {
                path, encryption, ..
            } => Ok(Arc::from(read_spooled_bytes(
                path, encryption, stage, session,
            )?)),
        }
    }

    pub fn into_arc(self, stage: &'static str, session: &SessionContext) -> Result<Arc<[u8]>> {
        match self {
            Self::InMemory(bytes) => Ok(bytes),
            Self::Spooled {
                path,
                encryption,
                lifecycle,
            } => {
                let bytes = read_spooled_bytes(&path, &encryption, stage, session)?;
                if lifecycle.delete_on_materialize {
                    delete_spooled_file(
                        &path,
                        lifecycle.secure_delete_on_materialize,
                        stage,
                        session,
                    )?;
                }
                Ok(Arc::from(bytes))
            }
        }
    }

    pub fn dispose(self, stage: &'static str, session: &SessionContext) -> Result<()> {
        match self {
            Self::InMemory(_) => Ok(()),
            Self::Spooled {
                path, lifecycle, ..
            } => {
                if lifecycle.delete_on_materialize {
                    delete_spooled_file(
                        &path,
                        lifecycle.secure_delete_on_materialize,
                        stage,
                        session,
                    )?;
                }
                Ok(())
            }
        }
    }
}

fn read_spooled_bytes(
    path: &Path,
    encryption: &SpoolEncryption,
    stage: &'static str,
    session: &SessionContext,
) -> Result<Vec<u8>> {
    let bytes = std::fs::read(path).map_err(|err| {
        AsxError::new(
            ErrorCode::TransportFailure,
            format!("failed to read spooled body {}: {err}", path.display()),
            ErrorContext::for_session(stage, session),
        )
    })?;

    match encryption {
        SpoolEncryption::Plaintext => Ok(bytes),
        SpoolEncryption::Aes256Gcm { key } => {
            let min_len = SPOOLED_AES256_GCM_MAGIC.len()
                + SPOOLED_AES256_GCM_NONCE_LEN
                + SPOOLED_AES256_GCM_TAG_LEN;
            if bytes.len() < min_len {
                return Err(AsxError::new(
                    ErrorCode::DecryptionFailed,
                    format!(
                        "spooled encrypted body {} is too short for AES-GCM envelope",
                        path.display()
                    ),
                    ErrorContext::for_session(stage, session),
                ));
            }

            let magic = &bytes[..SPOOLED_AES256_GCM_MAGIC.len()];
            if magic != SPOOLED_AES256_GCM_MAGIC {
                return Err(AsxError::new(
                    ErrorCode::DecryptionFailed,
                    format!(
                        "spooled encrypted body {} has invalid envelope magic",
                        path.display()
                    ),
                    ErrorContext::for_session(stage, session),
                ));
            }

            let nonce_start = SPOOLED_AES256_GCM_MAGIC.len();
            let nonce_end = nonce_start + SPOOLED_AES256_GCM_NONCE_LEN;
            let tag_start = bytes.len() - SPOOLED_AES256_GCM_TAG_LEN;
            let nonce = &bytes[nonce_start..nonce_end];
            let ciphertext = &bytes[nonce_end..tag_start];
            let tag = &bytes[tag_start..];

            decrypt_spooled_aes256_gcm(path, key.as_ref(), nonce, ciphertext, tag, stage, session)
        }
    }
}

#[cfg(any(feature = "as2", feature = "as4", feature = "async-ocsp"))]
fn decrypt_spooled_aes256_gcm(
    path: &Path,
    key: &[u8],
    nonce: &[u8],
    ciphertext: &[u8],
    tag: &[u8],
    stage: &'static str,
    session: &SessionContext,
) -> Result<Vec<u8>> {
    openssl::symm::decrypt_aead(
        openssl::symm::Cipher::aes_256_gcm(),
        key,
        Some(nonce),
        &[],
        ciphertext,
        tag,
    )
    .map_err(|err| {
        AsxError::new(
            ErrorCode::DecryptionFailed,
            format!(
                "failed to decrypt spooled encrypted body {}: {err}",
                path.display()
            ),
            ErrorContext::for_session(stage, session),
        )
    })
}

#[cfg(not(any(feature = "as2", feature = "as4", feature = "async-ocsp")))]
fn decrypt_spooled_aes256_gcm(
    path: &Path,
    _key: &[u8],
    _nonce: &[u8],
    _ciphertext: &[u8],
    _tag: &[u8],
    stage: &'static str,
    session: &SessionContext,
) -> Result<Vec<u8>> {
    Err(AsxError::new(
        ErrorCode::PolicyViolation,
        format!(
            "spool AES-256-GCM decryption unavailable for {} without crypto protocol features",
            path.display()
        ),
        ErrorContext::for_session(stage, session),
    ))
}

fn delete_spooled_file(
    path: &Path,
    secure_delete: bool,
    stage: &'static str,
    session: &SessionContext,
) -> Result<()> {
    if secure_delete {
        use std::io::{Seek, SeekFrom, Write};

        let mut file = std::fs::OpenOptions::new()
            .read(true)
            .write(true)
            .open(path)
            .map_err(|err| {
                AsxError::new(
                    ErrorCode::TransportFailure,
                    format!(
                        "failed to open spooled file {} for secure delete: {err}",
                        path.display()
                    ),
                    ErrorContext::for_session(stage, session),
                )
            })?;
        let file_len = file
            .metadata()
            .map_err(|err| {
                AsxError::new(
                    ErrorCode::TransportFailure,
                    format!(
                        "failed to stat spooled file {} for secure delete: {err}",
                        path.display()
                    ),
                    ErrorContext::for_session(stage, session),
                )
            })?
            .len();

        file.seek(SeekFrom::Start(0)).map_err(|err| {
            AsxError::new(
                ErrorCode::TransportFailure,
                format!(
                    "failed to seek spooled file {} for secure delete: {err}",
                    path.display()
                ),
                ErrorContext::for_session(stage, session),
            )
        })?;

        let zeroes = vec![0u8; 8192];
        let mut remaining = file_len;
        while remaining > 0 {
            let write_len =
                usize::try_from(remaining.min(zeroes.len() as u64)).unwrap_or(zeroes.len());
            file.write_all(&zeroes[..write_len]).map_err(|err| {
                AsxError::new(
                    ErrorCode::TransportFailure,
                    format!(
                        "failed to overwrite spooled file {} for secure delete: {err}",
                        path.display()
                    ),
                    ErrorContext::for_session(stage, session),
                )
            })?;
            remaining -= write_len as u64;
        }
        file.flush().map_err(|err| {
            AsxError::new(
                ErrorCode::TransportFailure,
                format!(
                    "failed to flush overwritten spooled file {}: {err}",
                    path.display()
                ),
                ErrorContext::for_session(stage, session),
            )
        })?;
        file.sync_all().map_err(|err| {
            AsxError::new(
                ErrorCode::TransportFailure,
                format!(
                    "failed to sync overwritten spooled file {}: {err}",
                    path.display()
                ),
                ErrorContext::for_session(stage, session),
            )
        })?;
    }

    std::fs::remove_file(path).map_err(|err| {
        AsxError::new(
            ErrorCode::TransportFailure,
            format!("failed to remove spooled file {}: {err}", path.display()),
            ErrorContext::for_session(stage, session),
        )
    })
}

impl SessionContext {
    /// Start building a [`SessionContext`] incrementally.
    pub fn builder(
        session_id: impl Into<String>,
        partner_id: impl Into<String>,
    ) -> SessionContextBuilder {
        SessionContextBuilder::new(session_id, partner_id)
    }

    pub fn new(
        session_id: impl Into<String>,
        partner_id: impl Into<String>,
        profile_name: impl Into<String>,
    ) -> Result<Self> {
        let session_id = session_id.into();
        let partner_id = partner_id.into();
        let profile_name = profile_name.into();

        if session_id.trim().is_empty() {
            return Err(AsxError::new(
                ErrorCode::InvalidInput,
                "session_id must not be empty",
                ErrorContext::new("session_context_init"),
            ));
        }
        if partner_id.trim().is_empty() {
            return Err(AsxError::new(
                ErrorCode::InvalidInput,
                "partner_id must not be empty",
                ErrorContext::new("session_context_init").with_session_id(&session_id),
            ));
        }
        if profile_name.trim().is_empty() {
            return Err(AsxError::new(
                ErrorCode::InvalidInput,
                "profile_name must not be empty",
                ErrorContext::new("session_context_init")
                    .with_session_and_partner(&session_id, &partner_id),
            ));
        }

        Ok(Self {
            metadata: SessionMetadata::default(),
            cert_handle: Arc::new(CertHandle::new(format!("cert:{partner_id}"))),
            correlation_scope: CorrelationScope {
                root_id: format!("corr:{session_id}"),
                parent_message_id: None,
                traceparent: None,
            },
            session_id,
            partner_id,
            profile_name,
        })
    }

    /// Validate and set the session's certificate/trust configuration (builder method).
    ///
    /// This is the primary way to attach certificate material when constructing a
    /// session.  For rotation on an already-live session, prefer
    /// [`rotate_cert_handle`] which takes `&mut self`.
    ///
    /// The `trust_anchors_cache` is always reset so that a handle constructed via
    /// struct update syntax (`..old_handle`) does not carry a stale parsed-anchor
    /// cache from its origin.
    ///
    /// [`rotate_cert_handle`]: Self::rotate_cert_handle
    pub fn with_cert_handle(mut self, cert_handle: CertHandle) -> Result<Self> {
        Self::validate_cert_handle_fields(&cert_handle, "session_context_cert_update", &self)?;
        self.cert_handle = Arc::new(CertHandle {
            #[cfg(any(feature = "as2", feature = "as4"))]
            trust_anchors_cache: TrustAnchorCache::default(),
            ..cert_handle
        });
        Ok(self)
    }

    /// Atomically rotate the certificate/trust configuration on a live session.
    ///
    /// Unlike [`with_cert_handle`], this takes `&mut self` and therefore works
    /// on an already-constructed session without rebuilding it.  The `session_id`,
    /// `partner_id`, profile, reliability queues, and event subscriptions are
    /// preserved.
    ///
    /// Any in-flight verification calls that were already dispatched continue
    /// using the previous `Arc<CertHandle>` snapshot until they complete; newly
    /// accepted messages see the updated configuration immediately.
    ///
    /// # Errors
    ///
    /// Returns `InvalidInput` if `cert_handle` fails the same validation as
    /// [`with_cert_handle`].
    ///
    /// [`with_cert_handle`]: Self::with_cert_handle
    pub fn rotate_cert_handle(&mut self, cert_handle: CertHandle) -> Result<()> {
        Self::validate_cert_handle_fields(&cert_handle, "session_context_cert_rotate", self)?;
        self.cert_handle = Arc::new(CertHandle {
            #[cfg(any(feature = "as2", feature = "as4"))]
            trust_anchors_cache: TrustAnchorCache::default(),
            ..cert_handle
        });
        Ok(())
    }

    fn validate_cert_handle_fields(
        cert_handle: &CertHandle,
        stage: &'static str,
        session: &SessionContext,
    ) -> Result<()> {
        if cert_handle.key_id.trim().is_empty() {
            return Err(AsxError::new(
                ErrorCode::InvalidInput,
                "cert handle key_id must not be empty",
                ErrorContext::for_session(stage, session),
            ));
        }
        if cert_handle
            .trust_anchor_pems
            .iter()
            .any(|pem| pem.trim().is_empty())
            || cert_handle
                .revocation_crl_pems
                .iter()
                .any(|pem| pem.trim().is_empty())
            || cert_handle
                .stapled_ocsp_responses_der
                .iter()
                .any(Vec::is_empty)
            || cert_handle
                .responder_ocsp_responses_der
                .iter()
                .any(Vec::is_empty)
        {
            return Err(AsxError::new(
                ErrorCode::InvalidInput,
                "cert handle PKIX/OCSP material must not contain empty entries",
                ErrorContext::for_session(stage, session),
            ));
        }
        Ok(())
    }

    pub fn with_effective_policy_snapshot_json(
        mut self,
        snapshot_json: impl Into<String>,
    ) -> Result<Self> {
        let snapshot_json = snapshot_json.into();
        if snapshot_json.trim().is_empty() {
            return Err(AsxError::new(
                ErrorCode::InvalidInput,
                "effective policy snapshot JSON must not be empty",
                ErrorContext::for_session("session_context_metadata_update", &self),
            ));
        }
        self.metadata.effective_policy_snapshot_json = Some(snapshot_json);
        Ok(self)
    }

    pub fn effective_policy_snapshot_json(&self) -> Option<&str> {
        self.metadata.effective_policy_snapshot_json.as_deref()
    }

    /// Return whether this session is explicitly marked as startup-validated
    /// for strict-runtime protocol entry point enforcement.
    pub fn strict_runtime_bootstrap_validated(&self) -> bool {
        self.metadata.strict_runtime_bootstrap_validated
    }

    /// Return a cloned session marked with strict-runtime bootstrap validation state.
    pub fn with_strict_runtime_bootstrap_validated(mut self, validated: bool) -> Self {
        self.metadata.strict_runtime_bootstrap_validated = validated;
        self
    }

    /// Set strict-runtime bootstrap validation state on an existing session.
    pub fn set_strict_runtime_bootstrap_validated(&mut self, validated: bool) {
        self.metadata.strict_runtime_bootstrap_validated = validated;
    }

    pub fn session_id(&self) -> &str {
        &self.session_id
    }

    pub fn partner_id(&self) -> &str {
        &self.partner_id
    }

    pub fn profile_name(&self) -> &str {
        &self.profile_name
    }

    pub fn cert_handle(&self) -> &CertHandle {
        self.cert_handle.as_ref()
    }

    pub fn correlation_scope(&self) -> &CorrelationScope {
        &self.correlation_scope
    }

    /// Attach an inbound W3C Trace Context `traceparent` header value to this
    /// session so that every [`ScopedAsxEvent`] emitted during processing
    /// carries the upstream trace identifier.
    ///
    /// Typically called by the embedder's HTTP handler after parsing the
    /// inbound request with [`parse_as2_ingress_request`] or
    /// [`parse_as4_ingress_request`]:
    ///
    /// ```rust,ignore
    /// let ingress = parse_as4_ingress_request(&http_request)?;
    /// let session = SessionContext::new("s1", "partner", "strict")?
    ///     .with_incoming_traceparent(ingress.traceparent.as_deref());
    /// ```
    ///
    /// Passing `None` is a no-op (leaves any previously set value unchanged
    /// because inbound absence of the header should not clear a manually set
    /// value).
    ///
    /// [`ScopedAsxEvent`]: crate::observability::ScopedAsxEvent
    /// [`parse_as2_ingress_request`]: crate::transport::ingress::parse_as2_ingress_request
    /// [`parse_as4_ingress_request`]: crate::transport::ingress::parse_as4_ingress_request
    pub fn with_incoming_traceparent(mut self, traceparent: Option<&str>) -> Self {
        if let Some(tp) = traceparent {
            self.correlation_scope.traceparent = Some(Arc::from(tp));
        }
        self
    }

    /// Construct a minimal `SessionContext` for unit tests.
    ///
    /// Uses `OcspMode::Disabled` and soft-fail to avoid network I/O in tests.
    /// Prefer this over manually building `SessionContext` with `new()` in test code.
    #[cfg(any(test, feature = "testing"))]
    pub fn for_testing(session_id: impl Into<String>, partner_id: impl Into<String>) -> Self {
        let session_id = session_id.into();
        let partner_id = partner_id.into();
        Self {
            metadata: SessionMetadata::default(),
            cert_handle: Arc::new(CertHandle {
                ocsp_mode: OcspMode::Disabled,
                ocsp_failure_mode: OcspFailureMode::SoftFail,
                ..CertHandle::new(format!("cert:{partner_id}"))
            }),
            correlation_scope: CorrelationScope {
                root_id: format!("corr:{session_id}"),
                parent_message_id: None,
                traceparent: None,
            },
            session_id,
            partner_id,
            profile_name: "test".into(),
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn escape_xml_prevents_injection() {
        // Test ampersand
        assert_eq!(escape_xml("A&B"), "A&amp;B");
        // Test less-than
        assert_eq!(escape_xml("A<B"), "A&lt;B");
        // Test greater-than
        assert_eq!(escape_xml("A>B"), "A&gt;B");
        // Test double quote
        assert_eq!(escape_xml("A\"B"), "A&quot;B");
        // Test combined injection attempt
        assert_eq!(
            escape_xml("msg<inject>B&C\"D"),
            "msg&lt;inject&gt;B&amp;C&quot;D"
        );
        // Test empty string
        assert_eq!(escape_xml(""), "");
        // Test string with no special chars
        assert_eq!(escape_xml("hello-world"), "hello-world");
        // XML 1.0 §2.2: NUL and other forbidden control chars must be stripped.
        assert_eq!(escape_xml("ab\x00cd"), "abcd");
        assert_eq!(escape_xml("\x01\x08\x0B\x0C\x0E\x1F\x7F"), "");
        // NUL inside markup-requiring content
        assert_eq!(escape_xml("a\x00<b\x00>"), "a&lt;b&gt;");
    }

    #[test]
    fn error_code_strings_are_stable() {
        assert_eq!(ErrorCode::TransportFailure.as_str(), "transport_failure");
        assert_eq!(ErrorCode::InteropViolation.as_str(), "interop_violation");
    }

    #[test]
    fn session_context_validation_rejects_empty_values() {
        assert!(SessionContext::new("", "p", "strict").is_err());
        assert!(SessionContext::new("s", "", "strict").is_err());
        assert!(SessionContext::new("s", "p", "").is_err());
    }

    #[test]
    fn session_context_has_deterministic_default_handles() {
        let session = SessionContext::new("s1", "partner-a", "strict").expect("session");
        assert_eq!(session.cert_handle().key_id, "cert:partner-a");
        assert_eq!(session.correlation_scope().root_id, "corr:s1");
        assert!(session.effective_policy_snapshot_json().is_none());
    }

    #[test]
    fn session_context_builder_supports_incremental_configuration() {
        let cert = CertHandle {
            trust_anchor_pems: vec!["anchor-pem".into()],
            ..CertHandle::new("partner-key")
        };

        let session = SessionContext::builder("s-builder", "partner-z")
            .profile_name("peppol")
            .cert_handle(cert)
            .effective_policy_snapshot_json("{\"mode\":\"Strict\"}")
            .correlation_scope("corr-custom", Some("parent-1".into()))
            .build()
            .expect("builder session");

        assert_eq!(session.session_id(), "s-builder");
        assert_eq!(session.partner_id(), "partner-z");
        assert_eq!(session.profile_name(), "peppol");
        assert_eq!(session.cert_handle().key_id, "partner-key");
        assert_eq!(session.correlation_scope().root_id, "corr-custom");
        assert_eq!(
            session.correlation_scope().parent_message_id.as_deref(),
            Some("parent-1")
        );
        assert_eq!(
            session.effective_policy_snapshot_json(),
            Some("{\"mode\":\"Strict\"}")
        );
    }

    #[test]
    fn session_context_builder_rejects_blank_correlation_root() {
        let err = SessionContext::builder("s-builder", "partner-z")
            .correlation_scope("  ", None)
            .build()
            .expect_err("must reject blank correlation root");
        assert_eq!(err.code, ErrorCode::InvalidInput);
    }

    #[test]
    fn session_context_metadata_attaches_snapshot_json() {
        let session = SessionContext::new("s1", "partner-a", "strict")
            .expect("session")
            .with_effective_policy_snapshot_json("{\"resolved_mode\":\"Strict\"}")
            .expect("snapshot json");

        assert_eq!(
            session.effective_policy_snapshot_json(),
            Some("{\"resolved_mode\":\"Strict\"}")
        );
    }

    #[test]
    fn session_context_metadata_rejects_empty_snapshot_json() {
        let err = SessionContext::new("s1", "partner-a", "strict")
            .expect("session")
            .with_effective_policy_snapshot_json("   ")
            .expect_err("must reject blank snapshot");
        assert_eq!(err.code, ErrorCode::InvalidInput);
    }

    #[test]
    fn cert_handle_new_has_expected_defaults() {
        let h = CertHandle::new("my-key");
        assert_eq!(h.key_id, "my-key");
        assert!(h.trust_anchor_pems.is_empty());
        assert_eq!(h.ocsp_mode, OcspMode::ResponderOnly);
        assert_eq!(h.ocsp_failure_mode, OcspFailureMode::HardFail);
    }

    #[test]
    fn cert_handle_struct_update_syntax_works() {
        let base = CertHandle::new("base-key");
        let updated = CertHandle {
            trust_anchor_pems: vec!["fake-pem".into()],
            ocsp_mode: OcspMode::Disabled,
            ..base
        };
        assert_eq!(updated.key_id, "base-key");
        assert_eq!(updated.trust_anchor_pems, vec!["fake-pem".to_string()]);
        assert_eq!(updated.ocsp_mode, OcspMode::Disabled);
    }

    #[test]
    fn rotate_cert_handle_preserves_session_identity() {
        let mut session =
            SessionContext::new("rotate-session", "partner-b", "strict").expect("session");
        let original_session_id = session.session_id().to_string();
        let original_partner_id = session.partner_id().to_string();
        let original_root_id = session.correlation_scope().root_id.clone();

        let new_cert = CertHandle {
            trust_anchor_pems: vec!["new-anchor-pem".into()],
            ..CertHandle::new("new-key")
        };
        session.rotate_cert_handle(new_cert).expect("rotate");

        assert_eq!(session.session_id(), original_session_id);
        assert_eq!(session.partner_id(), original_partner_id);
        assert_eq!(session.correlation_scope().root_id, original_root_id);
        assert_eq!(session.cert_handle().key_id, "new-key");
        assert_eq!(
            session.cert_handle().trust_anchor_pems,
            vec!["new-anchor-pem".to_string()]
        );
    }

    #[test]
    fn rotate_cert_handle_rejects_empty_key_id() {
        let mut session = SessionContext::new("s1", "p1", "strict").expect("session");
        let bad_cert = CertHandle::new("");
        assert!(session.rotate_cert_handle(bad_cert).is_err());
    }

    #[test]
    fn arc_cert_handle_clone_shares_same_pointer() {
        let session = SessionContext::new("s1", "p1", "strict").expect("session");
        let clone = session.clone();
        // Both sessions share the same Arc<CertHandle> pointer — O(1) clone.
        assert!(Arc::ptr_eq(&session.cert_handle, &clone.cert_handle));
    }

    #[test]
    #[cfg(any(feature = "as2", feature = "as4"))]
    fn with_cert_handle_resets_trust_anchor_cache() {
        // Verify that cache is NOT shared between different CertHandles
        // even when constructed via struct update syntax.
        let old_cert = CertHandle::new("old-key");
        // Force the cache to be populated on old_cert (simulate a warm cache).
        // We can't easily populate it here since trust_anchors_x509 needs real PEM,
        // but we can verify that a new CertHandle::new() gives a fresh cache Arc.
        let new_cert = CertHandle {
            trust_anchor_pems: vec!["some-pem".into()],
            ..CertHandle::new("new-key")
        };
        // The cache Arcs should be different instances.
        assert!(!Arc::ptr_eq(
            &old_cert.trust_anchors_cache.0,
            &new_cert.trust_anchors_cache.0
        ));
    }
}
