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 - Patch]
+## [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.
 
 ### Fixed
 

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/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,38 @@ 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) 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 +394,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 +492,66 @@ 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 {
+                    self.pending_verified_spawns.insert(
+                        name,
+                        PendingVerifiedSpawn {
+                            invocation_id: invoke.invocation_id,
+                            deadline: Instant::now() + VERIFIED_SPAWN_READY_TIMEOUT,
+                        },
+                    );
+                }
+                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,13 +595,41 @@ 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;
+        if outcome == super::relaycast_events::ReleaseOutcome::Released {
+            match deregister_fleet_agent(
+                &self.fleet_control_tx,
+                &mut 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");
+                    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 => {

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,