agentmux_srv/sagas/
orphan_reaper.rs

1// Copyright 2026, AgentMux Corp.
2// SPDX-License-Identifier: Apache-2.0
3//
4// Orphan reaper — docs/specs/SPEC_AGENT_PANE_CLOSE_GRACEFUL_SHUTDOWN_2026_09_18.md §4.9.
5//
6// An agent block that is in a tab's `block_ids` but in no pane of that tab
7// is a bug, not a state: its process keeps running and nothing on screen
8// can reach it (the Manoz incident, spec §2). The close paths no longer
9// produce one (#3402, #3414), but other ways remain — a create-then-place
10// interrupted between its two steps, a stale whole-tree push
11// (SPEC_PANE_TABS_REDUCER_COMMANDS_2026_09_18.md §2) — and orphans from
12// before those fixes are still on disk. This is the backstop, not the fix.
13//
14// Every sweep finds agent blocks that are unplaced in a tab with a loaded
15// layout. A block is reaped (the ordinary graceful close, `close_pane`) only
16// once it has stayed unplaced for `MIN_ORPHAN_AGE` across sweeps — the
17// window in which a block is legitimately unplaced (created with
18// `pane.open { skip_placement }` and not yet pushed onto a stack, or
19// mid-drag) is well under a second.
20//
21// Never reaped:
22// - non-agent blocks (terminals etc. — narrower scope first; widen once this
23//   has run in the field);
24// - sub-blocks (their parent is a block, not a pane — they sit in a parent's
25//   `subblockids`);
26// - blocks already closing;
27// - blocks in a tab whose layout tree is empty — "no tree yet" can't be told
28//   apart from "not loaded yet" from here.
29
30use std::collections::{HashMap, HashSet};
31use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering};
32use std::time::{Duration, Instant};
33
34use crate::backend::blockcontroller;
35use crate::backend::obj::Block;
36use crate::server::AppState;
37
38/// First sweep waits for the frontend to load and push its layouts.
39const FIRST_SWEEP_DELAY: Duration = Duration::from_secs(120);
40const SWEEP_INTERVAL: Duration = Duration::from_secs(60);
41/// How long a block must stay unplaced before it is reaped.
42pub(crate) const MIN_ORPHAN_AGE: Duration = Duration::from_secs(90);
43
44/// Diagnostic-only counters updated at the end of every `sweep()` — for the
45/// periodic `mem_attribution` log (`sysinfo.rs`) to correlate commit/handle
46/// growth against reaper activity. A `reaper_candidates`/`reaper_first_seen`
47/// that climbs sweep over sweep without a matching rise in
48/// `reaper_reaped_cumulative` would mean the reaper keeps finding (but not
49/// managing to close) the same growing set of blocks — worth seeing
50/// directly rather than inferred from commit numbers alone.
51static LAST_CANDIDATES: AtomicUsize = AtomicUsize::new(0);
52static LAST_FIRST_SEEN_LEN: AtomicUsize = AtomicUsize::new(0);
53static REAPED_CUMULATIVE: AtomicU64 = AtomicU64::new(0);
54
55/// `(candidates found this sweep, first_seen entries carried across sweeps,
56/// cumulative blocks reaped since process start)`.
57#[cfg(any(windows, test))]
58pub fn last_sweep_stats() -> (usize, usize, u64) {
59    (
60        LAST_CANDIDATES.load(Ordering::Relaxed),
61        LAST_FIRST_SEEN_LEN.load(Ordering::Relaxed),
62        REAPED_CUMULATIVE.load(Ordering::Relaxed),
63    )
64}
65
66/// Start the periodic sweep. Called once from `main` after `AppState` exists.
67pub fn install(state: &AppState) {
68    let state = state.clone();
69    tokio::spawn(async move {
70        tokio::time::sleep(FIRST_SWEEP_DELAY).await;
71        let mut first_seen: HashMap<String, Instant> = HashMap::new();
72        loop {
73            let reaped = sweep(&state, &mut first_seen, Instant::now(), MIN_ORPHAN_AGE).await;
74            if !reaped.is_empty() {
75                tracing::warn!(count = reaped.len(), blocks = ?reaped, "orphan_reaper: closed unplaced agent blocks");
76            }
77            tokio::time::sleep(SWEEP_INTERVAL).await;
78        }
79    });
80}
81
82/// One sweep. `first_seen` carries each candidate's first sighting across
83/// sweeps; a block that got placed (or deleted) in between drops out of it.
84/// Returns the block ids it closed.
85pub(crate) async fn sweep(
86    state: &AppState,
87    first_seen: &mut HashMap<String, Instant>,
88    now: Instant,
89    min_age: Duration,
90) -> Vec<String> {
91    let candidates = unplaced_agent_blocks(state).await;
92    let candidate_count = candidates.len();
93    first_seen.retain(|id, _| candidates.contains(id));
94    let mut due = Vec::new();
95    for id in candidates {
96        let seen = *first_seen.entry(id.clone()).or_insert(now);
97        if now.duration_since(seen) >= min_age {
98            due.push(id);
99        }
100    }
101
102    let mut reaped = Vec::new();
103    for id in due {
104        tracing::warn!(block_id = %id, "orphan_reaper: agent block has been in no pane for too long — closing it");
105        match super::close_pane::run(state, vec![id.clone()]).await {
106            Ok(_) => {
107                first_seen.remove(&id);
108                reaped.push(id);
109            }
110            Err(e) => tracing::warn!(block_id = %id, error = %e, "orphan_reaper: close failed; will retry next sweep"),
111        }
112    }
113
114    LAST_CANDIDATES.store(candidate_count, Ordering::Relaxed);
115    LAST_FIRST_SEEN_LEN.store(first_seen.len(), Ordering::Relaxed);
116    if !reaped.is_empty() {
117        REAPED_CUMULATIVE.fetch_add(reaped.len() as u64, Ordering::Relaxed);
118    }
119    reaped
120}
121
122/// Agent blocks that are in their tab's `block_ids` but in no leaf of the
123/// tab's layout, minus everything the module doc lists as never reaped.
124async fn unplaced_agent_blocks(state: &AppState) -> HashSet<String> {
125    let unplaced: Vec<(String, String)> = {
126        let s = state.srv_state.lock().await;
127        let mut out = Vec::new();
128        for tab in s.tabs.values() {
129            let Some(root) = tab.rootnode.as_ref() else { continue };
130            let mut placed = HashSet::new();
131            collect_members(root, &mut placed);
132            out.extend(
133                tab.block_ids
134                    .iter()
135                    .filter(|b| !placed.contains(*b))
136                    .map(|b| (tab.tab_id.clone(), b.clone())),
137            );
138        }
139        out
140    };
141    // A block the backend has queued for placement (`agent.open`, docked
142    // `pane.open`, redock) is in no leaf until a frontend applies the action
143    // and pushes its tree back — indefinitely, if no window is showing that
144    // tab. That is pending placement, not an orphan.
145    let pending: HashSet<String> = unplaced
146        .iter()
147        .map(|(tab_id, _)| tab_id)
148        .collect::<HashSet<_>>()
149        .into_iter()
150        .flat_map(|tab_id| pending_placement_blocks(state, tab_id))
151        .collect();
152    let unplaced: Vec<String> = unplaced
153        .into_iter()
154        .map(|(_, b)| b)
155        .filter(|b| !pending.contains(b))
156        .collect();
157    if unplaced.is_empty() {
158        return HashSet::new();
159    }
160
161    let blocks: Vec<Block> = state.mstore.get_all::<Block>().unwrap_or_default();
162    let sub_blocks: HashSet<&String> = blocks
163        .iter()
164        .filter_map(|b| b.subblockids.as_ref())
165        .flatten()
166        .collect();
167    let agent_blocks: HashSet<&String> = blocks
168        .iter()
169        .filter(|b| crate::backend::obj::meta_get_string(&b.meta, "view", "") == "agent")
170        .map(|b| &b.oid)
171        .collect();
172
173    unplaced
174        .into_iter()
175        .filter(|id| agent_blocks.contains(id) && !sub_blocks.contains(id) && !blockcontroller::is_closing(id))
176        .collect()
177}
178
179/// Blocks named by a queued, not-yet-applied placement action
180/// (`pendingbackendactions`) in this tab's layout row.
181fn pending_placement_blocks(state: &AppState, tab_id: &str) -> Vec<String> {
182    let Ok(tab) = state.mstore.must_get::<crate::backend::obj::Tab>(tab_id) else {
183        return Vec::new();
184    };
185    let Ok(layout) = state.mstore.must_get::<crate::backend::obj::LayoutState>(&tab.layoutstate) else {
186        return Vec::new();
187    };
188    layout
189        .pendingbackendactions
190        .unwrap_or_default()
191        .into_iter()
192        .filter(|a| a.actiontype != "delete")
193        .map(|a| a.blockid)
194        .collect()
195}
196
197fn collect_members(node: &agentmux_common::LayoutNode, out: &mut HashSet<String>) {
198    if let Some(data) = node.data.as_ref() {
199        out.extend(crate::backend::layout::leaf_members(data));
200    }
201    for child in &node.children {
202        collect_members(child, out);
203    }
204}
205
206#[cfg(test)]
207mod tests {
208    use super::*;
209    use crate::server::tests::test_state;
210    use agentmux_common::ipc::{Command, Event};
211
212    async fn dispatch_apply(state: &AppState, cmd: Command) -> Vec<Event> {
213        let events = crate::server::service::dispatch_to_reducer(state, cmd).await;
214        for ev in &events {
215            crate::persist_subscriber::apply_event_to_mstore(ev, &state.mstore).unwrap();
216        }
217        events
218    }
219
220    async fn seed_tab(state: &AppState) -> String {
221        let ws = dispatch_apply(state, Command::CreateWorkspace { name: "w".into() })
222            .await
223            .iter()
224            .find_map(|e| match e {
225                Event::WorkspaceCreated { workspace_id, .. } => Some(workspace_id.clone()),
226                _ => None,
227            })
228            .unwrap();
229        dispatch_apply(state, Command::CreateTab { workspace_id: ws, name: "t".into() })
230            .await
231            .iter()
232            .find_map(|e| match e {
233                Event::TabCreated { tab_id, .. } => Some(tab_id.clone()),
234                _ => None,
235            })
236            .unwrap()
237    }
238
239    async fn seed_block(state: &AppState, tab_id: &str, view: &str) -> String {
240        dispatch_apply(
241            state,
242            Command::CreateBlock {
243                tab_id: tab_id.to_string(),
244                meta: serde_json::json!({ "view": view }),
245            },
246        )
247        .await
248        .iter()
249        .find_map(|e| match e {
250            Event::BlockCreated { block_id, .. } => Some(block_id.clone()),
251            _ => None,
252        })
253        .unwrap()
254    }
255
256    async fn place(state: &AppState, tab_id: &str, block_ids: &[&String]) {
257        let children = block_ids
258            .iter()
259            .map(|b| agentmux_common::LayoutNode {
260                id: format!("n-{b}"),
261                data: Some(agentmux_common::LayoutNodeData {
262                    block_id: (*b).clone(),
263                    ..Default::default()
264                }),
265                ..Default::default()
266            })
267            .collect();
268        dispatch_apply(
269            state,
270            Command::LayoutSetTree {
271                tab_id: tab_id.to_string(),
272                new_tree: Some(agentmux_common::LayoutNode {
273                    id: "root".into(),
274                    children,
275                    ..Default::default()
276                }),
277                correlation_id: String::new(),
278                slices: None,
279            },
280        )
281        .await;
282    }
283
284    /// The incident shape: an agent block in the tab, in no pane. It is left
285    /// alone until it has been unplaced for the minimum age, then closed.
286    #[tokio::test]
287    async fn an_unplaced_agent_block_is_reaped_only_after_the_minimum_age() {
288        let state = test_state();
289        let tab = seed_tab(&state).await;
290        let placed = seed_block(&state, &tab, "agent").await;
291        let orphan = seed_block(&state, &tab, "agent").await;
292        place(&state, &tab, &[&placed]).await;
293
294        let t0 = Instant::now();
295        let mut seen = HashMap::new();
296        let age = Duration::from_secs(90);
297
298        assert!(sweep(&state, &mut seen, t0, age).await.is_empty(), "first sighting: too young");
299        assert!(state.srv_state.lock().await.blocks.contains_key(&orphan));
300
301        let reaped = sweep(&state, &mut seen, t0 + age, age).await;
302        assert_eq!(reaped, vec![orphan.clone()]);
303        let s = state.srv_state.lock().await;
304        assert!(!s.blocks.contains_key(&orphan), "orphan closed");
305        assert!(s.blocks.contains_key(&placed), "placed block untouched");
306    }
307
308    /// `last_sweep_stats()`'s cumulative-reaped counter is process-wide
309    /// (shared with every other test in this binary, since it's a plain
310    /// static) and other tests in this module call `sweep` concurrently —
311    /// so this only asserts the one property that's safe under that:
312    /// monotonic, non-decreasing, and increases by at least what THIS
313    /// sweep reaped. It deliberately does not assert an exact value for
314    /// `candidates`/`first_seen`, which the last-writer-wins semantics of a
315    /// shared static make inherently racy to pin down under parallel tests.
316    #[tokio::test]
317    async fn last_sweep_stats_reaped_cumulative_is_monotonic_across_a_real_reap() {
318        let state = test_state();
319        let tab = seed_tab(&state).await;
320        // A placed block is required for the tab to have a real layout
321        // tree at all — an empty tree is indistinguishable from "not
322        // loaded yet" and the whole tab is skipped (see
323        // `a_tab_with_no_layout_tree_is_skipped`), so an unplaced block on
324        // its own would never even become a candidate.
325        let placed = seed_block(&state, &tab, "agent").await;
326        let orphan = seed_block(&state, &tab, "agent").await;
327        place(&state, &tab, &[&placed]).await;
328
329        let (_, _, before) = last_sweep_stats();
330        let t0 = Instant::now();
331        let mut seen = HashMap::new();
332        let age = Duration::from_secs(90);
333        sweep(&state, &mut seen, t0, age).await; // first sighting, not yet due
334        let reaped = sweep(&state, &mut seen, t0 + age, age).await;
335        assert_eq!(reaped, vec![orphan]);
336
337        let (_, _, after) = last_sweep_stats();
338        assert!(
339            after >= before + 1,
340            "cumulative reaped count must have grown by at least the one block this sweep reaped: before={before}, after={after}"
341        );
342    }
343
344    /// A block that gets placed between sweeps (skip_placement then a stack
345    /// push, or a drag) starts over — never reaped.
346    #[tokio::test]
347    async fn a_block_placed_between_sweeps_is_forgotten() {
348        let state = test_state();
349        let tab = seed_tab(&state).await;
350        let a = seed_block(&state, &tab, "agent").await;
351        let fresh = seed_block(&state, &tab, "agent").await;
352        place(&state, &tab, &[&a]).await;
353
354        let t0 = Instant::now();
355        let mut seen = HashMap::new();
356        let age = Duration::from_secs(90);
357        sweep(&state, &mut seen, t0, age).await;
358        assert!(seen.contains_key(&fresh));
359
360        place(&state, &tab, &[&a, &fresh]).await;
361        assert!(sweep(&state, &mut seen, t0 + age, age).await.is_empty());
362        assert!(!seen.contains_key(&fresh), "placed → dropped from the watch list");
363        assert!(state.srv_state.lock().await.blocks.contains_key(&fresh));
364    }
365
366    #[tokio::test]
367    async fn non_agent_blocks_and_sub_blocks_are_never_reaped() {
368        let state = test_state();
369        let tab = seed_tab(&state).await;
370        let placed = seed_block(&state, &tab, "agent").await;
371        let term = seed_block(&state, &tab, "term").await;
372        let sub = seed_block(&state, &tab, "agent").await;
373        place(&state, &tab, &[&placed]).await;
374        // Make `sub` a sub-block of `placed`.
375        let mut parent = state.mstore.must_get::<Block>(&placed).unwrap();
376        parent.subblockids = Some(vec![sub.clone()]);
377        state.mstore.update(&mut parent).unwrap();
378
379        let t0 = Instant::now();
380        let mut seen = HashMap::new();
381        let age = Duration::from_secs(90);
382        sweep(&state, &mut seen, t0, age).await;
383        assert!(sweep(&state, &mut seen, t0 + age, age).await.is_empty());
384        let s = state.srv_state.lock().await;
385        assert!(s.blocks.contains_key(&term) && s.blocks.contains_key(&sub));
386    }
387
388    /// `agent.open` queues the new block's placement for the frontend; with
389    /// no window showing the tab it stays queued. Not an orphan.
390    #[tokio::test]
391    async fn a_block_with_a_queued_placement_is_not_reaped() {
392        let state = test_state();
393        let tab = seed_tab(&state).await;
394        let placed = seed_block(&state, &tab, "agent").await;
395        let opening = seed_block(&state, &tab, "agent").await;
396        place(&state, &tab, &[&placed]).await;
397        crate::server::service::layout_helpers::queue_target_layout_insert(&state, &tab, &opening)
398            .await
399            .unwrap();
400
401        let t0 = Instant::now();
402        let mut seen = HashMap::new();
403        let age = Duration::from_secs(90);
404        sweep(&state, &mut seen, t0, age).await;
405        assert!(sweep(&state, &mut seen, t0 + age, age).await.is_empty());
406        assert!(state.srv_state.lock().await.blocks.contains_key(&opening));
407    }
408
409    /// No tree at all can't be told apart from "layout not loaded yet".
410    #[tokio::test]
411    async fn a_tab_with_no_layout_tree_is_skipped() {
412        let state = test_state();
413        let tab = seed_tab(&state).await;
414        let a = seed_block(&state, &tab, "agent").await;
415
416        let t0 = Instant::now();
417        let mut seen = HashMap::new();
418        let age = Duration::from_secs(90);
419        sweep(&state, &mut seen, t0, age).await;
420        assert!(sweep(&state, &mut seen, t0 + age, age).await.is_empty());
421        assert!(state.srv_state.lock().await.blocks.contains_key(&a));
422    }
423
424    #[tokio::test]
425    async fn a_background_stack_member_counts_as_placed() {
426        let state = test_state();
427        let tab = seed_tab(&state).await;
428        let visible = seed_block(&state, &tab, "agent").await;
429        let background = seed_block(&state, &tab, "agent").await;
430        dispatch_apply(
431            &state,
432            Command::LayoutSetTree {
433                tab_id: tab.clone(),
434                new_tree: Some(agentmux_common::LayoutNode {
435                    id: "pane".into(),
436                    data: Some(agentmux_common::LayoutNodeData {
437                        block_id: visible.clone(),
438                        block_stack: vec![visible.clone(), background.clone()],
439                        active_block_id: visible.clone(),
440                        ..Default::default()
441                    }),
442                    ..Default::default()
443                }),
444                correlation_id: String::new(),
445                slices: None,
446            },
447        )
448        .await;
449
450        let t0 = Instant::now();
451        let mut seen = HashMap::new();
452        let age = Duration::from_secs(90);
453        sweep(&state, &mut seen, t0, age).await;
454        assert!(sweep(&state, &mut seen, t0 + age, age).await.is_empty());
455        assert!(state.srv_state.lock().await.blocks.contains_key(&background));
456    }
457}
agentmux_srv/sagas/orphan_reaper.rs

agentmux_srv/sagas/
orphan_reaper.rs
