diff --git a/docs/agent-lifecycle/persistence.md b/docs/agent-lifecycle/persistence.md index 5ad13f56..c5808cb1 100644 --- a/docs/agent-lifecycle/persistence.md +++ b/docs/agent-lifecycle/persistence.md @@ -278,7 +278,8 @@ Under `/var/lib/hyperhive/agents//`: logs an error naming both paths: claude loads only `CLAUDE.md` while one exists in its cwd or above. The harness also publishes the file to the swarm, which reads that copy, and applies the git diffs the - swarm sends to edit it. This file stays the only copy that counts: + swarm sends to edit it. A new agent's first start writes the text given at + create, only when no file exists. This file stays the only copy that counts: [`docs/swarm/README.md`](../swarm/README.md#an-agents-agentsmd). - `harness/` — harness-internal ephemeral state; not intended for agent consumption. Bind-mounted to `/agents//harness` diff --git a/docs/swarm/README.md b/docs/swarm/README.md index 9e0e4f2f..6c756914 100644 --- a/docs/swarm/README.md +++ b/docs/swarm/README.md @@ -328,6 +328,19 @@ publishes `""`. `GET /api/agents//agents-md` answers `{content, revision}` from that copy, so it works while the agent isn't running. A 404 means the agent hasn't published one: its harness hasn't run since it began to. +**Initial text.** `POST /api/agents` and `swarmctl agent create --agents-md ` +take an optional `agents_md`, the text the file starts with, up to 64 KiB; a +longer text is a 400. The controller stores it under the key `` of the +`agents-md-initial` bucket, which only it writes, before it queues the creation. +At start the harness reads its own key, with one direct get on +`$JS.API.DIRECT.GET.KV_agents-md-initial.$KV.agents-md-initial.`, and +writes the text to `AGENTS.md` only when no file exists there. It does this +before the first mirror publish and the first turn. An existing file is never +replaced, so a restart or a re-create keeps the agent's edits; a later change +to the file never reaches the key. Destroying the agent deletes the key. When +the harness can't read the key it starts without the text and logs a warning. +The text holds no secret, so nothing goes through the secret store. + **Editing.** `PUT /api/agents//agents-md` takes `{base_revision, diff}`. The diff is a unified diff that names `a/AGENTS.md` and `b/AGENTS.md`, made against the text at `base_revision`. swarm-controller sends it as a request on diff --git a/docs/tools/swarmctl-cli.md b/docs/tools/swarmctl-cli.md index 6ec9956c..e032ad98 100644 --- a/docs/tools/swarmctl-cli.md +++ b/docs/tools/swarmctl-cli.md @@ -101,6 +101,9 @@ No approval gate guards this: the caller is root on the controller's host, or lo * `--hive ` — Hive in this swarm to deploy the agent to. Required, and deliberately not defaulted: it's an *address* — the hive that gets the deploy message — and only the operator knows which one they mean. The controller checks it against the swarm's hive roster and names the known hives if it misses. +* `--agents-md ` — A text file the agent's `AGENTS.md` starts with. + + The agent writes it on its first start, and only when it has no `AGENTS.md` yet: re-creating an agent never replaces an existing file. The controller refuses a file over 64 KiB. * `--controller-socket ` — swarm-controller's unix socket. Supplied by the nix module that installs this binary, from the same `socketPath` option the daemon binds; falls back to `SWARM_CONTROLLER_SOCKET`. diff --git a/hive-agent/Cargo.toml b/hive-agent/Cargo.toml index 11e68f9c..37133008 100644 --- a/hive-agent/Cargo.toml +++ b/hive-agent/Cargo.toml @@ -49,6 +49,7 @@ swarm-queue-client = { workspace = true, features = [ "agent-schedules", "agent-inference", "agent-permissions", + "agents-md-initial", "operator-profiles", ] } # This agent's own queue secret, read from the store under its own certificate. diff --git a/hive-agent/src/main.rs b/hive-agent/src/main.rs index 6d210163..c0a7dfb2 100644 --- a/hive-agent/src/main.rs +++ b/hive-agent/src/main.rs @@ -505,6 +505,8 @@ 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(); + // Before the mirror and before any turn reads the file. + swarm_agents_md::seed_initial().await; // At start and on every change, so the swarm can read it while the // agent is stopped. swarm_agents_md::spawn(); diff --git a/hive-agent/src/swarm_agents_md.rs b/hive-agent/src/swarm_agents_md.rs index a968afe4..292358e8 100644 --- a/hive-agent/src/swarm_agents_md.rs +++ b/hive-agent/src/swarm_agents_md.rs @@ -29,6 +29,7 @@ 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 swarm_queue_client::agents_md_initial::Initial; use tokio::sync::Notify; use crate::claude_md_watch::agents_md_path; @@ -94,6 +95,103 @@ pub fn spawn() { }); } +/// How many times, and for how long each, the key is read at start before the +/// agent starts without the text. +const SEED_ATTEMPTS: u32 = 3; +const SEED_READ_TIMEOUT: Duration = Duration::from_secs(5); +const SEED_RETRY_WAIT: Duration = Duration::from_secs(1); + +/// Write the text the swarm stored for this agent at create to `AGENTS.md`, +/// when the file does not exist. Done before anything reads the file, so the +/// agent's first turn already has it, and before the mirror, so the swarm +/// sees it from the first publish. +/// +/// An existing file is never touched, so a restart or a re-created agent +/// keeps its edits. The read of the key is bounded to about +/// [`SEED_ATTEMPTS`] × [`SEED_READ_TIMEOUT`], so a queue that refuses it delays +/// boot by seconds; the agent then starts without the text and a later start +/// tries again. +pub async fn seed_initial() { + let agent = crate::identity::label(); + let dir = crate::paths::state_dir(); + if agent.is_empty() || exists(&agents_md_path(&dir)) { + return; + } + let client = crate::swarm_queue::client().await; + let read = read_bounded(SEED_ATTEMPTS, SEED_READ_TIMEOUT, SEED_RETRY_WAIT, || { + swarm_queue_client::agents_md_initial::read_own(&client, &agent) + }) + .await; + match read { + Ok(initial) => { + if let Err(e) = seed(&dir, initial.as_ref()) { + tracing::warn!(error = %e, "writing the initial AGENTS.md failed"); + } + } + Err(e) => { + tracing::warn!(error = %e, "reading the initial AGENTS.md failed; starting without it"); + } + } +} + +/// `read`, tried up to `attempts` times with `wait` between, each cut off +/// after `timeout`. The first answer wins; the last failure is the error. +async fn read_bounded( + attempts: u32, + timeout: Duration, + wait: Duration, + mut read: F, +) -> Result +where + F: FnMut() -> Fut, + Fut: Future>, + E: std::fmt::Display, +{ + let mut last = String::new(); + for attempt in 0..attempts { + if attempt > 0 { + tokio::time::sleep(wait).await; + } + match tokio::time::timeout(timeout, read()).await { + Ok(Ok(value)) => return Ok(value), + Ok(Err(e)) => last = e.to_string(), + Err(_) => last = format!("no answer within {timeout:?}"), + } + } + Err(last) +} + +/// Whether anything, a dangling link included, is at `path`. +fn exists(path: &Path) -> bool { + std::fs::symlink_metadata(path).is_ok() +} + +/// Create `AGENTS.md` in `state_dir` with `initial`'s text, unless a file is +/// already there. The text goes to a temporary file in the same directory +/// first and is linked into place without replacing anything +/// (`persist_noclobber`), so a failed write leaves no file and a file written +/// meanwhile is kept. +fn seed(state_dir: &Path, initial: Option<&Initial>) -> std::io::Result<()> { + use std::os::unix::fs::PermissionsExt as _; + + let Some(initial) = initial else { + return Ok(()); + }; + std::fs::create_dir_all(state_dir)?; + let mut temp = tempfile::NamedTempFile::new_in(state_dir)?; + temp.write_all(initial.content.as_bytes())?; + temp.as_file() + .set_permissions(std::fs::Permissions::from_mode(0o644))?; + match temp.persist_noclobber(agents_md_path(state_dir)) { + Ok(_) => { + tracing::info!(bytes = initial.content.len(), "wrote the initial AGENTS.md"); + Ok(()) + } + Err(e) if e.error.kind() == std::io::ErrorKind::AlreadyExists => Ok(()), + Err(e) => Err(e.error), + } +} + /// What became of one mirror write. #[derive(Debug)] enum Published { @@ -503,4 +601,95 @@ mod tests { assert_eq!(second.revision, revision(AFTER.as_bytes())); task.abort(); } + + fn initial(text: &str) -> swarm_queue_client::agents_md_initial::Initial { + swarm_queue_client::agents_md_initial::Initial { + content: text.to_owned(), + } + } + + /// The stored text becomes the file when there is none, and the mirror's + /// revision then covers it. + #[test] + fn the_initial_text_is_written_when_there_is_no_file() { + let dir = tempfile::tempdir().expect("tempdir"); + super::seed(dir.path(), Some(&initial(BEFORE))).expect("seed"); + assert_eq!(file(&dir), BEFORE); + // Presence control for the next two: nothing stored writes nothing. + let empty = tempfile::tempdir().expect("tempdir"); + super::seed(empty.path(), None).expect("seed"); + assert!(!empty.path().join("AGENTS.md").exists()); + } + + /// A file the agent already has, edited or not, is never replaced. + #[test] + fn an_existing_file_is_not_overwritten_by_the_initial_text() { + let dir = state(AFTER); + super::seed(dir.path(), Some(&initial(BEFORE))).expect("seed"); + assert_eq!(file(&dir), AFTER); + let empty = state(""); + super::seed(empty.path(), Some(&initial(BEFORE))).expect("seed"); + assert_eq!(file(&empty), "", "an emptied file is still the agent's"); + } + + /// A second start, with the key still stored, keeps the agent's edits. + #[test] + fn a_second_seed_keeps_what_the_agent_did_with_the_file() { + let dir = tempfile::tempdir().expect("tempdir"); + super::seed(dir.path(), Some(&initial(BEFORE))).expect("seed"); + std::fs::write(dir.path().join("AGENTS.md"), AFTER).expect("edit"); + super::seed(dir.path(), Some(&initial(BEFORE))).expect("seed again"); + assert_eq!(file(&dir), AFTER); + } + + /// A seed that cannot finish leaves no `AGENTS.md`, so the next start + /// tries again. The state dir's parent is a file, so no temp file can be + /// made; the leftover directory listing is empty either way. + #[test] + fn a_seed_that_fails_leaves_no_file_and_no_temp_file() { + let dir = tempfile::tempdir().expect("tempdir"); + let blocker = dir.path().join("blocker"); + std::fs::write(&blocker, "x").expect("write"); + super::seed(&blocker.join("state"), Some(&initial(BEFORE))).expect_err("no dir"); + assert!(!blocker.join("state").exists()); + // Control: a good dir leaves exactly AGENTS.md, no temp file. + let ok = tempfile::tempdir().expect("tempdir"); + super::seed(ok.path(), Some(&initial(BEFORE))).expect("seed"); + let names: Vec<_> = std::fs::read_dir(ok.path()) + .expect("ls") + .map(|e| e.expect("entry").file_name()) + .collect(); + assert_eq!(names, ["AGENTS.md"]); + } + + /// A read that never answers costs `attempts` × `timeout` plus the waits, + /// not a startup retry schedule, and the last failure is reported. + #[tokio::test(start_paused = true)] + async fn a_read_that_never_answers_is_given_up_after_the_bound() { + let started = tokio::time::Instant::now(); + let out = super::read_bounded(3, Duration::from_secs(5), Duration::from_secs(1), || { + std::future::pending::>() + }) + .await; + assert!(out.unwrap_err().contains("no answer within")); + assert_eq!(started.elapsed(), Duration::from_secs(17)); + } + + /// A later attempt that answers wins over an earlier failure. + #[tokio::test(start_paused = true)] + async fn a_read_that_fails_once_and_then_answers_returns_the_answer() { + let calls = std::sync::atomic::AtomicU32::new(0); + let out = super::read_bounded(3, Duration::from_secs(5), Duration::from_secs(1), || { + let n = calls.fetch_add(1, std::sync::atomic::Ordering::SeqCst); + async move { + if n == 0 { + Err("down".to_owned()) + } else { + Ok(7) + } + } + }) + .await; + assert_eq!(out, Ok(7)); + } } diff --git a/nix/host-modules/swarm-nats.nix b/nix/host-modules/swarm-nats.nix index 14ed391e..69fab3de 100644 --- a/nix/host-modules/swarm-nats.nix +++ b/nix/host-modules/swarm-nats.nix @@ -909,6 +909,10 @@ in # the harness and the subagent daemon read the tool permissions # it holds. "--agent-token-publish-subject ${lib.escapeShellArg "\$\$JS.API.DIRECT.GET.KV_agent-permissions.\$\$KV.agent-permissions.{agent}"}" + # Its own key of the `agents-md-initial` KV bucket + # (`swarm_queue_client::agents_md_initial`), on the same terms: the + # harness reads the text its `AGENTS.md` starts with. + "--agent-token-publish-subject ${lib.escapeShellArg "\$\$JS.API.DIRECT.GET.KV_agents-md-initial.\$\$KV.agents-md-initial.{agent}"}" # Every agent also reads the operator list, the one key of the # `operator-profiles` bucket, with one direct get. That grant # names no agent, so the responder adds it itself diff --git a/nix/module-eval/nats-authelia.nix b/nix/module-eval/nats-authelia.nix index d725cecb..026765d2 100644 --- a/nix/module-eval/nats-authelia.nix +++ b/nix/module-eval/nats-authelia.nix @@ -505,6 +505,7 @@ let "'$$JS.API.DIRECT.GET.KV_agent-schedules.$$KV.agent-schedules.{agent}'" "'$$JS.API.DIRECT.GET.KV_agent-inference.$$KV.agent-inference.{agent}'" "'$$JS.API.DIRECT.GET.KV_agent-permissions.$$KV.agent-permissions.{agent}'" + "'$$JS.API.DIRECT.GET.KV_agents-md-initial.$$KV.agents-md-initial.{agent}'" ]; } { diff --git a/swarm-controller/README.md b/swarm-controller/README.md index 346b0071..0176caf7 100644 --- a/swarm-controller/README.md +++ b/swarm-controller/README.md @@ -62,7 +62,9 @@ The swarm's control plane. The swarm UI and `swarmctl` are its clients. published it into the `agents-md` KV bucket (`GET /api/agents/{name}/agents-md`), and edits to it as git diffs, which this daemon hands to the agent's harness to apply - (`PUT /api/agents/{name}/agents-md`) → + (`PUT /api/agents/{name}/agents-md`). `POST /api/agents` can set the text the + file starts with (`agents_md`), stored in the `agents-md-initial` bucket for + the harness to write when no file exists → [`docs/swarm/README.md`](../docs/swarm/README.md#an-agents-agentsmd). - **Review classifier** — try-outs from the swarm UI and reviews on review requests, and the newest 100 finished runs, one document in the diff --git a/swarm-controller/src/agent_permissions.rs b/swarm-controller/src/agent_permissions.rs index 251f457e..8bb65c28 100644 --- a/swarm-controller/src/agent_permissions.rs +++ b/swarm-controller/src/agent_permissions.rs @@ -812,6 +812,7 @@ mod tests { inference: None, permissions, archetype: None, + agents_md: None, }), ) }; @@ -873,6 +874,7 @@ mod tests { inference: None, permissions: None, archetype: None, + agents_md: None, }), ) .await diff --git a/swarm-controller/src/agents_md_initial.rs b/swarm-controller/src/agents_md_initial.rs new file mode 100644 index 00000000..235ddc2c --- /dev/null +++ b/swarm-controller/src/agents_md_initial.rs @@ -0,0 +1,252 @@ +//! The text an agent's `AGENTS.md` starts with: one key per agent in +//! `agents-md-initial` (`swarm_queue_client::agents_md_initial`), which this +//! daemon alone writes. `POST /api/agents` stores it before it queues the +//! creation; a destroy deletes it. The agent's harness reads its own key at +//! start and writes the file only when none exists. + +use problem_details::ProblemDetails; +use swarm_queue_client::agents_md_initial::{self, Initial, MAX_BYTES}; + +use crate::routes::{AppState, error_problem}; +use crate::schedules::{queue, read_or_default, write_revision}; + +/// How many times a losing writer re-reads and re-applies, as in +/// `wanted::MAX_ATTEMPTS`. +const MAX_ATTEMPTS: usize = 5; + +fn queue_problem(e: &anyhow::Error) -> ProblemDetails { + tracing::warn!(error = %format!("{e:#}"), "agents-md initial text: queue operation failed"); + error_problem( + axum::http::StatusCode::INTERNAL_SERVER_ERROR, + &format!("{e:#}"), + ) +} + +/// Refuse a text longer than [`MAX_BYTES`]. +pub(crate) fn check(content: &str) -> Result<(), String> { + if content.len() > MAX_BYTES { + return Err(format!( + "agents_md is {} bytes; the limit is {MAX_BYTES}", + content.len() + )); + } + Ok(()) +} + +/// What [`store`] replaced, for [`restore`]; `None` when it wrote nothing. +pub(crate) type Replaced = Option>; + +/// Record that `agent`'s `AGENTS.md` starts with `content`. +pub(crate) async fn store( + client: &async_nats::Client, + agent: &str, + content: &str, +) -> Result { + let store = agents_md_initial::open_or_create(client) + .await + .map_err(|e| queue_problem(&e.into()))?; + let value = Initial { + content: content.to_owned(), + }; + for _ in 0..MAX_ATTEMPTS { + let (current, revision, undecodable) = read_or_default::(&store, agent) + .await + .map_err(|e| queue_problem(&e))?; + // An absent key reads as the default, which is not "stored": a first + // write of an empty text still has to land. + if current == value && !undecodable && revision.is_some() { + return Ok(None); + } + if let Some(written) = write_revision(&store, agent, &value, revision) + .await + .map_err(|e| queue_problem(&e))? + { + tracing::info!(%agent, bytes = content.len(), "set agent's initial AGENTS.md at create"); + return Ok(Some(crate::schedules::Replaced::new( + revision.map(|_| current), + written, + ))); + } + } + Err(queue_problem(&anyhow::anyhow!( + "gave up writing the initial AGENTS.md of {agent} after {MAX_ATTEMPTS} conflicting writes" + ))) +} + +/// Undo what [`store`] did, unless another write has landed since. Best +/// effort: a failure is logged, since the caller is already on its way to an +/// error of its own. +pub(crate) async fn restore(client: &async_nats::Client, agent: &str, replaced: Replaced) { + let Some(replaced) = replaced else { return }; + let undone = match agents_md_initial::open_or_create(client).await { + Ok(store) => replaced.undo(&store, agent).await, + Err(e) => Err(anyhow::anyhow!("{e:?}")), + }; + match undone { + Ok(true) => {} + Ok(false) => { + tracing::info!(%agent, "left the agent's initial AGENTS.md: another write landed since"); + } + Err(e) => { + tracing::warn!(%agent, error = %format!("{e:#}"), "could not put back the agent's initial AGENTS.md"); + } + } +} + +/// Delete a destroyed agent's key, so the next agent of that name does not +/// start from this one's text. A failure is logged and does not stop the +/// destroy, as in `agent_archetype::clear_on_destroy`. +pub(crate) async fn clear_on_destroy(state: &AppState, hive: &str, agent: &str) { + let cleared = async { + let client = queue(state).map_err(|(_, why)| anyhow::anyhow!(why))?; + agents_md_initial::open_or_create(&client) + .await? + .delete(agent) + .await?; + anyhow::Ok(()) + } + .await; + match cleared { + Ok(()) => tracing::info!(hive, agent, "destroyed agent's initial AGENTS.md cleared"), + Err(e) => tracing::error!( + hive, + agent, + error = %format!("{e:#}"), + "clearing this agent's initial AGENTS.md FAILED; it was destroyed anyway, so a new \ + agent of the same name starts from this text. Declaring the agent destroyed again \ + re-runs this." + ), + } +} + +#[cfg(test)] +mod tests { + use axum::extract::{Path, State}; + use axum::http::StatusCode; + + use super::*; + use crate::test_nats::TestNats; + + async fn published(client: &async_nats::Client, agent: &str) -> Option { + let store = agents_md_initial::open_or_create(client) + .await + .expect("bucket"); + store + .get(agent) + .await + .expect("read") + .map(|v| String::from_utf8(v.to_vec()).expect("utf-8")) + } + + fn request(name: &str, agents_md: Option) -> crate::routes::agents::CreateAgentRequest { + crate::routes::agents::CreateAgentRequest { + name: name.to_owned(), + hive: "pr1ma".to_owned(), + limits: None, + inference: None, + permissions: None, + archetype: None, + agents_md, + } + } + + #[test] + fn a_text_over_the_limit_is_refused_and_one_at_it_passes() { + assert_eq!(check(&"a".repeat(MAX_BYTES)), Ok(())); + let why = check(&"a".repeat(MAX_BYTES + 1)).unwrap_err(); + assert!(why.contains("the limit is"), "{why}"); + } + + /// `POST /api/agents` with `agents_md` stores it under the new agent's + /// key; a text over the limit is refused before anything is stored or + /// queued. + #[tokio::test] + async fn create_stores_the_text_it_is_given_and_refuses_one_too_big() { + let nats = TestNats::start(); + let client = nats.client().await; + let state = crate::agent_limits::tests::declared(&client).await; + + let problem = crate::routes::agents::create_agent( + State(state.clone()), + axum::Json(request("iris", Some("a".repeat(MAX_BYTES + 1)))), + ) + .await + .expect_err("too big"); + assert_eq!(problem.status, Some(StatusCode::BAD_REQUEST), "{problem:?}"); + assert_eq!(published(&client, "iris").await, None); + let queued = state + .jobq + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .graph() + .nodes() + .count(); + assert_eq!(queued, 0, "nothing is queued for a refused create"); + + let _ = crate::routes::agents::create_agent( + State(state.clone()), + axum::Json(request("iris", Some("# iris\n".to_owned()))), + ) + .await + .expect("created"); + assert_eq!( + published(&client, "iris").await.as_deref(), + Some(r##"{"content":"# iris\n"}"##) + ); + + // Control: no `agents_md` writes no key. + let _ = crate::routes::agents::create_agent( + State(state.clone()), + axum::Json(request("damocles", None)), + ) + .await + .expect("created"); + assert_eq!(published(&client, "damocles").await, None); + } + + /// An empty text is a text: the key is written, so the agent gets an + /// empty file rather than none. + #[tokio::test] + async fn an_empty_text_is_stored() { + let nats = TestNats::start(); + let client = nats.client().await; + let state = crate::agent_limits::tests::declared(&client).await; + let _ = crate::routes::agents::create_agent( + State(state.clone()), + axum::Json(request("iris", Some(String::new()))), + ) + .await + .expect("created"); + assert_eq!( + published(&client, "iris").await.as_deref(), + Some(r#"{"content":""}"#) + ); + } + + /// Destroying an agent deletes its key, so a new agent of that name does + /// not start from the old text. + #[tokio::test] + async fn a_recreated_agent_does_not_start_from_the_destroyed_ones_text() { + let nats = TestNats::start(); + let client = nats.client().await; + let state = crate::agent_limits::tests::declared(&client).await; + let _ = crate::routes::agents::create_agent( + State(state.clone()), + axum::Json(request("atlas", Some("old".to_owned()))), + ) + .await + .expect("created"); + assert!(published(&client, "atlas").await.is_some()); + + let _ = crate::routes::agents::set_agent_state( + State(state.clone()), + Path(("pr1ma".to_owned(), "atlas".to_owned())), + axum::Json(crate::routes::agents::SetAgentStateRequest { + state: swarm_queue_client::wanted::AgentState::Destroyed, + }), + ) + .await + .expect("destroyed"); + assert_eq!(published(&client, "atlas").await, None); + } +} diff --git a/swarm-controller/src/jobq.rs b/swarm-controller/src/jobq.rs index 908da079..54cb4ee6 100644 --- a/swarm-controller/src/jobq.rs +++ b/swarm-controller/src/jobq.rs @@ -1688,6 +1688,7 @@ mod tests { inference: None, permissions: None, archetype: None, + agents_md: None, }), ) .await @@ -1810,6 +1811,7 @@ mod tests { inference: None, permissions: None, archetype: None, + agents_md: None, }), ) .await @@ -1861,6 +1863,7 @@ mod tests { inference: None, permissions: None, archetype: None, + agents_md: None, }), ) .await @@ -1920,6 +1923,7 @@ mod tests { inference: None, permissions: None, archetype: None, + agents_md: None, }), ) .await @@ -2126,6 +2130,7 @@ mod tests { inference: None, permissions: None, archetype: None, + agents_md: None, }), ) .await diff --git a/swarm-controller/src/main.rs b/swarm-controller/src/main.rs index 59b114b2..1e927f79 100644 --- a/swarm-controller/src/main.rs +++ b/swarm-controller/src/main.rs @@ -33,6 +33,7 @@ mod agent_renewal; mod agent_state_stream; mod agent_status; mod agents_md; +mod agents_md_initial; mod alerts; mod archetypes; mod auth; diff --git a/swarm-controller/src/routes/agents.rs b/swarm-controller/src/routes/agents.rs index 49cff3aa..c8fd7218 100644 --- a/swarm-controller/src/routes/agents.rs +++ b/swarm-controller/src/routes/agents.rs @@ -16,8 +16,8 @@ use crate::jobq::{SwarmNodeKind, SwarmResourceKind, declare_agent_job}; use crate::routes::hives::HiveEntry; use crate::routes::{AppState, StatusUnavailable, error_problem, wanted_error_status}; use crate::{ - agent_archetype, agent_identity, agent_limits, agent_permissions, agent_status, auth, - inference, schedules, wanted, + agent_archetype, agent_identity, agent_limits, agent_permissions, agent_status, + agents_md_initial, auth, inference, schedules, wanted, }; /// The swarm's agent roster — every agent the swarm holds an identity for. @@ -214,6 +214,7 @@ pub(crate) async fn set_agent_state( revoke_or_complain(&hive, &agent).await; agent_permissions::clear_on_destroy(&state, &hive, &agent).await; agent_archetype::clear_on_destroy(&state, &hive, &agent).await; + agents_md_initial::clear_on_destroy(&state, &hive, &agent).await; } Ok(Json(render(&declaration))) } @@ -516,6 +517,13 @@ pub(crate) struct CreateAgentRequest { #[serde(default)] #[schema(value_type = Option, example = json!({"source": "hyperhive", "name": "reviewer"}))] pub(crate) archetype: Option, + /// The text the agent's `AGENTS.md` starts with, at most + /// `swarm_queue_client::agents_md_initial::MAX_BYTES` bytes. The agent's + /// harness writes it on its first start, and only when the agent has no + /// `AGENTS.md` yet: an existing file is never replaced. Absent leaves the + /// file to the harness. + #[serde(default)] + pub(crate) agents_md: Option, } /// Queue the whole agent-creation job graph for `name` — two independent @@ -549,10 +557,10 @@ pub(crate) struct CreateAgentRequest { request_body = CreateAgentRequest, responses( (status = 200, description = "job chain queued", body = CreateAgentResponse), - (status = 400, description = "`name` or `hive` is not a valid identifier, `name` is new and reserved or refused by the forge, `hive` is not in this swarm, `limits` or `inference` holds a value the matching `PUT` refuses, a permission is listed twice, or `archetype` is not one the stored archetype list holds (problem+json)", body = String), + (status = 400, description = "`name` or `hive` is not a valid identifier, `name` is new and reserved or refused by the forge, `hive` is not in this swarm, `limits` or `inference` holds a value the matching `PUT` refuses, a permission is listed twice, `archetype` is not one the stored archetype list holds, or `agents_md` is over the size limit (problem+json)", body = String), (status = 409, description = "the swarm has already placed `name` on a different hive (problem+json)", body = String), (status = 422, description = "an unknown field in the body, in `limits`, or in `inference`, an `effort` that is not one of the five levels, or a `permissions` entry that is not a permission name"), - (status = 503, description = "the swarm queue is not connected, so whether `name` is placed on another hive is unknown, or `limits`, `inference`, `permissions` or `archetype` is set and cannot be stored (problem+json)", body = String), + (status = 503, description = "the swarm queue is not connected, so whether `name` is placed on another hive is unknown, or `limits`, `inference`, `permissions`, `archetype` or `agents_md` is set and cannot be stored (problem+json)", body = String), (status = 500, description = "the job chain could not be queued, or another hive's wanted state could not be read (problem+json)", body = String), ), tag = "agents" @@ -567,7 +575,7 @@ pub(crate) async fn create_agent( // Settings are checked before anything is created, so a bad value stops // the request rather than leaving an agent without them. - let settings = CreateSettings::parse(req.limits, req.inference, req.permissions, req.archetype) + let settings = CreateSettings::parse(&req) .map_err(|why| error_problem(axum::http::StatusCode::BAD_REQUEST, &why))?; // Two checks, and the second is the one that makes the field worth @@ -665,6 +673,7 @@ pub(crate) async fn create_agent( settings.inference.as_ref(), settings.permissions.as_ref(), settings.archetype.as_ref(), + settings.agents_md.as_deref(), ) .await? } @@ -706,18 +715,25 @@ struct CreateSettings { inference: Option, permissions: Option, archetype: Option, + agents_md: Option, } impl CreateSettings { /// Each setting given, or why its `PUT` would refuse the first bad one. /// An archetype is checked here for its shape only; whether the swarm /// lists it is [`agent_archetype::check`]. - fn parse( - limits: Option, - inference: Option, - permissions: Option>, - archetype: Option, - ) -> Result { + fn parse(req: &CreateAgentRequest) -> Result { + let CreateAgentRequest { + limits, + inference, + permissions, + archetype, + agents_md, + .. + } = req.clone(); + if let Some(text) = &agents_md { + agents_md_initial::check(text)?; + } if let Some(a) = &archetype { for part in [&a.source, &a.name] { hive_types::Ident::parse(part) @@ -731,6 +747,7 @@ impl CreateSettings { .transpose()?, permissions: permissions.map(agent_permissions::at_create).transpose()?, archetype, + agents_md, }) } @@ -740,6 +757,7 @@ impl CreateSettings { || self.inference.is_some() || self.permissions.is_some() || self.archetype.is_some() + || self.agents_md.is_some() } /// The queue to store the settings through, `None` when none is given. @@ -767,10 +785,11 @@ struct StoredSettings { inference: inference::Replaced, permissions: agent_permissions::Replaced, archetype: agent_archetype::Replaced, + agents_md: agents_md_initial::Replaced, } impl StoredSettings { - /// Store whichever of `limits`, `permissions`, `inference` and `archetype` are + /// Store whichever of `limits`, `permissions`, `inference`, `archetype` and `agents_md` are /// given, in that order. A failure on one puts back the ones stored before it. async fn store( client: &async_nats::Client, @@ -779,6 +798,7 @@ impl StoredSettings { inference: Option<&swarm_queue_client::agent_inference::AgentInference>, permissions: Option<&swarm_queue_client::agent_permissions::AgentPermissions>, archetype: Option<&swarm_queue_client::agent_archetype::ArchetypeRef>, + agents_md: Option<&str>, ) -> Result { let mut stored = Self::default(); if let Some(limits) = limits { @@ -811,6 +831,15 @@ impl StoredSettings { } } } + if let Some(text) = agents_md { + match agents_md_initial::store(client, agent, text).await { + Ok(replaced) => stored.agents_md = replaced, + Err(e) => { + stored.roll_back(client, agent).await; + return Err(e); + } + } + } Ok(stored) } @@ -819,6 +848,7 @@ impl StoredSettings { inference::restore(client, agent, self.inference).await; agent_permissions::restore(client, agent, self.permissions).await; agent_archetype::restore(client, agent, self.archetype).await; + agents_md_initial::restore(client, agent, self.agents_md).await; } } @@ -1238,6 +1268,7 @@ mod tests { inference: None, permissions: None, archetype: None, + agents_md: None, }), ) .await @@ -1280,6 +1311,7 @@ mod tests { inference: None, permissions: None, archetype: None, + agents_md: None, }), ) .await @@ -1444,6 +1476,7 @@ mod tests { Some(&inference), None, None, + None, ) .await; assert!(stored.is_err(), "{agent}: the inference write must fail"); @@ -1583,6 +1616,7 @@ mod tests { inference: None, permissions: None, archetype: None, + agents_md: None, }), ) .await diff --git a/swarm-queue-client/Cargo.toml b/swarm-queue-client/Cargo.toml index b7a590c0..90d0987c 100644 --- a/swarm-queue-client/Cargo.toml +++ b/swarm-queue-client/Cargo.toml @@ -62,6 +62,10 @@ agent-permissions = ["dep:hive-sh4re", "async-nats/jetstream"] # `agent_archetype::read`, for hive-c0re, which reads the archetype of each # agent it provisions or rebuilds. Explicit for the same reason as `notices`. agent-archetype = ["async-nats/jetstream"] +# `agents_md_initial::read_own`, for the agent harness, which reads its own +# initial `AGENTS.md` text with one direct get. Explicit for the same reason as +# `notices`. +agents-md-initial = ["async-nats/jetstream"] [dependencies] # Bare (no `kv`/`jetstream`) unless a consumer opts into the `kv` feature diff --git a/swarm-queue-client/src/agents_md_initial.rs b/swarm-queue-client/src/agents_md_initial.rs new file mode 100644 index 00000000..1a47b6dc --- /dev/null +++ b/swarm-queue-client/src/agents_md_initial.rs @@ -0,0 +1,140 @@ +//! The text an agent's `AGENTS.md` starts with: one key per agent in the +//! `agents-md-initial` KV bucket, set when the agent is created. +//! +//! **swarm-controller is the only writer**: `POST /api/agents` stores the key +//! before it queues the agent's creation, and a destroy deletes it. +//! +//! **The agent's harness reads its own key** when it starts, and writes the +//! text to `AGENTS.md` only when the file does not exist. The file in the +//! agent's state dir stays the truth ([`crate::agents_md`]); this key is not +//! read again once the file exists, and later edits never reach it. +//! +//! Keyed by the agent alone, like [`crate::agent_archetype`]. The agent's +//! credential is granted one direct get on its own key and nothing else of +//! the bucket. + +/// The KV bucket the initial texts are stored in, keyed by agent name. +pub const BUCKET: &str = "agents-md-initial"; + +/// The stream backing [`BUCKET`]. `create_key_value` prefixes `KV_`. +pub const STREAM: &str = "KV_agents-md-initial"; + +/// The longest initial text, in bytes. The agent mirrors the file into +/// [`crate::agents_md`] whole, so the limit keeps it well under the queue's +/// payload limit. +pub const MAX_BYTES: usize = 64 * 1024; + +/// The subject `agent`'s entry is stored under. +#[must_use] +pub fn key_subject(agent: &str) -> String { + format!("$KV.{BUCKET}.{agent}") +} + +/// The request subject that reads `agent`'s current entry. +#[must_use] +pub fn direct_get_subject(agent: &str) -> String { + format!("$JS.API.DIRECT.GET.{STREAM}.{}", key_subject(agent)) +} + +/// The value under an agent's key. +#[derive(Debug, Clone, Default, PartialEq, Eq, serde::Serialize, serde::Deserialize)] +pub struct Initial { + /// The text `AGENTS.md` starts with. + pub content: String, +} + +/// Open the bucket for writing, creating it if nothing has yet. +/// +/// **Controller-side only.** `history: 1`: the agent wants the current value. +#[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-initial", + "The text each agent's AGENTS.md starts with, set by the swarm controller".to_owned(), + ) + .await +} + +/// Read `agent`'s own entry with one direct get. `None` when the key is +/// absent or deleted, or when no bucket exists yet. +/// +/// # Errors +/// +/// A failed request or an undecodable value. A request the queue refused for +/// permissions reaches the client as a timeout, not as a refusal. +#[cfg(feature = "agents-md-initial")] +pub async fn read_own( + client: &async_nats::Client, + agent: &str, +) -> Result, Box> { + use async_nats::jetstream::stream::DirectGetErrorKind; + + let js = async_nats::jetstream::new(client.clone()); + let stream = js.get_stream_no_info(STREAM).await?; + let message = match stream.direct_get_last_for_subject(key_subject(agent)).await { + Ok(message) => message, + Err(e) if matches!(e.kind(), DirectGetErrorKind::NotFound) => return Ok(None), + Err(e) + if matches!( + e.kind(), + DirectGetErrorKind::ErrorResponse(status, _) + if status == async_nats::StatusCode::NO_RESPONDERS + ) => + { + return Ok(None); + } + Err(e) => return Err(e.into()), + }; + Ok(decode( + message + .headers + .get("KV-Operation") + .map(async_nats::HeaderValue::as_str), + &message.payload, + )?) +} + +/// One stored message as a value: a KV delete or purge is stored as an empty +/// marker carrying the `KV-Operation` header, and reads as no entry. +#[cfg(feature = "agents-md-initial")] +fn decode(operation: Option<&str>, payload: &[u8]) -> serde_json::Result> { + if matches!(operation, Some("DEL" | "PURGE")) { + return Ok(None); + } + serde_json::from_slice(payload).map(Some) +} + +#[cfg(test)] +mod tests { + use super::*; + + /// The names the grant in `swarm-nats-auth` is built from. + #[test] + fn names_match_the_queue_grant() { + assert_eq!(STREAM, format!("KV_{BUCKET}")); + assert_eq!(key_subject("iris"), "$KV.agents-md-initial.iris"); + assert_eq!( + direct_get_subject("iris"), + "$JS.API.DIRECT.GET.KV_agents-md-initial.$KV.agents-md-initial.iris" + ); + } + + /// A stored text reads back, a delete is no entry. + #[cfg(feature = "agents-md-initial")] + #[test] + fn decoding_reads_the_text_and_a_delete_as_none() { + assert_eq!( + decode(None, br##"{"content":"# hi\n","later":1}"##).unwrap(), + Some(Initial { + content: "# hi\n".to_owned() + }) + ); + assert_eq!(decode(Some("DEL"), b"").unwrap(), None); + assert_eq!(decode(Some("PURGE"), b"").unwrap(), None); + } +} diff --git a/swarm-queue-client/src/lib.rs b/swarm-queue-client/src/lib.rs index 534e933e..cc74a575 100644 --- a/swarm-queue-client/src/lib.rs +++ b/swarm-queue-client/src/lib.rs @@ -214,6 +214,11 @@ pub mod agent_icon; /// `kv`. pub mod agents_md; +/// The text each agent's `AGENTS.md` starts with, set at create. The names and +/// types are unconditional; opening the bucket needs `kv`, reading it +/// `agents-md-initial`. +pub mod agents_md_initial; + /// 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`. diff --git a/swarmctl/src/agent.rs b/swarmctl/src/agent.rs index b71fa1a5..20827fc3 100644 --- a/swarmctl/src/agent.rs +++ b/swarmctl/src/agent.rs @@ -41,6 +41,9 @@ struct CreateAgentRequest<'a> { /// endpoint and not defaulted anywhere, because it is only knowable /// from the operator making the choice. hive: &'a str, + /// The text the agent's `AGENTS.md` starts with; left out when not given. + #[serde(skip_serializing_if = "Option::is_none")] + agents_md: Option<&'a str>, } /// Run `swarmctl agent create`. @@ -50,19 +53,30 @@ struct CreateAgentRequest<'a> { /// built here keeps that cost inside the one arm that incurs it, the same /// way `PathArgs::resolve` is called only in the arms that need the /// deployment's env vars. -pub(crate) fn create(target: &Target, name: &str, hive: &str) -> Result<()> { +pub(crate) fn create( + target: &Target, + name: &str, + hive: &str, + agents_md: Option<&Path>, +) -> Result<()> { // Client-side before the round trip, so a typo is an immediate local // error rather than a 400 the operator has to wait for. The controller // validates both again — that is the gate that matters, this is the // one that is fast. let name = parse_ident(name, "agent name")?; let hive = parse_ident(hive, "hive")?; + let agents_md = agents_md + .map(|path| { + std::fs::read_to_string(path) + .with_context(|| format!("reading --agents-md {}", path.display())) + }) + .transpose()?; let rt = tokio::runtime::Builder::new_current_thread() .enable_all() .build() .context("starting a tokio runtime for the controller request")?; - let resp = rt.block_on(post_create(target, &name, &hive))?; + let resp = rt.block_on(post_create(target, &name, &hive, agents_md.as_deref()))?; println!("queued: job node {}", resp.node_id); println!( @@ -114,11 +128,20 @@ pub(crate) fn mint_forge_token(target: &Target, name: &str) -> Result<()> { } /// One `POST /api/agents` round trip at the controller. -async fn post_create(target: &Target, name: &str, hive: &str) -> Result { +async fn post_create( + target: &Target, + name: &str, + hive: &str, + agents_md: Option<&str>, +) -> Result { post( target, "/api/agents", - &CreateAgentRequest { name, hive }, + &CreateAgentRequest { + name, + hive, + agents_md, + }, "create-agent", ) .await diff --git a/swarmctl/src/main.rs b/swarmctl/src/main.rs index 35352ca2..e6ec9442 100644 --- a/swarmctl/src/main.rs +++ b/swarmctl/src/main.rs @@ -426,6 +426,13 @@ struct AgentCreateArgs { /// hive roster and names the known hives if it misses. #[arg(long, value_name = "HIVE")] hive: String, + /// A text file the agent's `AGENTS.md` starts with. + /// + /// The agent writes it on its first start, and only when it has no + /// `AGENTS.md` yet: re-creating an agent never replaces an existing file. + /// The controller refuses a file over 64 KiB. + #[arg(long, value_name = "FILE")] + agents_md: Option, /// swarm-controller's unix socket. /// /// Supplied by the nix module that installs this binary, from the same @@ -668,7 +675,7 @@ fn main() -> Result<()> { command: AgentVerb::Create(args), } => { let target = remote.target(args.controller_socket)?; - agent::create(&target, &args.name, &args.hive) + agent::create(&target, &args.name, &args.hive, args.agents_md.as_deref()) } // Same socket-resolution reasoning as `Create` above. Verb::Agent {