gemini_genai_rs/transport/
codec.rs

1//! Message codec — encode commands, decode server messages.
2
3use base64::Engine;
4
5use crate::protocol::messages::*;
6use crate::protocol::types::*;
7use crate::session::SessionCommand;
8
9/// Error during encoding or decoding.
10#[derive(Debug, thiserror::Error, Clone)]
11pub enum CodecError {
12    /// Failed to serialize a client message to JSON.
13    #[error("Serialization error: {0}")]
14    Serialize(String),
15    /// Failed to deserialize a server message from JSON.
16    #[error("Deserialization error: {0}")]
17    Deserialize(String),
18    /// Server sent bytes that are not valid UTF-8.
19    #[error("Invalid UTF-8")]
20    InvalidUtf8,
21}
22
23/// Encodes client commands into wire bytes and decodes server bytes into messages.
24///
25/// The default implementation is [`JsonCodec`], which serializes commands as JSON
26/// and parses server responses via [`ServerMessage::parse`].
27///
28/// # Implementors
29///
30/// - [`JsonCodec`] -- Standard JSON codec. Encodes setup messages, audio (base64),
31///   text, tool responses, and activity signals. Decodes server JSON into
32///   [`ServerMessage`] variants. Handles platform-specific wire stripping
33///   (e.g., removing `scheduling` fields for Vertex AI).
34pub trait Codec: Send + Sync + 'static {
35    /// Encode the initial setup message for the given session configuration.
36    fn encode_setup(&self, config: &SessionConfig) -> Result<Vec<u8>, CodecError>;
37    /// Encode a session command into wire bytes.
38    fn encode_command(
39        &self,
40        cmd: &SessionCommand,
41        config: &SessionConfig,
42    ) -> Result<Vec<u8>, CodecError>;
43    /// Decode raw bytes from the server into a `ServerMessage`.
44    fn decode_message(&self, data: &[u8]) -> Result<ServerMessage, CodecError>;
45}
46
47/// Default JSON codec — current behavior extracted from connection.rs.
48pub struct JsonCodec;
49
50impl Codec for JsonCodec {
51    fn encode_setup(&self, config: &SessionConfig) -> Result<Vec<u8>, CodecError> {
52        serde_json::to_vec(&config.to_setup_message())
53            .map_err(|e| CodecError::Serialize(e.to_string()))
54    }
55
56    fn encode_command(
57        &self,
58        cmd: &SessionCommand,
59        config: &SessionConfig,
60    ) -> Result<Vec<u8>, CodecError> {
61        match cmd {
62            SessionCommand::SendAudio(data) => {
63                let encoded = base64::engine::general_purpose::STANDARD.encode(data);
64                let msg = RealtimeInputMessage {
65                    realtime_input: RealtimeInputPayload {
66                        media_chunks: Vec::new(),
67                        audio: Some(Blob {
68                            mime_type: config.input_audio_format.mime_type().to_string(),
69                            data: encoded,
70                        }),
71                        video: None,
72                        audio_stream_end: None,
73                        text: None,
74                    },
75                };
76                serde_json::to_vec(&msg).map_err(|e| CodecError::Serialize(e.to_string()))
77            }
78            SessionCommand::SendText(text) => {
79                let msg = ClientContentMessage {
80                    client_content: ClientContentPayload {
81                        turns: vec![Content::user(text)],
82                        turn_complete: Some(true),
83                    },
84                };
85                serde_json::to_vec(&msg).map_err(|e| CodecError::Serialize(e.to_string()))
86            }
87            SessionCommand::SendToolResponse(responses) => {
88                let function_responses = if config.supports_async_tools() {
89                    responses.clone()
90                } else {
91                    responses
92                        .iter()
93                        .map(|r| {
94                            let mut r = r.clone();
95                            r.scheduling = None;
96                            r
97                        })
98                        .collect()
99                };
100                let msg = ToolResponseMessage {
101                    tool_response: ToolResponsePayload { function_responses },
102                };
103                serde_json::to_vec(&msg).map_err(|e| CodecError::Serialize(e.to_string()))
104            }
105            // Explicit activity signals are suppressed unless the caller turned
106            // *off* automatic detection. The two are mutually exclusive on the
107            // wire: sending one while server VAD is enabled draws a close frame
108            // — code 1007, "Explicit activity control is not supported when
109            // automatic activity detection is enabled" — and the session dies.
110            //
111            // This is not a hypothetical. `LiveHandle::send_audio` runs
112            // client-side VAD and calls `user_speech_started` on the first
113            // speech frame, which lands here as `ActivityStart`. With server VAD
114            // on by default, *every* audio session was killed the moment the
115            // user began speaking: the socket closed, the loop reconnected, the
116            // next utterance killed it again. Text sessions were unaffected,
117            // which is why nothing caught it — every Live test in this workspace
118            // drove sessions with `send_text`.
119            //
120            // An empty encoding is the codec's existing idiom for "this does not
121            // belong on this wire"; the session loop skips empty payloads.
122            SessionCommand::ActivityStart if config.automatic_activity_detection_enabled() => {
123                Ok(Vec::new())
124            }
125            SessionCommand::ActivityEnd if config.automatic_activity_detection_enabled() => {
126                Ok(Vec::new())
127            }
128            SessionCommand::ActivityStart => {
129                let msg = ActivitySignalMessage {
130                    realtime_input: ActivitySignalPayload {
131                        activity_start: Some(ActivityStart {}),
132                        activity_end: None,
133                    },
134                };
135                serde_json::to_vec(&msg).map_err(|e| CodecError::Serialize(e.to_string()))
136            }
137            SessionCommand::ActivityEnd => {
138                let msg = ActivitySignalMessage {
139                    realtime_input: ActivitySignalPayload {
140                        activity_start: None,
141                        activity_end: Some(ActivityEnd {}),
142                    },
143                };
144                serde_json::to_vec(&msg).map_err(|e| CodecError::Serialize(e.to_string()))
145            }
146            SessionCommand::SendClientContent {
147                turns,
148                turn_complete,
149            } => {
150                let msg = ClientContentMessage {
151                    client_content: ClientContentPayload {
152                        turns: turns.clone(),
153                        turn_complete: Some(*turn_complete),
154                    },
155                };
156                serde_json::to_vec(&msg).map_err(|e| CodecError::Serialize(e.to_string()))
157            }
158            SessionCommand::SendVideo(data) => {
159                let encoded = base64::engine::general_purpose::STANDARD.encode(data);
160                let msg = RealtimeInputMessage {
161                    realtime_input: RealtimeInputPayload {
162                        media_chunks: Vec::new(),
163                        audio: None,
164                        video: Some(Blob {
165                            mime_type: "image/jpeg".to_string(),
166                            data: encoded,
167                        }),
168                        audio_stream_end: None,
169                        text: None,
170                    },
171                };
172                serde_json::to_vec(&msg).map_err(|e| CodecError::Serialize(e.to_string()))
173            }
174            SessionCommand::UpdateInstruction(instruction) => {
175                let msg = ClientContentMessage {
176                    client_content: ClientContentPayload {
177                        turns: vec![Content {
178                            role: Some(Role::System),
179                            parts: vec![Part::Text {
180                                text: instruction.clone(),
181                            }],
182                        }],
183                        turn_complete: Some(false),
184                    },
185                };
186                serde_json::to_vec(&msg).map_err(|e| CodecError::Serialize(e.to_string()))
187            }
188            SessionCommand::Disconnect => Ok(Vec::new()),
189        }
190    }
191
192    fn decode_message(&self, data: &[u8]) -> Result<ServerMessage, CodecError> {
193        let text = std::str::from_utf8(data).map_err(|_| CodecError::InvalidUtf8)?;
194        ServerMessage::parse(text).map_err(|e| CodecError::Deserialize(e.to_string()))
195    }
196}
197
198#[cfg(test)]
199mod tests {
200    use super::*;
201
202    fn test_config() -> SessionConfig {
203        SessionConfig::new("test-key")
204            .model(GeminiModel::Gemini2_0FlashLive)
205            .voice(Voice::Puck)
206    }
207
208    // -----------------------------------------------------------------------
209    // Encode tests
210    // -----------------------------------------------------------------------
211
212    #[test]
213    fn json_codec_encode_setup() {
214        let codec = JsonCodec;
215        let config = test_config();
216        let bytes = codec.encode_setup(&config).unwrap();
217        let json = String::from_utf8(bytes).unwrap();
218        assert!(json.contains("\"setup\""), "should contain setup key");
219        assert!(
220            json.contains("gemini-2.0-flash-live-001"),
221            "should contain model name"
222        );
223    }
224
225    #[test]
226    fn json_codec_encode_send_text() {
227        let codec = JsonCodec;
228        let config = test_config();
229        let cmd = SessionCommand::SendText("Hello, world!".to_string());
230        let bytes = codec.encode_command(&cmd, &config).unwrap();
231        let json = String::from_utf8(bytes).unwrap();
232        assert!(
233            json.contains("\"clientContent\""),
234            "should contain clientContent"
235        );
236        assert!(
237            json.contains("Hello, world!"),
238            "should contain the text payload"
239        );
240        assert!(
241            json.contains("\"turnComplete\""),
242            "should contain turnComplete"
243        );
244    }
245
246    #[test]
247    fn json_codec_encode_send_audio() {
248        let codec = JsonCodec;
249        let config = test_config();
250        let audio_data = vec![1u8, 2, 3, 4];
251        let cmd = SessionCommand::SendAudio(audio_data);
252        let bytes = codec.encode_command(&cmd, &config).unwrap();
253        let json = String::from_utf8(bytes).unwrap();
254        assert!(
255            json.contains("\"realtimeInput\""),
256            "should contain realtimeInput"
257        );
258        assert!(json.contains("\"audio\""), "should contain audio field");
259        assert!(
260            json.contains("audio/pcm"),
261            "should contain the audio mime type"
262        );
263        // base64 of [1,2,3,4] is "AQIDBA=="
264        assert!(
265            json.contains("AQIDBA=="),
266            "should contain base64-encoded data"
267        );
268    }
269
270    #[test]
271    fn json_codec_encode_tool_response() {
272        let codec = JsonCodec;
273        let config = test_config();
274        let cmd = SessionCommand::SendToolResponse(vec![FunctionResponse {
275            name: "get_weather".to_string(),
276            response: serde_json::json!({"temp": 22}),
277            id: Some("call-1".to_string()),
278            scheduling: None,
279        }]);
280        let bytes = codec.encode_command(&cmd, &config).unwrap();
281        let json = String::from_utf8(bytes).unwrap();
282        assert!(
283            json.contains("\"toolResponse\""),
284            "should contain toolResponse"
285        );
286        assert!(
287            json.contains("\"functionResponses\""),
288            "should contain functionResponses"
289        );
290        assert!(
291            json.contains("get_weather"),
292            "should contain the function name"
293        );
294    }
295
296    #[test]
297    fn json_codec_strips_scheduling_for_vertex() {
298        let codec = JsonCodec;
299        let config = SessionConfig::from_vertex("proj", "us-central1", "token")
300            .model(GeminiModel::Gemini2_0FlashLive);
301        let cmd = SessionCommand::SendToolResponse(vec![FunctionResponse {
302            name: "search".to_string(),
303            response: serde_json::json!({"ok": true}),
304            id: Some("call-1".to_string()),
305            scheduling: Some(FunctionResponseScheduling::WhenIdle),
306        }]);
307        let bytes = codec.encode_command(&cmd, &config).unwrap();
308        let json = String::from_utf8(bytes).unwrap();
309        assert!(
310            !json.contains("scheduling"),
311            "Vertex AI should strip scheduling from tool responses"
312        );
313    }
314
315    #[test]
316    fn json_codec_preserves_scheduling_for_google_ai() {
317        let codec = JsonCodec;
318        let config = test_config();
319        let cmd = SessionCommand::SendToolResponse(vec![FunctionResponse {
320            name: "search".to_string(),
321            response: serde_json::json!({"ok": true}),
322            id: Some("call-1".to_string()),
323            scheduling: Some(FunctionResponseScheduling::WhenIdle),
324        }]);
325        let bytes = codec.encode_command(&cmd, &config).unwrap();
326        let json = String::from_utf8(bytes).unwrap();
327        assert!(
328            json.contains("WHEN_IDLE"),
329            "Google AI should preserve scheduling in tool responses"
330        );
331    }
332
333    #[test]
334    fn manual_activity_signals_are_suppressed_under_automatic_detection() {
335        // The default config leaves server VAD on, and the two are mutually
336        // exclusive on the wire: emitting one draws a 1007 close and kills the
337        // session mid-utterance. An empty encoding is the codec's idiom for
338        // "not on this wire"; the session loop skips it.
339        let codec = JsonCodec;
340        let config = test_config();
341        assert!(config.automatic_activity_detection_enabled());
342
343        for cmd in [SessionCommand::ActivityStart, SessionCommand::ActivityEnd] {
344            let bytes = codec.encode_command(&cmd, &config).unwrap();
345            assert!(
346                bytes.is_empty(),
347                "explicit activity control must not reach the wire while the \
348                 server is detecting activity itself; got {:?}",
349                String::from_utf8_lossy(&bytes)
350            );
351        }
352    }
353
354    #[test]
355    fn manual_activity_signals_are_sent_when_automatic_detection_is_off() {
356        let codec = JsonCodec;
357        let mut config = test_config();
358        config.realtime_input_config = Some(crate::protocol::types::RealtimeInputConfig {
359            automatic_activity_detection: Some(
360                crate::protocol::types::AutomaticActivityDetection {
361                    disabled: Some(true),
362                    start_of_speech_sensitivity: None,
363                    end_of_speech_sensitivity: None,
364                    prefix_padding_ms: None,
365                    silence_duration_ms: None,
366                },
367            ),
368            activity_handling: None,
369            turn_coverage: None,
370        });
371        assert!(!config.automatic_activity_detection_enabled());
372
373        let json = String::from_utf8(
374            codec
375                .encode_command(&SessionCommand::ActivityStart, &config)
376                .unwrap(),
377        )
378        .unwrap();
379        assert!(json.contains("\"activityStart\""), "{json}");
380        assert!(!json.contains("\"activityEnd\""), "{json}");
381
382        let json = String::from_utf8(
383            codec
384                .encode_command(&SessionCommand::ActivityEnd, &config)
385                .unwrap(),
386        )
387        .unwrap();
388        assert!(json.contains("\"activityEnd\""), "{json}");
389        assert!(!json.contains("\"activityStart\""), "{json}");
390    }
391
392    #[test]
393    fn json_codec_encode_client_content() {
394        let codec = JsonCodec;
395        let config = test_config();
396        let cmd = SessionCommand::SendClientContent {
397            turns: vec![Content::user("context message")],
398            turn_complete: false,
399        };
400        let bytes = codec.encode_command(&cmd, &config).unwrap();
401        let json = String::from_utf8(bytes).unwrap();
402        assert!(
403            json.contains("\"clientContent\""),
404            "should contain clientContent"
405        );
406        assert!(
407            json.contains("context message"),
408            "should contain the text content"
409        );
410        assert!(
411            json.contains("\"turnComplete\":false"),
412            "should contain turnComplete set to false"
413        );
414    }
415
416    #[test]
417    fn json_codec_encode_send_video() {
418        let codec = JsonCodec;
419        let config = test_config();
420        let cmd = SessionCommand::SendVideo(vec![0xFF, 0xD8, 0xFF]); // JPEG magic bytes
421        let bytes = codec.encode_command(&cmd, &config).unwrap();
422        let json: serde_json::Value = serde_json::from_slice(&bytes).unwrap();
423        assert_eq!(
424            json["realtimeInput"]["video"]["mimeType"].as_str().unwrap(),
425            "image/jpeg"
426        );
427        assert!(json["realtimeInput"]["video"]["data"].is_string());
428    }
429
430    #[test]
431    fn json_codec_encode_update_instruction() {
432        let codec = JsonCodec;
433        let config = test_config();
434        let cmd = SessionCommand::UpdateInstruction("New instruction".into());
435        let bytes = codec.encode_command(&cmd, &config).unwrap();
436        let json: serde_json::Value = serde_json::from_slice(&bytes).unwrap();
437        let turns = &json["clientContent"]["turns"];
438        assert_eq!(turns[0]["role"], "system");
439        assert_eq!(turns[0]["parts"][0]["text"], "New instruction");
440    }
441
442    #[test]
443    fn json_codec_encode_disconnect() {
444        let codec = JsonCodec;
445        let config = test_config();
446        let cmd = SessionCommand::Disconnect;
447        let bytes = codec.encode_command(&cmd, &config).unwrap();
448        assert!(bytes.is_empty(), "Disconnect should produce empty bytes");
449    }
450
451    // -----------------------------------------------------------------------
452    // Decode tests
453    // -----------------------------------------------------------------------
454
455    #[test]
456    fn json_codec_decode_setup_complete() {
457        let codec = JsonCodec;
458        let json = r#"{"setupComplete":{"sessionResumption":{"handle":"abc123"}}}"#;
459        let msg = codec.decode_message(json.as_bytes()).unwrap();
460        match msg {
461            ServerMessage::SetupComplete(sc) => {
462                let handle = sc.setup_complete.session_resumption.unwrap().handle;
463                assert_eq!(handle, Some("abc123".to_string()));
464            }
465            _ => panic!("Expected SetupComplete"),
466        }
467    }
468
469    #[test]
470    fn json_codec_decode_server_content() {
471        let codec = JsonCodec;
472        let json = r#"{
473            "serverContent": {
474                "modelTurn": {
475                    "parts": [{"text": "Hello! How can I help?"}]
476                },
477                "turnComplete": true
478            }
479        }"#;
480        let msg = codec.decode_message(json.as_bytes()).unwrap();
481        match msg {
482            ServerMessage::ServerContent(sc) => {
483                assert!(sc.server_content.turn_complete.unwrap_or(false));
484                let turn = sc.server_content.model_turn.unwrap();
485                assert_eq!(turn.parts.len(), 1);
486                match &turn.parts[0] {
487                    Part::Text { text } => assert_eq!(text, "Hello! How can I help?"),
488                    _ => panic!("Expected text part"),
489                }
490            }
491            _ => panic!("Expected ServerContent"),
492        }
493    }
494
495    #[test]
496    fn json_codec_decode_tool_call() {
497        let codec = JsonCodec;
498        let json = r#"{
499            "toolCall": {
500                "functionCalls": [
501                    {"name": "get_weather", "args": {"city": "London"}, "id": "call-1"}
502                ]
503            }
504        }"#;
505        let msg = codec.decode_message(json.as_bytes()).unwrap();
506        match msg {
507            ServerMessage::ToolCall(tc) => {
508                assert_eq!(tc.tool_call.function_calls.len(), 1);
509                assert_eq!(tc.tool_call.function_calls[0].name, "get_weather");
510            }
511            _ => panic!("Expected ToolCall"),
512        }
513    }
514
515    #[test]
516    fn json_codec_decode_invalid_utf8() {
517        let codec = JsonCodec;
518        let bad_bytes: &[u8] = &[0xFF, 0xFE, 0xFD];
519        let result = codec.decode_message(bad_bytes);
520        match result {
521            Err(CodecError::InvalidUtf8) => {} // expected
522            other => panic!("Expected CodecError::InvalidUtf8, got {:?}", other),
523        }
524    }
525}
gemini_genai_rs/transport/codec.rs

gemini_genai_rs/transport/
codec.rs
