slice-5: scalability seams — addressing, admission, drain, events (review B1/M1-M7) (#14)

Co-authored-by: Aaron D. Lee <himself@adlee.work>
Co-committed-by: Aaron D. Lee <himself@adlee.work>

crates/rutster-call-model/src/lib.rs

@@ -57,6 +57,14 @@ mod tests {
         // Zero-sized type — confirms the "no extra allocation" claim.
         assert_eq!(std::mem::size_of::<TapHandle>(), 0);
     }
+
+    #[test]
+    fn channel_carries_wall_clock_start() {
+        let before = std::time::SystemTime::now();
+        let ch = Channel::new_inbound();
+        let after = std::time::SystemTime::now();
+        assert!(ch.started_at >= before && ch.started_at <= after);
+    }
 }
 
 use std::time::Instant;
@@ -150,6 +158,13 @@ pub struct Channel {
     /// user could change mid-call. The monotonic clock is the right
     /// tool for "has this peer been silent for 60 seconds?"
     pub created_at: Instant,
+    /// Wall-clock start (slice-5, review M3). `created_at: Instant` above
+    /// measures elapsed time (idle timeout); THIS field anchors the CDR —
+    /// "when did the call start" as a timestamp a bill or an audit can
+    /// use. A monotonic Instant cannot be converted to wall-clock after
+    /// the fact, which is why both exist: Instant for arithmetic,
+    /// SystemTime for the record.
+    pub started_at: std::time::SystemTime,
     /// NEW (slice-2, spec §5.2, §6): None until Connected, set on Connected,
     /// cleared on Closing. State invariant — see the table below.
     pub tap: Option<TapHandle>,
@@ -163,6 +178,7 @@ impl Channel {
             state: ChannelState::New,
             direction: Direction::Inbound,
             created_at: Instant::now(),
+            started_at: std::time::SystemTime::now(),
             tap: None,
         }
     }

crates/rutster-media/src/lib.rs

@@ -44,7 +44,7 @@ pub use reflex::{
     AdvisoryEvent, LocalVadReflex, Reflex, ReflexMetrics, ReflexMetricsSnapshot,
     VAD_DEBOUNCE_FRAMES, VAD_RMS_THRESHOLD,
 };
-pub use rtc_session::{RtcSession, RtcSessionError};
+pub use rtc_session::{MediaAddressConfig, RtcSession, RtcSessionError};
 
 use thiserror::Error;
 

crates/rutster-media/src/rtc_session.rs

@@ -17,7 +17,7 @@
 //! (which owns any carrier SIP/SDP, outside the trust boundary). rutster's
 //! WebRTC-ICE-coupled SDP lives entirely in str0m.
 
-use std::net::SocketAddr;
+use std::net::{IpAddr, SocketAddr, UdpSocket};
 use std::time::{Duration, Instant};
 
 use rutster_call_model::{Channel, ChannelId, ChannelState};
@@ -57,6 +57,55 @@ pub enum RtcSessionError {
 use str0m::change::SdpOffer;
 use str0m::media::Mid;
 
+/// Where the media socket binds vs. what address peers are told to reach.
+///
+/// # Why bind and advertised are SEPARATE concepts (slice-5, review B1)
+///
+/// Behind cloud NAT / a 1:1 elastic IP, the address a peer must send RTP
+/// to is NOT the local bind address — and RTP cannot ride the HTTP load
+/// balancer, so every instance must advertise its own reachable address.
+/// str0m also rejects the unspecified address (`0.0.0.0`) as a candidate,
+/// which forces the split at the type level: you may bind wildcard, but
+/// you must then say what to advertise.
+///
+/// `port_range` bounds the UDP ports (inclusive) so a fleet's security
+/// groups can be written; `None` keeps the OS-ephemeral behavior.
+#[derive(Debug, Clone)]
+pub struct MediaAddressConfig {
+    pub bind_ip: IpAddr,
+    pub advertised_ip: Option<IpAddr>,
+    pub port_range: Option<(u16, u16)>,
+}
+
+impl Default for MediaAddressConfig {
+    /// Loopback + ephemeral — byte-for-byte the pre-slice-5 behavior, so
+    /// every existing test and the dev loop run unchanged.
+    fn default() -> Self {
+        Self {
+            bind_ip: IpAddr::from([127, 0, 0, 1]),
+            advertised_ip: None,
+            port_range: None,
+        }
+    }
+}
+
+/// Bind the first free UDP port in `[lo, hi]` (inclusive).
+///
+/// Sequential scan — O(range) worst case, cold path only (session
+/// construction). The OS is the allocator: a failed bind means "taken,"
+/// so no shared allocator state is needed across sessions or threads.
+fn bind_in_range(ip: IpAddr, lo: u16, hi: u16) -> std::io::Result<UdpSocket> {
+    let mut last_err = None;
+    for port in lo..=hi {
+        match UdpSocket::bind(SocketAddr::new(ip, port)) {
+            Ok(s) => return Ok(s),
+            Err(e) => last_err = Some(e),
+        }
+    }
+    Err(last_err
+        .unwrap_or_else(|| std::io::Error::new(std::io::ErrorKind::AddrInUse, "empty port range")))
+}
+
 /// The per-peer media owner (spec §3.1, §4.5).
 ///
 /// # Ownership / sharing
@@ -81,11 +130,16 @@ pub struct RtcSession {
     pub(crate) pipe: Box<dyn AudioPipe + Send>,
     /// Local UDP socket str0m sends `Transmit` packets out on and
     /// receives `Input::Receive` packets from. Bound to an ephemeral
-    /// port at construction; the local candidate passed to str0m at
-    /// offer-accept time uses this address.
+    /// port at construction. The SDP candidate passed to the peer uses
+    /// the separate `advertised_addr` field, not this socket's address.
     pub(crate) socket: std::net::UdpSocket,
     /// Local socket address — cached because `local_addr()` is a syscall.
     pub(crate) local_addr: SocketAddr,
+    /// The address written into our ICE host candidate — advertised IP
+    /// (if configured) + the actually-bound port. `local_addr` stays the
+    /// real socket address (loop_driver's `Receive::new` needs it);
+    /// `advertised_addr` is what the SDP answer tells the peer.
+    pub(crate) advertised_addr: SocketAddr,
     /// Mid of the audio m-line we accepted. Set during `accept_offer`.
     /// Slice 1 has exactly one m-line; multi-m-line arrives with video.
     pub(crate) audio_mid: Option<Mid>,
@@ -114,7 +168,48 @@ impl RtcSession {
     /// `for_server` split; the body is identical (binding a UDP socket
     /// on `0.0.0.0:0`, constructing the `Rtc` + codecs).
     pub fn new() -> Result<Self, RtcSessionError> {
-        Self::new_internal()
+        Self::new_with_config(&MediaAddressConfig::default())
+    }
+
+    /// Construct with explicit addressing (slice-5, review B1). The
+    /// binary threads `MediaThreadOpts.media_cfg` here; `new()` keeps the
+    /// loopback default for tests and the dev loop.
+    pub fn new_with_config(cfg: &MediaAddressConfig) -> Result<Self, RtcSessionError> {
+        // Fail-fast: a wildcard bind with nothing to advertise would die
+        // later inside accept_offer (str0m rejects 0.0.0.0 candidates).
+        // Cold path → a real error, not an expect().
+        if cfg.bind_ip.is_unspecified() && cfg.advertised_ip.is_none() {
+            return Err(RtcSessionError::Socket(std::io::Error::new(
+                std::io::ErrorKind::InvalidInput,
+                "RUTSTER_MEDIA_BIND_IP is unspecified (0.0.0.0/::) — set RUTSTER_MEDIA_ADVERTISED_IP",
+            )));
+        }
+        let socket = match cfg.port_range {
+            Some((lo, hi)) => bind_in_range(cfg.bind_ip, lo, hi)?,
+            None => UdpSocket::bind(SocketAddr::new(cfg.bind_ip, 0))?,
+        };
+        socket.set_nonblocking(true)?;
+        let local_addr = socket.local_addr()?;
+        let advertised_addr =
+            SocketAddr::new(cfg.advertised_ip.unwrap_or(cfg.bind_ip), local_addr.port());
+
+        let rtc = Rtc::new(Instant::now());
+
+        Ok(Self {
+            channel: Channel::new_inbound(),
+            rtc,
+            decoder: OpusDecoder::new()?,
+            encoder: OpusEncoder::new()?,
+            pipe: Box::new(EchoAudioPipe::new()),
+            socket,
+            local_addr,
+            advertised_addr,
+            audio_mid: None,
+            next_timeout: None,
+            last_rx: Instant::now(),
+            last_outbound_at: Instant::now(),
+            next_media_time: str0m::media::MediaTime::ZERO,
+        })
     }
 
     /// Replace the default `EchoAudioPipe` with a real tap pipe (slice-2,
@@ -163,41 +258,6 @@ impl RtcSession {
         self.pipe.clear_playout_ring();
     }
 
-    fn new_internal() -> Result<Self, RtcSessionError> {
-        // Bind an ephemeral UDP socket. We use std::net::UdpSocket and
-        // drive it non-blocking from tokio rather than tokio's UdpSocket:
-        // str0m operates on raw `Receive` values and yields `Transmit`
-        // values, both of which are plain structs — no async needed.
-        // Setting non-blocking lets us `recv_from` without blocking.
-        //
-        // We bind to 127.0.0.1 (not 0.0.0.0) because the local candidate
-        // we hand to str0m uses this socket's address, and str0m's
-        // `Candidate::host` rejects the unspecified address 0.0.0.0 as
-        // invalid. Slice 1 is loopback-only; production binding (when
-        // the binary lands in Task 5) can rebind to a real interface
-        // address before `accept_offer` constructs the candidate.
-        let socket = std::net::UdpSocket::bind("127.0.0.1:0")?;
-        socket.set_nonblocking(true)?;
-        let local_addr = socket.local_addr()?;
-
-        let rtc = Rtc::new(Instant::now());
-
-        Ok(Self {
-            channel: Channel::new_inbound(),
-            rtc,
-            decoder: OpusDecoder::new()?,
-            encoder: OpusEncoder::new()?,
-            pipe: Box::new(EchoAudioPipe::new()),
-            socket,
-            local_addr,
-            audio_mid: None,
-            next_timeout: None,
-            last_rx: Instant::now(),
-            last_outbound_at: Instant::now(),
-            next_media_time: str0m::media::MediaTime::ZERO,
-        })
-    }
-
     /// Returns the session's `ChannelId` — also the REST API session id
     /// surfaced at `/v1/sessions/{id}` (spec §4.5).
     pub fn channel_id(&self) -> ChannelId {
@@ -233,11 +293,10 @@ impl RtcSession {
         // this candidate's address + the ICE creds it generates in the SDP
         // answer. `add_local_candidate` returns `Option<&Candidate>` —
         // `None` means str0m rejected it (log + continue; not fatal).
-        let candidate = str0m::Candidate::host(self.local_addr, "udp")
-            .expect("host candidate from bound UDP socket");
-        // ^-- expect is acceptable here: this is construction (cold path),
-        // not the hot path. A bound UDP socket always yields a valid
-        // host candidate; only an absurd Protocol parse fails.
+        let candidate = str0m::Candidate::host(self.advertised_addr, "udp")
+            .expect("host candidate from validated advertised address");
+        // ^-- expect stays acceptable: new_with_config rejected the only
+        // input (unspecified IP) that str0m refuses, so this is const-ish.
         if self.rtc.add_local_candidate(candidate).is_none() {
             tracing::warn!(channel_id = %self.channel.id, "str0m rejected local candidate");
         }
@@ -352,4 +411,60 @@ a=candidate:1 1 UDP 2113667327 192.168.1.42 50000 typ host\r
         let _ = session.accept_offer(BROWSER_SDP_OFFER).expect("answer");
         assert_eq!(session.channel_state(), ChannelState::Connecting);
     }
+
+    #[test]
+    fn default_config_binds_loopback_ephemeral() {
+        // Byte-for-byte behavioral compatibility: new() == new_with_config(default).
+        let s = RtcSession::new_with_config(&MediaAddressConfig::default()).expect("session");
+        assert!(s.local_addr.ip().is_loopback());
+    }
+
+    #[test]
+    fn port_range_is_respected_and_exhaustion_errors() {
+        let cfg = MediaAddressConfig {
+            bind_ip: "127.0.0.1".parse().unwrap(),
+            advertised_ip: None,
+            port_range: Some((41500, 41501)), // room for exactly two sessions
+        };
+        let a = RtcSession::new_with_config(&cfg).expect("first");
+        let b = RtcSession::new_with_config(&cfg).expect("second");
+        assert!((41500..=41501).contains(&a.local_addr.port()));
+        assert!((41500..=41501).contains(&b.local_addr.port()));
+        // Range full → cold-path Socket error, not a panic.
+        assert!(matches!(
+            RtcSession::new_with_config(&cfg),
+            Err(RtcSessionError::Socket(_))
+        ));
+    }
+
+    #[test]
+    fn sdp_answer_advertises_the_advertised_ip_not_the_bind_ip() {
+        let cfg = MediaAddressConfig {
+            bind_ip: "127.0.0.1".parse().unwrap(),
+            advertised_ip: Some("203.0.113.7".parse().unwrap()), // TEST-NET-3
+            port_range: None,
+        };
+        let mut s = RtcSession::new_with_config(&cfg).expect("session");
+        let answer = s.accept_offer(BROWSER_SDP_OFFER).expect("answer");
+        assert!(
+            answer.contains("203.0.113.7"),
+            "candidate must carry advertised IP"
+        );
+        assert!(
+            !answer.contains("127.0.0.1 "),
+            "bind IP must not leak into the candidate"
+        );
+    }
+
+    #[test]
+    fn unspecified_bind_without_advertised_ip_is_rejected() {
+        // str0m's Candidate::host rejects 0.0.0.0 — fail at construction
+        // with a config error, not at offer-accept with an expect().
+        let cfg = MediaAddressConfig {
+            bind_ip: "0.0.0.0".parse().unwrap(),
+            advertised_ip: None,
+            port_range: None,
+        };
+        assert!(RtcSession::new_with_config(&cfg).is_err());
+    }
 }

crates/rutster-tap/src/protocol.rs

@@ -15,6 +15,29 @@
 //! implement a byte-parser. ~33% wire overhead (65 KB/s at 24 kHz mono
 //! i16) is negligible at slice-2's scale; brain-authoring ergonomics
 //! dominate. A future-rung `v: 2` may negotiate a binary mode (spec §9).
+//!
+//! ## v2 reservations (documented 2026-07-04; implemented at the version bump)
+//!
+//! The 2026-07-04 scalability review (finding M4) identified the missing
+//! reconnect semantics as the most calcifying gap in the system — wire
+//! protocols are the hardest surface to retrofit. v1 stays frozen; the
+//! v2 negotiation MUST carry:
+//!
+//! - **`hello.resume_token`** — an opaque token the brain returned on a
+//!   prior accept for this session, plus a **connection epoch** counter.
+//!   Lets a brain fleet distinguish "resume this conversation" (state
+//!   held / recoverable) from "unknown session" (context lost).
+//! - **`hello_ack.resume: accepted | rejected`** — the brain's explicit
+//!   answer. A rejected resume tells the core the brain is amnesiac, so
+//!   the core (not the caller's ears) decides: re-prime, escalate, or end.
+//! - **`bye.reason: terminal | transient`** — today every close funnels
+//!   into infinite re-dial (5s cap, forever); a brain that deliberately
+//!   ends a session is re-dialed for the rest of the call. `terminal`
+//!   exits the retry loop.
+//!
+//! Anything in this list changes BOTH sides of the wire — which is why it
+//! is reserved here, in the protocol's own doc, rather than in a review
+//! doc nobody re-reads at implementation time.
 
 use rutster_media::PcmFrame;
 use serde::ser::{SerializeStruct, Serializer};

crates/rutster/src/config.rs

@@ -0,0 +1,189 @@
+//! # config — pure env-parsing helpers (slice-5)
+//!
+//! Every knob is a pure function over `Option<String>` / `&str` so tests
+//! never mutate process env (env mutation in tests races across the
+//! parallel test harness — the same reason `api_key.rs` needs its
+//! ENV_MUTEX). `main.rs` is the only caller that touches `std::env`.
+//!
+//! Closes 2026-07-04 scalability review "HTTP bind hardcoded" (minor).
+
+use std::net::{IpAddr, SocketAddr};
+
+use rutster_media::MediaAddressConfig;
+
+/// Resolve the HTTP bind address from `RUTSTER_HTTP_BIND`.
+///
+/// `None` → the historical default `0.0.0.0:8080`. Invalid input is a
+/// hard error (fail-fast at startup — an operator typo must not silently
+/// bind the default and hide behind an LB health check).
+pub fn http_bind(raw: Option<String>) -> Result<SocketAddr, String> {
+    match raw {
+        None => Ok("0.0.0.0:8080".parse().expect("static default parses")),
+        Some(s) => s
+            .parse()
+            .map_err(|e| format!("RUTSTER_HTTP_BIND {s:?} is not a socket address: {e}")),
+    }
+}
+
+/// Parse `RUTSTER_MEDIA_PORT_RANGE` — `"lo-hi"`, inclusive, lo ≤ hi.
+pub fn parse_port_range(raw: &str) -> Result<(u16, u16), String> {
+    let (lo, hi) = raw
+        .split_once('-')
+        .ok_or_else(|| format!("RUTSTER_MEDIA_PORT_RANGE {raw:?}: expected \"lo-hi\""))?;
+    let lo: u16 = lo
+        .trim()
+        .parse()
+        .map_err(|e| format!("RUTSTER_MEDIA_PORT_RANGE lo: {e}"))?;
+    let hi: u16 = hi
+        .trim()
+        .parse()
+        .map_err(|e| format!("RUTSTER_MEDIA_PORT_RANGE hi: {e}"))?;
+    if lo > hi {
+        return Err(format!("RUTSTER_MEDIA_PORT_RANGE {raw:?}: lo > hi"));
+    }
+    Ok((lo, hi))
+}
+
+/// Assemble `MediaAddressConfig` from the three `RUTSTER_MEDIA_*` vars.
+pub fn media_address_config(
+    bind_ip: Option<String>,
+    advertised_ip: Option<String>,
+    port_range: Option<String>,
+) -> Result<MediaAddressConfig, String> {
+    let mut cfg = MediaAddressConfig::default();
+    if let Some(s) = bind_ip {
+        cfg.bind_ip = s
+            .parse::<IpAddr>()
+            .map_err(|e| format!("RUTSTER_MEDIA_BIND_IP: {e}"))?;
+    }
+    if let Some(s) = advertised_ip {
+        cfg.advertised_ip = Some(
+            s.parse::<IpAddr>()
+                .map_err(|e| format!("RUTSTER_MEDIA_ADVERTISED_IP: {e}"))?,
+        );
+    }
+    if let Some(s) = port_range {
+        cfg.port_range = Some(parse_port_range(&s)?);
+    }
+    Ok(cfg)
+}
+
+/// `RUTSTER_MAX_SESSIONS` — admission cap. Default 64 (placeholder until
+/// the ADR-0010 benchmark measures the real per-node ceiling).
+pub fn max_sessions(raw: Option<String>) -> Result<usize, String> {
+    match raw {
+        None => Ok(64),
+        Some(s) => s
+            .parse()
+            .map_err(|e| format!("RUTSTER_MAX_SESSIONS {s:?}: {e}")),
+    }
+}
+
+/// `RUTSTER_DRAIN_DEADLINE_SECS` — how long shutdown waits for in-flight
+/// calls before the hard stop. Default 0 = today's instant shutdown (dev
+/// loop); production sets minutes. Calls are non-migratable by design, so
+/// drain-then-terminate is the ONLY graceful scale-in shape.
+pub fn drain_deadline(raw: Option<String>) -> Result<std::time::Duration, String> {
+    match raw {
+        None => Ok(std::time::Duration::ZERO),
+        Some(s) => s
+            .parse::<u64>()
+            .map(std::time::Duration::from_secs)
+            .map_err(|e| format!("RUTSTER_DRAIN_DEADLINE_SECS {s:?}: {e}")),
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn http_bind_defaults_when_unset() {
+        assert_eq!(
+            http_bind(None).unwrap(),
+            "0.0.0.0:8080".parse::<SocketAddr>().unwrap()
+        );
+    }
+
+    #[test]
+    fn http_bind_parses_override() {
+        assert_eq!(
+            http_bind(Some("127.0.0.1:9090".into())).unwrap(),
+            "127.0.0.1:9090".parse::<SocketAddr>().unwrap()
+        );
+    }
+
+    #[test]
+    fn http_bind_rejects_garbage_with_var_name_in_error() {
+        let err = http_bind(Some("not-an-addr".into())).unwrap_err();
+        assert!(err.contains("RUTSTER_HTTP_BIND"));
+    }
+
+    #[test]
+    fn port_range_parses_inclusive_pair() {
+        assert_eq!(parse_port_range("10000-20000").unwrap(), (10000, 20000));
+    }
+
+    #[test]
+    fn port_range_rejects_inverted_and_garbage() {
+        assert!(parse_port_range("20000-10000").is_err());
+        assert!(parse_port_range("10000").is_err());
+        assert!(parse_port_range("a-b").is_err());
+    }
+
+    #[test]
+    fn media_address_config_defaults_to_loopback_ephemeral() {
+        let cfg = media_address_config(None, None, None).unwrap();
+        assert!(cfg.bind_ip.is_loopback());
+        assert!(cfg.advertised_ip.is_none());
+        assert!(cfg.port_range.is_none());
+    }
+
+    #[test]
+    fn max_sessions_defaults_when_unset() {
+        assert_eq!(max_sessions(None).unwrap(), 64);
+    }
+
+    #[test]
+    fn max_sessions_parses_override() {
+        assert_eq!(max_sessions(Some("128".into())).unwrap(), 128);
+    }
+
+    #[test]
+    fn max_sessions_rejects_garbage_with_var_name_in_error() {
+        let err = max_sessions(Some("zero".into())).unwrap_err();
+        assert!(err.contains("RUTSTER_MAX_SESSIONS"));
+    }
+
+    #[test]
+    fn drain_deadline_defaults_to_zero_when_unset() {
+        assert_eq!(drain_deadline(None).unwrap(), std::time::Duration::ZERO);
+    }
+
+    #[test]
+    fn drain_deadline_parses_seconds() {
+        assert_eq!(
+            drain_deadline(Some("300".into())).unwrap(),
+            std::time::Duration::from_secs(300)
+        );
+    }
+
+    #[test]
+    fn drain_deadline_rejects_garbage_with_var_name_in_error() {
+        let err = drain_deadline(Some("soon".into())).unwrap_err();
+        assert!(err.contains("RUTSTER_DRAIN_DEADLINE_SECS"));
+    }
+
+    #[test]
+    fn media_address_config_parses_all_three() {
+        let cfg = media_address_config(
+            Some("0.0.0.0".into()),
+            Some("203.0.113.7".into()),
+            Some("10000-10100".into()),
+        )
+        .unwrap();
+        assert!(cfg.bind_ip.is_unspecified());
+        assert_eq!(cfg.advertised_ip.unwrap().to_string(), "203.0.113.7");
+        assert_eq!(cfg.port_range, Some((10000, 10100)));
+    }
+}

crates/rutster/src/event_sink.rs

@@ -0,0 +1,72 @@
+//! # event_sink — durable-lifecycle seam (slice-5, review M3)
+//!
+//! ADR-0005 promises call lifecycle → Valkey streams → durable CDR
+//! pipeline. Until that lands, lifecycle events die with the process —
+//! an OOM-killed node erases all evidence of its calls. This module is
+//! the SEAM: the media thread emits `CallEvent`s through `EventSink`;
+//! today's impl logs, the ADR-0005 impl will publish. The emission
+//! points and the event shape are the part that calcifies — the backend
+//! is plumbing.
+
+use std::sync::Arc;
+use std::time::SystemTime;
+
+use rutster_call_model::ChannelId;
+use rutster_tap::MetricsSnapshot;
+
+/// CDR-shaped lifecycle events. `SessionEnded` carries both timestamps so
+/// a consumer computes duration without correlating two events (a node
+/// can die between them — the whole point of emitting durably).
+#[derive(Debug, Clone)]
+pub enum CallEvent {
+    SessionRegistered {
+        id: ChannelId,
+        at: SystemTime,
+    },
+    SessionConnected {
+        id: ChannelId,
+        at: SystemTime,
+    },
+    SessionEnded {
+        id: ChannelId,
+        started_at: SystemTime,
+        ended_at: SystemTime,
+        reason: EndReason,
+        /// Tap counters at teardown — fulfills the tap_engine.rs promise
+        /// ("read a MetricsSnapshot for log/CDR emission on teardown").
+        /// Snapshot is taken just before the engine task is torn down, so
+        /// counts may trail the final frame by a tick — fine for a CDR.
+        tap_metrics: Option<MetricsSnapshot>,
+    },
+}
+
+/// Why the session ended — the CDR disposition embryo.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum EndReason {
+    /// DELETE /v1/sessions/:id (API-driven hangup).
+    Deleted,
+    /// The session closed itself (peer close / idle timeout).
+    Closed,
+    /// Node shutdown dropped it (the mass-hangup path drain exists to avoid).
+    Shutdown,
+}
+
+/// The seam. CONTRACT: `emit` is called from the media std::thread —
+/// implementations MUST NOT block or perform I/O inline. Buffer to your
+/// own task (the future Valkey impl does channel → tokio publisher).
+pub trait EventSink: Send + Sync {
+    fn emit(&self, event: CallEvent);
+}
+
+/// Log-backed sink: structured tracing on target "rutster::events". Not
+/// durable — the placeholder that makes the emission points real today.
+pub struct TracingEventSink;
+
+impl EventSink for TracingEventSink {
+    fn emit(&self, event: CallEvent) {
+        tracing::info!(target: "rutster::events", event = ?event, "call event");
+    }
+}
+
+/// Convenience alias for the opts field / spawn signature.
+pub type SharedEventSink = Arc<dyn EventSink>;

crates/rutster/src/lib.rs

@@ -22,6 +22,8 @@
 //! - [slice-1 spec §4](../../docs/superpowers/specs/2026-06-28-slice-1-webrtc-loopback-design.md)
 //! - [ARCHITECTURE.md](../../docs/ARCHITECTURE.md) — fused vertical.
 
+pub mod config;
+pub mod event_sink;
 pub mod media_thread;
 pub mod routes;
 pub mod session_map;

crates/rutster/src/main.rs

@@ -37,26 +37,87 @@ async fn main() {
     // from the spawned MediaThread below.
     let (placeholder_tx, _placeholder_rx) = mpsc::channel(1);
     let app_state = AppState::new(placeholder_tx, default_tap_url);
+
+    let media_cfg = rutster::config::media_address_config(
+        std::env::var("RUTSTER_MEDIA_BIND_IP").ok(),
+        std::env::var("RUTSTER_MEDIA_ADVERTISED_IP").ok(),
+        std::env::var("RUTSTER_MEDIA_PORT_RANGE").ok(),
+    )
+    .expect("RUTSTER_MEDIA_* env config invalid");
+    let opts = rutster::media_thread::MediaThreadOpts {
+        media_cfg,
+        max_sessions: rutster::config::max_sessions(std::env::var("RUTSTER_MAX_SESSIONS").ok())
+            .expect("RUTSTER_MAX_SESSIONS must be a number"),
+        // sink: TracingEventSink via Default — ADR-0005 Valkey impl replaces it at the same seam.
+        ..Default::default()
+    };
     let (app_state, media_thread) = app_state
-        .spawn_media_thread(tokio::runtime::Handle::current())
+        .spawn_media_thread_with(opts, tokio::runtime::Handle::current())
         .expect("media thread spawn");
 
-    let addr: SocketAddr = "0.0.0.0:8080".parse().expect("valid addr");
+    // RUTSTER_HTTP_BIND: per-instance bind is the first thing any LB /
+    // compose template parametrizes (slice-5; matches the existing
+    // RUTSTER_TAP_BIND pattern in rutster-brain-realtime).
+    let addr: SocketAddr = rutster::config::http_bind(std::env::var("RUTSTER_HTTP_BIND").ok())
+        .expect("RUTSTER_HTTP_BIND must be host:port");
+    let drain_deadline =
+        rutster::config::drain_deadline(std::env::var("RUTSTER_DRAIN_DEADLINE_SECS").ok())
+            .expect("RUTSTER_DRAIN_DEADLINE_SECS must be seconds");
     info!(%addr, "listening");
     let listener = tokio::net::TcpListener::bind(addr).await.unwrap();
+
+    // Two-phase shutdown (slice-5, review M1):
+    //   SIGTERM → Drain (stop admitting; keep serving HTTP for in-flight
+    //   calls' DELETE/offer; readyz flips so the LB pulls us) → deadline
+    //   → HTTP quiesce → media_thread.shutdown() (the hard stop).
+    let (http_stop_tx, http_stop_rx) = tokio::sync::oneshot::channel::<()>();
+    let drain_cmd_tx = app_state.cmd_tx.clone();
+    tokio::spawn(async move {
+        shutdown_signal().await;
+        if !drain_deadline.is_zero() {
+            let (reply, drained) = tokio::sync::oneshot::channel();
+            // One deadline bounds the WHOLE drain round trip — send AND
+            // completion. A wedged media thread can stall the send itself
+            // (full command channel), and the shutdown promise is "bounded
+            // by the deadline" against exactly that node. Same principle
+            // as readyz's whole-round-trip timeout (commit 5ce18bf).
+            let outcome = tokio::time::timeout(drain_deadline, async {
+                if drain_cmd_tx
+                    .send(rutster::media_thread::MediaCmd::Drain { reply })
+                    .await
+                    .is_ok()
+                {
+                    let _ = drained.await;
+                }
+            })
+            .await;
+            match outcome {
+                Ok(()) => info!("drain complete; proceeding to shutdown"),
+                Err(_) => info!(?drain_deadline, "drain deadline hit; shutting down anyway"),
+            }
+        }
+        let _ = http_stop_tx.send(());
+    });
+
     axum::serve(listener, router(app_state))
-        .with_graceful_shutdown(shutdown_signal())
+        .with_graceful_shutdown(async {
+            let _ = http_stop_rx.await;
+        })
         .await
         .unwrap();
 
     media_thread.shutdown();
 }
 
-/// Ctrl-C / SIGTERM handler (spec §4.5). After the signal, `main` returns
-/// from `axum::serve(...)` and calls `media_thread.shutdown()`, which sends
-/// `MediaCmd::Shutdown` to the dedicated thread. The thread drains its
-/// session map, exits its loop, and is joined. No in-flight call
-/// preservation story in the dev loop.
+/// Ctrl-C / SIGTERM handler (spec §4.5). This future resolves the instant
+/// the signal arrives — it does NOT wait for the drain. The caller (`main`,
+/// slice-5 review M1) is the two-phase shutdown: after this resolves, it
+/// sends `MediaCmd::Drain` and keeps `axum::serve` running so in-flight
+/// calls can still hit DELETE/offer while readyz flips the LB away from
+/// this node; once the map empties or `RUTSTER_DRAIN_DEADLINE_SECS` elapses
+/// (whichever first — default 0 skips this wait entirely, preserving the
+/// old instant-shutdown dev loop), HTTP quiesces and `media_thread.shutdown()`
+/// sends `MediaCmd::Shutdown` for the hard stop.
 async fn shutdown_signal() {
     let ctrl_c = async {
         tokio::signal::ctrl_c()

crates/rutster/src/media_thread.rs

@@ -38,14 +38,40 @@ const META_TICK: Duration = Duration::from_millis(10);
 /// Capacity for the command channel from axum to the media thread.
 const CMD_CHANNEL_CAPACITY: usize = 64;
 
+/// Knobs the binary resolves from env and hands to the media thread.
+/// Manual Default (not derive) because `sink: Arc<dyn EventSink>` has no
+/// meaningful derived default — `Arc::new(TracingEventSink)` is a
+/// deliberate choice, not a zero value. Defaults reproduce the
+/// pre-slice-5 single-node dev behavior exactly.
+pub struct MediaThreadOpts {
+    pub media_cfg: rutster_media::MediaAddressConfig,
+    /// Admission cap (review M2). 64 is a placeholder until the ADR-0010
+    /// benchmark produces the real per-node number — the gauge below is
+    /// that benchmark's primary readout.
+    pub max_sessions: usize,
+    /// Lifecycle event seam (Task 7, review M3). Default is the log-backed
+    /// `TracingEventSink`; ADR-0005's Valkey publisher swaps in here.
+    pub sink: crate::event_sink::SharedEventSink,
+}
+
+impl Default for MediaThreadOpts {
+    fn default() -> Self {
+        Self {
+            media_cfg: rutster_media::MediaAddressConfig::default(),
+            max_sessions: 64,
+            sink: std::sync::Arc::new(crate::event_sink::TracingEventSink),
+        }
+    }
+}
+
 /// Commands axum sends to the media thread (cold-path only — NEVER on
 /// the 20ms tick). The thread owns RtcSessions exclusively; this is the
 /// ONLY entry point for axum-side mutation.
 #[derive(Debug)]
 pub enum MediaCmd {
     /// Construct a fresh RtcSession, store it under a new ChannelId, reply.
-    /// The thread constructs RtcSession::new() (keeps all RtcSession
-    /// construction on the thread that owns it).
+    /// The thread constructs RtcSession::new_with_config(&opts.media_cfg)
+    /// (keeps all RtcSession construction on the thread that owns it).
     Register {
         tap_url: url::Url,
         reply: oneshot::Sender<Result<ChannelId, String>>,
@@ -65,6 +91,29 @@ pub enum MediaCmd {
     },
     /// Graceful shutdown — drain + drop + join.
     Shutdown { reply: oneshot::Sender<()> },
+    /// Enter drain (review M1): reject new Registers ("draining" → 503),
+    /// keep ticking existing sessions, fire `reply` when the map empties.
+    /// The deadline lives with the CALLER (main.rs times out and proceeds
+    /// to Shutdown) — the thread just reports emptiness; that keeps
+    /// "how long is too long" an operator policy, not thread logic.
+    Drain { reply: oneshot::Sender<()> },
+    /// Cold-path capacity/health snapshot (readyz + the autoscaling
+    /// signal; review M2). Cheap: counters the loop already maintains.
+    Stats { reply: oneshot::Sender<MediaStats> },
+}
+
+/// What the node tells the platform about itself. `serde::Serialize`
+/// because /readyz returns it verbatim as JSON.
+#[derive(Debug, Clone, serde::Serialize)]
+pub struct MediaStats {
+    pub sessions: usize,
+    pub max_sessions: usize,
+    pub draining: bool,
+    /// Ticks whose work exceeded META_TICK — the saturation signal.
+    /// Overload degrades EVERY call at once (missed 20ms deadlines), so
+    /// this must trend at ~0; an autoscaler scales out on its slope.
+    pub tick_overruns: u64,
+    pub last_tick_micros: u64,
 }
 
 /// The handle returned to the binary. Clone the `cmd_tx` per-session;
@@ -95,6 +144,7 @@ impl MediaThread {
     /// crash" policy.
     pub fn spawn(
         default_tap_url: url::Url,
+        opts: MediaThreadOpts,
         tokio_handle: tokio::runtime::Handle,
     ) -> Result<Self, std::io::Error> {
         let (cmd_tx, cmd_rx) = mpsc::channel(CMD_CHANNEL_CAPACITY);
@@ -106,7 +156,13 @@ impl MediaThread {
         let join = std::thread::Builder::new()
             .name("rutster-media".into())
             .spawn(move || {
-                run_media_thread(cmd_rx, default_tap_url, tokio_handle, cmd_tx_for_thread);
+                run_media_thread(
+                    cmd_rx,
+                    default_tap_url,
+                    opts,
+                    tokio_handle,
+                    cmd_tx_for_thread,
+                );
             })?;
         Ok(Self {
             cmd_tx,
@@ -158,39 +214,74 @@ struct ThreadSession {
 fn run_media_thread(
     mut cmd_rx: mpsc::Receiver<MediaCmd>,
     default_tap_url: url::Url,
+    opts: MediaThreadOpts,
     tokio_handle: tokio::runtime::Handle,
     media_cmd_tx: mpsc::Sender<MediaCmd>,
 ) {
     let mut sessions: HashMap<ChannelId, ThreadSession> = HashMap::new();
+    let max_sessions = opts.max_sessions;
+    let sink = opts.sink.clone();
+    let mut tick_overruns: u64 = 0;
+    let mut last_tick_micros: u64 = 0;
+    let mut draining = false;
+    let mut drain_done: Option<oneshot::Sender<()>> = None;
     info!("media thread started");
 
     loop {
+        let tick_started = Instant::now();
+
         // === Step 1: drain ALL pending commands (cold path) BEFORE ticking. ===
         while let Ok(cmd) = cmd_rx.try_recv() {
             match cmd {
-                MediaCmd::Register { tap_url, reply } => match RtcSession::new() {
-                    Ok(session) => {
-                        let id = session.channel_id();
-                        let app_state = crate::session_map::AppState::new(
-                            media_cmd_tx.clone(),
-                            default_tap_url.clone(),
-                        );
-                        sessions.insert(
-                            id,
-                            ThreadSession {
-                                rtc: session,
-                                tap_url,
-                                app_state,
-                                tap_conn: None,
-                            },
-                        );
-                        let _ = reply.send(Ok(id));
-                        debug!(channel_id = %id, "session registered");
+                MediaCmd::Register { tap_url, reply } => {
+                    // Drain (review M1) takes priority over the capacity
+                    // check below: once draining, EVERY new Register is
+                    // shed regardless of headroom — we're bleeding out,
+                    // not admitting.
+                    if draining {
+                        let _ = reply.send(Err("draining: not accepting new sessions".into()));
+                        continue;
                     }
-                    Err(e) => {
-                        let _ = reply.send(Err(format!("RtcSession::new: {e}")));
+                    // Admission control (review M2): shed the marginal
+                    // call with a crisp 503 instead of degrading every
+                    // in-flight call's 20ms budget. "node full" is the
+                    // routes-layer contract string.
+                    if sessions.len() >= max_sessions {
+                        let _ = reply.send(Err(format!(
+                            "node full: {} sessions (max {max_sessions})",
+                            sessions.len()
+                        )));
+                        continue;
                     }
-                },
+                    match RtcSession::new_with_config(&opts.media_cfg) {
+                        Ok(session) => {
+                            let id = session.channel_id();
+                            let started_at = session.channel.started_at;
+                            let app_state = crate::session_map::AppState::new(
+                                media_cmd_tx.clone(),
+                                default_tap_url.clone(),
+                            );
+                            sessions.insert(
+                                id,
+                                ThreadSession {
+                                    rtc: session,
+                                    tap_url,
+                                    app_state,
+                                    tap_conn: None,
+                                },
+                            );
+                            sink.emit(crate::event_sink::CallEvent::SessionRegistered {
+                                id,
+                                at: started_at,
+                            });
+                            let _ = reply.send(Ok(id));
+                            debug!(channel_id = %id, "session registered");
+                        }
+                        Err(e) => {
+                            let _ = reply.send(Err(format!("RtcSession::new: {e}")));
+                        }
+                    }
+                }
                 MediaCmd::AcceptOffer { id, sdp, reply } => {
                     let result = match sessions.get_mut(&id) {
                         Some(s) => s.rtc.accept_offer(&sdp).map_err(|e| format!("{e}")),
@@ -200,21 +291,21 @@ fn run_media_thread(
                 }
                 MediaCmd::Delete { id, reply } => {
                     if let Some(mut s) = sessions.remove(&id) {
-                        if let Some(mut conn) = s.tap_conn.take() {
-                            let _ = conn.close_tx.send(());
-                            let teardown = tokio_handle.block_on(tokio::time::timeout(
-                                Duration::from_millis(750),
-                                &mut conn.join,
-                            ));
-                            match teardown {
-                                Ok(Ok(())) => {
-                                    info!(channel_id = %id, "tap engine torn down via Delete (graceful)");
-                                }
-                                _ => {
-                                    conn.join.abort();
-                                    info!(channel_id = %id, "tap engine torn down via Delete (abort after timeout)");
-                                }
-                            }
+                        // Snapshot BEFORE take(): after the take the conn
+                        // (and its metrics Arc) belongs to the teardown task.
+                        let tap_metrics = s.tap_conn.as_ref().map(|c| c.metrics.snapshot());
+                        sink.emit(crate::event_sink::CallEvent::SessionEnded {
+                            id,
+                            started_at: s.rtc.channel.started_at,
+                            ended_at: std::time::SystemTime::now(),
+                            reason: crate::event_sink::EndReason::Deleted,
+                            tap_metrics,
+                        });
+                        if let Some(conn) = s.tap_conn.take() {
+                            // Hand the 750ms bounded teardown to tokio —
+                            // the tick loop never waits on brain I/O
+                            // (review M7; see spawn_tap_teardown docs).
+                            crate::tap_engine::spawn_tap_teardown(&tokio_handle, id, conn);
                         }
                         s.rtc.channel.tap = None;
                         s.rtc.channel.state = rutster_call_model::ChannelState::Closed;
@@ -226,10 +317,36 @@ fn run_media_thread(
                         "media thread shutdown; dropping {} sessions",
                         sessions.len()
                     );
+                    for (id, s) in sessions.iter() {
+                        sink.emit(crate::event_sink::CallEvent::SessionEnded {
+                            id: *id,
+                            started_at: s.rtc.channel.started_at,
+                            ended_at: std::time::SystemTime::now(),
+                            reason: crate::event_sink::EndReason::Shutdown,
+                            tap_metrics: s.tap_conn.as_ref().map(|c| c.metrics.snapshot()),
+                        });
+                    }
                     sessions.clear();
                     let _ = reply.send(());
                     return;
                 }
+                MediaCmd::Stats { reply } => {
+                    let _ = reply.send(MediaStats {
+                        sessions: sessions.len(),
+                        max_sessions,
+                        draining,
+                        tick_overruns,
+                        last_tick_micros,
+                    });
+                }
+                MediaCmd::Drain { reply } => {
+                    draining = true;
+                    if sessions.is_empty() {
+                        let _ = reply.send(());
+                    } else {
+                        drain_done = Some(reply);
+                    }
+                }
             }
         }
 
@@ -286,6 +403,10 @@ fn run_media_thread(
                     session.rtc.channel.tap = Some(rutster_call_model::TapHandle::new());
                     session.tap_conn = Some(conn);
                     info!(channel_id = %id, "tap engine + reflex + local VAD wired on Connected");
+                    sink.emit(crate::event_sink::CallEvent::SessionConnected {
+                        id: *id,
+                        at: std::time::SystemTime::now(),
+                    });
                     continue;
                 }
             }
@@ -295,12 +416,37 @@ fn run_media_thread(
             }
         }
         for id in closed_ids {
-            sessions.remove(&id);
+            if let Some(s) = sessions.remove(&id) {
+                sink.emit(crate::event_sink::CallEvent::SessionEnded {
+                    id,
+                    started_at: s.rtc.channel.started_at,
+                    ended_at: std::time::SystemTime::now(),
+                    reason: crate::event_sink::EndReason::Closed,
+                    tap_metrics: s.tap_conn.as_ref().map(|c| c.metrics.snapshot()),
+                });
+            }
             debug!(channel_id = %id, "session evicted after close");
         }
 
-        // === Step 3: sleep META_TICK. ===
-        std::thread::sleep(META_TICK);
+        // Drain completion check — after eviction so a tick that closes
+        // the last session completes the drain in the same iteration.
+        if draining && sessions.is_empty() {
+            if let Some(done) = drain_done.take() {
+                let _ = done.send(());
+            }
+        }
+
+        // === Step 3: compensated sleep (review M2). ===
+        // The old fixed sleep(META_TICK) silently stretched the effective
+        // tick to 10ms + work; under load that pushes every session past
+        // its 20ms outbound deadline SIMULTANEOUSLY. Sleep only the
+        // remainder, and count the ticks where there was none to sleep.
+        let elapsed = tick_started.elapsed();
+        last_tick_micros = elapsed.as_micros() as u64;
+        if elapsed >= META_TICK {
+            tick_overruns += 1;
+        }
+        std::thread::sleep(META_TICK.saturating_sub(elapsed));
     }
 }
 
@@ -308,12 +454,156 @@ fn run_media_thread(
 mod tests {
     use super::*;
 
+    #[test]
+    fn register_rejects_when_at_capacity_and_stats_reports() {
+        let rt = tokio::runtime::Runtime::new().unwrap();
+        let handle = rt.handle().clone();
+        let url = url::Url::parse("ws://127.0.0.1:8081/echo").unwrap();
+        let opts = MediaThreadOpts {
+            max_sessions: 1,
+            ..Default::default()
+        };
+        let thread = MediaThread::spawn(url.clone(), opts, handle).expect("spawn");
+
+        let register = |thread: &MediaThread| {
+            let (reply, rx) = oneshot::channel();
+            thread
+                .cmd_tx()
+                .blocking_send(MediaCmd::Register {
+                    tap_url: url::Url::parse("ws://127.0.0.1:1/echo").unwrap(),
+                    reply,
+                })
+                .unwrap();
+            rx.blocking_recv().expect("reply")
+        };
+
+        assert!(register(&thread).is_ok(), "first session fits");
+        let err = register(&thread).expect_err("second must be rejected");
+        assert!(err.contains("node full"), "routes contract string: {err}");
+
+        let (reply, rx) = oneshot::channel();
+        thread
+            .cmd_tx()
+            .blocking_send(MediaCmd::Stats { reply })
+            .unwrap();
+        let stats = rx.blocking_recv().expect("stats");
+        assert_eq!(stats.sessions, 1);
+        assert_eq!(stats.max_sessions, 1);
+        assert!(!stats.draining);
+
+        thread.shutdown();
+    }
+
+    #[test]
+    fn drain_rejects_new_sessions_and_completes_when_map_empties() {
+        let rt = tokio::runtime::Runtime::new().unwrap();
+        let handle = rt.handle().clone();
+        let url = url::Url::parse("ws://127.0.0.1:8081/echo").unwrap();
+        let thread =
+            MediaThread::spawn(url.clone(), MediaThreadOpts::default(), handle).expect("spawn");
+
+        // one live session
+        let (reply, rx) = oneshot::channel();
+        thread
+            .cmd_tx()
+            .blocking_send(MediaCmd::Register {
+                tap_url: url::Url::parse("ws://127.0.0.1:1/echo").unwrap(),
+                reply,
+            })
+            .unwrap();
+        let id = rx.blocking_recv().unwrap().unwrap();
+
+        // drain: must NOT complete while the session lives
+        let (drain_reply, mut drain_rx) = oneshot::channel();
+        thread
+            .cmd_tx()
+            .blocking_send(MediaCmd::Drain { reply: drain_reply })
+            .unwrap();
+        std::thread::sleep(Duration::from_millis(50));
+        assert!(
+            drain_rx.try_recv().is_err(),
+            "drain must wait for in-flight sessions"
+        );
+
+        // new registers are shed with the contract string
+        let (reply, rx) = oneshot::channel();
+        thread
+            .cmd_tx()
+            .blocking_send(MediaCmd::Register {
+                tap_url: url::Url::parse("ws://127.0.0.1:1/echo").unwrap(),
+                reply,
+            })
+            .unwrap();
+        let err = rx.blocking_recv().unwrap().expect_err("draining rejects");
+        assert!(err.contains("draining"), "{err}");
+
+        // deleting the last session completes the drain
+        let (reply, rx) = oneshot::channel();
+        thread
+            .cmd_tx()
+            .blocking_send(MediaCmd::Delete { id, reply })
+            .unwrap();
+        rx.blocking_recv().unwrap();
+        assert!(
+            drain_rx.blocking_recv().is_ok(),
+            "drain completes once the map is empty"
+        );
+
+        thread.shutdown();
+    }
+
+    #[test]
+    fn lifecycle_events_emit_registered_then_ended_deleted() {
+        use crate::event_sink::{CallEvent, EndReason, EventSink};
+        struct TestSink(std::sync::Mutex<Vec<CallEvent>>);
+        impl EventSink for TestSink {
+            fn emit(&self, event: CallEvent) {
+                self.0.lock().unwrap().push(event);
+            }
+        }
+        let sink = std::sync::Arc::new(TestSink(std::sync::Mutex::new(Vec::new())));
+
+        let rt = tokio::runtime::Runtime::new().unwrap();
+        let handle = rt.handle().clone();
+        let url = url::Url::parse("ws://127.0.0.1:8081/echo").unwrap();
+        let opts = MediaThreadOpts {
+            sink: sink.clone(),
+            ..Default::default()
+        };
+        let thread = MediaThread::spawn(url, opts, handle).expect("spawn");
+
+        let (reply, rx) = oneshot::channel();
+        thread
+            .cmd_tx()
+            .blocking_send(MediaCmd::Register {
+                tap_url: url::Url::parse("ws://127.0.0.1:1/echo").unwrap(),
+                reply,
+            })
+            .unwrap();
+        let id = rx.blocking_recv().unwrap().unwrap();
+        let (reply, rx) = oneshot::channel();
+        thread
+            .cmd_tx()
+            .blocking_send(MediaCmd::Delete { id, reply })
+            .unwrap();
+        rx.blocking_recv().unwrap();
+        thread.shutdown();
+
+        let events = sink.0.lock().unwrap();
+        assert!(matches!(&events[0], CallEvent::SessionRegistered { id: eid, .. } if *eid == id));
+        assert!(events.iter().any(|e| matches!(
+            e,
+            CallEvent::SessionEnded { id: eid, reason: EndReason::Deleted, .. } if *eid == id
+        )));
+    }
+
     #[test]
     fn media_thread_register_and_shutdown_round_trips() {
         let rt = tokio::runtime::Runtime::new().unwrap();
         let handle = rt.handle().clone();
         let url = url::Url::parse("ws://127.0.0.1:8081/echo").unwrap();
-        let thread = MediaThread::spawn(url, handle).expect("media thread spawn in test");
+        let thread = MediaThread::spawn(url, MediaThreadOpts::default(), handle)
+            .expect("media thread spawn in test");
         let (reply, rx) = oneshot::channel();
         thread
             .cmd_tx()

crates/rutster/src/routes.rs

@@ -55,6 +55,12 @@ pub async fn create_session(
             (StatusCode::OK, body).into_response()
         }
         Err(e) => {
+            // Contract strings from the media thread (see media_thread.rs):
+            // capacity/drain rejections are retryable-elsewhere → 503 so an
+            // LB retries the next node; everything else stays 500.
+            if e.contains("node full") || e.contains("draining") {
+                return (StatusCode::SERVICE_UNAVAILABLE, e).into_response();
+            }
             tracing::error!(error = ?e, "session create failed");
             StatusCode::INTERNAL_SERVER_ERROR.into_response()
         }
@@ -147,10 +153,55 @@ pub async fn index() -> Response {
         .into_response()
 }
 
+/// GET /healthz — liveness only: the process and HTTP stack are up.
+/// Deliberately does NOT consult the media thread — liveness and
+/// readiness are different probes with different restart semantics.
+pub async fn healthz() -> Response {
+    (StatusCode::OK, "ok").into_response()
+}
+
+/// GET /readyz — "can this node accept a NEW call right now?"
+/// LB target membership + autoscaler both read this. 503 when draining,
+/// at capacity, or when the media thread doesn't answer Stats in 250ms
+/// (a wedged/dead media thread previously left GET / answering 200 —
+/// the zombie-node failure from the 2026-07-04 review).
+pub async fn readyz(State(state): State<AppState>) -> Response {
+    let (reply, rx) = tokio::sync::oneshot::channel();
+    // One timeout bounds the WHOLE round trip — send AND reply. A wedged
+    // media thread can stall the send (full command channel), not just
+    // the reply, and the probe's 250ms promise must hold in that case too.
+    let stats = match tokio::time::timeout(std::time::Duration::from_millis(250), async {
+        state
+            .cmd_tx
+            .send(crate::media_thread::MediaCmd::Stats { reply })
+            .await
+            .ok()?;
+        rx.await.ok()
+    })
+    .await
+    {
+        Ok(Some(s)) => s,
+        // Elapsed timeout, send failure (thread gone), or dropped reply —
+        // all mean the media thread cannot vouch for itself right now.
+        _ => {
+            return (StatusCode::SERVICE_UNAVAILABLE, "media thread unresponsive").into_response();
+        }
+    };
+    let ready = !stats.draining && stats.sessions < stats.max_sessions;
+    let code = if ready {
+        StatusCode::OK
+    } else {
+        StatusCode::SERVICE_UNAVAILABLE
+    };
+    (code, Json(stats)).into_response()
+}
+
 /// Build the axum router.
 pub fn router(state: AppState) -> Router {
     Router::new()
         .route("/", get(index))
+        .route("/healthz", get(healthz))
+        .route("/readyz", get(readyz))
         // `POST /v1/sessions` creates; `DELETE /v1/sessions/:id` destroys
         // (note the `:id` — deleting the collection root has no meaning and
         // would extract a missing `:id` path parameter, so the two routes
@@ -165,6 +216,7 @@ pub fn router(state: AppState) -> Router {
 #[cfg(test)]
 mod tests {
     use super::*;
+    use tower::ServiceExt;
 
     #[test]
     fn ws_loopback_accepted() {
@@ -193,4 +245,31 @@ mod tests {
         let r = resolve_tap_url(None, &default).unwrap();
         assert_eq!(r.as_str(), "ws://127.0.0.1:8081/echo");
     }
+
+    #[tokio::test]
+    async fn readyz_503_when_media_thread_gone_but_healthz_200() {
+        // Default AppState: closed placeholder channel = dead media thread.
+        let app = router(AppState::default());
+        let ready = app
+            .clone()
+            .oneshot(
+                axum::http::Request::builder()
+                    .uri("/readyz")
+                    .body(axum::body::Body::empty())
+                    .unwrap(),
+            )
+            .await
+            .unwrap();
+        assert_eq!(ready.status(), StatusCode::SERVICE_UNAVAILABLE);
+        let health = app
+            .oneshot(
+                axum::http::Request::builder()
+                    .uri("/healthz")
+                    .body(axum::body::Body::empty())
+                    .unwrap(),
+            )
+            .await
+            .unwrap();
+        assert_eq!(health.status(), StatusCode::OK);
+    }
 }

crates/rutster/src/session_map.rs

@@ -80,16 +80,41 @@ impl AppState {
         let _ = rx.await;
     }
 
-    /// Spawn the dedicated media thread for this state.
+    /// Spawn the dedicated media thread with default options.
     ///
-    /// Returns the state with the real command-channel sender wired in,
-    /// plus the `MediaThread` handle. Routes on the returned `AppState`
-    /// will send `MediaCmd` to the running media thread.
+    /// Test-facing convenience: most tests don't care about addressing,
+    /// admission caps, or event sinks, so this skips straight to
+    /// `MediaThreadOpts::default()`. Production always goes through
+    /// [`AppState::spawn_media_thread_with`] instead, so that env-resolved
+    /// config actually reaches the thread. Returns the state with the real
+    /// command-channel sender wired in, plus the `MediaThread` handle;
+    /// routes on the returned `AppState` will send `MediaCmd` to the
+    /// running media thread.
     pub fn spawn_media_thread(
-        mut self,
+        self,
         tokio_handle: tokio::runtime::Handle,
     ) -> Result<(Self, MediaThread), std::io::Error> {
-        let thread = MediaThread::spawn(self.default_tap_url.clone(), tokio_handle)?;
+        self.spawn_media_thread_with(
+            crate::media_thread::MediaThreadOpts::default(),
+            tokio_handle,
+        )
+    }
+
+    /// Spawn the dedicated media thread with explicit options — the
+    /// production path.
+    ///
+    /// Unlike [`AppState::spawn_media_thread`], this threads a caller-built
+    /// `MediaThreadOpts` (media addressing, admission cap, event sink) all
+    /// the way to the media thread, so `main` can resolve those from env
+    /// vars and have them actually take effect. Returns the state with the
+    /// real command-channel sender installed, plus the `MediaThread`
+    /// handle.
+    pub fn spawn_media_thread_with(
+        mut self,
+        opts: crate::media_thread::MediaThreadOpts,
+        tokio_handle: tokio::runtime::Handle,
+    ) -> Result<(Self, MediaThread), std::io::Error> {
+        let thread = MediaThread::spawn(self.default_tap_url.clone(), opts, tokio_handle)?;
         self.cmd_tx = thread.cmd_tx();
         Ok((self, thread))
     }

crates/rutster/src/tap_engine.rs

@@ -111,6 +111,33 @@ pub struct TapConn {
     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).
 ///
@@ -456,6 +483,47 @@ 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();

crates/rutster/tests/api_integration.rs

@@ -7,7 +7,7 @@
 
 use axum::body::Body;
 use axum::http::{Request, StatusCode};
-use rutster::media_thread::MediaThread;
+use rutster::media_thread::{MediaThread, MediaThreadOpts};
 use rutster::session_map::AppState;
 use tower::ServiceExt; // enables `oneshot` on the Router for sync tests
 
@@ -23,8 +23,12 @@ fn app_state_with_thread() -> (AppState, MediaThread) {
     let default = default_tap_url();
     let (placeholder_tx, _placeholder_rx) = tokio::sync::mpsc::channel(1);
     let mut state = AppState::new(placeholder_tx, default.clone());
-    let media_thread =
-        MediaThread::spawn(default, tokio::runtime::Handle::current()).expect("media thread spawn");
+    let media_thread = MediaThread::spawn(
+        default,
+        MediaThreadOpts::default(),
+        tokio::runtime::Handle::current(),
+    )
+    .expect("media thread spawn");
     state.cmd_tx = media_thread.cmd_tx();
     (state, media_thread)
 }
@@ -54,6 +58,31 @@ async fn post_v1_sessions_returns_a_session_id() {
         .unwrap();
 }
 
+#[tokio::test]
+async fn create_session_returns_503_when_node_full() {
+    use rutster::media_thread::MediaThreadOpts;
+    let state = rutster::session_map::AppState::default();
+    let opts = MediaThreadOpts {
+        max_sessions: 0, // a node that can never admit — the LB-shed path
+        ..Default::default()
+    };
+    let (state, _thread) = state
+        .spawn_media_thread_with(opts, tokio::runtime::Handle::current())
+        .expect("spawn");
+    let app = rutster::routes::router(state);
+    let resp = app
+        .oneshot(
+            Request::builder()
+                .method("POST")
+                .uri("/v1/sessions")
+                .body(Body::empty())
+                .unwrap(),
+        )
+        .await
+        .unwrap();
+    assert_eq!(resp.status(), StatusCode::SERVICE_UNAVAILABLE);
+}
+
 #[tokio::test]
 async fn get_root_serves_html() {
     let (state, media_thread) = app_state_with_thread();
@@ -74,3 +103,31 @@ async fn get_root_serves_html() {
         .await
         .unwrap();
 }
+
+#[tokio::test]
+async fn readyz_200_with_stats_json_when_thread_alive() {
+    let state = rutster::session_map::AppState::default();
+    let (state, media_thread) = state
+        .spawn_media_thread(tokio::runtime::Handle::current())
+        .expect("spawn");
+    let app = rutster::routes::router(state);
+    let resp = app
+        .oneshot(
+            Request::builder()
+                .uri("/readyz")
+                .body(Body::empty())
+                .unwrap(),
+        )
+        .await
+        .unwrap();
+    assert_eq!(resp.status(), StatusCode::OK);
+    let body = axum::body::to_bytes(resp.into_body(), 64 * 1024)
+        .await
+        .unwrap();
+    let v: serde_json::Value = serde_json::from_slice(&body).unwrap();
+    assert_eq!(v["draining"], false);
+    assert!(v["max_sessions"].as_u64().unwrap() > 0);
+    tokio::task::spawn_blocking(move || media_thread.shutdown())
+        .await
+        .unwrap();
+}