From c41429e473b6e334a42b069dcc8b470b29aee987 Mon Sep 17 00:00:00 2001 From: damocles Date: Mon, 9 Nov 2026 14:43:32 +0100 Subject: [PATCH] hive-matrix-mcp: sync with sliding sync when the server has it, classic /sync otherwise --- docs/4-implementation/agent-tools/matrix.md | 14 ++ hive-matrix-mcp/src/lib.rs | 1 + hive-matrix-mcp/src/main.rs | 43 ++++-- hive-matrix-mcp/src/sliding.rs | 155 ++++++++++++++++++++ 4 files changed, 198 insertions(+), 15 deletions(-) create mode 100644 hive-matrix-mcp/src/sliding.rs diff --git a/docs/4-implementation/agent-tools/matrix.md b/docs/4-implementation/agent-tools/matrix.md index c7c007cf6..6597c6e68 100644 --- a/docs/4-implementation/agent-tools/matrix.md +++ b/docs/4-implementation/agent-tools/matrix.md @@ -115,6 +115,20 @@ Silently exits when the primary account has no token yet; the appears, and, on an agent with a store, a timer restarts it while it's down so it re-reads a token the swarm minted into the store. +**Sync protocol**: when the homeserver advertises sliding sync (MSC4186, +`org.matrix.simplified_msc3575` in `/versions`) the daemon uses it: one +growing list over every joined room, the newest event per room and only +the state it reads (name, alias, encryption, create, power levels, lazy +members), plus the receipts, to-device, E2EE and account-data extensions. +That's the protocol Element X uses, and it moves far fewer bytes than +classic `/sync`. A server without it gets classic `/sync` with a trimmed +filter (no presence or typing, lazy-loaded members, 20-event timelines). +After 5 failed sliding-sync polls in a row the daemon also falls back to +`/sync`. Set `HIVE_MATRIX_SYNC=classic` in the daemon's environment to +force `/sync`. The unread check doesn't depend on the protocol: it reads +the agent's own receipt from the store and the recent events with a live +`/messages` call. + **Wake path**: after every sync the daemon upserts a harness-local todo on the in-agent socket for each unread room (keyed by its room id) and each pending invite (below). A new or changed todo wakes the agent with diff --git a/hive-matrix-mcp/src/lib.rs b/hive-matrix-mcp/src/lib.rs index 91500c10a..7fae6d360 100644 --- a/hive-matrix-mcp/src/lib.rs +++ b/hive-matrix-mcp/src/lib.rs @@ -14,5 +14,6 @@ pub mod handlers; pub mod mcp; pub mod paths; pub mod protocol; +pub mod sliding; pub mod timeline; pub mod wake; diff --git a/hive-matrix-mcp/src/main.rs b/hive-matrix-mcp/src/main.rs index 3b80bc4a5..bad0ad846 100644 --- a/hive-matrix-mcp/src/main.rs +++ b/hive-matrix-mcp/src/main.rs @@ -42,7 +42,7 @@ use tokio::time::Instant; use hive_matrix_mcp::accounts::{AccountCfg, Registry}; use hive_matrix_mcp::client::PermanentBringUpError; use hive_matrix_mcp::credential::Token; -use hive_matrix_mcp::{accounts, client, credential, mcp, paths, timeline, wake}; +use hive_matrix_mcp::{accounts, client, credential, mcp, paths, sliding, timeline, wake}; #[derive(Parser)] #[command(name = "hive-matrix-daemon", about = "matrix-sdk client + MCP daemon")] @@ -781,22 +781,35 @@ fn account_sync_loop(client: &Client, tag: Option) -> SyncLoop { // each sweep (see the sweep fns) so re-invites / new messages re-push. let invite_notified = Arc::new(tokio::sync::Mutex::new(std::collections::HashSet::new())); let unread_notified = Arc::new(tokio::sync::Mutex::new(std::collections::HashSet::new())); + // Both loops run this after each response. + let sweep = move || { + let client = cb_client.clone(); + let invite_notified = invite_notified.clone(); + let unread_notified = unread_notified.clone(); + let tag = tag.clone(); + async move { + timeline::sweep_invites(&client, &invite_notified, tag.as_deref()).await; + timeline::sweep_unread(&client, &unread_notified, tag.as_deref()).await; + } + }; Box::pin(async move { + if sliding::wanted(&sync_client).await { + tracing::info!("syncing with sliding sync"); + match sliding::run(&sync_client, &sweep).await { + Ok(()) => return Ok(()), + Err(e) => tracing::warn!(error = %e, "sliding sync gave up, falling back to /sync"), + } + } else { + tracing::info!("syncing with classic /sync"); + } sync_client - .sync_with_callback( - SyncSettings::default().filter(sync_filter()), - move |_response| { - let client = cb_client.clone(); - let invite_notified = invite_notified.clone(); - let unread_notified = unread_notified.clone(); - let tag = tag.clone(); - async move { - timeline::sweep_invites(&client, &invite_notified, tag.as_deref()).await; - timeline::sweep_unread(&client, &unread_notified, tag.as_deref()).await; - matrix_sdk::LoopCtrl::Continue - } - }, - ) + .sync_with_callback(SyncSettings::default().filter(sync_filter()), |_response| { + let sweep = sweep(); + async move { + sweep.await; + matrix_sdk::LoopCtrl::Continue + } + }) .await .context("matrix-sdk sync loop exited")?; Ok(()) diff --git a/hive-matrix-mcp/src/sliding.rs b/hive-matrix-mcp/src/sliding.rs new file mode 100644 index 000000000..5c7e999a1 --- /dev/null +++ b/hive-matrix-mcp/src/sliding.rs @@ -0,0 +1,155 @@ +//! The sliding-sync (MSC4186) sync loop, the cheap alternative to classic +//! `/sync`: one growing list over every joined room, the newest event per +//! room and a minimal set of state, instead of up to 20 events and the state +//! delta of every room that moved. It fills the same matrix-sdk store, so the +//! unread check (the agent's own receipt plus a live `/messages`) is +//! unchanged. +//! +//! [`wanted`] decides whether to use it; `main`'s loop falls back to `/sync` +//! when it is not wanted or [`run`] gives up. + +use std::future::Future; +use std::time::Duration; + +use anyhow::{Result, bail}; +use futures_util::StreamExt as _; +use matrix_sdk::Client; +use matrix_sdk::ruma::events::StateEventType; +use matrix_sdk::sliding_sync::{SlidingSyncList, SlidingSyncMode, Version}; + +/// `classic` forces `/sync`; anything else (or unset) lets the server's +/// advertised support decide. +const MODE_ENV: &str = "HIVE_MATRIX_SYNC"; + +/// Rooms asked for per growing-list step. The list grows until it covers every +/// joined room, so this only sets how many rooms the first responses carry. +const BATCH: u32 = 50; + +/// Consecutive failed polls after which [`run`] gives up and the caller falls +/// back to `/sync`. +const MAX_CONSECUTIVE_ERRORS: u32 = 5; + +/// Pause after a failed poll, so a dead server is not hammered. +const ERROR_PAUSE: Duration = Duration::from_secs(5); + +/// Whether this account should sync with sliding sync: the server advertises +/// MSC4186 and the operator has not forced `/sync`. +pub async fn wanted(client: &Client) -> bool { + let forced_classic = std::env::var(MODE_ENV).is_ok_and(|v| v.eq_ignore_ascii_case("classic")); + let advertised = client + .available_sliding_sync_versions() + .await + .iter() + .any(|v| matches!(v, Version::Native)); + decide(forced_classic, advertised) +} + +fn decide(forced_classic: bool, advertised: bool) -> bool { + advertised && !forced_classic +} + +/// The state the daemon reads from a room: its name, whether it is +/// encrypted, who is in it (lazily) and the power levels sends depend on. +fn required_state() -> Vec<(StateEventType, String)> { + [ + (StateEventType::RoomName, ""), + (StateEventType::RoomCanonicalAlias, ""), + (StateEventType::RoomEncryption, ""), + (StateEventType::RoomCreate, ""), + (StateEventType::RoomPowerLevels, ""), + (StateEventType::RoomMember, "$LAZY"), + (StateEventType::RoomMember, "$ME"), + ] + .into_iter() + .map(|(t, k)| (t, k.to_owned())) + .collect() +} + +/// Poll sliding sync for `client`, calling `after_update` once per response. +/// Returns `Err` once 5 polls in a row have failed, so the caller can fall +/// back to `/sync`. +pub async fn run(client: &Client, after_update: F) -> Result<()> +where + F: Fn() -> Fut, + Fut: Future, +{ + let list = SlidingSyncList::builder("all-rooms") + .sync_mode(SlidingSyncMode::new_growing(BATCH)) + .timeline_limit(1) + .required_state(required_state()); + let sync = client + .sliding_sync("hive-matrix")? + .version(Version::Native) + .with_all_extensions() + .add_list(list) + .build() + .await?; + let stream = sync.sync(); + futures_util::pin_mut!(stream); + let mut errors = ErrorRun::default(); + while let Some(update) = stream.next().await { + match update { + Ok(_) => { + errors.ok(); + after_update().await; + } + Err(e) => { + tracing::warn!(error = %e, "sliding sync poll failed"); + if errors.failed() { + bail!("sliding sync failed {MAX_CONSECUTIVE_ERRORS} polls in a row: {e}"); + } + tokio::time::sleep(ERROR_PAUSE).await; + } + } + } + bail!("sliding sync stream ended") +} + +/// Counts failed polls in a row; a good poll resets it. +#[derive(Default)] +struct ErrorRun(u32); + +impl ErrorRun { + fn ok(&mut self) { + self.0 = 0; + } + + /// Records a failure; true once there are [`MAX_CONSECUTIVE_ERRORS`] in a row. + fn failed(&mut self) -> bool { + self.0 += 1; + self.0 >= MAX_CONSECUTIVE_ERRORS + } +} + +#[cfg(test)] +mod tests { + use super::{ErrorRun, MAX_CONSECUTIVE_ERRORS, decide, required_state}; + use matrix_sdk::ruma::events::StateEventType; + + #[test] + fn sliding_sync_needs_server_support_and_no_opt_out() { + assert!(decide(false, true)); + assert!(!decide(false, false), "a server without MSC4186 gets /sync"); + assert!(!decide(true, true), "the operator can force /sync"); + } + + #[test] + fn gives_up_only_after_a_run_of_failures() { + let mut run = ErrorRun::default(); + for _ in 1..MAX_CONSECUTIVE_ERRORS { + assert!(!run.failed()); + } + run.ok(); + for _ in 1..MAX_CONSECUTIVE_ERRORS { + assert!(!run.failed(), "a good poll resets the count"); + } + assert!(run.failed()); + } + + #[test] + fn the_room_state_asks_for_encryption_and_lazy_members() { + let state = required_state(); + assert!(state.contains(&(StateEventType::RoomEncryption, String::new()))); + assert!(state.contains(&(StateEventType::RoomMember, "$LAZY".to_owned()))); + } +}