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](../../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<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`:

```rust
//! # 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`:

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

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

```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, 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:

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

```rust
    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:

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

```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='<new>'`)
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<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)**

```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<MediaStats> },
```

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

```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<String>) -> Result<std::time::Duration, String>` — 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<oneshot::Sender<()>> = 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<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:

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

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

```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<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**

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

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