From bcbd92e6ef9649f69034956c9bdfc9c25ab9b40c Mon Sep 17 00:00:00 2001 From: Proactive Runtime Bot Date: Sat, 8 Aug 2026 00:59:21 +0200 Subject: [PATCH 1/7] feat: verify persona fleet spawns --- CHANGELOG.md | 6 +- crates/broker/src/protocol.rs | 8 ++ crates/broker/src/runtime/event_loop.rs | 3 + crates/broker/src/runtime/fleet.rs | 89 ++++++++++++++++++- crates/broker/src/runtime/init.rs | 2 + crates/broker/src/runtime/maintenance.rs | 60 ++++++++++++- crates/broker/src/runtime/relaycast_events.rs | 60 ++++++++++++- crates/broker/src/runtime/worker_events.rs | 12 ++- .../src/cli/agent-relay-mcp.startup.test.ts | 34 +++++++ packages/cli/src/cli/agent-relay-mcp.ts | 89 +++++++++++++++++-- packages/fleet/README.md | 26 ++++++ packages/fleet/src/serve-node.test.ts | 44 +++++++++ packages/fleet/src/serve-node.ts | 3 +- packages/harness-driver/src/agent-handle.ts | 56 ++++++++++++ .../harness-driver/src/agent-result.test.ts | 25 ++++++ 15 files changed, 503 insertions(+), 14 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 37687938f..0962257d5 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/crates/broker/src/protocol.rs b/crates/broker/src/protocol.rs index 66e0e30ae..a74186c7a 100644 --- a/crates/broker/src/protocol.rs +++ b/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> { + 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)] diff --git a/crates/broker/src/runtime/event_loop.rs b/crates/broker/src/runtime/event_loop.rs index 15c98c832..b92971061 100644 --- a/crates/broker/src/runtime/event_loop.rs +++ b/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, pub(super) pending_requests: HashMap, + /// 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, /// Per-worker PTY resize ownership (single-resizer policy, see #1247). /// /// A shared PTY has exactly one size, so letting every attached client diff --git a/crates/broker/src/runtime/fleet.rs b/crates/broker/src/runtime/fleet.rs index be2daaf1e..1c7deb1cb 100644 --- a/crates/broker/src/runtime/fleet.rs +++ b/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,54 @@ 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. + let _ = + deregister_fleet_agent(&self.fleet_control_tx, &self.fleet_delivery_book, &name) + .await; + prune_fleet_agent_state( + &self.fleet_control_tx, + &mut self.fleet_inventory, + &mut self.fleet_delivery_book, + &name, + ) + .await; self.reply_action_error(&invoke.invocation_id, "spawn_failed") .await; } @@ -512,6 +590,13 @@ impl BrokerRuntime { &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 => { diff --git a/crates/broker/src/runtime/init.rs b/crates/broker/src/runtime/init.rs index 14c5effa3..ea7ffef6f 100644 --- a/crates/broker/src/runtime/init.rs +++ b/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 = 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, diff --git a/crates/broker/src/runtime/maintenance.rs b/crates/broker/src/runtime/maintenance.rs index a76eca180..dae3f1041 100644 --- a/crates/broker/src/runtime/maintenance.rs +++ b/crates/broker/src/runtime/maintenance.rs @@ -22,11 +22,13 @@ impl BrokerRuntime { let pending_deliveries = &mut self.pending_deliveries; let dead_letters = &mut self.dead_letters; let pending_requests = &mut self.pending_requests; + let pending_verified_spawns = &mut self.pending_verified_spawns; let delivery_states = &mut self.delivery_states; let resize_owners = &mut self.resize_owners; let agent_result_tokens = &mut self.agent_result_tokens; let delivery_retry_interval = self.delivery_retry_interval; let shutdown = &self.shutdown; + let default_workspace = &self.default_workspace; let now = Instant::now(); @@ -93,6 +95,50 @@ impl BrokerRuntime { } } + let expired_verified_spawns: Vec<(WorkerName, String)> = pending_verified_spawns + .iter() + .filter(|(_, pending)| pending.deadline <= now) + .map(|(name, pending)| (name.clone(), pending.invocation_id.clone())) + .collect(); + for (name, invocation_id) in &expired_verified_spawns { + pending_verified_spawns.remove(name); + let _ = super::relaycast_events::release_worker_locally( + name.clone(), + default_workspace, + workers, + state, + paths, + telemetry, + sdk_out_tx, + pending_deliveries, + dead_letters, + pending_requests, + delivery_states, + agent_result_tokens, + ) + .await; + let _ = + super::fleet::deregister_fleet_agent(fleet_control_tx, fleet_delivery_book, name) + .await; + super::fleet::prune_fleet_agent_state( + fleet_control_tx, + fleet_inventory, + fleet_delivery_book, + name, + ) + .await; + let _ = fleet_control_tx + .send(FleetControlCommand::Send( + crate::fleet_wire::BrokerToRelaycast::ActionResult( + super::fleet::verified_spawn_failed_result( + invocation_id.clone(), + "spawn_readiness_timeout", + ), + ), + )) + .await; + } + let exited = match workers.reap_exited().await { Ok(v) => v, Err(e) => { @@ -100,8 +146,20 @@ impl BrokerRuntime { vec![] } }; - let mut fleet_load_changed = !exited.is_empty(); + let mut fleet_load_changed = !expired_verified_spawns.is_empty() || !exited.is_empty(); for (name, code, signal, exit_reason) in &exited { + if let Some(pending) = pending_verified_spawns.remove(name) { + let _ = fleet_control_tx + .send(FleetControlCommand::Send( + crate::fleet_wire::BrokerToRelaycast::ActionResult( + super::fleet::verified_spawn_failed_result( + pending.invocation_id, + "spawn_harness_not_ready", + ), + ), + )) + .await; + } let lifecycle_reason = exit_reason.as_deref().unwrap_or("worker_exited"); if (code.is_some_and(|code| code != 0) || signal.is_some()) && state diff --git a/crates/broker/src/runtime/relaycast_events.rs b/crates/broker/src/runtime/relaycast_events.rs index 294bd61b8..e85a6ef76 100644 --- a/crates/broker/src/runtime/relaycast_events.rs +++ b/crates/broker/src/runtime/relaycast_events.rs @@ -131,6 +131,22 @@ fn relaycast_harness_config(value: &Value) -> Result bool { + config + .metadata() + .and_then(|metadata| metadata.get(snake).or_else(|| metadata.get(camel))) + .and_then(Value::as_bool) + .unwrap_or(false) +} + +pub(super) fn relaycast_spawn_verifies_ready(value: &Value) -> bool { + relaycast_harness_config(value) + .ok() + .flatten() + .as_ref() + .is_some_and(|config| harness_metadata_flag(config, "verify_ready", "verifyReady")) +} + /// Bind a freshly HTTP-registered agent to this broker's relaycast node so it /// becomes `locationType='via_node'`. /// @@ -399,6 +415,13 @@ pub(super) async fn spawn_worker_from_request( return; } }; + let require_node_registration = harness_config.as_ref().is_some_and(|config| { + harness_metadata_flag( + config, + "require_node_registration", + "requireNodeRegistration", + ) + }); let runtime = harness_config .as_ref() .map(ResolvedHarnessConfig::runtime) @@ -466,7 +489,9 @@ pub(super) async fn spawn_worker_from_request( // the worker MCP never re-registers over HTTP. Falls back to HTTP // pre-registration when node binding is unavailable. let worker_relay_key = { - if let Some(token) = relaycast_ws_spawn_token(ws_value) { + if let Some(token) = + relaycast_ws_spawn_token(ws_value).filter(|_| !require_node_registration) + { seed_supplied_agent_token(workspace_http, &name, &token); Some(token) } else { @@ -487,6 +512,14 @@ pub(super) async fn spawn_worker_from_request( Some(token.token) } Err(node_error) => { + if require_node_registration { + tracing::warn!( + worker = %name, + error = %node_error, + "rejecting verified spawn because node agent.register failed" + ); + return; + } tracing::warn!( worker = %name, error = %node_error, @@ -706,6 +739,31 @@ mod tests { assert!(error.contains("harnessId is not supported")); } + #[test] + fn verified_spawn_contract_is_read_from_harness_metadata() { + let verified = json!({ + "harness_config": { + "runtime": "pty", + "command": "codex", + "args": [], + "metadata": { + "verify_ready": true, + "require_node_registration": true + } + } + }); + let ordinary = json!({ + "harness_config": { + "runtime": "pty", + "command": "codex", + "args": [] + } + }); + + assert!(relaycast_spawn_verifies_ready(&verified)); + assert!(!relaycast_spawn_verifies_ready(&ordinary)); + } + /// Regression guard for the v5.0.1 firehose control path. /// /// In relaycast v5 `WsEvent` ends in `#[serde(other)] Unknown`, so an diff --git a/crates/broker/src/runtime/worker_events.rs b/crates/broker/src/runtime/worker_events.rs index ae6b76b5b..7c77f0ac3 100644 --- a/crates/broker/src/runtime/worker_events.rs +++ b/crates/broker/src/runtime/worker_events.rs @@ -1,4 +1,4 @@ -use super::fleet::refresh_fleet_inventory_session_ref; +use super::fleet::{refresh_fleet_inventory_session_ref, verified_spawn_ready_result}; use super::*; use crate::worker::AgentWorkState; @@ -322,6 +322,7 @@ impl BrokerRuntime { let dead_letters = &mut self.dead_letters; let terminal_failed_deliveries = &mut self.terminal_failed_deliveries; let pending_requests = &mut self.pending_requests; + let pending_verified_spawns = &mut self.pending_verified_spawns; let delivery_retry_interval = self.delivery_retry_interval; let fleet_control_tx = &self.fleet_control_tx; let fleet_inventory = &mut self.fleet_inventory; @@ -884,6 +885,15 @@ impl BrokerRuntime { }), ) .await; + if let Some(pending) = pending_verified_spawns.remove(&name) { + let _ = fleet_control_tx + .send(FleetControlCommand::Send( + crate::fleet_wire::BrokerToRelaycast::ActionResult( + verified_spawn_ready_result(pending.invocation_id, &name), + ), + )) + .await; + } } else if msg_type == "agent_idle" { let idle_secs = value .get("payload") diff --git a/packages/cli/src/cli/agent-relay-mcp.startup.test.ts b/packages/cli/src/cli/agent-relay-mcp.startup.test.ts index eedf4393d..c9c8168c4 100644 --- a/packages/cli/src/cli/agent-relay-mcp.startup.test.ts +++ b/packages/cli/src/cli/agent-relay-mcp.startup.test.ts @@ -145,6 +145,12 @@ async function loadAgentRelayMcpModule(options: LoadOptions = {}) { actionName: name, input, })), + getInvocation: vi.fn(async (name: string, invocationId: string) => ({ + invocationId, + actionName: name, + status: 'completed', + output: { spawned: true, ready: true }, + })), }, send: vi.fn(async (channel: string, text: string) => ({ id: 'msg_1', channel, text })), messages: vi.fn(async () => []), @@ -228,6 +234,21 @@ async function loadAgentRelayMcpModule(options: LoadOptions = {}) { const AgentRelayMock = vi.fn(function (this: unknown) { return { nodes: { list: agentRelayNodesList }, + messaging: { + commands: { + invoke: vi.fn(async (name: string, input: unknown) => ({ + invocationId: 'inv_1', + actionName: name, + input, + })), + getInvocation: vi.fn(async (name: string, invocationId: string) => ({ + invocationId, + actionName: name, + status: 'completed', + output: { spawned: true, ready: true }, + })), + }, + }, }; }) as any; @@ -510,6 +531,19 @@ describe('createAgentRelayMcpServer', () => { }, }); + const personaSpawnResult = await server.tools.get('spawn')?.handler({ + name: 'IntegrationExpert', + persona: 'nango-integrations', + task: 'Fix the sync', + cwd: '/workspace/project', + target_node: 'node-a', + }); + expect(personaSpawnResult.structuredContent.invocation).toEqual({ + invocationId: 'inv_1', + actionName: 'spawn', + status: 'completed', + output: { spawned: true, ready: true }, + }); const toolsList = await server.listToolsHandler?.({}, {}); expect(toolsList?.tools).toEqual([ { diff --git a/packages/cli/src/cli/agent-relay-mcp.ts b/packages/cli/src/cli/agent-relay-mcp.ts index d24af51b3..64d2c2080 100644 --- a/packages/cli/src/cli/agent-relay-mcp.ts +++ b/packages/cli/src/cli/agent-relay-mcp.ts @@ -62,6 +62,52 @@ function withExitAfterTaskInstruction(task: string): string { return `${task}\n\n${EXIT_AFTER_TASK_INSTRUCTION}`; } +const PERSONA_SPAWN_TIMEOUT_MS = 130_000; +const PERSONA_SPAWN_POLL_MS = 250; + +type InvocationReader = { + getInvocation(name: string, invocationId: string): Promise; +}; + +function recordValue(value: unknown): Record { + return value !== null && typeof value === 'object' && !Array.isArray(value) + ? (value as Record) + : {}; +} + +function invocationText(record: Record, camel: string, snake?: string): string | undefined { + const value = record[camel] ?? (snake ? record[snake] : undefined); + return typeof value === 'string' && value.trim() ? value : undefined; +} + +async function waitForPersonaSpawn( + actions: InvocationReader, + ackValue: unknown, + timeoutMs = PERSONA_SPAWN_TIMEOUT_MS +): Promise { + const ack = recordValue(ackValue); + const actionName = invocationText(ack, 'actionName', 'action_name') ?? 'spawn'; + const invocationId = invocationText(ack, 'invocationId', 'invocation_id'); + if (!invocationId) throw new Error('Persona spawn did not return an invocation id.'); + + const deadline = Date.now() + timeoutMs; + for (;;) { + const invocation = await actions.getInvocation(actionName, invocationId); + const record = recordValue(invocation); + const status = invocationText(record, 'status')?.toLowerCase(); + if (status === 'completed' || status === 'succeeded' || status === 'success') { + return invocation; + } + if (status === 'failed' || status === 'error' || status === 'cancelled' || status === 'canceled') { + throw new Error(invocationText(record, 'error') ?? `Persona spawn ${status}.`); + } + if (Date.now() >= deadline) { + throw new Error('Persona spawn timed out before broker registration and harness readiness.'); + } + await new Promise((resolve) => setTimeout(resolve, PERSONA_SPAWN_POLL_MS)); + } +} + export const AGENT_RELAY_MCP_INSTRUCTIONS = `You are an AI agent in a collaborative workspace powered by Agent Relay. You can communicate with other agents using these MCP tools: ## Coordination rule @@ -86,7 +132,8 @@ export const AGENT_RELAY_MCP_INSTRUCTIONS = `You are an AI agent in a collaborat ## Fleet - Use "query_nodes" to find fleet nodes by capability or name -- Use "spawn" to invoke the fleet spawn action on an eligible node +- Use "spawn" with either a CLI or an AgentWorkforce persona to invoke the fleet spawn action on an eligible node +- Persona spawns require a node exposing "spawn:persona" through @agentworkforce/local-surface's defineWorkforcePersonaSpawnNode ## Best Practices - Check your inbox regularly for new messages and mentions @@ -692,14 +739,20 @@ function registerAgentRelayTools( { title: 'Spawn Agent', description: - 'Invoke the fleet spawn action, optionally targeting a specific node. ' + - 'Returns an `invocation` record acknowledging the request. The action runs asynchronously, so this confirms the spawn was queued, not that the worker is running.', + 'Invoke the fleet spawn action with either a raw `cli` or an AgentWorkforce `persona` name/path. ' + + 'Persona requests route to a node exposing `spawn:persona` (for example, `defineWorkforcePersonaSpawnNode` from `@agentworkforce/local-surface`) and return only after broker registration and harness readiness are verified. Raw CLI requests retain asynchronous acknowledgement behavior.', inputSchema: { name: z.string().describe('Agent name'), cli: z .enum(['claude', 'codex', 'gemini', 'aider', 'goose', 'grok', 'opencode']) - .describe('AI CLI to launch'), + .optional() + .describe('AI CLI to launch; mutually exclusive with persona'), + persona: z + .string() + .optional() + .describe('AgentWorkforce persona id or JSON path; mutually exclusive with cli'), task: z.string().optional().describe('Initial task instructions'), + cwd: z.string().optional().describe('Project cwd used for persona registry resolution'), channel: z.string().optional().describe('Channel to join'), channels: z.array(z.string()).optional().describe('Channels to join'), model: z.string().optional().describe('Model powering the worker'), @@ -715,21 +768,43 @@ function registerAgentRelayTools( openWorldHint: true, }, }, - async ({ name, cli, task, channel, channels, model, session_ref, target_node, as }) => { + async ({ name, cli, persona, task, cwd, channel, channels, model, session_ref, target_node, as }) => { const actions = getAgentClient(as).actions; if (!actions) { throw new Error('spawn requires an agent-scoped Relaycast actions client.'); } + if (Boolean(cli) === Boolean(persona)) { + throw new Error('spawn requires exactly one of `cli` or `persona`.'); + } + if (persona && model) { + throw new Error('Persona harness and model come from the persona spec; omit `model`.'); + } + if (persona && session_ref) { + throw new Error('Persona session settings come from the persona launch plan; omit `session_ref`.'); + } const actionInput = { name, - cli, + ...(cli ? { cli } : { persona, capability: 'spawn:persona' }), ...(task ? { task } : {}), + ...(persona && cwd ? { cwd } : {}), ...(model ? { model } : {}), ...(session_ref ? { session_ref } : {}), ...(target_node ? { target_node } : {}), ...((channels ?? (channel ? [channel] : undefined)) ? { channels: channels ?? [channel] } : {}), }; - return jsonContent({ invocation: await actions.invoke('spawn', actionInput) }); + if (!persona) { + return jsonContent({ invocation: await actions.invoke('spawn', actionInput) }); + } + const session = getSession(); + const agentToken = as ? session.agents.get(as)?.agentToken : session.agentToken; + if (!agentToken) { + throw new Error('Persona spawn requires a registered agent identity.'); + } + const relay = new AgentRelay({ agentToken, baseUrl }); + const invocation = await relay.messaging.commands.invoke('spawn', actionInput); + return jsonContent({ + invocation: await waitForPersonaSpawn(relay.messaging.commands, invocation), + }); } ); diff --git a/packages/fleet/README.md b/packages/fleet/README.md index e1c62d840..961b59f61 100644 --- a/packages/fleet/README.md +++ b/packages/fleet/README.md @@ -73,6 +73,32 @@ await running.stop(); await serveNode({ definition, connection }); ``` +### AgentWorkforce personas + +The built-in `spawn:` capabilities launch raw harnesses. An +AgentWorkforce persona also carries its standing instructions, installed skills, +MCP servers, harness, model, and harness settings, so it must be resolved and +prepared as a unit on the target node. + +Use `defineWorkforcePersonaSpawnNode` from `@agentworkforce/local-surface` to +advertise the `spawn:persona` capability: + +```ts +import { serveNode } from '@agent-relay/fleet'; +import { defineWorkforcePersonaSpawnNode } from '@agentworkforce/local-surface'; + +const definition = defineWorkforcePersonaSpawnNode({ + nodeName: 'workforce-personas', + cwd: process.cwd(), +}); + +await serveNode({ definition, connection }); +``` + +Callers can then pass `persona` (an id or JSON path) instead of `cli` to the +Agent Relay `spawn` MCP tool. Persona spawns are SDK-backed and are reported as +successful only after node registration and the harness readiness handshake. + ### Logging The node runtime emits structured events — each capability it registers and every diff --git a/packages/fleet/src/serve-node.test.ts b/packages/fleet/src/serve-node.test.ts index fe2800424..39ae78b7c 100644 --- a/packages/fleet/src/serve-node.test.ts +++ b/packages/fleet/src/serve-node.test.ts @@ -279,6 +279,50 @@ describe('serveNode', () => { await running.stop(); }); + it('lets a spawn-prefixed action delegate to its resolved runtime instead of a shadow harness', async () => { + const node = defineNode({ + name: 'p', + capabilities: { + 'spawn:persona': action({}, async (_input, ctx) => + ctx.spawnAgent({ + agent: { + name: 'persona-worker', + runtime: 'pty', + cli: 'codex', + model: 'persona-model', + }, + }) + ), + }, + }); + const running = startServeNode({ definition: node, connection, reconnect: false }); + const sock = socket(); + sock.open(); + sock.emit(acceptAll(sock.lastRegister())); + await flush(); + + sock.emit({ + v: 1, + type: 'action.invoke', + invocation_id: 'inv_persona', + action: 'spawn:persona', + input: { persona: 'nango-integrations' }, + }); + await flush(); + + const [nodeSpawn] = sock.sentOfType('node.spawn'); + expect(nodeSpawn).toBeTruthy(); + expect(nodeSpawn.input).toMatchObject({ + name: 'persona-worker', + cli: 'codex', + model: 'persona-model', + }); + expect(nodeSpawn.input).not.toHaveProperty('capability'); + sock.emit({ v: 1, id: nodeSpawn.id, type: 'reply', ok: true, data: { name: 'persona-worker' } }); + await flush(); + await running.stop(); + }); + it('reconciles declared triggers with the injected client on registration', async () => { const node = defineNode({ name: 'p', diff --git a/packages/fleet/src/serve-node.ts b/packages/fleet/src/serve-node.ts index b7af3a44b..9af6fef2f 100644 --- a/packages/fleet/src/serve-node.ts +++ b/packages/fleet/src/serve-node.ts @@ -312,7 +312,8 @@ function makeContext( // harness identity lives in the capability name, not the handler's transformed // `cli` (which is the executable to run — for a stub, an arbitrary command), so // carry it as the delegated spawn's capacity key. - const shadowedHarness = capabilityName.startsWith('spawn:') + const capability = options.definition.capabilities[capabilityName]; + const shadowedHarness = capability?.kind === 'spawn' && capabilityName.startsWith('spawn:') ? capabilityName.slice('spawn:'.length) : undefined; return { diff --git a/packages/harness-driver/src/agent-handle.ts b/packages/harness-driver/src/agent-handle.ts index 032c86a56..3580cac4b 100644 --- a/packages/harness-driver/src/agent-handle.ts +++ b/packages/harness-driver/src/agent-handle.ts @@ -41,6 +41,17 @@ export interface AgentIdleInfo { exit?: AgentExitInfo; } +export interface AgentReadyInfo { + /** `'ready'` after worker_ready, `'exited'` if startup died, or `'timeout'`. */ + reason: 'ready' | 'exited' | 'timeout'; + /** Runtime reported by the ready handshake. */ + runtime?: AgentRuntime; + /** Harness process id reported by the ready handshake. */ + pid?: number; + /** Exit details when the worker died before readiness. */ + exit?: AgentExitInfo; +} + export interface AgentResultInfo { /** `'result'` on a submitted result, `'exited'` if the agent exited first, `'timeout'` otherwise. */ reason: 'result' | 'exited' | 'timeout'; @@ -93,6 +104,51 @@ export class SpawnedAgentHandle implements SpawnAgentResult { return this.exit?.signal; } + /** + * Resolve only after the broker has received the harness `worker_ready` + * handshake. A successful `/api/spawn` response proves process creation, + * not harness readiness, so callers that advertise a running agent should + * gate on this method and release on `exited` / `timeout`. + */ + waitForReady(timeoutMs = 90_000): Promise { + this.client.connectEvents(); + + return new Promise((resolve) => { + let timer: ReturnType | undefined; + let unsub: () => void = () => undefined; + let settled = false; + const settle = (info: AgentReadyInfo) => { + if (settled) return; + settled = true; + if (timer) clearTimeout(timer); + unsub(); + resolve(info); + }; + unsub = this.client.onEvent((event: BrokerEvent) => { + if (event.kind === 'worker_ready' && event.name === this.name) { + settle({ reason: 'ready', runtime: event.runtime, pid: event.pid }); + return; + } + const exit = matchExit(event, this.name); + if (exit) settle({ reason: 'exited', exit }); + }); + + // Subscribe before replaying history so worker_ready cannot land in the + // gap between the history check and listener registration. + const replayed = this.client.getLastEvent('worker_ready', this.name); + if (replayed?.kind === 'worker_ready') { + settle({ reason: 'ready', runtime: replayed.runtime, pid: replayed.pid }); + return; + } + const alreadyExited = this.exit; + if (alreadyExited) { + settle({ reason: 'exited', exit: alreadyExited }); + return; + } + timer = setTimeout(() => settle({ reason: 'timeout' }), timeoutMs); + }); + } + /** * Resolve when the agent exits (with `code` / `signal` when the broker reports * them), or with `{ reason: 'timeout' }` if `timeoutMs` elapses first. Replays diff --git a/packages/harness-driver/src/agent-result.test.ts b/packages/harness-driver/src/agent-result.test.ts index 8704c98fa..082a4e80e 100644 --- a/packages/harness-driver/src/agent-result.test.ts +++ b/packages/harness-driver/src/agent-result.test.ts @@ -107,6 +107,31 @@ describe('SpawnedAgentHandle.waitForResult', () => { // ── lifecycle helpers ────────────────────────────────────────────────────── describe('SpawnedAgentHandle lifecycle helpers', () => { + it('waits for the harness worker_ready handshake', async () => { + const stub = createStubClient(); + const handle = createHandle(stub); + + const pending = handle.waitForReady(); + stub.emit({ kind: 'worker_ready', name: 'someone-else', runtime: 'pty' } as BrokerEvent); + stub.emit({ kind: 'worker_ready', name: 'worker', runtime: 'pty', pid: 42 } as BrokerEvent); + + await expect(pending).resolves.toEqual({ reason: 'ready', runtime: 'pty', pid: 42 }); + }); + + it('reports an exit before readiness and times out when no handshake arrives', async () => { + const exitedStub = createStubClient(); + const exitedHandle = createHandle(exitedStub); + const exited = exitedHandle.waitForReady(); + exitedStub.emit({ kind: 'agent_exited', name: 'worker', code: 1 } as BrokerEvent); + await expect(exited).resolves.toEqual({ + reason: 'exited', + exit: { reason: 'exited', code: 1, signal: undefined }, + }); + + const timeoutHandle = createHandle(createStubClient()); + await expect(timeoutHandle.waitForReady(5)).resolves.toEqual({ reason: 'timeout' }); + }); + it('exposes prior exit info and replays it for waitForExit', async () => { const stub = createStubClient([ { kind: 'agent_exited', name: 'worker', code: 7, signal: 'SIGTERM' } as BrokerEvent, From 5ac1323bd4fa1bf8f103002a645c032323217245 Mon Sep 17 00:00:00 2001 From: "github-actions[bot]" Date: Fri, 7 Aug 2026 23:00:58 +0000 Subject: [PATCH 2/7] style: auto-format with Prettier --- packages/fleet/src/serve-node.ts | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git a/packages/fleet/src/serve-node.ts b/packages/fleet/src/serve-node.ts index 9af6fef2f..078f1a141 100644 --- a/packages/fleet/src/serve-node.ts +++ b/packages/fleet/src/serve-node.ts @@ -313,9 +313,10 @@ function makeContext( // `cli` (which is the executable to run — for a stub, an arbitrary command), so // carry it as the delegated spawn's capacity key. const capability = options.definition.capabilities[capabilityName]; - const shadowedHarness = capability?.kind === 'spawn' && capabilityName.startsWith('spawn:') - ? capabilityName.slice('spawn:'.length) - : undefined; + const shadowedHarness = + capability?.kind === 'spawn' && capabilityName.startsWith('spawn:') + ? capabilityName.slice('spawn:'.length) + : undefined; return { node: { ...info, From 8696810e9f30b4e4e79a8d242fea550a870985aa Mon Sep 17 00:00:00 2001 From: Proactive Runtime Bot Date: Sat, 8 Aug 2026 01:03:58 +0200 Subject: [PATCH 3/7] feat: advertise dynamic spawn delegation --- packages/fleet/src/index.ts | 3 +++ 1 file changed, 3 insertions(+) diff --git a/packages/fleet/src/index.ts b/packages/fleet/src/index.ts index 3e081b21a..f212df830 100644 --- a/packages/fleet/src/index.ts +++ b/packages/fleet/src/index.ts @@ -8,6 +8,9 @@ import type { JsonValue, } from '@agent-relay/harness-driver/protocol'; +/** Runtime compatibility marker for dynamic `spawn:*` action delegation. */ +export const FLEET_DYNAMIC_SPAWN_DELEGATION = true; + export type MaybePromise = T | Promise; export interface FleetNodeInfo { From 924aeacdf1ba97511425d0f61a592beed66825fd Mon Sep 17 00:00:00 2001 From: Miya Date: Sat, 8 Aug 2026 22:35:47 +0200 Subject: [PATCH 4/7] fix: harden verified persona spawn lifecycle --- crates/broker/src/runtime/fleet.rs | 23 ++++++----- crates/broker/src/runtime/maintenance.rs | 39 ++++++++++++++----- crates/broker/src/runtime/relaycast_events.rs | 5 ++- crates/broker/src/runtime/worker_events.rs | 22 ++++++----- .../src/cli/agent-relay-mcp.startup.test.ts | 12 ++++++ packages/cli/src/cli/agent-relay-mcp.ts | 13 ++++++- packages/harness-driver/src/agent-handle.ts | 5 ++- 7 files changed, 86 insertions(+), 33 deletions(-) diff --git a/crates/broker/src/runtime/fleet.rs b/crates/broker/src/runtime/fleet.rs index 1c7deb1cb..fd6354ab7 100644 --- a/crates/broker/src/runtime/fleet.rs +++ b/crates/broker/src/runtime/fleet.rs @@ -530,16 +530,19 @@ impl BrokerRuntime { } else { // A registration can succeed before process creation fails. Undo // that authoritative identity before reporting the failed launch. - let _ = - deregister_fleet_agent(&self.fleet_control_tx, &self.fleet_delivery_book, &name) - .await; - prune_fleet_agent_state( - &self.fleet_control_tx, - &mut self.fleet_inventory, - &mut self.fleet_delivery_book, - &name, - ) - .await; + 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; } diff --git a/crates/broker/src/runtime/maintenance.rs b/crates/broker/src/runtime/maintenance.rs index dae3f1041..e3b5d1c74 100644 --- a/crates/broker/src/runtime/maintenance.rs +++ b/crates/broker/src/runtime/maintenance.rs @@ -117,16 +117,19 @@ impl BrokerRuntime { agent_result_tokens, ) .await; - let _ = - super::fleet::deregister_fleet_agent(fleet_control_tx, fleet_delivery_book, name) - .await; - super::fleet::prune_fleet_agent_state( - fleet_control_tx, - fleet_inventory, - fleet_delivery_book, - name, - ) - .await; + match super::fleet::deregister_fleet_agent(fleet_control_tx, fleet_delivery_book, name).await { + Ok(_) => super::fleet::prune_fleet_agent_state( + fleet_control_tx, + fleet_inventory, + fleet_delivery_book, + name, + ) + .await, + Err(error) => { + tracing::warn!(worker = %name, %error, "retaining fleet identity after readiness timeout cleanup"); + super::fleet::prune_fleet_inventory_entry(fleet_control_tx, fleet_inventory, name).await; + } + } let _ = fleet_control_tx .send(FleetControlCommand::Send( crate::fleet_wire::BrokerToRelaycast::ActionResult( @@ -149,6 +152,22 @@ impl BrokerRuntime { let mut fleet_load_changed = !expired_verified_spawns.is_empty() || !exited.is_empty(); for (name, code, signal, exit_reason) in &exited { if let Some(pending) = pending_verified_spawns.remove(name) { + // A failed verified launch has no owner after its action is + // failed. Do not let the normal supervisor revive it later. + workers.supervisor.unregister(name); + match super::fleet::deregister_fleet_agent(fleet_control_tx, fleet_delivery_book, name).await { + Ok(_) => super::fleet::prune_fleet_agent_state( + fleet_control_tx, + fleet_inventory, + fleet_delivery_book, + name, + ) + .await, + Err(error) => { + tracing::warn!(worker = %name, %error, "retaining fleet identity after early verified-spawn exit"); + super::fleet::prune_fleet_inventory_entry(fleet_control_tx, fleet_inventory, name).await; + } + } let _ = fleet_control_tx .send(FleetControlCommand::Send( crate::fleet_wire::BrokerToRelaycast::ActionResult( diff --git a/crates/broker/src/runtime/relaycast_events.rs b/crates/broker/src/runtime/relaycast_events.rs index e85a6ef76..d86ddf9a7 100644 --- a/crates/broker/src/runtime/relaycast_events.rs +++ b/crates/broker/src/runtime/relaycast_events.rs @@ -489,8 +489,9 @@ pub(super) async fn spawn_worker_from_request( // the worker MCP never re-registers over HTTP. Falls back to HTTP // pre-registration when node binding is unavailable. let worker_relay_key = { - if let Some(token) = - relaycast_ws_spawn_token(ws_value).filter(|_| !require_node_registration) + if let Some(token) = relaycast_ws_spawn_token(ws_value).filter(|_| { + !require_node_registration && !relaycast_spawn_verifies_ready(ws_value) + }) { seed_supplied_agent_token(workspace_http, &name, &token); Some(token) diff --git a/crates/broker/src/runtime/worker_events.rs b/crates/broker/src/runtime/worker_events.rs index 7c77f0ac3..6b17aaf5d 100644 --- a/crates/broker/src/runtime/worker_events.rs +++ b/crates/broker/src/runtime/worker_events.rs @@ -614,6 +614,19 @@ impl BrokerRuntime { .map(str::to_string), ); } + // Resolve the verified Fleet action before optional SDK + // notifications and initial-task work. A congested SDK + // output queue must not turn a ready worker into an + // action timeout. + if let Some(pending) = pending_verified_spawns.remove(&name) { + let _ = fleet_control_tx + .send(FleetControlCommand::Send( + crate::fleet_wire::BrokerToRelaycast::ActionResult( + verified_spawn_ready_result(pending.invocation_id, &name), + ), + )) + .await; + } let _ = send_event( sdk_out_tx, json!({ @@ -885,15 +898,6 @@ impl BrokerRuntime { }), ) .await; - if let Some(pending) = pending_verified_spawns.remove(&name) { - let _ = fleet_control_tx - .send(FleetControlCommand::Send( - crate::fleet_wire::BrokerToRelaycast::ActionResult( - verified_spawn_ready_result(pending.invocation_id, &name), - ), - )) - .await; - } } else if msg_type == "agent_idle" { let idle_secs = value .get("payload") diff --git a/packages/cli/src/cli/agent-relay-mcp.startup.test.ts b/packages/cli/src/cli/agent-relay-mcp.startup.test.ts index c9c8168c4..688aa695a 100644 --- a/packages/cli/src/cli/agent-relay-mcp.startup.test.ts +++ b/packages/cli/src/cli/agent-relay-mcp.startup.test.ts @@ -544,6 +544,18 @@ describe('createAgentRelayMcpServer', () => { status: 'completed', output: { spawned: true, ready: true }, }); + expect(messaging.commands.invoke).toHaveBeenCalledWith( + expect.objectContaining({ + actionName: 'spawn', + actionInput: { + name: 'IntegrationExpert', + persona: 'nango-integrations', + task: 'Fix the sync', + cwd: '/workspace/project', + target_node: 'node-a', + }, + }) + ); const toolsList = await server.listToolsHandler?.({}, {}); expect(toolsList?.tools).toEqual([ { diff --git a/packages/cli/src/cli/agent-relay-mcp.ts b/packages/cli/src/cli/agent-relay-mcp.ts index 64d2c2080..cbca07f37 100644 --- a/packages/cli/src/cli/agent-relay-mcp.ts +++ b/packages/cli/src/cli/agent-relay-mcp.ts @@ -92,7 +92,18 @@ async function waitForPersonaSpawn( const deadline = Date.now() + timeoutMs; for (;;) { - const invocation = await actions.getInvocation(actionName, invocationId); + let invocation: unknown; + try { + invocation = await actions.getInvocation(actionName, invocationId); + } catch (error) { + const message = error instanceof Error ? error.message : String(error); + if (/invalid.?agent.?token|unauthori[sz]ed|forbidden/i.test(message)) throw error; + if (Date.now() >= deadline) { + throw new Error('Persona spawn timed out before broker registration and harness readiness.'); + } + await new Promise((resolve) => setTimeout(resolve, PERSONA_SPAWN_POLL_MS)); + continue; + } const record = recordValue(invocation); const status = invocationText(record, 'status')?.toLowerCase(); if (status === 'completed' || status === 'succeeded' || status === 'success') { diff --git a/packages/harness-driver/src/agent-handle.ts b/packages/harness-driver/src/agent-handle.ts index 3580cac4b..48ffa7f3d 100644 --- a/packages/harness-driver/src/agent-handle.ts +++ b/packages/harness-driver/src/agent-handle.ts @@ -72,6 +72,7 @@ export class SpawnedAgentHandle implements SpawnAgentResult { readonly runtime: AgentRuntime; readonly sessionId?: string; readonly pid?: number; + private readonly readyNotBefore = Date.now(); constructor( result: SpawnAgentResult, @@ -135,7 +136,9 @@ export class SpawnedAgentHandle implements SpawnAgentResult { // Subscribe before replaying history so worker_ready cannot land in the // gap between the history check and listener registration. - const replayed = this.client.getLastEvent('worker_ready', this.name); + const replayed = this.client + .queryEvents({ kind: 'worker_ready', name: this.name, since: this.readyNotBefore, limit: 1 }) + .at(-1); if (replayed?.kind === 'worker_ready') { settle({ reason: 'ready', runtime: replayed.runtime, pid: replayed.pid }); return; From 83448820dca7727538af62d94a64b42fb1ba678f Mon Sep 17 00:00:00 2001 From: "github-actions[bot]" Date: Sat, 8 Aug 2026 20:36:07 +0000 Subject: [PATCH 5/7] style: auto-format Rust code with cargo fmt --- crates/broker/src/runtime/fleet.rs | 27 ++++++--- crates/broker/src/runtime/maintenance.rs | 58 +++++++++++++------ crates/broker/src/runtime/relaycast_events.rs | 5 +- 3 files changed, 60 insertions(+), 30 deletions(-) diff --git a/crates/broker/src/runtime/fleet.rs b/crates/broker/src/runtime/fleet.rs index fd6354ab7..cb269ea49 100644 --- a/crates/broker/src/runtime/fleet.rs +++ b/crates/broker/src/runtime/fleet.rs @@ -530,17 +530,26 @@ impl BrokerRuntime { } 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, + 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; + prune_fleet_inventory_entry( + &self.fleet_control_tx, + &mut self.fleet_inventory, + &name, + ) + .await; } } self.reply_action_error(&invoke.invocation_id, "spawn_failed") diff --git a/crates/broker/src/runtime/maintenance.rs b/crates/broker/src/runtime/maintenance.rs index e3b5d1c74..0424b2c4e 100644 --- a/crates/broker/src/runtime/maintenance.rs +++ b/crates/broker/src/runtime/maintenance.rs @@ -117,17 +117,26 @@ impl BrokerRuntime { agent_result_tokens, ) .await; - match super::fleet::deregister_fleet_agent(fleet_control_tx, fleet_delivery_book, name).await { - Ok(_) => super::fleet::prune_fleet_agent_state( - fleet_control_tx, - fleet_inventory, - fleet_delivery_book, - name, - ) - .await, + match super::fleet::deregister_fleet_agent(fleet_control_tx, fleet_delivery_book, name) + .await + { + Ok(_) => { + super::fleet::prune_fleet_agent_state( + fleet_control_tx, + fleet_inventory, + fleet_delivery_book, + name, + ) + .await + } Err(error) => { tracing::warn!(worker = %name, %error, "retaining fleet identity after readiness timeout cleanup"); - super::fleet::prune_fleet_inventory_entry(fleet_control_tx, fleet_inventory, name).await; + super::fleet::prune_fleet_inventory_entry( + fleet_control_tx, + fleet_inventory, + name, + ) + .await; } } let _ = fleet_control_tx @@ -155,17 +164,30 @@ impl BrokerRuntime { // A failed verified launch has no owner after its action is // failed. Do not let the normal supervisor revive it later. workers.supervisor.unregister(name); - match super::fleet::deregister_fleet_agent(fleet_control_tx, fleet_delivery_book, name).await { - Ok(_) => super::fleet::prune_fleet_agent_state( - fleet_control_tx, - fleet_inventory, - fleet_delivery_book, - name, - ) - .await, + match super::fleet::deregister_fleet_agent( + fleet_control_tx, + fleet_delivery_book, + name, + ) + .await + { + Ok(_) => { + super::fleet::prune_fleet_agent_state( + fleet_control_tx, + fleet_inventory, + fleet_delivery_book, + name, + ) + .await + } Err(error) => { tracing::warn!(worker = %name, %error, "retaining fleet identity after early verified-spawn exit"); - super::fleet::prune_fleet_inventory_entry(fleet_control_tx, fleet_inventory, name).await; + super::fleet::prune_fleet_inventory_entry( + fleet_control_tx, + fleet_inventory, + name, + ) + .await; } } let _ = fleet_control_tx diff --git a/crates/broker/src/runtime/relaycast_events.rs b/crates/broker/src/runtime/relaycast_events.rs index d86ddf9a7..50ef9caec 100644 --- a/crates/broker/src/runtime/relaycast_events.rs +++ b/crates/broker/src/runtime/relaycast_events.rs @@ -489,9 +489,8 @@ pub(super) async fn spawn_worker_from_request( // the worker MCP never re-registers over HTTP. Falls back to HTTP // pre-registration when node binding is unavailable. let worker_relay_key = { - if let Some(token) = relaycast_ws_spawn_token(ws_value).filter(|_| { - !require_node_registration && !relaycast_spawn_verifies_ready(ws_value) - }) + if let Some(token) = relaycast_ws_spawn_token(ws_value) + .filter(|_| !require_node_registration && !relaycast_spawn_verifies_ready(ws_value)) { seed_supplied_agent_token(workspace_http, &name, &token); Some(token) From f8cb0b21c073fe1969d3658040eeb16d65aa4810 Mon Sep 17 00:00:00 2001 From: Miya Date: Sat, 8 Aug 2026 22:52:49 +0200 Subject: [PATCH 6/7] fix: complete verified persona spawn lifecycle --- crates/broker/src/runtime/fleet.rs | 27 ++++++-- crates/broker/src/runtime/maintenance.rs | 64 ++++++++++--------- crates/broker/src/runtime/relaycast_events.rs | 2 +- crates/broker/src/runtime/worker_events.rs | 26 ++++---- .../src/cli/agent-relay-mcp.startup.test.ts | 1 + packages/cli/src/cli/agent-relay-mcp.ts | 8 ++- packages/harness-driver/src/agent-handle.ts | 43 +++++++++---- .../harness-driver/src/agent-result.test.ts | 19 +++++- packages/harness-driver/src/client.ts | 6 +- 9 files changed, 128 insertions(+), 68 deletions(-) diff --git a/crates/broker/src/runtime/fleet.rs b/crates/broker/src/runtime/fleet.rs index cb269ea49..77bbdf78e 100644 --- a/crates/broker/src/runtime/fleet.rs +++ b/crates/broker/src/runtime/fleet.rs @@ -595,13 +595,26 @@ 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, diff --git a/crates/broker/src/runtime/maintenance.rs b/crates/broker/src/runtime/maintenance.rs index 0424b2c4e..773c4abe6 100644 --- a/crates/broker/src/runtime/maintenance.rs +++ b/crates/broker/src/runtime/maintenance.rs @@ -160,6 +160,7 @@ impl BrokerRuntime { }; let mut fleet_load_changed = !expired_verified_spawns.is_empty() || !exited.is_empty(); for (name, code, signal, exit_reason) in &exited { + let mut retain_fleet_identity = false; if let Some(pending) = pending_verified_spawns.remove(name) { // A failed verified launch has no owner after its action is // failed. Do not let the normal supervisor revive it later. @@ -171,23 +172,10 @@ impl BrokerRuntime { ) .await { - Ok(_) => { - super::fleet::prune_fleet_agent_state( - fleet_control_tx, - fleet_inventory, - fleet_delivery_book, - name, - ) - .await - } + Ok(_) => {} Err(error) => { tracing::warn!(worker = %name, %error, "retaining fleet identity after early verified-spawn exit"); - super::fleet::prune_fleet_inventory_entry( - fleet_control_tx, - fleet_inventory, - name, - ) - .await; + retain_fleet_identity = true; } } let _ = fleet_control_tx @@ -333,13 +321,22 @@ impl BrokerRuntime { tracing::warn!(path = %paths.state.display(), error = %error, "failed to persist broker state"); } } - super::fleet::prune_fleet_agent_state( - fleet_control_tx, - fleet_inventory, - fleet_delivery_book, - name, - ) - .await; + if retain_fleet_identity { + super::fleet::prune_fleet_inventory_entry( + fleet_control_tx, + fleet_inventory, + name, + ) + .await; + } else { + super::fleet::prune_fleet_agent_state( + fleet_control_tx, + fleet_inventory, + fleet_delivery_book, + name, + ) + .await; + } } None => { // Not supervised — original behavior @@ -398,13 +395,22 @@ impl BrokerRuntime { tracing::warn!(path = %paths.state.display(), error = %error, "failed to persist broker state"); } } - super::fleet::prune_fleet_agent_state( - fleet_control_tx, - fleet_inventory, - fleet_delivery_book, - name, - ) - .await; + if retain_fleet_identity { + super::fleet::prune_fleet_inventory_entry( + fleet_control_tx, + fleet_inventory, + name, + ) + .await; + } else { + super::fleet::prune_fleet_agent_state( + fleet_control_tx, + fleet_inventory, + fleet_delivery_book, + name, + ) + .await; + } } } } diff --git a/crates/broker/src/runtime/relaycast_events.rs b/crates/broker/src/runtime/relaycast_events.rs index 50ef9caec..bb5825bac 100644 --- a/crates/broker/src/runtime/relaycast_events.rs +++ b/crates/broker/src/runtime/relaycast_events.rs @@ -512,7 +512,7 @@ pub(super) async fn spawn_worker_from_request( Some(token.token) } Err(node_error) => { - if require_node_registration { + if require_node_registration || relaycast_spawn_verifies_ready(ws_value) { tracing::warn!( worker = %name, error = %node_error, diff --git a/crates/broker/src/runtime/worker_events.rs b/crates/broker/src/runtime/worker_events.rs index 6b17aaf5d..d9090ebcb 100644 --- a/crates/broker/src/runtime/worker_events.rs +++ b/crates/broker/src/runtime/worker_events.rs @@ -614,19 +614,6 @@ impl BrokerRuntime { .map(str::to_string), ); } - // Resolve the verified Fleet action before optional SDK - // notifications and initial-task work. A congested SDK - // output queue must not turn a ready worker into an - // action timeout. - if let Some(pending) = pending_verified_spawns.remove(&name) { - let _ = fleet_control_tx - .send(FleetControlCommand::Send( - crate::fleet_wire::BrokerToRelaycast::ActionResult( - verified_spawn_ready_result(pending.invocation_id, &name), - ), - )) - .await; - } let _ = send_event( sdk_out_tx, json!({ @@ -757,6 +744,19 @@ impl BrokerRuntime { .get(&name) .map(|handle| handle.spec.runtime == AgentRuntime::Pty) .unwrap_or(false); + // Resolve the verified Fleet action before optional SDK + // notifications and initial-task work. A congested SDK + // output queue must not turn a ready worker into an + // action timeout. + if let Some(pending) = pending_verified_spawns.remove(&name) { + let _ = fleet_control_tx + .send(FleetControlCommand::Send( + crate::fleet_wire::BrokerToRelaycast::ActionResult( + verified_spawn_ready_result(pending.invocation_id, &name), + ), + )) + .await; + } let interactive_hold_replayed = is_pty_worker && delivery_states .get(&name) diff --git a/packages/cli/src/cli/agent-relay-mcp.startup.test.ts b/packages/cli/src/cli/agent-relay-mcp.startup.test.ts index 688aa695a..f41a79892 100644 --- a/packages/cli/src/cli/agent-relay-mcp.startup.test.ts +++ b/packages/cli/src/cli/agent-relay-mcp.startup.test.ts @@ -550,6 +550,7 @@ describe('createAgentRelayMcpServer', () => { actionInput: { name: 'IntegrationExpert', persona: 'nango-integrations', + capability: 'spawn:persona', task: 'Fix the sync', cwd: '/workspace/project', target_node: 'node-a', diff --git a/packages/cli/src/cli/agent-relay-mcp.ts b/packages/cli/src/cli/agent-relay-mcp.ts index cbca07f37..fb8364325 100644 --- a/packages/cli/src/cli/agent-relay-mcp.ts +++ b/packages/cli/src/cli/agent-relay-mcp.ts @@ -17,6 +17,7 @@ import { createAgentClient, createRealtimeClient, createWorkspaceClient, + isInvalidAgentTokenError, } from '@agent-relay/sdk'; import { z } from 'zod'; import { initTelemetry, shutdown as shutdownTelemetry } from './telemetry/index.js'; @@ -97,7 +98,12 @@ async function waitForPersonaSpawn( invocation = await actions.getInvocation(actionName, invocationId); } catch (error) { const message = error instanceof Error ? error.message : String(error); - if (/invalid.?agent.?token|unauthori[sz]ed|forbidden/i.test(message)) throw error; + if ( + isInvalidAgentTokenError(error) || + /invalid.?agent.?token|unauthori[sz]ed|forbidden/i.test(message) + ) { + throw error; + } if (Date.now() >= deadline) { throw new Error('Persona spawn timed out before broker registration and harness readiness.'); } diff --git a/packages/harness-driver/src/agent-handle.ts b/packages/harness-driver/src/agent-handle.ts index 48ffa7f3d..c6edacb19 100644 --- a/packages/harness-driver/src/agent-handle.ts +++ b/packages/harness-driver/src/agent-handle.ts @@ -23,6 +23,8 @@ import type { HarnessDriverClient } from './client.js'; import type { AgentRuntime, BrokerEvent } from './protocol.js'; import type { SpawnAgentResult } from './types.js'; +type SequencedBrokerEvent = BrokerEvent & { seq?: number }; + export interface AgentExitInfo { /** `'exited'` when the agent exited; `'timeout'` when the wait elapsed first. */ reason: 'exited' | 'timeout'; @@ -72,11 +74,12 @@ export class SpawnedAgentHandle implements SpawnAgentResult { readonly runtime: AgentRuntime; readonly sessionId?: string; readonly pid?: number; - private readonly readyNotBefore = Date.now(); constructor( result: SpawnAgentResult, - private readonly client: HarnessDriverClient + private readonly client: HarnessDriverClient, + /** Last broker event sequence observed immediately before the spawn request. */ + private readonly eventSeqBeforeSpawn = 0 ) { this.name = result.name; this.runtime = result.runtime; @@ -86,11 +89,11 @@ export class SpawnedAgentHandle implements SpawnAgentResult { /** Exit info if the agent has already exited (from broker event history), else `undefined`. */ get exit(): AgentExitInfo | undefined { - const exited = this.client.getLastEvent('agent_exited', this.name); + const exited = this.lastEvent('agent_exited'); if (exited && exited.kind === 'agent_exited') { return { reason: 'exited', code: exited.code, signal: exited.signal }; } - const exit = this.client.getLastEvent('agent_exit', this.name); + const exit = this.lastEvent('agent_exit'); if (exit && exit.kind === 'agent_exit') { return { reason: 'exited' }; } @@ -126,18 +129,19 @@ export class SpawnedAgentHandle implements SpawnAgentResult { resolve(info); }; unsub = this.client.onEvent((event: BrokerEvent) => { - if (event.kind === 'worker_ready' && event.name === this.name) { + if (this.isCurrentGeneration(event) && event.kind === 'worker_ready' && event.name === this.name) { settle({ reason: 'ready', runtime: event.runtime, pid: event.pid }); return; } - const exit = matchExit(event, this.name); + const exit = this.isCurrentGeneration(event) ? matchExit(event, this.name) : undefined; if (exit) settle({ reason: 'exited', exit }); }); // Subscribe before replaying history so worker_ready cannot land in the // gap between the history check and listener registration. const replayed = this.client - .queryEvents({ kind: 'worker_ready', name: this.name, since: this.readyNotBefore, limit: 1 }) + .queryEvents({ kind: 'worker_ready', name: this.name }) + .filter((event) => this.isCurrentGeneration(event)) .at(-1); if (replayed?.kind === 'worker_ready') { settle({ reason: 'ready', runtime: replayed.runtime, pid: replayed.pid }); @@ -173,7 +177,7 @@ export class SpawnedAgentHandle implements SpawnAgentResult { resolve(info); }; const unsub = this.client.onEvent((event: BrokerEvent) => { - const exit = matchExit(event, this.name); + const exit = this.isCurrentGeneration(event) ? matchExit(event, this.name) : undefined; if (exit) settle(exit); }); if (timeoutMs !== undefined) { @@ -205,11 +209,11 @@ export class SpawnedAgentHandle implements SpawnAgentResult { // `agentIdle` event bus is only populated by call-site hooks (not broker // events) in direct-client usage, so it must not be used here. const unsub = this.client.onEvent((event: BrokerEvent) => { - if (event.kind === 'agent_idle' && event.name === this.name) { + if (this.isCurrentGeneration(event) && event.kind === 'agent_idle' && event.name === this.name) { settle({ reason: 'idle', idleSecs: event.idle_secs }); return; } - const exit = matchExit(event, this.name); + const exit = this.isCurrentGeneration(event) ? matchExit(event, this.name) : undefined; if (exit) settle({ reason: 'exited', exit }); }); if (timeoutMs !== undefined) { @@ -235,7 +239,7 @@ export class SpawnedAgentHandle implements SpawnAgentResult { .queryEvents({ kind: 'agent_result', name: this.name }) .find( (event): event is Extract => - event.kind === 'agent_result' && event.name === this.name + this.isCurrentGeneration(event) && event.kind === 'agent_result' && event.name === this.name ); if (replayed) { return Promise.resolve(toResultInfo(replayed)); @@ -251,11 +255,11 @@ export class SpawnedAgentHandle implements SpawnAgentResult { resolve(info); }; const unsub = this.client.onEvent((event: BrokerEvent) => { - if (event.kind === 'agent_result' && event.name === this.name) { + if (this.isCurrentGeneration(event) && event.kind === 'agent_result' && event.name === this.name) { settle(toResultInfo(event)); return; } - const exit = matchExit(event, this.name); + const exit = this.isCurrentGeneration(event) ? matchExit(event, this.name) : undefined; if (exit) settle({ reason: 'exited', exit }); }); if (timeoutMs !== undefined) { @@ -268,6 +272,19 @@ export class SpawnedAgentHandle implements SpawnAgentResult { release(reason?: string): Promise<{ name: string }> { return this.client.release(this.name, reason); } + + private isCurrentGeneration(event: BrokerEvent): boolean { + if (this.eventSeqBeforeSpawn === 0) return true; + const seq = (event as SequencedBrokerEvent).seq; + return typeof seq === 'number' && seq > this.eventSeqBeforeSpawn; + } + + private lastEvent(kind: string): BrokerEvent | undefined { + return this.client + .queryEvents({ kind, name: this.name }) + .filter((event) => this.isCurrentGeneration(event)) + .at(-1); + } } /** Map an `agent_result` broker event to the public result info shape. */ diff --git a/packages/harness-driver/src/agent-result.test.ts b/packages/harness-driver/src/agent-result.test.ts index 082a4e80e..2dda22b54 100644 --- a/packages/harness-driver/src/agent-result.test.ts +++ b/packages/harness-driver/src/agent-result.test.ts @@ -47,8 +47,12 @@ function createStubClient(history: BrokerEvent[] = []) { return stub; } -function createHandle(stub: ReturnType, name = 'worker') { - return new SpawnedAgentHandle({ name, runtime: 'pty' }, stub as unknown as HarnessDriverClient); +function createHandle(stub: ReturnType, name = 'worker', eventSeqBeforeSpawn = 0) { + return new SpawnedAgentHandle( + { name, runtime: 'pty' }, + stub as unknown as HarnessDriverClient, + eventSeqBeforeSpawn + ); } const resultEvent = (name: string, data: unknown, final = true): BrokerEvent => @@ -118,6 +122,17 @@ describe('SpawnedAgentHandle lifecycle helpers', () => { await expect(pending).resolves.toEqual({ reason: 'ready', runtime: 'pty', pid: 42 }); }); + it("ignores a reused name's prior generation when replaying worker_ready", async () => { + const staleReady = { kind: 'worker_ready', name: 'worker', runtime: 'pty', seq: 12 } as BrokerEvent; + const stub = createStubClient([staleReady]); + const handle = createHandle(stub, 'worker', 12); + + const pending = handle.waitForReady(); + stub.emit({ kind: 'worker_ready', name: 'worker', runtime: 'pty', pid: 43, seq: 13 } as BrokerEvent); + + await expect(pending).resolves.toEqual({ reason: 'ready', runtime: 'pty', pid: 43 }); + }); + it('reports an exit before readiness and times out when no handshake arrives', async () => { const exitedStub = createStubClient(); const exitedHandle = createHandle(exitedStub); diff --git a/packages/harness-driver/src/client.ts b/packages/harness-driver/src/client.ts index 71a867c96..0e5c8ced8 100644 --- a/packages/harness-driver/src/client.ts +++ b/packages/harness-driver/src/client.ts @@ -578,13 +578,14 @@ export class HarnessDriverClient { const t0 = Date.now(); const resolvedInput = await this.runBeforeSpawn(beforeCtx); try { + const eventSeqBeforeSpawn = await this.currentEventSeq().catch(() => 0); const rawResult = await this.transport.request('/api/spawn', { method: 'POST', body: JSON.stringify(buildSpawnPtyBody(resolvedInput)), }); const result = SpawnAgentResultSchema.parse(rawResult); await this.emitAfterSpawn(beforeCtx, resolvedInput, t0, result, undefined); - return new SpawnedAgentHandle(result, this); + return new SpawnedAgentHandle(result, this, eventSeqBeforeSpawn); } catch (err) { await this.emitAfterSpawn(beforeCtx, resolvedInput, t0, undefined, err); throw err; @@ -620,13 +621,14 @@ export class HarnessDriverClient { } try { + const eventSeqBeforeSpawn = await this.currentEventSeq().catch(() => 0); const rawResult = await this.transport.request('/api/spawn', { method: 'POST', body: JSON.stringify(buildSpawnCliBody(resolvedInput, transport)), }); const result = SpawnAgentResultSchema.parse(rawResult); await this.emitAfterSpawn(beforeCtx, resolvedInput, t0, result, undefined); - return new SpawnedAgentHandle(result, this); + return new SpawnedAgentHandle(result, this, eventSeqBeforeSpawn); } catch (err) { await this.emitAfterSpawn(beforeCtx, resolvedInput, t0, undefined, err); throw err; From bbb70df78880e4b2079363807b19d0198a14a561 Mon Sep 17 00:00:00 2001 From: "github-actions[bot]" Date: Sat, 8 Aug 2026 20:54:08 +0000 Subject: [PATCH 7/7] style: auto-format Rust code with cargo fmt --- crates/broker/src/runtime/fleet.rs | 16 ++++++++++++---- 1 file changed, 12 insertions(+), 4 deletions(-) diff --git a/crates/broker/src/runtime/fleet.rs b/crates/broker/src/runtime/fleet.rs index 77bbdf78e..5050a9b4b 100644 --- a/crates/broker/src/runtime/fleet.rs +++ b/crates/broker/src/runtime/fleet.rs @@ -596,8 +596,12 @@ impl BrokerRuntime { self.pty_observability.remove(&name); if outcome == super::relaycast_events::ReleaseOutcome::Released { - match deregister_fleet_agent(&self.fleet_control_tx, &mut self.fleet_delivery_book, &name) - .await + match deregister_fleet_agent( + &self.fleet_control_tx, + &mut self.fleet_delivery_book, + &name, + ) + .await { Ok(_) => { prune_fleet_agent_state( @@ -610,8 +614,12 @@ impl BrokerRuntime { } 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; + prune_fleet_inventory_entry( + &self.fleet_control_tx, + &mut self.fleet_inventory, + &name, + ) + .await; } } }