Agent: one turn path, tools and history in their own modules; SDK: one parser

agent/src/history.rs

@@ -0,0 +1,81 @@
+//! History window selection and context-size error detection.
+
+use crate::llm::Entry;
+
+pub fn is_too_large(error: &str) -> bool {
+    let Some((status, body)) = error
+        .strip_prefix("HTTP ")
+        .and_then(|error| error.split_once(':'))
+    else {
+        return false;
+    };
+    if !matches!(status.split_whitespace().next(), Some("400" | "413")) {
+        return false;
+    }
+    let body = body.to_ascii_lowercase();
+    body.contains("context_length_exceeded")
+        || body.contains("maximum context length")
+        || body.contains("context length exceeded")
+        || body.contains("too large")
+        || (body.contains("too long")
+            && ["context", "input", "prompt"]
+                .iter()
+                .any(|term| body.contains(term)))
+}
+
+pub fn next_turn_start(history: &[Entry], from: usize, current: usize) -> Option<usize> {
+    let end = current.min(history.len().checked_sub(1)?);
+    (from..=end).find(|&index| matches!(&history[index], Entry::User { .. }))
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn detects_context_size_errors_only_for_http_400_or_413() {
+        assert!(is_too_large("HTTP 400: context_length_exceeded"));
+        assert!(is_too_large("HTTP 413: prompt is too long"));
+        assert!(is_too_large("HTTP 400: maximum context length"));
+        assert!(is_too_large("HTTP 400: request is too large"));
+        assert!(!is_too_large("HTTP 500: prompt is too long"));
+        assert!(!is_too_large("HTTP 400: invalid token"));
+    }
+
+    #[test]
+    fn history_skip_advances_by_whole_user_turns() {
+        let history = vec![
+            Entry::User {
+                text: "first".into(),
+            },
+            Entry::ToolCall {
+                id: "a".into(),
+                name: "tool".into(),
+                arguments: "{}".into(),
+            },
+            Entry::ToolResult {
+                id: "a".into(),
+                output: "done".into(),
+            },
+            Entry::User {
+                text: "second".into(),
+            },
+            Entry::ToolCall {
+                id: "b".into(),
+                name: "tool".into(),
+                arguments: "{}".into(),
+            },
+            Entry::ToolResult {
+                id: "b".into(),
+                output: "done".into(),
+            },
+            Entry::User {
+                text: "current".into(),
+            },
+        ];
+        assert_eq!(next_turn_start(&history, 0, 6), Some(0));
+        assert_eq!(next_turn_start(&history, 1, 6), Some(3));
+        assert_eq!(next_turn_start(&history, 4, 6), Some(6));
+        assert_eq!(next_turn_start(&history, usize::MAX, 6).unwrap_or(6), 6);
+    }
+}

agent/src/main.rs

@@ -1,8 +1,11 @@
 //! The Sausage agent actor.
 
+mod history;
 mod llm;
+mod tools;
 
-use llm::{Config, Entry, Tool};
+use history::{is_too_large, next_turn_start};
+use llm::{Config, Entry};
 use sausage_actor::{Inbox, Message, Value, json, publish, send};
 use std::collections::{HashSet, VecDeque};
 use std::io::Write;
@@ -19,80 +22,6 @@ fn main() {
     wstd::runtime::block_on(async { Agent::new().run().await });
 }
 
-fn tools() -> Vec<Tool> {
-    vec![
-        Tool {
-            name: "send_message",
-            description: "Send an actor a JSON body with a `type`. `broker` and `cargo` are built-in actors; see `/actor/broker/public/PROTOCOL.md`. With `wait_seconds`, the reply's `reply_to` matches the returned ID. Returns the ID, reply, and other messages received.",
-            parameters: json!({
-                "type": "object",
-                "properties": {
-                    "to": {"type": "string", "description": "Actor ID"},
-                    "body": {"type": "object", "description": "JSON object with a type field"},
-                    "wait_seconds": {"type": "number", "description": "Seconds to wait for a reply. Default 0."}
-                },
-                "required": ["to", "body"]
-            }),
-        },
-        Tool {
-            name: "read_file",
-            description: "Read a text file under `/data`, `/public`, or `/actor/<id>`. Reads all bytes by default; `offset` and `max_bytes` are optional.",
-            parameters: json!({
-                "type": "object",
-                "properties": {
-                    "path": {"type": "string"},
-                    "offset": {"type": "integer", "description": "Byte offset (default 0)"},
-                    "max_bytes": {"type": "integer", "description": "Optional maximum number of bytes to read"}
-                },
-                "required": ["path"]
-            }),
-        },
-        Tool {
-            name: "write_file",
-            description: "Write text under `/data` or `/public`. Parent folders are created.",
-            parameters: json!({
-                "type": "object",
-                "properties": {"path": {"type": "string"}, "content": {"type": "string"}},
-                "required": ["path", "content"]
-            }),
-        },
-        Tool {
-            name: "list_dir",
-            description: "List entries in a directory under `/data`, `/public`, or `/actor/<id>`. Returns each name, type, and size.",
-            parameters: json!({"type": "object", "properties": {"path": {"type": "string"}}, "required": ["path"]}),
-        },
-        Tool {
-            name: "copy_dir",
-            description: "Copy a directory tree, for example `/actor/<id>/source` to `/public/drafts/<id>`. Skips `target/` folders.",
-            parameters: json!({
-                "type": "object",
-                "properties": {"from": {"type": "string"}, "to": {"type": "string"}},
-                "required": ["from", "to"]
-            }),
-        },
-    ]
-}
-
-fn copy_tree(from: &str, to: &str) -> std::io::Result<u64> {
-    std::fs::create_dir_all(to)?;
-    let mut count = 0;
-    for entry in std::fs::read_dir(from)? {
-        let entry = entry?;
-        let name = entry.file_name().to_string_lossy().to_string();
-        let src = format!("{from}/{name}");
-        let dst = format!("{to}/{name}");
-        if entry.file_type()?.is_dir() {
-            if name != "target" {
-                count += copy_tree(&src, &dst)?;
-            }
-        } else {
-            std::fs::copy(src, dst)?;
-            count += 1;
-        }
-    }
-    Ok(count)
-}
-
 struct Agent {
     inbox: Inbox,
     history: Vec<Entry>,
@@ -100,7 +29,6 @@ struct Agent {
     active: bool,
     queue: VecDeque<Message>,
     notes: Vec<Message>,
-    shutdown: bool,
 }
 
 impl Agent {
@@ -125,7 +53,6 @@ impl Agent {
             active,
             queue: VecDeque::new(),
             notes: Vec::new(),
-            shutdown: false,
         }
     }
 
@@ -151,6 +78,10 @@ impl Agent {
         let _ = std::fs::remove_file(TURN);
     }
 
+    fn closing(&self) -> bool {
+        self.inbox.is_closed()
+    }
+
     fn repair_dangling_tools(&mut self) {
         let answered: HashSet<String> = self
             .history
@@ -207,41 +138,50 @@ impl Agent {
 
     async fn run(mut self) {
         self.repair_dangling_tools();
-        self.resume().await;
+        if self.active {
-        while !self.shutdown {
+            self.turn(None).await;
+            if self.closing() && self.active {
+                return;
+            }
+        }
+        loop {
             let message = match self.queue.pop_front() {
                 Some(message) => message,
                 None => match self.inbox.recv().await {
                     Some(message) => message,
-                    None => {
+                    None => break,
-                        self.shutdown = true;
-                        break;
-                    }
                 },
             };
             if starts_turn(&message) {
-                self.turn(message).await;
+                self.turn(Some(message)).await;
+                if self.closing() && self.active {
+                    break;
+                }
             } else {
                 self.notes.push(message);
             }
         }
     }
 
-    async fn resume(&mut self) {
+    async fn turn(&mut self, message: Option<Message>) {
-        if !self.active {
+        let current_start = if message.is_some() {
-            return;
+            self.active = true;
-        }
+            let _ = std::fs::write(TURN, "true");
-        let current_start = self
+            self.history.len()
-            .history
+        } else {
-            .iter()
+            let current_start = self
-            .rposition(|entry| matches!(entry, Entry::User { text } if !text.starts_with("[note from your runtime]")))
+                .history
-            .unwrap_or(self.history.len());
+                .iter()
-        let version = std::env::var("SAUSAGE_VERSION").unwrap_or_default();
+                .rposition(|entry| matches!(entry, Entry::User { text } if !text.starts_with("[note from your runtime]")))
-        self.add_history(Entry::User {
+                .unwrap_or(self.history.len());
-            text: format!(
+            let version = std::env::var("SAUSAGE_VERSION").unwrap_or_default();
-                "[note from your runtime] you restarted and now run version {version}. Continue the active turn."
+            self.add_history(Entry::User {
-            ),
+                text: format!(
-        });
+                    "[note from your runtime] you restarted and now run version {version}. Continue the active turn."
+                ),
+            });
+            current_start
+        };
         let cfg = match load_config() {
             Ok(cfg) => cfg,
             Err(error) => {
@@ -249,40 +189,25 @@ impl Agent {
                 return;
             }
         };
-        self.steps(&cfg, current_start).await;
+        if let Some(message) = message {
-        if !self.shutdown {
+            let text = self.turn_text(&message);
-            self.finish_turn();
+            self.notes.clear();
+            self.add_history(Entry::User { text });
         }
-    }
-
-    async fn turn(&mut self, message: Message) {
-        self.active = true;
-        let _ = std::fs::write(TURN, "true");
-        let cfg = match load_config() {
-            Ok(cfg) => cfg,
-            Err(error) => {
-                self.not_configured(error);
-                return;
-            }
-        };
-        let text = self.turn_text(&message);
-        self.notes.clear();
-        let current_start = self.history.len();
-        self.add_history(Entry::User { text });
         self.steps(&cfg, current_start).await;
-        if !self.shutdown {
+        if !self.closing() {
             self.finish_turn();
         }
     }
 
     async fn steps(&mut self, cfg: &Config, current_start: usize) {
-        let tools = tools();
+        let tools = tools::definitions();
         loop {
-            if self.inbox.is_closed() {
+            if self.closing() {
-                self.shutdown = true;
                 return;
             }
-            let start = history_start(&self.history, self.history_skip, current_start);
+            let start = next_turn_start(&self.history, self.history_skip, current_start)
+                .unwrap_or(current_start);
             if start != self.history_skip {
                 self.history_skip = start;
                 let _ = std::fs::write(HISTORY_SKIP, start.to_string());
@@ -292,7 +217,9 @@ impl Agent {
                 Ok(entries) => entries,
                 Err(e) => {
                     if is_too_large(&e) {
-                        if let Some(next) = next_user_start(&self.history, start, current_start) {
+                        if let Some(next) =
+                            next_turn_start(&self.history, start.saturating_add(1), current_start)
+                        {
                             self.history_skip = next;
                             let _ = std::fs::write(HISTORY_SKIP, next.to_string());
                             continue;
@@ -304,9 +231,6 @@ impl Agent {
                         phase: None,
                     });
                     let _ = publish(json!({"type": "text", "text": text}));
-                    if self.inbox.is_closed() {
-                        self.shutdown = true;
-                    }
                     return;
                 }
             };
@@ -329,16 +253,14 @@ impl Agent {
                 }
                 self.add_history(entry);
             }
-            if self.inbox.is_closed() {
+            if self.closing() {
-                self.shutdown = true;
                 return;
             }
             if calls.is_empty() {
                 return;
             }
             for (id, name, args) in calls {
-                if self.shutdown || self.inbox.is_closed() {
+                if self.closing() {
-                    self.shutdown = true;
                     return;
                 }
                 let output = self.tool(&name, &args).await;
@@ -346,10 +268,6 @@ impl Agent {
                     id,
                     output: output.to_string(),
                 });
-                if self.shutdown || self.inbox.is_closed() {
-                    self.shutdown = true;
-                    return;
-                }
             }
         }
     }
@@ -367,71 +285,17 @@ impl Agent {
         while let Some(message) = self.inbox.try_recv() {
             self.stash(message, others);
         }
-        if self.inbox.is_closed() {
-            self.shutdown = true;
-        }
     }
 
     async fn tool(&mut self, name: &str, args: &str) -> Value {
-        let a: Value = match serde_json::from_str(args) {
+        let args: Value = match serde_json::from_str(args) {
-            Ok(value) => value,
+            Ok(args) => args,
-            Err(e) => return json!({"error": format!("arguments are not valid JSON: {e}")}),
+            Err(error) => return json!({"error": format!("arguments are not valid JSON: {error}")}),
         };
-        let path = |key: &str| a[key].as_str().unwrap_or_default().to_string();
+        if name == "send_message" {
-        let err = |e: std::io::Error| json!({"error": e.to_string()});
+            return self.send_message(&args).await;
-        match name {
-            "send_message" => self.send_message(&a).await,
-            "read_file" => {
-                let p = path("path");
-                match std::fs::read(&p) {
-                    Ok(bytes) => {
-                        let offset = (a["offset"].as_u64().unwrap_or(0) as usize).min(bytes.len());
-                        let end = a["max_bytes"].as_u64().map_or(bytes.len(), |max| {
-                            offset.saturating_add(max as usize).min(bytes.len())
-                        });
-                        json!({
-                            "path": p, "size": bytes.len(), "offset": offset,
-                            "content": String::from_utf8_lossy(&bytes[offset..end]),
-                            "truncated": end < bytes.len(),
-                        })
-                    }
-                    Err(e) => err(e),
-                }
-            }
-            "list_dir" => match std::fs::read_dir(path("path")) {
-                Ok(rd) => {
-                    let mut entries: Vec<Value> = rd
-                        .filter_map(|entry| entry.ok())
-                        .map(|entry| {
-                            let meta = entry.metadata().ok();
-                            json!({
-                                "name": entry.file_name().to_string_lossy(),
-                                "type": if meta.as_ref().is_some_and(|m| m.is_dir()) { "dir" } else { "file" },
-                                "size": meta.map(|m| m.len()),
-                            })
-                        })
-                        .collect();
-                    entries.sort_by(|x, y| x["name"].as_str().cmp(&y["name"].as_str()));
-                    json!({"entries": entries})
-                }
-                Err(e) => err(e),
-            },
-            "write_file" => {
-                let p = path("path");
-                if let Some(parent) = std::path::Path::new(&p).parent() {
-                    let _ = std::fs::create_dir_all(parent);
-                }
-                match std::fs::write(&p, a["content"].as_str().unwrap_or_default()) {
-                    Ok(()) => json!({"written": p}),
-                    Err(e) => err(e),
-                }
-            }
-            "copy_dir" => match copy_tree(&path("from"), &path("to")) {
-                Ok(count) => json!({"copied_files": count}),
-                Err(e) => err(e),
-            },
-            other => json!({"error": format!("unknown tool {other}")}),
         }
+        tools::run(name, &args).unwrap_or_else(|error| json!({"error": error.to_string()}))
     }
 
     async fn send_message(&mut self, args: &Value) -> Value {
@@ -454,10 +318,7 @@ impl Agent {
                         break;
                     }
                     Ok(Some(message)) => self.stash(message, &mut others),
-                    Ok(None) => {
+                    Ok(None) => break,
-                        self.shutdown = true;
-                        break;
-                    }
                     Err(_) => break,
                 }
             }
@@ -476,93 +337,3 @@ fn load_config() -> Result<Config, String> {
 fn starts_turn(message: &Message) -> bool {
     message.kind() == "text" && message.reply_to.is_none()
 }
-
-fn is_too_large(error: &str) -> bool {
-    let Some((status, body)) = error
-        .strip_prefix("HTTP ")
-        .and_then(|error| error.split_once(':'))
-    else {
-        return false;
-    };
-    let status = status.split_whitespace().next().unwrap_or("");
-    if !matches!(status, "400" | "413") {
-        return false;
-    }
-    let body = body.to_ascii_lowercase();
-    body.contains("context_length_exceeded")
-        || body.contains("maximum context length")
-        || body.contains("context length exceeded")
-        || body.contains("too large")
-        || (body.contains("too long")
-            && ["context", "input", "prompt"]
-                .iter()
-                .any(|term| body.contains(term)))
-}
-
-fn history_start(history: &[Entry], skip: usize, current: usize) -> usize {
-    let Some(last) = history.len().checked_sub(1) else {
-        return 0;
-    };
-    let current = current.min(last);
-    (skip..=current)
-        .find(|&index| matches!(&history[index], Entry::User { .. }))
-        .unwrap_or(current)
-}
-
-fn next_user_start(history: &[Entry], start: usize, current: usize) -> Option<usize> {
-    let end = current.min(history.len().checked_sub(1)?);
-    (start.saturating_add(1)..=end).find(|&index| matches!(&history[index], Entry::User { .. }))
-}
-
-#[cfg(test)]
-mod tests {
-    use super::*;
-
-    #[test]
-    fn detects_context_size_errors_only_for_http_400_or_413() {
-        assert!(is_too_large("HTTP 400: context_length_exceeded"));
-        assert!(is_too_large("HTTP 413: prompt is too long"));
-        assert!(is_too_large("HTTP 400: maximum context length"));
-        assert!(is_too_large("HTTP 400: request is too large"));
-        assert!(!is_too_large("HTTP 500: prompt is too long"));
-        assert!(!is_too_large("HTTP 400: invalid token"));
-    }
-
-    #[test]
-    fn history_skip_advances_by_whole_user_turns() {
-        let history = vec![
-            Entry::User {
-                text: "first".into(),
-            },
-            Entry::ToolCall {
-                id: "a".into(),
-                name: "tool".into(),
-                arguments: "{}".into(),
-            },
-            Entry::ToolResult {
-                id: "a".into(),
-                output: "done".into(),
-            },
-            Entry::User {
-                text: "second".into(),
-            },
-            Entry::ToolCall {
-                id: "b".into(),
-                name: "tool".into(),
-                arguments: "{}".into(),
-            },
-            Entry::ToolResult {
-                id: "b".into(),
-                output: "done".into(),
-            },
-            Entry::User {
-                text: "current".into(),
-            },
-        ];
-        assert_eq!(next_user_start(&history, 0, 6), Some(3));
-        assert_eq!(history_start(&history, 1, 6), 3);
-        assert_eq!(next_user_start(&history, 3, 6), Some(6));
-        assert_eq!(history_start(&history, usize::MAX, 6), 6);
-        assert_eq!(next_user_start(&history, 6, 6), None);
-    }
-}

agent/src/tools.rs

@@ -0,0 +1,138 @@
+//! Tool descriptions and file operations for the agent.
+
+use crate::llm::Tool;
+use sausage_actor::{Value, json};
+use std::io;
+
+pub fn definitions() -> Vec<Tool> {
+    vec![
+        Tool {
+            name: "send_message",
+            description: "Send an actor a JSON body with a `type`. `broker` and `cargo` are built-in actors; see `/actor/broker/public/PROTOCOL.md`. With `wait_seconds`, the reply's `reply_to` matches the returned ID. Returns the ID, reply, and other messages received.",
+            parameters: schema(
+                json!({
+                    "to": {"type": "string", "description": "Actor ID"},
+                    "body": {"type": "object", "description": "JSON object with a type field"},
+                    "wait_seconds": {"type": "number", "description": "Seconds to wait for a reply. Default 0."}
+                }),
+                &["to", "body"],
+            ),
+        },
+        Tool {
+            name: "read_file",
+            description: "Read a text file under `/data`, `/public`, or `/actor/<id>`. Reads all bytes by default; `offset` and `max_bytes` are optional.",
+            parameters: schema(
+                json!({
+                    "path": {"type": "string"},
+                    "offset": {"type": "integer", "description": "Byte offset (default 0)"},
+                    "max_bytes": {"type": "integer", "description": "Optional maximum number of bytes to read"}
+                }),
+                &["path"],
+            ),
+        },
+        Tool {
+            name: "write_file",
+            description: "Write text under `/data` or `/public`. Parent folders are created.",
+            parameters: schema(
+                json!({"path": {"type": "string"}, "content": {"type": "string"}}),
+                &["path", "content"],
+            ),
+        },
+        Tool {
+            name: "list_dir",
+            description: "List entries in a directory under `/data`, `/public`, or `/actor/<id>`. Returns each name, type, and size.",
+            parameters: schema(json!({"path": {"type": "string"}}), &["path"]),
+        },
+        Tool {
+            name: "copy_dir",
+            description: "Copy a directory tree, for example `/actor/<id>/source` to `/public/drafts/<id>`. Skips `target/` folders.",
+            parameters: schema(
+                json!({"from": {"type": "string"}, "to": {"type": "string"}}),
+                &["from", "to"],
+            ),
+        },
+    ]
+}
+
+fn schema(properties: Value, required: &[&str]) -> Value {
+    json!({"type": "object", "properties": properties, "required": required})
+}
+
+pub fn run(name: &str, args: &Value) -> io::Result<Value> {
+    match name {
+        "read_file" => read_file(args),
+        "write_file" => write_file(args),
+        "list_dir" => list_dir(args),
+        "copy_dir" => copy_dir(args),
+        other => Ok(json!({"error": format!("unknown tool {other}")})),
+    }
+}
+
+fn read_file(args: &Value) -> io::Result<Value> {
+    let path = args["path"].as_str().unwrap_or_default();
+    let bytes = std::fs::read(path)?;
+    let offset = (args["offset"].as_u64().unwrap_or(0) as usize).min(bytes.len());
+    let end = args["max_bytes"].as_u64().map_or(bytes.len(), |max| {
+        offset.saturating_add(max as usize).min(bytes.len())
+    });
+    Ok(json!({
+        "path": path,
+        "size": bytes.len(),
+        "offset": offset,
+        "content": String::from_utf8_lossy(&bytes[offset..end]),
+        "truncated": end < bytes.len(),
+    }))
+}
+
+fn write_file(args: &Value) -> io::Result<Value> {
+    let path = args["path"].as_str().unwrap_or_default();
+    if let Some(parent) = std::path::Path::new(path).parent() {
+        let _ = std::fs::create_dir_all(parent);
+    }
+    std::fs::write(path, args["content"].as_str().unwrap_or_default())?;
+    Ok(json!({"written": path}))
+}
+
+fn list_dir(args: &Value) -> io::Result<Value> {
+    let mut entries: Vec<Value> = std::fs::read_dir(args["path"].as_str().unwrap_or_default())?
+        .filter_map(|entry| entry.ok())
+        .map(|entry| {
+            let metadata = entry.metadata().ok();
+            json!({
+                "name": entry.file_name().to_string_lossy(),
+                "type": if metadata.as_ref().is_some_and(|meta| meta.is_dir()) { "dir" } else { "file" },
+                "size": metadata.map(|meta| meta.len()),
+            })
+        })
+        .collect();
+    entries.sort_by(|left, right| left["name"].as_str().cmp(&right["name"].as_str()));
+    Ok(json!({"entries": entries}))
+}
+
+fn copy_dir(args: &Value) -> io::Result<Value> {
+    let count = copy_tree(
+        args["from"].as_str().unwrap_or_default(),
+        args["to"].as_str().unwrap_or_default(),
+    )?;
+    Ok(json!({"copied_files": count}))
+}
+
+fn copy_tree(from: &str, to: &str) -> io::Result<u64> {
+    std::fs::create_dir_all(to)?;
+    let mut count = 0;
+    for entry in std::fs::read_dir(from)? {
+        let entry = entry?;
+        let name = entry.file_name().to_string_lossy().to_string();
+        let src = format!("{from}/{name}");
+        let dst = format!("{to}/{name}");
+        if entry.file_type()?.is_dir() {
+            if name != "target" {
+                count += copy_tree(&src, &dst)?;
+            }
+        } else {
+            std::fs::copy(src, dst)?;
+            count += 1;
+        }
+    }
+    Ok(count)
+}

sdk/src/lib.rs

@@ -26,6 +26,14 @@ impl Message {
     }
 }
 
+fn parse(line: &[u8]) -> Option<Message> {
+    serde_json::from_slice::<Message>(line).ok()
+}
+
+fn is_shutdown(message: &Message) -> bool {
+    message.from == "broker" && message.kind() == "shutdown"
+}
+
 /// Return this actor's ID from `SAUSAGE_ACTOR`, or an empty string if unset.
 pub fn me() -> String {
     std::env::var("SAUSAGE_ACTOR").unwrap_or_default()
@@ -39,17 +47,13 @@ pub fn messages() -> impl Iterator<Item = Message> {
     let mut lines = stdin.lock().lines();
     std::iter::from_fn(move || {
         loop {
-            let line = match lines.next()? {
+            let Ok(line) = lines.next()? else {
-                Ok(line) => line,
-                Err(_) => continue,
-            };
-            let Ok(message) = serde_json::from_str::<Message>(&line) else {
                 continue;
             };
-            if message.from == "broker" && message.kind() == "shutdown" {
+            let Some(message) = parse(line.as_bytes()) else {
-                return None;
+                continue;
-            }
+            };
-            return Some(message);
+            return (!is_shutdown(&message)).then_some(message);
         }
     })
 }
@@ -153,11 +157,12 @@ async fn read_messages(tx: async_channel::Sender<Message>) {
         }
         buf.extend_from_slice(&chunk[..n]);
         while let Some(pos) = buf.iter().position(|b| *b == b'\n') {
-            let line: Vec<_> = buf.drain(..=pos).collect();
+            let message = parse(&buf[..pos]);
-            let Ok(message) = serde_json::from_slice::<Message>(&line) else {
+            buf.drain(..=pos);
+            let Some(message) = message else {
                 continue;
             };
-            if message.from == "broker" && message.kind() == "shutdown" {
+            if is_shutdown(&message) {
                 tx.close();
                 return;
             }
@@ -166,13 +171,6 @@ async fn read_messages(tx: async_channel::Sender<Message>) {
             }
         }
     }
-    if let Ok(message) = serde_json::from_slice::<Message>(&buf) {
-        if message.from == "broker" && message.kind() == "shutdown" {
-            tx.close();
-            return;
-        }
-        let _ = tx.send(message).await;
-    }
     tx.close();
 }