publish operators' forge-profile pronouns to kv and list them in the agent prompt
swarm-controller reads each member of authelia's operator group's forge profile every five minutes and keeps one key per operator in the new operator-profiles bucket. Every agent may list and read that bucket, and hive-agent renders it into the system prompt at harness start, falling back to today's prompt when the bucket is empty or doesn't answer within five seconds. Fixes #5383
This commit is contained in:
atlas
mara
parent
a2b4320cc4
commit
6942f8fd27
16 changed files with 681 additions and 35 deletions
|
|
@ -71,6 +71,13 @@ The swarm's control plane. The swarm UI and `swarmctl` are its clients.
|
|||
`review-classifier-settings` KV bucket, which this daemon alone reads and
|
||||
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 →
|
||||
[`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.
|
||||
- **Alerts** — the swarm Grafana's current alerts (`GET /api/alerts`), read
|
||||
|
|
|
|||
|
|
@ -53,6 +53,7 @@ pub mod docs_grant;
|
|||
pub mod hive_token;
|
||||
pub mod legacy_tokens;
|
||||
pub mod objects;
|
||||
pub mod operator_profiles;
|
||||
pub mod pull_diff;
|
||||
pub mod runner_token;
|
||||
pub mod site_admin;
|
||||
|
|
|
|||
290
swarm-controller/src/forge/operator_profiles.rs
Normal file
290
swarm-controller/src/forge/operator_profiles.rs
Normal file
|
|
@ -0,0 +1,290 @@
|
|||
//! 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
|
||||
//! 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.
|
||||
|
||||
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 super::Client;
|
||||
use super::agent_token::is_not_found;
|
||||
use crate::auth::AuthBridge;
|
||||
|
||||
/// How often [`spawn`] re-reads the operator group and their profiles.
|
||||
const RECONCILE_INTERVAL: std::time::Duration = std::time::Duration::from_mins(5);
|
||||
|
||||
/// Where an operator's pronouns come from.
|
||||
pub(crate) trait Profiles {
|
||||
/// The pronouns on `name`'s forge profile: `None` when unset or when the
|
||||
/// forge has no such account.
|
||||
fn pronouns(&self, name: &str) -> impl Future<Output = Result<Option<String>>> + Send;
|
||||
}
|
||||
|
||||
impl Profiles for Client {
|
||||
async fn pronouns(&self, name: &str) -> Result<Option<String>> {
|
||||
match self.api.call(|api| api.user_get(name).into_future()).await {
|
||||
Ok(user) => Ok(user
|
||||
.pronouns
|
||||
.map(|p| p.trim().to_owned())
|
||||
.filter(|p| !p.is_empty())),
|
||||
Err(e) if is_not_found(&e) => Ok(None),
|
||||
Err(e) => Err(e).with_context(|| format!("read forge user {name}")),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// 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.
|
||||
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.
|
||||
///
|
||||
/// # Errors
|
||||
/// When the bucket cannot be listed, read or written. A forge read that fails
|
||||
/// is logged and leaves that operator's key alone.
|
||||
pub(crate) async fn sync(
|
||||
operators: &[String],
|
||||
forge: &impl Profiles,
|
||||
store: &Store,
|
||||
) -> Result<Outcome> {
|
||||
let mut stale: BTreeSet<String> = store
|
||||
.keys()
|
||||
.await
|
||||
.context("listing the operator-profiles bucket")?
|
||||
.try_collect()
|
||||
.await
|
||||
.context("listing the operator-profiles bucket")?;
|
||||
let mut outcome = Outcome::default();
|
||||
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"
|
||||
);
|
||||
outcome.kept += 1;
|
||||
continue;
|
||||
}
|
||||
};
|
||||
let profile = 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::<OperatorProfile>(&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 {
|
||||
store
|
||||
.delete(&name)
|
||||
.await
|
||||
.with_context(|| format!("removing the profile of {name}"))?;
|
||||
outcome.removed += 1;
|
||||
}
|
||||
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.
|
||||
async fn pass(forge: &Client, bridge: &AuthBridge, queue: &async_nats::Client) -> Result<Outcome> {
|
||||
let operators = bridge
|
||||
.list_operators()
|
||||
.await
|
||||
.context("reading authelia's operator group")?;
|
||||
let store = operator_profiles::open_or_create(queue)
|
||||
.await
|
||||
.context("opening the operator-profiles bucket")?;
|
||||
sync(&operators, forge, &store).await
|
||||
}
|
||||
|
||||
/// Publish the operators' profiles now and every [`RECONCILE_INTERVAL`]
|
||||
/// after. A pass that fails is logged and retried on the next tick; it never
|
||||
/// stops the daemon.
|
||||
pub fn spawn(forge: Arc<Client>, bridge: Arc<AuthBridge>, queue: async_nats::Client) {
|
||||
tokio::spawn(async move {
|
||||
let mut ticker = tokio::time::interval(RECONCILE_INTERVAL);
|
||||
loop {
|
||||
ticker.tick().await;
|
||||
let Some(outcome) =
|
||||
super::within_forge_deadline("operator profiles", pass(&forge, &bridge, &queue))
|
||||
.await
|
||||
else {
|
||||
continue;
|
||||
};
|
||||
match outcome {
|
||||
Ok(outcome) if outcome.written > 0 || outcome.removed > 0 => tracing::info!(
|
||||
written = outcome.written,
|
||||
removed = outcome.removed,
|
||||
kept = outcome.kept,
|
||||
"operator profiles: bucket updated"
|
||||
),
|
||||
Ok(outcome) => tracing::debug!(kept = outcome.kept, "operator profiles: unchanged"),
|
||||
Err(e) => tracing::warn!(
|
||||
error = %format!("{e:#}"),
|
||||
retry_in_s = RECONCILE_INTERVAL.as_secs(),
|
||||
"operator profiles: pass failed; retrying next tick"
|
||||
),
|
||||
}
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use std::collections::BTreeMap;
|
||||
|
||||
use super::*;
|
||||
|
||||
/// A forge answering from a table: `Err` for a name whose read fails.
|
||||
struct Fake(BTreeMap<&'static str, Result<Option<&'static str>, ()>>);
|
||||
|
||||
impl Profiles for Fake {
|
||||
async fn pronouns(&self, name: &str) -> Result<Option<String>> {
|
||||
match self.0.get(name) {
|
||||
Some(Ok(p)) => Ok(p.map(str::to_owned)),
|
||||
Some(Err(())) => anyhow::bail!("the forge refused the read"),
|
||||
None => Ok(None),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
fn names(names: &[&str]) -> Vec<String> {
|
||||
names.iter().map(|&n| n.to_owned()).collect()
|
||||
}
|
||||
|
||||
async fn stored(store: &Store) -> Vec<OperatorProfile> {
|
||||
let keys: Vec<String> = 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
|
||||
}
|
||||
|
||||
fn profile(name: &str, pronouns: Option<&str>) -> OperatorProfile {
|
||||
OperatorProfile {
|
||||
name: name.to_owned(),
|
||||
pronouns: pronouns.map(str::to_owned),
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn profiles_are_written_and_a_departed_operator_is_removed() {
|
||||
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)
|
||||
.await
|
||||
.unwrap();
|
||||
assert_eq!(outcome.written, 2);
|
||||
assert_eq!(
|
||||
stored(&store).await,
|
||||
[profile("mara", Some("she/her")), profile("sol", None)]
|
||||
);
|
||||
|
||||
// An unchanged profile is not rewritten.
|
||||
let outcome = sync(&names(&["mara", "sol"]), &forge, &store)
|
||||
.await
|
||||
.unwrap();
|
||||
assert_eq!(outcome, Outcome::default());
|
||||
|
||||
let outcome = sync(&names(&["mara"]), &forge, &store).await.unwrap();
|
||||
assert_eq!(outcome.removed, 1);
|
||||
assert_eq!(stored(&store).await, [profile("mara", Some("she/her"))]);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn a_failed_forge_read_keeps_the_stored_value() {
|
||||
let nats = crate::test_nats::TestNats::start();
|
||||
let client = nats.client().await;
|
||||
let store = operator_profiles::open_or_create(&client).await.unwrap();
|
||||
let reachable = Fake(
|
||||
[
|
||||
("mara", Ok(Some("she/her"))),
|
||||
("sol", Ok(Some("they/them"))),
|
||||
]
|
||||
.into(),
|
||||
);
|
||||
sync(&names(&["mara", "sol"]), &reachable, &store)
|
||||
.await
|
||||
.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)
|
||||
.await
|
||||
.unwrap();
|
||||
assert_eq!(outcome.kept, 1);
|
||||
assert_eq!(outcome.removed, 0);
|
||||
assert_eq!(
|
||||
stored(&store).await,
|
||||
[
|
||||
profile("mara", Some("she/her")),
|
||||
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*"));
|
||||
}
|
||||
}
|
||||
|
|
@ -2863,14 +2863,20 @@ fn spawn_matrix_account_backfill(
|
|||
/// `internal/docs` grants, and
|
||||
/// agents' tokens (which also
|
||||
/// sweeps legacy tokens; see `forge::legacy_tokens`) —
|
||||
/// once there is a forge client. Lifted out of `main` for
|
||||
/// once there is a forge client, and the operators' profiles when the bridge
|
||||
/// and the queue are wired up too. Lifted out of `main` for
|
||||
/// `clippy::too_many_lines`.
|
||||
fn spawn_forge_workers(
|
||||
jobq: &Arc<Mutex<hive_jobq::scheduler::Scheduler<SwarmNodeKind, SwarmResourceKind>>>,
|
||||
client: Arc<forge::Client>,
|
||||
hives: &[HiveEntry],
|
||||
auth: Option<Arc<auth::AuthBridge>>,
|
||||
queue: Option<async_nats::Client>,
|
||||
) {
|
||||
forge::objects::spawn(Arc::clone(&client));
|
||||
if let (Some(bridge), Some(queue)) = (auth, queue) {
|
||||
forge::operator_profiles::spawn(Arc::clone(&client), bridge, queue);
|
||||
}
|
||||
forge::runner_token::spawn(Arc::clone(&client));
|
||||
let sched = Arc::clone(jobq);
|
||||
let hive_names = hives.iter().map(|h| h.name.clone()).collect();
|
||||
|
|
@ -3417,6 +3423,8 @@ async fn start_forge(
|
|||
jobq: Arc<Mutex<hive_jobq::scheduler::Scheduler<SwarmNodeKind, SwarmResourceKind>>>,
|
||||
hives: Vec<HiveEntry>,
|
||||
webhook_secret: Option<Arc<str>>,
|
||||
auth: Option<Arc<auth::AuthBridge>>,
|
||||
queue: Option<async_nats::Client>,
|
||||
) {
|
||||
let url = match forge::url_from_env() {
|
||||
Ok(Some(url)) => url,
|
||||
|
|
@ -3427,7 +3435,7 @@ async fn start_forge(
|
|||
}
|
||||
};
|
||||
forge::enable(url, forge_client, move |client| {
|
||||
spawn_forge_workers(&jobq, Arc::clone(&client), &hives);
|
||||
spawn_forge_workers(&jobq, Arc::clone(&client), &hives, auth, queue);
|
||||
config_prs.get_or_init(|| config_pr::spawn(Arc::clone(&client)));
|
||||
register_swarm_webhooks(client, webhook_secret);
|
||||
})
|
||||
|
|
@ -3549,6 +3557,8 @@ async fn main() -> Result<()> {
|
|||
Arc::clone(&jobq),
|
||||
hives.clone(),
|
||||
webhook_secret.clone(),
|
||||
auth.clone(),
|
||||
status.as_ref().map(|s| s.queue_client()),
|
||||
)
|
||||
.await;
|
||||
|
||||
|
|
|
|||
Loading…
Reference in a new issue