use std::collections::{BTreeMap, HashMap, HashSet};
use std::pin::Pin;
use std::sync::{
    Arc, RwLock, RwLockReadGuard, RwLockWriteGuard,
    atomic::{AtomicBool, AtomicU64, Ordering},
};
use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};

use async_trait::async_trait;
use cgka_traits::transport::Timestamp;
use cgka_traits::{
    MemberId, MessageId, TransportAccountActivation, TransportAdapter, TransportAdapterError,
    TransportDelivery, TransportDeliveryPlane, TransportEndpoint, TransportEndpointFailure,
    TransportEndpointFailureKind, TransportEndpointRejectionCategory, TransportGroupSubscription,
    TransportGroupSync, TransportPublishFailure, TransportPublishReport, TransportPublishRequest,
};
use futures::{FutureExt, Stream, StreamExt, stream::FuturesUnordered};
use nostr_sdk::NotificationUpdate;
use nostr_sdk::prelude::{
    Client as NostrSdkClient, ClientNotification, Filter, Kind, PublicKey, RelayMessage, RelayUrl,
    SubscriptionId, Timestamp as NostrTimestamp,
};
use rand::RngCore;
use serde::{Deserialize, Serialize};
use tokio::sync::{Mutex, Notify, broadcast, mpsc};
use tokio::task::JoinHandle;
use tokio::time::{timeout, timeout_at};
use transport_nostr_adapter::{
    AccountSubscriptionEose, AccountSubscriptionInput, NostrAcquisitionCancellation,
    NostrAcquisitionError, NostrAcquisitionRequest, NostrAcquisitionResult,
    NostrNotificationLossBounds, NostrNotificationLossFloor, NostrPublishOutcome,
    NostrReconciliationItem, NostrReconciliationSummary, NostrRelayClient,
    NostrRelayNotificationSource, NostrSdkNotification, NostrSdkRelayClient, NostrSdkRelayHealth,
    NostrSubscription, NostrTransportAdapter, RelayExportConsent, RelayLabelResolution,
    RelayRegistrationOutcome, SubscriptionAttempt,
};

pub use transport_nostr_adapter::{
    DeliveryRoute, MAX_DELIVERY_LOSS_ROUTES, RelayLiveness, RelayLivenessStatus,
    ReplayAdmissionStatus, ReplayFence, ReplayFencePlan, ReplayFenceTicket,
};
use transport_nostr_peeler::NostrTransportEvent;

mod delivery_spill;
mod directory;
mod directory_wire;
mod known_limits;
#[cfg(test)]
pub(crate) mod publish_accounting_tests;
mod safety;
mod telemetry;
#[cfg(any(test, feature = "test-support"))]
pub mod test_support;

pub use known_limits::{KnownLimitPublishClient, RelayEventLimits};
use known_limits::{event_json_bytes, join_refusals, refuse_over_known_limits, refused_everywhere};
pub use safety::{RelayEndpointClassification, RelayEndpointPolicy, retired_relay_hosts};
pub use telemetry::{
    EngineReorgMetrics, RelayRollupEntry, RelayTelemetryRollup, RelayTelemetrySnapshot,
};

/// The opt-in that relay-identity resolution requires. The app's export
/// configuration implements it, so the plane does not depend on the app's
/// config types.
pub trait RelayExportGate: Send + Sync {
    /// Export is opted in with a TLS (or loopback) endpoint, authorization
    /// and resource metadata.
    fn export_allowed(&self) -> bool;
}

/// Every relay client the plane builds routes through the process SOCKS5
/// configuration: the same adapter builder `marmot-app`'s
/// `network_proxy::nostr_builder` returns, called here so the plane does not
/// depend on the app for it.
pub(crate) fn nostr_builder() -> nostr_sdk::prelude::ClientBuilder {
    transport_nostr_adapter::network_proxy::nostr_builder(
        transport_nostr_adapter::network_proxy::socks5_proxy(),
    )
}

pub use delivery_spill::{AccountDeliverySpillStore, DeliverySpillDisposition};
pub use directory::{
    DirectoryEventQuery, DirectoryFetchOutcome, DirectoryFetchRequest, DirectoryInspectionError,
    DirectoryRelayEventRecord, DirectoryRelayFetcher, DirectoryRelayPlane, DirectoryRelayStats,
    DirectorySubscriptionFilter, DirectorySubscriptionSyncSummary, DirectorySyncBatch,
    DirectorySyncPlan, NostrSdkDirectoryRelayFetcher,
};
pub use safety::{RelaySafetyPolicy, canonical_relay, recovery_required_endpoints, same_relay};
pub use telemetry::rollup_from_snapshots;

// Re-exported so the in-tree `tests` module (which uses `super::*`) keeps
// reaching these names unchanged after the split moved their only non-test
// uses into the submodules above.
#[cfg(test)]
pub use cgka_traits::TransportPublishTarget;
#[cfg(test)]
pub use transport_nostr_adapter::{
    DurationHistogramSnapshot, NostrAdapterMetrics, RelayDeliverySpread, RelaySyncSnapshot,
};

pub const ACCOUNT_DELIVERY_BUFFER: usize = 1024;
const DIRECTORY_EVENT_BUFFER: usize = 1024;
pub(crate) const DIRECTORY_RELAY_CONNECT_WAIT: Duration = Duration::from_secs(5);
const RELAY_PLANE_SHUTDOWN_WAIT: Duration = Duration::from_secs(2);
const RELAY_PLANE_TASK_ABORT_WAIT: Duration = Duration::from_millis(250);
const RELAY_NOTIFICATION_RESTART_INITIAL_BACKOFF: Duration = Duration::from_millis(100);
const RELAY_NOTIFICATION_RESTART_MAX_BACKOFF: Duration = Duration::from_secs(30);
const RELAY_NOTIFICATION_RESTART_HEALTHY_RUNTIME: Duration = Duration::from_secs(5);
/// Notifications a relay consumer may queue for its event worker before the
/// backlog counts as receiver loss. A catch-up rebuild replays every stored
/// event inside the lookback window as a raw relay copy, plus a deduplicated
/// delivery for events the SDK client has not seen, so ordinary catch-ups
/// queue far more than a few hundred. Loss keeps the account route open but
/// costs a comparison recovery pass, because the SDK client has already marked
/// the lost events seen and will not deliver them again. Match the pinned SDK
/// client's per-receiver notification buffer, which bounded the inline
/// consumer this queue replaced.
const RELAY_NOTIFICATION_EVENT_QUEUE_CAPACITY: usize = 4096;

const ACCOUNT_PUBLISH_WAIT: Duration = Duration::from_secs(20);

/// The process-wide relay plane every account shares.
///
/// Construct it inside a Tokio runtime. The constructors start the
/// notification router on the current runtime; without one, construction
/// still succeeds but nothing is routed (`spawn_router` returns early). Taking
/// an explicit runtime handle is a later change.
#[derive(Clone)]
pub struct MarmotRelayPlane {
    inner: Arc<MarmotRelayPlaneInner>,
}

struct MarmotRelayPlaneInner {
    subscription_rebuild_lookback: Option<Duration>,
    relay_safety: RelaySafetyPolicy,
    transport: Arc<RelayPlaneTransport>,
    directory: DirectoryRelayPlane,
    directory_subscription_sync: Mutex<()>,
    /// One signer-bound publisher per account. Setup publishes and the managed
    /// account worker share this client so the worker can reuse the same relay
    /// pool instead of constructing another TCP/TLS/WebSocket stack.
    account_publish_clients: std::sync::Mutex<HashMap<String, Arc<dyn NostrRelayClient>>>,
}

/// The router and notification forwarders own clones of `transport` and each waits on a channel only `transport`
/// can close, so they would keep the transport, its SDK clients and their preallocated caches alive for the life
/// of the process after the last plane handle is gone. A plane dropped without [`MarmotRelayPlane::shutdown`]
/// (every short-lived `MarmotApp`, such as one per `wnd` Execute request) leaked several MB that way.
///
/// Every task slot is locked only through a `&MarmotRelayPlane` (`spawn_router`, `set_transport_signer`, account
/// retirement, `shutdown`), and that borrow keeps this inner alive, so no slot can be held here and `try_lock`
/// always succeeds. The router and forwarders hold `transport` but never lock the slots.
impl Drop for MarmotRelayPlaneInner {
    fn drop(&mut self) {
        let transport = &self.transport;
        transport.shutting_down.store(true, Ordering::SeqCst);
        for slot in [
            &transport.router,
            &transport.notification_forwarder,
            &transport.directory_notification_forwarder,
        ] {
            if let Ok(mut slot) = slot.try_lock()
                && let Some(handle) = slot.take()
            {
                handle.abort();
            }
        }
        if let Ok(mut forwarders) = transport.account_notification_forwarders.try_lock() {
            for (_, handle) in forwarders.drain() {
                handle.abort();
            }
        }
    }
}

struct RelayPlaneTransport {
    adapter: NostrTransportAdapter,
    sdk_relay_client: Option<NostrSdkRelayClient>,
    /// An injected relay client's own relay messages
    /// ([`NostrRelayClient::relay_notification_source`]), read by the same
    /// supervised consumer as a single SDK client's.
    injected_notifications: Option<Arc<dyn NostrRelayNotificationSource>>,
    directory_client: Option<NostrSdkClient>,
    directory_events: broadcast::Sender<DirectoryRelayPlaneEvent>,
    account_deliveries: RwLock<HashMap<MemberId, AccountDeliveryRoute>>,
    /// Each account's overflow coordination, kept for the process lifetime. A
    /// spill writer can outlive its route; every later adapter for the account
    /// shares its fence, wakeup and pending loss instead of starting fresh.
    account_overflow_states: std::sync::Mutex<HashMap<MemberId, Arc<AccountDeliveryOverflowState>>>,
    account_delivery_metrics: Arc<AccountDeliveryMetrics>,
    router: Mutex<Option<JoinHandle<()>>>,
    notification_forwarder: Mutex<Option<JoinHandle<()>>>,
    account_notification_forwarders: Mutex<HashMap<MemberId, JoinHandle<()>>>,
    directory_notification_forwarder: Mutex<Option<JoinHandle<()>>>,
    notification_forwarder_health: Arc<RelayNotificationForwarderHealth>,

    shutting_down: AtomicBool,
}

#[derive(Clone, Debug)]
pub enum DirectoryRelayPlaneEvent {
    Record(DirectoryRelayEventRecord),
    RecoveryRequired,
}

#[derive(Default)]
struct RelayNotificationForwarderHealth {
    running_count: AtomicU64,
    restarts: AtomicU64,
    lag_incidents: AtomicU64,
    lagged_notifications: AtomicU64,
    panics: AtomicU64,
    unexpected_exits: AtomicU64,
}

#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
struct RelayNotificationForwarderHealthSnapshot {
    running: bool,
    restarts: u64,
    lag_incidents: u64,
    lagged_notifications: u64,
    panics: u64,
    unexpected_exits: u64,
}

#[derive(PartialEq, Eq)]
struct IncrementalActivation {
    inbox_endpoints: Vec<TransportEndpoint>,
    since: Timestamp,
}

#[derive(Clone)]
pub struct MarmotRelayPlaneAccountAdapter {
    account_id: MemberId,
    relay_plane: MarmotRelayPlane,
    publish_client: Arc<dyn NostrRelayClient>,
    anonymous_publish_client: Option<NostrSdkRelayClient>,
    delivery_rx: Arc<Mutex<mpsc::Receiver<AccountDeliveryEvent>>>,
    delivery_overflow: Arc<AccountDeliveryOverflowState>,
    /// The queue generation `delivery_rx` belongs to.
    delivery_epoch: u64,
    incremental_activation: Arc<Mutex<Option<IncrementalActivation>>>,
    /// Known relay event limits; without a source every limit is unknown.
    event_limits: Option<Arc<dyn RelayEventLimits>>,
}

#[derive(Clone)]
struct AccountDeliveryRoute {
    sender: mpsc::Sender<AccountDeliveryEvent>,
    overflow: Arc<AccountDeliveryOverflowState>,
    /// The queue generation `sender` feeds.
    epoch: u64,
    recovery_marker: Option<AccountDeliveryRecoveryMarker>,
    spill: Option<Arc<delivery_spill::AccountDeliverySpill>>,
}

/// Persists a queue-loss generation: its marker token, dropped count, the
/// earliest wire `created_at` among the dropped deliveries (`None` if any is
/// unknown), and each route they arrived on with the earliest `created_at`
/// dropped there (`None` if any had no route or time, or they touched more
/// than [`MAX_DELIVERY_LOSS_ROUTES`]). Each write covers every drop under its
/// token: the plane never writes a token whose earlier drops it did not count,
/// so new loss always raises the count.
pub type AccountDeliveryRecoveryMarker = Arc<
    dyn Fn(
            u64,
            u64,
            Option<u64>,
            Option<&[(DeliveryRoute, u64)]>,
        ) -> Result<(), AccountDeliveryRecoveryMarkerError>
        + Send
        + Sync
        + 'static,
>;

#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum AccountDeliveryRecoveryMarkerError {
    Retryable,
    Closed,
}

#[derive(Debug)]
enum AccountDeliveryEvent {
    Delivery(Box<TransportDelivery>),
    Overflow { generation: u64 },
}

/// Aggregate, privacy-safe description of one unresolved per-account queue
/// overflow generation. The account identity never leaves the private route
/// registry; callers see only counts, queue depth, and elapsed time.
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
pub struct AccountDeliveryOverflow {
    pub generation: u64,
    pub marker_token: u64,
    pub dropped: u64,
    /// Earliest wire `created_at` among this generation's dropped
    /// deliveries, or `None` when any of them is unknown.
    pub earliest_dropped: Option<u64>,
    pub notification_losses: u64,
    pub notification_token: u64,
    /// Lowest REQ `since` floor among this generation's notification or replay
    /// losses, or `None` when any loss's floor is unknown.
    pub notification_floor: Option<u64>,
    pub queue_depth: usize,
    pub elapsed_ms: u64,
}

/// Cumulative placements one account queue made, as counts only.
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
pub struct AccountDeliveryPlacementCounts {
    pub spilled_below_floor: u64,
    pub spilled_queue_full: u64,
    pub spill_already_seen: u64,
    pub queue_dropped: u64,
}

#[derive(Debug)]
pub enum AccountDeliveryReceive {
    Delivery(Box<TransportDelivery>),
    Overflow(AccountDeliveryOverflow),
}

/// What woke a waiting account consumer.
#[derive(Debug)]
pub enum AccountDeliveryWait {
    Received(AccountDeliveryReceive),
    /// Spilled rows became durable while the queue was empty.
    SpillReady,
    Closed,
}

/// Keeps one account's delivery-loss state alive and compares it by identity.
#[cfg(any(test, feature = "test-support"))]
pub struct DeliveryOverflowIdentity(Arc<AccountDeliveryOverflowState>);

#[cfg(any(test, feature = "test-support"))]
impl DeliveryOverflowIdentity {
    pub fn same_as(&self, other: &Self) -> bool {
        Arc::ptr_eq(&self.0, &other.0)
    }
}

#[derive(Default)]
struct AccountDeliveryOverflowState {
    inner: std::sync::Mutex<AccountDeliveryOverflowInner>,
    metrics: Arc<AccountDeliveryMetrics>,
    /// Notified by the spill writer after spilled rows become durable. Lives
    /// here, not on the adapter, so a replaced adapter still hears a writer
    /// that started under its predecessor.
    spill_ready: Arc<Notify>,
}

#[derive(Default)]
struct AccountDeliveryOverflowInner {
    generation: u64,
    pending: bool,
    signal_queued: bool,
    marker_signal_pending: bool,
    recovery_in_progress: bool,
    dropped: u64,
    notification_losses: u64,
    notification_token: u64,
    notification_imported: u64,
    queue_depth: usize,
    started_at: Option<Instant>,
    recovery_started_at: Option<Instant>,
    marker_token: u64,
    /// The generation carries a durable marker adopted after a restart or a
    /// failed acknowledgment, whose earlier drops this process never counted.
    /// Its first new drop takes a fresh token, so every token's count grows
    /// from zero and a new drop always raises the count storage sees.
    marker_adopted: bool,
    marker_in_progress: bool,
    marker_durable: bool,
    marker_closed: bool,
    /// Deliveries handed to the spill writer but not yet durable or lost.
    spill_in_flight: u64,
    /// A spill writer recorded loss after its route was retired, so no queue
    /// has carried its signal yet.
    retired_loss_unsignalled: bool,
    /// Running minimum of the dropped deliveries' wire `created_at`, and
    /// whether any dropped delivery's time was unknown.
    earliest_dropped: Option<u64>,
    earliest_dropped_unknown: bool,
    /// Running minimum `created_at` per route among the dropped deliveries,
    /// and whether some dropped delivery's route or time was unknown or the
    /// routes passed [`MAX_DELIVERY_LOSS_ROUTES`], which empties the map.
    dropped_routes: BTreeMap<DeliveryRoute, u64>,
    dropped_routes_unknown: bool,
    /// Running minimum of notification and replay losses' REQ floors, and
    /// whether any loss's floor was unknown.
    notification_floor: Option<u64>,
    notification_floor_unknown: bool,
    /// Running minimum REQ floor per route among those losses, and whether
    /// some loss's routes were unknown or they passed
    /// [`MAX_DELIVERY_LOSS_ROUTES`], which empties the map.
    notification_routes: BTreeMap<DeliveryRoute, u64>,
    notification_routes_unknown: bool,
    /// What the account queue holds, and the restart floors a cursor commit
    /// must keep over it.
    admission: AccountDeliveryAdmission,
}

/// The account queue as a restart would see it, and the restart floors a
/// transport-cursor commit guards. It sits under the overflow lock, which the
/// router holds while it places each delivery, so every commit's decision is
/// ordered against every placement: a delivery queued first caps the commit,
/// and one placed later sees the floor the commit raised.
///
/// Every value is a restart `since`: the group-route `since` a restart builds
/// from a cursor, which is the cursor minus the rebuild lookback. A delivery's
/// key is the lowest such `since` that still fetches it again: its
/// `created_at`, or for the inbox its `created_at` plus the NIP-59 widening
/// the inbox REQ adds.
#[derive(Debug, Default)]
struct AccountDeliveryAdmission {
    /// The queue generation `queued` describes. A replaced route's queue
    /// dies with its receiver.
    epoch: u64,
    /// Keys of the deliveries in the account queue, and of those its
    /// consumer took but has not yet ingested durably, with their counts.
    queued: BTreeMap<u64, u32>,
    /// The key in `queued` each taken delivery holds until its consumer
    /// releases it, by event. A delivery whose ingest failed is never
    /// released, so its key keeps capping commits for the rest of the queue
    /// generation, unless a redelivery of the same event is released.
    taken: HashMap<MessageId, u64>,
    /// The persisted cursor's restart `since`. Every seal raises it before
    /// its commit saves, so it covers a cursor still being written too.
    durable_since: Option<u64>,
    /// The part of `durable_since` that drain checkpoints, settled loss and
    /// the cursor the account opened with made durable, which is where a
    /// design without live promotion would have it. A delivery keyed between
    /// the two is one that only a live promotion, or a commit still saving,
    /// stopped a restart from fetching, so the router spills it instead of
    /// queueing it. `None` until the account's first settled cursor: before
    /// any cursor a restart relied on its comparison, not the cursor, for
    /// everything, so while this is `None` every delivery below
    /// `durable_since` is spilled.
    settled_since: Option<u64>,
}

impl AccountDeliveryAdmission {
    /// One delivery keyed `key` no longer caps a commit.
    fn remove_queued(&mut self, key: u64) {
        if let std::collections::btree_map::Entry::Occupied(mut count) = self.queued.entry(key) {
            if *count.get() > 1 {
                *count.get_mut() -= 1;
            } else {
                count.remove();
            }
        }
    }
}

/// Where the router puts one delivery.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
enum AccountDeliveryPlacement {
    Queue,
    /// The durable spill. The cursor fence already counts it.
    Spill,
    /// Queue loss, because no spill can take it.
    Omit,
}

/// The lowest restart `since` that fetches `delivery` again.
fn account_delivery_restart_key(delivery: &TransportDelivery) -> u64 {
    let created_at = delivery.message.timestamp.0;
    if delivery.source.plane == TransportDeliveryPlane::AccountInbox {
        created_at.saturating_add(transport_nostr_adapter::NIP59_TIMESTAMP_TWEAK_SECS)
    } else {
        created_at
    }
}

/// What one omission charges to the current loss generation.
#[derive(Clone, Copy, Debug)]
enum AccountDeliveryLossCharge<'a> {
    /// A delivery the queue omitted, with its wire `created_at` and route if
    /// known.
    Drop {
        created_at: Option<u64>,
        route: Option<DeliveryRoute>,
    },
    /// A notification or staged-replay loss, bounded by the REQ floors that
    /// could have delivered the lost history, overall and per route.
    Notification {
        loss: &'a NostrNotificationLossBounds,
    },
}

#[derive(Default)]
struct AccountDeliveryMetrics {
    max_queue_depth: AtomicU64,
    dropped: AtomicU64,
    spilled: AtomicU64,
    spill_already_seen: AtomicU64,
    /// Deliveries `place` sent to the spill because a cursor commit had
    /// raised the restart floor past them, and because the queue was full.
    /// Counted at placement; `spilled` counts what the writer stored.
    spill_diverted_below_floor: AtomicU64,
    spill_diverted_queue_full: AtomicU64,
    recovery_attempts: AtomicU64,
    recovery_successes: AtomicU64,
    recovery_failures: AtomicU64,
    recovery_elapsed_ms: AtomicU64,
}

#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
struct AccountDeliveryMetricsSnapshot {
    queue_depth: usize,
    max_queue_depth: u64,
    dropped: u64,
    spilled: u64,
    spill_already_seen: u64,
    recovery_attempts: u64,
    recovery_successes: u64,
    recovery_failures: u64,
    recovery_elapsed_ms: u64,
}

impl AccountDeliveryMetrics {
    fn snapshot(
        &self,
        routes: &RwLock<HashMap<MemberId, AccountDeliveryRoute>>,
    ) -> AccountDeliveryMetricsSnapshot {
        let queue_depth = account_deliveries_read(routes)
            .values()
            .map(|route| {
                route
                    .sender
                    .max_capacity()
                    .saturating_sub(route.sender.capacity())
            })
            .fold(0_usize, usize::saturating_add);
        AccountDeliveryMetricsSnapshot {
            queue_depth,
            max_queue_depth: self.max_queue_depth.load(Ordering::Relaxed),
            dropped: self.dropped.load(Ordering::Relaxed),
            spilled: self.spilled.load(Ordering::Relaxed),
            spill_already_seen: self.spill_already_seen.load(Ordering::Relaxed),
            recovery_attempts: self.recovery_attempts.load(Ordering::Relaxed),
            recovery_successes: self.recovery_successes.load(Ordering::Relaxed),
            recovery_failures: self.recovery_failures.load(Ordering::Relaxed),
            recovery_elapsed_ms: self.recovery_elapsed_ms.load(Ordering::Relaxed),
        }
    }
}

impl AccountDeliveryOverflowState {
    fn observe_queue_depth(&self, queue_depth: usize) {
        let depth = u64::try_from(queue_depth).unwrap_or(u64::MAX);
        self.metrics
            .max_queue_depth
            .fetch_max(depth, Ordering::Relaxed);
    }

    /// Decide where one delivery goes, under the lock every cursor commit
    /// decides under. A full queue, or a key below the durable floor and not
    /// below the settled one, sends it to the spill, or to loss when there is
    /// none. Before the first settled cursor every key below the durable
    /// floor goes there. A spilled delivery is counted in the cursor fence
    /// here, before the hand-off holds it, so no commit can pass it while it
    /// is in neither place.
    fn place(
        &self,
        epoch: u64,
        key: u64,
        queue_full: bool,
        spill: bool,
    ) -> AccountDeliveryPlacement {
        let mut state = self.inner.lock().unwrap_or_else(|p| p.into_inner());
        let admission = &mut state.admission;
        let below_durable_floor = admission.durable_since.is_some_and(|durable| {
            key < durable && admission.settled_since.is_none_or(|settled| settled <= key)
        });
        if !queue_full && !below_durable_floor {
            if admission.epoch == epoch {
                *admission.queued.entry(key).or_default() += 1;
            }
            return AccountDeliveryPlacement::Queue;
        }
        if spill {
            state.spill_in_flight = state.spill_in_flight.saturating_add(1);
            RelayNotificationForwarderHealth::increment(
                if queue_full {
                    &self.metrics.spill_diverted_queue_full
                } else {
                    &self.metrics.spill_diverted_below_floor
                },
                1,
            );
            AccountDeliveryPlacement::Spill
        } else {
            AccountDeliveryPlacement::Omit
        }
    }

    /// A delivery `place` queued never reached the queue: its send failed.
    fn unqueue(&self, epoch: u64, key: u64) {
        let mut state = self.inner.lock().unwrap_or_else(|p| p.into_inner());
        let admission = &mut state.admission;
        if admission.epoch == epoch {
            admission.remove_queued(key);
        }
    }

    /// The consumer took a queued delivery. It keeps its key, so it caps
    /// every commit until the consumer releases it: until then its ingest may
    /// still fail, and a restart must fetch it again. An event holds one key
    /// at a time, so a redelivery of one whose ingest failed takes over the
    /// pin that failure left.
    fn take(&self, epoch: u64, key: u64, id: &MessageId) {
        let mut state = self.inner.lock().unwrap_or_else(|p| p.into_inner());
        let admission = &mut state.admission;
        if admission.epoch != epoch {
            return;
        }
        let spare = match admission.taken.get_mut(id) {
            Some(held) => {
                let spare = (*held).max(key);
                *held = (*held).min(key);
                spare
            }
            None => {
                admission.taken.insert(id.clone(), key);
                return;
            }
        };
        admission.remove_queued(spare);
    }

    /// A taken delivery's ingest is durable, or its consumer dropped it on
    /// purpose, so it no longer caps a commit.
    fn release(&self, epoch: u64, id: &MessageId) {
        let mut state = self.inner.lock().unwrap_or_else(|p| p.into_inner());
        let admission = &mut state.admission;
        if admission.epoch != epoch {
            return;
        }
        if let Some(key) = admission.taken.remove(id) {
            admission.remove_queued(key);
        }
    }

    /// Start the generation of a new route's queue. The previous queue died
    /// with its receiver, and nothing in it can be taken any more. Its
    /// deliveries, and those its consumer took and never released, went with
    /// it: the new route's subscriptions start from a cursor no commit moved
    /// past them, so they fetch them again.
    fn open_queue(&self) -> u64 {
        let mut state = self.inner.lock().unwrap_or_else(|p| p.into_inner());
        let admission = &mut state.admission;
        admission.epoch = admission.epoch.wrapping_add(1);
        admission.queued.clear();
        admission.taken.clear();
        admission.epoch
    }

    /// How far a commit may promote the cursor, decided under the placement
    /// lock: `candidate`, capped at the lowest key the persisted cursor still
    /// covers of a delivery queued or taken and not yet released, or of a
    /// frame the adapter's fence holds, plus the lookback. `None` while loss
    /// or a spill hand-off is pending, and for a replaced adapter, whose
    /// queue is not the one tracked here. The restart
    /// floor rises here, before the commit's save starts, so any delivery
    /// placed while that save runs and falls below it goes to the spill. A
    /// settled commit moves the settled floor up once its save succeeds; a
    /// live one leaves it, so its spilling continues.
    fn seal_cursor(
        &self,
        epoch: u64,
        lookback: Option<u64>,
        candidate: Option<u64>,
        replay: &mut ReplayAdmissionStatus<'_>,
        authorize: impl FnOnce(&ReplayAdmissionStatus<'_>) -> bool,
    ) -> Option<u64> {
        let mut state = self.inner.lock().unwrap_or_else(|p| p.into_inner());
        replay.spill_pending = state.spill_in_flight > 0;
        if replay.pending || replay.spill_pending || !authorize(replay) {
            return None;
        }
        if state.pending || state.spill_in_flight > 0 || state.admission.epoch != epoch {
            return None;
        }
        let candidate = candidate?;
        let Some(lookback) = lookback else {
            // A full-history plane rebuilds unfloored, so no cursor can hide
            // a delivery from a restart.
            return Some(candidate);
        };
        let admission = &mut state.admission;
        let covered = admission.durable_since.unwrap_or(0);
        // A frame a fence holds in the adapter is in no queue yet: it caps
        // the commit like a queued delivery.
        let held = replay.fence_held_since(covered);
        let cap = admission
            .queued
            .range(covered..)
            .next()
            .map(|(key, _)| *key)
            .into_iter()
            .chain(held)
            .min()
            .map(|key| key.saturating_add(lookback));
        let reached = cap.map_or(candidate, |cap| candidate.min(cap));
        let since = reached.saturating_sub(lookback);
        admission.durable_since = Some(admission.durable_since.map_or(since, |d| d.max(since)));
        Some(reached)
    }

    /// A commit's save failed, so `restored` is still the persisted cursor:
    /// lower the restart floor its seal raised back to it. The router queues
    /// what that floor covers again, and the next seal is capped by it.
    /// Deliveries spilled meanwhile stay in the spill.
    fn unseal_cursor(&self, lookback: Option<u64>, restored: Option<u64>) {
        let Some(lookback) = lookback else {
            return;
        };
        let mut state = self.inner.lock().unwrap_or_else(|p| p.into_inner());
        let admission = &mut state.admission;
        admission.durable_since = restored
            .map(|restored| restored.saturating_sub(lookback))
            .max(admission.settled_since);
    }

    /// Record a cursor that is durable without live promotion: one a settled
    /// commit reached on its own, never one an earlier live promotion left
    /// persisted, so the settled floor stays at or below where a design
    /// without live promotion would have it. `opened` is the cursor the
    /// account opened with: it seeds the settled floor only when this process
    /// has none, because a reopened client inherits whatever live promotion an
    /// earlier one made.
    fn settle_cursor(&self, lookback: Option<u64>, reached: Option<u64>, opened: bool) {
        let (Some(lookback), Some(reached)) = (lookback, reached) else {
            return;
        };
        let since = reached.saturating_sub(lookback);
        let mut state = self.inner.lock().unwrap_or_else(|p| p.into_inner());
        let admission = &mut state.admission;
        admission.durable_since = Some(admission.durable_since.map_or(since, |d| d.max(since)));
        admission.settled_since = match admission.settled_since {
            Some(settled) if opened => Some(settled),
            Some(settled) => Some(settled.max(since)),
            None => Some(since),
        };
    }

    fn finish_spill(&self, settled: u64, stored: u64, already_seen: u64) {
        let mut state = self.inner.lock().unwrap_or_else(|p| p.into_inner());
        state.spill_in_flight = state.spill_in_flight.saturating_sub(settled);
        RelayNotificationForwarderHealth::increment(&self.metrics.spilled, stored);
        RelayNotificationForwarderHealth::increment(&self.metrics.spill_already_seen, already_seen);
    }

    /// Record an omitted delivery and return the generation only when this
    /// caller must enqueue the generation's control record.
    fn record_drop(
        &self,
        queue_depth: usize,
        created_at: Option<u64>,
        route: Option<DeliveryRoute>,
    ) -> Option<u64> {
        self.record_loss(
            queue_depth,
            AccountDeliveryLossCharge::Drop { created_at, route },
        )
    }

    /// Start a new generation's dropped-delivery bounds.
    fn reset_dropped(state: &mut AccountDeliveryOverflowInner) {
        state.dropped = 0;
        state.earliest_dropped = None;
        state.earliest_dropped_unknown = false;
        state.dropped_routes.clear();
        state.dropped_routes_unknown = false;
    }

    /// Record notification or replay loss with its REQ floors, returning the
    /// generation only when this caller must enqueue its control record.
    fn record_notification_loss(&self, loss: &NostrNotificationLossBounds) -> Option<u64> {
        self.record_loss(0, AccountDeliveryLossCharge::Notification { loss })
    }

    fn record_loss(&self, queue_depth: usize, charge: AccountDeliveryLossCharge) -> Option<u64> {
        let mut state = self
            .inner
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        if !state.pending {
            state.generation = state.generation.saturating_add(1);
            state.pending = true;
            Self::reset_dropped(&mut state);
            Self::reset_notification_loss(&mut state);
            state.started_at = Some(Instant::now());
            state.marker_token = rand::rngs::OsRng.next_u64() & i64::MAX as u64;
            state.marker_adopted = false;
            state.marker_in_progress = false;
            state.marker_durable = false;
            state.marker_closed = false;
        }
        match charge {
            AccountDeliveryLossCharge::Notification { loss } => {
                state.notification_losses = state.notification_losses.saturating_add(1);
                // Each loss mints a token, so a changed floor always imports as
                // new loss and bumps the obligation's revision.
                state.notification_token = rand::rngs::OsRng.next_u64() & i64::MAX as u64;
                match loss.floor.since_seconds() {
                    Some(since) => {
                        state.notification_floor =
                            Some(state.notification_floor.map_or(since, |f| f.min(since)))
                    }
                    None => state.notification_floor_unknown = true,
                }
                match &loss.routes {
                    Some(routes) if !state.notification_routes_unknown => {
                        for (route, since) in routes {
                            let floor = state.notification_routes.entry(*route).or_insert(*since);
                            *floor = (*floor).min(*since);
                        }
                        if state.notification_routes.len() > MAX_DELIVERY_LOSS_ROUTES {
                            state.notification_routes.clear();
                            state.notification_routes_unknown = true;
                        }
                    }
                    Some(_) => {}
                    None => {
                        state.notification_routes.clear();
                        state.notification_routes_unknown = true;
                    }
                }
            }
            AccountDeliveryLossCharge::Drop { created_at, route } => {
                if std::mem::take(&mut state.marker_adopted) {
                    state.marker_token = rand::rngs::OsRng.next_u64() & i64::MAX as u64;
                }
                state.dropped = state.dropped.saturating_add(1);
                match created_at {
                    Some(at) => {
                        state.earliest_dropped =
                            Some(state.earliest_dropped.map_or(at, |e| e.min(at)))
                    }
                    None => state.earliest_dropped_unknown = true,
                }
                match (route, created_at) {
                    (Some(route), Some(at)) if !state.dropped_routes_unknown => {
                        let earliest = state.dropped_routes.entry(route).or_insert(at);
                        *earliest = (*earliest).min(at);
                        if state.dropped_routes.len() > MAX_DELIVERY_LOSS_ROUTES {
                            state.dropped_routes.clear();
                            state.dropped_routes_unknown = true;
                        }
                    }
                    (Some(_), Some(_)) => {}
                    _ => {
                        state.dropped_routes.clear();
                        state.dropped_routes_unknown = true;
                    }
                }
                state.marker_durable = false;
                RelayNotificationForwarderHealth::increment(&self.metrics.dropped, 1);
            }
        }
        state.queue_depth = state.queue_depth.max(queue_depth);
        self.observe_queue_depth(queue_depth);
        if state.signal_queued {
            None
        } else {
            state.signal_queued = true;
            Some(state.generation)
        }
    }

    /// A new generation starts with no notification loss.
    fn reset_notification_loss(state: &mut AccountDeliveryOverflowInner) {
        state.notification_losses = 0;
        state.notification_token = 0;
        state.notification_imported = 0;
        state.notification_floor = None;
        state.notification_floor_unknown = false;
        state.notification_routes.clear();
        state.notification_routes_unknown = false;
    }

    fn record_retired_drop(&self, created_at: Option<u64>, route: Option<DeliveryRoute>) {
        self.record_drop(0, created_at, route);
        self.inner
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner())
            .retired_loss_unsignalled = true;
    }

    /// A control record queued on a route whose receiver is gone died with
    /// that queue, yet `signal_queued` still claims it, so `finish_recovery`
    /// could never clear the generation. Mark the signal unsent, as for loss a
    /// retired writer recorded while no route existed, so that the next route
    /// carries it. A deferred signal counts too: its marker writer holds the
    /// dead queue's sender. This assumes one live adapter per account, as the
    /// worker drops its client before it reopens.
    fn release_signal_of_closed_route(&self, sender: &mpsc::Sender<AccountDeliveryEvent>) {
        if !sender.is_closed() {
            return;
        }
        let mut state = self
            .inner
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        if state.pending && state.signal_queued {
            state.retired_loss_unsignalled = true;
        }
    }

    /// Claim the signal for loss a retired writer recorded while no route
    /// existed, or whose control record died with a replaced route's queue.
    /// Any other pending loss already had its route.
    fn claim_retired_loss_signal(&self) -> Option<u64> {
        let mut state = self
            .inner
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        if !std::mem::take(&mut state.retired_loss_unsignalled) || !state.pending {
            return None;
        }
        state.signal_queued = true;
        state.marker_signal_pending = false;
        Some(state.generation)
    }

    fn cancel_signal(&self, generation: u64) {
        let mut state = self
            .inner
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        if state.generation == generation {
            state.signal_queued = false;
            state.marker_signal_pending = false;
        }
    }

    fn defer_marker_signal(&self, generation: u64) {
        let mut state = self.inner.lock().unwrap_or_else(|p| p.into_inner());
        if state.generation == generation {
            state.marker_signal_pending = true;
        }
    }

    fn take_durable_marker_signal(&self, generation: u64) -> bool {
        let mut state = self.inner.lock().unwrap_or_else(|p| p.into_inner());
        if state.generation == generation
            && state.marker_signal_pending
            && (state.marker_durable || state.marker_closed)
            && !state.marker_in_progress
        {
            state.marker_signal_pending = false;
            true
        } else {
            false
        }
    }

    /// Claim the one account-local marker worker for the current generation.
    /// Each increased queue count must become durable. One writer aggregates
    /// changes while active; notification incidents belong to the account worker.
    fn start_marker_persistence(&self) -> bool {
        let mut state = self
            .inner
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        if state.dropped == 0
            || state.marker_durable
            || state.marker_closed
            || state.marker_in_progress
        {
            return false;
        }
        state.marker_in_progress = true;
        true
    }

    #[cfg(test)]
    fn marker_barrier_complete(&self) -> bool {
        let state = self
            .inner
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        state.dropped == 0 || state.marker_durable || state.marker_closed
    }

    /// Persist the generation marker on one account-local blocking worker while
    /// the shared relay router continues serving every other route. Retryable
    /// failures retain only this task and aggregate later omissions into the
    /// generation counter; terminal storage closure releases the task.
    async fn persist_marker_before_drop(&self, marker: AccountDeliveryRecoveryMarker) {
        let generation = {
            let state = self
                .inner
                .lock()
                .unwrap_or_else(|poisoned| poisoned.into_inner());
            state.generation
        };
        loop {
            let (marker_token, dropped, earliest, routes) = {
                let state = self
                    .inner
                    .lock()
                    .unwrap_or_else(|poisoned| poisoned.into_inner());
                (
                    state.marker_token,
                    state.dropped,
                    state
                        .earliest_dropped
                        .filter(|_| !state.earliest_dropped_unknown),
                    (!state.dropped_routes_unknown).then(|| {
                        state
                            .dropped_routes
                            .iter()
                            .map(|(route, at)| (*route, *at))
                            .collect::<Vec<_>>()
                    }),
                )
            };
            let marker = marker.clone();
            match tokio::task::spawn_blocking(move || {
                marker(marker_token, dropped, earliest, routes.as_deref())
            })
            .await
            {
                Ok(Ok(())) => {
                    let mut state = self
                        .inner
                        .lock()
                        .unwrap_or_else(|poisoned| poisoned.into_inner());
                    if state.generation == generation {
                        if state.dropped != dropped {
                            continue;
                        }
                        state.marker_in_progress = false;
                        state.marker_durable = true;
                    }
                    return;
                }
                Ok(Err(AccountDeliveryRecoveryMarkerError::Closed)) => {
                    let mut state = self
                        .inner
                        .lock()
                        .unwrap_or_else(|poisoned| poisoned.into_inner());
                    if state.generation == generation {
                        state.marker_in_progress = false;
                        state.marker_closed = true;
                        tracing::debug!(
                            target: "marmot_app::relay_plane",
                            method = "persist_marker_before_drop",
                            error_kind = "storage_closed",
                            "account delivery overflow marker worker stopped after storage closure",
                        );
                    }
                    return;
                }
                Ok(Err(AccountDeliveryRecoveryMarkerError::Retryable)) | Err(_) => {
                    tracing::warn!(
                        target: "marmot_app::relay_plane",
                        method = "persist_marker_before_drop",
                        error_kind = "overflow_marker_persist_failed",
                        "durable account delivery overflow marker write failed; retrying",
                    );
                    tokio::time::sleep(Duration::from_millis(100)).await;
                }
            }
            let state = self
                .inner
                .lock()
                .unwrap_or_else(|poisoned| poisoned.into_inner());
            if state.generation != generation {
                return;
            }
        }
    }

    fn notification_persisted(&self, observed: AccountDeliveryOverflow) {
        let mut state = self
            .inner
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        if state.generation == observed.generation
            && state.notification_token == observed.notification_token
        {
            state.notification_imported = observed.notification_losses;
        }
    }

    fn pending_snapshot(&self) -> Option<AccountDeliveryOverflow> {
        let state = self
            .inner
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        (state.pending && !state.recovery_in_progress).then(|| Self::snapshot(&state))
    }

    fn consume_signal(&self, generation: u64) -> AccountDeliveryOverflow {
        let mut state = self
            .inner
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        if state.generation == generation {
            state.signal_queued = false;
        }
        Self::snapshot(&state)
    }

    fn start_recovery(&self, durable_marker_token: u64) -> AccountDeliveryOverflow {
        let mut state = self
            .inner
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        // A durable marker can outlive the process-local route that detected
        // it. Recreate a synthetic pending generation on reopen so completion
        // still has a compare-and-clear guard against new local overflow.
        if !state.pending {
            state.generation = state.generation.saturating_add(1);
            state.pending = true;
            Self::reset_dropped(&mut state);
            // The resolved generation's lags must not charge this one's floor.
            Self::reset_notification_loss(&mut state);
            state.queue_depth = 0;
            state.started_at = Some(Instant::now());
            // This path exists only because the durable database marker was
            // loaded after a restart; the process-local generation is already
            // covered before its first recovery subscription is issued.
            state.marker_token = durable_marker_token;
            state.marker_adopted = true;
            state.marker_durable = true;
            state.marker_closed = false;
        }
        state.recovery_in_progress = true;
        state.recovery_started_at = Some(Instant::now());
        RelayNotificationForwarderHealth::increment(&self.metrics.recovery_attempts, 1);
        Self::snapshot(&state)
    }

    fn restore_recovery_guard(&self, durable_marker_token: u64) {
        let mut state = self
            .inner
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        if !state.pending {
            state.generation = state.generation.saturating_add(1);
            state.pending = true;
            Self::reset_dropped(&mut state);
            Self::reset_notification_loss(&mut state);
            state.queue_depth = 0;
            state.started_at = Some(Instant::now());
            state.marker_token = durable_marker_token;
            state.marker_adopted = true;
            state.marker_durable = true;
            state.marker_closed = false;
            state.recovery_in_progress = false;
        }
    }

    /// Whether `observed` is still exactly the pending generation, with every
    /// omission it counts durable and imported and no control record queued,
    /// so the account owner may settle it.
    fn settles(state: &AccountDeliveryOverflowInner, observed: &AccountDeliveryOverflow) -> bool {
        state.pending
            && state.generation == observed.generation
            && state.marker_token == observed.marker_token
            && state.dropped == observed.dropped
            && state.notification_losses == observed.notification_losses
            && state.notification_imported == state.notification_losses
            && (state.dropped == 0 || state.marker_durable)
            && !state.marker_in_progress
            && !state.marker_closed
            && !state.signal_queued
    }

    fn finish_recovery(&self, attempt: AccountDeliveryOverflow) -> Option<u64> {
        let mut state = self
            .inner
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        let resolved = Self::settles(&state, &attempt);
        if resolved {
            state.pending = false;
            state.recovery_in_progress = false;
            state.marker_in_progress = false;
            state.marker_durable = false;
            state.marker_closed = false;
            state.started_at = None;
            let elapsed_ms = state
                .recovery_started_at
                .take()
                .map(|started| started.elapsed().as_millis() as u64)
                .unwrap_or_default();
            return Some(elapsed_ms);
        }
        None
    }

    /// Clear the generation the account owner retired as "history may be
    /// incomplete", under the same exact-generation guard as `finish_recovery`.
    /// It is not a recovery success: no success is counted, and an attempt
    /// still in flight records its own failure when it ends. Returns true when
    /// no generation is pending any more.
    fn retire_recovery(&self, observed: Option<AccountDeliveryOverflow>) -> bool {
        let mut state = self
            .inner
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        if !state.pending {
            return true;
        }
        if !observed.is_some_and(|observed| Self::settles(&state, &observed)) {
            return false;
        }
        state.pending = false;
        state.marker_durable = false;
        state.marker_closed = false;
        state.started_at = None;
        true
    }

    fn pending_generation(&self) -> Option<AccountDeliveryOverflow> {
        let state = self
            .inner
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        state.pending.then(|| Self::snapshot(&state))
    }

    fn record_recovery_success(&self, elapsed_ms: u64) {
        RelayNotificationForwarderHealth::increment(&self.metrics.recovery_successes, 1);
        RelayNotificationForwarderHealth::increment(&self.metrics.recovery_elapsed_ms, elapsed_ms);
    }

    fn fail_recovery(&self) {
        let mut state = self
            .inner
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        state.recovery_in_progress = false;
        let elapsed_ms = state
            .recovery_started_at
            .take()
            .map(|started| started.elapsed().as_millis() as u64)
            .unwrap_or_default();
        RelayNotificationForwarderHealth::increment(&self.metrics.recovery_failures, 1);
        RelayNotificationForwarderHealth::increment(&self.metrics.recovery_elapsed_ms, elapsed_ms);
    }

    fn blocks_ordinary_eose(&self) -> bool {
        let state = self
            .inner
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        state.pending && !state.recovery_in_progress
    }

    fn snapshot(state: &AccountDeliveryOverflowInner) -> AccountDeliveryOverflow {
        AccountDeliveryOverflow {
            generation: state.generation,
            marker_token: state.marker_token,
            dropped: state.dropped,
            earliest_dropped: state
                .earliest_dropped
                .filter(|_| !state.earliest_dropped_unknown),
            notification_losses: state.notification_losses,
            notification_token: state.notification_token,
            notification_floor: state
                .notification_floor
                .filter(|_| !state.notification_floor_unknown),
            queue_depth: state.queue_depth,
            elapsed_ms: state
                .started_at
                .map(|started| started.elapsed().as_millis() as u64)
                .unwrap_or_default(),
        }
    }
}

#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
pub struct RelayPlaneHealth {
    pub sdk_backed: bool,
    pub total_relays: usize,
    pub initialized: usize,
    pub pending: usize,
    pub connecting: usize,
    pub connected: usize,
    pub disconnected: usize,
    pub terminated: usize,
    pub banned: usize,
    pub sleeping: usize,
    pub connection_attempts: usize,
    pub connection_successes: usize,
    pub notification_forwarder_running: bool,
    /// Relay notification consumers restarted after a lag or an unexpected
    /// exit. A lag resumes the same SDK receiver and leaves account delivery
    /// open; an unexpected exit replaces the receiver and closes account
    /// delivery, so the account worker reconnects.
    pub notification_forwarder_restarts: u64,
    /// Consumer lags, each recorded as notification loss on the affected
    /// accounts and recovered by comparison.
    pub notification_forwarder_lag_incidents: u64,
    /// Notifications those lags skipped or abandoned. A count of
    /// notifications, including raw relay copies, not of lost events.
    pub notification_forwarder_lagged_notifications: u64,
    pub notification_forwarder_panics: u64,
    pub notification_forwarder_unexpected_exits: u64,
    /// Current queued account-delivery records across active accounts.
    #[serde(default)]
    pub account_delivery_queue_depth: usize,
    /// High-water queue depth for any account since this plane started.
    #[serde(default)]
    pub account_delivery_max_queue_depth: u64,
    /// Deliveries neither queued nor spilled, and covered by an explicit
    /// recovery generation instead.
    #[serde(default)]
    pub account_delivery_dropped: u64,
    /// Deliveries stored in the durable account spill instead of the queue:
    /// because it was full, or because a transport-cursor checkpoint still
    /// saving, or a live promotion alone, had put them below the floor a
    /// restart fetches from.
    #[serde(default)]
    pub account_delivery_spilled: u64,
    /// Spill candidates discarded because the account had already seen them.
    #[serde(default)]
    pub account_delivery_spill_already_seen: u64,
    #[serde(default)]
    pub account_delivery_recovery_attempts: u64,
    #[serde(default)]
    pub account_delivery_recovery_successes: u64,
    #[serde(default)]
    pub account_delivery_recovery_failures: u64,
    /// Aggregate wall-clock time spent in completed/failed recovery attempts.
    #[serde(default)]
    pub account_delivery_recovery_elapsed_ms: u64,
    pub directory_inflight_fetches: usize,
    pub directory_active_subscriptions: usize,
    #[serde(default)]
    pub directory_auth_required_routes: usize,
    pub directory_completed_fetches: usize,
    pub directory_coalesced_waiters: usize,
    pub directory_failed_fetches: usize,
    pub directory_completed_subscription_syncs: usize,
    pub directory_subscriptions_created: usize,
    pub directory_subscriptions_removed: usize,
}

impl MarmotRelayPlane {
    /// Call inside a Tokio runtime; see [`MarmotRelayPlane`].
    pub fn runtime_default(subscription_rebuild_lookback: Duration) -> Self {
        Self::from_sdk(Some(subscription_rebuild_lookback), false)
    }

    /// Production runtime plane whose relay-safety chokepoint admits loopback
    /// endpoints only when `allow_loopback` is set
    /// (`MarmotAppConfig::allow_loopback_relay_endpoints`, off by default).
    ///
    /// Call inside a Tokio runtime; see [`MarmotRelayPlane`].
    pub fn runtime_default_with_loopback(
        subscription_rebuild_lookback: Duration,
        allow_loopback: bool,
    ) -> Self {
        Self::from_sdk(Some(subscription_rebuild_lookback), allow_loopback)
    }

    /// Call inside a Tokio runtime; see [`MarmotRelayPlane`].
    pub fn full_history() -> Self {
        Self::from_sdk(None, false)
    }

    /// Full-history plane whose relay-safety chokepoint admits loopback
    /// endpoints only when `allow_loopback` is set
    /// (`MarmotAppConfig::allow_loopback_relay_endpoints`, off by default).
    ///
    /// Call inside a Tokio runtime; see [`MarmotRelayPlane`].
    pub fn full_history_with_loopback(allow_loopback: bool) -> Self {
        Self::from_sdk(None, allow_loopback)
    }

    /// Call inside a Tokio runtime; see [`MarmotRelayPlane`].
    pub fn with_subscription_rebuild_lookback(lookback: Duration) -> Self {
        Self::from_sdk(Some(lookback), false)
    }

    /// Call inside a Tokio runtime; see [`MarmotRelayPlane`].
    pub fn new(
        subscription_rebuild_lookback: Option<Duration>,
        relay_client: Arc<dyn NostrRelayClient>,
    ) -> Self {
        Self::new_with_loopback(subscription_rebuild_lookback, relay_client, false)
    }

    /// Call inside a Tokio runtime; see [`MarmotRelayPlane`].
    #[doc(hidden)]
    pub fn new_with_loopback(
        subscription_rebuild_lookback: Option<Duration>,
        relay_client: Arc<dyn NostrRelayClient>,
        allow_loopback: bool,
    ) -> Self {
        let notifications = relay_client.relay_notification_source();
        let adapter = NostrTransportAdapter::new(relay_client);
        Self::from_adapter(
            subscription_rebuild_lookback,
            adapter,
            None,
            notifications,
            None,
            Arc::new(NostrSdkDirectoryRelayFetcher::standalone()),
            allow_loopback,
        )
    }

    #[cfg(any(test, feature = "test-support"))]
    #[doc(hidden)]
    pub fn new_with_directory_fetcher_for_test(
        subscription_rebuild_lookback: Option<Duration>,
        relay_client: Arc<dyn NostrRelayClient>,
        directory_fetcher: Arc<dyn DirectoryRelayFetcher>,
        allow_loopback: bool,
    ) -> Self {
        let notifications = relay_client.relay_notification_source();
        Self::from_adapter(
            subscription_rebuild_lookback,
            NostrTransportAdapter::new(relay_client),
            None,
            notifications,
            None,
            directory_fetcher,
            allow_loopback,
        )
    }

    fn from_sdk(subscription_rebuild_lookback: Option<Duration>, allow_loopback: bool) -> Self {
        let directory_client = directory::anonymous_directory_client();
        let relay_client = NostrSdkRelayClient::multi_account();
        let adapter = NostrTransportAdapter::new(Arc::new(relay_client.clone()));
        Self::from_adapter(
            subscription_rebuild_lookback,
            adapter,
            Some(relay_client),
            None,
            Some(directory_client.clone()),
            Arc::new(NostrSdkDirectoryRelayFetcher::standalone()),
            allow_loopback,
        )
    }

    fn from_adapter(
        subscription_rebuild_lookback: Option<Duration>,
        adapter: NostrTransportAdapter,
        sdk_relay_client: Option<NostrSdkRelayClient>,
        injected_notifications: Option<Arc<dyn NostrRelayNotificationSource>>,
        directory_client: Option<NostrSdkClient>,
        directory_fetcher: Arc<dyn DirectoryRelayFetcher>,
        allow_loopback: bool,
    ) -> Self {
        let transport = Arc::new(RelayPlaneTransport {
            adapter,
            sdk_relay_client,
            injected_notifications,
            directory_client,
            directory_events: broadcast::channel(DIRECTORY_EVENT_BUFFER).0,
            account_deliveries: RwLock::new(HashMap::new()),
            account_overflow_states: std::sync::Mutex::new(HashMap::new()),
            account_delivery_metrics: Arc::new(AccountDeliveryMetrics::default()),
            router: Mutex::new(None),
            notification_forwarder: Mutex::new(None),
            account_notification_forwarders: Mutex::new(HashMap::new()),
            directory_notification_forwarder: Mutex::new(None),
            notification_forwarder_health: Arc::new(RelayNotificationForwarderHealth::default()),
            shutting_down: AtomicBool::new(false),
        });
        let this = Self {
            inner: Arc::new(MarmotRelayPlaneInner {
                subscription_rebuild_lookback,
                relay_safety: RelaySafetyPolicy::with_allow_loopback(allow_loopback),
                transport,
                directory: DirectoryRelayPlane::new(directory_fetcher),
                directory_subscription_sync: Mutex::new(()),
                account_publish_clients: std::sync::Mutex::new(HashMap::new()),
            }),
        };
        this.spawn_router();
        this
    }

    /// The account's signer-bound publish client, built on first use with the
    /// plane's proxy-aware builder and shared by every later caller.
    ///
    /// The signer is ignored on a cache hit. A caller that replaces an
    /// account's signer must call [`Self::evict_account_publish_client`]
    /// first; registering an external signer and dropping an account's
    /// caches do.
    #[doc(hidden)]
    pub fn account_publish_client(
        &self,
        account_id_hex: &str,
        signer: Arc<dyn transport_nostr_peeler::MarmotNostrSigner>,
    ) -> Arc<dyn NostrRelayClient> {
        let mut clients = self
            .inner
            .account_publish_clients
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        clients
            .entry(account_id_hex.to_owned())
            .or_insert_with(|| {
                let client = nostr_builder()
                    .authenticator(nostr_sdk::authenticator::SignerAuthenticator::new(
                        transport_nostr_peeler::SdkSigner(signer.clone()),
                    ))
                    .build();
                let account_id =
                    MemberId::new(hex::decode(account_id_hex).expect("validated account identity"));
                Arc::new(NostrSdkRelayClient::with_account_signer(
                    client, account_id, signer,
                ))
            })
            .clone()
    }

    /// Forgets the account's publish client, so the next
    /// [`Self::account_publish_client`] builds one with the current signer.
    #[doc(hidden)]
    pub fn evict_account_publish_client(&self, account_id_hex: &str) {
        self.inner
            .account_publish_clients
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner())
            .remove(account_id_hex);
    }

    /// Build an account adapter without durable queue-overflow recovery.
    ///
    /// Production callers must use `account_adapter_with_recovery_marker` with
    /// a marker. This compatibility constructor is retained for tests and
    /// embedders that do not persist account projection state.
    pub fn account_adapter(
        &self,
        account_id: MemberId,
        publish_client: Arc<dyn NostrRelayClient>,
    ) -> MarmotRelayPlaneAccountAdapter {
        self.account_adapter_with_recovery_marker(account_id, publish_client, None, None)
    }

    #[doc(hidden)]
    pub fn account_adapter_with_recovery_marker(
        &self,
        account_id: MemberId,
        publish_client: Arc<dyn NostrRelayClient>,
        recovery_marker: Option<AccountDeliveryRecoveryMarker>,
        spill_store: Option<AccountDeliverySpillStore>,
    ) -> MarmotRelayPlaneAccountAdapter {
        self.spawn_router();
        // Keep one slot reserved for the overflow control record. Ordinary
        // deliveries stop at ACCOUNT_DELIVERY_BUFFER, so a full account queue
        // can always carry the explicit recovery signal without awaiting the
        // slow consumer or blocking the shared router.
        let (delivery_tx, delivery_rx) = mpsc::channel(ACCOUNT_DELIVERY_BUFFER + 1);
        let signal_tx = delivery_tx.clone();
        let mut routes = account_deliveries_write(&self.inner.transport.account_deliveries);
        let delivery_overflow = self
            .inner
            .transport
            .account_overflow_states
            .lock()
            .unwrap_or_else(|p| p.into_inner())
            .entry(account_id.clone())
            .or_insert_with(|| {
                Arc::new(AccountDeliveryOverflowState {
                    inner: std::sync::Mutex::new(AccountDeliveryOverflowInner::default()),
                    metrics: self.inner.transport.account_delivery_metrics.clone(),
                    spill_ready: Arc::default(),
                })
            })
            .clone();
        // The route lock orders the new generation against the router, which
        // reads a route's generation with the route itself.
        let delivery_epoch = delivery_overflow.open_queue();
        let replaced = routes.insert(
            account_id.clone(),
            AccountDeliveryRoute {
                sender: delivery_tx,
                overflow: delivery_overflow.clone(),
                epoch: delivery_epoch,
                spill: spill_store.map(|store| {
                    delivery_spill::AccountDeliverySpill::new(
                        store,
                        account_id.clone(),
                        Arc::downgrade(&self.inner.transport),
                        delivery_overflow.clone(),
                        recovery_marker.clone(),
                    )
                }),
                recovery_marker,
            },
        );
        if let Some(replaced) = replaced {
            delivery_overflow.release_signal_of_closed_route(&replaced.sender);
        }
        // Loss a retired writer recorded while no route existed has not been
        // signalled to any consumer, nor has a record that died with the
        // replaced route's queue. The route lock orders this against that
        // writer's own check for a route.
        if let Some(generation) = delivery_overflow.claim_retired_loss_signal() {
            enqueue_account_delivery_overflow_signal(&signal_tx, &delivery_overflow, generation);
        }
        drop(routes);
        // Never share a socket with the authenticated receive pool or Welcome
        // publisher. Adapter clones retain this account's anonymous WRITE pool.
        let anonymous_publish_client = self
            .inner
            .transport
            .sdk_relay_client
            .as_ref()
            .map(|_| NostrSdkRelayClient::from_builder(nostr_builder()));
        MarmotRelayPlaneAccountAdapter {
            account_id,
            relay_plane: self.clone(),
            publish_client,
            anonymous_publish_client,
            delivery_rx: Arc::new(Mutex::new(delivery_rx)),
            delivery_overflow,
            delivery_epoch,
            incremental_activation: Arc::new(Mutex::new(None)),
            event_limits: None,
        }
    }

    /// Place one delivery exactly as the router does, synchronously, so a
    /// test can land it at a chosen point inside a cursor commit.
    #[cfg(any(test, feature = "test-support"))]
    #[doc(hidden)]
    pub fn route_account_delivery_for_test(&self, delivery: TransportDelivery) {
        route_account_delivery(&self.inner.transport, delivery);
    }

    #[doc(hidden)]
    pub fn sanitize_relay_endpoints(
        &self,
        endpoints: Vec<TransportEndpoint>,
        context: &str,
    ) -> Result<Vec<TransportEndpoint>, String> {
        self.inner
            .relay_safety
            .sanitize_endpoints(endpoints, context)
    }

    /// Check the production relay dial policy before a history request may
    /// register an endpoint absent from this account's current SDK pool.
    pub async fn acquire_history(
        &self,
        request: NostrAcquisitionRequest,
        cancellation: NostrAcquisitionCancellation,
    ) -> Result<NostrAcquisitionResult, NostrAcquisitionError> {
        request.validate()?;
        let checked = self
            .inner
            .relay_safety
            .sanitize_endpoints(request.endpoints.clone(), "history acquisition")
            .map_err(|_| NostrAcquisitionError::InvalidRequest)?;
        if checked.len() != request.endpoints.len() {
            return Err(NostrAcquisitionError::InvalidRequest);
        }
        self.inner
            .transport
            .adapter
            .acquire_history(request, cancellation)
            .await
    }

    /// Classify caller-owned relay URLs under the exact policy used at every
    /// relay-plane dial boundary. Results preserve input order and cardinality.
    pub fn classify_relay_endpoints(
        &self,
        endpoints: Vec<String>,
    ) -> Vec<RelayEndpointClassification> {
        self.inner.relay_safety.classify_endpoints(endpoints)
    }

    pub fn subscription_rebuild_since(
        &self,
        last_transport_timestamp: Option<u64>,
    ) -> Option<Timestamp> {
        let lookback = self.inner.subscription_rebuild_lookback?;
        let last_transport_timestamp = last_transport_timestamp?;
        // The persisted cursor is advanced from the sender-controlled inbound
        // `created_at`; a far-future value would push `since` past the present,
        // so relays return no present-dated events and reception silently halts
        // forever (the cursor is persisted and monotonic, so it survives
        // restarts — mdk#182).
        let now = SystemTime::now()
            .duration_since(UNIX_EPOCH)
            .unwrap_or_default()
            .as_secs();
        // A cursor detectably in the future is corrupted, not authoritative.
        // Merely clamping it to wall-clock would yield `since = now - lookback`
        // and permanently skip any valid backlog older than the (short,
        // production-default 120s) lookback for an account whose cursor was
        // poisoned before the write-side clamp existed. Treat it as untrusted
        // and request a full-history replay (`None`) so the catch-up range is
        // never silently dropped; the write side then heals the stored value
        // back below wall-clock. A cursor at or behind wall-clock is trusted
        // and used as-is.
        if last_transport_timestamp > now {
            return None;
        }
        Some(Timestamp(
            last_transport_timestamp.saturating_sub(lookback.as_secs()),
        ))
    }

    /// The subscription-rebuild lookback in seconds, if this plane rebuilds
    /// from the durable cursor. `None` means the plane rebuilds with full
    /// history (no `since` floor). Surfaced for the `subscription_rebuild`
    /// forensic audit row so an analyzer sees the window subtracted from the
    /// cursor to derive the `since` floor.
    pub fn subscription_rebuild_lookback_secs(&self) -> Option<u64> {
        self.inner
            .subscription_rebuild_lookback
            .map(|lookback| lookback.as_secs())
    }

    /// Take the finished lag-lost EOSE repair reports this account records:
    /// its own receiver's, and, when `shared` is set, those of a receiver
    /// shared across accounts, which the first account to take them records
    /// once. An account with no v5 recorder takes only its own, so it never
    /// consumes a shared report another account could record.
    /// Drain the relay refusals of `account`'s live subscriptions since the
    /// last drain, for its `subscription_refused` audit row.
    #[doc(hidden)]
    pub fn take_subscription_refusals(
        &self,
        account: &MemberId,
    ) -> transport_nostr_adapter::SubscriptionRefusalReports {
        self.inner
            .transport
            .adapter
            .take_subscription_refusals(account)
    }

    /// Drain the per-relay subscription-registration outcomes `account`
    /// accumulated since its previous drain, for its `subscription_rebuild`
    /// forensic audit row.
    ///
    /// Delegates to the SDK relay client, which records each subscribe's
    /// per-endpoint acceptance bucketed by account. The drain is account-scoped
    /// so concurrent account workers sharing this one relay plane each attribute
    /// their own registrations to their own audit row; a group shared across
    /// accounts registers once, attributed to whichever account's client
    /// subscribed (an acceptable diagnostic attribution). A plane built on a
    /// custom (non-SDK) relay client does not track registration outcomes, so
    /// this returns empty for it — the audit row then carries `since`/`lookback`
    /// without relay rows.
    pub async fn take_subscription_registrations(
        &self,
        account: &MemberId,
    ) -> Vec<RelayRegistrationOutcome> {
        if let Some(sdk_relay_client) = &self.inner.transport.sdk_relay_client {
            return sdk_relay_client
                .take_subscription_registrations(account)
                .await;
        }
        Vec::new()
    }

    /// Register the immutable NIP-42 context before any account subscription.
    /// A custom injected relay client owns its own authentication behavior.
    pub async fn set_transport_signer(
        &self,
        account_id: &MemberId,
        signer: Arc<dyn transport_nostr_peeler::MarmotNostrSigner>,
    ) -> Result<(), TransportAdapterError> {
        if let Some(sdk_relay_client) = &self.inner.transport.sdk_relay_client {
            let account_client = sdk_relay_client
                .register_account(account_id.clone(), signer)
                .await?;
            let mut forwarders = self
                .inner
                .transport
                .account_notification_forwarders
                .lock()
                .await;
            if forwarders
                .get(account_id)
                .is_none_or(JoinHandle::is_finished)
                && !self.inner.transport.shutting_down.load(Ordering::SeqCst)
            {
                if forwarders.remove(account_id).is_some() {
                    recover_relay_notification_forwarder_scoped(
                        &self.inner.transport,
                        RelayNotificationConsumerExit::Closed,
                        NostrNotificationLossBounds::account_wide(
                            NostrNotificationLossFloor::Unbounded,
                        ),
                        Some(account_id),
                    );
                }
                forwarders.insert(
                    account_id.clone(),
                    spawn_relay_notification_forwarder_scoped(
                        account_client,
                        self.inner.transport.clone(),
                        Some(account_id.clone()),
                    ),
                );
            }
        }
        Ok(())
    }

    /// Retire one account session's immutable SDK authenticator and its
    /// notification forwarder before the same worker reopens that account. The
    /// reopened session registers a new signer, which the SDK refuses while
    /// the previous context is live. Keep the delivery route so the
    /// replacement adapter inherits unresolved loss evidence.
    ///
    /// This runs on the reopening worker and never outlives it. If the worker
    /// is cancelled first, the reaper's `deactivate_account_context` retires
    /// the same context before a replacement may register; detached work here
    /// could instead remove that replacement's context.
    #[doc(hidden)]
    pub async fn retire_account_session_transport(&self, account_id: &MemberId) {
        let forwarder = self
            .inner
            .transport
            .account_notification_forwarders
            .lock()
            .await
            .remove(account_id);
        if let Some(mut forwarder) = forwarder {
            forwarder.abort();
            let _ = timeout(RELAY_PLANE_TASK_ABORT_WAIT, &mut forwarder).await;
        }
        if let Some(sdk) = &self.inner.transport.sdk_relay_client {
            let _ = timeout(RELAY_PLANE_SHUTDOWN_WAIT, sdk.remove_account(account_id)).await;
        }
    }

    /// Retire a stopped account worker's subscriptions and immutable SDK
    /// authenticator before a replacement worker can register a fresh signer.
    #[doc(hidden)]
    pub async fn deactivate_account_context(
        &self,
        account_id: &MemberId,
    ) -> Result<(), TransportAdapterError> {
        let removed =
            account_deliveries_write(&self.inner.transport.account_deliveries).remove(account_id);
        if let Some(removed) = removed {
            // Once the worker has stopped, its queue is gone, and so is any
            // control record in it.
            removed
                .overflow
                .release_signal_of_closed_route(&removed.sender);
        }
        self.inner
            .transport
            .adapter
            .deactivate_account(account_id)
            .await?;
        if let Some(sdk) = &self.inner.transport.sdk_relay_client {
            sdk.remove_account(account_id).await;
        }
        if let Some(handle) = self
            .inner
            .transport
            .account_notification_forwarders
            .lock()
            .await
            .remove(account_id)
        {
            handle.abort();
        }
        Ok(())
    }

    pub async fn relay_health(&self) -> RelayPlaneHealth {
        let directory = self.inner.directory.stats().await;
        let forwarder = self
            .inner
            .transport
            .notification_forwarder_health
            .snapshot();
        let account_delivery = self
            .inner
            .transport
            .account_delivery_metrics
            .snapshot(&self.inner.transport.account_deliveries);
        if let Some(sdk_relay_client) = &self.inner.transport.sdk_relay_client {
            return RelayPlaneHealth::from_sdk(
                sdk_relay_client.relay_health().await,
                directory,
                forwarder,
                account_delivery,
            );
        }
        RelayPlaneHealth::from_directory(directory, account_delivery)
    }

    /// Wake local relay-state readers on status or account-pool changes.
    pub fn relay_liveness_changes(&self) -> Option<tokio::sync::watch::Receiver<u64>> {
        self.inner
            .transport
            .sdk_relay_client
            .as_ref()
            .map(NostrSdkRelayClient::liveness_changes)
    }

    /// Reconnect within the account's own pool after applying the dial policy.
    pub async fn reconnect_relay(
        &self,
        account_id: &MemberId,
        endpoint: TransportEndpoint,
    ) -> Result<RelayLiveness, TransportAdapterError> {
        if self.inner.transport.shutting_down.load(Ordering::SeqCst) {
            return Err(TransportAdapterError::Subscription(
                "relay plane is shutting down".into(),
            ));
        }
        let endpoint = self
            .inner
            .relay_safety
            .sanitize_endpoints(vec![endpoint], "relay reconnect")
            .map_err(TransportAdapterError::Subscription)?
            .pop()
            .expect("one sanitized relay endpoint");
        let client = self
            .account_history_relay_client(account_id)
            .await
            .ok_or_else(|| {
                TransportAdapterError::Subscription("account relay pool is unavailable".into())
            })?;
        client.reconnect_relay(&endpoint).await
    }

    /// Snapshot the device-local relay telemetry for local inspection.
    ///
    /// Aggregate and privacy-safe: counts, millisecond histogram buckets, and
    /// opaque relay indices only. There is a single shared adapter per device,
    /// so these counters already span every local account. Resolving the opaque
    /// indices to relay URLs is reserved for the opt-in export path.
    pub async fn relay_telemetry(&self) -> RelayTelemetrySnapshot {
        let adapter = &self.inner.transport.adapter;
        RelayTelemetrySnapshot {
            metrics: adapter.metrics().await,
            delivery_spread: adapter.delivery_spread().await,
            sync: adapter.relay_sync().await,
            health: self.relay_health().await,
        }
    }

    /// Resolve opaque relay indices to relay endpoints — the export label
    /// boundary.
    ///
    /// Crate-private and reachable only through the exporter. It returns `None`
    /// unless [`RelayExportGate::export_allowed`] holds (the same gate the
    /// app's exporter constructor applies); only then does it mint
    /// a [`RelayExportConsent`] and ask the adapter to reverse-map indices to
    /// relay URLs. No other code path turns a device-local index into a relay
    /// URL. See the privacy contract in `relay-observability.md`.
    #[doc(hidden)]
    pub async fn resolve_relay_labels(
        &self,
        gate: &dyn RelayExportGate,
    ) -> Option<RelayLabelResolution> {
        // Same gate as the exporter constructor: resolution cannot happen
        // unless export is opted in with a TLS/loopback endpoint, auth, and
        // resource metadata.
        if !gate.export_allowed() {
            return None;
        }
        let consent = RelayExportConsent::affirm();
        Some(
            self.inner
                .transport
                .adapter
                .resolve_relay_labels(consent)
                .await,
        )
    }

    /// Aggregate the device-local per-relay telemetry into one export-ready
    /// rollup, optionally folding in engine-side reorg metrics.
    ///
    /// Keyed by opaque relay index — no relay URLs. The single shared adapter
    /// already merges across local accounts, so today this is a near-passthrough
    /// reshaping; it is the seam where multi-account dedup and engine metrics are
    /// combined for export. `engine` is `None` until the parallel
    /// `observed_reorg_rate` workstream lands.
    pub async fn telemetry_rollup(
        &self,
        engine: Option<EngineReorgMetrics>,
    ) -> RelayTelemetryRollup {
        let adapter = &self.inner.transport.adapter;
        let spread = adapter.delivery_spread().await;
        let sync = adapter.relay_sync().await;
        let metrics = adapter.metrics().await;
        let health = self.relay_health().await;
        rollup_from_snapshots(spread, sync, metrics, health, engine)
    }

    #[doc(hidden)]
    pub async fn fetch_directory_events(
        &self,
        endpoints: Vec<TransportEndpoint>,
        queries: Vec<DirectoryEventQuery>,
    ) -> Result<Vec<DirectoryRelayEventRecord>, String> {
        let endpoints = self
            .inner
            .relay_safety
            .sanitize_endpoints(endpoints, "directory fetch")?;
        self.inner
            .directory
            .fetch_events(DirectoryFetchRequest::new(endpoints, queries)?)
            .await
    }

    /// Bounded read of public events for a message-content reference. Relay
    /// hints come from other people's messages, so every endpoint passes the
    /// discovered-endpoint safety rule; unsafe ones are dropped, not fatal.
    #[doc(hidden)]
    pub async fn fetch_public_events(
        &self,
        endpoints: Vec<TransportEndpoint>,
        filters: Vec<Filter>,
        max_events: usize,
    ) -> Result<directory::PublicEventFetch, String> {
        let endpoints = self
            .inner
            .relay_safety
            .retain_safe_endpoints(endpoints, "public event fetch");
        if endpoints.is_empty() {
            return Err("public event fetch: no safe relay endpoints".to_owned());
        }
        directory::fetch_public_events(&endpoints, filters, max_events).await
    }

    #[doc(hidden)]
    pub async fn fetch_directory_events_with_completion(
        &self,
        endpoints: Vec<TransportEndpoint>,
        queries: Vec<DirectoryEventQuery>,
    ) -> Result<DirectoryFetchOutcome, String> {
        let endpoints = self
            .inner
            .relay_safety
            .sanitize_endpoints(endpoints, "directory fetch")?;
        self.inner
            .directory
            .fetch_events_with_completion(DirectoryFetchRequest::new(endpoints, queries)?)
            .await
    }

    #[doc(hidden)]
    pub async fn inspect_directory_events(
        &self,
        endpoint: TransportEndpoint,
        query: DirectoryEventQuery,
        signer: Option<Arc<dyn transport_nostr_peeler::MarmotNostrSigner>>,
    ) -> Result<Vec<DirectoryRelayEventRecord>, directory::DirectoryInspectionError> {
        let endpoints = self
            .inner
            .relay_safety
            .sanitize_endpoints(vec![endpoint], "onboarding inspection")
            .map_err(|_| directory::DirectoryInspectionError::InvalidRequest)?;
        self.inner
            .directory
            .inspect_events(
                DirectoryFetchRequest::new(endpoints, vec![query])
                    .map_err(|_| directory::DirectoryInspectionError::InvalidRequest)?,
                signer,
            )
            .await
    }

    /// Narrow discovered relay endpoints to the safe ones, dropping the rest.
    ///
    /// Unlike the fail-closed sanitize on the dial path, this is for endpoints
    /// another account published; see
    /// [`RelaySafetyPolicy::retain_safe_endpoints`]. What survives is still
    /// sanitized at the dial chokepoint.
    #[doc(hidden)]
    pub fn retain_safe_discovered_endpoints(
        &self,
        endpoints: Vec<TransportEndpoint>,
        context: &str,
    ) -> Vec<TransportEndpoint> {
        self.inner
            .relay_safety
            .retain_safe_endpoints(endpoints, context)
    }

    #[doc(hidden)]
    pub fn subscribe_directory_events(&self) -> broadcast::Receiver<DirectoryRelayPlaneEvent> {
        self.inner.transport.directory_events.subscribe()
    }

    #[doc(hidden)]
    pub async fn sync_directory_user_subscriptions(
        &self,
        plan: DirectorySyncPlan,
        force_rebuild: bool,
    ) -> Result<DirectorySubscriptionSyncSummary, String> {
        let _sync_guard = self.inner.directory_subscription_sync.lock().await;
        self.spawn_router();
        let endpoints = self
            .inner
            .relay_safety
            .sanitize_endpoints(plan.endpoints, "directory subscription")?;
        if plan.batches.is_empty() || endpoints.is_empty() {
            if let Some(client) = &self.inner.transport.directory_client {
                let (_, stale) = self
                    .inner
                    .directory
                    .subscription_diff(&HashSet::new())
                    .await;
                for subscription_id in stale {
                    client
                        .unsubscribe(&SubscriptionId::new(subscription_id))
                        .await
                        .map_err(|_| "directory unsubscribe failed".to_owned())?;
                }
            }
            return self
                .inner
                .directory
                .replace_subscriptions(HashMap::new())
                .await;
        }
        let directory_client = self
            .inner
            .transport
            .directory_client
            .as_ref()
            .ok_or_else(|| "directory subscription requires SDK relay plane".to_owned())?;
        let relay_urls = endpoints
            .iter()
            .map(|endpoint| {
                RelayUrl::parse(endpoint.as_str())
                    .map_err(|_| "directory subscription: invalid relay endpoint".to_owned())
            })
            .collect::<Result<Vec<_>, _>>()?;
        for relay_url in &relay_urls {
            directory_client
                .add_relay(relay_url.clone())
                .await
                .map_err(|_| "directory subscription add relay failed".to_owned())?;
            timeout(
                DIRECTORY_RELAY_CONNECT_WAIT,
                directory_client.connect_relay(relay_url.clone()),
            )
            .await
            .map_err(|_| "directory subscription connect relay timed out".to_owned())?
            .map_err(|_| "directory subscription connect relay failed".to_owned())?;
        }
        let endpoints_changed = self
            .inner
            .directory
            .set_subscription_endpoints(&relay_urls)
            .await;

        let desired_ids = plan
            .batches
            .iter()
            .map(|batch| batch.subscription_id.clone())
            .collect::<HashSet<_>>();
        let (mut to_add, to_remove) = self.inner.directory.subscription_diff(&desired_ids).await;
        if force_rebuild || endpoints_changed {
            to_add = desired_ids;
            self.inner.directory.mark_rebuild_pending(&to_add).await;
        }
        for subscription_id in &to_remove {
            directory_client
                .unsubscribe(&SubscriptionId::new(subscription_id.clone()))
                .await
                .map_err(|_| "directory unsubscribe failed".to_owned())?;
        }
        if force_rebuild || endpoints_changed {
            for subscription_id in &to_add {
                directory_client
                    .unsubscribe(&SubscriptionId::new(subscription_id.clone()))
                    .await
                    .map_err(|_| "directory unsubscribe failed".to_owned())?;
            }
        }
        // The validation filter persisted for every batch (added or already
        // active) is keyed on the same canonical-hex authors and kinds the SDK
        // subscription is issued with, so a live notification is only forwarded
        // into the directory cache when it matches an active subscription's
        // requested authors and kinds (mdk#709).
        let mut desired = HashMap::with_capacity(plan.batches.len());
        let mut subscriptions_created = 0;
        for batch in &plan.batches {
            let authors = batch
                .authors
                .iter()
                .map(|author| PublicKey::parse(author).map_err(|_| "invalid directory author"))
                .collect::<Result<Vec<_>, _>>()?;
            let kinds = batch
                .kinds
                .iter()
                .map(|kind| {
                    u16::try_from(*kind)
                        .map(Kind::from)
                        .map_err(|_| format!("unsupported Nostr kind {kind}"))
                })
                .collect::<Result<Vec<_>, _>>()?;
            // Canonical lowercase hex matches the `event.pubkey` form a forwarded
            // SDK event carries, so the membership check is exact.
            let filter_authors = authors.iter().map(PublicKey::to_hex).collect::<Vec<_>>();
            let validation_filter =
                DirectorySubscriptionFilter::new(filter_authors, batch.kinds.clone());
            desired.insert(batch.subscription_id.clone(), validation_filter.clone());
            if !to_add.contains(&batch.subscription_id) {
                continue;
            }
            self.inner
                .directory
                .clear_auth_required(&batch.subscription_id)
                .await;
            let mut filter = Filter::new()
                .authors(authors)
                .kinds(kinds)
                .limit(batch.authors.len().saturating_mul(batch.kinds.len()).max(1));
            if let Some(since) = batch.since {
                filter = filter.since(NostrTimestamp::from_secs(since));
            }
            let previous = self
                .inner
                .directory
                .record_subscription_filter(batch.subscription_id.clone(), validation_filter)
                .await;
            let subscription = directory_client
                .subscribe(nostr_sdk::prelude::ReqTarget::manual(
                    relay_urls
                        .iter()
                        .cloned()
                        .map(|url| (url, vec![filter.clone()])),
                ))
                .with_id(SubscriptionId::new(batch.subscription_id.clone()))
                .await;
            if !subscription.is_ok_and(|output| !output.success.is_empty()) {
                self.inner
                    .directory
                    .restore_failed_subscription_filter(&batch.subscription_id, previous)
                    .await;
                return Err("directory subscription registered on no relays".to_owned());
            }
            subscriptions_created += usize::from(previous.is_none());
            self.inner
                .directory
                .mark_subscription_installed(&batch.subscription_id)
                .await;
        }

        self.inner
            .directory
            .complete_subscription_sync(desired, subscriptions_created)
            .await
    }

    /// Whether [`Self::shutdown`] has begun. It is set before the relay pools
    /// are torn down, so a request that failed because of that teardown sees
    /// it set.
    pub fn is_shutting_down(&self) -> bool {
        self.inner.transport.shutting_down.load(Ordering::SeqCst)
    }

    pub async fn shutdown(&self) {
        self.inner
            .transport
            .shutting_down
            .store(true, Ordering::SeqCst);
        if let Some(sdk_relay_client) = &self.inner.transport.sdk_relay_client {
            let _ = timeout(
                RELAY_PLANE_SHUTDOWN_WAIT,
                sdk_relay_client.shutdown_accounts(),
            )
            .await;
            let timed_out = timeout(RELAY_PLANE_SHUTDOWN_WAIT, sdk_relay_client.shutdown())
                .await
                .is_err();
            if timed_out {
                tracing::warn!(
                    target: "marmot_app::relay_plane",
                    method = "shutdown",
                    "SDK relay pool shutdown timed out",
                );
            }
        }
        if let Some(directory_client) = &self.inner.transport.directory_client {
            let _ = timeout(RELAY_PLANE_SHUTDOWN_WAIT, directory_client.shutdown()).await;
        }
        account_deliveries_write(&self.inner.transport.account_deliveries).clear();
        if let Some(handle) = self.inner.transport.router.lock().await.take() {
            let mut handle = handle;
            handle.abort();
            let _ = timeout(RELAY_PLANE_TASK_ABORT_WAIT, &mut handle).await;
        }
        if let Some(handle) = self
            .inner
            .transport
            .notification_forwarder
            .lock()
            .await
            .take()
        {
            let mut handle = handle;
            handle.abort();
            let _ = timeout(RELAY_PLANE_TASK_ABORT_WAIT, &mut handle).await;
        }
        let account_forwarders = self
            .inner
            .transport
            .account_notification_forwarders
            .lock()
            .await
            .drain()
            .map(|(_, handle)| handle)
            .collect::<Vec<_>>();
        for mut handle in account_forwarders {
            handle.abort();
            let _ = timeout(RELAY_PLANE_TASK_ABORT_WAIT, &mut handle).await;
        }
        if let Some(handle) = self
            .inner
            .transport
            .directory_notification_forwarder
            .lock()
            .await
            .take()
        {
            let mut handle = handle;
            handle.abort();
            let _ = timeout(RELAY_PLANE_TASK_ABORT_WAIT, &mut handle).await;
        }
        self.inner
            .transport
            .notification_forwarder_health
            .running_count
            .store(0, Ordering::SeqCst);
    }

    fn spawn_router(&self) {
        if self.inner.transport.shutting_down.load(Ordering::SeqCst) {
            return;
        }
        let Ok(handle) = tokio::runtime::Handle::try_current() else {
            return;
        };
        if let Ok(mut notification_forwarder) =
            self.inner.transport.notification_forwarder.try_lock()
        {
            let needs_forwarder = match notification_forwarder.as_ref() {
                None => true,
                Some(forwarder) => forwarder.is_finished(),
            };
            if needs_forwarder
                && let Some(sdk_relay_client) = &self.inner.transport.sdk_relay_client
                && !sdk_relay_client.is_multi_account()
            {
                if sdk_relay_client.client().is_shutdown() {
                    notification_forwarder.take();
                } else {
                    if notification_forwarder.take().is_some() {
                        recover_relay_notification_forwarder(
                            &self.inner.transport,
                            RelayNotificationConsumerExit::Closed,
                        );
                    }
                    *notification_forwarder = Some(spawn_relay_notification_forwarder(
                        sdk_relay_client.clone(),
                        self.inner.transport.clone(),
                    ));
                }
            } else if needs_forwarder
                && let Some(source) = &self.inner.transport.injected_notifications
            {
                if notification_forwarder.take().is_some() {
                    recover_relay_notification_forwarder(
                        &self.inner.transport,
                        RelayNotificationConsumerExit::Closed,
                    );
                }
                let source: Arc<dyn RelayNotificationSource> =
                    Arc::new(InjectedRelayNotificationSource(source.clone()));
                // Read from now on: a fake relay answers a REQ as it is sent,
                // which can be before the consumer task first runs.
                let receiver = source.notifications();
                *notification_forwarder = Some(spawn_relay_notification_supervisor_reading(
                    source,
                    self.inner.transport.clone(),
                    None,
                    Some(receiver),
                ));
            }
        }
        if let Ok(mut forwarder) = self
            .inner
            .transport
            .directory_notification_forwarder
            .try_lock()
            && forwarder.as_ref().is_none_or(JoinHandle::is_finished)
            && let Some(client) = &self.inner.transport.directory_client
            && !client.is_shutdown()
        {
            if forwarder.take().is_some() {
                let _ = self
                    .inner
                    .transport
                    .directory_events
                    .send(DirectoryRelayPlaneEvent::RecoveryRequired);
            }
            *forwarder = Some(spawn_directory_notification_forwarder(
                client.notifications_with_gaps(),
                self.inner.transport.directory_events.clone(),
                self.inner.directory.clone(),
            ));
        }
        let Ok(mut router) = self.inner.transport.router.try_lock() else {
            return;
        };
        if router.is_some() {
            return;
        }
        let transport = self.inner.transport.clone();
        let adapter = transport.adapter.clone();
        let handle = handle.spawn(async move {
            loop {
                tokio::select! {
                    delivery = adapter.receive() => match delivery {
                        Ok(Some(delivery)) => {
                            let account_id = delivery.account_id.clone();
                            let message_id = delivery.message.id.clone();
                            // Install the queue/spill pin or charge loss before
                            // releasing the staged replay's admission fence.
                            route_account_delivery(&transport, delivery);
                            adapter.ack_replay_delivery(&account_id, &message_id).await;
                        }
                        Ok(None) | Err(_) => break,
                    },
                    () = adapter.replay_loss_notified() => {
                        let account_ids = account_deliveries_read(&transport.account_deliveries)
                            .keys()
                            .cloned()
                            .collect::<Vec<_>>();
                        for account_id in account_ids {
                            let status = adapter.take_replay_status(&account_id).await;
                            if let Some(loss) = status.loss {
                                // Taking the status consumes the adapter's
                                // evidence: import it before another await.
                                import_replay_loss(&transport, &account_id, loss);
                            }
                        }
                    }
                }
            }
        });
        *router = Some(handle);
    }

    #[cfg(any(test, feature = "test-support"))]
    #[doc(hidden)]
    pub async fn handle_relay_event_for_test(
        &self,
        relay_event: transport_nostr_adapter::NostrRelayEvent,
    ) -> Result<usize, TransportAdapterError> {
        self.inner
            .transport
            .adapter
            .handle_relay_event(relay_event)
            .await
    }

    #[cfg(any(test, feature = "test-support"))]
    #[doc(hidden)]
    pub fn set_account_delivery_recovery_marker_for_test(
        &self,
        account_id: &MemberId,
        marker: AccountDeliveryRecoveryMarker,
    ) -> bool {
        let mut routes = account_deliveries_write(&self.inner.transport.account_deliveries);
        let Some(route) = routes.get_mut(account_id) else {
            return false;
        };
        route.recovery_marker = Some(marker);
        true
    }

    /// Exercise the loss tier directly: a full queue omits instead of spilling.
    #[cfg(any(test, feature = "test-support"))]
    #[doc(hidden)]
    pub fn disable_account_delivery_spill_for_test(&self, account_id: &MemberId) -> bool {
        let mut routes = account_deliveries_write(&self.inner.transport.account_deliveries);
        routes
            .get_mut(account_id)
            .map(|route| route.spill = None)
            .is_some()
    }

    /// Drive every managed account worker through its receive-error reconnect
    /// path: an unexpected notification consumer exit closes inbound delivery.
    #[cfg(any(test, feature = "test-support"))]
    #[doc(hidden)]
    pub fn simulate_notification_consumer_exit_for_test(&self) {
        recover_relay_notification_forwarder(
            &self.inner.transport,
            RelayNotificationConsumerExit::Closed,
        );
    }

    /// Record a lag on one account's notification consumer, as its scoped
    /// forwarder does with the floor its SDK context read at the lag. Inbound
    /// delivery stays open.
    #[cfg(any(test, feature = "test-support"))]
    #[doc(hidden)]
    pub fn simulate_notification_lag_for_test(
        &self,
        account_id: &MemberId,
        skipped_notifications: u64,
        floor: NostrNotificationLossFloor,
    ) {
        self.simulate_routed_notification_lag_for_test(
            account_id,
            skipped_notifications,
            NostrNotificationLossBounds::account_wide(floor),
        );
    }

    /// [`Self::simulate_notification_lag_for_test`] with the per-route floors
    /// the SDK context read at the lag.
    #[cfg(any(test, feature = "test-support"))]
    #[doc(hidden)]
    pub fn simulate_routed_notification_lag_for_test(
        &self,
        account_id: &MemberId,
        skipped_notifications: u64,
        bounds: NostrNotificationLossBounds,
    ) {
        recover_relay_notification_forwarder_scoped(
            &self.inner.transport,
            RelayNotificationConsumerExit::Lagged(skipped_notifications),
            bounds,
            Some(account_id),
        );
    }

    /// The SDK relay client whose pool carries `account_id`'s REQs, for the
    /// recovery runner's sweeps and comparisons (mdk#2660). `None` without an
    /// SDK client or an account context. It dials nothing: its pool admits
    /// relays only through the dial-safety chokepoint.
    #[doc(hidden)]
    pub async fn account_history_relay_client(
        &self,
        account_id: &MemberId,
    ) -> Option<NostrSdkRelayClient> {
        self.inner
            .transport
            .sdk_relay_client
            .as_ref()?
            .account_relay_client(account_id)
            .await
    }

    /// `account_id`'s deliveries for one fetched history event, as the
    /// account adapter's `recovered_deliveries` gives them. The caller admits
    /// them directly, never through the live queue.
    #[doc(hidden)]
    pub async fn recovered_deliveries_for_account(
        &self,
        account_id: &MemberId,
        event: transport_nostr_adapter::NostrRelayEvent,
    ) -> Result<Vec<TransportDelivery>, TransportAdapterError> {
        self.inner
            .transport
            .adapter
            .reconciled_deliveries(account_id, event)
            .await
    }

    /// The SDK relay client the plane runs on, for tests that build a whole
    /// app and so cannot reach the plane's private state.
    #[cfg(any(test, feature = "test-support"))]
    #[doc(hidden)]
    pub fn sdk_relay_client_for_test(&self) -> Option<NostrSdkRelayClient> {
        self.inner.transport.sdk_relay_client.clone()
    }

    /// Unexpected notification-consumer exits the plane has recorded.
    #[cfg(any(test, feature = "test-support"))]
    #[doc(hidden)]
    pub fn notification_forwarder_unexpected_exits_for_test(&self) -> u64 {
        self.inner
            .transport
            .notification_forwarder_health
            .snapshot()
            .unexpected_exits
    }
}

impl RelayPlaneHealth {
    fn from_sdk(
        health: NostrSdkRelayHealth,
        directory: DirectoryRelayStats,
        forwarder: RelayNotificationForwarderHealthSnapshot,
        account_delivery: AccountDeliveryMetricsSnapshot,
    ) -> Self {
        Self {
            sdk_backed: true,
            total_relays: health.total_relays,
            initialized: health.initialized,
            pending: health.pending,
            connecting: health.connecting,
            connected: health.connected,
            disconnected: health.disconnected,
            terminated: health.terminated,
            banned: health.banned,
            sleeping: health.sleeping,
            connection_attempts: health.connection_attempts,
            connection_successes: health.connection_successes,
            notification_forwarder_running: forwarder.running,
            notification_forwarder_restarts: forwarder.restarts,
            notification_forwarder_lag_incidents: forwarder.lag_incidents,
            notification_forwarder_lagged_notifications: forwarder.lagged_notifications,
            notification_forwarder_panics: forwarder.panics,
            notification_forwarder_unexpected_exits: forwarder.unexpected_exits,
            account_delivery_queue_depth: account_delivery.queue_depth,
            account_delivery_max_queue_depth: account_delivery.max_queue_depth,
            account_delivery_dropped: account_delivery.dropped,
            account_delivery_spilled: account_delivery.spilled,
            account_delivery_spill_already_seen: account_delivery.spill_already_seen,
            account_delivery_recovery_attempts: account_delivery.recovery_attempts,
            account_delivery_recovery_successes: account_delivery.recovery_successes,
            account_delivery_recovery_failures: account_delivery.recovery_failures,
            account_delivery_recovery_elapsed_ms: account_delivery.recovery_elapsed_ms,
            directory_inflight_fetches: directory.inflight_fetches,
            directory_active_subscriptions: directory.active_subscriptions,
            directory_auth_required_routes: directory.auth_required_routes,
            directory_completed_fetches: directory.completed_fetches,
            directory_coalesced_waiters: directory.coalesced_waiters,
            directory_failed_fetches: directory.failed_fetches,
            directory_completed_subscription_syncs: directory.completed_subscription_syncs,
            directory_subscriptions_created: directory.subscriptions_created,
            directory_subscriptions_removed: directory.subscriptions_removed,
        }
    }

    fn from_directory(
        directory: DirectoryRelayStats,
        account_delivery: AccountDeliveryMetricsSnapshot,
    ) -> Self {
        Self {
            account_delivery_queue_depth: account_delivery.queue_depth,
            account_delivery_max_queue_depth: account_delivery.max_queue_depth,
            account_delivery_dropped: account_delivery.dropped,
            account_delivery_spilled: account_delivery.spilled,
            account_delivery_spill_already_seen: account_delivery.spill_already_seen,
            account_delivery_recovery_attempts: account_delivery.recovery_attempts,
            account_delivery_recovery_successes: account_delivery.recovery_successes,
            account_delivery_recovery_failures: account_delivery.recovery_failures,
            account_delivery_recovery_elapsed_ms: account_delivery.recovery_elapsed_ms,
            directory_inflight_fetches: directory.inflight_fetches,
            directory_active_subscriptions: directory.active_subscriptions,
            directory_auth_required_routes: directory.auth_required_routes,
            directory_completed_fetches: directory.completed_fetches,
            directory_coalesced_waiters: directory.coalesced_waiters,
            directory_failed_fetches: directory.failed_fetches,
            directory_completed_subscription_syncs: directory.completed_subscription_syncs,
            directory_subscriptions_created: directory.subscriptions_created,
            directory_subscriptions_removed: directory.subscriptions_removed,
            ..Self::default()
        }
    }
}

impl RelayNotificationForwarderHealth {
    fn snapshot(&self) -> RelayNotificationForwarderHealthSnapshot {
        RelayNotificationForwarderHealthSnapshot {
            running: self.running_count.load(Ordering::Relaxed) > 0,
            restarts: self.restarts.load(Ordering::Relaxed),
            lag_incidents: self.lag_incidents.load(Ordering::Relaxed),
            lagged_notifications: self.lagged_notifications.load(Ordering::Relaxed),
            panics: self.panics.load(Ordering::Relaxed),
            unexpected_exits: self.unexpected_exits.load(Ordering::Relaxed),
        }
    }

    fn increment(counter: &AtomicU64, value: u64) {
        let _ = counter.fetch_update(Ordering::Relaxed, Ordering::Relaxed, |current| {
            Some(current.saturating_add(value))
        });
    }
}

#[derive(Clone, Copy, Debug, PartialEq, Eq)]
enum RelayNotificationConsumerExit {
    Shutdown,
    Lagged(u64),
    Closed,
}

type RelayNotificationStream =
    Pin<Box<dyn Stream<Item = NotificationUpdate<NostrSdkNotification>> + Send>>;

type DirectoryNotificationStream =
    Pin<Box<dyn Stream<Item = NotificationUpdate<ClientNotification>> + Send>>;

struct RelayNotificationConsumerOutcome {
    receiver: RelayNotificationStream,
    exit: RelayNotificationConsumerExit,
    /// The source's REQ floors, read where a lag was recorded. Unbounded for
    /// every other exit.
    lag_bounds: NostrNotificationLossBounds,
}

impl RelayNotificationConsumerOutcome {
    fn new(receiver: RelayNotificationStream, exit: RelayNotificationConsumerExit) -> Self {
        Self {
            receiver,
            exit,
            lag_bounds: NostrNotificationLossBounds::account_wide(
                NostrNotificationLossFloor::Unbounded,
            ),
        }
    }

    /// Record `abandoned` notifications as receiver loss, reading the floors
    /// at that point.
    fn lagged(
        receiver: RelayNotificationStream,
        source: &dyn RelayNotificationSource,
        abandoned: u64,
    ) -> Self {
        Self {
            receiver,
            exit: RelayNotificationConsumerExit::Lagged(abandoned),
            lag_bounds: source.record_loss(abandoned),
        }
    }
}

#[derive(Clone, Copy, Debug, PartialEq, Eq)]
struct RelayNotificationRestartBackoff {
    next: Duration,
}

impl Default for RelayNotificationRestartBackoff {
    fn default() -> Self {
        Self {
            next: RELAY_NOTIFICATION_RESTART_INITIAL_BACKOFF,
        }
    }
}

impl RelayNotificationRestartBackoff {
    fn delay_after_failure(&mut self, consumer_runtime: Duration) -> Duration {
        if consumer_runtime >= RELAY_NOTIFICATION_RESTART_HEALTHY_RUNTIME {
            self.next = RELAY_NOTIFICATION_RESTART_INITIAL_BACKOFF;
        }
        let delay = self.next;
        self.next = self
            .next
            .saturating_mul(2)
            .min(RELAY_NOTIFICATION_RESTART_MAX_BACKOFF);
        delay
    }
}

trait RelayNotificationSource: Send + Sync {
    fn notifications(&self) -> RelayNotificationStream;
    fn is_shutdown(&self) -> bool;
    /// Record receiver loss and return the lowest REQ `since` that could have
    /// delivered the lost notifications, overall and per route. A source
    /// without that evidence reports it unbounded.
    fn record_loss(&self, _skipped: u64) -> NostrNotificationLossBounds {
        NostrNotificationLossBounds::account_wide(NostrNotificationLossFloor::Unbounded)
    }
    /// Native sources may already have charged their own gap before yielding.
    fn record_lag(&self, skipped: u64, abandoned: u64) -> NostrNotificationLossBounds {
        self.record_loss(skipped.saturating_add(abandoned))
    }
    fn receiver_replaced(&self) {}
    /// Classify a relay's `CLOSED` reason as this source's SDK client settles
    /// the REQ. Without a known client the authenticator is unknown, which
    /// keeps an `auth-required:` close pending.
    fn subscription_refusal(&self, reason: &str) -> transport_nostr_adapter::SubscriptionRefusal {
        transport_nostr_adapter::sdk_subscription_refusal(reason, None)
    }
    #[cfg(test)]
    fn worker_failure_hook(&self) -> Option<Arc<NotificationWorkerFailureHook>> {
        None
    }
    #[cfg(test)]
    fn notification_queued(&self) {}
}

#[cfg(test)]
struct NotificationWorkerFailureHook {
    entered: tokio::sync::Notify,
    release: tokio::sync::Notify,
    fail_once: AtomicBool,
    /// Hold the first notification like `fail_once`, then handle it normally:
    /// a busy, not failed, event worker.
    stall_once: AtomicBool,
}

struct SdkRelayNotificationSource {
    client: NostrSdkRelayClient,
}

impl RelayNotificationSource for SdkRelayNotificationSource {
    fn notifications(&self) -> RelayNotificationStream {
        self.client.notifications()
    }

    fn is_shutdown(&self) -> bool {
        self.client.client().is_shutdown()
    }

    fn record_loss(&self, skipped: u64) -> NostrNotificationLossBounds {
        self.client.record_notification_gap(skipped);
        // Fence all retained source/worker frames before the supervisor resumes
        // this same receiver. A live REQ may never emit another EOSE.
        self.client.notification_receiver_replaced();
        // Read at the lag: a later subscription change could raise it past
        // an event this gap lost.
        self.client.notification_loss_bounds()
    }

    fn record_lag(&self, _skipped: u64, abandoned: u64) -> NostrNotificationLossBounds {
        // Native source loss has already advanced the ordered reader frontier.
        // Only the frontend worker's additional abandoned tail remains to charge.
        self.client.record_notification_gap(abandoned);
        self.client.notification_loss_bounds()
    }

    fn subscription_refusal(&self, reason: &str) -> transport_nostr_adapter::SubscriptionRefusal {
        self.client.subscription_refusal(reason)
    }
}

/// An injected relay client's messages. It has no SDK registry or REQ floors
/// to consult: a `CLOSED` keeps `auth-required:` pending, and a lag is loss
/// from no known floor.
struct InjectedRelayNotificationSource(Arc<dyn NostrRelayNotificationSource>);

impl RelayNotificationSource for InjectedRelayNotificationSource {
    fn notifications(&self) -> RelayNotificationStream {
        self.0.notifications()
    }

    /// The stream itself reports a shutdown.
    fn is_shutdown(&self) -> bool {
        false
    }
}

fn spawn_relay_notification_forwarder(
    sdk_relay_client: NostrSdkRelayClient,
    transport: Arc<RelayPlaneTransport>,
) -> JoinHandle<()> {
    spawn_relay_notification_forwarder_scoped(sdk_relay_client, transport, None)
}

fn spawn_relay_notification_forwarder_scoped(
    sdk_relay_client: NostrSdkRelayClient,
    transport: Arc<RelayPlaneTransport>,
    account_id: Option<MemberId>,
) -> JoinHandle<()> {
    let source: Arc<dyn RelayNotificationSource> = Arc::new(SdkRelayNotificationSource {
        client: sdk_relay_client,
    });
    spawn_relay_notification_supervisor_scoped(source, transport, account_id)
}

/// Public directory interests have their own unauthenticated SDK client and
/// receiver. A gap here invalidates directory coverage without interrupting
/// an account's live delivery or attributing the loss to an account.
fn spawn_directory_notification_forwarder(
    mut receiver: DirectoryNotificationStream,
    events: broadcast::Sender<DirectoryRelayPlaneEvent>,
    directory: DirectoryRelayPlane,
) -> JoinHandle<()> {
    tokio::spawn(async move {
        loop {
            match receiver.next().await {
                Some(NotificationUpdate::Notification(ClientNotification::Event {
                    relay_url,
                    subscription_id,
                    event,
                })) => {
                    // SDK admission and seen-state are not authentication at
                    // MDK's directory boundary (mdk#2521).
                    if event.verify().is_err() {
                        continue;
                    }
                    let subscription_id = subscription_id.to_string();
                    if let Ok(event) = NostrTransportEvent::from_nostr_event(&event)
                        && directory
                            .accepts_live_event_from(
                                &subscription_id,
                                relay_url.as_str(),
                                &event.pubkey,
                                event.kind,
                            )
                            .await
                    {
                        let _ = events.send(DirectoryRelayPlaneEvent::Record(
                            DirectoryRelayEventRecord {
                                endpoints: vec![TransportEndpoint(relay_url.to_string())],
                                event,
                            },
                        ));
                    }
                }
                Some(NotificationUpdate::Notification(ClientNotification::Message {
                    relay_url,
                    message,
                })) => {
                    if let RelayMessage::Closed {
                        subscription_id,
                        message,
                    } = *message
                        && message.starts_with("auth-required:")
                        && directory
                            .mark_auth_required(subscription_id.as_str(), relay_url.as_str())
                            .await
                    {
                        let count = directory.stats().await.auth_required_routes;
                        tracing::warn!(
                            target: "marmot_app::relay_plane",
                            method = "spawn_directory_notification_forwarder",
                            auth_required_routes = count,
                            "anonymous directory request requires authentication",
                        );
                    }
                }
                Some(NotificationUpdate::Notification(ClientNotification::Shutdown)) => break,
                Some(NotificationUpdate::Lagged { skipped }) => {
                    tracing::warn!(
                        target: "marmot_app::relay_plane",
                        method = "spawn_directory_notification_forwarder",
                        skipped_notifications = skipped,
                        "directory SDK receiver lost notifications",
                    );
                    let _ = events.send(DirectoryRelayPlaneEvent::RecoveryRequired);
                }
                None => {
                    break;
                }
            }
        }
    })
}

#[cfg(test)]
fn spawn_relay_notification_supervisor(
    source: Arc<dyn RelayNotificationSource>,
    transport: Arc<RelayPlaneTransport>,
) -> JoinHandle<()> {
    spawn_relay_notification_supervisor_scoped(source, transport, None)
}

fn spawn_relay_notification_supervisor_scoped(
    source: Arc<dyn RelayNotificationSource>,
    transport: Arc<RelayPlaneTransport>,
    account_id: Option<MemberId>,
) -> JoinHandle<()> {
    spawn_relay_notification_supervisor_reading(source, transport, account_id, None)
}

/// [`spawn_relay_notification_supervisor_scoped`], whose first consumer reads
/// `receiver` when one is given instead of asking the source for one.
fn spawn_relay_notification_supervisor_reading(
    source: Arc<dyn RelayNotificationSource>,
    transport: Arc<RelayPlaneTransport>,
    account_id: Option<MemberId>,
    receiver: Option<RelayNotificationStream>,
) -> JoinHandle<()> {
    tokio::spawn(async move {
        let _running = RunningSupervisorGuard::new(transport.notification_forwarder_health.clone());
        let mut receiver = receiver;
        let mut restart_backoff = RelayNotificationRestartBackoff::default();
        loop {
            let adapter = transport.adapter.clone();
            let source_for_consumer = source.clone();
            let next_receiver = receiver.take();
            let consumer_started_at = Instant::now();
            if next_receiver.is_none() {
                source.receiver_replaced();
            }
            let consumer_account_id = account_id.clone();
            let mut consumer = tokio::spawn(async move {
                let receiver = next_receiver.unwrap_or_else(|| source_for_consumer.notifications());
                run_relay_notification_consumer_scoped(
                    receiver,
                    adapter,
                    consumer_account_id,
                    source_for_consumer,
                )
                .await
            });
            let abort_on_drop = AbortTaskOnDrop(consumer.abort_handle());
            match (&mut consumer).await {
                Ok(outcome) => {
                    drop(abort_on_drop);
                    receiver = Some(outcome.receiver);
                    if outcome.exit == RelayNotificationConsumerExit::Shutdown
                        || transport.shutting_down.load(Ordering::SeqCst)
                        || source.is_shutdown()
                    {
                        if matches!(outcome.exit, RelayNotificationConsumerExit::Lagged(_)) {
                            recover_relay_notification_forwarder_scoped(
                                &transport,
                                outcome.exit,
                                outcome.lag_bounds,
                                account_id.as_ref(),
                            );
                        }
                        break;
                    }
                    // A lag resumes the retained receiver at once; only an
                    // unexpected exit replaces it after a backoff.
                    recover_relay_notification_forwarder_scoped(
                        &transport,
                        outcome.exit,
                        outcome.lag_bounds,
                        account_id.as_ref(),
                    );
                    if outcome.exit == RelayNotificationConsumerExit::Closed {
                        receiver = None;
                        tokio::time::sleep(
                            restart_backoff.delay_after_failure(consumer_started_at.elapsed()),
                        )
                        .await;
                    }
                }
                Err(join_error) => {
                    drop(abort_on_drop);
                    if transport.shutting_down.load(Ordering::SeqCst) || source.is_shutdown() {
                        break;
                    }
                    RelayNotificationForwarderHealth::increment(
                        &transport.notification_forwarder_health.panics,
                        u64::from(join_error.is_panic()),
                    );
                    receiver = None;
                    recover_relay_notification_forwarder_scoped(
                        &transport,
                        RelayNotificationConsumerExit::Closed,
                        NostrNotificationLossBounds::account_wide(
                            NostrNotificationLossFloor::Unbounded,
                        ),
                        account_id.as_ref(),
                    );
                    tokio::time::sleep(
                        restart_backoff.delay_after_failure(consumer_started_at.elapsed()),
                    )
                    .await;
                }
            }
        }
    })
}

struct RunningSupervisorGuard(Arc<RelayNotificationForwarderHealth>);

impl RunningSupervisorGuard {
    fn new(health: Arc<RelayNotificationForwarderHealth>) -> Self {
        health.running_count.fetch_add(1, Ordering::SeqCst);
        Self(health)
    }
}

impl Drop for RunningSupervisorGuard {
    fn drop(&mut self) {
        // Terminal shutdown may already have reset the aggregate to zero.
        let _ = self
            .0
            .running_count
            .fetch_update(Ordering::SeqCst, Ordering::SeqCst, |count| {
                Some(count.saturating_sub(1))
            });
    }
}

struct AbortTaskOnDrop(tokio::task::AbortHandle);

impl Drop for AbortTaskOnDrop {
    fn drop(&mut self) {
        self.0.abort();
    }
}

#[cfg(test)]
async fn run_relay_notification_consumer(
    mut receiver: RelayNotificationStream,
    adapter: NostrTransportAdapter,
) -> RelayNotificationConsumerOutcome {
    loop {
        match receiver.next().await {
            Some(NotificationUpdate::Notification(notification)) => {
                let should_shutdown =
                    handle_relay_notification(notification, &adapter, None, &|reason| {
                        transport_nostr_adapter::sdk_subscription_refusal(reason, None)
                    })
                    .await;
                if should_shutdown {
                    return RelayNotificationConsumerOutcome::new(
                        receiver,
                        RelayNotificationConsumerExit::Shutdown,
                    );
                }
            }
            Some(NotificationUpdate::Lagged { skipped }) => {
                return RelayNotificationConsumerOutcome::new(
                    receiver,
                    RelayNotificationConsumerExit::Lagged(skipped),
                );
            }
            None => {
                return RelayNotificationConsumerOutcome::new(
                    receiver,
                    RelayNotificationConsumerExit::Closed,
                );
            }
        }
    }
}

/// Keep reading the SDK receiver while account delivery or telemetry awaits.
/// The bounded queue is an event lane; a full queue becomes a typed gap before
/// the reader can block, and the watch control lane advances independently.
fn abandoned_notification_lane(
    receiver: RelayNotificationStream,
    pending: &AtomicU64,
    source: &dyn RelayNotificationSource,
) -> RelayNotificationConsumerOutcome {
    let abandoned = pending.load(Ordering::SeqCst);
    if abandoned > 0 {
        RelayNotificationConsumerOutcome::lagged(receiver, source, abandoned)
    } else {
        RelayNotificationConsumerOutcome::new(receiver, RelayNotificationConsumerExit::Closed)
    }
}

async fn run_relay_notification_consumer_scoped(
    mut receiver: RelayNotificationStream,
    adapter: NostrTransportAdapter,
    account_id: Option<MemberId>,
    source: Arc<dyn RelayNotificationSource>,
) -> RelayNotificationConsumerOutcome {
    let (sender, mut event_rx) = mpsc::channel(RELAY_NOTIFICATION_EVENT_QUEUE_CAPACITY);
    // Channel capacity returns to its maximum when a panicked worker drops
    // the receiver, even though queued work was discarded. Count admitted
    // items independently until each worker operation actually finishes.
    let pending = Arc::new(AtomicU64::new(0));
    let worker_pending = pending.clone();
    #[cfg(test)]
    let worker_failure_hook = source.worker_failure_hook();
    let classifier = Arc::clone(&source);
    let mut worker = tokio::spawn(async move {
        while let Some(notification) = event_rx.recv().await {
            #[cfg(test)]
            if let Some(hook) = &worker_failure_hook {
                let fail = hook.fail_once.swap(false, Ordering::SeqCst);
                if fail || hook.stall_once.swap(false, Ordering::SeqCst) {
                    hook.entered.notify_one();
                    hook.release.notified().await;
                    if fail {
                        panic!("injected relay notification worker failure");
                    }
                }
            }
            let classify = |reason: &str| classifier.subscription_refusal(reason);
            if handle_relay_notification(notification, &adapter, account_id.as_ref(), &classify)
                .await
            {
                worker_pending.fetch_sub(1, Ordering::SeqCst);
                return true;
            }
            worker_pending.fetch_sub(1, Ordering::SeqCst);
        }
        false
    });
    let abort_on_drop = AbortTaskOnDrop(worker.abort_handle());
    loop {
        tokio::select! {
            worker_result = &mut worker => {
                drop(abort_on_drop);
                let mut outcome = abandoned_notification_lane(receiver, &pending, source.as_ref());
                if outcome.exit == RelayNotificationConsumerExit::Closed
                    && matches!(worker_result, Ok(true))
                {
                    outcome.exit = RelayNotificationConsumerExit::Shutdown;
                }
                return outcome;
            }
            update = receiver.next() => match update {
                Some(NotificationUpdate::Notification(notification)) => {
                    if matches!(notification, NostrSdkNotification::Shutdown) {
                        let abandoned = pending.load(Ordering::SeqCst);
                        return if abandoned > 0 {
                            RelayNotificationConsumerOutcome::lagged(
                                receiver,
                                source.as_ref(),
                                abandoned,
                            )
                        } else {
                            RelayNotificationConsumerOutcome::new(
                                receiver,
                                RelayNotificationConsumerExit::Shutdown,
                            )
                        };
                    }
                    pending.fetch_add(1, Ordering::SeqCst);
                    if let Err(error) = sender.try_send(notification) {
                        pending.fetch_sub(1, Ordering::SeqCst);
                        let skipped = match error {
                            mpsc::error::TrySendError::Full(_) => {
                                1 + pending.load(Ordering::SeqCst)
                            }
                            mpsc::error::TrySendError::Closed(_) => 1,
                        };
                        return RelayNotificationConsumerOutcome::lagged(
                            receiver,
                            source.as_ref(),
                            skipped,
                        );
                    }
                    #[cfg(test)]
                    source.notification_queued();
                }
                Some(NotificationUpdate::Lagged { skipped }) => {
                    let abandoned = pending.load(Ordering::SeqCst);
                    let lag_bounds = source.record_lag(skipped, abandoned);
                    return RelayNotificationConsumerOutcome {
                        receiver,
                        exit: RelayNotificationConsumerExit::Lagged(skipped.saturating_add(abandoned)),
                        lag_bounds,
                    };
                }
                None => {
                    let abandoned = pending.load(Ordering::SeqCst);
                    return if abandoned > 0 {
                        RelayNotificationConsumerOutcome::lagged(
                            receiver,
                            source.as_ref(),
                            abandoned,
                        )
                    } else {
                        RelayNotificationConsumerOutcome::new(
                            receiver,
                            RelayNotificationConsumerExit::Closed,
                        )
                    };
                }
            }
        }
    }
}

async fn handle_relay_notification(
    notification: NostrSdkNotification,
    adapter: &NostrTransportAdapter,
    account_id: Option<&MemberId>,
    classify_closed: &(dyn Fn(&str) -> transport_nostr_adapter::SubscriptionRefusal + Sync),
) -> bool {
    let shutdown = transport_nostr_adapter::dispatch_sdk_replay(
        notification,
        adapter,
        account_id,
        classify_closed,
    )
    .await;
    if shutdown {
        tracing::debug!(
            target: "marmot_app::relay_plane",
            method = "handle_relay_notification",
            "SDK relay pool shutdown observed"
        );
    }
    shutdown
}

fn recover_relay_notification_forwarder(
    transport: &Arc<RelayPlaneTransport>,
    exit: RelayNotificationConsumerExit,
) {
    recover_relay_notification_forwarder_scoped(
        transport,
        exit,
        NostrNotificationLossBounds::account_wide(NostrNotificationLossFloor::Unbounded),
        None,
    );
}

/// Record a consumer lag, or close account delivery after an unexpected
/// consumer exit. `lag_bounds` are the source's REQ floors read at the lag.
fn recover_relay_notification_forwarder_scoped(
    transport: &Arc<RelayPlaneTransport>,
    exit: RelayNotificationConsumerExit,
    lag_bounds: NostrNotificationLossBounds,
    account_id: Option<&MemberId>,
) {
    let account_count = match account_id {
        Some(account_id) => usize::from(
            account_deliveries_read(&transport.account_deliveries).contains_key(account_id),
        ),
        None => account_deliveries_read(&transport.account_deliveries).len(),
    };
    match exit {
        RelayNotificationConsumerExit::Lagged(skipped) => {
            // The live subscriptions did not fail; only this consumer fell
            // behind, and it resumes on the same receiver. Keep every route
            // open and charge the loss to each account's current generation.
            // A shared receiver's floors cannot be attributed to one account.
            let bounds = if account_id.is_some() {
                lag_bounds
            } else {
                NostrNotificationLossBounds::account_wide(NostrNotificationLossFloor::Unbounded)
            };
            let routes = account_deliveries_read(&transport.account_deliveries);
            for (route_account_id, route) in routes.iter() {
                if account_id.is_some_and(|account_id| account_id != route_account_id) {
                    continue;
                }
                record_account_loss(&route.overflow, Some(&route.sender), &bounds);
            }
            drop(routes);
            let health = &transport.notification_forwarder_health;
            RelayNotificationForwarderHealth::increment(&health.restarts, 1);
            RelayNotificationForwarderHealth::increment(&health.lag_incidents, 1);
            RelayNotificationForwarderHealth::increment(&health.lagged_notifications, skipped);
            tracing::warn!(
                target: "marmot_app::relay_plane",
                method = "recover_relay_notification_forwarder",
                skipped_notifications = skipped,
                affected_accounts = account_count,
                bounded = bounds.floor.since_seconds().is_some(),
                routed = bounds.routes.is_some(),
                "relay notification consumer lagged; recording notification loss for comparison recovery",
            );
        }
        RelayNotificationConsumerExit::Closed => {
            // The receiver itself is gone. Latch account-local evidence before
            // closing receivers, and keep each loss handle in the account
            // registry so replacement adapters inherit the fence; the shared
            // router never awaits account database I/O.
            let mut routes = account_deliveries_write(&transport.account_deliveries);
            for (route_account_id, route) in routes.iter_mut() {
                if account_id.is_some_and(|account_id| account_id != route_account_id) {
                    continue;
                }
                // A queued control record dies with the closed queue. The
                // account worker persists pending loss before returning the
                // close signal.
                let generation = route
                    .overflow
                    .inner
                    .lock()
                    .unwrap_or_else(|poisoned| poisoned.into_inner())
                    .generation;
                route.overflow.cancel_signal(generation);
                let (closed_sender, closed_receiver) = mpsc::channel(1);
                drop(closed_receiver);
                route.sender = closed_sender;
            }
            drop(routes);
            let health = &transport.notification_forwarder_health;
            RelayNotificationForwarderHealth::increment(&health.restarts, 1);
            RelayNotificationForwarderHealth::increment(&health.unexpected_exits, 1);
            tracing::warn!(
                target: "marmot_app::relay_plane",
                method = "recover_relay_notification_forwarder",
                affected_accounts = account_count,
                "relay notification consumer exited unexpectedly; restarting inbound delivery",
            );
        }
        RelayNotificationConsumerExit::Shutdown => {}
    }
}

impl MarmotRelayPlaneAccountAdapter {
    /// Send no event over a relay's known limit (mdk#2334).
    #[doc(hidden)]
    pub fn with_event_limits(mut self, limits: Arc<dyn RelayEventLimits>) -> Self {
        self.event_limits = Some(limits);
        self
    }

    #[doc(hidden)]
    pub async fn publish_signed_event(
        &self,
        fallback: &dyn NostrRelayClient,
        endpoints: &[TransportEndpoint],
        event: &NostrTransportEvent,
        required_acks: usize,
    ) -> Result<NostrPublishOutcome, TransportAdapterError> {
        let Some(publisher) = self
            .anonymous_publish_client
            .as_ref()
            .filter(|_| event.sig.is_some())
        else {
            // Unsigned events and custom transports retain their supplied publisher.
            return fallback
                .publish_event_for_account(&self.account_id, endpoints, event, required_acks)
                .await;
        };
        let deadline = tokio::time::Instant::now() + ACCOUNT_PUBLISH_WAIT;
        let required_acks = required_acks.max(1);
        let unknown = |endpoint: &TransportEndpoint| TransportEndpointFailure {
            size_class: None,
            endpoint: endpoint.clone(),
            reason: "publish acknowledgement unknown".to_owned(),
            kind: TransportEndpointFailureKind::PossiblyExposed,
            rejection_category: None,
        };
        let mut outcome = NostrPublishOutcome::default();
        let mut pending = HashSet::new();
        let mut publishes = FuturesUnordered::new();
        let mut retrying = HashSet::new();
        let mut retries = FuturesUnordered::new();
        for endpoint in endpoints {
            if !pending.insert(endpoint) {
                continue;
            }
            // The SDK aggregate error cannot carry partial acknowledgements.
            // Race endpoint results instead so public and authenticated receipts
            // can satisfy the same quorum without discarding either.
            publishes.push(async move {
                let result = publisher
                    .publish_event(std::slice::from_ref(endpoint), event, 1)
                    .await;
                (endpoint, false, result)
            });
        }
        while !publishes.is_empty() || !retries.is_empty() {
            let (is_retry, next) = tokio::select! {
                next = timeout_at(deadline, publishes.next()), if !publishes.is_empty() => (false, next),
                next = timeout_at(deadline, retries.next()), if !retries.is_empty() => (true, next),
            };
            match next {
                Ok(Some((endpoint, earlier_exposure, result))) => {
                    if is_retry {
                        retrying.remove(endpoint);
                    } else {
                        pending.remove(endpoint);
                    }
                    let preserve_prior_exposure = |mut failure: TransportEndpointFailure| {
                        if earlier_exposure && failure.endpoint == *endpoint {
                            failure.kind = TransportEndpointFailureKind::PossiblyExposed;
                            failure.size_class = None;
                        }
                        failure
                    };
                    match result {
                        Ok(receipt) => {
                            let endpoint_accepted = receipt
                                .accepted
                                .iter()
                                .any(|accepted| accepted.endpoint == *endpoint);
                            outcome.message_id = receipt.message_id.or(outcome.message_id);
                            outcome.accepted.extend(receipt.accepted);
                            outcome.failed.extend(
                                receipt
                                    .failed
                                    .into_iter()
                                    .filter(|failure| {
                                        // A matching ACK settles this endpoint even
                                        // if a supplied receipt retains old failures.
                                        !(earlier_exposure
                                            && endpoint_accepted
                                            && failure.endpoint == *endpoint)
                                    })
                                    .map(preserve_prior_exposure),
                            );
                        }
                        Err(error) => {
                            outcome.message_id =
                                error.publish_message_id().cloned().or(outcome.message_id);
                            let auth_required = !is_retry
                                && error.publish_endpoint_failures().iter().any(|failure| {
                                    failure.endpoint == *endpoint
                                        && failure.rejection_category
                                            == Some(
                                                TransportEndpointRejectionCategory::AuthRequired,
                                            )
                                });
                            if auth_required {
                                let earlier_exposure =
                                    error.publish_endpoint_failures().iter().any(|failure| {
                                        failure.endpoint == *endpoint
                                            && failure.kind
                                                == TransportEndpointFailureKind::PossiblyExposed
                                    });
                                retrying.insert(endpoint);
                                retries.push(async move {
                                    (
                                        endpoint,
                                        earlier_exposure,
                                        fallback
                                            .publish_event_for_account(
                                                &self.account_id,
                                                std::slice::from_ref(endpoint),
                                                event,
                                                1,
                                            )
                                            .await,
                                    )
                                });
                            } else if error.publish_endpoint_failures().is_empty() {
                                if !is_retry {
                                    return Err(error);
                                }
                                outcome.failed.push(unknown(endpoint));
                            } else {
                                outcome.failed.extend(
                                    error
                                        .publish_endpoint_failures()
                                        .iter()
                                        .cloned()
                                        .map(preserve_prior_exposure),
                                );
                            }
                        }
                    }
                    if outcome.accepted.len() >= required_acks {
                        // Quorum cancels unfinished publications, including
                        // authenticated fallbacks. Their initial rejection no
                        // longer proves non-exposure: retain the same unknown
                        // evidence as deadline cancellation for exact-event retry.
                        publishes.clear();
                        retries.clear();
                        outcome.failed.extend(pending.drain().map(unknown));
                        outcome.failed.extend(retrying.drain().map(unknown));
                    }
                }
                Ok(None) => {}
                Err(_) => {
                    // This includes in-flight authenticated retries. Keeping
                    // their earlier AuthRequired would falsely prove no send.
                    outcome.failed.extend(pending.drain().map(unknown));
                    outcome.failed.extend(retrying.drain().map(unknown));
                    break;
                }
            }
        }
        if outcome.accepted.len() >= required_acks {
            return Ok(outcome);
        }
        let mut failure = TransportPublishFailure::with_endpoint_failures(
            format!(
                "insufficient publish acknowledgements: accepted {} of required {}",
                outcome.accepted.len(),
                required_acks
            ),
            outcome.failed,
        );
        failure.message_id = outcome
            .message_id
            .or_else(|| hex::decode(&event.id).ok().map(cgka_traits::MessageId::new));
        Err(TransportAdapterError::PublishEndpoints(failure))
    }

    /// Explicit catch-up must reissue even settled, matching subscriptions.
    /// Cancellation leaves reuse disabled until an activation succeeds.
    #[doc(hidden)]
    pub async fn require_fresh_activation(&self) {
        *self.incremental_activation.lock().await = None;
    }

    /// The account this adapter is bound to — the `MemberId` every subscription
    /// issued through it carries (activation and group sync reject any other
    /// id), and the key its registrations bucket under on the shared relay
    /// plane. Draining the `subscription_rebuild` row uses this so a rebuild is
    /// attributed to exactly the account whose subscribes produced it.
    #[doc(hidden)]
    pub fn account_id(&self) -> &MemberId {
        &self.account_id
    }

    /// Read only the live activation ordinal for worker-owned comparison
    /// admission. An SDK connection generation is a different lifetime.
    #[doc(hidden)]
    pub async fn account_subscription_attempt(&self) -> Option<SubscriptionAttempt> {
        self.relay_plane
            .inner
            .transport
            .adapter
            .account_subscription_attempt(&self.account_id)
            .await
    }

    #[doc(hidden)]
    pub async fn reconcile_inbox_history(
        &self,
        endpoints: Vec<TransportEndpoint>,
        local_items: &[NostrReconciliationItem],
        reconcile_since: u64,
        reconcile_until: u64,
        progress: &dyn transport_nostr_adapter::NostrReconciliationProgress,
        deadline: tokio::time::Instant,
    ) -> Result<
        Option<(
            NostrReconciliationSummary,
            Vec<transport_nostr_adapter::NostrRelayEvent>,
        )>,
        TransportAdapterError,
    > {
        let Some(client) = &self.relay_plane.inner.transport.sdk_relay_client else {
            return Ok(None);
        };
        let activation = self
            .relay_plane
            .inner
            .relay_safety
            .sanitize_activation(TransportAccountActivation {
                account_id: self.account_id.clone(),
                inbox_endpoints: endpoints,
                group_subscriptions: Vec::new(),
                since: None,
            })
            .map_err(TransportAdapterError::Subscription)?;
        let result = client
            .reconcile_subscription(
                NostrSubscription::AccountInbox {
                    account_id: self.account_id.clone(),
                    endpoints: activation.inbox_endpoints,
                    since: None,
                    // A one-shot reconciliation REQ is not owned by an account
                    // activation and never feeds the replay-coverage gate.
                    attempt: SubscriptionAttempt::INITIAL,
                },
                local_items,
                reconcile_since,
                reconcile_until,
                progress,
                deadline,
            )
            .await;
        let metric = result
            .as_ref()
            .map(|(summary, _)| summary.clone())
            .unwrap_or_default();
        self.relay_plane
            .inner
            .transport
            .adapter
            .record_reconciliation(&metric)
            .await;
        Ok(Some(result?))
    }

    #[doc(hidden)]
    pub async fn reconcile_group_history(
        &self,
        group: TransportGroupSubscription,
        local_items: &[NostrReconciliationItem],
        reconcile_since: u64,
        reconcile_until: u64,
        progress: &dyn transport_nostr_adapter::NostrReconciliationProgress,
        deadline: tokio::time::Instant,
    ) -> Result<
        Option<(
            NostrReconciliationSummary,
            Vec<transport_nostr_adapter::NostrRelayEvent>,
        )>,
        TransportAdapterError,
    > {
        let Some(client) = &self.relay_plane.inner.transport.sdk_relay_client else {
            return Ok(None);
        };
        let sync = self
            .relay_plane
            .inner
            .relay_safety
            .sanitize_group_sync(TransportGroupSync {
                account_id: self.account_id.clone(),
                group_subscriptions: vec![group],
                since: None,
            })
            .map_err(TransportAdapterError::Subscription)?;
        let group = sync.group_subscriptions.into_iter().next().ok_or_else(|| {
            TransportAdapterError::Subscription(
                "reconciliation group subscription was empty".to_owned(),
            )
        })?;
        let result = client
            .reconcile_subscription(
                NostrSubscription::Group {
                    account_id: self.account_id.clone(),
                    group_id: group.group_id,
                    transport_group_id: group.transport_group_id,
                    endpoints: group.endpoints,
                    since: None,
                    // A one-shot reconciliation REQ is not owned by an account
                    // activation and never feeds the replay-coverage gate.
                    attempt: SubscriptionAttempt::INITIAL,
                },
                local_items,
                reconcile_since,
                reconcile_until,
                progress,
                deadline,
            )
            .await;
        let metric = result
            .as_ref()
            .map(|(summary, _)| summary.clone())
            .unwrap_or_default();
        self.relay_plane
            .inner
            .transport
            .adapter
            .record_reconciliation(&metric)
            .await;
        Ok(Some(result?))
    }

    /// This account's deliveries for one owned comparison event. The worker
    /// admits them directly, never through the live queue.
    #[doc(hidden)]
    pub async fn recovered_deliveries(
        &self,
        event: transport_nostr_adapter::NostrRelayEvent,
    ) -> Result<Vec<TransportDelivery>, TransportAdapterError> {
        self.relay_plane
            .inner
            .transport
            .adapter
            .reconciled_deliveries(&self.account_id, event)
            .await
    }

    /// Install a group's post-join maintenance REQ, floored at `since` so a
    /// notification lag while it is live stays bounded. `None` requests the
    /// group's full history.
    #[doc(hidden)]
    pub async fn install_group_maintenance_subscription(
        &self,
        group: TransportGroupSubscription,
        recovery_attempt: u64,
        since: Option<Timestamp>,
    ) -> Result<String, TransportAdapterError> {
        let sync = self
            .relay_plane
            .inner
            .relay_safety
            .sanitize_group_sync(TransportGroupSync {
                account_id: self.account_id.clone(),
                group_subscriptions: vec![group],
                since: None,
            })
            .map_err(TransportAdapterError::Subscription)?;
        let group = sync.group_subscriptions.into_iter().next().ok_or_else(|| {
            TransportAdapterError::Subscription(
                "maintenance group subscription was empty".to_owned(),
            )
        })?;
        self.relay_plane
            .inner
            .transport
            .adapter
            .install_group_maintenance_recovery_subscription(
                &self.account_id,
                &group,
                recovery_attempt,
                since,
            )
            .await
    }

    #[doc(hidden)]
    pub async fn group_maintenance_any_eose(&self, subscription_id: &str) -> Option<bool> {
        self.relay_plane
            .inner
            .transport
            .adapter
            .subscription_any_eose(subscription_id)
            .await
    }

    /// Whether everything the post-join maintenance session delivered so far
    /// has been ingested. See [`NostrTransportAdapter::subscription_placed`].
    #[doc(hidden)]
    pub async fn group_maintenance_placed(&self, subscription_id: &str) -> Option<bool> {
        self.relay_plane
            .inner
            .transport
            .adapter
            .subscription_placed(subscription_id)
            .await
    }

    /// Operational capability identity only; SDK presence proves neither
    /// endpoint availability nor exhaustive historical coverage.
    #[doc(hidden)]
    pub fn recovery_comparison_capability_key(&self) -> &'static [u8] {
        if self.relay_plane.inner.transport.sdk_relay_client.is_some() {
            b"sdk-bounded-comparison-v1"
        } else {
            b"no-sdk-reconciliation-v1"
        }
    }

    #[doc(hidden)]
    pub async fn group_maintenance_endpoint_eose(
        &self,
        subscription_id: &str,
        endpoint: &TransportEndpoint,
    ) -> Option<bool> {
        self.relay_plane
            .inner
            .transport
            .adapter
            .subscription_endpoint_eose(subscription_id, endpoint)
            .await
    }

    /// End-of-stored-events progress across this account activation's frozen
    /// endpoint coverage snapshot.
    ///
    /// The epoch-gap backfill drain reads this to tell a relay that has
    /// finished replaying stored history from one that has simply gone quiet.
    #[doc(hidden)]
    pub async fn account_subscription_eose(&self) -> AccountSubscriptionEose {
        let mut eose = self
            .relay_plane
            .inner
            .transport
            .adapter
            .account_subscription_eose(&self.account_id)
            .await;
        if self.delivery_overflow.blocks_ordinary_eose() {
            eose.with_eose = 0;
        }
        eose
    }

    /// Current replies may finish even when the recovery owner still owes the
    /// original query's history. Use only for waits and subscription lifecycle;
    /// cursor and coverage qualification retain their independent guards.
    #[doc(hidden)]
    pub async fn account_subscription_input(&self) -> AccountSubscriptionInput {
        let mut input = self
            .relay_plane
            .inner
            .transport
            .adapter
            .account_subscription_input(&self.account_id)
            .await;
        if self.delivery_overflow.blocks_ordinary_eose() {
            input.complete = false;
            input.any = false;
        }
        input
    }

    #[doc(hidden)]
    pub async fn receive_account_delivery(
        &self,
    ) -> Result<Option<AccountDeliveryReceive>, TransportAdapterError> {
        let event = self.delivery_rx.lock().await.recv().await;
        Ok(event.map(|event| self.account_delivery_receive(event)))
    }

    /// The next queued item, if one is ready now.
    #[doc(hidden)]
    pub fn try_receive_account_delivery(&self) -> Option<AccountDeliveryReceive> {
        let event = self.delivery_rx.try_lock().ok()?.try_recv().ok()?;
        Some(self.account_delivery_receive(event))
    }

    /// Wait for a queued item, or for spilled rows to become durable. A
    /// pending spill wakeup wins, so a continuously ready live queue cannot
    /// hide it; it fires once per writer batch, so live input is not starved.
    #[doc(hidden)]
    pub async fn receive_account_delivery_or_spill(&self) -> AccountDeliveryWait {
        let mut delivery_rx = self.delivery_rx.lock().await;
        tokio::select! {
            biased;
            () = self.delivery_overflow.spill_ready.notified() => AccountDeliveryWait::SpillReady,
            event = delivery_rx.recv() => match event {
                Some(event) => AccountDeliveryWait::Received(self.account_delivery_receive(event)),
                None => AccountDeliveryWait::Closed,
            },
        }
    }

    fn account_delivery_receive(&self, event: AccountDeliveryEvent) -> AccountDeliveryReceive {
        match event {
            AccountDeliveryEvent::Delivery(delivery) => {
                // Taken, it still caps every commit until its consumer
                // releases it: a failed ingest's checkpoint must not pass it.
                self.delivery_overflow.take(
                    self.delivery_epoch,
                    account_delivery_restart_key(&delivery),
                    &delivery.message.id,
                );
                AccountDeliveryReceive::Delivery(delivery)
            }
            AccountDeliveryEvent::Overflow { generation } => {
                AccountDeliveryReceive::Overflow(self.delivery_overflow.consume_signal(generation))
            }
        }
    }

    fn cursor_lookback_secs(&self) -> Option<u64> {
        self.relay_plane.subscription_rebuild_lookback_secs()
    }

    /// Borrow wire admission while retaining its source-generation guard.
    /// Even with no active sources, the returned status borrows this account
    /// and adapter; absence is not a synthetic replay completion certificate.
    #[doc(hidden)]
    pub async fn replay_admission(&self) -> ReplayAdmissionStatus<'_> {
        let mut status = self
            .relay_plane
            .inner
            .transport
            .adapter
            .take_replay_status(&self.account_id)
            .await;
        if let Some(loss) = status.loss {
            import_replay_loss(&self.relay_plane.inner.transport, &self.account_id, loss);
        }
        let placement = self
            .delivery_overflow
            .inner
            .lock()
            .unwrap_or_else(|p| p.into_inner());
        status.spill_pending = placement.spill_in_flight > 0;
        status
    }

    /// Test hook: advance this account's replay generation, as a source change
    /// would.
    #[cfg(feature = "test-support")]
    pub async fn change_replay_sources_for_test(&self) {
        self.relay_plane
            .inner
            .transport
            .adapter
            .change_replay_sources_for_test(&self.account_id)
            .await;
    }

    /// Observe source revisions and admission progress without scheduling IO.
    #[doc(hidden)]
    pub async fn replay_changes(&self) -> Option<tokio::sync::watch::Receiver<u64>> {
        self.relay_plane
            .inner
            .transport
            .adapter
            .replay_changes(&self.account_id)
            .await
    }

    /// Hold this account's groups' live frames behind the owner's fences from
    /// now on (mdk#2660). Call it before activation, so the first counts are
    /// held too.
    #[doc(hidden)]
    pub async fn enable_replay_fencing(&self) {
        self.relay_plane
            .inner
            .transport
            .adapter
            .enable_replay_fencing(&self.account_id)
            .await;
    }

    /// Apply the owner's fences on this account's groups' live frames, as of
    /// the status `plan`'s ticket came from. Schedules no IO.
    #[doc(hidden)]
    pub async fn fence_replay(&self, plan: &ReplayFencePlan) {
        self.relay_plane
            .inner
            .transport
            .adapter
            .fence_replay(&self.account_id, plan)
            .await;
    }

    /// Register ordered native input before activation. Retained group spill
    /// joins the recovered range before the owner decrypts any of its frames.
    #[doc(hidden)]
    pub async fn open_replay_ingress(
        &self,
        retained: Vec<TransportDelivery>,
    ) -> Result<transport_nostr_adapter::ReplayIngress, TransportAdapterError> {
        self.relay_plane
            .inner
            .transport
            .adapter
            .open_replay_ingress(&self.account_id, retained)
            .await
    }

    /// How far a commit may promote the transport cursor, decided under the
    /// router's placement lock: at most `candidate`, and never past a
    /// delivery still queued, or taken and not yet released, that the
    /// persisted cursor still lets a restart fetch. `None` while loss or a
    /// spill hand-off is pending. The caller persists the larger of this and
    /// its current cursor. The restart floor rises here, before the commit's
    /// save runs, so a delivery that arrives during that save and falls below
    /// it is spilled rather than queued.
    /// `authorize` consults only cached account-owner policy while the source
    /// guard and placement lock are held; it must not perform DB or network IO.
    #[doc(hidden)]
    pub fn seal_transport_cursor(
        &self,
        candidate: Option<u64>,
        authorize: impl FnOnce(&ReplayAdmissionStatus<'_>) -> bool,
    ) -> Option<u64> {
        // Staged replay is not under the router's placement lock yet. A
        // synchronous checkpoint must not pass it. Contention conservatively
        // defers the cursor; never block an account worker to read admission.
        let mut replay = self
            .relay_plane
            .inner
            .transport
            .adapter
            .take_replay_status(&self.account_id)
            .now_or_never()?;
        if let Some(loss) = replay.loss {
            import_replay_loss(&self.relay_plane.inner.transport, &self.account_id, loss);
        }
        self.delivery_overflow.seal_cursor(
            self.delivery_epoch,
            self.cursor_lookback_secs(),
            candidate,
            &mut replay,
            authorize,
        )
    }

    /// The delivery of event `id` no longer needs a restart to fetch it: its
    /// ingest is durable, or its consumer dropped it on purpose, as a
    /// duplicate or as input the account keeps no trace of by design. Its
    /// consumer calls this before the save that follows, so a committed
    /// delivery never holds the cursor back. A failed ingest keeps its pin
    /// until a redelivery succeeds, retained post-ingest projection completes,
    /// or its queue generation ends. An event this
    /// queue holds no pin for, such as one read back from the spill, has
    /// nothing to release.
    #[doc(hidden)]
    pub fn release_account_delivery(&self, id: &MessageId) {
        self.delivery_overflow.release(self.delivery_epoch, id);
    }

    /// A sealed commit's save failed and `restored` is still the persisted
    /// cursor. Undo the floor its seal raised.
    #[doc(hidden)]
    pub fn unseal_transport_cursor(&self, restored: Option<u64>) {
        self.delivery_overflow
            .unseal_cursor(self.cursor_lookback_secs(), restored);
    }

    /// A settled commit (a drain checkpoint, settled loss or a retired
    /// notice) saved what its seal reached, so the settled floor rises to it
    /// and the router stops spilling below. `reached` is what the seal
    /// returned, not the cursor the commit persisted, which may be an earlier
    /// live promotion's. A live commit never settles.
    #[doc(hidden)]
    pub fn settle_transport_cursor(&self, reached: Option<u64>) {
        self.delivery_overflow
            .settle_cursor(self.cursor_lookback_secs(), reached, false);
    }

    /// Record the cursor the account opened with.
    #[doc(hidden)]
    pub fn open_transport_cursor(&self, persisted: Option<u64>) {
        self.delivery_overflow
            .settle_cursor(self.cursor_lookback_secs(), persisted, true);
    }

    /// Whether the account has a settled cursor floor: the one it opened
    /// with, or one a drain checkpoint, settled loss or retired notice made
    /// durable. Only the account worker settles, so this cannot change
    /// between the worker's read and its next seal.
    #[doc(hidden)]
    pub fn transport_cursor_settled(&self) -> bool {
        self.delivery_overflow
            .inner
            .lock()
            .unwrap_or_else(|p| p.into_inner())
            .admission
            .settled_since
            .is_some()
    }

    /// Cumulative account-queue placement counts for the transport-cursor
    /// audit row. Relaxed counters: a reader sees a recent value, never a
    /// torn one, and the row reports differences between two reads.
    #[doc(hidden)]
    pub fn delivery_placement_counts(&self) -> AccountDeliveryPlacementCounts {
        let metrics = &self.delivery_overflow.metrics;
        AccountDeliveryPlacementCounts {
            spilled_below_floor: metrics.spill_diverted_below_floor.load(Ordering::Relaxed),
            spilled_queue_full: metrics.spill_diverted_queue_full.load(Ordering::Relaxed),
            spill_already_seen: metrics.spill_already_seen.load(Ordering::Relaxed),
            queue_dropped: metrics.dropped.load(Ordering::Relaxed),
        }
    }

    /// Process-local overflow evidence becomes visible at the exact omission,
    /// before marker I/O or the queued control record can complete.
    #[doc(hidden)]
    pub fn notification_loss_persisted(&self, observed: AccountDeliveryOverflow) {
        self.delivery_overflow.notification_persisted(observed);
    }

    /// Unacknowledged loss, or a spilled delivery that is not yet durable,
    /// keeps the transport cursor from advancing.
    #[doc(hidden)]
    pub fn delivery_loss_blocks_cursor(&self) -> bool {
        let state = self
            .delivery_overflow
            .inner
            .lock()
            .unwrap_or_else(|p| p.into_inner());
        state.pending || state.spill_in_flight > 0
    }

    #[doc(hidden)]
    pub fn unpersisted_notification_loss(&self) -> Option<AccountDeliveryOverflow> {
        let state = self
            .delivery_overflow
            .inner
            .lock()
            .unwrap_or_else(|p| p.into_inner());
        (state.pending && state.notification_losses > state.notification_imported)
            .then(|| AccountDeliveryOverflowState::snapshot(&state))
    }

    /// The routes `observed`'s notification and replay loss reached, each with
    /// the lowest REQ floor it could have lost from, or `None` when some loss
    /// in its generation reached unknown routes. A generation only adds routes
    /// and lowers their floors, so reading after the snapshot still bounds
    /// every loss the snapshot counted. A later generation's routes say
    /// nothing about this one.
    #[doc(hidden)]
    pub fn notification_loss_routes(
        &self,
        observed: &AccountDeliveryOverflow,
    ) -> Option<Vec<(DeliveryRoute, u64)>> {
        let state = self
            .delivery_overflow
            .inner
            .lock()
            .unwrap_or_else(|p| p.into_inner());
        (state.generation == observed.generation
            && !state.notification_routes_unknown
            && !state.notification_routes.is_empty())
        .then(|| {
            state
                .notification_routes
                .iter()
                .map(|(route, since)| (*route, *since))
                .collect()
        })
    }

    /// The routes `observed`'s dropped deliveries arrived on, each with the
    /// earliest `created_at` dropped there, or `None` when some drop in its
    /// generation had no route or time, or they touched more than
    /// [`MAX_DELIVERY_LOSS_ROUTES`]. The worker passes them with an
    /// observation past the durable marker count, whose extra drops no marker
    /// write has named yet. As for [`Self::notification_loss_routes`], reading
    /// after the snapshot still bounds every drop it counted, and another
    /// generation's or token's routes say nothing about these.
    #[doc(hidden)]
    pub fn delivery_loss_routes(
        &self,
        observed: &AccountDeliveryOverflow,
    ) -> Option<Vec<(DeliveryRoute, u64)>> {
        let state = self
            .delivery_overflow
            .inner
            .lock()
            .unwrap_or_else(|p| p.into_inner());
        (state.generation == observed.generation
            && state.marker_token == observed.marker_token
            && !state.dropped_routes_unknown
            && !state.dropped_routes.is_empty())
        .then(|| {
            state
                .dropped_routes
                .iter()
                .map(|(route, earliest)| (*route, *earliest))
                .collect()
        })
    }

    /// Whether a queued or deferred control record will still report the
    /// pending loss generation to this account's consumer.
    #[doc(hidden)]
    pub fn delivery_overflow_signal_outstanding(&self) -> bool {
        let state = self
            .delivery_overflow
            .inner
            .lock()
            .unwrap_or_else(|p| p.into_inner());
        state.pending && state.signal_queued
    }

    #[doc(hidden)]
    pub fn pending_delivery_overflow(&self) -> Option<AccountDeliveryOverflow> {
        self.delivery_overflow.pending_snapshot()
    }

    /// This account's delivery-loss state, so a test can check that a
    /// replacement adapter inherited it rather than starting fresh.
    #[cfg(any(test, feature = "test-support"))]
    #[doc(hidden)]
    pub fn delivery_overflow_identity_for_test(&self) -> DeliveryOverflowIdentity {
        DeliveryOverflowIdentity(self.delivery_overflow.clone())
    }

    /// Notification losses this account's delivery route has recorded,
    /// whether or not a loss is still pending.
    #[cfg(any(test, feature = "test-support"))]
    #[doc(hidden)]
    pub fn notification_losses_for_test(&self) -> u64 {
        self.delivery_overflow
            .inner
            .lock()
            .unwrap_or_else(|p| p.into_inner())
            .notification_losses
    }

    /// Begin (or resume after process restart) the recovery job required by a
    /// durable account-delivery loss marker.
    #[doc(hidden)]
    pub fn start_delivery_overflow_recovery(
        &self,
        durable_marker_token: u64,
    ) -> AccountDeliveryOverflow {
        self.delivery_overflow.start_recovery(durable_marker_token)
    }

    /// Resolve only the exact overflow prefix the replay started against. A
    /// newer omitted delivery keeps the generation pending and forces another
    /// unfloored attempt.
    #[doc(hidden)]
    pub fn finish_delivery_overflow_recovery(
        &self,
        attempt: AccountDeliveryOverflow,
    ) -> Option<u64> {
        self.delivery_overflow.finish_recovery(attempt)
    }

    #[doc(hidden)]
    pub fn restore_delivery_overflow_guard(&self, durable_marker_token: u64) {
        self.delivery_overflow
            .restore_recovery_guard(durable_marker_token);
    }

    #[doc(hidden)]
    pub fn record_delivery_overflow_recovery_success(&self, elapsed_ms: u64) {
        self.delivery_overflow.record_recovery_success(elapsed_ms);
    }

    /// The pending loss generation, including one a recovery attempt holds.
    #[doc(hidden)]
    pub fn pending_delivery_overflow_generation(&self) -> Option<AccountDeliveryOverflow> {
        self.delivery_overflow.pending_generation()
    }

    /// Release the cursor fence for loss the account owner retired, only if
    /// `observed` is still exactly the pending generation. Never counts a
    /// recovery success. Returns true when no plane loss remains pending.
    #[doc(hidden)]
    pub fn retire_delivery_overflow(&self, observed: Option<AccountDeliveryOverflow>) -> bool {
        self.delivery_overflow.retire_recovery(observed)
    }

    #[doc(hidden)]
    pub fn fail_delivery_overflow_recovery(&self) {
        self.delivery_overflow.fail_recovery();
    }

    #[doc(hidden)]
    pub async fn remove_group_maintenance_subscription(
        &self,
        subscription_id: &str,
    ) -> Result<(), TransportAdapterError> {
        self.relay_plane
            .inner
            .transport
            .adapter
            .remove_group_maintenance_recovery_subscription(&self.account_id, subscription_id)
            .await
    }
}

#[async_trait]
impl TransportAdapter for MarmotRelayPlaneAccountAdapter {
    async fn activate_account(
        &self,
        activation: TransportAccountActivation,
    ) -> Result<(), TransportAdapterError> {
        if activation.account_id != self.account_id {
            return Err(TransportAdapterError::AccountNotActive(
                activation.account_id,
            ));
        }
        let activation = self
            .relay_plane
            .inner
            .relay_safety
            .sanitize_activation(activation)
            .map_err(TransportAdapterError::Subscription)?;
        let incremental = activation.since.map(|since| IncrementalActivation {
            inbox_endpoints: activation.inbox_endpoints.clone(),
            since,
        });
        let mut previous = self.incremental_activation.lock().await;
        let reuse = incremental.is_some()
            && *previous == incremental
            && self.account_subscription_input().await.complete;
        // Cancellation or failure must force the next call through activation's
        // unconditional orphan-REQ cleanup. None always requests full history.
        *previous = None;
        let adapter = &self.relay_plane.inner.transport.adapter;
        if reuse {
            adapter
                .sync_account_groups(TransportGroupSync {
                    account_id: activation.account_id,
                    group_subscriptions: activation.group_subscriptions,
                    since: activation.since,
                })
                .await?;
        } else {
            adapter.activate_account(activation).await?;
        }
        *previous = incremental;
        Ok(())
    }

    async fn sync_account_groups(
        &self,
        sync: TransportGroupSync,
    ) -> Result<(), TransportAdapterError> {
        if sync.account_id != self.account_id {
            return Err(TransportAdapterError::AccountNotActive(sync.account_id));
        }
        let sync = self
            .relay_plane
            .inner
            .relay_safety
            .sanitize_group_sync(sync)
            .map_err(TransportAdapterError::Subscription)?;
        self.relay_plane
            .inner
            .transport
            .adapter
            .sync_account_groups(sync)
            .await
    }

    async fn deactivate_account(&self, account_id: &MemberId) -> Result<(), TransportAdapterError> {
        if account_id != &self.account_id {
            return Err(TransportAdapterError::AccountNotActive(account_id.clone()));
        }
        let mut activation = self.incremental_activation.lock().await;
        *activation = None;
        self.relay_plane
            .deactivate_account_context(account_id)
            .await
    }

    async fn publish(
        &self,
        request: TransportPublishRequest,
    ) -> Result<TransportPublishReport, TransportAdapterError> {
        if request.account_id != self.account_id {
            return Err(TransportAdapterError::AccountNotActive(request.account_id));
        }
        let request = self
            .relay_plane
            .inner
            .relay_safety
            .sanitize_publish_request(request)
            .map_err(TransportAdapterError::Publish)?;
        request.validate_envelope_matches_target()?;
        let event = NostrTransportEvent::from_transport_message(&request.message)
            .map_err(|e| TransportAdapterError::Publish(format!("Nostr payload: {e}")))?;
        // An event over a relay's known limit is never sent there: the relay
        // would refuse it, or close the socket on the oversized frame, which
        // reads as a possible exposure and resends the event (mdk#2334).
        let (endpoints, refused) = refuse_over_known_limits(
            self.event_limits.as_deref(),
            request.target.endpoints(),
            event_json_bytes(&event),
        )
        .await;
        if endpoints.is_empty() && !refused.is_empty() {
            // No relay was contacted, so no publish attempt is counted.
            return Err(refused_everywhere(
                request.target.endpoints(),
                refused,
                &event,
            ));
        }
        // Refused endpoints count against the request's threshold in the
        // report; the endpoints still sent to can meet at most their number.
        let required_acks = if refused.is_empty() {
            request.required_acks
        } else {
            request.required_acks.min(endpoints.len())
        };
        // Welcome fanout keeps its account publisher; a blocked inbox write
        // cannot stall the independent anonymous group publisher.
        let publication = async {
            if matches!(
                request.target,
                cgka_traits::TransportPublishTarget::Group { .. }
            ) {
                self.publish_signed_event(
                    self.publish_client.as_ref(),
                    &endpoints,
                    &event,
                    required_acks,
                )
                .await
            } else {
                self.publish_client
                    .publish_event_for_account(&self.account_id, &endpoints, &event, required_acks)
                    .await
            }
        };
        let outcome = match self
            .relay_plane
            .inner
            .transport
            .adapter
            .publish_event_with_client(publication, request.required_acks)
            .await
        {
            Ok(outcome) => outcome,
            Err(error) => {
                return Err(join_refusals(error, request.target.endpoints(), refused));
            }
        };
        let local_fanout_endpoints = if !outcome.accepted.is_empty() {
            outcome
                .accepted
                .iter()
                .map(|receipt| receipt.endpoint.clone())
                .collect::<Vec<_>>()
        } else if outcome.failed.is_empty() {
            endpoints
        } else {
            Vec::new()
        };
        if !local_fanout_endpoints.is_empty() {
            let mut local_message = request.message.clone();
            if let Some(message_id) = outcome.message_id.clone() {
                local_message.id = message_id;
            }
            self.relay_plane
                .inner
                .transport
                .adapter
                .deliver_local_publish(&local_message, &local_fanout_endpoints)
                .await?;
        }
        let mut report = publish_report_from_outcome(outcome, request);
        report.failed.extend(refused);
        Ok(report)
    }

    async fn receive(&self) -> Result<Option<TransportDelivery>, TransportAdapterError> {
        match self.receive_account_delivery().await? {
            Some(AccountDeliveryReceive::Delivery(delivery)) => Ok(Some(*delivery)),
            Some(AccountDeliveryReceive::Overflow(_)) => Err(TransportAdapterError::Other(
                "account delivery overflow recovery required".to_owned(),
            )),
            None => Ok(None),
        }
    }

    async fn size_class_over_known_limits(
        &self,
        message: &cgka_traits::TransportMessage,
        target: &cgka_traits::TransportPublishTarget,
    ) -> Option<cgka_traits::EventSizeClass> {
        // An unreadable payload is refused by `publish` itself.
        let event = NostrTransportEvent::from_transport_message(message).ok()?;
        // Judge the endpoints `publish` can reach, by the rule it applies to
        // each request, and refuse them by the limits it applies.
        let endpoints = self
            .relay_plane
            .inner
            .relay_safety
            .reachable_publish_endpoints(target);
        let (_, refused) = refuse_over_known_limits(
            self.event_limits.as_deref(),
            &endpoints,
            event_json_bytes(&event),
        )
        .await;
        cgka_traits::EventSizeClass::from_rejected_targets(&endpoints, &refused)
    }
}

/// Hand one delivery to its account without awaiting the account queue: a
/// single account whose receiver has stalled (full buffer) must not block the
/// shared router and back-pressure delivery for every other account (and,
/// upstream, the relay notification pipeline).
///
/// A full queue hands the delivery to the account's durable spill, and so
/// does a delivery that a transport-cursor checkpoint still saving, or a live
/// promotion alone, stopped a restart from fetching again: the queue never
/// holds one of those. Only when the spill cannot take it does the delivery
/// join an explicit loss generation, using the one channel slot reserved for
/// its control record.
fn route_account_delivery(transport: &RelayPlaneTransport, delivery: TransportDelivery) {
    let Some(route) = account_deliveries_read(&transport.account_deliveries)
        .get(&delivery.account_id)
        .cloned()
    else {
        return;
    };
    let queue_depth = route
        .sender
        .max_capacity()
        .saturating_sub(route.sender.capacity());
    route.overflow.observe_queue_depth(queue_depth);
    let created_at = delivery.message.timestamp.0;
    let delivery_route = DeliveryRoute::of(&delivery);
    let key = account_delivery_restart_key(&delivery);
    match route.overflow.place(
        route.epoch,
        key,
        route.sender.capacity() <= 1,
        route.spill.is_some(),
    ) {
        AccountDeliveryPlacement::Queue => {
            match route
                .sender
                .try_send(AccountDeliveryEvent::Delivery(Box::new(delivery)))
            {
                Ok(()) => {
                    let queue_depth = route
                        .sender
                        .max_capacity()
                        .saturating_sub(route.sender.capacity());
                    route.overflow.observe_queue_depth(queue_depth);
                }
                Err(mpsc::error::TrySendError::Full(_)) => {
                    route.overflow.unqueue(route.epoch, key);
                    // Only this router writes the route, and it reserves one
                    // control slot above, so reaching Full here indicates a
                    // violated queue invariant rather than ordinary
                    // backpressure.
                    tracing::warn!(
                        target: "marmot_app::relay_plane",
                        method = "route_account_delivery",
                        error_kind = "reserved_overflow_slot_unavailable",
                        "account delivery queue invariant failed",
                    );
                }
                Err(mpsc::error::TrySendError::Closed(_)) => {
                    route.overflow.unqueue(route.epoch, key);
                }
            }
        }
        AccountDeliveryPlacement::Spill => {
            let spilled = route
                .spill
                .as_ref()
                .is_some_and(|spill| spill.offer(delivery));
            if !spilled {
                // Record the loss before releasing the placement's fence, so
                // the cursor stays fenced throughout.
                omit_account_delivery(&route, Some(created_at), delivery_route);
                route.overflow.finish_spill(1, 0, 0);
            }
        }
        AccountDeliveryPlacement::Omit => {
            omit_account_delivery(&route, Some(created_at), delivery_route)
        }
    }
}

/// Omit one delivery from a full account queue into the current loss
/// generation, persisting its evidence before the control record is queued.
fn omit_account_delivery(
    route: &AccountDeliveryRoute,
    created_at: Option<u64>,
    delivery_route: Option<DeliveryRoute>,
) {
    let queue_depth = route
        .sender
        .max_capacity()
        .saturating_sub(route.sender.capacity());
    let signal_generation = route
        .overflow
        .record_drop(queue_depth, created_at, delivery_route);
    if let Some(marker) = route.recovery_marker.clone() {
        persist_queue_loss(&route.sender, &route.overflow, marker, signal_generation);
    } else if let Some(generation) = signal_generation {
        enqueue_account_delivery_overflow_signal(&route.sender, &route.overflow, generation);
    }
    tracing::warn!(
        target: "marmot_app::relay_plane",
        method = "omit_account_delivery",
        queue_depth,
        "omitting transport delivery: account delivery queue overflow recovery required",
    );
}

/// Count persistence is independent of whether a control record is already
/// queued. The one writer takes any deferred signal only after its latest
/// watermark is durable, including signals requested while that writer ran.
fn persist_queue_loss(
    sender: &mpsc::Sender<AccountDeliveryEvent>,
    overflow: &Arc<AccountDeliveryOverflowState>,
    marker: AccountDeliveryRecoveryMarker,
    signal: Option<u64>,
) {
    let generation = overflow
        .inner
        .lock()
        .unwrap_or_else(|p| p.into_inner())
        .generation;
    if let Some(generation) = signal {
        overflow.defer_marker_signal(generation);
    }
    if overflow.start_marker_persistence() {
        let sender = sender.clone();
        let overflow = overflow.clone();
        tokio::spawn(async move {
            overflow.persist_marker_before_drop(marker).await;
            if overflow.take_durable_marker_signal(generation) {
                enqueue_account_delivery_overflow_signal(&sender, &overflow, generation);
            }
        });
    } else if overflow.take_durable_marker_signal(generation) {
        enqueue_account_delivery_overflow_signal(sender, overflow, generation);
    }
}

/// Persist loss recorded after its route was retired. There is no queue for
/// a signal; whichever session comes next recovers from the durable marker.
fn persist_retired_queue_loss(
    overflow: &Arc<AccountDeliveryOverflowState>,
    marker: AccountDeliveryRecoveryMarker,
) {
    if overflow.start_marker_persistence() {
        let overflow = overflow.clone();
        tokio::spawn(async move { overflow.persist_marker_before_drop(marker).await });
    }
}

/// Charge replay loss to the account's process-lifetime fence, even if its
/// route was retired while the adapter status was being awaited.
fn import_replay_loss(
    transport: &RelayPlaneTransport,
    account_id: &MemberId,
    floor: NostrNotificationLossFloor,
) {
    // Staged replay loss is not attributed to routes.
    let loss = NostrNotificationLossBounds::account_wide(floor);
    let routes = account_deliveries_read(&transport.account_deliveries);
    if let Some(route) = routes.get(account_id) {
        record_account_loss(&route.overflow, Some(&route.sender), &loss);
    } else if let Some(overflow) = transport
        .account_overflow_states
        .lock()
        .unwrap_or_else(|p| p.into_inner())
        .get(account_id)
    {
        record_account_loss(overflow, None, &loss);
    }
}

fn record_account_loss(
    overflow: &AccountDeliveryOverflowState,
    sender: Option<&mpsc::Sender<AccountDeliveryEvent>>,
    loss: &NostrNotificationLossBounds,
) {
    // A queued or deferred control record already carries this generation;
    // the worker imports subsequent losses from the same pending snapshot.
    if let Some(generation) = overflow.record_notification_loss(loss)
        && let Some(sender) = sender
    {
        enqueue_account_delivery_overflow_signal(sender, overflow, generation);
    }
    if sender.is_none_or(mpsc::Sender::is_closed) {
        overflow
            .inner
            .lock()
            .unwrap_or_else(|p| p.into_inner())
            .retired_loss_unsignalled = true;
    }
}

fn enqueue_account_delivery_overflow_signal(
    sender: &mpsc::Sender<AccountDeliveryEvent>,
    overflow: &AccountDeliveryOverflowState,
    generation: u64,
) {
    match sender.try_send(AccountDeliveryEvent::Overflow { generation }) {
        Ok(()) => {}
        Err(error) => {
            overflow.cancel_signal(generation);
            tracing::warn!(
                target: "marmot_app::relay_plane",
                method = "spawn_router",
                error_kind = match error {
                    mpsc::error::TrySendError::Full(_) => "queue_full",
                    mpsc::error::TrySendError::Closed(_) => "queue_closed",
                },
                "could not enqueue account delivery overflow recovery signal",
            );
        }
    }
}

fn account_deliveries_read(
    deliveries: &RwLock<HashMap<MemberId, AccountDeliveryRoute>>,
) -> RwLockReadGuard<'_, HashMap<MemberId, AccountDeliveryRoute>> {
    deliveries
        .read()
        .unwrap_or_else(|poisoned| poisoned.into_inner())
}

fn account_deliveries_write(
    deliveries: &RwLock<HashMap<MemberId, AccountDeliveryRoute>>,
) -> RwLockWriteGuard<'_, HashMap<MemberId, AccountDeliveryRoute>> {
    deliveries
        .write()
        .unwrap_or_else(|poisoned| poisoned.into_inner())
}

fn publish_report_from_outcome(
    outcome: NostrPublishOutcome,
    request: TransportPublishRequest,
) -> TransportPublishReport {
    TransportPublishReport {
        message_id: outcome.message_id.unwrap_or(request.message.id),
        accepted: outcome.accepted,
        failed: outcome.failed,
        required_acks: request.required_acks,
    }
}

#[cfg(test)]
mod tests;