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.<agent>` key. git needs no new package: agent containers enable programs.git, and the harness unit's PATH includes /run/current-system/sw.
This commit is contained in:
atlas
mara
parent
f4f459f0b2
commit
88674eb63a
22 changed files with 1296 additions and 47 deletions
|
|
@ -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")
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -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<S: Surface>(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.
|
||||
|
|
|
|||
437
hive-agent/src/swarm_agents_md.rs
Normal file
437
hive-agent/src/swarm_agents_md.rs
Normal file
|
|
@ -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 <len>\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<Vec<u8>> {
|
||||
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<Mirror> {
|
||||
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<F, Fut>(state_dir: &Path, poll: Duration, wake: &Notify, mut publish: F)
|
||||
where
|
||||
F: FnMut(Mirror) -> Fut,
|
||||
Fut: Future<Output = Published>,
|
||||
{
|
||||
let mut written: Option<String> = None;
|
||||
// A failure that repeats every poll is logged once, until it changes.
|
||||
let mut last_warning: Option<String> = 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 `<state_dir>/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::<Vec<_>>(),
|
||||
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 <args>` 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<Result<(), String>> {
|
||||
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<Mirror>) -> Option<Mirror> {
|
||||
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::<Mirror>();
|
||||
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();
|
||||
}
|
||||
}
|
||||
|
|
@ -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<F>(
|
||||
payload: &[u8],
|
||||
cancel: impl FnOnce(swarm_queue_client::AgentControl) -> bool,
|
||||
) -> Option<swarm_queue_client::AgentControlReply> {
|
||||
match serde_json::from_slice::<swarm_queue_client::AgentControl>(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<Vec<u8>>
|
||||
where
|
||||
F: Future<Output = swarm_queue_client::agents_md::ApplyReply>,
|
||||
{
|
||||
use swarm_queue_client::AgentControl;
|
||||
|
||||
let encoded = match serde_json::from_slice::<AgentControl>(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<DeliveredMessage>, agent: String) {
|
||||
|
|
@ -156,6 +168,7 @@ fn delivered(payload: &[u8]) -> Option<DeliveredMessage> {
|
|||
#[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<u8> {
|
||||
|
|
@ -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);
|
||||
}
|
||||
}
|
||||
|
|
|
|||
Loading…
Reference in a new issue