diff --git a/agent/src/main.rs b/agent/src/main.rs index 4349233..0b0a333 100644 --- a/agent/src/main.rs +++ b/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::(&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::(&s).map_err(|_| ())) else { + return; + }; + // Every tool call needs a result, or the APIs reject the history. + let answered: std::collections::HashSet = self + .history + .iter() + .filter_map(|e| match e { + Entry::ToolResult { id, .. } => Some(id.clone()), + _ => None, + }) + .collect(); + let dangling: Vec = 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 = 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, } } } diff --git a/cli/src/broker.rs b/cli/src/broker.rs index d82b999..f185e68 100644 --- a/cli/src/broker.rs +++ b/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"; diff --git a/cli/src/main.rs b/cli/src/main.rs index b01c560..f4bc2e0 100644 --- a/cli/src/main.rs +++ b/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(()) } diff --git a/cli/src/runtime.rs b/cli/src/runtime.rs index d7f57a1..fb050ea 100644 --- a/cli/src/runtime.rs +++ b/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> { 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::() { 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. diff --git a/examples/bench/Cargo.toml b/examples/bench/Cargo.toml new file mode 100644 index 0000000..16ab922 --- /dev/null +++ b/examples/bench/Cargo.toml @@ -0,0 +1,7 @@ +[package] +name = "bench" +version = "0.1.0" +edition = "2024" + +[dependencies] +serde_json = "1" diff --git a/examples/bench/src/main.rs b/examples/bench/src/main.rs new file mode 100644 index 0000000..61c1bc3 --- /dev/null +++ b/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::(&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::(&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(); + } + _ => {} + } + } +} diff --git a/examples/fstest/Cargo.toml b/examples/fstest/Cargo.toml new file mode 100644 index 0000000..0b6b178 --- /dev/null +++ b/examples/fstest/Cargo.toml @@ -0,0 +1,7 @@ +[package] +name = "fstest" +version = "0.1.0" +edition = "2024" + +[dependencies] +serde_json = "1" diff --git a/examples/fstest/src/main.rs b/examples/fstest/src/main.rs new file mode 100644 index 0000000..1ac1d70 --- /dev/null +++ b/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 = 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::(&line) else { continue }; + let b = &msg["body"]; + let path = b["path"].as_str().unwrap_or_default().to_string(); + let r = |res: std::io::Result| 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 = 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(); + } +} diff --git a/examples/notes-v2/Cargo.toml b/examples/notes-v2/Cargo.toml new file mode 100644 index 0000000..2b0aa5f --- /dev/null +++ b/examples/notes-v2/Cargo.toml @@ -0,0 +1,7 @@ +[package] +name = "notes" +version = "0.1.0" +edition = "2024" + +[dependencies] +serde_json = "1" diff --git a/examples/notes-v2/src/main.rs b/examples/notes-v2/src/main.rs new file mode 100644 index 0000000..4d539d5 --- /dev/null +++ b/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 = 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::(&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(¬es); + 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(); + } +} diff --git a/examples/notes-v2/tests/basic.jsonl b/examples/notes-v2/tests/basic.jsonl new file mode 100644 index 0000000..40d794a --- /dev/null +++ b/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"}} diff --git a/examples/ticker/Cargo.toml b/examples/ticker/Cargo.toml new file mode 100644 index 0000000..dea727b --- /dev/null +++ b/examples/ticker/Cargo.toml @@ -0,0 +1,7 @@ +[package] +name = "ticker" +version = "0.1.0" +edition = "2024" + +[dependencies] +serde_json = "1" diff --git a/examples/ticker/src/main.rs b/examples/ticker/src/main.rs new file mode 100644 index 0000000..2aeef21 --- /dev/null +++ b/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)); + } +}