From e111807f59d4c0f888943a79b9178421f9b33210 Mon Sep 17 00:00:00 2001 From: atlas Date: Sun, 18 Oct 2026 12:33:35 +0200 Subject: [PATCH] swarm-wire: take agent_schedules's wire types --- hive-agent/src/schedule_timer.rs | 3 +- swarm-controller/src/schedules.rs | 5 +- swarm-nats-auth/src/policy.rs | 6 +- swarm-queue-client/src/agent_schedules.rs | 155 +--------------------- swarm-queue-client/src/lib.rs | 7 +- swarm-wire/src/agent_inference.rs | 8 +- swarm-wire/src/agent_schedules.rs | 147 ++++++++++++++++++++ swarm-wire/src/lib.rs | 1 + 8 files changed, 170 insertions(+), 162 deletions(-) create mode 100644 swarm-wire/src/agent_schedules.rs diff --git a/hive-agent/src/schedule_timer.rs b/hive-agent/src/schedule_timer.rs index a6f4f4c2..5fac9a6d 100644 --- a/hive-agent/src/schedule_timer.rs +++ b/hive-agent/src/schedule_timer.rs @@ -31,7 +31,8 @@ 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 swarm_queue_client::agent_schedules::{Read, read_own}; +use swarm_wire::agent_schedules::{AgentSchedules, Schedule}; use tokio::sync::Notify; use crate::todos::Todos; diff --git a/swarm-controller/src/schedules.rs b/swarm-controller/src/schedules.rs index a1299682..0d014417 100644 --- a/swarm-controller/src/schedules.rs +++ b/swarm-controller/src/schedules.rs @@ -7,7 +7,7 @@ //! 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 +//! (`swarm_wire::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 @@ -30,7 +30,8 @@ use serde::{Deserialize, Serialize}; use swarm_controller_api::schedules::{ AddScheduleRequest, EditScheduleRequest, IntervalEdit, ScheduleView, }; -use swarm_queue_client::agent_schedules::{self, AgentSchedules, Schedule}; +use swarm_queue_client::agent_schedules; +use swarm_wire::agent_schedules::{AgentSchedules, Schedule}; use crate::agent_renewal::unix_now; use crate::routes::{AppState, error_problem}; diff --git a/swarm-nats-auth/src/policy.rs b/swarm-nats-auth/src/policy.rs index 6473809d..244d5eed 100644 --- a/swarm-nats-auth/src/policy.rs +++ b/swarm-nats-auth/src/policy.rs @@ -881,7 +881,7 @@ mod tests { // The read reaches only this bucket. for other in [ swarm_wire::agent_inference::direct_get_subject("atlas"), - swarm_queue_client::agent_schedules::direct_get_subject("atlas"), + swarm_wire::agent_schedules::direct_get_subject("atlas"), "$JS.API.DIRECT.GET.KV_hive-wanted-beta.$KV.hive-wanted-beta.beta".to_owned(), ] { assert!(!may(&other), "hive alpha may publish {other}"); @@ -2012,7 +2012,7 @@ mod tests { /// and no way to open, watch or write the bucket. #[test] fn an_agent_may_read_only_its_own_schedule_key() { - use swarm_queue_client::agent_schedules::{STREAM, direct_get_subject, key_subject}; + use swarm_wire::agent_schedules::{STREAM, direct_get_subject, key_subject}; let p = policy() .with_agent_token_subjects(vec![ @@ -2078,7 +2078,7 @@ mod tests { format!("$JS.API.STREAM.INFO.{STREAM}"), format!("$JS.API.CONSUMER.CREATE.{STREAM}"), key_subject("atlas"), - swarm_queue_client::agent_schedules::direct_get_subject("atlas"), + swarm_wire::agent_schedules::direct_get_subject("atlas"), ] { assert!( !atlas.publish.iter().any(|g| covers(g, &refused)), diff --git a/swarm-queue-client/src/agent_schedules.rs b/swarm-queue-client/src/agent_schedules.rs index fcfdf9f0..f55d9e7b 100644 --- a/swarm-queue-client/src/agent_schedules.rs +++ b/swarm-queue-client/src/agent_schedules.rs @@ -1,93 +1,6 @@ -//! 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 `swarm_wire::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) - } -} +//! The connection side of the per-agent schedule KV bucket: opening it for +//! writing and the agent's own direct read. The bucket's names, subjects and +//! stored types are `swarm_wire::agent_schedules`. /// Open the schedule bucket for writing, creating it if nothing has yet. /// @@ -99,7 +12,7 @@ pub async fn open_or_create( let js = async_nats::jetstream::new(client.clone()); crate::kv::open_or_create( &js, - BUCKET, + swarm_wire::agent_schedules::BUCKET, "agent-schedule", "Scheduled prompts the swarm controller declares for each agent".to_owned(), ) @@ -111,7 +24,7 @@ pub async fn open_or_create( #[derive(Debug)] pub enum Read { /// The bucket answered; an absent or deleted key is an empty list. - Current(AgentSchedules), + Current(swarm_wire::agent_schedules::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. @@ -130,6 +43,7 @@ pub async fn read_own( agent: &str, ) -> Result> { use async_nats::jetstream::stream::DirectGetErrorKind; + use swarm_wire::agent_schedules::{AgentSchedules, STREAM, key_subject}; let js = async_nats::jetstream::new(client.clone()); let stream = js.get_stream_no_info(STREAM).await?; @@ -160,60 +74,3 @@ fn is_no_responders(e: &async_nats::jetstream::stream::DirectGetError) -> bool { 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 f22e01d7..063b3402 100644 --- a/swarm-queue-client/src/lib.rs +++ b/swarm-queue-client/src/lib.rs @@ -215,9 +215,10 @@ pub mod agents_md; /// `agents-md-initial`. pub mod agents_md_initial; -/// The bucket swarm-controller declares each agent's scheduled prompts in — -/// one key per agent, read by that agent alone. The names are unconditional; -/// writing needs `kv` and reading needs `agent-schedules`. +/// Opening and reading the bucket swarm-controller declares each agent's +/// scheduled prompts in — one key per agent, read by that agent alone. The +/// names and stored types are `swarm_wire::agent_schedules`; writing needs `kv` +/// and reading needs `agent-schedules`. pub mod agent_schedules; /// The bucket swarm-controller sets each agent's model, effort and subagent diff --git a/swarm-wire/src/agent_inference.rs b/swarm-wire/src/agent_inference.rs index 43aeaf65..b5586aea 100644 --- a/swarm-wire/src/agent_inference.rs +++ b/swarm-wire/src/agent_inference.rs @@ -7,11 +7,11 @@ //! no swarm. //! //! **An agent reads with one `DIRECT.GET` on its own key, not a watch**, for -//! the reason `swarm_queue_client::agent_schedules` gives: the direct get names the key -//! in its subject ([`direct_get_subject`]), which is -//! what lets the queue grant scope the read to one agent. +//! the reason [`crate::agent_schedules`] gives: the direct get names the key +//! in its subject ([`direct_get_subject`]), which is what lets the queue grant +//! scope the read to one agent. //! -//! Keyed by the agent alone, like `swarm_queue_client::agent_schedules`: an agent can +//! Keyed by the agent alone, like [`crate::agent_schedules`]: an agent can //! move between hives. /// The KV bucket inference settings are published into, keyed by agent name. diff --git a/swarm-wire/src/agent_schedules.rs b/swarm-wire/src/agent_schedules.rs new file mode 100644 index 00000000..12fb7ad1 --- /dev/null +++ b/swarm-wire/src/agent_schedules.rs @@ -0,0 +1,147 @@ +//! 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 ([`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 ([`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: +//! `swarm_queue_client::agent_schedules::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) + } +} + +#[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-wire/src/lib.rs b/swarm-wire/src/lib.rs index 546735a7..adc9713a 100644 --- a/swarm-wire/src/lib.rs +++ b/swarm-wire/src/lib.rs @@ -8,6 +8,7 @@ pub mod agent_icon; pub mod agent_inference; pub mod agent_limits; +pub mod agent_schedules; pub mod agent_login; pub mod agent_status; pub mod builder;