spiffe 0.19.0

use super::builder::{ReconnectConfig, ResourceLimits};
use super::errors::{MetricsErrorKind, X509SourceError};
use super::limits::{select_svid, validate_limits_and_record_metric};
use super::metrics::MetricsRecorder;
use super::supervisor::initial_sync_with_retry;
use crate::bundle::BundleSource;
use crate::cert::Certificate;
use crate::prelude::warn;
use crate::svid::SvidSource;
use crate::workload_api::supervisor_common::{SourceOwner, SupervisorTask};
use crate::workload_api::x509_context::X509Context;
use crate::x509_source::types::{ClientFactory, SvidPicker};
use crate::{TrustDomain, X509Bundle, X509BundleSet, X509SourceBuilder, X509Svid};
use arc_swap::ArcSwap;
use std::cmp::Ordering as CmpOrdering;
use std::fmt::Debug;
use std::future::Future;
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::sync::Arc;
use std::sync::OnceLock;
use std::time::{Duration, SystemTime, UNIX_EPOCH};
use tokio::sync::watch;
use tokio_util::sync::CancellationToken;

use crate::WorkloadApiError;

/// Handle for receiving update notifications from an [`X509Source`].
///
/// This type wraps the underlying notification mechanism and provides a clean
/// API for detecting when the X.509 context has been updated.
///
/// Cloning this handle creates another receiver that shares the same update
/// stream. Each receiver observes the latest sequence number; if a receiver
/// is slow to consume updates, intermediate sequence numbers may be skipped
/// (this is the standard behavior of `watch` channels).
///
/// # Examples
///
/// ```no_run
/// # use spiffe::X509Source;
/// # async fn example() -> Result<(), Box<dyn std::error::Error>> {
/// let source = X509Source::new().await?;
/// let mut updates = source.updated();
///
/// // Wait for an update
/// updates.changed().await?;
/// println!("Update sequence: {}", updates.last());
///
/// // Wait for another update
/// updates.changed().await?;
/// println!("New sequence: {}", updates.last());
/// # Ok(())
/// # }
/// ```
#[derive(Clone, Debug)]
pub struct X509SourceUpdates {
    rx: watch::Receiver<u64>,
    shutdown: CancellationToken,
}

impl X509SourceUpdates {
    /// Waits for the next update and returns the new sequence number.
    ///
    /// This method waits for the next rotation after initial synchronization.
    /// The initial sync does not trigger a notification; only subsequent updates
    /// are notified.
    ///
    /// This method will return an error if the source has been closed or
    /// the internal update task has terminated (e.g., due to shutdown or
    /// an internal error).
    ///
    /// # Errors
    ///
    /// Returns [`X509SourceError::Closed`] if the source has been shut down
    /// or the internal update task has terminated. This can occur due to:
    /// - Explicit shutdown via [`X509Source::shutdown`] or [`X509Source::shutdown_with_timeout`]
    /// - Internal task termination (e.g., supervisor task panic)
    pub async fn changed(&mut self) -> Result<u64, X509SourceError> {
        if self.rx.has_changed().unwrap_or(false) {
            self.rx
                .changed()
                .await
                .map_err(|watch::error::RecvError { .. }| X509SourceError::Closed)?;
            return Ok(*self.rx.borrow());
        }

        if self.shutdown.is_cancelled() {
            return Err(X509SourceError::Closed);
        }

        tokio::select! {
            biased;
            result = self.rx.changed() => {
                result.map_err(|watch::error::RecvError { .. }| X509SourceError::Closed)?;
                Ok(*self.rx.borrow())
            }
            () = self.shutdown.cancelled() => Err(X509SourceError::Closed),
        }
    }

    /// Returns the last sequence number without waiting.
    ///
    /// This method never blocks and always returns immediately with the
    /// current sequence number.
    pub fn last(&self) -> u64 {
        *self.rx.borrow()
    }

    /// Waits for the sequence number to satisfy a predicate.
    ///
    /// This method first checks whether the source is closed, then checks the
    /// current sequence number; if the predicate is already satisfied, it returns
    /// immediately without waiting. Otherwise, it repeatedly calls `changed()`
    /// until the predicate returns `true`.
    ///
    /// # Errors
    ///
    /// Returns an error if the source has been closed.
    pub async fn wait_for<F>(&mut self, mut f: F) -> Result<u64, X509SourceError>
    where
        F: FnMut(&u64) -> bool,
    {
        if self.shutdown.is_cancelled() {
            return Err(X509SourceError::Closed);
        }

        let current = self.last();
        if f(&current) {
            return Ok(current);
        }
        loop {
            let seq = self.changed().await?;
            if f(&seq) {
                return Ok(seq);
            }
        }
    }
}

/// Live source of X.509 SVIDs and bundles from the SPIFFE Workload API.
///
/// `X509Source` performs an initial sync before returning from [`X509Source::new`] or
/// [`X509SourceBuilder::build`]. Updates are applied atomically and can be observed via
/// [`X509Source::updated`].
///
/// The source automatically:
/// - Maintains a cached view of the latest X.509 materials
/// - Handles SVID and bundle rotation transparently
/// - Reconnects with exponential backoff on transient failures
/// - Validates resource limits before publishing updates
/// - Publishes valid complete trust even without a usable selected SVID
/// - Withdraws permission until a valid complete response reauthorizes its material RPC
///
/// Use [`X509Source::shutdown`] or [`X509Source::shutdown_configured`] to stop the supervisor.
///
#[derive(Clone, Debug)]
pub struct X509Source {
    inner: Arc<Inner>,
    _shutdown_guard: Arc<SourceOwner>,
}

/// One coherent authoritative X.509 observation, including final state after closure.
///
/// The revision changes on accepted material changes, withdrawal, and recovery. Closure
/// is separate: `context()` and `svid()` describe final permission, while source getters
/// additionally reject closure. Previously acquired credentials remain caller-owned.
#[derive(Clone, Debug)]
pub struct X509SourceState {
    ctx: Arc<X509Context>,
    selected: Option<Arc<X509Svid>>,
    expiry_unix: i64,
    revision: u64,
    withdrawal: Option<Arc<WorkloadApiError>>,
    closed: bool,
}

impl X509SourceState {
    /// Authoritative material revision (initial synchronization is revision zero).
    pub const fn revision(&self) -> u64 {
        self.revision
    }
    /// Whether refresh has permanently ended or shutdown was requested.
    pub const fn is_closed(&self) -> bool {
        self.closed
    }
    /// Permission error that withdrew this material RPC, if any.
    pub const fn withdrawal(&self) -> Option<&Arc<WorkloadApiError>> {
        self.withdrawal.as_ref()
    }
    /// Current permitted context, independent of local identity selectability and closure.
    ///
    /// # Errors
    /// Returns `Withdrawn` until a valid complete response reauthorizes the RPC.
    pub fn context(&self) -> Result<Arc<X509Context>, X509SourceError> {
        self.assert_permitted()?;
        Ok(Arc::clone(&self.ctx))
    }
    /// Picker-selected identity, if permitted and unexpired at acquisition time.
    ///
    /// # Errors
    /// Returns `Withdrawn` or `NoSuitableSvid`. An unavailable wall clock fails closed.
    pub fn svid(&self) -> Result<Arc<X509Svid>, X509SourceError> {
        self.svid_at(unix_timestamp_now())
    }
    fn svid_at(&self, now: Option<i64>) -> Result<Arc<X509Svid>, X509SourceError> {
        self.assert_permitted()?;
        if now.is_none_or(|now| self.expiry_unix <= now) {
            return Err(X509SourceError::NoSuitableSvid);
        }
        self.selected.clone().ok_or(X509SourceError::NoSuitableSvid)
    }
    /// Whether this observation currently permits a usable local identity.
    pub fn is_usable(&self) -> bool {
        self.svid().is_ok()
    }
    fn assert_permitted(&self) -> Result<(), X509SourceError> {
        if let Some(reason) = &self.withdrawal {
            return Err(X509SourceError::Withdrawn(Arc::clone(reason)));
        }
        Ok(())
    }
}

type Snapshot = X509SourceState;

pub(super) struct Inner {
    // Atomically replaced context, cached selection, permission, and revision.
    snapshot: ArcSwap<Snapshot>,

    // Policy for selecting an SVID when multiple are present.
    svid_picker: Option<Box<dyn SvidPicker>>,
    limits: ResourceLimits,

    // Supervisor configuration and dependencies.
    reconnect: ReconnectConfig,
    make_client: ClientFactory,
    metrics: Option<Arc<dyn MetricsRecorder>>,

    // Lifecycle / shutdown.
    closed: AtomicBool,
    supervisor_running: AtomicBool,
    cancel: CancellationToken,
    shutdown_timeout: Option<Duration>,

    // Update notifications (monotonic sequence).
    update_seq: AtomicU64,
    update_tx: watch::Sender<u64>,

    // Supervisor task handle (joined/aborted at shutdown).
    supervisor: OnceLock<SupervisorTask>,
}

impl Inner {
    pub(super) const fn reconnect(&self) -> ReconnectConfig {
        self.reconnect
    }
    pub(super) fn metrics(&self) -> Option<&dyn MetricsRecorder> {
        self.metrics.as_deref()
    }
    pub(super) fn make_client(&self) -> &ClientFactory {
        &self.make_client
    }
}

impl Debug for Inner {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("X509Source")
            .field("snapshot", &"<ArcSwap<Snapshot>>")
            .field(
                "svid_picker",
                &self.svid_picker.as_ref().map(|_| "<SvidPicker>"),
            )
            .field("reconnect", &self.reconnect)
            .field("limits", &self.limits)
            .field("make_client", &"<ClientFactory>")
            .field(
                "metrics",
                &self.metrics.as_ref().map(|_| "<MetricsRecorder>"),
            )
            .field("shutdown_timeout", &self.shutdown_timeout)
            .field("closed", &self.closed.load(Ordering::Relaxed))
            .field(
                "supervisor_running",
                &self.supervisor_running.load(Ordering::Relaxed),
            )
            .field("cancel", &self.cancel)
            .field("update_seq", &self.update_seq)
            .field("update_tx", &"<watch::Sender<u64>>")
            .field("supervisor", &"<OnceLock<SupervisorTask>>")
            .finish()
    }
}

impl X509Source {
    #[cfg(test)]
    pub(super) const fn inner_for_test(&self) -> &Arc<Inner> {
        &self.inner
    }

    /// Creates an `X509Source` using the default Workload API endpoint.
    ///
    /// The endpoint is resolved from `SPIFFE_ENDPOINT_SOCKET`. The source selects the default
    /// X.509 SVID when multiple SVIDs are available.
    ///
    /// On success, the returned source is already synchronized with the agent and will keep
    /// updating in the background until it is closed.
    ///
    /// # Errors
    ///
    /// Returns an [`X509SourceError`] if:
    /// - the Workload API endpoint cannot be resolved or connected to,
    /// - the initial synchronization with the Workload API does not complete successfully,
    /// - or no suitable X.509 SVID can be selected from the received context.
    pub async fn new() -> Result<Self, X509SourceError> {
        X509SourceBuilder::new().build().await
    }

    /// Creates a builder for configuring an [`X509Source`].
    ///
    /// The builder allows customizing how the source connects to the SPIFFE
    /// Workload API and how X.509 material is managed (e.g. endpoint selection,
    /// reconnection behavior, resource limits).
    ///
    /// # Examples
    ///
    /// ```no_run
    /// use spiffe::X509Source;
    ///
    /// # async fn example() -> Result<(), Box<dyn std::error::Error>> {
    /// let source = X509Source::builder()
    ///     .endpoint("unix:///tmp/spire-agent/public/api.sock")
    ///     .build()
    ///     .await?;
    ///
    /// # Ok(())
    /// # }
    /// ```
    pub fn builder() -> X509SourceBuilder {
        X509SourceBuilder::new()
    }

    /// Returns a handle for receiving update notifications.
    ///
    /// The handle yields a monotonically increasing sequence number on each
    /// authoritative material change, withdrawal, or recovery. This can be used to detect when
    /// the context has changed without polling.
    ///
    /// **Note:** The initial sequence number is 0. Notifications are sent when the selected
    /// SVID or the order-insensitive SVID/bundle material changes. Reordering non-selected
    /// SVIDs or bundle authorities does not notify. The initial sync does not trigger a
    /// notification, and neither do re-deliveries of unchanged material (e.g. on reconnect
    /// after a dropped stream or agent restart). A byte-identical valid response following
    /// withdrawal restores permission and advances the sequence.
    ///
    /// # Examples
    ///
    /// ```no_run
    /// # use spiffe::X509Source;
    /// # async fn example() -> Result<(), Box<dyn std::error::Error>> {
    /// let source = X509Source::new().await?;
    /// let mut updates = source.updated();
    /// // Wait for the first update notification
    /// updates.changed().await?;
    /// println!("Update sequence: {}", updates.last());
    /// # Ok(())
    /// # }
    /// ```
    pub fn updated(&self) -> X509SourceUpdates {
        X509SourceUpdates {
            rx: self.inner.update_tx.subscribe(),
            shutdown: self.inner.cancel.clone(),
        }
    }

    /// Returns `true` if the source appears healthy and can likely provide an SVID.
    ///
    /// This method checks that:
    /// - The source is not closed or cancelled
    /// - The update supervisor is still running
    /// - The most recently accepted update produced a selectable SVID
    /// - That selected SVID's leaf certificate `notAfter` is still in the future
    ///
    /// **Note:** This check is inherently racy. Between calling `is_healthy()` and
    /// `svid()`, the source may be shut down or the context may change. Use this
    /// for best-effort health checks (e.g., monitoring), not for synchronization.
    /// If you need a guaranteed check, call `svid()` directly and handle the error.
    /// `build()` / builder `build()` only indicates that initial sync completed successfully;
    /// `is_healthy()` is a runtime health signal, not a construction-success signal. There
    /// may be a brief scheduler-dependent window immediately after construction where this
    /// method returns `false` until the background supervisor task is first polled.
    ///
    /// If a [`MetricsRecorder`] or picker callback panics, the supervisor task terminates,
    /// update handles close, and this method reports `false` afterward. Rebuild the source
    /// to resume updates.
    ///
    /// # Examples
    ///
    /// ```no_run
    /// # use spiffe::X509Source;
    /// # async fn example() -> Result<(), Box<dyn std::error::Error>> {
    /// let source = X509Source::new().await?;
    ///
    /// if source.is_healthy() {
    ///     println!("Source appears healthy");
    /// } else {
    ///     println!("Source is unhealthy");
    /// }
    /// # Ok(())
    /// # }
    /// ```
    pub fn is_healthy(&self) -> bool {
        if self.inner.closed.load(Ordering::Acquire)
            || self.inner.cancel.is_cancelled()
            || !self.inner.supervisor_running.load(Ordering::Acquire)
        {
            return false;
        }

        let Some(now) = unix_timestamp_now() else {
            // If the wall clock is unavailable, health is reported conservatively as
            // unhealthy. Validation is more optimistic to avoid rejecting otherwise
            // usable initial sync data.
            return false;
        };

        {
            let snapshot = self.inner.snapshot.load();
            snapshot.withdrawal.is_none()
                && snapshot.selected.is_some()
                && snapshot.expiry_unix > now
        }
    }

    /// Returns the current X.509 context (SVID + bundles) as a single value.
    ///
    /// # Errors
    ///
    /// Returns an [`X509SourceError`] if the X.509 context is not available or
    /// cannot be constructed.
    pub fn x509_context(&self) -> Result<Arc<X509Context>, X509SourceError> {
        self.assert_open()?;
        self.state().context()
    }

    /// Acquires one coherent security view. The atomic snapshot load is the material
    /// acquisition linearization point. This accessor also works after terminal closure
    /// so consumers can apply an explicit final-material policy without resurrecting trust.
    /// The returned observation is immutable; reacquire it for current authorization.
    pub fn state(&self) -> X509SourceState {
        let mut state = self.inner.snapshot.load_full().as_ref().clone();
        state.closed =
            self.inner.closed.load(Ordering::Acquire) || self.inner.cancel.is_cancelled();
        state
    }

    /// Returns the current X.509 SVID selected by the picker (or default).
    ///
    /// # Errors
    ///
    /// Returns [`X509SourceError`] if the source is closed or no SVID is available.
    pub fn svid(&self) -> Result<Arc<X509Svid>, X509SourceError> {
        self.assert_open()?;
        self.state().svid()
    }

    /// Returns the current SVID, or `None` if unavailable.
    ///
    /// Returns `None` instead of an error
    /// when the SVID cannot be retrieved. Use this when `None` is an acceptable
    /// value for your use case.
    ///
    /// **Note:** This method swallows all errors, including `Closed`. If you need
    /// to detect shutdown, use [`X509Source::svid`] instead.
    ///
    /// # Examples
    ///
    /// ```no_run
    /// # use spiffe::X509Source;
    /// # async fn example() -> Result<(), Box<dyn std::error::Error>> {
    /// let source = X509Source::new().await?;
    ///
    /// if let Some(svid) = source.try_svid() {
    ///     println!("Got SVID: {}", svid.spiffe_id());
    /// } else {
    ///     println!("No SVID available");
    /// }
    /// # Ok(())
    /// # }
    /// ```
    pub fn try_svid(&self) -> Option<Arc<X509Svid>> {
        self.svid().ok()
    }

    /// Returns the current X.509 bundle set.
    ///
    /// # Errors
    ///
    /// Returns [`X509SourceError`] if the source is closed.
    pub fn bundle_set(&self) -> Result<Arc<X509BundleSet>, X509SourceError> {
        self.assert_open()?;
        Ok(Arc::clone(self.state().context()?.bundle_set()))
    }

    /// Returns the current bundle for the trust domain, or `None` if unavailable.
    ///
    /// Returns `None` instead of an error
    /// when the bundle cannot be retrieved. Use this when `None` is an acceptable
    /// value for your use case.
    ///
    /// **Note:** This method swallows all errors, including `Closed`. If you need
    /// to detect shutdown, use [`X509Source::bundle_for_trust_domain`] instead.
    ///
    /// # Examples
    ///
    /// ```no_run
    /// # use spiffe::{TrustDomain, X509Source};
    /// # async fn example() -> Result<(), Box<dyn std::error::Error>> {
    /// let source = X509Source::new().await?;
    /// let trust_domain = TrustDomain::new("example.org")?;
    ///
    /// if let Some(bundle) = source.try_bundle_for_trust_domain(&trust_domain) {
    ///     println!("Got bundle for {}", trust_domain);
    /// } else {
    ///     println!("No bundle available for {}", trust_domain);
    /// }
    /// # Ok(())
    /// # }
    /// ```
    pub fn try_bundle_for_trust_domain(&self, td: &TrustDomain) -> Option<Arc<X509Bundle>> {
        self.bundle_for_trust_domain(td).ok().flatten()
    }

    /// Requests cancellation and waits for the source supervisor to terminate.
    ///
    /// The first poll closes admission and requests cancellation. Dropping this
    /// shutdown future preserves that request and the source's join ownership.
    /// Concurrent and repeated callers all observe actual supervisor completion.
    ///
    /// This can wait indefinitely if cleanup cannot finish. For a graceful
    /// deadline followed by abort and join, use [`X509Source::shutdown_with_timeout`]
    /// or [`X509Source::shutdown_configured`].
    ///
    /// Completion covers the source supervisor, including destruction of its
    /// directly owned futures. It does not join dependency-owned transport workers.
    pub async fn shutdown(&self) {
        self.join_supervisor(None).await;
    }

    async fn join_supervisor(&self, timeout: Option<Duration>) -> bool {
        self.inner.closed.store(true, Ordering::Release);
        self.inner.cancel.cancel();
        // Production builds install the task synchronously before returning a source.
        // Test-only sources may intentionally have no supervisor.
        let Some(task) = self.inner.supervisor.get() else {
            return false;
        };
        let (outcome, timed_out) = task.join(timeout).await;
        if !timed_out && task.report_failure_once(outcome) {
            warn!("Error joining supervisor task during shutdown: {outcome:?}");
            self.inner
                .record_error(MetricsErrorKind::SupervisorJoinFailed);
        }
        if timed_out {
            warn!("Shutdown grace expired; supervisor aborted and joined");
        }
        timed_out
    }

    /// Requests cancellation, then aborts and joins after a graceful deadline.
    ///
    /// Each caller has its own `timeout`. When its grace period expires, it
    /// requests abort and continues waiting for actual supervisor completion.
    /// Another caller can observe that completion within its own grace period
    /// and return `Ok(())`. Repeated calls after completion return promptly.
    ///
    /// Dropping a shutdown future never loses join ownership. The grace period
    /// is not an absolute wall-clock bound: abort cannot preempt synchronous
    /// callbacks or destructors, and this method waits for them to finish.
    ///
    /// # Errors
    ///
    /// Returns [`X509SourceError::ShutdownTimeout`] after joining if this caller's
    /// grace period expired. Supervisor panic counts as termination under the
    /// existing error contract.
    ///
    /// # Examples
    ///
    /// ```no_run
    /// use spiffe::X509Source;
    /// use std::time::Duration;
    /// # async fn example() -> Result<(), Box<dyn std::error::Error>> {
    /// let source = X509Source::new().await?;
    /// // Shutdown with 10 second timeout (graceful, then abort if needed)
    /// source.shutdown_with_timeout(Duration::from_secs(10)).await?;
    /// # Ok(())
    /// # }
    /// ```
    pub async fn shutdown_with_timeout(&self, timeout: Duration) -> Result<(), X509SourceError> {
        if self.join_supervisor(Some(timeout)).await {
            Err(X509SourceError::ShutdownTimeout)
        } else {
            Ok(())
        }
    }

    /// Shuts down the supervisor using the builder's graceful deadline.
    ///
    /// The default is 30 seconds. With `None`, this waits indefinitely, like
    /// [`X509Source::shutdown`]. With `Some`, it has the same abort-then-join and
    /// cancellation guarantees as [`X509Source::shutdown_with_timeout`].
    ///
    /// # Errors
    ///
    /// Returns [`X509SourceError::ShutdownTimeout`] after joining if this caller's
    /// configured grace period expired.
    pub async fn shutdown_configured(&self) -> Result<(), X509SourceError> {
        if let Some(timeout) = self.inner.shutdown_timeout {
            self.shutdown_with_timeout(timeout).await
        } else {
            self.shutdown().await;
            Ok(())
        }
    }
}

impl X509Source {
    pub(super) async fn build_with(
        make_client: ClientFactory,
        svid_picker: Option<Box<dyn SvidPicker>>,
        reconnect: ReconnectConfig,
        limits: ResourceLimits,
        metrics: Option<Arc<dyn MetricsRecorder>>,
        shutdown_timeout: Option<Duration>,
        initial_sync_timeout: Option<Duration>,
    ) -> Result<Self, X509SourceError> {
        let reconnect = super::builder::normalize_reconnect(reconnect);

        let (update_tx, _update_rx) = watch::channel(0u64);
        let cancel = CancellationToken::new();
        let shutdown_guard = Arc::new(SourceOwner::new(cancel.clone()));

        let initial_sync = initial_sync_with_retry(
            &make_client,
            svid_picker.as_deref(),
            &cancel,
            reconnect,
            limits,
            metrics.as_deref(),
        );
        let (initial_ctx, selected_svid) =
            initial_sync_with_timeout(initial_sync, &cancel, initial_sync_timeout).await?;
        let snapshot = Arc::new(Snapshot {
            ctx: initial_ctx,
            expiry_unix: selected_svid.expiry_unix(),
            selected: Some(selected_svid),
            revision: 0,
            withdrawal: None,
            closed: false,
        });

        let inner = Arc::new(Inner {
            snapshot: ArcSwap::from(snapshot),
            svid_picker,
            reconnect,
            make_client,
            limits,
            metrics,
            shutdown_timeout,
            closed: AtomicBool::new(false),
            supervisor_running: AtomicBool::new(false),
            cancel,
            update_seq: AtomicU64::new(0),
            update_tx,
            supervisor: OnceLock::new(),
        });

        let task_inner = Arc::clone(&inner);
        let token = task_inner.cancel.clone();
        let guard_inner = Arc::clone(&task_inner);
        let terminate_on_drop = SupervisorTerminationGuard { inner: guard_inner };
        let handle = tokio::spawn(async move {
            let _terminate_on_drop = terminate_on_drop;
            task_inner.supervisor_running.store(true, Ordering::Release);
            task_inner.run_update_supervisor(token).await;
        });

        let task = SupervisorTask::new(handle);
        shutdown_guard.install_abort(task.abort_handle());
        let installed = inner.supervisor.set(task);
        debug_assert!(installed.is_ok(), "source supervisor installed only once");

        Ok(Self {
            inner,
            _shutdown_guard: shutdown_guard,
        })
    }

    /// Test-only constructor that creates an `X509Source` with a provided initial context
    /// without spawning the supervisor task or performing initial sync.
    ///
    /// This allows deterministic unit tests without requiring a real Workload API client.
    #[cfg(test)]
    pub(super) fn new_for_test(
        initial_ctx: Arc<X509Context>,
        reconnect: ReconnectConfig,
        limits: ResourceLimits,
        metrics: Option<Arc<dyn MetricsRecorder>>,
        svid_picker: Option<Box<dyn SvidPicker>>,
    ) -> Self {
        // Normalize reconnect config at the boundary (same as new_with)
        let reconnect = super::builder::normalize_reconnect(reconnect);

        let (update_tx, _update_rx) = watch::channel(0u64);
        let cancel = CancellationToken::new();
        let shutdown_guard = Arc::new(SourceOwner::new(cancel.clone()));

        let make_client: ClientFactory =
            Arc::new(|| Box::pin(async move { Err(WorkloadApiError::EmptyResponse) }));
        let selected = select_svid(&initial_ctx, svid_picker.as_deref());
        let selected_svid_expires_at_unix = selected.as_ref().map_or(0, |svid| svid.expiry_unix());
        let snapshot = Arc::new(Snapshot {
            selected,
            ctx: initial_ctx,
            expiry_unix: selected_svid_expires_at_unix,
            revision: 0,
            withdrawal: None,
            closed: false,
        });

        let inner = Inner {
            snapshot: ArcSwap::from(snapshot),
            svid_picker,
            reconnect,
            make_client,
            limits,
            metrics,
            shutdown_timeout: None,
            closed: AtomicBool::new(false),
            supervisor_running: AtomicBool::new(false),
            cancel,
            update_seq: AtomicU64::new(0),
            update_tx,
            supervisor: OnceLock::new(),
        };

        Self {
            inner: Arc::new(inner),
            _shutdown_guard: shutdown_guard,
        }
    }

    fn assert_open(&self) -> Result<(), X509SourceError> {
        if self.inner.closed.load(Ordering::Acquire) || self.inner.cancel.is_cancelled() {
            return Err(X509SourceError::Closed);
        }
        Ok(())
    }
}

struct SupervisorTerminationGuard {
    inner: Arc<Inner>,
}

#[cfg(test)]
impl SupervisorTerminationGuard {
    fn new(inner: Arc<Inner>) -> Self {
        inner.supervisor_running.store(true, Ordering::Release);
        Self { inner }
    }
}

impl Drop for SupervisorTerminationGuard {
    fn drop(&mut self) {
        self.inner
            .supervisor_running
            .store(false, Ordering::Release);
        self.inner.cancel.cancel();
    }
}

/// Outcome of a successfully validated update.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub(super) enum ApplyUpdateResult {
    /// The incoming context differed and was stored, notified, and recorded.
    Applied,
    /// The incoming context matched the held material; nothing was stored or notified.
    Unchanged,
}

fn unix_timestamp_now() -> Option<i64> {
    let duration = SystemTime::now().duration_since(UNIX_EPOCH).ok()?;
    i64::try_from(duration.as_secs()).ok()
}

impl Inner {
    pub(super) fn record_error(&self, kind: MetricsErrorKind) {
        if let Some(metrics) = self.metrics.as_deref() {
            metrics.record_error(kind);
        }
    }

    pub(super) fn record_update(&self) {
        if let Some(metrics) = self.metrics.as_deref() {
            metrics.record_update();
        }
    }

    pub(super) fn apply_update(
        &self,
        new_ctx: Arc<X509Context>,
    ) -> Result<ApplyUpdateResult, X509SourceError> {
        if let Err(e) =
            validate_limits_and_record_metric(&new_ctx, self.limits, self.metrics.as_deref())
        {
            self.record_error(MetricsErrorKind::UpdateRejected);
            return Err(e);
        }
        // Empty mandatory SVID lists are rejected; an empty optional federation is removal.
        if new_ctx.svids().is_empty() {
            self.record_error(MetricsErrorKind::UpdateRejected);
            return Err(X509SourceError::Source(WorkloadApiError::EmptyResponse));
        }
        let selected = select_svid(&new_ctx, self.svid_picker.as_deref());
        let current = self.snapshot.load_full();
        if current.withdrawal.is_none()
            && current.selected == selected
            && same_context_material(&current.ctx, &new_ctx)
        {
            return Ok(ApplyUpdateResult::Unchanged);
        }
        self.snapshot.store(Arc::new(Snapshot {
            expiry_unix: selected.as_ref().map_or(0, |svid| svid.expiry_unix()),
            ctx: new_ctx,
            selected,
            revision: current.revision + 1,
            withdrawal: None,
            closed: false,
        }));
        self.notify_update();
        self.record_update();
        Ok(ApplyUpdateResult::Applied)
    }

    pub(super) fn withdraw(&self, error: &WorkloadApiError) {
        let reason = match error {
            WorkloadApiError::PermissionDenied(message) => {
                WorkloadApiError::PermissionDenied(message.clone())
            }
            WorkloadApiError::NoIdentityIssued => WorkloadApiError::NoIdentityIssued,
            _ => return,
        };
        let current = self.snapshot.load_full();
        if current.withdrawal.is_some() {
            return;
        }
        let mut withdrawn = current.as_ref().clone();
        withdrawn.revision += 1;
        withdrawn.withdrawal = Some(Arc::new(reason));
        self.snapshot.store(Arc::new(withdrawn));
        self.notify_update();
    }

    pub(super) fn notify_update(&self) {
        let next = self.update_seq.fetch_add(1, Ordering::Relaxed) + 1;
        let _prev = self.update_tx.send_replace(next);
    }
}

/// Returns true when two contexts select the same SVID and carry the same SVID multiset and
/// bundle set.
///
/// Exercises the update equality rules in tests: re-delivery that only reshuffles
/// non-selected SVIDs or bundle authority order is unchanged. A reorder that changes
/// the default or picker-selected SVID remains material.
///
/// SVID equality includes the full certificate chain: intermediate differences are treated as
/// material changes because TLS presentation uses the complete chain.
#[cfg(test)]
fn same_material_for_update(
    current: &X509Context,
    incoming: &X509Context,
    picker: Option<&dyn SvidPicker>,
    incoming_selected: &X509Svid,
) -> bool {
    if !select_svid(current, picker).is_some_and(|selected| selected.as_ref() == incoming_selected)
    {
        return false;
    }
    same_context_material(current, incoming)
}

fn same_context_material(current: &X509Context, incoming: &X509Context) -> bool {
    if !bundle_set_equal_for_update(current.bundle_set(), incoming.bundle_set()) {
        return false;
    }
    if current.svids().len() != incoming.svids().len() {
        return false;
    }

    let mut left: Vec<&X509Svid> = current.svids().iter().map(AsRef::as_ref).collect();
    let mut right: Vec<&X509Svid> = incoming.svids().iter().map(AsRef::as_ref).collect();
    left.sort_unstable_by(|a, b| cmp_svid_for_update_dedupe(a, b));
    right.sort_unstable_by(|a, b| cmp_svid_for_update_dedupe(a, b));
    left == right
}

fn bundle_set_equal_for_update(a: &X509BundleSet, b: &X509BundleSet) -> bool {
    if a.len() != b.len() {
        return false;
    }

    a.iter().all(|(trust_domain, bundle)| {
        b.get(trust_domain)
            .is_some_and(|other| bundle_equal_for_update(bundle.as_ref(), other.as_ref()))
    })
}

fn bundle_equal_for_update(a: &X509Bundle, b: &X509Bundle) -> bool {
    a.trust_domain() == b.trust_domain()
        && authority_set_equal_for_update(a.authorities(), b.authorities())
}

fn authority_set_equal_for_update(a: &[Certificate], b: &[Certificate]) -> bool {
    if a.len() != b.len() {
        return false;
    }

    let mut left: Vec<&[u8]> = a.iter().map(Certificate::as_bytes).collect();
    let mut right: Vec<&[u8]> = b.iter().map(Certificate::as_bytes).collect();
    left.sort_unstable();
    right.sort_unstable();
    left == right
}

/// Total order over SVIDs covering every field that [`X509Svid`] equality compares.
///
/// `cert_chain()` starts with the leaf, so comparing it also discriminates leaf differences,
/// and `expiry_unix` is derived from the leaf. Ordering must agree with equality: SVIDs that
/// compare `Equal` here are interchangeable when the sorted lists are compared element-wise.
fn cmp_svid_for_update_dedupe(a: &X509Svid, b: &X509Svid) -> CmpOrdering {
    a.spiffe_id()
        .cmp(b.spiffe_id())
        .then_with(|| a.hint().cmp(&b.hint()))
        .then_with(|| cmp_cert_chain_for_update(a.cert_chain(), b.cert_chain()))
        .then_with(|| a.private_key().as_ref().cmp(b.private_key().as_ref()))
}

fn cmp_cert_chain_for_update(a: &[Certificate], b: &[Certificate]) -> CmpOrdering {
    a.iter()
        .map(Certificate::as_bytes)
        .cmp(b.iter().map(Certificate::as_bytes))
}

async fn initial_sync_with_timeout<T, F>(
    initial_sync: F,
    cancel: &CancellationToken,
    timeout: Option<Duration>,
) -> Result<T, X509SourceError>
where
    F: Future<Output = Result<T, X509SourceError>>,
{
    let Some(timeout) = timeout else {
        return initial_sync.await;
    };

    match tokio::time::timeout(timeout, initial_sync).await {
        Ok(result) => result,
        Err(_elapsed) => {
            cancel.cancel();
            Err(X509SourceError::InitialSyncTimeout)
        }
    }
}

impl SvidSource for X509Source {
    type Item = X509Svid;
    type Error = X509SourceError;

    fn svid(&self) -> Result<Arc<Self::Item>, Self::Error> {
        Self::svid(self)
    }
}

impl BundleSource for X509Source {
    type Item = X509Bundle;
    type Error = X509SourceError;

    fn bundle_for_trust_domain(
        &self,
        trust_domain: &TrustDomain,
    ) -> Result<Option<Arc<Self::Item>>, Self::Error> {
        self.assert_open()?;
        Ok(self.state().context()?.bundle_set().get(trust_domain))
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::collections::HashMap;
    use std::sync::Mutex;

    fn updates_for_test(rx: watch::Receiver<u64>) -> X509SourceUpdates {
        X509SourceUpdates {
            rx,
            shutdown: CancellationToken::new(),
        }
    }

    fn test_x509_context() -> Arc<X509Context> {
        let cert_bytes = include_bytes!("../../tests/testdata/svid/x509/1-svid-chain.der");
        let key_bytes = include_bytes!("../../tests/testdata/svid/x509/1-key.der");
        let svid = Arc::new(X509Svid::parse_from_der(cert_bytes, key_bytes).unwrap());
        Arc::new(X509Context::new([svid], Arc::new(X509BundleSet::new())))
    }

    struct HintPicker(&'static str);

    impl SvidPicker for HintPicker {
        fn pick_svid(&self, svids: &[Arc<X509Svid>]) -> Option<usize> {
            svids.iter().position(|svid| svid.hint() == Some(self.0))
        }
    }

    struct IndexPicker(usize);

    impl SvidPicker for IndexPicker {
        fn pick_svid(&self, svids: &[Arc<X509Svid>]) -> Option<usize> {
            (self.0 < svids.len()).then_some(self.0)
        }
    }

    /// Returns `(full_chain, leaf_only)` SVIDs that differ only by their intermediates.
    fn full_and_leaf_only_svids() -> (Arc<X509Svid>, Arc<X509Svid>) {
        use crate::cert::parsing::to_certificate_vec;

        let chain = include_bytes!("../../tests/testdata/svid/x509/1-svid-chain.der");
        let key = include_bytes!("../../tests/testdata/svid/x509/1-key.der");
        let certs = to_certificate_vec(chain).unwrap();
        let first_cert = certs
            .first()
            .expect("fixture must include leaf certificate");
        assert!(
            certs.len() > 1,
            "fixture must include intermediates to exercise chain-length differences"
        );

        let full = Arc::new(X509Svid::parse_from_der(chain, key).unwrap());
        let leaf_only = Arc::new(X509Svid::parse_from_der(first_cert.as_bytes(), key).unwrap());
        assert_ne!(
            full.as_ref(),
            leaf_only.as_ref(),
            "full-chain and leaf-only SVIDs must differ under structural equality"
        );
        (full, leaf_only)
    }

    #[track_caller]
    fn assert_apply_update_notifies_for_chain_change(initial: X509Context, incoming: X509Context) {
        let initial = Arc::new(initial);
        let incoming = Arc::new(incoming);
        assert!(!same_material_for_update(
            initial.as_ref(),
            incoming.as_ref(),
            None,
            incoming.default_svid().unwrap().as_ref(),
        ));

        let metrics = Arc::new(OrderingMetricsRecorder::new());
        let source = X509Source::new_for_test(
            initial,
            ReconnectConfig::default(),
            ResourceLimits::default(),
            Some(Arc::<OrderingMetricsRecorder>::clone(&metrics)),
            None,
        );
        metrics.set_updates(source.updated());

        let result = source
            .inner
            .apply_update(Arc::clone(&incoming))
            .expect("intermediate chain change should be accepted");
        assert_eq!(result, ApplyUpdateResult::Applied);
        assert_eq!(
            source.updated().last(),
            1,
            "intermediate chain changes must notify because TLS uses the full chain"
        );
        assert_eq!(metrics.update_sequences(), vec![Some(1)]);
        assert_eq!(source.x509_context().unwrap().as_ref(), incoming.as_ref());
    }

    fn supervisor_running_guard_for_test(source: &X509Source) -> SupervisorTerminationGuard {
        SupervisorTerminationGuard::new(Arc::clone(&source.inner))
    }

    async fn terminate_supervisor_for_test(terminate_guard: SupervisorTerminationGuard) {
        tokio::spawn(async move {
            let _terminate_on_drop = terminate_guard;
        })
        .await
        .expect("supervisor termination task should not panic");
    }

    #[tokio::test]
    async fn initial_sync_timeout_returns_timeout_and_cancels_token() {
        let cancel = CancellationToken::new();

        let result = initial_sync_with_timeout(
            std::future::pending::<Result<(), X509SourceError>>(),
            &cancel,
            Some(Duration::ZERO),
        )
        .await;

        assert!(matches!(result, Err(X509SourceError::InitialSyncTimeout)));
        assert!(cancel.is_cancelled());
    }

    #[tokio::test]
    async fn initial_sync_timeout_allows_success_before_timeout() {
        let cancel = CancellationToken::new();

        let result = initial_sync_with_timeout(
            async { Ok::<_, X509SourceError>("synced") },
            &cancel,
            Some(Duration::from_secs(60)),
        )
        .await;

        assert_eq!(result.unwrap(), "synced");
        assert!(!cancel.is_cancelled());
    }

    #[tokio::test]
    async fn initial_sync_without_timeout_waits_for_future() {
        let cancel = CancellationToken::new();

        let result =
            initial_sync_with_timeout(async { Ok::<_, X509SourceError>("synced") }, &cancel, None)
                .await;

        assert_eq!(result.unwrap(), "synced");
        assert!(!cancel.is_cancelled());
    }

    fn set_selected_svid_expiry_for_test(source: &X509Source, expiry_unix: i64) {
        let snapshot = source.inner.snapshot.load();
        source.inner.snapshot.store(Arc::new(Snapshot {
            expiry_unix,
            ..snapshot.as_ref().clone()
        }));
    }

    #[tokio::test]
    async fn test_wait_for_immediate_satisfaction() {
        let (tx, rx) = watch::channel(5u64);
        let mut updates = updates_for_test(rx);

        // Predicate is already satisfied (current value is 5, which is > 3)
        let result = updates.wait_for(|&seq| seq > 3).await;
        assert!(result.is_ok());
        assert_eq!(result.unwrap(), 5);

        // Update the value
        let _unused: Result<_, _> = tx.send(10);

        // Wait for predicate to be satisfied again (should return immediately with new value)
        let result = updates.wait_for(|&seq| seq > 8).await;
        assert!(result.is_ok());
        assert_eq!(result.unwrap(), 10);
    }

    #[tokio::test]
    async fn test_wait_for_waits_when_not_satisfied() {
        let (tx, rx) = watch::channel(1u64);
        let mut updates = updates_for_test(rx);

        // Spawn a task to update the value after a short delay
        let tx_clone = tx.clone();
        tokio::spawn(async move {
            tokio::time::sleep(Duration::from_millis(50)).await;
            let _unused: Result<_, _> = tx_clone.send(5);
        });

        // Predicate is not satisfied initially (1 is not > 3)
        // Should wait and then return when value becomes 5
        let result = tokio::time::timeout(Duration::from_secs(1), updates.wait_for(|&seq| seq > 3))
            .await
            .expect("Should complete within timeout");
        assert!(result.is_ok());
        assert_eq!(result.unwrap(), 5);
    }

    #[tokio::test]
    async fn test_updated_only_notifies_on_rotations_after_initial_sync() {
        // Verify that updated().changed() only notifies on rotations after initial sync,
        // not on the initial sync itself. The initial sequence number is 0.
        let (tx, rx) = watch::channel(0u64);
        let mut updates = updates_for_test(rx.clone());

        // Initial value is 0, so changed() should wait for an update
        let tx_clone = tx.clone();
        tokio::spawn(async move {
            tokio::time::sleep(Duration::from_millis(50)).await;
            // Simulate first rotation after initial sync
            let _unused: Result<_, _> = tx_clone.send(1);
        });

        // Should wait and then return when value becomes 1 (first rotation)
        let result = tokio::time::timeout(Duration::from_secs(1), updates.changed())
            .await
            .expect("Should complete within timeout");
        assert!(result.is_ok());
        assert_eq!(result.unwrap(), 1);
        assert_eq!(updates.last(), 1);
    }

    #[tokio::test]
    async fn test_updated_initial_sequence_is_zero() {
        // Verify that the initial sequence number is 0
        let (_tx, rx) = watch::channel(0u64);
        let updates = updates_for_test(rx);
        assert_eq!(updates.last(), 0);
    }

    #[tokio::test]
    async fn test_updated_subscribes_at_current_sequence() {
        let source = X509Source::new_for_test(
            test_x509_context(),
            ReconnectConfig::default(),
            ResourceLimits::default(),
            None,
            None,
        );

        source.inner.notify_update();

        let mut updates = source.updated();
        assert_eq!(updates.last(), 1);
        assert!(
            tokio::time::timeout(Duration::from_millis(20), updates.changed())
                .await
                .is_err(),
            "new subscribers should wait for updates after subscription"
        );

        source.inner.notify_update();
        assert_eq!(updates.changed().await.unwrap(), 2);
    }

    /// With no `updated()` / `watch` subscribers yet, `notify_update` must still advance the
    /// shared sequence; the first `updated()` should observe that value.
    #[test]
    fn test_notify_update_sequence_before_first_subscriber() {
        let source = X509Source::new_for_test(
            test_x509_context(),
            ReconnectConfig::default(),
            ResourceLimits::default(),
            None,
            None,
        );
        source.inner.notify_update();
        let updates = source.updated();
        assert_eq!(updates.last(), 1);
    }

    #[tokio::test]
    async fn test_updates_changed_returns_closed_after_shutdown() {
        let source = X509Source::new_for_test(
            test_x509_context(),
            ReconnectConfig::default(),
            ResourceLimits::default(),
            None,
            None,
        );
        let mut updates = source.updated();

        source.shutdown().await;

        let result = tokio::time::timeout(Duration::from_secs(1), updates.changed())
            .await
            .expect("changed should return after shutdown");
        assert!(matches!(result, Err(X509SourceError::Closed)));
    }

    #[tokio::test]
    async fn test_updates_wait_for_returns_closed_after_shutdown() {
        let source = X509Source::new_for_test(
            test_x509_context(),
            ReconnectConfig::default(),
            ResourceLimits::default(),
            None,
            None,
        );
        let mut updates = source.updated();

        source.shutdown().await;

        let result = tokio::time::timeout(Duration::from_secs(1), updates.wait_for(|&seq| seq > 0))
            .await
            .expect("wait_for should return after shutdown");
        assert!(matches!(result, Err(X509SourceError::Closed)));
    }

    #[tokio::test]
    async fn wait_for_returns_closed_after_shutdown_even_when_predicate_matches_current() {
        let source = X509Source::new_for_test(
            test_x509_context(),
            ReconnectConfig::default(),
            ResourceLimits::default(),
            None,
            None,
        );
        let mut updates = source.updated();
        let current = updates.last();

        source.shutdown().await;

        let result = updates.wait_for(|seq| *seq == current).await;
        assert!(matches!(result, Err(X509SourceError::Closed)));
    }

    #[tokio::test]
    async fn test_updates_changed_delivers_pending_update_before_closed() {
        let source = X509Source::new_for_test(
            test_x509_context(),
            ReconnectConfig::default(),
            ResourceLimits::default(),
            None,
            None,
        );
        let mut updates = source.updated();

        source.inner.notify_update();
        source.inner.cancel.cancel();

        assert_eq!(updates.changed().await.unwrap(), 1);
        let result = tokio::time::timeout(Duration::from_secs(1), updates.changed())
            .await
            .expect("changed should return closed after pending update is observed");
        assert!(matches!(result, Err(X509SourceError::Closed)));
    }

    #[tokio::test]
    async fn test_updates_changed_returns_closed_after_last_source_drop() {
        let source = X509Source::new_for_test(
            test_x509_context(),
            ReconnectConfig::default(),
            ResourceLimits::default(),
            None,
            None,
        );
        let mut updates = source.updated();

        drop(source);

        let result = tokio::time::timeout(Duration::from_secs(1), updates.changed())
            .await
            .expect("changed should return after last source handle is dropped");
        assert!(matches!(result, Err(X509SourceError::Closed)));
    }

    #[tokio::test]
    async fn test_dropping_last_source_handle_cancels_supervisor_token() {
        let source = X509Source::new_for_test(
            test_x509_context(),
            ReconnectConfig::default(),
            ResourceLimits::default(),
            None,
            None,
        );
        let clone = source.clone();
        let task_inner = Arc::clone(&source.inner);
        let token = task_inner.cancel.clone();
        let (stopped_tx, stopped_rx) = tokio::sync::oneshot::channel();

        tokio::spawn(async move {
            token.cancelled().await;
            let _unused: Result<_, _> = stopped_tx.send(());
            drop(task_inner);
        });

        drop(source);
        // One public handle still holds the shutdown guard, so cancellation must not fire yet.
        assert!(!clone.inner.cancel.is_cancelled());

        drop(clone);
        tokio::time::timeout(Duration::from_secs(1), stopped_rx)
            .await
            .expect("supervisor token should be cancelled when last source handle is dropped")
            .expect("supervisor observer should send stop notification");
    }

    #[tokio::test]
    async fn test_supervisor_termination_marks_unhealthy_and_closes_updates() {
        let source = X509Source::new_for_test(
            test_x509_context(),
            ReconnectConfig::default(),
            ResourceLimits::default(),
            None,
            None,
        );
        let running_guard = supervisor_running_guard_for_test(&source);
        let mut changed_updates = source.updated();
        let mut wait_updates = source.updated();

        assert!(
            source.is_healthy(),
            "cached SVID should be healthy while supervisor is running"
        );

        terminate_supervisor_for_test(running_guard).await;

        assert!(
            !source.is_healthy(),
            "source must be unhealthy after supervisor termination"
        );
        let changed = tokio::time::timeout(Duration::from_secs(1), changed_updates.changed())
            .await
            .expect("changed should stop waiting after supervisor termination");
        assert!(matches!(changed, Err(X509SourceError::Closed)));

        let waited = tokio::time::timeout(
            Duration::from_secs(1),
            wait_updates.wait_for(|&seq| seq > 0),
        )
        .await
        .expect("wait_for should stop waiting after supervisor termination");
        assert!(matches!(waited, Err(X509SourceError::Closed)));
    }

    #[tokio::test]
    async fn wait_for_returns_closed_after_supervisor_termination_even_when_predicate_matches_current(
    ) {
        let source = X509Source::new_for_test(
            test_x509_context(),
            ReconnectConfig::default(),
            ResourceLimits::default(),
            None,
            None,
        );
        let running_guard = supervisor_running_guard_for_test(&source);
        let mut updates = source.updated();
        let current = updates.last();

        terminate_supervisor_for_test(running_guard).await;

        let result = updates.wait_for(|seq| *seq == current).await;
        assert!(matches!(result, Err(X509SourceError::Closed)));
    }

    #[test]
    fn test_is_healthy_returns_false_when_selected_svid_is_expired() {
        let source = X509Source::new_for_test(
            test_x509_context(),
            ReconnectConfig::default(),
            ResourceLimits::default(),
            None,
            None,
        );
        let _running_guard = supervisor_running_guard_for_test(&source);
        let now = unix_timestamp_now().expect("system time should be after UNIX epoch");

        assert!(
            source.is_healthy(),
            "source should be healthy while selected SVID expiry is in the future"
        );

        set_selected_svid_expiry_for_test(&source, now);
        assert!(
            !source.is_healthy(),
            "source should be unhealthy once selected SVID expiry is reached"
        );
    }

    #[tokio::test]
    async fn is_healthy_false_after_shutdown() {
        let source = X509Source::new_for_test(
            test_x509_context(),
            ReconnectConfig::default(),
            ResourceLimits::default(),
            None,
            None,
        );
        let _running_guard = supervisor_running_guard_for_test(&source);

        assert!(source.is_healthy());
        source.shutdown().await;
        assert!(!source.is_healthy());
    }

    #[test]
    fn is_healthy_false_after_cancel() {
        let source = X509Source::new_for_test(
            test_x509_context(),
            ReconnectConfig::default(),
            ResourceLimits::default(),
            None,
            None,
        );
        let _running_guard = supervisor_running_guard_for_test(&source);

        assert!(source.is_healthy());
        source.inner.cancel.cancel();
        assert!(!source.is_healthy());
    }

    /// Test metrics recorder that counts error recordings by kind.
    struct TestMetricsRecorder {
        counts: Arc<Mutex<HashMap<MetricsErrorKind, u64>>>,
    }

    impl TestMetricsRecorder {
        fn new() -> Self {
            Self {
                counts: Arc::new(Mutex::new(HashMap::new())),
            }
        }

        fn count(&self, kind: MetricsErrorKind) -> u64 {
            *self.counts.lock().unwrap().get(&kind).unwrap_or(&0)
        }
    }

    impl MetricsRecorder for TestMetricsRecorder {
        fn record_update(&self) {}
        fn record_reconnect(&self) {}
        fn record_error(&self, kind: MetricsErrorKind) {
            *self.counts.lock().unwrap().entry(kind).or_insert(0) += 1;
        }
    }

    struct OrderingMetricsRecorder {
        updates: Mutex<Option<X509SourceUpdates>>,
        update_sequences: Mutex<Vec<Option<u64>>>,
    }

    impl OrderingMetricsRecorder {
        fn new() -> Self {
            Self {
                updates: Mutex::new(None),
                update_sequences: Mutex::new(Vec::new()),
            }
        }

        fn set_updates(&self, updates: X509SourceUpdates) {
            *self.updates.lock().unwrap() = Some(updates);
        }

        fn update_sequences(&self) -> Vec<Option<u64>> {
            self.update_sequences.lock().unwrap().clone()
        }

        fn record_observed_update(&self) {
            let sequence = self
                .updates
                .lock()
                .unwrap()
                .as_ref()
                .map(X509SourceUpdates::last);
            self.update_sequences.lock().unwrap().push(sequence);
        }
    }

    impl MetricsRecorder for OrderingMetricsRecorder {
        fn record_update(&self) {
            self.record_observed_update();
        }

        fn record_reconnect(&self) {}
        fn record_error(&self, _kind: MetricsErrorKind) {}
    }

    struct PanickingOrderingMetricsRecorder {
        inner: OrderingMetricsRecorder,
    }

    impl PanickingOrderingMetricsRecorder {
        fn new() -> Self {
            Self {
                inner: OrderingMetricsRecorder::new(),
            }
        }

        fn set_updates(&self, updates: X509SourceUpdates) {
            self.inner.set_updates(updates);
        }

        fn update_sequences(&self) -> Vec<Option<u64>> {
            self.inner.update_sequences()
        }
    }

    impl MetricsRecorder for PanickingOrderingMetricsRecorder {
        fn record_update(&self) {
            self.inner.record_observed_update();
            panic!("intentional metrics panic for update ordering test");
        }

        fn record_reconnect(&self) {}
        fn record_error(&self, _kind: MetricsErrorKind) {}
    }

    #[test]
    fn test_apply_update_notifies_before_recording_success_metrics() {
        let metrics = Arc::new(OrderingMetricsRecorder::new());
        let source = X509Source::new_for_test(
            Arc::new(X509Context::new([], Arc::new(X509BundleSet::new()))),
            ReconnectConfig::default(),
            ResourceLimits::default(),
            Some(Arc::<OrderingMetricsRecorder>::clone(&metrics)),
            None,
        );
        metrics.set_updates(source.updated());

        let result = source.inner.apply_update(test_x509_context());

        assert_eq!(
            result.expect("valid X.509 update should be accepted"),
            ApplyUpdateResult::Applied
        );
        assert_eq!(metrics.update_sequences(), vec![Some(1)]);
        assert_eq!(source.updated().last(), 1);
        assert_eq!(source.x509_context().unwrap().svids().len(), 1);
    }

    #[test]
    fn test_apply_update_skips_notify_for_unchanged_context() {
        // The Workload API re-delivers the current context as the first item on every
        // new stream (initial sync, reconnect after a dropped stream, agent restart, etc.).
        // apply_update must not treat re-delivery of an identical context as a rotation.
        let metrics = Arc::new(OrderingMetricsRecorder::new());
        let source = X509Source::new_for_test(
            test_x509_context(),
            ReconnectConfig::default(),
            ResourceLimits::default(),
            Some(Arc::<OrderingMetricsRecorder>::clone(&metrics)),
            None,
        );
        metrics.set_updates(source.updated());

        // Same context (by value) delivered again, e.g. on reconnect.
        let result = source.inner.apply_update(test_x509_context());

        assert_eq!(
            result.expect("re-delivery of an unchanged context should be accepted"),
            ApplyUpdateResult::Unchanged
        );
        assert_eq!(
            source.updated().last(),
            0,
            "no notification should be sent for an unchanged context"
        );
        assert!(
            metrics.update_sequences().is_empty(),
            "record_update should not be called for an unchanged context"
        );

        // A genuine rotation must still notify, even if it only changes the bundle set
        // (e.g. a new federated trust bundle) while keeping the same SVID.
        let mut rotated_bundle_set = X509BundleSet::new();
        rotated_bundle_set.add_bundle(X509Bundle::new(TrustDomain::new("example.org").unwrap()));
        let rotated_ctx = Arc::new(X509Context::new(
            test_x509_context().svids().to_vec(),
            Arc::new(rotated_bundle_set),
        ));

        let result = source
            .inner
            .apply_update(rotated_ctx)
            .expect("valid rotation should be accepted");
        assert_eq!(result, ApplyUpdateResult::Applied);
        assert_eq!(
            source.updated().last(),
            1,
            "a genuine rotation should still notify"
        );
        assert_eq!(metrics.update_sequences(), vec![Some(1)]);
    }

    #[test]
    fn same_material_for_update_detects_default_svid_change() {
        let cert = include_bytes!("../../tests/testdata/svid/x509/1-svid-chain.der");
        let key = include_bytes!("../../tests/testdata/svid/x509/1-key.der");
        let svid_a = Arc::new(
            X509Svid::parse_from_der_with_hint(cert, key, Some("internal".into())).unwrap(),
        );
        let svid_b = Arc::new(
            X509Svid::parse_from_der_with_hint(cert, key, Some("external".into())).unwrap(),
        );
        assert_ne!(
            svid_a.as_ref(),
            svid_b.as_ref(),
            "fixture SVIDs must differ so order-sensitive equality would fail"
        );

        let mut bundle_set = X509BundleSet::new();
        bundle_set.add_bundle(X509Bundle::new(TrustDomain::new("example.org").unwrap()));
        let bundle_set = Arc::new(bundle_set);

        let ordered = X509Context::new(
            [Arc::clone(&svid_a), Arc::clone(&svid_b)],
            Arc::clone(&bundle_set),
        );
        let reversed = X509Context::new([svid_b, svid_a], bundle_set);

        assert_ne!(
            ordered, reversed,
            "public X509Context equality remains order-sensitive"
        );
        assert!(!same_material_for_update(
            &ordered,
            &reversed,
            None,
            reversed.default_svid().unwrap().as_ref(),
        ));
    }

    #[test]
    fn test_apply_update_notifies_for_intermediate_chain_change() {
        let (full, leaf_only) = full_and_leaf_only_svids();
        let mut bundle_set = X509BundleSet::new();
        bundle_set.add_bundle(X509Bundle::new(TrustDomain::new("example.org").unwrap()));
        let bundle_set = Arc::new(bundle_set);

        let leaf_only_ctx = X509Context::new([leaf_only], Arc::clone(&bundle_set));
        let with_chain = X509Context::new([full], bundle_set);

        assert_apply_update_notifies_for_chain_change(leaf_only_ctx, with_chain);
    }

    #[test]
    fn test_apply_update_notifies_for_non_selected_svid_chain_change() {
        let chain = include_bytes!("../../tests/testdata/svid/x509/1-svid-chain.der");
        let key = include_bytes!("../../tests/testdata/svid/x509/1-key.der");
        let (full, leaf_only) = full_and_leaf_only_svids();
        let mut bundle_set = X509BundleSet::new();
        bundle_set.add_bundle(X509Bundle::new(TrustDomain::new("example.org").unwrap()));
        let bundle_set = Arc::new(bundle_set);

        // The chain change is in a non-selected SVID so the selected-SVID check passes and
        // the multiset comparison is what must detect it.
        let selected = Arc::new(
            X509Svid::parse_from_der_with_hint(chain, key, Some("selected".into())).unwrap(),
        );
        let leaf_only_ctx =
            X509Context::new([Arc::clone(&selected), leaf_only], Arc::clone(&bundle_set));
        let with_chain = X509Context::new([selected, full], bundle_set);

        assert_apply_update_notifies_for_chain_change(leaf_only_ctx, with_chain);
    }

    #[test]
    fn same_material_for_update_ignores_tail_order_when_chain_differs() {
        use crate::cert::parsing::to_certificate_vec;

        let chain = include_bytes!("../../tests/testdata/svid/x509/1-svid-chain.der");
        let key = include_bytes!("../../tests/testdata/svid/x509/1-key.der");
        let certs = to_certificate_vec(chain).unwrap();
        let first_cert = certs
            .first()
            .expect("fixture must include leaf certificate");

        let full = Arc::new(X509Svid::parse_from_der(chain, key).unwrap());
        let leaf_only = Arc::new(X509Svid::parse_from_der(first_cert.as_bytes(), key).unwrap());
        assert_ne!(
            full.as_ref(),
            leaf_only.as_ref(),
            "fixtures must differ only by presented chain"
        );

        let mut bundle_set = X509BundleSet::new();
        bundle_set.add_bundle(X509Bundle::new(TrustDomain::new("example.org").unwrap()));
        let bundle_set = Arc::new(bundle_set);

        let selected = Arc::new(
            X509Svid::parse_from_der_with_hint(chain, key, Some("selected".into())).unwrap(),
        );
        let ordered = X509Context::new(
            [
                Arc::clone(&selected),
                Arc::clone(&full),
                Arc::clone(&leaf_only),
            ],
            Arc::clone(&bundle_set),
        );
        let reversed = X509Context::new([selected, leaf_only, full], bundle_set);

        assert_ne!(
            ordered, reversed,
            "public X509Context equality remains order-sensitive"
        );
        assert!(
            same_material_for_update(
                &ordered,
                &reversed,
                None,
                reversed.default_svid().unwrap().as_ref(),
            ),
            "equivalent SVID multisets should compare equal even when matching requires the chain"
        );
    }

    #[test]
    fn same_material_for_update_ignores_bundle_authority_order() {
        use crate::cert::parsing::to_certificate_vec;

        let cert = include_bytes!("../../tests/testdata/svid/x509/1-svid-chain.der");
        let key = include_bytes!("../../tests/testdata/svid/x509/1-key.der");
        let svid = Arc::new(X509Svid::parse_from_der(cert, key).unwrap());
        let certs = to_certificate_vec(cert).unwrap();
        let [first, second, ..] = certs.as_slice() else {
            panic!("fixture must include multiple certificates to exercise authority order");
        };

        let trust_domain = TrustDomain::new("example.org").unwrap();
        let mut bundle_a = X509Bundle::new(trust_domain.clone());
        bundle_a
            .add_authority(first.as_bytes())
            .expect("authority should parse");
        bundle_a
            .add_authority(second.as_bytes())
            .expect("second authority should parse");

        let mut bundle_b = X509Bundle::new(trust_domain);
        bundle_b
            .add_authority(second.as_bytes())
            .expect("authority should parse");
        bundle_b
            .add_authority(first.as_bytes())
            .expect("second authority should parse");
        assert_ne!(bundle_a, bundle_b);

        let mut bundle_set_a = X509BundleSet::new();
        bundle_set_a.add_bundle(bundle_a);
        let mut bundle_set_b = X509BundleSet::new();
        bundle_set_b.add_bundle(bundle_b);

        let ctx_a = X509Context::new([Arc::clone(&svid)], Arc::new(bundle_set_a));
        let ctx_b = X509Context::new([svid], Arc::new(bundle_set_b));

        assert!(same_material_for_update(
            &ctx_a,
            &ctx_b,
            None,
            ctx_b.default_svid().unwrap().as_ref(),
        ));
    }

    #[test]
    fn test_apply_update_applies_reorder_that_changes_default_svid() {
        let cert = include_bytes!("../../tests/testdata/svid/x509/1-svid-chain.der");
        let key = include_bytes!("../../tests/testdata/svid/x509/1-key.der");
        let svid_a = Arc::new(
            X509Svid::parse_from_der_with_hint(cert, key, Some("internal".into())).unwrap(),
        );
        let svid_b = Arc::new(
            X509Svid::parse_from_der_with_hint(cert, key, Some("external".into())).unwrap(),
        );

        let mut bundle_set = X509BundleSet::new();
        bundle_set.add_bundle(X509Bundle::new(TrustDomain::new("example.org").unwrap()));
        let bundle_set = Arc::new(bundle_set);

        let initial = Arc::new(X509Context::new(
            [Arc::clone(&svid_a), Arc::clone(&svid_b)],
            Arc::clone(&bundle_set),
        ));
        let reordered = Arc::new(X509Context::new([Arc::clone(&svid_b), svid_a], bundle_set));

        let metrics = Arc::new(OrderingMetricsRecorder::new());
        let source = X509Source::new_for_test(
            initial,
            ReconnectConfig::default(),
            ResourceLimits::default(),
            Some(Arc::<OrderingMetricsRecorder>::clone(&metrics)),
            None,
        );
        metrics.set_updates(source.updated());

        let result = source
            .inner
            .apply_update(Arc::clone(&reordered))
            .expect("new default SVID should be accepted");
        assert_eq!(result, ApplyUpdateResult::Applied);
        assert_eq!(source.updated().last(), 1);
        assert_eq!(metrics.update_sequences(), vec![Some(1)]);
        assert_eq!(source.x509_context().unwrap().as_ref(), reordered.as_ref());
        assert_eq!(source.svid().unwrap().as_ref(), svid_b.as_ref());
    }

    #[test]
    fn test_apply_update_suppresses_tail_only_svid_reorder() {
        let cert = include_bytes!("../../tests/testdata/svid/x509/1-svid-chain.der");
        let key = include_bytes!("../../tests/testdata/svid/x509/1-key.der");
        let make_svid = |hint: &str| {
            Arc::new(X509Svid::parse_from_der_with_hint(cert, key, Some(hint.into())).unwrap())
        };
        let selected = make_svid("selected");
        let tail_a = make_svid("tail-a");
        let tail_b = make_svid("tail-b");
        let bundle_set = Arc::new(X509BundleSet::new());
        let initial = Arc::new(X509Context::new(
            [
                Arc::clone(&selected),
                Arc::clone(&tail_a),
                Arc::clone(&tail_b),
            ],
            Arc::clone(&bundle_set),
        ));
        let reordered = Arc::new(X509Context::new([selected, tail_b, tail_a], bundle_set));
        let source = X509Source::new_for_test(
            Arc::clone(&initial),
            ReconnectConfig::default(),
            ResourceLimits::default(),
            None,
            None,
        );

        let result = source.inner.apply_update(reordered).unwrap();

        assert_eq!(result, ApplyUpdateResult::Unchanged);
        assert_eq!(source.updated().last(), 0);
        assert_eq!(source.x509_context().unwrap().as_ref(), initial.as_ref());
    }

    #[test]
    fn test_apply_update_suppresses_reorder_when_picker_selection_is_unchanged() {
        let cert = include_bytes!("../../tests/testdata/svid/x509/1-svid-chain.der");
        let key = include_bytes!("../../tests/testdata/svid/x509/1-key.der");
        let selected = Arc::new(
            X509Svid::parse_from_der_with_hint(cert, key, Some("selected".into())).unwrap(),
        );
        let other =
            Arc::new(X509Svid::parse_from_der_with_hint(cert, key, Some("other".into())).unwrap());
        let bundle_set = Arc::new(X509BundleSet::new());
        let initial = Arc::new(X509Context::new(
            [Arc::clone(&selected), Arc::clone(&other)],
            Arc::clone(&bundle_set),
        ));
        let reordered = Arc::new(X509Context::new([other, selected], bundle_set));
        let source = X509Source::new_for_test(
            Arc::clone(&initial),
            ReconnectConfig::default(),
            ResourceLimits::default(),
            None,
            Some(Box::new(HintPicker("selected"))),
        );

        let result = source.inner.apply_update(reordered).unwrap();

        assert_eq!(result, ApplyUpdateResult::Unchanged);
        assert_eq!(source.updated().last(), 0);
        assert_eq!(source.x509_context().unwrap().as_ref(), initial.as_ref());
        assert_eq!(source.svid().unwrap().hint(), Some("selected"));
    }

    #[test]
    fn test_apply_update_applies_reorder_when_picker_selection_changes() {
        let cert = include_bytes!("../../tests/testdata/svid/x509/1-svid-chain.der");
        let key = include_bytes!("../../tests/testdata/svid/x509/1-key.der");
        let svid_a =
            Arc::new(X509Svid::parse_from_der_with_hint(cert, key, Some("a".into())).unwrap());
        let svid_b =
            Arc::new(X509Svid::parse_from_der_with_hint(cert, key, Some("b".into())).unwrap());
        let bundle_set = Arc::new(X509BundleSet::new());
        let initial = Arc::new(X509Context::new(
            [Arc::clone(&svid_a), Arc::clone(&svid_b)],
            Arc::clone(&bundle_set),
        ));
        let reordered = Arc::new(X509Context::new(
            [Arc::clone(&svid_b), Arc::clone(&svid_a)],
            bundle_set,
        ));
        let source = X509Source::new_for_test(
            initial,
            ReconnectConfig::default(),
            ResourceLimits::default(),
            None,
            Some(Box::new(IndexPicker(1))),
        );

        let result = source.inner.apply_update(Arc::clone(&reordered)).unwrap();

        assert_eq!(result, ApplyUpdateResult::Applied);
        assert_eq!(source.updated().last(), 1);
        assert_eq!(source.x509_context().unwrap().as_ref(), reordered.as_ref());
        assert_eq!(source.svid().unwrap().as_ref(), svid_a.as_ref());
    }

    #[test]
    fn test_apply_update_publishes_and_notifies_before_panicking_success_metrics() {
        let metrics = Arc::new(PanickingOrderingMetricsRecorder::new());
        let source = X509Source::new_for_test(
            Arc::new(X509Context::new([], Arc::new(X509BundleSet::new()))),
            ReconnectConfig::default(),
            ResourceLimits::default(),
            Some(Arc::<PanickingOrderingMetricsRecorder>::clone(&metrics)),
            None,
        );
        metrics.set_updates(source.updated());

        let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
            source.inner.apply_update(test_x509_context())
        }));

        result.expect_err("metrics recorder should panic after notification");
        assert_eq!(metrics.update_sequences(), vec![Some(1)]);
        assert_eq!(source.updated().last(), 1);
        assert_eq!(source.x509_context().unwrap().svids().len(), 1);
    }

    #[test]
    fn test_metrics_recorded_exactly_once_per_rejected_update() {
        // Verify that UpdateRejected and limit metrics are recorded exactly once
        // per rejected update, with no double-counting.
        use super::super::builder::{ReconnectConfig, ResourceLimits};
        use crate::workload_api::x509_context::X509Context;
        use crate::{TrustDomain, X509Bundle, X509BundleSet, X509Svid};
        use std::sync::Arc;

        // Load test fixture SVID
        let cert_bytes = include_bytes!("../../tests/testdata/svid/x509/1-svid-chain.der");
        let key_bytes = include_bytes!("../../tests/testdata/svid/x509/1-key.der");
        let svid = Arc::new(X509Svid::parse_from_der(cert_bytes, key_bytes).unwrap());

        let metrics = Arc::new(TestMetricsRecorder::new());
        let limits = ResourceLimits {
            max_svids: Some(100),
            max_bundles: Some(0), // Limit that will be exceeded
            max_bundle_der_bytes: Some(1000),
        };

        // Create a context with 1 bundle (exceeds max_bundles=0)
        let trust_domain = TrustDomain::new("example.org").unwrap();
        let bundle = X509Bundle::new(trust_domain);
        let mut bundle_set = X509BundleSet::new();
        bundle_set.add_bundle(bundle);

        let ctx = X509Context::new([svid], Arc::new(bundle_set));

        // Create source using test seam
        let source = {
            let metrics = Arc::clone(&metrics);
            X509Source::new_for_test(
                Arc::new(X509Context::new([], Arc::new(X509BundleSet::new()))),
                ReconnectConfig::default(),
                limits,
                Some(metrics),
                None,
            )
        };

        // Apply update that should be rejected due to max_bundles limit
        let result = source.inner.apply_update(Arc::new(ctx));

        // Should fail with ResourceLimitExceeded
        assert!(matches!(
            result,
            Err(X509SourceError::ResourceLimitExceeded {
                kind: super::super::errors::LimitKind::MaxBundles,
                ..
            })
        ));

        // Verify metrics recorded exactly once
        assert_eq!(metrics.count(MetricsErrorKind::LimitMaxBundles), 1);
        assert_eq!(metrics.count(MetricsErrorKind::UpdateRejected), 1);
        // Verify no other limit metrics were recorded
        assert_eq!(metrics.count(MetricsErrorKind::LimitMaxSvids), 0);
        assert_eq!(metrics.count(MetricsErrorKind::LimitMaxBundleDerBytes), 0);
    }

    #[test]
    fn test_apply_update_accepts_trust_removal_with_expired_selected_svid() {
        use super::super::builder::{ReconnectConfig, ResourceLimits};
        use crate::workload_api::x509_context::X509Context;
        use crate::{TrustDomain, X509Bundle, X509BundleSet, X509Svid};
        use std::sync::Arc;

        let good_trust_domain = TrustDomain::new("good.example.org").unwrap();
        let mut good_bundle_set = X509BundleSet::new();
        good_bundle_set.add_bundle(X509Bundle::new(good_trust_domain.clone()));
        let initial_ctx = Arc::new(X509Context::new(
            test_x509_context().svids().to_vec(),
            Arc::new(good_bundle_set),
        ));

        let source = X509Source::new_for_test(
            Arc::clone(&initial_ctx),
            ReconnectConfig::default(),
            ResourceLimits::default(),
            None,
            None,
        );

        // A rotated update carrying a fresh (different) bundle, but whose selected SVID
        // is already expired.
        let expired_cert_bytes =
            include_bytes!("../../tests/testdata/svid/x509/expired-svid-chain.der");
        let expired_key_bytes = include_bytes!("../../tests/testdata/svid/x509/expired-key.der");
        let expired_svid = Arc::new(
            X509Svid::parse_from_der(expired_cert_bytes, expired_key_bytes)
                .expect("expired fixture should parse as an X509-SVID"),
        );
        let new_trust_domain = TrustDomain::new("new.example.org").unwrap();
        let mut new_bundle_set = X509BundleSet::new();
        new_bundle_set.add_bundle(X509Bundle::new(new_trust_domain.clone()));
        let expired_ctx = Arc::new(X509Context::new([expired_svid], Arc::new(new_bundle_set)));

        let result = source.inner.apply_update(Arc::clone(&expired_ctx));
        assert_eq!(result.unwrap(), ApplyUpdateResult::Applied);
        assert!(matches!(
            source.svid(),
            Err(X509SourceError::NoSuitableSvid)
        ));
        assert_eq!(
            source.x509_context().unwrap().as_ref(),
            expired_ctx.as_ref()
        );
        let current_bundles = source.bundle_set().unwrap();
        assert!(current_bundles.get(&good_trust_domain).is_none());
        assert!(current_bundles.get(&new_trust_domain).is_some());
        assert_eq!(source.updated().last(), 1);
    }

    #[test]
    fn test_new_with_normalizes_reconnect_config() {
        // Verify that X509Source::new_with() normalizes reconnect config at the authoritative boundary.
        use super::super::builder::{ReconnectConfig, ResourceLimits};
        use crate::workload_api::x509_context::X509Context;
        use crate::{X509BundleSet, X509Svid};
        use std::sync::Arc;
        use std::time::Duration;

        // Load test fixture SVID
        let cert_bytes = include_bytes!("../../tests/testdata/svid/x509/1-svid-chain.der");
        let key_bytes = include_bytes!("../../tests/testdata/svid/x509/1-key.der");
        let svid = Arc::new(X509Svid::parse_from_der(cert_bytes, key_bytes).unwrap());

        // Create a context with valid SVID and bundle
        let ctx = X509Context::new([svid], Arc::new(X509BundleSet::new()));

        // Create reconnect config with inverted min/max (min > max)
        let inverted_reconnect = ReconnectConfig {
            min_backoff: Duration::from_secs(10),
            max_backoff: Duration::from_secs(1),
        };

        // Create source using test seam with inverted reconnect config
        let source = X509Source::new_for_test(
            Arc::new(ctx),
            inverted_reconnect,
            ResourceLimits::default(),
            None,
            None,
        );

        // Verify that reconnect config was normalized (swapped)
        assert_eq!(source.inner.reconnect.min_backoff, Duration::from_secs(1));
        assert_eq!(source.inner.reconnect.max_backoff, Duration::from_secs(10));
    }

    #[test]
    fn test_initial_sync_validation_records_correct_metrics() {
        // Verify that validation records limit metrics and NoSuitableSvid,
        // but does NOT record UpdateRejected (that's recorded by apply_update).
        use super::super::builder::ResourceLimits;
        use super::super::limits::validate_context;
        use crate::workload_api::x509_context::X509Context;
        use crate::{TrustDomain, X509Bundle, X509BundleSet, X509Svid};
        use std::sync::Arc;

        // Load test fixture SVID
        let cert_bytes = include_bytes!("../../tests/testdata/svid/x509/1-svid-chain.der");
        let key_bytes = include_bytes!("../../tests/testdata/svid/x509/1-key.der");
        let svid = Arc::new(X509Svid::parse_from_der(cert_bytes, key_bytes).unwrap());

        let metrics = Arc::new(TestMetricsRecorder::new());
        let limits = ResourceLimits {
            max_svids: Some(100),
            max_bundles: Some(0), // Limit that will be exceeded
            max_bundle_der_bytes: Some(1000),
        };

        // Create a context with 1 bundle (exceeds max_bundles=0)
        let trust_domain = TrustDomain::new("example.org").unwrap();
        let bundle = X509Bundle::new(trust_domain);
        let mut bundle_set = X509BundleSet::new();
        bundle_set.add_bundle(bundle);

        let ctx = X509Context::new([svid], Arc::new(bundle_set));

        // Validate context (simulating validation used in both initial sync and updates)
        let result = validate_context(
            &ctx,
            None, // No picker
            limits,
            Some(metrics.as_ref()),
        );

        // Should fail with ResourceLimitExceeded
        assert!(matches!(
            result,
            Err(X509SourceError::ResourceLimitExceeded {
                kind: super::super::errors::LimitKind::MaxBundles,
                ..
            })
        ));

        // Verify limit metric was recorded
        assert_eq!(metrics.count(MetricsErrorKind::LimitMaxBundles), 1);
        // Verify UpdateRejected was NOT recorded (that's recorded by apply_update, not validate_context)
        assert_eq!(metrics.count(MetricsErrorKind::UpdateRejected), 0);
        // Verify no other limit metrics were recorded
        assert_eq!(metrics.count(MetricsErrorKind::LimitMaxSvids), 0);
        assert_eq!(metrics.count(MetricsErrorKind::LimitMaxBundleDerBytes), 0);
    }

    #[test]
    fn test_resource_limits_unlimited() {
        // Verify that ResourceLimits::unlimited() creates limits with all fields set to None.
        use super::super::builder::ResourceLimits;

        let unlimited = ResourceLimits::unlimited();
        assert_eq!(unlimited.max_svids, None);
        assert_eq!(unlimited.max_bundles, None);
        assert_eq!(unlimited.max_bundle_der_bytes, None);
    }

    #[test]
    fn test_resource_limits_default_limits() {
        // Verify that ResourceLimits::default_limits() creates limits with conservative defaults.
        use super::super::builder::ResourceLimits;

        let limits = ResourceLimits::default_limits();
        assert_eq!(limits.max_svids, Some(100));
        assert_eq!(limits.max_bundles, Some(200));
        assert_eq!(limits.max_bundle_der_bytes, Some(4 * 1024 * 1024)); // 4MB
    }

    #[test]
    fn test_resource_limits_mixed() {
        // Verify that ResourceLimits can have mixed unlimited and limited fields.
        use super::super::builder::ResourceLimits;

        let mixed = ResourceLimits {
            max_svids: Some(50),
            max_bundles: None,                       // Unlimited
            max_bundle_der_bytes: Some(1024 * 1024), // 1MB
        };

        assert_eq!(mixed.max_svids, Some(50));
        assert_eq!(mixed.max_bundles, None);
        assert_eq!(mixed.max_bundle_der_bytes, Some(1024 * 1024));
    }
    #[test]
    fn picker_failure_publishes_current_trust_without_identity_fallback() {
        let initial = test_x509_context();
        let mut initial_svids = initial.svids().to_vec();
        initial_svids.push(Arc::clone(initial.svids().first().unwrap()));
        let source = X509Source::new_for_test(
            Arc::new(X509Context::new(
                initial_svids,
                Arc::new(X509BundleSet::new()),
            )),
            ReconnectConfig::default(),
            ResourceLimits::default(),
            None,
            Some(Box::new(IndexPicker(1))),
        );
        source.svid().unwrap();
        let incoming = initial;
        assert_eq!(
            source.inner.apply_update(Arc::clone(&incoming)).unwrap(),
            ApplyUpdateResult::Applied
        );
        source.svid().unwrap_err();
        assert_eq!(source.x509_context().unwrap().as_ref(), incoming.as_ref());
        assert_eq!(source.updated().last(), 1);
    }
    #[test]
    fn coherent_state_retains_its_identity_and_bundles_across_publications() {
        let (full, leaf) = full_and_leaf_only_svids();
        let mut bundles = X509BundleSet::new();
        bundles.add_bundle(
            X509Bundle::parse_from_der(
                TrustDomain::new("example.org").unwrap(),
                include_bytes!("../../tests/testdata/bundle/x509/cert1.der"),
            )
            .unwrap(),
        );
        let initial = Arc::new(X509Context::new(
            vec![Arc::clone(&full), Arc::clone(&leaf)],
            Arc::new(bundles),
        ));
        let source = X509Source::new_for_test(
            Arc::clone(&initial),
            ReconnectConfig::default(),
            ResourceLimits::default(),
            None,
            Some(Box::new(IndexPicker(1))),
        );
        let before = source.state();
        let incoming = Arc::new(X509Context::new(
            vec![Arc::clone(&leaf), Arc::clone(&full)],
            Arc::new(X509BundleSet::new()),
        ));
        source.inner.apply_update(Arc::clone(&incoming)).unwrap();
        let after = source.state();
        for (state, expected, revision) in [(&before, &initial, 0), (&after, &incoming, 1)] {
            let context = state.context().unwrap();
            assert_eq!(context.as_ref(), expected.as_ref());
            assert_eq!(state.svid().unwrap(), *context.svids().get(1).unwrap());
            assert_eq!(state.revision(), revision);
        }
        assert!(!before.context().unwrap().bundle_set().is_empty());
        assert!(after.context().unwrap().bundle_set().is_empty());
    }

    #[test]
    fn state_expiry_and_unavailable_clock_affect_identity_without_reverting_trust() {
        let source = X509Source::new_for_test(
            test_x509_context(),
            ReconnectConfig::default(),
            ResourceLimits::default(),
            None,
            None,
        );
        let state = source.state();
        state.svid_at(Some(state.expiry_unix - 1)).unwrap();
        assert!(matches!(
            state.svid_at(Some(state.expiry_unix)),
            Err(X509SourceError::NoSuitableSvid)
        ));
        assert!(matches!(
            state.svid_at(None),
            Err(X509SourceError::NoSuitableSvid)
        ));
        state.context().unwrap();
    }
}

#[cfg(test)]
mod lifecycle_tests {
    use super::*;
    use crate::workload_api::lifecycle_test_support::*;
    use crate::WorkloadApiClient;
    use tokio::sync::oneshot;

    type Source = X509Source;
    type SourceError = X509SourceError;

    async fn build(factory: Factory, timeout: Option<Duration>) -> Source {
        Source::builder()
            .client_factory(factory)
            .shutdown_timeout(timeout)
            .build()
            .await
            .unwrap()
    }

    fn observer(source: &Source) -> tokio::task::AbortHandle {
        source.inner.supervisor.get().unwrap().abort_handle()
    }

    fn controlled(
        timeout: Option<Duration>,
    ) -> (Source, oneshot::Sender<()>, oneshot::Receiver<()>) {
        controlled_with_metrics(timeout, None)
    }

    #[expect(
        clippy::used_underscore_binding,
        reason = "Test fixture installs abort permission in the public-owner drop guard"
    )]
    fn controlled_with_metrics(
        timeout: Option<Duration>,
        metrics: Option<Arc<dyn MetricsRecorder>>,
    ) -> (Source, oneshot::Sender<()>, oneshot::Receiver<()>) {
        let svid = Arc::new(
            X509Svid::parse_from_der(
                include_bytes!("../../tests/testdata/svid/x509/1-svid-chain.der"),
                include_bytes!("../../tests/testdata/svid/x509/1-key.der"),
            )
            .unwrap(),
        );
        let mut source = Source::new_for_test(
            Arc::new(X509Context::new([svid], Arc::new(X509BundleSet::new()))),
            ReconnectConfig::default(),
            ResourceLimits::default(),
            metrics,
            None,
        );
        Arc::get_mut(&mut source.inner).unwrap().shutdown_timeout = timeout;
        let (release, released) = oneshot::channel();
        let (dropped, destroyed) = oneshot::channel();
        let guard = SupervisorTerminationGuard {
            inner: Arc::clone(&source.inner),
        };
        let drop = OnDrop(Some(dropped));
        let task = tokio::spawn(async move {
            let _guard = guard;
            let _drop = drop;
            let _result = released.await;
        });
        let task = SupervisorTask::new(task);
        source._shutdown_guard.install_abort(task.abort_handle());
        source.inner.supervisor.set(task).unwrap();
        (source, release, destroyed)
    }

    fn restore(source: &Source, initial: &Arc<Snapshot>) {
        source.inner.apply_update(Arc::clone(&initial.ctx)).unwrap();
    }

    include!("../workload_api/source_lifecycle_tests.rs");
}

#[cfg(test)]
mod startup_tests {
    use super::{
        MetricsErrorKind, MetricsRecorder, X509Source as Source, X509SourceError as SourceError,
    };
    use crate::workload_api::lifecycle_test_support::Factory;
    const RPC: &str = "FetchX509SVID";
    include!("../workload_api/source_startup_tests.rs");
}