From 77cffccd2c7f916ef8f1a69baa792dccf3ba3827 Mon Sep 17 00:00:00 2001
From: Miya <khaliqgant+miya@gmail.com>
Date: Mon, 14 Sep 2026 05:13:15 +0200
Subject: [PATCH 1/9] fix(broker): reconnect when inventory acknowledgements
 stop

Session-Id: 01a09dbd-b8ff-7072-927d-2f9f2c403790

Session-Id: 01a09dbd-b8ff-7072-927d-2f9f2c403790
---
 CHANGELOG.md                                  |   6 +-
 crates/broker/src/node_control.rs             | 635 ++++++++++++++++--
 .../1591-application-ack-reconnect/case.json  |  26 +
 .../1591-application-ack-reconnect/run.mjs    | 215 ++++++
 4 files changed, 830 insertions(+), 52 deletions(-)
 create mode 100644 tests/relayflows/cases/1591-application-ack-reconnect/case.json
 create mode 100644 tests/relayflows/cases/1591-application-ack-reconnect/run.mjs

diff --git a/CHANGELOG.md b/CHANGELOG.md
index a4ef197bae..6778493763 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -5,7 +5,11 @@ All notable changes to Agent Relay will be documented in this file.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
-## [Unreleased]
+## [Unreleased - Patch]
+
+### Fixed
+
+- Broker node connections recover when inventory acknowledgements stop even while WebSocket pongs continue, with bounded retries during outages.
 
 ## [12.1.0] - 2026-09-12
 
diff --git a/crates/broker/src/node_control.rs b/crates/broker/src/node_control.rs
index 5114831562..71e2297b5f 100644
--- a/crates/broker/src/node_control.rs
+++ b/crates/broker/src/node_control.rs
@@ -46,6 +46,13 @@ const INVENTORY_REFRESH_INTERVAL: Duration = Duration::from_secs(60);
 /// heartbeat intervals so three consecutive lost pings are tolerated before a
 /// reconnect.
 const READ_IDLE_TIMEOUT: Duration = Duration::from_secs(48);
+/// Bound every node-control WebSocket handshake so a half-open TCP/TLS path
+/// cannot stall the reconnect loop before it reaches the capped backoff.
+const CONNECT_TIMEOUT: Duration = Duration::from_secs(15);
+/// Bound each control-plane write. A full send buffer otherwise parks the
+/// entire select loop and prevents both transport and application deadlines
+/// from being observed.
+const WRITE_TIMEOUT: Duration = Duration::from_secs(10);
 const INITIAL_RECONNECT_DELAY: Duration = Duration::from_secs(1);
 const MAX_RECONNECT_DELAY: Duration = Duration::from_secs(30);
 const REGISTER_AGENT_PENDING_TTL: Duration = Duration::from_secs(300);
@@ -1716,13 +1723,19 @@ pub(crate) async fn run_node_control_client(
         if matches!(result, ControlRunResult::Shutdown) {
             return;
         }
-        if matches!(result, ControlRunResult::Disconnected) {
-            // A real connection was established and then dropped, so the current
-            // token authenticated successfully. Reset the 401 counter only here —
-            // NOT after a successful re-mint — so a tight loop where each mint
-            // succeeds but the engine keeps 401-ing `/v1/node/ws` still
-            // accumulates toward the cap instead of resetting on every iteration.
+        let application_ready = matches!(
+            result,
+            ControlRunResult::Disconnected {
+                application_ready: true
+            }
+        );
+        if application_ready {
+            // A correlated inventory.sync acknowledgement proves the engine's
+            // application processed this session. Reset outage state only after
+            // that proof — a transport handshake followed by a pre-ready drop
+            // must preserve both the 401 history and exponential backoff.
             consecutive_unauthorized = 0;
+            reconnect_delay = INITIAL_RECONNECT_DELAY;
         }
         if matches!(result, ControlRunResult::Unauthorized) {
             // The engine rejected our current node token. Re-mint a fresh one
@@ -1767,15 +1780,36 @@ pub(crate) async fn run_node_control_client(
                 );
             }
         }
+        let transition = match result {
+            ControlRunResult::ConnectFailed => "connect_failed",
+            ControlRunResult::Disconnected { .. } => "disconnected",
+            ControlRunResult::Unauthorized => "unauthorized",
+            ControlRunResult::Shutdown => unreachable!("shutdown returned above"),
+        };
+        tracing::warn!(
+            target = "relay_broker::fleet",
+            node_id = %config.node_id,
+            transition,
+            reconnect_delay_ms = reconnect_delay.as_millis(),
+            "node-control transition: unhealthy; reconnect scheduled"
+        );
         let _ = event_tx.send(FleetControlEvent::Disconnected).await;
         tokio::time::sleep(reconnect_delay).await;
-        reconnect_delay = (reconnect_delay * 2).min(MAX_RECONNECT_DELAY);
+        if !application_ready {
+            reconnect_delay = (reconnect_delay * 2).min(MAX_RECONNECT_DELAY);
+        }
     }
 }
 
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
 enum ControlRunResult {
-    Disconnected,
+    /// The WebSocket session never completed its transport handshake.
+    ConnectFailed,
+    Disconnected {
+        /// True only after a correlated inventory.sync acknowledgement proved
+        /// that the engine application processed this connection.
+        application_ready: bool,
+    },
     /// The `/v1/node/ws` handshake was rejected with HTTP 401/Unauthorized,
     /// i.e. the current node token is stale or scoped to a different
     /// workspace/engine and must be re-minted before retrying.
@@ -1794,6 +1828,66 @@ fn connect_error_is_unauthorized(error: &tokio_tungstenite::tungstenite::Error)
     )
 }
 
+/// Application-level liveness is proven by an engine reply to the correlated
+/// `inventory.sync` request the broker already sends periodically. WebSocket
+/// pong traffic is deliberately excluded: an intermediary or a socket task can
+/// keep answering pings even after the node-control application stops applying
+/// heartbeats and inventory.
+struct ApplicationLiveness {
+    deadline: Duration,
+    last_acknowledged: Instant,
+    pending_inventory_syncs: VecDeque<String>,
+    ready: bool,
+}
+
+impl ApplicationLiveness {
+    fn new(deadline: Duration) -> Self {
+        Self {
+            deadline,
+            last_acknowledged: Instant::now(),
+            pending_inventory_syncs: VecDeque::new(),
+            ready: false,
+        }
+    }
+
+    fn track_inventory_sync(&mut self, id: String) {
+        self.pending_inventory_syncs.push_back(id);
+    }
+
+    /// Returns `Some(true)` for the first successful application acknowledgement,
+    /// `Some(false)` for later ones, and `None` for unrelated replies.
+    fn acknowledge(&mut self, id: &str) -> Option<bool> {
+        let acknowledged_index = self
+            .pending_inventory_syncs
+            .iter()
+            .position(|pending_id| pending_id == id)?;
+        self.pending_inventory_syncs.drain(..=acknowledged_index);
+        let became_ready = !self.ready;
+        self.ready = true;
+        self.last_acknowledged = Instant::now();
+        // Relaycast serializes control work for a node, so this reply proves all
+        // older probes were processed. Preserve newer probes: their later error
+        // replies must still replace an unhealthy control session.
+        Some(became_ready)
+    }
+
+    fn reject(&mut self, id: &str) -> bool {
+        let Some(index) = self
+            .pending_inventory_syncs
+            .iter()
+            .position(|pending_id| pending_id == id)
+        else {
+            return false;
+        };
+        self.pending_inventory_syncs.remove(index);
+        true
+    }
+
+    fn idle(&self) -> Duration {
+        self.last_acknowledged.elapsed()
+    }
+}
+
 async fn run_connected_once(
     config: &FleetControlConfig,
     command_rx: &mut mpsc::Receiver<FleetControlCommand>,
@@ -1804,10 +1898,10 @@ async fn run_connected_once(
     inventory_refresh_interval: Duration,
 ) -> ControlRunResult {
     let Some(mut node_register) = registration.clone() else {
-        return ControlRunResult::Disconnected;
+        return ControlRunResult::ConnectFailed;
     };
     let Some(node_token) = config.node_token.as_deref() else {
-        return ControlRunResult::Disconnected;
+        return ControlRunResult::ConnectFailed;
     };
 
     // A fresh provider instance per connection: reconnecting with a new
@@ -1825,7 +1919,7 @@ async fn run_connected_once(
         Ok(request) => request,
         Err(error) => {
             tracing::warn!(target = "relay_broker::fleet", error = %error, "invalid fleet node ws url");
-            return ControlRunResult::Disconnected;
+            return ControlRunResult::ConnectFailed;
         }
     };
     let header = format!("Bearer {}", node_token.trim());
@@ -1835,7 +1929,7 @@ async fn run_connected_once(
         }
         Err(error) => {
             tracing::warn!(target = "relay_broker::fleet", error = %error, "invalid fleet node token header");
-            return ControlRunResult::Disconnected;
+            return ControlRunResult::ConnectFailed;
         }
     }
 
@@ -1854,21 +1948,49 @@ async fn run_connected_once(
         }
     }
 
-    let (ws, _) = match tokio_tungstenite::connect_async(request).await {
-        Ok(connected) => connected,
-        Err(error) => {
+    let (ws, _) = match tokio::time::timeout(
+        CONNECT_TIMEOUT,
+        tokio_tungstenite::connect_async(request),
+    )
+    .await
+    {
+        Ok(Ok(connected)) => connected,
+        Ok(Err(error)) => {
             tracing::warn!(target = "relay_broker::fleet", url = %config.ws_url, error = %error, "fleet node ws connect failed");
             if connect_error_is_unauthorized(&error) {
                 return ControlRunResult::Unauthorized;
             }
-            return ControlRunResult::Disconnected;
+            return ControlRunResult::ConnectFailed;
+        }
+        Err(_) => {
+            tracing::warn!(
+                target = "relay_broker::fleet",
+                url = %config.ws_url,
+                timeout_secs = CONNECT_TIMEOUT.as_secs(),
+                "fleet node ws connect attempt timed out"
+            );
+            return ControlRunResult::ConnectFailed;
         }
     };
+    tracing::info!(
+        target = "relay_broker::fleet",
+        node_id = %config.node_id,
+        "node-control transition: transport connected; awaiting application acknowledgement"
+    );
     let _ = event_tx.send(FleetControlEvent::Connected).await;
     let (mut sink, mut stream) = ws.split();
     let mut pending_agent_registrations: HashMap<String, PendingAgentRegistration> = HashMap::new();
     let mut pending_deregistrations: HashMap<String, oneshot::Sender<Result<(), String>>> =
         HashMap::new();
+    let read_idle_timeout = config.read_idle_timeout.unwrap_or(READ_IDLE_TIMEOUT);
+    // inventory.sync is the existing request/reply application probe. Allow two
+    // complete refresh periods before declaring it unacknowledged, while never
+    // making the application deadline tighter than the transport deadline.
+    let application_liveness_timeout = inventory_refresh_interval
+        .checked_mul(2)
+        .unwrap_or(Duration::MAX)
+        .max(read_idle_timeout);
+    let mut application_liveness = ApplicationLiveness::new(application_liveness_timeout);
 
     if send_wire(
         &mut sink,
@@ -1877,10 +1999,21 @@ async fn run_connected_once(
     .await
     .is_err()
     {
-        return ControlRunResult::Disconnected;
+        return ControlRunResult::Disconnected {
+            application_ready: false,
+        };
     }
-    if !send_inventory_sync(&mut sink, inventory, &mut pending_agent_registrations).await {
-        return ControlRunResult::Disconnected;
+    if !send_inventory_sync(
+        &mut sink,
+        inventory,
+        &mut pending_agent_registrations,
+        &mut application_liveness,
+    )
+    .await
+    {
+        return ControlRunResult::Disconnected {
+            application_ready: false,
+        };
     }
     if send_wire(
         &mut sink,
@@ -1889,13 +2022,14 @@ async fn run_connected_once(
     .await
     .is_err()
     {
-        return ControlRunResult::Disconnected;
+        return ControlRunResult::Disconnected {
+            application_ready: false,
+        };
     }
 
     // The idle check runs on the heartbeat tick, so the tick must be shorter
     // than the window it polices; an overridden (test) window keeps that ratio.
-    let read_idle_timeout_value = config.read_idle_timeout.unwrap_or(READ_IDLE_TIMEOUT);
-    let mut heartbeat = tokio::time::interval(HEARTBEAT_INTERVAL.min(read_idle_timeout_value / 4));
+    let mut heartbeat = tokio::time::interval(HEARTBEAT_INTERVAL.min(read_idle_timeout / 4));
     heartbeat.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
     let mut inventory_refresh = tokio::time::interval(inventory_refresh_interval);
     inventory_refresh.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
@@ -1903,7 +2037,6 @@ async fn run_connected_once(
     // already renewed the lease, so schedule the first refresh one full period
     // from now instead of duplicating it on connection setup.
     inventory_refresh.tick().await;
-    let read_idle_timeout = read_idle_timeout_value;
     let mut last_inbound = Instant::now();
 
     loop {
@@ -1918,7 +2051,7 @@ async fn run_connected_once(
                         node_register = next.clone();
                         *registration = Some(next.clone());
                         if send_wire(&mut sink, &BrokerToRelaycast::NodeRegister(next)).await.is_err() {
-                            return ControlRunResult::Disconnected;
+                            return ControlRunResult::Disconnected { application_ready: application_liveness.ready };
                         }
                     }
                     Some(FleetControlCommand::UpdateInventory(next)) => {
@@ -1927,10 +2060,11 @@ async fn run_connected_once(
                             &mut sink,
                             inventory,
                             &mut pending_agent_registrations,
+                            &mut application_liveness,
                         )
                         .await
                         {
-                            return ControlRunResult::Disconnected;
+                            return ControlRunResult::Disconnected { application_ready: application_liveness.ready };
                         }
                     }
                     Some(FleetControlCommand::UpdateLoad(next)) => {
@@ -1938,12 +2072,12 @@ async fn run_connected_once(
                     }
                     Some(FleetControlCommand::HeartbeatNow) => {
                         if send_wire(&mut sink, &BrokerToRelaycast::NodeHeartbeat(load.heartbeat(&node_register))).await.is_err() {
-                            return ControlRunResult::Disconnected;
+                            return ControlRunResult::Disconnected { application_ready: application_liveness.ready };
                         }
                     }
                     Some(FleetControlCommand::Send(message)) => {
                         if send_wire(&mut sink, &message).await.is_err() {
-                            return ControlRunResult::Disconnected;
+                            return ControlRunResult::Disconnected { application_ready: application_liveness.ready };
                         }
                     }
                     Some(FleetControlCommand::DeregisterAgent { mut request, reply }) => {
@@ -1952,7 +2086,7 @@ async fn run_connected_once(
                         pending_deregistrations.retain(|_, pending| !pending.is_closed());
                         pending_deregistrations.insert(request_id, reply);
                         if send_wire(&mut sink, &BrokerToRelaycast::AgentDeregister(request)).await.is_err() {
-                            return ControlRunResult::Disconnected;
+                            return ControlRunResult::Disconnected { application_ready: application_liveness.ready };
                         }
                     }
                     Some(FleetControlCommand::RegisterAgent { mut request, reply }) => {
@@ -1971,7 +2105,7 @@ async fn run_connected_once(
                         );
                         if send_wire(&mut sink, &BrokerToRelaycast::AgentRegister(request)).await.is_err() {
                             drain_agent_registrations(&mut pending_agent_registrations, "node_control_disconnected");
-                            return ControlRunResult::Disconnected;
+                            return ControlRunResult::Disconnected { application_ready: application_liveness.ready };
                         }
                     }
                     Some(FleetControlCommand::Shutdown) | None => {
@@ -1999,17 +2133,31 @@ async fn run_connected_once(
                         "no inbound node-control frame within the read-idle window; reconnecting"
                     );
                     drain_agent_registrations(&mut pending_agent_registrations, "node_control_disconnected");
-                    return ControlRunResult::Disconnected;
+                    return ControlRunResult::Disconnected { application_ready: application_liveness.ready };
+                }
+                let application_idle = application_liveness.idle();
+                if application_idle >= application_liveness.deadline {
+                    tracing::warn!(
+                        target = "relay_broker::fleet",
+                        node_id = %config.node_id,
+                        application_idle_ms = application_idle.as_millis(),
+                        application_deadline_ms = application_liveness.deadline.as_millis(),
+                        transport_idle_ms = idle.as_millis(),
+                        pending_inventory_syncs = application_liveness.pending_inventory_syncs.len(),
+                        "node-control application acknowledgement deadline exceeded while transport remained active; reconnecting"
+                    );
+                    drain_agent_registrations(&mut pending_agent_registrations, "node_control_disconnected");
+                    return ControlRunResult::Disconnected { application_ready: application_liveness.ready };
                 }
                 if send_wire(&mut sink, &BrokerToRelaycast::NodeHeartbeat(load.heartbeat(&node_register))).await.is_err() {
                     drain_agent_registrations(&mut pending_agent_registrations, "node_control_disconnected");
-                    return ControlRunResult::Disconnected;
+                    return ControlRunResult::Disconnected { application_ready: application_liveness.ready };
                 }
                 // Guarantees the peer owes us a frame every interval, so an idle
                 // engine is distinguishable from a dead connection.
-                if sink.send(Message::Ping(Vec::new())).await.is_err() {
+                if send_ws_frame(&mut sink, Message::Ping(Vec::new())).await.is_err() {
                     drain_agent_registrations(&mut pending_agent_registrations, "node_control_disconnected");
-                    return ControlRunResult::Disconnected;
+                    return ControlRunResult::Disconnected { application_ready: application_liveness.ready };
                 }
             }
             _ = inventory_refresh.tick() => {
@@ -2017,32 +2165,43 @@ async fn run_connected_once(
                     &mut sink,
                     inventory,
                     &mut pending_agent_registrations,
+                    &mut application_liveness,
                 )
                 .await
                 {
-                    return ControlRunResult::Disconnected;
+                    return ControlRunResult::Disconnected { application_ready: application_liveness.ready };
                 }
             }
             message = stream.next() => {
                 let Some(message) = message else {
                     drain_agent_registrations(&mut pending_agent_registrations, "node_control_disconnected");
-                    return ControlRunResult::Disconnected;
+                    return ControlRunResult::Disconnected { application_ready: application_liveness.ready };
                 };
                 let message = match message {
                     Ok(message) => message,
                     Err(error) => {
                         tracing::warn!(target = "relay_broker::fleet", error = %error, "fleet node ws read failed");
                         drain_agent_registrations(&mut pending_agent_registrations, "node_control_disconnected");
-                        return ControlRunResult::Disconnected;
+                        return ControlRunResult::Disconnected { application_ready: application_liveness.ready };
                     }
                 };
                 // Any frame proves the peer is still there — including the pong
                 // answering our ping, which is the only traffic a healthy but
                 // idle engine is guaranteed to send.
                 last_inbound = Instant::now();
-                if !handle_server_message(message, event_tx, &mut pending_agent_registrations, &mut pending_deregistrations, &mut sink).await {
+                if !handle_server_message(
+                    message,
+                    event_tx,
+                    &mut pending_agent_registrations,
+                    &mut pending_deregistrations,
+                    &mut application_liveness,
+                    &config.node_id,
+                    &mut sink,
+                )
+                .await
+                {
                     drain_agent_registrations(&mut pending_agent_registrations, "node_control_disconnected");
-                    return ControlRunResult::Disconnected;
+                    return ControlRunResult::Disconnected { application_ready: application_liveness.ready };
                 }
             }
         }
@@ -2053,16 +2212,18 @@ async fn send_inventory_sync<S>(
     sink: &mut S,
     inventory: &[InventoryAgent],
     pending_agent_registrations: &mut HashMap<String, PendingAgentRegistration>,
+    application_liveness: &mut ApplicationLiveness,
 ) -> bool
 where
     S: Sink<Message> + Unpin,
     S::Error: std::error::Error + Send + Sync + 'static,
 {
+    let request_id = format!("inventory_sync_{}", Uuid::new_v4().simple());
     if send_wire(
         sink,
         &BrokerToRelaycast::InventorySync(InventorySync {
             v: FLEET_WIRE_VERSION,
-            id: None,
+            id: Some(request_id.clone()),
             agents: inventory.to_vec(),
         }),
     )
@@ -2072,6 +2233,7 @@ where
         drain_agent_registrations(pending_agent_registrations, "node_control_disconnected");
         return false;
     }
+    application_liveness.track_inventory_sync(request_id);
 
     true
 }
@@ -2081,6 +2243,8 @@ async fn handle_server_message<S>(
     event_tx: &mpsc::Sender<FleetControlEvent>,
     pending_agent_registrations: &mut HashMap<String, PendingAgentRegistration>,
     pending_deregistrations: &mut HashMap<String, oneshot::Sender<Result<(), String>>>,
+    application_liveness: &mut ApplicationLiveness,
+    node_id: &str,
     sink: &mut S,
 ) -> bool
 where
@@ -2100,7 +2264,24 @@ where
                     return true;
                 }
 
-                complete_agent_registration(reply, pending_agent_registrations, sink).await
+                if !reply.ok && application_liveness.reject(&reply.id) {
+                    return false;
+                }
+                match application_liveness.acknowledge(&reply.id) {
+                    Some(became_ready) => {
+                        if became_ready {
+                            tracing::info!(
+                            target = "relay_broker::fleet",
+                            node_id,
+                            "node-control transition: application acknowledgement received; control link ready"
+                        );
+                        }
+                        true
+                    }
+                    None => {
+                        complete_agent_registration(reply, pending_agent_registrations, sink).await
+                    }
+                }
             }
             Ok(RelaycastToBroker::Error(error)) => {
                 if let Some(pending) = pending_deregistrations.remove(&error.id) {
@@ -2122,12 +2303,16 @@ where
                 if error.code == "invalid_message" {
                     fail_unsupported_channel_isolation(&error.message, pending_agent_registrations);
                 }
+                let rejected_liveness_probe = application_liveness.reject(&error.id);
                 fail_agent_registration(
                     &error.id,
                     format!("{}: {}", error.code, error.message),
                     pending_agent_registrations,
                 );
-                true
+                // A reply proves the application is responsive, but rejecting
+                // the authoritative inventory probe means the control session
+                // is not healthy enough to advertise; replace it immediately.
+                !rejected_liveness_probe
             }
             Ok(other) => event_tx
                 .send(FleetControlEvent::Message(other))
@@ -2154,11 +2339,10 @@ where
     S::Error: std::error::Error + Send + Sync + 'static,
 {
     let request_id = reply.id.clone();
-    // The engine replies to every node-control request (`node.register`,
-    // `inventory.sync`, ...) with a `reply` frame, but only `agent.register`
-    // replies correspond to a pending registration. Those non-agent replies
-    // carry a fresh engine-minted snowflake id (the broker sends those frames
-    // without an `id`), so they never match `request_id`. To stay robust we:
+    // The engine replies to node-control requests such as `node.register`, but
+    // only `agent.register` replies correspond to a pending registration.
+    // Correlated `inventory.sync` replies have already been consumed by the
+    // application-liveness tracker above. To stay robust we:
     //   1. match on the echoed request id (the happy path), then
     //   2. fall back to matching the validated reply `data.name` against a
     //      pending entry (covers an engine that drops/regenerates the id), and
@@ -2182,7 +2366,7 @@ where
         tracing::debug!(
             target = "relay_broker::fleet",
             id = %request_id,
-            "node-control reply did not match a pending agent.register (likely a node.register/inventory.sync reply)"
+            "node-control reply did not match a pending agent.register (likely a node.register reply)"
         );
         return true;
     };
@@ -2314,8 +2498,28 @@ where
     S::Error: std::error::Error + Send + Sync + 'static,
 {
     let text = serde_json::to_string(message)?;
-    sink.send(Message::Text(text)).await?;
-    Ok(())
+    send_ws_frame(sink, Message::Text(text)).await
+}
+
+async fn send_ws_frame<S>(sink: &mut S, message: Message) -> Result<()>
+where
+    S: Sink<Message> + Unpin,
+    S::Error: std::error::Error + Send + Sync + 'static,
+{
+    match tokio::time::timeout(WRITE_TIMEOUT, sink.send(message)).await {
+        Ok(result) => {
+            result?;
+            Ok(())
+        }
+        Err(_) => {
+            tracing::warn!(
+                target = "relay_broker::fleet",
+                timeout_secs = WRITE_TIMEOUT.as_secs(),
+                "fleet node websocket write timed out; treating control link as unhealthy"
+            );
+            Err(anyhow::anyhow!("fleet node websocket write timed out"))
+        }
+    }
 }
 
 pub(crate) fn delivery_ack(agent: impl Into<String>, up_to_seq: u64) -> BrokerToRelaycast {
@@ -3347,6 +3551,8 @@ mod tests {
                     &events,
                     &mut pending,
                     &mut deregistrations,
+                    &mut ApplicationLiveness::new(Duration::from_secs(1)),
+                    "node-test",
                     &mut sink
                 )
                 .await
@@ -3368,6 +3574,57 @@ mod tests {
         assert_eq!(pending.len(), 1);
     }
 
+    #[test]
+    fn application_readiness_requires_successful_correlated_inventory_reply() {
+        let mut liveness = ApplicationLiveness::new(Duration::from_secs(1));
+        liveness.track_inventory_sync("inventory-rejected".to_string());
+        assert!(liveness.reject("inventory-rejected"));
+        assert!(
+            !liveness.ready,
+            "an error reply must not make the link ready"
+        );
+
+        liveness.track_inventory_sync("inventory-acknowledged".to_string());
+        assert_eq!(liveness.acknowledge("unrelated"), None);
+        assert_eq!(liveness.acknowledge("inventory-acknowledged"), Some(true));
+        assert!(liveness.ready);
+    }
+
+    #[tokio::test]
+    async fn rejected_inventory_reply_does_not_make_application_ready() {
+        let mut liveness = ApplicationLiveness::new(Duration::from_secs(1));
+        liveness.track_inventory_sync("inventory-rejected".to_string());
+        let (events, _receiver) = mpsc::channel(1);
+        let healthy = handle_server_message(
+            Message::Text(json!({"v": 1, "type": "reply", "id": "inventory-rejected", "ok": false, "data": {}}).to_string()),
+            &events,
+            &mut HashMap::new(),
+            &mut HashMap::new(),
+            &mut liveness,
+            "node-test",
+            &mut futures_util::sink::drain(),
+        ).await;
+        assert!(!healthy, "a rejected inventory must replace the connection");
+        assert!(
+            !liveness.ready,
+            "a rejection is not an application acknowledgement"
+        );
+    }
+
+    #[test]
+    fn acknowledged_probe_preserves_newer_probe_for_rejection() {
+        let mut liveness = ApplicationLiveness::new(Duration::from_secs(1));
+        liveness.track_inventory_sync("inventory-a".to_string());
+        liveness.track_inventory_sync("inventory-b".to_string());
+
+        assert_eq!(liveness.acknowledge("inventory-a"), Some(true));
+        assert_eq!(
+            liveness.pending_inventory_syncs,
+            VecDeque::from(["inventory-b".to_string()])
+        );
+        assert!(liveness.reject("inventory-b"));
+    }
+
     #[test]
     fn expire_agent_registrations_bounds_pending_map() {
         let created_at = Instant::now();
@@ -4105,7 +4362,12 @@ mod tests {
         .await
         .expect("mock node-control session should finish");
 
-        assert_eq!(result, ControlRunResult::Disconnected);
+        assert_eq!(
+            result,
+            ControlRunResult::Disconnected {
+                application_ready: false
+            }
+        );
         server.await.unwrap();
     }
 
@@ -4183,10 +4445,281 @@ mod tests {
         .await
         .expect("mock node-control session should finish");
 
-        assert_eq!(result, ControlRunResult::Disconnected);
+        assert_eq!(
+            result,
+            ControlRunResult::Disconnected {
+                application_ready: false
+            }
+        );
+        server.await.unwrap();
+    }
+
+    /// Regression for relay#1591: transport traffic must not mask an application
+    /// control-plane failure. The peer keeps the TCP/WebSocket connection open,
+    /// continuously polls it (so tungstenite answers every WebSocket ping with a
+    /// pong), and drains every node frame, acknowledges the initial inventory, then stops acknowledging later
+    /// application requests. The node-control session must still declare the link dead.
+    #[tokio::test]
+    async fn node_control_disconnects_when_application_acks_stop_but_socket_stays_live() {
+        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+        let ws_url = format!("ws://{}/v1/node/ws", listener.local_addr().unwrap());
+        let (_command_tx, mut command_rx) = mpsc::channel(4);
+        let (event_tx, _event_rx) = mpsc::channel(4);
+        let mut registration = Some(build_node_register(
+            &test_manifest(),
+            "node-test",
+            "host-test",
+            "broker/test",
+            None,
+        ));
+        let mut inventory = Vec::new();
+        let mut load = FleetLoadSnapshot {
+            active_agents: 3,
+            max_agents: 4,
+            handlers_live: true,
+            active_agent_names: vec![
+                "agent-a".to_string(),
+                "agent-b".to_string(),
+                "agent-c".to_string(),
+            ],
+        };
+
+        let server = tokio::spawn(async move {
+            let (stream, _) = listener.accept().await.unwrap();
+            let mut ws = accept_async(stream).await.unwrap();
+            let mut saw_inventory = false;
+            let mut saw_heartbeat = false;
+
+            while let Some(frame) = ws.next().await {
+                let Ok(frame) = frame else { break };
+                match frame {
+                    Message::Text(text) => match serde_json::from_str::<BrokerToRelaycast>(&text)
+                        .expect("valid node-control frame")
+                    {
+                        BrokerToRelaycast::InventorySync(sync) => {
+                            if !saw_inventory {
+                                ws.send(Message::Text(
+                                    json!({
+                                        "v": 1, "type": "reply", "id": sync.id.unwrap(),
+                                        "ok": true, "data": {"reconciled": 0}
+                                    })
+                                    .to_string(),
+                                ))
+                                .await
+                                .unwrap();
+                            }
+                            saw_inventory = true;
+                        }
+                        BrokerToRelaycast::NodeHeartbeat(heartbeat) => {
+                            saw_heartbeat = true;
+                            assert_eq!(heartbeat.active_agents, 3);
+                        }
+                        _ => {}
+                    },
+                    Message::Close(_) => break,
+                    _ => {}
+                }
+            }
+
+            assert!(saw_inventory, "client must send an application request");
+            assert!(saw_heartbeat, "client process must keep heartbeating");
+        });
+
+        let result = tokio::time::timeout(
+            Duration::from_secs(2),
+            run_connected_once(
+                &FleetControlConfig {
+                    ws_url,
+                    node_token: Some("nt_test".to_string()),
+                    node_id: "node-test".to_string(),
+                    node_name: "host-test".to_string(),
+                    broker_version: "broker/test".to_string(),
+                    token_minter: None,
+                    session_token: None,
+                    read_idle_timeout: Some(Duration::from_millis(400)),
+                },
+                &mut command_rx,
+                &event_tx,
+                &mut registration,
+                &mut inventory,
+                &mut load,
+                Duration::from_millis(100),
+            ),
+        )
+        .await
+        .expect("application-level liveness deadline did not fire");
+
+        assert_eq!(
+            result,
+            ControlRunResult::Disconnected {
+                application_ready: true
+            }
+        );
+        server.await.unwrap();
+    }
+
+    /// Must-not-fire control arm for the application deadline above. Under the
+    /// same 400ms deadline and 100ms probe cadence, correlated inventory replies
+    /// keep the session healthy for several complete deadline windows.
+    #[tokio::test]
+    async fn node_control_stays_connected_while_application_acks_continue() {
+        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+        let ws_url = format!("ws://{}/v1/node/ws", listener.local_addr().unwrap());
+        let (command_tx, mut command_rx) = mpsc::channel(4);
+        let (event_tx, _event_rx) = mpsc::channel(4);
+        let mut registration = Some(build_node_register(
+            &test_manifest(),
+            "node-test",
+            "host-test",
+            "broker/test",
+            None,
+        ));
+        let mut inventory = Vec::new();
+        let mut load = FleetLoadSnapshot {
+            active_agents: 3,
+            max_agents: 4,
+            handlers_live: true,
+            active_agent_names: vec![
+                "agent-a".to_string(),
+                "agent-b".to_string(),
+                "agent-c".to_string(),
+            ],
+        };
+
+        let server = tokio::spawn(async move {
+            let (stream, _) = listener.accept().await.unwrap();
+            let mut ws = accept_async(stream).await.unwrap();
+            while let Some(frame) = ws.next().await {
+                let Ok(frame) = frame else { break };
+                let Message::Text(text) = frame else { continue };
+                let BrokerToRelaycast::InventorySync(sync) =
+                    serde_json::from_str::<BrokerToRelaycast>(&text)
+                        .expect("valid node-control frame")
+                else {
+                    continue;
+                };
+                let id = sync.id.expect("inventory liveness probe id");
+                if ws
+                    .send(Message::Text(
+                        serde_json::to_string(&RelaycastToBroker::Reply(
+                            crate::fleet_wire::Reply {
+                                v: FLEET_WIRE_VERSION,
+                                id,
+                                ok: true,
+                                data: json!({ "reconciled": sync.agents.len() }),
+                            },
+                        ))
+                        .unwrap(),
+                    ))
+                    .await
+                    .is_err()
+                {
+                    break;
+                }
+            }
+        });
+
+        let shutdown = tokio::spawn(async move {
+            tokio::time::sleep(Duration::from_millis(1200)).await;
+            command_tx
+                .send(FleetControlCommand::Shutdown)
+                .await
+                .unwrap();
+        });
+        let result = tokio::time::timeout(
+            Duration::from_secs(2),
+            run_connected_once(
+                &FleetControlConfig {
+                    ws_url,
+                    node_token: Some("nt_test".to_string()),
+                    node_id: "node-test".to_string(),
+                    node_name: "host-test".to_string(),
+                    broker_version: "broker/test".to_string(),
+                    token_minter: None,
+                    session_token: None,
+                    read_idle_timeout: Some(Duration::from_millis(400)),
+                },
+                &mut command_rx,
+                &event_tx,
+                &mut registration,
+                &mut inventory,
+                &mut load,
+                Duration::from_millis(100),
+            ),
+        )
+        .await
+        .expect("acknowledged application link should remain connected");
+
+        assert_eq!(result, ControlRunResult::Shutdown);
+        shutdown.await.unwrap();
         server.await.unwrap();
     }
 
+    /// Transport handshakes do not make a control session healthy. If the peer
+    /// repeatedly drops each socket before acknowledging inventory.sync, the
+    /// reconnect delay must continue growing instead of resetting to one second.
+    #[tokio::test]
+    async fn pre_ready_disconnects_preserve_exponential_reconnect_backoff() {
+        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+        let ws_url = format!("ws://{}/v1/node/ws", listener.local_addr().unwrap());
+        let (command_tx, command_rx) = mpsc::channel(8);
+        let (event_tx, _event_rx) = mpsc::channel(8);
+
+        let client = tokio::spawn(run_node_control_client(
+            FleetControlConfig {
+                ws_url,
+                node_token: Some("nt_test".to_string()),
+                node_id: "node-test".to_string(),
+                node_name: "host-test".to_string(),
+                broker_version: "broker/test".to_string(),
+                token_minter: None,
+                session_token: None,
+                read_idle_timeout: Some(Duration::from_millis(400)),
+            },
+            command_rx,
+            event_tx,
+        ));
+        command_tx
+            .send(FleetControlCommand::RegisterNode {
+                manifest: test_manifest(),
+                resume_cursor: None,
+            })
+            .await
+            .unwrap();
+
+        let started = Instant::now();
+        let mut accepted_at = Vec::new();
+        let mut shutdown_ws = None;
+        for attempt in 0..3 {
+            let (stream, _) = tokio::time::timeout(Duration::from_secs(5), listener.accept())
+                .await
+                .expect("client did not make the next reconnect attempt")
+                .unwrap();
+            let ws = accept_async(stream).await.unwrap();
+            accepted_at.push(started.elapsed());
+            if attempt < 2 {
+                drop(ws);
+            } else {
+                shutdown_ws = Some(ws);
+                command_tx
+                    .send(FleetControlCommand::Shutdown)
+                    .await
+                    .unwrap();
+            }
+        }
+
+        client.await.unwrap();
+        drop(shutdown_ws);
+        assert!(
+            accepted_at[1].saturating_sub(accepted_at[0]) >= Duration::from_millis(900),
+            "first pre-ready failure must retain the one-second backoff: {accepted_at:?}"
+        );
+        assert!(
+            accepted_at[2].saturating_sub(accepted_at[1]) >= Duration::from_millis(1800),
+            "second pre-ready failure must grow to the two-second backoff: {accepted_at:?}"
+        );
+    }
+
     /// A blackholed `/v1/node/ws` — the socket still accepts writes, but the
     /// engine never sends another frame — must be detected and reconnected.
     ///
diff --git a/tests/relayflows/cases/1591-application-ack-reconnect/case.json b/tests/relayflows/cases/1591-application-ack-reconnect/case.json
new file mode 100644
index 0000000000..87b5f6807d
--- /dev/null
+++ b/tests/relayflows/cases/1591-application-ack-reconnect/case.json
@@ -0,0 +1,26 @@
+{
+  "version": 1,
+  "id": "1591-application-ack-reconnect",
+  "kind": "bugfix",
+  "title": "Reconnect when application acknowledgements stop",
+  "runner": {
+    "command": [
+      "node",
+      "tests/relayflows/cases/1591-application-ack-reconnect/run.mjs"
+    ]
+  },
+  "timeoutSeconds": 240,
+  "expected": {
+    "base": {
+      "outcome": "bug",
+      "signature": "application_ack_stall_not_detected"
+    },
+    "head": {
+      "outcome": "fixed",
+      "signature": "application_ack_stall_reconnects"
+    }
+  },
+  "requirements": [
+    "broker-linux-x64"
+  ]
+}
diff --git a/tests/relayflows/cases/1591-application-ack-reconnect/run.mjs b/tests/relayflows/cases/1591-application-ack-reconnect/run.mjs
new file mode 100644
index 0000000000..2c2d01661a
--- /dev/null
+++ b/tests/relayflows/cases/1591-application-ack-reconnect/run.mjs
@@ -0,0 +1,215 @@
+import { execFileSync, spawn } from 'node:child_process';
+import { createHash } from 'node:crypto';
+import http from 'node:http';
+import { access, mkdir, mkdtemp, readFile, rm, writeFile } from 'node:fs/promises';
+import { constants } from 'node:fs';
+import { tmpdir } from 'node:os';
+import path from 'node:path';
+import { fileURLToPath } from 'node:url';
+
+const CASE_ID = '1591-application-ack-reconnect';
+const required = (name) => {
+  const value = process.env[name];
+  if (!value) throw new Error(`Missing ${name}`);
+  return value;
+};
+const arm = required('RELAY_PR_PROOF_ARM');
+if (!['base', 'head'].includes(arm)) throw new Error('Invalid proof arm');
+const targetDir = path.resolve(required('RELAY_PR_PROOF_TARGET_DIR'));
+const harnessDir = path.resolve(required('RELAY_PR_PROOF_HARNESS_DIR'));
+const expectedSha = required(arm === 'base' ? 'RELAY_PR_PROOF_BASE_SHA' : 'RELAY_PR_PROOF_HEAD_SHA');
+const headSha = required('RELAY_PR_PROOF_HEAD_SHA');
+const shaAt = (directory) => execFileSync('git', ['-C', directory, 'rev-parse', 'HEAD'], { encoding: 'utf8' }).trim();
+if (shaAt(targetDir) !== expectedSha || shaAt(harnessDir) !== headSha) throw new Error('Exact target/harness SHA mismatch');
+if (fileURLToPath(import.meta.url) !== path.join(harnessDir, 'tests/relayflows/cases', CASE_ID, 'run.mjs')) throw new Error('Runner must come from exact-head harness');
+const binary = path.resolve(required('RELAY_PR_PROOF_BROKER_BINARY'));
+await access(binary, constants.R_OK | constants.X_OK);
+const binarySha256 = createHash('sha256').update(await readFile(binary)).digest('hex');
+const resultPath = path.resolve(required('RELAY_PR_PROOF_RESULT_PATH'));
+const scratch = await mkdtemp(path.join(tmpdir(), 'relayflow-inventory-ack-'));
+const stateDir = path.join(scratch, 'state');
+await mkdir(stateDir);
+const sockets = new Set();
+const connections = [];
+const started = performance.now();
+let broker;
+let stderr = '';
+let fixtureError;
+const sendJson = (response, data) => {
+  response.writeHead(200, { 'content-type': 'application/json' });
+  response.end(JSON.stringify({ ok: true, data }));
+};
+const server = http.createServer(async (request, response) => {
+  try {
+    const chunks = [];
+    for await (const chunk of request) chunks.push(chunk);
+    const body = chunks.length ? JSON.parse(Buffer.concat(chunks)) : {};
+    const pathname = new URL(request.url, 'http://proof.invalid').pathname;
+    if (request.method === 'POST' && pathname === '/v1/agents') {
+      sendJson(response, { id: 'agent_proof_broker', workspace_id: 'ws_proof', name: body.name,
+        token: 'at_proof', status: 'active', created_at: '2026-09-14T00:00:00Z' });
+    } else {
+      sendJson(response, {});
+    }
+  } catch (error) { fixtureError = error; response.destroy(); }
+});
+server.on('connection', (socket) => {
+  sockets.add(socket);
+  socket.on('error', () => {});
+  socket.once('close', () => sockets.delete(socket));
+});
+server.on('upgrade', (request, socket, initialData) => {
+  const key = request.headers['sec-websocket-key'];
+  if (typeof key !== 'string') { socket.destroy(); return; }
+  const accept = createHash('sha1').update(key + '258EAFA5-E914-47DA-95CA-C5AB0DC85B11').digest('base64');
+  socket.write(`HTTP/1.1 101 Switching Protocols\r\nUpgrade: websocket\r\nConnection: Upgrade\r\nSec-WebSocket-Accept: ${accept}\r\n\r\n`);
+  const isNode = new URL(request.url, 'http://proof.invalid').pathname === '/v1/node/ws';
+  const connection = { connectedMs: performance.now() - started, pongs: 0, inventory: 0, acknowledgements: 0, heartbeats: 0 };
+  if (isNode) connections.push(connection);
+  attachFrameReader(socket, (frame) => {
+    try {
+      if (frame.opcode === 0x9) {
+        sendFrame(socket, 0xa, frame.payload);
+        if (isNode) connection.pongs++;
+        return;
+      }
+      if (frame.opcode !== 0x1 || !isNode) return;
+      const message = JSON.parse(frame.payload);
+      if (message.type === 'node.heartbeat') connection.heartbeats++;
+      if (message.type === 'node.register') {
+        sendText(socket, { v: 1, type: 'reply', id: message.id ?? 'legacy-register', ok: true, data: {} });
+      }
+      if (message.type === 'inventory.sync') {
+        connection.inventory++;
+        // Accept the initial inventory, then stall only the first session's
+        // application. Its transport continues answering every ping. A new
+        // session is healthy so an unnecessary reconnect loop is observable.
+        if (connection !== connections[0] || connection.inventory === 1) {
+          sendText(socket, { v: 1, type: 'reply', id: message.id ?? 'legacy-inventory', ok: true, data: { reconciled: 0 } });
+          connection.acknowledgements++;
+        }
+      }
+    } catch (error) { fixtureError = error; }
+  }, initialData);
+});
+try {
+  await new Promise((resolve) => server.listen(0, '127.0.0.1', resolve));
+  broker = spawn(binary, ['init', '--instance-name', 'relayflow-inventory-ack', '--workspace-key', 'rk_proof',
+    '--state-dir', stateDir, '--api-port', '0', '--channels', ''], {
+    cwd: scratch,
+    env: { ...process.env, RELAYCAST_BASE_URL: `http://127.0.0.1:${server.address().port}`,
+      RELAY_NODE_ID: 'node_proof_inventory_ack', RELAY_NODE_TOKEN: 'nt_proof',
+      RELAY_BROKER_API_KEY: 'br_proof', AGENT_RELAY_NO_DEBUG_FILES: '1' },
+    stdio: ['ignore', 'ignore', 'pipe'],
+  });
+  broker.stderr.on('data', (chunk) => { stderr = `${stderr}${chunk}`.slice(-8000); });
+  // Uses production 60s inventory cadence and 120s application deadline. No
+  // source rewriting or Cargo is permitted in the protected artifact runner.
+  const deadline = performance.now() + 150_000;
+  while (performance.now() < deadline) {
+    if (fixtureError) throw fixtureError;
+    if (broker.exitCode !== null) throw new Error(`Broker exited (${broker.exitCode}): ${stderr}`);
+    await new Promise((resolve) => setTimeout(resolve, 250));
+  }
+  const first = connections[0];
+  if (!first || first.inventory < 2 || first.pongs < 6 || first.acknowledgements !== 1 || first.heartbeats < 6) {
+    throw new Error(`Application/transport fixture controls missing: ${JSON.stringify(connections)}`);
+  }
+  const reconnected = connections.length === 2;
+  if (connections.length > 2) throw new Error('Healthy replacement unexpectedly reconnected again');
+  if (reconnected) {
+    const gap = connections[1].connectedMs - first.connectedMs;
+    if (gap < 110_000 || gap > 145_000 || connections[1].acknowledgements < 1) {
+      throw new Error(`Unexpected reconnect timing/readiness: ${JSON.stringify(connections)}`);
+    }
+  }
+  await mkdir(path.dirname(resultPath), { recursive: true });
+  await writeFile(resultPath, `${JSON.stringify({ version: 1, caseId: CASE_ID, arm,
+    outcome: reconnected ? 'fixed' : 'bug',
+    signature: reconnected ? 'application_ack_stall_reconnects' : 'application_ack_stall_not_detected',
+    details: 'Exact broker artifact against loopback HTTP/WebSocket fixture; initial inventory accepted, later inventory acknowledgements withheld while pongs continue for 150 seconds.',
+    evidence: { targetSha: expectedSha, harnessSha: headSha, binarySha256, connections },
+  }, null, 2)}\n`);
+} finally {
+  if (broker && broker.exitCode === null) {
+    broker.kill('SIGTERM');
+    await Promise.race([new Promise((resolve) => broker.once('exit', resolve)), new Promise((resolve) => setTimeout(resolve, 5000))]);
+    if (broker.exitCode === null) broker.kill('SIGKILL');
+  }
+  for (const socket of sockets) socket.destroy();
+  await new Promise((resolve) => server.close(resolve));
+  await rm(scratch, { recursive: true, force: true });
+}
+
+function attachFrameReader(socket, onFrame, initialData = Buffer.alloc(0)) {
+  let buffered = Buffer.from(initialData);
+  const consume = (chunk) => {
+    buffered = Buffer.concat([buffered, chunk]);
+    while (true) {
+      const decoded = decodeFrame(buffered);
+      if (!decoded) return;
+      buffered = buffered.subarray(decoded.consumed);
+      onFrame(decoded);
+    }
+  };
+  socket.on('data', consume);
+  if (buffered.length > 0) consume(Buffer.alloc(0));
+}
+
+function decodeFrame(buffer) {
+  if (buffer.length < 2) return undefined;
+  const opcode = buffer[0] & 0x0f;
+  const masked = (buffer[1] & 0x80) !== 0;
+  let length = buffer[1] & 0x7f;
+  let offset = 2;
+  if (length === 126) {
+    if (buffer.length < 4) return undefined;
+    length = buffer.readUInt16BE(2);
+    offset = 4;
+  } else if (length === 127) {
+    if (buffer.length < 10) return undefined;
+    const wideLength = buffer.readBigUInt64BE(2);
+    if (wideLength > BigInt(Number.MAX_SAFE_INTEGER)) throw new Error('WebSocket frame is too large.');
+    length = Number(wideLength);
+    offset = 10;
+  }
+  let mask;
+  if (masked) {
+    if (buffer.length < offset + 4) return undefined;
+    mask = buffer.subarray(offset, offset + 4);
+    offset += 4;
+  }
+  if (buffer.length < offset + length) return undefined;
+  const payload = Buffer.from(buffer.subarray(offset, offset + length));
+  if (mask) {
+    for (let index = 0; index < payload.length; index += 1) {
+      payload[index] ^= mask[index % 4];
+    }
+  }
+  return { opcode, payload, consumed: offset + length };
+}
+
+function sendText(socket, value) {
+  if (!socket || socket.destroyed) return false;
+  sendFrame(socket, 0x1, Buffer.from(JSON.stringify(value)));
+  return true;
+}
+
+function sendFrame(socket, opcode, payload) {
+  const length = payload.length;
+  let header;
+  if (length < 126) {
+    header = Buffer.from([0x80 | opcode, length]);
+  } else if (length <= 0xffff) {
+    header = Buffer.alloc(4);
+    header[0] = 0x80 | opcode;
+    header[1] = 126;
+    header.writeUInt16BE(length, 2);
+  } else {
+    header = Buffer.alloc(10);
+    header[0] = 0x80 | opcode;
+    header[1] = 127;
+    header.writeBigUInt64BE(BigInt(length), 2);
+  }
+  socket.write(Buffer.concat([header, payload]));
+}

From ed1dd82231a42b92b81d38a8eb601786ee663f0f Mon Sep 17 00:00:00 2001
From: Miya <khaliqgant+miya@gmail.com>
Date: Mon, 14 Sep 2026 05:15:23 +0200
Subject: [PATCH 2/9] test(broker): respect strict inventory reply wire
 validation

Session-Id: 01a09dbd-b8ff-7072-927d-2f9f2c403790

Session-Id: 01a09dbd-b8ff-7072-927d-2f9f2c403790
---
 crates/broker/src/node_control.rs | 18 +++++++++---------
 1 file changed, 9 insertions(+), 9 deletions(-)

diff --git a/crates/broker/src/node_control.rs b/crates/broker/src/node_control.rs
index 71e2297b5f..2189ce7a03 100644
--- a/crates/broker/src/node_control.rs
+++ b/crates/broker/src/node_control.rs
@@ -1603,8 +1603,8 @@ pub(crate) async fn run_node_control_client(
     let mut reconnect_delay = INITIAL_RECONNECT_DELAY;
     // Bound re-minting so a persistently-rejecting engine can't spin a tight
     // mint loop. This counter increments on every consecutive `/v1/node/ws` 401
-    // and only resets once a connection actually establishes (the `Disconnected`
-    // arm below) — NOT on a successful re-mint. So repeated 401s accumulate
+    // and only resets once a correlated inventory acknowledgement proves the
+    // application processed this connection — NOT on a successful re-mint. So repeated 401s accumulate
     // toward [`MAX_UNAUTHORIZED_BEFORE_GIVING_UP`] even when each mint succeeds,
     // and each retry honors the backoff sleep at the bottom of the loop.
     let mut consecutive_unauthorized: u32 = 0;
@@ -1750,8 +1750,8 @@ pub(crate) async fn run_node_control_client(
                         // the next connect attempt so a server that 401s every
                         // freshly minted token can't be hammered. The counter is
                         // intentionally NOT reset here; it only resets once a
-                        // connection actually establishes (the `Disconnected` arm
-                        // above), so repeated 401s still accumulate toward the cap
+                        // correlated inventory reply establishes application readiness
+                        // (the `Disconnected` arm above), so repeated 401s still accumulate toward the cap
                         // even when each mint succeeds.
                         config.node_token = Some(fresh);
                         // Mirror the fresh token to the HTTP session so a provider
@@ -2264,9 +2264,6 @@ where
                     return true;
                 }
 
-                if !reply.ok && application_liveness.reject(&reply.id) {
-                    return false;
-                }
                 match application_liveness.acknowledge(&reply.id) {
                     Some(became_ready) => {
                         if became_ready {
@@ -3591,7 +3588,7 @@ mod tests {
     }
 
     #[tokio::test]
-    async fn rejected_inventory_reply_does_not_make_application_ready() {
+    async fn malformed_inventory_reply_does_not_make_application_ready() {
         let mut liveness = ApplicationLiveness::new(Duration::from_secs(1));
         liveness.track_inventory_sync("inventory-rejected".to_string());
         let (events, _receiver) = mpsc::channel(1);
@@ -3604,7 +3601,10 @@ mod tests {
             "node-test",
             &mut futures_util::sink::drain(),
         ).await;
-        assert!(!healthy, "a rejected inventory must replace the connection");
+        assert!(
+            healthy,
+            "malformed frames are ignored until the liveness deadline"
+        );
         assert!(
             !liveness.ready,
             "a rejection is not an application acknowledgement"

From 7f6595d788a4af3c8cfacd966d7bd97456537bcf Mon Sep 17 00:00:00 2001
From: Miya <khaliqgant+miya@gmail.com>
Date: Mon, 14 Sep 2026 05:17:19 +0200
Subject: [PATCH 3/9] style: format application acknowledgement proof case

Session-Id: 01a09dbd-b8ff-7072-927d-2f9f2c403790

Session-Id: 01a09dbd-b8ff-7072-927d-2f9f2c403790
---
 .../1591-application-ack-reconnect/case.json  |   9 +-
 .../1591-application-ack-reconnect/run.mjs    | 181 +++++++++++++-----
 2 files changed, 133 insertions(+), 57 deletions(-)

diff --git a/tests/relayflows/cases/1591-application-ack-reconnect/case.json b/tests/relayflows/cases/1591-application-ack-reconnect/case.json
index 87b5f6807d..b0a80f330c 100644
--- a/tests/relayflows/cases/1591-application-ack-reconnect/case.json
+++ b/tests/relayflows/cases/1591-application-ack-reconnect/case.json
@@ -4,10 +4,7 @@
   "kind": "bugfix",
   "title": "Reconnect when application acknowledgements stop",
   "runner": {
-    "command": [
-      "node",
-      "tests/relayflows/cases/1591-application-ack-reconnect/run.mjs"
-    ]
+    "command": ["node", "tests/relayflows/cases/1591-application-ack-reconnect/run.mjs"]
   },
   "timeoutSeconds": 240,
   "expected": {
@@ -20,7 +17,5 @@
       "signature": "application_ack_stall_reconnects"
     }
   },
-  "requirements": [
-    "broker-linux-x64"
-  ]
+  "requirements": ["broker-linux-x64"]
 }
diff --git a/tests/relayflows/cases/1591-application-ack-reconnect/run.mjs b/tests/relayflows/cases/1591-application-ack-reconnect/run.mjs
index 2c2d01661a..0bbc16b82e 100644
--- a/tests/relayflows/cases/1591-application-ack-reconnect/run.mjs
+++ b/tests/relayflows/cases/1591-application-ack-reconnect/run.mjs
@@ -19,12 +19,17 @@ const targetDir = path.resolve(required('RELAY_PR_PROOF_TARGET_DIR'));
 const harnessDir = path.resolve(required('RELAY_PR_PROOF_HARNESS_DIR'));
 const expectedSha = required(arm === 'base' ? 'RELAY_PR_PROOF_BASE_SHA' : 'RELAY_PR_PROOF_HEAD_SHA');
 const headSha = required('RELAY_PR_PROOF_HEAD_SHA');
-const shaAt = (directory) => execFileSync('git', ['-C', directory, 'rev-parse', 'HEAD'], { encoding: 'utf8' }).trim();
-if (shaAt(targetDir) !== expectedSha || shaAt(harnessDir) !== headSha) throw new Error('Exact target/harness SHA mismatch');
-if (fileURLToPath(import.meta.url) !== path.join(harnessDir, 'tests/relayflows/cases', CASE_ID, 'run.mjs')) throw new Error('Runner must come from exact-head harness');
+const shaAt = (directory) =>
+  execFileSync('git', ['-C', directory, 'rev-parse', 'HEAD'], { encoding: 'utf8' }).trim();
+if (shaAt(targetDir) !== expectedSha || shaAt(harnessDir) !== headSha)
+  throw new Error('Exact target/harness SHA mismatch');
+if (fileURLToPath(import.meta.url) !== path.join(harnessDir, 'tests/relayflows/cases', CASE_ID, 'run.mjs'))
+  throw new Error('Runner must come from exact-head harness');
 const binary = path.resolve(required('RELAY_PR_PROOF_BROKER_BINARY'));
 await access(binary, constants.R_OK | constants.X_OK);
-const binarySha256 = createHash('sha256').update(await readFile(binary)).digest('hex');
+const binarySha256 = createHash('sha256')
+  .update(await readFile(binary))
+  .digest('hex');
 const resultPath = path.resolve(required('RELAY_PR_PROOF_RESULT_PATH'));
 const scratch = await mkdtemp(path.join(tmpdir(), 'relayflow-inventory-ack-'));
 const stateDir = path.join(scratch, 'state');
@@ -46,12 +51,21 @@ const server = http.createServer(async (request, response) => {
     const body = chunks.length ? JSON.parse(Buffer.concat(chunks)) : {};
     const pathname = new URL(request.url, 'http://proof.invalid').pathname;
     if (request.method === 'POST' && pathname === '/v1/agents') {
-      sendJson(response, { id: 'agent_proof_broker', workspace_id: 'ws_proof', name: body.name,
-        token: 'at_proof', status: 'active', created_at: '2026-09-14T00:00:00Z' });
+      sendJson(response, {
+        id: 'agent_proof_broker',
+        workspace_id: 'ws_proof',
+        name: body.name,
+        token: 'at_proof',
+        status: 'active',
+        created_at: '2026-09-14T00:00:00Z',
+      });
     } else {
       sendJson(response, {});
     }
-  } catch (error) { fixtureError = error; response.destroy(); }
+  } catch (error) {
+    fixtureError = error;
+    response.destroy();
+  }
 });
 server.on('connection', (socket) => {
   sockets.add(socket);
@@ -60,49 +74,96 @@ server.on('connection', (socket) => {
 });
 server.on('upgrade', (request, socket, initialData) => {
   const key = request.headers['sec-websocket-key'];
-  if (typeof key !== 'string') { socket.destroy(); return; }
-  const accept = createHash('sha1').update(key + '258EAFA5-E914-47DA-95CA-C5AB0DC85B11').digest('base64');
-  socket.write(`HTTP/1.1 101 Switching Protocols\r\nUpgrade: websocket\r\nConnection: Upgrade\r\nSec-WebSocket-Accept: ${accept}\r\n\r\n`);
+  if (typeof key !== 'string') {
+    socket.destroy();
+    return;
+  }
+  const accept = createHash('sha1')
+    .update(key + '258EAFA5-E914-47DA-95CA-C5AB0DC85B11')
+    .digest('base64');
+  socket.write(
+    `HTTP/1.1 101 Switching Protocols\r\nUpgrade: websocket\r\nConnection: Upgrade\r\nSec-WebSocket-Accept: ${accept}\r\n\r\n`
+  );
   const isNode = new URL(request.url, 'http://proof.invalid').pathname === '/v1/node/ws';
-  const connection = { connectedMs: performance.now() - started, pongs: 0, inventory: 0, acknowledgements: 0, heartbeats: 0 };
+  const connection = {
+    connectedMs: performance.now() - started,
+    pongs: 0,
+    inventory: 0,
+    acknowledgements: 0,
+    heartbeats: 0,
+  };
   if (isNode) connections.push(connection);
-  attachFrameReader(socket, (frame) => {
-    try {
-      if (frame.opcode === 0x9) {
-        sendFrame(socket, 0xa, frame.payload);
-        if (isNode) connection.pongs++;
-        return;
-      }
-      if (frame.opcode !== 0x1 || !isNode) return;
-      const message = JSON.parse(frame.payload);
-      if (message.type === 'node.heartbeat') connection.heartbeats++;
-      if (message.type === 'node.register') {
-        sendText(socket, { v: 1, type: 'reply', id: message.id ?? 'legacy-register', ok: true, data: {} });
-      }
-      if (message.type === 'inventory.sync') {
-        connection.inventory++;
-        // Accept the initial inventory, then stall only the first session's
-        // application. Its transport continues answering every ping. A new
-        // session is healthy so an unnecessary reconnect loop is observable.
-        if (connection !== connections[0] || connection.inventory === 1) {
-          sendText(socket, { v: 1, type: 'reply', id: message.id ?? 'legacy-inventory', ok: true, data: { reconciled: 0 } });
-          connection.acknowledgements++;
+  attachFrameReader(
+    socket,
+    (frame) => {
+      try {
+        if (frame.opcode === 0x9) {
+          sendFrame(socket, 0xa, frame.payload);
+          if (isNode) connection.pongs++;
+          return;
+        }
+        if (frame.opcode !== 0x1 || !isNode) return;
+        const message = JSON.parse(frame.payload);
+        if (message.type === 'node.heartbeat') connection.heartbeats++;
+        if (message.type === 'node.register') {
+          sendText(socket, { v: 1, type: 'reply', id: message.id ?? 'legacy-register', ok: true, data: {} });
         }
+        if (message.type === 'inventory.sync') {
+          connection.inventory++;
+          // Accept the initial inventory, then stall only the first session's
+          // application. Its transport continues answering every ping. A new
+          // session is healthy so an unnecessary reconnect loop is observable.
+          if (connection !== connections[0] || connection.inventory === 1) {
+            sendText(socket, {
+              v: 1,
+              type: 'reply',
+              id: message.id ?? 'legacy-inventory',
+              ok: true,
+              data: { reconciled: 0 },
+            });
+            connection.acknowledgements++;
+          }
+        }
+      } catch (error) {
+        fixtureError = error;
       }
-    } catch (error) { fixtureError = error; }
-  }, initialData);
+    },
+    initialData
+  );
 });
 try {
   await new Promise((resolve) => server.listen(0, '127.0.0.1', resolve));
-  broker = spawn(binary, ['init', '--instance-name', 'relayflow-inventory-ack', '--workspace-key', 'rk_proof',
-    '--state-dir', stateDir, '--api-port', '0', '--channels', ''], {
-    cwd: scratch,
-    env: { ...process.env, RELAYCAST_BASE_URL: `http://127.0.0.1:${server.address().port}`,
-      RELAY_NODE_ID: 'node_proof_inventory_ack', RELAY_NODE_TOKEN: 'nt_proof',
-      RELAY_BROKER_API_KEY: 'br_proof', AGENT_RELAY_NO_DEBUG_FILES: '1' },
-    stdio: ['ignore', 'ignore', 'pipe'],
+  broker = spawn(
+    binary,
+    [
+      'init',
+      '--instance-name',
+      'relayflow-inventory-ack',
+      '--workspace-key',
+      'rk_proof',
+      '--state-dir',
+      stateDir,
+      '--api-port',
+      '0',
+      '--channels',
+      '',
+    ],
+    {
+      cwd: scratch,
+      env: {
+        ...process.env,
+        RELAYCAST_BASE_URL: `http://127.0.0.1:${server.address().port}`,
+        RELAY_NODE_ID: 'node_proof_inventory_ack',
+        RELAY_NODE_TOKEN: 'nt_proof',
+        RELAY_BROKER_API_KEY: 'br_proof',
+        AGENT_RELAY_NO_DEBUG_FILES: '1',
+      },
+      stdio: ['ignore', 'ignore', 'pipe'],
+    }
+  );
+  broker.stderr.on('data', (chunk) => {
+    stderr = `${stderr}${chunk}`.slice(-8000);
   });
-  broker.stderr.on('data', (chunk) => { stderr = `${stderr}${chunk}`.slice(-8000); });
   // Uses production 60s inventory cadence and 120s application deadline. No
   // source rewriting or Cargo is permitted in the protected artifact runner.
   const deadline = performance.now() + 150_000;
@@ -112,7 +173,13 @@ try {
     await new Promise((resolve) => setTimeout(resolve, 250));
   }
   const first = connections[0];
-  if (!first || first.inventory < 2 || first.pongs < 6 || first.acknowledgements !== 1 || first.heartbeats < 6) {
+  if (
+    !first ||
+    first.inventory < 2 ||
+    first.pongs < 6 ||
+    first.acknowledgements !== 1 ||
+    first.heartbeats < 6
+  ) {
     throw new Error(`Application/transport fixture controls missing: ${JSON.stringify(connections)}`);
   }
   const reconnected = connections.length === 2;
@@ -124,16 +191,30 @@ try {
     }
   }
   await mkdir(path.dirname(resultPath), { recursive: true });
-  await writeFile(resultPath, `${JSON.stringify({ version: 1, caseId: CASE_ID, arm,
-    outcome: reconnected ? 'fixed' : 'bug',
-    signature: reconnected ? 'application_ack_stall_reconnects' : 'application_ack_stall_not_detected',
-    details: 'Exact broker artifact against loopback HTTP/WebSocket fixture; initial inventory accepted, later inventory acknowledgements withheld while pongs continue for 150 seconds.',
-    evidence: { targetSha: expectedSha, harnessSha: headSha, binarySha256, connections },
-  }, null, 2)}\n`);
+  await writeFile(
+    resultPath,
+    `${JSON.stringify(
+      {
+        version: 1,
+        caseId: CASE_ID,
+        arm,
+        outcome: reconnected ? 'fixed' : 'bug',
+        signature: reconnected ? 'application_ack_stall_reconnects' : 'application_ack_stall_not_detected',
+        details:
+          'Exact broker artifact against loopback HTTP/WebSocket fixture; initial inventory accepted, later inventory acknowledgements withheld while pongs continue for 150 seconds.',
+        evidence: { targetSha: expectedSha, harnessSha: headSha, binarySha256, connections },
+      },
+      null,
+      2
+    )}\n`
+  );
 } finally {
   if (broker && broker.exitCode === null) {
     broker.kill('SIGTERM');
-    await Promise.race([new Promise((resolve) => broker.once('exit', resolve)), new Promise((resolve) => setTimeout(resolve, 5000))]);
+    await Promise.race([
+      new Promise((resolve) => broker.once('exit', resolve)),
+      new Promise((resolve) => setTimeout(resolve, 5000)),
+    ]);
     if (broker.exitCode === null) broker.kill('SIGKILL');
   }
   for (const socket of sockets) socket.destroy();

From e3e3ad3cd89f25faf5649d79415758843c658ed6 Mon Sep 17 00:00:00 2001
From: Miya <khaliqgant+miya@gmail.com>
Date: Mon, 14 Sep 2026 05:26:23 +0200
Subject: [PATCH 4/9] fix(proof): read broker artifact without separate access
 check

Session-Id: 01a09dbd-b8ff-7072-927d-2f9f2c403790

Session-Id: 01a09dbd-b8ff-7072-927d-2f9f2c403790
---
 .../cases/1591-application-ack-reconnect/run.mjs           | 7 ++++---
 1 file changed, 4 insertions(+), 3 deletions(-)

diff --git a/tests/relayflows/cases/1591-application-ack-reconnect/run.mjs b/tests/relayflows/cases/1591-application-ack-reconnect/run.mjs
index 0bbc16b82e..176137788c 100644
--- a/tests/relayflows/cases/1591-application-ack-reconnect/run.mjs
+++ b/tests/relayflows/cases/1591-application-ack-reconnect/run.mjs
@@ -1,8 +1,7 @@
 import { execFileSync, spawn } from 'node:child_process';
 import { createHash } from 'node:crypto';
 import http from 'node:http';
-import { access, mkdir, mkdtemp, readFile, rm, writeFile } from 'node:fs/promises';
-import { constants } from 'node:fs';
+import { mkdir, mkdtemp, readFile, rm, writeFile } from 'node:fs/promises';
 import { tmpdir } from 'node:os';
 import path from 'node:path';
 import { fileURLToPath } from 'node:url';
@@ -26,7 +25,6 @@ if (shaAt(targetDir) !== expectedSha || shaAt(harnessDir) !== headSha)
 if (fileURLToPath(import.meta.url) !== path.join(harnessDir, 'tests/relayflows/cases', CASE_ID, 'run.mjs'))
   throw new Error('Runner must come from exact-head harness');
 const binary = path.resolve(required('RELAY_PR_PROOF_BROKER_BINARY'));
-await access(binary, constants.R_OK | constants.X_OK);
 const binarySha256 = createHash('sha256')
   .update(await readFile(binary))
   .digest('hex');
@@ -161,6 +159,9 @@ try {
       stdio: ['ignore', 'ignore', 'pipe'],
     }
   );
+  broker.once('error', (error) => {
+    fixtureError = error;
+  });
   broker.stderr.on('data', (chunk) => {
     stderr = `${stderr}${chunk}`.slice(-8000);
   });

From 07e314a650d3e206a4134f72c9cc9003e8219892 Mon Sep 17 00:00:00 2001
From: Miya <khaliqgant+miya@gmail.com>
Date: Tue, 15 Sep 2026 01:00:22 +0200
Subject: [PATCH 5/9] test(proof): validate broker registration fixture payload

Session-Id: 01a09dbd-b8ff-7072-927d-2f9f2c403790

Session-Id: 01a09dbd-b8ff-7072-927d-2f9f2c403790
---
 .../pr-proof-inventory-http-fixture.test.ts   | 55 +++++++++++++++++++
 .../http-fixture.mjs                          | 47 ++++++++++++++++
 .../1591-application-ack-reconnect/run.mjs    | 25 +++------
 3 files changed, 110 insertions(+), 17 deletions(-)
 create mode 100644 tests/fixtures/pr-proof-inventory-http-fixture.test.ts
 create mode 100644 tests/relayflows/cases/1591-application-ack-reconnect/http-fixture.mjs

diff --git a/tests/fixtures/pr-proof-inventory-http-fixture.test.ts b/tests/fixtures/pr-proof-inventory-http-fixture.test.ts
new file mode 100644
index 0000000000..b8f22e1672
--- /dev/null
+++ b/tests/fixtures/pr-proof-inventory-http-fixture.test.ts
@@ -0,0 +1,55 @@
+import { describe, expect, it } from 'vitest';
+import {
+  BROKER_NAME,
+  BROKER_TYPE,
+  BROKER_IDENTITY_HASH,
+  brokerHttpResponse,
+} from '../relayflows/cases/1591-application-ack-reconnect/http-fixture.mjs';
+
+const registration = () => ({
+  name: BROKER_NAME,
+  type: BROKER_TYPE,
+  metadata: { identity_key: BROKER_IDENTITY_HASH },
+});
+
+describe('inventory liveness HTTP fixture (unit only)', () => {
+  it('preserves the valid broker registration response', () => {
+    expect(brokerHttpResponse('POST', '/v1/agents', registration())).toEqual({
+      status: 200,
+      body: {
+        ok: true,
+        data: {
+          id: 'agent_proof_broker',
+          workspace_id: 'ws_proof',
+          name: BROKER_NAME,
+          token: 'at_proof',
+          status: 'active',
+          created_at: '2026-09-14T00:00:00Z',
+        },
+      },
+    });
+  });
+  it.each([
+    null,
+    {},
+    { ...registration(), name: 'wrong' },
+    { ...registration(), type: 'agent' },
+    { ...registration(), metadata: {} },
+    { ...registration(), metadata: { identity_key: 'wrong' } },
+  ])('rejects invalid registration %j', (body) => {
+    expect(brokerHttpResponse('POST', '/v1/agents', body)).toMatchObject({
+      status: 400,
+      body: { ok: false, error: { code: 'fixture_registration_invalid' } },
+    });
+  });
+  it.each([
+    ['GET', '/v1/agents'],
+    ['POST', '/v1/agent'],
+    ['PATCH', '/v1/agents/worker'],
+  ])('rejects unexpected route %s %s', (method, route) => {
+    expect(brokerHttpResponse(method, route, registration())).toMatchObject({
+      status: 404,
+      body: { ok: false },
+    });
+  });
+});
diff --git a/tests/relayflows/cases/1591-application-ack-reconnect/http-fixture.mjs b/tests/relayflows/cases/1591-application-ack-reconnect/http-fixture.mjs
new file mode 100644
index 0000000000..fc9753a93d
--- /dev/null
+++ b/tests/relayflows/cases/1591-application-ack-reconnect/http-fixture.mjs
@@ -0,0 +1,47 @@
+import { createHash } from 'node:crypto';
+
+export const BROKER_NAME = 'relayflow-inventory-ack';
+export const BROKER_TYPE = 'human';
+// Synthetic isolated-fixture identity, never a real workspace credential.
+export const BROKER_IDENTITY = 'relayflow-inventory-ack-fixture-identity';
+export const BROKER_IDENTITY_HASH = createHash('sha256').update(BROKER_IDENTITY).digest('hex');
+
+/** Validate the HTTP bootstrap independently of the WebSocket liveness oracle. */
+export function brokerHttpResponse(method, pathname, body) {
+  if (method !== 'POST' || pathname !== '/v1/agents') {
+    return {
+      status: 404,
+      body: { ok: false, error: { code: 'fixture_route_not_found', message: 'Unexpected fixture route' } },
+    };
+  }
+  if (
+    body?.name !== BROKER_NAME ||
+    body?.type !== BROKER_TYPE ||
+    body?.metadata?.identity_key !== BROKER_IDENTITY_HASH
+  ) {
+    return {
+      status: 400,
+      body: {
+        ok: false,
+        error: {
+          code: 'fixture_registration_invalid',
+          message: 'Unexpected registration name, type, or identity metadata',
+        },
+      },
+    };
+  }
+  return {
+    status: 200,
+    body: {
+      ok: true,
+      data: {
+        id: 'agent_proof_broker',
+        workspace_id: 'ws_proof',
+        name: BROKER_NAME,
+        token: 'at_proof',
+        status: 'active',
+        created_at: '2026-09-14T00:00:00Z',
+      },
+    },
+  };
+}
diff --git a/tests/relayflows/cases/1591-application-ack-reconnect/run.mjs b/tests/relayflows/cases/1591-application-ack-reconnect/run.mjs
index 176137788c..d317d6677a 100644
--- a/tests/relayflows/cases/1591-application-ack-reconnect/run.mjs
+++ b/tests/relayflows/cases/1591-application-ack-reconnect/run.mjs
@@ -6,6 +6,8 @@ import { tmpdir } from 'node:os';
 import path from 'node:path';
 import { fileURLToPath } from 'node:url';
 
+import { BROKER_NAME, BROKER_TYPE, BROKER_IDENTITY, brokerHttpResponse } from './http-fixture.mjs';
+
 const CASE_ID = '1591-application-ack-reconnect';
 const required = (name) => {
   const value = process.env[name];
@@ -38,28 +40,15 @@ const started = performance.now();
 let broker;
 let stderr = '';
 let fixtureError;
-const sendJson = (response, data) => {
-  response.writeHead(200, { 'content-type': 'application/json' });
-  response.end(JSON.stringify({ ok: true, data }));
-};
 const server = http.createServer(async (request, response) => {
   try {
     const chunks = [];
     for await (const chunk of request) chunks.push(chunk);
     const body = chunks.length ? JSON.parse(Buffer.concat(chunks)) : {};
     const pathname = new URL(request.url, 'http://proof.invalid').pathname;
-    if (request.method === 'POST' && pathname === '/v1/agents') {
-      sendJson(response, {
-        id: 'agent_proof_broker',
-        workspace_id: 'ws_proof',
-        name: body.name,
-        token: 'at_proof',
-        status: 'active',
-        created_at: '2026-09-14T00:00:00Z',
-      });
-    } else {
-      sendJson(response, {});
-    }
+    const result = brokerHttpResponse(request.method, pathname, body);
+    response.writeHead(result.status, { 'content-type': 'application/json' });
+    response.end(JSON.stringify(result.body));
   } catch (error) {
     fixtureError = error;
     response.destroy();
@@ -136,7 +125,7 @@ try {
     [
       'init',
       '--instance-name',
-      'relayflow-inventory-ack',
+      BROKER_NAME,
       '--workspace-key',
       'rk_proof',
       '--state-dir',
@@ -151,6 +140,8 @@ try {
       env: {
         ...process.env,
         RELAYCAST_BASE_URL: `http://127.0.0.1:${server.address().port}`,
+        RELAY_AGENT_TYPE: BROKER_TYPE,
+        RELAY_AGENT_IDENTITY_KEY: BROKER_IDENTITY,
         RELAY_NODE_ID: 'node_proof_inventory_ack',
         RELAY_NODE_TOKEN: 'nt_proof',
         RELAY_BROKER_API_KEY: 'br_proof',

From 3b105f65353dc88ff2ed4287b8178c5189293382 Mon Sep 17 00:00:00 2001
From: "github-actions[bot]" <github-actions[bot]@users.noreply.github.com>
Date: Tue, 15 Sep 2026 15:55:58 +0000
Subject: [PATCH 6/9] style: auto-format Rust code with cargo fmt

---
 crates/broker/src/node_control.rs | 8 ++++++--
 1 file changed, 6 insertions(+), 2 deletions(-)

diff --git a/crates/broker/src/node_control.rs b/crates/broker/src/node_control.rs
index 78952c2fa4..1b882200f0 100644
--- a/crates/broker/src/node_control.rs
+++ b/crates/broker/src/node_control.rs
@@ -2424,8 +2424,12 @@ where
                                     true
                                 }
                                 None => {
-                                    complete_agent_registration(reply, pending_agent_registrations, sink)
-                                        .await
+                                    complete_agent_registration(
+                                        reply,
+                                        pending_agent_registrations,
+                                        sink,
+                                    )
+                                    .await
                                 }
                             }
                         }

From bc32abb30363038a9a38b12023a60a72190012a2 Mon Sep 17 00:00:00 2001
From: agentrelaybot <agentrelaybot@agentrelay.dev>
Date: Tue, 15 Sep 2026 09:12:25 -0700
Subject: [PATCH 7/9] fix(broker): repair post-merge compile/clippy breakage in
 node_control

- registration_tests.rs still asserted the pre-#1770 unit-style
  ControlRunResult::Disconnected; this fixture never sends a
  correlated inventory.sync reply (it exercises the registration
  gate, not the ack-liveness deadline), so application_ready is
  always false here.
- handle_server_message grew to 8 parameters once #1769's
  registration-gate and #1770's application-liveness params were
  combined; allow clippy::too_many_arguments to match existing
  precedent elsewhere in this crate (snippets.rs, pty_worker.rs).

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
---
 crates/broker/src/node_control.rs                    |  1 +
 crates/broker/src/node_control/registration_tests.rs | 10 +++++++++-
 2 files changed, 10 insertions(+), 1 deletion(-)

diff --git a/crates/broker/src/node_control.rs b/crates/broker/src/node_control.rs
index 769e3323a8..d2adbdb200 100644
--- a/crates/broker/src/node_control.rs
+++ b/crates/broker/src/node_control.rs
@@ -2435,6 +2435,7 @@ where
     true
 }
 
+#[allow(clippy::too_many_arguments)]
 async fn handle_server_message<S>(
     message: Message,
     event_tx: &mpsc::Sender<FleetControlEvent>,
diff --git a/crates/broker/src/node_control/registration_tests.rs b/crates/broker/src/node_control/registration_tests.rs
index efb19788b1..f0297548cc 100644
--- a/crates/broker/src/node_control/registration_tests.rs
+++ b/crates/broker/src/node_control/registration_tests.rs
@@ -246,7 +246,15 @@ async fn registration_gate_case(response: &str) {
     })
     .await
     .expect("registration case must terminate after its bounded protocol exchange");
-    assert_eq!(result, ControlRunResult::Disconnected);
+    // Only a correlated `inventory.sync` reply proves application liveness,
+    // and this fixture never sends one (accepted or not) — it exercises the
+    // registration gate, not the inventory-ack liveness deadline.
+    assert_eq!(
+        result,
+        ControlRunResult::Disconnected {
+            application_ready: false,
+        }
+    );
     assert!(
         event_rx.try_recv().is_err(),
         "no unexpected or duplicate runtime events"

From 2152ad74d69d6b1caa0f62a712172eeebd2cd8ef Mon Sep 17 00:00:00 2001
From: agentrelaybot <agentrelaybot@agentrelay.dev>
Date: Tue, 15 Sep 2026 10:37:25 -0700
Subject: [PATCH 8/9] fix(broker): stop Shutdown from stranding behind a stuck
 registration gate
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

register_node_session() blocks up to read_idle_timeout waiting for a
node.register reply before the main command loop in run_connected_once
ever starts. If the peer never replies (unreachable relaycast, or a
peer that accepts the transport handshake but never registers), a
Shutdown command sent during that window was never observed: the
outer reconnect loop just kept retrying registration forever, each
attempt re-entering the same gate before command_rx was ever polled
again. Reproduced locally as a genuine hang in
pre_ready_disconnects_preserve_exponential_reconnect_backoff (cargo
test on this file never completed).

register_node_session now races command_rx alongside the wire wait.
Shutdown (or a closed channel) ends the wait immediately. Every other
command received during the window is queued and replayed, in order,
through a new handle_connected_command() helper shared with the main
select! loop — so an UpdateInventory or RegisterAgent that arrives
mid-registration still gets exactly the same wire round trip and
sync/ack behavior it would have gotten had it arrived a moment later,
after the registration reply, rather than being silently dropped or
folded early into state that hasn't been sent yet.

Verified locally (no toolchain available in CI's sandbox for this
session, so validated directly): full agent-relay-broker test suite
(1157 tests, debug and release) passes, including the previously-hung
test in isolation; cargo clippy -D warnings and cargo fmt --check are
clean.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
---
 crates/broker/src/node_control.rs | 395 +++++++++++++++++++++---------
 1 file changed, 273 insertions(+), 122 deletions(-)

diff --git a/crates/broker/src/node_control.rs b/crates/broker/src/node_control.rs
index d2adbdb200..e935ff6c50 100644
--- a/crates/broker/src/node_control.rs
+++ b/crates/broker/src/node_control.rs
@@ -2007,12 +2007,24 @@ impl Drop for ProbeSessionGuard<'_> {
 /// engine can reject node.register while leaving the socket open; sending
 /// inventory or heartbeats then updates a fallback provider and masks the loss.
 /// Only this request's successful reply opens the application delivery path.
+///
+/// Also races `command_rx` so a `Shutdown` (or a closed command channel) can
+/// interrupt the wait. Without this, a `Shutdown` sent while a connection is
+/// mid-registration is stranded: the main command loop below only starts once
+/// registration resolves, so on a peer that never replies the caller would
+/// retry this gate forever and never observe the request to stop. Every other
+/// command received during the wait is queued in the returned `Vec` rather
+/// than applied here, so the caller can replay it through the exact same path
+/// the main loop uses once this gate opens — preserving the ordering an
+/// `UpdateInventory`/`RegisterAgent`/etc. would have had if it had simply
+/// arrived a moment later, after the registration reply.
 async fn register_node_session<S, R>(
     sink: &mut S,
     stream: &mut R,
     registration: &mut NodeRegister,
     config: &FleetControlConfig,
-) -> bool
+    command_rx: &mut mpsc::Receiver<FleetControlCommand>,
+) -> (Option<bool>, Vec<FleetControlCommand>)
 where
     S: Sink<Message> + Unpin,
     S::Error: std::error::Error + Send + Sync + 'static,
@@ -2026,47 +2038,218 @@ where
         .read_idle_timeout
         .unwrap_or(Duration::from_secs(10))
         .min(Duration::from_secs(10));
-    let accepted = tokio::time::timeout(deadline, async {
+    let mut deferred_commands: Vec<FleetControlCommand> = Vec::new();
+    let outcome = tokio::time::timeout(deadline, async {
         if send_wire(sink, &BrokerToRelaycast::NodeRegister(registration.clone())).await.is_err() {
-            return false;
+            return Some(false);
         }
-        while let Some(Ok(message)) = stream.next().await {
-            match message {
-                Message::Text(text) => {
-                    if let Some(probe) = config.probe.as_ref() { probe.record_text_frame(); }
-                    match serde_json::from_str::<RelaycastToBroker>(&text) {
-                        Ok(frame) => {
-                            if let Some(probe) = config.probe.as_ref() { probe.record_frame(&frame); }
-                            match frame {
-                                RelaycastToBroker::Reply(reply) if reply.id == id => return reply.ok,
-                                RelaycastToBroker::Error(error) => {
-                                    tracing::error!(code = %error.code, "node registration rejected; reconnecting without advertising delivery readiness");
-                                    return false;
+        loop {
+            tokio::select! {
+                message = stream.next() => {
+                    let Some(Ok(message)) = message else { return Some(false); };
+                    match message {
+                        Message::Text(text) => {
+                            if let Some(probe) = config.probe.as_ref() { probe.record_text_frame(); }
+                            match serde_json::from_str::<RelaycastToBroker>(&text) {
+                                Ok(frame) => {
+                                    if let Some(probe) = config.probe.as_ref() { probe.record_frame(&frame); }
+                                    match frame {
+                                        RelaycastToBroker::Reply(reply) if reply.id == id => return Some(reply.ok),
+                                        RelaycastToBroker::Error(error) => {
+                                            tracing::error!(code = %error.code, "node registration rejected; reconnecting without advertising delivery readiness");
+                                            return Some(false);
+                                        }
+                                        // The engine replies before replaying deliveries. Never
+                                        // acknowledge or inject a frame on an unaccepted provider.
+                                        RelaycastToBroker::Deliver(_) | RelaycastToBroker::ActionInvoke(_) => return Some(false),
+                                        _ => {},
+                                    }
+                                }
+                                Err(error) => {
+                                    if let Some(probe) = config.probe.as_ref() { probe.record_parse_failure(&error.to_string(), &text); }
                                 }
-                                // The engine replies before replaying deliveries. Never
-                                // acknowledge or inject a frame on an unaccepted provider.
-                                RelaycastToBroker::Deliver(_) | RelaycastToBroker::ActionInvoke(_) => return false,
-                                _ => {},
                             }
                         }
-                        Err(error) => {
-                            if let Some(probe) = config.probe.as_ref() { probe.record_parse_failure(&error.to_string(), &text); }
+                        Message::Ping(payload) => {
+                            if sink.send(Message::Pong(payload)).await.is_err() { return Some(false); }
                         }
+                        Message::Close(_) => return Some(false),
+                        _ => {},
                     }
                 }
-                Message::Ping(payload) => {
-                    if sink.send(Message::Pong(payload)).await.is_err() { return false; }
+                command = command_rx.recv() => {
+                    match command {
+                        Some(FleetControlCommand::Shutdown) | None => return None,
+                        Some(other) => deferred_commands.push(other),
+                    }
                 }
-                Message::Close(_) => return false,
-                _ => {},
             }
         }
-        false
-    }).await.unwrap_or(false);
-    if !accepted {
+    }).await;
+    let accepted = match outcome {
+        Ok(result) => result,
+        Err(_) => Some(false),
+    };
+    if accepted == Some(false) {
         tracing::warn!("node registration was not accepted within its deadline; delivery unavailable, reconnecting");
     }
-    accepted
+    (accepted, deferred_commands)
+}
+
+/// Handles one `FleetControlCommand` on an already-registered session — shared
+/// by the main command loop in `run_connected_once` and the replay of commands
+/// deferred while `register_node_session` was still waiting on the wire, so
+/// both paths apply a command identically. `ControlFlow::Break` carries the
+/// `ControlRunResult` the caller should return immediately (a disconnect or
+/// shutdown); `ControlFlow::Continue` means the session stays up.
+#[allow(clippy::too_many_arguments)]
+async fn handle_connected_command<S>(
+    command: Option<FleetControlCommand>,
+    sink: &mut S,
+    config: &FleetControlConfig,
+    provider: &FleetProviderIdentity,
+    node_register: &NodeRegister,
+    registration: &mut Option<NodeRegister>,
+    load: &mut FleetLoadSnapshot,
+    inventory: &mut Vec<InventoryAgent>,
+    pending_agent_registrations: &mut HashMap<String, PendingAgentRegistration>,
+    pending_deregistrations: &mut HashMap<String, oneshot::Sender<Result<(), String>>>,
+    application_liveness: &mut ApplicationLiveness,
+) -> std::ops::ControlFlow<ControlRunResult>
+where
+    S: Sink<Message> + Unpin,
+    S::Error: std::error::Error + Send + Sync + 'static,
+{
+    use std::ops::ControlFlow::{Break, Continue};
+    match command {
+        Some(FleetControlCommand::RegisterNode {
+            manifest,
+            resume_cursor,
+        }) => {
+            load.max_agents = manifest.max_agents.unwrap_or(load.max_agents);
+            load.handlers_live = true;
+            let mut next = build_node_register(
+                &manifest,
+                &config.node_id,
+                &config.node_name,
+                &config.broker_version,
+                resume_cursor,
+            );
+            next.provider = Some(provider.clone());
+            // Reopen the provider session for a new manifest. Running
+            // the registration gate inside this active socket would
+            // consume replies belonging to in-flight agent requests.
+            *registration = Some(next);
+            drain_agent_registrations(pending_agent_registrations, "node_control_reconfiguring");
+            for (_, pending) in pending_deregistrations.drain() {
+                let _ = pending.send(Err("node_control_reconfiguring".to_string()));
+            }
+            Break(ControlRunResult::Disconnected {
+                application_ready: application_liveness.ready,
+            })
+        }
+        Some(FleetControlCommand::UpdateInventory(next)) => {
+            *inventory = next;
+            if !send_inventory_sync(
+                sink,
+                inventory,
+                pending_agent_registrations,
+                application_liveness,
+            )
+            .await
+            {
+                return Break(ControlRunResult::Disconnected {
+                    application_ready: application_liveness.ready,
+                });
+            }
+            Continue(())
+        }
+        Some(FleetControlCommand::UpdateLoad(next)) => {
+            *load = next;
+            Continue(())
+        }
+        Some(FleetControlCommand::HeartbeatNow) => {
+            if send_wire(
+                sink,
+                &BrokerToRelaycast::NodeHeartbeat(load.heartbeat(node_register)),
+            )
+            .await
+            .is_err()
+            {
+                return Break(ControlRunResult::Disconnected {
+                    application_ready: application_liveness.ready,
+                });
+            }
+            Continue(())
+        }
+        Some(FleetControlCommand::Send(message)) => {
+            // A `delivery_ack` is the engine's only evidence that a
+            // frame was consumed. The runtime records the *decision*
+            // to ack before handing it here and cannot wait for the
+            // wire, so the probe learns the outcome at the one place
+            // that knows it. See `NodeDeliveryProbe::record_ack_sent`.
+            let is_ack = matches!(message, BrokerToRelaycast::DeliveryAck(_));
+            let sent = send_wire(sink, &message).await;
+            if let (true, Some(probe)) = (is_ack, config.probe.as_ref()) {
+                if sent.is_ok() {
+                    probe.record_ack_sent();
+                } else {
+                    probe.record_ack_send_failed();
+                }
+            }
+            if sent.is_err() {
+                return Break(ControlRunResult::Disconnected {
+                    application_ready: application_liveness.ready,
+                });
+            }
+            Continue(())
+        }
+        Some(FleetControlCommand::DeregisterAgent { mut request, reply }) => {
+            let request_id = format!("agent_deregister_{}", Uuid::new_v4().simple());
+            request.id = Some(request_id.clone());
+            pending_deregistrations.retain(|_, pending| !pending.is_closed());
+            pending_deregistrations.insert(request_id, reply);
+            if send_wire(sink, &BrokerToRelaycast::AgentDeregister(request))
+                .await
+                .is_err()
+            {
+                return Break(ControlRunResult::Disconnected {
+                    application_ready: application_liveness.ready,
+                });
+            }
+            Continue(())
+        }
+        Some(FleetControlCommand::RegisterAgent { mut request, reply }) => {
+            let request_id = request
+                .id
+                .clone()
+                .unwrap_or_else(|| format!("agent_register_{}", Uuid::new_v4().simple()));
+            request.id = Some(request_id.clone());
+            pending_agent_registrations.insert(
+                request_id,
+                PendingAgentRegistration {
+                    isolates_channels: request.auto_join_general == Some(false),
+                    name: request.name.clone(),
+                    reply,
+                    created_at: Instant::now(),
+                },
+            );
+            if send_wire(sink, &BrokerToRelaycast::AgentRegister(request))
+                .await
+                .is_err()
+            {
+                drain_agent_registrations(pending_agent_registrations, "node_control_disconnected");
+                return Break(ControlRunResult::Disconnected {
+                    application_ready: application_liveness.ready,
+                });
+            }
+            Continue(())
+        }
+        Some(FleetControlCommand::Shutdown) | None => {
+            drain_agent_registrations(pending_agent_registrations, "node_control_shutdown");
+            Break(ControlRunResult::Shutdown)
+        }
+    }
 }
 
 async fn run_connected_once(
@@ -2174,11 +2357,23 @@ async fn run_connected_once(
         .max(read_idle_timeout);
     let mut application_liveness = ApplicationLiveness::new(application_liveness_timeout);
 
-    if !register_node_session(&mut sink, &mut stream, &mut node_register, config).await {
-        return ControlRunResult::Disconnected {
-            application_ready: false,
-        };
-    }
+    let deferred_commands = match register_node_session(
+        &mut sink,
+        &mut stream,
+        &mut node_register,
+        config,
+        command_rx,
+    )
+    .await
+    {
+        (None, _) => return ControlRunResult::Shutdown,
+        (Some(false), _) => {
+            return ControlRunResult::Disconnected {
+                application_ready: false,
+            };
+        }
+        (Some(true), deferred_commands) => deferred_commands,
+    };
     *registration = Some(node_register.clone());
     let _ = event_tx.send(FleetControlEvent::Connected).await;
     if !send_inventory_sync(
@@ -2217,97 +2412,53 @@ async fn run_connected_once(
     inventory_refresh.tick().await;
     let mut last_inbound = Instant::now();
 
+    // Commands that arrived while `register_node_session` was still waiting on
+    // the wire are replayed here, in the order received, through the exact
+    // same handling the main loop below uses — so an `UpdateInventory` that
+    // raced the registration reply still triggers its own `inventory.sync`
+    // push (rather than silently folding into the one already sent) and a
+    // `RegisterAgent`/`DeregisterAgent` still gets a wire round trip instead
+    // of the early rejection `register_node_session` gives commands it can't
+    // service pre-connection.
+    for deferred in deferred_commands {
+        if let std::ops::ControlFlow::Break(result) = handle_connected_command(
+            Some(deferred),
+            &mut sink,
+            config,
+            &provider,
+            &node_register,
+            registration,
+            load,
+            inventory,
+            &mut pending_agent_registrations,
+            &mut pending_deregistrations,
+            &mut application_liveness,
+        )
+        .await
+        {
+            return result;
+        }
+    }
+
     loop {
         tokio::select! {
             command = command_rx.recv() => {
-                match command {
-                    Some(FleetControlCommand::RegisterNode { manifest, resume_cursor }) => {
-                        load.max_agents = manifest.max_agents.unwrap_or(load.max_agents);
-                        load.handlers_live = true;
-                        let mut next = build_node_register(&manifest, &config.node_id, &config.node_name, &config.broker_version, resume_cursor);
-                        next.provider = Some(provider.clone());
-                        // Reopen the provider session for a new manifest. Running
-                        // the registration gate inside this active socket would
-                        // consume replies belonging to in-flight agent requests.
-                        *registration = Some(next);
-                        drain_agent_registrations(&mut pending_agent_registrations, "node_control_reconfiguring");
-                        for (_, pending) in pending_deregistrations.drain() {
-                            let _ = pending.send(Err("node_control_reconfiguring".to_string()));
-                        }
-                        return ControlRunResult::Disconnected { application_ready: application_liveness.ready };
-                    }
-                    Some(FleetControlCommand::UpdateInventory(next)) => {
-                        *inventory = next;
-                        if !send_inventory_sync(
-                            &mut sink,
-                            inventory,
-                            &mut pending_agent_registrations,
-                            &mut application_liveness,
-                        )
-                        .await
-                        {
-                            return ControlRunResult::Disconnected { application_ready: application_liveness.ready };
-                        }
-                    }
-                    Some(FleetControlCommand::UpdateLoad(next)) => {
-                        *load = next;
-                    }
-                    Some(FleetControlCommand::HeartbeatNow) => {
-                        if send_wire(&mut sink, &BrokerToRelaycast::NodeHeartbeat(load.heartbeat(&node_register))).await.is_err() {
-                            return ControlRunResult::Disconnected { application_ready: application_liveness.ready };
-                        }
-                    }
-                    Some(FleetControlCommand::Send(message)) => {
-                        // A `delivery_ack` is the engine's only evidence that a
-                        // frame was consumed. The runtime records the *decision*
-                        // to ack before handing it here and cannot wait for the
-                        // wire, so the probe learns the outcome at the one place
-                        // that knows it. See `NodeDeliveryProbe::record_ack_sent`.
-                        let is_ack = matches!(message, BrokerToRelaycast::DeliveryAck(_));
-                        let sent = send_wire(&mut sink, &message).await;
-                        if let (true, Some(probe)) = (is_ack, config.probe.as_ref()) {
-                            if sent.is_ok() {
-                                probe.record_ack_sent();
-                            } else {
-                                probe.record_ack_send_failed();
-                            }
-                        }
-                        if sent.is_err() {
-                            return ControlRunResult::Disconnected { application_ready: application_liveness.ready };
-                        }
-                    }
-                    Some(FleetControlCommand::DeregisterAgent { mut request, reply }) => {
-                        let request_id = format!("agent_deregister_{}", Uuid::new_v4().simple());
-                        request.id = Some(request_id.clone());
-                        pending_deregistrations.retain(|_, pending| !pending.is_closed());
-                        pending_deregistrations.insert(request_id, reply);
-                        if send_wire(&mut sink, &BrokerToRelaycast::AgentDeregister(request)).await.is_err() {
-                            return ControlRunResult::Disconnected { application_ready: application_liveness.ready };
-                        }
-                    }
-                    Some(FleetControlCommand::RegisterAgent { mut request, reply }) => {
-                        let request_id = request.id.clone().unwrap_or_else(|| {
-                            format!("agent_register_{}", Uuid::new_v4().simple())
-                        });
-                        request.id = Some(request_id.clone());
-                        pending_agent_registrations.insert(
-                            request_id,
-                            PendingAgentRegistration {
-                                isolates_channels: request.auto_join_general == Some(false),
-                                name: request.name.clone(),
-                                reply,
-                                created_at: Instant::now(),
-                            },
-                        );
-                        if send_wire(&mut sink, &BrokerToRelaycast::AgentRegister(request)).await.is_err() {
-                            drain_agent_registrations(&mut pending_agent_registrations, "node_control_disconnected");
-                            return ControlRunResult::Disconnected { application_ready: application_liveness.ready };
-                        }
-                    }
-                    Some(FleetControlCommand::Shutdown) | None => {
-                        drain_agent_registrations(&mut pending_agent_registrations, "node_control_shutdown");
-                        return ControlRunResult::Shutdown;
-                    }
+                if let std::ops::ControlFlow::Break(result) = handle_connected_command(
+                    command,
+                    &mut sink,
+                    config,
+                    &provider,
+                    &node_register,
+                    registration,
+                    load,
+                    inventory,
+                    &mut pending_agent_registrations,
+                    &mut pending_deregistrations,
+                    &mut application_liveness,
+                )
+                .await
+                {
+                    return result;
                 }
             }
             _ = heartbeat.tick() => {

From 6acaf23f1866f0fd1b923ccb5a2327f85f80cd82 Mon Sep 17 00:00:00 2001
From: agentrelaybot <agentrelaybot@agentrelay.dev>
Date: Tue, 15 Sep 2026 11:01:56 -0700
Subject: [PATCH 9/9] fix(broker): address Cursor Bugbot and CodeRabbit
 findings on the registration-gate fix
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Two real issues in 2152ad74d, both caught by automated PR review:

- Cursor Bugbot (high severity): a rejected/timed-out registration, or a
  Shutdown seen mid-registration, discarded register_node_session's
  deferred_commands wholesale instead of finalizing them. RegisterAgent/
  DeregisterAgent callers would hang until their own reply timeout instead
  of getting an immediate error, and an UpdateInventory/UpdateLoad update
  was lost rather than carried into the next connection attempt. The same
  gap existed a second time in the replay loop: breaking out partway
  through (e.g. a wire write failing) silently dropped whatever was still
  queued behind it. Both paths now finalize the untouched remainder via a
  new fail_deferred_commands() — local-state updates are preserved,
  pending replies get an explicit rejection instead of silence.

- CodeRabbit: application_liveness.acknowledge() was called for any
  correlated inventory.sync Reply regardless of reply.ok, so a relaycast
  that keeps explicitly rejecting inventory.sync would still read as
  "ready" — undermining the application-liveness check this PR exists to
  add. Only reply.ok == true now acknowledges; ok == false instead calls
  the existing reject() path, matching how a RelaycastToBroker::Error on
  the same id is already handled.

Verified locally (cargo test -p agent-relay-broker --lib: 1157 passed;
cargo clippy -D warnings and cargo fmt --check clean) since this
sandbox has no toolchain for CI to use directly.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
---
 crates/broker/src/node_control.rs | 69 +++++++++++++++++++++++++++++--
 1 file changed, 65 insertions(+), 4 deletions(-)

diff --git a/crates/broker/src/node_control.rs b/crates/broker/src/node_control.rs
index e935ff6c50..b2d8488e34 100644
--- a/crates/broker/src/node_control.rs
+++ b/crates/broker/src/node_control.rs
@@ -2366,8 +2366,12 @@ async fn run_connected_once(
     )
     .await
     {
-        (None, _) => return ControlRunResult::Shutdown,
-        (Some(false), _) => {
+        (None, deferred_commands) => {
+            fail_deferred_commands(deferred_commands, "node_control_shutdown", inventory, load);
+            return ControlRunResult::Shutdown;
+        }
+        (Some(false), deferred_commands) => {
+            fail_deferred_commands(deferred_commands, "node_not_registered", inventory, load);
             return ControlRunResult::Disconnected {
                 application_ready: false,
             };
@@ -2420,7 +2424,8 @@ async fn run_connected_once(
     // `RegisterAgent`/`DeregisterAgent` still gets a wire round trip instead
     // of the early rejection `register_node_session` gives commands it can't
     // service pre-connection.
-    for deferred in deferred_commands {
+    let mut deferred_commands = deferred_commands.into_iter();
+    for deferred in deferred_commands.by_ref() {
         if let std::ops::ControlFlow::Break(result) = handle_connected_command(
             Some(deferred),
             &mut sink,
@@ -2436,6 +2441,17 @@ async fn run_connected_once(
         )
         .await
         {
+            // The command that broke out already got its outcome (a wire
+            // failure, or a Shutdown handled like any other command here);
+            // anything still queued behind it in this replay never got a
+            // turn and must not be silently dropped, the same as a rejected
+            // registration's leftovers above.
+            let reason = if matches!(result, ControlRunResult::Shutdown) {
+                "node_control_shutdown"
+            } else {
+                "node_control_disconnected"
+            };
+            fail_deferred_commands(deferred_commands.collect(), reason, inventory, load);
             return result;
         }
     }
@@ -2629,7 +2645,17 @@ where
                                 return true;
                             }
 
-                            match application_liveness.acknowledge(&reply.id) {
+                            // A correlated reply proves the transport is alive, but only
+                            // `ok: true` proves the application processed it. Crediting a
+                            // rejection as an acknowledgement would let a relaycast that
+                            // keeps refusing inventory.sync still read as "ready" — the
+                            // exact failure mode this liveness check exists to catch.
+                            let liveness_outcome = if reply.ok {
+                                application_liveness.acknowledge(&reply.id)
+                            } else {
+                                application_liveness.reject(&reply.id).then_some(false)
+                            };
+                            match liveness_outcome {
                                 Some(became_ready) => {
                                     if became_ready {
                                         tracing::info!(
@@ -2871,6 +2897,41 @@ fn drain_agent_registrations(
     }
 }
 
+/// Finalizes commands `register_node_session` deferred but that this
+/// connection attempt cannot service — either the wire gate rejected/timed
+/// out, or a `Shutdown` cut the wait short. Silently dropping these would
+/// strand `RegisterAgent`/`DeregisterAgent` callers until their own reply
+/// timeout and lose an `UpdateInventory`/`UpdateLoad` update entirely, since
+/// (unlike a command still sitting in `command_rx`) they were already taken
+/// out of the channel. `RegisterNode`/`Send`/`HeartbeatNow` are dropped, the
+/// same as `handle_disconnected_command` does before the first connection.
+fn fail_deferred_commands(
+    commands: Vec<FleetControlCommand>,
+    reason: &str,
+    inventory: &mut Vec<InventoryAgent>,
+    load: &mut FleetLoadSnapshot,
+) {
+    for command in commands {
+        match command {
+            FleetControlCommand::RegisterAgent { reply, .. } => {
+                let _ = reply.send(Err(reason.to_string()));
+            }
+            FleetControlCommand::DeregisterAgent { reply, .. } => {
+                let _ = reply.send(Err(reason.to_string()));
+            }
+            // Preserved for the next connection attempt rather than lost:
+            // these only update local state, so there is no wire round trip
+            // to retry, just a value to carry forward.
+            FleetControlCommand::UpdateInventory(next) => *inventory = next,
+            FleetControlCommand::UpdateLoad(next) => *load = next,
+            FleetControlCommand::RegisterNode { .. }
+            | FleetControlCommand::Send(_)
+            | FleetControlCommand::HeartbeatNow => {}
+            FleetControlCommand::Shutdown => {}
+        }
+    }
+}
+
 async fn send_wire<S>(sink: &mut S, message: &BrokerToRelaycast) -> Result<()>
 where
     S: Sink<Message> + Unpin,