diff --git a/crates/buzz-agent/src/config.rs b/crates/buzz-agent/src/config.rs index 35e0caa625..70146fb70c 100644 --- a/crates/buzz-agent/src/config.rs +++ b/crates/buzz-agent/src/config.rs @@ -969,7 +969,7 @@ fn parse_openai_api(raw: Option<&str>) -> Result { "realtime" => Ok(OpenAiApi::Realtime), "auto" | "" => Ok(OpenAiApi::Auto), other => Err(format!( - "config: OPENAI_COMPAT_API={other} not supported (use auto|chat|responses)" + "config: OPENAI_COMPAT_API={other} not supported (use auto|chat|responses|realtime)" )), } } diff --git a/crates/buzz-agent/src/realtime.rs b/crates/buzz-agent/src/realtime.rs index 662c684c29..a78bdd009c 100644 --- a/crates/buzz-agent/src/realtime.rs +++ b/crates/buzz-agent/src/realtime.rs @@ -215,7 +215,8 @@ impl RealtimeSender { impl RealtimeReceiver { /// Read one bounded JSON event. Unknown event types remain available to the - /// session driver; errors propagate, and binary audio is not this protocol. + /// session driver, including recoverable provider errors. Transport failures + /// propagate; binary audio is not this protocol. /// The owner sets idle/session deadlines and must not log raw events. pub async fn next_event(&mut self) -> Result { if self.failed { @@ -237,9 +238,6 @@ impl RealtimeReceiver { if event["type"].as_str().is_none() { return Err(error("event type missing")); } - if event["type"] == "error" { - return Err(error("provider returned an error event")); - } self.failed = false; return Ok(event); } diff --git a/crates/buzz-agent/src/realtime_session.rs b/crates/buzz-agent/src/realtime_session.rs index f9e52c29e9..9aa35a56f7 100644 --- a/crates/buzz-agent/src/realtime_session.rs +++ b/crates/buzz-agent/src/realtime_session.rs @@ -176,7 +176,10 @@ impl RealtimeSession { "type":"realtime", "model":ctx.effective_model, "instructions":ctx.system_prompt, "tools":tools, "output_modalities":[if audio_output { "audio" } else { "text" }], - "audio":{"input":{"turn_detection":null}}, + "audio":{ + "input":{"turn_detection":null,"format":{"type":"audio/pcm","rate":24000}}, + "output":{"format":{"type":"audio/pcm","rate":24000}}, + }, })) .await?; let deadline = Instant::now() + ctx.cfg.llm_timeout; @@ -184,9 +187,11 @@ impl RealtimeSession { if updated["type"] != "session.updated" || updated["session"]["output_modalities"] != json!([if audio_output { "audio" } else { "text" }]) - || updated["session"]["audio"]["input"].get("turn_detection") != Some(&Value::Null) + || updated["session"]["audio"]["input"] + .get("turn_detection") + .is_some_and(|value| !value.is_null()) { - return Err(error("provider did not acknowledge manual text session")); + return Err(error("provider did not acknowledge manual session")); } } let live = self.live.as_mut().ok_or_else(|| error("missing session"))?; @@ -438,8 +443,16 @@ impl Live { biased; _ = ctx.cancel.changed() => Err(AgentError::Cancelled), result = tokio::time::timeout_at(deadline.min(self.expires), self.events.recv()) => { - result.map_err(|_| error("response deadline exceeded"))? - .ok_or_else(|| error("session disconnected"))? + let event = result.map_err(|_| error("response deadline exceeded"))? + .ok_or_else(|| error("session disconnected"))??; + if event["type"] == "error" { + // This manual driver cannot reconcile rejected mutations yet. + // Keep that policy here, not in the reusable transport. + let message = event["error"]["message"].as_str().unwrap_or("unspecified provider error"); + let bounded: String = message.chars().filter(|c| !c.is_control()).take(512).collect(); + return Err(error(&format!("provider rejected request: {bounded}"))); + } + Ok(event) } } } diff --git a/crates/buzz-agent/tests/realtime.rs b/crates/buzz-agent/tests/realtime.rs index 6d9a5d9c7e..ca1e6ef71c 100644 --- a/crates/buzz-agent/tests/realtime.rs +++ b/crates/buzz-agent/tests/realtime.rs @@ -148,11 +148,6 @@ async fn invalid_frames_poison_connection_without_echoing_provider_data() { for message in [ Message::Text("not-json secret".into()), Message::Text(json!({"not_type":"secret"}).to_string().into()), - Message::Text( - json!({"type":"error","error":{"message":"secret"}}) - .to_string() - .into(), - ), Message::Binary(vec![1, 2].into()), Message::Text("x".repeat(1024 * 1024 + 1).into()), ] { @@ -278,3 +273,34 @@ async fn pending_provider_read_does_not_block_cancellation_write() { assert_eq!(terminal["response"]["status"], "cancelled"); server.await.unwrap(); } + +#[tokio::test] +async fn provider_error_preserves_event_and_connection() { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let endpoint = format!("ws://{}", listener.local_addr().unwrap()); + let event = json!({"type":"error","event_id":"server_error", "error":{"type":"invalid_request_error","code":"invalid_value","message":"Unknown voice","event_id":"client_update"}}); + let expected = event.clone(); + let server = tokio::spawn(async move { + let (tcp, _) = listener.accept().await.unwrap(); + let mut ws = accept_async(tcp).await.unwrap(); + for event in [ + json!({"type":"session.created"}), + event, + json!({"type":"session.updated"}), + ] { + ws.send(Message::Text(event.to_string().into())) + .await + .unwrap(); + } + }); + let connection = RealtimeConnection::connect(&endpoint, "test") + .await + .unwrap(); + let (_sender, mut reader) = connection.split(); + assert_eq!(reader.next_event().await.unwrap(), expected); + assert_eq!( + reader.next_event().await.unwrap()["type"], + "session.updated" + ); + server.await.unwrap(); +} diff --git a/crates/buzz-agent/tests/realtime_acp.rs b/crates/buzz-agent/tests/realtime_acp.rs index 6308dbba80..2b3da2b08c 100644 --- a/crates/buzz-agent/tests/realtime_acp.rs +++ b/crates/buzz-agent/tests/realtime_acp.rs @@ -23,7 +23,7 @@ async fn recv(ws: &mut Ws) -> Value { .unwrap(); serde_json::from_str(msg.to_text().unwrap()).unwrap() } -async fn modality(ws: &mut Ws) { +async fn modality_ack(ws: &mut Ws) { let update = recv(ws).await; assert_eq!(update["type"], "session.update"); send( @@ -70,7 +70,7 @@ async fn connect(listener: TcpListener) -> Ws { json!({"type":"session.updated", "session":update["session"]}), ) .await; - modality(&mut ws).await; + modality_ack(&mut ws).await; assert_eq!(recv(&mut ws).await["type"], "conversation.item.create"); assert_eq!(recv(&mut ws).await["type"], "response.create"); ws @@ -132,7 +132,7 @@ async fn realtime_acp_permission_and_persistent_conversation() { assert_eq!(recv(&mut ws).await["type"], "response.create"); text(&mut ws, "r2").await; // A second ACP prompt must use the SAME socket, not a new session. - modality(&mut ws).await; + modality_ack(&mut ws).await; assert_eq!(recv(&mut ws).await["type"], "conversation.item.create"); assert_eq!(recv(&mut ws).await["type"], "response.create"); text(&mut ws, "r3").await; @@ -256,12 +256,18 @@ async fn realtime_acp_audio_content_roundtrip() { .await; let update = recv(&mut ws).await; assert_eq!(update["session"]["output_modalities"], json!(["audio"])); + for direction in ["input", "output"] { + assert_eq!( + update["session"]["audio"][direction]["format"], + json!({"type":"audio/pcm","rate":24000}) + ); + } send( &mut ws, json!({"type":"session.updated","session":update["session"]}), ) .await; - modality(&mut ws).await; + modality_ack(&mut ws).await; let append = recv(&mut ws).await; assert_eq!(append["type"], "input_audio_buffer.append"); assert_eq!( @@ -339,3 +345,82 @@ async fn realtime_disconnect_revokes_pending_permission_without_executing() { assert!(!dir.path().join("calls.log").exists()); server.await.unwrap(); } + +#[tokio::test] +async fn realtime_session_acknowledgment_is_not_an_echo_contract() { + for (modality, detection, accepted) in [ + ("text", None, true), + ("text", Some(Value::Null), true), + ("audio", None, false), + ("text", Some(json!({"type":"server_vad"})), false), + ] { + let dir = tempfile::tempdir().unwrap(); + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let url = format!("ws://{}/v1/realtime", listener.local_addr().unwrap()); + let server = tokio::spawn(async move { + let (socket, _) = listener.accept().await.unwrap(); + let mut ws = accept_async(socket).await.unwrap(); + send( + &mut ws, + json!({"type":"session.created","session":{"id":"s1"}}), + ) + .await; + assert_eq!(recv(&mut ws).await["type"], "session.update"); + let mut session = + json!({"type":"realtime","output_modalities":[modality],"audio":{"input":{}}}); + if let Some(detection) = detection { + session["audio"]["input"]["turn_detection"] = detection; + } + send(&mut ws, json!({"type":"session.updated","session":session})).await; + if accepted { + modality_ack(&mut ws).await; + assert_eq!(recv(&mut ws).await["type"], "conversation.item.create"); + assert_eq!(recv(&mut ws).await["type"], "response.create"); + text(&mut ws, "r1").await; + } + tokio::time::sleep(Duration::from_millis(200)).await; + }); + let mut h = Harness::spawn_with_env(&url, &options()).await; + let sid = init(&mut h, dir.path()).await; + let p = prompt(&mut h, &sid).await; + let result = h.recv_until(|v| v["id"] == p).await; + if accepted { + assert_eq!(result["result"]["stopReason"], "end_turn", "{result}"); + } else { + assert!( + result["error"]["message"] + .as_str() + .unwrap() + .contains("acknowledge manual session"), + "{result}" + ); + } + assert!(!dir.path().join("calls.log").exists()); + server.await.unwrap(); + } +} + +#[tokio::test] +async fn realtime_rejection_reports_provider_reason_without_tool_effects() { + let dir = tempfile::tempdir().unwrap(); + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let url = format!("ws://{}/v1/realtime", listener.local_addr().unwrap()); + let server = tokio::spawn(async move { + let mut ws = connect(listener).await; + send(&mut ws, json!({"type":"error", "error":{"code":"invalid_value","message":"Unsupported output voice"}})).await; + tokio::time::sleep(Duration::from_millis(200)).await; + }); + let mut h = Harness::spawn_with_env(&url, &options()).await; + let sid = init(&mut h, dir.path()).await; + let p = prompt(&mut h, &sid).await; + let result = h.recv_until(|v| v["id"] == p).await; + assert!( + result["error"]["message"] + .as_str() + .unwrap() + .contains("Unsupported output voice"), + "{result}" + ); + assert!(!dir.path().join("calls.log").exists()); + server.await.unwrap(); +}