From 71d62e9e52570918476e8f9829fad83cbb793c87 Mon Sep 17 00:00:00 2001 From: Will Washburn Date: Wed, 1 Jul 2026 07:34:40 -0700 Subject: [PATCH 1/5] fix(harness-driver): poll broker handshake for full startupTimeoutMs MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit HarnessDriverClient.spawn() polled the broker's startup handshake for a fixed 10 attempts × 1s (~10s), ignoring the caller's startupTimeoutMs. When the Relaycast handshake took longer than ~10s on a cold or slow network, the loop exhausted its attempts and threw the broker's retryable "Broker is starting" 503 even though the broker was healthy and still warming up. This failed the release smoke tests, which skipped the internal @agent-relay/* publish jobs, which then made "Publish Main Package" time out waiting for deps that were never published. Poll until the full startup budget (default 45s) elapses instead. 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. Co-Authored-By: Claude Opus 4.8 (1M context) --- CHANGELOG.md | 1 + packages/harness-driver/src/client.ts | 13 ++++++++++--- 2 files changed, 11 insertions(+), 3 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 4e9fb4648..be0d2cdaf 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. A slow-but-healthy Relaycast handshake that keeps answering `503` while warming up is no longer misreported as a spawn failure (which was failing the release smoke tests and blocking publishes). - `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/packages/harness-driver/src/client.ts b/packages/harness-driver/src/client.ts index d98b5dd1d..8019df83d 100644 --- a/packages/harness-driver/src/client.ts +++ b/packages/harness-driver/src/client.ts @@ -374,15 +374,22 @@ 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...`); + 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)); } } From 597d04555f03c1a5d49eef40043ae52bf34560f6 Mon Sep 17 00:00:00 2001 From: Will Washburn Date: Wed, 1 Jul 2026 07:46:52 -0700 Subject: [PATCH 2/5] Update CHANGELOG.md Co-authored-by: gemini-code-assist[bot] <176961590+gemini-code-assist[bot]@users.noreply.github.com> --- CHANGELOG.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index be0d2cdaf..6946a884b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -26,7 +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. A slow-but-healthy Relaycast handshake that keeps answering `503` while warming up is no longer misreported as a spawn failure (which was failing the release smoke tests and blocking publishes). +- `HarnessDriverClient.spawn()` now polls the broker's startup handshake for the full `startupTimeoutMs` budget (default 45s) instead of a fixed ~10s, preventing slow-but-healthy handshakes from being misreported as spawn failures. - `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`. From 38abb3c188bc308f7da6670a3aefdc4057f33834 Mon Sep 17 00:00:00 2001 From: Will Washburn Date: Wed, 1 Jul 2026 08:27:19 -0700 Subject: [PATCH 3/5] =?UTF-8?q?fix(broker):=20address=20PR=20review=20?= =?UTF-8?q?=E2=80=94=20workspace-scoped=20impersonation,=20drive=20queued?= =?UTF-8?q?=20event,=20steer/thread,=20structured=20503?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - api.rs/worker.rs: gate sender impersonation on workspace membership, not just name. `has_worker_in_workspace` requires the custodial worker to belong to the workspace being published into, so a worker attached to workspace A can't be impersonated (registering/rotating its token) in workspace B. - fleet.rs: re-emit the `delivery_queued` broker event when a node delivery is held under manual_flush. The removed local send path emitted it and `attach --drive` counts it to show pending messages; node delivery is the only path now, so without this queued messages were invisible until a later flush. - ws.rs: a `Steer` send with a `thread_id` can't be honored (`AgentClient::reply` takes no injection mode). Downgrade to a normal reply as before, but log a warning and document it instead of dropping steer silently. - client.ts: match the startup-handshake 503 on the structured `HarnessDriverProtocolError` fields (`status`/`code`) with a message-regex fallback, so a custom broker 503 body still retries. - CHANGELOG: trim the spawn-timeout entry to impact-first per the style guide. Co-Authored-By: Claude Opus 4.8 (1M context) --- CHANGELOG.md | 2 +- crates/broker/src/relaycast/ws.rs | 16 +++++++++++++++- crates/broker/src/runtime/api.rs | 17 +++++++++++------ crates/broker/src/runtime/fleet.rs | 19 +++++++++++++++++++ crates/broker/src/worker.rs | 26 ++++++++++++++++++++++++++ packages/harness-driver/src/client.ts | 9 +++++++-- 6 files changed, 79 insertions(+), 10 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 6946a884b..dcb061e99 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -26,7 +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, preventing slow-but-healthy handshakes from being misreported as spawn failures. +- `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..7537d9f1d 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, 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 8019df83d..e9af477a2 100644 --- a/packages/harness-driver/src/client.ts +++ b/packages/harness-driver/src/client.ts @@ -386,8 +386,13 @@ export class HarnessDriverClient { 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'); + // 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)); From 11ac64b25d6d8ced4d93da61df0dcdd5f292c5bc Mon Sep 17 00:00:00 2001 From: Will Washburn Date: Wed, 1 Jul 2026 08:32:03 -0700 Subject: [PATCH 4/5] fix(broker): only thread-reply on real Relaycast message ids MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Since the send handler now posts via AgentClient::reply when thread_id is present, a caller echoing back a synthetic id — the broker-minted http_* event_id from /api/send, or a #channel / direct:* grouping key from /api/threads — would make Relaycast reject the reply and fail the whole send. Gate the reply path on is_relaycast_reply_target (reusing the existing synthetic-event-id classifier plus the channel/DM grouping-key prefixes); non-message-id thread_ids fall back to a plain post so the message is still delivered, just unthreaded. Co-Authored-By: Claude Opus 4.8 (1M context) --- crates/broker/src/runtime/api.rs | 20 ++++++++++++- crates/broker/src/runtime/delivery.rs | 41 +++++++++++++++++++++++++++ 2 files changed, 60 insertions(+), 1 deletion(-) diff --git a/crates/broker/src/runtime/api.rs b/crates/broker/src/runtime/api.rs index 7537d9f1d..039690123 100644 --- a/crates/broker/src/runtime/api.rs +++ b/crates/broker/src/runtime/api.rs @@ -760,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, @@ -768,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..e5e370981 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,30 @@ 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:?}"); + } + } +} From 7235a1ca7d9793ff22e2074728639a7403061f59 Mon Sep 17 00:00:00 2001 From: "github-actions[bot]" Date: Wed, 1 Jul 2026 15:45:45 +0000 Subject: [PATCH 5/5] style: auto-format Rust code with cargo fmt --- crates/broker/src/runtime/delivery.rs | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/crates/broker/src/runtime/delivery.rs b/crates/broker/src/runtime/delivery.rs index e5e370981..3dc4c4c44 100644 --- a/crates/broker/src/runtime/delivery.rs +++ b/crates/broker/src/runtime/delivery.rs @@ -940,7 +940,10 @@ mod reply_target_tests { "cont_load_1", "flush_1", ] { - assert!(!is_relaycast_reply_target(id), "expected non-target: {id:?}"); + assert!( + !is_relaycast_reply_target(id), + "expected non-target: {id:?}" + ); } } }