//! # TapEngine — per-session WSS task supervisor (spec §4.3, §5.1)
//!
//! Spawned by `session_map::drive_all_sessions` when the poll task observes
//! `channel.state == Connected && channel.tap.is_none()` (spec §5.1 step 3).
//! Owns the WSS connection lifecycle + the bounded-backoff reconnect policy.
//!
//! # Why this is NOT the step-4 forbidden "dedicated timing thread"
//!
//! ARCHITECTURE.md mandates "dedicated timing threads, not the shared
//! tokio pool" for the *timed media work* (the 20 ms loop). The TapEngine
//! is cold-path network I/O — connect_async, ws.send, ws.next — on tokio's
//! shared pool. Slice-1 §3.4's scoped deviation (media loop on tokio) is
//! unchanged; adding a cold-path I/O supervisor doesn't widen it.
//!
//! # Return shape
//!
//! `spawn_tap_engine` returns the tuple `(TapAudioPipe, TapConn)`. The
//! `TapAudioPipe` (the seam object) is wired into `RtcSession` via
//! `RtcSession::with_pipe(...)`; its mpsc ends are MOVED into the pipe.
//! `TapConn` holds only the engine-control handles (`close_tx`, `join`,
//! `metrics`) — the mpsc pairs are not split between the two (the earlier
//! brief sketch had `TapConn` carrying them; the structural review in
//! Task 7's plan corrected this so the pipe is the single owner of the
//! channel ends the loop_driver touches indirectly through `AudioSource`/
//! `AudioSink`).

use std::sync::Arc;
use std::sync::atomic::Ordering;
use std::time::Duration;

use futures_util::FutureExt;
use rutster_call_model::ChannelId;
use rutster_tap::tap_client::{
    FunctionCallEvent, FunctionCallOutputEvent, TapClientError, run_tap_client,
};
use rutster_tap::{TapAudioPipe, TapMetrics};
use tokio::sync::{Mutex, mpsc, oneshot};
use tokio::task::JoinHandle;
use tokio_tungstenite::tungstenite::client::IntoClientRequest;
use tracing::{info, warn};
use url::Url;

use crate::tool_registry::{HangupTool, ToolRegistry};

/// Capacity for the two mpsc channels between TapAudioPipe and TapClient.
/// Large enough that a slow brain tick doesn't drop on every cycle;
/// small enough that a runaway brain doesn't accumulate seconds of audio.
const TAP_MPSC_CAPACITY: usize = 32;

/// Connect timeout — bounded dial per spec §4.3 step 1 (2 s).
const CONNECT_TIMEOUT: Duration = Duration::from_secs(2);

/// The cold-path state the binary holds per session (looked up by ChannelId).
///
/// Holds ONLY the engine-control handles — the mpsc ends that drive
/// `TapAudioPipe` are owned by the `TapAudioPipe` itself (returned alongside
/// `TapConn` from `spawn_tap_engine`). This split keeps `TapConn` cheap
/// to store in a `DashMap<ChannelId, TapConn>` even though the actual
/// data flow happens through the `TapAudioPipe` wired into `RtcSession`.
///
/// # slice-2 §5.3 step 4 — `flush_rx` side-channel
///
/// On brain disconnect / WS error / graceful close, the engine task
/// signals a playout-ring flush by `try_send(())` on `flush_tx`. The
/// binary's poll task drains `flush_rx` (after each `run_poll_once`)
/// and calls `RtcSession::clear_playout_ring()`. This drops stale brain
/// audio queued before the disconnect — preventing "stale bleed-through"
/// at reconnect. `Option<Receiver>` because it's only `Some` after the
/// engine spawns; on a freshly-initialized `TapConn` it's `None` (no
/// flush Rx has been plumbed).
pub struct TapConn {
    /// Close signal — fired on `Channel::Closing`. `oneshot::Sender` is the
    /// sending half; dropping it without `send(())` would also close the
    /// receiver (`Err(Closed)` on the await), but an explicit `send(())`
    /// is the documented teardown trigger (spec §5.1 step 5).
    pub close_tx: oneshot::Sender<()>,
    /// Task handle — abort on session teardown to guarantee the engine
    /// task stops even if the close signal is stuck in a `select!` arm.
    /// Per AGENTS.md: `abort()` is the safety net, `close_tx` is the gentle path.
    pub join: JoinHandle<()>,
    /// Shared metrics — also held by `TapAudioPipe` + `TapClient`. Storing
    /// here lets the binary read a `MetricsSnapshot` for log/CDR emission
    /// on teardown (spec §3.8 + the eventual CDR/ringbus emitter).
    pub metrics: Arc<TapMetrics>,
    /// Flush side-channel — drained by the binary's poll task after each
    /// `run_poll_once` (slice-2 spec §5.3 step 4). `Some` only on conns
    /// returned from `spawn_tap_engine`; the engine task owns the paired
    /// `flush_tx` and signals a flush after each failed pump cycle.
    pub flush_rx: Option<mpsc::Receiver<()>>,
    /// slice-3 §5.2 tool-call side-channel (brain → core): the binary's
    /// poll task drains `function_call` events from here + dispatches via
    /// [`tool_registry`]. Engine owns the paired `tx_function_call` Sender
    /// moved into `run_tap_client`. `Option` to keep the type constructible
    /// in tests that don't plumb it; always `Some` on conns returned from
    /// `spawn_tap_engine` (matches `flush_rx`'s posture).
    pub rx_function_call: Option<mpsc::Receiver<FunctionCallEvent>>,
    /// slice-3 §5.2 tool-call side-channel (core → brain): the binary's
    /// poll task writes `function_call_output` events here (one per
    /// `ToolRegistry::dispatch` result). The engine owns the paired
    /// `rx_function_call_output` Receiver moved into `run_tap_client` so
    /// the engine's pump loop forwards each event as a tap WS frame.
    pub tx_function_call_output: Option<mpsc::Sender<FunctionCallOutputEvent>>,
    /// slice-3 §6 — per-channel tool registry. One registry per active
    /// session (spec §6.2 "keyed by ChannelId"); `HangupTool` is the only
    /// tool wired in slice-3 (§6.3). `Arc<Mutex<_>>` because the binary's
    /// poll task mutates the registry (dispatch) while the TapConn is
    /// shared across the poll task + the routes layer (potential future
    /// `tools.update` forwarding). `Mutex` (not `RwLock`) because dispatch
    /// is effectively exclusive — there's no read-heavy workload to
    /// optimize.
    pub tool_registry: Arc<Mutex<ToolRegistry>>,
}

/// Tear down one session's tap engine WITHOUT blocking the caller.
///
/// # Why this must never run inline on the media thread (review M7)
///
/// The old Delete arm did `tokio_handle.block_on(timeout(750ms, join))`
/// inside the tick loop: one teardown with an unresponsive brain froze
/// the 20 ms loop for EVERY live call on the node (~37 missed frames),
/// and batched Deletes stacked sequentially. Teardown is brain I/O —
/// exactly what the tick loop must never wait on. Same discipline as the
/// tap pipe's try_send posture.
pub fn spawn_tap_teardown(tokio_handle: &tokio::runtime::Handle, id: ChannelId, mut conn: TapConn) {
    tokio_handle.spawn(async move {
        // Gentle path first (AGENTS.md: close_tx is the documented
        // trigger, abort is the safety net).
        let _ = conn.close_tx.send(());
        match tokio::time::timeout(Duration::from_millis(750), &mut conn.join).await {
            Ok(Ok(())) => {
                info!(channel_id = %id, "tap engine torn down (graceful)");
            }
            _ => {
                conn.join.abort();
                info!(channel_id = %id, "tap engine torn down (abort after timeout)");
            }
        }
    });
}

/// Spawn the TapEngine task for one session. Dials `tap_url`, runs the pump
/// loop, reconnects with bounded backoff on failure (spec §4.3, §5.2).
///
/// Returns BOTH the `TapAudioPipe` (the seam object to wire into
/// `RtcSession` via `RtcSession::with_pipe(pipe)`) AND the `TapConn`
/// (the engine-control handle to store in the binary's tap registry for
/// teardown).
///
/// # Why a tuple return (not a single struct holding both?)
///
/// The two values have different lifetimes + consumers: `TapAudioPipe`
/// is consumed by `RtcSession::with_pipe` immediately and lives behind
/// `RtcSession.pipe`; `TapConn` lives in the binary's tap registry and
/// outlives `TapAudioPipe` (it's the teardown handle). Coupling them in a
/// single struct would force the registry to also own the pipe, splitting
/// ownership of `RtcSession`'s internals across two modules — exactly the
/// pattern the slice-2 plan's structural review warned against.
pub fn spawn_tap_engine(
    session_id: ChannelId,
    tap_url: Url,
    app_state: crate::session_map::AppState,
    advisory_tx: mpsc::Sender<rutster_media::AdvisoryEvent>,
) -> (TapAudioPipe, TapConn) {
    // Two mpsc channels. The naming convention is "from the engine's POV":
    //  - `tx_pcm_in`/`rx_pcm_in`: peer PCM flowing INTO the engine (sink side
    //    of TapAudioPipe calls `tx_pcm_in.try_send(frame)`; engine task owns
    //    `rx_pcm_in` and feeds it to run_tap_client).
    //  - `tx_audio_out`/`rx_audio_out`: brain PCM flowing OUT of the engine
    //    (engine task owns `tx_audio_out` and feeds it to run_tap_client;
    //    TapAudioPipe holds `rx_audio_out` and drains it into the playout
    //    ring on each `next_pcm_frame` call).
    let (tx_pcm_in, rx_pcm_in) = mpsc::channel(TAP_MPSC_CAPACITY);
    let (tx_audio_out, rx_audio_out) = mpsc::channel(TAP_MPSC_CAPACITY);
    let (close_tx, close_rx) = oneshot::channel::<()>();
    // slice-2 §5.3 step 4 flush side-channel: empty payload, the existence
    // of the signal IS the message. Capacity 8 is comfortably larger than
    // the engine's per-reconnect backoff cadence; `try_send` is non-blocking
    // so a slow-draining `flush_rx` (poll task busy) doesn't stall the
    // engine task. The signal is naturally idempotent (multiple pending
    // flush signals collapse to one `clear_playout_ring` call from the
    // binary side).
    let (flush_tx, flush_rx) = mpsc::channel::<()>(8);
    // slice-3 §5.2 tool-call side-channels. Same capacity-shape rationale
    // as `flush_tx`: `try_send`-safe on both ends, idempotent drain. One
    // extra pair of mpsc halves vs slice-2 — same cycle in the binary's
    // poll task (drain alongside `flush_rx`), same pump-arm in run_tap_client.
    let (tx_function_call, rx_function_call) =
        mpsc::channel::<FunctionCallEvent>(TAP_MPSC_CAPACITY);
    let (tx_function_call_output, rx_function_call_output) =
        mpsc::channel::<FunctionCallOutputEvent>(TAP_MPSC_CAPACITY);
    let metrics = TapMetrics::new();

    // slice-3 §6.2: per-channel tool registry. The engine constructs it
    // (one per active session — spec "keyed by ChannelId, one registry per
    // active channel"). `HangupTool` is the only wired tool in slice-3
    // (§6.3); other tool names reply `not_implemented` via dispatch.
    // The registry is `Arc<Mutex<ToolRegistry>>` so the binary's poll task
    // can `dispatch` on it while the TapConn is shared (the future
    // `tools.update`-forwarding path will also share this handle).
    let mut registry = ToolRegistry::new();
    registry.register(Box::new(HangupTool::new(app_state, session_id)));
    let tool_registry = Arc::new(Mutex::new(registry));

    // Clone metrics three ways: the engine task, the TapAudioPipe, and
    // the TapConn handle each hold their own `Arc<TapMetrics>` refcount.
    // All three views must observe the same counters — `Arc` is the
    // canonical Rust pattern for shared-ownership shared-mutation (the
    // counters live behind `AtomicU64` inside `TapMetrics`, so the
    // shared mutation is sound without a `Mutex`).
    let metrics_for_pipe = metrics.clone();
    let metrics_for_conn = metrics.clone();
    let tool_registry_for_task = tool_registry.clone();

    let join = tokio::spawn(async move {
        run_engine_loop(
            session_id,
            tap_url,
            rx_pcm_in,
            tx_audio_out,
            close_rx,
            flush_tx,
            tx_function_call,
            rx_function_call_output,
            advisory_tx,
            metrics,
        )
        .await;
    });

    // The pipe owns `tx_pcm_in` (drains peer PCM, sends to engine via mpsc)
    // and `rx_audio_out` (drains engine's audio_out frames → playout ring).
    let pipe = TapAudioPipe::new(tx_pcm_in, rx_audio_out, metrics_for_pipe);

    let conn = TapConn {
        close_tx,
        join,
        metrics: metrics_for_conn,
        flush_rx: Some(flush_rx),
        rx_function_call: Some(rx_function_call),
        tx_function_call_output: Some(tx_function_call_output),
        tool_registry: tool_registry_for_task,
    };

    (pipe, conn)
}

/// The engine's main loop: connect → handshake → pump, with bounded
/// exponential backoff (250 ms → 500 ms → 1 s → 2 s → cap at 5 s) on
/// any non-close failure. The `close` oneshot is shared across reconnect
/// attempts — one close signal per session, no fresh oneshot per dial.
///
/// # Deviation from the brief's verbatim sketch (Note A in plan)
///
/// The brief's `run_engine_loop` body had dead-code `let _ = pump_close;`
/// and convoluted `connect_async` future plumbing inside a `timeout(...)`
/// that wasn't actually awaited. The contract is what matters — connect,
/// run the pump, backoff, retry — so we use the cleaner multiply-select
/// pattern the plan's Note A spelled out: a single `tokio::select!` over
/// (a) the close receiver and (b) `connect_brain`, then a second select
/// over the close receiver and the backoff sleep.
//
// clippy::too_many_arguments: slice-3 §5.2 added two more mpsc halves to
// slice-2's engine-loop signature (the tool-call side-channel ends). Same
// rationale as `run_tap_client` above: each arg is a distinct channel
// end with a distinct lifetime owner. Wrapping them in a struct would
// obscure the channel-pair structure. Suppress per AGENTS.md's documented
// inline-rationale exception.
#[allow(clippy::too_many_arguments)]
async fn run_engine_loop(
    session_id: ChannelId,
    tap_url: Url,
    mut rx_pcm_in: mpsc::Receiver<rutster_media::PcmFrame>,
    tx_audio_out: mpsc::Sender<rutster_media::PcmFrame>,
    mut close: oneshot::Receiver<()>,
    flush_tx: mpsc::Sender<()>,
    tx_function_call: mpsc::Sender<FunctionCallEvent>,
    mut rx_function_call_output: mpsc::Receiver<FunctionCallOutputEvent>,
    tx_advisory: mpsc::Sender<rutster_media::AdvisoryEvent>,
    metrics: Arc<TapMetrics>,
) {
    let mut backoff = Backoff::default();

    loop {
        // === Step 1: dial the brain, with the close signal racing the
        // connect future. `biased` makes close the priority arm — if the
        // binary tears the session down mid-dial we exit immediately
        // rather than waiting on connect_async to time out or succeed. ===
        let ws_result = tokio::select! {
            biased;
            _ = &mut close => {
                info!(%session_id, "close signal during connect; exiting");
                return;
            }
            // Boxed + pinned via `.map`/`boxed()` so the select arms share
            // a common Future output type (connect_async returns a different
            // concrete type than the close future; without boxing, select!
            // can't unify them on stable Rust).
            ws_result = connect_brain(&tap_url).boxed() => ws_result,
        };

        let should_flush = match ws_result {
            Ok(ws) => {
                // Successful connect → reset backoff + run the pump loop.
                backoff.reset();
                info!(%session_id, %tap_url, "brain WS connected; running pump");

                // === Step 2: run the pump loop until close/error. ===
                // `close` is a shared `&mut oneshot::Receiver` — same
                // signal across reconnect attempts (Task 4's design).
                // The slice-3 §5.2 tool-call side-channels are also shared
                // across reconnect attempts — a brain reconnect mid-call
                // must keep the binary's tool-registry dispatch flowing
                // (a function_call the brain proposed right before it
                // dropped should still get a function_call_output reply
                // on the next connect).
                let pump_result = run_tap_client(
                    ws,
                    session_id,
                    &mut rx_pcm_in,
                    tx_audio_out.clone(),
                    tx_function_call.clone(),
                    &mut rx_function_call_output,
                    tx_advisory.clone(),
                    metrics.clone(),
                    &mut close,
                )
                .await;

                match pump_result {
                    Ok(()) => {
                        // Brain WS stream ended cleanly (e.g. brain sent bye).
                        // Per spec §4.3 step 4, we still backoff + retry:
                        // the brain may have restarted. The disconnect
                        // invalidates any playout already buffered — signal
                        // a flush (spec §5.3 step 4).
                        info!(%session_id, "tap client closed gracefully; backing off");
                        true
                    }
                    Err(TapClientError::Closed) => {
                        // Close signal from the binary (Channel::Closing).
                        // `run_tap_client` already sent `session_end` + did
                        // the bounded bye handshake + closed the WS; we're
                        // done — do NOT retry or flush (we're tearing down).
                        info!(%session_id, "tap client closed via core signal; exiting");
                        return;
                    }
                    Err(e) => {
                        // WS/protocol error → backoff + retry per spec §4.3.
                        // The disconnect invalidates the playout ring —
                        // flush before backoff (spec §5.3 step 4).
                        warn!(error = ?e, %session_id, "tap client error; backing off");
                        true
                    }
                }
            }
            Err(e) => {
                warn!(error = ?e, %session_id, "tap connect failed; backing off");
                // Failed dial → no live WS ever existed this cycle, so the
                // playout ring has nothing new to flush. Skip the signal
                // (avoids spurious flushes on every retry-attempt-fails
                // cycle of a down brain). The previous successful session's
                // flush was already signaled on its disconnect.
                false
            }
        };

        // === Step 3: signal flush + backoff before retry (spec §4.3, §5.2,
        // §5.3 step 4). ===
        if should_flush {
            // Non-blocking send: if the channel is full (binary's poll task
            // hasn't drained yet), the next pump-cycle's flush signal will
            // retry. Idempotent — multiple pending signals collapse to one
            // `clear_playout_ring` call on the binary side.
            let _ = flush_tx.try_send(());
        }
        metrics.reconnect_attempts.fetch_add(1, Ordering::Relaxed);
        let delay = backoff.next_delay();
        tokio::select! {
            biased;
            _ = &mut close => {
                info!(%session_id, "close signal during backoff; exiting");
                return;
            }
            _ = tokio::time::sleep(delay) => {}
        }
    }
}

/// Dial the brain. Returns the live `WebSocketStream` (no TLS — slice-2 is
/// `ws://` loopback only per spec §4.4; `wss://` is rejected at
/// `POST /v1/sessions` time before we ever get here).
///
/// `connect_async` returns `WebSocketStream<MaybeTlsStream<TcpStream>>` for
/// `ws://` URLs (the `MaybeTlsStream` resolves to the no-TLS variant). The
/// concrete type is `tokio_tungstenite::WebSocketStream<tokio_tungstenite::MaybeTlsStream<tokio::net::TcpStream>>`.
/// We surface that explicit type rather than `impl Trait` because
/// `run_tap_client` is generic over `T: AsyncRead + AsyncWrite + Unpin`,
/// so a concrete type annotates + documents the wire shape for learners
/// without obscuring it behind `impl`.
///
/// # Why `IntoClientRequest` (not just `connect_async(&url)`)
///
/// `connect_async` accepts anything that implements `IntoClientRequest`.
/// `&Url` does, but going through `into_client_request()` first surfaces
/// any URL-shape error before the network round-trip — easier to attribute
/// (URL parse failure is a config error, not a network error).
async fn connect_brain(
    tap_url: &Url,
) -> Result<
    tokio_tungstenite::WebSocketStream<tokio_tungstenite::MaybeTlsStream<tokio::net::TcpStream>>,
    TapClientError,
> {
    let request = tap_url
        .as_str()
        .into_client_request()
        .map_err(TapClientError::Ws)?;
    // Bounded dial — `tokio::time::timeout` returns `Result<Ok(inner),
    // Elapsed>`; map the elapsed to a HandshakeTimeout-shaped Ws error so
    // the engine loop's backoff path sees it through the same `Err(e)`
    // arm as a true WS failure (one code path, two causes).
    let connect_fut = tokio_tungstenite::connect_async(request);
    let ws_result = tokio::time::timeout(CONNECT_TIMEOUT, connect_fut)
        .await
        .map_err(|_| TapClientError::Ws(tokio_tungstenite::tungstenite::Error::AlreadyClosed))?;
    let (ws, _response) = ws_result?;
    Ok(ws)
}

/// Bounded exponential backoff (spec §4.3, §5.2):
/// 250 ms → 500 ms → 1 s → 2 s → cap at 5 s. Infinite retries.
///
/// # Why not `std::time::Duration::saturating_mul`?
///
/// `Duration * 2` panics on overflow (Rust 1.x); `.min(cap)` keeps us
/// bounded. The explicit cap at 5 s means even after 1e6 retries the
/// delay stays at 5 s — important for the infinite-retry posture (spec
/// §5.2: "the brain WILL come back; the call SHOULD outlast the brain
/// outage").
struct Backoff {
    count: u32,
    current: Duration,
}

impl Default for Backoff {
    fn default() -> Self {
        Self {
            count: 0,
            current: Duration::from_millis(250),
        }
    }
}

impl Backoff {
    /// Return the current delay and advance the backoff state.
    ///
    /// Matches the spec §4.3 step 4 enumeration exactly:
    /// 250ms → 500ms → 1s → 2s → 5s cap (forever). The "cap at 5s" line
    /// means jump directly to 5s after the 2s step — NOT doubling to 4s
    /// first and then clamping (that would inject a 4s step the spec
    /// doesn't enumerate). The conditional below skips the intermediate
    /// 4s value that naïve `(current * 2).min(5s)` would produce.
    fn next_delay(&mut self) -> Duration {
        let d = self.current;
        // Below the 2s step: keep doubling (250 → 500 → 1s → 2s).
        // At or above 2s: jump straight to the 5s cap (no 4s step).
        self.current = if self.current < Duration::from_secs(2) {
            self.current * 2
        } else {
            Duration::from_secs(5)
        };
        self.count += 1;
        d
    }
    /// Reset after a successful connect + handshake.
    fn reset(&mut self) {
        self.count = 0;
        self.current = Duration::from_millis(250);
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use rutster_media::AudioSource;

    #[test]
    fn spawn_tap_teardown_returns_immediately_and_aborts_stuck_engine() {
        let rt = tokio::runtime::Runtime::new().unwrap();
        // A "stuck brain": the engine task never finishes on its own. Its
        // guard's Drop fires on abort — our proof the teardown completed.
        struct DropGuard(std::sync::mpsc::Sender<()>);
        impl Drop for DropGuard {
            fn drop(&mut self) {
                let _ = self.0.send(());
            }
        }
        let (dropped_tx, dropped_rx) = std::sync::mpsc::channel();
        let join = rt.spawn(async move {
            let _guard = DropGuard(dropped_tx);
            std::future::pending::<()>().await;
        });
        let (close_tx, _close_rx) = oneshot::channel();
        let conn = TapConn {
            close_tx,
            join,
            metrics: TapMetrics::new(),
            flush_rx: None,
            rx_function_call: None,
            tx_function_call_output: None,
            tool_registry: Arc::new(Mutex::new(ToolRegistry::default())),
        };

        let started = std::time::Instant::now();
        spawn_tap_teardown(rt.handle(), rutster_call_model::ChannelId::new(), conn);
        // The whole point (review M7): the caller — the media TICK LOOP —
        // must not wait out the 750ms brain timeout.
        assert!(
            started.elapsed() < Duration::from_millis(100),
            "teardown must not block the caller"
        );
        // …but the stuck engine task must still get aborted after the cap.
        dropped_rx
            .recv_timeout(Duration::from_secs(2))
            .expect("stuck engine task was aborted (guard dropped)");
    }

    #[test]
    fn backoff_doubles_until_cap() {
        let mut b = Backoff::default();
        // Spec §4.3 step 4 + §5.2 enumeration:
        //   250ms → 500ms → 1s → 2s → 5s cap (no intermediate 4s step from
        //   naïve doubling). Infinite retries at the cap.
        assert_eq!(b.next_delay(), Duration::from_millis(250)); // step 1
        assert_eq!(b.next_delay(), Duration::from_millis(500)); // step 2
        assert_eq!(b.next_delay(), Duration::from_secs(1)); // step 3
        assert_eq!(b.next_delay(), Duration::from_secs(2)); // step 4
        assert_eq!(b.next_delay(), Duration::from_secs(5)); // step 5: cap
        // Cap invariant: stays at 5s forever (infinite retries, per spec §5.2).
        assert_eq!(b.next_delay(), Duration::from_secs(5)); // step 6
        assert_eq!(b.next_delay(), Duration::from_secs(5)); // step 7
        assert_eq!(b.next_delay(), Duration::from_secs(5)); // step 8+
    }

    #[test]
    fn backoff_reset_returns_to_floor() {
        let mut b = Backoff::default();
        let _ = b.next_delay();
        let _ = b.next_delay();
        b.reset();
        assert_eq!(b.next_delay(), Duration::from_millis(250));
    }

    #[tokio::test]
    async fn spawn_returns_pipe_with_live_mpsc_ends() {
        // Smoke test: spawning the engine immediately returns a TapAudioPipe
        // the binary can wire into RtcSession. We don't actually wire a brain
        // here — the engine task will spin connect_brain (failing on a
        // non-existent server) and back off; the metrics counter should
        // increment. We abort the task on test drop to avoid leak.
        let id = ChannelId::new();
        let url = Url::parse("ws://127.0.0.1:1/echo").unwrap(); // port 1 = unreachable
        let (advisory_tx, _advisory_rx) = mpsc::channel::<rutster_media::AdvisoryEvent>(16);
        let (mut pipe, conn) = spawn_tap_engine(
            id,
            url,
            crate::session_map::AppState::default(),
            advisory_tx,
        );
        // TapAudioPipe is the seam object — should default to silent underflow.
        assert!(pipe.next_pcm_frame().is_none());
        // TapConn carries the close oneshot + JoinHandle + metrics.
        let _ = conn.close_tx.send(());
        conn.join.abort();
    }

    /// slice-3 §5.2: the TapConn returned by `spawn_tap_engine` must carry
    /// the tool-call side-channel ends so the binary's poll task can drain
    /// `function_call` events (→ `ToolRegistry::dispatch`) and write
    /// `function_call_output` replies. This is a structural smoke test —
    /// it pins that the plumbing lands (the binary's poll-task drain needs
    /// all three: a ready `rx_function_call`, a ready `tx_function_call_output`,
    /// and a `tool_registry` scoped to this session).
    #[tokio::test]
    async fn spawn_returns_tap_conn_with_function_call_side_channels() {
        let id = ChannelId::new();
        let url = Url::parse("ws://127.0.0.1:1/echo").unwrap(); // unreachable brain
        let (advisory_tx, _advisory_rx) = mpsc::channel::<rutster_media::AdvisoryEvent>(16);
        let (_pipe, conn) = spawn_tap_engine(
            id,
            url,
            crate::session_map::AppState::default(),
            advisory_tx,
        );

        // rx_function_call: Some(Receiver) — engine owns the paired Sender.
        assert!(
            conn.rx_function_call.is_some(),
            "TapConn must carry a drainable rx_function_call"
        );
        // tx_function_call_output: Some(Sender) — engine owns the paired
        // Receiver on the pump-side (run_tap_client).
        assert!(
            conn.tx_function_call_output.is_some(),
            "TapConn must carry a writeable tx_function_call_output"
        );
        // tool_registry: present (slice-3 §6.2 — one registry per active
        // channel; HangupTool is the only wired tool).
        let reg = conn.tool_registry.lock().await;
        assert_eq!(reg.catalog().len(), 1, "only hangup is wired in slice-3");
        assert_eq!(reg.catalog()[0]["name"], "hangup");

        let _ = conn.close_tx.send(());
        conn.join.abort();
    }

    /// slice-4: spawn_tap_engine takes advisory_tx as a parameter because
    /// the media thread owns the channel and needs to clone the sender to
    /// both the engine and the LocalVadReflex wrapper.
    #[tokio::test]
    async fn spawn_accepts_advisory_tx_parameter() {
        let id = ChannelId::new();
        let url = Url::parse("ws://127.0.0.1:1/echo").unwrap();
        let (advisory_tx, _advisory_rx) = mpsc::channel::<rutster_media::AdvisoryEvent>(16);
        let (_pipe, conn) = spawn_tap_engine(
            id,
            url,
            crate::session_map::AppState::default(),
            advisory_tx,
        );
        let _ = conn.close_tx.send(());
        conn.join.abort();
    }
}
