diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 774c338..5b79bb7 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -41,17 +41,18 @@ jobs: steps: - uses: actions/checkout@v4 - # Seam gate (slice-4 §7 #3 / PORT_PLAN phasing): + # Seam gate (slice-4 §7 #3; re-pinned slice-5): # - # `loop_driver.rs` and `rtc_session.rs` are the hot-path media seam - # preserved from slice-3. They MUST remain byte-identical across - # slice-4 so the reflex wrapper can be added without touching the - # core poll loop. If a legitimate change is needed, update the pinned - # blob hashes below in the same PR — loud and reviewable by design. - - name: Seam gate — loop_driver + rtc_session byte-identical to slice-3 + # `loop_driver.rs` stays byte-identical to slice-3 — the hot-path + # poll loop is untouched. `rtc_session.rs` was re-pinned in slice-5 + # for MediaAddressConfig (2026-07-04 scalability review, B1: + # bind/advertised address split + port range). If a legitimate + # change is needed, update the pinned blob hashes below in the same + # PR — loud and reviewable by design. + - name: Seam gate — loop_driver frozen (slice-3) + rtc_session pinned (slice-5) run: | EXPECTED_LOOP_DRIVER='744bf314edf7f4925c8bb3bd0f5176dbc88f8113' - EXPECTED_RTC_SESSION='a4c9f2ae64e56c08e1990956391514929535b526' + EXPECTED_RTC_SESSION='f47d63b9a2883d37066a93c9daa0e2cf8816bec4' GOT_LOOP_DRIVER=$(git hash-object crates/rutster-media/src/loop_driver.rs) GOT_RTC_SESSION=$(git hash-object crates/rutster-media/src/rtc_session.rs) if [ "$GOT_LOOP_DRIVER" != "$EXPECTED_LOOP_DRIVER" ]; then diff --git a/crates/rutster-call-model/src/lib.rs b/crates/rutster-call-model/src/lib.rs index 6770fd2..7e052f5 100644 --- a/crates/rutster-call-model/src/lib.rs +++ b/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::(), 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, @@ -163,6 +178,7 @@ impl Channel { state: ChannelState::New, direction: Direction::Inbound, created_at: Instant::now(), + started_at: std::time::SystemTime::now(), tap: None, } } diff --git a/crates/rutster-media/src/lib.rs b/crates/rutster-media/src/lib.rs index 4798783..51373b2 100644 --- a/crates/rutster-media/src/lib.rs +++ b/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; diff --git a/crates/rutster-media/src/rtc_session.rs b/crates/rutster-media/src/rtc_session.rs index a4c9f2a..f47d63b 100644 --- a/crates/rutster-media/src/rtc_session.rs +++ b/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, + 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 { + 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, /// 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, @@ -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::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 { + // 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 { - // 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()); + } } diff --git a/crates/rutster-tap/src/protocol.rs b/crates/rutster-tap/src/protocol.rs index 5724ce8..2888d04 100644 --- a/crates/rutster-tap/src/protocol.rs +++ b/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}; diff --git a/crates/rutster/src/config.rs b/crates/rutster/src/config.rs new file mode 100644 index 0000000..96af499 --- /dev/null +++ b/crates/rutster/src/config.rs @@ -0,0 +1,189 @@ +//! # config — pure env-parsing helpers (slice-5) +//! +//! Every knob is a pure function over `Option` / `&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) -> Result { + 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, + advertised_ip: Option, + port_range: Option, +) -> Result { + let mut cfg = MediaAddressConfig::default(); + if let Some(s) = bind_ip { + cfg.bind_ip = s + .parse::() + .map_err(|e| format!("RUTSTER_MEDIA_BIND_IP: {e}"))?; + } + if let Some(s) = advertised_ip { + cfg.advertised_ip = Some( + s.parse::() + .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) -> Result { + 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) -> Result { + match raw { + None => Ok(std::time::Duration::ZERO), + Some(s) => s + .parse::() + .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::().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::().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))); + } +} diff --git a/crates/rutster/src/event_sink.rs b/crates/rutster/src/event_sink.rs new file mode 100644 index 0000000..8eb2f2d --- /dev/null +++ b/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, + }, +} + +/// 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; diff --git a/crates/rutster/src/lib.rs b/crates/rutster/src/lib.rs index 0d89088..5781ac8 100644 --- a/crates/rutster/src/lib.rs +++ b/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; diff --git a/crates/rutster/src/main.rs b/crates/rutster/src/main.rs index b72f0d7..09b93b8 100644 --- a/crates/rutster/src/main.rs +++ b/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() diff --git a/crates/rutster/src/media_thread.rs b/crates/rutster/src/media_thread.rs index 26c7406..ad05721 100644 --- a/crates/rutster/src/media_thread.rs +++ b/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` 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>, @@ -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 }, +} + +/// 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 { 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, default_tap_url: url::Url, + opts: MediaThreadOpts, tokio_handle: tokio::runtime::Handle, media_cmd_tx: mpsc::Sender, ) { let mut sessions: HashMap = 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> = 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>); + 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() diff --git a/crates/rutster/src/routes.rs b/crates/rutster/src/routes.rs index cea1551..3f02131 100644 --- a/crates/rutster/src/routes.rs +++ b/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) -> 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); + } } diff --git a/crates/rutster/src/session_map.rs b/crates/rutster/src/session_map.rs index 05eb55d..7c7265a 100644 --- a/crates/rutster/src/session_map.rs +++ b/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)) } diff --git a/crates/rutster/src/tap_engine.rs b/crates/rutster/src/tap_engine.rs index 94e30fc..6fcb831 100644 --- a/crates/rutster/src/tap_engine.rs +++ b/crates/rutster/src/tap_engine.rs @@ -111,6 +111,33 @@ pub struct TapConn { pub tool_registry: Arc>, } +/// 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(); diff --git a/crates/rutster/tests/api_integration.rs b/crates/rutster/tests/api_integration.rs index 70081b8..0c2cc9e 100644 --- a/crates/rutster/tests/api_integration.rs +++ b/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(); +} diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index 5c27d75..e236fba 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -32,8 +32,9 @@ carrier SIP trunk ─► media termination (RTP/SRTP + local real-time reflexes) ``` **Horizontal platform** concerns are services *around* the core, independently scaled: number -inventory, billing rollup, analytics, multi-region orchestration, the management API, and the agent -brain itself. +inventory, billing rollup, analytics, multi-region orchestration, call placement (the +session→node directory — which node owns which live call, the routing layer above N cores), +the management API, and the agent brain itself. This **replaces the founding three-plane framing**: diff --git a/docs/QUICKSTART.md b/docs/QUICKSTART.md index f866fd2..eb95644 100644 --- a/docs/QUICKSTART.md +++ b/docs/QUICKSTART.md @@ -83,7 +83,7 @@ RUST_LOG=rutster=debug cargo run | Browser shows no mic prompt | Another tab/app holding the mic, or mic permissions disabled for `localhost`. Check browser settings. | | `ICE connection failed` in the browser | Shouldn't happen on loopback (host candidates only). If it does, check the server console for the str0m error. | | Click Start call, nothing happens | Open the browser console (F12). The page logs ICE state + connection state to a `
` element. Look for the failure there. |
-| Port 8080 already in use | Another process holding the port. Either stop it or edit `crates/rutster/src/main.rs` to bind a different port. |
+| Port 8080 already in use | Set `RUTSTER_HTTP_BIND`, e.g. `RUTSTER_HTTP_BIND=0.0.0.0:8090 cargo run -p rutster` |
 
 The browser test page at `GET /` is a single self-contained HTML file
 with inline JS — no build step. View source to see exactly what the
diff --git a/docs/adr/0005-event-bus.md b/docs/adr/0005-event-bus.md
index 8f0d94a..460c586 100644
--- a/docs/adr/0005-event-bus.md
+++ b/docs/adr/0005-event-bus.md
@@ -53,6 +53,10 @@ Recommending "Redis" undercuts the data-ownership pillar ([ADR-0002](0002-north-
    persistence (RDB/AOF) is async-ish — fine for transport / replay / fleeting retention, **wrong**
    for "the CDR that proves what we billed." CDR and recordings emit **durably to object storage**
    in their own services; the bus only *flows* events into that pipeline and lets services *react*.
+3. **Constraint 2 governs the durable CDR, not enforcement counters.** The spend gate's
+   live accounting state (spend/pacing/rate counters) is ephemeral control state and DOES
+   belong in Valkey KV at N>1 — see the
+   [ADR-0009 amendment 2026-07-04](0009-spend-gate-honest-rescope.md#amendment-2026-07-04--enforcement-locality--accounting-locality).
 
 ## Consequences
 
diff --git a/docs/adr/0009-spend-gate-honest-rescope.md b/docs/adr/0009-spend-gate-honest-rescope.md
index ff7e16a..8e37c8f 100644
--- a/docs/adr/0009-spend-gate-honest-rescope.md
+++ b/docs/adr/0009-spend-gate-honest-rescope.md
@@ -82,6 +82,32 @@ security-constitutive), but its guarantees are restated precisely:
   [market feature scan F5](../reviews/2026-07-03-market-feature-scan.md), since integrator unit
   economics run on cost-per-contained-call).
 
+## Amendment 2026-07-04 — enforcement locality ≠ accounting locality
+
+- **Status:** Proposed (pending maintainer ratification)
+- **Origin:** [2026-07-04 scalability & infra-fit review](../reviews/2026-07-04-scalability-infra-review.md),
+  finding M5.
+
+Guarantee 2 above prescribes where the **check** sits — in-process with the tap and the
+provider call-control client. It is silent on where the **accounting state** lives, and at
+N>1 core instances that silence becomes a correctness trap: per-instance counters make
+every spend/pacing cap silently N× the fleet intent, and toll-fraud thresholds never trip
+because attempts spread across instances.
+
+Clarification, binding on the step-6 implementation:
+
+1. **Enforcement is in-process** (unchanged — constitutive; guarantee 2 stands).
+2. **Accounting is shared.** The gate is built against a ledger trait with atomic
+   check-and-reserve semantics from day one: an in-memory implementation for single-node,
+   a Valkey-backed one ([ADR-0005](0005-event-bus.md)) for fleets. The check path never
+   assumes counter locality.
+3. **ADR-0005 constraint 2 is not a counter-argument.** "The bus is not the source of
+   truth for billing- or call-loss-critical state" governs the *durable CDR*. Live
+   enforcement counters (spend/pacing/rate state) are ephemeral control state — exactly
+   what Valkey KV is for. Losing them on a Valkey restart degrades fail-safe (re-count
+   from zero, provider-side caps as the backstop per this ADR's deployment guidance) —
+   not to billing corruption.
+
 ## References
 
 - [ADR-0002](0002-north-star-and-fused-core.md) — the pillar this amends
diff --git a/docs/reviews/2026-07-04-scalability-infra-review.md b/docs/reviews/2026-07-04-scalability-infra-review.md
new file mode 100644
index 0000000..40b4971
--- /dev/null
+++ b/docs/reviews/2026-07-04-scalability-infra-review.md
@@ -0,0 +1,249 @@
+# 2026-07-04 — Scalability & infra-fit review
+
+**Scope:** what single-process assumptions are baked into the code today that will fight
+horizontal scaling, autoscaling, and load balancing later — and how rutster's shape maps onto
+infra overall. **Not in scope:** the fused per-call vertical itself (deliberate, ADR-0002) or
+documented later-rung deferrals with clean seams.
+
+**Method:** 7-lens multi-agent survey over every crate (43 raw findings), dedup (17), adversarial
+per-finding verification (each verifier read the cited code with a default-skeptical stance),
+plus first-hand re-verification of 5 findings whose verifiers were lost to a session limit.
+2 findings were rejected outright (one factually wrong, one a documented deferral); several were
+downgraded. Severity: **blocker** = correctness breaks or scaling impossible at N>1 (or even N=1
+in cloud); **major** = missing seam that calcifies as code accretes; **minor** = trivial when the
+time comes.
+
+---
+
+## The infra model this system actually implies
+
+Before the findings: the frame. rutster's fused vertical means **a call is pinned to a process
+for its whole life** — DTLS/ICE state, jitter buffer, reflex state, the playout ring. Calls are
+non-migratable, same as they were on Asterisk. That is not a defect; it defines the scaling model:
+
+1. **Placement, not load balancing.** New calls get *placed* on a node (least-loaded /
+   bin-packed); every subsequent operation on that call must reach the *owning* node. A
+   round-robin L7 LB over the REST surface is the wrong mental model and breaks at N=2 today.
+   The proven patterns: (a) a shared session→node directory (Valkey — already ratified,
+   ADR-0005) behind a thin routing/API tier, or (b) the create-response carries a node-addressed
+   URL (the LiveKit/Janus pattern). The CPaaS trunk path (ADR-0007 layer 1) converges on the
+   same decision point: Twilio/Telnyx hand the media-stream URL out **per call at answer time**,
+   so trunk dispatch and API routing are the *same* placement problem with the same answer.
+
+2. **Capacity is tick-budget, and it must be measured to be scheduled.** The calls-per-node
+   bound is the media thread exhausting its 10ms meta-tick. CPU% is the wrong autoscaling
+   signal for a real-time engine (it lags and it lies); the right signals are
+   **calls-in-flight** and **tick-overrun rate** (deadline misses). Neither exists today.
+   Convergence worth exploiting: ADR-0010's pulled-forward benchmark/sim harness needs exactly
+   this gauge — its primary readout should be "max sessions at 1 correctness traps** are worse than the loud ones: a DELETE that 204s on the
+   wrong node while the call (and its spend) keeps running; per-instance spend caps that
+   quietly become N× the fleet cap; CDRs that evaporate with an OOM-kill. These bite after
+   things *appear* to work.
+
+---
+
+## Blockers
+
+### B1. Media addressing: loopback bind, no advertised-address concept, no port range
+`crates/rutster-media/src/rtc_session.rs:179` binds `127.0.0.1:0`; `accept_offer`
+(rtc_session.rs:236) advertises the raw local socket as the **sole** ICE host candidate. No
+srflx/STUN, no NAT 1:1 advertised-IP config, no port-range allocator; `RtcSession::new()` is
+zero-arg, and nothing in the chain main → `MediaThread::spawn` → `MediaCmd::Register` can carry
+addressing config. The file is CI-seam-frozen (ci.yml:51–64 — a deliberate, updatable speed
+bump, but it means the seam change is a "loud" PR by design).
+**Why blocker:** RTP cannot ride the HTTP LB; even a single cloud instance behind NAT is
+unreachable — the SDP answer says `127.0.0.1:`. Fleet firewalling needs a bounded
+port range. *(Adversarially verified: CONFIRMED, structurally baked in — str0m rejects 0.0.0.0
+candidates, so this needs a bind-vs-advertised split, a new concept, not a constant swap.)*
+**Fix shape:** `MediaAddressConfig { bind_interface, advertised_address, udp_port_range }`
+threaded through the construction chain; advertised addr feeds `Candidate::host`, bind addr
+feeds the socket; update the seam-gate hashes in the same PR.
+
+### B2. Session placement: ownership is process-local, with a silent-failure DELETE
+The only registry of live calls is the media thread's private
+`HashMap` (media_thread.rs:164). Routes reach it over a
+process-local mpsc. ADR-0005's Valkey has **zero code presence** — no dep, no crate, no trait.
+At N=2 behind any LB:
+- `POST /v1/sessions/:id/offer` on a non-owning node → "not found" → **404** (loud failure);
+- `DELETE /v1/sessions/:id` on a non-owning node → `sessions.remove` finds nothing, reply is
+  `()` regardless, route returns **204 unconditionally** (routes.rs:131–138) — the hangup
+  *reports success while the call keeps running*, burning trunk minutes and brain tokens. Once
+  the spend gate exists, "kill this runaway call" failing silently is a security-posture hole,
+  not just a routing bug.
+*(Verified first-hand after the workflow's verifier was lost to a session limit; every line
+re-checked.)*
+**Fix shape:** session directory (ChannelId → node) written at Register/Delete — Valkey per
+ADR-0005 is the obvious backend — plus one of: routing tier, consistent-hash LB, or
+node-addressed URLs returned at create. The `MediaCmd` seam gives the writes a clean landing
+spot; the missing piece is the *concept*, which also appears nowhere in ARCHITECTURE.md's
+horizontal-platform list.
+
+## Major
+
+### M1. No graceful drain — SIGTERM hard-drops every in-flight call
+`MediaCmd::Shutdown` is `sessions.clear(); return` (media_thread.rs:224–232) — it even skips
+the 750ms per-session tap teardown that `Delete` performs. axum's graceful shutdown drains
+HTTP requests; calls are not HTTP requests. main.rs:55–59 says it outright: "No in-flight call
+preservation story in the dev loop." *(CONFIRMED; not yet structurally baked — the fix is
+contained plumbing today — but it must land before the planned threadpool-shard graduation of
+this exact thread, or the two-state running/dead lifecycle gets baked into the shards, the
+readiness surface, and the trunk routing.)*
+**Fix shape:** a `Drain` lifecycle state: reject new Registers (503), flip readiness off, keep
+ticking until the map empties or a deadline, then Shutdown.
+
+### M2. No admission control, no capacity signal
+Register inserts unconditionally; the fixed `sleep(META_TICK)` (media_thread.rs:303) ignores
+how long the tick took, so saturation silently stretches the tick and degrades **every** call's
+20ms pacing at once — the worst overload mode for a real-time engine, and invisible: the
+session count surfaces exactly once, in the shutdown log. A node can never say "full" and an
+autoscaler has nothing to read. *(CONFIRMED. Notably: this repo documents its deferrals
+meticulously, and this one is nowhere documented — a genuinely silent gap. ADR-0010's benchmark
+needs the tick-lag gauge to produce its headline number.)*
+**Fix shape:** configurable max-sessions in Register (Err → 503), per-tick elapsed-vs-budget
+gauge, `MediaCmd::Stats { reply }` for the readiness/metrics surface.
+
+### M3. ADR-0005 exists only on paper — lifecycle events die with the process
+No bus, no `EventSink` trait, nothing durable: register/Connected/evict/Shutdown are tracing
+lines; the `Delete` handler drops `TapConn` without ever reading `conn.metrics`, breaking the
+in-code promise at tap_engine.rs:82–84 ("the eventual CDR/ringbus emitter"); `Channel` carries
+only a **monotonic `Instant`** — there is no wall-clock timestamp in the entire production
+workspace from which a CDR could even be built. An OOM-killed node erases all evidence of its
+calls. *(CONFIRMED, leaning baked-in: the "tracing is the record" assumption is reproducing
+slice-over-slice, and thread-shard graduation will multiply the emission points.)*
+**Fix shape:** minimal `EventSink` trait owned by the media thread (log-backed now, Valkey
+streams later), CDR-shaped started/ended events with wall-clock time + metrics snapshots.
+
+### M4. Tap protocol has no resume or terminal semantics — the calcifying one
+Brain bye, error, and stream-end all funnel into infinite re-dial (5s cap, forever); every
+reconnect restarts `seq_egress = 0` with a hello carrying no epoch/resume token; the reference
+brain acks any hello and opens a fresh OpenAI session. Mid-call reconnect = silently amnesiac
+brain **today, even at N=1**; a brain that deliberately ends a session gets re-dialed for the
+rest of the call. At brain-fleet scale: reconnects land on different instances with no
+protocol-level way to detect lost context, plus re-dial storms. *(CONFIRMED, baked in — this is
+a **wire-protocol** change, v2 + brain-side changes, the most expensive kind of retrofit. The
+protocol being versioned is the one mercy.)*
+**Fix shape:** resume token/connection epoch in hello + resume-ack/reject from the brain +
+terminal-vs-transient bye reason that exits the retry loop.
+
+### M5. Spend-gate accounting locality is undefined — and one ADR sentence steers it wrong
+The crate is a stub (nothing baked yet), but ADR-0009 prescribes where the *check* sits without
+distinguishing enforcement point from **accounting ledger**. Per-instance counters on a fleet =
+N× every cap, and toll-fraud thresholds that never trip because attempts spread across nodes.
+The subtle trap found in verification: ADR-0005's "the bus is NOT the source of truth for
+billing-critical state" is the sentence most likely to push the step-6 implementer toward local
+counters — but enforcement counters (rate/spend state) are not billing truth (durable CDR);
+they're different things with different stores. *(WEAK/major-borderline — nothing accretes yet;
+the fix today is one sentence.)*
+**Fix shape:** amend ADR-0009 now: "in-process enforcement, **shared accounting**" — gate built
+against a ledger trait with atomic check-and-reserve (in-memory impl for N=1, Valkey for N>1).
+
+### M6. Metrics are write-only — nothing aggregates, nothing exports
+`TapMetrics`/`ReflexMetrics` are per-session atomics with a snapshot method that production
+code never calls; no process-level registry, no calls-in-flight gauge, no Prometheus/OTel
+export. An LB health check and an autoscaler would both be reading a static HTML page today.
+*(Verified first-hand. The atomics+snapshot shape is export-friendly — the missing piece is
+the aggregation registry and one scrape endpoint, medium plumbing.)*
+
+### M7. Session teardown stalls every live call on the node
+`MediaCmd::Delete` runs `tokio_handle.block_on(timeout(750ms, &mut conn.join))`
+(media_thread.rs:205–217) **on the media thread, inside the tick loop**. One teardown with a
+slow/unresponsive brain freezes the 20ms loop for every other call on the node — up to ~37
+missed frames each; several Deletes drained in one batch stack sequentially. Under fleet churn
+(hangups are constant in a call center) this is a per-node isolation failure that worsens with
+density. *(Verified first-hand.)*
+**Fix shape:** hand teardown to a tokio task (fire close_tx, spawn the bounded join await);
+the tick loop should never block on brain I/O — same discipline the tap pipe already follows.
+
+## Minor
+
+- **Readiness/liveness absent** (routes.rs:151): only probe-able route is `GET /` static HTML —
+  returns 200 while the media thread is dead (its panic isn't monitored; sessions just 500).
+  Downgraded to minor by verification: the `MediaCmd` seam makes `/healthz` + `/readyz` trivial
+  additions, nothing accretes around their absence. Lands naturally with M1/M2.
+- **Trunk dispatch** (rutster-trunk stub): the "which instance gets this call's media fork"
+  question. Downgraded: Twilio/Telnyx hand out the stream URL per call by design, so answer-time
+  binding is the *default* usage — this needs one design constraint written into the step-5
+  spec, not code today. It is, however, the same placement concept as B2 — solve once.
+- **Tap URL loopback-only + wss:// not compiled** (routes.rs:87, workspace Cargo.toml:48):
+  documented step-6 deferral with the per-session override seam already in place. Non-obvious
+  detail from verification: no crate enables a TLS feature on tokio-tungstenite, so relaxing the
+  validator alone wouldn't help — the dial capability itself is compiled out, and the env
+  default goes through the same validator (no escape hatch).
+- **HTTP bind hardcoded** `0.0.0.0:8080` (main.rs:44); QUICKSTART literally says "edit main.rs
+  to change the port." `RUTSTER_TAP_BIND` in the brain binary is the pattern to copy.
+- **Brain endpoint static per-call URL** (tap_engine.rs:271): downgraded on verification —
+  `connect_async` re-resolves DNS each dial, so a brain fleet behind DNS/VIP works today; the
+  real calcification is M4's missing resume semantics, not URL indirection.
+
+## Rejected in verification (for the record)
+
+- *"EnvFilter `rutster=info` silences library-crate warnings"* — factually wrong;
+  tracing-subscriber matches by string prefix, so `rutster` covers `rutster_media::*` et al.
+  (verified empirically against the locked 0.3.23).
+- *"No packaging/compose artifact"* — documented later rung (DEVELOPMENT.md:172), purely
+  additive, nothing accretes around its absence.
+
+## What's genuinely clean (don't break these)
+
+- **`ChannelId` = UUIDv4** — globally unique, fleet-safe, doubles as the wire session id.
+- **The `MediaCmd` command channel** — the load-bearing seam. Drain, stats, admission,
+  directory writes, and event emission all have a natural landing spot because of it.
+- **Bounded backpressure with drop-counting on both tap directions** (32-frame mpscs, 5-frame
+  playout ring, drop-oldest, every drop counted) — a slow brain cannot eat unbounded memory.
+- **Per-call `tap_url` override** — per-call brain routing already exists.
+- **Versioned tap protocol** (`v:1` enforced at decode) — M4 has an evolution path.
+- **Zero production global statics** in the workspace.
+- **Tap is outbound from the call-owning node** — connection follows call ownership; exactly
+  right for multi-node.
+
+## Sequencing — what to bake in now vs. later
+
+The point is not to build the fleet now. It's that a handful of *concepts* are cheap to plant
+today and expensive to retrofit once slices accrete around their absence:
+
+**Now / next slice (cheap seams, prevents calcification):**
+1. `MediaAddressConfig` (B1) — also unblocks the first cloud demo at N=1.
+2. Admission cap + tick-lag gauge + `MediaCmd::Stats` (M2) — double-billed to ADR-0010's
+   benchmark harness, which needs the same instrumentation.
+3. Drain vocabulary (M1) + `/healthz` + `/readyz` — before thread-sharding.
+4. `EventSink` trait + wall-clock timestamps on `Channel` (M3).
+5. One-sentence ADR-0009 amendment: in-process enforcement, shared accounting (M5); clarify
+   ADR-0005's "not source of truth" ≠ "no enforcement counters in Valkey."
+6. Reserve resume-token + terminal-bye in the tap protocol's v2 plan (M4) — wire semantics
+   are the most calcifying surface in the whole system.
+7. Move Delete teardown off the tick loop (M7). Trivial, and it's a latency bug today.
+8. `RUTSTER_HTTP_BIND` (copy the existing pattern).
+
+**When N>1 actually lands (design then, not now — but write the concepts down):**
+- Valkey session directory + placement/routing tier (B2) — add "call placement" to
+  ARCHITECTURE.md's horizontal-platform list so the concept exists on paper.
+- Trunk step-5 spec constraint: media-stream URL handed out per call at answer time (placement
+  decision shared with B2).
+- Metrics aggregation registry + scrape endpoint (M6).
+- TLS on tap dial + validator policy (documented step 6).
+
+---
+*Method note: multi-agent review (7 survey lenses → dedup → adversarial verify → severity
+calibration), ~1M tokens of subagent reading; 5 of 17 findings re-verified by hand after their
+verifiers hit a session limit. All file:line citations checked against working tree @ d696536.*
diff --git a/docs/superpowers/plans/2026-07-04-slice-5-scalability-seams.md b/docs/superpowers/plans/2026-07-04-slice-5-scalability-seams.md
new file mode 100644
index 0000000..1ad258d
--- /dev/null
+++ b/docs/superpowers/plans/2026-07-04-slice-5-scalability-seams.md
@@ -0,0 +1,1829 @@
+# Slice 5 — Scalability Seams — Implementation Plan
+
+> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.
+
+**Goal:** Plant the horizontal-scaling seams identified in the
+[2026-07-04 scalability & infra-fit review](../../reviews/2026-07-04-scalability-infra-review.md)
+— media addressing (B1), admission/capacity signals (M2), drain lifecycle (M1), event
+sink + wall-clock CDR fields (M3), non-blocking teardown (M7), health/readiness probes,
+config surface, and the ADR-0009 accounting amendment — **before** further slices accrete
+around their absence.
+
+**Naming note:** this is *not* spearhead step 5 (rented-transport trunk). It is an
+infra-seams slice that pre-paves step 4½ (ADR-0010 benchmark — Task 3's tick-lag gauge is
+its primary readout) and step 5 (the trunk adapter inherits the advertised-address +
+placement concepts).
+
+**Architecture:** No new services, no Valkey yet, no fleet. Every change is a seam *inside
+the existing fused vertical*: a config struct where a literal was, an enum variant on the
+existing `MediaCmd` command channel, a trait where tracing lines were. Single-node behavior
+with default config is byte-for-byte-equivalent-or-better; the deliverable is that N>1,
+NAT, and autoscaling stop being *design changes* and become *plumbing*.
+
+**Tech Stack:** existing workspace only — Rust stable + 1.85, `str0m` 0.21, `axum` 0.7,
+`tokio`, `std::thread` media loop, `serde`. **No new dependencies.**
+
+## Global Constraints
+
+- **License:** GPL-3.0-or-later on every crate manifest (ADR-0004).
+- **DCO:** every commit signed off — `git commit -s` (AGENTS.md Git workflow). Signoff
+  identity = the human maintainer's git config, not the agent.
+- **Seam gate (CHANGED THIS SLICE):** `loop_driver.rs` stays **byte-identical**
+  (hash `744bf314edf7f4925c8bb3bd0f5176dbc88f8113`). `rtc_session.rs` is **re-pinned by
+  Task 2 only** — Task 2 updates `EXPECTED_RTC_SESSION` in `.github/workflows/ci.yml` in
+  the same commit as the code change. **No other task may touch either file.**
+- **Hot-path policy:** never `?`-propagate on the 20 ms loop; match-and-continue;
+  "drop + observe." No `unwrap()`/`expect()` outside tests/const-init/cold-startup.
+- **Learner-facing comments:** this project OVERRIDES the no-comments default. Every new
+  public item gets `///` docs; every new module gets `//!` docs citing the review finding
+  it closes (e.g. "closes 2026-07-04 review B1"). Snippets below show the load-bearing
+  comments; implementers keep that density.
+- **Cold-path error strings are a wire contract within the binary:** routes match
+  `e.contains("not found")` today; Tasks 3–4 add `"node full"` and `"draining"` to that
+  contract. Don't reword one side without the other.
+- **Env naming:** `RUTSTER_*`, parsed by pure functions in `crates/rutster/src/config.rs`
+  (testable without env mutation — take `Option`/`&str` inputs).
+- **CI gates:** `cargo fmt --check`, `cargo clippy --all -- -D warnings`,
+  `cargo test --all` (stable + 1.85), `cargo deny check`.
+- **Branch/PR:** branch `slice-5/scalability-seams` off `main`; PR via `tea` (not `gh`).
+
+## File Structure
+
+### New files
+
+| Path | Responsibility |
+|---|---|
+| `crates/rutster/src/config.rs` | Pure env-parsing helpers: `http_bind`, `parse_port_range`, `media_address_config`, `max_sessions`, `drain_deadline`. |
+| `crates/rutster/src/event_sink.rs` | `EventSink` trait + `CallEvent`/`EndReason` + `TracingEventSink` (log-backed impl; Valkey impl lands with ADR-0005 wiring, later rung). |
+
+### Modified files
+
+| Path | What changes |
+|---|---|
+| `crates/rutster/src/main.rs` | Env-driven HTTP bind; builds `MediaThreadOpts` from env; two-phase shutdown (drain → deadline → hard stop). |
+| `crates/rutster/src/lib.rs` | `pub mod config;` + `pub mod event_sink;`. |
+| `crates/rutster-media/src/rtc_session.rs` | **Task 2 ONLY.** `MediaAddressConfig`, `new_with_config`, `bind_in_range`, `advertised_addr` field, candidate uses advertised addr. Re-pin CI hash. |
+| `crates/rutster-media/src/lib.rs` | Re-export `MediaAddressConfig`. |
+| `crates/rutster/src/media_thread.rs` | `MediaThreadOpts`; admission cap; tick-lag gauge + compensated sleep; `MediaCmd::{Stats, Drain}`; non-blocking Delete teardown; `EventSink` emissions. (Tasks 2, 3, 4, 6, 7 — **serial**.) |
+| `crates/rutster/src/session_map.rs` | `spawn_media_thread_with(opts, …)`; existing `spawn_media_thread` delegates with defaults (test call sites unchanged). |
+| `crates/rutster/src/routes.rs` | 503 mapping for `"node full"`/`"draining"`; `GET /healthz`; `GET /readyz`. |
+| `crates/rutster/src/tap_engine.rs` | `spawn_tap_teardown` free function (moved out of the tick loop). |
+| `crates/rutster-call-model/src/lib.rs` | `Channel.started_at: SystemTime` (wall-clock CDR anchor; `created_at: Instant` stays for idle math). |
+| `crates/rutster/tests/api_integration.rs` | 503-when-full + healthz/readyz integration tests. |
+| `.github/workflows/ci.yml` | Task 2 re-pins `EXPECTED_RTC_SESSION`. |
+| `docs/QUICKSTART.md` | Env-var table replaces "edit main.rs to change the port". |
+| `docs/adr/0009-spend-gate-honest-rescope.md` | Amendment 2026-07-04: enforcement locality vs accounting locality. |
+| `docs/adr/0005-event-bus.md` | One-line cross-ref under Constraints. |
+| `docs/ARCHITECTURE.md` | "call placement (session→node directory)" added to the horizontal-platform list. |
+| `crates/rutster-tap/src/protocol.rs` | Module-doc "v2 reservations" section (resume token, terminal bye). Doc-only. |
+
+### SEAM-INVARIANT files
+
+- `crates/rutster-media/src/loop_driver.rs` — **byte-identical, all tasks.**
+- `crates/rutster-media/src/rtc_session.rs` — untouchable by every task except Task 2.
+
+## Task ordering (for multi-agent dispatch)
+
+`media_thread.rs` is the contention file — Tasks 3 → 4 → 6 → 7 are **strictly serial** on
+one dev. Everything else parallelizes:
+
+- **Task 1** — config.rs foundation + `RUTSTER_HTTP_BIND`. Land first (Task 2 extends config.rs).
+- **Task 2** — `MediaAddressConfig` + seam re-pin. Depends on Task 1 (config.rs exists). Parallel with 3.
+- **Task 3** — admission cap + tick-lag + `MediaCmd::Stats`. Depends on Task 2 (`MediaThreadOpts` exists — Task 2 introduces it).
+- **Task 4** — drain lifecycle. Depends on Task 3 (extends `MediaStats` with `draining`).
+- **Task 5** — healthz/readyz. Depends on Tasks 3+4.
+- **Task 6** — non-blocking tap teardown. Depends on Task 4 (same file; serial).
+- **Task 7** — `EventSink` + wall-clock. Depends on Task 6 (same file; serial).
+- **Task 8** — ADR-0009 amendment + ADR-0005 note + ARCHITECTURE.md line. Parallel-safe anytime.
+- **Task 9** — tap protocol v2 reservations doc. Parallel-safe anytime.
+
+Parallelizable-now filler: QUICKSTART env table (after Task 1), LEARNING.md pointers to
+`config.rs`/`event_sink.rs` (after Tasks 1/7), `cargo doc` render check (after Task 7).
+
+---
+
+### Task 1: `config.rs` + `RUTSTER_HTTP_BIND`
+
+**Files:**
+- Create: `crates/rutster/src/config.rs`
+- Modify: `crates/rutster/src/lib.rs` (add `pub mod config;`)
+- Modify: `crates/rutster/src/main.rs:44` (replace hardcoded bind)
+- Modify: `docs/QUICKSTART.md` (env table; drop "edit main.rs" advice)
+- Test: inline `#[cfg(test)]` in `config.rs`
+
+**Interfaces:**
+- Consumes: nothing new.
+- Produces: `pub fn http_bind(raw: Option) -> Result`
+  (later tasks add more parsers to this module).
+
+- [ ] **Step 1: Write the failing tests**
+
+Create `crates/rutster/src/config.rs`:
+
+```rust
+//! # config — pure env-parsing helpers (slice-5)
+//!
+//! Every knob is a pure function over `Option` / `&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::SocketAddr;
+
+/// 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) -> Result {
+    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}")),
+    }
+}
+
+#[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::().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::().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"));
+    }
+}
+```
+
+- [ ] **Step 2: Run tests to verify they fail (module not declared)**
+
+Run: `cargo test -p rutster config::`
+Expected: compile error — `config` not in `lib.rs` yet. Add to `crates/rutster/src/lib.rs`:
+
+```rust
+pub mod config;
+```
+
+- [ ] **Step 3: Run tests to verify they pass**
+
+Run: `cargo test -p rutster config::`
+Expected: `3 passed`
+
+- [ ] **Step 4: Wire into `main.rs`**
+
+Replace (main.rs:44):
+
+```rust
+    let addr: SocketAddr = "0.0.0.0:8080".parse().expect("valid addr");
+```
+
+with:
+
+```rust
+    // 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");
+```
+
+(`use std::net::SocketAddr;` already imported at main.rs:14.)
+
+- [ ] **Step 5: Update QUICKSTART**
+
+In `docs/QUICKSTART.md`, replace the troubleshooting row "edit `crates/rutster/src/main.rs`
+to bind a different port" with:
+
+```markdown
+| Port 8080 already in use | Set `RUTSTER_HTTP_BIND`, e.g. `RUTSTER_HTTP_BIND=0.0.0.0:8090 cargo run -p rutster` |
+```
+
+- [ ] **Step 6: Full check + commit**
+
+Run: `cargo fmt && cargo clippy --all -- -D warnings && cargo test -p rutster`
+Expected: clean, all tests pass.
+
+```bash
+git add crates/rutster/src/config.rs crates/rutster/src/lib.rs crates/rutster/src/main.rs docs/QUICKSTART.md
+git commit -s -m "slice-5: RUTSTER_HTTP_BIND env config via new config module"
+```
+
+---
+
+### Task 2: `MediaAddressConfig` — bind/advertised address + port range (review B1)
+
+**Files:**
+- Modify: `crates/rutster-media/src/rtc_session.rs` (**the seam-frozen file — this task re-pins it**)
+- Modify: `crates/rutster-media/src/lib.rs` (re-export)
+- Modify: `crates/rutster/src/media_thread.rs` (introduce `MediaThreadOpts`; Register uses config)
+- Modify: `crates/rutster/src/session_map.rs` (add `spawn_media_thread_with`)
+- Modify: `crates/rutster/src/config.rs` (parsers), `crates/rutster/src/main.rs` (env wiring)
+- Modify: `.github/workflows/ci.yml:54` (`EXPECTED_RTC_SESSION` re-pin, same commit)
+- Test: inline tests in `rtc_session.rs` + `config.rs`
+
+**Interfaces:**
+- Consumes: `config.rs` module (Task 1).
+- Produces:
+  - `pub struct MediaAddressConfig { pub bind_ip: IpAddr, pub advertised_ip: Option, pub port_range: Option<(u16, u16)> }` + manual `Default` (loopback/None/None = today's behavior), re-exported from `rutster_media`.
+  - `RtcSession::new_with_config(cfg: &MediaAddressConfig) -> Result`; `RtcSession::new()` delegates with defaults.
+  - `pub struct MediaThreadOpts { pub media_cfg: MediaAddressConfig }` + manual `Default` (in `media_thread.rs`; Tasks 3/7 add fields — **manual** `Default`, no derive, so later `Arc` fits).
+  - `MediaThread::spawn(default_tap_url: url::Url, opts: MediaThreadOpts, tokio_handle: tokio::runtime::Handle)` (signature change; existing test call sites updated in this task).
+  - `AppState::spawn_media_thread_with(self, opts: MediaThreadOpts, tokio_handle) -> Result<(Self, MediaThread), std::io::Error>`; existing `spawn_media_thread` delegates with `MediaThreadOpts::default()` so integration tests don't churn.
+  - `config::parse_port_range(&str) -> Result<(u16, u16), String>`, `config::media_address_config(bind_ip, advertised_ip, port_range: Option ×3) -> Result`.
+  - Env: `RUTSTER_MEDIA_BIND_IP` (default `127.0.0.1`), `RUTSTER_MEDIA_ADVERTISED_IP` (default: bind IP), `RUTSTER_MEDIA_PORT_RANGE` (`"lo-hi"` inclusive; default: OS-ephemeral).
+
+- [ ] **Step 1: Write failing tests in `rtc_session.rs`** (append inside the existing `#[cfg(test)] mod tests`)
+
+```rust
+    #[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());
+    }
+```
+
+- [ ] **Step 2: Run to verify failure**
+
+Run: `cargo test -p rutster-media rtc_session`
+Expected: compile error — `MediaAddressConfig` / `new_with_config` undefined.
+
+- [ ] **Step 3: Implement in `rtc_session.rs`**
+
+Add after the `RtcSessionError` enum (below line 55):
+
+```rust
+use std::net::{IpAddr, SocketAddr, UdpSocket};
+
+/// 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,
+    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 {
+    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")
+    }))
+}
+```
+
+Change the struct: add a field after `local_addr` (line 88):
+
+```rust
+    /// 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,
+```
+
+Replace `new()` / `new_internal()` (lines 116–199):
+
+```rust
+    pub fn new() -> Result {
+        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 {
+        // 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,
+        })
+    }
+```
+
+In `accept_offer` (line 236) change the candidate to the advertised address:
+
+```rust
+        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.
+```
+
+Keep the old `use std::net::SocketAddr;` import (line 20) — fold into the new `use` line.
+In `crates/rutster-media/src/lib.rs` add:
+
+```rust
+pub use rtc_session::MediaAddressConfig;
+```
+
+- [ ] **Step 4: Run media tests**
+
+Run: `cargo test -p rutster-media`
+Expected: all pass, including the four new tests.
+
+- [ ] **Step 5: Introduce `MediaThreadOpts` and thread the config (binary crate)**
+
+In `media_thread.rs`, after `CMD_CHANNEL_CAPACITY` (line 39):
+
+```rust
+/// Knobs the binary resolves from env and hands to the media thread.
+/// Manual Default (not derive) because later slices add non-derivable
+/// fields (Task 7's `Arc`). Defaults reproduce the
+/// pre-slice-5 single-node dev behavior exactly.
+pub struct MediaThreadOpts {
+    pub media_cfg: rutster_media::MediaAddressConfig,
+}
+
+impl Default for MediaThreadOpts {
+    fn default() -> Self {
+        Self {
+            media_cfg: rutster_media::MediaAddressConfig::default(),
+        }
+    }
+}
+```
+
+`MediaThread::spawn` (line 96) gains the param and forwards it:
+
+```rust
+    pub fn spawn(
+        default_tap_url: url::Url,
+        opts: MediaThreadOpts,
+        tokio_handle: tokio::runtime::Handle,
+    ) -> Result {
+        let (cmd_tx, cmd_rx) = mpsc::channel(CMD_CHANNEL_CAPACITY);
+        let cmd_tx_for_thread = cmd_tx.clone();
+        let join = std::thread::Builder::new()
+            .name("rutster-media".into())
+            .spawn(move || {
+                run_media_thread(cmd_rx, default_tap_url, opts, tokio_handle, cmd_tx_for_thread);
+            })?;
+        Ok(Self {
+            cmd_tx,
+            join: Some(join),
+        })
+    }
+```
+
+`run_media_thread` (line 158) gains `opts: MediaThreadOpts` after `default_tap_url`, and
+the Register arm (line 171) constructs with it:
+
+```rust
+                MediaCmd::Register { tap_url, reply } => {
+                    match RtcSession::new_with_config(&opts.media_cfg) {
+```
+
+(the rest of the arm is unchanged). Update the in-file test (line 316):
+
+```rust
+        let thread =
+            MediaThread::spawn(url, MediaThreadOpts::default(), handle).expect("media thread spawn in test");
+```
+
+In `session_map.rs`, replace `spawn_media_thread` (lines 88–95) with the pair:
+
+```rust
+    /// Spawn with defaults — the test-facing convenience; production goes
+    /// through `spawn_media_thread_with` so env config reaches the thread.
+    pub fn spawn_media_thread(
+        self,
+        tokio_handle: tokio::runtime::Handle,
+    ) -> Result<(Self, MediaThread), std::io::Error> {
+        self.spawn_media_thread_with(crate::media_thread::MediaThreadOpts::default(), tokio_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))
+    }
+```
+
+- [ ] **Step 6: config.rs parsers + failing tests first**
+
+Append tests to `config.rs`:
+
+```rust
+    #[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 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)));
+    }
+```
+
+Run: `cargo test -p rutster config::` — expected: compile failure. Then implement:
+
+```rust
+use std::net::IpAddr;
+
+use rutster_media::MediaAddressConfig;
+
+/// 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,
+    advertised_ip: Option,
+    port_range: Option,
+) -> Result {
+    let mut cfg = MediaAddressConfig::default();
+    if let Some(s) = bind_ip {
+        cfg.bind_ip = s.parse::().map_err(|e| format!("RUTSTER_MEDIA_BIND_IP: {e}"))?;
+    }
+    if let Some(s) = advertised_ip {
+        cfg.advertised_ip =
+            Some(s.parse::().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)
+}
+```
+
+Run: `cargo test -p rutster config::` — expected: all pass.
+
+- [ ] **Step 7: Wire main.rs**
+
+In `main.rs`, after the `default_tap_url` block (line 35), build opts and use the `_with`
+spawn:
+
+```rust
+    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 };
+```
+
+and change the spawn call (line 40):
+
+```rust
+    let (app_state, media_thread) = app_state
+        .spawn_media_thread_with(opts, tokio::runtime::Handle::current())
+        .expect("media thread spawn");
+```
+
+- [ ] **Step 8: Re-pin the CI seam gate — SAME COMMIT as the code**
+
+```bash
+cargo fmt   # hash must be of the final formatted content
+git add crates/rutster-media/src/rtc_session.rs
+git hash-object crates/rutster-media/src/rtc_session.rs
+```
+
+Paste the printed hash into `.github/workflows/ci.yml:54` (`EXPECTED_RTC_SESSION=''`)
+and replace the comment block (lines 44–50) with:
+
+```yaml
+      # Seam gate (slice-4 §7 #3; re-pinned slice-5):
+      #
+      # `loop_driver.rs` stays byte-identical to slice-3 — the hot-path
+      # poll loop is untouched. `rtc_session.rs` was re-pinned in slice-5
+      # for MediaAddressConfig (2026-07-04 scalability review, B1:
+      # bind/advertised address split + port range). If a legitimate
+      # change is needed, update the pinned blob hashes below in the same
+      # PR — loud and reviewable by design.
+```
+
+Verify locally exactly as CI will:
+
+```bash
+test "$(git hash-object crates/rutster-media/src/loop_driver.rs)" = "744bf314edf7f4925c8bb3bd0f5176dbc88f8113" && echo LOOP_DRIVER_OK
+```
+
+Expected: `LOOP_DRIVER_OK`.
+
+- [ ] **Step 9: Full check + commit**
+
+Run: `cargo fmt --check && cargo clippy --all -- -D warnings && cargo test --all`
+Expected: clean.
+
+```bash
+git add -A
+git commit -s -m "slice-5: MediaAddressConfig — advertised addr + port range (review B1)
+
+Re-pins the rtc_session.rs seam hash (loud by design). loop_driver.rs
+unchanged. Default config reproduces pre-slice-5 behavior exactly."
+```
+
+---
+
+### Task 3: Admission cap + tick-lag gauge + `MediaCmd::Stats` (review M2)
+
+**Files:**
+- Modify: `crates/rutster/src/media_thread.rs` (cap, gauge, compensated sleep, Stats)
+- Modify: `crates/rutster/src/routes.rs:56-60` (503 mapping)
+- Modify: `crates/rutster/src/config.rs` + `crates/rutster/src/main.rs` (`RUTSTER_MAX_SESSIONS`)
+- Test: `media_thread.rs` inline + `crates/rutster/tests/api_integration.rs`
+
+**Interfaces:**
+- Consumes: `MediaThreadOpts` (Task 2).
+- Produces:
+  - `MediaThreadOpts.max_sessions: usize` (manual `Default` → `64`).
+  - `MediaCmd::Stats { reply: oneshot::Sender }`.
+  - `#[derive(Debug, Clone, serde::Serialize)] pub struct MediaStats { pub sessions: usize, pub max_sessions: usize, pub draining: bool, pub tick_overruns: u64, pub last_tick_micros: u64 }` — `draining` is hardwired `false` until Task 4.
+  - Register error strings `"node full"` (routes contract → 503).
+  - `config::max_sessions(raw: Option) -> Result`.
+
+- [ ] **Step 1: Failing tests (media_thread.rs test module)**
+
+```rust
+    #[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();
+    }
+```
+
+Run: `cargo test -p rutster media_thread` — expected: compile failure (no `max_sessions`,
+no `Stats`).
+
+- [ ] **Step 2: Implement**
+
+`MediaThreadOpts` gains the field (Default → 64, with a comment: "64 is a placeholder
+until the ADR-0010 benchmark produces the real per-node number — the gauge below is that
+benchmark's primary readout"). Add to `MediaCmd`:
+
+```rust
+    /// Cold-path capacity/health snapshot (readyz + the autoscaling
+    /// signal; review M2). Cheap: counters the loop already maintains.
+    Stats { reply: oneshot::Sender },
+```
+
+Add next to `MediaCmd`:
+
+```rust
+/// 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,
+}
+```
+
+In `run_media_thread`, before the loop:
+
+```rust
+    let max_sessions = opts.max_sessions;
+    let mut tick_overruns: u64 = 0;
+    let mut last_tick_micros: u64 = 0;
+```
+
+Register arm gains the admission check as its first line:
+
+```rust
+                MediaCmd::Register { tap_url, reply } => {
+                    // 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) {
+                        // ... existing arm body unchanged ...
+```
+
+Stats arm (in the command drain):
+
+```rust
+                MediaCmd::Stats { reply } => {
+                    let _ = reply.send(MediaStats {
+                        sessions: sessions.len(),
+                        max_sessions,
+                        draining: false, // Task 4 wires the real flag
+                        tick_overruns,
+                        last_tick_micros,
+                    });
+                }
+```
+
+Replace the tail of the loop (`std::thread::sleep(META_TICK);`, line 303) with the
+instrumented, compensated sleep — the tick start is captured **before** the command drain
+(first line inside `loop {`):
+
+```rust
+        let tick_started = Instant::now();
+```
+
+and the tail becomes:
+
+```rust
+        // === 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));
+```
+
+- [ ] **Step 3: Run the new test**
+
+Run: `cargo test -p rutster media_thread`
+Expected: PASS (both media_thread tests).
+
+- [ ] **Step 4: routes 503 mapping + integration test**
+
+In `routes.rs` `create_session` (lines 56–60), replace the error arm:
+
+```rust
+        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()
+        }
+```
+
+Append to `crates/rutster/tests/api_integration.rs` (match the file's existing
+`tower::ServiceExt::oneshot` pattern for building requests):
+
+```rust
+#[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(
+            http::Request::builder()
+                .method("POST")
+                .uri("/v1/sessions")
+                .body(axum::body::Body::empty())
+                .unwrap(),
+        )
+        .await
+        .unwrap();
+    assert_eq!(resp.status(), http::StatusCode::SERVICE_UNAVAILABLE);
+}
+```
+
+Run: `cargo test -p rutster --test api_integration`
+Expected: PASS.
+
+- [ ] **Step 5: `RUTSTER_MAX_SESSIONS` in config.rs + main.rs**
+
+config.rs (test first: `max_sessions(None) == Ok(64)`, `Some("128") == Ok(128)`,
+`Some("zero")` errs mentioning the var name):
+
+```rust
+/// `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) -> Result {
+    match raw {
+        None => Ok(64),
+        Some(s) => s.parse().map_err(|e| format!("RUTSTER_MAX_SESSIONS {s:?}: {e}")),
+    }
+}
+```
+
+main.rs opts construction becomes:
+
+```rust
+    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"),
+    };
+```
+
+- [ ] **Step 6: Full check + commit**
+
+Run: `cargo fmt --check && cargo clippy --all -- -D warnings && cargo test --all`
+
+```bash
+git add -A
+git commit -s -m "slice-5: admission cap, tick-lag gauge, MediaCmd::Stats (review M2)"
+```
+
+---
+
+### Task 4: Drain lifecycle (review M1)
+
+**Files:**
+- Modify: `crates/rutster/src/media_thread.rs` (Drain variant, draining flag)
+- Modify: `crates/rutster/src/main.rs` (two-phase shutdown)
+- Modify: `crates/rutster/src/config.rs` (`RUTSTER_DRAIN_DEADLINE_SECS`)
+- Test: `media_thread.rs` inline
+
+**Interfaces:**
+- Consumes: Task 3's `MediaStats`/Register-reject shape.
+- Produces:
+  - `MediaCmd::Drain { reply: oneshot::Sender<()> }` — flips draining on; reply fires when the session map empties (immediately if already empty).
+  - Register rejection string `"draining"` (routes 503 contract — mapping already landed in Task 3).
+  - `MediaStats.draining` reports the real flag.
+  - `config::drain_deadline(raw: Option) -> Result` — default `0s` (dev loop keeps today's instant shutdown; operators opt into real drain).
+
+- [ ] **Step 1: Failing test**
+
+```rust
+    #[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();
+    }
+```
+
+Run: `cargo test -p rutster media_thread` — expected: compile failure (no `Drain`).
+
+- [ ] **Step 2: Implement**
+
+`MediaCmd` gains:
+
+```rust
+    /// 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<()> },
+```
+
+`run_media_thread` state, next to the counters:
+
+```rust
+    let mut draining = false;
+    let mut drain_done: Option> = None;
+```
+
+Register arm — the draining check goes FIRST (before the capacity check):
+
+```rust
+                    if draining {
+                        let _ = reply.send(Err("draining: not accepting new sessions".into()));
+                        continue;
+                    }
+```
+
+Drain arm (in the command drain):
+
+```rust
+                MediaCmd::Drain { reply } => {
+                    draining = true;
+                    if sessions.is_empty() {
+                        let _ = reply.send(());
+                    } else {
+                        drain_done = Some(reply);
+                    }
+                }
+```
+
+Stats arm: `draining: false` → `draining,`. After the eviction sweep
+(`for id in closed_ids { ... }`, line ~297), append:
+
+```rust
+        // 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: Run test**
+
+Run: `cargo test -p rutster media_thread` — expected: PASS.
+
+- [ ] **Step 4: config parser + two-phase shutdown in main.rs**
+
+config.rs (test first: `None → 0s`, `Some("300") → 300s`, garbage errs with var name):
+
+```rust
+/// `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) -> Result {
+    match raw {
+        None => Ok(std::time::Duration::ZERO),
+        Some(s) => s
+            .parse::()
+            .map(std::time::Duration::from_secs)
+            .map_err(|e| format!("RUTSTER_DRAIN_DEADLINE_SECS {s:?}: {e}")),
+    }
+}
+```
+
+main.rs — restructure the serve/shutdown tail (lines 44–52). The key ordering change:
+**drain runs while axum still serves** (in-flight calls still need DELETE/offer during
+bleed-out; the LB pulls the node via readyz, Task 5), and only then does HTTP quiesce:
+
+```rust
+    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();
+            if drain_cmd_tx
+                .send(rutster::media_thread::MediaCmd::Drain { reply })
+                .await
+                .is_ok()
+            {
+                match tokio::time::timeout(drain_deadline, drained).await {
+                    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(async {
+            let _ = http_stop_rx.await;
+        })
+        .await
+        .unwrap();
+
+    media_thread.shutdown();
+```
+
+(The `shutdown_signal` doc comment at main.rs:55–59 — delete the sentence "No in-flight
+call preservation story in the dev loop." and describe the two-phase flow instead.)
+
+- [ ] **Step 5: Full check + commit**
+
+Run: `cargo fmt --check && cargo clippy --all -- -D warnings && cargo test --all`
+
+```bash
+git add -A
+git commit -s -m "slice-5: drain lifecycle — SIGTERM bleeds out calls (review M1)"
+```
+
+---
+
+### Task 5: `GET /healthz` + `GET /readyz`
+
+> **Deviation note (post-review):** the snippets below contain two bugs found and fixed
+> in commit 5ce18bf during execution — the readyz timeout must wrap the SEND as well as
+> the reply await, and the integration test must shut its `MediaThread` down. Copy from
+> the landed code, not these snippets.
+
+**Files:**
+- Modify: `crates/rutster/src/routes.rs` (two routes + handlers)
+- Test: `routes.rs` inline (default-AppState zombie case) + `crates/rutster/tests/api_integration.rs` (live case)
+
+**Interfaces:**
+- Consumes: `MediaCmd::Stats`, `MediaStats` (Tasks 3–4).
+- Produces:
+  - `GET /healthz` → 200 `"ok"` always (liveness: process + HTTP stack up).
+  - `GET /readyz` → 200 + `MediaStats` JSON iff the media thread answers Stats within 250 ms AND `!draining` AND `sessions < max_sessions`; else 503 (body: the JSON if available, `"media thread unresponsive"` if not). Readiness = "can this node take a NEW call."
+
+- [ ] **Step 1: Failing tests**
+
+`routes.rs` test module (the zombie case — this is the *point* of readyz: a dead media
+thread must flip readiness off while the HTTP listener still answers):
+
+```rust
+    #[tokio::test]
+    async fn readyz_503_when_media_thread_gone_but_healthz_200() {
+        use tower::ServiceExt;
+        // 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);
+    }
+```
+
+`api_integration.rs` (live case):
+
+```rust
+#[tokio::test]
+async fn readyz_200_with_stats_json_when_thread_alive() {
+    let state = rutster::session_map::AppState::default();
+    let (state, _thread) = state
+        .spawn_media_thread(tokio::runtime::Handle::current())
+        .expect("spawn");
+    let app = rutster::routes::router(state);
+    let resp = app
+        .oneshot(
+            http::Request::builder()
+                .uri("/readyz")
+                .body(axum::body::Body::empty())
+                .unwrap(),
+        )
+        .await
+        .unwrap();
+    assert_eq!(resp.status(), http::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);
+}
+```
+
+Run: `cargo test -p rutster` — expected: FAIL (404 on the routes).
+
+- [ ] **Step 2: Implement handlers + routes**
+
+routes.rs:
+
+```rust
+/// 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) -> Response {
+    let (reply, rx) = tokio::sync::oneshot::channel();
+    if state
+        .cmd_tx
+        .send(crate::media_thread::MediaCmd::Stats { reply })
+        .await
+        .is_err()
+    {
+        return (StatusCode::SERVICE_UNAVAILABLE, "media thread unresponsive").into_response();
+    }
+    let stats = match tokio::time::timeout(std::time::Duration::from_millis(250), rx).await {
+        Ok(Ok(s)) => s,
+        _ => {
+            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()
+}
+```
+
+Router additions (in `router()`, before `.with_state`):
+
+```rust
+        .route("/healthz", get(healthz))
+        .route("/readyz", get(readyz))
+```
+
+- [ ] **Step 3: Run tests**
+
+Run: `cargo test -p rutster && cargo test -p rutster --test api_integration`
+Expected: PASS.
+
+- [ ] **Step 4: Commit**
+
+```bash
+git add -A
+git commit -s -m "slice-5: /healthz + /readyz — readiness reads MediaCmd::Stats"
+```
+
+---
+
+### Task 6: Non-blocking tap teardown (review M7)
+
+**Files:**
+- Modify: `crates/rutster/src/tap_engine.rs` (new `spawn_tap_teardown` free fn + test)
+- Modify: `crates/rutster/src/media_thread.rs:201-223` (Delete arm calls it)
+
+**Interfaces:**
+- Consumes: `TapConn` (existing).
+- Produces: `pub fn spawn_tap_teardown(tokio_handle: &tokio::runtime::Handle, id: ChannelId, conn: TapConn)` — returns immediately; the 750 ms bounded close_tx→join→abort sequence runs on a tokio task.
+
+- [ ] **Step 1: Failing test (tap_engine.rs test module)**
+
+```rust
+    #[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)");
+    }
+```
+
+Run: `cargo test -p rutster tap_engine` — expected: compile failure (`spawn_tap_teardown`
+undefined).
+
+- [ ] **Step 2: Implement in tap_engine.rs**
+
+```rust
+/// 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)");
+            }
+        }
+    });
+}
+```
+
+Note: `close_tx` is consumed by `send`, but `conn` is only partially moved inside the
+async block — Rust allows the field-by-field moves because the closure owns `conn`. If
+the borrow checker objects on the `&mut conn.join` + later `conn.join.abort()` pair,
+destructure instead: `let TapConn { close_tx, mut join, .. } = conn;`.
+
+- [ ] **Step 3: Rewire the Delete arm (media_thread.rs:201-223)**
+
+```rust
+                MediaCmd::Delete { id, reply } => {
+                    if let Some(mut s) = sessions.remove(&id) {
+                        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;
+                    }
+                    let _ = reply.send(());
+                }
+```
+
+- [ ] **Step 4: Run the full suite (the existing tap/barge/realtime integration tests are the regression net for live-engine teardown)**
+
+Run: `cargo test --all`
+Expected: PASS — `tap_integration`, `barge_in_integration`, `realtime_integration` all green.
+
+- [ ] **Step 5: Commit**
+
+```bash
+git add -A
+git commit -s -m "slice-5: Delete teardown off the tick loop (review M7)"
+```
+
+---
+
+### Task 7: `EventSink` + wall-clock `started_at` (review M3)
+
+**Files:**
+- Modify: `crates/rutster-call-model/src/lib.rs` (Channel gains `started_at: SystemTime`)
+- Create: `crates/rutster/src/event_sink.rs`
+- Modify: `crates/rutster/src/lib.rs` (`pub mod event_sink;`)
+- Modify: `crates/rutster/src/media_thread.rs` (opts gains sink; emissions at Register/Connected/Delete/evict/Shutdown)
+- Test: call-model inline; `media_thread.rs` inline with a channel-backed TestSink
+
+**Interfaces:**
+- Consumes: `MetricsSnapshot` (rutster-tap), `ChannelId`, `MediaThreadOpts`.
+- Produces:
+  - `Channel.started_at: std::time::SystemTime` (set in `new_inbound`; `created_at: Instant` STAYS — monotonic for idle math, wall-clock for CDR; the doc comments explain the pairing).
+  - `pub trait EventSink: Send + Sync { fn emit(&self, event: CallEvent); }` — **contract:** `emit` is called from the media `std::thread` and MUST NOT block or do I/O inline (a Valkey impl buffers via channel to a tokio task; that impl lands with ADR-0005 wiring, later rung).
+  - `#[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_metrics: Option } }`.
+  - `#[derive(Debug, Clone, Copy, PartialEq, Eq)] pub enum EndReason { Deleted, Closed, Shutdown }`.
+  - `pub struct TracingEventSink;` — emits one structured `tracing::info!(target: "rutster::events", …)` per event (the log-backed placeholder for the bus).
+  - `MediaThreadOpts.sink: std::sync::Arc` (Default → `Arc::new(TracingEventSink)`).
+
+- [ ] **Step 1: call-model — failing test then field**
+
+Test (call-model tests module):
+
+```rust
+    #[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);
+    }
+```
+
+Implement: `Channel` gains
+
+```rust
+    /// 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,
+```
+
+and `new_inbound()` sets `started_at: std::time::SystemTime::now(),`.
+
+Run: `cargo test -p rutster-call-model` — expected: PASS.
+
+- [ ] **Step 2: event_sink.rs (types + TracingEventSink)**
+
+```rust
+//! # 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,
+    },
+}
+
+/// 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;
+```
+
+Add `pub mod event_sink;` to `crates/rutster/src/lib.rs`.
+(`MetricsSnapshot` must be re-exported from `rutster_tap` — it lives in `metrics.rs`;
+check `crates/rutster-tap/src/lib.rs` re-exports it, add `pub use metrics::MetricsSnapshot;`
+if absent.)
+
+- [ ] **Step 3: Failing media_thread test with a TestSink**
+
+```rust
+    #[test]
+    fn lifecycle_events_emit_registered_then_ended_deleted() {
+        use crate::event_sink::{CallEvent, EndReason, EventSink};
+        struct TestSink(std::sync::Mutex>);
+        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
+        )));
+    }
+```
+
+Run: `cargo test -p rutster media_thread` — expected: compile failure (no `sink` field).
+
+- [ ] **Step 4: Implement emissions**
+
+`MediaThreadOpts` gains `pub sink: crate::event_sink::SharedEventSink,`; `Default` sets
+`sink: std::sync::Arc::new(crate::event_sink::TracingEventSink),`. In `run_media_thread`
+let-bind `let sink = opts.sink.clone();` before the loop. Emissions:
+
+- Register arm, after `sessions.insert(...)` / before `reply`:
+
+```rust
+                        sink.emit(crate::event_sink::CallEvent::SessionRegistered {
+                            id,
+                            at: started_at,
+                        });
+```
+
+(read `let started_at = session.rtc.channel.started_at;` — take it before the
+`sessions.insert` moves the session, i.e. right after `let id = session.channel_id();`).
+
+- Connected transition — after the existing `info!(channel_id = %id, "tap engine + reflex
+  + local VAD wired on Connected");` and **before the `continue;`** that ends the block:
+
+```rust
+                    sink.emit(crate::event_sink::CallEvent::SessionConnected {
+                        id: *id,
+                        at: std::time::SystemTime::now(),
+                    });
+```
+
+- Delete arm — full rewritten arm (ORDER MATTERS: the metrics snapshot must be read
+  **before** `s.tap_conn.take()` empties the field, or `tap_metrics` is silently `None`):
+
+```rust
+                MediaCmd::Delete { id, reply } => {
+                    if let Some(mut s) = sessions.remove(&id) {
+                        // 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() {
+                            crate::tap_engine::spawn_tap_teardown(&tokio_handle, id, conn);
+                        }
+                        s.rtc.channel.tap = None;
+                        s.rtc.channel.state = rutster_call_model::ChannelState::Closed;
+                    }
+                    let _ = reply.send(());
+                }
+```
+
+- Eviction sweep — change the loop to fetch state before removal:
+
+```rust
+        for id in closed_ids {
+            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");
+        }
+```
+
+- Shutdown arm, before `sessions.clear()`:
+
+```rust
+                    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()),
+                        });
+                    }
+```
+
+main.rs opts construction gains nothing (Default supplies TracingEventSink) — add a
+comment noting the seam: `// sink: TracingEventSink via Default — ADR-0005 Valkey impl replaces it at the same seam.`
+
+- [ ] **Step 5: Run + commit**
+
+Run: `cargo fmt --check && cargo clippy --all -- -D warnings && cargo test --all`
+Expected: PASS.
+
+```bash
+git add -A
+git commit -s -m "slice-5: EventSink seam + wall-clock started_at (review M3)
+
+Lifecycle events now flow through a trait the ADR-0005 Valkey publisher
+can implement; Channel carries SystemTime so a CDR can exist."
+```
+
+---
+
+### Task 8: ADR-0009 amendment + ADR-0005 note + ARCHITECTURE.md placement line
+
+**Files:**
+- Modify: `docs/adr/0009-spend-gate-honest-rescope.md` (append amendment before References)
+- Modify: `docs/adr/0005-event-bus.md` (one-line note under Constraints)
+- Modify: `docs/ARCHITECTURE.md:34-36` (horizontal list)
+
+- [ ] **Step 1: Append to ADR-0009** (between "Consequences" and "References"):
+
+```markdown
+## Amendment 2026-07-04 — enforcement locality ≠ accounting locality
+
+- **Status:** Proposed (pending maintainer ratification)
+- **Origin:** [2026-07-04 scalability & infra-fit review](../reviews/2026-07-04-scalability-infra-review.md),
+  finding M5.
+
+Guarantee 2 above prescribes where the **check** sits — in-process with the tap and the
+provider call-control client. It is silent on where the **accounting state** lives, and at
+N>1 core instances that silence becomes a correctness trap: per-instance counters make
+every spend/pacing cap silently N× the fleet intent, and toll-fraud thresholds never trip
+because attempts spread across instances.
+
+Clarification, binding on the step-6 implementation:
+
+1. **Enforcement is in-process** (unchanged — constitutive; guarantee 2 stands).
+2. **Accounting is shared.** The gate is built against a ledger trait with atomic
+   check-and-reserve semantics from day one: an in-memory implementation for single-node,
+   a Valkey-backed one ([ADR-0005](0005-event-bus.md)) for fleets. The check path never
+   assumes counter locality.
+3. **ADR-0005 constraint 2 is not a counter-argument.** "The bus is not the source of
+   truth for billing- or call-loss-critical state" governs the *durable CDR*. Live
+   enforcement counters (spend/pacing/rate state) are ephemeral control state — exactly
+   what Valkey KV is for. Losing them on a Valkey restart degrades fail-safe (re-count
+   from zero, provider-side caps as the backstop per this ADR's deployment guidance) —
+   not to billing corruption.
+```
+
+- [ ] **Step 2: ADR-0005 cross-ref** — append to the "Constraints" list:
+
+```markdown
+3. **Constraint 2 governs the durable CDR, not enforcement counters.** The spend gate's
+   live accounting state (spend/pacing/rate counters) is ephemeral control state and DOES
+   belong in Valkey KV at N>1 — see the
+   [ADR-0009 amendment 2026-07-04](0009-spend-gate-honest-rescope.md#amendment-2026-07-04--enforcement-locality--accounting-locality).
+```
+
+- [ ] **Step 3: ARCHITECTURE.md** — in the "Horizontal platform" sentence (lines 34–36), change
+
+```markdown
+inventory, billing rollup, analytics, multi-region orchestration, the management API, and the agent
+brain itself.
+```
+
+to
+
+```markdown
+inventory, billing rollup, analytics, multi-region orchestration, call placement (the
+session→node directory — which node owns which live call, the routing layer above N cores),
+the management API, and the agent brain itself.
+```
+
+- [ ] **Step 4: Verify links render + commit**
+
+Run: `ls docs/adr/ && grep -c "Amendment 2026-07-04" docs/adr/0009-spend-gate-honest-rescope.md`
+Expected: `1`.
+
+```bash
+git add docs/adr/0009-spend-gate-honest-rescope.md docs/adr/0005-event-bus.md docs/ARCHITECTURE.md
+git commit -s -m "docs(adr): ADR-0009 amendment — shared spend accounting (review M5)"
+```
+
+---
+
+### Task 9: Tap protocol v2 reservations (doc-only; review M4)
+
+**Files:**
+- Modify: `crates/rutster-tap/src/protocol.rs` (module doc `//!` section only — NO code)
+
+- [ ] **Step 1: Append to the module doc** (after the existing wire-contract prose):
+
+```rust
+//! ## 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.
+```
+
+- [ ] **Step 2: Verify doc renders + commit**
+
+Run: `cargo doc --no-deps -p rutster-tap 2>&1 | tail -2 && cargo test -p rutster-tap`
+Expected: doc builds clean; tests untouched and green.
+
+```bash
+git add crates/rutster-tap/src/protocol.rs
+git commit -s -m "docs(tap): reserve v2 resume/terminal-bye semantics (review M4)"
+```
+
+---
+
+## Done criteria (slice level)
+
+1. `cargo fmt --check && cargo clippy --all -- -D warnings && cargo test --all && cargo deny check` — green on stable + 1.85.
+2. Seam gate: `loop_driver.rs` hash unchanged; `rtc_session.rs` re-pinned once (Task 2).
+3. With **no env vars set**, the dev loop behaves exactly as before slice-5 (loopback media, port 8080, instant Ctrl-C shutdown) — the seams cost nothing until configured.
+4. With `RUTSTER_MEDIA_ADVERTISED_IP` + `RUTSTER_MEDIA_PORT_RANGE` + `RUTSTER_HTTP_BIND` + `RUTSTER_MAX_SESSIONS` + `RUTSTER_DRAIN_DEADLINE_SECS` set, the binary is deployable behind NAT with an LB reading `/readyz` — the review's B1/M1/M2 exit criteria.
+5. Every call leaves a `rutster::events` record with wall-clock start/end + reason + tap counters (M3's seam, log-backed).
+6. PR via `tea`, squash-merge, DCO-signed commits throughout.
+```