refactor(#2449): write power intent via a SetWanted DAG node, not a pre-submit side effect

The durable 'wanted' power intent was written by submit::{start,stop,
restart,graceful_restart,graceful_stop} as a synchronous pre-submit side
effect, then read by the DAG's tail Reconcile. That's not crash-safe
(a crash between the write and the enqueue loses it) and, with agent now
per-node, can't be per-agent in a DAG that spans agents.

Move it into the DAG as a head SetWanted node:
- NodeKind::SetWanted { up } + run_set_wanted executor (fails the node on
  a write error, unlike the old warn-and-continue, so a stale intent
  never reaches Reconcile).
- LEASE-NEEDING, not lease-exempt: it takes the agent lease so a power-op
  DAG's intent-write + reconcile is atomic per-agent. If it were exempt,
  two racing ops (restart vs stop) would run both intent-writes up front
  and clobber each other before either reconciled — defeating the point
  of moving the write into the DAG. (In stale_start the lease is thus held
  across the head Prebuild, but that's a no-op there: the agent is down so
  prebuild is skipped.)
- templates: explicit SetWanted node 0 on restart/graceful_restart/
  graceful_stop, plus dedicated start/stop templates (SetWanted -> Reconcile)
  and stale_start (SetWanted(Up) -> rebuild subgraph, reusing rebuild_nodes).
  No compose helper / rebuild variant. reconcile_only is now boot-only.
- submit.rs: drop the set_wanted side effect; the stale-rev shape decision
  (start vs stale_start) stays submit-side.

All 33 job_queue tests pass (shape/lease tests updated for the head node).

hive-c0re/src/job_queue/submit.rs

@@ -1,16 +1,15 @@
 //! Request-level submit API — the surface the dashboard POST handlers,
-//! the MCP socket handlers, and `hivectl` paths call. Owns the
-//! submit-time side effects the DAG templates deliberately don't:
-//! writing the durable `wanted` power intent (synchronously,
-//! last-writer-wins) before the DAG whose `Reconcile` reads it, and
-//! upgrading a stale start to a full rebuild. Every helper emits a
-//! fresh queue snapshot so the dashboard shows the new DAG immediately.
+//! the MCP socket handlers, and `hivectl` paths call. The durable
+//! `wanted` power intent is now written by a `SetWanted` DAG node at the
+//! head of each power-op template (not a pre-submit side effect); the
+//! only submit-time logic left is the stale-start *shape* decision
+//! (`start` vs `stale_start`). Every helper emits a fresh queue snapshot
+//! so the dashboard shows the new DAG immediately.
 
 use std::sync::Arc;
 
-use super::{Source, Template, templates};
+use super::{Source, templates};
 use crate::coordinator::Coordinator;
-use crate::power::Wanted;
 
 fn submit_and_emit(coord: &Arc<Coordinator>, spec: super::DagSpec) -> u64 {
     let id = coord
@@ -21,12 +20,6 @@ fn submit_and_emit(coord: &Arc<Coordinator>, spec: super::DagSpec) -> u64 {
     id
 }
 
-fn set_wanted(coord: &Arc<Coordinator>, agent: &str, wanted: Wanted) {
-    if let Err(e) = coord.power.set(agent, wanted) {
-        tracing::warn!(%agent, wanted = wanted.as_str(), error = ?e, "agent_power: set failed");
-    }
-}
-
 /// Manual/approval-independent rebuild (always relocks the agent's
 /// meta input — cascade children are built by the scheduler's fan-out
 /// instead of this surface).
@@ -35,20 +28,20 @@ pub fn rebuild(coord: &Arc<Coordinator>, agent: &str, source: Source, reason: St
 }
 
 /// Restart: mechanical stop + converge to `wanted = Up`. The intent
-/// write matters when `wanted` drifted `Offline` under a running
-/// agent — the old `kill + start` always ended up, and an operator
-/// asking for a restart plainly wants it running, not a stop.
+/// write is the template's head `SetWanted(Up)` node — it matters when
+/// `wanted` drifted `Offline` under a running agent (an operator asking
+/// for a restart plainly wants it running, not a stop).
 pub fn restart(coord: &Arc<Coordinator>, agent: &str, source: Source, reason: String) -> u64 {
-    set_wanted(coord, agent, Wanted::Up);
     submit_and_emit(coord, templates::restart(agent, source, reason))
 }
 
-/// Start: persist `wanted = Up`, then reconcile. A stale rev marker
-/// upgrades the start to a full rebuild (whose tail `Reconcile` does
-/// the start) so the container always comes up on current derivations
-/// — the old fast-lane `run_start` upgrade, moved to submit time.
+/// Start: `SetWanted(Up)` (a DAG node now) then reconcile. A stale rev
+/// marker upgrades the start to a rebuild-then-start (`stale_start`, whose
+/// tail `Reconcile` does the start) so the container always comes up on
+/// current derivations — the old fast-lane `run_start` upgrade, still a
+/// submit-time *shape* decision (which template), while the intent write
+/// itself is now the template's head node.
 pub fn start(coord: &Arc<Coordinator>, agent: &str, source: Source, reason: String) -> u64 {
-    set_wanted(coord, agent, Wanted::Up);
     let stored = std::fs::read_to_string(crate::paths::applied_rev_marker(agent)).ok();
     let stale = crate::auto_update::current_flake_rev(&coord.hyperhive_flake)
         .is_some_and(|rev| stored.as_deref() != Some(rev.as_str()));
@@ -56,60 +49,34 @@ pub fn start(coord: &Arc<Coordinator>, agent: &str, source: Source, reason: Stri
         tracing::info!(%agent, "start: rev stale — upgrading to rebuild+start");
         return submit_and_emit(
             coord,
-            templates::rebuild(
-                agent,
-                source,
-                format!("{reason} (stale — rebuild+start)"),
-                None,
-                true,
-            ),
+            templates::stale_start(agent, source, format!("{reason} (stale — rebuild+start)")),
         );
     }
-    submit_and_emit(
-        coord,
-        templates::reconcile_only(
-            Template::Start,
-            agent,
-            source,
-            reason,
-            Some(crate::coordinator::TransientKind::Starting),
-        ),
-    )
+    submit_and_emit(coord, templates::start(agent, source, reason))
 }
 
-/// Hard stop: persist `wanted = Offline`, then reconcile (kill +
+/// Hard stop: `SetWanted(Offline)` (a DAG node now) then reconcile (kill +
 /// unregister + `Killed` event).
 pub fn stop(coord: &Arc<Coordinator>, agent: &str, source: Source, reason: String) -> u64 {
-    set_wanted(coord, agent, Wanted::Offline);
-    submit_and_emit(
-        coord,
-        templates::reconcile_only(
-            Template::Stop,
-            agent,
-            source,
-            reason,
-            Some(crate::coordinator::TransientKind::Stopping),
-        ),
-    )
+    submit_and_emit(coord, templates::stop(agent, source, reason))
 }
 
-/// Graceful stop: persist `wanted = Offline`, then signal → drain →
-/// reconcile (the actual stop).
+/// Graceful stop: signal → drain → reconcile (the actual stop). The head
+/// `SetWanted(Offline)` node writes the intent as part of the DAG.
 pub fn graceful_stop(coord: &Arc<Coordinator>, agent: &str, source: Source, reason: String) -> u64 {
-    set_wanted(coord, agent, Wanted::Offline);
     submit_and_emit(coord, templates::graceful_stop(agent, source, reason))
 }
 
-/// Graceful restart: persist `wanted = Up`, then signal → drain →
-/// mechanical stop → reconcile (starts it back up) — one atomic DAG,
-/// no client-side "await the stop DAG then submit a start DAG" split.
+/// Graceful restart: signal → drain → mechanical stop → reconcile (starts
+/// it back up) — one atomic DAG, no client-side "await the stop DAG then
+/// submit a start DAG" split. The head `SetWanted(Up)` node writes the
+/// intent as part of the DAG.
 pub fn graceful_restart(
     coord: &Arc<Coordinator>,
     agent: &str,
     source: Source,
     reason: String,
 ) -> u64 {
-    set_wanted(coord, agent, Wanted::Up);
     submit_and_emit(coord, templates::graceful_restart(agent, source, reason))
 }