agentmux_srv/backend/blockcontroller/persistent/status.rs
1// Copyright 2026, AgentMux Corp.
2// SPDX-License-Identifier: Apache-2.0
3
4//! Runtime status: the `BlockControllerRuntimeStatus` snapshot, its publication
5//! to the frontend, the status heartbeat, and turn-activity marking.
6
7use super::*;
8
9impl PersistentSubprocessController {
10 pub(super) fn set_status(inner: &mut PersistentInner, status: &str) {
11 inner.proc_status = status.to_string();
12 inner.status_version += 1;
13 // The argv belongs to a live process; once it is gone there is nothing
14 // to compare a selection against.
15 if status != STATUS_RUNNING {
16 inner.spawn_runtime = None;
17 inner.effective_runtime = None;
18 }
19 }
20
21 pub(super) fn get_status_snapshot(&self) -> BlockControllerRuntimeStatus {
22 let inner = self.inner.lock().unwrap();
23 super::super::agent_runtime_status(
24 &self.block_id,
25 inner.status_version,
26 &inner.proc_status,
27 inner.proc_exit_code,
28 self.health_monitor.is_active_turn(),
29 )
30 }
31
32 pub(super) fn publish_status(&self) {
33 if let Some(ref broker) = self.broker {
34 let status = self.get_status_snapshot();
35 super::super::publish_controller_status(broker, &status);
36 // Alongside every status change, so the two cannot disagree about
37 // whether a process is running: what it was spawned with, and
38 // whether a restart is already on its way.
39 publish_runtime_event(&self.inner, broker, &self.block_id);
40 }
41 }
42
43 /// Announce only what the process runs (or that none does), without
44 /// repeating the controller status. For the paths that move the status off
45 /// "running" without publishing it themselves (`stop`).
46 pub(super) fn publish_runtime(&self) {
47 if let Some(ref broker) = self.broker {
48 publish_runtime_event(&self.inner, broker, &self.block_id);
49 }
50 }
51
52 /// Periodic (low-frequency) republish of the current controllerstatus
53 /// while a turn is active — a self-healing backstop independent of
54 /// `publish_controller_status`'s `persist: 1` (which only helps a
55 /// reconnecting subscriber) and the frontend's focus-triggered reconcile
56 /// (which only fires on a background→foreground transition). If a single
57 /// live push is missed for some other reason (e.g. a throttled/backgrounded
58 /// renderer coalescing WS messages) while the window stays foregrounded
59 /// and connected the whole time, nothing else corrects it until the turn
60 /// actually ends. This shrinks that window from "forever" to at most one
61 /// heartbeat interval. `HEARTBEAT_SECS` is well below the frontend's own
62 /// `STUCK_THRESHOLD_MS` (45s, diagnostic-only) and `LIVENESS_RECOVERY_MS`
63 /// (180s, force-recovery) so a missed push self-heals long before either
64 /// of those fire. See REPORT_LOGIN_PERSIST_FAILURE_AND_STUCK_WORKING_2026_07_27.md
65 /// §4 item 5. Builds the snapshot with `agent_runtime_status` rather than
66 /// `get_status_snapshot`, since the spawned task only holds cloned `Arc`s,
67 /// not `&self`.
68 ///
69 /// A latent duplicate-loop race is an existing, already-accepted
70 /// contract (reagent P2 on the PR that introduced this function): if a
71 /// turn ends and a new one starts again within one
72 /// `HEARTBEAT_SECS` window, the old loop hasn't yet woken up to observe
73 /// `is_active_turn() == false` and break, so both the old and new loop
74 /// can run concurrently for that window. Harmless — `publish_status`
75 /// republishing the same (or a slightly stale) snapshot twice is
76 /// idempotent from the frontend's point of view — so this is accepted
77 /// rather than fixed, matching the pre-existing pattern.
78 pub(super) fn spawn_status_heartbeat(&self) {
79 const HEARTBEAT_SECS: u64 = 20;
80 self.spawn_status_heartbeat_with_interval(tokio::time::Duration::from_secs(HEARTBEAT_SECS));
81 }
82
83 /// Interval parameterized out of `spawn_status_heartbeat` so a test can
84 /// drive it with a short interval instead of waiting out `HEARTBEAT_SECS`.
85 pub(super) fn spawn_status_heartbeat_with_interval(&self, heartbeat_interval: tokio::time::Duration) {
86 let inner = Arc::clone(&self.inner);
87 let block_id = self.block_id.clone();
88 let broker = self.broker.clone();
89 let health_monitor = Arc::clone(&self.health_monitor);
90 tokio::spawn(async move {
91 let Some(broker) = broker else { return };
92 let mut interval = tokio::time::interval(heartbeat_interval);
93 interval.tick().await; // first tick is immediate; publish_status() already ran at turn start
94 loop {
95 interval.tick().await;
96 let still_active = health_monitor.is_active_turn();
97 let status = {
98 let g = inner.lock().unwrap();
99 super::super::agent_runtime_status(
100 &block_id,
101 g.status_version,
102 &g.proc_status,
103 g.proc_exit_code,
104 still_active,
105 )
106 };
107 super::super::publish_controller_status(&broker, &status);
108 // Publish the final `turn_active: false` snapshot BEFORE
109 // exiting, not just break silently — reagent P1: this
110 // heartbeat exists specifically to backstop a missed live
111 // turn-end push (the exact stuck-"Working"-forever bug this
112 // PR fixes). Breaking without this last publish meant the
113 // one case it's most needed for — the terminal push being
114 // the one that got dropped — was the one case it didn't
115 // help.
116 if !still_active {
117 break;
118 }
119 }
120 });
121 }
122
123 /// The CLI started a pass by itself and the tracker has marked it active
124 /// (`mark_turn_active_from_cli`): publish the flip and keep the status
125 /// heartbeat going for it, as for a pass srv started.
126 pub(super) fn cli_started_pass(&self) {
127 self.spawn_status_heartbeat();
128 self.publish_status();
129 }
130
131 /// Marks a turn active (re-arming the status heartbeat only if it was
132 /// previously idle) and publishes the resulting status flip. Shared by
133 /// `send_message` and `retry_after_resume_failure` — both represent "a
134 /// user message is about to be delivered," just via different spawn
135 /// paths. The heartbeat is re-armed conditionally: a mid-turn steering
136 /// send already has one running, so re-spawning on every call would
137 /// leak duplicate heartbeat tasks.
138 pub(super) fn mark_turn_active_and_publish(&self) {
139 let was_active = self.health_monitor.mark_turn_active_for_queued();
140 if !was_active {
141 self.spawn_status_heartbeat();
142 }
143 self.publish_status();
144 }
145}
146
147/// Announce what the process runs, from state the caller already holds locked.
148/// `restart_pending` covers both a restart deferred to the end of the turn and
149/// one already committed.
150pub(super) fn announce_runtime(broker: &crate::backend::mps::Broker, block_id: &str, inner: &PersistentInner) {
151 crate::backend::agent_runtime::publish_agent_runtime(
152 broker,
153 block_id,
154 inner.spawn_runtime.as_ref(),
155 inner.restart_when_idle || inner.restart_pending,
156 inner.effective_runtime.as_ref(),
157 );
158}
159
160/// [`announce_runtime`], taking the lock itself. EVERY path that moves the
161/// status off "running" must call one of the two: the menu judges its
162/// selection against the LAST announcement, so a process that ended without
163/// one would go on being judged (ReAgent P1 on #4149).
164pub(super) fn publish_runtime_event(
165 inner: &Mutex<PersistentInner>,
166 broker: &crate::backend::mps::Broker,
167 block_id: &str,
168) {
169 announce_runtime(broker, block_id, &inner.lock().unwrap());
170}