From 47689e7f850f9336b5400763bf4d6a729d5b22ee Mon Sep 17 00:00:00 2001 From: Christopher Sardegna Date: Mon, 20 Jul 2026 23:39:01 -0700 Subject: [PATCH 1/8] feat: implement automatic recovery snapshot mechanism with debounce and cadence intervals --- src/harness/mod.rs | 4 +- src/session.rs | 272 +++++++++++++++++++++++++++----- src/supervisor.rs | 169 ++++++++++++++++++++ src/supervisor_capture_tests.rs | 70 ++++++++ src/supervisor_tests.rs | 194 +++++++++++++++++++++++ tests/common/mod.rs | 5 + 6 files changed, 675 insertions(+), 39 deletions(-) diff --git a/src/harness/mod.rs b/src/harness/mod.rs index 54b3a3e..313d586 100644 --- a/src/harness/mod.rs +++ b/src/harness/mod.rs @@ -31,8 +31,8 @@ use std::{ pub use claude::Claude; pub use codex::Codex; -// Test-only: `testutil::write_rollout` derives day directories from it. -#[cfg(test)] +// `session`'s recovery stems and labels derive their UTC dates from it, as +// does the shared rollout fixture in `testutil`. pub(crate) use codex::civil_from_days; pub use grok::Grok; diff --git a/src/session.rs b/src/session.rs index 0735b19..b2290d6 100644 --- a/src/session.rs +++ b/src/session.rs @@ -8,6 +8,7 @@ use std::{ os::unix::fs::{DirBuilderExt, OpenOptionsExt, PermissionsExt}, path::{Path, PathBuf}, sync::atomic::{AtomicU64, Ordering}, + time::SystemTime, }; /// One recipe entry. Entries without a group or name serialize as strings; @@ -75,9 +76,8 @@ pub fn sessions_dir(root: Option) -> Option { /// Missing versions are interpreted as version 1; unsupported versions fail. const FORMAT_VERSION: u64 = 1; -/// Serialize the versioned wrapped schema. The stored name distinguishes -/// names that sanitize to the same filename. -fn to_json(name: &str, cfg: &SessionConfig) -> String { +/// Build the `dirs` object alone: the recipe body without the wrapper. +fn dirs_json(cfg: &SessionConfig) -> jzon::JsonValue { let mut dirs = jzon::JsonValue::new_object(); for (dir, entries) in cfg { let mut arr = jzon::JsonValue::new_array(); @@ -100,13 +100,26 @@ fn to_json(name: &str, cfg: &SessionConfig) -> String { } let _ = dirs.insert(dir, arr); } + dirs +} + +/// Serialize the versioned wrapped schema. The stored name distinguishes +/// names that sanitize to the same filename. +fn to_json(name: &str, cfg: &SessionConfig) -> String { let mut obj = jzon::JsonValue::new_object(); let _ = obj.insert("version", FORMAT_VERSION); let _ = obj.insert("name", name); - let _ = obj.insert("dirs", dirs); + let _ = obj.insert("dirs", dirs_json(cfg)); obj.pretty(2) } +/// Serialize only the recipe body, for change detection. Excludes the wrapper +/// because its `name` field is volatile in recovery snapshots (the label +/// carries the write time): equal recipes must fingerprint equal. +pub fn fingerprint_json(cfg: &SessionConfig) -> String { + dirs_json(cfg).dump() +} + /// Parse wrapped and flat schemas, returning the stored name when present. /// A wrapped file has an object-valued `dirs`; flat files have entry arrays at /// the top level, including when a directory is literally named `dirs`. @@ -183,9 +196,9 @@ fn from_json(text: &str) -> io::Result<(Option, SessionConfig)> { /// Distinguishes concurrent savers' temp files within one process. static TMP_SEQ: AtomicU64 = AtomicU64::new(0); -pub fn save_in(dir: &Path, name: &str, cfg: &SessionConfig) -> io::Result { - // Create missing directories with 0700 and restrict the session directory - // itself to 0700. Existing parent directories remain unchanged. +/// Create missing directories with 0700 and restrict `dir` itself to 0700. +/// Existing parent directories remain unchanged. +fn ensure_private_dir(dir: &Path) -> io::Result<()> { fs::DirBuilder::new() .recursive(true) .mode(0o700) @@ -193,40 +206,21 @@ pub fn save_in(dir: &Path, name: &str, cfg: &SessionConfig) -> io::Result { - if let Ok((Some(stored), _)) = from_json(&text) - && stored != trimmed - { - return Err(io::Error::new( - io::ErrorKind::AlreadyExists, - format!( - "session \"{trimmed}\" collides with existing \"{stored}\" \ - (both map to {file_name})" - ), - )); - } - } - Err(e) if e.kind() == io::ErrorKind::NotFound => {} - Err(e) => return Err(e), - } + Ok(()) +} - // Write a private temp file in the session directory, sync its contents, - // then atomically rename it over the recipe. The `.tmp` suffix keeps it - // out of `list_in`, and the rename gives the target mode 0600. +/// Write `contents` to `/` atomically: a private temp file in +/// `dir`, synced, then renamed over the target. The `.tmp` suffix keeps the +/// temp out of `list_in`, and the rename gives the target mode 0600. +fn write_atomic(dir: &Path, file_name: &str, contents: &str) -> io::Result { + let file = dir.join(file_name); let pid = std::process::id(); let (mut tmp_file, tmp) = loop { let n = TMP_SEQ.fetch_add(1, Ordering::Relaxed); - // Shorten the recipe portion so the decorated temporary filename stays + // Shorten the target portion so the decorated temporary filename stays // within the 255-byte component limit. let suffix = format!(".{pid}.{n}.tmp"); - let stem = prefix_bytes(&file_name, 254 - suffix.len()); + let stem = prefix_bytes(file_name, 254 - suffix.len()); let candidate = dir.join(format!(".{stem}{suffix}")); match fs::OpenOptions::new() .write(true) @@ -240,8 +234,8 @@ pub fn save_in(dir: &Path, name: &str, cfg: &SessionConfig) -> io::Result io::Result io::Result { + ensure_private_dir(dir)?; + let trimmed = name.trim(); + let file_name = format!("{}.json", sanitize(name)); + let file = dir.join(&file_name); + + // Refuse a stored-name mismatch because distinct names can sanitize to the + // same filename. Files without a parseable stored name remain overwritable. + match fs::read_to_string(&file) { + Ok(text) => { + if let Ok((Some(stored), _)) = from_json(&text) + && stored != trimmed + { + return Err(io::Error::new( + io::ErrorKind::AlreadyExists, + format!( + "session \"{trimmed}\" collides with existing \"{stored}\" \ + (both map to {file_name})" + ), + )); + } + } + Err(e) if e.kind() == io::ErrorKind::NotFound => {} + Err(e) => return Err(e), + } + + write_atomic(dir, &file_name, &to_json(trimmed, cfg)) +} + pub fn load_in(dir: &Path, name: &str) -> io::Result { let file = dir.join(format!("{}.json", sanitize(name))); from_json(&fs::read_to_string(file)?).map(|(_, cfg)| cfg) @@ -279,6 +302,92 @@ pub fn list_in(dir: &Path) -> Vec { names } +// --- recovery snapshots: the supervisor's automatic fleet backups, written +// under `/recovery` in the ordinary wrapped format. Phase 1 +// only writes; listing and loading them arrive in later phases. --------------- + +/// Snapshots kept per recovery directory; older ones are pruned after each +/// write. +const RECOVERY_KEEP: usize = 10; + +/// Recovery-snapshot directory under a session root. Created 0700 on first +/// write; `list_in` never descends into it (directories fail its `.json` +/// extension filter), so snapshots stay out of the session picker. +pub fn recovery_dir(sessions_root: &Path) -> PathBuf { + sessions_root.join("recovery") +} + +/// UTC civil time for `t`: (year, month, day, hour, minute, second). +/// UTC because the recovery stems must sort lexically by age: local time +/// repeats an hour at DST fall-back. +fn civil_utc(t: SystemTime) -> (i64, u32, u32, u64, u64, u64) { + let secs = t + .duration_since(SystemTime::UNIX_EPOCH) + .unwrap_or_default() + .as_secs(); + let (y, m, d) = crate::harness::civil_from_days((secs / 86_400) as i64); + let tod = secs % 86_400; + (y, m, d, tod / 3600, (tod % 3600) / 60, tod % 60) +} + +/// Incarnation filename stem `-` from the supervisor's +/// start time and process id. The pid separates a daemon from a concurrent +/// `--foreground` client sharing one config root; the leading UTC stamp makes +/// lexical order age order, which pruning relies on. Digits and dashes only, +/// so the stem needs no `sanitize` pass. +pub fn recovery_stem(start: SystemTime, pid: u32) -> String { + let (y, m, d, hh, mm, ss) = civil_utc(start); + format!("{y:04}{m:02}{d:02}-{hh:02}{mm:02}{ss:02}-{pid}") +} + +/// Human label stored in a snapshot's `name` field: `autosaved ` (UTC) from the write time. The label exists so a recovery file +/// copied by hand into `sessions/` becomes an ordinary, sensibly-named +/// session with no tooling: `list_in` shows the stored name, and loading it +/// needs nothing new. +pub fn recovery_label(now: SystemTime) -> String { + let (y, m, d, hh, mm, _) = civil_utc(now); + format!("autosaved {y:04}-{m:02}-{d:02} {hh:02}:{mm:02}") +} + +/// Write one recovery snapshot through the same atomic temp-rename, 0600, and +/// version-1 path as `save_in`, then prune. `save_in`'s stored-name collision +/// check does not apply: the incarnation owns `.json` outright and +/// always overwrites it. +pub fn save_recovery_in( + dir: &Path, + file_stem: &str, + name: &str, + cfg: &SessionConfig, +) -> io::Result { + ensure_private_dir(dir)?; + let file = write_atomic(dir, &format!("{file_stem}.json"), &to_json(name, cfg))?; + prune_recovery(dir); + Ok(file) +} + +/// Best-effort prune: keep the newest [`RECOVERY_KEEP`] snapshots by filename +/// (stems are UTC timestamps, so lexical order is age order) and ignore every +/// error -- a snapshot that cannot be removed must not fail the write that +/// just succeeded. +fn prune_recovery(dir: &Path) { + let Ok(entries) = fs::read_dir(dir) else { + return; + }; + let mut snapshots: Vec = entries + .flatten() + .map(|e| e.path()) + .filter(|p| p.extension().and_then(|s| s.to_str()) == Some("json")) + .collect(); + if snapshots.len() <= RECOVERY_KEEP { + return; + } + snapshots.sort(); + for old in &snapshots[..snapshots.len() - RECOVERY_KEEP] { + let _ = fs::remove_file(old); + } +} + #[cfg(test)] mod tests { use super::*; @@ -671,4 +780,93 @@ mod tests { } let _ = fs::remove_dir_all(&dir); } + + /// 2026-07-14 09:30:15 UTC, matching `civil_from_days`'s test vector. + fn recovery_instant() -> SystemTime { + SystemTime::UNIX_EPOCH + std::time::Duration::from_secs(1_784_021_415) + } + + /// The stem is `-`; the label is the write minute. + #[test] + fn recovery_stem_and_label_render_utc() { + assert_eq!( + recovery_stem(recovery_instant(), 4242), + "20260714-093015-4242" + ); + assert_eq!( + recovery_label(recovery_instant()), + "autosaved 2026-07-14 09:30" + ); + } + + /// A snapshot round-trips through `load_in` with version 1, the human + /// label, and the private-permission idiom of ordinary saves. + #[test] + fn recovery_snapshot_round_trips_with_version_and_label() { + let base = temp("session_recovery_roundtrip"); + let rec = recovery_dir(&base); + let mut cfg = SessionConfig::new(); + cfg.insert("~/p".into(), vec![gne("cargo run", "api", "server")]); + let stem = recovery_stem(recovery_instant(), 4242); + let label = recovery_label(recovery_instant()); + + let file = save_recovery_in(&rec, &stem, &label, &cfg).unwrap(); + assert_eq!(file, rec.join("20260714-093015-4242.json")); + let text = fs::read_to_string(&file).unwrap(); + assert!(text.contains("\"version\": 1"), "{text}"); + let (stored, parsed) = from_json(&text).unwrap(); + assert_eq!(stored.as_deref(), Some("autosaved 2026-07-14 09:30")); + assert_eq!(parsed, cfg); + assert_eq!(load_in(&rec, &stem).unwrap(), cfg); + + let mode = |p: &Path| fs::metadata(p).unwrap().permissions().mode() & 0o777; + assert_eq!(mode(&rec), 0o700); + assert_eq!(mode(&file), 0o600); + let _ = fs::remove_dir_all(&base); + } + + /// Prune keeps exactly the newest [`RECOVERY_KEEP`] snapshots by filename. + #[test] + fn recovery_prune_keeps_the_newest_ten() { + let base = temp("session_recovery_prune"); + let rec = recovery_dir(&base); + let mut cfg = SessionConfig::new(); + cfg.insert("~/p".into(), vec![e("vim")]); + for i in 1..=12u32 { + let stem = format!("20260714-0930{i:02}-77"); + save_recovery_in(&rec, &stem, "autosaved 2026-07-14 09:30", &cfg).unwrap(); + } + + let mut names: Vec = fs::read_dir(&rec) + .unwrap() + .flatten() + .map(|e| e.file_name().into_string().unwrap()) + .collect(); + names.sort(); + let expected: Vec = (3..=12u32) + .map(|i| format!("20260714-0930{i:02}-77.json")) + .collect(); + assert_eq!(names, expected, "prune must drop exactly the oldest two"); + let _ = fs::remove_dir_all(&base); + } + + /// The `recovery/` subdirectory never appears in a session listing: + /// directories fail `list_in`'s `.json` extension filter. + #[test] + fn list_ignores_the_recovery_subdirectory() { + let dir = temp("session_list_recovery"); + let mut cfg = SessionConfig::new(); + cfg.insert("~/p".into(), vec![e("vim")]); + save_in(&dir, "real", &cfg).unwrap(); + save_recovery_in( + &recovery_dir(&dir), + &recovery_stem(recovery_instant(), 7), + &recovery_label(recovery_instant()), + &cfg, + ) + .unwrap(); + + assert_eq!(list_in(&dir), vec!["real".to_string()]); + let _ = fs::remove_dir_all(&dir); + } } diff --git a/src/supervisor.rs b/src/supervisor.rs index 77083d7..7b44eff 100644 --- a/src/supervisor.rs +++ b/src/supervisor.rs @@ -89,6 +89,18 @@ fn effective_scrollback(flag: Option, env: Option<&str>) -> usize { /// that do not exit after SIGTERM. const KILL_GRACE: Duration = Duration::from_secs(2); +/// Quiet period after a structural recipe mutation before the recovery +/// snapshot writes, coalescing a burst (a session load spawning many tasks) +/// into one write. +const RECOVERY_DEBOUNCE: Duration = Duration::from_secs(2); + +/// Interval between recovery content-comparison passes. Resume-ID resolution +/// is pull-based (`current_resume_id` reads the capture file during +/// serialization; no event fires when an ID appears), so only re-serializing +/// on a cadence can observe agent conversation-ID drift between structural +/// mutations. +const RECOVERY_CADENCE: Duration = Duration::from_secs(60); + /// Maximum stored label length in Unicode scalar values after normalization, /// shared by group and display-name assignments. const MAX_LABEL_CHARS: usize = 64; @@ -155,6 +167,58 @@ fn harness_home(env: &[(OsString, OsString)], h: &dyn harness::Harness) -> Optio val(h.home_env_var()).or_else(|| Some(val("HOME")?.join(h.home_dot_dir()))) } +/// State for the automatic fleet-recovery snapshot writer. Recovery is +/// insurance bolted beside the supervision path, never in it: every branch +/// that cannot write resolves toward not disturbing supervision, silently. +/// Quit, `--kill`, and SIGTERM teardowns deliberately neither write nor +/// delete snapshots -- files persisting past an "oops" is the point. +struct Recovery { + /// Armed in production. Test builds start disarmed because supervisors + /// built with real launch contexts tick inside many unrelated tests, + /// which would otherwise land snapshots in the developer's real config + /// root; recovery tests arm explicitly via `set_recovery_timing`. + enabled: bool, + /// Set by the six structural recipe mutations, cleared by the next due + /// pass (written, unchanged, empty, or unwritable alike -- see + /// `Supervisor::maybe_write_recovery`). + dirty: bool, + /// The most recent structural mutation: the debounce anchor. + last_mutation: Option, + /// The last cadence pass, due or not. + last_cadence: Instant, + /// FNV-1a of the last successfully written recipe fingerprint; `None` + /// until the first write. + last_hash: Option, + /// Incarnation filename stem, fixed at construction from the supervisor's + /// start time and pid (see `session::recovery_stem`): every write of this + /// incarnation replaces its own file. + stem: String, + /// One-notice latch: a persistently failing root (e.g. an unwritable + /// config directory) reports once per failure streak, not per attempt. + failing: bool, + /// `RECOVERY_DEBOUNCE`/`RECOVERY_CADENCE` in production; fields so tests + /// shrink them instead of sleeping through real seconds (the `kill_grace` + /// pattern). + debounce: Duration, + cadence: Duration, +} + +impl Recovery { + fn new() -> Recovery { + Recovery { + enabled: !cfg!(test), + dirty: false, + last_mutation: None, + last_cadence: Instant::now(), + last_hash: None, + stem: session::recovery_stem(std::time::SystemTime::now(), std::process::id()), + failing: false, + debounce: RECOVERY_DEBOUNCE, + cadence: RECOVERY_CADENCE, + } + } +} + pub struct Supervisor { tasks: Vec, /// Removed tasks whose process groups may still be winding down: TERMed at @@ -196,6 +260,8 @@ pub struct Supervisor { /// Capture assets keyed by canonicalized root and reused for this /// supervisor's lifetime. capture: BTreeMap, + /// Automatic fleet-recovery snapshot state. + recovery: Recovery, } impl Supervisor { @@ -214,6 +280,7 @@ impl Supervisor { waker: Arc::new(Mutex::new(None)), kill_grace: KILL_GRACE, capture: BTreeMap::new(), + recovery: Recovery::new(), } } @@ -228,6 +295,15 @@ impl Supervisor { self.kill_grace = grace; } + /// Arm the recovery writer with short timings so snapshot tests run in + /// milliseconds. Test builds start disarmed (see [`Recovery::enabled`]). + #[cfg(test)] + pub fn set_recovery_timing(&mut self, debounce: Duration, cadence: Duration) { + self.recovery.enabled = true; + self.recovery.debounce = debounce; + self.recovery.cadence = cadence; + } + /// Install the sender the current serving loop waits on, so task reader /// threads (present and future; they share this one slot) wake it on output. pub fn set_waker(&self, tx: Sender) { @@ -260,6 +336,25 @@ impl Supervisor { /// Apply one client request. Fire-and-forget: any result (a save/load /// notice, a spawn failure) is queued as `Event::Status`, never returned. pub fn apply(&mut self, cmd: Command) { + // The six structural recipe mutations arm the recovery writer; the + // burst coalesces behind `RECOVERY_DEBOUNCE` in `tick`. `Tag` is + // deliberately absent: tags are not recipe state (a `SessionEntry` + // stores cmd, group, and name only), so a tag flip cannot change the + // snapshot. Arming keys on the command, not its outcome -- a refused + // spawn or unknown-id assignment costs one fingerprint comparison in + // the next pass, which then skips the write. + if matches!( + &cmd, + Command::Spawn { .. } + | Command::Remove { .. } + | Command::Restart { .. } + | Command::SetGroup { .. } + | Command::SetName { .. } + | Command::LoadSession { .. } + ) { + self.recovery.dirty = true; + self.recovery.last_mutation = Some(Instant::now()); + } match cmd { Command::Spawn { command, @@ -497,6 +592,80 @@ impl Supervisor { self.events.push(Event::Screen(view)); } } + + self.maybe_write_recovery(now); + } + + /// Write the recovery snapshot when a pass is due. Two schedules share + /// the write: a debounce pass follows a structural-mutation burst, and a + /// cadence pass re-serializes on an interval to observe pull-resolved + /// resume-ID drift (see [`RECOVERY_CADENCE`]). Recovery is insurance + /// beside the supervision path: every refusal here is silent by design. + fn maybe_write_recovery(&mut self, now: Instant) { + if !self.recovery.enabled { + return; + } + let debounce_due = self.recovery.dirty + && self + .recovery + .last_mutation + .is_some_and(|t| now.duration_since(t) >= self.recovery.debounce); + let cadence_due = now.duration_since(self.recovery.last_cadence) >= self.recovery.cadence; + if !(debounce_due || cadence_due) { + return; + } + if cadence_due { + self.recovery.last_cadence = now; + } + // An empty fleet never writes: the snapshot worth recovering is + // exactly the one a quit-with-zero-tasks pass would clobber. + // Clearing `dirty` is not deferral; the next mutation re-arms. + if self.tasks.is_empty() { + self.recovery.dirty = false; + return; + } + // No resolvable config root: nothing to write to, nothing to report. + let Some(root) = self.sessions_root() else { + self.recovery.dirty = false; + return; + }; + // Give finished tasks their exit scrape before serialization reads + // resume IDs, exactly as `save_session` does. + for t in &mut self.tasks { + scrape_now(t); + } + let cfg = self.session_config(); + // Fingerprint the recipe body, not the wrapped file: the stored + // label carries the write time, so hashing the full serialization + // would report a change every minute. + let hash = fnv1a_hex(session::fingerprint_json(&cfg).as_bytes()); + if self.recovery.last_hash.as_ref() == Some(&hash) { + self.recovery.dirty = false; + return; + } + let label = session::recovery_label(std::time::SystemTime::now()); + match session::save_recovery_in( + &session::recovery_dir(&root), + &self.recovery.stem, + &label, + &cfg, + ) { + Ok(_) => { + self.recovery.last_hash = Some(hash); + self.recovery.failing = false; + } + Err(e) => { + // Degrade silently, but say so once per failure streak. The + // cadence pass retries because `last_hash` still names the + // last *written* state; `dirty` clears below either way, so + // a broken root costs one attempt per interval, not per tick. + if !self.recovery.failing { + self.recovery.failing = true; + self.status(format!("recovery snapshot failed: {e}")); + } + } + } + self.recovery.dirty = false; } /// Hand the client every event queued since the last drain. diff --git a/src/supervisor_capture_tests.rs b/src/supervisor_capture_tests.rs index 5dad1d7..d3a8ec6 100644 --- a/src/supervisor_capture_tests.rs +++ b/src/supervisor_capture_tests.rs @@ -1103,3 +1103,73 @@ fn non_agent_entries_survive_save_as_plain_strings() { ); let _ = std::fs::remove_dir_all(&dir); } + +/// The cadence pass picks up conversation-ID drift that no structural +/// mutation announces: resume IDs resolve pull-style at serialization time, +/// so only re-serializing on the interval can see a capture file change. A +/// settled recipe then stops writing (the snapshot's mtime holds still). +#[test] +fn recovery_cadence_rewrites_on_capture_drift_and_skips_when_static() { + let dir = scratch("cap_recovery_cadence"); + let (bin, runtime, config) = (dir.join("bin"), dir.join("run"), dir.join("config")); + install_stub(&bin, "claude", &dir); + let mut s = Supervisor::new(24, 80, 2000); + s.set_launch_context(agent_ctx_plus( + &bin, + &runtime, + dir.clone(), + &[("FLEETCOM_CONFIG_DIR", &config)], + )); + s.set_recovery_timing(Duration::from_millis(20), Duration::from_millis(100)); + spawn(&mut s, "claude", dir.clone()); + let _ = wait_argv(&mut s, &dir.join("argv")); + + let rec = config.join("sessions").join("recovery"); + let snapshot = |rec: &Path| -> Option<(PathBuf, String)> { + let p = std::fs::read_dir(rec).ok()?.flatten().next()?.path(); + let text = std::fs::read_to_string(&p).ok()?; + Some((p, text)) + }; + // The debounced spawn write lands first, carrying the pinned ID only. + assert!( + wait_until(Duration::from_secs(5), || { + s.tick(); + snapshot(&rec).is_some() + }), + "the spawn snapshot never landed" + ); + assert!(!snapshot(&rec).unwrap().1.contains(CAP_OTHER)); + + // Drift the conversation: the capture file now reports a different ID. + // No structural mutation follows, so only the cadence pass can see it. + let cap = s.tasks[0].capture_file.clone().expect("capture file set"); + std::fs::write( + &cap, + format!( + r#"{{"session_id":"{CAP_OTHER}","hook_event_name":"SessionStart","source":"clear"}}"# + ), + ) + .unwrap(); + assert!( + wait_until(Duration::from_secs(5), || { + s.tick(); + snapshot(&rec).is_some_and(|(_, t)| t.contains(CAP_OTHER)) + }), + "the cadence pass never picked up the drifted ID" + ); + + // A settled recipe writes nothing more across several intervals. + let (path, _) = snapshot(&rec).unwrap(); + let mtime = std::fs::metadata(&path).unwrap().modified().unwrap(); + let rewritten = wait_until(Duration::from_millis(600), || { + s.tick(); + std::fs::metadata(&path).unwrap().modified().unwrap() != mtime + }); + assert!( + !rewritten, + "an unchanged recipe must not rewrite the snapshot" + ); + let names: Vec<_> = std::fs::read_dir(&rec).unwrap().flatten().collect(); + assert_eq!(names.len(), 1, "one incarnation owns one snapshot file"); + let _ = std::fs::remove_dir_all(&dir); +} diff --git a/src/supervisor_tests.rs b/src/supervisor_tests.rs index 0326d1f..7e7128f 100644 --- a/src/supervisor_tests.rs +++ b/src/supervisor_tests.rs @@ -1608,5 +1608,199 @@ fn key_command_encodes_against_live_cursor_mode() { let _ = std::fs::remove_dir_all(&dir); } +// --- recovery-snapshot writer ------------------------------------------- + +/// Supervisor wired to `config` with the recovery writer armed at short +/// timings (test builds start disarmed; see `Recovery::enabled`). +fn recovery_sup(config: &Path, cwd: PathBuf, debounce: Duration, cadence: Duration) -> Supervisor { + let mut s = Supervisor::new(24, 80, 2000); + s.set_launch_context(config_ctx(config, cwd, &[])); + s.set_recovery_timing(debounce, cadence); + s +} + +/// Sorted recovery-snapshot filenames under `config`'s session root. +fn recovery_files(config: &Path) -> Vec { + let mut names: Vec = std::fs::read_dir(config.join("sessions").join("recovery")) + .map(|it| { + it.flatten() + .map(|e| e.file_name().to_string_lossy().into_owned()) + .collect() + }) + .unwrap_or_default(); + names.sort(); + names +} + +/// Read and clear the writer's dirty flag. +fn take_dirty(s: &mut Supervisor) -> bool { + std::mem::replace(&mut s.recovery.dirty, false) +} + +/// The six structural recipe mutations arm the writer; `Tag` does not (tags +/// are not recipe state). Arming keys on the command, not its outcome, so a +/// refused `Restart` and a missing `LoadSession` recipe still arm. +#[test] +fn recovery_arms_on_structural_mutations_not_tag() { + let mut s = sup(24, 80); + assert!(!s.recovery.dirty, "a fresh supervisor starts clean"); + + spawn(&mut s, "sleep 30", here()); + assert!(take_dirty(&mut s), "Spawn must arm"); + let id = first_id(&mut s); + + s.recovery.dirty = false; // first_id ticks; reassert a clean baseline + s.apply(Command::Tag { id, on: true }); + assert!( + !take_dirty(&mut s), + "Tag is not recipe state and must not arm" + ); + + s.apply(Command::SetGroup { + id, + group: Some("api".into()), + }); + assert!(take_dirty(&mut s), "SetGroup must arm"); + + s.apply(Command::SetName { + id, + name: Some("server".into()), + }); + assert!(take_dirty(&mut s), "SetName must arm"); + + s.apply(Command::Restart { id }); + assert!(take_dirty(&mut s), "Restart must arm"); + + s.apply(Command::LoadSession { + name: "ghost".into(), + }); + assert!(take_dirty(&mut s), "LoadSession must arm"); + + s.apply(Command::Remove { id }); + assert!(take_dirty(&mut s), "Remove must arm"); +} + +/// A burst of mutations coalesces behind the debounce into one snapshot +/// carrying the whole burst: no write lands inside the quiet period, and one +/// file holds all three commands afterward. +#[test] +fn recovery_debounce_coalesces_a_mutation_burst() { + let dir = scratch("recovery_debounce"); + let config = dir.join("config"); + let mut s = recovery_sup( + &config, + dir.clone(), + Duration::from_millis(500), + Duration::from_secs(600), + ); + spawn(&mut s, "sleep 30", dir.clone()); + spawn(&mut s, "sleep 31", dir.clone()); + spawn(&mut s, "sleep 32", dir.clone()); + s.tick(); + assert!( + recovery_files(&config).is_empty(), + "a write inside the debounce window defeats coalescing" + ); + + assert!( + wait_until(Duration::from_secs(5), || { + s.tick(); + !recovery_files(&config).is_empty() + }), + "the debounced snapshot never landed" + ); + let files = recovery_files(&config); + assert_eq!( + files.len(), + 1, + "a burst must produce one snapshot: {files:?}" + ); + let text = + std::fs::read_to_string(config.join("sessions").join("recovery").join(&files[0])).unwrap(); + for cmd in ["sleep 30", "sleep 31", "sleep 32"] { + assert!(text.contains(cmd), "snapshot must carry {cmd:?}: {text}"); + } + assert!(!s.recovery.dirty, "a completed pass clears the flag"); + let _ = std::fs::remove_dir_all(&dir); +} + +/// An empty fleet never writes -- idle from birth, and after a remove-all +/// that empties the fleet inside the debounce window. Quitting with zero +/// tasks must not clobber the snapshot a user would want back. +#[test] +fn recovery_empty_fleet_never_writes() { + let dir = scratch("recovery_empty"); + let config = dir.join("config"); + let mut s = recovery_sup( + &config, + dir.clone(), + Duration::from_millis(100), + Duration::from_millis(200), + ); + // Idle and empty: both schedules cross without writing. + assert!( + !wait_until(Duration::from_millis(600), || { + s.tick(); + !recovery_files(&config).is_empty() + }), + "an idle empty fleet must never write" + ); + + // Remove-all before any pass runs: writes happen only in `tick`, so + // reading the id from the task set directly keeps this deterministic. + spawn(&mut s, "sleep 30", dir.clone()); + let id = s.tasks[0].id; + s.apply(Command::Remove { id }); + assert!( + !wait_until(Duration::from_millis(600), || { + s.tick(); + !recovery_files(&config).is_empty() + }), + "a fleet emptied before the pass must never write" + ); + let _ = std::fs::remove_dir_all(&dir); +} + +/// A persistently unwritable root degrades silently after ONE status notice, +/// and the fleet stays supervised throughout. +#[test] +fn recovery_write_failure_notices_once_and_keeps_supervising() { + let dir = scratch("recovery_fail"); + let config = dir.join("config"); + // A plain file where `recovery/` must go fails every write attempt. + std::fs::create_dir_all(config.join("sessions")).unwrap(); + std::fs::write(config.join("sessions").join("recovery"), "not a dir").unwrap(); + let mut s = recovery_sup( + &config, + dir.clone(), + Duration::from_millis(10), + Duration::from_millis(50), + ); + spawn(&mut s, "sleep 30", dir.clone()); + + // Cross many debounce and cadence intervals, counting notices. + let mut notices = 0usize; + let deadline = Instant::now() + Duration::from_millis(500); + while Instant::now() < deadline { + s.tick(); + notices += s + .drain() + .iter() + .filter(|e| matches!(e, Event::Status(m) if m.contains("recovery snapshot failed"))) + .count(); + std::thread::sleep(Duration::from_millis(10)); + } + assert_eq!(notices, 1, "persistent failure must notice exactly once"); + + s.tick(); + assert!( + s.drain() + .iter() + .any(|e| matches!(e, Event::Tasks(v) if v.len() == 1)), + "a failing writer must never disturb supervision" + ); + let _ = std::fs::remove_dir_all(&dir); +} + #[path = "supervisor_capture_tests.rs"] mod capture; diff --git a/tests/common/mod.rs b/tests/common/mod.rs index f4524d2..0361933 100644 --- a/tests/common/mod.rs +++ b/tests/common/mod.rs @@ -196,6 +196,11 @@ pub fn start_daemon_raw( let mut cmd = Command::new(env!("CARGO_BIN_EXE_fleetcom")); cmd.arg("--daemon") .env("FLEETCOM_RUNTIME_DIR", &dir) + // Pin the config root beside the runtime dir: the daemon's automatic + // recovery snapshots must never land in the developer's real config + // directory. Tests that care set their own value via `configure` + // (later `env` calls override this one). + .env("FLEETCOM_CONFIG_DIR", dir.join("config")) .stdin(Stdio::null()) .stdout(Stdio::null()) .stderr(Stdio::null()); From a0b324d7b74708631b70e4df5abaaa4f8b197e3c Mon Sep 17 00:00:00 2001 From: Christopher Sardegna Date: Mon, 20 Jul 2026 23:47:29 -0700 Subject: [PATCH 2/8] feat(protocol): recovery-snapshot wire surface (v9) --- src/app.rs | 4 +- src/protocol.rs | 194 ++++++++++++++++++++++++++++++++++++++-- src/session.rs | 158 +++++++++++++++++++++++++++++++- src/supervisor.rs | 114 +++++++++++++++++------ src/supervisor_tests.rs | 161 ++++++++++++++++++++++++++++++++- tests/common/mod.rs | 2 +- 6 files changed, 592 insertions(+), 41 deletions(-) diff --git a/src/app.rs b/src/app.rs index afb9bdd..77b740e 100644 --- a/src/app.rs +++ b/src/app.rs @@ -601,7 +601,9 @@ impl App { self.focused_screen = Some(s); } Event::Status(s) => self.status = Some(s), - Event::Sessions(names) => { + // Recovery entries are ignored here until the picker learns + // to show them (the client work behind this event's v9 shape). + Event::Sessions { names, .. } => { // A shorter list can land while the picker is open; clamp // the selection before it can index past the end. self.session_sel = self.session_sel.min(names.len().saturating_sub(1)); diff --git a/src/protocol.rs b/src/protocol.rs index ccf8e60..0ecb7bb 100644 --- a/src/protocol.rs +++ b/src/protocol.rs @@ -12,7 +12,8 @@ use base64::{Engine as _, engine::general_purpose::STANDARD as B64}; use crate::frame::{KIND_CONTROL, KIND_HELLO, KIND_SCREEN}; /// Wire-protocol version; the handshake rejects mismatched peers. -pub const PROTOCOL_VERSION: u32 = 8; +/// v9 added recovery entries to the `sessions` event and `Command::LoadRecovery`. +pub const PROTOCOL_VERSION: u32 = 9; /// Environment and working directory supplied by the launching client. #[derive(Debug, Clone, PartialEq)] @@ -99,6 +100,10 @@ pub enum Command { SaveSession { name: String }, /// Spawn every command in a named recipe, each in its (existing) dir. LoadSession { name: String }, + /// Spawn every command in a recovery snapshot, addressed by the filename + /// stem from [`Event::Sessions`]. The daemon validates the stem before it + /// touches the filesystem: the stem is a wire string, not a trusted path. + LoadRecovery { stem: String }, /// Ask for the saved recipe names; answered with `Event::Sessions`. Listing /// is core-side like save/load, so the picker shows the same dir they use. ListSessions, @@ -187,8 +192,28 @@ pub enum Event { Screen(ScreenView), /// A one-line notice for the status line (save/load result, spawn error). Status(String), - /// Saved session-recipe names, sorted: the reply to `ListSessions`. - Sessions(Vec), + /// The reply to `ListSessions`: saved session-recipe names (sorted) and + /// recovery snapshots (newest first). + Sessions { + names: Vec, + recovery: Vec, + }, +} + +/// One recovery snapshot in [`Event::Sessions`]: the wire's view of a file +/// under the session root's `recovery/` directory. `stem` is the load key a +/// [`Command::LoadRecovery`] sends back; the rest is picker display state. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct RecoveryEntry { + /// Incarnation filename stem, `-`. + pub stem: String, + /// The snapshot's stored human label, `autosaved `. + pub label: String, + /// Command count across the snapshot's directories. + pub tasks: u32, + /// Seconds since the snapshot file's mtime; 0 when the mtime is unreadable + /// or in the future. + pub age_secs: u64, } /// Process-derived lifecycle state, independent of the user's `tagged` intent. @@ -390,6 +415,30 @@ fn str_vec(v: &jzon::JsonValue) -> Option> { Some(out) } +/// Decode the `recovery` array of a `sessions` event, total by construction: +/// a missing key or non-array value reads as no entries, and a malformed +/// member (non-string stem or label, missing field, count past `u32`) drops +/// that member alone. Unlike [`str_vec`], positions carry no meaning here -- +/// each entry names its own file by stem -- so one junk member from a +/// malformed peer costs itself, not the event. +fn recovery_vec(v: &jzon::JsonValue) -> Vec { + let mut out = Vec::new(); + for m in v.members() { + let entry = || -> Option { + Some(RecoveryEntry { + stem: m["stem"].as_str()?.to_string(), + label: m["label"].as_str()?.to_string(), + tasks: u32::try_from(m["tasks"].as_u64()?).ok()?, + age_secs: m["age"].as_u64()?, + }) + }; + if let Some(e) = entry() { + out.push(e); + } + } + out +} + fn lifecycle_str(l: Lifecycle) -> &'static str { match l { Lifecycle::Active => "active", @@ -620,6 +669,10 @@ pub fn encode_command(cmd: &Command) -> (u8, Vec) { let _ = o.insert("t", "load"); let _ = o.insert("name", name.as_str()); } + Command::LoadRecovery { stem } => { + let _ = o.insert("t", "recover"); + let _ = o.insert("stem", stem.as_str()); + } Command::ListSessions => { let _ = o.insert("t", "list"); } @@ -764,6 +817,9 @@ pub fn decode_command(kind: u8, payload: &[u8]) -> Option { "load" => Command::LoadSession { name: v["name"].as_str()?.to_string(), }, + "recover" => Command::LoadRecovery { + stem: v["stem"].as_str()?.to_string(), + }, "list" => Command::ListSessions, "shutdown" => Command::Shutdown, _ => return None, @@ -816,14 +872,24 @@ pub fn encode_event(ev: &Event) -> (u8, Vec) { let _ = o.insert("msg", msg.as_str()); (KIND_CONTROL, o.dump().into_bytes()) } - Event::Sessions(names) => { + Event::Sessions { names, recovery } => { let mut arr = jzon::JsonValue::new_array(); for n in names { let _ = arr.push(n.as_str()); } + let mut rec = jzon::JsonValue::new_array(); + for r in recovery { + let mut m = jzon::JsonValue::new_object(); + let _ = m.insert("stem", r.stem.as_str()); + let _ = m.insert("label", r.label.as_str()); + let _ = m.insert("tasks", u64::from(r.tasks)); + let _ = m.insert("age", r.age_secs); + let _ = rec.push(m); + } let mut o = jzon::JsonValue::new_object(); let _ = o.insert("t", "sessions"); let _ = o.insert("names", arr); + let _ = o.insert("recovery", rec); (KIND_CONTROL, o.dump().into_bytes()) } Event::Screen(sv) => { @@ -906,7 +972,10 @@ pub fn decode_event(kind: u8, payload: &[u8]) -> Option { Some(Event::Tasks(views)) } "status" => Some(Event::Status(v["msg"].as_str()?.to_string())), - "sessions" => Some(Event::Sessions(str_vec(&v["names"])?)), + "sessions" => Some(Event::Sessions { + names: str_vec(&v["names"])?, + recovery: recovery_vec(&v["recovery"]), + }), _ => None, } } @@ -1063,6 +1132,9 @@ mod tests { Command::LoadSession { name: "home".into(), }, + Command::LoadRecovery { + stem: "20260714-093015-4242".into(), + }, Command::ListSessions, Command::Shutdown, ]; @@ -1525,13 +1597,123 @@ mod tests { Vec::new(), vec!["my session".to_string(), "café ☕".to_string()], ] { - let ev = Event::Sessions(names); + let ev = Event::Sessions { + names, + recovery: Vec::new(), + }; let (k, p) = encode_event(&ev); assert_eq!(k, KIND_CONTROL); assert_eq!(decode_event(k, &p).as_ref(), Some(&ev), "round-trip {ev:?}"); } } + /// Recovery entries round-trip beside the names, and the encoded object + /// carries exactly the `{stem, label, tasks, age}` members. + #[test] + fn sessions_recovery_entries_round_trip_and_pin_the_wire_shape() { + let ev = Event::Sessions { + names: vec!["work".to_string()], + recovery: vec![ + RecoveryEntry { + stem: "20260715-070000-22".into(), + label: "autosaved 2026-07-15 07:00".into(), + tasks: 3, + age_secs: 42, + }, + RecoveryEntry { + stem: "20260714-093015-11".into(), + label: "autosaved 2026-07-14 09:30".into(), + tasks: 1, + age_secs: 90_000, + }, + ], + }; + let (k, p) = encode_event(&ev); + assert_eq!(k, KIND_CONTROL); + assert_eq!( + std::str::from_utf8(&p).unwrap(), + r#"{"t":"sessions","names":["work"],"recovery":[{"stem":"20260715-070000-22","label":"autosaved 2026-07-15 07:00","tasks":3,"age":42},{"stem":"20260714-093015-11","label":"autosaved 2026-07-14 09:30","tasks":1,"age":90000}]}"# + ); + assert_eq!(decode_event(k, &p), Some(ev)); + } + + /// A `sessions` frame without a `recovery` key -- or with a non-array + /// value there -- decodes to an empty entry list, never a dropped event. + #[test] + fn sessions_frame_without_recovery_key_decodes_empty() { + for json in [ + r#"{"t":"sessions","names":["a"]}"#, + r#"{"t":"sessions","names":["a"],"recovery":null}"#, + r#"{"t":"sessions","names":["a"],"recovery":"junk"}"#, + ] { + assert_eq!( + decode_event(KIND_CONTROL, json.as_bytes()), + Some(Event::Sessions { + names: vec!["a".to_string()], + recovery: Vec::new(), + }), + "should tolerate {json}" + ); + } + } + + /// A malformed recovery member drops alone: non-string stems and labels, + /// missing fields, counts past `u32`, negative numbers, and flat strings + /// all cost their member, while well-formed neighbors survive. + #[test] + fn malformed_recovery_members_drop_without_rejecting_the_event() { + let json = r#"{"t":"sessions","names":[],"recovery":[ + {"stem":5,"label":"x","tasks":1,"age":0}, + {"label":"x","tasks":1,"age":0}, + {"stem":"s1","tasks":1,"age":0}, + {"stem":"s2","label":7,"tasks":1,"age":0}, + {"stem":"s3","label":"x","age":0}, + {"stem":"s4","label":"x","tasks":4294967296,"age":0}, + {"stem":"s5","label":"x","tasks":-1,"age":0}, + {"stem":"s6","label":"x","tasks":1,"age":-3}, + {"stem":"s7","label":"x","tasks":1}, + "flat", + {"stem":"good","label":"autosaved","tasks":2,"age":7} + ]}"#; + assert_eq!( + decode_event(KIND_CONTROL, json.as_bytes()), + Some(Event::Sessions { + names: Vec::new(), + recovery: vec![RecoveryEntry { + stem: "good".into(), + label: "autosaved".into(), + tasks: 2, + age_secs: 7, + }], + }) + ); + } + + /// `LoadRecovery` pins its wire form, and a missing or non-string stem + /// rejects the command: the daemon never sees an unvalidatable stem. + #[test] + fn load_recovery_wire_form() { + let (k, p) = encode_command(&Command::LoadRecovery { + stem: "20260714-093015-4242".into(), + }); + assert_eq!(k, KIND_CONTROL); + assert_eq!( + std::str::from_utf8(&p).unwrap(), + r#"{"t":"recover","stem":"20260714-093015-4242"}"# + ); + for json in [ + r#"{"t":"recover"}"#, + r#"{"t":"recover","stem":5}"#, + r#"{"t":"recover","stem":null}"#, + ] { + assert_eq!( + decode_command(KIND_CONTROL, json.as_bytes()), + None, + "should reject {json}" + ); + } + } + /// The `Screen` event keeps its formatted bytes intact through the raw tail, /// including non-UTF-8 bytes (0xFF) an ANSI stream really contains. #[test] diff --git a/src/session.rs b/src/session.rs index b2290d6..8e1efac 100644 --- a/src/session.rs +++ b/src/session.rs @@ -11,6 +11,8 @@ use std::{ time::SystemTime, }; +use crate::protocol::RecoveryEntry; + /// One recipe entry. Entries without a group or name serialize as strings; /// other entries use objects whose optional fields are written only when set. #[derive(Debug, Clone, PartialEq, Eq)] @@ -303,8 +305,8 @@ pub fn list_in(dir: &Path) -> Vec { } // --- recovery snapshots: the supervisor's automatic fleet backups, written -// under `/recovery` in the ordinary wrapped format. Phase 1 -// only writes; listing and loading them arrive in later phases. --------------- +// under `/recovery` in the ordinary wrapped format, and listed +// and loaded by stem for the wire (`list_recovery_in`/`load_recovery_in`). ---- /// Snapshots kept per recovery directory; older ones are pruned after each /// write. @@ -366,6 +368,71 @@ pub fn save_recovery_in( Ok(file) } +/// List recovery snapshots under `dir` as wire entries, newest first (stems +/// lead with a UTC stamp, so descending lexical order is ascending age). +/// Unreadable and unparseable files are skipped silently: one corrupt +/// snapshot must not empty the picker of the intact ones beside it. A file +/// without a stored name labels as its stem, matching `list_in`. +pub fn list_recovery_in(dir: &Path) -> Vec { + let mut out = Vec::new(); + if let Ok(entries) = fs::read_dir(dir) { + for e in entries.flatten() { + let p = e.path(); + if p.extension().and_then(|s| s.to_str()) != Some("json") { + continue; + } + let Some(stem) = p.file_stem().and_then(|s| s.to_str()) else { + continue; + }; + let Ok((stored, cfg)) = fs::read_to_string(&p).and_then(|t| from_json(&t)) else { + continue; + }; + // Saturate both derived numbers: a count past `u32` pins to the + // maximum, and an unreadable or future mtime reads as age 0. + let tasks = + u32::try_from(cfg.values().map(Vec::len).sum::()).unwrap_or(u32::MAX); + let age_secs = fs::metadata(&p) + .and_then(|m| m.modified()) + .ok() + .and_then(|mtime| SystemTime::now().duration_since(mtime).ok()) + .map_or(0, |d| d.as_secs()); + out.push(RecoveryEntry { + stem: stem.to_string(), + label: stored.unwrap_or_else(|| stem.to_string()), + tasks, + age_secs, + }); + } + } + out.sort_by(|a, b| b.stem.cmp(&a.stem)); + out +} + +/// Whether a wire-supplied stem may be joined into a recovery directory. +/// Separators cover traversal (`../x`, `a/b`); rejecting `.` outright also +/// covers extension smuggling and dot-files. Stems the writer generates are +/// digits and dashes only (see [`recovery_stem`]), so a legitimate stem never +/// trips this. +fn valid_recovery_stem(stem: &str) -> bool { + !stem.is_empty() && !stem.contains(['/', '\\', '.']) +} + +/// Load one recovery snapshot by filename stem. `load_in` does not fit here: +/// it maps a user-typed session *name* through `sanitize` into a filename, +/// while this addresses a file by the exact stem the daemon listed -- and the +/// stem arrives over the wire, so it is validated, never rewritten. A stem +/// this module would not have written fails as `InvalidInput` before any +/// path is built from it. +pub fn load_recovery_in(dir: &Path, stem: &str) -> io::Result { + if !valid_recovery_stem(stem) { + return Err(io::Error::new( + io::ErrorKind::InvalidInput, + format!("invalid recovery stem {stem:?}"), + )); + } + from_json(&fs::read_to_string(dir.join(format!("{stem}.json")))?).map(|(_, cfg)| cfg) +} + /// Best-effort prune: keep the newest [`RECOVERY_KEEP`] snapshots by filename /// (stems are UTC timestamps, so lexical order is age order) and ignore every /// error -- a snapshot that cannot be removed must not fail the write that @@ -869,4 +936,91 @@ mod tests { assert_eq!(list_in(&dir), vec!["real".to_string()]); let _ = fs::remove_dir_all(&dir); } + + /// Listing returns newest-first entries with labels and task counts; a + /// corrupt snapshot is skipped, not fatal; a missing directory lists as + /// empty. + #[test] + fn recovery_listing_is_newest_first_and_skips_corrupt_files() { + let base = temp("session_recovery_list"); + let rec = recovery_dir(&base); + assert!( + list_recovery_in(&rec).is_empty(), + "a missing recovery dir must list empty" + ); + + let mut one = SessionConfig::new(); + one.insert("~/a".into(), vec![e("vim")]); + let mut three = SessionConfig::new(); + three.insert("~/a".into(), vec![e("vim"), e("top")]); + three.insert("~/b".into(), vec![e("make")]); + save_recovery_in( + &rec, + "20260714-093015-11", + "autosaved 2026-07-14 09:30", + &one, + ) + .unwrap(); + save_recovery_in( + &rec, + "20260715-070000-22", + "autosaved 2026-07-15 07:00", + &three, + ) + .unwrap(); + fs::write(rec.join("20260716-000000-33.json"), "{not json").unwrap(); + + let entries = list_recovery_in(&rec); + assert_eq!( + entries.len(), + 2, + "the corrupt snapshot must drop alone: {entries:?}" + ); + assert_eq!(entries[0].stem, "20260715-070000-22"); + assert_eq!(entries[0].label, "autosaved 2026-07-15 07:00"); + assert_eq!(entries[0].tasks, 3); + assert_eq!(entries[1].stem, "20260714-093015-11"); + assert_eq!(entries[1].label, "autosaved 2026-07-14 09:30"); + assert_eq!(entries[1].tasks, 1); + assert!( + entries.iter().all(|en| en.age_secs < 3600), + "just-written files must read near-zero ages: {entries:?}" + ); + let _ = fs::remove_dir_all(&base); + } + + /// `load_recovery_in` loads by exact stem; a stem carrying a separator or + /// dot -- or nothing at all -- fails as `InvalidInput` before any path is + /// built, and an unknown stem reads as `NotFound`. + #[test] + fn load_recovery_in_loads_by_stem_and_rejects_traversal() { + let base = temp("session_recovery_load"); + let rec = recovery_dir(&base); + let mut cfg = SessionConfig::new(); + cfg.insert("~/p".into(), vec![gne("cargo run", "api", "server")]); + save_recovery_in( + &rec, + "20260714-093015-11", + "autosaved 2026-07-14 09:30", + &cfg, + ) + .unwrap(); + + assert_eq!(load_recovery_in(&rec, "20260714-093015-11").unwrap(), cfg); + for bad in ["../x", "a/b", "a.b", "a\\b", ""] { + let err = load_recovery_in(&rec, bad).unwrap_err(); + assert_eq!( + err.kind(), + io::ErrorKind::InvalidInput, + "stem {bad:?} must be refused, got {err}" + ); + } + assert_eq!( + load_recovery_in(&rec, "20990101-000000-1") + .unwrap_err() + .kind(), + io::ErrorKind::NotFound + ); + let _ = fs::remove_dir_all(&base); + } } diff --git a/src/supervisor.rs b/src/supervisor.rs index 7b44eff..3b8c00d 100644 --- a/src/supervisor.rs +++ b/src/supervisor.rs @@ -178,7 +178,7 @@ struct Recovery { /// which would otherwise land snapshots in the developer's real config /// root; recovery tests arm explicitly via `set_recovery_timing`. enabled: bool, - /// Set by the six structural recipe mutations, cleared by the next due + /// Set by the seven structural recipe mutations, cleared by the next due /// pass (written, unchanged, empty, or unwritable alike -- see /// `Supervisor::maybe_write_recovery`). dirty: bool, @@ -336,7 +336,7 @@ impl Supervisor { /// Apply one client request. Fire-and-forget: any result (a save/load /// notice, a spawn failure) is queued as `Event::Status`, never returned. pub fn apply(&mut self, cmd: Command) { - // The six structural recipe mutations arm the recovery writer; the + // The seven structural recipe mutations arm the recovery writer; the // burst coalesces behind `RECOVERY_DEBOUNCE` in `tick`. `Tag` is // deliberately absent: tags are not recipe state (a `SessionEntry` // stores cmd, group, and name only), so a tag flip cannot change the @@ -351,6 +351,7 @@ impl Supervisor { | Command::SetGroup { .. } | Command::SetName { .. } | Command::LoadSession { .. } + | Command::LoadRecovery { .. } ) { self.recovery.dirty = true; self.recovery.last_mutation = Some(Instant::now()); @@ -461,6 +462,7 @@ impl Supervisor { } Command::SaveSession { name } => self.save_session(&name), Command::LoadSession { name } => self.load_session(&name), + Command::LoadRecovery { stem } => self.load_recovery(&stem), Command::ListSessions => self.list_sessions(), Command::Shutdown => self.shutdown_all(), } @@ -950,40 +952,31 @@ impl Supervisor { } /// Answer `ListSessions` with the recipe names under this connection's - /// session root (sorted by `list_in`); no root reads as no sessions. + /// session root (sorted by `list_in`) and the recovery snapshots under its + /// `recovery/` directory (newest first by `list_recovery_in`); no root + /// reads as neither. fn list_sessions(&mut self) { - let names = self + let (names, recovery) = self .sessions_root() - .map(|root| session::list_in(&root)) + .map(|root| { + ( + session::list_in(&root), + session::list_recovery_in(&session::recovery_dir(&root)), + ) + }) .unwrap_or_default(); - self.events.push(Event::Sessions(names)); + self.events.push(Event::Sessions { names, recovery }); } - /// Spawn every command in the named session, each in its (existing) dir. + /// Spawn every entry of a loaded recipe, each in its (existing) dir. /// Missing dirs are skipped rather than spawning tasks doomed to fail on - /// chdir. - fn load_session(&mut self, name: &str) { - let Some(root) = self.sessions_root() else { - self.status("load failed: no config directory available"); - return; - }; - let cfg = match session::load_in(&root, name) { - Ok(c) => c, - // Preserve load errors; only a missing file maps to "not found". - Err(e) if e.kind() == io::ErrorKind::NotFound => { - self.status(format!("session '{name}' not found")); - return; - } - Err(e) => { - self.status(format!("session '{name}' failed to load: {e}")); - return; - } - }; - let Some(launch) = self.launch_or_refuse() else { - return; - }; + /// chdir. Returns `(spawned, skipped, failed)` for the caller's notice, or + /// `None` when no launch context is installed (already refused with its + /// own notice). + fn materialize(&mut self, cfg: &SessionConfig) -> Option<(usize, usize, usize)> { + let launch = self.launch_or_refuse()?; let (mut spawned, mut skipped, mut failed) = (0usize, 0usize, 0usize); - for (dir, entries) in &cfg { + for (dir, entries) in cfg { let resolved = path::resolve(&launch.cwd, dir); if !resolved.is_dir() { skipped += entries.len(); @@ -1008,6 +1001,30 @@ impl Supervisor { } } } + Some((spawned, skipped, failed)) + } + + /// Spawn every command in the named session via `materialize`. + fn load_session(&mut self, name: &str) { + let Some(root) = self.sessions_root() else { + self.status("load failed: no config directory available"); + return; + }; + let cfg = match session::load_in(&root, name) { + Ok(c) => c, + // Preserve load errors; only a missing file maps to "not found". + Err(e) if e.kind() == io::ErrorKind::NotFound => { + self.status(format!("session '{name}' not found")); + return; + } + Err(e) => { + self.status(format!("session '{name}' failed to load: {e}")); + return; + } + }; + let Some((spawned, skipped, failed)) = self.materialize(&cfg) else { + return; + }; // Omit zero buckets, except report zero tasks for an empty recipe. let mut parts = Vec::new(); if spawned > 0 || (skipped == 0 && failed == 0) { @@ -1023,6 +1040,45 @@ impl Supervisor { } self.status(format!("loaded '{name}': {}", parts.join(", "))); } + + /// Spawn every command in a recovery snapshot, addressed by the validated + /// filename stem. Mirrors `load_session` except for the notice: a clean + /// load steers the user toward `SaveSession`, because the snapshot's + /// writer overwrites its own file and pruning ages the rest out -- + /// naming the fleet is what makes it durable. + fn load_recovery(&mut self, stem: &str) { + let Some(root) = self.sessions_root() else { + self.status("load failed: no config directory available"); + return; + }; + let cfg = match session::load_recovery_in(&session::recovery_dir(&root), stem) { + Ok(c) => c, + // Preserve load errors; only a missing file maps to "not found". + Err(e) if e.kind() == io::ErrorKind::NotFound => { + self.status(format!("recovery snapshot '{stem}' not found")); + return; + } + Err(e) => { + self.status(format!("recovery snapshot '{stem}' failed to load: {e}")); + return; + } + }; + let Some((_, skipped, failed)) = self.materialize(&cfg) else { + return; + }; + // The success notice is fixed; problem buckets append after it so a + // partial materialization is never reported as clean. + let mut msg = String::from("loaded recovery snapshot; save to name it"); + if skipped > 0 { + msg.push_str(&format!( + ", {skipped} skipped (missing dir, task limit, or command too long)" + )); + } + if failed > 0 { + msg.push_str(&format!(", {failed} failed to spawn")); + } + self.status(msg); + } } #[cfg(test)] diff --git a/src/supervisor_tests.rs b/src/supervisor_tests.rs index 7e7128f..3b85384 100644 --- a/src/supervisor_tests.rs +++ b/src/supervisor_tests.rs @@ -1174,7 +1174,7 @@ fn session_commands_use_the_launch_context_config_dir() { let evs = s.drain(); assert!( evs.iter() - .any(|e| matches!(e, Event::Sessions(n) if n == &["ctx".to_string()])), + .any(|e| matches!(e, Event::Sessions { names, .. } if names == &["ctx".to_string()])), "list must see the recipe save just wrote; got {evs:?}" ); @@ -1637,7 +1637,7 @@ fn take_dirty(s: &mut Supervisor) -> bool { std::mem::replace(&mut s.recovery.dirty, false) } -/// The six structural recipe mutations arm the writer; `Tag` does not (tags +/// The seven structural recipe mutations arm the writer; `Tag` does not (tags /// are not recipe state). Arming keys on the command, not its outcome, so a /// refused `Restart` and a missing `LoadSession` recipe still arm. #[test] @@ -1676,10 +1676,167 @@ fn recovery_arms_on_structural_mutations_not_tag() { }); assert!(take_dirty(&mut s), "LoadSession must arm"); + s.apply(Command::LoadRecovery { + stem: "20990101-000000-1".into(), + }); + assert!(take_dirty(&mut s), "LoadRecovery must arm"); + s.apply(Command::Remove { id }); assert!(take_dirty(&mut s), "Remove must arm"); } +/// `ListSessions` answers with the recovery snapshots newest first beside the +/// recipe names, both from the connection's session root. +#[test] +fn list_sessions_includes_recovery_snapshots_newest_first() { + let dir = scratch("recovery_list_wire"); + let config = dir.join("config"); + let mut s = Supervisor::new(24, 80, 2000); + s.set_launch_context(config_ctx(&config, dir.clone(), &[])); + + let rec = config.join("sessions").join("recovery"); + let entry = |cmd: &str| SessionEntry { + cmd: cmd.into(), + group: None, + name: None, + }; + let mut one = SessionConfig::new(); + one.insert("~/a".into(), vec![entry("vim")]); + let mut two = SessionConfig::new(); + two.insert("~/a".into(), vec![entry("vim"), entry("top")]); + session::save_recovery_in( + &rec, + "20260714-093015-11", + "autosaved 2026-07-14 09:30", + &one, + ) + .unwrap(); + session::save_recovery_in( + &rec, + "20260715-070000-22", + "autosaved 2026-07-15 07:00", + &two, + ) + .unwrap(); + + s.apply(Command::ListSessions); + let evs = s.drain(); + let (names, recovery) = evs + .iter() + .find_map(|e| match e { + Event::Sessions { names, recovery } => Some((names, recovery)), + _ => None, + }) + .expect("a Sessions reply"); + assert!(names.is_empty(), "no recipes were saved; got {names:?}"); + let summary: Vec<(&str, &str, u32)> = recovery + .iter() + .map(|r| (r.stem.as_str(), r.label.as_str(), r.tasks)) + .collect(); + assert_eq!( + summary, + vec![ + ("20260715-070000-22", "autosaved 2026-07-15 07:00", 2), + ("20260714-093015-11", "autosaved 2026-07-14 09:30", 1), + ], + "snapshots must list newest first with labels and task counts" + ); + let _ = std::fs::remove_dir_all(&dir); +} + +/// `LoadRecovery` materializes the snapshot through the session-load path -- +/// groups and names included -- and reports the fixed notice steering the +/// user toward `SaveSession`. +#[test] +fn load_recovery_materializes_the_fleet_and_notices() { + let dir = scratch("recovery_load_wire"); + let config = dir.join("config"); + let mut s = Supervisor::new(24, 80, 2000); + s.set_launch_context(config_ctx(&config, dir.clone(), &[])); + + let mut cfg = SessionConfig::new(); + cfg.insert( + dir.to_string_lossy().into_owned(), + vec![ + SessionEntry { + cmd: "sleep 30".into(), + group: Some("api".into()), + name: None, + }, + SessionEntry { + cmd: "sleep 31".into(), + group: None, + name: Some("web".into()), + }, + ], + ); + session::save_recovery_in( + &config.join("sessions").join("recovery"), + "20260714-093015-11", + "autosaved 2026-07-14 09:30", + &cfg, + ) + .unwrap(); + + s.apply(Command::LoadRecovery { + stem: "20260714-093015-11".into(), + }); + let evs = s.drain(); + assert!( + evs.iter().any(|e| matches!( + e, + Event::Status(m) if m == "loaded recovery snapshot; save to name it" + )), + "a clean load must report exactly the rename-steering notice; got {evs:?}" + ); + assert_eq!(s.tasks.len(), 2, "both snapshot commands must spawn"); + let by_cmd = |s: &Supervisor, cmd: &str| { + let t = s + .tasks + .iter() + .find(|t| t.command == cmd) + .unwrap_or_else(|| panic!("task '{cmd}' missing after load")); + (t.group.clone(), t.name.clone()) + }; + assert_eq!(by_cmd(&s, "sleep 30"), (Some("api".into()), None)); + assert_eq!(by_cmd(&s, "sleep 31"), (None, Some("web".into()))); + let _ = std::fs::remove_dir_all(&dir); +} + +/// An unknown stem reads as not-found and a traversal-shaped stem is refused +/// before any path is built; neither panics or spawns anything. +#[test] +fn load_recovery_refuses_unknown_and_traversal_stems() { + let dir = scratch("recovery_load_refuse"); + let config = dir.join("config"); + let mut s = Supervisor::new(24, 80, 2000); + s.set_launch_context(config_ctx(&config, dir.clone(), &[])); + + s.apply(Command::LoadRecovery { + stem: "20990101-000000-1".into(), + }); + assert!( + s.drain().iter().any(|e| matches!( + e, + Event::Status(m) if m == "recovery snapshot '20990101-000000-1' not found" + )), + "an unknown stem must read as not-found" + ); + + s.apply(Command::LoadRecovery { + stem: "../x".into(), + }); + assert!( + s.drain().iter().any(|e| matches!( + e, + Event::Status(m) if m.starts_with("recovery snapshot '../x' failed to load:") + )), + "a traversal stem must be refused, not probed" + ); + assert!(s.tasks.is_empty(), "refused loads must spawn nothing"); + let _ = std::fs::remove_dir_all(&dir); +} + /// A burst of mutations coalesces behind the debounce into one snapshot /// carrying the whole burst: no write lands inside the quiet period, and one /// file holds all three commands afterward. diff --git a/tests/common/mod.rs b/tests/common/mod.rs index 0361933..15abb9a 100644 --- a/tests/common/mod.rs +++ b/tests/common/mod.rs @@ -17,7 +17,7 @@ use std::{ /// The protocol version this test suite speaks; must track /// `protocol::PROTOCOL_VERSION` (drift fails the handshake, loudly). -pub const PROTOCOL_VERSION: u32 = 8; +pub const PROTOCOL_VERSION: u32 = 9; /// One frame of the given kind: `[u32 len][kind][payload]`. pub fn frame(kind: u8, payload: &[u8]) -> Vec { From c82aed39ed09672da7437ac346909153b8b5697f Mon Sep 17 00:00:00 2001 From: Christopher Sardegna Date: Mon, 20 Jul 2026 23:58:38 -0700 Subject: [PATCH 3/8] feat(ui): recovery page in the session picker; document recovery snapshots --- README.md | 2 +- docs/commands.md | 8 +- docs/sessions.md | 17 ++++ src/app.rs | 93 ++++++++++++++---- src/app_tests.rs | 239 +++++++++++++++++++++++++++++++++++++++++++++++ src/ui.rs | 131 ++++++++++++++++++++++---- 6 files changed, 455 insertions(+), 35 deletions(-) diff --git a/README.md b/README.md index cc4f345..dcbe0f8 100644 --- a/README.md +++ b/README.md @@ -28,7 +28,7 @@ Running several long-lived commands is pesky once they span terminal panes or ne - Runs each command in its own PTY and groups tasks by state, working directory, or named group. - Delegates tasks to a daemon, so a disconnecting client stops nothing. -- Saves and reloads task recipes: directories, commands, group assignments, and display names. +- Saves and reloads task recipes: directories, commands, group assignments, and display names — and snapshots the running fleet automatically, so a lost fleet is recoverable. - Reruns a completed task in place, keeping its identity, group, and name. - Preserves `claude`, `codex`, and `grok` conversations, so saved or rerun tasks resume instead of starting fresh. diff --git a/docs/commands.md b/docs/commands.md index 98a0b39..b522001 100644 --- a/docs/commands.md +++ b/docs/commands.md @@ -33,7 +33,7 @@ | `r` | Rerun a finished task; supported agent tasks use the captured resume command | | `X` | Kill a running task (`TERM`, then `KILL` after 2 s), or remove a finished one | | `w` | Save the current tasks as a session | -| `o` | Load a saved session | +| `o` | Load a saved session or a recovery snapshot (opens the [session picker](#the-o-session-picker)) | | `q` (or `Ctrl-C`) | Disconnect; leave the daemon and tasks running | | `Q` | Quit; kill the tasks and stop the daemon | @@ -130,6 +130,12 @@ Typing filters the rows; `Backspace` deletes one character and the matches re-fi The daemon normalizes every group name received from the picker or a [session](sessions.md) file. It removes control characters, trims surrounding whitespace, and caps the result at 64 characters. An empty result or the exact name `Unassigned` means no group, preventing a user-defined name from colliding with the reserved section. Comparison remains case-sensitive, so `unassigned` is a valid group name. +## The `o` session picker + +`o` opens a bottom panel listing the saved [sessions](sessions.md): `↑`/`↓` move the highlight, `Enter` loads, `Esc` cancels. While [recovery snapshots](sessions.md#recovery) exist, the hint adds `tab recovery (N)` and `Tab` (or `Shift-Tab`) flips the panel to them; `Tab` again returns to the saved list. Each list keeps its own highlight. With no snapshots, `Tab` does nothing and the hint omits it. + +A recovery row reads ` ago · task(s) ·