swarm-controller: declare a new agent paused at creation

Creating an agent queued its identity, forge repo and deploy, but never
wrote a wanted-state declaration for it — so the agent showed up in
swarm-ui as "no declaration", and the hive brought it up with nothing
saying whether it should be driving turns.

Add a `SetAgentWanted` job node that declares the agent `Paused` in its
hive's wanted-state bucket, using the same `WantedWriter::set` primitive
the per-agent state HTTP handler already uses. A fresh agent therefore
sits paused until the operator explicitly flips it to `Up`.

The node is a root — it needs only the hive and agent names known at
request time — but the deploy trigger now waits on it, so the pause is
in the store before the hive brings the container up rather than landing
some time after a freshly deployed agent has already started taking
turns.

swarm-controller/src/main.rs

@@ -105,6 +105,12 @@ enum SwarmNodeKind {
     /// its hive's shared queue credential, so the document cannot be rendered
     /// without knowing which hive the agent belongs to.
     MintAgentIdentity { hive: String, agent: String },
+    /// Declare `agent` on `hive` as `Paused` in the swarm's wanted-state
+    /// store, so a freshly created agent does not start driving turns the
+    /// moment it's deployed — the operator has to explicitly flip it to `Up`.
+    /// Carries the hive for the same reason `TriggerDeploy`/`MintAgentIdentity`
+    /// do: the wanted-state bucket is keyed per hive.
+    SetAgentWanted { hive: String, agent: String },
     /// Tell `hive` to rebuild `agent`, by publishing on the swarm's deploy
     /// subject. The one node kind whose effect leaves this host.
     ///
@@ -124,6 +130,7 @@ impl hive_jobq_wire::WireNode for SwarmNodeKind {
             SwarmNodeKind::AddRepoMember { .. } => "add_repo_member".to_owned(),
             SwarmNodeKind::InitAgentConfigRepo { .. } => "init_agent_config_repo".to_owned(),
             SwarmNodeKind::MintAgentIdentity { .. } => "mint_agent_identity".to_owned(),
+            SwarmNodeKind::SetAgentWanted { .. } => "set_agent_wanted".to_owned(),
             SwarmNodeKind::TriggerDeploy { .. } => "trigger_deploy".to_owned(),
         }
     }
@@ -144,7 +151,8 @@ impl hive_jobq_wire::WireNode for SwarmNodeKind {
                 serde_json::json!({ "agent": agent })
             }
             SwarmNodeKind::TriggerDeploy { hive, agent }
-            | SwarmNodeKind::MintAgentIdentity { hive, agent } => {
+            | SwarmNodeKind::MintAgentIdentity { hive, agent }
+            | SwarmNodeKind::SetAgentWanted { hive, agent } => {
                 serde_json::json!({ "agent": agent, "hive": hive })
             }
         }
@@ -186,8 +194,22 @@ struct WorkerDeps {
     /// a per-node chance to read one. `None` on a host the operator has not
     /// given an authority — see `agent_identity::Authority::from_env`.
     agent_ca: Option<std::sync::Arc<agent_identity::Authority>>,
+    /// The wanted-state writer, for nodes that declare what a hive should
+    /// converge an agent to. Shares the status reader's queue connection —
+    /// see `wanted_writer`. `None` exactly when no swarm queue is configured
+    /// on this host, same as `queue`.
+    wanted: Option<std::sync::Arc<wanted::WantedWriter>>,
 }
 
+/// What [`SwarmNodeKind::SetAgentWanted`] declares a brand-new agent to be.
+///
+/// A named constant rather than the literal inline, because this *is* the
+/// decision the node exists to carry — writing a declaration at all is the
+/// mechanism, `Paused` is the policy — and a test can pin a name where it
+/// cannot pin a value written into a KV bucket no unit test has a server for.
+const NEW_AGENT_WANTED_STATE: swarm_queue_client::wanted::AgentState =
+    swarm_queue_client::wanted::AgentState::Paused;
+
 /// Run a claimed node's actual work. Mirrors `hive-c0re/src/job_queue/
 /// exec.rs::run_node`'s role exactly — the one place a `SwarmNodeKind`
 /// variant turns into a real effect.
@@ -286,6 +308,14 @@ async fn run_swarm_node(
                 }
             }
         },
+        SwarmNodeKind::SetAgentWanted { hive, agent } => match deps.wanted {
+            None => Outcome::Failed(
+                "no swarm queue is configured on this host, so no wanted-state \
+                 declaration can be written"
+                    .to_owned(),
+            ),
+            Some(writer) => declare_new_agent(&writer, &hive, &agent).await,
+        },
         SwarmNodeKind::TriggerDeploy { hive, agent } => match deps.queue {
             None => Outcome::Failed(
                 "no swarm queue is configured on this host, so no hive can be told to deploy"
@@ -297,6 +327,25 @@ async fn run_swarm_node(
     (builder, outcome)
 }
 
+/// Declare a brand-new agent at [`NEW_AGENT_WANTED_STATE`].
+///
+/// A named function beside [`publish_deploy`] rather than the two lines
+/// inline, for the same reason that one is: `run_swarm_node`'s match is a
+/// per-variant index, and every arm that grows a body past a call pushes the
+/// next reader further from the variant they came to read.
+async fn declare_new_agent(
+    writer: &wanted::WantedWriter,
+    hive: &str,
+    agent: &str,
+) -> hive_jobq::scheduler::Outcome {
+    use hive_jobq::scheduler::Outcome;
+
+    match writer.set(hive, agent, NEW_AGENT_WANTED_STATE).await {
+        Ok(_) => Outcome::Done,
+        Err(e) => Outcome::Failed(format!("{e:#}")),
+    }
+}
+
 /// Publish one deploy request and report whether it left this process.
 ///
 /// The flush is not belt-and-braces: `publish` hands the message to the
@@ -1316,6 +1365,15 @@ fn declare_agent_job(
             agent: agent.to_owned(),
         })
         .after_ok(create_identity);
+    // Declared before the deploy trigger so the pause is visible in the
+    // wanted-state store before the hive brings the container up — see the
+    // node's own doc comment for why "before", not just "eventually". Needs
+    // no parent of its own: the declaration is written from the hive/agent
+    // names known at request time, same as `create_forge_user`.
+    let set_wanted = b.node(SwarmNodeKind::SetAgentWanted {
+        hive: hive.to_owned(),
+        agent: agent.to_owned(),
+    });
     // Last, and specifically after the config repo is seeded: the hive
     // deploys by reading that repo, so a deploy asked for any earlier would
     // find nothing to build. This is the edge that makes creating an agent at
@@ -1327,13 +1385,21 @@ fn declare_agent_job(
     // overtake the mint — but a host with no authority configured must still
     // create agents exactly as it does today. `after_ok` there would turn an
     // unconfigured option into an agent nobody runs.
+    //
+    // `after_ok` on the declaration, though: an agent deployed without its
+    // pause landing first is the race this node exists to close, and the
+    // only way the declaration fails without a live queue is a host that has
+    // no queue at all — on which `TriggerDeploy` has nothing to publish to
+    // either, so the strict edge cancels a node that could not have
+    // succeeded.
     let _trigger_deploy = b
         .node(SwarmNodeKind::TriggerDeploy {
             hive: hive.to_owned(),
             agent: agent.to_owned(),
         })
         .after_ok(init_config)
-        .after_any(mint_identity);
+        .after_any(mint_identity)
+        .after_ok(set_wanted);
     vec![create_identity.guid()]
 }
 
@@ -1664,6 +1730,7 @@ async fn main() -> Result<()> {
         forge: forge_client.clone(),
         queue: status.as_ref().map(|s| s.queue_client()),
         agent_ca: load_agent_authority(),
+        wanted: wanted_writer(status.as_ref()),
     };
 
     let jobq = Arc::new(Mutex::new(hive_jobq::scheduler::Scheduler::new(
@@ -1882,8 +1949,9 @@ mod tests {
             }]),
             links: std::sync::Arc::new(Vec::new()),
             status: None,
-            // No queue, for the same reason as `status`: these tests drive
-            // agent creation, which publishes no declaration.
+            // No queue, for the same reason as `status`: these tests build
+            // the creation graph and read its shape, and never run the node
+            // that would write a declaration.
             wanted: None,
             agent_status: None,
             jobq: std::sync::Arc::clone(&sched),
@@ -2031,6 +2099,7 @@ mod tests {
             forge: None,
             queue: None,
             agent_ca: None,
+            wanted: None,
         };
         let runner =
             hive_jobq::scheduler::Scheduler::claim_next(&sched, move |id, kind, builder| {
@@ -2084,6 +2153,7 @@ mod tests {
             forge: None,
             queue: None,
             agent_ca: None,
+            wanted: None,
         };
         let runner =
             hive_jobq::scheduler::Scheduler::claim_next(&sched, move |id, kind, builder| {
@@ -2137,6 +2207,7 @@ mod tests {
             forge: None,
             queue: None,
             agent_ca: None,
+            wanted: None,
         };
         let runner =
             hive_jobq::scheduler::Scheduler::claim_next(&sched, move |id, kind, builder| {
@@ -2159,6 +2230,137 @@ mod tests {
         );
     }
 
+    /// Fourth sibling, same shape and the same caveat once more: with no
+    /// queue configured this reaches only the graceful-absence-is-failure
+    /// branch. Writing the declaration for real needs a `JetStream` server,
+    /// which no test in this crate has — `wanted::apply`'s own unit tests
+    /// cover what gets written, and [`a_new_agent_is_declared_paused`] below
+    /// pins the state this node asks for.
+    #[tokio::test]
+    async fn set_agent_wanted_node_runs_end_to_end_and_fails_without_a_queue() {
+        let mut sched = hive_jobq::scheduler::Scheduler::new(
+            hive_jobq::Graph::new(),
+            hive_jobq::resources::ResourceTable::new(),
+        );
+        let id = sched
+            .append(
+                SwarmNodeKind::SetAgentWanted {
+                    hive: "pr1ma".to_owned(),
+                    agent: "atlas".to_owned(),
+                },
+                Vec::new(),
+                None,
+            )
+            .expect("insert");
+        let sched = std::sync::Arc::new(std::sync::Mutex::new(sched));
+
+        let deps = WorkerDeps {
+            auth: None,
+            forge: None,
+            queue: None,
+            agent_ca: None,
+            wanted: None,
+        };
+        let runner =
+            hive_jobq::scheduler::Scheduler::claim_next(&sched, move |id, kind, builder| {
+                run_swarm_node(id, kind, builder, deps)
+            })
+            .expect("the node just inserted is runnable");
+        runner
+            .await
+            .1
+            .expect("no growth declared, nothing to reject");
+
+        let guard = sched.lock().unwrap();
+        let node = guard.graph().node(id).expect("node still present");
+        assert_eq!(node.state, hive_jobq::State::Failed);
+        assert!(
+            node.error.as_deref().unwrap_or_default().contains("queue"),
+            "expected a no-queue-configured error, got {:?}",
+            node.error
+        );
+    }
+
+    /// The whole point of the node: a freshly created agent is declared
+    /// `Paused`, so it does not start driving turns the moment its hive
+    /// brings it up. Any other state here is the bug this fixes.
+    #[test]
+    fn a_new_agent_is_declared_paused() {
+        assert_eq!(
+            crate::NEW_AGENT_WANTED_STATE,
+            swarm_queue_client::wanted::AgentState::Paused
+        );
+    }
+
+    /// Same reasoning as `a_mint_node_renders_both_the_agent_and_the_hive`:
+    /// this variant carries a hive too, and has to have joined `data`'s
+    /// two-field or-pattern rather than the agent-only one.
+    #[test]
+    fn a_set_wanted_node_renders_both_the_agent_and_the_hive() {
+        use hive_jobq_wire::WireNode as _;
+
+        let kind = SwarmNodeKind::SetAgentWanted {
+            hive: "pr1ma".to_owned(),
+            agent: "atlas".to_owned(),
+        };
+        assert_eq!(kind.label(), "set_agent_wanted");
+        let data = kind.data(1);
+        assert_eq!(data["agent"], "atlas");
+        assert_eq!(data["hive"], "pr1ma");
+    }
+
+    /// The ordering the fix depends on: the pause has to be in the
+    /// wanted-state store *before* the hive is told to deploy, or a freshly
+    /// created agent runs undeclared for the window between the two.
+    ///
+    /// Asserted on the graph `create_agent` builds, same as the mint edge's
+    /// test above and for the same reason — the edge is one line in a
+    /// builder closure whose absence changes nothing observable until a real
+    /// agent boots and starts taking turns nobody asked for.
+    #[tokio::test]
+    async fn the_deploy_waits_for_the_pause_to_be_declared() {
+        use hive_jobq_wire::WireNode as _;
+
+        let (state, sched) = state_with_roster();
+        let _queued = super::create_agent(
+            axum::extract::State(state),
+            axum::Json(super::CreateAgentRequest {
+                name: "atlas".to_owned(),
+                hive: "pr1ma".to_owned(),
+            }),
+        )
+        .await
+        .expect("a hive in the roster must be accepted");
+
+        let guard = sched
+            .lock()
+            .unwrap_or_else(std::sync::PoisonError::into_inner);
+        let graph = guard.graph();
+        let id_of = |label: &str| {
+            graph
+                .nodes()
+                .find(|n| n.payload.label() == label)
+                .unwrap_or_else(|| panic!("the graph holds a {label} node"))
+                .id
+        };
+        let set_wanted = id_of("set_agent_wanted");
+        let deploy = graph.node(id_of("trigger_deploy")).expect("just found");
+
+        let when = deploy
+            .deps
+            .iter()
+            .find_map(|d| match d {
+                hive_jobq::Dep::Node { id, when } if *id == set_wanted => Some(*when),
+                _ => None,
+            })
+            .expect("the deploy waits for the declaration");
+        assert!(
+            !when.accepts(hive_jobq::TerminalState::Failed),
+            "a declaration that never landed must not be deployed past; this \
+             edge has to be `after_ok`, not `after_any`"
+        );
+    }
+
     /// The node carries the hive, and the viewer has to see it. The `data`
     /// match is an or-pattern on purpose (see its own comment), and this is
     /// the assertion that the new variant joined the two-field arm rather