surma/sausage
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

SDK sausage-actor; examples and agent on the SDK; agent without limits

Browse Source 
- 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)
This commit is contained in:
surma
2026-10-15 11:17:14 +01:00
parent c9d9864743
commit 7f76eab161
22 changed files with 607 additions and 518 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Cargo.lock
Generated
+7 -1
View File
@@ -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"
agent/Cargo.toml
+1 -1
View File
@@ -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"
agent/src/llm.rs
+8 -15
View File
@@ -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<String>,
#[serde(default = "default_max_tokens")]
pub max_tokens: u32,
}
fn default_max_tokens() -> u32 {
16000
#[serde(default)]
pub max_tokens: Option<u32>,
}
#[derive(Serialize, Deserialize, Clone, Debug)]
@@ -57,17 +52,13 @@ async fn post(url: &str, headers: &[(&str, String)], body: &Value) -> Result<Val
req = req.header(*k, v.as_str());
}
let req = req.body(Body::from(body.to_string())).map_err(|e| format!("request: {e}"))?;
let mut client = Client::new();
client.set_first_byte_timeout(Duration::from_secs(600));
client.set_between_bytes_timeout(Duration::from_secs(600));
let client = Client::new();
let resp = client.send(req).await.map_err(|e| format!("HTTP request failed: {e:?}"))?;
let status = resp.status();
let mut body = resp.into_body();
let text = body.str_contents().await.map_err(|e| format!("reading the response failed: {e:?}"))?.to_string();
if !status.is_success() {
let mut t = text;
t.truncate(2000);
return Err(format!("HTTP {status}: {t}"));
return Err(format!("HTTP {status}: {text}"));
}
serde_json::from_str(&text).map_err(|e| format!("invalid JSON in response: {e}"))
}
@@ -99,8 +90,10 @@ async fn openai(cfg: &Config, system: &str, tools: &[Tool], history: &[Entry]) -
"tools": tools,
"store": false,
"include": ["reasoning.encrypted_content"],
"max_output_tokens": cfg.max_tokens,
});
if let Some(max_tokens) = cfg.max_tokens {
body["max_output_tokens"] = json!(max_tokens);
}
if let Some(effort) = &cfg.reasoning_effort {
body["reasoning"] = json!({"effort": effort});
}
@@ -170,7 +163,7 @@ async fn anthropic(cfg: &Config, system: &str, tools: &[Tool], history: &[Entry]
"system": system,
"messages": messages,
"tools": tools,
"max_tokens": cfg.max_tokens,
"max_tokens": cfg.max_tokens.unwrap_or(32000),
});
let url = format!("{}/v1/messages", cfg.base_url);
let resp = post(
agent/src/main.rs
+186 -246
View File
@@ -1,85 +1,33 @@
//! The Sausage agent: a normal WASM actor that does inference over HTTP and
//! grows the world through messages to `cargo` and `broker`.
//!
//! Configuration: `/data/config.json` (see `llm::Config`).
//! Conversation: `/data/history.json` (one shared conversation).
//! The Sausage agent actor.
mod llm;
use std::collections::VecDeque;
use async_channel::{Receiver, Sender};
use llm::{Config, Entry, Tool};
use serde_json::{Value, json};
use sausage_actor::{Inbox, Message, Value, json, reply, send};
use std::collections::{HashSet, VecDeque};
use std::io::Write;
use wstd::future::FutureExt;
use wstd::io::{AsyncRead, AsyncWrite};
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 HISTORY: &str = "/data/history.jsonl";
const TURN: &str = "/data/turn.json";
const MAX_STEPS: usize = 40;
const MAX_HISTORY: usize = 600;
const MAX_TOOL_OUTPUT: usize = 30_000;
const MAX_NOTES: usize = 50;
fn main() {
wstd::runtime::block_on(async {
let (tx, rx) = async_channel::unbounded();
wstd::runtime::spawn(read_stdin(tx)).detach();
Agent::new(rx).run().await;
});
}
/// Read NDJSON messages from stdin into the channel.
async fn read_stdin(tx: Sender<Value>) {
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<u8> = buf.drain(..=pos).collect();
if let Ok(v) = serde_json::from_slice::<Value>(&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<Tool> {
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> {
},
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<Tool> {
"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> {
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> {
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<u64> {
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<u64> {
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<Value>,
stdout: wstd::io::Stdout,
inbox: Inbox,
history: Vec<Entry>,
/// Text messages that start a new turn later.
queue: VecDeque<Value>,
/// Other messages that arrived while the agent was idle.
notes: Vec<Value>,
seq: u64,
queue: VecDeque<Message>,
notes: Vec<Message>,
shutdown: bool,
}
impl Agent {
fn new(rx: Receiver<Value>) -> 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::<Value>(&s).ok()) else {
let Ok(turn) = std::fs::read_to_string(TURN) else {
return;
};
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);
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
let answered: HashSet<String> = 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<String> = 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<Value>) {
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<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"));
let notes = self
.notes
.drain(..)
.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}");
}
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<String> = 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<Value>) {
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<Value> = 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()
}
agent/src/prompt.md
+47 -63
View File
@@ -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 `"<sender>/<id>"`. 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/<id>/public`: the public files of every actor, read-only.
- `/actor/<id>/source`: the source of the running version of every actor, read-only.
- `/actor/<id>/status.json` and `/actor/<id>/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":"<new-id>","wasm":"<path>","source":"<folder>","built_by":"cargo"}`
- `{"type":"upgrade","actor":"<id>","wasm":"<path>","source":"<folder>","built_by":"cargo"}`
- `{"type":"rollback","actor":"<id>"}` (optional `"version"`)
- `{"type":"stop","actor":"<id>"}`, `{"type":"start","actor":"<id>"}`, `{"type":"remove","actor":"<id>"}` (archives, never deletes)
- `{"type":"subscribe","actor":"<id>"}`, `{"type":"unsubscribe","actor":"<id>"}`, `{"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 `<source>/tests/*.jsonl`. Each line is `{"send": <body>}` or `{"expect": <subset of the reply body>}` (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":"<folder with Cargo.toml>"}` → `{"type":"built","wasm":"/actor/cargo/public/builds/<n>/<name>.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/<name>/`: `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/<name>` 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::<Value>(&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/<name>/`. 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/<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.
## Filesystem
Your private state lives under `/data`. Your public files live under `/public`. Other actors' `data/` directories are hidden.
The host will provide `/actor/<id>/` with `actor.wasm`, `actor.json`, `source/`, `public/`, `status.json`, and `stderr.log`. Each `versions/<n>/` 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`.
examples/bench/Cargo.toml
+3 -1
View File
@@ -4,4 +4,6 @@ version = "0.1.0"
edition = "2024"
[dependencies]
serde_json = "1"
sausage-actor = { version = "0.1", path = "../../sdk" }
[workspace]
examples/bench/src/main.rs
+32 -39
View File
@@ -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::<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();
}
_ => {}
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)
}),
);
}
}
examples/crasher/Cargo.toml
+2
View File
@@ -4,3 +4,5 @@ version = "0.1.0"
edition = "2024"
[dependencies]
[workspace]
examples/echo/Cargo.toml
+3 -1
View File
@@ -4,4 +4,6 @@ version = "0.1.0"
edition = "2024"
[dependencies]
serde_json = "1"
sausage-actor = { version = "0.1", path = "../../sdk" }
[workspace]
examples/echo/src/main.rs
+3 -11
View File
@@ -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::<Value>(&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}));
}
}
examples/fetch/Cargo.toml
+3 -1
View File
@@ -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]
examples/fetch/src/main.rs
+37 -57
View File
@@ -1,77 +1,57 @@
//! Fetches URLs over HTTP(S) and writes the body to /public/fetch/<n>.
//! 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<RefCell<Vec<String>>>;
async fn flush(out: &Out) {
let lines: Vec<String> = 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<u8> = buf.drain(..=pos).collect();
let Ok(msg) = serde_json::from_slice::<Value>(&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"}));
}
_ => {}
}
}
});
examples/fstest/Cargo.toml
+3 -1
View File
@@ -4,4 +4,6 @@ version = "0.1.0"
edition = "2024"
[dependencies]
serde_json = "1"
sausage-actor = { version = "0.1", path = "../../sdk" }
[workspace]
examples/fstest/src/main.rs
+24 -33
View File
@@ -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<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);
let mut open = HashMap::<String, std::fs::File>::new();
for msg in messages() {
let b = &msg.body;
let path = b["path"].as_str().unwrap_or_default();
let r = |result: std::io::Result<Value>| 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<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"),
"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);
}
}
examples/notes-v2/Cargo.toml
+3
View File
@@ -4,4 +4,7 @@ version = "0.1.0"
edition = "2024"
[dependencies]
sausage-actor = { version = "0.1", path = "../../sdk" }
serde_json = "1"
[workspace]
examples/notes-v2/src/main.rs
+11 -19
View File
@@ -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<String> = 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::<Value>(&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(&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}),
"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);
}
}
examples/notes/Cargo.toml
+3
View File
@@ -4,4 +4,7 @@ version = "0.1.0"
edition = "2024"
[dependencies]
sausage-actor = { version = "0.1", path = "../../sdk" }
serde_json = "1"
[workspace]
examples/notes/src/main.rs
+10 -17
View File
@@ -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<String> = 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::<Value>(&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(&notes);
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);
}
}
examples/ticker/Cargo.toml
+3 -1
View File
@@ -4,4 +4,6 @@ version = "0.1.0"
edition = "2024"
[dependencies]
serde_json = "1"
sausage-actor = { version = "0.1", path = "../../sdk" }
[workspace]
examples/ticker/src/main.rs
+7 -10
View File
@@ -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));
}
}
sdk/Cargo.toml
+8
View File
@@ -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 }
sdk/src/lib.rs
+203 -1
View File
@@ -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<String>,
/// 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<Item = Message> {
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::<Message>(&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 (`<actor>/<own-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<u128> = 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<Message>,
}
#[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<Message> {
self.rx.recv().await.ok()
}
/// Return the next available message without waiting.
pub fn try_recv(&self) -> Option<Message> {
self.rx.try_recv().ok()
}
}
#[cfg(feature = "async")]
async fn read_messages(tx: async_channel::Sender<Message>) {
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::<Message>(&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::<Message>(&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());
}
}