From c919d2eeff0bf606a215141822571ca1fdbba5d3 Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 2 Sep 2026 04:42:44 +0000 Subject: [PATCH 1/6] fix(broker): accept context.update node frames and act on delivery failures and identity takeover MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The relaycast engine pushes ephemeral `context.update` frames to every ws-kind node, but `ServerToNode` had no variant for them, so every frame failed to parse and was logged as `invalid fleet node ws frame`. Among the events lost that way were the `delivery.failed` / `delivery.deferred` notices the engine sends to the SENDING agent when a recipient could not be reached (relay#1615), and `agent.identity_taken_over`. - fleet_wire: add `ContextUpdate` + `ContextTopic` mirroring the engine's canonical schema, forward-compatible (no `deny_unknown_fields`) like the other inbound frames, plus the canonical fixture and round-trip coverage. - runtime/fleet: route the parsed frame off the existing fleet-control channel. `delivery.failed`/`delivery.deferred` for a hosted agent now emit the same `BrokerEvent::MessageDeliveryFailed` the broker's own dead-letter path uses (engine reason + target agent name) plus an info log; `agent.identity_taken_over` drops the worker's cached Relaycast registration via `forget_agent_registration` so the next operation re-registers instead of first hitting a 401. Everything else is ignored at debug — never as an invalid frame. - RelayFlow proof case 1615-context-update-frames stands up a dependency-free fake Relaycast (HTTP + /v1/node/ws) and observes the exact base/head broker binaries. Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_01MB3K5bK7Jc5HM92fsZRUyS --- .../2026-09/traj_h0xx33q5a1ga/summary.md | 37 ++ .../2026-09/traj_h0xx33q5a1ga/trajectory.json | 69 +++ CHANGELOG.md | 7 +- crates/broker/src/fleet_wire.rs | 103 ++++- crates/broker/src/node_control.rs | 62 +++ crates/broker/src/relaycast/ws.rs | 11 + crates/broker/src/runtime/fleet.rs | 145 +++++- crates/broker/src/runtime/tests.rs | 209 ++++++++- .../fixtures/fleet-wire/context.update.json | 15 + crates/broker/tests/fleet_wire_fixtures.rs | 10 +- .../1615-context-update-frames/case.json | 21 + .../cases/1615-context-update-frames/run.mjs | 433 ++++++++++++++++++ 12 files changed, 1114 insertions(+), 8 deletions(-) create mode 100644 .agentworkforce/trajectories/completed/2026-09/traj_h0xx33q5a1ga/summary.md create mode 100644 .agentworkforce/trajectories/completed/2026-09/traj_h0xx33q5a1ga/trajectory.json create mode 100644 crates/broker/tests/fixtures/fleet-wire/context.update.json create mode 100644 tests/relayflows/cases/1615-context-update-frames/case.json create mode 100644 tests/relayflows/cases/1615-context-update-frames/run.mjs 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..77658a8f0 --- /dev/null +++ b/.agentworkforce/trajectories/completed/2026-09/traj_h0xx33q5a1ga/summary.md @@ -0,0 +1,37 @@ +# 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, surfaces delivery.failed/deferred to the sending worker as message_delivery_failed, and drops the cached registration on agent.identity_taken_over + +**Approach:** Standard approach + +--- + +## 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 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..8f8699cc9 --- /dev/null +++ b/.agentworkforce/trajectories/completed/2026-09/traj_h0xx33q5a1ga/trajectory.json @@ -0,0 +1,69 @@ +{ + "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, surfaces delivery.failed/deferred to the sending worker as message_delivery_failed, and drops the cached registration on agent.identity_taken_over", + "approach": "Standard approach", + "confidence": 0.85 + }, + "commits": [], + "filesChanged": [], + "projectId": "AgentWorkforce/relay", + "tags": [], + "_trace": { + "startRef": "6d5199ff103cb5f4ff6adf0a3fa32a788646a9bc", + "endRef": "6d5199ff103cb5f4ff6adf0a3fa32a788646a9bc" + } +} \ No newline at end of file diff --git a/CHANGELOG.md b/CHANGELOG.md index df38251d3..fb45247bf 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,7 +5,12 @@ All notable changes to Agent Relay will be documented in this file. The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/), and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). -## [Unreleased] +## [Unreleased - Patch] + +### Fixed + +- Broker now accepts Relaycast `context.update` node frames instead of logging every one as an invalid frame, and surfaces `delivery.failed`/`delivery.deferred` to the sending agent as a `message_delivery_failed` event so a DM to an unreachable agent is no longer silently lost. +- 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.0] - 2026-09-02 diff --git a/crates/broker/src/fleet_wire.rs b/crates/broker/src/fleet_wire.rs index db105f154..0f79fac29 100644 --- a/crates/broker/src/fleet_wire.rs +++ b/crates/broker/src/fleet_wire.rs @@ -601,6 +601,42 @@ 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 { + pub v: FleetWireVersion, + pub topic: ContextTopic, + 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>, + pub data: Value, +} + #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] #[serde(deny_unknown_fields)] pub struct Ping { @@ -742,6 +778,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 +798,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 +1363,65 @@ 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"); + } + + #[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 02707eaa9..d3c822fbe 100644 --- a/crates/broker/src/node_control.rs +++ b/crates/broker/src/node_control.rs @@ -762,6 +762,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 @@ -2193,6 +2206,55 @@ 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:?}"), + } + } + + #[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 86d92a6ba..7d1347a28 100644 --- a/crates/broker/src/relaycast/ws.rs +++ b/crates/broker/src/relaycast/ws.rs @@ -170,6 +170,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 00e4f8500..b4f9eab01 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}, @@ -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, @@ -764,12 +789,128 @@ 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). + async fn handle_fleet_context_update(&mut self, update: ContextUpdate) { + match (update.topic, update.event.as_str()) { + (ContextTopic::Agent, "delivery.failed" | "delivery.deferred") => { + let workers = self.fleet_context_update_workers(&update); + if workers.is_empty() { + debug_ignored_context_update(&update, "no matching worker"); + return; + } + self.surface_fleet_delivery_problem(&update, &workers).await; + } + (ContextTopic::Agent, "agent.identity_taken_over") => { + let workers = self.fleet_context_update_workers(&update); + if workers.is_empty() { + debug_ignored_context_update(&update, "no matching worker"); + return; + } + self.handle_fleet_identity_taken_over(&update, &workers); + } + _ => debug_ignored_context_update(&update, "event not actionable on this broker"), + } + } + + /// 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. + 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 could not land a + /// message it sent — the recipient was offline, its node was gone, or the + /// delivery was deferred. 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. + 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; + } + } + + /// 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 ff1203253..b53dfa228 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, @@ -1481,6 +1484,208 @@ async fn terminal_disposition_helpers_remove_withheld_fleet_ack_state() { } } +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 surfaced through the same channel, with the engine's +/// event name kept in the reason so a consumer can tell the two apart. +#[tokio::test] +async fn relaycast_delivery_deferral_is_labelled_as_deferred() { + 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.deferred", + &["agt_sender"], + json!({ + "delivery_id": "del_remote_2", + "message_id": "msg_remote_2", + "available_at": "2026-09-02T00:00:00.000Z", + "reason": "recipient_busy", + }), + )) + .await; + + let frame = tokio::time::timeout(Duration::from_secs(1), fixture._sdk_out_rx.recv()) + .await + .expect("a deferral must be emitted") + .expect("sdk_out_tx should remain open"); + assert_eq!(frame.payload["kind"], "message_delivery_failed"); + assert_eq!( + frame.payload["lastError"], + "relaycast delivery.deferred: recipient_busy" + ); + // No target agent on a deferral payload: the event still has to name a + // target rather than panicking or emitting an empty one. + assert_eq!(frame.payload["to"], "unknown"); +} + +/// 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..fdc253667 --- /dev/null +++ b/crates/broker/tests/fixtures/fleet-wire/context.update.json @@ -0,0 +1,15 @@ +{ + "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..9ca748a1f --- /dev/null +++ b/tests/relayflows/cases/1615-context-update-frames/run.mjs @@ -0,0 +1,433 @@ +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. + RUST_LOG: 'info,relay_broker::runtime::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(); + }); +} From 0e6f9f4b5e69f341a50709f367631ba76a93f8c2 Mon Sep 17 00:00:00 2001 From: "github-actions[bot]" Date: Wed, 2 Sep 2026 04:44:58 +0000 Subject: [PATCH 2/6] style: auto-format with Prettier --- crates/broker/tests/fixtures/fleet-wire/context.update.json | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) diff --git a/crates/broker/tests/fixtures/fleet-wire/context.update.json b/crates/broker/tests/fixtures/fleet-wire/context.update.json index fdc253667..8ecd3d443 100644 --- a/crates/broker/tests/fixtures/fleet-wire/context.update.json +++ b/crates/broker/tests/fixtures/fleet-wire/context.update.json @@ -4,9 +4,7 @@ "topic": "presence", "event": "agent.status.active", "channel_id": null, - "agent_ids": [ - "agt_01J7FLEET000000000000101" - ], + "agent_ids": ["agt_01J7FLEET000000000000101"], "data": { "agent_id": "agt_01J7FLEET000000000000202", "agent_name": "planner", From e22a7a712025bd2b325d607bff54f5a00c01cc06 Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 2 Sep 2026 12:08:50 +0000 Subject: [PATCH 3/6] fix(broker): keep deferred deliveries non-terminal and fix proof-case log filter MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Review follow-up on relay#1615. - runtime/fleet: `delivery.deferred` no longer emits `BrokerEvent::MessageDeliveryFailed`. A deferred delivery stays queued for a later `available_at` retry, so reporting it as a failure invites the sender to resend and duplicate the message the engine still holds. It now takes a log-only path (`log_fleet_delivery_deferral`, info with worker/target/ available_at/reason); only `delivery.failed` is surfaced as an event. - RelayFlow case 1615-context-update-frames: widen `RUST_LOG` so the routed marker survives the filter regardless of which prefix the call site uses. The event's tracing target is its module path (`relay_broker::runtime::fleet`) — `target = "relay_broker::fleet"` in the macro is a structured field, not the metadata target — so the directive now enables both. - Docstrings on the new `context.update` surface (fleet_wire fields and tests, node_control test, runtime helpers). - Trajectory traj_h0xx33q5a1ga: record the commit, the product files it touched, a real start/end trace range, and the verification actually run. - CHANGELOG: the bullet now only claims `delivery.failed` is surfaced. Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_01MB3K5bK7Jc5HM92fsZRUyS --- .../2026-09/traj_h0xx33q5a1ga/summary.md | 25 +++++++- .../2026-09/traj_h0xx33q5a1ga/trajectory.json | 25 ++++++-- CHANGELOG.md | 2 +- crates/broker/src/fleet_wire.rs | 9 +++ crates/broker/src/node_control.rs | 3 + crates/broker/src/runtime/fleet.rs | 63 ++++++++++++++++--- crates/broker/src/runtime/tests.rs | 59 ++++++++++------- .../cases/1615-context-update-frames/run.mjs | 8 ++- 8 files changed, 151 insertions(+), 43 deletions(-) diff --git a/.agentworkforce/trajectories/completed/2026-09/traj_h0xx33q5a1ga/summary.md b/.agentworkforce/trajectories/completed/2026-09/traj_h0xx33q5a1ga/summary.md index 77658a8f0..28c8781d6 100644 --- a/.agentworkforce/trajectories/completed/2026-09/traj_h0xx33q5a1ga/summary.md +++ b/.agentworkforce/trajectories/completed/2026-09/traj_h0xx33q5a1ga/summary.md @@ -10,9 +10,9 @@ ## Summary -Broker parses Relaycast context.update node frames, surfaces delivery.failed/deferred to the sending worker as message_delivery_failed, and drops the cached registration on agent.identity_taken_over +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:** Standard approach +**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. --- @@ -35,3 +35,24 @@ Broker parses Relaycast context.update node frames, surfaces delivery.failed/def - 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` + +Traced range: `6d5199ff103cb5f4ff6adf0a3fa32a788646a9bc` .. `c919d2eeff0bf606a215141822571ca1fdbba5d3` + +## 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 index 8f8699cc9..813d07190 100644 --- a/.agentworkforce/trajectories/completed/2026-09/traj_h0xx33q5a1ga/trajectory.json +++ b/.agentworkforce/trajectories/completed/2026-09/traj_h0xx33q5a1ga/trajectory.json @@ -54,16 +54,29 @@ } ], "retrospective": { - "summary": "Broker parses Relaycast context.update node frames, surfaces delivery.failed/deferred to the sending worker as message_delivery_failed, and drops the cached registration on agent.identity_taken_over", - "approach": "Standard approach", + "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": [], - "filesChanged": [], + "commits": [ + "c919d2eeff0bf606a215141822571ca1fdbba5d3" + ], + "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": "6d5199ff103cb5f4ff6adf0a3fa32a788646a9bc" + "endRef": "c919d2eeff0bf606a215141822571ca1fdbba5d3" } -} \ No newline at end of file +} diff --git a/CHANGELOG.md b/CHANGELOG.md index 995a2e186..1833e629a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,7 +9,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Fixed -- Broker now accepts Relaycast `context.update` node frames instead of logging every one as an invalid frame, and surfaces `delivery.failed`/`delivery.deferred` to the sending agent as a `message_delivery_failed` event so a DM to an unreachable agent is no longer silently lost. +- Broker now accepts Relaycast `context.update` node frames instead of logging every one as an invalid frame, and surfaces `delivery.failed` to the sending agent as a `message_delivery_failed` event so a DM to an unreachable agent is no longer silently lost. - 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.1] - 2026-09-02 diff --git a/crates/broker/src/fleet_wire.rs b/crates/broker/src/fleet_wire.rs index 0f79fac29..acc4f21e2 100644 --- a/crates/broker/src/fleet_wire.rs +++ b/crates/broker/src/fleet_wire.rs @@ -621,8 +621,12 @@ pub enum ContextTopic { /// 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 @@ -634,6 +638,8 @@ pub struct ContextUpdate { /// 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, } @@ -1413,6 +1419,9 @@ mod tests { 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!({ diff --git a/crates/broker/src/node_control.rs b/crates/broker/src/node_control.rs index d3c822fbe..aea155c35 100644 --- a/crates/broker/src/node_control.rs +++ b/crates/broker/src/node_control.rs @@ -2242,6 +2242,9 @@ mod tests { } } + /// `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(); diff --git a/crates/broker/src/runtime/fleet.rs b/crates/broker/src/runtime/fleet.rs index b4f9eab01..9c438deeb 100644 --- a/crates/broker/src/runtime/fleet.rs +++ b/crates/broker/src/runtime/fleet.rs @@ -802,9 +802,14 @@ impl BrokerRuntime { /// 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" | "delivery.deferred") => { + (ContextTopic::Agent, "delivery.failed") => { let workers = self.fleet_context_update_workers(&update); if workers.is_empty() { debug_ignored_context_update(&update, "no matching worker"); @@ -812,6 +817,14 @@ impl BrokerRuntime { } self.surface_fleet_delivery_problem(&update, &workers).await; } + (ContextTopic::Agent, "delivery.deferred") => { + let workers = self.fleet_context_update_workers(&update); + if workers.is_empty() { + debug_ignored_context_update(&update, "no matching worker"); + return; + } + self.log_fleet_delivery_deferral(&update, &workers); + } (ContextTopic::Agent, "agent.identity_taken_over") => { let workers = self.fleet_context_update_workers(&update); if workers.is_empty() { @@ -827,7 +840,7 @@ impl BrokerRuntime { /// 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. - fn fleet_context_update_workers(&self, update: &ContextUpdate) -> Vec { + 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 { @@ -841,12 +854,16 @@ impl BrokerRuntime { workers } - /// Tell the SENDING agent (and operators) that Relaycast could not land a - /// message it sent — the recipient was offline, its node was gone, or the - /// delivery was deferred. 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. + /// 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()); @@ -886,6 +903,36 @@ impl BrokerRuntime { } } + /// 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 diff --git a/crates/broker/src/runtime/tests.rs b/crates/broker/src/runtime/tests.rs index b53dfa228..3cab42af2 100644 --- a/crates/broker/src/runtime/tests.rs +++ b/crates/broker/src/runtime/tests.rs @@ -1484,6 +1484,8 @@ 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, @@ -1548,10 +1550,12 @@ async fn relaycast_delivery_failure_reaches_the_sending_worker() { ); } -/// A deferred delivery is surfaced through the same channel, with the engine's -/// event name kept in the reason so a consumer can tell the two apart. +/// 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_is_labelled_as_deferred() { +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()); @@ -1560,33 +1564,40 @@ async fn relaycast_delivery_deferral_is_labelled_as_deferred() { .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(context_update_event( - ContextTopic::Agent, - "delivery.deferred", - &["agt_sender"], - json!({ - "delivery_id": "del_remote_2", - "message_id": "msg_remote_2", - "available_at": "2026-09-02T00:00:00.000Z", - "reason": "recipient_busy", - }), + .handle_fleet_control_event(FleetControlEvent::Message( + RelaycastToBroker::ContextUpdate(update), )) .await; - let frame = tokio::time::timeout(Duration::from_secs(1), fixture._sdk_out_rx.recv()) - .await - .expect("a deferral must be emitted") - .expect("sdk_out_tx should remain open"); - assert_eq!(frame.payload["kind"], "message_delivery_failed"); - assert_eq!( - frame.payload["lastError"], - "relaycast delivery.deferred: recipient_busy" + 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" ); - // No target agent on a deferral payload: the event still has to name a - // target rather than panicking or emitting an empty one. - assert_eq!(frame.payload["to"], "unknown"); } /// Frames naming an agent this broker does not host, and events it has no use diff --git a/tests/relayflows/cases/1615-context-update-frames/run.mjs b/tests/relayflows/cases/1615-context-update-frames/run.mjs index 9ca748a1f..d946eae7e 100644 --- a/tests/relayflows/cases/1615-context-update-frames/run.mjs +++ b/tests/relayflows/cases/1615-context-update-frames/run.mjs @@ -256,8 +256,12 @@ try { 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. - RUST_LOG: 'info,relay_broker::runtime::fleet=debug', + // 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', }, } From 2769f9e3bdb2d3206b716cb7f8cf59025d05dd10 Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 2 Sep 2026 12:49:43 +0000 Subject: [PATCH 4/6] refactor(broker): share the context.update worker guard and lead the changelog with impact Review follow-up on relay#1615. - runtime/fleet: the `delivery.failed`, `delivery.deferred` and `agent.identity_taken_over` arms of `handle_fleet_context_update` each repeated the same resolve-then-debug-drop guard. Extracted it as `resolve_context_update_workers`, which returns `None` (after logging the shared "no matching worker" debug drop) when this broker hosts none of the addressed agents, so each arm is a single `let Some(workers) = ... else`. Behaviour is identical; `fleet_context_update_workers` stays `pub(super)` because `runtime/tests.rs` exercises it directly. - CHANGELOG: lead the `delivery.failed` bullet with the user-visible impact (a DM to an unreachable agent is no longer silently lost) and demote the frame-parsing fix to the parenthetical it is. - Trajectory traj_h0xx33q5a1ga: extend the record to the review follow-up commit e22a7a7 (commits list, `_trace.endRef`, summary Commits/Traced range). e22a7a7 touched no product file c919d2e had not, so `filesChanged` is unchanged. Hand-edited and re-validated with `agent-trajectories doctor` because `trail` cannot amend a completed trajectory. Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_01MB3K5bK7Jc5HM92fsZRUyS --- .../2026-09/traj_h0xx33q5a1ga/summary.md | 9 +++++- .../2026-09/traj_h0xx33q5a1ga/trajectory.json | 5 +-- CHANGELOG.md | 2 +- crates/broker/src/runtime/fleet.rs | 31 ++++++++++++------- 4 files changed, 31 insertions(+), 16 deletions(-) diff --git a/.agentworkforce/trajectories/completed/2026-09/traj_h0xx33q5a1ga/summary.md b/.agentworkforce/trajectories/completed/2026-09/traj_h0xx33q5a1ga/summary.md index 28c8781d6..46b0d8d9d 100644 --- a/.agentworkforce/trajectories/completed/2026-09/traj_h0xx33q5a1ga/summary.md +++ b/.agentworkforce/trajectories/completed/2026-09/traj_h0xx33q5a1ga/summary.md @@ -41,8 +41,15 @@ Broker parses Relaycast context.update node frames via a new ServerToNode::Conte ## Commits - `c919d2eeff0bf606a215141822571ca1fdbba5d3` +- `e22a7a712025bd2b325d607bff54f5a00c01cc06` -Traced range: `6d5199ff103cb5f4ff6adf0a3fa32a788646a9bc` .. `c919d2eeff0bf606a215141822571ca1fdbba5d3` +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 diff --git a/.agentworkforce/trajectories/completed/2026-09/traj_h0xx33q5a1ga/trajectory.json b/.agentworkforce/trajectories/completed/2026-09/traj_h0xx33q5a1ga/trajectory.json index 813d07190..ac85b2abb 100644 --- a/.agentworkforce/trajectories/completed/2026-09/traj_h0xx33q5a1ga/trajectory.json +++ b/.agentworkforce/trajectories/completed/2026-09/traj_h0xx33q5a1ga/trajectory.json @@ -59,7 +59,8 @@ "confidence": 0.85 }, "commits": [ - "c919d2eeff0bf606a215141822571ca1fdbba5d3" + "c919d2eeff0bf606a215141822571ca1fdbba5d3", + "e22a7a712025bd2b325d607bff54f5a00c01cc06" ], "filesChanged": [ "CHANGELOG.md", @@ -77,6 +78,6 @@ "tags": [], "_trace": { "startRef": "6d5199ff103cb5f4ff6adf0a3fa32a788646a9bc", - "endRef": "c919d2eeff0bf606a215141822571ca1fdbba5d3" + "endRef": "e22a7a712025bd2b325d607bff54f5a00c01cc06" } } diff --git a/CHANGELOG.md b/CHANGELOG.md index 41fe6884d..aaf69bbea 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -10,7 +10,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Fixed - `node agent message flush` and `node agent message auto` now unblock a held queue that could previously never drain, instead of reporting `flushed: 0` forever and leaving the agent unable to receive anything later. A parked message whose Relaycast identity has been retired is dead-lettered with a reason and is visible in `node deadletters`; injection failures and out-of-order sequences are still held for retry. -- Broker now accepts Relaycast `context.update` node frames instead of logging every one as an invalid frame, and surfaces `delivery.failed` to the sending agent as a `message_delivery_failed` event so a DM to an unreachable agent is no longer silently lost. +- 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 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. ### Added diff --git a/crates/broker/src/runtime/fleet.rs b/crates/broker/src/runtime/fleet.rs index 22a755de3..a67600bd7 100644 --- a/crates/broker/src/runtime/fleet.rs +++ b/crates/broker/src/runtime/fleet.rs @@ -810,33 +810,40 @@ impl BrokerRuntime { async fn handle_fleet_context_update(&mut self, update: ContextUpdate) { match (update.topic, update.event.as_str()) { (ContextTopic::Agent, "delivery.failed") => { - let workers = self.fleet_context_update_workers(&update); - if workers.is_empty() { - debug_ignored_context_update(&update, "no matching worker"); + let Some(workers) = self.resolve_context_update_workers(&update) else { return; - } + }; self.surface_fleet_delivery_problem(&update, &workers).await; } (ContextTopic::Agent, "delivery.deferred") => { - let workers = self.fleet_context_update_workers(&update); - if workers.is_empty() { - debug_ignored_context_update(&update, "no matching worker"); + 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 workers = self.fleet_context_update_workers(&update); - if workers.is_empty() { - debug_ignored_context_update(&update, "no matching worker"); + 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. From 8e4ea08a1a4271b0ce92f19e9850608fc3d7d641 Mon Sep 17 00:00:00 2001 From: "github-actions[bot]" Date: Thu, 3 Sep 2026 13:36:55 +0000 Subject: [PATCH 5/6] style: auto-format with Prettier --- .trajectories/compacted/release-11.10.2.json | 19 ++++--------------- .trajectories/compacted/release-11.10.2.md | 8 +++++++- 2 files changed, 11 insertions(+), 16 deletions(-) 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 From 0248739ce1cd965c8f9a56062210e60906abe47c Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 3 Sep 2026 13:56:19 +0000 Subject: [PATCH 6/6] docs(changelog): split the context.update fix into one bullet per change Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_01MB3K5bK7Jc5HM92fsZRUyS --- CHANGELOG.md | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index e5d52927b..23c10d8cc 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,7 +9,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### 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 instead of being logged as invalid). +- 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