From 69a060cddb4ed641d43f0def2562e806917549a8 Mon Sep 17 00:00:00 2001 From: Relayflow Date: Tue, 22 Sep 2026 18:47:10 +0000 Subject: [PATCH 1/2] fix(fleet): request and report explicit spawn readiness --- .../compact_wx0ujsgwtaqd_2026-09-22.json | 37 +++ .../compact_wx0ujsgwtaqd_2026-09-22.md | 18 ++ CHANGELOG.md | 8 +- crates/broker/src/node_control.rs | 12 +- crates/broker/src/pty_worker.rs | 2 +- crates/broker/src/runtime/fleet.rs | 34 ++- crates/broker/src/runtime/maintenance.rs | 2 +- crates/broker/src/runtime/relaycast_events.rs | 13 + crates/broker/src/runtime/worker_events.rs | 3 + packages/cli/src/cli/commands/fleet.test.ts | 155 ++++++------ packages/cli/src/cli/commands/fleet.ts | 6 +- .../cli/lib/fleet-spawn-confirmation.test.ts | 59 ++++- .../cli/src/cli/lib/spawn-lifecycle.test.ts | 32 +++ packages/cli/src/cli/lib/spawn-lifecycle.ts | 12 +- packages/fleet/src/serve-node.test.ts | 20 ++ packages/sdk/src/messaging/placement.test.mts | 121 +++++++++ .../sdk/src/messaging/relaycast-placement.ts | 3 +- packages/sdk/src/messaging/relaycast.ts | 36 ++- packages/sdk/src/messaging/types.ts | 10 + summary.md | 18 ++ .../broker/fleet-spawn-readiness.test.ts | 237 ++++++++++++++++++ 21 files changed, 745 insertions(+), 93 deletions(-) create mode 100644 .agentworkforce/trajectories/compacted/compact_wx0ujsgwtaqd_2026-09-22.json create mode 100644 .agentworkforce/trajectories/compacted/compact_wx0ujsgwtaqd_2026-09-22.md create mode 100644 packages/cli/src/cli/lib/spawn-lifecycle.test.ts create mode 100644 summary.md create mode 100644 tests/integration/broker/fleet-spawn-readiness.test.ts diff --git a/.agentworkforce/trajectories/compacted/compact_wx0ujsgwtaqd_2026-09-22.json b/.agentworkforce/trajectories/compacted/compact_wx0ujsgwtaqd_2026-09-22.json new file mode 100644 index 0000000000..0c01063a55 --- /dev/null +++ b/.agentworkforce/trajectories/compacted/compact_wx0ujsgwtaqd_2026-09-22.json @@ -0,0 +1,37 @@ +{ + "id": "compact_wx0ujsgwtaqd", + "version": 1, + "type": "compacted", + "compactedAt": "2026-09-22T18:45:48.437Z", + "sourceTrajectories": [ + "traj_4yi8dr1zi7ud" + ], + "dateRange": { + "start": "2026-09-22T18:37:03.545Z", + "end": "2026-09-22T18:45:47.737Z" + }, + "summary": { + "totalDecisions": 1, + "totalEvents": 1, + "uniqueAgents": [ + "default" + ] + }, + "decisionGroups": [ + { + "category": "other", + "decisions": [ + { + "question": "Keep action-result reconnect durability out of this fix", + "chosen": "Keep action-result reconnect durability out of this fix", + "reasoning": "Reviewed plan requires readiness contract repair plus diagnostics; engine replay idempotency is unproven.", + "fromTrajectory": "traj_4yi8dr1zi7ud" + } + ] + } + ], + "keyLearnings": [], + "keyFindings": [], + "filesAffected": [], + "commits": [] +} \ No newline at end of file diff --git a/.agentworkforce/trajectories/compacted/compact_wx0ujsgwtaqd_2026-09-22.md b/.agentworkforce/trajectories/compacted/compact_wx0ujsgwtaqd_2026-09-22.md new file mode 100644 index 0000000000..11925ffb57 --- /dev/null +++ b/.agentworkforce/trajectories/compacted/compact_wx0ujsgwtaqd_2026-09-22.md @@ -0,0 +1,18 @@ +# Trajectory Compaction: Sep 22, 2026 - Sep 22, 2026 + +## Summary +- Sessions: 1 +- Decisions: 1 +- Events: 1 +- Agents: default +- Files: 0 +- Commits: 0 + +## Other +- Keep action-result reconnect durability out of this fix -> Keep action-result reconnect durability out of this fix (traj_4yi8dr1zi7ud) + +## Key Learnings +- None + +## Key Findings +- None \ No newline at end of file diff --git a/CHANGELOG.md b/CHANGELOG.md index 2f8707793c..e43d81dbf9 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,10 +5,16 @@ 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] + +### Changed + +- Targeted `fleet spawn` now waits for harness readiness; the broker releases workers that miss its 90-second readiness window. Confirmed targeted spawns require `--confirm-timeout` of at least 90000ms. ### Fixed +- Targeted `fleet spawn` requests explicit readiness proof, preventing healthy launches from being rejected for missing proof; unconfirmed launches report `ready:false` while obsolete handlers remain rejected. + - Broker `manual_flush` recovery now replays a missing cumulative-ACK predecessor without duplicating an already-completed PTY injection, restores it ahead of parked successors, and reports the head/ACK/received sequence gap plus the reconciliation action in `message flush` and `message auto` results. ## [12.4.1] - 2026-09-22 diff --git a/crates/broker/src/node_control.rs b/crates/broker/src/node_control.rs index a4890bb605..4371e3b094 100644 --- a/crates/broker/src/node_control.rs +++ b/crates/broker/src/node_control.rs @@ -1727,7 +1727,13 @@ fn handle_disconnected_command( Some(FleetControlCommand::RegisterAgent { reply, .. }) => { let _ = reply.send(Err(register_agent_error.to_string())); } - Some(FleetControlCommand::Send(_)) | Some(FleetControlCommand::HeartbeatNow) => {} + Some(FleetControlCommand::Send(message)) => { + if let BrokerToRelaycast::ActionResult(result) = &message { + tracing::warn!(invocation_id = %result.invocation_id, frame_kind = "action.result", + "dropping fleet frame while disconnected"); + } + } + Some(FleetControlCommand::HeartbeatNow) => {} Some(FleetControlCommand::Shutdown) | None => return DisconnectedCommandOutcome::Shutdown, } DisconnectedCommandOutcome::Handled @@ -2258,6 +2264,10 @@ where } } if sent.is_err() { + if let BrokerToRelaycast::ActionResult(result) = &message { + tracing::warn!(invocation_id = %result.invocation_id, frame_kind = "action.result", + "dropping fleet frame after wire send failure"); + } return Break(ControlRunResult::Disconnected { application_ready: application_liveness.ready, }); diff --git a/crates/broker/src/pty_worker.rs b/crates/broker/src/pty_worker.rs index d7232f61cd..b1c42f4c10 100644 --- a/crates/broker/src/pty_worker.rs +++ b/crates/broker/src/pty_worker.rs @@ -224,7 +224,7 @@ const STARTUP_READY_WARNING: Duration = Duration::from_secs(25); /// that never receives its task is a total loss, while a brief typed a little /// early is recoverable. Bounded below `WORKER_READY_DEADLINE` (90s) so the work /// is released before an unready harness is reaped. -const STARTUP_READY_TIMEOUT: Duration = Duration::from_secs(60); +pub(crate) const STARTUP_READY_TIMEOUT: Duration = Duration::from_secs(60); const STARTUP_BUFFER_MAX: usize = 12_000; const STARTUP_BUFFER_KEEP: usize = 8_000; const CODEX_STARTUP_SETTLE: Duration = Duration::from_secs(1); diff --git a/crates/broker/src/runtime/fleet.rs b/crates/broker/src/runtime/fleet.rs index 5ed5350f5c..388115db7e 100644 --- a/crates/broker/src/runtime/fleet.rs +++ b/crates/broker/src/runtime/fleet.rs @@ -151,6 +151,7 @@ pub(super) fn close_terminal_sessions_for_worker( pub(super) struct PendingVerifiedSpawn { pub(super) invocation_id: String, pub(super) deadline: Instant, + pub(super) started: Instant, pub(super) generation: Uuid, } @@ -1432,6 +1433,9 @@ impl BrokerRuntime { /// to this node). Replies with `action.result { output }` on success or /// `{ error }` on failure. async fn handle_fleet_action_spawn(&mut self, invoke: ActionInvoke) { + let started = Instant::now(); + let verify_ready = super::relaycast_events::relaycast_spawn_verifies_ready(&invoke.input); + tracing::info!(invocation_id = %invoke.invocation_id, verify_ready, "fleet spawn received"); let Some(name) = action_invoke_agent_name(&invoke) else { self.reply_action_error(&invoke.invocation_id, "spawn_missing_agent_name") .await; @@ -1538,7 +1542,8 @@ impl BrokerRuntime { self.publish_fleet_load(true).await; - let verify_ready = super::relaycast_events::relaycast_spawn_verifies_ready(&ws_value); + tracing::info!(invocation_id = %invoke.invocation_id, worker = %name, verify_ready, + elapsed_ms = started.elapsed().as_millis() as u64, "fleet spawn launch returned"); let spawn_outcome = fleet_spawn_outcome(spawn_result, &name, self.workers.is_worker_live(&name)); @@ -1560,6 +1565,9 @@ impl BrokerRuntime { (worker.ready_at.is_some(), worker.generation) }; if already_ready { + tracing::info!(invocation_id = %invoke.invocation_id, worker = %name, + verify_ready, elapsed_ms = started.elapsed().as_millis() as u64, + "sending verified fleet spawn result"); self.send_fleet_action_result(verified_spawn_ready_result( invoke.invocation_id, &name, @@ -1571,6 +1579,7 @@ impl BrokerRuntime { PendingVerifiedSpawn { invocation_id: invoke.invocation_id, deadline: Instant::now() + VERIFIED_SPAWN_READY_TIMEOUT, + started, generation, }, ); @@ -1749,6 +1758,7 @@ impl BrokerRuntime { } async fn send_fleet_action_result(&self, result: ActionResult) { + tracing::info!(invocation_id = %result.invocation_id, "sending fleet action result"); let _ = self .fleet_control_tx .send(FleetControlCommand::Send(BrokerToRelaycast::ActionResult( @@ -1929,6 +1939,7 @@ pub(super) fn confirm_pending_delivery_and_resolve_fleet_ack( ((!already_held).then_some(pending), resolved) } +/// Every spawn success includes `ready`; true is reserved for proven harness readiness. fn fleet_spawn_action_result( invocation_id: &str, name: &WorkerName, @@ -1936,7 +1947,7 @@ fn fleet_spawn_action_result( ) -> ActionResult { let result = match spawn_result { Ok(()) => ActionResultPayload::Output(ActionResultOutput { - output: json!({ "spawned": true, "name": name.as_str() }), + output: json!({ "spawned": true, "ready": false, "name": name.as_str() }), }), Err(error) => ActionResultPayload::Error(ActionResultError { error: format!("spawn_failed: {error:#}"), @@ -3137,6 +3148,25 @@ mod tests { ); } + #[test] + fn spawn_success_always_declares_readiness() { + let name = WorkerName::from("Probe"); + let unverified = fleet_spawn_action_result("inv-launch", &name, Ok(())); + let verified = verified_spawn_ready_result("inv-ready".into(), &name); + for (result, ready) in [(unverified, false), (verified, true)] { + let ActionResultPayload::Output(output) = result.result else { + panic!("live spawn must succeed"); + }; + assert_eq!( + output.output, + json!({"spawned": true, "ready": ready, "name": "Probe"}) + ); + } + // CLI/SDK default confirmation budget is 120s (fleet.ts / relaycast.ts). + assert!(crate::pty_worker::STARTUP_READY_TIMEOUT < VERIFIED_SPAWN_READY_TIMEOUT); + assert!(VERIFIED_SPAWN_READY_TIMEOUT < Duration::from_secs(120)); + } + 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/maintenance.rs b/crates/broker/src/runtime/maintenance.rs index e82fe861b0..1b530db5dc 100644 --- a/crates/broker/src/runtime/maintenance.rs +++ b/crates/broker/src/runtime/maintenance.rs @@ -234,7 +234,7 @@ impl BrokerRuntime { let owned = workers.owned_spawn_generations.get(name).cloned(); let completion = super::fleet::verified_spawn_failed_result( invocation_id.clone(), - "spawn_readiness_timeout", + "spawn_readiness_timeout: worker released after failing to reach harness readiness", ); if let Some((_, http)) = owned { super::identity_cleanup::schedule_identity_cleanup( diff --git a/crates/broker/src/runtime/relaycast_events.rs b/crates/broker/src/runtime/relaycast_events.rs index 6e131282f0..64c0050971 100644 --- a/crates/broker/src/runtime/relaycast_events.rs +++ b/crates/broker/src/runtime/relaycast_events.rs @@ -1478,6 +1478,19 @@ mod tests { assert!(error.contains("harnessId is not supported")); } + #[test] + fn placement_spawn_requests_harness_readiness() { + // Literal SDK placementActionInput payload; persona stays engine-owned. + let mut payload = json!({"capability":"spawn:claude", "cli":"claude", + "node":"node-a", "target_node":"node-a", "name":"Probe", "verify_ready":true}); + assert!(relaycast_spawn_verifies_ready(&payload)); + payload.as_object_mut().unwrap().remove("verify_ready"); + assert!(!relaycast_spawn_verifies_ready(&payload)); + payload["capability"] = json!("spawn:persona"); + payload["cli"] = json!("persona"); + assert!(!relaycast_spawn_verifies_ready(&payload)); + } + #[test] fn verified_spawn_contract_is_read_from_request_or_harness_metadata() { let verified = json!({ diff --git a/crates/broker/src/runtime/worker_events.rs b/crates/broker/src/runtime/worker_events.rs index 0ad77b89bf..3e4ee83416 100644 --- a/crates/broker/src/runtime/worker_events.rs +++ b/crates/broker/src/runtime/worker_events.rs @@ -1391,6 +1391,9 @@ impl BrokerRuntime { .then(|| pending_verified_spawns.remove(&name)) .flatten(); if let Some(pending) = pending { + tracing::info!(invocation_id = %pending.invocation_id, worker = %name, + verify_ready = true, elapsed_ms = pending.started.elapsed().as_millis() as u64, + "sending verified fleet spawn result"); let _ = fleet_control_tx .send(FleetControlCommand::Send( crate::fleet_wire::BrokerToRelaycast::ActionResult( diff --git a/packages/cli/src/cli/commands/fleet.test.ts b/packages/cli/src/cli/commands/fleet.test.ts index f229c1936d..5e5c96eab6 100644 --- a/packages/cli/src/cli/commands/fleet.test.ts +++ b/packages/cli/src/cli/commands/fleet.test.ts @@ -800,7 +800,14 @@ describe('fleet command support', () => { invocationId: 'inv_targeted', actionName: 'spawn', node: { name: 'sf-mini' }, - placement: { capability: 'spawn:codex', node: 'sf-mini', attempts: 1, queued: false }, + placement: { + capability: 'spawn:codex', + node: 'sf-mini', + attempts: 1, + queued: false, + state: 'ready', + confirmed: true, + }, })), }; const createAgentRelay = vi.fn(() => ({ messaging: { placement } })); @@ -918,8 +925,9 @@ describe('fleet command support', () => { }) ); expect(createFleetWorkspaceClient).not.toHaveBeenCalled(); + expect(logs).toHaveLength(1); expect(JSON.parse(logs[0]!)).toMatchObject({ - invocation: { invocationId: 'inv_targeted' }, + invocation: { invocationId: 'inv_targeted', placement: { state: 'ready', confirmed: true } }, }); }); @@ -930,76 +938,81 @@ describe('fleet command support', () => { // evidence. With `--no-confirm`, the top-level invocation has no terminal // `status`, so a naive replacement would downgrade a confirmed SDK // `accepted` placement to `unconfirmed_may_be_running`. - it('preserves the SDK placement state and confirmed flag on a targeted --no-confirm spawn', async () => { - const placement = { - spawn: vi.fn(async () => ({ - invocationId: 'inv_no_confirm', - actionName: 'spawn', - node: { name: 'sf-mini' }, - placement: { - capability: 'spawn:codex', - node: 'sf-mini', - attempts: 1, - queued: false, - state: 'accepted', - confirmed: false, + it.each(['invoked', 'completed'])( + 'preserves the SDK placement for a targeted --no-confirm %s ack', + async (status) => { + const placement = { + spawn: vi.fn(async () => ({ + invocationId: 'inv_no_confirm', + status, + output: { spawned: true, ready: false }, + actionName: 'spawn', + node: { name: 'sf-mini' }, + placement: { + capability: 'spawn:codex', + node: 'sf-mini', + attempts: 1, + queued: false, + state: 'accepted', + confirmed: false, + }, + })), + }; + const createAgentRelay = vi.fn(() => ({ messaging: { placement } })); + const logs: string[] = []; + const program = new Command(); + program.exitOverride(); + registerFleetCommands(program, { + resolveSandboxRepository: () => undefined, + sdk: { + createAgentRelay: createAgentRelay as never, + createWorkspaceRelay: vi.fn() as never, + createWorkspace: vi.fn() as never, + log: (message: unknown) => logs.push(String(message)), + error: vi.fn(), + exit: vi.fn() as never, }, - })), - }; - const createAgentRelay = vi.fn(() => ({ messaging: { placement } })); - const logs: string[] = []; - const program = new Command(); - program.exitOverride(); - registerFleetCommands(program, { - resolveSandboxRepository: () => undefined, - sdk: { - createAgentRelay: createAgentRelay as never, - createWorkspaceRelay: vi.fn() as never, - createWorkspace: vi.fn() as never, - log: (message: unknown) => logs.push(String(message)), - error: vi.fn(), - exit: vi.fn() as never, - }, - createFleetWorkspaceClient: vi.fn() as never, - log: () => undefined, - warn: () => undefined, - error: () => undefined, - }); + createFleetWorkspaceClient: vi.fn() as never, + log: () => undefined, + warn: () => undefined, + error: () => undefined, + }); - await program.parseAsync( - [ - 'fleet', - 'spawn', - 'codex', - '--name', - 'api-worker', - '--task', - 'ACK and wait', - '--target-node', - 'sf-mini', - '--no-confirm', - '--workspace-key', - 'rk_live_test', - '--token', - 'at_live_lead', - ], - { from: 'user' } - ); + await program.parseAsync( + [ + 'fleet', + 'spawn', + 'codex', + '--name', + 'api-worker', + '--task', + 'ACK and wait', + '--target-node', + 'sf-mini', + '--no-confirm', + '--workspace-key', + 'rk_live_test', + '--token', + 'at_live_lead', + ], + { from: 'user' } + ); - const printed = JSON.parse(logs[0]!); - // The SDK's own evidence (state: 'accepted', confirmed: false) must - // survive untouched... - expect(printed.invocation.placement).toMatchObject({ - capability: 'spawn:codex', - node: 'sf-mini', - state: 'accepted', - confirmed: false, - }); - // ...augmented with the normalized dispatch evidence and invocation id, - // not replaced by them. - expect(printed.invocation.placement.dispatchState).toBeDefined(); - expect(printed.invocation.placement.state).not.toBe('unconfirmed_may_be_running'); - }); + const printed = JSON.parse(logs[0]!); + // The SDK's own evidence (state: 'accepted', confirmed: false) must + // survive untouched... + expect(printed.invocation.placement).toMatchObject({ + capability: 'spawn:codex', + node: 'sf-mini', + state: 'accepted', + confirmed: false, + }); + // ...augmented with the normalized dispatch evidence and invocation id, + // not replaced by them. + expect(printed.invocation.placement.dispatchState).toBeDefined(); + expect(printed.invocation.placement.state).not.toBe('unconfirmed_may_be_running'); + } + ); it('terminates an accepted-but-unconfirmed live invocation without inviting a blind retry', async () => { const invocationId = 'inv_223936432626290688'; @@ -3461,7 +3474,7 @@ describe('fleet command support', () => { expect(call).not.toHaveProperty('confirmTimeoutMs'); }); - it('fleet spawn rejects a non-numeric --confirm-timeout', async () => { + it.each(['soon', '30000', '89999'])('fleet spawn rejects invalid --confirm-timeout %s', async (timeout) => { const placement = { spawn: vi.fn() }; const program = new Command(); program.exitOverride(); @@ -3494,7 +3507,7 @@ describe('fleet command support', () => { '--node', 'sf-mini', '--confirm-timeout', - 'soon', + timeout, '--workspace-key', 'rk_live_test', '--token', diff --git a/packages/cli/src/cli/commands/fleet.ts b/packages/cli/src/cli/commands/fleet.ts index c447f50e70..63dac08829 100644 --- a/packages/cli/src/cli/commands/fleet.ts +++ b/packages/cli/src/cli/commands/fleet.ts @@ -449,7 +449,7 @@ export function registerFleetCommands( ) .option( '--confirm-timeout ', - 'How long a targeted spawn waits for the node to confirm the launch', + 'How long a targeted spawn waits for harness readiness (minimum 90000ms)', '120000' ) ).action(async (cli: string, options: Record) => { @@ -539,6 +539,10 @@ export function registerFleetCommands( ); const confirmTimeoutText = optionalText(options.confirmTimeout, 'Confirm timeout') ?? '120000'; const confirmTimeoutMs = Number(confirmTimeoutText); + // Broker VERIFIED_SPAWN_READY_TIMEOUT is 90s; default confirmation is 120s. + if ((targetNode || useSandbox) && options.confirm !== false && confirmTimeoutMs < 90_000) { + throw new Error('--confirm-timeout must be at least 90000ms for verified targeted spawns.'); + } if (!Number.isFinite(confirmTimeoutMs) || confirmTimeoutMs <= 0) { throw new Error('--confirm-timeout must be a positive number of milliseconds.'); } diff --git a/packages/cli/src/cli/lib/fleet-spawn-confirmation.test.ts b/packages/cli/src/cli/lib/fleet-spawn-confirmation.test.ts index b80c3dd8ce..83937278bd 100644 --- a/packages/cli/src/cli/lib/fleet-spawn-confirmation.test.ts +++ b/packages/cli/src/cli/lib/fleet-spawn-confirmation.test.ts @@ -193,7 +193,7 @@ describe('fleet spawn confirmation is observable from the requester (#1430)', () expect((error as RelayPlacementError).state).toBe('failed'); expect((error as RelayPlacementError).invocationId).toBe('inv-1430'); expect((error as RelayPlacementError).dispatchState).toBe('dispatched'); - expect((error as RelayPlacementError).message).toContain('spawned:true and ready:true proof'); + expect((error as RelayPlacementError).message).toContain('verify_ready'); }); // VACUITY CONTROL — without `confirm` the invocation is never read back, so @@ -318,3 +318,60 @@ describe('fleet spawn confirmation is observable from the requester (#1430)', () expect((error as Error).message).not.toContain('transient socket reset'); }); }); + +describe('targeted spawn readiness contract', () => { + it('requests readiness from a healthy remote claude node within the default budget', async () => { + const { client, invoke } = createClient(async (_name, invocationId) => ({ + invocation_id: invocationId, + status: 'completed', + output: { spawned: true, ready: invoke.mock.calls[0]?.[1]?.verify_ready === true }, + })); + const started = Date.now(); + const ack = await client.placement.spawn(spawnInput({ confirm: true })); + expect(ack.placement.state).toBe('ready'); + expect(ack.placement.confirmed).toBe(true); + expect(Date.now() - started).toBeLessThan(120_000); + expect(invoke).toHaveBeenCalledTimes(1); + }); + + it.each([false, true])('accepts an explicit unverified launch ack with ready=%s', async (ready) => { + const { client, invoke, reader } = createClient(); + invoke.mockResolvedValueOnce({ + invocation_id: 'launch', + status: 'completed', + output: { spawned: true, ready }, + } as never); + const ack = await client.placement.spawn(spawnInput({ confirm: false })); + expect(ack.placement).toMatchObject({ state: 'accepted', confirmed: false }); + expect(invoke.mock.calls[0]?.[1]).not.toHaveProperty('verify_ready'); + expect(reader).not.toHaveBeenCalled(); + }); + + it('rejects an obsolete handler missing the ready boolean even without confirmation', async () => { + const { client, invoke } = createClient(); + invoke.mockResolvedValueOnce({ + invocation_id: 'old', + status: 'completed', + output: { spawned: true }, + } as never); + await expect(client.placement.spawn(spawnInput({ confirm: false }))).rejects.toMatchObject({ + code: 'spawn_failed', + message: expect.stringContaining('verify_ready'), + }); + }); + + it('leaves persona verification unchanged', async () => { + const { client, invoke } = createClient(async () => ({ + status: 'completed', + output: { spawned: true, ready: true }, + })); + // Expose persona capacity only for this fixture. + LIVE_NODE.capabilities.push({ name: 'spawn:persona', kind: 'spawn' }); + try { + await client.placement.spawn(spawnInput({ capability: 'spawn:persona', confirm: true })); + expect(invoke.mock.calls[0]?.[1]).not.toHaveProperty('verify_ready'); + } finally { + LIVE_NODE.capabilities.pop(); + } + }); +}); diff --git a/packages/cli/src/cli/lib/spawn-lifecycle.test.ts b/packages/cli/src/cli/lib/spawn-lifecycle.test.ts new file mode 100644 index 0000000000..e325681a8e --- /dev/null +++ b/packages/cli/src/cli/lib/spawn-lifecycle.test.ts @@ -0,0 +1,32 @@ +import { describe, expect, it } from 'vitest'; +import { sanitizedSpawnReceipt, spawnLifecycleState } from './spawn-lifecycle.js'; + +describe('spawn lifecycle readiness modes', () => { + it.each([true, false])('requires a readiness boolean in mode %s', (verifyReady) => { + expect(spawnLifecycleState({ status: 'completed', output: { spawned: true } }, verifyReady)).toBe( + 'failed' + ); + expect(spawnLifecycleState({ status: 'completed', output: { ready: true } }, verifyReady)).toBe('failed'); + }); + it.each([true, false])('reports an unverified launch as accepted with ready=%s', (ready) => { + expect(spawnLifecycleState({ status: 'completed', output: { spawned: true, ready } }, false)).toBe( + 'accepted' + ); + }); + it('keeps readiness mandatory by default for MCP', () => { + expect(spawnLifecycleState({ status: 'completed', output: { spawned: true, ready: false } })).toBe( + 'failed' + ); + expect(spawnLifecycleState({ status: 'completed', output: { spawned: true, ready: true } })).toBe( + 'ready' + ); + }); + it('preserves ready:false in sanitized receipts', () => { + expect( + sanitizedSpawnReceipt({ + status: 'completed', + output: { spawned: true, ready: false, secret: 'hidden' }, + }) + ).toEqual({ status: 'completed', output: { spawned: true, ready: false } }); + }); +}); diff --git a/packages/cli/src/cli/lib/spawn-lifecycle.ts b/packages/cli/src/cli/lib/spawn-lifecycle.ts index fc23bd9a79..437f9ab5d9 100644 --- a/packages/cli/src/cli/lib/spawn-lifecycle.ts +++ b/packages/cli/src/cli/lib/spawn-lifecycle.ts @@ -25,17 +25,19 @@ function dispatchEvidence(value: Record): SpawnDispatchState { return 'unknown'; } -export function spawnLifecycleState(value: Record): SpawnLifecycleState { +export function spawnLifecycleState(value: Record, verifyReady = true): SpawnLifecycleState { const status = text(value.status)?.toLowerCase(); if (status && SUCCESS.has(status)) { const output = value.output !== null && typeof value.output === 'object' ? (value.output as Record) : value; - // A terminal success status without explicit launch and readiness proof is - // a failed spawn, not a live-but-uncertain one. Genuine uncertainty is - // reserved for non-terminal acknowledgements and confirmation timeouts. - return output.spawned === true && output.ready === true ? 'ready' : 'failed'; + // Terminal success must prove launch and declare readiness in either mode. + // MCP callers default to requiring readiness; unverified callers only accept + // launch. Uncertainty is reserved for pending actions and confirmation timeouts. + if (output.spawned !== true || typeof output.ready !== 'boolean') return 'failed'; + if (!verifyReady) return 'accepted'; + return output.ready === true ? 'ready' : 'failed'; } if (status && FAILURE.has(status)) return 'failed'; if (status === 'accepted') return 'accepted'; diff --git a/packages/fleet/src/serve-node.test.ts b/packages/fleet/src/serve-node.test.ts index b2892220cf..756f62ddd3 100644 --- a/packages/fleet/src/serve-node.test.ts +++ b/packages/fleet/src/serve-node.test.ts @@ -150,6 +150,26 @@ describe('serveNode', () => { await running.stop(); }); + it.each([true, false])( + 'rejects unverified delegated output only when verifyReady=%s', + async (verifyReady) => { + const fetchMock = vi.fn(async () => + Response.json({ data: { status: 'completed', output: { spawned: true, ready: false } } }) + ); + vi.stubGlobal('fetch', fetchMock); + const { running, sock, delegation } = await delegateForConfirmation(verifyReady); + sock.emit({ v: 1, id: delegation.id, type: 'reply', ok: true, data: { invocation_id: 'child' } }); + await vi.waitFor(() => expect(sock.sentOfType('action.result')).toHaveLength(1)); + const result = sock.sentOfType('action.result')[0]!; + if (verifyReady) expect(result.error).toBeTruthy(); + else { + expect(fetchMock).not.toHaveBeenCalled(); + expect(result.output).toMatchObject({ ready: false }); + } + await running.stop(); + } + ); + it.each(['pending', 'dispatched', 'invoked', 'running', 'read-timeout'])( 'keeps confirming through %s without dispatching another child', async (state) => { diff --git a/packages/sdk/src/messaging/placement.test.mts b/packages/sdk/src/messaging/placement.test.mts index 17707462a0..61385eab31 100644 --- a/packages/sdk/src/messaging/placement.test.mts +++ b/packages/sdk/src/messaging/placement.test.mts @@ -128,6 +128,7 @@ describe('RelaycastMessagingClient placement', () => { repo: 'relay', ttl_override_ms: 60, cli: 'claude', + verify_ready: true, }); }); @@ -278,6 +279,7 @@ describe('RelaycastMessagingClient placement', () => { capability: 'spawn:claude', ttl_override_ms: 60, cli: 'claude', + verify_ready: true, }); }); @@ -1266,3 +1268,122 @@ describe('RelaycastMessagingClient placement', () => { }); }); }); + +const LIVE_NODE = { + id: 'node_a', + name: 'node-a', + status: 'online', + live: true, + handlers_live: true, + capabilities: [{ name: 'spawn:claude', kind: 'spawn' }], + repo_keys: ['relay'], +}; + +function contractClient( + getInvocation?: (name: string, invocationId: string) => Promise, + acceptedInvocationId = 'inv-1430' +) { + const invoke = vi.fn(async (name: string, input?: Record) => ({ + invocation_id: acceptedInvocationId, + 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 contractSpawnInput(overrides: Record = {}) { + return { + capability: 'spawn:claude', + node: 'node-a', + repo: 'relay', + input: { name: 'worker-1430' }, + ...overrides, + }; +} + +describe('targeted spawn readiness contract', () => { + it('requests readiness from a healthy remote claude node within the default budget', async () => { + const { client, invoke } = contractClient(async (_name, invocationId) => ({ + invocation_id: invocationId, + status: 'completed', + output: { spawned: true, ready: invoke.mock.calls[0]?.[1]?.verify_ready === true }, + })); + const started = Date.now(); + const ack = await client.placement.spawn(contractSpawnInput({ confirm: true })); + expect(ack.placement.state).toBe('ready'); + expect(ack.placement.confirmed).toBe(true); + expect(Date.now() - started).toBeLessThan(120_000); + expect(invoke).toHaveBeenCalledTimes(1); + }); + + it.each([false, true])('accepts an explicit unverified launch ack with ready=%s', async (ready) => { + const { client, invoke, reader } = contractClient(); + invoke.mockResolvedValueOnce({ + invocation_id: 'launch', + status: 'completed', + output: { spawned: true, ready }, + } as never); + const ack = await client.placement.spawn(contractSpawnInput({ confirm: false })); + expect(ack.placement).toMatchObject({ state: 'accepted', confirmed: false }); + expect(invoke.mock.calls[0]?.[1]).not.toHaveProperty('verify_ready'); + expect(reader).not.toHaveBeenCalled(); + }); + + it('rejects an obsolete handler missing the ready boolean even without confirmation', async () => { + const { client, invoke } = contractClient(); + invoke.mockResolvedValueOnce({ + invocation_id: 'old', + status: 'completed', + output: { spawned: true }, + } as never); + await expect(client.placement.spawn(contractSpawnInput({ confirm: false }))).rejects.toMatchObject({ + code: 'spawn_failed', + message: expect.stringContaining('verify_ready'), + }); + }); + + it('leaves persona verification unchanged', async () => { + const { client, invoke } = contractClient(async () => ({ + status: 'completed', + output: { spawned: true, ready: true }, + })); + // Expose persona capacity only for this fixture. + LIVE_NODE.capabilities.push({ name: 'spawn:persona', kind: 'spawn' }); + try { + await client.placement.spawn(contractSpawnInput({ capability: 'spawn:persona', confirm: true })); + expect(invoke.mock.calls[0]?.[1]).not.toHaveProperty('verify_ready'); + } finally { + LIVE_NODE.capabilities.pop(); + } + }); +}); diff --git a/packages/sdk/src/messaging/relaycast-placement.ts b/packages/sdk/src/messaging/relaycast-placement.ts index 6ca375f83f..26d76ce9fa 100644 --- a/packages/sdk/src/messaging/relaycast-placement.ts +++ b/packages/sdk/src/messaging/relaycast-placement.ts @@ -153,10 +153,11 @@ export function placementActionName(capability: string): string { export function placementActionInput( input: Record | undefined, - placement: { capability: string; node?: string; repo?: string; ttlMs: number } + placement: { capability: string; node?: string; repo?: string; ttlMs: number; verifyReady?: boolean } ): Record { const payload = { ...(input ?? {}) }; payload.capability = placement.capability; + if (placement.verifyReady) payload.verify_ready = true; if (placement.node) { payload.node = placement.node; payload.target_node = placement.node; diff --git a/packages/sdk/src/messaging/relaycast.ts b/packages/sdk/src/messaging/relaycast.ts index add93e55ba..0da0f6c0e6 100644 --- a/packages/sdk/src/messaging/relaycast.ts +++ b/packages/sdk/src/messaging/relaycast.ts @@ -143,10 +143,26 @@ const CONFIRM_SUCCESS_STATUSES = new Set(['completed', 'succeeded', 'success']); */ const CONFIRM_FAILURE_STATUSES = new Set(['failed', 'error', 'denied', 'cancelled', 'canceled']); -function hasExplicitSpawnReadinessProof(value: { output?: Record | null }): boolean { +function hasSpawnReadinessProof(value: { output?: Record | null }): boolean { return value.output?.spawned === true && value.output?.ready === true; } +function hasSpawnLaunchProof(value: { output?: Record | null }): boolean { + return value.output?.spawned === true && typeof value.output?.ready === 'boolean'; +} + +function spawnProofError( + node: string, + status: string, + action: string, + output?: Record | null +): string { + if (output?.spawned === true && typeof output.ready !== 'boolean') { + return `node '${node}' handler did not honour verify_ready: missing explicit spawned:true and ready:true proof; upgrade to a release containing Relay PR #1708`; + } + return `node '${node}' reported ${status} for ${action} without explicit spawned:true and ready:true proof`; +} + /** Distinguishes "the read outlived its budget" from any value a read returns. */ const READ_TIMED_OUT = Symbol('relay.confirm.readTimedOut'); @@ -730,7 +746,12 @@ export class RelaycastMessagingClient implements RelayMessagingClient { // a node can die between this read and action invocation, after // which the engine treats it as a targeted queued placement. const clientMustTarget = Boolean(targetNode || repo || sandboxOnly); + // Persona resolves a nested child in the engine; its contract is unchanged. + const verifyReady = + input.verifyReady ?? + (input.confirm !== false && capability.startsWith('spawn:') && capability !== 'spawn:persona'); const actionInput = placementActionInput(input.input, { + verifyReady, capability, ...(clientMustTarget ? { node: decision.node.name } : {}), repo, @@ -772,11 +793,13 @@ export class RelaycastMessagingClient implements RelayMessagingClient { capability.startsWith('spawn:') && ackStatus && CONFIRM_SUCCESS_STATUSES.has(ackStatus) && - !hasExplicitSpawnReadinessProof({ output: ackOutput }) + !(capability === 'spawn:persona' + ? hasSpawnReadinessProof({ output: ackOutput }) + : hasSpawnLaunchProof({ output: ackOutput })) ) { throw new RelayPlacementError( 'spawn_failed', - `node '${placedNodeLabel}' reported ${ackStatus} for ${actionName} without explicit spawned:true and ready:true proof`, + spawnProofError(placedNodeLabel, ackStatus, actionName, ackOutput), { capability, node: placedNodeLabel, @@ -990,13 +1013,10 @@ export class RelaycastMessagingClient implements RelayMessagingClient { const invocation = outcome.value; const status = invocation?.status?.toLowerCase(); if (status && CONFIRM_SUCCESS_STATUSES.has(status)) { - if ( - errorContext.capability.startsWith('spawn:') && - !hasExplicitSpawnReadinessProof(invocation) - ) { + if (errorContext.capability.startsWith('spawn:') && !hasSpawnReadinessProof(invocation)) { throw new RelayPlacementError( 'spawn_failed', - `node '${context.node}' reported ${status} for ${actionName} without explicit spawned:true and ready:true proof`, + spawnProofError(context.node, status, actionName, invocation.output), { ...errorContext, state: 'failed', diff --git a/packages/sdk/src/messaging/types.ts b/packages/sdk/src/messaging/types.ts index 5cb245bb21..062c372b52 100644 --- a/packages/sdk/src/messaging/types.ts +++ b/packages/sdk/src/messaging/types.ts @@ -647,8 +647,18 @@ export interface RelaySpawnPlacementInput { * 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. + * Confirmed spawn success requires spawned:true and ready:true. Without + * confirmation, terminal launch success still requires spawned:true and a + * boolean ready field; ready:false means accepted, not verified readiness. */ confirm?: boolean; + /** + * Request proven harness readiness from the broker. Defaults to confirm !== false + * for spawn harnesses, and false for spawn:persona (engine-owned child input). + * Verified success carries spawned:true, ready:true. Unverified success carries + * spawned:true, ready:false; confirmation still requires ready:true. + */ + verifyReady?: 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 diff --git a/summary.md b/summary.md new file mode 100644 index 0000000000..8de7513e63 --- /dev/null +++ b/summary.md @@ -0,0 +1,18 @@ +Targeted fleet spawns now request the broker's verified-readiness contract, so a healthy worker can complete confirmation instead of being rejected for missing `ready:true`. Unverified broker success explicitly returns `ready:false`; `--no-confirm` accepts that launch evidence while continuing to reject obsolete handlers that omit readiness entirely. Persona defaults and MCP readiness requirements remain unchanged. + +The CLI rejects confirmed targeted-spawn timeouts below the broker's 90-second readiness window. Readiness timeout errors explain that the worker was released. Invocation-correlated spawn timing and dropped-result diagnostics help investigate the separate `spawn_unconfirmed` report. The changelog is raised to `[Unreleased - Minor]` as specified by the reviewed plan. + +Validation: + +- Full Vitest suite: 3,496 passed, 24 skipped (`env -u RELAY_BASE_URL npm --ignore-scripts test`, after building packages). Two ambient-URL failures disappeared with that environment override removed. The final additional completed-ack CLI regression passed in the 74-test fleet command suite. +- SDK placement suite: 47 passed, using a temporary Vitest configuration because the root configuration excludes SDK tests. +- Rust suite: 1,302 unit tests and 18 integration tests passed, 5 ignored (`env -u GIT_CONFIG_COUNT cargo test -p agent-relay-broker`). The environment's forced `core.hooksPath=/dev/null` caused four hook-test failures before its removal. +- `npm run typecheck` passed. `npm run lint` passed with 108 existing warnings. +- Real broker integration: `node --test dist/fleet-spawn-readiness.test.js` passed in approximately 1.1 seconds. A loopback engine forwards the SDK invocation to the real broker, which registers exactly one worker, launches a Claude stub in a real PTY, and returns `{spawned:true, ready:true}` through node control and requester polling. +- Mutate-to-red: removing `payload.verify_ready = true` from the built SDK made that same real-broker test fail with `spawn_failed`; restoring it passed. Removing `ready:false` from the Rust unverified result made `spawn_success_always_declares_readiness` fail; restoring it passed. The requester regression also failed against the original source before implementation. +- The broad `npm run test:integration:broker` run encountered continuity/event failures and stalled; it was stopped. The isolated readiness integration above passes. This is not a claim that the full broker integration suite passed. + +Remaining validation and follow-up: + +- Live two-node logged-in Claude, Codex, Gemini, Muse, and Devin checks were not run; the deterministic PTY proof does not establish real-harness login or prompt-detection behavior. +- The separate 120-second result silence remains open, per reviewed-plan.md. Investigate wire loss, slow spawn, and invocation-ID mismatch using the new logs. Action-result retention/replay is deferred until engine-side replay idempotency is established; no reconnect queue is included here. diff --git a/tests/integration/broker/fleet-spawn-readiness.test.ts b/tests/integration/broker/fleet-spawn-readiness.test.ts new file mode 100644 index 0000000000..60a838f94a --- /dev/null +++ b/tests/integration/broker/fleet-spawn-readiness.test.ts @@ -0,0 +1,237 @@ +/** Real broker/PTY, loopback engine, deterministic Claude stub; no live credentials. */ +import assert from 'node:assert/strict'; +import { spawn } from 'node:child_process'; +import { mkdtemp, mkdir, rm, writeFile } from 'node:fs/promises'; +import http from 'node:http'; +import { tmpdir } from 'node:os'; +import path from 'node:path'; +import { test } from 'node:test'; +import { RelaycastMessagingClient } from '@agent-relay/sdk'; +import { WebSocketServer, type WebSocket } from 'ws'; + +test( + 'targeted Claude spawn confirms real broker readiness inside the default budget', + { timeout: 130_000 }, + async () => { + const binary = + process.env.AGENT_RELAY_BROKER_BINARY ?? path.resolve('../../../target/debug/agent-relay-broker'); + const directory = await mkdtemp(path.join(tmpdir(), 'fleet-ready-')); + const bin = path.join(directory, 'bin'); + await mkdir(bin); + await writeFile(path.join(bin, 'claude'), "#!/bin/sh\nprintf '%s\\n' '->pty:ready'\nsleep 120\n", { + mode: 0o755, + }); + const receipts = new Map>(); + let nodeSocket: WebSocket | undefined; + let registrations = 0; + const server = http.createServer(async (request, response) => { + const chunks: Buffer[] = []; + for await (const chunk of request) chunks.push(Buffer.from(chunk)); + const body = chunks.length ? JSON.parse(Buffer.concat(chunks).toString()) : {}; + response.setHeader('content-type', 'application/json'); + if (request.method === 'GET' && request.url?.startsWith('/v1/agents/')) { + response.end( + JSON.stringify({ + ok: true, + data: { + id: 'worker-1', + name: 'Probe', + workspace_id: 'fixture-workspace', + status: 'online', + channels: [{ name: 'general' }, { name: 'engineering' }], + metadata: {}, + }, + }) + ); + return; + } + if (request.url?.endsWith('/members')) { + response.end( + JSON.stringify({ + ok: true, + data: [ + { + agent_id: 'worker-1', + agent_name: 'Probe', + role: 'member', + joined_at: '2026-01-01T00:00:00Z', + }, + ], + }) + ); + return; + } + response.end( + JSON.stringify({ + ok: true, + data: + request.url === '/v1/agents' + ? { + id: 'fixture-broker', + workspace_id: 'fixture-workspace', + name: body.name, + token: 'at_fixture', + status: 'online', + created_at: '2026-01-01T00:00:00Z', + } + : { + id: 'channel-fixture', + name: body.name ?? 'general', + workspace_id: 'fixture-workspace', + created_at: '2026-01-01T00:00:00Z', + created_by: 'fixture-broker', + is_archived: false, + topic: null, + members: [], + member_count: 1, + }, + }) + ); + }); + const wss = new WebSocketServer({ server }); + wss.on('connection', (socket, request) => { + const isNode = request.url?.startsWith('/v1/node/ws'); + socket.on('message', (raw) => { + const frame = JSON.parse(raw.toString()); + if (isNode && frame.type === 'inventory.sync') nodeSocket = socket; + if (frame.type === 'action.result') + receipts.set(frame.invocation_id, { + invocation_id: frame.invocation_id, + status: frame.error ? 'failed' : 'completed', + output: frame.output, + error: frame.error, + }); + if (['node.register', 'inventory.sync', 'agent.register', 'agent.deregister'].includes(frame.type)) { + if (frame.type === 'agent.register') registrations++; + socket.send( + JSON.stringify({ + v: 1, + type: 'reply', + id: frame.id, + ok: true, + data: + frame.type === 'agent.register' + ? { + agent_id: `worker-${registrations}`, + token: 'at_worker_fixture', + name: frame.name, + delivery_ack_seq: 0, + } + : {}, + }) + ); + } + }); + }); + await new Promise((resolve) => server.listen(0, '127.0.0.1', resolve)); + const address = server.address(); + assert(address && typeof address !== 'string'); + const baseUrl = `http://127.0.0.1:${address.port}`; + const broker = spawn( + binary, + [ + 'init', + '--instance-name', + 'ready-node', + '--workspace-key', + 'rk_fixture', + '--state-dir', + directory, + '--api-port', + '0', + '--channels', + '', + ], + { + cwd: directory, + env: { + PATH: `${bin}:${process.env.PATH}`, + HOME: directory, + TMPDIR: directory, + RELAYCAST_BASE_URL: baseUrl, + RELAY_BASE_URL: baseUrl, + RELAY_BROKER_API_KEY: 'br_fixture', + RELAY_NODE_ID: 'node-ready', + RELAY_NODE_TOKEN: 'nt_fixture', + AGENT_RELAY_BROKER_LOG: 'stderr', + AGENT_RELAY_TELEMETRY_DISABLED: '1', + AGENT_RELAY_NO_DEBUG_FILES: '1', + }, + stdio: ['ignore', 'pipe', 'pipe'], + } + ); + let logs = ''; + let spawnError: Error | undefined; + broker.on('error', (error) => { + spawnError = error; + }); + broker.stdout.on('data', (chunk) => { + logs = (logs + chunk).slice(-16000); + if (process.env.FLEET_TEST_DEBUG) process.stderr.write(chunk); + }); + broker.stderr.on('data', (chunk) => { + logs = (logs + chunk).slice(-16000); + if (process.env.FLEET_TEST_DEBUG) process.stderr.write(chunk); + }); + try { + const deadline = Date.now() + 10_000; + while (!nodeSocket && Date.now() < deadline && !spawnError && broker.exitCode === null) { + await new Promise((resolve) => setTimeout(resolve, 20)); + } + assert.ifError(spawnError); + assert(nodeSocket, `broker did not connect: ${logs}`); + const node = { + id: 'node-ready', + name: 'ready-node', + status: 'online', + live: true, + handlers_live: true, + capabilities: [{ name: 'spawn:claude', kind: 'spawn' }], + }; + let invocations = 0; + const client = new RelaycastMessagingClient({ + relaycast: { nodes: { list: async () => [node], get: async () => node } } as never, + agentClient: { + actions: { + invoke: async (_name: string, input: Record) => { + const id = `inv-${++invocations}`; + nodeSocket!.send( + JSON.stringify({ v: 1, type: 'action.invoke', invocation_id: id, action: 'spawn', input }) + ); + return { invocation_id: id, status: 'invoked', handler_node_id: node.id }; + }, + getInvocation: async (_name: string, id: string) => + receipts.get(id) ?? { invocation_id: id, status: 'invoked' }, + }, + } as never, + }); + const started = Date.now(); + const result = await client.placement.spawn({ + capability: 'spawn:claude', + node: node.name, + confirm: true, + input: { name: 'Probe' }, + }); + assert.equal(result.placement.state, 'ready', logs); + assert(Date.now() - started < 120_000); + assert.equal(invocations, 1); + assert.equal(registrations, 1); + assert.deepEqual(result.confirmation?.output, { spawned: true, ready: true, name: 'Probe' }); + } catch (error) { + throw new Error(`${String(error)}\nBroker output:\n${logs}`, { cause: error }); + } finally { + if (broker.exitCode === null && !spawnError) { + const exited = new Promise((resolve) => broker.once('exit', resolve)); + broker.kill('SIGTERM'); + const timer = setTimeout(() => broker.kill('SIGKILL'), 2000); + await exited; + clearTimeout(timer); + } + for (const socket of wss.clients) socket.terminate(); + await new Promise((resolve) => wss.close(() => resolve())); + server.closeAllConnections(); + await new Promise((resolve) => server.close(() => resolve())); + await rm(directory, { recursive: true, force: true }); + } + } +); From 7c2546fc90635b281f88d4cc587c1e75cbebb027 Mon Sep 17 00:00:00 2001 From: Relayflow Date: Tue, 22 Sep 2026 19:13:17 +0000 Subject: [PATCH 2/2] fix(sdk): resolve the spawn readiness mode once and judge only what was asked MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Review of #1843 found the new readiness contract was requested and judged by three predicates that disagreed. This resolves it once, per request. - `placement.spawn` derived `verify_ready` from `confirm !== false`, so the documented default (`confirm` omitted) asked the broker to hold the action open and release the worker at 90s while the caller returned immediately — killing a worker that would otherwise have survived. Gate it on `confirm === true`, matching the two predicates that consume it. - The confirmation poll demanded `ready:true` for every `spawn:*` invocation regardless of the mode requested, so `verifyReady: false` failed a healthy launch with the issue's own error string. Thread the resolved mode through and judge the requested contract on both the ack and poll paths. Persona stays pinned to proven readiness: its engine-owned handler never reads `verify_ready` but does report readiness itself. - `state: 'ready'` is now reserved for a confirmation that actually verified readiness; a launch-only confirmation reports `confirmed` + `accepted`. - The "did not honour verify_ready" message no longer fires on the path that never sent `verify_ready`; there it names the missing `ready` boolean. - `--confirm-timeout` floor raised to 95000ms. The broker's 90s readiness window starts after launch, registration and token minting, so equal budgets do not nest and 90000 reported a released worker as `spawn_unconfirmed`. The floor check now runs after the numeric guard so `-5` reports what is actually wrong with it. - The readiness-timeout failure result bypasses `send_fleet_action_result`; log its correlation fields so the most-investigated outcome is traceable. - Drop the unreachable `verifyReady` parameter on `spawnLifecycleState`, use the suite's binary-resolution helpers in the broker integration test, remove the SDK placement arms that CI never runs, and drop the committed `summary.md` workflow artifact. Co-Authored-By: Claude Opus 5 --- CHANGELOG.md | 4 +- crates/broker/src/runtime/maintenance.rs | 7 + packages/cli/src/cli/commands/fleet.test.ts | 84 ++++++++---- packages/cli/src/cli/commands/fleet.ts | 33 ++++- .../cli/lib/fleet-spawn-confirmation.test.ts | 114 +++++++++++++++-- .../cli/src/cli/lib/spawn-lifecycle.test.ts | 22 ++-- packages/cli/src/cli/lib/spawn-lifecycle.ts | 15 ++- packages/sdk/src/messaging/placement.test.mts | 121 ------------------ packages/sdk/src/messaging/relaycast.ts | 69 ++++++++-- packages/sdk/src/messaging/types.ts | 29 +++-- summary.md | 18 --- .../broker/fleet-spawn-readiness.test.ts | 18 ++- 12 files changed, 305 insertions(+), 229 deletions(-) delete mode 100644 summary.md diff --git a/CHANGELOG.md b/CHANGELOG.md index e43d81dbf9..44ca92659f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,12 +9,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Changed -- Targeted `fleet spawn` now waits for harness readiness; the broker releases workers that miss its 90-second readiness window. Confirmed targeted spawns require `--confirm-timeout` of at least 90000ms. +- Targeted `fleet spawn` now waits for harness readiness; the broker releases workers that miss its 90-second readiness window. Confirmed targeted spawns require `--confirm-timeout` of at least 95000ms. ### Fixed - Targeted `fleet spawn` requests explicit readiness proof, preventing healthy launches from being rejected for missing proof; unconfirmed launches report `ready:false` while obsolete handlers remain rejected. - +- `@agent-relay/sdk` `placement.spawn` only asks a node to verify readiness when it will wait for the answer, and confirms against the contract it requested, so `confirm` omitted no longer arms a 90-second readiness kill switch and `verifyReady: false` no longer fails a healthy launch. - Broker `manual_flush` recovery now replays a missing cumulative-ACK predecessor without duplicating an already-completed PTY injection, restores it ahead of parked successors, and reports the head/ACK/received sequence gap plus the reconciliation action in `message flush` and `message auto` results. ## [12.4.1] - 2026-09-22 diff --git a/crates/broker/src/runtime/maintenance.rs b/crates/broker/src/runtime/maintenance.rs index 1b530db5dc..aac7a0c22a 100644 --- a/crates/broker/src/runtime/maintenance.rs +++ b/crates/broker/src/runtime/maintenance.rs @@ -236,6 +236,13 @@ impl BrokerRuntime { invocation_id.clone(), "spawn_readiness_timeout: worker released after failing to reach harness readiness", ); + // This result goes out through identity cleanup or the fleet channel + // directly rather than `send_fleet_action_result`, so it would + // otherwise be the one spawn outcome missing the correlation log — + // and it is the outcome most likely to be investigated. + tracing::info!(invocation_id = %invocation_id, worker = %name, verify_ready = true, + deferred_to_identity_cleanup = owned.is_some(), + "sending fleet action result"); if let Some((_, http)) = owned { super::identity_cleanup::schedule_identity_cleanup( workers, diff --git a/packages/cli/src/cli/commands/fleet.test.ts b/packages/cli/src/cli/commands/fleet.test.ts index 5e5c96eab6..311528d09a 100644 --- a/packages/cli/src/cli/commands/fleet.test.ts +++ b/packages/cli/src/cli/commands/fleet.test.ts @@ -3474,11 +3474,18 @@ describe('fleet command support', () => { expect(call).not.toHaveProperty('confirmTimeoutMs'); }); - it.each(['soon', '30000', '89999'])('fleet spawn rejects invalid --confirm-timeout %s', async (timeout) => { - const placement = { spawn: vi.fn() }; + function confirmTimeoutHarness() { + const placement = { + spawn: vi.fn(async () => ({ + invocationId: 'inv_timeout_floor', + actionName: 'spawn', + node: { name: 'sf-mini' }, + placement: { capability: 'spawn:codex', node: 'sf-mini', attempts: 1, queued: false }, + })), + }; + const errors: string[] = []; const program = new Command(); program.exitOverride(); - const errors: unknown[] = []; registerFleetCommands(program, { resolveSandboxRepository: () => undefined, sdk: { @@ -3486,7 +3493,7 @@ describe('fleet command support', () => { createWorkspaceRelay: vi.fn() as never, createWorkspace: vi.fn() as never, log: () => undefined, - error: (message: unknown) => errors.push(message), + error: (message: unknown) => errors.push(String(message)), exit: vi.fn() as never, }, createFleetWorkspaceClient: vi.fn() as never, @@ -3494,30 +3501,57 @@ describe('fleet command support', () => { warn: () => undefined, error: () => undefined, }); + return { program, placement, errors }; + } + + function confirmTimeoutArgv(timeout: string): string[] { + return [ + 'fleet', + 'spawn', + 'codex', + '--name', + 'api-worker', + '--task', + 'ACK and wait', + '--node', + 'sf-mini', + '--confirm-timeout', + timeout, + '--workspace-key', + 'rk_live_test', + '--token', + 'at_live_lead', + ]; + } - await program.parseAsync( - [ - 'fleet', - 'spawn', - 'codex', - '--name', - 'api-worker', - '--task', - 'ACK and wait', - '--node', - 'sf-mini', - '--confirm-timeout', - timeout, - '--workspace-key', - 'rk_live_test', - '--token', - 'at_live_lead', - ], - { from: 'user' } - ); + // The floor and the non-numeric guard produce different errors, and each + // arm asserts the one it should get: a value that is simply not a number + // must not be reported as being below the floor. + it.each([ + ['soon', 'must be a positive number of milliseconds'], + ['-5', 'must be a positive number of milliseconds'], + ['30000', 'must be at least 95000ms'], + ['94999', 'must be at least 95000ms'], + ])('fleet spawn rejects --confirm-timeout %s', async (timeout, expected) => { + const { program, placement, errors } = confirmTimeoutHarness(); + + await program.parseAsync(confirmTimeoutArgv(timeout), { from: 'user' }); expect(placement.spawn).not.toHaveBeenCalled(); - expect(String(errors.join('\n'))).toContain('--confirm-timeout'); + expect(errors.join('\n')).toContain(expected); + }); + + // The first accepted value. The broker's own readiness window is 90000ms and + // starts after the launch work completes, so a budget at the floor is the + // smallest one that can still contain it. + it('fleet spawn accepts --confirm-timeout at the floor', async () => { + const { program, placement, errors } = confirmTimeoutHarness(); + + await program.parseAsync(confirmTimeoutArgv('95000'), { from: 'user' }); + + expect(errors).toEqual([]); + expect(placement.spawn).toHaveBeenCalledTimes(1); + expect(placement.spawn.mock.calls[0]![0]).toMatchObject({ confirm: true, confirmTimeoutMs: 95_000 }); }); describe('local default spawn', () => { diff --git a/packages/cli/src/cli/commands/fleet.ts b/packages/cli/src/cli/commands/fleet.ts index 63dac08829..1f607bb067 100644 --- a/packages/cli/src/cli/commands/fleet.ts +++ b/packages/cli/src/cli/commands/fleet.ts @@ -74,6 +74,21 @@ const FLEET_CLIS = new Set([ 'opencode', 'devin', ]); +/** + * Floor for `--confirm-timeout` on a verified targeted spawn. + * + * The broker holds a verified spawn open for its 90s + * `VERIFIED_SPAWN_READY_TIMEOUT` (`crates/broker/src/runtime/fleet.rs`), but + * starts that clock only after `spawn_worker_from_request` returns — i.e. after + * process creation, the stability window, agent registration and token minting. + * The requester's budget starts earlier, at the dispatch ack, so the two + * windows only nest when the requester's is strictly larger. The 5s margin + * covers the invoke round-trip, that launch work, and one + * `DEFAULT_CONFIRM_POLL_MS` (500ms) confirmation poll. Below this floor a worker + * the broker already released with `spawn_readiness_timeout` is reported as + * `spawn_unconfirmed` — the failure shape this confirmation exists to remove. + */ +const MIN_VERIFIED_CONFIRM_TIMEOUT_MS = 95_000; const CLOUD_SANDBOX_ID_PATTERN = /^sbx_[0-9a-f]{8}-[0-9a-f]{4}-[1-5][0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$/; @@ -449,7 +464,7 @@ export function registerFleetCommands( ) .option( '--confirm-timeout ', - 'How long a targeted spawn waits for harness readiness (minimum 90000ms)', + `How long a targeted spawn waits for harness readiness (minimum ${MIN_VERIFIED_CONFIRM_TIMEOUT_MS}ms)`, '120000' ) ).action(async (cli: string, options: Record) => { @@ -539,13 +554,21 @@ export function registerFleetCommands( ); const confirmTimeoutText = optionalText(options.confirmTimeout, 'Confirm timeout') ?? '120000'; const confirmTimeoutMs = Number(confirmTimeoutText); - // Broker VERIFIED_SPAWN_READY_TIMEOUT is 90s; default confirmation is 120s. - if ((targetNode || useSandbox) && options.confirm !== false && confirmTimeoutMs < 90_000) { - throw new Error('--confirm-timeout must be at least 90000ms for verified targeted spawns.'); - } if (!Number.isFinite(confirmTimeoutMs) || confirmTimeoutMs <= 0) { throw new Error('--confirm-timeout must be a positive number of milliseconds.'); } + // Checked after the numeric guard so a negative value reports what is + // actually wrong with it rather than the floor. + if ( + (targetNode || useSandbox) && + options.confirm !== false && + confirmTimeoutMs < MIN_VERIFIED_CONFIRM_TIMEOUT_MS + ) { + throw new Error( + `--confirm-timeout must be at least ${MIN_VERIFIED_CONFIRM_TIMEOUT_MS}ms for verified targeted spawns; ` + + "the node's own readiness window is 90000ms and starts after the launch completes." + ); + } let sandbox: EnsureCloudFleetSandboxResult | undefined; let sandboxRepository: SandboxRepositorySelection | undefined; diff --git a/packages/cli/src/cli/lib/fleet-spawn-confirmation.test.ts b/packages/cli/src/cli/lib/fleet-spawn-confirmation.test.ts index 83937278bd..13a7b8082c 100644 --- a/packages/cli/src/cli/lib/fleet-spawn-confirmation.test.ts +++ b/packages/cli/src/cli/lib/fleet-spawn-confirmation.test.ts @@ -320,20 +320,38 @@ describe('fleet spawn confirmation is observable from the requester (#1430)', () }); describe('targeted spawn readiness contract', () => { - it('requests readiness from a healthy remote claude node within the default budget', async () => { + it('requests readiness from a healthy remote claude node and confirms it', async () => { const { client, invoke } = createClient(async (_name, invocationId) => ({ invocation_id: invocationId, status: 'completed', + // The broker only proves readiness when it was asked to. output: { spawned: true, ready: invoke.mock.calls[0]?.[1]?.verify_ready === true }, })); - const started = Date.now(); const ack = await client.placement.spawn(spawnInput({ confirm: true })); + expect(invoke.mock.calls[0]?.[1]).toMatchObject({ verify_ready: true }); expect(ack.placement.state).toBe('ready'); expect(ack.placement.confirmed).toBe(true); - expect(Date.now() - started).toBeLessThan(120_000); expect(invoke).toHaveBeenCalledTimes(1); }); + // MUST-FIRE for the `confirm`-omitted default. Requesting broker-side + // verification commits the broker to holding the action open and *releasing + // the worker* if readiness never arrives. A caller that does not wait for + // that answer must not ask for it, or the default dispatch silently acquires + // a 90-second kill switch nobody is watching. + it('does not request broker verification when confirmation is omitted', async () => { + const { client, invoke, reader } = createClient(); + invoke.mockResolvedValueOnce({ + invocation_id: 'default', + status: 'completed', + output: { spawned: true, ready: false }, + } as never); + const ack = await client.placement.spawn(spawnInput()); + expect(invoke.mock.calls[0]?.[1]).not.toHaveProperty('verify_ready'); + expect(ack.placement).toMatchObject({ state: 'accepted', confirmed: false }); + expect(reader).not.toHaveBeenCalled(); + }); + it.each([false, true])('accepts an explicit unverified launch ack with ready=%s', async (ready) => { const { client, invoke, reader } = createClient(); invoke.mockResolvedValueOnce({ @@ -347,31 +365,103 @@ describe('targeted spawn readiness contract', () => { expect(reader).not.toHaveBeenCalled(); }); - it('rejects an obsolete handler missing the ready boolean even without confirmation', async () => { + // MUST-FIRE for `verifyReady: false` with confirmation. The requester told + // the broker not to verify readiness, so judging the result against + // `ready:true` would reject a healthy launch with the exact error string from + // the issue — through an option the SDK itself offers. + it('confirms a verifyReady:false spawn against the launch contract it asked for', async () => { + const { client, invoke, reader } = createClient(async (_name, invocationId) => ({ + invocation_id: invocationId, + status: 'completed', + output: { spawned: true, ready: false }, + })); + const ack = await client.placement.spawn( + spawnInput({ confirm: true, verifyReady: false, confirmPollIntervalMs: 10 }) + ); + expect(invoke.mock.calls[0]?.[1]).not.toHaveProperty('verify_ready'); + expect(reader).toHaveBeenCalled(); + // Confirmed as launched, but never proven ready — `ready` is reserved for + // the mode that actually asked for readiness. + expect(ack.placement).toMatchObject({ state: 'accepted', confirmed: true }); + }); + + it('still rejects a confirmed readiness request that reports ready:false', async () => { + const { client } = createClient(async (_name, invocationId) => ({ + invocation_id: invocationId, + status: 'completed', + output: { spawned: true, ready: false }, + })); + await expect( + client.placement.spawn(spawnInput({ confirm: true, confirmPollIntervalMs: 10 })) + ).rejects.toMatchObject({ + code: 'spawn_failed', + message: expect.stringContaining('without explicit spawned:true and ready:true proof'), + }); + }); + + it('blames the missing ready boolean, not verify_ready, when readiness was not requested', async () => { const { client, invoke } = createClient(); invoke.mockResolvedValueOnce({ invocation_id: 'old', status: 'completed', output: { spawned: true }, } as never); - await expect(client.placement.spawn(spawnInput({ confirm: false }))).rejects.toMatchObject({ - code: 'spawn_failed', - message: expect.stringContaining('verify_ready'), - }); + const error = await client.placement.spawn(spawnInput({ confirm: false })).catch((e: unknown) => e); + expect(error).toBeInstanceOf(RelayPlacementError); + expect((error as RelayPlacementError).code).toBe('spawn_failed'); + expect((error as Error).message).toContain('without an explicit ready boolean'); + // This path never sent verify_ready, so naming it would send the next + // investigation to the wrong layer — which is how #1430 was first misread. + expect((error as Error).message).not.toContain('did not honour verify_ready'); }); - it('leaves persona verification unchanged', async () => { - const { client, invoke } = createClient(async () => ({ + it('names verify_ready when readiness was requested and the handler ignored it', async () => { + const { client } = createClient(async (_name, invocationId) => ({ + invocation_id: invocationId, status: 'completed', - output: { spawned: true, ready: true }, + output: { spawned: true }, })); + await expect( + client.placement.spawn(spawnInput({ confirm: true, confirmPollIntervalMs: 10 })) + ).rejects.toMatchObject({ + code: 'spawn_failed', + message: expect.stringContaining('did not honour verify_ready'), + }); + }); + + it('leaves persona engine-owned: no verify_ready on the wire', async () => { + const { client, invoke } = createClient(); // Expose persona capacity only for this fixture. LIVE_NODE.capabilities.push({ name: 'spawn:persona', kind: 'spawn' }); try { - await client.placement.spawn(spawnInput({ capability: 'spawn:persona', confirm: true })); + await client.placement.spawn(spawnInput({ capability: 'spawn:persona', confirm: false })); expect(invoke.mock.calls[0]?.[1]).not.toHaveProperty('verify_ready'); } finally { LIVE_NODE.capabilities.pop(); } }); + + // Persona never sees `verify_ready`, but its engine-owned handler still + // reports proven readiness, so relaxing it to the launch contract would + // accept a child that was dispatched and never came up. + it('keeps persona confirmation pinned to proven readiness', async () => { + const { client } = createClient(async (_name, invocationId) => ({ + invocation_id: invocationId, + status: 'completed', + output: { spawned: true, ready: false }, + })); + LIVE_NODE.capabilities.push({ name: 'spawn:persona', kind: 'spawn' }); + try { + await expect( + client.placement.spawn( + spawnInput({ capability: 'spawn:persona', confirm: true, confirmPollIntervalMs: 10 }) + ) + ).rejects.toMatchObject({ + code: 'spawn_failed', + message: expect.stringContaining('without explicit spawned:true and ready:true proof'), + }); + } finally { + LIVE_NODE.capabilities.pop(); + } + }); }); diff --git a/packages/cli/src/cli/lib/spawn-lifecycle.test.ts b/packages/cli/src/cli/lib/spawn-lifecycle.test.ts index e325681a8e..9b0a722f1f 100644 --- a/packages/cli/src/cli/lib/spawn-lifecycle.test.ts +++ b/packages/cli/src/cli/lib/spawn-lifecycle.test.ts @@ -1,19 +1,16 @@ import { describe, expect, it } from 'vitest'; import { sanitizedSpawnReceipt, spawnLifecycleState } from './spawn-lifecycle.js'; -describe('spawn lifecycle readiness modes', () => { - it.each([true, false])('requires a readiness boolean in mode %s', (verifyReady) => { - expect(spawnLifecycleState({ status: 'completed', output: { spawned: true } }, verifyReady)).toBe( - 'failed' - ); - expect(spawnLifecycleState({ status: 'completed', output: { ready: true } }, verifyReady)).toBe('failed'); - }); - it.each([true, false])('reports an unverified launch as accepted with ready=%s', (ready) => { - expect(spawnLifecycleState({ status: 'completed', output: { spawned: true, ready } }, false)).toBe( - 'accepted' - ); +describe('spawn lifecycle readiness', () => { + it('rejects a terminal success without launch and readiness proof', () => { + expect(spawnLifecycleState({ status: 'completed', output: { spawned: true } })).toBe('failed'); + expect(spawnLifecycleState({ status: 'completed', output: { ready: true } })).toBe('failed'); }); - it('keeps readiness mandatory by default for MCP', () => { + + // The broker's unverified success path declares `ready:false` rather than + // omitting the field. This receipt is only derived for callers that required + // readiness, so an honest "launched, not ready" is still not a ready spawn. + it('treats a declared ready:false as a failed spawn for readiness callers', () => { expect(spawnLifecycleState({ status: 'completed', output: { spawned: true, ready: false } })).toBe( 'failed' ); @@ -21,6 +18,7 @@ describe('spawn lifecycle readiness modes', () => { 'ready' ); }); + it('preserves ready:false in sanitized receipts', () => { expect( sanitizedSpawnReceipt({ diff --git a/packages/cli/src/cli/lib/spawn-lifecycle.ts b/packages/cli/src/cli/lib/spawn-lifecycle.ts index 437f9ab5d9..45042837f4 100644 --- a/packages/cli/src/cli/lib/spawn-lifecycle.ts +++ b/packages/cli/src/cli/lib/spawn-lifecycle.ts @@ -25,19 +25,20 @@ function dispatchEvidence(value: Record): SpawnDispatchState { return 'unknown'; } -export function spawnLifecycleState(value: Record, verifyReady = true): SpawnLifecycleState { +export function spawnLifecycleState(value: Record): SpawnLifecycleState { const status = text(value.status)?.toLowerCase(); if (status && SUCCESS.has(status)) { const output = value.output !== null && typeof value.output === 'object' ? (value.output as Record) : value; - // Terminal success must prove launch and declare readiness in either mode. - // MCP callers default to requiring readiness; unverified callers only accept - // launch. Uncertainty is reserved for pending actions and confirmation timeouts. - if (output.spawned !== true || typeof output.ready !== 'boolean') return 'failed'; - if (!verifyReady) return 'accepted'; - return output.ready === true ? 'ready' : 'failed'; + // Every caller of this receipt (MCP spawn, `fleet spawn --auto-place`) + // requires proven readiness. A terminal success without explicit launch and + // readiness proof is a failed spawn, not a live-but-uncertain one. Genuine + // uncertainty is reserved for non-terminal acknowledgements and confirmation + // timeouts. The unverified targeted path does not reach here: it keeps the + // SDK's own placement evidence (see `spawnInvocationWithMergedPlacement`). + return output.spawned === true && output.ready === true ? 'ready' : 'failed'; } if (status && FAILURE.has(status)) return 'failed'; if (status === 'accepted') return 'accepted'; diff --git a/packages/sdk/src/messaging/placement.test.mts b/packages/sdk/src/messaging/placement.test.mts index 61385eab31..17707462a0 100644 --- a/packages/sdk/src/messaging/placement.test.mts +++ b/packages/sdk/src/messaging/placement.test.mts @@ -128,7 +128,6 @@ describe('RelaycastMessagingClient placement', () => { repo: 'relay', ttl_override_ms: 60, cli: 'claude', - verify_ready: true, }); }); @@ -279,7 +278,6 @@ describe('RelaycastMessagingClient placement', () => { capability: 'spawn:claude', ttl_override_ms: 60, cli: 'claude', - verify_ready: true, }); }); @@ -1268,122 +1266,3 @@ describe('RelaycastMessagingClient placement', () => { }); }); }); - -const LIVE_NODE = { - id: 'node_a', - name: 'node-a', - status: 'online', - live: true, - handlers_live: true, - capabilities: [{ name: 'spawn:claude', kind: 'spawn' }], - repo_keys: ['relay'], -}; - -function contractClient( - getInvocation?: (name: string, invocationId: string) => Promise, - acceptedInvocationId = 'inv-1430' -) { - const invoke = vi.fn(async (name: string, input?: Record) => ({ - invocation_id: acceptedInvocationId, - 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 contractSpawnInput(overrides: Record = {}) { - return { - capability: 'spawn:claude', - node: 'node-a', - repo: 'relay', - input: { name: 'worker-1430' }, - ...overrides, - }; -} - -describe('targeted spawn readiness contract', () => { - it('requests readiness from a healthy remote claude node within the default budget', async () => { - const { client, invoke } = contractClient(async (_name, invocationId) => ({ - invocation_id: invocationId, - status: 'completed', - output: { spawned: true, ready: invoke.mock.calls[0]?.[1]?.verify_ready === true }, - })); - const started = Date.now(); - const ack = await client.placement.spawn(contractSpawnInput({ confirm: true })); - expect(ack.placement.state).toBe('ready'); - expect(ack.placement.confirmed).toBe(true); - expect(Date.now() - started).toBeLessThan(120_000); - expect(invoke).toHaveBeenCalledTimes(1); - }); - - it.each([false, true])('accepts an explicit unverified launch ack with ready=%s', async (ready) => { - const { client, invoke, reader } = contractClient(); - invoke.mockResolvedValueOnce({ - invocation_id: 'launch', - status: 'completed', - output: { spawned: true, ready }, - } as never); - const ack = await client.placement.spawn(contractSpawnInput({ confirm: false })); - expect(ack.placement).toMatchObject({ state: 'accepted', confirmed: false }); - expect(invoke.mock.calls[0]?.[1]).not.toHaveProperty('verify_ready'); - expect(reader).not.toHaveBeenCalled(); - }); - - it('rejects an obsolete handler missing the ready boolean even without confirmation', async () => { - const { client, invoke } = contractClient(); - invoke.mockResolvedValueOnce({ - invocation_id: 'old', - status: 'completed', - output: { spawned: true }, - } as never); - await expect(client.placement.spawn(contractSpawnInput({ confirm: false }))).rejects.toMatchObject({ - code: 'spawn_failed', - message: expect.stringContaining('verify_ready'), - }); - }); - - it('leaves persona verification unchanged', async () => { - const { client, invoke } = contractClient(async () => ({ - status: 'completed', - output: { spawned: true, ready: true }, - })); - // Expose persona capacity only for this fixture. - LIVE_NODE.capabilities.push({ name: 'spawn:persona', kind: 'spawn' }); - try { - await client.placement.spawn(contractSpawnInput({ capability: 'spawn:persona', confirm: true })); - expect(invoke.mock.calls[0]?.[1]).not.toHaveProperty('verify_ready'); - } finally { - LIVE_NODE.capabilities.pop(); - } - }); -}); diff --git a/packages/sdk/src/messaging/relaycast.ts b/packages/sdk/src/messaging/relaycast.ts index 0da0f6c0e6..4b8dfe20dd 100644 --- a/packages/sdk/src/messaging/relaycast.ts +++ b/packages/sdk/src/messaging/relaycast.ts @@ -151,16 +151,38 @@ function hasSpawnLaunchProof(value: { output?: Record | null }) return value.output?.spawned === true && typeof value.output?.ready === 'boolean'; } +/** + * The requester declares the mode and judges only what it asked for. A + * readiness request demands `ready:true`; a launch-only request accepts either + * boolean. Both demand `spawned:true`, so a handler that reports nothing still + * fails. Judging a launch-only request against `ready:true` would reject the + * broker's own honest `ready:false` success — the failure shape relay#1430 + * exists to remove. + */ +function hasSpawnProof( + value: { output?: Record | null }, + requireReadiness: boolean +): boolean { + return requireReadiness ? hasSpawnReadinessProof(value) : hasSpawnLaunchProof(value); +} + function spawnProofError( node: string, status: string, action: string, - output?: Record | null + output: Record | null | undefined, + requireReadiness: boolean ): string { if (output?.spawned === true && typeof output.ready !== 'boolean') { - return `node '${node}' handler did not honour verify_ready: missing explicit spawned:true and ready:true proof; upgrade to a release containing Relay PR #1708`; + // Only a readiness request sent `verify_ready`; on the launch-only path the + // accurate statement is that the handler omitted the `ready` boolean. + return requireReadiness + ? `node '${node}' handler did not honour verify_ready: missing explicit spawned:true and ready:true proof; upgrade to a release containing Relay PR #1708` + : `node '${node}' reported ${status} for ${action} without an explicit ready boolean; upgrade to a release containing Relay PR #1708`; } - return `node '${node}' reported ${status} for ${action} without explicit spawned:true and ready:true proof`; + return `node '${node}' reported ${status} for ${action} without explicit ${ + requireReadiness ? 'spawned:true and ready:true' : 'spawned:true and a ready boolean' + } proof`; } /** Distinguishes "the read outlived its budget" from any value a read returns. */ @@ -746,10 +768,22 @@ export class RelaycastMessagingClient implements RelayMessagingClient { // a node can die between this read and action invocation, after // which the engine treats it as a targeted queued placement. const clientMustTarget = Boolean(targetNode || repo || sandboxOnly); - // Persona resolves a nested child in the engine; its contract is unchanged. + // Asking the broker to verify readiness commits it to holding the + // action open until the harness reports ready and to *releasing the + // worker* if it never does. Only request that when this call will + // wait for the answer — `confirm: true`. Deriving it from + // `confirm !== false` instead would make the documented default + // (`confirm` omitted) request verification and then walk away, + // killing a worker that would otherwise have survived. + // Persona resolves a nested child in the engine, which never reads + // `verify_ready`; its handler proves readiness on its own. const verifyReady = input.verifyReady ?? - (input.confirm !== false && capability.startsWith('spawn:') && capability !== 'spawn:persona'); + (input.confirm === true && capability.startsWith('spawn:') && capability !== 'spawn:persona'); + // What this requester will accept as proof. Persona's engine-owned + // handler reports proven readiness even though it is never sent + // `verify_ready`, so its contract stays readiness-only. + const requireReadiness = verifyReady || capability === 'spawn:persona'; const actionInput = placementActionInput(input.input, { verifyReady, capability, @@ -793,13 +827,11 @@ export class RelaycastMessagingClient implements RelayMessagingClient { capability.startsWith('spawn:') && ackStatus && CONFIRM_SUCCESS_STATUSES.has(ackStatus) && - !(capability === 'spawn:persona' - ? hasSpawnReadinessProof({ output: ackOutput }) - : hasSpawnLaunchProof({ output: ackOutput })) + !hasSpawnProof({ output: ackOutput }, requireReadiness) ) { throw new RelayPlacementError( 'spawn_failed', - spawnProofError(placedNodeLabel, ackStatus, actionName, ackOutput), + spawnProofError(placedNodeLabel, ackStatus, actionName, ackOutput, requireReadiness), { capability, node: placedNodeLabel, @@ -822,6 +854,9 @@ export class RelaycastMessagingClient implements RelayMessagingClient { node: placedNodeLabel, repo, attempts, + // The mode this request actually asked the broker for, so the + // poll judges the same contract the invocation carried. + requireReadiness, // Defaults and validation live in confirmPlacementInvocation // so a direct caller cannot bypass them. timeoutMs: input.confirmTimeoutMs, @@ -838,7 +873,10 @@ export class RelaycastMessagingClient implements RelayMessagingClient { attempts, queued, confirmed: Boolean(confirmation), - state: confirmation ? 'ready' : 'accepted', + // `ready` means proven harness readiness. A confirmation that + // only judged the launch contract (`verifyReady: false`) is + // confirmed, not ready. + state: confirmation && requireReadiness ? 'ready' : 'accepted', }, ...(confirmation ? { confirmation } : {}), }; @@ -950,11 +988,13 @@ export class RelaycastMessagingClient implements RelayMessagingClient { node: string; repo?: string; attempts: number; + /** Whether this request asked the broker for proven harness readiness. */ + requireReadiness: boolean; timeoutMs?: number; pollIntervalMs?: number; } ): Promise { - const { timeoutMs, pollIntervalMs, ...errorContext } = context; + const { timeoutMs, pollIntervalMs, requireReadiness, ...errorContext } = context; const invocationId = ack.invocationId; // Dispatch evidence is fixed at ack time: a node id here means the engine // already routed the invocation to a node before this method starts @@ -1013,10 +1053,13 @@ export class RelaycastMessagingClient implements RelayMessagingClient { const invocation = outcome.value; const status = invocation?.status?.toLowerCase(); if (status && CONFIRM_SUCCESS_STATUSES.has(status)) { - if (errorContext.capability.startsWith('spawn:') && !hasSpawnReadinessProof(invocation)) { + if ( + errorContext.capability.startsWith('spawn:') && + !hasSpawnProof(invocation, requireReadiness) + ) { throw new RelayPlacementError( 'spawn_failed', - spawnProofError(context.node, status, actionName, invocation.output), + spawnProofError(context.node, status, actionName, invocation.output, requireReadiness), { ...errorContext, state: 'failed', diff --git a/packages/sdk/src/messaging/types.ts b/packages/sdk/src/messaging/types.ts index 062c372b52..ac40f60eb5 100644 --- a/packages/sdk/src/messaging/types.ts +++ b/packages/sdk/src/messaging/types.ts @@ -647,22 +647,35 @@ export interface RelaySpawnPlacementInput { * 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. - * Confirmed spawn success requires spawned:true and ready:true. Without - * confirmation, terminal launch success still requires spawned:true and a - * boolean ready field; ready:false means accepted, not verified readiness. + * Spawn success is judged against whichever contract this request asked the + * node for — see `verifyReady`. Either way a terminal success must carry + * `spawned:true` and an explicit `ready` boolean, so a handler that reports + * nothing is still a failure. */ confirm?: boolean; /** - * Request proven harness readiness from the broker. Defaults to confirm !== false - * for spawn harnesses, and false for spawn:persona (engine-owned child input). - * Verified success carries spawned:true, ready:true. Unverified success carries - * spawned:true, ready:false; confirmation still requires ready:true. + * Request proven harness readiness from the broker. Defaults to `true` for + * spawn harnesses when `confirm: true` is set, and `false` otherwise + * (including `spawn:persona`, whose child is engine-owned and never reads + * `verify_ready`). + * + * Verified success carries `spawned:true, ready:true`; unverified success + * carries `spawned:true, ready:false`. Confirmation judges whichever contract + * was requested, so `verifyReady: false` with `confirm: true` accepts a + * launched-but-not-yet-ready worker. + * + * Setting this to `true` without `confirm: true` asks the broker to hold the + * action open until the harness is ready — and to release the worker if it + * never is — while this call returns as soon as the dispatch is accepted. Only + * do that if something else reads the invocation back. */ verifyReady?: 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. + * for up to 90s, and starts that clock only once the launch completes). + * Defaults to 120000; `fleet spawn` enforces a 95000 floor for verified + * targeted spawns so the two windows nest. */ confirmTimeoutMs?: number; /** Poll cadence while awaiting confirmation. Defaults to 500. */ diff --git a/summary.md b/summary.md deleted file mode 100644 index 8de7513e63..0000000000 --- a/summary.md +++ /dev/null @@ -1,18 +0,0 @@ -Targeted fleet spawns now request the broker's verified-readiness contract, so a healthy worker can complete confirmation instead of being rejected for missing `ready:true`. Unverified broker success explicitly returns `ready:false`; `--no-confirm` accepts that launch evidence while continuing to reject obsolete handlers that omit readiness entirely. Persona defaults and MCP readiness requirements remain unchanged. - -The CLI rejects confirmed targeted-spawn timeouts below the broker's 90-second readiness window. Readiness timeout errors explain that the worker was released. Invocation-correlated spawn timing and dropped-result diagnostics help investigate the separate `spawn_unconfirmed` report. The changelog is raised to `[Unreleased - Minor]` as specified by the reviewed plan. - -Validation: - -- Full Vitest suite: 3,496 passed, 24 skipped (`env -u RELAY_BASE_URL npm --ignore-scripts test`, after building packages). Two ambient-URL failures disappeared with that environment override removed. The final additional completed-ack CLI regression passed in the 74-test fleet command suite. -- SDK placement suite: 47 passed, using a temporary Vitest configuration because the root configuration excludes SDK tests. -- Rust suite: 1,302 unit tests and 18 integration tests passed, 5 ignored (`env -u GIT_CONFIG_COUNT cargo test -p agent-relay-broker`). The environment's forced `core.hooksPath=/dev/null` caused four hook-test failures before its removal. -- `npm run typecheck` passed. `npm run lint` passed with 108 existing warnings. -- Real broker integration: `node --test dist/fleet-spawn-readiness.test.js` passed in approximately 1.1 seconds. A loopback engine forwards the SDK invocation to the real broker, which registers exactly one worker, launches a Claude stub in a real PTY, and returns `{spawned:true, ready:true}` through node control and requester polling. -- Mutate-to-red: removing `payload.verify_ready = true` from the built SDK made that same real-broker test fail with `spawn_failed`; restoring it passed. Removing `ready:false` from the Rust unverified result made `spawn_success_always_declares_readiness` fail; restoring it passed. The requester regression also failed against the original source before implementation. -- The broad `npm run test:integration:broker` run encountered continuity/event failures and stalled; it was stopped. The isolated readiness integration above passes. This is not a claim that the full broker integration suite passed. - -Remaining validation and follow-up: - -- Live two-node logged-in Claude, Codex, Gemini, Muse, and Devin checks were not run; the deterministic PTY proof does not establish real-harness login or prompt-detection behavior. -- The separate 120-second result silence remains open, per reviewed-plan.md. Investigate wire loss, slow spawn, and invocation-ID mismatch using the new logs. Action-result retention/replay is deferred until engine-side replay idempotency is established; no reconnect queue is included here. diff --git a/tests/integration/broker/fleet-spawn-readiness.test.ts b/tests/integration/broker/fleet-spawn-readiness.test.ts index 60a838f94a..e8d15e052c 100644 --- a/tests/integration/broker/fleet-spawn-readiness.test.ts +++ b/tests/integration/broker/fleet-spawn-readiness.test.ts @@ -5,16 +5,24 @@ import { mkdtemp, mkdir, rm, writeFile } from 'node:fs/promises'; import http from 'node:http'; import { tmpdir } from 'node:os'; import path from 'node:path'; -import { test } from 'node:test'; +import { test, type TestContext } from 'node:test'; import { RelaycastMessagingClient } from '@agent-relay/sdk'; import { WebSocketServer, type WebSocket } from 'ws'; +import { checkPrerequisites, resolveBinaryPath } from './utils/broker-harness.js'; + test( 'targeted Claude spawn confirms real broker readiness inside the default budget', { timeout: 130_000 }, - async () => { - const binary = - process.env.AGENT_RELAY_BROKER_BINARY ?? path.resolve('../../../target/debug/agent-relay-broker'); + async (t: TestContext) => { + // Skip rather than fail when the broker has not been built, matching every + // sibling suite here. + const missing = checkPrerequisites(); + if (missing) { + t.skip(missing); + return; + } + const binary = resolveBinaryPath(); const directory = await mkdtemp(path.join(tmpdir(), 'fleet-ready-')); const bin = path.join(directory, 'bin'); await mkdir(bin); @@ -205,7 +213,6 @@ test( }, } as never, }); - const started = Date.now(); const result = await client.placement.spawn({ capability: 'spawn:claude', node: node.name, @@ -213,7 +220,6 @@ test( input: { name: 'Probe' }, }); assert.equal(result.placement.state, 'ready', logs); - assert(Date.now() - started < 120_000); assert.equal(invocations, 1); assert.equal(registrations, 1); assert.deepEqual(result.confirmation?.output, { spawned: true, ready: true, name: 'Probe' });