agentmux_srv/backend/blockcontroller/
session_recovery.rs

1// Copyright 2026, AgentMux Corp.
2// SPDX-License-Identifier: Apache-2.0
3
4//! Session recovery after unclean shutdown (Phase 4.2 — ultra-long-sessions).
5//!
6//! When a persistent agent subprocess is running and the server is killed
7//! (crash, OS reboot, task-kill), the subprocess dies with it but the block's
8//! session history is preserved in FileStore. On next startup, we want to:
9//!
10//!   1. Detect that a session was running when the server died, so we can
11//!      surface it to the user as "this session was interrupted".
12//!   2. Let the user resume: the persistent controller already supports
13//!      `--resume <session_id>` on the next input, so the recovery UX is
14//!      simply "click resume, type your next message, get picked up mid-flight".
15//!
16//! The mechanism is a single meta key: `session:active_pid` (the PID). It's
17//! set on subprocess spawn, cleared on clean exit (graceful or killed). If
18//! this flag is still set when the server boots, the process is definitely
19//! gone (old PID from a dead process), so we transfer it to
20//! `session:was_interrupted = true`.
21//!
22//! `session:was_interrupted` is a frontend-only signal — the backend doesn't
23//! consume it. The frontend `AgentSessionNotices` renders a banner when it's set,
24//! and `service:update_object_meta` clears it when the user dismisses.
25//!
26//! A second, unrelated flag lives here for the same reason: `session:resume_failed`
27//! (SPEC_PANE_CLOSE_REOPEN_CONTINUITY_GUARANTEE_2026_07_27.md §4.2) marks a
28//! block whose `--resume <sid>` was rejected by the CLI (stale/unreachable
29//! session id — see `persistent/mod.rs`'s `poison_resume`) and silently fell
30//! through to a fresh conversation. Same shape, same frontend-only contract,
31//! different trigger (a resume rejection mid-session, not a stale PID found
32//! at boot) — kept in this file rather than a new module since it's the
33//! established home for "frontend-only session-state signal flags."
34//!
35//! Unlike `was_interrupted`, `resume_failed` is cleared by the backend as well
36//! as by the user: `clear_resume_failed` retracts it when the recovery scan
37//! that follows a rejection succeeds, so the banner can't claim the
38//! conversation was lost while the transcript's own session-outcome divider
39//! says it was resumed.
40
41use std::sync::Arc;
42
43use crate::backend::eventbus::EventBus;
44use crate::backend::obj::{Block, MetaMapType};
45use crate::backend::storage::store::Store;
46
47/// PID of the current running subprocess; 0 or missing = no process.
48pub const META_SESSION_ACTIVE_PID: &str = "session:active_pid";
49/// Set to `true` by startup scan when a pre-existing `active_pid` is found.
50pub const META_SESSION_WAS_INTERRUPTED: &str = "session:was_interrupted";
51/// Set to `true` when a `--resume <sid>` was rejected by the CLI and the
52/// controller silently started a fresh conversation instead.
53pub const META_SESSION_RESUME_FAILED: &str = "session:resume_failed";
54
55/// Record that a subprocess with `pid` has been spawned for `block_id`.
56/// Best-effort — logs on failure but never panics.
57pub fn mark_active_pid(mstore: &Arc<Store>, block_id: &str, pid: u32) {
58    let mut meta = MetaMapType::new();
59    meta.insert(META_SESSION_ACTIVE_PID.to_string(), serde_json::json!(pid));
60    // Clear any stale interrupt flag — we're running again.
61    meta.insert(META_SESSION_WAS_INTERRUPTED.to_string(), serde_json::Value::Null);
62    let oref_str = format!("block:{}", block_id);
63    if let Err(e) = crate::server::service::update_object_meta(mstore, &oref_str, &meta) {
64        tracing::warn!(block_id = %block_id, error = %e, "session_recovery: failed to mark active pid");
65    }
66}
67
68/// Mark that `block_id`'s `--resume <sid>` was rejected and the controller
69/// fell through to a fresh conversation. Best-effort — logs on failure but
70/// never panics, matching `mark_active_pid`'s contract. Called from
71/// `persistent/stderr_reader.rs`'s stderr reader the moment it detects the CLI's "No
72/// conversation found with session ID" line, right alongside the existing
73/// `core::persist_session_id(block_id, "", ...)` clear.
74///
75/// Broadcasts `waveobj:update` on success:
76/// this fires while the user may be actively watching the pane that just
77/// lost its resume, so — unlike `mark_active_pid`/`scan_orphans`, both of
78/// which run before any frontend subscriber could be watching (spawn time /
79/// server boot) — the write must reach an already-open `blockAtom` live, not
80/// only on the pane's next reload/reopen. Mirrors `persist_session_id`'s
81/// broadcast in `core.rs`. `event_bus: None` (tests / no live subscribers)
82/// silently skips the broadcast, matching every other call site's contract.
83pub fn mark_resume_failed(mstore: &Arc<Store>, event_bus: &Option<Arc<EventBus>>, block_id: &str) {
84    let mut meta = MetaMapType::new();
85    meta.insert(META_SESSION_RESUME_FAILED.to_string(), serde_json::json!(true));
86    let oref_str = format!("block:{}", block_id);
87    if let Err(e) = crate::server::service::update_object_meta(mstore, &oref_str, &meta) {
88        tracing::warn!(block_id = %block_id, error = %e, "session_recovery: failed to mark resume_failed");
89        return;
90    }
91    let Some(ref bus) = event_bus else {
92        return;
93    };
94    super::core::broadcast_block_update(mstore, bus, block_id);
95}
96
97/// Clear `session:resume_failed` — the counterpart to [`mark_resume_failed`],
98/// called when a resume attempt ultimately resolves as `Resumed`, or when a
99/// fresh spawn carries AgentMux's record of the conversation forward
100/// (`persistent/resume_retry.rs`'s `carry_continuation`).
101///
102/// [`mark_resume_failed`] fires from the stderr reader the instant the CLI
103/// says "No conversation found", which is *before*
104/// `persistent/resume_retry.rs`'s `retry_after_resume_failure` runs its
105/// recovery scan. In the common case that scan finds the real, live on-disk
106/// session and resumes it (live-confirmed in
107/// `docs/status/STATUS_STALE_RESUME_LIVE_REPRO_AND_FIX_PLAN_2026_08_23.md` §2),
108/// the conversation was NOT lost — but nothing cleared the flag, so the pane
109/// kept showing "Couldn't resume the previous conversation — started a new
110/// one" while the `agentmux_session_outcome` divider in the same transcript
111/// said `resumed`. Two disclosure surfaces contradicting each other is worse
112/// than either alone, so the flag now tracks the *resolved* outcome.
113///
114/// Skips both the write and the broadcast when the flag isn't currently set —
115/// a `Resumed` outcome is the overwhelmingly common case and almost never
116/// follows a failure, so the no-op path must not spam `waveobj:update` on
117/// every ordinary resume.
118pub fn clear_resume_failed(mstore: &Arc<Store>, event_bus: &Option<Arc<EventBus>>, block_id: &str) {
119    match mstore.get::<Block>(block_id) {
120        Ok(Some(block)) => {
121            let set = block
122                .meta
123                .get(META_SESSION_RESUME_FAILED)
124                .and_then(|v| v.as_bool())
125                .unwrap_or(false);
126            if !set {
127                return;
128            }
129        }
130        // Block missing or unreadable — nothing to clear, and the write
131        // below would fail anyway.
132        _ => return,
133    }
134
135    let mut meta = MetaMapType::new();
136    meta.insert(META_SESSION_RESUME_FAILED.to_string(), serde_json::Value::Null);
137    let oref_str = format!("block:{}", block_id);
138    if let Err(e) = crate::server::service::update_object_meta(mstore, &oref_str, &meta) {
139        tracing::warn!(block_id = %block_id, error = %e, "session_recovery: failed to clear resume_failed");
140        return;
141    }
142    tracing::info!(
143        block_id = %block_id,
144        "session_recovery: resume recovered — cleared the resume_failed disclosure"
145    );
146    let Some(ref bus) = event_bus else {
147        return;
148    };
149    // Same live-delivery requirement as `mark_resume_failed`: the user may be
150    // staring at the banner right now, so the clear must reach an already-open
151    // `blockAtom`, not only the pane's next reload.
152    super::core::broadcast_block_update(mstore, bus, block_id);
153}
154
155/// Clear `session:active_pid` — called when the subprocess exits for any reason.
156pub fn clear_active_pid(mstore: &Arc<Store>, block_id: &str) {
157    let mut meta = MetaMapType::new();
158    meta.insert(META_SESSION_ACTIVE_PID.to_string(), serde_json::Value::Null);
159    let oref_str = format!("block:{}", block_id);
160    if let Err(e) = crate::server::service::update_object_meta(mstore, &oref_str, &meta) {
161        tracing::warn!(block_id = %block_id, error = %e, "session_recovery: failed to clear active pid");
162    }
163}
164
165/// Clear `session:active_pid` **only if it still equals `expected_pid`** —
166/// a transactional compare-and-clear for exit-handlers that know which
167/// process instance they are cleaning up after.
168///
169/// Issue #2363: `PersistentSubprocessController`'s fallback respawn
170/// (`respawn_once_for_leftover_queue`) runs on a separate tokio task from
171/// the dying process's own exit-handler. The exit-handler's cleanup is
172/// gated on a spawn-generation check, but that gate is read once under the
173/// controller's lock while the fallback registers its own fresh PID in
174/// parallel — the unconditional [`clear_active_pid`] could then wipe the
175/// NEW process's registration, leaving it untracked for crash recovery
176/// until the next natural re-registration. The whole read-compare-clear
177/// here runs inside one `Store::with_tx` (a single connection lock across
178/// read+merge+write — see `object_helpers::update_object_meta`'s doc
179/// comment for the transactional contract), so a concurrent
180/// [`mark_active_pid`] strictly serializes against it: whichever commits
181/// second sees the other's completed write, and a mismatch means "a newer
182/// process already registered — leave it alone."
183///
184/// Best-effort like the unconditional variant: logs on failure, never
185/// panics. Returns nothing — callers do not branch on the outcome.
186pub fn clear_active_pid_if_pid(mstore: &Arc<Store>, block_id: &str, expected_pid: u32) {
187    let block_id_owned = block_id.to_string();
188    let result = mstore.with_tx(|tx| {
189        let mut block = tx.must_get::<Block>(&block_id_owned)?;
190        let current = block
191            .meta
192            .get(META_SESSION_ACTIVE_PID)
193            .and_then(|v| v.as_u64())
194            .unwrap_or(0);
195        if current != expected_pid as u64 {
196            // A newer process re-registered (or the pid was already
197            // cleared) — not ours to clear.
198            return Ok(false);
199        }
200        block.meta.remove(META_SESSION_ACTIVE_PID);
201        tx.update(&mut block)?;
202        Ok(true)
203    });
204    match result {
205        Ok(true) => {}
206        Ok(false) => {
207            tracing::info!(
208                block_id = %block_id,
209                expected_pid = expected_pid,
210                "session_recovery: active pid changed since this process registered — skipping clear"
211            );
212        }
213        Err(e) => {
214            tracing::warn!(block_id = %block_id, error = %e, "session_recovery: compare-and-clear active pid failed");
215        }
216    }
217}
218
219/// Scan all blocks at server startup. For any agent block that still has
220/// `session:active_pid` set, transfer the flag to `session:was_interrupted`.
221///
222/// Returns the number of orphaned sessions found.
223pub fn scan_orphans(mstore: &Arc<Store>) -> u32 {
224    let all_blocks = match mstore.get_all::<Block>() {
225        Ok(blocks) => blocks,
226        Err(e) => {
227            tracing::warn!(error = %e, "session_recovery: scan_orphans get_all failed");
228            return 0;
229        }
230    };
231
232    let mut count = 0u32;
233    for block in &all_blocks {
234        // Only agent panes
235        let view = block.meta.get("view").and_then(|v| v.as_str()).unwrap_or("");
236        if view != "agent" {
237            continue;
238        }
239
240        // Check for a stale active_pid
241        let pid = block
242            .meta
243            .get(META_SESSION_ACTIVE_PID)
244            .and_then(|v| v.as_u64())
245            .unwrap_or(0);
246        if pid == 0 {
247            continue;
248        }
249
250        // Transfer: clear active_pid, set was_interrupted
251        let mut update = MetaMapType::new();
252        update.insert(META_SESSION_ACTIVE_PID.to_string(), serde_json::Value::Null);
253        update.insert(
254            META_SESSION_WAS_INTERRUPTED.to_string(),
255            serde_json::json!(true),
256        );
257        let oref_str = format!("block:{}", block.oid);
258        match crate::server::service::update_object_meta(mstore, &oref_str, &update) {
259            Ok(()) => {
260                count += 1;
261                tracing::info!(
262                    block_id = %block.oid,
263                    stale_pid = pid,
264                    "session_recovery: marked orphaned session as interrupted"
265                );
266            }
267            Err(e) => {
268                tracing::warn!(
269                    block_id = %block.oid,
270                    error = %e,
271                    "session_recovery: failed to mark orphan"
272                );
273            }
274        }
275    }
276
277    if count > 0 {
278        tracing::info!(count = count, "session_recovery: scan_orphans complete");
279    }
280    count
281}
282
283#[cfg(test)]
284mod tests {
285    use super::*;
286    use crate::backend::obj::Block;
287
288    /// Verify that scan_orphans transfers active_pid → was_interrupted on
289    /// agent blocks, and leaves non-agent blocks + blocks without active_pid
290    /// untouched.
291    #[test]
292    fn test_scan_orphans_transfers_flag() {
293        let tmp = tempfile::tempdir().unwrap();
294        let mstore = Arc::new(Store::open(&tmp.path().join("objects.db")).unwrap());
295
296        // ORef parser requires valid UUIDs, so generate them inline.
297        let orphan_id = "11111111-1111-1111-1111-111111111111";
298        let clean_id = "22222222-2222-2222-2222-222222222222";
299        let term_id = "33333333-3333-3333-3333-333333333333";
300
301        // Agent block with stale active_pid — should be flagged.
302        let mut orphan = Block {
303            oid: orphan_id.to_string(),
304            parentoref: String::new(),
305            version: 1,
306            runtimeopts: None,
307            stickers: None,
308            meta: {
309                let mut m = MetaMapType::new();
310                m.insert("view".to_string(), serde_json::json!("agent"));
311                m.insert(META_SESSION_ACTIVE_PID.to_string(), serde_json::json!(12345u32));
312                m
313            },
314            subblockids: None,
315        };
316        mstore.insert(&mut orphan).unwrap();
317
318        // Agent block without active_pid — should be left alone.
319        let mut clean = Block {
320            oid: clean_id.to_string(),
321            parentoref: String::new(),
322            version: 1,
323            runtimeopts: None,
324            stickers: None,
325            meta: {
326                let mut m = MetaMapType::new();
327                m.insert("view".to_string(), serde_json::json!("agent"));
328                m
329            },
330            subblockids: None,
331        };
332        mstore.insert(&mut clean).unwrap();
333
334        // Non-agent block with active_pid (shouldn't exist in practice, but
335        // sanity check the view filter) — should be left alone.
336        let mut nonagent = Block {
337            oid: term_id.to_string(),
338            parentoref: String::new(),
339            version: 1,
340            runtimeopts: None,
341            stickers: None,
342            meta: {
343                let mut m = MetaMapType::new();
344                m.insert("view".to_string(), serde_json::json!("term"));
345                m.insert(META_SESSION_ACTIVE_PID.to_string(), serde_json::json!(67890u32));
346                m
347            },
348            subblockids: None,
349        };
350        mstore.insert(&mut nonagent).unwrap();
351
352        let count = scan_orphans(&mstore);
353        assert_eq!(count, 1, "exactly one agent orphan should be flagged");
354
355        // Verify orphan block meta
356        let after: Block = mstore.get(orphan_id).unwrap().unwrap();
357        assert!(
358            after.meta.get(META_SESSION_ACTIVE_PID).and_then(|v| v.as_u64()).unwrap_or(0) == 0,
359            "active_pid should be cleared"
360        );
361        assert_eq!(
362            after.meta.get(META_SESSION_WAS_INTERRUPTED).and_then(|v| v.as_bool()),
363            Some(true),
364            "was_interrupted should be true"
365        );
366
367        // Clean block untouched
368        let clean_after: Block = mstore.get(clean_id).unwrap().unwrap();
369        assert!(clean_after.meta.get(META_SESSION_WAS_INTERRUPTED).is_none());
370
371        // Non-agent block untouched (still has active_pid)
372        let term_after: Block = mstore.get(term_id).unwrap().unwrap();
373        assert_eq!(
374            term_after.meta.get(META_SESSION_ACTIVE_PID).and_then(|v| v.as_u64()),
375            Some(67890),
376        );
377    }
378
379    #[test]
380    fn test_mark_resume_failed_sets_the_flag() {
381        let tmp = tempfile::tempdir().unwrap();
382        let mstore = Arc::new(Store::open(&tmp.path().join("objects.db")).unwrap());
383
384        let block_id = "44444444-4444-4444-4444-444444444444";
385        let mut block = Block {
386            oid: block_id.to_string(),
387            parentoref: String::new(),
388            version: 1,
389            runtimeopts: None,
390            stickers: None,
391            meta: {
392                let mut m = MetaMapType::new();
393                m.insert("view".to_string(), serde_json::json!("agent"));
394                m
395            },
396            subblockids: None,
397        };
398        mstore.insert(&mut block).unwrap();
399
400        mark_resume_failed(&mstore, &None, block_id);
401
402        let after: Block = mstore.get(block_id).unwrap().unwrap();
403        assert_eq!(
404            after.meta.get(META_SESSION_RESUME_FAILED).and_then(|v| v.as_bool()),
405            Some(true),
406        );
407    }
408
409    /// reagent P1 on the original PR: mark_resume_failed wrote the meta but
410    /// never broadcast waveobj:update, so an already-open pane's blockAtom
411    /// never saw the flag live — the disclosure banner wouldn't render until
412    /// the pane was reloaded, defeating the point of a live-disclosure
413    /// signal. This asserts the broadcast actually reaches a connected
414    /// WebSocket client, not just that the DB write succeeded.
415    #[tokio::test]
416    async fn test_mark_resume_failed_broadcasts_waveobj_update() {
417        let tmp = tempfile::tempdir().unwrap();
418        let mstore = Arc::new(Store::open(&tmp.path().join("objects.db")).unwrap());
419        let event_bus = Arc::new(EventBus::new());
420
421        let block_id = "77777777-7777-7777-7777-777777777777";
422        let mut block = Block {
423            oid: block_id.to_string(),
424            parentoref: String::new(),
425            version: 1,
426            runtimeopts: None,
427            stickers: None,
428            meta: {
429                let mut m = MetaMapType::new();
430                m.insert("view".to_string(), serde_json::json!("agent"));
431                m
432            },
433            subblockids: None,
434        };
435        mstore.insert(&mut block).unwrap();
436
437        let mut receivers = event_bus.register_ws("test-conn", "test-tab");
438
439        mark_resume_failed(&mstore, &Some(event_bus.clone()), block_id);
440
441        let msg = tokio::time::timeout(std::time::Duration::from_secs(2), receivers.priority.recv())
442            .await
443            .expect("timed out waiting for waveobj:update broadcast")
444            .expect("priority channel closed");
445        assert_eq!(msg.get("eventtype").and_then(|v| v.as_str()), Some("waveobj:update"));
446        assert_eq!(msg.get("oref").and_then(|v| v.as_str()), Some(format!("block:{}", block_id).as_str()));
447        let obj = msg.get("data").and_then(|d| d.get("obj")).expect("data.obj present");
448        assert_eq!(
449            obj.get("meta")
450                .and_then(|m| m.get(META_SESSION_RESUME_FAILED))
451                .and_then(|v| v.as_bool()),
452            Some(true),
453            "broadcast payload must carry the flag, not just trigger a generic refetch",
454        );
455    }
456
457    fn agent_block(block_id: &str) -> Block {
458        Block {
459            oid: block_id.to_string(),
460            parentoref: String::new(),
461            version: 1,
462            runtimeopts: None,
463            stickers: None,
464            meta: {
465                let mut m = MetaMapType::new();
466                m.insert("view".to_string(), serde_json::json!("agent"));
467                m
468            },
469            subblockids: None,
470        }
471    }
472
473    /// The recovery scan that follows a rejected `--resume` usually succeeds
474    /// (live-confirmed in STATUS_STALE_RESUME_LIVE_REPRO_AND_FIX_PLAN_2026_08_23
475    /// §2). When it does, the banner `mark_resume_failed` already raised must
476    /// come back down — otherwise the pane claims the conversation was lost
477    /// while the transcript's own outcome divider says `resumed`.
478    #[test]
479    fn test_clear_resume_failed_retracts_the_flag() {
480        let tmp = tempfile::tempdir().unwrap();
481        let mstore = Arc::new(Store::open(&tmp.path().join("objects.db")).unwrap());
482
483        let block_id = "55555555-5555-5555-5555-555555555555";
484        let mut block = agent_block(block_id);
485        mstore.insert(&mut block).unwrap();
486
487        mark_resume_failed(&mstore, &None, block_id);
488        clear_resume_failed(&mstore, &None, block_id);
489
490        let after: Block = mstore.get(block_id).unwrap().unwrap();
491        assert_eq!(
492            after.meta.get(META_SESSION_RESUME_FAILED).and_then(|v| v.as_bool()),
493            None,
494            "a recovered resume must leave no resume_failed flag behind",
495        );
496    }
497
498    /// A `Resumed` outcome is the overwhelmingly common case and almost never
499    /// follows a failure, so the clear must not broadcast `waveobj:update` on
500    /// every ordinary resume just to write a null over an absent key.
501    #[tokio::test]
502    async fn test_clear_resume_failed_is_silent_when_the_flag_was_never_set() {
503        let tmp = tempfile::tempdir().unwrap();
504        let mstore = Arc::new(Store::open(&tmp.path().join("objects.db")).unwrap());
505        let event_bus = Arc::new(EventBus::new());
506
507        let block_id = "66666666-6666-6666-6666-666666666666";
508        let mut block = agent_block(block_id);
509        mstore.insert(&mut block).unwrap();
510        let version_before = mstore.get::<Block>(block_id).unwrap().unwrap().version;
511
512        let mut receivers = event_bus.register_ws("test-conn", "test-tab");
513
514        clear_resume_failed(&mstore, &Some(event_bus.clone()), block_id);
515
516        assert!(
517            tokio::time::timeout(
518                std::time::Duration::from_millis(200),
519                receivers.priority.recv(),
520            )
521            .await
522            .is_err(),
523            "no broadcast may fire when there was no flag to retract",
524        );
525        assert_eq!(
526            mstore.get::<Block>(block_id).unwrap().unwrap().version,
527            version_before,
528            "the block must not be rewritten when there was nothing to clear",
529        );
530    }
531
532    /// The retraction has the same live-delivery requirement as the flag
533    /// itself: the user may be looking at the banner right now, so an
534    /// already-open `blockAtom` must see it go away without a pane reload.
535    #[tokio::test]
536    async fn test_clear_resume_failed_broadcasts_waveobj_update() {
537        let tmp = tempfile::tempdir().unwrap();
538        let mstore = Arc::new(Store::open(&tmp.path().join("objects.db")).unwrap());
539        let event_bus = Arc::new(EventBus::new());
540
541        let block_id = "88888888-8888-8888-8888-888888888888";
542        let mut block = agent_block(block_id);
543        mstore.insert(&mut block).unwrap();
544        mark_resume_failed(&mstore, &None, block_id);
545
546        let mut receivers = event_bus.register_ws("test-conn", "test-tab");
547
548        clear_resume_failed(&mstore, &Some(event_bus.clone()), block_id);
549
550        let msg = tokio::time::timeout(std::time::Duration::from_secs(2), receivers.priority.recv())
551            .await
552            .expect("timed out waiting for waveobj:update broadcast")
553            .expect("priority channel closed");
554        assert_eq!(msg.get("eventtype").and_then(|v| v.as_str()), Some("waveobj:update"));
555        let obj = msg.get("data").and_then(|d| d.get("obj")).expect("data.obj present");
556        assert_eq!(
557            obj.get("meta")
558                .and_then(|m| m.get(META_SESSION_RESUME_FAILED))
559                .and_then(|v| v.as_bool()),
560            None,
561            "the broadcast payload must show the flag gone, not still set",
562        );
563    }
564}
agentmux_srv/backend/blockcontroller/session_recovery.rs

agentmux_srv/backend/blockcontroller/
session_recovery.rs
