Spike: fs/upgrade/bench examples, agent resumes interrupted turns, world lock, short trap reasons

agent/src/main.rs

@@ -18,6 +18,8 @@ use wstd::time::{Duration, Instant};
 const SYSTEM: &str = include_str!("prompt.md");
 const CONFIG: &str = "/data/config.json";
 const HISTORY: &str = "/data/history.json";
+/// An interrupted turn (sender and message ID), resumed at the next start.
+const TURN: &str = "/data/turn.json";
 const MAX_STEPS: usize = 40;
 const MAX_HISTORY: usize = 600;
 const MAX_TOOL_OUTPUT: usize = 30_000;
@@ -195,6 +197,7 @@ impl Agent {
 
     async fn run(mut self) {
         eprintln!("agent: started with {} history entries", self.history.len());
+        self.resume().await;
         while !self.shutdown {
             let msg = match self.queue.pop_front() {
                 Some(m) => m,
@@ -219,6 +222,49 @@ impl Agent {
         eprintln!("agent: stopped");
     }
 
+    /// Continue a turn that a restart (usually an upgrade of the agent itself)
+    /// interrupted.
+    async fn resume(&mut self) {
+        let Some(turn) = std::fs::read_to_string(TURN).ok().and_then(|s| serde_json::from_str::<Value>(&s).ok()) else {
+            return;
+        };
+        let _ = std::fs::remove_file(TURN);
+        let Ok(cfg) = std::fs::read_to_string(CONFIG).map_err(|_| ()).and_then(|s| serde_json::from_str::<Config>(&s).map_err(|_| ())) else {
+            return;
+        };
+        // Every tool call needs a result, or the APIs reject the history.
+        let answered: std::collections::HashSet<String> = self
+            .history
+            .iter()
+            .filter_map(|e| match e {
+                Entry::ToolResult { id, .. } => Some(id.clone()),
+                _ => None,
+            })
+            .collect();
+        let dangling: Vec<String> = self
+            .history
+            .iter()
+            .filter_map(|e| match e {
+                Entry::ToolCall { id, .. } if !answered.contains(id) => Some(id.clone()),
+                _ => None,
+            })
+            .collect();
+        for id in dangling {
+            self.history.push(Entry::ToolResult {
+                id,
+                output: json!({"error": "interrupted: the agent restarted before this tool finished"}).to_string(),
+            });
+        }
+        let version = std::env::var("SAUSAGE_VERSION").unwrap_or_default();
+        self.history.push(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."
+            ),
+        });
+        eprintln!("agent: resuming an interrupted turn");
+        self.steps(&cfg, turn["from"].clone(), turn["id"].clone()).await;
+    }
+
     fn save(&mut self) {
         if self.history.len() > MAX_HISTORY {
             let start = self.history.len() - MAX_HISTORY;
@@ -268,14 +314,25 @@ impl Agent {
                 return;
             }
         };
-        let mut text = format!("Message {} from {}:\n{}", id, from.as_str().unwrap_or("?"), msg["body"]["text"].as_str().unwrap_or(""));
+        let mut text = format!(
+            "Message {} from {}:\n{}",
+            id.as_str().unwrap_or("?"),
+            from.as_str().unwrap_or("?"),
+            msg["body"]["text"].as_str().unwrap_or("")
+        );
         if !self.notes.is_empty() {
             let notes: Vec<String> = self.notes.drain(..).map(|n| n.to_string()).collect();
             text = format!("Messages that arrived while you were idle:\n{}\n\n{text}", notes.join("\n"));
         }
         self.history.push(Entry::User { text });
-        let tools = tools();
+        self.steps(&cfg, from, id).await;
+    }
 
+    /// Run inference and tools until the model answers. If a shutdown
+    /// interrupts the turn, remember it in TURN so the next start resumes it.
+    async fn steps(&mut self, cfg: &Config, from: Value, id: Value) {
+        let tools = tools();
+        let cfg = cfg.clone();
         for _ in 0..MAX_STEPS {
             let started = Instant::now();
             let result = llm::complete(&cfg, SYSTEM, &tools, &self.history).await;
@@ -317,6 +374,8 @@ impl Agent {
             }
             self.save();
             if self.shutdown {
+                let _ = std::fs::write(TURN, json!({"from": from, "id": id}).to_string());
+                eprintln!("agent: turn interrupted by shutdown; it resumes after the restart");
                 return;
             }
         }
@@ -335,11 +394,15 @@ impl Agent {
             "wait_for_messages" => {
                 let secs = a["seconds"].as_f64().unwrap_or(10.0).clamp(0.0, 900.0);
                 let mut others = Vec::new();
-                if let Ok(Ok(m)) = self.rx.recv().timeout(Duration::from_secs_f64(secs)).await {
-                    self.stash(m, &mut others);
-                    while let Ok(m) = self.rx.try_recv() {
+                match self.rx.recv().timeout(Duration::from_secs_f64(secs)).await {
+                    Ok(Ok(m)) => {
                         self.stash(m, &mut others);
+                        while let Ok(m) = self.rx.try_recv() {
+                            self.stash(m, &mut others);
+                        }
                     }
+                    Ok(Err(_)) => self.shutdown = true,
+                    Err(_) => {}
                 }
                 json!({"messages": others})
             }
@@ -430,7 +493,12 @@ impl Agent {
                             break;
                         }
                     }
-                    _ => break,
+                    Ok(Err(_)) => {
+                        // stdin closed: the broker stops us.
+                        self.shutdown = true;
+                        break;
+                    }
+                    Err(_) => break,
                 }
             }
         }

cli/src/broker.rs

@@ -628,8 +628,12 @@ impl Supervisor {
     async fn start_instance(&mut self) {
         if self.component.is_none() {
             let path = self.world().version_dir(&self.id, self.version).join("actor.wasm");
+            let t = Instant::now();
             match self.broker.runtime.load(&path) {
-                Ok(c) => self.component = Some(c),
+                Ok(c) => {
+                    self.broker.log(&self.id, &format!("compiled version {} in {} ms", self.version, t.elapsed().as_millis()));
+                    self.component = Some(c);
+                }
                 Err(e) => {
                     self.broker.log(&self.id, &format!("cannot load version {}: {e:#}", self.version));
                     self.state = "broken";

cli/src/main.rs

@@ -147,6 +147,11 @@ async fn run(dir: PathBuf) -> Result<()> {
     if !world.exists() {
         bail!("{} is not a Sausage world. Create one with: sausage init {}", dir.display(), dir.display());
     }
+    // One host process per world.
+    let lock = std::fs::File::create(world.root.join(".lock"))?;
+    if lock.try_lock().is_err() {
+        bail!("another sausage process runs {}", dir.display());
+    }
     let runtime = runtime::Runtime::new()?;
     let broker = Broker::new(world.clone(), runtime)?;
 
@@ -165,6 +170,7 @@ async fn run(dir: PathBuf) -> Result<()> {
     console.run_input().await;
     println!("Stopping actors …");
     broker.shutdown_all().await;
+    drop(lock);
     Ok(())
 }
 

cli/src/runtime.rs

@@ -176,6 +176,24 @@ impl Instance {
     }
 }
 
+fn log_note(path: &Path, text: &str) {
+    use std::io::Write;
+    if let Ok(mut f) = std::fs::OpenOptions::new().create(true).append(true).open(path) {
+        let _ = writeln!(f, "[sausage] {text}");
+    }
+}
+
+/// The most useful line of a trap: the `wasm trap: ...` reason.
+fn short_trap(full: &str) -> String {
+    let line = full
+        .lines()
+        .find_map(|l| l.find("wasm trap:").map(|i| l[i..].to_string()))
+        .unwrap_or_else(|| full.lines().next().unwrap_or("trap").to_string());
+    let mut line = line.trim().to_string();
+    line.truncate(300);
+    line
+}
+
 impl Runtime {
     pub fn new() -> Result<Arc<Self>> {
         let mut config = Config::new();
@@ -251,6 +269,7 @@ impl Runtime {
 
         let rt = self.clone();
         let component = component.clone();
+        let stderr_log = cfg.stderr_log.clone();
         let task = tokio::spawn(async move {
             let mut store = Store::new(&rt.engine, state);
             store.epoch_deadline_async_yield_and_update(1);
@@ -265,7 +284,12 @@ impl Runtime {
                 Err(e) => match e.downcast_ref::<I32Exit>() {
                     Some(I32Exit(0)) => Exit::Clean,
                     Some(I32Exit(c)) => Exit::Code(*c),
-                    None => Exit::Trap(format!("{e:#}")),
+                    None => {
+                        // Keep the full backtrace in the log, and a short reason in the status.
+                        let full = format!("{e:?}");
+                        log_note(&stderr_log, &format!("trap details:\n{full}"));
+                        Exit::Trap(short_trap(&full))
+                    }
                 },
             };
             // Process exit closes every file, so a clean exit commits.

examples/bench/Cargo.toml

@@ -0,0 +1,7 @@
+[package]
+name = "bench"
+version = "0.1.0"
+edition = "2024"
+
+[dependencies]
+serde_json = "1"

examples/bench/src/main.rs

@@ -0,0 +1,45 @@
+//! {"type":"bench","target":"echo","n":1000}: ping-pong with the target, one
+//! message at a time, and report the round-trip times.
+use serde_json::{Value, json};
+use std::io::{BufRead, Write};
+use std::time::Instant;
+
+fn main() {
+    let mut out = std::io::stdout().lock();
+    let mut lines = std::io::stdin().lock().lines();
+    while let Some(Ok(line)) = lines.next() {
+        let Ok(msg) = serde_json::from_str::<Value>(&line) else { continue };
+        match msg["body"]["type"].as_str() {
+            Some("shutdown") => break,
+            Some("bench") => {
+                let target = msg["body"]["target"].as_str().unwrap_or("echo").to_string();
+                let n = msg["body"]["n"].as_u64().unwrap_or(1000);
+                let mut times = Vec::new();
+                let started = Instant::now();
+                for i in 0..n {
+                    let t = Instant::now();
+                    writeln!(out, "{}", json!({"to": target, "id": i, "body": {"type": "ping", "i": i}})).unwrap();
+                    out.flush().unwrap();
+                    // Wait for the reply.
+                    loop {
+                        let Some(Ok(l)) = lines.next() else { return };
+                        let Ok(r) = serde_json::from_str::<Value>(&l) else { continue };
+                        if r["reply_to"].as_str().is_some_and(|s| s.ends_with(&format!("/{i}"))) {
+                            break;
+                        }
+                    }
+                    times.push(t.elapsed().as_micros() as u64);
+                }
+                let total = started.elapsed().as_secs_f64();
+                times.sort();
+                let p = |q: f64| times[((times.len() as f64 - 1.0) * q) as usize];
+                let reply = json!({"type": "bench_result", "n": n, "total_s": total,
+                    "msgs_per_s": (n as f64 / total).round(),
+                    "p50_us": p(0.5), "p90_us": p(0.9), "p99_us": p(0.99), "max_us": p(1.0)});
+                writeln!(out, "{}", json!({"to": msg["from"], "reply_to": msg["id"], "body": reply})).unwrap();
+                out.flush().unwrap();
+            }
+            _ => {}
+        }
+    }
+}

examples/fstest/Cargo.toml

@@ -0,0 +1,7 @@
+[package]
+name = "fstest"
+version = "0.1.0"
+edition = "2024"
+
+[dependencies]
+serde_json = "1"

examples/fstest/src/main.rs

@@ -0,0 +1,47 @@
+//! Filesystem probe for the spike. Keeps files open across messages.
+use serde_json::{Value, json};
+use std::collections::HashMap;
+use std::io::{BufRead, Write};
+
+fn main() {
+    let mut open: HashMap<String, std::fs::File> = HashMap::new();
+    let mut out = std::io::stdout().lock();
+    for line in std::io::stdin().lock().lines() {
+        let Ok(line) = line else { break };
+        let Ok(msg) = serde_json::from_str::<Value>(&line) else { continue };
+        let b = &msg["body"];
+        let path = b["path"].as_str().unwrap_or_default().to_string();
+        let r = |res: std::io::Result<Value>| match res {
+            Ok(v) => v,
+            Err(e) => json!({"type": "error", "error": e.to_string()}),
+        };
+        let reply = match b["type"].as_str() {
+            Some("shutdown") => break,
+            Some("write") => r(std::fs::write(&path, b["text"].as_str().unwrap_or_default()).map(|_| json!({"type": "ok"}))),
+            Some("open") => r((|| {
+                let mut f = std::fs::File::create(&path)?;
+                f.write_all(b["text"].as_str().unwrap_or_default().as_bytes())?;
+                open.insert(path.clone(), f);
+                Ok(json!({"type": "ok", "open": true}))
+            })()),
+            Some("append") => r((|| {
+                let mut f = std::fs::OpenOptions::new().append(true).open(&path)?;
+                f.write_all(b["text"].as_str().unwrap_or_default().as_bytes())?;
+                open.insert(path.clone(), f);
+                Ok(json!({"type": "ok", "open": true}))
+            })()),
+            Some("close") => json!({"type": "ok", "closed": open.remove(&path).is_some()}),
+            Some("read") => r(std::fs::read_to_string(&path).map(|t| json!({"type": "content", "text": t}))),
+            Some("list") => r(std::fs::read_dir(&path).map(|rd| {
+                let mut names: Vec<String> = rd.filter_map(|e| e.ok()).map(|e| e.file_name().to_string_lossy().to_string()).collect();
+                names.sort();
+                json!({"type": "entries", "names": names})
+            })),
+            Some("remove") => r(std::fs::remove_file(&path).map(|_| json!({"type": "ok"}))),
+            Some("crash") => panic!("asked to crash"),
+            _ => json!({"type": "error", "error": "unknown message"}),
+        };
+        writeln!(out, "{}", json!({"to": msg["from"], "reply_to": msg["id"], "body": reply})).unwrap();
+        out.flush().unwrap();
+    }
+}

examples/notes-v2/Cargo.toml

@@ -0,0 +1,7 @@
+[package]
+name = "notes"
+version = "0.1.0"
+edition = "2024"
+
+[dependencies]
+serde_json = "1"

examples/notes-v2/src/main.rs

@@ -0,0 +1,39 @@
+//! Keeps a list of notes in /data/notes.json and publishes them as
+//! /public/notes.txt. Version 2: adds `count` and no longer crashes.
+use serde_json::{Value, json};
+use std::io::{BufRead, Write};
+
+const DATA: &str = "/data/notes.json";
+
+fn save(notes: &[String]) {
+    std::fs::write(DATA, serde_json::to_string(notes).unwrap()).unwrap();
+    std::fs::write("/public/notes.txt", notes.join("\n") + "\n").unwrap();
+}
+
+fn main() {
+    let mut notes: Vec<String> = std::fs::read_to_string(DATA)
+        .ok()
+        .and_then(|s| serde_json::from_str(&s).ok())
+        .unwrap_or_default();
+    let mut out = std::io::stdout().lock();
+    for line in std::io::stdin().lock().lines() {
+        let Ok(line) = line else { break };
+        let Ok(msg) = serde_json::from_str::<Value>(&line) else { continue };
+        let body = &msg["body"];
+        let reply = match body["type"].as_str() {
+            Some("shutdown") => break,
+            Some("add") => {
+                notes.push(body["text"].as_str().unwrap_or_default().to_string());
+                save(&notes);
+                json!({"type": "added", "count": notes.len()})
+            }
+            Some("list") => json!({"type": "notes", "items": notes}),
+            Some("count") => json!({"type": "count", "count": notes.len(), "version": 2}),
+            // Version 2 fixes the crash.
+            Some("crash") => json!({"type": "not_crashing", "version": 2}),
+            _ => json!({"type": "error", "reason": "unknown message"}),
+        };
+        writeln!(out, "{}", json!({"to": msg["from"], "reply_to": msg["id"], "body": reply})).unwrap();
+        out.flush().unwrap();
+    }
+}

examples/notes-v2/tests/basic.jsonl

@@ -0,0 +1,8 @@
+{"send": {"type": "add", "text": "test note"}}
+{"expect": {"type": "added"}}
+{"send": {"type": "list"}}
+{"expect": {"type": "notes"}}
+{"send": {"type": "count"}}
+{"expect": {"type": "count", "version": 2}}
+{"send": {"type": "crash"}}
+{"expect": {"type": "not_crashing"}}

examples/ticker/Cargo.toml

@@ -0,0 +1,7 @@
+[package]
+name = "ticker"
+version = "0.1.0"
+edition = "2024"
+
+[dependencies]
+serde_json = "1"

examples/ticker/src/main.rs

@@ -0,0 +1,16 @@
+//! Sends {"type":"add","text":"tick N"} to `notes` every 50 ms, N = 1..=200.
+//! Does not read stdin, so the broker kills it at shutdown; that is fine.
+use std::io::Write;
+fn main() {
+    let mut out = std::io::stdout().lock();
+    for n in 1..=200 {
+        let msg = serde_json::json!({"to": "notes", "body": {"type": "add", "text": format!("tick {n}")}});
+        writeln!(out, "{msg}").unwrap();
+        out.flush().unwrap();
+        std::thread::sleep(std::time::Duration::from_millis(50));
+    }
+    // Stay alive without doing anything.
+    loop {
+        std::thread::sleep(std::time::Duration::from_secs(3600));
+    }
+}