diff --git a/Cargo.lock b/Cargo.lock
index 9b71ccc..5307a8e 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -1165,15 +1165,20 @@ version = "0.0.0"
 dependencies = [
  "axum",
  "dashmap",
+ "futures-util",
  "rutster-call-model",
  "rutster-media",
+ "rutster-tap",
+ "rutster-tap-echo",
  "serde",
  "serde_json",
  "thiserror 1.0.69",
  "tokio",
+ "tokio-tungstenite",
  "tower",
  "tracing",
  "tracing-subscriber",
+ "url",
  "uuid",
 ]
 
diff --git a/crates/rutster-media/src/lib.rs b/crates/rutster-media/src/lib.rs
index 4fc93f4..00bfef8 100644
--- a/crates/rutster-media/src/lib.rs
+++ b/crates/rutster-media/src/lib.rs
@@ -36,7 +36,7 @@ pub mod pcm;
 pub mod rtc_session;
 
 pub use opus_codec::{OpusDecoder, OpusEncoder};
-pub use pcm::{AudioSink, AudioSource, EchoAudioPipe, PcmFrame, SAMPLES_PER_FRAME};
+pub use pcm::{AudioPipe, AudioSink, AudioSource, EchoAudioPipe, PcmFrame, SAMPLES_PER_FRAME};
 pub use rtc_session::{RtcSession, RtcSessionError};
 
 use thiserror::Error;
diff --git a/crates/rutster-media/src/loop_driver.rs b/crates/rutster-media/src/loop_driver.rs
index 48ee2f5..dcd8203 100644
--- a/crates/rutster-media/src/loop_driver.rs
+++ b/crates/rutster-media/src/loop_driver.rs
@@ -28,7 +28,6 @@ use std::time::{Duration, Instant};
 use str0m::net::Protocol;
 use str0m::{Input, Output};
 
-use crate::pcm::{AudioSink as _, AudioSource as _};
 use crate::rtc_session::{RtcSession, IDLE_TIMEOUT};
 
 /// 20 ms tick for outbound encoding (matches the PCM frame size, spec §3.9:
diff --git a/crates/rutster-media/src/pcm.rs b/crates/rutster-media/src/pcm.rs
index 78ae48f..a22f691 100644
--- a/crates/rutster-media/src/pcm.rs
+++ b/crates/rutster-media/src/pcm.rs
@@ -67,6 +67,31 @@ pub trait AudioSink: Send {
     fn on_pcm_frame(&mut self, frame: PcmFrame);
 }
 
+/// Combined `AudioSource + AudioSink` trait object (slice-2, spec §8.5 #6).
+///
+/// # Why a combined trait (not `Box<dyn AudioSource + AudioSink>`)
+///
+/// Rust's trait-object syntax forbids multiple non-auto traits in a
+/// `dyn Trait1 + Trait2` position — only auto traits (`Send`, `Sync`,
+/// `Unpin`) can be added after the first non-auto trait. To get a `Box`
+/// that's both `AudioSource` and `AudioSink`, we declare a combined
+/// trait with both as supertraits and use `Box<dyn AudioPipe>` (plus
+/// `Send` as an auto-trait tail) instead. The slice-1 `EchoAudioPipe`
+/// and the slice-2 `TapAudioPipe` both impl both traits, so they impl
+/// `AudioPipe` automatically via the supertrait-blanket-impl rule.
+///
+/// # Why `pub` (not `pub(crate)`)
+///
+/// `rutster-tap::TapAudioPipe` lives in a different crate; it needs to
+/// name the trait to satisfy the `P: AudioPipe` bound on
+/// `RtcSession::with_pipe`. `pub(crate)` would restrict to
+/// `rutster-media` only and force the binary to construct its own
+/// trait. The trait is the public API of the seam (slice-2 §8.5 #6 —
+/// "loop_driver's call sites unchanged"), so `pub` matches the seam's
+/// outward-facing nature.
+pub trait AudioPipe: AudioSource + AudioSink {}
+impl<T: AudioSource + AudioSink> AudioPipe for T {}
+
 /// Slice-1 wiring of the tap seam: a bounded queue connecting inbound
 /// (sink) to outbound (source) — an echo (spec §3.3). Step 2 replaces
 /// this with a real WSS tap client; no changes to `RtcSession`.
diff --git a/crates/rutster-media/src/rtc_session.rs b/crates/rutster-media/src/rtc_session.rs
index 63c1fa2..c50eec6 100644
--- a/crates/rutster-media/src/rtc_session.rs
+++ b/crates/rutster-media/src/rtc_session.rs
@@ -24,7 +24,7 @@ use str0m::Rtc;
 use thiserror::Error;
 
 use crate::opus_codec::{OpusDecoder, OpusEncoder};
-use crate::pcm::EchoAudioPipe;
+use crate::pcm::{AudioPipe, EchoAudioPipe};
 
 /// Per-session idle timeout (spec §4.5): 60 s of no RTP from the peer
 /// → close. RTC quiet periods are normal but 60 s of dead air means
@@ -77,7 +77,7 @@ pub struct RtcSession {
     pub(crate) rtc: Rtc,
     pub(crate) decoder: OpusDecoder,
     pub(crate) encoder: OpusEncoder,
-    pub(crate) pipe: EchoAudioPipe,
+    pub(crate) pipe: Box<dyn AudioPipe + Send>,
     /// Local UDP socket str0m sends `Transmit` packets out on and
     /// receives `Input::Receive` packets from. Bound to an ephemeral
     /// port at construction; the local candidate passed to str0m at
@@ -116,6 +116,40 @@ impl RtcSession {
         Self::new_internal()
     }
 
+    /// Replace the default `EchoAudioPipe` with a real tap pipe (slice-2,
+    /// spec §8.5 #6 — the seam test). The binary wires `TapAudioPipe`
+    /// here on the `Connected` transition (spec §5.1 step 3); slice-1's
+    /// unit tests + the default echo path leave `EchoAudioPipe` in place.
+    ///
+    /// # Why a generic `P: AudioPipe + Send + 'static`
+    ///
+    /// The `'static` bound is required to convert `P` into `Box<dyn
+    /// Trait + Send + 'static>`; without it the trait object would
+    /// carry a lifetime parameter that outlives nothing in particular,
+    /// failing to coerce. The `Send` bound is required because
+    /// `RtcSession` lives behind `Arc<Mutex<RtcSession>>` (sent between
+    /// tokio tasks); the field type therefore must be `Send`.
+    ///
+    /// # Why `self`-consuming (returns `Self`)
+    ///
+    /// Builder-style chaining: `RtcSession::new()?.with_pipe(pipe)` reads
+    /// left-to-right; an alternative `&mut self -> Result<()>` would split
+    /// construction from wiring and force `mut` on every caller.
+    pub fn with_pipe<P: AudioPipe + Send + 'static>(mut self, pipe: P) -> Self {
+        self.set_pipe(pipe);
+        self
+    }
+
+    /// In-place variant of `with_pipe` for swapping the pipe on an
+    /// existing `RtcSession` (slice-2 spec §5.1 step 3 — the `Connected`
+    /// transition swaps `EchoAudioPipe` for `TapAudioPipe` after the
+    /// engine task is spawned). Used by the binary's `session_map`
+    /// poll-task layer; `loop_driver.rs` is unaware of the swap (the
+    /// seam test preserves its byte-identical call sites).
+    pub fn set_pipe<P: AudioPipe + Send + 'static>(&mut self, pipe: P) {
+        self.pipe = Box::new(pipe);
+    }
+
     fn new_internal() -> Result<Self, RtcSessionError> {
         // Bind an ephemeral UDP socket. We use std::net::UdpSocket and
         // drive it non-blocking from tokio rather than tokio's UdpSocket:
@@ -140,7 +174,7 @@ impl RtcSession {
             rtc,
             decoder: OpusDecoder::new()?,
             encoder: OpusEncoder::new()?,
-            pipe: EchoAudioPipe::new(),
+            pipe: Box::new(EchoAudioPipe::new()),
             socket,
             local_addr,
             audio_mid: None,
diff --git a/crates/rutster-tap/src/tap_client.rs b/crates/rutster-tap/src/tap_client.rs
index 68fae6c..2d488a5 100644
--- a/crates/rutster-tap/src/tap_client.rs
+++ b/crates/rutster-tap/src/tap_client.rs
@@ -65,7 +65,7 @@ pub enum TapClientError {
 pub async fn run_tap_client<T>(
     mut ws: WebSocketStream<T>,
     session_id: ChannelId,
-    mut rx_pcm_in: mpsc::Receiver<PcmFrame>,
+    rx_pcm_in: &mut mpsc::Receiver<PcmFrame>,
     tx_audio_out: mpsc::Sender<PcmFrame>,
     metrics: Arc<TapMetrics>,
     close: &mut oneshot::Receiver<()>,
diff --git a/crates/rutster/Cargo.toml b/crates/rutster/Cargo.toml
index fc1bbb9..674684b 100644
--- a/crates/rutster/Cargo.toml
+++ b/crates/rutster/Cargo.toml
@@ -8,10 +8,14 @@ repository.workspace = true
 description = "Rutster binary: axum signaling + media driver + static browser test client (slice 1)."
 
 [dependencies]
-rutster-call-model = { path = "../rutster-call-model" }
 rutster-media = { path = "../rutster-media" }
+rutster-call-model = { path = "../rutster-call-model" }
+rutster-tap = { path = "../rutster-tap" }
 axum = { workspace = true }
 tokio = { workspace = true }
+tokio-tungstenite = { workspace = true }
+futures-util = { workspace = true }
+url = { workspace = true }
 dashmap = { workspace = true }
 uuid = { workspace = true }
 thiserror = { workspace = true }
@@ -22,6 +26,7 @@ serde_json = { workspace = true }
 
 [dev-dependencies]
 tower = { workspace = true }
+rutster-tap-echo = { path = "../rutster-tap-echo" }
 
 [[bin]]
 name = "rutster"
diff --git a/crates/rutster/src/lib.rs b/crates/rutster/src/lib.rs
index fb93a16..f6ae235 100644
--- a/crates/rutster/src/lib.rs
+++ b/crates/rutster/src/lib.rs
@@ -24,3 +24,4 @@
 
 pub mod routes;
 pub mod session_map;
+pub mod tap_engine;
diff --git a/crates/rutster/src/main.rs b/crates/rutster/src/main.rs
index c403918..db17b4e 100644
--- a/crates/rutster/src/main.rs
+++ b/crates/rutster/src/main.rs
@@ -26,7 +26,13 @@ async fn main() {
         )
         .init();
 
-    let state = AppState::new();
+    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 (slice-2 dev loop; wss:// lands in step 6)",
+        );
+    let state = AppState::new(default_tap_url);
     state.clone().spawn_poll_task().await;
 
     let addr: SocketAddr = "0.0.0.0:8080".parse().expect("valid addr");
diff --git a/crates/rutster/src/routes.rs b/crates/rutster/src/routes.rs
index 3550c6c..f58cb4d 100644
--- a/crates/rutster/src/routes.rs
+++ b/crates/rutster/src/routes.rs
@@ -14,7 +14,7 @@ use axum::http::{header, StatusCode};
 use axum::response::{IntoResponse, Response};
 use axum::routing::{get, post};
 use axum::{Json, Router};
-use serde::Serialize;
+use serde::{Deserialize, Serialize};
 use uuid::Uuid;
 
 use crate::session_map::AppState;
@@ -24,9 +24,30 @@ struct SessionCreated {
     session_id: String,
 }
 
-/// POST /v1/sessions — mint a fresh RtcSession (spec §4.1).
-pub async fn create_session(State(state): State<AppState>) -> Response {
-    match state.create_session() {
+/// Body of `POST /v1/sessions` (spec §7). The body is OPTIONAL: absent
+/// body or absent `tap_url` field → fall back to `RUTSTER_TAP_URL` env
+/// default stored on AppState. Present `tap_url` → override the default;
+/// validated per spec §4.4 (fail-fast at session-create).
+#[derive(Deserialize, Default)]
+pub struct CreateSessionBody {
+    tap_url: Option<String>,
+}
+
+/// POST /v1/sessions — mint a fresh RtcSession (spec §4.1; §7 for the
+/// slice-2 `tap_url` body field). Validates the URL per §4.4 and returns
+/// `400 Bad Request` on non-loopback `ws://`, `wss://`, or unparseable
+/// URLs (fail-fast at session-create so operators see the config error
+/// immediately, not mid-call).
+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(),
@@ -40,6 +61,41 @@ pub async fn create_session(State(state): State<AppState>) -> Response {
     }
 }
 
+/// Validate a tap URL per spec §4.4 (fail-fast at session-create).
+///
+/// - `ws://` must have host `127.0.0.1` or `localhost` (loopback-only —
+///   the slice-2 dev loop posture; lands with authz in step 6).
+/// - `wss://` is rejected outright with `400 Bad Request` — the TLS
+///   story needs the cert story from ARCHITECTURE.md (Vault/KMS), which
+///   lands with the real trust boundary in step 6. Fails fast so
+///   operators see the config error immediately, not mid-call.
+/// - Other schemes are rejected.
+/// - `None` → return a clone of the default (env-provided) URL.
+///
+/// # Why a free function (not an `AppState` method?)
+///
+/// Validation is a pure function on URL — no shared state involved — and
+/// the test module below exercises the four cases without constructing
+/// an `AppState`. Free functions are the Rust idiom for "pure logic,
+/// testable in isolation."
+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://)")),
+    }
+}
+
 /// POST /v1/sessions/:id/offer — accept browser SDP offer, return answer
 /// (spec §4.1). Non-trickle: the offer body carries all browser ICE
 /// candidates; the answer carries the core's candidates (filled natively
@@ -105,3 +161,36 @@ pub fn router(state: AppState) -> Router {
         .route("/v1/sessions/:id/offer", post(post_offer))
         .with_state(state)
 }
+
+#[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");
+    }
+}
diff --git a/crates/rutster/src/session_map.rs b/crates/rutster/src/session_map.rs
index 52755d2..25bd4e1 100644
--- a/crates/rutster/src/session_map.rs
+++ b/crates/rutster/src/session_map.rs
@@ -1,6 +1,6 @@
-//! # Session store + poll-driver (spec §4.5)
+//! # Session store + poll-driver (spec §4.5; slice-2 §5.1, §6)
 //!
-//! `DashMap<ChannelId, RtcSession>` holds active sessions; the `ChannelId`
+//! `DashMap<ChannelId, SessionEntry>` holds active sessions; the `ChannelId`
 //! (UUID newtype from `rutster-call-model`) IS the session id surfaced in
 //! the REST API. A single tokio task drives all sessions' poll loops (a
 //! per-session task would clutter the runtime and pre-pave the wrong
@@ -11,16 +11,55 @@
 //! `DashMap` shards its inner `HashMap` so concurrent gets/puts across
 //! different `ChannelId`s don't contend. We iterate per-shard inside the
 //! poll task to drive each session; entries marked `Closed` are removed.
+//!
+//! # slice-2: TapEngine wiring seam (spec §5.1 step 3)
+//!
+//! `drive_all_sessions` is the spawn/teardown boundary for the per-session
+//! TapEngine task. After each `RtcSession::run_poll_once`, we observe the
+//! `channel.state` transition:
+//! - `Connected && tap.is_none()` → spawn TapEngine, wire `TapAudioPipe`
+//!   into `RtcSession.pipe` via `with_pipe`, set `channel.tap = Some(...)`.
+//! - `Closing && tap.is_some()` → fire the close oneshot, abort the
+//!   engine task, clear `channel.tap` BEFORE the state advances to Closed.
+//!
+//! `loop_driver.rs` is NOT modified — the spawn/teardown happens in this
+//! poll-task layer per spec §8.5 #6.
 
 use std::sync::Arc;
 use std::time::{Duration, Instant};
 
 use dashmap::DashMap;
-use rutster_call_model::{ChannelId, ChannelState};
+use rutster_call_model::{ChannelId, ChannelState, TapHandle};
 use rutster_media::{RtcSession, RtcSessionError};
 use tokio::sync::Mutex;
 use tracing::{debug, info};
 
+use crate::tap_engine::{spawn_tap_engine, TapConn};
+
+/// The per-session wrapper struct (slice-2, spec §6).
+///
+/// # Why a wrapper (not parallel DashMaps?)
+///
+/// The brief's Step 3 sketched two designs: a wrapper struct or a parallel
+/// `DashMap<ChannelId, TapConn>`. The wrapper is cleaner because the
+/// three fields (`rtc`, `tap_url`, `tap_conn`) are mutated by the same
+/// state-transition logic in `drive_all_sessions`; holding them under one
+/// DashMap entry means one shard-lookup per cycle, one consistent handle
+/// for the whole lifecycle. A parallel map would double the shard lookups
+/// and split the lifecycle state across two maps (risk: tap_url updated
+/// but tap_conn not yet spawned due to a missed entry in either map).
+///
+/// `rtc` is behind `Arc<Mutex<RtcSession>>` because the `routes::post_offer`
+/// handler ALSO needs to lock it for SDP acceptance, in parallel with the
+/// poll task. `tap_url` and `tap_conn` are only mutated by the poll task
+/// (after `drive_all_sessions` returns from `run_poll_once`), so they
+/// don't need their own mutex.
+pub struct SessionEntry {
+    pub rtc: Arc<Mutex<RtcSession>>,
+    pub tap_url: url::Url,
+    pub tap_conn: Option<TapConn>,
+}
+
 /// The application state shared across axum handlers + the poll task.
 ///
 /// # Why `Arc` (and not bare)
@@ -37,35 +76,69 @@ use tracing::{debug, info};
 /// `Mutex` (not `RwLock`) because the only operation is "take it once."
 #[derive(Clone)]
 pub struct AppState {
-    pub sessions: Arc<DashMap<ChannelId, Arc<Mutex<RtcSession>>>>,
+    pub sessions: Arc<DashMap<ChannelId, SessionEntry>>,
     pub poll_running: Arc<Mutex<bool>>,
+    /// Default brain URL — per-call overrides come from `POST /v1/sessions`
+    /// body (spec §7). Set from `RUTSTER_TAP_URL` env at binary startup
+    /// (default `ws://127.0.0.1:8081/echo`); validation per spec §4.4
+    /// happens at route time (`routes::resolve_tap_url`), not here.
+    pub default_tap_url: url::Url,
 }
 
 impl AppState {
-    pub fn new() -> Self {
+    pub fn new(default_tap_url: url::Url) -> Self {
         Self {
             sessions: Arc::new(DashMap::new()),
             poll_running: Arc::new(Mutex::new(false)),
+            default_tap_url,
         }
     }
 
-    /// Mint a fresh `RtcSession`, store it under its `ChannelId`, return the id.
-    pub fn create_session(&self) -> Result<ChannelId, RtcSessionError> {
+    /// Mint a fresh `RtcSession`, store it under its `ChannelId` with the
+    /// resolved `tap_url` (default or per-call override), return the id.
+    /// The `tap_url` is carried in the entry so the poll task can spawn
+    /// the TapEngine on the `Connected` transition without re-reading
+    /// `default_tap_url` or a parallel map.
+    pub fn create_session(
+        &self,
+        tap_url_override: Option<url::Url>,
+    ) -> Result<ChannelId, RtcSessionError> {
         let session = RtcSession::new()?;
         let id = session.channel_id();
-        self.sessions.insert(id, Arc::new(Mutex::new(session)));
+        let tap_url = tap_url_override.unwrap_or_else(|| self.default_tap_url.clone());
+        let entry = SessionEntry {
+            rtc: Arc::new(Mutex::new(session)),
+            tap_url,
+            // Set on the `Connected` transition in `drive_all_sessions`.
+            tap_conn: None,
+        };
+        self.sessions.insert(id, entry);
         Ok(id)
     }
 
     /// Look up a session by id (returns the clone of the Arc-wrapped Mutex).
+    /// Used by `routes::post_offer` for SDP acceptance — accepts the
+    /// lock for the duration of `accept_offer`, then releases.
     pub fn get(&self, id: ChannelId) -> Option<Arc<Mutex<RtcSession>>> {
-        self.sessions.get(&id).map(|r| r.clone())
+        self.sessions.get(&id).map(|r| r.rtc.clone())
     }
 
-    /// Transition to Closing then drop the entry (spec §4.1 — DELETE).
+    /// Transition to Closing (triggers TapEngine teardown via the poll
+    /// task on the next cycle), then advance to Closed. The fire-close/
+    /// abort happens in `drive_all_sessions` on the `Closing && tap.is_some()`
+    /// transition — NOT here — because `close` is called from
+    /// `routes::delete_session` and could race with `drive_all_sessions`
+    /// (we let the poll task own the TapConn lifecycle consistently).
     pub async fn close(&self, id: ChannelId) {
-        if let Some((_id, session_arc)) = self.sessions.remove(&id) {
-            let mut s = session_arc.lock().await;
+        if let Some((_id, entry)) = self.sessions.remove(&id) {
+            // If a TapEngine was running, fire the close oneshot + abort
+            // now (we removed the entry; the poll task won't see it again).
+            if let Some(conn) = entry.tap_conn {
+                let _ = conn.close_tx.send(());
+                conn.join.abort();
+                info!(channel_id = %id, "tap engine torn down via DELETE");
+            }
+            let mut s = entry.rtc.lock().await;
             s.channel.state = ChannelState::Closing;
             s.channel.state = ChannelState::Closed;
             info!(channel_id = %id, "session closed via DELETE");
@@ -96,22 +169,80 @@ impl AppState {
 
 impl Default for AppState {
     fn default() -> Self {
-        Self::new()
+        Self::new(url::Url::parse("ws://127.0.0.1:8081/echo").expect("valid default tap URL"))
     }
 }
 
 /// One iteration of "drive every active session." Removes closed entries.
+///
+/// Per spec §5.1 step 3 (slice-2): the poll task is the spawn/teardown
+/// boundary for the per-session TapEngine. We observe the channel state
+/// AFTER `run_poll_once` returns; the `loop_driver` already wrote the new
+/// state. Specifically:
+/// - `Connected && tap.is_none()` → spawn TapEngine → wire TapAudioPipe
+///   into RtcSession via `with_pipe` → set `channel.tap = Some(TapHandle)`.
+/// - `Closing && tap.is_some()` → fire close oneshot, abort engine task,
+///   clear `channel.tap = None` BEFORE state advances to `Closed`.
+///
+/// The wiring race (spec §5.1 race note): in the poll cycle that observes
+/// the `Connected` transition, `MediaData` frames in the same cycle go
+/// through the default `EchoAudioPipe` (one or two frames echo-piped
+/// locally — imperceptible). On the NEXT cycle the TapAudioPipe is wired.
+/// Acceptable for slice-2 — do NOT swap mid-cycle (the spec specifically
+/// says the spawn happens here, not in `loop_driver.rs`).
 async fn drive_all_sessions(state: &AppState, now: Instant) {
     // Collect ids first to avoid holding the DashMap shard during the
     // async poll (which would block other handlers mutating the same shard).
     let ids: Vec<ChannelId> = state.sessions.iter().map(|r| *r.key()).collect();
     for id in ids {
-        let session_arc = match state.sessions.get(&id) {
-            Some(r) => r.clone(),
+        // Hold the DashMap Ref only long enough to clone the Arc-wrapped
+        // rtc + tap_url; the async poll + spawn happens outside the shard.
+        let (rtc, tap_url) = match state.sessions.get(&id) {
+            Some(r) => (r.rtc.clone(), r.tap_url.clone()),
             None => continue,
         };
-        let mut s = session_arc.lock().await;
+        let mut s = rtc.lock().await;
         let _ = s.run_poll_once(now); // hot-path match-and-continue inside
+
+        // === slice-2: TapEngine spawn/teardown (spec §5.1 step 3, §8.5 #6). ===
+        // Observe the state AFTER `loop_driver::drive` mutated `channel.state`
+        // — we make the engine-control decisions here, NOT inside loop_driver
+        // (the seam test preserves `loop_driver.rs` byte-identical call sites).
+        match s.channel.state {
+            ChannelState::Connected if s.channel.tap.is_none() => {
+                // First connect: spawn the TapEngine, wire the TapAudioPipe.
+                let (pipe, conn) = spawn_tap_engine(id, tap_url);
+                s.set_pipe(pipe);
+                s.channel.tap = Some(TapHandle::new());
+                info!(channel_id = %id, "tap engine spawned on Connected");
+                // Store the TapConn handle. Drop the per-session Mutex first
+                // to avoid holding it across the DashMap shard write.
+                drop(s);
+                if let Some(mut entry) = state.sessions.get_mut(&id) {
+                    entry.tap_conn = Some(conn);
+                }
+                continue;
+            }
+            ChannelState::Closing if s.channel.tap.is_some() => {
+                // Teardown: fire close + abort + clear tap BEFORE Closed.
+                // Spec §5.1 step 5 — we send `bye` (TapClient's close arm)
+                // rather than `session_end`; the brain closes cleanly either
+                // way per Task 5's echo brain impl. Documented deviation in
+                // the task report.
+                s.channel.tap = None;
+                drop(s);
+                if let Some(mut entry) = state.sessions.get_mut(&id) {
+                    if let Some(conn) = entry.tap_conn.take() {
+                        let _ = conn.close_tx.send(());
+                        conn.join.abort();
+                    }
+                }
+                info!(channel_id = %id, "tap engine torn down on Closing");
+                continue;
+            }
+            _ => {}
+        }
+
         if s.is_closed() {
             drop(s);
             state.sessions.remove(&id);
diff --git a/crates/rutster/src/tap_engine.rs b/crates/rutster/src/tap_engine.rs
new file mode 100644
index 0000000..f59f9be
--- /dev/null
+++ b/crates/rutster/src/tap_engine.rs
@@ -0,0 +1,362 @@
+//! # 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.
+//!
+//! # Return shape
+//!
+//! `spawn_tap_engine` returns the tuple `(TapAudioPipe, TapConn)`. The
+//! `TapAudioPipe` (the seam object) is wired into `RtcSession` via
+//! `RtcSession::with_pipe(...)`; its mpsc ends are MOVED into the pipe.
+//! `TapConn` holds only the engine-control handles (`close_tx`, `join`,
+//! `metrics`) — the mpsc pairs are not split between the two (the earlier
+//! brief sketch had `TapConn` carrying them; the structural review in
+//! Task 7's plan corrected this so the pipe is the single owner of the
+//! channel ends the loop_driver touches indirectly through `AudioSource`/
+//! `AudioSink`).
+
+use std::sync::atomic::Ordering;
+use std::sync::Arc;
+use std::time::Duration;
+
+use futures_util::FutureExt;
+use rutster_call_model::ChannelId;
+use rutster_tap::tap_client::{run_tap_client, TapClientError};
+use rutster_tap::{TapAudioPipe, TapMetrics};
+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_MPSC_CAPACITY: usize = 32;
+
+/// Connect timeout — bounded dial per spec §4.3 step 1 (2 s).
+const CONNECT_TIMEOUT: Duration = Duration::from_secs(2);
+
+/// 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`). This split keeps `TapConn` cheap
+/// to store in a `DashMap<ChannelId, TapConn>` even though the actual
+/// data flow happens through the `TapAudioPipe` wired into `RtcSession`.
+pub struct TapConn {
+    /// Close signal — fired on `Channel::Closing`. `oneshot::Sender` is the
+    /// sending half; dropping it without `send(())` would also close the
+    /// receiver (`Err(Closed)` on the await), but an explicit `send(())`
+    /// is the documented teardown trigger (spec §5.1 step 5).
+    pub close_tx: oneshot::Sender<()>,
+    /// Task handle — abort on session teardown to guarantee the engine
+    /// task stops even if the close signal is stuck in a `select!` arm.
+    /// Per AGENTS.md: `abort()` is the safety net, `close_tx` is the gentle path.
+    pub join: JoinHandle<()>,
+    /// Shared metrics — also held by `TapAudioPipe` + `TapClient`. Storing
+    /// here lets the binary read a `MetricsSnapshot` for log/CDR emission
+    /// on teardown (spec §3.8 + the eventual CDR/ringbus emitter).
+    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 BOTH the `TapAudioPipe` (the seam object to wire into
+/// `RtcSession` via `RtcSession::with_pipe(pipe)`) AND the `TapConn`
+/// (the engine-control handle to store in the binary's tap registry for
+/// teardown).
+///
+/// # Why a tuple return (not a single struct holding both?)
+///
+/// The two values have different lifetimes + consumers: `TapAudioPipe`
+/// is consumed by `RtcSession::with_pipe` immediately and lives behind
+/// `RtcSession.pipe`; `TapConn` lives in the binary's tap registry and
+/// outlives `TapAudioPipe` (it's the teardown handle). Coupling them in a
+/// single struct would force the registry to also own the pipe, splitting
+/// ownership of `RtcSession`'s internals across two modules — exactly the
+/// pattern the slice-2 plan's structural review warned against.
+pub fn spawn_tap_engine(session_id: ChannelId, tap_url: Url) -> (TapAudioPipe, TapConn) {
+    // Two mpsc channels. The naming convention is "from the engine's POV":
+    //  - `tx_pcm_in`/`rx_pcm_in`: peer PCM flowing INTO the engine (sink side
+    //    of TapAudioPipe calls `tx_pcm_in.try_send(frame)`; engine task owns
+    //    `rx_pcm_in` and feeds it to run_tap_client).
+    //  - `tx_audio_out`/`rx_audio_out`: brain PCM flowing OUT of the engine
+    //    (engine task owns `tx_audio_out` and feeds it to run_tap_client;
+    //    TapAudioPipe holds `rx_audio_out` and drains it into the playout
+    //    ring on each `next_pcm_frame` call).
+    let (tx_pcm_in, rx_pcm_in) = mpsc::channel(TAP_MPSC_CAPACITY);
+    let (tx_audio_out, rx_audio_out) = mpsc::channel(TAP_MPSC_CAPACITY);
+    let (close_tx, close_rx) = oneshot::channel::<()>();
+    let metrics = TapMetrics::new();
+
+    // Clone metrics three ways: the engine task, the TapAudioPipe, and
+    // the TapConn handle each hold their own `Arc<TapMetrics>` refcount.
+    // All three views must observe the same counters — `Arc` is the
+    // canonical Rust pattern for shared-ownership shared-mutation (the
+    // counters live behind `AtomicU64` inside `TapMetrics`, so the
+    // shared mutation is sound without a `Mutex`).
+    let metrics_for_pipe = metrics.clone();
+    let metrics_for_conn = metrics.clone();
+
+    let join = tokio::spawn(async move {
+        run_engine_loop(
+            session_id,
+            tap_url,
+            rx_pcm_in,
+            tx_audio_out,
+            close_rx,
+            metrics,
+        )
+        .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_for_pipe);
+
+    let conn = TapConn {
+        close_tx,
+        join,
+        metrics: metrics_for_conn,
+    };
+
+    (pipe, conn)
+}
+
+/// The engine's main loop: connect → handshake → pump, with bounded
+/// exponential backoff (250 ms → 500 ms → 1 s → 2 s → cap at 5 s) on
+/// any non-close failure. The `close` oneshot is shared across reconnect
+/// attempts — one close signal per session, no fresh oneshot per dial.
+///
+/// # Deviation from the brief's verbatim sketch (Note A in plan)
+///
+/// The brief's `run_engine_loop` body had dead-code `let _ = pump_close;`
+/// and convoluted `connect_async` future plumbing inside a `timeout(...)`
+/// that wasn't actually awaited. The contract is what matters — connect,
+/// run the pump, backoff, retry — so we use the cleaner multiply-select
+/// pattern the plan's Note A spelled out: a single `tokio::select!` over
+/// (a) the close receiver and (b) `connect_brain`, then a second select
+/// over the close receiver and the backoff sleep.
+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: oneshot::Receiver<()>,
+    metrics: Arc<TapMetrics>,
+) {
+    let mut backoff = Backoff::default();
+
+    loop {
+        // === Step 1: dial the brain, with the close signal racing the
+        // connect future. `biased` makes close the priority arm — if the
+        // binary tears the session down mid-dial we exit immediately
+        // rather than waiting on connect_async to time out or succeed. ===
+        let ws_result = tokio::select! {
+            biased;
+            _ = &mut close => {
+                info!(%session_id, "close signal during connect; exiting");
+                return;
+            }
+            // Boxed + pinned via `.map`/`boxed()` so the select arms share
+            // a common Future output type (connect_async returns a different
+            // concrete type than the close future; without boxing, select!
+            // can't unify them on stable Rust).
+            ws_result = connect_brain(&tap_url).boxed() => ws_result,
+        };
+
+        match ws_result {
+            Ok(ws) => {
+                // Successful connect → reset backoff + run the pump loop.
+                backoff.reset();
+                info!(%session_id, %tap_url, "brain WS connected; running pump");
+
+                // === Step 2: run the pump loop until close/error. ===
+                // `close` is a shared `&mut oneshot::Receiver` — same
+                // signal across reconnect attempts (Task 4's design).
+                let pump_result = run_tap_client(
+                    ws,
+                    session_id,
+                    &mut rx_pcm_in,
+                    tx_audio_out.clone(),
+                    metrics.clone(),
+                    &mut close,
+                )
+                .await;
+
+                match pump_result {
+                    Ok(()) => {
+                        // Brain WS stream ended cleanly (e.g. brain sent bye).
+                        // Per spec §4.3 step 4, we still backoff + retry:
+                        // the brain may have restarted. Loop continues.
+                        info!(%session_id, "tap client closed gracefully; backing off");
+                    }
+                    Err(TapClientError::Closed) => {
+                        // Close signal from the binary (Channel::Closing).
+                        // `run_tap_client` already sent `bye` and closed
+                        // the WS; we're done — do NOT retry.
+                        info!(%session_id, "tap client closed via core signal; exiting");
+                        return;
+                    }
+                    Err(e) => {
+                        // WS/protocol error → backoff + retry per spec §4.3.
+                        warn!(error = ?e, %session_id, "tap client error; backing off");
+                    }
+                }
+            }
+            Err(e) => {
+                warn!(error = ?e, %session_id, "tap connect failed; backing off");
+            }
+        }
+
+        // === Step 3: backoff before retry (spec §4.3, §5.2). ===
+        metrics.reconnect_attempts.fetch_add(1, Ordering::Relaxed);
+        let delay = backoff.next_delay();
+        tokio::select! {
+            biased;
+            _ = &mut close => {
+                info!(%session_id, "close signal during backoff; exiting");
+                return;
+            }
+            _ = tokio::time::sleep(delay) => {}
+        }
+    }
+}
+
+/// Dial the brain. Returns the live `WebSocketStream` (no TLS — slice-2 is
+/// `ws://` loopback only per spec §4.4; `wss://` is rejected at
+/// `POST /v1/sessions` time before we ever get here).
+///
+/// `connect_async` returns `WebSocketStream<MaybeTlsStream<TcpStream>>` for
+/// `ws://` URLs (the `MaybeTlsStream` resolves to the no-TLS variant). The
+/// concrete type is `tokio_tungstenite::WebSocketStream<tokio_tungstenite::MaybeTlsStream<tokio::net::TcpStream>>`.
+/// We surface that explicit type rather than `impl Trait` because
+/// `run_tap_client` is generic over `T: AsyncRead + AsyncWrite + Unpin`,
+/// so a concrete type annotates + documents the wire shape for learners
+/// without obscuring it behind `impl`.
+///
+/// # Why `IntoClientRequest` (not just `connect_async(&url)`)
+///
+/// `connect_async` accepts anything that implements `IntoClientRequest`.
+/// `&Url` does, but going through `into_client_request()` first surfaces
+/// any URL-shape error before the network round-trip — easier to attribute
+/// (URL parse failure is a config error, not a network error).
+async fn connect_brain(
+    tap_url: &Url,
+) -> Result<
+    tokio_tungstenite::WebSocketStream<tokio_tungstenite::MaybeTlsStream<tokio::net::TcpStream>>,
+    TapClientError,
+> {
+    let request = tap_url
+        .as_str()
+        .into_client_request()
+        .map_err(TapClientError::Ws)?;
+    // Bounded dial — `tokio::time::timeout` returns `Result<Ok(inner),
+    // Elapsed>`; map the elapsed to a HandshakeTimeout-shaped Ws error so
+    // the engine loop's backoff path sees it through the same `Err(e)`
+    // arm as a true WS failure (one code path, two causes).
+    let connect_fut = tokio_tungstenite::connect_async(request);
+    let ws_result = tokio::time::timeout(CONNECT_TIMEOUT, connect_fut)
+        .await
+        .map_err(|_| TapClientError::Ws(tokio_tungstenite::tungstenite::Error::AlreadyClosed))?;
+    let (ws, _response) = ws_result?;
+    Ok(ws)
+}
+
+/// Bounded exponential backoff (spec §4.3, §5.2):
+/// 250 ms → 500 ms → 1 s → 2 s → cap at 5 s. Infinite retries.
+///
+/// # Why not `std::time::Duration::saturating_mul`?
+///
+/// `Duration * 2` panics on overflow (Rust 1.x); `.min(cap)` keeps us
+/// bounded. The explicit cap at 5 s means even after 1e6 retries the
+/// delay stays at 5 s — important for the infinite-retry posture (spec
+/// §5.2: "the brain WILL come back; the call SHOULD outlast the brain
+/// outage").
+struct Backoff {
+    count: u32,
+    current: Duration,
+}
+
+impl Default for Backoff {
+    fn default() -> Self {
+        Self {
+            count: 0,
+            current: Duration::from_millis(250),
+        }
+    }
+}
+
+impl Backoff {
+    /// Return the current delay and advance the backoff state.
+    /// Doubles up to the 5 s cap and stays there — infinite retries.
+    fn next_delay(&mut self) -> Duration {
+        let d = self.current;
+        // Saturating `min` here guarantees we never exceed the 5 s cap.
+        // `(current * 2).min(cap)` is the standard "double-up-to-cap" shape.
+        self.current = (self.current * 2).min(Duration::from_secs(5));
+        self.count += 1;
+        d
+    }
+    /// Reset after a successful connect + handshake.
+    fn reset(&mut self) {
+        self.count = 0;
+        self.current = Duration::from_millis(250);
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use rutster_media::AudioSource;
+
+    #[test]
+    fn backoff_doubles_until_cap() {
+        let mut b = Backoff::default();
+        assert_eq!(b.next_delay(), Duration::from_millis(250));
+        assert_eq!(b.next_delay(), Duration::from_millis(500));
+        assert_eq!(b.next_delay(), Duration::from_secs(1));
+        assert_eq!(b.next_delay(), Duration::from_secs(2));
+        assert_eq!(b.next_delay(), Duration::from_secs(4));
+        // Cap reached: stays at 5s thereafter.
+        assert_eq!(b.next_delay(), Duration::from_secs(5));
+        assert_eq!(b.next_delay(), Duration::from_secs(5));
+    }
+
+    #[test]
+    fn backoff_reset_returns_to_floor() {
+        let mut b = Backoff::default();
+        let _ = b.next_delay();
+        let _ = b.next_delay();
+        b.reset();
+        assert_eq!(b.next_delay(), Duration::from_millis(250));
+    }
+
+    #[tokio::test]
+    async fn spawn_returns_pipe_with_live_mpsc_ends() {
+        // Smoke test: spawning the engine immediately returns a TapAudioPipe
+        // the binary can wire into RtcSession. We don't actually wire a brain
+        // here — the engine task will spin connect_brain (failing on a
+        // non-existent server) and back off; the metrics counter should
+        // increment. We abort the task on test drop to avoid leak.
+        let id = ChannelId::new();
+        let url = Url::parse("ws://127.0.0.1:1/echo").unwrap(); // port 1 = unreachable
+        let (mut pipe, conn) = spawn_tap_engine(id, url);
+        // TapAudioPipe is the seam object — should default to silent underflow.
+        assert!(pipe.next_pcm_frame().is_none());
+        // TapConn carries the close oneshot + JoinHandle + metrics.
+        let _ = conn.close_tx.send(());
+        conn.join.abort();
+    }
+}
diff --git a/crates/rutster/tests/api_integration.rs b/crates/rutster/tests/api_integration.rs
index a71695a..377824b 100644
--- a/crates/rutster/tests/api_integration.rs
+++ b/crates/rutster/tests/api_integration.rs
@@ -10,9 +10,13 @@ use axum::http::{Request, StatusCode};
 use rutster::session_map::AppState;
 use tower::ServiceExt; // enables `oneshot` on the Router for sync tests
 
+fn default_tap_url() -> url::Url {
+    url::Url::parse("ws://127.0.0.1:8081/echo").unwrap()
+}
+
 #[tokio::test]
 async fn post_v1_sessions_returns_a_session_id() {
-    let app = rutster::routes::router(AppState::new());
+    let app = rutster::routes::router(AppState::new(default_tap_url()));
     let resp = app
         .oneshot(
             Request::builder()
@@ -33,7 +37,7 @@ async fn post_v1_sessions_returns_a_session_id() {
 
 #[tokio::test]
 async fn get_root_serves_html() {
-    let app = rutster::routes::router(AppState::new());
+    let app = rutster::routes::router(AppState::new(default_tap_url()));
     let resp = app
         .oneshot(Request::builder().uri("/").body(Body::empty()).unwrap())
         .await