agentmux_srv/muxbus/delivery_status/
machine.rs

1// Copyright 2026, AgentMux Corp.
2// SPDX-License-Identifier: Apache-2.0
3
4//! The pure half of [`super`]: what state cloud delivery is in, given what
5//! the cloud subscriber and the credential broker last said, and what to do
6//! when it changes. No clock, no I/O; every call takes `now_ms`.
7
8use std::collections::BTreeMap;
9
10use crate::backend::rpc_types::{MuxBusDeliveryState, MuxBusDeliveryStatus};
11use crate::broker::CredentialState;
12
13/// How long cloud delivery must have been paused before the agents are told:
14/// a reconnect or a token refresh that recovers within this never reaches
15/// them. The status bar's amber "reconnecting" waits the same time.
16pub const PAUSE_NOTICE_MS: i64 = 2 * 60 * 1000;
17
18/// What the cloud subscriber's loop is doing, as it last reported.
19#[derive(Debug, Clone, PartialEq, Eq)]
20pub enum Link {
21    /// Nothing stored to sign in with: the loop waits for a sign-in.
22    NoSignIn,
23    /// A sign-in is stored but can't work anymore (the reason).
24    SignInStale(String),
25    /// Signed in, but the relay isn't reachable or the token can't be
26    /// refreshed for now (the reason). The loop retries on its own.
27    Unreachable(String),
28    /// A sign-in was just made and saved; the loop is on its way to the
29    /// relay. The broker may still hold the previous sign-in's refusal, which
30    /// says nothing about this one.
31    Connecting,
32    /// The WebSocket to the relay is open.
33    Connected,
34}
35
36/// Something [`Machine`] decided the runtime should do.
37#[derive(Debug, Clone, PartialEq, Eq)]
38pub enum Effect {
39    /// The status changed: tell the UI.
40    Changed(MuxBusDeliveryStatus),
41    /// A needs-sign-in episode began: one OS notification.
42    NotifySignInNeeded,
43    /// That episode ended: take the notification down if it's still up.
44    RetractSignInNeeded,
45    /// Cloud delivery has been paused for [`PAUSE_NOTICE_MS`]: tell the
46    /// agents, since `since_ms`, because of `reason`.
47    PauseNote { since_ms: i64, reason: &'static str },
48    /// Delivery came back after a pause the agents were told about.
49    Resumed { at_ms: i64 },
50}
51
52/// The state a set of inputs comes to, and why it isn't connected.
53pub fn derive(
54    session_known: bool,
55    link: &Link,
56    broker: Option<&CredentialState>,
57) -> (MuxBusDeliveryState, Option<String>) {
58    match link {
59        Link::Connected => (MuxBusDeliveryState::Connected, None),
60        _ if !session_known => (MuxBusDeliveryState::SignedOut, None),
61        Link::SignInStale(reason) => (MuxBusDeliveryState::NeedsSignIn, Some(reason.clone())),
62        Link::Connecting => (MuxBusDeliveryState::Reconnecting, Some("connecting".to_string())),
63        // Signed in before, and now nothing readable is stored: the sign-in
64        // was lost, not given up (a deliberate sign-out clears `session_known`).
65        Link::NoSignIn => (
66            MuxBusDeliveryState::NeedsSignIn,
67            Some("the stored sign-in can't be read".to_string()),
68        ),
69        Link::Unreachable(error) => match broker {
70            Some(CredentialState::NeedsReauth { rejected: true, reason, .. }) => {
71                (MuxBusDeliveryState::NeedsSignIn, Some(format!("the sign-in was refused: {reason}")))
72            }
73            _ => (MuxBusDeliveryState::Reconnecting, Some(error.clone())),
74        },
75    }
76}
77
78/// Why the agents' pause note says messages are paused.
79pub fn pause_reason(state: MuxBusDeliveryState) -> &'static str {
80    match state {
81        MuxBusDeliveryState::NeedsSignIn => "MuxBus needs a sign-in",
82        _ => "AgentMux can't reach MuxBus",
83    }
84}
85
86/// Cloud delivery's state over time, and the episodes that drive the
87/// notification and the agents' notes.
88#[derive(Debug, Clone)]
89pub struct Machine {
90    session_known: bool,
91    email: Option<String>,
92    link: Link,
93    /// Agents whose REST pull is failing (agent → reason) though the
94    /// WebSocket may be open: their messages can't be fetched, so that is
95    /// not connected either. An agent's entry clears when its own pull
96    /// answers, not another agent's.
97    pull_errors: BTreeMap<String, String>,
98    status: MuxBusDeliveryStatus,
99    /// When the current pause began; open until delivery is connected again.
100    /// A sign-out ends one only if the agents weren't told about it.
101    paused_since_ms: Option<i64>,
102    /// The agents were told about the current pause.
103    pause_noted: bool,
104    /// The current needs-sign-in episode has had its notification.
105    sign_in_notified: bool,
106}
107
108impl Machine {
109    pub fn new(now_ms: i64) -> Self {
110        Machine {
111            session_known: false,
112            email: None,
113            link: Link::NoSignIn,
114            pull_errors: BTreeMap::new(),
115            status: MuxBusDeliveryStatus { since_ms: now_ms, ..Default::default() },
116            paused_since_ms: None,
117            pause_noted: false,
118            sign_in_notified: false,
119        }
120    }
121
122    pub fn status(&self) -> &MuxBusDeliveryStatus {
123        &self.status
124    }
125
126    /// When the current pause began, if one is open.
127    #[cfg(test)]
128    pub fn paused_since_ms(&self) -> Option<i64> {
129        self.paused_since_ms
130    }
131
132    /// The subscriber found a sign-in to use (`email` when it could read one).
133    pub fn saw_session(&mut self, email: Option<String>, broker: Option<&CredentialState>, now_ms: i64) -> Vec<Effect> {
134        self.session_known = true;
135        if email.as_deref().is_some_and(|e| !e.is_empty()) {
136            self.email = email;
137        }
138        // There is a sign-in now, so the loop is on its way to connecting:
139        // not a lost one (which would notify).
140        if self.link == Link::NoSignIn {
141            self.link = Link::Unreachable("connecting".to_string());
142        }
143        self.update(broker, now_ms)
144    }
145
146    /// The user just signed in (`muxbus.login` saved it): connecting, whatever
147    /// the broker still says about the sign-in this one replaces.
148    pub fn fresh_sign_in(&mut self, email: Option<String>, now_ms: i64) -> Vec<Effect> {
149        self.session_known = true;
150        if email.as_deref().is_some_and(|e| !e.is_empty()) {
151            self.email = email;
152        }
153        self.link = Link::Connecting;
154        self.update(None, now_ms)
155    }
156
157    /// The user signed out on purpose: not a lost sign-in.
158    pub fn signed_out(&mut self, now_ms: i64) -> Vec<Effect> {
159        self.session_known = false;
160        self.email = None;
161        self.link = Link::NoSignIn;
162        self.pull_errors.clear();
163        self.update(None, now_ms)
164    }
165
166    pub fn set_link(&mut self, link: Link, broker: Option<&CredentialState>, now_ms: i64) -> Vec<Effect> {
167        self.link = link;
168        self.update(broker, now_ms)
169    }
170
171    /// The relay answered `agent`'s pull: delivery is healthy as of `now_ms`.
172    pub fn relay_answered(&mut self, agent: &str, broker: Option<&CredentialState>, now_ms: i64) -> Vec<Effect> {
173        self.status.last_ok_ms = Some(now_ms);
174        self.pull_errors.remove(agent);
175        self.update(broker, now_ms)
176    }
177
178    /// `agent`'s pull (fetch or claim) got no usable answer from the relay.
179    pub fn pull_failed(&mut self, agent: &str, reason: String, broker: Option<&CredentialState>, now_ms: i64) -> Vec<Effect> {
180        self.pull_errors.insert(agent.to_string(), reason);
181        self.update(broker, now_ms)
182    }
183
184    /// Forget pull failures of agents no longer subscribed: they won't pull
185    /// again to clear them.
186    pub fn keep_pull_errors_of(&mut self, agents: &[String]) {
187        self.pull_errors.retain(|agent, _| agents.contains(agent));
188    }
189
190    /// Re-derive (the broker may have moved on its own) and release the
191    /// pause note once it is due.
192    pub fn tick(&mut self, broker: Option<&CredentialState>, now_ms: i64) -> Vec<Effect> {
193        let mut out = self.update(broker, now_ms);
194        if let Some(since_ms) = self.paused_since_ms {
195            if !self.pause_noted && self.status.state.is_paused() && now_ms - since_ms >= PAUSE_NOTICE_MS {
196                self.pause_noted = true;
197                out.push(Effect::PauseNote { since_ms, reason: pause_reason(self.status.state) });
198            }
199        }
200        out
201    }
202
203    fn update(&mut self, broker: Option<&CredentialState>, now_ms: i64) -> Vec<Effect> {
204        let link = match (&self.link, self.pull_errors.values().next()) {
205            (Link::Connected, Some(e)) => Link::Unreachable(e.clone()),
206            (link, _) => link.clone(),
207        };
208        let (state, error) = derive(self.session_known, &link, broker);
209        let mut out = Vec::new();
210        let prev = self.status.state;
211        let changed = state != prev || error != self.status.last_error || self.email != self.status.account_email;
212        if state == MuxBusDeliveryState::Connected || prev == MuxBusDeliveryState::Connected {
213            self.status.last_ok_ms = Some(now_ms);
214        }
215        if state != prev {
216            self.status.since_ms = now_ms;
217            if state == MuxBusDeliveryState::NeedsSignIn && !self.sign_in_notified {
218                self.sign_in_notified = true;
219                out.push(Effect::NotifySignInNeeded);
220            }
221            if matches!(state, MuxBusDeliveryState::Connected | MuxBusDeliveryState::SignedOut) && self.sign_in_notified {
222                self.sign_in_notified = false;
223                out.push(Effect::RetractSignInNeeded);
224            }
225            // A sign-out ends a pause the agents were never told about, so a
226            // later sign-in starts a new one rather than dating it from the
227            // old outage. A pause they were told about stays open for its
228            // resume note.
229            if state == MuxBusDeliveryState::SignedOut && !self.pause_noted {
230                self.paused_since_ms = None;
231            }
232            if state.is_paused() && self.paused_since_ms.is_none() {
233                self.paused_since_ms = Some(now_ms);
234            }
235            if state == MuxBusDeliveryState::Connected {
236                if self.paused_since_ms.take().is_some() && self.pause_noted {
237                    out.push(Effect::Resumed { at_ms: now_ms });
238                }
239                self.pause_noted = false;
240            }
241        }
242        self.status.state = state;
243        self.status.last_error = error;
244        self.status.account_email = self.email.clone();
245        if changed {
246            out.insert(0, Effect::Changed(self.status.clone()));
247        }
248        out
249    }
250}
251
252#[cfg(test)]
253mod tests {
254    use super::*;
255    use MuxBusDeliveryState::*;
256
257    const T0: i64 = 1_791_000_000_000;
258
259    fn refused() -> CredentialState {
260        CredentialState::NeedsReauth { since_unix: 0, reason: "invalid_grant".into(), rejected: true }
261    }
262
263    fn piled_up() -> CredentialState {
264        CredentialState::NeedsReauth { since_unix: 0, reason: "network".into(), rejected: false }
265    }
266
267    fn connected_machine() -> Machine {
268        let mut m = Machine::new(T0);
269        m.saw_session(Some("a@b.c".into()), None, T0);
270        m.set_link(Link::Connected, None, T0);
271        m
272    }
273
274    fn kinds(effects: &[Effect]) -> Vec<&'static str> {
275        effects
276            .iter()
277            .map(|e| match e {
278                Effect::Changed(_) => "changed",
279                Effect::NotifySignInNeeded => "notify",
280                Effect::RetractSignInNeeded => "retract",
281                Effect::PauseNote { .. } => "pause",
282                Effect::Resumed { .. } => "resumed",
283            })
284            .collect()
285    }
286
287    #[test]
288    fn each_state_derives_from_its_inputs() {
289        assert_eq!(derive(true, &Link::Connected, Some(&refused())).0, Connected, "an open link wins");
290        assert_eq!(derive(false, &Link::NoSignIn, None).0, SignedOut);
291        assert_eq!(derive(false, &Link::Unreachable("x".into()), Some(&refused())).0, SignedOut);
292        assert_eq!(derive(true, &Link::NoSignIn, None).0, NeedsSignIn, "a sign-in that vanished");
293        assert_eq!(derive(true, &Link::SignInStale("pool".into()), None), (NeedsSignIn, Some("pool".into())));
294        assert_eq!(derive(true, &Link::Unreachable("timeout".into()), None), (Reconnecting, Some("timeout".into())));
295        assert_eq!(derive(true, &Link::Unreachable("x".into()), Some(&piled_up())).0, Reconnecting);
296        let (state, why) = derive(true, &Link::Unreachable("x".into()), Some(&refused()));
297        assert_eq!(state, NeedsSignIn);
298        assert!(why.unwrap().contains("invalid_grant"));
299    }
300
301    #[test]
302    fn a_channel_never_signed_in_stays_signed_out_and_quiet() {
303        let mut m = Machine::new(T0);
304        let fx = m.set_link(Link::NoSignIn, Some(&refused()), T0 + 1);
305        assert!(fx.is_empty(), "{fx:?}");
306        assert_eq!(m.status().state, SignedOut);
307        assert!(m.tick(Some(&refused()), T0 + 10 * PAUSE_NOTICE_MS).is_empty());
308    }
309
310    #[test]
311    fn transitions_record_when_they_happened() {
312        let mut m = connected_machine();
313        assert_eq!(m.status().state, Connected);
314        assert_eq!(m.status().account_email.as_deref(), Some("a@b.c"));
315        m.set_link(Link::Unreachable("closed".into()), None, T0 + 5_000);
316        assert_eq!((m.status().state, m.status().since_ms), (Reconnecting, T0 + 5_000));
317        assert_eq!(m.status().last_ok_ms, Some(T0 + 5_000), "last ok is when it left connected");
318        m.set_link(Link::SignInStale("refused".into()), None, T0 + 9_000);
319        assert_eq!((m.status().state, m.status().since_ms), (NeedsSignIn, T0 + 9_000));
320        m.set_link(Link::Connected, None, T0 + 20_000);
321        assert_eq!((m.status().state, m.status().since_ms), (Connected, T0 + 20_000));
322        let fx = m.signed_out(T0 + 30_000);
323        assert_eq!(m.status().state, SignedOut);
324        assert_eq!(m.status().account_email, None);
325        assert_eq!(kinds(&fx), ["changed"]);
326    }
327
328    #[test]
329    fn the_broker_alone_can_move_reconnecting_to_needs_sign_in() {
330        let mut m = connected_machine();
331        m.set_link(Link::Unreachable("closed".into()), None, T0 + 1);
332        let fx = m.tick(Some(&refused()), T0 + 2);
333        assert_eq!(m.status().state, NeedsSignIn);
334        assert_eq!(kinds(&fx), ["changed", "notify"]);
335    }
336
337    #[test]
338    fn one_notification_per_needs_sign_in_episode() {
339        let mut m = connected_machine();
340        let fx = m.set_link(Link::SignInStale("refused".into()), None, T0 + 1);
341        assert_eq!(kinds(&fx), ["changed", "notify"]);
342        // Flapping inside the episode does not notify again.
343        m.set_link(Link::Unreachable("x".into()), None, T0 + 2);
344        let fx = m.set_link(Link::SignInStale("refused".into()), None, T0 + 3);
345        assert_eq!(kinds(&fx), ["changed"]);
346        assert!(!kinds(&m.tick(None, T0 + 10 * PAUSE_NOTICE_MS)).contains(&"notify"));
347        // Connected ends it and takes the notification down; the next one notifies.
348        let fx = m.set_link(Link::Connected, None, T0 + 20 * PAUSE_NOTICE_MS);
349        assert!(kinds(&fx).contains(&"retract"));
350        let fx = m.set_link(Link::NoSignIn, None, T0 + 21 * PAUSE_NOTICE_MS);
351        assert_eq!(kinds(&fx), ["changed", "notify"]);
352    }
353
354    #[test]
355    fn signing_out_ends_the_sign_in_episode_without_another_notification() {
356        let mut m = connected_machine();
357        m.set_link(Link::SignInStale("refused".into()), None, T0 + 1);
358        let fx = m.signed_out(T0 + 2);
359        assert_eq!(kinds(&fx), ["changed", "retract"]);
360    }
361
362    #[test]
363    fn the_pause_note_waits_two_minutes_and_goes_out_once() {
364        let mut m = connected_machine();
365        let start = T0 + 1_000;
366        m.set_link(Link::Unreachable("closed".into()), None, start);
367        assert!(!kinds(&m.tick(None, start + PAUSE_NOTICE_MS - 1)).contains(&"pause"));
368        let fx = m.tick(None, start + PAUSE_NOTICE_MS);
369        assert_eq!(fx, [Effect::PauseNote { since_ms: start, reason: "AgentMux can't reach MuxBus" }]);
370        assert!(m.tick(None, start + 5 * PAUSE_NOTICE_MS).is_empty(), "once per pause");
371        // The pause carries on into needs-sign-in: still the same pause, no second note.
372        m.set_link(Link::SignInStale("refused".into()), None, start + 6 * PAUSE_NOTICE_MS);
373        assert!(!kinds(&m.tick(None, start + 9 * PAUSE_NOTICE_MS)).contains(&"pause"));
374        let fx = m.set_link(Link::Connected, None, start + 10 * PAUSE_NOTICE_MS);
375        assert!(fx.contains(&Effect::Resumed { at_ms: start + 10 * PAUSE_NOTICE_MS }));
376    }
377
378    #[test]
379    fn the_pause_reason_follows_the_state_at_note_time() {
380        let mut m = connected_machine();
381        m.set_link(Link::SignInStale("refused".into()), None, T0 + 1);
382        let fx = m.tick(None, T0 + 1 + PAUSE_NOTICE_MS);
383        assert!(fx.contains(&Effect::PauseNote { since_ms: T0 + 1, reason: "MuxBus needs a sign-in" }));
384    }
385
386    #[test]
387    fn a_blip_under_two_minutes_tells_the_agents_nothing() {
388        let mut m = connected_machine();
389        m.set_link(Link::Unreachable("closed".into()), None, T0 + 1);
390        m.tick(None, T0 + PAUSE_NOTICE_MS - 10);
391        let fx = m.set_link(Link::Connected, None, T0 + PAUSE_NOTICE_MS - 5);
392        assert_eq!(kinds(&fx), ["changed"], "no resume without a pause note");
393        assert!(m.tick(None, T0 + 10 * PAUSE_NOTICE_MS).is_empty());
394        assert_eq!(m.paused_since_ms(), None);
395    }
396
397    #[test]
398    fn a_pause_survives_a_sign_out_and_resumes_on_the_next_connect() {
399        let mut m = connected_machine();
400        m.set_link(Link::SignInStale("refused".into()), None, T0);
401        m.tick(None, T0 + PAUSE_NOTICE_MS);
402        m.signed_out(T0 + PAUSE_NOTICE_MS + 1);
403        assert!(m.tick(None, T0 + 4 * PAUSE_NOTICE_MS).is_empty(), "signed out: no further note");
404        let fx = m.saw_session(None, None, T0 + 5 * PAUSE_NOTICE_MS);
405        assert_eq!(m.status().state, Reconnecting, "signing in again is not a lost sign-in");
406        assert!(!kinds(&fx).contains(&"notify"));
407        let fx = m.set_link(Link::Connected, None, T0 + 5 * PAUSE_NOTICE_MS);
408        assert!(kinds(&fx).contains(&"resumed"));
409    }
410
411    #[test]
412    fn a_short_pause_ended_by_a_sign_out_is_not_dated_from_later() {
413        let mut m = connected_machine();
414        m.set_link(Link::Unreachable("closed".into()), None, T0);
415        m.signed_out(T0 + PAUSE_NOTICE_MS / 2);
416        assert_eq!(m.paused_since_ms(), None);
417        let later = T0 + 100 * PAUSE_NOTICE_MS;
418        m.saw_session(None, None, later);
419        assert!(!kinds(&m.tick(None, later + 1)).contains(&"pause"), "no note dated from the old outage");
420        let fx = m.tick(None, later + PAUSE_NOTICE_MS);
421        assert!(fx.contains(&Effect::PauseNote { since_ms: later, reason: pause_reason(Reconnecting) }));
422    }
423
424    #[test]
425    fn an_open_link_whose_pulls_fail_is_not_connected() {
426        let mut m = connected_machine();
427        m.pull_failed("a", "the relay answered 503 to a message fetch".into(), None, T0 + 1);
428        assert_eq!(m.status().state, Reconnecting);
429        assert_eq!(m.status().last_error.as_deref(), Some("the relay answered 503 to a message fetch"));
430        let fx = m.tick(None, T0 + 1 + PAUSE_NOTICE_MS);
431        assert!(kinds(&fx).contains(&"pause"), "the agents are told, as for any pause");
432        m.relay_answered("b", None, T0 + 2 * PAUSE_NOTICE_MS - 1);
433        assert_eq!(m.status().state, Reconnecting, "another agent's pull doesn't clear a's failure");
434        let fx = m.relay_answered("a", None, T0 + 2 * PAUSE_NOTICE_MS);
435        assert_eq!(m.status().state, Connected);
436        assert!(kinds(&fx).contains(&"resumed"));
437        // A failing pull while the link itself is down changes nothing more.
438        m.set_link(Link::Unreachable("closed".into()), None, T0 + 3 * PAUSE_NOTICE_MS);
439        m.pull_failed("a", "x".into(), None, T0 + 3 * PAUSE_NOTICE_MS + 1);
440        assert_eq!(m.status().last_error.as_deref(), Some("closed"));
441        // An agent that leaves can't clear its own failure; it is forgotten.
442        m.set_link(Link::Connected, None, T0 + 4 * PAUSE_NOTICE_MS);
443        assert_eq!(m.status().state, Reconnecting);
444        m.keep_pull_errors_of(&["b".to_string()]);
445        m.tick(None, T0 + 4 * PAUSE_NOTICE_MS + 1);
446        assert_eq!(m.status().state, Connected);
447    }
448
449    #[test]
450    fn a_fresh_sign_in_is_not_judged_by_the_refusal_it_replaces() {
451        let mut m = connected_machine();
452        m.set_link(Link::Unreachable("x".into()), Some(&refused()), T0 + 1);
453        assert_eq!(m.status().state, NeedsSignIn);
454        m.signed_out(T0 + 2);
455        let fx = m.fresh_sign_in(Some("a@b.c".into()), T0 + 3);
456        assert_eq!(m.status().state, Reconnecting);
457        assert!(!kinds(&fx).contains(&"notify"));
458        // The broker still holds the old refusal until the loop refreshes.
459        assert_eq!(m.tick(Some(&refused()), T0 + 4).len(), 0);
460        assert_eq!(m.status().state, Reconnecting);
461        m.set_link(Link::Connected, Some(&refused()), T0 + 5);
462        assert_eq!(m.status().state, Connected);
463    }
464
465    #[test]
466    fn an_unchanged_report_changes_nothing() {
467        let mut m = connected_machine();
468        assert!(m.set_link(Link::Connected, None, T0 + 1).is_empty());
469        assert!(m.saw_session(Some("a@b.c".into()), None, T0 + 2).is_empty());
470    }
471}
agentmux_srv/muxbus/delivery_status/machine.rs

agentmux_srv/muxbus/delivery_status/
machine.rs
