From 253f43158bb35c112d43f2b8529ee2a00f7baa47 Mon Sep 17 00:00:00 2001 From: Will Washburn Date: Tue, 30 Jun 2026 21:10:06 -0700 Subject: [PATCH 1/3] fix(broker): route all message delivery through Relaycast, no local bypass MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ListenApiRequest::Send (used by both the HTTP API and the agent-facing MCP send_message tool) resolved locally-attached worker targets and injected directly into their PTY, only falling back to publishing via Relaycast when no local target existed. Any client that only observes state through Relaycast (a hosted observer, a teammate's Pear, cross-device sync) would silently miss messages whenever sender and recipient happened to share a broker, even though the recipient visibly received and replied to them. All delivery is now relaycast-mediated unconditionally: the send handler always publishes via Relaycast and relies on node/fleet redelivery (handle_fleet_deliver) to reach locally-attached workers, exactly as it would for any other client, whether Relaycast is the hosted service or a local Relaycast host. Wire handle_fleet_deliver into the same per-worker InboundDeliveryMode choke point (queue_inbound_for_delivery_mode) the removed local-delivery path used, so manual_flush keeps working now that node delivery is the only delivery path — it was previously only honored for the local bypass and silently ignored for genuine Relaycast-originated deliveries. Removes the now-fully-dead local-target routing helpers from production use (routing.rs, worker.rs wrappers, http_api_local_delivery_timeout); kept rather than deleted, matching this crate's existing #[allow(dead_code)] convention for modules retained for their unit tests. Co-Authored-By: Claude Sonnet 5 --- crates/broker/src/lib.rs | 6 + crates/broker/src/runtime/api.rs | 408 +++++++---------------------- crates/broker/src/runtime/fleet.rs | 91 ++++++- crates/broker/src/runtime/mod.rs | 1 + crates/broker/src/runtime/util.rs | 5 + crates/broker/src/worker.rs | 38 --- 6 files changed, 189 insertions(+), 360 deletions(-) diff --git a/crates/broker/src/lib.rs b/crates/broker/src/lib.rs index 0afa6812a..48d19240c 100644 --- a/crates/broker/src/lib.rs +++ b/crates/broker/src/lib.rs @@ -41,6 +41,12 @@ pub(crate) mod redact; #[allow(dead_code)] pub(crate) mod relaycast; pub(crate) mod replay_buffer; +// Local-target routing helpers, kept for their unit tests but no longer +// called from production code: the HTTP/sidecar send path (runtime/api.rs) +// no longer resolves local targets and injects directly — it always +// publishes through Relaycast and lets node delivery (runtime/fleet.rs) +// redeliver, even to workers attached to this same broker. +#[allow(dead_code)] pub(crate) mod routing; pub(crate) mod runtime; #[allow(dead_code)] diff --git a/crates/broker/src/runtime/api.rs b/crates/broker/src/runtime/api.rs index f9a591fe4..66acc2fb5 100644 --- a/crates/broker/src/runtime/api.rs +++ b/crates/broker/src/runtime/api.rs @@ -695,11 +695,7 @@ impl BrokerRuntime { normalized_sender }; let event_id = format!("http_{}", Uuid::new_v4().simple()); - let priority = if normalized_to.starts_with('#') { 3 } else { 2 }; - let mut delivered = 0usize; - let mut delivery_errors = 0usize; let request_start = Instant::now(); - let local_delivery_timeout = http_api_local_delivery_timeout(); let relaycast_timeout = http_api_relaycast_send_timeout(); let event_emit_timeout = http_api_event_emit_timeout(); @@ -718,356 +714,128 @@ impl BrokerRuntime { }), ); - let targets = if normalized_to.starts_with('#') { - workers.worker_names_for_channel_delivery( - &normalized_to, - &delivery_from, - Some(&selected_workspace_id), - ) - } else { - workers.worker_names_for_direct_target( - &normalized_to, - &delivery_from, - Some(&selected_workspace_id), - ) - }; - + // All delivery is relaycast-mediated, with no local-injection + // shortcut and no fallback switch on whether a recipient + // happens to be attached to this broker. Even when the + // target is a worker running right here, we publish to + // Relaycast (cloud-hosted or a local Relaycast host — either + // way, wherever Relaycast is) and let it redeliver over the + // node control plane (see `handle_fleet_deliver`), exactly + // as it would for any other client. A broker-local shortcut + // would let a message reach a worker's PTY without Relaycast + // ever seeing it, so anything that only observes state + // through Relaycast (a hosted observer, a teammate's Pear, + // cross-device sync) would silently miss messages that the + // sender's own terminal still showed a reply to. tracing::info!( target = "relay_broker::http_api", event_id = %event_id, to = %normalized_to, delivery_from = %delivery_from, - target_count = %targets.len(), - "resolved HTTP API send targets" + ui_from = %ui_from, + relaycast_timeout_ms = %relaycast_timeout.as_millis(), + "publishing to relaycast" ); + let relaycast_start = Instant::now(); + match timeout( + relaycast_timeout, + selected_workspace.http_client.send_with_mode( + &normalized_to, + &text, + mode.clone(), + ), + ) + .await + { + Ok(Ok(())) => { + tracing::info!( + target = "relay_broker::http_api", - for worker_name in targets { - // Inbound-delivery queue: every inbound message - // enters the per-worker FIFO first. `auto_inject` - // drains immediately; `manual_flush` holds and - // counts as delivered so the HTTP caller's ack - // semantics are unchanged. We pass the FULL - // routing context so any drain reproduces the - // original delivery (channel/thread/workspace - // /priority/mode), not a stripped-down DM. - let queue_result = queue_inbound_for_delivery_mode( - delivery_states, - workers, - &worker_name, - InboundContext { - from: &delivery_from, - body: &text, - target: &normalized_to, - thread_id: thread_id.as_deref(), - workspace_id: Some(selected_workspace_id.as_str()), - workspace_alias: selected_workspace_alias.as_deref(), - priority, - mode: mode.clone(), - event_id: Some(&event_id), - }, - ); - if let Some(dropped_from) = &queue_result.evicted_from { - let _ = send_broker_event( + event_id = %event_id, + to = %normalized_to, + relaycast_ms = %relaycast_start.elapsed().as_millis(), + "relaycast publish succeeded" + ); + emit_http_api_event_with_timeout( sdk_out_tx, - delivery_dropped_event_for_eviction(&worker_name, dropped_from), + json!({ + "kind": "relay_inbound", + "event_id": event_id, + "from": ui_from, + "target": normalized_to, + "body": text, + "thread_id": thread_id.clone(), + "workspace_id": selected_workspace_id.clone(), + "workspace_alias": selected_workspace_alias.clone(), + }), + event_emit_timeout, ) .await; - } - match queue_result.outcome { - InboundQueueOutcome::Queued => { - delivered = delivered.saturating_add(1); - tracing::info!( - target = "relay_broker::http_api", - event_id = %event_id, - to = %normalized_to, - worker = %worker_name, - "queued local delivery (manual_flush inbound delivery mode)" - ); - let _ = send_event( - sdk_out_tx, - json!({ - "kind":"delivery_queued", - "name":&worker_name, - "event_id":&event_id, - "from":&delivery_from, - "target":&normalized_to, - "reason":"inbound_delivery_manual_flush", - }), - ) - .await; - continue; - } - InboundQueueOutcome::DrainNow(to_drain) => { - for queued in to_drain { - let queued_event_id = queued.event_id.as_deref().unwrap_or(""); - let is_current = - queued.event_id.as_deref() == Some(event_id.as_str()); - match timeout( - local_delivery_timeout, - try_inject_pending_relay_message( - workers, - pending_deliveries, - &worker_name, - &queued, - delivery_retry_interval, - ), - ) - .await - { - Ok(Ok(_)) => { - if is_current { - delivered = delivered.saturating_add(1); - } - } - Ok(Err(error)) => { - if is_current { - delivery_errors = delivery_errors.saturating_add(1); - } - tracing::warn!( - target = "relay_broker::http_api", - - event_id = %queued_event_id, - to = %queued.target, - worker = %worker_name, - error = %error, - "local delivery attempt failed" - ); - } - Err(_) => { - if is_current { - delivery_errors = delivery_errors.saturating_add(1); - } - tracing::warn!( - target = "relay_broker::http_api", - - event_id = %queued_event_id, - to = %queued.target, - worker = %worker_name, - timeout_ms = %local_delivery_timeout.as_millis(), - "local delivery attempt timed out" - ); - } - } - } - continue; - } - InboundQueueOutcome::WorkerMissing => { - // Fall through so the standard - // not-found accounting path runs. - } - } - match timeout( - local_delivery_timeout, - queue_and_try_delivery_raw( - workers, - pending_deliveries, - &worker_name, - &event_id, - &delivery_from, - &normalized_to, - &text, - thread_id.clone(), - Some(selected_workspace_id.clone()), - selected_workspace_alias.clone(), - priority, - mode.clone(), - delivery_retry_interval, - ), - ) - .await - { - Ok(Ok(_)) => { - delivered = delivered.saturating_add(1); - } - Ok(Err(error)) => { - delivery_errors = delivery_errors.saturating_add(1); - tracing::warn!( - target = "relay_broker::http_api", - - event_id = %event_id, - to = %normalized_to, - worker = %worker_name, - error = %error, - "local delivery attempt failed" - ); - } - Err(_) => { - delivery_errors = delivery_errors.saturating_add(1); + if reply + .send(Ok(json!({ + "success": true, + "event_id": event_id, + "relaycast_published": true, + "local": false, + "workspace_id": selected_workspace_id, + "workspace_alias": selected_workspace_alias, + }))) + .is_err() + { tracing::warn!( target = "relay_broker::http_api", event_id = %event_id, - to = %normalized_to, - worker = %worker_name, - timeout_ms = %local_delivery_timeout.as_millis(), - "local delivery attempt timed out" + "broker HTTP API reply channel closed before relaycast response" ); } } - } - - if delivered > 0 { - tracing::info!( - target = "relay_broker::http_api", - - event_id = %event_id, - to = %normalized_to, - delivery_from = %delivery_from, - ui_from = %ui_from, - delivered = %delivered, - "local delivery succeeded" - ); - emit_http_api_event_with_timeout( - sdk_out_tx, - json!({ - "kind": "relay_inbound", - "event_id": event_id, - "from": ui_from, - "target": normalized_to, - "body": text, - "thread_id": thread_id.clone(), - "workspace_id": selected_workspace_id.clone(), - "workspace_alias": selected_workspace_alias.clone(), - }), - event_emit_timeout, - ) - .await; - if reply - .send(Ok(json!({ - "success": true, - "event_id": event_id, - "delivered": delivered, - "local": true, - "workspace_id": selected_workspace_id, - "workspace_alias": selected_workspace_alias, - }))) - .is_err() - { + Ok(Err(error)) => { tracing::warn!( target = "relay_broker::http_api", event_id = %event_id, - "broker HTTP API reply channel closed before local delivery response" + to = %normalized_to, + relaycast_ms = %relaycast_start.elapsed().as_millis(), + error = %error, + "relaycast publish failed" ); - } - } else { - tracing::info!( - target = "relay_broker::http_api", - - event_id = %event_id, - to = %normalized_to, - mode = ?mode, - delivery_errors = %delivery_errors, - delivery_from = %delivery_from, - ui_from = %ui_from, - relaycast_timeout_ms = %relaycast_timeout.as_millis(), - "no local deliveries succeeded; forwarding to relaycast" - ); - let relaycast_start = Instant::now(); - match timeout( - relaycast_timeout, - selected_workspace.http_client.send_with_mode( - &normalized_to, - &text, - mode.clone(), - ), - ) - .await - { - Ok(Ok(())) => { - tracing::info!( - target = "relay_broker::http_api", - - event_id = %event_id, - to = %normalized_to, - relaycast_ms = %relaycast_start.elapsed().as_millis(), - "relaycast publish succeeded" - ); - emit_http_api_event_with_timeout( - sdk_out_tx, - json!({ - "kind": "relay_inbound", - "event_id": event_id, - "from": ui_from, - "target": normalized_to, - "body": text, - "thread_id": thread_id.clone(), - "workspace_id": selected_workspace_id.clone(), - "workspace_alias": selected_workspace_alias.clone(), - }), - event_emit_timeout, - ) - .await; - if reply - .send(Ok(json!({ - "success": true, - "event_id": event_id, - "relaycast_published": true, - "local": false, - "workspace_id": selected_workspace_id, - "workspace_alias": selected_workspace_alias, - }))) - .is_err() - { - tracing::warn!( - target = "relay_broker::http_api", - - event_id = %event_id, - "broker HTTP API reply channel closed before relaycast response" - ); - } - } - Ok(Err(error)) => { + if reply + .send(Err(format!("Relaycast publish failed: {error}"))) + .is_err() + { tracing::warn!( target = "relay_broker::http_api", event_id = %event_id, - to = %normalized_to, - relaycast_ms = %relaycast_start.elapsed().as_millis(), - error = %error, - "relaycast publish failed" + "broker HTTP API reply channel closed before relaycast failure response" ); - let not_found = format!("Agent \"{}\" not found", normalized_to); - if reply - .send(Err(format!( - "{not_found} and Relaycast publish failed: {error}" - ))) - .is_err() - { - tracing::warn!( - target = "relay_broker::http_api", - - event_id = %event_id, - "broker HTTP API reply channel closed before relaycast failure response" - ); - } } - Err(_) => { + } + Err(_) => { + tracing::warn!( + target = "relay_broker::http_api", + + event_id = %event_id, + to = %normalized_to, + relaycast_timeout_ms = %relaycast_timeout.as_millis(), + relaycast_ms = %relaycast_start.elapsed().as_millis(), + "relaycast publish timed out" + ); + if reply + .send(Err(format!( + "Relaycast publish timed out after {}ms", + relaycast_timeout.as_millis() + ))) + .is_err() + { tracing::warn!( target = "relay_broker::http_api", event_id = %event_id, - to = %normalized_to, - relaycast_timeout_ms = %relaycast_timeout.as_millis(), - relaycast_ms = %relaycast_start.elapsed().as_millis(), - "relaycast publish timed out" + "broker HTTP API reply channel closed before relaycast timeout response" ); - let not_found = format!("Agent \"{}\" not found", normalized_to); - if reply - .send(Err(format!( - "{not_found} and Relaycast publish timed out after {}ms", - relaycast_timeout.as_millis() - ))) - .is_err() - { - tracing::warn!( - target = "relay_broker::http_api", - - event_id = %event_id, - "broker HTTP API reply channel closed before relaycast timeout response" - ); - } } } } diff --git a/crates/broker/src/runtime/fleet.rs b/crates/broker/src/runtime/fleet.rs index cb79bc3f2..61a0eb861 100644 --- a/crates/broker/src/runtime/fleet.rs +++ b/crates/broker/src/runtime/fleet.rs @@ -388,9 +388,96 @@ impl BrokerRuntime { .and_then(Value::as_str) .unwrap_or(""); match classify_fleet_delivery(payload_type) { + // Route through the same per-worker InboundDeliveryMode choke + // point as the HTTP/sidecar send path (`queue_inbound_for_delivery_mode` + // in runtime/delivery.rs). Node delivery is the ONLY delivery + // path now that direct local injection has been removed from + // `ListenApiRequest::Send`, so if manual_flush isn't honored + // here, it's never honored anywhere. FleetDeliverySurfacing::Inject => { - let relay_delivery = self.fleet_relay_delivery(deliver); - self.workers.deliver(&deliver.agent, relay_delivery).await + let fields = fleet_delivery_fields(&deliver.payload, &deliver.agent); + let injection_mode = match deliver.mode { + DeliveryMode::Wait => MessageInjectionMode::Wait, + DeliveryMode::Steer => MessageInjectionMode::Steer, + }; + let priority = fields + .priority + .unwrap_or(if fields.target.starts_with('#') { 3 } else { 2 }); + let queue_result = queue_inbound_for_delivery_mode( + &mut self.delivery_states, + &self.workers, + &deliver.agent, + InboundContext { + from: &fields.from, + body: &fields.body, + target: &fields.target, + thread_id: fields.thread_id.as_deref(), + workspace_id: self.default_workspace_id.as_deref(), + workspace_alias: self.default_workspace.workspace_alias.as_deref(), + priority, + mode: injection_mode, + event_id: Some(&deliver.msg_id), + }, + ); + if let Some(dropped_from) = &queue_result.evicted_from { + let _ = send_broker_event( + &self.sdk_out_tx, + delivery_dropped_event_for_eviction(&deliver.agent, dropped_from), + ) + .await; + } + match queue_result.outcome { + InboundQueueOutcome::Queued => { + tracing::info!( + target = "relay_broker::fleet", + agent = %deliver.agent, + delivery_id = %deliver.delivery_id, + msg_id = %deliver.msg_id, + "queued node delivery (manual_flush inbound delivery mode)" + ); + Ok(()) + } + InboundQueueOutcome::DrainNow(to_drain) => { + // Mirrors the HTTP send path: drain may surface older + // backlog alongside the message this specific `deliver` + // frame is for. Only a failure injecting THIS delivery's + // own message should withhold the ack (causing the + // engine to redeliver it); backlog injection failures + // are logged and otherwise don't block the ack, since + // their own delivery frames already governed their acks. + let mut current_result = Ok(()); + for queued in to_drain { + let is_current = + queued.event_id.as_deref() == Some(deliver.msg_id.as_str()); + if let Err(error) = try_inject_pending_relay_message( + &mut self.workers, + &mut self.pending_deliveries, + &deliver.agent, + &queued, + self.delivery_retry_interval, + ) + .await + { + if is_current { + current_result = Err(error); + } else { + tracing::warn!( + target = "relay_broker::fleet", + agent = %deliver.agent, + from = %queued.from, + error = %error, + "failed to inject drained backlog message" + ); + } + } + } + current_result + } + InboundQueueOutcome::WorkerMissing => { + let relay_delivery = self.fleet_relay_delivery(deliver); + self.workers.deliver(&deliver.agent, relay_delivery).await + } + } } FleetDeliverySurfacing::AckOnly => { tracing::info!( diff --git a/crates/broker/src/runtime/mod.rs b/crates/broker/src/runtime/mod.rs index bddcd9e73..b0c472e92 100644 --- a/crates/broker/src/runtime/mod.rs +++ b/crates/broker/src/runtime/mod.rs @@ -61,6 +61,7 @@ use crate::{broker, listen_api, worker_request}; const DEFAULT_DELIVERY_RETRY_MS: u64 = 1_000; const MAX_DELIVERY_RETRIES: u32 = 10; const THREAD_HISTORY_LIMIT: usize = 1_000; +#[allow(dead_code)] // only http_api_local_delivery_timeout's default; see its own allow const DEFAULT_HTTP_API_LOCAL_DELIVERY_TIMEOUT_MS: u64 = 3_000; const DEFAULT_HTTP_API_RELAYCAST_SEND_TIMEOUT_MS: u64 = 20_000; const DEFAULT_HTTP_API_EVENT_EMIT_TIMEOUT_MS: u64 = 200; diff --git a/crates/broker/src/runtime/util.rs b/crates/broker/src/runtime/util.rs index 070a1a617..98d75e8fa 100644 --- a/crates/broker/src/runtime/util.rs +++ b/crates/broker/src/runtime/util.rs @@ -237,6 +237,11 @@ pub(crate) fn delivery_retry_interval() -> Duration { Duration::from_millis(ms.max(50)) } +// No longer called from production code — the HTTP/sidecar send path +// (runtime/api.rs) no longer attempts direct local delivery, so there's +// nothing left to bound with a "local delivery" timeout. Kept (with its +// env-var override still covered by unit tests) rather than deleted. +#[allow(dead_code)] pub(crate) fn http_api_local_delivery_timeout() -> Duration { let ms = std::env::var("AGENT_RELAY_HTTP_API_LOCAL_DELIVERY_TIMEOUT_MS") .ok() diff --git a/crates/broker/src/worker.rs b/crates/broker/src/worker.rs index 32c566237..5ff7149bd 100644 --- a/crates/broker/src/worker.rs +++ b/crates/broker/src/worker.rs @@ -29,7 +29,6 @@ use tokio::{ use crate::{ cli::command_parse::{normalize_cli_name, parse_cli_command}, - routing, runtime::headless_provider_cli_name, spawner::terminate_child, }; @@ -975,37 +974,6 @@ impl WorkerRegistry { } Ok(exited) } - - pub(crate) fn routing_workers(&self) -> Vec> { - self.workers - .iter() - .map(|(name, handle)| routing::RoutingWorker { - name, - channels: &handle.spec.channels, - workspace_id: handle.workspace_id.as_deref(), - }) - .collect() - } - - pub(crate) fn worker_names_for_channel_delivery( - &self, - channel: &str, - from: &str, - workspace_id: Option<&str>, - ) -> Vec { - let workers = self.routing_workers(); - routing::worker_names_for_channel_delivery(&workers, channel, from, workspace_id) - } - - pub(crate) fn worker_names_for_direct_target( - &self, - target: &str, - from: &str, - workspace_id: Option<&str>, - ) -> Vec { - let workers = self.routing_workers(); - routing::worker_names_for_direct_target(&workers, target, from, workspace_id) - } } fn release_policy_arg(policy: Option<&HarnessReleasePolicy>) -> &'static str { @@ -1720,12 +1688,6 @@ mod tests { assert_eq!(release_grace_for_spec(&spec), APP_SERVER_RELEASE_GRACE); } - #[test] - fn routing_workers_empty_when_no_workers() { - let reg = make_registry(vec![]); - assert!(reg.routing_workers().is_empty()); - } - #[test] fn prepare_claude_session_args_generates_uuid_session_id() { let mut args = Vec::new(); From b1b058826232837b9227c8e466366ba95ec79751 Mon Sep 17 00:00:00 2001 From: Will Washburn Date: Tue, 30 Jun 2026 21:27:08 -0700 Subject: [PATCH 2/3] fix(broker): preserve sender identity and thread grouping on Relaycast publish MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Review feedback (coderabbitai, cubic-dev-ai) on #1219: since the send handler now always publishes through Relaycast (no local-injection fallback), send_with_mode's from/thread_id being dropped at the relay boundary went from a narrow, rarely-hit gap to affecting every message — breaking sender attribution and thread/conversation grouping downstream. - Impersonate the original `from` via registered_agent_client_as (same impersonation-by-design pattern already used by mark_read_as_agent) instead of always posting as the broker's own registered identity. - Post via AgentClient::reply when thread_id is present so Relaycast actually records a thread reply, instead of a plain channel post that drops the parent-message link entirely. Co-Authored-By: Claude Sonnet 5 --- crates/broker/src/relaycast/ws.rs | 56 +++++++++++++++++++++++++------ crates/broker/src/runtime/api.rs | 2 ++ 2 files changed, 48 insertions(+), 10 deletions(-) diff --git a/crates/broker/src/relaycast/ws.rs b/crates/broker/src/relaycast/ws.rs index 75d6ea3e0..8d670a0da 100644 --- a/crates/broker/src/relaycast/ws.rs +++ b/crates/broker/src/relaycast/ws.rs @@ -239,18 +239,26 @@ impl RelaycastHttpClient { /// Send a direct message to a named agent via the Relaycast REST API. pub async fn send_dm(&self, to: &str, text: &str) -> Result<()> { - self.send_dm_with_mode(to, text, MessageInjectionMode::Wait) + self.send_dm_with_mode(to, text, MessageInjectionMode::Wait, &self.agent_name) .await } /// Send a direct message with explicit injection mode via the Relaycast REST API. + /// + /// `from` is impersonated via [`registered_agent_client_as`] rather than + /// always posting as this broker's own registered identity (the same + /// impersonation-by-design pattern as [`mark_read_as_agent`]) — a DM + /// forwarded from a locally-attached worker must be attributed to that + /// worker's own Relaycast identity, not the broker's, or sender identity + /// is lost at the relay boundary. pub async fn send_dm_with_mode( &self, to: &str, text: &str, mode: MessageInjectionMode, + from: &str, ) -> Result<()> { - let agent_client = self.registered_agent_client().await?; + let agent_client = self.registered_agent_client_as(from, None).await?; let relay_mode = match mode { MessageInjectionMode::Wait => relaycast::MessageInjectionMode::Wait, MessageInjectionMode::Steer => relaycast::MessageInjectionMode::Steer, @@ -464,31 +472,53 @@ impl RelaycastHttpClient { /// Smart send: routes to channel or DM based on `#` prefix. pub async fn send(&self, to: &str, text: &str) -> Result<()> { - self.send_with_mode(to, text, MessageInjectionMode::Wait) + self.send_with_mode(to, text, MessageInjectionMode::Wait, &self.agent_name, None) .await } /// Smart send with explicit injection mode. + /// + /// `from` is impersonated via [`registered_agent_client_as`] (see + /// [`send_dm_with_mode`]) so the Relaycast-recorded sender matches the + /// original request's `from` rather than always this broker's own + /// identity — this is the only delivery path now (no local-injection + /// bypass), so every send's sender attribution flows through here. + /// + /// `thread_id`, when present, is a Relaycast message id to reply to + /// (channel targets only — Relaycast DMs have no thread concept): posting + /// via [`AgentClient::reply`] instead of a plain channel post is what + /// actually creates real thread/conversation grouping on the Relaycast + /// side, as opposed to passing an opaque value the server doesn't + /// interpret as a reply. pub async fn send_with_mode( &self, to: &str, text: &str, mode: MessageInjectionMode, + from: &str, + thread_id: Option<&str>, ) -> Result<()> { if to.starts_with('#') { - let agent_client = self.registered_agent_client().await?; + let agent_client = self.registered_agent_client_as(from, None).await?; let relay_mode = match mode { MessageInjectionMode::Wait => relaycast::MessageInjectionMode::Wait, MessageInjectionMode::Steer => relaycast::MessageInjectionMode::Steer, }; - agent_client - .send_with_mode(to, text, None, None, relay_mode, None) - .await - .map_err(|e| anyhow::anyhow!("relaycast send_to_channel failed: {e}"))?; + if let Some(thread_id) = thread_id { + agent_client + .reply(thread_id, text, None, None) + .await + .map_err(|e| anyhow::anyhow!("relaycast thread reply failed: {e}"))?; + } else { + agent_client + .send_with_mode(to, text, None, None, relay_mode, None) + .await + .map_err(|e| anyhow::anyhow!("relaycast send_to_channel failed: {e}"))?; + } return Ok(()); } - self.send_dm_with_mode(to, text, mode).await + self.send_dm_with_mode(to, text, mode, from).await } } @@ -639,7 +669,13 @@ mod tests { let client = seeded_http_client(&server.base_url()); client - .send_with_mode("worker-a", "interrupt", MessageInjectionMode::Steer) + .send_with_mode( + "worker-a", + "interrupt", + MessageInjectionMode::Steer, + "broker", + None, + ) .await .expect("relaycast DM steer send should succeed"); } diff --git a/crates/broker/src/runtime/api.rs b/crates/broker/src/runtime/api.rs index 66acc2fb5..c8a89f446 100644 --- a/crates/broker/src/runtime/api.rs +++ b/crates/broker/src/runtime/api.rs @@ -744,6 +744,8 @@ impl BrokerRuntime { &normalized_to, &text, mode.clone(), + &delivery_from, + thread_id.as_deref(), ), ) .await From bc2a2cdb7bbe2a8dc9997ca5925b06253a85ae05 Mon Sep 17 00:00:00 2001 From: Will Washburn Date: Tue, 30 Jun 2026 21:36:19 -0700 Subject: [PATCH 3/3] fix(broker): don't impersonate an unvalidated sender on Relaycast publish MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit registered_agent_client_as's underlying register_agent_token will silently register a brand-new Relaycast agent under whatever name it's given, or — if that name already exists (409) — ROTATE its token, invalidating whatever token that agent was already using. mark_read_as_agent (the existing precedent for this pattern) is safe because every caller passes a name the broker itself spawned; the new send path is not, since delivery_from is caller-supplied HTTP API / MCP input with no validation that it's actually a local worker. Only impersonate `from` when it's a name this broker has custodial responsibility for (a worker it spawned); fall back to the broker's own identity for anything else. Falling back is always safe — impersonating an arbitrary string risks silently disconnecting an unrelated agent that happens to share the name. Also tightens the doc comments on registered_agent_client_as/send_with_mode/ send_dm_with_mode to spell out this precondition instead of describing impersonation as an unconditionally safe pattern. Co-Authored-By: Claude Sonnet 5 --- crates/broker/src/relaycast/ws.rs | 36 +++++++++++++++++++++++-------- crates/broker/src/runtime/api.rs | 20 ++++++++++++++++- 2 files changed, 46 insertions(+), 10 deletions(-) diff --git a/crates/broker/src/relaycast/ws.rs b/crates/broker/src/relaycast/ws.rs index 8d670a0da..c8a1aa3c4 100644 --- a/crates/broker/src/relaycast/ws.rs +++ b/crates/broker/src/relaycast/ws.rs @@ -131,6 +131,19 @@ impl RelaycastHttpClient { .map_err(|error| anyhow::anyhow!("{error}")) } + /// Authenticate as `agent_name` rather than this broker's own identity. + /// + /// **Callers must only pass a name this broker has custodial + /// responsibility for** (a worker it spawned, or its own identity) — + /// this is not a safe way to relay an arbitrary, caller-supplied sender + /// label. Underneath, `AgentRegistrationClient::register_agent_token` + /// either registers a brand-new Relaycast agent under `agent_name` if + /// none exists, or — if one already does (409) — ROTATES its token, + /// invalidating whatever token that agent was already using. Passing an + /// unvalidated `agent_name` therefore risks silently disconnecting an + /// unrelated, already-registered agent that happens to share the name. + /// Validate against known-local names first; fall back to this + /// broker's own identity (`registered_agent_client`) for anything else. async fn registered_agent_client_as( &self, agent_name: &str, @@ -245,12 +258,13 @@ impl RelaycastHttpClient { /// Send a direct message with explicit injection mode via the Relaycast REST API. /// - /// `from` is impersonated via [`registered_agent_client_as`] rather than - /// always posting as this broker's own registered identity (the same - /// impersonation-by-design pattern as [`mark_read_as_agent`]) — a DM - /// forwarded from a locally-attached worker must be attributed to that - /// worker's own Relaycast identity, not the broker's, or sender identity - /// is lost at the relay boundary. + /// `from` is authenticated via [`registered_agent_client_as`] rather than + /// always posting as this broker's own registered identity, so a DM + /// forwarded from a locally-attached worker is attributed to that + /// worker's own Relaycast identity instead of losing sender identity at + /// the relay boundary. **The caller must validate `from` first** — + /// see [`registered_agent_client_as`]'s doc comment for why passing an + /// arbitrary, unvalidated sender label here is unsafe. pub async fn send_dm_with_mode( &self, to: &str, @@ -478,11 +492,15 @@ impl RelaycastHttpClient { /// Smart send with explicit injection mode. /// - /// `from` is impersonated via [`registered_agent_client_as`] (see + /// `from` is authenticated via [`registered_agent_client_as`] (see /// [`send_dm_with_mode`]) so the Relaycast-recorded sender matches the /// original request's `from` rather than always this broker's own - /// identity — this is the only delivery path now (no local-injection - /// bypass), so every send's sender attribution flows through here. + /// identity. **The caller must validate `from` first** (a locally-known + /// worker name or this broker's own identity) — see + /// [`registered_agent_client_as`]'s doc comment for why an arbitrary, + /// unvalidated sender label is unsafe to pass here. This is the only + /// delivery path now (no local-injection bypass), so every send's + /// sender attribution flows through here. /// /// `thread_id`, when present, is a Relaycast message id to reply to /// (channel targets only — Relaycast DMs have no thread concept): posting diff --git a/crates/broker/src/runtime/api.rs b/crates/broker/src/runtime/api.rs index c8a89f446..4f2b2933a 100644 --- a/crates/broker/src/runtime/api.rs +++ b/crates/broker/src/runtime/api.rs @@ -699,6 +699,23 @@ impl BrokerRuntime { let relaycast_timeout = http_api_relaycast_send_timeout(); let event_emit_timeout = http_api_event_emit_timeout(); + // Only impersonate `from` on the Relaycast publish (see + // send_with_mode's doc comment) when it's a name this broker + // actually has custodial responsibility for: a worker it + // spawned, or its own identity. `delivery_from` is otherwise + // caller-supplied and unvalidated — impersonating an + // arbitrary string would let any HTTP API caller silently + // register a brand-new Relaycast agent under that name, or + // worse, ROTATE (and thereby invalidate) the live token of + // an unrelated, already-registered agent that happens to + // share the name. Falling back to the broker's own identity + // is always safe; impersonation is not. + let publish_from = if workers.has_worker(&delivery_from) { + delivery_from.as_str() + } else { + workspace_self_name.as_str() + }; + record_thread_history_event( recent_thread_messages, json!({ @@ -733,6 +750,7 @@ impl BrokerRuntime { event_id = %event_id, to = %normalized_to, delivery_from = %delivery_from, + publish_from = %publish_from, ui_from = %ui_from, relaycast_timeout_ms = %relaycast_timeout.as_millis(), "publishing to relaycast" @@ -744,7 +762,7 @@ impl BrokerRuntime { &normalized_to, &text, mode.clone(), - &delivery_from, + publish_from, thread_id.as_deref(), ), )