From 69c76a4515ddd1d20b4d1aeb3b445223c888924b Mon Sep 17 00:00:00 2001 From: Phelipe Lorran <112786693+httpsphl@users.noreply.github.com> Date: Wed, 30 Sep 2026 19:55:50 -0300 Subject: [PATCH] perf(chat): send live text in batches, not one event per token Every text_delta became its own chat.delta: at tens of tokens a second per busy bot, each app connection serialized a frame and re-rendered the chat for each one, and a burst could make a connection lag behind the event bus and be dropped. The stream reader now gathers live text and sends it as one chat.delta 40 ms after the first piece, at 4 KB, before any other event of the bot (so the text always comes before the reply that replaces it) and when the process exits. The event bus holds 4096 events instead of 1024. --- crates/botloftd/src/chat/events.rs | 29 ++++++-------- crates/botloftd/src/chat/mod.rs | 53 ++++++++++++++++++++++--- crates/botloftd/src/state.rs | 2 +- crates/botloftd/src/supervisor/spawn.rs | 24 ++++++++++- crates/botloftd/tests/chat.rs | 8 +++- crates/botloftd/tests/common/mod.rs | 7 ++++ crates/botloftd/tests/live_text.rs | 47 ++++++++++++++++++++++ docs/spec.md | 2 +- 8 files changed, 144 insertions(+), 28 deletions(-) create mode 100644 crates/botloftd/tests/live_text.rs 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