agentmux_srv/muxbus/

cloud_subscriber.rs

// Copyright 2026, AgentMux Corp.
// SPDX-License-Identifier: Apache-2.0

//! Cloud push subscription — replaces per-agent polling with a single
//! sidecar-level WebSocket to muxbus-ws.agentmux.ai.
//!
//! When MUXBUS_TOKEN is present, the subscriber:
//!   1. Opens wss://muxbus-ws.agentmux.ai with the bearer token.
//!   2. Listens for { type: "inject_available" } broadcast wake signals (zero metadata).
//!   3. On each signal, pulls pending injections for every locally-registered
//!      agent: one POST /reactive/pending-batch for all of them, or one
//!      GET /reactive/pending/:id per agent on a relay without the batch route.
//!   4. Claims all pending injections via REST POST /reactive/ack (atomic
//!      pending->delivered transition server-side) *before* delivering — only
//!      ids this call actually won the claim on get delivered via
//!      ReactiveHandler. A claimed injection that fails local delivery is
//!      released back to pending via REST POST /reactive/release so it's
//!      retried, rather than silently dropped.
//!   5. Reconnects with exponential back-off on any disconnect.
//!
//! The server broadcasts to ALL connected sidecars on every injection, so a subscriber
//! cannot correlate the signal to any particular agent, account, or timing.
//!
//! Agents register/unregister via add_agent()/remove_agent().
//! No MUXBUS_TOKEN → subscriber stays idle.

use std::collections::HashSet;
use std::sync::{Arc, Mutex, OnceLock};
use std::time::Duration;

use futures_util::{SinkExt, StreamExt};
use serde::{Deserialize, Serialize};
use tokio::sync::mpsc;
use tokio_tungstenite::{connect_async_tls_with_config, tungstenite::Message};

use crate::backend::reactive::handler::get_global_handler;
use crate::backend::reactive::types::InjectionRequest;
use crate::backend::storage::store::Store;
use crate::broker::RefreshErrorKind;
use crate::muxbus::delivery_status;

mod delivery;
use delivery::{claim_and_deliver, PendingResp};

// The relay's WebSocket endpoint has its own domain, and the handshake is at
// the domain root: no path suffix. The relay side is designed in the private
// cloud repo.
/// Fallback when the cloud publishes no settings (`discovery::ws_url`).
pub(crate) const MUXBUS_WS_URL: &str = "wss://muxbus-ws.agentmux.ai";
pub(crate) const MUXBUS_REST_URL: &str = "https://muxbus.agentmux.ai";
const RECONNECT_DELAY_SECS: u64 = 5;
const MAX_RECONNECT_DELAY_SECS: u64 = 60;
// How often a loop parked on a stale sign-in re-reads the stored credential
// (no network beyond the cached discovery document) in case another channel
// sharing the store signed in.
const STALE_RECHECK_SECS: u64 = 60;
// The relay drops a WebSocket that has been silent in both directions for
// 10 minutes and sends no keepalive of its own. Ping well under that so a
// quiet connection (no inject_available traffic) survives indefinitely. The
// ping is an app-level ClientMsg::Ping frame, not a WS-protocol-level
// Message::Ping — raw protocol ping/pong control frames are not reliably
// relayed, so a normal data frame is what actually keeps the connection
// alive in production (see ClientMsg::Ping's doc comment).
const CLIENT_PING_INTERVAL_SECS: u64 = 240;
// How often the broker's background sweep re-checks the stored MuxBus
// credential and proactively refreshes it if it's nearing expiry — now the
// SOLE proactive-refresh trigger (see `run_loop`'s registration call and the
// broker module doc); replaces what used to be two independent, uncoordinated
// checks (the outer reconnect loop's `load_valid_token`, and an inline check
// on every WS ping tick). Cheap (one DB read) when not actually stale, so a
// short interval is fine.
const BROKER_SWEEP_INTERVAL_SECS: u64 = 60;

// Max age (seconds) a reagent-signed WAN jekt's `reagent_ts_secs` may be
// from "now" and still verify — anti-replay, same purpose as
// server/reactive.rs's JEKT_SIG_MAX_AGE_SECS for host-tier (reagentx P1 on
// PR #2570). Wider than host-tier's 300s because this covers real network
// delivery latency, not a same-machine call — matches the review
// notifications' own delivery window.
const REAGENT_SIG_MAX_AGE_SECS: i64 = 600;

use agentmux_common::time::now_ms;
use agentmux_common::time::now_secs as now_unix_secs;

/// Is `ts_secs` (a reagent signature's claimed signing time) within
/// `REAGENT_SIG_MAX_AGE_SECS` of `now`? Extracted as a pure function (takes
/// `now` explicitly rather than calling `now_unix_secs()` itself) so it's
/// unit-testable without mocking the system clock. `ts_secs <= 0` is never
/// fresh — a zero/negative timestamp only happens for a malformed or
/// missing claim, never a genuine signature.
fn reagent_sig_is_fresh(ts_secs: i64, now: i64) -> bool {
    ts_secs > 0 && (now - ts_secs).abs() <= REAGENT_SIG_MAX_AGE_SECS
}

// ── Wire protocol ─────────────────────────────────────────────────────────────

#[derive(Serialize)]
#[serde(tag = "type", rename_all = "snake_case")]
enum ClientMsg {
    Subscribe { agents: Vec<String> },
    #[serde(rename = "subscribe:add")]
    SubscribeAdd { agents: Vec<String> },
    #[serde(rename = "subscribe:remove")]
    SubscribeRemove { agents: Vec<String> },
    // Ack removed — ACK is sent via REST POST /reactive/ack, not WS
    // App-level keepalive — see CLIENT_PING_INTERVAL_SECS. Raw WS-protocol
    // Ping/Pong control frames are not reliably relayed in production, so the
    // keepalive must be a normal data frame the relay can see and reply to.
    Ping,
}

#[derive(Deserialize)]
#[serde(tag = "type", rename_all = "snake_case")]
enum ServerMsg {
    // Zero-metadata wake signal. The sidecar polls all its registered agents on receipt.
    // No agent_id, no injection_id — subscribers gain no information from this frame.
    InjectAvailable,
    Subscribed {
        #[allow(dead_code)]
        agents: Vec<String>,
    },
    Pong,
    Error {
        message: String,
    },
    Evicted {
        #[allow(dead_code)]
        agents: Vec<String>,
    },
    #[serde(other)]
    Unknown,
}

// ── Control channel ───────────────────────────────────────────────────────────

enum CtrlMsg {
    AddAgent(String),
    RemoveAgent(String),
    ReloadToken,
}

// ── Public struct ─────────────────────────────────────────────────────────────

pub struct CloudSubscriber {
    ctrl_tx: mpsc::UnboundedSender<CtrlMsg>,
    /// In-memory copy of subscribed agents, kept in sync with the WS loop
    /// so add_agent/remove_agent can be called before the WS connects.
    agents: Arc<Mutex<HashSet<String>>>,
}

static GLOBAL_SUBSCRIBER: OnceLock<CloudSubscriber> = OnceLock::new();

pub fn get_global_subscriber() -> Option<&'static CloudSubscriber> {
    GLOBAL_SUBSCRIBER.get()
}

impl CloudSubscriber {
    /// Initialize the global subscriber and start the background WS loop.
    /// Should be called once at startup. No-op if already initialized.
    pub fn init_global(mstore: Arc<Store>) {
        let (ctrl_tx, ctrl_rx) = mpsc::unbounded_channel::<CtrlMsg>();
        let agents = Arc::new(Mutex::new(HashSet::<String>::new()));
        // Agents that registered before this subscriber existed (an isolated
        // channel creates it lazily at `muxbus.login`) were never told to it.
        let handler = get_global_handler();
        let registered: Vec<String> = handler
            .list_agents()
            .into_iter()
            .map(|r| r.agent_id)
            .collect();
        let seed = agents_to_seed(&registered, |a| handler.has_live_name(a));
        if !seed.is_empty() {
            tracing::info!(
                count = seed.len(),
                "cloud_subscriber: subscribing agents that registered before it started"
            );
            agents.lock().unwrap().extend(seed);
        }
        let subscriber = CloudSubscriber {
            ctrl_tx,
            agents: agents.clone(),
        };
        if GLOBAL_SUBSCRIBER.set(subscriber).is_err() {
            return; // already initialized
        }
        tokio::spawn(run_loop(mstore, agents, ctrl_rx));
    }

    /// Notify the WS loop that a new agent is registered locally.
    /// Normalizes to lowercase and skips the WS send if already subscribed,
    /// so calling this on every COMMAND_AGENT_INPUT turn is safe.
    pub fn add_agent(&self, agent_id: &str) {
        let key = agent_id.to_lowercase();
        let mut agents = self.agents.lock().unwrap();
        if agents.contains(&key) {
            return;
        }
        agents.insert(key.clone());
        drop(agents);
        let _ = self.ctrl_tx.send(CtrlMsg::AddAgent(key));
    }

    /// Notify the WS loop that an agent has been unregistered.
    pub fn remove_agent(&self, agent_id: &str) {
        let key = agent_id.to_lowercase();
        self.agents.lock().unwrap().remove(&key);
        let _ = self.ctrl_tx.send(CtrlMsg::RemoveAgent(key));
    }

    /// Unsubscribe `agent_id` if no live block holds it (`is_live`), and
    /// say whether it did. The check, the removal and the `RemoveAgent`
    /// send all happen under the `agents` lock, which `add_agent` also
    /// takes: a respawn either registered first (the check sees it live and
    /// nothing is dropped) or its `add_agent` waits, then finds the key gone
    /// and queues `AddAgent` after this `RemoveAgent` — the WS loop applies
    /// them in order and it ends subscribed (Codex P2 ×2 on #3897).
    /// `is_live` must not take the `agents` lock; it may take the reactive
    /// handler's (which never takes this one).
    pub fn drop_if_orphaned(&self, agent_id: &str, is_live: impl Fn(&str) -> bool) -> bool {
        let key = agent_id.to_lowercase();
        let mut agents = self.agents.lock().unwrap();
        if !agents.contains(&key) || is_live(&key) {
            return false;
        }
        agents.remove(&key);
        let _ = self.ctrl_tx.send(CtrlMsg::RemoveAgent(key));
        true
    }

    /// [`release_agent_now`]'s body, with the relay and token injectable.
    /// The removal and its `RemoveAgent` are queued together under the
    /// `agents` lock, like [`Self::remove_agent`] — so an agent re-added
    /// while the release is awaited queues `AddAgent` after it and ends
    /// subscribed (ReAgent P1 on #3899). The awaited release then runs
    /// under `wan_lease`'s per-agent lock, serialized with the RemoveAgent
    /// arm's own: when this returns, whichever of the two ran has finished,
    /// and neither releases an agent that was subscribed again.
    async fn release_now(
        &self,
        agent_id: &str,
        base: &str,
        token: impl std::future::Future<Output = String>,
    ) -> RelayRelease {
        let key = agent_id.to_lowercase();
        let was_subscribed = {
            let mut agents = self.agents.lock().unwrap();
            let removed = agents.remove(&key);
            if removed {
                let _ = self.ctrl_tx.send(CtrlMsg::RemoveAgent(key.clone()));
            }
            removed
        };
        // Not subscribed is not "no lease" (Codex P1 on #3908): a stopped
        // pane's exit may already have unsubscribed it and queued a release
        // that runs detached. Release here too — idempotent, and under the
        // per-agent op lock, so this also waits for a release in flight.
        let held = super::wan_lease::is_held(&key);
        if !was_subscribed && !held {
            // Wait out a release already in flight; nothing is held here.
            super::wan_lease::release(base, &key, "", &reqwest::Client::new(), || true).await;
            return RelayRelease::NotSubscribed;
        }
        let token = token.await;
        let still_gone = || !self.agents.lock().unwrap().contains(&key);
        // Re-added meanwhile: then this srv kept the lease.
        if super::wan_lease::release(base, &key, &token, &reqwest::Client::new(), still_gone).await {
            RelayRelease::Released
        } else {
            RelayRelease::Kept
        }
    }

    /// Called after muxbus.login completes — trigger a fresh WS connection
    /// with the newly stored token.
    pub fn reload_token(&self) {
        let _ = self.ctrl_tx.send(CtrlMsg::ReloadToken);
    }

    /// Snapshot of the agents this sidecar has subscribed to the cloud relay
    /// (Tier-4), sorted for stable output. Empty when no MUXBUS_TOKEN is set
    /// (cloud disabled). Read-only; used by the discovery endpoint.
    pub fn subscribed_agents(&self) -> Vec<String> {
        let mut v: Vec<String> = self.agents.lock().unwrap().iter().cloned().collect();
        v.sort();
        v
    }
}

// ── Background loop ───────────────────────────────────────────────────────────

async fn run_loop(
    mstore: Arc<Store>,
    agents: Arc<Mutex<HashSet<String>>>,
    mut ctrl_rx: mpsc::UnboundedReceiver<CtrlMsg>,
) {
    let http = reqwest::Client::builder()
        .user_agent(super::relay::USER_AGENT)
        .build()
        .unwrap_or_default();
    let mut delay_secs = RECONNECT_DELAY_SECS;

    // Register this credential with the global broker scheduler once, up
    // front. The broker's own background sweep (spawned inside
    // `crate::broker::init_global`) now owns proactive refresh entirely —
    // it keeps the stored token fresh even while this WS session is
    // disconnected/backing off, which the old ping-tick-only check could not.
    let scheduler = crate::broker::init_global(Duration::from_secs(BROKER_SWEEP_INTERVAL_SECS));
    {
        let mstore_fresh = mstore.clone();
        let mstore_refresh = mstore.clone();
        let http_refresh = http.clone();
        scheduler
            .register(
                crate::muxbus::CREDENTIAL_ID,
                // muxbus_is_fresh, not muxbus_load — reagent P2 on #2260:
                // register()'s own contract requires is_fresh to be
                // side-effect-free (called under the per-id lock on every
                // ensure_fresh/sweep tick), but muxbus_load's lazy-migration
                // branch can perform a keychain write + SQL update for a
                // legacy row. muxbus_is_fresh shares the same read logic
                // with that branch disabled.
                //
                // spawn_blocking, not a direct call — reagent P1 on #2260:
                // muxbus_is_fresh does a synchronous OS-keychain read, which
                // can hang on a slow/unresponsive Secret Service D-Bus
                // daemon (a scenario headless Linux specifically has to
                // handle). Without this, that hang stalls the tokio worker
                // thread `run_sweep_loop` runs on, and via the scheduler's
                // held per-id lock, can block OTHER credentials' refreshes
                // too — the exact convention this codebase already applies
                // to keychain reads elsewhere (see
                // app_api/mod.rs's account_validate_impl).
                move || {
                    let mstore = mstore_fresh.clone();
                    Box::pin(async move {
                        tokio::task::spawn_blocking(move || mstore.muxbus_is_fresh())
                            .await
                            .unwrap_or(false)
                    })
                },
                move || {
                    let mstore = mstore_refresh.clone();
                    let http = http_refresh.clone();
                    Box::pin(async move {
                        // Held until this attempt ends, across the token request.
                        let lock_store = mstore.clone();
                        let _leader = tokio::task::spawn_blocking(move || lock_store.muxbus_refresh_lock())
                            .await
                            .map_err(|e| RefreshErrorKind::Transient(format!("refresh lock task: {e}")))?
                            .map_err(|e| RefreshErrorKind::Transient(e.to_string()))?;
                        let load_store = mstore.clone();
                        let current = tokio::task::spawn_blocking(move || load_store.muxbus_load())
                            .await
                            .map_err(|e| RefreshErrorKind::Transient(format!("muxbus_load task: {e}")))?
                            .map_err(|e| RefreshErrorKind::Transient(e.to_string()))?
                            // Nothing stored at all — not a transient storage
                            // blip, there's genuinely no credential to refresh;
                            // only a fresh login produces one.
                            .ok_or_else(|| {
                                RefreshErrorKind::PermanentAuthFailure(
                                    "no muxbus credentials stored".to_string(),
                                )
                            })?;
                        // Another process refreshed while this one waited.
                        if current.is_valid() && !current.nearly_expired() {
                            return Ok(());
                        }
                        if current.refresh_token.is_empty() {
                            // Same reasoning: no refresh_token will ever
                            // appear on its own — only re-login fixes it.
                            return Err(RefreshErrorKind::PermanentAuthFailure(
                                "no refresh_token stored".to_string(),
                            ));
                        }
                        // Issued by a client the cloud no longer publishes:
                        // its pool is gone, so don't refresh against it
                        // (SPEC_CLOUD_SETTINGS_DISCOVERY_2026_09_27.md §3.3).
                        let discovered = crate::muxbus::discovery::cloud_settings(&http).await;
                        if crate::muxbus::discovery::client_id_superseded(discovered.as_ref(), &current.client_id) {
                            return Err(RefreshErrorKind::PermanentAuthFailure(
                                "signed in to a user pool AgentMux Cloud no longer uses".to_string(),
                            ));
                        }
                        // Preserve-on-failure: `refresh_token` returning Err
                        // here means `muxbus_save` is simply never called —
                        // the last-known-good credential in the store is
                        // left untouched rather than overwritten with a
                        // failed/partial result.
                        let refreshed = crate::muxbus::pkce::refresh_token(&current, &http)
                            .await
                            .map_err(classify_refresh_token_error)?;
                        if !refreshed_same_account(&current, &refreshed) {
                            return Err(RefreshErrorKind::PermanentAuthFailure(
                                "the refreshed sign-in belongs to another account".to_string(),
                            ));
                        }
                        let save_store = mstore.clone();
                        tokio::task::spawn_blocking(move || save_store.muxbus_save(&refreshed))
                            .await
                            .map_err(|e| RefreshErrorKind::Transient(format!("muxbus_save task: {e}")))?
                            .map_err(|e| RefreshErrorKind::Transient(e.to_string()))
                    })
                },
            )
            .await;
    }
    // The presence publisher reads the sign-in through this registration. A
    // first login on a fresh channel signals it before the registration above
    // has run, so signal again now that it has (#4463).
    crate::muxbus::wan_presence::sign_in_changed();

    // The stale sign-in reason last logged, so a parked loop logs it once.
    let mut stale_logged: Option<String> = None;

    loop {
        // Load token — refresh if expired.
        // Two distinct None cases:
        //   a) No credentials in DB at all → wait for muxbus.login ReloadToken signal
        //   b) Credentials exist but refresh failed transiently → back off and retry
        // has_stored_creds = true only when a retry is meaningful:
        //   - access_token present AND still valid (use it now), OR
        //   - access_token expired but refresh_token present (can refresh).
        // An expired token with no refresh_token is a permanent failure → park like no-creds.
        //
        // A real load ERROR (keychain locked/unavailable, distinct from
        // muxbus_load's Ok(None) "genuinely nothing stored" — see that
        // function's own reagent-fixed error handling) must NOT collapse to
        // has_stored_creds=false: that would park indefinitely waiting for a
        // muxbus.login the user was never actually missing, instead of
        // backing off and retrying once the transient failure clears.
        // spawn_blocking — reagent P1 on #2260: muxbus_load does a
        // synchronous OS-keychain read; a hang there (slow/unresponsive
        // Secret Service D-Bus daemon) must not stall this loop's tokio
        // worker thread.
        let has_stored_creds_load = {
            let mstore = mstore.clone();
            tokio::task::spawn_blocking(move || mstore.muxbus_load()).await
        };
        let nothing_stored = matches!(has_stored_creds_load, Ok(Ok(None)));
        let (has_stored_creds, stale) = match has_stored_creds_load {
            Ok(Ok(Some(c))) => {
                // A torn sign-in keeps only its refresh token; it is still a
                // sign-in this channel had (and repairs on refresh).
                if !c.access_token.is_empty() || !c.refresh_token.is_empty() {
                    delivery_status::saw_session(Some(c.user_email.clone()));
                }
                (usable_or_refreshable(&c), crate::muxbus::stale_sign_in(&c, &http).await)
            }
            Ok(Ok(None)) => (false, None),
            Ok(Err(e)) => {
                tracing::warn!(error = %e, "cloud_subscriber: muxbus_load failed — assuming credentials exist and retrying with backoff");
                delivery_status::saw_session(None);
                (true, None)
            }
            Err(e) => {
                tracing::warn!(error = %e, "cloud_subscriber: muxbus_load task panicked — assuming credentials exist and retrying with backoff");
                delivery_status::saw_session(None);
                (true, None)
            }
        };
        // A sign-in that can't work anymore: stop reconnecting and wait for a
        // new one (SPEC_CLOUD_SETTINGS_DISCOVERY_2026_09_27.md §3.3).
        // `muxbus.status` reports it to the UI as "Sign in again". Rechecked
        // now and then from local state only, for a sign-in made by another
        // channel sharing the store (which sends no ReloadToken here).
        if let Some(reason) = stale {
            delivery_status::set_link(delivery_status::Link::SignInStale(reason.clone()));
            if stale_logged.as_deref() != Some(reason.as_str()) {
                tracing::warn!(reason = %reason, "cloud_subscriber: AgentMux Cloud sign-in needs renewing — not reconnecting until the user signs in again");
                stale_logged = Some(reason);
            }
            wait_for_sign_in(&mut ctrl_rx, Duration::from_secs(STALE_RECHECK_SECS)).await;
            delay_secs = RECONNECT_DELAY_SECS;
            continue;
        }
        stale_logged = None;
        // Nothing stored: wait for a sign-in. Rechecked now and then, since a
        // sign-in by another process of this channel sends no ReloadToken.
        if nothing_stored {
            delivery_status::set_link(delivery_status::Link::NoSignIn);
            wait_for_sign_in(&mut ctrl_rx, Duration::from_secs(STALE_RECHECK_SECS)).await;
            delay_secs = RECONNECT_DELAY_SECS;
            continue;
        }
        let token = match load_valid_token(&mstore, &scheduler).await {
            Some(t) => t,
            None if !has_stored_creds => {
                // Nothing usable: the broker now knows (NeedsReauth), and the
                // stale check above parks the next pass until a sign-in.
                delivery_status::set_link(delivery_status::Link::NoSignIn);
                wait_for_sign_in(&mut ctrl_rx, Duration::from_secs(STALE_RECHECK_SECS)).await;
                delay_secs = RECONNECT_DELAY_SECS;
                continue;
            }
            None => {
                // Credentials present but refresh failed transiently — back off and retry.
                // Loop inside the select! so AddAgent/RemoveAgent messages drain without
                // cancelling the sleep; only ReloadToken (or sleep expiry) breaks out.
                tracing::warn!(
                    "cloud_subscriber: token refresh failed, retrying in {}s",
                    delay_secs
                );
                delivery_status::set_link(delivery_status::Link::Unreachable(
                    "couldn't refresh the sign-in".to_string(),
                ));
                let sleep = tokio::time::sleep(Duration::from_secs(delay_secs));
                tokio::pin!(sleep);
                'backoff: loop {
                    tokio::select! {
                        _ = &mut sleep => break 'backoff,
                        Some(msg) = ctrl_rx.recv() => match msg {
                            CtrlMsg::ReloadToken => {
                                delay_secs = RECONNECT_DELAY_SECS;
                                break 'backoff;
                            }
                            CtrlMsg::AddAgent(_) | CtrlMsg::RemoveAgent(_) => {
                                // agents mutex already updated; continue sleeping
                            }
                        }
                    }
                }
                delay_secs = (delay_secs * 2).min(MAX_RECONNECT_DELAY_SECS);
                continue;
            }
        };

        let session_start = std::time::Instant::now();

        match connect_and_run(&token, agents.clone(), &mut ctrl_rx, &mstore, &http).await {
            Ok(()) => {
                // Apply the same >30s healthy-session guard as the error branch: a server
                // that immediately closes after handshake must not suppress back-off.
                if session_start.elapsed().as_secs() > 30 {
                    delay_secs = RECONNECT_DELAY_SECS;
                }
                tracing::info!("cloud_subscriber: disconnected cleanly");
                delivery_status::set_link(delivery_status::Link::Unreachable("disconnected".to_string()));
            }
            Err(e) => {
                // If the session was healthy for >30s before the error, treat it as
                // network instability rather than a persistent failure and reset back-off.
                if session_start.elapsed().as_secs() > 30 {
                    delay_secs = RECONNECT_DELAY_SECS;
                }
                tracing::warn!("cloud_subscriber: error: {e}, reconnecting in {}s", delay_secs);
                delivery_status::set_link(delivery_status::Link::Unreachable(e));
            }
        }

        // Post-disconnect back-off. Loop inside the select! so AddAgent/RemoveAgent
        // messages drain without cancelling the sleep; only ReloadToken breaks out early.
        let sleep = tokio::time::sleep(Duration::from_secs(delay_secs));
        tokio::pin!(sleep);
        'reconnect: loop {
            tokio::select! {
                _ = &mut sleep => break 'reconnect,
                Some(msg) = ctrl_rx.recv() => match msg {
                    CtrlMsg::ReloadToken => {
                        delay_secs = RECONNECT_DELAY_SECS;
                        break 'reconnect;
                    }
                    CtrlMsg::AddAgent(_) | CtrlMsg::RemoveAgent(_) => {
                        // agents mutex already updated; continue sleeping
                    }
                }
            }
        }
        delay_secs = (delay_secs * 2).min(MAX_RECONNECT_DELAY_SECS);
    }
}

/// A loaded sign-in that can be used now or refreshed: a valid access token,
/// or a refresh token (all a torn sign-in keeps, see `storage::muxbus`).
fn usable_or_refreshable(c: &crate::muxbus::MuxBusCredentials) -> bool {
    c.is_valid() || !c.refresh_token.is_empty()
}

/// Whether a refresh came back for the account the saved sign-in names. A
/// torn set could pair a refresh token with another account's row; a row
/// with no account recorded (an older sign-in) can't be checked.
fn refreshed_same_account(current: &crate::muxbus::MuxBusCredentials, refreshed: &crate::muxbus::MuxBusCredentials) -> bool {
    if current.user_sub.is_empty() {
        return true;
    }
    let sub = crate::muxbus::pkce::token_sub(&refreshed.id_token);
    let sub = if sub.is_empty() { crate::muxbus::pkce::token_sub(&refreshed.access_token) } else { sub };
    sub == current.user_sub
}

/// Park until a sign-in (`ReloadToken`) or `recheck` elapses, whichever comes
/// first. AddAgent/RemoveAgent drain without waking it: they already updated
/// the `agents` set the next connection subscribes from.
async fn wait_for_sign_in(ctrl_rx: &mut mpsc::UnboundedReceiver<CtrlMsg>, recheck: Duration) {
    let sleep = tokio::time::sleep(recheck);
    tokio::pin!(sleep);
    loop {
        tokio::select! {
            _ = &mut sleep => return,
            msg = ctrl_rx.recv() => match msg {
                Some(CtrlMsg::ReloadToken) => return,
                Some(CtrlMsg::AddAgent(_) | CtrlMsg::RemoveAgent(_)) => {}
                // Closed: nothing can wake us early; wait out the recheck.
                None => {
                    (&mut sleep).await;
                    return;
                }
            }
        }
    }
}

async fn connect_and_run(
    token: &str,
    agents: Arc<Mutex<HashSet<String>>>,
    ctrl_rx: &mut mpsc::UnboundedReceiver<CtrlMsg>,
    mstore: &Arc<Store>,
    http: &reqwest::Client,
) -> Result<(), String> {
    use tokio_tungstenite::tungstenite::ClientRequestBuilder;

    // Where the REST calls of this session go: the relay's pending pulls,
    // acks and releases, and the WAN agent lease. Resolved once per session.
    let base = super::relay::rest_base_url();
    let base = base.as_str();

    // Issue #2091: a hand-built `http::Request` (as this used to be, via
    // `Request::builder()...body(())`) is NOT auto-completed by tungstenite.
    // `IntoClientRequest for Request` is a bare passthrough (`Ok(self)`) —
    // the Host/Connection/Upgrade/Sec-WebSocket-Version/Sec-WebSocket-Key
    // headers this comment used to claim were "added automatically" are only
    // generated when tungstenite builds the request itself from a bare
    // `Uri`/`&str`/`String` (`impl IntoClientRequest for Uri`, which calls
    // `generate_key()`). A hand-built Request with only `Authorization` set
    // therefore had no `Sec-WebSocket-Key` at all, and tungstenite's own
    // `generate_request()` requires it to already be present — every
    // connection attempt failed handshake validation on its OWN outgoing
    // request before a single byte reached the network, logged as "Missing,
    // duplicated or incorrect header sec-websocket-key". This has been
    // broken since the file's first commit; reproduced in isolation
    // (a `ws_probe` example, removed after confirming the fix) and
    // confirmed fixed by switching to `ClientRequestBuilder` — tungstenite's
    // own purpose-built API for "URI + extra headers": it builds the request
    // from the `Uri` (generating the required headers, including a fresh
    // `Sec-WebSocket-Key`) and layers `with_header` calls on top, so nothing
    // required is ever missing.
    // The cloud's published WebSocket URL, else the compiled one
    // (docs/specs/SPEC_CLOUD_SETTINGS_DISCOVERY_2026_09_27.md).
    let ws_url = super::discovery::ws_url(http).await;
    tracing::info!("cloud_subscriber: connecting to {}", ws_url);
    let uri: tokio_tungstenite::tungstenite::http::Uri =
        ws_url.parse().map_err(|e| format!("parse url: {e}"))?;
    let request = ClientRequestBuilder::new(uri)
        .with_header("Authorization", format!("Bearer {}", token));

    let (ws_stream, _) = connect_async_tls_with_config(request, None, false, None)
        .await
        .map_err(|e| format!("connect: {e}"))?;

    tracing::info!("cloud_subscriber: WebSocket connected");
    delivery_status::set_link(delivery_status::Link::Connected);
    let (mut write, mut read) = ws_stream.split();

    // Initial subscription for all currently-registered agents
    let initial_agents: Vec<String> = agents.lock().unwrap().iter().cloned().collect();
    let sub_msg = serde_json::to_string(&ClientMsg::Subscribe { agents: initial_agents })
        .map_err(|e| format!("serialize subscribe: {e}"))?;
    write.send(Message::Text(sub_msg.into()))
        .await
        .map_err(|e| format!("send subscribe: {e}"))?;

    // Pull anything that reached the relay before this connection existed, or
    // before its agents were subscribed: the relay only wakes us for what
    // arrives from now on. A wake is a best-effort broadcast, and this loop
    // pulled only on one (CATCH_UP_EVERY, `agents_to_seed`).
    if catch_up(base, token, http, &agents, mstore).await {
        return Ok(());
    }

    let mut ping_interval = tokio::time::interval(Duration::from_secs(CLIENT_PING_INTERVAL_SECS));
    // Keeps every registered agent's relay lease renewed between wake
    // signals (the relay's lease TTL is 60 s; wakes can be far apart).
    let mut lease_interval = tokio::time::interval(super::wan_lease::RENEW_EVERY);
    lease_interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
    ping_interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
    ping_interval.tick().await; // first tick fires immediately — consume it, we just connected
    // The safety net under the wake signal: pull on a timer too.
    let mut catch_up_interval = tokio::time::interval(CATCH_UP_EVERY);
    catch_up_interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
    catch_up_interval.tick().await; // the connect-time pull above was the first

    loop {
        tokio::select! {
            // Keepalive — see CLIENT_PING_INTERVAL_SECS. Proactive credential
            // freshness is no longer piggybacked on this tick: the broker's
            // own background sweep (registered once in `run_loop`, see
            // `crate::broker`) now keeps `db_muxbus_credentials` fresh on its
            // own schedule regardless of whether a WS session happens to be
            // open, so a long-lived session no longer needs to special-case
            // its own credential check here.
            _ = lease_interval.tick() => {
                let mut registered: Vec<String> = agents.lock().unwrap().iter().cloned().collect();
                // Never renew a lease for an agent no live block holds: drop
                // the leaked subscription instead. A drop goes through the
                // RemoveAgent arm below, which unsubscribes and releases the
                // lease; `drop_if_orphaned` re-checks liveness so a respawn
                // registering meanwhile keeps its subscription.
                let handler = get_global_handler();
                let orphans = orphaned_subscriptions(&registered, |a| handler.has_live_name(a));
                if !orphans.is_empty() {
                    if let Some(sub) = get_global_subscriber() {
                        for agent_id in &orphans {
                            if sub.drop_if_orphaned(agent_id, |a| handler.has_live_name(a)) {
                                tracing::warn!(
                                    agent = %agent_id,
                                    "cloud_subscriber: subscribed agent has no live pane — dropped it and released its WAN lease"
                                );
                            }
                        }
                    }
                    registered.retain(|a| !orphans.contains(a));
                }
                let subscribed = &agents;
                // One request renews every held lease that is due
                // (`renew_held_batch`); anything it leaves out, and every
                // agent on a relay without the batch route, goes one by one.
                let shared = shared_token_now(mstore, token).await;
                let batched = super::wan_lease::renew_held_batch(base, &registered, &shared, http, |a| {
                    subscribed.lock().unwrap().iter().any(|s| s.eq_ignore_ascii_case(a))
                })
                .await;
                for (agent_id, outcome) in &batched {
                    if let super::wan_lease::Outcome::HeldElsewhere(where_) = outcome {
                        crate::backend::agent_admission::fence_agent_named(agent_id, where_);
                    }
                }
                registered.retain(|a| !batched.contains_key(&a.trim().to_lowercase()));
                futures_util::future::join_all(registered.iter().map(|agent_id| async move {
                    let per_agent = crate::muxbus::agent_credentials::ensure_agent_credential(agent_id, mstore, http).await;
                    let agent_token = match per_agent {
                        Some(t) => t,
                        None => shared_token_now(mstore, token).await,
                    };
                    // Re-checked under the lease's per-agent lock: a Take over
                    // may have released it since this list was copied.
                    let still_subscribed = || subscribed.lock().unwrap().contains(agent_id);
                    if let super::wan_lease::Outcome::HeldElsewhere(where_) =
                        super::wan_lease::ensure(base, agent_id, &agent_token, http, still_subscribed).await
                    {
                        crate::backend::agent_admission::fence_agent_named(agent_id, &where_);
                    }
                }))
                .await;
            }

            _ = catch_up_interval.tick() => {
                if catch_up(base, token, http, &agents, mstore).await {
                    return Ok(());
                }
            }

            _ = ping_interval.tick() => {
                let ping_msg = serde_json::to_string(&ClientMsg::Ping)
                    .map_err(|e| format!("serialize ping: {e}"))?;
                if let Err(e) = write.send(Message::Text(ping_msg.into())).await {
                    return Err(format!("send ping: {e}"));
                }
            }

            // Incoming WebSocket message from cloud
            msg = read.next() => {
                match msg {
                    None => return Ok(()), // server closed
                    Some(Err(e)) => return Err(format!("ws recv: {e}")),
                    Some(Ok(Message::Text(text))) => {
                        if let Ok(server_msg) = serde_json::from_str::<ServerMsg>(&text) {
                            match handle_server_msg(server_msg, base, token, http, &agents, mstore).await {
                                Ok(()) => {}
                                // Eviction or expired token — close stream to trigger reconnect.
                                Err(ref e) if e.starts_with("reconnect:") => {
                                    tracing::info!("cloud_subscriber: {e}, reconnecting");
                                    return Ok(());
                                }
                                Err(e) => {
                                    tracing::warn!("cloud_subscriber: handle msg error: {e}");
                                }
                            }
                        }
                    }
                    Some(Ok(Message::Ping(data))) => {
                        let _ = write.send(Message::Pong(data)).await;
                    }
                    Some(Ok(Message::Close(_))) => return Ok(()),
                    Some(Ok(_)) => {} // binary frames etc — ignore
                }
            }

            // Control messages from register/unregister calls
            ctrl = ctrl_rx.recv() => {
                match ctrl {
                    None => return Ok(()), // channel closed
                    Some(CtrlMsg::AddAgent(id)) => {
                        let msg = serde_json::to_string(&ClientMsg::SubscribeAdd { agents: vec![id] })
                            .unwrap_or_default();
                        let _ = write.send(Message::Text(msg.into())).await;
                        // A message may already be waiting for this agent (it
                        // reached the relay before the agent was subscribed).
                        if catch_up(base, token, http, &agents, mstore).await {
                            return Ok(());
                        }
                    }
                    Some(CtrlMsg::RemoveAgent(id)) => {
                        // The agent went away here: let another instance have it.
                        {
                            let (id, base, token, http, mstore, agents) =
                                (id.clone(), base.to_string(), token.to_string(), http.clone(), mstore.clone(), agents.clone());
                            tokio::spawn(async move {
                                let token = shared_token_now(&mstore, &token).await;
                                // Skip if it was subscribed again meanwhile: its
                                // fresh claim must not be released (Codex P2, #3897).
                                let still_gone = || !agents.lock().unwrap().contains(&id);
                                super::wan_lease::release(&base, &id, &token, &http, still_gone).await
                            });
                        }
                        let msg = serde_json::to_string(&ClientMsg::SubscribeRemove { agents: vec![id] })
                            .unwrap_or_default();
                        let _ = write.send(Message::Text(msg.into())).await;
                    }
                    Some(CtrlMsg::ReloadToken) => {
                        // Close and let the outer loop reconnect with the new token
                        let _ = write.send(Message::Close(None)).await;
                        return Ok(());
                    }
                }
            }
        }
    }
}

/// How often the subscriber pulls for every registered agent whether or not a
/// wake signal arrived. The relay's wake is best-effort (a lost one is
/// meant to be caught by the client's own poll), but this client had no
/// poll interval, so a jekt that reached the relay before its agent was
/// subscribed waited for the next unrelated wake: 8m46s on 2026-09-30. The
/// lease is already renewed every [`super::wan_lease::RENEW_EVERY`], so a
/// catch-up costs about one pending GET per agent: roughly 0.5 requests a
/// minute per agent at this interval.
const CATCH_UP_EVERY: Duration = Duration::from_secs(120);

/// The agents to put in the subscription when it is created: every locally
/// registered agent that a live pane still holds, lowercased and deduplicated.
/// Registered agents used to be missed. On an isolated channel the subscriber
/// is created lazily at `muxbus.login`, with an empty set, and an agent that
/// registered earlier (while there was no subscriber to tell) joined only on
/// its next input turn, so nothing reached it until then.
fn agents_to_seed(registered: &[String], is_live: impl Fn(&str) -> bool) -> Vec<String> {
    let mut seed: Vec<String> = registered
        .iter()
        .map(|a| a.to_lowercase())
        .filter(|a| is_live(a))
        .collect();
    seed.sort();
    seed.dedup();
    seed
}

/// Pull pending injections for every registered agent now. The same work an
/// `InjectAvailable` wake does, run on connect, when an agent is added, and on
/// a timer ([`CATCH_UP_EVERY`]) so a missed wake costs minutes at worst.
/// Returns true when the session should reconnect (the shared token expired).
async fn catch_up(
    base: &str,
    token: &str,
    http: &reqwest::Client,
    agents: &Arc<Mutex<HashSet<String>>>,
    mstore: &Arc<Store>,
) -> bool {
    match handle_server_msg(ServerMsg::InjectAvailable, base, token, http, agents, mstore).await {
        Ok(()) => false,
        Err(e) if e.starts_with("reconnect:") => {
            tracing::info!("cloud_subscriber: {e}, reconnecting");
            true
        }
        Err(e) => {
            tracing::warn!("cloud_subscriber: catch-up pull error: {e}");
            false
        }
    }
}

/// Handle a message from the cloud server.
/// The WS wake signal carries zero metadata — the sidecar polls all its registered
/// agents via REST to find pending injections, so a compromised subscriber gains nothing.
async fn handle_server_msg(
    msg: ServerMsg,
    base: &str,
    token: &str,
    http: &reqwest::Client,
    agents: &Arc<Mutex<HashSet<String>>>,
    mstore: &Arc<Store>,
) -> Result<(), String> {
    match msg {
        ServerMsg::InjectAvailable => {
            // Collect currently-registered agents without holding the lock across awaits.
            let registered: Vec<String> = agents.lock().unwrap().iter().cloned().collect();
            let handler = get_global_handler();

            // Concurrent, not sequential: each agent's sync can incur up to
            // two CREDENTIAL_HTTP_TIMEOUT-bounded HTTP calls (credential
            // fetch + pending fetch, worst case an unrevoked-then-retried
            // claim too) — sequentially awaiting N agents compounds that
            // into an N-times-longer block of this whole select! loop
            // (pings, incoming messages, ctrl signals all wait on it).
            // join_all bounds the wall-clock cost to the slowest single
            // agent's chain regardless of N. reagentx P1 (round 4) on
            // PR #2342.
            // One request pulls for every agent (`sync_agents_batched`); a
            // relay without the batch route, or a single agent, is pulled one
            // agent at a time as before.
            let outcomes = match sync_agents_batched(base, &registered, token, http, mstore, handler).await {
                Some(outcomes) => outcomes,
                None => {
                    futures_util::future::join_all(
                        registered
                            .iter()
                            .map(|agent_id| sync_agent_reactive(base, agent_id, token, http, mstore, handler)),
                    )
                    .await
                }
            };

            if outcomes
                .iter()
                .any(|o| matches!(o, AgentSyncOutcome::ReconnectSharedTokenExpired))
            {
                // Shared account-level token expired during session — reconnect to refresh.
                return Err("reconnect:token_expired".to_string());
            }
        }
        ServerMsg::Error { message } => {
            tracing::warn!("cloud_subscriber: server error: {message}");
        }
        ServerMsg::Evicted { .. } => {
            // Another sidecar evicted us. Reconnect to re-subscribe (reclaim).
            return Err("reconnect:evicted".to_string());
        }
        ServerMsg::Pong | ServerMsg::Subscribed { .. } | ServerMsg::Unknown => {}
    }
    Ok(())
}

enum AgentSyncOutcome {
    /// This agent's sync finished (successfully or with an already-logged,
    /// non-fatal error) — nothing further to signal to the caller.
    Ok,
    /// The SHARED account-level token (not a per-agent credential) was
    /// rejected with a genuinely recoverable-by-reconnect status (401,
    /// expired) — the whole WS session's auth is stale and a reconnect
    /// gives `load_valid_token` a chance to mint a fresh token via
    /// `ensure_fresh` before retrying. Only ever produced when no
    /// per-agent credential was in play for the rejected call, and NOT
    /// produced for a 403 on the shared token — see
    /// `shared_token_rejection_outcome`'s doc comment for why a binding
    /// mismatch (403) can't be fixed by reconnecting and must not tear
    /// down the whole session.
    ReconnectSharedTokenExpired,
}

/// Does `status` mean "this credential is not accepted for this request,"
/// regardless of the specific reason? 401 = expired/invalid token. 403 =
/// the cloud's agent-binding check rejected it (this credential is bound to a
/// different agent_id than the one it's being used for). Both mean the
/// SAME thing operationally for a poller CURRENTLY USING A PER-AGENT
/// CREDENTIAL: that credential can't be used for this agent, so
/// invalidate-and-retry-with-the-shared-token applies identically to either
/// status. Before this, only 401 triggered that recovery path — a 403 fell
/// into the generic non-2xx branch, silently stalling that agent's WAN
/// delivery for the rest of the poll cycle with no retry and no
/// user-visible error. Once the retry has fallen back to the SHARED token
/// itself, 401 and 403 diverge again — see `shared_token_rejection_outcome`,
/// which is what actually decides that case, not this function.
fn is_credential_rejected(status: reqwest::StatusCode) -> bool {
    status == reqwest::StatusCode::UNAUTHORIZED || status == reqwest::StatusCode::FORBIDDEN
}

/// What to do when the SHARED account-level token itself (not a per-agent
/// credential — that case is already handled by the invalidate-and-retry
/// branch above) is rejected. 401 and 403 are NOT interchangeable here even
/// though both satisfy `is_credential_rejected`:
///
/// - 401 (expired) → `ReconnectSharedTokenExpired`. Reconnecting is
///   productive: the outer loop's `load_valid_token` calls
///   `RefreshScheduler::ensure_fresh` first, which can mint a genuinely new,
///   accepted access token from the refresh_token before retrying.
/// - 403 (the agent-binding check rejected this account for `agent_id`) →
///   `AgentSyncOutcome::Ok` (skip this agent this cycle, log a warning).
///   `MuxBusCredentials::is_valid()` is purely expiry-based, so the still
///   time-valid access token survives `load_valid_token` unchanged after a
///   reconnect — and even a genuine refresh would mint a new token for the
///   same binding, so nothing about tearing down and reconnecting the WS
///   session can fix a binding mismatch. Treating it like a 401 anyway
///   (the original extension of this function) churned the whole shared
///   session through repeated reconnects with the identical rejected token,
///   starving `InjectAvailable` delivery for every OTHER registered agent
///   too — reagentx P1 on PR #2573. Matches this project's pre-403-handling
///   behavior for exactly this case (log-only, no reconnect); only the
///   per-agent-credential branch above actually gained 403 recovery.
fn shared_token_rejection_outcome(status: reqwest::StatusCode, agent_id: &str) -> AgentSyncOutcome {
    if status == reqwest::StatusCode::FORBIDDEN {
        tracing::warn!(
            agent_id = %agent_id,
            "cloud_subscriber: shared token rejected (binding mismatch) for this agent — skipping this cycle, not reconnecting"
        );
        // Not reconnecting, but this agent's messages aren't arriving either.
        delivery_status::pull_failed(agent_id, "the relay refused this agent's messages (403)");
        return AgentSyncOutcome::Ok;
    }
    AgentSyncOutcome::ReconnectSharedTokenExpired
}

/// One registered agent's full pending-fetch → claim → deliver cycle for an
/// `InjectAvailable` broadcast. Extracted from the loop body it used to be
/// so `handle_server_msg` can run every agent's sync concurrently via
/// `join_all` instead of sequentially — see the call site's own comment.
///
/// Retries once with the shared `token` if a per-agent credential is
/// rejected (401 or 403 — see `is_credential_rejected`) — without this,
/// invalidating the credential alone left this agent's pending injection
/// undelivered until an unrelated InjectAvailable broadcast happened to
/// fire again (no periodic resync exists). reagentx P1 (round 4) on
/// PR #2342; 403 coverage added later (without it, a genuine binding
/// mismatch degrades to a silent stall instead of falling back the same
/// way an expired token already does).
///
/// `base` is the relay's REST base URL ([`super::relay::rest_base_url`]); a
/// parameter so a test can point one sync at a fake relay.
async fn sync_agent_reactive(
    base: &str,
    agent_id: &str,
    token: &str,
    http: &reqwest::Client,
    mstore: &Arc<Store>,
    handler: &'static crate::backend::reactive::handler::ReactiveHandler,
) -> AgentSyncOutcome {
    // Prefer a credential bound to exactly this agent_id over the shared
    // account-level token — see agent_credentials.rs. Falls back to `token`
    // (today's self-declared behavior) whenever this agent isn't
    // provisioned yet or provisioning fails, so rollout never blocks
    // delivery. `using_per_agent` tracks which one is currently in play so
    // a 401 can be attributed correctly and, for a per-agent credential,
    // retried once with the shared token instead of just giving up.
    let per_agent_token = crate::muxbus::agent_credentials::ensure_agent_credential(agent_id, mstore, http).await;
    let mut agent_token = match per_agent_token.clone() {
        Some(t) => t,
        None => shared_token_now(mstore, token).await,
    };
    let mut using_per_agent = per_agent_token.is_some();

    // One live instance per agent across the WAN
    // (SPEC_AGENT_SINGLE_LIVE_INSTANCE_2026_09_24 §4.5): claim or renew this
    // agent's relay lease before pulling. If another instance holds it, do
    // not take its jekts — and fence the local holder of this agent, if any
    // (the newcomer yields). Fails open on any error (wan_lease::Outcome::Unknown).
    if let super::wan_lease::Outcome::HeldElsewhere(where_) =
        super::wan_lease::ensure(base, agent_id, &agent_token, http, || is_subscribed(agent_id)).await
    {
        crate::backend::agent_admission::fence_agent_named(agent_id, &where_);
        return AgentSyncOutcome::Ok;
    }

    let url = format!("{}/reactive/pending/{}", base, agent_id);
    let body: PendingResp = loop {
        // For telling a 409 from before a Take over apart (note_not_holder_since).
        let resp_started = std::time::Instant::now();
        let resp = match http
            .get(&url)
            .header("Authorization", format!("Bearer {}", agent_token))
            .header("X-Agent-ID", agent_id)
            .header("X-Agent-Instance", super::wan_lease::instance_id())
            .send()
            .await
        {
            Ok(r) => r,
            Err(e) => {
                tracing::warn!(agent_id = %agent_id, error = %e, "cloud_subscriber: fetch pending failed");
                delivery_status::pull_failed(agent_id, "couldn't fetch messages from the relay");
                return AgentSyncOutcome::Ok;
            }
        };

        if is_credential_rejected(resp.status()) {
            if using_per_agent {
                // The per-agent credential's cached token is stale/revoked,
                // or the credential itself is binding-mismatched server-side
                // even though it looked locally valid — invalidate it (so
                // the next agent_credentials call re-fetches instead of
                // retrying the same rejected token) and retry THIS fetch
                // immediately with the shared token, rather than tearing
                // down the whole shared session (which would starve
                // delivery for every OTHER agent too). 403 additionally
                // records a cooldown — see
                // invalidate_binding_mismatched_credential's doc comment for
                // why a plain token-clear isn't enough for that status.
                if resp.status() == reqwest::StatusCode::FORBIDDEN {
                    crate::muxbus::agent_credentials::invalidate_binding_mismatched_credential(agent_id, mstore);
                } else {
                    crate::muxbus::agent_credentials::invalidate_cached_token(agent_id, mstore);
                }
                tracing::warn!(
                    agent_id = %agent_id,
                    status = %resp.status(),
                    "cloud_subscriber: per-agent credential rejected — invalidated, retrying with shared token",
                );
                agent_token = shared_token_now(mstore, token).await;
                using_per_agent = false;
                continue;
            }
            return shared_token_rejection_outcome(resp.status(), agent_id);
        }
        if resp.status() == reqwest::StatusCode::CONFLICT {
            // The relay's lease fence: another instance holds this agent.
            let body = resp.json::<serde_json::Value>().await.unwrap_or_default();
            if let Some(where_) = super::wan_lease::note_not_holder_since(agent_id, &body, resp_started) {
                crate::backend::agent_admission::fence_agent_named(agent_id, &where_);
            }
            return AgentSyncOutcome::Ok;
        }
        if !resp.status().is_success() {
            tracing::warn!(
                status = %resp.status(),
                agent_id = %agent_id,
                "cloud_subscriber: fetch pending non-2xx"
            );
            delivery_status::pull_failed(agent_id, format!("the relay answered {} to a message fetch", resp.status().as_u16()));
            return AgentSyncOutcome::Ok;
        }

        break match resp.json().await {
            Ok(b) => b,
            Err(e) => {
                tracing::warn!(error = %e, "cloud_subscriber: parse pending failed");
                delivery_status::pull_failed(agent_id, "the relay's message list couldn't be read");
                return AgentSyncOutcome::Ok;
            }
        };
    };

    claim_and_deliver(base, agent_id, token, http, mstore, handler, body, agent_token, using_per_agent).await
}

/// Every agent in `registered` pulled in one request
/// (`POST /reactive/pending-batch`) instead of one
/// `GET /reactive/pending/:agent` each, then claimed and delivered exactly as
/// [`sync_agent_reactive`] does, for each agent with messages. The relay
/// answers each agent as its single route would; this is what keeps an
/// install's relay traffic from growing with every open agent.
///
/// `None` when the batch isn't used: fewer than two agents (one pull either
/// way), or a relay without the route. The caller then pulls one agent at a
/// time, as before. An agent the batch couldn't answer (a failed request, an
/// entry missing) is pulled on its own here.
async fn sync_agents_batched(
    base: &str,
    registered: &[String],
    token: &str,
    http: &reqwest::Client,
    mstore: &Arc<Store>,
    handler: &'static crate::backend::reactive::handler::ReactiveHandler,
) -> Option<Vec<AgentSyncOutcome>> {
    if registered.len() < 2 || super::relay::batch_routes_missing(base) {
        return None;
    }
    let shared = shared_token_now(mstore, token).await;

    // The lease first, as for a single pull: claimed or renewed (no request
    // while a renewal is fresh), and no pull for an agent another instance
    // holds (SPEC_AGENT_SINGLE_LIVE_INSTANCE_2026_09_24 §4.5).
    let leases = futures_util::future::join_all(registered.iter().map(|agent_id| {
        let shared = &shared;
        async move { super::wan_lease::ensure(base, agent_id, shared, http, || is_subscribed(agent_id)).await }
    }))
    .await;
    let mut pull: Vec<&String> = Vec::new();
    for (agent_id, lease) in registered.iter().zip(leases) {
        match lease {
            super::wan_lease::Outcome::HeldElsewhere(where_) => {
                crate::backend::agent_admission::fence_agent_named(agent_id, &where_);
            }
            _ => pull.push(agent_id),
        }
    }

    let mut answers = serde_json::Map::new();
    let started = std::time::Instant::now();
    for chunk in pull.chunks(super::relay::MAX_BATCH_AGENTS) {
        let resp = http
            .post(format!("{base}/reactive/pending-batch"))
            .header("Authorization", format!("Bearer {shared}"))
            .header("X-Agent-Instance", super::wan_lease::instance_id())
            .json(&serde_json::json!({ "agent_ids": chunk }))
            .send()
            .await;
        match resp {
            Ok(r) if matches!(r.status().as_u16(), 404 | 405) && answers.is_empty() => {
                super::relay::note_batch_routes_missing(base);
                return None;
            }
            Ok(r) if r.status().is_success() => match r.json::<serde_json::Value>().await {
                Ok(v) => {
                    if let Some(results) = v.get("results").and_then(|r| r.as_object()) {
                        answers.extend(results.clone());
                    }
                }
                Err(e) => tracing::warn!(error = %e, "cloud_subscriber: parse batch pull failed"),
            },
            Ok(r) => tracing::warn!(status = %r.status(), "cloud_subscriber: batch pull non-2xx; pulling those agents one by one"),
            Err(e) => tracing::warn!(error = %e, "cloud_subscriber: batch pull failed; pulling those agents one by one"),
        }
    }

    let outcomes = futures_util::future::join_all(pull.into_iter().map(|agent_id| {
        let entry = answers.get(&agent_id.trim().to_lowercase()).cloned();
        let shared = &shared;
        async move {
            let Some(entry) = entry else {
                return sync_agent_reactive(base, agent_id, token, http, mstore, handler).await;
            };
            match entry.get("error").and_then(|e| e.as_str()) {
                Some("not_holder") => {
                    // The relay's lease fence, as a 409 on a single pull.
                    if let Some(where_) = super::wan_lease::note_not_holder_since(agent_id, &entry, started) {
                        crate::backend::agent_admission::fence_agent_named(agent_id, &where_);
                    }
                    AgentSyncOutcome::Ok
                }
                Some(_) => {
                    delivery_status::pull_failed(agent_id, "the relay couldn't read this agent's messages just now");
                    AgentSyncOutcome::Ok
                }
                None => match serde_json::from_value::<PendingResp>(entry) {
                    Ok(body) if body.injections.is_empty() => {
                        delivery_status::fetched(agent_id, 0, &body.expired);
                        AgentSyncOutcome::Ok
                    }
                    Ok(body) => {
                        // Claimed with the agent's own credential when it has
                        // one, as a single pull would be.
                        let own = crate::muxbus::agent_credentials::ensure_agent_credential(agent_id, mstore, http).await;
                        let using_per_agent = own.is_some();
                        let agent_token = own.unwrap_or_else(|| shared.clone());
                        claim_and_deliver(base, agent_id, token, http, mstore, handler, body, agent_token, using_per_agent).await
                    }
                    Err(e) => {
                        tracing::warn!(agent_id = %agent_id, error = %e, "cloud_subscriber: parse pending failed");
                        delivery_status::pull_failed(agent_id, "the relay's message list couldn't be read");
                        AgentSyncOutcome::Ok
                    }
                },
            }
        }
    }))
    .await;
    Some(outcomes)
}

/// The shared account token for an HTTP call made during a connection: loaded
/// fresh (`fresh`), the same way `wan_publish` and `relay::relay_token` load
/// theirs. Not the connection's `token`: that one is loaded once at connect, a
/// desktop (PKCE) access token lives 15 minutes and a connection up to 2
/// hours, so after minute 15 every call made with it is a 401. For the W3-S
/// key directory that turned every same-account WAN jekt `network-claimed`;
/// for a lease release it left the lease held until it expired. The
/// connection's token is only the fallback when no fresh one can be loaded,
/// and otherwise only authenticates the WebSocket handshake.
async fn fresh_shared_token<F, Fut>(fresh: F, connection_token: &str) -> String
where
    F: FnOnce() -> Fut,
    Fut: std::future::Future<Output = Option<String>>,
{
    fresh().await.unwrap_or_else(|| connection_token.to_string())
}

/// Whether a spawn's exit drops its agent from the cloud subscription.
/// `registration_was_ours`: its own nonce matched. That alone was the old
/// rule, and a registration already gone (`false`) read as "a newer spawn
/// owns it" — so the agent stayed subscribed, and the lease tick renewed
/// its WAN lease for as long as the srv ran, fencing it on every other
/// instance (charlie, 2026-09-26). Keep it only while a live block holds it.
pub(crate) fn drop_cloud_subscription_on_exit(registration_was_ours: bool, still_held: bool) -> bool {
    registration_was_ours || !still_held
}

/// Subscribed agents no live block holds. Every way into the subscription
/// (`add_agent`) runs with a live registration, so one without is a leak
/// from a teardown that missed it; renewing its WAN lease would fence the
/// agent everywhere else for as long as this srv runs.
fn orphaned_subscriptions(registered: &[String], is_live: impl Fn(&str) -> bool) -> Vec<String> {
    registered.iter().filter(|a| !is_live(a)).cloned().collect()
}

/// Drop `agent` from the relay subscription and release its WAN lease,
/// awaiting the release (the `RemoveAgent` arm's is fire-and-forget) — for a
/// Take over, whose requester claims the lease right after. `false` if it
/// wasn't subscribed. `id_store` is where muxbus credentials live.
/// What a Take over's relay release did.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum RelayRelease {
    /// Not in the relay subscription: no relay lease to release.
    NotSubscribed,
    /// Unsubscribed and the lease released (or none was held).
    Released,
    /// Re-added while releasing: the lease was kept.
    Kept,
}

/// Is `agent` in the relay subscription? `true` when there is no
/// subscriber (tests), so a lease check fails open as before.
pub(crate) fn is_subscribed(agent: &str) -> bool {
    get_global_subscriber().is_none_or(|s| s.agents.lock().unwrap().contains(&agent.to_lowercase()))
}

/// Take over's relay step on the requester ([`super::wan_lease::take_over`]),
/// with a fresh shared token. `Unknown` (fail open) when this instance has no
/// MuxBus session or no token can be loaded.
pub(crate) async fn take_over_lease_now(
    id_store: &Arc<Store>,
    http: &reqwest::Client,
    agent: &str,
) -> super::wan_lease::Outcome {
    if get_global_subscriber().is_none() || agent.is_empty() {
        return super::wan_lease::Outcome::Unknown;
    }
    let token = shared_token_now(id_store, "").await;
    if token.is_empty() {
        return super::wan_lease::Outcome::Unknown;
    }
    super::wan_lease::take_over(&super::relay::rest_base_url(), agent, &token, http).await
}

pub(crate) async fn release_agent_now(id_store: &Arc<Store>, agent: &str) -> RelayRelease {
    let Some(sub) = get_global_subscriber() else { return RelayRelease::NotSubscribed };
    sub.release_now(agent, &super::relay::rest_base_url(), shared_token_now(id_store, "")).await
}

/// [`fresh_shared_token`] against the broker and `mstore` (the store muxbus
/// credentials live in — `CloudSubscriber::init_global`'s `id_store`).
async fn shared_token_now(mstore: &Arc<Store>, connection_token: &str) -> String {
    fresh_shared_token(
        || async {
            let scheduler = crate::broker::get_global()?;
            load_valid_token(mstore, &scheduler).await
        },
        connection_token,
    )
    .await
}

/// Load a valid (non-expired) access token via the broker, refreshing first
/// if the stored credential is missing/stale. Returns None if no credentials
/// are stored, the token is expired and refresh fails, or the refresh_token
/// is absent — `ensure_fresh`'s own single-flight guard means this can run
/// concurrently with the broker's background sweep for the same credential
/// without either one duplicating the other's refresh attempt.
///
/// pub(crate): also used by agent_credentials.rs to authenticate the
/// POST /agents/provision call, which requires the human's own user-level
/// (PKCE) token, not an agent-bound M2M one.
pub(crate) async fn load_valid_token(
    mstore: &Arc<Store>,
    scheduler: &crate::broker::RefreshScheduler,
) -> Option<String> {
    if let Err(e) = scheduler.ensure_fresh(crate::muxbus::CREDENTIAL_ID).await {
        tracing::warn!(error = %e, "cloud_subscriber: token refresh failed");
    }
    // Preserve-on-failure means a failed refresh leaves the last-known-good
    // credential in place — re-read regardless of ensure_fresh's outcome and
    // fall back to it if it's still within its validity window, matching the
    // pre-broker behavior: a transient refresh failure should not prevent
    // connecting with a token that's still technically valid.
    //
    // spawn_blocking — reagent P1 on #2260: same synchronous-keychain-read
    // concern as every other muxbus_load call site in this file.
    let load_store = mstore.clone();
    let creds = tokio::task::spawn_blocking(move || load_store.muxbus_load())
        .await
        .ok()
        .and_then(|r| r.ok())
        .flatten();
    match creds {
        Some(c) if c.is_valid() => Some(c.access_token),
        _ => None,
    }
}

/// Classify a `pkce::refresh_token` failure for the broker's
/// `RefreshErrorKind` — `NoRefreshToken`/`Rejected` in the 4xx range mean
/// the credential itself is the problem (only a fresh `muxbus.login` fixes
/// it) *except* 408/429, which mean the request itself was throttled or
/// timed out, not that the refresh_token was rejected — treating those as
/// permanent would strand the credential in `NeedsReauth` past a temporary
/// rate limit (reagent P2 on #2275). Everything else (network blips, 5xx,
/// response parse failures) is worth retrying on the next sweep tick.
fn classify_refresh_token_error(e: crate::muxbus::pkce::RefreshTokenError) -> RefreshErrorKind {
    use crate::muxbus::pkce::RefreshTokenError;
    match &e {
        RefreshTokenError::NoRefreshToken => RefreshErrorKind::PermanentAuthFailure(e.to_string()),
        RefreshTokenError::Rejected { status, .. }
            if (400..500).contains(status) && !matches!(status, 408 | 429) =>
        {
            RefreshErrorKind::PermanentAuthFailure(e.to_string())
        }
        RefreshTokenError::Rejected { .. }
        | RefreshTokenError::Network(_)
        | RefreshTokenError::ParseFailed(_) => RefreshErrorKind::Transient(e.to_string()),
    }
}

#[cfg(test)]
mod tests {
    use super::classify_refresh_token_error;
    use super::is_credential_rejected;
    use super::shared_token_rejection_outcome;
    use super::AgentSyncOutcome;
    use super::{reagent_sig_is_fresh, REAGENT_SIG_MAX_AGE_SECS};
    use crate::broker::RefreshErrorKind;
    use crate::muxbus::pkce::RefreshTokenError;
    use tokio_tungstenite::tungstenite::client::IntoClientRequest;
    use tokio_tungstenite::tungstenite::ClientRequestBuilder;

    fn jwt(sub: &str) -> String {
        use base64::Engine;
        let payload = base64::engine::general_purpose::URL_SAFE_NO_PAD.encode(format!(r#"{{"sub":"{sub}"}}"#));
        format!("e30.{payload}.sig")
    }

    /// A torn sign-in keeps only its refresh token; that is still worth a
    /// refresh, where nothing at all is not.
    #[test]
    fn a_sign_in_with_only_a_refresh_token_can_still_be_refreshed() {
        use crate::muxbus::MuxBusCredentials;
        let in_an_hour = agentmux_common::time::now_secs() + 3600;
        let usable = |access: &str, refresh: &str, expires_at: i64| {
            super::usable_or_refreshable(&MuxBusCredentials {
                access_token: access.into(),
                refresh_token: refresh.into(),
                expires_at,
                ..Default::default()
            })
        };
        assert!(usable("", "rt", 0), "torn: refresh token only");
        assert!(usable("at", "", in_an_hour), "valid access token");
        assert!(usable("at", "rt", 0), "expired, refreshable");
        assert!(!usable("at", "", 0), "expired, nothing to refresh with");
        assert!(!usable("", "", 0), "nothing");
    }

    /// A repaired sign-in is kept only for the account the saved row names.
    #[test]
    fn a_refresh_for_another_account_is_refused() {
        use crate::muxbus::MuxBusCredentials;
        let current = |sub: &str| MuxBusCredentials { user_sub: sub.into(), ..Default::default() };
        let refreshed = |id: String, access: String| MuxBusCredentials {
            id_token: id,
            access_token: access,
            ..Default::default()
        };
        assert!(super::refreshed_same_account(&current("user-a"), &refreshed(jwt("user-a"), jwt("user-a"))));
        assert!(!super::refreshed_same_account(&current("user-a"), &refreshed(jwt("user-b"), jwt("user-b"))));
        assert!(
            super::refreshed_same_account(&current("user-a"), &refreshed(String::new(), jwt("user-a"))),
            "no id token: the access token's"
        );
        assert!(!super::refreshed_same_account(&current("user-a"), &refreshed("x".into(), "y".into())), "unreadable");
        assert!(
            super::refreshed_same_account(&current(""), &refreshed(jwt("user-b"), jwt("user-b"))),
            "an older row with no account recorded can't be checked"
        );
    }

    // The connection's token is loaded once at connect, and a desktop (PKCE)
    // access token lives 15 min while a connection lasts up to 2 h. Handing
    // it to the key directory made every lookup after minute 15 a 401, read
    // as "couldn't check" — every same-account WAN jekt arrived
    // `network-claimed`; a lease release with it left the lease held. Calls
    // made during a connection must get a freshly loaded token.
    #[tokio::test]
    async fn calls_during_a_connection_get_a_fresh_token_not_the_connections() {
        let token = super::fresh_shared_token(|| async { Some("fresh".to_string()) }, "stale-connection-token").await;
        assert_eq!(token, "fresh");
    }

    // Charlie 2026-09-26: a spawn's exit found its block's registration
    // already gone, took that for "a newer spawn owns it" and kept the
    // agent in the cloud subscription — whose lease tick then renewed its
    // WAN lease for as long as the srv ran, fencing the agent everywhere.
    #[test]
    fn an_exit_drops_the_subscription_when_nobody_holds_the_agent() {
        assert!(super::drop_cloud_subscription_on_exit(false, false), "no registration and no live holder");
        assert!(super::drop_cloud_subscription_on_exit(true, false));
        assert!(!super::drop_cloud_subscription_on_exit(false, true), "a newer spawn holds it: keep");
    }

    // A stale sign-in parks the loop instead of reconnecting with backoff
    // (SPEC_CLOUD_SETTINGS_DISCOVERY_2026_09_27.md §3.3): agents coming and
    // going must not wake it, only a sign-in or the slow local recheck.
    #[tokio::test]
    async fn a_parked_loop_ignores_agent_traffic_and_waits_for_the_recheck() {
        use std::time::{Duration, Instant};
        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
        for i in 0..20 {
            tx.send(super::CtrlMsg::AddAgent(format!("a{i}"))).unwrap();
            tx.send(super::CtrlMsg::RemoveAgent(format!("a{i}"))).unwrap();
        }
        let started = Instant::now();
        super::wait_for_sign_in(&mut rx, Duration::from_millis(300)).await;
        assert!(started.elapsed() >= Duration::from_millis(300), "woke before the recheck");
    }

    #[tokio::test]
    async fn a_sign_in_wakes_a_parked_loop_at_once() {
        use std::time::Duration;
        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
        tx.send(super::CtrlMsg::AddAgent("a".into())).unwrap();
        tx.send(super::CtrlMsg::ReloadToken).unwrap();
        tokio::time::timeout(Duration::from_secs(5), super::wait_for_sign_in(&mut rx, Duration::from_secs(3600)))
            .await
            .expect("ReloadToken must end the wait");
    }

    fn test_subscriber(
        agents: &[&str],
    ) -> (super::CloudSubscriber, tokio::sync::mpsc::UnboundedReceiver<super::CtrlMsg>) {
        let (ctrl_tx, ctrl_rx) = tokio::sync::mpsc::unbounded_channel();
        let set = agents.iter().map(|a| a.to_string()).collect::<std::collections::HashSet<_>>();
        (super::CloudSubscriber { ctrl_tx, agents: std::sync::Arc::new(std::sync::Mutex::new(set)) }, ctrl_rx)
    }

    // Codex P2 on #3897: "no live block" can be read just before a respawn
    // registers; its add_agent then finds the key still present and sends
    // nothing, and an unconditional removal after that drops the live
    // agent's subscription. The drop must re-check liveness after removing.
    #[test]
    fn dropping_an_orphan_keeps_an_agent_that_registered_meanwhile() {
        let (sub, mut rx) = test_subscriber(&["opaz"]);
        // Liveness flips to true between the caller's check and the drop.
        assert!(!sub.drop_if_orphaned("Opaz", |_| true));
        assert!(sub.agents.lock().unwrap().contains("opaz"), "still subscribed");
        assert!(rx.try_recv().is_err(), "no RemoveAgent sent");
    }

    // Codex P2 (second pass) on #3897: a respawn registering after the
    // re-check but before RemoveAgent is sent would queue AddAgent first,
    // and the WS loop would process the stale removal last. An add_agent
    // racing a drop must queue after it.
    #[test]
    fn an_add_racing_a_drop_is_queued_after_the_removal() {
        let (sub, mut rx) = test_subscriber(&["opaz"]);
        let sub = std::sync::Arc::new(sub);
        let (entered_tx, entered_rx) = std::sync::mpsc::channel::<()>();
        let adder = {
            let sub = sub.clone();
            std::thread::spawn(move || {
                entered_rx.recv().unwrap();
                sub.add_agent("Opaz");
            })
        };
        let dropped = sub.drop_if_orphaned("Opaz", |_| {
            // The respawn starts registering while liveness is checked.
            entered_tx.send(()).unwrap();
            std::thread::sleep(std::time::Duration::from_millis(50));
            false
        });
        adder.join().unwrap();
        assert!(dropped);
        assert!(matches!(rx.try_recv(), Ok(super::CtrlMsg::RemoveAgent(k)) if k == "opaz"), "removal first");
        assert!(matches!(rx.try_recv(), Ok(super::CtrlMsg::AddAgent(k)) if k == "opaz"), "then the re-add");
        assert!(sub.agents.lock().unwrap().contains("opaz"), "and it ends subscribed");
    }

    // ReAgent P1 (second pass) on #3899: a Take over release sent its
    // RemoveAgent only after the relay round trip, so an agent re-added in
    // that window queued AddAgent first and was then unsubscribed. The
    // removal must be queued with the set change, before any await.
    #[tokio::test]
    async fn a_take_over_release_queues_its_removal_before_a_re_add() {
        let (sub, mut rx) = test_subscriber(&["opaz"]);
        let sub = std::sync::Arc::new(sub);
        let re_adder = sub.clone();
        let released = sub
            .release_now("Opaz", "http://127.0.0.1:9", async move {
                // The agent is reopened while the fresh token is fetched.
                re_adder.add_agent("Opaz");
                String::new()
            })
            .await;
        assert_eq!(released, super::RelayRelease::Kept, "re-added meanwhile: the lease was kept");
        assert!(matches!(rx.try_recv(), Ok(super::CtrlMsg::RemoveAgent(k)) if k == "opaz"), "removal first");
        assert!(matches!(rx.try_recv(), Ok(super::CtrlMsg::AddAgent(k)) if k == "opaz"), "then the re-add");
        assert!(sub.agents.lock().unwrap().contains("opaz"), "and it ends subscribed");
    }

    // ReAgent P2 on #3908: "not subscribed" (e.g. no relay) must not read
    // as "re-added, lease kept".
    // Codex P1 on #3908: the stopped pane's own exit may already have
    // unsubscribed the agent and queued RemoveAgent, whose release runs
    // detached. "Not in the set" is not "no lease": release it here too
    // (idempotently) and answer only once it's gone.
    #[tokio::test]
    async fn a_take_over_release_still_releases_a_lease_the_exit_left_behind() {
        let agent = format!("agent-{}", uuid::Uuid::new_v4());
        crate::muxbus::wan_lease::mark_held_for_test(&agent);
        let (sub, mut rx) = test_subscriber(&[]);
        let outcome = sub.release_now(&agent, "http://127.0.0.1:9", async { String::new() }).await;
        assert_eq!(outcome, super::RelayRelease::Released);
        assert!(!crate::muxbus::wan_lease::is_held(&agent), "the lease is let go before answering");
        assert!(rx.try_recv().is_err(), "no second RemoveAgent");
    }

    #[tokio::test]
    async fn a_take_over_release_of_an_unsubscribed_agent_says_so() {
        let (sub, _rx) = test_subscriber(&[]);
        let outcome = sub.release_now("Opaz", "http://127.0.0.1:9", async { String::new() }).await;
        assert_eq!(outcome, super::RelayRelease::NotSubscribed);
    }

    #[test]
    fn dropping_an_orphan_removes_it_and_tells_the_ws_loop() {
        let (sub, mut rx) = test_subscriber(&["opaz"]);
        assert!(sub.drop_if_orphaned("Opaz", |_| false));
        assert!(!sub.agents.lock().unwrap().contains("opaz"));
        assert!(matches!(rx.try_recv(), Ok(super::CtrlMsg::RemoveAgent(k)) if k == "opaz"));
        assert!(!sub.drop_if_orphaned("Opaz", |_| false), "already gone");
    }

    #[test]
    fn the_lease_tick_finds_subscriptions_no_live_block_holds() {
        let registered = vec!["opaz".to_string(), "maricon".to_string()];
        let orphans = super::orphaned_subscriptions(&registered, |name| name == "maricon");
        assert_eq!(orphans, vec!["opaz".to_string()]);
        assert!(super::orphaned_subscriptions(&registered, |_| true).is_empty());
    }

    #[tokio::test]
    async fn calls_during_a_connection_fall_back_to_the_connections_token() {
        let token = super::fresh_shared_token(|| async { None }, "connection-token").await;
        assert_eq!(token, "connection-token");
    }

    // An agent-binding rejection (403) must trigger the exact same
    // invalidate-and-retry-with-shared-token recovery as an expired token
    // (401), not silently fall through to a stalled, unretried delivery.
    #[test]
    fn treats_401_as_credential_rejected() {
        assert!(is_credential_rejected(reqwest::StatusCode::UNAUTHORIZED));
    }

    #[test]
    fn treats_403_as_credential_rejected() {
        assert!(is_credential_rejected(reqwest::StatusCode::FORBIDDEN));
    }

    #[test]
    fn does_not_treat_other_client_errors_as_credential_rejected() {
        // 400/404/429 etc. are real errors but not "this credential doesn't
        // work here" — retrying with a different credential wouldn't help,
        // so these must keep falling into the generic non-2xx branch
        // (logged, not retried) rather than triggering credential swap.
        for status in [
            reqwest::StatusCode::BAD_REQUEST,
            reqwest::StatusCode::NOT_FOUND,
            reqwest::StatusCode::TOO_MANY_REQUESTS,
        ] {
            assert!(!is_credential_rejected(status), "{status} must not be treated as a credential rejection");
        }
    }

    // reagentx P1 on PR #2573: a 403 on the SHARED token itself (binding
    // mismatch, not expiry) must NOT reconnect the whole session — the still
    // time-valid token survives `load_valid_token` unchanged, so reconnecting
    // just replays the identical rejected token forever, starving every
    // other registered agent's delivery too.
    #[test]
    fn shared_token_403_does_not_reconnect() {
        let outcome = shared_token_rejection_outcome(reqwest::StatusCode::FORBIDDEN, "agent-x");
        assert!(matches!(outcome, AgentSyncOutcome::Ok));
    }

    #[test]
    fn shared_token_401_still_reconnects() {
        let outcome = shared_token_rejection_outcome(reqwest::StatusCode::UNAUTHORIZED, "agent-x");
        assert!(matches!(outcome, AgentSyncOutcome::ReconnectSharedTokenExpired));
    }

    #[test]
    fn does_not_treat_success_or_server_error_as_credential_rejected() {
        for status in [reqwest::StatusCode::OK, reqwest::StatusCode::INTERNAL_SERVER_ERROR] {
            assert!(!is_credential_rejected(status));
        }
    }

    // Anti-replay for reagent-signed WAN jekts.
    #[test]
    fn reagent_sig_exactly_now_is_fresh() {
        assert!(reagent_sig_is_fresh(1_000, 1_000));
    }

    #[test]
    fn reagent_sig_within_the_window_is_fresh() {
        assert!(reagent_sig_is_fresh(1_000, 1_000 + REAGENT_SIG_MAX_AGE_SECS));
        assert!(reagent_sig_is_fresh(1_000, 1_000 - REAGENT_SIG_MAX_AGE_SECS));
    }

    #[test]
    fn reagent_sig_just_past_the_window_is_not_fresh() {
        assert!(!reagent_sig_is_fresh(1_000, 1_000 + REAGENT_SIG_MAX_AGE_SECS + 1));
        assert!(!reagent_sig_is_fresh(1_000, 1_000 - REAGENT_SIG_MAX_AGE_SECS - 1));
    }

    #[test]
    fn reagent_sig_far_in_the_past_is_not_fresh() {
        // The exact scenario this exists to stop: a captured signature from
        // hours ago resubmitted under a new injection id.
        assert!(!reagent_sig_is_fresh(1_000, 1_000 + 3600));
    }

    #[test]
    fn reagent_sig_zero_or_negative_timestamp_is_never_fresh() {
        assert!(!reagent_sig_is_fresh(0, 0));
        assert!(!reagent_sig_is_fresh(-1, 0));
    }

    #[test]
    fn no_refresh_token_is_permanent() {
        assert!(matches!(
            classify_refresh_token_error(RefreshTokenError::NoRefreshToken),
            RefreshErrorKind::PermanentAuthFailure(_)
        ));
    }

    #[test]
    fn a_4xx_rejection_is_permanent() {
        assert!(matches!(
            classify_refresh_token_error(RefreshTokenError::Rejected {
                status: 400,
                body: "invalid_grant".into(),
            }),
            RefreshErrorKind::PermanentAuthFailure(_)
        ));
    }

    #[test]
    fn a_5xx_rejection_is_transient() {
        assert!(matches!(
            classify_refresh_token_error(RefreshTokenError::Rejected {
                status: 503,
                body: "unavailable".into(),
            }),
            RefreshErrorKind::Transient(_)
        ));
    }

    #[test]
    fn rate_limit_and_timeout_rejections_are_transient_not_permanent() {
        // A 429/408 means the request was throttled/timed out, not that the
        // refresh_token itself was rejected — must not strand the
        // credential in NeedsReauth past a temporary rate limit.
        assert!(matches!(
            classify_refresh_token_error(RefreshTokenError::Rejected {
                status: 429,
                body: "rate limited".into(),
            }),
            RefreshErrorKind::Transient(_)
        ));
        assert!(matches!(
            classify_refresh_token_error(RefreshTokenError::Rejected {
                status: 408,
                body: "request timeout".into(),
            }),
            RefreshErrorKind::Transient(_)
        ));
    }

    #[test]
    fn network_and_parse_errors_are_transient() {
        assert!(matches!(
            classify_refresh_token_error(RefreshTokenError::Network("timeout".into())),
            RefreshErrorKind::Transient(_)
        ));
        assert!(matches!(
            classify_refresh_token_error(RefreshTokenError::ParseFailed("bad json".into())),
            RefreshErrorKind::Transient(_)
        ));
    }

    /// Regression for issue #2091: every connection attempt failed the
    /// WebSocket handshake with "Missing, duplicated or incorrect header
    /// sec-websocket-key" from the moment this file was first written,
    /// because the old code hand-built an `http::Request` with only
    /// `Authorization` set — tungstenite's `IntoClientRequest for Request`
    /// is a bare passthrough, so none of Host/Connection/Upgrade/
    /// Sec-WebSocket-Version/Sec-WebSocket-Key were ever added, and
    /// tungstenite's handshake code requires all five to already be present
    /// on the outgoing request. This test builds the request the exact way
    /// `connect_and_run` now does and asserts every required header exists
    /// — it would have failed against the old `Request::builder()` code and
    /// passes against `ClientRequestBuilder`. No network access needed: this
    /// only inspects the request tungstenite would send, before any I/O.
    #[test]
    fn ws_request_carries_every_header_the_handshake_requires() {
        let uri: tokio_tungstenite::tungstenite::http::Uri =
            "wss://muxbus-ws.agentmux.ai".parse().unwrap();
        let request = ClientRequestBuilder::new(uri)
            .with_header("Authorization", "Bearer test-token")
            .into_client_request()
            .expect("request should build");

        let headers = request.headers();
        for required in ["Host", "Connection", "Upgrade", "Sec-WebSocket-Version", "Sec-WebSocket-Key"] {
            assert!(
                headers.contains_key(required),
                "missing required WebSocket handshake header: {required}"
            );
        }
        assert_eq!(headers.get("Authorization").unwrap(), "Bearer test-token");
    }

    // ── The relay's lease fence, end to end against a fake relay ──────────
    // (SPEC_AGENT_SINGLE_LIVE_INSTANCE_2026_09_24 §4.5). Each 409 the relay
    // can answer while another instance holds the agent must fence this
    // process's local holder and take nothing.

    use std::sync::{Arc, Mutex};

    type Seen = Arc<Mutex<Vec<String>>>;

    /// A fake muxbus relay answering `"METHOD /path"` with the scripted
    /// status and JSON body (anything else: 404), recording every request.
    /// Every script answers the lease claim: a 404 there would switch the WAN
    /// tier off for the whole test process (`wan_lease::UNSUPPORTED_UNTIL`).
    async fn fake_relay(routes: Vec<(String, u16, serde_json::Value)>) -> (String, Seen) {
        let seen: Seen = Arc::default();
        let app = axum::Router::new().fallback({
            let seen = Arc::clone(&seen);
            move |req: axum::extract::Request| {
                let (seen, routes) = (Arc::clone(&seen), routes.clone());
                async move {
                    let line = format!("{} {}", req.method(), req.uri().path());
                    seen.lock().unwrap().push(line.clone());
                    let (status, body) = routes
                        .iter()
                        .find(|(route, _, _)| *route == line)
                        .map(|(_, status, body)| (*status, body.clone()))
                        .unwrap_or((404, serde_json::Value::Null));
                    (axum::http::StatusCode::from_u16(status).unwrap(), axum::Json(body))
                }
            }
        });
        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
        let addr = listener.local_addr().unwrap();
        tokio::spawn(async move { axum::serve(listener, app).await.unwrap() });
        (format!("http://{addr}"), seen)
    }

    /// Hold `agent` in this process, as a running agent pane does. The
    /// receiver gets the reason when the holder is fenced.
    fn hold_locally(
        agent: &str,
    ) -> (tempfile::TempDir, crate::backend::agent_admission::HeldAgentLease, std::sync::mpsc::Receiver<String>) {
        let tmp = tempfile::tempdir().unwrap();
        let store = Arc::new(crate::registry::LeaseStore::open_with_ttl(tmp.path(), 60_000).unwrap());
        let (tx, rx) = std::sync::mpsc::channel();
        let tx = Mutex::new(tx);
        let held = crate::backend::agent_admission::acquire(
            &store,
            &format!("uid-{agent}"),
            agent,
            &Arc::from("boot-test"),
            "block-test",
            None,
            move |why| {
                let _ = tx.lock().unwrap().send(why);
            },
        )
        .unwrap();
        (tmp, held, rx)
    }

    fn not_holder() -> serde_json::Value {
        serde_json::json!({
            "error": "not_holder",
            "held_by": { "host": "desk", "channel": "stable", "version": "0.58.0" },
        })
    }

    fn lease_granted() -> (String, u16, serde_json::Value) {
        ("POST /agents/lease".into(), 200, serde_json::json!({ "ok": true, "epoch": 1 }))
    }

    /// Run one `InjectAvailable` sync of `agent` against the relay at `base`.
    async fn sync(base: &str, agent: &str) {
        let mstore = Arc::new(crate::backend::storage::store::Store::open_in_memory().unwrap());
        let handler = crate::backend::reactive::handler::get_global_handler();
        let outcome = super::sync_agent_reactive(base, agent, "test-token", &reqwest::Client::new(), &mstore, handler).await;
        assert!(matches!(outcome, AgentSyncOutcome::Ok));
    }

    fn assert_fenced(agent: &str, rx: &std::sync::mpsc::Receiver<String>) {
        let why = rx.try_recv().expect("the local holder was fenced");
        assert!(why.contains("taken over") && why.contains("computer desk"), "{why}");
        assert_eq!(
            crate::muxbus::wan_lease::held_elsewhere(agent).as_deref(),
            Some("computer desk, channel stable, v0.58.0"),
            "remembered for the early admission check"
        );
    }

    #[tokio::test]
    async fn a_lease_held_elsewhere_fences_the_local_holder_before_any_pull() {
        let agent = format!("agent-{}", uuid::Uuid::new_v4());
        let (_tmp, _held, rx) = hold_locally(&agent);
        let (base, seen) = fake_relay(vec![("POST /agents/lease".into(), 409, not_holder())]).await;

        sync(&base, &agent).await;

        assert_fenced(&agent, &rx);
        assert_eq!(*seen.lock().unwrap(), ["POST /agents/lease"], "no pending pull");
    }

    #[tokio::test]
    async fn a_409_on_the_pending_pull_fences_the_local_holder_and_claims_nothing() {
        let agent = format!("agent-{}", uuid::Uuid::new_v4());
        let (_tmp, _held, rx) = hold_locally(&agent);
        let pending = format!("GET /reactive/pending/{agent}");
        let (base, seen) = fake_relay(vec![lease_granted(), (pending.clone(), 409, not_holder())]).await;

        sync(&base, &agent).await;

        assert_fenced(&agent, &rx);
        assert_eq!(*seen.lock().unwrap(), ["POST /agents/lease", pending.as_str()], "no ack");
    }

    /// ReAgent P1 on #3746: the lease can change hands between the pull and
    /// the ack. A 409 on the ack must fence like one on the pull, not be
    /// logged as a generic error while the local copy keeps running.
    #[tokio::test]
    async fn a_409_on_the_ack_fences_the_local_holder_and_delivers_nothing() {
        let agent = format!("agent-{}", uuid::Uuid::new_v4());
        let (_tmp, _held, rx) = hold_locally(&agent);
        let pending = format!("GET /reactive/pending/{agent}");
        let (base, seen) = fake_relay(vec![
            lease_granted(),
            (pending.clone(), 200, serde_json::json!({ "injections": [{ "id": "inj-1", "message": "hello" }] })),
            ("POST /reactive/ack".into(), 409, not_holder()),
        ])
        .await;

        sync(&base, &agent).await;

        assert_fenced(&agent, &rx);
        assert_eq!(
            *seen.lock().unwrap(),
            ["POST /agents/lease", pending.as_str(), "POST /reactive/ack"],
            "nothing delivered, so nothing released"
        );
    }

    // ── One pull for every agent (POST /reactive/pending-batch) ──────────

    /// Run one `InjectAvailable` pull of `agents` through the batch path.
    async fn sync_batched(base: &str, agents: &[String]) -> Option<usize> {
        let mstore = Arc::new(crate::backend::storage::store::Store::open_in_memory().unwrap());
        let handler = crate::backend::reactive::handler::get_global_handler();
        super::sync_agents_batched(base, agents, "test-token", &reqwest::Client::new(), &mstore, handler)
            .await
            .map(|outcomes| outcomes.len())
    }

    #[tokio::test]
    async fn one_request_pulls_for_every_agent_and_only_agents_with_messages_are_claimed() {
        let [a, b, c, d] = std::array::from_fn(|_| format!("agent-{}", uuid::Uuid::new_v4()));
        let (_tmp, _held, rx) = hold_locally(&b);
        let results = serde_json::json!({ "results": {
            a.clone(): { "agent_id": a.clone(), "injections": [{ "id": "inj-1", "message": "hello" }], "count": 1, "expired": [] },
            b.clone(): not_holder(),
            c.clone(): { "agent_id": c.clone(), "injections": [], "count": 0, "expired": [] },
            // d is missing from the answer: pulled on its own.
        }});
        let pd = format!("GET /reactive/pending/{d}");
        let (base, seen) = fake_relay(vec![
            lease_granted(),
            ("POST /reactive/pending-batch".into(), 200, results),
            ("POST /reactive/ack".into(), 200, serde_json::json!({ "acknowledged": [], "delivered_at": "2026-10-11T00:00:00Z" })),
            (pd.clone(), 200, serde_json::json!({ "injections": [] })),
        ])
        .await;

        let agents = vec![a.clone(), b.clone(), c.clone(), d.clone()];
        assert_eq!(sync_batched(&base, &agents).await, Some(4));

        assert_fenced(&b, &rx);
        let seen = seen.lock().unwrap();
        let count = |line: &str| seen.iter().filter(|s| *s == line).count();
        assert_eq!(count("POST /reactive/pending-batch"), 1, "{seen:?}");
        assert_eq!(count("POST /reactive/ack"), 1, "only {a} had messages: {seen:?}");
        assert_eq!(count(&pd), 1, "{d}, missing from the answer, pulled on its own: {seen:?}");
        assert!(
            !seen.iter().any(|s| s.starts_with("GET /reactive/pending/") && *s != pd),
            "no single pull for the agents the batch answered: {seen:?}"
        );
    }

    #[tokio::test]
    async fn an_unavailable_entry_is_a_failed_pull_for_that_agent_only() {
        let [a, b] = std::array::from_fn(|_| format!("agent-{}", uuid::Uuid::new_v4()));
        let results = serde_json::json!({ "results": {
            a.clone(): { "error": "unavailable" },
            b.clone(): { "agent_id": b.clone(), "injections": [], "count": 0, "expired": [] },
        }});
        let (base, seen) = fake_relay(vec![lease_granted(), ("POST /reactive/pending-batch".into(), 200, results)]).await;

        assert_eq!(sync_batched(&base, &[a.clone(), b.clone()]).await, Some(2));
        assert!(
            !seen.lock().unwrap().iter().any(|s| s.starts_with("GET /reactive/pending/")),
            "retried on the next pull, not at once"
        );
    }

    #[tokio::test]
    async fn a_relay_without_the_batch_route_is_pulled_one_agent_at_a_time() {
        let [a, b] = std::array::from_fn(|_| format!("agent-{}", uuid::Uuid::new_v4()));
        let (pa, pb) = (format!("GET /reactive/pending/{a}"), format!("GET /reactive/pending/{b}"));
        let (base, seen) = fake_relay(vec![
            lease_granted(),
            (pa.clone(), 200, serde_json::json!({ "injections": [] })),
            (pb.clone(), 200, serde_json::json!({ "injections": [] })),
        ])
        .await;
        let agents = Arc::new(Mutex::new(HashSet::from([a.clone(), b.clone()])));
        let mstore = Arc::new(crate::backend::storage::store::Store::open_in_memory().unwrap());

        assert!(!catch_up(&base, "test-token", &reqwest::Client::new(), &agents, &mstore).await);
        assert!(crate::muxbus::relay::batch_routes_missing(&base));
        assert!(!catch_up(&base, "test-token", &reqwest::Client::new(), &agents, &mstore).await);

        let seen = seen.lock().unwrap();
        assert_eq!(seen.iter().filter(|s| *s == "POST /reactive/pending-batch").count(), 1, "asked once, then remembered: {seen:?}");
        assert_eq!(seen.iter().filter(|s| **s == pa || **s == pb).count(), 4, "both agents, both times: {seen:?}");
    }

    #[tokio::test]
    async fn a_single_agent_is_pulled_with_the_single_route() {
        let a = format!("agent-{}", uuid::Uuid::new_v4());
        let (base, seen) = fake_relay(vec![lease_granted()]).await;
        assert_eq!(sync_batched(&base, &[a]).await, None);
        assert!(seen.lock().unwrap().is_empty());
    }

    // ── Catch-up pull: a message must not wait for an unrelated wake ──────
    use super::{agents_to_seed, catch_up, CATCH_UP_EVERY};
    use std::collections::HashSet;
    use std::time::Duration;

    // (2026-09-30: a jekt to manoz reached the relay at 23:12:14 and was
    // delivered at 23:21:00, when someone else's traffic woke the subscriber.)

    #[test]
    fn the_seed_is_every_live_registered_agent_lowercased_and_deduplicated() {
        let registered: Vec<String> = ["Alice", "alice", "Bob", "Carol"].map(String::from).to_vec();
        let got = agents_to_seed(&registered, |a| a != "carol");
        assert_eq!(got, ["alice", "bob"], "Carol has no live pane; Alice appears once, lowercased");
    }

    #[test]
    fn an_empty_registry_seeds_nothing() {
        assert!(agents_to_seed(&[], |_| true).is_empty());
    }

    #[tokio::test]
    async fn a_catch_up_pulls_pending_for_every_registered_agent() {
        let (a, b) = (format!("agent-{}", uuid::Uuid::new_v4()), format!("agent-{}", uuid::Uuid::new_v4()));
        let (pa, pb) = (format!("GET /reactive/pending/{a}"), format!("GET /reactive/pending/{b}"));
        let (base, seen) = fake_relay(vec![
            lease_granted(),
            (pa.clone(), 200, serde_json::json!({ "injections": [] })),
            (pb.clone(), 200, serde_json::json!({ "injections": [] })),
        ])
        .await;
        let agents = Arc::new(Mutex::new(HashSet::from([a.clone(), b.clone()])));
        let mstore = Arc::new(crate::backend::storage::store::Store::open_in_memory().unwrap());

        let reconnect = catch_up(&base, "test-token", &reqwest::Client::new(), &agents, &mstore).await;

        assert!(!reconnect, "nothing went wrong, so no reconnect");
        let seen = seen.lock().unwrap();
        assert!(seen.contains(&pa), "pulled for {a}: {seen:?}");
        assert!(seen.contains(&pb), "pulled for {b}: {seen:?}");
    }

    #[tokio::test]
    async fn a_catch_up_with_no_registered_agents_asks_the_relay_for_nothing() {
        let (base, seen) = fake_relay(vec![lease_granted()]).await;
        let agents = Arc::new(Mutex::new(HashSet::new()));
        let mstore = Arc::new(crate::backend::storage::store::Store::open_in_memory().unwrap());

        assert!(!catch_up(&base, "test-token", &reqwest::Client::new(), &agents, &mstore).await);

        assert!(seen.lock().unwrap().is_empty());
    }

    #[test]
    fn the_fallback_interval_is_short_enough_to_matter_and_long_enough_to_be_cheap() {
        assert!(CATCH_UP_EVERY >= Duration::from_secs(30), "faster than this is polling");
        assert!(CATCH_UP_EVERY <= Duration::from_secs(300), "slower than this is the 8m46s we are fixing");
    }

    /// The real session loop, against a local fake relay and WebSocket, for the
    /// exact failure of 2026-09-30: a jekt is already waiting for an agent
    /// when the session starts, and NO wake signal ever arrives. The relay
    /// never sends one here, so any pull is the catch-up, not a wake.
    #[tokio::test]
    async fn a_session_pulls_what_is_waiting_on_connect_and_when_an_agent_is_added() {
        use futures_util::StreamExt;
        // Process-global env, shared with the pkce tests that set the same URL.
        let _serial = crate::muxbus::pkce::TEST_SERIAL
            .lock()
            .unwrap_or_else(|e| e.into_inner());

        let (a, b) = (format!("agent-{}", uuid::Uuid::new_v4()), format!("agent-{}", uuid::Uuid::new_v4()));
        let (pa, pb) = (format!("GET /reactive/pending/{a}"), format!("GET /reactive/pending/{b}"));
        let (rest, seen) = fake_relay(vec![
            lease_granted(),
            (pa.clone(), 200, serde_json::json!({ "injections": [] })),
            (pb.clone(), 200, serde_json::json!({ "injections": [] })),
        ])
        .await;

        // A WebSocket relay that accepts, records what the client says, and
        // never sends a wake.
        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
        let ws_url = format!("ws://{}", listener.local_addr().unwrap());
        let frames: Seen = Arc::default();
        let server = tokio::spawn({
            let frames = Arc::clone(&frames);
            async move {
                let (stream, _) = listener.accept().await.unwrap();
                let mut ws = tokio_tungstenite::accept_async(stream).await.unwrap();
                while let Some(Ok(msg)) = ws.next().await {
                    if let tokio_tungstenite::tungstenite::Message::Text(t) = msg {
                        frames.lock().unwrap().push(t.to_string());
                    }
                }
            }
        });

        std::env::set_var("AGENTMUX_MUXBUS_REST_URL", &rest);
        std::env::set_var("AGENTMUX_MUXBUS_WS_URL", &ws_url);

        let agents = Arc::new(Mutex::new(HashSet::from([a.clone()])));
        let (ctrl_tx, mut ctrl_rx) = tokio::sync::mpsc::unbounded_channel::<super::CtrlMsg>();
        let mstore = Arc::new(crate::backend::storage::store::Store::open_in_memory().unwrap());
        let session = tokio::spawn({
            let (agents, mstore) = (Arc::clone(&agents), Arc::clone(&mstore));
            async move {
                super::connect_and_run("test-token", agents, &mut ctrl_rx, &mstore, &reqwest::Client::new()).await
            }
        });

        async fn within_5s(what: &str, mut cond: impl FnMut() -> bool) {
            let deadline = std::time::Instant::now() + Duration::from_secs(5);
            while !cond() {
                assert!(std::time::Instant::now() < deadline, "timed out waiting for {what}");
                tokio::time::sleep(Duration::from_millis(20)).await;
            }
        }

        // 1. Connected, subscribed, and pulled for the agent with no wake sent.
        within_5s(&pa, || seen.lock().unwrap().contains(&pa)).await;
        assert!(
            frames.lock().unwrap().iter().any(|f| f.contains("\"subscribe\"") && f.contains(&a)),
            "subscribed {a}: {:?}",
            frames.lock().unwrap()
        );

        // 2. An agent added later is pulled for at once, again with no wake.
        agents.lock().unwrap().insert(b.clone());
        ctrl_tx.send(super::CtrlMsg::AddAgent(b.clone())).unwrap();
        within_5s(&pb, || seen.lock().unwrap().contains(&pb)).await;

        session.abort();
        server.abort();
        std::env::remove_var("AGENTMUX_MUXBUS_REST_URL");
        std::env::remove_var("AGENTMUX_MUXBUS_WS_URL");
    }
}