diff --git a/docs/coordinator.md b/docs/coordinator.md
index 115d8ec2..b4e6b4c2 100644
--- a/docs/coordinator.md
+++ b/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
diff --git a/hive-c0re/src/actions.rs b/hive-c0re/src/actions.rs
index 7333f387..11c94c2f 100644
--- a/hive-c0re/src/actions.rs
+++ b/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();
 }
diff --git a/hive-c0re/src/dashboard/lifecycle_ops.rs b/hive-c0re/src/dashboard/lifecycle_ops.rs
index 87e7c2d0..e12723fa 100644
--- a/hive-c0re/src/dashboard/lifecycle_ops.rs
+++ b/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()
 }
 
diff --git a/hive-c0re/src/dashboard/meta_inputs.rs b/hive-c0re/src/dashboard/meta_inputs.rs
index 3efe6b74..3f6705d8 100644
--- a/hive-c0re/src/dashboard/meta_inputs.rs
+++ b/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()
 }
 
diff --git a/hive-c0re/src/dashboard/permissions.rs b/hive-c0re/src/dashboard/permissions.rs
index 601ca400..d542fc8c 100644
--- a/hive-c0re/src/dashboard/permissions.rs
+++ b/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())
diff --git a/hive-c0re/src/dashboard/topology.rs b/hive-c0re/src/dashboard/topology.rs
index 8e8422fb..37ec5045 100644
--- a/hive-c0re/src/dashboard/topology.rs
+++ b/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())
 }
diff --git a/hive-c0re/src/job_queue/exec.rs b/hive-c0re/src/job_queue/exec.rs
index fe0c4074..d8866327 100644
--- a/hive-c0re/src/job_queue/exec.rs
+++ b/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)
 }
diff --git a/hive-c0re/src/job_queue/mod.rs b/hive-c0re/src/job_queue/mod.rs
index b9727493..04ed00af 100644
--- a/hive-c0re/src/job_queue/mod.rs
+++ b/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
diff --git a/hive-c0re/src/job_queue/model.rs b/hive-c0re/src/job_queue/model.rs
index f1f7936d..57cf0dca 100644
--- a/hive-c0re/src/job_queue/model.rs
+++ b/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 { .. } => "",
         }
     }
 
diff --git a/hive-c0re/src/job_queue/submit.rs b/hive-c0re/src/job_queue/power.rs
similarity index 51%
rename from hive-c0re/src/job_queue/submit.rs
rename to hive-c0re/src/job_queue/power.rs
index 3ec3e11b..b1606a57 100644
--- a/hive-c0re/src/job_queue/submit.rs
+++ b/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)
 }
diff --git a/hive-c0re/src/job_queue/scheduler.rs b/hive-c0re/src/job_queue/scheduler.rs
index 3c19c529..f60f2028 100644
--- a/hive-c0re/src/job_queue/scheduler.rs
+++ b/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(),
diff --git a/hive-c0re/src/job_queue/templates.rs b/hive-c0re/src/job_queue/templates.rs
index 536093fc..b90827b4 100644
--- a/hive-c0re/src/job_queue/templates.rs
+++ b/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`
diff --git a/hive-c0re/src/job_queue/tests.rs b/hive-c0re/src/job_queue/tests.rs
index 4bc199a9..41e45018 100644
--- a/hive-c0re/src/job_queue/tests.rs
+++ b/hive-c0re/src/job_queue/tests.rs
@@ -4,7 +4,7 @@
 //! that graph (wire projection, history retention, error truncation).
 //!
 //! **Nothing here runs a node.** Everything a template declares is in the graph
-//! the moment `submit` returns, so the assertions read it there. Whether the
+//! the moment `insert` returns, so the assertions read it there. Whether the
 //! scheduler then honours those declarations — cascade, roll-up, grant
 //! borrow/release, fairness, the `Finishing` gate — is `hive_jobq`'s property
 //! and is tested in `hive_jobq`, against its own primitives rather than through
@@ -17,36 +17,64 @@
 use super::model::NodeKind;
 use super::*;
 
-/// Submit a declared shape with the metadata every mechanics test uses.
-/// `Source::Manual` because none of these exercise provenance — the tests that
-/// do name their own source at the call site.
-fn submit(q: &JobQueue, reason: &str, declare: impl FnOnce(&JobBuilder)) -> u64 {
-    q.submit(Source::Manual, reason.to_owned(), declare)
+/// Insert a declared job, naming nothing — the shape assertions read the whole
+/// graph. A test that needs a handle calls `q.insert` directly and names the
+/// node it cares about.
+fn insert(q: &JobQueue, declare: impl FnOnce(&JobBuilder)) {
+    q.insert_job(|b| {
+        declare(b);
+        Vec::new()
+    })
+    .expect("valid shape");
+}
+
+/// Insert a declared job and hand back the ids of the nodes it **named**, in
+/// the order it named them.
+///
+/// This is the handle that replaced the DAG id: there is no container to point
+/// at any more, so a test that needs to cancel a job or read its state names
+/// the roots it cares about — exactly what production does with the ids
+/// [`JobQueue::insert_job`] returns.
+/// Handed back as raw `u64`, the same form production passes to `cancel` and
+/// `node_subtrees` — a `NodeId` cannot be fabricated, so the read surface takes
+/// raw ids and searches for them.
+fn insert_named(
+    q: &JobQueue,
+    declare: impl FnOnce(&JobBuilder) -> Vec<hive_jobq::NodeGuid>,
+) -> Vec<u64> {
+    q.insert_job(declare)
         .expect("valid shape")
+        .into_iter()
+        .map(NodeId::get)
+        .collect()
 }
 
 fn ident(s: &str) -> hive_types::Ident {
     hive_types::Ident::parse(s).expect("valid test ident")
 }
 
-fn rebuild(builder: &JobBuilder, agent: &str) {
-    templates::rebuild(builder, agent, true);
+fn rebuild(builder: &JobBuilder, agent: &str) -> Vec<hive_jobq::NodeGuid> {
+    templates::rebuild(builder, agent, true)
 }
 
 /// Restart shape with every agent treated as **running** — the online
 /// shape (`[Signal→Drain→] StopForUpdate → Reconcile`, no `SetWanted` head)
 /// most queue-mechanics tests assume. Mirrors the pre-dynamic
-/// `templates::restart` (which is now the state-aware `submit::restart_nodes`).
-fn restart_online(builder: &JobBuilder, agents: &[&str], graceful: bool) {
+/// `templates::restart` (which is now the state-aware `power::restart_nodes`).
+fn restart_online(
+    builder: &JobBuilder,
+    agents: &[&str],
+    graceful: bool,
+) -> Vec<hive_jobq::NodeGuid> {
     let targets: Vec<(String, bool)> = agents.iter().map(|a| ((*a).to_owned(), true)).collect();
-    submit::restart_nodes(builder, &targets, graceful);
+    power::restart_nodes(builder, &targets, graceful)
 }
 
 /// Stop shape with every agent treated as **running** — the online shape
 /// (`SetWanted → [Signal→Drain→](graceful) Reconcile`).
-fn stop_online(builder: &JobBuilder, agents: &[&str], graceful: bool) {
+fn stop_online(builder: &JobBuilder, agents: &[&str], graceful: bool) -> Vec<hive_jobq::NodeGuid> {
     let targets: Vec<(String, bool)> = agents.iter().map(|a| ((*a).to_owned(), true)).collect();
-    submit::stop_nodes(builder, &targets, graceful);
+    power::stop_nodes(builder, &targets, graceful)
 }
 
 // `Claimed` / `ClaimReady` / `CompleteNode` lived here: a claim snapshot type
@@ -62,8 +90,8 @@ fn stop_online(builder: &JobBuilder, agents: &[&str], graceful: bool) {
 #[derive(Debug, PartialEq, Eq)]
 struct Declared {
     kind: &'static str,
-    /// Parent kind, or `None` when the node hangs directly under the DAG
-    /// container (i.e. it is a group root).
+    /// Parent kind, or `None` when the node is a group root — which is now a
+    /// genuine `parent = None`, not "hangs under the container".
     parent: Option<&'static str>,
     /// Kinds this node declared a node-dep on, in declaration order, each with
     /// the outcome set that satisfies it.
@@ -96,33 +124,35 @@ fn when_tag(when: hive_jobq::DepWhen) -> String {
     .join("|")
 }
 
-/// Every work node under `dag`, in insertion order, as its declared shape.
+/// Every work node in the graph, in insertion order, as its declared shape.
 ///
 /// **This is what the template tests are actually about.** A template's output
-/// is fully determined the moment `submit` returns: the kinds, the parent
+/// is fully determined the moment `insert` returns: the kinds, the parent
 /// nesting and the dep edges are all sitting in the graph. Reading them here
 /// keeps the assertion on c0re's own product. Whether the scheduler then
 /// *honours* those edges — runs a chain serially, holds a grant across a
 /// subtree — is `hive_jobq`'s property and is tested in `hive_jobq`.
-fn declared_shape(q: &JobQueue, dag: u64) -> Vec<Declared> {
-    declared_shape_filtered(q, dag, &|_| true)
+fn declared_shape(q: &JobQueue) -> Vec<Declared> {
+    declared_shape_filtered(q, &|_| true)
 }
 
-fn declared_shape_filtered(
-    q: &JobQueue,
-    dag: u64,
-    keep: &dyn Fn(&NodeKind) -> bool,
-) -> Vec<Declared> {
+/// Every node in the graph, since each test inserts into a fresh [`JobQueue`].
+///
+/// This used to take a DAG id and filter by `root_of(n) == Some(container)`.
+/// With no container node there is no per-DAG root to filter on — and nothing
+/// to exclude either, because every node in the graph is now real work. Tests
+/// that insert more than one job name a node per job and assert on the ids
+/// [`JobQueue::insert`] hands back.
+fn declared_shape_filtered(q: &JobQueue, keep: &dyn Fn(&NodeKind) -> bool) -> Vec<Declared> {
     let sched = q.sched().lock().expect("job_queue mutex poisoned");
     let graph = sched.graph();
-    let root = graph.resolve_id(dag).expect("dag id is a real node id");
     let kind_of = |id: NodeId| graph.node(id).map(|n| n.payload.as_str());
     graph
         .nodes()
-        .filter(|n| n.id != root && graph.root_of(n.id) == Some(root) && keep(&n.payload))
+        .filter(|n| keep(&n.payload))
         .map(|n| Declared {
             kind: n.payload.as_str(),
-            parent: n.parent.filter(|p| *p != root).and_then(kind_of),
+            parent: n.parent.and_then(kind_of),
             after: n
                 .deps
                 .iter()
@@ -135,78 +165,66 @@ fn declared_shape_filtered(
         .collect()
 }
 
-/// The id of the one node of `kind` under `dag`, for the resource assertions.
+/// The id of the one node of `kind`, for the resource assertions.
 ///
 /// Panics unless there is exactly one — every caller is about a shape where the
 /// kind is unique, so two would mean the assertion had quietly stopped being
 /// about the node the test names.
-fn node_of(q: &JobQueue, dag: u64, kind: &str) -> hive_jobq::NodeId {
+fn node_of(q: &JobQueue, kind: &str) -> hive_jobq::NodeId {
     let sched = q.sched().lock().expect("job_queue mutex poisoned");
     let graph = sched.graph();
-    let root = graph.resolve_id(dag).expect("dag id is a real node id");
     let mut found: Vec<_> = graph
         .nodes()
-        .filter(|n| n.id != root && graph.root_of(n.id) == Some(root) && n.payload.as_str() == kind)
+        .filter(|n| n.payload.as_str() == kind)
         .map(|n| n.id)
         .collect();
     assert_eq!(
         found.len(),
         1,
-        "expected exactly one {kind} node in the dag"
+        "expected exactly one {kind} node in the graph"
     );
     found.pop().expect("checked above")
 }
 
-/// The payload of the one node of `kind` under `dag`, for assertions about what
-/// a node *carries* rather than how it is wired.
-fn payload_of(q: &JobQueue, dag: u64, kind: &str) -> NodeKind {
-    let id = node_of(q, dag, kind);
+/// The payload of the one node of `kind`, for assertions about what a node
+/// *carries* rather than how it is wired.
+fn payload_of(q: &JobQueue, kind: &str) -> NodeKind {
+    let id = node_of(q, kind);
     let sched = q.sched().lock().expect("job_queue mutex poisoned");
     sched.graph().node(id).expect("node exists").payload.clone()
 }
 
-/// Kinds of every node under `dag` still `Pending` — the nodes that could yet
-/// run. Stronger than asking the scheduler what is *ready right now*: a node
+/// Kinds of every node still `Pending` — the nodes that could yet run.
+/// Stronger than asking the scheduler what is *ready right now*: a node
 /// blocked on a dep is not ready but is very much still alive.
-fn pending_kinds(q: &JobQueue, dag: u64) -> Vec<&'static str> {
-    pending_kinds_filtered(q, dag, &|_| true)
+fn pending_kinds(q: &JobQueue) -> Vec<&'static str> {
+    pending_kinds_filtered(q, &|_| true)
 }
 
 /// [`pending_kinds`] restricted to the nodes whose payload names `agent`.
-fn pending_kinds_for(q: &JobQueue, dag: u64, agent: &str) -> Vec<&'static str> {
-    pending_kinds_filtered(q, dag, &|kind: &NodeKind| kind.agent() == agent)
+fn pending_kinds_for(q: &JobQueue, agent: &str) -> Vec<&'static str> {
+    pending_kinds_filtered(q, &|kind: &NodeKind| kind.agent() == agent)
 }
 
-/// The payloads of every node under `dag` still `Pending`, for the cases where
-/// *which* of a family of same-kind nodes survived is the assertion — a
-/// template emits one tail per outcome and they differ only in what they carry.
-fn pending_payloads(q: &JobQueue, dag: u64) -> Vec<NodeKind> {
+/// The payloads of every node still `Pending`, for the cases where *which* of a
+/// family of same-kind nodes survived is the assertion — a template emits one
+/// tail per outcome and they differ only in what they carry.
+fn pending_payloads(q: &JobQueue) -> Vec<NodeKind> {
     let sched = q.sched().lock().expect("job_queue mutex poisoned");
     let graph = sched.graph();
-    let root = graph.resolve_id(dag).expect("dag id is a real node id");
     graph
         .nodes()
-        .filter(|n| n.id != root && graph.root_of(n.id) == Some(root) && n.state == State::Pending)
+        .filter(|n| n.state == State::Pending)
         .map(|n| n.payload.clone())
         .collect()
 }
 
-fn pending_kinds_filtered(
-    q: &JobQueue,
-    dag: u64,
-    keep: &dyn Fn(&NodeKind) -> bool,
-) -> Vec<&'static str> {
+fn pending_kinds_filtered(q: &JobQueue, keep: &dyn Fn(&NodeKind) -> bool) -> Vec<&'static str> {
     let sched = q.sched().lock().expect("job_queue mutex poisoned");
     let graph = sched.graph();
-    let root = graph.resolve_id(dag).expect("dag id is a real node id");
     graph
         .nodes()
-        .filter(|n| {
-            n.id != root
-                && graph.root_of(n.id) == Some(root)
-                && n.state == State::Pending
-                && keep(&n.payload)
-        })
+        .filter(|n| n.state == State::Pending && keep(&n.payload))
         .map(|n| n.payload.as_str())
         .collect()
 }
@@ -245,20 +263,19 @@ fn declared_resources(q: &JobQueue, node_id: hive_jobq::NodeId) -> Vec<Resource>
         .collect()
 }
 
-/// The resources declared by **every** node of `kind` under `dag`, one row per
+/// The resources declared by **every** node of `kind`, one row per
 /// node, sorted so the rows read as a set rather than an insertion order.
 ///
 /// The per-agent templates emit several nodes of one kind — one per agent — and
 /// what makes them concurrent is that each holds only its *own* agent's lease.
 /// That is a statement about the whole family, so it needs all the rows, not
 /// [`declared_resources`]'s single node.
-fn declared_resources_of_kind(q: &JobQueue, dag: u64, kind: &str) -> Vec<Vec<Resource>> {
+fn declared_resources_of_kind(q: &JobQueue, kind: &str) -> Vec<Vec<Resource>> {
     let sched = q.sched().lock().expect("job_queue mutex poisoned");
     let graph = sched.graph();
-    let root = graph.resolve_id(dag).expect("dag id is a real node id");
     let mut rows: Vec<Vec<Resource>> = graph
         .nodes()
-        .filter(|n| n.id != root && graph.root_of(n.id) == Some(root) && n.payload.as_str() == kind)
+        .filter(|n| n.payload.as_str() == kind)
         .map(|n| {
             n.deps
                 .iter()
@@ -279,8 +296,8 @@ fn declared_resources_of_kind(q: &JobQueue, dag: u64, kind: &str) -> Vec<Vec<Res
 /// cannot tell them apart — two `set_wanted` rows look identical. Slicing by
 /// agent is what makes "the fresh agent goes straight to reconcile while the
 /// stale one rebuilds first" expressible as a declared shape.
-fn declared_shape_for(q: &JobQueue, dag: u64, agent: &str) -> Vec<Declared> {
-    declared_shape_filtered(q, dag, &|kind: &NodeKind| kind.agent() == agent)
+fn declared_shape_for(q: &JobQueue, agent: &str) -> Vec<Declared> {
+    declared_shape_filtered(q, &|kind: &NodeKind| kind.agent() == agent)
 }
 
 fn state_of(q: &JobQueue, dag_id: u64) -> State {
@@ -307,63 +324,68 @@ fn dag_count(q: &JobQueue) -> usize {
         .count()
 }
 
-// ---- submit (dedup removed — every submit is a fresh DAG) ----
+// ---- insert (dedup removed — every insert is a fresh job) ----
+//
+// `submit_assigns_distinct_ids` lived here and is gone. Its whole body was
+// "two inserts get different container ids" — an assertion about the id
+// allocation of a node type this issue deleted, and in any case `hive_jobq`'s
+// property rather than c0re's. What the tests below keep is the part that was
+// about c0re: **no dedup**, now read off the group count instead of off an id.
 
-#[test]
-fn submit_assigns_distinct_ids() {
-    let q = JobQueue::new(1);
-    let first = submit(&q, "first", |builder| rebuild(builder, "agent-a"));
-    let second = submit(&q, "second", |builder| rebuild(builder, "agent-b"));
-    assert_ne!(first, second);
-    assert_eq!(dag_count(&q), 2);
-}
-
-/// Submit-time dedup was removed with the agent-per-node refactor (a
-/// multi-agent DAG has no single agent to key a dedup on), so an identical
-/// resubmit — same template + agent, still queued — now enqueues a distinct
-/// DAG instead of collapsing into the pending one. Whether any dedup needs
+/// Insert-time dedup was removed with the agent-per-node refactor (a
+/// multi-agent job has no single agent to key a dedup on), so an identical
+/// re-insert — same template + agent, still queued — now enqueues a distinct
+/// group instead of collapsing into the pending one. Whether any dedup needs
 /// reintroducing is tracked as a follow-up.
 #[test]
 fn identical_resubmit_is_a_distinct_dag() {
     let q = JobQueue::new(1);
-    let first = submit(&q, "first", |builder| rebuild(builder, "agent-a"));
-    let resubmit = submit(&q, "again", |builder| rebuild(builder, "agent-a"));
-    assert_ne!(first, resubmit, "no dedup: identical resubmit is a new DAG");
-    assert_eq!(dag_count(&q), 2);
+    let first = insert_named(&q, |builder| rebuild(builder, "agent-a"));
+    let after_first = dag_count(&q);
+    let resubmit = insert_named(&q, |builder| rebuild(builder, "agent-a"));
+    assert_ne!(
+        first, resubmit,
+        "no dedup: an identical re-insert declares its own nodes"
+    );
+    // Counted as a doubling rather than against a literal: one rebuild is
+    // several group roots now, and pinning the number here would make this
+    // test fail on any shape change while saying nothing about dedup.
+    assert_eq!(
+        dag_count(&q),
+        after_first * 2,
+        "the re-insert added its own roots instead of collapsing into the pending ones"
+    );
 }
 
 #[test]
 fn distinct_submits_never_collapse() {
     let q = JobQueue::new(1);
-    let rebuild_a = submit(&q, "r", |builder| rebuild(builder, "agent-a"));
-    let rebuild_b = submit(&q, "r", |builder| rebuild(builder, "agent-b"));
-    let restart_a = submit(&q, "r", |builder| {
-        restart_online(builder, &["agent-a"], false);
-    });
+    let rebuild_a = insert_named(&q, |builder| rebuild(builder, "agent-a"));
+    let one_rebuild = dag_count(&q);
+    let rebuild_b = insert_named(&q, |builder| rebuild(builder, "agent-b"));
+    let two_rebuilds = dag_count(&q);
+    let restart_a = insert_named(&q, |builder| restart_online(builder, &["agent-a"], false));
     assert_ne!(rebuild_a, rebuild_b);
     assert_ne!(rebuild_a, restart_a);
-    assert_eq!(dag_count(&q), 3);
+    assert_eq!(
+        two_rebuilds,
+        one_rebuild * 2,
+        "two rebuilds, nothing merged"
+    );
+    assert_eq!(
+        dag_count(&q),
+        two_rebuilds + restart_a.len(),
+        "a restart of an agent that already has a queued rebuild is still its own group"
+    );
 }
 
-#[test]
-fn resubmit_while_running_is_new_dag() {
-    // The "while running" is not load-bearing and used to be staged by claiming
-    // a node first. `submit` appends a container and inserts the declared
-    // group; it never consults the state of any existing node, so whether an
-    // earlier DAG is running cannot change the outcome. What is actually being
-    // asserted — no dedup, ever — is `identical_resubmit_is_a_distinct_dag`.
-    //
-    // Kept as the *named* case because "a config bump mid-build must not be
-    // swallowed" is the scenario people worry about, and a reader looking for
-    // it should find it.
-    let q = JobQueue::new(1);
-    let a = submit(&q, "first", |builder| rebuild(builder, "agent-a"));
-    let again = submit(&q, "config bumped during build", |builder| {
-        rebuild(builder, "agent-a");
-    });
-    assert_ne!(a, again);
-    assert_eq!(dag_count(&q), 2);
-}
+// `resubmit_while_running_is_new_dag` lived here: the same two inserts as
+// above, kept under a second name so a reader looking for "a config bump
+// mid-build must not be swallowed" would find it. It asserted nothing the
+// test above doesn't — `insert_job` never consults the state of an existing
+// node, so "while running" could not change the outcome and was never staged.
+// The scenario is named in that test's doc instead; a duplicate test is a
+// second place for the same fact to rot.
 
 // ---- malformed specs: no longer expressible ----
 //
@@ -387,7 +409,7 @@ fn resubmit_while_running_is_new_dag() {
 #[test]
 fn rebuild_chain_is_declared_serial() {
     // Was `rebuild_chain_claims_in_dep_order`, which drove the whole DAG to
-    // observe an order that is fully declared the moment `submit` returns.
+    // observe an order that is fully declared the moment `insert` returns.
     //
     // ⚠️ The old name was also wrong about the mechanism, and reading it rather
     // than the graph is how you'd stay wrong: **only part of this chain is dep
@@ -401,9 +423,11 @@ fn rebuild_chain_is_declared_serial() {
     // The non-graceful shape asserted here has no quiesce chain, so nothing here
     // is concurrent — but the name would mislead about the graceful one.
     let q = JobQueue::new(1);
-    let id = submit(&q, "r", |builder| rebuild(builder, "agent-a"));
+    insert(&q, |builder| {
+        rebuild(builder, "agent-a");
+    });
     assert_eq!(
-        declared_shape(&q, id),
+        declared_shape(&q),
         vec![
             row("meta_sync", None, &[]),
             // The brace: holds the lease + slot for everything nested below it.
@@ -468,20 +492,14 @@ fn rebuild_chain_is_declared_serial() {
 #[test]
 fn graceful_rebuild_chain_drains_before_stopping() {
     let q = JobQueue::new(1);
-    let id = q
-        .submit(
-            Source::AutoUpdate,
-            "sweep".to_owned(),
-            |builder: &JobBuilder| {
-                templates::graceful_rebuild_nodes(builder, "agent-a", true, None);
-            },
-        )
-        .expect("valid shape");
+    insert(&q, |builder| {
+        templates::graceful_rebuild_nodes(builder, "agent-a", true, None);
+    });
     // Asserted as full rows, not just kinds: the kind list is identical whether
     // the quiesce chain runs beside the build or nested under it, so a
     // kind-only assertion cannot see the bug this shape exists to fix.
     assert_eq!(
-        declared_shape(&q, id),
+        declared_shape(&q),
         vec![
             row("meta_sync", None, &[]),
             row("agent_window", None, &[("meta_sync", "done")]),
@@ -530,11 +548,11 @@ fn non_graceful_rebuild_has_no_signal_or_drain() {
     // job keeps its nodes to itself and inserts them, so what it built is
     // observable where it matters — in what the scheduler runs.
     let q = JobQueue::new(1);
-    let id = submit(&q, "manual", |builder| {
+    insert(&q, |builder| {
         templates::rebuild_nodes(builder, "agent-a", true, None);
     });
     assert_eq!(
-        declared_shape(&q, id)
+        declared_shape(&q)
             .iter()
             .map(|d| d.kind)
             .collect::<Vec<_>>(),
@@ -567,8 +585,10 @@ fn rebuild_chain_declares_its_resources_on_the_brace() {
     // `a_contended_resource_goes_to_the_oldest_waiter`. What is c0re's is
     // *which* nodes contend in the first place — a declaration, asserted here.)
     let q = JobQueue::new(1);
-    let id = submit(&q, "r", |builder| rebuild(builder, "agent-a"));
-    let res = |kind: &str| declared_resources(&q, node_of(&q, id, kind));
+    insert(&q, |builder| {
+        rebuild(builder, "agent-a");
+    });
+    let res = |kind: &str| declared_resources(&q, node_of(&q, kind));
 
     let agent = || Resource::Agent("agent-a".to_owned());
     assert_eq!(
@@ -610,20 +630,25 @@ fn rebuild_chain_declares_its_resources_on_the_brace() {
 // ---- per-agent lease ----
 
 #[test]
-fn multi_agent_restart_is_one_dag_with_concurrent_per_agent_subgraphs() {
+fn multi_agent_restart_declares_concurrent_per_agent_subgraphs() {
     let q = JobQueue::new(4);
-    let id = submit(&q, "hive-wide", |builder| {
-        restart_online(builder, &["agent-a", "agent-b"], false);
+    let roots = insert_named(&q, |builder| {
+        restart_online(builder, &["agent-a", "agent-b"], false)
     });
-    // A hive-wide restart is ONE DAG, not one-per-agent.
-    assert_eq!(dag_count(&q), 1);
+    // Was "a hive-wide restart is ONE DAG": one container over both agents.
+    // With the container gone it is one *insert* over N independent groups —
+    // which is the same claim about the operator's action and a better one
+    // about the graph, since independence is what lets them run at once.
+    // Asserted against the named count rather than a literal: the point is
+    // that every root the job named is top-level, with nothing above it.
+    assert_eq!(dag_count(&q), roots.len(), "every named root is top-level");
     // Each agent's subgraph head (StopForUpdate, since both are running) is a
     // group root with no deps, so nothing orders them against each other; and
     // each declares only its OWN agent's lease, so nothing makes them contend.
     // Those two declared facts are what "they run concurrently" *means* here —
     // that a scheduler then does run independent, resource-disjoint roots at
     // once is hive_jobq's property, tested there.
-    let heads: Vec<_> = declared_shape(&q, id)
+    let heads: Vec<_> = declared_shape(&q)
         .into_iter()
         .filter(|d| d.kind == "stop_for_update")
         .collect();
@@ -636,7 +661,7 @@ fn multi_agent_restart_is_one_dag_with_concurrent_per_agent_subgraphs() {
         "both per-agent heads are independent group roots"
     );
     assert_eq!(
-        declared_resources_of_kind(&q, id, "stop_for_update"),
+        declared_resources_of_kind(&q, "stop_for_update"),
         vec![
             vec![Resource::Agent("agent-a".to_owned())],
             vec![Resource::Agent("agent-b".to_owned())],
@@ -646,18 +671,18 @@ fn multi_agent_restart_is_one_dag_with_concurrent_per_agent_subgraphs() {
 }
 
 #[test]
-fn multi_agent_stop_is_one_dag_with_concurrent_per_agent_subgraphs() {
+fn multi_agent_stop_declares_concurrent_per_agent_subgraphs() {
     let q = JobQueue::new(4);
-    let id = submit(&q, "hive-wide stop", |builder| {
-        stop_online(builder, &["agent-a", "agent-b"], false);
+    let roots = insert_named(&q, |builder| {
+        stop_online(builder, &["agent-a", "agent-b"], false)
     });
-    // A hive-wide stop is ONE DAG, not one-per-agent.
-    assert_eq!(dag_count(&q), 1);
+    // See the restart case above for why this is a named-root count now.
+    assert_eq!(dag_count(&q), roots.len(), "every named root is top-level");
     // Same declared story as the restart case above: each agent's subgraph head
     // is a group root with no node-deps, holding only its own agent's lease.
     // Independent roots on disjoint resources is what "concurrently" means at
     // this layer — the running of them is hive_jobq's.
-    let heads: Vec<_> = declared_shape(&q, id)
+    let heads: Vec<_> = declared_shape(&q)
         .into_iter()
         .filter(|d| d.kind == "set_wanted")
         .collect();
@@ -667,7 +692,7 @@ fn multi_agent_stop_is_one_dag_with_concurrent_per_agent_subgraphs() {
         "both per-agent stop subgraph heads are independent group roots"
     );
     assert_eq!(
-        declared_resources_of_kind(&q, id, "set_wanted"),
+        declared_resources_of_kind(&q, "set_wanted"),
         vec![
             vec![Resource::Agent("agent-a".to_owned())],
             vec![Resource::Agent("agent-b".to_owned())],
@@ -677,27 +702,30 @@ fn multi_agent_stop_is_one_dag_with_concurrent_per_agent_subgraphs() {
 }
 
 #[test]
-fn multi_agent_start_one_dag_folds_per_agent_stale_rebuild() {
+fn multi_agent_start_folds_per_agent_stale_rebuild() {
     let q = JobQueue::new(4);
     // fresh: offline + not stale → SetWanted → Reconcile.
     // stale: offline + stale → SetWanted → «rebuild subgraph».
-    let id = submit(&q, "hive-wide start", |builder| {
-        submit::start_nodes(
+    let roots = insert_named(&q, |builder| {
+        power::start_nodes(
             builder,
             &[
                 ("fresh".to_owned(), false, false),
                 ("stale".to_owned(), false, true),
             ],
-        );
+        )
     });
-    // One DAG spanning both agents.
-    assert_eq!(dag_count(&q), 1);
+    // One insert spanning both agents — and an *uneven* number of roots, which
+    // is the shape this test is about: the fresh agent names one, the stale one
+    // names four (its rebuild chains behind `SetWanted` rather than nesting
+    // under it, so the head alone would report the start done mid-rebuild).
+    assert_eq!(dag_count(&q), roots.len(), "every named root is top-level");
     // The fold is a *declared* difference, readable the moment submit returns:
     // both agents get a `SetWanted(Up)` group root, but the fresh agent's
     // subgraph ends at the Reconcile behind it while the stale agent's carries
     // the whole rebuild chain in between.
     assert_eq!(
-        declared_shape_for(&q, id, "fresh"),
+        declared_shape_for(&q, "fresh"),
         vec![
             row("set_wanted", None, &[]),
             row("reconcile", Some("set_wanted"), &[]),
@@ -705,7 +733,7 @@ fn multi_agent_start_one_dag_folds_per_agent_stale_rebuild() {
         "a fresh agent is intent + convergence, nothing in between"
     );
     assert_eq!(
-        declared_shape_for(&q, id, "stale"),
+        declared_shape_for(&q, "stale"),
         vec![
             row("set_wanted", None, &[]),
             row("meta_sync", None, &[("set_wanted", "done")]),
@@ -741,31 +769,32 @@ fn offline_agents_skip_mechanical_nodes_but_keep_reconcile() {
     // read and node exec.
     let q = JobQueue::new(4);
     // Offline graceful stop → SetWanted(Off) → Reconcile (no Signal/Drain).
-    let stop = submit(&q, "stop down", |builder| {
-        submit::stop_nodes(builder, &[("down".to_owned(), false)], true);
+    let stop = insert_named(&q, |builder| {
+        power::stop_nodes(builder, &[("down".to_owned(), false)], true)
     });
     // Offline restart → a lone Reconcile (no SetWanted, no StopForUpdate):
     // nothing to bounce, and restart never rewrites intent, so the tail
     // Reconcile converges the down agent to its existing `wanted`.
-    let restart = submit(&q, "restart down", |builder| {
-        submit::restart_nodes(builder, &[("down2".to_owned(), false)], true);
+    let restart = insert_named(&q, |builder| {
+        power::restart_nodes(builder, &[("down2".to_owned(), false)], true)
     });
-    let shape = |id: u64| -> Vec<String> {
-        // The group's work nodes: its subtree minus the container itself,
-        // which the generic view carries as an ordinary node.
-        q.node_subtrees(&[id])
+    // The whole group, **root included** — the root is a work node now
+    // (`SetWanted` for the stop, the lone `Reconcile` for the restart), not a
+    // container to be filtered out. One id per agent, which is what the chain
+    // named.
+    let shape = |roots: &[u64]| -> Vec<String> {
+        q.node_subtrees(roots)
             .iter()
-            .filter(|n| n.id != id)
             .map(|n| n.payload.label.clone())
             .collect()
     };
     assert_eq!(
-        shape(stop),
+        shape(&stop),
         vec!["set_wanted".to_owned(), "reconcile".to_owned()],
         "offline graceful stop skips the signal/drain quiesce, keeps Reconcile"
     );
     assert_eq!(
-        shape(restart),
+        shape(&restart),
         vec!["reconcile".to_owned()],
         "offline restart is a lone Reconcile (no SetWanted head, nothing to stop)"
     );
@@ -782,22 +811,16 @@ fn boot_sweep_nodes_declare_their_own_resources() {
     // meta commit inside another node's staged deploy window. Nothing failed to
     // compile; only an exhaustive caller list would have caught it.
     let q = JobQueue::new(4);
-    let id = q
-        .submit(
-            Source::AutoUpdate,
-            "boot".to_owned(),
-            |builder: &JobBuilder| {
-                crate::workers::auto_update::boot_nodes(
-                    builder,
-                    true,
-                    vec!["stale-agent".to_owned()],
-                    vec!["drifted-agent".to_owned()],
-                );
-            },
-        )
-        .expect("valid shape");
+    insert(&q, |builder| {
+        crate::workers::auto_update::boot_nodes(
+            builder,
+            true,
+            vec!["stale-agent".to_owned()],
+            vec!["drifted-agent".to_owned()],
+        );
+    });
 
-    let mut lock = declared_resources(&q, node_of(&q, id, "meta_lock"));
+    let mut lock = declared_resources(&q, node_of(&q, "meta_lock"));
     lock.sort_by_key(|r| format!("{r:?}"));
     assert_eq!(
         lock,
@@ -806,7 +829,7 @@ fn boot_sweep_nodes_declare_their_own_resources() {
     );
 
     assert_eq!(
-        declared_resources(&q, node_of(&q, id, "reconcile")),
+        declared_resources(&q, node_of(&q, "reconcile")),
         vec![Resource::Agent("drifted-agent".to_owned())],
         "a boot Reconcile touches the container, so it holds that agent's lease"
     );
@@ -901,8 +924,10 @@ fn rebuild_reconcile_waits_for_the_whole_build_subtree() {
     // **did** flatten this chain, and the guarantee survives only because the
     // roll-up point moved with it.
     let q = JobQueue::new(1);
-    let id = submit(&q, "r", |builder| rebuild(builder, "agent-a"));
-    let shape = declared_shape(&q, id);
+    insert(&q, |builder| {
+        rebuild(builder, "agent-a");
+    });
+    let shape = declared_shape(&q);
     let parent_of = |kind: &str| {
         shape
             .iter()
@@ -973,7 +998,7 @@ fn rebuild_reconcile_waits_for_the_whole_build_subtree() {
 #[test]
 fn a_fanned_out_mechanical_node_declares_its_agent_lease() {
     let q = JobQueue::new(4);
-    let id = submit(&q, "fan-out", |builder| {
+    insert(&q, |builder| {
         templates::fanned_out_mechanical(
             builder,
             NodeKind::Start {
@@ -981,9 +1006,9 @@ fn a_fanned_out_mechanical_node_declares_its_agent_lease() {
             },
         );
     });
-    assert_eq!(declared_shape(&q, id), vec![row("start", None, &[])]);
+    assert_eq!(declared_shape(&q), vec![row("start", None, &[])]);
     assert_eq!(
-        declared_resources(&q, node_of(&q, id, "start")),
+        declared_resources(&q, node_of(&q, "start")),
         vec![Resource::Agent("agent-a".to_owned())],
         "the fanned-out node carries the lease itself, rather than relying on \
          whoever happened to fan it out"
@@ -1008,19 +1033,13 @@ fn a_fanned_out_mechanical_node_declares_its_agent_lease() {
 fn a_meta_lock_grows_one_rebuild_subgraph_per_agent() {
     let q = JobQueue::new(4);
     let agents = vec!["alice".to_owned(), "bob".to_owned()];
-    let id = q
-        .submit(
-            Source::AutoUpdate,
-            "sweep".to_owned(),
-            |builder: &JobBuilder| {
-                templates::grown_graceful_rebuilds(builder, &agents, true);
-            },
-        )
-        .expect("valid shape");
+    insert(&q, |builder| {
+        templates::grown_graceful_rebuilds(builder, &agents, true);
+    });
 
     // One chain per agent, each an independent group root — so the two rebuild
     // concurrently, each on its own lease.
-    let shape = declared_shape(&q, id);
+    let shape = declared_shape(&q);
     let heads: Vec<_> = shape
         .iter()
         .filter(|d| d.kind == "meta_sync")
@@ -1046,21 +1065,40 @@ fn a_meta_lock_grows_one_rebuild_subgraph_per_agent() {
 #[test]
 fn cancel_clears_queued_dag() {
     let q = JobQueue::new(1);
-    let id = submit(&q, "r", |builder| rebuild(builder, "agent-a"));
-    assert!(q.cancel(id), "fully-queued dag cancels");
-    // The operator sees `Cancelled` the moment the cancel returns — the spared
-    // tail is still `Pending`, and a DAG must not read `Queued` back to the
-    // operator who just cancelled it (the dashboard renders this roll-up from
-    // the snapshot `post_rebuild_queue_cancel` emits synchronously).
-    assert_eq!(state_of(&q, id), State::Cancelled, "no stale Queued gap");
-    // Neither `EmitRebuilt` tail accepts a *dropped* dependency — the ok one is
-    // `AFTER_OK`, the failure one keys on elimination — so both are cancelled
-    // with the work and **nothing is left that could still run**: no node is
-    // spared, so a rebuild that never ran emits nothing.
+    let roots = insert_named(&q, |builder| rebuild(builder, "agent-a"));
+    let [head, brace, tail] = roots.as_slice() else {
+        panic!("a rebuild names three roots, got {roots:?}")
+    };
+    assert!(q.cancel(*head), "fully-queued dag cancels");
+    // The operator sees `Cancelled` the moment the cancel returns — a group must
+    // not read `Queued` back to the operator who just cancelled it (the
+    // dashboard renders this roll-up from the snapshot
+    // `post_rebuild_queue_cancel` emits synchronously).
+    assert_eq!(state_of(&q, *head), State::Cancelled, "no stale Queued gap");
+    // Both `EmitRebuilt` tails go with it — the ok one is `AFTER_OK`, the
+    // failure one keys on elimination — so a rebuild that never ran emits
+    // nothing.
+    //
+    // 🎯 **But `Reconcile` survives, and that is the finding.** Its edge onto
+    // the brace accepts `done|failed|skipped`, and a cancel-cascade *skips* the
+    // brace rather than cancelling it — so the edge is satisfied and the tail
+    // stays claimable. With a container above them all, one cancel took the
+    // whole group; a job is its roots now, and dropping it means dropping every
+    // id the insert returned. That is what the ids are for.
+    assert_eq!(
+        pending_kinds(&q),
+        vec!["reconcile"],
+        "the convergence tail outlives its head's cancel"
+    );
     assert!(
-        pending_kinds(&q, id).is_empty(),
-        "a dropped rebuild leaves nothing alive, got {:?}",
-        pending_kinds(&q, id)
+        !q.cancel(*brace),
+        "the brace was eliminated with the head — there is nothing left to cancel"
+    );
+    assert!(q.cancel(*tail), "the surviving tail cancels on its own id");
+    assert!(
+        pending_kinds(&q).is_empty(),
+        "cancelling every named root leaves nothing alive, got {:?}",
+        pending_kinds(&q)
     );
 }
 
@@ -1075,36 +1113,26 @@ fn cancel_clears_queued_dag() {
 #[test]
 fn cancel_drops_one_agents_branch_leaving_the_rest() {
     let q = JobQueue::new(2);
-    let id = submit(&q, "r", |builder| {
-        restart_online(builder, &["agent-a", "agent-b"], false);
+    let roots = insert_named(&q, |builder| {
+        restart_online(builder, &["agent-a", "agent-b"], false)
     });
-    // Per-agent subgraphs hang directly off the container, one per agent.
-    // ⚠️ `parent` is the *graph* parent here, not a DAG-relative one: what
-    // the typed view called a parentless group root is a direct child of the
-    // container node in the generic view.
-    let snap = q.node_subtrees(&[id]);
-    let a_root = snap
-        .iter()
-        .find(|n| {
-            n.parent == Some(id)
-                && n.payload
-                    .data
-                    .get("agent")
-                    .and_then(serde_json::Value::as_str)
-                    == Some("agent-a")
-        })
-        .expect("agent-a has a group root");
+    // One root per agent, **in the order the chain named them** — that ordering
+    // is `insert_job`'s contract, and it is what replaced digging the right
+    // subgraph out of a snapshot by matching on its payload's agent field.
+    let [a_root, _b_root] = roots.as_slice() else {
+        panic!("a two-agent restart names one root per agent, got {roots:?}")
+    };
 
-    assert!(q.cancel(a_root.id), "an interior/group root cancels alone");
+    assert!(q.cancel(*a_root), "an interior/group root cancels alone");
 
     // agent-a's subgraph is gone; agent-b's is untouched and still alive.
     assert!(
-        pending_kinds_for(&q, id, "agent-a").is_empty(),
+        pending_kinds_for(&q, "agent-a").is_empty(),
         "agent-a's branch was dropped whole, got {:?}",
-        pending_kinds_for(&q, id, "agent-a")
+        pending_kinds_for(&q, "agent-a")
     );
     assert!(
-        !pending_kinds_for(&q, id, "agent-b").is_empty(),
+        !pending_kinds_for(&q, "agent-b").is_empty(),
         "agent-b's branch survives its sibling's cancel"
     );
 }
@@ -1124,23 +1152,26 @@ fn cancel_drops_one_agents_branch_leaving_the_rest() {
 /// deliberately-stopped as far as reconcile and crash-watch are concerned.
 #[test]
 fn cancelled_power_op_runs_no_compensating_node() {
-    /// Submit-cancel-assert for one power op. Taking the already-submitted DAG
-    /// id is what removes the need to put three differently-typed recipes in
-    /// one array: each caller submits its own spec, so no closure type has to
-    /// be erased to a boxed one.
-    fn assert_cancels_clean(q: &JobQueue, id: u64, writes_intent: bool, case: &str) {
-        // Read the intent head off the submitted DAG rather than out of the
-        // spec: a declared job holds its own nodes and inserts them.
-        let has_intent = declared_shape(q, id).iter().any(|d| d.kind == "set_wanted");
+    /// Insert-cancel-assert for one power op. Taking the roots the job already
+    /// named is what removes the need to put three differently-typed recipes in
+    /// one array: each caller inserts its own, so no closure type has to be
+    /// erased to a boxed one.
+    fn assert_cancels_clean(q: &JobQueue, roots: &[u64], writes_intent: bool, case: &str) {
+        // Read the intent head off the inserted nodes rather than out of a
+        // spec: a declared job holds its own nodes and inserts them. The queue
+        // is fresh per case, so the whole graph is this one op.
+        let has_intent = declared_shape(q).iter().any(|d| d.kind == "set_wanted");
         assert_eq!(has_intent, writes_intent, "{case}: intent head");
-        assert!(q.cancel(id), "{case}: cancelled while queued");
-        assert_eq!(state_of(q, id), State::Cancelled);
+        for root in roots {
+            assert!(q.cancel(*root), "{case}: cancelled while queued");
+            assert_eq!(state_of(q, *root), State::Cancelled);
+        }
         // Nothing is left that *could* run. Asserting on the pending set rather
         // than on "what is ready this instant" also covers a node that is alive
         // but blocked — which is exactly what a leftover compensating node
         // would look like.
         assert_eq!(
-            pending_kinds(q, id),
+            pending_kinds(q),
             Vec::<&str>::new(),
             "{case}: a power op emits no tail node, so a cancelled one leaves nothing"
         );
@@ -1152,22 +1183,20 @@ fn cancelled_power_op_runs_no_compensating_node() {
             let case = format!("graceful={graceful} running={running}");
 
             let q = JobQueue::new(1);
-            let id = submit(&q, "bounce", |builder| {
-                submit::restart_nodes(builder, &targets, graceful);
+            let roots = insert_named(&q, |builder| {
+                power::restart_nodes(builder, &targets, graceful)
             });
-            assert_cancels_clean(&q, id, false, &format!("restart {case}"));
+            assert_cancels_clean(&q, &roots, false, &format!("restart {case}"));
 
             let q = JobQueue::new(1);
-            let id = submit(&q, "stop", |builder| {
-                submit::stop_nodes(builder, &targets, graceful);
-            });
-            assert_cancels_clean(&q, id, true, &format!("stop {case}"));
+            let roots = insert_named(&q, |builder| power::stop_nodes(builder, &targets, graceful));
+            assert_cancels_clean(&q, &roots, true, &format!("stop {case}"));
 
             let q = JobQueue::new(1);
-            let id = submit(&q, "start", |builder| {
-                submit::start_nodes(builder, &[("agent-a".to_owned(), running, false)]);
+            let roots = insert_named(&q, |builder| {
+                power::start_nodes(builder, &[("agent-a".to_owned(), running, false)])
             });
-            assert_cancels_clean(&q, id, true, &format!("start {case}"));
+            assert_cancels_clean(&q, &roots, true, &format!("start {case}"));
         }
     }
 }
@@ -1185,16 +1214,21 @@ fn cancelled_power_op_runs_no_compensating_node() {
 #[test]
 fn cancelled_dag_still_runs_its_approval_tail() {
     let q = JobQueue::new(1);
-    let id = submit(&q, "approval #7", |builder| {
+    insert(&q, |builder| {
         templates::approval_deploy(builder, "agent-a", 7);
     });
-    assert!(q.cancel(id), "fully-queued dag cancels");
+    // Found by kind, not returned: `approval_deploy` deliberately names
+    // nothing, because nothing polls it — the approval row is how an operator
+    // follows a deploy, so the template is fire-and-forget in production and a
+    // test must not make it return an id it would otherwise have no use for.
+    let window = node_of(&q, "deploy_window").get();
+    assert!(q.cancel(window), "fully-queued dag cancels");
     // The `Cancelled` tail is the only node whose edge accepts a dropped
     // dependency, so it is the only one `cancel` spares — and *which* tail
     // survives is the whole assertion: the template emits one per outcome and
     // the spared one names how the approval row is about to be resolved.
     // Nothing computes it, so reading the survivor is reading the answer.
-    let spared = pending_payloads(&q, id);
+    let spared = pending_payloads(&q);
     assert!(
         matches!(
             spared.as_slice(),
@@ -1205,18 +1239,20 @@ fn cancelled_dag_still_runs_its_approval_tail() {
         ),
         "only the cancelled-outcome tail is spared, got {spared:?}"
     );
-    // ⚠️ `Finishing`, not `Cancelled` — and the change is a **fix**, not a
-    // regression. This used to read the host-side `DagView::rollup_state`,
-    // which flattened the spared tail away and reported the group settled
-    // while a node of it was still pending. The root's own state is the
-    // scheduler's answer: `Finishing` means "own logic done, children still
-    // running", and the tail this test exists to protect *is* such a child.
-    // A group that still has work to do does not read terminal.
-    assert_eq!(state_of(&q, id), State::Finishing);
+    // ⚠️ `Cancelled`, and it reads terminal **while the spared tail is still
+    // pending** — the one place this differs from the container era, where the
+    // cancel landed on a node *above* the window and the window rolled up
+    // `Finishing`. Here the operator cancels the window itself, so its own
+    // state is `Cancelled` however its subtree is doing. Deliberately asserted
+    // rather than routed around: the group's card goes terminal while a
+    // bookkeeping node runs on. That is acceptable for the tail this test
+    // protects (it resolves the approval row and nothing waits on it), and it
+    // would not be for work an operator expects to still be watching.
+    assert_eq!(state_of(&q, window), State::Cancelled);
     // An unrelated group landing in the same graph doesn't disturb this one's
     // state — a root rolls up its own subtree, not the graph.
-    let _other = submit(&q, "r", |builder| rebuild(builder, "agent-b"));
-    assert_eq!(state_of(&q, id), State::Finishing);
+    let _other = insert_named(&q, |builder| rebuild(builder, "agent-b"));
+    assert_eq!(state_of(&q, window), State::Cancelled);
 }
 
 // ---- approval deploy subtree ----
@@ -1229,12 +1265,12 @@ fn cancelled_dag_still_runs_its_approval_tail() {
 #[test]
 fn deploy_dag_runs_phases_in_order_and_tails_a_failed_apply() {
     let q = JobQueue::new(1);
-    let id = submit(&q, "approval #7", |builder| {
+    insert(&q, |builder| {
         templates::approval_deploy(builder, "agent-a", 7);
     });
 
     assert_eq!(
-        declared_shape(&q, id),
+        declared_shape(&q),
         vec![
             // The window is the group root and holds the meta window for the
             // whole subtree; the three phases are its sub-nodes.
@@ -1287,12 +1323,12 @@ fn deploy_apply_grows_rebuild_subgraph_and_finalizes_after_it() {
     // children run (`a_completing_node_grows_the_work_it_declared`,
     // `parent_parks_in_finishing_until_children_roll_up`).
     let q = JobQueue::new(1);
-    let id = submit(&q, "deploy graft", |builder| {
+    insert(&q, |builder| {
         templates::deploy_rebuild_nodes(builder, "agent-a", 11);
     });
 
     assert_eq!(
-        declared_shape(&q, id),
+        declared_shape(&q),
         vec![
             row("meta_sync", None, &[]),
             row("agent_window", None, &[("meta_sync", "done")]),
@@ -1467,11 +1503,11 @@ fn error_truncation_cuts_on_a_char_boundary() {
 #[test]
 fn graceful_stop_shape_signal_drain_reconcile() {
     let q = JobQueue::new(1);
-    let id = submit(&q, "graceful", |builder| {
+    insert(&q, |builder| {
         stop_online(builder, &["agent-a"], true);
     });
     assert_eq!(
-        declared_shape(&q, id),
+        declared_shape(&q),
         vec![
             // The whole stop hangs under `set_wanted`: the durable intent is
             // written first, and the mechanical steps are its sub-nodes.
@@ -1494,8 +1530,8 @@ fn graceful_stop_shape_signal_drain_reconcile() {
     // `exec.rs`.
     assert_eq!(
         [
-            declared_resources(&q, node_of(&q, id, "signal")),
-            declared_resources(&q, node_of(&q, id, "drain")),
+            declared_resources(&q, node_of(&q, "signal")),
+            declared_resources(&q, node_of(&q, "drain")),
         ],
         [vec![], vec![]],
         "the quiesce pair borrows the brace's lease and declares nothing"
@@ -1505,11 +1541,11 @@ fn graceful_stop_shape_signal_drain_reconcile() {
 #[test]
 fn spawn_shape_provision_create_dropin_reconcile() {
     let q = JobQueue::new(1);
-    let id = submit(&q, "approval #7 spawn", |builder| {
+    insert(&q, |builder| {
         templates::spawn(builder, "newbie", 7);
     });
     assert_eq!(
-        declared_shape(&q, id),
+        declared_shape(&q),
         vec![
             row("provision", None, &[]),
             row("create", Some("provision"), &[]),
@@ -1530,7 +1566,7 @@ fn spawn_shape_provision_create_dropin_reconcile() {
 #[test]
 fn perm_change_shape_prefixes_rebuild_chain() {
     let q = JobQueue::new(1);
-    let id = submit(&q, "perm", |builder| {
+    insert(&q, |builder| {
         templates::perm_change(
             builder,
             "agent-a",
@@ -1541,7 +1577,7 @@ fn perm_change_shape_prefixes_rebuild_chain() {
         );
     });
     assert_eq!(
-        declared_shape(&q, id)
+        declared_shape(&q)
             .iter()
             .map(|d| d.kind)
             .collect::<Vec<_>>(),
@@ -1569,15 +1605,15 @@ fn reparent_shape_is_a_lone_agentless_meta_window_node() {
     // `MetaLock`, and it must declare the meta window — a topology commit
     // must not land inside another node's staged deploy window.
     let q = JobQueue::new(1);
-    let id = submit(&q, "set-parent", |builder| {
+    insert(&q, |builder| {
         templates::reparent(builder, vec![(ident("alice"), Some(ident("bob")))]);
     });
     assert_eq!(
-        declared_shape(&q, id),
+        declared_shape(&q),
         vec![row("reparent", None, &[])],
         "one node, no rebuild subgraph"
     );
-    let node = node_of(&q, id, "reparent");
+    let node = node_of(&q, "reparent");
     assert_eq!(
         declared_resources(&q, node),
         vec![Resource::MetaWindow],
@@ -1593,15 +1629,15 @@ fn reparent_bulk_shape_carries_every_move_on_one_node() {
     // request is the reason a single node was chosen in the first place.
     let moves = vec![(ident("alice"), Some(ident("bob"))), (ident("carol"), None)];
     let q = JobQueue::new(1);
-    let id = submit(&q, "set-parent-bulk", |builder| {
+    insert(&q, |builder| {
         templates::reparent(builder, moves.clone());
     });
     assert_eq!(
-        declared_shape(&q, id),
+        declared_shape(&q),
         vec![row("reparent", None, &[])],
         "one node for the whole request, not one per move"
     );
-    let NodeKind::Reparent { moves: got } = payload_of(&q, id, "reparent") else {
+    let NodeKind::Reparent { moves: got } = payload_of(&q, "reparent") else {
         panic!("expected a Reparent node");
     };
     assert_eq!(got, moves, "every move rides the single node");
diff --git a/hive-c0re/src/server.rs b/hive-c0re/src/server.rs
index 26799a9a..8e534ba1 100644
--- a/hive-c0re/src/server.rs
+++ b/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 {
diff --git a/hive-c0re/src/socket_server/lifecycle_handlers.rs b/hive-c0re/src/socket_server/lifecycle_handlers.rs
index 46446c12..c0261405 100644
--- a/hive-c0re/src/socket_server/lifecycle_handlers.rs
+++ b/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
 }
 
diff --git a/hive-c0re/src/workers/auto_update.rs b/hive-c0re/src/workers/auto_update.rs
index 8132f0b7..0084b102 100644
--- a/hive-c0re/src/workers/auto_update.rs
+++ b/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();
 }
diff --git a/hive-host-sock/src/jobs.rs b/hive-host-sock/src/jobs.rs
index 7502277b..7930807e 100644
--- a/hive-host-sock/src/jobs.rs
+++ b/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.