diff --git a/crates/buzz-agent/README.md b/crates/buzz-agent/README.md index 0a4d1010c4..b60db1de7f 100644 --- a/crates/buzz-agent/README.md +++ b/crates/buzz-agent/README.md @@ -431,6 +431,12 @@ Existing clients keep manual turns unless they opt in: for client-initiated interruption. Positions cannot exceed emitted samples. 6. Send `.../close` to end the live prompt, or use ordinary `session/cancel`. +Each live spoken reply allows up to two minutes of audio (or the provider's +smaller advertised limit). Reaching this cap cancels that reply, clears queued +playback, and truncates its history to the reported playback position. A +`response_limited` update reports `max_output_audio`; the microphone session +stays open and the next reply gets a fresh budget. + When `ready.input` is true, `.../input` accepts the same session/stream IDs and optional `context` and `text` strings (each at most 16 KiB). Context replaces a pending snapshot without generating a response or changing session instructions. diff --git a/crates/buzz-agent/src/realtime_session.rs b/crates/buzz-agent/src/realtime_session.rs index 4837b83b49..524d0a74d4 100644 --- a/crates/buzz-agent/src/realtime_session.rs +++ b/crates/buzz-agent/src/realtime_session.rs @@ -22,6 +22,8 @@ use crate::{ }; const SESSION_LIMIT: Duration = Duration::from_secs(60 * 60); +// Streamed replies do not accumulate the short WAV input/output buffer. +const MAX_LIVE_AUDIO_SAMPLES: u64 = 24_000 * 120; const MAX_CALLS: usize = 64; const MAX_SESSION_IDS: usize = 4096; // Aggregate content budget: UTF-8 text, base64 images, and decoded PCM bytes. @@ -206,7 +208,7 @@ impl RealtimeSession { duplex_audio: false, backchannels: false, input_context: false, - output_audio_limit: MAX_PCM as u64 / 2, + output_audio_limit: MAX_LIVE_AUDIO_SAMPLES, }); let live = self.live.as_mut().ok_or_else(|| error("missing session"))?; let mut tools = ctx.mcp.tools(); @@ -245,7 +247,7 @@ impl RealtimeSession { .as_u64() .filter(|n| (1..=24000 * 3600).contains(n)) .ok_or_else(|| error("invalid provider live audio budget"))?; - live.output_audio_limit = limit; + live.output_audio_limit = limit.min(MAX_LIVE_AUDIO_SAMPLES); } if updated["type"] != "session.updated" || ctx.cfg.thinking_effort.is_some_and(|effort| { diff --git a/crates/buzz-agent/src/realtime_session/media.rs b/crates/buzz-agent/src/realtime_session/media.rs index 456dfcb912..a8eca9d3a5 100644 --- a/crates/buzz-agent/src/realtime_session/media.rs +++ b/crates/buzz-agent/src/realtime_session/media.rs @@ -533,11 +533,17 @@ impl State { let pcm = STANDARD .decode(delta) .map_err(|_| error("invalid audio base64"))?; - if pcm.is_empty() - || !pcm.len().is_multiple_of(2) - || output.emitted + pcm.len() as u64 / 2 > live.output_audio_limit - { - return Err(error("media output limit")); + if pcm.is_empty() || !pcm.len().is_multiple_of(2) { + return Err(error("invalid media PCM: expected nonempty 16-bit samples")); + } + if output.emitted + pcm.len() as u64 / 2 > live.output_audio_limit { + // Use the existing cancellation/playback/truncation fence so the + // next turn retains only speech the user actually heard. + self.interrupt(live, ctx.wire, ctx.session_id, &ctx.run_id) + .await?; + media::notify(ctx.wire, ctx.session_id, &ctx.run_id, + json!({"type":"response_limited","responseId":response,"reason":"max_output_audio"})).await?; + return Ok(None); } media::notify( ctx.wire, @@ -583,7 +589,7 @@ impl State { if limited { media::notify(ctx.wire, ctx.session_id, &ctx.run_id, json!({"type":"response_limited","responseId":response["id"],"reason":response["status_details"]["reason"]})).await?; } - for item in items.iter().filter(|_| !limited) { + for item in items.iter().filter(|_| !limited && !self.interrupted) { live.remember(field(item, "id")?)?; if item["type"] == "function_call" { if item["status"] != "completed" { diff --git a/crates/buzz-agent/tests/realtime_acp.rs b/crates/buzz-agent/tests/realtime_acp.rs index d3f31279d4..d3552c1d4e 100644 --- a/crates/buzz-agent/tests/realtime_acp.rs +++ b/crates/buzz-agent/tests/realtime_acp.rs @@ -1,5 +1,7 @@ //! Real ACP process -> standard Realtime WS -> existing permission/MCP seam. mod common; +#[path = "realtime_acp/live_limits.rs"] +mod live_limits; use base64::{engine::general_purpose::STANDARD, Engine}; use common::{approve_permission, Harness}; diff --git a/crates/buzz-agent/tests/realtime_acp/live_limits.rs b/crates/buzz-agent/tests/realtime_acp/live_limits.rs new file mode 100644 index 0000000000..277e146b42 --- /dev/null +++ b/crates/buzz-agent/tests/realtime_acp/live_limits.rs @@ -0,0 +1,142 @@ +use super::*; + +#[tokio::test] +async fn live_audio_budget_is_two_minutes_per_reply_and_recovers_after_cutoff() { + for advertised in [None, Some(48_000), Some(24_000 * 300)] { + let samples = advertised.unwrap_or(24_000 * 120).min(24_000 * 120); + 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":"limits"}}), + ) + .await; + let mut update = recv(&mut ws).await; + if let Some(limit) = advertised { + update["session"]["frankie"] = json!({"max_output_audio_samples":limit}); + } + send( + &mut ws, + json!({"type":"session.updated","session":update["session"]}), + ) + .await; + // Full budget twice proves ordinary replies reset it; a third reply + // proves a cutoff does not poison the persistent conversation. + for turn in 0..3 { + assert_eq!(recv(&mut ws).await["type"], "input_audio_buffer.append"); + send( + &mut ws, + json!({"type":"input_audio_buffer.committed","item_id":format!("input{turn}")}), + ) + .await; + assert_eq!(recv(&mut ws).await["type"], "response.create"); + let id = format!("r{turn}"); + let item = format!("a{turn}"); + created(&mut ws, &id).await; + // Below the 1 MiB wire frame cap; duration is a sample count, + // so testing two minutes does not require sleeping two minutes. + for _ in 0..8 { + send(&mut ws, json!({"type":"response.output_audio.delta","response_id":id,"item_id":item,"content_index":0,"delta":STANDARD.encode(vec![0u8;samples / 8 * 2])})).await; + } + if turn == 1 { + send(&mut ws, json!({"type":"response.output_audio.delta","response_id":id,"item_id":item,"content_index":0,"delta":"AAA="})).await; + let cancel = recv(&mut ws).await; + assert_eq!(cancel["type"], "response.cancel"); + assert_eq!(cancel["response_id"], id); + // Completion may already be in flight when cancellation arrives. + let status = if advertised.is_some() { + "completed" + } else { + "cancelled" + }; + send(&mut ws, json!({"type":"response.done","response":{"id":id,"status":status,"output":[call("discarded")]}})).await; + let truncate = recv(&mut ws).await; + assert_eq!(truncate["type"], "conversation.item.truncate"); + assert_eq!(truncate["item_id"], item); + assert_eq!(truncate["audio_end_ms"], samples / 24); + send(&mut ws, json!({"type":"conversation.item.truncated","item_id":item,"content_index":0,"audio_end_ms":samples / 24})).await; + } else { + done(&mut ws, &id, vec![]).await; + } + } + let _ = tokio::time::timeout(Duration::from_secs(10), ws.next()).await; + }); + let mut h = Harness::spawn_with_env(&url, &options()).await; + let i = h.send("initialize", json!({"protocolVersion":1,"clientCapabilities":{"_meta":{"buzz":{"realtimeAudio":1}}}})).await; + h.recv_until(|v| v["id"] == i).await; + let i = h + .send("session/new", json!({"cwd":dir.path(),"mcpServers":[]})) + .await; + let sid = h.recv_until(|v| v["id"] == i).await["result"]["sessionId"] + .as_str() + .unwrap() + .to_owned(); + let prompt = h + .send( + "session/prompt", + json!({"sessionId":sid,"prompt":[],"_meta":{"buzz":{"realtimeAudio":1}}}), + ) + .await; + let ready = h + .recv_until(|v| v["params"]["update"]["type"] == "ready" || v["id"] == prompt) + .await; + let stream = ready["params"]["streamId"].as_str().unwrap(); + for turn in 0..3 { + h.send("_buzz/unstable/realtime/append", json!({"sessionId":sid,"streamId":stream,"sequence":turn,"data":STANDARD.encode(vec![0u8;960])})).await; + let mut emitted = 0; + let mut limited = false; + let mut cleared = false; + loop { + let event = h.recv().await; + assert_ne!(event["id"], prompt, "live session ended: {event}"); + assert_ne!(event["method"], "session/request_permission"); + let update = &event["params"]["update"]; + assert_ne!(update["sessionUpdate"], "tool_call"); + match update["type"].as_str() { + Some("audio") => { + assert_eq!(update["startSample"], emitted); + emitted += STANDARD + .decode(update["data"].as_str().unwrap()) + .unwrap() + .len() + / 2; + assert!(emitted <= samples); + } + Some("clear") => { + assert_eq!(turn, 1); + cleared = true; + h.send("_buzz/unstable/realtime/playback", json!({"sessionId":sid,"streamId":stream,"playback":{"responseId":format!("r{turn}"),"itemId":format!("a{turn}"),"contentIndex":0,"playedSamples":emitted,"stopped":true}})).await; + } + Some("response_limited") => { + assert_eq!(update["reason"], "max_output_audio"); + limited = true; + } + Some("response_done") => break, + _ => {} + } + } + assert_eq!(emitted, samples); + assert_eq!(limited, turn == 1); + assert_eq!(cleared, turn == 1); + if turn != 1 { + let i = h.send("_buzz/unstable/realtime/playback", json!({"sessionId":sid,"streamId":stream,"playback":{"responseId":format!("r{turn}"),"itemId":format!("a{turn}"),"contentIndex":0,"playedSamples":emitted}})).await; + assert!(h.recv_until(|v| v["id"] == i).await.get("error").is_none()); + } + } + h.send( + "_buzz/unstable/realtime/close", + json!({"sessionId":sid,"streamId":stream}), + ) + .await; + assert_eq!( + h.recv_until(|v| v["id"] == prompt).await["result"]["stopReason"], + "end_turn" + ); + server.await.unwrap(); + h.shutdown().await; + } +}