vinzenz/hyperhive
vinzenz/hyperhive
Watch
0
0
Fork
You've already forked hyperhive
 0
Code Activity

Compare commits

base: vinzenz:49015eb55350a92182874c357aff8e3e7c165a9b
Branches Tags
vinzenz:main
..
compare: vinzenz:a2b4320cc4197ffdbdb7c7799eb2265124fb281d
Branches Tags
vinzenz:main
17 changed files with 36 additions and 735 deletions
Download patch file Download diff file Expand all files Collapse all files

3
docs/getting-started/setup.md
View file

@ -89,9 +89,6 @@ swarmctl forge make-admin mara
If the forge already has a local account with your username, the first
SSO login asks for its forge password once, to link the two.
Agents learn your pronouns from the **Pronouns** field of your forge profile
settings. Leave it empty and agents refer to you by name only.
## 4 · Ruth
hive-c0re creates no agent at startup. You create ruth, the manager agent,

16
docs/turn-loop/claude-invocation.md
View file

@ -255,21 +255,13 @@ needs to `chown` a bind mount), once at startup:
`hive-agent/prompts/system.md` by `hive-agent`'s `prompt::render`:
HTML-comment markers (`<!-- role:agent -->...<!-- /role:agent -->`,
same for `role:manager`) gate the role-specific blocks; both flavors
share everything else. Five placeholders are then
share everything else. Four placeholders are then
substituted: `{label}` (short agent name), `{qualified_label}`
(hive-qualified `name@domain` form),
`{hive_identity}` (for example `` on hive `pr1ma` ``; empty when
`HYPERHIVE_HIVE_NAME` is unset), `{swarm_identity}` (same
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 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
pronoun change reaches an agent at its next harness start, since
the prompt renders once per start.
`HYPERHIVE_HIVE_NAME` is unset), and `{swarm_identity}` (same
shape for the swarm). The template names no pronouns for the
operator, and tells agents to call them the operator.
When you enable `services.hyperhive.agent.docs.enable`, `HIVE_DOCS_DIR` is present
in the environment and `render()` appends a one-sentence pointer
telling the agent where the docs live.

10
hive-agent/Cargo.toml
View file

@ -38,16 +38,14 @@ rusqlite.workspace = true
schemars.workspace = true
serde.workspace = true
serde_json.workspace = true
# `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.
# `agent-schedules` and `agent-inference`: the schedule ticker and the
# inference poller each read their own key 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 = [
"agent-schedules",
"agent-inference",
"agent-permissions",
"operator-profiles",
] }
# This agent's own queue secret, read from the store under its own certificate.
swarm-secret-client.workspace = true

2
hive-agent/prompts/system.md
View file

@ -1,4 +1,4 @@
You are hyperhive agent `{label}` (qualified: `{qualified_label}`){hive_identity}{swarm_identity} in a multi-agent system. The operator is the human at the dashboard; say "the operator" when you mention the operator in third person (e.g. when relaying to a peer or the manager). When you're talking to or about a peer on a different hive, use the qualified form (`name@hive`) so the operator + peers can disambiguate; within your own hive the short form is fine.{operators}
You are hyperhive agent `{label}` (qualified: `{qualified_label}`){hive_identity}{swarm_identity} in a multi-agent system. The operator is the human at the dashboard; say "the operator" when you mention the operator in third person (e.g. when relaying to a peer or the manager). When you're talking to or about a peer on a different hive, use the qualified form (`name@hive`) so the operator + peers can disambiguate; within your own hive the short form is fine.
Tools (hyperhive surface). Full signature + behavior for each comes from the tool's own MCP description (you already received it via the MCP tool schema) — this is just the map of what exists and which ones are gated, so you know where to look:

163
hive-agent/src/prompt.rs
View file

@ -5,17 +5,11 @@
//! `docs/turn-loop/claude-invocation.md::On-boot files` (`claude-system-prompt.md`).
use std::path::{Path, PathBuf};
use std::time::Duration;
use anyhow::{Context, Result};
use swarm_queue_client::operator_profiles::{self, OperatorProfile};
/// How long [`write_system_prompt`] waits for the operator list before
/// rendering without it.
const OPERATORS_TIMEOUT: Duration = Duration::from_secs(5);
/// Assemble the system prompt for a given label + optional hive /
/// swarm display names + the swarm's operators. Pure function — no I/O. Splits out from
/// swarm display names. Pure function — no I/O. Splits out from
/// [`write_system_prompt`] so the marker logic + substitution is unit-testable
/// in isolation. The caller supplies the template body so tests can pass an
/// inline fixture and production reads it once at harness startup via
@ -30,7 +24,6 @@ pub fn render(
hive_name: Option<&str>,
swarm_name: Option<&str>,
docs_dir: Option<&str>,
operators: &[OperatorProfile],
) -> String {
let body = filter_role_blocks(template, "agent");
let qualified = crate::identity::qualify(label);
@ -44,8 +37,7 @@ pub fn render(
.replace("{label}", label)
.replace("{qualified_label}", &qualified)
.replace("{hive_identity}", &hive_identity)
.replace("{swarm_identity}", &swarm_identity)
.replace("{operators}", &operators_line(operators));
.replace("{swarm_identity}", &swarm_identity);
// When the reference docs are mounted in-container (`services.hyperhive.agent.docs.enable`
// wires `HIVE_DOCS_DIR` + `claude --add-dir`), append a single pointer
// sentence so the agent knows they exist. Additive: it doesn't replace the
@ -61,63 +53,6 @@ pub fn render(
}
}
/// The `{operators}` substitution: a paragraph naming each operator, with their
/// pronouns where set, or nothing when there are none.
fn operators_line(operators: &[OperatorProfile]) -> String {
if operators.is_empty() {
return String::new();
}
let names: Vec<String> = operators
.iter()
.map(|o| match &o.pronouns {
Some(pronouns) => format!("{} ({pronouns})", o.name),
None => o.name.clone(),
})
.collect();
format!("\n\nOperators: {}.", names.join(", "))
}
/// The swarm's operators, or none when the operator-profile bucket cannot be
/// read within [`OPERATORS_TIMEOUT`]: the prompt then reads as it does with no
/// operator list.
async fn operators() -> Vec<OperatorProfile> {
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<Vec<OperatorProfile>, Box<dyn std::error::Error + Send + Sync>> {
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`.
async fn within_timeout<E: std::fmt::Display>(
timeout: Duration,
read: impl Future<Output = std::result::Result<Vec<OperatorProfile>, E>>,
) -> Vec<OperatorProfile> {
match tokio::time::timeout(timeout, read).await {
Ok(Ok(operators)) => operators,
Ok(Err(e)) => {
tracing::warn!(error = %e, "prompt: operator profiles unreadable; rendering without them");
Vec::new()
}
Err(_) => {
tracing::warn!(
timeout_s = timeout.as_secs(),
"prompt: no operator profiles within the timeout; rendering without them"
);
Vec::new()
}
}
}
/// Walk `template` line-by-line, stripping `<!-- role:X -->` /
/// `<!-- /role:X -->` marker lines and suppressing the lines inside a block
/// unless `X == target`. A close marker ends the current block. Production
@ -202,7 +137,6 @@ pub async fn write_system_prompt(_socket: &Path, label: &str) -> Result<PathBuf>
hive_name.as_deref(),
swarm_name.as_deref(),
docs_dir.as_deref(),
&operators().await,
);
let path = parent.join("claude-system-prompt.md");
tokio::fs::write(&path, body).await?;
@ -300,7 +234,7 @@ shared closer
#[test]
fn render_substitutes_label_and_names_no_operator_pronouns() {
// Real template's first agent line.
let rendered = render(&PRODUCTION_TEMPLATE, "alice", None, None, None, &[]);
let rendered = render(&PRODUCTION_TEMPLATE, "alice", None, None, None);
assert!(rendered.contains("hyperhive agent `alice`"));
assert!(rendered.contains("say \"the operator\""));
assert!(!rendered.contains("pronoun"));
@ -311,7 +245,7 @@ shared closer
#[test]
fn render_no_role_markers_in_output() {
// No raw role markers should survive into the rendered prompt.
let rendered = render(&PRODUCTION_TEMPLATE, "alice", None, None, None, &[]);
let rendered = render(&PRODUCTION_TEMPLATE, "alice", None, None, None);
assert!(!rendered.contains("<!-- role:"));
assert!(!rendered.contains("<!-- /role:"));
// Shared tools appear.
@ -321,13 +255,13 @@ shared closer
#[test]
fn render_uses_agent_opener() {
let rendered = render(&PRODUCTION_TEMPLATE, "alice", None, None, None, &[]);
let rendered = render(&PRODUCTION_TEMPLATE, "alice", None, None, None);
assert!(rendered.starts_with("You are hyperhive agent"));
}
#[test]
fn render_points_agents_at_the_work_dir() {
let rendered = render(&PRODUCTION_TEMPLATE, "alice", None, None, None, &[]);
let rendered = render(&PRODUCTION_TEMPLATE, "alice", None, None, None);
assert!(rendered.contains("`/agents/alice/work/`"), "{rendered}");
}
@ -346,7 +280,7 @@ You are hyperhive agent `{label}` (qualified: `{qualified_label}`){hive_identity
#[test]
fn render_substitutes_hive_identity_when_set() {
let rendered = render(IDENTITY_FIXTURE, "alice", Some("pr1ma"), None, None, &[]);
let rendered = render(IDENTITY_FIXTURE, "alice", Some("pr1ma"), None, None);
assert!(rendered.contains("on hive `pr1ma`"), "{rendered}");
// swarm clause stays absent when only hive is set.
assert!(!rendered.contains("in swarm"));
@ -357,14 +291,7 @@ You are hyperhive agent `{label}` (qualified: `{qualified_label}`){hive_identity
#[test]
fn render_substitutes_swarm_identity_when_set() {
let rendered = render(
IDENTITY_FIXTURE,
"ruth",
None,
Some("constellat1on"),
None,
&[],
);
let rendered = render(IDENTITY_FIXTURE, "ruth", None, Some("constellat1on"), None);
assert!(rendered.contains("in swarm `constellat1on`"));
assert!(!rendered.contains("on hive"));
}
@ -377,7 +304,6 @@ You are hyperhive agent `{label}` (qualified: `{qualified_label}`){hive_identity
Some("pr1ma"),
Some("constellat1on"),
None,
&[],
);
// Order: hive then swarm, both inline before "in a multi-agent
// system" — keeps the opener grammar intact.
@ -388,7 +314,7 @@ You are hyperhive agent `{label}` (qualified: `{qualified_label}`){hive_identity
fn render_omits_identity_when_unset() {
// None / None must round-trip the non-identity opener verbatim
// — single-hive deployments see zero diff.
let rendered = render(IDENTITY_FIXTURE, "alice", None, None, None, &[]);
let rendered = render(IDENTITY_FIXTURE, "alice", None, None, None);
assert!(!rendered.contains("on hive"));
assert!(!rendered.contains("in swarm"));
assert!(!rendered.contains("{hive_identity}"));
@ -401,7 +327,7 @@ You are hyperhive agent `{label}` (qualified: `{qualified_label}`){hive_identity
// through `identity::hive_name()` as None (the accessor
// filters empty), but `render` should still no-op on a
// direct `Some("")` from a test fixture or a future caller.
let rendered = render(IDENTITY_FIXTURE, "alice", Some(""), Some(""), None, &[]);
let rendered = render(IDENTITY_FIXTURE, "alice", Some(""), Some(""), None);
assert!(!rendered.contains("on hive"));
assert!(!rendered.contains("in swarm"));
}
@ -414,7 +340,6 @@ You are hyperhive agent `{label}` (qualified: `{qualified_label}`){hive_identity
None,
None,
Some("/run/hive-docs"),
&[],
);
// Key on a phrase unique to the pointer sentence — "reference docs"
// alone also appears in the /knowledge blurb of the base template.
@ -427,77 +352,11 @@ You are hyperhive agent `{label}` (qualified: `{qualified_label}`){hive_identity
#[test]
fn render_no_docs_pointer_when_docs_dir_absent_or_empty() {
for docs in [None, Some("")] {
let rendered = render(&PRODUCTION_TEMPLATE, "alice", None, None, docs, &[]);
let rendered = render(&PRODUCTION_TEMPLATE, "alice", None, None, docs);
assert!(
!rendered.contains("mounted read-only at"),
"docs pointer must not appear for {docs:?}:\n{rendered}"
);
}
}
fn operator(name: &str, pronouns: Option<&str>) -> OperatorProfile {
OperatorProfile {
name: name.to_owned(),
pronouns: pronouns.map(str::to_owned),
}
}
const OPERATORS_FIXTURE: &str = "say \"the operator\".{operators}\n\nnext paragraph\n";
#[test]
fn render_lists_operators_with_their_pronouns() {
let operators = [operator("mara", Some("she/her")), operator("sol", None)];
let rendered = render(OPERATORS_FIXTURE, "alice", None, None, None, &operators);
assert_eq!(
rendered,
"say \"the operator\".\n\nOperators: mara (she/her), sol.\n\nnext paragraph\n"
);
let rendered = render(&PRODUCTION_TEMPLATE, "alice", None, None, None, &operators);
assert!(
rendered.contains("Operators: mara (she/her), sol."),
"{rendered}"
);
}
/// No operators renders the template as it reads without the placeholder,
/// byte for byte.
#[test]
fn render_with_no_operators_is_the_template_without_the_placeholder() {
let rendered = render(OPERATORS_FIXTURE, "alice", None, None, None, &[]);
assert_eq!(rendered, "say \"the operator\".\n\nnext paragraph\n");
assert!(PRODUCTION_TEMPLATE.contains("{operators}"));
assert_eq!(
render(&PRODUCTION_TEMPLATE, "alice", None, None, None, &[]),
render(
&PRODUCTION_TEMPLATE.replace("{operators}", ""),
"alice",
None,
None,
None,
&[]
)
);
}
/// An unreachable or failing bucket reads as no operators, so the prompt
/// renders as it does with none.
#[tokio::test(start_paused = true)]
async fn an_unread_operator_list_is_empty() {
let hangs = std::future::pending::<std::result::Result<Vec<OperatorProfile>, String>>();
let start = tokio::time::Instant::now();
assert_eq!(within_timeout(OPERATORS_TIMEOUT, hangs).await, []);
assert!(
start.elapsed() < OPERATORS_TIMEOUT * 2,
"{:?}",
start.elapsed()
);
let fails = async { Err::<Vec<OperatorProfile>, _>("no bucket".to_owned()) };
assert_eq!(within_timeout(OPERATORS_TIMEOUT, fails).await, []);
// Control: a read that answers in time is what comes back.
let answers = async { Ok::<_, String>(vec![operator("mara", Some("she/her"))]) };
assert_eq!(
within_timeout(OPERATORS_TIMEOUT, answers).await,
[operator("mara", Some("she/her"))]
);
}
}

8
nix/host-modules/hive-options-removed.nix
View file

@ -3,8 +3,7 @@
#
# A definition of any of them is ignored and produces a warning, so a host
# that still sets one keeps evaluating. The agents' system prompt
# (hive-agent/prompts/system.md) takes each operator's pronouns from their
# forge profile, the
# (hive-agent/prompts/system.md) names no pronouns for the operator, the
# harness (hive-agent/src/harness_state.rs) holds the per-model context
# windows it uses until a turn reports the model's own, every OTEL exporter
# exports every 60 seconds (claude at its SDK default, the rest by a fixed
@ -45,8 +44,9 @@ in
config.warnings =
removed "c0re.operatorPronouns" cfg.c0re.operatorPronouns ''
agents' system prompts take each operator's pronouns from the Pronouns
field of that operator's forge profile.
agents' system prompts name no pronouns for the operator and say "the
operator". Pronouns will come from a per-operator profile setting, which
is an open hyperhive forge issue.
''
++ removed "c0re.contextWindowTokens" cfg.c0re.contextWindowTokens ''
the agent harness uses the context window the model reports after each

4
nix/host-modules/swarm-nats.nix
View file

@ -905,10 +905,6 @@ 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 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}"
];
# The store identity arrives by `LoadCredential` and is named

2
swarm-controller/Cargo.toml
View file

@ -150,8 +150,6 @@ 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

8
swarm-controller/README.md
View file

@ -71,14 +71,6 @@ 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, 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.
- **Alerts** — the swarm Grafana's current alerts (`GET /api/alerts`), read

1
swarm-controller/src/forge.rs
View file

@ -53,7 +53,6 @@ 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;

287
swarm-controller/src/forge/operator_profiles.rs
View file

@ -1,287 +0,0 @@
//! 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 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::sync::Arc;
use anyhow::{Context, Result};
use async_nats::jetstream::kv::Store;
use swarm_queue_client::operator_profiles::{self, KEY, 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 {
/// The list differed from the stored one and was written.
written: bool,
/// Operators whose profile could not be read, left as stored.
kept: usize,
}
/// Bring [`KEY`] in `store` to the list of `operators`, each with its
/// pronouns as `forge` reports them, sorted by name.
///
/// # Errors
/// 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<Outcome> {
let stored: Vec<OperatorProfile> = store
.get(KEY)
.await
.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 {
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 entry"
);
outcome.kept += 1;
stored
.iter()
.find(|o| o.name == *name)
.and_then(|o| o.pronouns.clone())
}
};
list.push(OperatorProfile {
name: name.clone(),
pronouns,
});
}
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
.put(KEY, value.into())
.await
.context("writing the operator list")?;
outcome.written = true;
}
Ok(outcome)
}
/// 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<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 => tracing::info!(
kept = outcome.kept,
"operator profiles: operator list 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 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 {
OperatorProfile {
name: name.to_owned(),
pronouns: pronouns.map(str::to_owned),
}
}
#[tokio::test]
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(&["sol", "mara"]), &forge, &store)
.await
.unwrap();
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 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!(outcome.written);
assert_eq!(stored(&store).await, [profile("mara", Some("she/her"))]);
}
#[tokio::test]
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();
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.
// `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, 2);
assert_eq!(
stored(&store).await,
[
profile("mara", Some("she/her")),
profile("ren", None),
profile("sol", Some("he/him"))
]
);
}
/// 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, []);
}
}

14
swarm-controller/src/main.rs
View file

@ -2863,20 +2863,14 @@ 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, and the operators' profiles when the bridge
/// and the queue are wired up too. Lifted out of `main` for
/// once there is a forge client. 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();
@ -3423,8 +3417,6 @@ 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,
@ -3435,7 +3427,7 @@ async fn start_forge(
}
};
forge::enable(url, forge_client, move |client| {
spawn_forge_workers(&jobq, Arc::clone(&client), &hives, auth, queue);
spawn_forge_workers(&jobq, Arc::clone(&client), &hives);
config_prs.get_or_init(|| config_pr::spawn(Arc::clone(&client)));
register_swarm_webhooks(client, webhook_secret);
})
@ -3557,8 +3549,6 @@ async fn main() -> Result<()> {
Arc::clone(&jobq),
hives.clone(),
webhook_secret.clone(),
auth.clone(),
status.as_ref().map(|s| s.queue_client()),
)
.await;

2
swarm-nats-auth/src/agent_token.rs
View file

@ -118,8 +118,6 @@ mod tests {
"$SWARM.term.atlas".to_owned(),
"$SWARM.agent-state.atlas".to_owned(),
"$KV.agent-icons.atlas".to_owned(),
"$JS.API.DIRECT.GET.KV_operator-profiles.$KV.operator-profiles.operators"
.to_owned(),
]
);
}

87
swarm-nats-auth/src/policy.rs
View file

@ -173,22 +173,12 @@ impl Policy {
/// must already be a single `[A-Za-z0-9_-]` segment, which the token parse
/// guarantees, so it cannot widen a subject with `.`, `*` or `>`.
pub fn agent_token_permissions(&self, agent: &str) -> Option<Permissions> {
let mut publish: Vec<String> = self
let publish: Vec<String> = self
.agent_token_subjects
.iter()
.map(|s| s.replace(AGENT_PLACEHOLDER, agent))
.collect();
if publish.is_empty() {
return None;
}
// 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.is_empty()).then(|| Permissions {
publish,
subscribe: Some(vec![
format!("{}.>", swarm_queue_client::agent_inbox_prefix(agent)),
@ -734,18 +724,6 @@ mod tests {
.expect("the default policy is valid")
}
/// 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<String> {
let shared = swarm_queue_client::operator_profiles::direct_get_subject();
grant
.publish
.iter()
.filter(|s| **s != shared)
.cloned()
.collect()
}
#[test]
fn the_reader_can_open_and_list_the_agent_status_bucket() {
let subjects = policy().reader_subjects();
@ -1900,8 +1878,6 @@ mod tests {
vec![
"$SWARM.term.atlas".to_owned(),
"$SWARM.agent-state.atlas".to_owned(),
"$JS.API.DIRECT.GET.KV_operator-profiles.$KV.operator-profiles.operators"
.to_owned(),
]
);
}
@ -1915,12 +1891,12 @@ mod tests {
.with_agent_token_subjects(vec!["$KV.agent-icons.{agent}".to_owned()])
.expect("a per-agent template is valid");
let atlas = p.agent_token_permissions("atlas").expect("configured");
assert_eq!(templated(&atlas), vec![subject("atlas")]);
assert_eq!(atlas.publish, vec![subject("atlas")]);
assert!(!atlas.publish.contains(&subject("argus")));
// Control: the same template does grant argus its own key, so atlas
// lacking it is atlas's grant being scoped.
let argus = p.agent_token_permissions("argus").expect("configured");
assert_eq!(templated(&argus), vec![subject("argus")]);
assert_eq!(argus.publish, vec![subject("argus")]);
}
/// The subagent template as `swarm-nats.nix` spells it expands to the
@ -1934,10 +1910,10 @@ mod tests {
.with_agent_token_subjects(vec!["$SWARM.term.{agent}.sub.>".to_owned()])
.expect("a per-agent template is valid");
let g = p.agent_token_permissions("atlas").expect("configured");
assert_eq!(templated(&g), vec![stream_subjects("atlas")]);
assert_eq!(g.publish, vec![stream_subjects("atlas")]);
assert!(subject("atlas", "scout").starts_with(g.publish[0].trim_end_matches('>')));
let argus = p.agent_token_permissions("argus").expect("configured");
assert_eq!(templated(&argus), vec![stream_subjects("argus")]);
assert_eq!(argus.publish, vec![stream_subjects("argus")]);
assert_ne!(g.publish, argus.publish);
let stream = stream_name("atlas");
@ -1965,11 +1941,11 @@ mod tests {
])
.expect("a per-agent template is valid");
let atlas = p.agent_token_permissions("atlas").expect("configured");
assert_eq!(templated(&atlas), vec![direct_get_subject("atlas")]);
assert_eq!(atlas.publish, vec![direct_get_subject("atlas")]);
// Control: argus's own expansion is a different subject, so atlas
// lacking it is the grant being scoped, not the template being empty.
let argus = p.agent_token_permissions("argus").expect("configured");
assert_eq!(templated(&argus), vec![direct_get_subject("argus")]);
assert_eq!(argus.publish, vec![direct_get_subject("argus")]);
assert!(!atlas.publish.contains(&direct_get_subject("argus")));
for refused in [
@ -2013,9 +1989,9 @@ mod tests {
])
.expect("a per-agent template is valid");
let atlas = p.agent_token_permissions("atlas").expect("configured");
assert_eq!(templated(&atlas), vec![direct_get_subject("atlas")]);
assert_eq!(atlas.publish, vec![direct_get_subject("atlas")]);
let argus = p.agent_token_permissions("argus").expect("configured");
assert_eq!(templated(&argus), vec![direct_get_subject("argus")]);
assert_eq!(argus.publish, vec![direct_get_subject("argus")]);
for refused in [
direct_get_subject("argus"),
@ -2046,47 +2022,4 @@ mod tests {
fn with_no_agent_token_subject_configured_an_agent_token_is_refused() {
assert!(policy().agent_token_permissions("atlas").is_none());
}
/// 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_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 refused in [
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}"),
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, &direct_get_subject()))
);
}
}

4
swarm-queue-client/Cargo.toml
View file

@ -44,10 +44,6 @@ 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`, 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"]

5
swarm-queue-client/src/lib.rs
View file

@ -227,11 +227,6 @@ pub mod agent_limits;
/// and reading need `agent-permissions`.
pub mod agent_permissions;
/// The bucket swarm-controller publishes each operator's name and forge-profile
/// 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;
/// The login flag in an agent's turn-state header. No bucket: it travels
/// on [`agent_state_subject`] only.
pub mod agent_login;

155
swarm-queue-client/src/operator_profiles.rs
View file

@ -1,155 +0,0 @@
//! 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; one who
//! leaves it drops out of the list on the controller's next pass.
//!
//! **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.
pub const BUCKET: &str = "operator-profiles";
/// The stream backing [`BUCKET`]. `create_key_value` prefixes `KV_`.
pub const STREAM: &str = "KV_operator-profiles";
/// 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.
pub name: String,
/// The pronouns set on the operator's forge profile. `None` when unset or
/// when the operator has no forge account yet.
#[serde(default, skip_serializing_if = "Option::is_none")]
pub pronouns: Option<String>,
}
/// Open the profile bucket for writing, creating it if nothing has yet.
///
/// **Controller-side only.** `history: 1`: an agent wants the current value.
#[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,
BUCKET,
"operator-profiles",
"Each operator's name and forge-profile pronouns, set by the swarm controller".to_owned(),
)
.await
}
/// 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<OperatorProfile>),
/// 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 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(
client: &async_nats::Client,
) -> Result<Read, Box<dyn std::error::Error + Send + Sync>> {
use async_nats::jetstream::stream::DirectGetErrorKind;
let js = async_nats::jetstream::new(client.clone());
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()));
}
Ok(Read::Current(serde_json::from_slice(&message.payload)?))
}
#[cfg(test)]
mod tests {
use super::*;
/// The names the auth responder grants agents from.
#[test]
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 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!(
json,
r#"[{"name":"mara","pronouns":"she/her"},{"name":"sol"}]"#
);
assert_eq!(
serde_json::from_str::<Vec<OperatorProfile>>(&json).unwrap(),
list
);
}
}