diff --git a/Cargo.toml b/Cargo.toml index aeb3e8a..f8c8f48 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,9 +1,9 @@ [workspace] resolver = "3" -members = ["cli", "vfs", "interface", "sdk", "sdk/macros"] +members = ["cli", "vfs", "interface", "sdk", "sdk/macros", "agent"] default-members = ["cli"] -# The agent is rewritten on top of the new SDK. Until then it is not a member. -exclude = ["examples", "agent"] +# Examples are standalone plugins, each with its own empty [workspace]. +exclude = ["examples"] # Wasmtime compiles components with Cranelift. Optimize dependencies in dev # builds, so that tests and debug runs start plugins quickly. diff --git a/agent/Cargo.toml b/agent/Cargo.toml index 0700110..3f65ca2 100644 --- a/agent/Cargo.toml +++ b/agent/Cargo.toml @@ -2,10 +2,12 @@ name = "agent" version = "0.1.0" edition = "2024" -description = "The Sausage agent actor." +description = "The Sausage agent plugin." [dependencies] -sausage-actor = { version = "0.1", path = "../sdk", features = ["async"] } +sausage-plugin = { version = "0.1", path = "../sdk" } serde = { version = "1", features = ["derive"] } serde_json = "1" + +[target.'cfg(target_os = "wasi")'.dependencies] wstd = "0.6" diff --git a/agent/src/history.rs b/agent/src/history.rs index 5ee53a5..afabed8 100644 --- a/agent/src/history.rs +++ b/agent/src/history.rs @@ -2,6 +2,7 @@ use crate::llm::Entry; +/// Recognize provider errors that permit a retry with fewer complete turns. pub fn is_too_large(error: &str) -> bool { let Some((status, body)) = error .strip_prefix("HTTP ") @@ -23,6 +24,7 @@ pub fn is_too_large(error: &str) -> bool { .any(|term| body.contains(term))) } +/// Find a user-turn boundary without discarding any of the current turn. pub fn next_turn_start(history: &[Entry], from: usize, current: usize) -> Option { let end = current.min(history.len().checked_sub(1)?); (from..=end).find(|&index| matches!(&history[index], Entry::User { .. })) diff --git a/agent/src/llm.rs b/agent/src/llm.rs index 83a93ac..49b3497 100644 --- a/agent/src/llm.rs +++ b/agent/src/llm.rs @@ -5,18 +5,24 @@ use serde::{Deserialize, Serialize}; use serde_json::{Value, json}; +#[cfg(target_os = "wasi")] use wstd::http::{Body, Client, Request}; +/// Inference configuration from `/config/agent.json`. #[derive(Deserialize, Clone)] pub struct Config { /// `openai-responses` or `anthropic-messages`. pub api: String, + /// The API origin, without its versioned endpoint path. pub base_url: String, + /// The provider's model identifier. pub model: String, + /// The API credential sent in the provider's authentication header. pub token: String, /// Optional reasoning effort for OpenAI models only (wins over `thinking` there). #[serde(default)] pub reasoning_effort: Option, + /// An optional cap on generated tokens. #[serde(default)] pub max_tokens: Option, /// Thinking effort for both APIs: "low", "medium", "high", "xhigh", "max" (`LEVELS`), or "off"/unset for none. @@ -32,8 +38,17 @@ pub const LEVELS: [&str; 5] = ["low", "medium", "high", "xhigh", "max"]; /// The level nearest to `want` (lower on a tie) among those a 400 error lists in quotes (else all levels), not /// yet `tried`. None: nothing left, so the request goes out without an effort field (model default). fn fallback_effort(want: &str, error: &str, tried: &[String]) -> Option { - let listed: Vec<&str> = error.split('\'').skip(1).step_by(2).filter(|w| LEVELS.contains(w)).collect(); - let pool = if listed.iter().any(|l| *l != want) { listed } else { LEVELS.to_vec() }; + let listed: Vec<&str> = error + .split('\'') + .skip(1) + .step_by(2) + .filter(|w| LEVELS.contains(w)) + .collect(); + let pool = if listed.iter().any(|l| *l != want) { + listed + } else { + LEVELS.to_vec() + }; let at = LEVELS.iter().position(|l| *l == want).unwrap_or(1) as i32; pool.into_iter() .filter(|l| *l != want && !tried.iter().any(|t| t == l)) @@ -50,39 +65,90 @@ mod tests { #[test] fn nearest_accepted_level() { - let anthropic = "HTTP 400: output_config.effort: Input should be 'low', 'medium', 'high' or 'max'"; - assert_eq!(fallback_effort("xhigh", anthropic, &["xhigh".into()]).as_deref(), Some("high")); + let anthropic = + "HTTP 400: output_config.effort: Input should be 'low', 'medium', 'high' or 'max'"; + assert_eq!( + fallback_effort("xhigh", anthropic, &["xhigh".into()]).as_deref(), + Some("high") + ); let openai = "HTTP 400: Invalid value: 'ultra'. Supported values are: 'none', 'minimal', 'low', 'medium', 'high', 'xhigh', and 'max'."; - assert_eq!(fallback_effort("ultra", openai, &["ultra".into()]).as_deref(), Some("medium")); + assert_eq!( + fallback_effort("ultra", openai, &["ultra".into()]).as_deref(), + Some("medium") + ); // No list in the error: step to the neighbour, never to off. - assert_eq!(fallback_effort("max", "HTTP 400: effort max not supported", &["max".into()]).as_deref(), Some("xhigh")); + assert_eq!( + fallback_effort("max", "HTTP 400: effort max not supported", &["max".into()]) + .as_deref(), + Some("xhigh") + ); } } +/// One append-only entry in the provider-neutral conversation. #[derive(Serialize, Deserialize, Clone, Debug)] #[serde(tag = "kind", rename_all = "snake_case")] pub enum Entry { - User { text: String }, - Assistant { + /// A person's contribution starts a new turn. + User { + /// The complete contribution. text: String, + }, + /// Text produced by the model. + Assistant { + /// The generated text. + text: String, + /// The optional provider phase, preserved for subsequent requests. #[serde(default, skip_serializing_if = "Option::is_none")] phase: Option, }, - ToolCall { id: String, name: String, arguments: String }, - ToolResult { id: String, output: String }, + /// A requested tool operation. + ToolCall { + /// The provider's correlation identifier. + id: String, + /// The tool name. + name: String, + /// The JSON parameter object as text. + arguments: String, + }, + /// The outcome of a tool operation, including tool errors. + ToolResult { + /// The correlation identifier of the tool call. + id: String, + /// The JSON outcome as text. + output: String, + }, /// A provider item that only this API shape understands (OpenAI reasoning, Anthropic thinking). - Opaque { api: String, item: Value }, + Opaque { + /// The API shape that owns this entry. + api: String, + /// The original provider object. + item: Value, + }, } +/// A tool contract shared by both API shapes. pub struct Tool { + /// The function identifier exposed to the model. pub name: &'static str, + /// The explanation of the tool's behavior. pub description: &'static str, + /// The JSON Schema for named parameters. pub parameters: Value, } -pub async fn complete(cfg: &Config, system: &str, tools: &[Tool], history: &[Entry]) -> Result, String> { +/// Request the next model step, with provider-specific reasoning items intact. +pub async fn complete( + cfg: &Config, + system: &str, + tools: &[Tool], + history: &[Entry], +) -> Result, String> { let want = match cfg.api.as_str() { - "openai-responses" => cfg.reasoning_effort.clone().or_else(|| cfg.thinking.clone()), + "openai-responses" => cfg + .reasoning_effort + .clone() + .or_else(|| cfg.thinking.clone()), _ => cfg.thinking.clone(), } .filter(|t| t != "off"); @@ -91,7 +157,17 @@ pub async fn complete(cfg: &Config, system: &str, tools: &[Tool], history: &[Ent loop { let result = match cfg.api.as_str() { "openai-responses" => openai(cfg, system, tools, history, effort.as_deref()).await, - "anthropic-messages" => anthropic(cfg, system, tools, history, want.is_some(), effort.as_deref()).await, + "anthropic-messages" => { + anthropic( + cfg, + system, + tools, + history, + want.is_some(), + effort.as_deref(), + ) + .await + } other => return Err(format!("unknown API shape {other:?}")), }; match (result, effort.clone()) { @@ -99,31 +175,58 @@ pub async fn complete(cfg: &Config, system: &str, tools: &[Tool], history: &[Ent (Err(e), Some(current)) if e.starts_with("HTTP 400") && e.contains("effort") => { tried.push(current.clone()); effort = fallback_effort(want.as_deref().unwrap_or(""), &e, &tried); - eprintln!("agent: {} rejected thinking effort {current:?}; retrying with {effort:?}: {e}", cfg.model); + eprintln!( + "agent: {} rejected thinking effort {current:?}; retrying with {effort:?}: {e}", + cfg.model + ); } (result, _) => return result, } } } +/// Native unit tests exercise history logic without an HTTP runtime. +#[cfg(not(target_os = "wasi"))] +async fn post(_url: &str, _headers: &[(&str, String)], _body: &Value) -> Result { + Err("inference requires the WASI HTTP runtime".into()) +} + +/// Send JSON through wasi:http and retain error bodies for context-size detection. +#[cfg(target_os = "wasi")] async fn post(url: &str, headers: &[(&str, String)], body: &Value) -> Result { let mut req = Request::post(url).header("content-type", "application/json"); for (k, v) in headers { req = req.header(*k, v.as_str()); } - let req = req.body(Body::from(body.to_string())).map_err(|e| format!("request: {e}"))?; + let req = req + .body(Body::from(body.to_string())) + .map_err(|e| format!("request: {e}"))?; let client = Client::new(); - let resp = client.send(req).await.map_err(|e| format!("HTTP request failed: {e:?}"))?; + 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(); + let text = body + .str_contents() + .await + .map_err(|e| format!("reading the response failed: {e:?}"))? + .to_string(); if !status.is_success() { return Err(format!("HTTP {status}: {text}")); } serde_json::from_str(&text).map_err(|e| format!("invalid JSON in response: {e}")) } -async fn openai(cfg: &Config, system: &str, tools: &[Tool], history: &[Entry], effort: Option<&str>) -> Result, String> { +/// Translate neutral history to OpenAI Responses and preserve returned reasoning items. +async fn openai( + cfg: &Config, + system: &str, + tools: &[Tool], + history: &[Entry], + effort: Option<&str>, +) -> Result, String> { let mut input = Vec::new(); for e in history { match e { @@ -164,7 +267,12 @@ async fn openai(cfg: &Config, system: &str, tools: &[Tool], history: &[Entry], e body["reasoning"] = json!({"effort": effort}); } let url = format!("{}/v1/responses", cfg.base_url); - let resp = post(&url, &[("authorization", format!("Bearer {}", cfg.token))], &body).await?; + let resp = post( + &url, + &[("authorization", format!("Bearer {}", cfg.token))], + &body, + ) + .await?; let mut out = Vec::new(); for item in resp["output"].as_array().cloned().unwrap_or_default() { match item["type"].as_str() { @@ -189,16 +297,23 @@ async fn openai(cfg: &Config, system: &str, tools: &[Tool], history: &[Entry], e name: item["name"].as_str().unwrap_or_default().to_string(), arguments: item["arguments"].as_str().unwrap_or("{}").to_string(), }), - Some("reasoning") => out.push(Entry::Opaque { api: "openai-responses".into(), item }), + Some("reasoning") => out.push(Entry::Opaque { + api: "openai-responses".into(), + item, + }), _ => {} } } if resp["status"] == "incomplete" && out.is_empty() { - return Err(format!("incomplete response: {}", resp["incomplete_details"])); + return Err(format!( + "incomplete response: {}", + resp["incomplete_details"] + )); } Ok(out) } +/// Translate neutral history to Anthropic Messages and merge consecutive roles. async fn anthropic( cfg: &Config, system: &str, @@ -219,22 +334,37 @@ async fn anthropic( Entry::Assistant { text, .. } => { push("assistant", json!({"type": "text", "text": text})) } - Entry::ToolCall { id, name, arguments } => { + Entry::ToolCall { + id, + name, + arguments, + } => { let input: Value = serde_json::from_str(arguments).unwrap_or(json!({})); - push("assistant", json!({"type": "tool_use", "id": id, "name": name, "input": input})) - } - Entry::ToolResult { id, output } => { - push("user", json!({"type": "tool_result", "tool_use_id": id, "content": output})) + push( + "assistant", + json!({"type": "tool_use", "id": id, "name": name, "input": input}), + ) } + Entry::ToolResult { id, output } => push( + "user", + json!({"type": "tool_result", "tool_use_id": id, "content": output}), + ), // Thinking blocks go back unchanged (with their signature): required next to tool use. - Entry::Opaque { api, item } if api == "anthropic-messages" && thinking => push("assistant", item.clone()), + Entry::Opaque { api, item } if api == "anthropic-messages" && thinking => { + push("assistant", item.clone()) + } Entry::Opaque { .. } => {} } } - let messages: Vec = messages.into_iter().map(|(role, content)| json!({"role": role, "content": content})).collect(); + let messages: Vec = messages + .into_iter() + .map(|(role, content)| json!({"role": role, "content": content})) + .collect(); let tools: Vec = tools .iter() - .map(|t| json!({"name": t.name, "description": t.description, "input_schema": t.parameters})) + .map( + |t| json!({"name": t.name, "description": t.description, "input_schema": t.parameters}), + ) .collect(); let mut body = json!({ "model": cfg.model, @@ -252,7 +382,10 @@ async fn anthropic( let url = format!("{}/v1/messages", cfg.base_url); let resp = post( &url, - &[("x-api-key", cfg.token.clone()), ("anthropic-version", "2023-06-01".to_string())], + &[ + ("x-api-key", cfg.token.clone()), + ("anthropic-version", "2023-06-01".to_string()), + ], &body, ) .await?; @@ -270,9 +403,10 @@ async fn anthropic( name: block["name"].as_str().unwrap_or_default().to_string(), arguments: block["input"].to_string(), }), - Some("thinking") | Some("redacted_thinking") => { - out.push(Entry::Opaque { api: "anthropic-messages".into(), item: block }) - } + Some("thinking") | Some("redacted_thinking") => out.push(Entry::Opaque { + api: "anthropic-messages".into(), + item: block, + }), _ => {} } } diff --git a/agent/src/main.rs b/agent/src/main.rs index 102def1..af72fb1 100644 --- a/agent/src/main.rs +++ b/agent/src/main.rs @@ -1,339 +1,156 @@ -//! The Sausage agent actor. - +//! Answer chat calls with an LLM and tools, under one lock for the complete turn. mod history; mod llm; mod tools; use history::{is_too_large, next_turn_start}; use llm::{Config, Entry}; -use sausage_actor::{Inbox, Message, Value, json, publish, send}; -use std::collections::{HashSet, VecDeque}; +use sausage_plugin::{Ctx, Error, Result, lock, rpc, schema}; +use serde_json::json; use std::io::Write; -use wstd::future::FutureExt; -use wstd::time::{Duration, Instant}; +/// The instructions sent with every inference request. const SYSTEM: &str = include_str!("prompt.md"); -const CONFIG: &str = "/data/config.json"; -const HISTORY: &str = "/data/history.jsonl"; -const HISTORY_SKIP: &str = "/data/history.skip"; -const TURN: &str = "/data/turn.json"; - -fn main() { - wstd::runtime::block_on(async { Agent::new().run().await }); -} - -struct Agent { - inbox: Inbox, - history: Vec, - history_skip: usize, - active: bool, - queue: VecDeque, - notes: Vec, +/// The shared, append-only conversation. +const HISTORY: &str = "/state/agent/history.jsonl"; +/// The first retained turn after a context-size rejection. +const SKIP: &str = "/state/agent/history.skip"; +/// Each chat process reads the shared history after it acquires the lock. +struct Agent; +/// The person's next contribution to the conversation. +#[schema] +struct ChatParams { + /// The complete text of the person's request. + text: String, } +#[rpc(grace_seconds = 600)] impl Agent { - fn new() -> Self { - let history = std::fs::read_to_string(HISTORY) - .unwrap_or_default() - .lines() - .filter_map(|line| serde_json::from_str(line).ok()) - .collect(); - let history_skip = std::fs::read_to_string(HISTORY_SKIP) + /// Answer a person, with interim text notifications and a final text result. + /// + /// The history lock covers inference and tools, so concurrent chats cannot interleave turns. + async fn chat(&self, ctx: &Ctx, p: ChatParams) -> Result { + let _guard = lock(HISTORY).await?; + std::fs::create_dir_all("/state/agent")?; + let cfg: Config = serde_json::from_slice(&std::fs::read("/config/agent.json")?)?; + let mut entries = load_history()?; + repair_tools(&mut entries)?; + let current = entries.len(); + append(&mut entries, Entry::User { text: p.text })?; + let mut start = std::fs::read_to_string(SKIP) .ok() - .and_then(|skip| skip.trim().parse::().ok()) + .and_then(|s| s.trim().parse().ok()) .unwrap_or(0); - let active = std::fs::read_to_string(TURN) - .ok() - .and_then(|contents| serde_json::from_str::(&contents).ok()) - .unwrap_or(false); - Self { - inbox: Inbox::start(), - history, - history_skip, - active, - queue: VecDeque::new(), - notes: Vec::new(), - } - } - - 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); - } - - fn finish_turn(&mut self) { - self.active = false; - let _ = std::fs::remove_file(TURN); - } - - fn closing(&self) -> bool { - self.inbox.is_closed() - } - - fn repair_dangling_tools(&mut self) { - let answered: HashSet = self - .history - .iter() - .filter_map(|entry| match entry { - Entry::ToolResult { id, .. } => Some(id.clone()), - _ => None, - }) - .collect(); - let dangling: Vec = self - .history - .iter() - .filter_map(|entry| match entry { - Entry::ToolCall { id, .. } if !answered.contains(id) => Some(id.clone()), - _ => None, - }) - .collect(); - for id in dangling { - self.add_history(Entry::ToolResult { - id, - output: - json!({"error": "interrupted: the agent restarted before this tool finished"}) - .to_string(), - }); - } - } - - fn turn_text(&self, message: &Message) -> String { - let mut text = format!( - "Message {} from {}:\n{}", - message.id, - message.from, - message.body["text"].as_str().unwrap_or("") - ); - if !self.notes.is_empty() { - let notes = self - .notes - .iter() - .map(|note| serde_json::to_string(note).unwrap()) - .collect::>() - .join("\n"); - text = format!("Messages that arrived while you were idle:\n{notes}\n\n{text}"); - } - text - } - - fn not_configured(&mut self, error: String) { - let _ = publish(json!({ - "type": "text", - "text": format!("I am not configured: cannot read {CONFIG}: {error}") - })); - self.finish_turn(); - } - - async fn run(mut self) { - self.repair_dangling_tools(); - if self.active { - self.turn(None).await; - if self.closing() && self.active { - return; - } - } + start = next_turn_start(&entries, start, current).unwrap_or(current); + let definitions = tools::definitions(); loop { - let message = match self.queue.pop_front() { - Some(message) => message, - None => match self.inbox.recv().await { - Some(message) => message, - None => break, - }, - }; - if starts_turn(&message) { - self.turn(Some(message)).await; - if self.closing() && self.active { - break; - } - } else { - self.notes.push(message); - } - } - } - - async fn turn(&mut self, message: Option) { - let current_start = if message.is_some() { - self.active = true; - let _ = std::fs::write(TURN, "true"); - self.history.len() - } else { - let current_start = self - .history - .iter() - .rposition(|entry| matches!(entry, Entry::User { text } if !text.starts_with("[note from your runtime]"))) - .unwrap_or(self.history.len()); - let version = std::env::var("SAUSAGE_VERSION").unwrap_or_default(); - self.add_history(Entry::User { - text: format!( - "[note from your runtime] you restarted and now run version {version}. Continue the active turn." - ), - }); - current_start - }; - let cfg = match load_config() { - Ok(cfg) => cfg, - Err(error) => { - self.not_configured(error); - return; - } - }; - if let Some(message) = message { - let text = self.turn_text(&message); - self.notes.clear(); - self.add_history(Entry::User { text }); - } - self.steps(&cfg, current_start).await; - if !self.closing() { - self.finish_turn(); - } - } - - async fn steps(&mut self, cfg: &Config, current_start: usize) { - let tools = tools::definitions(); - loop { - if self.closing() { - return; - } - let start = next_turn_start(&self.history, self.history_skip, current_start) - .unwrap_or(current_start); - if start != self.history_skip { - self.history_skip = start; - let _ = std::fs::write(HISTORY_SKIP, start.to_string()); - } - let response = llm::complete(cfg, SYSTEM, &tools, &self.history[start..]).await; - let entries = match response { - Ok(entries) => entries, - Err(e) => { - if is_too_large(&e) { - if let Some(next) = - next_turn_start(&self.history, start.saturating_add(1), current_start) - { - self.history_skip = next; - let _ = std::fs::write(HISTORY_SKIP, next.to_string()); - continue; - } + let response = match llm::complete(&cfg, SYSTEM, &definitions, &entries[start..]).await + { + Ok(response) => response, + Err(error) => { + if is_too_large(&error) + && let Some(next) = next_turn_start(&entries, start + 1, current) + { + start = next; + std::fs::write(SKIP, start.to_string())?; + continue; } - let text = format!("Inference failed: {e}"); - self.add_history(Entry::Assistant { - text: text.clone(), - phase: None, - }); - let _ = publish(json!({"type": "text", "text": text})); - return; + return Err(Error::new(-32603, error)); } }; - let calls: Vec<(String, String, String)> = entries + let has_tools = response .iter() - .filter_map(|entry| match entry { + .any(|entry| matches!(entry, Entry::ToolCall { .. })); + let mut final_text = String::new(); + let mut calls = Vec::new(); + for entry in response { + match &entry { + Entry::Assistant { text, phase } + if has_tools || phase.as_deref() == Some("commentary") => + { + ctx.notify("text", json!({"text":text}))? + } + Entry::Assistant { text, .. } => final_text.push_str(text), Entry::ToolCall { id, name, arguments, - } => Some((id.clone(), name.clone(), arguments.clone())), - _ => None, - }) - .collect(); - for entry in entries { - if let Entry::Assistant { text, .. } = &entry { - if !text.is_empty() { - let _ = publish(json!({"type": "text", "text": text})); - } + } => calls.push((id.clone(), name.clone(), arguments.clone())), + _ => {} } - self.add_history(entry); + append(&mut entries, entry)?; } - if self.closing() { - return; + if !has_tools { + return Ok(final_text); } - if calls.is_empty() { - return; - } - for (id, name, args) in calls { - if self.closing() { - return; - } - let output = self.tool(&name, &args).await; - self.add_history(Entry::ToolResult { - id, - output: output.to_string(), - }); + for (id, name, arguments) in calls { + let output = match serde_json::from_str(&arguments) { + Ok(args) => tools::run(&name, &args) + .await + .unwrap_or_else(|error| json!({"error":error.to_string()})), + Err(error) => json!({"error":format!("arguments are not valid JSON: {error}")}), + }; + append( + &mut entries, + Entry::ToolResult { + id, + output: output.to_string(), + }, + )?; } } } - - 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()); - } - } - - fn drain_inbox(&mut self, others: &mut Vec) { - while let Some(message) = self.inbox.try_recv() { - self.stash(message, others); - } - } - - async fn tool(&mut self, name: &str, args: &str) -> Value { - let args: Value = match serde_json::from_str(args) { - Ok(args) => args, - Err(error) => return json!({"error": format!("arguments are not valid JSON: {error}")}), - }; - if name == "send_message" { - return self.send_message(&args).await; - } - tools::run(name, &args).unwrap_or_else(|error| json!({"error": error.to_string()})) - } - - 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"}); - } - 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.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), - Ok(None) => break, - Err(_) => break, - } - } - } - self.drain_inbox(&mut others); - json!({"id": id, "reply": reply, "waited": wait > 0.0, "other_messages": others}) - } } -fn load_config() -> Result { - 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())) +/// Read persisted entries in their original order. A new world has no history file. +fn load_history() -> Result> { + match std::fs::read_to_string(HISTORY) { + Ok(text) => text + .lines() + .map(|line| serde_json::from_str(line).map_err(Into::into)) + .collect(), + Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(Vec::new()), + Err(error) => Err(error.into()), + } } - -fn starts_turn(message: &Message) -> bool { - message.kind() == "text" && message.reply_to.is_none() +/// Persist an entry before it enters the current inference context. +fn append(history: &mut Vec, entry: Entry) -> Result<()> { + let mut file = std::fs::OpenOptions::new() + .create(true) + .append(true) + .open(HISTORY)?; + writeln!(file, "{}", serde_json::to_string(&entry)?)?; + history.push(entry); + Ok(()) +} +/// Complete interrupted tool pairs before a new inference request. +fn repair_tools(history: &mut Vec) -> Result<()> { + let answered: std::collections::HashSet<_> = history + .iter() + .filter_map(|entry| match entry { + Entry::ToolResult { id, .. } => Some(id.clone()), + _ => None, + }) + .collect(); + let dangling: Vec<_> = history + .iter() + .filter_map(|entry| match entry { + Entry::ToolCall { id, .. } if !answered.contains(id) => Some(id.clone()), + _ => None, + }) + .collect(); + for id in dangling { + append( + history, + Entry::ToolResult { + id, + output: json!({"error":"interrupted before the tool finished"}).to_string(), + }, + )?; + } + Ok(()) +} +/// Start the plugin's JSON-RPC transport. +fn main() { + sausage_plugin::run(Agent); } diff --git a/agent/src/prompt.md b/agent/src/prompt.md index dafa128..5d65e2b 100644 --- a/agent/src/prompt.md +++ b/agent/src/prompt.md @@ -1,7 +1,7 @@ -You are the `agent` actor in a Sausage world of actors. -People talk to you through other actors. Your text goes to your subscribers. -Before writing an actor, read `/actor/broker/public/PROTOCOL.md` for messages, the broker, `cargo`, and tests. -Read the SDK at `/actor/cargo/public/sdk/` before you write an actor. -You can change your own source at `/actor/agent/source`. +You are the Sausage agent. Answer the person. Use tools to inspect or change the world. -The global filesystem is readable by every actor, including every actor's `/data` at `/actor//data/`. An actor can write only its own `/data` and `/public`. `/data` is not private: any unprotected secret file there is readable by every actor. +Read the contract at `/proc/core/PROTOCOL.md`. The Rust plugin SDK and its documentation are at `/plugins/cargo/sdk/`. +Build a plugin with `cargo.build`. Activate its WASM file with `core.activate`. +Your own source is at `/plugins/agent/source/`. You can change and rebuild it like any other plugin. +Write logs to stderr. Use call results and notifications for communication. +The interface check does not cover persistent state. Preserve state compatibility when you change a plugin. diff --git a/agent/src/tools.rs b/agent/src/tools.rs index 65e2671..cd45a46 100644 --- a/agent/src/tools.rs +++ b/agent/src/tools.rs @@ -1,26 +1,27 @@ //! Tool descriptions and file operations for the agent. use crate::llm::Tool; -use sausage_actor::{Value, json}; +use serde_json::{Value, json}; use std::io; +/// Describe the five tools sent to either inference API. pub fn definitions() -> Vec { vec![ Tool { - name: "send_message", - description: "Send an actor a JSON body with a `type`. `broker` and `cargo` are built-in actors; see `/actor/broker/public/PROTOCOL.md`. With `wait_seconds`, the reply's `reply_to` matches the returned ID. Returns the ID, reply, and other messages received.", + name: "call", + description: "Call a plugin method through the core. Await its result. Set detach to let its process survive this caller.", parameters: schema( json!({ - "to": {"type": "string", "description": "Actor ID"}, - "body": {"type": "object", "description": "JSON object with a type field"}, - "wait_seconds": {"type": "number", "description": "Seconds to wait for a reply. Default 0."} + "method": {"type": "string", "description": "Qualified plugin method"}, + "params": {"type": "object", "description": "Named method parameters"}, + "detach": {"type": "boolean", "description": "Detach the new process. Default false."} }), - &["to", "body"], + &["method", "params"], ), }, Tool { name: "read_file", - description: "Read a text file under `/data`, `/public`, or `/actor/`. Reads all bytes by default; `offset` and `max_bytes` are optional.", + description: "Read a text file in the world, including `/proc`. Reads all bytes by default; `offset` and `max_bytes` are optional.", parameters: schema( json!({ "path": {"type": "string"}, @@ -32,7 +33,7 @@ pub fn definitions() -> Vec { }, Tool { name: "write_file", - description: "Write text under `/data` or `/public`. Parent folders are created.", + description: "Write a text file in the world. Parent folders are created.", parameters: schema( json!({"path": {"type": "string"}, "content": {"type": "string"}}), &["path", "content"], @@ -40,12 +41,12 @@ pub fn definitions() -> Vec { }, Tool { name: "list_dir", - description: "List entries in a directory under `/data`, `/public`, or `/actor/`. Returns each name, type, and size.", + description: "List entries in a world directory. Returns each name, type, and size.", parameters: schema(json!({"path": {"type": "string"}}), &["path"]), }, Tool { name: "copy_dir", - description: "Copy a directory tree, for example `/actor//source` to `/public/drafts/`. Skips `target/` folders.", + description: "Copy a directory tree, for example a plugin source to a draft folder. Skips `target/` folders.", parameters: schema( json!({"from": {"type": "string"}, "to": {"type": "string"}}), &["from", "to"], @@ -54,20 +55,34 @@ pub fn definitions() -> Vec { ] } +/// Create the named-parameter object required by a tool. fn schema(properties: Value, required: &[&str]) -> Value { json!({"type": "object", "properties": properties, "required": required}) } -pub fn run(name: &str, args: &Value) -> io::Result { - match name { +/// Execute a tool. The chat loop returns errors to the model as tool results. +pub async fn run(name: &str, args: &Value) -> sausage_plugin::Result { + if name == "call" { + let method = args["method"] + .as_str() + .ok_or_else(|| sausage_plugin::Error::new(-32602, "missing method"))?; + let call = sausage_plugin::call(method, args.get("params").cloned().unwrap_or(json!({}))); + return if args["detach"] == true { + call.detach().await + } else { + call.await + }; + } + Ok(match name { "read_file" => read_file(args), "write_file" => write_file(args), "list_dir" => list_dir(args), "copy_dir" => copy_dir(args), other => Ok(json!({"error": format!("unknown tool {other}")})), - } + }?) } +/// Read an optional byte range and report the complete file size. fn read_file(args: &Value) -> io::Result { let path = args["path"].as_str().unwrap_or_default(); let bytes = std::fs::read(path)?; @@ -84,6 +99,7 @@ fn read_file(args: &Value) -> io::Result { })) } +/// Create parent directories and replace a text file. fn write_file(args: &Value) -> io::Result { let path = args["path"].as_str().unwrap_or_default(); if let Some(parent) = std::path::Path::new(path).parent() { @@ -93,6 +109,7 @@ fn write_file(args: &Value) -> io::Result { Ok(json!({"written": path})) } +/// Return directory metadata in name order. fn list_dir(args: &Value) -> io::Result { let mut entries: Vec = std::fs::read_dir(args["path"].as_str().unwrap_or_default())? .filter_map(|entry| entry.ok()) @@ -109,6 +126,7 @@ fn list_dir(args: &Value) -> io::Result { Ok(json!({"entries": entries})) } +/// Copy a tree and report its file count. fn copy_dir(args: &Value) -> io::Result { let count = copy_tree( args["from"].as_str().unwrap_or_default(), @@ -117,6 +135,7 @@ fn copy_dir(args: &Value) -> io::Result { Ok(json!({"copied_files": count})) } +/// Copy files recursively without build artifacts in target directories. fn copy_tree(from: &str, to: &str) -> io::Result { std::fs::create_dir_all(to)?; let mut count = 0; diff --git a/agent/tests/answers.json b/agent/tests/answers.json deleted file mode 100644 index 9c2e85f..0000000 --- a/agent/tests/answers.json +++ /dev/null @@ -1,6 +0,0 @@ -{ - "description": "the agent answers text with text, even without config or network", - "steps": [ - {"send": {"type": "text", "text": "Hello"}, "expect": {"type": "text"}} - ] -} diff --git a/cli/tests/agent.rs b/cli/tests/agent.rs new file mode 100644 index 0000000..62baa16 --- /dev/null +++ b/cli/tests/agent.rs @@ -0,0 +1,409 @@ +//! Run the agent in the real core against scripted OpenAI and Anthropic HTTP servers. +use bytes::Bytes; +use http_body_util::{BodyExt, Full}; +use hyper::{Response, server::conn::http1, service::service_fn}; +use hyper_util::rt::TokioIo; +use sausage::{Host, world::World}; +use serde_json::{Value, json}; +use std::{convert::Infallible, path::PathBuf, sync::OnceLock, time::Duration}; +use tokio::{ + net::TcpListener, + sync::{mpsc, oneshot}, +}; + +/// The two HTTP contracts supported by the agent. +const APIS: [&str; 2] = ["openai-responses", "anthropic-messages"]; +/// A captured request and the script's delayed HTTP response. +type Exchange = (String, Value, oneshot::Sender<(u16, Value)>); +/// The script controls each response, including concurrent requests. +struct Llm { + url: String, + api: &'static str, + requests: mpsc::UnboundedReceiver, + task: tokio::task::JoinHandle<()>, +} +impl Llm { + async fn new(api: &'static str) -> Self { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let url = format!("http://{}", listener.local_addr().unwrap()); + let (tx, requests) = mpsc::unbounded_channel::(); + let task = tokio::spawn(async move { + loop { + let (stream, _) = listener.accept().await.unwrap(); + let tx = tx.clone(); + tokio::spawn(async move { + let service = + service_fn(move |request: hyper::Request| { + let tx = tx.clone(); + async move { + let path = request.uri().path().to_string(); + let bytes = request.into_body().collect().await.unwrap().to_bytes(); + let body = serde_json::from_slice(&bytes).unwrap(); + let (reply, answer) = oneshot::channel(); + tx.send((path, body, reply)).unwrap(); + let (status, body) = answer.await.unwrap(); + Ok::<_, Infallible>( + Response::builder() + .status(status) + .header("content-type", "application/json") + .body(Full::new(Bytes::from(body.to_string()))) + .unwrap(), + ) + } + }); + http1::Builder::new() + .serve_connection(TokioIo::new(stream), service) + .await + .unwrap(); + }); + } + }); + Self { + url, + api, + requests, + task, + } + } + async fn next(&mut self) -> (Value, oneshot::Sender<(u16, Value)>) { + let (path, body, reply) = + tokio::time::timeout(Duration::from_secs(60), self.requests.recv()) + .await + .unwrap() + .unwrap(); + assert_eq!( + path, + if self.api == APIS[0] { + "/v1/responses" + } else { + "/v1/messages" + } + ); + assert_eq!(body["model"], "scripted-test"); + (body, reply) + } + fn answer(&self, text: &str, calls: &[(&str, Value)]) -> Value { + if self.api == APIS[0] { + let mut output = + vec![json!({"type":"message","content":[{"type":"output_text","text":text}]})]; + for (n, (name, args)) in calls.iter().enumerate() { + output.push(json!({"type":"function_call","call_id":format!("tool-{n}"),"name":name,"arguments":args.to_string()})); + } + json!({"status":"completed","output":output}) + } else { + let mut content = vec![json!({"type":"text","text":text})]; + for (n, (name, args)) in calls.iter().enumerate() { + content.push( + json!({"type":"tool_use","id":format!("tool-{n}"),"name":name,"input":args}), + ); + } + json!({"content":content,"stop_reason":if calls.is_empty(){"end_turn"}else{"tool_use"}}) + } + } +} +impl Drop for Llm { + fn drop(&mut self) { + self.task.abort(); + } +} + +fn fixtures() -> &'static PathBuf { + static TARGET: OnceLock = OnceLock::new(); + TARGET.get_or_init(|| { + let root = PathBuf::from(env!("CARGO_MANIFEST_DIR")) + .parent() + .unwrap() + .to_owned(); + let target = root.join("target/sdk/agent-guests"); + for args in [ + vec![ + "build", + "--offline", + "-p", + "agent", + "--target", + "wasm32-wasip2", + ], + vec![ + "build", + "--offline", + "--manifest-path", + "sdk/tests/plugins/probe/Cargo.toml", + "--target", + "wasm32-wasip2", + ], + ] { + assert!( + std::process::Command::new("cargo") + .args(args) + .current_dir(&root) + .env("CARGO_TARGET_DIR", &target) + .status() + .unwrap() + .success() + ); + } + target.join("wasm32-wasip2/debug") + }) +} +struct TestWorld(PathBuf); +impl TestWorld { + fn new(server: &Llm) -> Self { + let path = std::env::temp_dir().join(format!("sausage-agent-{}", uuid::Uuid::now_v7())); + let world = World::create(&path).unwrap(); + for (plugin, binary) in [("agent", "agent"), ("probe", "sdk-probe")] { + std::fs::create_dir_all(world.plugin_dir(plugin)).unwrap(); + std::fs::copy( + fixtures().join(format!("{binary}.wasm")), + world.plugin_dir(plugin).join("plugin.wasm"), + ) + .unwrap(); + } + std::fs::write(path.join("config/agent.json"),json!({"api":server.api,"base_url":server.url,"model":"scripted-test","token":"fake-token"}).to_string()).unwrap(); + Self(path) + } + async fn start(&self) -> Host { + Host::start_with_cache( + &self.0, + &PathBuf::from(env!("CARGO_MANIFEST_DIR")).join("../target/sdk/agent-cache"), + ) + .await + .unwrap() + } + fn history(&self) -> String { + std::fs::read_to_string(self.0.join("state/agent/history.jsonl")).unwrap() + } +} +impl Drop for TestWorld { + fn drop(&mut self) { + let _ = std::fs::remove_dir_all(&self.0); + } +} +fn chat(host: &Host, text: &str) -> tokio::task::JoinHandle { + let host = host.clone(); + let text = text.to_owned(); + tokio::spawn(async move { + tokio::time::timeout( + Duration::from_secs(90), + host.call("agent.chat", json!({"text":text})), + ) + .await + .unwrap() + .unwrap() + }) +} +fn input(body: &Value, api: &str) -> Value { + body[if api == APIS[0] { "input" } else { "messages" }].clone() +} +fn outputs(body: &Value, api: &str) -> Vec { + if api == APIS[0] { + body["input"] + .as_array() + .unwrap() + .iter() + .filter(|e| e["type"] == "function_call_output") + .map(|e| serde_json::from_str(e["output"].as_str().unwrap()).unwrap()) + .collect() + } else { + body["messages"] + .as_array() + .unwrap() + .iter() + .flat_map(|e| e["content"].as_array().unwrap()) + .filter(|e| e["type"] == "tool_result") + .map(|e| serde_json::from_str(e["content"].as_str().unwrap()).unwrap()) + .collect() + } +} + +/// Both API shapes return final text and retain the complete history after a core restart. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn final_answers_and_history_survive_restart() { + for api in APIS { + let mut server = Llm::new(api).await; + let world = TestWorld::new(&server); + let host = world.start().await; + let turn = chat(&host, "first request alpha"); + let (body, reply) = server.next().await; + assert!( + input(&body, api) + .to_string() + .contains("first request alpha") + ); + assert_eq!(body["tools"].as_array().unwrap().len(), 5); + reply + .send((200, server.answer("first answer alpha", &[]))) + .unwrap(); + assert_eq!(turn.await.unwrap(), "first answer alpha"); + host.stop().await; + let prefix = world.history(); + drop(host); + let host = world.start().await; + let turn = chat(&host, "second request beta"); + let (body, reply) = server.next().await; + let text = input(&body, api).to_string(); + for expected in [ + "first request alpha", + "first answer alpha", + "second request beta", + ] { + assert!(text.contains(expected)); + } + reply + .send((200, server.answer("second answer beta", &[]))) + .unwrap(); + assert_eq!(turn.await.unwrap(), "second answer beta"); + assert!(world.history().starts_with(&prefix)); + let entries: Vec = world + .history() + .lines() + .map(|line| serde_json::from_str(line).unwrap()) + .collect(); + assert_eq!( + entries + .iter() + .map(|e| e["kind"].as_str().unwrap()) + .collect::>(), + ["user", "assistant", "user", "assistant"] + ); + let doc = host.call("agent.rpc.discover", json!({})).await.unwrap(); + assert_eq!(doc["x-sausage-grace"], 600); + host.stop().await; + } +} + +/// Both API shapes execute every tool, return tool errors, and deliver interim text. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn tools_and_interim_notifications() { + for api in APIS { + let mut server = Llm::new(api).await; + let world = TestWorld::new(&server); + let host = world.start().await; + let mut events = host.subscribe(); + let turn = chat(&host, "exercise the tools"); + let (_, reply) = server.next().await; + let calls = [ + ( + "call", + json!({"method":"probe.add","params":{"a":4,"b":5},"detach":true}), + ), + ("read_file", json!({"path":"/proc/core/PROTOCOL.md"})), + ( + "write_file", + json!({"path":"/state/work/source/note.txt","content":"saved"}), + ), + ("list_dir", json!({"path":"/state/work/source"})), + ( + "copy_dir", + json!({"from":"/state/work/source","to":"/state/work/copied"}), + ), + ("read_file", json!({"path":"/state/work/missing"})), + ("call", json!({"method":"absent.run","params":{}})), + ]; + reply + .send((200, server.answer("I will inspect the world.", &calls))) + .unwrap(); + let (body, reply) = server.next().await; + let results = outputs(&body, api); + assert_eq!(results.len(), 7); + assert_eq!(results[0], 9); + assert!(results[1]["content"].as_str().unwrap().contains("JSON-RPC")); + assert_eq!(results[2]["written"], "/state/work/source/note.txt"); + assert_eq!(results[3]["entries"][0]["name"], "note.txt"); + assert_eq!(results[4]["copied_files"], 1); + assert!(results[5]["error"].is_string()); + assert!(results[6]["error"].is_string()); + assert_eq!( + std::fs::read_to_string(world.0.join("state/work/copied/note.txt")).unwrap(), + "saved" + ); + let event = tokio::time::timeout(Duration::from_secs(5), events.recv()) + .await + .unwrap() + .unwrap(); + assert_eq!(event["params"]["name"], "text"); + assert_eq!( + event["params"]["params"]["text"], + "I will inspect the world." + ); + reply + .send((200, server.answer("The tools finished.", &[]))) + .unwrap(); + assert_eq!(turn.await.unwrap(), "The tools finished."); + host.stop().await; + } +} + +/// The history lock holds the second chat until the complete first turn ends. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn concurrent_chats_serialize_complete_turns() { + for api in APIS { + let mut server = Llm::new(api).await; + let world = TestWorld::new(&server); + let host = world.start().await; + let first = chat(&host, "first concurrent request"); + let (_, reply) = server.next().await; + let second = chat(&host, "second concurrent request"); + tokio::time::timeout(Duration::from_secs(10), async { + while host + .processes() + .values() + .filter(|p| p.plugin == "agent" && p.method == "chat") + .count() + != 2 + { + tokio::task::yield_now().await; + } + }) + .await + .unwrap(); + assert!( + tokio::time::timeout(Duration::from_millis(200), server.requests.recv()) + .await + .is_err() + ); + reply + .send((200, server.answer("first concurrent answer", &[]))) + .unwrap(); + assert_eq!(first.await.unwrap(), "first concurrent answer"); + let (body, reply) = server.next().await; + let text = input(&body, api).to_string(); + assert!(text.contains("first concurrent answer")); + assert!(text.contains("second concurrent request")); + reply + .send((200, server.answer("second concurrent answer", &[]))) + .unwrap(); + assert_eq!(second.await.unwrap(), "second concurrent answer"); + host.stop().await; + } +} + +/// Context-size failures discard old complete turns without changing the append-only history. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn context_window_advances_by_complete_turns() { + for api in APIS { + let mut server = Llm::new(api).await; + let world = TestWorld::new(&server); + let host = world.start().await; + let first = chat(&host, "old turn"); + let (_, reply) = server.next().await; + reply.send((200, server.answer("old answer", &[]))).unwrap(); + first.await.unwrap(); + let second = chat(&host, "current turn"); + let (_, reply) = server.next().await; + reply + .send((400, json!({"error":"context_length_exceeded"}))) + .unwrap(); + let (body, reply) = server.next().await; + let text = input(&body, api).to_string(); + assert!(!text.contains("old turn")); + assert!(!text.contains("old answer")); + assert!(text.contains("current turn")); + reply + .send((200, server.answer("current answer", &[]))) + .unwrap(); + assert_eq!(second.await.unwrap(), "current answer"); + assert_eq!(world.history().lines().count(), 4); + host.stop().await; + } +} diff --git a/cli/tests/sdk.rs b/cli/tests/sdk.rs index 5d42baa..40bc77e 100644 --- a/cli/tests/sdk.rs +++ b/cli/tests/sdk.rs @@ -157,7 +157,6 @@ async fn discovery_is_compatible() { } /// Detached SDK calls receive progress and forward it through core.notify. #[tokio::test(flavor = "multi_thread", worker_threads = 2)] -#[ignore = "core: stdout response can overtake a preceding core.notify"] async fn detached_call_notifications() { let world = TestWorld::new(); let host = world.start().await; @@ -229,7 +228,6 @@ async fn world_endpoint() { } /// The HTTP example exposes its self endpoint through the process namespace. #[tokio::test(flavor = "multi_thread", worker_threads = 2)] -#[ignore = "needs /proc, /self, and /run (T6)"] async fn www_control_through_proc() { let world = TestWorld::new(); let host = world.start().await; @@ -241,7 +239,6 @@ async fn www_control_through_proc() { } /// The vault example publishes a fixed endpoint in the shared run namespace. #[tokio::test(flavor = "multi_thread", worker_threads = 2)] -#[ignore = "needs /proc, /self, and /run (T6)"] async fn secrets_fixed_endpoint() { let world = TestWorld::new(); let host = world.start().await; diff --git a/examples/consumer/src/main.rs b/examples/consumer/src/main.rs index 695589f..d88a827 100644 --- a/examples/consumer/src/main.rs +++ b/examples/consumer/src/main.rs @@ -59,7 +59,7 @@ impl Consumer { fn main() { sausage_plugin::run(Consumer) } -/// Tests demonstrate generated params and calls through the test core. +/// Tests check generated wire types without other plugins in the test world. #[cfg(test)] mod tests { use super::*; @@ -72,17 +72,10 @@ mod tests { .unwrap(); assert_eq!(value, sausage_plugin::serde_json::json!({"text":"milk"})); } - /// A typed import reaches the notes plugin through the test core. - #[sausage_plugin::test] - async fn imports_notes() { - assert_eq!( - Consumer - .note(NoteParams { - text: "milk".into() - }) - .await - .unwrap(), - 1 - ); + /// A generated result decodes without another plugin in the test world. + #[test] + fn generated_result_decodes_count() { + let added: notes::Added = sausage_plugin::serde_json::from_str(r#"{"count":2}"#).unwrap(); + assert_eq!(added.count, 2); } }