fix(broker): accept context.update node frames and act on delivery failures and identity takeover

.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`

.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"
+ }
+}

.trajectories/compacted/release-11.10.2.json

@@ -3,19 +3,15 @@
  "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"
  },
  "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"
- ]
-}
+ "commits": ["e85f4ca23", "bf66a2e86", "13971aa8f", "1191af9bd", "cefc1ab4d", "2559e668a"]
+}

.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,20 +10,25 @@
 - 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)
 - Applied independent review findings without weakening the effect oracle -> Applied independent review findings without weakening the effect oracle (traj_7yref3wye283)
 - 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
+
+- None

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
 

crates/broker/src/fleet_wire.rs

@@ -601,6 +601,48 @@ pub struct ActionInvoke {
  pub agent_name: Option<String>,
 }
 
+/// 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<String>,
+ /// 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<Vec<String>>,
+ /// 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::<RelaycastToBroker>(json!({
+ "type": "context.update",
+ "v": 1,
+ "topic": "galaxy",
+ "event": "agent.status.active",
+ "data": {}
+ }))
+ .is_err());
+ }
 }