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.

Author: httpsphl

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

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;

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::<Vec<_>>()

crates/botloftd/src/chat/events.rs crates/botloftd/src/chat/claude.rscrates/botloftd/src/chat/events.rs renamed to 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);

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

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<TokenUsage>,
+    error: Option<String>,
+) {
+    items::add(
+        daemon,
+        bot,
+        ChatBody::Turn(TurnItem {
+            duration_ms,
+            tokens,
+            error,
+        }),
+    );
+}

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(),

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 |