From 49015eb55350a92182874c357aff8e3e7c165a9b Mon Sep 17 00:00:00 2001 From: atlas Date: Sat, 17 Oct 2026 12:19:43 +0200 Subject: [PATCH] keep the operator list under one key and read it with one direct get The agent read the operator-profiles bucket with a keys() listing, an ordered consumer whose grant (STREAM.INFO, CONSUMER.CREATE) and flow-control reply had never run against a permissioned queue. The controller now writes the whole list as one JSON array under one key, only when it changed, and the agent reads it with one DIRECT.GET the way agent_inference::read_own reads its key. The agent grant shrinks to that one subject. --- docs/turn-loop/claude-invocation.md | 4 +- hive-agent/Cargo.toml | 6 +- hive-agent/src/prompt.rs | 20 +- nix/host-modules/swarm-nats.nix | 5 +- swarm-controller/Cargo.toml | 2 + swarm-controller/README.md | 11 +- .../src/forge/operator_profiles.rs | 189 +++++++++--------- swarm-nats-auth/src/agent_token.rs | 6 +- swarm-nats-auth/src/policy.rs | 73 +++---- swarm-queue-client/Cargo.toml | 8 +- swarm-queue-client/src/lib.rs | 2 +- swarm-queue-client/src/operator_profiles.rs | 129 ++++++++---- 12 files changed, 254 insertions(+), 201 deletions(-) diff --git a/docs/turn-loop/claude-invocation.md b/docs/turn-loop/claude-invocation.md index 70b81e2c..28dc5548 100644 --- a/docs/turn-loop/claude-invocation.md +++ b/docs/turn-loop/claude-invocation.md @@ -263,8 +263,8 @@ needs to `chown` a bind mount), once at startup: shape for the swarm), and `{operators}`: a paragraph listing the swarm's operators with the pronouns from each one's forge profile, for example `Operators: mara (she/her), sol.`. The harness reads - them from the `operator-profiles` KV bucket, which - `swarm-controller` writes. The harness waits up to five seconds for + them with one direct get on the one key of the `operator-profiles` + KV bucket, which `swarm-controller` writes. The harness waits up to five seconds for them. When the bucket is empty or doesn't answer in time, `{operators}` is empty and the prompt names no operator. Either way the template tells agents to say "the operator" in third person. A diff --git a/hive-agent/Cargo.toml b/hive-agent/Cargo.toml index e75af5d6..ffd7e439 100644 --- a/hive-agent/Cargo.toml +++ b/hive-agent/Cargo.toml @@ -38,9 +38,9 @@ rusqlite.workspace = true schemars.workspace = true serde.workspace = true serde_json.workspace = true -# `agent-schedules` and `agent-inference`: the schedule ticker and the -# inference poller each read their own key with one direct get. -# `operator-profiles`: the prompt renderer lists the operator-profile bucket. +# `agent-schedules`, `agent-inference` and `operator-profiles`: the schedule +# ticker and the inference poller each read their own key, and the prompt +# renderer the operator list, with one direct get. # `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 = [ diff --git a/hive-agent/src/prompt.rs b/hive-agent/src/prompt.rs index 4828c216..5dccb8ea 100644 --- a/hive-agent/src/prompt.rs +++ b/hive-agent/src/prompt.rs @@ -81,12 +81,20 @@ fn operators_line(operators: &[OperatorProfile]) -> String { /// read within [`OPERATORS_TIMEOUT`]: the prompt then reads as it does with no /// operator list. async fn operators() -> Vec { - within_timeout(OPERATORS_TIMEOUT, async { - let client = crate::swarm_queue::client().await; - swarm_queue_client::ensure_connected(&client)?; - operator_profiles::read_all(&client).await - }) - .await + within_timeout(OPERATORS_TIMEOUT, read_operators()).await +} + +/// One read of the operator list over this agent's queue connection. +async fn read_operators() +-> std::result::Result, Box> { + let client = crate::swarm_queue::client().await; + swarm_queue_client::ensure_connected(&client)?; + match operator_profiles::read(&client).await? { + operator_profiles::Read::Current(operators) => Ok(operators), + operator_profiles::Read::NoBucket => { + Err("the operator-profiles bucket does not answer".into()) + } + } } /// `read`'s profiles, or none when it fails or outlasts `timeout`. diff --git a/nix/host-modules/swarm-nats.nix b/nix/host-modules/swarm-nats.nix index 288b0040..0b98752b 100644 --- a/nix/host-modules/swarm-nats.nix +++ b/nix/host-modules/swarm-nats.nix @@ -905,8 +905,9 @@ in # the harness and the subagent daemon read the tool permissions # it holds. "--agent-token-publish-subject ${lib.escapeShellArg "\$\$JS.API.DIRECT.GET.KV_agent-permissions.\$\$KV.agent-permissions.{agent}"}" - # Every agent also reads the whole `operator-profiles` bucket. - # That grant names no agent, so the responder adds it itself + # Every agent also reads the operator list, the one key of the + # `operator-profiles` bucket, with one direct get. That grant + # names no agent, so the responder adds it itself # (`swarm-nats-auth`'s `Policy::agent_token_permissions`). "--store-cert-role ${lib.escapeShellArg authCertRole}" ]; diff --git a/swarm-controller/Cargo.toml b/swarm-controller/Cargo.toml index bae6e814..a18270d9 100644 --- a/swarm-controller/Cargo.toml +++ b/swarm-controller/Cargo.toml @@ -150,6 +150,8 @@ tempfile = "3" tower = { version = "0.5", features = ["util"] } # A layer that captures `review_classifier::runs`'s log event in a test. tracing-subscriber.workspace = true +# `operator_profiles::read`, the agent's read, run against what `sync` wrote. +swarm-queue-client = { workspace = true, features = ["operator-profiles"] } [lints] workspace = true diff --git a/swarm-controller/README.md b/swarm-controller/README.md index abee7e89..d5b88243 100644 --- a/swarm-controller/README.md +++ b/swarm-controller/README.md @@ -72,11 +72,12 @@ The swarm's control plane. The swarm UI and `swarmctl` are its clients. writes (`GET`/`PUT /api/review-classifier/settings`) → [`docs/swarm/review-classifier.md`](../docs/swarm/review-classifier.md#swarm-level-settings). - **Operator profiles** — each operator's name and the pronouns on their - forge profile, one key per member of authelia's operator group in the - `operator-profiles` KV bucket, which this daemon alone writes, at start and - every five minutes. A forge read that fails keeps the stored value, and a - name that leaves the group loses its key. Every agent lists the bucket into - its system prompt → + forge profile, for every member of authelia's operator group, as one list + under the one key of the `operator-profiles` KV bucket. This daemon alone + writes it, at start and every five minutes, and only when the list changed. + A forge read that fails keeps that operator's stored entry, and a name that + leaves the group drops out of the list. Every agent reads the key into its + system prompt → [`docs/turn-loop/claude-invocation.md`](../docs/turn-loop/claude-invocation.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/forge/operator_profiles.rs b/swarm-controller/src/forge/operator_profiles.rs index a592069f..24875e13 100644 --- a/swarm-controller/src/forge/operator_profiles.rs +++ b/swarm-controller/src/forge/operator_profiles.rs @@ -1,21 +1,21 @@ -//! Each operator's forge-profile pronouns, published to the `operator-profiles` -//! KV bucket (`swarm_queue_client::operator_profiles`) at start and every -//! [`RECONCILE_INTERVAL`] after. Every agent reads that bucket into its system +//! Each operator's forge-profile pronouns, published as one list under the +//! `operator-profiles` KV bucket's one key +//! (`swarm_queue_client::operator_profiles`) at start and every +//! [`RECONCILE_INTERVAL`] after. Every agent reads that key into its system //! prompt. //! //! An operator is a member of authelia's operator group, read the way -//! `crate::matrix_account::operator_admin` reads it. A pass writes a key only -//! when its value changed and removes the key of a name no longer in the -//! group. An operator whose forge profile could not be read keeps the value -//! the bucket holds; one with no forge account yet is listed with no pronouns. +//! `crate::matrix_account::operator_admin` reads it. A pass writes the key +//! only when the list changed, so a name no longer in the group drops out of +//! it. An operator whose forge profile could not be read keeps the entry the +//! stored list holds (name only when it holds none); one with no forge account +//! yet is listed with no pronouns. -use std::collections::BTreeSet; use std::sync::Arc; use anyhow::{Context, Result}; use async_nats::jetstream::kv::Store; -use futures_util::TryStreamExt as _; -use swarm_queue_client::operator_profiles::{self, OperatorProfile}; +use swarm_queue_client::operator_profiles::{self, KEY, OperatorProfile}; use super::Client; use super::agent_token::is_not_found; @@ -47,91 +47,68 @@ impl Profiles for Client { /// What one pass did. #[derive(Debug, Default, PartialEq, Eq)] pub(crate) struct Outcome { - written: usize, - removed: usize, - /// Operators whose profile could not be read, left as they were. + /// The list differed from the stored one and was written. + written: bool, + /// Operators whose profile could not be read, left as stored. kept: usize, } -/// Whether `name` can be a key: one subject token of the characters NATS -/// accepts in one. -fn is_key(name: &str) -> bool { - !name.is_empty() - && name - .bytes() - .all(|b| b.is_ascii_alphanumeric() || b == b'-' || b == b'_') -} - -/// Bring `store` to one key per name in `operators`, each holding that -/// operator's pronouns as `forge` reports them. +/// Bring [`KEY`] in `store` to the list of `operators`, each with its +/// pronouns as `forge` reports them, sorted by name. /// /// # Errors -/// When the bucket cannot be listed, read or written. A forge read that fails -/// is logged and leaves that operator's key alone. +/// When the key cannot be read or written. A forge read that fails is logged +/// and keeps that operator's stored entry. A stored value this build cannot +/// decode reads as an empty list, so the pass replaces it. pub(crate) async fn sync( operators: &[String], forge: &impl Profiles, store: &Store, ) -> Result { - let mut stale: BTreeSet = store - .keys() + let stored: Vec = store + .get(KEY) .await - .context("listing the operator-profiles bucket")? - .try_collect() - .await - .context("listing the operator-profiles bucket")?; + .context("reading the operator list")? + .and_then(|value| serde_json::from_slice(&value).ok()) + .unwrap_or_default(); let mut outcome = Outcome::default(); + let mut list = Vec::with_capacity(operators.len()); for name in operators { - stale.remove(name); - if !is_key(name) { - tracing::warn!( - operator = %name, - "operator profiles: the username is no legal key; skipped" - ); - continue; - } let pronouns = match forge.pronouns(name).await { Ok(pronouns) => pronouns, Err(e) => { tracing::warn!( operator = %name, error = %format!("{e:#}"), - "operator profiles: forge profile unreadable; keeping the stored value" + "operator profiles: forge profile unreadable; keeping the stored entry" ); outcome.kept += 1; - continue; + stored + .iter() + .find(|o| o.name == *name) + .and_then(|o| o.pronouns.clone()) } }; - let profile = OperatorProfile { + list.push(OperatorProfile { name: name.clone(), pronouns, - }; - let stored = store - .get(name) - .await - .with_context(|| format!("reading the profile of {name}"))? - .and_then(|value| serde_json::from_slice::(&value).ok()); - if stored.as_ref() != Some(&profile) { - let value = serde_json::to_vec(&profile).context("encoding an operator profile")?; - store - .put(name, value.into()) - .await - .with_context(|| format!("writing the profile of {name}"))?; - outcome.written += 1; - } + }); } - for name in stale { + list.sort_by(|a, b| a.name.cmp(&b.name)); + list.dedup_by(|a, b| a.name == b.name); + if list != stored { + let value = serde_json::to_vec(&list).context("encoding the operator list")?; store - .delete(&name) + .put(KEY, value.into()) .await - .with_context(|| format!("removing the profile of {name}"))?; - outcome.removed += 1; + .context("writing the operator list")?; + outcome.written = true; } Ok(outcome) } -/// One pass: read the operator group, then [`sync`] the bucket to it. Fails -/// as a whole, changing nothing, when the group cannot be read. +/// One pass: read the operator group, then [`sync`] the key to it. Fails as a +/// whole, changing nothing, when the group cannot be read. async fn pass(forge: &Client, bridge: &AuthBridge, queue: &async_nats::Client) -> Result { let operators = bridge .list_operators() @@ -158,11 +135,9 @@ pub fn spawn(forge: Arc, bridge: Arc, queue: async_nats::Cli continue; }; match outcome { - Ok(outcome) if outcome.written > 0 || outcome.removed > 0 => tracing::info!( - written = outcome.written, - removed = outcome.removed, + Ok(outcome) if outcome.written => tracing::info!( kept = outcome.kept, - "operator profiles: bucket updated" + "operator profiles: operator list updated" ), Ok(outcome) => tracing::debug!(kept = outcome.kept, "operator profiles: unchanged"), Err(e) => tracing::warn!( @@ -199,18 +174,8 @@ mod tests { } async fn stored(store: &Store) -> Vec { - let keys: Vec = store.keys().await.unwrap().try_collect().await.unwrap(); - let mut out = Vec::new(); - for key in keys { - let value = store - .get(&key) - .await - .unwrap() - .expect("a listed key has a value"); - out.push(serde_json::from_slice(&value).unwrap()); - } - out.sort_by(|a: &OperatorProfile, b| a.name.cmp(&b.name)); - out + let value = store.get(KEY).await.unwrap().expect("the list is written"); + serde_json::from_slice(&value).unwrap() } fn profile(name: &str, pronouns: Option<&str>) -> OperatorProfile { @@ -221,34 +186,36 @@ mod tests { } #[tokio::test] - async fn profiles_are_written_and_a_departed_operator_is_removed() { + async fn the_list_is_written_once_and_a_departed_operator_drops_out() { let nats = crate::test_nats::TestNats::start(); let client = nats.client().await; let store = operator_profiles::open_or_create(&client).await.unwrap(); let forge = Fake([("mara", Ok(Some("she/her"))), ("sol", Ok(None))].into()); - let outcome = sync(&names(&["mara", "sol"]), &forge, &store) + let outcome = sync(&names(&["sol", "mara"]), &forge, &store) .await .unwrap(); - assert_eq!(outcome.written, 2); + assert!(outcome.written); assert_eq!( stored(&store).await, [profile("mara", Some("she/her")), profile("sol", None)] ); + let revision = store.entry(KEY).await.unwrap().unwrap().revision; - // An unchanged profile is not rewritten. + // An unchanged list is not rewritten. let outcome = sync(&names(&["mara", "sol"]), &forge, &store) .await .unwrap(); assert_eq!(outcome, Outcome::default()); + assert_eq!(store.entry(KEY).await.unwrap().unwrap().revision, revision); let outcome = sync(&names(&["mara"]), &forge, &store).await.unwrap(); - assert_eq!(outcome.removed, 1); + assert!(outcome.written); assert_eq!(stored(&store).await, [profile("mara", Some("she/her"))]); } #[tokio::test] - async fn a_failed_forge_read_keeps_the_stored_value() { + async fn a_failed_forge_read_keeps_the_stored_entry() { let nats = crate::test_nats::TestNats::start(); let client = nats.client().await; let store = operator_profiles::open_or_create(&client).await.unwrap(); @@ -264,27 +231,57 @@ mod tests { .unwrap(); // Control: the same sync does change a value the forge reads anew. - let failing = Fake([("mara", Err(())), ("sol", Ok(Some("he/him")))].into()); - let outcome = sync(&names(&["mara", "sol"]), &failing, &store) + // `ren` is new and unreadable, so it is listed by name only. + let failing = Fake( + [ + ("mara", Err(())), + ("sol", Ok(Some("he/him"))), + ("ren", Err(())), + ] + .into(), + ); + let outcome = sync(&names(&["mara", "sol", "ren"]), &failing, &store) .await .unwrap(); - assert_eq!(outcome.kept, 1); - assert_eq!(outcome.removed, 0); + assert_eq!(outcome.kept, 2); assert_eq!( stored(&store).await, [ profile("mara", Some("she/her")), + profile("ren", None), profile("sol", Some("he/him")) ] ); } - #[test] - fn a_dotted_or_empty_name_is_no_key() { - assert!(is_key("mara")); - assert!(is_key("op_2-x")); - assert!(!is_key("")); - assert!(!is_key("ma.ra")); - assert!(!is_key("ma*")); + /// The agent's read (`operator_profiles::read`, one direct get) finds + /// what a pass wrote, an empty list for a bucket with no key yet, and + /// `NoBucket` before the bucket exists. + #[tokio::test] + async fn the_agent_read_finds_the_list_and_falls_back_without_one() { + use swarm_queue_client::operator_profiles::{Read, read}; + + let nats = crate::test_nats::TestNats::start(); + let client = nats.client().await; + assert!(matches!(read(&client).await.unwrap(), Read::NoBucket)); + + let store = operator_profiles::open_or_create(&client).await.unwrap(); + let Read::Current(empty) = read(&client).await.unwrap() else { + panic!("the bucket exists"); + }; + assert_eq!(empty, []); + + let forge = Fake([("mara", Ok(Some("she/her")))].into()); + sync(&names(&["mara"]), &forge, &store).await.unwrap(); + let Read::Current(list) = read(&client).await.unwrap() else { + panic!("the bucket exists"); + }; + assert_eq!(list, [profile("mara", Some("she/her"))]); + + store.delete(KEY).await.unwrap(); + let Read::Current(deleted) = read(&client).await.unwrap() else { + panic!("the bucket exists"); + }; + assert_eq!(deleted, []); } } diff --git a/swarm-nats-auth/src/agent_token.rs b/swarm-nats-auth/src/agent_token.rs index b7a9df2c..a1cc1d20 100644 --- a/swarm-nats-auth/src/agent_token.rs +++ b/swarm-nats-auth/src/agent_token.rs @@ -118,10 +118,8 @@ mod tests { "$SWARM.term.atlas".to_owned(), "$SWARM.agent-state.atlas".to_owned(), "$KV.agent-icons.atlas".to_owned(), - "$JS.API.STREAM.INFO.KV_operator-profiles".to_owned(), - "$JS.API.CONSUMER.CREATE.KV_operator-profiles".to_owned(), - "$JS.API.CONSUMER.CREATE.KV_operator-profiles.>".to_owned(), - "$JS.API.DIRECT.GET.KV_operator-profiles.>".to_owned(), + "$JS.API.DIRECT.GET.KV_operator-profiles.$KV.operator-profiles.operators" + .to_owned(), ] ); } diff --git a/swarm-nats-auth/src/policy.rs b/swarm-nats-auth/src/policy.rs index d72985e2..c3482dd4 100644 --- a/swarm-nats-auth/src/policy.rs +++ b/swarm-nats-auth/src/policy.rs @@ -181,7 +181,13 @@ impl Policy { if publish.is_empty() { return None; } - publish.extend(Self::operator_profiles_read()); + // The operator list (`swarm_queue_client::operator_profiles`), which + // every agent renders into its system prompt: one direct get on the + // bucket's one key, so no `STREAM.INFO`, no `CONSUMER.CREATE` and no + // `$KV.` write. Every agent reads the same key, so it is no `{agent}` + // template and is granted here rather than through + // `--agent-token-publish-subject`. + publish.push(swarm_queue_client::operator_profiles::direct_get_subject()); Some(Permissions { publish, subscribe: Some(vec![ @@ -200,26 +206,6 @@ impl Policy { }) } - /// Reading every key of the operator-profile bucket - /// (`swarm_queue_client::operator_profiles::read_all`), which an agent - /// renders into its system prompt. - /// - /// The whole bucket, since every agent reads every operator. Not an - /// `{agent}` template, so it is granted here rather than through - /// `--agent-token-publish-subject`. `STREAM.INFO` opens the bucket, both - /// `CONSUMER.CREATE` forms are the `keys()` listing (an ephemeral consumer - /// carries no name, and `>` never matches zero tokens), and `DIRECT.GET` - /// reads each key. No `$KV.` subject: the controller is the only writer. - fn operator_profiles_read() -> [String; 4] { - let stream = swarm_queue_client::operator_profiles::STREAM; - [ - format!("$JS.API.STREAM.INFO.{stream}"), - format!("$JS.API.CONSUMER.CREATE.{stream}"), - format!("$JS.API.CONSUMER.CREATE.{stream}.>"), - format!("$JS.API.DIRECT.GET.{stream}.>"), - ] - } - /// The permissions for `client_id`, or `None` when no rule matches. /// /// `None` is a denial. It is not "grant nothing and let them connect": @@ -751,11 +737,11 @@ mod tests { /// An agent grant's publish list without the operator-profile read every /// agent holds, so a test can compare the templated part exactly. fn templated(grant: &Permissions) -> Vec { - let shared = Policy::operator_profiles_read(); + let shared = swarm_queue_client::operator_profiles::direct_get_subject(); grant .publish .iter() - .filter(|s| !shared.contains(s)) + .filter(|s| **s != shared) .cloned() .collect() } @@ -1914,10 +1900,8 @@ mod tests { vec![ "$SWARM.term.atlas".to_owned(), "$SWARM.agent-state.atlas".to_owned(), - "$JS.API.STREAM.INFO.KV_operator-profiles".to_owned(), - "$JS.API.CONSUMER.CREATE.KV_operator-profiles".to_owned(), - "$JS.API.CONSUMER.CREATE.KV_operator-profiles.>".to_owned(), - "$JS.API.DIRECT.GET.KV_operator-profiles.>".to_owned(), + "$JS.API.DIRECT.GET.KV_operator-profiles.$KV.operator-profiles.operators" + .to_owned(), ] ); } @@ -2063,43 +2047,46 @@ mod tests { assert!(policy().agent_token_permissions("atlas").is_none()); } - /// Every agent reads every operator's profile, and only reads it: the - /// bucket opens, lists and answers a direct get, and nothing writes, - /// reshapes or deletes it. + /// Every agent reads the operator list with one direct get on its one key, + /// and nothing else of the bucket: no lookup, no listing, no other key and + /// no write. #[test] - fn an_agent_may_list_and_read_every_operator_profile_and_write_none() { - use swarm_queue_client::operator_profiles::{BUCKET, STREAM}; + fn an_agent_may_read_only_the_operator_list_key() { + use swarm_queue_client::operator_profiles::{ + BUCKET, STREAM, direct_get_subject, key_subject, + }; let p = policy() .with_agent_token_subjects(vec!["$SWARM.term.{agent}".to_owned()]) .expect("a per-agent template is valid"); let atlas = p.agent_token_permissions("atlas").expect("configured"); + assert!(atlas.publish.contains(&direct_get_subject())); let may = |subject: &str| atlas.publish.iter().any(|g| covers(g, subject)); - for allowed in [ - format!("$JS.API.STREAM.INFO.{STREAM}"), - format!("$JS.API.CONSUMER.CREATE.{STREAM}"), - format!("$JS.API.CONSUMER.CREATE.{STREAM}.c1.$KV.{BUCKET}.>"), - format!("$JS.API.DIRECT.GET.{STREAM}.$KV.{BUCKET}.mara"), - ] { - assert!(may(&allowed), "atlas may not publish {allowed}"); - } for refused in [ - format!("$KV.{BUCKET}.mara"), + key_subject(), + format!("$JS.API.DIRECT.GET.{STREAM}"), + format!("$JS.API.DIRECT.GET.{STREAM}.$KV.{BUCKET}.mara"), + format!("$JS.API.STREAM.INFO.{STREAM}"), format!("$JS.API.STREAM.CREATE.{STREAM}"), format!("$JS.API.STREAM.UPDATE.{STREAM}"), format!("$JS.API.STREAM.DELETE.{STREAM}"), format!("$JS.API.STREAM.PURGE.{STREAM}"), - "$JS.API.DIRECT.GET.KV_agent-inference.$KV.agent-inference.argus".to_owned(), + format!("$JS.API.STREAM.MSG.GET.{STREAM}"), + format!("$JS.API.CONSUMER.CREATE.{STREAM}"), + format!("$JS.API.CONSUMER.CREATE.{STREAM}.c1.$KV.{BUCKET}.>"), ] { assert!(!may(&refused), "atlas may publish {refused}"); } + // Control: the matcher does match, so the refusals above are not + // vacuous. + assert!(may(&direct_get_subject())); // The hive's own client is not the reader and gets none of it. let hive = p.permissions("hive-alpha").expect("admitted"); assert!( !hive .publish .iter() - .any(|g| covers(g, &format!("$JS.API.STREAM.INFO.{STREAM}"))) + .any(|g| covers(g, &direct_get_subject())) ); } } diff --git a/swarm-queue-client/Cargo.toml b/swarm-queue-client/Cargo.toml index b844cb40..46825254 100644 --- a/swarm-queue-client/Cargo.toml +++ b/swarm-queue-client/Cargo.toml @@ -44,10 +44,10 @@ agent-schedules = ["async-nats/jetstream"] # `agent_inference::read_own`, for the agent harness, which reads its own # inference key the same way. Explicit for the same reason as `notices`. agent-inference = ["async-nats/jetstream"] -# `operator_profiles::read_all`, for the agent harness, which lists every -# operator's profile when it renders its system prompt: a `keys()` listing, so -# the KV client rather than a bare direct get. -operator-profiles = ["async-nats/kv"] +# `operator_profiles::read`, for the agent harness, which reads the operator +# list with one direct get when it renders its system prompt. Explicit for the +# same reason as `notices`. +operator-profiles = ["async-nats/jetstream"] # `ToSchema` on the review classifier's shared types, for swarm-controller, # whose HTTP API serves the same shapes the queue carries. schema = ["dep:utoipa"] diff --git a/swarm-queue-client/src/lib.rs b/swarm-queue-client/src/lib.rs index 343f677c..6970c024 100644 --- a/swarm-queue-client/src/lib.rs +++ b/swarm-queue-client/src/lib.rs @@ -228,7 +228,7 @@ pub mod agent_limits; pub mod agent_permissions; /// The bucket swarm-controller publishes each operator's name and forge-profile -/// pronouns in — one key per operator, read by every agent. The names are +/// pronouns in — one key holding the whole list, read by every agent. The names are /// unconditional; writing needs `kv` and reading needs `operator-profiles`. pub mod operator_profiles; diff --git a/swarm-queue-client/src/operator_profiles.rs b/swarm-queue-client/src/operator_profiles.rs index f93823ea..41931407 100644 --- a/swarm-queue-client/src/operator_profiles.rs +++ b/swarm-queue-client/src/operator_profiles.rs @@ -1,25 +1,44 @@ -//! The operator-profile KV bucket: one key per operator, holding the name and -//! the pronouns from that operator's own forge profile. +//! The operator-profile KV bucket: one key, +//! [`crate::operator_profiles::KEY`], holding every operator's name and the +//! pronouns from that operator's own forge profile, as one JSON array. //! //! **swarm-controller is the only writer**, and creates the bucket on its -//! first write. An operator is a member of authelia's operator group, and the -//! controller removes the key of one who leaves it. +//! first write. An operator is a member of authelia's operator group; one who +//! leaves it drops out of the list on the controller's next pass. //! -//! **Every agent reads the whole bucket**, read-only, when it renders its -//! system prompt (`read_all`). Unlike the per-agent buckets, nothing here is -//! one agent's, so the grant covers every key and the listing a `keys()` -//! consumer needs. +//! **Every agent reads the key with one `DIRECT.GET`**, read-only, when it +//! renders its system prompt +//! ([`crate::operator_profiles::direct_get_subject`]). One key rather than one +//! per operator so that the read is a single request on one literal subject, +//! the shape of [`crate::agent_inference`]'s read: no `STREAM.INFO`, no +//! consumer. -/// The KV bucket operator profiles are published into, keyed by operator name. +/// The KV bucket operator profiles are published into. pub const BUCKET: &str = "operator-profiles"; /// The stream backing [`BUCKET`]. `create_key_value` prefixes `KV_`. pub const STREAM: &str = "KV_operator-profiles"; -/// One operator's profile: the value under `BUCKET`/``. +/// The one key in [`BUCKET`]: the whole operator list. +pub const KEY: &str = "operators"; + +/// The subject [`KEY`] is stored under. +#[must_use] +pub fn key_subject() -> String { + format!("$KV.{BUCKET}.{KEY}") +} + +/// The request subject that reads [`KEY`]'s current entry, and so the one +/// subject the queue grants every agent in this bucket. +#[must_use] +pub fn direct_get_subject() -> String { + format!("$JS.API.DIRECT.GET.{STREAM}.{}", key_subject()) +} + +/// One operator's profile: an element of the array under [`KEY`]. #[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)] pub struct OperatorProfile { - /// The operator's authelia username, which is also the key. + /// The operator's authelia username. pub name: String, /// The pronouns set on the operator's forge profile. `None` when unset or /// when the operator has no forge account yet. @@ -44,30 +63,56 @@ pub async fn open_or_create( .await } -/// Every operator's profile, sorted by name. +/// What reading [`KEY`] found. +#[cfg(feature = "operator-profiles")] +#[derive(Debug)] +pub enum Read { + /// The bucket answered; an absent or deleted key is no operators. + Current(Vec), + /// Nothing answers for the bucket's stream: the controller has not written + /// it yet, or `JetStream` is not serving it right now. + NoBucket, +} + +/// Read the operator list with the one subject the agent grant covers. /// /// # Errors /// -/// A missing bucket, a failed request or an undecodable value. A request the -/// queue refused for permissions reaches the client as a timeout, not as a -/// refusal. +/// 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 = "operator-profiles")] -pub async fn read_all( +pub async fn read( client: &async_nats::Client, -) -> Result, Box> { - use futures_util::TryStreamExt as _; +) -> Result> { + use async_nats::jetstream::stream::DirectGetErrorKind; let js = async_nats::jetstream::new(client.clone()); - let store = js.get_key_value(BUCKET).await?; - let keys: Vec = store.keys().await?.try_collect().await?; - let mut profiles = Vec::with_capacity(keys.len()); - for key in keys { - if let Some(value) = store.get(&key).await? { - profiles.push(serde_json::from_slice::(&value)?); + let stream = js.get_stream_no_info(STREAM).await?; + let message = match stream.direct_get_last_for_subject(key_subject()).await { + Ok(message) => message, + Err(e) if matches!(e.kind(), DirectGetErrorKind::NotFound) => { + return Ok(Read::Current(Vec::new())); } + Err(e) + if matches!( + e.kind(), + DirectGetErrorKind::ErrorResponse(status, _) + if status == async_nats::StatusCode::NO_RESPONDERS + ) => + { + 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(Vec::new())); } - profiles.sort_by(|a, b| a.name.cmp(&b.name)); - Ok(profiles) + Ok(Read::Current(serde_json::from_slice(&message.payload)?)) } #[cfg(test)] @@ -76,21 +121,35 @@ mod tests { /// The names the auth responder grants agents from. #[test] - fn the_stream_is_the_bucket_s() { + fn names_match_the_queue_grant() { assert_eq!(STREAM, format!("KV_{BUCKET}")); + assert_eq!(key_subject(), "$KV.operator-profiles.operators"); + assert_eq!( + direct_get_subject(), + "$JS.API.DIRECT.GET.KV_operator-profiles.$KV.operator-profiles.operators" + ); } #[test] fn unset_pronouns_are_left_out_of_the_value() { - let profile = OperatorProfile { - name: "mara".to_owned(), - pronouns: None, - }; - let json = serde_json::to_string(&profile).unwrap(); - assert_eq!(json, r#"{"name":"mara"}"#); + let list = vec![ + OperatorProfile { + name: "mara".to_owned(), + pronouns: Some("she/her".to_owned()), + }, + OperatorProfile { + name: "sol".to_owned(), + pronouns: None, + }, + ]; + let json = serde_json::to_string(&list).unwrap(); assert_eq!( - serde_json::from_str::(&json).unwrap(), - profile + json, + r#"[{"name":"mara","pronouns":"she/her"},{"name":"sol"}]"# + ); + assert_eq!( + serde_json::from_str::>(&json).unwrap(), + list ); } }