From 7f76eab1619ff5833d5d1f97bd0b4e0b638685ab Mon Sep 17 00:00:00 2001 From: Surma Date: Thu, 15 Oct 2026 11:17:14 +0100 Subject: [PATCH] SDK sausage-actor; examples and agent on the SDK; agent without limits - sausage-actor: Message, messages(), send/reply/publish, async Inbox - agent: append-only /data/history.jsonl, no caps on history, steps, waits, tool output, or reads; max_tokens only when configured (Anthropic: 32000) --- Cargo.lock | 8 +- agent/Cargo.toml | 2 +- agent/src/llm.rs | 23 +- agent/src/main.rs | 432 +++++++++++++++------------------- agent/src/prompt.md | 110 ++++----- examples/bench/Cargo.toml | 4 +- examples/bench/src/main.rs | 71 +++--- examples/crasher/Cargo.toml | 2 + examples/echo/Cargo.toml | 4 +- examples/echo/src/main.rs | 14 +- examples/fetch/Cargo.toml | 4 +- examples/fetch/src/main.rs | 94 +++----- examples/fstest/Cargo.toml | 4 +- examples/fstest/src/main.rs | 57 ++--- examples/notes-v2/Cargo.toml | 3 + examples/notes-v2/src/main.rs | 30 +-- examples/notes/Cargo.toml | 3 + examples/notes/src/main.rs | 27 +-- examples/ticker/Cargo.toml | 4 +- examples/ticker/src/main.rs | 17 +- sdk/Cargo.toml | 8 + sdk/src/lib.rs | 204 +++++++++++++++- 22 files changed, 607 insertions(+), 518 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 47263c0..8fced98 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -15,7 +15,7 @@ dependencies = [ name = "agent" version = "0.1.0" dependencies = [ - "async-channel", + "sausage-actor", "serde", "serde_json", "wstd", @@ -1601,6 +1601,12 @@ dependencies = [ [[package]] name = "sausage-actor" version = "0.1.0" +dependencies = [ + "async-channel", + "serde", + "serde_json", + "wstd", +] [[package]] name = "scopeguard" diff --git a/agent/Cargo.toml b/agent/Cargo.toml index a720c9c..0700110 100644 --- a/agent/Cargo.toml +++ b/agent/Cargo.toml @@ -5,7 +5,7 @@ edition = "2024" description = "The Sausage agent actor." [dependencies] -async-channel = "2" +sausage-actor = { version = "0.1", path = "../sdk", features = ["async"] } serde = { version = "1", features = ["derive"] } serde_json = "1" wstd = "0.6" diff --git a/agent/src/llm.rs b/agent/src/llm.rs index 53d80d2..f22d839 100644 --- a/agent/src/llm.rs +++ b/agent/src/llm.rs @@ -6,7 +6,6 @@ use serde::{Deserialize, Serialize}; use serde_json::{Value, json}; use wstd::http::{Body, Client, Request}; -use wstd::time::Duration; #[derive(Deserialize, Clone)] pub struct Config { @@ -18,12 +17,8 @@ pub struct Config { /// Optional reasoning effort for OpenAI models ("low", "medium", ...). #[serde(default)] pub reasoning_effort: Option, - #[serde(default = "default_max_tokens")] - pub max_tokens: u32, -} - -fn default_max_tokens() -> u32 { - 16000 + #[serde(default)] + pub max_tokens: Option, } #[derive(Serialize, Deserialize, Clone, Debug)] @@ -57,17 +52,13 @@ async fn post(url: &str, headers: &[(&str, String)], body: &Value) -> Result) { - let mut stdin = wstd::io::stdin(); - let mut buf = Vec::new(); - let mut chunk = vec![0u8; 16 * 1024]; - loop { - match stdin.read(&mut chunk).await { - Ok(0) | Err(_) => break, - Ok(n) => { - buf.extend_from_slice(&chunk[..n]); - while let Some(pos) = buf.iter().position(|b| *b == b'\n') { - let line: Vec = buf.drain(..=pos).collect(); - if let Ok(v) = serde_json::from_slice::(&line) - && tx.send(v).await.is_err() - { - return; - } - } - } - } - } - tx.close(); -} - -fn truncate(mut s: String, max: usize) -> String { - if s.len() > max { - let mut end = max; - while !s.is_char_boundary(end) { - end -= 1; - } - let total = s.len(); - s.truncate(end); - s.push_str(&format!("\n… (truncated, {total} bytes in total)")); - } - s + wstd::runtime::block_on(async { Agent::new().run().await }); } fn tools() -> Vec { vec![ Tool { name: "send_message", - description: "Send a message to an actor. With wait_seconds > 0, wait for the reply (a message whose reply_to is the ID of this message). Returns the message ID, the reply (or null), and other messages that arrived meanwhile.", + description: "Send a message to an actor. With wait_seconds > 0, wait for its reply. Returns the message ID, the reply, and other messages that arrived meanwhile.", parameters: json!({ "type": "object", "properties": { "to": {"type": "string", "description": "Actor ID"}, - "body": {"type": "object", "description": "Message body. Must have a \"type\" field."}, + "body": {"type": "object", "description": "Message body. Must have a type field."}, "wait_seconds": {"type": "number", "description": "Seconds to wait for the reply. 0 = do not wait."} }, "required": ["to", "body"] @@ -87,7 +35,7 @@ fn tools() -> Vec { }, Tool { name: "wait_for_messages", - description: "Wait up to `seconds` for messages to arrive and return them.", + description: "Wait up to seconds for messages to arrive and return them.", parameters: json!({ "type": "object", "properties": {"seconds": {"type": "number"}}, @@ -102,7 +50,7 @@ fn tools() -> Vec { "properties": { "path": {"type": "string"}, "offset": {"type": "integer", "description": "Byte offset (default 0)"}, - "max_bytes": {"type": "integer", "description": "Default 30000"} + "max_bytes": {"type": "integer", "description": "Optional maximum number of bytes to read"} }, "required": ["path"] }), @@ -110,15 +58,11 @@ fn tools() -> Vec { Tool { name: "list_dir", description: "List a folder.", - parameters: json!({ - "type": "object", - "properties": {"path": {"type": "string"}}, - "required": ["path"] - }), + parameters: json!({"type": "object", "properties": {"path": {"type": "string"}}, "required": ["path"]}), }, Tool { name: "write_file", - description: "Write a text file (only under /data or /public). Creates parent folders.", + description: "Write a text file under /data or /public. Creates parent folders.", parameters: json!({ "type": "object", "properties": {"path": {"type": "string"}, "content": {"type": "string"}}, @@ -137,18 +81,14 @@ fn tools() -> Vec { Tool { name: "remove_path", description: "Remove a file or folder under /data or /public.", - parameters: json!({ - "type": "object", - "properties": {"path": {"type": "string"}}, - "required": ["path"] - }), + parameters: json!({"type": "object", "properties": {"path": {"type": "string"}}, "required": ["path"]}), }, ] } fn copy_tree(from: &str, to: &str) -> std::io::Result { std::fs::create_dir_all(to)?; - let mut n = 0; + 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(); @@ -156,87 +96,94 @@ fn copy_tree(from: &str, to: &str) -> std::io::Result { let dst = format!("{to}/{name}"); if entry.file_type()?.is_dir() { if name != "target" { - n += copy_tree(&src, &dst)?; + count += copy_tree(&src, &dst)?; } } else { - std::fs::copy(&src, &dst)?; - n += 1; + std::fs::copy(src, dst)?; + count += 1; } } - Ok(n) + Ok(count) } struct Agent { - rx: Receiver, - stdout: wstd::io::Stdout, + inbox: Inbox, history: Vec, - /// Text messages that start a new turn later. - queue: VecDeque, - /// Other messages that arrived while the agent was idle. - notes: Vec, - seq: u64, + queue: VecDeque, + notes: Vec, shutdown: bool, } impl Agent { - fn new(rx: Receiver) -> Self { + fn new() -> Self { let history = std::fs::read_to_string(HISTORY) - .ok() - .and_then(|s| serde_json::from_str(&s).ok()) - .unwrap_or_default(); + .unwrap_or_default() + .lines() + .filter_map(|line| serde_json::from_str(line).ok()) + .collect(); Self { - rx, - stdout: wstd::io::stdout(), + inbox: Inbox::start(), history, queue: VecDeque::new(), notes: Vec::new(), - seq: 0, shutdown: false, } } + fn add_history(&mut self, entry: Entry) { + let line = serde_json::to_string(&entry).unwrap(); + match std::fs::OpenOptions::new() + .create(true) + .append(true) + .open(HISTORY) + { + Ok(mut file) => { + if let Err(e) = writeln!(file, "{line}") { + eprintln!("agent: cannot append history: {e}"); + } + } + Err(e) => eprintln!("agent: cannot append history: {e}"), + } + self.history.push(entry); + } + 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, - None => match self.rx.recv().await { - Ok(m) => m, - Err(_) => break, + Some(msg) => msg, + None => match self.inbox.recv().await { + Some(msg) => msg, + None => break, }, }; - if msg["body"]["type"] == "shutdown" { - break; - } if starts_turn(&msg) { self.turn(msg).await; } else { self.notes.push(msg); - if self.notes.len() > MAX_NOTES { - self.notes.remove(0); - } } } - self.save(); - 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 { + 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); - 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 + + let answered: HashSet = self .history .iter() - .filter_map(|e| match e { + .filter_map(|entry| match entry { Entry::ToolResult { id, .. } => Some(id.clone()), _ => None, }) @@ -244,164 +191,164 @@ impl Agent { let dangling: Vec = self .history .iter() - .filter_map(|e| match e { + .filter_map(|entry| match entry { Entry::ToolCall { id, .. } if !answered.contains(id) => Some(id.clone()), _ => None, }) .collect(); for id in dangling { - self.history.push(Entry::ToolResult { + self.add_history(Entry::ToolResult { id, - output: json!({"error": "interrupted: the agent restarted before this tool finished"}).to_string(), + 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 { + 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." ), }); - eprintln!("agent: resuming an interrupted turn"); - self.steps(&cfg, turn["from"].clone(), turn["id"].clone()).await; + 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; } - fn save(&mut self) { - if self.history.len() > MAX_HISTORY { - let start = self.history.len() - MAX_HISTORY; - let cut = (start..self.history.len()) - .find(|i| matches!(self.history[*i], Entry::User { .. })) - .unwrap_or(start); - self.history.drain(..cut); - } - if let Ok(s) = serde_json::to_string(&self.history) - && let Err(e) = std::fs::write(HISTORY, s) + 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())) { - eprintln!("agent: cannot save history: {e}"); - } - } - - async fn emit(&mut self, msg: Value) { - let mut line = msg.to_string(); - line.push('\n'); - if self.stdout.write_all(line.as_bytes()).await.is_err() || self.stdout.flush().await.is_err() { - eprintln!("agent: cannot write to stdout"); - } - } - - async fn reply(&mut self, to: &Value, reply_to: &Value, text: String) { - self.emit(json!({"to": to, "reply_to": reply_to, "body": {"type": "text", "text": text}})).await; - } - - /// Sort a message that arrived during a turn. - fn stash(&mut self, m: Value, others: &mut Vec) { - if m["body"]["type"] == "shutdown" { - self.shutdown = true; - } else if starts_turn(&m) { - others.push(json!({"queued_for_a_later_turn": true, "message": m})); - self.queue.push_back(m); - } else { - others.push(m); - } - } - - async fn turn(&mut self, msg: Value) { - let from = msg["from"].clone(); - let id = msg["id"].clone(); - let cfg: Config = match std::fs::read_to_string(CONFIG).map_err(|e| e.to_string()).and_then(|s| serde_json::from_str(&s).map_err(|e| e.to_string())) { - Ok(c) => c, + Ok(cfg) => cfg, Err(e) => { - self.reply(&from, &id, format!("I am not configured: cannot read {CONFIG}: {e}")).await; + let _ = reply( + &message, + json!({"type": "text", "text": format!("I am not configured: cannot read {CONFIG}: {e}")}), + ); return; } }; let mut text = format!( "Message {} from {}:\n{}", - id.as_str().unwrap_or("?"), - from.as_str().unwrap_or("?"), - msg["body"]["text"].as_str().unwrap_or("") + message.id, + message.from, + message.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")); + let notes = self + .notes + .drain(..) + .map(|note| serde_json::to_string(¬e).unwrap()) + .collect::>() + .join("\n"); + text = format!("Messages that arrived while you were idle:\n{notes}\n\n{text}"); } - self.history.push(Entry::User { text }); - self.steps(&cfg, from, id).await; + self.add_history(Entry::User { text }); + self.steps(&cfg, &message).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) { + async fn steps(&mut self, cfg: &Config, message: &Message) { 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; - let entries = match result { - Ok(e) => e, + loop { + let entries = match llm::complete(cfg, SYSTEM, &tools, &self.history).await { + Ok(entries) => entries, Err(e) => { - eprintln!("agent: inference failed: {e}"); - self.history.push(Entry::Assistant { text: format!("(inference failed: {e})") }); - self.save(); - self.reply(&from, &id, format!("Inference failed: {e}")).await; + self.add_history(Entry::Assistant { + text: format!("(inference failed: {e})"), + }); + let _ = reply( + message, + json!({"type": "text", "text": format!("Inference failed: {e}")}), + ); return; } }; - eprintln!("agent: inference took {:?}", started.elapsed()); let calls: Vec<(String, String, String)> = entries .iter() - .filter_map(|e| match e { - Entry::ToolCall { id, name, arguments } => Some((id.clone(), name.clone(), arguments.clone())), + .filter_map(|entry| match entry { + Entry::ToolCall { + id, + name, + arguments, + } => Some((id.clone(), name.clone(), arguments.clone())), _ => None, }) .collect(); let answer: Vec = entries .iter() - .filter_map(|e| match e { + .filter_map(|entry| match entry { Entry::Assistant { text } => Some(text.clone()), _ => None, }) .collect(); - self.history.extend(entries); + for entry in entries { + self.add_history(entry); + } if calls.is_empty() { - self.save(); - let answer = if answer.is_empty() { "(no answer)".to_string() } else { answer.join("\n\n") }; - self.reply(&from, &id, answer).await; + let answer = if answer.is_empty() { + "(no answer)".to_string() + } else { + answer.join("\n\n") + }; + let _ = reply(message, json!({"type": "text", "text": answer})); return; } - for (call_id, name, args) in calls { + for (id, name, args) in calls { let output = self.tool(&name, &args).await; - self.history.push(Entry::ToolResult { id: call_id, output: truncate(output.to_string(), MAX_TOOL_OUTPUT) }); + self.add_history(Entry::ToolResult { + id, + output: output.to_string(), + }); } - 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"); + let _ = std::fs::write( + TURN, + json!({"from": message.from, "id": message.id}).to_string(), + ); return; } } - self.reply(&from, &id, format!("I stopped after {MAX_STEPS} steps. Tell me to continue if you want.")).await; + } + + 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); + } else { + others.push(serde_json::to_value(message).unwrap()); + } } async fn tool(&mut self, name: &str, args: &str) -> Value { let a: Value = match serde_json::from_str(args) { - Ok(v) => v, + Ok(value) => value, Err(e) => return json!({"error": format!("arguments are not valid JSON: {e}")}), }; - let path = |k: &str| a[k].as_str().unwrap_or_default().to_string(); + 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, "wait_for_messages" => { - let secs = a["seconds"].as_f64().unwrap_or(10.0).clamp(0.0, 900.0); + let seconds = a["seconds"].as_f64().unwrap_or(10.0).max(0.0); let mut others = Vec::new(); - 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); + match self + .inbox + .recv() + .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(Err(_)) => self.shutdown = true, + Ok(None) => self.shutdown = true, Err(_) => {} } json!({"messages": others}) @@ -411,8 +358,9 @@ impl Agent { match std::fs::read(&p) { Ok(bytes) => { let offset = (a["offset"].as_u64().unwrap_or(0) as usize).min(bytes.len()); - let max = a["max_bytes"].as_u64().unwrap_or(30_000) as usize; - let end = (offset + max).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]), @@ -425,12 +373,11 @@ impl Agent { "list_dir" => match std::fs::read_dir(path("path")) { Ok(rd) => { let mut entries: Vec = rd - .filter_map(|e| e.ok()) - .take(500) - .map(|e| { - let meta = e.metadata().ok(); + .filter_map(|entry| entry.ok()) + .map(|entry| { + let meta = entry.metadata().ok(); json!({ - "name": e.file_name().to_string_lossy(), + "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()), }) @@ -452,13 +399,17 @@ impl Agent { } } "copy_dir" => match copy_tree(&path("from"), &path("to")) { - Ok(n) => json!({"copied_files": n}), + Ok(count) => json!({"copied_files": count}), Err(e) => err(e), }, "remove_path" => { let p = path("path"); - let r = if std::path::Path::new(&p).is_dir() { std::fs::remove_dir_all(&p) } else { std::fs::remove_file(&p) }; - match r { + let result = if std::path::Path::new(&p).is_dir() { + std::fs::remove_dir_all(&p) + } else { + std::fs::remove_file(&p) + }; + match result { Ok(()) => json!({"removed": p}), Err(e) => err(e), } @@ -467,34 +418,32 @@ impl Agent { } } - async fn send_message(&mut self, a: &Value) -> Value { - let Some(to) = a["to"].as_str() else { return json!({"error": "missing \"to\""}) }; - if !a["body"].is_object() { + async fn send_message(&mut self, args: &Value) -> Value { + let Some(to) = args["to"].as_str() else { + return json!({"error": "missing \"to\""}); + }; + if !args["body"].is_object() { return json!({"error": "\"body\" must be a JSON object"}); } - self.seq += 1; - let own = format!("a{}-{}", wstd_now_ms(), self.seq); - let full = format!("agent/{own}"); - self.emit(json!({"to": to, "id": own, "body": a["body"]})).await; - let wait = a["wait_seconds"].as_f64().unwrap_or(0.0).clamp(0.0, 900.0); + let id = send(to, args["body"].clone()); + let wait = args["wait_seconds"].as_f64().unwrap_or(0.0).max(0.0); let mut reply = Value::Null; let mut others = Vec::new(); if wait > 0.0 { let deadline = Instant::now() + Duration::from_secs_f64(wait); loop { - match self.rx.recv().timeout(deadline).await { - Ok(Ok(m)) => { - if m["reply_to"] == full.as_str() { - reply = m; - break; - } - self.stash(m, &mut others); + match self.inbox.recv().timeout(deadline).await { + Ok(Some(message)) if message.reply_to.as_deref() == Some(&id) => { + reply = serde_json::to_value(message).unwrap(); + break; + } + Ok(Some(message)) => { + self.stash(message, &mut others); if self.shutdown { break; } } - Ok(Err(_)) => { - // stdin closed: the broker stops us. + Ok(None) => { self.shutdown = true; break; } @@ -502,19 +451,10 @@ impl Agent { } } } - json!({"id": full, "reply": reply, "waited": wait > 0.0, "other_messages": others}) + json!({"id": id, "reply": reply, "waited": wait > 0.0, "other_messages": others}) } } -/// A text message that is not a reply starts a turn. Replies never do, so -/// two actors cannot keep the agent in an endless conversation. -fn starts_turn(m: &Value) -> bool { - m["body"]["type"] == "text" && m.get("reply_to").is_none_or(Value::is_null) -} - -fn wstd_now_ms() -> u64 { - std::time::SystemTime::now() - .duration_since(std::time::UNIX_EPOCH) - .map(|d| d.as_millis() as u64) - .unwrap_or(0) +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 b934235..f640741 100644 --- a/agent/src/prompt.md +++ b/agent/src/prompt.md @@ -1,80 +1,64 @@ -You are the agent of a Sausage world. Your actor ID is `agent`. +You are the `agent` actor in a Sausage world. Your actor ID is `agent`. -Sausage is a system of actors. Each actor is a WASI program (a `.wasm` file). Actors talk only through messages. People talk to you through other actors, for example `console` (a terminal) or later `matrix`. You help them by using actors, and by writing, testing, and upgrading actors. You can also change your own code. +Sausage is a system of actors. Each actor is a WASM program that sends JSON messages through a broker. Help people use actors, write new actors, and upgrade them when asked. -# Messages +## Messages and the SDK -Every message has a sender (`from`), an ID, an optional `reply_to` (the ID of the message it answers), and a JSON `body`. A body is an object with a `"type"` field. There is no request/response: a reply is a new message with `reply_to` set. Delivery is at most once. +Actors read one JSON message per line. Each message has `id`, `from`, optional `reply_to`, and `body`. The broker routes a message with `to`. Without `to`, the broker publishes it to subscribers. A reply is a new message with `reply_to` set to the original ID. -A WASM actor reads one JSON object per line on stdin: -`{"id":"m-17","from":"console","body":{...}}` -and writes one JSON object per line on stdout: -`{"to":"console","reply_to":"m-17","body":{...}}` -Without `"to"`, the broker publishes the message to the subscribers of the sender. An actor may set its own `"id"`. The broker then uses `"/"`. stdout is only for messages. Logs go to stderr. The broker sends `{"type":"shutdown"}` before it stops an actor. EOF on stdin means "stop now". +Use `sausage-actor = "0.1"` in a Rust actor's `Cargo.toml`. -# Filesystem - -Each actor sees: -- `/data`: private state, read-write. Only the actor itself sees it. -- `/public`: its public files, read-write. -- `/actor//public`: the public files of every actor, read-only. -- `/actor//source`: the source of the running version of every actor, read-only. -- `/actor//status.json` and `/actor//stderr.log`: state and stderr log of every actor. - -`ls /actor` lists all actors. A write becomes visible to other actors only when the writer closes the file. Send large or binary data as a file path, not in a message. - -# Built-in actors - -`broker` (only accepts `.wasm` files, knows no source language): -- `{"type":"spawn","actor":"","wasm":"","source":"","built_by":"cargo"}` -- `{"type":"upgrade","actor":"","wasm":"","source":"","built_by":"cargo"}` -- `{"type":"rollback","actor":""}` (optional `"version"`) -- `{"type":"stop","actor":""}`, `{"type":"start","actor":""}`, `{"type":"remove","actor":""}` (archives, never deletes) -- `{"type":"subscribe","actor":""}`, `{"type":"unsubscribe","actor":""}`, `{"type":"list"}` - -Before an upgrade or spawn, the broker tests the candidate on a copy of the actor's `/data` and `/public`, with no HTTP access. The tests are message transcripts in `/tests/*.jsonl`. Each line is `{"send": }` or `{"expect": }` (optional `"to"`). Without test files, the candidate must start and stay up for 2 seconds. If a new version crashes 3 times within 30 seconds, the broker rolls it back automatically. The broker keeps the message that crashed an actor and delivers it again after the next upgrade. - -`cargo` (built-in Rust compiler): -- `{"type":"build","source":""}` → `{"type":"built","wasm":"/actor/cargo/public/builds//.wasm",...}` or `{"type":"build_failed","errors":"..."}`. A build can take a minute. Wait up to 600 seconds for the reply. - -# Writing an actor in Rust - -1. Write the crate to `/public/drafts//`: `Cargo.toml`, `src/main.rs`, and `tests/*.jsonl`. Use `edition = "2024"`. Use `serde_json` for messages. -2. Send `build` to `cargo` with the folder path `/actor/agent/public/drafts/` and wait for the reply. Fix compiler errors and build again. -3. Send `spawn` (new actor) or `upgrade` (existing actor) to `broker` with the `wasm` path from the build reply and the same `source` folder. Wait for the reply. -4. Send a message to the new actor to check it. - -A simple actor that only reacts to messages can block on stdin: +A simple actor reads messages with `messages()` and replies with `reply()`: ```rust -use serde_json::{json, Value}; -use std::io::{BufRead, Write}; +use sausage_actor::{json, messages, reply}; fn main() { - 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("ping") => json!({"type": "pong"}), - _ => json!({"type": "error", "reason": "unknown message"}), - }; - writeln!(out, "{}", json!({"to": msg["from"], "reply_to": msg["id"], "body": reply})).unwrap(); - out.flush().unwrap(); + for message in messages() { + if message.kind() == "ping" { + let _ = reply(&message, json!({"type": "pong"})); + } } } ``` -An actor that needs HTTP or timers and must also read messages uses the async `wstd` crate (`wstd = "0.6"`): `#[wstd::main] async fn main()`, `wstd::io::stdin()` (`AsyncRead`), `wstd::io::stdout()` (`AsyncWrite`, `write_all`, `flush`), `wstd::http::{Client, Request, Body}` (`Client::new().send(Request::get(url).body(Body::empty())?).await?`, then `response.into_body().contents().await?`), `wstd::task::sleep(wstd::time::Duration::from_secs(n)).await`, and `wstd::runtime::spawn(future)`. Actors have outgoing HTTP and HTTPS only, no raw sockets. Threads are not available. Store state in `/data` so it survives restarts and upgrades. A new version must read the data format of the old version. +Use `Inbox` for an actor that needs HTTP or timers. Enable the SDK's `async` feature. Use `wstd` for HTTP: -# Changing yourself +```rust +use sausage_actor::{Inbox, json, reply}; +use wstd::http::{Body, Client, Request}; -Your own source is at `/actor/agent/source`. To change yourself, copy it to `/public/drafts/agent`, edit the copy, build it, and upgrade `agent`. The upgrade restarts you, and the conversation continues from `/data`. Be careful: if the new version cannot answer, the person must roll you back through the console. Change yourself only when the person asks for it. +fn main() { + wstd::runtime::block_on(async { + let inbox = Inbox::start(); + while let Some(message) = inbox.recv().await { + if message.kind() != "fetch" { + continue; + } + 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()})); + } + }); +} +``` -# How to work +## Writing and testing actors -- Use tools to look at the real state before you claim anything. Read files instead of guessing. -- Keep answers short and concrete. Say what you changed and whether it worked. -- Your final text in a turn goes back to the person who wrote to you. +Write each Rust actor under `/public/drafts//`. Add `Cargo.toml`, `src/main.rs`, and tests under `tests/`. + +The broker tests each `tests/*.json` file before an upgrade or spawn. Each file has a `steps` list. A step can have `send`, `expect`, `from`, and `to`. An expected body matches a subset of the reply body. Tests have no HTTP access. + +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. + +## Filesystem + +Your private state lives under `/data`. Your public files live under `/public`. Other actors' `data/` directories are hidden. + +The host will provide `/actor//` with `actor.wasm`, `actor.json`, `source/`, `public/`, `status.json`, and `stderr.log`. Each `versions//` contains `actor.wasm`, `source/`, and `meta.json`. + +Keep state in `/data` so it survives restarts and upgrades. Read the old data format when you upgrade an actor. + +Your source lives at `/actor/agent/source`. To change yourself, copy it to `/public/drafts/agent`, build it with `cargo`, and send an `upgrade` request to `broker`. The upgrade restarts you. Your turn resumes from `/data/turn.json`. diff --git a/examples/bench/Cargo.toml b/examples/bench/Cargo.toml index 16ab922..a3ac152 100644 --- a/examples/bench/Cargo.toml +++ b/examples/bench/Cargo.toml @@ -4,4 +4,6 @@ version = "0.1.0" edition = "2024" [dependencies] -serde_json = "1" +sausage-actor = { version = "0.1", path = "../../sdk" } + +[workspace] diff --git a/examples/bench/src/main.rs b/examples/bench/src/main.rs index 61c1bc3..eafdeff 100644 --- a/examples/bench/src/main.rs +++ b/examples/bench/src/main.rs @@ -1,45 +1,38 @@ -//! {"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}; +//! Ping-pong with a target and report the round-trip times. +use sausage_actor::{json, messages, reply, send}; 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(); - } - _ => {} + let mut inbox = messages(); + while let Some(message) = inbox.next() { + if message.kind() != "bench" { + continue; } + let target = message.body["target"].as_str().unwrap_or("echo"); + let n = message.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(); + let id = send(target, json!({"type": "ping", "i": i})); + loop { + let Some(reply) = inbox.next() else { return }; + if reply.reply_to.as_deref() == Some(id.as_str()) { + 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( + &message, + 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) + }), + ); } } diff --git a/examples/crasher/Cargo.toml b/examples/crasher/Cargo.toml index af6eb0b..fdb8dc7 100644 --- a/examples/crasher/Cargo.toml +++ b/examples/crasher/Cargo.toml @@ -4,3 +4,5 @@ version = "0.1.0" edition = "2024" [dependencies] + +[workspace] diff --git a/examples/echo/Cargo.toml b/examples/echo/Cargo.toml index 46c5a87..910b0a8 100644 --- a/examples/echo/Cargo.toml +++ b/examples/echo/Cargo.toml @@ -4,4 +4,6 @@ version = "0.1.0" edition = "2024" [dependencies] -serde_json = "1" +sausage-actor = { version = "0.1", path = "../../sdk" } + +[workspace] diff --git a/examples/echo/src/main.rs b/examples/echo/src/main.rs index 8e54d0f..c983c10 100644 --- a/examples/echo/src/main.rs +++ b/examples/echo/src/main.rs @@ -1,15 +1,7 @@ -use serde_json::{json, Value}; -use std::io::{BufRead, Write}; +use sausage_actor::{json, messages, reply}; fn main() { - let mut out = std::io::stdout().lock(); - for line in std::io::stdin().lock().lines() { - let Ok(msg) = serde_json::from_str::(&line.unwrap()) else { continue }; - if msg["body"]["type"] == "shutdown" { - break; - } - let env = json!({"to": msg["from"], "reply_to": msg["id"], "body": {"type": "echo", "got": msg["body"]}}); - writeln!(out, "{env}").unwrap(); - out.flush().unwrap(); + for message in messages() { + let _ = reply(&message, json!({"type": "echo", "got": message.body})); } } diff --git a/examples/fetch/Cargo.toml b/examples/fetch/Cargo.toml index 895db74..5843776 100644 --- a/examples/fetch/Cargo.toml +++ b/examples/fetch/Cargo.toml @@ -4,5 +4,7 @@ version = "0.1.0" edition = "2024" [dependencies] -serde_json = "1" +sausage-actor = { version = "0.1", path = "../../sdk", features = ["async"] } wstd = "0.6" + +[workspace] diff --git a/examples/fetch/src/main.rs b/examples/fetch/src/main.rs index f94d44c..2ad4da7 100644 --- a/examples/fetch/src/main.rs +++ b/examples/fetch/src/main.rs @@ -1,77 +1,57 @@ -//! Fetches URLs over HTTP(S) and writes the body to /public/fetch/. -//! Reads stdin and runs requests at the same time: while a fetch runs, -//! `ping` still answers at once. -use serde_json::{Value, json}; -use std::cell::RefCell; -use std::rc::Rc; +//! Fetch URLs over HTTP while answering pings. +use sausage_actor::{Inbox, Message, json, me, reply}; use wstd::http::{Body, Client, Request}; -use wstd::io::{AsyncRead, AsyncWrite}; -type Out = Rc>>; - -async fn flush(out: &Out) { - let lines: Vec = out.borrow_mut().drain(..).collect(); - let mut stdout = wstd::io::stdout(); - for l in lines { - let _ = stdout.write_all(format!("{l}\n").as_bytes()).await; - } - let _ = stdout.flush().await; -} - -async fn fetch(n: u64, msg: Value, out: Out) { - let url = msg["body"]["url"].as_str().unwrap_or_default().to_string(); +async fn fetch(n: u64, message: Message) { + let url = message.body["url"].as_str().unwrap_or_default(); let started = wstd::time::Instant::now(); let result = async { - let req = Request::get(&url).body(Body::empty()).map_err(|e| e.to_string())?; - let resp = Client::new().send(req).await.map_err(|e| format!("{e:?}"))?; - let status = resp.status().as_u16(); - let bytes = resp.into_body().contents().await.map_err(|e| format!("{e:?}"))?.to_vec(); + let request = Request::get(url) + .body(Body::empty()) + .map_err(|e| e.to_string())?; + let response = Client::new() + .send(request) + .await + .map_err(|e| format!("{e:?}"))?; + let status = response.status().as_u16(); + let bytes = response + .into_body() + .contents() + .await + .map_err(|e| format!("{e:?}"))? + .to_vec(); std::fs::create_dir_all("/public/fetch").map_err(|e| e.to_string())?; let path = format!("/public/fetch/{n}"); - std::fs::write(&path, &bytes).map_err(|e| e.to_string())?; - Ok::<_, String>((status, bytes.len(), path)) + std::fs::write(path, &bytes).map_err(|e| e.to_string())?; + Ok::<_, String>((status, bytes.len())) } .await; let ms = started.elapsed().as_millis(); let body = match result { - Ok((status, size, path)) => { - let id = std::env::var("SAUSAGE_ACTOR").unwrap_or("fetch".into()); - json!({"type": "fetched", "status": status, "bytes": size, "path": path.replacen("/public", &format!("/actor/{id}/public"), 1), "ms": ms}) + Ok((status, size)) => { + let id = me(); + let actor = if id.is_empty() { "fetch" } else { &id }; + json!({"type": "fetched", "status": status, "bytes": size, "path": format!("/actor/{actor}/public/fetch/{n}"), "ms": ms}) } - Err(e) => json!({"type": "error", "reason": e, "ms": ms}), + Err(reason) => json!({"type": "error", "reason": reason, "ms": ms}), }; - out.borrow_mut().push(json!({"to": msg["from"], "reply_to": msg["id"], "body": body}).to_string()); - flush(&out).await; + let _ = reply(&message, body); } fn main() { wstd::runtime::block_on(async { - let out: Out = Rc::new(RefCell::new(Vec::new())); - let mut stdin = wstd::io::stdin(); - let mut buf = Vec::new(); - let mut chunk = vec![0u8; 8192]; - let mut n = 0u64; - loop { - let Ok(len) = stdin.read(&mut chunk).await else { break }; - if len == 0 { - break; - } - buf.extend_from_slice(&chunk[..len]); - while let Some(pos) = buf.iter().position(|b| *b == b'\n') { - let line: Vec = buf.drain(..=pos).collect(); - let Ok(msg) = serde_json::from_slice::(&line) else { continue }; - match msg["body"]["type"].as_str() { - Some("shutdown") => return, - Some("fetch") => { - n += 1; - wstd::runtime::spawn(fetch(n, msg, out.clone())).detach(); - } - Some("ping") => { - out.borrow_mut().push(json!({"to": msg["from"], "reply_to": msg["id"], "body": {"type": "pong"}}).to_string()); - flush(&out).await; - } - _ => {} + 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(); } + "ping" => { + let _ = reply(&message, json!({"type": "pong"})); + } + _ => {} } } }); diff --git a/examples/fstest/Cargo.toml b/examples/fstest/Cargo.toml index 0b6b178..dda05a5 100644 --- a/examples/fstest/Cargo.toml +++ b/examples/fstest/Cargo.toml @@ -4,4 +4,6 @@ version = "0.1.0" edition = "2024" [dependencies] -serde_json = "1" +sausage-actor = { version = "0.1", path = "../../sdk" } + +[workspace] diff --git a/examples/fstest/src/main.rs b/examples/fstest/src/main.rs index 1ac1d70..e81a240 100644 --- a/examples/fstest/src/main.rs +++ b/examples/fstest/src/main.rs @@ -1,47 +1,38 @@ -//! Filesystem probe for the spike. Keeps files open across messages. -use serde_json::{Value, json}; -use std::collections::HashMap; -use std::io::{BufRead, Write}; +//! Filesystem probe. Keeps files open across messages. +use sausage_actor::{Value, json, messages, reply}; +use std::{collections::HashMap, io::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); + let mut open = HashMap::::new(); + for msg in messages() { + let b = &msg.body; + let path = b["path"].as_str().unwrap_or_default(); + let r = |result: std::io::Result| result.unwrap_or_else(|e| json!({"type": "error", "error": e.to_string()})); + let response = match msg.kind() { + "write" => r(std::fs::write(path, b["text"].as_str().unwrap_or_default()).map(|_| json!({"type": "ok"}))), + "open" => r((|| { + let mut file = std::fs::File::create(path)?; + file.write_all(b["text"].as_str().unwrap_or_default().as_bytes())?; + open.insert(path.into(), file); 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); + "append" => r((|| { + let mut file = std::fs::OpenOptions::new().append(true).open(path)?; + file.write_all(b["text"].as_str().unwrap_or_default().as_bytes())?; + open.insert(path.into(), file); 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| { + "close" => json!({"type": "ok", "closed": open.remove(path).is_some()}), + "read" => r(std::fs::read_to_string(path).map(|text| json!({"type": "content", "text": text}))), + "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"), + "remove" => r(std::fs::remove_file(path).map(|_| json!({"type": "ok"}))), + "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(); + let _ = reply(&msg, response); } } diff --git a/examples/notes-v2/Cargo.toml b/examples/notes-v2/Cargo.toml index 2b0aa5f..3ca0202 100644 --- a/examples/notes-v2/Cargo.toml +++ b/examples/notes-v2/Cargo.toml @@ -4,4 +4,7 @@ version = "0.1.0" edition = "2024" [dependencies] +sausage-actor = { version = "0.1", path = "../../sdk" } serde_json = "1" + +[workspace] diff --git a/examples/notes-v2/src/main.rs b/examples/notes-v2/src/main.rs index 4d539d5..3952b6e 100644 --- a/examples/notes-v2/src/main.rs +++ b/examples/notes-v2/src/main.rs @@ -1,7 +1,5 @@ -//! 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}; +//! Version 2 adds a note count and fixes the crash command. +use sausage_actor::{json, messages, reply}; const DATA: &str = "/data/notes.json"; @@ -13,27 +11,21 @@ fn save(notes: &[String]) { fn main() { let mut notes: Vec = std::fs::read_to_string(DATA) .ok() - .and_then(|s| serde_json::from_str(&s).ok()) + .and_then(|text| serde_json::from_str(&text).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") => { + for message in messages() { + let body = &message.body; + let response = match message.kind() { + "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}), + "list" => json!({"type": "notes", "items": notes}), + "count" => json!({"type": "count", "count": notes.len(), "version": 2}), + "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(); + let _ = reply(&message, response); } } diff --git a/examples/notes/Cargo.toml b/examples/notes/Cargo.toml index 2b0aa5f..3ca0202 100644 --- a/examples/notes/Cargo.toml +++ b/examples/notes/Cargo.toml @@ -4,4 +4,7 @@ version = "0.1.0" edition = "2024" [dependencies] +sausage-actor = { version = "0.1", path = "../../sdk" } serde_json = "1" + +[workspace] diff --git a/examples/notes/src/main.rs b/examples/notes/src/main.rs index e63137d..a695596 100644 --- a/examples/notes/src/main.rs +++ b/examples/notes/src/main.rs @@ -1,7 +1,5 @@ -//! Keeps a list of notes in /data/notes.json and publishes them as -//! /public/notes.txt. -use serde_json::{Value, json}; -use std::io::{BufRead, Write}; +//! Keep notes in `/data/notes.json` and publish them at `/public/notes.txt`. +use sausage_actor::{json, messages, reply}; const DATA: &str = "/data/notes.json"; @@ -13,25 +11,20 @@ fn save(notes: &[String]) { fn main() { let mut notes: Vec = std::fs::read_to_string(DATA) .ok() - .and_then(|s| serde_json::from_str(&s).ok()) + .and_then(|text| serde_json::from_str(&text).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") => { + for message in messages() { + let body = &message.body; + let response = match message.kind() { + "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("crash") => panic!("asked to crash"), + "list" => json!({"type": "notes", "items": notes}), + "crash" => panic!("asked to crash"), _ => json!({"type": "error", "reason": "unknown message"}), }; - writeln!(out, "{}", json!({"to": msg["from"], "reply_to": msg["id"], "body": reply})).unwrap(); - out.flush().unwrap(); + let _ = reply(&message, response); } } diff --git a/examples/ticker/Cargo.toml b/examples/ticker/Cargo.toml index dea727b..98760a4 100644 --- a/examples/ticker/Cargo.toml +++ b/examples/ticker/Cargo.toml @@ -4,4 +4,6 @@ version = "0.1.0" edition = "2024" [dependencies] -serde_json = "1" +sausage-actor = { version = "0.1", path = "../../sdk" } + +[workspace] diff --git a/examples/ticker/src/main.rs b/examples/ticker/src/main.rs index 2aeef21..758f390 100644 --- a/examples/ticker/src/main.rs +++ b/examples/ticker/src/main.rs @@ -1,16 +1,13 @@ -//! 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; +//! Send 200 notes to `notes`, 50 ms apart, then stay alive. +use sausage_actor::{json, send}; +use std::time::Duration; + 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)); + let _ = send("notes", json!({"type": "add", "text": format!("tick {n}")})); + std::thread::sleep(Duration::from_millis(50)); } - // Stay alive without doing anything. loop { - std::thread::sleep(std::time::Duration::from_secs(3600)); + std::thread::sleep(Duration::from_secs(3600)); } } diff --git a/sdk/Cargo.toml b/sdk/Cargo.toml index 7664bf0..a72480c 100644 --- a/sdk/Cargo.toml +++ b/sdk/Cargo.toml @@ -4,4 +4,12 @@ version = "0.1.0" edition = "2024" description = "SDK for Sausage actors." +[features] +default = [] +async = ["dep:async-channel", "dep:wstd"] + [dependencies] +async-channel = { version = "2", optional = true } +serde = { version = "1", features = ["derive"] } +serde_json = "1" +wstd = { version = "0.6", optional = true } diff --git a/sdk/src/lib.rs b/sdk/src/lib.rs index 78ef62c..bca225e 100644 --- a/sdk/src/lib.rs +++ b/sdk/src/lib.rs @@ -1 +1,203 @@ -//! SDK for Sausage actors (placeholder). +//! Small runtime helpers for Sausage actors. + +pub use serde_json::{Value, json}; + +use serde::{Deserialize, Serialize}; +use std::io::BufRead; + +/// A message that an actor receives from the broker. +#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)] +pub struct Message { + /// The message ID. + pub id: String, + /// The ID of the sending actor. + pub from: String, + /// The ID of the message that this message answers. + #[serde(skip_serializing_if = "Option::is_none")] + pub reply_to: Option, + /// The message body. + pub body: Value, +} + +impl Message { + /// Return the body's `type` string, or an empty string when it is missing. + pub fn kind(&self) -> &str { + self.body["type"].as_str().unwrap_or("") + } +} + +/// 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() +} + +/// Read messages from stdin until EOF or a shutdown message from the broker. +/// +/// Invalid JSON lines are skipped. +pub fn messages() -> impl Iterator { + let stdin = std::io::stdin(); + 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 { + continue; + }; + if message.from == "broker" && message.kind() == "shutdown" { + return None; + } + return Some(message); + } + }) +} + +/// Send a message to `to` and return its full ID (`/`). +pub fn send(to: &str, body: Value) -> String { + let id = next_id(); + write_message(json!({"to": to, "id": id, "body": body})); + format!("{}/{id}", me()) +} + +/// Reply to `message` and return the new message's full ID. +pub fn reply(message: &Message, body: Value) -> String { + let id = next_id(); + write_message(json!({ + "to": message.from, + "id": id, + "reply_to": message.id, + "body": body, + })); + format!("{}/{id}", me()) +} + +/// Publish a message to this actor's subscribers and return its full ID. +pub fn publish(body: Value) -> String { + let id = next_id(); + write_message(json!({"id": id, "body": body})); + format!("{}/{id}", me()) +} + +fn write_message(message: Value) { + use std::io::Write; + + let mut stdout = std::io::stdout().lock(); + writeln!(stdout, "{message}").unwrap(); + stdout.flush().unwrap(); +} + +fn next_id() -> String { + use std::sync::OnceLock; + use std::sync::atomic::{AtomicUsize, Ordering}; + use std::time::{SystemTime, UNIX_EPOCH}; + + static START: OnceLock = OnceLock::new(); + static SEQUENCE: AtomicUsize = AtomicUsize::new(0); + let start = START.get_or_init(|| { + SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap_or_default() + .as_nanos() + }); + format!("{start}-{}", SEQUENCE.fetch_add(1, Ordering::Relaxed)) +} + +/// An asynchronous inbox for actors that use HTTP or timers. +#[cfg(feature = "async")] +pub struct Inbox { + rx: async_channel::Receiver, +} + +#[cfg(feature = "async")] +impl Inbox { + /// Start a task that reads stdin and returns an inbox for its messages. + pub fn start() -> Self { + let (tx, rx) = async_channel::unbounded(); + wstd::runtime::spawn(read_messages(tx)).detach(); + Self { rx } + } + + /// Wait for the next message, or return `None` at EOF or shutdown. + pub async fn recv(&self) -> Option { + self.rx.recv().await.ok() + } + + /// Return the next available message without waiting. + pub fn try_recv(&self) -> Option { + self.rx.try_recv().ok() + } +} + +#[cfg(feature = "async")] +async fn read_messages(tx: async_channel::Sender) { + use wstd::io::AsyncRead; + + let mut stdin = wstd::io::stdin(); + let mut buf = Vec::new(); + let mut chunk = [0; 8192]; + loop { + let Ok(n) = stdin.read(&mut chunk).await else { + break; + }; + if n == 0 { + break; + } + 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 { + continue; + }; + if message.from == "broker" && message.kind() == "shutdown" { + tx.close(); + return; + } + if tx.send(message).await.is_err() { + return; + } + } + } + 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(); +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn kind_reads_body_type() { + let message: Message = serde_json::from_value(json!({ + "id": "broker/1", + "from": "broker", + "body": {"type": "ping"} + })) + .unwrap(); + assert_eq!(message.kind(), "ping"); + assert_eq!(message.reply_to, None); + } + + #[test] + fn kind_is_empty_when_type_is_missing() { + let message: Message = serde_json::from_value(json!({ + "id": "broker/1", + "from": "broker", + "body": {} + })) + .unwrap(); + assert_eq!(message.kind(), ""); + } + + #[test] + fn generated_ids_differ() { + assert_ne!(next_id(), next_id()); + } +}