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            // Vertex AI documents a `system`-role turn for this; Google AI
175            // closes the session over one (1007), so there the update goes
176            // out as a user-role turn that says what it is. `turnComplete:
177            // false` either way: the model takes it in without answering.
178            SessionCommand::UpdateInstruction(instruction) => {
179                let (role, text) = if config.supports_system_role_updates() {
180                    (Role::System, instruction.clone())
181                } else {
182                    (
183                        Role::User,
184                        format!(
185                            "System instruction update, which replaces your previous \
186                             instructions: {instruction}"
187                        ),
188                    )
189                };
190                let msg = ClientContentMessage {
191                    client_content: ClientContentPayload {
192                        turns: vec![Content {
193                            role: Some(role),
194                            parts: vec![Part::Text { text }],
195                        }],
196                        turn_complete: Some(false),
197                    },
198                };
199                serde_json::to_vec(&msg).map_err(|e| CodecError::Serialize(e.to_string()))
200            }
201            SessionCommand::Disconnect => Ok(Vec::new()),
202        }
203    }
204
205    fn decode_message(&self, data: &[u8]) -> Result<ServerMessage, CodecError> {
206        let text = std::str::from_utf8(data).map_err(|_| CodecError::InvalidUtf8)?;
207        ServerMessage::parse(text).map_err(|e| CodecError::Deserialize(e.to_string()))
208    }
209}
210
211#[cfg(test)]
212mod tests {
213    use super::*;
214
215    fn test_config() -> SessionConfig {
216        SessionConfig::new("test-key")
217            .model(ModelId::from_static("models/gemini-2.0-flash-live-001"))
218            .voice(Voice::Puck)
219    }
220
221    // -----------------------------------------------------------------------
222    // Encode tests
223    // -----------------------------------------------------------------------
224
225    #[test]
226    fn json_codec_encode_setup() {
227        let codec = JsonCodec;
228        let config = test_config();
229        let bytes = codec.encode_setup(&config).unwrap();
230        let json = String::from_utf8(bytes).unwrap();
231        assert!(json.contains("\"setup\""), "should contain setup key");
232        assert!(
233            json.contains("gemini-2.0-flash-live-001"),
234            "should contain model name"
235        );
236    }
237
238    #[test]
239    fn json_codec_encode_send_text() {
240        let codec = JsonCodec;
241        let config = test_config();
242        let cmd = SessionCommand::SendText("Hello, world!".to_string());
243        let bytes = codec.encode_command(&cmd, &config).unwrap();
244        let json = String::from_utf8(bytes).unwrap();
245        assert!(
246            json.contains("\"clientContent\""),
247            "should contain clientContent"
248        );
249        assert!(
250            json.contains("Hello, world!"),
251            "should contain the text payload"
252        );
253        assert!(
254            json.contains("\"turnComplete\""),
255            "should contain turnComplete"
256        );
257    }
258
259    #[test]
260    fn json_codec_encode_send_audio() {
261        let codec = JsonCodec;
262        let config = test_config();
263        let audio_data = vec![1u8, 2, 3, 4];
264        let cmd = SessionCommand::SendAudio(audio_data.into());
265        let bytes = codec.encode_command(&cmd, &config).unwrap();
266        let json = String::from_utf8(bytes).unwrap();
267        assert!(
268            json.contains("\"realtimeInput\""),
269            "should contain realtimeInput"
270        );
271        assert!(json.contains("\"audio\""), "should contain audio field");
272        assert!(
273            json.contains("audio/pcm"),
274            "should contain the audio mime type"
275        );
276        // base64 of [1,2,3,4] is "AQIDBA=="
277        assert!(
278            json.contains("AQIDBA=="),
279            "should contain base64-encoded data"
280        );
281    }
282
283    #[test]
284    fn json_codec_encode_tool_response() {
285        let codec = JsonCodec;
286        let config = test_config();
287        let cmd = SessionCommand::SendToolResponse(vec![FunctionResponse {
288            name: "get_weather".to_string(),
289            response: serde_json::json!({"temp": 22}),
290            id: Some("call-1".to_string()),
291            scheduling: None,
292        }]);
293        let bytes = codec.encode_command(&cmd, &config).unwrap();
294        let json = String::from_utf8(bytes).unwrap();
295        assert!(
296            json.contains("\"toolResponse\""),
297            "should contain toolResponse"
298        );
299        assert!(
300            json.contains("\"functionResponses\""),
301            "should contain functionResponses"
302        );
303        assert!(
304            json.contains("get_weather"),
305            "should contain the function name"
306        );
307    }
308
309    #[test]
310    fn json_codec_strips_scheduling_for_vertex() {
311        let codec = JsonCodec;
312        let config = SessionConfig::from_vertex("proj", "us-central1", "token")
313            .model(ModelId::from_static("models/gemini-2.0-flash-live-001"));
314        let cmd = SessionCommand::SendToolResponse(vec![FunctionResponse {
315            name: "search".to_string(),
316            response: serde_json::json!({"ok": true}),
317            id: Some("call-1".to_string()),
318            scheduling: Some(FunctionResponseScheduling::WhenIdle),
319        }]);
320        let bytes = codec.encode_command(&cmd, &config).unwrap();
321        let json = String::from_utf8(bytes).unwrap();
322        assert!(
323            !json.contains("scheduling"),
324            "Vertex AI should strip scheduling from tool responses"
325        );
326    }
327
328    #[test]
329    fn json_codec_preserves_scheduling_for_google_ai() {
330        let codec = JsonCodec;
331        let config = test_config();
332        let cmd = SessionCommand::SendToolResponse(vec![FunctionResponse {
333            name: "search".to_string(),
334            response: serde_json::json!({"ok": true}),
335            id: Some("call-1".to_string()),
336            scheduling: Some(FunctionResponseScheduling::WhenIdle),
337        }]);
338        let bytes = codec.encode_command(&cmd, &config).unwrap();
339        let json = String::from_utf8(bytes).unwrap();
340        assert!(
341            json.contains("WHEN_IDLE"),
342            "Google AI should preserve scheduling in tool responses"
343        );
344    }
345
346    #[test]
347    fn manual_activity_signals_are_suppressed_under_automatic_detection() {
348        // The default config leaves server VAD on, and the two are mutually
349        // exclusive on the wire: emitting one draws a 1007 close and kills the
350        // session mid-utterance. An empty encoding is the codec's idiom for
351        // "not on this wire"; the session loop skips it.
352        let codec = JsonCodec;
353        let config = test_config();
354        assert!(config.automatic_activity_detection_enabled());
355
356        for cmd in [SessionCommand::ActivityStart, SessionCommand::ActivityEnd] {
357            let bytes = codec.encode_command(&cmd, &config).unwrap();
358            assert!(
359                bytes.is_empty(),
360                "explicit activity control must not reach the wire while the \
361                 server is detecting activity itself; got {:?}",
362                String::from_utf8_lossy(&bytes)
363            );
364        }
365    }
366
367    #[test]
368    fn manual_activity_signals_are_sent_when_automatic_detection_is_off() {
369        let codec = JsonCodec;
370        let mut config = test_config();
371        config.realtime_input_config = Some(crate::protocol::types::RealtimeInputConfig {
372            automatic_activity_detection: Some(
373                crate::protocol::types::AutomaticActivityDetection {
374                    disabled: Some(true),
375                    start_of_speech_sensitivity: None,
376                    end_of_speech_sensitivity: None,
377                    prefix_padding_ms: None,
378                    silence_duration_ms: None,
379                },
380            ),
381            activity_handling: None,
382            turn_coverage: None,
383        });
384        assert!(!config.automatic_activity_detection_enabled());
385
386        let json = String::from_utf8(
387            codec
388                .encode_command(&SessionCommand::ActivityStart, &config)
389                .unwrap(),
390        )
391        .unwrap();
392        assert!(json.contains("\"activityStart\""), "{json}");
393        assert!(!json.contains("\"activityEnd\""), "{json}");
394
395        let json = String::from_utf8(
396            codec
397                .encode_command(&SessionCommand::ActivityEnd, &config)
398                .unwrap(),
399        )
400        .unwrap();
401        assert!(json.contains("\"activityEnd\""), "{json}");
402        assert!(!json.contains("\"activityStart\""), "{json}");
403    }
404
405    #[test]
406    fn json_codec_encode_client_content() {
407        let codec = JsonCodec;
408        let config = test_config();
409        let cmd = SessionCommand::SendClientContent {
410            turns: vec![Content::user("context message")],
411            turn_complete: false,
412        };
413        let bytes = codec.encode_command(&cmd, &config).unwrap();
414        let json = String::from_utf8(bytes).unwrap();
415        assert!(
416            json.contains("\"clientContent\""),
417            "should contain clientContent"
418        );
419        assert!(
420            json.contains("context message"),
421            "should contain the text content"
422        );
423        assert!(
424            json.contains("\"turnComplete\":false"),
425            "should contain turnComplete set to false"
426        );
427    }
428
429    #[test]
430    fn json_codec_encode_send_video() {
431        let codec = JsonCodec;
432        let config = test_config();
433        let cmd = SessionCommand::SendVideo(vec![0xFF, 0xD8, 0xFF].into()); // JPEG magic bytes
434        let bytes = codec.encode_command(&cmd, &config).unwrap();
435        let json: serde_json::Value = serde_json::from_slice(&bytes).unwrap();
436        assert_eq!(
437            json["realtimeInput"]["video"]["mimeType"].as_str().unwrap(),
438            "image/jpeg"
439        );
440        assert!(json["realtimeInput"]["video"]["data"].is_string());
441    }
442
443    #[test]
444    fn json_codec_encode_update_instruction_google_ai() {
445        // A `system` role closes a Google AI session (1007), so the update
446        // travels as a user turn that names itself, without completing the turn.
447        let codec = JsonCodec;
448        let config = test_config();
449        assert!(!config.supports_system_role_updates());
450        let cmd = SessionCommand::UpdateInstruction("New instruction".into());
451        let bytes = codec.encode_command(&cmd, &config).unwrap();
452        let json: serde_json::Value = serde_json::from_slice(&bytes).unwrap();
453        let turns = &json["clientContent"]["turns"];
454        assert_eq!(turns[0]["role"], "user");
455        let text = turns[0]["parts"][0]["text"].as_str().unwrap();
456        assert!(text.starts_with("System instruction update"), "{text}");
457        assert!(text.ends_with("New instruction"), "{text}");
458        assert_eq!(json["clientContent"]["turnComplete"], false);
459    }
460
461    #[test]
462    fn json_codec_encode_update_instruction_vertex() {
463        let codec = JsonCodec;
464        let config = SessionConfig::from_vertex("p", "us-central1", "t");
465        let cmd = SessionCommand::UpdateInstruction("New instruction".into());
466        let bytes = codec.encode_command(&cmd, &config).unwrap();
467        let json: serde_json::Value = serde_json::from_slice(&bytes).unwrap();
468        let turns = &json["clientContent"]["turns"];
469        assert_eq!(turns[0]["role"], "system");
470        assert_eq!(turns[0]["parts"][0]["text"], "New instruction");
471        assert_eq!(json["clientContent"]["turnComplete"], false);
472    }
473
474    #[test]
475    fn json_codec_encode_disconnect() {
476        let codec = JsonCodec;
477        let config = test_config();
478        let cmd = SessionCommand::Disconnect;
479        let bytes = codec.encode_command(&cmd, &config).unwrap();
480        assert!(bytes.is_empty(), "Disconnect should produce empty bytes");
481    }
482
483    // -----------------------------------------------------------------------
484    // Decode tests
485    // -----------------------------------------------------------------------
486
487    #[test]
488    fn json_codec_decode_setup_complete() {
489        let codec = JsonCodec;
490        let json = r#"{"setupComplete":{"sessionResumption":{"handle":"abc123"}}}"#;
491        let msg = codec.decode_message(json.as_bytes()).unwrap();
492        match msg {
493            ServerMessage::SetupComplete(sc) => {
494                let handle = sc.setup_complete.session_resumption.unwrap().handle;
495                assert_eq!(handle, Some("abc123".to_string()));
496            }
497            _ => panic!("Expected SetupComplete"),
498        }
499    }
500
501    #[test]
502    fn json_codec_decode_server_content() {
503        let codec = JsonCodec;
504        let json = r#"{
505            "serverContent": {
506                "modelTurn": {
507                    "parts": [{"text": "Hello! How can I help?"}]
508                },
509                "turnComplete": true
510            }
511        }"#;
512        let msg = codec.decode_message(json.as_bytes()).unwrap();
513        match msg {
514            ServerMessage::ServerContent(sc) => {
515                assert!(sc.server_content.turn_complete.unwrap_or(false));
516                let turn = sc.server_content.model_turn.unwrap();
517                assert_eq!(turn.parts.len(), 1);
518                match &turn.parts[0] {
519                    Part::Text { text } => assert_eq!(text, "Hello! How can I help?"),
520                    _ => panic!("Expected text part"),
521                }
522            }
523            _ => panic!("Expected ServerContent"),
524        }
525    }
526
527    #[test]
528    fn json_codec_decode_tool_call() {
529        let codec = JsonCodec;
530        let json = r#"{
531            "toolCall": {
532                "functionCalls": [
533                    {"name": "get_weather", "args": {"city": "London"}, "id": "call-1"}
534                ]
535            }
536        }"#;
537        let msg = codec.decode_message(json.as_bytes()).unwrap();
538        match msg {
539            ServerMessage::ToolCall(tc) => {
540                assert_eq!(tc.tool_call.function_calls.len(), 1);
541                assert_eq!(tc.tool_call.function_calls[0].name, "get_weather");
542            }
543            _ => panic!("Expected ToolCall"),
544        }
545    }
546
547    #[test]
548    fn json_codec_decode_invalid_utf8() {
549        let codec = JsonCodec;
550        let bad_bytes: &[u8] = &[0xFF, 0xFE, 0xFD];
551        let result = codec.decode_message(bad_bytes);
552        match result {
553            Err(CodecError::InvalidUtf8) => {} // expected
554            other => panic!("Expected CodecError::InvalidUtf8, got {other:?}"),
555        }
556    }
557}
gemini_genai_rs/transport/codec.rs

gemini_genai_rs/transport/
codec.rs
