feat(#2349): fan reconcile's start/stop out as first-class dag nodes

hive-c0re/src/job_queue/exec.rs

@@ -29,6 +29,15 @@ pub const GRACEFUL_STOP_TIMEOUT: std::time::Duration = std::time::Duration::from
 pub struct NodeOutput {
     /// Agents to fan child `Rebuild` DAGs out for (`MetaLock` only).
     pub fanout: Vec<String>,
+    /// Mechanical sub-step nodes to append into *this same* DAG at
+    /// runtime, each depending `AfterOk` on the emitting node — e.g. a
+    /// `Reconcile` planner emitting a `Start` / `Stop`. A separate
+    /// channel from `fanout` (which appends whole *child* DAGs) so the
+    /// sub-step stays a first-class node in the same DAG and the
+    /// lease-window transient is held across it. The scheduler applies
+    /// these *before* the emitting node's completion so the DAG never
+    /// rolls terminal with the appended work still pending.
+    pub append_nodes: Vec<NodeKind>,
 }
 
 /// Step-label + build-log sink for one claimed node.
@@ -76,7 +85,9 @@ pub(super) async fn run_node(coord: &Arc<Coordinator>, claim: &Claim) -> Result<
         NodeKind::MetaLock { sweep, fanout } => {
             run_meta_lock(coord, claim, &ctx, *sweep, fanout.clone()).await
         }
-        NodeKind::Reconcile => run_reconcile(coord, claim, &ctx).await,
+        NodeKind::Reconcile => run_reconcile(coord, claim).await,
+        NodeKind::Start => run_start(coord, claim, &ctx).await,
+        NodeKind::Stop => run_stop(coord, claim, &ctx).await,
         NodeKind::StopForUpdate => run_stop_for_update(coord, claim, &ctx).await,
         NodeKind::Signal => Ok(run_signal(coord, claim, &ctx)),
         NodeKind::Drain => run_drain(coord, claim, &ctx).await,
@@ -216,6 +227,7 @@ async fn run_meta_lock(
         }
         return Ok(NodeOutput {
             fanout: fanout.unwrap_or_default(),
+            ..Default::default()
         });
     }
     let _progress = coord.meta_update_guard();
@@ -230,63 +242,84 @@ async fn run_meta_lock(
         Some(list) => list,
         None => meta_update_cascade_agents(&claim.inputs).await,
     };
-    Ok(NodeOutput { fanout: cascade })
+    Ok(NodeOutput {
+        fanout: cascade,
+        ..Default::default()
+    })
 }
 
-/// Idempotent power converge: `wanted` (durable intent) vs observed.
+/// Idempotent power-converge *planner*: compare `wanted` (durable
-async fn run_reconcile(
+/// intent) against observed state and, when they diverge, fan the
-    coord: &Arc<Coordinator>,
+/// mechanical `Start` / `Stop` out as a first-class node appended to
-    claim: &Claim,
+/// *this* DAG (`NodeOutput::append_nodes`). Does no container work
-    ctx: &Ctx<'_>,
+/// itself — the sub-step becomes visible in the DAG and the
-) -> Result<NodeOutput> {
+/// lease-window transient (or the sub-step's own node-local guard)
+/// rides across it.
+async fn run_reconcile(coord: &Arc<Coordinator>, claim: &Claim) -> Result<NodeOutput> {
     let name = &claim.agent;
     let running = crate::lifecycle::is_running(name).await;
     let wanted = coord.power.get_or_seed(name, running)?;
-    match reconcile_action(wanted, running) {
+    let append_nodes = match reconcile_action(wanted, running) {
-        ReconcileAction::Start => {
+        ReconcileAction::Start => vec![NodeKind::Start],
-            // Node-local transient only when the DAG holds none (the
+        ReconcileAction::Stop => vec![NodeKind::Stop],
-            // boot-reconcile template); lease-window guards otherwise
-            // already cover this node.
-            let _guard = claim
-                .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 = 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?;
-            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;
-        }
-        ReconcileAction::Stop => {
-            let _guard = claim
-                .transient
-                .is_none()
-                .then(|| coord.transient_guard(name, crate::coordinator::TransientKind::Stopping));
-            ctx.step("nixos-container stop");
-            crate::lifecycle::kill(name).await?;
-            coord.unregister_agent(name);
-            coord.notify_manager(&hive_sh4re::HelperEvent::Killed {
-                agent: name.clone(),
-            });
-            coord.rescan_containers_and_emit().await;
-        }
         ReconcileAction::Noop => {
             tracing::debug!(%name, wanted = wanted.as_str(), running, "reconcile: noop");
+            Vec::new()
         }
-    }
+    };
+    Ok(NodeOutput {
+        append_nodes,
+        ..Default::default()
+    })
+}
+
+/// 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>, claim: &Claim, ctx: &Ctx<'_>) -> Result<NodeOutput> {
+    let name = &claim.agent;
+    // Node-local transient only when the DAG holds none (the
+    // boot-reconcile template); a rebuild/spawn/etc. DAG's lease-window
+    // transient already covers this node.
+    let _guard = claim
+        .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 = 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?;
+    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;
+    Ok(NodeOutput::default())
+}
+
+/// Mechanical container stop — the sub-step a `Reconcile` planner fans
+/// out when it observes `wanted = Offline` and the container up.
+async fn run_stop(coord: &Arc<Coordinator>, claim: &Claim, ctx: &Ctx<'_>) -> Result<NodeOutput> {
+    let name = &claim.agent;
+    let _guard = claim
+        .transient
+        .is_none()
+        .then(|| coord.transient_guard(name, crate::coordinator::TransientKind::Stopping));
+    ctx.step("nixos-container stop");
+    crate::lifecycle::kill(name).await?;
+    coord.unregister_agent(name);
+    coord.notify_manager(&hive_sh4re::HelperEvent::Killed {
+        agent: name.clone(),
+    });
+    coord.rescan_containers_and_emit().await;
     Ok(NodeOutput::default())
 }
 

hive-c0re/src/job_queue/mod.rs

@@ -169,6 +169,40 @@ impl JobQueue {
         ids
     }
 
+    /// Append a node into a *live* (non-terminal) DAG at runtime,
+    /// depending `AfterOk` on `dep_on` (the node that emitted it). Lets a
+    /// planner node — e.g. [`NodeKind::Reconcile`] — fan a mechanical
+    /// sub-step ([`NodeKind::Start`] / [`NodeKind::Stop`]) out as a
+    /// first-class node in the *same* DAG.
+    ///
+    /// Must be called *before* the emitting node's [`Self::complete_node`]
+    /// so the DAG doesn't roll terminal with the new node still pending —
+    /// that keeps the lease-window transient held across the sub-step and
+    /// lets the appended node's `AfterOk` dep resolve as soon as the
+    /// emitter settles `Done`. No-op (returns `None`) if the DAG is gone.
+    pub fn append_node(&self, dag_id: u64, kind: NodeKind, dep_on: NodeId) -> Option<NodeId> {
+        let mut inner = self.inner.lock().expect("job_queue mutex poisoned");
+        let dag = inner.dags.iter_mut().find(|d| d.id == dag_id)?;
+        let new_id: NodeId = u32::try_from(dag.nodes.len()).unwrap_or(u32::MAX);
+        dag.nodes.push(Node {
+            id: new_id,
+            kind,
+            deps: vec![model::Dep {
+                on: dep_on,
+                when: DepWhen::AfterOk,
+            }],
+            state: State::Queued,
+            step: None,
+            build_log_id: None,
+            started_at: None,
+            finished_at: None,
+            error: None,
+        });
+        drop(inner);
+        self.notify.notify_one();
+        Some(new_id)
+    }
+
     fn dedup_target<'a>(inner: &'a mut Inner, spec: &DagSpec) -> Option<&'a mut Dag> {
         inner.dags.iter_mut().find(|d| {
             d.rollup() == State::Queued

hive-c0re/src/job_queue/model.rs

@@ -90,10 +90,21 @@ pub enum NodeKind {
         sweep: bool,
         fanout: Option<Vec<String>>,
     },
-    /// Idempotent power converge: read `wanted` + observed state;
+    /// Idempotent power converge *planner*: read `wanted` + observed
-    /// start if `Up` & down (with cold-start fallback), stop if
+    /// state and decide the action (start if `Up` & down, stop if
-    /// `Offline` & up, else noop.
+    /// `Offline` & up, else noop). The mechanical work is not done in
+    /// this node — it fans a child [`NodeKind::Start`] / [`NodeKind::Stop`]
+    /// DAG out at runtime so the sub-step is a first-class DAG node.
     Reconcile,
+    /// Mechanical container start: the start preamble (runtime dir +
+    /// drop-ins), `start_with_fallback`, MCP listener registration, and
+    /// the manager kick. Fanned out by a [`NodeKind::Reconcile`] that
+    /// observed `wanted = Up` and the container down.
+    Start,
+    /// Mechanical container stop: `nixos-container` kill, MCP listener
+    /// unregister, and the `Killed` manager notify. Fanned out by a
+    /// [`NodeKind::Reconcile`] that observed `wanted = Offline` and up.
+    Stop,
     /// Mechanical `nixos-container stop` for the profile swap. Never
     /// touches `wanted`. Noop if already stopped.
     StopForUpdate,
@@ -125,6 +136,8 @@ impl NodeKind {
             NodeKind::Create => "create",
             NodeKind::MetaLock { .. } => "meta_lock",
             NodeKind::Reconcile => "reconcile",
+            NodeKind::Start => "start",
+            NodeKind::Stop => "stop",
             NodeKind::StopForUpdate => "stop_for_update",
             NodeKind::Signal => "signal",
             NodeKind::Drain => "drain",

hive-c0re/src/job_queue/scheduler.rs

@@ -92,14 +92,32 @@ async fn handle_completion(
     done: NodeDone,
 ) {
     let NodeDone { claim, result } = done;
-    let (queue_result, fanout) = match result {
+    match result {
         Ok(output) => {
             tracing::info!(
                 dag = claim.dag_id,
                 node = claim.node_id,
                 "job_queue: node done"
             );
-            (Ok(()), output.fanout)
+            // Append any in-DAG sub-step nodes (e.g. a `Reconcile`
+            // planner's `Start` / `Stop`) BEFORE completing this node, so
+            // completing it doesn't roll the DAG terminal while the
+            // appended work is still pending — that keeps the lease-window
+            // transient held across the sub-step. Each depends `AfterOk`
+            // on this node, so it becomes ready the instant this one
+            // settles `Done` just below.
+            for kind in output.append_nodes {
+                coord
+                    .job_queue
+                    .append_node(claim.dag_id, kind, claim.node_id);
+            }
+            coord
+                .job_queue
+                .complete_node(claim.dag_id, claim.node_id, Ok(()));
+            if !output.fanout.is_empty() {
+                let specs = fanout_specs(&claim, output.fanout);
+                coord.job_queue.append_children(specs);
+            }
         }
         Err(e) => {
             let msg = format!("{e:#}");
@@ -111,15 +129,10 @@ async fn handle_completion(
                 error = %msg,
                 "job_queue: node failed"
             );
-            (Err(msg), Vec::new())
+            coord
+                .job_queue
+                .complete_node(claim.dag_id, claim.node_id, Err(msg));
         }
-    };
-    coord
-        .job_queue
-        .complete_node(claim.dag_id, claim.node_id, queue_result);
-    if !fanout.is_empty() {
-        let specs = fanout_specs(&claim, fanout);
-        coord.job_queue.append_children(specs);
     }
     process_terminals(coord, transients).await;
     coord.emit_rebuild_queue_snapshot();