diff --git a/hive-c0re/src/actions.rs b/hive-c0re/src/actions.rs index 55a65eae..d89defea 100644 --- a/hive-c0re/src/actions.rs +++ b/hive-c0re/src/actions.rs @@ -159,9 +159,7 @@ pub async fn run_approval_apply_commit( approval_id: i64, ) -> Result<()> { let approval = fetch_approval_for_worker(coord, approval_id, ApprovalKind::ApplyCommit)?; - // Runtime dir creation is handled inside lifecycle::rebuild_no_meta's - // spawn path (first-spawn) or is already present for rebuilds. - let agent_dir = Coordinator::agent_dir(&approval.agent); + let agent_dir = coord.ensure_runtime(&approval.agent)?; let applied_dir = Coordinator::agent_applied_dir(&approval.agent); coord.set_queue_step(queue_entry_id, "apply commit"); let (result, terminal_tag, is_first_spawn) = @@ -194,7 +192,7 @@ pub async fn run_approval_merge_config_pr( approval_id: i64, ) -> Result<()> { let approval = fetch_approval_for_worker(coord, approval_id, ApprovalKind::MergeConfigPr)?; - let agent_dir = Coordinator::agent_dir(&approval.agent); + let agent_dir = coord.ensure_runtime(&approval.agent)?; let applied_dir = Coordinator::agent_applied_dir(&approval.agent); coord.set_queue_step(queue_entry_id, "merge config pr"); let (result, terminal_tag) = diff --git a/hive-c0re/src/coordinator.rs b/hive-c0re/src/coordinator.rs index 16e7d241..97d984e9 100644 --- a/hive-c0re/src/coordinator.rs +++ b/hive-c0re/src/coordinator.rs @@ -545,11 +545,14 @@ impl Coordinator { } } - /// Assemble the per-agent filesystem paths for `name`. `agent_dir` - /// is the runtime directory (`/run/hyperhive/agents/`), obtained - /// from `Coordinator::agent_dir(name)` (pure path) or from - /// `lifecycle::ensure_agent_runtime_dir` + `agent_dir` when the dir - /// must be created. All other paths are derived statically from `name`. + /// Assemble the per-agent filesystem paths for `name`. The caller + /// must supply `agent_dir` (from `ensure_runtime`) since that + /// creates the tmpfs entry on first call. All other paths are + /// derived statically from `name`. + /// + /// ```ignore + /// let paths = Coordinator::agent_paths(name, coord.ensure_runtime(name)?); + /// ``` #[must_use] pub fn agent_paths(name: &str, agent_dir: PathBuf) -> AgentPaths { AgentPaths { @@ -1450,16 +1453,11 @@ impl Coordinator { Self::agent_dir(name).join("mcp.sock") } - /// Ensure a runtime dir + (for sub-agents) per-agent socket exists. - /// - /// **Prefer the split form (event-driven):** - /// - dir creation → `lifecycle::ensure_agent_runtime_dir(name)` - /// - listener registration → `register_agent(name)` at each explicit - /// lifecycle event (spawn, reconcile-start). `mcp_sockets::sync_on_start` - /// covers the daemon-restart case. - /// - /// This method is kept as a convenience shim for any remaining callers - /// that need the combined semantics in one call. + /// Ensure a runtime dir + (for sub-agents) per-agent socket exists. For + /// the manager, `socket_server::start_manager` owns the socket — just return + /// the dir. For sub-agents this is `register_agent` (creates a fresh + /// listener bound to `socket_path(name)`). Source directory of the + /// `/run/hive/mcp.sock` bind that ends up in `set_nspawn_flags`. pub fn ensure_runtime(self: &Arc, name: &str) -> Result { if name == crate::lifecycle::MANAGER_NAME { let dir = Self::agent_dir(name); diff --git a/hive-c0re/src/job_queue/exec.rs b/hive-c0re/src/job_queue/exec.rs index 51b35358..9a229459 100644 --- a/hive-c0re/src/job_queue/exec.rs +++ b/hive-c0re/src/job_queue/exec.rs @@ -97,10 +97,9 @@ async fn run_prebuild( relock: bool, ) -> Result { 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 (event-driven: registered at start/create). - let agent_dir = Coordinator::agent_dir(name); + let agent_dir = coord + .ensure_runtime(name) + .with_context(|| format!("ensure_runtime {name}"))?; let hive = coord.hive_env(); let paths = Coordinator::agent_paths(name, agent_dir); crate::lifecycle::prepare_rebuild_dirs(name, &paths).await?; @@ -132,9 +131,7 @@ async fn run_prebuild( /// `Reconcile` runs after this node terminal ok *or* fail. async fn run_swap(coord: &Arc, claim: &Claim, ctx: &Ctx<'_>) -> Result { let name = &claim.agent; - // Swap runs on an already-existing (stopped) container — runtime dir - // and listener were created earlier. Pure path accessor suffices. - let agent_dir = Coordinator::agent_dir(name); + let agent_dir = coord.ensure_runtime(name)?; let hive = coord.hive_env(); let paths = Coordinator::agent_paths(name, agent_dir); let result = @@ -181,7 +178,7 @@ async fn run_swap(coord: &Arc, claim: &Claim, ctx: &Ctx<'_>) -> Res /// build+create — no prebuild needed). async fn run_create(coord: &Arc, claim: &Claim, ctx: &Ctx<'_>) -> Result { let name = &claim.agent; - let agent_dir = Coordinator::agent_dir(name); + let agent_dir = coord.ensure_runtime(name)?; let hive = coord.hive_env(); let paths = Coordinator::agent_paths(name, agent_dir); ctx.step("nixos-container create"); @@ -189,9 +186,6 @@ async fn run_create(coord: &Arc, claim: &Claim, ctx: &Ctx<'_>) -> R // (sync_agents commit) before `nixos-container create` — hold the // 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 (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()) @@ -251,22 +245,21 @@ async fn run_reconcile( .transient .is_none() .then(|| coord.transient_guard(name, crate::coordinator::TransientKind::Starting)); - // Run the typed start preamble: ensures the runtime dir - // 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. - let agent_dir = Coordinator::agent_dir(name); + // Converge the ephemeral host-side state before the start: + // `/run/hyperhive/agents/` + `/run/hive-agent/` + // (both nspawn bind sources — tmpfs, empty after a host + // reboot; nspawn refuses to start with a missing source) + // and the resource-limits drop-in under + // `/run/systemd/system/`. Rebuild DAGs get this from their + // Prebuild/Swap nodes; the bare-Reconcile templates (boot + // reconcile, plain start/restart) otherwise start with + // nothing under /run and fail. + let agent_dir = coord.ensure_runtime(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?; + crate::lifecycle::write_dropins(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)?; + crate::lifecycle::start_with_fallback(name).await?; coord.kick_agent(name, "container started"); coord.rescan_containers_and_emit().await; } @@ -343,10 +336,7 @@ async fn run_drain(coord: &Arc, claim: &Claim, ctx: &Ctx<'_>) -> Re /// `set_nspawn_flags` + `set_resource_limits` + daemon-reload. async fn run_write_dropin(coord: &Arc, claim: &Claim) -> Result { let name = &claim.agent; - // write_dropins only needs the path value to build AgentPaths; the - // dir doesn't need to exist at this point (created by ensure_agent_runtime_dir - // on the upstream Prebuild/Start node). - let agent_dir = Coordinator::agent_dir(name); + let agent_dir = coord.ensure_runtime(name)?; let hive = coord.hive_env(); let paths = Coordinator::agent_paths(name, agent_dir); crate::lifecycle::write_dropins(name, &hive, &paths).await?; diff --git a/hive-c0re/src/lib.rs b/hive-c0re/src/lib.rs index c2156ed4..3b20bd5e 100644 --- a/hive-c0re/src/lib.rs +++ b/hive-c0re/src/lib.rs @@ -50,6 +50,6 @@ pub use stores::{ approvals, audit_log, broker, build_logs, db, operator_questions, power, scheduled_prompts, }; pub use workers::{ - agent_sockets, auto_update, crash_watch, knowledge, mcp_sockets, reminder_scheduler, + agent_sockets, auto_update, crash_watch, knowledge, reminder_scheduler, scheduled_prompts_worker, }; diff --git a/hive-c0re/src/lifecycle/mod.rs b/hive-c0re/src/lifecycle/mod.rs index 207976b0..cf91a945 100644 --- a/hive-c0re/src/lifecycle/mod.rs +++ b/hive-c0re/src/lifecycle/mod.rs @@ -275,9 +275,6 @@ async fn port_collision(self_name: &str) -> Option { pub async fn spawn(name: &str, hive: &HiveEnv, paths: &AgentPaths) -> Result<()> { create_container(name, hive, paths).await?; - // Runtime dir must exist before nixos-container start (nspawn bind-mount - // source). Create it here so callers don't need a separate preamble step. - ensure_agent_runtime_dir(name)?; write_dropins(name, hive, paths).await?; priv_run("start", name).await } @@ -423,59 +420,17 @@ pub async fn start(name: &str) -> Result<()> { priv_run("start", name).await } -/// Opaque token produced by [`converge_start_preamble`]. -/// [`start_with_fallback`] requires this as proof that the pre-start -/// preamble (runtime dir + drop-ins) ran. Dropping the token without -/// calling `start_with_fallback` is a no-op. -#[must_use = "call lifecycle::start_with_fallback(token) to start the container"] -pub struct StartableAgent { - name: String, -} - -/// Run the per-agent start preamble: ensure the runtime dir exists and write -/// the nspawn / resource-limits drop-ins. Returns a [`StartableAgent`] token -/// as typed proof that the preamble ran; pass it to [`start_with_fallback`]. -/// Callers that omit this step cannot call `start_with_fallback` — the type -/// system makes forgetting the preamble a compile error. -/// -/// # Errors -/// -/// Returns an error if `ensure_agent_runtime_dir` or `write_dropins` fails. -pub async fn converge_start_preamble( - name: &str, - hive: &HiveEnv, - paths: &AgentPaths, -) -> Result { - ensure_agent_runtime_dir(name)?; - write_dropins(name, hive, paths).await?; - Ok(StartableAgent { - name: name.to_owned(), - }) -} - /// Start with the cold-start fallback: when a plain start fails (the /// activation-error shape), retry once via stop + kill + start before /// giving up. Used by the queue's fast-lane `Start` handler and the /// inline start-after-rebuild path. /// See `docs/coordinator.md::Cold-start fallback`. /// -/// Requires a [`StartableAgent`] token from [`converge_start_preamble`] -/// to prove the preamble ran. For internal use within this module (where -/// the preamble is already enforced structurally) call -/// `start_with_fallback_inner` directly. -/// /// # Errors /// /// Propagates the retry's start error (annotated with the original /// failure) when the fallback also fails. -pub async fn start_with_fallback(token: StartableAgent) -> Result<()> { - start_with_fallback_inner(&token.name).await -} - -/// Internal implementation of the cold-start fallback. Used by -/// [`start_with_fallback`] (public, token-gated) and by -/// [`rebuild_no_meta`] where the preamble is already enforced structurally. -async fn start_with_fallback_inner(name: &str) -> Result<()> { +pub async fn start_with_fallback(name: &str) -> Result<()> { validate(name)?; if let Err(start_err) = priv_run("start", name).await { let container = container_name(name); @@ -622,9 +577,7 @@ pub async fn rebuild_no_meta( return Ok(true); } on_step("nixos-container start"); - // write_dropins was called above; use the inner fn directly - // since the preamble is enforced structurally in this path. - start_with_fallback_inner(name).await?; + start_with_fallback(name).await?; } Ok(false) } else { @@ -632,8 +585,6 @@ pub async fn rebuild_no_meta( // See `docs/coordinator.md::Spawn path`. on_step("nixos-container create"); priv_run("create", name).await?; - // Runtime dir must exist before nixos-container start. - ensure_agent_runtime_dir(name)?; write_dropins(name, hive, paths).await?; on_step("nixos-container start"); priv_run("start", name).await?; @@ -798,23 +749,6 @@ pub async fn sync_tmpfiles() { } } -/// Ensure the per-agent runtime directory `/run/hyperhive/agents/` -/// exists. The directory is also written by `SyncAgentTmpfiles` (run at -/// boot + spawn/destroy), but explicit creation in start/spawn paths guards -/// against races where hive-c0re starts a container before tmpfiles.d has -/// applied the new entry. -/// -/// Pure filesystem op — no `Coordinator` dependency — so callers that only -/// need the dir do not have to hold an `Arc`. -/// -/// # Errors -/// Returns an error if `create_dir_all` fails. -pub fn ensure_agent_runtime_dir(name: &str) -> Result<()> { - let dir = std::path::PathBuf::from(format!("/run/hyperhive/agents/{name}")); - std::fs::create_dir_all(&dir) - .with_context(|| format!("create agent runtime dir {}", dir.display())) -} - /// Build the per-line callback for `create_container_streaming` / /// `update_container_streaming`. Both ops share identical dispatch logic /// (stdout → info + `append_stdout`, stderr → warn + `append_stderr`); this diff --git a/hive-c0re/src/main.rs b/hive-c0re/src/main.rs index 66741bbe..b9752362 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, HiveEnv, ServeConfig}; use hive_c0re::{ agent_sockets, auto_update, broker, client, crash_watch, dashboard, dashboard_events, forge, - job_queue, knowledge, matrix, mcp_sockets, migrate, reminder_scheduler, - scheduled_prompts_worker, server, socket_server, + job_queue, knowledge, matrix, migrate, reminder_scheduler, scheduled_prompts_worker, server, + socket_server, }; #[derive(Parser)] @@ -420,13 +420,6 @@ 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 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/server.rs b/hive-c0re/src/server.rs index bf0b3779..f0913c85 100644 --- a/hive-c0re/src/server.rs +++ b/hive-c0re/src/server.rs @@ -202,21 +202,14 @@ async fn dispatch(req: &HostRequest, coord: Arc) -> HostResponse { /// registration and notifying the manager on failure. async fn handle_spawn(coord: &Arc, name: &str) -> Result { tracing::info!(%name, "spawn"); - let agent_dir = Coordinator::agent_dir(name); + let agent_dir = coord.ensure_runtime(name)?; let hive = coord.hive_env(); let paths = Coordinator::agent_paths(name, agent_dir); - // lifecycle::spawn creates the runtime dir internally before start. - // MCP listener registration is event-driven: bind immediately on - // success so the harness can connect on its first turn without - // waiting for any poll interval. match lifecycle::spawn(name, &hive, &paths).await { Ok(()) => { if let Err(e) = coord.power.set(name, crate::power::Wanted::Up) { tracing::warn!(%name, error = ?e, "agent_power: set wanted=up failed"); } - // Bind the MCP listener now that the container is starting up. - // The harness connects to this socket on its first turn. - coord.register_agent(name)?; coord.notify_manager(&hive_sh4re::HelperEvent::Spawned { agent: name.to_owned(), ok: true, @@ -227,8 +220,8 @@ async fn handle_spawn(coord: &Arc, name: &str) -> Result { - // Spawn failed: register_agent was never called, so there is - // nothing to unregister. Notify the manager and propagate. + // Roll back socket registration if container creation failed. + coord.unregister_agent(name); coord.notify_manager(&hive_sh4re::HelperEvent::Spawned { agent: name.to_owned(), ok: false, diff --git a/hive-c0re/src/workers/auto_update.rs b/hive-c0re/src/workers/auto_update.rs index 1c7a909c..590ff450 100644 --- a/hive-c0re/src/workers/auto_update.rs +++ b/hive-c0re/src/workers/auto_update.rs @@ -153,9 +153,7 @@ pub async fn ensure_root_agent(coord: &Arc) -> Result<()> { return Ok(()); } tracing::info!("manager container missing — spawning"); - // lifecycle::spawn creates the runtime dir internally; no manual - // ensure_agent_runtime_dir needed here. - let runtime = Coordinator::agent_dir(MANAGER_NAME); + let runtime = coord.ensure_runtime(MANAGER_NAME)?; let hive = coord.hive_env(); let paths = Coordinator::agent_paths(MANAGER_NAME, runtime); lifecycle::spawn(MANAGER_NAME, &hive, &paths).await?; diff --git a/hive-c0re/src/workers/mcp_sockets.rs b/hive-c0re/src/workers/mcp_sockets.rs deleted file mode 100644 index ee0f3de3..00000000 --- a/hive-c0re/src/workers/mcp_sockets.rs +++ /dev/null @@ -1,52 +0,0 @@ -//! MCP socket listener boot sync. -//! -//! 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. -//! -//! 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; - -/// One-shot MCP listener sync run at daemon startup. -/// -/// 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::warn!(error = ?e, "mcp_sockets: startup sync failed to list agents; MCP listeners may be missing until next start"); - return; - } - }; - let registered = coord.list_agents(); - for container in running { - let Some(name) = container.strip_prefix(crate::lifecycle::AGENT_PREFIX) else { - continue; - }; - if !registered.contains(&name.to_owned()) { - if let Err(e) = coord.register_agent(name) { - tracing::warn!( - agent = %name, - error = ?e, - "mcp_sockets: startup register_agent failed" - ); - } else { - tracing::debug!(agent = %name, "mcp_sockets: registered listener on startup"); - } - } - } -} diff --git a/hive-c0re/src/workers/mod.rs b/hive-c0re/src/workers/mod.rs index 75619f0b..98654ea3 100644 --- a/hive-c0re/src/workers/mod.rs +++ b/hive-c0re/src/workers/mod.rs @@ -1,13 +1,12 @@ //! Background tasks and periodic sweeps: crash/login watcher, the //! reminder and scheduled-prompt delivery loops, boot-time auto-update -//! reconcile, the agent-sockets.json writer loop, the MCP socket listener -//! reconcile loop, and knowledge-repo sync. Each submodule is re-exported -//! at the crate root, so `crate::crash_watch::…` etc. keep working unchanged. +//! reconcile, the agent-sockets.json writer loop, and knowledge-repo +//! sync. Each submodule is re-exported at the crate root, so +//! `crate::crash_watch::…` etc. keep working unchanged. pub mod agent_sockets; pub mod auto_update; pub mod crash_watch; pub mod knowledge; -pub mod mcp_sockets; pub mod reminder_scheduler; pub mod scheduled_prompts_worker;