Slice 3 — OpenAI Realtime brain: swap echo for the brain #4
@@ -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);
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -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<T>(
|
||||
mut ws: WebSocketStream<T>,
|
||||
session_id: ChannelId,
|
||||
rx_pcm_in: &mut mpsc::Receiver<PcmFrame>,
|
||||
tx_audio_out: mpsc::Sender<PcmFrame>,
|
||||
tx_function_call: mpsc::Sender<FunctionCallEvent>,
|
||||
rx_function_call_output: &mut mpsc::Receiver<FunctionCallOutputEvent>,
|
||||
metrics: Arc<TapMetrics>,
|
||||
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<u64>,
|
||||
tx_audio_out: &mpsc::Sender<PcmFrame>,
|
||||
tx_function_call: &mpsc::Sender<FunctionCallEvent>,
|
||||
metrics: &Arc<TapMetrics>,
|
||||
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::<FunctionCallEvent>(8);
|
||||
let (tx_audio_out, _rx_audio_out) = mpsc::channel::<PcmFrame>(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<u64> = 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::<FunctionCallEvent>(8);
|
||||
let (tx_audio_out, _rx_audio_out) = mpsc::channel::<PcmFrame>(8);
|
||||
let metrics = Arc::new(TapMetrics::new());
|
||||
|
||||
let wire = crate::protocol::encode_speech_started(2, 200).unwrap();
|
||||
let mut last_seq: Option<u64> = 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));
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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<rutster_tap::FunctionCallEvent>;
|
||||
let tool_registry: Arc<tokio::sync::Mutex<crate::tool_registry::ToolRegistry>>;
|
||||
let tx_out: Option<mpsc::Sender<rutster_tap::FunctionCallOutputEvent>>;
|
||||
{
|
||||
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<Sender>`; `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::<FunctionCallEvent>(8);
|
||||
let (tx_fco, mut rx_fco) = tokio::sync::mpsc::channel::<FunctionCallOutputEvent>(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"
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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<mpsc::Receiver<()>>,
|
||||
/// 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<mpsc::Receiver<FunctionCallEvent>>,
|
||||
/// 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<mpsc::Sender<FunctionCallOutputEvent>>,
|
||||
/// 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<Mutex<_>>` 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<Mutex<ToolRegistry>>,
|
||||
}
|
||||
|
||||
/// 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::<FunctionCallEvent>(TAP_MPSC_CAPACITY);
|
||||
let (tx_function_call_output, rx_function_call_output) =
|
||||
mpsc::channel::<FunctionCallOutputEvent>(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<Mutex<ToolRegistry>>` 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<TapMetrics>` 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<rutster_media::PcmFrame>,
|
||||
mut close: oneshot::Receiver<()>,
|
||||
flush_tx: mpsc::Sender<()>,
|
||||
tx_function_call: mpsc::Sender<FunctionCallEvent>,
|
||||
mut rx_function_call_output: mpsc::Receiver<FunctionCallOutputEvent>,
|
||||
metrics: Arc<TapMetrics>,
|
||||
) {
|
||||
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();
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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
|
||||
|
||||
Reference in New Issue
Block a user