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

derive backend hosts only from the operator's save; the backfill only drops

Browse source
This commit is contained in:
atlas 2026-10-29 02:26:47 +01:00 • committed by mara
commit 0ad1dad8b5
2 changed files with 229 additions and 108 deletions
Download patch file Download diff file Expand all files Collapse all files

293
swarm-controller/src/backend_egress.rs
View file

@ -3,21 +3,22 @@
//! `ANTHROPIC_BASE_URL` points at a provider other than Anthropic reaches that
//! provider through its hive's proxy.
//!
//! The controller sees the environment where it is written (`backend_env`) and
//! in the store, so it derives here and a hive renders the entries like any
//! other. An entry is marked `derived`: [`follow_env`] replaces an agent's
//! derived entries on every write or delete of its environment, and
//! [`spawn_backfill`] reconciles all of them from the stored environments at
//! start. Each host goes through `egress_allow::checked`, the check an
//! operator's own entry gets.
//! An entry is marked `derived` and a hive renders it like any other. Only the
//! operator's save or delete of the environment through the controller creates
//! one ([`follow_env`]): the agent can create its own `backend-env` leaf while
//! it is empty, so what the store holds is not evidence that an operator chose
//! it. [`spawn_backfill`] therefore only drops: it keeps a derived entry while
//! the stored environment still names its exact host, and never adds one. Each
//! host goes through `egress_allow::checked`, the check an operator's own
//! entry gets.
//!
//! Only the store's copy is read. An agent still on the hand-placed
//! `backendEnvironmentFile` alone has no derived entry. The opencode preset's
//! provider URL (`acp.opencode.provider.baseUrl`) is host configuration, not
//! part of the environment, so it is never derived either.
//! An agent still on the hand-placed `backendEnvironmentFile` alone has no
//! derived entry, and neither does an environment saved before this existed
//! until the operator saves it once more. The opencode preset's provider URL
//! (`acp.opencode.provider.baseUrl`) is host configuration, not part of the
//! environment, so it is never derived either.
use std::collections::{BTreeMap, BTreeSet};
use std::sync::Arc;
use std::time::Duration;
use problem_details::ProblemDetails;
@ -28,7 +29,6 @@ use crate::egress_allow::{base_domains, checked, modify};
use crate::linked_accounts::AccountStore;
use crate::routes::{AppState, error_problem, wanted_error_status};
use crate::schedules::queue;
use crate::wanted::WantedWriter;
/// `ANTHROPIC_BASE_URL` and every variable named `<PROVIDER>_BASE_URL`, which
/// is where a provider's address sits for an ACP agent that loads the same
@ -90,7 +90,18 @@ fn host_of(url: &str) -> Option<&str> {
/// a name the base list already allows) is logged by variable name and left
/// out. The value is never logged: a URL can carry credentials.
fn derive(agent: &str, env: &str, base: &[String]) -> Vec<Entry> {
let mut entries = BTreeSet::new();
let mut entries: Vec<Entry> = derive_named(agent, env, base)
.into_iter()
.map(|(_, entry)| entry)
.collect();
entries.sort();
entries.dedup();
entries
}
/// [`fn@derive`] with the variable each entry came from, one per variable.
fn derive_named<'a>(agent: &str, env: &'a str, base: &[String]) -> Vec<(&'a str, Entry)> {
let mut entries = Vec::new();
for (name, value) in url_assignments(env) {
let Some(host) = host_of(value) else {
tracing::info!(%agent, variable = name, "backend egress: not an http(s) URL; no entry");
@ -102,24 +113,25 @@ fn derive(agent: &str, env: &str, base: &[String]) -> Vec<Entry> {
derived: false,
};
match checked(&request, base) {
Ok(entry) => {
entries.insert(Entry {
Ok(entry) => entries.push((
name,
Entry {
derived: true,
..entry
});
}
},
)),
Err(why) => {
tracing::info!(%agent, variable = name, %why, "backend egress: no entry for this host");
}
}
}
entries.into_iter().collect()
entries
}
/// `list` with, for each agent in `desired`, its derived entries replaced by
/// the given ones. Entries of other agents and operator-added entries are
/// untouched. A full list takes no more derived entries.
fn reconciled(mut list: AllowList, desired: &BTreeMap<String, Vec<Entry>>) -> AllowList {
fn replaced(mut list: AllowList, desired: &BTreeMap<String, Vec<Entry>>) -> AllowList {
list.entries
.retain(|e| !(e.derived && e.agent.as_ref().is_some_and(|a| desired.contains_key(a))));
for entry in desired.values().flatten() {
@ -158,7 +170,7 @@ pub(crate) async fn follow_env(
)]);
let updated = async {
let client = queue(state).map_err(|(_, why)| anyhow::anyhow!(why))?;
modify(&client, |list| reconciled(list, &desired)).await
modify(&client, |list| replaced(list, &desired)).await
}
.await;
match updated {
@ -178,9 +190,26 @@ pub(crate) async fn follow_env(
Ok(())
}
/// Reconcile every agent's derived entries from the environments in `store`:
/// the live agents, and every agent that has a derived entry now. An agent
/// whose environment cannot be read keeps its entries.
/// `list` without the derived entries of the agents in `still_named` whose host
/// is not in that agent's set. Nothing is added; entries of agents not in the
/// map and operator-added entries are untouched.
fn pruned(mut list: AllowList, still_named: &BTreeMap<String, BTreeSet<String>>) -> AllowList {
list.entries.retain(|e| {
let Some(hosts) = e.agent.as_ref().and_then(|a| still_named.get(a)) else {
return true;
};
!e.derived || hosts.contains(&e.domain)
});
list
}
/// Drop the derived entries whose host the stored environment no longer names.
///
/// Only the agents that have a derived entry are looked at. An agent whose
/// stored environment is gone loses all of them; one whose environment cannot
/// be read keeps them. A host the stored environment names that is not
/// already derived is left out and logged by variable name, since whoever
/// wrote the store may be the agent itself.
///
/// Answers how many agents could not be read.
///
@ -189,28 +218,45 @@ pub(crate) async fn follow_env(
pub(crate) async fn reconcile(
client: &async_nats::Client,
store: &impl AccountStore,
live: &BTreeSet<String>,
) -> anyhow::Result<usize> {
let mut agents = live.clone();
if let Some(list) = crate::egress_allow::current(client).await? {
agents.extend(
list.entries
.iter()
.filter(|e| e.derived)
.filter_map(|e| e.agent.clone()),
);
let Some(current) = crate::egress_allow::current(client).await? else {
return Ok(0);
};
let mut derived: BTreeMap<String, BTreeSet<String>> = BTreeMap::new();
for entry in current.entries.iter().filter(|e| e.derived) {
if let Some(agent) = &entry.agent {
derived
.entry(agent.clone())
.or_default()
.insert(entry.domain.clone());
}
}
let base = base_domains();
let mut desired = BTreeMap::new();
let mut still_named = BTreeMap::new();
let mut unreadable = 0;
for agent in agents {
for (agent, kept) in derived {
let Ok(path) = backend_env::env_path(&agent) else {
continue;
};
match store.read_optional::<BackendEnv>(&path).await {
Ok(stored) => {
let entries = stored.map_or_else(Vec::new, |s| derive(&agent, &s.env, &base));
desired.insert(agent, entries);
let named = stored.map_or_else(Vec::new, |s| {
derive_named(&agent, &s.env, &base)
.into_iter()
.map(|(variable, entry)| (variable.to_owned(), entry.domain))
.collect()
});
for (variable, domain) in &named {
if !kept.contains(domain) {
tracing::warn!(
%agent,
%variable,
"backend egress: the agent's stored env names a host not derived \
(only an operator save derives)"
);
}
}
still_named.insert(agent, named.into_iter().map(|(_, d)| d).collect());
}
Err(e) => {
tracing::warn!(%agent, error = %e, "backend egress: reading the backend env failed; its entries stay");
@ -218,34 +264,25 @@ pub(crate) async fn reconcile(
}
}
}
if modify(client, |list| reconciled(list, &desired)).await? {
let derived: usize = desired.values().map(Vec::len).sum();
if modify(client, |list| pruned(list, &still_named)).await? {
tracing::info!(
agents = desired.len(),
derived,
"backend egress: backfilled the allow list"
agents = still_named.len(),
"backend egress: dropped derived entries the stored env no longer names"
);
}
Ok(unreadable)
}
/// Run [`reconcile`] once the queue and the store answer, retrying with a
/// growing wait until a pass reads every agent. One pass at start: the
/// routes keep the entries after it.
pub(crate) fn spawn_backfill(
client: async_nats::Client,
wanted: Arc<WantedWriter>,
hives: Vec<String>,
) {
/// growing wait until a pass reads every agent that has an entry.
pub(crate) fn spawn_backfill(client: async_nats::Client) {
tokio::spawn(async move {
let mut wait = Duration::from_secs(5);
loop {
let pass = async {
swarm_queue_client::ensure_connected(&client)?;
let store = crate::store::connect().await?;
let live = crate::subagent_term::live_agents(&wanted, &hives).await;
anyhow::ensure!(!live.is_empty(), "no live agent is declared yet");
reconcile(&client, &store, &live).await
reconcile(&client, &store).await
}
.await;
match pass {
@ -266,6 +303,8 @@ pub(crate) fn spawn_backfill(
#[cfg(test)]
mod tests {
use std::sync::Arc;
use super::*;
use crate::agent_limits::tests::declared;
use crate::linked_accounts::tests::FakeStore;
@ -376,7 +415,7 @@ mod tests {
}
#[test]
fn reconciling_replaces_only_the_named_agents_derived_entries() {
fn replacing_replaces_only_the_named_agents_derived_entries() {
let list = AllowList {
entries: vec![
derived("old.example.com", "atlas"),
@ -389,7 +428,7 @@ mod tests {
"atlas".to_owned(),
vec![derived("new.example.com", "atlas")],
)]);
let after = reconciled(list, &desired);
let after = replaced(list, &desired);
assert_eq!(
after.entries,
[
@ -520,47 +559,126 @@ mod tests {
assert_eq!(list_of(&client).await.len(), 1);
}
const OPERATOR: &str = "ANTHROPIC_BASE_URL=https://llm.example.com/v1\n";
fn leaf(agent: &str) -> String {
format!("swarm/agents/{agent}/backend-env")
}
async fn write_leaf(store: &FakeStore, agent: &str, env: &str) {
store
.write(
&leaf(agent),
&BackendEnv {
env: env.to_owned(),
},
)
.await
.expect("written");
}
/// The agent creates its own leaf while it is empty, so a stored value is
/// not evidence of an operator's choice: the backfill adds nothing for it.
#[tokio::test]
async fn the_backfill_derives_from_stored_envs_and_a_second_pass_changes_nothing() {
async fn a_leaf_created_outside_the_route_yields_no_entry() {
let nats = TestNats::start();
let client = nats.client().await;
let store = Arc::new(
FakeStore::default()
.with(
"swarm/agents/atlas/backend-env",
json!({"env": "ANTHROPIC_BASE_URL=https://llm.example.com/v1\n"}),
)
.with(
"swarm/agents/iris/backend-env",
json!({"env": "ANTHROPIC_API_KEY=sk-live\n"}),
),
);
let live: BTreeSet<String> = ["atlas", "iris", "scribe"].map(str::to_owned).into();
let store = Arc::new(FakeStore::default());
let state = swarm(&client, &store).await;
write_leaf(
&store,
"atlas",
"ANTHROPIC_BASE_URL=https://evil.example.net\n",
)
.await;
assert_eq!(reconcile(&client, &*store, &live).await.expect("pass"), 0);
// With no list at all, the pass creates nothing.
assert_eq!(reconcile(&client, &*store).await.expect("pass"), 0);
assert!(
swarm_queue_client::egress_allow::open(&client)
.await
.expect("the queue answers")
.is_none()
);
// With a list that holds only an operator's entry, nothing is added.
modify(&client, |mut list| {
list.entries.push(manual("github.com", "atlas"));
list
})
.await
.expect("seed");
assert_eq!(reconcile(&client, &*store).await.expect("pass"), 0);
assert_eq!(list_of(&client).await, [manual("github.com", "atlas")]);
// Control: the operator's save of the same text derives the host.
put(
&state,
"atlas",
"ANTHROPIC_BASE_URL=https://evil.example.net\n",
)
.await;
assert_eq!(
list_of(&client).await,
[derived("llm.example.com", "atlas")]
[
derived("evil.example.net", "atlas"),
manual("github.com", "atlas")
]
);
}
let published = published(&client).await;
assert_eq!(reconcile(&client, &*store, &live).await.expect("again"), 0);
assert_eq!(
self::published(&client).await,
published,
"the second pass writes nothing"
);
/// A pass keeps an entry whose exact host the stored env still names,
/// drops one it no longer names, and never adds the new host.
#[tokio::test]
async fn the_backfill_keeps_a_named_host_drops_a_stale_one_and_adds_nothing() {
let nats = TestNats::start();
let client = nats.client().await;
let store = Arc::new(FakeStore::default());
let state = swarm(&client, &store).await;
put(&state, "atlas", OPERATOR).await;
let entry = derived("llm.example.com", "atlas");
// An agent that stops being live but keeps its leaf keeps its entry;
// one whose leaf is gone loses it.
let gone: BTreeSet<String> = BTreeSet::new();
reconcile(&client, &*store, &gone).await.expect("pass");
assert_eq!(
list_of(&client).await,
[derived("llm.example.com", "atlas")]
);
let empty = FakeStore::default();
reconcile(&client, &empty, &gone).await.expect("pass");
// Control: unchanged content keeps the entry, and a second pass writes
// nothing.
assert_eq!(reconcile(&client, &*store).await.expect("pass"), 0);
assert_eq!(list_of(&client).await, [entry]);
let before = published(&client).await;
reconcile(&client, &*store).await.expect("again");
assert_eq!(published(&client).await, before);
// The stored text changed outside the route: the old host goes, the
// new one does not come.
write_leaf(
&store,
"atlas",
"ANTHROPIC_BASE_URL=https://other.example.net\n",
)
.await;
reconcile(&client, &*store).await.expect("pass");
assert_eq!(list_of(&client).await, []);
}
/// The leaf is deleted outside the controller and created again by the
/// agent with another host before the pass: the old entry goes and the new
/// host is not derived. A leaf that is simply gone drops its entries too.
#[tokio::test]
async fn a_leaf_deleted_and_recreated_by_the_agent_loses_the_entry_and_gains_none() {
let nats = TestNats::start();
let client = nats.client().await;
let state = swarm(&client, &Arc::default()).await;
put(&state, "atlas", OPERATOR).await;
put(&state, "iris", OPERATOR).await;
assert_eq!(list_of(&client).await.len(), 2, "control: both derived");
let recreated = FakeStore::default();
write_leaf(
&recreated,
"atlas",
"ANTHROPIC_BASE_URL=https://evil.example.net\n",
)
.await;
// iris's leaf is gone, atlas's was created again by atlas.
reconcile(&client, &recreated).await.expect("pass");
assert_eq!(list_of(&client).await, []);
}
@ -576,8 +694,7 @@ mod tests {
)
.await;
let live: BTreeSet<String> = ["atlas"].map(str::to_owned).into();
let unreadable = reconcile(&client, &FakeStore::denying(), &live)
let unreadable = reconcile(&client, &FakeStore::denying())
.await
.expect("pass");
assert_eq!(unreadable, 1);

17
swarm-controller/src/startup.rs
View file

@ -387,7 +387,7 @@ pub(crate) fn spawn_per_agent_upkeep(
spawn_subagent_streams(status, hives);
spawn_schedule_reconcile(status, hives);
spawn_agent_permissions(status, hives);
spawn_backend_egress_backfill(status, hives);
spawn_backend_egress_backfill(status);
crate::attic_copy::spawn(hives.iter().map(|h| h.name.clone()).collect());
crate::proxy_credential::spawn(hives.iter().map(|h| h.name.clone()).collect());
if let Some(status) = status {
@ -405,15 +405,12 @@ fn spawn_agent_permissions(status: Option<&Arc<status::StatusReader>>, hives: &[
agent_permissions::spawn(status.queue_client(), wanted, hives);
}
/// Derive each agent's allowed-domain entries from the backend environments in
/// the store when a swarm queue is wired up. Same shape as
/// [`spawn_subagent_streams`].
fn spawn_backend_egress_backfill(status: Option<&Arc<status::StatusReader>>, hives: &[HiveEntry]) {
let (Some(status), Some(wanted)) = (status, wanted_writer(status)) else {
return;
};
let hives = hives.iter().map(|h| h.name.clone()).collect();
crate::backend_egress::spawn_backfill(status.queue_client(), wanted, hives);
/// Drop the allowed-domain entries derived from backend environments that no
/// longer name their host, when a swarm queue is wired up.
fn spawn_backend_egress_backfill(status: Option<&Arc<status::StatusReader>>) {
if let Some(status) = status {
crate::backend_egress::spawn_backfill(status.queue_client());
}
}
/// Start rewriting every agent's schedule key from the definitions when a