rutster/docs/superpowers/plans/2026-06-28-slice-2-agent-tap.md

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

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

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:

[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

[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

//! # 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

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

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

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

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

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

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

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

cargo test -p rutster-tap

Expected: PASS — all 10 tests in protocol::tests + crate_compiles.

• Step 5: Run fmt + clippy + deny

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

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

//! # 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

//! # 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

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:

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

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:

#[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.)

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

cargo test -p rutster-tap

Expected: PASS.

• Step 6: Run fmt + clippy

cargo fmt --check
cargo clippy -p rutster-tap -- -D warnings

• Step 7: Commit

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

//! # 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

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

cargo fmt --check
cargo clippy -p rutster-tap -- -D warnings

• Step 4: Commit

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

• Step 1: Write crates/rutster-tap-echo/src/lib.rs (replace skeleton)

//! # 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)

//! 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

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

cargo fmt --check
cargo clippy -p rutster-tap-echo -- -D warnings

• Step 5: Run the standalone binary as a smoke test

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

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:

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

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

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:

#[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

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

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

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

[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

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

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

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

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

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:

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn ws_loopback_accepted() {
        let default = url::Url::parse("ws://127.0.0.1:8081/echo").unwrap();
        let r = resolve_tap_url(Some("ws://localhost:9001/x"), &default).unwrap();
        assert_eq!(r.host_str(), Some("localhost"));
    }

    #[test]
    fn ws_non_loopback_rejected() {
        let default = url::Url::parse("ws://127.0.0.1:8081/echo").unwrap();
        let err = resolve_tap_url(Some("ws://example.com:8081/echo"), &default).unwrap_err();
        assert!(err.contains("loopback"));
    }

    #[test]
    fn wss_rejected_fail_fast() {
        let default = url::Url::parse("ws://127.0.0.1:8081/echo").unwrap();
        let err = resolve_tap_url(Some("wss://127.0.0.1:8081/echo"), &default).unwrap_err();
        assert!(err.contains("step 6"));
    }

    #[test]
    fn default_used_when_override_absent() {
        let default = url::Url::parse("ws://127.0.0.1:8081/echo").unwrap();
        let r = resolve_tap_url(None, &default).unwrap();
        assert_eq!(r.as_str(), "ws://127.0.0.1:8081/echo");
    }
}

• Step 7: Run cargo test --all + fix compile errors iteratively. The structural change to AppState.sessions (Box field on RtcSession + per-session TapConn storage) cascades through slice-1's tests. Fix each failure; do NOT suppress with #[allow]. Each fix is a behavioral-preservation commit boundary within the task.

• Step 8: Run fmt + clippy + deny

cargo fmt --check
cargo clippy --all -- -D warnings
cargo deny check

• Step 9: Commit

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

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

cargo test -p rutster

Expected: the smoke test passes.

• Step 2: Write examples/echo_brain/echo_brain.py

#!/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

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

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

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:

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

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

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