perf(chat): send live text in batches, not one event per token

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

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<u8>,
     /// 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);
     }
 }

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`.

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<ProcessEvent>,
 ) {
+    // When the live text gathered so far goes out (spec 8.3).
+    let mut flush_at: Option<Instant> = 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);

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.");

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<Value> {
+        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);

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<u8> {
+    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."]);
+}

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