From 222fc332fb93640d46626ea8f366864822f62d95 Mon Sep 17 00:00:00 2001 From: khaliqgant Date: Wed, 7 Oct 2026 09:39:34 -0700 Subject: [PATCH 1/6] fix(broker): record a spawned worker's cli, host and owner Fleet workers registered through a node carried no owner or machine, so Agent Relay Desktop listed them under "Other live agents" even on the person's own machine, and a worker named without a harness prefix got no icon. The declared-metadata PATCH now also carries `cli`, `host` (the hostname without `.local`) and `owner_hash` (SHA-256 of the Cloud user id), the keys the desktop records for its own sessions. The user id comes from AGENT_RELAY_USER_ID or, for a service-started broker, the CLI's cloud-identity.json. Co-Authored-By: Claude Opus 5.5 (1M context) --- CHANGELOG.md | 4 + crates/broker/src/relaycast/mod.rs | 4 +- crates/broker/src/relaycast/ws.rs | 161 +++++++++++++++++- crates/broker/src/runtime/api.rs | 1 + crates/broker/src/runtime/fleet.rs | 17 +- crates/broker/src/runtime/relaycast_events.rs | 2 + crates/broker/src/runtime/tests.rs | 14 +- 7 files changed, 192 insertions(+), 11 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 6056d36972..0f3a25ceef 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Fixed + +- Workers a fleet broker spawns now record their CLI (`cli`), the machine's name (`host`) and the signed-in Cloud person who runs the broker (`owner_hash`) in their Relaycast metadata. These are the keys Agent Relay Desktop records for the sessions it registers, so the desktop lists your fleet workers as your agents on that machine, with their harness icon, instead of under "Other live agents". The broker reads the person from `AGENT_RELAY_USER_ID` or, for a broker started by a service manager, from the CLI's stored `cloud-identity.json`. + ## [13.1.3] - 2026-10-07 ### Fixed diff --git a/crates/broker/src/relaycast/mod.rs b/crates/broker/src/relaycast/mod.rs index 38b6c2a16b..9b8494a879 100644 --- a/crates/broker/src/relaycast/mod.rs +++ b/crates/broker/src/relaycast/mod.rs @@ -27,6 +27,6 @@ pub(crate) use workspace::{ pub(crate) use ws::{ format_worker_preregistration_error, register_new_spawn_identity, registration_retry_after_secs, retry_agent_registration, retry_agent_registration_create_only, - RegRetryOutcome, RelaycastHttpClient, RelaycastRegistrationError, WsControl, - MAX_AGENT_REGISTRATION_ELAPSED, MAX_AGENT_REGISTRATION_OUTER_TIMEOUT, + spawned_worker_metadata, RegRetryOutcome, RelaycastHttpClient, RelaycastRegistrationError, + WsControl, MAX_AGENT_REGISTRATION_ELAPSED, MAX_AGENT_REGISTRATION_OUTER_TIMEOUT, }; diff --git a/crates/broker/src/relaycast/ws.rs b/crates/broker/src/relaycast/ws.rs index 6d51b1856e..6c7a5e6f1a 100644 --- a/crates/broker/src/relaycast/ws.rs +++ b/crates/broker/src/relaycast/ws.rs @@ -920,16 +920,22 @@ impl RelaycastHttpClient { /// /// Callers treat this as best-effort: the agent is registered and running /// either way, so a failure here must be logged, not fatal. + /// + /// `spawned` is what this broker knows about the worker it spawned (see + /// [`spawned_worker_metadata`]); it rides the same PATCH so a spawn costs + /// one request either way. pub async fn publish_declared_metadata( &self, agent_name: &str, declared: &AgentRegistrationMetadata, + spawned: &serde_json::Map, ) -> std::result::Result<(), RelaycastRegistrationError> { let name = agent_name.trim(); if name.is_empty() { return Err(RelaycastRegistrationError::InvalidAgentName); } - let declared_metadata = declared_metadata_map(declared); + let mut declared_metadata = spawned.clone(); + declared_metadata.extend(declared_metadata_map(declared)); if declared_metadata.is_empty() { return Ok(()); } @@ -2968,6 +2974,83 @@ async fn register_new_spawn_identity_inner( } } +/// What a broker knows about a worker it spawned, as roster metadata: the +/// CLI it runs (`cli`), this machine's name (`host`) and the signed-in Cloud +/// person who runs this broker (`owner_hash`, the SHA-256 hex of their Cloud +/// user id). These are the keys Agent Relay Desktop stamps on the sessions it +/// registers, so a worker spawned on someone's fleet node is listed as theirs +/// and on that machine instead of as an unowned agent. A key whose value is +/// unknown is omitted, never sent blank. +pub fn spawned_worker_metadata(cli: &str) -> serde_json::Map { + let host = hostname::get() + .ok() + .and_then(|name| name.into_string().ok()); + spawned_worker_metadata_from(cli, host.as_deref(), signed_in_user_id().as_deref()) +} + +fn spawned_worker_metadata_from( + cli: &str, + host: Option<&str>, + user_id: Option<&str>, +) -> serde_json::Map { + let mut metadata = serde_json::Map::new(); + let cli = crate::cli::command_parse::parse_cli_command(cli) + .map(|(command, _)| crate::cli::command_parse::normalize_cli_name(&command)) + .unwrap_or_default(); + if !cli.is_empty() { + metadata.insert("cli".into(), Value::String(cli)); + } + // The desktop's label: no trailing dot or `.local` (macOS Bonjour names). + if let Some(host) = host.map(|host| host.trim().trim_end_matches('.')) { + let host = match host.len().checked_sub(".local".len()) { + Some(cut) + if host.is_char_boundary(cut) && host[cut..].eq_ignore_ascii_case(".local") => + { + &host[..cut] + } + _ => host, + }; + if !host.is_empty() { + metadata.insert("host".into(), Value::String(host.to_string())); + } + } + if let Some(user_id) = user_id.map(str::trim).filter(|id| !id.is_empty()) { + let hash = Sha256::digest(user_id.as_bytes()) + .iter() + .map(|byte| format!("{byte:02x}")) + .collect::(); + metadata.insert("owner_hash".into(), Value::String(hash)); + } + metadata +} + +/// The Cloud user this broker runs for: `AGENT_RELAY_USER_ID` when the CLI +/// published it, otherwise the CLI's stored sign-in. A broker started by a +/// service manager has no CLI parent, so the file is the common case there. +fn signed_in_user_id() -> Option { + let from_env = std::env::var("AGENT_RELAY_USER_ID") + .ok() + .map(|id| id.trim().to_string()) + .filter(|id| !id.is_empty()); + from_env.or_else(|| { + let data_dir = std::env::var("AGENT_RELAY_DATA_DIR") + .ok() + .map(|dir| dir.trim().to_string()) + .filter(|dir| !dir.is_empty()) + .map(std::path::PathBuf::from) + .or_else(|| dirs::home_dir().map(|home| home.join(".agentworkforce/relay")))?; + let file = std::fs::read_to_string(data_dir.join("cloud-identity.json")).ok()?; + let identity: Value = serde_json::from_str(&file).ok()?; + identity["userId"] + .as_str() + .map(str::trim) + .filter(|id| { + !id.is_empty() && id.len() <= 128 && id.chars().all(|c| (' '..='~').contains(&c)) + }) + .map(str::to_string) + }) +} + /// The declared fields alone, trimmed, with blanks omitted. /// /// Omitting rather than sending `""` matters because both callers merge this @@ -4421,6 +4504,7 @@ mod tests { role: None, objective: Some("Publish registration metadata".to_string()), }, + &serde_json::Map::new(), ) .await .expect("publishing declared metadata should succeed"); @@ -4454,6 +4538,7 @@ mod tests { project: Some(" ".to_string()), ..Default::default() }, + &serde_json::Map::new(), ) .await .expect("an empty declaration is a no-op, not an error"); @@ -4462,6 +4547,80 @@ mod tests { any_write.assert_hits(0); } + /// The worker's CLI, machine and owner ride the declared-metadata PATCH, + /// so the roster can show whose worker it is, where it runs and its icon. + #[tokio::test] + async fn publish_declared_metadata_carries_the_spawned_worker_keys() { + let server = MockServer::start(); + let update = server.mock(|when, then| { + when.method(PATCH) + .path("/v1/agents/worker-a") + .json_body(json!({ + "metadata": { + "cli": "claude", + "host": "Finn-Mac-Mini", + "owner_hash": "abc", + "objective": "Check the roster" + } + })); + then.status(200).json_body(json!({ + "ok": true, + "data": { + "id": "agent_worker_a", + "name": "worker-a", + "type": "agent", + "status": "online", + "persona": null, + "metadata": {} + } + })); + }); + let spawned = json!({"cli": "claude", "host": "Finn-Mac-Mini", "owner_hash": "abc"}); + + let client = seeded_http_client(&server.base_url()); + client + .publish_declared_metadata( + "worker-a", + &AgentRegistrationMetadata { + objective: Some("Check the roster".to_string()), + ..Default::default() + }, + spawned.as_object().unwrap(), + ) + .await + .expect("publishing spawned worker metadata should succeed"); + + update.assert_hits(1); + } + + #[test] + fn spawned_worker_metadata_matches_the_desktop_keys() { + let metadata = super::spawned_worker_metadata_from( + "/opt/homebrew/bin/claude --model opus", + Some("Finn-Mac-Mini.local."), + Some(" user-1 "), + ); + assert_eq!( + serde_json::Value::Object(metadata), + json!({ + "cli": "claude", + "host": "Finn-Mac-Mini", + // SHA-256 of "user-1", the desktop's `identity_hash`. + "owner_hash": "c6c289e49e9c05b2145860387b73bcb18df43fb09a1e4a4a9713c76c88bb541b" + }) + ); + } + + #[test] + fn spawned_worker_metadata_omits_what_it_does_not_know() { + let metadata = super::spawned_worker_metadata_from(" ", Some(" .local "), Some("")); + assert!(metadata.is_empty(), "unexpected keys: {metadata:?}"); + let metadata = super::spawned_worker_metadata_from("codex", Some("éabcde"), None); + assert_eq!(metadata["host"], "éabcde"); + let metadata = super::spawned_worker_metadata_from("codex", None, None); + assert_eq!(serde_json::Value::Object(metadata), json!({"cli": "codex"})); + } + /// A presence update used to call POST /v1/agents/release with no reason. /// That invalidated the credential of a participant that could still be /// running, and left an unattributable `release.reason = null` record. The diff --git a/crates/broker/src/runtime/api.rs b/crates/broker/src/runtime/api.rs index 9c027dbc5b..6d08db5208 100644 --- a/crates/broker/src/runtime/api.rs +++ b/crates/broker/src/runtime/api.rs @@ -802,6 +802,7 @@ impl BrokerRuntime { super::fleet::spawn_declared_metadata_publish( relaycast_http, name.as_str(), + cli.as_str(), registration_metadata, ); } diff --git a/crates/broker/src/runtime/fleet.rs b/crates/broker/src/runtime/fleet.rs index 94910d78f9..b9722a7845 100644 --- a/crates/broker/src/runtime/fleet.rs +++ b/crates/broker/src/runtime/fleet.rs @@ -2384,18 +2384,27 @@ pub(super) async fn reconcile_blocked_flush_predecessor( /// either — a failure is logged at error level with the agent name and the /// underlying error, and it is not retried, because the honest signal is worth /// more than a hidden retry loop on a non-critical publish. +/// +/// The same PATCH carries the worker's CLI, this machine's name and the +/// broker's signed-in owner, so the roster can show whose agent it is, where +/// it runs and which CLI's icon to draw. pub(super) fn spawn_declared_metadata_publish( relaycast_http: &RelaycastHttpClient, name: &str, + cli: &str, declared: AgentRegistrationMetadata, ) { - if declared.is_empty() { + let spawned = crate::relaycast::spawned_worker_metadata(cli); + if declared.is_empty() && spawned.is_empty() { return; } let http = relaycast_http.clone(); let agent = name.to_string(); tokio::spawn(async move { - match http.publish_declared_metadata(&agent, &declared).await { + match http + .publish_declared_metadata(&agent, &declared, &spawned) + .await + { Ok(()) => tracing::debug!( worker = %agent, "published declared workforce metadata for spawned agent" @@ -2404,8 +2413,8 @@ pub(super) fn spawn_declared_metadata_publish( worker = %agent, error = %error, "failed to publish declared workforce metadata; the agent is registered and \ - running but its declared organization/project/workstream/role/objective are \ - not visible to the engine" + running but its declared organization/project/workstream/role/objective and \ + its cli/host/owner are not visible to the engine" ), } }); diff --git a/crates/broker/src/runtime/relaycast_events.rs b/crates/broker/src/runtime/relaycast_events.rs index ad5a2dd8a8..b3f8031614 100644 --- a/crates/broker/src/runtime/relaycast_events.rs +++ b/crates/broker/src/runtime/relaycast_events.rs @@ -869,6 +869,7 @@ pub(super) async fn spawn_worker_from_request( super::fleet::spawn_declared_metadata_publish( workspace_http, name.as_str(), + cli.as_str(), registration_metadata, ); let relay_key = token.token.clone(); @@ -906,6 +907,7 @@ pub(super) async fn spawn_worker_from_request( super::fleet::spawn_declared_metadata_publish( workspace_http, name.as_str(), + cli.as_str(), registration_metadata, ); tracing::info!( diff --git a/crates/broker/src/runtime/tests.rs b/crates/broker/src/runtime/tests.rs index c8d55338b7..7be745e1e3 100644 --- a/crates/broker/src/runtime/tests.rs +++ b/crates/broker/src/runtime/tests.rs @@ -8463,13 +8463,19 @@ async fn assert_http_spawn_metadata_publication(supplied_token: bool, valid_cwd: then.status(200) .json_body(json!({"ok":true,"data":{"channels":[]}})); }); + // Partial: the same PATCH also carries this machine's `host` and, when + // the test machine is signed in, its `owner_hash`. let metadata = server.mock(|when, then| { when.method(PATCH) .path("/v1/agents/metadata-worker") - .json_body(json!({"metadata":{ - "organization":"demo-org", "project":"demo-project", - "workstream":"subscriptions", "role":"reviewer", "objective":"prove delivery" - }})); + .json_body_partial( + json!({"metadata":{ + "organization":"demo-org", "project":"demo-project", + "workstream":"subscriptions", "role":"reviewer", "objective":"prove delivery", + "cli":"cat" + }}) + .to_string(), + ); then.status(200) .json_body(json!({"ok":true,"data":identity})); }); From 5134478d3d8e6db7c7ada7c894dd128cc561258a Mon Sep 17 00:00:00 2001 From: khaliqgant Date: Wed, 7 Oct 2026 10:06:03 -0700 Subject: [PATCH 2/6] fix(broker): publish worker metadata after launch, from the stored sign-in Review follow-ups: - Publish once the worker launched, on every hosted credential path of the action spawn (node bind, HTTP fallback and a supplied token), as the API spawn already did. - Record the CLI the worker runs: a PTY harness's own command. - Read the owner from cloud-identity.json at each spawn and stop using the inherited AGENT_RELAY_USER_ID, so a sign-out applies to the next worker and every id passes the CLI's validation. - Resolve the hostname and the identity off the event loop. - Match the spawn test's PATCH body in full; mark the release Patch. Co-Authored-By: Claude Opus 5.5 (1M context) --- CHANGELOG.md | 4 +- crates/broker/src/relaycast/ws.rs | 62 ++++++++++++------- crates/broker/src/runtime/api.rs | 2 +- crates/broker/src/runtime/fleet.rs | 25 ++++++-- crates/broker/src/runtime/relaycast_events.rs | 26 ++++---- crates/broker/src/runtime/tests.rs | 26 +++++--- 6 files changed, 88 insertions(+), 57 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 0f3a25ceef..bcd01361d3 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,11 +5,11 @@ All notable changes to Agent Relay will be documented in this file. The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/), and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). -## [Unreleased] +## [Unreleased - Patch] ### Fixed -- Workers a fleet broker spawns now record their CLI (`cli`), the machine's name (`host`) and the signed-in Cloud person who runs the broker (`owner_hash`) in their Relaycast metadata. These are the keys Agent Relay Desktop records for the sessions it registers, so the desktop lists your fleet workers as your agents on that machine, with their harness icon, instead of under "Other live agents". The broker reads the person from `AGENT_RELAY_USER_ID` or, for a broker started by a service manager, from the CLI's stored `cloud-identity.json`. +- Workers started with `agent-relay fleet spawn` now appear in Agent Relay Desktop as your agents, on the machine that runs them, with their CLI's icon, instead of under "Other live agents". ## [13.1.3] - 2026-10-07 diff --git a/crates/broker/src/relaycast/ws.rs b/crates/broker/src/relaycast/ws.rs index 6c7a5e6f1a..b3dab880c5 100644 --- a/crates/broker/src/relaycast/ws.rs +++ b/crates/broker/src/relaycast/ws.rs @@ -945,7 +945,8 @@ impl RelaycastHttpClient { agent_name: name.to_string(), detail: "SDK relay client not initialized".to_string(), })?; - // Send ONLY the declared keys. `PATCH /v1/agents/:name` merges them over + // Send ONLY the declared keys and the spawned worker's `cli`, `host` and + // `owner_hash`. `PATCH /v1/agents/:name` merges them over // the record's existing metadata server-side — verified in the engine at // both the ref fleet-e2e pins (v7.0.0, eb7563ff) and relaycast `main` // (`packages/engine/src/routes/agent.ts`: @@ -3024,31 +3025,33 @@ fn spawned_worker_metadata_from( metadata } -/// The Cloud user this broker runs for: `AGENT_RELAY_USER_ID` when the CLI -/// published it, otherwise the CLI's stored sign-in. A broker started by a -/// service manager has no CLI parent, so the file is the common case there. +/// The Cloud user this broker runs for: the CLI's stored sign-in, read at +/// each spawn, so signing in or out takes effect for the next worker without a +/// broker restart. `AGENT_RELAY_USER_ID` is deliberately not used: it is a +/// copy taken when the broker started, which outlives a sign-out, and a broker +/// started by a service manager has none. fn signed_in_user_id() -> Option { - let from_env = std::env::var("AGENT_RELAY_USER_ID") + let data_dir = std::env::var("AGENT_RELAY_DATA_DIR") .ok() - .map(|id| id.trim().to_string()) - .filter(|id| !id.is_empty()); - from_env.or_else(|| { - let data_dir = std::env::var("AGENT_RELAY_DATA_DIR") - .ok() - .map(|dir| dir.trim().to_string()) - .filter(|dir| !dir.is_empty()) - .map(std::path::PathBuf::from) - .or_else(|| dirs::home_dir().map(|home| home.join(".agentworkforce/relay")))?; - let file = std::fs::read_to_string(data_dir.join("cloud-identity.json")).ok()?; - let identity: Value = serde_json::from_str(&file).ok()?; - identity["userId"] - .as_str() - .map(str::trim) - .filter(|id| { - !id.is_empty() && id.len() <= 128 && id.chars().all(|c| (' '..='~').contains(&c)) - }) - .map(str::to_string) - }) + .map(|dir| dir.trim().to_string()) + .filter(|dir| !dir.is_empty()) + .map(std::path::PathBuf::from) + .or_else(|| dirs::home_dir().map(|home| home.join(".agentworkforce/relay")))?; + let file = std::fs::read_to_string(data_dir.join("cloud-identity.json")).ok()?; + user_id_from_identity(&file) +} + +/// The `userId` of a stored identity, held to the CLI's own rules: printable +/// ASCII, at most 128 bytes once trimmed. +fn user_id_from_identity(file: &str) -> Option { + let identity: Value = serde_json::from_str(file).ok()?; + identity["userId"] + .as_str() + .map(str::trim) + .filter(|id| { + !id.is_empty() && id.len() <= 128 && id.chars().all(|c| (' '..='~').contains(&c)) + }) + .map(str::to_string) } /// The declared fields alone, trimmed, with blanks omitted. @@ -4621,6 +4624,17 @@ mod tests { assert_eq!(serde_json::Value::Object(metadata), json!({"cli": "codex"})); } + #[test] + fn stored_identity_user_id_follows_the_cli_rules() { + let id = |file: &str| super::user_id_from_identity(file); + assert_eq!(id(r#"{"userId":" user-1 "}"#).as_deref(), Some("user-1")); + assert_eq!(id(r#"{"userId":"bad\u0007id"}"#), None); + assert_eq!(id(&format!(r#"{{"userId":"{}"}}"#, "u".repeat(129))), None); + assert_eq!(id(r#"{"userId":""}"#), None); + assert_eq!(id(r#"{"email":"a@b"}"#), None); + assert_eq!(id("not json"), None); + } + /// A presence update used to call POST /v1/agents/release with no reason. /// That invalidated the credential of a participant that could still be /// running, and left an unattributable `release.reason = null` record. The diff --git a/crates/broker/src/runtime/api.rs b/crates/broker/src/runtime/api.rs index 6d08db5208..ee2bd9d919 100644 --- a/crates/broker/src/runtime/api.rs +++ b/crates/broker/src/runtime/api.rs @@ -802,7 +802,7 @@ impl BrokerRuntime { super::fleet::spawn_declared_metadata_publish( relaycast_http, name.as_str(), - cli.as_str(), + &effective_spec, registration_metadata, ); } diff --git a/crates/broker/src/runtime/fleet.rs b/crates/broker/src/runtime/fleet.rs index b9722a7845..a2a17d65e2 100644 --- a/crates/broker/src/runtime/fleet.rs +++ b/crates/broker/src/runtime/fleet.rs @@ -2391,16 +2391,22 @@ pub(super) async fn reconcile_blocked_flush_predecessor( pub(super) fn spawn_declared_metadata_publish( relaycast_http: &RelaycastHttpClient, name: &str, - cli: &str, + spec: &crate::protocol::AgentSpec, declared: AgentRegistrationMetadata, ) { - let spawned = crate::relaycast::spawned_worker_metadata(cli); - if declared.is_empty() && spawned.is_empty() { - return; - } let http = relaycast_http.clone(); let agent = name.to_string(); + let cli = launched_cli(spec).to_string(); tokio::spawn(async move { + // The hostname call and the identity file read block; keep them off + // the runtime's async workers too. + let spawned = + tokio::task::spawn_blocking(move || crate::relaycast::spawned_worker_metadata(&cli)) + .await + .unwrap_or_default(); + if declared.is_empty() && spawned.is_empty() { + return; + } match http .publish_declared_metadata(&agent, &declared, &spawned) .await @@ -2420,6 +2426,15 @@ pub(super) fn spawn_declared_metadata_publish( }); } +/// The CLI a launched worker actually runs: a PTY harness's own command, +/// which can differ from the requested `cli`, else the requested `cli`. +fn launched_cli(spec: &crate::protocol::AgentSpec) -> &str { + match spec.harness_config.as_ref() { + Some(crate::protocol::ResolvedHarnessConfig::Pty(config)) => config.command.as_str(), + _ => spec.cli.as_deref().unwrap_or_default(), + } +} + /// Bind an agent to this node by sending node-control `agent.register` and /// awaiting the engine reply with the minted agent token. This is the single /// "register agent via node" step both the `/api/spawn` path and the node diff --git a/crates/broker/src/runtime/relaycast_events.rs b/crates/broker/src/runtime/relaycast_events.rs index b3f8031614..cb1c002b2a 100644 --- a/crates/broker/src/runtime/relaycast_events.rs +++ b/crates/broker/src/runtime/relaycast_events.rs @@ -866,12 +866,6 @@ pub(super) async fn spawn_worker_from_request( worker = %name, "bound agent to node via agent.register for action.invoke spawn" ); - super::fleet::spawn_declared_metadata_publish( - workspace_http, - name.as_str(), - cli.as_str(), - registration_metadata, - ); let relay_key = token.token.clone(); node_registered_agent_id = Some(token.agent_id.clone()); fleet_registration = Some((token, invocation_id.clone(), session_ref.clone())); @@ -901,15 +895,6 @@ pub(super) async fn spawn_worker_from_request( .await { Ok(token) => { - // Declared metadata is published over the agent API - // exactly as on the node path; registration itself - // stays on the cache- and rate-limit-aware call. - super::fleet::spawn_declared_metadata_publish( - workspace_http, - name.as_str(), - cli.as_str(), - registration_metadata, - ); tracing::info!( worker = %name, "pre-registered agent via broker for WS spawn" @@ -1019,6 +1004,17 @@ pub(super) async fn spawn_worker_from_request( .await { Ok(effective_spec) => { + // Every hosted credential path (node bind, HTTP fallback or a + // supplied token) publishes declared metadata, only once the worker + // launched, as the API spawn does. + if worker_relay_key.is_some() { + super::fleet::spawn_declared_metadata_publish( + workspace_http, + name.as_str(), + &effective_spec, + registration_metadata, + ); + } if owns_identity { if let Some(worker) = workers.workers.get(&name) { workers diff --git a/crates/broker/src/runtime/tests.rs b/crates/broker/src/runtime/tests.rs index 7be745e1e3..f180aa61ba 100644 --- a/crates/broker/src/runtime/tests.rs +++ b/crates/broker/src/runtime/tests.rs @@ -8463,19 +8463,25 @@ async fn assert_http_spawn_metadata_publication(supplied_token: bool, valid_cwd: then.status(200) .json_body(json!({"ok":true,"data":{"channels":[]}})); }); - // Partial: the same PATCH also carries this machine's `host` and, when - // the test machine is signed in, its `owner_hash`. + // The same PATCH carries this machine's `host` and, when the test machine + // is signed in, its `owner_hash`; both are exactly what the broker + // computes for this process, so the body is still matched in full. + let mut expected = serde_json::Map::new(); + expected.extend(crate::relaycast::spawned_worker_metadata("cat")); + expected.extend( + json!({ + "organization":"demo-org", "project":"demo-project", + "workstream":"subscriptions", "role":"reviewer", "objective":"prove delivery" + }) + .as_object() + .unwrap() + .clone(), + ); + assert_eq!(expected["cli"], "cat"); let metadata = server.mock(|when, then| { when.method(PATCH) .path("/v1/agents/metadata-worker") - .json_body_partial( - json!({"metadata":{ - "organization":"demo-org", "project":"demo-project", - "workstream":"subscriptions", "role":"reviewer", "objective":"prove delivery", - "cli":"cat" - }}) - .to_string(), - ); + .json_body(json!({ "metadata": expected })); then.status(200) .json_body(json!({"ok":true,"data":identity})); }); From 29f28bda7775cb583d0a557bedb6daf6825e3e09 Mon Sep 17 00:00:00 2001 From: khaliqgant Date: Wed, 7 Oct 2026 10:22:57 -0700 Subject: [PATCH 3/6] fix(broker): log a failed worker metadata lookup instead of dropping it Co-Authored-By: Claude Opus 5.5 (1M context) --- crates/broker/src/runtime/fleet.rs | 21 +++++++++++++++++---- 1 file changed, 17 insertions(+), 4 deletions(-) diff --git a/crates/broker/src/runtime/fleet.rs b/crates/broker/src/runtime/fleet.rs index a2a17d65e2..f12db8a6cc 100644 --- a/crates/broker/src/runtime/fleet.rs +++ b/crates/broker/src/runtime/fleet.rs @@ -2400,10 +2400,23 @@ pub(super) fn spawn_declared_metadata_publish( tokio::spawn(async move { // The hostname call and the identity file read block; keep them off // the runtime's async workers too. - let spawned = - tokio::task::spawn_blocking(move || crate::relaycast::spawned_worker_metadata(&cli)) - .await - .unwrap_or_default(); + // A failed lookup still publishes the declared fields, and says so. + let spawned = match tokio::task::spawn_blocking(move || { + crate::relaycast::spawned_worker_metadata(&cli) + }) + .await + { + Ok(spawned) => spawned, + Err(error) => { + tracing::error!( + worker = %agent, + error = %error, + "could not read the spawned worker's cli/host/owner; publishing its \ + declared workforce metadata without them" + ); + serde_json::Map::new() + } + }; if declared.is_empty() && spawned.is_empty() { return; } From 7296571363935efce3be3ebae56f264f2ab20a0a Mon Sep 17 00:00:00 2001 From: khaliqgant Date: Wed, 7 Oct 2026 10:30:27 -0700 Subject: [PATCH 4/6] fix(broker): record a provider-only headless worker's CLI Review follow-up: a headless spawn that names only its provider has no `cli`, so the roster got none. Fall back to the provider's CLI. Co-Authored-By: Claude Opus 5.5 (1M context) --- crates/broker/src/runtime/fleet.rs | 39 ++++++++++++++++++++++++++++-- 1 file changed, 37 insertions(+), 2 deletions(-) diff --git a/crates/broker/src/runtime/fleet.rs b/crates/broker/src/runtime/fleet.rs index f12db8a6cc..a946318223 100644 --- a/crates/broker/src/runtime/fleet.rs +++ b/crates/broker/src/runtime/fleet.rs @@ -2440,11 +2440,46 @@ pub(super) fn spawn_declared_metadata_publish( } /// The CLI a launched worker actually runs: a PTY harness's own command, -/// which can differ from the requested `cli`, else the requested `cli`. +/// which can differ from the requested `cli`, else the requested `cli`, else +/// the CLI a provider-only headless spawn runs. fn launched_cli(spec: &crate::protocol::AgentSpec) -> &str { match spec.harness_config.as_ref() { Some(crate::protocol::ResolvedHarnessConfig::Pty(config)) => config.command.as_str(), - _ => spec.cli.as_deref().unwrap_or_default(), + _ => spec + .cli + .as_deref() + .or_else(|| { + spec.provider + .as_ref() + .map(crate::runtime::headless::headless_provider_cli_name) + }) + .unwrap_or_default(), + } +} + +#[cfg(test)] +mod launched_cli_tests { + use super::launched_cli; + use crate::protocol::AgentSpec; + use serde_json::json; + + fn spec(value: serde_json::Value) -> AgentSpec { + serde_json::from_value(value).expect("a valid agent spec") + } + + #[test] + fn the_roster_cli_is_what_the_worker_runs() { + let requested = spec(json!({"name": "w", "runtime": "pty", "cli": "codex"})); + assert_eq!(launched_cli(&requested), "codex"); + let harness = spec(json!({ + "name": "w", "runtime": "pty", "cli": "codex", + "harnessConfig": {"runtime": "pty", "command": "/usr/local/bin/claude"} + })); + assert_eq!(launched_cli(&harness), "/usr/local/bin/claude"); + let headless = spec(json!({"name": "w", "runtime": "headless", "provider": "opencode"})); + assert_eq!(launched_cli(&headless), "opencode"); + let unknown = spec(json!({"name": "w", "runtime": "pty"})); + assert_eq!(launched_cli(&unknown), ""); } } From 03227995dcfc1842f8525c0640357c9030070e7d Mon Sep 17 00:00:00 2001 From: khaliqgant Date: Wed, 7 Oct 2026 11:47:00 -0700 Subject: [PATCH 5/6] test(broker): de-flake delivery retry and integration release - delivery_retry_transient_blip_emits_failed_event_for_present_worker: macOS sometimes accepts a write to the exited child's pipe, which resets the consecutive-failure budget, so a fixed retry count failed three CI runs in a row. Count failures the way the retry path does and require the terminal failure exactly at the consecutive cap. - Broker integration harness: an API release reports an identity cleanup it could not confirm yet (Relaycast `d1_pressure`) at once and retries it every 5s, expecting the caller to release again. The harness now does, for up to 60s. Reproduced on main's broker too. Co-Authored-By: Claude Opus 5.5 (1M context) --- crates/broker/src/runtime/tests.rs | 42 +++++++++++-------- .../broker/utils/broker-harness.ts | 17 +++++++- 2 files changed, 40 insertions(+), 19 deletions(-) diff --git a/crates/broker/src/runtime/tests.rs b/crates/broker/src/runtime/tests.rs index f180aa61ba..0ec26c0795 100644 --- a/crates/broker/src/runtime/tests.rs +++ b/crates/broker/src/runtime/tests.rs @@ -3786,8 +3786,15 @@ async fn delivery_retry_transient_blip_emits_failed_event_for_present_worker() { }, )]); + // A write to the exited child's pipe usually fails, but some platforms + // (macOS in particular) accept an occasional write after exit. An accepted + // write resets the consecutive-failure budget, so count failures the way + // the retry path does: the delivery must fail exactly when + // MAX_DELIVERY_RETRIES writes in a row have failed, and never earlier. let mut final_outcome = None; - for retry_index in 1..=MAX_DELIVERY_RETRIES + 1 { + let mut consecutive_failures = 0; + let mut total_attempts = 0; + for _ in 0..MAX_DELIVERY_RETRIES * 4 { match retry_pending_delivery( &DeliveryId::new("del_blip"), &mut workers, @@ -3800,35 +3807,34 @@ async fn delivery_retry_transient_blip_emits_failed_event_for_present_worker() { let DeliveryAttemptOutcome::Failed { ref pending, .. } = outcome else { unreachable!(); }; - assert_eq!(pending.attempts, MAX_DELIVERY_RETRIES); - // Some platforms can accept a final pipe write after the child exits, - // so terminal failure may arrive on the immediate post-cap check. - assert!( - retry_index >= MAX_DELIVERY_RETRIES, - "delivery should not fail before the retry cap is exhausted" + total_attempts += 1; + consecutive_failures += 1; + assert_eq!( + consecutive_failures, MAX_DELIVERY_RETRIES, + "delivery must fail exactly when the consecutive retry cap is exhausted" ); + assert_eq!(pending.failed_attempts, MAX_DELIVERY_RETRIES); + assert_eq!(pending.attempts, total_attempts); final_outcome = Some(outcome); break; } Ok(DeliveryAttemptOutcome::Attempted { attempts, .. }) => { - assert!( - attempts <= MAX_DELIVERY_RETRIES, - "retry attempts must stay within the retry cap" - ); - assert!( - retry_index <= MAX_DELIVERY_RETRIES, - "the retry after the cap should return a terminal failure" - ); + total_attempts += 1; + consecutive_failures = 0; + assert_eq!(attempts, total_attempts); } Ok(DeliveryAttemptOutcome::Noop) => { + total_attempts += 1; + consecutive_failures += 1; assert!( - retry_index < MAX_DELIVERY_RETRIES, - "the final bounded retry should return a terminal failure" + consecutive_failures < MAX_DELIVERY_RETRIES, + "the write that exhausts the cap should return a terminal failure" ); let pending = pending_deliveries .get("del_blip") .expect("delivery remains pending before terminal failure"); - assert_eq!(pending.attempts, retry_index); + assert_eq!(pending.attempts, total_attempts); + assert_eq!(pending.failed_attempts, consecutive_failures); assert!(pending .last_error .as_deref() diff --git a/tests/integration/broker/utils/broker-harness.ts b/tests/integration/broker/utils/broker-harness.ts index 3ecbcadd4f..652a904042 100644 --- a/tests/integration/broker/utils/broker-harness.ts +++ b/tests/integration/broker/utils/broker-harness.ts @@ -316,7 +316,22 @@ export class BrokerHarness { * Release an agent by name via the low-level client. */ async releaseAgent(name: string): Promise<{ name: string }> { - return this.client.release(name); + // The broker reports an identity cleanup it could not confirm yet (for + // example Relaycast answering `d1_pressure`) at once, keeps retrying it + // every 5s, and expects API callers to release again. Until its retry + // lands, a repeated release answers "cleanup is in progress". + const deadline = Date.now() + 60_000; + for (;;) { + try { + return await this.client.release(name); + } catch (error) { + const message = error instanceof Error ? error.message : String(error); + const pendingCleanup = + /owned identity cleanup unconfirmed/.test(message) || /cleanup is in progress/.test(message); + if (!pendingCleanup || Date.now() >= deadline) throw error; + await new Promise((resolve) => setTimeout(resolve, 3_000)); + } + } } /** From 990fa112d4fea80f42c37e4cd4cbd89ba8dfa4ed Mon Sep 17 00:00:00 2001 From: khaliqgant Date: Wed, 7 Oct 2026 11:51:14 -0700 Subject: [PATCH 6/6] test(broker): expect the real attempt count in the delivery-failed event Review follow-up: with an accepted post-exit write, the total attempts exceed the consecutive cap, so the failed event and dead letter carry the total, not MAX_DELIVERY_RETRIES. Co-Authored-By: Claude Opus 5.5 (1M context) --- crates/broker/src/runtime/tests.rs | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/crates/broker/src/runtime/tests.rs b/crates/broker/src/runtime/tests.rs index 0ec26c0795..173bf1e089 100644 --- a/crates/broker/src/runtime/tests.rs +++ b/crates/broker/src/runtime/tests.rs @@ -3876,7 +3876,7 @@ async fn delivery_retry_transient_blip_emits_failed_event_for_present_worker() { assert_eq!(frame.payload["to"], worker_name); assert_eq!( frame.payload["attempts"].as_u64(), - Some(u64::from(MAX_DELIVERY_RETRIES)) + Some(u64::from(total_attempts)) ); let last_error = frame.payload["lastError"].as_str().unwrap_or_default(); assert!( @@ -3901,7 +3901,7 @@ async fn delivery_retry_transient_blip_emits_failed_event_for_present_worker() { ); let entry = dead_letters.get("del_blip").expect("dead letter by id"); assert_eq!(entry.delivery.body, "transient auth blip"); - assert_eq!(entry.attempts, MAX_DELIVERY_RETRIES); + assert_eq!(entry.attempts, total_attempts); } #[tokio::test]