jerry/buzz
1
0
Fork 0
mirror of https://github.com/block/buzz.git synced 2026-10-28 21:11:48 +00:00
Code Issues Packages Projects Releases Wiki Activity

agent: accept view context and text during realtime sessions

Browse Source 
Signed-off-by: Codex <codex@openai.com>
This commit is contained in:
Codex
2026-09-12 23:55:22 -04:00
parent 9bab300881
commit 049fed56bb
6 changed files with 262 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files

9
crates/buzz-agent/README.md
View File

@@ -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,

9
crates/buzz-agent/src/realtime.rs
View File

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

30
crates/buzz-agent/src/realtime_media.rs
View File

@@ -25,6 +25,10 @@ pub(crate) enum Action {
},
Playback(Playback),
Interrupt(Option<Playback>),
Input {
context: Option<String>,
text: Option<String>,
},
Close,
}
@@ -52,6 +56,10 @@ struct Params {
playback_data: Option<String>,
#[serde(default)]
playback: Option<Playback>,
#[serde(default)]
context: Option<String>,
#[serde(default)]
text: Option<String>,
}
pub(crate) struct Ingress {
@@ -89,6 +97,28 @@ pub(crate) fn enabled(meta: &Value) -> bool {
}
fn action(operation: &str, params: &mut Params) -> Result<Action, &'static str> {
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");
}

3
crates/buzz-agent/src/realtime_session.rs
View File

@@ -51,6 +51,7 @@ struct Live {
active_response: Option<String>,
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()

63
crates/buzz-agent/src/realtime_session/media.rs
View File

@@ -23,6 +23,13 @@ struct State {
response_requested: bool,
fence_deadline: Option<Instant>,
response_deadline: Option<Instant>,
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 {

151
crates/buzz-agent/tests/realtime_acp.rs
View File

@@ -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();