diff --git a/.agentworkforce/trajectories/completed/2026-09/traj_h0xx33q5a1ga/summary.md b/.agentworkforce/trajectories/completed/2026-09/traj_h0xx33q5a1ga/summary.md new file mode 100644 index 000000000..46b0d8d9d --- /dev/null +++ b/.agentworkforce/trajectories/completed/2026-09/traj_h0xx33q5a1ga/summary.md @@ -0,0 +1,65 @@ +# Trajectory: Accept Relaycast context.update node frames in the broker (relay#1615) + +> **Status:** ✅ Completed +> **Task:** 1615 +> **Confidence:** 85% +> **Started:** September 2, 2026 at 04:39 AM +> **Completed:** September 2, 2026 at 04:42 AM + +--- + +## Summary + +Broker parses Relaycast context.update node frames via a new ServerToNode::ContextUpdate variant, surfaces Relaycast delivery problems to the sending worker as message_delivery_failed, and drops the cached Relaycast registration on agent.identity_taken_over. (Review follow-up on the same PR narrowed the surfaced event to delivery.failed only: delivery.deferred is non-terminal — it stays queued for a later available_at retry — so it is now log-only.) Verified with `cargo fmt --all --check` (clean), `cargo clippy -p agent-relay-broker --all-targets` (no warnings), and `cargo test -p agent-relay-broker`: 1047 passed / 0 failed / 4 ignored in the lib unit suite, plus 12 continuity, 1 fleet_wire_fixtures and 3 journal_lock_cli integration tests, and 0 doc-tests. RelayFlow case 1615-context-update-frames was run on both compiled arms — base (origin/main) yields outcome bug / context_update_rejected_as_invalid_frame, head yields outcome fixed / context_update_accepted_and_routed. + +**Approach:** Mirror the engine's canonical context.update schema in fleet_wire (forward-compatible, no deny_unknown_fields), route the parsed frame through the existing fleet-control channel, and prove base/head behaviour with a dependency-free fake Relaycast in a RelayFlow case. + +--- + +## Key Decisions + +### Reused BrokerEvent::MessageDeliveryFailed for Relaycast delivery.failed/deferred instead of a new event kind +- **Chose:** Reused BrokerEvent::MessageDeliveryFailed for Relaycast delivery.failed/deferred instead of a new event kind +- **Reasoning:** SDK/dashboard consumers already render message_delivery_failed; a new kind would need client changes to be visible, defeating the point of surfacing the failure to the sending agent + +### Reused RelaycastHttpClient::forget_agent_registration for agent.identity_taken_over +- **Chose:** Reused RelaycastHttpClient::forget_agent_registration for agent.identity_taken_over +- **Reasoning:** That cache is the state the takeover/registration paths in relaycast/ws.rs already consult; a parallel stale flag would drift + +--- + +## Chapters + +### 1. Work +*Agent: default* + +- Reused BrokerEvent::MessageDeliveryFailed for Relaycast delivery.failed/deferred instead of a new event kind: Reused BrokerEvent::MessageDeliveryFailed for Relaycast delivery.failed/deferred instead of a new event kind +- Reused RelaycastHttpClient::forget_agent_registration for agent.identity_taken_over: Reused RelaycastHttpClient::forget_agent_registration for agent.identity_taken_over + +--- + +## Commits + +- `c919d2eeff0bf606a215141822571ca1fdbba5d3` +- `e22a7a712025bd2b325d607bff54f5a00c01cc06` + +Traced range: `6d5199ff103cb5f4ff6adf0a3fa32a788646a9bc` .. `e22a7a712025bd2b325d607bff54f5a00c01cc06` + +> This record was hand-amended (JSON + this summary edited directly): `trail` +> has no amend command for a completed trajectory, so extending the traced +> range to the review follow-up commit `e22a7a7` could not be done through the +> tool. `e22a7a7` touched no product file that `c919d2e` had not already +> touched, so **Files Changed** below is unchanged. + +## Files Changed + +- `CHANGELOG.md` +- `crates/broker/src/fleet_wire.rs` +- `crates/broker/src/node_control.rs` +- `crates/broker/src/relaycast/ws.rs` +- `crates/broker/src/runtime/fleet.rs` +- `crates/broker/src/runtime/tests.rs` +- `crates/broker/tests/fixtures/fleet-wire/context.update.json` +- `crates/broker/tests/fleet_wire_fixtures.rs` +- `tests/relayflows/cases/1615-context-update-frames/case.json` +- `tests/relayflows/cases/1615-context-update-frames/run.mjs` diff --git a/.agentworkforce/trajectories/completed/2026-09/traj_h0xx33q5a1ga/trajectory.json b/.agentworkforce/trajectories/completed/2026-09/traj_h0xx33q5a1ga/trajectory.json new file mode 100644 index 000000000..ac85b2abb --- /dev/null +++ b/.agentworkforce/trajectories/completed/2026-09/traj_h0xx33q5a1ga/trajectory.json @@ -0,0 +1,83 @@ +{ + "id": "traj_h0xx33q5a1ga", + "version": 1, + "task": { + "title": "Accept Relaycast context.update node frames in the broker (relay#1615)", + "source": { + "system": "plain", + "id": "1615" + } + }, + "status": "completed", + "startedAt": "2026-09-02T04:39:35.333Z", + "completedAt": "2026-09-02T04:42:32.494Z", + "agents": [ + { + "name": "default", + "role": "lead", + "joinedAt": "2026-09-02T04:39:43.528Z" + } + ], + "chapters": [ + { + "id": "chap_lhxpvnoz5nzx", + "title": "Work", + "agentName": "default", + "startedAt": "2026-09-02T04:39:43.528Z", + "endedAt": "2026-09-02T04:42:32.494Z", + "events": [ + { + "ts": 1788323983530, + "type": "decision", + "content": "Reused BrokerEvent::MessageDeliveryFailed for Relaycast delivery.failed/deferred instead of a new event kind: Reused BrokerEvent::MessageDeliveryFailed for Relaycast delivery.failed/deferred instead of a new event kind", + "raw": { + "question": "Reused BrokerEvent::MessageDeliveryFailed for Relaycast delivery.failed/deferred instead of a new event kind", + "chosen": "Reused BrokerEvent::MessageDeliveryFailed for Relaycast delivery.failed/deferred instead of a new event kind", + "alternatives": [], + "reasoning": "SDK/dashboard consumers already render message_delivery_failed; a new kind would need client changes to be visible, defeating the point of surfacing the failure to the sending agent" + }, + "significance": "high" + }, + { + "ts": 1788323985386, + "type": "decision", + "content": "Reused RelaycastHttpClient::forget_agent_registration for agent.identity_taken_over: Reused RelaycastHttpClient::forget_agent_registration for agent.identity_taken_over", + "raw": { + "question": "Reused RelaycastHttpClient::forget_agent_registration for agent.identity_taken_over", + "chosen": "Reused RelaycastHttpClient::forget_agent_registration for agent.identity_taken_over", + "alternatives": [], + "reasoning": "That cache is the state the takeover/registration paths in relaycast/ws.rs already consult; a parallel stale flag would drift" + }, + "significance": "high" + } + ] + } + ], + "retrospective": { + "summary": "Broker parses Relaycast context.update node frames via a new ServerToNode::ContextUpdate variant, surfaces Relaycast delivery problems to the sending worker as message_delivery_failed, and drops the cached Relaycast registration on agent.identity_taken_over. (Review follow-up on the same PR narrowed the surfaced event to delivery.failed only: delivery.deferred is non-terminal \u2014 it stays queued for a later available_at retry \u2014 so it is now log-only.) Verified with `cargo fmt --all --check` (clean), `cargo clippy -p agent-relay-broker --all-targets` (no warnings), and `cargo test -p agent-relay-broker`: 1047 passed / 0 failed / 4 ignored in the lib unit suite, plus 12 continuity, 1 fleet_wire_fixtures and 3 journal_lock_cli integration tests, and 0 doc-tests. RelayFlow case 1615-context-update-frames was run on both compiled arms \u2014 base (origin/main) yields outcome bug / context_update_rejected_as_invalid_frame, head yields outcome fixed / context_update_accepted_and_routed.", + "approach": "Mirror the engine's canonical context.update schema in fleet_wire (forward-compatible, no deny_unknown_fields), route the parsed frame through the existing fleet-control channel, and prove base/head behaviour with a dependency-free fake Relaycast in a RelayFlow case.", + "confidence": 0.85 + }, + "commits": [ + "c919d2eeff0bf606a215141822571ca1fdbba5d3", + "e22a7a712025bd2b325d607bff54f5a00c01cc06" + ], + "filesChanged": [ + "CHANGELOG.md", + "crates/broker/src/fleet_wire.rs", + "crates/broker/src/node_control.rs", + "crates/broker/src/relaycast/ws.rs", + "crates/broker/src/runtime/fleet.rs", + "crates/broker/src/runtime/tests.rs", + "crates/broker/tests/fixtures/fleet-wire/context.update.json", + "crates/broker/tests/fleet_wire_fixtures.rs", + "tests/relayflows/cases/1615-context-update-frames/case.json", + "tests/relayflows/cases/1615-context-update-frames/run.mjs" + ], + "projectId": "AgentWorkforce/relay", + "tags": [], + "_trace": { + "startRef": "6d5199ff103cb5f4ff6adf0a3fa32a788646a9bc", + "endRef": "e22a7a712025bd2b325d607bff54f5a00c01cc06" + } +} diff --git a/.trajectories/compacted/release-11.10.2.json b/.trajectories/compacted/release-11.10.2.json index 151c30136..ad207dfb6 100644 --- a/.trajectories/compacted/release-11.10.2.json +++ b/.trajectories/compacted/release-11.10.2.json @@ -3,9 +3,7 @@ "version": 1, "type": "compacted", "compactedAt": "2026-09-03T06:52:02.784Z", - "sourceTrajectories": [ - "traj_7yref3wye283" - ], + "sourceTrajectories": ["traj_7yref3wye283"], "dateRange": { "start": "2026-09-02T10:25:41.382Z", "end": "2026-09-02T12:08:45.907Z" @@ -13,9 +11,7 @@ "summary": { "totalDecisions": 7, "totalEvents": 11, - "uniqueAgents": [ - "default" - ] + "uniqueAgents": ["default"] }, "decisionGroups": [ { @@ -103,12 +99,5 @@ "tests/relayflows/cases/1638-attach-input-replay/case.json", "tests/relayflows/cases/1638-attach-input-replay/run.mjs" ], - "commits": [ - "e85f4ca23", - "bf66a2e86", - "13971aa8f", - "1191af9bd", - "cefc1ab4d", - "2559e668a" - ] -} \ No newline at end of file + "commits": ["e85f4ca23", "bf66a2e86", "13971aa8f", "1191af9bd", "cefc1ab4d", "2559e668a"] +} diff --git a/.trajectories/compacted/release-11.10.2.md b/.trajectories/compacted/release-11.10.2.md index e8680f9be..fc7dee8b6 100644 --- a/.trajectories/compacted/release-11.10.2.md +++ b/.trajectories/compacted/release-11.10.2.md @@ -1,6 +1,7 @@ # Trajectory Compaction: Sep 2, 2026 - Sep 2, 2026 ## Summary + - Sessions: 1 - Decisions: 7 - Events: 11 @@ -9,6 +10,7 @@ - Commits: 6 ## Api + - Treat Relaycast publication and recipient reachability as separate observations -> Treat Relaycast publication and recipient reachability as separate observations (traj_7yref3wye283) - Keep the reachability probe outside the publication timeout -> Keep the reachability probe outside the publication timeout (traj_7yref3wye283) - Keep the synchronous reachability snapshot bounded at five seconds -> Keep the synchronous reachability snapshot bounded at five seconds (traj_7yref3wye283) @@ -16,13 +18,17 @@ - Cancel reachability observation when publication fails -> Cancel reachability observation when publication fails (traj_7yref3wye283) ## Security + - Resolve Relaycast @self before reachability probing -> Resolve Relaycast @self before reachability probing (traj_7yref3wye283) ## Other + - Treat legacy away as reachable -> Treat legacy away as reachable (traj_7yref3wye283) ## Key Learnings + - None ## Key Findings -- None \ No newline at end of file + +- None diff --git a/CHANGELOG.md b/CHANGELOG.md index c28342cef..23c10d8cc 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,7 +5,13 @@ 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 - Minor] + +### Fixed + +- A DM to an unreachable agent is no longer silently lost: the broker now surfaces Relaycast `delivery.failed` to the sending agent as a `message_delivery_failed` event. +- Relaycast `context.update` node frames are now parsed and routed instead of being logged as invalid. +- Broker drops a worker's cached Relaycast registration when Relaycast reports `agent.identity_taken_over`, so the next operation re-registers instead of failing on a revoked token. ## [11.10.2] - 2026-09-03 diff --git a/crates/broker/src/fleet_wire.rs b/crates/broker/src/fleet_wire.rs index db105f154..acc4f21e2 100644 --- a/crates/broker/src/fleet_wire.rs +++ b/crates/broker/src/fleet_wire.rs @@ -601,6 +601,48 @@ pub struct ActionInvoke { pub agent_name: Option, } +/// Scope of an ephemeral `context.update` fan-out. Mirrors the engine's +/// `FleetContextUpdateMessageSchema` topic enum. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum ContextTopic { + Presence, + Channel, + Thread, + Agent, +} + +/// Ephemeral, best-effort context fan-out (server -> broker). Unlike +/// [`Deliver`] it is never acked and never redelivered, so a broker that +/// cannot make sense of one simply ignores it. +/// +/// Inbound (server -> broker): intentionally NOT `deny_unknown_fields` for the +/// same forward-compatibility reason as `Deliver` above — a new top-level field +/// must not make the whole frame unparseable. +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +pub struct ContextUpdate { + /// Fleet wire version the engine framed this update with. + pub v: FleetWireVersion, + /// Scope of the fan-out; decides how `event` and `data` are interpreted. + pub topic: ContextTopic, + /// Engine event name within `topic`, e.g. `delivery.failed`. Free-form on + /// purpose — an unrecognized event must be ignored, never rejected. + pub event: String, + /// Nullable on the wire (the engine always sends the key, `null` when the + /// event is not channel-scoped), so this is a plain `Option` that both + /// accepts and re-emits `null` rather than the presence-only optional used + /// by outbound frames. + #[serde(default)] + pub channel_id: Option, + /// Agents hosted on THIS node that the event concerns. The engine groups + /// its fan-out per node/provider before sending. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub agent_ids: Option>, + /// Open per-event payload. The engine adds keys over time, so readers pull + /// the ones they understand instead of deserializing a fixed shape. + pub data: Value, +} + #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] #[serde(deny_unknown_fields)] pub struct Ping { @@ -742,6 +784,8 @@ pub enum ServerToNode { Deliver(Deliver), #[serde(rename = "action.invoke")] ActionInvoke(ActionInvoke), + #[serde(rename = "context.update")] + ContextUpdate(ContextUpdate), #[serde(rename = "ping")] Ping(Ping), #[serde(rename = "reply")] @@ -760,8 +804,8 @@ mod tests { use super::{ validate_agent_register_reply_data, validate_finite_nonnegative_f64, ActionResult, ActionResultError, ActionResultPayload, AgentRegister, AgentRegistrationMetadata, - BrokerToRelaycast, Deliver, DeliveryMode, Error, FleetCapability, NodeHeartbeat, - RelaycastToBroker, Reply, FLEET_WIRE_VERSION, + BrokerToRelaycast, ContextTopic, Deliver, DeliveryMode, Error, FleetCapability, + NodeHeartbeat, RelaycastToBroker, Reply, FLEET_WIRE_VERSION, }; #[test] @@ -1325,4 +1369,68 @@ mod tests { let decoded: RelaycastToBroker = serde_json::from_value(value).unwrap(); assert_eq!(decoded, msg); } + + /// The canonical engine fixture must parse as a `context.update`, not fall + /// through to the "invalid fleet node ws frame" path. Kept in lockstep with + /// `tests/fixtures/fleet-wire/context.update.json`. + #[test] + fn parses_the_canonical_context_update_fixture() { + let raw = include_str!("../tests/fixtures/fleet-wire/context.update.json"); + let decoded: RelaycastToBroker = serde_json::from_str(raw).expect("fixture must parse"); + + let RelaycastToBroker::ContextUpdate(update) = decoded else { + panic!("context.update must decode to the ContextUpdate variant"); + }; + assert_eq!(update.topic, ContextTopic::Presence); + assert_eq!(update.event, "agent.status.active"); + assert_eq!(update.channel_id, None); + assert_eq!( + update.agent_ids.as_deref(), + Some(["agt_01J7FLEET000000000000101".to_string()].as_slice()) + ); + assert_eq!(update.data["agent_name"], "planner"); + + let encoded: Value = + serde_json::to_value(RelaycastToBroker::ContextUpdate(update)).unwrap(); + let fixture: Value = serde_json::from_str(raw).unwrap(); + assert_eq!(encoded, fixture, "context.update must round-trip verbatim"); + } + + /// Forward compatibility: an unknown top-level field (and an unknown + /// `event` string) must not make the frame unparseable, or the broker is + /// back to logging a legitimate engine frame as invalid. + #[test] + fn context_update_tolerates_unknown_fields_and_events() { + let decoded: RelaycastToBroker = serde_json::from_value(json!({ + "type": "context.update", + "v": 1, + "topic": "agent", + "event": "some.future.event", + "agent_ids": ["agt_1"], + "data": {"anything": true}, + "future_field": "ignored" + })) + .expect("unknown fields must not fail the frame"); + + let RelaycastToBroker::ContextUpdate(update) = decoded else { + panic!("expected a context.update"); + }; + assert_eq!(update.topic, ContextTopic::Agent); + assert_eq!(update.event, "some.future.event"); + } + + /// `topic` is a closed enum mirroring the engine's schema, so an unknown + /// scope is a genuine protocol mismatch and must fail loudly rather than + /// being silently coerced into one of the known topics. + #[test] + fn context_update_rejects_unknown_topics() { + assert!(serde_json::from_value::(json!({ + "type": "context.update", + "v": 1, + "topic": "galaxy", + "event": "agent.status.active", + "data": {} + })) + .is_err()); + } } diff --git a/crates/broker/src/node_control.rs b/crates/broker/src/node_control.rs index b2d5ef0f7..9bb0796cb 100644 --- a/crates/broker/src/node_control.rs +++ b/crates/broker/src/node_control.rs @@ -798,6 +798,19 @@ impl FleetDeliveryBook { .map(|binding| binding.agent_id.as_str()) } + /// Return the agent name currently bound to an immutable identity. + /// + /// The inverse of [`Self::active_agent_id`]. Ephemeral `context.update` + /// frames address agents by id only, so routing one back to a worker on + /// this broker needs this direction of the same authoritative binding. + pub(crate) fn active_agent_name(&self, agent_id: &str) -> Option<&str> { + let agent = self.active_agent_names_by_id.get(agent_id)?; + self.active_agent_bindings_by_name + .get(agent) + .filter(|binding| binding.authoritative && binding.agent_id == agent_id) + .map(|_| agent.as_str()) + } + /// Seed Relaycast's cumulative cursor after identity authority is bound. /// /// The immutable `agent_id` is the key: a later agent reusing the same name @@ -2285,6 +2298,58 @@ mod tests { } } + /// Ephemeral `context.update` frames must reach the runtime on the same + /// channel `deliver`/`action.invoke` use. Before relay#1615 they fell + /// through to the `Err` branch and were logged as an invalid frame. + #[tokio::test] + async fn context_update_reaches_the_runtime_instead_of_the_invalid_frame_path() { + let (event_tx, mut event_rx) = mpsc::channel(4); + let mut pending_agent_registrations = HashMap::new(); + let mut sink = futures_util::sink::drain::(); + + let raw = include_str!("../tests/fixtures/fleet-wire/context.update.json"); + assert!( + handle_server_message( + Message::Text(raw.to_string()), + &event_tx, + &mut pending_agent_registrations, + &mut sink, + ) + .await, + "a context.update must never close the node-control socket" + ); + + match event_rx + .try_recv() + .expect("context.update must be forwarded") + { + FleetControlEvent::Message(RelaycastToBroker::ContextUpdate(update)) => { + assert_eq!(update.event, "agent.status.active"); + assert_eq!( + update.agent_ids.as_deref(), + Some(["agt_01J7FLEET000000000000101".to_string()].as_slice()) + ); + } + other => panic!("expected a forwarded context.update, got {other:?}"), + } + } + + /// `active_agent_name` is the id -> name direction ephemeral + /// `context.update` routing depends on: it must resolve only authoritative + /// live bindings, and must stop resolving once the agent is removed. + #[test] + fn delivery_book_resolves_a_worker_name_from_an_authoritative_agent_id() { + let mut book = FleetDeliveryBook::default(); + book.bind_authoritative_identity("agent-a", "agent-a-id"); + + assert_eq!(book.active_agent_name("agent-a-id"), Some("agent-a")); + assert_eq!(book.active_agent_name("unknown-id"), None); + + // A retired identity must not keep resolving to the live worker. + book.remove_agent("agent-a"); + assert_eq!(book.active_agent_name("agent-a-id"), None); + } + #[test] fn derive_node_id_is_stable_for_same_seed_and_cwd() { let a = derive_node_id("node_seed123", "/Users/will/Projects/relay", "workspace-a"); diff --git a/crates/broker/src/relaycast/ws.rs b/crates/broker/src/relaycast/ws.rs index 803481848..d8503f6ca 100644 --- a/crates/broker/src/relaycast/ws.rs +++ b/crates/broker/src/relaycast/ws.rs @@ -210,6 +210,17 @@ impl RelaycastHttpClient { } } + /// Test hook: read the SDK token cache this client consults before it + /// registers. Lets tests prove `forget_agent_registration` really dropped a + /// worker's credential rather than asserting on a parallel flag. + #[cfg(test)] + pub(crate) fn cached_agent_token(&self, agent_name: &str) -> Option { + self.registration + .as_ref() + .as_ref() + .and_then(|registration| registration.cached_agent_token(agent_name)) + } + pub fn registration_block_remaining(&self, agent_name: &str) -> Option { self.registration .as_ref() diff --git a/crates/broker/src/runtime/fleet.rs b/crates/broker/src/runtime/fleet.rs index d5917dd55..de76f0a90 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, AgentRegistrationMetadata, BrokerToRelaycast, Deliver, - DeliveryMode, RelaycastToBroker, FLEET_WIRE_VERSION, + AgentDeregister, AgentRegister, AgentRegistrationMetadata, BrokerToRelaycast, ContextTopic, + ContextUpdate, Deliver, DeliveryMode, RelaycastToBroker, FLEET_WIRE_VERSION, }, listen_api::{DeliveryRouteError, ListenApiRequest, SetInboundDeliveryModeOk}, node_control::{delivery_ack, handler_unavailable_result, DeliveryDecision, ReceiptAckability}, @@ -34,6 +34,31 @@ pub(super) struct FleetInventoryRetry { generation: Uuid, retry_after: Instant, } + +/// First non-blank string among `keys` in an ephemeral `context.update` +/// payload. The engine's per-event payloads are open JSON objects, so callers +/// name the keys they understand in preference order. +fn context_update_string(data: &Value, keys: &[&str]) -> Option { + keys.iter() + .filter_map(|key| data.get(*key)) + .filter_map(Value::as_str) + .map(str::trim) + .find(|value| !value.is_empty()) + .map(ToOwned::to_owned) +} + +/// A `context.update` this broker has no use for. Debug, never warn: the frame +/// is legitimate and unacked, and the engine fans these out continuously. +fn debug_ignored_context_update(update: &ContextUpdate, why: &str) { + tracing::debug!( + target = "relay_broker::fleet", + topic = ?update.topic, + event = %update.event, + reason = why, + "ignoring relaycast context.update" + ); +} + pub(super) fn try_send_terminal( terminal_control_tx: &mpsc::Sender, message: TerminalToCloud, @@ -819,12 +844,182 @@ impl BrokerRuntime { FleetControlEvent::Message(RelaycastToBroker::ActionInvoke(invoke)) => { self.handle_fleet_action_invoke(invoke).await; } + FleetControlEvent::Message(RelaycastToBroker::ContextUpdate(update)) => { + self.handle_fleet_context_update(update).await; + } FleetControlEvent::Message(RelaycastToBroker::Ping(_)) | FleetControlEvent::Message(RelaycastToBroker::Reply(_)) | FleetControlEvent::Message(RelaycastToBroker::Error(_)) => {} } } + /// Ephemeral engine fan-out (`context.update`). These frames are + /// best-effort and never acked, so an event this broker has no use for is + /// dropped at debug — it is a legitimate frame and must never be logged as + /// invalid (relay#1615). + /// + /// Only terminal outcomes reach the sending agent as events: + /// `delivery.failed` is surfaced, while `delivery.deferred` (still queued + /// for a later `available_at` retry) is logged only, so a sender is never + /// prompted to resend a message the engine still intends to deliver. + async fn handle_fleet_context_update(&mut self, update: ContextUpdate) { + match (update.topic, update.event.as_str()) { + (ContextTopic::Agent, "delivery.failed") => { + let Some(workers) = self.resolve_context_update_workers(&update) else { + return; + }; + self.surface_fleet_delivery_problem(&update, &workers).await; + } + (ContextTopic::Agent, "delivery.deferred") => { + let Some(workers) = self.resolve_context_update_workers(&update) else { + return; + }; + self.log_fleet_delivery_deferral(&update, &workers); + } + (ContextTopic::Agent, "agent.identity_taken_over") => { + let Some(workers) = self.resolve_context_update_workers(&update) else { + return; + }; + self.handle_fleet_identity_taken_over(&update, &workers); + } + _ => debug_ignored_context_update(&update, "event not actionable on this broker"), + } + } + + /// Resolve the workers an actionable `context.update` applies to, logging + /// the debug drop shared by every actionable arm when this broker hosts + /// none of the addressed agents. Returns `None` when there is nothing to + /// act on, so callers can simply bail. + fn resolve_context_update_workers(&self, update: &ContextUpdate) -> Option> { + let workers = self.fleet_context_update_workers(update); + if workers.is_empty() { + debug_ignored_context_update(update, "no matching worker"); + return None; + } + Some(workers) + } + + /// Resolve `agent_ids` to workers this broker actually hosts. The engine + /// addresses ephemeral events by immutable agent id only, so this walks the + /// same authoritative binding `agent.register` established. + pub(super) fn fleet_context_update_workers(&self, update: &ContextUpdate) -> Vec { + let mut workers: Vec = Vec::new(); + for agent_id in update.agent_ids.iter().flatten() { + let Some(name) = self.fleet_delivery_book.active_agent_name(agent_id) else { + continue; + }; + let name = WorkerName::from(name); + if self.workers.workers.contains_key(&name) && !workers.contains(&name) { + workers.push(name); + } + } + workers + } + + /// Tell the SENDING agent (and operators) that Relaycast gave up on a + /// message it sent — the recipient was offline or its node was gone. + /// Without this a broker-hosted agent that DMs an unreachable agent never + /// learns (relay#1615). Reuses the same `message_delivery_failed` event the + /// broker's own dead-letter path emits, so existing SDK/dashboard consumers + /// surface it with no client change. + /// + /// Terminal outcomes only (`delivery.failed`): a deferred delivery is still + /// queued and must not be reported as a failure — see + /// [`Self::log_fleet_delivery_deferral`]. + async fn surface_fleet_delivery_problem(&self, update: &ContextUpdate, workers: &[WorkerName]) { + let reason = context_update_string(&update.data, &["reason", "error"]) + .unwrap_or_else(|| "unspecified".to_string()); + let last_error = format!("relaycast {}: {reason}", update.event); + let target = context_update_string(&update.data, &["target_agent_name", "target_agent_id"]) + .unwrap_or_else(|| "unknown".to_string()); + let delivery_id = + context_update_string(&update.data, &["delivery_id"]).map(DeliveryId::from); + let message_id = context_update_string(&update.data, &["message_id"]).map(EventId::from); + + for name in workers { + tracing::info!( + target = "relay_broker::fleet", + worker = %name, + event = %update.event, + target_agent = %target, + reason = %reason, + delivery_id = delivery_id.as_deref().unwrap_or(""), + message_id = message_id.as_deref().unwrap_or(""), + "relaycast could not deliver a message this broker's agent sent" + ); + // Best-effort: a closed SDK channel must not abort the remaining + // notifications, and this frame is never acked back to the engine. + let _ = send_broker_event( + &self.sdk_out_tx, + BrokerEvent::MessageDeliveryFailed { + name: name.clone(), + delivery_id: delivery_id.clone(), + event_id: message_id.clone(), + from: name.to_string(), + to: MessageTarget::from(target.as_str()), + attempts: 1, + last_error: last_error.clone(), + }, + ) + .await; + } + } + + /// Record that Relaycast has postponed — not abandoned — a message one of + /// this broker's agents sent. A deferred delivery stays queued for a later + /// `available_at` retry, so it deliberately emits no `BrokerEvent`: telling + /// the sender it "failed" would invite a resend and duplicate the message + /// the engine is still holding. + fn log_fleet_delivery_deferral(&self, update: &ContextUpdate, workers: &[WorkerName]) { + let reason = context_update_string(&update.data, &["reason", "error"]) + .unwrap_or_else(|| "unspecified".to_string()); + let target = context_update_string(&update.data, &["target_agent_name", "target_agent_id"]) + .unwrap_or_else(|| "unknown".to_string()); + let available_at = + context_update_string(&update.data, &["available_at"]).unwrap_or_default(); + let delivery_id = context_update_string(&update.data, &["delivery_id"]).unwrap_or_default(); + let message_id = context_update_string(&update.data, &["message_id"]).unwrap_or_default(); + + for name in workers { + tracing::info!( + target = "relay_broker::fleet", + worker = %name, + event = %update.event, + target_agent = %target, + available_at = %available_at, + reason = %reason, + delivery_id = %delivery_id, + message_id = %message_id, + "relaycast deferred a message this broker's agent sent; it stays queued for retry" + ); + } + } + + /// Another node reclaimed one of this broker's agent identities. Drop the + /// cached registration so the next operation re-registers (which routes + /// through the audited takeover path in `relaycast::ws`) instead of first + /// spending a round trip on a 401 with the now-revoked token. + fn handle_fleet_identity_taken_over(&self, update: &ContextUpdate, workers: &[WorkerName]) { + let actor = context_update_string(&update.data, &["actor"]) + .unwrap_or_else(|| "unknown".to_string()); + let reason = context_update_string(&update.data, &["reason"]) + .unwrap_or_else(|| "unspecified".to_string()); + let node_id = context_update_string(&update.data, &["node_id"]).unwrap_or_default(); + + for name in workers { + self.relaycast_http.forget_agent_registration(name); + tracing::warn!( + target = "relay_broker::fleet", + worker = %name, + actor = %actor, + reason = %reason, + node_id = %node_id, + "relaycast reported this agent identity was taken over; dropped the cached \ + registration so the next operation re-registers" + ); + } + } + async fn handle_fleet_deliver(&mut self, deliver: Deliver) { let decision = self.fleet_delivery_book.observe(&deliver); let up_to_seq = match plan_fleet_delivery(decision) { diff --git a/crates/broker/src/runtime/tests.rs b/crates/broker/src/runtime/tests.rs index d667190bf..8161bd779 100644 --- a/crates/broker/src/runtime/tests.rs +++ b/crates/broker/src/runtime/tests.rs @@ -6,12 +6,15 @@ use std::{ time::{Duration, Instant}, }; -use crate::fleet_wire::{BrokerToRelaycast, Deliver, DeliveryMode, FLEET_WIRE_VERSION}; +use crate::fleet_wire::{ + BrokerToRelaycast, ContextTopic, ContextUpdate, Deliver, DeliveryMode, RelaycastToBroker, + FLEET_WIRE_VERSION, +}; use crate::ids::{ AgentId, ChannelName, DeliveryId, EventId, MessageTarget, WorkerName, WorkspaceAlias, WorkspaceId, }; -use crate::node_control::{FleetControlCommand, FleetDeliveryBook}; +use crate::node_control::{FleetControlCommand, FleetControlEvent, FleetDeliveryBook}; use crate::protocol::{ AgentSpec, BrokerEvent, DeliveryReadAckStatus, HarnessReleasePolicy, HeadlessHarnessConfig, HeadlessHarnessDriver, MessageInjectionMode, NativeHarnessConfig, ProtocolEnvelope, @@ -1906,6 +1909,219 @@ async fn terminal_disposition_helpers_remove_withheld_fleet_ack_state() { } } +/// Build a fleet-control event carrying one ephemeral `context.update`, as the +/// node-control socket would hand it to the runtime. +fn context_update_event( + topic: ContextTopic, + event: &str, + agent_ids: &[&str], + data: Value, +) -> FleetControlEvent { + FleetControlEvent::Message(RelaycastToBroker::ContextUpdate(ContextUpdate { + v: FLEET_WIRE_VERSION, + topic, + event: event.to_string(), + channel_id: None, + agent_ids: Some(agent_ids.iter().map(|id| (*id).to_string()).collect()), + data, + })) +} + +/// relay#1615: Relaycast tells the SENDING agent when a message could not +/// land. The broker must surface that on the SDK/dashboard stream instead of +/// dropping the frame — a broker-hosted agent that DMs an offline agent used to +/// never learn. +#[tokio::test] +async fn relaycast_delivery_failure_reaches_the_sending_worker() { + let worker_name = "worker-a"; + let registry = make_worker_registry_with_worker(worker_name).await; + let mut fixture = worker_event_runtime_fixture(registry, HashMap::new()); + fixture + .runtime + .fleet_delivery_book + .bind_authoritative_identity(worker_name, "agt_sender"); + + fixture + .runtime + .handle_fleet_control_event(context_update_event( + ContextTopic::Agent, + "delivery.failed", + &["agt_sender"], + json!({ + "delivery_id": "del_remote_1", + "message_id": "msg_remote_1", + "target_agent_id": "agt_target", + "target_agent_name": "planner", + "reason": "recipient_offline", + "error": "agent has no live node", + "retryable": false, + }), + )) + .await; + + let frame = tokio::time::timeout(Duration::from_secs(1), fixture._sdk_out_rx.recv()) + .await + .expect("a delivery failure must be emitted") + .expect("sdk_out_tx should remain open"); + assert_eq!(frame.payload["kind"], "message_delivery_failed"); + assert_eq!(frame.payload["name"], worker_name); + assert_eq!(frame.payload["from"], worker_name); + assert_eq!(frame.payload["to"], "planner"); + assert_eq!(frame.payload["delivery_id"], "del_remote_1"); + assert_eq!(frame.payload["event_id"], "msg_remote_1"); + assert_eq!( + frame.payload["lastError"], + "relaycast delivery.failed: recipient_offline" + ); +} + +/// A deferred delivery is still queued for a later `available_at` retry, so it +/// must NOT reach the sender as `message_delivery_failed` — that would invite a +/// resend and duplicate the message the engine is still holding. The broker +/// takes the log-only path instead and emits nothing. +#[tokio::test] +async fn relaycast_delivery_deferral_emits_no_event() { + let worker_name = "worker-a"; + let registry = make_worker_registry_with_worker(worker_name).await; + let mut fixture = worker_event_runtime_fixture(registry, HashMap::new()); + fixture + .runtime + .fleet_delivery_book + .bind_authoritative_identity(worker_name, "agt_sender"); + + let update = ContextUpdate { + v: FLEET_WIRE_VERSION, + topic: ContextTopic::Agent, + event: "delivery.deferred".to_string(), + channel_id: None, + agent_ids: Some(vec!["agt_sender".to_string()]), + data: json!({ + "delivery_id": "del_remote_2", + "message_id": "msg_remote_2", + "available_at": "2026-09-02T00:00:00.000Z", + "reason": "recipient_busy", + }), + }; + + // The deferral resolves to this broker's worker, so the log-only path runs + // rather than the `no matching worker` drop. + assert_eq!( + fixture.runtime.fleet_context_update_workers(&update), + vec![WorkerName::from(worker_name)] + ); + + fixture + .runtime + .handle_fleet_control_event(FleetControlEvent::Message( + RelaycastToBroker::ContextUpdate(update), + )) + .await; + + assert!( + tokio::time::timeout(Duration::from_millis(100), fixture._sdk_out_rx.recv()) + .await + .is_err(), + "a deferred delivery is not terminal and must not emit message_delivery_failed" + ); +} + +/// Frames naming an agent this broker does not host, and events it has no use +/// for, are dropped silently — never emitted, never logged as invalid. +#[tokio::test] +async fn unrelated_context_updates_emit_nothing() { + let worker_name = "worker-a"; + let registry = make_worker_registry_with_worker(worker_name).await; + let mut fixture = worker_event_runtime_fixture(registry, HashMap::new()); + fixture + .runtime + .fleet_delivery_book + .bind_authoritative_identity(worker_name, "agt_sender"); + + // An agent hosted somewhere else. + fixture + .runtime + .handle_fleet_control_event(context_update_event( + ContextTopic::Agent, + "delivery.failed", + &["agt_elsewhere"], + json!({"reason": "recipient_offline"}), + )) + .await; + // A topic/event this broker does not act on. + fixture + .runtime + .handle_fleet_control_event(context_update_event( + ContextTopic::Presence, + "agent.status.active", + &["agt_sender"], + json!({"status": "active"}), + )) + .await; + + assert!( + tokio::time::timeout(Duration::from_millis(100), fixture._sdk_out_rx.recv()) + .await + .is_err(), + "unrelated context.update frames must not emit broker events" + ); +} + +/// `agent.identity_taken_over` means another node reclaimed this worker's +/// identity. Drop the cached registration so the next operation re-registers +/// instead of first spending a round trip on a 401. +#[tokio::test] +async fn identity_takeover_drops_the_cached_registration() { + let worker_name = "worker-a"; + let registry = make_worker_registry_with_worker(worker_name).await; + let mut fixture = worker_event_runtime_fixture(registry, HashMap::new()); + fixture + .runtime + .fleet_delivery_book + .bind_authoritative_identity(worker_name, "agt_sender"); + fixture + .runtime + .relaycast_http + .seed_agent_token(worker_name, "at_now_revoked"); + assert_eq!( + fixture + .runtime + .relaycast_http + .cached_agent_token(worker_name), + Some("at_now_revoked".to_string()) + ); + + fixture + .runtime + .handle_fleet_control_event(context_update_event( + ContextTopic::Agent, + "agent.identity_taken_over", + &["agt_sender"], + json!({ + "agent_id": "agt_sender", + "agent_name": worker_name, + "actor": "other-node", + "reason": "operator reclaimed the name", + "node_id": "node-other", + }), + )) + .await; + + assert_eq!( + fixture + .runtime + .relaycast_http + .cached_agent_token(worker_name), + None, + "the revoked token must not survive a takeover notification" + ); + assert!( + tokio::time::timeout(Duration::from_millis(100), fixture._sdk_out_rx.recv()) + .await + .is_err(), + "a takeover notification is a log/state change, not a delivery event" + ); +} + // Full runtime/channel companion for the terminal-disposition coverage above. // Each real disposal path removes the pending delivery first; a late matching // worker `delivery_ack` is then driven through `BrokerRuntime::handle_worker_event`. diff --git a/crates/broker/tests/fixtures/fleet-wire/context.update.json b/crates/broker/tests/fixtures/fleet-wire/context.update.json new file mode 100644 index 000000000..8ecd3d443 --- /dev/null +++ b/crates/broker/tests/fixtures/fleet-wire/context.update.json @@ -0,0 +1,13 @@ +{ + "v": 1, + "type": "context.update", + "topic": "presence", + "event": "agent.status.active", + "channel_id": null, + "agent_ids": ["agt_01J7FLEET000000000000101"], + "data": { + "agent_id": "agt_01J7FLEET000000000000202", + "agent_name": "planner", + "status": "active" + } +} diff --git a/crates/broker/tests/fleet_wire_fixtures.rs b/crates/broker/tests/fleet_wire_fixtures.rs index 813647cb1..1871c10c4 100644 --- a/crates/broker/tests/fleet_wire_fixtures.rs +++ b/crates/broker/tests/fleet_wire_fixtures.rs @@ -16,7 +16,14 @@ const NODE_TO_SERVER_TYPES: &[&str] = &[ "inventory.sync", ]; -const SERVER_TO_NODE_TYPES: &[&str] = &["deliver", "action.invoke", "ping", "reply", "error"]; +const SERVER_TO_NODE_TYPES: &[&str] = &[ + "deliver", + "action.invoke", + "context.update", + "ping", + "reply", + "error", +]; const EXPECTED_FIXTURE_FILES: &[&str] = &[ "action.invoke.json", @@ -24,6 +31,7 @@ const EXPECTED_FIXTURE_FILES: &[&str] = &[ "action.result.output.json", "agent.deregister.json", "agent.register.json", + "context.update.json", "deliver.json", "delivery.ack.json", "error.json", diff --git a/tests/relayflows/cases/1615-context-update-frames/case.json b/tests/relayflows/cases/1615-context-update-frames/case.json new file mode 100644 index 000000000..4df429e7d --- /dev/null +++ b/tests/relayflows/cases/1615-context-update-frames/case.json @@ -0,0 +1,21 @@ +{ + "version": 1, + "id": "1615-context-update-frames", + "kind": "bugfix", + "title": "Accept Relaycast context.update node frames instead of rejecting them", + "requirements": ["broker-linux-x64"], + "runner": { + "command": ["node", "tests/relayflows/cases/1615-context-update-frames/run.mjs"] + }, + "timeoutSeconds": 600, + "expected": { + "base": { + "outcome": "bug", + "signature": "context_update_rejected_as_invalid_frame" + }, + "head": { + "outcome": "fixed", + "signature": "context_update_accepted_and_routed" + } + } +} diff --git a/tests/relayflows/cases/1615-context-update-frames/run.mjs b/tests/relayflows/cases/1615-context-update-frames/run.mjs new file mode 100644 index 000000000..d946eae7e --- /dev/null +++ b/tests/relayflows/cases/1615-context-update-frames/run.mjs @@ -0,0 +1,437 @@ +import { execFileSync, spawn } from 'node:child_process'; +import { constants as fsConstants } from 'node:fs'; +import { access, mkdtemp, mkdir, rm, writeFile } from 'node:fs/promises'; +import { tmpdir } from 'node:os'; +import path from 'node:path'; +import process from 'node:process'; +import { fileURLToPath } from 'node:url'; + +// relay#1615: Relaycast pushes ephemeral `context.update` frames to every +// ws-kind node. The base broker's `ServerToNode` union had no variant for them, +// so every frame failed to parse and was logged as `invalid fleet node ws +// frame` — including the `delivery.failed` / `delivery.deferred` events that +// tell a sending agent its message never landed. The head broker parses the +// frame and routes it into the runtime's delivery-problem handler. +const CASE_ID = '1615-context-update-frames'; +const INVALID_FRAME_MARKER = 'invalid fleet node ws frame'; +const ROUTED_MARKER = 'ignoring relaycast context.update'; +const OBSERVATION_TIMEOUT_MS = 60_000; + +const targetDir = requiredDirectory('RELAY_PR_PROOF_TARGET_DIR'); +const harnessDir = requiredDirectory('RELAY_PR_PROOF_HARNESS_DIR'); +const binaryPath = await requiredExecutable('RELAY_PR_PROOF_BROKER_BINARY'); +const resultPath = requiredValue('RELAY_PR_PROOF_RESULT_PATH'); +const arm = requiredValue('RELAY_PR_PROOF_ARM'); + +if (arm !== 'base' && arm !== 'head') { + throw new Error(`RELAY_PR_PROOF_ARM must be base or head, received ${JSON.stringify(arm)}.`); +} + +const expectedSha = + arm === 'base' ? process.env.RELAY_PR_PROOF_BASE_SHA : process.env.RELAY_PR_PROOF_HEAD_SHA; +if (!expectedSha) throw new Error(`Missing expected ${arm} SHA.`); +const targetSha = execFileSync('git', ['-C', targetDir, 'rev-parse', 'HEAD'], { + encoding: 'utf8', +}).trim(); +if (targetSha !== expectedSha) { + throw new Error(`Target checkout ${targetSha} does not match exact ${arm} SHA ${expectedSha}.`); +} + +const runnerPath = fileURLToPath(import.meta.url); +if (!isWithin(harnessDir, runnerPath)) { + throw new Error('The RelayFlow runner must execute from the exact-head harness checkout.'); +} + +const probeDir = await mkdtemp(path.join(tmpdir(), 'relayflow-1615-')); +const stateDir = path.join(probeDir, 'state'); +const serverPath = path.join(probeDir, 'fake-relaycast.mjs'); + +// A dependency-free Relaycast stand-in: the handful of HTTP routes the broker +// touches on the way to node control, plus an RFC 6455 server on +// /v1/node/ws that answers `node.register` and then pushes exactly one +// `context.update` (topic `agent`, event `delivery.failed`). +const serverSource = String.raw`import crypto from 'node:crypto'; +import http from 'node:http'; + +const WS_GUID = '258EAFA5-E914-47DA-95CA-C5AB0DC85B11'; +const CONTEXT_UPDATE = { + v: 1, + type: 'context.update', + topic: 'agent', + event: 'delivery.failed', + channel_id: null, + agent_ids: ['agt_relayflow_sender'], + data: { + delivery_id: 'del_relayflow_probe', + message_id: 'msg_relayflow_probe', + target_agent_id: 'agt_relayflow_target', + target_agent_name: 'planner', + reason: 'recipient_offline', + error: 'agent has no live node', + retryable: false, + }, +}; + +function encodeTextFrame(text) { + const payload = Buffer.from(text, 'utf8'); + const length = payload.length; + let header; + if (length < 126) { + header = Buffer.from([0x81, length]); + } else if (length < 65536) { + header = Buffer.alloc(4); + header[0] = 0x81; + header[1] = 126; + header.writeUInt16BE(length, 2); + } else { + header = Buffer.alloc(10); + header[0] = 0x81; + header[1] = 127; + header.writeBigUInt64BE(BigInt(length), 2); + } + return Buffer.concat([header, payload]); +} + +function createFrameReader(onText) { + let buffer = Buffer.alloc(0); + return (chunk) => { + buffer = Buffer.concat([buffer, chunk]); + for (;;) { + if (buffer.length < 2) return; + const opcode = buffer[0] & 0x0f; + const masked = (buffer[1] & 0x80) !== 0; + let length = buffer[1] & 0x7f; + let offset = 2; + if (length === 126) { + if (buffer.length < offset + 2) return; + length = buffer.readUInt16BE(offset); + offset += 2; + } else if (length === 127) { + if (buffer.length < offset + 8) return; + length = Number(buffer.readBigUInt64BE(offset)); + offset += 8; + } + let mask = null; + if (masked) { + if (buffer.length < offset + 4) return; + mask = buffer.subarray(offset, offset + 4); + offset += 4; + } + if (buffer.length < offset + length) return; + const payload = Buffer.from(buffer.subarray(offset, offset + length)); + buffer = buffer.subarray(offset + length); + if (mask) for (let i = 0; i < payload.length; i += 1) payload[i] ^= mask[i % 4]; + if (opcode === 0x1) onText(payload.toString('utf8')); + } + }; +} + +const server = http.createServer((request, response) => { + let body = ''; + request.on('data', (chunk) => { + body += chunk; + }); + request.on('end', () => { + const url = request.url.split('?')[0]; + const send = (data) => { + response.writeHead(200, { 'content-type': 'application/json' }); + response.end(JSON.stringify({ ok: true, data })); + }; + if (request.method === 'POST' && url === '/v1/agents') { + let parsed = {}; + try { + parsed = JSON.parse(body || '{}'); + } catch {} + send({ + id: 'agt_relayflow_broker', + workspace_id: 'ws_relayflow', + name: parsed.name ?? 'broker', + token: 'at_relayflow_broker', + status: 'online', + created_at: '2026-09-01T00:00:00.000Z', + }); + return; + } + if (url === '/v1/agents' || url === '/v1/channels') { + send([]); + return; + } + if (url.startsWith('/v1/agents/')) { + send({ id: 'agt_relayflow_other', name: 'other', status: 'offline', metadata: {} }); + return; + } + send({}); + }); +}); + +server.on('upgrade', (request, socket) => { + const key = request.headers['sec-websocket-key']; + if (!key) { + socket.destroy(); + return; + } + const accept = crypto.createHash('sha1').update(key + WS_GUID).digest('base64'); + socket.write( + 'HTTP/1.1 101 Switching Protocols\r\n' + + 'Upgrade: websocket\r\nConnection: Upgrade\r\n' + + 'Sec-WebSocket-Accept: ' + accept + '\r\n\r\n' + ); + // Only the node-control socket drives the probe; the broker also opens a + // separate terminal socket, which is accepted and then left idle. + const isNodeControl = request.url.split('?')[0] === '/v1/node/ws'; + let pushed = false; + socket.on('error', () => {}); + socket.on( + 'data', + createFrameReader((text) => { + if (!isNodeControl) return; + let frame; + try { + frame = JSON.parse(text); + } catch { + return; + } + if (frame.type !== 'node.register' || pushed) return; + pushed = true; + socket.write( + encodeTextFrame( + JSON.stringify({ v: 1, type: 'reply', id: frame.id ?? 'node-register', ok: true, data: {} }) + ) + ); + // Give the broker's control client a beat to finish the registration + // handshake before the ephemeral frame arrives. + setTimeout(() => { + socket.write(encodeTextFrame(JSON.stringify(CONTEXT_UPDATE))); + process.stdout.write(JSON.stringify({ pushed: true }) + '\n'); + }, 500); + }) + ); +}); + +server.listen(0, '127.0.0.1', () => { + const address = server.address(); + if (!address || typeof address === 'string') throw new Error('Expected a TCP address.'); + process.stdout.write(JSON.stringify({ port: address.port }) + '\n'); +}); + +process.once('SIGTERM', () => server.close(() => process.exit(0))); +`; + +let server; +let broker; +try { + await mkdir(stateDir, { recursive: true }); + await writeFile(serverPath, serverSource, { encoding: 'utf8', mode: 0o600 }); + server = spawn(process.execPath, [serverPath], { + cwd: probeDir, + stdio: ['ignore', 'pipe', 'pipe'], + }); + let serverStderr = ''; + server.stderr.on('data', (chunk) => { + serverStderr += chunk.toString(); + }); + const pushedFrame = { seen: false }; + const port = await waitForServerReady(server, pushedFrame); + + broker = spawn( + binaryPath, + [ + 'init', + '--instance-name', + 'relayflow-1615-node', + '--api-port', + '0', + '--channels', + 'general', + '--state-dir', + stateDir, + ], + { + cwd: probeDir, + stdio: ['ignore', 'pipe', 'pipe'], + env: { + ...process.env, + RELAY_API_KEY: 'rk_relayflow_1615_probe', + RELAYCAST_BASE_URL: `http://127.0.0.1:${port}`, + RELAY_NODE_TOKEN: 'nt_relayflow_1615_probe', + AGENT_RELAY_BROKER_LOG: 'stderr', + // `invalid fleet node ws frame` is a warning; the accepted-frame path + // logs at debug because an ephemeral frame is never an error. The + // event's tracing target is its module path (`relay_broker::runtime:: + // fleet`) — the `target = "relay_broker::fleet"` in the macro is a + // structured field, not the metadata target — so enable both prefixes + // rather than betting the proof on which one the call site uses. + RUST_LOG: 'info,relay_broker::runtime::fleet=debug,relay_broker::fleet=debug', + RELAY_TELEMETRY_DISABLED: '1', + }, + } + ); + + let brokerStderr = ''; + let brokerStdout = ''; + broker.stdout.on('data', (chunk) => { + brokerStdout += chunk.toString(); + }); + broker.stderr.on('data', (chunk) => { + brokerStderr += chunk.toString(); + }); + + const observedAt = await waitForObservation( + () => { + if (brokerStderr.includes(INVALID_FRAME_MARKER)) return 'rejected'; + if (pushedFrame.seen && brokerStderr.includes(ROUTED_MARKER)) return 'routed'; + return null; + }, + broker, + OBSERVATION_TIMEOUT_MS + ); + + let outcome; + let signature; + let details; + if (observedAt === 'rejected') { + outcome = 'bug'; + signature = 'context_update_rejected_as_invalid_frame'; + details = + 'The compiled base broker could not parse the Relaycast context.update (topic agent, event ' + + `delivery.failed) and logged it as "${INVALID_FRAME_MARKER}", so the sending agent was never told ` + + 'its message failed to land.'; + } else if (observedAt === 'routed') { + outcome = 'fixed'; + signature = 'context_update_accepted_and_routed'; + details = + 'The compiled head broker parsed the Relaycast context.update (topic agent, event ' + + 'delivery.failed) and routed it into the runtime delivery-problem handler; no ' + + `"${INVALID_FRAME_MARKER}" was logged.`; + } else { + throw new Error( + `Unexpected compiled context.update observation: ${JSON.stringify({ + arm, + pushed: pushedFrame.seen, + brokerExit: broker.exitCode, + stdout: brokerStdout.slice(-2_000), + stderr: brokerStderr.slice(-4_000), + serverStderr: serverStderr.slice(-2_000), + })}.` + ); + } + + await mkdir(path.dirname(resultPath), { recursive: true }); + await writeFile( + resultPath, + `${JSON.stringify({ version: 1, caseId: CASE_ID, arm, outcome, signature, details })}\n`, + 'utf8' + ); +} finally { + await terminate(broker); + await terminate(server); + await rm(probeDir, { recursive: true, force: true }); +} + +function requiredValue(name) { + const value = process.env[name]?.trim(); + if (!value) throw new Error(`Missing required environment variable ${name}.`); + return value; +} + +function requiredDirectory(name) { + return path.resolve(requiredValue(name)); +} + +async function requiredExecutable(name) { + const candidate = path.resolve(requiredValue(name)); + try { + await access(candidate, fsConstants.R_OK | fsConstants.X_OK); + } catch { + throw new Error(`${name} must name a readable executable file.`); + } + return candidate; +} + +function isWithin(directory, candidate) { + const relative = path.relative(directory, candidate); + return ( + relative === '' || + (!relative.startsWith(`..${path.sep}`) && relative !== '..' && !path.isAbsolute(relative)) + ); +} + +async function terminate(child) { + if (!child || child.exitCode !== null) return; + child.kill('SIGTERM'); + await Promise.race([ + new Promise((resolve) => child.once('exit', resolve)), + new Promise((resolve) => setTimeout(resolve, 5_000)), + ]); + if (child.exitCode === null) child.kill('SIGKILL'); +} + +// The fake Relaycast announces its port on the first stdout line and the +// context.update push on the second, so the probe never guesses at timing. +function waitForServerReady(child, pushedFrame) { + return new Promise((resolve, reject) => { + let stdout = ''; + let resolved = false; + const timer = setTimeout( + () => reject(new Error('fake Relaycast did not report a listening port')), + 15_000 + ); + + child.stdout.on('data', (chunk) => { + stdout += chunk.toString(); + for (;;) { + const newline = stdout.indexOf('\n'); + if (newline < 0) return; + const line = stdout.slice(0, newline); + stdout = stdout.slice(newline + 1); + let parsed; + try { + parsed = JSON.parse(line); + } catch (error) { + clearTimeout(timer); + reject(new Error(`fake Relaycast emitted invalid readiness: ${error.message}`)); + return; + } + if (parsed.pushed) { + pushedFrame.seen = true; + continue; + } + if (!Number.isInteger(parsed.port) || parsed.port <= 0) { + clearTimeout(timer); + reject(new Error(`fake Relaycast reported an invalid port ${JSON.stringify(parsed.port)}`)); + return; + } + clearTimeout(timer); + resolved = true; + resolve(parsed.port); + } + }); + child.once('exit', (code, signal) => { + clearTimeout(timer); + if (!resolved) { + reject(new Error(`fake Relaycast exited before readiness (${signal ?? code ?? 'unknown'})`)); + } + }); + }); +} + +// Poll the accumulated broker log until one of the two mutually exclusive +// markers appears, the broker dies, or the bound elapses. +function waitForObservation(check, child, timeoutMs) { + return new Promise((resolve) => { + const deadline = Date.now() + timeoutMs; + const poll = () => { + const observation = check(); + if (observation) { + resolve(observation); + return; + } + if (child.exitCode !== null || Date.now() >= deadline) { + resolve(null); + return; + } + setTimeout(poll, 250); + }; + poll(); + }); +}