knowledge: ping every agent when internal/knowledge changes

This commit is contained in:
damocles 2026-08-02 03:09:55 +02:00
commit 78b90604a1
3 changed files with 77 additions and 10 deletions

View file

@ -129,8 +129,9 @@ pub(super) async fn post_webhook_knowledge(
return (StatusCode::OK, "ignored").into_response();
}
tracing::info!("webhook/knowledge: pull triggered by push to {expected_repo}");
tokio::spawn(async {
if let Err(e) = crate::knowledge::pull().await {
let coord = state.coord.clone();
tokio::spawn(async move {
if let Err(e) = crate::knowledge::pull(&coord).await {
tracing::warn!(error = ?e, "webhook/knowledge: pull failed");
}
});

View file

@ -376,11 +376,12 @@ async fn cmd_serve(
// 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();
let knowledge_coord = coord.clone();
tokio::spawn(async move {
// Initial pull — reconcile any commits that landed while c0re
// was offline. Not fed to the health tracker: a startup miss is
// expected (the clone may not exist yet) and is logged at debug.
if let Err(e) = knowledge::pull().await {
if let Err(e) = knowledge::pull(&knowledge_coord).await {
tracing::debug!(error = ?e, "knowledge: startup pull skipped (no clone yet?)");
}
// Persistent-failure → banner. An hourly sweep that keeps failing for
@ -393,7 +394,7 @@ async fn cmd_serve(
loop {
tokio::select! {
() = tokio::time::sleep(interval) => {
match knowledge::pull().await {
match knowledge::pull(&knowledge_coord).await {
Ok(()) => health.record_ok(),
Err(e) => {
tracing::warn!(error = ?e, "knowledge: periodic pull failed");

View file

@ -16,6 +16,7 @@ use std::collections::BTreeMap;
use anyhow::{Context, Result};
use forgejo_api::structs::{CreateHookOption, CreateHookOptionConfig, CreateHookOptionType};
use crate::coordinator::Coordinator;
use crate::forge::forge_git_url;
pub const ORG: &str = "internal";
@ -241,28 +242,92 @@ pub async fn ensure_webhook(
Ok(())
}
/// Current `HEAD` sha of the local clone, best-effort. `None` on any
/// failure (used only to detect whether a pull actually moved `HEAD` —
/// worth skipping the broadcast over, not worth failing the pull for).
async fn head_sha() -> Option<String> {
let out = tokio::process::Command::new("git")
.args(["-C", LOCAL_DIR, "rev-parse", "HEAD"])
.output()
.await
.ok()?;
out.status
.success()
.then(|| String::from_utf8_lossy(&out.stdout).trim().to_owned())
}
/// Pull the latest changes in the local clone. Called from the webhook
/// 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<()> {
///
/// When the pull actually moves `HEAD` (a real change, not a no-op),
/// broadcasts a short `git diff --stat` summary to every live agent's
/// inbox via `coord` — inbox-only, no forced wake, and shared by both
/// call sites since the broadcast lives in here rather than in each
/// caller.
pub async fn pull(coord: &Coordinator) -> Result<()> {
// Sanity: if the clone is missing (e.g. storage was wiped), refuse
// to pull and let the caller decide whether to re-clone.
let git_dir = std::path::Path::new(LOCAL_DIR).join(".git");
if !git_dir.exists() {
anyhow::bail!("knowledge: {LOCAL_DIR}/.git not found — clone first");
}
let before = head_sha().await;
let out = tokio::process::Command::new("git")
.args(["-C", LOCAL_DIR, "pull", "--ff-only"])
.output()
.await
.context("git pull knowledge")?;
if out.status.success() {
let stdout = String::from_utf8_lossy(&out.stdout).trim().to_owned();
tracing::info!(result = %stdout, "knowledge: pull succeeded");
Ok(())
} else {
if !out.status.success() {
let stderr = String::from_utf8_lossy(&out.stderr).trim().to_owned();
anyhow::bail!("git pull {ORG}/{REPO} failed: {stderr}")
}
let stdout = String::from_utf8_lossy(&out.stdout).trim().to_owned();
tracing::info!(result = %stdout, "knowledge: pull succeeded");
let after = head_sha().await;
if let (Some(before), Some(after)) = (before, after)
&& before != after
{
broadcast_change(coord, &before, &after).await;
}
Ok(())
}
/// Diff `before..after` in the local clone and broadcast a one-line-per-file
/// summary to every live agent. Best-effort — a diff or send failure is
/// logged, never propagated (the pull itself already succeeded).
async fn broadcast_change(coord: &Coordinator, before: &str, after: &str) {
let diff = tokio::process::Command::new("git")
.args([
"-C",
LOCAL_DIR,
"diff",
"--stat",
&format!("{before}..{after}"),
])
.output()
.await;
let stat = match diff {
Ok(out) if out.status.success() => String::from_utf8_lossy(&out.stdout).trim().to_owned(),
Ok(out) => {
let stderr = String::from_utf8_lossy(&out.stderr).trim().to_owned();
tracing::warn!(%stderr, "knowledge: diff --stat failed; broadcasting without it");
String::new()
}
Err(e) => {
tracing::warn!(error = ?e, "knowledge: diff --stat failed; broadcasting without it");
String::new()
}
};
let body = if stat.is_empty() {
"[system] /knowledge updated — see the repo for what changed.".to_owned()
} else {
format!("[system] /knowledge updated:\n{stat}")
};
let errors = coord.broadcast_send(hive_sh4re::SYSTEM_SENDER, &body);
if !errors.is_empty() {
tracing::warn!(?errors, "knowledge: broadcast had per-agent failures");
}
}