diff --git a/CHANGELOG.md b/CHANGELOG.md index 6056d3697..bcd01361d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,7 +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 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/mod.rs b/crates/broker/src/relaycast/mod.rs index 38b6c2a16..9b8494a87 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 6d51b1856..b3dab880c 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(()); } @@ -939,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`: @@ -2968,6 +2975,85 @@ 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: 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 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()?; + 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. /// /// Omitting rather than sending `""` matters because both callers merge this @@ -4421,6 +4507,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 +4541,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 +4550,91 @@ 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"})); + } + + #[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 9c027dbc5..ee2bd9d91 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(), + &effective_spec, registration_metadata, ); } diff --git a/crates/broker/src/runtime/fleet.rs b/crates/broker/src/runtime/fleet.rs index 94910d78f..a94631822 100644 --- a/crates/broker/src/runtime/fleet.rs +++ b/crates/broker/src/runtime/fleet.rs @@ -2384,18 +2384,46 @@ 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, + spec: &crate::protocol::AgentSpec, declared: AgentRegistrationMetadata, ) { - if declared.is_empty() { - return; - } let http = relaycast_http.clone(); let agent = name.to_string(); + let cli = launched_cli(spec).to_string(); tokio::spawn(async move { - match http.publish_declared_metadata(&agent, &declared).await { + // The hostname call and the identity file read block; keep them off + // the runtime's async workers too. + // 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; + } + match http + .publish_declared_metadata(&agent, &declared, &spawned) + .await + { Ok(()) => tracing::debug!( worker = %agent, "published declared workforce metadata for spawned agent" @@ -2404,13 +2432,57 @@ 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" ), } }); } +/// The CLI a launched worker actually runs: a PTY harness's own command, +/// 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() + .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), ""); + } +} + /// 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 ad5a2dd8a..cb1c002b2 100644 --- a/crates/broker/src/runtime/relaycast_events.rs +++ b/crates/broker/src/runtime/relaycast_events.rs @@ -866,11 +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(), - 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())); @@ -900,14 +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(), - registration_metadata, - ); tracing::info!( worker = %name, "pre-registered agent via broker for WS spawn" @@ -1017,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 c8d55338b..173bf1e08 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() @@ -3870,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!( @@ -3895,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] @@ -8463,13 +8469,25 @@ async fn assert_http_spawn_metadata_publication(supplied_token: bool, valid_cwd: then.status(200) .json_body(json!({"ok":true,"data":{"channels":[]}})); }); + // 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(json!({"metadata":{ - "organization":"demo-org", "project":"demo-project", - "workstream":"subscriptions", "role":"reviewer", "objective":"prove delivery" - }})); + .json_body(json!({ "metadata": expected })); then.status(200) .json_body(json!({"ok":true,"data":identity})); }); diff --git a/tests/integration/broker/utils/broker-harness.ts b/tests/integration/broker/utils/broker-harness.ts index 3ecbcadd4..652a90404 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)); + } + } } /**