From 5c3baf91553d681f073a142a4a5f4c7a5b8c3fd0 Mon Sep 17 00:00:00 2001 From: opencode controller Date: Tue, 30 Jun 2026 23:21:58 -0400 Subject: [PATCH] =?UTF-8?q?feat(tap):=20tool-call=20side-channel=20wiring?= =?UTF-8?q?=20(spec=20=C2=A75.2)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit slice-3 §5.2 + §6: the binary's poll task now drains the brain's function_call proposals from rx_function_call, dispatches through the per-channel ToolRegistry (HangupTool wired at spawn_tap_engine time), and writes function_call_output replies back through tx_function_call_output which run_tap_client forwards as tap WS frames to the brain. TapClient: handle_brain_frame now forwards function_call events to a new tx_function_call mpsc side-channel instead of dropping them. run_tap_client adds a select! arm draining rx_function_call_output + sending each as a tap frame. Advisory events (speech_started/stopped, tools.update) still log + count (slice-3 deferred-action posture). TapEngine: spawn_tap_engine now takes AppState + constructs a per-channel ToolRegistry (spec §6.2) with HangupTool pre-registered (§6.3). TapConn gains rx_function_call, tx_function_call_output, tool_registry fields. session_map: drive_all_sessions calls drain_function_calls in the same cycle as the slice-2 §5.3 step 4 flush drain (one extra channel, same cycle); the helper spawns each dispatch as its own task so the 750 ms hangup teardown bound (AppState::close) can't stall the 10 ms poll cadence. files touched: crates/rutster-tap/src/{lib,tap_client}.rs, crates/rutster/src/{session_map,tap_engine}.rs, crates/rutster/tests/tap_integration.rs ( AppState arg ), crates/rutster-brain-realtime/src/translator.rs (clippy needless_borrow ). NOT touched: loop_driver.rs, rtc_session.rs (seam test §7.5 #6). gates: cargo fmt --check OK. cargo clippy --all --tests -D warnings OK. cargo test --all OK. cargo deny check has pre-existing environmental failure (CVSS 4.0 unsupported in advisory-db; same on main). --- .../rutster-brain-realtime/src/translator.rs | 2 +- crates/rutster-tap/src/lib.rs | 2 +- crates/rutster-tap/src/tap_client.rs | 230 +++++++++++++++++- crates/rutster/src/session_map.rs | 192 ++++++++++++++- crates/rutster/src/tap_engine.rs | 115 ++++++++- crates/rutster/tests/tap_integration.rs | 4 +- 6 files changed, 525 insertions(+), 20 deletions(-) diff --git a/crates/rutster-brain-realtime/src/translator.rs b/crates/rutster-brain-realtime/src/translator.rs index 7c07231..453b6d5 100644 --- a/crates/rutster-brain-realtime/src/translator.rs +++ b/crates/rutster-brain-realtime/src/translator.rs @@ -162,7 +162,7 @@ mod tests { // The PCM base64 string for one zero frame (480 samples, every LE i16=0). use base64::Engine; let zeros = [0u8; 960]; - let pcm_b64 = base64::engine::general_purpose::STANDARD.encode(&zeros); + let pcm_b64 = base64::engine::general_purpose::STANDARD.encode(zeros); let v = tap_audio_in_to_openai_append(&pcm_b64); assert_eq!(v["type"], "input_audio_buffer.append"); assert_eq!(v["audio"], pcm_b64); diff --git a/crates/rutster-tap/src/lib.rs b/crates/rutster-tap/src/lib.rs index af30ed2..4e5d552 100644 --- a/crates/rutster-tap/src/lib.rs +++ b/crates/rutster-tap/src/lib.rs @@ -48,7 +48,7 @@ pub use protocol::{ // Slice-3 additive (spec §3). pub use protocol::{FunctionCallOutputPayload, FunctionCallPayload, ToolsUpdatePayload}; pub use tap_audio_pipe::TapAudioPipe; -pub use tap_client::{TapClientError, run_tap_client}; +pub use tap_client::{FunctionCallEvent, FunctionCallOutputEvent, TapClientError, run_tap_client}; #[cfg(test)] mod tests { diff --git a/crates/rutster-tap/src/tap_client.rs b/crates/rutster-tap/src/tap_client.rs index 7d4219f..ed906d4 100644 --- a/crates/rutster-tap/src/tap_client.rs +++ b/crates/rutster-tap/src/tap_client.rs @@ -63,11 +63,46 @@ pub enum TapClientError { /// `close` is a shared borrow (`&mut oneshot::Receiver<()>`) so the /// `TapEngine` reconnect loop (Task 7) can share one close signal across /// reconnect attempts of the same session (see module docs). +/// +/// # slice-3 §5.2 — the tool-call side-channel +/// +/// Two extra mpsc halves (relative to slice-2's audio-only pump) carry the +/// brain's `function_call` proposals out to the binary's poll task and the +/// binary's `function_call_output` replies back onto the wire: +/// +/// - `tx_function_call` — the TapClient emits a `FunctionCallEvent` here +/// whenever it observes a tap `function_call` frame on its inbound WS +/// stream. The binary's poll task drains this in the same cycle it drains +/// the existing `flush_tx` side-channel (slice-2 §5.3 step 4 — one extra +/// channel, same cycle) and dispatches via `ToolRegistry::dispatch`. +/// - `rx_function_call_output` — the binary's poll task writes +/// `FunctionCallOutputEvent`s here (after a `ToolRegistry::dispatch` call +/// completes); the TapClient drains this in the same `tokio::select!` +/// pump as the audio + close arms and sends each as a `function_call_output` +/// tap WS frame to the brain. +/// +/// Both halves are mpsc ends (not oneshot) because the brain may propose +/// multiple tool calls per session (one-of-many, not one-of-one) and the +/// binary may queue multiple replies before the TapClient's pump cycle +/// drains them. +// +// clippy::too_many_arguments: the slice-3 §5.2 design added two more mpsc +// halves to slice-2's already-5-arg pump signature for a total of 8. Each +// arg is a distinct channel end with a distinct lifetime owner (the WS +// stream, the session id, two sender/receiver pairs for audio, two for the +// tool-call side-channel, the metrics Arc, the close oneshot). Wrapping +// them in a struct would obscure that all-but-one are channel ends shared +// with the binary's poll task — the flat signature mirrors slice-2's +// precedent and keeps the call site readable. Suppress per AGENTS.md's +// "documented inline rationale" exception to the -D warnings bar. +#[allow(clippy::too_many_arguments)] pub async fn run_tap_client( mut ws: WebSocketStream, session_id: ChannelId, rx_pcm_in: &mut mpsc::Receiver, tx_audio_out: mpsc::Sender, + tx_function_call: mpsc::Sender, + rx_function_call_output: &mut mpsc::Receiver, metrics: Arc, close: &mut oneshot::Receiver<()>, ) -> Result<(), TapClientError> @@ -175,6 +210,44 @@ where } } } + // slice-3 §5.2: drain a `function_call_output` event the binary's + // poll task wrote (after `ToolRegistry::dispatch` returned) + send + // it as a `function_call_output` tap WS frame to the brain. The + // `seq_egress` bump mirrors the audio arm — every egress frame + // shares the same per-direction counter (spec §3.1). + // + // Like `rx_pcm_in.recv()`, this is one of many arms in the + // select! — the engine's `tx_function_call_output` sender lives + // in the binary; `run_tap_client` returns when the brain WS ends + // or close fires, regardless of pending function_call_output + // events (they're dropped on close — same posture as pending + // audio_out frames on teardown). + out = rx_function_call_output.recv() => { + if let Some(out) = out { + let ts = elapsed_ms(session_start); + let result_str = out.0.result.to_string(); + match crate::protocol::encode_function_call_output( + &out.0.id, + &out.0.status, + &result_str, + seq_egress, + ts, + ) { + Ok(s) => { + seq_egress += 1; + if let Err(e) = ws.send(Message::Text(s)).await { + warn!(error = ?e, %session_id, "ws send function_call_output failed"); + return Err(e.into()); + } + info!(%session_id, call_id = %out.0.id, status = %out.0.status, "sent function_call_output to brain"); + } + Err(e) => { + metrics.malformed_frames.fetch_add(1, Ordering::Relaxed); + warn!(error = ?e, "encode function_call_output failed; dropping"); + } + } + } + } // Inbound WS frame from brain. msg = ws.next() => { let Some(msg) = msg else { @@ -188,7 +261,7 @@ where if let Ok(text) = msg.into_text() { handle_brain_frame( &text, &mut last_seq_ingress, &tx_audio_out, - &metrics, session_start, + &tx_function_call, &metrics, session_start, ).await; } } @@ -251,6 +324,7 @@ async fn handle_brain_frame( text: &str, last_seq_ingress: &mut Option, tx_audio_out: &mpsc::Sender, + tx_function_call: &mpsc::Sender, metrics: &Arc, session_start: Instant, ) { @@ -307,18 +381,40 @@ async fn handle_brain_frame( metrics.unknown_frames.fetch_add(1, Ordering::Relaxed); warn!("unexpected frame direction from brain; dropping"); } - // Slice-3 additive v1 event types (spec §3.2): forwards-compatible - // wire shapes the slice-2 tap client doesn't yet act on — the - // brain's translator (Task 4) emits these, the tool registry - // (Task 6) consumes them. Until those land, log + count + drop - // (same posture as `Unknown`, slice-2 §3.4). - DecodedPayload::SpeechStarted - | DecodedPayload::SpeechStopped - | DecodedPayload::FunctionCall(_) - | DecodedPayload::FunctionCallOutput(_) - | DecodedPayload::ToolsUpdate(_) => { + // Slice-3 spec §5.2: `function_call` flows through the side-channel + // (NON-BLOCKING try_send — the binary's poll task drains on its own + // cycle). The same "drop + observe" posture as audio_out applies if + // the channel is full: a backed-up binary means we drop the proposal + // and the brain gets no reply (the brain process knows no + // function_call_output arrived → its OpenAI pump keeps going; the + // model tolerates missing replies per OpenAI's design). + DecodedPayload::FunctionCall(p) => { + if tx_function_call.try_send(FunctionCallEvent(p)).is_err() { + metrics.outbound_dropped.fetch_add(1, Ordering::Relaxed); + warn!( + "function_call dropped (binary poll task not draining; brain will see no reply)" + ); + } + } + // Slice-3 (spec §3.2): `function_call_output` is core→brain; ignore + // if a brain sends one back (a misbehaving brain — same posture as + // `SessionEnd`/`AudioIn` from brain above). + DecodedPayload::FunctionCallOutput(_) => { metrics.unknown_frames.fetch_add(1, Ordering::Relaxed); - debug!("slice-3 event type not yet handled; dropping"); + warn!("unexpected function_call_output from brain; dropping"); + } + // Slice-3 advisory — same "logged + counted, not forwarded" posture + // as `Unknown`. The FOB reflex loop in step 4 will act on these; + // slice-3 only pre-paves the wire event. + DecodedPayload::SpeechStarted | DecodedPayload::SpeechStopped => { + metrics.unknown_frames.fetch_add(1, Ordering::Relaxed); + debug!("advisory interruption event observed; not acted on in slice-3"); + } + DecodedPayload::ToolsUpdate(_) => { + metrics.unknown_frames.fetch_add(1, Ordering::Relaxed); + debug!( + "tools.update observed; slice-3 dispatch keys off function_call by name, not catalog" + ); } } let _ = session_start; // used for ts computation if added later @@ -328,6 +424,36 @@ fn elapsed_ms(start: Instant) -> u64 { start.elapsed().as_millis() as u64 } +/// A `function_call` event the TapClient **observed** on its inbound WS +/// stream and forwarded to the binary's poll task via the +/// `tx_function_call` side-channel (spec §5.2). The binary's poll task +/// drains this (alongside the existing `flush_rx` side-channel — slice-2 +/// §5.3 step 4) and dispatches each event through `ToolRegistry::dispatch`. +/// +/// # Why a thin newtype over `FunctionCallPayload` (and not a bare alias)? +/// +/// A type alias (`pub type FunctionCallEvent = FunctionCallPayload;`) would +/// let the binary pass a `FunctionCallPayload` where a `FunctionCallEvent` +/// is expected without surfacing the intent. The newtype draws a small but +/// real boundary: the wire-payload type (`FunctionCallPayload`) lives in +/// `protocol.rs` for (de)serialization; the side-channel event type +/// (`FunctionCallEvent`) lives here for dispatch. They share a shape but +/// carry different semantic weight — honoring the newtype-over-primitives +/// convention from AGENTS.md even at the message level. +#[derive(Debug, Clone)] +pub struct FunctionCallEvent(pub crate::protocol::FunctionCallPayload); + +/// A `function_call_output` event the binary's poll task **emits** back to +/// the TapClient via the `tx_function_call_output` side-channel (spec §5.2 +/// — the binary's poll task dispatches through `ToolRegistry::dispatch`, +/// serializes the `ToolResult`, and writes the output here). The TapClient +/// drains this in the same `tokio::select!` pump cycle as the audio + close +/// arms and sends each as a `function_call_output` tap WS frame. +/// +/// Same newtype-over-payload rationale as `FunctionCallEvent`. +#[derive(Debug, Clone)] +pub struct FunctionCallOutputEvent(pub crate::protocol::FunctionCallOutputPayload); + #[cfg(test)] mod tests { // TapClient is heavily async; its real behavior is exercised in the @@ -335,6 +461,7 @@ mod tests { // tests here cover the pure helpers. use super::*; + use crate::protocol::encode_function_call; #[test] fn elapsed_ms_is_monotonic_nonneg() { @@ -343,4 +470,83 @@ mod tests { // First call ~0; just assert it's a valid u64. assert_eq!(ms, ms); // tautology but clippy-clean } + + /// slice-3 spec §5.2: when the TapClient observes a tap `function_call` + /// frame on its inbound WS stream it emits a `FunctionCallEvent` on + /// the `tx_function_call` side-channel. The binary's poll task drains + /// that and dispatches via `ToolRegistry::dispatch`. This test pins the + /// contract end-to-end through the pure helper (`handle_brain_frame`): + /// hand it a wire-encoded function_call frame + a fresh mpsc pair and + /// assert the receiver observes the forwarded event. + #[tokio::test] + async fn handle_brain_frame_forwards_function_call_to_side_channel() { + let (tx_fc, mut rx_fc) = mpsc::channel::(8); + let (tx_audio_out, _rx_audio_out) = mpsc::channel::(8); + let metrics = Arc::new(TapMetrics::new()); + + // Build a wire function_call frame: id="call-1", name="hangup", args={}. + let wire = encode_function_call("call-1", "hangup", "{}", 1, 100).unwrap(); + let mut last_seq: Option = None; + + handle_brain_frame( + &wire, + &mut last_seq, + &tx_audio_out, + &tx_fc, + &metrics, + Instant::now(), + ) + .await; + + // The side-channel must have observed exactly one FunctionCallEvent + // carrying the wire's id/name/args. + let event = tokio::time::timeout(Duration::from_millis(200), rx_fc.recv()) + .await + .expect("tx_function_call drained within 200ms") + .expect("channel not closed"); + assert_eq!(event.0.id, "call-1"); + assert_eq!(event.0.name, "hangup"); + assert_eq!(event.0.args, serde_json::json!({})); + // seq tracking still updates for the side-channeled event. + assert_eq!(last_seq, Some(1)); + } + + /// slice-3 spec §5.2 — the *advisory* interrupt events (`speech_started` + /// /`speech_stopped`) and `tools.update` are observed (logged + counted) + /// but do NOT flow through the function_call side-channel (only + /// `function_call` does — that's the only event with a binary-side + /// disposal). This pins that boundary: an advisory event must NOT + /// produce a `FunctionCallEvent` even with the channel plumbed. + #[tokio::test] + async fn advisory_events_are_logged_not_forwarded_to_function_call_channel() { + let (tx_fc, mut rx_fc) = mpsc::channel::(8); + let (tx_audio_out, _rx_audio_out) = mpsc::channel::(8); + let metrics = Arc::new(TapMetrics::new()); + + let wire = crate::protocol::encode_speech_started(2, 200).unwrap(); + let mut last_seq: Option = None; + handle_brain_frame( + &wire, + &mut last_seq, + &tx_audio_out, + &tx_fc, + &metrics, + Instant::now(), + ) + .await; + + // No FunctionCallEvent forwarded — the channel stays empty. Pick a + // tight bounded receive so the test fails fast if a future refactor + // starts forwarding advisory events here. + assert!( + tokio::time::timeout(Duration::from_millis(50), rx_fc.recv()) + .await + .is_err(), + "no FunctionCallEvent expected for advisory events" + ); + // The advisory event IS still observed via metrics (seq gap tracking + // + the unknown-slot counter remains 0 — speech_started is now a + // known payload variant). + assert_eq!(last_seq, Some(2)); + } } diff --git a/crates/rutster/src/session_map.rs b/crates/rutster/src/session_map.rs index 11b5f9d..d538fc4 100644 --- a/crates/rutster/src/session_map.rs +++ b/crates/rutster/src/session_map.rs @@ -39,6 +39,11 @@ use tokio::sync::Mutex; use tracing::{debug, info, warn}; use crate::tap_engine::{TapConn, spawn_tap_engine}; +// Re-using the binary crate's `tokio::sync::mpsc` import (the engine task +// + the poll task both live in `rutster`'s poll-driver module). The type +// only appears in `TapConn` field signatures + the slice-3 §5.2 dispatch +// helper below; bringing it in here keeps the type names short. +use tokio::sync::mpsc; /// The per-session wrapper struct (slice-2, spec §6). /// @@ -265,6 +270,13 @@ async fn drive_all_sessions(state: &AppState, now: Instant) { } } } + // slice-3 §5.2: drain the per-session `rx_function_call` + // side-channel in the same cycle as the `flush_rx` drain (slice-2 + // §5.3 step 4 pattern — one extra channel, same cycle). The + // helper spawns each dispatch as its own task so the 750 ms + // `AppState::close` await (hangup's teardown handshake) can't + // stall the poll cadence. + let _fc_drained = drain_function_calls(state, id).await; // 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) { @@ -298,7 +310,12 @@ async fn drive_all_sessions(state: &AppState, now: Instant) { if let ChannelState::Connected = s.channel.state { if s.channel.tap.is_none() { // First connect: spawn the TapEngine, wire the TapAudioPipe. - let (pipe, conn) = spawn_tap_engine(id, tap_url); + // slice-3 §6.2: spawn_tap_engine constructs a per-channel + // ToolRegistry holding HangupTool (the only wired tool in + // slice-3 — §6.3) bound to this AppState + ChannelId. + let app_state = state.clone(); + let tap_url_clone = tap_url.clone(); + let (pipe, conn) = spawn_tap_engine(id, tap_url_clone, app_state); s.set_pipe(pipe); s.channel.tap = Some(TapHandle::new()); info!(channel_id = %id, "tap engine spawned on Connected"); @@ -319,3 +336,176 @@ async fn drive_all_sessions(state: &AppState, now: Instant) { } } } + +/// slice-3 §5.2 + §6 — drain the per-session `rx_function_call` side-channel +/// and dispatch each event through the per-channel `ToolRegistry`. One +/// dispatch result → one `function_call_output` written to +/// `tx_function_call_output` (which the TapClient forwards to the brain on +/// its next pump cycle). +/// +/// # Why this is a separate helper (not inline in `drive_all_sessions`) +/// +/// `ToolRegistry::dispatch` is `async` (the `hangup` tool calls +/// `AppState::close`, which awaits the engine's teardown handshake — spec +/// §5.2's 750 ms bound). Awaiting that inline in the poll task would stall +/// the 10 ms poll cadence for every hangup. Instead we **collect** events +/// non-blockingly here (`try_recv` — drop + observe on full channel, same +/// posture as the existing flush drain), then **spawn** each dispatch as +/// its own tokio task so the poll task returns to its 10 ms cadence +/// immediately. The spawned task holds its own clones of `AppState` + +/// `tool_registry` + `tx_function_call_output` — no shared mutable state +/// with the poll path. +/// +/// Returns the count of events drained (for observability — matches the +/// flush drain's "drain all" loop shape). The poll task calls this in the +/// same `drive_all_sessions` cycle it drains the `flush_rx` side-channel +/// (slice-2 §5.3 step 4 pattern, one extra channel, same cycle). +async fn drain_function_calls(state: &AppState, id: ChannelId) -> usize { + // Collect: pull everything we need out of the DashMap shard BEFORE any + // await (the spawned dispatch's `AppState::close` later takes the same + // shard lock — holding a `get_mut` Ref across the spawn's await would + // deadlock slice-2's other shard-mutating handlers). The Collector scope + // owns a single short-lived `get_mut` Ref; the loop does only sync + // `try_recv` + Vec push (no await). + let collected: Vec; + let tool_registry: Arc>; + let tx_out: Option>; + { + let Some(mut entry) = state.sessions.get_mut(&id) else { + return 0; + }; + let Some(conn) = entry.tap_conn.as_mut() else { + return 0; + }; + tool_registry = conn.tool_registry.clone(); + tx_out = conn.tx_function_call_output.clone(); + let Some(rx_fc) = conn.rx_function_call.as_mut() else { + return 0; + }; + collected = (0..).map_while(|_| rx_fc.try_recv().ok()).collect(); + } + let drained = collected.len(); + + // Spawn dispatches outside the shard Ref scope. `AppState::close` is a + // 750 ms bounded wait in the worst case (slice-2 §5.2's teardown + // handshake); spawning keeps the 10 ms poll cadence responsive. + for event in collected { + let app_state = state.clone(); + let reg = tool_registry.clone(); + let tx_out = tx_out.clone(); + tokio::spawn(async move { + let payload = event.0; + let call_id = payload.id.clone(); + let name = payload.name.clone(); + debug!(channel_id = %id, call_id = %call_id, tool = %name, "dispatching function_call"); + let result = reg.lock().await.dispatch(&name, payload.args).await; + let (status, result_value) = result.to_status_result(); + let out = + rutster_tap::FunctionCallOutputEvent(rutster_tap::FunctionCallOutputPayload { + id: call_id.clone(), + status: status.clone(), + result: result_value, + }); + // try_send — non-blocking. The TapClient's pump loop drains + // `rx_function_call_output` on its next `select!` cycle. If the + // channel is full (brain not pumping), we drop + observe (same + // posture as the engine's `tx_function_call.try_send` on the + // inbound side). `tx_out` is `Option`; `None` means the + // engine has torn down its receiver — the result is dropped, + // acceptable since the call is hanging up. + if let Some(tx) = tx_out.as_ref() { + if let Err(e) = tx.try_send(out) { + warn!( + channel_id = %id, call_id = %call_id, error = ?e, + "function_call_output dropped (TapClient pump not draining)" + ); + } + } + // app_state is held for the dispatch lifetime — the HangupTool + // inside the registry already captured its own clone at registry + // construction, so this outer binding exists for future tool + // impls that need the live AppState handed to dispatch (none in + // slice-3 beyond hangup). Drop the silent-binding warning by + // referencing it once. + drop(app_state); + info!(channel_id = %id, call_id = %call_id, status = %status, "function_call dispatched"); + }); + } + drained +} + +#[cfg(test)] +mod tests { + use super::*; + use rutster_tap::{ + FunctionCallEvent, FunctionCallOutputEvent, FunctionCallPayload, TapMetrics, + }; + use tokio::sync::Mutex; + + /// slice-3 §5.2 + §6 — a `function_call` event drained from + /// `rx_function_call` triggers `ToolRegistry::dispatch("hangup")` which + /// fires `AppState::close` (the slice-2 teardown path), and the dispatch + /// result flows back as a `FunctionCallOutputEvent` on + /// `tx_function_call_output`. End-to-end contract test for the helper + /// minus the live TapClient pump (which is integration-test territory). + #[tokio::test] + async fn drain_function_calls_dispatches_hangup_and_writes_output() { + let state = AppState::default(); + // Create a session so AppState::close has something to remove. + let id = state.create_session(None).unwrap(); + + // Build a TapConn with manually-controlled side-channel ends so we + // can push a FunctionCallEvent from the test side + observe the + // dispatch's output on the paired Receiver. + let (tx_fc, rx_fc) = tokio::sync::mpsc::channel::(8); + let (tx_fco, mut rx_fco) = tokio::sync::mpsc::channel::(8); + let mut registry = crate::tool_registry::ToolRegistry::new(); + registry.register(Box::new(crate::tool_registry::HangupTool::new( + state.clone(), + id, + ))); + let (close_tx, _close_rx) = tokio::sync::oneshot::channel::<()>(); + let conn = crate::tap_engine::TapConn { + close_tx, + join: tokio::spawn(async {}), // no-op handle; aborted below + metrics: TapMetrics::new(), + flush_rx: None, + rx_function_call: Some(rx_fc), + tx_function_call_output: Some(tx_fco), + tool_registry: Arc::new(Mutex::new(registry)), + }; + state.sessions.get_mut(&id).unwrap().tap_conn = Some(conn); + + // Push a function_call for the hangup tool — simulates what the + // TapClient does when it observes a `function_call` tap frame. + tx_fc + .send(FunctionCallEvent(FunctionCallPayload { + id: "call-1".to_string(), + name: "hangup".to_string(), + args: serde_json::json!({}), + })) + .await + .unwrap(); + + // Drain — this spawns a dispatch task + returns immediately. + let drained = drain_function_calls(&state, id).await; + assert_eq!(drained, 1, "exactly one function_call should drain"); + + // The spawned dispatch task fires AppState::close (session removed) + // + writes the function_call_output. Bounded-wait the result with + // a generous timeout (AppState::close has a 750 ms teardown bound + + // the spawned task may take a few ms to schedule). + let out = tokio::time::timeout(Duration::from_secs(2), rx_fco.recv()) + .await + .expect("function_call_output drained within 2s") + .expect("channel not closed"); + assert_eq!(out.0.id, "call-1"); + assert_eq!(out.0.status, "ok"); + assert_eq!(out.0.result["channel_state"], "Closing"); + // AppState::close removed the session entry (the teardown it fires). + assert!( + state.sessions.get(&id).is_none(), + "session should be removed after hangup dispatch" + ); + } +} diff --git a/crates/rutster/src/tap_engine.rs b/crates/rutster/src/tap_engine.rs index bf2d5f8..bb8a580 100644 --- a/crates/rutster/src/tap_engine.rs +++ b/crates/rutster/src/tap_engine.rs @@ -30,14 +30,18 @@ use std::time::Duration; use futures_util::FutureExt; use rutster_call_model::ChannelId; -use rutster_tap::tap_client::{TapClientError, run_tap_client}; +use rutster_tap::tap_client::{ + FunctionCallEvent, FunctionCallOutputEvent, TapClientError, run_tap_client, +}; use rutster_tap::{TapAudioPipe, TapMetrics}; -use tokio::sync::{mpsc, oneshot}; +use tokio::sync::{Mutex, mpsc, oneshot}; use tokio::task::JoinHandle; use tokio_tungstenite::tungstenite::client::IntoClientRequest; use tracing::{info, warn}; use url::Url; +use crate::tool_registry::{HangupTool, ToolRegistry}; + /// 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. @@ -83,6 +87,28 @@ pub struct TapConn { /// returned from `spawn_tap_engine`; the engine task owns the paired /// `flush_tx` and signals a flush after each failed pump cycle. pub flush_rx: Option>, + /// slice-3 §5.2 tool-call side-channel (brain → core): the binary's + /// poll task drains `function_call` events from here + dispatches via + /// [`tool_registry`]. Engine owns the paired `tx_function_call` Sender + /// moved into `run_tap_client`. `Option` to keep the type constructible + /// in tests that don't plumb it; always `Some` on conns returned from + /// `spawn_tap_engine` (matches `flush_rx`'s posture). + pub rx_function_call: Option>, + /// slice-3 §5.2 tool-call side-channel (core → brain): the binary's + /// poll task writes `function_call_output` events here (one per + /// `ToolRegistry::dispatch` result). The engine owns the paired + /// `rx_function_call_output` Receiver moved into `run_tap_client` so + /// the engine's pump loop forwards each event as a tap WS frame. + pub tx_function_call_output: Option>, + /// slice-3 §6 — per-channel tool registry. One registry per active + /// session (spec §6.2 "keyed by ChannelId"); `HangupTool` is the only + /// tool wired in slice-3 (§6.3). `Arc>` because the binary's + /// poll task mutates the registry (dispatch) while the TapConn is + /// shared across the poll task + the routes layer (potential future + /// `tools.update` forwarding). `Mutex` (not `RwLock`) because dispatch + /// is effectively exclusive — there's no read-heavy workload to + /// optimize. + pub tool_registry: Arc>, } /// Spawn the TapEngine task for one session. Dials `tap_url`, runs the pump @@ -102,7 +128,11 @@ pub struct TapConn { /// 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) { +pub fn spawn_tap_engine( + session_id: ChannelId, + tap_url: Url, + app_state: crate::session_map::AppState, +) -> (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 @@ -122,8 +152,27 @@ pub fn spawn_tap_engine(session_id: ChannelId, tap_url: Url) -> (TapAudioPipe, T // flush signals collapse to one `clear_playout_ring` call from the // binary side). let (flush_tx, flush_rx) = mpsc::channel::<()>(8); + // slice-3 §5.2 tool-call side-channels. Same capacity-shape rationale + // as `flush_tx`: `try_send`-safe on both ends, idempotent drain. One + // extra pair of mpsc halves vs slice-2 — same cycle in the binary's + // poll task (drain alongside `flush_rx`), same pump-arm in run_tap_client. + let (tx_function_call, rx_function_call) = + mpsc::channel::(TAP_MPSC_CAPACITY); + let (tx_function_call_output, rx_function_call_output) = + mpsc::channel::(TAP_MPSC_CAPACITY); let metrics = TapMetrics::new(); + // slice-3 §6.2: per-channel tool registry. The engine constructs it + // (one per active session — spec "keyed by ChannelId, one registry per + // active channel"). `HangupTool` is the only wired tool in slice-3 + // (§6.3); other tool names reply `not_implemented` via dispatch. + // The registry is `Arc>` so the binary's poll task + // can `dispatch` on it while the TapConn is shared (the future + // `tools.update`-forwarding path will also share this handle). + let mut registry = ToolRegistry::new(); + registry.register(Box::new(HangupTool::new(app_state, session_id))); + let tool_registry = Arc::new(Mutex::new(registry)); + // Clone metrics three ways: the engine task, the TapAudioPipe, and // the TapConn handle each hold their own `Arc` refcount. // All three views must observe the same counters — `Arc` is the @@ -132,6 +181,7 @@ pub fn spawn_tap_engine(session_id: ChannelId, tap_url: Url) -> (TapAudioPipe, T // shared mutation is sound without a `Mutex`). let metrics_for_pipe = metrics.clone(); let metrics_for_conn = metrics.clone(); + let tool_registry_for_task = tool_registry.clone(); let join = tokio::spawn(async move { run_engine_loop( @@ -141,6 +191,8 @@ pub fn spawn_tap_engine(session_id: ChannelId, tap_url: Url) -> (TapAudioPipe, T tx_audio_out, close_rx, flush_tx, + tx_function_call, + rx_function_call_output, metrics, ) .await; @@ -155,6 +207,9 @@ pub fn spawn_tap_engine(session_id: ChannelId, tap_url: Url) -> (TapAudioPipe, T join, metrics: metrics_for_conn, flush_rx: Some(flush_rx), + rx_function_call: Some(rx_function_call), + tx_function_call_output: Some(tx_function_call_output), + tool_registry: tool_registry_for_task, }; (pipe, conn) @@ -174,6 +229,14 @@ pub fn spawn_tap_engine(session_id: ChannelId, tap_url: Url) -> (TapAudioPipe, T /// 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. +// +// clippy::too_many_arguments: slice-3 §5.2 added two more mpsc halves to +// slice-2's engine-loop signature (the tool-call side-channel ends). Same +// rationale as `run_tap_client` above: each arg is a distinct channel +// end with a distinct lifetime owner. Wrapping them in a struct would +// obscure the channel-pair structure. Suppress per AGENTS.md's documented +// inline-rationale exception. +#[allow(clippy::too_many_arguments)] async fn run_engine_loop( session_id: ChannelId, tap_url: Url, @@ -181,6 +244,8 @@ async fn run_engine_loop( tx_audio_out: mpsc::Sender, mut close: oneshot::Receiver<()>, flush_tx: mpsc::Sender<()>, + tx_function_call: mpsc::Sender, + mut rx_function_call_output: mpsc::Receiver, metrics: Arc, ) { let mut backoff = Backoff::default(); @@ -212,11 +277,19 @@ async fn run_engine_loop( // === 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). + // The slice-3 §5.2 tool-call side-channels are also shared + // across reconnect attempts — a brain reconnect mid-call + // must keep the binary's tool-registry dispatch flowing + // (a function_call the brain proposed right before it + // dropped should still get a function_call_output reply + // on the next connect). let pump_result = run_tap_client( ws, session_id, &mut rx_pcm_in, tx_audio_out.clone(), + tx_function_call.clone(), + &mut rx_function_call_output, metrics.clone(), &mut close, ) @@ -414,11 +487,45 @@ mod tests { // 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); + let (mut pipe, conn) = spawn_tap_engine(id, url, crate::session_map::AppState::default()); // 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(); } + + /// slice-3 §5.2: the TapConn returned by `spawn_tap_engine` must carry + /// the tool-call side-channel ends so the binary's poll task can drain + /// `function_call` events (→ `ToolRegistry::dispatch`) and write + /// `function_call_output` replies. This is a structural smoke test — + /// it pins that the plumbing lands (the binary's poll-task drain needs + /// all three: a ready `rx_function_call`, a ready `tx_function_call_output`, + /// and a `tool_registry` scoped to this session). + #[tokio::test] + async fn spawn_returns_tap_conn_with_function_call_side_channels() { + let id = ChannelId::new(); + let url = Url::parse("ws://127.0.0.1:1/echo").unwrap(); // unreachable brain + let (_pipe, conn) = spawn_tap_engine(id, url, crate::session_map::AppState::default()); + + // rx_function_call: Some(Receiver) — engine owns the paired Sender. + assert!( + conn.rx_function_call.is_some(), + "TapConn must carry a drainable rx_function_call" + ); + // tx_function_call_output: Some(Sender) — engine owns the paired + // Receiver on the pump-side (run_tap_client). + assert!( + conn.tx_function_call_output.is_some(), + "TapConn must carry a writeable tx_function_call_output" + ); + // tool_registry: present (slice-3 §6.2 — one registry per active + // channel; HangupTool is the only wired tool). + let reg = conn.tool_registry.lock().await; + assert_eq!(reg.catalog().len(), 1, "only hangup is wired in slice-3"); + assert_eq!(reg.catalog()[0]["name"], "hangup"); + + let _ = conn.close_tx.send(()); + conn.join.abort(); + } } diff --git a/crates/rutster/tests/tap_integration.rs b/crates/rutster/tests/tap_integration.rs index e73140c..634ee19 100644 --- a/crates/rutster/tests/tap_integration.rs +++ b/crates/rutster/tests/tap_integration.rs @@ -26,6 +26,7 @@ use std::sync::atomic::Ordering; use std::time::Duration; use futures_util::{SinkExt, StreamExt}; +use rutster::session_map::AppState; use rutster::tap_engine::spawn_tap_engine; use rutster_call_model::ChannelId; use rutster_media::{AudioPipe, AudioSink, AudioSource}; @@ -76,7 +77,8 @@ async fn reconnect_after_brain_kill_resumes_audio_and_flushes_playout() { // flush side-channel gets drained manually below to exercise the // Fix-3 playout-ring-flush contract end-to-end. let session_id = ChannelId::new(); - let (mut pipe, mut conn) = spawn_tap_engine(session_id, url); + let app_state = AppState::default(); + let (mut pipe, mut conn) = spawn_tap_engine(session_id, url, app_state); // 3. Push TWO frames with the same marker `samples[0] = 7` back-to-back // before the kill so the playout ring has buffered content that the