slither 0.3.0

//! §16.2's `Endpoint`, its builder, and `Connecting`.
//!
//! What an endpoint *is*, and the `LocalSet` it must be built inside, are in
//! [the shell module docs](super). What is here is §16.3's thin client: the
//! command-channel verbs, `connect()`'s synchronous 0-DH half, and the
//! builder that spawns the driver.

use std::cell::RefCell;
use std::future::{Future, poll_fn};
use std::net::SocketAddr;
use std::pin::Pin;
use std::rc::Rc;
use std::task::{Context, Poll};

use crate::config::Config;
use crate::core::{ConnectionId, Endpoint as CoreEndpoint, ToEndpoint};
use crate::error::ConnectError;
use crate::identity::{Identity, PublicKeyOf};

use super::connection::Connection;
use super::driver::Driver;
use super::shared::{Command, PendingOutcome, PendingSlot, Shell, now};
use super::staged::Intro;
use super::wire::Wire;

/// One endpoint: a [`Wire`], an [`Identity`], and everything §16.1's object
/// model hangs off them.
///
/// # It is a handle
///
/// The driver lives while any handle lives (§16.3). An `Endpoint` is one;
/// so is a [`Connecting`] (ruling 62) and so is a [`Connection`]. A staged
/// object is **not**, and neither is a `closed()` future.
///
/// Dropping an `Endpoint` while a `Connection` lives does **not** tear
/// anything down — that is documentation obligation #4, and it is the
/// opposite of the obvious guess.
///
/// # One connection per remote static
///
/// [`connect`](Self::connect) to a static that already has a live
/// connection, or an outbound attempt still in flight, returns
/// [`ConnectError::AlreadyConnected`]. **Reconnecting is `close()` then
/// dial**, not `connect()` again — documentation obligation #1.
pub struct Endpoint<I: Identity> {
    shell: Shell<I>,
}

impl<I: Identity> Endpoint<I> {
    /// Start building an endpoint.
    ///
    /// The builder is where §16.3's `Wire` is supplied, so an application
    /// that needs its own socket options, a dual-stack arrangement, a
    /// tunnel or a simulator installs one without forking the crate.
    pub fn builder<W: Wire>() -> EndpointBuilder<I, W> {
        EndpointBuilder::new()
    }

    /// Wait for the next introduction (§6.2, §6.3).
    ///
    /// `None` means the endpoint is closed — the driver has stopped. On a
    /// live endpoint this never resolves until a peer initiates.
    ///
    /// Costs **0 DH**: an [`Intro`] is msg1's parked bytes and its source,
    /// nothing more. The ladder starts at
    /// [`Intro::read_identity`](super::staged::Intro::read_identity).
    ///
    /// # Keep calling it — documentation obligation #6
    ///
    /// **This is a loop for the lifetime of the endpoint, not one call per
    /// connection.** §6.5: *"Every application SHOULD treat `accept()` as a
    /// loop for the lifetime of its endpoint — diallers and responders
    /// alike."* Two conditions are visible only as a later [`Intro`], and
    /// nothing else repairs them.
    ///
    /// A **lost msg2**. msg2 is never retransmitted — every retransmit is a
    /// completely fresh initiation (§5.5) — so one dropped msg2 leaves this
    /// endpoint holding a live, never-confirmed connection while the peer
    /// re-offers a fresh introduction every `RETRANSMIT_BASE` (~5 s) until
    /// `HANDSHAKE_GIVEUP` (90 s). **No error announces it**: this call
    /// *succeeded*, and the connection it produced simply never carries
    /// anything. A **restarted peer** (§6.8) is the same shape against a
    /// zombie connection that keeps running untouched.
    ///
    /// Admitting the fresh introduction **is** the replacement (§5.4),
    /// under §6.4's §16.1 guard. Against a connection this endpoint
    /// **accepted** — replacement basis `Some(t)`, the new initiation's
    /// timestamp strictly greater — the old connection's handle sees
    /// [`ConnectionLost::Replaced`](crate::error::ConnectionLost::Replaced)
    /// and the new chain completes. Against one it **dialled** the basis is
    /// `None`, so no initiation can replace it:
    /// [`AcceptError::Stale`](crate::error::AcceptError::Stale) comes back
    /// and the live connection is untouched. Documentation obligation #6 in
    /// the [crate docs](crate) states both halves.
    ///
    /// # Cancel-safety
    ///
    /// Dropping the future before it resolves takes nothing: the
    /// introduction is handed to the next caller. If it was handed over in
    /// the same instant the future was dropped, the `Intro` is dropped with
    /// it — which is §6.2's silent reject, the documented meaning of
    /// dropping a staged object, not a loss.
    pub async fn accept(&self) -> Option<Intro<I>> {
        let mut pending = None;
        poll_fn(|cx| self.poll_accept(cx, &mut pending)).await
    }

    /// The one implementation of [`accept`](Self::accept) — **ruling 229**.
    ///
    /// §16.11 lists the `Stream`/`Sink` faces and ruling 229 added
    /// `incoming` to that list, which had named seven faces and none of them
    /// the endpoint's. The reason it was missed is this function's absence:
    /// `accept()` is on ruling 53's **channel** side — §6.2 requires the DH
    /// to land on the driver task — so it was the one verb with no `poll_*`
    /// behind it, and an adapter built on the `async fn` would have had to
    /// **box and store an in-flight future**, which is precisely the cost
    /// §16.3 names as what ruling 53 exists to avoid. A face that could not
    /// be built the way the section builds every other face is a face that
    /// quietly falls off the section's list.
    ///
    /// So this is ruling 53's *"written once"* rule applied to the verb that
    /// escaped it: `accept()` above is `poll_fn` over this, and
    /// `compat::Incoming` holds the same `Option<Receiver>` this takes —
    /// a `oneshot::Receiver` is `Unpin`, so the adapter stays `Unpin` too.
    ///
    /// # `pending` is the caller's, and dropping it is §6.2's silent reject
    ///
    /// The slot holds the outstanding request between polls. Dropping it
    /// with a request in flight — dropping the `accept()` future, or the
    /// `Incoming` adapter — may drop one `Intro` that the driver handed over
    /// in that same instant, which is *"the documented meaning of dropping a
    /// staged object, not a loss"*.
    ///
    /// # The slot is cleared before `Ready`
    ///
    /// Not tidiness: `tokio::sync::oneshot::Receiver` **panics** when polled
    /// after it has completed. Clearing it is also what lets a caller poll
    /// again for the *next* introduction, which is what the `Stream` face
    /// does.
    ///
    /// `None` means the endpoint is closed, exactly as [`accept`](Self::accept)
    /// documents, and it stays `None`: a stopped driver is answered on the
    /// first branch, and a driver that stops with a request outstanding
    /// drops the sender, which resolves the receiver in error.
    pub(crate) fn poll_accept(
        &self,
        cx: &mut Context<'_>,
        pending: &mut Option<tokio::sync::oneshot::Receiver<Intro<I>>>,
    ) -> Poll<Option<Intro<I>>> {
        let rx = match pending {
            Some(rx) => rx,
            None => {
                if self.shell.driver_stopped() {
                    return Poll::Ready(None);
                }
                let (tx, rx) = tokio::sync::oneshot::channel();
                self.shell.send(Command::Accept(tx));
                pending.insert(rx)
            }
        };
        match Pin::new(rx).poll(cx) {
            Poll::Pending => Poll::Pending,
            Poll::Ready(outcome) => {
                *pending = None;
                Poll::Ready(outcome.ok())
            }
        }
    }
}

/// §16.2's `connect()`, defined **only where the suite's pattern has no
/// pre-shared key** — every [`channel!`](crate::channel) suite.
///
/// A [`channel_psk!`](crate::channel_psk) suite gets
/// [`connect_with`](Endpoint::connect_with) instead, and gets it as the
/// only dial: the bound is what makes "there is no PSK-shaped default to
/// fall into" a compile error rather than a convention (ruling 280).
impl<I: Identity> Endpoint<I>
where
    I::Suite: crate::packet::Handshake<Psk = ()>,
{
    /// Dial `remote_static` at `remote` (§5.5).
    ///
    /// # Example
    ///
    /// A first connect, whole: bind a socket, build the endpoint, dial,
    /// send one message, close.
    ///
    /// ```no_run
    /// # use slither::prelude::*;
    /// # use rand_chacha::ChaCha20Rng;
    /// # use rand_chacha::rand_core::SeedableRng;
    /// # slither::channel! { pub MySuite<P256, ChaChaPoly, Blake2b>; }
    /// # fn seeded() -> Result<ChaCha20Rng, Box<dyn std::error::Error>> {
    /// #     let mut seed = [0u8; 32];
    /// #     getrandom::fill(&mut seed)?;
    /// #     Ok(ChaCha20Rng::from_seed(seed))
    /// # }
    /// # fn main() -> Result<(), Box<dyn std::error::Error>> {
    /// # let identity: SoftwareIdentity<MySuite> = SoftwareIdentity::generate(seeded()?)?;
    /// # let peer: SoftwareIdentity<MySuite> = SoftwareIdentity::generate(seeded()?)?;
    /// // The peer's `public_static()` reached you out of band; `peer_addr`
    /// // is where to send. slither never learns a static key from the wire.
    /// # let peer_key = *peer.public_static();
    /// # let peer_addr: std::net::SocketAddr = "203.0.113.7:9000".parse()?;
    /// block_on(async move {
    ///     // A `tokio::net::UdpSocket` is a `Wire` out of the box, and
    ///     // `block_on` is the current-thread runtime plus the `LocalSet`
    ///     // the `!Send` driver needs.
    ///     let socket = tokio::net::UdpSocket::bind("0.0.0.0:0").await?;
    ///     let endpoint = Endpoint::builder()
    ///         .identity(identity)
    ///         .wire(socket)
    ///         .config(Config::new())
    ///         .build();
    ///
    ///     // Synchronous, 0 DH: this mints the pending and answers
    ///     // `AlreadyConnected` here, before any await. Awaiting the
    ///     // [`Connecting`] is what runs §6.1's handshake.
    ///     let conn = endpoint.connect(peer_addr, peer_key)?.await?;
    ///
    ///     conn.send_message(b"hello").await?;
    ///     conn.acked().await?;
    ///     conn.close(slither::constants::NO_ERROR, b"done").await;
    ///     Ok(())
    /// })
    /// # }
    /// ```
    ///
    /// **Not `async`** — §16.2 declares it so, and rulings 87 and 90 are why
    /// it can be. Ruling 87 settled the signature on the grounds that
    /// "`connect()` performs no DH — §6.1's initiator costs are paid when
    /// msg1 is built, on the driver; the verb itself only mints the
    /// pending". **Ruling 90 is the core factoring that makes that sentence
    /// true**: `core::Endpoint::mint_pending` costs 0 DH, so this verb calls
    /// it here, synchronously, and `core::Endpoint::start_attempt` spends
    /// §6.1's two on the driver where §6.2 requires them.
    ///
    /// So [`ConnectError::AlreadyConnected`] arrives **before any await**,
    /// and it arrives from the **endpoint core's own static map** — §16.1's
    /// NONE/PENDING/LIVE test read from the one authority, not from a
    /// shell-side copy of it. There is no second record of
    /// one-connection-per-static to disagree with the first.
    ///
    /// That same synchronous call is what makes ruling 50's
    /// cancellation-ordering **MUST** structural rather than a discipline:
    /// [`Connecting::drop`] retires the pending in that same map, in the
    /// same instant, so
    ///
    /// ```text
    /// drop(connecting);                    // cancels
    /// let retry = endpoint.connect(a, k)?; // succeeds, on the very next line
    /// ```
    ///
    /// works **with no advance of the clock between them** — which is the
    /// `timeout()` idiom every consumer writes.
    ///
    /// # A connection you are not using dies — S5
    ///
    /// **Do not connect ahead of need.** slither has no idle state: a
    /// session over which neither side sends dies at `DEAD_TIMEOUT` (25 s),
    /// and this is deliberate rather than a gap. Dialling at start-up so the
    /// path is "warm" for a request several minutes later gets you a dead
    /// connection and a surprising error at the moment you first try to use
    /// it.
    ///
    /// Two shapes work. **Dial when you need it** — the handshake is one
    /// round trip. Or **keep it alive by using it**: §7.5's keepalive
    /// machinery holds a session open only while there is traffic to hold
    /// open, so an application-level heartbeat is what makes a long-lived
    /// idle connection a real thing.
    ///
    /// This is the single most surprising behaviour for a new consumer, and
    /// it is the one with no code fix — the alternative is sessions that
    /// outlive their usefulness and a `!Send` driver that never quiesces.
    pub fn connect(
        &self,
        remote: SocketAddr,
        remote_static: PublicKeyOf<I>,
    ) -> Result<Connecting<I>, ConnectError> {
        self.connect_with(remote, remote_static, ())
    }
}

impl<I: Identity> Endpoint<I> {
    /// [`connect`](Endpoint::connect), naming the pre-shared key.
    ///
    /// Available on **every** suite; on an `IK` one the key is `()` and
    /// `connect` is the same call with it elided. On a
    /// [`channel_psk!`](crate::channel_psk) suite this is the *only* dial —
    /// `connect` is defined only where `Psk = ()`, so there is no
    /// PSK-shaped default to fall into (ruling 280).
    ///
    /// Every word of [`connect`](Endpoint::connect)'s documentation applies
    /// unchanged, including the two that matter most: **a connection you
    /// are not using dies**, and a `Connecting` dropped before it resolves
    /// cancels the dial synchronously.
    ///
    /// # The key is held for the dial's whole life
    ///
    /// §5.5 rebuilds msg1 on a **fresh ephemeral** every retransmit, so the
    /// PSK is re-mixed each time; it is released when the pending is, on
    /// success, give-up or cancel. It is also the key §6.6's internal
    /// tie-break completes with, should this peer's own initiation cross
    /// ours — that path has no application in the loop to ask, and the key
    /// this call named is the right one by construction.
    ///
    /// # A wrong key is not a distinguishable outcome
    ///
    /// The peer answers a bad PSK the way it answers a bad identity: msg1's
    /// tail tag fails and nothing is sent. From here that is silence, and
    /// the dial ends at [`ConnectError::TimedOut`] like any unanswered
    /// handshake. There is no "wrong PSK" error, deliberately — §18.1 gives
    /// an unauthenticated peer no oracle.
    pub fn connect_with(
        &self,
        remote: SocketAddr,
        remote_static: PublicKeyOf<I>,
        psk: crate::identity::PskOf<I>,
    ) -> Result<Connecting<I>, ConnectError> {
        let (id, core) = {
            let mut state = self.shell.state.borrow_mut();
            if state.driver_stopped {
                // Unreachable while this `Endpoint` lives — it is a handle,
                // and the driver runs while any handle does. `ConnectError`
                // deliberately has no `EndpointDropped` (ruling 62); a
                // local fault with 0 DH spent is what `Local` names.
                return Err(ConnectError::Local);
            }
            let minted = state
                .endpoint
                .mint_pending(now(), remote, remote_static.clone(), psk)?;
            // §16.4: every mutating core call is followed by a drain. This
            // one emits nothing; see `ShellState::drain_endpoint`.
            state.drain_endpoint();
            minted
        };

        let slot = Rc::new(RefCell::new(PendingSlot::new()));
        self.shell.send(Command::Connect {
            id,
            core: Box::new(core),
            remote,
            remote_static,
            slot: Rc::clone(&slot),
        });

        Ok(Connecting::new(self.shell.clone(), slot, id))
    }
}

impl<I: Identity> std::fmt::Debug for Endpoint<I> {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("Endpoint").finish_non_exhaustive()
    }
}

impl<I: Identity> Drop for Endpoint<I> {
    fn drop(&mut self) {
        self.shell.release();
    }
}

/// An outbound attempt in flight — §16.2's `connect()` future.
///
/// You get one from [`Endpoint::connect`] and you await it; the example on
/// that method is the whole shape.
///
/// # It is a handle, and dropping it cancels
///
/// **[RATIFIED 2026/08/14 — rulings 62 and 50]** A `Connecting` owns an
/// in-flight protocol attempt: a pending, its index, and §5.5's retransmit
/// train. So the driver lives while one lives, and dropping it **cancels
/// the attempt immediately** — the train stops, the pending and its index
/// go (§17.3), its dialled address leaves §6.5's hint set (§17.4), and the
/// static leaves PENDING for NONE (§5.4).
///
/// **Nothing is transmitted.** An attempt that never completed has no
/// session to close and no wire signal to send, as in §15.4's
/// endpoint-dropped row. If the peer already answered and installed a
/// half-open session, it is *not* told: it reaps it at `DEAD_TIMEOUT`
/// (25 s) in silence.
///
/// The cancellation is ordered **ahead of any endpoint verb issued after
/// the drop returns**, so an immediate redial cannot observe the corpse.
pub struct Connecting<I: Identity> {
    shell: Shell<I>,
    slot: Rc<RefCell<PendingSlot<I>>>,
    /// Minted synchronously by `mint_pending` (ruling 90), so it exists from
    /// the instant this handle does — which is what lets `Drop` retire the
    /// pending in the core rather than ask the driver to.
    id: ConnectionId,
    /// Set once this future has handed its result out, so `Drop` knows the
    /// attempt is no longer in flight.
    resolved: bool,
}

impl<I: Identity> Connecting<I> {
    fn new(shell: Shell<I>, slot: Rc<RefCell<PendingSlot<I>>>, id: ConnectionId) -> Self {
        shell.acquire();
        Self {
            shell,
            slot,
            id,
            resolved: false,
        }
    }
}

impl<I: Identity> Future for Connecting<I> {
    type Output = Result<Connection<I::Suite>, ConnectError>;

    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
        let this = self.get_mut();
        let mut slot = this.slot.borrow_mut();
        match std::mem::replace(&mut slot.outcome, PendingOutcome::Waiting) {
            PendingOutcome::Waiting => {
                slot.waker = Some(cx.waker().clone());
                Poll::Pending
            }
            PendingOutcome::Ready(connection) => {
                this.resolved = true;
                Poll::Ready(Ok(connection))
            }
            PendingOutcome::Failed(error) => {
                this.resolved = true;
                Poll::Ready(Err(error))
            }
        }
    }
}

impl<I: Identity> std::fmt::Debug for Connecting<I> {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("Connecting").finish_non_exhaustive()
    }
}

impl<I: Identity> Drop for Connecting<I> {
    fn drop(&mut self) {
        // Only a *waiting* attempt is in flight. One that already resolved
        // — including one resolved into the slot and never polled — is not,
        // and cancelling it would free a static its live `Connection` still
        // holds. An unpolled `Ready` drops its `Connection` with the slot,
        // which is the ordinary last-handle drop.
        let in_flight =
            !self.resolved && matches!(self.slot.borrow().outcome, PendingOutcome::Waiting);

        if in_flight {
            // Ruling 50's MUST, and after ruling 90 it is discharged in the
            // **core's own map**, synchronously, right here.
            //
            // §16.4 lists no cancel verb; `Retired` is the event whose
            // documented effects are exactly cancellation's — it stops
            // §5.5's train, frees the pending index (§17.3), takes the
            // dialled address out of §6.5's hint set (§17.4), releases the
            // §17.1 pin, and **returns the static to NONE**. It costs 0 DH,
            // which is the whole reason a handle may call it: §6.2 puts DH
            // on the driver, and there is none here.
            //
            // `our_index: 0` is provably safe: `IndexTables::mint` draws a
            // random **nonzero** `u32`, so no live route is ever keyed on
            // it, and a `connect()` that never installed a session has no
            // session index to name.
            //
            // Doing it here rather than on the driver is what makes the
            // ordering **structural**: an immediate redial calls
            // `mint_pending` on the very map this line just released, with
            // no clock advance and no driver turn in between. A version
            // that only queued the command would answer that redial
            // `AlreadyConnected`.
            {
                let mut state = self.shell.state.borrow_mut();
                state.endpoint.handle_connection_event(
                    now(),
                    self.id,
                    ToEndpoint::Retired { our_index: 0 },
                );
                // §16.4's drain contract; this call emits nothing.
                state.drain_endpoint();
            }
            // What is left for the driver is its own bookkeeping: the
            // `ConnRecord`, and the wake that lets it recompute deadlines.
            self.shell.send(Command::Cancel(self.id));
        }

        self.shell.release();
    }
}

/// Builds an [`Endpoint`] and spawns its driver (§16.2).
///
/// # A `LocalSet` is required
///
/// [`build`](Self::build) calls `tokio::task::spawn_local`, which **panics
/// outside a `LocalSet`** — with slither's own message, naming
/// [`block_on`](crate::compat::block_on) and `LocalSet::run_until` (see
/// [`build`](Self::build)'s panics section). That is §16.3's architecture,
/// not an implementation detail: the driver is a single `!Send` actor
/// because a DH provider is not required to be `Send` — a hardware-backed
/// static key is the case the seam exists for — and a `Wire` is not
/// required to be `Send` either.
///
/// ```no_run
/// # use slither::prelude::*;
/// # async fn doc<I, W>(identity: I, wire: W)
/// # where
/// #     I: Identity + 'static,
/// #     W: Wire + 'static,
/// # {
/// let local = tokio::task::LocalSet::new();
/// local
///     .run_until(async move {
///         let endpoint = Endpoint::builder()
///             .identity(identity)
///             .wire(wire)
///             .build();
///         // … dial and accept through `endpoint` …
///         drop(endpoint);
///     })
///     .await;
/// # }
/// ```
pub struct EndpointBuilder<I: Identity, W: Wire> {
    identity: Option<I>,
    wire: Option<W>,
    config: Config,
    rng_seed: Option<[u8; 32]>,
}

impl<I: Identity, W: Wire> EndpointBuilder<I, W> {
    fn new() -> Self {
        Self {
            identity: None,
            wire: None,
            config: Config::new(),
            rng_seed: None,
        }
    }

    /// The static-key seam (§2.4). Required.
    #[must_use]
    pub fn identity(mut self, identity: I) -> Self {
        self.identity = Some(identity);
        self
    }

    /// The datagram substrate (§16.3). Required.
    ///
    /// A `tokio::net::UdpSocket` is a [`Wire`] out of the box; so is
    /// `testutil::FlakyWire`, which is what makes §16.10's kernel-free
    /// drivability work.
    #[must_use]
    pub fn wire(mut self, wire: W) -> Self {
        self.wire = Some(wire);
        self
    }

    /// Endpoint configuration. Defaults to [`Config::new`].
    #[must_use]
    pub fn config(mut self, config: Config) -> Self {
        self.config = config;
        self
    }

    /// Seed §16.6's one endpoint RNG explicitly.
    ///
    /// # This is a test-only facility
    ///
    /// §16.6: session and pending indices **MUST** be unpredictable to an
    /// off-path observer — index unpredictability is load-bearing for
    /// §5.5's on-path-only completion spend and §15.2's authenticated-only
    /// linger reply. A caller-chosen seed is therefore **security-relevant**
    /// and is documented as such here rather than being quietly available:
    /// with it, one root seed reproduces every index, every jitter draw and
    /// every connection sub-seed. Without it the endpoint seeds from OS
    /// entropy, which is what a production endpoint must do.
    #[must_use]
    pub fn rng_seed(mut self, seed: [u8; 32]) -> Self {
        self.rng_seed = Some(seed);
        self
    }

    /// Spawn the driver and return the endpoint (§16.3).
    ///
    /// Named `build` for the builder convention, but note what it does:
    /// **it spawns a task**. The driver is the endpoint — there is no
    /// separate `run()` for a caller to forget.
    ///
    /// # Panics
    ///
    /// If [`identity`](Self::identity) or [`wire`](Self::wire) was not
    /// supplied, or if this is called outside a `tokio::task::LocalSet` —
    /// both are programming errors rather than runtime conditions, and
    /// neither is recoverable at the call site.
    ///
    /// The `LocalSet` case panics with **slither's own message**, which
    /// names [`block_on`](crate::compat::block_on) and
    /// `tokio::task::LocalSet::run_until` as the two fixes: `spawn_local`'s
    /// own wording names tokio and no slither symbol, and a consumer who has
    /// just written `#[tokio::main]` needs to be told which of the two to
    /// reach for. Under `panic = "abort"` the guard cannot intercept —
    /// `catch_unwind` does not catch an aborting panic — so that profile
    /// still shows tokio's raw *"`spawn_local` called from outside of a
    /// `task::LocalSet`"*.
    ///
    /// Seeding from OS entropy can also fail; `getrandom` panics there, and
    /// an endpoint that silently continued with a predictable seed would
    /// violate §16.6.
    ///
    /// The `'static` bounds are the spawned task's, not a `Send`
    /// requirement: `spawn_local` needs the future to outlive any borrow,
    /// and **no `Send` bound appears anywhere on this path** (§16.3).
    #[must_use]
    pub fn build(self) -> Endpoint<I>
    where
        I: 'static,
        W: 'static,
    {
        let identity = self
            .identity
            .expect("Endpoint::builder() requires an identity");
        let wire = self.wire.expect("Endpoint::builder() requires a wire");
        let rng_seed = self.rng_seed.unwrap_or_else(|| {
            let mut seed = [0u8; 32];
            getrandom::fill(&mut seed).expect("OS entropy for the endpoint RNG (§16.6)");
            seed
        });

        let core = CoreEndpoint::new(super::shared::now(), self.config, identity, rng_seed);
        let (shell, commands) = Shell::new(core);
        let endpoint = Endpoint {
            shell: shell.clone(),
        };
        // The `Endpoint` is the first handle; count it before the driver
        // can observe a zero.
        endpoint.shell.acquire();

        // **[RATIFIED 2026/08/19 — round 44 §8(ii)]** `spawn_local`'s own
        // panic names tokio and no slither symbol, so the consumer who has
        // just written `#[tokio::main]` is told nothing about what to do.
        // There is no public tokio predicate for "am I inside a `LocalSet`"
        // — a `LocalSet` runs on a multi-thread runtime via `run_until`, so
        // a runtime-flavour check would falsely panic on a valid
        // configuration — which leaves catching the panic as the only way to
        // answer the question that was actually asked.
        let driver = Driver::new(wire, shell, commands).run();
        let spawned = std::panic::catch_unwind(std::panic::AssertUnwindSafe(move || {
            tokio::task::spawn_local(driver);
        }));
        if spawned.is_err() {
            panic!(
                "slither: Endpoint::builder()…build() was called outside a \
                 `tokio::task::LocalSet`. The driver is a single `!Send` task \
                 (a DH provider and a Wire are not required to be `Send`), so \
                 it is spawned with `tokio::task::spawn_local`, which needs a \
                 current-thread runtime with a `LocalSet` — `#[tokio::main]` \
                 alone is not one. Wrap your code in \
                 `slither::prelude::block_on(async {{ … }})`, or build the \
                 endpoint inside \
                 `tokio::task::LocalSet::new().run_until(…)`."
            );
        }
        endpoint
    }
}

impl<I: Identity, W: Wire> Default for EndpointBuilder<I, W> {
    fn default() -> Self {
        Self::new()
    }
}

impl<I: Identity, W: Wire> std::fmt::Debug for EndpointBuilder<I, W> {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("EndpointBuilder")
            .field("identity", &self.identity.is_some())
            .field("wire", &self.wire.is_some())
            .field("seeded", &self.rng_seed.is_some())
            .finish_non_exhaustive()
    }
}

#[cfg(test)]
mod tests {
    //! The `LocalSet` guard's pin — round 44 §8(ii).
    //!
    //! Not a protocol test: it asserts that the *diagnostic* a consumer
    //! meets on the first mistake they can make is slither's own, and names
    //! the two ways out.

    use rand_chacha::ChaCha20Rng;
    use rand_chacha::rand_core::SeedableRng;

    use super::Endpoint;
    use crate::identity::SoftwareIdentity;
    use crate::packet::ReferenceSuite;

    /// `build()` on a plain current-thread runtime — a runtime, but no
    /// `LocalSet`, which is exactly what `#[tokio::main]` hands a consumer —
    /// panics with **slither's** message rather than tokio's.
    ///
    /// The three `contains` assertions are what separate the guard from its
    /// absence: tokio's own text is *"`spawn_local` called from outside of a
    /// `task::LocalSet` or `runtime::LocalRuntime`"*, which contains none of
    /// the three strings below. Deleting the guard, or dropping any of
    /// `slither::prelude::block_on` / `tokio::task::LocalSet` /
    /// `Endpoint::builder` from its wording, turns this red.
    #[test]
    fn build_outside_a_localset_panics_with_slithers_own_message() {
        let rt = tokio::runtime::Builder::new_current_thread()
            .enable_all()
            .build()
            .expect("a current-thread runtime");

        let payload = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
            rt.block_on(async {
                let identity: SoftwareIdentity<ReferenceSuite> =
                    SoftwareIdentity::generate(ChaCha20Rng::from_seed([0x44; 32]))
                        .expect("generate an identity");
                let socket = tokio::net::UdpSocket::bind("127.0.0.1:0")
                    .await
                    .expect("bind a socket");
                let _endpoint = Endpoint::builder().identity(identity).wire(socket).build();
            });
        }))
        .expect_err("build() outside a LocalSet must panic");

        let message = payload
            .downcast_ref::<String>()
            .map(String::as_str)
            .or_else(|| payload.downcast_ref::<&'static str>().copied())
            .expect("a string panic payload");

        assert!(
            message.contains("slither::prelude::block_on"),
            "the guard must name the fix; got: {message}"
        );
        assert!(
            message.contains("tokio::task::LocalSet"),
            "the guard must name what is missing; got: {message}"
        );
        assert!(
            message.contains("Endpoint::builder"),
            "the guard must name the call site; got: {message}"
        );
    }
}