diff --git a/CHANGELOG.md b/CHANGELOG.md index 4e9fb4648..dcb061e99 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -26,6 +26,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Fixed +- `HarnessDriverClient.spawn()` now polls the broker's startup handshake for the full `startupTimeoutMs` budget (default 45s) instead of a fixed ~10s, so a slow-but-healthy Relaycast handshake that keeps answering `503` while warming up is no longer misreported as a spawn failure. - `agent-relay integration subscribe` now resolves provider-native `--resource` values through relayfile before binding, so Slack channel names, GitHub repos, Linear team keys, and Telegram chats bind to matching relayfile VFS globs while explicit `/`-prefixed globs still work. - `agent-relay integration subscribe` is now idempotent and supports multiple resources/channels per provider. Each inbound webhook is scoped to its `(provider, resource)` binding (not one-per-provider), so subscribing a second Slack channel — or two sources into the same relay channel — no longer collides on the unique `(workspace, webhook name)` index or clobbers the other binding's webhook. Re-subscribing creates the replacement webhook/subscription before retiring the old one, so a transient failure can't leave you with no working binding; a failed cleanup now warns instead of being silently swallowed. The relay channel id is normalized (`#general` → `general`) consistently across the webhook, subscription filter, relayfile bind, and writeback-secret lookup, and `listBindings` now maps relayfile's `pathGlob` field so unsubscribe/replace match correctly. - `agent-relay-broker` bootstrap `node.register` no longer advertises a generic `"spawn"` capability. Because the engine does not treat bare `"spawn"` as a placement capability (only `spawn:*`), it materialized a `spawn` action pinned to whichever node bootstrapped first, which then hijacked capability-based spawn placement for the whole workspace — every `spawn` invoke was dispatched to that node, ignoring `cli`/`target_node`/least-loaded routing. The pre-sidecar descriptor now carries no capabilities; real `spawn:*`/action capabilities arrive on the sidecar's `node.register`. diff --git a/crates/broker/src/relaycast/ws.rs b/crates/broker/src/relaycast/ws.rs index c8a1aa3c4..54fa8018d 100644 --- a/crates/broker/src/relaycast/ws.rs +++ b/crates/broker/src/relaycast/ws.rs @@ -507,7 +507,9 @@ impl RelaycastHttpClient { /// 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. + /// interpret as a reply. `reply` takes no injection mode, so a threaded + /// reply is always delivered with Wait semantics — a `Steer` request with + /// a `thread_id` is downgraded to a normal reply (logged, not dropped). pub async fn send_with_mode( &self, to: &str, @@ -523,6 +525,18 @@ impl RelaycastHttpClient { MessageInjectionMode::Steer => relaycast::MessageInjectionMode::Steer, }; if let Some(thread_id) = thread_id { + // `AgentClient::reply` has no injection-mode parameter, so a + // threaded reply is always delivered with Wait semantics. + // `Steer` can't be honored on a reply; downgrade rather than + // drop the message, but log it so the loss of steer is visible + // instead of silent. + if matches!(mode, MessageInjectionMode::Steer) { + tracing::warn!( + target = "relay_broker::relaycast", + thread_id = %thread_id, + "steer injection mode is not supported on threaded replies; delivering as a normal reply" + ); + } agent_client .reply(thread_id, text, None, None) .await diff --git a/crates/broker/src/runtime/api.rs b/crates/broker/src/runtime/api.rs index 4f2b2933a..039690123 100644 --- a/crates/broker/src/runtime/api.rs +++ b/crates/broker/src/runtime/api.rs @@ -709,12 +709,17 @@ impl BrokerRuntime { // 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() - }; + // is always safe; impersonation is not. The worker must also + // belong to the workspace we're publishing into — a worker + // attached to another attached workspace is not ours to + // impersonate here (it would register/rotate that name in the + // wrong Relaycast workspace). + let publish_from = + if workers.has_worker_in_workspace(&delivery_from, &selected_workspace_id) { + delivery_from.as_str() + } else { + workspace_self_name.as_str() + }; record_thread_history_event( recent_thread_messages, @@ -755,6 +760,24 @@ impl BrokerRuntime { relaycast_timeout_ms = %relaycast_timeout.as_millis(), "publishing to relaycast" ); + // Only forward `thread_id` to the Relaycast publish when it's a + // real message id we can reply to. Broker-minted synthetic ids + // (`http_*`) and channel/DM grouping keys (`#general`, + // `direct:*`) that a client may echo back from `/api/send` or + // `/api/threads` aren't reply targets — Relaycast would reject + // the reply and fail the whole send. Fall back to a plain post + // (unthreaded) for those, preserving delivery. + let reply_thread_id = thread_id + .as_deref() + .filter(|tid| is_relaycast_reply_target(tid)); + if thread_id.is_some() && reply_thread_id.is_none() { + tracing::debug!( + target = "relay_broker::http_api", + event_id = %event_id, + thread_id = ?thread_id, + "thread_id is not a Relaycast message id; publishing without a thread reply" + ); + } let relaycast_start = Instant::now(); match timeout( relaycast_timeout, @@ -763,7 +786,7 @@ impl BrokerRuntime { &text, mode.clone(), publish_from, - thread_id.as_deref(), + reply_thread_id, ), ) .await diff --git a/crates/broker/src/runtime/delivery.rs b/crates/broker/src/runtime/delivery.rs index c0dde6acb..3dc4c4c44 100644 --- a/crates/broker/src/runtime/delivery.rs +++ b/crates/broker/src/runtime/delivery.rs @@ -215,6 +215,20 @@ pub(crate) fn delivery_read_ack_is_relaycast_message(event_id: &EventId) -> bool synthetic_delivery_read_ack_reason(event_id).is_none() } +/// True when `thread_id` is a real Relaycast message id we can `reply()` to, +/// as opposed to a broker-minted synthetic event id (`http_`/`init_`/… — see +/// [`synthetic_delivery_read_ack_reason`]) or a channel/DM grouping key +/// (`#channel`, `direct:*`) that `/api/threads` can surface. Relaycast rejects +/// a reply to anything that isn't a real message id, so the publish path must +/// fall back to a plain post for these rather than fail the whole send. +pub(crate) fn is_relaycast_reply_target(thread_id: &str) -> bool { + let id = thread_id.trim(); + if id.is_empty() || id.starts_with('#') || id.starts_with("direct:") { + return false; + } + synthetic_delivery_read_ack_reason(&EventId::new(id)).is_none() +} + pub(crate) fn seed_supplied_agent_token( relaycast_http: &RelaycastHttpClient, agent_name: &str, @@ -903,3 +917,33 @@ pub(crate) fn clear_pending_delivery_if_event_matches( } None } + +#[cfg(test)] +mod reply_target_tests { + use super::is_relaycast_reply_target; + + #[test] + fn real_message_ids_are_reply_targets() { + assert!(is_relaycast_reply_target("msg_abc123")); + assert!(is_relaycast_reply_target("evt_01hxyz")); + } + + #[test] + fn synthetic_and_grouping_ids_are_not_reply_targets() { + for id in [ + "", + " ", + "#general", + "direct:alice", + "http_deadbeef", + "init_task", + "cont_load_1", + "flush_1", + ] { + assert!( + !is_relaycast_reply_target(id), + "expected non-target: {id:?}" + ); + } + } +} diff --git a/crates/broker/src/runtime/fleet.rs b/crates/broker/src/runtime/fleet.rs index 61a0eb861..1f4e0bf2d 100644 --- a/crates/broker/src/runtime/fleet.rs +++ b/crates/broker/src/runtime/fleet.rs @@ -435,6 +435,25 @@ impl BrokerRuntime { msg_id = %deliver.msg_id, "queued node delivery (manual_flush inbound delivery mode)" ); + // Surface the hold as a `delivery_queued` event, as the + // now-removed local send path did. `attach --drive` + // counts these to show pending messages; node delivery + // is the only delivery path now, so this is the only + // place the event can originate. The `name` field is + // what scopes it to the worker on the consumer side. + let _ = send_event( + &self.sdk_out_tx, + json!({ + "kind": "delivery_queued", + "name": deliver.agent.as_str(), + "event_id": deliver.msg_id.as_str(), + "delivery_id": deliver.delivery_id.as_str(), + "from": fields.from.as_str(), + "target": fields.target.as_str(), + "reason": "inbound_delivery_manual_flush", + }), + ) + .await; Ok(()) } InboundQueueOutcome::DrainNow(to_drain) => { diff --git a/crates/broker/src/worker.rs b/crates/broker/src/worker.rs index 5ff7149bd..823fc3b55 100644 --- a/crates/broker/src/worker.rs +++ b/crates/broker/src/worker.rs @@ -209,6 +209,25 @@ impl WorkerRegistry { self.workers.contains_key(name) } + /// True when a worker named `name` exists and either has no recorded + /// workspace or belongs to `workspace_id`. Gates sender impersonation on + /// Relaycast publish: a worker attached to workspace A must not be + /// impersonated when publishing into workspace B, which would register or + /// rotate that name's token in the wrong workspace. + pub(crate) fn has_worker_in_workspace( + &self, + name: &str, + workspace_id: &crate::ids::WorkspaceId, + ) -> bool { + match self.workers.get(name) { + Some(handle) => match &handle.workspace_id { + Some(worker_ws) => worker_ws == workspace_id, + None => true, + }, + None => false, + } + } + pub(crate) fn worker_pid(&self, name: &str) -> Option { self.workers.get(name).and_then(|h| h.child.id()) } @@ -1568,6 +1587,13 @@ mod tests { assert!(!reg.has_worker("nonexistent")); } + #[test] + fn has_worker_in_workspace_returns_false_for_unknown() { + let reg = make_registry(vec![]); + let workspace = crate::ids::WorkspaceId::new("ws_1".to_string()); + assert!(!reg.has_worker_in_workspace("nonexistent", &workspace)); + } + #[test] fn worker_log_path_rejects_path_traversal() { let reg = make_registry(vec![]); diff --git a/packages/harness-driver/src/client.ts b/packages/harness-driver/src/client.ts index d98b5dd1d..e9af477a2 100644 --- a/packages/harness-driver/src/client.ts +++ b/packages/harness-driver/src/client.ts @@ -374,15 +374,27 @@ export class HarnessDriverClient { onStep?.('Waiting for broker session handshake...'); let session: SessionInfo | undefined; - for (let attempt = 0; attempt < 10; attempt++) { + // The Relaycast handshake can take many seconds on a cold or slow network, + // during which the startup-only API answers 503. Poll for the full startup + // budget (`timeoutMs`) rather than a fixed attempt count so a slow-but- + // healthy handshake isn't misreported as a spawn failure. The `brokerExited` + // race still surfaces a dead broker immediately, so this only extends how + // long we wait on a broker that is alive and warming up. + const handshakeDeadline = Date.now() + timeoutMs; + for (let attempt = 0; ; attempt++) { try { session = await Promise.race([client.getSession(), brokerExited]); break; } catch (err) { - const message = err instanceof Error ? err.message : String(err); - const is503 = message.includes('503') || message.includes('Service Unavailable'); - if (!is503 || attempt >= 9) throw err; - onStep?.(`Broker still starting (handshake attempt ${attempt + 1}/10), retrying in 1s...`); + // The broker's startup-only API returns a structured 503 + // (`http_503`) while it warms up. Prefer the typed fields over the + // formatted message, which the broker is free to customize. + const is503 = + err instanceof HarnessDriverProtocolError + ? err.status === 503 || err.code === 'http_503' + : /503|Service Unavailable/.test(err instanceof Error ? err.message : String(err)); + if (!is503 || Date.now() >= handshakeDeadline) throw err; + onStep?.(`Broker still starting (handshake attempt ${attempt + 1}), retrying in 1s...`); await new Promise((resolve) => setTimeout(resolve, 1000)); } }