agentmux_srv/backend/
agent_state.rs

1// Copyright 2026, AgentMux Corp.
2// SPDX-License-Identifier: Apache-2.0
3
4//! Whether each agent is working, waiting on the user, idle, stopped or
5//! failed, and since when: the `agent_status` the LAN fleet feed,
6//! `GET /agentmux/reactive/agent-names` and (through the feed's snapshot) the
7//! cloud presence record carry (agentmux-mobile's
8//! SPEC_AGENT_STATUS_AND_LIVE_PANE_FEED_2026_10_07 §3, §13.1).
9//!
10//! [`decide`] is the rule, pure; [`current_state`] reads its inputs from the
11//! block's controller and the block's meta; a [`SinceTracker`] remembers when
12//! each block's state began. Every caller goes through
13//! [`agent_status_of_block`] or a [`SinceTracker`] of its own, so the feed
14//! and the names route agree on `since_ms`.
15
16use std::collections::HashMap;
17use std::sync::LazyLock;
18
19use serde::{Deserialize, Serialize};
20
21use crate::backend::blockcontroller;
22use crate::backend::storage::store::Store;
23
24/// Block meta the agent pane sets while a question waits for the user.
25pub const META_AWAITING_USER: &str = "term:awaiting_user";
26
27/// An agent's state as the wire reports it.
28#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, ts_rs::TS)]
29#[ts(export, export_to = "../../../frontend/types/rpc/")]
30#[serde(rename_all = "lowercase")]
31pub enum AgentState {
32    Working,
33    /// A turn is running and a question waits for the user.
34    Waiting,
35    Idle,
36    /// The process exited cleanly.
37    Stopped,
38    /// The process exited with an error.
39    Error,
40}
41
42impl AgentState {
43    pub fn as_str(self) -> &'static str {
44        match self {
45            Self::Working => "working",
46            Self::Waiting => "waiting",
47            Self::Idle => "idle",
48            Self::Stopped => "stopped",
49            Self::Error => "error",
50        }
51    }
52}
53
54/// A state and when it began (unix ms, this machine's clock).
55#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
56pub struct AgentStatus {
57    pub state: AgentState,
58    pub since_ms: u64,
59}
60
61/// What [`decide`] looks at, as read from the controller and the block.
62#[derive(Debug, Clone, Copy, Default)]
63pub struct StateInputs<'a> {
64    pub controller_type: &'a str,
65    /// `BlockControllerRuntimeStatus::turn_active`; an absent field is false.
66    pub turn_active: bool,
67    /// `shellprocstatus`: `init`, `running` or `done`.
68    pub process_status: &'a str,
69    pub exit_code: i32,
70    pub awaiting_user: bool,
71}
72
73/// The state, or `None` when nothing tracks this agent's turns (a terminal
74/// pane: `shell` / `cmd`, or any controller that isn't an agent's) or its
75/// process state is unknown. Saying nothing beats a wrong `idle`.
76///
77/// A turn wins over the process state, as in the process broker's
78/// `lifecycle_from`. The `subprocess` controller runs one process per turn,
79/// so its process exiting cleanly between turns is `idle`, not `stopped`.
80pub fn decide(inputs: &StateInputs) -> Option<AgentState> {
81    let per_turn_process = match inputs.controller_type {
82        blockcontroller::BLOCK_CONTROLLER_SUBPROCESS => true,
83        blockcontroller::BLOCK_CONTROLLER_PERSISTENT
84        | blockcontroller::BLOCK_CONTROLLER_ACP
85        | blockcontroller::BLOCK_CONTROLLER_APP_SERVER => false,
86        _ => return None,
87    };
88    if inputs.turn_active {
89        return Some(if inputs.awaiting_user { AgentState::Waiting } else { AgentState::Working });
90    }
91    match inputs.process_status {
92        blockcontroller::STATUS_RUNNING | blockcontroller::STATUS_INIT => Some(AgentState::Idle),
93        blockcontroller::STATUS_DONE if inputs.exit_code != 0 => Some(AgentState::Error),
94        blockcontroller::STATUS_DONE if per_turn_process => Some(AgentState::Idle),
95        blockcontroller::STATUS_DONE => Some(AgentState::Stopped),
96        _ => None,
97    }
98}
99
100/// The block's state now: its controller's runtime status and the block's
101/// `term:awaiting_user`. `None` without a controller, or when the block
102/// can't be read.
103pub fn current_state(mstore: &Store, block_id: &str) -> Option<AgentState> {
104    let controller = blockcontroller::get_controller(block_id)?;
105    let status = controller.get_runtime_status();
106    let block = mstore.get::<crate::backend::obj::Block>(block_id).ok()??;
107    decide(&StateInputs {
108        controller_type: controller.controller_type(),
109        turn_active: status.turn_active,
110        process_status: &status.shellprocstatus,
111        exit_code: status.shellprocexitcode,
112        awaiting_user: crate::backend::obj::meta_get_bool(&block.meta, META_AWAITING_USER, false),
113    })
114}
115
116/// When each block's current state began. A state seen for the first time
117/// (srv start, a new agent) begins when it is seen; a state that changes
118/// begins again; `None` forgets the block, so a stopped and restarted agent
119/// does not inherit an old start.
120#[derive(Default)]
121pub struct SinceTracker {
122    by_block: parking_lot::Mutex<HashMap<String, AgentStatus>>,
123}
124
125impl SinceTracker {
126    /// Record `state` for `block_id` as seen at `now_ms`; returns it with the
127    /// time it began.
128    pub fn record(&self, block_id: &str, state: Option<AgentState>, now_ms: u64) -> Option<AgentStatus> {
129        let mut by_block = self.by_block.lock();
130        let Some(state) = state else {
131            by_block.remove(block_id);
132            return None;
133        };
134        match by_block.get(block_id) {
135            Some(known) if known.state == state => Some(*known),
136            _ => {
137                let status = AgentStatus { state, since_ms: now_ms };
138                by_block.insert(block_id.to_string(), status);
139                Some(status)
140            }
141        }
142    }
143
144    /// Drop every block `keep` rejects (no longer listed).
145    pub fn retain(&self, keep: impl Fn(&str) -> bool) {
146        self.by_block.lock().retain(|block_id, _| keep(block_id));
147    }
148
149    #[cfg(test)]
150    pub fn len(&self) -> usize {
151        self.by_block.lock().len()
152    }
153}
154
155static TRACKER: LazyLock<SinceTracker> = LazyLock::new(SinceTracker::default);
156
157/// The tracker the live fleet source and the names route share.
158pub fn global_tracker() -> &'static SinceTracker {
159    &TRACKER
160}
161
162/// The agent on `block_id`: its state now and since when.
163pub fn agent_status_of_block(mstore: &Store, block_id: &str) -> Option<AgentStatus> {
164    TRACKER.record(block_id, current_state(mstore, block_id), agentmux_common::time::now_ms_u64())
165}
166
167#[cfg(test)]
168pub(crate) mod test_support {
169    use std::any::Any;
170    use std::sync::Arc;
171
172    use crate::backend::blockcontroller::{BlockControllerRuntimeStatus, BlockInputUnion, Controller};
173    use crate::backend::obj::MetaMapType;
174
175    /// A registered controller that reports a fixed runtime status.
176    pub struct StubController {
177        pub block_id: String,
178        pub controller_type: String,
179        pub status: BlockControllerRuntimeStatus,
180    }
181
182    impl Controller for StubController {
183        fn start(&self, _: MetaMapType, _: Option<serde_json::Value>, _: bool) -> Result<(), String> {
184            Ok(())
185        }
186        fn stop(&self, _: bool, _: &str) -> Result<(), String> {
187            Ok(())
188        }
189        fn get_runtime_status(&self) -> BlockControllerRuntimeStatus {
190            self.status.clone()
191        }
192        fn send_input(&self, _: BlockInputUnion, _: Option<u64>) -> Result<(), String> {
193            Ok(())
194        }
195        fn controller_type(&self) -> &str {
196            &self.controller_type
197        }
198        fn block_id(&self) -> &str {
199            &self.block_id
200        }
201        fn as_any(&self) -> &dyn Any {
202            self
203        }
204    }
205
206    /// Register a stub on `block_id`; remove it with
207    /// `blockcontroller::delete_controller`.
208    pub fn register_stub(block_id: &str, controller_type: &str, process_status: &str, turn_active: bool) {
209        crate::backend::blockcontroller::register_controller(
210            block_id,
211            Arc::new(StubController {
212                block_id: block_id.to_string(),
213                controller_type: controller_type.to_string(),
214                status: BlockControllerRuntimeStatus {
215                    blockid: block_id.to_string(),
216                    shellprocstatus: process_status.to_string(),
217                    turn_active,
218                    ..Default::default()
219                },
220            }),
221        );
222    }
223}
224
225#[cfg(test)]
226mod tests {
227    use super::*;
228    use crate::backend::blockcontroller::{
229        BLOCK_CONTROLLER_ACP, BLOCK_CONTROLLER_APP_SERVER, BLOCK_CONTROLLER_CMD, BLOCK_CONTROLLER_PERSISTENT,
230        BLOCK_CONTROLLER_SHELL, BLOCK_CONTROLLER_SUBPROCESS, STATUS_DONE, STATUS_INIT,
231        STATUS_RUNNING,
232    };
233
234    fn inputs<'a>(controller_type: &'a str, turn_active: bool, process_status: &'a str) -> StateInputs<'a> {
235        StateInputs { controller_type, turn_active, process_status, exit_code: 0, awaiting_user: false }
236    }
237
238    #[test]
239    fn one_case_per_state() {
240        let p = BLOCK_CONTROLLER_PERSISTENT;
241        assert_eq!(decide(&inputs(p, true, STATUS_RUNNING)), Some(AgentState::Working));
242        assert_eq!(
243            decide(&StateInputs { awaiting_user: true, ..inputs(p, true, STATUS_RUNNING) }),
244            Some(AgentState::Waiting)
245        );
246        assert_eq!(decide(&inputs(p, false, STATUS_RUNNING)), Some(AgentState::Idle));
247        assert_eq!(decide(&inputs(p, false, STATUS_INIT)), Some(AgentState::Idle));
248        assert_eq!(decide(&inputs(p, false, STATUS_DONE)), Some(AgentState::Stopped));
249        assert_eq!(
250            decide(&StateInputs { exit_code: 1, ..inputs(p, false, STATUS_DONE) }),
251            Some(AgentState::Error)
252        );
253    }
254
255    #[test]
256    fn a_question_outside_a_turn_is_not_waiting() {
257        let left_over = StateInputs { awaiting_user: true, ..inputs(BLOCK_CONTROLLER_PERSISTENT, false, STATUS_RUNNING) };
258        assert_eq!(decide(&left_over), Some(AgentState::Idle));
259    }
260
261    #[test]
262    fn absent_turn_active_means_no_turn() {
263        // A turn that ended is an absent field: it deserializes to false.
264        let status: blockcontroller::BlockControllerRuntimeStatus =
265            serde_json::from_value(serde_json::json!({"blockid": "b", "shellprocstatus": "running"})).unwrap();
266        assert!(!status.turn_active);
267        assert_eq!(
268            decide(&inputs(BLOCK_CONTROLLER_PERSISTENT, status.turn_active, &status.shellprocstatus)),
269            Some(AgentState::Idle)
270        );
271    }
272
273    #[test]
274    fn terminal_panes_and_unknown_controllers_report_nothing() {
275        for t in [BLOCK_CONTROLLER_SHELL, BLOCK_CONTROLLER_CMD, "", "future"] {
276            assert_eq!(decide(&inputs(t, false, STATUS_RUNNING)), None, "{t}");
277            assert_eq!(decide(&inputs(t, true, STATUS_RUNNING)), None, "{t} mid-turn");
278        }
279    }
280
281    #[test]
282    fn every_agent_controller_reports_a_state() {
283        for t in [BLOCK_CONTROLLER_PERSISTENT, BLOCK_CONTROLLER_ACP, BLOCK_CONTROLLER_APP_SERVER, BLOCK_CONTROLLER_SUBPROCESS] {
284            assert_eq!(decide(&inputs(t, true, STATUS_RUNNING)), Some(AgentState::Working), "{t}");
285        }
286    }
287
288    #[test]
289    fn an_unknown_process_status_reports_nothing() {
290        assert_eq!(decide(&inputs(BLOCK_CONTROLLER_PERSISTENT, false, "")), None);
291        assert_eq!(decide(&inputs(BLOCK_CONTROLLER_PERSISTENT, false, "weird")), None);
292    }
293
294    #[test]
295    fn a_per_turn_process_exiting_cleanly_is_idle() {
296        let s = BLOCK_CONTROLLER_SUBPROCESS;
297        assert_eq!(decide(&inputs(s, false, STATUS_DONE)), Some(AgentState::Idle));
298        assert_eq!(decide(&StateInputs { exit_code: 2, ..inputs(s, false, STATUS_DONE) }), Some(AgentState::Error));
299    }
300
301    #[test]
302    fn states_serialize_lowercase() {
303        for s in [AgentState::Working, AgentState::Waiting, AgentState::Idle, AgentState::Stopped, AgentState::Error] {
304            assert_eq!(serde_json::to_value(s).unwrap(), serde_json::json!(s.as_str()));
305        }
306        assert_eq!(
307            serde_json::to_value(AgentStatus { state: AgentState::Working, since_ms: 7 }).unwrap(),
308            serde_json::json!({"state": "working", "since_ms": 7})
309        );
310    }
311
312    #[test]
313    fn since_is_kept_while_the_state_holds_and_reset_on_a_change() {
314        let t = SinceTracker::default();
315        let working = Some(AgentState::Working);
316        assert_eq!(t.record("b", working, 100).unwrap().since_ms, 100, "first seen");
317        assert_eq!(t.record("b", working, 200).unwrap().since_ms, 100, "unchanged");
318        assert_eq!(
319            t.record("b", Some(AgentState::Idle), 300),
320            Some(AgentStatus { state: AgentState::Idle, since_ms: 300 })
321        );
322        assert_eq!(t.record("b", working, 400).unwrap().since_ms, 400, "back to working starts again");
323    }
324
325    #[test]
326    fn no_state_forgets_the_block() {
327        let t = SinceTracker::default();
328        t.record("b", Some(AgentState::Idle), 100);
329        assert_eq!(t.record("b", None, 200), None);
330        assert_eq!(t.len(), 0);
331        assert_eq!(t.record("b", Some(AgentState::Idle), 300).unwrap().since_ms, 300, "not the old start");
332    }
333
334    #[test]
335    fn retain_drops_unlisted_blocks() {
336        let t = SinceTracker::default();
337        t.record("a", Some(AgentState::Idle), 1);
338        t.record("b", Some(AgentState::Idle), 1);
339        t.retain(|b| b == "a");
340        assert_eq!(t.len(), 1);
341        assert_eq!(t.record("a", Some(AgentState::Idle), 5).unwrap().since_ms, 1);
342    }
343
344    fn store() -> Store {
345        Store::open_in_memory().expect("in-memory store")
346    }
347
348    fn insert_block(store: &Store, meta: &[(&str, serde_json::Value)]) -> String {
349        let mut block = crate::backend::obj::Block { oid: uuid::Uuid::new_v4().to_string(), ..Default::default() };
350        for (k, v) in meta {
351            block.meta.insert(k.to_string(), v.clone());
352        }
353        store.insert(&mut block).expect("insert block");
354        block.oid
355    }
356
357    #[test]
358    fn current_state_reads_the_controller_and_the_block() {
359        let store = store();
360        let asking = insert_block(&store, &[(META_AWAITING_USER, serde_json::json!(true))]);
361        test_support::register_stub(&asking, BLOCK_CONTROLLER_PERSISTENT, STATUS_RUNNING, true);
362        assert_eq!(current_state(&store, &asking), Some(AgentState::Waiting));
363
364        let quiet = insert_block(&store, &[]);
365        test_support::register_stub(&quiet, BLOCK_CONTROLLER_PERSISTENT, STATUS_RUNNING, false);
366        assert_eq!(current_state(&store, &quiet), Some(AgentState::Idle));
367
368        let terminal = insert_block(&store, &[]);
369        test_support::register_stub(&terminal, BLOCK_CONTROLLER_SHELL, STATUS_RUNNING, false);
370        assert_eq!(current_state(&store, &terminal), None);
371
372        let no_controller = insert_block(&store, &[]);
373        assert_eq!(current_state(&store, &no_controller), None);
374
375        let unreadable = format!("missing-{}", uuid::Uuid::new_v4());
376        test_support::register_stub(&unreadable, BLOCK_CONTROLLER_PERSISTENT, STATUS_RUNNING, true);
377        assert_eq!(current_state(&store, &unreadable), None, "the block can't be read");
378
379        for b in [&asking, &quiet, &terminal, &unreadable] {
380            blockcontroller::delete_controller(b);
381        }
382    }
383}
agentmux_srv/backend/agent_state.rs

agentmux_srv/backend/
agent_state.rs
