//! 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 swarm_queue_client::agents_md_initial::Initial; 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; }); } /// 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 { 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 }; } if let Some(mode) = non_regular_mode(&apply.diff) { return ApplyReply::Rejected { error: format!( "the diff sets file mode {mode}; AGENTS.md must stay a regular file (100644)" ), }; } 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:?}"), }; } // Belt and braces for the mode check above: never read through a link. if !std::fs::symlink_metadata(©).is_ok_and(|m| m.file_type().is_file()) { return ApplyReply::Rejected { error: "the diff left AGENTS.md as something other than a regular file".to_owned(), }; } 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), } } /// The first file mode a diff header sets that is not `100644`, if any. fn non_regular_mode(diff: &str) -> Option<&str> { diff.lines() .filter_map(|l| { l.strip_prefix("new file mode ") .or_else(|| l.strip_prefix("new mode ")) }) .map(str::trim) .find(|m| *m != "100644") } /// 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(["-c", "core.symlinks=false"]) .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"); } /// A symlink diff must not make the harness read the link's target into /// AGENTS.md, which the mirror then publishes. #[test] fn a_diff_that_creates_a_symlink_is_refused_and_nothing_is_read_through_it() { let secret = tempfile::tempdir().expect("tempdir"); let target = secret.path().join("secret"); std::fs::write(&target, "do not publish\n").expect("write"); let link = format!( "diff --git a/AGENTS.md b/AGENTS.md\nnew file mode 120000\n--- /dev/null\n+++ b/AGENTS.md\n@@ -0,0 +1 @@\n+{}\n\\ No newline at end of file\n", target.display() ); // Fresh create. let dir = tempfile::tempdir().expect("tempdir"); let reply = apply_diff(dir.path(), &apply("", &link)); assert!(matches!(reply, ApplyReply::Rejected { .. }), "{reply:?}"); assert!(!dir.path().join("AGENTS.md").exists()); // Typechange of an existing file. let dir = state(BEFORE); let typechange = format!( "diff --git a/AGENTS.md b/AGENTS.md\ndeleted file mode 100644\n--- a/AGENTS.md\n+++ /dev/null\n@@ -1,2 +0,0 @@\n-# atlas\n-keep notes short\n{link}" ); let reply = apply_diff(dir.path(), &apply(BEFORE, &typechange)); assert!(matches!(reply, ApplyReply::Rejected { .. }), "{reply:?}"); assert_eq!(file(&dir), BEFORE); } /// The mode check is the first layer; with it out of the way git must /// still not write a link (`core.symlinks=false`) and the regular-file /// check must hold. #[test] fn a_link_never_survives_the_apply_even_without_the_mode_check() { let dir = tempfile::tempdir().expect("tempdir"); let work = tempfile::tempdir().expect("tempdir"); let diff = "diff --git a/AGENTS.md b/AGENTS.md\nnew file mode 120000\n--- /dev/null\n+++ b/AGENTS.md\n@@ -0,0 +1 @@\n+/etc/hostname\n\\ No newline at end of file\n"; super::git(work.path(), &["apply"], diff) .expect("git runs") .expect("applies"); let meta = std::fs::symlink_metadata(work.path().join("AGENTS.md")).expect("exists"); assert!(meta.file_type().is_file(), "git wrote a link"); drop(dir); } /// 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(); } 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)); } }