derive per-agent egress entries from operator-linked accounts
This commit is contained in:
parent
7eb9db47d6
commit
9193f37963
14 changed files with 852 additions and 38 deletions
|
|
@ -349,9 +349,9 @@ host. It never decrypts TLS, so it holds no signing key. It caches nothing.
|
|||
you add the entry, and `hive-c0re` checks each name again and skips one that
|
||||
fails, or that the base list or the canary covers, or that another entry of
|
||||
the same list covers. A name that resolves to a private address stays a 403.
|
||||
The controller also adds entries itself, and only from the operator's save
|
||||
of an agent's API-key environment in the swarm UI: it takes the host from
|
||||
`ANTHROPIC_BASE_URL` and from every variable named `*_BASE_URL` (the
|
||||
The controller also adds entries itself, and only from the operator's own
|
||||
actions in the swarm UI. Saving an agent's API-key environment makes it take
|
||||
the host from `ANTHROPIC_BASE_URL` and from every variable named `*_BASE_URL` (the
|
||||
address an ACP agent that loads the same file reads), and keeps one
|
||||
`derived` entry for that agent per host. Each host passes the same check as
|
||||
an operator's entry, so an IP address, `localhost` and a base-list name get
|
||||
|
|
@ -366,6 +366,19 @@ host. It never decrypts TLS, so it holds no signing key. It caches nothing.
|
|||
`acp.opencode.provider.baseUrl`, get no derived entry, so add one in the
|
||||
swarm UI. A `hive-c0re` that predates the field renders a derived entry
|
||||
like any other.
|
||||
Linking a GitHub token, an external matrix account or an external forge
|
||||
account adds a derived entry for that agent per host it reaches, and
|
||||
deleting the account removes it: `github.com` and `api.github.com` for
|
||||
GitHub, and the host of the matrix account's `homeserver` or the forge
|
||||
account's `url`. Each entry's `source` (`link:github`,
|
||||
`link:matrix:<account>`, `link:forge:<label>`) keeps it apart from the
|
||||
environment's, which a save of the environment leaves alone; an entry
|
||||
without a `source` counts as the environment's. Only the controller writes
|
||||
these leaves, so when it starts it also adds the entries of accounts linked
|
||||
before this existed and drops those whose account no longer exists, and logs
|
||||
the counts (`link egress: entries match the linked accounts`). A store it
|
||||
can't read changes nothing. An `agent.nix` `matrixAccounts` homeserver gets
|
||||
no entry, since the controller never reads it; add one in the swarm UI.
|
||||
The swarm's list adds to the hive's `allowedDomains`: `allowedFile` holds
|
||||
those names plus the swarm-wide entries, so removing an entry never removes a
|
||||
name the hive lists itself. Until the controller has a list (it predates the
|
||||
|
|
|
|||
|
|
@ -15,8 +15,9 @@
|
|||
// internet flag reaches any name the proxy's private-range rules leave, so an
|
||||
// entry matters only to a restricted one.
|
||||
//
|
||||
// An entry the controller keeps from the agent's backend environment (`derived`)
|
||||
// shows as automatic with no remove control; the route refuses to remove it.
|
||||
// An entry the controller keeps from the agent's backend environment or a linked
|
||||
// account (`derived`) shows as automatic with no remove control; the route
|
||||
// refuses to remove it.
|
||||
//
|
||||
// Its state belongs to one agent: callers key it by agent name.
|
||||
import { useEffect, useState } from "preact/hooks";
|
||||
|
|
@ -131,7 +132,7 @@ export function AllowedDomains({ agent }: { agent: string }) {
|
|||
value={e.domain}
|
||||
title={
|
||||
e.derived === true
|
||||
? "kept from this agent's backend environment; change that variable to change it"
|
||||
? "kept from this agent's backend environment or a linked account; change or unlink that to change it"
|
||||
: undefined
|
||||
}
|
||||
/>{" "}
|
||||
|
|
|
|||
|
|
@ -423,6 +423,7 @@ mod tests {
|
|||
domain: domain.to_owned(),
|
||||
agent: agent.map(str::to_owned),
|
||||
derived: false,
|
||||
source: None,
|
||||
}
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -19,8 +19,9 @@ pub struct AllowedDomain {
|
|||
#[cfg_attr(feature = "schema", schema(example = "atlas"))]
|
||||
pub agent: Option<String>,
|
||||
/// Set on an entry the controller keeps from the agent's backend
|
||||
/// environment (the host of its `ANTHROPIC_BASE_URL`). The list shows it
|
||||
/// as automatic; adding or removing one is refused.
|
||||
/// environment (the host of its `ANTHROPIC_BASE_URL`) or from an account an
|
||||
/// operator linked to it. The list shows it as automatic; adding or
|
||||
/// removing one is refused.
|
||||
#[serde(default, skip_serializing_if = "std::ops::Not::not")]
|
||||
pub derived: bool,
|
||||
}
|
||||
|
|
|
|||
|
|
@ -19,16 +19,18 @@
|
|||
//! environment, so it is never derived either.
|
||||
|
||||
use std::collections::{BTreeMap, BTreeSet};
|
||||
use std::sync::Arc;
|
||||
use std::time::Duration;
|
||||
|
||||
use problem_details::ProblemDetails;
|
||||
use swarm_secret_client::backend_env::{self, BackendEnv};
|
||||
use swarm_wire::egress_allow::{AllowList, Entry, MAX_ENTRIES};
|
||||
use swarm_wire::egress_allow::{AllowList, BACKEND_ENV_SOURCE, Entry, MAX_ENTRIES};
|
||||
|
||||
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
|
||||
|
|
@ -71,7 +73,7 @@ fn unquoted(value: &str) -> &str {
|
|||
/// The host of `url`, as written: its scheme (`http` or `https`, or none),
|
||||
/// user information, port, path, query and fragment are dropped. Nothing is
|
||||
/// validated; the result goes through `egress_allow::checked`.
|
||||
fn host_of(url: &str) -> Option<&str> {
|
||||
pub(crate) fn host_of(url: &str) -> Option<&str> {
|
||||
let rest = match url.split_once("://") {
|
||||
Some(("http" | "https", rest)) => rest,
|
||||
Some(_) => return None,
|
||||
|
|
@ -117,6 +119,7 @@ fn derive_named<'a>(agent: &str, env: &'a str, base: &[String]) -> Vec<(&'a str,
|
|||
name,
|
||||
Entry {
|
||||
derived: true,
|
||||
source: Some(BACKEND_ENV_SOURCE.to_owned()),
|
||||
..entry
|
||||
},
|
||||
)),
|
||||
|
|
@ -128,12 +131,14 @@ fn derive_named<'a>(agent: &str, env: &'a str, base: &[String]) -> Vec<(&'a str,
|
|||
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
|
||||
/// `list` with, for each agent in `desired`, its entries derived from the backend
|
||||
/// environment replaced by the given ones. Entries of other agents,
|
||||
/// operator-added entries and entries derived from linked accounts are
|
||||
/// untouched. A full list takes no more derived entries.
|
||||
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))));
|
||||
list.entries.retain(|e| {
|
||||
!(e.from_backend_env() && e.agent.as_ref().is_some_and(|a| desired.contains_key(a)))
|
||||
});
|
||||
for entry in desired.values().flatten() {
|
||||
if list.entries.len() >= MAX_ENTRIES {
|
||||
tracing::warn!(
|
||||
|
|
@ -190,15 +195,15 @@ pub(crate) async fn follow_env(
|
|||
Ok(())
|
||||
}
|
||||
|
||||
/// `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.
|
||||
/// `list` without the backend-env 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, operator-added entries and link-derived 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)
|
||||
!e.from_backend_env() || hosts.contains(&e.domain)
|
||||
});
|
||||
list
|
||||
}
|
||||
|
|
@ -223,7 +228,7 @@ pub(crate) async fn reconcile(
|
|||
return Ok(0);
|
||||
};
|
||||
let mut derived: BTreeMap<String, BTreeSet<String>> = BTreeMap::new();
|
||||
for entry in current.entries.iter().filter(|e| e.derived) {
|
||||
for entry in current.entries.iter().filter(|e| e.from_backend_env()) {
|
||||
if let Some(agent) = &entry.agent {
|
||||
derived
|
||||
.entry(agent.clone())
|
||||
|
|
@ -273,16 +278,24 @@ pub(crate) async fn reconcile(
|
|||
Ok(unreadable)
|
||||
}
|
||||
|
||||
/// Run [`reconcile`] once the queue and the store answer, retrying with a
|
||||
/// growing wait until a pass reads every agent that has an entry.
|
||||
pub(crate) fn spawn_backfill(client: async_nats::Client) {
|
||||
/// Run [`reconcile`] and `link_egress::reconcile` once the queue and the store
|
||||
/// answer, retrying with a growing wait until a pass reads every agent that
|
||||
/// has an entry and some hive has declared an agent.
|
||||
pub(crate) fn spawn_backfill(
|
||||
client: async_nats::Client,
|
||||
wanted: Arc<WantedWriter>,
|
||||
hives: Vec<String>,
|
||||
) {
|
||||
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?;
|
||||
reconcile(&client, &store).await
|
||||
let live = crate::subagent_term::live_agents(&wanted, &hives).await;
|
||||
let unreadable = reconcile(&client, &store).await?
|
||||
+ crate::link_egress::reconcile(&client, &store, &live).await?;
|
||||
Ok::<_, anyhow::Error>(unreadable + usize::from(live.is_empty()))
|
||||
}
|
||||
.await;
|
||||
match pass {
|
||||
|
|
@ -395,6 +408,7 @@ mod tests {
|
|||
domain: "llm.example.com".to_owned(),
|
||||
agent: Some("atlas".to_owned()),
|
||||
derived: true,
|
||||
source: Some(BACKEND_ENV_SOURCE.to_owned()),
|
||||
}]
|
||||
);
|
||||
}
|
||||
|
|
@ -404,12 +418,14 @@ mod tests {
|
|||
domain: domain.to_owned(),
|
||||
agent: Some(agent.to_owned()),
|
||||
derived: false,
|
||||
source: None,
|
||||
}
|
||||
}
|
||||
|
||||
fn derived(domain: &str, agent: &str) -> Entry {
|
||||
Entry {
|
||||
derived: true,
|
||||
source: Some(BACKEND_ENV_SOURCE.to_owned()),
|
||||
..manual(domain, agent)
|
||||
}
|
||||
}
|
||||
|
|
@ -440,6 +456,68 @@ mod tests {
|
|||
);
|
||||
}
|
||||
|
||||
fn linked(domain: &str, agent: &str, source: &str) -> Entry {
|
||||
Entry {
|
||||
derived: true,
|
||||
source: Some(source.to_owned()),
|
||||
..manual(domain, agent)
|
||||
}
|
||||
}
|
||||
|
||||
/// A derived entry written before sources existed.
|
||||
fn untagged(domain: &str, agent: &str) -> Entry {
|
||||
Entry {
|
||||
source: None,
|
||||
..derived(domain, agent)
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_backend_env_save_leaves_link_entries_and_replaces_untagged_ones() {
|
||||
let list = AllowList {
|
||||
entries: vec![
|
||||
linked("github.com", "atlas", "link:github"),
|
||||
linked("hs.example.org", "atlas", "link:matrix:work"),
|
||||
untagged("old.example.com", "atlas"),
|
||||
],
|
||||
};
|
||||
let desired = BTreeMap::from([(
|
||||
"atlas".to_owned(),
|
||||
vec![derived("new.example.com", "atlas")],
|
||||
)]);
|
||||
let after = replaced(list, &desired);
|
||||
assert_eq!(
|
||||
after.entries,
|
||||
[
|
||||
linked("github.com", "atlas", "link:github"),
|
||||
linked("hs.example.org", "atlas", "link:matrix:work"),
|
||||
derived("new.example.com", "atlas"),
|
||||
]
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn pruning_the_backend_env_leaves_link_entries_and_drops_untagged_ones() {
|
||||
let list = AllowList {
|
||||
entries: vec![
|
||||
linked("github.com", "atlas", "link:github"),
|
||||
untagged("old.example.com", "atlas"),
|
||||
derived("kept.example.com", "atlas"),
|
||||
],
|
||||
};
|
||||
let still_named = BTreeMap::from([(
|
||||
"atlas".to_owned(),
|
||||
BTreeSet::from(["kept.example.com".to_owned()]),
|
||||
)]);
|
||||
assert_eq!(
|
||||
pruned(list, &still_named).entries,
|
||||
[
|
||||
linked("github.com", "atlas", "link:github"),
|
||||
derived("kept.example.com", "atlas"),
|
||||
]
|
||||
);
|
||||
}
|
||||
|
||||
/// The state of a test swarm whose `pr1ma` declares `atlas`, with `store`
|
||||
/// as the secret store.
|
||||
async fn swarm(client: &async_nats::Client, store: &Arc<FakeStore>) -> AppState {
|
||||
|
|
|
|||
|
|
@ -30,8 +30,8 @@ use crate::schedules::{live, queue, read_or_default, refused, write_revision};
|
|||
const MAX_ATTEMPTS: usize = 5;
|
||||
|
||||
/// Why an operator can neither add nor remove an entry the controller derived.
|
||||
const DERIVED_REFUSED: &str = "this entry is kept automatically from the agent's backend environment; \
|
||||
change or remove that variable instead";
|
||||
const DERIVED_REFUSED: &str = "this entry is kept automatically from the agent's backend environment or a linked \
|
||||
account; change or remove that variable, or unlink the account, instead";
|
||||
|
||||
/// Set from the hive module's `baseAllowedDomains` default by
|
||||
/// `nix/host-modules/swarm-controller.nix`: the names, separated by spaces,
|
||||
|
|
@ -69,6 +69,7 @@ pub(crate) fn checked(req: &AllowedDomain, base: &[String]) -> Result<Entry, Str
|
|||
domain,
|
||||
agent,
|
||||
derived: false,
|
||||
source: None,
|
||||
})
|
||||
}
|
||||
|
||||
|
|
@ -551,6 +552,7 @@ mod tests {
|
|||
domain: format!("host{i}.example.com"),
|
||||
agent: None,
|
||||
derived: false,
|
||||
source: None,
|
||||
})
|
||||
.collect(),
|
||||
};
|
||||
|
|
@ -558,6 +560,7 @@ mod tests {
|
|||
domain: "github.com".to_owned(),
|
||||
agent: None,
|
||||
derived: false,
|
||||
source: None,
|
||||
};
|
||||
let why = added(full.clone(), &new).err().expect("a full list");
|
||||
assert!(why.contains("1000"), "{why}");
|
||||
|
|
|
|||
|
|
@ -82,6 +82,15 @@ pub async fn put_forge_account(
|
|||
tracing::warn!(path = %secret_path, error = ?e, "linking the forge account failed");
|
||||
e.problem(&existing(&agent, &label))
|
||||
})?;
|
||||
crate::link_egress::added(
|
||||
&state,
|
||||
&agent,
|
||||
&crate::link_egress::Link::Forge {
|
||||
label: &label,
|
||||
url: &account.url,
|
||||
},
|
||||
)
|
||||
.await?;
|
||||
|
||||
tracing::info!(%hive, %agent, %label, url = %account.url, "forge account stored");
|
||||
Ok(Json(PutForgeAccountResponse { url: account.url }))
|
||||
|
|
|
|||
|
|
@ -65,6 +65,7 @@ pub async fn put_github_account(
|
|||
tracing::warn!(path = %secret_path, error = ?e, "linking the github token failed");
|
||||
e.problem(&existing(&agent))
|
||||
})?;
|
||||
crate::link_egress::added(&state, &agent, &crate::link_egress::Link::Github).await?;
|
||||
|
||||
tracing::info!(%hive, %agent, "github token stored");
|
||||
Ok(StatusCode::NO_CONTENT)
|
||||
|
|
|
|||
622
swarm-controller/src/link_egress.rs
Normal file
622
swarm-controller/src/link_egress.rs
Normal file
|
|
@ -0,0 +1,622 @@
|
|||
//! The hosts an agent's linked accounts reach, kept as derived entries of the
|
||||
//! egress allow list (`crate::egress_allow`), so a restricted agent with a
|
||||
//! linked GitHub, external matrix or external forge account reaches that
|
||||
//! service through its hive's proxy.
|
||||
//!
|
||||
//! An entry is marked `derived` and carries the account as its `source`
|
||||
//! (`link:github`, `link:matrix:<account>`, `link:forge:<label>`), so the
|
||||
//! backend environment's entries (`crate::backend_egress`) and each account's
|
||||
//! are replaced and dropped independently. A hive renders it like any other.
|
||||
//!
|
||||
//! The controller alone writes the three leaves the entries come from; the
|
||||
//! agent's store policy grants it `create` on its own `backend-env` leaf and
|
||||
//! nothing else (`swarm-secret-client/src/policy.rs`), so what these leaves
|
||||
//! hold is the operator's choice, unlike the backend environment's.
|
||||
//! [`added`] and [`removed`] follow the operator's link and unlink, and
|
||||
//! [`reconcile`] makes the entries match the leaves at start: it adds the
|
||||
//! entries of accounts linked before this existed and drops those of accounts
|
||||
//! whose leaf is gone, and an agent whose leaves cannot be read keeps what it
|
||||
//! has.
|
||||
//!
|
||||
//! The `homeserver` and `url` of an account go through
|
||||
//! [`crate::backend_egress::host_of`] and `egress_allow::checked`, the check
|
||||
//! an operator's own entry gets. An agent's `matrixAccounts.*.homeserver` in
|
||||
//! its own configuration is not read: the controller never sees it.
|
||||
|
||||
use std::collections::{BTreeMap, BTreeSet};
|
||||
|
||||
use problem_details::ProblemDetails;
|
||||
use swarm_wire::egress_allow::{AllowList, Entry, LINK_SOURCE_PREFIX, MAX_ENTRIES};
|
||||
|
||||
use crate::backend_egress::host_of;
|
||||
use crate::egress_allow::{base_domains, checked, current, modify};
|
||||
use crate::linked_accounts::{AccountKind, AccountStore, linked_accounts};
|
||||
use crate::routes::{AppState, error_problem, wanted_error_status};
|
||||
use crate::schedules::queue;
|
||||
|
||||
/// What the GitHub token is used against: `gh`'s API and git over HTTPS.
|
||||
const GITHUB_HOSTS: [&str; 2] = ["github.com", "api.github.com"];
|
||||
|
||||
/// An account the operator links to an agent.
|
||||
pub(crate) enum Link<'a> {
|
||||
Github,
|
||||
Matrix {
|
||||
account: &'a str,
|
||||
homeserver: Option<&'a str>,
|
||||
},
|
||||
Forge {
|
||||
label: &'a str,
|
||||
url: &'a str,
|
||||
},
|
||||
}
|
||||
|
||||
pub(crate) fn github_source() -> String {
|
||||
format!("{LINK_SOURCE_PREFIX}github")
|
||||
}
|
||||
|
||||
pub(crate) fn matrix_source(account: &str) -> String {
|
||||
format!("{LINK_SOURCE_PREFIX}matrix:{account}")
|
||||
}
|
||||
|
||||
pub(crate) fn forge_source(label: &str) -> String {
|
||||
format!("{LINK_SOURCE_PREFIX}forge:{label}")
|
||||
}
|
||||
|
||||
impl Link<'_> {
|
||||
fn source(&self) -> String {
|
||||
match self {
|
||||
Self::Github => github_source(),
|
||||
Self::Matrix { account, .. } => matrix_source(account),
|
||||
Self::Forge { label, .. } => forge_source(label),
|
||||
}
|
||||
}
|
||||
|
||||
/// The entries for `agent`: one per host that passes
|
||||
/// `egress_allow::checked`. A host that does not (an IP address,
|
||||
/// `localhost`, a name the base list already allows) is logged and left
|
||||
/// out. The URL is never logged: it can carry credentials.
|
||||
fn entries(&self, agent: &str, base: &[String]) -> Vec<Entry> {
|
||||
let source = self.source();
|
||||
let hosts: Vec<Option<&str>> = match self {
|
||||
Self::Github => GITHUB_HOSTS.iter().map(|h| Some(*h)).collect(),
|
||||
Self::Matrix { homeserver, .. } => homeserver.iter().map(|u| host_of(u)).collect(),
|
||||
Self::Forge { url, .. } => vec![host_of(url)],
|
||||
};
|
||||
let mut entries = Vec::new();
|
||||
for host in hosts {
|
||||
let Some(host) = host else {
|
||||
tracing::warn!(%agent, %source, "link egress: not an http(s) URL; no entry");
|
||||
continue;
|
||||
};
|
||||
let request = swarm_controller_api::egress::AllowedDomain {
|
||||
domain: host.to_owned(),
|
||||
agent: Some(agent.to_owned()),
|
||||
derived: false,
|
||||
};
|
||||
match checked(&request, base) {
|
||||
Ok(entry) => entries.push(Entry {
|
||||
derived: true,
|
||||
source: Some(source.clone()),
|
||||
..entry
|
||||
}),
|
||||
Err(why) => {
|
||||
tracing::warn!(%agent, %source, %why, "link egress: no entry for this host");
|
||||
}
|
||||
}
|
||||
}
|
||||
entries
|
||||
}
|
||||
}
|
||||
|
||||
/// `list` without the entries `stale` names, plus `add`. A full list takes no
|
||||
/// more entries.
|
||||
fn replaced(mut list: AllowList, stale: impl Fn(&Entry) -> bool, add: &[Entry]) -> AllowList {
|
||||
list.entries.retain(|e| !stale(e));
|
||||
for entry in add {
|
||||
if list.entries.len() >= MAX_ENTRIES {
|
||||
tracing::warn!(
|
||||
domain = %entry.domain,
|
||||
"link egress: the allow list is full; no entry for this host"
|
||||
);
|
||||
break;
|
||||
}
|
||||
list.entries.push(entry.clone());
|
||||
}
|
||||
list
|
||||
}
|
||||
|
||||
/// Whether `e` is `agent`'s entry derived from exactly `source`.
|
||||
fn is_of(e: &Entry, agent: &str, source: &str) -> bool {
|
||||
e.derived && e.agent.as_deref() == Some(agent) && e.source.as_deref() == Some(source)
|
||||
}
|
||||
|
||||
/// Replace `agent`'s entries of `source` with `entries`. A host without a
|
||||
/// queue (`state.status` unset) has no allow list, so nothing happens.
|
||||
async fn follow(
|
||||
state: &AppState,
|
||||
agent: &str,
|
||||
source: &str,
|
||||
entries: Vec<Entry>,
|
||||
stored: &str,
|
||||
) -> Result<(), ProblemDetails> {
|
||||
if state.status.is_none() {
|
||||
return Ok(());
|
||||
}
|
||||
let updated = async {
|
||||
let client = queue(state).map_err(|(_, why)| anyhow::anyhow!(why))?;
|
||||
modify(&client, |list| {
|
||||
replaced(list, |e| is_of(e, agent, source), &entries)
|
||||
})
|
||||
.await
|
||||
}
|
||||
.await;
|
||||
match updated {
|
||||
Ok(true) => tracing::info!(%agent, %source, "link egress: derived entries updated"),
|
||||
Ok(false) => {}
|
||||
Err(e) => {
|
||||
tracing::warn!(%agent, %source, error = %format!("{e:#}"), "link egress: updating the allow list failed");
|
||||
return Err(error_problem(
|
||||
wanted_error_status(&e),
|
||||
&format!(
|
||||
"the account is {stored}, but the allowed domains were not updated \
|
||||
({e:#}); the controller corrects them when it next starts"
|
||||
),
|
||||
));
|
||||
}
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Give `agent` the entries of the account `link` just stored.
|
||||
///
|
||||
/// # Errors
|
||||
/// A problem naming that the account is linked but the list was not updated.
|
||||
pub(crate) async fn added(
|
||||
state: &AppState,
|
||||
agent: &str,
|
||||
link: &Link<'_>,
|
||||
) -> Result<(), ProblemDetails> {
|
||||
let entries = link.entries(agent, &base_domains());
|
||||
follow(state, agent, &link.source(), entries, "linked").await
|
||||
}
|
||||
|
||||
/// Take from `agent` the entries of the account stored at `source`, which was
|
||||
/// just deleted.
|
||||
///
|
||||
/// # Errors
|
||||
/// A problem naming that the account is deleted but the list was not updated.
|
||||
pub(crate) async fn removed(
|
||||
state: &AppState,
|
||||
agent: &str,
|
||||
source: &str,
|
||||
) -> Result<(), ProblemDetails> {
|
||||
follow(state, agent, source, Vec::new(), "deleted").await
|
||||
}
|
||||
|
||||
/// The entries the accounts stored for `agent` call for.
|
||||
async fn wanted_entries(
|
||||
store: &impl AccountStore,
|
||||
agent: &str,
|
||||
base: &[String],
|
||||
) -> Result<Vec<Entry>, swarm_secret_client::Error> {
|
||||
let mut entries = Vec::new();
|
||||
for account in linked_accounts(store, agent).await? {
|
||||
let link = match account.kind {
|
||||
AccountKind::Github => Link::Github,
|
||||
AccountKind::Matrix if !account.reserved => Link::Matrix {
|
||||
account: &account.name,
|
||||
homeserver: account.host.as_deref(),
|
||||
},
|
||||
AccountKind::Forgejo => match account.host.as_deref() {
|
||||
Some(url) => Link::Forge {
|
||||
label: &account.name,
|
||||
url,
|
||||
},
|
||||
None => continue,
|
||||
},
|
||||
AccountKind::Matrix | AccountKind::Backend => continue,
|
||||
};
|
||||
entries.extend(link.entries(agent, base));
|
||||
}
|
||||
Ok(entries)
|
||||
}
|
||||
|
||||
/// `list` with, for each agent in `desired`, its link-derived entries replaced
|
||||
/// by the given ones. Entries of other agents, operator-added entries and
|
||||
/// entries derived from the backend environment are untouched.
|
||||
fn synced(list: AllowList, desired: &BTreeMap<String, Vec<Entry>>) -> AllowList {
|
||||
let add: Vec<Entry> = desired.values().flatten().cloned().collect();
|
||||
replaced(
|
||||
list,
|
||||
|e| e.from_link() && e.agent.as_ref().is_some_and(|a| desired.contains_key(a)),
|
||||
&add,
|
||||
)
|
||||
}
|
||||
|
||||
/// Make the link-derived entries match the stored accounts of every agent in
|
||||
/// `live` and of every agent that has one: add what a stored account calls for,
|
||||
/// drop what no stored account does. Logs the counts.
|
||||
///
|
||||
/// An agent whose accounts cannot be read keeps its entries and is counted in
|
||||
/// the answer, which is how many agents that was.
|
||||
///
|
||||
/// # Errors
|
||||
/// A queue failure reading or writing the list.
|
||||
pub(crate) async fn reconcile(
|
||||
client: &async_nats::Client,
|
||||
store: &impl AccountStore,
|
||||
live: &BTreeSet<String>,
|
||||
) -> anyhow::Result<usize> {
|
||||
let before = current(client).await?.unwrap_or_default();
|
||||
let mut agents = live.clone();
|
||||
agents.extend(
|
||||
before
|
||||
.entries
|
||||
.iter()
|
||||
.filter(|e| e.from_link())
|
||||
.filter_map(|e| e.agent.clone()),
|
||||
);
|
||||
let base = base_domains();
|
||||
let mut desired = BTreeMap::new();
|
||||
let mut unreadable = 0;
|
||||
for agent in agents {
|
||||
match wanted_entries(store, &agent, &base).await {
|
||||
Ok(entries) => {
|
||||
desired.insert(agent, entries);
|
||||
}
|
||||
Err(e) => {
|
||||
tracing::warn!(%agent, error = %e, "link egress: reading the linked accounts failed; its entries stay");
|
||||
unreadable += 1;
|
||||
}
|
||||
}
|
||||
}
|
||||
let had: BTreeSet<&Entry> = before.entries.iter().filter(|e| e.from_link()).collect();
|
||||
let after = synced(before.clone(), &desired);
|
||||
let has: BTreeSet<&Entry> = after.entries.iter().filter(|e| e.from_link()).collect();
|
||||
let (added, removed) = (has.difference(&had).count(), had.difference(&has).count());
|
||||
modify(client, |list| synced(list, &desired)).await?;
|
||||
tracing::info!(
|
||||
agents = desired.len(),
|
||||
unreadable,
|
||||
added,
|
||||
removed,
|
||||
"link egress: entries match the linked accounts"
|
||||
);
|
||||
Ok(unreadable)
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use std::sync::Arc;
|
||||
|
||||
use axum::extract::{Path, State};
|
||||
use serde_json::json;
|
||||
|
||||
use super::*;
|
||||
use crate::agent_limits::tests::declared;
|
||||
use crate::linked_accounts::tests::FakeStore;
|
||||
use crate::test_nats::TestNats;
|
||||
|
||||
fn base() -> Vec<String> {
|
||||
vec!["api.anthropic.com".to_owned()]
|
||||
}
|
||||
|
||||
fn hosts(link: &Link<'_>) -> Vec<String> {
|
||||
link.entries("atlas", &base())
|
||||
.into_iter()
|
||||
.map(|e| e.domain)
|
||||
.collect()
|
||||
}
|
||||
|
||||
fn entry(domain: &str, agent: &str, source: Option<&str>) -> Entry {
|
||||
Entry {
|
||||
domain: domain.to_owned(),
|
||||
agent: Some(agent.to_owned()),
|
||||
derived: source.is_some(),
|
||||
source: source.map(str::to_owned),
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn each_kind_yields_the_hosts_its_service_is_reached_at() {
|
||||
let github = Link::Github.entries("atlas", &base());
|
||||
assert_eq!(
|
||||
github,
|
||||
[
|
||||
entry("github.com", "atlas", Some("link:github")),
|
||||
entry("api.github.com", "atlas", Some("link:github")),
|
||||
]
|
||||
);
|
||||
// A port, a path, a trailing slash and a scheme-less value all reduce
|
||||
// to the bare host.
|
||||
for homeserver in [
|
||||
"https://Matrix.Example.org:8448/_matrix/",
|
||||
"https://matrix.example.org/",
|
||||
"matrix.example.org",
|
||||
] {
|
||||
let link = Link::Matrix {
|
||||
account: "work",
|
||||
homeserver: Some(homeserver),
|
||||
};
|
||||
assert_eq!(hosts(&link), ["matrix.example.org"], "{homeserver}");
|
||||
assert_eq!(
|
||||
link.entries("atlas", &base()),
|
||||
[entry(
|
||||
"matrix.example.org",
|
||||
"atlas",
|
||||
Some("link:matrix:work")
|
||||
)]
|
||||
);
|
||||
}
|
||||
let forge = Link::Forge {
|
||||
label: "oss",
|
||||
url: "https://forge.example.org:3000/sub",
|
||||
};
|
||||
assert_eq!(
|
||||
forge.entries("atlas", &base()),
|
||||
[entry("forge.example.org", "atlas", Some("link:forge:oss"))]
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_host_the_swarm_would_refuse_gets_no_entry() {
|
||||
for url in [
|
||||
"http://10.0.0.5:3000",
|
||||
"https://localhost",
|
||||
"ftp://forge.example.org",
|
||||
"https://",
|
||||
"https://*.example.org",
|
||||
// The base list already allows it.
|
||||
"https://api.anthropic.com",
|
||||
] {
|
||||
let forge = Link::Forge { label: "x", url };
|
||||
assert_eq!(hosts(&forge), Vec::<String>::new(), "{url:?}");
|
||||
}
|
||||
// Controls: a public name gets one, and a matrix account with no
|
||||
// homeserver has none to derive.
|
||||
let forge = Link::Forge {
|
||||
label: "x",
|
||||
url: "https://forge.example.org",
|
||||
};
|
||||
assert_eq!(hosts(&forge), ["forge.example.org"]);
|
||||
let matrix = Link::Matrix {
|
||||
account: "work",
|
||||
homeserver: None,
|
||||
};
|
||||
assert_eq!(hosts(&matrix), Vec::<String>::new());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn unlinking_removes_only_that_sources_entries() {
|
||||
let list = AllowList {
|
||||
entries: vec![
|
||||
entry("github.com", "atlas", Some("link:github")),
|
||||
entry("github.com", "iris", Some("link:github")),
|
||||
entry("forge.example.org", "atlas", Some("link:forge:oss")),
|
||||
entry("forge.example.org", "atlas", Some("link:forge:work")),
|
||||
entry("llm.example.com", "atlas", Some("backend-env")),
|
||||
entry("old.example.com", "atlas", Some("backend-env")),
|
||||
entry("github.com", "atlas", None),
|
||||
],
|
||||
};
|
||||
let after = replaced(
|
||||
list.clone(),
|
||||
|e| is_of(e, "atlas", &forge_source("oss")),
|
||||
&[],
|
||||
);
|
||||
assert_eq!(after.entries.len(), list.entries.len() - 1);
|
||||
assert!(!after.entries.contains(&entry(
|
||||
"forge.example.org",
|
||||
"atlas",
|
||||
Some("link:forge:oss")
|
||||
)));
|
||||
// Control: the same lookup finds the entry it removed.
|
||||
assert!(
|
||||
list.entries
|
||||
.iter()
|
||||
.any(|e| is_of(e, "atlas", "link:forge:oss"))
|
||||
);
|
||||
// The manual entry for the same name as a derived one stays.
|
||||
let after = replaced(list, |e| is_of(e, "atlas", &github_source()), &[]);
|
||||
assert!(after.entries.contains(&entry("github.com", "atlas", None)));
|
||||
assert!(
|
||||
!after
|
||||
.entries
|
||||
.contains(&entry("github.com", "atlas", Some("link:github")))
|
||||
);
|
||||
}
|
||||
|
||||
/// The state of a test swarm whose `pr1ma` declares `atlas`, with `store`
|
||||
/// as the secret store.
|
||||
async fn swarm(client: &async_nats::Client, store: &Arc<FakeStore>) -> AppState {
|
||||
AppState {
|
||||
accounts: crate::linked_accounts::Accounts::Fake(Arc::clone(store)),
|
||||
..declared(client).await
|
||||
}
|
||||
}
|
||||
|
||||
async fn list_of(client: &async_nats::Client) -> Vec<Entry> {
|
||||
let mut entries = current(client)
|
||||
.await
|
||||
.expect("the queue answers")
|
||||
.map(|l| l.entries)
|
||||
.unwrap_or_default();
|
||||
entries.sort();
|
||||
entries
|
||||
}
|
||||
|
||||
fn sorted(mut entries: Vec<Entry>) -> Vec<Entry> {
|
||||
entries.sort();
|
||||
entries
|
||||
}
|
||||
|
||||
fn github_entries(agent: &str) -> Vec<Entry> {
|
||||
GITHUB_HOSTS
|
||||
.iter()
|
||||
.map(|h| entry(h, agent, Some("link:github")))
|
||||
.collect()
|
||||
}
|
||||
|
||||
fn path(parts: &str) -> Path<(String, String)> {
|
||||
Path(("pr1ma".to_owned(), parts.to_owned()))
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn linking_and_unlinking_through_the_routes_moves_the_entries() {
|
||||
let nats = TestNats::start();
|
||||
let client = nats.client().await;
|
||||
let store = Arc::default();
|
||||
let state = swarm(&client, &store).await;
|
||||
|
||||
let _ = crate::github_account::put_github_account(
|
||||
State(state.clone()),
|
||||
path("atlas"),
|
||||
axum::Json(serde_json::from_value(json!({"token": "t"})).expect("a body")),
|
||||
)
|
||||
.await
|
||||
.expect("linked");
|
||||
let _ = crate::forge_account::put_forge_account(
|
||||
State(state.clone()),
|
||||
Path(("pr1ma".to_owned(), "atlas".to_owned(), "oss".to_owned())),
|
||||
axum::Json(
|
||||
serde_json::from_value(
|
||||
json!({"url": "https://forge.example.org:3000/", "token": "t"}),
|
||||
)
|
||||
.expect("a body"),
|
||||
),
|
||||
)
|
||||
.await
|
||||
.expect("linked");
|
||||
let _ = crate::matrix_account::put_matrix_account(
|
||||
State(state.clone()),
|
||||
Path(("pr1ma".to_owned(), "atlas".to_owned(), "work".to_owned())),
|
||||
axum::Json(
|
||||
serde_json::from_value(json!({
|
||||
"mode": "token", "token": "t", "homeserver": "https://matrix.example.org/"
|
||||
}))
|
||||
.expect("a body"),
|
||||
),
|
||||
)
|
||||
.await
|
||||
.expect("linked");
|
||||
let forge = entry("forge.example.org", "atlas", Some("link:forge:oss"));
|
||||
let matrix = entry("matrix.example.org", "atlas", Some("link:matrix:work"));
|
||||
let mut all = github_entries("atlas");
|
||||
all.extend([forge.clone(), matrix.clone()]);
|
||||
assert_eq!(list_of(&client).await, sorted(all));
|
||||
|
||||
// The backend environment's own save and delete leave them alone.
|
||||
crate::backend_env::put_backend_env(
|
||||
State(state.clone()),
|
||||
path("atlas"),
|
||||
axum::Json(
|
||||
serde_json::from_value(
|
||||
json!({"env": "ANTHROPIC_BASE_URL=https://llm.example.com\n"}),
|
||||
)
|
||||
.expect("a body"),
|
||||
),
|
||||
)
|
||||
.await
|
||||
.expect("stored");
|
||||
crate::linked_accounts::delete_backend_env(State(state.clone()), path("atlas"))
|
||||
.await
|
||||
.expect("deleted");
|
||||
assert_eq!(list_of(&client).await.len(), 4);
|
||||
|
||||
crate::linked_accounts::delete_github_account(State(state.clone()), path("atlas"))
|
||||
.await
|
||||
.expect("deleted");
|
||||
assert_eq!(
|
||||
list_of(&client).await,
|
||||
sorted(vec![forge.clone(), matrix.clone()])
|
||||
);
|
||||
crate::linked_accounts::delete_forge_account(
|
||||
State(state.clone()),
|
||||
Path(("pr1ma".to_owned(), "atlas".to_owned(), "oss".to_owned())),
|
||||
)
|
||||
.await
|
||||
.expect("deleted");
|
||||
assert_eq!(list_of(&client).await, [matrix]);
|
||||
crate::linked_accounts::delete_matrix_account(
|
||||
State(state.clone()),
|
||||
Path(("pr1ma".to_owned(), "atlas".to_owned(), "work".to_owned())),
|
||||
axum::extract::Query(serde_json::from_value(json!({})).expect("a query")),
|
||||
)
|
||||
.await
|
||||
.expect("deleted");
|
||||
assert_eq!(list_of(&client).await, []);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn the_backfill_adds_entries_for_existing_links_and_prunes_orphans() {
|
||||
let nats = TestNats::start();
|
||||
let client = nats.client().await;
|
||||
let store =
|
||||
FakeStore::default().with("swarm/agents/atlas/github-token", json!({"value": "t"}));
|
||||
let store = store.with(
|
||||
"swarm/agents/atlas/matrix/work",
|
||||
json!({"value": "t", "homeserver": "https://matrix.example.org"}),
|
||||
);
|
||||
// The agent's own account has a homeserver of its own and no entry.
|
||||
let store = store.with(
|
||||
"swarm/agents/atlas/matrix/main",
|
||||
json!({"value": "t", "homeserver": "https://hive.example.org"}),
|
||||
);
|
||||
modify(&client, |mut list| {
|
||||
list.entries.extend([
|
||||
// An orphan: iris has no leaf for it.
|
||||
entry("github.com", "iris", Some("link:github")),
|
||||
// An orphan on a live agent whose forge leaf is gone.
|
||||
entry("forge.example.org", "atlas", Some("link:forge:old")),
|
||||
entry("llm.example.com", "atlas", Some("backend-env")),
|
||||
entry("crates.io", "atlas", None),
|
||||
]);
|
||||
list
|
||||
})
|
||||
.await
|
||||
.expect("seed");
|
||||
|
||||
let live = BTreeSet::from(["atlas".to_owned()]);
|
||||
assert_eq!(reconcile(&client, &store, &live).await.expect("pass"), 0);
|
||||
let mut want = github_entries("atlas");
|
||||
want.extend([
|
||||
entry("matrix.example.org", "atlas", Some("link:matrix:work")),
|
||||
entry("llm.example.com", "atlas", Some("backend-env")),
|
||||
entry("crates.io", "atlas", None),
|
||||
]);
|
||||
assert_eq!(list_of(&client).await, sorted(want.clone()));
|
||||
|
||||
// A second pass changes nothing.
|
||||
reconcile(&client, &store, &live).await.expect("again");
|
||||
assert_eq!(list_of(&client).await, sorted(want));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn a_store_error_prunes_nothing_and_adds_nothing() {
|
||||
let nats = TestNats::start();
|
||||
let client = nats.client().await;
|
||||
let seeded = vec![
|
||||
entry("github.com", "iris", Some("link:github")),
|
||||
entry("forge.example.org", "atlas", Some("link:forge:old")),
|
||||
];
|
||||
modify(&client, |mut list| {
|
||||
list.entries.extend(seeded.clone());
|
||||
list
|
||||
})
|
||||
.await
|
||||
.expect("seed");
|
||||
|
||||
let live = BTreeSet::from(["atlas".to_owned()]);
|
||||
let unreadable = reconcile(&client, &FakeStore::denying(), &live)
|
||||
.await
|
||||
.expect("pass");
|
||||
assert_eq!(unreadable, 2, "atlas and iris");
|
||||
assert_eq!(list_of(&client).await, sorted(seeded.clone()));
|
||||
|
||||
// Control: a store that answers, and holds nothing, prunes both.
|
||||
reconcile(&client, &FakeStore::default(), &live)
|
||||
.await
|
||||
.expect("pass");
|
||||
assert_eq!(list_of(&client).await, []);
|
||||
}
|
||||
}
|
||||
|
|
@ -54,17 +54,17 @@ pub enum AccountKind {
|
|||
/// One account linked to an agent. Never carries the credential.
|
||||
#[derive(Clone, Debug, PartialEq, Eq, Serialize, ToSchema)]
|
||||
pub struct LinkedAccount {
|
||||
kind: AccountKind,
|
||||
pub(crate) kind: AccountKind,
|
||||
/// The matrix account name, the forge label, `github`, or `backend-env`.
|
||||
#[schema(example = "main")]
|
||||
name: String,
|
||||
pub(crate) name: String,
|
||||
/// The matrix homeserver, the forge base URL, or `github.com`. `None` for a
|
||||
/// matrix account stored without a homeserver.
|
||||
#[schema(example = "https://matrix.example.org")]
|
||||
host: Option<String>,
|
||||
pub(crate) host: Option<String>,
|
||||
/// The agent's own matrix account, which the swarm mints and an operator
|
||||
/// does not link.
|
||||
reserved: bool,
|
||||
pub(crate) reserved: bool,
|
||||
}
|
||||
|
||||
/// The store calls a listing, a link or a delete makes, so a test can stand in
|
||||
|
|
@ -489,6 +489,8 @@ pub async fn delete_matrix_account(
|
|||
tracing::warn!(%hive, %agent, %account, error = ?e, "deleting the matrix account failed");
|
||||
e.problem()
|
||||
})?;
|
||||
crate::link_egress::removed(&state, &agent, &crate::link_egress::matrix_source(&account))
|
||||
.await?;
|
||||
|
||||
tracing::info!(%hive, %agent, %account, revoked = query.revoke, "matrix account deleted");
|
||||
Ok(StatusCode::NO_CONTENT)
|
||||
|
|
@ -536,6 +538,7 @@ pub async fn delete_forge_account(
|
|||
tracing::warn!(%hive, %agent, %label, error = ?e, "deleting the forge account failed");
|
||||
e.problem()
|
||||
})?;
|
||||
crate::link_egress::removed(&state, &agent, &crate::link_egress::forge_source(&label)).await?;
|
||||
|
||||
tracing::info!(%hive, %agent, %label, "forge account deleted");
|
||||
Ok(StatusCode::NO_CONTENT)
|
||||
|
|
@ -578,6 +581,7 @@ pub async fn delete_github_account(
|
|||
tracing::warn!(%hive, %agent, error = ?e, "deleting the github token failed");
|
||||
e.problem()
|
||||
})?;
|
||||
crate::link_egress::removed(&state, &agent, &crate::link_egress::github_source()).await?;
|
||||
|
||||
tracing::info!(%hive, %agent, "github token deleted");
|
||||
Ok(StatusCode::NO_CONTENT)
|
||||
|
|
|
|||
|
|
@ -55,6 +55,7 @@ mod held_deploys;
|
|||
mod inference;
|
||||
mod issue_report;
|
||||
mod jobq;
|
||||
mod link_egress;
|
||||
mod linked_accounts;
|
||||
mod matrix_account;
|
||||
mod message_stream;
|
||||
|
|
|
|||
|
|
@ -180,6 +180,15 @@ pub async fn put_matrix_account(
|
|||
let user_id = link_matrix(&store, &secret_path, &existing(&agent, &account), &req)
|
||||
.await
|
||||
.map_err(|b| *b)?;
|
||||
crate::link_egress::added(
|
||||
&state,
|
||||
&agent,
|
||||
&crate::link_egress::Link::Matrix {
|
||||
account: &account,
|
||||
homeserver: req.homeserver.as_deref(),
|
||||
},
|
||||
)
|
||||
.await?;
|
||||
|
||||
tracing::info!(%hive, %agent, %account, "credential stored");
|
||||
Ok(Json(PutMatrixAccountResponse { user_id }))
|
||||
|
|
|
|||
|
|
@ -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);
|
||||
spawn_backend_egress_backfill(status, hives);
|
||||
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 {
|
||||
|
|
@ -406,11 +406,14 @@ fn spawn_agent_permissions(status: Option<&Arc<status::StatusReader>>, 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());
|
||||
}
|
||||
/// longer name their host, and make the entries derived from linked accounts
|
||||
/// match the stored accounts, when a swarm queue is wired up.
|
||||
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);
|
||||
}
|
||||
|
||||
/// Start rewriting every agent's schedule key from the definitions when a
|
||||
|
|
|
|||
|
|
@ -45,11 +45,48 @@ pub struct Entry {
|
|||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
||||
pub agent: Option<String>,
|
||||
/// The controller derived this entry from the agent's backend environment
|
||||
/// and replaces it whenever that changes; an operator can neither add nor
|
||||
/// remove one. A hive renders it like any other entry, and a reader that
|
||||
/// predates the field ignores it.
|
||||
/// or a linked account and replaces it whenever that changes; an operator
|
||||
/// can neither add nor remove one. A hive renders it like any other entry,
|
||||
/// and a reader that predates the field ignores it.
|
||||
#[serde(default, skip_serializing_if = "std::ops::Not::not")]
|
||||
pub derived: bool,
|
||||
/// What a derived entry follows: [`BACKEND_ENV_SOURCE`], or a name starting
|
||||
/// with [`LINK_SOURCE_PREFIX`] for an account the operator linked. Absent
|
||||
/// on an entry written before the field existed, which counts as
|
||||
/// [`BACKEND_ENV_SOURCE`]. A reader that predates the field ignores it.
|
||||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
||||
pub source: Option<String>,
|
||||
}
|
||||
|
||||
/// The [`Entry::source`] of an entry derived from the agent's backend
|
||||
/// environment.
|
||||
pub const BACKEND_ENV_SOURCE: &str = "backend-env";
|
||||
|
||||
/// What the [`Entry::source`] of an entry derived from a linked account starts
|
||||
/// with: `link:github`, `link:matrix:<account>` or `link:forge:<label>`.
|
||||
pub const LINK_SOURCE_PREFIX: &str = "link:";
|
||||
|
||||
impl Entry {
|
||||
/// Whether the controller derived this entry from the agent's backend
|
||||
/// environment, which an entry without a source counts as.
|
||||
#[must_use]
|
||||
pub fn from_backend_env(&self) -> bool {
|
||||
self.derived
|
||||
&& self
|
||||
.source
|
||||
.as_deref()
|
||||
.is_none_or(|s| s == BACKEND_ENV_SOURCE)
|
||||
}
|
||||
|
||||
/// Whether the controller derived this entry from a linked account.
|
||||
#[must_use]
|
||||
pub fn from_link(&self) -> bool {
|
||||
self.derived
|
||||
&& self
|
||||
.source
|
||||
.as_deref()
|
||||
.is_some_and(|s| s.starts_with(LINK_SOURCE_PREFIX))
|
||||
}
|
||||
}
|
||||
|
||||
/// The value under [`BUCKET`]/[`KEY`]. No `deny_unknown_fields`: a field a
|
||||
|
|
@ -229,11 +266,13 @@ mod tests {
|
|||
domain: "github.com".to_owned(),
|
||||
agent: None,
|
||||
derived: false,
|
||||
source: None,
|
||||
},
|
||||
Entry {
|
||||
domain: "crates.io".to_owned(),
|
||||
agent: Some("atlas".to_owned()),
|
||||
derived: false,
|
||||
source: None,
|
||||
},
|
||||
],
|
||||
};
|
||||
|
|
@ -255,6 +294,7 @@ mod tests {
|
|||
domain: "llm.example.com".to_owned(),
|
||||
agent: Some("atlas".to_owned()),
|
||||
derived: true,
|
||||
source: None,
|
||||
};
|
||||
let wire = r#"{"domain":"llm.example.com","agent":"atlas","derived":true}"#;
|
||||
assert_eq!(serde_json::to_string(&derived).unwrap(), wire);
|
||||
|
|
@ -265,4 +305,32 @@ mod tests {
|
|||
assert!(!manual.derived);
|
||||
assert_eq!(manual.domain, "a.io");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_source_round_trips_and_an_entry_without_one_is_from_the_backend_env() {
|
||||
let entry = |source: Option<&str>| Entry {
|
||||
domain: "github.com".to_owned(),
|
||||
agent: Some("atlas".to_owned()),
|
||||
derived: true,
|
||||
source: source.map(str::to_owned),
|
||||
};
|
||||
let linked = entry(Some("link:github"));
|
||||
let wire =
|
||||
r#"{"domain":"github.com","agent":"atlas","derived":true,"source":"link:github"}"#;
|
||||
assert_eq!(serde_json::to_string(&linked).unwrap(), wire);
|
||||
assert_eq!(serde_json::from_str::<Entry>(wire).unwrap(), linked);
|
||||
assert!(linked.from_link() && !linked.from_backend_env());
|
||||
|
||||
assert!(entry(None).from_backend_env() && !entry(None).from_link());
|
||||
assert!(entry(Some(BACKEND_ENV_SOURCE)).from_backend_env());
|
||||
// A source this build does not know belongs to neither.
|
||||
let other = entry(Some("later:x"));
|
||||
assert!(!other.from_backend_env() && !other.from_link());
|
||||
// Only a derived entry has a source.
|
||||
let manual = Entry {
|
||||
derived: false,
|
||||
..entry(Some("link:github"))
|
||||
};
|
||||
assert!(!manual.from_backend_env() && !manual.from_link());
|
||||
}
|
||||
}
|
||||
|
|
|
|||
Loading…
Reference in a new issue