From 88674eb63ac5766730b62d400c8f3b7c968f847c Mon Sep 17 00:00:00 2001 From: atlas Date: Sun, 18 Oct 2026 03:11:15 +0200 Subject: [PATCH] read an agent's agents.md through the swarm and edit it with git diffs The agent's state/AGENTS.md stays the truth and its harness the only writer. The harness mirrors the file into a new `agents-md` KV bucket, and swarm-controller serves that copy, so a read works while the agent is stopped. An edit is a unified diff plus the revision it was made against, sent over the agent's existing control subject; the harness refuses a stale base, applies the diff with `git apply --check` then `git apply` to a temp copy, and writes the file only when both pass. Fixes #5519. Files: - swarm-queue-client/src/agents_md.rs, lib.rs: the bucket name and subject, the mirror value, the `ApplyAgentsMd` control command and its reply. `AgentControl` loses `Copy` because the new variant carries text. - hive-agent/src/swarm_agents_md.rs, main.rs: the mirror task (start, 10 s re-read, immediate republish after an edit) and the apply. - hive-agent/src/swarm_input.rs: the control loop answers both commands. - hive-agent/src/claude_md_watch.rs: its AGENTS.md path is shared. - hive-agent/Cargo.toml, Cargo.lock: sha1 for the git blob id (already in the lock), tempfile moves from dev to runtime for the apply directory. - swarm-controller/src/agents_md.rs, main.rs: the bucket reader and GET/PUT /api/agents/{name}/agents-md; test AppState literals get the new field. - swarm-controller/src/agent_control.rs: the sink hands back raw reply bytes, so each command decodes its own reply type. - swarm-controller-api/src/agents_md.rs: the route bodies, shared with swarmctl. - swarmctl/src/agents_md.rs, main.rs: `swarmctl agent agents-md get|apply`. - nix/host-modules/swarm-nats.nix, nix/module-eval/nats-authelia.nix: the agent's publish grant on its own `$KV.agents-md.` key. git needs no new package: agent containers enable programs.git, and the harness unit's PATH includes /run/current-system/sw. --- Cargo.lock | 1 + hive-agent/Cargo.toml | 6 +- hive-agent/src/claude_md_watch.rs | 2 +- hive-agent/src/main.rs | 4 + hive-agent/src/swarm_agents_md.rs | 437 ++++++++++++++++++++++++ hive-agent/src/swarm_input.rs | 115 +++++-- nix/host-modules/swarm-nats.nix | 4 + nix/module-eval/nats-authelia.nix | 1 + swarm-controller-api/src/agents_md.rs | 51 +++ swarm-controller-api/src/lib.rs | 1 + swarm-controller/src/agent_control.rs | 25 +- swarm-controller/src/agent_limits.rs | 1 + swarm-controller/src/agents_md.rs | 436 +++++++++++++++++++++++ swarm-controller/src/forge_account.rs | 1 + swarm-controller/src/github_account.rs | 1 + swarm-controller/src/linked_accounts.rs | 1 + swarm-controller/src/main.rs | 21 +- swarm-controller/src/matrix_account.rs | 1 + swarm-queue-client/src/agents_md.rs | 103 ++++++ swarm-queue-client/src/lib.rs | 13 +- swarmctl/src/agents_md.rs | 68 ++++ swarmctl/src/main.rs | 50 +++ 22 files changed, 1296 insertions(+), 47 deletions(-) create mode 100644 hive-agent/src/swarm_agents_md.rs create mode 100644 swarm-controller-api/src/agents_md.rs create mode 100644 swarm-controller/src/agents_md.rs create mode 100644 swarm-queue-client/src/agents_md.rs create mode 100644 swarmctl/src/agents_md.rs diff --git a/Cargo.lock b/Cargo.lock index 8d768a19..8ecce245 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1688,6 +1688,7 @@ dependencies = [ "schemars", "serde", "serde_json", + "sha1", "swarm-queue-client", "swarm-secret-client", "tempfile", diff --git a/hive-agent/Cargo.toml b/hive-agent/Cargo.toml index ffd7e439..11e68f9c 100644 --- a/hive-agent/Cargo.toml +++ b/hive-agent/Cargo.toml @@ -38,6 +38,8 @@ rusqlite.workspace = true schemars.workspace = true serde.workspace = true serde_json.workspace = true +# The git blob id of AGENTS.md, the revision an edit from the swarm names. +sha1 = "0.10" # `agent-schedules`, `agent-inference` and `operator-profiles`: the schedule # ticker and the inference poller each read their own key, and the prompt # renderer the operator list, with one direct get. @@ -51,6 +53,9 @@ swarm-queue-client = { workspace = true, features = [ ] } # This agent's own queue secret, read from the store under its own certificate. swarm-secret-client.workspace = true +# The directory an AGENTS.md edit is applied in, and the file it is written +# through. +tempfile = "3" tokio.workspace = true tokio-stream.workspace = true tower-http.workspace = true @@ -59,7 +64,6 @@ tracing.workspace = true # `test-util` for `#[tokio::test(start_paused = true)]`: `swarm_agent_state`'s # refresh test waits out its one-minute interval on paused time. [dev-dependencies] -tempfile = "3" tokio = { workspace = true, features = ["test-util"] } tower = { version = "0.5", features = ["util"] } # A layer that captures `swarm_queue`'s exhausted-connect log event in a test. diff --git a/hive-agent/src/claude_md_watch.rs b/hive-agent/src/claude_md_watch.rs index 8f88958c..80ee12e9 100644 --- a/hive-agent/src/claude_md_watch.rs +++ b/hive-agent/src/claude_md_watch.rs @@ -89,7 +89,7 @@ fn reconcile(todos: &Todos, wake: &Notify, summary: Option<&str>) { /// `AGENTS.md`'s path: the top of the agent's own state dir, matching /// where `claude --print` runs (`turn::session_cwd`) and therefore where /// its own auto-load discovers the file. -fn agents_md_path(state_dir: &Path) -> std::path::PathBuf { +pub(crate) fn agents_md_path(state_dir: &Path) -> std::path::PathBuf { state_dir.join("AGENTS.md") } diff --git a/hive-agent/src/main.rs b/hive-agent/src/main.rs index 93024155..6d210163 100644 --- a/hive-agent/src/main.rs +++ b/hive-agent/src/main.rs @@ -35,6 +35,7 @@ mod stats; mod subagent_role_watch; mod swarm_agent_icon; mod swarm_agent_state; +mod swarm_agents_md; mod swarm_inference; mod swarm_input; mod swarm_permissions; @@ -504,6 +505,9 @@ async fn serve_main(socket: &Path, poll_ms: u64) -> Result<()> { swarm_agent_state::spawn(&bus, login_state.clone()); // Once per start: a new icon only arrives with a new harness process. swarm_agent_icon::spawn(); + // At start and on every change, so the swarm can read it while the + // agent is stopped. + swarm_agents_md::spawn(); // The model and effort the swarm sets for this agent, over its local ones. tokio::spawn(swarm_inference::run(bus.clone())); // The permissions the swarm grants this agent, read before each turn. diff --git a/hive-agent/src/swarm_agents_md.rs b/hive-agent/src/swarm_agents_md.rs new file mode 100644 index 00000000..fe744747 --- /dev/null +++ b/hive-agent/src/swarm_agents_md.rs @@ -0,0 +1,437 @@ +//! This agent's `AGENTS.md`, mirrored into the swarm's `agents-md` bucket and +//! edited by diffs the swarm sends. +//! +//! **The file in the state dir is the truth**, and this harness is the only +//! swarm-side writer of it. The mirror is a copy for reading while the agent +//! is stopped; nothing reads the mirror back into the file. +//! +//! **Mirror.** The file's text and revision (its git blob id) go to +//! [`swarm_queue_client::agents_md::subject`] at start, whenever a re-read +//! every [`POLL`] finds a new revision, and straight after an applied edit. +//! The write is a `JetStream` publish to the key's subject, as +//! [`crate::swarm_agent_icon`] does it, so the grant is that one subject; a +//! failed write is retried on the next poll. +//! +//! **Edit.** [`handle`] answers an [`Apply`] from the control subject: it +//! refuses a diff made against another revision, applies the diff with +//! `git apply --check` and then `git apply` to a copy of the file in a +//! temporary directory, and writes the result over the file only when both +//! pass. A diff that does not apply is refused with git's message and the +//! file is left as it was. `git` comes from the container's `PATH` +//! (`programs.git` in the agent's NixOS config). + +use std::fmt::Write as _; +use std::future::Future; +use std::io::Write as _; +use std::path::Path; +use std::time::Duration; + +use async_nats::jetstream::context::PublishErrorKind; +use sha1::{Digest as _, Sha1}; +use swarm_queue_client::agents_md::{Apply, ApplyReply, Mirror}; +use tokio::sync::Notify; + +use crate::claude_md_watch::agents_md_path; + +/// How often the file is re-read for a change the agent made itself. +const POLL: Duration = Duration::from_secs(10); + +/// Wakes the mirror after an applied edit, so the swarm sees the new revision +/// without waiting out [`POLL`]. +static REPUBLISH: Notify = Notify::const_new(); + +/// The file name `git apply` sees, the `a/AGENTS.md` / `b/AGENTS.md` of a diff. +const FILE_NAME: &str = "AGENTS.md"; + +/// The git blob id of `bytes`: SHA-1 of `blob \0` and the bytes, what +/// `git hash-object` prints. +fn revision(bytes: &[u8]) -> String { + let mut hash = Sha1::new(); + hash.update(format!("blob {}\0", bytes.len())); + hash.update(bytes); + hash.finalize() + .iter() + .fold(String::with_capacity(40), |mut hex, b| { + // Writing to a `String` cannot fail. + let _ = write!(hex, "{b:02x}"); + hex + }) +} + +/// The file's bytes; a missing file reads as empty, so the swarm can create +/// one with a diff against `""`. +fn read_or_empty(path: &Path) -> std::io::Result> { + match std::fs::read(path) { + Ok(bytes) => Ok(bytes), + Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(Vec::new()), + Err(e) => Err(e), + } +} + +fn read_mirror(state_dir: &Path) -> std::io::Result { + let bytes = read_or_empty(&agents_md_path(state_dir))?; + Ok(Mirror { + revision: revision(&bytes), + content: String::from_utf8_lossy(&bytes).into_owned(), + }) +} + +/// Start the mirror, if this agent has a label. +pub fn spawn() { + let agent = crate::identity::label(); + if agent.is_empty() { + tracing::warn!("this agent has no label; not mirroring its AGENTS.md"); + return; + } + tokio::spawn(async move { + let client = crate::swarm_queue::client().await; + let js = async_nats::jetstream::new(client.clone()); + let subject = swarm_queue_client::agents_md::subject(&agent); + mirror(&crate::paths::state_dir(), POLL, &REPUBLISH, |m| { + publish(&client, &js, &subject, m) + }) + .await; + }); +} + +/// What became of one mirror write. +#[derive(Debug)] +enum Published { + Acked, + /// The text is over the queue's payload limit; this revision is not + /// tried again. + TooLarge, + /// Anything else, rendered for the log; retried on the next poll. + Failed(String), +} + +/// Publish the file whenever its revision differs from the last one written, +/// checking every `poll` and whenever `wake` is notified. Runs forever. +async fn mirror(state_dir: &Path, poll: Duration, wake: &Notify, mut publish: F) +where + F: FnMut(Mirror) -> Fut, + Fut: Future, +{ + let mut written: Option = None; + // A failure that repeats every poll is logged once, until it changes. + let mut last_warning: Option = None; + loop { + let warning = match read_mirror(state_dir) { + Ok(m) if written.as_ref() != Some(&m.revision) => { + let revision = m.revision.clone(); + match publish(m).await { + Published::Acked => { + tracing::info!(%revision, "published this agent's AGENTS.md"); + written = Some(revision); + None + } + Published::TooLarge => { + written = Some(revision); + Some( + "AGENTS.md is over the queue's payload limit; not mirroring it" + .to_owned(), + ) + } + Published::Failed(e) => Some(format!("mirroring AGENTS.md failed: {e}")), + } + } + Ok(_) => None, + Err(e) => Some(format!("reading AGENTS.md failed: {e}")), + }; + if warning.is_some() && warning != last_warning { + tracing::warn!(warning = warning.as_deref(), "agents.md mirror"); + } + last_warning = warning; + tokio::select! { + () = tokio::time::sleep(poll) => {} + () = wake.notified() => {} + } + } +} + +/// One acked write of `mirror` to `subject`. +async fn publish( + client: &async_nats::Client, + js: &async_nats::jetstream::Context, + subject: &str, + mirror: Mirror, +) -> Published { + // An unconnected client buffers a publish rather than failing it. + if let Err(e) = swarm_queue_client::ensure_connected(client) { + return Published::Failed(swarm_queue_client::chain(&e)); + } + let payload = serde_json::to_vec(&mirror).expect("a mirror encodes"); + let ack = match js.publish(subject.to_owned(), payload.into()).await { + Ok(ack) => ack, + Err(e) if e.kind() == PublishErrorKind::MaxPayloadExceeded => return Published::TooLarge, + Err(e) => return Published::Failed(e.to_string()), + }; + // `StreamNotFound` here is the bucket not existing yet. + match ack.await { + Ok(_) => Published::Acked, + Err(e) => Published::Failed(e.to_string()), + } +} + +/// Answer an edit command, and wake the mirror when it changed the file. +pub async fn handle(apply: Apply) -> ApplyReply { + apply_and_wake(crate::paths::state_dir(), apply, &REPUBLISH).await +} + +async fn apply_and_wake(state_dir: std::path::PathBuf, apply: Apply, wake: &Notify) -> ApplyReply { + let reply = tokio::task::spawn_blocking(move || apply_diff(&state_dir, &apply)) + .await + .unwrap_or_else(|e| ApplyReply::Failed { + error: format!("the apply task failed: {e}"), + }); + if matches!(reply, ApplyReply::Applied { .. }) { + wake.notify_one(); + } + reply +} + +/// Apply `apply` to `/AGENTS.md`, or say why not. +fn apply_diff(state_dir: &Path, apply: &Apply) -> ApplyReply { + let failed = |what: &str, e: &dyn std::fmt::Display| ApplyReply::Failed { + error: format!("{what}: {e}"), + }; + let path = agents_md_path(state_dir); + let current = match read_or_empty(&path) { + Ok(bytes) => bytes, + Err(e) => return failed("reading AGENTS.md", &e), + }; + let at = revision(¤t); + if at != apply.base_revision { + return ApplyReply::Stale { revision: at }; + } + let work = match tempfile::tempdir() { + Ok(dir) => dir, + Err(e) => return failed("creating a directory to apply in", &e), + }; + let copy = work.path().join(FILE_NAME); + if let Err(e) = std::fs::write(©, ¤t) { + return failed("copying AGENTS.md to apply to", &e); + } + for args in [&["apply", "--check"][..], &["apply"]] { + match git(work.path(), args, &apply.diff) { + Ok(Ok(())) => {} + Ok(Err(error)) => return ApplyReply::Rejected { error }, + Err(e) => return failed("running git", &e), + } + } + // The diff may name other paths; `git apply` keeps them inside the + // directory, and only a diff that leaves exactly `AGENTS.md` is an edit + // of it. + let left = match std::fs::read_dir(work.path()) { + Ok(entries) => entries + .filter_map(Result::ok) + .map(|e| e.file_name().to_string_lossy().into_owned()) + .collect::>(), + Err(e) => return failed("listing the applied files", &e), + }; + if left != [FILE_NAME] { + return ApplyReply::Rejected { + error: format!("the diff must change AGENTS.md and nothing else; it left {left:?}"), + }; + } + let edited = match std::fs::read(©) { + Ok(bytes) => bytes, + Err(e) => return failed("reading the applied AGENTS.md", &e), + }; + if let Err(e) = replace(state_dir, &path, &edited) { + return failed("writing AGENTS.md", &e); + } + ApplyReply::Applied { + revision: revision(&edited), + } +} + +/// Run `git ` in `dir` with `diff` on stdin. The outer error is git not +/// running; the inner one is git's stderr when it exits non-zero. +fn git(dir: &Path, args: &[&str], diff: &str) -> std::io::Result> { + let mut child = std::process::Command::new("git") + .args(args) + .current_dir(dir) + // Outside a repository `git apply` patches plain files, as `patch` + // does; this keeps it from finding one above the directory. + .env("GIT_CEILING_DIRECTORIES", dir.parent().unwrap_or(dir)) + .env_remove("GIT_DIR") + .env_remove("GIT_WORK_TREE") + .stdin(std::process::Stdio::piped()) + .stdout(std::process::Stdio::null()) + .stderr(std::process::Stdio::piped()) + .spawn()?; + child + .stdin + .take() + .expect("stdin is piped") + .write_all(diff.as_bytes())?; + let out = child.wait_with_output()?; + Ok(if out.status.success() { + Ok(()) + } else { + Err(String::from_utf8_lossy(&out.stderr).trim().to_owned()) + }) +} + +/// Write `bytes` over `path` by renaming a file written beside it, so a +/// reader sees the old text or the new, never part of one. +fn replace(state_dir: &Path, path: &Path, bytes: &[u8]) -> std::io::Result<()> { + let mut tmp = tempfile::NamedTempFile::new_in(state_dir)?; + tmp.write_all(bytes)?; + tmp.persist(path).map_err(|e| e.error)?; + Ok(()) +} + +#[cfg(test)] +mod tests { + use std::time::Duration; + + use swarm_queue_client::agents_md::{Apply, ApplyReply, Mirror}; + use tokio::sync::Notify; + + use super::{Published, apply_and_wake, apply_diff, mirror, revision}; + + const BEFORE: &str = "# atlas\nkeep notes short\n"; + const AFTER: &str = "# atlas\nkeep notes very short\n"; + const DIFF: &str = "diff --git a/AGENTS.md b/AGENTS.md\n\ + --- a/AGENTS.md\n\ + +++ b/AGENTS.md\n\ + @@ -1,2 +1,2 @@\n \ + # atlas\n\ + -keep notes short\n\ + +keep notes very short\n"; + + fn state(text: &str) -> tempfile::TempDir { + let dir = tempfile::tempdir().expect("tempdir"); + std::fs::write(dir.path().join("AGENTS.md"), text).expect("write"); + dir + } + + fn file(dir: &tempfile::TempDir) -> String { + std::fs::read_to_string(dir.path().join("AGENTS.md")).expect("read") + } + + fn apply(base: &str, diff: &str) -> Apply { + Apply { + base_revision: revision(base.as_bytes()), + diff: diff.to_owned(), + } + } + + /// The revision is git's own blob id, so a client can compute it with + /// `git hash-object`. + #[test] + fn the_revision_is_the_git_blob_id() { + assert_eq!( + revision(b"hello\n"), + "ce013625030ba8dba906f756967f9e9ca394464a" + ); + assert_eq!(revision(b""), "e69de29bb2d1d6434b8b29ae775ad8c2e48c5391"); + } + + #[test] + fn a_diff_against_the_current_revision_is_applied() { + let dir = state(BEFORE); + let reply = apply_diff(dir.path(), &apply(BEFORE, DIFF)); + assert_eq!( + reply, + ApplyReply::Applied { + revision: revision(AFTER.as_bytes()) + } + ); + assert_eq!(file(&dir), AFTER); + } + + #[test] + fn a_diff_against_another_revision_is_refused_with_the_current_one() { + let dir = state(BEFORE); + let reply = apply_diff(dir.path(), &apply("something older\n", DIFF)); + assert_eq!( + reply, + ApplyReply::Stale { + revision: revision(BEFORE.as_bytes()) + } + ); + assert_eq!(file(&dir), BEFORE); + } + + /// The base revision matches but the hunk does not: git's refusal comes + /// back and the file is untouched. + #[test] + fn a_diff_that_does_not_apply_is_refused_with_git_s_message() { + let text = "# atlas\nsomething else entirely\n"; + let dir = state(text); + let ApplyReply::Rejected { error } = apply_diff(dir.path(), &apply(text, DIFF)) else { + panic!("applied a diff that does not fit"); + }; + assert!(error.contains("patch does not apply"), "{error}"); + assert_eq!(file(&dir), text); + } + + #[test] + fn a_diff_that_touches_another_file_is_refused() { + let dir = state(BEFORE); + let other = "diff --git a/other.md b/other.md\n\ + new file mode 100644\n\ + --- /dev/null\n\ + +++ b/other.md\n\ + @@ -0,0 +1 @@\n\ + +hi\n"; + let ApplyReply::Rejected { error } = apply_diff(dir.path(), &apply(BEFORE, other)) else { + panic!("applied a diff to another file"); + }; + assert!(error.contains("other.md"), "{error}"); + assert_eq!(file(&dir), BEFORE); + assert!(!dir.path().join("other.md").exists()); + } + + /// No file reads as empty, so a diff against `""` creates it. + #[test] + fn a_missing_file_is_created_from_a_diff_against_empty() { + let dir = tempfile::tempdir().expect("tempdir"); + let create = "--- a/AGENTS.md\n+++ b/AGENTS.md\n@@ -0,0 +1 @@\n+# atlas\n"; + let reply = apply_diff(dir.path(), &apply("", create)); + assert!(matches!(reply, ApplyReply::Applied { .. }), "{reply:?}"); + assert_eq!(file(&dir), "# atlas\n"); + } + + /// The next mirror write, or `None` after ten seconds without one. + async fn next(rx: &mut tokio::sync::mpsc::UnboundedReceiver) -> Option { + tokio::time::timeout(Duration::from_secs(10), rx.recv()) + .await + .ok() + .flatten() + } + + /// The mirror publishes at start, and again straight after an applied + /// edit: the poll here is an hour, so only the wake can explain the + /// second write. + #[tokio::test] + async fn an_applied_edit_is_republished_without_waiting_for_the_poll() { + let dir = state(BEFORE); + let wake = std::sync::Arc::new(Notify::new()); + let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel::(); + let state_dir = dir.path().to_owned(); + let mirror_wake = std::sync::Arc::clone(&wake); + let task = tokio::spawn(async move { + mirror(&state_dir, Duration::from_hours(1), &mirror_wake, |m| { + tx.send(m).expect("test listening"); + std::future::ready(Published::Acked) + }) + .await; + }); + + let first = next(&mut rx).await.expect("published at start"); + assert_eq!(first.content, BEFORE); + assert_eq!(first.revision, revision(BEFORE.as_bytes())); + + let reply = apply_and_wake(dir.path().to_owned(), apply(BEFORE, DIFF), &wake).await; + assert!(matches!(reply, ApplyReply::Applied { .. }), "{reply:?}"); + + let second = next(&mut rx).await.expect("republished"); + assert_eq!(second.content, AFTER); + assert_eq!(second.revision, revision(AFTER.as_bytes())); + task.abort(); + } +} diff --git a/hive-agent/src/swarm_input.rs b/hive-agent/src/swarm_input.rs index 858f355c..5f9ba949 100644 --- a/hive-agent/src/swarm_input.rs +++ b/hive-agent/src/swarm_input.rs @@ -21,10 +21,12 @@ //! request that times out as "a subscriber held it". //! //! **Commands are a separate subject and are answered.** The swarm UI's -//! cancel arrives on [`swarm_queue_client::agent_control_subject`], never on -//! the input subject, so nothing a person types can be read as a command. The -//! reply says whether a turn was actually stopped; the agent's grant allows -//! answering only the request it was just sent (`allow_responses`). +//! cancel and `AGENTS.md` edits arrive on +//! [`swarm_queue_client::agent_control_subject`], never on the input subject, +//! so nothing a person types can be read as a command. A cancel's reply says +//! whether a turn was actually stopped, an edit's whether it was applied +//! ([`crate::swarm_agents_md`]); the agent's grant allows answering only the +//! request it was just sent (`allow_responses`). use futures_util::{Stream, StreamExt as _}; use hive_sh4re::inbox::DeliveredMessage; @@ -70,35 +72,45 @@ async fn run_control( tracing::warn!("swarm control: command without a reply subject dropped"); continue; }; - let Some(answer) = act(&msg.payload, |command| match command { - swarm_queue_client::AgentControl::Cancel => { - crate::web_ui::cancel_turn(&bus, &interrupted, "swarm UI: cancel") - } - }) else { + let Some(answer) = act( + &msg.payload, + || crate::web_ui::cancel_turn(&bus, &interrupted, "swarm UI: cancel"), + crate::swarm_agents_md::handle, + ) + .await + else { continue; }; - let payload = serde_json::to_vec(&answer).expect("a reply encodes"); - if let Err(e) = client.publish(reply, payload.into()).await { + if let Err(e) = client.publish(reply, answer.into()).await { tracing::warn!(error = %e, "swarm control: answering failed"); } } } -/// The answer to a command payload, running `cancel` for a cancel; `None` -/// (logged) when the payload is not a command, which gets no answer. -fn act( +/// The encoded answer to a command payload: `cancel` runs for a cancel and +/// `apply` for an `AGENTS.md` edit. `None` (logged) when the payload is not a +/// command, which gets no answer. +async fn act( payload: &[u8], - cancel: impl FnOnce(swarm_queue_client::AgentControl) -> bool, -) -> Option { - match serde_json::from_slice::(payload) { - Ok(command) => Some(swarm_queue_client::AgentControlReply { - acted: cancel(command), - }), + cancel: impl FnOnce() -> bool, + apply: impl FnOnce(swarm_queue_client::agents_md::Apply) -> F, +) -> Option> +where + F: Future, +{ + use swarm_queue_client::AgentControl; + + let encoded = match serde_json::from_slice::(payload) { + Ok(AgentControl::Cancel) => { + serde_json::to_vec(&swarm_queue_client::AgentControlReply { acted: cancel() }) + } + Ok(AgentControl::ApplyAgentsMd(edit)) => serde_json::to_vec(&apply(edit).await), Err(e) => { tracing::warn!(error = %e, "swarm control: undecodable command dropped"); - None + return None; } - } + }; + Some(encoded.expect("a reply encodes")) } async fn run(tx: UnboundedSender, agent: String) { @@ -156,6 +168,7 @@ fn delivered(payload: &[u8]) -> Option { #[cfg(test)] mod tests { use super::{act, forward}; + use swarm_queue_client::agents_md::{Apply, ApplyReply}; use swarm_queue_client::{AgentControl, AgentControlReply, AgentInput}; fn payload(from: &str, body: &str) -> Vec { @@ -204,20 +217,58 @@ mod tests { forward(payloads, &tx).await; } - #[test] - fn a_cancel_is_answered_with_whether_it_acted() { + #[tokio::test] + async fn a_cancel_is_answered_with_whether_it_acted() { for acted in [true, false] { - let got = act(br#""cancel""#, |c| { - assert_eq!(c, AgentControl::Cancel); - acted - }); - assert_eq!(got, Some(AgentControlReply { acted })); + let got = act( + br#""cancel""#, + || acted, + |_| async { panic!("applied an edit for a cancel") }, + ) + .await + .expect("answered"); + let got: AgentControlReply = serde_json::from_slice(&got).expect("a cancel reply"); + assert_eq!(got, AgentControlReply { acted }); } } - #[test] - fn a_payload_that_is_not_a_command_is_not_answered_or_acted_on() { - let got = act(b"pause", |_| panic!("ran a command that was not one")); + #[tokio::test] + async fn an_edit_is_answered_with_the_apply_s_reply() { + let edit = Apply { + base_revision: "abc".to_owned(), + diff: "a diff".to_owned(), + }; + let payload = + serde_json::to_vec(&AgentControl::ApplyAgentsMd(edit.clone())).expect("encodes"); + let got = act( + &payload, + || panic!("cancelled for an edit"), + |got| async move { + assert_eq!(got, edit); + ApplyReply::Applied { + revision: "def".to_owned(), + } + }, + ) + .await + .expect("answered"); + let got: ApplyReply = serde_json::from_slice(&got).expect("an apply reply"); + assert_eq!( + got, + ApplyReply::Applied { + revision: "def".to_owned() + } + ); + } + + #[tokio::test] + async fn a_payload_that_is_not_a_command_is_not_answered_or_acted_on() { + let got = act( + b"pause", + || panic!("ran a command that was not one"), + |_| async { panic!("ran a command that was not one") }, + ) + .await; assert_eq!(got, None); } } diff --git a/nix/host-modules/swarm-nats.nix b/nix/host-modules/swarm-nats.nix index 0b98752b..14ed391e 100644 --- a/nix/host-modules/swarm-nats.nix +++ b/nix/host-modules/swarm-nats.nix @@ -884,6 +884,10 @@ in # subject directly instead of opening the bucket, so it needs # no `$JS.API.*` subject; swarm-controller creates the bucket. "--agent-token-publish-subject ${lib.escapeShellArg "\$\$KV.agent-icons.{agent}"}" + # Its own key in the `agents-md` KV bucket + # (`swarm_queue_client::agents_md`), on the same terms: the + # harness mirrors its `AGENTS.md` there for the swarm to read. + "--agent-token-publish-subject ${lib.escapeShellArg "\$\$KV.agents-md.{agent}"}" # Its subagents' terminal rows # (`swarm_queue_client::subagent_term`), publish only and no # `$JS.API.*` subject: swarm-controller creates the diff --git a/nix/module-eval/nats-authelia.nix b/nix/module-eval/nats-authelia.nix index fe6dfe10..d725cecb 100644 --- a/nix/module-eval/nats-authelia.nix +++ b/nix/module-eval/nats-authelia.nix @@ -500,6 +500,7 @@ let "'$$SWARM.term.{agent}'" "'$$SWARM.agent-state.{agent}'" "'$$KV.agent-icons.{agent}'" + "'$$KV.agents-md.{agent}'" "'$$SWARM.term.{agent}.sub.>'" "'$$JS.API.DIRECT.GET.KV_agent-schedules.$$KV.agent-schedules.{agent}'" "'$$JS.API.DIRECT.GET.KV_agent-inference.$$KV.agent-inference.{agent}'" diff --git a/swarm-controller-api/src/agents_md.rs b/swarm-controller-api/src/agents_md.rs new file mode 100644 index 00000000..d901afba --- /dev/null +++ b/swarm-controller-api/src/agents_md.rs @@ -0,0 +1,51 @@ +//! `GET` and `PUT /api/agents/{name}/agents-md`: an agent's `AGENTS.md`, +//! read from the copy its harness publishes and edited by a diff the harness +//! applies. + +use serde::{Deserialize, Serialize}; + +/// Success body of `GET /api/agents/{name}/agents-md`. +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[cfg_attr(feature = "schema", derive(utoipa::ToSchema))] +pub struct AgentsMdView { + /// The file's text as the agent last published it; `""` for an agent + /// with no `AGENTS.md`. + pub content: String, + /// The git blob id of that text, what `git hash-object` prints for it. + pub revision: String, +} + +/// Body of `PUT /api/agents/{name}/agents-md`. +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[cfg_attr(feature = "schema", derive(utoipa::ToSchema))] +pub struct AgentsMdEdit { + /// The [`AgentsMdView::revision`] the diff was made against. + pub base_revision: String, + /// A unified diff as `git diff` writes it, naming `a/AGENTS.md` and + /// `b/AGENTS.md`. + pub diff: String, +} + +/// Success body of `PUT /api/agents/{name}/agents-md`. +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[cfg_attr(feature = "schema", derive(utoipa::ToSchema))] +pub struct AgentsMdSaved { + /// The file's revision after the edit. + pub revision: String, +} + +#[cfg(test)] +mod tests { + use super::AgentsMdEdit; + + #[test] + fn an_edit_round_trips() { + let edit = AgentsMdEdit { + base_revision: "e69de29bb2d1d6434b8b29ae775ad8c2e48c5391".to_owned(), + diff: "--- a/AGENTS.md\n".to_owned(), + }; + let wire = r#"{"base_revision":"e69de29bb2d1d6434b8b29ae775ad8c2e48c5391","diff":"--- a/AGENTS.md\n"}"#; + assert_eq!(serde_json::to_string(&edit).unwrap(), wire); + assert_eq!(serde_json::from_str::(wire).unwrap(), edit); + } +} diff --git a/swarm-controller-api/src/lib.rs b/swarm-controller-api/src/lib.rs index e80a2c67..fc9c47d1 100644 --- a/swarm-controller-api/src/lib.rs +++ b/swarm-controller-api/src/lib.rs @@ -5,5 +5,6 @@ //! `schema` feature each type also derives `utoipa::ToSchema`. pub mod agents; +pub mod agents_md; pub mod forge; pub mod schedules; diff --git a/swarm-controller/src/agent_control.rs b/swarm-controller/src/agent_control.rs index 5a3af000..d1c8c1bf 100644 --- a/swarm-controller/src/agent_control.rs +++ b/swarm-controller/src/agent_control.rs @@ -27,8 +27,8 @@ use crate::{AppState, error_problem}; /// What became of one command. #[derive(Debug, PartialEq, Eq)] pub(crate) enum Outcome { - /// The agent answered. - Answered(AgentControlReply), + /// The agent answered with this payload, a reply to the command sent. + Answered(Vec), /// Nothing subscribes to the agent's control subject. NotRunning, /// A subscriber held the command and did not answer in time. @@ -50,10 +50,7 @@ impl ControlSink for async_nats::Client { .payload(payload.into()) .timeout(Some(swarm_queue_client::AGENT_CONTROL_REPLY_DEADLINE)); match self.send_request(subject, request).await { - Ok(reply) => match serde_json::from_slice(&reply.payload) { - Ok(reply) => Outcome::Answered(reply), - Err(e) => Outcome::Failed(format!("the agent's answer was unreadable: {e}")), - }, + Ok(reply) => Outcome::Answered(reply.payload.to_vec()), Err(e) => match e.kind() { RequestErrorKind::TimedOut => Outcome::Unconfirmed, RequestErrorKind::NoResponders => Outcome::NotRunning, @@ -117,7 +114,7 @@ pub(crate) async fn post_agent_cancel( /// The agent and who is asking, or the refusal. Everything decided before the /// queue is touched. -fn target( +pub(crate) fn target( headers: &HeaderMap, agent: &str, ) -> Result<(String, String), (StatusCode, &'static str)> { @@ -143,6 +140,12 @@ async fn send( })?; match sink.request(subject.clone(), payload).await { Outcome::Answered(reply) => { + let reply: AgentControlReply = serde_json::from_slice(&reply).map_err(|e| { + error_problem( + StatusCode::INTERNAL_SERVER_ERROR, + &format!("{agent}'s answer was unreadable: {e}"), + ) + })?; tracing::info!(%agent, %from, ?command, acted = reply.acted, "agent command answered"); Ok(Json(CancelResponse { acted: reply.acted })) } @@ -231,10 +234,14 @@ mod tests { async fn the_answer_is_relayed_from_the_agent_s_own_control_subject() { let answers: [(bool, fn() -> Outcome); 2] = [ (true, || { - Outcome::Answered(AgentControlReply { acted: true }) + Outcome::Answered( + serde_json::to_vec(&AgentControlReply { acted: true }).expect("encodes"), + ) }), (false, || { - Outcome::Answered(AgentControlReply { acted: false }) + Outcome::Answered( + serde_json::to_vec(&AgentControlReply { acted: false }).expect("encodes"), + ) }), ]; for (acted, answer) in answers { diff --git a/swarm-controller/src/agent_limits.rs b/swarm-controller/src/agent_limits.rs index d801db4c..ac1b56f8 100644 --- a/swarm-controller/src/agent_limits.rs +++ b/swarm-controller/src/agent_limits.rs @@ -493,6 +493,7 @@ pub(crate) mod tests { wanted: Some(Arc::new(WantedWriter::new(client.clone()))), agent_status: None, agent_icons: None, + agents_md: None, reported_inference: None, agent_login: None, jobq: Arc::new(std::sync::Mutex::new(hive_jobq::scheduler::Scheduler::new( diff --git a/swarm-controller/src/agents_md.rs b/swarm-controller/src/agents_md.rs new file mode 100644 index 00000000..f082bb49 --- /dev/null +++ b/swarm-controller/src/agents_md.rs @@ -0,0 +1,436 @@ +//! `GET` and `PUT /api/agents/{name}/agents-md`: an agent's `AGENTS.md`. +//! +//! **The file in the agent's state dir is the truth, and only its harness +//! writes it.** A read serves the copy the harness publishes into the +//! `agents-md` bucket at start and on every change +//! ([`swarm_queue_client::agents_md`]), so it answers for a stopped agent +//! too. An edit is a git diff sent to the harness as a command on its +//! control subject ([`crate::agent_control`]); the harness applies it to the +//! file and says whether it did, so an edit needs the agent running. +//! +//! No hive is on either path. + +use axum::Json; +use axum::extract::{Path, State}; +use axum::http::{HeaderMap, StatusCode}; +use axum::response::{IntoResponse as _, Response}; +use problem_details::ProblemDetails; +use swarm_controller_api::agents_md::{AgentsMdEdit, AgentsMdSaved, AgentsMdView}; +use swarm_queue_client::AgentControl; +use swarm_queue_client::agents_md::{Apply, ApplyReply, Mirror}; + +use crate::agent_control::{ControlSink, Outcome}; +use crate::{AppState, error_problem, wanted_error_status}; + +/// Reads the `agents-md` bucket, the way [`crate::agent_icon::AgentIconReader`] +/// reads the icon bucket, and creates it: the agents writing it cannot. +pub struct AgentsMdReader { + client: async_nats::Client, + store: tokio::sync::OnceCell, +} + +impl AgentsMdReader { + #[must_use] + pub fn new(client: async_nats::Client) -> Self { + Self { + client, + store: tokio::sync::OnceCell::new(), + } + } + + async fn store( + &self, + ) -> std::result::Result<&async_nats::jetstream::kv::Store, swarm_queue_client::Error> { + self.store + .get_or_try_init(|| swarm_queue_client::agents_md::open_or_create(&self.client)) + .await + } + + /// Open the bucket, creating it, as soon as the queue is connected. Until + /// then every agent's write is refused and retried. + pub async fn create_when_connected(&self) { + const POLL: std::time::Duration = std::time::Duration::from_secs(5); + loop { + if swarm_queue_client::ensure_connected(&self.client).is_ok() { + match self.store().await { + Ok(_) => return, + Err(e) => tracing::warn!( + error = swarm_queue_client::chain(&e), + "creating the agents-md bucket failed; retrying" + ), + } + } + tokio::time::sleep(POLL).await; + } + } + + /// `agent`'s published `AGENTS.md`, or `None` when it has published none. + async fn get(&self, agent: &str) -> anyhow::Result> { + use anyhow::Context as _; + + // An unconnected client hangs on a JetStream request rather than + // failing it. + swarm_queue_client::ensure_connected(&self.client)?; + let store = self.store().await?; + let Some(bytes) = store + .get(agent) + .await + .with_context(|| format!("reading the agents-md entry for {agent}"))? + else { + return Ok(None); + }; + let mirror = serde_json::from_slice(&bytes) + .with_context(|| format!("the agents-md entry for {agent} is not a mirror"))?; + Ok(Some(mirror)) + } +} + +/// Read an agent's `AGENTS.md` as its harness last published it. +#[utoipa::path( + get, + path = "/api/agents/{name}/agents-md", + params(("name" = String, Path, description = "agent name")), + responses( + (status = 200, description = "the file's text and revision, as the agent last \ + published them; answered while the agent is stopped too", body = AgentsMdView), + (status = 400, description = "not an identifier (problem+json)", body = String), + (status = 404, description = "this agent has published no AGENTS.md: its harness \ + has not run since it started mirroring the file (problem+json)", body = String), + (status = 500, description = "the bucket could not be read (problem+json)", body = String), + (status = 503, description = "no swarm queue is wired up, or it is not connected \ + (problem+json)", body = String), + ), + tag = "agents" +)] +pub(crate) async fn get_agents_md( + State(state): State, + Path(name): Path, +) -> Result, ProblemDetails> { + let Some(reader) = state.agents_md.as_ref() else { + return Err(error_problem( + StatusCode::SERVICE_UNAVAILABLE, + "no swarm queue is wired up on this host", + )); + }; + let agent = hive_types::Ident::parse(&name) + .map_err(|reason| error_problem(StatusCode::BAD_REQUEST, reason))? + .into_string(); + let mirror = reader.get(&agent).await.map_err(|e| { + tracing::warn!(%agent, error = %format!("{e:#}"), "reading the agents-md entry failed"); + error_problem(wanted_error_status(&e), &format!("{e:#}")) + })?; + let Some(Mirror { content, revision }) = mirror else { + return Err(error_problem( + StatusCode::NOT_FOUND, + "this agent has published no AGENTS.md", + )); + }; + Ok(Json(AgentsMdView { content, revision })) +} + +/// The revision a 409 names, as a problem+json extension member. +#[derive(serde::Serialize)] +struct CurrentRevision { + revision: String, +} + +/// Edit an agent's `AGENTS.md` with a git diff, which the agent's harness +/// applies to the file. +#[utoipa::path( + put, + path = "/api/agents/{name}/agents-md", + params(("name" = String, Path, description = "agent whose AGENTS.md to edit")), + request_body = AgentsMdEdit, + responses( + (status = 200, description = "the diff is applied; the body is the file's new \ + revision", body = AgentsMdSaved), + (status = 400, description = "not an identifier (problem+json)", body = String), + (status = 401, description = "the request carries no `Remote-User` (problem+json)", + body = String), + (status = 409, description = "the file is no longer at `base_revision`, so nothing \ + was applied; `revision` in the problem+json body is the one it is at", body = String), + (status = 422, description = "`git apply` refused the diff, or it touches something \ + other than AGENTS.md; `detail` is git's message and the file is unchanged \ + (problem+json)", body = String), + (status = 503, description = "the agent is not running, or its harness is not \ + connected to the swarm queue, or no swarm queue is configured on this host or it \ + could not be reached (problem+json)", body = String), + (status = 504, description = "the harness held the edit and did not answer in time, \ + so whether it was applied is unknown (problem+json)", body = String), + (status = 500, description = "the queue refused or failed the request, or the harness \ + could not read or write the file (problem+json)", body = String), + ), + tag = "agents" +)] +pub(crate) async fn put_agents_md( + State(state): State, + Path(agent): Path, + headers: HeaderMap, + Json(edit): Json, +) -> Result, Response> { + let problem = |s: StatusCode, d: &str| error_problem(s, d).into_response(); + let (agent, from) = + crate::agent_control::target(&headers, &agent).map_err(|(s, d)| problem(s, d))?; + let Some(status) = state.status.as_ref() else { + return Err(problem( + StatusCode::SERVICE_UNAVAILABLE, + "no swarm queue is configured on this host", + )); + }; + let client = status.queue_client(); + swarm_queue_client::ensure_connected(&client).map_err(|e| { + problem( + StatusCode::SERVICE_UNAVAILABLE, + &swarm_queue_client::chain(&e), + ) + })?; + send(&client, &agent, &from, edit).await +} + +async fn send( + sink: &impl ControlSink, + agent: &str, + from: &str, + edit: AgentsMdEdit, +) -> Result, Response> { + let problem = |s: StatusCode, d: &str| error_problem(s, d).into_response(); + let subject = swarm_queue_client::agent_control_subject(agent); + let command = AgentControl::ApplyAgentsMd(Apply { + base_revision: edit.base_revision, + diff: edit.diff, + }); + let payload = serde_json::to_vec(&command).map_err(|e| { + problem( + StatusCode::INTERNAL_SERVER_ERROR, + &format!("encoding the edit failed: {e}"), + ) + })?; + let reply = match sink.request(subject.clone(), payload).await { + Outcome::Answered(reply) => serde_json::from_slice::(&reply).map_err(|e| { + problem( + StatusCode::INTERNAL_SERVER_ERROR, + &format!("{agent}'s answer was unreadable: {e}"), + ) + })?, + Outcome::NotRunning => { + return Err(problem( + StatusCode::SERVICE_UNAVAILABLE, + &format!( + "{agent} is not running: its harness is not listening for commands, so \ + AGENTS.md cannot be edited until it is" + ), + )); + } + Outcome::Unconfirmed => { + tracing::warn!(%agent, %from, "agents-md edit unanswered"); + return Err(problem( + StatusCode::GATEWAY_TIMEOUT, + &format!("{agent} did not answer in time: whether the edit was applied is unknown"), + )); + } + Outcome::Failed(e) => { + tracing::warn!(%agent, %subject, error = %e, "agents-md edit failed"); + return Err(problem( + StatusCode::INTERNAL_SERVER_ERROR, + &format!("sending to {subject} failed: {e}"), + )); + } + }; + match reply { + ApplyReply::Applied { revision } => { + tracing::info!(%agent, %from, %revision, "agents-md edit applied"); + Ok(Json(AgentsMdSaved { revision })) + } + ApplyReply::Stale { revision } => { + Err(ProblemDetails::from_status_code(StatusCode::CONFLICT) + .with_detail(format!( + "{agent}'s AGENTS.md changed since the edit was made: it is at {revision}. \ + Reload it and make the edit again" + )) + .with_extensions(CurrentRevision { revision }) + .into_response()) + } + ApplyReply::Rejected { error } => Err(problem( + StatusCode::UNPROCESSABLE_ENTITY, + &format!("the diff does not apply to {agent}'s AGENTS.md: {error}"), + )), + ApplyReply::Failed { error } => { + tracing::warn!(%agent, %error, "agents-md edit failed in the harness"); + Err(problem( + StatusCode::INTERNAL_SERVER_ERROR, + &format!("{agent} could not apply the edit: {error}"), + )) + } + } +} + +#[cfg(test)] +mod tests { + use std::sync::Mutex; + + use axum::http::StatusCode; + use swarm_controller_api::agents_md::AgentsMdEdit; + use swarm_queue_client::AgentControl; + use swarm_queue_client::agents_md::{Apply, ApplyReply}; + + use super::send; + use crate::agent_control::{ControlSink, Outcome}; + + /// Answers every command with one fixed outcome and records what it got. + struct FakeSink { + answer: Box Outcome + Send + Sync>, + got: Mutex)>>, + } + + impl FakeSink { + fn replying(reply: &ApplyReply) -> Self { + let bytes = serde_json::to_vec(reply).expect("encodes"); + Self::new(move || Outcome::Answered(bytes.clone())) + } + + fn new(answer: impl Fn() -> Outcome + Send + Sync + 'static) -> Self { + Self { + answer: Box::new(answer), + got: Mutex::default(), + } + } + } + + impl ControlSink for FakeSink { + async fn request(&self, subject: String, payload: Vec) -> Outcome { + self.got + .lock() + .expect("unpoisoned") + .push((subject, payload)); + (self.answer)() + } + } + + fn edit() -> AgentsMdEdit { + AgentsMdEdit { + base_revision: "abc".to_owned(), + diff: "--- a/AGENTS.md\n".to_owned(), + } + } + + async fn body(resp: axum::response::Response) -> serde_json::Value { + let bytes = axum::body::to_bytes(resp.into_body(), usize::MAX) + .await + .expect("body"); + serde_json::from_slice(&bytes).expect("json") + } + + #[tokio::test] + async fn an_edit_goes_to_the_agent_s_control_subject_and_its_revision_comes_back() { + let sink = FakeSink::replying(&ApplyReply::Applied { + revision: "def".to_owned(), + }); + let saved = send(&sink, "atlas", "mara", edit()).await.expect("applied"); + assert_eq!(saved.0.revision, "def"); + let got = sink.got.lock().expect("unpoisoned"); + assert_eq!(got[0].0, "$SWARM.control.atlas"); + let sent: AgentControl = serde_json::from_slice(&got[0].1).expect("a command"); + assert_eq!( + sent, + AgentControl::ApplyAgentsMd(Apply { + base_revision: "abc".to_owned(), + diff: "--- a/AGENTS.md\n".to_owned(), + }) + ); + } + + #[tokio::test] + async fn a_stale_base_is_409_naming_the_current_revision() { + let sink = FakeSink::replying(&ApplyReply::Stale { + revision: "fff".to_owned(), + }); + let resp = send(&sink, "atlas", "mara", edit()) + .await + .expect_err("stale"); + assert_eq!(resp.status(), StatusCode::CONFLICT); + assert_eq!(body(resp).await["revision"], "fff"); + } + + #[tokio::test] + async fn a_diff_git_refuses_is_422_with_git_s_message() { + let sink = FakeSink::replying(&ApplyReply::Rejected { + error: "error: patch failed: AGENTS.md:1".to_owned(), + }); + let resp = send(&sink, "atlas", "mara", edit()) + .await + .expect_err("rejected"); + assert_eq!(resp.status(), StatusCode::UNPROCESSABLE_ENTITY); + let detail = body(resp).await["detail"] + .as_str() + .expect("detail") + .to_owned(); + assert!(detail.contains("patch failed: AGENTS.md:1"), "{detail}"); + } + + #[tokio::test] + async fn no_subscriber_is_503_and_the_rest_are_504_and_500() { + let status = |sink: FakeSink| async move { + send(&sink, "atlas", "mara", edit()) + .await + .expect_err("not a success") + .status() + }; + assert_eq!( + status(FakeSink::new(|| Outcome::NotRunning)).await, + StatusCode::SERVICE_UNAVAILABLE + ); + assert_eq!( + status(FakeSink::new(|| Outcome::Unconfirmed)).await, + StatusCode::GATEWAY_TIMEOUT + ); + assert_eq!( + status(FakeSink::new(|| Outcome::Failed("refused".to_owned()))).await, + StatusCode::INTERNAL_SERVER_ERROR + ); + assert_eq!( + status(FakeSink::replying(&ApplyReply::Failed { + error: "disk full".to_owned() + })) + .await, + StatusCode::INTERNAL_SERVER_ERROR + ); + // A cancel's answer is not an edit's. + assert_eq!( + status(FakeSink::new(|| Outcome::Answered( + br#"{"acted":true}"#.to_vec() + ))) + .await, + StatusCode::INTERNAL_SERVER_ERROR + ); + } + + /// Through a real `nats-server`: nothing is served before the harness + /// publishes, and its publish, straight to the key's subject as the + /// harness sends it, is what a read returns. + #[tokio::test] + async fn a_read_serves_what_the_harness_published() { + use swarm_queue_client::agents_md::{Mirror, subject}; + + let nats = crate::test_nats::TestNats::start(); + let client = nats.client().await; + let reader = super::AgentsMdReader::new(client.clone()); + reader.create_when_connected().await; + assert_eq!(reader.get("atlas").await.expect("read"), None); + + let mirror = Mirror { + content: "# atlas\n".to_owned(), + revision: "abc".to_owned(), + }; + let js = async_nats::jetstream::new(client.clone()); + js.publish( + subject("atlas"), + serde_json::to_vec(&mirror).expect("encodes").into(), + ) + .await + .expect("sent") + .await + .expect("acked"); + assert_eq!(reader.get("atlas").await.expect("read"), Some(mirror)); + assert_eq!(reader.get("iris").await.expect("read"), None); + } +} diff --git a/swarm-controller/src/forge_account.rs b/swarm-controller/src/forge_account.rs index 0856130a..836a07f0 100644 --- a/swarm-controller/src/forge_account.rs +++ b/swarm-controller/src/forge_account.rs @@ -153,6 +153,7 @@ mod tests { wanted: None, agent_status: None, agent_icons: None, + agents_md: None, reported_inference: None, agent_login: None, jobq: std::sync::Arc::new(std::sync::Mutex::new(hive_jobq::scheduler::Scheduler::new( diff --git a/swarm-controller/src/github_account.rs b/swarm-controller/src/github_account.rs index fb0d124f..f560b105 100644 --- a/swarm-controller/src/github_account.rs +++ b/swarm-controller/src/github_account.rs @@ -123,6 +123,7 @@ mod tests { wanted: None, agent_status: None, agent_icons: None, + agents_md: None, reported_inference: None, agent_login: None, jobq: std::sync::Arc::new(std::sync::Mutex::new(hive_jobq::scheduler::Scheduler::new( diff --git a/swarm-controller/src/linked_accounts.rs b/swarm-controller/src/linked_accounts.rs index 1ad46944..f8787cd1 100644 --- a/swarm-controller/src/linked_accounts.rs +++ b/swarm-controller/src/linked_accounts.rs @@ -962,6 +962,7 @@ pub(crate) mod tests { wanted: None, agent_status: None, agent_icons: None, + agents_md: None, reported_inference: None, agent_login: None, jobq: std::sync::Arc::new(std::sync::Mutex::new(hive_jobq::scheduler::Scheduler::new( diff --git a/swarm-controller/src/main.rs b/swarm-controller/src/main.rs index 60fd4d6f..ce88efd3 100644 --- a/swarm-controller/src/main.rs +++ b/swarm-controller/src/main.rs @@ -55,6 +55,7 @@ mod agent_permissions; mod agent_renewal; mod agent_state_stream; mod agent_status; +mod agents_md; mod alerts; mod archetypes; mod auth; @@ -954,6 +955,9 @@ struct AppState { /// connection, several consumers" rationale as `agent_status` above, /// and `None` in exactly the state `status` is. agent_icons: Option>, + /// Each agent's published `AGENTS.md`, on `status`'s connection like + /// `agent_icons`, and `None` in exactly the state `status` is. + agents_md: Option>, /// What each agent last reported running on, from its turn-state /// headers on `status`'s connection. `None` in exactly the state /// `status` is. @@ -1325,6 +1329,17 @@ fn agent_icon_reader( Some(reader) } +/// The `AGENTS.md` reader, sharing the status reader's connection, and +/// creating the bucket, as [`agent_icon_reader`] does for icons. +fn agents_md_reader( + status: Option<&Arc>, +) -> Option> { + let reader = Arc::new(agents_md::AgentsMdReader::new(status?.queue_client())); + let creating = Arc::clone(&reader); + tokio::spawn(async move { creating.create_when_connected().await }); + Some(reader) +} + /// What each agent last reported running on, read on the status reader's /// connection — same rationale as [`wanted_writer`]. fn reported_inference( @@ -3612,6 +3627,7 @@ async fn main() -> Result<()> { wanted: wanted_writer(status.as_ref()), agent_status: agent_status_reader(status.as_ref()), agent_icons: agent_icon_reader(status.as_ref()), + agents_md: agents_md_reader(status.as_ref()), reported_inference: reported_inference(status.as_ref()), agent_login: agent_login(status.as_ref()), runs: review_classifier::runs::Runs::Live(status.as_ref().map(|s| s.queue_client())), @@ -3630,8 +3646,7 @@ async fn main() -> Result<()> { alerts: alerts::from_env(), }; - let app = build_app(state); - axum::serve(listener, app) + axum::serve(listener, build_app(state)) .await .context("serving swarm-controller") } @@ -3654,6 +3669,7 @@ fn build_app(state: AppState) -> axum::Router { .routes(routes!(get_jobq_rollup)) .routes(routes!(get_agent_config_pr)) .routes(routes!(get_agent_icon)) + .routes(routes!(agents_md::get_agents_md, agents_md::put_agents_md)) .routes(routes!( inference::get_agent_inference, inference::put_agent_inference, @@ -3833,6 +3849,7 @@ mod tests { wanted: None, agent_status: None, agent_icons: None, + agents_md: None, reported_inference: None, agent_login: None, jobq: std::sync::Arc::clone(&sched), diff --git a/swarm-controller/src/matrix_account.rs b/swarm-controller/src/matrix_account.rs index a79082a1..0905682f 100644 --- a/swarm-controller/src/matrix_account.rs +++ b/swarm-controller/src/matrix_account.rs @@ -457,6 +457,7 @@ mod tests { wanted: None, agent_status: None, agent_icons: None, + agents_md: None, reported_inference: None, agent_login: None, jobq: std::sync::Arc::new(std::sync::Mutex::new(hive_jobq::scheduler::Scheduler::new( diff --git a/swarm-queue-client/src/agents_md.rs b/swarm-queue-client/src/agents_md.rs new file mode 100644 index 00000000..6e267c7e --- /dev/null +++ b/swarm-queue-client/src/agents_md.rs @@ -0,0 +1,103 @@ +//! An agent's `AGENTS.md`, as the swarm sees it: a read mirror in the +//! `agents-md` KV bucket, and the edit command that changes the file. +//! +//! **The file in the agent's state dir is the truth.** The agent's harness is +//! the only writer of both the file and its key here: it publishes the file's +//! text and [`crate::agents_md::Mirror::revision`] when it starts and whenever +//! the file changes, so the swarm can show the text while the agent is +//! stopped. An edit from the swarm is a [`crate::agents_md::Apply`] sent on the +//! agent's control subject, which the harness applies to the file itself and +//! answers with a [`crate::agents_md::ApplyReply`]. +//! +//! Keyed by the agent alone, for the reason [`crate::agent_icon`] gives. The +//! auth callout grants each agent's credential `$KV.agents-md.` and no +//! other key; the writer publishes to [`crate::agents_md::subject`] without +//! opening the bucket, and swarm-controller creates it. + +#[cfg(feature = "kv")] +use crate::Error; + +/// The KV bucket the mirrors are published into, keyed by agent name. +pub const BUCKET: &str = "agents-md"; + +/// The subject `agent`'s mirror is written on: `$KV..`. +#[must_use] +pub fn subject(agent: &str) -> String { + format!("$KV.{BUCKET}.{agent}") +} + +/// The value under an agent's key: its `AGENTS.md` as last published. +#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)] +pub struct Mirror { + /// The file's text. An agent with no `AGENTS.md` publishes `""`. + pub content: String, + /// The git blob id of the file's bytes, what `git hash-object` prints for + /// it. An [`Apply`] names the revision its diff was made against. + pub revision: String, +} + +/// An edit to the agent's `AGENTS.md`: a unified diff, as `git diff` writes +/// it (`a/AGENTS.md` and `b/AGENTS.md`), against the text at +/// [`Self::base_revision`]. +#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)] +pub struct Apply { + /// The [`Mirror::revision`] the diff was made against. The harness + /// refuses the edit when the file is no longer at it. + pub base_revision: String, + /// The diff. + pub diff: String, +} + +/// What the harness answers an [`Apply`] with. +#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)] +#[serde(rename_all = "snake_case", tag = "outcome")] +pub enum ApplyReply { + /// The diff is applied; the file is now at `revision`. + Applied { revision: String }, + /// The file is not at the base revision, so nothing was applied. It is at + /// `revision`. + Stale { revision: String }, + /// `git apply` refused the diff, or the diff touches something other than + /// `AGENTS.md`. The file is unchanged; `error` is git's own message or + /// says what the diff touched. + Rejected { error: String }, + /// The harness could not read or write the file, or run git. + Failed { error: String }, +} + +/// Open the bucket, creating it if nothing has yet. `history: 1`, as for +/// [`crate::agent_icon::open_or_create`]: the swarm wants the current text. +#[cfg(feature = "kv")] +pub async fn open_or_create( + client: &async_nats::Client, +) -> Result { + let js = async_nats::jetstream::new(client.clone()); + crate::kv::open_or_create( + &js, + BUCKET, + "agents-md", + "Each agent's AGENTS.md, as its harness last published it".to_owned(), + ) + .await +} + +#[cfg(test)] +mod tests { + use super::{ApplyReply, subject}; + + /// One token after the bucket, the agent name alone: what the per-agent + /// grant `$KV.agents-md.{agent}` expands to. + #[test] + fn the_published_subject_carries_the_agent_alone() { + assert_eq!(subject("iris"), "$KV.agents-md.iris"); + } + + #[test] + fn a_reply_names_its_outcome() { + let wire = serde_json::to_string(&ApplyReply::Stale { + revision: "abc".to_owned(), + }) + .expect("encodes"); + assert_eq!(wire, r#"{"outcome":"stale","revision":"abc"}"#); + } +} diff --git a/swarm-queue-client/src/lib.rs b/swarm-queue-client/src/lib.rs index e11f1f23..ee0ef4ea 100644 --- a/swarm-queue-client/src/lib.rs +++ b/swarm-queue-client/src/lib.rs @@ -209,6 +209,11 @@ pub mod agent_token; /// placement stays out of the key, and for who may write it. pub mod agent_icon; +/// The bucket each agent mirrors its `AGENTS.md` into, and the edit command +/// for it. The names and types are unconditional; opening the bucket needs +/// `kv`. +pub mod agents_md; + /// The bucket swarm-controller declares each agent's scheduled prompts in — /// one key per agent, read by that agent alone. The names are unconditional; /// writing needs `kv` and reading needs `agent-schedules`. @@ -427,14 +432,18 @@ pub fn client_inbox_prefix(client_id: &str) -> String { /// The harness acts on a command only because the controller is the one /// client granted publish on the subject, so there is no sender field to /// trust or forge here, unlike [`AgentInput::from`]. -#[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize, serde::Deserialize)] +#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)] #[serde(rename_all = "snake_case")] pub enum AgentControl { /// Interrupt the turn in flight, as the agent's own cancel button does. + /// Answered with an [`AgentControlReply`]. Cancel, + /// Apply a diff to the agent's `AGENTS.md`. Answered with an + /// [`agents_md::ApplyReply`]. + ApplyAgentsMd(agents_md::Apply), } -/// What an agent answers an [`AgentControl`] request with. +/// What an agent answers an [`AgentControl::Cancel`] request with. /// /// The controller sends a command as a request, so a reply is the only /// confirmation the operator gets that the harness took it. No reply within diff --git a/swarmctl/src/agents_md.rs b/swarmctl/src/agents_md.rs new file mode 100644 index 00000000..dd7f3ba2 --- /dev/null +++ b/swarmctl/src/agents_md.rs @@ -0,0 +1,68 @@ +//! `swarmctl agent agents-md` — read an agent's `AGENTS.md` and edit it with +//! a diff, through swarm-controller's `/api/agents/{name}/agents-md` routes. +//! +//! The read is the copy the agent's harness last published, so it works +//! while the agent is stopped. An edit is applied by the harness to the file +//! itself, so it needs the agent running. + +use std::path::Path; + +use anyhow::{Context as _, Result}; +use swarm_controller_api::agents_md::{AgentsMdEdit, AgentsMdSaved, AgentsMdView}; + +use crate::agent::{parse_ident, send}; +use crate::remote::Target; + +fn runtime() -> Result { + tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .context("starting a tokio runtime for the controller request") +} + +/// The agent's `AGENTS.md` path. The name is checked first, so it cannot +/// reshape the URI. +fn member(agent: &str) -> Result { + let agent = parse_ident(agent, "agent name")?; + Ok(format!("/api/agents/{agent}/agents-md")) +} + +/// Run `swarmctl agent agents-md get`: the text on stdout, unchanged, and its +/// revision on stderr, so stdout redirected to a file is the file. +pub(crate) fn get(target: &Target, agent: &str) -> Result<()> { + let uri = member(agent)?; + let view: AgentsMdView = runtime()?.block_on(send( + target, + hyper::Method::GET, + &uri, + None::<&()>, + "agents-md-get", + ))?; + eprintln!("revision {}", view.revision); + print!("{}", view.content); + Ok(()) +} + +/// Run `swarmctl agent agents-md apply`: send the diff in `diff_file`, made +/// against `base`, and print the file's new revision. +pub(crate) fn apply(target: &Target, agent: &str, base: &str, diff_file: &Path) -> Result<()> { + let uri = member(agent)?; + let diff = std::fs::read_to_string(diff_file) + .with_context(|| format!("reading the diff {}", diff_file.display()))?; + let edit = AgentsMdEdit { + base_revision: base.to_owned(), + diff, + }; + let saved: AgentsMdSaved = runtime()?.block_on(send( + target, + hyper::Method::PUT, + &uri, + Some(&edit), + "agents-md-apply", + ))?; + println!( + "applied; {agent}'s AGENTS.md is at revision {}", + saved.revision + ); + Ok(()) +} diff --git a/swarmctl/src/main.rs b/swarmctl/src/main.rs index 18991c38..d2dd9967 100644 --- a/swarmctl/src/main.rs +++ b/swarmctl/src/main.rs @@ -26,6 +26,7 @@ //! for the same reason `hivectl` does not link `hive-c0re`. mod agent; +mod agents_md; mod forge; mod inference; mod login; @@ -302,6 +303,43 @@ enum AgentVerb { #[command(subcommand)] command: PermissionsVerb, }, + /// Show an agent's AGENTS.md, or edit it with a diff. + /// + /// `get` shows the copy the agent last published, so it works while the + /// agent is stopped. `apply` hands the diff to the agent, which applies + /// it to the file with `git apply`; it needs the agent running. + AgentsMd { + #[command(subcommand)] + command: AgentsMdVerb, + }, +} + +#[derive(Subcommand)] +enum AgentsMdVerb { + /// Print an agent's AGENTS.md on stdout, and its revision on stderr. + Get(InferenceAgentArgs), + /// Apply a unified diff to an agent's AGENTS.md. + /// + /// The diff names `a/AGENTS.md` and `b/AGENTS.md`, as + /// `diff -u a/AGENTS.md b/AGENTS.md` writes it. The agent refuses it when + /// the file is no longer at `--base`, or when it doesn't apply, and + /// leaves the file as it was. + Apply(AgentsMdApplyArgs), +} + +#[derive(Args)] +struct AgentsMdApplyArgs { + /// Agent name. + agent: String, + /// The file holding the diff. + diff_file: PathBuf, + /// The revision the diff was made against, as `get` printed it. It is + /// the git blob id of that text, so `git hash-object` of the unedited + /// file prints it too. + #[arg(long, value_name = "REVISION")] + base: String, + #[command(flatten)] + socket: ControllerSocketArg, } #[derive(Subcommand)] @@ -645,6 +683,9 @@ fn main() -> Result<()> { Verb::Agent { command: AgentVerb::Permissions { command }, } => permissions(&remote, command), + Verb::Agent { + command: AgentVerb::AgentsMd { command }, + } => agents_md(&remote, command), // Same socket-resolution reasoning as `Create` above. Verb::Forge { command: ForgeVerb::MakeAdmin(args), @@ -724,6 +765,15 @@ fn permissions(remote: &RemoteArgs, command: PermissionsVerb) -> Result<()> { } } +fn agents_md(remote: &RemoteArgs, command: AgentsMdVerb) -> Result<()> { + match command { + AgentsMdVerb::Get(a) => agents_md::get(&a.socket.resolve(remote)?, &a.agent), + AgentsMdVerb::Apply(a) => { + agents_md::apply(&a.socket.resolve(remote)?, &a.agent, &a.base, &a.diff_file) + } + } +} + fn schedule(remote: &RemoteArgs, command: ScheduleVerb) -> Result<()> { match command { ScheduleVerb::Add(a) => schedule::add(