From 60742e5825fa442e7126ad8ebdb42cfb3539b466 Mon Sep 17 00:00:00 2001 From: Phelipe Lorran <112786693+httpsphl@users.noreply.github.com> Date: Fri, 9 Oct 2026 17:29:58 -0300 Subject: [PATCH] refactor(agent): each agent reads its own output (A1b) Agent gains live_text and handle_output; chat::events becomes chat::claude, the Claude decoder, and the generic parts (reply, tool start and end, end of turn) go through chat::sink. StreamReader takes the bot's agent. No change for Claude bots. --- crates/botloftd/src/agent/claude.rs | 13 ++- crates/botloftd/src/agent/mod.rs | 14 +++ crates/botloftd/src/chat/account.rs | 4 +- .../src/chat/{events.rs => claude.rs} | 80 ++++------------- crates/botloftd/src/chat/mod.rs | 14 ++- crates/botloftd/src/chat/sink.rs | 86 +++++++++++++++++++ crates/botloftd/src/supervisor/start.rs | 5 +- docs/spec.md | 2 +- 8 files changed, 148 insertions(+), 70 deletions(-) rename crates/botloftd/src/chat/{events.rs => claude.rs} (73%) create mode 100644 crates/botloftd/src/chat/sink.rs diff --git a/crates/botloftd/src/agent/claude.rs b/crates/botloftd/src/agent/claude.rs index 1543ee96..a4ed31ef 100644 --- a/crates/botloftd/src/agent/claude.rs +++ b/crates/botloftd/src/agent/claude.rs @@ -3,11 +3,14 @@ use std::ffi::OsString; +use botloft_core::ids::BotId; use botloft_core::protocol::AgentKind; use bytes::Bytes; -use serde_json::json; +use serde_json::{Value, json}; use super::{Agent, LaunchPlan, Turn}; +use crate::chat::claude; +use crate::state::Daemon; /// Tools the bot uses without asking: its crew tools (spec 7.4). const ALLOWED_TOOLS: &str = "mcp__botloft"; @@ -107,6 +110,14 @@ impl Agent for ClaudeAgent { line.push(b'\n'); Bytes::from(line) } + + fn live_text<'a>(&self, event: &'a Value) -> Option<&'a str> { + claude::live_text(event) + } + + fn handle_output(&self, daemon: &Daemon, bot: &BotId, generation: u64, event: &Value) { + claude::apply(daemon, bot, generation, event); + } } #[cfg(test)] diff --git a/crates/botloftd/src/agent/mod.rs b/crates/botloftd/src/agent/mod.rs index 2ceff246..5ea0a201 100644 --- a/crates/botloftd/src/agent/mod.rs +++ b/crates/botloftd/src/agent/mod.rs @@ -9,11 +9,15 @@ mod claude; use std::ffi::OsString; use std::path::Path; +use botloft_core::ids::BotId; use botloft_core::protocol::{AgentKind, BotEffort, BotModel, PermissionMode}; use bytes::Bytes; +use serde_json::Value; pub use claude::ClaudeAgent; +use crate::state::Daemon; + /// One image that goes in the turn, already read and encoded. #[derive(Debug, Clone, PartialEq, Eq)] pub struct TurnImage { @@ -55,6 +59,16 @@ pub trait Agent: Send + Sync + 'static { /// One turn as the bytes written to the process's stdin; `uuid` is what /// the agent gives back when the turn begins. fn encode_turn(&self, uuid: &str, turn: &Turn) -> Bytes; + + /// The piece of reply text one line of output carries while the reply is + /// still being written, if that is what it is (spec 8.3). + fn live_text<'a>(&self, event: &'a Value) -> Option<&'a str>; + + /// What one line of output (already JSON) means for the chat and the + /// bot's state: it saves items and tells the supervisor, through + /// `chat::sink` for what every agent has and its own code for the rest. + /// Lines from a process that was replaced are ignored (spec 30). + fn handle_output(&self, daemon: &Daemon, bot: &BotId, generation: u64, event: &Value); } static CLAUDE: ClaudeAgent = ClaudeAgent; diff --git a/crates/botloftd/src/chat/account.rs b/crates/botloftd/src/chat/account.rs index 0dea8e43..76bcb73f 100644 --- a/crates/botloftd/src/chat/account.rs +++ b/crates/botloftd/src/chat/account.rs @@ -7,7 +7,7 @@ use botloft_core::protocol::{ }; use serde_json::Value; -use super::{events, items}; +use super::{claude, items}; use crate::state::Daemon; /// Errors that restarting cannot fix: Claude Code needs the owner. @@ -31,7 +31,7 @@ pub(super) fn failed_turn( error: &str, event: &Value, ) { - let detail = events::blocks(event) + let detail = claude::blocks(event) .iter() .filter_map(|block| block["text"].as_str()) .collect::>() diff --git a/crates/botloftd/src/chat/events.rs b/crates/botloftd/src/chat/claude.rs similarity index 73% rename from crates/botloftd/src/chat/events.rs rename to crates/botloftd/src/chat/claude.rs index 80db5186..c4456d0a 100644 --- a/crates/botloftd/src/chat/events.rs +++ b/crates/botloftd/src/chat/claude.rs @@ -1,19 +1,18 @@ -//! What each stream-json event means for the chat and the bot's state +//! Claude Code's stream-json, decoded into the chat (spec 8.1, 30): what each +//! event means for the chat and the bot's state //! (spec 8.1). Fields are read defensively: the format is not documented //! and grows between Claude Code versions. -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, ReplyItem, TokenUsage, ToolItem, ToolStatus, TurnItem}; +use botloft_core::protocol::TokenUsage; use serde_json::Value; use tracing::{debug, warn}; -use super::{account, control, items}; +use super::{account, control, sink}; use crate::state::{Daemon, Event}; use crate::{context, routines, service}; -pub(super) fn apply(daemon: &Daemon, bot: &BotId, generation: u64, event: &Value) { +pub(crate) fn apply(daemon: &Daemon, bot: &BotId, generation: u64, event: &Value) { // Lines from a process that was replaced say nothing about the new one. if !daemon.supervisor.is_current(bot, generation) { return; @@ -74,7 +73,7 @@ fn init(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> { +pub(crate) fn live_text(event: &Value) -> Option<&str> { if event["type"] != "stream_event" || !event["parent_tool_use_id"].is_null() { return None; } @@ -105,36 +104,14 @@ fn assistant(daemon: &Daemon, bot: &BotId, generation: u64, event: &Value) { context::used(daemon, bot, &event["message"]); for block in blocks(event) { match block["type"].as_str() { - Some("text") => { - let text = block["text"].as_str().unwrap_or_default().trim(); - if !text.is_empty() { - items::add( - daemon, - bot, - ChatBody::Reply(ReplyItem { - text: text.to_owned(), - }), - ); - } - } - Some("tool_use") => { - let name = block["name"].as_str().unwrap_or("tool"); - let input = &block["input"]; - items::add( - daemon, - bot, - ChatBody::Tool(ToolItem { - tool_use_id: block["id"].as_str().unwrap_or_default().to_owned(), - name: name.to_owned(), - summary: tool_summary(name, input), - explanation: tool_explanation(name, input), - input: clip(&input.to_string(), tool_input_max(name)), - status: ToolStatus::Running, - output: None, - file: tool_file(name, input), - }), - ); - } + Some("text") => sink::reply(daemon, bot, block["text"].as_str().unwrap_or_default()), + Some("tool_use") => sink::tool_started( + daemon, + bot, + block["id"].as_str().unwrap_or_default(), + block["name"].as_str().unwrap_or("tool"), + &block["input"], + ), _ => {} } } @@ -167,25 +144,8 @@ fn tool_results(daemon: &Daemon, bot: &BotId, event: &Value) { continue; }; crate::screens::tool_done(daemon, bot, id); - let found = daemon.store().tool_item(bot, id); - let Ok(Some(item)) = found else { - continue; - }; - let ChatBody::Tool(tool) = item.body else { - continue; - }; let failed = block["is_error"].as_bool() == Some(true); - let output = result_text(&block["content"]); - let body = ChatBody::Tool(ToolItem { - status: if failed { - ToolStatus::Failed - } else { - ToolStatus::Done - }, - output: (!output.is_empty()).then(|| clip(&output, TOOL_OUTPUT_MAX)), - ..tool - }); - items::update(daemon, &item.id, body); + sink::tool_finished(daemon, bot, id, failed, &result_text(&block["content"])); } } @@ -221,14 +181,12 @@ fn result(daemon: &Daemon, bot: &BotId, generation: u64, event: &Value) { }); // A compaction ends like a turn; its notice already says what happened. if !context::turn_ended(daemon, bot, event) { - items::add( + sink::turn_finished( daemon, bot, - ChatBody::Turn(TurnItem { - duration_ms: event["duration_ms"].as_u64().unwrap_or_default(), - tokens: tokens(&event["usage"], reloaded), - error, - }), + event["duration_ms"].as_u64().unwrap_or_default(), + tokens(&event["usage"], reloaded), + error, ); } routines::turn_ended(daemon, bot, failed); diff --git a/crates/botloftd/src/chat/mod.rs b/crates/botloftd/src/chat/mod.rs index 45d25d06..2be530af 100644 --- a/crates/botloftd/src/chat/mod.rs +++ b/crates/botloftd/src/chat/mod.rs @@ -3,15 +3,17 @@ //! state changes. mod account; +pub(crate) mod claude; pub(crate) mod control; -mod events; pub(crate) mod items; mod session; +pub(crate) mod sink; use botloft_core::ids::BotId; use botloft_core::protocol::ChatDelta; use tracing::debug; +use crate::agent::Agent; use crate::state::{Daemon, Event}; /// Longest line kept; anything longer is dropped whole (spec 8.1). @@ -29,6 +31,8 @@ const LIVE_TEXT_MAX: usize = 4 << 10; pub struct StreamReader { bot: BotId, generation: u64, + /// Reads the lines: they are the agent's own protocol. + agent: &'static dyn Agent, buffer: Vec, /// Inside a line that grew past [`LINE_MAX`]; ends at its newline. skipping: bool, @@ -37,10 +41,11 @@ pub struct StreamReader { } impl StreamReader { - pub fn new(bot: BotId, generation: u64) -> Self { + pub fn new(bot: BotId, generation: u64, agent: &'static dyn Agent) -> Self { Self { bot, generation, + agent, buffer: Vec::new(), skipping: false, live: String::new(), @@ -105,7 +110,7 @@ impl StreamReader { return; } }; - if let Some(text) = events::live_text(&event) { + if let Some(text) = self.agent.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) { @@ -118,6 +123,7 @@ impl StreamReader { } // Whatever comes next follows the text before it. self.flush(daemon); - events::apply(daemon, &self.bot, self.generation, &event); + self.agent + .handle_output(daemon, &self.bot, self.generation, &event); } } diff --git a/crates/botloftd/src/chat/sink.rs b/crates/botloftd/src/chat/sink.rs new file mode 100644 index 00000000..54a97da7 --- /dev/null +++ b/crates/botloftd/src/chat/sink.rs @@ -0,0 +1,86 @@ +//! What any agent's output turns into in the chat (spec 8, 30): a reply, a +//! tool that starts and finishes, the end of a turn. An agent decodes its own +//! stream and calls these; saving the items and telling the app is the same +//! for all of them. + +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, ReplyItem, TokenUsage, ToolItem, ToolStatus, TurnItem}; +use serde_json::Value; + +use super::items; +use crate::state::Daemon; + +/// The bot's reply text; nothing for an empty one. +pub(crate) fn reply(daemon: &Daemon, bot: &BotId, text: &str) { + let text = text.trim(); + if !text.is_empty() { + items::add( + daemon, + bot, + ChatBody::Reply(ReplyItem { + text: text.to_owned(), + }), + ); + } +} + +/// A tool the bot began to use, shown as running. +pub(crate) fn tool_started(daemon: &Daemon, bot: &BotId, id: &str, name: &str, input: &Value) { + items::add( + daemon, + bot, + ChatBody::Tool(ToolItem { + tool_use_id: id.to_owned(), + name: name.to_owned(), + summary: tool_summary(name, input), + explanation: tool_explanation(name, input), + input: clip(&input.to_string(), tool_input_max(name)), + status: ToolStatus::Running, + output: None, + file: tool_file(name, input), + }), + ); +} + +/// The tool `id` finished, with what it printed. Does nothing for a tool the +/// chat does not know. +pub(crate) fn tool_finished(daemon: &Daemon, bot: &BotId, id: &str, failed: bool, output: &str) { + let Ok(Some(item)) = daemon.store().tool_item(bot, id) else { + return; + }; + let ChatBody::Tool(tool) = item.body else { + return; + }; + let body = ChatBody::Tool(ToolItem { + status: if failed { + ToolStatus::Failed + } else { + ToolStatus::Done + }, + output: (!output.is_empty()).then(|| clip(output, TOOL_OUTPUT_MAX)), + ..tool + }); + items::update(daemon, &item.id, body); +} + +/// The line that closes a turn: how long it took, its tokens when the agent +/// says, and the error it ended with, if any. +pub(crate) fn turn_finished( + daemon: &Daemon, + bot: &BotId, + duration_ms: u64, + tokens: Option, + error: Option, +) { + items::add( + daemon, + bot, + ChatBody::Turn(TurnItem { + duration_ms, + tokens, + error, + }), + ); +} diff --git a/crates/botloftd/src/supervisor/start.rs b/crates/botloftd/src/supervisor/start.rs index f557578a..d2254e70 100644 --- a/crates/botloftd/src/supervisor/start.rs +++ b/crates/botloftd/src/supervisor/start.rs @@ -14,6 +14,7 @@ use tracing::{debug, warn}; use super::slot::Running; use super::spawn::{Launch, launch_spec, pump, settle}; use super::{ClaudeStatus, Inner, Supervisor}; +use crate::agent::{self, ClaudeAgent}; use crate::chat::{StreamReader, control}; use crate::context; use crate::runtime::Process; @@ -156,7 +157,9 @@ impl Supervisor { slot.state = BotState::Launching; self.announce(&bot.id, slot); debug!(bot = %bot.id, generation, pid = ?process.pid, resumed = launch.resumed, "bot started"); - let reader = StreamReader::new(bot.id.clone(), generation); + // The bot launched, so its agent runs. + let agent = agent::of(bot.agent).unwrap_or(&ClaudeAgent); + let reader = StreamReader::new(bot.id.clone(), generation, agent); tokio::spawn(pump(self.daemon.clone(), reader, process.events)); tokio::spawn(settle( self.daemon.clone(), diff --git a/docs/spec.md b/docs/spec.md index 7a21de52..5d7329ae 100644 --- a/docs/spec.md +++ b/docs/spec.md @@ -2643,7 +2643,7 @@ Decisão e contexto: `docs/adr/0003-agentes.md`. Um bot tem um **agente**, `Bot. | Passo | O que entra | Estado | |---|---|---| | **A0** | ADR, esta seção, `bots.agent` no banco e no protocolo; `bots.create` com `codex` ou `agy` dá `-32004` ("not available yet") | feito | -| **A1** | A interface `Agent` no daemon (`crates/botloftd/src/agent`), com o Claude por trás e os testes de hoje sem mudança. Em fatias: **A1a** argumentos do processo, ambiente extra e codificação do turno no stdin (feito); **A1b** leitura da saída em eventos normalizados; **A1c** descoberta e login, arquivos da pasta do bot e aprovações | em andamento | +| **A1** | A interface `Agent` no daemon (`crates/botloftd/src/agent`), com o Claude por trás e os testes de hoje sem mudança. Em fatias: **A1a** argumentos do processo, ambiente extra e codificação do turno no stdin (feito); **A1b** leitura da saída: cada agente decodifica o próprio fluxo (`Agent::handle_output`, `Agent::live_text`) e grava no chat por `chat::sink` (resposta, ferramenta que começa e termina, fim de turno), o que todo agente tem; o que é só do Claude (contexto, compactação, parte do plano, rascunho de telas) fica no decodificador dele, `chat/claude.rs` (feito); **A1c** descoberta e login, arquivos da pasta do bot e aprovações | em andamento | | **A2** | AGY: lançamento, entrada e saída, MCP, aprovações, isolamento, modelos e esforço, testados com o `agy` real | a fazer | | **A3** | Codex por `codex app-server` | a fazer | | **A4** | Seletor de agente no app, conta e login de cada agente, paridade de rotinas, navegador, desktop e perguntas | a fazer |