diff --git a/crates/botloftd/src/chat/events.rs b/crates/botloftd/src/chat/events.rs index c4f4be70..2fb3939c 100644 --- a/crates/botloftd/src/chat/events.rs +++ b/crates/botloftd/src/chat/events.rs @@ -5,9 +5,7 @@ use botloft_core::chat::{TOOL_OUTPUT_MAX, clip, tool_file, tool_input_max, tool_summary}; use botloft_core::command::tool_explanation; use botloft_core::ids::BotId; -use botloft_core::protocol::{ - ChatBody, ChatDelta, ReplyItem, TokenUsage, ToolItem, ToolStatus, TurnItem, -}; +use botloft_core::protocol::{ChatBody, ReplyItem, TokenUsage, ToolItem, ToolStatus, TurnItem}; use serde_json::Value; use tracing::{debug, warn}; @@ -37,10 +35,7 @@ pub(super) fn apply(daemon: &Daemon, bot: &BotId, generation: u64, event: &Value "system" if subtype == Some("status") => context::status(daemon, bot, event), "system" if subtype == Some("compact_boundary") => context::compacted(daemon, bot, event), "control_response" => control::answered(daemon, bot, event), - "stream_event" if !from_subagent => { - delta(daemon, bot, event); - crate::screens::stream(daemon, bot, event); - } + "stream_event" if !from_subagent => crate::screens::stream(daemon, bot, event), "assistant" if !from_subagent => assistant(daemon, bot, generation, event), "user" if event["isReplay"].as_bool() == Some(true) => { let uuid = event["uuid"].as_str(); @@ -77,19 +72,19 @@ fn init(daemon: &Daemon, bot: &BotId, event: &Value) { } } -fn delta(daemon: &Daemon, bot: &BotId, event: &Value) { +/// The piece of reply text a `stream_event` carries, if that is what it is +/// (spec 8.3). A subagent's text stays out of the chat. +pub(super) fn live_text(event: &Value) -> Option<&str> { + if event["type"] != "stream_event" || !event["parent_tool_use_id"].is_null() { + return None; + } let inner = &event["event"]; if inner["type"] != "content_block_delta" || inner["delta"]["type"] != "text_delta" { - return; - } - if let Some(text) = inner["delta"]["text"].as_str() - && !text.is_empty() - { - daemon.emit(Event::ChatDelta(ChatDelta { - bot_id: bot.clone(), - text: text.to_owned(), - })); + return None; } + inner["delta"]["text"] + .as_str() + .filter(|text| !text.is_empty()) } pub(super) fn blocks(event: &Value) -> &[Value] { diff --git a/crates/botloftd/src/chat/mod.rs b/crates/botloftd/src/chat/mod.rs index f4644316..45d25d06 100644 --- a/crates/botloftd/src/chat/mod.rs +++ b/crates/botloftd/src/chat/mod.rs @@ -9,13 +9,21 @@ pub(crate) mod items; mod session; use botloft_core::ids::BotId; +use botloft_core::protocol::ChatDelta; use tracing::debug; -use crate::state::Daemon; +use crate::state::{Daemon, Event}; /// Longest line kept; anything longer is dropped whole (spec 8.1). const LINE_MAX: usize = 8 << 20; +/// Live text waits at most this long to go out with the pieces after it +/// (spec 8.3): one `chat.delta` per token cost every app a render. +pub const LIVE_TEXT_EVERY: std::time::Duration = std::time::Duration::from_millis(40); + +/// Live text this long goes out without waiting. +const LIVE_TEXT_MAX: usize = 4 << 10; + /// Splits one process's stdout into lines. Lines may arrive split across /// chunks, and one chunk may hold several. pub struct StreamReader { @@ -24,6 +32,8 @@ pub struct StreamReader { buffer: Vec, /// Inside a line that grew past [`LINE_MAX`]; ends at its newline. skipping: bool, + /// Live text not sent yet. + live: String, } impl StreamReader { @@ -33,6 +43,7 @@ impl StreamReader { generation, buffer: Vec::new(), skipping: false, + live: String::new(), } } @@ -44,6 +55,21 @@ impl StreamReader { self.generation } + /// True while live text waits for [`StreamReader::flush`]. + pub fn holds_live_text(&self) -> bool { + !self.live.is_empty() + } + + /// Sends the live text gathered so far as one `chat.delta`. + pub fn flush(&mut self, daemon: &Daemon) { + if !self.live.is_empty() { + daemon.emit(Event::ChatDelta(ChatDelta { + bot_id: self.bot.clone(), + text: std::mem::take(&mut self.live), + })); + } + } + pub fn feed(&mut self, daemon: &Daemon, data: &[u8]) { let mut rest = data; while let Some(end) = rest.iter().position(|byte| *byte == b'\n') { @@ -67,14 +93,31 @@ impl StreamReader { } } - fn line(&self, daemon: &Daemon) { + fn line(&mut self, daemon: &Daemon) { let line = self.buffer.strip_suffix(b"\r").unwrap_or(&self.buffer); if line.iter().all(u8::is_ascii_whitespace) { return; } - match serde_json::from_slice(line) { - Ok(event) => events::apply(daemon, &self.bot, self.generation, &event), - Err(_) => debug!(bot = %self.bot, "stream line is not JSON"), + let event = match serde_json::from_slice(line) { + Ok(event) => event, + Err(_) => { + debug!(bot = %self.bot, "stream line is not JSON"); + return; + } + }; + if let Some(text) = events::live_text(&event) { + // Lines from a process that was replaced say nothing about the + // new one. + if daemon.supervisor.is_current(&self.bot, self.generation) { + self.live.push_str(text); + if self.live.len() >= LIVE_TEXT_MAX { + self.flush(daemon); + } + } + return; } + // Whatever comes next follows the text before it. + self.flush(daemon); + events::apply(daemon, &self.bot, self.generation, &event); } } diff --git a/crates/botloftd/src/state.rs b/crates/botloftd/src/state.rs index 2c2e1420..6a443d2e 100644 --- a/crates/botloftd/src/state.rs +++ b/crates/botloftd/src/state.rs @@ -52,7 +52,7 @@ pub enum Event { } /// Events buffered per connection before a slow client is dropped. -const EVENT_BUFFER: usize = 1024; +const EVENT_BUFFER: usize = 4096; /// Settings for what bots may ask of the owner (spec 6, `[bots]`). How /// long a request waits for the owner is live: `LiveSettings`. diff --git a/crates/botloftd/src/supervisor/spawn.rs b/crates/botloftd/src/supervisor/spawn.rs index 84f89e1c..291e0016 100644 --- a/crates/botloftd/src/supervisor/spawn.rs +++ b/crates/botloftd/src/supervisor/spawn.rs @@ -13,7 +13,7 @@ use tracing::{debug, warn}; use super::slot::{Running, StopIntent}; use super::{ClaudeStatus, Inner, STABLE_AFTER, Supervisor}; -use crate::chat::{StreamReader, control}; +use crate::chat::{LIVE_TEXT_EVERY, StreamReader, control}; use crate::platform; use crate::runtime::claude::Claude; use crate::runtime::{ProcessEvent, SpawnSpec}; @@ -291,19 +291,39 @@ async fn pump( mut reader: StreamReader, mut events: mpsc::Receiver, ) { + // When the live text gathered so far goes out (spec 8.3). + let mut flush_at: Option = None; let code = loop { - match events.recv().await { + let event = tokio::select! { + event = events.recv() => event, + () = tokio::time::sleep_until(flush_at.unwrap_or_else(Instant::now)), + if flush_at.is_some() => + { + flush_at = None; + if let Some(daemon) = daemon.upgrade() { + reader.flush(&daemon); + } + continue; + } + }; + match event { Some(ProcessEvent::Output(data)) => { let Some(daemon) = daemon.upgrade() else { return; }; reader.feed(&daemon, &data); + if !reader.holds_live_text() { + flush_at = None; + } else if flush_at.is_none() { + flush_at = Some(Instant::now() + LIVE_TEXT_EVERY); + } } Some(ProcessEvent::Exited(code)) => break code, None => break None, } }; if let Some(daemon) = daemon.upgrade() { + reader.flush(&daemon); daemon .supervisor .on_exit(&daemon, reader.bot(), reader.generation(), code); diff --git a/crates/botloftd/tests/chat.rs b/crates/botloftd/tests/chat.rs index 62c49e41..39b26f4b 100644 --- a/crates/botloftd/tests/chat.rs +++ b/crates/botloftd/tests/chat.rs @@ -73,8 +73,12 @@ async fn what_the_bot_prints_becomes_the_chat() { p.emit(stream::text("All 12 tests pass.")).await; p.emit(stream::result(false)).await; - assert_eq!(c.app.notification("chat.delta").await["text"], "Run"); - assert_eq!(c.app.notification("chat.delta").await["text"], "ning."); + // Pieces written close together may arrive as one (spec 8.3). + let mut live = String::new(); + while live != "Running." { + let delta = c.app.notification("chat.delta").await; + live.push_str(delta["text"].as_str().expect("text")); + } let reply = kind(&mut c.app, "reply").await; assert_eq!(reply["item"]["body"]["text"], "Running."); assert_eq!(reply["activity"]["text"], "Running."); diff --git a/crates/botloftd/tests/common/mod.rs b/crates/botloftd/tests/common/mod.rs index edd7da5c..7b6c894f 100644 --- a/crates/botloftd/tests/common/mod.rs +++ b/crates/botloftd/tests/common/mod.rs @@ -288,6 +288,13 @@ impl Client { } } + /// The notifications named `method` that came before the last one + /// waited for, without waiting for more. + pub fn queued(&self, method: &str) -> Vec { + let named = self.notifications.iter().filter(|n| n["method"] == method); + named.map(|n| n["params"].clone()).collect() + } + /// Drops the notifications named `method` that came so far. pub fn forget(&mut self, method: &str) { self.notifications.retain(|n| n["method"] != method); diff --git a/crates/botloftd/tests/live_text.rs b/crates/botloftd/tests/live_text.rs new file mode 100644 index 00000000..94bd6a19 --- /dev/null +++ b/crates/botloftd/tests/live_text.rs @@ -0,0 +1,47 @@ +//! Live text (spec 8.3): pieces of a reply that arrive close together reach +//! the app as one `chat.delta`, always before what the bot does next. + +mod common; + +use common::bots::two_bots; +use common::stream; +use serde_json::Value; + +/// Several stream-json events in one chunk of stdout. +fn chunk(events: &[Value]) -> Vec { + events + .iter() + .flat_map(|event| format!("{event}\n").into_bytes()) + .collect() +} + +#[tokio::test] +async fn pieces_that_arrive_together_go_out_as_one() { + let mut c = two_bots().await; + let pieces = ["Hel", "lo, ", "world"].map(stream::delta); + c.lead_process.output(&chunk(&pieces)).await; + + let delta = c.app.notification("chat.delta").await; + assert_eq!(delta["text"], "Hello, world"); +} + +#[tokio::test] +async fn live_text_goes_out_before_the_reply_it_becomes() { + let mut c = two_bots().await; + let events = [ + stream::delta("All "), + stream::delta("done."), + stream::text("All done."), + ]; + c.lead_process.output(&chunk(&events)).await; + + loop { + let changed = c.app.notification("chat.item").await; + if changed["item"]["body"]["kind"] == "reply" { + break; + } + } + let before = c.app.queued("chat.delta"); + let texts: Vec<_> = before.iter().map(|delta| delta["text"].clone()).collect(); + assert_eq!(texts, ["All done."]); +} diff --git a/docs/spec.md b/docs/spec.md index 084d4bd6..9fab7dd8 100644 --- a/docs/spec.md +++ b/docs/spec.md @@ -357,7 +357,7 @@ Todo item tem `id` (`cht_`), `botId`, `kind`, `createdAt` e `updatedAt`. ### 8.3 Ao vivo - `chat.item {item, activity}`: item novo ou atualizado (tool que terminou, aprovação respondida). `activity` é a nova linha da conversa na barra lateral, quando mudou. -- `chat.delta {botId, text}`: pedaço do texto que o bot está escrevendo, na ordem. O app junta os pedaços num balão provisório, trocado pelo `reply` quando ele chega. Um app que conecta no meio de um turno não vê o texto parcial já passado, só o que vier depois e o `reply` final. +- `chat.delta {botId, text}`: pedaço do texto que o bot está escrevendo, na ordem. O daemon junta os pedaços que chegam em até 40 ms (ou até 4 KB) num só `chat.delta`, e manda o que juntou antes de qualquer outro evento do mesmo bot: um evento por token custava a cada app uma renderização. O app junta os pedaços num balão provisório, trocado pelo `reply` quando ele chega. Um app que conecta no meio de um turno não vê o texto parcial já passado, só o que vier depois e o `reply` final. ### 8.4 Arquivos do bot