swarm: subagent-status KV bucket, the run state of each subagent

One key per subagent (<agent>.<subagent>) holding running/finished, since,
exit, model, effort and the publish time. The controller creates the bucket,
and the agent credential may publish its own keys. Nothing reads or writes
it yet; the daemon publisher and the controller list follow.

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
+(`<agent>.<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.<agent>.*` 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

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}.<subagent>`, 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

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}'"

swarm-controller/src/subagent_term.rs

@@ -73,6 +73,12 @@ pub(crate) fn spawn(client: async_nats::Client, wanted: Arc<WantedWriter>, 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

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`.

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<async_nats::jetstream::kv::Store, crate::Error> {
+    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
+}

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;
 

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.<agent>.*` 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**, `<agent>.<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<i32>,
+    /// The model it runs; `None` on the harness's default.
+    pub model: Option<String>,
+    /// Its reasoning effort; `None` when not set.
+    pub effort: Option<String>,
+    /// 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`: `<agent>.<subagent>`.
+#[must_use]
+pub fn key(agent: &str, subagent: &str) -> String {
+    format!("{agent}.{subagent}")
+}
+
+/// The subject `subagent`'s status is written on: `$KV.<bucket>.<key>`.
+///
+/// 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}"#
+        );
+    }
+}