bacnet-endpoint 0.15.0

//! Single-owner endpoint session: one ingress + shared coordinator.
//!
//! Lifecycle lives here ONLY. Role handles expose no lifecycle methods.
//!
//! # Lifecycle
//!
//! [`EndpointSession::new`] owns `transport` without starting it. The owner
//! drives [`start`](EndpointSession::start) once, then
//! [`stop`](EndpointSession::stop) once; both take `&mut self` so only the
//! owner can drive them. `Drop` aborts dispatch/ingress/role work without
//! orphaning, but `stop()` remains the graceful path that joins termination
//! and reports the [`SessionExit`].
//!
//! # Cancellation and drop
//!
//! Outbound leases use RAII guards. Ordinary `read_property*` calls release their
//! exact lease when cancelled. Source-audited calls transfer to bounded session
//! ownership before egress admission, so dropping their caller does not cancel
//! admitted work. `stop()` seals admission, closes roles, joins dispatch and owned
//! workers, then stops ingress. Undelivered audit records may be lost on shutdown.
//! Before reporting termination, it settles forgotten durable writes and awaits
//! queued save attempts with the database guard released. Canceling this wait
//! preserves one settlement task for the next `stop()` call to join.
//! Cloned role handles keep working until shutdown, then fail closed; after
//! the session drops, [`Weak`](std::sync::Weak) upgrade fails and every role
//! call reports shutdown.
//!
//! The session keeps its object database through `stop()` and lets go of it
//! when it drops. Every endpoint holder of the database (the session, the
//! server role's responder, a cloned server role handle, the source Audit
//! runtime and the session's tasks) lets go through
//! [`drop_database_off_runtime`], so whichever goes last, in async code the
//! objects and any durable saves they wait for drop on Tokio's blocking pool,
//! not on a runtime worker (#1561).
//!
//! ```compile_fail,E0596
//! // Lifecycle is owner-exclusive: `stop` takes `&mut self`, so a shared
//! // borrow cannot drive shutdown.
//! # use bacnet_transport::loopback::LoopbackTransport;
//! # use bacnet_endpoint::session::EndpointSession;
//! # fn forbidden(session: &EndpointSession<LoopbackTransport>) {
//! session.stop();
//! # }
//! ```
//!
//! # Thread safety
//!
//! [`EndpointSession`] is `Send` when its transport is `Send`: the dispatch
//! task, channels, and shared counters are all `Send`-owned. Shared `&self`
//! borrows (`client`, `server`, counters, `broadcast_i_am`) are safe because
//! lifecycle mutation is owner-exclusive (`&mut`) or via atomics/async
//! mutexes; role handles synchronize through the shared session token.

use std::sync::Arc;
use std::sync::atomic::{AtomicU8, Ordering};

use bacnet_client::EndpointRequester;
use bacnet_client::client::ClientConfig;
use bacnet_encoding::apdu::Apdu;
use bacnet_endpoint_core::coordinator::{
    AdmissionOutcome, CoordinatorError, OutboundTransactionCoordinator,
};
use bacnet_endpoint_core::endpoint_ingress::{
    ClassifierExit, EndpointIngress, PolicyOutcome, PolicyReason,
};
use bacnet_network::layer::ReceivedApdu;
use bacnet_network::network_number::LocalNetworkNumber;
use bacnet_objects::database::ObjectDatabase;
use bacnet_objects::traits::BACnetObject;
use bacnet_transport::port::TransportPort;
use bacnet_types::enums::{ObjectType, PropertyIdentifier};
use bacnet_types::error::Error;
use bacnet_types::primitives::{ObjectIdentifier, PropertyValue};
use tokio::sync::{Mutex, RwLock, mpsc, oneshot};
use tokio::task::JoinHandle;

#[path = "source_profile.rs"]
mod source_profile;
#[path = "source_reporter.rs"]
pub(crate) mod source_reporter;
#[path = "session_start.rs"]
mod startup;

#[path = "device_writes.rs"]
mod device_writes;
pub use device_writes::{RESPONDER_OPT_IN_SERVICES, RESPONDER_SERVICES, ResponderOptIn};
#[path = "registered_port.rs"]
mod registered_port;

use crate::roles::{
    ClientRoleHandle, ServerRoleHandle, SessionToken, admit_once, decode_terminal,
    inbound_canonical_peer, is_requester_lease,
};
use bacnet_server::server::{
    __endpoint_EndpointResponder as EndpointResponder,
    __endpoint_NotificationTransactions as NotificationTransactions,
    __endpoint_settle_durable_writes as settle_durable_writes, drop_database_off_runtime,
};

#[derive(Clone, Copy, PartialEq, Eq)]
enum Lifecycle {
    Ready = 0,
    Running = 1,
    Stopped = 2,
    Stopping = 3,
}

/// Which roles the session composes.
///
/// Chosen at [`EndpointSession::new`] time (or via the endpoint builders'
/// `role()` setter) and fixed for the session lifetime. Determines which
/// role handles exist after [`start`](EndpointSession::start):
///
/// - [`ClientOnly`](SessionRole::ClientOnly): `client()` is `Some`, `server()`
///   is `None`. Inbound requests are counted (`no_server_role`) and their
///   one-use reply sender released without sending.
/// - [`ServerOnly`](SessionRole::ServerOnly): `server()` is `Some`,
///   `client()` is `None`. Terminal responses with no consumer release their
///   exact lease and count `no_client_role`.
/// - [`Both`](SessionRole::Both): both handles are present above the one
///   shared coordinator.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum SessionRole {
    /// Client requester only.
    ClientOnly,
    /// Server responder + notification pool only.
    ServerOnly,
    /// Both sibling roles above one ingress/coordinator.
    Both,
}

/// Session tuning (bounded queues, client timers, the responder's read work
/// limit).
///
/// Typed at the public boundary: every field is validated where it matters
/// (`queue_capacity == 0`, `read_work_limit == 0` or a
/// `min_request_interval_ms` past an hour fails [`EndpointSession::new`];
/// APDU/timer values flow into the client role config unchanged).
///
/// ```
/// use bacnet_endpoint::session::SessionConfig;
///
/// let config = SessionConfig::default();
/// assert!(config.queue_capacity > 0);
/// assert_eq!(config.max_apdu_length, 480);
/// assert_eq!(config.read_work_limit, 256);
/// assert_eq!(config.min_request_interval_ms, 0);
/// ```
#[derive(Clone, Debug)]
pub struct SessionConfig {
    /// Bounded capacity for each ingress queue + egress channel.
    ///
    /// Must be greater than zero; [`EndpointSession::new`] returns
    /// [`Error::Encoding`] otherwise.
    pub queue_capacity: usize,
    /// Client APDU timeout (ms).
    pub apdu_timeout_ms: u64,
    /// Client APDU retries.
    pub apdu_retries: u8,
    /// Client max APDU length.
    ///
    /// Standalone default is 480. Composing a
    /// [`DeviceIdentity`](crate::identity::DeviceIdentity) via
    /// [`device`](EndpointSession::device) overrides this at start with the
    /// identity's ceiling resolved against the transport's stable local
    /// receive capacity (480 for standard MS/TP, 1476 for extended MS/TP).
    pub max_apdu_length: u16,
    /// Result rows one ReadProperty served by the server role may expand
    /// (default 256).
    ///
    /// The endpoint's counterpart of the server's
    /// [`ReadPropertyMultipleBudget::max_result_elements`](bacnet_server::server::ReadPropertyMultipleBudget::max_result_elements),
    /// with the same default. A read counts its own row, and a Group's
    /// Present_Value adds one row per member property after ALL, REQUIRED
    /// and OPTIONAL expand. A read past the limit is answered with an Abort
    /// carrying OUT_OF_RESOURCES before any member is read.
    ///
    /// Must be greater than zero; [`EndpointSession::new`] returns
    /// [`Error::Encoding`] otherwise.
    pub read_work_limit: usize,
    /// Least time, in milliseconds, between the client role's confirmed
    /// requests to one destination (default 0: no pacing), measured as
    /// [`ClientConfig::min_request_interval_ms`] says for `BACnetClient`
    /// (#1542).
    ///
    /// Only new confirmed requests wait: a retry keeps its request's turn,
    /// and replies, notifications and unconfirmed requests go at once.
    /// Stopping or dropping the session ends a wait at once with the
    /// shutdown error. At
    /// most [`MAX_MIN_REQUEST_INTERVAL_MS`](bacnet_client::client::MAX_MIN_REQUEST_INTERVAL_MS),
    /// an hour; [`EndpointSession::new`] returns [`Error::Encoding`] past it.
    pub min_request_interval_ms: u64,
}

impl Default for SessionConfig {
    fn default() -> Self {
        Self {
            queue_capacity: 16,
            apdu_timeout_ms: 1_000,
            apdu_retries: 0,
            max_apdu_length: 480,
            read_work_limit: bacnet_server::server::ReadPropertyMultipleBudget::default()
                .max_result_elements,
            min_request_interval_ms: 0,
        }
    }
}

/// Snapshot of policy-outcome ownership (session-owned, bounded).
///
/// The session counts every classifier/policy outcome instead of dropping it
/// silently. Sampled via [`policy_counters`](EndpointSession::policy_counters).
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
pub struct PolicyCountersSnapshot {
    /// Ingress classifier outcomes owned by the session.
    pub ingress_policy: u64,
    /// Inbound requests with no server role (client-only).
    pub no_server_role: u64,
    /// Terminal responses with no client/notification role.
    pub no_client_role: u64,
    /// Terminal responses no lease claimed (unknown/peer/service/policy).
    pub unclaimed_terminal: u64,
    /// Responder declined (group/unsupported/closed).
    pub responder_declined: u64,
}

struct PolicyCounters {
    ingress_policy: u64,
    no_server_role: u64,
    no_client_role: u64,
    unclaimed_terminal: u64,
    responder_declined: u64,
}

struct SessionShared {
    token: Arc<SessionToken>,
    counters: Mutex<PolicyCounters>,
}

/// One owned single-device endpoint session.
///
/// Owns ONE [`EndpointIngress`], the shared [`OutboundTransactionCoordinator`],
/// role registration, policy-outcome ownership, timers, egress and
/// termination. Start-once/stop-once; `Drop` aborts without orphaning.
///
/// An optional [`DeviceIdentity`](crate::identity::DeviceIdentity) is the
/// single source for I-Am + Device readback + role limits when composed via
/// [`device`](Self::device). Standalone sessions without an
/// identity keep the `SessionConfig` 480 default untouched.
///
/// ```no_run
/// use bacnet_endpoint::session::{EndpointSession, SessionConfig, SessionRole};
/// use bacnet_transport::loopback::LoopbackTransport;
///
/// # #[tokio::main]
/// # async fn main() -> Result<(), bacnet_types::error::Error> {
/// let (transport, _peer) = LoopbackTransport::pair(vec![0x01], vec![0x02]);
/// let mut session =
///     EndpointSession::new(transport, SessionRole::Both, SessionConfig::default())?;
/// session.start().await?;
/// assert!(session.is_running());
/// session.stop().await?;
/// # Ok(())
/// # }
/// ```
///
/// `T` defaults to [`AnyTransport`](bacnet_transport::any::AnyTransport), so
/// `EndpointSession` alone names a session over any built-in transport;
/// `EndpointSession<BipTransport>` and the like name one over a concrete
/// transport, as the builders return.
pub struct EndpointSession<T: TransportPort + 'static = bacnet_transport::any::AnyTransport> {
    shared: Arc<SessionShared>,
    coordinator: Arc<OutboundTransactionCoordinator>,
    ingress: Option<EndpointIngress<T>>,
    requester: Option<EndpointRequester>,
    responder: Option<Arc<EndpointResponder>>,
    notifications: Option<Arc<NotificationTransactions>>,
    client_handle: Option<ClientRoleHandle>,
    server_handle: Option<ServerRoleHandle>,
    dispatch_task: Option<JoinHandle<SessionExit>>,
    network_number_task: Option<JoinHandle<()>>,
    cancel_tx: Option<oneshot::Sender<()>>,
    lifecycle: AtomicU8,
    role: SessionRole,
    #[allow(dead_code)]
    config: SessionConfig,
    client_config: ClientConfig,
    database: Option<Arc<RwLock<ObjectDatabase>>>,
    source_audit_reporter: Option<ObjectIdentifier>,
    source_audit: Option<Arc<crate::source_audit::SourceAudit>>,
    pub(crate) source_audit_bindings: Vec<(ObjectIdentifier, std::net::SocketAddrV4)>,
    source_recipient: Option<Arc<crate::source_audit::recipient::SourceRecipient>>,
    stop_exit: Option<Result<SessionExit, Error>>,
    durable_settlement: Option<JoinHandle<()>>,
    ingress_stopped: bool,
    identity: Option<crate::identity::DeviceIdentity>,
    local_receive_capacity: u16,
    bip_local_address: Option<std::net::SocketAddrV4>,
    registered_network_port: Option<ObjectIdentifier>,
    registered_port_lease: std::sync::Weak<()>,
    device_write_authorizer: Option<bacnet_server::mutation::DeviceWriteAuthorizer>,
    pub(crate) reinitialize: Option<bacnet_server::server::ReinitializeHandler>,
    pub(crate) reinit_password: Option<String>,
    egress: Option<bacnet_endpoint_core::endpoint_ingress::EndpointEgress>,
    /// The B/IP transport's own exchanges, when the B/IP builder built it.
    pub(crate) bip_status: Option<bacnet_transport::bip::BvlcClientStatus>,
    /// The MS/TP transport's counters, when the MS/TP builder built it.
    pub(crate) mstp_diagnostics: Option<bacnet_transport::mstp::MstpDiagnostics>,
}

/// Terminal state of the session dispatch task.
///
/// Returned by [`stop`](EndpointSession::stop): either an explicit stop won
/// the race, the ingress classifier exited, or all dispatch receivers closed.
#[derive(Debug)]
pub enum SessionExit {
    /// Explicit `stop()` won the race.
    Cancelled,
    /// Ingress classifier exited (input closed / policy route full/closed).
    Ingress(ClassifierExit),
    /// Dispatch receivers all closed.
    ReceiversClosed,
}

impl<T: TransportPort + 'static> EndpointSession<T> {
    /// Creates a session owning `transport` (not yet started).
    ///
    /// Returns [`Error::Encoding`] when `config.queue_capacity` or
    /// `config.read_work_limit` is zero, or `config.min_request_interval_ms`
    /// is more than an hour. Normally built via
    /// [`BipEndpointBuilder`](crate::bip::BipEndpointBuilder),
    /// [`ScEndpointBuilder`](crate::sc::ScEndpointBuilder), or
    /// [`MstpEndpointBuilder`](crate::mstp::MstpEndpointBuilder) instead of
    /// directly.
    ///
    /// ```
    /// use bacnet_endpoint::session::{EndpointSession, SessionConfig, SessionRole};
    /// use bacnet_transport::loopback::LoopbackTransport;
    ///
    /// let (transport, _peer) = LoopbackTransport::pair(vec![0x01], vec![0x02]);
    /// let bad = SessionConfig {
    ///     queue_capacity: 0,
    ///     ..SessionConfig::default()
    /// };
    /// assert!(EndpointSession::new(transport, SessionRole::Both, bad).is_err());
    /// ```
    pub fn new(transport: T, role: SessionRole, config: SessionConfig) -> Result<Self, Error> {
        if config.queue_capacity == 0 {
            return Err(Error::Encoding(
                "endpoint session queue capacity must be greater than zero".into(),
            ));
        }
        if config.read_work_limit == 0 {
            return Err(Error::Encoding(
                "endpoint session read work limit must be greater than zero".into(),
            ));
        }
        let interval = config.min_request_interval_ms;
        if interval > bacnet_client::client::MAX_MIN_REQUEST_INTERVAL_MS {
            return Err(Error::Encoding(format!(
                "endpoint session min-request-interval {interval} ms is more than {} ms",
                bacnet_client::client::MAX_MIN_REQUEST_INTERVAL_MS
            )));
        }
        let coordinator = Arc::new(OutboundTransactionCoordinator::new());
        let token = SessionToken::new(Arc::clone(&coordinator));
        let client_config = ClientConfig {
            apdu_timeout_ms: config.apdu_timeout_ms,
            apdu_retries: config.apdu_retries,
            max_apdu_length: config.max_apdu_length,
            min_request_interval_ms: interval,
            ..ClientConfig::default()
        };
        let local_receive_capacity = transport.local_receive_apdu_capacity();
        Ok(Self {
            local_receive_capacity,
            shared: Arc::new(SessionShared {
                token,
                counters: Mutex::new(PolicyCounters {
                    ingress_policy: 0,
                    no_server_role: 0,
                    no_client_role: 0,
                    unclaimed_terminal: 0,
                    responder_declined: 0,
                }),
            }),
            coordinator,
            ingress: Some(EndpointIngress::new(transport, config.queue_capacity)),
            requester: None,
            responder: None,
            notifications: None,
            client_handle: None,
            server_handle: None,
            dispatch_task: None,
            network_number_task: None,
            cancel_tx: None,
            lifecycle: AtomicU8::new(Lifecycle::Ready as u8),
            role,
            config,
            client_config,
            database: None,
            source_audit_reporter: None,
            source_audit: None,
            source_audit_bindings: Vec::new(),
            source_recipient: None,
            stop_exit: None,
            durable_settlement: None,
            ingress_stopped: false,
            identity: None,
            bip_local_address: None,
            registered_network_port: None,
            registered_port_lease: std::sync::Weak::new(),
            device_write_authorizer: None,
            reinitialize: None,
            reinit_password: None,
            egress: None,
            bip_status: None,
            mstp_diagnostics: None,
        })
    }

    /// Attaches the local object database (before start).
    ///
    /// Must be called before [`start`](Self::start); when the server role is
    /// composed without an identity or database, an empty one is used. With a
    /// [`DeviceIdentity`](crate::identity::DeviceIdentity), omitting the database
    /// requests stock generation after receive-capacity resolution. Explicitly
    /// supplied databases, including empty ones, are preserved. Their selected
    /// Device must match the resolved identity and declare
    /// segmentation NONE, matching the narrow endpoint roles.
    /// The session retains ownership even in `ClientOnly` mode and shares this
    /// same database with the responder in `Both` mode.
    ///
    /// # Panics
    /// Panics if startup has already consumed the session configuration.
    pub fn with_database(mut self, db: ObjectDatabase) -> Self {
        self.assert_configurable();
        self.database = Some(Arc::new(RwLock::new(db)));
        self
    }

    /// Select the sole source Audit Reporter in the attached local database.
    ///
    /// Activates bounded source READ reporting on direct IPv4 B/IP. The built-in
    /// Device must have a typed Audit recipient provision, even at Audit_Level
    /// NONE. Device choices use immutable builder route bindings; direct Address
    /// choices need no binding. Monitored_Objects must be absent. `Both` requires
    /// an explicit [`with_device_writes`](Self::with_device_writes) authorizer;
    /// `ClientOnly` permits trusted local changes and `ServerOnly` is rejected.
    /// Standalone BACnetClient source reporting remains unsupported.
    ///
    /// [`start`](Self::start) requires a concrete built-in local Device
    /// ([`ObjectDatabase::local_device`]: the lowest instance when the database
    /// holds several) matching the optional identity, and one selected Reporter capability with
    /// no conflicting source Reporter. Preflight validation changes no source flags
    /// and leaves configuration retryable; post-ingress failure joins terminal cleanup.
    /// Successful startup installs the Device
    /// recipient mutation owner and source projection without sending traffic.
    /// Device/Reporter membership is protected until owned frames quiesce. The
    /// private adapter forwards the original object behavior when sealed/released.
    /// Repeated calls before start replace the selection.
    ///
    /// ```compile_fail,E0603
    /// use bacnet_endpoint::session::source_reporter::SourceReporter;
    /// ```
    ///
    /// ```no_run
    /// use std::net::{Ipv4Addr, SocketAddrV4};
    /// use bacnet_endpoint::{bip::BipEndpointBuilder, DeviceIdentity, SessionRole};
    /// use bacnet_objects::{audit::AuditReporterObject, traits::BACnetObject};
    /// use bacnet_types::{constructed::BACnetRecipient, enums::ObjectType, primitives::ObjectIdentifier};
    /// # async fn example() -> Result<(), bacnet_types::error::Error> {
    /// let identity = DeviceIdentity::new(123, 42)?;
    /// let mut db = identity.build_database()?;
    /// let logger = ObjectIdentifier::new(ObjectType::DEVICE, 999)?;
    /// db.get_mut(&identity.device_oid()).unwrap().device_authority_internal().unwrap()
    ///     .provision_audit_recipient(BACnetRecipient::Device(logger))?;
    /// let reporter = AuditReporterObject::new(1, "Source configuration")?;
    /// let reporter_oid = reporter.object_identifier();
    /// db.add(Box::new(reporter))?;
    /// let mut session = BipEndpointBuilder::new(Ipv4Addr::LOCALHOST, 0, Ipv4Addr::BROADCAST)
    ///     .role(SessionRole::ClientOnly).database(db).device(identity)
    ///     .source_audit_device_binding(logger, SocketAddrV4::new(Ipv4Addr::LOCALHOST, 47808))
    ///     .build_session()?.with_source_audit_reporter(reporter_oid);
    /// session.start().await?; // Startup emits no audit traffic.
    /// session.stop().await?;
    /// # Ok(())
    /// # }
    /// ```
    ///
    /// # Panics
    /// Panics if startup has already consumed the session configuration.
    pub fn with_source_audit_reporter(mut self, reporter: ObjectIdentifier) -> Self {
        self.assert_configurable();
        self.source_audit_reporter = Some(reporter);
        self
    }

    /// Composes the single Device identity (before start).
    ///
    /// Truth direction: identity overrides the standalone `SessionConfig`
    /// 480 default for the client role max-APDU; timers/retries stay from
    /// `SessionConfig`. Startup resolves the ceiling against the owned link.
    /// An absent database requests stock generation; an explicitly supplied
    /// database is preserved and its selected Device must match the resolved
    /// identity. Sessions without an identity keep the 480 default.
    ///
    /// # Panics
    /// Panics if startup has already consumed the session configuration.
    pub fn device(mut self, identity: crate::identity::DeviceIdentity) -> Self {
        self.assert_configurable();
        self.client_config.max_apdu_length = identity.max_apdu_length();
        self.config.max_apdu_length = identity.max_apdu_length();
        self.identity = Some(identity);
        self
    }

    fn assert_configurable(&self) {
        assert_eq!(
            self.lifecycle.load(Ordering::Acquire),
            Lifecycle::Ready as u8,
            "endpoint configuration must precede startup"
        );
    }

    /// Starts ingress, roles and the single dispatch consumer once.
    ///
    /// Start-once: a second call returns
    /// [`Error::Encoding`] without
    /// binding again. Takes `&mut self` so only the owner can start.
    /// Source-profile validation runs before lifecycle consumption or ingress
    /// startup; its errors leave the session ready for correction and retry.
    /// Profile initialization errors after ingress starts cancel and join ingress
    /// before returning, leaving the session terminal. Canceling that cleanup
    /// leaves a stopping session whose teardown can resume with [`stop`](Self::stop).
    pub async fn start(&mut self) -> Result<(), Error> {
        if self.lifecycle.load(Ordering::Acquire) != Lifecycle::Ready as u8 {
            return Err(Error::Encoding(
                "endpoint session cannot be started more than once".into(),
            ));
        }
        let (device_target, source_routes) = self.prepare_start().await?;
        if self.lifecycle.compare_exchange(
            Lifecycle::Ready as u8,
            Lifecycle::Running as u8,
            Ordering::AcqRel,
            Ordering::Acquire,
        ) != Ok(Lifecycle::Ready as u8)
        {
            return Err(Error::Encoding(
                "endpoint session cannot be started more than once".into(),
            ));
        }
        let ingress = self
            .ingress
            .as_mut()
            .ok_or_else(|| Error::Encoding("endpoint session ingress owner is missing".into()))?;
        let receivers = match ingress.start().await {
            Ok(receivers) => receivers,
            Err(error) => {
                let _ = self.stop().await;
                return Err(error);
            }
        };
        if let Err(error) = self
            .publish_registered_port(
                (receivers.bip_port.clone(), receivers.bip6_port.clone()),
                receivers.egress.local_network_number(),
            )
            .await
        {
            let _ = self.stop().await;
            return Err(error);
        }
        let bip_local_address = receivers.bip_local_address;
        if let Err(error) = self.start_roles(receivers, source_routes, device_target) {
            // Keep cleanup ownership in self before awaiting. Cancellation leaves
            // Stopping plus intact joins; stop/Drop can still finish teardown.
            let _ = self.stop().await;
            return Err(error);
        }
        self.bip_local_address = bip_local_address;
        Ok(())
    }

    /// Stops dispatch, roles and ingress once; joins termination.
    ///
    /// Returns the dispatch [`SessionExit`]. A canceled stop retains its joins
    /// and sealed membership protection; a later call finishes teardown. Calling before
    /// `start()` or twice returns
    /// [`Error::Encoding`]. Takes
    /// `&mut self` so only the owner can stop; cloned role handles observe
    /// shutdown and fail closed.
    ///
    /// With an attached database, waits for its lock and for queued durable save
    /// attempts, including corrections for abandoned staged writes. Release any
    /// application-held database guard so shutdown can progress. A stalled backend
    /// keeps this call pending; completion does not certify backend success or
    /// power-loss durability. The database remains attached after termination.
    pub async fn stop(&mut self) -> Result<SessionExit, Error> {
        let state = self.lifecycle.load(Ordering::Acquire);
        if state != Lifecycle::Running as u8 && state != Lifecycle::Stopping as u8 {
            return Err(Error::Encoding("endpoint session is not running".into()));
        }
        self.lifecycle
            .store(Lifecycle::Stopping as u8, Ordering::Release);
        self.shared.token.shutdown();
        if let Some(task) = self.network_number_task.as_mut() {
            task.abort();
            let _ = task.await;
        }
        self.network_number_task = None;
        if let Some(source) = &self.source_audit {
            source.close();
        }
        if let Some(requester) = &self.requester {
            requester.close();
        }
        if let Some(responder) = &self.responder {
            responder.close();
        }
        if let Some(notifications) = &self.notifications {
            notifications.close();
        }
        if let Some(cancel) = self.cancel_tx.take() {
            let _ = cancel.send(());
        }
        if self.stop_exit.is_none() {
            self.stop_exit = Some(match self.dispatch_task.as_mut() {
                Some(task) => task.await.map_err(|error| {
                    Error::Encoding(format!("endpoint session dispatch failed: {error}"))
                }),
                None => Ok(SessionExit::ReceiversClosed),
            });
            self.dispatch_task.take();
        }
        if let Some(notifications) = &self.notifications {
            while let Some(result) = notifications.join_next().await {
                NotificationTransactions::observe(Some(result));
            }
        }
        if !self.ingress_stopped {
            if let Some(ingress) = self.ingress.as_mut()
                && let Err(error) = ingress.stop().await
                && self.stop_exit.as_ref().is_some_and(Result::is_ok)
            {
                self.stop_exit = Some(Err(error));
            }
            self.ingress_stopped = true;
        }
        // The owner remains in the session through every await. The DB barrier
        // observes completed synchronous commits; queued source readers recheck
        // the sealed weak sink after taking this guard and cannot reenter it.
        if let Some(runtime) = &self.source_recipient {
            runtime.uninstall(
                &mut *self
                    .database
                    .as_ref()
                    .expect("source database")
                    .write()
                    .await,
            );
        }
        self.source_recipient.take();
        self.source_audit.take();
        self.notifications.take();
        self.requester.take();
        self.responder.take();
        self.client_handle = None;
        self.server_handle = None;
        if self.durable_settlement.is_none()
            && let Some(db) = &self.database
        {
            self.durable_settlement = Some(tokio::spawn(settle_durable_writes(Arc::clone(db))));
        }
        if let Some(task) = self.durable_settlement.as_mut() {
            // Keep the one settlement owner across canceled stop waiters.
            let result = task.await;
            self.durable_settlement = None;
            result.map_err(|error| {
                Error::Encoding(format!("endpoint durable settlement failed: {error}"))
            })?;
        }
        self.lifecycle
            .store(Lifecycle::Stopped as u8, Ordering::Release);
        self.stop_exit.take().expect("joined dispatch")
    }

    /// Borrows the client role (`None` when not composed).
    ///
    /// Returns `None` for [`ServerOnly`](SessionRole::ServerOnly) sessions,
    /// and after [`stop`](Self::stop) (handles are released at termination).
    pub fn client(&self) -> Option<&ClientRoleHandle> {
        self.client_handle.as_ref()
    }

    /// Borrows the server role (`None` when not composed).
    ///
    /// Returns `None` for [`ClientOnly`](SessionRole::ClientOnly) sessions,
    /// and after [`stop`](Self::stop).
    pub fn server(&self) -> Option<&ServerRoleHandle> {
        self.server_handle.as_ref()
    }

    /// Clones the client role handle with its session binding.
    ///
    /// The clone holds only a [`Weak`](std::sync::Weak) session token: it
    /// works while the session runs and fails closed after `stop()`/drop.
    /// Used to hand the client role to tasks that outlive the borrow.
    pub fn cloned_client_handle(&self) -> Option<ClientRoleHandle> {
        self.client_handle.clone()
    }

    /// Clones the server role handle with its session binding.
    ///
    /// Same [`Weak`](std::sync::Weak)-token semantics as
    /// [`cloned_client_handle`](Self::cloned_client_handle); see
    /// [`is_session_alive`](crate::roles::ServerRoleHandle::is_session_alive).
    pub fn cloned_server_handle(&self) -> Option<ServerRoleHandle> {
        self.server_handle.clone()
    }

    /// Samples policy-outcome ownership counters.
    ///
    /// Every classifier/policy outcome is counted, never silently dropped.
    pub async fn policy_counters(&self) -> PolicyCountersSnapshot {
        let counters = self.shared.counters.lock().await;
        PolicyCountersSnapshot {
            ingress_policy: counters.ingress_policy,
            no_server_role: counters.no_server_role,
            no_client_role: counters.no_client_role,
            unclaimed_terminal: counters.unclaimed_terminal,
            responder_declined: counters.responder_declined,
        }
    }

    /// Samples the shared outbound lease count.
    ///
    /// Returns `usize::MAX` only when the coordinator lock is poisoned
    /// (internal invariant violation surfaced honestly, never hidden).
    pub fn active_leases(&self) -> usize {
        self.coordinator.active_count().unwrap_or(usize::MAX)
    }

    /// Shares the device-wide coordinator (dispatch + tests only).
    #[doc(hidden)]
    pub fn coordinator(&self) -> Arc<OutboundTransactionCoordinator> {
        Arc::clone(&self.coordinator)
    }

    /// Borrows the composed identity (narrow admin; no lifecycle).
    ///
    /// RB-15 narrow admin borrow: roles hold no transport/session lifecycle;
    /// only the session owner exposes identity + counters + coordinator.
    /// A role handle attempting `stop`/transport access fails at compile
    /// time (no such method) and at runtime its post-stop calls fail closed.
    pub fn identity(&self) -> Option<&crate::identity::DeviceIdentity> {
        self.identity.as_ref()
    }

    /// Actual announced B/IP address after successful startup.
    /// Absent before publication, while stopping, after stop, or for other links.
    pub fn bip_local_address(&self) -> Option<std::net::SocketAddrV4> {
        (self.is_running() && self.egress.as_ref().is_some_and(|egress| egress.is_open()))
            .then_some(self.bip_local_address)
            .flatten()
    }

    /// How the transport's foreign-device registration is going (#938):
    /// whether it holds a live lease, the time left on it, and the last
    /// attempt's result and outcome, a BBMD's refusal included. It is there
    /// for a session [`BipEndpointBuilder`](crate::bip::BipEndpointBuilder)
    /// built, in any mode and after stop, and `None` for any other: take a
    /// [`BipTransport::bvlc_client_status`](bacnet_transport::bip::BipTransport::bvlc_client_status)
    /// before handing a transport to [`Self::new`].
    pub fn foreign_registration(
        &self,
    ) -> Option<bacnet_transport::bip::ForeignRegistrationSnapshot> {
        let status = self.bip_status.as_ref()?;
        Some(status.snapshot().foreign_registration)
    }

    /// The MS/TP transport's counters (#958): a handle to the same
    /// [`MstpDiagnostics`](bacnet_transport::mstp::MstpDiagnostics) the
    /// transport updates, readable in any state, after stop included. It is
    /// there for a session [`MstpEndpointBuilder`](crate::mstp::MstpEndpointBuilder)
    /// built, and `None` for any other: take
    /// [`MstpTransport::diagnostics`](bacnet_transport::mstp::MstpTransport::diagnostics)
    /// before handing a transport to [`Self::new`].
    pub fn mstp_diagnostics(&self) -> Option<bacnet_transport::mstp::MstpDiagnostics> {
        self.mstp_diagnostics.clone()
    }

    /// Returns the composed session role (narrow admin).
    pub fn session_role(&self) -> SessionRole {
        self.role
    }

    /// Returns true while the session dispatch is running (narrow admin).
    pub fn is_running(&self) -> bool {
        self.lifecycle.load(Ordering::Acquire) == Lifecycle::Running as u8
    }

    /// Broadcasts an I-Am consistent with the composed identity.
    ///
    /// Missing-identity path: errors without sending (no invented device).
    /// Present-identity path: encodes [`encode_iam_apdu`](crate::identity::encode_iam_apdu)
    /// and sends one local-broadcast Unconfirmed-Request via the session
    /// egress. Field-for-field identical to the server discovery I-Am built
    /// from [`server_config`](crate::identity::server_config).
    /// Transport-dependent: B/IP reaches the local subnet broadcast; SC
    /// relays via the hub as a broadcast NPDU. Fails when the session is not
    /// running.
    pub async fn broadcast_i_am(&self) -> Result<(), Error> {
        let identity = self
            .identity
            .as_ref()
            .ok_or_else(|| Error::Encoding("endpoint I-Am requires a composed identity".into()))?;
        let egress = self
            .egress
            .as_ref()
            .ok_or_else(|| Error::Encoding("endpoint session is not running".into()))?;
        if !self.is_running() {
            return Err(Error::Encoding("endpoint session is not running".into()));
        }
        let database = self
            .database
            .as_ref()
            .ok_or_else(|| Error::Encoding("endpoint I-Am has no Device database".into()))?;
        let database = database.read().await;
        if !database
            .get(&identity.device_oid())
            .is_some_and(|object| object.object_identifier() == identity.device_oid())
        {
            return Err(Error::Encoding("endpoint I-Am Device is absent".into()));
        }
        let apdu = crate::identity::encode_iam_apdu(identity)?;
        egress
            .send_apdu(
                apdu,
                bacnet_endpoint_core::endpoint_ingress::EndpointApduDestination::LocalBroadcast,
                false,
                bacnet_npdu::NetworkPriority::Normal,
                Vec::new(),
            )
            .await
    }
}

#[cfg(test)]
#[path = "source_reporter_tests.rs"]
mod source_reporter_tests;

#[cfg(test)]
#[path = "device_write_tests.rs"]
mod device_write_tests;

impl<T: TransportPort + 'static> Drop for EndpointSession<T> {
    fn drop(&mut self) {
        // Synchronous abort path: never orphan dispatch/ingress/role work.
        // `stop()` remains the graceful path; Drop only seals + aborts.
        self.shared.token.shutdown();
        if let Some(task) = &self.network_number_task {
            task.abort();
        }
        if let Some(source) = self.source_audit.take() {
            source.close();
        }
        if let Some(requester) = self.requester.take() {
            requester.close();
        }
        if let Some(responder) = self.responder.take() {
            responder.close();
        }
        if let Some(notifications) = self.notifications.take() {
            notifications.close();
        }
        if let Some(cancel) = self.cancel_tx.take() {
            let _ = cancel.send(());
        }
        if let Some(task) = self.dispatch_task.take() {
            task.abort();
        }
        if let Some(task) = self.durable_settlement.take() {
            task.abort();
        }
        // The session's own handle, kept through stop(), goes off the
        // runtime if it is the last (#1561). Every other endpoint holder
        // (responder, source Audit, the aborted tasks' handles) lets go the
        // same way when it drops, whichever goes last.
        if let Some(db) = self.database.take() {
            drop(drop_database_off_runtime(db));
        }
    }
}

struct DispatchParts {
    inbound: mpsc::Receiver<ReceivedApdu>,
    terminal: mpsc::Receiver<ReceivedApdu>,
    policy: mpsc::Receiver<PolicyOutcome>,
    requester: Option<EndpointRequester>,
    responder: Option<Arc<EndpointResponder>>,
    notifications: Option<Arc<NotificationTransactions>>,
    coordinator: Arc<OutboundTransactionCoordinator>,
    shared: Arc<SessionShared>,
    /// The layer's network number, which the session's Number owner
    /// publishes; answers relayed with it as SNET match like direct ones.
    local_network: LocalNetworkNumber,
}

#[path = "session_dispatch.rs"]
mod dispatch;
use dispatch::{DispatchEvent, next_event};

async fn dispatch_loop(mut parts: DispatchParts, mut cancel: oneshot::Receiver<()>) -> SessionExit {
    // One consumer retains all receivers and joins; cancellation stays first.
    let mut next = 0;
    loop {
        let event = tokio::select! {
            biased;
            _ = &mut cancel => return SessionExit::Cancelled,
            event = next_event(&mut parts, &mut next) => event,
        };
        match event {
            DispatchEvent::Worker(joined) => NotificationTransactions::observe(Some(joined)),
            DispatchEvent::Inbound(received) => handle_inbound(&mut parts, received).await,
            DispatchEvent::Terminal(received) => handle_terminal(&mut parts, received).await,
            DispatchEvent::Policy(outcome) => handle_ingress_policy(&parts.shared, outcome).await,
            DispatchEvent::Closed => return SessionExit::ReceiversClosed,
        }
    }
}

async fn handle_inbound(parts: &mut DispatchParts, received: ReceivedApdu) {
    // Preserve the full envelope structurally (raw + effective group,
    // attributes, ingress identity, provenance) — no new decisions here
    // beyond role presence + responder scope.
    let _peer = inbound_canonical_peer(&received, parts.local_network.get());
    let _link_group = received.link_layer_group;
    let _is_group = received.is_group;
    let _attributes = received.data_attributes.clone();
    let _provenance = &received.provenance;
    let _ingress = received.ingress_network;
    let Some(responder) = parts.responder.as_ref() else {
        let mut counters = parts.shared.counters.lock().await;
        counters.no_server_role += 1;
        // One-use reply ownership: release a lone reply sender without
        // sending bytes, mirroring queue-admission drop semantics.
        if let Some(reply_tx) = received.reply_tx {
            drop(reply_tx);
        }
        return;
    };
    // RB-17 deferred-reply wiring (minimal correlation, session-side): when
    // the one-shot suspension arm is set and this request carries the MS/TP
    // one-use prompt reply sender, drop the sender BEFORE the responder sees
    // it. Dropping releases the MAC to send ReplyPostponed; the responder
    // then observes `reply_tx: None` and answers through the single
    // EndpointEgress → queue_npdu path, which the MAC transmits as
    // DataNotExpectingReply at the next token opportunity with the identical
    // wire invoke ID + destination. Prompt (sender present, unarmed) and
    // token-owned (sender absent) stay distinct by construction: the
    // responder never sees both for one request. No second serial owner, no
    // MAC state duplication, no transport change.
    let mut received = received;
    if !received.provenance.is_direct_peer()
        && received.direct_response.is_none()
        && received.reply_tx.is_some()
        && parts.shared.token.take_suspend()
    {
        let _ = received.reply_tx.take();
    }
    match responder.handle(received).await {
        Ok(true) | Ok(false) => {}
        Err(_) => {
            let mut counters = parts.shared.counters.lock().await;
            counters.responder_declined += 1;
        }
    }
}

async fn handle_terminal(parts: &mut DispatchParts, received: ReceivedApdu) {
    let Some(apdu) = decode_terminal(&received) else {
        let mut counters = parts.shared.counters.lock().await;
        counters.unclaimed_terminal += 1;
        if let Some(reply_tx) = received.reply_tx {
            drop(reply_tx);
        }
        return;
    };
    // Inbound server transactions never allocate here: requests are not
    // terminal traffic. A Confirmed/Unconfirmed request in this queue is a
    // classifier violation → policy-owned, no coordinator interaction.
    if matches!(
        apdu,
        Apdu::ConfirmedRequest(_) | Apdu::UnconfirmedRequest(_)
    ) {
        let mut counters = parts.shared.counters.lock().await;
        counters.unclaimed_terminal += 1;
        if let Some(reply_tx) = received.reply_tx {
            drop(reply_tx);
        }
        return;
    }
    // Exactly one shared-coordinator admit per terminal APDU (exact-once
    // terminal claim). Equal numeric IDs are unambiguous: the admitted
    // lease owner selects requester vs notification.
    let outcome = admit_once(
        &parts.coordinator,
        &received,
        parts.local_network.get(),
        &apdu,
    );
    let admission = match outcome {
        Ok(AdmissionOutcome::Admitted(admission)) => admission,
        Ok(_) => {
            let mut counters = parts.shared.counters.lock().await;
            counters.unclaimed_terminal += 1;
            if let Some(reply_tx) = received.reply_tx {
                drop(reply_tx);
            }
            return;
        }
        Err(CoordinatorError::StatePoisoned) => {
            let mut counters = parts.shared.counters.lock().await;
            counters.unclaimed_terminal += 1;
            if let Some(reply_tx) = received.reply_tx {
                drop(reply_tx);
            }
            return;
        }
    };
    if is_requester_lease(&admission) {
        let Some(requester) = parts.requester.as_ref() else {
            // Lease was admitted but has no consumer: release the exact
            // lease so exhaustion → release → reuse still holds.
            let _ = parts.coordinator.release(admission.token());
            let mut counters = parts.shared.counters.lock().await;
            counters.no_client_role += 1;
            if let Some(reply_tx) = received.reply_tx {
                drop(reply_tx);
            }
            return;
        };
        // `complete_pre_admitted` delivers via the exact token without
        // re-admitting; a `false` means the TSM no longer owns it (lost
        // consumer / stale), the coordinator lease was already completed
        // inside the TSM path on success.
        let _ = requester
            .complete_pre_admitted(admission, apdu, received)
            .await;
    } else {
        let Some(notifications) = parts.notifications.as_ref() else {
            let _ = parts.coordinator.release(admission.token());
            let mut counters = parts.shared.counters.lock().await;
            counters.no_client_role += 1;
            if let Some(reply_tx) = received.reply_tx {
                drop(reply_tx);
            }
            return;
        };
        let _ = notifications.complete_pre_admitted(admission, &apdu);
        // Notification replies carry no reply sender; drop one if present.
        if let Some(reply_tx) = received.reply_tx {
            drop(reply_tx);
        }
    }
}

async fn handle_ingress_policy(shared: &Arc<SessionShared>, outcome: PolicyOutcome) {
    // Policy-outcome ownership: the session counts every classifier outcome
    // (malformed/unsupported/route-full/route-closed) instead of dropping it.
    let mut counters = shared.counters.lock().await;
    counters.ingress_policy += 1;
    let _ = outcome.reason;
    // Policy outcomes own the full envelope including a possible one-use
    // reply sender; releasing here preserves single-consumer + one-use
    // ownership without sending bytes.
    if let Some(reply_tx) = outcome.received.reply_tx {
        drop(reply_tx);
    }
    let _ = PolicyReason::MalformedApdu;
}

#[cfg(test)]
#[path = "source_read_tests.rs"]
mod source_read_tests;

#[cfg(test)]
#[path = "registered_port_bip6_tests.rs"]
mod registered_port_bip6_tests;
#[cfg(test)]
#[path = "registered_port_lifetime_tests.rs"]
mod registered_port_lifetime_tests;
#[cfg(test)]
#[path = "registered_port_wire_tests.rs"]
mod registered_port_wire_tests;

#[cfg(test)]
#[path = "network_number_tests.rs"]
mod network_number_tests;

/// Reruns a test whose closing rebind loses the freed port to another
/// process (#1419).
#[cfg(test)]
#[path = "../tests/support/port_retry.rs"]
mod port_retry;

#[cfg(test)]
#[path = "local_network_tests.rs"]
mod local_network_tests;

#[cfg(test)]
#[path = "read_work_limit_tests.rs"]
mod read_work_limit_tests;

#[cfg(test)]
#[path = "client_pacing_tests.rs"]
mod client_pacing_tests;

#[path = "device_identity.rs"]
mod device_identity;