From 69cf80a114f49b5d611a079c8dbd57d204d1d76b Mon Sep 17 00:00:00 2001 From: Surma Date: Thu, 15 Oct 2026 11:54:41 +0100 Subject: [PATCH] Agent: durable turn and queue state, visible inbox closure, OpenAI phase; prompt fixes; fetch/bench fixes --- agent/src/llm.rs | 25 ++- agent/src/main.rs | 290 +++++++++++++++++++++++---------- agent/src/prompt.md | 23 ++- examples/bench/src/main.rs | 4 + examples/bench/tests/zero.json | 9 + examples/fetch/src/main.rs | 35 +++- sdk/src/lib.rs | 13 +- 7 files changed, 296 insertions(+), 103 deletions(-) create mode 100644 examples/bench/tests/zero.json diff --git a/agent/src/llm.rs b/agent/src/llm.rs index f22d839..64e976c 100644 --- a/agent/src/llm.rs +++ b/agent/src/llm.rs @@ -25,7 +25,11 @@ pub struct Config { #[serde(tag = "kind", rename_all = "snake_case")] pub enum Entry { User { text: String }, - Assistant { text: String }, + Assistant { + text: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + phase: Option, + }, ToolCall { id: String, name: String, arguments: String }, ToolResult { id: String, output: String }, /// A provider item that only this API shape understands (OpenAI reasoning). @@ -68,7 +72,13 @@ async fn openai(cfg: &Config, system: &str, tools: &[Tool], history: &[Entry]) - for e in history { match e { Entry::User { text } => input.push(json!({"role": "user", "content": text})), - Entry::Assistant { text } => input.push(json!({"role": "assistant", "content": text})), + Entry::Assistant { text, phase } => { + let mut item = json!({"role": "assistant", "content": text}); + if let Some(phase) = phase { + item["phase"] = json!(phase); + } + input.push(item); + } Entry::ToolCall { id, name, arguments } => { input.push(json!({"type": "function_call", "call_id": id, "name": name, "arguments": arguments})) } @@ -113,8 +123,9 @@ async fn openai(cfg: &Config, system: &str, tools: &[Tool], history: &[Entry]) - .collect() }) .unwrap_or_default(); - if !text.is_empty() { - out.push(Entry::Assistant { text }); + let phase = item["phase"].as_str().map(str::to_string); + if !text.is_empty() || phase.is_some() { + out.push(Entry::Assistant { text, phase }); } } Some("function_call") => out.push(Entry::ToolCall { @@ -142,7 +153,9 @@ async fn anthropic(cfg: &Config, system: &str, tools: &[Tool], history: &[Entry] for e in history { match e { Entry::User { text } => push("user", json!({"type": "text", "text": text})), - Entry::Assistant { text } => push("assistant", json!({"type": "text", "text": text})), + Entry::Assistant { text, .. } => { + push("assistant", json!({"type": "text", "text": text})) + } Entry::ToolCall { id, name, arguments } => { let input: Value = serde_json::from_str(arguments).unwrap_or(json!({})); push("assistant", json!({"type": "tool_use", "id": id, "name": name, "input": input})) @@ -178,7 +191,7 @@ async fn anthropic(cfg: &Config, system: &str, tools: &[Tool], history: &[Entry] Some("text") => { let text = block["text"].as_str().unwrap_or_default().to_string(); if !text.is_empty() { - out.push(Entry::Assistant { text }); + out.push(Entry::Assistant { text, phase: None }); } } Some("tool_use") => out.push(Entry::ToolCall { diff --git a/agent/src/main.rs b/agent/src/main.rs index a109329..a94cb6e 100644 --- a/agent/src/main.rs +++ b/agent/src/main.rs @@ -4,6 +4,7 @@ mod llm; use llm::{Config, Entry, Tool}; use sausage_actor::{Inbox, Message, Value, json, reply, send}; +use serde::{Deserialize, Serialize}; use std::collections::{HashSet, VecDeque}; use std::io::Write; use wstd::future::FutureExt; @@ -106,11 +107,18 @@ fn copy_tree(from: &str, to: &str) -> std::io::Result { Ok(count) } +#[derive(Default, Deserialize, Serialize)] +#[serde(default)] +struct TurnState { + active: Option, + queue: VecDeque, + notes: Vec, +} + struct Agent { inbox: Inbox, history: Vec, - queue: VecDeque, - notes: Vec, + pending: TurnState, shutdown: bool, } @@ -121,13 +129,32 @@ impl Agent { .lines() .filter_map(|line| serde_json::from_str(line).ok()) .collect(); - Self { + let pending = std::fs::read_to_string(TURN) + .ok() + .and_then(|contents| { + let value: Value = serde_json::from_str(&contents).ok()?; + let mut pending = serde_json::from_value::(value.clone()).ok()?; + if pending.active.is_none() { + if let (Some(id), Some(from)) = (value["id"].as_str(), value["from"].as_str()) { + pending.active = Some(Message { + id: id.to_string(), + from: from.to_string(), + reply_to: None, + body: Value::Null, + }); + } + } + Some(pending) + }) + .unwrap_or_default(); + let agent = Self { inbox: Inbox::start(), history, - queue: VecDeque::new(), - notes: Vec::new(), + pending, shutdown: false, - } + }; + agent.persist_turn(); + agent } fn add_history(&mut self, entry: Entry) { @@ -147,39 +174,27 @@ impl Agent { self.history.push(entry); } - async fn run(mut self) { - self.resume().await; - while !self.shutdown { - let msg = match self.queue.pop_front() { - Some(msg) => msg, - None => match self.inbox.recv().await { - Some(msg) => msg, - None => break, - }, - }; - if starts_turn(&msg) { - self.turn(msg).await; - } else { - self.notes.push(msg); + fn persist_turn(&self) { + if self.pending.active.is_none() + && self.pending.queue.is_empty() + && self.pending.notes.is_empty() + { + if let Err(e) = std::fs::remove_file(TURN) { + if e.kind() != std::io::ErrorKind::NotFound { + eprintln!("agent: cannot remove turn state: {e}"); + } } + } else if let Err(e) = std::fs::write(TURN, serde_json::to_vec(&self.pending).unwrap()) { + eprintln!("agent: cannot save turn state: {e}"); } } - async fn resume(&mut self) { - let Ok(turn) = std::fs::read_to_string(TURN) else { - return; - }; - let Ok(turn) = serde_json::from_str::(&turn) else { - return; - }; - let Ok(contents) = std::fs::read_to_string(CONFIG) else { - return; - }; - let Ok(cfg) = serde_json::from_str::(&contents) else { - return; - }; - let _ = std::fs::remove_file(TURN); + fn finish_turn(&mut self) { + self.pending.active = None; + self.persist_turn(); + } + fn repair_dangling_tools(&mut self) { let answered: HashSet = self .history .iter() @@ -204,62 +219,135 @@ impl Agent { .to_string(), }); } - 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}. The restart interrupted your last turn (probably your own upgrade). Check the state you need, then finish the turn and answer the original message." - ), - }); - let message = Message { - id: turn["id"].as_str().unwrap_or_default().to_string(), - from: turn["from"].as_str().unwrap_or_default().to_string(), - reply_to: None, - body: Value::Null, - }; - self.steps(&cfg, &message).await; } - async fn turn(&mut self, message: Message) { - let cfg = match std::fs::read_to_string(CONFIG) - .map_err(|e| e.to_string()) - .and_then(|contents| serde_json::from_str(&contents).map_err(|e| e.to_string())) - { - Ok(cfg) => cfg, - Err(e) => { - let _ = reply( - &message, - json!({"type": "text", "text": format!("I am not configured: cannot read {CONFIG}: {e}")}), - ); - return; - } - }; + fn turn_text(&self, message: &Message) -> String { let mut text = format!( "Message {} from {}:\n{}", message.id, message.from, message.body["text"].as_str().unwrap_or("") ); - if !self.notes.is_empty() { + if !self.pending.notes.is_empty() { let notes = self + .pending .notes - .drain(..) - .map(|note| serde_json::to_string(¬e).unwrap()) + .iter() + .map(|note| serde_json::to_string(note).unwrap()) .collect::>() .join("\n"); text = format!("Messages that arrived while you were idle:\n{notes}\n\n{text}"); } - self.add_history(Entry::User { text }); + text + } + + fn has_turn_user(&self, message: &Message) -> bool { + let marker = format!("Message {} from {}:\n", message.id, message.from); + self.history.iter().any(|entry| match entry { + Entry::User { text } => text.contains(&marker), + _ => false, + }) + } + + fn not_configured(&mut self, message: &Message, error: String) { + let _ = reply( + message, + json!({"type": "text", "text": format!("I am not configured: cannot read {CONFIG}: {error}")}), + ); + self.finish_turn(); + } + + async fn run(mut self) { + self.repair_dangling_tools(); + self.resume().await; + while !self.shutdown { + let message = match self.pending.queue.pop_front() { + Some(message) => message, + None => match self.inbox.recv().await { + Some(message) => message, + None => { + self.shutdown = true; + break; + } + }, + }; + if starts_turn(&message) { + self.turn(message).await; + } else { + self.pending.notes.push(message); + self.persist_turn(); + } + } + self.persist_turn(); + } + + async fn resume(&mut self) { + let Some(message) = self.pending.active.clone() else { + return; + }; + if !self.has_turn_user(&message) { + let text = self.turn_text(&message); + self.add_history(Entry::User { text }); + self.pending.notes.clear(); + self.persist_turn(); + } + 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}. The restart interrupted your last turn (probably your own upgrade). Check the state you need, then finish the turn and answer the original message." + ), + }); + let cfg = match load_config() { + Ok(cfg) => cfg, + Err(error) => { + self.not_configured(&message, error); + return; + } + }; self.steps(&cfg, &message).await; + if !self.shutdown { + self.finish_turn(); + } + } + + async fn turn(&mut self, message: Message) { + self.pending.active = Some(message.clone()); + self.persist_turn(); + let cfg = match load_config() { + Ok(cfg) => cfg, + Err(error) => { + self.not_configured(&message, error); + return; + } + }; + let text = self.turn_text(&message); + self.add_history(Entry::User { text }); + self.pending.notes.clear(); + self.persist_turn(); + self.steps(&cfg, &message).await; + if !self.shutdown { + self.finish_turn(); + } } async fn steps(&mut self, cfg: &Config, message: &Message) { let tools = tools(); loop { - let entries = match llm::complete(cfg, SYSTEM, &tools, &self.history).await { + if self.inbox.is_closed() { + self.preserve_closed_messages(); + return; + } + let response = llm::complete(cfg, SYSTEM, &tools, &self.history).await; + if self.inbox.is_closed() { + self.preserve_closed_messages(); + return; + } + let entries = match response { Ok(entries) => entries, Err(e) => { self.add_history(Entry::Assistant { text: format!("(inference failed: {e})"), + phase: None, }); let _ = reply( message, @@ -282,7 +370,7 @@ impl Agent { let answer: Vec = entries .iter() .filter_map(|entry| match entry { - Entry::Assistant { text } => Some(text.clone()), + Entry::Assistant { text, .. } if !text.is_empty() => Some(text.clone()), _ => None, }) .collect(); @@ -299,18 +387,24 @@ impl Agent { return; } for (id, name, args) in calls { + if self.shutdown || self.inbox.is_closed() { + if !self.shutdown { + self.preserve_closed_messages(); + } + return; + } let output = self.tool(&name, &args).await; self.add_history(Entry::ToolResult { id, output: output.to_string(), }); - } - if self.shutdown { - let _ = std::fs::write( - TURN, - json!({"from": message.from, "id": message.id}).to_string(), - ); - return; + if self.shutdown { + return; + } + if self.inbox.is_closed() { + self.preserve_closed_messages(); + return; + } } } } @@ -318,12 +412,36 @@ impl Agent { fn stash(&mut self, message: Message, others: &mut Vec) { if starts_turn(&message) { others.push(json!({"queued_for_a_later_turn": true, "message": message})); - self.queue.push_back(message); + self.pending.queue.push_back(message); + self.persist_turn(); } else { others.push(serde_json::to_value(message).unwrap()); } } + fn drain_inbox(&mut self, others: &mut Vec) { + while let Some(message) = self.inbox.try_recv() { + self.stash(message, others); + } + if self.inbox.is_closed() { + self.shutdown = true; + } + self.persist_turn(); + } + + fn preserve_closed_messages(&mut self) { + while let Some(message) = self.inbox.try_recv() { + if starts_turn(&message) { + self.pending.queue.push_back(message); + } else { + self.pending.notes.push(message); + } + self.persist_turn(); + } + self.shutdown = self.inbox.is_closed(); + self.persist_turn(); + } + async fn tool(&mut self, name: &str, args: &str) -> Value { let a: Value = match serde_json::from_str(args) { Ok(value) => value, @@ -342,15 +460,11 @@ impl Agent { .timeout(Duration::from_secs_f64(seconds)) .await { - Ok(Some(message)) => { - self.stash(message, &mut others); - while let Some(message) = self.inbox.try_recv() { - self.stash(message, &mut others); - } - } + Ok(Some(message)) => self.stash(message, &mut others), Ok(None) => self.shutdown = true, Err(_) => {} } + self.drain_inbox(&mut others); json!({"messages": others}) } "read_file" => { @@ -437,12 +551,7 @@ impl Agent { reply = serde_json::to_value(message).unwrap(); break; } - Ok(Some(message)) => { - self.stash(message, &mut others); - if self.shutdown { - break; - } - } + Ok(Some(message)) => self.stash(message, &mut others), Ok(None) => { self.shutdown = true; break; @@ -451,10 +560,17 @@ impl Agent { } } } + self.drain_inbox(&mut others); json!({"id": id, "reply": reply, "waited": wait > 0.0, "other_messages": others}) } } +fn load_config() -> Result { + std::fs::read_to_string(CONFIG) + .map_err(|e| e.to_string()) + .and_then(|contents| serde_json::from_str(&contents).map_err(|e| e.to_string())) +} + fn starts_turn(message: &Message) -> bool { message.kind() == "text" && message.reply_to.is_none() } diff --git a/agent/src/prompt.md b/agent/src/prompt.md index f640741..a54cd4e 100644 --- a/agent/src/prompt.md +++ b/agent/src/prompt.md @@ -22,7 +22,15 @@ fn main() { } ``` -Use `Inbox` for an actor that needs HTTP or timers. Enable the SDK's `async` feature. Use `wstd` for HTTP: +Use `Inbox` for an actor that needs HTTP or timers. Add these dependencies to `Cargo.toml`: + +```toml +[dependencies] +sausage-actor = { version = "0.1", features = ["async"] } +wstd = "0.6" +``` + +Use `wstd` for HTTP: ```rust use sausage_actor::{Inbox, json, reply}; @@ -38,8 +46,9 @@ fn main() { let url = message.body["url"].as_str().unwrap(); let request = Request::get(url).body(Body::empty()).unwrap(); let response = Client::new().send(request).await.unwrap(); - let text = response.into_body().str_contents().await.unwrap(); - let _ = reply(&message, json!({"type": "fetched", "text": text.to_string()})); + let mut body = response.into_body(); + let text = body.str_contents().await.unwrap().to_string(); + let _ = reply(&message, json!({"type": "fetched", "text": text})); } }); } @@ -53,6 +62,14 @@ The broker tests each `tests/*.json` file before an upgrade or spawn. Each file Send `{"type":"build","source":"/actor/agent/public/drafts/"}` to `cargo`. Then send `spawn` or `upgrade` to `broker` with the returned WASM path and the source folder. Check the actor with a message. +Send broker requests with `body.type` set to the operation name. The fields are: +- `spawn`, `upgrade`: `actor`, `wasm`, `source`, `built_by`. +- `rollback`: `actor`, optional `version`. +- `stop`, `start`, `remove`, `subscribe`, `unsubscribe`: `actor`. +- `scan`, `list`: no extra fields. + +The full contract is at `/actor/broker/public/PROTOCOL.md`. + ## Filesystem Your private state lives under `/data`. Your public files live under `/public`. Other actors' `data/` directories are hidden. diff --git a/examples/bench/src/main.rs b/examples/bench/src/main.rs index eafdeff..38a9c10 100644 --- a/examples/bench/src/main.rs +++ b/examples/bench/src/main.rs @@ -10,6 +10,10 @@ fn main() { } let target = message.body["target"].as_str().unwrap_or("echo"); let n = message.body["n"].as_u64().unwrap_or(1000); + if n == 0 { + let _ = reply(&message, json!({"type": "error", "reason": "n must be greater than 0"})); + continue; + } let mut times = Vec::new(); let started = Instant::now(); for i in 0..n { diff --git a/examples/bench/tests/zero.json b/examples/bench/tests/zero.json new file mode 100644 index 0000000..cbf7328 --- /dev/null +++ b/examples/bench/tests/zero.json @@ -0,0 +1,9 @@ +{ + "description": "reject zero benchmark samples", + "steps": [ + { + "send": {"type": "bench", "n": 0}, + "expect": {"type": "error", "reason": "n must be greater than 0"} + } + ] +} diff --git a/examples/fetch/src/main.rs b/examples/fetch/src/main.rs index 2ad4da7..5416586 100644 --- a/examples/fetch/src/main.rs +++ b/examples/fetch/src/main.rs @@ -2,6 +2,32 @@ use sausage_actor::{Inbox, Message, json, me, reply}; use wstd::http::{Body, Client, Request}; +fn next_fetch_id() -> Result { + let previous = match std::fs::read_to_string("/data/fetch-sequence") { + Ok(value) => value.parse::().map_err(|e| e.to_string())?, + Err(e) if e.kind() == std::io::ErrorKind::NotFound => existing_fetch_id()?, + Err(e) => return Err(e.to_string()), + }; + let next = previous + .checked_add(1) + .ok_or_else(|| "fetch ID space exhausted".to_string())?; + std::fs::write("/data/fetch-sequence", next.to_string()).map_err(|e| e.to_string())?; + Ok(next) +} + +fn existing_fetch_id() -> Result { + let entries = match std::fs::read_dir("/public/fetch") { + Ok(entries) => entries, + Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(0), + Err(e) => return Err(e.to_string()), + }; + Ok(entries + .filter_map(Result::ok) + .filter_map(|entry| entry.file_name().to_str()?.parse().ok()) + .max() + .unwrap_or(0)) +} + async fn fetch(n: u64, message: Message) { let url = message.body["url"].as_str().unwrap_or_default(); let started = wstd::time::Instant::now(); @@ -41,12 +67,13 @@ async fn fetch(n: u64, message: Message) { fn main() { wstd::runtime::block_on(async { let inbox = Inbox::start(); - let mut n = 0; while let Some(message) = inbox.recv().await { match message.kind() { - "fetch" => { - n += 1; - wstd::runtime::spawn(fetch(n, message)).detach(); + "fetch" => match next_fetch_id() { + Ok(n) => wstd::runtime::spawn(fetch(n, message)).detach(), + Err(reason) => { + let _ = reply(&message, json!({"type": "error", "reason": reason})); + } } "ping" => { let _ = reply(&message, json!({"type": "pong"})); diff --git a/sdk/src/lib.rs b/sdk/src/lib.rs index bca225e..09ec822 100644 --- a/sdk/src/lib.rs +++ b/sdk/src/lib.rs @@ -108,15 +108,17 @@ fn next_id() -> String { #[cfg(feature = "async")] pub struct Inbox { rx: async_channel::Receiver, + _reader: wstd::runtime::Task<()>, } #[cfg(feature = "async")] impl Inbox { - /// Start a task that reads stdin and returns an inbox for its messages. + /// Start the stdin reader inside the wstd runtime and return its inbox. + /// The inbox owns the reader task and must be dropped to cancel it. pub fn start() -> Self { let (tx, rx) = async_channel::unbounded(); - wstd::runtime::spawn(read_messages(tx)).detach(); - Self { rx } + let reader = wstd::runtime::spawn(read_messages(tx)); + Self { rx, _reader: reader } } /// Wait for the next message, or return `None` at EOF or shutdown. @@ -128,6 +130,11 @@ impl Inbox { pub fn try_recv(&self) -> Option { self.rx.try_recv().ok() } + + /// Return whether the reader closed the inbox, even when messages remain buffered. + pub fn is_closed(&self) -> bool { + self.rx.is_closed() + } } #[cfg(feature = "async")]