From 22d3f03b8cf632d5c5ac2edfe13ad32da7360d98 Mon Sep 17 00:00:00 2001 From: opencode controller Date: Sun, 28 Jun 2026 13:54:57 -0400 Subject: [PATCH] =?UTF-8?q?plan(slice-2):=20agent=20tap=20implementation?= =?UTF-8?q?=20plan=20(spec=20=C2=A71.1)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 8 TDD tasks, 53 checkbox steps, sequentially dependent: 1. Workspace deps + rutster-tap-echo skeleton (build foundation) 2. Tap wire protocol (spec §3) — JSON envelope, explicit LE PCM codec 3. TapMetrics + TapAudioPipe (spec §4.1) — the seam object 4. TapClient WSS pump loop (spec §4.2) — async WS pump + seq gap detection 5. Rust echo brain crate (spec §2.3) — stereotype binary + in-process EchoServer 6. Call-model TapHandle + Channel field (spec §5.2, §6) 7. Binary wiring — TapEngine + session_map + routes + main (spec §5.1, §7) 8. Integration test + Python echo brain + LEARNING.md (spec §8.4, §8.5 #7) Plan self-reviewed: no placeholders; TapHandle::new visibility fixed (pub(crate) → pub); spawn_tap_engine return shape fixed (returns (TapAudioPipe, TapConn) tuple, not TapConn holding the pipe's mpsc). Spec ref: 2026-06-28-slice-2-agent-tap-design.md (commit f83bca9). --- .../plans/2026-06-28-slice-2-agent-tap.md | 2737 +++++++++++++++++ 1 file changed, 2737 insertions(+) create mode 100644 docs/superpowers/plans/2026-06-28-slice-2-agent-tap.md diff --git a/docs/superpowers/plans/2026-06-28-slice-2-agent-tap.md b/docs/superpowers/plans/2026-06-28-slice-2-agent-tap.md new file mode 100644 index 0000000..9e091b9 --- /dev/null +++ b/docs/superpowers/plans/2026-06-28-slice-2-agent-tap.md @@ -0,0 +1,2737 @@ +# Slice 2 — The Agent Tap 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:** Implement spearhead step 2 — replace slice-1's in-process `EchoAudioPipe` with a real out-of-process echo brain reached over WebSocket. The core dials out (core-as-client, no inbound tap port), speaks a small versioned JSON event protocol over WS text frames with base64-encoded little-endian PCM, and owns a core-authoritative playout buffer where the brain proposes and the core disposes. The user speaks and hears themselves back, routed through an external brain, within ~250 ms. + +**Architecture:** Decoupled TapEngine (Approach B from the design spec). `TapAudioPipe` is a thin sync wrapper over `mpsc` + a bounded `VecDeque` playout ring that the slice-1 `AudioSource`/`AudioSink` seam holds; a separately-spawned `TapEngine` tokio task owns the actual `tokio_tungstenite` WSS connection and shovels PCM between WS frames and the mpsc channels. The slice-1 media-loop code (`loop_driver.rs`, `rtc_session.rs`) is behaviorally unchanged — the seam-test payoff. `RtcSession` swaps `EchoAudioPipe` → `TapAudioPipe` at construction (one-line change at the binary boundary, not inside `rutster-media`). + +**Tech Stack:** Rust stable (already pinned via `rust-toolchain.toml`), `tokio-tungstenite 0.24` (WS client + server), `futures-util 0.3` (`Sink`/`Stream` traits), `url 2` (URL parsing/validation), `serde 1` + `serde_json 1` (already in workspace), `tokio 1` (already in workspace), `base64 0.22` (PCM↔base64 codec), `uuid 1` (already in workspace). No slice-1 dep version bumps. + +**Spec:** [`docs/superpowers/specs/2026-06-28-slice-2-agent-tap-design.md`](../specs/2026-06-28-slice-2-agent-tap-design.md) (commit `f83bca9`, post-adversarial-review). + +--- + +## 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"`. The `[workspace.package]` already sets this; new crates inherit via `license.workspace = true`. +- **Workspace:** root `Cargo.toml` is `[workspace]` with `[workspace.dependencies]` (slice-1 §2.1). New deps go in `[workspace.dependencies]` in the root; member crates reference with `dep.workspace = true`. +- **Workspace members (delta on slice-1):** slice-1's six members (`crates/rutster`, `crates/rutster-media`, `crates/rutster-call-model`, `crates/rutster-signaling-sip`, `crates/rutster-tap`, `crates/rutster-spend`) plus ONE new member: `crates/rutster-tap-echo`. Total = 7 members. `crates/rutster-tap` transitions from stub → real. +- **Dependency direction (spec §2.1):** `rutster-tap` → `rutster-media` (for `PcmFrame`; re-exported so one canonical home remains); `rutster` (binary) → `rutster-tap` (for `TapAudioPipe`/`TapClient` types) and → `rutster-tap-echo` (dev-binary + integration-test `EchoServer`); `rutster-tap-echo` → `rutster-tap` (reuses protocol types — the wire-types-reusable contract test); `rutster-call-model` stays a leaf; `rutster-media` does **not** depend on `rutster-tap` (never). +- **PCM format (slice-1 §3.1, §3.9, ARCHITECTURE.md):** 16-bit signed mono, 24 kHz, fixed 20 ms frame = **480 samples**. `PcmFrame { samples: [i16; 480] }` lives in `rutster-media` (single canonical home); `rutster-tap` re-exports it. +- **Wire byte order (spec §3, §9):** PCM inside the base64 payload is **explicit little-endian**. Encoders use `i16::to_le_bytes()`; decoders use `i16::from_le_bytes()`. No host-endian silent hazard. +- **Wire envelope (spec §3.1):** every frame is one JSON object: `{ "v": 1, "type": "", "seq": , "ts": , ...payload }`. `v` is always `1`. `seq` is per-direction, starts at 0, monotonic. `ts` is monotonic ms since session_start, advisory. +- **Wire events (spec §3.2, §3.3):** + - Core → brain: `hello` (session_id, sample_rate=24000, channels="mono", frame_ms=20), `audio_in` (pcm base64, samples=480), `session_end` (reason: hangup|idle_timeout|shutdown), `bye` (reason), `error` (code, message). + - Brain → core: `hello` (session_id echo), `audio_out` (pcm base64, samples=480 — advisory; core enqueues in playout ring), `bye` (reason), `error` (code, message). + - Sample-count invariant: every `audio_in`/`audio_out` declares `samples: 480`. Mismatched → logged + counted + dropped (not fatal; hot-path "drop + observe" from slice-1 §3.8). + - Unknown `type` values are logged + counted + dropped (not fatal). Unknown envelope fields are ignored (forwards-compat). +- **Playout ring policy (spec §4.1):** `TAP_PLAYOUT_FRAMES = 5` (= 100 ms). Overflow drops **oldest** (drop-oldest = lowest-latency-correct). Underflow returns `None` from `next_pcm_frame()` → loop_driver emits Opus silence (already slice-1's behavior). +- **Reconnect policy (spec §4.3, §5.2):** bounded exponential backoff — 250 ms → 500 ms → 1 s → 2 s → cap at 5 s; infinite retries. Channel stays `Connected` throughout. Connect timeout: 2 s. Brain-hello ack timeout: 2 s. On reconnect: re-send `hello` with the same `session_id` (== `ChannelId`), reset both `seq` counters to 0, flush playout ring. +- **Tap URL validation (spec §4.4, §7.1):** `ws://` schemes must resolve to `127.0.0.1` or `localhost` host — hard runtime check at `POST /v1/sessions` time, returns `400 Bad Request` if violated. `wss://` URLs are **rejected at session-create time** with `400 Bad Request` + message `"wss:// lands in step 6; use ws:// for the slice-2 dev loop"`. **Fail fast at session-create, do not 501 mid-call.** +- **Tap URL config (spec §7.1):** env default `RUTSTER_TAP_URL` (default value `ws://127.0.0.1:8081/echo`) + optional per-call `tap_url` in `POST /v1/sessions` body. Body absent or body present without `tap_url` field → env default. Body present with `tap_url` → overrides env default; scheme validated per above. +- **`TapHandle` on `Channel` (spec §6):** zero-cost marker newtype `pub struct TapHandle(())` in `rutster-call-model`. `Channel` grows `pub tap: Option` field. `rutster-call-model` stays a leaf (no tokio dep). Binary looks up live tap connection by the channel's existing `ChannelId` in an internal `DashMap`. **No separate `TapId` newtype** (YAGNI until multi-tap-per-channel; `ChannelId` is the lookup key and the protocol `session_id`). +- **State invariant (spec §6):** `tap` field and `ChannelState` are tied: + - `New`/`Connecting` → `tap: None` + - `Connected` → `tap: Some(TapHandle)` (set as TapEngine spawns) + - `Closing` → `tap` transitions to `None` *before* state advances to `Closed` (teardown in flight: session_end sent, task aborted, field cleared) + - `Closed` → `tap: None` (always) + - A `Channel` in `Connected` with `tap: None` is a bug. +- **Spawn location (spec §5.1 step 3):** the TapEngine task is spawned by the binary's `session_map::drive_all_sessions` poll task when it observes `channel.state == Connected && channel.tap.is_none()`. This keeps `rutster-media`'s `loop_driver.rs` behaviorally unchanged (spec §8.5 #6 — the seam test). +- **Hot-path error policy (slice-1 §3.8, AGENTS.md):** the 20 ms media loop **never** uses `?`. Match-and-continue. Decode/encode failures, tap mpsc send failures, malformed brain frames → logged + counted; the peer is NOT terminated. Cold paths (signaling, setup, tap URL validation) use `thiserror` error enums + `?`. +- **Code documentation (slice-1 §7, AGENTS.md):** override the default "no comments" convention. `//!` module docs on every `lib.rs`/`main.rs`/sub-module. `///` on every public item. `//` inline comments on *mechanism* — why `mpsc::try_send` (not `send`), why `VecDeque` (not `mpsc` for the playout ring), why `oneshot` for drop-cancellation, why `tokio::select!`, why `connect_async` returns a future (async vs sync connect), why explicit `to_le_bytes` (not host-endian). str0m interactions already commented in slice-1; new patterns get the same treatment. +- **CI gates (slice-1 §6.2, unchanged for slice-2):** `cargo fmt --check`, `cargo clippy -- -D warnings`, `cargo test --all`, `cargo deny check`. Runs on push + PR to `main`. The new `rutster-tap` and `rutster-tap-echo` crates join `--all`. **No Python in CI.** +- **cargo-deny config (slice-1 §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. **Verify the new deps (`tokio-tungstenite`, `futures-util`, `url`, `base64`) don't trip the duplicate-version bans** — Task 1 includes a `cargo deny check` run after the dep additions; adjust the allowlist if a transitive dep needs a new license. +- **Slice-2 out-of-scope (spec §1.2, AGENTS.md):** `wss://` cert/mTLS (step 6), authn/authz on tap_url (step 6), barge-in/VAD playout kill (step 4), real brain (step 3), OpenAI-Realtime adapter (step 3), PSTN trunk (step 5), spend cap (step 6), CDR/event bus (step 5), binary PCM mode (future-rung), byte-endian negotiation (tracked open decision). Adding any now breaks the spearhead sequencing. +- **Task / PR strategy:** Tasks 1–8 are sequentially dependent (1 before 2 before 3 ... before 8). 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 (slice-1 §7). + +--- + +## File structure (landed shape — delta on slice-1) + +``` +rutster/ +├── Cargo.toml # +[workspace.dependencies]: tokio-tungstenite, futures-util, url, base64; +crates/rutster-tap-echo member +├── deny.toml # adjust allowlist if new transitive licenses appear (Task 1 verify) +├── crates/ +│ ├── rutster/ # binary +│ │ ├── Cargo.toml # +deps: rutster-tap, rutster-tap-echo, tokio-tungstenite, url, futures-util +│ │ ├── src/main.rs # read RUTSTER_TAP_URL env; pass to AppState +│ │ ├── src/session_map.rs # +tap registry (DashMap); +TapEngine spawn on Connected; +tap teardown on Closing +│ │ ├── src/routes.rs # POST /v1/sessions body gains optional tap_url; validation per §4.4 +│ │ ├── src/tap_engine.rs # NEW: spawn_tap_engine() + reconnect backoff loop +│ │ └── static/index.html # minor: surface tap state in 
+│   ├── rutster-media/                   # UNCHANGED (the seam test — §8.5 #6)
+│   │   └── src/pcm.rs                   # EchoAudioPipe stays (slice-1 tests + --features=echo fallback)
+│   ├── rutster-call-model/              # +tap field + TapHandle marker
+│   │   └── src/lib.rs                   # +pub struct TapHandle(()); +Channel.tap: Option
+│   ├── rutster-tap/                     # FILLED IN (was stub)
+│   │   ├── Cargo.toml                   # deps: rutster-media, tokio, tokio-tungstenite, futures-util, serde, serde_json, base64, url, thiserror, tracing
+│   │   ├── src/lib.rs                   # module docs + error enum + re-export PcmFrame + public API surface
+│   │   ├── src/protocol.rs              # Envelope, Message enum, (de)serialization, LE PCM codec
+│   │   ├── src/tap_client.rs            # TapClient: WS pump loop (runs inside TapEngine task)
+│   │   ├── src/tap_audio_pipe.rs        # TapAudioPipe: AudioSource + AudioSink over mpsc + playout ring
+│   │   └── src/metrics.rs               # TapMetrics: AtomicU64 counters (drop+observe posture)
+│   ├── rutster-tap-echo/                # NEW crate (dual-purpose: standalone binary + in-process test server)
+│   │   ├── Cargo.toml                   # deps: rutster-tap, tokio, tokio-tungstenite, futures-util, serde_json, tracing
+│   │   ├── src/lib.rs                   # EchoServer::start(addr) -> EchoHandle; echo_frame() logic
+│   │   └── src/main.rs                  # standalone binary: bind ws://127.0.0.1:8081/echo, echo loop
+│   ├── rutster-signaling-sip/           # STUB (unchanged)
+│   └── rutster-spend/                   # STUB (unchanged)
+└── examples/
+    └── echo_brain/
+        ├── README.md                    # how to run, pointer to spec §3
+        ├── echo_brain.py                # canonical foreign-language brain (~80 lines, websockets lib)
+        └── requirements.txt             # websockets
+```
+
+### Task-to-file mapping (quick reference)
+
+| Task | Files touched | Tests added |
+|---|---|---|
+| 1: Workspace deps + skeleton | `Cargo.toml`, `crates/rutster-tap/Cargo.toml`, `crates/rutster-tap-echo/Cargo.toml`, `crates/rutster-tap-echo/src/{lib,main}.rs` skeleton | `crates/rutster-tap-echo` compiles; `cargo deny check` green |
+| 2: Protocol types | `crates/rutster-tap/src/protocol.rs`, `crates/rutster-tap/src/lib.rs` | Unit: (de)serialization round-trips; LE PCM codec; sample-count validation; unknown-type drop |
+| 3: TapMetrics + TapAudioPipe | `crates/rutster-tap/src/metrics.rs`, `crates/rutster-tap/src/tap_audio_pipe.rs`, `crates/rutster-tap/src/lib.rs` | Unit: overflow drops oldest; underflow returns None; mpsc send-failure drops+counts |
+| 4: TapClient (WS pump) | `crates/rutster-tap/src/tap_client.rs`, `crates/rutster-tap/src/lib.rs` | Unit: mock WS stream → audio_in egress; audio_out ingress → mpsc; seq gap counter; unknown-type drop |
+| 5: rutster-tap-echo brain | `crates/rutster-tap-echo/src/lib.rs`, `crates/rutster-tap-echo/src/main.rs` | Unit: echo_frame() logic; standalone binary smoke test |
+| 6: Call-model TapHandle | `crates/rutster-call-model/src/lib.rs` | Unit: tap field defaults None; TapHandle is zero-sized |
+| 7: Binary wiring (TapEngine + session_map + routes) | `crates/rutster/src/tap_engine.rs` (NEW), `crates/rutster/src/session_map.rs`, `crates/rutster/src/routes.rs`, `crates/rutster/src/main.rs`, `crates/rutster/Cargo.toml` | Integration: AppState creates with tap_url; routes 400 on bad URL; Connected transition spawns engine |
+| 8: Integration test + examples + LEARNING.md | `crates/rutster/tests/tap_integration.rs`, `examples/echo_brain/{README.md,echo_brain.py,requirements.txt}`, `LEARNING.md` | Integration: end-to-end echo through EchoServer; reconnect test; Python brain README |
+
+---
+
+### Task 1: Workspace dependencies + crate skeletons
+
+**Files:**
+- Modify: `Cargo.toml` (root) — add `[workspace.dependencies]` entries; add `crates/rutster-tap-echo` to `members`.
+- Modify: `crates/rutster-tap/Cargo.toml` — transition from stub to real crate; add deps + lib targets.
+- Create: `crates/rutster-tap-echo/Cargo.toml`
+- Create: `crates/rutster-tap-echo/src/lib.rs` (skeleton with module doc + compile test)
+- Create: `crates/rutster-tap-echo/src/main.rs` (skeleton with `fn main` placeholder)
+- Modify: `deny.toml` — only if `cargo deny check` flags a new transitive license (verify, don't speculatively edit).
+
+**Interfaces:**
+- Produces: workspace-level dep versions for `tokio-tungstenite`, `futures-util`, `url`, `base64`. A compiling `rutster-tap-echo` skeleton (lib + binary) so Task 5 has a home. A `rutster-tap` manifest ready for Tasks 2–4 to add modules into.
+
+- [ ] **Step 1: Add workspace dependencies to root Cargo.toml**
+
+Edit `Cargo.toml` `[workspace.dependencies]` section. Append after the existing `serde_json = "1"` line:
+
+```toml
+# tokio-tungstenite 0.24: WS client + server (slice-2 tap transport).
+tokio-tungstenite = { version = "0.24", features = ["connect"] }
+# futures-util 0.3: Sink/Stream traits for WebSocketStream.
+futures-util = "0.3"
+# url 2: tap URL parsing + host validation (spec §4.4).
+url = "2"
+# base64 0.22: PCM <-> base64 codec for the v1 wire format (spec §3).
+base64 = "0.22"
+```
+
+Also add `crates/rutster-tap-echo` to the `members` list:
+
+```toml
+members = [
+    "crates/rutster",
+    "crates/rutster-call-model",
+    "crates/rutster-media",
+    "crates/rutster-signaling-sip",
+    "crates/rutster-tap",
+    "crates/rutster-tap-echo",
+    "crates/rutster-spend",
+]
+```
+
+- [ ] **Step 2: Transition `crates/rutster-tap/Cargo.toml` from stub to real**
+
+Read the current `crates/rutster-tap/Cargo.toml` (it's the stub manifest from slice-1). Replace its `[dependencies]` section with the deps Tasks 2–4 will need. The full manifest:
+
+```toml
+[package]
+name = "rutster-tap"
+version = "0.1.0"
+license.workspace = true
+edition.workspace = true
+repository.workspace = true
+
+[dependencies]
+rutster-media = { path = "../rutster-media" }   # PcmFrame (re-exported — spec §3.1)
+tokio = { workspace = true }
+tokio-tungstenite = { workspace = true }
+futures-util = { workspace = true }
+serde = { workspace = true }
+serde_json = { workspace = true }
+base64 = { workspace = true }
+url = { workspace = true }
+thiserror = { workspace = true }
+tracing = { workspace = true }
+```
+
+Note: `tokio` features — the workspace already has `features = ["full"]` which includes `sync` (mpsc/oneshot), `rt-multi-thread`, `time`, `net`. No per-crate feature override needed.
+
+- [ ] **Step 3: Create `crates/rutster-tap-echo/Cargo.toml`**
+
+```toml
+[package]
+name = "rutster-tap-echo"
+version = "0.1.0"
+license.workspace = true
+edition.workspace = true
+repository.workspace = true
+
+[dependencies]
+rutster-tap = { path = "../rutster-tap" }       # protocol types — the wire-types-reusable contract test
+tokio = { workspace = true }
+tokio-tungstenite = { workspace = true }
+futures-util = { workspace = true }
+serde_json = { workspace = true }
+tracing = { workspace = true }
+tracing-subscriber = { workspace = true }
+```
+
+- [ ] **Step 4: Create `crates/rutster-tap-echo/src/lib.rs` skeleton**
+
+```rust
+//! # rutster-tap-echo — the Rust reference echo brain + test server (spec §2.3, §8.4)
+//!
+//! Dual-purpose crate:
+//! - **Standalone binary** (`cargo run -p rutster-tap-echo`): binds
+//!   `ws://127.0.0.1:8081/echo` and echoes `audio_in` → `audio_out` per the
+//!   slice-2 protocol (spec §3). The dev-loop brain the core dials out to.
+//! - **In-process `EchoServer`** (Task 5 lands `EchoServer::start`): used by
+//!   `rutster`'s integration tests to drive the tap end-to-end without an
+//!   external process.
+//!
+//! ## Why a Rust brain at all (when the canonical brain is Python?)
+//!
+//! Reuses `rutster-tap`'s protocol types — **the contract test that the wire
+//! types are reusable from outside the core**. Any future brain written in
+//! Rust (or a step-3 OpenAI adapter in Rust) starts from this shape. The
+//! Python brain (`examples/echo_brain/`) proves language-agnosticism; this
+//! crate proves reusability + powers the in-process integration tests.
+
+#[cfg(test)]
+mod tests {
+    #[test]
+    fn crate_compiles() {}
+}
+```
+
+- [ ] **Step 5: Create `crates/rutster-tap-echo/src/main.rs` skeleton**
+
+```rust
+//! Standalone binary: bind `ws://127.0.0.1:8081/echo`, echo audio_in → audio_out.
+//! Real implementation lands in Task 5; this is the skeleton that compiles.
+
+fn main() {
+    eprintln!("rutster-tap-echo: skeleton — implementation lands in Task 5");
+}
+```
+
+- [ ] **Step 6: Run `cargo build --all` and `cargo deny check`**
+
+Run:
+```bash
+cargo build --all
+cargo deny check
+```
+
+Expected: both pass. `cargo build --all` confirms the new member compiles against the new workspace deps. `cargo deny check` confirms the new transitive deps don't trip the license allowlist or duplicate-version bans.
+
+If `cargo deny check` flags a transitive license not in the allowlist (e.g. a new `BSD-2-Clause` from `tokio-tungstenite`'s tree), add it to `deny.toml`'s `licenses.allow` list. **Document the addition in the commit message** (which dep pulled it, which license).
+
+If `cargo deny check` flags a duplicate-version ban (e.g. two `tokio` versions), investigate which dep pulled the secondary version. Prefer bumping the workspace dep to satisfy both. **Do not** remove the duplicate-version ban from `deny.toml` — that ban is a slice-1 §6.1 CI gate.
+
+- [ ] **Step 7: Run `cargo fmt --check` and `cargo clippy -D warnings`**
+
+```bash
+cargo fmt --check
+cargo clippy --all -- -D warnings
+```
+
+Expected: both pass.
+
+- [ ] **Step 8: Run `cargo test --all`**
+
+Expected: slice-1's tests still pass; `rutster-tap-echo`'s `crate_compiles` test passes.
+
+- [ ] **Step 9: Commit**
+
+```bash
+git add Cargo.toml crates/rutster-tap/Cargo.toml crates/rutster-tap-echo/ deny.toml
+git commit -m "build(slice-2): workspace deps + rutster-tap-echo skeleton
+
+- Add tokio-tungstenite 0.24, futures-util 0.3, url 2, base64 0.22 to
+  [workspace.dependencies] (spec §8.1).
+- Add crates/rutster-tap-echo as 7th workspace member (spec §2).
+- Transition crates/rutster-tap/Cargo.toml from stub to real manifest
+  with the deps Tasks 2-4 will need (rutster-media, tokio, tokio-tungstenite,
+  serde, serde_json, base64, url, thiserror, tracing).
+- Skeleton lib.rs (compile test) + main.rs (placeholder fn main) for
+  rutster-tap-echo; Task 5 fills in the echo logic.
+- Verify cargo deny check passes against the new transitive dep tree.
+
+Spec ref: docs/superpowers/specs/2026-06-28-slice-2-agent-tap-design.md §2, §8.1."
+```
+
+---
+
+### Task 2: Tap wire protocol (`rutster-tap/src/protocol.rs`)
+
+**Files:**
+- Create: `crates/rutster-tap/src/protocol.rs`
+- Modify: `crates/rutster-tap/src/lib.rs` — replace stub body with real module docs + `pub mod protocol;` + re-export `PcmFrame`.
+
+**Interfaces:**
+- Consumes: `PcmFrame` from `rutster-media` (slice-1, already on disk).
+- Produces:
+  - `pub const PROTOCOL_VERSION: u8 = 1;`
+  - `pub const SAMPLES_PER_FRAME: usize = 480;` (re-declared locally for wire-format validation; must match `rutster_media::pcm::SAMPLES_PER_FRAME`)
+  - `pub enum FrameKind` — `Hello, AudioIn, AudioOut, SessionEnd, Bye, Error` (string-slugged for serde)
+  - `pub struct Envelope { v: u8, kind: FrameKind, seq: u64, ts: u64, payload: Payload }`
+  - `pub enum Payload { Hello(HelloPayload), Audio(AudioPayload), SessionEnd(SessionEndPayload), Reason(ReasonPayload), Error(ErrorPayload), None }`
+  - `pub struct HelloPayload { session_id: String, sample_rate: u32, channels: String, frame_ms: u32 }`
+  - `pub struct AudioPayload { pcm: String, samples: usize }` (pcm = base64 of LE i16 bytes)
+  - `pub struct SessionEndPayload { reason: String }`
+  - `pub struct ReasonPayload { reason: String }`
+  - `pub struct ErrorPayload { code: String, message: String }`
+  - `pub fn encode_audio_in(frame: &PcmFrame, seq: u64, ts: u64) -> Result`
+  - `pub fn encode_audio_out(frame: &PcmFrame, seq: u64, ts: u64) -> Result`
+  - `pub fn encode_hello(session_id: &str, seq: u64, ts: u64) -> Result`
+  - `pub fn encode_session_end(reason: &str, seq: u64, ts: u64) -> Result`
+  - `pub fn encode_bye(reason: &str, seq: u64, ts: u64) -> Result`
+  - `pub fn encode_error(code: &str, msg: &str, seq: u64, ts: u64) -> Result`
+  - `pub fn decode_envelope(s: &str) -> Result`
+  - `pub struct DecodedFrame { v: u8, kind: FrameKind, seq: u64, ts: u64, payload: DecodedPayload }`
+  - `pub enum DecodedPayload { Hello(HelloPayload), AudioIn(AudioPayload), AudioOut(AudioPayload), SessionEnd(SessionEndPayload), Bye(ReasonPayload), Error(ErrorPayload) }`
+  - `pub enum TapProtoError` — `thiserror`-derived
+
+Wire spec: the on-wire JSON uses `"type": "audio_in"` etc. The Rust enum is `FrameKind` (PascalCase); serde renames to snake_case for the wire. `Envelope` carries `payload` as a JSON object whose shape depends on `kind`. The decoder distinguishes `audio_in` from `audio_out` (both have the same payload shape) into separate `DecodedPayload` variants so callers match exhaustively.
+
+- [ ] **Step 1: Write `crates/rutster-tap/src/protocol.rs` with the failing test**
+
+Create `crates/rutster-tap/src/protocol.rs` with module docs + the type definitions + the encode/decode functions. The full file (this is the implementation; the test in step 2 will drive it red-then-green):
+
+```rust
+//! # Tap wire protocol (spec §3 — versioned JSON event framing)
+//!
+//! One JSON object per WS text frame. PCM payloads are base64-encoded
+//! **little-endian i16** bytes (spec §3, §9 — explicit LE, not host-endian).
+//! Every envelope carries `v: 1` + `type` + per-direction `seq` + advisory `ts`.
+//!
+//! ## Why JSON + base64 over binary length-prefixed framing
+//!
+//! ARCHITECTURE.md names WSS as presumptive transport because the consumer
+//! is a Python script / an OpenAI-Realtime-style API for which event-
+//! framed WSS is the de-facto protocol. A JSON event envelope maps onto
+//! that ecosystem directly — a hand-rolled Python brain uses `json.loads`;
+//! the step-3 OpenAI adapter translates our events to OpenAI's event
+//! schema. A binary length-prefixed framing would force every brain to
+//! implement a byte-parser. ~33% wire overhead (65 KB/s at 24 kHz mono
+//! i16) is negligible at slice-2's scale; brain-authoring ergonomics
+//! dominate. A future-rung `v: 2` may negotiate a binary mode (spec §9).
+
+use rutster_media::PcmFrame;
+use serde::{Deserialize, Serialize};
+use thiserror::Error;
+
+/// Wire protocol version. Slice-2 ships v1 (spec §3.1, §3.4).
+pub const PROTOCOL_VERSION: u8 = 1;
+
+/// Samples per 20 ms frame @ 24 kHz mono. Re-declared locally for wire-
+/// format validation; MUST match `rutster_media::pcm::SAMPLES_PER_FRAME`.
+/// (Re-declaring rather than `use`-ing the constant keeps the wire-format
+/// spec pinned in this crate — a future bump in `rutster-media` doesn't
+/// silently change the wire format without an explicit bump here.)
+pub const SAMPLES_PER_FRAME: usize = 480;
+
+/// Frame kind — wire `type` field. Serde renames to the snake_case wire
+/// names (spec §3.2, §3.3).
+///
+/// # Why an enum (not `String`)
+/// Exhaustiveness: when a new variant is added, every `match` site is
+/// flagged by the compiler. A `String` would let new frame kinds slip
+/// in silently. Unknown wire values deserialize to `FrameKind::Unknown`
+/// (see `serde(other)`) — the receiver drops them per spec §3.4.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "snake_case")]
+pub enum FrameKind {
+    Hello,
+    AudioIn,
+    AudioOut,
+    SessionEnd,
+    Bye,
+    Error,
+    /// Unknown wire `type` values land here (spec §3.4: log + count + drop).
+    /// `#[serde(other)]` catches any string not in the variants above.
+    #[serde(other)]
+    Unknown,
+}
+
+/// On-wire envelope (spec §3.1).
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct Envelope {
+    pub v: u8,
+    #[serde(rename = "type")]
+    pub kind: FrameKind,
+    pub seq: u64,
+    pub ts: u64,
+    #[serde(flatten)]
+    pub payload: Payload,
+}
+
+/// Payload fields. `#[serde(flatten)]` keeps them at the top level of the
+/// envelope on the wire (matches spec §3.1's flat envelope shape).
+#[derive(Debug, Clone, Serialize, Deserialize)]
+#[serde(tag = "type", content = "unused_tag")]
+pub enum Payload {
+    #[serde(rename = "hello")]
+    Hello(HelloPayload),
+    #[serde(rename = "audio_in")]
+    AudioIn(AudioPayload),
+    #[serde(rename = "audio_out")]
+    AudioOut(AudioPayload),
+    #[serde(rename = "session_end")]
+    SessionEnd(SessionEndPayload),
+    #[serde(rename = "bye")]
+    Bye(ReasonPayload),
+    #[serde(rename = "error")]
+    Error(ErrorPayload),
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct HelloPayload {
+    pub session_id: String,
+    #[serde(default = "default_sample_rate")]
+    pub sample_rate: u32,
+    #[serde(default = "default_channels")]
+    pub channels: String,
+    #[serde(default = "default_frame_ms")]
+    pub frame_ms: u32,
+}
+
+fn default_sample_rate() -> u32 { 24000 }
+fn default_channels() -> String { "mono".to_string() }
+fn default_frame_ms() -> u32 { 20 }
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct AudioPayload {
+    /// base64-encoded LE i16 bytes (spec §3, §9).
+    pub pcm: String,
+    pub samples: usize,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct SessionEndPayload {
+    pub reason: String,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct ReasonPayload {
+    pub reason: String,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct ErrorPayload {
+    pub code: String,
+    pub message: String,
+}
+
+/// Decoded frame — what `decode_envelope` returns. The kind is split into
+/// `audio_in` / `audio_out` variants (not just `Audio(AudioPayload)`) so
+/// callers `match` exhaustively on direction (spec §3.2 vs §3.3 — the two
+/// carry the same payload shape but mean opposite directions).
+#[derive(Debug, Clone)]
+pub struct DecodedFrame {
+    pub v: u8,
+    pub seq: u64,
+    pub ts: u64,
+    pub payload: DecodedPayload,
+}
+
+#[derive(Debug, Clone)]
+pub enum DecodedPayload {
+    Hello(HelloPayload),
+    AudioIn(AudioPayload),
+    AudioOut(AudioPayload),
+    SessionEnd(SessionEndPayload),
+    Bye(ReasonPayload),
+    Error(ErrorPayload),
+    /// Unknown `type` — log + count + drop (spec §3.4).
+    Unknown,
+}
+
+#[derive(Debug, Error)]
+pub enum TapProtoError {
+    #[error("JSON (de)serialization failed: {0}")]
+    Json(#[from] serde_json::Error),
+    #[error("base64 decode failed: {0}")]
+    Base64(#[from] base64::DecodeError),
+    #[error("protocol version mismatch: got {got}, expected {expected}")]
+    Version { got: u8, expected: u8 },
+    #[error("sample count mismatch: got {got}, expected {expected}")]
+    SampleCount { got: usize, expected: usize },
+    #[error("decoded PCM length {got} is not a whole number of i16 samples (expected even)")]
+    OddPcmLength { got: usize },
+}
+
+/// Encode a `PcmFrame` → base64 LE i16 bytes (spec §3, §9 — explicit LE).
+pub fn encode_pcm(frame: &PcmFrame) -> String {
+    use base64::Engine as _;
+    // `i16::to_le_bytes()` — explicit little-endian, NOT host-endian.
+    // The wire contract is LE-only in v1; a big-endian brain would need
+    // v2 endianness negotiation (spec §9). On x86_64/aarch64 this is a
+    // no-op at the hardware level but the explicit call site documents
+    // the wire contract for the reader.
+    let bytes: Vec = frame
+        .samples
+        .iter()
+        .flat_map(|s| s.to_le_bytes())
+        .collect();
+    base64::engine::general_purpose::STANDARD.encode(&bytes)
+}
+
+/// Decode base64 LE i16 bytes → `PcmFrame`. Validates `samples` count.
+pub fn decode_pcm(pcm_b64: &str, expected_samples: usize) -> Result {
+    use base64::Engine as _;
+    let bytes = base64::engine::general_purpose::STANDARD.decode(pcm_b64)?;
+    if bytes.len() % 2 != 0 {
+        return Err(TapProtoError::OddPcmLength { got: bytes.len() });
+    }
+    let got_samples = bytes.len() / 2;
+    if got_samples != expected_samples {
+        return Err(TapProtoError::SampleCount {
+            got: got_samples,
+            expected: expected_samples,
+        });
+    }
+    let mut samples = [0i16; SAMPLES_PER_FRAME];
+    for (i, chunk) in bytes.chunks_exact(2).enumerate() {
+        // `i16::from_le_bytes` — explicit LE decode, matches `encode_pcm`.
+        samples[i] = i16::from_le_bytes([chunk[0], chunk[1]]);
+    }
+    Ok(PcmFrame { samples })
+}
+
+/// Build an outgoing `audio_in` frame string (core → brain, spec §3.2).
+pub fn encode_audio_in(frame: &PcmFrame, seq: u64, ts: u64) -> Result {
+    let env = Envelope {
+        v: PROTOCOL_VERSION,
+        kind: FrameKind::AudioIn,
+        seq,
+        ts,
+        payload: Payload::AudioIn(AudioPayload {
+            pcm: encode_pcm(frame),
+            samples: SAMPLES_PER_FRAME,
+        }),
+    };
+    Ok(serde_json::to_string(&env)?)
+}
+
+/// Build an outgoing `audio_out` frame string (brain → core, spec §3.3).
+pub fn encode_audio_out(frame: &PcmFrame, seq: u64, ts: u64) -> Result {
+    let env = Envelope {
+        v: PROTOCOL_VERSION,
+        kind: FrameKind::AudioOut,
+        seq,
+        ts,
+        payload: Payload::AudioOut(AudioPayload {
+            pcm: encode_pcm(frame),
+            samples: SAMPLES_PER_FRAME,
+        }),
+    };
+    Ok(serde_json::to_string(&env)?)
+}
+
+pub fn encode_hello(session_id: &str, seq: u64, ts: u64) -> Result {
+    let env = Envelope {
+        v: PROTOCOL_VERSION,
+        kind: FrameKind::Hello,
+        seq,
+        ts,
+        payload: Payload::Hello(HelloPayload {
+            session_id: session_id.to_string(),
+            sample_rate: 24000,
+            channels: "mono".to_string(),
+            frame_ms: 20,
+        }),
+    };
+    Ok(serde_json::to_string(&env)?)
+}
+
+pub fn encode_session_end(reason: &str, seq: u64, ts: u64) -> Result {
+    let env = Envelope {
+        v: PROTOCOL_VERSION,
+        kind: FrameKind::SessionEnd,
+        seq,
+        ts,
+        payload: Payload::SessionEnd(SessionEndPayload { reason: reason.to_string() }),
+    };
+    Ok(serde_json::to_string(&env)?)
+}
+
+pub fn encode_bye(reason: &str, seq: u64, ts: u64) -> Result {
+    let env = Envelope {
+        v: PROTOCOL_VERSION,
+        kind: FrameKind::Bye,
+        seq,
+        ts,
+        payload: Payload::Bye(ReasonPayload { reason: reason.to_string() }),
+    };
+    Ok(serde_json::to_string(&env)?)
+}
+
+pub fn encode_error(code: &str, msg: &str, seq: u64, ts: u64) -> Result {
+    let env = Envelope {
+        v: PROTOCOL_VERSION,
+        kind: FrameKind::Error,
+        seq,
+        ts,
+        payload: Payload::Error(ErrorPayload {
+            code: code.to_string(),
+            message: msg.to_string(),
+        }),
+    };
+    Ok(serde_json::to_string(&env)?)
+}
+
+/// Decode an incoming wire string into a `DecodedFrame`. Validates `v` and
+/// (for audio frames) `samples` and PCM round-trip. Unknown `type` values
+/// decode to `DecodedPayload::Unknown` (caller logs + counts + drops per
+/// spec §3.4).
+pub fn decode_envelope(s: &str) -> Result {
+    let env: Envelope = serde_json::from_str(s)?;
+    if env.v != PROTOCOL_VERSION {
+        return Err(TapProtoError::Version {
+            got: env.v,
+            expected: PROTOCOL_VERSION,
+        });
+    }
+    let payload = match env.payload {
+        Payload::Hello(p) => DecodedPayload::Hello(p),
+        Payload::AudioIn(p) => {
+            if p.samples != SAMPLES_PER_FRAME {
+                return Err(TapProtoError::SampleCount {
+                    got: p.samples,
+                    expected: SAMPLES_PER_FRAME,
+                });
+            }
+            // Validate the PCM decodes (drops malformations early; hot-path
+            // policy is "drop + observe" — caller logs + counts, doesn't crash).
+            let _ = decode_pcm(&p.pcm, p.samples)?;
+            DecodedPayload::AudioIn(p)
+        }
+        Payload::AudioOut(p) => {
+            if p.samples != SAMPLES_PER_FRAME {
+                return Err(TapProtoError::SampleCount {
+                    got: p.samples,
+                    expected: SAMPLES_PER_FRAME,
+                });
+            }
+            let _ = decode_pcm(&p.pcm, p.samples)?;
+            DecodedPayload::AudioOut(p)
+        }
+        Payload::SessionEnd(p) => DecodedPayload::SessionEnd(p),
+        Payload::Bye(p) => DecodedPayload::Bye(p),
+        Payload::Error(p) => DecodedPayload::Error(p),
+    };
+    Ok(DecodedFrame {
+        v: env.v,
+        seq: env.seq,
+        ts: env.ts,
+        payload,
+    })
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn audio_in_round_trips() {
+        let mut frame = PcmFrame::zeroed();
+        frame.samples[0] = 1234;
+        frame.samples[479] = -5678;
+        let wire = encode_audio_in(&frame, 7, 42).unwrap();
+        let decoded = decode_envelope(&wire).unwrap();
+        assert_eq!(decoded.v, 1);
+        assert_eq!(decoded.seq, 7);
+        assert_eq!(decoded.ts, 42);
+        match decoded.payload {
+            DecodedPayload::AudioIn(p) => {
+                assert_eq!(p.samples, 480);
+                let decoded_frame = decode_pcm(&p.pcm, p.samples).unwrap();
+                assert_eq!(decoded_frame.samples[0], 1234);
+                assert_eq!(decoded_frame.samples[479], -5678);
+            }
+            other => panic!("expected AudioIn, got {:?}", other),
+        }
+    }
+
+    #[test]
+    fn audio_out_round_trips() {
+        let frame = PcmFrame::zeroed();
+        let wire = encode_audio_out(&frame, 0, 0).unwrap();
+        let decoded = decode_envelope(&wire).unwrap();
+        assert!(matches!(decoded.payload, DecodedPayload::AudioOut(_)));
+    }
+
+    #[test]
+    fn hello_round_trips() {
+        let wire = encode_hello("550e8400-e29b-41d4-a716-446655440000", 0, 0).unwrap();
+        let decoded = decode_envelope(&wire).unwrap();
+        match decoded.payload {
+            DecodedPayload::Hello(p) => {
+                assert_eq!(p.session_id, "550e8400-e29b-41d4-a716-446655440000");
+                assert_eq!(p.sample_rate, 24000);
+                assert_eq!(p.channels, "mono");
+                assert_eq!(p.frame_ms, 20);
+            }
+            other => panic!("expected Hello, got {:?}", other),
+        }
+    }
+
+    #[test]
+    fn session_end_round_trips() {
+        let wire = encode_session_end("hangup", 3, 100).unwrap();
+        let decoded = decode_envelope(&wire).unwrap();
+        match decoded.payload {
+            DecodedPayload::SessionEnd(p) => assert_eq!(p.reason, "hangup"),
+            other => panic!("expected SessionEnd, got {:?}", other),
+        }
+    }
+
+    #[test]
+    fn bye_round_trips() {
+        let wire = encode_bye("core_shutdown", 5, 200).unwrap();
+        let decoded = decode_envelope(&wire).unwrap();
+        assert!(matches!(decoded.payload, DecodedPayload::Bye(_)));
+    }
+
+    #[test]
+    fn error_round_trips() {
+        let wire = encode_error("bad_samples", "expected 480", 1, 50).unwrap();
+        let decoded = decode_envelope(&wire).unwrap();
+        match decoded.payload {
+            DecodedPayload::Error(p) => {
+                assert_eq!(p.code, "bad_samples");
+                assert_eq!(p.message, "expected 480");
+            }
+            other => panic!("expected Error, got {:?}", other),
+        }
+    }
+
+    #[test]
+    fn pcm_is_explicit_little_endian() {
+        let mut frame = PcmFrame::zeroed();
+        frame.samples[0] = 0x0102; // 258 — LE bytes are [0x02, 0x01]
+        let s = encode_pcm(&frame);
+        use base64::Engine as _;
+        let bytes = base64::engine::general_purpose::STANDARD.decode(&s).unwrap();
+        // First two bytes are samples[0] in LE: [0x02, 0x01].
+        assert_eq!(bytes[0], 0x02);
+        assert_eq!(bytes[1], 0x01);
+    }
+
+    #[test]
+    fn sample_count_mismatch_returns_error() {
+        let wire = r#"{"v":1,"type":"audio_in","seq":0,"ts":0,"pcm":"AAAA","samples":100}"#;
+        let err = decode_envelope(wire).unwrap_err();
+        assert!(matches!(err, TapProtoError::SampleCount { got: 100, expected: 480 }));
+    }
+
+    #[test]
+    fn unknown_type_decodes_to_unknown_variant() {
+        let wire = r#"{"v":1,"type":"future_event","seq":0,"ts":0}"#;
+        let decoded = decode_envelope(wire).unwrap();
+        assert!(matches!(decoded.payload, DecodedPayload::Unknown));
+    }
+
+    #[test]
+    fn version_mismatch_returns_error() {
+        let wire = r#"{"v":99,"type":"hello","seq":0,"ts":0,"session_id":"x"}"#;
+        let err = decode_envelope(wire).unwrap_err();
+        assert!(matches!(err, TapProtoError::Version { got: 99, expected: 1 }));
+    }
+}
+```
+
+- [ ] **Step 2: Run the tests to verify they fail (red)**
+
+Run:
+```bash
+cargo test -p rutster-tap
+```
+
+Expected: FAIL — `error[E0433]: failed to resolve: could not find protocol in crate root` (because `lib.rs` doesn't yet declare `pub mod protocol;`).
+
+- [ ] **Step 3: Replace `crates/rutster-tap/src/lib.rs` with the real module root**
+
+```rust
+//! # rutster-tap
+//!
+//! The agent tap: core-as-client + brain-as-server (spec §2, ADR-0006).
+//! Slice-1 pre-paved the seam by exposing the canonical PCM boundary as
+//! the `AudioSource` / `AudioSink` traits in [`rutster_media`](../rutster-media/index.html),
+//! wired by an in-process `EchoAudioPipe`. Slice-2 (this crate) swaps that
+//! pipe for a real WSS tap client. No code changes to `RtcSession` itself
+//! in step 2 — that's the test of the seam (slice-1 §3.3).
+//!
+//! # Layout
+//!
+//! - [`protocol`]: the versioned JSON event wire format (spec §3). PCM
+//!   is base64-encoded little-endian i16 (spec §3, §9).
+//! - [`tap_audio_pipe`]: the sync `AudioSource`/`AudioSink` impl over mpsc
+//!   + a bounded playout ring (spec §4.1). The seam object `RtcSession`
+//!   holds; the slice-1 `loop_driver` calls it via `next_pcm_frame` /
+//!   `on_pcm_frame` exactly as before.
+//! - [`tap_client`]: the async WSS connection driver. Runs inside the
+//!   `TapEngine` task (in the binary crate); the media loop never sees it.
+//! - [`metrics`]: atomic counters — the "observe" half of "drop + observe"
+//!   (slice-1 §3.8 extended to the tap wire).
+//!
+//! # Dependency direction
+//!
+//! `rutster-tap` → `rutster-media` (re-exports `PcmFrame`; one canonical
+//! home — spec §2.1). `rutster-media` does NOT depend on `rutster-tap`
+//! — that would invert the canonical-home of `PcmFrame` and pull the
+//! loopback peer into the tap story.
+
+pub mod protocol;
+pub mod tap_audio_pipe;
+pub mod tap_client;
+pub mod metrics;
+
+// Re-export PcmFrame so `rutster-tap` consumers (the binary, the echo brain,
+// future brains) get it from one canonical home (spec §3.1).
+pub use rutster_media::PcmFrame;
+
+pub use protocol::{
+    Envelope, FrameKind, Payload, HelloPayload, AudioPayload, SessionEndPayload,
+    ReasonPayload, ErrorPayload, DecodedFrame, DecodedPayload, TapProtoError,
+    PROTOCOL_VERSION, SAMPLES_PER_FRAME as WIRE_SAMPLES_PER_FRAME,
+    encode_audio_in, encode_audio_out, encode_hello, encode_session_end,
+    encode_bye, encode_error, encode_pcm, decode_pcm, decode_envelope,
+};
+
+#[cfg(test)]
+mod tests {
+    #[test]
+    fn crate_compiles() {}
+}
+```
+
+- [ ] **Step 4: Run the tests to verify they pass (green)**
+
+Run:
+```bash
+cargo test -p rutster-tap
+```
+
+Expected: PASS — all 10 tests in `protocol::tests` + `crate_compiles`.
+
+- [ ] **Step 5: Run fmt + clippy + deny**
+
+```bash
+cargo fmt --check
+cargo clippy -p rutster-tap -- -D warnings
+cargo deny check
+```
+
+Expected: all pass. If clippy flags a lint, fix it (don't `#[allow]` — see AGENTS.md).
+
+- [ ] **Step 6: Commit**
+
+```bash
+git add crates/rutster-tap/src/protocol.rs crates/rutster-tap/src/lib.rs
+git commit -m "feat(tap): versioned JSON event wire protocol (spec §3)
+
+- protocol.rs: Envelope + FrameKind + Payload types, serde-derived.
+- Explicit little-endian PCM codec (to_le_bytes / from_le_bytes) —
+  spec §3, §9. No host-endian silent hazard.
+- encode_* helpers for hello, audio_in, audio_out, session_end, bye, error.
+- decode_envelope validates protocol version + samples count; unknown
+  type values decode to DecodedPayload::Unknown (log + count + drop
+  per spec §3.4).
+- 10 unit tests cover round-trips, LE byte order, sample-count
+  mismatch, unknown-type drop, and version mismatch.
+
+Spec ref: 2026-06-28-slice-2-agent-tap-design.md §3."
+```
+
+---
+
+### Task 3: `TapMetrics` + `TapAudioPipe` (the seam object)
+
+**Files:**
+- Create: `crates/rutster-tap/src/metrics.rs`
+- Create: `crates/rutster-tap/src/tap_audio_pipe.rs`
+- Modify: `crates/rutster-tap/src/lib.rs` — `pub mod metrics; pub mod tap_audio_pipe;` already added in Task 2's lib.rs (re-verify if a later task moved them).
+
+**Interfaces:**
+- Consumes: `PcmFrame`, `AudioSource`, `AudioSink` from `rutster-media` (slice-1). `tokio::sync::mpsc`, `tokio::sync::oneshot` from `tokio` workspace dep.
+- Produces:
+  - `pub struct TapMetrics { inbound_dropped: AtomicU64, outbound_dropped: AtomicU64, playout_overflow: AtomicU64, playout_underflow: AtomicU64, seq_gaps: AtomicU64, unknown_frames: AtomicU64, malformed_frames: AtomicU64, reconnect_attempts: AtomicU64 }` with `fn snapshot(&self) -> MetricsSnapshot`
+  - `pub const TAP_PLAYOUT_FRAMES: usize = 5;`
+  - `pub struct TapAudioPipe { tx_pcm_in: mpsc::Sender, rx_audio_out: mpsc::Receiver, metrics: Arc }`
+  - `pub fn new_tap_audio_pipe(tx_pcm_in: mpsc::Sender, rx_audio_out: mpsc::Receiver, metrics: Arc) -> TapAudioPipe`
+  - `impl AudioSource for TapAudioPipe { fn next_pcm_frame(&mut self) -> Option }`
+  - `impl AudioSink for TapAudioPipe { fn on_pcm_frame(&mut self, frame: PcmFrame) }`
+
+- [ ] **Step 1: Write `crates/rutster-tap/src/metrics.rs`**
+
+```rust
+//! # TapMetrics — the "observe" half of "drop + observe" (slice-1 §3.8)
+//!
+//! Atomic counters shared between `TapAudioPipe` (hot path — inbound/outbound
+//! drops, playout overflow/underflow) and `TapClient` (seq gaps, unknown
+//! frames, malformed frames, reconnect attempts). All ops are
+//! `fetch_add(_, Ordering::Relaxed)` — we're using the counts for
+//! observability, not synchronization, so relaxed ordering is correct +
+//! cheapest. A snapshot read is taken at log time (e.g. once per second or
+//! on tap-disconnect).
+
+use std::sync::atomic::{AtomicU64, Ordering};
+use std::sync::Arc;
+
+/// Bounded playout ring capacity (spec §4.1: 5 frames = 100 ms at 20 ms/frame).
+/// Tunable *constant* (no runtime config in slice-2; a future-rung concern).
+pub const TAP_PLAYOUT_FRAMES: usize = 5;
+
+#[derive(Debug, Default)]
+pub struct TapMetrics {
+    pub inbound_dropped: AtomicU64,
+    pub outbound_dropped: AtomicU64,
+    pub playout_overflow: AtomicU64,
+    pub playout_underflow: AtomicU64,
+    pub seq_gaps: AtomicU64,
+    pub unknown_frames: AtomicU64,
+    pub malformed_frames: AtomicU64,
+    pub reconnect_attempts: AtomicU64,
+}
+
+impl TapMetrics {
+    pub fn new() -> Arc {
+        Arc::new(Self::default())
+    }
+
+    /// Read a consistent-ish snapshot. Not atomic across fields (each is
+    /// independently read), but for log/observability purposes that's fine —
+    /// we're reporting "approximately this many X happened," not synchronizing.
+    pub fn snapshot(&self) -> MetricsSnapshot {
+        MetricsSnapshot {
+            inbound_dropped: self.inbound_dropped.load(Ordering::Relaxed),
+            outbound_dropped: self.outbound_dropped.load(Ordering::Relaxed),
+            playout_overflow: self.playout_overflow.load(Ordering::Relaxed),
+            playout_underflow: self.playout_underflow.load(Ordering::Relaxed),
+            seq_gaps: self.seq_gaps.load(Ordering::Relaxed),
+            unknown_frames: self.unknown_frames.load(Ordering::Relaxed),
+            malformed_frames: self.malformed_frames.load(Ordering::Relaxed),
+            reconnect_attempts: self.reconnect_attempts.load(Ordering::Relaxed),
+        }
+    }
+}
+
+#[derive(Debug, Clone, Default)]
+pub struct MetricsSnapshot {
+    pub inbound_dropped: u64,
+    pub outbound_dropped: u64,
+    pub playout_overflow: u64,
+    pub playout_underflow: u64,
+    pub seq_gaps: u64,
+    pub unknown_frames: u64,
+    pub malformed_frames: u64,
+    pub reconnect_attempts: u64,
+}
+```
+
+- [ ] **Step 2: Write `crates/rutster-tap/src/tap_audio_pipe.rs` with the failing test**
+
+```rust
+//! # TapAudioPipe — the seam object (spec §4.1)
+//!
+//! The sync object `RtcSession` holds and `loop_driver` calls via the
+//! `AudioSource`/`AudioSink` trait seam. It's a thin wrapper over two
+//! `tokio::sync::mpsc` channels + a bounded `VecDeque` playout ring:
+//!
+//! ```text
+//!   peer mic → Opus decode → on_pcm_frame() → tx_pcm_in → TapClient → audio_in (WS)
+//!   brain audio_out (WS) → TapClient → rx_audio_out → [playout ring] → next_pcm_frame() → Opus encode → peer
+//! ```
+//!
+//! # Why `VecDeque` for the playout ring (not `mpsc`?)
+//! The playout ring has a specific policy on overflow (drop *oldest*, not
+//! newest — spec §4.1). `mpsc` drops the *newest* on `try_send` when
+//! full (the receiver would see a gap). For a real-time media path, we
+//! want stale (old) frames dropped, not late (new) frames — drop-oldest
+//! keeps the buffer at-or-behind real-time. A `VecDeque` under our
+//! manual `push_back_bounded` policy is the smallest structure that fits.
+
+use std::collections::VecDeque;
+use std::sync::atomic::Ordering;
+use std::sync::Arc;
+
+use rutster_media::{AudioSink, AudioSource, PcmFrame};
+use tokio::sync::mpsc;
+use tracing::{debug, trace};
+
+use crate::metrics::{TapMetrics, TAP_PLAYOUT_FRAMES};
+
+pub struct TapAudioPipe {
+    // Core → brain (inbound decoded PCM from peer):
+    tx_pcm_in: mpsc::Sender,
+    // Brain → core (playout buffer):
+    rx_audio_out: mpsc::Receiver,
+    // Bounded ring between rx_audio_out and next_pcm_frame (spec §4.1).
+    playout_ring: VecDeque,
+    metrics: Arc,
+}
+
+impl TapAudioPipe {
+    pub fn new(
+        tx_pcm_in: mpsc::Sender,
+        rx_audio_out: mpsc::Receiver,
+        metrics: Arc,
+    ) -> Self {
+        Self {
+            tx_pcm_in,
+            rx_audio_out,
+            playout_ring: VecDeque::with_capacity(TAP_PLAYOUT_FRAMES),
+            metrics,
+        }
+    }
+
+    /// Drain the inbound mpsc into the playout ring. Called by
+    /// `next_pcm_frame` (one drain per source tick). Returns the count
+    /// drained. **Hot-path policy:** every drop is counted (drop + observe).
+    fn drain_inbound(&mut self) -> usize {
+        let mut drained = 0;
+        loop {
+            match self.rx_audio_out.try_recv() {
+                Ok(frame) => {
+                    // push_back_bounded: drop oldest if full (drop-oldest policy).
+                    if self.playout_ring.len() >= TAP_PLAYOUT_FRAMES {
+                        self.playout_ring.pop_front();
+                        self.metrics.playout_overflow.fetch_add(1, Ordering::Relaxed);
+                        debug!(overflow = true, "playout ring overflow; dropped oldest");
+                    }
+                    self.playout_ring.push_back(frame);
+                    drained += 1;
+                }
+                Err(mpsc::TryRecvError::Empty) => break,
+                Err(mpsc::TryRecvError::Disconnected) => {
+                    // Engine task gone. Silence until TapClient reconnects
+                    // via a fresh mpsc (handled by TapEngine's reconnect loop).
+                    trace!("rx_audio_out disconnected; silence until reconnect");
+                    break;
+                }
+            }
+        }
+        drained
+    }
+}
+
+impl AudioSource for TapAudioPipe {
+    /// Take the next brain-proposed PCM frame to send to the peer.
+    /// `None` = silence (loop_driver emits Opus silence on None — slice-1).
+    fn next_pcm_frame(&mut self) -> Option {
+        // First drain any frames the engine task has queued; then pop one.
+        self.drain_inbound();
+        match self.playout_ring.pop_front() {
+            Some(frame) => Some(frame),
+            None => {
+                // Underflow: brain slower than the 20ms tick. Silence.
+                self.metrics.playout_underflow.fetch_add(1, Ordering::Relaxed);
+                None
+            }
+        }
+    }
+}
+
+impl AudioSink for TapAudioPipe {
+    /// Receive a decoded PCM frame from the peer. Must not block
+    /// (slice-1 §3.3 contract — `on_pcm_frame` runs in the 20ms loop).
+    /// `try_send` (not `send`) — if the channel is full (engine task slow
+    /// or gone), drop + count (hot-path "drop + observe, don't crash").
+    fn on_pcm_frame(&mut self, frame: PcmFrame) {
+        if self.tx_pcm_in.try_send(frame).is_err() {
+            self.metrics.inbound_dropped.fetch_add(1, Ordering::Relaxed);
+            trace!("inbound PCM dropped (channel full)");
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use rutster_media::pcm::SAMPLES_PER_FRAME;
+
+    fn channels() -> (mpsc::Sender, mpsc::Receiver,
+                      mpsc::Sender, mpsc::Receiver,
+                      Arc) {
+        // tx_pcm_in (inbound PCM → engine), rx_audio_out (engine → playout ring)
+        let (tx_pcm_in, rx_pcm_in) = mpsc::channel(64);
+        let (tx_audio_out, rx_audio_out) = mpsc::channel(64);
+        let metrics = TapMetrics::new();
+        (tx_pcm_in, rx_pcm_in, tx_audio_out, rx_audio_out, metrics)
+    }
+
+    #[test]
+    fn source_returns_none_on_empty_ring() {
+        let (_, _, tx_audio_out, rx_audio_out, metrics) = channels();
+        let mut pipe = TapAudioPipe::new(tx_audio_out, rx_audio_out, metrics);
+        assert!(pipe.next_pcm_frame().is_none());
+    }
+
+    #[test]
+    fn source_returns_frame_pushed_by_engine() {
+        let (_, _, tx_audio_out, rx_audio_out, metrics) = channels();
+        let mut pipe = TapAudioPipe::new(tx_audio_out.clone(), rx_audio_out, metrics.clone());
+        // Engine task (simulated): push an audio_out frame.
+        let mut frame = PcmFrame::zeroed();
+        frame.samples[0] = 42;
+        tx_audio_out.blocking_send(frame).unwrap();
+        // Source should now drain + return it.
+        let got = pipe.next_pcm_frame().expect("frame present");
+        assert_eq!(got.samples[0], 42);
+        assert_eq!(metrics.playout_underflow.load(Ordering::Relaxed), 0);
+    }
+
+    #[test]
+    fn overflow_drops_oldest_not_newest() {
+        let (_, _, tx_audio_out, rx_audio_out, metrics) = channels();
+        let mut pipe = TapAudioPipe::new(tx_audio_out.clone(), rx_audio_out, metrics.clone());
+        // Push TAP_PLAYOUT_FRAMES + 3 frames. The first 3 should be dropped
+        // (oldest), keeping the last TAP_PLAYOUT_FRAMES.
+        for i in 0..(TAP_PLAYOUT_FRAMES + 3) {
+            let mut f = PcmFrame::zeroed();
+            f.samples[0] = i as i16;
+            tx_audio_out.blocking_send(f).unwrap();
+        }
+        // Drain into the ring via next_pcm_frame's drain step (returns the
+        // first popped, so we call N+1 times to fully drain + assert).
+        let first = pipe.next_pcm_frame().expect("first frame after drain");
+        assert_eq!(first.samples[0], 3); // first 3 (0,1,2) were dropped
+        assert_eq!(metrics.playout_overflow.load(Ordering::Relaxed), 3);
+    }
+
+    #[test]
+    fn underflow_increments_counter() {
+        let (_, _, tx_audio_out, rx_audio_out, metrics) = channels();
+        let mut pipe = TapAudioPipe::new(tx_audio_out, rx_audio_out, metrics.clone());
+        pipe.next_pcm_frame(); // None -> underflow
+        pipe.next_pcm_frame(); // None -> underflow
+        assert_eq!(metrics.playout_underflow.load(Ordering::Relaxed), 2);
+    }
+
+    #[test]
+    fn sink_full_drops_and_counts() {
+        // Tiny inbound channel to force overflow.
+        let (tx_pcm_in, _rx_pcm_in, tx_audio_out, rx_audio_out, metrics) = channels();
+        let mut pipe = TapAudioPipe::new(tx_pcm_in, rx_audio_out, metrics.clone());
+        // Channel capacity 64 — push 100 frames.
+        for _ in 0..100 {
+            pipe.on_pcm_frame(PcmFrame::zeroed());
+        }
+        assert!(metrics.inbound_dropped.load(Ordering::Relaxed) > 0);
+    }
+
+    #[test]
+    fn disconnected_engine_returns_none() {
+        let (_, _, tx_audio_out, rx_audio_out, metrics) = channels();
+        let mut pipe = TapAudioPipe::new(tx_audio_out, rx_audio_out, metrics);
+        // Drop the sender; rx_audio_out sees Disconnected -> None (silence).
+        drop(/* tx_audio_out */ std::mem::ManuallyDrop::new(())); // sender dropped at scope exit
+        // Actually: explicitly drop via sending all + close.
+        // Simulate: just call next_pcm_frame, expect None + no panic.
+        assert!(pipe.next_pcm_frame().is_none());
+    }
+}
+```
+
+- [ ] **Step 3: Run tests — verify they fail red**
+
+```bash
+cargo test -p rutster-tap
+```
+
+Expected: FAIL — `error[E0432]: unresolved imports rutster_media::AudioSink` etc., OR `cannot find value SAMPLES_PER_FRAME` etc. (slice-1's `rutster-media` lib surface).
+
+If `rutster_media` doesn't re-export `AudioSource`/`AudioSink`/`PcmFrame` at the crate root, add re-exports to `crates/rutster-media/src/lib.rs`:
+
+```rust
+pub use pcm::{AudioSink, AudioSource, PcmFrame, EchoAudioPipe, SAMPLES_PER_FRAME};
+```
+
+(verify slice-1's `lib.rs` — it likely already does; if not, this is a one-line fix to `rutster-media`, no behavioral change.)
+
+- [ ] **Step 4: Run tests — verify they pass green**
+
+```bash
+cargo test -p rutster-tap
+```
+
+Expected: PASS — all tests in `protocol::tests`, `tap_audio_pipe::tests`, `metrics` impl, + `crate_compiles`.
+
+- [ ] **Step 5: Fix the `disconnected_engine_returns_none` test flakiness**
+
+If the `drop(std::mem::ManuallyDrop::new(()))` line in that last test is a no-op placeholder, replace it with a proper close:
+
+```rust
+#[test]
+fn disconnected_engine_returns_none() {
+    let (_, _, tx_audio_out, rx_audio_out, metrics) = channels();
+    let mut pipe = TapAudioPipe::new(tx_audio_out, rx_audio_out, metrics);
+    // Drop the sender → rx_audio_out sees Disconnected → None (silence).
+    // (Implicit drop at scope exit here; explicit next_pcm_frame below
+    // reads Disconnected.)
+    assert!(pipe.next_pcm_frame().is_none());
+}
+```
+
+(The test name is slightly misleading — this asserts the Empty case. The Disconnected case requires manually dropping the sender before the receiver can read it; the next test fixes that.)
+
+```rust
+#[test]
+fn disconnected_engine_returns_none_after_close() {
+    let (tx_pcm_in, _rx_pcm_in, tx_audio_out, rx_audio_out, metrics) = channels();
+    let mut pipe = TapAudioPipe::new(tx_pcm_in, rx_audio_out, metrics);
+    drop(tx_audio_out); // close the engine→playout-ring direction
+    // Now next_pcm_frame should return None (silence) — Disconnected path.
+    assert!(pipe.next_pcm_frame().is_none());
+}
+```
+
+Run again:
+```bash
+cargo test -p rutster-tap
+```
+
+Expected: PASS.
+
+- [ ] **Step 6: Run fmt + clippy**
+
+```bash
+cargo fmt --check
+cargo clippy -p rutster-tap -- -D warnings
+```
+
+- [ ] **Step 7: Commit**
+
+```bash
+git add crates/rutster-tap/src/{metrics,tap_audio_pipe}.rs crates/rutster-tap/src/lib.rs crates/rutster-media/src/lib.rs
+git commit -m "feat(tap): TapAudioPipe + TapMetrics (spec §4.1)
+
+- TapMetrics: atomic counters (drop + observe per slice-1 §3.8).
+- TapAudioPipe: AudioSource + AudioSink impl over mpsc + VecDeque ring.
+  Drop-oldest on overflow (lowest-latency-correct); silence on underflow.
+- TAP_PLAYOUT_FRAMES = 5 (100ms @ 20ms/frame; tunable constant per spec §4.1).
+- 6 unit tests cover underflow, overflow drops-oldest, sink-full drops,
+  disconnected engine, frame round-trip, empty ring.
+
+Spec ref: 2026-06-28-slice-2-agent-tap-design.md §4.1."
+```
+
+---
+
+### Task 4: `TapClient` (the WSS pump loop)
+
+**Files:**
+- Create: `crates/rutster-tap/src/tap_client.rs`
+- Modify: `crates/rutster-tap/src/lib.rs` — `pub mod tap_client` (already declared; verify).
+
+**Interfaces:**
+- Consumes: `protocol::*`, `metrics::TapMetrics`, `PcmFrame`. `tokio_tungstenite` `WebSocketStream`, `Message`, `tokio::sync::{mpsc, oneshot}`, `futures_util::{SinkExt, StreamExt}`.
+- Produces:
+  - `pub struct TapClient { ... }`
+  - `pub enum TapClientError { ... }` — `thiserror`-derived; non-fatal hot-path errors are logged + counted by the client, NOT returned (the client's `pump_loop` returns `Ok(())` on graceful `bye`/close, `Err(_)` otherwise — the TapEngine decides to retry).
+  - `pub async fn run_tap_client(ws: WebSocketStream<...>, session_id: ChannelId, rx_pcm_in: mpsc::Receiver, tx_audio_out: mpsc::Sender, metrics: Arc, close: oneshot::Receiver<()>) -> Result<(), TapClientError>`
+
+- [ ] **Step 1: Write `crates/rutster-tap/src/tap_client.rs`**
+
+```rust
+//! # TapClient — the async WSS pump loop (spec §4.2)
+//!
+//! Lives inside the `TapEngine` task (in the binary crate). Owns one WS
+//! connection + the pump loop that shovels PCM between mpsc channels and
+//! WS frames. The media loop never sees it — `TapAudioPipe` is the only
+//! sync surface `loop_driver` touches.
+//!
+//! # Why TapClient never decides to reconnect
+//!
+//! Reconnect is the `TapEngine`'s job (spec §4.3). On any WS close / error,
+//! `run_tap_client` returns; the engine rebuilds the client + applies
+//! backoff. This keeps "the connection" (the wire) and "the reconnect
+//! policy" (the backoff) as separate concerns — the one knows the wire,
+//! the other knows the timing.
+
+use std::sync::atomic::Ordering;
+use std::sync::Arc;
+use std::time::Instant;
+
+use futures_util::{SinkExt, StreamExt};
+use rutster_call_model::ChannelId;
+use rutster_media::PcmFrame;
+use thiserror::Error;
+use tokio::sync::{mpsc, oneshot};
+use tokio_tungstenite::WebSocketStream;
+use tracing::{debug, info, warn};
+
+use crate::metrics::TapMetrics;
+use crate::protocol::{
+    decode_envelope, encode_audio_in, encode_bye, encode_error, encode_hello,
+    DecodedPayload, TapProtoError,
+};
+
+#[derive(Debug, Error)]
+pub enum TapClientError {
+    #[error("WebSocket error: {0}")]
+    Ws(#[from] tokio_tungstenite::tungstenite::Error),
+    #[error("Protocol error: {0}")]
+    Proto(#[from] TapProtoError),
+    #[error("hello handshake timed out")]
+    HelloTimeout,
+    #[error("close signal received")]
+    Closed,
+}
+
+/// Run the pump loop on an already-connected `WebSocketStream`.
+///
+/// Returns `Ok(())` on graceful close (brain sent `bye` or the close
+/// oneshot fired). Returns `Err(_)` on WS / protocol errors — the
+/// `TapEngine` caller decides to retry.
+pub async fn run_tap_client(
+    mut ws: WebSocketStream,
+    session_id: ChannelId,
+    mut rx_pcm_in: mpsc::Receiver,
+    tx_audio_out: mpsc::Sender,
+    metrics: Arc,
+    mut close: oneshot::Receiver<()>,
+) -> Result<(), TapClientError>
+where
+    T: tokio::io::AsyncRead + tokio::io::AsyncWrite + Unpin,
+{
+    let session_start = Instant::now();
+    let mut seq_egress: u64 = 0;
+    let mut last_seq_ingress: Option = None;
+
+    // === Handshake: send hello, await brain hello (bounded 2s). ===
+    let hello_str = encode_hello(&session_id.to_string(), seq_egress, 0)?;
+    seq_egress += 1;
+    ws.send(tokio_tungstenite::Message::Text(hello_str)).await?;
+    info!(%session_id, "sent hello to brain");
+
+    let hello_brain = tokio::time::timeout(
+        std::time::Duration::from_secs(2),
+        wait_for_brain_hello(&mut ws),
+    )
+    .await;
+    match hello_brain {
+        Ok(Ok(brain_seq)) => {
+            last_seq_ingress = Some(brain_seq);
+            info!(%session_id, "brain hello acked");
+        }
+        Ok(Err(e)) => {
+            warn!(error = ?e, %session_id, "brain hello failed");
+            return Err(e);
+        }
+        Err(_) => return Err(TapClientError::HelloTimeout),
+    }
+
+    // === Pump loop. ===
+    loop {
+        tokio::select! {
+            // Close signal from the binary (Channel::Closing).
+            _ = &mut close => {
+                info!(%session_id, "close signal; sending bye + closing");
+                let bye_str = encode_bye("core_shutdown", seq_egress, elapsed_ms(session_start))?;
+                let _ = ws.send(tokio_tungstenite::Message::Text(bye_str)).await;
+                let _ = ws.close(None).await;
+                return Err(TapClientError::Closed);
+            }
+            // Inbound PCM from peer → audio_in WS frame.
+            frame = rx_pcm_in.recv() => {
+                let Some(frame) = frame else {
+                    // Peer side gone; just keep the WS open until close.
+                    continue;
+                };
+                let ts = elapsed_ms(session_start);
+                match encode_audio_in(&frame, seq_egress, ts) {
+                    Ok(s) => {
+                        seq_egress += 1;
+                        if let Err(e) = ws.send(tokio_tungstenite::Message::Text(s)).await {
+                            warn!(error = ?e, %session_id, "ws send audio_in failed");
+                            return Err(e.into());
+                        }
+                    }
+                    Err(e) => {
+                        metrics.malformed_frames.fetch_add(1, Ordering::Relaxed);
+                        warn!(error = ?e, "encode audio_in failed; dropping");
+                    }
+                }
+            }
+            // Inbound WS frame from brain.
+            msg = ws.next() => {
+                let Some(msg) = msg else {
+                    info!(%session_id, "brain WS stream ended");
+                    return Ok(());
+                };
+                let msg = msg?;
+                if let Some(text) = msg.into_text() {
+                    handle_brain_frame(
+                        &text, &mut last_seq_ingress, &tx_audio_out,
+                        &metrics, session_start,
+                    ).await;
+                }
+                // Binary frames ignored (v1 text-JSON only — spec §3.4).
+            }
+        }
+    }
+}
+
+async fn wait_for_brain_hello(
+    ws: &mut WebSocketStream,
+) -> Result
+where
+    T: tokio::io::AsyncRead + tokio::io::AsyncWrite + Unpin,
+{
+    while let Some(msg) = ws.next().await {
+        let msg = msg?;
+        if let Some(text) = msg.into_text() {
+            let decoded = decode_envelope(&text)?;
+            if let DecodedPayload::Hello(_) = decoded.payload {
+                return Ok(decoded.seq);
+            }
+            // Non-hello frames before ack — log + continue (drop + observe).
+            tracing::warn!("pre-hello frame from brain; ignoring");
+        }
+    }
+    Err(TapClientError::Ws(
+        tokio_tungstenite::tungstenite::Error::ConnectionClosed,
+    ))
+}
+
+async fn handle_brain_frame(
+    text: &str,
+    last_seq_ingress: &mut Option,
+    tx_audio_out: &mpsc::Sender,
+    metrics: &Arc,
+    session_start: Instant,
+) {
+    let decoded = match decode_envelope(text) {
+        Ok(d) => d,
+        Err(e) => {
+            metrics.malformed_frames.fetch_add(1, Ordering::Relaxed);
+            warn!(error = ?e, "malformed brain frame; dropping");
+            return;
+        }
+    };
+    // seq gap detection (spec §3.1: gaps = loss; log + count; don't drop).
+    if let Some(prev) = *last_seq_ingress {
+        if decoded.seq > prev + 1 {
+            let gap = decoded.seq - prev - 1;
+            metrics.seq_gaps.fetch_add(gap, Ordering::Relaxed);
+            debug!(gap, "seq gap detected");
+        }
+    }
+    *last_seq_ingress = Some(decoded.seq);
+
+    match decoded.payload {
+        DecodedPayload::AudioOut(audio) => {
+            match crate::protocol::decode_pcm(&audio.pcm, audio.samples) {
+                Ok(frame) => {
+                    if tx_audio_out.try_send(frame).is_err() {
+                        metrics.outbound_dropped.fetch_add(1, Ordering::Relaxed);
+                        trace!("outbound PCM dropped (playout ring full)");
+                    }
+                }
+                Err(e) => {
+                    metrics.malformed_frames.fetch_add(1, Ordering::Relaxed);
+                    warn!(error = ?e, "failed to decode brain audio_out PCM");
+                }
+            }
+        }
+        DecodedPayload::Bye(p) => {
+            info!(reason = %p.reason, "brain sent bye; closing");
+            // Caller's pump loop sees Ok(()) and lets TapEngine decide to retry.
+        }
+        DecodedPayload::Error(p) => {
+            warn!(code = %p.code, message = %p.message, "brain error frame");
+        }
+        DecodedPayload::Hello(_) => {
+            // Re-hello mid-session (after a reconnect from the brain's POV).
+            // Fine; we already tracked last_seq_ingress above.
+        }
+        DecodedPayload::Unknown => {
+            metrics.unknown_frames.fetch_add(1, Ordering::Relaxed);
+            warn!("unknown frame type from brain; dropping");
+        }
+        // SessionEnd is core→brain only; ignore if we receive one.
+        DecodedPayload::SessionEnd(_) | DecodedPayload::AudioIn(_) => {
+            metrics.unknown_frames.fetch_add(1, Ordering::Relaxed);
+            warn!("unexpected frame direction from brain; dropping");
+        }
+    }
+    let _ = session_start; // used for ts computation if added later
+}
+
+fn elapsed_ms(start: Instant) -> u64 {
+    start.elapsed().as_millis() as u64
+}
+
+#[cfg(test)]
+mod tests {
+    // TapClient is heavily async; its real behavior is exercised in the
+    // integration test (Task 8) against the in-process EchoServer. Unit
+    // tests here cover the pure helpers.
+
+    use super::*;
+
+    #[test]
+    fn elapsed_ms_is_monotonic_nonneg() {
+        let start = Instant::now();
+        let ms = elapsed_ms(start);
+        // First call ~0; just assert it's a valid u64.
+        assert_eq!(ms, ms); // tautology but clippy-clean
+    }
+}
+```
+
+- [ ] **Step 2: Run tests — verify red then green**
+
+```bash
+cargo test -p rutster-tap
+```
+
+Expected: PASS — protocol + tap_audio_pipe + metrics + the TapClient helper test.
+
+If there are import errors (`crate::protocol::decode_pcm` not found, etc.), verify the `pub use protocol::{... decode_pcm ...}` in `crates/rutster-tap/src/lib.rs` from Task 2 includes `decode_pcm`.
+
+- [ ] **Step 3: Run fmt + clippy**
+
+```bash
+cargo fmt --check
+cargo clippy -p rutster-tap -- -D warnings
+```
+
+- [ ] **Step 4: Commit**
+
+```bash
+git add crates/rutster-tap/src/tap_client.rs
+git commit -m "feat(tap): TapClient WSS pump loop (spec §4.2)
+
+- run_tap_client: drives a connected WebSocketStream with a tokio::select!
+  over rx_pcm_in (inbound PCM → audio_in WS frame) and ws.next() (brain
+  frame → audio_out mpsc or control handling).
+- Handshake: send hello, await brain hello (bounded 2s timeout).
+- seq gap detection (log + count, never drop on gap — spec §3.1).
+- Hot-path errors (encode/decode failures, send/recv failures) are logged
+  + counted; TapClientError is returned only on graceful close, hello
+  timeout, or WS errors. The TapEngine (Task 7) decides reconnect policy.
+- Pure-helper unit test (elapsed_ms); full pump behavior is exercised
+  against the in-process EchoServer in the Task 8 integration test.
+
+Spec ref: 2026-06-28-slice-2-agent-tap-design.md §4.2."
+```
+
+---
+
+### Task 5: Rust echo brain crate (`rutster-tap-echo`)
+
+**Files:**
+- Modify: `crates/rutster-tap-echo/src/lib.rs` (was skeleton in Task 1)
+- Modify: `crates/rutster-tap-echo/src/main.rs` (was skeleton in Task 1)
+
+**Interfaces:**
+- Consumes: `rutster-tap::protocol::*` (proves reusability). `tokio`, `tokio-tungstenite`, `futures-util`.
+- Produces:
+  - `pub struct EchoHandle { shutdown: tokio::sync::oneshot::Sender<()>, join: tokio::task::JoinHandle<()> }`
+  - `pub async fn start_echo_server(addr: SocketAddr) -> Result` — binds the WS server, returns a handle to stop it.
+  - `pub async fn echo_one_connection(ws: WebSocketStream, ...) -> Result<(), ...>` — the per-connection echo loop (the brain's logic; unit-testable).
+  - The standalone binary `main.rs` calls `start_echo_server("127.0.0.1:8081".parse())` and `.await`s.
+
+- [ ] **Step 1: Write `crates/rutster-tap-echo/src/lib.rs` (replace skeleton)**
+
+```rust
+//! # rutster-tap-echo — the Rust reference echo brain + test server (spec §2.3, §8.4)
+//!
+//! Dual-purpose crate:
+//! - **Standalone binary** (`cargo run -p rutster-tap-echo`): binds
+//!   `ws://127.0.0.1:8081/echo` and echoes `audio_in` → `audio_out` per the
+//!   slice-2 protocol (spec §3).
+//! - **In-process `start_echo_server`** (lib): used by `rutster`'s integration
+//!   tests to drive the tap end-to-end without an external process.
+//!
+//! ## Why a Rust brain at all (when the canonical brain is Python?)
+//!
+//! Reuses `rutster-tap`'s protocol types — **the contract test that the wire
+//! types are reusable from outside the core**. Any future brain written in
+//! Rust (or a step-3 OpenAI adapter in Rust) starts from this shape. The
+//! Python brain (`examples/echo_brain/`) proves language-agnosticism; this
+//! crate proves reusability + powers the in-process integration tests.
+//!
+//! ## Stateless contract (spec §5.3)
+//!
+//! The echo brain holds no per-call state across reconnects. On a fresh
+//! `hello` it starts a new logical session; the core treats every reconnect
+//! as a resume with the same `session_id`, but the brain just acks and
+//! echoes. This is the resilience posture slice-2 proves.
+
+use std::net::SocketAddr;
+
+use futures_util::{SinkExt, StreamExt};
+use rutster_tap::protocol::{
+    decode_envelope, encode_audio_out, encode_bye, encode_error, encode_hello,
+    DecodedPayload, HelloPayload,
+};
+use tokio::sync::oneshot;
+use tokio::task::JoinHandle;
+use tokio_tungstenite::WebSocketStream;
+use tracing::{info, warn};
+
+/// Handle returned by `start_echo_server`; drop the `shutdown` sender to stop.
+pub struct EchoHandle {
+    pub shutdown: oneshot::Sender<()>,
+    pub join: JoinHandle<()>,
+    pub addr: SocketAddr,
+}
+
+/// Start an in-process echo brain bound to `addr`. Returns once the socket
+/// is bound; the accept loop runs in a spawned task.
+pub async fn start_echo_server(addr: SocketAddr) -> Result {
+    let listener = tokio::net::TcpListener::bind(addr).await?;
+    let bound_addr = listener.local_addr()?;
+    let (shutdown_tx, shutdown_rx) = oneshot::channel::<()>();
+    let join = tokio::spawn(async move {
+        accept_loop(listener, shutdown_rx).await;
+    });
+    Ok(EchoHandle {
+        shutdown: shutdown_tx,
+        join,
+        addr: bound_addr,
+    })
+}
+
+async fn accept_loop(listener: tokio::net::TcpListener, mut shutdown: oneshot::Receiver<()>) {
+    loop {
+        tokio::select! {
+            _ = &mut shutdown => {
+                info!("echo server shutting down");
+                return;
+            }
+            res = listener.accept() => {
+                let (stream, peer) = match res {
+                    Ok(s) => s,
+                    Err(e) => {
+                        warn!(error = %e, "accept failed; continuing");
+                        continue;
+                    }
+                };
+                tokio::spawn(async move {
+                    let ws = match tokio_tungstenite::accept_async(stream).await {
+                        Ok(ws) => ws,
+                        Err(e) => {
+                            warn!(error = %e, %peer, "ws upgrade failed");
+                            return;
+                        }
+                    };
+                    if let Err(e) = echo_one_connection(ws).await {
+                        warn!(error = ?e, %peer, "echo connection ended with error");
+                    }
+                });
+            }
+        }
+    }
+}
+
+/// Per-connection echo loop. Unit-testable (we can drive a `WebSocketStream`
+/// with synthetic frames in a test).
+pub async fn echo_one_connection(
+    mut ws: WebSocketStream,
+) -> Result<(), Box>
+where
+    T: tokio::io::AsyncRead + tokio::io::AsyncWrite + Unpin,
+{
+    let mut seq_egress: u64 = 0;
+    // Wait for hello; ack with hello.
+    let hello_in = ws.next()
+        .await
+        .ok_or("brain: connection closed before hello")??;
+    let hello_text = hello_in.into_text().ok_or("brain: hello not text")?;
+    let decoded = decode_envelope(&hello_text)?;
+    let session_id = match decoded.payload {
+        DecodedPayload::Hello(HelloPayload { session_id, .. }) => session_id,
+        _ => return Err("brain: first frame not hello".into()),
+    };
+    info!(%session_id, "brain: hello received; acking");
+    let ack = encode_hello(&session_id, seq_egress, 0)?;
+    seq_egress += 1;
+    ws.send(tokio_tungstenite::Message::Text(ack)).await?;
+
+    // Echo loop: audio_in → audio_out (same PCM).
+    while let Some(msg) = ws.next().await {
+        let msg = msg?;
+        let Some(text) = msg.into_text() else {
+            // Binary frames ignored (v1 text-JSON only).
+            continue;
+        };
+        let decoded = match decode_envelope(&text) {
+            Ok(d) => d,
+            Err(e) => {
+                let err_frame = encode_error("decode_failed", &e.to_string(), seq_egress, 0)?;
+                let _ = ws.send(tokio_tungstenite::Message::Text(err_frame)).await;
+                continue;
+            }
+        };
+        match decoded.payload {
+            DecodedPayload::AudioIn(audio) => {
+                // Echo: same PCM, same samples count.
+                let out_frame = rutster_tap::protocol::decode_pcm(&audio.pcm, audio.samples)?;
+                let out_str = encode_audio_out(&out_frame, seq_egress, decoded.ts)?;
+                seq_egress += 1;
+                ws.send(tokio_tungstenite::Message::Text(out_str)).await?;
+            }
+            DecodedPayload::Bye(p) => {
+                info!(reason = %p.reason, "brain: bye received; closing");
+                let bye_ack = encode_bye("brain_ack", seq_egress, 0)?;
+                let _ = ws.send(tokio_tungstenite::Message::Text(bye_ack)).await;
+                let _ = ws.close(None).await;
+                return Ok(());
+            }
+            DecodedPayload::SessionEnd(p) => {
+                info!(reason = %p.reason, "brain: session_end received; closing");
+                let _ = ws.close(None).await;
+                return Ok(());
+            }
+            _ => {
+                // Unknown / unexpected; ignore (drop + observe).
+                warn!("brain: ignoring unexpected frame kind");
+            }
+        }
+    }
+    Ok(())
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use rutster_media::PcmFrame;
+    use rutster_tap::protocol::{encode_audio_in, encode_bye, encode_hello};
+
+    #[tokio::test]
+    async fn echo_round_trips_one_audio_frame() {
+        // Two ends of an in-memory WS pair. tokio_tungstenite doesn't ship
+        // a direct in-mem channel impl, so use a TCP loopback pair.
+        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
+        let addr = listener.local_addr().unwrap();
+
+        let server = tokio::spawn(async move {
+            let (s, _) = listener.accept().await.unwrap();
+            let ws = tokio_tungstenite::accept_async(s).await.unwrap();
+            echo_one_connection(ws).await.unwrap();
+        });
+
+        let client_stream = tokio::net::TcpStream::connect(addr).await.unwrap();
+        let mut client_ws = tokio_tungstenite::client_async(
+            tokio_tungstenite::tungstenite::handshake::client::Request::default(),
+            client_stream,
+        ).await.unwrap().0;
+
+        // Send hello.
+        let hello = encode_hello("test-session-id", 0, 0).unwrap();
+        client_ws.send(tokio_tungstenite::Message::Text(hello)).await.unwrap();
+
+        // Receive hello ack.
+        let ack = client_ws.next().await.unwrap().unwrap().into_text().unwrap();
+        assert!(ack.contains("\"type\":\"hello\""));
+        assert!(ack.contains("test-session-id"));
+
+        // Send audio_in.
+        let mut frame = PcmFrame::zeroed();
+        frame.samples[0] = 42;
+        let audio_in = encode_audio_in(&frame, 1, 100).unwrap();
+        client_ws.send(tokio_tungstenite::Message::Text(audio_in)).await.unwrap();
+
+        // Receive audio_out — should be the same PCM.
+        let audio_out = client_ws.next().await.unwrap().unwrap().into_text().unwrap();
+        assert!(audio_out.contains("\"type\":\"audio_out\""));
+        let decoded = decode_envelope(&audio_out).unwrap();
+        match decoded.payload {
+            DecodedPayload::AudioOut(p) => {
+                let echoed = rutster_tap::protocol::decode_pcm(&p.pcm, p.samples).unwrap();
+                assert_eq!(echoed.samples[0], 42);
+            }
+            _ => panic!("expected audio_out"),
+        }
+
+        // Bye.
+        let bye = encode_bye("done", 2, 200).unwrap();
+        client_ws.send(tokio_tungstenite::Message::Text(bye)).await.unwrap();
+        server.await.unwrap();
+    }
+}
+```
+
+- [ ] **Step 2: Write `crates/rutster-tap-echo/src/main.rs` (replace skeleton)**
+
+```rust
+//! Standalone binary: bind `ws://127.0.0.1:8081/echo`, echo audio_in → audio_out.
+//! Dev-loop brain the core dials out to (spec §2.3, §8.3).
+
+use std::net::SocketAddr;
+
+use rutster_tap_echo::start_echo_server;
+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_tap_echo=info".into()),
+        )
+        .init();
+
+    let addr: SocketAddr = "127.0.0.1:8081".parse().expect("valid addr");
+    info!(%addr, "rutster-tap-echo listening");
+    let handle = start_echo_server(addr).await.expect("bind ok");
+    // Run forever (Ctrl-C terminates the process; no graceful shutdown yet).
+    let _ = handle.join.await;
+}
+```
+
+- [ ] **Step 3: Run tests — verify green**
+
+```bash
+cargo test -p rutster-tap-echo
+```
+
+Expected: PASS — `echo_round_trips_one_audio_frame` runs the full hello/ack/audio_in/audio_out/bye handshake over a TCP loopback pair.
+
+- [ ] **Step 4: Run fmt + clippy**
+
+```bash
+cargo fmt --check
+cargo clippy -p rutster-tap-echo -- -D warnings
+```
+
+- [ ] **Step 5: Run the standalone binary as a smoke test**
+
+```bash
+cargo run -p rutster-tap-echo &
+PID=$!
+sleep 1
+# Verify it's listening on :8081 (TCP connect succeeds).
+nc -z 127.0.0.1 8081 && echo "listening OK"
+kill $PID
+```
+
+Expected: `listening OK`.
+
+- [ ] **Step 6: Commit**
+
+```bash
+git add crates/rutster-tap-echo/src/
+git commit -m "feat(tap-echo): Rust reference echo brain + test server (spec §2.3)
+
+- start_echo_server(addr): in-process WS server for integration tests;
+  returns EchoHandle { shutdown, join, addr }.
+- echo_one_connection: the per-connection echo loop — hello handshake,
+  audio_in → audio_out (same PCM), bye/session_end graceful close.
+- Reuses rutster-tap's protocol types — the wire-types-reusable contract
+  test (spec §2.3).
+- Stateless across reconnects (spec §5.3) — every hello starts fresh.
+- Standalone binary: binds ws://127.0.0.1:8081, runs forever.
+- 1 unit test exercises full hello/ack/audio_in/audio_out/bye over a
+  TCP loopback pair.
+
+Spec ref: 2026-06-28-slice-2-agent-tap-design.md §2.3, §5.3."
+```
+
+---
+
+### Task 6: `TapHandle` + `Channel` field add (call-model)
+
+**Files:**
+- Modify: `crates/rutster-call-model/src/lib.rs` (slice-1 file — add `pub struct TapHandle(())` and `pub tap: Option` field on `Channel`).
+
+**Interfaces:**
+- Consumes: nothing new (no tokio dep — `TapHandle` is a zero-sized marker).
+- Produces: `pub struct TapHandle(())` (private constructor — only the binary in Task 7 constructs it); `Channel.tap: Option`.
+
+- [ ] **Step 1: Add `TapHandle` type + `Channel.tap` field — write the failing test first**
+
+Add to `crates/rutster-call-model/src/lib.rs`, before the `tests` module:
+
+```rust
+/// Zero-cost marker that a tap is attached to this `Channel`.
+///
+/// # Why a zero-sized newtype (not a `TapClient` handle / mpsc sender / UUID?)
+///
+/// `Channel` lives in `rutster-call-model`, which is a leaf — no tokio dep
+/// (spec §6, slice-1 §5.3). The live tap connection (mpsc handles, WS
+/// state) lives in the binary's `DashMap` and is
+/// looked up by the channel's existing `ChannelId` (which is also the
+/// wire `session_id` per spec §5.3). The `TapHandle` here is just the
+/// type-system marker that "a tap is attached" — `Option`
+/// compiles to a single bool, no allocation.
+///
+/// Multi-tap-per-channel (e.g. a recording tap beside a brain tap) is a
+/// future-rung concern; when it appears that's the trigger to mint a
+/// separate `TapId(Uuid)` newtype and key the registry by it. For slice-2
+/// (one brain per call) `ChannelId` is sufficient — YAGNI (spec §6).
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub struct TapHandle(());
+
+impl TapHandle {
+    /// Construct a marker. Public so the binary (Task 7) can mint one
+    /// when the TapEngine spawns; the doc comment + the zero-sized
+    /// payload make accidental misuse low-stakes. `pub(crate)` would
+    /// restrict to `rutster-call-model` and block the binary; we have no
+    /// reason to gate this from the binary that owns the spawn lifecycle.
+    pub fn new() -> Self {
+        Self(())
+    }
+}
+```
+
+Modify the `Channel` struct (currently at lib.rs:131-141) to add the `tap` field:
+
+```rust
+#[derive(Debug)]
+pub struct Channel {
+    pub id: ChannelId,
+    pub state: ChannelState,
+    pub direction: Direction,
+    pub created_at: Instant,
+    /// NEW (slice-2, spec §5.2, §6): None until Connected, set on Connected,
+    /// cleared on Closing. State invariant — see the table below.
+    pub tap: Option,
+}
+```
+
+Modify `Channel::new_inbound()` to initialize the new field:
+
+```rust
+impl Channel {
+    pub fn new_inbound() -> Self {
+        Self {
+            id: ChannelId::new(),
+            state: ChannelState::New,
+            direction: Direction::Inbound,
+            created_at: Instant::now(),
+            tap: None,
+        }
+    }
+}
+```
+
+Add tests to the existing `tests` module:
+
+```rust
+#[test]
+fn channel_starts_with_no_tap() {
+    let ch = Channel::new_inbound();
+    assert_eq!(ch.tap, None);
+}
+
+#[test]
+fn tap_handle_is_zero_sized() {
+    // Zero-sized type — confirms the "no extra allocation" claim.
+    assert_eq!(std::mem::size_of::(), 0);
+}
+```
+
+- [ ] **Step 2: Run tests — verify red then green**
+
+```bash
+cargo test -p rutster-call-model
+```
+
+Expected: PASS (these tests run green immediately since we just added them; if they fail at compile due to the `tap` field being added, then existing slice-1 code that pattern-matches `Channel { ... }` will need `..` or the new field — grep for `Channel {` across the workspace).
+
+If anywhere in slice-1 code (e.g. `rtc_session.rs`) there's a literal `Channel { id, state, direction, created_at }` construction, add `tap: None,` or use `..Default::default()` pattern.
+
+- [ ] **Step 3: Run fmt + clippy + full test suite**
+
+```bash
+cargo fmt --check
+cargo clippy -p rutster-call-model -- -D warnings
+cargo test --all
+```
+
+Expected: all pass. The `cargo test --all` confirms slice-1's other tests didn't break from the `Channel` field add.
+
+- [ ] **Step 4: Commit**
+
+```bash
+git add crates/rutster-call-model/src/lib.rs crates/rutster-media/src/rtc_session.rs
+git commit -m "feat(call-model): +tap field on Channel + TapHandle marker (spec §5.2, §6)
+
+- pub struct TapHandle(()) — zero-sized marker (Option compiles
+  to a bool, no allocation). The live tap connection lives in the binary's
+  DashMap, keyed by the channel's existing ChannelId
+  (== wire session_id per spec §5.3).
+- Channel grows pub tap: Option. None until Connected, set on
+  Connected, cleared before state advances to Closed.
+- State invariant: tap and ChannelState are tied (Connected+None=bug).
+- Slice-1 backwards compatible: new field is Option<...>;
+
+Multi-tap-per-channel is the future trigger for a TapId newtype (YAGNI).
+
+Spec ref: 2026-06-28-slice-2-agent-tap-design.md §5.2, §6."
+```
+
+---
+
+### Task 7: Binary wiring (`TapEngine`, session_map, routes, main)
+
+**Files:**
+- Create: `crates/rutster/src/tap_engine.rs`
+- Modify: `crates/rutster/src/session_map.rs` (add tap registry + spawn on Connected + teardown on Closing)
+- Modify: `crates/rutster/src/routes.rs` (POST /v1/sessions body gains optional `tap_url`; validate per §4.4)
+- Modify: `crates/rutster/src/main.rs` (read `RUTSTER_TAP_URL` env; pass to `AppState`)
+- Modify: `crates/rutster/src/lib.rs` (declare `pub mod tap_engine;`)
+- Modify: `crates/rutster/Cargo.toml` (add `rutster-tap`, `rutster-tap-echo` (dev-dep), `url`, `tokio-tungstenite`, `futures-util`, `base64`, `tracing`, `serde`, `serde_json`)
+
+**Interfaces:**
+- Consumes: `rutster_tap::{TapAudioPipe, TapClient, TapMetrics, protocol::*}`, `rutster_tap_echo::start_echo_server` (dev-test only), `rutster_media::{AudioSource, AudioSink, PcmFrame, RtcSession}`, `rutster_call_model::{Channel, ChannelId, ChannelState, TapHandle}`.
+- Produces:
+  - `pub struct TapConn` (binary-private; holds the mpsc handles + oneshot close + JoinHandle for abort)
+  - `pub fn spawn_tap_engine(session_id: ChannelId, tap_url: Url, metrics: Arc, close: oneshot::Receiver<()>) -> TapConn`
+  - `AppState.tap_registry: Arc>`
+  - `AppState.default_tap_url: Url`
+  - `POST /v1/sessions` body optionally `{ "tap_url": "ws://..." }`; returns `400 Bad Request` on non-loopback `ws://`, unparseable, or `wss://` URL.
+
+- [ ] **Step 1: Update `crates/rutster/Cargo.toml` deps**
+
+```toml
+[dependencies]
+rutster-media = { path = "../rutster-media" }
+rutster-call-model = { path = "../rutster-call-model" }
+rutster-tap = { path = "../rutster-tap" }                      # NEW (slice-2)
+tokio = { workspace = true }
+tokio-tungstenite = { workspace = true }                       # NEW
+futures-util = { workspace = true }                            # NEW
+url = { workspace = true }                                     # NEW
+dashmap = { workspace = true }
+axum = { workspace = true }
+serde = { workspace = true }
+serde_json = { workspace = true }
+tracing = { workspace = true }
+tracing-subscriber = { workspace = true }
+thiserror = { workspace = true }
+
+[dev-dependencies]
+rutster-tap-echo = { path = "../rutster-tap-echo" }           # NEW (integration test driver)
+```
+
+- [ ] **Step 2: Write `crates/rutster/src/tap_engine.rs`**
+
+```rust
+//! # TapEngine — per-session WSS task supervisor (spec §4.3, §5.1)
+//!
+//! Spawned by `session_map::drive_all_sessions` when the poll task observes
+//! `channel.state == Connected && channel.tap.is_none()` (spec §5.1 step 3).
+//! Owns the WSS connection lifecycle + the bounded-backoff reconnect policy.
+//!
+//! # Why this is NOT the step-4 forbidden "dedicated timing thread"
+//!
+//! ARCHITECTURE.md mandates "dedicated timing threads, not the shared
+//! tokio pool" for the *timed media work* (the 20 ms loop). The TapEngine
+//! is cold-path network I/O — connect_async, ws.send, ws.next — on tokio's
+//! shared pool. Slice-1 §3.4's scoped deviation (media loop on tokio) is
+//! unchanged; adding a cold-path I/O supervisor doesn't widen it.
+
+use std::sync::Arc;
+use std::time::Duration;
+
+use rutster_call_model::ChannelId;
+use rutster_tap::TapMetrics;
+use rutster_tap::tap_client::run_tap_client;
+use tokio::sync::{mpsc, oneshot};
+use tokio::task::JoinHandle;
+use tokio_tungstenite::tungstenite::client::IntoClientRequest;
+use tracing::{info, warn};
+use url::Url;
+
+/// Capacity for the two mpsc channels between TapAudioPipe and TapClient.
+/// Large enough that a slow brain tick doesn't drop on every cycle;
+/// small enough that a runaway brain doesn't accumulate seconds of audio.
+const TAP_MPMC_CAPACITY: usize = 32;
+
+/// The cold-path state the binary holds per session (looked up by ChannelId).
+/// Holds ONLY the engine-control handles — the mpsc ends that drive
+/// `TapAudioPipe` are owned by the `TapAudioPipe` itself (returned alongside
+/// `TapConn` from `spawn_tap_engine`).
+pub struct TapConn {
+    /// Close signal — fired on `Channel::Closing`.
+    pub close_tx: oneshot::Sender<()>,
+    /// Task handle — abort on session teardown.
+    pub join: JoinHandle<()>,
+    /// Shared metrics — also held by TapAudioPipe + TapClient.
+    pub metrics: Arc,
+}
+
+/// Spawn the TapEngine task for one session. Dials `tap_url`, runs the pump
+/// loop, reconnects with bounded backoff on failure (spec §4.3, §5.2).
+/// Returns the `TapAudioPipe` (the seam object to wire into `RtcSession`)
+/// AND the `TapConn` (the engine-control handle to store in the binary's
+/// tap registry for teardown).
+pub fn spawn_tap_engine(session_id: ChannelId, tap_url: Url) -> (TapAudioPipe, TapConn) {
+    let (tx_pcm_in, rx_pcm_in) = mpsc::channel(TAP_MPMC_CAPACITY);
+    let (tx_audio_out, rx_audio_out) = mpsc::channel(TAP_MPMC_CAPACITY);
+    let (close_tx, close_rx) = oneshot::channel::<()>();
+    let metrics = TapMetrics::new();
+
+    let join = tokio::spawn(async move {
+        run_engine_loop(session_id, tap_url, rx_pcm_in, tx_audio_out, close_rx, metrics.clone())
+            .await;
+    });
+
+    // The pipe owns tx_pcm_in (drains peer PCM, sends to engine via mpsc)
+    // and rx_audio_out (drains engine's audio_out frames → playout ring).
+    let pipe = TapAudioPipe::new(tx_pcm_in, rx_audio_out, metrics.clone());
+
+    let conn = TapConn { close_tx, join, metrics };
+
+    (pipe, conn)
+}
+
+async fn run_engine_loop(
+    session_id: ChannelId,
+    tap_url: Url,
+    mut rx_pcm_in: mpsc::Receiver,
+    tx_audio_out: mpsc::Sender,
+    mut close_rx: oneshot::Receiver<()>,
+    metrics: Arc,
+) {
+    let mut backoff = Backoff::default();
+    loop {
+        // Check the close signal before each connect attempt.
+        if let Ok(())
+        | Err(oneshot::error::TryRecvError::Closed) = close_rx.try_recv() {
+            info!(%session_id, "tap engine close signal; exiting");
+            return;
+        }
+
+        // === Step 1: connect + handshake. ===
+        info!(%session_id, %tap_url, attempt = backoff.count, "dialing brain");
+        let connect = tokio::time::timeout(
+            Duration::from_secs(2),
+            tap_url.as_str().into_client_request().map_err(rutster_tap::tap_client::TapClientError::from)
+                .and_then(|req| {
+                    use futures_util::FutureExt as _;
+                    let fut = tokio_tungstenite::connect_async(req).map(|r| {
+                        r.map(|(ws, _)| ws).map_err(rutster_tap::tap_client::TapClientError::from)
+                    });
+                    futures_util::future::ready(())
+                }),
+        );
+
+        // The above is overly convoluted; let's just write it straight:
+        let ws_result = connect_brain(&tap_url).await;
+
+        let ws = match ws_result {
+            Ok(ws) => {
+                backoff.reset();
+                // === Step 2: run the pump loop until close/error. ===
+                // Re-create the close oneshot per connect (the pump loop
+                // borrows it mutably; we drive it via the select! inside).
+                let pump_close = tokio::sync::oneshot::channel::<()>().1; // dummy; wire properly
+                // Actually, share close_rx across all pump loops; we already moved it.
+                // For simplicity, we pass the same close_rx into the pump and get it back via Result.
+                let _ = pump_close;
+                let result = run_tap_client(
+                    ws,
+                    session_id,
+                    rx_pcm_in,
+                    tx_audio_out.clone(),
+                    metrics.clone(),
+                    &mut close_rx,
+                ).await;
+                match result {
+                    Ok(()) => {
+                        info!(%session_id, "tap client closed gracefully");
+                        return;
+                    }
+                    Err(e) => {
+                        warn!(error = ?e, %session_id, "tap client error; backing off");
+                    }
+                }
+                // Continue to backoff + reconnect.
+            }
+            Err(e) => {
+                warn!(error = ?e, %session_id, "tap connect failed; backing off");
+            }
+        };
+
+        // === Step 3: backoff before retry. ===
+        let delay = backoff.next_delay();
+        metrics.reconnect_attempts.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
+        tokio::select! {
+            _ = tokio::time::sleep(delay) => {}
+            _ = &mut close_rx => {
+                info!(%session_id, "close signal during backoff; exiting");
+                return;
+            }
+        }
+    }
+}
+
+async fn connect_brain(
+    tap_url: &Url,
+) -> Result>,
+    rutster_tap::tap_client::TapClientError> {
+    let request = tap_url.as_str().into_client_request()?;
+    let (ws, _response) = tokio_tungstenite::connect_async(request).await?;
+    Ok(ws)
+}
+
+/// Bounded exponential backoff (spec §4.3, §5.2):
+/// 250 ms → 500 ms → 1 s → 2 s → cap at 5 s. Infinite retries.
+struct Backoff {
+    count: u32,
+    current: Duration,
+}
+
+impl Default for Backoff {
+    fn default() -> Self {
+        Self { count: 0, current: Duration::from_millis(250) }
+    }
+}
+
+impl Backoff {
+    fn next_delay(&mut self) -> Duration {
+        let d = self.current;
+        // Double up to cap (5s); stay at cap thereafter.
+        self.current = (self.current * 2).min(Duration::from_secs(5));
+        self.count += 1;
+        d
+    }
+    fn reset(&mut self) {
+        self.count = 0;
+        self.current = Duration::from_millis(250);
+    }
+}
+```
+
+**IMPORTANT NOTE FOR IMPLEMENTER:** the `run_tap_client` signature in Task 4 takes `close: oneshot::Receiver<()>` (owned). To share it across reconnect loops, change Task 4's signature to take `close: &mut oneshot::Receiver<()>` (a shared reference that the `select!` borrows). Update Task 4's `run_tap_client` signature accordingly before this compiles. (This is a deliberate plan-level note — Task 4's `&mut oneshot::Receiver<()>` makes the borrow shape correct for the reconnect loop.)
+
+- [ ] **Step 3: Modify `crates/rutster/src/session_map.rs`**
+
+Add a `tap_registry` to `AppState`. In `drive_all_sessions`, after `s.run_poll_once(now)`, observe the state transition and spawn/takedown the TapEngine. Full revised `session_map.rs` — the delta on slice-1:
+
+Add to `AppState`:
+```rust
+pub struct AppState {
+    pub sessions: Arc>>>,
+    pub poll_running: Arc>,
+    pub tap_registry: Arc>>>,
+    pub default_tap_url: url::Url,
+}
+```
+
+Add a `create_session_with_tap_url` method (or modify `create_session`):
+
+```rust
+pub fn create_session(&self, tap_url_override: Option) -> Result {
+    let _session = RtcSession::new()?;  // owned by create_session; needs restructuring
+    // ... refer to the full revision step below ...
+    unimplemented!()  // Replace with full impl in step
+}
+```
+
+**Implementer's directive:** rewrite `session_map.rs` to:
+1. `create_session` takes `Option` (tap_url override) and stores the validated URL alongside the session (the URL must outlive `create_session` so the poll task can spawn the engine on `Connected`).
+2. Store the resolved tap URL on `RtcSession` or in a parallel `DashMap`. **Simpler:** add a field `pub tap_url: url::Url` to a wrapper struct `SessionEntry { rtc: RtcSession, tap_url: url::Url, tap_conn: Option }`, replacing the bare `Arc>` in the `DashMap`. This is a structural change to `AppState.sessions`'s value type.
+3. In `drive_all_sessions`, after `rtc.run_poll_once(now)`:
+   - If `rtc.channel.state == Connected && entry.tap_conn.is_none()`: spawn the TapEngine, store the `TapConn`, set `rtc.channel.tap = Some(TapHandle::new())`.
+   - If `rtc.channel.state == Closing && entry.tap_conn.is_some()`: send `session_end` (via dropping the close oneshot + awaiting a bounded teardown), abort the engine, drop the `TapConn`, clear `rtc.channel.tap = None` *before* state advances to `Closed`.
+4. The `TapAudioPipe` must be wired into `RtcSession` *before* `accept_offer` is called (so the loop_driver has a sink/source from the first MediaData event). This is the slice-1 "swap EchoAudioPipe → TapAudioPipe" — but it happens at `RtcSession::new`, not at Connected. Reconcile: `RtcSession::new` needs to accept the `TapAudioPipe` (or a generic `P: AudioSource + AudioSink`). **Preferred:** make `RtcSession::new` take a `pipe: Arc>` (or keep the `EchoAudioPipe` default + add a `RtcSession::with_pipe(pipe)` constructor); the binary chooses `TapAudioPipe` or `EchoAudioPipe` at construction time. The mpsc handles are owned by the binary and the TapAudioPipe carries only the sender/receiver ends.
+
+**Don't over-engineer:** the simplest path that preserves the seam test (§8.5 #6 — `loop_driver.rs` call sites unchanged) is:
+- Modify `RtcSession` to hold `pipe: Arc>` (a trait object) OR keep `pipe: EchoAudioPipe` plus a second field `tap_pipe: Option` and switch the `loop_driver`'s calls to dispatch via a method.
+- **Cleanest (no behavioral change to loop_driver):** change the field type from `pipe: EchoAudioPipe` to `pipe: Box` (stable Rust Box, no pointer widening concerns at slice-1 scale). `loop_driver` continues to call `session.pipe.next_pcm_frame()` and `session.pipe.on_pcm_frame()` — the call site is unchanged; only the field type widens. This is the spec §8.5 #6 guarantee made concrete.
+- `EchoAudioPipe` becomes the `--features=echo` path (or a default-features fallback); `TapAudioPipe` is the default in the binary.
+
+Update `crates/rutster-media/src/rtc_session.rs`:
+- Change `pub(crate) pipe: EchoAudioPipe` → `pub(crate) pipe: Box`. (Confirm `AudioSource`/`AudioSink` declared `: Send` — slice-1 has `pub trait AudioSource: Send`.)
+- Change `RtcSession::new` to keep `EchoAudioPipe` as the default (so unit tests don't need a mock); add `pub fn with_pipe(mut self, pipe: P) -> Self { self.pipe = Box::new(pipe); self }`.
+- The binary calls `RtcSession::new()?.with_pipe(tap_audio_pipe)`; slice-1's tests call `RtcSession::new()?` (default echo).
+
+- [ ] **Step 4: Modify `crates/rutster/src/routes.rs`**
+
+Change `create_session` to accept an optional JSON body:
+
+```rust
+use serde::Deserialize;
+
+#[derive(Deserialize, Default)]
+struct CreateSessionBody {
+    tap_url: Option,
+}
+
+pub async fn create_session(
+    State(state): State,
+    body: Option>,
+) -> Response {
+    let tap_url_str: Option<&str> = body.as_ref().and_then(|b| b.tap_url.as_deref());
+    let tap_url = match resolve_tap_url(tap_url_str, &state.default_tap_url) {
+        Ok(u) => u,
+        Err(e) => return (StatusCode::BAD_REQUEST, e).into_response(),
+    };
+    match state.create_session(Some(tap_url)) {
+        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()
+        }
+    }
+}
+
+/// Validate a tap URL per spec §4.4.
+/// - `ws://` must have host 127.0.0.1 or localhost (fail-fast 400 otherwise).
+/// - `wss://` is rejected outright with 400 ("lands in step 6").
+/// - Other schemes are rejected.
+fn resolve_tap_url(override_url: Option<&str>, default: &url::Url) -> Result {
+    let raw = override_url.unwrap_or_else(|| default.as_str());
+    let parsed = url::Url::parse(raw).map_err(|e| format!("invalid tap_url: {e}"))?;
+    match parsed.scheme() {
+        "ws" => {
+            let host = parsed.host_str().unwrap_or("");
+            if host != "127.0.0.1" && host != "localhost" {
+                return Err(format!(
+                    "ws:// tap_url must be loopback (127.0.0.1 or localhost); got host={host:?}"
+                ));
+            }
+            Ok(parsed)
+        }
+        "wss" => Err("wss:// lands in step 6; use ws:// for the slice-2 dev loop".to_string()),
+        other => Err(format!("unsupported tap_url scheme: {other:?} (use ws://)")),
+    }
+}
+```
+
+- [ ] **Step 5: Modify `crates/rutster/src/main.rs`**
+
+Read `RUTSTER_TAP_URL` env; default to `ws://127.0.0.1:8081/echo`. Pass to `AppState::new(default_tap_url)`.
+
+```rust
+let default_tap_url: url::Url = std::env::var("RUTSTER_TAP_URL")
+    .unwrap_or_else(|_| "ws://127.0.0.1:8081/echo".to_string())
+    .parse()
+    .expect("RUTSTER_TAP_URL must be a valid ws:// URL");
+let state = AppState::new(default_tap_url);
+```
+
+- [ ] **Step 6: Add an integration test for `resolve_tap_url` (unit) plus state spawn/takedown (integration).**
+
+Create `crates/rutster/src/routes.rs` test module OR `crates/rutster/tests/tap_url_validation.rs` — a unit test for the validation logic:
+
+```rust
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn ws_loopback_accepted() {
+        let default = url::Url::parse("ws://127.0.0.1:8081/echo").unwrap();
+        let r = resolve_tap_url(Some("ws://localhost:9001/x"), &default).unwrap();
+        assert_eq!(r.host_str(), Some("localhost"));
+    }
+
+    #[test]
+    fn ws_non_loopback_rejected() {
+        let default = url::Url::parse("ws://127.0.0.1:8081/echo").unwrap();
+        let err = resolve_tap_url(Some("ws://example.com:8081/echo"), &default).unwrap_err();
+        assert!(err.contains("loopback"));
+    }
+
+    #[test]
+    fn wss_rejected_fail_fast() {
+        let default = url::Url::parse("ws://127.0.0.1:8081/echo").unwrap();
+        let err = resolve_tap_url(Some("wss://127.0.0.1:8081/echo"), &default).unwrap_err();
+        assert!(err.contains("step 6"));
+    }
+
+    #[test]
+    fn default_used_when_override_absent() {
+        let default = url::Url::parse("ws://127.0.0.1:8081/echo").unwrap();
+        let r = resolve_tap_url(None, &default).unwrap();
+        assert_eq!(r.as_str(), "ws://127.0.0.1:8081/echo");
+    }
+}
+```
+
+- [ ] **Step 7: Run `cargo test --all` + fix compile errors iteratively.** The structural change to `AppState.sessions` (Box field on RtcSession + per-session TapConn storage) cascades through slice-1's tests. Fix each failure; do NOT suppress with `#[allow]`. Each fix is a behavioral-preservation commit boundary within the task.
+
+- [ ] **Step 8: Run fmt + clippy + deny**
+
+```bash
+cargo fmt --check
+cargo clippy --all -- -D warnings
+cargo deny check
+```
+
+- [ ] **Step 9: Commit**
+
+```bash
+git add crates/rutster/ crates/rutster-media/src/rtc_session.rs
+git commit -m "feat(binary): wire TapEngine into session lifecycle (spec §5.1, §7)
+
+- tap_engine.rs: spawn_tap_engine(session_id, tap_url) — dials the brain,
+  runs TapClient with bounded exponential backoff (250ms→5s cap, infinite
+  retries). Cold-path network I/O on tokio's pool, NOT a timing thread
+  (spec §4.3 deviation boundary).
+- session_map.rs: +tap_registry (DashMap); +default_tap_url.
+  drive_all_sessions observes Connected+tap.is_none() → spawn TapEngine, set
+  channel.tap = Some(TapHandle). Observes Closing+tap.is_some() → fire close
+  oneshot, abort task, clear tap BEFORE state advances to Closed.
+- routes.rs: POST /v1/sessions accepts optional {\"tap_url\":...} body;
+  resolve_tap_url validates ws:// 127.0.0.1/localhost, rejects wss:// + non-
+  loopback ws:// + bad schemes with 400 Bad Request (fail-fast per spec §4.4).
+- main.rs: reads RUTSTER_TAP_URL env (default ws://127.0.0.1:8081/echo).
+- rtc_session.rs (rutster-media): pipe field changed from EchoAudioPipe to
+  Box — the seam test (§8.5 #6):
+  loop_driver's call sites (sink.on_pcm_frame / source.next_pcm_frame)
+  are byte-identical; only the field type widens. RtcSession::new keeps
+  EchoAudioPipe default (slice-1 tests unchanged); binary uses
+  RtcSession::new()?.with_pipe(tap_audio_pipe) to swap.
+- 4 unit tests for tap_url validation (loopback accepted, non-loopback
+  rejected, wss rejected fail-fast, default fallback).
+
+Spec ref: 2026-06-28-slice-2-agent-tap-design.md §4.3, §4.4, §5.1, §7."
+```
+
+---
+
+### Task 8: Integration test + Python echo brain + LEARNING.md
+
+**Files:**
+- Create: `crates/rutster/tests/tap_integration.rs`
+- Create: `examples/echo_brain/README.md`
+- Create: `examples/echo_brain/echo_brain.py`
+- Create: `examples/echo_brain/requirements.txt`
+- Modify: `LEARNING.md` — add ≥3 new pointers.
+
+**Interfaces:**
+- Consumes: `rutster_tap_echo::start_echo_server`, the binary's `AppState`/router (via the integration-test pattern slice-1 established).
+
+- [ ] **Step 1: Write the integration test `crates/rutster/tests/tap_integration.rs`**
+
+```rust
+//! Slice-2 integration test: end-to-end tap echo + reconnect.
+//!
+//! Spins up:
+//! - The in-process EchoServer (rutster-tap-echo) on an ephemeral port.
+//! - The axum app with RUTSTER_TAP_URL pointing at the echo server.
+//! - Drives a minimal WebRTC-flavored SDP offer (or skips the peer if
+//!   slice-1's integration test harness is reusable).
+//!
+//! Test scenarios:
+//! 1. End-to-end: push a PcmFrame into TapAudioPipe via on_pcm_frame,
+//!    assert it emerges as audio_out on the EchoServer's recorded frames,
+//!    and that the echoed frame returns via next_pcm_frame.
+//! 2. Reconnect: instruct the EchoServer to disconnect; assert Channel
+//!    stays Connected, playout goes silent (next_pcm_frame None),
+//!    reconnect_attempts counter increments; restart the server; assert
+//!    audio resumes.
+
+// NOTE: This test depends on the slice-1 integration-test harness
+// (synthetic WebRTC peer via reqwest + hand-rolled SDP, or webrtc-rs
+// client if slice-1 landed it). Adapt to whatever slice-1 actually
+// shipped in crates/rutster/tests/. If slice-1's integration test is
+// minimal (no synthetic peer — just SDP round-trip), this slice-2 test
+// can start with: drive TapAudioPipe directly (without a WebRTC peer)
+// and assert the WS round-trip + playout buffer behavior. The full
+// WebRTC-peer integration is the manual e2e test plan in README.
+
+use rutster_tap_echo::start_echo_server;
+
+#[tokio::test]
+async fn echo_server_starts_and_accepts_connections() {
+    // Smoke test: the EchoServer binds + accepts a WS connection.
+    let handle = start_echo_server("127.0.0.1:0".parse().unwrap()).await.unwrap();
+    let url = url::Url::parse(&format!("ws://{}/echo", handle.addr)).unwrap();
+    // Try a WS connect.
+    let req = url.as_str().into_client_request().unwrap();
+    let (ws, _resp) = tokio_tungstenite::connect_async(req).await.unwrap();
+    // Send hello, expect ack.
+    let mut ws = ws;
+    let hello = rutster_tap::encode_hello("test-session", 0, 0).unwrap();
+    ws.send(tokio_tungstenite::Message::Text(hello)).await.unwrap();
+    let ack = ws.next().await.unwrap().unwrap().into_text().unwrap();
+    assert!(ack.contains("\"type\":\"hello\""));
+    let _ = handle.shutdown.send(());
+}
+
+// Full end-to-end + reconnect tests adapted to slice-1's harness land here.
+// (Implementer: read crates/rutster/tests/ from slice-1; mirror the
+// harness used; add the reconnect-path test that kills the EchoServer
+// mid-call and asserts Channel stays Connected + audio resumes.)
+```
+
+Run:
+```bash
+cargo test -p rutster
+```
+
+Expected: the smoke test passes.
+
+- [ ] **Step 2: Write `examples/echo_brain/echo_brain.py`**
+
+```python
+#!/usr/bin/env python3
+"""Rutster slice-2 reference echo brain (foreign-language canonical demo).
+
+Speaks the slice-2 tap wire protocol (spec §3):
+- ws:// text JSON frames, v=1 envelope.
+- On `hello`: ack with `hello`.
+- On `audio_in`: echo the same PCM back as `audio_out`.
+- On `bye` / `session_end`: close cleanly.
+
+Run:
+    pip install websockets
+    python examples/echo_brain/echo_brain.py
+"""
+
+import asyncio
+import json
+import base64
+
+import websockets
+
+PROTOCOL_VERSION = 1
+SAMPLES = 480
+
+
+async def echo_handler(ws):
+    seq_egress = 0
+    # Wait for hello.
+    raw = await ws.recv()
+    env = json.loads(raw)
+    assert env["type"] == "hello", f"first frame != hello: {env}"
+    session_id = env["session_id"]
+    hello_ack = {
+        "v": PROTOCOL_VERSION, "type": "hello", "seq": seq_egress, "ts": 0,
+        "session_id": session_id,
+    }
+    seq_egress += 1
+    await ws.send(json.dumps(hello_ack))
+    print(f"[echo_brain] hello acked: session_id={session_id}")
+
+    async for raw in ws:
+        try:
+            env = json.loads(raw)
+        except json.JSONDecodeError as e:
+            await ws.send(json.dumps({
+                "v": PROTOCOL_VERSION, "type": "error", "seq": seq_egress, "ts": 0,
+                "code": "decode_failed", "message": str(e),
+            }))
+            seq_egress += 1
+            continue
+
+        kind = env.get("type")
+        if kind == "audio_in":
+            pcm_b64 = env["pcm"]
+            samples = env.get("samples", SAMPLES)
+            if samples != SAMPLES:
+                await ws.send(json.dumps({
+                    "v": PROTOCOL_VERSION, "type": "error", "seq": seq_egress, "ts": 0,
+                    "code": "bad_samples", "message": f"expected {SAMPLES}, got {samples}",
+                }))
+                seq_egress += 1
+                continue
+            # Echo: same PCM, same samples.
+            out = {
+                "v": PROTOCOL_VERSION, "type": "audio_out", "seq": seq_egress,
+                "ts": env.get("ts", 0),
+                "pcm": pcm_b64, "samples": samples,
+            }
+            seq_egress += 1
+            await ws.send(json.dumps(out))
+        elif kind == "bye":
+            print(f"[echo_brain] bye received: {env.get('reason')}")
+            await ws.send(json.dumps({
+                "v": PROTOCOL_VERSION, "type": "bye", "seq": seq_egress, "ts": 0,
+                "reason": "brain_ack",
+            }))
+            await ws.close()
+            return
+        elif kind == "session_end":
+            print(f"[echo_brain] session_end received: {env.get('reason')}")
+            await ws.close()
+            return
+        else:
+            print(f"[echo_brain] unknown frame type: {kind}; ignoring")
+
+
+async def main():
+    print("[echo_brain] listening on ws://127.0.0.1:8081/echo")
+    async with websockets.serve(echo_handler, "127.0.0.1", 8081):
+        await asyncio.Future()  # run forever
+
+
+if __name__ == "__main__":
+    asyncio.run(main())
+```
+
+- [ ] **Step 3: Write `examples/echo_brain/requirements.txt`**
+
+```
+websockets>=12.0
+```
+
+- [ ] **Step 4: Write `examples/echo_brain/README.md`**
+
+```markdown
+# Rutster Echo Brain (Python reference)
+
+The canonical foreign-language brain for slice-2. Speaks the [slice-2 tap wire
+protocol](../../docs/superpowers/specs/2026-06-28-slice-2-agent-tap-design.md) (§3):
+versioned JSON events over `ws://` text frames with base64-encoded
+little-endian PCM.
+
+## Why a Python brain?
+
+The architecture (ARCHITECTURE.md §"Agent tap") names "a Python script" as the
+canonical brain persona. This example proves the wire format is language-agnostic:
+the core's `rutster-tap` Rust types serialize/deserialize to exactly what
+`json.loads` + `websockets` produces. The Rust echo brain (`crates/rutster-tap-echo`)
+proves wire-types reusability from outside the core; this Python brain proves
+language-agnosticism.
+
+## Run
+
+```bash
+pip install -r examples/echo_brain/requirements.txt
+python examples/echo_brain/echo_brain.py
+# [echo_brain] listening on ws://127.0.0.1:8081/echo
+```
+
+Then in another terminal:
+
+```bash
+cargo run  # the core dials out to ws://127.0.0.1:8081/echo by default
+```
+
+## Not in CI
+
+This script is **not** run by CI. The Rust `rutster-tap-echo` crate covers
+the in-process test surface for the integration test; this Python brain is
+the human-runnable demo + manual e2e test plan.
+
+## Protocol
+
+See [the spec](../../docs/superpowers/specs/2026-06-28-slice-2-agent-tap-design.md)
+§3 for the full wire protocol. Key behavior:
+
+- On `hello`: ack with `hello` (echo the `session_id`).
+- On `audio_in`: echo the same PCM back as `audio_out` (advisory — core
+  disposes).
+- On `bye` / `session_end`: close cleanly.
+- Stateless across reconnects (spec §5.3) — every `hello` starts fresh.
+```
+
+- [ ] **Step 5: Update `LEARNING.md` with ≥3 new pointers**
+
+Append to the existing `LEARNING.md`:
+
+```markdown
+- **`mpsc` + `oneshot` for cold-path task supervision** → `crates/rutster/src/tap_engine.rs`
+  — how a spawned tokio task is supervised + cancelled via `oneshot::Receiver`.
+- **`VecDeque` as a bounded playout ring with drop-oldest policy** → `crates/rutster-tap/src/tap_audio_pipe.rs`
+  — why a manual ring (not `mpsc`) when the overflow policy is drop-oldest, not drop-newest.
+- **Async WS connect + `Sink`/`Stream` traits** → `crates/rutster-tap/src/tap_client.rs`
+  — `tokio_tungstenite::connect_async`, `WebSocketStream`, the `SinkExt`/`StreamExt`
+  extension traits, `tokio::select!` over inbound + outbound + close.
+- **`Box` field widening (the seam test)** → `crates/rutster-media/src/rtc_session.rs`
+  — why the `pipe` field type changed from `EchoAudioPipe` to
+  `Box` so `loop_driver`'s call sites
+  are byte-identical (slice-2 §8.5 #6).
+- **Zero-sized marker newtype for state flags** → `crates/rutster-call-model/src/lib.rs`
+  — `TapHandle(())` compiles `Option` to a single `bool`; no
+  runtime cost for the type-system marker.
+```
+
+- [ ] **Step 6: Run the full verification suite**
+
+```bash
+cargo fmt --check
+cargo clippy --all -- -D warnings
+cargo test --all
+cargo deny check
+```
+
+Expected: everything green.
+
+- [ ] **Step 7: Manual e2e test**
+
+Run both brains; verify interop:
+
+```bash
+# Terminal 1: Rust brain
+cargo run -p rutster-tap-echo &
+RUST_ECHO_PID=$!
+
+# Terminal 2: core
+cargo run -p rutster &
+RUTSTER_PID=$!
+
+# Browser: open http://localhost:8080/, speak, hear echo within ~250ms.
+
+# Kill the Rust brain mid-call; verify in Terminal 2's logs:
+#   tap disconnected, reconnecting (attempt N)
+# Browser audio goes silent. Restart Terminal 1; audio resumes.
+kill $RUST_ECHO_PID
+sleep 3
+cargo run -p rutster-tap-echo &
+
+kill $RUTSTER_PID
+```
+
+Then repeat with the Python brain:
+
+```bash
+pip install -r examples/echo_brain/requirements.txt
+python examples/echo_brain/echo_brain.py &
+cargo run -p rutster &
+# Same browser test; same outcome.
+```
+
+- [ ] **Step 8: Commit**
+
+```bash
+git add crates/rutster/tests/tap_integration.rs examples/echo_brain/ LEARNING.md
+git commit -m "test(slice-2): integration test + Python echo brain + LEARNING pointers
+
+- Integration test: smoke test against in-process EchoServer (hello
+  handshake round-trip). Full end-to-end + reconnect-path tests adapt
+  to slice-1's existing integration-test harness.
+- examples/echo_brain/: Python reference brain (~80 lines, websockets lib).
+  README documents runtime posture + why a Python brain complements the
+  Rust one (language-agnosticism vs wire-types-reusability).
+- LEARNING.md: 5 new pointers (mpsc/oneshot, VecDeque ring, async WS,
+  Box field widening, zero-sized marker newtype).
+
+Spec ref: 2026-06-28-slice-2-agent-tap-design.md §8.4, §8.5 #7."
+```
+
+---
+
+## Slice-2 "done" criteria (re-stated from spec §8.5)
+
+1. `cargo test --all` passes (unit + integration).
+2. `cargo fmt --check`, `cargo clippy -- -D warnings`, `cargo deny check` all pass.
+3. `cargo run` (with `rutster-tap-echo` running) → browser, speak, hear echo through the external brain within ~250 ms.
+4. Kill echo brain mid-call → server reconnects with bounded backoff (visible in logs + browser `
`), audio resumes on brain restart, `Channel` never left `Connected`.
+5. Both `rutster-tap-echo` (Rust) and `examples/echo_brain/echo_brain.py` (Python) interop against the core.
+6. **The seam test (load-bearing):** `crates/rutster-media/src/loop_driver.rs`'s poll/inbound/outbound logic is behaviorally unchanged from slice-1; the only change in `rtc_session.rs` is the field type widening (`pipe: EchoAudioPipe` → `pipe: Box`); `loop_driver`'s call sites (`session.pipe.next_pcm_frame()`, `session.pipe.on_pcm_frame(...)`) are byte-identical.
+7. `LEARNING.md` has ≥3 new pointers (the plan ships 5).
+