jobq: one insertion entry point, and make it atomic

Three findings from the operator's review, all correct.

1. Two insert_job's. Graph::insert_job had no caller outside hive-jobq's
   own tests -- production only ever went through Scheduler::insert_job.
   It existed because the graph-level one got written first. Deleted; the
   tests moved onto a Scheduler, which is where insertion belongs anyway.

2. insert_job was not atomic, and the previous commit made that worse: a
   forward edge or forward parent surfaced mid-loop, leaving the nodes
   before it in the graph, and resolve_wanted ran after every insert, so
   an unknown handle failed once the whole job was already committed.
   The module documented this under "Partial insertion" instead of fixing
   it -- prose describing a hole is not a design.

   All three are decidable from what the builder holds, so
   check_declaration_order now runs before the first insert and the loop
   indexes ids directly. A malformed job leaves the graph untouched.
   What remains mid-insert is the graph's own rejection (out-of-group
   dep, empty DepWhen); closing that needs a dry-run validate on Graph,
   which is a separate change.

3. DagSpec no longer boxes its recipe: it is generic over the closure,
   which travels from the template that built it straight into submit.
   The box bought type inference, and paying for it costs annotations --
   `|b: &Job|` at each declaration site (the field needs an HRTB, and an
   unannotated closure binds one lifetime) and `+ use<>` on each
   returning signature (or the opaque type captures the caller's borrows).
   Erasure is still needed where several recipe shapes share one type:
   the boxed Declare stays for the executor's append_subgraph, and a test
   table uses an erase() helper.

hive-c0re/src/job_queue/tests.rs

@@ -9,15 +9,29 @@
 use super::model::NodeKind;
 use super::*;
 
-fn submit(q: &JobQueue, spec: DagSpec) -> u64 {
+fn submit<F: FnOnce(&Job)>(q: &JobQueue, spec: DagSpec<F>) -> u64 {
     q.submit(spec).expect("valid spec")
 }
 
+/// Erase a spec's recipe to the boxed [`Declare`] so specs of *different*
+/// shapes can share one type — e.g. a table of `(name, spec)` cases.
+///
+/// Production never needs this: each submit path builds one spec and hands it
+/// straight to `submit`, so the concrete closure type is known end to end. A
+/// test table is the case where several shapes must be one type.
+fn erase<F: FnOnce(&Job) + Send + 'static>(spec: DagSpec<F>) -> DagSpec<Declare> {
+    DagSpec {
+        source: spec.source,
+        reason: spec.reason,
+        declare: Box::new(spec.declare),
+    }
+}
+
 fn ident(s: &str) -> hive_types::Ident {
     hive_types::Ident::parse(s).expect("valid test ident")
 }
 
-fn rebuild(agent: &str, reason: &str) -> DagSpec {
+fn rebuild(agent: &str, reason: &str) -> DagSpec<impl FnOnce(&Job) + use<>> {
     templates::rebuild(agent, Source::Manual, reason.to_owned(), true)
 }
 
@@ -25,14 +39,22 @@ fn rebuild(agent: &str, reason: &str) -> DagSpec {
 /// 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_spec`).
-fn restart_online(agents: &[&str], graceful: bool, reason: &str) -> DagSpec {
+fn restart_online(
+    agents: &[&str],
+    graceful: bool,
+    reason: &str,
+) -> DagSpec<impl FnOnce(&Job) + use<>> {
     let targets: Vec<(String, bool)> = agents.iter().map(|a| ((*a).to_owned(), true)).collect();
     submit::restart_spec(&targets, graceful, Source::Manual, reason.to_owned())
 }
 
 /// Stop DAG spec with every agent treated as **running** — the online shape
 /// (`SetWanted → [Signal→Drain→](graceful) Reconcile`).
-fn stop_online(agents: &[&str], graceful: bool, reason: &str) -> DagSpec {
+fn stop_online(
+    agents: &[&str],
+    graceful: bool,
+    reason: &str,
+) -> DagSpec<impl FnOnce(&Job) + use<>> {
     let targets: Vec<(String, bool)> = agents.iter().map(|a| ((*a).to_owned(), true)).collect();
     submit::stop_spec(&targets, graceful, Source::Manual, reason.to_owned())
 }
@@ -197,7 +219,7 @@ fn graceful_rebuild_chain_drains_before_stopping() {
         DagSpec {
             source: Source::AutoUpdate,
             reason: "sweep".to_owned(),
-            declare: Box::new(|b| {
+            declare: Box::new(|b: &Job| {
                 templates::rebuild_nodes(
                     b,
                     "agent-a",
@@ -246,7 +268,7 @@ fn non_graceful_rebuild_has_no_signal_or_drain() {
         DagSpec {
             source: Source::Manual,
             reason: "manual".to_owned(),
-            declare: Box::new(|b| {
+            declare: Box::new(|b: &Job| {
                 templates::rebuild_nodes(
                     b,
                     "agent-a",
@@ -695,7 +717,7 @@ fn append_subgraph_roots_on_emitter_and_rebases_local_deps() {
     let spec = DagSpec {
         source: Source::AutoUpdate,
         reason: "sweep".to_owned(),
-        declare: Box::new(|b| {
+        declare: Box::new(|b: &Job| {
             let _lock = templates::node(
                 b,
                 NodeKind::MetaLock {
@@ -715,7 +737,7 @@ fn append_subgraph_roots_on_emitter_and_rebases_local_deps() {
     // match the sweep arm of `run_meta_lock` or this stops tracking production.
     let subgraph = |agent: &str| -> Declare {
         let agent = agent.to_owned();
-        Box::new(move |b| {
+        Box::new(move |b: &Job| {
             templates::rebuild_nodes(
                 b,
                 &agent,
@@ -821,7 +843,7 @@ fn meta_update_grows_cascade_in_dag() {
     // Simulate the executor growing the cascade in-DAG (`relock = false` — a
     // cascade child must not re-lock and revert the parent's bump).
     for agent in ["alice", "bob"] {
-        let declare: Declare = Box::new(move |b| {
+        let declare: Declare = Box::new(move |b: &Job| {
             templates::rebuild_nodes(
                 b,
                 agent,
@@ -1073,25 +1095,37 @@ fn cancelled_power_op_runs_no_compensating_node() {
     for graceful in [false, true] {
         for running in [false, true] {
             let targets = vec![("agent-a".to_owned(), running)];
+            // Erased to `DagSpec<Declare>`: three different recipe types have to
+            // sit in one array.
             let cases = [
                 (
                     "restart",
                     false,
-                    submit::restart_spec(&targets, graceful, Source::Manual, "bounce".to_owned()),
+                    erase(submit::restart_spec(
+                        &targets,
+                        graceful,
+                        Source::Manual,
+                        "bounce".to_owned(),
+                    )),
                 ),
                 (
                     "stop",
                     true,
-                    submit::stop_spec(&targets, graceful, Source::Manual, "stop".to_owned()),
+                    erase(submit::stop_spec(
+                        &targets,
+                        graceful,
+                        Source::Manual,
+                        "stop".to_owned(),
+                    )),
                 ),
                 (
                     "start",
                     true,
-                    submit::start_spec(
+                    erase(submit::start_spec(
                         &[("agent-a".to_owned(), running, false)],
                         Source::Manual,
                         "start".to_owned(),
-                    ),
+                    )),
                 ),
             ];
             for (name, writes_intent, spec) in cases {