From c636eb08038d57554848d66ac22947d7d42c1693 Mon Sep 17 00:00:00 2001 From: Surma Date: Thu, 15 Oct 2026 14:22:51 +0100 Subject: [PATCH] Agent: one turn path, tools and history in their own modules; SDK: one parser --- agent/src/history.rs | 81 ++++++++++ agent/src/main.rs | 349 ++++++++----------------------------------- agent/src/tools.rs | 138 +++++++++++++++++ sdk/src/lib.rs | 36 +++-- 4 files changed, 296 insertions(+), 308 deletions(-) create mode 100644 agent/src/history.rs create mode 100644 agent/src/tools.rs diff --git a/agent/src/history.rs b/agent/src/history.rs new file mode 100644 index 0000000..5ee53a5 --- /dev/null +++ b/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 { + 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); + } +} diff --git a/agent/src/main.rs b/agent/src/main.rs index a3dc664..102def1 100644 --- a/agent/src/main.rs +++ b/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 { - 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/`. 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/`. 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//source` to `/public/drafts/`. 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 { - 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, @@ -100,7 +29,6 @@ struct Agent { active: bool, queue: VecDeque, notes: Vec, - 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 = self .history @@ -207,41 +138,50 @@ impl Agent { async fn run(mut self) { self.repair_dangling_tools(); - self.resume().await; - while !self.shutdown { + if self.active { + 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 => { - self.shutdown = true; - break; - } + None => 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) { - if !self.active { - return; - } - let current_start = self - .history - .iter() - .rposition(|entry| matches!(entry, Entry::User { text } if !text.starts_with("[note from your runtime]"))) - .unwrap_or(self.history.len()); - let version = std::env::var("SAUSAGE_VERSION").unwrap_or_default(); - self.add_history(Entry::User { - text: format!( - "[note from your runtime] you restarted and now run version {version}. Continue the active turn." - ), - }); + async fn turn(&mut self, message: Option) { + let current_start = if message.is_some() { + self.active = true; + let _ = std::fs::write(TURN, "true"); + self.history.len() + } else { + let current_start = self + .history + .iter() + .rposition(|entry| matches!(entry, Entry::User { text } if !text.starts_with("[note from your runtime]"))) + .unwrap_or(self.history.len()); + let version = std::env::var("SAUSAGE_VERSION").unwrap_or_default(); + 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 !self.shutdown { - self.finish_turn(); + if let Some(message) = message { + let text = self.turn_text(&message); + 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() { - self.shutdown = true; + if self.closing() { 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() { - self.shutdown = true; + if self.closing() { return; } if calls.is_empty() { return; } for (id, name, args) in calls { - if self.shutdown || self.inbox.is_closed() { - self.shutdown = true; + if self.closing() { 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) { - Ok(value) => value, - Err(e) => return json!({"error": format!("arguments are not valid JSON: {e}")}), + let args: Value = match serde_json::from_str(args) { + Ok(args) => args, + 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(); - let err = |e: std::io::Error| json!({"error": e.to_string()}); - 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 = 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}")}), + if name == "send_message" { + return self.send_message(&args).await; } + 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) => { - self.shutdown = true; - break; - } + Ok(None) => break, Err(_) => break, } } @@ -476,93 +337,3 @@ fn load_config() -> Result { 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 { - 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); - } -} diff --git a/agent/src/tools.rs b/agent/src/tools.rs new file mode 100644 index 0000000..65e2671 --- /dev/null +++ b/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 { + 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/`. 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/`. 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//source` to `/public/drafts/`. 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 { + 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 { + 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 { + 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 { + let mut entries: Vec = 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 { + 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 { + 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) +} diff --git a/sdk/src/lib.rs b/sdk/src/lib.rs index 09ec822..844023b 100644 --- a/sdk/src/lib.rs +++ b/sdk/src/lib.rs @@ -26,6 +26,14 @@ impl Message { } } +fn parse(line: &[u8]) -> Option { + serde_json::from_slice::(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 { let mut lines = stdin.lock().lines(); std::iter::from_fn(move || { loop { - let line = match lines.next()? { - Ok(line) => line, - Err(_) => continue, - }; - let Ok(message) = serde_json::from_str::(&line) else { + let Ok(line) = lines.next()? else { continue; }; - if message.from == "broker" && message.kind() == "shutdown" { - return None; - } - return Some(message); + let Some(message) = parse(line.as_bytes()) else { + continue; + }; + return (!is_shutdown(&message)).then_some(message); } }) } @@ -153,11 +157,12 @@ async fn read_messages(tx: async_channel::Sender) { } 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 Ok(message) = serde_json::from_slice::(&line) else { + let message = parse(&buf[..pos]); + 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) { } } } - if let Ok(message) = serde_json::from_slice::(&buf) { - if message.from == "broker" && message.kind() == "shutdown" { - tx.close(); - return; - } - let _ = tx.send(message).await; - } tx.close(); }