diff --git a/CHANGELOG.md b/CHANGELOG.md index 3ea13aa6e..3d1e0f0b5 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -17,6 +17,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Fixed - `agent-relay node up --background` now preserves persisted Cloud enrollment credentials and node identity through detached startup, and fails instead of reporting healthy when the enrolled node cannot connect. +- `agent-relay-broker` now waits for a worker readiness handshake before completing a fleet spawn action, so a CLI that exits during startup produces `action.failed` instead of a false `spawned: true` result. +- Strict-name MCP registration now fails atomically when another session already owns the identity, rather than rotating that session's token and causing both sessions to invalidate each other. +- Broker dead-lettering now emits a structured warning with the worker, delivery, attempt count, and reason, making persisted undeliverable messages visible to operators. ## [10.6.1] - 2026-07-16 diff --git a/crates/broker/src/runtime/dead_letter.rs b/crates/broker/src/runtime/dead_letter.rs index d06f8c28f..f2617f04a 100644 --- a/crates/broker/src/runtime/dead_letter.rs +++ b/crates/broker/src/runtime/dead_letter.rs @@ -230,6 +230,15 @@ pub(crate) async fn dead_letter_pending_delivery( attempts: entry.attempts, reason: entry.reason.clone(), }; + tracing::warn!( + target = "agent_relay::broker", + worker = %entry.worker_name, + delivery_id = %entry.delivery.delivery_id, + event_id = %entry.delivery.event_id, + attempts = entry.attempts, + reason = %entry.reason, + "delivery moved to dead-letter queue" + ); dead_letters.push(entry); let _ = send_broker_event(sdk_out_tx, event).await; } diff --git a/crates/broker/src/worker.rs b/crates/broker/src/worker.rs index fafde5f06..37fa63d46 100644 --- a/crates/broker/src/worker.rs +++ b/crates/broker/src/worker.rs @@ -23,7 +23,7 @@ use serde_json::{json, Value}; use tokio::{ io::{AsyncBufReadExt, AsyncWriteExt, BufReader}, process::{Child, ChildStdin, Command}, - sync::mpsc, + sync::{mpsc, oneshot}, time::timeout, }; @@ -41,6 +41,11 @@ const APP_SERVER_AUTH_ENV_KEYS: [&str; 4] = [ ]; const DEFAULT_RELEASE_GRACE: Duration = Duration::from_secs(2); const APP_SERVER_RELEASE_GRACE: Duration = Duration::from_secs(35); +/// A spawned shim must prove it can receive `init_worker` and report readiness +/// before the caller is told that the agent exists. PTY workers have their own +/// 25-second readiness fallback, so this leaves enough headroom for that frame +/// to traverse the broker pipe. +const WORKER_STARTUP_READY_TIMEOUT: Duration = Duration::from_secs(30); // Working/idle activity inference from PTY output comes from the // harness-agnostic `relay-pty` crate. @@ -89,6 +94,7 @@ pub(crate) struct WorkerRegistry { event_tx: mpsc::Sender, worker_env: Vec<(String, String)>, worker_logs_dir: PathBuf, + worker_program_override: Option, pub(crate) initial_tasks: HashMap, pub(crate) supervisor: Supervisor, pub(crate) metrics: MetricsCollector, @@ -114,12 +120,32 @@ impl WorkerRegistry { event_tx, worker_env, worker_logs_dir, + worker_program_override: None, initial_tasks: HashMap::new(), supervisor: Supervisor::new(), metrics: MetricsCollector::new(broker_start), } } + #[cfg(test)] + fn set_worker_program_override(&mut self, program: PathBuf) { + self.worker_program_override = Some(program); + } + + async fn remove_failed_startup(&mut self, name: &WorkerName) { + let Some(mut handle) = self.workers.remove(name) else { + return; + }; + self.initial_tasks.remove(name); + if let Err(error) = terminate_child(&mut handle.child, DEFAULT_RELEASE_GRACE).await { + tracing::warn!( + worker = %name, + error = %error, + "failed to terminate worker after startup readiness failure" + ); + } + } + pub(crate) fn worker_log_path(&self, worker_name: &str) -> Option { // Reject path traversal: slashes, backslashes, null bytes, and ".." components if worker_name.contains('/') @@ -297,8 +323,11 @@ impl WorkerRegistry { "spawning worker" ); - let mut command = - Command::new(std::env::current_exe().context("failed to locate current executable")?); + let worker_program = self.worker_program_override.clone(); + let mut command = Command::new( + worker_program + .unwrap_or(std::env::current_exe().context("failed to locate current executable")?), + ); let mut harness_env: Vec<(String, String)> = Vec::new(); let mut suppress_worker_env: Vec<&'static str> = Vec::new(); let mut initial_harness_pid: Option = None; @@ -800,6 +829,7 @@ impl WorkerRegistry { let stdout = child.stdout.take().context("worker missing stdout pipe")?; let stderr = child.stderr.take().context("worker missing stderr pipe")?; let log_file = self.worker_log_path(&spec.name); + let (startup_ready_tx, startup_ready_rx) = oneshot::channel(); spawn_worker_reader( self.event_tx.clone(), @@ -808,6 +838,7 @@ impl WorkerRegistry { stdout, true, log_file.clone(), + Some(startup_ready_tx), ); spawn_worker_reader( self.event_tx.clone(), @@ -816,6 +847,7 @@ impl WorkerRegistry { stderr, false, log_file, + None, ); let handle = WorkerHandle { @@ -833,15 +865,38 @@ impl WorkerRegistry { }; self.workers.insert(spec.name.clone(), handle); - self.send_to_worker( - &spec.name, - "init_worker", - None, - json!({ - "agent": spec, - }), - ) - .await?; + if let Err(error) = self + .send_to_worker( + &spec.name, + "init_worker", + None, + json!({ + "agent": spec, + }), + ) + .await + { + self.remove_failed_startup(&spec.name).await; + return Err(error).context("failed to initialise worker during startup"); + } + + let startup_result = match timeout(WORKER_STARTUP_READY_TIMEOUT, startup_ready_rx).await { + Ok(Ok(Ok(()))) => Ok(()), + Ok(Ok(Err(error))) => Err(anyhow::anyhow!(error)), + Ok(Err(_)) => Err(anyhow::anyhow!( + "worker '{}' closed its startup stream before worker_ready", + spec.name + )), + Err(_) => Err(anyhow::anyhow!( + "worker '{}' did not emit worker_ready within {} seconds", + spec.name, + WORKER_STARTUP_READY_TIMEOUT.as_secs() + )), + }; + if let Err(error) = startup_result { + self.remove_failed_startup(&spec.name).await; + return Err(error).context("worker failed startup readiness handshake"); + } tracing::info!( target = "broker::spawn", @@ -1486,6 +1541,7 @@ fn spawn_worker_reader( reader: R, parse_json: bool, log_file_path: Option, + startup_ready_tx: Option>>, ) where R: tokio::io::AsyncRead + Unpin + Send + 'static, { @@ -1535,6 +1591,7 @@ fn spawn_worker_reader( } tokio::spawn(async move { + let mut startup_ready_tx = startup_ready_tx; let mut log_file = match log_file_path.as_ref() { Some(path) => match tokio::fs::OpenOptions::new() .create(true) @@ -1562,6 +1619,21 @@ fn spawn_worker_reader( while let Ok(Some(line)) = lines.next_line().await { if parse_json { if let Ok(value) = serde_json::from_str::(&line) { + let msg_type = value.get("type").and_then(Value::as_str); + if let Some(readiness_tx) = startup_ready_tx.take() { + match msg_type { + Some("worker_ready") => { + let _ = readiness_tx.send(Ok(())); + } + Some("worker_exited") => { + let _ = readiness_tx.send(Err(format!( + "worker '{}' exited before worker_ready", + name + ))); + } + _ => startup_ready_tx = Some(readiness_tx), + } + } if value .get("type") .and_then(Value::as_str) @@ -1636,13 +1708,19 @@ fn spawn_worker_reader( break; } } + if let Some(readiness_tx) = startup_ready_tx { + let _ = readiness_tx.send(Err(format!( + "worker '{}' closed its startup stream before worker_ready", + name + ))); + } }); } #[cfg(test)] mod tests { use super::*; - use crate::protocol::{AppServerHarnessAuth, AppServerHarnessHost}; + use crate::protocol::{AppServerHarnessAuth, AppServerHarnessHost, PtyHarnessConfig}; fn make_registry(env: Vec<(String, String)>) -> WorkerRegistry { let (tx, _rx) = mpsc::channel::(16); @@ -1688,6 +1766,72 @@ mod tests { assert_eq!(reg.env_value("MISSING"), None); } + #[cfg(unix)] + #[tokio::test] + async fn immediately_exiting_cli_is_not_reported_as_spawned() { + use std::os::unix::fs::PermissionsExt; + + let temp = tempfile::tempdir().expect("temporary worker test directory"); + let shim = temp.path().join("pty-shim"); + // The broker launches its own `pty` subcommand. This shim waits for + // `init_worker`, then executes the deliberately failing CLI named in + // the harness config below (`false`). Before the readiness gate, this + // sequence returned `Ok` after the OS-level process spawn. + std::fs::write( + &shim, + "#!/bin/sh\nIFS= read -r init\nwhile [ \"$1\" != \"false\" ]; do shift; done\nexec \"$@\"\n", + ) + .expect("write pty shim"); + let mut permissions = std::fs::metadata(&shim) + .expect("pty shim metadata") + .permissions(); + permissions.set_mode(0o755); + std::fs::set_permissions(&shim, permissions).expect("make pty shim executable"); + + let mut registry = make_registry(Vec::new()); + registry.set_worker_program_override(shim); + let spec = AgentSpec { + name: WorkerName::from("immediate-exit"), + runtime: AgentRuntime::Pty, + provider: None, + cli: Some("false".to_string()), + session_id: None, + harness_config: Some(ResolvedHarnessConfig::Pty(PtyHarnessConfig { + command: "false".to_string(), + args: Vec::new(), + cwd: None, + env: None, + session_id: None, + delivery: None, + metadata: None, + })), + model: None, + cwd: None, + team: None, + shadow_of: None, + shadow_mode: None, + args: Vec::new(), + channels: Vec::new(), + restart_policy: None, + }; + + let result = tokio::time::timeout( + Duration::from_secs(2), + registry.spawn(spec, None, None, None, true, None, None), + ) + .await + .expect("startup readiness must resolve promptly"); + + assert!( + result.is_err(), + "a CLI that exits before emitting worker_ready must not be reported as spawned" + ); + assert!( + !registry.has_worker("immediate-exit"), + "failed startup must not leave a dead worker registered" + ); + } + fn make_app_server_config() -> HeadlessHarnessConfig { HeadlessHarnessConfig { driver: HeadlessHarnessDriver::AppServer, 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 1673dda51..2e56e05bd 100644 --- a/packages/cli/src/cli/agent-relay-mcp.startup.test.ts +++ b/packages/cli/src/cli/agent-relay-mcp.startup.test.ts @@ -30,6 +30,7 @@ async function loadAgentRelayMcpModule(options: LoadOptions = {}) { const telemetryShutdown = vi.fn(async () => undefined); const relayInstances: Array<{ config: Record; + register: ReturnType; registerOrRotate: ReturnType; agentsList: ReturnType; nodesList: ReturnType; @@ -164,6 +165,7 @@ async function loadAgentRelayMcpModule(options: LoadOptions = {}) { }); const RelayCast = vi.fn(function (this: unknown, config: Record) { + const register = vi.fn(async (input: { name: string; type?: string }) => behavior.registerImpl(input)); const registerOrRotate = vi.fn(async (input: { name: string; type?: string }) => behavior.registerImpl(input) ); @@ -183,9 +185,10 @@ async function loadAgentRelayMcpModule(options: LoadOptions = {}) { reason: input.reason ?? null, })); const as = vi.fn((token: string) => createAgentClient(token)); - relayInstances.push({ config, registerOrRotate, agentsList, nodesList, spawn, release, as }); + relayInstances.push({ config, register, registerOrRotate, agentsList, nodesList, spawn, release, as }); return { agents: { + register, registerOrRotate, list: agentsList, spawn, @@ -806,6 +809,26 @@ describe('resolveStdioBootstrapOptions', () => { }); expect(result.agentToken).toBe('at_live_minted'); }); + + it('uses conflict-safe registration for a strict worker bootstrap', async () => { + const { mod, mocks } = await loadAgentRelayMcpModule(); + + await mod.resolveStdioBootstrapOptions({ + apiKey: 'rk_live_workspace', + agentName: 'WorkerA', + agentType: 'agent', + strictAgentName: true, + }); + + const bootstrapRelay = mocks.relayInstances.find( + (instance) => instance.config.apiKey === 'rk_live_workspace' + ); + expect(bootstrapRelay?.register).toHaveBeenCalledWith({ + name: 'WorkerA', + type: 'agent', + }); + expect(bootstrapRelay?.registerOrRotate).not.toHaveBeenCalled(); + }); }); describe('startAgentRelayMcpStdio', () => { diff --git a/packages/cli/src/cli/agent-relay-mcp.test.ts b/packages/cli/src/cli/agent-relay-mcp.test.ts index f35dd4d04..0d08991f1 100644 --- a/packages/cli/src/cli/agent-relay-mcp.test.ts +++ b/packages/cli/src/cli/agent-relay-mcp.test.ts @@ -37,17 +37,19 @@ describe('registerAgentWithRebind', () => { }); }); - it('re-registers when the strict-named identity was dropped from the agents map', async () => { + it('uses conflict-safe registration when the strict-named identity was dropped from the agents map', async () => { // After an `agent_token_invalid` recovery, the active token is null and // the identity is missing from session.agents. The short-circuit must - // fall through to registerOrRotate instead of handing back the dead token. + // fall through to plain registration instead of handing back the dead + // token or rotating a different live session's token. const setSession = vi.fn(); - const registerOrRotate = vi.fn().mockResolvedValue({ + const register = vi.fn().mockResolvedValue({ id: 'agent_456', name: 'WorkerA', token: 'at_live_fresh', status: 'online', }); + const registerOrRotate = vi.fn(); const payload = await registerAgentWithRebind({ session: { @@ -59,14 +61,15 @@ describe('registerAgentWithRebind', () => { setSession, getRelay: () => ({ - agents: { registerOrRotate }, + agents: { register, registerOrRotate }, }) as never, name: 'WorkerA', strictAgentName: true, preferredAgentName: 'WorkerA', }); - expect(registerOrRotate).toHaveBeenCalledOnce(); + expect(register).toHaveBeenCalledOnce(); + expect(registerOrRotate).not.toHaveBeenCalled(); expect(payload).toMatchObject({ token: 'at_live_fresh', registered_name: 'WorkerA' }); }); @@ -74,12 +77,13 @@ describe('registerAgentWithRebind', () => { // Edge case: token is still set but the identity was evicted. The session // is in an inconsistent state, so a fresh registration is the safe path. const setSession = vi.fn(); - const registerOrRotate = vi.fn().mockResolvedValue({ + const register = vi.fn().mockResolvedValue({ id: 'agent_789', name: 'WorkerA', token: 'at_live_rotated', status: 'online', }); + const registerOrRotate = vi.fn(); await registerAgentWithRebind({ session: { @@ -91,14 +95,15 @@ describe('registerAgentWithRebind', () => { setSession, getRelay: () => ({ - agents: { registerOrRotate }, + agents: { register, registerOrRotate }, }) as never, name: 'WorkerA', strictAgentName: true, preferredAgentName: 'WorkerA', }); - expect(registerOrRotate).toHaveBeenCalledOnce(); + expect(register).toHaveBeenCalledOnce(); + expect(registerOrRotate).not.toHaveBeenCalled(); }); it('prefers the per-identity token from the agents map when available', async () => { @@ -126,14 +131,15 @@ describe('registerAgentWithRebind', () => { expect(payload).toMatchObject({ token: 'at_live_per_identity' }); }); - it('registers or rotates and updates the bound session token', async () => { + it('uses conflict-safe registration for a strict worker identity', async () => { const setSession = vi.fn(); - const registerOrRotate = vi.fn().mockResolvedValue({ + const register = vi.fn().mockResolvedValue({ id: 'agent_123', name: 'WorkerA', token: 'at_live_rotated', status: 'online', }); + const registerOrRotate = vi.fn(); const payload = await registerAgentWithRebind({ session: { @@ -145,6 +151,7 @@ describe('registerAgentWithRebind', () => { getRelay: () => ({ agents: { + register, registerOrRotate, }, }) as never, @@ -156,7 +163,7 @@ describe('registerAgentWithRebind', () => { preferredAgentName: 'WorkerA', }); - expect(registerOrRotate).toHaveBeenCalledWith({ + expect(register).toHaveBeenCalledWith({ name: 'WorkerA', type: 'agent', persona: 'Test worker', @@ -173,6 +180,34 @@ describe('registerAgentWithRebind', () => { registered_name: 'WorkerA', warnings: [], }); + expect(registerOrRotate).not.toHaveBeenCalled(); + }); + + it('does not rotate a strict identity when plain registration reports a name conflict', async () => { + const setSession = vi.fn(); + const conflict = Object.assign(new Error('agent name already exists'), { code: 'name_conflict' }); + const register = vi.fn().mockRejectedValue(conflict); + const registerOrRotate = vi.fn(); + + await expect( + registerAgentWithRebind({ + session: { + workspaceKey: 'rk_live_test', + agentToken: null, + agentName: 'WorkerA', + agents: new Map(), + }, + setSession, + getRelay: () => ({ agents: { register, registerOrRotate } }) as never, + name: 'WorkerA', + strictAgentName: true, + preferredAgentName: 'WorkerA', + }) + ).rejects.toBe(conflict); + + expect(register).toHaveBeenCalledOnce(); + expect(registerOrRotate).not.toHaveBeenCalled(); + expect(setSession).not.toHaveBeenCalled(); }); }); diff --git a/packages/cli/src/cli/agent-relay-mcp.ts b/packages/cli/src/cli/agent-relay-mcp.ts index 8e32e8140..836b54f82 100644 --- a/packages/cli/src/cli/agent-relay-mcp.ts +++ b/packages/cli/src/cli/agent-relay-mcp.ts @@ -343,7 +343,7 @@ export async function registerAgentWithRebind({ // If the session tracks per-identity agents, only short-circuit when the // strict-named identity is still registered. After an `agent_token_invalid` // recovery the entry is dropped from the map, which lets this fall through - // to a fresh registerOrRotate instead of handing back the dead token. + // to a conflict-safe registration instead of handing back the dead token. const cachedAgent = session.agents?.get(effectiveName); const knowsIdentities = session.agents !== undefined; if (!knowsIdentities || cachedAgent) { @@ -357,12 +357,19 @@ export async function registerAgentWithRebind({ } const relay = getRelay(); - const result = await relay.agents.registerOrRotate({ + const registration = { name: effectiveName, type: effectiveType, persona, metadata, - }); + }; + // A strict worker name is bound to one agent identity. Rotating that + // identity's token from another live MCP session silently disconnects the + // first session and creates a re-registration ping-pong. Plain registration + // is an atomic server-side conflict instead, so its token remains valid. + const result = strictAgentName + ? await relay.agents.register(registration) + : await relay.agents.registerOrRotate(registration); const reboundName = result.name?.trim() ? result.name : effectiveName; setSession({ agentToken: result.token, agentName: reboundName }); @@ -915,10 +922,15 @@ export async function resolveStdioBootstrapOptions( const relay = createWorkspaceClient({ workspaceKey, baseUrl: options.baseUrl }); - const registered = await relay.agents.registerOrRotate({ + const registration = { name: options.agentName, type: options.agentType, - }); + }; + // Match the tool path: a strict injected worker must never rotate the token + // held by another session with the same fixed identity during bootstrap. + const registered = options.strictAgentName + ? await relay.agents.register(registration) + : await relay.agents.registerOrRotate(registration); return { ...options, agentToken: registered.token, diff --git a/packages/sdk/src/messaging/thin-client.ts b/packages/sdk/src/messaging/thin-client.ts index 78d9d183e..00d2990b5 100644 --- a/packages/sdk/src/messaging/thin-client.ts +++ b/packages/sdk/src/messaging/thin-client.ts @@ -91,6 +91,8 @@ export interface RelayRawGroupConversation { export interface RelayWorkspaceThinClient { readonly agents: { list(options?: RelayListAgentsOptions): Promise; + /** Register a new agent and fail if that identity already exists. */ + register(input: RelayRegisterAgentInput): Promise; /** Register an agent, rotating its token when the name already exists. */ registerOrRotate(input: RelayRegisterAgentInput): Promise; /** Ask the relay to spawn a provider-backed worker. */