feat: verify AgentWorkforce persona spawns

CHANGELOG.md

@@ -5,7 +5,11 @@ All notable changes to Agent Relay will be documented in this file.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
-## [Unreleased]
+## [Unreleased - Minor]
+
+### Added
+
+- The Agent Relay `spawn` MCP tool accepts an AgentWorkforce `persona` id or path instead of a raw CLI, routes it to a `spawn:persona` fleet node, and waits for broker registration plus harness readiness before reporting success. `@agent-relay/fleet` documents the corresponding `defineWorkforcePersonaSpawnNode` setup.
 
 ## [11.4.3] - 2026-08-09
 

crates/broker/src/protocol.rs

@@ -220,6 +220,14 @@ impl ResolvedHarnessConfig {
             Self::Native(config) => Some(config.session_id.as_str()),
         }
     }
+
+    pub(crate) fn metadata(&self) -> Option<&HashMap<String, Value>> {
+        match self {
+            Self::Pty(config) => config.metadata.as_ref(),
+            Self::Headless(config) => config.metadata.as_ref(),
+            Self::Native(config) => config.metadata.as_ref(),
+        }
+    }
 }
 
 #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]

crates/broker/src/runtime/api.rs

@@ -599,6 +599,10 @@ impl BrokerRuntime {
                                 || effective_spec.shadow_mode.is_some(),
                         });
                         let pid = workers.harness_pid(&name);
+                        let generation = workers
+                            .workers
+                            .get(&name)
+                            .map(|handle| handle.generation.to_string());
                         state.agents.insert(
                             name.clone(),
                             broker::PersistedAgent {
@@ -665,6 +669,7 @@ impl BrokerRuntime {
                             "model": effective_spec.model.clone(),
                             "sessionId": effective_spec.session_id.clone(),
                             "pid": pid,
+                            "generation": generation.clone(),
                             "sessionId": effective_spec.session_id.clone(),
                             "pre_registered": worker_relay_key.is_some(),
                             "warning": preregistration_warning,
@@ -699,13 +704,18 @@ impl BrokerRuntime {
                 }
 
                 let result_id = format!("ar_{}", Uuid::new_v4().simple());
+                let generation = workers
+                    .workers
+                    .get(&agent_name)
+                    .map(|handle| handle.generation.to_string());
                 let payload = json!({
                     "kind": "agent_result",
                     "name": agent_name,
                     "result_id": result_id,
                     "data": data,
                     "final": final_result,
                     "metadata": metadata,
+                    "generation": generation,
                 });
                 let _ = send_event(sdk_out_tx, payload).await;
                 let _ = reply.send(Ok(json!({

crates/broker/src/runtime/event_loop.rs

@@ -222,6 +222,9 @@ pub(crate) struct BrokerRuntime {
     pub(super) dead_letters: DeadLetterStore,
     pub(super) terminal_failed_deliveries: HashSet<DeliveryId>,
     pub(super) pending_requests: HashMap<String, worker_request::PendingRequest>,
+    /// Persona/capability spawns whose action result is held until the harness
+    /// proves readiness with worker_ready. Keyed by the node-local worker name.
+    pub(super) pending_verified_spawns: HashMap<WorkerName, super::fleet::PendingVerifiedSpawn>,
     /// Per-worker PTY resize ownership (single-resizer policy, see #1247).
     ///
     /// A shared PTY has exactly one size, so letting every attached client

crates/broker/src/runtime/fleet.rs

@@ -9,6 +9,39 @@ use crate::{
 };
 
 const FLEET_AGENT_REGISTER_TIMEOUT: Duration = Duration::from_secs(30);
+const VERIFIED_SPAWN_READY_TIMEOUT: Duration = Duration::from_secs(90);
+
+#[derive(Debug, Clone)]
+pub(super) struct PendingVerifiedSpawn {
+    pub(super) invocation_id: String,
+    pub(super) deadline: Instant,
+    pub(super) generation: Uuid,
+}
+
+pub(super) fn verified_spawn_ready_result(
+    invocation_id: String,
+    name: &WorkerName,
+) -> ActionResult {
+    ActionResult {
+        v: FLEET_WIRE_VERSION,
+        id: None,
+        invocation_id,
+        result: ActionResultPayload::Output(ActionResultOutput {
+            output: json!({ "spawned": true, "ready": true, "name": name.as_str() }),
+        }),
+    }
+}
+
+pub(super) fn verified_spawn_failed_result(invocation_id: String, error: &str) -> ActionResult {
+    ActionResult {
+        v: FLEET_WIRE_VERSION,
+        id: None,
+        invocation_id,
+        result: ActionResultPayload::Error(ActionResultError {
+            error: error.to_string(),
+        }),
+    }
+}
 
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
 enum FleetDeliverySurfaceOutcome {
@@ -362,6 +395,13 @@ impl BrokerRuntime {
                 .await;
             return;
         };
+        if self.workers.workers.contains_key(&name)
+            || self.pending_verified_spawns.contains_key(&name)
+        {
+            self.reply_action_error(&invoke.invocation_id, "spawn_agent_name_in_use")
+                .await;
+            return;
+        }
         let cli = match action_invoke_string(&invoke.input, &["cli", "command", "provider"]) {
             Some(cli) => cli,
             None => {
@@ -453,15 +493,73 @@ impl BrokerRuntime {
 
         self.publish_fleet_load(true).await;
 
-        // `spawn_worker_from_request` does not return a result; treat presence of
-        // the worker as success so the engine's invocation resolves.
+        let verify_ready = super::relaycast_events::relaycast_spawn_verifies_ready(&ws_value);
+
+        // A verified spawn keeps the action open until the harness itself emits
+        // worker_ready. Process creation alone is not proof that the persona is
+        // usable; worker_events resolves this pending entry, while maintenance
+        // fails it after an early exit/readiness timeout and performs cleanup.
         if self.workers.workers.contains_key(&name) {
+            if verify_ready {
+                if self
+                    .workers
+                    .workers
+                    .get(&name)
+                    .is_some_and(|worker| worker.ready_at.is_some())
+                {
+                    self.send_fleet_action_result(verified_spawn_ready_result(
+                        invoke.invocation_id,
+                        &name,
+                    ))
+                    .await;
+                } else {
+                    let generation = self
+                        .workers
+                        .workers
+                        .get(&name)
+                        .expect("verified spawn worker must still exist")
+                        .generation;
+                    self.pending_verified_spawns.insert(
+                        name,
+                        PendingVerifiedSpawn {
+                            invocation_id: invoke.invocation_id,
+                            deadline: Instant::now() + VERIFIED_SPAWN_READY_TIMEOUT,
+                            generation,
+                        },
+                    );
+                }
+                return;
+            }
             self.reply_action_output(
                 &invoke.invocation_id,
                 json!({ "spawned": true, "name": name.as_str() }),
             )
             .await;
         } else {
+            // A registration can succeed before process creation fails. Undo
+            // that authoritative identity before reporting the failed launch.
+            match deregister_fleet_agent(&self.fleet_control_tx, &self.fleet_delivery_book, &name)
+                .await
+            {
+                Ok(_) => {
+                    prune_fleet_agent_state(
+                        &self.fleet_control_tx,
+                        &mut self.fleet_inventory,
+                        &mut self.fleet_delivery_book,
+                        &name,
+                    )
+                    .await
+                }
+                Err(error) => {
+                    tracing::warn!(worker = %name, %error, "retaining fleet identity after failed spawn cleanup");
+                    prune_fleet_inventory_entry(
+                        &self.fleet_control_tx,
+                        &mut self.fleet_inventory,
+                        &name,
+                    )
+                    .await;
+                }
+            }
             self.reply_action_error(&invoke.invocation_id, "spawn_failed")
                 .await;
         }
@@ -505,15 +603,45 @@ impl BrokerRuntime {
         self.resize_owners.remove(&name);
         self.pty_observability.remove(&name);
 
-        prune_fleet_agent_state(
-            &self.fleet_control_tx,
-            &mut self.fleet_inventory,
-            &mut self.fleet_delivery_book,
-            &name,
-        )
-        .await;
+        let mut deregistration_failed = false;
+        if outcome == super::relaycast_events::ReleaseOutcome::Released {
+            match deregister_fleet_agent(&self.fleet_control_tx, &self.fleet_delivery_book, &name)
+                .await
+            {
+                Ok(_) => {
+                    prune_fleet_agent_state(
+                        &self.fleet_control_tx,
+                        &mut self.fleet_inventory,
+                        &mut self.fleet_delivery_book,
+                        &name,
+                    )
+                    .await;
+                }
+                Err(error) => {
+                    tracing::warn!(worker = %name, %error, "retaining fleet identity after release cleanup");
+                    deregistration_failed = true;
+                    prune_fleet_inventory_entry(
+                        &self.fleet_control_tx,
+                        &mut self.fleet_inventory,
+                        &name,
+                    )
+                    .await;
+                }
+            }
+        }
+        if let Some(pending) = self.pending_verified_spawns.remove(&name) {
+            self.send_fleet_action_result(verified_spawn_failed_result(
+                pending.invocation_id,
+                "spawn_released_before_ready",
+            ))
+            .await;
+        }
         self.publish_fleet_load(true).await;
         match outcome {
+            super::relaycast_events::ReleaseOutcome::Released if deregistration_failed => {
+                self.reply_action_error(&invoke.invocation_id, "release_deregistration_failed")
+                    .await;
+            }
             super::relaycast_events::ReleaseOutcome::Released => {
                 self.reply_action_output(
                     &invoke.invocation_id,

crates/broker/src/runtime/init.rs

@@ -594,6 +594,7 @@ pub(crate) async fn run_init(cmd: InitCommand, telemetry: TelemetryClient) -> Re
     // so each new request/response route (`snapshot_pty`, `delivery-mode`,
     // `pending`, `flush`, ...) costs about five lines of broker plumbing.
     let pending_requests: HashMap<String, worker_request::PendingRequest> = HashMap::new();
+    let pending_verified_spawns = HashMap::new();
     // Per-worker inbound-delivery-mode + pending-relay-message queue. Lives
     // parallel to `workers.workers` so we can swap modes / inspect /
     // drain without touching `WorkerHandle` (which holds OS-level
@@ -679,6 +680,7 @@ pub(crate) async fn run_init(cmd: InitCommand, telemetry: TelemetryClient) -> Re
         dead_letters,
         terminal_failed_deliveries,
         pending_requests,
+        pending_verified_spawns,
         resize_owners: HashMap::new(),
         delivery_states,
         agent_result_tokens,