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..5050a9b4b 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,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 => { 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..773c4abe6 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,62 @@ 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; + 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( + 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 +158,37 @@ 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 { + 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. + workers.supervisor.unregister(name); + match super::fleet::deregister_fleet_agent( + fleet_control_tx, + fleet_delivery_book, + name, + ) + .await + { + Ok(_) => {} + Err(error) => { + tracing::warn!(worker = %name, %error, "retaining fleet identity after early verified-spawn exit"); + retain_fleet_identity = true; + } + } + 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 @@ -234,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 @@ -299,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 294bd61b8..bb5825bac 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 && !relaycast_spawn_verifies_ready(ws_value)) + { 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 || relaycast_spawn_verifies_ready(ws_value) { + 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..d9090ebcb 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; @@ -743,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 eedf4393d..f41a79892 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,32 @@ 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 }, + }); + expect(messaging.commands.invoke).toHaveBeenCalledWith( + expect.objectContaining({ + actionName: 'spawn', + actionInput: { + name: 'IntegrationExpert', + persona: 'nango-integrations', + capability: 'spawn:persona', + 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 d24af51b3..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'; @@ -62,6 +63,68 @@ 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 (;;) { + let invocation: unknown; + try { + invocation = await actions.getInvocation(actionName, invocationId); + } catch (error) { + const message = error instanceof Error ? error.message : String(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.'); + } + 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') { + 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 +149,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 +756,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 +785,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/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 { 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..078f1a141 100644 --- a/packages/fleet/src/serve-node.ts +++ b/packages/fleet/src/serve-node.ts @@ -312,9 +312,11 @@ 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:') - ? capabilityName.slice('spawn:'.length) - : undefined; + const capability = options.definition.capabilities[capabilityName]; + const shadowedHarness = + capability?.kind === 'spawn' && capabilityName.startsWith('spawn:') + ? capabilityName.slice('spawn:'.length) + : undefined; return { node: { ...info, diff --git a/packages/harness-driver/src/agent-handle.ts b/packages/harness-driver/src/agent-handle.ts index 032c86a56..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'; @@ -41,6 +43,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'; @@ -64,7 +77,9 @@ export class SpawnedAgentHandle implements SpawnAgentResult { 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; @@ -74,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' }; } @@ -93,6 +108,54 @@ 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 (this.isCurrentGeneration(event) && event.kind === 'worker_ready' && event.name === this.name) { + settle({ reason: 'ready', runtime: event.runtime, pid: event.pid }); + return; + } + 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 }) + .filter((event) => this.isCurrentGeneration(event)) + .at(-1); + 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 @@ -114,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) { @@ -146,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) { @@ -176,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)); @@ -192,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) { @@ -209,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 8704c98fa..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 => @@ -107,6 +111,42 @@ 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("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); + 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, 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;