22d3f03b8c
8 TDD tasks, 53 checkbox steps, sequentially dependent:
1. Workspace deps + rutster-tap-echo skeleton (build foundation)
2. Tap wire protocol (spec §3) — JSON envelope, explicit LE PCM codec
3. TapMetrics + TapAudioPipe (spec §4.1) — the seam object
4. TapClient WSS pump loop (spec §4.2) — async WS pump + seq gap detection
5. Rust echo brain crate (spec §2.3) — stereotype binary + in-process EchoServer
6. Call-model TapHandle + Channel field (spec §5.2, §6)
7. Binary wiring — TapEngine + session_map + routes + main (spec §5.1, §7)
8. Integration test + Python echo brain + LEARNING.md (spec §8.4, §8.5 #7)
Plan self-reviewed: no placeholders; TapHandle::new visibility fixed
(pub(crate) → pub); spawn_tap_engine return shape fixed (returns
(TapAudioPipe, TapConn) tuple, not TapConn holding the pipe's mpsc).
Spec ref: 2026-06-28-slice-2-agent-tap-design.md (commit f83bca9).
117 KiB
117 KiB
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 vialicense.workspace = true. - Workspace: root
Cargo.tomlis[workspace]with[workspace.dependencies](slice-1 §2.1). New deps go in[workspace.dependencies]in the root; member crates reference withdep.workspace = true. - Workspace members (delta on slice-1): slice-1's six members (
crates/rutster,crates/rutster-media,crates/rutster-call-model,crates/rutster-signaling-sip,crates/rutster-tap,crates/rutster-spend) plus ONE new member:crates/rutster-tap-echo. Total = 7 members.crates/rutster-taptransitions from stub → real. - Dependency direction (spec §2.1):
rutster-tap→rutster-media(forPcmFrame; re-exported so one canonical home remains);rutster(binary) →rutster-tap(forTapAudioPipe/TapClienttypes) and →rutster-tap-echo(dev-binary + integration-testEchoServer);rutster-tap-echo→rutster-tap(reuses protocol types — the wire-types-reusable contract test);rutster-call-modelstays a leaf;rutster-mediadoes not depend onrutster-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 inrutster-media(single canonical home);rutster-tapre-exports it. - Wire byte order (spec §3, §9): PCM inside the base64 payload is explicit little-endian. Encoders use
i16::to_le_bytes(); decoders usei16::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 }.vis always1.seqis per-direction, starts at 0, monotonic.tsis 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_outdeclaressamples: 480. Mismatched → logged + counted + dropped (not fatal; hot-path "drop + observe" from slice-1 §3.8). - Unknown
typevalues are logged + counted + dropped (not fatal). Unknown envelope fields are ignored (forwards-compat).
- Core → brain:
- Playout ring policy (spec §4.1):
TAP_PLAYOUT_FRAMES = 5(= 100 ms). Overflow drops oldest (drop-oldest = lowest-latency-correct). Underflow returnsNonefromnext_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
Connectedthroughout. Connect timeout: 2 s. Brain-hello ack timeout: 2 s. On reconnect: re-sendhellowith the samesession_id(==ChannelId), reset bothseqcounters to 0, flush playout ring. - Tap URL validation (spec §4.4, §7.1):
ws://schemes must resolve to127.0.0.1orlocalhosthost — hard runtime check atPOST /v1/sessionstime, returns400 Bad Requestif violated.wss://URLs are rejected at session-create time with400 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 valuews://127.0.0.1:8081/echo) + optional per-calltap_urlinPOST /v1/sessionsbody. Body absent or body present withouttap_urlfield → env default. Body present withtap_url→ overrides env default; scheme validated per above. TapHandleonChannel(spec §6): zero-cost marker newtypepub struct TapHandle(())inrutster-call-model.Channelgrowspub tap: Option<TapHandle>field.rutster-call-modelstays a leaf (no tokio dep). Binary looks up live tap connection by the channel's existingChannelIdin an internalDashMap<ChannelId, TapConn>. No separateTapIdnewtype (YAGNI until multi-tap-per-channel;ChannelIdis the lookup key and the protocolsession_id).- State invariant (spec §6):
tapfield andChannelStateare tied:New/Connecting→tap: NoneConnected→tap: Some(TapHandle)(set as TapEngine spawns)Closing→taptransitions toNonebefore state advances toClosed(teardown in flight: session_end sent, task aborted, field cleared)Closed→tap: None(always)- A
ChannelinConnectedwithtap: Noneis a bug.
- Spawn location (spec §5.1 step 3): the TapEngine task is spawned by the binary's
session_map::drive_all_sessionspoll task when it observeschannel.state == Connected && channel.tap.is_none(). This keepsrutster-media'sloop_driver.rsbehaviorally 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) usethiserrorerror enums +?. - Code documentation (slice-1 §7, AGENTS.md): override the default "no comments" convention.
//!module docs on everylib.rs/main.rs/sub-module.///on every public item.//inline comments on mechanism — whympsc::try_send(notsend), whyVecDeque(notmpscfor the playout ring), whyoneshotfor drop-cancellation, whytokio::select!, whyconnect_asyncreturns a future (async vs sync connect), why explicitto_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 tomain. The newrutster-tapandrutster-tap-echocrates 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 warningson advisories. Duplicate-version bans ontokio,serde,bytes,tracing. Sources:crates-ioonly. Verify the new deps (tokio-tungstenite,futures-util,url,base64) don't trip the duplicate-version bans — Task 1 includes acargo deny checkrun 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 tomainif 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-signaling-sip/ # STUB (unchanged)
│ └── rutster-spend/ # STUB (unchanged)
└── examples/
└── echo_brain/
├── README.md # how to run, pointer to spec §3
├── echo_brain.py # canonical foreign-language brain (~80 lines, websockets lib)
└── requirements.txt # websockets
Task-to-file mapping (quick reference)
| Task | Files touched | Tests added |
|---|---|---|
| 1: Workspace deps + skeleton | Cargo.toml, crates/rutster-tap/Cargo.toml, crates/rutster-tap-echo/Cargo.toml, crates/rutster-tap-echo/src/{lib,main}.rs skeleton |
crates/rutster-tap-echo compiles; cargo deny check green |
| 2: Protocol types | crates/rutster-tap/src/protocol.rs, crates/rutster-tap/src/lib.rs |
Unit: (de)serialization round-trips; LE PCM codec; sample-count validation; unknown-type drop |
| 3: TapMetrics + TapAudioPipe | crates/rutster-tap/src/metrics.rs, crates/rutster-tap/src/tap_audio_pipe.rs, crates/rutster-tap/src/lib.rs |
Unit: overflow drops oldest; underflow returns None; mpsc send-failure drops+counts |
| 4: TapClient (WS pump) | crates/rutster-tap/src/tap_client.rs, crates/rutster-tap/src/lib.rs |
Unit: mock WS stream → audio_in egress; audio_out ingress → mpsc; seq gap counter; unknown-type drop |
| 5: rutster-tap-echo brain | crates/rutster-tap-echo/src/lib.rs, crates/rutster-tap-echo/src/main.rs |
Unit: echo_frame() logic; standalone binary smoke test |
| 6: Call-model TapHandle | crates/rutster-call-model/src/lib.rs |
Unit: tap field defaults None; TapHandle is zero-sized |
| 7: Binary wiring (TapEngine + session_map + routes) | crates/rutster/src/tap_engine.rs (NEW), crates/rutster/src/session_map.rs, crates/rutster/src/routes.rs, crates/rutster/src/main.rs, crates/rutster/Cargo.toml |
Integration: AppState creates with tap_url; routes 400 on bad URL; Connected transition spawns engine |
| 8: Integration test + examples + LEARNING.md | crates/rutster/tests/tap_integration.rs, examples/echo_brain/{README.md,echo_brain.py,requirements.txt}, LEARNING.md |
Integration: end-to-end echo through EchoServer; reconnect test; Python brain README |
Task 1: Workspace dependencies + crate skeletons
Files:
- Modify:
Cargo.toml(root) — add[workspace.dependencies]entries; addcrates/rutster-tap-echotomembers. - 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 withfn mainplaceholder) - Modify:
deny.toml— only ifcargo deny checkflags a new transitive license (verify, don't speculatively edit).
Interfaces:
-
Produces: workspace-level dep versions for
tokio-tungstenite,futures-util,url,base64. A compilingrutster-tap-echoskeleton (lib + binary) so Task 5 has a home. Arutster-tapmanifest 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-signaling-sip",
"crates/rutster-tap",
"crates/rutster-tap-echo",
"crates/rutster-spend",
]
- Step 2: Transition
crates/rutster-tap/Cargo.tomlfrom 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.rsskeleton
//! # 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.rsskeleton
//! 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 --allandcargo 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 --checkandcargo 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-exportPcmFrame.
Interfaces:
- Consumes:
PcmFramefromrutster-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 matchrutster_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.rswith 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.rswith 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,AudioSinkfromrutster-media(slice-1).tokio::sync::mpsc,tokio::sync::oneshotfromtokioworkspace 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 }withfn snapshot(&self) -> MetricsSnapshotpub 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>) -> TapAudioPipeimpl 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.rswith 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_nonetest 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_tungsteniteWebSocketStream,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'spump_loopreturnsOk(())on gracefulbye/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.rscallsstart_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 — addpub struct TapHandle(())andpub tap: Option<TapHandle>field onChannel).
Interfaces:
-
Consumes: nothing new (no tokio dep —
TapHandleis 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
TapHandletype +Channel.tapfield — 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 optionaltap_url; validate per §4.4) - Modify:
crates/rutster/src/main.rs(readRUTSTER_TAP_URLenv; pass toAppState) - Modify:
crates/rutster/src/lib.rs(declarepub mod tap_engine;) - Modify:
crates/rutster/Cargo.toml(addrutster-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<()>) -> TapConnAppState.tap_registry: Arc<DashMap<ChannelId, TapConn>>AppState.default_tap_url: UrlPOST /v1/sessionsbody optionally{ "tap_url": "ws://..." }; returns400 Bad Requeston non-loopbackws://, unparseable, orwss://URL.
-
Step 1: Update
crates/rutster/Cargo.tomldeps
[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:
create_sessiontakesOption<Url>(tap_url override) and stores the validated URL alongside the session (the URL must outlivecreate_sessionso the poll task can spawn the engine onConnected).- Store the resolved tap URL on
RtcSessionor in a parallelDashMap<ChannelId, Url>. Simpler: add a fieldpub tap_url: url::Urlto a wrapper structSessionEntry { rtc: RtcSession, tap_url: url::Url, tap_conn: Option<TapConn> }, replacing the bareArc<Mutex<RtcSession>>in theDashMap. This is a structural change toAppState.sessions's value type. - In
drive_all_sessions, afterrtc.run_poll_once(now):- If
rtc.channel.state == Connected && entry.tap_conn.is_none(): spawn the TapEngine, store theTapConn, setrtc.channel.tap = Some(TapHandle::new()). - If
rtc.channel.state == Closing && entry.tap_conn.is_some(): sendsession_end(via dropping the close oneshot + awaiting a bounded teardown), abort the engine, drop theTapConn, clearrtc.channel.tap = Nonebefore state advances toClosed.
- If
- The
TapAudioPipemust be wired intoRtcSessionbeforeaccept_offeris 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 atRtcSession::new, not at Connected. Reconcile:RtcSession::newneeds to accept theTapAudioPipe(or a genericP: AudioSource + AudioSink). Preferred: makeRtcSession::newtake apipe: Arc<Mutex<...>>(or keep theEchoAudioPipedefault + add aRtcSession::with_pipe(pipe)constructor); the binary choosesTapAudioPipeorEchoAudioPipeat 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
RtcSessionto holdpipe: Arc<Mutex<dyn AudioSource + AudioSink>>(a trait object) OR keeppipe: EchoAudioPipeplus a second fieldtap_pipe: Option<TapAudioPipe>and switch theloop_driver's calls to dispatch via a method. - Cleanest (no behavioral change to loop_driver): change the field type from
pipe: EchoAudioPipetopipe: Box<dyn AudioSource + AudioSink>(stable Rust Box, no pointer widening concerns at slice-1 scale).loop_drivercontinues to callsession.pipe.next_pcm_frame()andsession.pipe.on_pcm_frame()— the call site is unchanged; only the field type widens. This is the spec §8.5 #6 guarantee made concrete. EchoAudioPipebecomes the--features=echopath (or a default-features fallback);TapAudioPipeis 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>. (ConfirmAudioSource/AudioSinkdeclared: Send— slice-1 haspub trait AudioSource: Send.) -
Change
RtcSession::newto keepEchoAudioPipeas the default (so unit tests don't need a mock); addpub 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 callRtcSession::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 toAppState.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'sAppState/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 withhello(echo thesession_id). - On
audio_in: echo the same PCM back asaudio_out(advisory — core disposes). - On
bye/session_end: close cleanly. - Stateless across reconnects (spec §5.3) — every
hellostarts 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)
cargo test --allpasses (unit + integration).cargo fmt --check,cargo clippy -- -D warnings,cargo deny checkall pass.cargo run(withrutster-tap-echorunning) → browser, speak, hear echo through the external brain within ~250 ms.- Kill echo brain mid-call → server reconnects with bounded backoff (visible in logs + browser
<pre>), audio resumes on brain restart,Channelnever leftConnected. - Both
rutster-tap-echo(Rust) andexamples/echo_brain/echo_brain.py(Python) interop against the core. - 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 inrtc_session.rsis 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. LEARNING.mdhas ≥3 new pointers (the plan ships 5).