fix: preserve broker spawn custody and dispatch compiled MCP calls once

.github/workflows/fleet-e2e.yml

@@ -65,15 +65,15 @@ jobs:
 
       # The relaycast engine lives in a separate repo. This E2E drives the
       # node-provider model and heartbeats whose load may be unreported for
-      # unbounded providers. This companion PR candidate supplies isolated
-      # registration, acknowledged cleanup and node-owned spawn status reads.
+      # unbounded providers. The pinned merged engine supplies isolated
+      # registration, its contract acknowledgement, and node-owned spawn status reads.
       # It is a test pin, not a released or deployed product dependency.
       - name: Checkout relaycast engine (node-provider model)
         uses: actions/checkout@v4
         with:
           persist-credentials: false
           repository: AgentWorkforce/relaycast
-          ref: a5c7ac66f64a650274fbe2c60e3cd5d3c9b87dcd # relaycast#387: verified served-node spawn completion
+          ref: b60a39da04b9ab8b50d9064afe03fd73417b158c # relaycast#427/#430: node contract and bounded channel queries
           path: relaycast-engine
 
       - name: Setup Node.js

CHANGELOG.md

@@ -13,6 +13,16 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Fixed
 
+- Harness driver event observers disconnect promptly even when the broker does not complete the WebSocket Close handshake.
+
+- Compiled `agent-relay mcp` starts one stdio server, preventing duplicate tool execution and duplicate message sends.
+
+- Agent release waits for confirmed node deregistration and clears the completed registration reservation, allowing the same name to resume while preserving its retained identity; explicit release retries recover after automatic cleanup retries are exhausted.
+
+- Fleet action completion, delivery acknowledgements, and inventory retries stay bounded under control-queue pressure; broker shutdown retains unconfirmed action outcomes for reconciliation.
+
+- Fresh broker API spawns verify the server registration contract, create identities under the authenticated broker provider, and retain generation custody across timeouts and cleanup; unresolved names remain reserved after restart.
+
 - HTTP agent spawn rejects failed Relaycast node binding, cleans up the newly registered identity, and publishes declared metadata for successful spawns using either new or supplied tokens.
 
 - `agent-relay node up` retries the narrowly transient Relaycast `workspace_busy` admission response while keeping unrelated rate limits terminal and preserving bounded startup diagnostics.

README.md

@@ -7,6 +7,23 @@ Let Claude Code message Codex. Let your Hyperagent talk to your Hermes agent. Gi
 
 Relay gives all your agents shared channels, threads, DMs, reactions, files, search, and realtime events without building chat infrastructure.
 
+## Broker API spawn admission
+
+Fresh hosted identities created by `/api/spawn` require an authenticated
+`node.register` acknowledgement with `registration_contract: "relay:node-registration-v1"`.
+Deploy a compatible Relaycast engine and provider-aware adapter before upgrading
+the broker. Missing support fails before identity creation; there is no HTTP
+registration fallback. Caller-supplied agent credentials retain precedence.
+
+Timed-out or disconnected registrations retain name custody. Intent records under
+`<state-dir>/team/worker-logs/registration-custody/` contain identity hashes, never
+raw tokens, and unresolved names remain blocked after restart. Reconcile uncertain
+remote creation or failed cleanup before reusing a name. The broker allows at most
+256 retained fresh registrations at once and rejects additional admission instead
+of evicting unresolved records. Registration and channel checks do not block
+unrelated broker requests or shutdown. Caller cancellation before admission
+prevents launch; contradictory replies preserve quarantine even during cleanup.
+
 ## Quick Start
 
 Relay requires Node.js 22 or newer.

crates/broker/Cargo.toml

@@ -47,6 +47,7 @@ nix = { version = "0.30", features = ["signal", "process", "term", "fs"] }
 libc = "0.2"
 
 [dev-dependencies]
+tokio = { version = "1.44", features = ["test-util"] }
 httpmock = "0.7"
 tempfile = "3.19"
 tower = { version = "0.5", features = ["util"] }

crates/broker/src/fleet_responses.rs

@@ -0,0 +1,259 @@
+//! Bounded local custody for fleet action outcomes. A socket flush is the
+//! settlement boundary here; this is not a server receipt or crash guarantee.
+use crate::fleet_wire::{ActionResult, ActionResultError, ActionResultPayload, FLEET_WIRE_VERSION};
+use std::{
+    collections::{HashMap, VecDeque},
+    sync::Mutex,
+};
+use tokio::sync::Notify;
+
+const ORDINARY_LIMIT: usize = 256;
+const MAX_RESULT_BYTES: usize = 64 * 1024;
+const MAX_INVOCATION_ID_BYTES: usize = 4096;
+
+#[derive(Debug, PartialEq, Eq)]
+pub(crate) enum Admission {
+    Accepted,
+    Rejected,
+    Full,
+    Duplicate,
+    Invalid,
+}
+#[derive(Default)]
+struct State {
+    entries: HashMap<String, Entry>,
+    ready: VecDeque<String>,
+    rejection: Option<String>,
+}
+struct Entry {
+    result: Option<ActionResult>,
+}
+#[derive(Default)]
+pub(crate) struct FleetResponses {
+    state: Mutex<State>,
+    changed: Notify,
+    capacity_changed: Notify,
+    stop_changed: Notify,
+    connected: std::sync::atomic::AtomicBool,
+    shutdown: std::sync::atomic::AtomicBool,
+}
+impl FleetResponses {
+    pub(crate) fn reserve(&self, id: &str) -> Admission {
+        if id.is_empty() || id.len() > MAX_INVOCATION_ID_BYTES {
+            return Admission::Invalid;
+        }
+        let mut state = self.state.lock().unwrap();
+        if state.entries.contains_key(id) {
+            return Admission::Duplicate;
+        }
+        let ordinary = state.entries.len() - usize::from(state.rejection.is_some());
+        let admission = if ordinary < ORDINARY_LIMIT {
+            Admission::Accepted
+        } else if state.rejection.is_none() {
+            state.rejection = Some(id.into());
+            Admission::Rejected
+        } else {
+            return Admission::Full;
+        };
+        state.entries.insert(id.into(), Entry { result: None });
+        admission
+    }
+    pub(crate) fn can_admit(&self) -> bool {
+        let state = self.state.lock().unwrap();
+        state.entries.len() < ORDINARY_LIMIT || state.rejection.is_none()
+    }
+    /// A reservation is acquired before any validation or mutation. Completion
+    /// transfers that same slot to FIFO output; it never waits for a channel.
+    pub(crate) fn complete(&self, mut result: ActionResult) {
+        if serde_json::to_vec(&result).map_or(true, |bytes| bytes.len() > MAX_RESULT_BYTES) {
+            result.result = ActionResultPayload::Error(ActionResultError {
+                error: "action_result_too_large".into(),
+            });
+        }
+        let mut state = self.state.lock().unwrap();
+        let entry = state
+            .entries
+            .get_mut(&result.invocation_id)
+            .expect("fleet result must own an admission reservation");
+        if entry.result.is_some() {
+            return;
+        } // one terminal decision per admission
+        let id = result.invocation_id.clone();
+        entry.result = Some(result);
+        state.ready.push_back(id);
+        drop(state);
+        self.changed.notify_one();
+    }
+    pub(crate) fn front(&self) -> Option<ActionResult> {
+        let state = self.state.lock().unwrap();
+        state
+            .ready
+            .front()
+            .and_then(|id| state.entries[id].result.clone())
+    }
+    /// Only the sole node socket writer may settle the FIFO head, after its
+    /// successful flush. Failed/ambiguous writes keep the original outcome.
+    pub(crate) fn flushed(&self, id: &str) {
+        let mut state = self.state.lock().unwrap();
+        assert_eq!(state.ready.front().map(String::as_str), Some(id));
+        state.ready.pop_front();
+        state.entries.remove(id);
+        if state.rejection.as_deref() == Some(id) {
+            state.rejection = None;
+        }
+        drop(state);
+        self.capacity_changed.notify_one();
+    }
+    pub(crate) async fn changed(&self) {
+        self.changed.notified().await;
+    }
+    pub(crate) async fn capacity_changed(&self) {
+        self.capacity_changed.notified().await;
+    }
+    pub(crate) async fn stopped(&self) {
+        if !self.stopping() {
+            self.stop_changed.notified().await;
+        }
+    }
+    pub(crate) fn connected(&self) -> bool {
+        self.connected.load(std::sync::atomic::Ordering::Acquire)
+    }
+    pub(crate) fn set_connected(&self, connected: bool) {
+        self.connected
+            .store(connected, std::sync::atomic::Ordering::Release);
+    }
+    pub(crate) fn stop(&self) {
+        self.shutdown
+            .store(true, std::sync::atomic::Ordering::Release);
+        self.stop_changed.notify_one();
+    }
+    pub(crate) fn stopping(&self) -> bool {
+        self.shutdown.load(std::sync::atomic::Ordering::Acquire)
+    }
+    pub(crate) fn cancel_pending(&self) {
+        let ids: Vec<_> = self
+            .state
+            .lock()
+            .unwrap()
+            .entries
+            .iter()
+            .filter(|(_, entry)| entry.result.is_none())
+            .map(|(id, _)| id.clone())
+            .collect();
+        for id in ids {
+            self.complete(ActionResult {
+                v: FLEET_WIRE_VERSION,
+                id: None,
+                invocation_id: id,
+                result: ActionResultPayload::Error(ActionResultError {
+                    error: "broker_shutdown_before_completion".into(),
+                }),
+            });
+        }
+    }
+    pub(crate) fn checkpoint(&self, root: &std::path::Path) -> std::io::Result<()> {
+        use std::io::Write;
+        let results = self.unresolved();
+        if results.is_empty() {
+            return Ok(());
+        }
+        std::fs::create_dir_all(root)?;
+        // Unique immutable evidence: never overwrite an earlier unresolved run.
+        let path = root.join(format!(
+            "fleet-unconfirmed-results-{}.json",
+            uuid::Uuid::new_v4()
+        ));
+        let mut options = std::fs::OpenOptions::new();
+        options.write(true).create_new(true);
+        #[cfg(unix)]
+        {
+            use std::os::unix::fs::OpenOptionsExt;
+            options.mode(0o600);
+        }
+        let mut file = options.open(&path)?;
+        file.write_all(&serde_json::to_vec(&results)?)?;
+        file.sync_all()?;
+        tracing::warn!(count = results.len(), path = %path.display(), "retained unconfirmed fleet outcomes; server settlement/reconciliation is not proven");
+        Ok(())
+    }
+    pub(crate) fn unresolved(&self) -> Vec<ActionResult> {
+        let state = self.state.lock().unwrap();
+        state
+            .ready
+            .iter()
+            .filter_map(|id| state.entries[id].result.clone())
+            .collect()
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    fn result(id: &str) -> ActionResult {
+        ActionResult {
+            v: FLEET_WIRE_VERSION,
+            id: None,
+            invocation_id: id.into(),
+            result: ActionResultPayload::Error(ActionResultError {
+                error: "original".into(),
+            }),
+        }
+    }
+    #[tokio::test]
+    async fn bounded_reservations_preserve_terminal_order_and_rejection_progress() {
+        let queue = FleetResponses::default();
+        for i in 0..ORDINARY_LIMIT {
+            assert_eq!(queue.reserve(&format!("pending-{i}")), Admission::Accepted);
+        }
+        assert_eq!(queue.reserve("reject"), Admission::Rejected);
+        assert_eq!(queue.reserve("held"), Admission::Full);
+        queue.complete(result("reject"));
+        assert_eq!(
+            queue.front().unwrap().invocation_id,
+            "reject",
+            "ready rejection must not wait behind unfinished work"
+        );
+        assert_eq!(queue.reserve("pending-0"), Admission::Duplicate);
+        queue.flushed("reject");
+        assert_eq!(queue.reserve("next-reject"), Admission::Rejected);
+        queue.complete(result("pending-2"));
+        queue.complete(result("pending-0"));
+        assert_eq!(queue.front().unwrap().invocation_id, "pending-2");
+        assert_eq!(
+            queue.front().unwrap().invocation_id,
+            "pending-2",
+            "failed writer leaves the original head owned"
+        );
+        queue.flushed("pending-2");
+        assert_eq!(queue.front().unwrap().invocation_id, "pending-0");
+        assert_eq!(queue.reserve("new"), Admission::Accepted);
+        queue.cancel_pending();
+        assert_eq!(queue.unresolved().len(), ORDINARY_LIMIT + 1);
+        let directory = tempfile::tempdir().unwrap();
+        queue.checkpoint(directory.path()).unwrap();
+        assert_eq!(std::fs::read_dir(directory.path()).unwrap().count(), 1);
+        queue.stop();
+        tokio::time::timeout(std::time::Duration::from_millis(100), queue.stopped())
+            .await
+            .unwrap();
+    }
+    #[test]
+    fn result_bytes_are_bounded_without_losing_correlation() {
+        let queue = FleetResponses::default();
+        assert_eq!(
+            queue.reserve(&"i".repeat(MAX_INVOCATION_ID_BYTES + 1)),
+            Admission::Invalid
+        );
+        assert_eq!(queue.reserve("original-id"), Admission::Accepted);
+        let mut response = result("original-id");
+        response.result = ActionResultPayload::Error(ActionResultError {
+            error: "x".repeat(MAX_RESULT_BYTES + 1),
+        });
+        queue.complete(response);
+        queue.complete(result("original-id"));
+        assert_eq!(queue.unresolved().len(), 1);
+        assert!(serde_json::to_string(&queue.front().unwrap())
+            .unwrap()
+            .contains("action_result_too_large"));
+    }
+}

crates/broker/src/lib.rs

@@ -23,6 +23,7 @@ pub(crate) mod conversation_log;
 pub(crate) mod dedup;
 #[allow(dead_code)]
 pub(crate) mod events;
+pub(crate) mod fleet_responses;
 pub(crate) mod listen_api;
 #[allow(dead_code)]
 pub(crate) mod metrics;
@@ -36,6 +37,7 @@ pub(crate) mod redact;
 #[allow(dead_code)]
 pub(crate) mod relaycast;
 pub(crate) mod replay_buffer;
+pub(crate) mod spawn_registration;
 // 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