From cfa49cbafdf3bde1e5be4fdfdea729ebcbeb62a3 Mon Sep 17 00:00:00 2001 From: Meli <5aaa86bce934fc3445fc254aab560a40923f10252f92107e665073dede0e04d3@buzz.block.builderlab.xyz> Date: Tue, 8 Sep 2026 22:18:22 -0400 Subject: [PATCH] fix(agent): negotiate PCM and preserve Realtime error events Explicitly request PCM24k on both audio directions. Accept absent or null manual turn detection and reject mismatched acknowledgment modalities. Return provider errors as transport events; the manual driver still fails closed on rejected mutations, now with a bounded provider diagnostic. Co-authored-by: Meli <5aaa86bce934fc3445fc254aab560a40923f10252f92107e665073dede0e04d3@buzz.block.builderlab.xyz> Signed-off-by: Meli <5aaa86bce934fc3445fc254aab560a40923f10252f92107e665073dede0e04d3@buzz.block.builderlab.xyz> --- crates/buzz-agent/src/config.rs | 2 +- crates/buzz-agent/src/realtime.rs | 6 +- crates/buzz-agent/src/realtime_session.rs | 23 ++++-- crates/buzz-agent/tests/realtime.rs | 36 +++++++-- crates/buzz-agent/tests/realtime_acp.rs | 93 ++++++++++++++++++++++- 5 files changed, 141 insertions(+), 19 deletions(-) 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(); +}