feat(tap): versioned JSON event wire protocol (spec §3)
- protocol.rs: Envelope + FrameKind + Payload types, serde-derived. - Explicit little-endian PCM codec (to_le_bytes / from_le_bytes) — spec §3, §9. No host-endian silent hazard. - encode_* helpers for hello, audio_in, audio_out, session_end, bye, error. - decode_envelope validates protocol version + samples count; unknown type values decode to DecodedPayload::Unknown (log + count + drop per spec §3.4). - 10 unit tests cover round-trips, LE byte order, sample-count mismatch, unknown-type drop, and version mismatch. Spec ref: 2026-06-28-slice-2-agent-tap-design.md §3.
This commit is contained in:
opencode controller
@@ -1,18 +1,44 @@
|
||||
//! # rutster-tap
|
||||
//!
|
||||
//! **Status:** stub. Fills in at spearhead step 2 (the tap itself).
|
||||
//!
|
||||
//! Slice 1 *pre-paves* the tap by exposing the canonical PCM boundary as
|
||||
//! The agent tap: core-as-client + brain-as-server (spec §2, ADR-0006).
|
||||
//! Slice-1 pre-paved the seam by exposing the canonical PCM boundary as
|
||||
//! the `AudioSource` / `AudioSink` traits in [`rutster_media`](../rutster-media/index.html),
|
||||
//! and wires an `EchoAudioPipe` between sink and source. Step 2 swaps that
|
||||
//! pipe for a real WSS tap client (core-as-client, brain-as-server —
|
||||
//! [ADR-0006](../../../docs/adr/0006-ingress-posture.md)). No code changes to
|
||||
//! `RtcSession` itself in step 2 — that's the test of the seam.
|
||||
//! wired by an in-process `EchoAudioPipe`. Slice-2 (this crate) swaps that
|
||||
//! pipe for a real WSS tap client. No code changes to `RtcSession` itself
|
||||
//! in step 2 — that's the test of the seam (slice-1 §3.3).
|
||||
//!
|
||||
//! This crate will, when filled in, re-export `PcmFrame` from
|
||||
//! `rutster-media` (one canonical home — spec §3.1) and ship the WSS
|
||||
//! tap client + the versioned framing protocol. It depends on nothing in
|
||||
//! the workspace in slice 1.
|
||||
//! # Layout
|
||||
//!
|
||||
//! - [`protocol`]: the versioned JSON event wire format (spec §3). PCM
|
||||
//! is base64-encoded little-endian i16 (spec §3, §9).
|
||||
//! - `tap_audio_pipe` (Task 3): the sync `AudioSource`/`AudioSink` impl over
|
||||
//! mpsc + a bounded playout ring (spec §4.1). The seam object `RtcSession`
|
||||
//! holds; the slice-1 `loop_driver` calls it via `next_pcm_frame` /
|
||||
//! `on_pcm_frame` exactly as before.
|
||||
//! - `tap_client` (Task 3/4): the async WSS connection driver. Runs inside the
|
||||
//! `TapEngine` task (in the binary crate); the media loop never sees it.
|
||||
//! - `metrics` (Task 4): atomic counters — the "observe" half of "drop +
|
||||
//! observe" (slice-1 §3.8 extended to the tap wire).
|
||||
//!
|
||||
//! # Dependency direction
|
||||
//!
|
||||
//! `rutster-tap` → `rutster-media` (re-exports `PcmFrame`; one canonical
|
||||
//! home — spec §2.1). `rutster-media` does NOT depend on `rutster-tap`
|
||||
//! — that would invert the canonical-home of `PcmFrame` and pull the
|
||||
//! loopback peer into the tap story.
|
||||
|
||||
pub mod protocol;
|
||||
|
||||
// Re-export PcmFrame so `rutster-tap` consumers (the binary, the echo brain,
|
||||
// future brains) get it from one canonical home (spec §3.1).
|
||||
pub use rutster_media::PcmFrame;
|
||||
|
||||
pub use protocol::{
|
||||
decode_envelope, decode_pcm, encode_audio_in, encode_audio_out, encode_bye, encode_error,
|
||||
encode_hello, encode_pcm, encode_session_end, AudioPayload, DecodedFrame, DecodedPayload,
|
||||
Envelope, ErrorPayload, FrameKind, HelloPayload, Payload, ReasonPayload, SessionEndPayload,
|
||||
TapProtoError, PROTOCOL_VERSION, SAMPLES_PER_FRAME as WIRE_SAMPLES_PER_FRAME,
|
||||
};
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
|
||||
599
crates/rutster-tap/src/protocol.rs
Normal file
599
crates/rutster-tap/src/protocol.rs
Normal file
@@ -0,0 +1,599 @@
|
||||
//! # Tap wire protocol (spec §3 — versioned JSON event framing)
|
||||
//!
|
||||
//! One JSON object per WS text frame. PCM payloads are base64-encoded
|
||||
//! **little-endian i16** bytes (spec §3, §9 — explicit LE, not host-endian).
|
||||
//! Every envelope carries `v: 1` + `type` + per-direction `seq` + advisory `ts`.
|
||||
//!
|
||||
//! ## Why JSON + base64 over binary length-prefixed framing
|
||||
//!
|
||||
//! ARCHITECTURE.md names WSS as presumptive transport because the consumer
|
||||
//! is a Python script / an OpenAI-Realtime-style API for which event-
|
||||
//! framed WSS is the de-facto protocol. A JSON event envelope maps onto
|
||||
//! that ecosystem directly — a hand-rolled Python brain uses `json.loads`;
|
||||
//! the step-3 OpenAI adapter translates our events to OpenAI's event
|
||||
//! schema. A binary length-prefixed framing would force every brain to
|
||||
//! implement a byte-parser. ~33% wire overhead (65 KB/s at 24 kHz mono
|
||||
//! i16) is negligible at slice-2's scale; brain-authoring ergonomics
|
||||
//! dominate. A future-rung `v: 2` may negotiate a binary mode (spec §9).
|
||||
|
||||
use rutster_media::PcmFrame;
|
||||
use serde::ser::{SerializeStruct, Serializer};
|
||||
use serde::{Deserialize, Serialize};
|
||||
use thiserror::Error;
|
||||
/// Wire protocol version. Slice-2 ships v1 (spec §3.1, §3.4).
|
||||
pub const PROTOCOL_VERSION: u8 = 1;
|
||||
|
||||
/// Samples per 20 ms frame @ 24 kHz mono. Re-declared locally for wire-
|
||||
/// format validation; MUST match `rutster_media::pcm::SAMPLES_PER_FRAME`.
|
||||
/// (Re-declaring rather than `use`-ing the constant keeps the wire-format
|
||||
/// spec pinned in this crate — a future bump in `rutster-media` doesn't
|
||||
/// silently change the wire format without an explicit bump here.)
|
||||
pub const SAMPLES_PER_FRAME: usize = 480;
|
||||
|
||||
/// Frame kind — wire `type` field. Serde renames to the snake_case wire
|
||||
/// names (spec §3.2, §3.3).
|
||||
///
|
||||
/// # Why an enum (not `String`)
|
||||
/// Exhaustiveness: when a new variant is added, every `match` site is
|
||||
/// flagged by the compiler. A `String` would let new frame kinds slip
|
||||
/// in silently. Unknown wire values deserialize to `FrameKind::Unknown`
|
||||
/// (see `serde(other)`) — the receiver drops them per spec §3.4.
|
||||
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
|
||||
#[serde(rename_all = "snake_case")]
|
||||
pub enum FrameKind {
|
||||
Hello,
|
||||
AudioIn,
|
||||
AudioOut,
|
||||
SessionEnd,
|
||||
Bye,
|
||||
Error,
|
||||
/// Unknown wire `type` values land here (spec §3.4: log + count + drop).
|
||||
/// `#[serde(other)]` catches any string not in the variants above.
|
||||
#[serde(other)]
|
||||
Unknown,
|
||||
}
|
||||
|
||||
/// On-wire envelope (spec §3.1).
|
||||
///
|
||||
/// # Serde strategy — manual `Serialize`, no derived `Deserialize`
|
||||
///
|
||||
/// The wire contract is ONE flat JSON object: payload fields (`session_id`,
|
||||
/// `pcm`, `samples`, `reason`, `code`, `message`, ...) are *siblings* of
|
||||
/// `v`, `type`, `seq`, `ts` — not nested under a `payload` key (spec §3.1).
|
||||
///
|
||||
/// The obvious serde approach — `#[derive(Serialize, Deserialize)]` on
|
||||
/// `Envelope` with `#[serde(flatten)] payload: Payload` where `Payload` is
|
||||
/// `#[serde(tag = "type")]` — **does not work**: both the envelope's
|
||||
/// `#[serde(rename = "type")] kind` field and the flattened payload's tag
|
||||
/// claim the same JSON `type` key, producing `duplicate field "type"` on
|
||||
/// serialize and `missing field "type"` for unknown wire values on
|
||||
/// deserialize (Payload has no `Unknown` variant to absorb them).
|
||||
///
|
||||
/// The fix applied here (recommended in the task brief): keep `Envelope` as
|
||||
/// the encoding-side type with a manual `Serialize` that writes payload
|
||||
/// fields flat as siblings of `v`/`type`/`seq`/`ts`. Deserialization goes
|
||||
/// through a separate `WireEnvelope` intermediate (private) that flattens
|
||||
/// leftover fields into a `serde_json::Map` and dispatches on `kind` to
|
||||
/// build the specific payload struct. This matches the wire spec exactly
|
||||
/// and lets `FrameKind::Unknown` (via `#[serde(other)]`) flow through to
|
||||
/// `DecodedPayload::Unknown` — which the brief's tagged-enum approach
|
||||
/// cannot do.
|
||||
#[derive(Debug, Clone)]
|
||||
pub struct Envelope {
|
||||
pub v: u8,
|
||||
pub kind: FrameKind,
|
||||
pub seq: u64,
|
||||
pub ts: u64,
|
||||
pub payload: Payload,
|
||||
}
|
||||
|
||||
/// Flatten envelope + payload into ONE JSON object (spec §3.1's flat
|
||||
/// shape). This is the wire byte contract: payload fields as siblings of
|
||||
/// `v`/`type`/`seq`/`ts`, never nested under a `payload` key.
|
||||
///
|
||||
/// The derived `#[derive(Serialize)]` + `#[serde(flatten)]` approach
|
||||
/// failed (see the type-level doc above for the duplicate-`type` diagnosis).
|
||||
/// Writing `Serialize` manually is the minimal-invasive fix: it controls
|
||||
/// the wire shape directly and avoids fighting serde's flatten-over-
|
||||
/// tagged-enum machinery.
|
||||
///
|
||||
/// Field order is pinned (`v`, `type`, `seq`, `ts`, then payload fields)
|
||||
/// to make the wire bytes deterministic — JSON object key order isn't
|
||||
/// semantically meaningful but pinning it helps snapshot tests and
|
||||
/// interop with brains that read key order for debugging.
|
||||
impl Serialize for Envelope {
|
||||
fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
|
||||
where
|
||||
S: Serializer,
|
||||
{
|
||||
let payload_field_count = match &self.payload {
|
||||
Payload::Hello(_) => 4, // session_id, sample_rate, channels, frame_ms
|
||||
Payload::AudioIn(_) | Payload::AudioOut(_) => 2, // pcm, samples
|
||||
Payload::SessionEnd(_) | Payload::Bye(_) => 1, // reason
|
||||
Payload::Error(_) => 2, // code, message
|
||||
};
|
||||
let mut st = serializer.serialize_struct("Envelope", 4 + payload_field_count)?;
|
||||
st.serialize_field("v", &self.v)?;
|
||||
// `FrameKind`'s own serde derive turns `AudioIn` → "audio_in" etc.
|
||||
// via its `#[serde(rename_all = "snake_case")]`. The wire key is
|
||||
// "type" (spec §3.1's name); Rust's field name `kind` doesn't
|
||||
// appear on the wire at all.
|
||||
st.serialize_field("type", &self.kind)?;
|
||||
st.serialize_field("seq", &self.seq)?;
|
||||
st.serialize_field("ts", &self.ts)?;
|
||||
match &self.payload {
|
||||
Payload::Hello(p) => {
|
||||
st.serialize_field("session_id", &p.session_id)?;
|
||||
st.serialize_field("sample_rate", &p.sample_rate)?;
|
||||
st.serialize_field("channels", &p.channels)?;
|
||||
st.serialize_field("frame_ms", &p.frame_ms)?;
|
||||
}
|
||||
Payload::AudioIn(p) | Payload::AudioOut(p) => {
|
||||
st.serialize_field("pcm", &p.pcm)?;
|
||||
st.serialize_field("samples", &p.samples)?;
|
||||
}
|
||||
Payload::SessionEnd(p) => {
|
||||
st.serialize_field("reason", &p.reason)?;
|
||||
}
|
||||
Payload::Bye(p) => {
|
||||
st.serialize_field("reason", &p.reason)?;
|
||||
}
|
||||
Payload::Error(p) => {
|
||||
st.serialize_field("code", &p.code)?;
|
||||
st.serialize_field("message", &p.message)?;
|
||||
}
|
||||
}
|
||||
st.end()
|
||||
}
|
||||
}
|
||||
|
||||
/// Payload — internal-only data carrier; never directly serialized.
|
||||
/// `Envelope`'s manual `Serialize` flattens payload fields into the
|
||||
/// envelope. `decode_envelope` builds payloads from a flat
|
||||
/// `serde_json::Map` (see `WireEnvelope` in `decode_envelope`).
|
||||
///
|
||||
/// `AudioIn` and `AudioOut` share the same `AudioPayload` shape because
|
||||
/// the wire format does (spec §3.2, §3.3); direction is carried by the
|
||||
/// envelope's `type` field, not by a payload tag. The decode side
|
||||
/// surfaces direction via [`DecodedPayload`] (separate `AudioIn`/
|
||||
/// `AudioOut` variants) so callers `match` exhaustively on direction.
|
||||
#[derive(Debug, Clone)]
|
||||
pub enum Payload {
|
||||
Hello(HelloPayload),
|
||||
AudioIn(AudioPayload),
|
||||
AudioOut(AudioPayload),
|
||||
SessionEnd(SessionEndPayload),
|
||||
Bye(ReasonPayload),
|
||||
Error(ErrorPayload),
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Serialize, Deserialize)]
|
||||
pub struct HelloPayload {
|
||||
pub session_id: String,
|
||||
#[serde(default = "default_sample_rate")]
|
||||
pub sample_rate: u32,
|
||||
#[serde(default = "default_channels")]
|
||||
pub channels: String,
|
||||
#[serde(default = "default_frame_ms")]
|
||||
pub frame_ms: u32,
|
||||
}
|
||||
|
||||
fn default_sample_rate() -> u32 {
|
||||
24000
|
||||
}
|
||||
fn default_channels() -> String {
|
||||
"mono".to_string()
|
||||
}
|
||||
fn default_frame_ms() -> u32 {
|
||||
20
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Serialize, Deserialize)]
|
||||
pub struct AudioPayload {
|
||||
/// base64-encoded LE i16 bytes (spec §3, §9).
|
||||
pub pcm: String,
|
||||
pub samples: usize,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Serialize, Deserialize)]
|
||||
pub struct SessionEndPayload {
|
||||
pub reason: String,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Serialize, Deserialize)]
|
||||
pub struct ReasonPayload {
|
||||
pub reason: String,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Serialize, Deserialize)]
|
||||
pub struct ErrorPayload {
|
||||
pub code: String,
|
||||
pub message: String,
|
||||
}
|
||||
|
||||
/// Decoded frame — what `decode_envelope` returns. The kind is split into
|
||||
/// `audio_in` / `audio_out` variants (not just `Audio(AudioPayload)`) so
|
||||
/// callers `match` exhaustively on direction (spec §3.2 vs §3.3 — the two
|
||||
/// carry the same payload shape but mean opposite directions).
|
||||
#[derive(Debug, Clone)]
|
||||
pub struct DecodedFrame {
|
||||
pub v: u8,
|
||||
pub seq: u64,
|
||||
pub ts: u64,
|
||||
pub payload: DecodedPayload,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone)]
|
||||
pub enum DecodedPayload {
|
||||
Hello(HelloPayload),
|
||||
AudioIn(AudioPayload),
|
||||
AudioOut(AudioPayload),
|
||||
SessionEnd(SessionEndPayload),
|
||||
Bye(ReasonPayload),
|
||||
Error(ErrorPayload),
|
||||
/// Unknown `type` — log + count + drop (spec §3.4).
|
||||
Unknown,
|
||||
}
|
||||
|
||||
#[derive(Debug, Error)]
|
||||
pub enum TapProtoError {
|
||||
#[error("JSON (de)serialization failed: {0}")]
|
||||
Json(#[from] serde_json::Error),
|
||||
#[error("base64 decode failed: {0}")]
|
||||
Base64(#[from] base64::DecodeError),
|
||||
#[error("protocol version mismatch: got {got}, expected {expected}")]
|
||||
Version { got: u8, expected: u8 },
|
||||
#[error("sample count mismatch: got {got}, expected {expected}")]
|
||||
SampleCount { got: usize, expected: usize },
|
||||
#[error("decoded PCM length {got} is not a whole number of i16 samples (expected even)")]
|
||||
OddPcmLength { got: usize },
|
||||
}
|
||||
|
||||
/// Encode a `PcmFrame` → base64 LE i16 bytes (spec §3, §9 — explicit LE).
|
||||
pub fn encode_pcm(frame: &PcmFrame) -> String {
|
||||
use base64::Engine as _;
|
||||
// `i16::to_le_bytes()` — explicit little-endian, NOT host-endian.
|
||||
// The wire contract is LE-only in v1; a big-endian brain would need
|
||||
// v2 endianness negotiation (spec §9). On x86_64/aarch64 this is a
|
||||
// no-op at the hardware level but the explicit call site documents
|
||||
// the wire contract for the reader.
|
||||
let bytes: Vec<u8> = frame.samples.iter().flat_map(|s| s.to_le_bytes()).collect();
|
||||
base64::engine::general_purpose::STANDARD.encode(&bytes)
|
||||
}
|
||||
|
||||
/// Decode base64 LE i16 bytes → `PcmFrame`. Validates `samples` count.
|
||||
pub fn decode_pcm(pcm_b64: &str, expected_samples: usize) -> Result<PcmFrame, TapProtoError> {
|
||||
use base64::Engine as _;
|
||||
let bytes = base64::engine::general_purpose::STANDARD.decode(pcm_b64)?;
|
||||
if bytes.len() % 2 != 0 {
|
||||
return Err(TapProtoError::OddPcmLength { got: bytes.len() });
|
||||
}
|
||||
let got_samples = bytes.len() / 2;
|
||||
if got_samples != expected_samples {
|
||||
return Err(TapProtoError::SampleCount {
|
||||
got: got_samples,
|
||||
expected: expected_samples,
|
||||
});
|
||||
}
|
||||
let mut samples = [0i16; SAMPLES_PER_FRAME];
|
||||
for (i, chunk) in bytes.chunks_exact(2).enumerate() {
|
||||
// `i16::from_le_bytes` — explicit LE decode, matches `encode_pcm`.
|
||||
samples[i] = i16::from_le_bytes([chunk[0], chunk[1]]);
|
||||
}
|
||||
Ok(PcmFrame { samples })
|
||||
}
|
||||
|
||||
/// Build an outgoing `audio_in` frame string (core → brain, spec §3.2).
|
||||
pub fn encode_audio_in(frame: &PcmFrame, seq: u64, ts: u64) -> Result<String, TapProtoError> {
|
||||
let env = Envelope {
|
||||
v: PROTOCOL_VERSION,
|
||||
kind: FrameKind::AudioIn,
|
||||
seq,
|
||||
ts,
|
||||
payload: Payload::AudioIn(AudioPayload {
|
||||
pcm: encode_pcm(frame),
|
||||
samples: SAMPLES_PER_FRAME,
|
||||
}),
|
||||
};
|
||||
Ok(serde_json::to_string(&env)?)
|
||||
}
|
||||
|
||||
/// Build an outgoing `audio_out` frame string (brain → core, spec §3.3).
|
||||
pub fn encode_audio_out(frame: &PcmFrame, seq: u64, ts: u64) -> Result<String, TapProtoError> {
|
||||
let env = Envelope {
|
||||
v: PROTOCOL_VERSION,
|
||||
kind: FrameKind::AudioOut,
|
||||
seq,
|
||||
ts,
|
||||
payload: Payload::AudioOut(AudioPayload {
|
||||
pcm: encode_pcm(frame),
|
||||
samples: SAMPLES_PER_FRAME,
|
||||
}),
|
||||
};
|
||||
Ok(serde_json::to_string(&env)?)
|
||||
}
|
||||
|
||||
pub fn encode_hello(session_id: &str, seq: u64, ts: u64) -> Result<String, TapProtoError> {
|
||||
let env = Envelope {
|
||||
v: PROTOCOL_VERSION,
|
||||
kind: FrameKind::Hello,
|
||||
seq,
|
||||
ts,
|
||||
payload: Payload::Hello(HelloPayload {
|
||||
session_id: session_id.to_string(),
|
||||
sample_rate: 24000,
|
||||
channels: "mono".to_string(),
|
||||
frame_ms: 20,
|
||||
}),
|
||||
};
|
||||
Ok(serde_json::to_string(&env)?)
|
||||
}
|
||||
|
||||
pub fn encode_session_end(reason: &str, seq: u64, ts: u64) -> Result<String, TapProtoError> {
|
||||
let env = Envelope {
|
||||
v: PROTOCOL_VERSION,
|
||||
kind: FrameKind::SessionEnd,
|
||||
seq,
|
||||
ts,
|
||||
payload: Payload::SessionEnd(SessionEndPayload {
|
||||
reason: reason.to_string(),
|
||||
}),
|
||||
};
|
||||
Ok(serde_json::to_string(&env)?)
|
||||
}
|
||||
|
||||
pub fn encode_bye(reason: &str, seq: u64, ts: u64) -> Result<String, TapProtoError> {
|
||||
let env = Envelope {
|
||||
v: PROTOCOL_VERSION,
|
||||
kind: FrameKind::Bye,
|
||||
seq,
|
||||
ts,
|
||||
payload: Payload::Bye(ReasonPayload {
|
||||
reason: reason.to_string(),
|
||||
}),
|
||||
};
|
||||
Ok(serde_json::to_string(&env)?)
|
||||
}
|
||||
|
||||
pub fn encode_error(code: &str, msg: &str, seq: u64, ts: u64) -> Result<String, TapProtoError> {
|
||||
let env = Envelope {
|
||||
v: PROTOCOL_VERSION,
|
||||
kind: FrameKind::Error,
|
||||
seq,
|
||||
ts,
|
||||
payload: Payload::Error(ErrorPayload {
|
||||
code: code.to_string(),
|
||||
message: msg.to_string(),
|
||||
}),
|
||||
};
|
||||
Ok(serde_json::to_string(&env)?)
|
||||
}
|
||||
|
||||
/// Decode an incoming wire string into a `DecodedFrame`. Validates `v` and
|
||||
/// (for audio frames) `samples` and PCM round-trip. Unknown `type` values
|
||||
/// decode to `DecodedPayload::Unknown` (caller logs + counts + drops per
|
||||
/// spec §3.4).
|
||||
///
|
||||
/// # Deserialize strategy — `WireEnvelope` + dispatch on `kind`
|
||||
///
|
||||
/// We can't deserialize straight into `Envelope` because `Envelope` has no
|
||||
/// derived `Deserialize` (see its type-level doc). Instead we deserialize
|
||||
/// into a private `WireEnvelope` whose `#[serde(flatten)] extra:
|
||||
/// Map<String, Value>` hoovers up every field not named `v`/`type`/`seq`/
|
||||
/// `ts` — that's the payload's fields, exactly the flat wire shape spec §3.1
|
||||
/// mandates. Then we dispatch on `kind` (recovered via `FrameKind`'s
|
||||
/// `#[serde(other)] Unknown` fallback) and deserialize the `extra` map
|
||||
/// into the specific payload struct.
|
||||
///
|
||||
/// This is the cleanest way to (a) keep the flat wire shape, (b) distinguish
|
||||
/// `audio_in` from `audio_out` (they share `AudioPayload` shape — only
|
||||
/// `kind` tells them apart), and (c) route unknown `type` strings to
|
||||
/// `DecodedPayload::Unknown` rather than erroring (the brief's tagged-enum
|
||||
/// approach can't do (c)).
|
||||
pub fn decode_envelope(s: &str) -> Result<DecodedFrame, TapProtoError> {
|
||||
let w: WireEnvelope = serde_json::from_str(s)?;
|
||||
if w.v != PROTOCOL_VERSION {
|
||||
return Err(TapProtoError::Version {
|
||||
got: w.v,
|
||||
expected: PROTOCOL_VERSION,
|
||||
});
|
||||
}
|
||||
let extra_value = serde_json::Value::Object(w.extra);
|
||||
let payload = match w.kind {
|
||||
FrameKind::Hello => {
|
||||
let p: HelloPayload = serde_json::from_value(extra_value)?;
|
||||
DecodedPayload::Hello(p)
|
||||
}
|
||||
FrameKind::AudioIn => {
|
||||
let p: AudioPayload = serde_json::from_value(extra_value)?;
|
||||
if p.samples != SAMPLES_PER_FRAME {
|
||||
return Err(TapProtoError::SampleCount {
|
||||
got: p.samples,
|
||||
expected: SAMPLES_PER_FRAME,
|
||||
});
|
||||
}
|
||||
// Validate the PCM decodes (drops malformations early; hot-path
|
||||
// policy is "drop + observe" — caller logs + counts, doesn't crash).
|
||||
let _ = decode_pcm(&p.pcm, p.samples)?;
|
||||
DecodedPayload::AudioIn(p)
|
||||
}
|
||||
FrameKind::AudioOut => {
|
||||
let p: AudioPayload = serde_json::from_value(extra_value)?;
|
||||
if p.samples != SAMPLES_PER_FRAME {
|
||||
return Err(TapProtoError::SampleCount {
|
||||
got: p.samples,
|
||||
expected: SAMPLES_PER_FRAME,
|
||||
});
|
||||
}
|
||||
let _ = decode_pcm(&p.pcm, p.samples)?;
|
||||
DecodedPayload::AudioOut(p)
|
||||
}
|
||||
FrameKind::SessionEnd => {
|
||||
let p: SessionEndPayload = serde_json::from_value(extra_value)?;
|
||||
DecodedPayload::SessionEnd(p)
|
||||
}
|
||||
FrameKind::Bye => {
|
||||
let p: ReasonPayload = serde_json::from_value(extra_value)?;
|
||||
DecodedPayload::Bye(p)
|
||||
}
|
||||
FrameKind::Error => {
|
||||
let p: ErrorPayload = serde_json::from_value(extra_value)?;
|
||||
DecodedPayload::Error(p)
|
||||
}
|
||||
FrameKind::Unknown => DecodedPayload::Unknown,
|
||||
};
|
||||
Ok(DecodedFrame {
|
||||
v: w.v,
|
||||
seq: w.seq,
|
||||
ts: w.ts,
|
||||
payload,
|
||||
})
|
||||
}
|
||||
|
||||
/// Private intermediate for `decode_envelope`: parses the four named
|
||||
/// envelope fields and flattens every other JSON key (the payload fields)
|
||||
/// into `extra`. The `extra` map is then dispatched on `kind` to build
|
||||
/// the specific payload struct (see `decode_envelope`).
|
||||
///
|
||||
/// `seq` and `ts` carry `#[serde(default)]` because spec §3.1 marks `ts`
|
||||
/// as advisory and a malformed peer might omit `seq`; we'd rather decode-
|
||||
/// and-observe than reject. (No slice-2 test exercises this, but it
|
||||
/// matches the "drop + observe" posture of spec §3.8.)
|
||||
#[derive(Deserialize)]
|
||||
struct WireEnvelope {
|
||||
v: u8,
|
||||
#[serde(rename = "type")]
|
||||
kind: FrameKind,
|
||||
#[serde(default)]
|
||||
seq: u64,
|
||||
#[serde(default)]
|
||||
ts: u64,
|
||||
#[serde(flatten)]
|
||||
extra: serde_json::Map<String, serde_json::Value>,
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
#[test]
|
||||
fn audio_in_round_trips() {
|
||||
let mut frame = PcmFrame::zeroed();
|
||||
frame.samples[0] = 1234;
|
||||
frame.samples[479] = -5678;
|
||||
let wire = encode_audio_in(&frame, 7, 42).unwrap();
|
||||
let decoded = decode_envelope(&wire).unwrap();
|
||||
assert_eq!(decoded.v, 1);
|
||||
assert_eq!(decoded.seq, 7);
|
||||
assert_eq!(decoded.ts, 42);
|
||||
match decoded.payload {
|
||||
DecodedPayload::AudioIn(p) => {
|
||||
assert_eq!(p.samples, 480);
|
||||
let decoded_frame = decode_pcm(&p.pcm, p.samples).unwrap();
|
||||
assert_eq!(decoded_frame.samples[0], 1234);
|
||||
assert_eq!(decoded_frame.samples[479], -5678);
|
||||
}
|
||||
other => panic!("expected AudioIn, got {:?}", other),
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn audio_out_round_trips() {
|
||||
let frame = PcmFrame::zeroed();
|
||||
let wire = encode_audio_out(&frame, 0, 0).unwrap();
|
||||
let decoded = decode_envelope(&wire).unwrap();
|
||||
assert!(matches!(decoded.payload, DecodedPayload::AudioOut(_)));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn hello_round_trips() {
|
||||
let wire = encode_hello("550e8400-e29b-41d4-a716-446655440000", 0, 0).unwrap();
|
||||
let decoded = decode_envelope(&wire).unwrap();
|
||||
match decoded.payload {
|
||||
DecodedPayload::Hello(p) => {
|
||||
assert_eq!(p.session_id, "550e8400-e29b-41d4-a716-446655440000");
|
||||
assert_eq!(p.sample_rate, 24000);
|
||||
assert_eq!(p.channels, "mono");
|
||||
assert_eq!(p.frame_ms, 20);
|
||||
}
|
||||
other => panic!("expected Hello, got {:?}", other),
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn session_end_round_trips() {
|
||||
let wire = encode_session_end("hangup", 3, 100).unwrap();
|
||||
let decoded = decode_envelope(&wire).unwrap();
|
||||
match decoded.payload {
|
||||
DecodedPayload::SessionEnd(p) => assert_eq!(p.reason, "hangup"),
|
||||
other => panic!("expected SessionEnd, got {:?}", other),
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn bye_round_trips() {
|
||||
let wire = encode_bye("core_shutdown", 5, 200).unwrap();
|
||||
let decoded = decode_envelope(&wire).unwrap();
|
||||
assert!(matches!(decoded.payload, DecodedPayload::Bye(_)));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn error_round_trips() {
|
||||
let wire = encode_error("bad_samples", "expected 480", 1, 50).unwrap();
|
||||
let decoded = decode_envelope(&wire).unwrap();
|
||||
match decoded.payload {
|
||||
DecodedPayload::Error(p) => {
|
||||
assert_eq!(p.code, "bad_samples");
|
||||
assert_eq!(p.message, "expected 480");
|
||||
}
|
||||
other => panic!("expected Error, got {:?}", other),
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn pcm_is_explicit_little_endian() {
|
||||
let mut frame = PcmFrame::zeroed();
|
||||
frame.samples[0] = 0x0102; // 258 — LE bytes are [0x02, 0x01]
|
||||
let s = encode_pcm(&frame);
|
||||
use base64::Engine as _;
|
||||
let bytes = base64::engine::general_purpose::STANDARD
|
||||
.decode(&s)
|
||||
.unwrap();
|
||||
// First two bytes are samples[0] in LE: [0x02, 0x01].
|
||||
assert_eq!(bytes[0], 0x02);
|
||||
assert_eq!(bytes[1], 0x01);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn sample_count_mismatch_returns_error() {
|
||||
let wire = r#"{"v":1,"type":"audio_in","seq":0,"ts":0,"pcm":"AAAA","samples":100}"#;
|
||||
let err = decode_envelope(wire).unwrap_err();
|
||||
assert!(matches!(
|
||||
err,
|
||||
TapProtoError::SampleCount {
|
||||
got: 100,
|
||||
expected: 480
|
||||
}
|
||||
));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn unknown_type_decodes_to_unknown_variant() {
|
||||
let wire = r#"{"v":1,"type":"future_event","seq":0,"ts":0}"#;
|
||||
let decoded = decode_envelope(wire).unwrap();
|
||||
assert!(matches!(decoded.payload, DecodedPayload::Unknown));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn version_mismatch_returns_error() {
|
||||
let wire = r#"{"v":99,"type":"hello","seq":0,"ts":0,"session_id":"x"}"#;
|
||||
let err = decode_envelope(wire).unwrap_err();
|
||||
assert!(matches!(
|
||||
err,
|
||||
TapProtoError::Version {
|
||||
got: 99,
|
||||
expected: 1
|
||||
}
|
||||
));
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user