Compare commits
29 changed files with 58 additions and 1492 deletions
|
|
@ -118,11 +118,11 @@ Swarm-level:
|
|||
**the socket's directory is its access control**; the crate README has
|
||||
the constraint and a unit test pins the path.
|
||||
- **`swarm-controller-api/`** — serde-only request and response bodies of
|
||||
the controller's HTTP API, shared with `swarmctl` (forge, schedules,
|
||||
`AGENTS.md` edits, and the agent-create and forge-token responses).
|
||||
the controller's HTTP API, shared with `swarmctl` (forge, schedules and
|
||||
the agent-create and forge-token responses).
|
||||
- **`swarmctl/`** — swarm operator CLI; runs as **root and acts directly**
|
||||
(README says why). Doesn't link `swarm-controller`: the forge, schedule
|
||||
and `AGENTS.md` bodies and the agent responses come from
|
||||
(README says why). Doesn't link `swarm-controller`: the forge and
|
||||
schedule bodies and the agent responses come from
|
||||
`swarm-controller-api`; the agent-create request, inference and
|
||||
permissions ones are a local mirror. ⚠️ Edits authelia's
|
||||
**`users.yml` in place**, shared with `swarm-authelia-bridge`; that
|
||||
|
|
|
|||
1
Cargo.lock
generated
1
Cargo.lock
generated
|
|
@ -1688,7 +1688,6 @@ dependencies = [
|
|||
"schemars",
|
||||
"serde",
|
||||
"serde_json",
|
||||
"sha1",
|
||||
"swarm-queue-client",
|
||||
"swarm-secret-client",
|
||||
"tempfile",
|
||||
|
|
|
|||
|
|
@ -276,10 +276,7 @@ Under `/var/lib/hyperhive/agents/<name>/`:
|
|||
`agents_md_migrate::run` renames a `state/CLAUDE.md` to `AGENTS.md`
|
||||
when no `AGENTS.md` exists. When both exist it renames neither and
|
||||
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:
|
||||
[`docs/swarm/README.md`](../swarm/README.md#an-agents-agentsmd).
|
||||
one exists in its cwd or above.
|
||||
- `harness/` — harness-internal ephemeral state; not intended for
|
||||
agent consumption. Bind-mounted to `/agents/<name>/harness`
|
||||
inside the container (`$HYPERHIVE_HARNESS_DIR`). Contents:
|
||||
|
|
|
|||
|
|
@ -278,19 +278,13 @@ The third thing an agent publishes is its **icon**, the same SVG its own
|
|||
`GET /icon` serves. It goes into the `agent-icons` KV bucket under the key
|
||||
`<agent>`, with no hive in it, so the swarm can show the icon of an agent that's
|
||||
stopped or has moved hives. The queue grants the agent's own credential
|
||||
`$KV.agent-icons.<agent>` and no other key of that bucket. The harness writes
|
||||
once per start, because a config change reaches an agent by restarting its
|
||||
container. An agent with no icon deletes its key. Swarm-side,
|
||||
`GET /api/agents/<name>/icon` serves
|
||||
`$KV.agent-icons.<agent>` and no other key. The harness writes once per start,
|
||||
because a config change reaches an agent by restarting its container. An agent
|
||||
with no icon deletes its key. Swarm-side, `GET /api/agents/<name>/icon` serves
|
||||
the stored bytes, and 404 means the agent has no icon. swarm-ui's agent cards
|
||||
load it as an `<img>` and show the dimmed hyperhive mark for an agent without
|
||||
one, as the hive dashboard does.
|
||||
|
||||
The fourth is its **`AGENTS.md`**, which goes into the `agents-md` bucket under
|
||||
the key `<agent>`, writable by the agent's own credential alone
|
||||
(`$KV.agents-md.<agent>`). The swarm reads that copy and edits the file through
|
||||
the agent; [An agent's AGENTS.md](#an-agents-agentsmd) has the details.
|
||||
|
||||
The queue also limits what an agent may **subscribe** to, because agents are the
|
||||
least trusted clients on the queue. An agent's own queue credential, which its
|
||||
subagent daemon also presents, may subscribe to its reply inbox
|
||||
|
|
@ -315,46 +309,6 @@ no replies, so every request it makes times out.
|
|||
|
||||
</details>
|
||||
|
||||
### An agent's AGENTS.md
|
||||
|
||||
The file in the agent's state dir, `state/AGENTS.md`, is the only copy that
|
||||
counts, and only the agent writes it: itself, or its harness applying an edit
|
||||
from the swarm. No hive is in either path.
|
||||
|
||||
**Reading.** The harness publishes the file's text and its revision, the git
|
||||
blob id that `git hash-object` prints for it, at start, within ten seconds of
|
||||
any change, and straight after an edit it applied. An agent with no `AGENTS.md`
|
||||
publishes `""`. `GET /api/agents/<name>/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.
|
||||
|
||||
**Editing.** `PUT /api/agents/<name>/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
|
||||
the agent's command subject `$SWARM.control.<name>`, as it sends a cancel. The
|
||||
harness refuses it when the file isn't at `base_revision`. Otherwise it runs
|
||||
`git apply --check` and then `git apply` on a copy of the file, and writes the
|
||||
result over the file only when both pass:
|
||||
|
||||
| Status | Meaning |
|
||||
| ------ | --------------------------------------------------------------------------------------------------------------------------------------- |
|
||||
| 200 | applied; the body is `{revision}`, the file's new revision |
|
||||
| 409 | the file isn't at `base_revision` any more; nothing applied, and `revision` in the body is the one it's at |
|
||||
| 422 | `git apply` refused the diff, or it touches a file other than `AGENTS.md`; `detail` carries git's message, and the file stays as it was |
|
||||
| 503 | the agent isn't running or isn't connected to the queue |
|
||||
| 504 | the harness didn't answer within ten seconds, so whether the edit applied is unknown |
|
||||
| 500 | the harness couldn't read or write the file or run git, or the queue failed the request |
|
||||
|
||||
The harness refuses a diff that sets any file mode other than `100644`, so an
|
||||
edit can't turn `AGENTS.md` into a symlink. `PUT` is operator-only, behind
|
||||
`Remote-User` like cancel: whoever can call it can rewrite the agent's
|
||||
instructions, so don't widen who holds it without revisiting this.
|
||||
|
||||
An edit needs the agent running, and a read of a stopped agent shows the text
|
||||
it last published. A harness older than this route doesn't know the command and
|
||||
drops it unanswered, which reads as a 504. `swarmctl agent agents-md get` and
|
||||
`apply` call the two routes.
|
||||
|
||||
### Swarm-wide forge objects
|
||||
|
||||
The controller also keeps the forge objects that are one per swarm, not
|
||||
|
|
|
|||
|
|
@ -16,9 +16,6 @@ This document contains the help content for the `swarmctl` command-line program.
|
|||
* [`swarmctl agent permissions get`↴](#swarmctl-agent-permissions-get)
|
||||
* [`swarmctl agent permissions set`↴](#swarmctl-agent-permissions-set)
|
||||
* [`swarmctl agent permissions clear`↴](#swarmctl-agent-permissions-clear)
|
||||
* [`swarmctl agent agents-md`↴](#swarmctl-agent-agents-md)
|
||||
* [`swarmctl agent agents-md get`↴](#swarmctl-agent-agents-md-get)
|
||||
* [`swarmctl agent agents-md apply`↴](#swarmctl-agent-agents-md-apply)
|
||||
* [`swarmctl schedule`↴](#swarmctl-schedule)
|
||||
* [`swarmctl schedule add`↴](#swarmctl-schedule-add)
|
||||
* [`swarmctl schedule list`↴](#swarmctl-schedule-list)
|
||||
|
|
@ -74,7 +71,6 @@ Manage agents across the swarm
|
|||
* `mint-forge-token` — Check one agent's forge token, and mint it if it's missing or stale
|
||||
* `inference` — Show the model, effort, subagent models and subagent cap the swarm sets for an agent, or change the model, effort and subagent models
|
||||
* `permissions` — Show or change the permissions the swarm grants an agent
|
||||
* `agents-md` — Show an agent's AGENTS.md, or edit it with a diff
|
||||
|
||||
|
||||
|
||||
|
|
@ -278,61 +274,6 @@ Clear the agent's permissions, so the default applies again
|
|||
|
||||
|
||||
|
||||
## `swarmctl agent agents-md`
|
||||
|
||||
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 isn't running. `apply` hands the diff to the agent, which applies it to the file with `git apply`; it needs the agent running.
|
||||
|
||||
**Usage:** `swarmctl agent agents-md <COMMAND>`
|
||||
|
||||
###### **Subcommands:**
|
||||
|
||||
* `get` — Print an agent's AGENTS.md on stdout, and its revision on stderr
|
||||
* `apply` — Apply a unified diff to an agent's AGENTS.md
|
||||
|
||||
|
||||
|
||||
## `swarmctl agent agents-md get`
|
||||
|
||||
Print an agent's AGENTS.md on stdout, and its revision on stderr
|
||||
|
||||
**Usage:** `swarmctl agent agents-md get [OPTIONS] <AGENT>`
|
||||
|
||||
###### **Arguments:**
|
||||
|
||||
* `<AGENT>` — Agent name
|
||||
|
||||
###### **Options:**
|
||||
|
||||
* `--controller-socket <PATH>` — 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`.
|
||||
|
||||
|
||||
|
||||
## `swarmctl agent agents-md apply`
|
||||
|
||||
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.
|
||||
|
||||
**Usage:** `swarmctl agent agents-md apply [OPTIONS] --base <REVISION> <AGENT> <DIFF_FILE>`
|
||||
|
||||
###### **Arguments:**
|
||||
|
||||
* `<AGENT>` — Agent name
|
||||
* `<DIFF_FILE>` — The file holding the diff
|
||||
|
||||
###### **Options:**
|
||||
|
||||
* `--base <REVISION>` — The revision you made the diff against, as `get` printed it. It's the git blob id of that text, so `git hash-object` of the unedited file prints it too
|
||||
* `--controller-socket <PATH>` — 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`.
|
||||
|
||||
|
||||
|
||||
## `swarmctl schedule`
|
||||
|
||||
Manage scheduled prompts.
|
||||
|
|
|
|||
|
|
@ -38,8 +38,6 @@ 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.
|
||||
|
|
@ -53,9 +51,6 @@ 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
|
||||
|
|
@ -64,6 +59,7 @@ 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.
|
||||
|
|
|
|||
|
|
@ -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.
|
||||
pub(crate) fn agents_md_path(state_dir: &Path) -> std::path::PathBuf {
|
||||
fn agents_md_path(state_dir: &Path) -> std::path::PathBuf {
|
||||
state_dir.join("AGENTS.md")
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -35,7 +35,6 @@ 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;
|
||||
|
|
@ -505,9 +504,6 @@ 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.
|
||||
|
|
|
|||
|
|
@ -1,506 +0,0 @@
|
|||
//! 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 };
|
||||
}
|
||||
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::<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:?}"),
|
||||
};
|
||||
}
|
||||
// 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 <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(["-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<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,12 +21,10 @@
|
|||
//! request that times out as "a subscriber held it".
|
||||
//!
|
||||
//! **Commands are a separate subject and are answered.** The swarm UI's
|
||||
//! 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`).
|
||||
//! 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`).
|
||||
|
||||
use futures_util::{Stream, StreamExt as _};
|
||||
use hive_sh4re::inbox::DeliveredMessage;
|
||||
|
|
@ -72,45 +70,35 @@ async fn run_control(
|
|||
tracing::warn!("swarm control: command without a reply subject dropped");
|
||||
continue;
|
||||
};
|
||||
let Some(answer) = act(
|
||||
&msg.payload,
|
||||
|| crate::web_ui::cancel_turn(&bus, &interrupted, "swarm UI: cancel"),
|
||||
crate::swarm_agents_md::handle,
|
||||
)
|
||||
.await
|
||||
else {
|
||||
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 {
|
||||
continue;
|
||||
};
|
||||
if let Err(e) = client.publish(reply, answer.into()).await {
|
||||
let payload = serde_json::to_vec(&answer).expect("a reply encodes");
|
||||
if let Err(e) = client.publish(reply, payload.into()).await {
|
||||
tracing::warn!(error = %e, "swarm control: answering failed");
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// 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>(
|
||||
/// 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(
|
||||
payload: &[u8],
|
||||
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),
|
||||
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),
|
||||
}),
|
||||
Err(e) => {
|
||||
tracing::warn!(error = %e, "swarm control: undecodable command dropped");
|
||||
return None;
|
||||
None
|
||||
}
|
||||
};
|
||||
Some(encoded.expect("a reply encodes"))
|
||||
}
|
||||
}
|
||||
|
||||
async fn run(tx: UnboundedSender<DeliveredMessage>, agent: String) {
|
||||
|
|
@ -168,7 +156,6 @@ 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> {
|
||||
|
|
@ -217,58 +204,20 @@ mod tests {
|
|||
forward(payloads, &tx).await;
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn a_cancel_is_answered_with_whether_it_acted() {
|
||||
#[test]
|
||||
fn a_cancel_is_answered_with_whether_it_acted() {
|
||||
for acted in [true, false] {
|
||||
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 });
|
||||
let got = act(br#""cancel""#, |c| {
|
||||
assert_eq!(c, AgentControl::Cancel);
|
||||
acted
|
||||
});
|
||||
assert_eq!(got, Some(AgentControlReply { acted }));
|
||||
}
|
||||
}
|
||||
|
||||
#[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;
|
||||
#[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"));
|
||||
assert_eq!(got, None);
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -884,10 +884,6 @@ 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
|
||||
|
|
|
|||
|
|
@ -500,7 +500,6 @@ 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}'"
|
||||
|
|
|
|||
|
|
@ -17,8 +17,7 @@ controller's `OpenAPI` spec. `swarmctl` leaves it off.
|
|||
## What's here
|
||||
|
||||
One module per endpoint family: `agents` (the responses of `POST /api/agents`
|
||||
and `POST /api/agents/{name}/forge-token`), `agents_md`
|
||||
(`/api/agents/{name}/agents-md`), `forge`
|
||||
and `POST /api/agents/{name}/forge-token`), `forge`
|
||||
(`POST /api/forge/users/{name}/admin`) and `schedules` (`/api/schedules`).
|
||||
The agent-create request, inference and permissions bodies are still mirrored
|
||||
by hand in `swarmctl`.
|
||||
|
|
|
|||
|
|
@ -1,51 +0,0 @@
|
|||
//! `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::<AgentsMdEdit>(wire).unwrap(), edit);
|
||||
}
|
||||
}
|
||||
|
|
@ -5,6 +5,5 @@
|
|||
//! `schema` feature each type also derives `utoipa::ToSchema`.
|
||||
|
||||
pub mod agents;
|
||||
pub mod agents_md;
|
||||
pub mod forge;
|
||||
pub mod schedules;
|
||||
|
|
|
|||
|
|
@ -58,12 +58,6 @@ The swarm's control plane. The swarm UI and `swarmctl` are its clients.
|
|||
(`GET`/`PUT`/`DELETE /api/agents/{name}/limits`); the hive running the
|
||||
agent reads it, and each cap replaces the hive's own answer for that field →
|
||||
[`docs/scheduler/coordinator.md`](../docs/scheduler/coordinator.md#container-resource-limits).
|
||||
- **Agent instructions** — each agent's `AGENTS.md` as its harness last
|
||||
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`) →
|
||||
[`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
|
||||
`review-classifier-runs` KV bucket, which this daemon alone writes
|
||||
|
|
|
|||
|
|
@ -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 with this payload, a reply to the command sent.
|
||||
Answered(Vec<u8>),
|
||||
/// The agent answered.
|
||||
Answered(AgentControlReply),
|
||||
/// Nothing subscribes to the agent's control subject.
|
||||
NotRunning,
|
||||
/// A subscriber held the command and did not answer in time.
|
||||
|
|
@ -50,7 +50,10 @@ 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) => Outcome::Answered(reply.payload.to_vec()),
|
||||
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}")),
|
||||
},
|
||||
Err(e) => match e.kind() {
|
||||
RequestErrorKind::TimedOut => Outcome::Unconfirmed,
|
||||
RequestErrorKind::NoResponders => Outcome::NotRunning,
|
||||
|
|
@ -114,7 +117,7 @@ pub(crate) async fn post_agent_cancel(
|
|||
|
||||
/// The agent and who is asking, or the refusal. Everything decided before the
|
||||
/// queue is touched.
|
||||
pub(crate) fn target(
|
||||
fn target(
|
||||
headers: &HeaderMap,
|
||||
agent: &str,
|
||||
) -> Result<(String, String), (StatusCode, &'static str)> {
|
||||
|
|
@ -140,12 +143,6 @@ 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 }))
|
||||
}
|
||||
|
|
@ -234,14 +231,10 @@ mod tests {
|
|||
async fn the_answer_is_relayed_from_the_agent_s_own_control_subject() {
|
||||
let answers: [(bool, fn() -> Outcome); 2] = [
|
||||
(true, || {
|
||||
Outcome::Answered(
|
||||
serde_json::to_vec(&AgentControlReply { acted: true }).expect("encodes"),
|
||||
)
|
||||
Outcome::Answered(AgentControlReply { acted: true })
|
||||
}),
|
||||
(false, || {
|
||||
Outcome::Answered(
|
||||
serde_json::to_vec(&AgentControlReply { acted: false }).expect("encodes"),
|
||||
)
|
||||
Outcome::Answered(AgentControlReply { acted: false })
|
||||
}),
|
||||
];
|
||||
for (acted, answer) in answers {
|
||||
|
|
|
|||
|
|
@ -493,7 +493,6 @@ 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(
|
||||
|
|
|
|||
|
|
@ -1,436 +0,0 @@
|
|||
//! `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<async_nats::jetstream::kv::Store>,
|
||||
}
|
||||
|
||||
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<Option<Mirror>> {
|
||||
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<AppState>,
|
||||
Path(name): Path<String>,
|
||||
) -> Result<Json<AgentsMdView>, 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<AppState>,
|
||||
Path(agent): Path<String>,
|
||||
headers: HeaderMap,
|
||||
Json(edit): Json<AgentsMdEdit>,
|
||||
) -> Result<Json<AgentsMdSaved>, 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<Json<AgentsMdSaved>, 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::<ApplyReply>(&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<dyn Fn() -> Outcome + Send + Sync>,
|
||||
got: Mutex<Vec<(String, Vec<u8>)>>,
|
||||
}
|
||||
|
||||
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<u8>) -> 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);
|
||||
}
|
||||
}
|
||||
|
|
@ -153,7 +153,6 @@ 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(
|
||||
|
|
|
|||
|
|
@ -123,7 +123,6 @@ 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(
|
||||
|
|
|
|||
|
|
@ -962,7 +962,6 @@ 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(
|
||||
|
|
|
|||
|
|
@ -55,7 +55,6 @@ mod agent_permissions;
|
|||
mod agent_renewal;
|
||||
mod agent_state_stream;
|
||||
mod agent_status;
|
||||
mod agents_md;
|
||||
mod alerts;
|
||||
mod archetypes;
|
||||
mod auth;
|
||||
|
|
@ -955,9 +954,6 @@ struct AppState {
|
|||
/// connection, several consumers" rationale as `agent_status` above,
|
||||
/// and `None` in exactly the state `status` is.
|
||||
agent_icons: Option<Arc<agent_icon::AgentIconReader>>,
|
||||
/// Each agent's published `AGENTS.md`, on `status`'s connection like
|
||||
/// `agent_icons`, and `None` in exactly the state `status` is.
|
||||
agents_md: Option<Arc<agents_md::AgentsMdReader>>,
|
||||
/// What each agent last reported running on, from its turn-state
|
||||
/// headers on `status`'s connection. `None` in exactly the state
|
||||
/// `status` is.
|
||||
|
|
@ -1329,17 +1325,6 @@ 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<status::StatusReader>>,
|
||||
) -> Option<Arc<agents_md::AgentsMdReader>> {
|
||||
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(
|
||||
|
|
@ -3627,7 +3612,6 @@ 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())),
|
||||
|
|
@ -3646,7 +3630,8 @@ async fn main() -> Result<()> {
|
|||
alerts: alerts::from_env(),
|
||||
};
|
||||
|
||||
axum::serve(listener, build_app(state))
|
||||
let app = build_app(state);
|
||||
axum::serve(listener, app)
|
||||
.await
|
||||
.context("serving swarm-controller")
|
||||
}
|
||||
|
|
@ -3669,7 +3654,6 @@ 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,
|
||||
|
|
@ -3849,7 +3833,6 @@ 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),
|
||||
|
|
|
|||
|
|
@ -457,7 +457,6 @@ 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(
|
||||
|
|
|
|||
|
|
@ -1,103 +0,0 @@
|
|||
//! 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.<agent>` 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.<bucket>.<key>`.
|
||||
#[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<async_nats::jetstream::kv::Store, Error> {
|
||||
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"}"#);
|
||||
}
|
||||
}
|
||||
|
|
@ -209,11 +209,6 @@ 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`.
|
||||
|
|
@ -432,18 +427,14 @@ 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, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
|
||||
#[derive(Debug, Clone, Copy, 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::Cancel`] request with.
|
||||
/// What an agent answers an [`AgentControl`] 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
|
||||
|
|
|
|||
|
|
@ -7,9 +7,8 @@ that talk to the controller, from anywhere as your own SSO account (see
|
|||
|
||||
Distinct from `hivectl`, which drives one hive's `hive-c0re` over its
|
||||
admin socket. This crate doesn't link `swarm-controller`, for the same
|
||||
reason `hivectl` doesn't link `hive-c0re`; the forge, schedule and
|
||||
`AGENTS.md` bodies and the agent responses both sides speak come from
|
||||
`swarm-controller-api`,
|
||||
reason `hivectl` doesn't link `hive-c0re`; the forge and schedule bodies
|
||||
and the agent responses both sides speak come from `swarm-controller-api`,
|
||||
as `hivectl`'s come from `hive-host-sock`.
|
||||
|
||||
## Why root, and why no socket
|
||||
|
|
|
|||
|
|
@ -1,68 +0,0 @@
|
|||
//! `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::Runtime> {
|
||||
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<String> {
|
||||
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(())
|
||||
}
|
||||
|
|
@ -26,7 +26,6 @@
|
|||
//! for the same reason `hivectl` does not link `hive-c0re`.
|
||||
|
||||
mod agent;
|
||||
mod agents_md;
|
||||
mod forge;
|
||||
mod inference;
|
||||
mod login;
|
||||
|
|
@ -303,43 +302,6 @@ 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 isn't running. `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 you made the diff against, as `get` printed it. It's
|
||||
/// 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)]
|
||||
|
|
@ -683,9 +645,6 @@ 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),
|
||||
|
|
@ -765,15 +724,6 @@ 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(
|
||||
|
|
|
|||
Loading…
Reference in a new issue