fix(agent): allow two-minute live replies without ending calls

Signed-off-by: Meli <5aaa86bce934fc3445fc254aab560a40923f10252f92107e665073dede0e04d3@buzz.block.builderlab.xyz>

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.

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| {

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" {

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};

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;
+    }
+}