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

stop provisioning the hive-chat room in the matrix sweep

Browse source
This commit is contained in:
damocles 2026-10-28 20:39:15 +01:00
commit c7f3706597
5 changed files with 37 additions and 371 deletions
Download patch file Download diff file Expand all files Collapse all files

3
hive-c0re/src/job_queue/exec.rs
View file

@ -412,8 +412,7 @@ fn matrix_sweep_warning(ctx: sweep_health::SweepFailure) -> String {
);
format!(
"matrix user/space sweep failing ({} consecutive, {age}) \
— some agents may be missing matrix accounts, space membership, \
or chat-room invites",
— some agents may be missing matrix accounts or space membership",
ctx.consecutive
)
}

2
hive-c0re/src/job_queue/resource.rs
View file

@ -32,7 +32,7 @@ pub enum Resource {
/// `.git/index.lock` and the remote-tracking ref lock.
KnowledgeTree,
/// The hive's matrix provisioning, held by every `MatrixSweep`. Two
/// overlapping passes on a hive with no persisted Space / chat-room id both
/// overlapping passes on a hive with no persisted Space id both
/// miss the by-name lookup and both `createRoom`, leaving a duplicate room
/// with agents invited to both.
MatrixSweep,

2
hive-c0re/src/main.rs
View file

@ -308,7 +308,7 @@ async fn cmd_serve(
}
});
// Matrix sweep (`matrix::ensure_all`): the hive's own account, the hive
// Space and chat room, and every agent container's invite to both. It
// Space, and every agent container's invite to it. It
// creates no agent accounts or tokens — swarm-controller mints those.
// No-op when the hive-matrix container isn't running. Re-submitted every
// 30 minutes; the startup pass is the boot `MatrixSweep` node.

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

@ -1,5 +1,5 @@
//! Optional matrix-tuwunel wiring: the hive's `@hive-<hive>:` sender token,
//! taken from the swarm secret store, and the hive Space, chat room and
//! taken from the swarm secret store, and the hive Space and
//! invites provisioned with it. No-op when no homeserver is configured, so
//! operators who haven't flipped `services.hyperhive.deploy.matrix.enable =
//! true` pay nothing.
@ -60,8 +60,8 @@ fn matrix_base() -> Result<&'static str> {
/// [`stored_sender_token`] reads it; hive-c0re never creates it.
///
/// An ordinary account, with no homeserver-admin standing: what it
/// provisions — the Space, the chat room, the invites — it provisions as
/// the creator of those rooms.
/// provisions — the Space and the invites — it provisions as the
/// creator of the Space.
///
/// # Errors
/// When `HYPERHIVE_HIVE_NAME` is unset or empty. That is the same variable
@ -86,16 +86,6 @@ pub fn hive_localpart() -> Result<String> {
/// being created on the next sweep.
pub const HIVE_SPACE_NAME: &str = "hive";
/// Display name of the default "hive chat" room — the `m.space.child` of
/// the hive Space that every agent + the operator can join. Plain text so
/// it stays rediscoverable by name (mirrors [`HIVE_SPACE_NAME`]) when the
/// persisted room-id file is lost, preventing duplicate chat rooms.
pub const HIVE_CHAT_ROOM_NAME: &str = "hive-chat";
/// Topic for the default hive chat room.
const HIVE_CHAT_ROOM_TOPIC: &str =
"Hive-wide chat for all agents and the operator. Auto-provisioned by hive-c0re.";
/// Host path for the appservice sender account's matrix access token. Outside every
/// purgeable path — not deleted by `destroy --purge` on any agent.
#[must_use]
@ -110,13 +100,6 @@ pub fn hive_space_room_id_path() -> PathBuf {
crate::paths::matrix_space_room_id()
}
/// Host path where the default hive chat room ID is persisted. Outside
/// every purgeable path — not deleted by `destroy --purge`.
#[must_use]
pub fn hive_chat_room_id_path() -> PathBuf {
crate::paths::matrix_chat_room_id()
}
/// Whether this hive has a homeserver to provision against.
///
/// **A configured API URL, not a local container**: whether the homeserver
@ -408,233 +391,6 @@ pub async fn ensure_hive_space(client: &reqwest::Client, sender_token: &str) ->
Ok(room_id)
}
/// Persist the hive chat room id to [`hive_chat_room_id_path()`] (0600).
fn persist_chat_room_id(room_id: &str) -> Result<()> {
let path = hive_chat_room_id_path();
crate::secret_file::write_secret_0600(&path, format!("{room_id}\n").as_bytes())
.with_context(|| format!("matrix: write chat room_id to {}", path.display()))
}
/// Scan the admin account's joined rooms for the canonical hive chat room:
/// a non-space room named [`HIVE_CHAT_ROOM_NAME`]. Mirrors
/// [`find_space_by_name`] so a lost room-id file recovers the existing chat
/// room instead of spawning a duplicate. `None` if the homeserver is
/// unreachable or no match exists.
async fn find_chat_room_by_name(client: &reqwest::Client, sender_token: &str) -> Option<String> {
find_joined_room(
client,
matrix_http()?,
sender_token,
false,
HIVE_CHAT_ROOM_NAME,
)
.await
}
/// Whether the state event still has to be written.
///
/// `current` is the room's existing content for this `(type, state_key)`,
/// or `None` when it could not be read. **Unreadable means write**: the
/// re-apply exists so a missing link gets repaired, and one redundant
/// event is cheaper than a hierarchy that never reconverges.
///
/// Comparison is `serde_json::Value` equality, which compares objects as
/// maps — key order in the response does not matter.
fn state_needs_write(current: Option<&serde_json::Value>, desired: &serde_json::Value) -> bool {
current != Some(desired)
}
/// Read the room's current content for one state event. `None` on any
/// 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.
///
/// 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,
room_id: &str,
event_type: &str,
state_key: &str,
) -> Option<serde_json::Value> {
match swarm_matrix_client::room_state(
client,
base,
sender_token,
room_id,
event_type,
state_key,
)
.await
{
Ok(content) => content,
Err(e) => {
tracing::warn!(
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"
);
None
}
}
}
/// PUT a state event into `room_id` using the sender token, **skipping the
/// write when the room already carries identical content**.
///
/// The read is not an optimisation. A PUT of identical content is a no-op
/// on the room's *state*, and the homeserver still appends an event to the
/// *timeline* — so "idempotent" was true one level too high. Downstream,
/// an event is unread activity, which is a todo, which is a turn: a caller
/// re-applying a link on a periodic sweep wakes every agent in the room on
/// that sweep's cadence, forever. Measured at ~30 minutes per wake per
/// agent before this guard existed.
///
/// The self-healing property the re-apply exists for is unaffected: a
/// missing or divergent link still gets written.
async fn set_room_state(
client: &reqwest::Client,
sender_token: &str,
room_id: &str,
event_type: &str,
state_key: &str,
content: &serde_json::Value,
) -> Result<()> {
let base = matrix_base()?;
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,
event_type,
"matrix: state already current, skipping PUT (no timeline event)"
);
return Ok(());
}
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
/// the hive Space, persisting its room id to [`hive_chat_room_id_path()`].
///
/// The room is a normal room (not an `m.space`) named [`HIVE_CHAT_ROOM_NAME`]
/// with a `restricted` join rule allowing any member of the hive Space to
/// join — so the operator (who is in the Space) and every agent can chat
/// without needing an explicit invite. It's linked bidirectionally to the
/// Space: `m.space.child` on the Space points at the room, `m.space.parent`
/// on the room points back. Joining a Space does NOT auto-join its children
/// (Matrix semantics), so this gives clients a concrete room to surface +
/// join instead of an empty Space.
///
/// Dedup mirrors [`ensure_hive_space`]: persisted id wins, else rediscover
/// by name, else create. The space-child link is re-checked on every call
/// so a recovered room reconverges its hierarchy link — and written only
/// when it differs, see [`set_room_state`].
///
/// # Errors
///
/// Returns an error if the homeserver is unreachable, `createRoom` fails,
/// or the room-ID file cannot be written. A failure wiring the space-child
/// link is logged but not fatal (the room still exists + is joinable).
pub async fn ensure_hive_chat_room(
client: &reqwest::Client,
sender_token: &str,
space_room_id: &str,
server_name: &str,
) -> Result<String> {
let base = matrix_base()?;
// 1. Stored room id wins (fast path). 2. Rediscover by name before
// creating (prevents duplicates after a state wipe). 3. Create.
let room_id = if let Some(id) = std::fs::read_to_string(hive_chat_room_id_path())
.ok()
.map(|s| s.trim().to_owned())
.filter(|s| !s.is_empty())
{
tracing::debug!(room_id = %id, "matrix: hive chat room already provisioned");
id
} else if let Some(id) = find_chat_room_by_name(client, sender_token).await {
persist_chat_room_id(&id)?;
tracing::info!(room_id = %id, "matrix: recovered hive chat room by name");
id
} else {
// `initial_state` is applied after the preset-derived state, so the
// restricted join rule overrides private_chat's invite-only default.
let body = serde_json::json!({
"name": HIVE_CHAT_ROOM_NAME,
"topic": HIVE_CHAT_ROOM_TOPIC,
// Pin the room version explicitly: the `restricted` join rule
// below needs room version >= 8. tuwunel's default is currently
// higher, but pinning makes the dependency explicit so a future
// homeserver-default change can't silently invalidate the
// restricted rule (which would quietly fall back to invite-only
// and break the "operator joins from the Space" path).
"room_version": "10",
"preset": "private_chat",
"visibility": "private",
"initial_state": [
{
"type": "m.room.join_rules",
"state_key": "",
"content": {
"join_rule": "restricted",
"allow": [
{ "type": "m.room_membership", "room_id": space_room_id }
]
}
},
{
"type": "m.space.parent",
"state_key": space_room_id,
"content": { "via": [server_name], "canonical": true }
}
]
});
let id = swarm_matrix_client::create_room(client, base, sender_token, &body)
.await
.context("matrix: create hive chat room")?;
persist_chat_room_id(&id)?;
tracing::info!(room_id = %id, "matrix: created hive chat room");
id
};
// Wire the Space → room child link. Without an `m.space.child` carrying
// a `via`, the room won't surface in the Space hierarchy. `suggested`
// hints clients to surface it prominently.
let child_content = serde_json::json!({
"via": [server_name],
"suggested": true,
});
if let Err(e) = set_room_state(
client,
sender_token,
space_room_id,
"m.space.child",
&room_id,
&child_content,
)
.await
{
tracing::warn!(error = ?e, "matrix: set m.space.child on hive space failed");
}
Ok(room_id)
}
/// Invite `@{localpart}:{server_name}` to `room_id` using the admin
/// account. Idempotent — treats already-member responses as success.
async fn invite_to_room(
@ -777,7 +533,7 @@ pub async fn invite_user(
}
/// 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.
/// the hive Space and invite every agent container to it.
/// Agents' own accounts are not created here: `swarm-controller` mints them
/// with the swarm's appservice token. Runs only as a `MatrixSweep` job node
/// (see `job_queue::templates::matrix_sweep`). No-op when the
@ -804,9 +560,8 @@ pub async fn ensure_all() -> bool {
}
};
// The hive's `@hive-<hive>:` account first, because everything below provisions
// THROUGH it (the Space, the chat room and every invite are sent with
// its token) — as an ordinary user that created those rooms, not as a
// homeserver admin.
// THROUGH it (the Space and every invite are sent with its token) — as
// an ordinary user that created the Space, not as a homeserver admin.
if let Err(e) = ensure_hive_user().await {
tracing::warn!(error = ?e, "matrix: ensure_hive_user failed");
ok = false;
@ -829,8 +584,8 @@ pub async fn ensure_all() -> bool {
ok
}
/// Provision the hive Space + default chat room and invite every agent
/// (+ the admin account) to both. Split out of [`ensure_all`] purely to
/// Provision the hive Space and invite every agent (+ the admin account)
/// to it. Split out of [`ensure_all`] purely to
/// keep that function under the `too_many_lines` threshold — this is the
/// tail half of the same sequential sweep and shares its aggregate-bool,
/// log-and-continue failure handling.
@ -878,40 +633,6 @@ async fn provision_space(client: &reqwest::Client, agent_names: &[String]) -> bo
ok = false;
}
}
// Provision the default hive chat room as an m.space.child of the Space
// and invite `@hive-<hive>:` + every agent. Joining a Space alone surfaces no
// rooms to chat in (Matrix semantics — children aren't auto-joined), so
// without this the Space is empty. The restricted join rule additionally
// lets the operator (a Space member) join from the Space hierarchy.
match ensure_hive_chat_room(client, &sender_token, &room_id, &server_name).await {
Ok(chat_room_id) => {
if let Err(e) = invite_to_room(
client,
&sender_token,
&chat_room_id,
&localpart,
&server_name,
)
.await
{
tracing::warn!(error = ?e, %localpart, "matrix: invite the hive account to the chat room failed");
ok = false;
}
for name in agent_names {
if let Err(e) =
invite_to_room(client, &sender_token, &chat_room_id, name, &server_name).await
{
tracing::warn!(%name, error = ?e, "matrix: invite agent to chat room failed");
ok = false;
}
}
}
Err(e) => {
tracing::warn!(error = ?e, "matrix: ensure_hive_chat_room failed");
ok = false;
}
}
ok
}
@ -956,43 +677,6 @@ mod tests {
assert_eq!(sender_source(None, None), SenderSource::Unavailable);
}
/// The steady state. This is the whole point of the guard: the sweep
/// runs forever, and every write it makes is a wake for every agent
/// in the room.
#[test]
fn state_matching_the_room_is_not_rewritten() {
let desired = serde_json::json!({ "via": ["example.org"], "suggested": true });
let current = desired.clone();
assert!(!state_needs_write(Some(&current), &desired));
}
/// Key order in the homeserver's response must not force a write —
/// the guard leans on `serde_json::Value` comparing objects as maps,
/// and a byte- or order-sensitive comparison would silently degrade
/// to writing every time while still looking correct.
#[test]
fn state_matching_but_reordered_is_not_rewritten() {
let desired = serde_json::json!({ "via": ["example.org"], "suggested": true });
let current = serde_json::json!({ "suggested": true, "via": ["example.org"] });
assert!(!state_needs_write(Some(&current), &desired));
}
#[test]
fn diverged_state_is_rewritten() {
let desired = serde_json::json!({ "via": ["example.org"], "suggested": true });
let current = serde_json::json!({ "via": ["old.example.org"], "suggested": true });
assert!(state_needs_write(Some(&current), &desired));
}
/// Fail-open, and deliberately so: an unreadable current state must
/// write. The re-apply exists to repair a missing link, so "we don't
/// know" has to behave like "it's missing", not like "it's fine".
#[test]
fn unknown_state_is_written() {
let desired = serde_json::json!({ "via": ["example.org"], "suggested": true });
assert!(state_needs_write(None, &desired));
}
/// A banned target or a sender without power gets a 403; with no
/// membership behind it, the invite did not happen.
#[test]

7
hive-c0re/src/paths.rs
View file

@ -110,13 +110,6 @@ pub fn matrix_space_room_id() -> PathBuf {
matrix_dir().join("space-room-id")
}
/// `matrix/chat-room-id` — persisted default "hive chat" room id (the
/// `m.space.child` of the hive Space every agent + the operator can join).
#[must_use]
pub fn matrix_chat_room_id() -> PathBuf {
matrix_dir().join("chat-room-id")
}
/// `run/` — runtime maps hive-c0re regenerates on every meta sync.
#[must_use]
pub fn run_dir() -> PathBuf {