diff --git a/CHANGELOG.md b/CHANGELOG.md index 18a64fe9e..5daab2104 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -26,6 +26,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - `register_agent` now writes supplied `metadata` or `persona` instead of silently discarding it when returning a cached token, and its new `verify_metadata` option reports whether the write actually persisted. - Spawned agents now launch Agent Relay MCP through an installed local `agent-relay` executable instead of cold `npx` resolution. - Spawn now fails before start with an actionable error when no usable Agent Relay MCP executable is available. +- Fleet `spawn:` actions resolve from the spawn's own verified result instead of bare worker-registry presence, so a node that registers a worker whose process dies during startup now returns `spawn_failed: ` with the startup exit status and worker log path rather than `spawned: true`. - A fleet node whose connection to the engine goes dead now reconnects on its own instead of disappearing from `agent-relay fleet nodes` until the broker is restarted. ## [11.6.1] - 2026-08-13 diff --git a/crates/broker/src/runtime/fleet.rs b/crates/broker/src/runtime/fleet.rs index 785539c16..1e84bf869 100644 --- a/crates/broker/src/runtime/fleet.rs +++ b/crates/broker/src/runtime/fleet.rs @@ -1128,7 +1128,7 @@ impl BrokerRuntime { let action_control_dedup_key = relaycast_spawn_control_dedup_key(workspace_id.as_str(), name.as_str()); - super::relaycast_events::spawn_worker_from_request( + let spawn_result = super::relaycast_events::spawn_worker_from_request( name.clone(), cli, task, @@ -1161,73 +1161,86 @@ impl BrokerRuntime { let verify_ready = super::relaycast_events::relaycast_spawn_verifies_ready(&ws_value); - // A verified spawn keeps the action open until the harness itself emits - // worker_ready. Process creation alone is not proof that the persona is - // usable; worker_events resolves this pending entry, while maintenance - // fails it after an early exit/readiness timeout and performs cleanup. - if self.workers.workers.contains_key(&name) { - if verify_ready { - if self - .workers - .workers - .get(&name) - .is_some_and(|worker| worker.ready_at.is_some()) - { - self.send_fleet_action_result(verified_spawn_ready_result( - invoke.invocation_id, - &name, - )) - .await; - } else { - let generation = self - .workers - .workers - .get(&name) - .expect("verified spawn worker must still exist") - .generation; - self.pending_verified_spawns.insert( - name, - PendingVerifiedSpawn { - invocation_id: invoke.invocation_id, - deadline: Instant::now() + VERIFIED_SPAWN_READY_TIMEOUT, - generation, - }, - ); + let spawn_outcome = + fleet_spawn_outcome(spawn_result, &name, self.workers.is_worker_live(&name)); + + match spawn_outcome { + Ok(()) => { + // 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 verify_ready { + let (already_ready, generation) = { + let worker = self + .workers + .workers + .get(&name) + .expect("verified spawn worker must still exist"); + (worker.ready_at.is_some(), worker.generation) + }; + if already_ready { + 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, + generation, + }, + ); + } + return; } - return; + self.send_fleet_action_result(fleet_spawn_action_result( + &invoke.invocation_id, + &name, + Ok(()), + )) + .await; } - 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) + Err(error) => { + // 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; + { + Ok(_) => { + prune_fleet_agent_state( + &self.fleet_control_tx, + &mut self.fleet_inventory, + &mut self.fleet_delivery_book, + &name, + ) + .await + } + Err(cleanup_error) => { + tracing::warn!(worker = %name, error = %cleanup_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") + self.send_fleet_action_result(fleet_spawn_action_result( + &invoke.invocation_id, + &name, + Err(error), + )) .await; + } } } @@ -1369,6 +1382,53 @@ impl BrokerRuntime { } } +/// Decide a fleet spawn action from the spawn's own verified outcome. +/// +/// `spawn_worker_from_request` returns only after the process-stability probe, +/// so an `Err` already carries the real startup exit status and worker log +/// path. Liveness is still required on the success path: the child can exit +/// between that probe and this decision, and registry presence would survive +/// that — reporting `spawned: true` for a dead worker is precisely the failure +/// this action exists to prevent, so the guard asks whether the process is +/// alive rather than whether a map entry exists. +/// +/// Split out from `handle_fleet_action_spawn` so the guard is testable without +/// a whole `BrokerRuntime`; an untestable guard is how the weaker +/// registry-presence check survived here in the first place. +fn fleet_spawn_outcome( + spawn_result: Result<()>, + name: &WorkerName, + worker_is_live: bool, +) -> Result<()> { + match spawn_result { + Ok(()) if !worker_is_live => Err(anyhow::anyhow!( + "agent '{name}' has no live worker process after spawn" + )), + other => other, + } +} + +fn fleet_spawn_action_result( + invocation_id: &str, + name: &WorkerName, + spawn_result: Result<()>, +) -> ActionResult { + let result = match spawn_result { + Ok(()) => ActionResultPayload::Output(ActionResultOutput { + output: json!({ "spawned": true, "name": name.as_str() }), + }), + Err(error) => ActionResultPayload::Error(ActionResultError { + error: format!("spawn_failed: {error}"), + }), + }; + ActionResult { + v: FLEET_WIRE_VERSION, + id: None, + invocation_id: invocation_id.to_string(), + result, + } +} + #[derive(Debug, Default, PartialEq, Eq)] pub(super) struct FlushPendingRelayResult { pub(super) flushed: usize, @@ -2056,6 +2116,98 @@ mod tests { use super::*; use crate::protocol::PtyHarnessConfig; + #[cfg(unix)] + #[tokio::test] + async fn fleet_spawn_result_uses_verified_failure_not_registry_presence() { + let temp = tempfile::tempdir().expect("test tempdir"); + let (event_tx, _event_rx) = mpsc::channel::(4); + let mut workers = WorkerRegistry::new( + event_tx, + Vec::new(), + temp.path().join("worker-logs"), + Instant::now(), + ); + let mut child = tokio::process::Command::new("sh") + .args(["-c", "exit 19"]) + .stdin(Stdio::piped()) + .stdout(Stdio::null()) + .stderr(Stdio::null()) + .spawn() + .expect("repro child should spawn"); + let _stdin = child.stdin.take().expect("repro child stdin"); + child.wait().await.expect("repro child should exit"); + let name = WorkerName::from("fleet-spawn-repro-1430"); + let mut spec = test_agent_spec(None, None); + spec.name = name.clone(); + let (command_tx, _command_rx) = mpsc::channel(4); + workers.workers.insert( + name.clone(), + WorkerHandle { + generation: Uuid::new_v4(), + spec, + parent: Some("Relaycast".to_string()), + workspace_id: None, + child, + command_tx, + harness_pid: None, + spawned_at: Instant::now(), + ready_at: None, + last_activity_at: Instant::now(), + context_budget_pct: None, + state: crate::worker::AgentWorkState::Working, + exit_reason: None, + }, + ); + + // The registered-but-dead worker: present in the map, process gone. + // These two lines are the whole point — a guard keyed on presence sees + // a healthy worker here, and a guard keyed on liveness does not. + assert!(workers.workers.contains_key(&name)); + assert!(!workers.is_worker_live(&name)); + + // MUST-FIRE: a successful `spawn_worker_from_request` is not enough if + // the process died between its stability probe and this decision. + let outcome = fleet_spawn_outcome(Ok(()), &name, workers.is_worker_live(&name)); + let error = + outcome.expect_err("a dead worker must not resolve the spawn action as success"); + assert!( + error.to_string().contains("no live worker process"), + "{error}" + ); + + // MUST-NOT-FIRE: a live worker with a successful spawn stays successful, + // so the guard above is not simply rejecting everything. + fleet_spawn_outcome(Ok(()), &name, true).expect("a live worker must resolve as success"); + + // A real launch failure keeps its detail rather than being replaced by + // the liveness message. + let propagated = fleet_spawn_outcome(Err(anyhow::anyhow!("exit status: 19")), &name, false) + .expect_err("a failed spawn must stay failed"); + assert!( + propagated.to_string().contains("exit status: 19"), + "{propagated}" + ); + + let result = fleet_spawn_action_result( + "inv-failed-1430", + &name, + Err(anyhow::anyhow!( + "agent '{name}' process exited during startup (exit status: 19); see worker log /tmp/{name}.log" + )), + ); + + let ActionResultPayload::Error(error) = result.result else { + panic!("a verified spawn failure must not produce spawned:true"); + }; + assert_eq!(result.invocation_id, "inv-failed-1430"); + assert_eq!( + error.error, + format!( + "spawn_failed: agent '{name}' process exited during startup (exit status: 19); see worker log /tmp/{name}.log" + ) + ); + } + fn test_agent_spec(session_id: Option<&str>, harness_session_id: Option<&str>) -> AgentSpec { AgentSpec { name: WorkerName::from("agent-a"), diff --git a/crates/broker/src/runtime/relaycast_events.rs b/crates/broker/src/runtime/relaycast_events.rs index 9aa9c40d8..66bf6f7f4 100644 --- a/crates/broker/src/runtime/relaycast_events.rs +++ b/crates/broker/src/runtime/relaycast_events.rs @@ -333,6 +333,8 @@ pub(super) async fn release_worker_locally( /// engine-dispatched spawn exits after its task identically to a local HTTP /// spawn. `control_dedup_key` carries the firehose control dedup key so the /// local spawn-echo dedup behaves identically. +/// Returns only after `WorkerRegistry::spawn` has completed its process +/// stability probe, preserving the detailed launch error for the action result. #[allow(clippy::too_many_arguments)] pub(super) async fn spawn_worker_from_request( name: WorkerName, @@ -360,7 +362,7 @@ pub(super) async fn spawn_worker_from_request( session_ref: Option, hosted_agent_event_tx: &mpsc::Sender, pty_observability: &mut HashMap, -) { +) -> Result<()> { let workspace_http = &workspace_state.http_client; eprintln!( "[agent-relay] received spawn request for '{}' (cli: {})", @@ -379,7 +381,7 @@ pub(super) async fn spawn_worker_from_request( "[agent-relay] ignoring spawn request for '{}' (broker self)", name ); - return; + anyhow::bail!("agent '{name}' is the broker self"); } let local_spawn_echo_key = relaycast_spawn_control_dedup_key(workspace_id, &name); if relaycast_ws_should_apply_local_spawn_echo_dedup(control_dedup_key, &local_spawn_echo_key) @@ -394,7 +396,7 @@ pub(super) async fn spawn_worker_from_request( "[agent-relay] dropping duplicate spawn request for '{}'", name ); - return; + anyhow::bail!("duplicate spawn request for agent '{name}'"); } let task = task.filter(|value| !value.trim().is_empty()); // Carry the requested model through so the launched CLI is @@ -413,7 +415,7 @@ pub(super) async fn spawn_worker_from_request( "[agent-relay] rejecting spawn request for '{}': {}", name, error ); - return; + return Err(anyhow::anyhow!(error)); } }; let require_node_registration = harness_config.as_ref().is_some_and(|config| { @@ -549,7 +551,9 @@ pub(super) async fn spawn_worker_from_request( error = %node_error, "rejecting verified spawn because node agent.register failed" ); - return; + anyhow::bail!( + "node agent.register failed for agent '{name}': {node_error}" + ); } tracing::warn!( worker = %name, @@ -761,6 +765,7 @@ pub(super) async fn spawn_worker_from_request( .await; tracing::info!(child = %name, pid = ?pid, "spawned worker via relaycast WS"); eprintln!("[agent-relay] spawned worker '{}' via relaycast", name); + Ok(()) } Err(e) => { let msg = e.to_string(); @@ -770,6 +775,7 @@ pub(super) async fn spawn_worker_from_request( tracing::error!(child = %name, error = %e, "failed to spawn worker via relaycast WS"); eprintln!("[agent-relay] failed to spawn '{}': {}", name, e); } + Err(e) } } } @@ -779,6 +785,120 @@ mod tests { use super::*; use ::relaycast::WsEvent; + #[cfg(unix)] + #[tokio::test] + async fn spawn_request_returns_the_verified_process_failure() { + let temp = tempfile::tempdir().expect("test tempdir"); + let (worker_event_tx, _worker_event_rx) = mpsc::channel::(4); + let mut workers = WorkerRegistry::new( + worker_event_tx, + Vec::new(), + temp.path().join("worker-logs"), + Instant::now(), + ); + let workspace_id = WorkspaceId::from("ws_test_1430".to_string()); + let (ws_control_tx, _ws_control_rx) = mpsc::channel::(4); + let workspace = RelayWorkspace { + workspace_id: workspace_id.clone(), + workspace_alias: None, + relay_workspace_key: "rk_live_test".to_string(), + self_name: "broker".to_string(), + self_agent_id: AgentId::from("agent_broker".to_string()), + self_names: HashSet::from(["broker".to_string()]), + self_agent_ids: HashSet::from([AgentId::from("agent_broker".to_string())]), + http_client: RelaycastHttpClient::new( + Some("http://127.0.0.1:9".to_string()), + "rk_live_test", + "broker", + "codex", + ), + ws_control_tx, + }; + let paths = ensure_ephemeral_paths(temp.path(), "fleet-spawn-1430") + .expect("ephemeral runtime paths"); + let mut state = broker::BrokerState::default(); + let telemetry = TelemetryClient::default(); + let (sdk_out_tx, _sdk_out_rx) = mpsc::channel(4); + let mut dedup = DedupCache::new(Duration::from_secs(60), 16); + let mut agent_spawn_count = 0; + let (fleet_control_tx, _fleet_control_rx) = mpsc::channel(4); + let mut fleet_delivery_book = FleetDeliveryBook::default(); + let mut fleet_inventory = HashMap::new(); + let (hosted_agent_event_tx, _hosted_agent_event_rx) = mpsc::channel(4); + let mut pty_observability = HashMap::new(); + let name = WorkerName::from("failed-native-worker-1430"); + let ws_value = json!({ + "token": "at_live_test_worker", + "agent": { + "harnessConfig": { + "runtime": "native", + "command": "sh", + // Exit immediately rather than after a fixed sleep: the + // assertion only needs the child to exit somewhere inside + // WORKER_SPAWN_STABILITY_WINDOW, and a fixed 50ms sleep left + // only ~200ms of margin against that 250ms window on a + // loaded shared CI runner (relay#1516). An immediate exit + // keeps the full window as margin without touching the + // production constant. + "args": ["-c", "exit 23"], + "sessionId": "native-failed-1430" + } + } + }); + let control_key = relaycast_spawn_control_dedup_key(&workspace_id, &name); + + let error = spawn_worker_from_request( + name.clone(), + "codex".to_string(), + None, + None, + None, + false, + &ws_value, + &workspace_id, + Some(&control_key), + &workspace, + &mut workers, + &mut state, + &paths, + &telemetry, + &sdk_out_tx, + &mut dedup, + &mut agent_spawn_count, + &fleet_control_tx, + &mut fleet_delivery_book, + &mut fleet_inventory, + "test-node", + Some("inv-failed-1430".to_string()), + None, + &hosted_agent_event_tx, + &mut pty_observability, + ) + .await + .expect_err("a sidecar that exits during the stability window must fail the spawn"); + + // Two rejection paths race here: the stability-window check + // ("process exited during startup") if the child is still alive when + // `send_to_worker("init_worker")` writes to it, or an EPIPE from that + // write ("failed writing frame to worker") if the child has already + // exited by then (see the comment on that error branch above, and the + // identical pattern in tests/integration/broker/cli-spawn.test.ts). + // Both are the correct rejection for a sidecar that dies on startup, + // so assert on whichever wins rather than pinning to one. Exit status + // and log-path detail are asserted deterministically at the requester + // level in fleet-spawn-confirmation.test.ts, which uses a fixture + // instead of a real process and cannot race. + let message = error.to_string(); + assert!( + message.contains("process exited during startup") + || message.contains("failed writing frame to worker"), + "{message}" + ); + assert!(!workers.has_worker(&name)); + assert_eq!(agent_spawn_count, 0); + assert!(!state.agents.contains_key(&name)); + } + #[test] fn relaycast_harness_config_accepts_inline_config() { let value = json!({ diff --git a/packages/cli/src/cli/commands/fleet.test.ts b/packages/cli/src/cli/commands/fleet.test.ts index d3824a7b8..b41b581a7 100644 --- a/packages/cli/src/cli/commands/fleet.test.ts +++ b/packages/cli/src/cli/commands/fleet.test.ts @@ -451,6 +451,10 @@ describe('fleet command support', () => { capability: 'spawn:codex', node: 'sf-mini', failFast: true, + // #1430: acceptance by the node is not evidence of a launch, so a + // targeted spawn asks the node to confirm unless told not to. + confirm: true, + confirmTimeoutMs: 120_000, input: { name: 'api-worker', cli: 'codex', @@ -500,6 +504,111 @@ describe('fleet command support', () => { }); }); + it('fleet spawn --no-confirm accepts an unconfirmed targeted dispatch', async () => { + const placement = { + spawn: vi.fn(async () => ({ + invocationId: 'inv_unconfirmed', + actionName: 'spawn', + node: { name: 'sf-mini' }, + placement: { + capability: 'spawn:codex', + node: 'sf-mini', + attempts: 1, + queued: false, + confirmed: false, + }, + })), + }; + const createAgentRelay = vi.fn(() => ({ messaging: { placement } })); + const program = new Command(); + program.exitOverride(); + registerFleetCommands(program, { + sdk: { + createAgentRelay: createAgentRelay as never, + createWorkspaceRelay: vi.fn() as never, + createWorkspace: vi.fn() as never, + log: () => undefined, + error: vi.fn(), + exit: vi.fn() as never, + }, + createFleetWorkspaceClient: vi.fn() as never, + log: () => undefined, + warn: () => undefined, + error: () => undefined, + }); + + await program.parseAsync( + [ + 'fleet', + 'spawn', + 'codex', + '--name', + 'api-worker', + '--task', + 'ACK and wait', + '--node', + 'sf-mini', + '--no-confirm', + '--workspace-key', + 'rk_live_test', + '--token', + 'at_live_lead', + ], + { from: 'user' } + ); + + // The escape hatch must actually disable confirmation, and must not smuggle + // a timeout through that would imply the caller is still waiting. + const call = placement.spawn.mock.calls[0]![0] as Record; + expect(call.confirm).toBe(false); + expect(call).not.toHaveProperty('confirmTimeoutMs'); + }); + + it('fleet spawn rejects a non-numeric --confirm-timeout', async () => { + const placement = { spawn: vi.fn() }; + const program = new Command(); + program.exitOverride(); + const errors: unknown[] = []; + registerFleetCommands(program, { + sdk: { + createAgentRelay: vi.fn(() => ({ messaging: { placement } })) as never, + createWorkspaceRelay: vi.fn() as never, + createWorkspace: vi.fn() as never, + log: () => undefined, + error: (message: unknown) => errors.push(message), + exit: vi.fn() as never, + }, + createFleetWorkspaceClient: vi.fn() as never, + log: () => undefined, + warn: () => undefined, + error: () => undefined, + }); + + await program.parseAsync( + [ + 'fleet', + 'spawn', + 'codex', + '--name', + 'api-worker', + '--task', + 'ACK and wait', + '--node', + 'sf-mini', + '--confirm-timeout', + 'soon', + '--workspace-key', + 'rk_live_test', + '--token', + 'at_live_lead', + ], + { from: 'user' } + ); + + expect(placement.spawn).not.toHaveBeenCalled(); + expect(String(errors.join('\n'))).toContain('--confirm-timeout'); + }); + it('fleet spawn uses workspace-scoped automatic placement when no node is named', async () => { const spawn = vi.fn(async () => ({ invocation_id: 'inv_auto', status: 'accepted' })); const createFleetWorkspaceClient = vi.fn(() => ({ agents: { spawn } })); diff --git a/packages/cli/src/cli/commands/fleet.ts b/packages/cli/src/cli/commands/fleet.ts index 52f098f2a..63a262b63 100644 --- a/packages/cli/src/cli/commands/fleet.ts +++ b/packages/cli/src/cli/commands/fleet.ts @@ -126,6 +126,15 @@ export function registerFleetCommands( .option('--role ', 'Declared role for workforce reporting') .option('--objective ', 'Declared objective (defaults to --task when omitted)') .option('--session-ref ', 'Session reference for a resumable targeted spawn') + .option( + '--no-confirm', + 'Report a targeted spawn as soon as the node accepts it, without waiting for the node to confirm the agent actually launched' + ) + .option( + '--confirm-timeout ', + 'How long a targeted spawn waits for the node to confirm the launch', + '120000' + ) ).action(async (cli: string, options: Record) => { await runSdk(deps.sdk, async () => { warnIfInferredFromProjectSession(options, deps.warn); @@ -146,6 +155,11 @@ export function registerFleetCommands( { organization, project, workstream, role, objective }, task ); + const confirmTimeoutText = optionalText(options.confirmTimeout, 'Confirm timeout') ?? '120000'; + const confirmTimeoutMs = Number(confirmTimeoutText); + if (!Number.isFinite(confirmTimeoutMs) || confirmTimeoutMs <= 0) { + throw new Error('--confirm-timeout must be a positive number of milliseconds.'); + } if (targetNode) { if (!resolveAgentToken(clientOptions)) { @@ -154,10 +168,17 @@ export function registerFleetCommands( ); } const relay = deps.sdk.createAgentRelay(clientOptions); + // Placement alone only proves the node accepted the dispatch. A node + // running an obsolete broker advertises `spawn:` capacity, acks + // the invocation and launches nothing, which is indistinguishable from + // success here — so wait for the node to confirm unless asked not to. + const confirm = options.confirm !== false; const invocation = await relay.messaging.placement.spawn({ capability: `spawn:${cli}`, node: targetNode, failFast: true, + confirm, + ...(confirm ? { confirmTimeoutMs: confirmTimeoutMs } : {}), input: { name, cli, diff --git a/packages/cli/src/cli/lib/fleet-spawn-confirmation.test.ts b/packages/cli/src/cli/lib/fleet-spawn-confirmation.test.ts new file mode 100644 index 000000000..68d30b3b2 --- /dev/null +++ b/packages/cli/src/cli/lib/fleet-spawn-confirmation.test.ts @@ -0,0 +1,262 @@ +import { RelayPlacementError, RelaycastMessagingClient } from '@agent-relay/sdk'; +import { describe, expect, it, vi } from 'vitest'; + +/** + * Gated proof for issue #1430. + * + * These arms live here, under `packages/cli`, rather than beside the other + * placement tests in `packages/sdk/src/messaging/placement.test.mts`, because + * the root vitest config — the only JS suite CI runs — excludes + * `packages/sdk/**`. The load-bearing arms have to sit in a suite that actually + * runs, or a later refactor reverts the fix with CI green, which is the very + * failure shape this change exists to prevent. + * + * Every arm is a deterministic requester-level fixture. None of them touches a + * real node, and none of them passes merely because a node replied correctly — + * the point under test is that the REQUESTER distinguishes "the node executed + * and confirmed" from "the engine accepted the dispatch". + */ + +const LIVE_NODE = { + id: 'node_a', + name: 'node-a', + status: 'online', + live: true, + capabilities: [{ name: 'spawn:claude', kind: 'spawn' }], + repo_keys: ['relay'], +}; + +function createClient(getInvocation?: (name: string, invocationId: string) => Promise) { + const invoke = vi.fn(async (name: string, input?: Record) => ({ + invocation_id: 'inv-1430', + action_name: name, + handler_node_id: 'node_a', + dispatched_node_id: 'node_a', + input, + // The engine accepted the dispatch. This is all the requester ever knew + // before this change, and it is identical whether or not anything launched. + status: 'invoked', + })); + const reader = vi.fn(getInvocation ?? (async () => undefined)); + const relaycast = { + agents: { + list: vi.fn(async () => []), + get: vi.fn(), + register: vi.fn(), + update: vi.fn(), + delete: vi.fn(), + presence: vi.fn(async () => []), + }, + channels: { list: vi.fn(async () => []), get: vi.fn() }, + messages: { list: vi.fn(async () => []), get: vi.fn(), thread: vi.fn(), reactions: vi.fn() }, + nodes: { + list: vi.fn(async () => [LIVE_NODE]), + get: vi.fn(async () => LIVE_NODE), + }, + }; + const agentClient = { + actions: { invoke, getInvocation: reader, completeInvocation: vi.fn() }, + }; + const client = new RelaycastMessagingClient({ + relaycast: relaycast as never, + agentClient: agentClient as never, + placementTtlMs: 60, + }); + return { client, invoke, reader }; +} + +function spawnInput(overrides: Record = {}) { + return { + capability: 'spawn:claude', + node: 'node-a', + repo: 'relay', + input: { name: 'worker-1430' }, + ...overrides, + }; +} + +describe('fleet spawn confirmation is observable from the requester (#1430)', () => { + // MUST-FIRE — the sf-mini shape. A node advertised `spawn:claude` capacity, + // accepted the invocation, and launched nothing; the invocation therefore + // never reaches a terminal state. Before this change the requester returned a + // successful placement ack here. Silence must now be a failure. + it('fails with spawn_unconfirmed when the node accepts but never reports a result', async () => { + const { client, reader } = createClient(async (name, invocationId) => ({ + invocation_id: invocationId, + action_name: name, + status: 'invoked', + })); + + const error = await client.placement + .spawn(spawnInput({ confirm: true, confirmTimeoutMs: 60, confirmPollIntervalMs: 10 })) + .catch((caught: unknown) => caught); + + expect(error).toBeInstanceOf(RelayPlacementError); + expect((error as RelayPlacementError).code).toBe('spawn_unconfirmed'); + expect((error as RelayPlacementError).node).toBe('node-a'); + expect((error as Error).message).toContain('never reported a result'); + expect(reader).toHaveBeenCalled(); + }); + + // MUST-FIRE — a node that reports its failure honestly still surfaced as + // success before this change, because nothing read the action result. The + // broker's detail (startup exit status and worker log path) must survive. + it('fails with spawn_failed and preserves the node-reported detail', async () => { + const { client } = createClient(async (name, invocationId) => ({ + invocation_id: invocationId, + action_name: name, + status: 'failed', + error: + "spawn_failed: agent 'worker-1430' process exited during startup (exit status: 19); see worker log /tmp/worker-1430.log", + })); + + const error = await client.placement + .spawn(spawnInput({ confirm: true, confirmTimeoutMs: 1_000, confirmPollIntervalMs: 10 })) + .catch((caught: unknown) => caught); + + expect(error).toBeInstanceOf(RelayPlacementError); + expect((error as RelayPlacementError).code).toBe('spawn_failed'); + expect((error as Error).message).toContain('exit status: 19'); + expect((error as Error).message).toContain('/tmp/worker-1430.log'); + }); + + // MUST-NOT-FIRE — a healthy node. This is the arm a repaired node represents: + // confirmation must not turn a working spawn into an error. + it('resolves when the node confirms the spawn completed', async () => { + const { client } = createClient(async (name, invocationId) => ({ + invocation_id: invocationId, + action_name: name, + status: 'completed', + output: { spawned: true, name: 'worker-1430' }, + })); + + const ack = await client.placement.spawn( + spawnInput({ confirm: true, confirmTimeoutMs: 1_000, confirmPollIntervalMs: 10 }) + ); + + expect(ack.placement.confirmed).toBe(true); + expect(ack.confirmation?.status).toBe('completed'); + }); + + // VACUITY CONTROL — without `confirm` the invocation is never read back, so + // the three arms above cannot be passing for some incidental reason. This is + // also the documented behaviour of the generic primitive: `placement.spawn` + // dispatches non-spawn capabilities too, so it does not wait by default. + it('does not read the invocation at all when confirmation is not requested', async () => { + const { client, reader } = createClient(); + + const ack = await client.placement.spawn(spawnInput()); + + expect(reader).not.toHaveBeenCalled(); + expect(ack.placement.confirmed).toBe(false); + expect(ack.confirmation).toBeUndefined(); + }); + + // An engine that cannot answer at all is not evidence of success either. + it('fails with spawn_unconfirmed when the invocation cannot be read', async () => { + const { client } = createClient(async () => { + throw new Error('getInvocation is not supported by this engine'); + }); + + const error = await client.placement + .spawn(spawnInput({ confirm: true, confirmTimeoutMs: 60, confirmPollIntervalMs: 10 })) + .catch((caught: unknown) => caught); + + expect(error).toBeInstanceOf(RelayPlacementError); + expect((error as RelayPlacementError).code).toBe('spawn_unconfirmed'); + expect((error as Error).message).toContain('getInvocation is not supported'); + }); + + // `denied` is terminal — the documented lifecycle is + // `invoked -> completed | failed | denied`. Treating it as pending would burn + // the whole timeout and then report the wrong code, losing the node's reason. + it('treats a denied invocation as a terminal failure, not as pending', async () => { + const { client } = createClient(async (name, invocationId) => ({ + invocation_id: invocationId, + action_name: name, + status: 'denied', + error: 'spawn:claude refused: node is draining', + })); + + const started = Date.now(); + const error = await client.placement + .spawn(spawnInput({ confirm: true, confirmTimeoutMs: 30_000, confirmPollIntervalMs: 10 })) + .catch((caught: unknown) => caught); + + expect(error).toBeInstanceOf(RelayPlacementError); + expect((error as RelayPlacementError).code).toBe('spawn_failed'); + expect((error as Error).message).toContain('node is draining'); + // Must resolve on the denial, not after the 30s budget. + expect(Date.now() - started).toBeLessThan(5_000); + }); + + // A non-finite or non-positive timeout must not degenerate the loop. Without + // normalization `Date.now() + NaN` is `NaN`, `Date.now() >= NaN` is never + // true, and the poll delay collapses to `NaN` — which `setTimeout` treats as + // 0. The loop then spins as fast as it can, forever: a silent hang inside the + // mechanism whose entire purpose is to stop silent waiting. + // + // The observable asserted here is the CADENCE, because that is what separates + // the two states quickly. Waiting for the normalized fallback budget to + // expire would mean a 120s test, so this deliberately does not do that; it + // checks that the deadline arithmetic stayed finite enough to keep pacing the + // reads, then lets the invocation go terminal so nothing is left pending. + // + // Measured honestly: `NaN`, `0` and `-1` each fail without the normalization. + // `Infinity` does NOT fail this arm — an infinite deadline still paces reads + // off the old 25ms floor rather than busy-spinning — so it is covered here + // only because it shares the clamp path. Its real effect (a finite budget + // instead of an unbounded one) is not separately observable inside a fast + // test, and this arm does not claim otherwise. + it.each([NaN, Infinity, 0, -1])( + 'does not degenerate into a busy-spin when confirmTimeoutMs is %p', + async (badTimeout) => { + let reads = 0; + let terminal = false; + const { client } = createClient(async (name, invocationId) => { + reads += 1; + return { + invocation_id: invocationId, + action_name: name, + status: terminal ? 'failed' : 'invoked', + ...(terminal ? { error: 'worker died' } : {}), + }; + }); + + const pending = client.placement + .spawn(spawnInput({ confirm: true, confirmTimeoutMs: badTimeout, confirmPollIntervalMs: 20 })) + .catch((caught: unknown) => caught); + + await new Promise((resolve) => setTimeout(resolve, 200)); + // ~10 reads at the requested 20ms cadence. An unnormalized deadline + // produces hundreds to thousands in the same window. + expect(reads).toBeGreaterThan(0); + expect(reads).toBeLessThan(40); + + terminal = true; + const error = await pending; + expect(error).toBeInstanceOf(RelayPlacementError); + expect((error as RelayPlacementError).code).toBe('spawn_failed'); + } + ); + + // A transient read failure followed by successful reads must not leave a + // stale "Last read error" implying reads are still failing. + it('does not report a stale read error after a later read succeeds', async () => { + let call = 0; + const { client } = createClient(async (name, invocationId) => { + call += 1; + if (call === 1) throw new Error('transient socket reset'); + return { invocation_id: invocationId, action_name: name, status: 'invoked' }; + }); + + const error = await client.placement + .spawn(spawnInput({ confirm: true, confirmTimeoutMs: 120, confirmPollIntervalMs: 10 })) + .catch((caught: unknown) => caught); + + expect(error).toBeInstanceOf(RelayPlacementError); + expect((error as RelayPlacementError).code).toBe('spawn_unconfirmed'); + expect(call).toBeGreaterThan(1); + expect((error as Error).message).not.toContain('transient socket reset'); + }); +}); diff --git a/packages/sdk/src/messaging/placement.test.mts b/packages/sdk/src/messaging/placement.test.mts index f4553204f..a3a722998 100644 --- a/packages/sdk/src/messaging/placement.test.mts +++ b/packages/sdk/src/messaging/placement.test.mts @@ -13,10 +13,14 @@ type RawNode = { function createClient( nodes: RawNode[], - options: { + { + getInvocation: getInvocationOverride, + ...options + }: { placementLog?: (message: string) => void; selfNodeName?: string; maxQueuedPlacements?: number; + getInvocation?: (name: string, invocationId: string) => Promise; } = {} ) { const invoke = vi.fn(async (name: string, input?: Record) => ({ @@ -50,10 +54,11 @@ function createClient( get: vi.fn(async (name: string) => nodes.find((node) => node.name === name) ?? null), }, }; + const getInvocation = vi.fn(getInvocationOverride ?? (async () => undefined)); const agentClient = { actions: { invoke, - getInvocation: vi.fn(), + getInvocation, completeInvocation: vi.fn(), }, }; @@ -63,9 +68,18 @@ function createClient( placementTtlMs: 60, ...options, }); - return { client, invoke, nodes }; + return { client, invoke, getInvocation, nodes }; } +const LIVE_NODE_A = { + id: 'node_a', + name: 'node-a', + status: 'online', + live: true, + capabilities: [{ name: 'spawn:claude', kind: 'spawn' }], + repo_keys: ['relay'], +}; + describe('RelaycastMessagingClient placement', () => { it('places a targeted spawn on the named live eligible node', async () => { const { client, invoke } = createClient([ @@ -520,4 +534,134 @@ describe('RelaycastMessagingClient placement', () => { expect(invoke).not.toHaveBeenCalled(); expect(logs.join('\n')).toContain('placement TTL expired'); }); + + // Issue #1430: a node running an obsolete broker advertises `spawn:` + // capacity, accepts the invocation, and launches nothing. Placement acceptance + // is therefore not evidence of a spawn, and the requester cannot assume the + // node is current enough to report its own failure. + describe('spawn confirmation (#1430)', () => { + it('does not confirm by default, so acceptance alone still resolves', async () => { + const { client, getInvocation } = createClient([LIVE_NODE_A]); + + const ack = await client.placement.spawn({ + capability: 'spawn:claude', + node: 'node-a', + repo: 'relay', + input: { name: 'worker-unconfirmed' }, + }); + + // Control arm for the tests below: without `confirm` the invocation is + // never read back, and the ack says so rather than implying a launch. + expect(getInvocation).not.toHaveBeenCalled(); + expect(ack.placement.confirmed).toBe(false); + expect(ack.confirmation).toBeUndefined(); + }); + + it('resolves when the node confirms the spawn completed', async () => { + const { client, getInvocation } = createClient([LIVE_NODE_A], { + getInvocation: async (name, invocationId) => ({ + invocation_id: invocationId, + action_name: name, + status: 'completed', + output: { spawned: true, name: 'worker-confirmed' }, + }), + }); + + const ack = await client.placement.spawn({ + capability: 'spawn:claude', + node: 'node-a', + repo: 'relay', + confirm: true, + input: { name: 'worker-confirmed' }, + }); + + expect(getInvocation).toHaveBeenCalled(); + expect(ack.placement.confirmed).toBe(true); + expect(ack.confirmation?.status).toBe('completed'); + }); + + // The sf-mini reproduction: capacity advertised, invocation accepted, no + // process, and no result ever reported. This must fail, not succeed. + it('fails with spawn_unconfirmed when the node accepts but never reports a result', async () => { + const { client } = createClient([LIVE_NODE_A], { + // An obsolete broker leaves the invocation non-terminal forever. + getInvocation: async (name, invocationId) => ({ + invocation_id: invocationId, + action_name: name, + status: 'invoked', + }), + }); + + const error = await client.placement + .spawn({ + capability: 'spawn:claude', + node: 'node-a', + repo: 'relay', + confirm: true, + confirmTimeoutMs: 60, + confirmPollIntervalMs: 10, + input: { name: 'worker-silent' }, + }) + .catch((caught: unknown) => caught); + + expect(error).toBeInstanceOf(RelayPlacementError); + expect((error as RelayPlacementError).code).toBe('spawn_unconfirmed'); + expect((error as RelayPlacementError).node).toBe('node-a'); + expect((error as Error).message).toContain('never reported a result'); + }); + + it('surfaces the node-reported failure detail as spawn_failed', async () => { + const { client } = createClient([LIVE_NODE_A], { + getInvocation: async (name, invocationId) => ({ + invocation_id: invocationId, + action_name: name, + status: 'failed', + error: + "spawn_failed: agent 'worker-dead' process exited during startup (exit status: 19); see worker log /tmp/worker-dead.log", + }), + }); + + const error = await client.placement + .spawn({ + capability: 'spawn:claude', + node: 'node-a', + repo: 'relay', + confirm: true, + confirmTimeoutMs: 1_000, + confirmPollIntervalMs: 10, + input: { name: 'worker-dead' }, + }) + .catch((caught: unknown) => caught); + + expect(error).toBeInstanceOf(RelayPlacementError); + expect((error as RelayPlacementError).code).toBe('spawn_failed'); + // The broker's detail (exit status + log path) must survive to the caller. + expect((error as Error).message).toContain('exit status: 19'); + expect((error as Error).message).toContain('/tmp/worker-dead.log'); + }); + + it('times out as spawn_unconfirmed when the invocation cannot be read at all', async () => { + const { client } = createClient([LIVE_NODE_A], { + getInvocation: async () => { + throw new Error('getInvocation is not supported by this engine'); + }, + }); + + const error = await client.placement + .spawn({ + capability: 'spawn:claude', + node: 'node-a', + repo: 'relay', + confirm: true, + confirmTimeoutMs: 60, + confirmPollIntervalMs: 10, + input: { name: 'worker-unreadable' }, + }) + .catch((caught: unknown) => caught); + + expect(error).toBeInstanceOf(RelayPlacementError); + expect((error as RelayPlacementError).code).toBe('spawn_unconfirmed'); + expect((error as Error).message).toContain('getInvocation is not supported'); + }); + }); }); diff --git a/packages/sdk/src/messaging/relaycast-placement.ts b/packages/sdk/src/messaging/relaycast-placement.ts index 32a605761..80af880e0 100644 --- a/packages/sdk/src/messaging/relaycast-placement.ts +++ b/packages/sdk/src/messaging/relaycast-placement.ts @@ -29,7 +29,20 @@ export type PlacementSelection = }; export class RelayPlacementError extends Error { - readonly code: 'capability_mismatch' | 'placement_queue_full' | 'placement_ttl_expired' | 'unmapped_repo'; + readonly code: + | 'capability_mismatch' + | 'placement_queue_full' + | 'placement_ttl_expired' + | 'unmapped_repo' + /** The node ran the action and reported a failure. */ + | 'spawn_failed' + /** + * The node accepted the invocation but never reported a terminal result. + * A node running an obsolete broker advertises `spawn:` capacity + * and acknowledges the dispatch without launching anything, which is + * otherwise indistinguishable from success at the requester. + */ + | 'spawn_unconfirmed'; readonly capability: string; readonly node?: string; readonly repo?: string; diff --git a/packages/sdk/src/messaging/relaycast.ts b/packages/sdk/src/messaging/relaycast.ts index 8354f41ec..edc960bef 100644 --- a/packages/sdk/src/messaging/relaycast.ts +++ b/packages/sdk/src/messaging/relaycast.ts @@ -122,6 +122,62 @@ import type { export { RelayPlacementError } from './relaycast-placement.js'; export type { RelaycastMessagingOptions } from './relaycast-client.js'; +const DEFAULT_CONFIRM_TIMEOUT_MS = 120_000; +const DEFAULT_CONFIRM_POLL_MS = 500; +/** `setTimeout` clamps anything larger, firing immediately instead of waiting. */ +const MAX_CONFIRM_TIMEOUT_MS = 2_147_483_647; + +/** Terminal statuses that mean the node ran the action successfully. */ +const CONFIRM_SUCCESS_STATUSES = new Set(['completed', 'succeeded', 'success']); +/** + * Terminal statuses that mean the node will not run the action. `denied` is a + * refusal (permission/authorization) and is terminal, not pending — see the + * documented lifecycle `invoked` → `completed` | `failed` | `denied` at + * `packages/sdk-swift/Sources/AgentRelaySDK/RelayRestClient.swift:26`, which + * that client surfaces as a non-retryable `action_denied` error. + */ +const CONFIRM_FAILURE_STATUSES = new Set(['failed', 'error', 'denied', 'cancelled', 'canceled']); + +/** Distinguishes "the read outlived its budget" from any value a read returns. */ +const READ_TIMED_OUT = Symbol('relay.confirm.readTimedOut'); + +type ConfirmReadOutcome = { ok: true; value: RelayActionInvocation } | { ok: false; error: string }; + +/** + * Coerce a caller-supplied duration to a usable one. + * + * `NaN` and `Infinity` are the dangerous inputs: `Date.now() + NaN` is `NaN`, + * and `Date.now() >= NaN` is never true, so an unnormalized value would make + * the confirmation loop wait forever — silently, which is the precise failure + * this confirmation exists to remove. + */ +function confirmDurationMs(value: number | undefined, fallback: number, max: number): number { + return typeof value === 'number' && Number.isFinite(value) && value > 0 ? Math.min(value, max) : fallback; +} + +/** + * Resolve a read, or give up on it once `ms` has elapsed. A `getInvocation` + * that outlives the remaining budget must not postpone the deadline check: the + * caller asked to stop waiting at a point in time, not after N more reads. + */ +async function raceConfirmRead( + read: Promise, + ms: number +): Promise { + let timer: ReturnType | undefined; + try { + return await Promise.race([ + read, + new Promise((resolve) => { + timer = setTimeout(() => resolve(READ_TIMED_OUT), ms); + }), + ]); + } finally { + // Leaving this pending would keep a CLI process alive for the full budget. + if (timer) clearTimeout(timer); + } +} + export class RelaycastMessagingClient implements RelayMessagingClient { readonly capabilities: RelayMessagingCapabilities; @@ -646,6 +702,22 @@ export class RelaycastMessagingClient implements RelayMessagingClient { ttlMs, }); const ack = await this.commands.invoke(actionName, actionInput); + // The ack proves only that the engine accepted the dispatch. Unless + // the caller asks for confirmation, a node that accepted the + // invocation and launched nothing resolves identically to a real + // spawn — see `confirm` on RelaySpawnPlacementInput. + const confirmation = input.confirm + ? await this.confirmPlacementInvocation(actionName, ack, { + capability, + node: decision.node.name, + repo, + attempts, + // Defaults and validation live in confirmPlacementInvocation + // so a direct caller cannot bypass them. + timeoutMs: input.confirmTimeoutMs, + pollIntervalMs: input.confirmPollIntervalMs, + }) + : undefined; return { ...ack, node: decision.node, @@ -655,7 +727,9 @@ export class RelaycastMessagingClient implements RelayMessagingClient { ...(repo ? { repo } : {}), attempts, queued, + confirmed: Boolean(confirmation), }, + ...(confirmation ? { confirmation } : {}), }; } @@ -737,6 +811,105 @@ export class RelaycastMessagingClient implements RelayMessagingClient { }, }; + /** + * Poll a dispatched invocation until the node reports a terminal result. + * + * Confirmation has to be observable from the requester, because the failure + * this guards against is a node that cannot report honestly: an obsolete + * broker advertises `spawn:` capacity, acks the invocation, and + * never launches or reports anything. Silence is therefore a failure, not a + * pending success — it times out as `spawn_unconfirmed`. + */ + private async confirmPlacementInvocation( + actionName: string, + ack: RelayActionInvocationAck, + context: { + capability: string; + node: string; + repo?: string; + attempts: number; + timeoutMs?: number; + pollIntervalMs?: number; + } + ): Promise { + const { timeoutMs, pollIntervalMs, ...errorContext } = context; + const invocationId = ack.invocationId; + if (!invocationId) { + throw new RelayPlacementError( + 'spawn_unconfirmed', + `node '${context.node}' accepted ${actionName} without returning an invocation id, so the dispatch cannot be confirmed`, + errorContext + ); + } + + // Normalized here rather than at the call site so every entry path is + // covered, including a direct SDK caller passing `NaN`. + const budgetMs = confirmDurationMs(timeoutMs, DEFAULT_CONFIRM_TIMEOUT_MS, MAX_CONFIRM_TIMEOUT_MS); + const cadenceMs = confirmDurationMs(pollIntervalMs, DEFAULT_CONFIRM_POLL_MS, budgetMs); + + // A missing actions API is a permanent misconfiguration, not a transient + // read failure. Polling it until the deadline would turn a clear error into + // a slow one that reads as an unresponsive node. + if (!this.agentClient?.actions) { + throw new RelayPlacementError( + 'spawn_unconfirmed', + `node '${context.node}' accepted ${actionName} (invocation ${invocationId}), but confirmation requires an agent-scoped client with the actions API, so the dispatch cannot be read back`, + errorContext + ); + } + + const deadline = Date.now() + budgetMs; + let lastReadError: string | undefined; + for (;;) { + const remainingMs = deadline - Date.now(); + if (remainingMs > 0) { + // Fold rejection into the value so abandoning a slow read below cannot + // surface as an unhandled rejection. + const read: Promise = this.commands.getInvocation(actionName, invocationId).then( + (value) => ({ ok: true, value }) as const, + (error) => ({ ok: false, error: error instanceof Error ? error.message : String(error) }) as const + ); + const outcome = await raceConfirmRead(read, remainingMs); + + if (outcome !== READ_TIMED_OUT) { + if (outcome.ok) { + // A later success must not report an earlier transient failure. + lastReadError = undefined; + const invocation = outcome.value; + const status = invocation?.status?.toLowerCase(); + if (status && CONFIRM_SUCCESS_STATUSES.has(status)) { + return invocation; + } + if (status && CONFIRM_FAILURE_STATUSES.has(status)) { + throw new RelayPlacementError( + 'spawn_failed', + invocation?.error?.trim() || `node '${context.node}' reported ${status} for ${actionName}`, + errorContext + ); + } + } else { + // A read failure is not evidence either way; keep polling until the + // deadline and report the last reason if we never get an answer. + lastReadError = outcome.error; + } + } + } + + if (Date.now() >= deadline) { + throw new RelayPlacementError( + 'spawn_unconfirmed', + `node '${context.node}' accepted ${actionName} (invocation ${invocationId}) but never reported a result within ${budgetMs}ms. ` + + `The node advertised capacity and acknowledged the dispatch; nothing confirmed that it launched. ` + + `Check that node's broker version, or re-run without confirmation to accept an unconfirmed dispatch.` + + (lastReadError ? ` Last read error: ${lastReadError}` : ''), + errorContext + ); + } + // Never sleep past the deadline: that would buy one more pointless read. + await delay(Math.max(0, Math.min(cadenceMs, deadline - Date.now()))); + } + } + readonly triggers = { list: async (): Promise => (await this.requireTriggers().list()).map(toRelayTrigger), create: async (input: RelayTriggerInput): Promise => diff --git a/packages/sdk/src/messaging/types.ts b/packages/sdk/src/messaging/types.ts index 6f329d99a..c13350231 100644 --- a/packages/sdk/src/messaging/types.ts +++ b/packages/sdk/src/messaging/types.ts @@ -605,6 +605,24 @@ export interface RelaySpawnPlacementInput { pollIntervalMs?: number; /** Fail immediately instead of queueing when no currently eligible node exists. */ failFast?: boolean; + /** + * Wait for the target node's terminal action result before resolving. + * + * Placement only proves the engine accepted the dispatch. A node running an + * obsolete broker advertises `spawn:` capacity, acknowledges the + * invocation, and launches nothing — without this the ack is identical to a + * real spawn. Defaults to `false` so plain dispatch keeps its semantics for + * non-spawn capabilities; agent-spawning callers should set it. + */ + confirm?: boolean; + /** + * How long to wait for that terminal result. Must exceed the node's own + * readiness window (the broker's `verify_ready` mode holds the action open + * for up to 90s). Defaults to 120000. + */ + confirmTimeoutMs?: number; + /** Poll cadence while awaiting confirmation. Defaults to 500. */ + confirmPollIntervalMs?: number; /** Placement log sink. Defaults to the client placement logger. */ log?: (message: string) => void; /** Reconcile hook for queue/fail visibility, e.g. Slack surfacing by callers. */ @@ -619,7 +637,15 @@ export interface RelaySpawnPlacementAck extends RelayActionInvocationAck { repo?: string; attempts: number; queued: boolean; + /** + * `true` only when the node reported a terminal success for this + * invocation. `false` means the dispatch was accepted but not confirmed — + * it was not observed to have launched anything. + */ + confirmed: boolean; }; + /** The node's terminal action result, present only when `confirm` was set. */ + confirmation?: RelayActionInvocation; } // ── Workspace ───────────────────────────────────────────────────────────────