diff --git a/crates/rutster-tap/src/lib.rs b/crates/rutster-tap/src/lib.rs index 0bc7b44..af30ed2 100644 --- a/crates/rutster-tap/src/lib.rs +++ b/crates/rutster-tap/src/lib.rs @@ -41,8 +41,12 @@ pub use protocol::{ AudioPayload, DecodedFrame, DecodedPayload, Envelope, ErrorPayload, FrameKind, HelloPayload, PROTOCOL_VERSION, Payload, ReasonPayload, SAMPLES_PER_FRAME as WIRE_SAMPLES_PER_FRAME, SessionEndPayload, TapProtoError, decode_envelope, decode_pcm, encode_audio_in, - encode_audio_out, encode_bye, encode_error, encode_hello, encode_pcm, encode_session_end, + encode_audio_out, encode_bye, encode_error, encode_function_call, encode_function_call_output, + encode_hello, encode_pcm, encode_session_end, encode_speech_started, encode_speech_stopped, + encode_tools_update, }; +// 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}; diff --git a/crates/rutster-tap/src/protocol.rs b/crates/rutster-tap/src/protocol.rs index 9e3e75b..5724ce8 100644 --- a/crates/rutster-tap/src/protocol.rs +++ b/crates/rutster-tap/src/protocol.rs @@ -47,6 +47,21 @@ pub enum FrameKind { SessionEnd, Bye, Error, + /// brain → core: user speech started (advisory; translated from OpenAI + /// `input_audio_buffer.speech_started`; spec §3.2). + SpeechStarted, + /// brain → core: user speech stopped (advisory; OpenAI + /// `input_audio_buffer.speech_stopped`; spec §3.2). + SpeechStopped, + /// brain → core: brain proposes a tool call (translated from OpenAI + /// `response.function_call_arguments.done`; spec §3.2). + FunctionCall, + /// core → brain: tool registry reply (spec §3.3). + FunctionCallOutput, + /// brain → core: brain declares its tool catalog on hello + on changes + /// (spec §3.2). + #[serde(rename = "tools.update")] + ToolsUpdate, /// Unknown wire `type` values land here (spec §3.4: log + count + drop). /// `#[serde(other)]` catches any string not in the variants above. #[serde(other)] @@ -111,6 +126,11 @@ impl Serialize for Envelope { Payload::AudioIn(_) | Payload::AudioOut(_) => 2, // pcm, samples Payload::SessionEnd(_) | Payload::Bye(_) => 1, // reason Payload::Error(_) => 2, // code, message + // Slice-3 adds: empty payload (0 fields), 3-field payloads, 1-field payload. + Payload::SpeechStarted | Payload::SpeechStopped => 0, + Payload::FunctionCall(_) => 3, // id, name, args + Payload::FunctionCallOutput(_) => 3, // id, status, result + Payload::ToolsUpdate(_) => 1, // tools }; let mut st = serializer.serialize_struct("Envelope", 4 + payload_field_count)?; st.serialize_field("v", &self.v)?; @@ -142,6 +162,23 @@ impl Serialize for Envelope { st.serialize_field("code", &p.code)?; st.serialize_field("message", &p.message)?; } + Payload::SpeechStarted | Payload::SpeechStopped => { + // No payload fields — the envelope's `v`/`type`/`seq`/`ts` + // is the whole message. The event name IS the message. + } + Payload::FunctionCall(p) => { + st.serialize_field("id", &p.id)?; + st.serialize_field("name", &p.name)?; + st.serialize_field("args", &p.args)?; + } + Payload::FunctionCallOutput(p) => { + st.serialize_field("id", &p.id)?; + st.serialize_field("status", &p.status)?; + st.serialize_field("result", &p.result)?; + } + Payload::ToolsUpdate(p) => { + st.serialize_field("tools", &p.tools)?; + } } st.end() } @@ -165,6 +202,16 @@ pub enum Payload { SessionEnd(SessionEndPayload), Bye(ReasonPayload), Error(ErrorPayload), + /// Slice-3 additive (spec §3.2): brain → core, advisory; empty payload + /// (the event name IS the message). + SpeechStarted, + SpeechStopped, + /// Slice-3 additive (spec §3.2). + FunctionCall(FunctionCallPayload), + /// Slice-3 additive (spec §3.3): core → brain reply. + FunctionCallOutput(FunctionCallOutputPayload), + /// Slice-3 additive (spec §3.2): brain → core catalog declaration. + ToolsUpdate(ToolsUpdatePayload), } #[derive(Debug, Clone, Serialize, Deserialize)] @@ -211,6 +258,42 @@ pub struct ErrorPayload { pub message: String, } +/// `function_call` payload (brain → core; spec §3.2). Carries the +/// brain-minted id + tool name + args. Args is a raw JSON Value (not a +/// typed struct) so any tool schema is allowed. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct FunctionCallPayload { + pub id: String, + pub name: String, + /// Raw JSON arguments — the tool registry dispatches by name and lets + /// each Tool impl parse the args itself. (OpenAI sends `arguments` as a + /// JSON string; the translator parses it back to a Value before + /// emitting the function_call tap frame.) + pub args: serde_json::Value, +} + +/// `function_call_output` payload (core → brain; spec §3.3). The reply for +/// a `function_call`. `status` is one of `"ok"`, `"error"`, `"not_implemented"`. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct FunctionCallOutputPayload { + pub id: String, + pub status: String, + pub result: serde_json::Value, +} + +/// `tools.update` payload (brain → core; spec §3.2). The brain declares its +/// tool catalog so the core's tool registry can validate function_call +/// events by name. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct ToolsUpdatePayload { + /// An array of tool descriptors (each has `name`, `description`, + /// `parameters`). The shape is intentionally permissive (a JSON array, + /// not a typed Vec) so the brain can declare schemas the + /// core doesn't know about — the core only checks the `name` field for + /// dispatch, ignores the rest. + pub tools: serde_json::Value, +} + /// 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 @@ -231,7 +314,17 @@ pub enum DecodedPayload { SessionEnd(SessionEndPayload), Bye(ReasonPayload), Error(ErrorPayload), - /// Unknown `type` — log + count + drop (spec §3.4). + /// Slice-3 (spec §3.2): the brain detected user speech started/stopped. + SpeechStarted, + SpeechStopped, + /// Slice-3 (spec §3.2): brain wants the core to execute a tool. + FunctionCall(FunctionCallPayload), + /// Slice-3 (spec §3.3): the core's tool-registry reply. + FunctionCallOutput(FunctionCallOutputPayload), + /// Slice-3 (spec §3.2): brain declares its catalog so the core can + /// validate function_call events. + ToolsUpdate(ToolsUpdatePayload), + /// Unknown `type` — log + count + drop (spec §3.4 of slice-2). Unknown, } @@ -369,6 +462,111 @@ pub fn encode_error(code: &str, msg: &str, seq: u64, ts: u64) -> Result Result { + let env = Envelope { + v: PROTOCOL_VERSION, + kind: FrameKind::SpeechStarted, + seq, + ts, + payload: Payload::SpeechStarted, + }; + Ok(serde_json::to_string(&env)?) +} + +/// Build `speech_stopped` (brain → core, advisory; spec §3.2). +pub fn encode_speech_stopped(seq: u64, ts: u64) -> Result { + let env = Envelope { + v: PROTOCOL_VERSION, + kind: FrameKind::SpeechStopped, + seq, + ts, + payload: Payload::SpeechStopped, + }; + Ok(serde_json::to_string(&env)?) +} + +/// Build `function_call` (brain → core; spec §3.2). `args_json_str` is the +/// raw JSON string the brain's translator parses from OpenAI's +/// `response.function_call_arguments.done.arguments` (which is itself a +/// JSON string in OpenAI's wire format). +pub fn encode_function_call( + id: &str, + name: &str, + args_json_str: &str, + seq: u64, + ts: u64, +) -> Result { + let args: serde_json::Value = if args_json_str.is_empty() { + serde_json::Value::Null + } else { + serde_json::from_str(args_json_str)? + }; + let env = Envelope { + v: PROTOCOL_VERSION, + kind: FrameKind::FunctionCall, + seq, + ts, + payload: Payload::FunctionCall(FunctionCallPayload { + id: id.to_string(), + name: name.to_string(), + args, + }), + }; + Ok(serde_json::to_string(&env)?) +} + +/// Build `function_call_output` (core → brain; spec §3.3). `result_json_str` +/// is the raw JSON string the tool-registry dispatch returns (serialized +/// into the result field of the payload). +pub fn encode_function_call_output( + id: &str, + status: &str, // "ok" | "error" | "not_implemented" + result_json_str: &str, + seq: u64, + ts: u64, +) -> Result { + let result: serde_json::Value = if result_json_str.is_empty() { + serde_json::Value::Null + } else { + serde_json::from_str(result_json_str)? + }; + let env = Envelope { + v: PROTOCOL_VERSION, + kind: FrameKind::FunctionCallOutput, + seq, + ts, + payload: Payload::FunctionCallOutput(FunctionCallOutputPayload { + id: id.to_string(), + status: status.to_string(), + result, + }), + }; + Ok(serde_json::to_string(&env)?) +} + +/// Build `tools.update` (brain → core; spec §3.2). `tools_json_str` is the +/// raw JSON array of tool descriptors. +pub fn encode_tools_update( + tools_json_str: &str, + seq: u64, + ts: u64, +) -> Result { + let tools: serde_json::Value = if tools_json_str.is_empty() { + serde_json::Value::Array(vec![]) + } else { + serde_json::from_str(tools_json_str)? + }; + let env = Envelope { + v: PROTOCOL_VERSION, + kind: FrameKind::ToolsUpdate, + seq, + ts, + payload: Payload::ToolsUpdate(ToolsUpdatePayload { tools }), + }; + 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 @@ -440,6 +638,20 @@ pub fn decode_envelope(s: &str) -> Result { let p: ErrorPayload = serde_json::from_value(extra_value)?; DecodedPayload::Error(p) } + FrameKind::SpeechStarted => DecodedPayload::SpeechStarted, + FrameKind::SpeechStopped => DecodedPayload::SpeechStopped, + FrameKind::FunctionCall => { + let p: FunctionCallPayload = serde_json::from_value(extra_value)?; + DecodedPayload::FunctionCall(p) + } + FrameKind::FunctionCallOutput => { + let p: FunctionCallOutputPayload = serde_json::from_value(extra_value)?; + DecodedPayload::FunctionCallOutput(p) + } + FrameKind::ToolsUpdate => { + let p: ToolsUpdatePayload = serde_json::from_value(extra_value)?; + DecodedPayload::ToolsUpdate(p) + } FrameKind::Unknown => DecodedPayload::Unknown, }; Ok(DecodedFrame { @@ -596,4 +808,96 @@ mod tests { } )); } + + /// Slice-3 additive event types must round-trip through (de)serialization + /// without breaking slice-2's v1 contract. Every new kind + payload. + #[test] + fn speech_started_round_trips() { + let s = encode_speech_started(7, 100).unwrap(); + assert!(s.contains("\"type\":\"speech_started\"")); + assert!(s.contains("\"v\":1")); + let d = decode_envelope(&s).unwrap(); + assert_eq!(d.seq, 7); + assert_eq!(d.ts, 100); + assert!(matches!(d.payload, DecodedPayload::SpeechStarted)); + } + + #[test] + fn speech_stopped_round_trips() { + let s = encode_speech_stopped(9, 200).unwrap(); + assert!(s.contains("\"type\":\"speech_stopped\"")); + let d = decode_envelope(&s).unwrap(); + assert_eq!(d.seq, 9); + assert!(matches!(d.payload, DecodedPayload::SpeechStopped)); + } + + #[test] + fn function_call_round_trips() { + let s = encode_function_call("abc-123", "hangup", "{}", 0, 0).unwrap(); + assert!(s.contains("\"type\":\"function_call\"")); + assert!(s.contains("\"id\":\"abc-123\"")); + assert!(s.contains("\"name\":\"hangup\"")); + let d = decode_envelope(&s).unwrap(); + match d.payload { + DecodedPayload::FunctionCall(p) => { + assert_eq!(p.id, "abc-123"); + assert_eq!(p.name, "hangup"); + } + other => panic!("expected FunctionCall, got {other:?}"), + } + } + + #[test] + fn function_call_output_round_trips() { + let s = + encode_function_call_output("abc-123", "ok", r#"{"channel_state":"Closing"}"#, 0, 0) + .unwrap(); + assert!(s.contains("\"type\":\"function_call_output\"")); + assert!(s.contains("\"status\":\"ok\"")); + let d = decode_envelope(&s).unwrap(); + match d.payload { + DecodedPayload::FunctionCallOutput(p) => { + assert_eq!(p.id, "abc-123"); + assert_eq!(p.status, "ok"); + } + other => panic!("expected FunctionCallOutput, got {other:?}"), + } + } + + #[test] + fn tools_update_round_trips() { + let tools_json = r#"[{"name":"hangup","description":"hang up the call"}]"#; + let s = encode_tools_update(tools_json, 0, 0).unwrap(); + assert!(s.contains("\"type\":\"tools.update\"")); + assert!(s.contains("\"tools\":[")); + let d = decode_envelope(&s).unwrap(); + match d.payload { + DecodedPayload::ToolsUpdate(p) => { + assert!(p.tools.is_array()); + assert_eq!(p.tools.as_array().unwrap().len(), 1); + } + other => panic!("expected ToolsUpdate, got {other:?}"), + } + } + + /// Forwards-compat: slice-2's echo brain sees the new types as unknown + /// (the `#[serde(other)]` on the old enum absorbed them). With the new + /// enum in place, the kinds decode to their new variants (rather than + /// Unknown). This test asserts the decode no longer drops them. + #[test] + fn new_kinds_decode_to_their_variants_not_unknown() { + for s in [ + encode_speech_started(0, 0).unwrap(), + encode_speech_stopped(0, 0).unwrap(), + encode_function_call("x", "x", "null", 0, 0).unwrap(), + encode_function_call_output("x", "ok", "null", 0, 0).unwrap(), + encode_tools_update("[]", 0, 0).unwrap(), + ] { + let d = decode_envelope(&s).unwrap(); + assert!( + !matches!(d.payload, DecodedPayload::Unknown), + "new event type decoded as Unknown: {s}" + ); + } + } } diff --git a/crates/rutster-tap/src/tap_client.rs b/crates/rutster-tap/src/tap_client.rs index 4700d95..7d4219f 100644 --- a/crates/rutster-tap/src/tap_client.rs +++ b/crates/rutster-tap/src/tap_client.rs @@ -307,6 +307,19 @@ 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(_) => { + metrics.unknown_frames.fetch_add(1, Ordering::Relaxed); + debug!("slice-3 event type not yet handled; dropping"); + } } let _ = session_start; // used for ts computation if added later }