fix(broker): route all delivery through Relaycast + honor spawn startup timeout (#1221)

* fix(harness-driver): poll broker handshake for full startupTimeoutMs

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) <noreply@anthropic.com>

* Update CHANGELOG.md

Co-authored-by: gemini-code-assist[bot] <176961590+gemini-code-assist[bot]@users.noreply.github.com>

* fix(broker): address PR review — workspace-scoped impersonation, drive queued event, steer/thread, structured 503

- 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) <noreply@anthropic.com>

* fix(broker): only thread-reply on real Relaycast message ids

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) <noreply@anthropic.com>

* style: auto-format Rust code with cargo fmt

---------

Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Co-authored-by: gemini-code-assist[bot] <176961590+gemini-code-assist[bot]@users.noreply.github.com>
Co-authored-by: github-actions[bot] <github-actions[bot]@users.noreply.github.com>

Author: 4 people

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

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

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

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:?}"
+            );
+        }
+    }
+}

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) => {

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<u32> {
         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![]);

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));
       }
     }