diff --git a/hive-c0re/src/job_queue/exec.rs b/hive-c0re/src/job_queue/exec.rs
index de9e1f7d..fa303189 100644
--- a/hive-c0re/src/job_queue/exec.rs
+++ b/hive-c0re/src/job_queue/exec.rs
@@ -10,8 +10,7 @@ use std::sync::Arc;
 
 use anyhow::{Context as _, Result};
 
-use super::Claim;
-use hive_jobq::TerminalState;
+use hive_jobq::{NodeId, TerminalState};
 
 use super::model::NodeKind;
 use super::resource::Resource;
@@ -43,22 +42,31 @@ pub const GRACEFUL_STOP_TIMEOUT: std::time::Duration = std::time::Duration::from
 /// future non-`Send`, which the scheduler's `tokio::spawn` rejects. So the
 /// growth executors below return *what to grow* and the declaration happens
 /// here, synchronously, between awaits.
+///
+/// The node is identified by its own id + payload rather than by a `Claim`
+/// side-struct: `kind` already carries the agent, and the DAG id is a
+/// derived read (`JobQueue::dag_of`) the three arms that need it take
+/// themselves. Nothing here needs a claim to exist as a type.
 pub(super) async fn run_node(
     coord: &Arc<Coordinator>,
     job: super::Job,
-    claim: &Claim,
+    id: NodeId,
+    kind: &NodeKind,
 ) -> (super::Job, Result<()>) {
+    // The agent this node targets rides the payload — empty for the agentless
+    // container kinds (`MetaLock`, `Dag`), which never read it.
+    let agent = kind.agent();
     // Every arm is `Result<()>`; the three that grow work declare into `job`
     // *synchronously*, after their own awaits have finished. Borrowing `&job`
     // 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 &claim.kind {
-        NodeKind::MetaSync { relock, .. } => run_meta_sync(coord, claim, *relock).await,
-        NodeKind::Prebuild { .. } => run_prebuild(claim).await,
-        NodeKind::Swap { .. } => run_swap(coord, claim).await,
-        NodeKind::PostSwap { .. } => run_post_swap(coord, claim).await,
-        NodeKind::Provision { .. } => run_provision(coord, claim).await,
-        NodeKind::Create { .. } => run_create(claim).await,
+    let result = match kind {
+        NodeKind::MetaSync { relock, .. } => run_meta_sync(coord, agent, *relock).await,
+        NodeKind::Prebuild { .. } => run_prebuild(agent, id).await,
+        NodeKind::Swap { .. } => run_swap(coord, agent, id).await,
+        NodeKind::PostSwap { .. } => run_post_swap(coord, agent).await,
+        NodeKind::Provision { .. } => run_provision(coord, agent).await,
+        NodeKind::Create { .. } => run_create(agent).await,
         NodeKind::MetaLock {
             sweep,
             fanout,
@@ -70,7 +78,7 @@ pub(super) async fn run_node(
                     super::templates::rebuild_nodes(&job, &agent, opts, None);
                 }
             }),
-        NodeKind::Reconcile { .. } => run_reconcile(coord, claim).await.map(|sub| {
+        NodeKind::Reconcile { .. } => run_reconcile(coord, agent).await.map(|sub| {
             if let Some(kind) = sub {
                 // `Start` / `Stop` declare the lease they run under. This node
                 // is their parent and holds it, so the declaration is a
@@ -81,38 +89,41 @@ pub(super) async fn run_node(
                 let _ = job.node(kind).needs(lease);
             }
         }),
-        NodeKind::Start { .. } => run_start(coord, claim).await,
-        NodeKind::Stop { .. } => run_stop(coord, claim).await,
-        NodeKind::StopForUpdate { .. } => run_stop_for_update(coord, claim).await,
+        NodeKind::Start { .. } => run_start(coord, agent).await,
+        NodeKind::Stop { .. } => run_stop(coord, agent).await,
+        NodeKind::StopForUpdate { .. } => run_stop_for_update(coord, agent).await,
         NodeKind::Signal { .. } => {
-            run_signal(coord, claim);
+            run_signal(coord, agent);
             Ok(())
         }
-        NodeKind::Drain { .. } => run_drain(coord, claim).await,
-        NodeKind::WriteDropin { .. } => run_write_dropin(coord, claim).await,
-        NodeKind::WritePermFile { .. } => run_write_perm_file(coord, claim).await,
-        NodeKind::Reparent { .. } => run_reparent(coord, claim).await,
+        NodeKind::Drain { .. } => run_drain(coord, agent).await,
+        NodeKind::WriteDropin { .. } => run_write_dropin(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::WritePermFile { payload, .. } => run_write_perm_file(coord, agent, payload).await,
+        NodeKind::Reparent { moves } => run_reparent(coord, moves).await,
         NodeKind::MergeVerify { approval_id, .. } => run_merge_verify(coord, *approval_id).await,
         NodeKind::DeployApply { approval_id, .. } => {
             run_deploy_apply(coord, *approval_id).await.map(|()| {
-                super::templates::deploy_rebuild_nodes(&job, claim.kind.agent(), *approval_id);
+                super::templates::deploy_rebuild_nodes(&job, agent, *approval_id);
             })
         }
         NodeKind::FinalizeDeploy { approval_id, .. } => {
             run_finalize_deploy(coord, *approval_id).await
         }
         NodeKind::DeployTail { approval_id, .. } => {
-            run_deploy_tail(coord, claim, *approval_id).await
+            run_deploy_tail(coord, coord.job_queue.dag_of(id), agent, *approval_id).await
         }
         NodeKind::ResolveApproval {
             approval_id,
             outcome,
-        } => run_resolve_approval(coord, claim, *approval_id, *outcome).await,
+        } => run_resolve_approval(coord, coord.job_queue.dag_of(id), *approval_id, *outcome).await,
         NodeKind::EmitRebuilt { ok, .. } => {
-            run_emit_rebuilt(coord, claim, *ok);
+            run_emit_rebuilt(coord, agent, coord.job_queue.dag_of(id), *ok);
             Ok(())
         }
-        NodeKind::SetWanted { up, .. } => run_set_wanted(coord, claim, *up),
+        NodeKind::SetWanted { up, .. } => run_set_wanted(coord, agent, *up),
         // The two nodes that carry no work of their own; completing either
         // lets it reach `Finishing` so the nodes under it start.
         // - `Dag`: pure grouping container. The DAG's terminal side effect, if
@@ -133,12 +144,12 @@ pub(super) async fn run_node(
 /// since the work already happened and failing the tail would only misreport it.
 async fn run_resolve_approval(
     coord: &Arc<Coordinator>,
-    claim: &Claim,
+    dag_id: Option<u64>,
     approval_id: i64,
     outcome: TerminalState,
 ) -> Result<()> {
     let reason = (outcome == TerminalState::Failed)
-        .then(|| coord.job_queue.first_error(claim.dag_id))
+        .then(|| dag_id.and_then(|dag| coord.job_queue.first_error(dag)))
         .flatten();
     crate::actions::resolve_approval_dag(coord, approval_id, outcome, reason.as_deref()).await;
     Ok(())
@@ -147,12 +158,12 @@ async fn run_resolve_approval(
 /// Emit this agent's `Rebuilt` manager event. `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*.
-fn run_emit_rebuilt(coord: &Arc<Coordinator>, claim: &Claim, ok: bool) {
+fn run_emit_rebuilt(coord: &Arc<Coordinator>, agent: &str, dag_id: Option<u64>, ok: bool) {
     coord.notify_manager(&hive_sh4re::HelperEvent::Rebuilt {
-        agent: claim.agent.clone(),
+        agent: agent.to_owned(),
         ok,
         note: (!ok)
-            .then(|| coord.job_queue.first_error(claim.dag_id))
+            .then(|| dag_id.and_then(|dag| coord.job_queue.first_error(dag)))
             .flatten(),
         sha: None,
         tag: None,
@@ -167,7 +178,7 @@ fn run_emit_rebuilt(coord: &Arc<Coordinator>, claim: &Claim, ok: bool) {
 /// warn-and-continue write, a failed write fails the node (cancel-downstream
 /// cancels the `Reconcile`) rather than letting it converge to a stale
 /// intent — that atomicity is the point of moving it into the DAG.
-fn run_set_wanted(coord: &Arc<Coordinator>, claim: &Claim, up: bool) -> Result<()> {
+fn run_set_wanted(coord: &Arc<Coordinator>, agent: &str, up: bool) -> Result<()> {
     let wanted = if up {
         crate::power::Wanted::Up
     } else {
@@ -175,8 +186,8 @@ fn run_set_wanted(coord: &Arc<Coordinator>, claim: &Claim, up: bool) -> Result<(
     };
     coord
         .power
-        .set(&claim.agent, wanted)
-        .with_context(|| format!("set wanted={} for agent {}", wanted.as_str(), claim.agent))?;
+        .set(agent, wanted)
+        .with_context(|| format!("set wanted={} for agent {agent}", wanted.as_str()))?;
     Ok(())
 }
 
@@ -189,8 +200,7 @@ fn run_set_wanted(coord: &Arc<Coordinator>, claim: &Claim, up: bool) -> Result<(
 /// Deliberately a separate node from the [`run_prebuild`] it feeds: that
 /// build takes minutes and only *reads* the store, so keeping the global
 /// window off it is what lets rebuilds of different agents overlap.
-async fn run_meta_sync(coord: &Arc<Coordinator>, claim: &Claim, relock: bool) -> Result<()> {
-    let name = &claim.agent;
+async fn run_meta_sync(coord: &Arc<Coordinator>, name: &str, relock: bool) -> 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).
@@ -217,15 +227,14 @@ async fn run_meta_sync(coord: &Arc<Coordinator>, claim: &Claim, relock: bool) ->
 /// container is already down: its only purpose is to shrink the swap's
 /// downtime window, so a stopped agent (no uptime to preserve) doesn't
 /// pay the double eval — `Swap` builds inline instead.
-async fn run_prebuild(claim: &Claim) -> Result<()> {
-    let name = &claim.agent;
+async fn run_prebuild(name: &str, id: NodeId) -> Result<()> {
     // Warm the toplevel build only when the container is up — the whole
     // point of prebuild is to shrink the swap's downtime window. A
     // stopped agent has no uptime to preserve, so skip the (expensive)
     // eval and let the downstream `Swap` build inline.
     if crate::lifecycle::is_running(name).await {
         let flake_ref = format!("{}#{name}", crate::paths::meta_root().display());
-        crate::lifecycle::prebuild_toplevel(name, &flake_ref, Some(claim.node_id.get())).await?;
+        crate::lifecycle::prebuild_toplevel(name, &flake_ref, Some(id.get())).await?;
     }
     Ok(())
 }
@@ -235,15 +244,13 @@ async fn run_prebuild(claim: &Claim) -> Result<()> {
 /// (rev marker, `Rebuilt` event, forge/matrix sync, kick, rescan).
 /// The recovery-start on failure is NOT here — the DAG's tail
 /// `Reconcile` runs after this node terminal ok *or* fail.
-async fn run_swap(coord: &Arc<Coordinator>, claim: &Claim) -> Result<()> {
-    let name = &claim.agent;
+async fn run_swap(coord: &Arc<Coordinator>, name: &str, id: NodeId) -> Result<()> {
     // Swap runs on an already-existing (stopped) container — runtime dir
     // and listener were created earlier. Pure path accessor suffices.
     let agent_dir = crate::paths::agent_runtime_dir(name);
     let hive = coord.hive_env();
     let paths = Coordinator::agent_paths(name, agent_dir);
-    let result =
-        crate::lifecycle::swap_update(name, &hive, &paths, Some(claim.node_id.get())).await;
+    let result = crate::lifecycle::swap_update(name, &hive, &paths, Some(id.get())).await;
     // On success the Ok-only bookkeeping tail (rev marker, forge/matrix
     // sync, kick, rescan, snapshot) runs in the sibling `PostSwap` node,
     // which deps `AfterOk(Swap)`. On failure `PostSwap` is cancel-cascaded
@@ -262,8 +269,7 @@ async fn run_swap(coord: &Arc<Coordinator>, claim: &Claim) -> Result<()> {
 /// means the profile swap succeeded. Store/forge/matrix 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_post_swap(coord: &Arc<Coordinator>, claim: &Claim) -> Result<()> {
-    let name = &claim.agent;
+async fn run_post_swap(coord: &Arc<Coordinator>, name: &str) -> Result<()> {
     if let Some(rev) = crate::auto_update::current_flake_rev(&coord.hyperhive_flake)
         && let Err(e) = std::fs::write(crate::paths::applied_rev_marker(name), rev)
     {
@@ -290,8 +296,7 @@ async fn run_post_swap(coord: &Arc<Coordinator>, claim: &Claim) -> Result<()> {
 /// subvolume, and the meta `sync_agents` registration. Runs under the
 /// deploy window (it declares `Resource::MetaWindow`) so its commit can't
 /// land inside another node's staged deploy window.
-async fn run_provision(coord: &Arc<Coordinator>, claim: &Claim) -> Result<()> {
-    let name = &claim.agent;
+async fn run_provision(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);
@@ -305,8 +310,8 @@ async fn run_provision(coord: &Arc<Coordinator>, claim: &Claim) -> Result<()> {
 /// dir creation and MCP listener registration are deferred to the tail
 /// `Reconcile` (`converge_start_preamble` + `register_agent`) so this
 /// node stays purely "create", not "create + start".
-async fn run_create(claim: &Claim) -> Result<()> {
-    crate::lifecycle::create_only(&claim.agent).await?;
+async fn run_create(name: &str) -> Result<()> {
+    crate::lifecycle::create_only(name).await?;
     Ok(())
 }
 
@@ -377,18 +382,17 @@ async fn run_meta_lock(
 /// guard) rides across it.
 /// Returns the mechanical node to fan out (`None` on a noop) rather than
 /// declaring it — the declaration has to happen outside any `.await`, see
-/// [`run_node`]. `NodeKind` carries the agent it targets, so `claim.agent` is
-/// stamped into the kind here.
-async fn run_reconcile(coord: &Arc<Coordinator>, claim: &Claim) -> Result<Option<NodeKind>> {
-    let name = &claim.agent;
+/// [`run_node`]. `NodeKind` carries the agent it targets, so this node's agent
+/// is stamped into the fanned-out kind here.
+async fn run_reconcile(coord: &Arc<Coordinator>, name: &str) -> Result<Option<NodeKind>> {
     let running = crate::lifecycle::is_running(name).await;
     let wanted = coord.power.get_or_seed(name, running)?;
     Ok(match reconcile_action(wanted, running) {
         ReconcileAction::Start => Some(NodeKind::Start {
-            agent: name.clone(),
+            agent: name.to_owned(),
         }),
         ReconcileAction::Stop => Some(NodeKind::Stop {
-            agent: name.clone(),
+            agent: name.to_owned(),
         }),
         ReconcileAction::Noop => {
             tracing::debug!(%name, wanted = wanted.as_str(), running, "reconcile: noop");
@@ -399,8 +403,7 @@ async fn run_reconcile(coord: &Arc<Coordinator>, claim: &Claim) -> Result<Option
 
 /// 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) -> Result<()> {
-    let name = &claim.agent;
+async fn run_start(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
@@ -428,14 +431,13 @@ async fn run_start(coord: &Arc<Coordinator>, claim: &Claim) -> Result<()> {
 
 /// 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) -> Result<()> {
-    let name = &claim.agent;
+async fn run_stop(coord: &Arc<Coordinator>, name: &str) -> Result<()> {
     // See `run_start`: no node-local guard — `Stop` reports `Stopping` from its
     // own kind now.
     crate::lifecycle::kill(name).await?;
     coord.unregister_agent(name);
     coord.notify_manager(&hive_sh4re::HelperEvent::Killed {
-        agent: name.clone(),
+        agent: name.to_owned(),
     });
     coord.rescan_containers_and_emit().await;
     Ok(())
@@ -443,8 +445,7 @@ async fn run_stop(coord: &Arc<Coordinator>, claim: &Claim) -> Result<()> {
 
 /// Mechanical stop for the profile swap. Never *changes* `wanted`;
 /// noop when already stopped.
-async fn run_stop_for_update(coord: &Arc<Coordinator>, claim: &Claim) -> Result<()> {
-    let name = &claim.agent;
+async fn run_stop_for_update(coord: &Arc<Coordinator>, name: &str) -> Result<()> {
     if crate::lifecycle::is_running(name).await {
         // Seed a missing agent_power row from the PRE-stop observation
         // — the DAG's tail `Reconcile` observes only the mechanically
@@ -468,19 +469,18 @@ async fn run_stop_for_update(coord: &Arc<Coordinator>, claim: &Claim) -> Result<
 /// `GRACEFUL_STOP_TIMEOUT`. Safe because the harness tests the marker at
 /// the top of its loop — a paused agent has no turn in flight, so there
 /// is nothing to checkpoint.
-fn run_signal(coord: &Arc<Coordinator>, claim: &Claim) {
-    if hive_types::Ident::parse(&claim.agent).is_ok_and(|a| Coordinator::is_paused(&a)) {
+fn run_signal(coord: &Arc<Coordinator>, name: &str) {
+    if hive_types::Ident::parse(name).is_ok_and(|a| Coordinator::is_paused(&a)) {
         return;
     }
-    coord.mark_graceful_stop(&claim.agent);
-    coord.kick_agent(&claim.agent, "graceful stop requested");
+    coord.mark_graceful_stop(name);
+    coord.kick_agent(name, "graceful stop requested");
 }
 
 /// Await the harness clearing the fence (`GracefulStopComplete`) or
 /// the timeout — either way the downstream `Reconcile` proceeds with
 /// the actual stop.
-async fn run_drain(coord: &Arc<Coordinator>, claim: &Claim) -> Result<()> {
-    let name = &claim.agent;
+async fn run_drain(coord: &Arc<Coordinator>, name: &str) -> Result<()> {
     let deadline = std::time::Instant::now() + GRACEFUL_STOP_TIMEOUT;
     while coord.is_graceful_stop_pending(name) {
         if std::time::Instant::now() >= deadline {
@@ -494,8 +494,7 @@ async fn run_drain(coord: &Arc<Coordinator>, claim: &Claim) -> Result<()> {
 }
 
 /// `set_nspawn_flags` + `set_resource_limits` + daemon-reload.
-async fn run_write_dropin(coord: &Arc<Coordinator>, claim: &Claim) -> Result<()> {
-    let name = &claim.agent;
+async fn run_write_dropin(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).
@@ -509,13 +508,12 @@ async fn run_write_dropin(coord: &Arc<Coordinator>, claim: &Claim) -> Result<()>
 /// Write + commit the perm file(s) (fused under `META_LOCK` so the
 /// working tree is never left dirty), then emit the P3RM1SS10NS-tab
 /// snapshots so the dashboard reflects the new assignment.
-async fn run_write_perm_file(coord: &Arc<Coordinator>, claim: &Claim) -> Result<()> {
+async fn run_write_perm_file(
+    coord: &Arc<Coordinator>,
+    name: &str,
+    payload: &super::model::PermPayload,
+) -> Result<()> {
     use super::model::PermPayload;
-    let name = &claim.agent;
-    // The perm file payload rides the node itself (the only consumer).
-    let NodeKind::WritePermFile { payload, .. } = &claim.kind else {
-        anyhow::bail!("run_write_perm_file on a non-WritePermFile node");
-    };
     // Runs under the deploy window (it declares `Resource::MetaWindow`): a
     // perm commit landing inside another node's staged prepare→finalize
     // window would sweep the staged deploy lock into its commit (the
@@ -555,10 +553,10 @@ async fn run_write_perm_file(coord: &Arc<Coordinator>, claim: &Claim) -> Result<
 /// the deploy window (it declares `Resource::MetaWindow`), same reasoning as
 /// `run_write_perm_file`: a topology commit landing inside another node's
 /// staged deploy window would sweep the staged lock into its commit.
-async fn run_reparent(coord: &Arc<Coordinator>, claim: &Claim) -> Result<()> {
-    let NodeKind::Reparent { moves } = &claim.kind else {
-        anyhow::bail!("run_reparent on a non-Reparent node");
-    };
+async fn run_reparent(
+    coord: &Arc<Coordinator>,
+    moves: &[(hive_types::Ident, Option<hive_types::Ident>)],
+) -> Result<()> {
     let refs: Vec<(&str, Option<&str>)> = moves
         .iter()
         .map(|(child, parent)| {
@@ -609,9 +607,13 @@ async fn run_finalize_deploy(coord: &Arc<Coordinator>, approval_id: i64) -> Resu
 ///
 /// Takes the agent from the node payload so the tail can still compensate when
 /// the approval row is gone (deny race, purge).
-async fn run_deploy_tail(coord: &Arc<Coordinator>, claim: &Claim, approval_id: i64) -> Result<()> {
-    crate::actions::run_deploy_tail(coord, Some(claim.dag_id), claim.kind.agent(), approval_id)
-        .await;
+async fn run_deploy_tail(
+    coord: &Arc<Coordinator>,
+    dag_id: Option<u64>,
+    agent: &str,
+    approval_id: i64,
+) -> Result<()> {
+    crate::actions::run_deploy_tail(coord, dag_id, agent, approval_id).await;
     Ok(())
 }
 
diff --git a/hive-c0re/src/job_queue/mod.rs b/hive-c0re/src/job_queue/mod.rs
index 2780ee6a..e5d2ebee 100644
--- a/hive-c0re/src/job_queue/mod.rs
+++ b/hive-c0re/src/job_queue/mod.rs
@@ -88,18 +88,6 @@ pub struct RunningTransient {
     pub since: DateTime<Utc>,
 }
 
-/// A node claimed for execution — everything the executor needs, snapshotted at
-/// claim time.
-#[derive(Debug, Clone)]
-pub struct Claim {
-    pub dag_id: u64,
-    pub node_id: NodeId,
-    pub kind: NodeKind,
-    /// The agent this node targets (its own, not a DAG-level field). Empty for
-    /// the agentless [`NodeKind::MetaLock`] + [`NodeKind::Dag`] container nodes.
-    pub agent: String,
-}
-
 /// An owned read-view of a DAG container's carried metadata ([`NodeKind::Dag`]).
 /// Derived on read from the container node — the data has a single home (the
 /// node payload); this is not a stored side-table.
@@ -245,88 +233,41 @@ impl JobQueue {
         Ok(container.get())
     }
 
-    /// Claim every currently-runnable node, acquiring its resources, and mark it
-    /// `Running`. Delegates readiness + resource acquisition to the crate's
-    /// settle loop; builds a [`Claim`] per started node from its payload + its
-    /// DAG container's metadata. The container node itself is claimed like any
-    /// other (its executor is an instant no-op that lets its subtree start).
-    pub fn claim_ready(&self) -> Vec<Claim> {
-        let mut inner = self.lock();
-        let inner = &mut *inner;
-        let started = inner.settle();
-        let mut claims = Vec::with_capacity(started.len());
-        for id in started {
-            let Some(node) = inner.graph().node(id) else {
-                continue;
-            };
-            let kind = node.payload.clone();
-            let agent = node.payload.agent().to_owned();
-            let Some(container) = inner.graph().root_of(id) else {
-                continue;
-            };
-            claims.push(Claim {
-                dag_id: container.get(),
-                node_id: id,
-                kind,
-                agent,
-            });
-            // `started_at` is stamped on the graph `Node` by the scheduler's
-            // transition to `Running` — no host-side copy needed.
-        }
-        claims
+    /// The scheduler itself, for `hive_jobq`'s run-loop seam
+    /// (`Scheduler::claim_next`), which takes exactly this type.
+    ///
+    /// Handing out the `Arc` rather than wrapping each crate call keeps the
+    /// host from growing a parallel API: the run loop uses `hive_jobq`'s
+    /// functions directly, and this module stays the thin glue it is being
+    /// reduced to.
+    pub(crate) fn sched(&self) -> &Arc<Mutex<Sched>> {
+        &self.sched
     }
 
-    /// Mark a claimed node terminal, recording its outcome + (truncated) error.
-    /// The crate releases the node's build slot immediately and cascades the
-    /// `AfterOk` failure cancellation + subtree lease release.
-    ///
-    /// Nothing is returned: a DAG's terminal side effects are its own tail nodes
-    /// ([`NodeKind::ResolveApproval`] / [`NodeKind::EmitRebuilt`]), which the
-    /// scheduler claims and runs like any other node.
-    pub fn complete_node(&self, node_id: NodeId, result: Result<(), String>) {
-        // Deliberately not `complete_node_growing(.., self.new_job())`: that
-        // would take the lock twice (once to mint an empty builder, once to
-        // complete) to express "grew nothing". The shared part is the outcome
-        // mapping, and that's a free fn.
-        let mut inner = self.lock();
-        inner.complete(node_id, outcome_of(result));
-        drop(inner);
-        self.notify.notify_one();
+    /// The DAG container id owning `node`, for log lines and the dashboard.
+    /// Derived from the graph rather than carried alongside the node — the
+    /// parent axis already knows it.
+    #[must_use]
+    pub fn dag_of(&self, node: NodeId) -> Option<u64> {
+        self.lock().graph().root_of(node).map(NodeId::get)
     }
 
     /// A builder for a node to declare more work into while it runs.
     ///
-    /// Handed to [`exec::run_node`] and returned to
-    /// [`JobQueue::complete_node_growing`]. Only `hive_jobq` can construct one,
-    /// which is why this goes through the scheduler rather than
-    /// `Job::default()`.
+    /// Handed to [`exec::run_node`] and returned to the crate's completion.
+    /// Only `hive_jobq` can construct one, which is why this goes through the
+    /// scheduler rather than `Job::default()`.
+    ///
+    /// The completion wrappers that used to live beside this (`complete_node`,
+    /// `complete_node_growing`) are **gone**: a node is completed inside the
+    /// future [`hive_jobq::scheduler::Scheduler::claim_next`] hands back, so
+    /// this layer has nothing left to wrap. The tests keep their own extension
+    /// trait for driving completions by hand.
     #[must_use]
     pub fn new_job(&self) -> Job {
         self.lock().new_job()
     }
 
-    /// [`JobQueue::complete_node`] plus the work the node declared while it ran.
-    ///
-    /// `grown` is inserted **under `node_id`** before the completion, so the DAG
-    /// cannot roll terminal with the appended work still pending — the property
-    /// the old two-call `append_subgraph` + `complete_node` sequence had to
-    /// arrange by hand at every call site.
-    pub fn complete_node_growing(&self, node_id: NodeId, result: Result<(), String>, grown: Job) {
-        let mut inner = self.lock();
-        // A rejected grown job is logged, not propagated: the node's own work
-        // already ran, and refusing to complete it here would both misreport
-        // that and wedge the DAG on a node stuck `Running`.
-        if let Err(e) = inner.complete_growing(node_id, outcome_of(result), grown) {
-            tracing::error!(
-                node = node_id.get(),
-                error = %e,
-                "job_queue: work grown by a completing node was rejected"
-            );
-        }
-        drop(inner);
-        self.notify.notify_one();
-    }
-
     /// Cancel a DAG that hasn't started yet: every work node is still `Pending`,
     /// so each is cancelled. `false` once any work node is running or terminal —
     /// an in-flight nix build isn't interruptible.
diff --git a/hive-c0re/src/job_queue/scheduler.rs b/hive-c0re/src/job_queue/scheduler.rs
index d1aec330..33982ae2 100644
--- a/hive-c0re/src/job_queue/scheduler.rs
+++ b/hive-c0re/src/job_queue/scheduler.rs
@@ -17,23 +17,17 @@
 //!
 //! In-DAG growth (a `MetaLock` growing rebuild subgraphs, a `Reconcile` fanning
 //! its `Start`/`Stop`) is declared onto the builder each node is handed, and
-//! inserted as part of completing that node — see `handle_completion`.
+//! inserted as part of completing that node. Completion itself is not this
+//! module's job any more: it happens *inside* the future
+//! [`hive_jobq::scheduler::Scheduler::claim_next`] hands back, so a node that
+//! ran but was never completed is not an expressible state here.
 
 use std::collections::HashMap;
 use std::sync::Arc;
 
 use super::exec;
-use super::{Claim, Job};
 use crate::coordinator::Coordinator;
 
-struct NodeDone {
-    claim: Claim,
-    /// Whatever the node declared into its builder while running — usually
-    /// nothing. Inserted under the node as part of completing it.
-    grown: Job,
-    result: anyhow::Result<()>,
-}
-
 /// Scheduler loop. Spawned once at hive-c0re startup from `main.rs`.
 ///
 /// Shutdown semantics: subscribes to `coord.shutdown_rx()`. On a true signal
@@ -56,7 +50,6 @@ struct NodeDone {
 /// reconverging silently.
 pub async fn run_worker(coord: Arc<Coordinator>) {
     let mut shutdown = coord.shutdown_rx();
-    let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel::<NodeDone>();
     // Last derived pill set we published, keyed by agent (its lease is cap-1,
     // so one pill each). Purely the previous value of a *derived* quantity —
     // it exists to spot transitions, since the dashboard wants edges
@@ -73,33 +66,69 @@ pub async fn run_worker(coord: Arc<Coordinator>) {
             return;
         }
         reconcile_transients(&coord, &mut transients);
-        let claims = coord.job_queue.claim_ready();
-        if !claims.is_empty() {
-            for claim in claims {
-                tracing::info!(
-                    dag = claim.dag_id,
-                    node = claim.node_id.get(),
-                    kind = claim.kind.as_str(),
-                    agent = %claim.agent,
-                    "job_queue: node running"
-                );
-                let coord = Arc::clone(&coord);
-                let tx = tx.clone();
-                tokio::spawn(async move {
-                    // The node's growth channel. Local state, so it costs
-                    // nothing to carry and holds no lock while the node runs.
-                    // The builder is passed by value and handed back: owned it
-                    // is `Send`, a `&Job` held across an await is not.
-                    let job = coord.job_queue.new_job();
-                    let (grown, result) = exec::run_node(&coord, job, &claim).await;
-                    // Send failure = scheduler gone (shutdown); drop.
-                    let _ = tx.send(NodeDone {
-                        claim,
-                        grown,
-                        result,
-                    });
-                });
-            }
+        // Claim exactly one node and get back the work that runs it. `Some`
+        // means something started, so there may be more runnable right now —
+        // loop again immediately. `None` means nothing is runnable and the
+        // loop parks below. That decision is the whole reason the crate hands
+        // back a task rather than an id.
+        let runner = {
+            // Two handles, deliberately: `sched` is the scheduler the crate
+            // locks, `node_coord` is what the node's own future captures. One
+            // binding can't do both — passing `coord.job_queue.sched()` borrows
+            // `coord` for the whole call while the `move` closure wants to take
+            // it.
+            let sched = Arc::clone(coord.job_queue.sched());
+            let node_coord = Arc::clone(&coord);
+            hive_jobq::scheduler::Scheduler::claim_next(&sched, move |id, kind, job| {
+                let coord = node_coord;
+                async move {
+                    tracing::info!(
+                        dag = coord.job_queue.dag_of(id).unwrap_or_default(),
+                        node = id.get(),
+                        kind = kind.as_str(),
+                        agent = %kind.agent(),
+                        "job_queue: node running"
+                    );
+                    let (grown, result) = exec::run_node(&coord, job, id, &kind).await;
+                    match &result {
+                        Ok(()) => tracing::info!(node = id.get(), "job_queue: node done"),
+                        Err(e) => tracing::warn!(
+                            node = id.get(),
+                            kind = kind.as_str(),
+                            agent = %kind.agent(),
+                            error = %format!("{e:#}"),
+                            grown_nodes = !grown.is_empty(),
+                            "job_queue: node failed"
+                        ),
+                    }
+                    let outcome = super::outcome_of(result.map_err(|e| format!("{e:#}")));
+                    // Growth is dropped on failure: a node that declared
+                    // follow-up work and *then* failed does not want it run —
+                    // failure cancel-cascades, so inserting it would only add
+                    // nodes to immediately cancel.
+                    let grown = if matches!(outcome, hive_jobq::scheduler::Outcome::Failed(_)) {
+                        coord.job_queue.new_job()
+                    } else {
+                        grown
+                    };
+                    (grown, outcome)
+                }
+            })
+        };
+        if let Some(runner) = runner {
+            let done_coord = Arc::clone(&coord);
+            tokio::spawn(async move {
+                // Completion happens inside `runner` — it cannot be forgotten
+                // here, which is why there is no completion channel any more.
+                let (id, grew) = runner.await;
+                if let Err(e) = grew {
+                    tracing::warn!(node = id.get(), error = %e, "job_queue: grown job rejected");
+                }
+                done_coord.emit_rebuild_queue_snapshot();
+                // Wake the loop: this node's completion may have unblocked
+                // dependents. Previously the completion channel did this.
+                done_coord.job_queue.notify.notify_one();
+            });
             // Newly-started owner nodes now hold their leases — surface the pills.
             reconcile_transients(&coord, &mut transients);
             coord.emit_rebuild_queue_snapshot();
@@ -113,63 +142,11 @@ pub async fn run_worker(coord: Arc<Coordinator>) {
                     return;
                 }
             }
-            Some(done) = rx.recv() => {
-                handle_completion(&coord, done);
-            }
             () = coord.job_queue.notify.notified() => {}
         }
     }
 }
 
-fn handle_completion(coord: &Arc<Coordinator>, done: NodeDone) {
-    let NodeDone {
-        claim,
-        grown,
-        result,
-    } = done;
-    match result {
-        Ok(()) => {
-            tracing::info!(
-                dag = claim.dag_id,
-                node = claim.node_id.get(),
-                "job_queue: node done"
-            );
-            // Whatever the node declared goes in under it as part of this
-            // completion, so the DAG cannot roll terminal while the appended
-            // work is still pending. Covers the multi-node case (a `MetaLock`
-            // growing per-agent rebuild subgraphs) and the single-node case (a
-            // `Reconcile` planner's `Start` / `Stop`) identically.
-            coord
-                .job_queue
-                .complete_node_growing(claim.node_id, Ok(()), grown);
-        }
-        Err(e) => {
-            let msg = format!("{e:#}");
-            tracing::warn!(
-                dag = claim.dag_id,
-                node = claim.node_id.get(),
-                kind = claim.kind.as_str(),
-                agent = %claim.agent,
-                error = %msg,
-                grown_nodes = !grown.is_empty(),
-                "job_queue: node failed"
-            );
-            // `grown` is deliberately dropped on failure. A node that declared
-            // follow-up work and *then* failed does not want that work run —
-            // failure cancel-cascades downstream, so inserting it would only
-            // add nodes to immediately cancel. This preserves the old shape,
-            // where growth could only be expressed on the success path at all;
-            // the difference is that it is now possible to declare and then
-            // fail, so the drop has to be a decision rather than an accident.
-            drop(grown);
-            coord.job_queue.complete_node(claim.node_id, Err(msg));
-        }
-    }
-    // The next loop iteration re-reconciles the transient pills against the
-    // post-completion lease state (a settled subgraph drops its pill).
-    coord.emit_rebuild_queue_snapshot();
-}
-
 /// Publish the transitions between the previously-derived pill set and the
 /// current one. `prev` is last loop's derived value, keyed by agent (an agent's
 /// lease is cap-1, so at most one pill each).
diff --git a/hive-c0re/src/job_queue/tests.rs b/hive-c0re/src/job_queue/tests.rs
index 802372a2..17e231fe 100644
--- a/hive-c0re/src/job_queue/tests.rs
+++ b/hive-c0re/src/job_queue/tests.rs
@@ -65,8 +65,96 @@ fn stop_online(
     submit::stop_spec(&targets, graceful, Source::Manual, reason.to_owned())
 }
 
+/// What a test observes about a node the settle loop just started: its id, its
+/// DAG, and its payload.
+///
+/// **Test-only, and deliberately not a production type.** `exec::run_node`
+/// takes `(NodeId, &NodeKind)` and derives the DAG id on the two arms that
+/// actually want it — nothing in production needs a claim snapshot to exist.
+/// The assertions here are about *which* node the graph let run, which does
+/// need the payload next to the id.
+#[derive(Debug, Clone)]
+struct Claimed {
+    dag_id: u64,
+    node_id: NodeId,
+    kind: NodeKind,
+    agent: String,
+}
+
+/// Drive one settle wave and report every node that started.
+///
+/// An **extension trait rather than a method on [`JobQueue`]**: production
+/// claims one node at a time ([`hive_jobq::scheduler::Scheduler::claim_next`])
+/// and has no use for a whole wave, so this must not be reachable from
+/// non-test code. `settle()` is that same claim primitive in a loop, so a test
+/// driving it here exercises the production path.
+trait ClaimReady {
+    fn claim_ready(&self) -> Vec<Claimed>;
+}
+
+impl ClaimReady for JobQueue {
+    fn claim_ready(&self) -> Vec<Claimed> {
+        let mut sched = self.sched().lock().expect("job_queue mutex poisoned");
+        let started = sched.settle();
+        started
+            .into_iter()
+            .filter_map(|node_id| {
+                let kind = sched.graph().node(node_id)?.payload.clone();
+                Some(Claimed {
+                    dag_id: sched.graph().root_of(node_id)?.get(),
+                    node_id,
+                    agent: kind.agent().to_owned(),
+                    kind,
+                })
+            })
+            .collect()
+    }
+}
+
+/// Drive a node terminal by hand.
+///
+/// Also an extension trait, for the same reason as [`ClaimReady`]: production
+/// completes a node **inside** the future
+/// [`hive_jobq::scheduler::Scheduler::claim_next`] hands back, so "run the node,
+/// then remember to complete it" is not an expressible sequence there — which
+/// was the whole point of the seam. These tests need to express it, because
+/// they exercise the graph without running any executor.
+trait CompleteNode {
+    fn complete_node(&self, node_id: NodeId, result: Result<(), String>);
+    fn complete_node_growing(&self, node_id: NodeId, result: Result<(), String>, grown: Job);
+}
+
+impl CompleteNode for JobQueue {
+    fn complete_node(&self, node_id: NodeId, result: Result<(), String>) {
+        self.sched()
+            .lock()
+            .expect("job_queue mutex poisoned")
+            .complete(node_id, outcome_of(result));
+        self.notify.notify_one();
+    }
+
+    fn complete_node_growing(&self, node_id: NodeId, result: Result<(), String>, grown: Job) {
+        // A rejected grown job is logged, not propagated: the node's own work
+        // already ran, and refusing to complete it here would both misreport
+        // that and wedge the DAG on a node stuck `Running`.
+        if let Err(e) = self
+            .sched()
+            .lock()
+            .expect("job_queue mutex poisoned")
+            .complete_growing(node_id, outcome_of(result), grown)
+        {
+            tracing::error!(
+                node = node_id.get(),
+                error = %e,
+                "job_queue: work grown by a completing node was rejected"
+            );
+        }
+        self.notify.notify_one();
+    }
+}
+
 /// Claim helper asserting exactly one node comes back.
-fn claim_one(q: &JobQueue) -> Claim {
+fn claim_one(q: &JobQueue) -> Claimed {
     let mut claims = q.claim_ready();
     assert_eq!(
         claims.len(),