Module cloud_subscriber 

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.

Modules

delivery 🔒

Claim → verify → deliver for one agent’s pulled injections: the half of a pull that comes after the relay answered, shared by the single pull (sync_agent_reactive) and the batch pull (sync_agents_batched).

Structs

CloudSubscriber

Enums

AgentSyncOutcome 🔒

ClientMsg 🔒

CtrlMsg 🔒

RelayRelease 🔒

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.

ServerMsg 🔒

Constants

BROKER_SWEEP_INTERVAL_SECS 🔒

CATCH_UP_EVERY 🔒

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.

CLIENT_PING_INTERVAL_SECS 🔒

MAX_RECONNECT_DELAY_SECS 🔒

MUXBUS_REST_URL 🔒

MUXBUS_WS_URL 🔒

Fallback when the cloud publishes no settings (discovery::ws_url).

REAGENT_SIG_MAX_AGE_SECS 🔒

RECONNECT_DELAY_SECS 🔒

STALE_RECHECK_SECS 🔒

Statics

GLOBAL_SUBSCRIBER 🔒

Functions

agents_to_seed 🔒

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.

catch_up 🔒

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

classify_refresh_token_error 🔒

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.

connect_and_run 🔒

drop_cloud_subscription_on_exit 🔒

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.

fresh_shared_token 🔒

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.

get_global_subscriber

handle_server_msg 🔒

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.

is_credential_rejected 🔒

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.

is_subscribed 🔒

Is agent in the relay subscription? true when there is no subscriber (tests), so a lease check fails open as before.

load_valid_token 🔒

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.

orphaned_subscriptions 🔒

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.

reagent_sig_is_fresh 🔒

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.

refreshed_same_account 🔒

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.

release_agent_now 🔒

run_loop 🔒

shared_token_now 🔒

fresh_shared_token against the broker and mstore (the store muxbus credentials live in — CloudSubscriber::init_global’s id_store).

shared_token_rejection_outcome 🔒

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:

sync_agent_reactive 🔒

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.

sync_agents_batched 🔒

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.

take_over_lease_now 🔒

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.

usable_or_refreshable 🔒

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

wait_for_sign_in 🔒

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.