Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 5 additions & 1 deletion CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -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
Comment thread
coderabbitai[bot] marked this conversation as resolved.

- 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

Expand Down
4 changes: 2 additions & 2 deletions crates/broker/src/relaycast/mod.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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,
};
177 changes: 175 additions & 2 deletions crates/broker/src/relaycast/ws.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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<String, Value>,
) -> 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));
Comment thread
cubic-dev-ai[bot] marked this conversation as resolved.
if declared_metadata.is_empty() {
return Ok(());
}
Expand All @@ -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`:
Expand Down Expand Up @@ -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<String, Value> {
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<String, Value> {
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::<String>();
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<String> {
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<String> {
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
Expand Down Expand Up @@ -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");
Expand Down Expand Up @@ -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");
Expand All @@ -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
Expand Down
1 change: 1 addition & 0 deletions crates/broker/src/runtime/api.rs
Original file line number Diff line number Diff line change
Expand Up @@ -802,6 +802,7 @@ impl BrokerRuntime {
super::fleet::spawn_declared_metadata_publish(
relaycast_http,
name.as_str(),
&effective_spec,
registration_metadata,
);
}
Expand Down
84 changes: 78 additions & 6 deletions crates/broker/src/runtime/fleet.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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"
Expand All @@ -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
Expand Down
Loading
Loading