From 73f1020a7eed878b34788df46b497c8a21f0732b Mon Sep 17 00:00:00 2001 From: atlas Date: Thu, 9 Jul 2026 00:54:37 +0200 Subject: [PATCH] refactor(#2290): replace mcp_sockets poll with event-driven register_agent MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit mara: the background worker is redundant if c0re knows when its own sockets go missing. damocles: 10s poll latency and redundancy are two faces of the same issue — poll adds a reconnect window and does redundant work when c0re could react directly. design: c0re owns the MCP listener lifecycle, so the only time a listener disappears without c0re knowing is when c0re itself restarts. - replace spawn_poll (recurring 10s loop) with sync_on_start (one-shot sweep at daemon boot): re-registers all running agents on startup after /run/hyperhive/agents/ is cleared by the tmpfs reset. - run_reconcile (reconcile-start path): add coord.register_agent(name) immediately after start_with_fallback — event-driven, no poll delay. - run_create already calls register_agent eagerly; kill/destroy paths already call unregister_agent — no changes needed there. tracker: #2290 --- hive-c0re/src/job_queue/exec.rs | 15 ++++---- hive-c0re/src/main.rs | 14 ++++---- hive-c0re/src/workers/mcp_sockets.rs | 52 ++++++++++++---------------- 3 files changed, 38 insertions(+), 43 deletions(-) diff --git a/hive-c0re/src/job_queue/exec.rs b/hive-c0re/src/job_queue/exec.rs index ff10eb71..ad9a99e2 100644 --- a/hive-c0re/src/job_queue/exec.rs +++ b/hive-c0re/src/job_queue/exec.rs @@ -99,7 +99,7 @@ async fn run_prebuild( let name = &claim.agent; // Prebuild runs while the agent is still up — the runtime dir and // MCP listener already exist. Use the pure path accessor; no need - // to re-register the listener (the mcp_sockets supervisor owns that). + // to re-register the listener (event-driven: registered at start/create). let agent_dir = Coordinator::agent_dir(name); let hive = coord.hive_env(); let paths = Coordinator::agent_paths(name, agent_dir); @@ -190,8 +190,8 @@ async fn run_create(coord: &Arc, claim: &Claim, ctx: &Ctx<'_>) -> R // deploy-window gate so that commit can't land inside another // node's staged deploy window. // Runtime dir creation and MCP listener registration are deferred to - // the tail Reconcile's converge_start_preamble / mcp_sockets supervisor - // so this node stays purely "provision + create", not "create + start". + // the tail Reconcile (converge_start_preamble + register_agent) so this + // node stays purely "provision + create", not "create + start". let _window = crate::meta::exclusive().await; crate::lifecycle::create_container(name, &hive, &paths).await?; Ok(NodeOutput::default()) @@ -255,15 +255,18 @@ async fn run_reconcile( // exists and writes the nspawn/resource-limits drop-ins. // The returned StartableAgent token is the only way to call // start_with_fallback — omitting this becomes a compile error. - // MCP listener registration is handled by mcp_sockets::spawn_poll - // (first tick immediate); the container boot takes longer than - // the 10 s interval so the listener is ready in time. let agent_dir = Coordinator::agent_dir(name); let hive = coord.hive_env(); let paths = Coordinator::agent_paths(name, agent_dir); let token = crate::lifecycle::converge_start_preamble(name, &hive, &paths).await?; ctx.step("nixos-container start"); crate::lifecycle::start_with_fallback(token).await?; + // Bind the MCP listener immediately after starting the container. + // The preamble created the runtime dir; the container is now + // coming up and will connect to this socket on its first turn. + // Event-driven (no background poll) — c0re owns the listener + // lifecycle, so register here rather than waiting for a sweep. + coord.register_agent(name)?; coord.kick_agent(name, "container started"); coord.rescan_containers_and_emit().await; } diff --git a/hive-c0re/src/main.rs b/hive-c0re/src/main.rs index 42ca5687..66741bbe 100644 --- a/hive-c0re/src/main.rs +++ b/hive-c0re/src/main.rs @@ -420,13 +420,13 @@ async fn cmd_serve( // is one stat per agent per tick. // See `docs/gateway.md::Per-agent unix-socket upstream`. agent_sockets::spawn_poll(); - // MCP socket listener reconcile loop: every 10s re-registers any - // running agent that lost its host-side MCP listener (e.g. after a - // hive-c0re restart cleared /run/hyperhive/agents/). First tick fires - // immediately so restarts re-register all running agents without delay. - // Decouples listener registration from the start path — start only needs - // lifecycle::ensure_agent_runtime_dir; the supervisor converges the rest. - mcp_sockets::spawn_poll(coord.clone()); + // MCP socket listener startup sync: one-shot sweep that re-registers any + // running agent container whose MCP listener was lost when hive-c0re + // restarted (Coordinator starts empty; /run/hyperhive/agents/ is tmpfs). + // After this, listener registration is event-driven: run_create / + // run_reconcile call register_agent on start; kill/destroy call + // unregister_agent. No recurring poll needed — c0re owns the listeners. + mcp_sockets::sync_on_start(coord.clone()).await; // Reminder scheduler: drains due reminders + handles // file_path payload persistence. See reminder_scheduler.rs. reminder_scheduler::spawn(coord.clone()); diff --git a/hive-c0re/src/workers/mcp_sockets.rs b/hive-c0re/src/workers/mcp_sockets.rs index fb4efb18..ee0f3de3 100644 --- a/hive-c0re/src/workers/mcp_sockets.rs +++ b/hive-c0re/src/workers/mcp_sockets.rs @@ -1,42 +1,34 @@ -//! MCP socket listener reconcile loop. +//! MCP socket listener boot sync. //! -//! Periodically checks that every running agent container has a bound -//! MCP listener registered in the `Coordinator`. Any agent that is running -//! but whose listener has gone (e.g. after a hive-c0re restart that cleared -//! `/run/hyperhive/agents/`) gets re-registered automatically. +//! On hive-c0re startup, any agent containers that survived the daemon restart +//! still have their bind-mount source dirs but no live MCP listener (the +//! `Coordinator` is freshly empty). `sync_on_start` does a one-shot sweep to +//! re-register all running agents. //! -//! Same shape as `agent_sockets::spawn_poll` — a simple 10 s tick loop that -//! converges "agent container running ⇒ MCP listener bound". The self-healing -//! guarantee means no callsite needs to call `register_agent` directly; -//! `lifecycle::ensure_agent_runtime_dir` (called inside `lifecycle::spawn` -//! and `converge_start_preamble`) creates the bind-mount source, and the -//! reconcile loop picks up the listener binding on the next tick. +//! After startup, listeners are managed event-driven: +//! - `run_create` calls `register_agent` eagerly on first-spawn. +//! - `run_reconcile` calls `register_agent` immediately after `start_with_fallback`. +//! - `kill`/`destroy` paths call `unregister_agent`. +//! +//! No recurring poll is needed because c0re owns the listener lifecycle — +//! a listener can only disappear when c0re itself restarts, which is exactly +//! the case `sync_on_start` covers. use std::sync::Arc; use crate::coordinator::Coordinator; -/// Spawn the MCP socket listener reconcile loop. +/// One-shot MCP listener sync run at daemon startup. /// -/// Every 10 s the loop lists running agent containers and calls -/// `register_agent` for any that lack a bound listener. The first tick fires -/// immediately so hive-c0re restarts re-register all running agents without -/// waiting a full interval. -pub fn spawn_poll(coord: Arc) { - tokio::spawn(async move { - let mut interval = tokio::time::interval(std::time::Duration::from_secs(10)); - loop { - interval.tick().await; - reconcile_once(&coord).await; - } - }); -} - -async fn reconcile_once(coord: &Arc) { +/// Iterates all currently-running agent containers and calls `register_agent` +/// for any that have no live listener in the `Coordinator`. Safe to call +/// concurrently with the rest of startup — `register_agent` is idempotent +/// (drops and rebinds) and the coordinator lock serialises concurrent calls. +pub async fn sync_on_start(coord: Arc) { let running = match crate::lifecycle::list().await { Ok(names) => names, Err(e) => { - tracing::debug!(error = ?e, "mcp_sockets poll: failed to list agents"); + tracing::warn!(error = ?e, "mcp_sockets: startup sync failed to list agents; MCP listeners may be missing until next start"); return; } }; @@ -50,10 +42,10 @@ async fn reconcile_once(coord: &Arc) { tracing::warn!( agent = %name, error = ?e, - "mcp_sockets poll: register_agent failed" + "mcp_sockets: startup register_agent failed" ); } else { - tracing::debug!(agent = %name, "mcp_sockets poll: registered missing listener"); + tracing::debug!(agent = %name, "mcp_sockets: registered listener on startup"); } } }