agentmux_srv/muxbus/delivery_status/
mod.rs

1// Copyright 2026, AgentMux Corp.
2// SPDX-License-Identifier: Apache-2.0
3
4//! Whether cloud (MuxBus) messages are reaching this channel's agents, made
5//! visible: a sign-in that stops working, or a relay that can't be reached,
6//! used to leave nothing but a log line while agents silently stopped getting
7//! review, CI and cross-machine messages.
8//!
9//! **State** (`muxbus.status`'s `delivery`, and the `muxbus:status` event on
10//! every change) is derived in [`machine::derive`] from two things only: what
11//! the cloud subscriber's loop last reported ([`set_link`], [`saw_session`])
12//! and the credential broker's state for the sign-in. It never reads the
13//! credential store itself.
14//!
15//! **What follows from it** ([`machine::Effect`]):
16//! - one OS notification when a needs-sign-in episode starts (the notify
17//!   Router's `CloudSignedOut` kind), taken down when it ends;
18//! - after [`machine::PAUSE_NOTICE_MS`] paused, one AgentMux system note to
19//!   every agent subscribed to the cloud, and one more when delivery resumes,
20//!   with what was delivered and what expired meanwhile ([`notes`]). The
21//!   notes are delivered locally by srv, never through the relay; see
22//!   [`notes::NOTE_MARKER`] for why one can't be forged.
23
24mod machine;
25pub mod notes;
26
27use std::sync::atomic::{AtomicBool, Ordering};
28use std::sync::{Arc, Mutex, OnceLock};
29use std::time::Duration;
30
31use agentmux_common::time::now_ms;
32pub use machine::Link;
33use machine::{Effect, Machine};
34use notes::{ExpiredItem, Note, NoteBook, RetryQueue};
35
36use crate::backend::eventbus::EventBus;
37use crate::backend::mps::Broker;
38use crate::backend::rpc_types::MuxBusDeliveryStatus;
39use crate::broker::CredentialState;
40
41/// The WebSocket event carrying a changed [`MuxBusDeliveryStatus`].
42pub const EVENT_MUXBUS_STATUS: &str = "muxbus:status";
43
44/// How often the state is re-derived (the broker can move on its own) and
45/// the time-based notes are checked.
46const TICK: Duration = Duration::from_secs(15);
47
48struct Sinks {
49    event_bus: Arc<EventBus>,
50    broker: Arc<Broker>,
51}
52
53struct Runtime {
54    machine: Mutex<Machine>,
55    book: Mutex<NoteBook>,
56    retry: Mutex<RetryQueue>,
57    sinks: OnceLock<Sinks>,
58}
59
60static RUNTIME: OnceLock<Runtime> = OnceLock::new();
61
62fn rt() -> &'static Runtime {
63    RUNTIME.get_or_init(|| Runtime {
64        machine: Mutex::new(Machine::new(now_ms())),
65        book: Mutex::new(NoteBook::default()),
66        retry: Mutex::new(RetryQueue::default()),
67        sinks: OnceLock::new(),
68    })
69}
70
71fn broker_state() -> Option<CredentialState> {
72    crate::broker::get_global().and_then(|s| s.state(crate::muxbus::CREDENTIAL_ID))
73}
74
75fn subscribed_agents() -> Vec<String> {
76    crate::muxbus::cloud_subscriber::get_global_subscriber()
77        .map(|s| s.subscribed_agents())
78        .unwrap_or_default()
79}
80
81/// Where status changes and the notification go. Call once at startup;
82/// starts the tick. Before this, state is still tracked but nothing is sent.
83pub fn install(event_bus: Arc<EventBus>, broker: Arc<Broker>) {
84    if rt().sinks.set(Sinks { event_bus, broker }).is_err() {
85        return;
86    }
87    if tokio::runtime::Handle::try_current().is_err() {
88        return;
89    }
90    tokio::spawn(async {
91        let mut tick = tokio::time::interval(TICK);
92        tick.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
93        loop {
94            tick.tick().await;
95            run_tick();
96        }
97    });
98}
99
100fn run_tick() {
101    let broker = broker_state();
102    let now = now_ms();
103    let agents = subscribed_agents();
104    let effects = {
105        let mut machine = rt().machine.lock().unwrap_or_else(|e| e.into_inner());
106        machine.keep_pull_errors_of(&agents);
107        machine.tick(broker.as_ref(), now)
108    };
109    apply(effects);
110    send_owed_sign_in_notice();
111    let retries = rt().retry.lock().unwrap_or_else(|e| e.into_inner()).take();
112    for (note, attempts) in retries {
113        deliver_one(note, attempts + 1);
114    }
115    let notes = {
116        let mut book = rt().book.lock().unwrap_or_else(|e| e.into_inner());
117        let mut notes = book.pause_late_joiners(&agents);
118        notes.extend(book.tick(now));
119        notes
120    };
121    deliver(notes);
122}
123
124/// The current state, for `muxbus.status`.
125pub fn status() -> MuxBusDeliveryStatus {
126    rt().machine.lock().unwrap_or_else(|e| e.into_inner()).status().clone()
127}
128
129/// The subscriber found a stored sign-in, or a sign-in just succeeded.
130pub fn saw_session(email: Option<String>) {
131    let broker = broker_state();
132    let effects = rt().machine.lock().unwrap_or_else(|e| e.into_inner()).saw_session(email, broker.as_ref(), now_ms());
133    apply(effects);
134}
135
136/// The user just signed in (`muxbus.login`).
137pub fn fresh_sign_in(email: Option<String>) {
138    let effects = rt().machine.lock().unwrap_or_else(|e| e.into_inner()).fresh_sign_in(email, now_ms());
139    apply(effects);
140}
141
142/// The user signed out (`muxbus.disconnect`).
143pub fn signed_out() {
144    let effects = rt().machine.lock().unwrap_or_else(|e| e.into_inner()).signed_out(now_ms());
145    apply(effects);
146}
147
148/// What the subscriber's loop is doing now.
149pub fn set_link(link: Link) {
150    let broker = broker_state();
151    let effects = rt().machine.lock().unwrap_or_else(|e| e.into_inner()).set_link(link, broker.as_ref(), now_ms());
152    apply(effects);
153}
154
155/// One agent's pull answered: `delivered` messages reached it and the relay
156/// reported `expired`. Sends its resume or expiry note, if it has one; call
157/// after the pull's own messages are delivered, so the note comes last.
158pub fn fetched(agent: &str, delivered: usize, expired: &[ExpiredItem]) {
159    let broker = broker_state();
160    let effects = rt().machine.lock().unwrap_or_else(|e| e.into_inner()).relay_answered(agent, broker.as_ref(), now_ms());
161    apply(effects);
162    let note = rt().book.lock().unwrap_or_else(|e| e.into_inner()).fetched(agent, delivered, expired);
163    deliver(note.into_iter().collect());
164}
165
166/// `agent`'s pull (fetch or claim) got no usable answer from the relay,
167/// though the WebSocket may be open: its messages aren't arriving.
168pub fn pull_failed(agent: &str, reason: impl Into<String>) {
169    let broker = broker_state();
170    let effects =
171        rt().machine.lock().unwrap_or_else(|e| e.into_inner()).pull_failed(agent, reason.into(), broker.as_ref(), now_ms());
172    apply(effects);
173}
174
175/// The episode's notification is owed but had no router to go to (no window
176/// had connected yet); [`run_tick`] sends it once one exists.
177static SIGN_IN_NOTICE_OWED: AtomicBool = AtomicBool::new(false);
178
179fn send_owed_sign_in_notice() {
180    if SIGN_IN_NOTICE_OWED.load(Ordering::SeqCst) {
181        if let Some(router) = router() {
182            if SIGN_IN_NOTICE_OWED.swap(false, Ordering::SeqCst) {
183                router.cloud_signed_out();
184            }
185        }
186    }
187}
188
189fn apply(effects: Vec<Effect>) {
190    for effect in effects {
191        match effect {
192            Effect::Changed(status) => {
193                tracing::info!(
194                    state = ?status.state,
195                    error = status.last_error.as_deref().unwrap_or(""),
196                    "muxbus delivery: state changed"
197                );
198                if let Some(sinks) = rt().sinks.get() {
199                    sinks.event_bus.broadcast_event(&crate::backend::eventbus::EventBus::typed_ws_event::<crate::backend::mps_events::MuxbusStatus>(&status, String::new()));
200                }
201            }
202            Effect::NotifySignInNeeded => {
203                SIGN_IN_NOTICE_OWED.store(true, Ordering::SeqCst);
204                send_owed_sign_in_notice();
205            }
206            Effect::RetractSignInNeeded => {
207                SIGN_IN_NOTICE_OWED.store(false, Ordering::SeqCst);
208                if let Some(router) = router() {
209                    router.resolve("", crate::backend::notify::policy::Family::Cloud);
210                }
211            }
212            Effect::PauseNote { since_ms, reason } => {
213                tracing::warn!(reason, "muxbus delivery: paused for two minutes, telling the agents");
214                let notes = rt().book.lock().unwrap_or_else(|e| e.into_inner()).pause(since_ms, reason, &subscribed_agents());
215                deliver(notes);
216            }
217            Effect::Resumed { at_ms } => {
218                rt().retry.lock().unwrap_or_else(|e| e.into_inner()).drop_pause_notes();
219                rt().book.lock().unwrap_or_else(|e| e.into_inner()).resumed(at_ms);
220            }
221        }
222    }
223}
224
225fn router() -> Option<Arc<crate::backend::notify::router::Router>> {
226    let sinks = rt().sinks.get()?;
227    let router = crate::backend::notify::router::get(&sinks.broker);
228    if router.is_none() {
229        tracing::debug!("muxbus delivery: no notification router yet (no window connected)");
230    }
231    router
232}
233
234fn deliver(notes: Vec<Note>) {
235    for note in notes {
236        deliver_one(note, 1);
237    }
238}
239
240/// Deliver `note` (its `attempt`-th try); a failure is retried on later ticks.
241fn deliver_one(note: Note, attempt: u32) {
242    let handler = crate::backend::reactive::handler::get_global_handler();
243    if let Err(e) = handler.deliver_system_note(&note.agent, &note.text) {
244        let agent = note.agent.clone();
245        if rt().retry.lock().unwrap_or_else(|e| e.into_inner()).failed(note, attempt) {
246            tracing::info!(agent = %agent, attempt, error = %e, "muxbus delivery: system note not delivered, will retry");
247        } else {
248            tracing::warn!(agent = %agent, error = %e, "muxbus delivery: system note not delivered, giving up");
249        }
250    }
251}
agentmux_srv/muxbus/delivery_status/mod.rs

agentmux_srv/muxbus/delivery_status/
mod.rs
