diff --git a/docs/4-implementation/swarm/controller.md b/docs/4-implementation/swarm/controller.md index 2a3cbcb4f..c729b694a 100644 --- a/docs/4-implementation/swarm/controller.md +++ b/docs/4-implementation/swarm/controller.md @@ -180,6 +180,16 @@ appearing there. The agent's grant is publish on its own subjects and limits are the controller's and never the agent's. Subagents publish output only and read nothing. +The daemon also publishes whether each subagent is running, one key per subagent +(`.`) in the `subagent-status` KV bucket, which swarm-controller +creates with one revision per key. The value is the subagent's `state` (`running` +or `finished`), `since`, `exit`, `model`, `effort` and `ts`, the time of the +publish. A running subagent republishes every 60 seconds, and a reader treats a +`running` value older than three of those as stale. The agent's credential may +publish `$KV.subagent-status..*` and nothing else of the bucket. Keeping +this apart from the rows means a list of subagents reads one small value each +and never a terminal stream. + The queue would refuse a row too large for its `max_payload` outright and take the connection down with it, so the harness drops such a row's body before sending and leaves a marker in its place; the summary, level and icon still diff --git a/nix/host-modules/swarm-nats.nix b/nix/host-modules/swarm-nats.nix index 2192a3403..82575c05e 100644 --- a/nix/host-modules/swarm-nats.nix +++ b/nix/host-modules/swarm-nats.nix @@ -988,6 +988,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.>"}" + # One key per subagent in the `subagent-status` KV bucket + # (`swarm_wire::subagent_status`): whether it is running. + # The key is `{agent}.`, so `*` is exactly one + # subagent token. As for the icon, a KV put is one publish + # and swarm-controller creates the bucket, so no + # `$$JS.API.*` subject. + "--agent-token-publish-subject ${lib.escapeShellArg "\$\$KV.subagent-status.{agent}.*"}" # 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 diff --git a/nix/module-eval/nats-authelia.nix b/nix/module-eval/nats-authelia.nix index 10ef03aea..ea1f6441d 100644 --- a/nix/module-eval/nats-authelia.nix +++ b/nix/module-eval/nats-authelia.nix @@ -544,6 +544,7 @@ let "'$$KV.agent-icons.{agent}'" "'$$KV.agents-md.{agent}'" "'$$SWARM.term.{agent}.sub.>'" + "'$$KV.subagent-status.{agent}.*'" "'$$JS.API.DIRECT.GET.KV_agent-schedules.$$KV.agent-schedules.{agent}'" "'$$JS.API.DIRECT.GET.KV_agent-inference.$$KV.agent-inference.{agent}'" "'$$JS.API.DIRECT.GET.KV_agent-permissions.$$KV.agent-permissions.{agent}'" diff --git a/swarm-controller/src/subagent_term.rs b/swarm-controller/src/subagent_term.rs index 57de40bd0..d540465ac 100644 --- a/swarm-controller/src/subagent_term.rs +++ b/swarm-controller/src/subagent_term.rs @@ -73,6 +73,12 @@ pub(crate) fn spawn(client: async_nats::Client, wanted: Arc, hives tracing::debug!("terminal streams: queue not connected; next pass"); continue; } + if let Err(e) = swarm_queue_client::subagent_status::open_or_create(&client).await { + tracing::warn!( + error = %swarm_queue_client::chain(&e), + "subagent status: opening the bucket failed; next pass" + ); + } for agent in live_agents(&wanted, &hives).await { if let Err(e) = swarm_queue_client::subagent_term::open_or_create(&client, &agent).await diff --git a/swarm-queue-client/src/lib.rs b/swarm-queue-client/src/lib.rs index 61101db41..b06d1a1bd 100644 --- a/swarm-queue-client/src/lib.rs +++ b/swarm-queue-client/src/lib.rs @@ -200,6 +200,11 @@ pub mod wanted; /// merge, are in `swarm_wire::agent_status`. pub mod agent_status; +/// Opening the bucket each agent's subagent daemon publishes the run state of +/// its subagents into: one key per subagent. The name, keys and value are +/// `swarm_wire::subagent_status`; opening the bucket needs `kv`. +pub mod subagent_status; + /// Opening the bucket agent icons are published into: one key per agent, with /// no hive in it, unlike [`agent_status`]. The name, media type and subject, /// and why the placement stays out of the key, are in `swarm_wire::agent_icon`. diff --git a/swarm-queue-client/src/subagent_status.rs b/swarm-queue-client/src/subagent_status.rs new file mode 100644 index 000000000..7658219e1 --- /dev/null +++ b/swarm-queue-client/src/subagent_status.rs @@ -0,0 +1,20 @@ +//! The connection side of the subagent status KV bucket: opening it. The +//! bucket's name, keys and value are `swarm_wire::subagent_status`. + +/// Open the subagent status bucket, creating it if nothing has yet. +/// +/// `history: 1`: a reader wants each subagent's current state, not every +/// state it has ever been in. +#[cfg(feature = "kv")] +pub async fn open_or_create( + client: &async_nats::Client, +) -> Result { + let js = async_nats::jetstream::new(client.clone()); + crate::kv::open_or_create( + &js, + swarm_wire::subagent_status::BUCKET, + "subagent-status", + "Whether each subagent is running, as its daemon last published".to_owned(), + ) + .await +} diff --git a/swarm-wire/src/lib.rs b/swarm-wire/src/lib.rs index ba2fc3ad4..589d0bd61 100644 --- a/swarm-wire/src/lib.rs +++ b/swarm-wire/src/lib.rs @@ -29,6 +29,7 @@ pub mod router_api; pub mod router_sink; pub mod router_todo; pub mod stt; +pub mod subagent_status; pub mod subagent_term; pub mod wanted; diff --git a/swarm-wire/src/subagent_status.rs b/swarm-wire/src/subagent_status.rs new file mode 100644 index 000000000..f24e99035 --- /dev/null +++ b/swarm-wire/src/subagent_status.rs @@ -0,0 +1,164 @@ +//! The run state of an agent's subagents: one key per subagent in the +//! `subagent-status` KV bucket, holding whether it is running and since when. +//! +//! The terminal rows of a subagent (`crate::subagent_term`) say what it +//! printed, not whether it is still going: a subagent that thinks quietly +//! publishes nothing, and one that finished leaves its last row behind. The +//! swarm UI used to guess from the age of the newest row. This bucket says it +//! outright, and keeps it apart from the rows so listing the subagents of an +//! agent reads one small value each and never touches a terminal stream. +//! +//! **Each agent's subagent daemon writes its own keys**, with the agent's own +//! queue credential. The auth callout grants that credential +//! `$KV.subagent-status..*` and nothing else of `JetStream`. The writer +//! publishes to [`subject`] directly instead of opening the bucket, as the +//! agent icon writer does (`crate::agent_icon`); `swarm-controller` creates the +//! bucket (`swarm_queue_client::subagent_status::open_or_create`). A write that +//! arrives before the bucket exists is refused with "no stream", and the +//! writer tries again on its next heartbeat. +//! +//! **Keyed by the agent and the subagent**, `.`, so an agent +//! moving between hives keeps its entries, for the reason +//! `crate::agent_icon` gives. A subagent name is a single subject token, the +//! same rule `crate::subagent_term::subagent_from_subject` applies to the +//! terminal subjects. + +use serde::{Deserialize, Serialize}; + +/// The KV bucket the statuses are published into. +/// +/// A constant and not an option, for the reason `crate::agent_status::BUCKET` +/// gives: writer and reader must name the same bucket. +pub const BUCKET: &str = "subagent-status"; + +/// How often the writer republishes a `running` status while the subagent's +/// process lives, so a writer that died shows up as stale instead of running +/// forever. +pub const HEARTBEAT_S: i64 = 60; + +/// A `running` status whose [`Status::ts`] is older than this is stale: the +/// writer missed three heartbeats in a row. Readers show it as not running. +pub const STALE_AFTER_S: i64 = 3 * HEARTBEAT_S; + +/// Whether a subagent's process is alive. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum State { + /// A run is in progress. + Running, + /// The last run ended; [`Status::exit`] says how. + Finished, +} + +/// The value under one subagent's key. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct Status { + /// Whether the subagent is running. + pub state: State, + /// Unix seconds when the subagent entered [`Self::state`]. + pub since: i64, + /// The process's exit status once it finished. `None` while running, and + /// when the process was killed by a signal. + pub exit: Option, + /// The model it runs; `None` on the harness's default. + pub model: Option, + /// Its reasoning effort; `None` when not set. + pub effort: Option, + /// Unix seconds when the writer published this value. A `running` status + /// refreshes it every [`HEARTBEAT_S`]. + pub ts: i64, +} + +impl Status { + /// Whether a reader at `now_s` (unix seconds) should show the subagent as + /// running: [`State::Running`] and refreshed within [`STALE_AFTER_S`]. + #[must_use] + pub fn is_running(&self, now_s: i64) -> bool { + self.state == State::Running && now_s - self.ts <= STALE_AFTER_S + } +} + +/// The key of `subagent` under `agent`: `.`. +#[must_use] +pub fn key(agent: &str, subagent: &str) -> String { + format!("{agent}.{subagent}") +} + +/// The subject `subagent`'s status is written on: `$KV..`. +/// +/// A KV put is a `JetStream` publish to this subject, so a writer that +/// publishes directly has to use the same subject the bucket's `get` reads. +#[must_use] +pub fn subject(agent: &str, subagent: &str) -> String { + format!("$KV.{BUCKET}.{}", key(agent, subagent)) +} + +/// The subagent a bucket key names under `agent`, or `None` for a key that is +/// not exactly one subagent token below `agent`. +#[must_use] +pub fn subagent_from_key<'a>(agent: &str, key: &'a str) -> Option<&'a str> { + let name = key.strip_prefix(agent)?.strip_prefix('.')?; + (!name.is_empty() && !name.contains('.')).then_some(name) +} + +#[cfg(test)] +mod tests { + use super::*; + + /// Two tokens after the bucket, the agent and the subagent: what the + /// per-agent grant `$KV.subagent-status.{agent}.*` expands to. + #[test] + fn the_published_subject_carries_the_agent_and_the_subagent() { + assert_eq!(subject("iris", "scout"), "$KV.subagent-status.iris.scout"); + } + + #[test] + fn a_key_names_one_subagent_of_its_agent_or_nothing() { + assert_eq!(subagent_from_key("iris", "iris.scout"), Some("scout")); + assert_eq!(subagent_from_key("iris", "iris.a.b"), None); + assert_eq!(subagent_from_key("iris", "iris."), None); + assert_eq!(subagent_from_key("iris", "irisx.scout"), None); + assert_eq!(subagent_from_key("iris", "other.scout"), None); + } + + fn status(state: State, ts: i64) -> Status { + Status { + state, + since: 1_000, + exit: None, + model: None, + effort: None, + ts, + } + } + + #[test] + fn a_running_status_goes_stale_after_three_missed_heartbeats() { + let s = status(State::Running, 10_000); + assert!(s.is_running(10_000)); + assert!(s.is_running(10_000 + STALE_AFTER_S)); + assert!(!s.is_running(10_000 + STALE_AFTER_S + 1)); + } + + #[test] + fn a_finished_status_is_never_running() { + assert!(!status(State::Finished, 10_000).is_running(10_000)); + } + + #[test] + fn the_wire_form_is_snake_case_json() { + let wire = serde_json::to_string(&Status { + state: State::Finished, + since: 5, + exit: Some(0), + model: Some("opus".to_owned()), + effort: None, + ts: 6, + }) + .expect("encodes"); + assert_eq!( + wire, + r#"{"state":"finished","since":5,"exit":0,"model":"opus","effort":null,"ts":6}"# + ); + } +}