From 77c4de01ed63b3471c0190a2e330fb6d065c3e9f Mon Sep 17 00:00:00 2001 From: Miya Date: Fri, 14 Aug 2026 11:20:46 +0200 Subject: [PATCH] feat(fleet): publish declared worker metadata MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Fleet spawn callers can declare an agent's organization, project, workstream, role, and objective through the CLI, the MCP `spawn` tool, or the Fleet DSL. Those fields are published onto the agent's Relaycast record so consumers can read a worker's identity instead of guessing it from the agent's name. `objective` falls back to the spawn's task when no explicit objective is given. The declared fields deliberately do NOT ride the node-control `agent.register` frame. The engine parses that frame with a `.strict()` schema (relaycast packages/types/src/fleet-wire.ts) that rejects unknown keys, and its rejection is sent with a freshly generated id, which the broker's id-keyed correlation never matches — so the registration waiter stalls for the full 30s FLEET_AGENT_REGISTER_TIMEOUT and then silently falls back to HTTP pre-registration. Instead the fields are published over the REST agent API once registration succeeds, on a detached task so nothing is added to the spawn await that runs inline on the runtime event loop. A failed publish never fails the spawn and is logged rather than retried. A serialization test asserts the exact key set of `agent.register` so re-adding a field there fails loudly instead of resurfacing as a stall, and a live-engine e2e proves the declared fields actually land. --- CHANGELOG.md | 5 + crates/broker/src/fleet_wire.rs | 158 ++++++++++++- crates/broker/src/listen_api.rs | 26 ++- crates/broker/src/relaycast/ws.rs | 212 +++++++++++++++++- crates/broker/src/runtime/api.rs | 18 ++ crates/broker/src/runtime/fleet.rs | 44 +++- crates/broker/src/runtime/mod.rs | 6 +- crates/broker/src/runtime/relaycast_events.rs | 15 ++ .../src/cli/agent-relay-mcp.protocol.test.ts | 10 + .../src/cli/agent-relay-mcp.startup.test.ts | 18 ++ packages/cli/src/cli/agent-relay-mcp.ts | 41 +++- packages/cli/src/cli/commands/fleet.test.ts | 68 +++++- packages/cli/src/cli/commands/fleet.ts | 20 +- .../src/cli/lib/registration-metadata.test.ts | 49 ++++ .../cli/src/cli/lib/registration-metadata.ts | 51 +++++ packages/fleet/src/index.test.ts | 69 +++++- packages/fleet/src/index.ts | 48 ++++ packages/fleet/src/serve-node.test.ts | 22 +- packages/fleet/src/serve-node.ts | 4 +- tests/e2e/fleet/fleet-e2e.test.ts | 88 ++++++++ tests/e2e/fleet/harness.ts | 20 ++ 21 files changed, 970 insertions(+), 22 deletions(-) create mode 100644 packages/cli/src/cli/lib/registration-metadata.test.ts create mode 100644 packages/cli/src/cli/lib/registration-metadata.ts diff --git a/CHANGELOG.md b/CHANGELOG.md index 3708ad4c7..9574d9f4b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -13,6 +13,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - Passing `--workspace-key` to the local or `--ssh-host` attach path is rejected because those paths authenticate with the broker instead. - `agent-relay-broker reclaim-legacy-identity`: restore restart reclaim for one offline agent registered before identity proofs were stamped. +- Fleet spawn callers can now declare an agent's organization, project, workstream, + role, and objective through the CLI, the MCP `spawn` tool, or the Fleet DSL. Relay + publishes those fields onto the agent's Relaycast record after registration; when + `objective` is omitted the spawn task is used, and hierarchy is never inferred from + the agent's name. ### Fixed diff --git a/crates/broker/src/fleet_wire.rs b/crates/broker/src/fleet_wire.rs index 3c75c3f84..a544e7e86 100644 --- a/crates/broker/src/fleet_wire.rs +++ b/crates/broker/src/fleet_wire.rs @@ -233,6 +233,89 @@ pub struct AgentRegister { skip_serializing_if = "Option::is_none" )] pub resumable: Option, + // NOTE: declared registration metadata is deliberately NOT a field here. + // The engine parses this frame with a `.strict()` schema + // (relaycast packages/types/src/fleet-wire.ts, FleetAgentRegisterMessageSchema) + // that rejects unknown keys, and its rejection carries a freshly generated + // id, so the broker's id-keyed correlation never matches and the waiter + // stalls for the full `FLEET_AGENT_REGISTER_TIMEOUT`. Declared metadata is + // published over the HTTP agent API after registration instead — see + // `RelaycastHttpClient::publish_declared_metadata`. +} + +/// Explicit organizational identity declared by a fleet spawn and published to +/// the engine as agent metadata after registration. These fields intentionally +/// never infer hierarchy from the agent name; `objective` is the original task +/// when a narrower value was not explicitly supplied. +#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct AgentRegistrationMetadata { + #[serde(default, skip_serializing_if = "Option::is_none")] + pub organization: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub project: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub workstream: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub role: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub objective: Option, +} + +impl AgentRegistrationMetadata { + pub fn is_empty(&self) -> bool { + self.organization.is_none() + && self.project.is_none() + && self.workstream.is_none() + && self.role.is_none() + && self.objective.is_none() + } + + /// Read declared hierarchy from either a flattened spawn action or its + /// `metadata` bag. `objective` falls back only to the supplied task — the + /// caller's own brief — and never to a name-derived convention. + pub fn from_spawn_input(input: &Value, task: Option<&str>) -> Self { + let objective = Self::declared_string(input, "objective") + .or_else(|| task.and_then(non_empty_string).map(ToOwned::to_owned)); + Self { + organization: Self::declared_string(input, "organization"), + project: Self::declared_string(input, "project"), + workstream: Self::declared_string(input, "workstream"), + role: Self::declared_string(input, "role"), + objective, + } + } + + fn declared_string(input: &Value, key: &str) -> Option { + let nested_agent = input.get("agent").and_then(Value::as_object); + for record in std::iter::once(input.as_object()).chain(std::iter::once(nested_agent)) { + let Some(record) = record else { + continue; + }; + if let Some(value) = record + .get(key) + .and_then(Value::as_str) + .and_then(non_empty_string) + { + return Some(value.to_string()); + } + if let Some(value) = record + .get("metadata") + .and_then(Value::as_object) + .and_then(|metadata| metadata.get(key)) + .and_then(Value::as_str) + .and_then(non_empty_string) + { + return Some(value.to_string()); + } + } + None + } +} + +fn non_empty_string(value: &str) -> Option<&str> { + let trimmed = value.trim(); + (!trimmed.is_empty()).then_some(trimmed) } #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] @@ -664,9 +747,9 @@ mod tests { use super::{ validate_agent_register_reply_data, validate_finite_nonnegative_f64, ActionResult, - ActionResultError, ActionResultPayload, AgentRegister, BrokerToRelaycast, Deliver, - DeliveryMode, Error, FleetCapability, NodeHeartbeat, RelaycastToBroker, Reply, - FLEET_WIRE_VERSION, + ActionResultError, ActionResultPayload, AgentRegister, AgentRegistrationMetadata, + BrokerToRelaycast, Deliver, DeliveryMode, Error, FleetCapability, NodeHeartbeat, + RelaycastToBroker, Reply, FLEET_WIRE_VERSION, }; #[test] @@ -691,6 +774,75 @@ mod tests { ); } + /// Regression guard for the whole reason declared metadata is published over + /// HTTP rather than on this frame. The engine parses `agent.register` with a + /// `.strict()` schema whose only keys are v/id/name/invocation_id/ + /// session_ref/resumable; ANY additional key makes it reject the frame with + /// a freshly generated id, which the broker's id-keyed correlation cannot + /// match, so the registration waiter stalls for the full 30s + /// `FLEET_AGENT_REGISTER_TIMEOUT`. Assert the serialized key set exactly, so + /// re-adding a field here fails loudly instead of surfacing as a timeout. + #[test] + fn agent_register_carries_no_keys_the_engine_schema_rejects() { + let msg = BrokerToRelaycast::AgentRegister(AgentRegister { + v: FLEET_WIRE_VERSION, + id: Some("register-1".to_string()), + name: "fleet-worker".to_string(), + invocation_id: Some("inv-1".to_string()), + session_ref: Some("sess-1".to_string()), + resumable: Some(true), + }); + + let value = serde_json::to_value(msg).unwrap(); + let mut keys: Vec<&str> = value + .as_object() + .expect("agent.register serializes to an object") + .keys() + .map(String::as_str) + .collect(); + keys.sort_unstable(); + assert_eq!( + keys, + vec![ + "id", + "invocation_id", + "name", + "resumable", + "session_ref", + "type", + "v", + ] + ); + } + + #[test] + fn spawn_metadata_preserves_explicit_objective_and_falls_back_to_task() { + let declared = AgentRegistrationMetadata::from_spawn_input( + &json!({ + "metadata": { + "organization": "Agent Workforce", + "project": "relay", + "workstream": "fleet", + "role": "implementation", + "objective": "Ship fleet metadata" + } + }), + Some("The wider initial brief"), + ); + assert_eq!(declared.objective.as_deref(), Some("Ship fleet metadata")); + assert_eq!(declared.organization.as_deref(), Some("Agent Workforce")); + + let fallback = AgentRegistrationMetadata::from_spawn_input( + &json!({"agent": {"metadata": {"project": "relay"}}}), + Some(" Publish declared registration metadata "), + ); + assert_eq!(fallback.project.as_deref(), Some("relay")); + assert_eq!( + fallback.objective.as_deref(), + Some("Publish declared registration metadata") + ); + } + #[test] fn action_result_allows_error_payloads() { let msg = BrokerToRelaycast::ActionResult(ActionResult { diff --git a/crates/broker/src/listen_api.rs b/crates/broker/src/listen_api.rs index f218ca654..7d38c18b1 100644 --- a/crates/broker/src/listen_api.rs +++ b/crates/broker/src/listen_api.rs @@ -11,6 +11,7 @@ use std::{ }; use crate::{ + fleet_wire::AgentRegistrationMetadata, ids::{ ChannelName, DeliveryId, MessageTarget, ThreadId, WorkerName, WorkspaceAlias, WorkspaceId, }, @@ -45,6 +46,7 @@ pub enum ListenApiRequest { model: Option, args: Vec, task: Option, + registration_metadata: AgentRegistrationMetadata, channels: Vec, cwd: Option, team: Option, @@ -1000,6 +1002,7 @@ async fn listen_api_spawn( }) .unwrap_or_default(); let task = body.get("task").and_then(Value::as_str).map(String::from); + let registration_metadata = AgentRegistrationMetadata::from_spawn_input(&body, task.as_deref()); let channels: Vec = body .get("channels") .and_then(Value::as_array) @@ -1112,6 +1115,7 @@ async fn listen_api_spawn( model, args, task, + registration_metadata, channels: channels.into_iter().map(ChannelName::from).collect(), cwd, team, @@ -3699,8 +3703,8 @@ mod auth_tests { use tower::ServiceExt; use super::{ - listen_api_router_with_auth, DeliveryRouteError, ListenApiConfig, ListenApiRequest, - PtyInputFrame, SetInboundDeliveryModeOk, + listen_api_router_with_auth, AgentRegistrationMetadata, DeliveryRouteError, + ListenApiConfig, ListenApiRequest, PtyInputFrame, SetInboundDeliveryModeOk, }; use crate::ids::{EventId, MessageTarget, ThreadId, WorkspaceAlias, WorkspaceId}; use crate::protocol::MessageInjectionMode; @@ -3868,6 +3872,7 @@ mod auth_tests { model, args, task, + registration_metadata, channels, cwd, team, @@ -3890,6 +3895,16 @@ mod auth_tests { assert_eq!(model.as_deref(), Some("o3")); assert_eq!(args, vec!["--fast".to_string()]); assert_eq!(task.as_deref(), Some("Ship it")); + assert_eq!( + registration_metadata, + AgentRegistrationMetadata { + organization: Some("Agent Workforce".to_string()), + project: Some("Relay".to_string()), + workstream: Some("fleet-metadata".to_string()), + role: Some("implementation".to_string()), + objective: Some("Publish registration metadata".to_string()), + } + ); assert_eq!( channels, vec!["general".to_string(), "engineering".to_string()] @@ -3929,6 +3944,13 @@ mod auth_tests { "model": "o3", "args": ["--fast"], "task": "Ship it", + "metadata": { + "organization": "Agent Workforce", + "project": "Relay", + "workstream": "fleet-metadata", + "role": "implementation", + "objective": "Publish registration metadata" + }, "channels": ["general", "engineering"], "cwd": "/tmp/project", "team": "core", diff --git a/crates/broker/src/relaycast/ws.rs b/crates/broker/src/relaycast/ws.rs index ac3fb8ebd..9d9383fb7 100644 --- a/crates/broker/src/relaycast/ws.rs +++ b/crates/broker/src/relaycast/ws.rs @@ -7,11 +7,11 @@ use relaycast::{ AgentClient, AgentRegistrationClient, AgentRegistrationError, AgentRegistrationRetryOutcome, CompleteInvocationRequest, CreateObserverTokenRequest, EmitSessionEventRequest, MessageListQuery, ObserverToken, RegisterActionRequest, RelayCast, RelayCastOptions, - RelayError, ReleaseAgentRequest, + RelayError, ReleaseAgentRequest, UpdateAgentRequest, }; use serde_json::Value; -use crate::protocol::MessageInjectionMode; +use crate::{fleet_wire::AgentRegistrationMetadata, protocol::MessageInjectionMode}; #[derive(Debug, Clone)] pub enum WsControl { @@ -143,6 +143,60 @@ impl RelaycastHttpClient { .await } + /// Publish caller-declared workforce metadata onto an already-registered + /// agent, merging it over whatever the engine already holds. + /// + /// This is how declared metadata reaches the engine on the node + /// registration path. It deliberately does NOT ride the `agent.register` + /// frame: that frame is parsed by a `.strict()` schema which rejects unknown + /// keys, and the rejection stalls the registration waiter for 30s (see the + /// note on `fleet_wire::AgentRegister`). The REST agent API has no such + /// restriction and already accepts a metadata bag, so the observable + /// outcome is the same over a transport every engine accepts. + /// + /// Callers treat this as best-effort: the agent is registered and running + /// either way, so a failure here must be logged, not fatal. + pub async fn publish_declared_metadata( + &self, + agent_name: &str, + declared: &AgentRegistrationMetadata, + ) -> std::result::Result<(), RelaycastRegistrationError> { + let name = agent_name.trim(); + if name.is_empty() { + return Err(RelaycastRegistrationError::InvalidAgentName); + } + let declared_metadata = declared_metadata_map(declared); + if declared_metadata.is_empty() { + return Ok(()); + } + let relay = self + .relay_client() + .ok_or_else(|| RelaycastRegistrationError::Transport { + agent_name: name.to_string(), + detail: "SDK relay client not initialized".to_string(), + })?; + // Read-merge-write rather than a bare overwrite: the engine owns other + // keys on this record (the `fleet` placement record among them) and a + // replacing update would drop them. + let existing = relay + .get_agent(name) + .await + .map_err(|error| registration_metadata_error(name, error))?; + let mut merged = existing.metadata; + merged.extend(declared_metadata); + relay + .update_agent( + name, + UpdateAgentRequest { + metadata: Some(merged), + ..Default::default() + }, + ) + .await + .map_err(|error| registration_metadata_error(name, error))?; + Ok(()) + } + async fn registered_agent_client(&self) -> Result { let registration = self .registration @@ -812,13 +866,66 @@ pub async fn retry_agent_registration( sdk_retry_agent_registration(registration, name, cli).await } +/// The declared fields alone, trimmed, with blanks omitted. +/// +/// Omitting rather than sending `""` matters because both callers merge this +/// over metadata the engine already holds: an empty value would overwrite an +/// engine-owned field with nothing. +fn declared_metadata_map(declared: &AgentRegistrationMetadata) -> serde_json::Map { + let mut metadata = serde_json::Map::new(); + let declared_fields = [ + ("organization", declared.organization.as_deref()), + ("project", declared.project.as_deref()), + ("workstream", declared.workstream.as_deref()), + ("role", declared.role.as_deref()), + ("objective", declared.objective.as_deref()), + ]; + for (key, value) in declared_fields { + let Some(value) = value.map(str::trim).filter(|value| !value.is_empty()) else { + continue; + }; + metadata.insert(key.to_string(), Value::String(value.to_string())); + } + metadata +} + +fn registration_metadata_error(agent_name: &str, error: RelayError) -> RelaycastRegistrationError { + match error { + RelayError::Api { + status: 429, + message, + code, + } => RelaycastRegistrationError::RateLimited { + agent_name: agent_name.to_string(), + retry_after_secs: 60, + detail: format!("{message} (code: {code})"), + }, + RelayError::Api { + status, + message, + code, + } => RelaycastRegistrationError::Api { + agent_name: agent_name.to_string(), + status, + detail: format!("{message} (code: {code})"), + }, + error => RelaycastRegistrationError::Transport { + agent_name: agent_name.to_string(), + detail: error.to_string(), + }, + } +} + #[cfg(test)] mod tests { - use httpmock::{Method::POST, MockServer}; + use httpmock::{ + Method::{GET, PATCH, POST}, + MockServer, + }; use relaycast::AgentRegistrationError; use serde_json::json; - use crate::ids::ChannelName; + use crate::{fleet_wire::AgentRegistrationMetadata, ids::ChannelName}; use super::{ format_worker_preregistration_error, registration_is_retryable, @@ -858,6 +965,103 @@ mod tests { assert!(message.contains("pre-register")); } + /// The node path's replacement for putting metadata on `agent.register`: + /// read the agent, merge the declared fields over what the engine already + /// holds, write it back. The PATCH body is matched exactly, so dropping an + /// engine-owned key would fail here. + #[tokio::test] + async fn publish_declared_metadata_merges_over_engine_owned_fields() { + let server = MockServer::start(); + let existing = server.mock(|when, then| { + when.method(GET).path("/v1/agents/worker-a"); + then.status(200).json_body(json!({ + "ok": true, + "data": { + "id": "agent_worker_a", + "name": "worker-a", + "type": "agent", + "status": "online", + "persona": null, + "metadata": { "cli": "codex", "fleet": { "node": "sf-mini" } } + } + })); + }); + let update = server.mock(|when, then| { + when.method(PATCH) + .path("/v1/agents/worker-a") + .json_body(json!({ + "metadata": { + "cli": "codex", + "fleet": { "node": "sf-mini" }, + "organization": "Agent Workforce", + "project": "Relay", + "objective": "Publish registration metadata" + } + })); + then.status(200).json_body(json!({ + "ok": true, + "data": { + "id": "agent_worker_a", + "name": "worker-a", + "type": "agent", + "status": "online", + "persona": null, + "metadata": {} + } + })); + }); + + let client = seeded_http_client(&server.base_url()); + client + .publish_declared_metadata( + "worker-a", + &AgentRegistrationMetadata { + organization: Some("Agent Workforce".to_string()), + project: Some(" Relay ".to_string()), + workstream: Some(" ".to_string()), + role: None, + objective: Some("Publish registration metadata".to_string()), + }, + ) + .await + .expect("publishing declared metadata should succeed"); + + existing.assert_hits(1); + update.assert_hits(1); + } + + /// Must-not-fire: nothing declared means no request at all. A spawn that + /// declares nothing should not pay for a read-modify-write, and must not + /// rewrite the agent's metadata with what it happens to already hold. + #[tokio::test] + async fn publish_declared_metadata_makes_no_request_when_nothing_is_declared() { + let server = MockServer::start(); + let any_read = server.mock(|when, then| { + when.method(GET).path("/v1/agents/worker-a"); + then.status(500); + }); + let any_write = server.mock(|when, then| { + when.method(PATCH).path("/v1/agents/worker-a"); + then.status(500); + }); + + let client = seeded_http_client(&server.base_url()); + client + .publish_declared_metadata( + "worker-a", + &AgentRegistrationMetadata { + organization: Some(String::new()), + project: Some(" ".to_string()), + ..Default::default() + }, + ) + .await + .expect("an empty declaration is a no-op, not an error"); + + any_read.assert_hits(0); + any_write.assert_hits(0); + } + #[tokio::test] async fn emit_agent_event_records_canonical_payload() { let server = MockServer::start(); diff --git a/crates/broker/src/runtime/api.rs b/crates/broker/src/runtime/api.rs index 7c3302f6c..ab33466aa 100644 --- a/crates/broker/src/runtime/api.rs +++ b/crates/broker/src/runtime/api.rs @@ -1,5 +1,6 @@ use super::fleet::try_send_terminal; use super::*; +use crate::relaycast::retry_agent_registration; use crate::terminal_control::TerminalToCloud; use relaycast::{ CreateObserverTokenRequest, ObserverScope, ObserverToken, ObserverTokenFilters, RelayError, @@ -293,6 +294,7 @@ impl BrokerRuntime { model, args, task, + registration_metadata, channels, cwd, team, @@ -391,6 +393,11 @@ impl BrokerRuntime { worker = %name, "bound agent to node via agent.register for HTTP spawn" ); + super::fleet::spawn_declared_metadata_publish( + relaycast_http, + name.as_str(), + registration_metadata, + ); let relay_key = token.token.clone(); fleet_registration = Some((token, None, session_ref)); Some(relay_key) @@ -401,9 +408,20 @@ impl BrokerRuntime { error = %node_error, "node agent.register unavailable; falling back to HTTP pre-registration" ); + // The ordinary cache-aware registration: it honours + // the SDK's cached token and rate-limit block, so a + // name already seeded by preflight or an earlier + // spawn is reused rather than re-created. Declared + // metadata is published separately, exactly as on + // the node path. match retry_agent_registration(relaycast_http, &name, Some(&cli)).await { Ok(token) => { + super::fleet::spawn_declared_metadata_publish( + relaycast_http, + name.as_str(), + registration_metadata, + ); // HTTP registration alone leaves the agent // without a node binding; the engine only // delivers to `via_node` agents in node-only diff --git a/crates/broker/src/runtime/fleet.rs b/crates/broker/src/runtime/fleet.rs index df6f50ef3..785539c16 100644 --- a/crates/broker/src/runtime/fleet.rs +++ b/crates/broker/src/runtime/fleet.rs @@ -2,8 +2,8 @@ use super::*; use crate::{ fleet_wire::{ ActionInvoke, ActionResult, ActionResultError, ActionResultOutput, ActionResultPayload, - AgentDeregister, AgentRegister, BrokerToRelaycast, Deliver, DeliveryMode, - RelaycastToBroker, FLEET_WIRE_VERSION, + AgentDeregister, AgentRegister, AgentRegistrationMetadata, BrokerToRelaycast, Deliver, + DeliveryMode, RelaycastToBroker, FLEET_WIRE_VERSION, }, listen_api::{DeliveryRouteError, ListenApiRequest, SetInboundDeliveryModeOk}, node_control::{delivery_ack, handler_unavailable_result, DeliveryDecision}, @@ -1450,6 +1450,46 @@ pub(super) async fn flush_pending_relay_messages( result } +/// Publish a spawn's declared workforce metadata onto the freshly registered +/// agent, on its own task. +/// +/// Detached on purpose. Both callers run inside the runtime event loop's +/// `handle_api_request`/spawn await, and anything awaited there stops the loop +/// answering API requests or observing SIGTERM for the duration. Registration +/// already carries that cost; a metadata field must not add to it. +/// +/// Best-effort by design: the agent is registered and running whether or not +/// this lands, so a failure here must never fail the spawn. It is not silent +/// 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. +pub(super) fn spawn_declared_metadata_publish( + relaycast_http: &RelaycastHttpClient, + name: &str, + declared: AgentRegistrationMetadata, +) { + if declared.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 { + Ok(()) => tracing::debug!( + worker = %agent, + "published declared workforce metadata for spawned agent" + ), + Err(error) => tracing::error!( + 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" + ), + } + }); +} + /// 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/mod.rs b/crates/broker/src/runtime/mod.rs index 7c5117aa0..d32a1dce3 100644 --- a/crates/broker/src/runtime/mod.rs +++ b/crates/broker/src/runtime/mod.rs @@ -37,9 +37,9 @@ use crate::{ }, relaycast::{ agent_identity_key, format_worker_preregistration_error, identity_key_fingerprint, - reclaim_legacy_identity, registration_retry_after_secs, retry_agent_registration, - stable_node_identity_key, AuthClient, MultiWorkspaceSession, RegRetryOutcome, - RelaycastHttpClient, WorkspaceInboundMessage, WorkspaceMembershipSummary, WsControl, + reclaim_legacy_identity, registration_retry_after_secs, stable_node_identity_key, + AuthClient, MultiWorkspaceSession, RegRetryOutcome, RelaycastHttpClient, + WorkspaceInboundMessage, WorkspaceMembershipSummary, WsControl, }, replay_buffer::{ReplayBuffer, DEFAULT_REPLAY_CAPACITY}, telemetry::{ActionSource, TelemetryClient, TelemetryEvent}, diff --git a/crates/broker/src/runtime/relaycast_events.rs b/crates/broker/src/runtime/relaycast_events.rs index 64f0d9e90..9aa9c40d8 100644 --- a/crates/broker/src/runtime/relaycast_events.rs +++ b/crates/broker/src/runtime/relaycast_events.rs @@ -490,6 +490,8 @@ pub(super) async fn spawn_worker_from_request( // the worker MCP never re-registers over HTTP. Falls back to HTTP // pre-registration when node binding is unavailable. let mut fleet_registration = None; + let registration_metadata = + crate::fleet_wire::AgentRegistrationMetadata::from_spawn_input(ws_value, task.as_deref()); let worker_relay_key = { if let Some(token) = relaycast_ws_spawn_token(ws_value) .filter(|_| !require_node_registration && !relaycast_spawn_verifies_ready(ws_value)) @@ -531,6 +533,11 @@ 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(); fleet_registration = Some((token, invocation_id.clone(), session_ref.clone())); Some(relay_key) @@ -557,6 +564,14 @@ pub(super) async fn spawn_worker_from_request( .await { Ok(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" diff --git a/packages/cli/src/cli/agent-relay-mcp.protocol.test.ts b/packages/cli/src/cli/agent-relay-mcp.protocol.test.ts index 8231309bc..ff64c141f 100644 --- a/packages/cli/src/cli/agent-relay-mcp.protocol.test.ts +++ b/packages/cli/src/cli/agent-relay-mcp.protocol.test.ts @@ -18,6 +18,16 @@ describe('Agent Relay MCP initialization', () => { 'Existing Relay participants are not local or built-in subagents' ); expect(client.getInstructions()).toContain('"send_dm"'); + + const tools = await client.listTools(); + const spawn = tools.tools.find((tool) => tool.name === 'spawn'); + expect(spawn?.inputSchema.properties).toMatchObject({ + organization: { type: 'string' }, + project: { type: 'string' }, + workstream: { type: 'string' }, + role: { type: 'string' }, + objective: { type: 'string' }, + }); } finally { await client.close(); await server.close(); diff --git a/packages/cli/src/cli/agent-relay-mcp.startup.test.ts b/packages/cli/src/cli/agent-relay-mcp.startup.test.ts index 70abe205f..3563372f9 100644 --- a/packages/cli/src/cli/agent-relay-mcp.startup.test.ts +++ b/packages/cli/src/cli/agent-relay-mcp.startup.test.ts @@ -570,6 +570,10 @@ describe('createAgentRelayMcpServer', () => { task: 'Implement a fix', channel: 'general', target_node: 'node-a', + organization: 'Agent Workforce', + project: 'Relay', + workstream: 'fleet-metadata', + role: 'implementer', }); expect(spawnResult.structuredContent.invocation).toEqual({ invocationId: 'inv_1', @@ -580,6 +584,11 @@ describe('createAgentRelayMcpServer', () => { task: 'Implement a fix', target_node: 'node-a', channels: ['general'], + organization: 'Agent Workforce', + project: 'Relay', + workstream: 'fleet-metadata', + role: 'implementer', + objective: 'Implement a fix', }, }); @@ -589,6 +598,10 @@ describe('createAgentRelayMcpServer', () => { task: 'Fix the sync', cwd: '/workspace/project', target_node: 'node-a', + organization: 'Agent Workforce', + project: 'Relay', + workstream: 'fleet-metadata', + role: 'integration specialist', }); expect(personaSpawnResult.structuredContent.invocation).toEqual({ invocationId: 'inv_1', @@ -603,6 +616,11 @@ describe('createAgentRelayMcpServer', () => { task: 'Fix the sync', cwd: '/workspace/project', target_node: 'node-a', + organization: 'Agent Workforce', + project: 'Relay', + workstream: 'fleet-metadata', + role: 'integration specialist', + objective: 'Fix the sync', }); const toolsList = await server.listToolsHandler?.({}, {}); // Assert protocol-level tool discovery: the wrapped tools/list response diff --git a/packages/cli/src/cli/agent-relay-mcp.ts b/packages/cli/src/cli/agent-relay-mcp.ts index dae030715..56535c754 100644 --- a/packages/cli/src/cli/agent-relay-mcp.ts +++ b/packages/cli/src/cli/agent-relay-mcp.ts @@ -20,6 +20,7 @@ import { isInvalidAgentTokenError, } from '@agent-relay/sdk'; import { z } from 'zod'; +import { declaredWorkforceMetadata } from './lib/registration-metadata.js'; import { initTelemetry, shutdown as shutdownTelemetry } from './telemetry/index.js'; import { RealtimeResourceBridge, SubscriptionManager, registerResourceDefinitions } from './mcp/resources.js'; import { jsonContent, jsonResult, textContent } from './mcp/tool-results.js'; @@ -533,6 +534,11 @@ type SpawnToolRequest = { channel?: string; channels?: string[]; model?: string; + organization?: string; + project?: string; + workstream?: string; + role?: string; + objective?: string; sessionRef?: string; targetNode?: string; }; @@ -565,6 +571,11 @@ function buildSpawnActionInput({ channel, channels, model, + organization, + project, + workstream, + role, + objective, sessionRef, targetNode, }: SpawnToolRequest): Record { @@ -575,6 +586,7 @@ function buildSpawnActionInput({ ...(task ? { task } : {}), ...(persona && cwd ? { cwd } : {}), ...(model ? { model } : {}), + ...declaredWorkforceMetadata({ organization, project, workstream, role, objective }, task), ...(sessionRef ? { session_ref: sessionRef } : {}), ...(targetNode ? { target_node: targetNode } : {}), ...(selectedChannels ? { channels: selectedChannels } : {}), @@ -921,6 +933,11 @@ function registerAgentRelayTools( channel: z.string().optional().describe('Channel to join'), channels: z.array(z.string()).optional().describe('Channels to join'), model: z.string().optional().describe('Model powering the worker'), + organization: z.string().optional().describe('Declared organization for workforce reporting'), + project: z.string().optional().describe('Declared project for workforce reporting'), + workstream: z.string().optional().describe('Declared workstream for workforce reporting'), + role: z.string().optional().describe('Declared role for workforce reporting'), + objective: z.string().optional().describe('Declared objective; defaults to task when omitted'), session_ref: z.string().optional().describe('Session reference for resumable spawns'), target_node: z.string().optional().describe('Optional target fleet node name'), ...identityOverrideInputShape, @@ -933,7 +950,24 @@ function registerAgentRelayTools( openWorldHint: true, }, }, - async ({ name, cli, persona, task, cwd, channel, channels, model, session_ref, target_node, as }) => { + async ({ + name, + cli, + persona, + task, + cwd, + channel, + channels, + model, + organization, + project, + workstream, + role, + objective, + session_ref, + target_node, + as, + }) => { const actions = requireSpawnActions(getAgentClient(as)); const request = { name, @@ -944,6 +978,11 @@ function registerAgentRelayTools( channel, channels, model, + organization, + project, + workstream, + role, + objective, sessionRef: session_ref, targetNode: target_node, }; diff --git a/packages/cli/src/cli/commands/fleet.test.ts b/packages/cli/src/cli/commands/fleet.test.ts index 52d1ef854..d3824a7b8 100644 --- a/packages/cli/src/cli/commands/fleet.test.ts +++ b/packages/cli/src/cli/commands/fleet.test.ts @@ -3,6 +3,7 @@ import os from 'node:os'; import path from 'node:path'; import { Command } from 'commander'; +import { defineNode, invokeNodeHandler, spawn as fleetSpawn } from '@agent-relay/fleet'; import { describe, expect, it, vi } from 'vitest'; // `fleet status` fetches the broker session (which carries the node token and @@ -11,7 +12,11 @@ import { describe, expect, it, vi } from 'vitest'; vi.mock('../lib/broker-lifecycle.js', () => ({ readBrokerConnection: vi.fn(() => ({ url: 'http://127.0.0.1:1', api_key: 'k', pid: 1, port: 1 })), })); -vi.mock('@agent-relay/harness-driver', () => ({ +// Only the driver client is stubbed. The rest of the module stays real so the +// fleet spawn handler can resolve a static harness config through the same code +// path a node runs. +vi.mock('@agent-relay/harness-driver', async (importOriginal) => ({ + ...(await importOriginal()), HarnessDriverClient: class { async getSession() { return { @@ -419,6 +424,14 @@ describe('fleet command support', () => { 'general', '--model', 'gpt-5', + '--organization', + 'Agent Workforce', + '--project', + 'Relay', + '--workstream', + 'fleet-metadata', + '--role', + 'implementation', '--session-ref', 'session-1', '--workspace-key', @@ -444,9 +457,43 @@ describe('fleet command support', () => { task: 'ACK and wait', channels: ['general'], model: 'gpt-5', + organization: 'Agent Workforce', + project: 'Relay', + workstream: 'fleet-metadata', + role: 'implementation', + objective: 'ACK and wait', session_ref: 'session-1', }, }); + // Exercise the actual two-package boundary: the CLI's targeted-placement + // input must be readable by the Fleet DSL spawn handler, which forwards it + // to the broker's registration path. + const fleetNode = defineNode({ + name: 'sf-mini', + capabilities: { + 'spawn:codex': fleetSpawn({ runtime: 'pty', command: 'codex' }), + }, + }); + const spawnAgent = vi.fn(async () => undefined); + // `mock.calls` is an array of argument lists, so the request object is the + // first argument of the first call — not the first call itself. + const [[{ input: handlerInput }]] = placement.spawn.mock.calls; + await invokeNodeHandler(fleetNode, 'spawn:codex', handlerInput, { + node: { name: fleetNode.name, capabilities: Object.keys(fleetNode.capabilities) }, + relay: { sendMessage: vi.fn() }, + spawnAgent, + }); + expect(spawnAgent).toHaveBeenCalledWith( + expect.objectContaining({ + registrationMetadata: { + organization: 'Agent Workforce', + project: 'Relay', + workstream: 'fleet-metadata', + role: 'implementation', + objective: 'ACK and wait', + }, + }) + ); expect(createFleetWorkspaceClient).not.toHaveBeenCalled(); expect(JSON.parse(logs[0]!)).toMatchObject({ invocation: { invocationId: 'inv_targeted' }, @@ -489,6 +536,16 @@ describe('fleet command support', () => { 'Reviewer', '--model', 'gpt-5', + '--organization', + 'Agent Workforce', + '--project', + 'Relay', + '--workstream', + 'fleet-metadata', + '--role', + 'reviewer', + '--objective', + 'Review the fleet change', '--workspace-key', 'rk_live_test', ], @@ -506,7 +563,14 @@ describe('fleet command support', () => { task: 'Review the diff', channel: 'general', persona: 'Reviewer', - metadata: { model: 'gpt-5' }, + metadata: { + model: 'gpt-5', + organization: 'Agent Workforce', + project: 'Relay', + workstream: 'fleet-metadata', + role: 'reviewer', + objective: 'Review the fleet change', + }, }); expect(JSON.parse(logs[0]!)).toEqual({ invocation: { invocation_id: 'inv_auto', status: 'accepted' }, diff --git a/packages/cli/src/cli/commands/fleet.ts b/packages/cli/src/cli/commands/fleet.ts index 066682ea4..52f098f2a 100644 --- a/packages/cli/src/cli/commands/fleet.ts +++ b/packages/cli/src/cli/commands/fleet.ts @@ -4,6 +4,7 @@ import { createWorkspaceClient, type RelayWorkspaceThinClient } from '@agent-rel import { withDefaults, type CoreDependencies } from './core.js'; import { readBrokerConnection } from '../lib/broker-lifecycle.js'; +import { declaredWorkforceMetadata } from '../lib/registration-metadata.js'; import { redactSecrets } from '../lib/redact.js'; import { resolveAgentToken, @@ -119,6 +120,11 @@ export function registerFleetCommands( .option('--channel ', 'Channel for the worker to join') .option('--persona ', 'Worker persona (automatic placement)') .option('--model ', 'Model powering the worker') + .option('--organization ', 'Declared organization for workforce reporting') + .option('--project ', 'Declared project for workforce reporting') + .option('--workstream ', 'Declared workstream for workforce reporting') + .option('--role ', 'Declared role for workforce reporting') + .option('--objective ', 'Declared objective (defaults to --task when omitted)') .option('--session-ref ', 'Session reference for a resumable targeted spawn') ).action(async (cli: string, options: Record) => { await runSdk(deps.sdk, async () => { @@ -130,7 +136,16 @@ export function registerFleetCommands( optionalText(options.targetNode, 'Target node') ?? optionalText(options.node, 'Node'); const channel = optionalText(options.channel, 'Channel'); const model = optionalText(options.model, 'Model'); + const organization = optionalText(options.organization, 'Organization'); + const project = optionalText(options.project, 'Project'); + const workstream = optionalText(options.workstream, 'Workstream'); + const role = optionalText(options.role, 'Role'); + const objective = optionalText(options.objective, 'Objective'); const sessionRef = optionalText(options.sessionRef, 'Session reference'); + const registrationMetadata = declaredWorkforceMetadata( + { organization, project, workstream, role, objective }, + task + ); if (targetNode) { if (!resolveAgentToken(clientOptions)) { @@ -149,6 +164,7 @@ export function registerFleetCommands( task, ...(channel ? { channels: [channel] } : {}), ...(model ? { model } : {}), + ...registrationMetadata, ...(sessionRef ? { session_ref: sessionRef } : {}), }, }); @@ -167,7 +183,9 @@ export function registerFleetCommands( task, ...(channel ? { channel } : {}), ...(persona ? { persona } : {}), - ...(model ? { metadata: { model } } : {}), + ...(model || Object.keys(registrationMetadata).length > 0 + ? { metadata: { ...(model ? { model } : {}), ...registrationMetadata } } + : {}), }); printJson(deps.sdk, { invocation }); }); diff --git a/packages/cli/src/cli/lib/registration-metadata.test.ts b/packages/cli/src/cli/lib/registration-metadata.test.ts new file mode 100644 index 000000000..4c923e4ad --- /dev/null +++ b/packages/cli/src/cli/lib/registration-metadata.test.ts @@ -0,0 +1,49 @@ +import { describe, expect, it } from 'vitest'; + +import { declaredWorkforceMetadata } from './registration-metadata.js'; + +describe('declaredWorkforceMetadata', () => { + it('trims declared fields and omits blank ones', () => { + expect( + declaredWorkforceMetadata({ + organization: ' Agent Workforce ', + project: '', + workstream: ' ', + role: 'implementer', + objective: 'Publish registration metadata', + }) + ).toEqual({ + organization: 'Agent Workforce', + role: 'implementer', + objective: 'Publish registration metadata', + }); + }); + + it('falls back to the task for objective, and prefers an explicit one', () => { + expect(declaredWorkforceMetadata({}, 'the initial brief')).toEqual({ + objective: 'the initial brief', + }); + expect(declaredWorkforceMetadata({ objective: 'ship it' }, 'the initial brief')).toEqual({ + objective: 'ship it', + }); + }); + + // `??` alone would keep the blank, suppress the fallback, and then drop the + // key at the trim step — leaving no objective at all. The broker filters + // blanks before its own fallback, so these two surfaces must agree. + it('treats a blank objective as absent so the task fallback still applies', () => { + expect(declaredWorkforceMetadata({ objective: ' ' }, 'the initial brief')).toEqual({ + objective: 'the initial brief', + }); + expect(declaredWorkforceMetadata({ objective: ' ' }, ' ')).toEqual({}); + }); + + // The guard callers rely on is `Object.keys(...).length > 0`, so "nothing + // declared" has to produce a genuinely empty object — an `objective: + // undefined` key would make that guard always true. + it('returns an empty object when nothing is declared', () => { + const metadata = declaredWorkforceMetadata({ organization: ' ' }, ' '); + expect(metadata).toEqual({}); + expect(Object.keys(metadata)).toHaveLength(0); + }); +}); diff --git a/packages/cli/src/cli/lib/registration-metadata.ts b/packages/cli/src/cli/lib/registration-metadata.ts new file mode 100644 index 000000000..36816c329 --- /dev/null +++ b/packages/cli/src/cli/lib/registration-metadata.ts @@ -0,0 +1,51 @@ +/** + * Caller-declared workforce identity attached to a spawn. + * + * One assembly shared by every surface that can start an agent — the MCP + * `spawn` tool and `agent-relay fleet spawn` — so the declared semantics cannot + * drift between them. It mirrors the broker's `AgentRegistrationMetadata` + * (`crates/broker/src/fleet_wire.rs`), including the rule that `objective` + * falls back to the spawn's task and is never derived from the agent's name. + */ +export interface DeclaredWorkforceInput { + organization?: string; + project?: string; + workstream?: string; + role?: string; + objective?: string; +} + +export type DeclaredWorkforceMetadata = Partial>; + +/** + * Build the declared metadata bag, omitting anything blank. + * + * Every key is conditional, `objective` included: emitting `objective: + * undefined` would leave the object non-empty when nothing was declared, so + * callers that branch on `Object.keys(...).length` would always take the + * "something was declared" path. + */ +export function declaredWorkforceMetadata( + input: DeclaredWorkforceInput, + task?: string +): DeclaredWorkforceMetadata { + // A blank declared objective counts as "not declared", so the task fallback + // still applies. `??` alone would keep the blank, suppress the fallback, and + // then drop the key at the trim step below — leaving no objective at all. The + // broker filters blanks the same way before its fallback + // (`AgentRegistrationMetadata::from_spawn_input` via `declared_string`). + const objective = input.objective?.trim() ? input.objective : task; + const declared: Array<[keyof DeclaredWorkforceInput, string | undefined]> = [ + ['organization', input.organization], + ['project', input.project], + ['workstream', input.workstream], + ['role', input.role], + ['objective', objective], + ]; + const metadata: DeclaredWorkforceMetadata = {}; + for (const [key, value] of declared) { + const trimmed = value?.trim(); + if (trimmed) metadata[key] = trimmed; + } + return metadata; +} diff --git a/packages/fleet/src/index.test.ts b/packages/fleet/src/index.test.ts index aa9e81f16..756bb5b44 100644 --- a/packages/fleet/src/index.test.ts +++ b/packages/fleet/src/index.test.ts @@ -57,7 +57,16 @@ describe('@agent-relay/fleet', () => { await invokeNodeHandler( node, 'spawn:codex', - { name: 'worker-a', model: 'gpt-5', session_ref: 'thread-1', task: 'ship it' }, + { + name: 'worker-a', + model: 'gpt-5', + session_ref: 'thread-1', + task: 'ship it', + organization: 'AgentWorkforce', + project: 'relay', + workstream: 'fleet-metadata', + role: 'implementer', + }, ctx ); @@ -71,11 +80,69 @@ describe('@agent-relay/fleet', () => { channels: ['general'], }), initialTask: 'ship it', + registrationMetadata: { + organization: 'AgentWorkforce', + project: 'relay', + workstream: 'fleet-metadata', + role: 'implementer', + objective: 'ship it', + }, skipRelayPrompt: false, invocationId: undefined, }); }); + // Blank declared values are dropped rather than forwarded as empty strings — + // the broker merges these over metadata the engine already holds, so an empty + // value would overwrite an engine-owned field. Matches the CLI's + // `declaredWorkforceMetadata` and the broker's `declared_metadata_map`. + it('trims declared metadata and omits blank values', async () => { + const node = defineNode({ + name: 'builder', + capabilities: { + 'spawn:codex': spawn({ runtime: 'pty', command: 'codex' }), + }, + }); + const ctx = stubContext(node.name, Object.keys(node.capabilities)); + + await invokeNodeHandler( + node, + 'spawn:codex', + { + name: 'worker-a', + task: 'ship it', + // `''` cannot reach here — the spawn input schema already rejects it + // with `min(1)`. Whitespace-only is the value that gets through, so + // that is what this asserts on. + organization: ' AgentWorkforce ', + workstream: ' ', + role: ' ', + }, + ctx + ); + + expect(ctx.spawnAgent).toHaveBeenCalledWith( + expect.objectContaining({ + registrationMetadata: { organization: 'AgentWorkforce', objective: 'ship it' }, + }) + ); + }); + + it('omits registration metadata entirely when nothing is declared', async () => { + const node = defineNode({ + name: 'builder', + capabilities: { + 'spawn:codex': spawn({ runtime: 'pty', command: 'codex' }), + }, + }); + const ctx = stubContext(node.name, Object.keys(node.capabilities)); + + await invokeNodeHandler(node, 'spawn:codex', { name: 'worker-a', role: ' ' }, ctx); + + const [request] = (ctx.spawnAgent as unknown as { mock: { calls: unknown[][] } }).mock.calls[0]; + expect(request).not.toHaveProperty('registrationMetadata'); + }); + it('threads invocation ids through concurrent spawn handlers', async () => { const node = defineNode({ name: 'builder', diff --git a/packages/fleet/src/index.ts b/packages/fleet/src/index.ts index f212df830..74c4fe848 100644 --- a/packages/fleet/src/index.ts +++ b/packages/fleet/src/index.ts @@ -31,9 +31,26 @@ export interface FleetScopedRelayClient { sendMessage(input: FleetRelaySendMessageInput): Promise; } +/** + * Declared organizational identity for a spawned worker. + * + * These are intentionally explicit rather than derived from the agent name: + * names are an implementation label, whereas this data is the caller's + * statement of where the work belongs. `objective` is normally supplied by a + * spawn's initial task when callers do not provide a narrower declaration. + */ +export interface FleetAgentRegistrationMetadata { + organization?: string; + project?: string; + workstream?: string; + role?: string; + objective?: string; +} + export interface FleetSpawnAgentInput { agent: AgentSpec; initialTask?: string; + registrationMetadata?: FleetAgentRegistrationMetadata; skipRelayPrompt?: boolean; invocationId?: string; } @@ -132,6 +149,11 @@ const spawnInputSchema = z cwd: z.string().min(1).optional(), args: z.array(z.string()).optional(), team: z.string().min(1).optional(), + organization: z.string().min(1).optional(), + project: z.string().min(1).optional(), + workstream: z.string().min(1).optional(), + role: z.string().min(1).optional(), + objective: z.string().min(1).optional(), skip_relay_prompt: z.boolean().optional(), }) .refine((input) => Boolean(input.name ?? input.agent), { @@ -237,6 +259,7 @@ export function spawn( const cwd = input.cwd ?? options.cwd ?? definition.cwd; const channels = input.channels ?? options.channels; const task = input.task; + const metadata = registrationMetadata(input, task); const harnessConfig = resolveStaticHarnessConfig({ name, cli: definition.command, @@ -267,6 +290,7 @@ export function spawn( return ctx.spawnAgent({ agent, ...(task !== undefined ? { initialTask: task } : {}), + ...(metadata ? { registrationMetadata: metadata } : {}), skipRelayPrompt: input.skip_relay_prompt ?? options.skipRelayPrompt ?? false, invocationId: ctx.invocationId, }); @@ -287,6 +311,30 @@ export function spawn( }; } +function registrationMetadata( + input: Pick, + task: string | undefined +): FleetAgentRegistrationMetadata | undefined { + // Trim and omit blanks, matching the CLI's `declaredWorkforceMetadata` and + // the broker's `declared_metadata_map`. A whitespace-only value is a declared + // nothing, and the broker merges these over metadata the engine already + // holds, so emitting one would overwrite an engine-owned field with an empty + // string. + const declared: Array<[keyof FleetAgentRegistrationMetadata, string | undefined]> = [ + ['organization', input.organization], + ['project', input.project], + ['workstream', input.workstream], + ['role', input.role], + ['objective', input.objective ?? task], + ]; + const metadata: FleetAgentRegistrationMetadata = {}; + for (const [key, value] of declared) { + const trimmed = value?.trim(); + if (trimmed) metadata[key] = trimmed; + } + return Object.keys(metadata).length > 0 ? metadata : undefined; +} + export function onMessage(input: OnMessageTriggerInput, actionName: string): FleetTriggerDescriptor { return { type: 'message', diff --git a/packages/fleet/src/serve-node.test.ts b/packages/fleet/src/serve-node.test.ts index 39ae78b7c..796c2bb03 100644 --- a/packages/fleet/src/serve-node.test.ts +++ b/packages/fleet/src/serve-node.test.ts @@ -231,7 +231,14 @@ describe('serveNode', () => { type: 'action.invoke', invocation_id: 'inv_3', action: 'spawn:codex', - input: { name: 'worker-a' }, + input: { + name: 'worker-a', + task: 'Implement fleet metadata', + organization: 'AgentWorkforce', + project: 'relay', + workstream: 'fleet-metadata', + role: 'implementer', + }, }); await flush(); @@ -239,7 +246,18 @@ describe('serveNode', () => { expect(nodeSpawn).toBeTruthy(); // The delegation carries `capability: 'codex'` (from the shadow name) so the // engine keys node capacity on `spawn:codex`, plus the executable `cli`. - expect(nodeSpawn.input).toMatchObject({ name: 'worker-a', cli: 'codex', capability: 'codex' }); + expect(nodeSpawn.input).toMatchObject({ + name: 'worker-a', + cli: 'codex', + capability: 'codex', + metadata: { + organization: 'AgentWorkforce', + project: 'relay', + workstream: 'fleet-metadata', + role: 'implementer', + objective: 'Implement fleet metadata', + }, + }); // Reply so the delegating handler resolves and the invocation completes. sock.emit({ v: 1, id: nodeSpawn.id, type: 'reply', ok: true, data: { name: 'worker-a' } }); await flush(); diff --git a/packages/fleet/src/serve-node.ts b/packages/fleet/src/serve-node.ts index 078f1a141..5ad607c4f 100644 --- a/packages/fleet/src/serve-node.ts +++ b/packages/fleet/src/serve-node.ts @@ -342,7 +342,8 @@ function makeContext( /** * Shape a fleet spawn request as the engine `node.spawn` input. Spawn fields are * flattened to the top level so the broker's spawn executor reads `name`/`cli`/ - * `task`. The engine's capacity placement keys on `capability` — the harness a + * `task`; declared registration metadata remains under `metadata` for the + * subsequent `agent.register`. The engine's capacity placement keys on `capability` — the harness a * `spawn:` shadow delegates to — which is distinct from the executable * `cli` when the shadow's harness command isn't itself the harness name. */ @@ -357,6 +358,7 @@ function buildSpawnInput( return { ...spawn.agent, ...(spawn.initialTask !== undefined ? { task: spawn.initialTask } : {}), + ...(spawn.registrationMetadata ? { metadata: spawn.registrationMetadata } : {}), skip_relay_prompt: spawn.skipRelayPrompt ?? false, ...(invocationId ? { invocation_id: invocationId } : {}), ...(shadowedHarness ? { capability: shadowedHarness } : {}), diff --git a/tests/e2e/fleet/fleet-e2e.test.ts b/tests/e2e/fleet/fleet-e2e.test.ts index 23b9dd86e..92db3ed20 100644 --- a/tests/e2e/fleet/fleet-e2e.test.ts +++ b/tests/e2e/fleet/fleet-e2e.test.ts @@ -10,6 +10,7 @@ import { delay, enrollNode, FleetNode, + getAgent, getFreePort, getInvocation, getNodes, @@ -669,6 +670,93 @@ describe.skipIf(!pre.ok)('two-node fleet scenario matrix', () => { await rx.disconnect(); }, 30_000); + + // The feature this PR exists for, asserted against a live engine rather than + // a mock. It is deliberately an e2e: the declared fields do NOT ride the + // `agent.register` frame (the engine parses that one with a strict schema that + // rejects unknown keys), they are published over the HTTP agent API on a + // separate task once registration succeeds. Only a real engine proves that the + // chosen transport is one the engine actually accepts. + it('declared workforce metadata reaches the engine as agent metadata', async () => { + const agent = 'worker-declared-metadata'; + const spawn = await invokeAction(engine, driverToken, 'spawn', { + cli: 'codex', + name: agent, + target_node: 'node-b', + task: 'ack and wait', + organization: 'Agent Workforce', + project: 'Relay', + workstream: 'fleet-metadata', + role: 'implementer', + }); + expect(spawn.status).toBe(201); + const settled = await waitFor( + async () => { + const inv = await getInvocation(engine, driverToken, 'spawn', spawn.invocationId!); + return inv.status === 'completed' || inv.status === 'failed' ? inv : null; + }, + { label: 'declared-metadata spawn settled', timeoutMs: 35_000 } + ); + expect(settled.status).toBe('completed'); + + // Poll rather than read once: the publish is intentionally detached from the + // spawn so it cannot extend the broker's inline registration await. + const metadata = await waitFor( + async () => { + const record = await getAgent(engine, workspaceKey, agent); + return record?.metadata?.organization ? record.metadata : null; + }, + { label: 'declared metadata published to the engine', timeoutMs: 20_000 } + ); + + expect(metadata).toMatchObject({ + organization: 'Agent Workforce', + project: 'Relay', + workstream: 'fleet-metadata', + role: 'implementer', + // No explicit objective was declared, so it falls back to the task. + objective: 'ack and wait', + }); + }, 70_000); + + // Control for the test above: a spawn that declares nothing must not acquire + // declared fields from anywhere — no name-derived hierarchy, no leakage from + // the previous spawn's publish. + it('a spawn that declares nothing gets no declared metadata', async () => { + const agent = 'worker-undeclared-metadata'; + const spawn = await invokeAction(engine, driverToken, 'spawn', { + cli: 'codex', + name: agent, + target_node: 'node-b', + }); + expect(spawn.status).toBe(201); + const settled = await waitFor( + async () => { + const inv = await getInvocation(engine, driverToken, 'spawn', spawn.invocationId!); + return inv.status === 'completed' || inv.status === 'failed' ? inv : null; + }, + { label: 'undeclared spawn settled', timeoutMs: 35_000 } + ); + // A control arm that accepts a FAILED spawn proves nothing: no agent was + // stamped because none was started. + expect(settled.status).toBe('completed'); + + const initial = await waitFor(async () => await getAgent(engine, workspaceKey, agent), { + label: 'undeclared agent visible', + timeoutMs: 20_000, + }); + // Give any (incorrect) publish the same window the positive test relies on. + await delay(3_000); + const final = await getAgent(engine, workspaceKey, agent); + // Assert against the FINAL read, and require it to exist — falling back to + // the earlier snapshot would let a broken final read pass this test on + // stale data. + expect(final).not.toBeNull(); + for (const key of ['organization', 'project', 'workstream', 'role', 'objective']) { + expect(initial.metadata ?? {}).not.toHaveProperty(key); + expect(final!.metadata ?? {}).not.toHaveProperty(key); + } + }, 70_000); }); /** diff --git a/tests/e2e/fleet/harness.ts b/tests/e2e/fleet/harness.ts index 04b5c412f..debb7835a 100644 --- a/tests/e2e/fleet/harness.ts +++ b/tests/e2e/fleet/harness.ts @@ -667,6 +667,26 @@ export async function listMessages( return items as Array<{ text: string }>; } +/** Read one agent's engine record, including the metadata bag. + * + * Returns null ONLY for 404 — "the agent does not exist yet", which callers + * poll on. Every other failure throws: mapping auth, server, and not-found + * errors all to null makes a broken read indistinguishable from a legitimately + * absent agent, and a `waitFor` polling on null would then time out (or an + * assertion would pass) for entirely the wrong reason. */ +export async function getAgent( + engine: EngineHandle, + workspaceKey: string, + name: string +): Promise<{ name: string; metadata?: Record } | null> { + const { status, body } = await engine.fetchJson(`/v1/agents/${name}`, { + headers: { authorization: `Bearer ${workspaceKey}` }, + }); + if (status === 404) return null; + if (status >= 300) throw new Error(`getAgent(${name}) ${status}: ${JSON.stringify(body)}`); + return body.data ?? null; +} + /** Release (delete) an agent, freeing its location — used to model a resumable * agent being released before a resume re-spawn. */ export async function releaseAgent(