feat(hive-c0re): replace rebuild queue with generic job-DAG queue

jobs are now DAGs of primitive nodes (prebuild, stop-for-update, swap,
reconcile, signal, drain, ...) driven by one scheduler with N build
slots + per-agent lifecycle leases. per-agent power intent (wanted
up/offline) is durable in agent_power.sqlite; Reconcile nodes converge
observed state to it. kills the graceful-stop watcher thread, the
deferred-start follow-up, and the cascade pre-enqueue (fan-out on
MetaLock completion instead). tracker: #2166

hive-c0re/src/job_queue/scheduler.rs

@@ -0,0 +1,150 @@
+//! The single scheduler task that drives all DAGs: claim every ready
+//! node (as many as the build slots / leases allow), spawn one
+//! executor task per claim, and on any completion re-evaluate.
+//! Concurrency comes from the build-slot count, not multiple workers.
+//!
+//! Also owns the two DAG-lifetime side channels the sync queue core
+//! can't hold itself:
+//! - the per-DAG transient guard (dashboard pill + crash-watch
+//!   suppression), created when a DAG acquires its agent lease and
+//!   dropped when the DAG settles terminal;
+//! - the `MetaLock` fan-out: appending child `Rebuild` DAGs once the
+//!   lock bump lands, so children build against the post-bump lock
+//!   (and a failed bump fans out nothing — replacing the old
+//!   pre-enqueue + cancel-children dance).
+
+use std::collections::HashMap;
+use std::sync::Arc;
+
+use super::exec::{self, NodeOutput};
+use super::{Claim, Source, Template, templates};
+use crate::coordinator::Coordinator;
+
+struct NodeDone {
+    claim: Claim,
+    result: anyhow::Result<NodeOutput>,
+}
+
+/// Scheduler loop. Spawned once at hive-c0re startup from `main.rs`.
+///
+/// Shutdown semantics: subscribes to `coord.shutdown_rx()`. On a true
+/// signal the loop exits immediately; already-running node tasks ride
+/// the runtime down with the process, and pending `Queued` DAGs are
+/// dropped — desired state is re-derived on next boot (boot sweep +
+/// reconcile), so the in-memory queue is deliberately not durable.
+pub async fn run_worker(coord: Arc<Coordinator>) {
+    let mut shutdown = coord.shutdown_rx();
+    let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel::<NodeDone>();
+    // DAG id → transient guard held for the lease window.
+    let mut transients: HashMap<u64, crate::coordinator::TransientGuard> = HashMap::new();
+    loop {
+        let claims = coord.job_queue.claim_ready();
+        if !claims.is_empty() {
+            for claim in claims {
+                if claim.lease_acquired
+                    && let Some(kind) = claim.transient
+                {
+                    transients.insert(claim.dag_id, coord.transient_guard(&claim.agent, kind));
+                }
+                tracing::info!(
+                    dag = claim.dag_id,
+                    node = claim.node_id,
+                    kind = claim.kind.as_str(),
+                    agent = %claim.agent,
+                    template = claim.template.as_str(),
+                    "job_queue: node running"
+                );
+                let coord = Arc::clone(&coord);
+                let tx = tx.clone();
+                tokio::spawn(async move {
+                    let result = exec::run_node(&coord, &claim).await;
+                    // Send failure = scheduler gone (shutdown); drop.
+                    let _ = tx.send(NodeDone { claim, result });
+                });
+            }
+            coord.emit_rebuild_queue_snapshot();
+            continue;
+        }
+        tokio::select! {
+            biased;
+            res = shutdown.changed() => {
+                if res.is_err() || *shutdown.borrow() {
+                    tracing::info!("job_queue: scheduler exiting on shutdown");
+                    return;
+                }
+            }
+            Some(done) = rx.recv() => {
+                handle_completion(&coord, &mut transients, done).await;
+            }
+            () = coord.job_queue.notify.notified() => {}
+        }
+    }
+}
+
+async fn handle_completion(
+    coord: &Arc<Coordinator>,
+    transients: &mut HashMap<u64, crate::coordinator::TransientGuard>,
+    done: NodeDone,
+) {
+    let NodeDone { claim, result } = done;
+    let (queue_result, fanout) = match result {
+        Ok(output) => {
+            tracing::info!(
+                dag = claim.dag_id,
+                node = claim.node_id,
+                "job_queue: node done"
+            );
+            (Ok(()), output.fanout)
+        }
+        Err(e) => {
+            let msg = format!("{e:#}");
+            tracing::warn!(
+                dag = claim.dag_id,
+                node = claim.node_id,
+                kind = claim.kind.as_str(),
+                agent = %claim.agent,
+                error = %msg,
+                "job_queue: node failed"
+            );
+            (Err(msg), Vec::new())
+        }
+    };
+    let report = 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);
+    }
+    for terminal in report.terminal {
+        // Drop the lease-window transient guard, then let the hook
+        // fire approval resolution / failure events.
+        transients.remove(&terminal.dag_id);
+        exec::on_dag_terminal(coord, &terminal).await;
+    }
+    coord.emit_rebuild_queue_snapshot();
+}
+
+/// Child `Rebuild` specs for a completed `MetaLock` fan-out, grouped
+/// under the parent via `parent_id`. Meta-update children skip the
+/// per-agent relock (it would revert the bump the parent just
+/// committed); sweep children relock like a manual rebuild.
+fn fanout_specs(claim: &Claim, agents: Vec<String>) -> Vec<super::DagSpec> {
+    let sweep = claim.template == Template::StartupSweep;
+    let (source, relock) = if sweep {
+        (Source::StartupSweep, true)
+    } else {
+        (Source::MetaUpdate, false)
+    };
+    let reason = if sweep {
+        "startup sweep".to_owned()
+    } else if let Some(approval_id) = claim.approval_id {
+        format!("approval #{approval_id} meta input cascade")
+    } else {
+        "meta-update cascade".to_owned()
+    };
+    agents
+        .into_iter()
+        .map(|agent| templates::rebuild(&agent, source, reason.clone(), Some(claim.dag_id), relock))
+        .collect()
+}