From 237a1388a45453271a7241c7fac01aee6fe711c3 Mon Sep 17 00:00:00 2001 From: opencode controller Date: Sun, 28 Jun 2026 17:20:00 -0400 Subject: [PATCH] =?UTF-8?q?feat(tap-echo):=20Rust=20reference=20echo=20bra?= =?UTF-8?q?in=20+=20test=20server=20(spec=20=C2=A72.3)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - start_echo_server(addr): in-process WS server for integration tests; returns EchoHandle { shutdown, join, addr }. - echo_one_connection: the per-connection echo loop — hello handshake, audio_in → audio_out (same PCM), bye/session_end graceful close. - Reuses rutster-tap's protocol types — the wire-types-reusable contract test (spec §2.3). - Stateless across reconnects (spec §5.3) — every hello starts fresh. - Standalone binary: binds ws://127.0.0.1:8081, runs forever. - 1 unit test exercises full hello/ack/audio_in/audio_out/bye over a TCP loopback pair. Spec ref: 2026-06-28-slice-2-agent-tap-design.md §2.3, §5.3. --- crates/rutster-tap-echo/src/lib.rs | 254 +++++++++++++++++++++++++++- crates/rutster-tap-echo/src/main.rs | 23 ++- 2 files changed, 268 insertions(+), 9 deletions(-) diff --git a/crates/rutster-tap-echo/src/lib.rs b/crates/rutster-tap-echo/src/lib.rs index 9ccf105..8a6ea25 100644 --- a/crates/rutster-tap-echo/src/lib.rs +++ b/crates/rutster-tap-echo/src/lib.rs @@ -3,10 +3,9 @@ //! Dual-purpose crate: //! - **Standalone binary** (`cargo run -p rutster-tap-echo`): binds //! `ws://127.0.0.1:8081/echo` and echoes `audio_in` → `audio_out` per the -//! slice-2 protocol (spec §3). The dev-loop brain the core dials out to. -//! - **In-process `EchoServer`** (Task 5 lands `EchoServer::start`): used by -//! `rutster`'s integration tests to drive the tap end-to-end without an -//! external process. +//! slice-2 protocol (spec §3). +//! - **In-process `start_echo_server`** (lib): used by `rutster`'s integration +//! tests to drive the tap end-to-end without an external process. //! //! ## Why a Rust brain at all (when the canonical brain is Python?) //! @@ -15,9 +14,252 @@ //! Rust (or a step-3 OpenAI adapter in Rust) starts from this shape. The //! Python brain (`examples/echo_brain/`) proves language-agnosticism; this //! crate proves reusability + powers the in-process integration tests. +//! +//! ## Stateless contract (spec §5.3) +//! +//! The echo brain holds no per-call state across reconnects. On a fresh +//! `hello` it starts a new logical session; the core treats every reconnect +//! as a resume with the same `session_id`, but the brain just acks and +//! echoes. This is the resilience posture slice-2 proves. + +use std::net::SocketAddr; + +use futures_util::{SinkExt, StreamExt}; +use rutster_tap::protocol::{ + decode_envelope, encode_audio_out, encode_bye, encode_error, encode_hello, DecodedPayload, + HelloPayload, +}; +use tokio::sync::oneshot; +use tokio::task::JoinHandle; +use tokio_tungstenite::WebSocketStream; +use tracing::{info, warn}; + +/// Handle returned by `start_echo_server`; drop the `shutdown` sender to stop. +pub struct EchoHandle { + pub shutdown: oneshot::Sender<()>, + pub join: JoinHandle<()>, + pub addr: SocketAddr, +} + +/// Start an in-process echo brain bound to `addr`. Returns once the socket +/// is bound; the accept loop runs in a spawned task. +pub async fn start_echo_server(addr: SocketAddr) -> Result { + let listener = tokio::net::TcpListener::bind(addr).await?; + let bound_addr = listener.local_addr()?; + let (shutdown_tx, shutdown_rx) = oneshot::channel::<()>(); + let join = tokio::spawn(async move { + accept_loop(listener, shutdown_rx).await; + }); + Ok(EchoHandle { + shutdown: shutdown_tx, + join, + addr: bound_addr, + }) +} + +async fn accept_loop(listener: tokio::net::TcpListener, mut shutdown: oneshot::Receiver<()>) { + loop { + tokio::select! { + _ = &mut shutdown => { + info!("echo server shutting down"); + return; + } + res = listener.accept() => { + let (stream, peer) = match res { + Ok(s) => s, + Err(e) => { + warn!(error = %e, "accept failed; continuing"); + continue; + } + }; + tokio::spawn(async move { + let ws = match tokio_tungstenite::accept_async(stream).await { + Ok(ws) => ws, + Err(e) => { + warn!(error = %e, %peer, "ws upgrade failed"); + return; + } + }; + if let Err(e) = echo_one_connection(ws).await { + warn!(error = ?e, %peer, "echo connection ended with error"); + } + }); + } + } + } +} + +/// Per-connection echo loop. Unit-testable (we can drive a `WebSocketStream` +/// with synthetic frames in a test). +pub async fn echo_one_connection( + mut ws: WebSocketStream, +) -> Result<(), Box> +where + T: tokio::io::AsyncRead + tokio::io::AsyncWrite + Unpin, +{ + let mut seq_egress: u64 = 0; + // Wait for hello; ack with hello. + let hello_in = ws + .next() + .await + .ok_or("brain: connection closed before hello")??; + // tungstenite 0.24: `Message::into_text()` returns `Result`, + // not `Option`. `map_err` collapses the error to a `&str` for + // the `Box` path. + let hello_text = hello_in.into_text().map_err(|_| "brain: hello not text")?; + let decoded = decode_envelope(&hello_text)?; + let session_id = match decoded.payload { + DecodedPayload::Hello(HelloPayload { session_id, .. }) => session_id, + _ => return Err("brain: first frame not hello".into()), + }; + info!(%session_id, "brain: hello received; acking"); + let ack = encode_hello(&session_id, seq_egress, 0)?; + seq_egress += 1; + ws.send(tokio_tungstenite::tungstenite::Message::Text(ack)) + .await?; + + // Echo loop: audio_in → audio_out (same PCM). + while let Some(msg) = ws.next().await { + let msg = msg?; + // `into_text()` on 0.24 returns `Result`, so `let Ok(...) else` + // is the natural pattern for "ignore binary / decode failures and + // continue the loop" (drop + observe, spec §3.8). + let Ok(text) = msg.into_text() else { + // Binary frames ignored (v1 text-JSON only). + continue; + }; + let decoded = match decode_envelope(&text) { + Ok(d) => d, + Err(e) => { + let err_frame = encode_error("decode_failed", &e.to_string(), seq_egress, 0)?; + let _ = ws + .send(tokio_tungstenite::tungstenite::Message::Text(err_frame)) + .await; + continue; + } + }; + match decoded.payload { + DecodedPayload::AudioIn(audio) => { + // Echo: same PCM, same samples count. + let out_frame = rutster_tap::protocol::decode_pcm(&audio.pcm, audio.samples)?; + let out_str = encode_audio_out(&out_frame, seq_egress, decoded.ts)?; + seq_egress += 1; + ws.send(tokio_tungstenite::tungstenite::Message::Text(out_str)) + .await?; + } + DecodedPayload::Bye(p) => { + info!(reason = %p.reason, "brain: bye received; closing"); + let bye_ack = encode_bye("brain_ack", seq_egress, 0)?; + let _ = ws + .send(tokio_tungstenite::tungstenite::Message::Text(bye_ack)) + .await; + let _ = ws.close(None).await; + return Ok(()); + } + DecodedPayload::SessionEnd(p) => { + info!(reason = %p.reason, "brain: session_end received; closing"); + let _ = ws.close(None).await; + return Ok(()); + } + _ => { + // Unknown / unexpected; ignore (drop + observe). + warn!("brain: ignoring unexpected frame kind"); + } + } + } + Ok(()) +} #[cfg(test)] mod tests { - #[test] - fn crate_compiles() {} + use super::*; + // `PcmFrame` is re-exported by `rutster-tap` from `rutster-media` (spec + // §3.1 — one canonical home). Test consumes the re-export rather than + // adding `rutster-media` as a second path dep on this crate. + use rutster_tap::protocol::{encode_audio_in, encode_bye, encode_hello}; + use rutster_tap::PcmFrame; + + #[tokio::test] + async fn echo_round_trips_one_audio_frame() { + // Two ends of an in-memory WS pair. tokio_tungstenite doesn't ship + // a direct in-mem channel impl, so use a TCP loopback pair. + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + + let server = tokio::spawn(async move { + let (s, _) = listener.accept().await.unwrap(); + let ws = tokio_tungstenite::accept_async(s).await.unwrap(); + echo_one_connection(ws).await.unwrap(); + }); + + let client_stream = tokio::net::TcpStream::connect(addr).await.unwrap(); + // tungstenite 0.24: `Request::default()` is not a usable default for + // `client_async`. Use `IntoClientRequest` impls (`&str` / `String` + // / `url::Url` all work). `format!("ws://...")` constructs the + // request in one line — simpler than building a `Request::builder()` + // chain. The test asserts `echo_one_connection`'s behavior, not the + // client-side handshake API. + let (mut client_ws, _resp) = + tokio_tungstenite::client_async(format!("ws://{addr}/echo"), client_stream) + .await + .unwrap(); + + // Send hello. + let hello = encode_hello("test-session-id", 0, 0).unwrap(); + client_ws + .send(tokio_tungstenite::tungstenite::Message::Text(hello)) + .await + .unwrap(); + + // Receive hello ack. + let ack = client_ws + .next() + .await + .unwrap() + .unwrap() + .into_text() + .unwrap(); + assert!(ack.contains("\"type\":\"hello\"")); + assert!(ack.contains("test-session-id")); + + // Send audio_in. + let mut frame = PcmFrame::zeroed(); + frame.samples[0] = 42; + let audio_in = encode_audio_in(&frame, 1, 100).unwrap(); + client_ws + .send(tokio_tungstenite::tungstenite::Message::Text(audio_in)) + .await + .unwrap(); + + // Receive audio_out — should be the same PCM. + let audio_out = client_ws + .next() + .await + .unwrap() + .unwrap() + .into_text() + .unwrap(); + assert!(audio_out.contains("\"type\":\"audio_out\"")); + let decoded = decode_envelope(&audio_out).unwrap(); + match decoded.payload { + DecodedPayload::AudioOut(p) => { + let echoed = rutster_tap::protocol::decode_pcm(&p.pcm, p.samples).unwrap(); + assert_eq!(echoed.samples[0], 42); + } + _ => panic!("expected audio_out"), + } + + // Bye. + let bye = encode_bye("done", 2, 200).unwrap(); + client_ws + .send(tokio_tungstenite::tungstenite::Message::Text(bye)) + .await + .unwrap(); + // Wait for the bye-ack so the server's `close(None)` flushes + // before the spawned task resolves (otherwise the client's next + // read might race the close). Receive one final message — either + // the bye-ack text frame or the close frame — then join. + let _ = client_ws.next().await; + server.await.unwrap(); + } } diff --git a/crates/rutster-tap-echo/src/main.rs b/crates/rutster-tap-echo/src/main.rs index 8331b8b..6a14693 100644 --- a/crates/rutster-tap-echo/src/main.rs +++ b/crates/rutster-tap-echo/src/main.rs @@ -1,6 +1,23 @@ //! Standalone binary: bind `ws://127.0.0.1:8081/echo`, echo audio_in → audio_out. -//! Real implementation lands in Task 5; this is the skeleton that compiles. +//! Dev-loop brain the core dials out to (spec §2.3, §8.3). -fn main() { - eprintln!("rutster-tap-echo: skeleton — implementation lands in Task 5"); +use std::net::SocketAddr; + +use rutster_tap_echo::start_echo_server; +use tracing::info; + +#[tokio::main] +async fn main() { + tracing_subscriber::fmt() + .with_env_filter( + tracing_subscriber::EnvFilter::try_from_default_env() + .unwrap_or_else(|_| "rutster_tap_echo=info".into()), + ) + .init(); + + let addr: SocketAddr = "127.0.0.1:8081".parse().expect("valid addr"); + info!(%addr, "rutster-tap-echo listening"); + let handle = start_echo_server(addr).await.expect("bind ok"); + // Run forever (Ctrl-C terminates the process; no graceful shutdown yet). + let _ = handle.join.await; }