alee/rutster
0
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity
Files
c86bb19c0ee66cd6ce99ea75a5c52aa7e7a9889b
rutster/docs/superpowers/plans/2026-06-28-slice-1-webrtc-loopback.md
opencode controller 3b7b0d6459 docs: propagate rutster-trunk rename + repo URL through plans/specs/DEVELOPMENT 
Mechanical propagation of the crate rename ( rutster-signaling-sip →
rutster-trunk ) and the repo URL fix ( github.com/anomalyco →
git.adlee.work/alee ) through the documents that name them:

- docs/DEVELOPMENT.md: crate-layout sketch + stub-crate description.
- docs/superpowers/plans/2026-06-28-slice-1-webrtc-loopback.md: the
  workspace members list in the plan's binding-values section —
  repository URL updated.
- docs/superpowers/plans/2026-06-28-slice-2-agent-tap.md: workspace
  members list in binding-values + crate-layout sketch in the file-
  structure section (rutster-signaling-sip → rutster-trunk).
- docs/superpowers/specs/2026-06-28-slice-2-agent-tap-design.md: §2.1
  workspace-layout sketch's STUB crate-row.

No content changes beyond the substitutions; the slice-2 spec/plan
body (protocol, TapAudioPipe, TapClient, TapEngine, lifecycle, done-
  criteria) is untouched.
2026-06-29 20:27:01 -04:00

2647 lines
110 KiB
Markdown
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
# 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<SdpOffer, SdpError>`).
- Then `let answer: SdpAnswer = rtc.sdp_api().accept_offer(parsed_offer)?;` — `accept_offer` takes the owned `SdpOffer`, returns `Result<SdpAnswer, RtcError>`. `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<Protocol>) -> Result<Candidate, IceError>` — `"udp"` literal works because `&str: TryInto<Protocol>`.
- Inbound audio events arrive via the Frame API as `Event::MediaData(MediaData)`. `MediaData.data: Arc<[u8]>` is the encoded Opus payload (NOT `Vec<u8>` — it's an atomically-refcounted boxed slice; pass `&media.data[..]` to the decoder).
- Outbound: `let writer: Option<Writer<'_>> = 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<Pt>` (recommended — matches the incoming payload params) OR `writer.payload_params()` returns `impl Iterator<Item = &PayloadParams>`, 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<Arc<[u8]>>` — pass `&opus_bytes[..]` or `Vec<u8>` (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<usize>` (returns samples-per-channel decoded). `encoder.encode_vec(&pcm[..480], /*max_size*/ 4000) -> Result<Vec<u8>>`. 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<Mutex<...>>` vs `Arc<RwLock<...>>`, why `Pin<Box<dyn Future>>`, 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": "<uuid>" }`; `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<ChannelId, RtcSession>` 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://git.adlee.work/alee/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.85 as of writing — bumped from 1.80 because `uuid 1.x → getrandom 0.4.x` requires Rust `edition = "2024"`, stabilized in 1.85. Confirm the latest stable at impl time with `rustc --version`). The MSRV is the edition-2024 floor; the CI matrix (Task 6) tests stable + MSRV.
```toml
# rust-toolchain.toml — pins the toolchain for reproducible builds.
[toolchain]
channel = "1.85"
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<MediaLeg>` — second consumer.
/// - `audiohooks: Vec<AudiohookHandle>` — escalation rung 2.
/// - `tap: Option<TapHandle>` — 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<PcmFrame>` / `OpusEncoder::encode(&PcmFrame) -> Option<Vec<u8>>` — 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<PcmFrame>;
}
/// 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<Mutex<...>>` 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<PcmFrame>,
}
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<PcmFrame> {
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<PcmFrame>` and encode returns
//! `Option<Vec<u8>>` — 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<PcmFrame>` would also work; a flat array keeps the
// reuse obvious.
pcm_buf: [i16; SAMPLES_PER_FRAME],
}
impl OpusDecoder {
pub fn new() -> Result<Self, opus::Error> {
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<PcmFrame> {
// 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<Self, opus::Error> {
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<u8>` 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<Vec<u8>> {
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<PcmFrame>/Option<Vec<u8>>
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<String, RtcSessionError>` — 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<Duration>` — 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<Mutex<RtcSession>>` in the
/// binary's `DashMap<ChannelId, RtcSession>` (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<Mutex<...>>` (not `Arc<RwLock<...>>`)
///
/// 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<Mid>,
/// Last deadline from `Rtc::poll_output` — the next time the loop
/// should wake the rtc with `Input::Timeout`.
pub(crate) next_timeout: Option<Instant>,
/// 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, RtcSessionError> {
Self::new_internal()
}
fn new_internal() -> Result<Self, RtcSessionError> {
// Bind an ephemeral UDP socket. We use std::net::UdpSocket and
// drive it non-blocking from tokio rather than tokio's UdpSocket:
// str0m operates on raw `Receive` values and yields `Transmit`
// values, both of which are plain structs — no async needed.
// Setting non-blocking lets us `recv_from` without blocking.
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<String, RtcSessionError> {
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<Duration> {
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<Duration> {
// === 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<Instant> = 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<Arc<[u8]>>)
// -> Result<(), RtcError>
// - pt: payload type for Opus. `writer.payload_params()`
// returns `impl Iterator<Item = &PayloadParams>`; 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<Writer<'_>>` — `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<Pt>` 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<ChannelId, RtcSession>` + 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<ChannelId, RtcSession>` 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<DashMap<ChannelId, Arc<Mutex<RtcSession>>>>,
pub poll_running: Arc<Mutex<bool>>,
}
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<ChannelId, RtcSessionError> {
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<Arc<Mutex<RtcSession>>> {
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<ChannelId> = 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": "<uuid>" }`.
//! - `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<AppState>) -> 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<AppState>,
Path(id_str): Path<String>,
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<AppState>,
Path(id_str): Path<String>,
) -> 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
<!doctype html>
<html lang="en">
<head>
<meta charset="utf-8">
<title>Rutster slice-1 — WebRTC loopback</title>
<style>
body { font: 14px/1.4 system-ui, sans-serif; max-width: 60ch; margin: 2rem auto; }
pre { background: #f4f4f4; padding: 1rem; overflow: auto; }
button { font: inherit; padding: 0.4rem 1rem; }
</style>
</head>
<body>
<h1>Rutster slice-1 — WebRTC loopback</h1>
<p>Speak; you should hear yourself back within ~200 ms.</p>
<button id="start">Start call</button>
<button id="mute" disabled>Mute mic</button>
<button id="hangup" disabled>Hang up</button>
<pre id="log"></pre>
<audio id="audio" autoplay></audio>
<script>
const log = (s) => { document.getElementById('log').textContent += s + '\n'; };
const $ = (id) => document.getElementById(id);
let pc, sessionId, localStream;
$('start').onclick = async () => {
$('start').disabled = true;
$('mute').disabled = false;
$('hangup').disabled = false;
localStream = await navigator.mediaDevices.getUserMedia({ audio: true });
// No STUN (host candidates only — spec §4.4, zero-dependency dev loop).
pc = new RTCPeerConnection({ iceServers: [] });
localStream.getTracks().forEach(t => pc.addTrack(t, localStream));
pc.oniceconnectionstatechange = () => log('ICE: ' + pc.iceConnectionState);
pc.onconnectionstatechange = () => log('conn: ' + pc.connectionState);
pc.ontrack = (e) => { $('audio').srcObject = e.streams[0]; };
const offer = await pc.createOffer();
await pc.setLocalDescription(offer);
// Non-trickle: wait for ICE gathering to complete so all candidates
// travel in the SDP offer POST.
await new Promise(r => {
if (pc.iceGatheringState === 'complete') r();
else { pc.onicegatheringstatechange = () => { if (pc.iceGatheringState === 'complete') r(); }; };
});
const resp = await fetch('/v1/sessions', { method: 'POST' });
const { session_id } = await resp.json();
sessionId = session_id;
log('session: ' + session_id);
const ans = await fetch(`/v1/sessions/${session_id}/offer`, {
method: 'POST',
headers: { 'Content-Type': 'application/sdp' },
body: pc.localDescription.sdp,
});
const answerSdp = await ans.text();
await pc.setRemoteDescription({ type: 'answer', sdp: answerSdp });
};
$('mute').onclick = () => {
const track = localStream.getAudioTracks()[0];
track.enabled = !track.enabled;
$('mute').textContent = track.enabled ? 'Mute mic' : 'Unmute mic';
};
$('hangup').onclick = async () => {
await fetch(`/v1/sessions/${sessionId}`, { method: 'DELETE' });
pc.close();
$('hangup').disabled = true;
$('mute').disabled = true;
$('start').disabled = false;
log('hung up');
};
</script>
</body>
</html>
```
- [ ] **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<ChannelId, Arc<Mutex<RtcSession>>> (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.85"]
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.85), 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<PcmFrame>`). Cold path: `thiserror`-derived enum + `?`. Hot
path: match-and-continue, never `?`, never panic — "drop + observe,
don't crash" (spec §3.8).
- **`Arc<Mutex<T>>` vs `Arc<RwLock<T>>` — when each is right** →
`crates/rutster/src/session_map.rs`. The `RtcSession` lives behind
`Arc<Mutex<...>>` 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<Mutex>
vs Arc<RwLock>, 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`.
Reference in New Issue View Git Blame Copy Permalink