diff --git a/CHANGELOG.md b/CHANGELOG.md index c86799ccb..38f94fefc 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -60,6 +60,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - The Bun-compiled `agent-relay` standalone binary now bundles workspace packages from their compiled JS instead of their `.d.ts`, so `node up` starts the implicit Fleet local node instead of failing with `Fleet local node skipped: … is not a function`. The `tsconfig` `paths` that mapped `@agent-relay/*` to declaration files (no runtime exports) were redundant with the npm workspace symlinks and have been removed. - `agent-relay` and `@agent-relay/sdk` require `@relaycast/sdk` `^4.1.2`, whose matching `@relaycast/types` package is now published, so publish installs resolve cleanly without pinning. - `agent-relay node up --config ` loads plain JavaScript node definitions without `jiti`, so the published Bun-compiled CLI can serve compiled JS node files. +- `agent-relay node up` reports `Broker started.` as soon as the workspace handshake completes: the broker no longer blocks its `/api/session` readiness on minting the node token (a Relaycast `create_node` round-trip). The node-control client mints the token in the background and publishes it to the session, so a slow node-token mint on a slow network no longer delays or fails startup. Serving a capability definition without an explicit `RELAY_NODE_TOKEN` waits briefly for the background-minted token instead of skipping the provider. - Spawned opencode worker agents no longer pause for interactive tool-approval prompts; the broker injects a wildcard allow-all permission block into every generated `opencode.json`, augmenting existing partial permission objects rather than replacing them. ### Added diff --git a/crates/broker/src/node_control.rs b/crates/broker/src/node_control.rs index f778a73c7..cc72a1502 100644 --- a/crates/broker/src/node_control.rs +++ b/crates/broker/src/node_control.rs @@ -61,10 +61,11 @@ pub(crate) struct FleetControlConfig { pub(crate) session_token: Option>>>, } -/// Re-mints a fresh node token via `POST /v1/nodes` and rewrites the -/// workspace-scoped cache, used to recover from a stale/rejected cached token -/// (HTTP 401 on the node-control handshake) instead of looping forever on the -/// same token. Mirrors the initial mint wired in `runtime::init::resolve_node_token`. +/// Mints node tokens via `POST /v1/nodes` and maintains the workspace-scoped +/// cache. Held by the node-control client, which uses it both for the initial +/// mint (when no token is cached — see [`NodeTokenMinter::mint`]) and to recover +/// from a stale/rejected cached token (HTTP 401 on the node-control handshake — +/// see [`NodeTokenMinter::remint`]) instead of looping forever on the same token. #[derive(Clone)] pub(crate) struct NodeTokenMinter { pub(crate) workspace_key: String, @@ -79,24 +80,11 @@ pub(crate) struct NodeTokenMinter { } impl NodeTokenMinter { - /// Discard the cached token for this workspace and mint a fresh one. Returns - /// the new token on success. On failure the caller surfaces a loud error and - /// backs off rather than looping on the rejected token. - async fn remint(&self) -> Option { - // Drop the rejected cache eagerly so a crash mid-mint doesn't leave the - // stale token behind for the next start. - if let Some(path) = self.token_path.as_deref() { - if let Err(error) = fs::remove_file(path) { - if error.kind() != std::io::ErrorKind::NotFound { - tracing::warn!( - target = "relay_broker::fleet", - node_id = %self.node_id, - error = %error, - "failed to clear rejected node token cache before re-mint" - ); - } - } - } + /// Mint a fresh node token via `POST /v1/nodes` and persist it to the + /// workspace-scoped cache. Returns the new token on success, or `None` after + /// logging the failure so the caller can back off and retry. Used for the + /// initial mint (no cached token) and as the shared body of [`Self::remint`]. + async fn mint(&self) -> Option { let request = create_node_request(&self.node_id, &self.node_name, &self.broker_version); match mint_node_token( &self.workspace_key, @@ -122,7 +110,7 @@ impl NodeTokenMinter { target = "relay_broker::fleet", node_id = %self.node_id, error = %error, - "failed to persist re-minted node token" + "failed to persist minted node token" ); } } @@ -130,7 +118,7 @@ impl NodeTokenMinter { target = "relay_broker::fleet", node_id = %self.node_id, workspace_id = %self.workspace_id, - "re-minted node token after node-control 401" + "minted node token via create_node" ); Some(token) } @@ -140,12 +128,33 @@ impl NodeTokenMinter { &self.node_id, &self.workspace_id, &error, - "failed to re-mint node token after node-control 401", + "failed to mint node token via create_node", ); None } } } + + /// Discard the cached token for this workspace and mint a fresh one. Returns + /// the new token on success. On failure the caller surfaces a loud error and + /// backs off rather than looping on the rejected token. + async fn remint(&self) -> Option { + // Drop the rejected cache eagerly so a crash mid-mint doesn't leave the + // stale token behind for the next start. + if let Some(path) = self.token_path.as_deref() { + if let Err(error) = fs::remove_file(path) { + if error.kind() != std::io::ErrorKind::NotFound { + tracing::warn!( + target = "relay_broker::fleet", + node_id = %self.node_id, + error = %error, + "failed to clear rejected node token cache before re-mint" + ); + } + } + } + self.mint().await + } } pub(crate) fn create_node_request( @@ -905,6 +914,53 @@ pub(crate) fn persist_node_token( Ok(()) } +/// Outcome of handling a control command received while the node is not yet +/// connected to `/v1/node/ws`. `Shutdown` means the command channel closed or a +/// `Shutdown` command arrived and the caller should return. +enum DisconnectedCommandOutcome { + Handled, + Shutdown, +} + +/// Apply a control command received while the node is disconnected, shared by the +/// three not-yet-connected wait points (pre-registration, mint backoff, and the +/// no-minter idle wait) so a new `FleetControlCommand` variant or state update +/// stays consistent across them. `register_agent_error` is the reason replied to +/// a `RegisterAgent` that can't be served yet: `node_not_registered` before the +/// node is registered, `node_token_missing` once registered but tokenless. +fn handle_disconnected_command( + command: Option, + config: &FleetControlConfig, + registration: &mut Option, + load: &mut FleetLoadSnapshot, + inventory: &mut Vec, + register_agent_error: &str, +) -> DisconnectedCommandOutcome { + match command { + Some(FleetControlCommand::RegisterNode { + manifest, + resume_cursor, + }) => { + load.max_agents = manifest.max_agents.unwrap_or(load.max_agents); + *registration = Some(build_node_register( + &manifest, + &config.node_id, + &config.node_name, + &config.broker_version, + resume_cursor, + )); + } + Some(FleetControlCommand::UpdateLoad(next)) => *load = next, + Some(FleetControlCommand::UpdateInventory(next)) => *inventory = next, + Some(FleetControlCommand::RegisterAgent { reply, .. }) => { + let _ = reply.send(Err(register_agent_error.to_string())); + } + Some(FleetControlCommand::Send(_)) | Some(FleetControlCommand::HeartbeatNow) => {} + Some(FleetControlCommand::Shutdown) | None => return DisconnectedCommandOutcome::Shutdown, + } + DisconnectedCommandOutcome::Handled +} + pub(crate) async fn run_node_control_client( mut config: FleetControlConfig, mut command_rx: mpsc::Receiver, @@ -924,27 +980,18 @@ pub(crate) async fn run_node_control_client( loop { while registration.is_none() { - match command_rx.recv().await { - Some(FleetControlCommand::RegisterNode { - manifest, - resume_cursor, - }) => { - load.max_agents = manifest.max_agents.unwrap_or(load.max_agents); - registration = Some(build_node_register( - &manifest, - &config.node_id, - &config.node_name, - &config.broker_version, - resume_cursor, - )); - } - Some(FleetControlCommand::UpdateLoad(next)) => load = next, - Some(FleetControlCommand::UpdateInventory(next)) => inventory = next, - Some(FleetControlCommand::Shutdown) | None => return, - Some(FleetControlCommand::RegisterAgent { reply, .. }) => { - let _ = reply.send(Err("node_not_registered".to_string())); - } - Some(FleetControlCommand::Send(_)) | Some(FleetControlCommand::HeartbeatNow) => {} + if matches!( + handle_disconnected_command( + command_rx.recv().await, + &config, + &mut registration, + &mut load, + &mut inventory, + "node_not_registered", + ), + DisconnectedCommandOutcome::Shutdown + ) { + return; } } @@ -955,29 +1002,81 @@ pub(crate) async fn run_node_control_client( .unwrap_or("") .is_empty() { - match command_rx.recv().await { - Some(FleetControlCommand::RegisterNode { - manifest, - resume_cursor, - }) => { - load.max_agents = manifest.max_agents.unwrap_or(load.max_agents); - registration = Some(build_node_register( - &manifest, - &config.node_id, - &config.node_name, - &config.broker_version, - resume_cursor, - )); + // No pre-supplied or cached token. Broker startup deliberately skips + // the network mint (it must not gate `/api/session` readiness), so the + // initial mint happens here, in the background. On success, publish the + // token to the shared HTTP session so `/api/session` starts reporting + // it, then fall through to connect. On failure, back off and retry + // rather than idling forever — realtime delivery self-heals once the + // engine is reachable. + if let Some(minter) = config.token_minter.as_ref() { + if let Some(fresh) = minter.mint().await { + config.node_token = Some(fresh); + if let Some(shared) = &config.session_token { + if let Ok(mut guard) = shared.write() { + guard.clone_from(&config.node_token); + } + } + // A successful mint proves the engine is reachable, so reset + // the backoff any earlier mint failures grew — the first + // `/v1/node/ws` connect should start from the minimum delay, + // not inherit a bloated one. + reconnect_delay = INITIAL_RECONNECT_DELAY; + } else { + tracing::warn!( + target = "relay_broker::fleet", + node_id = %config.node_id, + "node token mint failed; retrying after backoff (realtime delivery pending)" + ); + // Stay responsive during the backoff instead of a blind + // sleep: a spawn's `RegisterAgent` must get an immediate + // `node_token_missing` (so the caller falls back to HTTP + // register) rather than blocking on the 30s register timeout, + // and load/inventory updates must keep draining so the bounded + // control channel can't fill during a Relaycast outage. + let backoff = tokio::time::sleep(reconnect_delay); + tokio::pin!(backoff); + loop { + tokio::select! { + _ = &mut backoff => break, + command = command_rx.recv() => { + if matches!( + handle_disconnected_command( + command, + &config, + &mut registration, + &mut load, + &mut inventory, + "node_token_missing", + ), + DisconnectedCommandOutcome::Shutdown + ) { + return; + } + } + } + } + reconnect_delay = (reconnect_delay * 2).min(MAX_RECONNECT_DELAY); + continue; } - Some(FleetControlCommand::UpdateLoad(next)) => load = next, - Some(FleetControlCommand::UpdateInventory(next)) => inventory = next, - Some(FleetControlCommand::Shutdown) | None => return, - Some(FleetControlCommand::RegisterAgent { reply, .. }) => { - let _ = reply.send(Err("node_token_missing".to_string())); + } else { + // No minter available (e.g. no workspace RelayCast client). Can't + // self-recover; wait for a token to arrive via command. + if matches!( + handle_disconnected_command( + command_rx.recv().await, + &config, + &mut registration, + &mut load, + &mut inventory, + "node_token_missing", + ), + DisconnectedCommandOutcome::Shutdown + ) { + return; } - Some(FleetControlCommand::Send(_)) | Some(FleetControlCommand::HeartbeatNow) => {} + continue; } - continue; } let result = run_connected_once( @@ -2262,6 +2361,102 @@ mod tests { let _ = command_tx.send(FleetControlCommand::Shutdown).await; } + #[tokio::test] + async fn node_control_client_mints_initial_token_when_none_supplied() { + // Broker startup no longer mints the node token on the API-readiness + // path; the client mints it in the background. Started with no token but + // a minter, it must mint via create_node, publish the token to the shared + // HTTP session handle, and connect. + let mint_server = MockServer::start(); + let create_node = mint_server.mock(|when, then| { + when.method(POST).path("/v1/nodes"); + then.status(200).json_body(json!({ + "ok": true, + "data": { + "id": "node-test", + "name": "host-test", + "kind": "ws", + "role": "broker", + "version": "broker/test", + "status": "online", + "live": true, + "handlers_live": true, + "load": 0.0, + "active_agents": 0, + "max_agents": 0, + "created_at": "2026-06-30T00:00:00Z", + "token": "nt_minted_by_client" + } + })); + }); + + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let ws_url = format!("ws://{}/v1/node/ws", listener.local_addr().unwrap()); + let (command_tx, command_rx) = mpsc::channel(32); + let (event_tx, mut event_rx) = mpsc::channel(32); + let session_token = Arc::new(std::sync::RwLock::new(None)); + + tokio::spawn(run_node_control_client( + FleetControlConfig { + ws_url, + node_token: None, + node_id: "node-test".to_string(), + node_name: "host-test".to_string(), + broker_version: "broker/test".to_string(), + token_minter: Some(NodeTokenMinter { + workspace_key: "rk_live_test".to_string(), + workspace_id: "ws_test".to_string(), + base_url: Some(mint_server.base_url()), + node_id: "node-test".to_string(), + node_name: "host-test".to_string(), + broker_version: "broker/test".to_string(), + token_path: None, + }), + session_token: Some(session_token.clone()), + }, + command_rx, + event_tx, + )); + + let server = tokio::spawn(async move { + let (stream, _) = listener.accept().await.unwrap(); + let mut ws = accept_async(stream).await.unwrap(); + assert!(matches!( + next_node_to_server(&mut ws).await, + BrokerToRelaycast::NodeRegister(_) + )); + }); + + command_tx + .send(FleetControlCommand::RegisterNode { + manifest: test_manifest(), + resume_cursor: None, + }) + .await + .unwrap(); + + // Bounded: if the mint or connect path regresses, the client retries + // without ever emitting `Connected`, so an unbounded recv would hang the + // whole suite. Fail with an assertion instead. + let connected = tokio::time::timeout(Duration::from_secs(5), event_rx.recv()) + .await + .expect("node-control client should emit Connected within 5s") + .unwrap(); + assert_eq!(connected, FleetControlEvent::Connected); + tokio::time::timeout(Duration::from_secs(5), server) + .await + .unwrap() + .unwrap(); + + create_node.assert_hits(1); + assert_eq!( + session_token.read().unwrap().as_deref(), + Some("nt_minted_by_client"), + "the background mint must publish the token to the shared HTTP session" + ); + let _ = command_tx.send(FleetControlCommand::Shutdown).await; + } + #[tokio::test] async fn node_control_agent_register_timeout_late_success_deregisters() { let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); diff --git a/crates/broker/src/runtime/init.rs b/crates/broker/src/runtime/init.rs index 0f7c4377e..e8ce6c5e6 100644 --- a/crates/broker/src/runtime/init.rs +++ b/crates/broker/src/runtime/init.rs @@ -266,34 +266,16 @@ pub(crate) async fn run_init(cmd: InitCommand, telemetry: TelemetryClient) -> Re // cached token is only reused when both match, and so a re-mint after a // node-control 401 rewrites the correctly-scoped cache. let node_base_url = configured_base.clone(); - let node_token = resolve_node_token( - &node_id, - &node_name, - &broker_version, - &node_workspace_id, - &relay_workspace_key, - node_base_url.as_deref(), - ) - .await; - if node_token.is_none() { - // Node-only delivery requires this broker to be a functioning relaycast - // node: without a node token it cannot open /v1/node/ws, so the engine - // delivers nothing and every spawned agent is effectively unreachable. - // This is a hard operational fault, not a benign warning. We do NOT exit - // (a token may arrive via env/mint later), but make the failure mode - // unmistakable in logs and on stderr. - tracing::error!( - node_id = %node_id, - "NO NODE TOKEN: this broker is NOT a functioning relaycast node (env unset, no cached token, mint failed). \ - /v1/node/ws will not connect and realtime delivery will FAIL for every agent until a node token is available \ - (set RELAY_NODE_TOKEN or restore connectivity so a token can be minted)." - ); - eprintln!( - "[agent-relay] FATAL CONFIG: no node token available for node '{}'. \ - Realtime delivery is DISABLED until RELAY_NODE_TOKEN is set or a token can be minted.", - node_id - ); - } + // Resolve only the fast, local token sources here (RELAY_NODE_TOKEN override + // and the on-disk cache). The network mint (create_node) is deliberately NOT + // done on this path: it would block the broker's API readiness handoff below + // behind a Relaycast round-trip, delaying `/api/session` (and the CLI's + // "Broker started.") until the mint completes. When no token is cached, the + // node-control client mints one in the background (it holds the same minter) + // and publishes it to `session_node_token`, so realtime delivery still comes + // online without gating startup on it. + let node_token = + resolve_cached_node_token(&node_id, &node_workspace_id, node_base_url.as_deref()); let node_manifest = bootstrap_node_manifest(&node_name, &node_id, &broker_version); // Retain the node name for the runtime: the HTTP `bind_agent_to_node` // fallback (used when node-control `agent.register` is unavailable) binds @@ -309,14 +291,16 @@ pub(crate) async fn run_init(cmd: InitCommand, telemetry: TelemetryClient) -> Re // the broker's resolved token on the api-key-gated session so the CLI can // serve local config providers without a pre-enrolled RELAY_NODE_TOKEN (the // broker mints its own in that case). Alongside the workspace key already - // returned there, this stays within the local trust boundary. A shared handle - // (updated by the node-control re-mint path) keeps the session token current. + // returned there, this stays within the local trust boundary. Seeded with the + // cached token (if any); the node-control client writes through this shared + // handle when it mints (initially or after a re-mint), keeping it current. let session_node_token = std::sync::Arc::new(std::sync::RwLock::new(node_token.clone())); - // Wire a re-mint facility so a node-control 401 (stale/wrong-scoped token) - // discards the cached token and mints a fresh one, instead of looping - // forever on the rejected token. Mirrors the initial mint above. Absent when - // no workspace RelayCast client is available (then a 401 surfaces a hard - // error rather than recovering). + // Wire the token minter used by the node-control client both for the initial + // mint (when no token is cached, off the readiness path) and to recover from + // a node-control 401 (stale/wrong-scoped token) by discarding the cached + // token and minting a fresh one instead of looping forever on the rejected + // token. Absent when no workspace RelayCast client is available (then a 401 + // surfaces a hard error rather than recovering). let token_minter = Some(crate::node_control::NodeTokenMinter { workspace_key: relay_workspace_key.clone(), workspace_id: node_workspace_id.clone(), @@ -670,23 +654,19 @@ pub(crate) async fn run_init(cmd: InitCommand, telemetry: TelemetryClient) -> Re runtime.run().await } -/// Resolve the node token used to authenticate the `/v1/node/ws` connection. +/// Resolve the node token used to authenticate the `/v1/node/ws` connection +/// from the fast, local sources only, in precedence order: /// -/// Precedence: /// 1. `RELAY_NODE_TOKEN` env override (operator-supplied; never persisted). /// 2. A token previously minted for this exact `node_id` and cached on disk. -/// 3. A freshly minted token via `RelayCast::create_node` (workspace key), -/// persisted next to the node id for reuse on the next start. /// -/// Returns `None` only when no override or cache exists and minting is -/// impossible (no relay client) or fails; the caller logs and continues without -/// node delivery. -async fn resolve_node_token( +/// Returns `None` when neither exists. This never performs a network mint — that +/// stays off the broker's API-readiness path (see the call site) and is handled +/// in the background by the node-control client, which holds the same +/// [`crate::node_control::NodeTokenMinter`]. +fn resolve_cached_node_token( node_id: &str, - node_name: &str, - broker_version: &str, workspace_id: &str, - workspace_key: &str, base_url: Option<&str>, ) -> Option { if let Some(token) = std::env::var("RELAY_NODE_TOKEN") @@ -705,44 +685,7 @@ async fn resolve_node_token( return Some(token); } - let request = crate::node_control::create_node_request(node_id, node_name, broker_version); - match crate::node_control::mint_node_token( - workspace_key, - base_url, - request, - crate::node_control::MintNodeTokenLogContext { - node_id, - workspace_id, - }, - ) - .await - { - Ok(token) => { - if let Some(path) = token_path.as_deref() { - if let Err(error) = crate::node_control::persist_node_token( - path, - node_id, - workspace_id, - base_url, - &token, - ) { - tracing::warn!(node_id = %node_id, error = %error, "failed to persist minted node token"); - } - } - tracing::info!(node_id = %node_id, workspace_id = %workspace_id, "minted node token via create_node"); - Some(token) - } - Err(error) => { - crate::node_control::log_create_node_mint_error( - "relay_broker::fleet", - node_id, - workspace_id, - &error, - "failed to mint node token via create_node", - ); - None - } - } + None } fn callback_host_for_url(api_bind: &str, local_addr: SocketAddr) -> String { diff --git a/packages/cli/src/cli/lib/broker-lifecycle.test.ts b/packages/cli/src/cli/lib/broker-lifecycle.test.ts index 7c50da976..59ff12bc2 100644 --- a/packages/cli/src/cli/lib/broker-lifecycle.test.ts +++ b/packages/cli/src/cli/lib/broker-lifecycle.test.ts @@ -5,6 +5,7 @@ import { classifyBrokerStartStage, describeError, readNodeDeliveryStatus, + resolveNodeIdentityFromSession, waitForNodeDelivery, } from './broker-lifecycle.js'; @@ -366,3 +367,117 @@ describe('runUpCommand node-config gating', () => { expect(output).not.toMatch(/rk_live_|nt_live_/); }); }); + +describe('resolveNodeIdentityFromSession', () => { + const noSleep = vi.fn(async () => {}); + + it('returns identity immediately when the token is already present', async () => { + const getSession = vi.fn(async () => ({ + node_id: 'node-1', + node_name: 'host-1', + node_token: 'nt_live_ready', + })); + + const identity = await resolveNodeIdentityFromSession(getSession, { + awaitTokenMs: 15_000, + sleep: noSleep, + }); + + expect(identity).toEqual({ nodeId: 'node-1', nodeName: 'host-1', nodeToken: 'nt_live_ready' }); + expect(getSession).toHaveBeenCalledTimes(1); + expect(noSleep).not.toHaveBeenCalled(); + }); + + it('polls until the background-minted token appears', async () => { + const sessions = [ + { node_id: 'node-1', node_name: 'host-1' }, + { node_id: 'node-1', node_name: 'host-1' }, + { node_id: 'node-1', node_name: 'host-1', node_token: 'nt_live_late' }, + ]; + let call = 0; + const getSession = vi.fn(async () => sessions[Math.min(call++, sessions.length - 1)]); + const sleep = vi.fn(async () => {}); + + const identity = await resolveNodeIdentityFromSession(getSession, { + awaitTokenMs: 15_000, + sleep, + }); + + expect(identity).toEqual({ nodeId: 'node-1', nodeName: 'host-1', nodeToken: 'nt_live_late' }); + expect(getSession).toHaveBeenCalledTimes(3); + expect(sleep).toHaveBeenCalledTimes(2); + }); + + it('does not poll when awaitTokenMs is zero (explicit RELAY_NODE_TOKEN path)', async () => { + const getSession = vi.fn(async () => ({ node_id: 'node-1', node_name: 'host-1' })); + const sleep = vi.fn(async () => {}); + + const identity = await resolveNodeIdentityFromSession(getSession, { awaitTokenMs: 0, sleep }); + + expect(identity).toEqual({ nodeId: 'node-1', nodeName: 'host-1' }); + expect(getSession).toHaveBeenCalledTimes(1); + expect(sleep).not.toHaveBeenCalled(); + }); + + it('returns the best identity seen so far when the deadline elapses without a token', async () => { + let now = 1_000; + const nowSpy = vi.spyOn(Date, 'now').mockImplementation(() => now); + const getSession = vi.fn(async () => ({ node_id: 'node-1', node_name: 'host-1' })); + const sleep = vi.fn(async () => { + now += 200; // advance past the awaitTokenMs budget after the first sleep + }); + + const identity = await resolveNodeIdentityFromSession(getSession, { + awaitTokenMs: 150, + sleep, + }); + + expect(identity).toEqual({ nodeId: 'node-1', nodeName: 'host-1' }); + expect(getSession).toHaveBeenCalledTimes(2); + nowSpy.mockRestore(); + }); + + it('returns null when the broker never reports a node id', async () => { + const getSession = vi.fn(async () => ({})); + + const identity = await resolveNodeIdentityFromSession(getSession, { + awaitTokenMs: 15_000, + sleep: noSleep, + }); + + expect(identity).toBeNull(); + }); + + it('yields the last good identity when a later session read throws', async () => { + let call = 0; + const getSession = vi.fn(async () => { + if (call++ === 0) return { node_id: 'node-1', node_name: 'host-1' }; + throw new Error('connection reset'); + }); + const sleep = vi.fn(async () => {}); + + const identity = await resolveNodeIdentityFromSession(getSession, { + awaitTokenMs: 15_000, + sleep, + }); + + expect(identity).toEqual({ nodeId: 'node-1', nodeName: 'host-1' }); + }); + + it('bounds a stalled session read to the token-wait budget instead of hanging', async () => { + // getSession never resolves; without the per-read bound this would hang past + // the transport's 30s timeout. With a 60ms budget it must return ~promptly. + const getSession = vi.fn(() => new Promise<{ node_id?: string }>(() => {})); + const sleep = vi.fn(async () => {}); + + const start = Date.now(); + const identity = await resolveNodeIdentityFromSession(getSession, { + awaitTokenMs: 60, + sleep, + }); + + expect(identity).toBeNull(); + expect(Date.now() - start).toBeLessThan(1_000); + expect(getSession).toHaveBeenCalledTimes(1); + }); +}); diff --git a/packages/cli/src/cli/lib/broker-lifecycle.ts b/packages/cli/src/cli/lib/broker-lifecycle.ts index 37923ad7e..4d2bc4165 100644 --- a/packages/cli/src/cli/lib/broker-lifecycle.ts +++ b/packages/cli/src/cli/lib/broker-lifecycle.ts @@ -50,6 +50,10 @@ const CONNECTION_FILENAME = 'connection.json'; const STATUS_POLL_INTERVAL_MS = 500; const DETACHED_START_READY_TIMEOUT_MS = 10_000; const NODE_DELIVERY_READY_TIMEOUT_MS = 10_000; +// Bounded wait for the broker's background-minted node token to surface on +// `/api/session` when serving a capability definition without an explicit +// RELAY_NODE_TOKEN. +const NODE_TOKEN_WAIT_MS = 15_000; export interface BrokerConnection { url: string; @@ -305,24 +309,92 @@ export interface RunningNodeProviders { stop(): Promise; } +export interface BrokerNodeIdentity { + nodeId: string; + nodeName: string; + nodeToken?: string; +} + +interface SessionSnapshot { + node_id?: string; + node_name?: string; + node_token?: string; +} + +/** Reject if `promise` doesn't settle within `ms`, clearing the timer on settle. */ +function withTimeout(promise: Promise, ms: number): Promise { + return new Promise((resolve, reject) => { + const timer = setTimeout(() => reject(new Error('session read exceeded token-wait budget')), ms); + promise.then( + (value) => { + clearTimeout(timer); + resolve(value); + }, + (err) => { + clearTimeout(timer); + reject(err); + } + ); + }); +} + +/** + * Resolve the broker's node identity from repeated `/api/session` reads. + * + * The broker publishes its node id as soon as the workspace handshake completes, + * but mints the node token off the API-readiness path (in the background), so a + * freshly started broker can report `node_id` before `node_token`. When + * `awaitTokenMs` is set, poll until the token appears (or the budget elapses) so + * a provider that needs the broker-minted token isn't skipped over a startup + * race. A transient session-read error yields the best identity seen so far + * (identity without token), or `null` if no `node_id` was ever read. + */ +export async function resolveNodeIdentityFromSession( + getSession: () => Promise, + options: { awaitTokenMs?: number; sleep?: (ms: number) => Promise } = {} +): Promise { + const awaitTokenMs = options.awaitTokenMs ?? 0; + const sleep = options.sleep ?? ((ms: number) => new Promise((resolve) => setTimeout(resolve, ms))); + const deadline = Date.now() + awaitTokenMs; + let identity: BrokerNodeIdentity | null = null; + for (;;) { + let session: SessionSnapshot; + try { + // Bound each read to the remaining token-wait budget so a single stalled + // `/api/session` can't hold startup past `awaitTokenMs` (the transport's + // own request timeout is far longer). The non-await path issues one read + // and doesn't need the bound. + session = + awaitTokenMs > 0 + ? await withTimeout(getSession(), Math.max(0, deadline - Date.now())) + : await getSession(); + } catch { + return identity; + } + if (!session.node_id) return identity; + identity = { + nodeId: session.node_id, + nodeName: session.node_name ?? session.node_id, + ...(session.node_token ? { nodeToken: session.node_token } : {}), + }; + if (session.node_token || awaitTokenMs <= 0 || Date.now() >= deadline) { + return identity; + } + await sleep(250); + } +} + /** * Read the node id/name the broker registered as, from its HTTP session. The * capability providers attach to this same node so they share its identity. */ async function readBrokerNodeIdentity( - conn: BrokerConnection -): Promise<{ nodeId: string; nodeName: string; nodeToken?: string } | null> { + conn: BrokerConnection, + options: { awaitTokenMs?: number; sleep?: (ms: number) => Promise } = {} +): Promise { const client = new HarnessDriverClient({ baseUrl: conn.url, apiKey: conn.api_key }); try { - const session = await client.getSession(); - if (!session.node_id) return null; - return { - nodeId: session.node_id, - nodeName: session.node_name ?? session.node_id, - ...(session.node_token ? { nodeToken: session.node_token } : {}), - }; - } catch { - return null; + return await resolveNodeIdentityFromSession(() => client.getSession(), options); } finally { client.disconnect(); } @@ -356,7 +428,12 @@ async function startNodeCapabilityProviders( return undefined; } const baseUrl = deps.env.RELAY_BASE_URL?.trim(); - const identity = await readBrokerNodeIdentity(conn); + // Serving a definition needs the node token. When no explicit RELAY_NODE_TOKEN + // is set we rely on the broker's background-minted token, which can lag its + // node id by a Relaycast round-trip — wait a bounded window for it rather than + // racing the mint and skipping the provider. + const awaitTokenMs = deps.env.RELAY_NODE_TOKEN?.trim() ? 0 : NODE_TOKEN_WAIT_MS; + const identity = await readBrokerNodeIdentity(conn, { awaitTokenMs }); if (!identity) { deps.warn('Capability providers skipped: the broker did not report its node id yet.'); return undefined; @@ -1146,7 +1223,14 @@ export async function runUpCommand(options: UpOptions, deps: CoreDependencies): if (shouldSpawn && teamsConfig && teamsConfig.agents.length > 0) { vlog(deps, options.verbose, 'Waiting for broker node delivery (/v1/node/ws) before auto-spawning...'); - const delivery = await waitForNodeDelivery(relay, deps); + // Node delivery can't connect until the broker mints its node token, which + // now happens in the background after `Broker started.`. Budget for that + // mint window plus the connect so a slow mint doesn't abort auto-spawn. + const delivery = await waitForNodeDelivery( + relay, + deps, + NODE_TOKEN_WAIT_MS + NODE_DELIVERY_READY_TIMEOUT_MS + ); if (!delivery.ready) { deps.error('Refusing to auto-spawn agents because broker node delivery is not connected.'); deps.error(`Node delivery: ${formatNodeDeliveryStatus(delivery.status)}`);