diff --git a/hive-c0re/src/knowledge.rs b/hive-c0re/src/knowledge.rs index 2c147973..9701ec56 100644 --- a/hive-c0re/src/knowledge.rs +++ b/hive-c0re/src/knowledge.rs @@ -7,7 +7,9 @@ //! to the bind-mounted path inside the container. //! //! hive-c0re maintains the local clone. A Forgejo webhook notifies it -//! on push to main so agents always see an up-to-date snapshot. +//! on push to main so agents always see an up-to-date snapshot. The +//! webhook is auto-created by [`ensure_webhook`] at startup. A +//! periodic pull in `main.rs` provides a fallback cadence. use anyhow::{Context, Result}; @@ -133,10 +135,77 @@ async fn seed_readme(core_token: &str) -> Result<()> { } } +/// Ensure a Forgejo push webhook for `internal/knowledge` exists and +/// points at hive-c0re's `/webhook/knowledge` endpoint. Idempotent — +/// lists existing hooks first and skips creation when one is already +/// targeting the correct URL. `dashboard_port` is the TCP port +/// hive-c0re's dashboard listens on (default 7000); the webhook URL +/// is `http://127.0.0.1:/webhook/knowledge`. +/// +/// Called at startup alongside [`ensure_local_clone`]. No-op when the +/// core token is absent (forge not yet provisioned). +pub async fn ensure_webhook(core_token: &str, dashboard_port: u16) -> Result<()> { + const FORGE_HTTP: &str = "http://localhost:3000"; + let target_url = format!("http://127.0.0.1:{dashboard_port}/webhook/knowledge"); + let client = reqwest::Client::builder() + .timeout(std::time::Duration::from_secs(10)) + .build() + .context("build reqwest client for webhook setup")?; + + // List existing hooks — skip creation if ours is already there. + let list_url = format!("{FORGE_HTTP}/api/v1/repos/{ORG}/{REPO}/hooks"); + let resp = client + .get(&list_url) + .header("Authorization", format!("token {core_token}")) + .send() + .await + .with_context(|| format!("GET {list_url}"))?; + if resp.status().is_success() { + let hooks: Vec = resp.json().await.unwrap_or_default(); + let already_exists = hooks.iter().any(|h| { + h.get("config") + .and_then(|c| c.get("url")) + .and_then(|u| u.as_str()) + == Some(&target_url) + }); + if already_exists { + tracing::debug!(%target_url, "knowledge: push webhook already configured"); + return Ok(()); + } + } + + // Create the webhook. + let create_url = format!("{FORGE_HTTP}/api/v1/repos/{ORG}/{REPO}/hooks"); + let body = serde_json::json!({ + "type": "forgejo", + "config": { + "url": target_url, + "content_type": "json" + }, + "events": ["push"], + "active": true + }); + let resp = client + .post(&create_url) + .header("Authorization", format!("token {core_token}")) + .json(&body) + .send() + .await + .with_context(|| format!("POST {create_url}"))?; + let status = resp.status(); + if status.is_success() { + tracing::info!(%target_url, "knowledge: push webhook created"); + Ok(()) + } else { + let body = resp.text().await.unwrap_or_default(); + anyhow::bail!("create webhook for {ORG}/{REPO} failed ({status}): {body}") + } +} + /// Pull the latest changes in the local clone. Called from the webhook -/// handler on every push to `internal/knowledge` main. Uses -/// `--ff-only` so a force-push to the knowledge repo never wedges the -/// local copy silently. +/// handler on every push to `internal/knowledge` main, and periodically +/// from `main.rs` as a fallback. Uses `--ff-only` so a force-push to +/// the knowledge repo never wedges the local copy silently. pub async fn pull() -> Result<()> { // Sanity: if the clone is missing (e.g. storage was wiped), refuse // to pull and let the caller decide whether to re-clone. diff --git a/hive-c0re/src/main.rs b/hive-c0re/src/main.rs index 273de9ac..751db452 100644 --- a/hive-c0re/src/main.rs +++ b/hive-c0re/src/main.rs @@ -13,8 +13,8 @@ use hive_sh4re::{HostRequest, HostResponse}; use hive_c0re::coordinator::Coordinator; use hive_c0re::{ agent_sockets, auto_update, bash_tasks_vacuum, broker, client, crash_watch, dashboard, - dashboard_events, events_vacuum, forge, manager_server, matrix, migrate, rebuild_queue, - reminder_scheduler, scheduled_prompts_worker, server, stats_vacuum, + dashboard_events, events_vacuum, forge, knowledge, manager_server, matrix, migrate, + rebuild_queue, reminder_scheduler, scheduled_prompts_worker, server, stats_vacuum, }; #[derive(Parser)] @@ -262,13 +262,68 @@ async fn cmd_serve( tokio::spawn(async move { forge::ensure_all().await; }); + // Knowledge webhook setup: ensure the Forgejo push webhook for + // `internal/knowledge` exists so `pull()` fires on merge. Runs + // after forge::ensure_all so the core token + repo are present. + // No-op when the core token or forge are absent. + let webhook_port = dashboard_port; + tokio::spawn(async move { + if let Some(token) = forge::core_token() { + if let Err(e) = knowledge::ensure_webhook(&token, webhook_port).await { + tracing::warn!(error = ?e, "knowledge: ensure_webhook failed"); + } + } + }); + // Knowledge periodic pull: hourly fallback in case the webhook is + // missed (e.g. hive-c0re was down during a push). First fires at + // startup (immediate pull after the clone is already present). + let mut knowledge_shutdown = coord.shutdown_rx(); + tokio::spawn(async move { + // Initial pull — reconcile any commits that landed while c0re + // was offline, including the push that triggered #1244. + if let Err(e) = knowledge::pull().await { + tracing::debug!(error = ?e, "knowledge: startup pull skipped (no clone yet?)"); + } + let interval = std::time::Duration::from_hours(1); + loop { + tokio::select! { + () = tokio::time::sleep(interval) => { + if let Err(e) = knowledge::pull().await { + tracing::warn!(error = ?e, "knowledge: periodic pull failed"); + } + } + _ = knowledge_shutdown.changed() => { + tracing::info!("knowledge pull: shutdown signal received"); + break; + } + } + } + }); // Matrix user sweep: same shape — ensure every container has // an account on the local matrix-tuwunel homeserver with an // access_token persisted to `/matrix-token`. No-op when // the hive-matrix container isn't running. Backgrounded because // UIAA is a two-roundtrip dance per agent. + // + // Runs once at startup AND periodically every 30 minutes so that + // token files deleted by `hive-matrix-daemon` (stale-token + // recovery — `M_UNKNOWN_TOKEN`) get re-provisioned without + // requiring a hive-c0re restart. + let mut matrix_shutdown = coord.shutdown_rx(); tokio::spawn(async move { + let interval = std::time::Duration::from_mins(30); matrix::ensure_all().await; + loop { + tokio::select! { + () = tokio::time::sleep(interval) => { + matrix::ensure_all().await; + } + _ = matrix_shutdown.changed() => { + tracing::info!("matrix ensure_all: shutdown signal received"); + break; + } + } + } }); // Periodic broker vacuum: drop fully-acked messages older // than 30 days. Delivered-but-unacked rows (recoverable via diff --git a/hive-matrix-mcp/src/client.rs b/hive-matrix-mcp/src/client.rs index 97a6fc75..1db66db4 100644 --- a/hive-matrix-mcp/src/client.rs +++ b/hive-matrix-mcp/src/client.rs @@ -60,7 +60,41 @@ pub async fn build_and_restore( return Err(anyhow!("matrix token at {} is empty", token_file.display())); } - let (user_id, device_id) = whoami(homeserver, &token).await?; + let (user_id, device_id) = match whoami(homeserver, &token).await { + Ok(ids) => ids, + Err(e) => { + let msg = format!("{e:#}"); + if msg.contains("M_UNKNOWN_TOKEN") { + // Homeserver rejected our token — stale session after a homeserver + // state wipe or token expiry. Delete the token file (and the + // matrix-sdk sqlite state keyed to the now-invalid session) so + // hive-c0re's periodic `ensure_all` sweep re-provisions the account. + // Exit 0: systemd's Restart=on-failure must not loop us here; the + // systemd.paths watcher restarts us once the new token file appears. + tracing::warn!( + path = %token_file.display(), + "matrix token rejected (M_UNKNOWN_TOKEN); deleting stale token + \ + sdk state for re-provisioning" + ); + let _ = fs::remove_file(token_file).await; + if let Err(re) = fs::remove_dir_all(state_dir).await { + tracing::warn!( + path = %state_dir.display(), + err = %re, + "failed to remove sdk state dir; next startup may fail with stale state" + ); + } + // Exit 0 rather than returning Err: `hive-matrix-daemon` is a + // single-purpose process binary; the call site is before any + // tasks are spawned so there are no resources to clean up. + // Using exit(0) (not Err) keeps systemd's Restart=on-failure + // from looping — the systemd.paths watcher re-launches us + // once hive-c0re writes a fresh token file. + std::process::exit(0); + } + return Err(e); + } + }; fs::create_dir_all(state_dir) .await diff --git a/hive-matrix-mcp/src/main.rs b/hive-matrix-mcp/src/main.rs index d8bcda2e..24b551e8 100644 --- a/hive-matrix-mcp/src/main.rs +++ b/hive-matrix-mcp/src/main.rs @@ -13,6 +13,9 @@ //! Standalone-degraded boot: missing token file → exit 0 cleanly so //! systemd's `ConditionPathExists=` doesn't have to be perfectly //! synced with hive-c0re's token-provisioning timing. +//! +//! Stale-token recovery: handled in `client::build_and_restore` — see +//! that module for the M_UNKNOWN_TOKEN detection + cleanup flow. use anyhow::{Context, Result}; use matrix_sdk::config::SyncSettings;