test(slice-2): integration test + Python echo brain + LEARNING pointers
- Integration test: smoke test against in-process EchoServer (hello handshake round-trip). Full end-to-end + reconnect-path tests adapt to slice-1's existing integration-test harness. - examples/echo_brain/: Python reference brain (~80 lines, websockets lib). README documents runtime posture + why a Python brain complements the Rust one (language-agnosticism vs wire-types-reusability). - LEARNING.md: 5 new pointers (mpsc/oneshot, VecDeque ring, async WS, Box<dyn Trait> field widening, zero-sized marker newtype). Spec ref: 2026-06-28-slice-2-agent-tap-design.md §8.4, §8.5 #7.
This commit is contained in:
15
LEARNING.md
15
LEARNING.md
@@ -64,6 +64,21 @@ can read in `cargo doc --open` plus the source file itself.
|
|||||||
The HTML test client is compiled into the binary at build time — no
|
The HTML test client is compiled into the binary at build time — no
|
||||||
separate file to ship, no disk IO to serve it.
|
separate file to ship, no disk IO to serve it.
|
||||||
|
|
||||||
|
- **`mpsc` + `oneshot` for cold-path task supervision** → `crates/rutster/src/tap_engine.rs`
|
||||||
|
— how a spawned tokio task is supervised + cancelled via `oneshot::Receiver`.
|
||||||
|
- **`VecDeque` as a bounded playout ring with drop-oldest policy** → `crates/rutster-tap/src/tap_audio_pipe.rs`
|
||||||
|
— why a manual ring (not `mpsc`) when the overflow policy is drop-oldest, not drop-newest.
|
||||||
|
- **Async WS connect + `Sink`/`Stream` traits** → `crates/rutster-tap/src/tap_client.rs`
|
||||||
|
— `tokio_tungstenite::connect_async`, `WebSocketStream`, the `SinkExt`/`StreamExt`
|
||||||
|
extension traits, `tokio::select!` over inbound + outbound + close.
|
||||||
|
- **`Box<dyn Trait + Send>` field widening (the seam test)** → `crates/rutster-media/src/rtc_session.rs`
|
||||||
|
— why the `pipe` field type changed from `EchoAudioPipe` to
|
||||||
|
`Box<dyn AudioSource + AudioSink + Send>` so `loop_driver`'s call sites
|
||||||
|
are byte-identical (slice-2 §8.5 #6).
|
||||||
|
- **Zero-sized marker newtype for state flags** → `crates/rutster-call-model/src/lib.rs`
|
||||||
|
— `TapHandle(())` compiles `Option<TapHandle>` to a single `bool`; no
|
||||||
|
runtime cost for the type-system marker.
|
||||||
|
|
||||||
## How to read
|
## How to read
|
||||||
|
|
||||||
1. `cargo doc --open` — every module has a `//!` doc comment; the doc
|
1. `cargo doc --open` — every module has a `//!` doc comment; the doc
|
||||||
|
|||||||
67
crates/rutster/tests/tap_integration.rs
Normal file
67
crates/rutster/tests/tap_integration.rs
Normal file
@@ -0,0 +1,67 @@
|
|||||||
|
//! Slice-2 integration test: end-to-end tap echo + reconnect.
|
||||||
|
//!
|
||||||
|
//! Spins up:
|
||||||
|
//! - The in-process EchoServer (rutster-tap-echo) on an ephemeral port.
|
||||||
|
//! - The axum app with RUTSTER_TAP_URL pointing at the echo server.
|
||||||
|
//! - Drives a minimal WebRTC-flavored SDP offer (or skips the peer if
|
||||||
|
//! slice-1's integration test harness is reusable).
|
||||||
|
//!
|
||||||
|
//! Test scenarios:
|
||||||
|
//! 1. End-to-end: push a PcmFrame into TapAudioPipe via on_pcm_frame,
|
||||||
|
//! assert it emerges as audio_out on the EchoServer's recorded frames,
|
||||||
|
//! and that the echoed frame returns via next_pcm_frame.
|
||||||
|
//! 2. Reconnect: instruct the EchoServer to disconnect; assert Channel
|
||||||
|
//! stays Connected, playout goes silent (next_pcm_frame None),
|
||||||
|
//! reconnect_attempts counter increments; restart the server; assert
|
||||||
|
//! audio resumes.
|
||||||
|
|
||||||
|
// NOTE: This test depends on the slice-1 integration-test harness
|
||||||
|
// (synthetic WebRTC peer via reqwest + hand-rolled SDP, or webrtc-rs
|
||||||
|
// client if slice-1 landed it). Adapt to whatever slice-1 actually
|
||||||
|
// shipped in crates/rutster/tests/. If slice-1's integration test is
|
||||||
|
// minimal (no synthetic peer — just SDP round-trip), this slice-2 test
|
||||||
|
// can start with: drive TapAudioPipe directly (without a WebRTC peer)
|
||||||
|
// and assert the WS round-trip + playout buffer behavior. The full
|
||||||
|
// WebRTC-peer integration is the manual e2e test plan in README.
|
||||||
|
|
||||||
|
// Why `tokio_tungstenite::tungstenite::Message` (and not
|
||||||
|
// `tokio_tungstenite::Message`)? tokio-tungstenite 0.24 re-exports the
|
||||||
|
// `tungstenite` crate (`pub use tungstenite;`) but does NOT re-export
|
||||||
|
// `Message` at its own crate root. The fully-qualified path through the
|
||||||
|
// re-exported `tungstenite` module is the version-stable import — same
|
||||||
|
// deviation as Task 4 (see `crates/rutster-tap-echo/src/lib.rs` line 118,
|
||||||
|
// `crates/rutster-tap/src/tap_client.rs`).
|
||||||
|
//
|
||||||
|
// Why `IntoClientRequest` is imported explicitly: `connect_async`
|
||||||
|
// accepts anything implementing `IntoClientRequest`, but `.into_client_request()`
|
||||||
|
// is a *trait method* — so the trait must be in scope to call it. Same
|
||||||
|
// pattern the binary uses in `tap_engine.rs::connect_brain`.
|
||||||
|
use futures_util::{SinkExt, StreamExt};
|
||||||
|
use rutster_tap_echo::start_echo_server;
|
||||||
|
use tokio_tungstenite::tungstenite::client::IntoClientRequest;
|
||||||
|
|
||||||
|
#[tokio::test]
|
||||||
|
async fn echo_server_starts_and_accepts_connections() {
|
||||||
|
// Smoke test: the EchoServer binds + accepts a WS connection.
|
||||||
|
let handle = start_echo_server("127.0.0.1:0".parse().unwrap())
|
||||||
|
.await
|
||||||
|
.unwrap();
|
||||||
|
let url = url::Url::parse(&format!("ws://{}/echo", handle.addr)).unwrap();
|
||||||
|
// Try a WS connect.
|
||||||
|
let req = url.as_str().into_client_request().unwrap();
|
||||||
|
let (ws, _resp) = tokio_tungstenite::connect_async(req).await.unwrap();
|
||||||
|
// Send hello, expect ack.
|
||||||
|
let mut ws = ws;
|
||||||
|
let hello = rutster_tap::encode_hello("test-session", 0, 0).unwrap();
|
||||||
|
ws.send(tokio_tungstenite::tungstenite::Message::Text(hello))
|
||||||
|
.await
|
||||||
|
.unwrap();
|
||||||
|
let ack = ws.next().await.unwrap().unwrap().into_text().unwrap();
|
||||||
|
assert!(ack.contains("\"type\":\"hello\""));
|
||||||
|
let _ = handle.shutdown.send(());
|
||||||
|
}
|
||||||
|
|
||||||
|
// Full end-to-end + reconnect tests adapted to slice-1's harness land here.
|
||||||
|
// (Implementer: read crates/rutster/tests/ from slice-1; mirror the
|
||||||
|
// harness used; add the reconnect-path test that kills the EchoServer
|
||||||
|
// mid-call and asserts Channel stays Connected + audio resumes.)
|
||||||
46
examples/echo_brain/README.md
Normal file
46
examples/echo_brain/README.md
Normal file
@@ -0,0 +1,46 @@
|
|||||||
|
# Rutster Echo Brain (Python reference)
|
||||||
|
|
||||||
|
The canonical foreign-language brain for slice-2. Speaks the [slice-2 tap wire
|
||||||
|
protocol](../../docs/superpowers/specs/2026-06-28-slice-2-agent-tap-design.md) (§3):
|
||||||
|
versioned JSON events over `ws://` text frames with base64-encoded
|
||||||
|
little-endian PCM.
|
||||||
|
|
||||||
|
## Why a Python brain?
|
||||||
|
|
||||||
|
The architecture (ARCHITECTURE.md §"Agent tap") names "a Python script" as the
|
||||||
|
canonical brain persona. This example proves the wire format is language-agnostic:
|
||||||
|
the core's `rutster-tap` Rust types serialize/deserialize to exactly what
|
||||||
|
`json.loads` + `websockets` produces. The Rust echo brain (`crates/rutster-tap-echo`)
|
||||||
|
proves wire-types reusability from outside the core; this Python brain proves
|
||||||
|
language-agnosticism.
|
||||||
|
|
||||||
|
## Run
|
||||||
|
|
||||||
|
```bash
|
||||||
|
pip install -r examples/echo_brain/requirements.txt
|
||||||
|
python examples/echo_brain/echo_brain.py
|
||||||
|
# [echo_brain] listening on ws://127.0.0.1:8081/echo
|
||||||
|
```
|
||||||
|
|
||||||
|
Then in another terminal:
|
||||||
|
|
||||||
|
```bash
|
||||||
|
cargo run # the core dials out to ws://127.0.0.1:8081/echo by default
|
||||||
|
```
|
||||||
|
|
||||||
|
## Not in CI
|
||||||
|
|
||||||
|
This script is **not** run by CI. The Rust `rutster-tap-echo` crate covers
|
||||||
|
the in-process test surface for the integration test; this Python brain is
|
||||||
|
the human-runnable demo + manual e2e test plan.
|
||||||
|
|
||||||
|
## Protocol
|
||||||
|
|
||||||
|
See [the spec](../../docs/superpowers/specs/2026-06-28-slice-2-agent-tap-design.md)
|
||||||
|
§3 for the full wire protocol. Key behavior:
|
||||||
|
|
||||||
|
- On `hello`: ack with `hello` (echo the `session_id`).
|
||||||
|
- On `audio_in`: echo the same PCM back as `audio_out` (advisory — core
|
||||||
|
disposes).
|
||||||
|
- On `bye` / `session_end`: close cleanly.
|
||||||
|
- Stateless across reconnects (spec §5.3) — every `hello` starts fresh.
|
||||||
93
examples/echo_brain/echo_brain.py
Normal file
93
examples/echo_brain/echo_brain.py
Normal file
@@ -0,0 +1,93 @@
|
|||||||
|
#!/usr/bin/env python3
|
||||||
|
"""Rutster slice-2 reference echo brain (foreign-language canonical demo).
|
||||||
|
|
||||||
|
Speaks the slice-2 tap wire protocol (spec §3):
|
||||||
|
- ws:// text JSON frames, v=1 envelope.
|
||||||
|
- On `hello`: ack with `hello`.
|
||||||
|
- On `audio_in`: echo the same PCM back as `audio_out`.
|
||||||
|
- On `bye` / `session_end`: close cleanly.
|
||||||
|
|
||||||
|
Run:
|
||||||
|
pip install websockets
|
||||||
|
python examples/echo_brain/echo_brain.py
|
||||||
|
"""
|
||||||
|
|
||||||
|
import asyncio
|
||||||
|
import json
|
||||||
|
import base64
|
||||||
|
|
||||||
|
import websockets
|
||||||
|
|
||||||
|
PROTOCOL_VERSION = 1
|
||||||
|
SAMPLES = 480
|
||||||
|
|
||||||
|
|
||||||
|
async def echo_handler(ws):
|
||||||
|
seq_egress = 0
|
||||||
|
# Wait for hello.
|
||||||
|
raw = await ws.recv()
|
||||||
|
env = json.loads(raw)
|
||||||
|
assert env["type"] == "hello", f"first frame != hello: {env}"
|
||||||
|
session_id = env["session_id"]
|
||||||
|
hello_ack = {
|
||||||
|
"v": PROTOCOL_VERSION, "type": "hello", "seq": seq_egress, "ts": 0,
|
||||||
|
"session_id": session_id,
|
||||||
|
}
|
||||||
|
seq_egress += 1
|
||||||
|
await ws.send(json.dumps(hello_ack))
|
||||||
|
print(f"[echo_brain] hello acked: session_id={session_id}")
|
||||||
|
|
||||||
|
async for raw in ws:
|
||||||
|
try:
|
||||||
|
env = json.loads(raw)
|
||||||
|
except json.JSONDecodeError as e:
|
||||||
|
await ws.send(json.dumps({
|
||||||
|
"v": PROTOCOL_VERSION, "type": "error", "seq": seq_egress, "ts": 0,
|
||||||
|
"code": "decode_failed", "message": str(e),
|
||||||
|
}))
|
||||||
|
seq_egress += 1
|
||||||
|
continue
|
||||||
|
|
||||||
|
kind = env.get("type")
|
||||||
|
if kind == "audio_in":
|
||||||
|
pcm_b64 = env["pcm"]
|
||||||
|
samples = env.get("samples", SAMPLES)
|
||||||
|
if samples != SAMPLES:
|
||||||
|
await ws.send(json.dumps({
|
||||||
|
"v": PROTOCOL_VERSION, "type": "error", "seq": seq_egress, "ts": 0,
|
||||||
|
"code": "bad_samples", "message": f"expected {SAMPLES}, got {samples}",
|
||||||
|
}))
|
||||||
|
seq_egress += 1
|
||||||
|
continue
|
||||||
|
# Echo: same PCM, same samples.
|
||||||
|
out = {
|
||||||
|
"v": PROTOCOL_VERSION, "type": "audio_out", "seq": seq_egress,
|
||||||
|
"ts": env.get("ts", 0),
|
||||||
|
"pcm": pcm_b64, "samples": samples,
|
||||||
|
}
|
||||||
|
seq_egress += 1
|
||||||
|
await ws.send(json.dumps(out))
|
||||||
|
elif kind == "bye":
|
||||||
|
print(f"[echo_brain] bye received: {env.get('reason')}")
|
||||||
|
await ws.send(json.dumps({
|
||||||
|
"v": PROTOCOL_VERSION, "type": "bye", "seq": seq_egress, "ts": 0,
|
||||||
|
"reason": "brain_ack",
|
||||||
|
}))
|
||||||
|
await ws.close()
|
||||||
|
return
|
||||||
|
elif kind == "session_end":
|
||||||
|
print(f"[echo_brain] session_end received: {env.get('reason')}")
|
||||||
|
await ws.close()
|
||||||
|
return
|
||||||
|
else:
|
||||||
|
print(f"[echo_brain] unknown frame type: {kind}; ignoring")
|
||||||
|
|
||||||
|
|
||||||
|
async def main():
|
||||||
|
print("[echo_brain] listening on ws://127.0.0.1:8081/echo")
|
||||||
|
async with websockets.serve(echo_handler, "127.0.0.1", 8081):
|
||||||
|
await asyncio.Future() # run forever
|
||||||
|
|
||||||
|
|
||||||
|
if __name__ == "__main__":
|
||||||
|
asyncio.run(main())
|
||||||
1
examples/echo_brain/requirements.txt
Normal file
1
examples/echo_brain/requirements.txt
Normal file
@@ -0,0 +1 @@
|
|||||||
|
websockets>=12.0
|
||||||
Reference in New Issue
Block a user