test the remaining job-queue executors through the host seam

hive-c0re/src/job_queue/exec.rs

@@ -5,6 +5,7 @@
 //! the `Swap` tail); DAG-level failure handling is cancel-downstream in
 //! the queue.
 
+use std::path::PathBuf;
 use std::sync::{Arc, Mutex, OnceLock};
 
 use anyhow::{Context as _, Result};
@@ -13,6 +14,7 @@ use hive_jobq::NodeId;
 
 use super::model::NodeKind;
 use crate::coordinator::{AgentPaths, Coordinator, HiveEnv};
+use crate::meta_rev::MetaBuild;
 use crate::power::{ReconcileAction, reconcile_action};
 use crate::stats::sweep_health::{self, SweepHealth};
 
@@ -55,6 +57,48 @@ pub(super) trait Host: Sync {
     ) -> Result<()>;
     /// Write or remove the agent's pause marker.
     async fn set_paused(&self, agent: &hive_types::Ident, paused: bool) -> Result<()>;
+    /// Publish a lifecycle notice for the hive under the `core` subsystem.
+    async fn notify(&self, key: String, summary: String);
+    /// Re-read the agent's archetype from the swarm.
+    async fn refresh_archetype(&self, name: &str) -> Result<()>;
+    /// Rebuild preamble: port check, applied repo and state dirs.
+    async fn prepare_rebuild_dirs(&self, name: &str, paths: &AgentPaths) -> Result<()>;
+    /// Rewrite the host-side files that follow the agent set.
+    async fn refresh_agent_set(&self) -> Result<()>;
+    /// Pin the `internal/meta` commit the agent builds from.
+    async fn resolve_meta(&self, name: &str) -> Result<MetaBuild>;
+    /// Record `build` as what `Prebuild` and `Swap` build from.
+    fn record_meta(&self, name: &str, build: &MetaBuild) -> Result<()>;
+    /// The build `MetaSync` recorded, if any.
+    fn recorded_build(&self, name: &str) -> Option<MetaBuild>;
+    /// Mark the agent as running the pins of `build`.
+    fn mark_deployed(&self, name: &str, build: &MetaBuild) -> Result<()>;
+    /// First-spawn provisioning: repos, state subvolume, agent-set refresh.
+    async fn provision(&self, name: &str, paths: &AgentPaths) -> Result<()>;
+    /// `nixos-container create` at `rev`.
+    async fn create_only(&self, name: &str, rev: &str) -> Result<()>;
+    /// Write the nspawn and resource-limit drop-ins, then reload systemd.
+    async fn write_dropins(&self, name: &str, hive: &HiveEnv, paths: &AgentPaths) -> Result<()>;
+    /// Start preamble, then start with the cold-start fallback.
+    async fn start(&self, name: &str, hive: &HiveEnv, paths: &AgentPaths) -> Result<()>;
+    /// Bind the agent's MCP listener.
+    fn register_agent(&self, coord: &Arc<Coordinator>, name: &str) -> Result<()>;
+    /// `nixos-container destroy`.
+    async fn destroy(&self, name: &str) -> Result<()>;
+    /// The agent's ephemeral runtime dir.
+    fn runtime_dir(&self, name: &str) -> PathBuf;
+    /// Delete the agent's state subvolume through `hive-priv`.
+    async fn delete_state_subvolume(&self, name: &str) -> Result<()>;
+    /// The persistent directories a purge removes.
+    fn purge_dirs(&self, name: &str) -> Vec<PathBuf>;
+    /// Publish the hive's forge credential for nix.
+    async fn forge_ensure_all(&self);
+    /// One matrix user/space pass; whether every agent synced.
+    async fn matrix_ensure_all(&self) -> bool;
+    /// One `/knowledge` pull.
+    async fn knowledge_pull(&self, coord: &Arc<Coordinator>) -> Result<()>;
+    /// One swarm wanted-state pull.
+    async fn wanted_pull(&self, coord: &Arc<Coordinator>) -> Result<()>;
 }
 
 /// [`Host`] over the real system.
@@ -92,6 +136,100 @@ impl Host for RealHost {
     async fn set_paused(&self, agent: &hive_types::Ident, paused: bool) -> Result<()> {
         Coordinator::set_paused(agent, paused).await
     }
+
+    async fn notify(&self, key: String, summary: String) {
+        crate::swarm_notices::notify("core", Some(key), summary, None).await;
+    }
+
+    async fn refresh_archetype(&self, name: &str) -> Result<()> {
+        crate::archetypes::refresh(name).await
+    }
+
+    async fn prepare_rebuild_dirs(&self, name: &str, paths: &AgentPaths) -> Result<()> {
+        crate::lifecycle::prepare_rebuild_dirs(name, paths).await
+    }
+
+    async fn refresh_agent_set(&self) -> Result<()> {
+        let agents = crate::lifecycle::agents_for_meta_listing().await?;
+        crate::agent_set::refresh(&agents).await
+    }
+
+    async fn resolve_meta(&self, name: &str) -> Result<MetaBuild> {
+        crate::meta_rev::resolve(name).await
+    }
+
+    fn record_meta(&self, name: &str, build: &MetaBuild) -> Result<()> {
+        crate::meta_rev::record(name, build)
+    }
+
+    fn recorded_build(&self, name: &str) -> Option<MetaBuild> {
+        crate::meta_rev::recorded(name)
+    }
+
+    fn mark_deployed(&self, name: &str, build: &MetaBuild) -> Result<()> {
+        crate::meta_rev::mark_deployed(name, build)
+    }
+
+    async fn provision(&self, name: &str, paths: &AgentPaths) -> Result<()> {
+        crate::lifecycle::provision_container(name, paths).await
+    }
+
+    async fn create_only(&self, name: &str, rev: &str) -> Result<()> {
+        crate::lifecycle::create_only(name, rev).await
+    }
+
+    async fn write_dropins(&self, name: &str, hive: &HiveEnv, paths: &AgentPaths) -> Result<()> {
+        crate::lifecycle::write_dropins(name, hive, paths).await
+    }
+
+    async fn start(&self, name: &str, hive: &HiveEnv, paths: &AgentPaths) -> Result<()> {
+        let token = crate::lifecycle::converge_start_preamble(name, hive, paths).await?;
+        crate::lifecycle::start_with_fallback(token).await
+    }
+
+    fn register_agent(&self, coord: &Arc<Coordinator>, name: &str) -> Result<()> {
+        coord.register_agent(name).map(drop)
+    }
+
+    async fn destroy(&self, name: &str) -> Result<()> {
+        crate::lifecycle::destroy(name).await
+    }
+
+    fn runtime_dir(&self, name: &str) -> PathBuf {
+        crate::paths::agent_runtime_dir(name)
+    }
+
+    async fn delete_state_subvolume(&self, name: &str) -> Result<()> {
+        crate::priv_client::delete_agent_subvolume(name).await
+    }
+
+    fn purge_dirs(&self, name: &str) -> Vec<PathBuf> {
+        // A malformed name can't have a persistent state tree (the state dir is
+        // only ever created under a validated Ident), so only the applied dir
+        // is left to clear.
+        hive_types::Ident::parse(name)
+            .ok()
+            .map(|id| crate::paths::agent_state_dir(&id))
+            .into_iter()
+            .chain([crate::paths::applied_dir(name)])
+            .collect()
+    }
+
+    async fn forge_ensure_all(&self) {
+        crate::forge::ensure_all().await;
+    }
+
+    async fn matrix_ensure_all(&self) -> bool {
+        crate::matrix::ensure_all().await
+    }
+
+    async fn knowledge_pull(&self, coord: &Arc<Coordinator>) -> Result<()> {
+        crate::workers::knowledge::pull(coord).await
+    }
+
+    async fn wanted_pull(&self, coord: &Arc<Coordinator>) -> Result<()> {
+        crate::workers::wanted::pull(coord).await
+    }
 }
 
 /// Run one claimed node to completion. Called from a task the
@@ -130,19 +268,21 @@ pub(super) async fn run_node(
     // inside an `.await` would make this future non-`Send` (see above), so the
     // growth executors return what to grow rather than taking the builder.
     let result = match kind {
-        NodeKind::MetaSync { .. } => run_meta_sync(agent).await,
+        NodeKind::MetaSync { .. } => run_meta_sync(&RealHost, agent).await,
         NodeKind::Prebuild { .. } => run_prebuild(&RealHost, agent, id).await,
         NodeKind::Swap { .. } => run_swap(&RealHost, coord, agent, id).await,
-        NodeKind::RebuildBookkeeping { .. } => run_rebuild_bookkeeping(coord, agent).await,
-        NodeKind::Provision { .. } => run_provision(agent).await,
-        NodeKind::Create { .. } => run_create(agent).await,
-        NodeKind::DestroyContainer { .. } => run_destroy_container(coord, agent).await,
+        NodeKind::RebuildBookkeeping { .. } => {
+            run_rebuild_bookkeeping(&RealHost, coord, agent).await
+        }
+        NodeKind::Provision { .. } => run_provision(&RealHost, agent).await,
+        NodeKind::Create { .. } => run_create(&RealHost, agent).await,
+        NodeKind::DestroyContainer { .. } => run_destroy_container(&RealHost, coord, agent).await,
         NodeKind::PurgeState { .. } => {
-            run_purge_state(agent).await;
+            run_purge_state(&RealHost, agent).await;
             Ok(())
         }
         NodeKind::DestroyBookkeeping { .. } => {
-            run_destroy_bookkeeping(coord, agent).await;
+            run_destroy_bookkeeping(&RealHost, coord, agent).await;
             Ok(())
         }
 
@@ -151,7 +291,7 @@ pub(super) async fn run_node(
                 super::templates::fanned_out_mechanical(&builder, kind);
             }
         }),
-        NodeKind::Start { .. } => run_start(coord, agent).await,
+        NodeKind::Start { .. } => run_start(&RealHost, coord, agent).await,
         NodeKind::Stop { .. } => run_stop(&RealHost, coord, agent).await,
         NodeKind::StopForUpdate { .. } => run_stop_for_update(&RealHost, coord, agent).await,
         NodeKind::Signal { .. } => {
@@ -161,12 +301,12 @@ pub(super) async fn run_node(
         NodeKind::Drain { .. } => run_drain(coord, agent, GRACEFUL_STOP_TIMEOUT).await,
         NodeKind::PauseSignal { .. } => run_pause_signal(&RealHost, coord, agent).await,
         NodeKind::PauseDrain { .. } => run_pause_drain(coord, agent, PAUSE_ACK_TIMEOUT).await,
-        NodeKind::WriteDropin { .. } => run_write_dropin(coord, agent).await,
+        NodeKind::WriteDropin { .. } => run_write_dropin(&RealHost, coord, agent).await,
         // The payload rides the node and is destructured here, so the executor
         // takes it directly instead of re-matching the kind behind a `bail!`
         // that could never fire.
         NodeKind::EmitRebuilt { ok, .. } => {
-            run_emit_rebuilt(coord, agent, coord.job_queue.root_of(id), *ok).await;
+            run_emit_rebuilt(&RealHost, coord, agent, coord.job_queue.root_of(id), *ok).await;
             Ok(())
         }
         NodeKind::SetWanted { up, .. } => run_set_wanted(coord, agent, *up),
@@ -174,10 +314,12 @@ pub(super) async fn run_node(
         // `Finishing` so the nodes under it start. What they declare stays held
         // until their whole subtree settles.
         NodeKind::AgentWindow { .. } => Ok(()),
-        NodeKind::ForgeSweep => run_forge_sweep().await,
-        NodeKind::MatrixSweep => run_matrix_sweep().await,
-        NodeKind::KnowledgePull => run_knowledge_pull(coord).await,
-        NodeKind::WantedPull => run_wanted_pull(coord).await,
+        NodeKind::ForgeSweep => run_forge_sweep(&RealHost).await,
+        NodeKind::MatrixSweep => run_matrix_sweep(&RealHost, MATRIX_SWEEP_DEADLINE).await,
+        NodeKind::KnowledgePull => {
+            run_knowledge_pull(&RealHost, coord, KNOWLEDGE_PULL_DEADLINE).await
+        }
+        NodeKind::WantedPull => run_wanted_pull(&RealHost, coord).await,
     };
     (builder, result)
 }
@@ -187,19 +329,19 @@ pub(super) async fn run_node(
 /// is best-effort (a build re-publishes the credential before it fetches), so
 /// this wrapper has nothing left to report; it exists to make the step a
 /// visible unit of work.
-async fn run_forge_sweep() -> Result<()> {
-    crate::forge::ensure_all().await;
+async fn run_forge_sweep(host: &impl Host) -> Result<()> {
+    host.forge_ensure_all().await;
     Ok(())
 }
 
 /// Matrix user/space sweep as a DAG node — see [`NodeKind::MatrixSweep`].
 /// Every pass reports here, whoever submitted it: as the node's own error on
 /// the dashboard, and into the debounced warning.
-async fn run_matrix_sweep() -> Result<()> {
-    let ok = tokio::time::timeout(MATRIX_SWEEP_DEADLINE, crate::matrix::ensure_all())
+async fn run_matrix_sweep(host: &impl Host, deadline: std::time::Duration) -> Result<()> {
+    let ok = tokio::time::timeout(deadline, host.matrix_ensure_all())
         .await
         .unwrap_or_else(|_| {
-            tracing::warn!(deadline = ?MATRIX_SWEEP_DEADLINE, "matrix: sweep timed out");
+            tracing::warn!(?deadline, "matrix: sweep timed out");
             false
         });
     let mut health = matrix_sweep_health()
@@ -246,17 +388,18 @@ fn matrix_sweep_warning(ctx: sweep_health::SweepFailure) -> String {
 /// `/knowledge` pull as a DAG node — see [`NodeKind::KnowledgePull`]. Every
 /// pass reports here, whoever submitted it: as the node's own error on the
 /// dashboard, and into the debounced warning.
-async fn run_knowledge_pull(coord: &Arc<Coordinator>) -> Result<()> {
-    let result = tokio::time::timeout(
-        KNOWLEDGE_PULL_DEADLINE,
-        crate::workers::knowledge::pull(coord),
-    )
-    .await
-    .unwrap_or_else(|_| {
-        Err(anyhow::anyhow!(
-            "knowledge pull timed out after {KNOWLEDGE_PULL_DEADLINE:?}"
-        ))
-    });
+async fn run_knowledge_pull(
+    host: &impl Host,
+    coord: &Arc<Coordinator>,
+    deadline: std::time::Duration,
+) -> Result<()> {
+    let result = tokio::time::timeout(deadline, host.knowledge_pull(coord))
+        .await
+        .unwrap_or_else(|_| {
+            Err(anyhow::anyhow!(
+                "knowledge pull timed out after {deadline:?}"
+            ))
+        });
     let mut health = knowledge_pull_health()
         .lock()
         .unwrap_or_else(std::sync::PoisonError::into_inner);
@@ -299,16 +442,8 @@ fn knowledge_pull_health() -> &'static Mutex<SweepHealth> {
 /// successful outcome the worker logs, so an error reaching here means the
 /// read itself failed and this hive is running blind to its declaration —
 /// which is exactly the state worth seeing on the dashboard.
-async fn run_wanted_pull(coord: &Arc<Coordinator>) -> Result<()> {
-    crate::workers::wanted::pull(coord).await
-}
-
-/// Refresh the host-side agent-set files from whatever containers still exist
-/// on disk. Idempotent. Lives here because the destroy tail is its only
-/// caller; it moved with `destroy` when that became a DAG.
-async fn refresh_agent_set_after_lifecycle() -> Result<()> {
-    let agents = crate::lifecycle::agents_for_meta_listing().await?;
-    crate::agent_set::refresh(&agents).await
+async fn run_wanted_pull(host: &impl Host, coord: &Arc<Coordinator>) -> Result<()> {
+    host.wanted_pull(coord).await
 }
 
 /// `nixos-container destroy`, then drop the agent from the roster and clear
@@ -320,10 +455,14 @@ async fn refresh_agent_set_after_lifecycle() -> Result<()> {
 /// the roster claiming an agent that no longer exists. A destroy that leaves the
 /// container in place fails this node before un-registration, and its failure
 /// is what keeps the `after_ok` purge and bookkeeping from running.
-async fn run_destroy_container(coord: &Arc<Coordinator>, agent: &str) -> Result<()> {
-    crate::lifecycle::destroy(agent).await?;
+async fn run_destroy_container(
+    host: &impl Host,
+    coord: &Arc<Coordinator>,
+    agent: &str,
+) -> Result<()> {
+    host.destroy(agent).await?;
     coord.unregister_agent(agent);
-    let runtime = crate::paths::agent_runtime_dir(agent);
+    let runtime = host.runtime_dir(agent);
     if runtime.exists() {
         let _ = std::fs::remove_dir_all(&runtime);
     }
@@ -336,24 +475,15 @@ async fn run_destroy_container(coord: &Arc<Coordinator>, agent: &str) -> Result<
 /// deliberate — the container is already destroyed by the time this runs, so
 /// failing the node would leave the operator with a half-purged agent and a red
 /// DAG, when what they can actually act on is the log line naming the path.
-async fn run_purge_state(agent: &str) {
+async fn run_purge_state(host: &impl Host, agent: &str) {
     // The state root may be a btrfs subvolume: a subvolume root can't be
     // removed with rmdir/`remove_dir_all`, so delete it via hive-priv (root)
     // first. No-op for plain-dir agents — the loop below then handles the
     // plain-dir state root plus the applied dir.
-    if let Err(e) = crate::priv_client::delete_agent_subvolume(agent).await {
+    if let Err(e) = host.delete_state_subvolume(agent).await {
         tracing::warn!(error = ?e, %agent, "purge: delete state subvolume failed");
     }
-    // A malformed name can't have a persistent state tree (the state dir is
-    // only ever created under a validated Ident), so its removal is a no-op —
-    // skip the state-dir sweep and just clear the applied dir.
-    let state_dir = hive_types::Ident::parse(agent)
-        .ok()
-        .map(|id| crate::paths::agent_state_dir(&id));
-    for dir in state_dir
-        .into_iter()
-        .chain([crate::paths::applied_dir(agent)])
-    {
+    for dir in host.purge_dirs(agent) {
         if dir.exists()
             && let Err(e) = std::fs::remove_dir_all(&dir)
         {
@@ -367,9 +497,9 @@ async fn run_purge_state(agent: &str) {
 /// none of this is undoable — a failed agent-set refresh or power-store write is a
 /// bookkeeping drift to log, not a reason to red a DAG whose container is
 /// already gone.
-async fn run_destroy_bookkeeping(coord: &Arc<Coordinator>, agent: &str) {
+async fn run_destroy_bookkeeping(host: &impl Host, coord: &Arc<Coordinator>, agent: &str) {
     // Drop the agent from the topology and the gateway files.
-    if let Err(e) = refresh_agent_set_after_lifecycle().await {
+    if let Err(e) = host.refresh_agent_set().await {
         tracing::warn!(error = ?e, %agent, "agent_set refresh after destroy failed");
     }
     // Drop the durable power intent — a future agent of the same name seeds
@@ -377,11 +507,9 @@ async fn run_destroy_bookkeeping(coord: &Arc<Coordinator>, agent: &str) {
     if let Err(e) = coord.power.remove(agent) {
         tracing::warn!(%agent, error = ?e, "agent_power: remove on destroy failed");
     }
-    crate::swarm_notices::notify(
-        "core",
-        Some(format!("destroyed:{agent}")),
+    host.notify(
+        format!("destroyed:{agent}"),
         format!("agent '{agent}' destroyed"),
-        None,
     )
     .await;
     // Container row disappeared — rescan so the dashboard fires
@@ -394,12 +522,18 @@ async fn run_destroy_bookkeeping(coord: &Arc<Coordinator>, agent: &str) {
 /// Emit this agent's rebuild-complete todo. `ok` is not computed — it is which
 /// of the tail pair the graph let run. The failure note comes from the DAG's
 /// first failing node, since the branch knows *that* it failed but not *why*.
-async fn run_emit_rebuilt(coord: &Arc<Coordinator>, agent: &str, dag_id: Option<u64>, ok: bool) {
+async fn run_emit_rebuilt(
+    host: &impl Host,
+    coord: &Arc<Coordinator>,
+    agent: &str,
+    dag_id: Option<u64>,
+    ok: bool,
+) {
     let note = (!ok)
         .then(|| dag_id.and_then(|dag| coord.job_queue.first_error(dag)))
         .flatten();
     let summary = crate::coordinator::rebuilt_todo_summary(agent, ok, note.as_deref(), None, None);
-    crate::swarm_notices::notify("core", Some(format!("rebuilt:{agent}")), summary, None).await;
+    host.notify(format!("rebuilt:{agent}"), summary).await;
 }
 
 /// Write the agent's durable power intent — the DAG-node form of the old
@@ -428,7 +562,7 @@ fn run_set_wanted(coord: &Arc<Coordinator>, agent: &str, up: bool) -> Result<()>
 ///
 /// Fails when `internal/meta` does not pin the agent's current config yet:
 /// there is no other flake to build it from.
-async fn run_meta_sync(name: &str) -> Result<()> {
+async fn run_meta_sync(host: &impl Host, name: &str) -> Result<()> {
     // 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).
@@ -436,15 +570,14 @@ async fn run_meta_sync(name: &str) -> Result<()> {
     let paths = Coordinator::agent_paths(name, agent_dir)?;
     // A failed read keeps the archetype last stored, so the agent is built as
     // it was.
-    if let Err(e) = crate::archetypes::refresh(name).await {
+    if let Err(e) = host.refresh_archetype(name).await {
         tracing::warn!(%name, error = ?e, "rebuild: reading the swarm's archetype failed; keeping the stored one");
     }
-    crate::lifecycle::prepare_rebuild_dirs(name, &paths).await?;
-    let agents = crate::lifecycle::agents_for_meta_listing().await?;
-    crate::agent_set::refresh(&agents).await?;
+    host.prepare_rebuild_dirs(name, &paths).await?;
+    host.refresh_agent_set().await?;
     // Resolved once here so `Prebuild` and `Swap` build the same flake.
-    let build = crate::meta_rev::resolve(name).await?;
-    crate::meta_rev::record(name, &build)?;
+    let build = host.resolve_meta(name).await?;
+    host.record_meta(name, &build)?;
     Ok(())
 }
 
@@ -503,9 +636,13 @@ async fn run_swap(
 /// means the profile swap succeeded. Store/forge work only — no nix
 /// build (build-slot-exempt); the agent lease taken at `Swap` is still held
 /// (the whole chain up to `Reconcile` is one agent's subgraph).
-async fn run_rebuild_bookkeeping(coord: &Arc<Coordinator>, name: &str) -> Result<()> {
-    if let Some(build) = crate::meta_rev::recorded(name)
-        && let Err(e) = crate::meta_rev::mark_deployed(name, &build)
+async fn run_rebuild_bookkeeping(
+    host: &impl Host,
+    coord: &Arc<Coordinator>,
+    name: &str,
+) -> Result<()> {
+    if let Some(build) = host.recorded_build(name)
+        && let Err(e) = host.mark_deployed(name, &build)
     {
         tracing::warn!(%name, error = ?e, "write rev markers failed");
     }
@@ -528,10 +665,10 @@ async fn run_rebuild_bookkeeping(coord: &Arc<Coordinator>, name: &str) -> Result
 
 /// First-spawn pre-create provisioning: proposed/applied repos, state
 /// subvolume, and the agent-set refresh.
-async fn run_provision(name: &str) -> Result<()> {
+async fn run_provision(host: &impl Host, name: &str) -> Result<()> {
     let agent_dir = crate::paths::agent_runtime_dir(name);
     let paths = Coordinator::agent_paths(name, agent_dir)?;
-    crate::lifecycle::provision_container(name, &paths).await?;
+    host.provision(name, &paths).await?;
     Ok(())
 }
 
@@ -541,10 +678,10 @@ async fn run_provision(name: &str) -> Result<()> {
 /// `Reconcile` (`converge_start_preamble` + `register_agent`) so this
 /// node stays purely "create", not "create + start". Marks the agent as
 /// running the pins it was created from.
-async fn run_create(name: &str) -> Result<()> {
-    let build = crate::meta_rev::resolve(name).await?;
-    crate::lifecycle::create_only(name, &build.rev).await?;
-    crate::meta_rev::mark_deployed(name, &build)?;
+async fn run_create(host: &impl Host, name: &str) -> Result<()> {
+    let build = host.resolve_meta(name).await?;
+    host.create_only(name, &build.rev).await?;
+    host.mark_deployed(name, &build)?;
     Ok(())
 }
 
@@ -582,7 +719,7 @@ async fn run_reconcile(
 
 /// Mechanical container start — the sub-step a `Reconcile` planner fans
 /// out when it observes `wanted = Up` and the container down.
-async fn run_start(coord: &Arc<Coordinator>, name: &str) -> Result<()> {
+async fn run_start(host: &impl Host, coord: &Arc<Coordinator>, name: &str) -> Result<()> {
     // No node-local transient guard: the pill is derived from the running node
     // set, and `Start` reports `Starting` via `NodeKind::transient_kind`. This
     // used to take one "only when the DAG holds none", which was a second
@@ -595,14 +732,13 @@ async fn run_start(coord: &Arc<Coordinator>, name: &str) -> Result<()> {
     let agent_dir = crate::paths::agent_runtime_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?;
-    crate::lifecycle::start_with_fallback(token).await?;
+    host.start(name, &hive, &paths).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)?;
+    host.register_agent(coord, name)?;
     // No `kick_agent` here any more — `hive-agent` pushes its own "you
     // were just (re)started" todo on boot now, so this node doesn't need
     // to tell it. See the matching comment in `run_rebuild_bookkeeping`.
@@ -617,13 +753,8 @@ async fn run_stop(host: &impl Host, coord: &Arc<Coordinator>, name: &str) -> Res
     // own kind now.
     host.kill(name).await?;
     coord.unregister_agent(name);
-    crate::swarm_notices::notify(
-        "core",
-        Some(format!("killed:{name}")),
-        format!("agent '{name}' killed"),
-        None,
-    )
-    .await;
+    host.notify(format!("killed:{name}"), format!("agent '{name}' killed"))
+        .await;
     coord.rescan_containers_and_emit().await;
     Ok(())
 }
@@ -735,14 +866,14 @@ async fn run_pause_drain(
 }
 
 /// `set_nspawn_flags` + `set_resource_limits` + daemon-reload.
-async fn run_write_dropin(coord: &Arc<Coordinator>, name: &str) -> Result<()> {
+async fn run_write_dropin(host: &impl Host, coord: &Arc<Coordinator>, name: &str) -> Result<()> {
     // 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 = crate::paths::agent_runtime_dir(name);
     let hive = coord.hive_env();
     let paths = Coordinator::agent_paths(name, agent_dir)?;
-    crate::lifecycle::write_dropins(name, &hive, &paths).await?;
+    host.write_dropins(name, &hive, &paths).await?;
     Ok(())
 }