From 049fed56bbd1c669f55a737faaafd56b4cfaa631 Mon Sep 17 00:00:00 2001 From: Codex Date: Sat, 12 Sep 2026 23:55:22 -0400 Subject: [PATCH] agent: accept view context and text during realtime sessions Signed-off-by: Codex --- crates/buzz-agent/README.md | 9 ++ crates/buzz-agent/src/realtime.rs | 9 ++ crates/buzz-agent/src/realtime_media.rs | 30 ++++ crates/buzz-agent/src/realtime_session.rs | 3 + .../buzz-agent/src/realtime_session/media.rs | 63 +++++++- crates/buzz-agent/tests/realtime_acp.rs | 151 ++++++++++++++++++ 6 files changed, 262 insertions(+), 3 deletions(-) diff --git a/crates/buzz-agent/README.md b/crates/buzz-agent/README.md index 30782cdbad..0a4d1010c4 100644 --- a/crates/buzz-agent/README.md +++ b/crates/buzz-agent/README.md @@ -431,6 +431,15 @@ 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`. +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. +It is latched for the next utterance. A provider advertising +`frankie.input_context` attaches it before audio prefill; standard providers +receive a user context item beside the committed audio before response creation. +Typed text is accepted between turns and shares the pending context. Text sent +during speech, output, or tools is rejected without ending the call. + Provider VAD commits user turns. Buzz coordinates response creation, cancellation, playback truncation acknowledgment, and the existing tool executor and permission broker. Capture and control queues are separate and bounded; stale stream IDs, diff --git a/crates/buzz-agent/src/realtime.rs b/crates/buzz-agent/src/realtime.rs index bf44910bea..d8ac5c5bec 100644 --- a/crates/buzz-agent/src/realtime.rs +++ b/crates/buzz-agent/src/realtime.rs @@ -124,6 +124,15 @@ impl RealtimeConnection { } impl RealtimeSender { + /// Update context for the next input only when the provider advertises this capability. + pub(crate) async fn input_context( + &mut self, + revision: u64, + text: &str, + ) -> Result<(), AgentError> { + self.send(json!({"type":"frankie.input_context.update", "revision":revision, "text":text})) + .await + } /// Send a GA session update. Reject unsupported settings at the server, /// rather than translating them into model-specific wire fields. pub async fn update_session(&mut self, session: Value) -> Result<(), AgentError> { diff --git a/crates/buzz-agent/src/realtime_media.rs b/crates/buzz-agent/src/realtime_media.rs index 50a21a6ff2..e4023a2fe7 100644 --- a/crates/buzz-agent/src/realtime_media.rs +++ b/crates/buzz-agent/src/realtime_media.rs @@ -25,6 +25,10 @@ pub(crate) enum Action { }, Playback(Playback), Interrupt(Option), + Input { + context: Option, + text: Option, + }, Close, } @@ -52,6 +56,10 @@ struct Params { playback_data: Option, #[serde(default)] playback: Option, + #[serde(default)] + context: Option, + #[serde(default)] + text: Option, } pub(crate) struct Ingress { @@ -89,6 +97,28 @@ pub(crate) fn enabled(meta: &Value) -> bool { } fn action(operation: &str, params: &mut Params) -> Result { + if operation == "input" { + if params.sequence.is_some() + || params.data.is_some() + || params.playback.is_some() + || params.playback_data.is_some() + || (params.context.is_none() && params.text.is_none()) + || params.context.as_ref().is_some_and(|s| s.len() > 16384) + || params + .text + .as_ref() + .is_some_and(|s| s.trim().is_empty() || s.len() > 16384) + { + return Err("invalid realtime input"); + } + return Ok(Action::Input { + context: params.context.take(), + text: params.text.take(), + }); + } + if params.context.is_some() || params.text.is_some() { + return Err("unexpected input fields"); + } if operation != "append" && params.playback_data.is_some() { return Err("playback audio is only valid with capture"); } diff --git a/crates/buzz-agent/src/realtime_session.rs b/crates/buzz-agent/src/realtime_session.rs index 5ceca6ffc6..4837b83b49 100644 --- a/crates/buzz-agent/src/realtime_session.rs +++ b/crates/buzz-agent/src/realtime_session.rs @@ -51,6 +51,7 @@ struct Live { active_response: Option, duplex_audio: bool, backchannels: bool, + input_context: bool, output_audio_limit: u64, } @@ -204,6 +205,7 @@ impl RealtimeSession { active_response: None, duplex_audio: false, backchannels: false, + input_context: false, output_audio_limit: MAX_PCM as u64 / 2, }); let live = self.live.as_mut().ok_or_else(|| error("missing session"))?; @@ -237,6 +239,7 @@ impl RealtimeSession { let updated = live.next(ctx, deadline).await?; live.duplex_audio = updated["session"]["frankie"]["duplex_audio"] == true; live.backchannels = updated["session"]["frankie"]["backchannels"] == true; + live.input_context = updated["session"]["frankie"]["input_context"] == true; if let Some(limit) = updated["session"]["frankie"].get("max_output_audio_samples") { let limit = limit .as_u64() diff --git a/crates/buzz-agent/src/realtime_session/media.rs b/crates/buzz-agent/src/realtime_session/media.rs index 22c02b82ea..456dfcb912 100644 --- a/crates/buzz-agent/src/realtime_session/media.rs +++ b/crates/buzz-agent/src/realtime_session/media.rs @@ -23,6 +23,13 @@ struct State { response_requested: bool, fence_deadline: Option, response_deadline: Option, + context: (u64, String), + speech_context: (u64, String), + context_used: u64, + input_active: bool, + tools_active: bool, + text_sequence: u64, + committed_context: Option<(u64, String)>, } impl State { @@ -36,6 +43,13 @@ impl State { awaiting_playback: false, awaiting_truncate: false, pending_input: false, + context: (0, String::new()), + speech_context: (0, String::new()), + context_used: 0, + input_active: false, + tools_active: false, + text_sequence: 0, + committed_context: None, response_requested: false, fence_deadline: None, response_deadline: None, @@ -144,6 +158,14 @@ impl State { self.completed_playback = Some(output); } self.pending_input = false; + // Standard providers receive context beside the committed audio, before output. + // Frankie attaches it inside the input item before speculative prefill instead. + if let Some((revision, text)) = self.committed_context.take() { + if !live.input_context && revision > self.context_used { + live.sender.create_item(json!({"type":"message","role":"user","content":[{"type":"input_text","text":text}]})).await?; + self.context_used = revision; + } + } self.response_requested = true; self.response_deadline = Some(Instant::now() + ctx.cfg.llm_timeout); live.sender.create_response().await?; @@ -187,7 +209,7 @@ impl RealtimeSession { live.sender.create_item(json!({"type":"message","role":"user","content":parts})).await?; } media::notify(ctx.wire, ctx.session_id, &ctx.run_id, json!({"type":"ready","format":"s16le","rate":24000,"channels":1, - "maxFrameBytes":media::FRAME_BYTES,"queueFrames":media::QUEUE_FRAMES})).await?; + "maxFrameBytes":media::FRAME_BYTES,"queueFrames":media::QUEUE_FRAMES,"input":true})).await?; let mut state = State::new(); let mut heartbeat = tokio::time::interval(Duration::from_secs(15)); heartbeat.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); @@ -291,6 +313,31 @@ impl State { Ok(()) } Action::Close => Ok(()), + Action::Input { context, text } => { + if text.is_some() && (self.input_active || self.tools_active || self.pending_input + || self.response_requested || live.active_response.is_some() + || self.awaiting_playback || self.awaiting_truncate + || self.output.as_ref().is_some_and(|o| o.played < o.emitted)) { + return Err(error("Wait for the current turn to finish before sending text.")); + } + if let Some(context) = context { + let revision = self.context.0.checked_add(1).ok_or_else(|| error("input context sequence exhausted"))?; + if live.input_context { live.sender.input_context(revision, &context).await?; } + self.context = (revision, context); + } + if let Some(text) = text { + self.text_sequence = self.text_sequence.checked_add(1).ok_or_else(|| error("text input sequence exhausted"))?; + let id = format!("text_{}", self.text_sequence); + let content = if !live.input_context && self.context.0 > self.context_used { + format!("{}\n\n{}", self.context.1, text) + } else { text.clone() }; + live.sender.create_item(json!({"id":id,"type":"message","role":"user","content":[{"type":"input_text","text":content}]})).await?; + self.context_used = self.context.0; + media::notify(output, sid, stream, json!({"type":"input_transcript","itemId":id,"text":text})).await?; + self.pending_input = true; + } + Ok(()) + } } } .await; @@ -319,6 +366,8 @@ impl State { )) } Some("input_audio_buffer.speech_started") => { + self.input_active = true; + self.speech_context = self.context.clone(); // Also interrupt completed generation whose audio has not finished playing. if live.active_response.is_some() || self.response_requested @@ -369,6 +418,7 @@ impl State { .await?; } Some("input_audio_buffer.cleared") => { + self.input_active = false; media::notify( ctx.wire, ctx.session_id, @@ -378,6 +428,8 @@ impl State { .await?; } Some("input_audio_buffer.committed") => { + self.input_active = false; + self.committed_context = Some(self.speech_context.clone()); self.pending_input = true; } Some("conversation.item.truncated") => { @@ -597,6 +649,7 @@ impl State { let (cancel, local) = watch::channel(self.interrupted); let saved = std::mem::replace(ctx.cancel, local); let start = ctx.history.len(); + self.tools_active = true; let result = { let output = ctx.wire.clone(); let sid = ctx.session_id.to_owned(); @@ -636,6 +689,8 @@ impl State { match event { Some(Ok(event)) => match event["type"].as_str() { Some("input_audio_buffer.speech_started") => { + self.input_active = true; + self.speech_context = self.context.clone(); let _ = cancel.send(true); if let Err(e) = self.interrupt(live, &output, &sid, &stream).await { execute.await; break Err(e); @@ -654,9 +709,10 @@ impl State { let _ = cancel.send(true); execute.await; break Err(e); } } - Some("input_audio_buffer.committed") => self.pending_input = true, + Some("input_audio_buffer.committed") => { self.input_active = false; self.pending_input = true; self.committed_context = Some(self.speech_context.clone()); }, + Some("input_audio_buffer.cleared") => self.input_active = false, Some("error") => { let _ = cancel.send(true); execute.await; break Err(provider_error("provider rejected media operation", &event["error"])); } - Some("rate_limits.updated" | "conversation.item.created" | "conversation.item.added" | "conversation.item.done") => {}, + Some("rate_limits.updated" | "conversation.item.created" | "conversation.item.added" | "conversation.item.done" | "frankie.input_context.updated") => {}, _ => { let _ = cancel.send(true); execute.await; break Err(error("unexpected provider event during tools")); } }, _ => { let _ = cancel.send(true); execute.await; break Err(error("provider disconnected during tools")); } @@ -667,6 +723,7 @@ impl State { } }; *ctx.cancel = saved; + self.tools_active = false; result?; let results: Vec<_> = ctx.history.drain(start..).collect(); for result in results { diff --git a/crates/buzz-agent/tests/realtime_acp.rs b/crates/buzz-agent/tests/realtime_acp.rs index 3e3a440bba..d3f31279d4 100644 --- a/crates/buzz-agent/tests/realtime_acp.rs +++ b/crates/buzz-agent/tests/realtime_acp.rs @@ -593,6 +593,157 @@ async fn realtime_invalid_image_fails_before_connect_and_keeps_session_usable() h.shutdown().await; } +#[tokio::test] +async fn realtime_live_context_follows_input_without_resetting_history() { + for native in [false, true] { + 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":"persistent"}}), + ) + .await; + let mut session = recv(&mut ws).await["session"].clone(); + session["frankie"] = json!({"input_context":native}); + send(&mut ws, json!({"type":"session.updated","session":session})).await; + if native { + let context = recv(&mut ws).await; + assert_eq!( + context, + json!({"type":"frankie.input_context.update","revision":1,"text":"View A"}) + ); + send( + &mut ws, + json!({"type":"frankie.input_context.updated","revision":1}), + ) + .await; + } + assert_eq!(recv(&mut ws).await["type"], "input_audio_buffer.append"); + send(&mut ws, json!({"type":"input_audio_buffer.speech_started"})).await; + if native { + let context = recv(&mut ws).await; + assert_eq!(context["text"], "View B"); + assert_eq!(context["revision"], 2); + } + assert_eq!(recv(&mut ws).await["type"], "input_audio_buffer.append"); + send( + &mut ws, + json!({"type":"input_audio_buffer.committed","item_id":"spoken"}), + ) + .await; + if !native { + let context = recv(&mut ws).await; + assert_eq!(context["type"], "conversation.item.create"); + assert_eq!(context["item"]["content"][0]["text"], "View A"); + } + assert_eq!(recv(&mut ws).await["type"], "response.create"); + created(&mut ws, "spoken_reply").await; + done(&mut ws, "spoken_reply", vec![]).await; + for (index, expected) in [ + if native { + "First text" + } else { + "View B\n\nFirst text" + }, + "Second text", + ] + .iter() + .enumerate() + { + let text = recv(&mut ws).await; + assert_eq!(text["type"], "conversation.item.create"); + assert_eq!(text["item"]["role"], "user"); + assert_eq!(text["item"]["content"][0]["text"], *expected); + send( + &mut ws, + json!({"type":"conversation.item.created","item":text["item"]}), + ) + .await; + assert_eq!(recv(&mut ws).await["type"], "response.create"); + let id = format!("typed_reply_{index}"); + created(&mut ws, &id).await; + done(&mut ws, &id, vec![]).await; + } + let _ = ws.next().await; + }); + let mut h = Harness::spawn_with_env(&url, &options()).await; + let id = h.send("initialize", json!({"protocolVersion":1,"clientCapabilities":{"_meta":{"buzz":{"realtimeAudio":1}}}})).await; + h.recv_until(|v| v["id"] == id).await; + let id = h + .send("session/new", json!({"cwd":dir.path(),"mcpServers":[]})) + .await; + let sid = h.recv_until(|v| v["id"] == id).await["result"]["sessionId"].clone(); + 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; + assert_eq!(ready["params"]["update"]["input"], true); + let stream = ready["params"]["streamId"].clone(); + for invalid in [ + json!({"text":""}), + json!({"context":"x".repeat(16385)}), + json!({"context":"A","data":"AA=="}), + ] { + let mut params = invalid; + params["sessionId"] = sid.clone(); + params["streamId"] = stream.clone(); + let id = h.send("_buzz/unstable/realtime/input", params).await; + assert!(h.recv_until(|v| v["id"] == id).await.get("error").is_some()); + } + for (sequence, context) in ["View A", "View B"].iter().enumerate() { + let id = h + .send( + "_buzz/unstable/realtime/input", + json!({"sessionId":sid,"streamId":stream,"context":context}), + ) + .await; + assert!(h.recv_until(|v| v["id"] == id).await.get("error").is_none()); + h.send("_buzz/unstable/realtime/append", json!({"sessionId":sid,"streamId":stream,"sequence":sequence,"data":STANDARD.encode(vec![0u8;960])})).await; + h.recv_until(|v| { + v["params"]["update"]["type"] + == if sequence == 0 { + "speech_started" + } else { + "response_done" + } + }) + .await; + } + for text in ["First text", "Second text"] { + let id = h + .send( + "_buzz/unstable/realtime/input", + json!({"sessionId":sid,"streamId":stream,"text":text}), + ) + .await; + let input = h + .recv_until(|v| { + v["params"]["update"]["type"] == "input_transcript" || v["id"] == id + }) + .await; + assert_eq!(input["params"]["update"]["text"], text); + h.recv_until(|v| v["params"]["update"]["type"] == "response_done") + .await; + } + h.send( + "_buzz/unstable/realtime/close", + json!({"sessionId":sid,"streamId":stream}), + ) + .await; + h.recv_until(|v| v["id"] == prompt).await; + server.await.unwrap(); + } +} + #[tokio::test] async fn realtime_acp_live_duplex_and_playback_fence() { let dir = tempfile::tempdir().unwrap();