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 ); } }