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

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