diff --git a/Cargo.lock b/Cargo.lock
index 4e71f1ae..1f16b1e2 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -4909,6 +4909,7 @@ version = "0.1.0"
dependencies = [
"anyhow",
"bytes",
+ "chrono",
"clap",
"clap-markdown",
"clap_complete",
diff --git a/docs/README.md b/docs/README.md
index 196e6571..b7a25bc1 100644
--- a/docs/README.md
+++ b/docs/README.md
@@ -82,6 +82,8 @@ declarations.
[`swarm/secrets.md`](swarm/secrets.md) for where each file lives today.
- **How does the secret store come up, who writes its grants, how's it
sealed?** → [`swarm/bao.md`](swarm/bao.md).
+- **How does an agent get a prompt at set times, and why doesn't it fire
+ twice after downtime?** → [`swarm/schedules.md`](swarm/schedules.md).
## Scheduler, CI, observability
diff --git a/docs/swarm/README.md b/docs/swarm/README.md
index eb2fbd3a..bb0a9c7f 100644
--- a/docs/swarm/README.md
+++ b/docs/swarm/README.md
@@ -21,6 +21,8 @@ grafana, the controller, authelia, …) →
- **control plane** — `swarm-controller`: hive directory, agent roster, job
graph, agent creation. → [below](#swarm-controller)
+- **scheduled prompts** — named prompts for one or more agents at set times,
+ kept by the swarm and fired in each agent's container. → [`schedules.md`](schedules.md)
- **swarm UI** — the operator's day-to-day surface, on the swarm apex,
`admins` only. → [`ui.md`](ui.md)
- **shared services** — one forge, homeserver, SSO, queue and metrics/logs
diff --git a/docs/swarm/schedules.md b/docs/swarm/schedules.md
new file mode 100644
index 00000000..fe95a472
--- /dev/null
+++ b/docs/swarm/schedules.md
@@ -0,0 +1,150 @@
+# Scheduled prompts
+
+A scheduled prompt hands one or more agents the same text at set times: once,
+or every interval from a start time. Each prompt has a name, its title. The
+swarm keeps every prompt, and each agent it names fires it in its own
+container, so a fire needs neither the hive nor the swarm controller to be up
+at that moment, only the agent.
+
+Operators manage them with `swarmctl schedule` on the swarm controller's
+host. Agents can't create, change or remove one. For a one-off wake an agent
+sets itself, it uses `remind`, which stays inside its container.
+
+## Managing them
+
+```console
+# swarmctl schedule add --name standup --agent scribe --agent iris \
+ --body "post the standup summary" --at 2026-10-05T09:00:00+02:00 --every 1d
+added schedule kqzpmwab
+kqzpmwab "standup" every 1d from 2026-10-05T07:00:00Z next 2026-10-05T07:00:00Z agents: iris, scribe | post the standup summary
+
+# swarmctl schedule list
+# swarmctl schedule list --agent scribe
+# swarmctl schedule edit kqzpmwab --body "post the standup summary to the team room"
+# swarmctl schedule edit kqzpmwab --add-agent atlas --remove-agent iris
+# swarmctl schedule pause kqzpmwab
+# swarmctl schedule resume kqzpmwab
+# swarmctl schedule rm kqzpmwab
+```
+
+- `--agent` names an agent to fire the prompt in; repeat it for more. Each
+ must be an agent some hive's wanted state declares.
+- `--at` takes an RFC 3339 time with its offset, or unix seconds. Without it
+ the first fire is now. A time in the past fires its latest past time once,
+ as soon as each agent reads the prompt.
+- `--every` takes a count and a unit, `s`, `m`, `h` or `d`. Without it the
+ prompt fires once.
+- `schedule edit` changes only the flags you pass and keeps the id. `--once`
+ turns a recurring prompt into a one-shot. `--add-agent` and
+ `--remove-agent` change who gets it; removing the last agent fails, so
+ remove the prompt instead.
+- `schedule list` shows one row per prompt: id, name, cadence, start, next
+ fire time, agents, and the first line of the body.
+
+The full flag reference is [`swarmctl-cli.md`](../tools/swarmctl-cli.md).
+
+## When a prompt fires
+
+A prompt's fire times are its start time plus whole intervals. Each agent
+works them out from the prompt itself, so:
+
+- **Editing the text or the name doesn't move them.** Neither does changing
+ the agents. Only a new `--at` or `--every` does.
+- **An agent that was down or disconnected fires once.** When it's back, it
+ fires the latest time it missed and then carries on at the next one. The
+ earlier missed times don't fire.
+- **A restart refires nothing.** Each agent records the last time it fired
+ for each prompt in its state database, and fires a time only when it's
+ later than that. An agent removed from a prompt and added back keeps that
+ record, so it doesn't refire a time it already fired.
+- **A one-shot fires once**, even though it stays in `schedule list` until
+ you remove it. Moving its `--at` to a later time fires it again then.
+- **A paused prompt doesn't fire**, and the times that pass while it's paused
+ don't fire when it resumes. An agent marks those times as passed while it
+ watches the paused prompt; if the agent was down for the whole pause, it
+ fires the latest time it missed once, as above.
+
+A fire reaches the agent as a `schedule` loose end with the prompt's text,
+the same as a hive schedule's. The agent's turn loop wakes for it like any
+other loose end. If the agent hasn't dealt with the previous fire yet, the
+new fire replaces it rather than adding a second one; once the agent
+dismissed the previous one, the next fire opens a new one.
+
+Each agent reads its prompts every 10 seconds, so a fire can land up to 10
+seconds after its time. A change reaches each agent within the same window,
+or within a minute if the controller's first attempt to pass it on failed.
+
+## When a prompt doesn't fire
+
+- **The agent has no queue credential of its own.** The agent reads its
+ prompts with its own swarm-queue credential. An agent connected with its
+ hive's shared client logs `connected with the hive's shared client, which
+has no read grant on the agent-schedules bucket` and fires nothing.
+- **The agent's first connect with its own credential failed.** The harness
+ keeps the outcome of its first connect for the rest of its run: it falls
+ back to the hive's shared client (the case above) or stays disconnected,
+ and reads no prompts until it restarts.
+- **The agent hasn't read its prompts since it started.** It fires from the
+ last list it read; until the first read succeeds it fires nothing. A read
+ that fails later keeps the previous list.
+
+Implementation
+
+swarm-controller is the only client that writes either of the two KV buckets
+involved.
+
+`schedule-definitions` holds every prompt as one JSON document under the key
+`all`. Agents hold no grant on this bucket. Each change re-reads the key and
+writes against its revision, so a prompt exists for every one of its agents or for
+none of them, and two changes at once don't lose one.
+
+`agent-schedules` holds one key per agent: the prompts naming that agent,
+without their names or other agents.
+
+```json
+{
+ "schedules": [
+ {
+ "id": "kqzpmwab",
+ "body": "post the standup summary",
+ "anchor_unix": 1791183600,
+ "interval_seconds": 86400,
+ "paused": false
+ }
+ ]
+}
+```
+
+`interval_seconds` is absent for a one-shot. After each change the controller
+rewrites the key of every agent the change named before or after, and once a
+minute it rewrites every agent's key, so a rewrite that failed converges. Ids
+are 8 lowercase letters, the same in every agent's key, never reused, and so
+never equal to a hive schedule's numeric id.
+
+`swarmctl schedule` calls these routes over the controller's socket:
+
+| method | path | body |
+| -------- | --------------------- | ------------------------------------------------------------------------------------------------------------------------- |
+| `GET` | `/api/schedules` | —; answers every prompt, sorted by name then id |
+| `POST` | `/api/schedules` | `name`, `agents`, `body`, and optional `anchor_unix`, `interval_seconds`, `paused`; answers 201 with the prompt |
+| `PATCH` | `/api/schedules/` | any of `name`, `body`, `anchor_unix`, `interval_seconds` (`null` for a one-shot), `paused`, `add_agents`, `remove_agents` |
+| `DELETE` | `/api/schedules/` | —; answers 204 |
+
+Each prompt they answer with has `id`, `name`, `agents` (sorted), `body`,
+`anchor_unix`, `interval_seconds` (`null` for a one-shot), `paused`, and
+`next_fire_unix`: the next time on the prompt's grid, as of the moment the
+controller answers, `null` when paused or a one-shot whose time has passed.
+
+An agent's own queue credential grants exactly one subject in the swarm's
+JetStream: `$JS.API.DIRECT.GET.KV_agent-schedules.$KV.agent-schedules.`,
+the direct read of its own key. It can't read another agent's key, open or
+watch the bucket, or write to it. A KV watch would need a consumer, and a
+consumer create subject doesn't name a key, so granting one would let the
+agent read every key; the agent reads its key every 10 seconds instead.
+
+The harness (`hive-agent`'s `schedule_timer`) keeps the last fired time per
+prompt id in the `schedule_fired_slots` table of its state database, and
+pushes each fire as the todo `schedule`/`schedule:` with source
+`scheduled`, reopening an acknowledged one.
+
+ `
+
+###### **Subcommands:**
+
+* `add` — Add a scheduled prompt and print it with its new id
+* `list` — List scheduled prompts, one row each, with their agents and next fire time
+* `edit` — Change a scheduled prompt
+* `pause` — Stop a scheduled prompt firing. Times that pass while it's paused aren't fired when it resumes
+* `resume` — Let a paused scheduled prompt fire again, from its next time on
+* `rm` — Remove a scheduled prompt from every agent it names
+
+
+
+## `swarmctl schedule add`
+
+Add a scheduled prompt and print it with its new id.
+
+Without `--every` it fires once. Without `--at` its first fire is now.
+
+**Usage:** `swarmctl schedule add [OPTIONS] --name --agent --body `
+
+###### **Options:**
+
+* `--name ` — The prompt's title
+* `--agent ` — Agent to fire it in. Repeatable; at least one
+* `--body ` — The prompt, delivered to each agent verbatim
+* `--at ` — First fire time: an RFC 3339 time with its offset, such as `2026-10-05T09:00:00+02:00`, or unix seconds. Defaults to now
+* `--every ` — Time between fires: a count with a unit of `s`, `m`, `h` or `d`, such as `1h` or `7d`. Leave out for a one-shot
+* `--paused` — Add it paused
+* `--controller-socket ` — swarm-controller's unix socket.
+
+ Supplied by the nix module that installs this binary, from the same `socketPath` option the daemon binds; falls back to `SWARM_CONTROLLER_SOCKET`.
+
+
+
+## `swarmctl schedule list`
+
+List scheduled prompts, one row each, with their agents and next fire time
+
+**Usage:** `swarmctl schedule list [OPTIONS]`
+
+###### **Options:**
+
+* `--agent ` — Show only the prompts that fire in this agent
+* `--controller-socket ` — swarm-controller's unix socket.
+
+ Supplied by the nix module that installs this binary, from the same `socketPath` option the daemon binds; falls back to `SWARM_CONTROLLER_SOCKET`.
+
+
+
+## `swarmctl schedule edit`
+
+Change a scheduled prompt.
+
+Only the flags given change, and the id stays. `--at` and `--every` move the fire times; `--name`, `--body` and the agent flags don't.
+
+**Usage:** `swarmctl schedule edit [OPTIONS] `
+
+###### **Arguments:**
+
+* `` — The prompt's id, as `schedule list` shows it
+
+###### **Options:**
+
+* `--name ` — New title
+* `--body ` — New prompt text
+* `--at ` — New first fire time, in the form `schedule add --at` takes
+* `--every ` — New time between fires, in the form `schedule add --every` takes
+* `--once` — Make it a one-shot
+* `--add-agent ` — Agent to start firing it in. Repeatable
+* `--remove-agent ` — Agent to stop firing it in. Repeatable. Fails if the agent isn't a target, or if no agent would remain
+* `--controller-socket ` — swarm-controller's unix socket.
+
+ Supplied by the nix module that installs this binary, from the same `socketPath` option the daemon binds; falls back to `SWARM_CONTROLLER_SOCKET`.
+
+
+
+## `swarmctl schedule pause`
+
+Stop a scheduled prompt firing. Times that pass while it's paused aren't fired when it resumes
+
+**Usage:** `swarmctl schedule pause [OPTIONS] `
+
+###### **Arguments:**
+
+* `` — The prompt's id, as `schedule list` shows it
+
+###### **Options:**
+
+* `--controller-socket ` — swarm-controller's unix socket.
+
+ Supplied by the nix module that installs this binary, from the same `socketPath` option the daemon binds; falls back to `SWARM_CONTROLLER_SOCKET`.
+
+
+
+## `swarmctl schedule resume`
+
+Let a paused scheduled prompt fire again, from its next time on
+
+**Usage:** `swarmctl schedule resume [OPTIONS] `
+
+###### **Arguments:**
+
+* `` — The prompt's id, as `schedule list` shows it
+
+###### **Options:**
+
+* `--controller-socket ` — swarm-controller's unix socket.
+
+ Supplied by the nix module that installs this binary, from the same `socketPath` option the daemon binds; falls back to `SWARM_CONTROLLER_SOCKET`.
+
+
+
+## `swarmctl schedule rm`
+
+Remove a scheduled prompt from every agent it names
+
+**Usage:** `swarmctl schedule rm [OPTIONS] `
+
+###### **Arguments:**
+
+* `` — The prompt's id, as `schedule list` shows it
+
+###### **Options:**
+
+* `--controller-socket ` — swarm-controller's unix socket.
+
+ Supplied by the nix module that installs this binary, from the same `socketPath` option the daemon binds; falls back to `SWARM_CONTROLLER_SOCKET`.
+
+
+
## `swarmctl user`
Manage subjects in the swarm's SSO provider
diff --git a/hive-agent/Cargo.toml b/hive-agent/Cargo.toml
index 98afdae2..3f5063ed 100644
--- a/hive-agent/Cargo.toml
+++ b/hive-agent/Cargo.toml
@@ -37,10 +37,10 @@ rusqlite.workspace = true
schemars.workspace = true
serde.workspace = true
serde_json.workspace = true
-# Bare: `kv`/`notices` name buckets and streams this harness opens neither end
-# of. The terminal publisher is a plain core-subject publish, so it needs the
-# connect and the payload limit and nothing from JetStream.
-swarm-queue-client.workspace = true
+# `agent-schedules` alone: the schedule ticker reads its own key with one direct
+# get. `kv`/`notices` name buckets and streams this harness opens neither end
+# of, and the terminal publisher is a plain core-subject publish.
+swarm-queue-client = { workspace = true, features = ["agent-schedules"] }
# This agent's own queue secret, read from the store under its own certificate.
swarm-secret-client.workspace = true
tokio.workspace = true
diff --git a/hive-agent/src/main.rs b/hive-agent/src/main.rs
index b1fb9ad3..f51552bb 100644
--- a/hive-agent/src/main.rs
+++ b/hive-agent/src/main.rs
@@ -25,6 +25,7 @@ mod plugins;
mod prompt;
mod reminder_timer;
mod reminders;
+mod schedule_timer;
mod serve_common;
mod state_entry_watch;
mod stats;
@@ -479,6 +480,9 @@ fn spawn_todo_socket(
// Same shape again: nudge (and feed an OTEL gauge) on a
// grown CLAUDE.md instead of a grown state dir.
tokio::spawn(claude_md_watch::run(store.clone(), todo_wake.clone()));
+ // Swarm-scheduled prompts fire as todos too, from the list the
+ // swarm keeps for this agent.
+ tokio::spawn(schedule_timer::run(store.clone(), todo_wake.clone()));
Some(store)
}
Err(e) => {
diff --git a/hive-agent/src/schedule_timer.rs b/hive-agent/src/schedule_timer.rs
new file mode 100644
index 00000000..82e283f7
--- /dev/null
+++ b/hive-agent/src/schedule_timer.rs
@@ -0,0 +1,388 @@
+//! Fires this agent's swarm-scheduled prompts, in its own container.
+//!
+//! swarm-controller keeps the agent's list under its key in the
+//! `agent-schedules` KV bucket. This task reads that key every
+//! [`POLL_INTERVAL`] with the one subject the agent's own queue credential is
+//! granted there (`swarm_queue_client::agent_schedules::read_own`), and fires
+//! each due schedule as the same `schedule`-subsystem todo, keyed
+//! `schedule:`, that hive-c0re's schedule worker pushes. The wake path
+//! is the todo store's.
+//!
+//! **Fire times come from the schedule alone**: `anchor + k * interval`
+//! ([`Schedule::latest_slot`]). Rewriting the list, or changing a body,
+//! leaves them where they were.
+//!
+//! **No duplicates.** The last slot fired for each schedule id is kept in the
+//! state db ([`FiredSlots`]), and a slot fires only when it is later than
+//! that. After downtime the latest missed slot fires once and the earlier
+//! ones not at all; a restart refires nothing already recorded. A one-shot
+//! has one slot, so it fires once however long it stays listed. Records are
+//! never dropped when a schedule leaves the list: ids are never reused, and
+//! a list that reads empty for a moment must not erase what was fired.
+//!
+//! **Paused** schedules do not fire, and each slot that passes while one is
+//! paused is recorded as if fired, so resuming does not fire it.
+//!
+//! The list itself is held in memory only: until the first successful read
+//! after a start, nothing fires. Reading needs this agent's own credential;
+//! on the hive's shared client the task logs that and stops.
+
+use std::path::Path;
+use std::sync::{Arc, Mutex};
+use std::time::Duration;
+
+use anyhow::{Context as _, Result};
+use rusqlite::{Connection, OptionalExtension as _, params};
+use swarm_queue_client::agent_schedules::{AgentSchedules, Read, Schedule, read_own};
+use tokio::sync::Notify;
+
+use crate::swarm_queue::{Connection as QueueConnection, Presented};
+use crate::todos::Todos;
+
+/// How often the key is read and the list checked for due slots. Also the
+/// longest a fire can lag its slot, and a change can take to reach the agent.
+const POLL_INTERVAL: Duration = Duration::from_secs(10);
+
+/// The todo subsystem and source hive-c0re's schedule worker uses.
+const SUBSYSTEM: &str = "schedule";
+const SOURCE: &str = "scheduled";
+
+const SCHEMA: &str = r"
+CREATE TABLE IF NOT EXISTS schedule_fired_slots (
+ schedule_id TEXT PRIMARY KEY,
+ slot_unix INTEGER NOT NULL
+);
+";
+
+/// The last slot fired, or passed while paused, per schedule id.
+pub struct FiredSlots {
+ conn: Mutex,
+}
+
+impl FiredSlots {
+ /// Open (creating if needed) the table in the db at `path`.
+ ///
+ /// # Errors
+ ///
+ /// Propagates sqlite open / schema-apply failures.
+ pub fn open(path: &Path) -> Result {
+ let conn = Connection::open(path)
+ .with_context(|| format!("open schedule slots db {}", path.display()))?;
+ conn.execute_batch(SCHEMA)
+ .context("apply schedule slots schema")?;
+ Ok(Self {
+ conn: Mutex::new(conn),
+ })
+ }
+
+ fn last(&self, id: &str) -> Result> {
+ let conn = self
+ .conn
+ .lock()
+ .unwrap_or_else(std::sync::PoisonError::into_inner);
+ Ok(conn
+ .query_row(
+ "SELECT slot_unix FROM schedule_fired_slots WHERE schedule_id = ?1",
+ params![id],
+ |row| row.get(0),
+ )
+ .optional()?)
+ }
+
+ fn record(&self, id: &str, slot: i64) -> Result<()> {
+ let conn = self
+ .conn
+ .lock()
+ .unwrap_or_else(std::sync::PoisonError::into_inner);
+ conn.execute(
+ "INSERT INTO schedule_fired_slots (schedule_id, slot_unix) VALUES (?1, ?2) \
+ ON CONFLICT(schedule_id) DO UPDATE SET slot_unix = excluded.slot_unix",
+ params![id, slot],
+ )?;
+ Ok(())
+ }
+}
+
+/// What one schedule needs at `now`.
+#[derive(Debug, PartialEq, Eq)]
+enum Step {
+ /// Fire, then record the slot.
+ Fire(i64),
+ /// Record the slot without firing: it passed while paused.
+ Skip(i64),
+ Nothing,
+}
+
+fn decide(schedule: &Schedule, last: Option, now: i64) -> Step {
+ let Some(slot) = schedule.latest_slot(now) else {
+ return Step::Nothing;
+ };
+ if last.is_some_and(|last| last >= slot) {
+ return Step::Nothing;
+ }
+ if schedule.paused {
+ Step::Skip(slot)
+ } else {
+ Step::Fire(slot)
+ }
+}
+
+/// Apply [`decide`] to every schedule in `list`. `fire` returns whether the
+/// fire landed; a slot is recorded only after it did, so a failed fire is
+/// retried on the next tick.
+fn tick(
+ schedules: &AgentSchedules,
+ slots: &FiredSlots,
+ now: i64,
+ mut fire: impl FnMut(&Schedule) -> bool,
+) {
+ for schedule in &schedules.schedules {
+ let previous = match slots.last(&schedule.id) {
+ Ok(previous) => previous,
+ Err(e) => {
+ tracing::warn!(id = %schedule.id, error = ?e, "schedule: reading the fired slot failed");
+ continue;
+ }
+ };
+ let slot = match decide(schedule, previous, now) {
+ Step::Nothing => continue,
+ Step::Skip(slot) => slot,
+ Step::Fire(slot) => {
+ if !fire(schedule) {
+ continue;
+ }
+ tracing::info!(id = %schedule.id, slot, "schedule fired");
+ slot
+ }
+ };
+ if let Err(e) = slots.record(&schedule.id, slot) {
+ tracing::warn!(id = %schedule.id, slot, error = ?e, "schedule: recording the slot failed");
+ }
+ }
+}
+
+/// Push `schedule` as the todo hive-c0re's worker pushes, and wake the turn
+/// loop when it is new or reopened.
+fn push_todo(todos: &Todos, wake: &Notify, schedule: &Schedule) -> bool {
+ let key = format!("schedule:{}", schedule.id);
+ // `true`: each fire is a new occurrence, so an acked earlier one reopens.
+ match todos.upsert(SUBSYSTEM, Some(&key), &schedule.body, Some(SOURCE), true) {
+ Ok((_, changed)) => {
+ if changed {
+ wake.notify_one();
+ }
+ true
+ }
+ Err(e) => {
+ tracing::warn!(id = %schedule.id, error = ?e, "schedule: pushing the todo failed");
+ false
+ }
+ }
+}
+
+/// Run the ticker for the harness's lifetime, if this agent has a queue
+/// credential and a label to name its key with.
+pub async fn run(todos: Arc, wake: Arc) {
+ if !crate::swarm_queue::configured() {
+ return;
+ }
+ let agent = crate::identity::label();
+ if agent.is_empty() {
+ tracing::warn!("this agent has no label; not reading its swarm schedules");
+ return;
+ }
+ let slots = match FiredSlots::open(&crate::paths::state_db()) {
+ Ok(slots) => slots,
+ Err(e) => {
+ tracing::error!(error = ?e, "open schedule slots failed — swarm schedules disabled");
+ return;
+ }
+ };
+ let Some(QueueConnection { client, presented }) = crate::swarm_queue::client().await else {
+ return;
+ };
+ if presented != Presented::Agent {
+ tracing::warn!(
+ "connected with the hive's shared client, which has no read grant on the \
+ agent-schedules bucket; swarm schedules are not read"
+ );
+ return;
+ }
+ tracing::info!(%agent, "reading swarm schedules from the swarm queue");
+ let mut list: Option = None;
+ loop {
+ refresh(&client, &agent, &mut list).await;
+ if let Some(list) = &list {
+ let now = chrono::Utc::now().timestamp();
+ tick(list, &slots, now, |s| push_todo(&todos, &wake, s));
+ }
+ tokio::time::sleep(POLL_INTERVAL).await;
+ }
+}
+
+/// Replace `list` with the key's current value. On any failure the previous
+/// list stays, so schedules keep firing through a queue outage.
+async fn refresh(client: &async_nats::Client, agent: &str, list: &mut Option) {
+ // An unconnected client does not fail a JetStream request, it hangs on it.
+ if swarm_queue_client::ensure_connected(client).is_err() {
+ return;
+ }
+ match read_own(client, agent).await {
+ Ok(Read::Current(current)) => *list = Some(current),
+ Ok(Read::NoBucket) => {
+ tracing::debug!("schedule: the agent-schedules bucket does not answer yet");
+ }
+ Err(e) => tracing::warn!(error = %e, "schedule: reading this agent's schedules failed"),
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+
+ fn schedule(id: &str, anchor: i64, interval: Option) -> Schedule {
+ Schedule {
+ id: id.to_owned(),
+ body: format!("{id} body"),
+ anchor_unix: anchor,
+ interval_seconds: interval,
+ paused: false,
+ }
+ }
+
+ fn list(schedules: Vec) -> AgentSchedules {
+ AgentSchedules { schedules }
+ }
+
+ fn slots() -> FiredSlots {
+ FiredSlots::open(Path::new(":memory:")).unwrap()
+ }
+
+ /// The bodies fired by one tick at `now`.
+ fn fired(list: &AgentSchedules, slots: &FiredSlots, now: i64) -> Vec {
+ let mut out = Vec::new();
+ tick(list, slots, now, |s| {
+ out.push(s.body.clone());
+ true
+ });
+ out
+ }
+
+ #[test]
+ fn a_body_edit_does_not_move_the_cadence() {
+ let slots = slots();
+ let mut l = list(vec![schedule("a", 1000, Some(100))]);
+ assert_eq!(fired(&l, &slots, 1000), ["a body"]);
+ "edited".clone_into(&mut l.schedules[0].body);
+ assert!(fired(&l, &slots, 1050).is_empty());
+ assert!(fired(&l, &slots, 1099).is_empty());
+ assert_eq!(fired(&l, &slots, 1100), ["edited"]);
+ }
+
+ #[test]
+ fn an_offline_gap_of_several_slots_fires_once_then_rejoins_the_grid() {
+ let slots = slots();
+ let l = list(vec![schedule("a", 1000, Some(100))]);
+ assert_eq!(fired(&l, &slots, 1000), ["a body"]);
+ // Slots 1100 through 1500 pass unseen.
+ assert_eq!(fired(&l, &slots, 1550), ["a body"]);
+ assert!(fired(&l, &slots, 1560).is_empty());
+ assert!(fired(&l, &slots, 1599).is_empty());
+ assert_eq!(fired(&l, &slots, 1600), ["a body"]);
+ }
+
+ #[test]
+ fn a_restart_after_a_fire_does_not_refire() {
+ let dir = tempfile::tempdir().unwrap();
+ let db = dir.path().join("state.sqlite");
+ let l = list(vec![schedule("a", 1000, Some(100))]);
+ let before = FiredSlots::open(&db).unwrap();
+ assert_eq!(fired(&l, &before, 1010), ["a body"]);
+ drop(before);
+ let after = FiredSlots::open(&db).unwrap();
+ assert!(fired(&l, &after, 1020).is_empty());
+ // Control: the reopened store still fires the next slot.
+ assert_eq!(fired(&l, &after, 1100), ["a body"]);
+ }
+
+ /// An agent dropped from a schedule and added back gets the same id, so
+ /// the slot it already fired stays fired.
+ #[test]
+ fn leaving_and_rejoining_a_schedule_does_not_refire() {
+ let slots = slots();
+ let l = list(vec![schedule("a", 1000, Some(100))]);
+ assert_eq!(fired(&l, &slots, 1000), ["a body"]);
+ assert!(fired(&list(Vec::new()), &slots, 1020).is_empty());
+ assert!(fired(&l, &slots, 1040).is_empty());
+ assert_eq!(fired(&l, &slots, 1100), ["a body"]);
+ }
+
+ #[test]
+ fn a_list_refresh_does_not_refire() {
+ let slots = slots();
+ let l = list(vec![
+ schedule("a", 1000, Some(100)),
+ schedule("b", 900, None),
+ ]);
+ assert_eq!(fired(&l, &slots, 1000), ["a body", "b body"]);
+ let refreshed: AgentSchedules =
+ serde_json::from_slice(&serde_json::to_vec(&l).unwrap()).unwrap();
+ assert!(fired(&refreshed, &slots, 1001).is_empty());
+ // An empty read in between erases nothing.
+ assert!(fired(&list(Vec::new()), &slots, 1002).is_empty());
+ assert!(fired(&refreshed, &slots, 1003).is_empty());
+ }
+
+ #[test]
+ fn a_one_shot_fires_once() {
+ let slots = slots();
+ let l = list(vec![schedule("once", 1000, None)]);
+ assert!(fired(&l, &slots, 999).is_empty());
+ assert_eq!(fired(&l, &slots, 1000), ["once body"]);
+ assert!(fired(&l, &slots, 1001).is_empty());
+ assert!(fired(&l, &slots, 1_000_000).is_empty());
+ }
+
+ #[test]
+ fn a_paused_schedule_does_not_fire_or_catch_up() {
+ let slots = slots();
+ let mut l = list(vec![schedule("a", 1000, Some(100))]);
+ l.schedules[0].paused = true;
+ assert!(fired(&l, &slots, 1000).is_empty());
+ assert!(fired(&l, &slots, 1250).is_empty());
+ l.schedules[0].paused = false;
+ // Slot 1200 passed while paused, so resuming does not fire it.
+ assert!(fired(&l, &slots, 1260).is_empty());
+ assert_eq!(fired(&l, &slots, 1300), ["a body"]);
+ }
+
+ #[test]
+ fn a_failed_fire_is_retried_and_not_recorded() {
+ let slots = slots();
+ let l = list(vec![schedule("a", 1000, Some(100))]);
+ tick(&l, &slots, 1000, |_| false);
+ assert_eq!(fired(&l, &slots, 1005), ["a body"]);
+ }
+
+ /// The same todo hive-c0re's worker pushes, so the wake path does not
+ /// change: subsystem `schedule`, key `schedule:`, source
+ /// `scheduled`, and a reopen once the agent acked the previous fire.
+ #[test]
+ fn a_fire_is_the_hive_c0re_schedule_todo() {
+ let todos = Todos::open(Path::new(":memory:")).unwrap();
+ let wake = Notify::new();
+ let s = schedule("abcdefgh", 1000, Some(100));
+ assert!(push_todo(&todos, &wake, &s));
+ let listed = todos.list(Some(SUBSYSTEM)).unwrap();
+ assert_eq!(listed.len(), 1);
+ assert_eq!(
+ listed[0].subsystem_key.as_deref(),
+ Some("schedule:abcdefgh")
+ );
+ assert_eq!(listed[0].summary, "abcdefgh body");
+ assert_eq!(listed[0].source.as_deref(), Some(SOURCE));
+ todos.mark_done(listed[0].id).unwrap();
+ assert!(push_todo(&todos, &wake, &s));
+ assert_eq!(todos.list(Some(SUBSYSTEM)).unwrap().len(), 1);
+ }
+}
diff --git a/nix/host-modules/swarm-nats.nix b/nix/host-modules/swarm-nats.nix
index 493848b6..abf63aaf 100644
--- a/nix/host-modules/swarm-nats.nix
+++ b/nix/host-modules/swarm-nats.nix
@@ -850,6 +850,13 @@ in
# `term-sub-{agent}` stream that keeps them, so its subjects
# and limits are never the agent's to choose.
"--agent-token-publish-subject ${lib.escapeShellArg "\$\$SWARM.term.{agent}.sub.>"}"
+ # Reading its own key of the `agent-schedules` KV bucket
+ # (`swarm_queue_client::agent_schedules`), and nothing else of
+ # JetStream: a direct get names the key in its subject, so
+ # this one subject is one key. No `STREAM.INFO`, no
+ # `CONSUMER.CREATE` (a watch's create subject carries no key,
+ # so it would cover every agent's) and no `$KV` write.
+ "--agent-token-publish-subject ${lib.escapeShellArg "\$\$JS.API.DIRECT.GET.KV_agent-schedules.\$\$KV.agent-schedules.{agent}"}"
"--store-cert-role ${lib.escapeShellArg authCertRole}"
];
# Every credential arrives by `LoadCredential` and is named
diff --git a/nix/module-eval/nats-authelia.nix b/nix/module-eval/nats-authelia.nix
index 78b30f79..be098da3 100644
--- a/nix/module-eval/nats-authelia.nix
+++ b/nix/module-eval/nats-authelia.nix
@@ -165,7 +165,7 @@ let
# The whole per-agent grant, compared as a list rather than by infix: a
# wider subject added beside these (`$$JS.API.>`, another stream's name)
# would pass every presence check above.
- name = "a verified agent's grant is exactly its own subjects, with no JetStream API subject";
+ name = "a verified agent's grant is exactly its own subjects, and its one JetStream API subject reads its own schedule key";
ok =
let
args = lib.splitString " " (responderOf natsOldPath).serviceConfig.ExecStart;
@@ -183,6 +183,7 @@ let
"'$$SWARM.agent-state.{agent}'"
"'$$KV.agent-icons.{agent}'"
"'$$SWARM.term.{agent}.sub.>'"
+ "'$$JS.API.DIRECT.GET.KV_agent-schedules.$$KV.agent-schedules.{agent}'"
];
}
{
diff --git a/swarm-controller/README.md b/swarm-controller/README.md
index 566b0a4e..876483b3 100644
--- a/swarm-controller/README.md
+++ b/swarm-controller/README.md
@@ -41,6 +41,11 @@ The swarm's control plane. The swarm UI and `swarmctl` are its clients.
webhooks (`GET /api/config-prs`, `/api/agents/{name}/config-pr`).
- **Swarm-wide forge objects and webhooks** →
[`docs/swarm/README.md`](../docs/swarm/README.md#swarm-wide-forge-objects).
+- **Scheduled prompts** — named prompts for one or more agents, kept in
+ the `schedule-definitions` KV bucket and written into each agent's key of
+ `agent-schedules`, both of which this daemon alone writes
+ (`GET`/`POST /api/schedules`, `PATCH`/`DELETE /api/schedules/{id}`); each
+ agent fires them itself → [`docs/swarm/schedules.md`](../docs/swarm/schedules.md).
- **Relays** — each agent's terminal and turn-state header as SSE, agent
icons, the cross-repo issue report, and the UI's quick links.
diff --git a/swarm-controller/src/main.rs b/swarm-controller/src/main.rs
index 48115441..a1e90b92 100644
--- a/swarm-controller/src/main.rs
+++ b/swarm-controller/src/main.rs
@@ -57,6 +57,7 @@ mod matrix_account;
mod otel_http_client;
mod queue_identity;
mod read_policy;
+mod schedules;
mod status;
mod store;
mod subagent_term;
@@ -616,6 +617,7 @@ fn socket_path() -> PathBuf {
(name = "webhook", description = "swarm-wide forge webhook receipt"),
(name = "repos", description = "forge repo browsing (swarm-ui's issue-report page)"),
(name = "forge", description = "forge site admins (`swarmctl forge make-admin`)"),
+ (name = "schedules", description = "scheduled prompts (`swarmctl schedule`)"),
)
)]
struct ApiDoc;
@@ -933,8 +935,8 @@ fn error_problem(status: axum::http::StatusCode, detail: &str) -> problem_detail
problem_details::ProblemDetails::from_status_code(status).with_detail(detail)
}
-/// The status a [`wanted::WantedWriter`] or [`agent_icon::AgentIconReader`]
-/// failure should answer with.
+/// The status a [`wanted::WantedWriter`], [`agent_icon::AgentIconReader`] or
+/// `schedules` queue failure should answer with.
///
/// `WantedWriter::view`/`set` call `ensure_connected` internally and
/// propagate through `anyhow`, so the concrete
@@ -945,7 +947,7 @@ fn error_problem(status: axum::http::StatusCode, detail: &str) -> problem_detail
/// is reconnecting, or hasn't yet) and every other queue-backed route here
/// already answers 503 for it; anything else is a genuine failure to publish
/// or read, which stays 500.
-fn wanted_error_status(e: &anyhow::Error) -> axum::http::StatusCode {
+pub(crate) fn wanted_error_status(e: &anyhow::Error) -> axum::http::StatusCode {
match e.downcast_ref::() {
Some(swarm_queue_client::Error::NotConnected(_)) => {
axum::http::StatusCode::SERVICE_UNAVAILABLE
@@ -2409,6 +2411,23 @@ fn spawn_subagent_streams(status: Option<&Arc>, hives: &[H
subagent_term::spawn(status.queue_client(), wanted, hives);
}
+/// The per-agent background passes: subagent streams and schedule keys.
+/// Lifted out of `main` for `clippy::too_many_lines`.
+fn spawn_per_agent_upkeep(status: Option<&Arc>, hives: &[HiveEntry]) {
+ spawn_subagent_streams(status, hives);
+ spawn_schedule_reconcile(status, hives);
+}
+
+/// Start rewriting every agent's schedule key from the definitions when a
+/// swarm queue is wired up. Same shape as [`spawn_subagent_streams`].
+fn spawn_schedule_reconcile(status: Option<&Arc>, hives: &[HiveEntry]) {
+ let (Some(status), Some(wanted)) = (status, wanted_writer(status)) else {
+ return;
+ };
+ let hives = hives.iter().map(|h| h.name.clone()).collect();
+ schedules::spawn_reconcile(status.queue_client(), wanted, hives);
+}
+
/// Check an agent's forge token now, and mint one if it is missing or stale.
///
/// The periodic pass (`forge::agent_token::spawn`) does the same every five
@@ -2937,7 +2956,7 @@ async fn main() -> Result<()> {
let state_forge = keep_forge_for_state(forge_client, webhook_secret.clone());
spawn_agent_renewal(&jobq, wanted_writer(status.as_ref()), &hives);
- spawn_subagent_streams(status.as_ref(), &hives);
+ spawn_per_agent_upkeep(status.as_ref(), &hives);
// Before serving, because a hive whose role does not exist cannot log in,
// and one whose policy does not exist logs in able to read nothing —
// either way it cannot collect what this daemon writes for it. A store
@@ -3016,6 +3035,11 @@ fn build_app(state: AppState) -> axum::Router {
.routes(routes!(subagent_term::stream_subagent_term))
.routes(routes!(agent_state_stream::stream_agent_state))
.routes(routes!(agent_input::post_agent_input))
+ .routes(routes!(schedules::list_schedules, schedules::add_schedule))
+ .routes(routes!(
+ schedules::edit_schedule,
+ schedules::remove_schedule
+ ))
.routes(routes!(issue_report::get_repos))
.routes(routes!(issue_report::get_issue_report_all))
.routes(routes!(issue_report::get_issue_report))
diff --git a/swarm-controller/src/schedules.rs b/swarm-controller/src/schedules.rs
new file mode 100644
index 00000000..3ac00b9f
--- /dev/null
+++ b/swarm-controller/src/schedules.rs
@@ -0,0 +1,922 @@
+//! `/api/schedules` — the operator's scheduled prompts. Each has an id, a
+//! name, a body, a cadence and the agents it goes to.
+//!
+//! **Two buckets.** Every schedule in the swarm is one document under one key
+//! ([`DEFINITIONS_KEY`] in [`DEFINITIONS_BUCKET`]), which only this daemon
+//! reads or writes. Every change is one compare-and-set on that key, so a
+//! schedule exists for all of its agents or for none of them.
+//!
+//! Each agent reads only its own key of `agent-schedules`
+//! (`swarm_queue_client::agent_schedules`), which holds the schedules naming
+//! it, without their names or other agents. That key is a projection of the
+//! definitions: after a change this daemon rewrites the key of each agent the
+//! change touched, and [`spawn_reconcile`] rewrites every agent's key every
+//! [`RECONCILE_INTERVAL`], so a projection write that failed is repaired
+//! within that interval. A projection keeps each schedule's id, which the
+//! agent keys its fired-slot record on, so dropping an agent and adding it
+//! back does not refire a slot.
+
+use std::collections::BTreeSet;
+use std::sync::Arc;
+use std::time::Duration;
+
+use anyhow::Context as _;
+use async_nats::jetstream::kv::{CreateErrorKind, Operation, Store, UpdateErrorKind};
+use axum::Json;
+use axum::extract::{Path, State};
+use axum::http::StatusCode;
+use problem_details::ProblemDetails;
+use serde::{Deserialize, Deserializer, Serialize};
+use swarm_queue_client::agent_schedules::{self, AgentSchedules, Schedule};
+use utoipa::ToSchema;
+
+use crate::wanted::WantedWriter;
+use crate::{AppState, error_problem};
+
+/// The bucket holding every schedule definition. Agents hold no grant on it.
+pub const DEFINITIONS_BUCKET: &str = "schedule-definitions";
+
+/// The one key in [`DEFINITIONS_BUCKET`].
+pub const DEFINITIONS_KEY: &str = "all";
+
+/// How many times a losing writer re-reads and re-applies, as in
+/// `wanted::MAX_ATTEMPTS`.
+const MAX_ATTEMPTS: usize = 5;
+
+/// Length of a minted schedule id.
+const ID_LEN: usize = 8;
+
+/// How often [`spawn_reconcile`] rewrites every agent's key. Also the
+/// longest a failed projection write leaves an agent behind.
+const RECONCILE_INTERVAL: Duration = Duration::from_mins(1);
+
+/// Every schedule in the swarm: the value under [`DEFINITIONS_KEY`].
+#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
+struct Definitions {
+ #[serde(default)]
+ schedules: Vec,
+}
+
+/// One schedule as the operator defines it.
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+struct Definition {
+ id: String,
+ name: String,
+ agents: BTreeSet,
+ body: String,
+ anchor_unix: i64,
+ #[serde(default, skip_serializing_if = "Option::is_none")]
+ interval_seconds: Option,
+ #[serde(default)]
+ paused: bool,
+}
+
+impl Definition {
+ /// What an agent sees of this schedule.
+ fn for_agent(&self) -> Schedule {
+ Schedule {
+ id: self.id.clone(),
+ body: self.body.clone(),
+ anchor_unix: self.anchor_unix,
+ interval_seconds: self.interval_seconds,
+ paused: self.paused,
+ }
+ }
+
+ /// The next fire time after `now`, or `None` when nothing will fire:
+ /// paused, or a one-shot whose time has passed.
+ fn next_fire(&self, now: i64) -> Option {
+ if self.paused {
+ return None;
+ }
+ if now < self.anchor_unix {
+ return Some(self.anchor_unix);
+ }
+ let interval = i64::try_from(self.interval_seconds?).ok()?;
+ Some(self.for_agent().latest_slot(now)? + interval)
+ }
+}
+
+/// `agent`'s key: the schedules that name it, in definition order.
+fn projection(definitions: &Definitions, agent: &str) -> AgentSchedules {
+ AgentSchedules {
+ schedules: definitions
+ .schedules
+ .iter()
+ .filter(|d| d.agents.contains(agent))
+ .map(Definition::for_agent)
+ .collect(),
+ }
+}
+
+/// One schedule as the API shows it.
+#[derive(Debug, Serialize, ToSchema)]
+pub struct ScheduleView {
+ id: String,
+ /// The operator's title for it.
+ name: String,
+ /// The agents it fires in, sorted.
+ agents: Vec,
+ /// The prompt, delivered to each agent verbatim.
+ body: String,
+ /// The first fire time, unix seconds.
+ anchor_unix: i64,
+ /// Seconds between fires; `null` for a one-shot.
+ interval_seconds: Option,
+ paused: bool,
+ /// The next fire time on the schedule's grid, unix seconds, computed when
+ /// the response is built; `null` when paused or a one-shot whose time
+ /// has passed. Says nothing about whether an agent fired an earlier one.
+ next_fire_unix: Option,
+}
+
+impl ScheduleView {
+ fn new(d: &Definition, now: i64) -> Self {
+ Self {
+ id: d.id.clone(),
+ name: d.name.clone(),
+ agents: d.agents.iter().cloned().collect(),
+ body: d.body.clone(),
+ anchor_unix: d.anchor_unix,
+ interval_seconds: d.interval_seconds,
+ paused: d.paused,
+ next_fire_unix: d.next_fire(now),
+ }
+ }
+}
+
+/// A new schedule.
+#[derive(Debug, Clone, Deserialize, ToSchema)]
+pub struct AddScheduleRequest {
+ /// The title. Must not be empty.
+ name: String,
+ /// The agents to fire in, at least one. Each must be declared on a hive.
+ agents: Vec,
+ /// The prompt, delivered verbatim. Must not be empty.
+ body: String,
+ /// The first fire time, unix seconds. Defaults to now.
+ anchor_unix: Option,
+ /// Seconds between fires, at least 1. Absent for a one-shot.
+ interval_seconds: Option,
+ #[serde(default)]
+ paused: bool,
+}
+
+/// A partial change to one schedule; an absent field is left as it is.
+#[derive(Debug, Clone, Default, Deserialize, ToSchema)]
+pub struct EditScheduleRequest {
+ name: Option,
+ body: Option,
+ anchor_unix: Option,
+ /// A number sets the interval; `null` makes the schedule a one-shot.
+ #[serde(default)]
+ #[schema(value_type = Option)]
+ interval_seconds: IntervalEdit,
+ paused: Option,
+ /// Agents to add. Each must be declared on a hive; one already targeted
+ /// is left as it is.
+ #[serde(default)]
+ add_agents: Vec,
+ /// Agents to remove. Each must be a current target, and at least one
+ /// target must remain.
+ #[serde(default)]
+ remove_agents: Vec,
+}
+
+/// What an edit does to the interval: an absent field keeps it, `null` makes
+/// a one-shot, a number sets it.
+#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
+enum IntervalEdit {
+ #[default]
+ Keep,
+ Set(Option),
+}
+
+impl<'de> Deserialize<'de> for IntervalEdit {
+ fn deserialize>(d: D) -> Result {
+ Option::::deserialize(d).map(Self::Set)
+ }
+}
+
+/// Why a change was refused before it was written.
+#[derive(Debug, PartialEq, Eq)]
+enum Refusal {
+ Invalid(String),
+ NotDeclared(String),
+ NoSuchSchedule(String),
+}
+
+fn invalid(why: &str) -> Refusal {
+ Refusal::Invalid(why.to_owned())
+}
+
+fn check_text(value: &str, what: &str) -> Result<(), Refusal> {
+ if value.trim().is_empty() {
+ return Err(Refusal::Invalid(format!("the {what} is empty")));
+ }
+ Ok(())
+}
+
+fn check_interval(interval: Option) -> Result<(), Refusal> {
+ if interval == Some(0) {
+ return Err(invalid(
+ "interval_seconds must be at least 1; omit it for a one-shot",
+ ));
+ }
+ Ok(())
+}
+
+/// `names` as agent identifiers, each declared on some hive per `live`.
+fn check_agents(names: &[String], live: &BTreeSet) -> Result, Refusal> {
+ let mut agents = BTreeSet::new();
+ for name in names {
+ let agent = hive_types::Ident::parse(name)
+ .map(hive_types::Ident::into_string)
+ .map_err(|why| Refusal::Invalid(format!("agent {name:?}: {why}")))?;
+ if !live.contains(&agent) {
+ return Err(Refusal::NotDeclared(agent));
+ }
+ agents.insert(agent);
+ }
+ Ok(agents)
+}
+
+impl AddScheduleRequest {
+ fn into_definition(
+ self,
+ id: String,
+ now: i64,
+ live: &BTreeSet,
+ ) -> Result {
+ check_text(&self.name, "name")?;
+ check_text(&self.body, "body")?;
+ check_interval(self.interval_seconds)?;
+ let agents = check_agents(&self.agents, live)?;
+ if agents.is_empty() {
+ return Err(invalid("a schedule needs at least one agent"));
+ }
+ Ok(Definition {
+ id,
+ name: self.name,
+ agents,
+ body: self.body,
+ anchor_unix: self.anchor_unix.unwrap_or(now),
+ interval_seconds: self.interval_seconds,
+ paused: self.paused,
+ })
+ }
+}
+
+impl EditScheduleRequest {
+ /// Apply the edit to schedule `id`, returning it as changed and every
+ /// agent it named before or after.
+ fn apply(
+ self,
+ definitions: &mut Definitions,
+ id: &str,
+ live: &BTreeSet,
+ ) -> Result<(Definition, BTreeSet), Refusal> {
+ if let Some(name) = &self.name {
+ check_text(name, "name")?;
+ }
+ if let Some(body) = &self.body {
+ check_text(body, "body")?;
+ }
+ if let IntervalEdit::Set(interval) = self.interval_seconds {
+ check_interval(interval)?;
+ }
+ let added = check_agents(&self.add_agents, live)?;
+ let d = definitions
+ .schedules
+ .iter_mut()
+ .find(|d| d.id == id)
+ .ok_or_else(|| Refusal::NoSuchSchedule(id.to_owned()))?;
+ let mut agents = d.agents.clone();
+ agents.extend(added);
+ for name in &self.remove_agents {
+ if !agents.remove(name) {
+ return Err(Refusal::Invalid(format!(
+ "agent {name:?} is not a target of schedule {id:?}"
+ )));
+ }
+ }
+ if agents.is_empty() {
+ return Err(invalid(
+ "this would leave the schedule with no agents; remove the schedule instead",
+ ));
+ }
+ let touched = d.agents.union(&agents).cloned().collect();
+ d.agents = agents;
+ if let Some(name) = self.name {
+ d.name = name;
+ }
+ if let Some(body) = self.body {
+ d.body = body;
+ }
+ if let Some(anchor) = self.anchor_unix {
+ d.anchor_unix = anchor;
+ }
+ if let IntervalEdit::Set(interval) = self.interval_seconds {
+ d.interval_seconds = interval;
+ }
+ if let Some(paused) = self.paused {
+ d.paused = paused;
+ }
+ Ok((d.clone(), touched))
+ }
+}
+
+/// Remove schedule `id`, returning the agents it named.
+fn remove(definitions: &mut Definitions, id: &str) -> Result, Refusal> {
+ let at = definitions
+ .schedules
+ .iter()
+ .position(|d| d.id == id)
+ .ok_or_else(|| Refusal::NoSuchSchedule(id.to_owned()))?;
+ Ok(definitions.schedules.remove(at).agents)
+}
+
+/// A fresh id: lowercase letters only, so the agent's todo key
+/// `schedule:` can never equal one hive-c0re pushes for its numeric ids.
+fn mint_id(taken: &Definitions) -> anyhow::Result {
+ loop {
+ let mut bytes = [0u8; ID_LEN];
+ getrandom::fill(&mut bytes)
+ .map_err(|e| anyhow::anyhow!("drawing a schedule id from the kernel's CSPRNG: {e}"))?;
+ let id: String = bytes.iter().map(|b| char::from(b'a' + b % 26)).collect();
+ if !taken.schedules.iter().any(|d| d.id == id) {
+ return Ok(id);
+ }
+ }
+}
+
+/// The failure of one request, as the problem it answers with.
+enum Failure {
+ Refused(Refusal),
+ Queue(anyhow::Error),
+}
+
+impl From for Failure {
+ fn from(e: anyhow::Error) -> Self {
+ Self::Queue(e)
+ }
+}
+
+impl From for Failure {
+ fn from(r: Refusal) -> Self {
+ Self::Refused(r)
+ }
+}
+
+fn problem(failure: Failure) -> ProblemDetails {
+ match failure {
+ Failure::Refused(Refusal::Invalid(why)) => error_problem(StatusCode::BAD_REQUEST, &why),
+ Failure::Refused(Refusal::NotDeclared(agent)) => error_problem(
+ StatusCode::BAD_REQUEST,
+ &format!("no hive's wanted state declares an agent {agent:?} that is not destroyed"),
+ ),
+ Failure::Refused(Refusal::NoSuchSchedule(id)) => {
+ error_problem(StatusCode::NOT_FOUND, &format!("no schedule {id:?}"))
+ }
+ Failure::Queue(e) => {
+ tracing::warn!(error = %format!("{e:#}"), "schedules: queue failure");
+ error_problem(crate::wanted_error_status(&e), &format!("{e:#}"))
+ }
+ }
+}
+
+/// Open a bucket only this daemon writes, creating it if nothing has yet.
+async fn open_or_create_definitions(client: &async_nats::Client) -> anyhow::Result {
+ let js = async_nats::jetstream::new(client.clone());
+ if let Ok(store) = js.get_key_value(DEFINITIONS_BUCKET).await {
+ return Ok(store);
+ }
+ js.create_key_value(async_nats::jetstream::kv::Config {
+ bucket: DEFINITIONS_BUCKET.to_owned(),
+ description: "Every scheduled prompt the swarm controller keeps".to_owned(),
+ history: 1,
+ ..Default::default()
+ })
+ .await
+ .with_context(|| format!("creating the {DEFINITIONS_BUCKET} bucket"))
+}
+
+/// The value of `key` decoded, with the revision to write it back against.
+/// An absent or deleted key is the default value.
+async fn read(
+ store: &Store,
+ key: &str,
+) -> anyhow::Result<(T, Option)> {
+ let Some(entry) = store
+ .entry(key)
+ .await
+ .with_context(|| format!("reading {key}"))?
+ else {
+ return Ok((T::default(), None));
+ };
+ if entry.operation != Operation::Put {
+ return Ok((T::default(), Some(entry.revision)));
+ }
+ let value =
+ serde_json::from_slice(&entry.value).with_context(|| format!("{key} is not decodable"))?;
+ Ok((value, Some(entry.revision)))
+}
+
+/// Write `value` to `key` against `revision`. `Ok(false)` when another write
+/// landed first.
+async fn write(
+ store: &Store,
+ key: &str,
+ value: &T,
+ revision: Option,
+) -> anyhow::Result {
+ let encoded = serde_json::to_vec(value)?;
+ match revision {
+ Some(revision) => match store.update(key, encoded.into(), revision).await {
+ Ok(_) => Ok(true),
+ Err(e) if matches!(e.kind(), UpdateErrorKind::WrongLastRevision) => Ok(false),
+ Err(e) => Err(e).with_context(|| format!("writing {key}")),
+ },
+ None => match store.create(key, encoded.into()).await {
+ Ok(_) => Ok(true),
+ Err(e) if matches!(e.kind(), CreateErrorKind::AlreadyExists) => Ok(false),
+ Err(e) => Err(e).with_context(|| format!("writing {key}")),
+ },
+ }
+}
+
+/// Apply `change` to the definitions and write them back, re-reading and
+/// re-applying when another write landed in between.
+async fn commit(
+ store: &Store,
+ mut change: impl FnMut(&mut Definitions) -> Result,
+) -> Result {
+ for _ in 0..MAX_ATTEMPTS {
+ let (mut definitions, revision) = read::(store, DEFINITIONS_KEY).await?;
+ let out = change(&mut definitions)?;
+ if write(store, DEFINITIONS_KEY, &definitions, revision).await? {
+ return Ok(out);
+ }
+ tracing::debug!("schedules: definitions write lost a race, retrying");
+ }
+ Err(
+ anyhow::anyhow!("gave up writing the schedules after {MAX_ATTEMPTS} conflicting writes")
+ .into(),
+ )
+}
+
+/// Rewrite `agent`'s key from the current definitions.
+///
+/// The agent key's revision is read before the definitions, so a write based
+/// on definitions older than another writer's fails its revision check and
+/// is redone from fresh ones.
+async fn project(definitions: &Store, agents: &Store, agent: &str) -> anyhow::Result<()> {
+ for _ in 0..MAX_ATTEMPTS {
+ let (current, revision) = read::(agents, agent).await?;
+ let (all, _) = read::(definitions, DEFINITIONS_KEY).await?;
+ let wanted = projection(&all, agent);
+ if current == wanted {
+ return Ok(());
+ }
+ if write(agents, agent, &wanted, revision).await? {
+ return Ok(());
+ }
+ }
+ anyhow::bail!(
+ "gave up writing the schedules of {agent} after {MAX_ATTEMPTS} conflicting writes"
+ )
+}
+
+/// [`project`] each of `touched`, logging a failure: the definitions are
+/// already written, and [`spawn_reconcile`] repairs the agent's key.
+async fn project_all(client: &async_nats::Client, definitions: &Store, touched: &BTreeSet) {
+ let agents = match agent_schedules::open_or_create(client).await {
+ Ok(store) => store,
+ Err(e) => {
+ tracing::warn!(
+ error = %swarm_queue_client::chain(&e),
+ "schedules: opening the agent-schedules bucket failed; the reconcile pass retries"
+ );
+ return;
+ }
+ };
+ for agent in touched {
+ if let Err(e) = project(definitions, &agents, agent).await {
+ tracing::warn!(
+ %agent,
+ error = %format!("{e:#}"),
+ "schedules: rewriting the agent's key failed; the reconcile pass retries"
+ );
+ }
+ }
+}
+
+/// Rewrite every agent's key from the definitions, now and every
+/// [`RECONCILE_INTERVAL`] after: the agents the definitions name, and every
+/// live agent, so one dropped from its last schedule is emptied too.
+pub(crate) fn spawn_reconcile(
+ client: async_nats::Client,
+ wanted: Arc,
+ hives: Vec,
+) {
+ tokio::spawn(async move {
+ let mut ticker = tokio::time::interval(RECONCILE_INTERVAL);
+ loop {
+ ticker.tick().await;
+ if swarm_queue_client::ensure_connected(&client).is_err() {
+ continue;
+ }
+ let store = match open_or_create_definitions(&client).await {
+ Ok(store) => store,
+ Err(e) => {
+ tracing::warn!(error = %format!("{e:#}"), "schedules: reconcile could not open the definitions");
+ continue;
+ }
+ };
+ let all = match read::(&store, DEFINITIONS_KEY).await {
+ Ok((all, _)) => all,
+ Err(e) => {
+ tracing::warn!(error = %format!("{e:#}"), "schedules: reconcile could not read the definitions");
+ continue;
+ }
+ };
+ let mut agents = crate::subagent_term::live_agents(&wanted, &hives).await;
+ agents.extend(all.schedules.iter().flat_map(|d| d.agents.iter().cloned()));
+ project_all(&client, &store, &agents).await;
+ }
+ });
+}
+
+/// The connected queue client, or the problem.
+fn queue(state: &AppState) -> Result {
+ let status = state.status.as_ref().ok_or_else(|| {
+ (
+ StatusCode::SERVICE_UNAVAILABLE,
+ "no swarm queue is configured on this host".to_owned(),
+ )
+ })?;
+ let client = status.queue_client();
+ swarm_queue_client::ensure_connected(&client).map_err(|e| {
+ (
+ StatusCode::SERVICE_UNAVAILABLE,
+ swarm_queue_client::chain(&e),
+ )
+ })?;
+ Ok(client)
+}
+
+fn refused((status, detail): (StatusCode, String)) -> ProblemDetails {
+ error_problem(status, &detail)
+}
+
+/// Every agent some hive declares and has not destroyed.
+async fn live(state: &AppState) -> Result, ProblemDetails> {
+ let Some(wanted) = state.wanted.as_ref() else {
+ return Err(error_problem(
+ StatusCode::SERVICE_UNAVAILABLE,
+ "no swarm queue is configured on this host",
+ ));
+ };
+ let hives: Vec = state.hives.iter().map(|h| h.name.clone()).collect();
+ Ok(crate::subagent_term::live_agents(wanted, &hives).await)
+}
+
+#[utoipa::path(
+ get,
+ path = "/api/schedules",
+ responses(
+ (status = 200, description = "every schedule, sorted by name then id; empty when \
+ there are none", body = Vec),
+ (status = 503, description = "no swarm queue is configured on this host, or it is \
+ not connected (problem+json)", body = String),
+ (status = 500, description = "the schedules could not be read (problem+json)",
+ body = String),
+ ),
+ tag = "schedules"
+)]
+pub(crate) async fn list_schedules(
+ State(state): State,
+) -> Result>, ProblemDetails> {
+ let client = queue(&state).map_err(refused)?;
+ let store = open_or_create_definitions(&client)
+ .await
+ .map_err(|e| problem(e.into()))?;
+ let (mut all, _) = read::(&store, DEFINITIONS_KEY)
+ .await
+ .map_err(|e| problem(e.into()))?;
+ all.schedules
+ .sort_by(|a, b| (&a.name, &a.id).cmp(&(&b.name, &b.id)));
+ let now = unix_now();
+ Ok(Json(
+ all.schedules
+ .iter()
+ .map(|d| ScheduleView::new(d, now))
+ .collect(),
+ ))
+}
+
+#[utoipa::path(
+ post,
+ path = "/api/schedules",
+ request_body = AddScheduleRequest,
+ responses(
+ (status = 201, description = "the schedule as stored, with its new id",
+ body = ScheduleView),
+ (status = 400, description = "the name or body is empty, the interval is 0, no \
+ agent is named, or an agent is not an identifier or not declared on any hive \
+ (problem+json)", body = String),
+ (status = 503, description = "no swarm queue is configured on this host, or it is \
+ not connected (problem+json)", body = String),
+ (status = 500, description = "the schedules could not be written (problem+json)",
+ body = String),
+ ),
+ tag = "schedules"
+)]
+pub(crate) async fn add_schedule(
+ State(state): State,
+ Json(req): Json,
+) -> Result<(StatusCode, Json), ProblemDetails> {
+ let client = queue(&state).map_err(refused)?;
+ let live = live(&state).await?;
+ let now = unix_now();
+ let store = open_or_create_definitions(&client)
+ .await
+ .map_err(|e| problem(e.into()))?;
+ let added = commit(&store, |all| {
+ let id = mint_id(all)?;
+ let d = req.clone().into_definition(id, now, &live)?;
+ all.schedules.push(d.clone());
+ Ok(d)
+ })
+ .await
+ .map_err(problem)?;
+ tracing::info!(id = %added.id, agents = ?added.agents, "added a schedule");
+ project_all(&client, &store, &added.agents).await;
+ Ok((StatusCode::CREATED, Json(ScheduleView::new(&added, now))))
+}
+
+#[utoipa::path(
+ patch,
+ path = "/api/schedules/{id}",
+ params(("id" = String, Path, description = "schedule to change")),
+ request_body = EditScheduleRequest,
+ responses(
+ (status = 200, description = "the schedule as stored", body = ScheduleView),
+ (status = 400, description = "the name or body is empty, the interval is 0, an \
+ added agent is not declared on any hive, a removed agent is not a target, or \
+ no agent would remain (problem+json)", body = String),
+ (status = 404, description = "there is no schedule with this id (problem+json)",
+ body = String),
+ (status = 503, description = "no swarm queue is configured on this host, or it is \
+ not connected (problem+json)", body = String),
+ (status = 500, description = "the schedules could not be written (problem+json)",
+ body = String),
+ ),
+ tag = "schedules"
+)]
+pub(crate) async fn edit_schedule(
+ State(state): State,
+ Path(id): Path,
+ Json(req): Json,
+) -> Result, ProblemDetails> {
+ let client = queue(&state).map_err(refused)?;
+ let live = live(&state).await?;
+ let store = open_or_create_definitions(&client)
+ .await
+ .map_err(|e| problem(e.into()))?;
+ let (edited, touched) = commit(&store, |all| Ok(req.clone().apply(all, &id, &live)?))
+ .await
+ .map_err(problem)?;
+ tracing::info!(%id, "edited a schedule");
+ project_all(&client, &store, &touched).await;
+ Ok(Json(ScheduleView::new(&edited, unix_now())))
+}
+
+#[utoipa::path(
+ delete,
+ path = "/api/schedules/{id}",
+ params(("id" = String, Path, description = "schedule to remove")),
+ responses(
+ (status = 204, description = "removed; it fires no more"),
+ (status = 404, description = "there is no schedule with this id (problem+json)",
+ body = String),
+ (status = 503, description = "no swarm queue is configured on this host, or it is \
+ not connected (problem+json)", body = String),
+ (status = 500, description = "the schedules could not be written (problem+json)",
+ body = String),
+ ),
+ tag = "schedules"
+)]
+pub(crate) async fn remove_schedule(
+ State(state): State,
+ Path(id): Path,
+) -> Result {
+ let client = queue(&state).map_err(refused)?;
+ let store = open_or_create_definitions(&client)
+ .await
+ .map_err(|e| problem(e.into()))?;
+ let touched = commit(&store, |all| Ok(remove(all, &id)?))
+ .await
+ .map_err(problem)?;
+ tracing::info!(%id, "removed a schedule");
+ project_all(&client, &store, &touched).await;
+ Ok(StatusCode::NO_CONTENT)
+}
+
+fn unix_now() -> i64 {
+ std::time::SystemTime::now()
+ .duration_since(std::time::UNIX_EPOCH)
+ .map_or(0, |d| i64::try_from(d.as_secs()).unwrap_or(i64::MAX))
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+
+ fn live() -> BTreeSet {
+ ["atlas", "iris", "scribe"].map(str::to_owned).into()
+ }
+
+ fn standup() -> Definition {
+ Definition {
+ id: "abc".to_owned(),
+ name: "standup".to_owned(),
+ agents: ["atlas", "iris"].map(str::to_owned).into(),
+ body: "post the standup".to_owned(),
+ anchor_unix: 1000,
+ interval_seconds: Some(3600),
+ paused: false,
+ }
+ }
+
+ fn all() -> Definitions {
+ Definitions {
+ schedules: vec![standup()],
+ }
+ }
+
+ fn edit(json: &str) -> EditScheduleRequest {
+ serde_json::from_str(json).expect("decodes")
+ }
+
+ /// Each agent's key carries the schedules naming it, under the same id,
+ /// and nothing of the name or the other agents.
+ #[test]
+ fn an_agents_key_holds_the_schedules_naming_it() {
+ let mut defs = all();
+ let mut other = standup();
+ "xyz".clone_into(&mut other.id);
+ other.agents = ["scribe"].map(str::to_owned).into();
+ defs.schedules.push(other);
+ let atlas = projection(&defs, "atlas");
+ assert_eq!(atlas.schedules, vec![standup().for_agent()]);
+ assert_eq!(projection(&defs, "scribe").schedules[0].id, "xyz");
+ assert!(projection(&defs, "argus").schedules.is_empty());
+ let encoded = serde_json::to_string(&atlas).unwrap();
+ assert!(!encoded.contains("\"name\""), "{encoded}");
+ assert!(!encoded.contains("iris"), "{encoded}");
+ }
+
+ /// A body or name edit leaves anchor and interval, and so the fire
+ /// times, where they were, and keeps the id.
+ #[test]
+ fn a_text_edit_leaves_the_cadence_and_id_alone() {
+ let mut defs = all();
+ let (out, touched) = edit(r#"{"body":"retro","name":"retro"}"#)
+ .apply(&mut defs, "abc", &live())
+ .unwrap();
+ assert_eq!(
+ (out.id.as_str(), out.body.as_str(), out.name.as_str()),
+ ("abc", "retro", "retro")
+ );
+ assert_eq!((out.anchor_unix, out.interval_seconds), (1000, Some(3600)));
+ assert_eq!(touched, standup().agents);
+ }
+
+ /// Adding and removing targets keeps the id, and every agent on either
+ /// side of the change is rewritten.
+ #[test]
+ fn the_agent_set_changes_without_a_new_id() {
+ let mut defs = all();
+ let (out, touched) = edit(r#"{"add_agents":["scribe"],"remove_agents":["iris"]}"#)
+ .apply(&mut defs, "abc", &live())
+ .unwrap();
+ assert_eq!(out.id, "abc");
+ assert_eq!(out.agents, ["atlas", "scribe"].map(str::to_owned).into());
+ assert_eq!(
+ touched,
+ ["atlas", "iris", "scribe"].map(str::to_owned).into()
+ );
+ assert_eq!(defs.schedules, vec![out]);
+ }
+
+ #[test]
+ fn bad_edits_are_refused_before_anything_changes() {
+ for (json, id) in [
+ (r#"{"interval_seconds":0}"#, "abc"),
+ (r#"{"body":" "}"#, "abc"),
+ (r#"{"name":""}"#, "abc"),
+ (r#"{"add_agents":["nobody"]}"#, "abc"),
+ (r#"{"remove_agents":["scribe"]}"#, "abc"),
+ (r#"{"remove_agents":["atlas","iris"]}"#, "abc"),
+ (r#"{"body":"x"}"#, "nope"),
+ ] {
+ let mut defs = all();
+ assert!(edit(json).apply(&mut defs, id, &live()).is_err(), "{json}");
+ assert_eq!(defs, all(), "{json}");
+ }
+ }
+
+ #[test]
+ fn a_null_interval_makes_a_one_shot_and_an_absent_one_keeps_it() {
+ let mut defs = all();
+ let (kept, _) = edit(r#"{"paused":true}"#)
+ .apply(&mut defs, "abc", &live())
+ .unwrap();
+ assert_eq!(kept.interval_seconds, Some(3600));
+ assert!(kept.paused);
+ let (once, _) = edit(r#"{"interval_seconds":null}"#)
+ .apply(&mut defs, "abc", &live())
+ .unwrap();
+ assert_eq!(once.interval_seconds, None);
+ }
+
+ #[test]
+ fn removing_drops_only_the_named_schedule() {
+ let mut defs = all();
+ assert_eq!(
+ remove(&mut defs, "zzz"),
+ Err(Refusal::NoSuchSchedule("zzz".to_owned()))
+ );
+ assert_eq!(defs, all());
+ assert_eq!(remove(&mut defs, "abc").unwrap(), standup().agents);
+ assert!(defs.schedules.is_empty());
+ }
+
+ #[test]
+ fn an_add_needs_a_name_and_declared_agents_and_anchors_at_now() {
+ let add = |json: &str| {
+ serde_json::from_str::(json)
+ .unwrap()
+ .into_definition("id".to_owned(), 42, &live())
+ };
+ let d = add(r#"{"name":"n","agents":["iris","atlas","iris"],"body":"hi"}"#).unwrap();
+ assert_eq!(
+ (d.anchor_unix, d.interval_seconds, d.paused),
+ (42, None, false)
+ );
+ assert_eq!(d.agents, ["atlas", "iris"].map(str::to_owned).into());
+ assert!(add(r#"{"name":" ","agents":["iris"],"body":"hi"}"#).is_err());
+ assert!(add(r#"{"name":"n","agents":[],"body":"hi"}"#).is_err());
+ assert_eq!(
+ add(r#"{"name":"n","agents":["nobody"],"body":"hi"}"#),
+ Err(Refusal::NotDeclared("nobody".to_owned()))
+ );
+ assert!(add(r#"{"name":"n","agents":["Not An Ident"],"body":"hi"}"#).is_err());
+ }
+
+ #[test]
+ fn next_fire_is_the_next_grid_slot_or_none() {
+ let mut d = standup();
+ assert_eq!(d.next_fire(999), Some(1000));
+ assert_eq!(d.next_fire(1000), Some(4600));
+ assert_eq!(d.next_fire(4601), Some(8200));
+ d.paused = true;
+ assert_eq!(d.next_fire(999), None);
+ d.paused = false;
+ d.interval_seconds = None;
+ assert_eq!(d.next_fire(999), Some(1000));
+ assert_eq!(d.next_fire(1000), None);
+ }
+
+ #[test]
+ fn a_view_shows_the_wire_shape() {
+ let v = serde_json::to_value(ScheduleView::new(&standup(), 1500)).unwrap();
+ assert_eq!(
+ v,
+ serde_json::json!({
+ "id": "abc",
+ "name": "standup",
+ "agents": ["atlas", "iris"],
+ "body": "post the standup",
+ "anchor_unix": 1000,
+ "interval_seconds": 3600,
+ "paused": false,
+ "next_fire_unix": 4600,
+ })
+ );
+ }
+
+ /// Letters only: hive-c0re keys its todos `schedule:`, and a
+ /// swarm id that could be all digits could collapse into one of them.
+ #[test]
+ fn minted_ids_are_letters_and_avoid_taken_ones() {
+ let defs = all();
+ for _ in 0..100 {
+ let id = mint_id(&defs).unwrap();
+ assert_eq!(id.len(), ID_LEN);
+ assert!(id.chars().all(|c| c.is_ascii_lowercase()), "{id}");
+ assert_ne!(id, "abc");
+ }
+ }
+}
diff --git a/swarm-controller/src/subagent_term.rs b/swarm-controller/src/subagent_term.rs
index daaf9dc5..8b19caf2 100644
--- a/swarm-controller/src/subagent_term.rs
+++ b/swarm-controller/src/subagent_term.rs
@@ -72,7 +72,7 @@ pub(crate) fn spawn(client: async_nats::Client, wanted: Arc, hives
}
/// Every agent the readable declarations of `hives` keep alive.
-async fn live_agents(wanted: &WantedWriter, hives: &[String]) -> BTreeSet {
+pub(crate) async fn live_agents(wanted: &WantedWriter, hives: &[String]) -> BTreeSet {
let mut declarations = Vec::with_capacity(hives.len());
for hive in hives {
match wanted.view(hive).await {
diff --git a/swarm-nats-auth/src/main.rs b/swarm-nats-auth/src/main.rs
index 7a2ddb99..4e804931 100644
--- a/swarm-nats-auth/src/main.rs
+++ b/swarm-nats-auth/src/main.rs
@@ -363,12 +363,14 @@ mod tests {
"$KV.agent-icons.{agent}",
"--agent-token-publish-subject",
"$SWARM.term.{agent}.sub.>",
+ "--agent-token-publish-subject",
+ "$JS.API.DIRECT.GET.KV_agent-schedules.$KV.agent-schedules.{agent}",
"--store-cert-role",
"swarm-nats-auth",
])
.expect("the unit's own argument vector must parse");
assert_eq!(args.agent_client_suffix, "-agent");
- assert_eq!(args.agent_token_publish_subjects.len(), 4);
+ assert_eq!(args.agent_token_publish_subjects.len(), 5);
}
/// The control for the case above: an ordinary value parses through the
diff --git a/swarm-nats-auth/src/policy.rs b/swarm-nats-auth/src/policy.rs
index dbface27..15a32251 100644
--- a/swarm-nats-auth/src/policy.rs
+++ b/swarm-nats-auth/src/policy.rs
@@ -1290,6 +1290,77 @@ mod tests {
);
}
+ /// The schedule template as `swarm-nats.nix` spells it reads one key: the
+ /// agent's own. It covers no other agent's key, no other stream or bucket,
+ /// and no way to open, watch or write the bucket. The hive's shared agent
+ /// client gets none of it.
+ #[test]
+ fn an_agent_may_read_only_its_own_schedule_key() {
+ use swarm_queue_client::agent_schedules::{
+ BUCKET, STREAM, direct_get_subject, key_subject,
+ };
+
+ let p = policy_with_agent_subject()
+ .with_agent_token_subjects(vec![
+ "$JS.API.DIRECT.GET.KV_agent-schedules.$KV.agent-schedules.{agent}".to_owned(),
+ ])
+ .expect("a per-agent template is valid");
+ let atlas = p.agent_token_permissions("atlas").expect("configured");
+ assert_eq!(atlas.publish, vec![direct_get_subject("atlas")]);
+ // Control: argus's own expansion is a different subject, so atlas
+ // lacking it is the grant being scoped, not the template being empty.
+ let argus = p.agent_token_permissions("argus").expect("configured");
+ assert_eq!(argus.publish, vec![direct_get_subject("argus")]);
+ assert!(!atlas.publish.contains(&direct_get_subject("argus")));
+
+ let covers = |grant: &str, subject: &str| {
+ let g: Vec<&str> = grant.split('.').collect();
+ let s: Vec<&str> = subject.split('.').collect();
+ for (i, t) in g.iter().enumerate() {
+ match *t {
+ ">" => return s.len() > i,
+ "*" if i < s.len() => {}
+ t if s.get(i) == Some(&t) => {}
+ _ => return false,
+ }
+ }
+ g.len() == s.len()
+ };
+ for refused in [
+ direct_get_subject("argus"),
+ format!("$JS.API.DIRECT.GET.{STREAM}"),
+ format!("$JS.API.STREAM.INFO.{STREAM}"),
+ format!("$JS.API.STREAM.CREATE.{STREAM}"),
+ format!("$JS.API.STREAM.DELETE.{STREAM}"),
+ format!("$JS.API.STREAM.MSG.GET.{STREAM}"),
+ format!("$JS.API.CONSUMER.CREATE.{STREAM}"),
+ format!(
+ "$JS.API.CONSUMER.CREATE.{STREAM}.w.{}",
+ key_subject("atlas")
+ ),
+ key_subject("atlas"),
+ key_subject("argus"),
+ "$JS.API.DIRECT.GET.KV_agent-icons.$KV.agent-icons.atlas".to_owned(),
+ "$JS.API.DIRECT.GET.KV_hive-wanted-alpha.$KV.hive-wanted-alpha.alpha".to_owned(),
+ "$JS.API.INFO".to_owned(),
+ ] {
+ assert!(
+ !atlas.publish.iter().any(|g| covers(g, &refused)),
+ "atlas's grant covers {refused}"
+ );
+ }
+ // Control: the matcher does match, so the refusals above are not
+ // vacuous.
+ assert!(covers(&atlas.publish[0], &direct_get_subject("atlas")));
+ assert!(covers("$KV.*.>", &key_subject("argus")));
+
+ let shared = p.permissions("hive-alpha-agent").expect("alpha's agents");
+ assert!(
+ !shared.publish.iter().any(|s| s.contains(BUCKET)),
+ "the hive's shared agent client got a schedule subject: {shared:?}"
+ );
+ }
+
#[test]
fn an_agent_token_subject_without_the_placeholder_is_refused() {
let err = policy()
diff --git a/swarm-queue-client/Cargo.toml b/swarm-queue-client/Cargo.toml
index 7caec0bb..1fac3713 100644
--- a/swarm-queue-client/Cargo.toml
+++ b/swarm-queue-client/Cargo.toml
@@ -33,6 +33,10 @@ notices = ["async-nats/jetstream"]
# `subagent_term::open_or_create`, for swarm-controller, which creates every
# agent's subagent stream. Explicit for the same reason as `notices`.
subagent-term = ["async-nats/jetstream"]
+# `agent_schedules::read_own`, for the agent harness, which reads its own
+# schedule key without opening the bucket. Explicit for the same reason as
+# `notices`.
+agent-schedules = ["async-nats/jetstream"]
[dependencies]
# Bare (no `kv`/`jetstream`) unless a consumer opts into the `kv` feature
diff --git a/swarm-queue-client/src/agent_schedules.rs b/swarm-queue-client/src/agent_schedules.rs
new file mode 100644
index 00000000..f3a26677
--- /dev/null
+++ b/swarm-queue-client/src/agent_schedules.rs
@@ -0,0 +1,234 @@
+//! The per-agent schedule KV bucket: one key per agent, holding the scheduled
+//! prompts that fire in that agent.
+//!
+//! **swarm-controller is the only writer**, and creates the bucket on its
+//! first write. It keeps each prompt once, with its name and every agent it
+//! goes to, and writes each agent's key as the prompts naming that agent,
+//! without the name or the other agents. Each agent's harness reads its own
+//! key and fires the schedules itself; the fire times are a pure function of
+//! each schedule ([`crate::agent_schedules::Schedule::latest_slot`]), so
+//! rewriting the list never moves them.
+//!
+//! **An agent reads with one `DIRECT.GET` on its own key, not a watch.** A KV
+//! watch is an unnamed consumer, whose create subject carries no filter, so
+//! granting it would grant every agent's key. The direct get carries the key
+//! in its subject ([`crate::agent_schedules::direct_get_subject`]), which is
+//! what lets the queue grant scope the read to one agent. It is also the
+//! agent's whole `JetStream` grant: `read_own` builds the stream handle
+//! without a `STREAM.INFO`.
+//!
+//! Keyed by the agent alone, like [`crate::agent_icon`], and for the same
+//! reason: an agent can move between hives.
+
+/// The KV bucket schedules are published into, keyed by agent name.
+pub const BUCKET: &str = "agent-schedules";
+
+/// The stream backing [`BUCKET`]. `create_key_value` prefixes `KV_`.
+pub const STREAM: &str = "KV_agent-schedules";
+
+/// The subject `agent`'s entry is stored under.
+#[must_use]
+pub fn key_subject(agent: &str) -> String {
+ format!("$KV.{BUCKET}.{agent}")
+}
+
+/// The request subject that reads `agent`'s current entry, and so the one
+/// subject the queue grants that agent in this bucket.
+#[must_use]
+pub fn direct_get_subject(agent: &str) -> String {
+ format!("$JS.API.DIRECT.GET.{STREAM}.{}", key_subject(agent))
+}
+
+/// The prompts that fire in one agent: the value under `BUCKET`/``.
+#[derive(Debug, Clone, Default, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
+pub struct AgentSchedules {
+ #[serde(default)]
+ pub schedules: Vec,
+}
+
+/// One scheduled prompt.
+///
+/// Its fire times are `anchor_unix + k * interval_seconds` for `k >= 0`, or
+/// `anchor_unix` alone when there is no interval.
+#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
+pub struct Schedule {
+ /// Minted by the controller, the same in every agent the prompt names,
+ /// and never reused, so an agent's record of what it last fired for an
+ /// id cannot apply to a different schedule.
+ pub id: String,
+ /// The prompt, delivered verbatim as the todo's summary.
+ pub body: String,
+ /// The first fire time, unix seconds.
+ pub anchor_unix: i64,
+ /// Seconds between fires; `None` is a one-shot.
+ #[serde(default, skip_serializing_if = "Option::is_none")]
+ pub interval_seconds: Option,
+ /// A paused schedule does not fire, and the slots that pass while it is
+ /// paused are not fired when it resumes.
+ #[serde(default)]
+ pub paused: bool,
+}
+
+impl Schedule {
+ /// The latest fire time at or before `now`, or `None` while the first is
+ /// still ahead.
+ #[must_use]
+ pub fn latest_slot(&self, now: i64) -> Option {
+ if now < self.anchor_unix {
+ return None;
+ }
+ let Some(interval) = self
+ .interval_seconds
+ .filter(|i| *i > 0)
+ .and_then(|i| i64::try_from(i).ok())
+ else {
+ return Some(self.anchor_unix);
+ };
+ let elapsed = now - self.anchor_unix;
+ Some(self.anchor_unix + elapsed - elapsed % interval)
+ }
+}
+
+/// Open the schedule bucket for writing, creating it if nothing has yet.
+///
+/// **Controller-side only.** `history: 1`: an agent wants its current list.
+#[cfg(feature = "kv")]
+pub async fn open_or_create(
+ client: &async_nats::Client,
+) -> Result {
+ let js = async_nats::jetstream::new(client.clone());
+ match js.get_key_value(BUCKET).await {
+ Ok(store) => Ok(store),
+ Err(e) => {
+ tracing::info!(
+ bucket = BUCKET,
+ reason = %e,
+ "agent-schedule bucket not available, creating it"
+ );
+ js.create_key_value(async_nats::jetstream::kv::Config {
+ bucket: BUCKET.to_owned(),
+ description: "Scheduled prompts the swarm controller declares for each agent"
+ .to_owned(),
+ history: 1,
+ ..Default::default()
+ })
+ .await
+ .map_err(|source| crate::Error::CreateBucket {
+ bucket: BUCKET.to_owned(),
+ source,
+ })
+ }
+ }
+}
+
+/// What reading one agent's key found.
+#[cfg(feature = "agent-schedules")]
+#[derive(Debug)]
+pub enum Read {
+ /// The bucket answered; an absent or deleted key is an empty list.
+ Current(AgentSchedules),
+ /// Nothing answers for the bucket's stream: no schedule has been written
+ /// yet, or `JetStream` is not serving it right now. Not an empty list,
+ /// since the second case says nothing about the agent's schedules.
+ NoBucket,
+}
+
+/// Read `agent`'s current list with the one subject its grant covers.
+///
+/// # Errors
+///
+/// A failed request or an undecodable value. A request the queue refused for
+/// permissions reaches the client as a timeout, not as a refusal.
+#[cfg(feature = "agent-schedules")]
+pub async fn read_own(
+ client: &async_nats::Client,
+ agent: &str,
+) -> Result> {
+ use async_nats::jetstream::stream::DirectGetErrorKind;
+
+ let js = async_nats::jetstream::new(client.clone());
+ let stream = js.get_stream_no_info(STREAM).await?;
+ let message = match stream.direct_get_last_for_subject(key_subject(agent)).await {
+ Ok(message) => message,
+ Err(e) if matches!(e.kind(), DirectGetErrorKind::NotFound) => {
+ return Ok(Read::Current(AgentSchedules::default()));
+ }
+ Err(e) if is_no_responders(&e) => return Ok(Read::NoBucket),
+ Err(e) => return Err(e.into()),
+ };
+ // A KV delete or purge is stored as an empty marker carrying this header.
+ let removed = message
+ .headers
+ .get("KV-Operation")
+ .is_some_and(|op| matches!(op.as_str(), "DEL" | "PURGE"));
+ if removed {
+ return Ok(Read::Current(AgentSchedules::default()));
+ }
+ Ok(Read::Current(serde_json::from_slice(&message.payload)?))
+}
+
+#[cfg(feature = "agent-schedules")]
+fn is_no_responders(e: &async_nats::jetstream::stream::DirectGetError) -> bool {
+ matches!(
+ e.kind(),
+ async_nats::jetstream::stream::DirectGetErrorKind::ErrorResponse(status, _)
+ if status == async_nats::StatusCode::NO_RESPONDERS
+ )
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+
+ fn every(anchor_unix: i64, interval: Option) -> Schedule {
+ Schedule {
+ id: "abc".to_owned(),
+ body: "hi".to_owned(),
+ anchor_unix,
+ interval_seconds: interval,
+ paused: false,
+ }
+ }
+
+ /// The names the grant in `swarm-nats.nix` spells with `{agent}`.
+ #[test]
+ fn names_match_the_queue_grant() {
+ assert_eq!(STREAM, format!("KV_{BUCKET}"));
+ assert_eq!(key_subject("iris"), "$KV.agent-schedules.iris");
+ assert_eq!(
+ direct_get_subject("iris"),
+ "$JS.API.DIRECT.GET.KV_agent-schedules.$KV.agent-schedules.iris"
+ );
+ }
+
+ #[test]
+ fn slots_sit_on_the_anchor_grid() {
+ let s = every(1000, Some(60));
+ assert_eq!(s.latest_slot(999), None);
+ assert_eq!(s.latest_slot(1000), Some(1000));
+ assert_eq!(s.latest_slot(1059), Some(1000));
+ assert_eq!(s.latest_slot(1060), Some(1060));
+ assert_eq!(s.latest_slot(1000 + 60 * 7 + 5), Some(1000 + 60 * 7));
+ }
+
+ #[test]
+ fn a_one_shot_has_one_slot() {
+ let s = every(1000, None);
+ assert_eq!(s.latest_slot(999), None);
+ assert_eq!(s.latest_slot(1000), Some(1000));
+ assert_eq!(s.latest_slot(1_000_000), Some(1000));
+ }
+
+ /// Fields a newer controller adds must not cost an older agent its list,
+ /// and a list written without the optional fields decodes to a recurring
+ /// or one-shot schedule that is not paused.
+ #[test]
+ fn decoding_tolerates_missing_and_unknown_fields() {
+ let doc = r#"{"schedules":[{"id":"a","body":"b","anchor_unix":5,"note":"x"}],"rev":2}"#;
+ let decoded: AgentSchedules = serde_json::from_str(doc).expect("decodes");
+ let mut want = every(5, None);
+ "a".clone_into(&mut want.id);
+ "b".clone_into(&mut want.body);
+ assert_eq!(decoded.schedules, vec![want]);
+ }
+}
diff --git a/swarm-queue-client/src/lib.rs b/swarm-queue-client/src/lib.rs
index b3483509..c66c8393 100644
--- a/swarm-queue-client/src/lib.rs
+++ b/swarm-queue-client/src/lib.rs
@@ -196,6 +196,11 @@ pub mod agent_token;
/// placement stays out of the key, and for who may write it.
pub mod agent_icon;
+/// 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`.
+pub mod agent_schedules;
+
/// The subject the swarm controller publishes on when the hive-wide knowledge
/// repository has changed. One writer, many readers — every hive subscribes.
///
diff --git a/swarmctl/Cargo.toml b/swarmctl/Cargo.toml
index 5a85f2a7..25098920 100644
--- a/swarmctl/Cargo.toml
+++ b/swarmctl/Cargo.toml
@@ -10,6 +10,7 @@ path = "src/main.rs"
[dependencies]
anyhow.workspace = true
+chrono.workspace = true
# `agent create` — bare HTTP/1.1 client over swarm-controller's unix
# socket. No new workspace dependency: `bytes`, `http-body-util`,
# `hyper` and `hyper-util` are all already pinned in the root manifest.
diff --git a/swarmctl/README.md b/swarmctl/README.md
index 4771c024..5907ce03 100644
--- a/swarmctl/README.md
+++ b/swarmctl/README.md
@@ -153,3 +153,19 @@ person's on their first login through authelia, and until then this fails
saying so. Running it on a site admin changes nothing, and an agent's name is
refused. The response shape is mirrored in `src/forge.rs`, for the same
reason as `agent create`'s.
+
+## `schedule`
+
+```console
+# swarmctl schedule add --name standup --agent scribe --body "post the standup summary" --every 1d
+# swarmctl schedule list --agent scribe
+# swarmctl schedule edit kqzpmwab --at 2026-10-05T09:00:00+02:00 --add-agent iris
+# swarmctl schedule pause kqzpmwab
+# swarmctl schedule rm kqzpmwab
+```
+
+The `/api/schedules` routes on the swarm-controller, over the
+same socket as `agent create`; each verb waits for the stored result. What a
+schedule does once stored is in
+[`docs/swarm/schedules.md`](../docs/swarm/schedules.md). The shapes are
+mirrored in `src/schedule.rs`, for the same reason as `agent create`'s.
diff --git a/swarmctl/src/agent.rs b/swarmctl/src/agent.rs
index 5cada5d1..700ac7e1 100644
--- a/swarmctl/src/agent.rs
+++ b/swarmctl/src/agent.rs
@@ -184,8 +184,23 @@ pub(crate) async fn post(
request: &Req,
what: &str,
) -> Result {
- let body =
- serde_json::to_vec(request).with_context(|| format!("serialising the {what} request"))?;
+ send(socket, hyper::Method::POST, uri, Some(request), what).await
+}
+
+/// One request to `uri` over the controller's unix socket, with `request` as
+/// its JSON body when there is one. A `204` decodes as JSON `null`, so a
+/// caller expecting no body asks for `()`.
+pub(crate) async fn send(
+ socket: &Path,
+ method: hyper::Method,
+ uri: &str,
+ request: Option<&Req>,
+ what: &str,
+) -> Result {
+ let body = request
+ .map(serde_json::to_vec)
+ .transpose()
+ .with_context(|| format!("serialising the {what} request"))?;
let stream = UnixStream::connect(socket)
.await
@@ -193,19 +208,22 @@ pub(crate) async fn post(
let (mut sender, conn) = hyper::client::conn::http1::handshake(TokioIo::new(stream))
.await
.with_context(|| format!("HTTP handshake with {}", socket.display()))?;
- // Drives the connection's I/O; a one-shot POST has nothing to do with
+ // Drives the connection's I/O; a one-shot request has nothing to do with
// the join handle, and a connection that dies surfaces as an error on
// `send_request` or on reading the body.
tokio::spawn(conn);
- let req = hyper::Request::builder()
- .method(hyper::Method::POST)
+ let mut req = hyper::Request::builder()
+ .method(method)
.uri(uri)
// A unix socket has no authority of its own, but HTTP/1.1 requires
// the header; the controller routes on the path alone.
- .header(hyper::header::HOST, "localhost")
- .header(hyper::header::CONTENT_TYPE, "application/json")
- .body(Full::new(bytes::Bytes::from(body)))
+ .header(hyper::header::HOST, "localhost");
+ if body.is_some() {
+ req = req.header(hyper::header::CONTENT_TYPE, "application/json");
+ }
+ let req = req
+ .body(Full::new(bytes::Bytes::from(body.unwrap_or_default())))
.with_context(|| format!("building the {what} request"))?;
let resp = sender
@@ -223,6 +241,11 @@ pub(crate) async fn post(
if !status.is_success() {
bail!("{}", describe_error(status, &body));
}
+ let body = if status == hyper::StatusCode::NO_CONTENT {
+ bytes::Bytes::from_static(b"null")
+ } else {
+ body
+ };
serde_json::from_slice(&body).with_context(|| {
format!(
"swarm-controller answered {status} but its body is not a {what} response: {}",
diff --git a/swarmctl/src/main.rs b/swarmctl/src/main.rs
index 9e8d89d7..1afa1339 100644
--- a/swarmctl/src/main.rs
+++ b/swarmctl/src/main.rs
@@ -29,6 +29,7 @@
mod agent;
mod forge;
+mod schedule;
mod users;
use std::fs::{self, File, Permissions};
@@ -119,6 +120,16 @@ enum Verb {
#[command(subcommand)]
command: AgentVerb,
},
+ /// Manage scheduled prompts.
+ ///
+ /// The swarm keeps each prompt, and every agent it names fires it in its
+ /// own container. Fire times are the schedule's start time plus whole
+ /// intervals: changing the body never moves them, and an agent that was
+ /// down fires the latest missed one once.
+ Schedule {
+ #[command(subcommand)]
+ command: ScheduleVerb,
+ },
/// Manage subjects in the swarm's SSO provider.
User {
#[command(subcommand)]
@@ -276,6 +287,119 @@ struct ForgeMakeAdminArgs {
controller_socket: Option,
}
+#[derive(Subcommand)]
+enum ScheduleVerb {
+ /// Add a scheduled prompt and print it with its new id.
+ ///
+ /// Without `--every` it fires once. Without `--at` its first fire is now.
+ Add(ScheduleAddArgs),
+ /// List scheduled prompts, one row each, with their agents and next fire
+ /// time.
+ List(ScheduleListArgs),
+ /// Change a scheduled prompt.
+ ///
+ /// Only the flags given change, and the id stays. `--at` and `--every`
+ /// move the fire times; `--name`, `--body` and the agent flags don't.
+ Edit(ScheduleEditArgs),
+ /// Stop a scheduled prompt firing. Times that pass while it's paused
+ /// aren't fired when it resumes.
+ Pause(ScheduleIdArgs),
+ /// Let a paused scheduled prompt fire again, from its next time on.
+ Resume(ScheduleIdArgs),
+ /// Remove a scheduled prompt from every agent it names.
+ Rm(ScheduleIdArgs),
+}
+
+#[derive(Args)]
+struct ScheduleAddArgs {
+ /// The prompt's title.
+ #[arg(long, value_name = "TEXT")]
+ name: String,
+ /// Agent to fire it in. Repeatable; at least one.
+ #[arg(long = "agent", value_name = "AGENT", required = true)]
+ agents: Vec,
+ /// The prompt, delivered to each agent verbatim.
+ #[arg(long, value_name = "TEXT")]
+ body: String,
+ /// First fire time: an RFC 3339 time with its offset, such as
+ /// `2026-10-05T09:00:00+02:00`, or unix seconds. Defaults to now.
+ #[arg(long, value_name = "TIME")]
+ at: Option,
+ /// Time between fires: a count with a unit of `s`, `m`, `h` or `d`,
+ /// such as `1h` or `7d`. Leave out for a one-shot.
+ #[arg(long, value_name = "DURATION")]
+ every: Option,
+ /// Add it paused.
+ #[arg(long)]
+ paused: bool,
+ #[command(flatten)]
+ socket: ControllerSocketArg,
+}
+
+#[derive(Args)]
+struct ScheduleListArgs {
+ /// Show only the prompts that fire in this agent.
+ #[arg(long, value_name = "AGENT")]
+ agent: Option,
+ #[command(flatten)]
+ socket: ControllerSocketArg,
+}
+
+#[derive(Args)]
+struct ScheduleEditArgs {
+ /// The prompt's id, as `schedule list` shows it.
+ id: String,
+ /// New title.
+ #[arg(long, value_name = "TEXT")]
+ name: Option,
+ /// New prompt text.
+ #[arg(long, value_name = "TEXT")]
+ body: Option,
+ /// New first fire time, in the form `schedule add --at` takes.
+ #[arg(long, value_name = "TIME")]
+ at: Option,
+ /// New time between fires, in the form `schedule add --every` takes.
+ #[arg(long, value_name = "DURATION", conflicts_with = "once")]
+ every: Option,
+ /// Make it a one-shot.
+ #[arg(long)]
+ once: bool,
+ /// Agent to start firing it in. Repeatable.
+ #[arg(long = "add-agent", value_name = "AGENT")]
+ add_agents: Vec,
+ /// Agent to stop firing it in. Repeatable. Fails if the agent isn't a
+ /// target, or if no agent would remain.
+ #[arg(long = "remove-agent", value_name = "AGENT")]
+ remove_agents: Vec,
+ #[command(flatten)]
+ socket: ControllerSocketArg,
+}
+
+#[derive(Args)]
+struct ScheduleIdArgs {
+ /// The prompt's id, as `schedule list` shows it.
+ id: String,
+ #[command(flatten)]
+ socket: ControllerSocketArg,
+}
+
+#[derive(Args)]
+struct ControllerSocketArg {
+ /// 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`.
+ #[arg(long, value_name = "PATH")]
+ controller_socket: Option,
+}
+
+impl ControllerSocketArg {
+ fn resolve(self) -> Result {
+ path_from(self.controller_socket, "SWARM_CONTROLLER_SOCKET")
+ }
+}
+
#[derive(Subcommand)]
enum UserVerb {
/// Add a user, generating a password for them.
@@ -379,6 +503,8 @@ fn main() -> Result<()> {
let socket = path_from(args.controller_socket, "SWARM_CONTROLLER_SOCKET")?;
forge::make_admin(&socket, &args.name)
}
+ // Same socket-resolution reasoning as `Create` above.
+ Verb::Schedule { command } => schedule(command),
// Resolved lazily, inside the one arm that actually touches the
// deployment env vars — see the `MarkdownDocs` doc comment above
// for why an unconditional resolve up front would be wrong.
@@ -418,6 +544,38 @@ fn main() -> Result<()> {
}
}
+fn schedule(command: ScheduleVerb) -> Result<()> {
+ match command {
+ ScheduleVerb::Add(a) => schedule::add(
+ &a.socket.resolve()?,
+ &schedule::AddFlags {
+ name: &a.name,
+ agents: &a.agents,
+ body: &a.body,
+ at: a.at.as_deref(),
+ every: a.every.as_deref(),
+ paused: a.paused,
+ },
+ ),
+ ScheduleVerb::List(a) => schedule::list(&a.socket.resolve()?, a.agent.as_deref()),
+ ScheduleVerb::Edit(a) => {
+ let edit = schedule::EditRequest::new(schedule::EditFlags {
+ name: a.name,
+ body: a.body,
+ at: a.at.as_deref(),
+ every: a.every.as_deref(),
+ once: a.once,
+ add_agents: a.add_agents,
+ remove_agents: a.remove_agents,
+ })?;
+ schedule::edit(&a.socket.resolve()?, &a.id, &edit)
+ }
+ ScheduleVerb::Pause(a) => schedule::set_paused(&a.socket.resolve()?, &a.id, true),
+ ScheduleVerb::Resume(a) => schedule::set_paused(&a.socket.resolve()?, &a.id, false),
+ ScheduleVerb::Rm(a) => schedule::remove(&a.socket.resolve()?, &a.id),
+ }
+}
+
fn user_add(paths: &Paths, args: AddArgs) -> Result<()> {
users::validate_username(&args.username)?;
@@ -713,6 +871,23 @@ mod tests {
Cli::command().debug_assert();
}
+ /// A schedule with no agent would fire nowhere, so `--agent` is required
+ /// at parse time, and repeats collect into the target list.
+ #[test]
+ fn schedule_add_takes_one_or_more_agents() {
+ let base = ["swarmctl", "schedule", "add", "--name", "n", "--body", "b"];
+ assert!(Cli::try_parse_from(base).is_err());
+ let cli = Cli::try_parse_from(base.iter().copied().chain(["--agent", "a", "--agent", "b"]))
+ .expect("parses");
+ let Verb::Schedule {
+ command: ScheduleVerb::Add(args),
+ } = cli.command
+ else {
+ panic!("parsed as another verb");
+ };
+ assert_eq!(args.agents, ["a", "b"]);
+ }
+
/// `--hive` carries the address the agent is deployed to and the
/// endpoint has no default for it, so omitting it has to fail at parse
/// time rather than reach the controller as an empty string.
diff --git a/swarmctl/src/schedule.rs b/swarmctl/src/schedule.rs
new file mode 100644
index 00000000..b7becc07
--- /dev/null
+++ b/swarmctl/src/schedule.rs
@@ -0,0 +1,416 @@
+//! `swarmctl schedule` — the operator's scheduled prompts, through
+//! swarm-controller's `/api/schedules` routes.
+//!
+//! Each verb waits for the controller's answer, which is the list or the
+//! schedule as stored. Each agent the schedule names picks a change up on its
+//! next read of its schedule key and fires in its own container.
+//!
+//! The shapes below mirror the controller's own `ScheduleView` and request
+//! bodies, for the reason [`crate::agent`]'s module doc gives.
+
+use std::path::Path;
+
+use anyhow::{Context as _, Result, bail};
+use serde::{Deserialize, Serialize};
+
+use crate::agent::{parse_ident, send};
+
+/// One schedule as the controller shows it.
+#[derive(Debug, Deserialize)]
+struct ScheduleView {
+ id: String,
+ name: String,
+ agents: Vec,
+ body: String,
+ anchor_unix: i64,
+ interval_seconds: Option,
+ paused: bool,
+ next_fire_unix: Option,
+}
+
+/// Body of `POST /api/schedules`.
+#[derive(Serialize)]
+struct AddRequest<'a> {
+ name: &'a str,
+ agents: Vec,
+ body: &'a str,
+ #[serde(skip_serializing_if = "Option::is_none")]
+ anchor_unix: Option,
+ #[serde(skip_serializing_if = "Option::is_none")]
+ interval_seconds: Option,
+ paused: bool,
+}
+
+/// Body of `PATCH /api/schedules/{id}`. An absent field is left as it is.
+#[derive(Default, Serialize)]
+pub(crate) struct EditRequest {
+ #[serde(skip_serializing_if = "Option::is_none")]
+ name: Option,
+ #[serde(skip_serializing_if = "Option::is_none")]
+ body: Option,
+ #[serde(skip_serializing_if = "Option::is_none")]
+ anchor_unix: Option,
+ #[serde(skip_serializing_if = "IntervalEdit::is_keep")]
+ interval_seconds: IntervalEdit,
+ #[serde(skip_serializing_if = "Option::is_none")]
+ paused: Option,
+ #[serde(skip_serializing_if = "Vec::is_empty")]
+ add_agents: Vec,
+ #[serde(skip_serializing_if = "Vec::is_empty")]
+ remove_agents: Vec,
+}
+
+/// What an edit does to the interval. `Set(None)` is sent as `null`, which
+/// makes a one-shot; `Keep` is not sent at all.
+#[derive(Clone, Copy, Default)]
+enum IntervalEdit {
+ #[default]
+ Keep,
+ Set(Option),
+}
+
+impl IntervalEdit {
+ fn is_keep(&self) -> bool {
+ matches!(self, Self::Keep)
+ }
+}
+
+impl Serialize for IntervalEdit {
+ fn serialize(&self, s: S) -> Result {
+ match self {
+ Self::Keep | Self::Set(None) => s.serialize_none(),
+ Self::Set(Some(seconds)) => s.serialize_u64(*seconds),
+ }
+ }
+}
+
+/// What `schedule edit` was asked to change.
+pub(crate) struct EditFlags<'a> {
+ pub name: Option,
+ pub body: Option,
+ pub at: Option<&'a str>,
+ pub every: Option<&'a str>,
+ pub once: bool,
+ pub add_agents: Vec,
+ pub remove_agents: Vec,
+}
+
+impl EditRequest {
+ /// The edit `schedule edit` asks for. `once` and `every` are exclusive
+ /// in the CLI, so at most one of them is set here.
+ pub(crate) fn new(flags: EditFlags<'_>) -> Result {
+ let interval_seconds = match (flags.every, flags.once) {
+ (Some(every), _) => IntervalEdit::Set(Some(parse_every(every)?)),
+ (None, true) => IntervalEdit::Set(None),
+ (None, false) => IntervalEdit::Keep,
+ };
+ let edit = Self {
+ name: flags.name,
+ body: flags.body,
+ anchor_unix: flags.at.map(parse_at).transpose()?,
+ interval_seconds,
+ paused: None,
+ add_agents: agents(&flags.add_agents)?,
+ remove_agents: agents(&flags.remove_agents)?,
+ };
+ if edit.name.is_none()
+ && edit.body.is_none()
+ && edit.anchor_unix.is_none()
+ && edit.interval_seconds.is_keep()
+ && edit.add_agents.is_empty()
+ && edit.remove_agents.is_empty()
+ {
+ bail!(
+ "nothing to change: pass --name, --body, --at, --every, --once, --add-agent \
+ or --remove-agent"
+ );
+ }
+ Ok(edit)
+ }
+
+ fn paused(paused: bool) -> Self {
+ Self {
+ paused: Some(paused),
+ ..Self::default()
+ }
+ }
+}
+
+fn agents(names: &[String]) -> Result> {
+ names.iter().map(|n| parse_ident(n, "agent name")).collect()
+}
+
+fn runtime() -> Result {
+ tokio::runtime::Builder::new_current_thread()
+ .enable_io()
+ .build()
+ .context("starting a tokio runtime for the controller request")
+}
+
+const COLLECTION: &str = "/api/schedules";
+
+/// The schedule's path. The id is checked like an agent name, which every id
+/// the controller mints satisfies, so it cannot reshape the URI.
+fn member(id: &str) -> Result {
+ let id = parse_ident(id, "schedule id")?;
+ Ok(format!("{COLLECTION}/{id}"))
+}
+
+/// What `schedule add` was asked for.
+pub(crate) struct AddFlags<'a> {
+ pub name: &'a str,
+ pub agents: &'a [String],
+ pub body: &'a str,
+ pub at: Option<&'a str>,
+ pub every: Option<&'a str>,
+ pub paused: bool,
+}
+
+/// Run `swarmctl schedule add`.
+pub(crate) fn add(socket: &Path, flags: &AddFlags<'_>) -> Result<()> {
+ let request = AddRequest {
+ name: flags.name,
+ agents: agents(flags.agents)?,
+ body: flags.body,
+ anchor_unix: flags.at.map(parse_at).transpose()?,
+ interval_seconds: flags.every.map(parse_every).transpose()?,
+ paused: flags.paused,
+ };
+ let added: ScheduleView = runtime()?.block_on(send(
+ socket,
+ hyper::Method::POST,
+ COLLECTION,
+ Some(&request),
+ "schedule-add",
+ ))?;
+ println!("added schedule {}", added.id);
+ println!("{}", render(&added));
+ Ok(())
+}
+
+/// Run `swarmctl schedule list`, optionally only the schedules naming `agent`.
+pub(crate) fn list(socket: &Path, agent: Option<&str>) -> Result<()> {
+ let agent = agent.map(|a| parse_ident(a, "agent name")).transpose()?;
+ let schedules: Vec = runtime()?.block_on(send(
+ socket,
+ hyper::Method::GET,
+ COLLECTION,
+ None::<&()>,
+ "schedule-list",
+ ))?;
+ let shown: Vec<&ScheduleView> = schedules
+ .iter()
+ .filter(|s| agent.as_ref().is_none_or(|a| s.agents.contains(a)))
+ .collect();
+ if shown.is_empty() {
+ println!("no schedules");
+ }
+ for s in shown {
+ println!("{}", render(s));
+ }
+ Ok(())
+}
+
+/// Run `swarmctl schedule edit`.
+pub(crate) fn edit(socket: &Path, id: &str, edit: &EditRequest) -> Result<()> {
+ let edited: ScheduleView = runtime()?.block_on(send(
+ socket,
+ hyper::Method::PATCH,
+ &member(id)?,
+ Some(edit),
+ "schedule-edit",
+ ))?;
+ println!("{}", render(&edited));
+ Ok(())
+}
+
+/// Run `swarmctl schedule pause` or `resume`.
+pub(crate) fn set_paused(socket: &Path, id: &str, paused: bool) -> Result<()> {
+ edit(socket, id, &EditRequest::paused(paused))
+}
+
+/// Run `swarmctl schedule rm`.
+pub(crate) fn remove(socket: &Path, id: &str) -> Result<()> {
+ let uri = member(id)?;
+ runtime()?.block_on(send::<(), ()>(
+ socket,
+ hyper::Method::DELETE,
+ &uri,
+ None,
+ "schedule-rm",
+ ))?;
+ println!("removed schedule {id}");
+ Ok(())
+}
+
+/// `--at`: unix seconds, or an RFC 3339 time with its offset.
+fn parse_at(value: &str) -> Result {
+ if let Ok(unix) = value.parse::() {
+ return Ok(unix);
+ }
+ chrono::DateTime::parse_from_rfc3339(value)
+ .map(|t| t.timestamp())
+ .with_context(|| {
+ format!(
+ "--at {value:?} is neither unix seconds nor an RFC 3339 time such as \
+ 2026-10-05T09:00:00+02:00"
+ )
+ })
+}
+
+/// `--every`: a positive count of seconds, minutes, hours or days, such as
+/// `90s`, `15m`, `1h` or `7d`. A bare number is seconds.
+fn parse_every(value: &str) -> Result {
+ let (digits, unit) = match value.char_indices().last() {
+ Some((i, c)) if c.is_ascii_alphabetic() => (&value[..i], c),
+ _ => (value, 's'),
+ };
+ let scale = match unit {
+ 's' => 1,
+ 'm' => 60,
+ 'h' => 3600,
+ 'd' => 86_400,
+ _ => bail!("--every {value:?}: the unit must be s, m, h or d"),
+ };
+ let count: u64 = digits
+ .parse()
+ .with_context(|| format!("--every {value:?} is not a count followed by s, m, h or d"))?;
+ match count.checked_mul(scale) {
+ Some(0) => bail!("--every {value:?} is zero; leave --every out for a one-shot"),
+ Some(seconds) => Ok(seconds),
+ None => bail!("--every {value:?} is too long"),
+ }
+}
+
+fn utc(unix: i64) -> String {
+ chrono::DateTime::from_timestamp(unix, 0).map_or_else(
+ || unix.to_string(),
+ |t| t.format("%Y-%m-%dT%H:%M:%SZ").to_string(),
+ )
+}
+
+/// One row per schedule: id, name, cadence, next fire, agents, and the
+/// body's first line.
+fn render(s: &ScheduleView) -> String {
+ let cadence = match s.interval_seconds {
+ Some(i) => format!("every {}", every_text(i)),
+ None => "once".to_owned(),
+ };
+ let next = match (s.paused, s.next_fire_unix) {
+ (true, _) => "paused".to_owned(),
+ (false, Some(t)) => format!("next {}", utc(t)),
+ (false, None) => format!("fired at {}", utc(s.anchor_unix)),
+ };
+ let body = s.body.lines().next().unwrap_or_default();
+ format!(
+ "{} {:?} {cadence} from {} {next} agents: {} | {body}",
+ s.id,
+ s.name,
+ utc(s.anchor_unix),
+ s.agents.join(", ")
+ )
+}
+
+fn every_text(seconds: u64) -> String {
+ for (unit, scale) in [("d", 86_400), ("h", 3600), ("m", 60)] {
+ if seconds.is_multiple_of(scale) {
+ return format!("{}{unit}", seconds / scale);
+ }
+ }
+ format!("{seconds}s")
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+
+ fn flags() -> EditFlags<'static> {
+ EditFlags {
+ name: None,
+ body: None,
+ at: None,
+ every: None,
+ once: false,
+ add_agents: Vec::new(),
+ remove_agents: Vec::new(),
+ }
+ }
+
+ #[test]
+ fn every_takes_a_count_and_a_unit() {
+ assert_eq!(parse_every("90").unwrap(), 90);
+ assert_eq!(parse_every("90s").unwrap(), 90);
+ assert_eq!(parse_every("15m").unwrap(), 900);
+ assert_eq!(parse_every("1h").unwrap(), 3600);
+ assert_eq!(parse_every("7d").unwrap(), 604_800);
+ assert!(parse_every("0h").is_err());
+ assert!(parse_every("1w").is_err());
+ assert!(parse_every("h").is_err());
+ assert!(parse_every("").is_err());
+ }
+
+ #[test]
+ fn at_takes_unix_seconds_or_rfc3339() {
+ assert_eq!(parse_at("1700000000").unwrap(), 1_700_000_000);
+ assert_eq!(parse_at("1970-01-01T01:00:00+01:00").unwrap(), 0);
+ assert!(parse_at("tomorrow").is_err());
+ }
+
+ /// The edit body says exactly what the flags changed: a `--once` sends an
+ /// explicit `null` interval, and an unset flag sends nothing.
+ #[test]
+ fn an_edit_sends_only_what_was_asked() {
+ let body = |e: &EditRequest| serde_json::to_string(e).unwrap();
+ let e = EditRequest::new(EditFlags {
+ name: Some("hi".to_owned()),
+ ..flags()
+ })
+ .unwrap();
+ assert_eq!(body(&e), r#"{"name":"hi"}"#);
+ let e = EditRequest::new(EditFlags {
+ once: true,
+ ..flags()
+ })
+ .unwrap();
+ assert_eq!(body(&e), r#"{"interval_seconds":null}"#);
+ let e = EditRequest::new(EditFlags {
+ at: Some("5"),
+ every: Some("1m"),
+ add_agents: vec!["iris".to_owned()],
+ remove_agents: vec!["atlas".to_owned()],
+ ..flags()
+ })
+ .unwrap();
+ assert_eq!(
+ body(&e),
+ r#"{"anchor_unix":5,"interval_seconds":60,"add_agents":["iris"],"remove_agents":["atlas"]}"#
+ );
+ assert_eq!(body(&EditRequest::paused(true)), r#"{"paused":true}"#);
+ assert!(EditRequest::new(flags()).is_err());
+ assert!(
+ EditRequest::new(EditFlags {
+ add_agents: vec!["Not An Ident".to_owned()],
+ ..flags()
+ })
+ .is_err()
+ );
+ }
+
+ /// The controller's list item, as the wire shape documents it, decodes
+ /// and renders as one row naming every agent.
+ #[test]
+ fn a_listed_schedule_renders_as_one_row() {
+ let s: ScheduleView = serde_json::from_str(
+ r#"{"id":"abc","name":"standup","agents":["atlas","iris"],"body":"post it\nmore",
+ "anchor_unix":0,"interval_seconds":86400,"paused":false,"next_fire_unix":86400}"#,
+ )
+ .unwrap();
+ let row = render(&s);
+ assert!(!row.contains('\n'), "{row}");
+ assert!(row.contains("\"standup\""), "{row}");
+ assert!(row.contains("every 1d"), "{row}");
+ assert!(row.contains("next 1970-01-02T00:00:00Z"), "{row}");
+ assert!(row.contains("agents: atlas, iris"), "{row}");
+ assert!(row.ends_with("| post it"), "{row}");
+ }
+}