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/relaycast/ws.rs b/crates/broker/src/relaycast/ws.rs index 75d6ea3e0..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, @@ -239,18 +252,27 @@ 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 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, 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 +486,57 @@ 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 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. **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 + /// 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 +687,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 f9a591fe4..4f2b2933a 100644 --- a/crates/broker/src/runtime/api.rs +++ b/crates/broker/src/runtime/api.rs @@ -695,14 +695,27 @@ 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(); + // 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!({ @@ -718,356 +731,131 @@ 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" + publish_from = %publish_from, + 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(), + publish_from, + thread_id.as_deref(), + ), + ) + .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();