diff --git a/docs/LOOKOUT.md b/docs/LOOKOUT.md index ba7dcb429..f99f3f38f 100644 --- a/docs/LOOKOUT.md +++ b/docs/LOOKOUT.md @@ -173,6 +173,9 @@ those with subscriptions of their own: - a running turn that has printed nothing for 30 minutes, once per silence; a turn waiting on the person, on other turns or for a slot is not silent; +- a task's turns its goal (SONNY.md) held, once the goal ends: only the + last, with how it ended (``turn:calc.fix/4 completed, asked by goal, + and met its goal: its check exits 0``); - turns the person asked for in a task themselves, listed apart as theirs; - what was saved to or removed from the person's and the project's memory, by path, less what the subscriber saved itself: the folders its own diff --git a/docs/SONNY.md b/docs/SONNY.md index 6ee26fcd4..5a17b2fe5 100644 --- a/docs/SONNY.md +++ b/docs/SONNY.md @@ -286,6 +286,43 @@ its folder rather than trusting a compacted conversation for ids and times, report a source it could not read by name, re-check items right before posting, and record a post only once the destination confirms it. +## Goals + +A goal keeps an agent working until a command says its work is done, with +no judge and nothing in the daemon ([goals.rs](../sonny/src/lookout/goals.rs)). +An agent sets one with the `goal` tool, on itself or on a task it made: +`set {done_when, check, bot, budget_tokens, timeout_s}`, `blocked` WHY to +give its own up, `clear` NAME, or nothing to list them. The check is a +shell command (`cargo test`, `make lint`, `test -z "$(git status --short)"`) +that exits 0 once done; the lookout runs it in the agent's folder, as the +person and outside any sandbox like a trigger's `if`, though without the +provider keys an agent's shell goes without, so `set` waits for a verdict wherever the +agent's other calls do (`/approvals`). Each goal is +`STORE.lookout/goals/BOT.json`; setting the same goal again changes +nothing. + +Each time a turn of that agent completes, the lookout runs the check, for +up to `timeout_s` (600 when left out, at most 3,600), killing its process +group past that. While it fails, it queues the agent a turn (origin +`goal`) with its exit code and the last 60 lines (8 KiB) it printed: + +``` +Your goal is not met yet: `cargo test` exits 101. +... +Keep working until the tests pass. When no way remains, give the goal up with the goal tool's blocked, saying why. +``` + +The goal ends, and its file goes, at the first of: the check exits 0; the +agent's tokens since `set` reach `budget_tokens`; two turns in a row that +the goal started call no tool, since the agent only answers and another +turn would too; the agent calls `blocked`; or someone clears it or the +agent is deleted. A turn that fails or is stopped ends nothing and starts +nothing: the lead hears it as any turn. Its lead hears none of the turns +the goal held until the last, told with how the goal ended (below). That +line is the lookout's own words only: the agent's reason for giving up, +and a check it set itself, are its text, which the lead reads in the turn +or with `goal`. + ## Memory Memory is what earlier agents learned that a later one needs: decisions, diff --git a/sonny/src/approvals.rs b/sonny/src/approvals.rs index b8b765b3e..46083ef97 100644 --- a/sonny/src/approvals.rs +++ b/sonny/src/approvals.rs @@ -391,7 +391,8 @@ mod tests { "new_project", "subscriptions", "run_in_pane", - "triggers" + "triggers", + "goal" ]) ); } diff --git a/sonny/src/lookout/goals.rs b/sonny/src/lookout/goals.rs new file mode 100644 index 000000000..64f3797c0 --- /dev/null +++ b/sonny/src/lookout/goals.rs @@ -0,0 +1,809 @@ +//! Goals: an agent keeps working until a command says its work is done. +//! A goal is what done looks like, in words, and a check: a command that +//! exits 0 once it is. Each time a turn of the agent that has one +//! completes, the lookout runs the check in the agent's folder. While it +//! fails, the agent gets its output and works again, and its lead hears +//! nothing of those turns; once it passes, or the agent gives up, or the +//! goal's budget is spent, or two turns in a row the goal started call no +//! tool, the goal ends and the lead hears that turn with how it ended. +//! No model judges: a command says when the work is done, as it does in +//! CI. +//! +//! Each agent's goal is one file, `STORE.lookout/goals/BOT.json`, which +//! its `goal` tool writes (a lead's for its tasks too) and the lookout +//! reads again when a call changed it, and removes when the goal ends. A +//! check runs as the person, outside every sandbox, as a trigger's `if` +//! does, without the provider keys an agent's shell goes without: +//! setting one waits for a verdict where the agent's other calls do. +use agent_client::Client; +use serde_json::{Value, json}; +use std::collections::{HashMap, HashSet}; +use std::path::{Path, PathBuf}; +use std::time::Duration; + +/// How long a check may run unless the goal says, and at most. +pub const TIMEOUT_S: u64 = 10 * 60; +pub const TIMEOUT_MAX_S: u64 = 60 * 60; +/// How much of a failing check's output the agent is shown: its end. +const OUTPUT_LINES: usize = 60; +const OUTPUT_BYTES: usize = 8 * 1024; +/// Turns in a row the goal started that called no tool, after which it +/// stops: the agent only answers, and another turn would too. +const IDLE_TURNS: u32 = 2; +/// What a turn the goal started carries as its request ID, before the +/// agent and turn it follows: not the lookout's own note, since the +/// turn is the agent's work and its end is news. +pub(crate) const REQUEST: &str = "goal-"; + +#[derive(Clone, Debug, PartialEq)] +pub struct Goal { + pub bot: String, + pub bot_id: i64, + pub done_when: String, + pub check: String, + /// The agent's folder, where the check runs. + pub workspace: PathBuf, + pub set_by: String, + /// When it was set: a goal set again since is another goal. + pub set_ms: u64, + /// Tokens the agent may spend on it, counted from `tokens_at_set`. + pub budget_tokens: Option, + pub tokens_at_set: u64, + pub timeout_s: u64, + /// Why the agent gave up, told once its turn ends. + pub blocked: Option, +} + +impl Goal { + fn to_json(&self) -> Value { + json!({"bot": self.bot, "bot_id": self.bot_id, "done_when": self.done_when, "check": self.check, + "workspace": self.workspace, "set_by": self.set_by, "set_ms": self.set_ms, + "budget_tokens": self.budget_tokens, "tokens_at_set": self.tokens_at_set, + "timeout_s": self.timeout_s, "blocked": self.blocked}) + } + + fn from_json(value: &Value) -> Option { + let text = |key: &str| value[key].as_str().map(str::to_owned); + Some(Goal { + bot: text("bot")?, + bot_id: value["bot_id"].as_i64()?, + done_when: text("done_when")?, + check: text("check")?, + workspace: PathBuf::from(text("workspace")?), + set_by: text("set_by")?, + set_ms: value["set_ms"].as_u64()?, + budget_tokens: value["budget_tokens"].as_u64(), + tokens_at_set: value["tokens_at_set"].as_u64().unwrap_or(0), + timeout_s: value["timeout_s"].as_u64().unwrap_or(TIMEOUT_S), + blocked: text("blocked"), + }) + } + + /// What the agent and its lead are shown of it. + fn shown(&self) -> Value { + json!({"bot": self.bot, "done_when": self.done_when, "check": self.check, + "budget_tokens": self.budget_tokens, "timeout_s": self.timeout_s, "set_by": self.set_by, + "blocked": self.blocked}) + } +} + +pub fn dir(store: &Path) -> PathBuf { + super::subs::dir(store).join("goals") +} + +fn file(store: &Path, bot: &str) -> PathBuf { + dir(store).join(format!("{bot}.json")) +} + +pub fn read(store: &Path, bot: &str) -> Option { + let text = std::fs::read_to_string(file(store, bot)).ok()?; + Goal::from_json(&serde_json::from_str(&text).ok()?) +} + +pub fn all(store: &Path) -> HashMap { + let Ok(entries) = std::fs::read_dir(dir(store)) else { + return HashMap::new(); + }; + entries + .filter_map(|entry| { + let name = entry.ok()?.file_name().to_string_lossy().into_owned(); + let bot = name.strip_suffix(".json")?; + Some((bot.to_owned(), read(store, bot)?)) + }) + .collect() +} + +/// Written whole and moved in, so the lookout never reads half of one. +fn write(store: &Path, goal: &Goal) -> Result<(), String> { + let dir = dir(store); + std::fs::create_dir_all(&dir).map_err(|e| format!("{}: {e}", dir.display()))?; + let path = file(store, &goal.bot); + let partial = dir.join(format!(".{}.{}.partial", goal.bot, std::process::id())); + std::fs::write(&partial, goal.to_json().to_string()) + .and_then(|()| std::fs::rename(&partial, &path)) + .map_err(|e| { + let _ = std::fs::remove_file(&partial); + format!("{}: {e}", path.display()) + }) +} + +/// The goal ends: its file goes, unless it was set again since. +fn end(store: &Path, goal: &Goal) { + if read(store, &goal.bot).is_some_and(|now| now.set_ms == goal.set_ms) { + let _ = std::fs::remove_file(file(store, &goal.bot)); + } +} + +fn now_ms() -> u64 { + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map_or(0, |d| d.as_millis() as u64) +} + +/// The `goal` tool, as `sonny tool goal` runs it for `caller`: set one on +/// itself or on a task it made, give its own up, clear one, or list the +/// goals it and its tasks have. +pub async fn call( + store: &Path, + client: &Client, + caller: &str, + arguments: &Value, +) -> Result { + let target = |bot: &Value| bot.as_str().unwrap_or(caller).to_owned(); + // A goal goes on the caller or on an agent it made, whose record says + // so by id, since a name can be another bot's since: the record also + // gives its folder, id and spend. + let inspect = async |bot: &str| -> Result { + (client.request("inspect", json!({"bot": bot})).await) + .map_err(|e| format!("bot_not_found: {bot}: {e}")) + }; + let record = async |bot: &str| -> Result { + let record = inspect(bot).await?; + if bot != caller && record["created_by_id"] != inspect(caller).await?["bot_id"] { + return Err(format!( + "not_your_task: {bot} was not made by {caller}; a goal goes on you or on a task you made" + )); + } + Ok(record) + }; + if let Some(set) = arguments.get("set") { + let bot = target(&set["bot"]); + let record = record(&bot).await?; + let text = |key: &str| set[key].as_str().map(str::trim).unwrap_or_default(); + let (done_when, check) = (text("done_when"), text("check")); + if done_when.is_empty() || check.is_empty() { + return Err("invalid_goal: done_when and check are both needed".into()); + } + let timeout_s = set["timeout_s"].as_u64().unwrap_or(TIMEOUT_S); + if !(1..=TIMEOUT_MAX_S).contains(&timeout_s) { + return Err(format!("invalid_goal: timeout_s is 1 to {TIMEOUT_MAX_S}")); + } + let budget_tokens = set["budget_tokens"].as_u64().filter(|b| *b > 0); + let workspace = record["workspace"].as_str().unwrap_or("/"); + let goal = Goal { + bot: bot.clone(), + bot_id: record["bot_id"].as_i64().unwrap_or_default(), + done_when: done_when.to_owned(), + check: check.to_owned(), + workspace: PathBuf::from(workspace), + set_by: caller.to_owned(), + set_ms: now_ms(), + budget_tokens, + tokens_at_set: record["tokens_used"].as_u64().unwrap_or(0), + timeout_s, + blocked: None, + }; + // The same goal again changes nothing, its budget still counted + // from when it was first set. + if let Some(had) = read(store, &bot).filter(|had| { + ( + had.bot_id, + &had.done_when, + &had.check, + had.budget_tokens, + had.timeout_s, + ) == ( + goal.bot_id, + &goal.done_when, + &goal.check, + goal.budget_tokens, + goal.timeout_s, + ) && had.blocked.is_none() + }) { + return Ok(json!({"goal": had.shown(), "duplicate": true})); + } + write(store, &goal)?; + return Ok(json!({"goal": goal.shown(), "duplicate": false})); + } + if let Some(why) = arguments.get("blocked") { + let why = why.as_str().map(str::trim).unwrap_or_default(); + if why.is_empty() { + return Err("invalid_goal: blocked says why".into()); + } + let Some(mut goal) = read(store, caller) else { + return Err("no_goal: you have no goal to give up".into()); + }; + goal.blocked = Some(why.to_owned()); + write(store, &goal)?; + return Ok(json!({"goal": goal.shown(), "ends": "when this turn ends"})); + } + if let Some(bot) = arguments.get("clear") { + let bot = target(bot); + record(&bot).await?; + let had = read(store, &bot); + if let Some(goal) = &had { + end(store, goal); + } + return Ok(json!({"bot": bot, "cleared": had.is_some()})); + } + // Its own, and its tasks', whoever set them. + let me = inspect(caller).await?["bot_id"].clone(); + let mut mine = Vec::new(); + for goal in all(store).into_values() { + if goal.bot == caller + || inspect(&goal.bot) + .await + .is_ok_and(|task| task["created_by_id"] == me) + { + mine.push(goal.shown()); + } + } + Ok(json!({"goals": mine})) +} + +/// The `goal` tool's declaration. +pub fn tool(name: &str, tag: &str) -> Value { + json!({ + "name": name, + "description": "A goal keeps an agent working until a command says its work is done. Set one on \ + yourself, or on a task you made, with what done looks like and a check: a shell command, run in \ + that agent's folder, that exits 0 once it is (tests passing, a build, a file count). Each time \ + its turn completes, the check runs; while it fails, the agent is shown its output and works \ + again, and its lead hears nothing of those turns. The goal ends once the check passes, the \ + budget is spent, two turns it started call no tool, or the agent gives up with blocked and why, \ + and its lead hears that turn with how it ended. Setting the same goal again changes nothing; \ + with nothing given, the result lists your goals and your tasks'.", + "parameters": { + "type": "object", + "properties": { + "set": { + "type": "object", + "properties": { + "done_when": {"type": "string", "minLength": 1, + "description": "What done looks like, in words."}, + "check": {"type": "string", "minLength": 1, + "description": "The command, as sh -c runs it, that exits 0 once it is."}, + "bot": {"type": "string", "minLength": 1, + "description": "A task you made; you when left out."}, + "budget_tokens": {"type": "integer", "minimum": 1, + "description": "Tokens the agent may spend on it before it stops."}, + "timeout_s": {"type": "integer", "minimum": 1, "maximum": TIMEOUT_MAX_S, + "description": "How long the check may run; 600 when left out."} + }, + "required": ["done_when", "check"], + "additionalProperties": false + }, + "blocked": {"type": "string", "minLength": 1, + "description": "Give your own goal up, saying why no way remains."}, + "clear": {"type": "string", "minLength": 1, + "description": "Clear the goal of a task you made, or your own by your name."} + }, + "additionalProperties": false + }, + "answered_by": tag + }) +} + +/// How a turn the lookout held for a goal ends: `None` when the agent +/// works again and the turn is no news; else what its lead hears after +/// the turn's status, empty when the goal went meanwhile. +#[derive(Debug, PartialEq)] +pub struct Settled { + pub bot: String, + pub turn: i64, + pub outcome: Option, +} + +/// The lookout's goals, and what it follows of each agent's turns. +pub struct Goals { + store: PathBuf, + socket: PathBuf, + goals: HashMap, + /// Agents whose running turn called a tool. + called: HashSet, + /// Agents whose running turn the goal started. + continued: HashSet, + /// Agents whose goal's check runs: a turn that ends meanwhile is + /// plain news, and the check's own turn is checked in its place. + checking: HashMap, + /// Turns in a row the goal started that called no tool. + /// Keyed by the goal's `set_ms` and the last turn counted, so a turn + /// replayed after a reconnect is not counted again. + idle: HashMap, + /// The folder each turn was sent to: a later turn can move its agent, + /// and the check runs where the turn worked. + worked: HashMap<(String, i64), PathBuf>, +} + +/// A goal's check under way. +struct Check { + pursuit: tokio::task::AbortHandle, + /// The cursor of the turn's end, which a restart replays. + at: Option, + turn: i64, + /// The goal checked: one set since is checked afresh. + set_ms: u64, +} + +impl Goals { + pub fn new(store: &Path, socket: &Path) -> Goals { + Goals { + store: store.to_owned(), + socket: socket.to_owned(), + goals: all(store), + called: HashSet::new(), + continued: HashSet::new(), + checking: HashMap::new(), + idle: HashMap::new(), + worked: HashMap::new(), + } + } + + /// The check of `bot`'s `turn` said how its goal stands. + pub fn settled(&mut self, bot: &str, turn: i64) { + if self + .checking + .get(bot) + .is_some_and(|check| check.turn == turn) + { + self.checking.remove(bot); + } + } + + /// Checks under way answer a connection since dropped, never the next: + /// they stop, and the replay runs them again. + pub fn reconnected(&mut self) { + for (_, check) in self.checking.drain() { + check.pursuit.abort(); + } + } + + /// The end of the oldest turn whose check is under way: what the + /// lookout keeps for the next one is from before it, so a restart, or + /// one with no subscriber to hold the turn, runs its check again. + pub fn floor(&self) -> Option { + self.checking.values().filter_map(|check| check.at).min() + } + + pub fn reread(&mut self) { + self.goals = all(&self.store); + self.idle.retain(|bot, _| self.goals.contains_key(bot)); + } + + /// An event the lookout read: the turn it holds for a goal, whose news + /// waits for `done` to settle it, when its end is one. + /// `bot_id` is the id of the agent the event names, as the lookout + /// knows it now. + pub fn event( + &mut self, + event: &Value, + bot_id: Option, + live: bool, + client: &std::sync::Arc, + done: &tokio::sync::mpsc::Sender, + ) -> Option<(String, i64)> { + let bot = event["bot"].as_str()?; + let data = &event["data"]; + if let (Some("accepted" | "queued"), Some(turn), Some(workspace)) = ( + event["event"].as_str(), + event["turn"].as_i64(), + data["workspace"].as_str(), + ) { + self.worked + .insert((bot.to_owned(), turn), PathBuf::from(workspace)); + } + match event["event"].as_str()? { + "accepted" => { + self.called.remove(bot); + match data["request_id"] + .as_str() + .is_some_and(|id| id.starts_with(REQUEST)) + { + true => self.continued.insert(bot.to_owned()), + false => self.continued.remove(bot), + }; + None + } + "tool_started" => { + self.called.insert(bot.to_owned()); + None + } + // A new agent under the name has no goal of the last one's. + "deleted" => { + self.worked.retain(|(name, _), _| name != bot); + if let Some(check) = self.checking.remove(bot) { + check.pursuit.abort(); + } + if let Some(goal) = self.goals.remove(bot) { + end(&self.store, &goal); + } + None + } + "turn_finished" => { + let turn = event["turn"].as_i64()?; + let worked = self.worked.remove(&(bot.to_owned(), turn)); + if !live { + return None; + } + let status = data["status"].as_str().unwrap_or_default(); + // Its goal as it is now, which its lead may have set or + // cleared in the meantime. + match read(&self.store, bot) { + Some(goal) => self.goals.insert(bot.to_owned(), goal), + None => self.goals.remove(bot), + }; + let mut goal = self.goals.get(bot)?.clone(); + if let Some(worked) = worked { + goal.workspace = worked; + } + // Left by an agent deleted while the lookout was away, whose + // name another has since. + if bot_id.is_some_and(|id| id != goal.bot_id) { + end(&self.store, &goal); + self.goals.remove(bot); + return None; + } + match self.checking.get(bot) { + Some(check) if check.set_ms == goal.set_ms => return None, + // The goal it checks was set again: its turn is plain + // news, and this one is checked for the new goal. + Some(_) => { + let check = self.checking.remove(bot)?; + check.pursuit.abort(); + let _ = done.try_send(Settled { + bot: bot.to_owned(), + turn: check.turn, + outcome: Some(String::new()), + }); + } + None => {} + } + // An interrupted or failed turn pauses it: the next turn + // that completes is checked. A steer's is part of another. + if status != "completed" && goal.blocked.is_none() { + return None; + } + let idle = match self.continued.contains(bot) && !self.called.contains(bot) { + // Counted for this goal only: one set since starts at 0. + true => match self.idle.get(bot) { + Some((set_ms, at, idle)) if *set_ms == goal.set_ms && *at == turn => *idle, + Some((set_ms, _, idle)) if *set_ms == goal.set_ms => idle + 1, + _ => 1, + }, + false => 0, + }; + self.idle.insert(bot.to_owned(), (goal.set_ms, turn, idle)); + let (client, done, store, socket) = ( + client.clone(), + done.clone(), + self.store.clone(), + self.socket.clone(), + ); + let held = (bot.to_owned(), turn); + let set_ms = goal.set_ms; + let bot = bot.to_owned(); + let checking = bot.clone(); + let pursuit = tokio::spawn(async move { + let outcome = pursue(&client, &store, &socket, &goal, turn, idle).await; + let _ = done.send(Settled { bot, turn, outcome }).await; + }); + let check = Check { + pursuit: pursuit.abort_handle(), + at: event["cursor"].as_i64(), + turn, + set_ms, + }; + self.checking.insert(checking, check); + Some(held) + } + _ => None, + } + } +} + +/// After a turn of an agent with a goal: whether it works again, or how +/// the goal ended. +async fn pursue( + client: &Client, + store: &Path, + socket: &Path, + goal: &Goal, + turn: i64, + idle: u32, +) -> Option { + // What the lead hears is the lookout's own words only: the reason an + // agent gives and the check it may have set are its text, which the + // lead reads in the turn itself. + if goal.blocked.is_some() { + end(store, goal); + return Some(GAVE_UP.to_owned()); + } + let check = format!("`{}`", goal.check); + let (code, output) = run(goal, store, socket).await; + // Cleared or set again while the check ran: this turn is plain news; + // given up meanwhile, it ends so. + match read(store, &goal.bot) { + Some(now) if now.set_ms == goal.set_ms => { + if now.blocked.is_some() { + end(store, &now); + return Some(GAVE_UP.to_owned()); + } + } + _ => return Some(String::new()), + } + if code == Some(0) { + end(store, goal); + return Some(", and met its goal: its check exits 0".to_owned()); + } + let exits = match code { + Some(code) => format!("exits {code}"), + None => "did not finish".to_owned(), + }; + if let Some(budget) = goal.budget_tokens { + let record = client.request("inspect", json!({"bot": goal.bot})).await; + let used = (record.ok()) + .and_then(|r| r["tokens_used"].as_u64()) + .unwrap_or(0); + if used.saturating_sub(goal.tokens_at_set) >= budget { + end(store, goal); + return Some(format!( + ", and stopped at its goal's budget of {budget} tokens: its check still {exits}" + )); + } + } + if idle >= IDLE_TURNS { + end(store, goal); + return Some(format!( + ", and stopped pursuing its goal: {IDLE_TURNS} turns in a row it started called no tool, \ + and its check still {exits}" + )); + } + let prompt = format!( + "Your goal is not met yet: {check} {exits}.\n```\n{output}\n```\nKeep working until {}. When no way \ + remains, give the goal up with the goal tool's blocked, saying why.", + goal.done_when + ); + let asked = json!({"bot": goal.bot, "bot_id": goal.bot_id, "prompt": prompt, "delivery": "queue", + "origin": "goal", "request_id": format!("{REQUEST}{}-{turn}", goal.bot)}); + match client.request("submit", asked).await { + Ok(_) => None, + Err(error) => Some(format!( + ", and could not go on with its goal: {}", + error.code + )), + } +} + +const GAVE_UP: &str = ", and gave up its goal, saying why in that turn"; + +/// The check, in its own process group in the agent's folder, for up to +/// its timeout: its exit code, `None` when it did not finish, and the end +/// of what it printed. +async fn run(goal: &Goal, store: &Path, socket: &Path) -> (Option, String) { + let mut command = tokio::process::Command::new("sh"); + command + .arg("-c") + .arg(&goal.check) + .current_dir(&goal.workspace) + .env("AGENT_STORE", store) + .env("AGENT_SOCKET", socket) + .stdin(std::process::Stdio::null()) + .stdout(std::process::Stdio::piped()) + .stderr(std::process::Stdio::piped()) + .process_group(0) + .kill_on_drop(true); + // The provider keys are the daemon's: no agent's own command has them, + // nor does its check. + for key in crate::task::kept_keys() { + command.env_remove(key); + } + let mut child = match command.spawn() { + Ok(child) => child, + Err(error) => return (None, format!("the check did not start: {error}")), + }; + // Whatever the check started goes with it, however it ends: done, past + // its timeout, or its pursuit stopped. + let _group = child.id().map(|id| Group::new(id as i32)); + // Read as it prints, keeping only the end, so a check that prints + // without end holds a few KB, not all of it. + let (stdout, stderr) = (child.stdout.take(), child.stderr.take()); + let ran = async { + let (stdout, stderr) = tokio::join!(last(stdout), last(stderr)); + (child.wait().await, stdout, stderr) + }; + let (status, stdout, stderr) = + match tokio::time::timeout(Duration::from_secs(goal.timeout_s), ran).await { + Ok((Ok(status), stdout, stderr)) => (status, stdout, stderr), + Ok((Err(error), ..)) => return (None, format!("the check failed to run: {error}")), + Err(_) => { + return ( + None, + format!("the check ran past {} s and was stopped", goal.timeout_s), + ); + } + }; + let mut text = String::from_utf8_lossy(&stdout).into_owned(); + let errors = String::from_utf8_lossy(&stderr); + if !errors.trim().is_empty() { + if !text.is_empty() && !text.ends_with('\n') { + text.push('\n'); + } + text.push_str(&errors); + } + (status.code(), tail(&text)) +} + +/// The process groups of the checks running now. +static GROUPS: std::sync::Mutex> = std::sync::Mutex::new(Vec::new()); + +/// Stop every check under way: the lookout is giving way, and the one +/// that takes over runs them again. +pub fn stop() { + for group in GROUPS.lock().map(|g| g.clone()).unwrap_or_default() { + // SAFETY: a signal to a process group a check leads. + unsafe { libc::killpg(group, libc::SIGKILL) }; + } +} + +/// A check's process group, killed when dropped. +struct Group(i32); + +impl Group { + fn new(id: i32) -> Group { + if let Ok(mut groups) = GROUPS.lock() { + groups.push(id); + } + Group(id) + } +} + +impl Drop for Group { + fn drop(&mut self) { + // SAFETY: a signal to the process group the check leads. + unsafe { libc::killpg(self.0, libc::SIGKILL) }; + if let Ok(mut groups) = GROUPS.lock() { + groups.retain(|g| *g != self.0); + } + } +} + +/// The last `OUTPUT_BYTES` a pipe gave, read until it closes. +async fn last(pipe: Option) -> Vec { + use tokio::io::AsyncReadExt; + let Some(mut pipe) = pipe else { + return Vec::new(); + }; + let (mut kept, mut chunk) = (Vec::new(), [0u8; 8192]); + while let Ok(read) = pipe.read(&mut chunk).await + && read > 0 + { + kept.extend_from_slice(&chunk[..read]); + if kept.len() > 2 * OUTPUT_BYTES { + kept.drain(..kept.len() - OUTPUT_BYTES); + } + } + let from = kept.len().saturating_sub(OUTPUT_BYTES); + kept.split_off(from) +} + +/// The end of a check's output: its last lines, within a few KB. +fn tail(text: &str) -> String { + let lines: Vec<&str> = text.trim_end().lines().collect(); + let mut kept = lines[lines.len().saturating_sub(OUTPUT_LINES)..].join("\n"); + if kept.len() > OUTPUT_BYTES { + let mut from = kept.len() - OUTPUT_BYTES; + while !kept.is_char_boundary(from) { + from += 1; + } + kept = kept[from..].to_owned(); + } + // Fences in the output would end the block the agent reads it in. + kept.replace("```", "'''") +} + +#[cfg(test)] +mod tests { + use super::*; + + fn scratch(name: &str) -> PathBuf { + let dir = std::env::temp_dir().join(format!("sonny-goals-{name}-{}", std::process::id())); + let _ = std::fs::remove_dir_all(&dir); + std::fs::create_dir_all(&dir).unwrap(); + dir + } + + fn goal(store: &Path, bot: &str) -> Goal { + Goal { + bot: bot.into(), + bot_id: 7, + done_when: "tests pass".into(), + check: "true".into(), + workspace: store.parent().unwrap().to_owned(), + set_by: "p.lead".into(), + set_ms: 1, + budget_tokens: Some(10), + tokens_at_set: 3, + timeout_s: 5, + blocked: None, + } + } + + #[test] + fn a_goal_reads_back_as_written_and_ends_only_as_set() { + let dir = scratch("read"); + let store = dir.join("state.sqlite"); + let first = goal(&store, "p.fix"); + write(&store, &first).unwrap(); + assert_eq!(read(&store, "p.fix"), Some(first.clone())); + assert_eq!(all(&store).len(), 1); + // Set again since: the old one's end leaves the new one. + let again = Goal { + set_ms: 2, + ..first.clone() + }; + write(&store, &again).unwrap(); + end(&store, &first); + assert_eq!(read(&store, "p.fix"), Some(again.clone())); + end(&store, &again); + assert_eq!(read(&store, "p.fix"), None); + } + + #[test] + fn a_check_says_its_code_and_the_end_of_what_it_printed() { + let dir = scratch("run"); + let store = dir.join("state.sqlite"); + let runtime = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .unwrap(); + let mut failing = goal(&store, "p.fix"); + failing.check = "seq 1 100; echo '```' >&2; exit 3".into(); + let (code, output) = runtime.block_on(run(&failing, &store, &store)); + assert_eq!(code, Some(3)); + assert!(output.starts_with("42\n"), "{output}"); + assert!(output.ends_with("100\n'''"), "{output}"); + // One that prints without end keeps only the end of it. + let mut loud = goal(&store, "p.fix"); + loud.check = "head -c 50000000 /dev/zero | tr '\\0' x; echo; echo last".into(); + let (code, output) = runtime.block_on(run(&loud, &store, &store)); + assert_eq!(code, Some(0)); + assert!( + output.len() <= OUTPUT_BYTES && output.ends_with("\nlast"), + "{}", + output.len() + ); + let mut slow = goal(&store, "p.fix"); + slow.check = "sleep 30".into(); + slow.timeout_s = 1; + let (code, output) = runtime.block_on(run(&slow, &store, &store)); + assert_eq!(code, None); + assert!(output.contains("ran past 1 s"), "{output}"); + let mut here = goal(&store, "p.fix"); + here.check = format!("test \"$AGENT_STORE\" = '{}' && pwd", store.display()); + let (code, output) = runtime.block_on(run(&here, &store, &store)); + assert_eq!(code, Some(0)); + assert_eq!( + std::fs::canonicalize(output.trim()).unwrap(), + std::fs::canonicalize(&dir).unwrap() + ); + } + + #[test] + fn the_tail_keeps_the_last_lines_within_its_bytes() { + let long = "x".repeat(OUTPUT_BYTES * 2); + assert_eq!(tail(&long).len(), OUTPUT_BYTES); + let wide = "é".repeat(OUTPUT_BYTES); + assert!(tail(&wide).len() <= OUTPUT_BYTES); + assert_eq!(tail("a\nb\n\n"), "a\nb"); + } +} diff --git a/sonny/src/lookout/mod.rs b/sonny/src/lookout/mod.rs index 721d37ffc..457cce671 100644 --- a/sonny/src/lookout/mod.rs +++ b/sonny/src/lookout/mod.rs @@ -12,6 +12,7 @@ //! agents' calls of the tools Sonny answers (`served`). //! //! A subscriber hears it all in one note, which starts "Task updates: ". +pub(crate) mod goals; pub(crate) mod sources; pub(crate) mod subs; mod updates; @@ -369,13 +370,14 @@ async fn serve(settings: Settings) -> Result<(), String> { store: settings.store.clone(), records: HashMap::new(), subscribers: HashMap::new(), - socket: settings.socket, + socket: settings.socket.clone(), lookouts: HashMap::new(), log: PathBuf::from(log), seated: 0, live: false, cursor: None, saved: None, + goals: goals::Goals::new(&settings.store, &settings.socket), resume: std::fs::read_to_string(cursor_path(&settings.store)) .ok() .and_then(|text| text.trim().parse().ok()), @@ -395,6 +397,7 @@ async fn serve(settings: Settings) -> Result<(), String> { term.recv().await; crate::served::stop(); sources::stop(); + goals::stop(); std::process::exit(0); }); } @@ -441,6 +444,8 @@ struct Service { /// a lookout started again tells what came after it. cursor: Option, saved: Option, + /// The agents' goals, whose turns it holds until their checks say. + goals: goals::Goals, /// Where the last lookout left off: events after it are news. resume: Option, /// Lead `react` calls naming another agent, by bot, turn and call, @@ -617,11 +622,15 @@ impl Service { // a message whose read the dropped connection cut short is read // again. self.resume = self.floor().or(self.resume); + // Goal checks a dropped connection left under way answer the old + // channel, never this one: none is under way here yet. + let (checked, mut settled) = tokio::sync::mpsc::channel(64); for lookout in self.lookouts.values_mut() { // Reads a dropped connection left under way answer the old // channel, never this one. lookout.reading = 0; } + self.goals.reconnected(); let (done, mut results) = tokio::sync::mpsc::channel(64); // Often enough for a note whose news has settled; subscription // files less often. @@ -667,6 +676,16 @@ impl Service { self.records.remove(name); self.lookouts.remove(name); } + // A turn whose goal's check is to run is news only once + // it says how the goal stands. + let bot_id = (event["bot"].as_str()) + .and_then(|bot| self.records.get(bot)) + .and_then(|record| record["bot_id"].as_i64()); + if let Some((bot, turn)) = self.goals.event(&event, bot_id, live, &client, &checked) { + for lookout in self.lookouts.values_mut() { + lookout.updates.goal_turn(&bot, turn); + } + } for lookout in self.lookouts.values_mut() { lookout.event(&event); // The oldest event whose news is not told yet: a @@ -693,6 +712,12 @@ impl Service { continue; } } + Some(goals::Settled { bot, turn, outcome }) = settled.recv() => { + self.goals.settled(&bot, turn); + for lookout in self.lookouts.values_mut() { + lookout.updates.goal_settled(&bot, turn, outcome.as_deref()); + } + } // What sources say is news once the stream is live; until // then it waits in the channel. Some(event) = from_sources.recv(), if self.live => { @@ -716,6 +741,7 @@ impl Service { // one this lookout answered, else now and then. if served.subscribed() || reread.is_none_or(|at| now >= at + REREAD) { reread = Some(now); + self.goals.reread(); self.runner.reload(); self.reread(&client).await; } @@ -939,16 +965,19 @@ impl Service { } } - /// Up to just before the oldest news a lookout has not told, which a - /// restart or a reconnect takes up again: news only in memory, or a - /// read a dropped connection cut short, is told again, not lost. + /// Up to just before the oldest news a lookout has not told, or the + /// oldest turn whose goal's check runs, which a restart or a reconnect + /// takes up again: news only in memory, or a read or check a dropped + /// connection cut short, is told or run again, not lost. fn floor(&mut self) -> Option { for lookout in self.lookouts.values_mut() { if !lookout.untold() { lookout.from = None; } } - let floor = self.lookouts.values().filter_map(|l| l.from).min(); + let floor = (self.lookouts.values().filter_map(|l| l.from)) + .chain(self.goals.floor()) + .min(); match (self.cursor, floor) { (Some(at), Some(floor)) => Some(at.min(floor - 1)), (at, _) => at, @@ -1203,8 +1232,11 @@ impl Lookout { /// Who wrote a turn, as the note names them; `None` for the /// subscriber, whose words are no news to it. fn author(&self, data: &Value) -> Option { - // A lookout's own note to a lead another subscriber watches. - if (data["request_id"].as_str()).is_some_and(|id| id.starts_with("lookout-")) { + // A lookout's own note to a lead another subscriber watches, or a + // goal's next turn, whose end alone is news. + if (data["request_id"].as_str()) + .is_some_and(|id| id.starts_with("lookout-") || id.starts_with(goals::REQUEST)) + { return None; } match (data["from"]["bot"].as_str(), data["origin"].as_str()) { @@ -1406,9 +1438,12 @@ impl Lookout { } /// Whether anything it heard is not told yet: messages to read or - /// being read, or news. + /// being read, a turn waiting on its goal's check, or news. fn untold(&self) -> bool { - !self.due.is_empty() || self.reading > 0 || self.updates.news() + !self.due.is_empty() + || self.reading > 0 + || self.updates.goal_pending() + || self.updates.news() } /// Tell the subscriber what happened: at once, cutting into its turn, @@ -2068,6 +2103,7 @@ mod tests { resume: None, reacting: HashMap::new(), taken: HashMap::new(), + goals: goals::Goals::new(Path::new("/nonexistent"), Path::new("")), } } diff --git a/sonny/src/lookout/updates.rs b/sonny/src/lookout/updates.rs index 67d490f4d..fda2a9351 100644 --- a/sonny/src/lookout/updates.rs +++ b/sonny/src/lookout/updates.rs @@ -40,6 +40,8 @@ const REPLY_BYTES: usize = 80 * 1024; /// How long a trigger has to pass a task's answer on before the turn is /// news after all. const FORWARD: Duration = Duration::from_secs(15); +/// How long a goal's turn waits for its check, which says sooner. +const GOAL_HOLD: Duration = Duration::from_secs(super::goals::TIMEOUT_MAX_S + 60); const MEMORY_LINES: usize = 20; const ACTIVE: [&str; 5] = ["running", "waiting", "paced", "queued", "ready"]; const APPROVAL: &str = "waiting for the person's approval"; @@ -62,6 +64,8 @@ struct Turn { which: Option, /// The turn resolved the task, with the task's own `resolve_task`. resolved: bool, + /// How the task's goal ended with this turn. + goal: Option, /// What the task replied, ending the turn. reply: Option, } @@ -86,11 +90,18 @@ impl News { fn merge(&mut self, news: News) { fn one(had: Option, new: Option) -> Option { match (had, new) { - (Some(a), Some(t)) => Some(Turn { - first: a.first.min(t.first), - count: a.count + t.count, - ..if t.turn >= a.turn { t } else { a } - }), + (Some(a), Some(t)) => { + let (first, count) = (a.first.min(t.first), a.count + t.count); + let (newer, older) = if t.turn >= a.turn { (t, a) } else { (a, t) }; + // How a goal ended stays news when a later turn joins it. + let goal = newer.goal.clone().or(older.goal); + Some(Turn { + first, + count, + goal, + ..newer + }) + } (a, t) => a.or(t), } } @@ -175,8 +186,11 @@ pub struct Updates { from: HashMap, origin: HashMap, answer_to: HashMap, - /// Turns whose answer a trigger passes to the lead. + /// Turns whose answer a trigger passes to the lead, or whose end + /// waits for their goal's check. forwarding: HashMap, + /// Turns whose end waits for their goal's check, until it says. + goaled: HashSet, /// Approvals not yet answered: when they become news, and the call. approvals: HashMap)>, /// Each task's news, in the order it first came. @@ -241,6 +255,7 @@ impl Updates { origin: HashMap::new(), answer_to: HashMap::new(), forwarding: HashMap::new(), + goaled: HashSet::new(), approvals: HashMap::new(), tasks: Vec::new(), quiet: HashMap::new(), @@ -340,7 +355,19 @@ impl Updates { if let (Some(task), Some(of)) = (data["from"]["bot"].as_str(), data["from"]["turn"].as_i64()) { - self.forwarding.remove(&(task.to_owned(), of)); + let key = (task.to_owned(), of); + // How its goal ends is still to tell: only its + // answer went on. + match self.goaled.contains(&key) { + true => { + if let Some(forward) = self.forwarding.get_mut(&key) { + forward.reply = None; + } + } + false => { + self.forwarding.remove(&key); + } + } } } else if let Some(origin) = data["origin"].as_str() { self.origin.insert(key(turn), origin.to_owned()); @@ -438,11 +465,17 @@ impl Updates { .creator(name) .and_then(|c| self.bots.get(c)) .and_then(|c| c.id); - if answered.is_some() && answered == creator { + let hold = match self.goaled.contains(&key(turn)) { + // News once its goal's check says, or once a check + // could have run its longest. + true => Some(GOAL_HOLD), + false => (answered.is_some() && answered == creator).then_some(FORWARD), + }; + if let Some(hold) = hold { self.forwarding.insert( key(turn), Forward { - at: now + FORWARD, + at: now + hold, status, from, origin, @@ -563,6 +596,21 @@ impl Updates { from: Option, which: Option, origin: Option, + ) { + self.tell_ended(name, turn, status, from, which, origin, None); + } + + /// `tell`, with how the task's goal ended with the turn. + #[allow(clippy::too_many_arguments)] + fn tell_ended( + &mut self, + name: &str, + turn: i64, + status: &str, + from: Option, + which: Option, + origin: Option, + goal: Option, ) { // A note while the turn still runs leaves its resolve for the last. let resolved = ![APPROVAL, QUIET].contains(&status) @@ -572,7 +620,8 @@ impl Updates { return; } let handle = format!("turn:{name}/{turn}"); - if self.bots[&self.lead].waiting_on.contains(&handle) { + // A wait returns the turn, never how its goal ended. + if goal.is_none() && self.bots[&self.lead].waiting_on.contains(&handle) { return; } let by = match status { @@ -595,6 +644,7 @@ impl Updates { count: 1, which, resolved, + goal, reply: None, }; news.merge(match theirs { @@ -633,6 +683,7 @@ impl Updates { .map(|(key, _)| key.clone()) .collect(); for key in late { + self.goaled.remove(&key); if let Some(forward) = self.forwarding.remove(&key) { let Forward { status, @@ -658,14 +709,55 @@ impl Updates { } } + /// A turn about to end whose goal's check runs once it does: its end + /// waits for `goal_settled`, if it is the turn of a task whose updates + /// the lead hears; another's end is judged as any turn's. + pub fn goal_turn(&mut self, name: &str, turn: i64) { + if self.live && self.task(name) { + self.goaled.insert((name.to_owned(), turn)); + } + } + + /// Whether a turn waits for its goal's check: the replay after a + /// restart starts before it, so the check runs again. + pub fn goal_pending(&self) -> bool { + !self.goaled.is_empty() + } + + /// The goal's check of a turn said: no news when the task works again + /// (`None`), else the turn, with how its goal ended after its status. + pub fn goal_settled(&mut self, name: &str, turn: i64, outcome: Option<&str>) { + self.goaled.remove(&(name.to_owned(), turn)); + let Some(forward) = self.forwarding.remove(&(name.to_owned(), turn)) else { + return; + }; + let Some(outcome) = outcome else { + return; + }; + let Forward { + status, + from, + origin, + reply, + .. + } = forward; + let goal = Some(outcome.to_owned()).filter(|outcome| !outcome.is_empty()); + self.tell_ended(name, turn, &status, from, None, origin, goal); + if let Some(reply) = reply { + self.replied(name, turn, &reply); + } + } + /// The lead read these turns with its wait tool: what they said is no /// news to it any more. pub fn read(&mut self, handles: &HashSet) { for (name, news) in &mut self.tasks { for kind in [&mut news.act, &mut news.theirs] { + // How a goal ended is no part of what the wait read. if kind.as_ref().is_some_and(|t| { t.status != APPROVAL && t.status != QUIET + && t.goal.is_none() && handles.contains(&format!("turn:{name}/{}", t.turn)) }) { *kind = None; @@ -673,8 +765,11 @@ impl Updates { } } self.tasks.retain(|(_, news)| !news.empty()); - self.forwarding - .retain(|(name, turn), _| !handles.contains(&format!("turn:{name}/{turn}"))); + // A turn its goal's check holds stays: the wait read how it ended, + // not how its goal did. + self.forwarding.retain(|key, _| { + self.goaled.contains(key) || !handles.contains(&format!("turn:{}/{}", key.0, key.1)) + }); } /// Whether there is news for the lead. @@ -853,6 +948,7 @@ impl Updates { pub fn forget(&mut self) { self.tasks.clear(); self.forwarding.clear(); + self.goaled.clear(); self.approvals.clear(); } @@ -882,6 +978,7 @@ impl Updates { fn drop_news(&mut self, name: &str) { self.tasks.retain(|(task, _)| task != name); self.forwarding.retain(|(task, _), _| task != name); + self.goaled.retain(|(task, _)| task != name); self.approvals.retain(|(task, _), _| task != name); self.noted.retain(|(task, _)| task != name); } @@ -960,6 +1057,9 @@ fn line(name: &str, theirs: bool, t: &Turn) -> Line { if t.resolved { text.push_str(", and resolved itself"); } + if let Some(goal) = &t.goal { + text.push_str(goal); + } if t.count > 1 { text.push_str(&format!( ", and {} earlier since turn:{name}/{}", @@ -1261,6 +1361,123 @@ mod tests { assert!(updates.due()); } + #[test] + fn a_goal_s_turns_are_news_only_once_it_ends() { + let mut updates = fleet(); + let lead = json!({"from": {"bot": "shop.lead", "turn": 9}}); + let goal = json!({"origin": "goal"}); + // The check failed and the task works again: no news. + updates.goal_turn("shop.fix", 4); + turn(&mut updates, "shop.fix", 4, lead.clone(), "completed"); + assert!(!updates.due(), "held for its check"); + updates.goal_settled("shop.fix", 4, None); + assert!(!updates.due(), "it works again"); + // The check passed: the turn, with how the goal ended. + updates.goal_turn("shop.fix", 5); + turn(&mut updates, "shop.fix", 5, goal.clone(), "completed"); + updates.goal_settled("shop.fix", 5, Some(", and met its goal: its check exits 0")); + let told = updates.take(); + assert_eq!( + text(&told), + "- shop.fix: turn:shop.fix/5 completed, asked by goal, and met its goal: its check exits 0" + ); + updates.sent(told); + // How it ended stays news when a later turn joins it, whichever + // settles first. + updates.goal_turn("shop.fix", 6); + turn(&mut updates, "shop.fix", 6, goal.clone(), "completed"); + turn(&mut updates, "shop.fix", 7, lead.clone(), "completed"); + updates.goal_settled("shop.fix", 6, Some(", and met its goal: its check exits 0")); + turn(&mut updates, "shop.fix", 8, lead.clone(), "completed"); + let told = updates.take(); + assert!(text(&told).contains("met its goal"), "{}", text(&told)); + updates.sent(told); + // The lead's wait read the turn while its check ran: how the goal + // ended is still news. + updates.goal_turn("shop.fix", 9); + turn(&mut updates, "shop.fix", 9, goal.clone(), "completed"); + updates.read(&["turn:shop.fix/9".to_owned()].into()); + updates.goal_settled("shop.fix", 9, Some(", and met its goal: its check exits 0")); + let told = updates.take(); + assert!(text(&told).contains("turn:shop.fix/9 completed, asked by goal, and met its goal")); + updates.sent(told); + // Or read it once its check had said. + updates.goal_turn("shop.fix", 10); + turn(&mut updates, "shop.fix", 10, goal.clone(), "completed"); + updates.goal_settled( + "shop.fix", + 10, + Some(", and met its goal: its check exits 0"), + ); + updates.read(&["turn:shop.fix/10".to_owned()].into()); + let told = updates.take(); + assert!( + text(&told).contains("turn:shop.fix/10 completed, asked by goal, and met its goal") + ); + updates.sent(told); + // Its check said while the lead's wait still held the turn. + updates.goal_turn("shop.fix", 12); + turn(&mut updates, "shop.fix", 12, goal.clone(), "completed"); + updates + .bots + .get_mut("shop.lead") + .unwrap() + .waiting_on + .push("turn:shop.fix/12".to_owned()); + updates.goal_settled( + "shop.fix", + 12, + Some(", and met its goal: its check exits 0"), + ); + updates + .bots + .get_mut("shop.lead") + .unwrap() + .waiting_on + .clear(); + let told = updates.take(); + assert!( + text(&told).contains("turn:shop.fix/12 completed, asked by goal, and met its goal") + ); + updates.sent(told); + // A trigger passed its answer on while its check ran: how the goal + // ended is still news, without the answer. + updates.goal_turn("shop.fix", 11); + let fire = json!({"origin": "goal", "request_id": "trigger_t1.to.1"}); + turn(&mut updates, "shop.fix", 11, fire, "completed"); + updates.replied("shop.fix", 11, "passed on"); + let passed = json!({"from": {"bot": "shop.fix", "turn": 11}}); + updates.event( + &ev("shop.lead", Some(10), "accepted", passed), + Instant::now(), + ); + updates.goal_settled( + "shop.fix", + 11, + Some(", and met its goal: its check exits 0"), + ); + let told = text(&updates.take()); + assert!(told.contains("turn:shop.fix/11 completed"), "{told}"); + assert!( + told.contains("met its goal") && !told.contains("passed on"), + "{told}" + ); + // Cleared while its check ran: plain news. + updates.goal_turn("shop.docs", 2); + turn(&mut updates, "shop.docs", 2, lead, "completed"); + updates.goal_settled("shop.docs", 2, Some("")); + assert_eq!( + text(&updates.take()), + "- shop.docs: turn:shop.docs/2 completed, asked by you" + ); + // A check that never says lets the turn go once it could have run + // its longest. + updates.goal_turn("shop.docs", 3); + turn(&mut updates, "shop.docs", 3, goal, "completed"); + updates.expire(Instant::now() + GOAL_HOLD); + assert!(text(&updates.take()).starts_with("- shop.docs: turn:shop.docs/3 completed")); + } + #[test] fn a_message_that_started_a_turn_told_as_ended_is_not_told_twice() { let mut updates = fleet(); diff --git a/sonny/src/served.rs b/sonny/src/served.rs index 6f7a93835..7cbc4f4e9 100644 --- a/sonny/src/served.rs +++ b/sonny/src/served.rs @@ -30,6 +30,7 @@ pub const PROJECT: &str = "new_project"; pub const SUBSCRIPTIONS: &str = "subscriptions"; pub const TRIGGERS: &str = "triggers"; pub const MEMORY: &str = "memory"; +pub const GOAL: &str = "goal"; pub const EXEC: &str = crate::exec::NAME; /// The tools whose calls wait for a verdict wherever the agent's other @@ -37,9 +38,18 @@ pub const EXEC: &str = crate::exec::NAME; /// lead, outside the agent's sandbox as a pane's command is, a removal can /// drop work, a subscription sends other bots' words to the subscriber's /// judge, a trigger is a job that runs as the person, its `if` -/// command outside any sandbox, and a script calls MCP servers, which run -/// as the person too. -pub const GATED: [&str; 7] = [START, REMOVE, PROJECT, SUBSCRIPTIONS, RUN, TRIGGERS, EXEC]; +/// command outside any sandbox, a script calls MCP servers, which run +/// as the person too, and so does a goal's check. +pub const GATED: [&str; 8] = [ + START, + REMOVE, + PROJECT, + SUBSCRIPTIONS, + RUN, + TRIGGERS, + EXEC, + GOAL, +]; /// The most of what a call printed that its result holds, under the /// daemon's bound on an answer. @@ -59,7 +69,13 @@ pub fn tools(lead: bool) -> Vec { if lead { tools.extend([start(), remove(), project(), subscriptions()]); } - tools.extend([show(), run(), triggers(), memory()]); + tools.extend([ + show(), + run(), + triggers(), + memory(), + crate::lookout::goals::tool(GOAL, TAG), + ]); tools.extend(crate::exec::tool()); tools } @@ -468,6 +484,16 @@ async fn called(name: &str, arguments: &Value) -> Answer { .map(|listed| listed.to_string()) .map_err(|e| ("subscription_refused", e)) } + GOAL => { + let (store, socket) = crate::daemon_paths(None, None).map_err(|e| ("no_store", e))?; + let (client, _events) = Client::connect(&socket) + .await + .map_err(|e| ("goal_refused", e.to_string()))?; + crate::lookout::goals::call(&store, &client, &caller, arguments) + .await + .map(|said| said.to_string()) + .map_err(|e| ("goal_refused", e)) + } MEMORY if arguments["said"] == true => crate::said::read(arguments["since_ms"].as_i64()) .await .map(|said| said.to_string()) @@ -605,7 +631,8 @@ async fn serve( "call_id": call["call_id"]}); match answer { Ok(output) => { - if call["name"] == SUBSCRIPTIONS { + // Files the lookout reads, read now. + if call["name"] == SUBSCRIPTIONS || call["name"] == GOAL { subscribed.store(true, Ordering::Relaxed); } asked["output"] = json!(cut(output)) @@ -976,7 +1003,7 @@ mod tests { .filter(|name| name != EXEC) .collect() }; - assert_eq!(names(tools(false)), [SHOW, RUN, TRIGGERS, MEMORY]); + assert_eq!(names(tools(false)), [SHOW, RUN, TRIGGERS, MEMORY, GOAL]); assert_eq!( names(tools(true)), [ @@ -987,7 +1014,8 @@ mod tests { SHOW, RUN, TRIGGERS, - MEMORY + MEMORY, + GOAL ] ); for tool in tools(true).into_iter().filter(|t| t["name"] != EXEC) { diff --git a/sonny/src/task.rs b/sonny/src/task.rs index 3d6034314..77856b621 100644 --- a/sonny/src/task.rs +++ b/sonny/src/task.rs @@ -665,7 +665,7 @@ fn within(copy: &Path, prefix: &Path) -> PathBuf { /// The provider credentials this process may hold, as the daemon keeps /// them from a bot's shell: the well-known keys, Bedrock's, the key a /// provider in `AGENT_PROVIDER` names, and what `~/.agent/env` sets. -fn kept_keys() -> Vec { +pub(crate) fn kept_keys() -> Vec { let mut keys: Vec = [ "ANTHROPIC_API_KEY", "OPENAI_API_KEY", diff --git a/tests/test_served.py b/tests/test_served.py index 77ca7adfe..b00acb88e 100644 --- a/tests/test_served.py +++ b/tests/test_served.py @@ -291,6 +291,142 @@ def test_a_lead_made_for_a_subfolder_reads_its_own_memory_and_follows_its_folder [kept] = (self.path / '.agent/sonny/instructions').iterdir() self.assertTrue(kept.name.endswith(f'-{lead["bot_id"]}'), kept.name) + def note_for(self, bot, having, timeout=30): + """The first note the lookout gave `bot` that says `having`.""" + deadline = time.monotonic() + timeout + while time.monotonic() < deadline: + for turn in self.daemon.request('turns', bot=bot)['result']['turns']: + text = self.daemon.request('prompts', bot=bot, turn=turn['turn'])['result']['prompts'] + if text and text[0]['text'].startswith('Task updates: ') and having in text[0]['text']: + return text[0]['text'] + time.sleep(.2) + self.fail(f'{bot} heard nothing saying {having!r}') + + def goal_watch(self): + """calc.fix, calc.lead's task with the goal tool, watched by its + lead, which hears its turns; the lookout serving them.""" + lead = self.daemon.request('create', bot='calc.lead', workspace=str(self.path), + model='openai/synthetic-model', instructions='Lead.', + declare=[declared('goal')])['result'] + self.daemon.request('create', bot='calc.fix', workspace=str(self.path), model='openai/synthetic-model', + instructions='Task.', declare=[declared('goal')], created_by='calc.lead', + created_by_id=lead['bot_id']) + folder = self.path / 'state.sqlite.lookout' + folder.mkdir(exist_ok=True) + watch = {'name': 'tasks', 'tasks_of': [], 'bots': ['calc.fix'], 'on': ['said', 'ended'], 'updates': True} + (folder / 'calc.lead.json').write_text(json.dumps({'subscriptions': [watch]})) + return self.lookout() + + def lines_of(self, path, at_least, timeout=30): + """The lines of `path` once it has `at_least` of them.""" + deadline = time.monotonic() + timeout + while not path.exists() or len(path.read_text().splitlines()) < at_least: + self.assertLess(time.monotonic(), deadline, f'{path.name} has fewer than {at_least} lines') + time.sleep(.05) + return path.read_text().splitlines() + + def goal_run(self, check): + """calc.fix sets itself a goal with `check`: the lead's note once + the goal ends, and the prompts of calc.fix's turns.""" + self.goal_watch() + set_ = json.loads(self.call('calc.fix', 'goal', {'set': {'done_when': 'the flag exists', 'check': check}})) + self.assertIn('goal', set_, set_) + self.assertEqual(set_['goal']['check'], check) + note = self.note_for('calc.lead', 'its goal') + prompts = [self.daemon.request('prompts', bot='calc.fix', turn=t['turn'])['result']['prompts'][0]['text'] + for t in self.daemon.request('turns', bot='calc.fix')['result']['turns']] + return note, prompts + + def test_a_goal_keeps_its_agent_working_until_its_check_passes(self): + # Fails once, leaving the flag that passes it the next time. + note, prompts = self.goal_run('test -f flag || { touch flag; echo not yet; exit 1; }') + self.assertEqual(len(prompts), 2, prompts) + self.assertTrue(prompts[1].startswith('Your goal is not met yet: `test -f flag'), prompts[1]) + self.assertIn('exits 1.\n```\nnot yet\n```\nKeep working until the flag exists.', prompts[1]) + # The lead hears the turn that met it, and nothing of the one before. + self.assertIn('- calc.fix: turn:calc.fix/2 completed, asked by goal, and met its goal:', note) + self.assertNotIn('calc.fix/1', note) + self.assertFalse((self.path / 'state.sqlite.lookout/goals/calc.fix.json').exists()) + + def test_a_goal_whose_agent_only_answers_stops(self): + note, prompts = self.goal_run('false') + # The turn that set it called the tool; two that only answered end it. + self.assertEqual(len(prompts), 3, prompts) + self.assertIn('turn:calc.fix/3 completed, asked by goal, and stopped pursuing its goal: 2 turns in a row', + note) + self.assertIn('its check still exits 1', note) + + def test_a_goal_its_lead_sets_counts_from_the_task_turn_that_ends_next(self): + self.goal_watch() + ran = self.path / 'ran' + set_ = json.loads(self.call('calc.lead', 'goal', {'set': {'bot': 'calc.fix', 'done_when': 'it ran', + 'check': f'echo . >> {ran}'}})) + self.assertEqual(set_['goal']['bot'], 'calc.fix', set_) + # Ends at once, well before the lookout reads its goals again. + turn = self.daemon.request('submit', bot='calc.fix', request_id='work', prompt='Work.')['result']['turn'] + self.assertEqual(self.daemon.finished(turn)['data']['status'], 'completed') + note = self.note_for('calc.lead', 'its goal') + self.assertIn(f'- calc.fix: turn:calc.fix/{turn} completed, asked by the person, and met its goal:', note) + self.assertEqual(len(ran.read_text().splitlines()), 1) + + def test_a_turn_that_ends_while_its_goal_is_checked_starts_no_other_check(self): + self.goal_watch() + runs = self.path / 'runs' + # Slow, and passing the second time it runs. + check = f'echo . >> {runs}; sleep 2; test $(wc -l < {runs}) -ge 2' + self.call('calc.fix', 'goal', {'set': {'done_when': 'it ran twice', 'check': check}}) + self.lines_of(runs, 1) + # The person writes in while it runs; the turn ends before it does. + turn = self.daemon.request('submit', bot='calc.fix', request_id='more', prompt='More.')['result']['turn'] + self.assertEqual(self.daemon.finished(turn)['data']['status'], 'completed') + note = self.note_for('calc.lead', 'its goal') + self.assertIn('asked by goal, and met its goal:', note) + # One check for the turn that set it, one for the goal's own turn. + self.assertEqual(len(runs.read_text().splitlines()), 2) + + def test_a_goal_checked_when_its_lookout_gives_way_is_checked_by_the_next(self): + old = self.goal_watch() + runs = self.path / 'runs' + check = f'echo . >> {runs}; sleep 30; exit 1' + self.call('calc.fix', 'goal', {'set': {'done_when': 'it ran again', 'check': check}}) + self.lines_of(runs, 1) + # The cursor it keeps stays before the turn's end, which waits on + # the check. + [ended] = [e['cursor'] for e in self.daemon.request('events', bot='calc.fix', after=0, limit=256) + ['result']['events'] if e['event'] == 'turn_finished'] + saved = self.lines_of(self.path / 'state.sqlite.lookout.cursor', 1) + self.assertLess(int(saved[0]), ended) + # Another build's lookout takes over, stopping this one's check. + (self.path / 'state.sqlite.lookout.sonny').write_text('0.0.1 1') + (self.path / 'state.sqlite.lookout/goals/calc.fix.json').write_text(json.dumps(dict( + json.loads((self.path / 'state.sqlite.lookout/goals/calc.fix.json').read_text()), + check=f'echo . >> {runs}'))) + self.lookout() + old.wait(timeout=10) + # The turn that set it is checked again, and now passes. + note = self.note_for('calc.lead', 'its goal') + self.assertIn('- calc.fix: turn:calc.fix/1 completed, asked by the person, and met its goal:', note) + self.assertEqual(len(self.lines_of(runs, 2)), 2) + + def test_a_goal_no_one_watches_is_checked_again_after_a_restart(self): + self.daemon.request('create', bot='calc.lead', workspace=str(self.path), model='openai/synthetic-model', + instructions='Lead.', declare=[declared('goal')]) + old = self.lookout() + runs = self.path / 'runs' + self.call('calc.lead', 'goal', {'set': {'done_when': 'it ran again', 'check': f'echo . >> {runs}; sleep 30'}}) + self.lines_of(runs, 1) + [ended] = [e['cursor'] for e in self.daemon.request('events', bot='calc.lead', after=0, limit=256) + ['result']['events'] if e['event'] == 'turn_finished'] + # Past a save of the cursor, which stays before the turn's end + # though no subscriber holds it. + time.sleep(1.5) + saved = self.lines_of(self.path / 'state.sqlite.lookout.cursor', 1) + self.assertLess(int(saved[0]), ended) + old.kill() + old.wait(timeout=10) + self.lookout() + self.assertEqual(len(self.lines_of(runs, 2)), 2) + @unittest.skipIf(sys.platform == 'darwin', 'launchd keeps triggers on a Mac') def test_a_trigger_with_no_systemd_running_is_refused_and_says_why(self): self.daemon.request('create', bot='calc.lead', workspace=str(self.path), model='openai/synthetic-model',