Make agent creation swarm-only and refuse a name placed on another hive

swarm-controller's POST /api/agents now refuses (409) a name the swarm
has already placed on a different hive: a non-Destroyed declaration in
that hive's wanted state, or a SetAgentWanted node still queued for it.
The same name on the same hive is that agent being re-created and goes
through. A wanted state that cannot be read refuses (503/500) instead of
reading as "placed nowhere". Creations are serialised from that read to
the graph insert so two concurrent creations of one name cannot both
pass.

Hive-level creation is removed: hivectl `agent create` / `request-create`,
HostRequest::Spawn / RequestSpawn, the dashboard POST /api/request-spawn
route, and ApprovalKind::Spawn with its approve/resolve arms and the
approval-carrying `templates::spawn`. The swarm path (deploy request or
wanted-state sweep -> queue_first_deploy -> templates::first_deploy) used
none of them. Old `spawn` approval rows are skipped by collect_lenient,
as `init_config` rows were in a3b672d1.

policy.rs's comment on agent_object_name stated swarm-wide name
uniqueness as a fact; it now says where it is enforced and what that
check cannot see.

Refs #4396

swarm-controller/src/main.rs

@@ -686,6 +686,10 @@ struct AppState {
     /// "no forge configured on this host" shape every other
     /// forge-backed field here uses.
     forge: Option<Arc<forge::Client>>,
+    /// Held by `create_agent` from reading where a name is placed until its
+    /// graph is queued, so two creations of one name cannot both find it
+    /// unplaced. `tokio`'s mutex, unlike `jobq`'s: the read awaits the queue.
+    create_gate: Arc<tokio::sync::Mutex<()>>,
 }
 
 /// Env var the controller's NixOS module sets from
@@ -1352,7 +1356,9 @@ struct CreateAgentResponse {
     responses(
         (status = 200, description = "job chain queued", body = CreateAgentResponse),
         (status = 400, description = "`name` or `hive` is not a valid identifier, `name` is new and reserved or refused by the forge, or `hive` is not in this swarm (problem+json)", body = String),
-        (status = 500, description = "the job chain could not be queued (problem+json)", body = String),
+        (status = 409, description = "the swarm has already placed `name` on a different hive (problem+json)", body = String),
+        (status = 503, description = "the swarm queue is not connected, so whether `name` is placed on another hive is unknown (problem+json)", body = String),
+        (status = 500, description = "the job chain could not be queued, or another hive's wanted state could not be read (problem+json)", body = String),
     ),
     tag = "agents"
 )]
@@ -1424,10 +1430,25 @@ async fn create_agent(
         return Err(error_problem(axum::http::StatusCode::BAD_REQUEST, &detail));
     }
 
+    // Agent names are one swarm-wide namespace: identity, forge user, store
+    // path and matrix id all carry the bare name. The same name on the same
+    // hive is that agent being re-created, and goes through.
+    let _gate = state.create_gate.lock().await;
+    let declared = declarations_elsewhere(&state, &agent, &hive).await?;
     let mut sched = state
         .jobq
         .lock()
         .unwrap_or_else(std::sync::PoisonError::into_inner);
+    let queued = queued_placements(sched.graph());
+    let elsewhere = placed_elsewhere(&agent, &hive, &declared, &queued);
+    if !elsewhere.is_empty() {
+        let detail = format!(
+            "agent name {agent:?} is already placed on hive {} — agent names are unique across \
+             the swarm; destroy it there first to move it, or choose another name",
+            elsewhere.join(", ")
+        );
+        return Err(error_problem(axum::http::StatusCode::CONFLICT, &detail));
+    }
     let ids = sched
         .insert_job(None, |b| declare_agent_job(b, &agent, &hive))
         .map_err(|e| {
@@ -1445,6 +1466,81 @@ async fn create_agent(
     }))
 }
 
+/// Every hive but `hive`'s published wanted state, for [`placed_elsewhere`].
+///
+/// No queue wired up means no hive has been sent a declaration, so there is
+/// nothing to collide with. A queue that cannot be read refuses: an unread
+/// declaration and an absent one look the same, and reading "absent" is how
+/// a duplicate would get through.
+async fn declarations_elsewhere(
+    state: &AppState,
+    agent: &str,
+    hive: &str,
+) -> Result<Vec<(String, swarm_queue_client::wanted::HiveWanted)>, problem_details::ProblemDetails>
+{
+    let Some(writer) = state.wanted.as_deref() else {
+        return Ok(Vec::new());
+    };
+    let mut declared = Vec::new();
+    for other in state.hives.iter().filter(|h| h.name != hive) {
+        match writer.view(&other.name).await {
+            Ok(Some(declaration)) => declared.push((other.name.clone(), declaration)),
+            Ok(None) => {}
+            Err(e) => {
+                let detail = format!(
+                    "cannot tell whether {agent:?} already exists on hive {:?}, so it was not \
+                     created: {e:#}",
+                    other.name
+                );
+                return Err(error_problem(wanted_error_status(&e), &detail));
+            }
+        }
+    }
+    Ok(declared)
+}
+
+/// `(hive, agent)` of every `SetAgentWanted` node not yet settled: a creation
+/// accepted but not yet visible in the wanted state it is about to write.
+fn queued_placements(
+    graph: &hive_jobq::Graph<SwarmNodeKind, SwarmResourceKind>,
+) -> Vec<(String, String)> {
+    graph
+        .nodes()
+        .filter(|node| !node.state.is_terminal())
+        .filter_map(|node| match &node.payload {
+            SwarmNodeKind::SetAgentWanted { hive, agent } => Some((hive.clone(), agent.clone())),
+            _ => None,
+        })
+        .collect()
+}
+
+/// The hives other than `hive` the swarm has placed `agent` on, sorted.
+///
+/// A placement is a declaration in any state but `Destroyed`, or a queued
+/// one. A `Destroyed` agent is gone from that hive, so creating it on another
+/// moves it.
+fn placed_elsewhere(
+    agent: &str,
+    hive: &str,
+    declared: &[(String, swarm_queue_client::wanted::HiveWanted)],
+    queued: &[(String, String)],
+) -> Vec<String> {
+    use swarm_queue_client::wanted::AgentState;
+    let declared = declared.iter().filter(|(_, declaration)| {
+        declaration
+            .agents
+            .get(agent)
+            .is_some_and(|wanted| wanted.state != AgentState::Destroyed)
+    });
+    let queued = queued.iter().filter(|(_, queued)| queued == agent);
+    let hives: std::collections::BTreeSet<&str> = declared
+        .map(|(h, _)| h.as_str())
+        .chain(queued.map(|(h, _)| h.as_str()))
+        .filter(|h| *h != hive)
+        .collect();
+    hives.into_iter().map(str::to_owned).collect()
+}
+
 /// Every naming rule `agent` breaks, each as a sentence naming the rule.
 ///
 /// `reserved` is [`hive_types::reserved_names_raw`]: the list nix owns in
@@ -2554,6 +2650,7 @@ async fn main() -> Result<()> {
         swarm_name: load_swarm_name().map(Arc::from),
         auth,
         forge: state_forge,
+        create_gate: Arc::default(),
     };
 
     let app = build_app(state);
@@ -2710,6 +2807,7 @@ mod tests {
             // read is the verb that needs it, and it has its own test below.
             auth: None,
             forge: None,
+            create_gate: std::sync::Arc::default(),
         };
         (state, sched)
     }
@@ -2846,6 +2944,170 @@ mod tests {
         assert!(queued > 0, "an accepted creation must queue work");
     }
 
+    /// `state_with_roster` with a second hive, `sec0nd`.
+    fn state_with_two_hives() -> (super::AppState, SharedSched) {
+        let (state, sched) = state_with_roster();
+        let mut hives = (*state.hives).clone();
+        hives.push(HiveEntry {
+            name: "sec0nd".to_owned(),
+            domain: "sec0nd.example".to_owned(),
+        });
+        let state = super::AppState {
+            hives: std::sync::Arc::new(hives),
+            ..state
+        };
+        (state, sched)
+    }
+
+    async fn create(
+        state: &super::AppState,
+        name: &str,
+        hive: &str,
+    ) -> Result<(), problem_details::ProblemDetails> {
+        super::create_agent(
+            axum::extract::State(state.clone()),
+            axum::Json(super::CreateAgentRequest {
+                name: name.to_owned(),
+                hive: hive.to_owned(),
+            }),
+        )
+        .await
+        .map(|_| ())
+    }
+
+    fn node_count(sched: &SharedSched) -> usize {
+        sched
+            .lock()
+            .unwrap_or_else(std::sync::PoisonError::into_inner)
+            .graph()
+            .nodes()
+            .count()
+    }
+
+    /// A name queued for one hive is refused on another, by effect: 409 and
+    /// nothing more queued. The queued-but-unwritten placement is the one a
+    /// wanted-state read alone would miss.
+    #[tokio::test]
+    async fn a_name_placed_on_another_hive_is_refused_before_anything_is_queued() {
+        let (state, sched) = state_with_two_hives();
+        create(&state, "atlas", "pr1ma")
+            .await
+            .expect("the first creation of a name must be accepted");
+        let before = node_count(&sched);
+
+        let err = create(&state, "atlas", "sec0nd")
+            .await
+            .expect_err("the same name on another hive must be refused");
+        assert_eq!(
+            err.status,
+            Some(axum::http::StatusCode::CONFLICT),
+            "{err:?}"
+        );
+        assert!(
+            format!("{err:?}").contains("pr1ma"),
+            "the refusal should name the hive holding the name, got: {err:?}"
+        );
+        assert_eq!(
+            node_count(&sched),
+            before,
+            "a refused creation must queue no work"
+        );
+    }
+
+    /// The control for the refusal above: the same name on the SAME hive is
+    /// that agent being re-created, and queues again.
+    #[tokio::test]
+    async fn re_creating_an_agent_on_its_own_hive_is_accepted() {
+        let (state, sched) = state_with_two_hives();
+        create(&state, "atlas", "pr1ma")
+            .await
+            .expect("the first creation of a name must be accepted");
+        let before = node_count(&sched);
+
+        create(&state, "atlas", "pr1ma")
+            .await
+            .expect("re-creating an agent on its own hive must be accepted");
+        assert!(node_count(&sched) > before, "a re-creation must queue work");
+    }
+
+    /// A wanted state that cannot be read refuses rather than reading as
+    /// "placed nowhere", and queues nothing.
+    #[tokio::test]
+    async fn an_unreadable_wanted_state_refuses_creation() {
+        let (state, sched) = state_with_two_hives();
+        let state = super::AppState {
+            wanted: Some(std::sync::Arc::new(wanted::WantedWriter::new(
+                disconnected_client().await,
+            ))),
+            ..state
+        };
+
+        let err = create(&state, "atlas", "pr1ma")
+            .await
+            .expect_err("an unreadable placement must refuse");
+        assert_eq!(
+            err.status,
+            Some(axum::http::StatusCode::SERVICE_UNAVAILABLE),
+            "{err:?}"
+        );
+        assert_eq!(
+            node_count(&sched),
+            0,
+            "a refused creation must queue no work"
+        );
+    }
+
+    fn declaring(
+        hive: &str,
+        agent: &str,
+        state: swarm_queue_client::wanted::AgentState,
+    ) -> (String, swarm_queue_client::wanted::HiveWanted) {
+        let mut declaration = swarm_queue_client::wanted::HiveWanted::default();
+        declaration.agents.insert(
+            agent.to_owned(),
+            swarm_queue_client::wanted::AgentWanted { state },
+        );
+        (hive.to_owned(), declaration)
+    }
+
+    #[test]
+    fn a_live_declaration_on_another_hive_is_a_placement() {
+        use swarm_queue_client::wanted::AgentState;
+        for state in [AgentState::Up, AgentState::Offline, AgentState::Paused] {
+            let declared = [declaring("sec0nd", "atlas", state)];
+            assert_eq!(
+                super::placed_elsewhere("atlas", "pr1ma", &declared, &[]),
+                ["sec0nd"],
+                "{state:?}"
+            );
+        }
+    }
+
+    /// A destroyed agent has left its hive, so creating it elsewhere moves it.
+    #[test]
+    fn a_destroyed_declaration_is_not_a_placement() {
+        let declared = [declaring(
+            "sec0nd",
+            "atlas",
+            swarm_queue_client::wanted::AgentState::Destroyed,
+        )];
+        assert!(super::placed_elsewhere("atlas", "pr1ma", &declared, &[]).is_empty());
+    }
+
+    #[test]
+    fn placements_on_the_requested_hive_or_of_other_names_do_not_count() {
+        use swarm_queue_client::wanted::AgentState;
+        let declared = [
+            declaring("pr1ma", "atlas", AgentState::Up),
+            declaring("sec0nd", "argus", AgentState::Up),
+        ];
+        let queued = [
+            ("pr1ma".to_owned(), "atlas".to_owned()),
+            ("sec0nd".to_owned(), "argus".to_owned()),
+        ];
+        assert!(super::placed_elsewhere("atlas", "pr1ma", &declared, &queued).is_empty());
+    }
+
     /// Forgejo v16.0.5's reserved usernames our charset can spell, plus the
     /// bare `-` — the forge section of `nix/reserved-names.nix`. Read from
     /// the list nix exports, so dropping one from the file reds this.

swarm-controller/src/matrix_account.rs

@@ -580,6 +580,7 @@ mod tests {
             swarm_name: None,
             auth: None,
             forge: None,
+            create_gate: std::sync::Arc::default(),
         }
     }