wip(#3001): convert tests off the container id; drop Source + insert_group

The last of the DAG-container removal. `tests.rs` navigated by the id
`submit` returned, so removing the container removed the tests' way of
finding what they inserted; they name the roots they assert on now, which
is the same handle production uses.

Three findings the port surfaced, each a behaviour change rather than a
test fix:

- Cancelling a rebuild's head no longer drops the job. `Reconcile`'s edge
  accepts a skipped brace, and a cancel-cascade skips rather than cancels,
  so the tail stays claimable. Dropping a job means cancelling every id the
  insert returned.
- A directly-cancelled group root reads terminal while a spared tail still
  runs; the cancel used to land on a node above it, which rolled up
  Finishing instead.
- "One DAG per hive-wide op" is not expressible without a container. The
  three tests asserting it now assert that every named root is top-level,
  which is what makes the per-agent subgraphs concurrent.

Deletes two tests: one asserted only that two containers get distinct ids,
the other re-ran an existing case under a second name.

`Source`, `insert_group` and the stop path's `reason` string went dead with
the container and are removed with it.

docs/coordinator.md

@@ -106,7 +106,7 @@ subgraph each (independent roots, run concurrently on their own leases), not
 N separate DAGs.
 
 **These are built dynamically from each agent's live running state** (an
-async `lifecycle::is_running` read), so they live in `job_queue/submit.rs`,
+async `lifecycle::is_running` read), so they live in `job_queue/power.rs`,
 not the pure/sync `templates.rs`. Per-agent shape rule: `stop`/`start` carry
 a head `SetWanted` (intent) — `restart` does not; the tail `Reconcile`
 (convergence guarantee — cheap, noops when already converged) is ALWAYS
@@ -161,12 +161,12 @@ Notable collapses:
 Per-agent power *intent* — `wanted: Up | Offline` — is durable as the
 `agent_power` table in the coordinator DB (`hive-c0re/src/stores/power.rs`).
 `container_view` remains the observed *status*; `Reconcile` nodes converge the
-two. Setting `wanted` is never a queued node: the submit layer
-(`job_queue/submit.rs`) writes the row synchronously, then submits the DAG
+two. Setting `wanted` is never a queued node: the power layer
+(`job_queue/power.rs`) writes the row synchronously, then inserts the DAG
 whose `Reconcile` reads the fresh value — rapid toggles are last-writer-wins.
 Power toggles never commit to the meta repo. Every operator power surface —
 dashboard buttons, the MCP tools, and `hivectl stop/start/restart/kill` —
-rides the queue through that submit layer, so intent, lease serialization,
+rides the queue through that power layer, so intent, lease serialization,
 and crash-watch suppression can't drift per surface; the only direct starts
 left are the root-agent bootstrap and infra containers (no lease, no
 harness). Cancelling a still-queued power DAG reverts `wanted` to the

hive-c0re/src/actions.rs

@@ -55,13 +55,12 @@ pub async fn approve(coord: Arc<Coordinator>, id: i64) -> Result<()> {
             // nothing).
             let inputs: Vec<String> =
                 serde_json::from_str(&approval.commit_ref).unwrap_or_default();
-            let submitted = coord.job_queue.submit(
-                crate::job_queue::Source::Approval,
-                format!("approval #{id} meta input update"),
-                |b| crate::job_queue::templates::meta_update(b, inputs, Some(id)),
-            );
-            if let Err(e) = submitted {
-                return Err(e.context("submit meta-update dag"));
+            let inserted = coord.job_queue.insert_job(|b| {
+                crate::job_queue::templates::meta_update(b, inputs, Some(id));
+                Vec::new()
+            });
+            if let Err(e) = inserted {
+                return Err(e.context("insert meta-update dag"));
             }
             coord.emit_rebuild_queue_snapshot();
             Ok(())
@@ -75,13 +74,12 @@ pub async fn approve(coord: Arc<Coordinator>, id: i64) -> Result<()> {
             {
                 tracing::warn!(agent = %approval.agent, error = ?e, "agent_power: seed on spawn failed");
             }
-            let submitted = coord.job_queue.submit(
-                crate::job_queue::Source::Approval,
-                format!("approval #{id} spawn"),
-                |b| crate::job_queue::templates::spawn(b, approval.agent.as_str(), id),
-            );
-            if let Err(e) = submitted {
-                return Err(e.context("submit spawn dag"));
+            let inserted = coord.job_queue.insert_job(|b| {
+                crate::job_queue::templates::spawn(b, approval.agent.as_str(), id);
+                Vec::new()
+            });
+            if let Err(e) = inserted {
+                return Err(e.context("insert spawn dag"));
             }
             coord.emit_rebuild_queue_snapshot();
             Ok(())
@@ -106,12 +104,7 @@ pub async fn approve(coord: Arc<Coordinator>, id: i64) -> Result<()> {
             // `run_deploy_apply` (ff-merge, then grows the rebuild subgraph),
             // `run_finalize_deploy` (deploy tag + lock commit) and
             // `run_deploy_tail` (compensation + forge mirror).
-            enqueue_approval_rebuild(
-                &coord,
-                approval.agent.as_str(),
-                id,
-                format!("approval #{id} merge config pr"),
-            );
+            enqueue_approval_rebuild(&coord, approval.agent.as_str(), id);
             Ok(())
         }
     }
@@ -121,19 +114,12 @@ pub async fn approve(coord: Arc<Coordinator>, id: i64) -> Result<()> {
 /// dispatch arm — the work ends in a container rebuild routed through the
 /// queue. See [`crate::job_queue::templates::approval_deploy`] for the node
 /// shape; the executor dispatches each node to the `run_deploy_*` bodies below.
-fn enqueue_approval_rebuild(
-    coord: &Arc<Coordinator>,
-    agent: &str,
-    approval_id: i64,
-    reason: String,
-) {
-    if let Err(e) = coord
-        .job_queue
-        .submit(crate::job_queue::Source::Approval, reason, |b| {
-            crate::job_queue::templates::approval_deploy(b, agent, approval_id);
-        })
-    {
-        tracing::error!(%agent, approval_id, error = ?e, "submit approval deploy dag failed");
+fn enqueue_approval_rebuild(coord: &Arc<Coordinator>, agent: &str, approval_id: i64) {
+    if let Err(e) = coord.job_queue.insert_job(|b| {
+        crate::job_queue::templates::approval_deploy(b, agent, approval_id);
+        Vec::new()
+    }) {
+        tracing::error!(%agent, approval_id, error = ?e, "insert approval deploy dag failed");
     }
     coord.emit_rebuild_queue_snapshot();
 }

hive-c0re/src/dashboard/lifecycle_ops.rs

@@ -1,11 +1,12 @@
 //! Container lifecycle endpoints for the dashboard.
 //!
 //! Rebuild / restart / start / stop (hard + graceful) / update-all all
-//! submit DAGs to the job queue (`job_queue::submit`), so each shows a
-//! visible queued→running transient on the dashboard — a direct
-//! sub-second start/stop only flashed the badge. Start/stop also
-//! persist the agent's `wanted` power intent before submitting; the
-//! DAG's `Reconcile` converges to it. Destroy delegates to
+//! insert DAGs into the job queue — the power ops via
+//! [`crate::job_queue::power`], the static shapes straight through
+//! `JobQueue::insert` — so each shows a visible queued→running transient on
+//! the dashboard; a direct sub-second start/stop only flashed the badge.
+//! Start/stop also persist the agent's `wanted` power intent before
+//! inserting; the DAG's `Reconcile` converges to it. Destroy delegates to
 //! `actions::destroy` (optionally purging).
 
 use axum::{
@@ -27,7 +28,6 @@ pub(super) struct GracefulParams {
 }
 
 use super::{AppState, Ident, error_response, guard_agent_name, strip_container_prefix};
-use crate::job_queue::{Source, submit};
 use crate::{actions, lifecycle};
 
 /// Queue a rebuild DAG for `name`.
@@ -50,12 +50,13 @@ pub(super) async fn post_rebuild(
     if let Some(reject) = guard_agent_name(&state, &logical).await {
         return reject;
     }
-    submit::rebuild(
-        &state.coord,
-        &logical,
-        Source::Manual,
-        "manual via dashboard ↻ R3BU1LD button".to_owned(),
-    );
+    if let Err(e) = state.coord.job_queue.insert_job(|b| {
+        crate::job_queue::templates::rebuild(b, &logical, true);
+        Vec::new()
+    }) {
+        tracing::error!(agent = %logical, error = ?e, "rebuild: insert failed");
+    }
+    state.coord.emit_rebuild_queue_snapshot();
     (StatusCode::OK, "ok").into_response()
 }
 
@@ -92,13 +93,12 @@ pub(super) async fn post_kill(
         // timeout fallback to a hard stop). The agent's lifecycle
         // lease keeps it from racing an in-flight rebuild for the same
         // agent, and per-node progress surfaces on the queue snapshot.
-        submit::graceful_stop(
-            &state.coord,
-            &logical,
-            Source::Manual,
-            "manual via dashboard graceful stop".to_owned(),
-        )
-        .await;
+        if let Err(e) =
+            crate::job_queue::power::stop_many(&state.coord, std::slice::from_ref(&logical), true)
+                .await
+        {
+            tracing::error!(agent = %logical, error = ?e, "graceful stop: insert failed");
+        }
         return (StatusCode::OK, "ok").into_response();
     }
     // Manager is stoppable from the dashboard like any other
@@ -111,13 +111,12 @@ pub(super) async fn post_kill(
     // `socket_server.rs::Request::Kill` stays in place: a
     // manager calling Kill on its own container is self-suicide
     // mid-call, not a legitimate operator action.
-    submit::stop(
-        &state.coord,
-        &logical,
-        Source::Manual,
-        "manual via dashboard stop".to_owned(),
-    )
-    .await;
+    if let Err(e) =
+        crate::job_queue::power::stop_many(&state.coord, std::slice::from_ref(&logical), false)
+            .await
+    {
+        tracing::error!(agent = %logical, error = ?e, "stop: insert failed");
+    }
     (StatusCode::OK, "ok").into_response()
 }
 
@@ -149,22 +148,23 @@ pub(super) async fn post_restart(
         return reject;
     }
     if params.graceful {
-        submit::graceful_restart(
+        if let Err(e) = crate::job_queue::power::restart_many(
             &state.coord,
-            &logical,
-            Source::Manual,
-            "manual via dashboard graceful restart".to_owned(),
+            std::slice::from_ref(&logical),
+            true,
         )
-        .await;
+        .await
+        {
+            tracing::error!(agent = %logical, error = ?e, "graceful restart: insert failed");
+        }
         return (StatusCode::OK, "ok").into_response();
     }
-    submit::restart(
-        &state.coord,
-        &logical,
-        Source::Manual,
-        "manual via dashboard ↺ R3START button".to_owned(),
-    )
-    .await;
+    if let Err(e) =
+        crate::job_queue::power::restart_many(&state.coord, std::slice::from_ref(&logical), false)
+            .await
+    {
+        tracing::error!(agent = %logical, error = ?e, "restart: insert failed");
+    }
     (StatusCode::OK, "ok").into_response()
 }
 
@@ -226,13 +226,11 @@ pub(super) async fn post_start(
             return (StatusCode::OK, "ok").into_response();
         }
     }
-    submit::start(
-        &state.coord,
-        &logical,
-        Source::Manual,
-        "manual via dashboard start".to_owned(),
-    )
-    .await;
+    if let Err(e) =
+        crate::job_queue::power::start_many(&state.coord, std::slice::from_ref(&logical)).await
+    {
+        tracing::error!(agent = %logical, error = ?e, "start: insert failed");
+    }
     (StatusCode::OK, "ok").into_response()
 }
 
@@ -411,13 +409,14 @@ pub(super) async fn post_update_all(State(state): State<AppState>) -> Response {
         else {
             continue;
         };
-        submit::rebuild(
-            &state.coord,
-            &logical,
-            Source::Manual,
-            "manual via dashboard 🌀 UPDATE ALL".to_owned(),
-        );
+        if let Err(e) = state.coord.job_queue.insert_job(|b| {
+            crate::job_queue::templates::rebuild(b, &logical, true);
+            Vec::new()
+        }) {
+            tracing::error!(agent = %logical, error = ?e, "update-all: insert failed");
+        }
     }
+    state.coord.emit_rebuild_queue_snapshot();
     (StatusCode::OK, "ok").into_response()
 }
 

hive-c0re/src/dashboard/meta_inputs.rs

@@ -208,16 +208,18 @@ pub(super) async fn post_meta_update(
     if inputs.is_empty() {
         return error_response("meta-update: no inputs selected");
     }
-    let inputs_label = inputs.join(", ");
     // Cascade rebuild children fan out from the MetaLock node when the
     // lock bump lands — appended by the scheduler so they build against
     // the post-bump lock, and a failed bump simply fans out nothing.
-    crate::job_queue::submit::meta_update(
-        &state.coord,
-        inputs,
-        crate::job_queue::Source::Manual,
-        format!("meta-update via dashboard ({inputs_label})"),
-    );
+    state
+        .coord
+        .job_queue
+        .insert_job(|b| {
+            crate::job_queue::templates::meta_update(b, inputs, None);
+            Vec::new()
+        })
+        .expect("template-declared shapes are acyclic");
+    state.coord.emit_rebuild_queue_snapshot();
     (StatusCode::OK, "ok").into_response()
 }
 

hive-c0re/src/dashboard/permissions.rs

@@ -155,15 +155,21 @@ pub(super) async fn post_tool_groups(
     // META_LOCK inside the WritePermFile node, so concurrent
     // batch-apply actions for different agents never race on the
     // shared tool-groups.json.
-    crate::job_queue::submit::perm_change(
-        &state.coord,
-        &logical,
-        crate::job_queue::Source::Manual,
-        "tool-group change via permissions UI".to_owned(),
-        crate::job_queue::PermPayload::ToolGroups {
-            groups: body.groups.clone(),
-        },
-    );
+    state
+        .coord
+        .job_queue
+        .insert_job(|b| {
+            crate::job_queue::templates::perm_change(
+                b,
+                &logical,
+                crate::job_queue::PermPayload::ToolGroups {
+                    groups: body.groups.clone(),
+                },
+            );
+            Vec::new()
+        })
+        .expect("template-declared shapes are acyclic");
+    state.coord.emit_rebuild_queue_snapshot();
     tracing::info!(agent = %logical, groups = ?body.groups, "operator: set tool-groups via dashboard");
     Ok((StatusCode::OK, "ok").into_response())
 }
@@ -264,15 +270,21 @@ pub(super) async fn post_capabilities(
     // META_LOCK inside the WritePermFile node, so concurrent
     // batch-apply actions for different agents never race on the
     // shared capabilities.json.
-    crate::job_queue::submit::perm_change(
-        &state.coord,
-        &logical,
-        crate::job_queue::Source::Manual,
-        "capability change via dashboard".to_owned(),
-        crate::job_queue::PermPayload::Capabilities {
-            caps: body.caps.clone(),
-        },
-    );
+    state
+        .coord
+        .job_queue
+        .insert_job(|b| {
+            crate::job_queue::templates::perm_change(
+                b,
+                &logical,
+                crate::job_queue::PermPayload::Capabilities {
+                    caps: body.caps.clone(),
+                },
+            );
+            Vec::new()
+        })
+        .expect("template-declared shapes are acyclic");
+    state.coord.emit_rebuild_queue_snapshot();
     tracing::info!(agent = %logical, caps = ?body.caps, "operator: set capabilities via dashboard");
     Ok((StatusCode::OK, "ok").into_response())
 }
@@ -359,13 +371,19 @@ pub(super) async fn post_permissions(
     }
     // Phase 2 — submit one combined PermChange DAG per affected agent.
     for (logical, groups, caps) in staged {
-        crate::job_queue::submit::perm_change(
-            &state.coord,
-            &logical,
-            crate::job_queue::Source::Manual,
-            "batch permission change via permissions UI".to_owned(),
-            crate::job_queue::PermPayload::Combined { groups, caps },
-        );
+        state
+            .coord
+            .job_queue
+            .insert_job(|b| {
+                crate::job_queue::templates::perm_change(
+                    b,
+                    &logical,
+                    crate::job_queue::PermPayload::Combined { groups, caps },
+                );
+                Vec::new()
+            })
+            .expect("template-declared shapes are acyclic");
+        state.coord.emit_rebuild_queue_snapshot();
         tracing::info!(agent = %logical, "operator: batch perm change via dashboard");
     }
     Ok((StatusCode::OK, "ok").into_response())

hive-c0re/src/dashboard/topology.rs

@@ -21,7 +21,6 @@ use utoipa::ToSchema;
 use problem_details::ProblemDetails;
 
 use super::{AppState, error_problem};
-use crate::job_queue::{Source, submit};
 
 /// `POST /api/topology/set-parent` body. `child` is required.
 /// `new_parent` may be:
@@ -94,12 +93,15 @@ pub(super) async fn post_set_parent(
         new_parent = ?new_parent,
         "operator: set-parent via dashboard"
     );
-    submit::reparent(
-        &state.coord,
-        vec![(child, new_parent)],
-        Source::Manual,
-        "manual set-parent via dashboard".to_owned(),
-    );
+    state
+        .coord
+        .job_queue
+        .insert_job(|b| {
+            crate::job_queue::templates::reparent(b, vec![(child, new_parent)]);
+            Vec::new()
+        })
+        .expect("template-declared shapes are acyclic");
+    state.coord.emit_rebuild_queue_snapshot();
     Ok((StatusCode::OK, "ok").into_response())
 }
 
@@ -150,11 +152,14 @@ pub(super) async fn post_set_parent_bulk(
         .map_err(|e| error_problem(&e))?;
     let names: Vec<&str> = body.iter().map(|e| e.child.as_str()).collect();
     tracing::info!(agents = ?names, "operator: set-parent-bulk via dashboard");
-    submit::reparent(
-        &state.coord,
-        moves,
-        Source::Manual,
-        "manual set-parent-bulk via dashboard".to_owned(),
-    );
+    state
+        .coord
+        .job_queue
+        .insert_job(|b| {
+            crate::job_queue::templates::reparent(b, moves);
+            Vec::new()
+        })
+        .expect("template-declared shapes are acyclic");
+    state.coord.emit_rebuild_queue_snapshot();
     Ok((StatusCode::OK, "ok").into_response())
 }

hive-c0re/src/job_queue/exec.rs

@@ -53,7 +53,7 @@ pub(super) async fn run_node(
     kind: &NodeKind,
 ) -> (super::JobBuilder, Result<()>) {
     // The agent this node targets rides the payload — empty for the agentless
-    // container kinds (`MetaLock`, `Dag`), which never read it.
+    // kinds (`MetaLock`, `Reparent`), which never read it.
     let agent = kind.agent();
     // Every arm is `Result<()>`; the three that grow work declare into `builder`
     // *synchronously*, after their own awaits have finished. Borrowing `&builder`
@@ -115,26 +115,21 @@ pub(super) async fn run_node(
             run_finalize_deploy(coord, *approval_id).await
         }
         NodeKind::DeployTail { approval_id, .. } => {
-            run_deploy_tail(coord, coord.job_queue.dag_of(id), agent, *approval_id).await
+            run_deploy_tail(coord, coord.job_queue.root_of(id), agent, *approval_id).await
         }
         NodeKind::ResolveApproval {
             approval_id,
             outcome,
-        } => run_resolve_approval(coord, coord.job_queue.dag_of(id), *approval_id, *outcome).await,
+        } => run_resolve_approval(coord, coord.job_queue.root_of(id), *approval_id, *outcome).await,
         NodeKind::EmitRebuilt { ok, .. } => {
-            run_emit_rebuilt(coord, agent, coord.job_queue.dag_of(id), *ok).await;
+            run_emit_rebuilt(coord, agent, coord.job_queue.root_of(id), *ok).await;
             Ok(())
         }
         NodeKind::SetWanted { up, .. } => run_set_wanted(coord, agent, *up),
-        // The nodes that carry no work of their own; completing one lets it
-        // reach `Finishing` so the nodes under it start.
-        // - `Dag`: pure grouping container. The DAG's terminal side effect, if
-        //   any, is its own tail node in the graph.
-        // - `DeployWindow` / `AgentWindow`: pure resource holders (braces) —
-        //   what they declare stays held until their subtree settles.
-        NodeKind::Dag { .. } | NodeKind::DeployWindow { .. } | NodeKind::AgentWindow { .. } => {
-            Ok(())
-        }
+        // Braces carry no work of their own; completing one lets it reach
+        // `Finishing` so the nodes under it start. What they declare stays held
+        // until their whole subtree settles.
+        NodeKind::DeployWindow { .. } | NodeKind::AgentWindow { .. } => Ok(()),
     };
     (builder, result)
 }

hive-c0re/src/job_queue/mod.rs

@@ -10,17 +10,16 @@
 //!   carries the agent it targets ([`NodeKind::agent`]); the two resource
 //!   classes are [`resource::Resource`] (`BuildSlot` node-held, `Agent` lease
 //!   subtree-held), declared per node at its construction site;
-//! - a **DAG is a single container node** ([`NodeKind::Dag`], `parent = None`)
-//!   carrying the group's metadata, with the work nodes hung under it as
-//!   its subtree (the **parent axis** groups; `deps` order). So the container's
-//!   `NodeId` is the DAG id, its rolled-up state is the DAG state, and membership
-//!   is a graph walk — there are no host grouping side-tables. The lease is owned
-//!   by a subtree root and borrowed by its descendants (continuity);
-//! - per-DAG terminal work is an ordinary **tail node**
+//! - **a job has no container node.** A template declares its nodes and names
+//!   the roots it wants back; `insert_job` returns those ids. Grouping is the
+//!   parent axis (a root's rolled-up state *is* its subtree's), so membership is
+//!   a graph walk with no host-side side-tables. The lease is owned by a subtree
+//!   root and borrowed by its descendants (continuity);
+//! - terminal work is an ordinary **tail node**
 //!   ([`NodeKind::ResolveApproval`] / [`NodeKind::EmitRebuilt`]) that the builder
-//!   appends in [`templates`], edged onto the DAG's other group roots by the
-//!   outcome it reports. Templates emit one tail per outcome and the graph runs
-//!   exactly one, so nothing branches at runtime.
+//!   appends in [`templates`], edged onto the job's group roots by the outcome it
+//!   reports. Templates emit one tail per outcome and the graph runs exactly one,
+//!   so nothing branches at runtime.
 //!
 //! The queue is runtime-only (no persistence): an empty graph on boot; desired
 //! state is re-derived by the reconcile sweep. A single scheduler task
@@ -29,9 +28,9 @@
 
 pub mod exec;
 pub mod model;
+pub mod power;
 pub mod resource;
 pub mod scheduler;
-pub mod submit;
 pub mod templates;
 #[cfg(test)]
 mod tests;
@@ -46,7 +45,7 @@ use hive_jobq_wire::{GraphNode, GraphWire};
 use tokio::sync::Notify;
 
 pub use hive_jobq::TerminalState;
-pub use model::{NodeKind, PermPayload, Source, State};
+pub use model::{NodeKind, PermPayload, State};
 use resource::Resource;
 
 /// A job under construction: `hive_jobq`'s builder over this queue's payload
@@ -90,12 +89,10 @@ pub struct RunningTransient {
 
 /// The crate scheduler, specialised to this host's node + resource types.
 ///
-/// A **DAG is a single container node** ([`NodeKind::Dag`], `parent = None`)
-/// whose subtree is the DAG's work — so the container's `NodeId` is the DAG id,
-/// its rolled-up state is the DAG state, and there are no grouping side-tables:
-/// membership + meta are graph queries ([`container`] + the `hive_jobq::Graph`
-/// accessors, with the meta read straight off the container's payload). One
-/// shared crate [`Graph`] holds every DAG.
+/// A job is **just its nodes** — no container, no grouping side-tables. A
+/// root's rolled-up state is its subtree's, so membership is a graph walk and
+/// "which job is this node in" is [`JobQueue::root_of`]. One shared crate
+/// [`Graph`] holds every job's nodes.
 ///
 /// There is deliberately **no wrapper struct and no per-node side map**. The
 /// last map held the `build_logs` row id; that link now lives on the log row
@@ -140,35 +137,12 @@ fn outcome_of(result: Result<(), String>) -> Outcome {
     }
 }
 
-/// Insert a declared `job` into the shared graph, returning the inserted ids.
-///
-/// A node that declared no parent hangs under `group_parent` — the DAG
-/// container for a template, the emitting node for a runtime-appended
-/// subgraph. Templates declare the parent axis + sibling ordering directly, so
-/// there is no dep-on-root to drop and no lease to hoist: each node declares
-/// its own resources, and the crate's borrow model keeps a resource continuous
-/// across a subtree (a root owns it, descendants borrow it). Independent group
-/// roots carry no cross-links, so a multi-agent DAG's per-agent subgraphs run
-/// concurrently, each on its own lease.
-///
-/// # Errors
-/// Propagates a crate graph-insert error (malformed dep/parent / dep-scope).
-fn insert_group(
-    inner: &mut Sched,
-    declare: impl FnOnce(&JobBuilder),
-    group_parent: Option<NodeId>,
-) -> anyhow::Result<()> {
-    inner
-        .insert_job(group_parent, |b| {
-            declare(b);
-            // c0re names no handles: a DAG is addressed by its container node,
-            // which `submit` inserts itself, and nothing downstream looks an
-            // individual step up by id.
-            Vec::new()
-        })
-        .map_err(|e| anyhow::anyhow!("job_queue: graph insert failed: {e}"))?;
-    Ok(())
-}
+// `insert_group` lived here: a `group_parent`-taking insert whose only
+// remaining caller was the DAG container, everything under it. Runtime growth
+// never went through it — an executor declares into the builder `hive_jobq`
+// hands it, which parents the new work under the emitting node by
+// construction. With no container to be the other kind of parent, the
+// distinction it existed to express is gone.
 
 impl JobQueue {
     #[must_use]
@@ -188,40 +162,34 @@ impl JobQueue {
         self.sched.lock().expect("job_queue mutex poisoned")
     }
 
-    /// Submit a DAG: insert a [`NodeKind::Dag`] **container node** carrying the
-    /// group's metadata, then insert the template's nodes as its subtree (their
-    /// roots re-parented to the container). Returns the container's id as the
-    /// DAG id — its rolled-up state is the DAG state.
+    /// Insert a job's nodes into the shared graph, then wake the run loop.
     ///
-    /// The container is an ordinary node: it declares no resources, so the
-    /// scheduler claims it on the next pass, runs its (empty) logic and parks
-    /// it in `Finishing`, at which point its children become runnable. Nothing
-    /// here completes it by hand — a node with no work of its own still goes
-    /// the way every other node goes.
+    /// Deliberately named for the [`hive_jobq`] primitive it wraps, because
+    /// that is nearly all it is. **The wrapper earns its place on the wake**:
+    /// the crate is sync and runtime-free — it holds no `Notify` at all — so
+    /// the channel the run loop parks on belongs to the host, and something has
+    /// to ping it. Left to call sites, an insert whose ping was forgotten would
+    /// leave a correct DAG sitting unscheduled until an unrelated event
+    /// happened along; nothing would fail, and no test in isolation would see
+    /// it.
     ///
-    /// `source` and `reason` are the container node's own payload — they are
-    /// arguments here rather than fields of a spec struct because that is all
-    /// they ever were. `declare` is the recipe, taken by generic and run
-    /// against a builder `hive_jobq` owns: it goes from the template straight
-    /// into this call, so there is nothing to allocate for.
+    /// Returns exactly what the primitive returns: the ids of the nodes the
+    /// template named, in the order it named them.
     ///
     /// # Errors
     /// Propagates a graph-insert error (dependencies that aren't
     /// dependency-topological).
-    pub fn submit(
+    pub fn insert_job(
         &self,
-        source: Source,
-        reason: String,
-        declare: impl FnOnce(&JobBuilder),
-    ) -> anyhow::Result<u64> {
+        declare: impl FnOnce(&JobBuilder) -> Vec<hive_jobq::NodeGuid>,
+    ) -> anyhow::Result<Vec<NodeId>> {
         let mut inner = self.lock();
-        let container = inner
-            .append(NodeKind::Dag { source, reason }, Vec::new(), None)
-            .map_err(|e| anyhow::anyhow!("job_queue: container insert failed: {e}"))?;
-        insert_group(&mut inner, declare, Some(container))?;
+        let named = inner
+            .insert_job(None, declare)
+            .map_err(|e| anyhow::anyhow!("job_queue: graph insert failed: {e}"))?;
         drop(inner);
         self.notify.notify_one();
-        Ok(container.get())
+        Ok(named)
     }
 
     /// The scheduler itself, for `hive_jobq`'s run-loop seam
@@ -235,11 +203,16 @@ impl JobQueue {
         &self.sched
     }
 
-    /// 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.
+    /// The id of the **group root** `node` belongs to, for log lines and the
+    /// dashboard. Derived from the graph rather than carried alongside the node
+    /// — the parent axis already knows it.
+    ///
+    /// Was `dag_of`, when a job's nodes hung under a container node that *was*
+    /// the group. Without it the parent chain ends at whichever root the
+    /// template declared, so this answers "which root owns this node", not
+    /// "which DAG is this in" — there is no longer such a thing.
     #[must_use]
-    pub fn dag_of(&self, node: NodeId) -> Option<u64> {
+    pub fn root_of(&self, node: NodeId) -> Option<u64> {
         self.lock().graph().root_of(node).map(NodeId::get)
     }
 
@@ -424,9 +397,9 @@ fn find_node(sched: &Sched, id: u64) -> Option<NodeId> {
 /// root, plus the newest [`MAX_HISTORY_DAGS`] settled ones.
 ///
 /// Selected *structurally* — a root is a node with no parent. The typed
-/// projection this replaced keyed on `NodeKind::Dag` instead, which made the
-/// visible set depend on one host node kind; nothing here knows what a node
-/// means.
+/// projection this replaced keyed on the since-removed container kind instead,
+/// which made the visible set depend on one host node kind; nothing here knows
+/// what a node means.
 ///
 /// **This bound is load-bearing, not tidiness.** Nothing ever removes a node
 /// from the graph (bounded pruning is a Stage-C follow-up), so serving

hive-c0re/src/job_queue/model.rs

@@ -1,18 +1,18 @@
-//! Data model for the generic job-DAG queue: node kinds (the primitive
-//! operations), dependency edges, and the runtime `Dag` / `Node` store.
-//! The `Source` / `State` / `PermPayload` wire enums live in
-//! `hive_host_sock::jobs` (they travel on the host admin socket) and are
-//! re-exported here for the queue's internal use. The graph itself is
-//! served through `hive_jobq_wire`'s generic projection — there is no
-//! second, typed view of it any more.
+//! Data model for the generic job-DAG queue: the node kinds — the primitive
+//! operations — and what each one carries. The `State` / `PermPayload` wire
+//! enums live in `hive_host_sock::jobs` (they travel on the host admin
+//! socket) and are re-exported here for the queue's internal use. The graph
+//! itself is served through `hive_jobq_wire`'s generic projection — there is
+//! no second, typed view of it any more.
 //!
-//! Two levels: the **DAG** is the unit of cancel / approval-resolution
-//! and the dashboard group; the **node** is the unit of scheduling /
-//! execution / build-log, and carries its own `agent` (a
-//! DAG can span agents). See `docs/coordinator.md::Job queue` for the
-//! full design.
+//! **One level, not two.** The node is the unit of everything: scheduling,
+//! execution, build-log, cancel, and the dashboard group (a group root's
+//! subtree *is* the group). A DAG used to be a second level above it, with
+//! its own store and its own id; there is no container node any more, so a
+//! job is exactly the nodes it declared. See `docs/coordinator.md::Job queue`
+//! for the full design.
 
-pub use hive_host_sock::jobs::{PermPayload, Source, State};
+pub use hive_host_sock::jobs::{PermPayload, State};
 use serde::Serialize;
 
 use hive_jobq::TerminalState;
@@ -280,18 +280,6 @@ pub enum NodeKind {
     /// `Prebuild`, but that's a no-op there — the agent is down, so prebuild
     /// is skipped.)
     SetWanted { agent: String, up: bool },
-    /// The **DAG container** node: one per submitted DAG, carrying the group's
-    /// domain metadata. Every node hangs *under* it (its subtree), so
-    /// the container's `NodeId` **is** the DAG id and its rolled-up state **is**
-    /// the DAG state. Pure grouping — lease- and
-    /// build-slot-exempt; the executor instant-completes it (`Done`) so it
-    /// reaches `Finishing` and its children start.
-    ///
-    /// No `created_at` here: the graph stamps [`hive_jobq::Node::created_at`] on
-    /// every node at insert, so the container already has one. A second copy in
-    /// the payload would be the same instant recorded twice, with only this
-    /// variant's version reachable to a generic viewer.
-    Dag { source: Source, reason: String },
 }
 
 /// How a hive-c0re node describes itself to a generic graph viewer.
@@ -362,14 +350,13 @@ impl NodeKind {
             NodeKind::ResolveApproval { .. } => "resolve_approval",
             NodeKind::EmitRebuilt { .. } => "emit_rebuilt",
             NodeKind::SetWanted { .. } => "set_wanted",
-            NodeKind::Dag { .. } => "dag",
         }
     }
 
     /// The agent this node targets, or `""` for agentless kinds
     /// ([`NodeKind::MetaLock`] on the `hyperhive` pseudo-agent,
-    /// [`NodeKind::Reparent`] which can span multiple agents, and the
-    /// [`NodeKind::Dag`] container).
+    /// [`NodeKind::Reparent`] which can span multiple agents, and
+    /// [`NodeKind::ResolveApproval`] which acts on an approval row).
     #[must_use]
     pub fn agent(&self) -> &str {
         match self {
@@ -397,8 +384,7 @@ impl NodeKind {
             | NodeKind::SetWanted { agent, .. } => agent,
             NodeKind::MetaLock { .. }
             | NodeKind::Reparent { .. }
-            | NodeKind::ResolveApproval { .. }
-            | NodeKind::Dag { .. } => "",
+            | NodeKind::ResolveApproval { .. } => "",
         }
     }
 

hive-c0re/src/job_queue/power.rs

@@ -1,59 +1,27 @@
-//! Request-level submit API — the surface the dashboard POST handlers,
-//! the MCP socket handlers, and `hivectl` paths call.
+//! Power ops (`stop` / `start` / `restart`) — the DAG shapes whose per-agent
+//! form depends on **live** container state, so they cannot be static
+//! [`super::templates`] entries.
 //!
-//! The **power ops** (`stop` / `start` / `restart`) are built here, not in
-//! `templates.rs`: each agent's subgraph shape depends on its *live* running
-//! state, which needs an async `lifecycle::is_running` read that a pure/sync
-//! template can't do. So these fns are async — they read each agent's state,
-//! assemble a per-agent subgraph out of the shared pure primitives
-//! (`JobBuilder::node` + `templates::rebuild_nodes`), all declaring into ONE job
-//! (independent per-agent roots, concurrent on their own leases).
+//! The split is the purity line, not the subject matter: the `*_chain` /
+//! `*_nodes` builders below are pure (they take `running` / `stale` as
+//! parameters, which is what keeps them unit-testable without a container),
+//! and only the `*_many` entry points do the async `lifecycle::is_running`
+//! read that produces those parameters.
 //!
-//! Dynamic shape rule: `stop`/`start` carry a head `SetWanted(w)` (durable
-//! intent write) — `restart` does NOT (it bounces the container but leaves
-//! `wanted` untouched, so a deliberately-stopped agent isn't forced up). The
-//! tail `Reconcile` (the convergence guarantee — cheap, noops when already
-//! converged) is ALWAYS present; only the *mechanical* nodes
-//! (`Signal`/`Drain`/`StopForUpdate`) are state-conditional (skipped for a
-//! down agent — nothing to quiesce/stop). Keeping `Reconcile` in every shape
-//! closes the TOCTOU window: if an agent flips state between the `is_running`
-//! read and node execution, the tail `Reconcile` still converges it in-DAG,
-//! with `StopForUpdate`-noop as the backstop — no reliance on an external
-//! reconcile sweep. Every helper emits a fresh queue snapshot so the
-//! dashboard shows the new DAG immediately.
+//! Each entry point declares its group and inserts it. There is no metadata
+//! parameter and no container node: attribution is not something every caller
+//! has to invent, and a DAG is addressed by the nodes a template names.
 
 use std::sync::Arc;
 
-use super::model::NodeKind;
+use hive_jobq::NodeId;
+
 use super::resource::Resource;
 use super::templates::rebuild_nodes;
-use super::{JobBuilder, Source, templates};
+use super::{JobBuilder, NodeKind};
 use crate::coordinator::Coordinator;
 use crate::lifecycle;
 
-fn submit_and_emit(
-    coord: &Arc<Coordinator>,
-    source: Source,
-    reason: String,
-    declare: impl FnOnce(&JobBuilder),
-) -> u64 {
-    let id = coord
-        .job_queue
-        .submit(source, reason, declare)
-        .expect("template-declared shapes are acyclic");
-    coord.emit_rebuild_queue_snapshot();
-    id
-}
-
-/// Manual/approval-independent rebuild (always relocks the agent's
-/// meta input — the meta-update cascade grows its own rebuild subgraphs
-/// in-DAG instead of going through this surface).
-pub fn rebuild(coord: &Arc<Coordinator>, agent: &str, source: Source, reason: String) -> u64 {
-    submit_and_emit(coord, source, reason, |builder| {
-        templates::rebuild(builder, agent, true);
-    })
-}
-
 // ---- dynamic power-op DAG assembly ----------------------------------------
 //
 // The pure per-agent chain builders below take `running` (and `stale`)
@@ -67,7 +35,16 @@ pub fn rebuild(coord: &Arc<Coordinator>, agent: &str, source: Source, reason: St
 /// actually running (nothing to drain on a down container). The `Reconcile`
 /// stays even for a down agent so a race-up between the state read and exec
 /// is still stopped in-DAG.
-fn stop_chain(builder: &JobBuilder, agent: &str, graceful: bool, running: bool) {
+/// Returns the group root's guid, which is what the caller names so
+/// `insert_job` hands its id back — that id is how `hivectl` polls this agent's
+/// progress. A chain that returned nothing would insert correctly and leave the
+/// caller with nothing to wait on.
+fn stop_chain(
+    builder: &JobBuilder,
+    agent: &str,
+    graceful: bool,
+    running: bool,
+) -> hive_jobq::NodeGuid {
     // `SetWanted` is the group root and owns the agent lease; the mechanical
     // steps are its children (borrow the lease, run once it reaches `Finishing`,
     // dep-ordered among themselves).
@@ -96,13 +73,26 @@ fn stop_chain(builder: &JobBuilder, agent: &str, graceful: bool, running: bool)
             .needs(Resource::Agent(a()))
             .part_of(wanted);
     }
+    wanted.guid()
 }
 
 /// One agent's **start** subgraph. `SetWanted(Up)` head; a down + stale-rev
 /// agent gets the rebuild subgraph (its tail `Reconcile` starts it on
 /// current derivations), otherwise a plain `Reconcile` (which starts a down
 /// agent and noops an already-running one).
-fn start_chain(builder: &JobBuilder, agent: &str, running: bool, stale: bool) {
+///
+/// Returns **every** group root — see [`stop_chain`] for why they are named.
+///
+/// ⚠️ More than one in the stale branch: `rebuild_nodes` chains its roots
+/// *behind* `SetWanted` with `after_ok`, it does **not** nest them under it. So
+/// `SetWanted` rolls up only itself, and naming it alone would report the whole
+/// start finished while the rebuild was still running.
+fn start_chain(
+    builder: &JobBuilder,
+    agent: &str,
+    running: bool,
+    stale: bool,
+) -> Vec<hive_jobq::NodeGuid> {
     let wanted = builder
         .node(NodeKind::SetWanted {
             agent: agent.to_owned(),
@@ -110,10 +100,16 @@ fn start_chain(builder: &JobBuilder, agent: &str, running: bool, stale: bool) {
         })
         .needs(Resource::Agent(agent.to_owned()));
     if !running && stale {
-        // Rebuild subtree chained behind the `SetWanted` head. `MetaSync`,
-        // `Prebuild` + `Reconcile` are their own group roots (top-level, per
-        // `rebuild_nodes`).
-        rebuild_nodes(builder, agent, true, Some(wanted));
+        // Rebuild subtree chained behind the `SetWanted` head. `MetaSync`, the
+        // `AgentWindow` brace and `Reconcile` are their own group roots
+        // (top-level, per `rebuild_nodes`) — hence all four names.
+        let roots = rebuild_nodes(builder, agent, true, Some(wanted));
+        vec![
+            wanted.guid(),
+            roots.meta_sync.guid(),
+            roots.agent_window.guid(),
+            roots.reconcile.guid(),
+        ]
     } else {
         let _ = builder
             .node(NodeKind::Reconcile {
@@ -121,6 +117,7 @@ fn start_chain(builder: &JobBuilder, agent: &str, running: bool, stale: bool) {
             })
             .needs(Resource::Agent(agent.to_owned()))
             .part_of(wanted);
+        vec![wanted.guid()]
     }
 }
 
@@ -133,14 +130,22 @@ fn start_chain(builder: &JobBuilder, agent: &str, running: bool, stale: bool) {
 /// before `Reconcile`; a down agent gets just `Reconcile`, which
 /// converges to intent — a stopped (`wanted = Off`) agent stays stopped,
 /// a crashed (`wanted = Up`) agent comes back up.
-fn restart_chain(builder: &JobBuilder, agent: &str, graceful: bool, running: bool) {
+///
+/// Returns the group root's guid — see [`stop_chain`].
+fn restart_chain(
+    builder: &JobBuilder,
+    agent: &str,
+    graceful: bool,
+    running: bool,
+) -> hive_jobq::NodeGuid {
     let a = || agent.to_owned();
     if !running {
-        // Nothing to bounce — a lone Reconcile converges to intent.
-        let _ = builder
+        // Nothing to bounce — a lone Reconcile converges to intent, and is
+        // itself the root.
+        return builder
             .node(NodeKind::Reconcile { agent: a() })
-            .needs(Resource::Agent(a()));
-        return;
+            .needs(Resource::Agent(a()))
+            .guid();
     }
     // Running: mechanical stop then Reconcile. The first stop node is the group
     // ROOT (no SetWanted head) and owns the agent lease; the rest are its
@@ -176,6 +181,7 @@ fn restart_chain(builder: &JobBuilder, agent: &str, graceful: bool, running: boo
             .needs(Resource::Agent(a()))
             .part_of(signal)
             .after_ok(stop);
+        signal.guid()
     } else {
         let stop = builder
             .node(NodeKind::StopForUpdate { agent: a() })
@@ -184,6 +190,7 @@ fn restart_chain(builder: &JobBuilder, agent: &str, graceful: bool, running: boo
             .node(NodeKind::Reconcile { agent: a() })
             .needs(Resource::Agent(a()))
             .part_of(stop);
+        stop.guid()
     }
 }
 
@@ -201,10 +208,15 @@ fn restart_chain(builder: &JobBuilder, agent: &str, graceful: bool, running: boo
 // subgraph's indices onto another's used to be a function.
 
 /// Declare the stop DAG from explicit `(agent, running)` targets.
-pub(crate) fn stop_nodes(builder: &JobBuilder, targets: &[(String, bool)], graceful: bool) {
-    for (agent, running) in targets {
-        stop_chain(builder, agent, graceful, *running);
-    }
+pub(crate) fn stop_nodes(
+    builder: &JobBuilder,
+    targets: &[(String, bool)],
+    graceful: bool,
+) -> Vec<hive_jobq::NodeGuid> {
+    targets
+        .iter()
+        .map(|(agent, running)| stop_chain(builder, agent, graceful, *running))
+        .collect()
 }
 
 /// Assemble the start DAG from explicit `(agent, running, stale)` targets.
@@ -213,76 +225,68 @@ pub(crate) fn stop_nodes(builder: &JobBuilder, targets: &[(String, bool)], grace
 /// running under its lease, so a down+stale agent that grew a rebuild subgraph
 /// reports `rebuilding` during its swap and `starting` at its reconcile,
 /// without the DAG having to guess one label covering every target.
-pub(crate) fn start_nodes(builder: &JobBuilder, targets: &[(String, bool, bool)]) {
-    for (agent, running, stale) in targets {
-        start_chain(builder, agent, *running, *stale);
-    }
+pub(crate) fn start_nodes(
+    builder: &JobBuilder,
+    targets: &[(String, bool, bool)],
+) -> Vec<hive_jobq::NodeGuid> {
+    targets
+        .iter()
+        .flat_map(|(agent, running, stale)| start_chain(builder, agent, *running, *stale))
+        .collect()
 }
 
 /// Declare the restart DAG from explicit `(agent, running)` targets.
-pub(crate) fn restart_nodes(builder: &JobBuilder, targets: &[(String, bool)], graceful: bool) {
-    for (agent, running) in targets {
-        restart_chain(builder, agent, graceful, *running);
-    }
+pub(crate) fn restart_nodes(
+    builder: &JobBuilder,
+    targets: &[(String, bool)],
+    graceful: bool,
+) -> Vec<hive_jobq::NodeGuid> {
+    targets
+        .iter()
+        .map(|(agent, running)| restart_chain(builder, agent, graceful, *running))
+        .collect()
 }
 
-/// Restart a single agent. Thin wrapper over [`restart_many`].
-pub async fn restart(coord: &Arc<Coordinator>, agent: &str, source: Source, reason: String) -> u64 {
-    restart_many(coord, &[agent.to_owned()], false, source, reason).await
-}
+// ---- entry points ---------------------------------------------------------
+//
+// Each reads the live state its shape depends on, then declares + inserts.
 
-/// Graceful restart of a single agent (signal → drain → stop → reconcile,
-/// when running). Thin wrapper over [`restart_many`] with `graceful = true`.
-pub async fn graceful_restart(
-    coord: &Arc<Coordinator>,
-    agent: &str,
-    source: Source,
-    reason: String,
-) -> u64 {
-    restart_many(coord, &[agent.to_owned()], true, source, reason).await
-}
-
-/// Restart `agents` (one or many) in a **single** DAG — one per-agent
-/// subgraph each, built dynamically from live running state and run
-/// concurrently on their own leases. A running agent gets the stop→reconcile
-/// chain (`graceful` prepends signal→drain); a down agent gets just a lone
-/// `Reconcile` (nothing to stop). Restart never writes `wanted`, so the
+/// Restart `agents` in a **single** DAG — one per-agent subgraph each, built
+/// from live running state and run concurrently on their own leases. A running
+/// agent gets the stop→reconcile chain (`graceful` prepends signal→drain); a
+/// down agent gets a lone `Reconcile`. Restart never writes `wanted`, so the
 /// tail `Reconcile` converges each agent to its EXISTING intent — a
-/// deliberately-stopped agent stays down. The whole hive-wide
-/// `hivectl restart` is one DAG.
+/// deliberately-stopped agent stays down.
+///
+/// # Errors
+/// Propagates a graph-insert error.
 pub async fn restart_many(
     coord: &Arc<Coordinator>,
     agents: &[String],
     graceful: bool,
-    source: Source,
-    reason: String,
-) -> u64 {
+) -> anyhow::Result<Vec<NodeId>> {
     let mut targets = Vec::with_capacity(agents.len());
     for agent in agents {
         targets.push((agent.clone(), lifecycle::is_running(agent).await));
     }
-    submit_and_emit(coord, source, reason, |builder| {
-        restart_nodes(builder, &targets, graceful);
-    })
+    let ids = coord
+        .job_queue
+        .insert_job(|b| restart_nodes(b, &targets, graceful))?;
+    coord.emit_rebuild_queue_snapshot();
+    Ok(ids)
 }
 
-/// Start a single agent. Thin wrapper over [`start_many`].
-pub async fn start(coord: &Arc<Coordinator>, agent: &str, source: Source, reason: String) -> u64 {
-    start_many(coord, &[agent.to_owned()], source, reason).await
-}
-
-/// Start `agents` (one or many) in a **single** DAG — one per-agent subgraph
-/// each, built dynamically from live state and run concurrently on their own
-/// leases. A down agent gets `SetWanted(Up) → Reconcile` (or, rev stale, a
-/// rebuild-then-start so it comes up on current derivations); an already-
-/// running agent gets `SetWanted(Up) → Reconcile` (the reconcile noops). The
-/// whole hive-wide `hivectl start` is one DAG.
+/// Start `agents` in a **single** DAG. A down agent gets
+/// `SetWanted(Up) → Reconcile` (or, rev stale, a rebuild-then-start so it comes
+/// up on current derivations); an already-running agent gets the same shape
+/// with the reconcile noop'ing.
+///
+/// # Errors
+/// Propagates a graph-insert error.
 pub async fn start_many(
     coord: &Arc<Coordinator>,
     agents: &[String],
-    source: Source,
-    reason: String,
-) -> u64 {
+) -> anyhow::Result<Vec<NodeId>> {
     let current = crate::auto_update::current_flake_rev(&coord.hyperhive_flake);
     let mut targets = Vec::with_capacity(agents.len());
     for agent in agents {
@@ -296,88 +300,30 @@ pub async fn start_many(
         }
         targets.push((agent.clone(), running, stale));
     }
-    submit_and_emit(coord, source, reason, |builder| {
-        start_nodes(builder, &targets);
-    })
+    let ids = coord.job_queue.insert_job(|b| start_nodes(b, &targets))?;
+    coord.emit_rebuild_queue_snapshot();
+    Ok(ids)
 }
 
-/// Hard stop a single agent. Thin wrapper over [`stop_many`].
-pub async fn stop(coord: &Arc<Coordinator>, agent: &str, source: Source, reason: String) -> u64 {
-    stop_many(coord, &[agent.to_owned()], false, source, reason).await
-}
-
-/// Graceful stop of a single agent (signal → drain → reconcile, when
-/// running). Thin wrapper over [`stop_many`] with `graceful = true`.
-pub async fn graceful_stop(
-    coord: &Arc<Coordinator>,
-    agent: &str,
-    source: Source,
-    reason: String,
-) -> u64 {
-    stop_many(coord, &[agent.to_owned()], true, source, reason).await
-}
-
-/// Stop `agents` (one or many) in a **single** DAG — one per-agent subgraph
-/// each, built dynamically from live state and run concurrently on their own
-/// leases. A running agent gets `SetWanted(Off) → [Signal → Drain →](graceful)
-/// Reconcile`; a down agent gets just `SetWanted(Off) → Reconcile` (skips the
-/// pointless quiesce, keeps the Reconcile as the race-up backstop). The whole
-/// hive-wide `hivectl stop` is one DAG.
+/// Stop `agents` in a **single** DAG. A running agent gets
+/// `SetWanted(Off) → [Signal → Drain →](graceful) Reconcile`; a down agent
+/// skips the pointless quiesce but keeps the `Reconcile` as the race-up
+/// backstop.
+///
+/// # Errors
+/// Propagates a graph-insert error.
 pub async fn stop_many(
     coord: &Arc<Coordinator>,
     agents: &[String],
     graceful: bool,
-    source: Source,
-    reason: String,
-) -> u64 {
+) -> anyhow::Result<Vec<NodeId>> {
     let mut targets = Vec::with_capacity(agents.len());
     for agent in agents {
         targets.push((agent.clone(), lifecycle::is_running(agent).await));
     }
-    submit_and_emit(coord, source, reason, |builder| {
-        stop_nodes(builder, &targets, graceful);
-    })
-}
-
-/// Perm change: commit the JSON file(s) then rebuild.
-pub fn perm_change(
-    coord: &Arc<Coordinator>,
-    agent: &str,
-    source: Source,
-    reason: String,
-    payload: super::PermPayload,
-) -> u64 {
-    submit_and_emit(coord, source, reason, |builder| {
-        templates::perm_change(builder, agent, payload);
-    })
-}
-
-/// Meta-input lock bump; cascade rebuilds fan out on completion.
-pub fn meta_update(
-    coord: &Arc<Coordinator>,
-    inputs: Vec<String>,
-    source: Source,
-    reason: String,
-) -> u64 {
-    submit_and_emit(coord, source, reason, |builder| {
-        templates::meta_update(builder, inputs, None);
-    })
-}
-
-/// Topology move(s) as a queue DAG. `moves` is `(child, new_parent)` pairs —
-/// one entry for `set-parent`, N for `set-parent-bulk`. Fire-and-forget like
-/// everything else in this module: submits and returns a DAG id
-/// immediately, the caller learns the outcome async (dashboard job view /
-/// `hivectl`'s `QueueDag` poll). Wired from `server.rs`'s `HostRequest::
-/// SetParent` (hivectl) and `dashboard/topology.rs`'s `set-parent`/
-/// `set-parent-bulk` handlers.
-pub fn reparent(
-    coord: &Arc<Coordinator>,
-    moves: Vec<(hive_types::Ident, Option<hive_types::Ident>)>,
-    source: Source,
-    reason: String,
-) -> u64 {
-    submit_and_emit(coord, source, reason, |builder| {
-        templates::reparent(builder, moves);
-    })
+    let ids = coord
+        .job_queue
+        .insert_job(|b| stop_nodes(b, &targets, graceful))?;
+    coord.emit_rebuild_queue_snapshot();
+    Ok(ids)
 }

hive-c0re/src/job_queue/scheduler.rs

@@ -93,7 +93,7 @@ pub async fn run_worker(coord: Arc<Coordinator>) {
                 let coord = node_coord;
                 async move {
                     tracing::info!(
-                        dag = coord.job_queue.dag_of(id).unwrap_or_default(),
+                        dag = coord.job_queue.root_of(id).unwrap_or_default(),
                         node = id.get(),
                         kind = kind.as_str(),
                         agent = %kind.agent(),

hive-c0re/src/job_queue/templates.rs

@@ -17,8 +17,8 @@
 //!
 //! The hive-wide **power ops** (`stop` / `start` / `restart`) are NOT here:
 //! their per-agent shape depends on live running state (an async
-//! `lifecycle::is_running` read), so `submit.rs` assembles them out of the
-//! primitives this module exports ([`rebuild_nodes`]).
+//! `lifecycle::is_running` read), so [`super::power`] assembles them out of
+//! the primitives this module exports ([`rebuild_nodes`]).
 
 use hive_jobq::TerminalState;
 
@@ -317,13 +317,22 @@ pub(crate) fn deploy_rebuild_nodes(builder: &JobBuilder, agent: &str, approval_i
 /// children.
 ///
 /// Closed by an [`NodeKind::EmitRebuilt`] tail edged onto all three group-roots
-/// (`MetaSync`, `Prebuild`, `Reconcile`) — `Prebuild`'s roll-up carries the
-/// whole `StopForUpdate`→`Swap`→`RebuildBookkeeping` subtree, so those three cover every
-/// node. Edging `Reconcile` alone would not do: it is `AfterAny` `Prebuild`, so
-/// it reaches `Done` even after a failed swap and the tail would report success.
-pub fn rebuild(builder: &JobBuilder, agent: &str, relock: bool) {
+/// (`MetaSync`, the `AgentWindow` brace, `Reconcile`) — the brace's roll-up
+/// carries the whole `StopForUpdate`→`Swap`→`RebuildBookkeeping` subtree, so
+/// those three cover every node. Edging `Reconcile` alone would not do: it is
+/// `AfterAny` the brace, so it reaches `Done` even after a failed swap and the
+/// tail would report success.
+///
+/// Returns those three roots, so a caller that needs to wait on the rebuild can
+/// name them.
+pub fn rebuild(builder: &JobBuilder, agent: &str, relock: bool) -> Vec<hive_jobq::NodeGuid> {
     let roots = rebuild_nodes(builder, agent, relock, None);
     emit_rebuilt_tails(builder, agent, &roots.all());
+    vec![
+        roots.meta_sync.guid(),
+        roots.agent_window.guid(),
+        roots.reconcile.guid(),
+    ]
 }
 
 /// Approval-driven deploy (`MergeConfigPr`) as a phase subtree rather than the
@@ -483,10 +492,16 @@ pub fn meta_update(builder: &JobBuilder, inputs: Vec<String>, approval_id: Optio
 /// checks), so a parent move needs no container rebuild to take effect.
 /// No transient pill either — the node is agentless (no lease to hang one
 /// off of) and near-instant. No tail node: the write is the whole effect.
-pub fn reparent(builder: &JobBuilder, moves: Vec<(hive_types::Ident, Option<hive_types::Ident>)>) {
-    let _reparent = builder
+///
+/// Returns the single node's guid so a caller can wait on it.
+pub fn reparent(
+    builder: &JobBuilder,
+    moves: Vec<(hive_types::Ident, Option<hive_types::Ident>)>,
+) -> hive_jobq::NodeGuid {
+    builder
         .node(NodeKind::Reparent { moves })
-        .needs(Resource::MetaWindow);
+        .needs(Resource::MetaWindow)
+        .guid()
 }
 
 // The boot is assembled inline in `workers/auto_update.rs::submit_boot_tree`

hive-c0re/src/server.rs

@@ -180,13 +180,19 @@ async fn dispatch(req: &HostRequest, coord: Arc<Coordinator>) -> HostResponse {
                 // submit returns a DAG id immediately, the caller polls
                 // `QueueDag` (`hivectl`'s wait/progress loop) for the
                 // outcome instead of blocking here on the commit.
-                let id = crate::job_queue::submit::reparent(
-                    &coord,
-                    vec![(child.clone(), new_parent.clone())],
-                    crate::job_queue::Source::Manual,
-                    "manual set-parent via hivectl".to_owned(),
-                );
-                HostResponse::queued(vec![id])
+                let inserted = coord.job_queue.insert_job(|b| {
+                    vec![crate::job_queue::templates::reparent(
+                        b,
+                        vec![(child.clone(), new_parent.clone())],
+                    )]
+                });
+                match inserted {
+                    Ok(ids) => {
+                        coord.emit_rebuild_queue_snapshot();
+                        HostResponse::queued(ids.into_iter().map(hive_jobq::NodeId::get).collect())
+                    }
+                    Err(e) => HostResponse::error(format!("queue reparent: {e}")),
+                }
             }
             HostRequest::SetResourceLimits {
                 name,
@@ -777,39 +783,36 @@ enum Verb {
 }
 
 async fn submit_single(coord: &Arc<Coordinator>, name: &str, verb: Verb) -> HostResponse {
-    use crate::job_queue::{Source, submit};
-    let id = match verb {
+    use crate::job_queue::power;
+    // A single-target op is the N-target one with N = 1 — the shapes are
+    // identical, so there is no separate builder to keep in sync.
+    let targets = [name.to_owned()];
+    let ids = match verb {
         Verb::Kill => {
             tracing::info!(%name, "kill");
-            submit::stop(
-                coord,
-                name,
-                Source::Manual,
-                "manual kill via hivectl".to_owned(),
-            )
-            .await
+            power::stop_many(coord, &targets, false).await
         }
         Verb::Restart => {
             tracing::info!(%name, "restart");
-            submit::restart(
-                coord,
-                name,
-                Source::Manual,
-                "manual restart via hivectl".to_owned(),
-            )
-            .await
+            power::restart_many(coord, &targets, false).await
         }
         Verb::Rebuild => {
             tracing::info!(%name, "rebuild");
-            submit::rebuild(
-                coord,
-                name,
-                Source::Manual,
-                "manual rebuild via hivectl".to_owned(),
-            )
+            // Not a power op: a rebuild's shape doesn't depend on live state,
+            // so it is a plain template insert rather than a `*_many` gather.
+            let inserted = coord
+                .job_queue
+                .insert_job(|b| crate::job_queue::templates::rebuild(b, name, true));
+            if inserted.is_ok() {
+                coord.emit_rebuild_queue_snapshot();
+            }
+            inserted
         }
     };
-    HostResponse::queued(vec![id])
+    match ids {
+        Ok(ids) => HostResponse::queued(ids.into_iter().map(hive_jobq::NodeId::get).collect()),
+        Err(e) => HostResponse::error(format!("queue insert failed: {e}")),
+    }
 }
 
 /// Stop the given `agents` (resolved logical names) then `infra` containers
@@ -837,27 +840,17 @@ async fn handle_stop(
     let mut errors: Vec<String> = Vec::new();
     let mut queued: Vec<u64> = Vec::new();
 
-    // One DAG for all targeted agents — a per-agent stop subgraph each
+    // One insert for all targeted agents — a per-agent stop subgraph each
     // (`SetWanted(Offline) → [Signal → Drain →] Reconcile`), independent
-    // roots that run concurrently on their own leases. A hive-wide
-    // `hivectl stop` is now a single DAG, not N.
+    // roots that run concurrently on their own leases.
     if !agents.is_empty() {
-        let reason = if graceful {
-            "manual via hivectl graceful stop"
-        } else {
-            "manual via hivectl stop"
-        };
-        queued.push(
-            crate::job_queue::submit::stop_many(
-                coord,
-                agents,
-                graceful,
-                crate::job_queue::Source::Manual,
-                reason.to_owned(),
-            )
-            .await,
-        );
-        ok_items.extend(agents.iter().cloned());
+        match crate::job_queue::power::stop_many(coord, agents, graceful).await {
+            Ok(ids) => {
+                queued.extend(ids.into_iter().map(hive_jobq::NodeId::get));
+                ok_items.extend(agents.iter().cloned());
+            }
+            Err(e) => errors.push(format!("queue stop: {e}")),
+        }
     }
 
     // Agents go down before infra so they're not mid-request against a
@@ -944,16 +937,13 @@ async fn handle_start(
     // hivectl's wait loop.
     let mut queued: Vec<u64> = Vec::new();
     if !agents.is_empty() {
-        queued.push(
-            crate::job_queue::submit::start_many(
-                coord,
-                agents,
-                crate::job_queue::Source::Manual,
-                "manual via hivectl start".to_owned(),
-            )
-            .await,
-        );
-        ok_items.extend(agents.iter().cloned());
+        match crate::job_queue::power::start_many(coord, agents).await {
+            Ok(ids) => {
+                queued.extend(ids.into_iter().map(hive_jobq::NodeId::get));
+                ok_items.extend(agents.iter().cloned());
+            }
+            Err(e) => errors.push(format!("queue start: {e}")),
+        }
     }
 
     let mut resp = finish_lifecycle(ok_items, &errors);
@@ -982,28 +972,20 @@ async fn handle_restart_scoped(
     let mut errors: Vec<String> = Vec::new();
     let mut queued: Vec<u64> = Vec::new();
 
-    // One DAG for all targeted agents — a per-agent restart subgraph each
-    // (`SetWanted → [Signal → Drain →] StopForUpdate → Reconcile`),
-    // independent roots that run concurrently on their own leases. A
-    // hive-wide `hivectl restart` is now a single DAG, not N. No
-    // client-side stop-then-start composition — the whole restart survives
-    // a dropped connection because the DAG owns it.
+    // One insert for all targeted agents — a per-agent restart subgraph each
+    // (`[Signal → Drain →] StopForUpdate → Reconcile`; no `SetWanted`, since a
+    // restart converges to the agent's *existing* intent rather than rewriting
+    // it), independent roots running concurrently on their own leases. No
+    // client-side stop-then-start composition — the whole restart survives a
+    // dropped connection because the graph owns it.
     if !agents.is_empty() {
-        queued.push(
-            crate::job_queue::submit::restart_many(
-                coord,
-                &agents,
-                graceful,
-                crate::job_queue::Source::Manual,
-                if graceful {
-                    "manual via hivectl restart --graceful".to_owned()
-                } else {
-                    "manual restart via hivectl restart".to_owned()
-                },
-            )
-            .await,
-        );
-        ok_items.extend(agents.iter().cloned());
+        match crate::job_queue::power::restart_many(coord, &agents, graceful).await {
+            Ok(ids) => {
+                queued.extend(ids.into_iter().map(hive_jobq::NodeId::get));
+                ok_items.extend(agents.iter().cloned());
+            }
+            Err(e) => errors.push(format!("queue restart: {e}")),
+        }
     }
 
     for &container in &infra {

hive-c0re/src/socket_server/lifecycle_handlers.rs

@@ -20,13 +20,9 @@ pub(super) async fn handle_start(coord: &Arc<Coordinator>, agent: &str, name: &s
     // Persist `wanted = Up` and submit the Start DAG; the submit layer
     // upgrades a stale-rev start to a full rebuild so the container
     // runs current nix derivations before it starts.
-    crate::job_queue::submit::start(
-        coord,
-        name,
-        crate::job_queue::Source::Manual,
-        format!("agent `{agent}` start tool"),
-    )
-    .await;
+    if let Err(e) = crate::job_queue::power::start_many(coord, &[name.to_owned()]).await {
+        tracing::error!(%agent, %name, error = ?e, "start: insert failed");
+    }
     Response::Ok
 }
 
@@ -48,13 +44,9 @@ pub(super) async fn handle_restart(coord: &Arc<Coordinator>, agent: &str, name:
         return err;
     }
     tracing::info!(%agent, %name, "submit restart");
-    crate::job_queue::submit::restart(
-        coord,
-        name,
-        crate::job_queue::Source::Manual,
-        format!("agent `{agent}` restart tool"),
-    )
-    .await;
+    if let Err(e) = crate::job_queue::power::restart_many(coord, &[name.to_owned()], false).await {
+        tracing::error!(%agent, %name, error = ?e, "restart: insert failed");
+    }
     Response::Ok
 }
 
@@ -153,12 +145,13 @@ pub(super) fn handle_update(coord: &Arc<Coordinator>, agent: &str, name: &str) -
         return err;
     }
     tracing::info!(%agent, %name, "submit rebuild");
-    crate::job_queue::submit::rebuild(
-        coord,
-        name,
-        crate::job_queue::Source::Manual,
-        format!("agent `{agent}` update tool"),
-    );
+    if let Err(e) = coord.job_queue.insert_job(|b| {
+        crate::job_queue::templates::rebuild(b, name, true);
+        Vec::new()
+    }) {
+        tracing::error!(%agent, %name, error = ?e, "update: insert failed");
+    }
+    coord.emit_rebuild_queue_snapshot();
     Response::Ok
 }
 

hive-c0re/src/workers/auto_update.rs

@@ -109,12 +109,11 @@ pub async fn ensure_root_agent(coord: &Arc<Coordinator>) -> Result<()> {
             tracing::warn!(
                 "manager container exists but no applied flake — forcing rebuild to migrate"
             );
-            if let Err(e) = coord.job_queue.submit(
-                crate::job_queue::Source::AutoUpdate,
-                "manager migration: no applied flake".to_owned(),
-                |b| crate::job_queue::templates::rebuild(b, MANAGER_NAME, true),
-            ) {
-                tracing::warn!(error = ?e, "manager migration rebuild submit failed");
+            if let Err(e) = coord.job_queue.insert_job(|b| {
+                crate::job_queue::templates::rebuild(b, MANAGER_NAME, true);
+                Vec::new()
+            }) {
+                tracing::warn!(error = ?e, "manager migration rebuild insert failed");
             }
         } else {
             tracing::debug!("manager container already present");
@@ -378,18 +377,19 @@ fn submit_boot_tree(
     n_deferred: usize,
     n_skipped: usize,
 ) {
-    use crate::job_queue::Source;
-
-    // Fully-quiet boot (nothing stale, nothing drifted) submits nothing.
+    // Fully-quiet boot (nothing stale, nothing drifted) inserts nothing.
     if !any_stale && drifted.is_empty() {
         return;
     }
-    let reason = format!(
-        "boot: {} rebuild(s), {} reconcile(s), {} deferred (offline), {} up-to-date",
-        fanout.len(),
-        drifted.len(),
-        n_deferred,
-        n_skipped,
+    // The summary the sweep used to hand the container as its `reason` is a log
+    // line now: it was only ever stored on a node nobody read, and the counts
+    // are worth having where they can actually be seen.
+    tracing::info!(
+        rebuilds = fanout.len(),
+        reconciles = drifted.len(),
+        deferred = n_deferred,
+        up_to_date = n_skipped,
+        "boot: sweep"
     );
 
     // The sweep's own rebuild subgraphs emit their `Rebuilt` events as they
@@ -397,10 +397,11 @@ fn submit_boot_tree(
     // The subgraphs also carry their own per-agent crash-watch suppression
     // during their `Swap` (applied at claim time); a reconcile-only boot needs
     // no transient.
-    if let Err(e) = coord.job_queue.submit(Source::AutoUpdate, reason, |b| {
+    if let Err(e) = coord.job_queue.insert_job(|b| {
         boot_nodes(b, any_stale, fanout, drifted);
+        Vec::new()
     }) {
-        tracing::warn!(error = ?e, "boot: sweep DAG submit failed");
+        tracing::warn!(error = ?e, "boot: sweep DAG insert failed");
     }
     coord.emit_rebuild_queue_snapshot();
 }

hive-host-sock/src/jobs.rs

@@ -1,6 +1,11 @@
-//! Vocabulary hive-c0re's job queue shares with its clients: where a job
-//! came from ([`Source`]), what a permission change carries
-//! ([`PermPayload`]), and the scheduler's lifecycle [`State`].
+//! Vocabulary hive-c0re's job queue shares with its clients: what a
+//! permission change carries ([`PermPayload`]) and the scheduler's
+//! lifecycle [`State`].
+//!
+//! A `Source` enum lived here too — where a job came from, rendered as the
+//! "why" chip. It was a field on the DAG container, and it went with it: a
+//! job is its nodes now, and a node says what it does rather than who asked
+//! for it.
 //!
 //! **The typed `DagView`/`NodeView` projection that used to live here is
 //! gone.** One graph is served one way now — `hive_jobq_wire`'s generic
@@ -12,37 +17,6 @@
 
 use serde::{Deserialize, Serialize};
 
-/// Where the submit request originated — drives the "why" chip on the
-/// dashboard.
-#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
-#[serde(rename_all = "snake_case")]
-pub enum Source {
-    /// Operator action (dashboard button, CLI, manager tool).
-    Manual,
-    /// Meta-update cascade rebuild (grown into the meta-update DAG).
-    MetaUpdate,
-    /// Boot-time submission (the boot sweep DAG + boot reconciles).
-    AutoUpdate,
-    /// Crash recovery path (future use).
-    CrashRecover,
-    /// Operator approved a pending `Approval` row; `approval_id` on
-    /// the DAG points back at the source row.
-    Approval,
-}
-
-impl Source {
-    #[must_use]
-    pub fn as_str(self) -> &'static str {
-        match self {
-            Source::Manual => "manual",
-            Source::MetaUpdate => "meta_update",
-            Source::AutoUpdate => "auto_update",
-            Source::CrashRecover => "crash_recover",
-            Source::Approval => "approval",
-        }
-    }
-}
-
 pub use hive_jobq::State;
 
 /// Kind-specific payload for `Template::PermChange` DAGs.