# 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-trunk`, `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": "<event_name>", "seq": <uint>, "ts": <uint>, ...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<TapHandle>` 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<ChannelId, TapConn>`. **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<ChannelId, TapConn>); +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 <pre>
│   ├── 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<TapHandle>
│   ├── 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-trunk/           # 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-trunk",
    "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<String, TapProtoError>`
  - `pub fn encode_audio_out(frame: &PcmFrame, seq: u64, ts: u64) -> Result<String, TapProtoError>`
  - `pub fn encode_hello(session_id: &str, seq: u64, ts: u64) -> Result<String, TapProtoError>`
  - `pub fn encode_session_end(reason: &str, seq: u64, ts: u64) -> Result<String, TapProtoError>`
  - `pub fn encode_bye(reason: &str, seq: u64, ts: u64) -> Result<String, TapProtoError>`
  - `pub fn encode_error(code: &str, msg: &str, seq: u64, ts: u64) -> Result<String, TapProtoError>`
  - `pub fn decode_envelope(s: &str) -> Result<DecodedFrame, TapProtoError>`
  - `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<u8> = 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<PcmFrame, TapProtoError> {
    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<String, TapProtoError> {
    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<String, TapProtoError> {
    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<String, TapProtoError> {
    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<String, TapProtoError> {
    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<String, TapProtoError> {
    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<String, TapProtoError> {
    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<DecodedFrame, TapProtoError> {
    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<PcmFrame>, rx_audio_out: mpsc::Receiver<PcmFrame>, metrics: Arc<TapMetrics> }`
  - `pub fn new_tap_audio_pipe(tx_pcm_in: mpsc::Sender<PcmFrame>, rx_audio_out: mpsc::Receiver<PcmFrame>, metrics: Arc<TapMetrics>) -> TapAudioPipe`
  - `impl AudioSource for TapAudioPipe { fn next_pcm_frame(&mut self) -> Option<PcmFrame> }`
  - `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<Self> {
        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<PcmFrame>,
    // Brain → core (playout buffer):
    rx_audio_out: mpsc::Receiver<PcmFrame>,
    // Bounded ring between rx_audio_out and next_pcm_frame (spec §4.1).
    playout_ring: VecDeque<PcmFrame>,
    metrics: Arc<TapMetrics>,
}

impl TapAudioPipe {
    pub fn new(
        tx_pcm_in: mpsc::Sender<PcmFrame>,
        rx_audio_out: mpsc::Receiver<PcmFrame>,
        metrics: Arc<TapMetrics>,
    ) -> 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<PcmFrame> {
        // 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<PcmFrame>, mpsc::Receiver<PcmFrame>,
                      mpsc::Sender<PcmFrame>, mpsc::Receiver<PcmFrame>,
                      Arc<TapMetrics>) {
        // 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<PcmFrame>, tx_audio_out: mpsc::Sender<PcmFrame>, metrics: Arc<TapMetrics>, 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<T>(
    mut ws: WebSocketStream<T>,
    session_id: ChannelId,
    mut rx_pcm_in: mpsc::Receiver<PcmFrame>,
    tx_audio_out: mpsc::Sender<PcmFrame>,
    metrics: Arc<TapMetrics>,
    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<u64> = 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<T>(
    ws: &mut WebSocketStream<T>,
) -> Result<u64, TapClientError>
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<u64>,
    tx_audio_out: &mpsc::Sender<PcmFrame>,
    metrics: &Arc<TapMetrics>,
    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<EchoHandle, std::io::Error>` — binds the WS server, returns a handle to stop it.
  - `pub async fn echo_one_connection(ws: WebSocketStream<T>, ...) -> 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<EchoHandle, std::io::Error> {
    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<T>(
    mut ws: WebSocketStream<T>,
) -> Result<(), Box<dyn std::error::Error>>
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<TapHandle>` 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<TapHandle>`.

- [ ] **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<ChannelId, TapConn>` 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<TapHandle>`
/// 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<TapHandle>,
}
```

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::<TapHandle>(), 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<TapHandle> compiles
  to a bool, no allocation). The live tap connection lives in the binary's
  DashMap<ChannelId, TapConn>, keyed by the channel's existing ChannelId
  (== wire session_id per spec §5.3).
- Channel grows pub tap: Option<TapHandle>. 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<TapMetrics>, close: oneshot::Receiver<()>) -> TapConn`
  - `AppState.tap_registry: Arc<DashMap<ChannelId, TapConn>>`
  - `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<TapMetrics>,
}

/// 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<rutster_media::PcmFrame>,
    tx_audio_out: mpsc::Sender<rutster_media::PcmFrame>,
    mut close_rx: oneshot::Receiver<()>,
    metrics: Arc<TapMetrics>,
) {
    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<tokio_tungstenite::WebSocketStream<tokio_tungstenite::MaybeTlsStream<tokio::net::TcpStream>>,
    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<DashMap<ChannelId, Arc<Mutex<RtcSession>>>>,
    pub poll_running: Arc<Mutex<bool>>,
    pub tap_registry: Arc<DashMap<ChannelId, Arc<Mutex<crate::tap_engine::TapConn>>>>,
    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<url::Url>) -> Result<ChannelId, RtcSessionError> {
    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<Url>` (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<ChannelId, Url>`. **Simpler:** add a field `pub tap_url: url::Url` to a wrapper struct `SessionEntry { rtc: RtcSession, tap_url: url::Url, tap_conn: Option<TapConn> }`, replacing the bare `Arc<Mutex<RtcSession>>` 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<Mutex<...>>` (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<Mutex<dyn AudioSource + AudioSink>>` (a trait object) OR keep `pipe: EchoAudioPipe` plus a second field `tap_pipe: Option<TapAudioPipe>` 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<dyn AudioSource + AudioSink>` (stable Rust Box<dyn Trait>, 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<dyn AudioSource + AudioSink + Send>`. (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<P: AudioSource + AudioSink + Send + 'static>(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<String>,
}

pub async fn create_session(
    State(state): State<AppState>,
    body: Option<axum::Json<CreateSessionBody>>,
) -> 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<url::Url, String> {
    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<dyn> 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<ChannelId, TapConn>); +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<dyn AudioSource + AudioSink + Send> — 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<dyn Trait + Send>` field widening (the seam test)** → `crates/rutster-media/src/rtc_session.rs`
  — why the `pipe` field type changed from `EchoAudioPipe` to
  `Box<dyn AudioSource + AudioSink + Send>` 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<TapHandle>` 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<dyn Trait> 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 `<pre>`), 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<dyn AudioSource + AudioSink + Send>`); `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).