rutster/docs/superpowers/plans/2026-07-04-slice-5-scalability-seams.md

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 — 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<String>/&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<String>) -> Result<std::net::SocketAddr, String> (later tasks add more parsers to this module).

• Step 1: Write the failing tests

Create crates/rutster/src/config.rs:

//! # config — pure env-parsing helpers (slice-5)
//!
//! Every knob is a pure function over `Option<String>` / `&str` so tests
//! never mutate process env (env mutation in tests races across the
//! parallel test harness — the same reason `api_key.rs` needs its
//! ENV_MUTEX). `main.rs` is the only caller that touches `std::env`.
//!
//! Closes 2026-07-04 scalability review "HTTP bind hardcoded" (minor).

use std::net::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<String>) -> Result<SocketAddr, String> {
    match raw {
        None => Ok("0.0.0.0:8080".parse().expect("static default parses")),
        Some(s) => s
            .parse()
            .map_err(|e| format!("RUTSTER_HTTP_BIND {s:?} is not a socket address: {e}")),
    }
}

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

    #[test]
    fn http_bind_defaults_when_unset() {
        assert_eq!(
            http_bind(None).unwrap(),
            "0.0.0.0:8080".parse::<SocketAddr>().unwrap()
        );
    }

    #[test]
    fn http_bind_parses_override() {
        assert_eq!(
            http_bind(Some("127.0.0.1:9090".into())).unwrap(),
            "127.0.0.1:9090".parse::<SocketAddr>().unwrap()
        );
    }

    #[test]
    fn http_bind_rejects_garbage_with_var_name_in_error() {
        let err = http_bind(Some("not-an-addr".into())).unwrap_err();
        assert!(err.contains("RUTSTER_HTTP_BIND"));
    }
}

• 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:

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):

    let addr: SocketAddr = "0.0.0.0:8080".parse().expect("valid addr");

with:

    // 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:

| 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.

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<IpAddr>, 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<Self, RtcSessionError>; 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<dyn EventSink> 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<String> ×3) -> Result<MediaAddressConfig, String>.
  • 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)

    #[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):

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<IpAddr>,
    pub port_range: Option<(u16, u16)>,
}

impl Default for MediaAddressConfig {
    /// Loopback + ephemeral — byte-for-byte the pre-slice-5 behavior, so
    /// every existing test and the dev loop run unchanged.
    fn default() -> Self {
        Self {
            bind_ip: IpAddr::from([127, 0, 0, 1]),
            advertised_ip: None,
            port_range: None,
        }
    }
}

/// Bind the first free UDP port in `[lo, hi]` (inclusive).
///
/// Sequential scan — O(range) worst case, cold path only (session
/// construction). The OS is the allocator: a failed bind means "taken,"
/// so no shared allocator state is needed across sessions or threads.
fn bind_in_range(ip: IpAddr, lo: u16, hi: u16) -> std::io::Result<UdpSocket> {
    let mut last_err = None;
    for port in lo..=hi {
        match UdpSocket::bind(SocketAddr::new(ip, port)) {
            Ok(s) => return Ok(s),
            Err(e) => last_err = Some(e),
        }
    }
    Err(last_err.unwrap_or_else(|| {
        std::io::Error::new(std::io::ErrorKind::AddrInUse, "empty port range")
    }))
}

Change the struct: add a field after local_addr (line 88):

    /// 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):

    pub fn new() -> Result<Self, RtcSessionError> {
        Self::new_with_config(&MediaAddressConfig::default())
    }

    /// Construct with explicit addressing (slice-5, review B1). The
    /// binary threads `MediaThreadOpts.media_cfg` here; `new()` keeps the
    /// loopback default for tests and the dev loop.
    pub fn new_with_config(cfg: &MediaAddressConfig) -> Result<Self, RtcSessionError> {
        // Fail-fast: a wildcard bind with nothing to advertise would die
        // later inside accept_offer (str0m rejects 0.0.0.0 candidates).
        // Cold path → a real error, not an expect().
        if cfg.bind_ip.is_unspecified() && cfg.advertised_ip.is_none() {
            return Err(RtcSessionError::Socket(std::io::Error::new(
                std::io::ErrorKind::InvalidInput,
                "RUTSTER_MEDIA_BIND_IP is unspecified (0.0.0.0/::) — set RUTSTER_MEDIA_ADVERTISED_IP",
            )));
        }
        let socket = match cfg.port_range {
            Some((lo, hi)) => bind_in_range(cfg.bind_ip, lo, hi)?,
            None => UdpSocket::bind(SocketAddr::new(cfg.bind_ip, 0))?,
        };
        socket.set_nonblocking(true)?;
        let local_addr = socket.local_addr()?;
        let advertised_addr =
            SocketAddr::new(cfg.advertised_ip.unwrap_or(cfg.bind_ip), local_addr.port());

        let rtc = Rtc::new(Instant::now());

        Ok(Self {
            channel: Channel::new_inbound(),
            rtc,
            decoder: OpusDecoder::new()?,
            encoder: OpusEncoder::new()?,
            pipe: Box::new(EchoAudioPipe::new()),
            socket,
            local_addr,
            advertised_addr,
            audio_mid: None,
            next_timeout: None,
            last_rx: Instant::now(),
            last_outbound_at: Instant::now(),
            next_media_time: str0m::media::MediaTime::ZERO,
        })
    }

In accept_offer (line 236) change the candidate to the advertised address:

        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:

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):

/// 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<dyn EventSink>`). 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:

    pub fn spawn(
        default_tap_url: url::Url,
        opts: MediaThreadOpts,
        tokio_handle: tokio::runtime::Handle,
    ) -> Result<Self, std::io::Error> {
        let (cmd_tx, cmd_rx) = mpsc::channel(CMD_CHANNEL_CAPACITY);
        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:

                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):

        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:

    /// 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:

    #[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:

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<String>,
    advertised_ip: Option<String>,
    port_range: Option<String>,
) -> Result<MediaAddressConfig, String> {
    let mut cfg = MediaAddressConfig::default();
    if let Some(s) = bind_ip {
        cfg.bind_ip = s.parse::<IpAddr>().map_err(|e| format!("RUTSTER_MEDIA_BIND_IP: {e}"))?;
    }
    if let Some(s) = advertised_ip {
        cfg.advertised_ip =
            Some(s.parse::<IpAddr>().map_err(|e| format!("RUTSTER_MEDIA_ADVERTISED_IP: {e}"))?);
    }
    if let Some(s) = port_range {
        cfg.port_range = Some(parse_port_range(&s)?);
    }
    Ok(cfg)
}

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:

    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):

    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

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='<new>') and replace the comment block (lines 44–50) with:

      # 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:

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.

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<MediaStats> }.
  • #[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<String>) -> Result<usize, String>.

• Step 1: Failing tests (media_thread.rs test module)

    #[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:

    /// Cold-path capacity/health snapshot (readyz + the autoscaling
    /// signal; review M2). Cheap: counters the loop already maintains.
    Stats { reply: oneshot::Sender<MediaStats> },

Add next to MediaCmd:

/// 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:

    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:

                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):

                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 {):

        let tick_started = Instant::now();

and the tail becomes:

        // === 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:

        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):

#[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):

/// `RUTSTER_MAX_SESSIONS` — admission cap. Default 64 (placeholder until
/// the ADR-0010 benchmark measures the real per-node ceiling).
pub fn max_sessions(raw: Option<String>) -> Result<usize, String> {
    match raw {
        None => Ok(64),
        Some(s) => s.parse().map_err(|e| format!("RUTSTER_MAX_SESSIONS {s:?}: {e}")),
    }
}

main.rs opts construction becomes:

    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

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<String>) -> Result<std::time::Duration, String> — default 0s (dev loop keeps today's instant shutdown; operators opt into real drain).

• Step 1: Failing test

    #[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:

    /// 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:

    let mut draining = false;
    let mut drain_done: Option<oneshot::Sender<()>> = None;

Register arm — the draining check goes FIRST (before the capacity check):

                    if draining {
                        let _ = reply.send(Err("draining: not accepting new sessions".into()));
                        continue;
                    }

Drain arm (in the command drain):

                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:

        // 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):

/// `RUTSTER_DRAIN_DEADLINE_SECS` — how long shutdown waits for in-flight
/// calls before the hard stop. Default 0 = today's instant shutdown (dev
/// loop); production sets minutes. Calls are non-migratable by design, so
/// drain-then-terminate is the ONLY graceful scale-in shape.
pub fn drain_deadline(raw: Option<String>) -> Result<std::time::Duration, String> {
    match raw {
        None => Ok(std::time::Duration::ZERO),
        Some(s) => s
            .parse::<u64>()
            .map(std::time::Duration::from_secs)
            .map_err(|e| format!("RUTSTER_DRAIN_DEADLINE_SECS {s:?}: {e}")),
    }
}

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:

    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

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):

    #[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):

#[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:

/// GET /healthz — liveness only: the process and HTTP stack are up.
/// Deliberately does NOT consult the media thread — liveness and
/// readiness are different probes with different restart semantics.
pub async fn healthz() -> Response {
    (StatusCode::OK, "ok").into_response()
}

/// GET /readyz — "can this node accept a NEW call right now?"
/// LB target membership + autoscaler both read this. 503 when draining,
/// at capacity, or when the media thread doesn't answer Stats in 250ms
/// (a wedged/dead media thread previously left GET / answering 200 —
/// the zombie-node failure from the 2026-07-04 review).
pub async fn readyz(State(state): State<AppState>) -> Response {
    let (reply, rx) = tokio::sync::oneshot::channel();
    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):

        .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

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)

    #[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

/// 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)

                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

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<MetricsSnapshot> } }.
  • #[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<dyn EventSink> (Default → Arc::new(TracingEventSink)).

• Step 1: call-model — failing test then field

Test (call-model tests module):

    #[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

    /// 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)

//! # event_sink — durable-lifecycle seam (slice-5, review M3)
//!
//! ADR-0005 promises call lifecycle → Valkey streams → durable CDR
//! pipeline. Until that lands, lifecycle events die with the process —
//! an OOM-killed node erases all evidence of its calls. This module is
//! the SEAM: the media thread emits `CallEvent`s through `EventSink`;
//! today's impl logs, the ADR-0005 impl will publish. The emission
//! points and the event shape are the part that calcifies — the backend
//! is plumbing.

use std::sync::Arc;
use std::time::SystemTime;

use rutster_call_model::ChannelId;
use rutster_tap::MetricsSnapshot;

/// CDR-shaped lifecycle events. `SessionEnded` carries both timestamps so
/// a consumer computes duration without correlating two events (a node
/// can die between them — the whole point of emitting durably).
#[derive(Debug, Clone)]
pub enum CallEvent {
    SessionRegistered { id: ChannelId, at: SystemTime },
    SessionConnected { id: ChannelId, at: SystemTime },
    SessionEnded {
        id: ChannelId,
        started_at: SystemTime,
        ended_at: SystemTime,
        reason: EndReason,
        /// Tap counters at teardown — fulfills the tap_engine.rs promise
        /// ("read a MetricsSnapshot for log/CDR emission on teardown").
        /// Snapshot is taken just before the engine task is torn down, so
        /// counts may trail the final frame by a tick — fine for a CDR.
        tap_metrics: Option<MetricsSnapshot>,
    },
}

/// Why the session ended — the CDR disposition embryo.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum EndReason {
    /// DELETE /v1/sessions/:id (API-driven hangup).
    Deleted,
    /// The session closed itself (peer close / idle timeout).
    Closed,
    /// Node shutdown dropped it (the mass-hangup path drain exists to avoid).
    Shutdown,
}

/// The seam. CONTRACT: `emit` is called from the media std::thread —
/// implementations MUST NOT block or perform I/O inline. Buffer to your
/// own task (the future Valkey impl does channel → tokio publisher).
pub trait EventSink: Send + Sync {
    fn emit(&self, event: CallEvent);
}

/// Log-backed sink: structured tracing on target "rutster::events". Not
/// durable — the placeholder that makes the emission points real today.
pub struct TracingEventSink;

impl EventSink for TracingEventSink {
    fn emit(&self, event: CallEvent) {
        tracing::info!(target: "rutster::events", event = ?event, "call event");
    }
}

/// Convenience alias for the opts field / spawn signature.
pub type SharedEventSink = Arc<dyn EventSink>;

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

    #[test]
    fn lifecycle_events_emit_registered_then_ended_deleted() {
        use crate::event_sink::{CallEvent, EndReason, EventSink};
        struct TestSink(std::sync::Mutex<Vec<CallEvent>>);
        impl EventSink for TestSink {
            fn emit(&self, event: CallEvent) {
                self.0.lock().unwrap().push(event);
            }
        }
        let sink = std::sync::Arc::new(TestSink(std::sync::Mutex::new(Vec::new())));

        let rt = tokio::runtime::Runtime::new().unwrap();
        let handle = rt.handle().clone();
        let url = url::Url::parse("ws://127.0.0.1:8081/echo").unwrap();
        let opts = MediaThreadOpts {
            sink: sink.clone(),
            ..Default::default()
        };
        let thread = MediaThread::spawn(url, opts, handle).expect("spawn");

        let (reply, rx) = oneshot::channel();
        thread
            .cmd_tx()
            .blocking_send(MediaCmd::Register {
                tap_url: url::Url::parse("ws://127.0.0.1:1/echo").unwrap(),
                reply,
            })
            .unwrap();
        let id = rx.blocking_recv().unwrap().unwrap();
        let (reply, rx) = oneshot::channel();
        thread
            .cmd_tx()
            .blocking_send(MediaCmd::Delete { id, reply })
            .unwrap();
        rx.blocking_recv().unwrap();
        thread.shutdown();

        let events = sink.0.lock().unwrap();
        assert!(
            matches!(&events[0], CallEvent::SessionRegistered { id: eid, .. } if *eid == id)
        );
        assert!(events.iter().any(|e| matches!(
            e,
            CallEvent::SessionEnded { id: eid, reason: EndReason::Deleted, .. } if *eid == id
        )));
    }

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:

                        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 thecontinue;`** that ends the block:

                    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):

                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:

        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():

                    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.

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"):

## 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:

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

inventory, billing rollup, analytics, multi-region orchestration, the management API, and the agent
brain itself.

to

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.

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):

//! ## 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.

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.