hive-c0re: a swarm deploy for an unknown agent provisions it

Per mara on the PR: the issue is about a *new* agent, there is no
approval because the operator clicked create at swarm level, and most of
what a hive does on create is already done by the controller.

`spawn_nodes` splits out of `spawn` the way `rebuild_nodes` already
splits out of `rebuild`: two callers want the same four nodes and
disagree only about what closes them. `first_deploy` is that subgraph
with no approval tail, and the absence is the point — that tail exists
because an operator used to approve the spawn at the hive, and asking
again after they clicked create at swarm level asks the same person the
same question twice.

The handler's predicate is "does a container exist", not "is one
running". `agents_for_meta_listing` is `nixos-container list`, so a
stopped agent still counts. `Coordinator::list_agents` looks like the
right check and is the registered-MCP-socket set — a stopped agent is
absent from it, and this would then try to create over an existing
container.

Enumeration failure drops the request rather than guessing: without the
list this cannot tell first deploy from rebuild, and guessing "new" is
the destructive direction.

Still missing, and the reason this is not the whole change: the hive
seeds its own config repo with `git init` instead of cloning the one the
controller already created.

hive-c0re/src/job_queue/templates.rs

@@ -452,6 +452,15 @@ pub fn approval_deploy(builder: &JobBuilder, agent: &str, approval_id: i64) {
 /// `AfterAny` onto `Provision` — the DAG's only other group-root, so its roll-up
 /// already carries the whole cascade.
 pub fn spawn(builder: &JobBuilder, agent: &str, approval_id: i64) {
+    let provision = spawn_nodes(builder, agent);
+    resolve_approval_tails(builder, approval_id, provision);
+}
+
+/// The spawn subgraph with no tail, returning its group root.
+///
+/// Split out for the same reason [`rebuild_nodes`] is: two callers want the
+/// same four nodes and disagree only about what closes them.
+pub(crate) fn spawn_nodes<'a>(builder: &'a JobBuilder, agent: &str) -> Handle<'a> {
     let a = || agent.to_owned();
     let provision = builder
         .node(NodeKind::Provision { agent: a() })
@@ -470,8 +479,20 @@ pub fn spawn(builder: &JobBuilder, agent: &str, approval_id: i64) {
         .needs(Resource::Agent(a()))
         .part_of(create)
         .after_ok(dropin);
+    provision
+}
 
-    resolve_approval_tails(builder, approval_id, provision);
+/// First deploy of an agent this hive has never seen, asked for by the swarm.
+///
+/// [`spawn`] without the approval tail, and the absence is the point rather
+/// than an omission: that flow exists because an operator used to approve the
+/// spawn *at the hive*. When the swarm asks, the operator has already clicked
+/// create at swarm level — the deploy request carries that authorisation, and a
+/// second gate here would be asking the same person the same question twice.
+///
+/// Returns the group root so a caller can wait on the whole subtree.
+pub fn first_deploy(builder: &JobBuilder, agent: &str) -> Vec<hive_jobq::NodeGuid> {
+    vec![spawn_nodes(builder, agent).guid()]
 }
 
 /// Teardown: `Stop` → `DestroyContainer` → (`PurgeState`) → `DestroyBookkeeping`.

hive-c0re/src/swarm_status.rs

@@ -221,7 +221,7 @@ async fn drain_swarm_events(
                     tracing::warn!(subject = %deploy_subject, "swarm events: deploy subscription closed");
                     return;
                 };
-                handle_deploy_request(&coord, &msg.payload);
+                handle_deploy_request(&coord, &msg.payload).await;
             }
             _ = shutdown.changed() => {
                 tracing::info!("swarm events: shutdown signal received");
@@ -236,7 +236,10 @@ async fn drain_swarm_events(
 ///
 /// A payload that will not decode is worth a `warn`: the controller and this
 /// end share one type, so a decode failure means they disagree about it.
-fn handle_deploy_request(coord: &std::sync::Arc<crate::coordinator::Coordinator>, payload: &[u8]) {
+async fn handle_deploy_request(
+    coord: &std::sync::Arc<crate::coordinator::Coordinator>,
+    payload: &[u8],
+) {
     let request: swarm_queue_client::DeployRequest = match serde_json::from_slice(payload) {
         Ok(request) => request,
         Err(e) => {
@@ -244,17 +247,45 @@ fn handle_deploy_request(coord: &std::sync::Arc<crate::coordinator::Coordinator>
             return;
         }
     };
-    // The same insert the operator's own `rebuild` verb makes, relock and
-    // all: "deploy this agent" means here exactly what it already meant, and
-    // a swarm-triggered rebuild that quietly did something narrower would be
-    // a second definition of the word.
     let agent = request.agent;
-    let inserted = coord
-        .job_queue
-        .insert_job(|b| crate::job_queue::templates::rebuild(b, &agent, true));
+
+    // Which of the two meanings this request has is decided here, and the
+    // predicate is "does a container exist", not "is one running":
+    // `agents_for_meta_listing` is `nixos-container list`, so a stopped agent
+    // still counts. `Coordinator::list_agents` would *look* right and is the
+    // registered-MCP-socket set — a stopped agent is absent from it, and this
+    // would then try to create a container that already exists.
+    let known = match crate::lifecycle::agents_for_meta_listing().await {
+        Ok(agents) => agents.iter().any(|spec| spec.name == agent),
+        Err(e) => {
+            // Fail closed: without the list this cannot tell first deploy from
+            // rebuild, and guessing wrong in the "new" direction tries to
+            // create over an existing container.
+            tracing::warn!(%agent, error = ?e, "swarm events: cannot enumerate agents; deploy dropped");
+            return;
+        }
+    };
+
+    let inserted = if known {
+        // The same insert the operator's own `rebuild` verb makes, relock and
+        // all: "deploy this agent" means here exactly what it already meant, and
+        // a swarm-triggered rebuild that quietly did something narrower would be
+        // a second definition of the word.
+        coord
+            .job_queue
+            .insert_job(|b| crate::job_queue::templates::rebuild(b, &agent, true))
+    } else {
+        // First deploy. The swarm has already created the identity, the forge
+        // repo and its config; what is left is hive-local, and there is no
+        // approval to wait on because the operator's click at swarm level is
+        // the authorisation.
+        coord
+            .job_queue
+            .insert_job(|b| crate::job_queue::templates::first_deploy(b, &agent))
+    };
     match inserted {
         Ok(_) => {
-            tracing::info!(%agent, "swarm events: deploy requested by the swarm, rebuild queued");
+            tracing::info!(%agent, known, "swarm events: deploy requested by the swarm, queued");
             coord.emit_rebuild_queue_snapshot();
         }
         Err(e) => tracing::warn!(%agent, error = %e, "swarm events: queueing the deploy failed"),