Agent: durable turn and queue state, visible inbox closure, OpenAI phase; prompt fixes; fetch/bench fixes

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<String>,
+    },
     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() {
+                let phase = item["phase"].as_str().map(str::to_string);
-                    out.push(Entry::Assistant { text });
+                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 {

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<u64> {
     Ok(count)
 }
 
+#[derive(Default, Deserialize, Serialize)]
+#[serde(default)]
+struct TurnState {
+    active: Option<Message>,
+    queue: VecDeque<Message>,
+    notes: Vec<Message>,
+}
+
 struct Agent {
     inbox: Inbox,
     history: Vec<Entry>,
-    queue: VecDeque<Message>,
+    pending: TurnState,
-    notes: Vec<Message>,
     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::<TurnState>(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(),
+            pending,
-            notes: Vec::new(),
             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) {
+    fn persist_turn(&self) {
-        self.resume().await;
+        if self.pending.active.is_none()
-        while !self.shutdown {
+            && self.pending.queue.is_empty()
-            let msg = match self.queue.pop_front() {
+            && self.pending.notes.is_empty()
-                Some(msg) => msg,
+        {
-                None => match self.inbox.recv().await {
+            if let Err(e) = std::fs::remove_file(TURN) {
-                    Some(msg) => msg,
+                if e.kind() != std::io::ErrorKind::NotFound {
-                    None => break,
+                    eprintln!("agent: cannot remove turn state: {e}");
-                },
-            };
-            if starts_turn(&msg) {
-                self.turn(msg).await;
-            } else {
-                self.notes.push(msg);
                 }
             }
+        } 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) {
+    fn finish_turn(&mut self) {
-        let Ok(turn) = std::fs::read_to_string(TURN) else {
+        self.pending.active = None;
-            return;
+        self.persist_turn();
-        };
+    }
-        let Ok(turn) = serde_json::from_str::<Value>(&turn) else {
-            return;
-        };
-        let Ok(contents) = std::fs::read_to_string(CONFIG) else {
-            return;
-        };
-        let Ok(cfg) = serde_json::from_str::<Config>(&contents) else {
-            return;
-        };
-        let _ = std::fs::remove_file(TURN);
 
+    fn repair_dangling_tools(&mut self) {
         let answered: HashSet<String> = 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) {
+    fn turn_text(&self, message: &Message) -> String {
-        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;
-            }
-        };
         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(..)
+                .iter()
-                .map(|note| serde_json::to_string(&note).unwrap())
+                .map(|note| serde_json::to_string(note).unwrap())
                 .collect::<Vec<_>>()
                 .join("\n");
             text = format!("Messages that arrived while you were idle:\n{notes}\n\n{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<String> = 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,31 +387,61 @@ 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.inbox.is_closed() {
+                    self.preserve_closed_messages();
+                    return;
+                }
+            }
         }
     }
 
     fn stash(&mut self, message: Message, others: &mut Vec<Value>) {
         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<Value>) {
+        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)) => {
+                    Ok(Some(message)) => self.stash(message, &mut others),
-                        self.stash(message, &mut others);
-                        while let Some(message) = self.inbox.try_recv() {
-                            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)) => {
+                    Ok(Some(message)) => self.stash(message, &mut others),
-                        self.stash(message, &mut others);
-                        if self.shutdown {
-                            break;
-                        }
-                    }
                     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<Config, String> {
+    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()
 }

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 mut body = response.into_body();
-            let _ = reply(&message, json!({"type": "fetched", "text": text.to_string()}));
+            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/<name>"}` 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.

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 {

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"}
+    }
+  ]
+}

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<u64, String> {
+    let previous = match std::fs::read_to_string("/data/fetch-sequence") {
+        Ok(value) => value.parse::<u64>().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<u64, String> {
+    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" => {
+                "fetch" => match next_fetch_id() {
-                    n += 1;
+                    Ok(n) => wstd::runtime::spawn(fetch(n, message)).detach(),
-                    wstd::runtime::spawn(fetch(n, message)).detach();
+                    Err(reason) => {
+                        let _ = reply(&message, json!({"type": "error", "reason": reason}));
+                    }
                 }
                 "ping" => {
                     let _ = reply(&message, json!({"type": "pong"}));

sdk/src/lib.rs

@@ -108,15 +108,17 @@ fn next_id() -> String {
 #[cfg(feature = "async")]
 pub struct Inbox {
     rx: async_channel::Receiver<Message>,
+    _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();
+        let reader = wstd::runtime::spawn(read_messages(tx));
-        Self { rx }
+        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<Message> {
         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")]