offline_protocol/

transport_manager.rs

//! Transport management for multi-transport protocol.
//!
//! This module provides the TransportManager which manages multiple transports
//! and uses DORS (Dynamic Offline Relay Switch) to select the optimal transport
//! for each message.

use crate::constants::{HOP_COUNT_EMA_ALPHA, LATENCY_EMA_ALPHA, OBSERVED_STATS_COMPACT_THRESHOLD};
use crate::events::{DorsEscalationPhase, DorsEscalationReasonCode, DorsReasonCode, Event};
use crate::telemetry::routing::{RoutingDecision, RoutingPhase, RoutingReasonCode};
use crate::{Error, Result};
use chrono::Utc;
use offline_protocol_core::{Message, Username, WireCodec};
use offline_protocol_router::{
    display_routing_score, DorsConfig, EscalationTriggerReason, TransportScore, TransportSelector,
};
use offline_protocol_transport::{
    nostr::ResolveRequest, Error as TransportError, Transport, TransportMetrics, TransportStatus,
    TransportType,
};
use std::collections::{HashMap, HashSet};
use std::sync::Arc;
use std::time::Instant;
use tracing::{debug, warn};

/// Manages multiple transports and handles transport selection.
pub struct TransportManager {
    /// Available transports mapped by type.
    ///
    /// Transports are fully interior-mutable (every `Transport` method takes
    /// `&self`), so they are shared as plain `Arc`s — no outer mutex
    /// serializing sends against status/metrics reads.
    transports: HashMap<TransportType, Arc<dyn Transport>>,

    /// Current active transport.
    current_transport: Option<TransportType>,

    /// Transport selector (DORS).
    selector: TransportSelector,

    /// Locally observed delivery outcomes used to enrich transport metrics.
    observations: HashMap<TransportType, ObservedStats>,

    /// Optional callback for DORS lifecycle/decision events (OFF-258).
    dors_event_callback: Option<Arc<dyn Fn(Event) + Send + Sync>>,

    /// Optional callback for structured routing decisions. Fires alongside
    /// every legacy `dors_event_callback` emission so the telemetry sink
    /// receives the richer `RoutingDecision` shape while `EventCallback`
    /// consumers continue to see the flattened `Event::Dors*` variants.
    routing_decision_callback: Option<Arc<dyn Fn(RoutingDecision) + Send + Sync>>,

    /// When `true`, every emitted [`RoutingDecision`] carries the full
    /// per-transport [`TransportScore`] breakdown in its `scores` field.
    /// Mirrors `TelemetryConfig::routing_diagnostic` and is toggled by
    /// `OfflineProtocol::install_telemetry_sink`.
    routing_diagnostic: bool,

    /// Last escalation trigger event emitted (reason, time) for dedupe window.
    last_escalation_trigger_emitted: Option<(DorsEscalationReasonCode, std::time::Instant)>,

    /// Peers known to decode the binary wire codec (from their signed key
    /// package). Messages addressed to a listed peer are stamped
    /// [`WireCodec::BinaryV1`] at send; all others stay JSON.
    peer_binary_wire: HashSet<String>,

    /// Host-reported battery level for this device, if the platform publishes
    /// one. Merged into every transport's metrics that does not report its own
    /// — see [`TransportManager::set_device_battery`].
    device_battery: Option<u8>,

    /// Host-reported charging state, applied alongside [`Self::device_battery`].
    device_is_charging: bool,
}

/// Carriers whose links are peer-to-peer, in the order a forwarding caller
/// should prefer them.
///
/// These are the only transports that can carry a *mesh hop*: a frame handed
/// straight to a neighbor's radio. Wi-Fi Direct leads because a link that
/// exists at all is the higher-bandwidth one; BLE is the carrier that is
/// almost always present. Everything else reaches peers through
/// infrastructure, which is not a hop — it is an exit from the mesh.
const MESH_TRANSPORTS: &[TransportType] = &[TransportType::WiFiDirect, TransportType::BLE];

/// A peer reachable over a direct mesh link.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct MeshNeighbor {
    /// User-level id of the neighbor.
    pub peer_id: String,
    /// Signal strength for the link, when the carrier measures one.
    pub rssi: Option<i16>,
    /// The carrier holding the link.
    pub transport: TransportType,
}

impl MeshNeighbor {
    /// How good this link is, on a 0–100 scale.
    ///
    /// Derived from signal strength where the carrier reports it, and a
    /// mid-scale value where it does not — an unmeasured link should not
    /// outrank a measured good one, nor be discarded like a measured bad one.
    pub fn link_quality(&self) -> u8 {
        match self.rssi {
            Some(rssi) => offline_protocol_transport::LinkQuality::from_rssi(rssi).value(),
            None => offline_protocol_transport::DEFAULT_LINK_QUALITY_WITHOUT_RSSI,
        }
    }
}

/// Dedupe window: don't re-emit same escalation trigger reason within this duration.
const ESCALATION_TRIGGER_DEDUPE_SECS: u64 = 30;

/// Cap on the per-peer binary-wire capability set. Keyed by the wire-claimed
/// peer id, so — like `key_package_sent_to` — it is cleared at capacity to bound
/// a forged-sender flood; the only cost of forgetting a peer is re-learning its
/// capability from the next key package (one JSON send in the meantime).
const MAX_PEER_BINARY_WIRE_ENTRIES: usize = 4096;

/// Whether the platform reports each send's outcome back to this transport,
/// so its own `success_count` / `failure_count` are real delivery evidence.
/// Exhaustive on purpose: a new transport must decide which side it is on.
fn has_send_confirmation_loop(transport_type: TransportType) -> bool {
    match transport_type {
        TransportType::Internet | TransportType::Reticulum | TransportType::Nostr => true,
        TransportType::BLE | TransportType::WiFiDirect => false,
    }
}

#[derive(Debug, Default, Clone)]
struct ObservedStats {
    success_count: u32,
    failure_count: u32,
    latency_ema: Option<f32>,
    hop_ema: Option<f32>,
    last_success_at: Option<Instant>,
}

impl ObservedStats {
    fn record_success(&mut self, latency_ms: u32, hop_count: u8) {
        self.success_count = self.success_count.saturating_add(1);
        self.latency_ema = Some(update_ema(
            self.latency_ema,
            latency_ms as f32,
            LATENCY_EMA_ALPHA,
        ));
        self.hop_ema = Some(update_ema(
            self.hop_ema,
            hop_count as f32,
            HOP_COUNT_EMA_ALPHA,
        ));
        self.last_success_at = Some(Instant::now());
        self.compact();
    }

    fn record_failure(&mut self) {
        self.failure_count = self.failure_count.saturating_add(1);
        self.compact();
    }

    fn compact(&mut self) {
        let total = self.success_count.saturating_add(self.failure_count);
        if total > OBSERVED_STATS_COMPACT_THRESHOLD {
            self.success_count /= 2;
            self.failure_count /= 2;
        }
    }

    fn delivery_ratio(&self) -> Option<f32> {
        let total = self.success_count + self.failure_count;
        if total == 0 {
            None
        } else {
            Some(self.success_count as f32 / total as f32)
        }
    }

    fn drop_ratio(&self) -> Option<f32> {
        self.delivery_ratio()
            .map(|ratio| (1.0 - ratio).clamp(0.0, 1.0))
    }

    /// Merges the engine's delivery observations into `metrics`.
    ///
    /// A transport with a send confirmation loop (Internet, Reticulum,
    /// Nostr: the platform reports each send's outcome back) owns its
    /// success and failure counts, and they are kept when it has any. Every
    /// other transport has no outcome of its own to report, so the engine's
    /// acknowledgements are its delivery evidence and always win. The
    /// previous rule was value-based ("keep the transport's counts when they
    /// are non-zero"), so one stray count from BLE's reassembly path silently
    /// discarded every later acknowledgement for that transport (#557).
    fn apply_to_metrics(&self, transport_type: TransportType, metrics: &mut TransportMetrics) {
        if self.success_count + self.failure_count > 0 {
            let transport_owns_counts = has_send_confirmation_loop(transport_type)
                && metrics.success_count + metrics.failure_count > 0;
            if !transport_owns_counts {
                metrics.success_count = self.success_count;
                metrics.failure_count = self.failure_count;
                metrics.delivery_ratio = self.delivery_ratio();
                metrics.drop_rate = self.drop_ratio();
            }
        }

        if let Some(latency) = self.latency_ema {
            metrics.latency_ms = Some(latency.round().clamp(0.0, u32::MAX as f32) as u32);
        }

        if let Some(hop) = self.hop_ema {
            metrics.average_hop_count = Some(hop);
        }
    }
}

fn update_ema(current: Option<f32>, new_value: f32, alpha: f32) -> f32 {
    let alpha = alpha.clamp(0.0, 1.0);
    match current {
        Some(existing) => existing * (1.0 - alpha) + new_value * alpha,
        None => new_value,
    }
}

impl TransportManager {
    /// Creates a new transport manager with default configuration.
    pub fn new(selector: TransportSelector) -> Self {
        Self {
            transports: HashMap::new(),
            current_transport: None,
            selector,
            observations: HashMap::new(),
            dors_event_callback: None,
            routing_decision_callback: None,
            routing_diagnostic: false,
            last_escalation_trigger_emitted: None,
            peer_binary_wire: HashSet::new(),
            device_battery: None,
            device_is_charging: false,
        }
    }

    /// Records the host's battery reading for this device.
    ///
    /// Battery is a property of the *device*, not of any one radio, but every
    /// consumer inside the SDK reads it from the per-transport
    /// [`TransportMetrics`] map: DORS energy scoring, the relay
    /// promote/demote policy, the forwarding battery floor, and the telemetry
    /// device-capability snapshot all resolve it through
    /// `device_battery_from_available`. This setter is therefore the single
    /// host-facing feed, and it is merged into the two snapshot loops that
    /// build that map.
    ///
    /// `level` is clamped to 0-100. A transport that reports its own battery
    /// level keeps it: the merge fills the field, never overwrites it.
    pub fn set_device_battery(&mut self, level: u8, is_charging: bool) {
        self.device_battery = Some(level.min(100));
        self.device_is_charging = is_charging;
    }

    /// Returns the host-reported `(battery_level, is_charging)` pair, if the
    /// platform has published one.
    pub fn device_battery(&self) -> (Option<u8>, bool) {
        (self.device_battery, self.device_is_charging)
    }

    /// Fills the device-level battery fields on one transport's metrics.
    ///
    /// **Fill-if-absent, never overwrite.** A transport that publishes its own
    /// `battery_level` is reporting something the host feed cannot know better
    /// (a peripheral's own cell, say), so its reading wins; the host value is
    /// the fallback for the ordinary case where no transport reports one at
    /// all. `is_charging` is only applied alongside a battery level that came
    /// from the host, so a transport's own (level, charging) pair is never
    /// split across two sources — the same coherence rule
    /// `device_battery_from_available` applies when choosing between
    /// transports.
    fn apply_device_battery(&self, metrics: &mut TransportMetrics) {
        let Some(level) = self.device_battery else {
            return;
        };
        if metrics.battery_level.is_none() {
            metrics.battery_level = Some(level);
            metrics.is_charging = self.device_is_charging;
        }
    }

    /// Returns whether a peer is recorded as able to decode the binary wire
    /// codec (and would therefore be sent binary frames).
    pub fn peer_supports_binary_wire(&self, peer_id: &str) -> bool {
        self.peer_binary_wire.contains(peer_id)
    }

    /// Records whether a peer can decode our binary wire frames, learned from
    /// its signed key package. Messages addressed to a capable peer are stamped
    /// [`WireCodec::BinaryV1`] at send; all others stay JSON.
    pub fn mark_peer_binary_wire(&mut self, peer_id: &str, supported: bool) {
        if supported {
            // Bound the wire-keyed set: clear at capacity, then insert.
            if !self.peer_binary_wire.contains(peer_id)
                && self.peer_binary_wire.len() >= MAX_PEER_BINARY_WIRE_ENTRIES
            {
                self.peer_binary_wire.clear();
            }
            self.peer_binary_wire.insert(peer_id.to_string());
        } else {
            self.peer_binary_wire.remove(peer_id);
        }
    }

    /// Forwards a peer's advertised Nostr public key to the Nostr transport, so
    /// gift wraps addressed to them are sealed to a key only their install
    /// holds instead of to the publicly computable bootstrap key.
    ///
    /// `None` means "this peer advertises no Nostr key", which **forgets** any
    /// key already held rather than leaving it in place. That is what keeps the
    /// in-memory copy consistent with the durable record, whose downgrade
    /// semantics delete on an empty advertisement: retaining it here instead
    /// would keep sealing to a key the peer has stopped claiming, and survive
    /// only until the next restart — a divergence with no symptom.
    ///
    /// A no-op when the Nostr transport is not installed (the common case —
    /// `nostr_enabled` defaults off). The bound on the underlying map lives in
    /// the transport.
    pub fn mark_peer_nostr_pubkey(&mut self, peer_id: &str, pubkey_hex: Option<&str>) {
        let Some(transport) = self.transports.get(&TransportType::Nostr) else {
            return;
        };
        if let Some(nostr) = transport
            .as_any()
            .downcast_ref::<offline_protocol_transport::nostr::NostrTransport>()
        {
            match pubkey_hex {
                Some(pubkey_hex) => nostr.set_peer_nostr_pubkey(peer_id, pubkey_hex),
                None => nostr.forget_peer_nostr_pubkey(peer_id),
            }
        }
    }

    /// Forgets a peer's Nostr public key, so frames to them revert to the
    /// publicly computable bootstrap key.
    ///
    /// Used by the unblock clean slate, which declares everything learned about
    /// a peer stale. Reverting is the right fallback rather than refusing to
    /// send: the bootstrap key is derived from the peer's user id, so it works
    /// whatever install they are on now.
    pub fn clear_peer_nostr_pubkey(&mut self, peer_id: &str) {
        if let Some(nostr) = self.transports.get(&TransportType::Nostr).and_then(|t| {
            t.as_any()
                .downcast_ref::<offline_protocol_transport::nostr::NostrTransport>()
        }) {
            nostr.forget_peer_nostr_pubkey(peer_id);
        }
    }

    /// This install's Nostr public key, for advertising in outgoing key
    /// packages. `None` when the Nostr transport is not installed.
    pub fn nostr_public_key(&self) -> Option<String> {
        self.transports
            .get(&TransportType::Nostr)?
            .as_any()
            .downcast_ref::<offline_protocol_transport::nostr::NostrTransport>()
            .map(|nostr| nostr.public_key_hex())
    }

    /// Applies the sealed-envelope kill switch to the Nostr transport.
    pub fn set_nostr_sealing_enabled(&mut self, enabled: bool) {
        if let Some(nostr) = self.transports.get(&TransportType::Nostr).and_then(|t| {
            t.as_any()
                .downcast_ref::<offline_protocol_transport::nostr::NostrTransport>()
        }) {
            nostr.set_sealing_enabled(enabled);
        }
    }

    /// Applies the cold-contact kill switch to the Nostr transport.
    pub fn set_nostr_cold_contact_enabled(&mut self, enabled: bool) {
        if let Some(nostr) = self.nostr_transport() {
            nostr.set_cold_contact_enabled(enabled);
        }
    }

    /// Queues a key-package record for publication at `slot_id`.
    ///
    /// A no-op when the Nostr transport is not installed, which is the common
    /// case: publication is driven from the engine's tick, and that tick runs
    /// whether or not Nostr is enabled.
    ///
    /// Serializes through the transport's own chokepoint, which yields JSON for
    /// an unstamped message — and a published record must never be anything
    /// else. The binary codec is negotiated per peer from that peer's key
    /// package, and the reader of this record is by definition a stranger whose
    /// key package we have not seen. Emitting binary here would make the record
    /// unreadable by exactly the audience it exists for.
    pub fn publish_nostr_key_package(&mut self, slot_id: &str, message: &Message) -> Result<()> {
        let Some(nostr) = self.nostr_transport() else {
            return Ok(());
        };
        let payload = nostr.serialize_message(message)?;
        nostr.publish_key_package(slot_id, payload);
        Ok(())
    }

    /// Drains the slot ids whose publication never reached a relay.
    ///
    /// Empty when Nostr is not installed, which keeps the caller's tick free of
    /// a transport check it would otherwise need.
    pub fn take_failed_nostr_publications(&self) -> Vec<String> {
        self.nostr_transport()
            .map(|nostr| nostr.take_failed_publications())
            .unwrap_or_default()
    }

    /// Whether the Nostr transport is installed and publishing records.
    pub fn nostr_cold_contact_active(&self) -> bool {
        self.nostr_transport()
            .map(|nostr| nostr.cold_contact_enabled())
            .unwrap_or(false)
    }

    /// Enables or disables username discovery on the Nostr transport.
    pub fn set_nostr_username_discovery_enabled(&mut self, enabled: bool) {
        if let Some(nostr) = self.nostr_transport() {
            nostr.set_discovery_enabled(enabled);
        }
    }

    /// Whether the Nostr transport is installed and username discovery is on.
    ///
    /// Already accounts for the cold-contact coupling: the transport reports
    /// false when either switch is off.
    pub fn nostr_discovery_active(&self) -> bool {
        self.nostr_transport()
            .map(|nostr| nostr.discovery_enabled())
            .unwrap_or(false)
    }

    /// Queues this install's username claim for publication.
    pub fn publish_nostr_discovery_record(&mut self, username: Username, payload: Vec<u8>) {
        if let Some(nostr) = self.nostr_transport() {
            nostr.publish_discovery_record(username, payload);
        }
    }

    /// Queues a retraction of this install's username claim.
    pub fn retract_nostr_discovery_record(&mut self, username: Username, payload: Vec<u8>) {
        if let Some(nostr) = self.nostr_transport() {
            nostr.retract_discovery_record(username, payload);
        }
    }

    /// Drains the discovery tags whose publication never reached a relay.
    pub fn take_failed_nostr_discovery_publications(&self) -> Vec<String> {
        self.nostr_transport()
            .map(|nostr| nostr.take_failed_discovery_publications())
            .unwrap_or_default()
    }

    /// Drains the discovery tags whose publication a relay acknowledged.
    pub fn take_confirmed_nostr_discovery_publications(&self) -> Vec<String> {
        self.nostr_transport()
            .map(|nostr| nostr.take_confirmed_discovery_publications())
            .unwrap_or_default()
    }

    /// Whether a Nostr transport is installed at all.
    ///
    /// Distinct from [`Self::nostr_discovery_active`]: a retraction is owed
    /// whether or not discovery is still switched on, so the engine has to be
    /// able to tell "the feature is off" from "there is nothing here that could
    /// carry a tombstone". Without the second, queueing a retraction silently
    /// does nothing while the engine records it as done.
    pub fn has_nostr_transport(&self) -> bool {
        self.nostr_transport().is_some()
    }

    /// Requests resolution of the devices claiming `username`.
    ///
    /// A missing Nostr transport reports `Disabled`, which is what it is from
    /// the caller's side: no query was sent and no event will follow.
    pub fn resolve_nostr_username(&self, username: Username) -> ResolveRequest {
        self.nostr_transport()
            .map(|nostr| nostr.resolve_username(username))
            .unwrap_or(ResolveRequest::Disabled)
    }

    /// Withdraws a queued username lookup that was never issued to a relay.
    ///
    /// Returns whether one was removed. Called when the engine gives up on a
    /// lookup, so the name does not sit in the queue making every later request
    /// for it return "already queued" without ever answering.
    pub fn cancel_nostr_username_resolution(&self, username: &Username) -> bool {
        self.nostr_transport()
            .map(|nostr| nostr.cancel_username_resolution(username))
            .unwrap_or(false)
    }

    fn nostr_transport(&self) -> Option<&offline_protocol_transport::nostr::NostrTransport> {
        self.transports.get(&TransportType::Nostr).and_then(|t| {
            t.as_any()
                .downcast_ref::<offline_protocol_transport::nostr::NostrTransport>()
        })
    }

    /// Sets the callback for DORS decision events (dors_score_updated, dors_transport_selected,
    pub fn set_dors_event_callback(&mut self, callback: Option<Arc<dyn Fn(Event) + Send + Sync>>) {
        self.dors_event_callback = callback;
    }

    /// Sets the callback that receives structured
    /// [`crate::telemetry::RoutingDecision`] records.
    ///
    /// Fires at the same sites as `dors_event_callback`, carrying a richer
    /// shape suitable for the unified telemetry sink.
    ///
    /// Crate-private: the callback is installed together with the telemetry
    /// sink (`OfflineProtocol::enable_telemetry` installs the pipe's) and is
    /// never wired directly, which keeps
    /// [`crate::telemetry::TelemetryConfig`] the single source of truth for
    /// its configuration.
    pub(crate) fn set_routing_decision_callback(
        &mut self,
        callback: Option<Arc<dyn Fn(RoutingDecision) + Send + Sync>>,
    ) {
        self.routing_decision_callback = callback;
    }

    /// Enables or disables the routing-diagnostic detail level.
    ///
    /// When `true`, [`RoutingDecision::scores`] is populated with the full
    /// seven-factor [`TransportScore`] for every ranked transport on every
    /// emission. When `false` (default), the vector is empty and the hot
    /// path avoids the per-emission allocation. Mirrors
    /// `TelemetryConfig::routing_diagnostic`.
    ///
    /// Crate-private: see [`Self::set_routing_decision_callback`].
    pub(crate) fn set_routing_diagnostic(&mut self, enabled: bool) {
        self.routing_diagnostic = enabled;
    }

    fn emit_dors_event(&self, event: Event) {
        if let Some(ref cb) = self.dors_event_callback {
            cb(event);
        }
    }

    /// Lazy-emit a [`RoutingDecision`]. The `build` closure only runs when a
    /// callback is actually installed, so the no-sink path pays only an
    /// `Option::is_some` check — not a `Utc::now()` call and struct build.
    fn emit_routing_decision<F: FnOnce() -> RoutingDecision>(&self, build: F) {
        if let Some(ref cb) = self.routing_decision_callback {
            cb(build());
        }
    }

    fn escalation_trigger_reason_to_code(r: EscalationTriggerReason) -> DorsEscalationReasonCode {
        match r {
            EscalationTriggerReason::RetryThreshold => DorsEscalationReasonCode::RetryThreshold,
            EscalationTriggerReason::PoorSignal => DorsEscalationReasonCode::PoorSignal,
            EscalationTriggerReason::Congestion => DorsEscalationReasonCode::Congestion,
            EscalationTriggerReason::LowTtl => DorsEscalationReasonCode::LowTtl,
            EscalationTriggerReason::LowSuccessRate => DorsEscalationReasonCode::LowSuccessRate,
        }
    }

    fn dors_reason_code_to_routing(code: DorsReasonCode) -> RoutingReasonCode {
        match code {
            DorsReasonCode::InitialSelection => RoutingReasonCode::InitialSelection,
            DorsReasonCode::PrimarySelected => RoutingReasonCode::PrimarySelected,
            DorsReasonCode::PrimarySuccess => RoutingReasonCode::PrimarySuccess,
            DorsReasonCode::FallbackSuccess => RoutingReasonCode::FallbackSuccess,
            DorsReasonCode::EscalationApplied => RoutingReasonCode::EscalationApplied,
            DorsReasonCode::CurrentUnavailable => RoutingReasonCode::CurrentUnavailable,
        }
    }

    fn escalation_reason_code_to_routing(code: DorsEscalationReasonCode) -> RoutingReasonCode {
        match code {
            DorsEscalationReasonCode::FallbackSuccess => RoutingReasonCode::FallbackSuccess,
            DorsEscalationReasonCode::RetryThreshold => RoutingReasonCode::RetryThreshold,
            DorsEscalationReasonCode::PoorSignal => RoutingReasonCode::PoorSignal,
            DorsEscalationReasonCode::Congestion => RoutingReasonCode::Congestion,
            DorsEscalationReasonCode::LowTtl => RoutingReasonCode::LowTtl,
            DorsEscalationReasonCode::LowSuccessRate => RoutingReasonCode::LowSuccessRate,
        }
    }

    /// Builds a [`RoutingDecision`]. The `detailed_scores` slice is cloned
    /// into `scores` when `routing_diagnostic` is enabled on the manager;
    /// otherwise the vector stays empty (allocation-free).
    ///
    /// Takes `now_ms` by argument rather than calling `Utc::now()` per
    /// build so every decision emitted from a single `send()` cycle shares
    /// one timestamp — easier to correlate downstream and saves three to
    /// four syscalls on the send hot path.
    #[allow(clippy::too_many_arguments)]
    fn build_routing_decision(
        &self,
        now_ms: i64,
        phase: RoutingPhase,
        from: Option<TransportType>,
        to: Option<TransportType>,
        winning_score: Option<f32>,
        reason_code: Option<RoutingReasonCode>,
        detailed_scores: Option<&[(TransportType, TransportScore)]>,
    ) -> RoutingDecision {
        // Sanitize the internal fallback-demotion sentinel out of every score
        // that leaves the engine. DORS keeps Internet's `total` negative to rank
        // it last; telemetry/UniFFI consumers must see the real 0–125 quality
        // score instead. This is the single chokepoint for all RoutingDecision
        // emitters, so `winning_score` is also sanitized here defensively
        // (idempotent for an already-positive value).
        let scores = match (self.routing_diagnostic, detailed_scores) {
            (true, Some(detailed)) => detailed
                .iter()
                .map(|(t, s)| {
                    let mut s = s.clone();
                    s.total = display_routing_score(s.total);
                    (*t, s)
                })
                .collect(),
            _ => Vec::new(),
        };
        RoutingDecision {
            timestamp_ms: now_ms,
            phase,
            from,
            to,
            winning_score: winning_score.map(display_routing_score),
            reason_code,
            scores,
        }
    }

    /// Emit dors_escalation_triggered at trigger boundary (typed reason), deduped by reason + time window.
    fn emit_escalation_trigger_if_deduped(
        &mut self,
        now_ms: i64,
        reason_code: DorsEscalationReasonCode,
        from: TransportType,
        to: TransportType,
        reason_detail: Option<String>,
        detailed_scores: Option<&[(TransportType, TransportScore)]>,
    ) {
        let now = std::time::Instant::now();
        let emit = match &self.last_escalation_trigger_emitted {
            Some((last_code, last_at)) => {
                *last_code != reason_code
                    || last_at.elapsed().as_secs() >= ESCALATION_TRIGGER_DEDUPE_SECS
            }
            None => true,
        };
        if emit {
            self.last_escalation_trigger_emitted = Some((reason_code, now));
            self.emit_dors_event(Event::dors_escalation_triggered(
                DorsEscalationPhase::Triggered,
                from.to_string(),
                to.to_string(),
                reason_code,
                reason_detail,
            ));
            self.emit_routing_decision(|| {
                self.build_routing_decision(
                    now_ms,
                    RoutingPhase::Escalated,
                    Some(from),
                    Some(to),
                    None,
                    Some(Self::escalation_reason_code_to_routing(reason_code)),
                    detailed_scores,
                )
            });
        }
    }

    /// Adds a transport to the manager.
    ///
    /// # Arguments
    ///
    /// * `transport_type` - Type of transport to add
    /// * `transport` - The transport implementation
    pub fn add_transport(&mut self, transport_type: TransportType, transport: Box<dyn Transport>) {
        self.transports.insert(transport_type, Arc::from(transport));
    }

    /// Sends a message through the best available transport, with fallback.
    ///
    /// DORS selects a primary transport (applying hysteresis, cooldown, and
    /// stability checks). If the primary's `send()` fails synchronously, the
    /// remaining transports are tried in descending score order before
    /// returning an error. Each synchronous failure is recorded via
    /// `record_retry_failure` so DORS can adjust future scoring.
    ///
    /// # Internet transport — fallback does NOT apply
    ///
    /// Internet `send()` enqueues the message and returns `Ok(())`
    /// immediately. The actual wire-level outcome is reported asynchronously
    /// via the `confirm_sent` / `report_send_failure` confirmation loop on
    /// `InternetTransport`. Because `send()` never fails synchronously, the
    /// fallback loop below will **not** trigger for Internet. If the
    /// WebSocket is silently broken (status still `Available` but sends
    /// fail), messages will be enqueued, detected as failures by the
    /// confirmation-loop timeout, and counted in DORS metrics — but they
    /// are **not** retried on an alternative transport by this method.
    /// Retry/re-routing for those messages is handled by the higher-level
    /// outbox retry mechanism.
    pub fn send(&mut self, message: &Message) -> Result<()> {
        // Stamp the negotiated wire codec for this recipient. The clone happens
        // only on the binary path (peers known to support it); JSON and unknown
        // peers reuse the borrowed message. The internet transport ignores the
        // stamp (its relay bridge is JSON-only), and mesh transports only ever
        // emit to a recipient that is a directly connected peer, so `recipient`
        // is the physical peer whose capability was recorded — keying by it
        // matches the same peer-string the MTU path already keys by.
        //
        // The codec travels on the message, and both `Transport::send` and this
        // method take `&Message` as public API, so carrying the stamp across
        // that boundary needs an owned message — hence this clone. Note it is a
        // full-message clone: it copies `binary_content` (up to the max message
        // size), not just the envelope, so for media-sized payloads sent to a
        // binary-capable peer it is a real payload memcpy on the hot path. It is
        // a deliberate tradeoff: threading the codec as a separate argument would
        // shave the clone but is a breaking change to the public `Transport`
        // trait every downstream implementor relies on. The transport's own
        // queue-clone is unaffected either way. If profiling ever flags this,
        // stamp via a lightweight wrapper rather than cloning the payload.
        let stamped = self.stamp_negotiated_wire_codec(message);
        let message = stamped.as_ref().unwrap_or(message);

        let available = self.get_available_transports();
        if available.is_empty() {
            return Err(Error::Transport(TransportError::TransportNotAvailable(
                "No available transport".to_string(),
            )));
        }

        // Single timestamp shared by every RoutingDecision emitted from this
        // send cycle — cheaper than per-emit `Utc::now()` and lets downstream
        // consumers group decisions by timestamp.
        let now_ms = Utc::now().timestamp_millis();

        let previous = self.current_transport;

        // DORS selects the primary transport (applies hysteresis/cooldown/stability).
        let primary = self
            .selector
            .select_transport(message, &available)
            .ok_or_else(|| {
                Error::Transport(TransportError::TransportNotAvailable(
                    "No suitable transport selected from available transports".to_string(),
                ))
            })?;

        // Emit DORS observability: scores and selection (before send attempt).
        // At the default (non-diagnostic) tier we project the detailed
        // breakdown down to `(transport, total)` — a single scoring pass
        // feeds both the legacy event and the routing record.
        let detailed_scores: Option<Vec<(TransportType, TransportScore)>> =
            if self.routing_diagnostic && self.routing_decision_callback.is_some() {
                Some(self.selector.score_and_rank_detailed(message, &available))
            } else {
                None
            };
        let detailed_slice = detailed_scores.as_deref();
        let scored: Vec<(TransportType, f32)> = match detailed_slice {
            Some(d) => d.iter().map(|(t, s)| (*t, s.total)).collect(),
            None => self.selector.score_and_rank(message, &available),
        };
        // Sanitize the fallback-demotion sentinel out of the emitted scores (the
        // `scored` vec itself keeps the raw totals — the fallback loop below
        // relies on Internet ranking last).
        let scores: Vec<(String, f32)> = scored
            .iter()
            .map(|(t, s)| (t.to_string(), display_routing_score(*s)))
            .collect();
        self.emit_dors_event(Event::dors_score_updated(scores));

        self.emit_routing_decision(|| {
            self.build_routing_decision(
                now_ms,
                RoutingPhase::ScoreUpdated,
                previous,
                None,
                None,
                None,
                detailed_slice,
            )
        });
        // Telemetry-only; sanitize the fallback-demotion sentinel so the emitted
        // selection score is the real 0–125 quality value, not the negative rank.
        let primary_score = scored
            .iter()
            .find(|(t, _)| *t == primary)
            .map(|(_, s)| display_routing_score(*s))
            .unwrap_or(0.0);
        let selection_reason = if previous.is_none() {
            DorsReasonCode::InitialSelection
        } else {
            DorsReasonCode::PrimarySelected
        };
        self.emit_dors_event(Event::dors_transport_selected(
            previous.map(|t| t.to_string()),
            primary.to_string(),
            selection_reason,
            primary_score,
        ));
        self.emit_routing_decision(|| {
            self.build_routing_decision(
                now_ms,
                RoutingPhase::Selected,
                previous,
                Some(primary),
                Some(primary_score),
                Some(Self::dors_reason_code_to_routing(selection_reason)),
                detailed_slice,
            )
        });

        // Try the primary transport first.
        let mut primary_refusal = None;
        let primary_result = {
            let transport = self
                .transports
                .get(&primary)
                .ok_or_else(|| Error::Other(format!("Transport {:?} not found", primary)))?;
            transport.send(message)
        };

        match primary_result {
            Ok(()) => {
                self.current_transport = Some(primary);
                if previous != Some(primary) {
                    let reason_code = if previous.is_some_and(|p| !available.contains_key(&p)) {
                        DorsReasonCode::CurrentUnavailable
                    } else {
                        DorsReasonCode::PrimarySuccess
                    };
                    self.emit_dors_event(Event::dors_transport_switched(
                        previous.map(|t| t.to_string()),
                        primary.to_string(),
                        reason_code,
                        None,
                    ));
                    self.emit_routing_decision(|| {
                        self.build_routing_decision(
                            now_ms,
                            RoutingPhase::Switched,
                            previous,
                            Some(primary),
                            Some(primary_score),
                            Some(Self::dors_reason_code_to_routing(reason_code)),
                            detailed_slice,
                        )
                    });
                }
                return Ok(());
            }
            Err(e) => {
                let is_peer_routing = matches!(e, TransportError::PeerNotReachable(_));
                if is_peer_routing {
                    debug!(
                        transport = ?primary,
                        error = %e,
                        "Peer not reachable via primary transport, trying fallback"
                    );
                    // Kept as the terminal error until a fallback replaces it.
                    // Without it, a primary that refused the recipient with
                    // nothing left to try ended as `SendFailed("All transports
                    // failed")`, which every classifier reads as a carrier that
                    // broke mid-send: the deferral said `transport_send_failed`
                    // for a recipient no carrier could address.
                    primary_refusal = Some(e);
                } else {
                    warn!(
                        transport = ?primary,
                        error = %e,
                        "Primary transport send failed, trying fallback"
                    );
                    self.selector.record_retry_failure(primary);
                }
            }
        }

        // Primary failed — try remaining transports in score order.
        // Reuse the scored ranking computed above; record_retry_failure only
        // mutates retry_counts which score_and_rank does not read.
        let mut last_error = primary_refusal;
        let mut attempted: Vec<TransportType> = vec![primary];

        for (transport_type, _score) in &scored {
            if *transport_type == primary {
                continue;
            }

            // Emit escalation trigger at DORS boundary (typed reason), deduped.
            if primary == TransportType::BLE && *transport_type == TransportType::WiFiDirect {
                if let Some(trigger_reason) = self.selector.escalation_trigger_reason() {
                    let reason_code = Self::escalation_trigger_reason_to_code(trigger_reason);
                    self.emit_escalation_trigger_if_deduped(
                        now_ms,
                        reason_code,
                        primary,
                        *transport_type,
                        None,
                        detailed_slice,
                    );
                }
            }

            let transport = match self.transports.get(transport_type) {
                Some(t) => t,
                None => continue,
            };

            attempted.push(*transport_type);

            match transport.send(message) {
                Ok(()) => {
                    self.current_transport = Some(*transport_type);
                    self.selector.set_current_transport(*transport_type);
                    // Emit switch only when active transport actually changed (fallback success).
                    if previous != Some(*transport_type) {
                        let reason_code = if primary == TransportType::BLE
                            && *transport_type == TransportType::WiFiDirect
                        {
                            DorsReasonCode::EscalationApplied
                        } else {
                            DorsReasonCode::FallbackSuccess
                        };
                        self.emit_dors_event(Event::dors_transport_switched(
                            previous.map(|t| t.to_string()),
                            transport_type.to_string(),
                            reason_code,
                            Some("primary send failed, fallback succeeded".to_string()),
                        ));
                        let fallback = *transport_type;
                        self.emit_routing_decision(|| {
                            self.build_routing_decision(
                                now_ms,
                                RoutingPhase::Switched,
                                previous,
                                Some(fallback),
                                None,
                                Some(Self::dors_reason_code_to_routing(reason_code)),
                                detailed_slice,
                            )
                        });
                    }
                    // Escalation applied only when BLE→WiFi fallback actually succeeded.
                    if primary == TransportType::BLE && *transport_type == TransportType::WiFiDirect
                    {
                        self.emit_dors_event(Event::dors_escalation_triggered(
                            DorsEscalationPhase::Applied,
                            primary.to_string(),
                            transport_type.to_string(),
                            DorsEscalationReasonCode::FallbackSuccess,
                            Some("primary BLE send failed, fallback to WiFi succeeded".to_string()),
                        ));
                        let escalated = *transport_type;
                        self.emit_routing_decision(|| {
                            self.build_routing_decision(
                                now_ms,
                                RoutingPhase::Escalated,
                                Some(primary),
                                Some(escalated),
                                None,
                                Some(Self::escalation_reason_code_to_routing(
                                    DorsEscalationReasonCode::FallbackSuccess,
                                )),
                                detailed_slice,
                            )
                        });
                    }
                    return Ok(());
                }
                Err(e) => {
                    if !matches!(e, TransportError::PeerNotReachable(_)) {
                        warn!(
                            transport = ?transport_type,
                            error = %e,
                            "Fallback transport send failed, trying next"
                        );
                        self.selector.record_retry_failure(*transport_type);
                    }
                    last_error = Some(e);
                }
            }
        }

        let terminal_error = last_error.unwrap_or_else(|| {
            TransportError::SendFailed(format!("All transports failed (tried {:?})", attempted))
        });
        Err(Error::Transport(terminal_error))
    }

    /// Sends a message via a specific transport, bypassing DORS selection.
    ///
    /// # Arguments
    ///
    /// * `message` - The message to send
    /// * `transport_type` - The transport to use
    ///
    /// # Returns
    ///
    /// Returns Ok(()) if sent successfully, Err otherwise.
    pub fn send_via_transport(
        &mut self,
        message: &Message,
        transport_type: TransportType,
    ) -> Result<()> {
        let transport = self
            .transports
            .get(&transport_type)
            .ok_or_else(|| Error::Other(format!("Transport {:?} not found", transport_type)))?;

        if transport.status() != TransportStatus::Available {
            return Err(Error::Other(format!(
                "Transport {:?} is not available",
                transport_type
            )));
        }

        // Preserved as `Error::Transport`, not flattened into `Error::Other`.
        // Flattening discarded the variant every classifier keys on —
        // `SessionStateError::classify` and `Protocol::is_no_carrier_error`
        // both answer on `Error::Transport(_)`, so on this path they saw
        // `Unknown`/`false` and the rendered string was the only surviving
        // signal. That is exactly the string that must stop being read.
        // Stamped exactly as `send` does. This path carries ordinary DMs
        // whenever a recipient is a live mesh neighbour, so skipping the stamp
        // would downgrade binary-capable peers to JSON on the mesh while the
        // DORS path kept them on binary.
        let stamped = self.stamp_negotiated_wire_codec(message);
        let message = stamped.as_ref().unwrap_or(message);
        transport.send(message).map_err(Error::Transport)?;

        // Only update current transport after successful send
        self.current_transport = Some(transport_type);

        Ok(())
    }

    /// Lists every peer reachable over a direct mesh link, with the transport
    /// that link runs on.
    ///
    /// Only the peer-to-peer carriers are consulted: a mesh neighbor is one we
    /// can hand a frame to over our own radio, which is what makes it a
    /// candidate next hop. Infrastructure carriers (Internet, Nostr, Reticulum)
    /// reach peers through something else, so they report no links and appear
    /// here as nothing — a message routed through them is not taking a mesh
    /// hop, it is leaving the mesh.
    ///
    /// A peer visible on more than one carrier is reported once, on the first
    /// carrier that claims it; [`Self::send_to_neighbor`] resolves the same way,
    /// so the pair stay consistent.
    pub fn mesh_neighbors(&self) -> Vec<MeshNeighbor> {
        let mut neighbors: Vec<MeshNeighbor> = Vec::new();

        for transport_type in MESH_TRANSPORTS {
            let Some(transport) = self.transports.get(transport_type) else {
                continue;
            };
            for link in transport.connected_peers() {
                if neighbors.iter().any(|n| n.peer_id == link.peer_id) {
                    continue;
                }
                neighbors.push(MeshNeighbor {
                    peer_id: link.peer_id,
                    rssi: link.rssi,
                    transport: *transport_type,
                });
            }
        }

        neighbors
    }

    /// Returns an owned copy stamped with the negotiated binary codec, when
    /// this recipient has one and the message is not already stamped.
    ///
    /// `None` means the borrowed message is already correct, which is the
    /// common case (JSON peers, unknown peers, already-stamped messages) and
    /// costs no allocation.
    ///
    /// Shared by every send path rather than living inline in [`Self::send`].
    /// The codec is negotiated per recipient, so a path that skips the stamp
    /// silently downgrades a binary-capable peer to JSON — a regression with no
    /// error and no event, visible only as larger frames on the wire.
    fn stamp_negotiated_wire_codec(&self, message: &Message) -> Option<Message> {
        if message.wire_codec() == WireCodec::BinaryV1
            || !self.peer_binary_wire.contains(message.recipient.as_str())
        {
            return None;
        }
        let mut stamped = message.clone();
        stamped.set_wire_codec(WireCodec::BinaryV1);
        Some(stamped)
    }

    /// The mesh carrier holding a live link straight to `peer_id`, if any.
    ///
    /// Ordered by mesh-carrier preference, so a peer reachable over both answers
    /// with the higher-bandwidth carrier. This is ground truth rather than a
    /// claim: it is read from the transport's live link view at the moment it
    /// is asked, never remembered.
    pub fn mesh_transport_for(&self, peer_id: &str) -> Option<TransportType> {
        MESH_TRANSPORTS.iter().copied().find(|transport_type| {
            self.transports
                .get(transport_type)
                .filter(|transport| transport.status() == TransportStatus::Available)
                .is_some_and(|transport| {
                    transport
                        .connected_peers()
                        .iter()
                        .any(|link| link.peer_id == peer_id)
                })
        })
    }

    /// Whether the mesh carrier `transport_type` holds a live link straight
    /// to `peer_id`.
    ///
    /// [`Self::mesh_transport_for`] answers with the preferred carrier only,
    /// so a peer linked over both never names the second; this asks about one
    /// carrier. `false` for a carrier that is not registered, not available,
    /// or not a mesh carrier.
    pub fn holds_mesh_link(&self, transport_type: TransportType, peer_id: &str) -> bool {
        MESH_TRANSPORTS.contains(&transport_type)
            && self
                .transports
                .get(&transport_type)
                .filter(|transport| transport.status() == TransportStatus::Available)
                .is_some_and(|transport| {
                    transport
                        .connected_peers()
                        .iter()
                        .any(|link| link.peer_id == peer_id)
                })
    }

    /// Every available carrier that does its own routing.
    ///
    /// The enumerated form of [`Self::has_infrastructure_carrier`], for callers
    /// that hold a per-recipient fact and must ask each carrier separately
    /// rather than accepting "one of them is up" for every recipient.
    pub fn available_infrastructure_carriers(&self) -> Vec<TransportType> {
        self.transports
            .iter()
            .filter(|(transport_type, transport)| {
                !MESH_TRANSPORTS.contains(transport_type)
                    && transport.status() == TransportStatus::Available
            })
            .map(|(transport_type, _)| *transport_type)
            .collect()
    }

    /// Whether a frame for `peer_id` can plausibly arrive without being carried
    /// by other devices.
    ///
    /// This exists because **a send returning `Ok` is not evidence of
    /// reachability**. BLE and Wi-Fi Direct refuse a recipient they hold no
    /// link to, but the internet transport and Reticulum enqueue for any
    /// recipient and report success, so a frame handed to them for someone
    /// neither can reach is reported as sent and never arrives. Anything that
    /// decides "the mesh has to carry this" from a send failure therefore never
    /// fires on a device where one of those carriers is up. Asking this instead
    /// keeps that decision on facts we can check.
    ///
    /// Two ways the answer is yes:
    ///
    /// - An **infrastructure** carrier is available. Internet, Nostr and
    ///   Reticulum do their own routing and reach peers we hold no radio link
    ///   to, so a direct send is a real delivery attempt. (Reticulum counts
    ///   here because it is a routing network in its own right — its `Ok` means
    ///   "accepted for routing", which is as much as any relay-style carrier
    ///   promises.)
    /// - The peer is one of our own **mesh neighbors**, so a mesh carrier can
    ///   hand the frame straight over.
    ///
    /// Checked in that order so the common online case costs a status scan and
    /// never enumerates links.
    ///
    /// **This answer is deliberately peer-blind on its first arm**, and that is
    /// a real limit rather than an oversight: an infrastructure carrier being up
    /// says nothing about whether *this* recipient is on it. Callers that hold a
    /// peer-specific fact must not settle for this answer — the relay's own
    /// `recipient_unreachable` verdict is exactly such a fact, and the park it
    /// drives offers the frame to the mesh regardless of what this returns (see
    /// `park_unreachable_dm`).
    pub fn can_reach_without_carrying(&self, peer_id: &str) -> bool {
        self.has_infrastructure_carrier() || self.mesh_can_address(peer_id)
    }

    /// Whether any carrier that does its own routing is available.
    ///
    /// Peer-blind by nature: these carriers reach peers we hold no radio link
    /// to, so there is no per-peer link to consult. Note the status is
    /// bridge-reported and means "the platform says this carrier is up", not
    /// "the relay connection is authenticated" — a distinction that matters
    /// wherever this stands in for reachability.
    pub fn has_infrastructure_carrier(&self) -> bool {
        self.transports.iter().any(|(transport_type, transport)| {
            !MESH_TRANSPORTS.contains(transport_type)
                && transport.status() == TransportStatus::Available
        })
    }

    /// Ids of messages some carrier has accepted but not yet started writing
    /// (see [`Transport::queued_message_ids`]).
    pub fn queued_message_ids(&self) -> std::collections::HashSet<String> {
        self.transports
            .values()
            .flat_map(|transport| transport.queued_message_ids())
            .collect()
    }

    /// Whether a mesh carrier holds a live link straight to `peer_id`.
    ///
    /// The peer-specific half of [`Self::can_reach_without_carrying`], split out
    /// because callers that already know infrastructure is up still need to ask
    /// whether the mesh can hand a frame over without anyone carrying it.
    pub fn mesh_can_address(&self, peer_id: &str) -> bool {
        self.mesh_transport_for(peer_id).is_some()
    }

    /// Whether `transport_type` is a short-range carrier that can only address
    /// peers it holds a live link to — as opposed to an infrastructure carrier
    /// that does its own routing.
    ///
    /// A frame that arrived over one of these travelled either from a neighbor
    /// or through them, which is what makes it evidence about the route back.
    pub fn is_mesh_transport(transport_type: TransportType) -> bool {
        MESH_TRANSPORTS.contains(&transport_type)
    }

    /// Whether `transport_type` can actually put a frame in front of `peer_id`.
    ///
    /// The narrower question [`Self::can_reach_without_carrying`] answers across
    /// every carrier, asked of one. It exists for the same reason: a transport
    /// returning `Ok` is not evidence the frame can arrive. A **mesh** carrier
    /// can only address a peer it holds a live link to: BLE and Wi-Fi Direct
    /// both refuse a send to anyone else, and asking first lets a caller choose
    /// another route before it spends a send, rather than after the refusal.
    /// **Infrastructure** carriers do their own routing, so for them the answer
    /// is yes whenever they are available.
    ///
    /// Callers that would otherwise trust a preferred transport — replying on
    /// the link a message arrived over, above all — must ask this first, or the
    /// reply dies on exactly the multi-hop path that made forwarding necessary.
    pub fn can_address_via(&self, transport_type: TransportType, peer_id: &str) -> bool {
        let Some(transport) = self.transports.get(&transport_type) else {
            return false;
        };
        if transport.status() != TransportStatus::Available {
            return false;
        }
        if !MESH_TRANSPORTS.contains(&transport_type) {
            return true;
        }
        transport
            .connected_peers()
            .iter()
            .any(|link| link.peer_id == peer_id)
    }

    /// Hands `message` to a specific mesh neighbor, whatever the message's own
    /// recipient is.
    ///
    /// This is the forwarding primitive: the frame crosses one link, to the
    /// peer named by `peer_id`. Because that peer is the physical target, the
    /// binary wire codec is negotiated against **its** capabilities rather than
    /// the recipient's — stamping a forwarded frame binary because the distant
    /// recipient supports it could put bytes a neighbor cannot decode onto that
    /// neighbor's link.
    ///
    /// Returns the transport the frame went out on.
    pub fn send_to_neighbor(&self, peer_id: &str, message: &Message) -> Result<TransportType> {
        let stamped;
        let message = if message.wire_codec() != WireCodec::BinaryV1
            && self.peer_binary_wire.contains(peer_id)
        {
            let mut m = message.clone();
            m.set_wire_codec(WireCodec::BinaryV1);
            stamped = m;
            &stamped
        } else if message.wire_codec() == WireCodec::BinaryV1
            && !self.peer_binary_wire.contains(peer_id)
        {
            // The frame arrived stamped binary (or was stamped for a different
            // peer) but this hop cannot decode it. Fall back to the JSON floor
            // rather than writing bytes the neighbor will drop.
            let mut m = message.clone();
            m.set_wire_codec(WireCodec::Json);
            stamped = m;
            &stamped
        } else {
            message
        };

        let mut last_error: Option<String> = None;

        for transport_type in MESH_TRANSPORTS {
            let Some(transport) = self.transports.get(transport_type) else {
                continue;
            };
            if transport.status() != TransportStatus::Available {
                continue;
            }
            match transport.send_to_peer(peer_id, message) {
                Ok(()) => return Ok(*transport_type),
                Err(e) => last_error = Some(e.to_string()),
            }
        }

        Err(Error::Transport(TransportError::PeerNotReachable(format!(
            "no mesh transport holds a link to {}{}",
            peer_id,
            last_error.map(|e| format!(" ({})", e)).unwrap_or_default()
        ))))
    }

    /// Attempts to receive a message from any transport.
    ///
    /// # Returns
    ///
    /// Returns Ok(Some((TransportType, Message))) if a message was received, Ok(None) if no message available.
    pub fn receive(&self) -> Result<Option<(TransportType, Message)>> {
        // Check all transports for messages
        for (transport_type, transport) in &self.transports {
            match transport.receive() {
                Ok(Some(message)) => {
                    debug!(
                        transport = ?transport_type,
                        sender = %message.sender,
                        recipient = %message.recipient,
                        "TransportManager found message"
                    );
                    return Ok(Some((*transport_type, message)));
                }
                Ok(None) => {
                    // No message from this transport, continue checking others
                }
                Err(e) => {
                    warn!(
                        transport = ?transport_type,
                        error = %e,
                        "Transport receive error"
                    );
                }
            }
        }
        Ok(None)
    }

    /// Gets metrics for all available transports.
    ///
    /// # Returns
    ///
    /// Returns a map of transport type to metrics.
    pub fn get_available_transports(&self) -> HashMap<TransportType, TransportMetrics> {
        let mut available = HashMap::new();

        for (transport_type, transport) in &self.transports {
            let maybe_metrics = {
                let status = transport.status();
                if status == TransportStatus::Available {
                    Some(transport.metrics())
                } else {
                    debug!(
                        transport = ?transport_type,
                        status = ?status,
                        "Transport excluded from available set"
                    );
                    None
                }
            };

            if let Some(mut metrics) = maybe_metrics {
                if let Some(stats) = self.observations.get(transport_type) {
                    stats.apply_to_metrics(*transport_type, &mut metrics);
                }
                self.apply_device_battery(&mut metrics);
                available.insert(*transport_type, metrics);
            }
        }

        available
    }

    /// Gets the current active transport type.
    pub fn current_transport(&self) -> Option<TransportType> {
        self.current_transport
    }

    /// Returns the current status of every registered transport, including
    /// those not currently [`TransportStatus::Available`].
    ///
    /// Used by the telemetry aggregator to detect state transitions; unlike
    /// [`TransportManager::get_available_transports`], this only reads each
    /// transport's status with no metrics work.
    pub fn get_all_transport_statuses(&self) -> HashMap<TransportType, TransportStatus> {
        let mut out = HashMap::with_capacity(self.transports.len());
        for (transport_type, transport) in &self.transports {
            out.insert(*transport_type, transport.status());
        }
        out
    }

    /// Returns the per-transport status map AND the available-only metrics
    /// map in a single pass.
    ///
    /// Equivalent to calling [`Self::get_all_transport_statuses`] and
    /// [`Self::get_available_transports`] back-to-back, but reads each
    /// transport's status once instead of twice. Used by the per-tick
    /// telemetry aggregator.
    pub fn snapshot_status_and_available(
        &self,
    ) -> (
        HashMap<TransportType, TransportStatus>,
        HashMap<TransportType, TransportMetrics>,
    ) {
        let mut statuses = HashMap::with_capacity(self.transports.len());
        let mut available = HashMap::new();

        for (transport_type, transport) in &self.transports {
            let status = transport.status();
            let metrics = if status == TransportStatus::Available {
                Some(transport.metrics())
            } else {
                None
            };
            statuses.insert(*transport_type, status);
            if let Some(mut metrics) = metrics {
                if let Some(stats) = self.observations.get(transport_type) {
                    stats.apply_to_metrics(*transport_type, &mut metrics);
                }
                self.apply_device_battery(&mut metrics);
                available.insert(*transport_type, metrics);
            }
        }

        (statuses, available)
    }

    /// Starts all transports.
    pub fn start(&mut self) -> Result<()> {
        for (transport_type, transport) in &self.transports {
            transport.start().map_err(|e| {
                Error::Other(format!(
                    "Failed to start transport {:?}: {}",
                    transport_type, e
                ))
            })?;
        }
        Ok(())
    }

    /// Stops all transports.
    pub fn stop(&mut self) -> Result<()> {
        for (transport_type, transport) in &self.transports {
            transport.stop().map_err(|e| {
                Error::Other(format!(
                    "Failed to stop transport {:?}: {}",
                    transport_type, e
                ))
            })?;
        }
        Ok(())
    }

    /// Gets a shared handle to a specific transport.
    pub fn get_transport(&self, transport_type: TransportType) -> Option<Arc<dyn Transport>> {
        self.transports.get(&transport_type).cloned()
    }

    /// Removes a transport from the manager.
    ///
    /// # Arguments
    ///
    /// * `transport_type` - Type of transport to remove
    pub fn remove_transport(&mut self, transport_type: TransportType) {
        self.transports.remove(&transport_type);

        // Clear current transport if it was the one removed
        if self.current_transport == Some(transport_type) {
            self.current_transport = None;
        }

        self.observations.remove(&transport_type);
    }

    /// Gets a list of all active transport types.
    ///
    /// # Returns
    ///
    /// Returns a vector of transport types that are currently added to the manager.
    pub fn get_active_transports(&self) -> Vec<TransportType> {
        self.transports.keys().copied().collect()
    }

    /// Checks if DORS suggests escalating from BLE to WiFi Direct.
    pub fn should_escalate_to_wifi(&self) -> bool {
        self.selector.should_escalate_to_wifi()
    }

    /// Records a retry failure for the given transport.
    pub fn record_retry_failure(&mut self, transport_type: TransportType) {
        self.selector.record_retry_failure(transport_type);
    }

    /// Resets retry count for the given transport after successful delivery.
    pub fn reset_retry_count(&mut self, transport_type: TransportType) {
        self.selector.reset_retry_count(transport_type);
    }

    /// Records a successful end-to-end delivery for the given transport.
    pub fn record_delivery_success(
        &mut self,
        transport_type: TransportType,
        latency_ms: u32,
        hop_count: u8,
    ) {
        let stats = self.observations.entry(transport_type).or_default();
        stats.record_success(latency_ms, hop_count);
    }

    /// Records a delivery failure (after exhausting retries) for the given transport.
    pub fn record_delivery_failure(&mut self, transport_type: TransportType) {
        let stats = self.observations.entry(transport_type).or_default();
        stats.record_failure();
    }

    #[cfg(test)]
    pub(crate) fn observed_failure_count(&self, transport_type: TransportType) -> u32 {
        self.observations
            .get(&transport_type)
            .map_or(0, |stats| stats.failure_count)
    }

    /// Updates the DORS selector configuration at runtime, preserving
    /// accumulated state (transport history, retry counts, signal tracking).
    pub fn update_selector_config(&mut self, config: DorsConfig) {
        self.selector.update_config(config);
    }

    /// Returns the DORS configuration transport scoring runs against.
    pub fn selector_config(&self) -> &DorsConfig {
        self.selector.config()
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use offline_protocol_core::{AppId, UserId};
    use offline_protocol_router::DorsConfig;
    use offline_protocol_transport::{mock::MockTransport, TransportType};
    use std::sync::Mutex;

    fn create_test_message() -> Message {
        Message::new(
            UserId::new("alice").unwrap(),
            UserId::new("bob").unwrap(),
            AppId::new("test").unwrap(),
            "Test message",
        )
    }

    #[test]
    fn test_transport_manager_creation() {
        let selector = TransportSelector::with_config(DorsConfig::default());
        let manager = TransportManager::new(selector);
        assert_eq!(manager.transports.len(), 0);
        assert!(manager.current_transport().is_none());
    }

    #[test]
    fn test_add_transport() {
        let selector = TransportSelector::with_config(DorsConfig::default());
        let mut manager = TransportManager::new(selector);

        let transport = Box::new(MockTransport::new(TransportType::BLE));
        manager.add_transport(TransportType::BLE, transport);

        assert_eq!(manager.transports.len(), 1);
    }

    #[test]
    fn test_start_stop_through_shared_transport_handle() {
        let selector = TransportSelector::with_config(DorsConfig::default());
        let mut manager = TransportManager::new(selector);
        manager.add_transport(
            TransportType::BLE,
            Box::new(MockTransport::new(TransportType::BLE)),
        );

        // The handle shares the same transport the manager drives: start()
        // and stop() go through &self, no per-transport mutex involved.
        let handle = manager
            .get_transport(TransportType::BLE)
            .expect("transport registered");
        assert_eq!(handle.status(), TransportStatus::Unavailable);

        manager.start().unwrap();
        assert_eq!(handle.status(), TransportStatus::Available);

        manager.stop().unwrap();
        assert_eq!(handle.status(), TransportStatus::Disconnected);
    }

    #[test]
    fn test_send_message() {
        let selector = TransportSelector::with_config(DorsConfig::default());
        let mut manager = TransportManager::new(selector);

        let transport = MockTransport::new(TransportType::BLE);
        transport.start().unwrap();
        manager.add_transport(TransportType::BLE, Box::new(transport));

        let message = create_test_message();
        let result = manager.send(&message);
        assert!(result.is_ok());
    }

    #[test]
    fn send_stamps_binary_only_for_capable_peer() {
        let selector = TransportSelector::with_config(DorsConfig::default());
        let mut manager = TransportManager::new(selector);
        let transport = MockTransport::new(TransportType::BLE);
        transport.start().unwrap();
        manager.add_transport(TransportType::BLE, Box::new(transport));

        // Unknown peer -> JSON (the default).
        manager.send(&create_test_message()).unwrap();
        // Peer recorded as binary-capable -> the send is stamped binary.
        manager.mark_peer_binary_wire("bob", true);
        manager.send(&create_test_message()).unwrap();

        let mock = manager
            .transports
            .get(&TransportType::BLE)
            .unwrap()
            .as_any()
            .downcast_ref::<MockTransport>()
            .unwrap();
        let sent = mock.sent_messages();
        assert_eq!(sent.len(), 2);
        assert_eq!(sent[0].wire_codec(), WireCodec::Json);
        assert_eq!(sent[1].wire_codec(), WireCodec::BinaryV1);
    }

    #[test]
    fn send_reverts_to_json_after_peer_downgrades() {
        let selector = TransportSelector::with_config(DorsConfig::default());
        let mut manager = TransportManager::new(selector);
        let transport = MockTransport::new(TransportType::BLE);
        transport.start().unwrap();
        manager.add_transport(TransportType::BLE, Box::new(transport));

        // Learn the peer is binary-capable -> the next send is stamped binary.
        manager.mark_peer_binary_wire("bob", true);
        assert!(manager.peer_supports_binary_wire("bob"));
        manager.send(&create_test_message()).unwrap();

        // A later key package that no longer advertises support downgrades the
        // peer (e.g. after a session reset, or the codec being turned off on
        // their side); subsequent sends must fall back to JSON, not keep
        // emitting binary a legacy build could not decode.
        manager.mark_peer_binary_wire("bob", false);
        assert!(!manager.peer_supports_binary_wire("bob"));
        manager.send(&create_test_message()).unwrap();

        let mock = manager
            .transports
            .get(&TransportType::BLE)
            .unwrap()
            .as_any()
            .downcast_ref::<MockTransport>()
            .unwrap();
        let sent = mock.sent_messages();
        assert_eq!(sent.len(), 2);
        assert_eq!(sent[0].wire_codec(), WireCodec::BinaryV1);
        assert_eq!(sent[1].wire_codec(), WireCodec::Json);
    }

    #[test]
    fn test_dors_events_emitted_when_callback_set() {
        let selector = TransportSelector::with_config(DorsConfig::default());
        let mut manager = TransportManager::new(selector);

        let transport = MockTransport::new(TransportType::BLE);
        transport.start().unwrap();
        manager.add_transport(TransportType::BLE, Box::new(transport));

        let events: std::sync::Arc<Mutex<Vec<Event>>> = std::sync::Arc::new(Mutex::new(Vec::new()));
        let events_clone = events.clone();
        manager.set_dors_event_callback(Some(Arc::new(move |e| {
            events_clone.lock().unwrap().push(e);
        })));

        let message = create_test_message();
        let _ = manager.send(&message);

        let captured = events.lock().unwrap();
        assert!(
            captured
                .iter()
                .any(|e| matches!(e, crate::events::Event::DorsScoreUpdated { .. })),
            "expected dors_score_updated"
        );
        assert!(
            captured
                .iter()
                .any(|e| { matches!(e, crate::events::Event::DorsTransportSelected { .. }) }),
            "expected dors_transport_selected"
        );
    }

    #[test]
    fn test_receive_message() {
        let selector = TransportSelector::with_config(DorsConfig::default());
        let mut manager = TransportManager::new(selector);

        let transport = MockTransport::new(TransportType::BLE);
        transport.start().unwrap();
        let message = create_test_message();
        transport.queue_message(message.clone());

        manager.add_transport(TransportType::BLE, Box::new(transport));

        let received = manager.receive().unwrap();
        assert!(matches!(received, Some((TransportType::BLE, _))));
    }

    #[test]
    fn test_record_delivery_success_enriches_metrics() {
        let selector = TransportSelector::with_config(DorsConfig::default());
        let mut manager = TransportManager::new(selector);

        let transport = MockTransport::new(TransportType::BLE);
        transport.start().unwrap();
        manager.add_transport(TransportType::BLE, Box::new(transport));

        manager.record_delivery_success(TransportType::BLE, 150, 2);

        let metrics = manager.get_available_transports();
        let ble_metrics = metrics.get(&TransportType::BLE).expect("metrics");

        assert_eq!(ble_metrics.success_count, 1);
        assert_eq!(ble_metrics.failure_count, 0);
        assert!(ble_metrics.delivery_ratio.expect("delivery ratio") > 0.99);
        assert_eq!(ble_metrics.average_hop_count, Some(2.0));
        assert_eq!(ble_metrics.latency_ms, Some(150));
    }

    #[test]
    fn test_record_delivery_failure_enriches_metrics() {
        let selector = TransportSelector::with_config(DorsConfig::default());
        let mut manager = TransportManager::new(selector);

        let transport = MockTransport::new(TransportType::BLE);
        transport.start().unwrap();
        manager.add_transport(TransportType::BLE, Box::new(transport));

        manager.record_delivery_failure(TransportType::BLE);
        let metrics = manager.get_available_transports();
        let ble_metrics = metrics.get(&TransportType::BLE).expect("metrics");

        assert_eq!(ble_metrics.success_count, 0);
        assert_eq!(ble_metrics.failure_count, 1);
        assert!(ble_metrics.drop_rate.expect("drop ratio") > 0.99);
    }

    /// A mesh transport with no send confirmation loop takes the engine's
    /// delivery evidence even when it carries a stray count of its own: the
    /// value-based rule discarded every acknowledgement after one (#557).
    #[test]
    fn engine_observations_win_over_a_mesh_transports_own_counts() {
        let selector = TransportSelector::with_config(DorsConfig::default());
        let mut manager = TransportManager::new(selector);

        let transport = MockTransport::new(TransportType::BLE);
        transport.start().unwrap();
        transport.set_metrics(TransportMetrics {
            failure_count: 1,
            delivery_ratio: Some(0.0),
            drop_rate: Some(1.0),
            ..Default::default()
        });
        manager.add_transport(TransportType::BLE, Box::new(transport));

        manager.record_delivery_success(TransportType::BLE, 150, 1);

        let metrics = manager.get_available_transports();
        let ble = metrics.get(&TransportType::BLE).expect("metrics");
        assert_eq!((ble.success_count, ble.failure_count), (1, 0));
        assert_eq!(ble.delivery_ratio, Some(1.0));
        assert_eq!(ble.drop_rate, Some(0.0));
    }

    /// A transport with a confirmation loop owns its counts: the platform's
    /// per-send outcomes are better evidence than the engine's
    /// acknowledgements and must not be overwritten.
    #[test]
    fn a_confirmation_loop_transport_keeps_its_own_counts() {
        let selector = TransportSelector::with_config(DorsConfig::default());
        let mut manager = TransportManager::new(selector);

        let transport = MockTransport::new(TransportType::Internet);
        transport.start().unwrap();
        transport.set_metrics(TransportMetrics {
            success_count: 5,
            failure_count: 1,
            delivery_ratio: Some(5.0 / 6.0),
            drop_rate: Some(1.0 / 6.0),
            ..Default::default()
        });
        manager.add_transport(TransportType::Internet, Box::new(transport));

        manager.record_delivery_failure(TransportType::Internet);

        let metrics = manager.get_available_transports();
        let internet = metrics.get(&TransportType::Internet).expect("metrics");
        assert_eq!((internet.success_count, internet.failure_count), (5, 1));
        assert_eq!(internet.delivery_ratio, Some(5.0 / 6.0));
    }

    /// End to end over a real BLE transport: an inbound partial message
    /// evicted at capacity must not read as a 0% delivery ratio, so a
    /// handful of sends afterwards neither trips the low-success-rate
    /// escalation nor hides the engine's acknowledgements (#557).
    #[test]
    fn a_dropped_inbound_fragment_does_not_escalate_ble() {
        use offline_protocol_transport::constants::BLE_MAX_FRAGMENT_ASSEMBLIES;
        use offline_protocol_transport::{BleTransport, PeerDevice};

        let selector = TransportSelector::with_config(DorsConfig::default());
        let mut manager = TransportManager::new(selector);

        let ble = BleTransport::new("alice-device");
        ble.start().unwrap();
        ble.on_peer_discovered(PeerDevice {
            device_id: "bob".to_string(),
            address: "AA:BB:CC:DD:EE:FF".to_string(),
            rssi: -60,
            last_seen: std::time::SystemTime::now(),
            connected: true,
        });
        manager.add_transport(TransportType::BLE, Box::new(ble));
        let wifi = MockTransport::new(TransportType::WiFiDirect);
        wifi.start().unwrap();
        manager.add_transport(TransportType::WiFiDirect, Box::new(wifi));

        // Fill the reassembly cache with first fragments of multi-fragment
        // messages, then one more: the cache evicts a partial assembly.
        let ble = manager.get_transport(TransportType::BLE).expect("ble");
        let ble_ref = ble.as_any().downcast_ref::<BleTransport>().expect("ble");
        for _ in 0..=BLE_MAX_FRAGMENT_ASSEMBLIES {
            let large = Message::new(
                UserId::new("bob").unwrap(),
                UserId::new("alice").unwrap(),
                AppId::new("test").unwrap(),
                "x".repeat(2_000),
            );
            let fragments = ble_ref.fragment_message(&large).expect("fragments");
            assert!(fragments.len() > 1, "the message must span fragments");
            ble.on_fragment_received_from(fragments[0].clone(), "bob".to_string())
                .expect("fragment accepted");
        }
        drop(ble);

        // More sends than the low-success-rate sample floor.
        for _ in 0..8 {
            manager.send(&create_test_message()).expect("send");
        }

        assert!(
            !manager.should_escalate_to_wifi(),
            "a receive-side reassembly drop must not read as a failing BLE link"
        );
        let metrics = manager.get_available_transports();
        assert_ne!(
            metrics[&TransportType::BLE].delivery_ratio,
            Some(0.0),
            "BLE must not report a 0% delivery ratio with no failed delivery"
        );

        manager.record_delivery_success(TransportType::BLE, 120, 1);
        let metrics = manager.get_available_transports();
        assert_eq!(metrics[&TransportType::BLE].delivery_ratio, Some(1.0));
    }

    /// A carrier that refuses the recipient with nothing left to fall back to
    /// ends the send as `PeerNotReachable`, not as a carrier that broke
    /// mid-send. The deferral token and every classifier read this variant.
    #[test]
    fn a_sole_carrier_refusing_the_recipient_ends_as_peer_not_reachable() {
        let selector = TransportSelector::with_config(DorsConfig::default());
        let mut manager = TransportManager::new(selector);

        let transport = MockTransport::new(TransportType::BLE);
        transport.start().unwrap();
        transport.set_reject_unknown_recipients(true);
        manager.add_transport(TransportType::BLE, Box::new(transport));

        let err = manager.send(&create_test_message()).unwrap_err();
        assert!(
            matches!(err, Error::Transport(TransportError::PeerNotReachable(_))),
            "got {err:?}"
        );
    }

    /// Two transports, neither reporting a battery level of its own — the
    /// ordinary case, since no platform transport publishes one. Both must
    /// come back carrying the host reading, because every downstream consumer
    /// (DORS energy scoring, the relay role, the forwarding floor) resolves
    /// battery through this map and would otherwise take its unknown branch.
    #[test]
    fn test_device_battery_fills_transports_that_report_none() {
        let selector = TransportSelector::with_config(DorsConfig::default());
        let mut manager = TransportManager::new(selector);

        for transport_type in [TransportType::BLE, TransportType::WiFiDirect] {
            let transport = MockTransport::new(transport_type);
            transport.start().unwrap();
            manager.add_transport(transport_type, Box::new(transport));
        }

        assert_eq!(manager.device_battery(), (None, false));
        manager.set_device_battery(42, true);
        assert_eq!(manager.device_battery(), (Some(42), true));

        let metrics = manager.get_available_transports();
        for transport_type in [TransportType::BLE, TransportType::WiFiDirect] {
            let m = metrics.get(&transport_type).expect("metrics");
            assert_eq!(m.battery_level, Some(42));
            assert!(m.is_charging);
        }

        // The one-pass snapshot builds the same map and must agree with it —
        // the telemetry aggregator and the relay-role evaluation read this one.
        let (_statuses, available) = manager.snapshot_status_and_available();
        let m = available.get(&TransportType::BLE).expect("metrics");
        assert_eq!(m.battery_level, Some(42));
        assert!(m.is_charging);
    }

    /// A transport that publishes its own battery reading keeps it, and keeps
    /// its own charging state with it: mixing a level from one source with a
    /// charging flag from another is what produces the spurious
    /// charging-state flips `device_battery_from_available` documents.
    #[test]
    fn test_transport_reported_battery_wins_over_device_feed() {
        let selector = TransportSelector::with_config(DorsConfig::default());
        let mut manager = TransportManager::new(selector);

        let transport = MockTransport::new(TransportType::BLE);
        transport.start().unwrap();
        transport.set_metrics(TransportMetrics {
            battery_level: Some(7),
            is_charging: false,
            ..Default::default()
        });
        manager.add_transport(TransportType::BLE, Box::new(transport));

        manager.set_device_battery(90, true);

        let metrics = manager.get_available_transports();
        let m = metrics.get(&TransportType::BLE).expect("metrics");
        assert_eq!(m.battery_level, Some(7));
        assert!(!m.is_charging);
    }

    #[test]
    fn test_device_battery_is_clamped() {
        let selector = TransportSelector::with_config(DorsConfig::default());
        let mut manager = TransportManager::new(selector);
        manager.set_device_battery(240, false);
        assert_eq!(manager.device_battery(), (Some(100), false));
    }
}