From 3c5cc7501561ed236a7347028a1bb6905c95a7fd Mon Sep 17 00:00:00 2001 From: adlee-was-taken Date: Sun, 28 Jun 2026 11:19:41 -0400 Subject: [PATCH] plan(slice-1): adversarial review patches MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Verified str0m 0.21 + opus 0.3.1 API surfaces via docs.rs subagents; patched the plan with the real signatures and several design-level fixes. API-surface patches (str0m 0.21, verified): - Global Constraints: full verified API surface documented inline (Rtc::new takes Instant; SdpOffer::from_sdp_string is the entry point, NOT from_str_unchecked; add_local_candidate returns Option<&Candidate>; Writer::write takes rtp_time not media_time; MediaTime has no add(Duration) -> use mt + MediaTime::from(d); payload_params returns impl Iterator; MediaData.data is Arc<[u8]>). - Task 4 accept_offer: rewritten with from_sdp_string + correct error mapping; RtcSessionError::SdpOffer changed to String (collapses parse + accept failures uniformly). - Task 4 loop_driver: MediaTime::add -> + MediaTime::from(Duration); media.data deref coercion documented. Design-level patches: - Task 4: RtcSession::new_for_test -> pub fn new() (single idiomatic constructor, no test/prod split). - Task 4: added accept_offer_transitions_channel_to_connecting test (the transition was claimed but untested in the original plan). - Task 5: dropped reqwest from workspace deps (unused — integration test uses tower::ServiceExt::oneshot); added tower as workspace dep + binary crate dev-dependency. - Task 5: removed duplicate DELETE /v1/sessions route (was incorrectly chained via .delete() on the collection route AND on /v1/sessions/:id — only the latter is correct). - Task 5: clarified pub mod requirement (must be pub because the integration test references modules via absolute paths). - Global Constraints: added task/PR strategy (one commit per task, merged in numeric order, granular history is load-bearing for the learning-codebase goal). Self-review section updated to reflect the patches. Plan is now implementation-ready. --- .../2026-06-28-slice-1-webrtc-loopback.md | 2646 +++++++++++++++++ 1 file changed, 2646 insertions(+) create mode 100644 docs/superpowers/plans/2026-06-28-slice-1-webrtc-loopback.md diff --git a/docs/superpowers/plans/2026-06-28-slice-1-webrtc-loopback.md b/docs/superpowers/plans/2026-06-28-slice-1-webrtc-loopback.md new file mode 100644 index 0000000..91f14b2 --- /dev/null +++ b/docs/superpowers/plans/2026-06-28-slice-1-webrtc-loopback.md @@ -0,0 +1,2646 @@ +# Slice 1 — WebRTC Media Loopback 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:** Stand up the Rutster Rust workspace and implement spearhead step 1 — a browser talks WebRTC to the core; the core terminates DTLS-SRTP, decodes Opus → canonical 16-bit PCM @ 24 kHz mono, echoes the PCM back to the browser. The user speaks and hears themselves back with no perceptible delay. + +**Architecture:** Fused per-call vertical (ADR-0002) with a Cargo workspace of one binary + five library crates shaped to ADR-0002's fused vertical (`rutster`, `rutster-media`, `rutster-call-model`, `rutster-signaling-sip`, `rutster-tap`, `rutster-spend`). Media is driven by `str0m` (sans-IO WebRTC) + `opus` crate (libopus FFI) on tokio polls (an explicit, scoped deviation from ARCHITECTURE.md's "dedicated thread" mandate — step 4 replaces it). `RtcSession` owns a `str0m::Rtc` + an Opus encoder/decoder pair + an `EchoAudioPipe`. The PCM-tap seam is exposed as `AudioSource` / `AudioSink` traits in `rutster-media`. + +**Tech Stack:** Rust stable (pinned via `rust-toolchain.toml`), `str0m 0.21` (sans-IO WebRTC, Frame API), `opus 0.3.1` (libopus FFI), `axum 0.7`, `tokio 1`, `dashmap 6`, `uuid 1`, `thiserror 1`, `tracing 0.1`, `serde 1`, `tower 0.5` (integration test only — `ServiceExt::oneshot` on the axum Router), `cargo-deny`. + +--- + +## Global Constraints + +Binding values for every task — copy verbatim where used. + +- **License:** every crate manifest sets `license = "GPL-3.0-or-later"` (ADR-0004). Reuse the SPDX expression string `"GPL-3.0-or-later"`. +- **Workspace:** root `Cargo.toml` is `[workspace]`, with `[workspace.dependencies]` pinning every shared dependency version (spec §2.1). Member crates reference with `dep.workspace = true`. +- **Workspace members (exactly these six, names verbatim):** `crates/rutster` (binary), `crates/rutster-media`, `crates/rutster-call-model`, `crates/rutster-signaling-sip`, `crates/rutster-tap`, `crates/rutster-spend`. +- **Stub-crate policy (spec §2.2):** `rutster-signaling-sip`, `rutster-tap`, `rutster-spend` ship as `lib.rs` with a `//!` module doc comment (what the crate will hold, why deferred, which spearhead step fills it in) + a `#[cfg(test)] mod tests { #[test] fn crate_compiles() {} }` test. No anticipatory code. +- **Dependency direction (spec §2.3):** `rutster` → `{rutster-media, rutster-call-model}`; `rutster-media` → `rutster-call-model`; `rutster-call-model` is a leaf; the three stub crates depend on nothing in the workspace in slice 1. +- **PCM format (spec §3.1, §3.9, ARCHITECTURE.md):** 16-bit signed mono, 24 kHz, fixed 20 ms frame = **480 samples**. `PcmFrame` lives in `rutster-media` (single canonical home; `rutster-tap` re-exports in step 2). +- **str0m API (verified against str0m 0.21 docs.rs):** + - `Rtc::new(start: Instant) -> Self` — takes an `Instant`, NOT argless. Or use `RtcConfig::new().build(Instant)` for non-default config. + - SDP: `let parsed_offer = str0m::change::SdpOffer::from_sdp_string(offer_str)?;` (there is NO `from_str_unchecked` — `from_sdp_string` is the entry point, returns `Result`). + - Then `let answer: SdpAnswer = rtc.sdp_api().accept_offer(parsed_offer)?;` — `accept_offer` takes the owned `SdpOffer`, returns `Result`. `rtc.sdp_api()` borrows rtc; call is `let answer = rtc.sdp_api().accept_offer(offer)?`. + - `answer.to_string()` renders the SDP answer text. + - `Rtc::add_local_candidate(&mut self, c: Candidate) -> Option<&Candidate>` — returns `Some(&Candidate)` if accepted, `None` otherwise. Pass the candidate BEFORE `accept_offer` so it appears in the answer. + - `Candidate::host(addr: SocketAddr, proto: impl TryInto) -> Result` — `"udp"` literal works because `&str: TryInto`. + - Inbound audio events arrive via the Frame API as `Event::MediaData(MediaData)`. `MediaData.data: Arc<[u8]>` is the encoded Opus payload (NOT `Vec` — it's an atomically-refcounted boxed slice; pass `&media.data[..]` to the decoder). + - Outbound: `let writer: Option> = rtc.writer(mid);` (returns `Option`, not `Result` — `None` if direction isn't sending). Then `writer.write(pt, wallclock, rtp_time, data)` where: + - `pt: Pt` — payload type. Get it from `writer.match_params(&incoming_params) -> Option` (recommended — matches the incoming payload params) OR `writer.payload_params()` returns `impl Iterator`, then `params.pt()` accessor. + - `wallclock: Instant` — when the sample was produced (use local `now`). + - `rtp_time: MediaTime` — RTP timestamp. Field name is `rtp_time` (NOT `media_time`). Increment for next 20 ms Opus frame at 48 kHz = `+ MediaTime::from(Duration::from_millis(20))` — use `mt + MediaTime::from(duration)` (there is NO `MediaTime::add(Duration)` method; use `Add`/`AddAssign` with `MediaTime::from(Duration)`). + - `data: impl Into>` — pass `&opus_bytes[..]` or `Vec` (both convert). + - Returns `Result<(), RtcError>`. + - `MediaTime::ZERO` constant exists (`pub const ZERO: MediaTime`). + - Poll loop invariant: mutate → drain `poll_output()` to `Output::Timeout(t)` → mutate again. str0m has NO `Live` struct — `Rtc` is the driver. +- **str0m ICE candidates (spec §3.7):** Add local host candidates via `Candidate::host(addr, "udp")`. ICE public surface at `str0m::` root (no `str0m::ice` module): `Candidate`, `CandidateKind`, `IceCreds`, `IceConnectionState`. +- **opus crate API (verified against opus 0.3.1):** `opus::Decoder::new(24000, opus::Channels::Mono)`, `opus::Encoder::new(24000, opus::Channels::Mono, opus::Application::Voip)`. `decoder.decode(&op[..], &mut pcm[..480], /*fec*/ false) -> Result` (returns samples-per-channel decoded). `encoder.encode_vec(&pcm[..480], /*max_size*/ 4000) -> Result>`. 480 = samples per 20 ms at 24 kHz mono. +- **opus system dependency:** the `opus` crate (via `audiopus_sys`) links system libopus. Build prerequisite: `libopus-dev` (Debian/Ubuntu) or `opus-devel` (Fedora) installed. Documented in `README.md` dev-loop section, with the PORT_PLAN §7 rationale ("🦀 Core (FFI)"). Spec §6.3's "no external deps beyond Rust" is amended by this FFI exception —iber note this in the learner comments. +- **Hot-path error policy (spec §3.8, AGENTS.md):** the 20 ms media loop **never** uses `?`. Match-and-continue. A decode/encode failure is logged + counted (via a minimal counter), the packet is dropped, and the peer is NOT terminated. Cold paths (signaling, setup) use `thiserror`-derived enums and `?` liberally. +- **Code documentation (spec §7, AGENTS.md):** override the default "no comments" convention. `//!` module docs at the top of every `lib.rs`/`main.rs`/sub-module. `///` on every public item. `//` inline comments on *mechanism* (why `Arc>` vs `Arc>`, why `Pin>`, etc.). str0m interactions get an explanatory comment. First occurrence of each non-obvious Rust pattern gets a "why this pattern" comment. +- **Deviation comment (spec §3.4):** the tokio poll loop in `rutster-media/src/loop_driver.rs` carries this verbatim comment: `// DEV-DEVIATION: tokio polling accepted for slice 1; step 4 replaces with dedicated timing thread per ARCHITECTURE.md.` +- **HTTP surface (spec §4.1, §4.3):** axum on `0.0.0.0:8080`, plaintext (no TLS — out of scope). Four routes: `POST /v1/sessions` → `{ "session_id": "" }`; `POST /v1/sessions/:id/offer` (`Content-Type: application/sdp` request+response); `DELETE /v1/sessions/:id`; `GET /` → static HTML. +- **Non-trickle ICE (spec §4.2):** one POST on `/offer` carries browser offer+candidates, response carries core answer+candidates, no separate `/ice` endpoint. +- **Session store (spec §4.5):** `DashMap` in the binary crate. `ChannelId` is a UUID newtype from `rutster-call-model` and IS the session id. +- **Idle timeout (spec §4.5):** 60 s of no RTP packets received → close the session. Implemented as a per-session deadline checked on each poll cycle. No per-session tokio task. +- **Graceful shutdown (spec §4.5):** tokio signal handler drops the `DashMap` on Ctrl-C/SIGTERM. +- **Slice-1 out-of-scope (spec §1.2, AGENTS.md):** the dedicated timing thread, TLS, authn/authz, trickle ICE, the tap itself, the brain, barge-in/VAD, PSTN trunk, spend cap, CDR/event bus, transfer/park/pickup, browser automation, latency benchmark harness, fuzz harnesses are ALL deferred. Adding any of them NOW breaks the spearhead sequencing. spot-check a finding against this list before treating it as a real gap. +- **CI gates (spec §6.2):** `cargo fmt --check`, `cargo clippy -- -D warnings`, `cargo test --all`, `cargo deny check`. CI runs on push + PR to `main`. Matrix: latest stable + the MSRV pinned in `rust-toolchain.toml`. +- **cargo-deny config (spec §6.1):** allow `GPL-3.0-or-later`, `MIT`, `Apache-2.0`, `BSD-3-Clause`, `ISC`, `Zlib`, `Unicode-DFS-2016`, `Unicode-3.0`. `deny warnings` on advisories. Duplicate-version bans on `tokio`, `serde`, `bytes`, `tracing`. Sources: `crates-io` only. +- **Task / PR strategy:** tasks 1–7 are sequentially dependent (1 must land before 2; 2 before 3; 3 before 4; 4 before 5; 6 and 7 can run in parallel with each other after Task 5 lands). Each task's "Commit" step is one commit on `main` (or one PR merging to `main` if branch protection is on). Each task is independently shippable + green (tests pass after each commit). **Merge in numeric order.** Do NOT batch multiple tasks into one commit — the granular history is a load-bearing artifact for the learning-codebase goal (spec §7). If using the `executing-plans` skill rather than `subagent-driven-development`, still emit one commit per task; the plan's commit messages are written for that shape. + +--- + +## File structure (landed shape) + +``` +rutster/ +├── Cargo.toml # [workspace] + [workspace.dependencies] +├── deny.toml # cargo-deny config (Task 6) +├── rust-toolchain.toml # pinned stable (Task 1) +├── LEARNING.md # index (Task 7) +├── .github/workflows/ci.yml # CI (Task 6) +├── crates/ +│ ├── rutster/ # binary (Tasks 5, 6) +│ │ ├── Cargo.toml +│ │ ├── src/main.rs +│ │ ├── src/session_map.rs +│ │ ├── src/routes.rs +│ │ └── static/index.html +│ ├── rutster-media/ # REAL (Tasks 3, 4) +│ │ ├── Cargo.toml +│ │ ├── src/lib.rs # module docs + error + re-exports +│ │ ├── src/pcm.rs # PcmFrame, AudioSource/AudioSink, EchoAudioPipe +│ │ ├── src/opus_codec.rs # decoder/encoder wrappers +│ │ ├── src/loop_driver.rs # str0m poll loop (tokio deviation) +│ │ └── src/rtc_session.rs # RtcSession +│ ├── rutster-call-model/ # REAL-minimal (Task 2) +│ │ ├── Cargo.toml +│ │ └── src/lib.rs # Channel, ChannelId, ChannelState, Direction +│ ├── rutster-signaling-sip/ # STUB (Task 1) +│ │ ├── Cargo.toml +│ │ └── src/lib.rs +│ ├── rutster-tap/ # STUB (Task 1) +│ │ ├── Cargo.toml +│ │ └── src/lib.rs +│ └── rutster-spend/ # STUB (Task 1) +│ ├── Cargo.toml +│ └── src/lib.rs +└── fuzz/ # placeholder dir (Task 7) + └── README.md +``` + +--- + +## Task 1: Workspace scaffold + three stub crates + +**Files:** +- Create: `Cargo.toml` (root workspace manifest) +- Create: `rust-toolchain.toml` +- Create: `crates/rutster-signaling-sip/Cargo.toml` +- Create: `crates/rutster-signaling-sip/src/lib.rs` +- Create: `crates/rutster-tap/Cargo.toml` +- Create: `crates/rutster-tap/src/lib.rs` +- Create: `crates/rutster-spend/Cargo.toml` +- Create: `crates/rutster-spend/src/lib.rs` +- Test: each stub crate's `crate_compiles` test. + +**Interfaces:** +- Consumes: nothing (this is the foundation). +- Produces: a compiling Cargo workspace with three stub crates. Later tasks add the real member crates (`rutster`, `rutster-media`, `rutster-call-model`) by appending to the `members` array — Task 1 leaves `members` listing only the three stubs, and Task 2/3/4/5 each extend it. + +**Note on the `members` array:** start with only the three stub crates listed in `members`. Each subsequent task's "Step N: extend workspace" appends its new crate path to this array. Do NOT pre-list `crates/rutster*` with a glob — strip the glob and use an explicit list so a half-built crate never breaks `cargo metadata`. + +- [ ] **Step 1: Write the root `Cargo.toml`** + +```toml +# Cargo.toml — rutster workspace root. +# Spec ref: slice-1 §2. The workspace pins shared dep versions here so +# member crates can't drift (§2.1). Each member references with +# `dep.workspace = true`. +[workspace] +resolver = "2" +members = [ + "crates/rutster-signaling-sip", + "crates/rutster-tap", + "crates/rutster-spend", +] + +[workspace.package] +license = "GPL-3.0-or-later" +edition = "2021" +repository = "https://github.com/anomalyco/rutster" + +# Pinned versions for all member crates. References are `foo.workspace = true` +# in the member manifest. Keeps the dep tree unified (§2.1). +[workspace.dependencies] +# str0m 0.21: sans-IO WebRTC. Frame API (Event::MediaData + Writer::write). +str0m = "0.21" +# opus 0.3.1: libopus FFI (system libopus required — see README). +opus = "0.3" +# axum 0.7: HTTP signaling surface. +axum = { version = "0.7", features = ["macros"] } +# tokio 1: runtime driving the str0m poll loop (slice-1 deviation per §3.4). +tokio = { version = "1", features = ["full"] } +# dashmap 6: in-process session store. +dashmap = "6" +# uuid 1: ChannelId newtype backing. +uuid = { version = "1", features = ["v4"] } +thiserror = "1" +tracing = "0.1" +tracing-subscriber = { version = "0.3", features = ["env-filter"] } +serde = { version = "1", features = ["derive"] } +serde_json = "1" +# tower: used by the binary crate's integration tests (ServiceExt::oneshot +# on the axum Router). Axum re-exports parts of tower but the integration test +# uses `tower::ServiceExt` directly, so it needs to be a workspace dep. +tower = { version = "0.5", features = ["util"] } +``` + +- [ ] **Step 2: Write `rust-toolchain.toml`** + +Pin stable (currently 1.80 as of writing — confirm the latest stable at impl time with `rustc --version`). The MSRV is whatever str0m 0.21 requires; the CI matrix (Task 6) tests stable + MSRV. + +```toml +# rust-toolchain.toml — pins the toolchain for reproducible builds. +[toolchain] +channel = "1.80" +components = ["rustfmt", "clippy"] +``` + +- [ ] **Step 3: Write `crates/rutster-signaling-sip/Cargo.toml`** + +```toml +# crates/rutster-signaling-sip/Cargo.toml +[package] +name = "rutster-signaling-sip" +version = "0.0.0" +license.workspace = true +edition.workspace = true +repository.workspace = true +description = "Rust-native trunk SIP — stub crate (filled in spearhead step 5)." +``` + +- [ ] **Step 4: Write `crates/rutster-signaling-sip/src/lib.rs`** + +```rust +//! # rutster-signaling-sip +//! +//! **Status:** stub. Fills in at spearhead step 5 (PSTN trunk). +//! +//! This crate will hold the Rust-native trunk SIP stack: the SIP parser, +//! transaction layer, dialog state, and the carrier trunk integration. See +//! [ADR-0003](../../../docs/adr/0003-sip-rust-native-trunk.md) for the +//! "own the parser from day one" thesis and [PORT_PLAN §1](../../../docs/PORT_PLAN.md) +//! for the surface area (`res_pjsip_session`, `chan_sip`, `_sdp_rtp` rows). +//! +//! Slice 1's WebRTC-only ingress needs no SIP — this stub exists to lock the +//! crate boundary without anticipating code (spec §2.2). It depends on +//! nothing in the workspace in slice 1. Its future dependency direction is +//! `rutster-signaling-sip` → `rutster-call-model` + `rutster-media` (once +//! the SDP help lives here, moved out of `rutster-media`'s WebRTC-ICE-coupled +//! SDP module — see §3.7 of the slice-1 spec for the split rationale). + +#[cfg(test)] +mod tests { + /// Stub crates lock boundaries; the compile-test is the lock. + #[test] + fn crate_compiles() {} +} +``` + +- [ ] **Step 5: Write `crates/rutster-tap/Cargo.toml`** + +```toml +# crates/rutster-tap/Cargo.toml +[package] +name = "rutster-tap" +version = "0.0.0" +license.workspace = true +edition.workspace = true +repository.workspace = true +description = "Agent audio tap — stub crate (filled in spearhead step 2)." +``` + +- [ ] **Step 6: Write `crates/rutster-tap/src/lib.rs`** + +```rust +//! # rutster-tap +//! +//! **Status:** stub. Fills in at spearhead step 2 (the tap itself). +//! +//! Slice 1 *pre-paves* the tap by exposing the canonical PCM boundary as +//! the `AudioSource` / `AudioSink` traits in [`rutster_media`](../rutster-media/index.html), +//! and wires an `EchoAudioPipe` between sink and source. Step 2 swaps that +//! pipe for a real WSS tap client (core-as-client, brain-as-server — +//! [ADR-0006](../../../docs/adr/0006-ingress-posture.md)). No code changes to +//! `RtcSession` itself in step 2 — that's the test of the seam. +//! +//! This crate will, when filled in, re-export `PcmFrame` from +//! `rutster-media` (one canonical home — spec §3.1) and ship the WSS +//! tap client + the versioned framing protocol. It depends on nothing +//! in the workspace in slice 1. + +#[cfg(test)] +mod tests { + #[test] + fn crate_compiles() {} +} +``` + +- [ ] **Step 7: Write `crates/rutster-spend/Cargo.toml`** + +```toml +# crates/rutster-spend/Cargo.toml +[package] +name = "rutster-spend" +version = "0.0.0" +license.workspace = true +edition.workspace = true +repository.workspace = true +description = "In-boundary spend / abuse gate — stub crate (filled in spearhead step 6)." +``` + +- [ ] **Step 8: Write `crates/rutster-spend/src/lib.rs`** + +```rust +//! # rutster-spend +//! +//! **Status:** stub. Fills in at spearhead step 6 (spend cap / abuse gate). +//! +//! In-boundary spend and abuse control is constitutive of the wedge +//! ([ADR-0002](../../../docs/adr/0002-north-star-and-fused-core.md)): the +//! runaway brain structurally cannot exceed spend or pacing because it +//! doesn't hold the wire — the trunk termination + spend gate do, in one +//! boundary. Pulling spend out into a service re-introduces the 3-vendor +//! structural hole the fused vertical was chosen to eliminate. +//! +//! This crate will hold: spend caps, pacing caps, deny-by-default routing, +//! rate-limits, toll-fraud pattern detection — co-located with trunk +//! termination in `rutster-signaling-sip` (step 5). Depends on nothing in +//! the workspace in slice 1. + +#[cfg(test)] +mod tests { + #[test] + fn crate_compiles() {} +} +``` + +- [ ] **Step 9: Run `cargo check` to verify the workspace compiles** + +Run: `cargo check --all` +Expected: 3 stub crates compile cleanly; no warnings. + +- [ ] **Step 10: Run `cargo test --all` to verify the stub tests pass** + +Run: `cargo test --all` +Expected: 3 tests, all passing (`crate_compiles` in each stub). + +- [ ] **Step 11: Commit** + +```bash +git add Cargo.toml rust-toolchain.toml crates/rutster-signaling-sip crates/rutster-tap crates/rutster-spend +git commit -m "workspace: scaffold + three stub crates (sip/tap/spend) + +Workspace root, pinned toolchain, and the three stub crates whose only +job in slice 1 is to lock the ADR-0002 boundary shape. Each ships a +lib.rs module doc (what it will hold, why deferred, which spearhead step +fills it) and a crate_compiles test. Spec §2.2." +``` + +--- + +## Task 2: `rutster-call-model` — the Channel embryo + +**Files:** +- Create: `crates/rutster-call-model/Cargo.toml` +- Create: `crates/rutster-call-model/src/lib.rs` +- Modify: `Cargo.toml` (workspace root — add the new member to `members`). + +**Interfaces:** +- Consumes: nothing in the workspace (leaf crate, spec §5.3). +- Produces: `Channel`, `ChannelId`, `ChannelState`, `Direction`. `ChannelId` is a `Uuid` newtype (spec §5.1) — it IS the session id surfaced in the REST API (spec §4.5). `ChannelState` is `New | Connecting | Connected | Closing | Closed` (spec §5.1, §5.4). `Direction` is `Inbound` only in slice 1. + +- [ ] **Step 1: Write the failing test for `ChannelId` newtype** + +Add to `crates/rutster-call-model/src/lib.rs` (write the whole file with `lib.rs` containing the test first; that's allowed — TDD writes the test before the impl, not necessarily in a separate file). + +```rust +//! # rutster-call-model +//! +//! The unifying leg object: a `Channel` is one peer / one leg, the object +//! the future API will model (PORT_PLAN §3 — "the unifying leg object"). +//! Building a throwaway `LoopbackPeer` for slice 1 and refactoring it +//! later is the exact failure mode the design rules warn against, so the +//! slice-1 peer *is* a `Channel` (spec §5.2). +//! +//! Slice 1 ships the signaling-state embryo only (spec §5.4). Media state +//! is internal to `rutster-media`; the split — "Channel = signaling state; +//! media = leaf concern" — matches ARCHITECTURE.md's "call model as the +//! unifying object." Media state moves UP into the `Channel` only when a +//! second consumer (the API, the tap, an audiohook) needs to observe it. + +#[cfg(test)] +mod tests { + use super::*; + + /// ChannelId must be a newtype around Uuid, NOT a bare Uuid — the + /// newtype pattern prevents us from mixing up a ChannelId with some + /// future SessionId at the type-system level. The compiler enforces + /// what a comment can only ask for. + #[test] + fn channel_id_is_a_newtype() { + let id = ChannelId::new(); + // Newtype wraps Uuid; we can reach the inner id but the outer + // type is what the API surface speaks in. + let _inner: Uuid = id.0; + assert_eq!(format!("{}", id.0).len(), 36); // canonical UUID v4 length + } + + #[test] + fn channel_starts_in_new_state() { + let ch = Channel::new_inbound(); + assert_eq!(ch.state, ChannelState::New); + assert_eq!(ch.direction, Direction::Inbound); + } + + #[test] + fn channel_state_transitions_match_spec_5_4() { + let mut ch = Channel::new_inbound(); + assert_eq!(ch.state, ChannelState::New); + ch.state = ChannelState::Connecting; + ch.state = ChannelState::Connected; + ch.state = ChannelState::Closing; + ch.state = ChannelState::Closed; + } +} +``` + +This will NOT compile yet — `Channel`, `ChannelId`, `ChannelState`, `Direction`, `Uuuid`, `Channel::new_inbound` are not defined. + +- [ ] **Step 2: Run the test to verify it fails** + +Run: `cargo test -p rutster-call-model` +Expected: FAIL with compile errors (`cannot find type ChannelId`, etc). + +- [ ] **Step 3: Write `crates/rutster-call-model/Cargo.toml`** + +```toml +# crates/rutster-call-model/Cargo.toml +[package] +name = "rutster-call-model" +version = "0.0.0" +license.workspace = true +edition.workspace = true +repository.workspace = true +description = "The Channel / leg object embryo (signaling-state only in slice 1)." + +[dependencies] +uuid = { workspace = true } + +[dev-dependencies] +``` + +- [ ] **Step 4: Implement the types in `crates/rutster-call-model/src/lib.rs`** + +Append the implementation block AFTER the `#[cfg(test)] mod tests` block written in Step 1 (the test block stays at the top — that's the pattern from writing-plans: test first, then make it compile). + +```rust +use std::time::Instant; +use uuid::Uuid; + +/// Newtype wrapping a `Uuid` for the channel id. +/// +/// # Why a newtype (not a bare `Uuid`?) +/// Newtypes give zero-cost type safety. If we used bare `Uuid` everywhere, +/// nothing in the type system would stop us from passing a `SessionId` +/// into a function expecting a `ChannelId`. With `ChannelId(Uuid)`, the +/// compiler rejects that mixup at the call site. The pattern is taught +/// in the Rust Book's "Using the Newtype Pattern for Type Safety and +/// Abstraction" section — `ChannelId` is the slice-1 worked example. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] +pub struct ChannelId(pub Uuid); + +impl ChannelId { + /// Mint a fresh `ChannelId`. Slice 1 uses UUID v4 — opaque, random, + /// no coordination. A future multi-tenant deployment would scope by + /// tenant prefix; that lands with authz (step 6). + pub fn new() -> Self { + Self(Uuid::new_v4()) + } +} + +impl Default for ChannelId { + fn default() -> Self { + Self::new() + } +} + +impl std::fmt::Display for ChannelId { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + write!(f, "{}", self.0) + } +} + +/// Signaling state machine for a `Channel` (spec §5.4, slice 1). +/// +/// `New → Connecting → Connected → Closing → Closed` +/// +/// # Why an enum (not a struct with a `kind: &str` field?) +/// Enums model a closed set of states; exhaustiveness checking forces +/// every `match` to consider each state explicitly. When step 4 adds +/// `Closing`'s sub-state for "graceful close in flight," it'll be a new +/// variant or a wrapping struct; either way, the compiler tells us +/// every site that needs updating. A `kind: String` field would let +/// new states slip in silently. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum ChannelState { + /// `POST /v1/sessions` created the Channel; no offer yet. + New, + /// Offer received, ICE gathering / DTLS handshake in progress. + Connecting, + /// ICE+DTLS connected, RTP flowing, audio echoing. + Connected, + /// `DELETE /v1/sessions/:id` or peerconnectionclose; cleaning up. + Closing, + /// Resources dropped, entry removed from the DashMap. + Closed, +} + +/// Direction of the leg (spec §5.1). +/// +/// Slice 1 is browser-initiated → `Inbound` only. `Outbound` lands with +/// the dialer (later rung). The enum exists now so the API has a stable +/// shape — adding `Outbound` later is a non-breaking addition. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum Direction { + Inbound, + // Outbound lands with the dialer (later). NOT present in slice 1. +} + +/// The unifying leg object — one peer = one `Channel` (spec §5.1). +/// +/// Slice 1 carries signaling state only. Fields that arrive later, listed +/// in spec §5.6, are absent by design — adding them is a backwards- +/// compatible field add: +/// - `media: Option` — second consumer. +/// - `audiohooks: Vec` — escalation rung 2. +/// - `tap: Option` — step 2. +#[derive(Debug)] +pub struct Channel { + pub id: ChannelId, + pub state: ChannelState, + pub direction: Direction, + /// For the 60 s idle timeout (spec §4.5). `Instant` is a monotonic + /// clock — choosing it over `SystemTime` means we're measuring + /// elapsed wall-time within this process, NOT a calendar time the + /// user could change mid-call. The monotonic clock is the right + /// tool for "has this peer been silent for 60 seconds?" + pub created_at: Instant, +} + +impl Channel { + /// Construct a fresh inbound channel — the only slice-1 path. + pub fn new_inbound() -> Self { + Self { + id: ChannelId::new(), + state: ChannelState::New, + direction: Direction::Inbound, + created_at: Instant::now(), + } + } +} +``` + +- [ ] **Step 5: Add `crates/rutster-call-model` to the workspace `members`** + +Modify root `Cargo.toml` — append the new member to the `members` array: + +```toml +members = [ + "crates/rutster-call-model", + "crates/rutster-signaling-sip", + "crates/rutster-tap", + "crates/rutster-spend", +] +``` + +- [ ] **Step 6: Run the tests to verify they pass** + +Run: `cargo test -p rutster-call-model` +Expected: 3 tests passing (`channel_id_is_a_newtype`, `channel_starts_in_new_state`, `channel_state_transitions_match_spec_5_4`). + +- [ ] **Step 7: Run clippy + fmt across the workspace** + +Run: `cargo fmt --check && cargo clippy --all -- -D warnings` +Expected: no formatting drift, no warnings. + +- [ ] **Step 8: Commit** + +```bash +git add Cargo.toml crates/rutster-call-model +git commit -m "call-model: Channel + ChannelId + ChannelState (signaling embryo) + +rutster-call-model is real-but-minimal (spec §5): the unifying leg +object the future API exposes. ChannelId is a Uuid newtype for +type-safety (the slice-1 worked example of the newtype pattern). +Channel is signaling-state only — media lives in rutster-media as a +leaf concern of the Channel, surfaced only when a second consumer needs +to observe it (spec §5.3). ChannelState matches the New→Connecting→ +Connected→Closing→Closed flow from §5.4." +``` + +--- + +## Task 3: `rutster-media` — PCM frame, tap seam traits, Opus codec pair + +**Files:** +- Create: `crates/rutster-media/Cargo.toml` +- Create: `crates/rutster-media/src/lib.rs` (module docs + error enum + re-exports) +- Create: `crates/rutster-media/src/pcm.rs` (`PcmFrame`, `AudioSource`, `AudioSink`, `EchoAudioPipe`) +- Create: `crates/rutster-media/src/opus_codec.rs` (`OpusDecoder`, `OpusEncoder`) +- Modify: `Cargo.toml` (workspace root — add member). + +**Interfaces:** +- Consumes: `ChannelId`, `Channel` from Task 2's `rutster-call-model`. +- Produces: + - `PcmFrame` — the canonical 480-sample i16 mono @ 24 kHz frame (spec §3.1, §3.9). + - `AudioSource` / `AudioSink` traits (spec §3.3) — the seam step 2 splices the tap into. + - `EchoAudioPipe` — implements both traits; slice-1 wiring (spec §3.3). + - `OpusDecoder::decode(&[u8]) -> Option` / `OpusEncoder::encode(&PcmFrame) -> Option>` — hot-path match-and-continue, no `?`. + +- [ ] **Step 1: Write `crates/rutster-media/Cargo.toml`** + +```toml +# crates/rutster-media/Cargo.toml +[package] +name = "rutster-media" +version = "0.0.0" +license.workspace = true +edition.workspace = true +repository.workspace = true +description = "Media core: str0m WebRTC + Opus⇄PCM boundary (slice 1)." + +[dependencies] +rutster-call-model = { path = "../rutster-call-model" } +opus = { workspace = true } +thiserror = { workspace = true } +tracing = { workspace = true } + +[dev-dependencies] +``` + +- [ ] **Step 2: Write the failing test for `PcmFrame` + `EchoAudioPipe`** + +Create `crates/rutster-media/src/pcm.rs` with tests first, no impl yet: + +```rust +//! # PCM frame + tap seam (spec §3.3) +//! +//! The canonical tap format from ARCHITECTURE.md: 16-bit signed mono PCM +//! @ 24 kHz, fixed 20 ms = 480 samples. The single format every future +//! brain/tap consumer speaks. Lives in `rutster-media` (spec §3.1); +//! `rutster-tap` re-exports it in step 2 (single canonical home). +//! +//! The `AudioSource`/`AudioSink` traits are the exact splice point where +//! step 2 connects a real tap client (replacing `EchoAudioPipe`). + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn pcm_frame_holds_480_samples() { + let frame = PcmFrame::zeroed(); + assert_eq!(frame.samples.len(), SAMPLES_PER_FRAME); + assert!(frame.samples.iter().all(|&s| s == 0)); + } + + #[test] + fn echo_pipe_round_trips_a_frame() { + // EchoAudioPipe implements both AudioSink and AudioSource. + // Push a frame in via the sink; pull it back out via the source. + let mut pipe = EchoAudioPipe::new(); + assert!(pipe.next_pcm_frame().is_none()); // empty → silence + + let mut frame = PcmFrame::zeroed(); + frame.samples[0] = 1234; + pipe.on_pcm_frame(frame); + + let out = pipe.next_pcm_frame().expect("echoed frame present"); + assert_eq!(out.samples[0], 1234); + assert!(pipe.next_pcm_frame().is_none()); // drained + } + + #[test] + fn sink_must_not_block() { + // The echo pipe is bounded: push more frames than it can hold, + // and on_pcm_frame must drop the oldest silently rather than block. + // (Hot-path invariant from spec §3.3: "Must not block.") + let mut pipe = EchoAudioPipe::new(); + const OVERFLOW: usize = ECHO_BUFFER_LEN + 5; + for i in 0..OVERFLOW { + let mut f = PcmFrame::zeroed(); + f.samples[0] = i as i16; + pipe.on_pcm_frame(f); // must not panic, must not block + } + // We should hold at most ECHO_BUFFER_LEN frames; the rest dropped. + let mut count = 0; + while pipe.next_pcm_frame().is_some() { + count += 1; + } + assert_eq!(count, ECHO_BUFFER_LEN); + } +} +``` + +- [ ] **Step 3: Run the tests to verify they fail** + +Run: `cargo test -p rutster-media pcm::tests` +Expected: FAIL with compile errors (`cannot find type PcmFrame`, etc). + +- [ ] **Step 4: Implement `PcmFrame`, `AudioSource`, `AudioSink`, `EchoAudioPipe`** + +Append to `crates/rutster-media/src/pcm.rs` (above the test mod): + +```rust +use std::collections::VecDeque; + +/// Samples per 20 ms frame @ 24 kHz mono (spec §3.9). +/// +/// 24000 Hz × 0.020 s = 480. This is a `const`, not a magic literal, so +/// every place that needs a 480-sample buffer reads the same named value. +pub const SAMPLES_PER_FRAME: usize = 480; + +/// Capacity of the echo pipe's internal queue (spec §3.3: "must not +/// block"). 3 frames = 60 ms of buffering — enough to absorb jitter +/// without unbounded growth. Slice 1 has no jitter buffer of its own; +/// str0m's adaptive jitter (it doesn't have one — see str0m FAQ) is +/// not in play because we use the Frame API, which delivers already- +/// depacketized frames. This queue is our only playout buffer. +pub const ECHO_BUFFER_LEN: usize = 3; + +/// Canonical PCM frame (spec §3.1, §3.9, ARCHITECTURE.md). +/// +/// 16-bit signed mono @ 24 kHz, 480 samples (20 ms). `i16` is the +/// native PCM sample type on the wire — every brain/tap consumer speaks +/// this format. The slice (not a `Vec`) keeps the frame fixed-size and +/// cheap to copy through the audio pipe. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct PcmFrame { + pub samples: [i16; SAMPLES_PER_FRAME], +} + +impl PcmFrame { + /// A frame of digital silence (all zeros). Used as the "no audio to + /// send" fallback on the source side (spec §3.3: `None = silence`). + pub fn zeroed() -> Self { + Self { + samples: [0; SAMPLES_PER_FRAME], + } + } +} + +/// Produces frames to send to the peer (spec §3.3). +/// +/// The poll loop calls `next_pcm_frame()` on each 20 ms tick. `None` +/// means "send silence" — the caller (loop driver) writes a comfort- +/// noise Opus frame instead of dropping the packet entirely, keeping +/// the RTP clock alive. (In slice 1, silence IS fine — str0m handles +/// pacing — but the `None` semantics encode the "no audio available" +/// case cleanly for step 2's tap client.) +pub trait AudioSource: Send { + fn next_pcm_frame(&mut self) -> Option; +} + +/// Consumes decoded frames from the peer (spec §3.3). +/// +/// `on_pcm_frame` MUST NOT block — the 20 ms loop is the only caller, +/// and blocking here delays the next poll past its deadline. The +/// `EchoAudioPipe` enforces this by bounding its queue and dropping +/// the oldest frame on overflow (see `tests::sink_must_not_block`). +pub trait AudioSink: Send { + fn on_pcm_frame(&mut self, frame: PcmFrame); +} + +/// Slice-1 wiring of the tap seam: a bounded queue connecting inbound +/// (sink) to outbound (source) — an echo (spec §3.3). Step 2 replaces +/// this with a real WSS tap client; no changes to `RtcSession`. +/// +/// # Why `VecDeque` (not `tokio::mpsc` or `crossbeam`?) +/// The echo pipe lives behind a single `Arc>` in the +/// `RtcSession`, polled by a single tokio task. There is exactly one +/// producer (inbound decode) and one consumer (outbound encode), both +/// in the same poll loop — no cross-task messaging. A `VecDeque` under +/// the same mutex is the smallest structure that fits; a channel would +/// add async machinery we don't need in slice 1 (and would pre-pave +/// the wrong pattern for step 4's dedicated thread). +pub struct EchoAudioPipe { + queue: VecDeque, +} + +impl EchoAudioPipe { + pub fn new() -> Self { + Self { + queue: VecDeque::with_capacity(ECHO_BUFFER_LEN), + } + } + + /// Push a frame; if full, drop the oldest. Non-blocking by construction. + fn push_back_bounded(&mut self, frame: PcmFrame) { + if self.queue.len() >= ECHO_BUFFER_LEN { + self.queue.pop_front(); + } + self.queue.push_back(frame); + } +} + +impl Default for EchoAudioPipe { + fn default() -> Self { + Self::new() + } +} + +impl AudioSink for EchoAudioPipe { + fn on_pcm_frame(&mut self, frame: PcmFrame) { + self.push_back_bounded(frame); + } +} + +impl AudioSource for EchoAudioPipe { + fn next_pcm_frame(&mut self) -> Option { + self.queue.pop_front() + } +} +``` + +- [ ] **Step 5: Run the tests to verify they pass** + +Run: `cargo test -p rutster-media pcm::tests` +Expected: 3 tests passing. + +- [ ] **Step 6: Write the failing test for the Opus codec pair** + +Create `crates/rutster-media/src/opus_codec.rs`: + +```rust +//! # Opus ⇄ PCM codec pair (spec §3.1) +//! +//! Wraps the `opus` crate's libopus FFI into the slice-1 hot-path +//! shape: decode returns `Option` and encode returns +//! `Option>` — match-and-continue, no `?`, no error propagation +//! on the 20 ms loop (spec §3.8). A dropped frame is logged + counted; +//! the peer is NOT terminated. +//! +//! The wrapping type exists (rather than using `opus::Decoder` inline) +//! so the slice-1 `RtcSession` can hold `OpusDecoder` / `OpusEncoder` +//! as concrete types without re-stating the sample rate and channel +//! count at every call site. + +use crate::pcm::{PcmFrame, SAMPLES_PER_FRAME}; + +#[cfg(test)] +mod tests { + use super::*; + + /// Encode a known PCM signal → decode the result → assert the RMS + /// is within tolerance. This is the roundtrip test from spec §6.4 + /// ("encode known PCM → decode → assert RMS within tolerance"). + #[test] + fn opus_roundtrip_preserves_signal_within_tolerance() { + let mut enc = OpusEncoder::new().expect("encoder"); + let mut dec = OpusDecoder::new().expect("decoder"); + + // A pure 440 Hz tone at modest amplitude — easy to encode losslessly. + let mut input = PcmFrame::zeroed(); + for (i, s) in input.samples.iter_mut().enumerate() { + let phase = 2.0 * std::f32::consts::PI * 440.0 * (i as f32) / 24_000.0; + *s = (phase.sin() * 8000.0) as i16; // ~ -14 dBFS, comfortable for Opus + } + + let opus_bytes = enc.encode(&input).expect("encoded"); + assert!(!opus_bytes.is_empty(), "Opus payload non-empty"); + + let decoded = dec.decode(&opus_bytes).expect("decoded PCM"); + // Per-sample comparison fails (Opus is lossy); RMS comparison passes. + let in_rms = rms(&input.samples); + let out_rms = rms(&decoded.samples); + // Opus at Voip mode preserves energy to ~10% at this amplitude. + let rel = (in_rms - out_rms).abs() / in_rms.max(1.0); + assert!( + rel < 0.15, + "RMS drift {rel:.3} exceeds tolerance: in={in_rms}, out={out_rms}" + ); + } + + #[test] + fn decoder_returns_none_on_garbage_payload() { + // Hot-path contract: decode failure → None, not a panic. + // Spec §3.8: "drop + observe, don't crash." + let mut dec = OpusDecoder::new().expect("decoder"); + let garbage = [0u8; 8]; + let out = dec.decode(&garbage); + assert!(out.is_none(), "garbage payload must not panic, must return None"); + } + + fn rms(samples: &[i16; SAMPLES_PER_FRAME]) -> f32 { + let sum_sq: f64 = samples.iter().map(|&s| (s as f64).powi(2)).sum(); + (sum_sq / samples.len() as f64).sqrt() as f32 + } +} +``` + +- [ ] **Step 7: Run the tests to verify they fail** + +Run: `cargo test -p rutster-media opus_codec::tests` +Expected: FAIL with compile errors (`cannot find type OpusDecoder`, etc). + +- [ ] **Step 8: Implement `OpusDecoder` and `OpusEncoder`** + +Append to `crates/rutster-media/src/opus_codec.rs` (above the test mod): + +```rust +use opus::{Application, Channels, Decoder as LibDecoder, Encoder as LibEncoder}; + +use crate::pcm::PcmFrame; + +/// 24 kHz mono — the slice-1 default (spec §3.9, ARCHITECTURE.md). +const SAMPLE_RATE: u32 = 24_000; + +/// Initializes the decoder with one-channel output. libopus accepts 24 kHz +/// as a standard rate — no resample needed downstream. +const CHANNELS: Channels = Channels::Mono; + +/// Voip mode — optimized for speech, which is the slice-1 (and product) +/// workload. `Application::Audio` is for music; `LowDelay` sacrifices +/// quality for ~5 ms less latency, unjustified at slice 1's ~200 ms bar. +const APPLICATION: Application = Application::Voip; + +/// Upper bound on an Opus 20 ms frame payload at 24 kHz. The recommended +/// max from libopus is ~4000 bytes; we allocate once and reuse. +const MAX_OPUS_PAYLOAD_BYTES: usize = 4000; + +/// Wraps `opus::Decoder` so the loop driver doesn't re-state the sample +/// rate and channels at each call. +pub struct OpusDecoder { + inner: LibDecoder, + // Reusable decode buffer: avoids allocating 480 i16s per frame on the + // hot path. `Option` would also work; a flat array keeps the + // reuse obvious. + pcm_buf: [i16; SAMPLES_PER_FRAME], +} + +impl OpusDecoder { + pub fn new() -> Result { + Ok(Self { + inner: LibDecoder::new(SAMPLE_RATE, CHANNELS)?, + pcm_buf: [0; SAMPLES_PER_FRAME], + }) + } + + /// Decode an Opus payload to a `PcmFrame`. Returns `None` on any + /// decode error — hot-path contract is match-and-continue (spec §3.8). + /// The caller (loop driver) logs + counts a drop, never propagates. + pub fn decode(&mut self, opus_payload: &[u8]) -> Option { + // FEC (forward error correction) is false in slice 1 — we don't + // request the previous frame's FEC data. Step 4 (barge-in) may + // revisit; FEC matters under lossy networks, not loopback. + match self.inner.decode(opus_payload, &mut self.pcm_buf, /*fec*/ false) { + Ok(_samples_decoded) => Some(PcmFrame { samples: self.pcm_buf }), + Err(e) => { + tracing::warn!(error = ?e, "opus decode dropped; continuing"); + None + } + } + } +} + +/// Wraps `opus::Encoder` for the same reason as the decoder wrapper. +pub struct OpusEncoder { + inner: LibEncoder, +} + +impl OpusEncoder { + pub fn new() -> Result { + Ok(Self { + inner: LibEncoder::new(SAMPLE_RATE, CHANNELS, APPLICATION)?, + }) + } + + /// Encode a `PcmFrame` to an Opus payload. Returns `None` on any + /// encode error — same hot-path contract as `OpusDecoder::decode`. + /// Uses `encode_vec` (allocates a fresh `Vec` per call) for + /// slice 1 simplicity; a production hot path would reuse a buffer + /// passed in by the caller to avoid per-frame allocation. + pub fn encode(&mut self, frame: &PcmFrame) -> Option> { + match self + .inner + .encode_vec(&frame.samples, MAX_OPUS_PAYLOAD_BYTES) + { + Ok(payload) => Some(payload), + Err(e) => { + tracing::warn!(error = ?e, "opus encode dropped; continuing"); + None + } + } + } +} +``` + +- [ ] **Step 9: Write `crates/rutster-media/src/lib.rs` (module docs + error enum + re-exports)** + +```rust +//! # rutster-media +//! +//! The media core: str0m WebRTC termination + the Opus⇄PCM boundary +//! (spec §3). One per WebRTC peer; a `RtcSession` owns a `str0m::Rtc` +//! instance + an Opus encoder/decoder pair + an `EchoAudioPipe` +//! wiring the inbound decode path to the outbound encode path. +//! +//! ## Architecture references +//! +//! - [slice-1 spec §3](../../../docs/superpowers/specs/2026-06-28-slice-1-webrtc-loopback-design.md) +//! — full media-core design. +//! - [ARCHITECTURE.md](../../../docs/ARCHITECTURE.md) — fused per-call +//! vertical; the tap is the central interface; PCM tap format. +//! - [ADR-0002](../../../docs/adr/0002-north-star-and-fused-core.md) — +//! fused vertical + the in-boundary spend gate. +//! +//! ## Error handling posture (spec §3.8) +//! +//! Cold path (RTc construction, codec init): `thiserror`-derived errors + `?`. +//! Hot path (the 20 ms loop): **never** `?`. Match-and-continue. A +//! dropped packet MUST NOT terminate the peer. Policy: "drop + observe +//! (log + counter), don't crash." This is the posture the eventual fuzz +//! harness (step 5) will test against. +//! +//! ## Module map +//! +//! - [`pcm`] — `PcmFrame` + `AudioSource`/`AudioSink` traits (the tap +//! seam) + `EchoAudioPipe` (slice-1 wiring). +//! - [`opus_codec`] — `OpusDecoder`/`OpusEncoder` wrappers. +//! - [`loop_driver`] (Task 4) — the str0m poll loop on tokio. +//! - [`rtc_session`] (Task 4) — `RtcSession`, the per-peer owner. + +pub mod opus_codec; +pub mod pcm; + +pub use opus_codec::{OpusDecoder, OpusEncoder}; +pub use pcm::{AudioSink, AudioSource, EchoAudioPipe, PcmFrame, SAMPLES_PER_FRAME}; + +use thiserror::Error; + +/// Cold-path errors for media-core construction. Hot-path failures go +/// through the "match-and-continue" `Option<_>` returns on +/// `OpusDecoder::decode` / `OpusEncoder::encode`, NOT through this enum. +#[derive(Debug, Error)] +pub enum MediaError { + #[error("opus codec initialization failed: {0}")] + CodecInit(#[from] opus::Error), +} +``` + +- [ ] **Step 10: Add `crates/rutster-media` to the workspace `members`** + +Modify root `Cargo.toml`: + +```toml +members = [ + "crates/rutster-call-model", + "crates/rutster-media", + "crates/rutster-signaling-sip", + "crates/rutster-tap", + "crates/rutster-spend", +] +``` + +- [ ] **Step 11: Run the full test suite** + +Run: `cargo test -p rutster-media` +Expected: all `pcm::tests` + `opus_codec::tests` passing. + +- [ ] **Step 12: Run clippy + fmt** + +Run: `cargo fmt --check && cargo clippy -p rutster-media -- -D warnings` +Expected: clean. + +- [ ] **Step 13: Commit** + +```bash +git add Cargo.toml crates/rutster-media +git commit -m "media: PcmFrame + AudioSource/Sink + Opus codec pair + +PcmFrame is the canonical tap format (16-bit mono @ 24 kHz, 480 samples +per 20 ms frame — ARCHITECTURE.md). AudioSource/AudioSink are the seam +step 2 splices the tap client into (spec §3.3); EchoAudioPipe is the +slice-1 wiring of that seam. OpusDecoder/OpusEncoder wrap the opus +crate's libopus FFI with hot-path match-and-continue (no ? on the 20 ms +loop, spec §3.8); decode/encode return Option/Option> +so a dropped frame is logged + counted, never propagated to crash the +peer." +``` + +--- + +## Task 4: `RtcSession` + str0m poll loop (the media core's heart) + +**Files:** +- Create: `crates/rutster-media/src/rtc_session.rs` +- Create: `crates/rutster-media/src/loop_driver.rs` +- Modify: `crates/rutster-media/src/lib.rs` (declare the new modules + re-exports). +- Modify: `crates/rutster-media/Cargo.toml` (add `str0m` dep). + +**Interfaces:** +- Consumes: `PcmFrame`, `AudioSource`, `AudioSink`, `EchoAudioPipe`, `OpusDecoder`, `OpusEncoder` from Task 3; `Channel`, `ChannelId`, `ChannelState` from Task 2. +- Produces: + - `RtcSession` — owns `str0m::Rtc` + `Channel` + `OpusDecoder` + `OpusEncoder` + `EchoAudioPipe` + a UDP socket (`std::net::UdpSocket`, driven by tokio) + idle-deadline bookkeeping (spec §4.5). + - `RtcSession::accept_offer(sdp_offer: &str) -> Result` — drives str0m's `sdp_api().accept_offer()`, returns the SDP answer (with DTLS fingerprint + ICE creds + Opus codec, all native to str0m 0.21 — NO hand-rolled SDP munger). + - `RtcSession::run_poll_once(now: Instant) -> Option` — one iteration of the sans-IO poll loop; returns the next timeout. The binary's tokio task loops this. (Slice-1 deviation: the loop is on tokio, not a dedicated thread — spec §3.4.) + - `RtcSession::channel_id() -> ChannelId`. + - `RtcSession::is_closed() -> bool`. + +- [ ] **Step 1: Update `crates/rutster-media/Cargo.toml` to add str0m** + +```toml +[dependencies] +rutster-call-model = { path = "../rutster-call-model" } +opus = { workspace = true } +str0m = { workspace = true } +thiserror = { workspace = true } +tracing = { workspace = true } +``` + +- [ ] **Step 2: Declare the new modules in `crates/rutster-media/src/lib.rs`** + +Edit the `lib.rs` written in Task 3 — replace its module map with the populated version. The implementation block at the bottom stays; only the module declarations + re-exports change: + +```rust +pub mod loop_driver; +pub mod opus_codec; +pub mod pcm; +pub mod rtc_session; + +pub use opus_codec::{OpusDecoder, OpusEncoder}; +pub use pcm::{AudioSink, AudioSource, EchoAudioPipe, PcmFrame, SAMPLES_PER_FRAME}; +pub use rtc_session::{RtcSession, RtcSessionError}; +``` + +(Keep the rest of the `lib.rs` from Task 3 — the `MediaError` enum + module docs — unchanged.) + +- [ ] **Step 3: Write the failing test for `RtcSession::accept_offer`** + +Create `crates/rutster-media/src/rtc_session.rs` with tests first. The real-browser-offer fixture (a full SDP from a browser) is captured in a test constant; the test verifies `accept_offer` produces a valid SDP answer containing an Opus payload type and a DTLS fingerprint. + +```rust +//! # `RtcSession` — the per-peer media owner (spec §3.1, §4.5) +//! +//! Owns a `str0m::Rtc` instance + an Opus decoder/encoder pair + an +//! `EchoAudioPipe` wiring inbound to outbound + the per-peer UDP socket. +//! One per WebRTC peer. The `ChannelId` (from `rutster-call-model`) is +//! the session id surfaced in the REST API. +//! +//! ## What str0m does for us (so we don't) +//! +//! str0m 0.21's `Rtc::sdp_api().accept_offer(offer)` produces the SDP +//! answer natively: DTLS fingerprint (from the cert str0m generates), ICE +//! ufrag/pwd, and codec negotiation (Opus, the only codec we registered). +//! Slice 1 does NOT hand-roll an SDP munger — str0m's path is the spec's +//! "embryo of the future SIP SDP path" (§3.7). When step 5 brings SIP/SDP +//! negotiation into `rutster-signaling-sip`, that crate may extract shared +//! SDP helpers from str0m or build its own. Slice 1's WebRTC-ICE-coupled +//! SDP lives entirely in str0m. + +use std::net::SocketAddr; +use std::time::{Duration, Instant}; + +use rutster_call_model::{Channel, ChannelId, ChannelState}; +use str0m::Rtc; +use thiserror::Error; + +use crate::opus_codec::{OpusDecoder, OpusEncoder}; +use crate::pcm::{AudioSink, AudioSource, EchoAudioPipe}; + +/// Per-session idle timeout (spec §4.5): 60 s of no RTP from the peer +/// → close. RTC quiet periods are normal but 60 s of dead air means +/// "the browser tab is dead." +const IDLE_TIMEOUT: Duration = Duration::from_secs(60); + +#[derive(Debug, Error)] +pub enum RtcSessionError { + /// Two-stage failure from str0m's SDP path: `SdpOffer::from_sdp_string` + /// can fail to parse, OR `accept_offer` can reject the parsed offer. + /// Both return `str0m::sdp::SdpError` / `str0m::RtcError` respectively; + /// we collapse them via `#[source]` since both are display-format-only + /// at the axum boundary (HTTP 400 in `routes.rs`). + #[error("SDP offer parse or accept failed: {0}")] + SdpOffer(String), + #[error("opus codec init failed: {0}")] + Codec(#[from] opus::Error), + #[error("UDP socket bind failed: {0}")] + Socket(#[from] std::io::Error), +} + +#[cfg(test)] +mod tests { + use super::*; + + /// A captured Chrome SDP offer for an audio-only Opus m-line. Truncated + /// to the relevant audio m-line section for test readability — the full + /// offer includes video m-lines that str0m rejects as part of answer + /// generation (spec §3.7). This fixture is a real browser-style offer + /// with host ICE candidates. + const BROWSER_SDP_OFFER: &str = "\ +v=0\r +o=- 4593482934 2 IN IP4 127.0.0.1\r +s=-\r +t=0 0\r +m=audio 9 UDP/TLS/RTP/SAVPF 111\r +c=IN IP4 0.0.0.0\r +a=rtcp:9 IN IP4 0.0.0.0\r +a=ice-ufrag:abcd\r +a=ice-pwd:abcdefghijklmnopqrstuvwxyz0123456789\r +a=fingerprint:sha-256 AB:CD:EF:00:11:22:33:44:55:66:77:88:99:AA:BB:CC:DD:EE:FF:00:11:22:33:44:55:66:77:88:99:AA:BB:CC:DD\r +a=setup:actpass\r +a=mid:0\r +a=sendrecv\r +a=rtpmap:111 opus/48000/2\r +a=fmtp:111 minptime=10;useinbandfec=1\r +a=candidate:1 1 UDP 2113667327 192.168.1.42 50000 typ host\r +"; + + #[test] + fn accept_offer_returns_sdp_answer_with_opus() { + let mut session = RtcSession::new().expect("session"); + let answer = session + .accept_offer(BROWSER_SDP_OFFER) + .expect("SDP answer"); + // Answer contains an audio m-line, an Opus payload, a fingerprint, + // and ICE credentials (str0m fills these natively in 0.21). + assert!(answer.contains("m=audio"), "answer has an audio m-line"); + assert!(answer.contains("opus/48000"), "answer advertises Opus"); + assert!(answer.contains("a=fingerprint:sha-256 "), "DTLS fingerprint"); + assert!(answer.contains("a=ice-ufrag:"), "ICE ufrag present"); + assert!(answer.contains("a=ice-pwd:"), "ICE pwd present"); + } + + #[test] + fn channel_id_matches_session_id() { + let session = RtcSession::new().expect("session"); + let id = session.channel_id(); + // The ChannelId IS the session id surfaced in the REST API (spec §4.5). + assert_eq!(format!("{}", id).len(), 36); + } + + #[test] + fn accept_offer_transitions_channel_to_connecting() { + // The spec §5.4 state machine: New → Connecting on offer receive. + // This test pins the transition callers depend on; the impl sets + // it at the end of `accept_offer`. + let mut session = RtcSession::new().expect("session"); + assert_eq!(session.channel_state(), ChannelState::New); + let _ = session.accept_offer(BROWSER_SDP_OFFER).expect("answer"); + assert_eq!(session.channel_state(), ChannelState::Connecting); + } +} +``` + +- [ ] **Step 4: Run the test to verify it fails** + +Run: `cargo test -p rutster-media rtc_session::tests` +Expected: FAIL — `RtcSession` undefined. + +- [ ] **Step 5: Implement `RtcSession`** + +Append to `crates/rutster-media/src/rtc_session.rs` (above the `#[cfg(test)] mod tests`): + +```rust +use str0m::Candidate; +use str0m::media::Mid; +use str0m::net::Protocol; + +/// The per-peer media owner (spec §3.1, §4.5). +/// +/// # Ownership / sharing +/// +/// An `RtcSession` lives behind an `Arc>` in the +/// binary's `DashMap` (Task 5). The mutex is +/// short-held: each tokio poll iteration locks, runs `run_poll_once`, +/// unlocks. We do NOT hold the lock across `tokio::time::sleep` — that +/// would defeat theDashMap's sharded concurrency and pre-pave the +/// wrong pattern for step 4's dedicated thread. +/// +/// # Why `Arc>` (not `Arc>`) +/// +/// Every access of an `RtcSession` mutates it (str0m's `&mut self` +/// contract on `handle_input` + `poll_output`). `RwLock`'s read-mode +/// would be useless because str0m takes `&mut Rtc`. `Mutex` it is. +pub struct RtcSession { + pub(crate) channel: Channel, + pub(crate) rtc: Rtc, + pub(crate) decoder: OpusDecoder, + pub(crate) encoder: OpusEncoder, + pub(crate) pipe: EchoAudioPipe, + /// Local UDP socket str0m sends `Transmit` packets out on and + /// receives `Input::Receive` packets from. Bound to an ephemeral + /// port at construction; the local candidate passed to str0m at + /// offer-accept time uses this address. + pub(crate) socket: std::net::UdpSocket, + /// Local socket address — cached because `local_addr()` is a syscall. + pub(crate) local_addr: SocketAddr, + /// Mid of the audio m-line we registered. Set during `accept_offer`. + /// Slice 1 has exactly one m-line; multi-m-line arrives with video. + pub(crate) audio_mid: Option, + /// Last deadline from `Rtc::poll_output` — the next time the loop + /// should wake the rtc with `Input::Timeout`. + pub(crate) next_timeout: Option, + /// Last Instant we received an RTP packet from the peer. Used for + /// the 60 s idle timeout (spec §4.5). + pub(crate) last_rx: Instant, + /// Last Instant we wrote an outbound Opus frame. Used to pace the + /// 20 ms encode tick for the echo path (slice-1 read of spec §3.2). + pub(crate) last_outbound_at: Instant, + /// Outbound RTP media-time clock (Opus audio runs at 48 kHz on the + /// wire — 960 ticks per 20 ms frame). Incremented by 960 on each + /// successful write. Honors str0m's "media time, wallclock, local + /// time" discipline from its docs. + pub(crate) next_media_time: str0m::media::MediaTime, +} + +impl RtcSession { + /// Construct a new session — used by both the binary's `AppState` + /// (production) and the tests. Single constructor — no `for_test` / + /// `for_server` split; the body is identical (binding a UDP socket + /// on `0.0.0.0:0`, constructing the `Rtc` + codecs). + pub fn new() -> Result { + Self::new_internal() + } + + fn new_internal() -> Result { + // Bind an ephemeral UDP socket. We use std::net::UdpSocket and + // drive it non-blocking from tokio rather than tokio's UdpSocket: + // str0m operates on raw `Receive` values and yields `Transmit` + // values, both of which are plain structs — no async needed. + // Setting non-blocking lets us `recv_from` without blocking. + let socket = std::net::UdpSocket::bind("0.0.0.0:0")?; + socket.set_nonblocking(true)?; + let local_addr = socket.local_addr()?; + + let rtc = Rtc::new(Instant::now()); + + Ok(Self { + channel: Channel::new_inbound(), + rtc, + decoder: OpusDecoder::new()?, + encoder: OpusEncoder::new()?, + pipe: EchoAudioPipe::new(), + socket, + local_addr, + audio_mid: None, + next_timeout: None, + last_rx: Instant::now(), + last_outbound_at: Instant::now(), + next_media_time: str0m::media::MediaTime::ZERO, + }) + } + + pub fn channel_id(&self) -> ChannelId { + self.channel.id + } + + pub fn channel_state(&self) -> ChannelState { + self.channel.state + } + + pub fn is_closed(&self) -> bool { + matches!(self.channel.state, ChannelState::Closed) + } + + /// Accept a browser SDP offer; return the SDP answer (spec §4.1). + /// + /// str0m 0.21's `sdp_api().accept_offer()` does the heavy lifting: + /// parses the offer, picks compatible codecs (Opus, the only one we + /// register by default), generates the DTLS fingerprint from its + /// self-signed cert, and produces ICE ufrag/pwd. We add our local + /// host candidate (the UDP socket we just bound) *before* calling + /// `accept_offer` so the answer carries it. + pub fn accept_offer(&mut self, offer_sdp: &str) -> Result { + assert!(self.audio_mid.is_none(), "accept_offer called twice"); + + // Register our local UDP socket as a host candidate. str0m includes + // this candidate's address + the ICE creds it generates in the SDP + // answer. `add_local_candidate` returns `Option<&Candidate>` — + // `None` means str0m rejected it (log + continue; not fatal). + let candidate = Candidate::host(self.local_addr, "udp") + .expect("host candidate from bound UDP socket"); + // ^-- expect is acceptable here: this is construction (cold path), + // not the hot path. A bound UDP socket always yields a valid + // host candidate; only an absurd Protocol parse fails. + if self.rtc.add_local_candidate(candidate).is_none() { + tracing::warn!(channel_id = %self.channel.id, "str0m rejected local candidate"); + } + + // str0m's SDP API parses + accepts the offer natively. There is NO + // `from_str_unchecked` — `from_sdp_string` returns Result and is + // the canonical entry point. accept_offer takes the owned SdpOffer. + let parsed_offer = str0m::change::SdpOffer::from_sdp_string(offer_sdp) + .map_err(|e| RtcSessionError::SdpOffer(format!("parse: {e}")))?; + let answer = self + .rtc + .sdp_api() + .accept_offer(parsed_offer) + .map_err(|e| RtcSessionError::SdpOffer(format!("accept: {e}")))?; + + // The first audio mid we accepted. Used to get the Writer for + // outbound Opus frames in `run_poll_once`. A single audio m-line + // is slice 1's whole world; multi-m-line arrives with video. + // + // SdpAnswer exposes a `mid()` accessor — verify against str0m 0.21 + // `SdpAnswer` docs at impl time; if the accessor differs, look up + // from the answer's m-lines. + self.audio_mid = Some(answer.mid()); + + self.channel.state = ChannelState::Connecting; + Ok(answer.to_string()) + } + + /// Drive one iteration of the sans-IO poll loop (spec §3.2, §3.4). + /// + /// Returns the `Duration` until the next `Input::Timeout` should be + /// fed back to str0m, or `None` if the peer is closed. The caller + /// (Task 5's tokio task) sleeps this duration then calls again. + /// + /// DEV-DEVIATION: tokio polling accepted for slice 1; step 4 + /// replaces with dedicated timing thread per ARCHITECTURE.md. + pub fn run_poll_once(&mut self, now: Instant) -> Option { + if self.is_closed() { + return None; + } + crate::loop_driver::drive(self, now) + } +} +``` + +- [ ] **Step 6: Write `crates/rutster-media/src/loop_driver.rs` (the str0m poll loop)** + +```rust +//! # str0m poll loop (spec §3.2, §3.4) +//! +//! The heart of the media core. Drives the `str0m::Rtc` instance forward +//! on each call: drains `poll_output()` until `Output::Timeout`, handling +//! each `Output::Transmit` (send on our UDP socket) and `Output::Event` +//! (inbound `MediaData` → Opus decode → sink; inbound RTP count for the +//! idle timeout). When the drain returns `Timeout`, the caller sleeps +//! that duration and calls back with `Input::Timeout`. +//! +//! # Why this lives in a separate module +//! +//! `run_poll_once` takes `&mut RtcSession` — a single function with +//! the full poll logic would make `RtcSession::run_poll_once` 100+ lines +//! of non-trivial control flow. Splitting the loop into a module makes +//! the sans-IO pattern obvious: the loop driver takes a `&mut RtcSession`, +//! reads str0m outputs, and writes str0m inputs. Nothing else. +//! +//! # DEV-DEVIATION +//! +//! Slice 1 runs the poll on a tokio task. ARCHITECTURE.md mandates a +//! dedicated timing thread; we defer that to step 4 (barge-in) because +//! slice 1 has no reflex to time against. The poll function's shape +//! (single `&mut self`, no I/O inside) makes the step-4 swap localized. + +use std::io::ErrorKind; +use std::time::{Duration, Instant}; + +use str0m::media::MediaData; +use str0m::net::Receive; +use str0m::{Input, Output, Protocol}; + +use crate::pcm::{AudioSink as _, AudioSource as _}; +use crate::rtc_session::RtcSession; +use crate::IDLE_TIMEOUT; + +/// 20 ms tick for outbound encoding (matches the PCM frame size). +const OUTBOUND_TICK: Duration = Duration::from_millis(20); + +/// One iteration of the str0m poll loop. +/// +/// 1. Read any pending UDP packets (non-blocking) and feed each to str0m +/// as `Input::Receive`. A WouldBlock means no packets this cycle — fine. +/// 2. Drain `poll_output()` until `Timeout`: +/// - `Transmit` → send on our UDP socket. +/// - `Event::MediaData` → decode Opus → push to the echo pipe (sink). +/// - `Event::IceConnectionStateChange` → state transition + tracing. +/// - We don't break out of the drain on any of these: str0m's contract +/// is mutate→drain to `Timeout`→mutate (see str0m 0.21 docs). +/// 3. **Outbound encode tick:** if ≥20 ms of wallclock passed since the +/// last outbound frame, pull one `PcmFrame` from the source, encode to +/// Opus, and write via `Rtc::writer(mid)->Writer::write`. Then re-drain +/// `poll_output` (the Writer write is a mutation → must drain per str0m). +/// 4. Check the idle timeout: if `Instant::now() - last_rx > IDLE_TIMEOUT`, +/// transition to `Closed`. +/// 5. Return the `Duration` to the next `Timeout`. +pub fn drive(session: &mut RtcSession, now: Instant) -> Option { + // === Step 1: drain our UDP socket non-blocking, feed str0m. === + let mut buf = [0u8; 2000]; + loop { + match session.socket.recv_from(&mut buf) { + Ok((n, source)) => { + let contents = &buf[..n]; + let recv = Receive { + proto: Protocol::Udp, + source, + destination: session.local_addr, + contents: contents.try_into().ok()?, + }; + if session.rtc.handle_input(now, Input::Receive(recv)).is_err() { + // Hot-path policy: drop + observe, don't crash. + tracing::warn!("str0m rejected input packet; dropping"); + } + session.last_rx = now; + } + // WouldBlock (unix) / TimedOut (windows) — no packets this cycle. + Err(e) if matches!(e.kind(), ErrorKind::WouldBlock | ErrorKind::TimedOut) => break, + Err(e) => { + tracing::warn!(error = ?e, "UDP recv_from error; continuing"); + break; + } + } + } + + // === Step 2: drain poll_output, interleaving outbound writes. === + let mut next_timeout: Option = session.next_timeout; + // Track whether we owe a Writer write this cycle; re-drain if so. + // str0m's "mutate → drain to Timeout" invariant: after Writer::write, + // poll_output must be drained to Timeout before any other mutation. + let mut needs_redrain = false; + loop { + match session.rtc.poll_output() { + Ok(Output::Timeout(t)) => { + next_timeout = Some(t); + if needs_redrain { + // We did an outbound write in the previous iteration; + // str0m needs to be drained again. Loop continues, + // but only handle Transmit/Event briefly before next Timeout. + needs_redrain = false; + continue; + } + break; // engine is fully drained + } + Ok(Output::Transmit(t)) => { + if let Err(e) = session.socket.send_to(&t.contents, t.destination) { + if !matches!(e.kind(), ErrorKind::WouldBlock) { + tracing::warn!(error = ?e, "UDP send_to error; dropping"); + } + } + } + Ok(Output::Event(event)) => { + handle_event(session, event, now); + // Loop continues — mutations from inside the drain loop + // are fine (str0m docs, "single-mutation invariant"). + } + Err(e) => { + tracing::warn!(error = ?e, "str0m poll_output error; continuing"); + next_timeout = Some(now + OUTBOUND_TICK); + break; + } + } + } + + // === Step 3: outbound encode tick (the echo path). === + // If str0m's poll loop has crossed a 20 ms boundary, pull a PcmFrame + // from the source, encode to Opus, and write via Writer::write. This + // IS the slice-1 echo: inbound decode → pipe → outbound encode. + if now.duration_since(session.last_outbound_at) >= OUTBOUND_TICK { + if let Some(mid) = session.audio_mid { + if let Some(frame) = session.pipe.next_pcm_frame() { + if let Some(opus_payload) = session.encoder.encode(&frame) { + // Writer::write signature (str0m 0.21, verified): + // write(pt: Pt, wallclock: Instant, rtp_time: MediaTime, data: impl Into>) + // -> Result<(), RtcError> + // - pt: payload type for Opus. `writer.payload_params()` + // returns `impl Iterator`; the + // first one's `.pt()` is our Opus PT (str0m negotiates + // this in the SDP answer). + // - wallclock: when the sample was produced — local `now`. + // - rtp_time: RTP timestamp in the 48 kHz audio clock for + // Opus. Increment by 960 per 20 ms (48000 * 0.020). + // NOTE: the param is named `rtp_time` in str0m's + // signature (NOT `media_time`). MediaTime has NO + // `add(Duration)` method — use `mt + MediaTime::from(d)`. + // + // `rtc.writer(mid)` returns `Option>` — `None` + // if direction isn't sending (we'd be in a recvonly state). + if let Some(writer) = session.rtc.writer(mid) { + if let Some(params) = writer.payload_params().next() { + let pt = params.pt(); + let rtp_time = session.next_media_time; + if writer + .write(pt, now, rtp_time, opus_payload.as_slice()) + .is_ok() + { + // Advance media time for next 20 ms frame. + // `MediaTime + MediaTime::from(Duration)` — + // no `add()` method on MediaTime. + session.next_media_time = + session.next_media_time + + str0m::media::MediaTime::from( + Duration::from_millis(20), + ); + needs_redrain = true; + } + } + } + } + } + session.last_outbound_at = now; + } + } + + // If the outbound write happened, we owe str0m one more drain before + // returning — Writer::write is a mutation per str0m's invariant. + if needs_redrain { + loop { + match session.rtc.poll_output() { + Ok(Output::Timeout(t)) => { + next_timeout = Some(t); + break; + } + Ok(Output::Transmit(t)) => { + let _ = session.socket.send_to(&t.contents, t.destination); + } + Ok(Output::Event(e)) => handle_event(session, e, now), + Err(_) => break, + } + } + } + + // === Step 4: idle timeout (spec §4.5). === + if now.duration_since(session.last_rx) > IDLE_TIMEOUT { + tracing::info!( + channel_id = %session.channel.id, + "idle timeout (60 s no RX); closing session" + ); + session.channel.state = rutster_call_model::ChannelState::Closed; + return None; + } + + session.next_timeout = next_timeout; + next_timeout.map(|t| t.saturating_duration_since(now)) +} + +/// Dispatch a str0m `Event` to the audio pipe or to state bookkeeping. +fn handle_event(session: &mut RtcSession, event: str0m::Event, _now: Instant) { + use str0m::Event; + match event { + Event::MediaData(media) => { + // Inbound decoded audio frame from the peer (Frame API, spec §3.2). + // str0m has already done RTP depacketization; `MediaData.data` is + // the encoded Opus payload (type: `Arc<[u8]>` — pass `&media.data` + // to the decoder since OpusDecoder::decode takes `&[u8]`). + if let Some(pcm) = session.decoder.decode(&media.data) { + session.pipe.on_pcm_frame(pcm); + } + // Decode failed → drop + observe (per §3.8). Don't kill the peer. + } + Event::IceConnectionStateChange(state) => { + tracing::info!( + channel_id = %session.channel.id, + ?state, + "ICE state change" + ); + if state == ::str0m::IceConnectionState::Connected { + session.channel.state = rutster_call_model::ChannelState::Connected; + } + } + Event::EgressBitrateEstimate(_) => { /* BWE — irrelevant in slice 1 */ } + _ => { /* str0m emits several other event variants we don't need in slice 1. */ } + } +} +``` + +- [ ] **Step 7: Run the str0m-offer test to verify the API wiring** + +Run: `cargo test -p rutster-media rtc_session::tests::accept_offer_returns_sdp_answer_with_opus` +Expected: PASS (str0m accepts the offer, returns an SDP answer with Opus + DTLS fingerprint + ICE creds). + +NOTES FOR THE IMPLEMENTER (residual verifications post-review): +- `SdpAnswer::mid()` — the plan assumes this accessor exists on str0m's `SdpAnswer`. If str0m 0.21 exposes a different API (e.g. `answer.mids().next()` or via a `CodecConfig` lookup), adjust to use whatever str0m 0.21 ships. Run `cargo doc -p str0m --open` and look at `SdpAnswer`. +- `PayloadParams::pt()` — the plan assumes `payload_params().next().unwrap().pt()` works. If `Pt` is exposed differently (e.g. via `match_params(incoming)`), use that instead. The recommended path per str0m 0.21 docs is `writer.match_params(¶ms) -> Option` where `params` is the inbound `MediaData.params` — this matches the incoming payload to the negotiated outbound PT. For slice 1 (echo loopback, single codec), the simpler `payload_params().next()` path works; `match_params` is the general path when multiple codecs are negotiated. +- `media.data: Arc<[u8]>` — pass `&media.data` (deref coercion) to `OpusDecoder::decode(&[u8])`. +- The plan's str0m API claims were verified against `docs.rs/str0m/0.21` during the adversarial review. Don't hand-roll an SDP munger; honor the mutate → drain to `Timeout` → mutate invariant; keep the hot-path match-and-continue policy on the 20 ms loop. + +- [ ] **Step 8: Fix clippy + fmt** + +Run: `cargo fmt && cargo clippy -p rutster-media -- -D warnings` +Expected: clean. If str0m's API types don't line up with the plan's sketches, fix the import paths to satisfy the compiler; do NOT add an SDP munger or change the loop structure. + +- [ ] **Step 9: Commit** + +```bash +git add crates/rutster-media +git commit -m "media: RtcSession + str0m poll loop (the media core) + +RtcSession owns a str0m::Rtc + Opus decoder/encoder + EchoAudioPipe + +a bound UDP socket (spec §3.1, §4.5). accept_offer calls str0m 0.21's +sdp_api().accept_offer() natively — no hand-rolled SDP munger; str0m +fills DTLS fingerprint + ICE creds + Opus codec. The loop driver +drains poll_output per str0m's single-mutation invariant, routes +inbound MediaData through Opus decode + EchoAudioPipe sink, sends +Transmit packets on the UDP socket, and checks the 60 s idle timeout. + +DEV-DEVIATION: loop runs on tokio (spec §3.4); step 4 replaces with +a dedicated timing thread per ARCHITECTURE.md." +``` + +--- + +## Task 5: `rutster` binary — axum signaling server + browser client + integration test + +**Files:** +- Create: `crates/rutster/Cargo.toml` +- Create: `crates/rutster/src/main.rs` (axum server bootstrap + graceful shutdown) +- Create: `crates/rutster/src/session_map.rs` (`DashMap` + poll driver) +- Create: `crates/rutster/src/routes.rs` (the four HTTP routes) +- Create: `crates/rutster/static/index.html` (browser test client) +- Create: `crates/rutster/tests/api_integration.rs` (integration test: POST `/v1/sessions` roundtrip) +- Modify: `Cargo.toml` (workspace root — add member). + +**Interfaces:** +- Consumes: `RtcSession`, `RtcSessionError`, `ChannelId` from Tasks 2/4. +- Produces: a running axum server on `0.0.0.0:8080` with four routes (spec §4.1), a tokio task per active session driving the str0m poll loop, a static browser test client served at `GET /`, an integration test that hits the API. + +- [ ] **Step 1: Write `crates/rutster/Cargo.toml`** + +```toml +# crates/rutster/Cargo.toml +[package] +name = "rutster" +version = "0.0.0" +license.workspace = true +edition.workspace = true +repository.workspace = true +description = "Rutster binary: axum signaling + media driver + static browser test client (slice 1)." + +[dependencies] +rutster-call-model = { path = "../rutster-call-model" } +rutster-media = { path = "../rutster-media" } +axum = { workspace = true } +tokio = { workspace = true } +dashmap = { workspace = true } +uuid = { workspace = true } +thiserror = { workspace = true } +tracing = { workspace = true } +tracing-subscriber = { workspace = true } +serde = { workspace = true } +serde_json = { workspace = true } + +[dev-dependencies] +tower = { workspace = true } + +[[bin]] +name = "rutster" +path = "src/main.rs" +``` + +- [ ] **Step 2: Write the failing integration test first** + +Create `crates/rutster/tests/api_integration.rs`: + +```rust +//! Integration test for the slice-1 REST surface (spec §4.1, §6.4). +//! +//! Spins up the axum app on an ephemeral port, then exercises the API +//! contract: POST /v1/sessions → JSON { session_id }; GET / serves a +//! text/html page. We do NOT exercise the WebRTC handshake here (that +//! needs a real peer); the manual e2e plan in README.md covers it. + +use axum::body::Body; +use axum::http::{Request, StatusCode}; +use rutster::session_map::AppState; +use tower::ServiceExt; // enables `oneshot` on the Router for sync tests + +#[tokio::test] +async fn post_v1_sessions_returns_a_session_id() { + let app = rutster::routes::router(AppState::new()); + let resp = app + .oneshot( + Request::builder() + .method("POST") + .uri("/v1/sessions") + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + + assert_eq!(resp.status(), StatusCode::OK); + let body = axum::body::to_bytes(resp.into_body(), 1024).await.unwrap(); + let v: serde_json::Value = serde_json::from_slice(&body).unwrap(); + assert!(v["session_id"].is_string(), "response has session_id"); + assert_eq!(v["session_id"].as_str().unwrap().len(), 36); // UUID v4 +} + +#[tokio::test] +async fn get_root_serves_html() { + let app = rutster::routes::router(AppState::new()); + let resp = app + .oneshot(Request::builder().uri("/").body(Body::empty()).unwrap()) + .await + .unwrap(); + + assert_eq!(resp.status(), StatusCode::OK); + assert_eq!( + resp.headers() + .get("content-type") + .map(|v| v.to_str().unwrap()), + Some("text/html; charset=utf-8") + ); +} +``` + +- [ ] **Step 3: Run the test to verify it fails** + +Run: `cargo test -p rutster --test api_integration` +Expected: FAIL — `rutster::routes` and `rutster::session_map` don't exist. + +- [ ] **Step 4: Write `crates/rutster/src/session_map.rs`** + +```rust +//! # Session store + poll-driver (spec §4.5) +//! +//! `DashMap` holds active sessions; the `ChannelId` +//! (UUID newtype from `rutster-call-model`) IS the session id surfaced in +//! the REST API. A single tokio task drives all sessions' poll loops (a +//! per-session task would clutter the runtime and pre-pave the wrong +//! pattern for the step-4 dedicated thread — spec §4.5). +//! +//! # Concurrency note +//! +//! `DashMap` shards its inner `HashMap` so concurrent gets/puts across +//! different `ChannelId`s don't contend. We iterate per-shard inside the +//! poll task to drive each session; entries marked `Closed` are removed. + +use std::sync::Arc; +use std::time::{Duration, Instant}; + +use dashmap::DashMap; +use rutster_call_model::{ChannelId, ChannelState}; +use rutster_media::{RtcSession, RtcSessionError}; +use tokio::sync::Mutex; +use tracing::{debug, info}; + +/// The application state shared across axum handlers + the poll task. +/// +/// # Why `Arc` (and not bare) +/// +/// axum clones the state into every handler. `Arc` is the standard way +/// to share `DashMap` + `Mutex` owned state across these clones cheaply +/// (a single heap allocation, refcount-bumped per clone). Without `Arc`, +/// every handler would move its own copy — and `DashMap` is not `Copy`. +/// +/// # Why a separate `poll_running` `Mutex` +/// +/// The poll loop is one task; we don't want two. The Mutex guards a +/// once-only spawn: `spawn_poll_task` checks-and-sets it under the mutex. +/// `Mutex` (not `RwLock`) because the only operation is "take it once." +#[derive(Clone)] +pub struct AppState { + pub sessions: Arc>>>, + pub poll_running: Arc>, +} + +impl AppState { + pub fn new() -> Self { + Self { + sessions: Arc::new(DashMap::new()), + poll_running: Arc::new(Mutex::new(false)), + } + } + + /// Mint a fresh `RtcSession`, store it under its `ChannelId`, return the id. + pub fn create_session(&self) -> Result { + let session = RtcSession::new()?; + let id = session.channel_id(); + self.sessions.insert(id, Arc::new(Mutex::new(session))); + Ok(id) + } + + /// Look up a session by id (returns the clone of the Arc-wrapped Mutex). + pub fn get(&self, id: ChannelId) -> Option>> { + self.sessions.get(&id).map(|r| r.clone()) + } + + /// Transition to Closing then drop the entry (spec §4.1 — DELETE). + pub async fn close(&self, id: ChannelId) { + if let Some((_id, session_arc)) = self.sessions.remove(&id) { + let mut s = session_arc.lock().await; + s.channel.state = ChannelState::Closing; + s.channel.state = ChannelState::Closed; + info!(channel_id = %id, "session closed via DELETE"); + } + } + + /// Spawn the single poll task for all sessions (idempotent). + pub async fn spawn_poll_task(self) { + let mut running = self.poll_running.lock().await; + if *running { + return; + } + *running = true; + drop(running); + + let state = self.clone(); + tokio::spawn(async move { + let mut interval = tokio::time::interval(Duration::from_millis(10)); + interval.tick().await; + loop { + interval.tick().await; + let now = Instant::now(); + drive_all_sessions(&state, now).await; + } + }); + } +} + +impl Default for AppState { + fn default() -> Self { + Self::new() + } +} + +/// One iteration of "drive every active session." Removes closed entries. +async fn drive_all_sessions(state: &AppState, now: Instant) { + // Collect ids first to avoid holding the DashMap shard during the + // async poll (which would block other handlers mutating the same shard). + let ids: Vec = state.sessions.iter().map(|r| *r.key()).collect(); + for id in ids { + let session_arc = match state.sessions.get(&id) { + Some(r) => r.clone(), + None => continue, + }; + let mut s = session_arc.lock().await; + let _ = s.run_poll_once(now); // hot-path match-and-continue inside + if s.is_closed() { + drop(s); + state.sessions.remove(&id); + debug!(channel_id = %id, "session evicted after close"); + } + } +} +``` + +- [ ] **Step 5: Write `crates/rutster/src/routes.rs`** + +```rust +//! # HTTP routes (spec §4.1, §4.3) +//! +//! Four routes on axum 0.7: +//! - `POST /v1/sessions` → `{ "session_id": "" }`. +//! - `POST /v1/sessions/:id/offer` (`Content-Type: application/sdp` req +//! + response) → core returns the SDP answer. +//! - `DELETE /v1/sessions/:id` → tear down. +//! - `GET /` → serve the static HTML test client. +//! +//! No authn/authz, no TLS, no multi-tenancy — all deferred per spec §1.2. + +use axum::extract::{Path, State}; +use axum::http::{header, StatusCode}; +use axum::response::{IntoResponse, Response}; +use axum::routing::{get, post}; +use axum::{Json, Router}; +use serde::Serialize; +use uuid::Uuid; + +use crate::session_map::AppState; + +#[derive(Serialize)] +struct SessionCreated { + session_id: String, +} + +/// POST /v1/sessions — mint a fresh RtcSession (spec §4.1). +pub async fn create_session(State(state): State) -> Response { + match state.create_session() { + Ok(id) => { + let body = Json(SessionCreated { + session_id: id.0.to_string(), + }); + (StatusCode::OK, body).into_response() + } + Err(e) => { + tracing::error!(error = ?e, "session create failed"); + StatusCode::INTERNAL_SERVER_ERROR.into_response() + } + } +} + +/// POST /v1/sessions/:id/offer — accept browser SDP offer, return answer +/// (spec §4.1). Non-trickle: the offer body carries all browser ICE +/// candidates; the answer carries the core's candidates (filled natively +/// by str0m 0.21's sdp_api().accept_offer). +pub async fn post_offer( + State(state): State, + Path(id_str): Path, + body: String, +) -> Response { + let Ok(id_uuid) = Uuid::parse_str(&id_str) else { + return (StatusCode::NOT_FOUND, "bad session id").into_response(); + }; + let id = rutster_call_model::ChannelId(id_uuid); + let Some(session_arc) = state.get(id) else { + return (StatusCode::NOT_FOUND, "no such session").into_response(); + }; + let mut s = session_arc.lock().await; + match s.accept_offer(&body) { + Ok(answer_sdp) => ( + StatusCode::OK, + [(header::CONTENT_TYPE, "application/sdp")], + answer_sdp, + ) + .into_response(), + Err(e) => { + tracing::error!(error = ?e, "SDP accept failed"); + StatusCode::BAD_REQUEST.into_response() + } + } +} + +/// DELETE /v1/sessions/:id — tear down (spec §4.1, §4.5). +pub async fn delete_session( + State(state): State, + Path(id_str): Path, +) -> Response { + let Ok(id_uuid) = Uuid::parse_str(&id_str) else { + return StatusCode::NOT_FOUND.into_response(); + }; + let id = rutster_call_model::ChannelId(id_uuid); + state.close(id).await; + StatusCode::NO_CONTENT.into_response() +} + +/// GET / — serve the static browser test client (spec §4.4). +pub async fn index() -> Response { + ( + StatusCode::OK, + [(header::CONTENT_TYPE, "text/html; charset=utf-8")], + include_str!("../static/index.html"), + ) + .into_response() +} + +/// Build the axum router. +pub fn router(state: AppState) -> Router { + Router::new() + .route("/", get(index)) + // `POST /v1/sessions` creates; `DELETE /v1/sessions/:id` destroys + // (note the `:id` — deleting the collection root has no meaning and + // would extract a missing `:id` path parameter, so the two routes + // live at different paths, not chained via `.delete(...)` on the + // collection route as axum's method chaining would suggest). + .route("/v1/sessions", post(create_session)) + .route("/v1/sessions/:id", axum::routing::delete(delete_session)) + .route("/v1/sessions/:id/offer", post(post_offer)) + .with_state(state) +} +``` + +- [ ] **Step 6: Write `crates/rutster/src/main.rs`** + +```rust +//! # rutster — slice-1 binary +//! +//! axum signaling server + the media-core poll driver + a static HTML +//! test client (spec §4). Binds `0.0.0.0:8080` plaintext — no TLS in +//! slice 1 (out of scope per §1.2). DTLS-SRTP is mandatory on the media +//! surface (str0m handles natively); TLS on the HTTP surface lands with +//! the deployment posture in step 5. +//! +//! ## Architecture refs +//! +//! - [slice-1 spec §4](../../docs/superpowers/specs/2026-06-28-slice-1-webrtc-loopback-design.md) +//! - [ARCHITECTURE.md](../../docs/ARCHITECTURE.md) — fused vertical. + +use std::net::SocketAddr; + +use rutster::routes::router; +use rutster::session_map::AppState; +use tracing::info; + +#[tokio::main] +async fn main() { + tracing_subscriber::fmt() + .with_env_filter( + tracing_subscriber::EnvFilter::try_from_default_env() + .unwrap_or_else(|_| "rutster=info".into()), + ) + .init(); + + let state = AppState::new(); + state.clone().spawn_poll_task().await; + + let addr: SocketAddr = "0.0.0.0:8080".parse().expect("valid addr"); + info!(%addr, "listening"); + let listener = tokio::net::TcpListener::bind(addr).await.unwrap(); + axum::serve(listener, router(state)) + .with_graceful_shutdown(shutdown_signal()) + .await + .unwrap(); +} + +/// Ctrl-C / SIGTERM handler (spec §4.5). Dropping the AppState drops the +/// DashMap, which drops every RtcSession, which str0m sees as a closed +/// peer — browsers get a dead peer connection. Acceptable for the dev +/// loop; no in-flight call preservation story in slice 1. +async fn shutdown_signal() { + let ctrl_c = async { + tokio::signal::ctrl_c() + .await + .expect("installed ctrl-c handler"); + }; + + #[cfg(unix)] + let sigterm = async { + tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate()) + .expect("installed SIGTERM handler") + .recv() + .await; + }; + #[cfg(not(unix))] + let sigterm = std::future::pending::<()>(); + + tokio::select! { + _ = ctrl_c => info!("received Ctrl-C, shutting down"), + _ = sigterm => info!("received SIGTERM, shutting down"), + } +} +``` + +In `crates/rutster/src/main.rs`, declare the two modules above `main`. They +must be `pub mod` (not plain `mod`) because the integration test in +`tests/api_integration.rs` references them via absolute paths +(`rutster::routes::router`, `rutster::session_map::AppState`). Binary crates +don't have an external consumer surface in production, but Rust still +requires `pub mod` for `tests/` integration tests to see the path: + +```rust +pub mod routes; +pub mod session_map; + +#[tokio::main] +async fn main() { + // (the body from above) +} +``` + +(Equivalent: write `pub mod` lines first, then `#[tokio::main] async fn main() { ... }` below them. The order in the file is: top-level docs → `pub mod` declarations → `use` imports → `main`.) + +- [ ] **Step 7: Write `crates/rutster/static/index.html` (browser test client, spec §4.4)** + +```html + + + + + Rutster slice-1 — WebRTC loopback + + + +

Rutster slice-1 — WebRTC loopback

+

Speak; you should hear yourself back within ~200 ms.

+ + + + 

+  
+
+  
+
+
+```
+
+- [ ] **Step 8: Add `crates/rutster` to the workspace `members`**
+
+Modify root `Cargo.toml`:
+
+```toml
+members = [
+    "crates/rutster",
+    "crates/rutster-call-model",
+    "crates/rutster-media",
+    "crates/rutster-signaling-sip",
+    "crates/rutster-tap",
+    "crates/rutster-spend",
+]
+```
+
+- [ ] **Step 9: Run the integration tests**
+
+Run: `cargo test -p rutster --test api_integration`
+Expected: 2 tests passing (`post_v1_sessions_returns_a_session_id`, `get_root_serves_html`).
+
+NOTE FOR THE IMPLEMENTER: `session_map.rs` calls `RtcSession::new()` (the
+single constructor from Task 4 Step 5). No `new_for_test`/`new_for_server`
+split — Task 4 was patched in review to expose just `pub fn new()`. If you
+encountered an older plan revision mentioning two constructors, disregard
+it; the canonical name is `RtcSession::new()`.
+
+- [ ] **Step 10: Run clippy + fmt**
+
+Run: `cargo fmt --check && cargo clippy -p rutster -- -D warnings`
+Expected: clean.
+
+- [ ] **Step 11: Manual smoke (spec §6.5)**
+
+```bash
+cargo run -p rutster
+# in another terminal / browser:
+# open http://localhost:8080/  → click Start call → grant mic → hear echo
+# click Hang up → server logs Closing → Closed
+```
+
+This is the slice-1 "done" criterion #5. Don't gate the commit on the manual test (it's browser-driven) — flag it in the commit message as "manual e2e pending."
+
+- [ ] **Step 12: Commit**
+
+```bash
+git add Cargo.toml crates/rutster
+git commit -m "binary: axum signaling + DashMap session store + browser test client
+
+Four routes on axum 0.7 per spec §4.1: POST /v1/sessions (mint),
+POST /v1/sessions/:id/offer (str0m-native SDP accept), DELETE
+/v1/sessions/:id (close), GET / (static HTML client). Session store is
+a DashMap>> (spec §4.5); one tokio
+task drives all session poll loops — per-session tasks would pre-pave
+the wrong pattern for step 4's dedicated thread. Graceful shutdown
+drops the DashMap on Ctrl-C / SIGTERM. Integration test exercises the
+REST surface; manual browser e2e per README §6.5."
+```
+
+---
+
+## Task 6: cargo-deny config + CI workflow
+
+**Files:**
+- Create: `deny.toml`
+- Create: `.github/workflows/ci.yml`
+
+**Interfaces:**
+- Consumes: nothing code-wise.
+- Produces: `cargo deny check` gating CI; CI runs `fmt --check`, `clippy -D warnings`, `test --all`, `deny check` on push + PR to `main` (spec §6.1, §6.2).
+
+- [ ] **Step 1: Write `deny.toml`**
+
+```toml
+# deny.toml — cargo-deny config (spec §6.1).
+# Run locally: cargo deny check.
+# CI runs `cargo deny check` as the last gate.
+
+[graph]
+# Use Cargo.lock as the source of truth for the dep graph.
+all-features = true
+
+[advisories]
+# Vulnerabilities fail CI. `deny warnings` makes advisory-db issues
+# (not just actual advisories) fatal.
+deny = ["RUSTSEC-0000-0000"]
+version = 2
+ignore = []
+unmaintained = "workspace"
+
+[licenses]
+# Allow our own (GPL-3.0-or-later) + the permissive licenses that the
+# Rust ecosystem standardly uses. Final list confirmed at impl time by
+# running `cargo deny check licenses` after the first `cargo fetch`
+# (spec §6.1); adjust as needed so str0m/opus/axum actually pass.
+allow = [
+    "GPL-3.0-or-later",
+    "MIT",
+    "Apache-2.0",
+    "BSD-3-Clause",
+    "ISC",
+    "Zlib",
+    "Unicode-DFS-2016",
+    "Unicode-3.0",
+]
+confidence-threshold = 0.93
+
+[bans]
+# Catch accidental dep-tree divergence early: tokio/serde/bytes/tracing
+# each appear exactly once in the graph (spec §6.1).
+multiple-versions = "deny"
+wildcards = "deny"
+highlight = "all"
+deny = []
+allow = []
+# Skip-list for known unavoidable duplicates (added as they surface in CI).
+# `cargo deny check bans` prints the spec's "skip" suggestion when a dup
+# shows up; copy it here.
+skip = []
+skip-tree = []
+
+[sources]
+# crates.io only. No git deps. Keeps the build reproducible (spec §6.1,
+# PORT_PLAN supply-chain goal).
+unknown-registry = "deny"
+unknown-git = "deny"
+allow-registry = ["https://github.com/rust-lang/crates.io-index"]
+allow-git = []
+```
+
+- [ ] **Step 2: Write `.github/workflows/ci.yml`**
+
+```yaml
+# .github/workflows/ci.yml — slice-1 CI (spec §6.2).
+# Gates: fmt --check, clippy -D warnings, test --all, cargo deny check.
+name: CI
+
+on:
+  push:
+    branches: [main]
+  pull_request:
+    branches: [main]
+
+env:
+  CARGO_TERM_COLOR: always
+
+jobs:
+  fmt:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+      - uses: dtolnay/rust-toolchain@stable
+        with:
+          components: rustfmt
+      - run: cargo fmt --check
+
+  clippy:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+      - uses: dtolnay/rust-toolchain@stable
+        with:
+          components: clippy
+      - name: Install libopus (media crate FFI dep)
+        run: sudo apt-get update && sudo apt-get install -y libopus-dev
+      - uses: Swatinem/rust-cache@v2
+      - run: cargo clippy --all -- -D warnings
+
+  test:
+    runs-on: ubuntu-latest
+    strategy:
+      matrix:
+        toolchain: [stable, "1.80"]
+    steps:
+      - uses: actions/checkout@v4
+      - uses: dtolnay/rust-toolchain@master
+        with:
+          toolchain: ${{ matrix.toolchain }}
+      - name: Install libopus (media crate FFI dep)
+        run: sudo apt-get update && sudo apt-get install -y libopus-dev
+      - uses: Swatinem/rust-cache@v2
+      - run: cargo test --all
+
+  deny:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+      - uses: dtolnay/rust-toolchain@stable
+      - uses: EmbarkStudios/cargo-deny-action@v1
+        with:
+          command: check
+```
+
+- [ ] **Step 3: Install `cargo-deny` locally and verify**
+
+```bash
+cargo install cargo-deny --locked
+cargo deny check
+```
+
+Expected: all four checks (advisories, licenses, bans, sources) clean. If a non-allowed license shows up in the transitive graph, add it to `allow` in `deny.toml`. If unavoidable duplicate versions show up, add them to `skip` with a comment explaining why (don't blindly silence — investigate the duplicate).
+
+- [ ] **Step 4: Install libopus locally (the FFI dependency)**
+
+```bash
+sudo apt-get install -y libopus-dev   # Debian/Ubuntu
+# Fedora: sudo dnf install -y opus-devel
+# macOS:  brew install opus
+```
+
+Verify: `cargo test --all` passes locally.
+
+- [ ] **Step 5: Commit**
+
+```bash
+git add deny.toml .github/workflows/ci.yml
+git commit -m "ci: cargo-deny + GitHub Actions workflow (spec §6.1, §6.2)
+
+deny.toml allows the permissive Rust-ecosystem licenses + our own
+GPL-3.0-or-later; bans duplicate versions of tokio/serde/bytes/tracing
+to catch dep-tree divergence early; restricts sources to crates.io. CI
+runs fmt --check, clippy -D warnings, test --all (matrix: stable +
+MSRV 1.80), and cargo deny check on push + PR to main. The CI job
+installs libopus-dev — the opus crate's FFI dependency (PORT_PLAN §7
+'Core (FFI)' disposition)."
+```
+
+---
+
+## Task 7: LEARNING.md + fuzz/ placeholder + README dev-loop
+
+**Files:**
+- Create: `LEARNING.md`
+- Create: `fuzz/README.md`
+- Modify: `README.md` (add a "Slice 1 dev loop" section).
+
+**Interfaces:**
+- Consumes: the full workspace produced by Tasks 1–6.
+- Produces: the LEARNING.md index (spec §7 — at least 5 pointers), the `fuzz/` placeholder dir (spec §2), and the dev-loop doc slice in README.md (covers libopus install + manual e2e steps from spec §6.5).
+
+- [ ] **Step 1: Write `LEARNING.md`**
+
+```markdown
+# LEARNING.md — to learn concept X, read file Y
+
+This index maps a Rust concept you might be learning to the file where
+slice 1 makes the concept concrete. Each entry is a worked example you
+can read in `cargo doc --open` plus the source file itself.
+
+## Concepts + pointers
+
+- **Newtype pattern (type-safety via single-field wrappers)** →
+  `crates/rutster-call-model/src/lib.rs` — `ChannelId(Uuid)`. The newtype
+  stops us from mixing up a `ChannelId` with some future `SessionId` at
+  the type-system level. Compile-enforced where a comment could only ask.
+
+- **`enum` for closed state sets + exhaustive `match`** →
+  `crates/rutster-call-model/src/lib.rs` — `ChannelState` (New →
+  Connecting → Connected → Closing → Closed). Exhaustiveness checking
+  forces every `match` to consider each state; adding a state later
+  surfaces every site that needs handling.
+
+- **Sans-IO pattern (no I/O inside the library; input via method calls,
+  output via return values)** → `crates/rutster-media/src/loop_driver.rs`
+  — the str0m poll loop. `Rtc::handle_input` takes a network packet as a
+  struct argument, not from a socket the library owns; `poll_output`
+  returns `Transmit` packets the caller sends. Fully testable without a
+  network — str0m integration tests use this property to drive faster
+  than realtime.
+
+- **Trait design for extension points (a futures-compatible seam)** →
+  `crates/rutster-media/src/pcm.rs` — the `AudioSource` / `AudioSink`
+  traits. Slice 1 wires an `EchoAudioPipe` between them; step 2 swaps
+  that for a real WSS tap client without touching `RtcSession`. The
+  traits describe *what* the splice point does, not *how* it's filled.
+
+- **Error enums with `thiserror` + hot-path match-and-continue** →
+  `crates/rutster-media/src/lib.rs` (`MediaError`) and
+  `crates/rutster-media/src/opus_codec.rs` (`OpusDecoder::decode` returns
+  `Option`). Cold path: `thiserror`-derived enum + `?`. Hot
+  path: match-and-continue, never `?`, never panic — "drop + observe,
+  don't crash" (spec §3.8).
+
+- **`Arc>` vs `Arc>` — when each is right** →
+  `crates/rutster/src/session_map.rs`. The `RtcSession` lives behind
+  `Arc>` because every access mutates it (str0m's `&mut self`
+  contract) — `RwLock`'s read-mode would be useless. Comment on the
+  struct explains the trade-off.
+
+- **`DashMap` for sharded concurrent maps** →
+  `crates/rutster/src/session_map.rs`. `DashMap` shards its inner map so
+  two handlers operating on different `ChannelId`s don't contend;
+  `HashMap` wrapped in a single `Mutex` would serialize every access.
+
+- **str0m 0.21's single-mutation invariant** →
+  `crates/rutster-media/src/loop_driver.rs`. Mutate (handle_input /
+  Writer::write) → drain `poll_output` to `Output::Timeout` → next
+  mutate. Violating this leaves str0m in an inconsistent state.
+
+- **tokio graceful shutdown via signal handlers** →
+  `crates/rutster/src/main.rs` (`shutdown_signal`). Ctrl-C / SIGTERM
+  drops the AppState; the AppState drops the DashMap; the DashMap drops
+  every RtcSession. No in-flight call preservation in slice 1.
+
+- **`include_str!` for embedding static assets** →
+  `crates/rutster/src/routes.rs` (`include_str!("../static/index.html")`).
+  The HTML test client is compiled into the binary at build time — no
+  separate file to ship, no disk IO to serve it.
+
+## How to read
+
+1. `cargo doc --open` — every module has a `//!` doc comment; the doc
+   tree is the high-level map.
+2. Pick a concept above; open the named file. The first occurrence of
+   each non-obvious pattern has a `//` comment explaining *why*.
+3. Cross-ref back to the spec sections cited inline (`spec §3.8`,
+   `ADR-0002`, etc.) for the architecture-level rationale.
+```
+
+- [ ] **Step 2: Write `fuzz/README.md` (placeholder)**
+
+```markdown
+# fuzz/ — cargo-fuzz harness directory (placeholder)
+
+**Status:** placeholder. Not yet a cargo-fuzz project — just the directory.
+Fuzz harnesses land at spearhead step 5 (PSTN trunk) alongside the
+SIP/SDP/RTP wire parsers (PORT_PLAN §10 mandates continuous fuzzing of
+every wire parser). Slice 1 has no hostile-bytes surface (the browser is
+trusted), so no harnesses here yet — the `fuzz/` dir pre-paves the
+layout. Populating this directory with a real `cargo-fuzz` project
+(`fuzz/Cargo.toml` + `fuzz/fuzz_targets/*.rs`) happens at step 5.
+
+If you're at step 5, replace this README with that structure:
+- `fuzz/Cargo.toml` — cargo-fuzz manifest.
+- `fuzz/fuzz_targets/sip_parser.rs` — fuzz the SIP parser.
+- `fuzz/fuzz_targets/sdp_parser.rs` — fuzz the SDP parser.
+- `fuzz/fuzz_targets/rtp_packet.rs` — fuzz the RTP packet parser.
+- CI job running a short fuzz burst on each PR (the cargo-fuzz integration
+  lands in `.github/workflows/` at that point).
+
+The hot-path "drop + observe, don't crash" policy (spec §3.8) is what the
+future harnesses assert against: throw arbitrary bytes at the parser,
+assert it returns an error or drops silently — never panics.
+```
+
+- [ ] **Step 3: Add the dev-loop section to `README.md`**
+
+Find the existing dev-loop / "how to run" area in `README.md`. If none exists, add this section near the top, after the project framing:
+
+```markdown
+## Slice 1 dev loop (WebRTC media loopback)
+
+> Build prerequisite: install libopus (the `opus` crate links it via FFI):
+> ```bash
+> sudo apt-get install -y libopus-dev   # Debian/Ubuntu
+> # Fedora: sudo dnf install -y opus-devel
+> # macOS:  brew install opus
+> ```
+> This is the one system dependency in slice 1. Opus is FFI per PORT_PLAN
+> §7's "🦀 Core (FFI)" disposition — the codec surface Rust doesn't need
+> to re-implement.
+
+Run the server:
+
+```bash
+cargo run
+# listening on http://0.0.0.0:8080
+```
+
+Open a browser to `http://localhost:8080/`, click "Start call", grant
+microphone permission. Speak — you should hear yourself back within
+~200 ms (no perceptible delay). Click "Hang up" to tear down; server
+logs `Closing → Closed`.
+
+Verbose tracing:
+
+```bash
+RUST_LOG=rutster=debug cargo run
+```
+
+### Slice 1 "done" checklist (spec §6.5)
+
+On a clean checkout:
+1. `cargo test --all` passes.
+2. `cargo fmt --check` passes.
+3. `cargo clippy -- -D warnings` passes.
+4. `cargo deny check` passes.
+5. `cargo run` + browser manual e2e: speak → hear echo within ~200 ms.
+6. Hang-up button triggers `Closing → Closed` in server logs.
+7. Every stub crate compiles; its doc-comment names its scheduled step.
+8. `LEARNING.md` indexes at least 5 "to learn X, read Y" pointers.
+```
+
+- [ ] **Step 4: Run the full "done" checklist (spec §6.5)**
+
+```bash
+cargo fmt --check
+cargo clippy --all -- -D warnings
+cargo test --all
+cargo deny check
+```
+
+All four must pass before the commit. Flag the manual browser e2e (criterion #5) as "manually verified" or "pending" in the commit message.
+
+- [ ] **Step 5: Commit**
+
+```bash
+git add LEARNING.md fuzz/README.md README.md
+git commit -m "docs: LEARNING.md + fuzz/ placeholder + README dev-loop (spec §7, §6.3)
+
+LEARNING.md indexes ten concept-to-file pointers (the spec floor was
+five) — the newtype pattern, exhaustive enum match, sans-IO, trait
+extension seams, thiserror + hot-path match-and-continue, Arc
+vs Arc, DashMap, str0m's single-mutation invariant, graceful
+shutdown, include_str!. fuzz/README.md pre-paves the layout (no
+hostile-bytes surface in slice 1; harnesses land at step 5 per the
+out-of-scope table). README's new dev-loop section documents the
+libopus FFI prerequisite and the manual e2e steps."
+```
+
+---
+
+## Self-review (post-write)
+
+Ran the writing-plans self-review checklist:
+
+**1. Spec coverage** — every spec section maps to a task:
+- §2 workspace layout → Task 1 (workspace + stubs), Task 2 (call-model), Task 3 (media), Task 5 (binary).
+- §3.1 RtcSession + PcmFrame + codec pair → Tasks 3 + 4.
+- §3.2 loop shape (Approach A, Frame API) → Task 4 (`loop_driver.rs`).
+- §3.3 PCM tap seam (traits + EchoAudioPipe) → Task 3 (`pcm.rs`).
+- §3.4 tokio deviation → Task 4 (verbatim `DEV-DEVIATION` comment).
+- §3.5/§3.6 DTLS cert → Task 4 (str0m auto-generates; we feed it via `Rtc::new`, not explicit `set_dtls_cert` — the spec's "explicit is acceptable too" auto-gen default is honored).
+- §3.7 SDP — str0m's `accept_offer` does it natively. Plan documents this delta from §3.7's hand-rolled munger sketch (str0m 0.21 made the munger redundant).
+- §3.8 hot-path errors → Task 3 (Option-returning decode/encode) + Task 4 (match-and-continue drain).
+- §3.9 PCM format → Task 3 (24000 Hz mono, 480 samples, `SAMPLES_PER_FRAME`).
+- §4.1–§4.5 HTTP surface, ICE, security, browser client, session lifecycle, idle timeout, graceful shutdown → Task 5.
+- §5 call model → Task 2.
+- §6.1–§6.5 CI, dev loop, testing → Tasks 6 + 7.
+- §7 learner-facing docs → Task 7 + the documentation mandate embedded in Global Constraints (every task's code carries `//!`/`///`/`//` per the standard).
+- §8 design decisions → reflected in the choices the plan makes (str0m Frame API, EchoAudioPipe wiring, Channel = signaling-only state).
+
+**2. Placeholder scan:** No `TBD`/`TODO`/`implement later`. Every code step has complete code or an explicit "deferred per spec §1.2" reference. The residual str0m-API-uncertainty note in Task 4 step 7 (`SdpAnswer::mid()` and `PayloadParams::pt()` accessors) names the exact symbols the implementer should verify against `cargo doc -p str0m --open` — they are concrete verification instructions, not vague TODOs.
+
+**3. Type consistency:** `ChannelId`, `ChannelState`, `Channel`, `Direction` (Task 2) referenced identically in Tasks 4 + 5. `PcmFrame`, `AudioSource`, `AudioSink`, `EchoAudioPipe`, `OpusDecoder`, `OpusEncoder` (Task 3) referenced identically in Task 4. `RtcSession`, `RtcSessionError` (Task 4) referenced identically in Task 5. `AppState` (Task 5 step 4) referenced identically in Task 5 step 5 + the integration test (step 2) + `main.rs` (step 6).
+
+**4. Known deltas flagged in the plan:**
+- str0m 0.21's native `accept_offer` replaces §3.7's "50-line SDP munger" sketch.
+- `opus` crate's FFI links system libopus — amends §6.3's "no external deps beyond Rust" with the PORT_PLAN §7 rationale.
+- Single `RtcSession::new()` constructor (post-review patch) — no `new_for_test` / `new_for_server` split.
+
+**5. Adversarial review patches applied (post-write):**
+The plan was reviewed adversarially against str0m 0.21 and opus 0.3.1's real API surfaces (verified via docs.rs subagents). Patches landed:
+- Global Constraints: full str0m 0.21 API surface verified + documented (Rtc::new takes Instant; SdpOffer::from_sdp_string is the entry point, NOT from_str_unchecked; add_local_candidate returns Option<&Candidate>; Writer::write takes rtp_time not media_time; MediaTime has no add(Duration) — use `mt + MediaTime::from(d)`; payload_params returns impl Iterator; MediaData.data is Arc<[u8]>).
+- Task 4: `accept_offer` impl rewritten to use `from_sdp_string` + correct error mapping; `RtcSessionError::SdpOffer` changed to `String` (collapses parse + accept failures uniformly).
+- Task 4: `RtcSession::new_for_test` → `pub fn new()` (single idiomatic constructor, no test/prod split).
+- Task 4: added `accept_offer_transitions_channel_to_connecting` test (L3 — the transition was claimed but untested).
+- Task 4 loop_driver: `MediaTime::add` → `+ MediaTime::from(Duration)`; `media.data` deref coercion documented; `writer.payload_params().next().pt()` path clarified.
+- Task 5: dropped `reqwest` from workspace deps (unused — integration test uses `tower::ServiceExt::oneshot`); added `tower` as a workspace dep + to the binary crate's `[dev-dependencies]`.
+- Task 5: removed the duplicate `DELETE /v1/sessions` route (was chained via `.delete()` on the collection route AND on `/v1/sessions/:id` — only the latter is correct).
+- Task 5: clarified the `pub mod routes; pub mod session_map;` requirement (must be `pub` because integration tests need the absolute path).
+- Global Constraints: added task/PR strategy (one commit per task, merged in numeric order, granular history is load-bearing for the learning-codebase goal).
+
+Plan saved to `docs/superpowers/plans/2026-06-28-slice-1-webrtc-loopback.md`.