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

hive-c0re: provision rooms through swarm-matrix-client

Browse source
This commit is contained in:
atlas 2026-10-18 07:22:41 +02:00 • committed by mara
commit 1d22041f82
5 changed files with 530 additions and 254 deletions
Download patch file Download diff file Expand all files Collapse all files

1
hive-c0re/Cargo.toml
View file

@ -61,6 +61,7 @@ swarm-queue-client = { workspace = true, features = [
"agent-permissions",
"agent-archetype",
] }
swarm-matrix-client.workspace = true
swarm-secret-client.workspace = true
tokio.workspace = true
tokio-stream.workspace = true

400
hive-c0re/src/matrix.rs
View file

@ -10,7 +10,6 @@
use std::path::PathBuf;
use anyhow::{Context, Result};
use reqwest::StatusCode;
/// Client-server API base this daemon provisions against, from
/// `HIVE_MATRIX_API_URL` (set by the hyperhive NixOS module from
@ -46,8 +45,6 @@ fn matrix_base() -> Result<&'static str> {
this path should have been gated on matrix::is_present()",
)
}
/// HTTP timeout for the sweep's homeserver round-trips.
const HTTP_TIMEOUT_SECS: u64 = 10;
/// Matrix localpart this hive acts as. Not an agent; has no state dir.
///
@ -130,12 +127,6 @@ pub fn is_present() -> bool {
matrix_http().is_some()
}
/// Percent-encode a matrix room ID for use in a URL path segment.
/// Only `:` needs encoding; `!` and alphanumerics are path-safe.
fn encode_room_id_for_url(room_id: &str) -> String {
room_id.replace(':', "%3A")
}
/// Bring the hive's `@hive-<hive>:` sender token at [`sender_token_path()`] in
/// line with the swarm secret store.
///
@ -318,57 +309,43 @@ async fn find_joined_room(
want_space: bool,
name: &str,
) -> Option<String> {
let joined_url = format!("{base}/_matrix/client/v3/joined_rooms");
let joined: serde_json::Value = client
.get(&joined_url)
.bearer_auth(sender_token)
.send()
.await
.ok()?
.json()
let rooms = swarm_matrix_client::joined_rooms(client, base, sender_token)
.await
.ok()?;
let rooms = joined["joined_rooms"].as_array()?;
for room in rooms {
let Some(room_id) = room.as_str() else {
continue;
};
let encoded = encode_room_id_for_url(room_id);
for room_id in rooms {
// m.room.create `type` says whether this is a Space.
let create_url = format!("{base}/_matrix/client/v3/rooms/{encoded}/state/m.room.create/");
let is_space = match client
.get(&create_url)
.bearer_auth(sender_token)
.send()
let is_space = keyless_state(client, base, sender_token, &room_id, "m.room.create")
.await
{
Ok(r) if r.status().is_success() => r
.json::<serde_json::Value>()
.await
.ok()
.is_some_and(|c| c["type"].as_str() == Some("m.space")),
_ => false,
};
.is_some_and(|c| c["type"].as_str() == Some("m.space"));
if is_space != want_space {
continue;
}
// …and named `name` (m.room.name `name`).
let name_url = format!("{base}/_matrix/client/v3/rooms/{encoded}/state/m.room.name/");
let name_matches = match client.get(&name_url).bearer_auth(sender_token).send().await {
Ok(r) if r.status().is_success() => r
.json::<serde_json::Value>()
.await
.ok()
.is_some_and(|n| n["name"].as_str() == Some(name)),
_ => false,
};
let name_matches = keyless_state(client, base, sender_token, &room_id, "m.room.name")
.await
.is_some_and(|n| n["name"].as_str() == Some(name));
if name_matches {
return Some(room_id.to_owned());
return Some(room_id);
}
}
None
}
/// The content of a state event with an empty state key, or `None` when the
/// room has none or the read failed.
async fn keyless_state(
client: &reqwest::Client,
base: &str,
sender_token: &str,
room_id: &str,
event_type: &str,
) -> Option<serde_json::Value> {
swarm_matrix_client::room_state(client, base, sender_token, room_id, event_type, "")
.await
.ok()
.flatten()
}
/// Scan the admin account's joined rooms for the canonical hive Space: the
/// `m.space` whose name is [`HIVE_SPACE_NAME`]. Returns the first match
/// (deterministic per the homeserver's joined-rooms order) so a lost
@ -417,32 +394,15 @@ pub async fn ensure_hive_space(client: &reqwest::Client, sender_token: &str) ->
}
// 3. Create the space (plain hardcoded name, no alias).
let url = format!("{base}/_matrix/client/v3/createRoom");
let body = serde_json::json!({
"name": HIVE_SPACE_NAME,
"creation_content": { "type": "m.space" },
"preset": "private_chat",
"visibility": "private",
});
let resp = client
.post(&url)
.bearer_auth(sender_token)
.json(&body)
.send()
let room_id = swarm_matrix_client::create_room(client, base, sender_token, &body)
.await
.context("matrix: POST /createRoom (hive space)")?;
let status = resp.status();
let json = resp
.json::<serde_json::Value>()
.await
.context("matrix: parse /createRoom response")?;
if !status.is_success() {
anyhow::bail!("matrix: createRoom HTTP {status}, body: {json}");
}
let room_id = json["room_id"]
.as_str()
.with_context(|| format!("matrix: createRoom missing room_id: {json}"))?
.to_owned();
.context("matrix: create hive space")?;
persist_space_room_id(&room_id)?;
tracing::info!(%room_id, "matrix: created hive space");
Ok(room_id)
@ -485,43 +445,40 @@ fn state_needs_write(current: Option<&serde_json::Value>, desired: &serde_json::
}
/// Read the room's current content for one state event. `None` on any
/// failure — absent state, transport error, or an unparseable body are
/// all "we don't know", and [`state_needs_write`] turns that into a write.
/// failure — absent state, transport error, a refusal, or an unparseable body
/// are all "we don't know", and [`state_needs_write`] turns that into a write.
///
/// One failure is louder than the rest, and it is the only one that hides:
/// **a non-404 HTTP failure means the read broke while the write still
/// works**, so the guard degrades to writing every time and the sweep
/// resumes waking the hive with nothing else to show for it. A transport
/// error needs no warning of its own — the PUT immediately after it fails
/// too, loudly — and a 404 is the expected first-setup case.
/// A failure other than absent state is logged at `warn`: **a read that breaks
/// while the write still works** makes the guard degrade to writing every
/// time, and the sweep resumes waking the hive with nothing else to show for
/// it. An absent state event (404) is the expected first-setup case.
async fn current_room_state(
client: &reqwest::Client,
base: &str,
sender_token: &str,
url: &str,
room_id: &str,
event_type: &str,
state_key: &str,
) -> Option<serde_json::Value> {
let resp = match client.get(url).bearer_auth(sender_token).send().await {
Ok(resp) => resp,
match swarm_matrix_client::room_state(
client,
base,
sender_token,
room_id,
event_type,
state_key,
)
.await
{
Ok(content) => content,
Err(e) => {
tracing::debug!(error = ?e, url, "matrix: state read unreachable; writing");
return None;
}
};
let status = resp.status();
if !status.is_success() {
if status != reqwest::StatusCode::NOT_FOUND {
tracing::warn!(
%status,
url,
error = %format!("{e:#}"),
%room_id,
event_type,
"matrix: state read failed while writes still work — the \
skip-if-unchanged guard is off and every sweep will re-emit"
);
}
return None;
}
match resp.json::<serde_json::Value>().await {
Ok(body) => Some(body),
Err(e) => {
tracing::debug!(error = ?e, url, "matrix: state read body unparseable; writing");
None
}
}
@ -549,11 +506,8 @@ async fn set_room_state(
content: &serde_json::Value,
) -> Result<()> {
let base = matrix_base()?;
let encoded_room = encode_room_id_for_url(room_id);
let encoded_key = encode_room_id_for_url(state_key);
let url =
format!("{base}/_matrix/client/v3/rooms/{encoded_room}/state/{event_type}/{encoded_key}");
let current = current_room_state(client, sender_token, &url).await;
let current =
current_room_state(client, base, sender_token, room_id, event_type, state_key).await;
if !state_needs_write(current.as_ref(), content) {
tracing::debug!(
%room_id,
@ -562,19 +516,17 @@ async fn set_room_state(
);
return Ok(());
}
let resp = client
.put(&url)
.bearer_auth(sender_token)
.json(content)
.send()
.await
.with_context(|| format!("matrix: PUT state {event_type} into {room_id}"))?;
let status = resp.status();
if status.is_success() {
return Ok(());
}
let body = resp.json::<serde_json::Value>().await.unwrap_or_default();
anyhow::bail!("matrix: set state {event_type} in {room_id}: HTTP {status}, body: {body}")
swarm_matrix_client::put_state(
client,
base,
sender_token,
room_id,
event_type,
state_key,
content,
)
.await
.context("matrix: set room state")
}
/// Create (or recover) the default hive chat room and wire it as a child of
@ -622,7 +574,6 @@ pub async fn ensure_hive_chat_room(
} else {
// `initial_state` is applied after the preset-derived state, so the
// restricted join rule overrides private_chat's invite-only default.
let url = format!("{base}/_matrix/client/v3/createRoom");
let body = serde_json::json!({
"name": HIVE_CHAT_ROOM_NAME,
"topic": HIVE_CHAT_ROOM_TOPIC,
@ -653,25 +604,9 @@ pub async fn ensure_hive_chat_room(
}
]
});
let resp = client
.post(&url)
.bearer_auth(sender_token)
.json(&body)
.send()
let id = swarm_matrix_client::create_room(client, base, sender_token, &body)
.await
.context("matrix: POST /createRoom (hive chat room)")?;
let status = resp.status();
let json = resp
.json::<serde_json::Value>()
.await
.context("matrix: parse /createRoom response (chat room)")?;
if !status.is_success() {
anyhow::bail!("matrix: createRoom (chat) HTTP {status}, body: {json}");
}
let id = json["room_id"]
.as_str()
.with_context(|| format!("matrix: createRoom (chat) missing room_id: {json}"))?
.to_owned();
.context("matrix: create hive chat room")?;
persist_chat_room_id(&id)?;
tracing::info!(room_id = %id, "matrix: created hive chat room");
id
@ -719,36 +654,30 @@ async fn invite_to_room(
async fn room_membership(
client: &reqwest::Client,
sender_token: &str,
encoded_room_id: &str,
room_id: &str,
user_id: &str,
) -> Option<String> {
let base = matrix_http()?;
// `:` must be percent-encoded in both the room-id and user-id path
// segments; `@` and `!` are permitted path characters per RFC 3986.
let encoded_user = user_id.replace(':', "%3A");
let url = format!(
"{base}/_matrix/client/v3/rooms/{encoded_room_id}/state/m.room.member/{encoded_user}"
);
let resp = client
.get(&url)
.bearer_auth(sender_token)
.send()
.await
.ok()?;
if !resp.status().is_success() {
// 404 = no membership event yet; anything else we treat as "unknown"
// and let the caller fall through to the invite attempt.
return None;
}
let body = resp.json::<serde_json::Value>().await.ok()?;
body["membership"].as_str().map(ToOwned::to_owned)
// A refused or failed read is "unknown": the caller falls through to the
// invite attempt.
let member = swarm_matrix_client::room_state(
client,
base,
sender_token,
room_id,
"m.room.member",
user_id,
)
.await
.ok()??;
member["membership"].as_str().map(ToOwned::to_owned)
}
/// Invite a fully-qualified Matrix user id (`@user:server`) to `room_id`
/// using the sender token. Idempotent: a user who is already a member or
/// already has a pending invite is left untouched (no fresh invite is sent,
/// so they are not re-notified). A refused invite is an error unless the
/// user turns out to be invited or joined anyway — see [`refused_invite`].
/// user turns out to be invited or joined anyway — see [`forbidden_invite`].
async fn invite_user_id(
client: &reqwest::Client,
sender_token: &str,
@ -756,61 +685,45 @@ async fn invite_user_id(
user_id: &str,
) -> Result<()> {
let base = matrix_base()?;
// `:` must be percent-encoded in the room-id path segment; `!` is
// permitted in URL path characters per RFC 3986.
let encoded_room_id = room_id.replace(':', "%3A");
// Skip the invite entirely when the user is already invited or joined.
// Re-POSTing an invite to a pending member re-sends the invite event,
// which re-notifies the agent on every provisioning sweep.
if let Some(membership) = room_membership(client, sender_token, &encoded_room_id, user_id).await
if let Some(membership) = room_membership(client, sender_token, room_id, user_id).await
&& matches!(membership.as_str(), "invite" | "join")
{
tracing::debug!(%user_id, %room_id, %membership, "matrix: invite skipped (already a member/invited)");
return Ok(());
}
let url = format!("{base}/_matrix/client/v3/rooms/{encoded_room_id}/invite");
let resp = client
.post(&url)
.bearer_auth(sender_token)
.json(&serde_json::json!({ "user_id": user_id }))
.send()
match swarm_matrix_client::invite(client, base, sender_token, room_id, user_id)
.await
.with_context(|| format!("matrix: POST /rooms/.../invite for {user_id}"))?;
let status = resp.status();
if status.is_success() {
tracing::debug!(%user_id, %room_id, "matrix: invited to room");
return Ok(());
.context("matrix: invite")?
{
swarm_matrix_client::Invited::Sent => {
tracing::debug!(%user_id, %room_id, "matrix: invited to room");
Ok(())
}
swarm_matrix_client::Invited::Forbidden => {
// The pre-check misses a member when its read failed or a
// concurrent invite landed after it, so a refusal re-reads the
// membership.
let membership = room_membership(client, sender_token, room_id, user_id).await;
forbidden_invite(membership.as_deref(), user_id, room_id)
}
}
// The pre-check misses a member when its read failed or a concurrent
// invite landed after it, so a refusal re-reads the membership.
let membership = if status == StatusCode::FORBIDDEN {
room_membership(client, sender_token, &encoded_room_id, user_id).await
} else {
None
};
refused_invite(resp, membership.as_deref(), user_id, room_id).await
}
/// Outcome of an invite POST the homeserver did not accept. A 403 is success
/// only when `membership`, re-read after the refusal, shows the user invited
/// or joined: `M_FORBIDDEN` covers both "already in the room" and real
/// refusals (banned target, sender lacks power), so the errcode can't tell
/// them apart.
async fn refused_invite(
resp: reqwest::Response,
membership: Option<&str>,
user_id: &str,
room_id: &str,
) -> Result<()> {
let status = resp.status();
if status == StatusCode::FORBIDDEN && matches!(membership, Some("invite" | "join")) {
/// Outcome of an invite the homeserver refused with 403. That is success only
/// when `membership`, re-read after the refusal, shows the user invited or
/// joined: `M_FORBIDDEN` covers both "already in the room" and real refusals
/// (banned target, sender lacks power), so the errcode can't tell them apart.
fn forbidden_invite(membership: Option<&str>, user_id: &str, room_id: &str) -> Result<()> {
if matches!(membership, Some("invite" | "join")) {
tracing::debug!(%user_id, %room_id, "matrix: invite refused, user already member/invited");
return Ok(());
}
let body = resp.json::<serde_json::Value>().await.unwrap_or_default();
anyhow::bail!("matrix: invite {user_id} to {room_id}: HTTP {status}, body: {body}")
anyhow::bail!("matrix: invite {user_id} to {room_id}: HTTP 403 and the user is not invited or joined")
}
/// Invite an arbitrary Matrix user to the hive Space (default) or an
@ -841,7 +754,9 @@ pub async fn invite_user(
// Resolve the room: explicit override (id or #alias) wins; otherwise
// the persisted hive Space.
let room_id = match room_override {
Some(r) if r.starts_with('#') => resolve_room_alias(client, sender_token, r).await?,
Some(r) if r.starts_with('#') => {
swarm_matrix_client::resolve_alias(client, matrix_base()?, sender_token, r).await?
}
Some(r) if r.starts_with('!') => r.to_owned(),
Some(r) => anyhow::bail!(
"matrix: --room {r:?} is neither a room id nor an alias; \
@ -859,32 +774,6 @@ pub async fn invite_user(
Ok(room_id)
}
/// Resolve a `#alias:server` to its room id via the directory API.
async fn resolve_room_alias(
client: &reqwest::Client,
sender_token: &str,
alias: &str,
) -> Result<String> {
let base = matrix_base()?;
let encoded = alias.replace('#', "%23").replace(':', "%3A");
let url = format!("{base}/_matrix/client/v3/directory/room/{encoded}");
let resp = client
.get(&url)
.bearer_auth(sender_token)
.send()
.await
.with_context(|| format!("matrix: GET directory for {alias}"))?;
let status = resp.status();
let json = resp.json::<serde_json::Value>().await.unwrap_or_default();
if !status.is_success() {
anyhow::bail!("matrix: resolve alias {alias}: HTTP {status}, body: {json}");
}
json["room_id"]
.as_str()
.map(str::to_owned)
.ok_or_else(|| anyhow::anyhow!("matrix: alias {alias} response missing room_id: {json}"))
}
/// Make sure the hive's own `@hive-<hive>:` account exists, then provision
/// the hive Space and chat room and invite every agent container to both.
/// Agents' own accounts are not created here: `swarm-controller` mints them
@ -905,10 +794,7 @@ pub async fn ensure_all() -> bool {
}
let mut ok = true;
// One HTTP client for the whole sweep.
let client = match reqwest::Client::builder()
.timeout(std::time::Duration::from_secs(HTTP_TIMEOUT_SECS))
.build()
{
let client = match swarm_matrix_client::client() {
Ok(c) => c,
Err(e) => {
tracing::warn!(error = ?e, "matrix: build HTTP client failed; skipping sweep");
@ -1105,54 +991,60 @@ mod tests {
assert!(state_needs_write(None, &desired));
}
/// A homeserver response built in memory, so no client (and no TLS
/// roots) is needed to drive the response-handling halves.
fn response(status: u16, body: &'static str) -> reqwest::Response {
axum::http::Response::builder()
.status(status)
.body(body)
.expect("valid mock response")
.into()
}
const FORBIDDEN_BODY: &str = r#"{"errcode":"M_FORBIDDEN","error":"refused"}"#;
/// A banned target or a sender without power gets exactly this 403;
/// with no membership behind it, the invite did not happen.
#[tokio::test]
async fn a_refused_invite_without_membership_is_an_error() {
/// A banned target or a sender without power gets a 403; with no
/// membership behind it, the invite did not happen.
#[test]
fn a_refused_invite_without_membership_is_an_error() {
for membership in [None, Some("ban"), Some("leave")] {
let outcome =
refused_invite(response(403, FORBIDDEN_BODY), membership, "@a:x", "!r:x").await;
let outcome = forbidden_invite(membership, "@a:x", "!r:x");
assert!(outcome.is_err(), "membership {membership:?} must not pass");
}
}
#[tokio::test]
async fn a_refused_invite_for_a_member_is_success() {
#[test]
fn a_refused_invite_for_a_member_is_success() {
for membership in ["invite", "join"] {
refused_invite(
response(403, FORBIDDEN_BODY),
Some(membership),
"@a:x",
"!r:x",
)
.await
.expect("already invited/joined is the goal state");
forbidden_invite(Some(membership), "@a:x", "!r:x")
.expect("already invited/joined is the goal state");
}
}
/// Membership only excuses a 403; any other failure stays a failure.
/// The homeserver's reply body reaches no error: `invite_user_id` is
/// driven against a stub whose 500 and 403 bodies carry a marker.
#[tokio::test]
async fn a_non_403_invite_failure_or_garbage_body_is_an_error() {
for (status, body) in [(500, "not json"), (403, "not json"), (429, "{}")] {
let outcome = refused_invite(response(status, body), None, "@a:x", "!r:x").await;
assert!(outcome.is_err(), "HTTP {status} {body:?} must fail");
async fn an_invite_refusal_error_does_not_quote_the_response_body() {
const BODY: &str = r#"{"errcode":"M_UNKNOWN","error":"free-form-marker"}"#;
for status in [500_u16, 403] {
let app = axum::Router::new().fallback(move || async move {
(
axum::http::StatusCode::from_u16(status).expect("a status"),
[(axum::http::header::CONTENT_TYPE, "application/json")],
BODY,
)
});
let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
.await
.expect("bind");
let base = format!("http://{}", listener.local_addr().expect("local addr"));
let server = tokio::spawn(async move {
axum::serve(listener, app).await.expect("serve");
});
let client = swarm_matrix_client::client().expect("client");
let outcome = swarm_matrix_client::invite(&client, &base, "tok", "!r:x", "@a:x").await;
server.abort();
let message = match (status, outcome) {
(403, Ok(swarm_matrix_client::Invited::Forbidden)) => {
format!("{:#}", forbidden_invite(None, "@a:x", "!r:x").unwrap_err())
}
(500, Err(e)) => format!("{e:#}"),
(status, other) => panic!("HTTP {status}: unexpected {other:?}"),
};
assert!(message.contains(&status.to_string()), "{message}");
assert!(!message.contains("free-form-marker"), "{message}");
}
let outcome = refused_invite(response(500, "{}"), Some("join"), "@a:x", "!r:x").await;
assert!(outcome.is_err(), "a 500 is not excused by membership");
}
/// Serve a fake homeserver on loopback: `!space` is an `m.space` and
/// `!chat` a plain room, both named "shared"; `!decoy` is a plain room
/// named "other". Returns the base URL and the server task.