From 187bcb01d6f4d8071eded156f72f5908d9dee108 Mon Sep 17 00:00:00 2001 From: Miguel Palhas Date: Mon, 24 Aug 2026 13:04:30 +0100 Subject: [PATCH] feat(maestro): refresh the blocked-work indicator in the watcher (#83) run_blocked_work_pass rides the liveness tick but probes on a slower cadence (DEFAULT_BLOCKED_WORK_INTERVAL, 15s): each probe is three git subprocesses per session. Sessions without a worktree are filtered out before any probe, so their indicator costs nothing. A failed probe keeps the previous value; settled exits clear it. --- crates/maestro-daemon/src/session.rs | 86 ++++++++++++- .../maestro-daemon/tests/blocked_work_cost.rs | 116 ++++++++++++++++++ .../maestro-daemon/tests/session_manager.rs | 64 +++++++++- 3 files changed, 264 insertions(+), 2 deletions(-) create mode 100644 crates/maestro-daemon/tests/blocked_work_cost.rs diff --git a/crates/maestro-daemon/src/session.rs b/crates/maestro-daemon/src/session.rs index 7d0c44e..fd27f50 100644 --- a/crates/maestro-daemon/src/session.rs +++ b/crates/maestro-daemon/src/session.rs @@ -33,6 +33,13 @@ pub const EVENTS_CAPACITY: usize = maestro_config::EVENTS_CAPACITY; /// How often the liveness watcher polls tmux for sessions that disappeared. pub const DEFAULT_LIVENESS_INTERVAL: Duration = Duration::from_secs(2); +/// How often the liveness watcher re-probes worktrees for the blocked-work +/// indicator (#83). Each probe costs three git subprocesses per session, so +/// the pass rides the same tick as everything else but runs on a slower +/// cadence: an indicator a few seconds stale still warns before the kill +/// attempt, which is all #83 needs. +pub const DEFAULT_BLOCKED_WORK_INTERVAL: Duration = Duration::from_secs(15); + /// Per-session daemon-local bookkeeping that complements the public `Session`. #[derive(Debug)] pub struct SessionRecord { @@ -492,6 +499,17 @@ impl SessionManager { /// `SessionManager` clone is dropped — call this once from the daemon /// startup path after [`Self::recover`] has populated the map. pub fn start_liveness_watcher(&self, interval: Duration) { + self.start_liveness_watcher_with_blocked_work(interval, DEFAULT_BLOCKED_WORK_INTERVAL); + } + + /// [`Self::start_liveness_watcher`] with the blocked-work cadence (#83) + /// made explicit. Tests pass a small `blocked_work_interval` to watch the + /// probe actually happen. + pub fn start_liveness_watcher_with_blocked_work( + &self, + interval: Duration, + blocked_work_interval: Duration, + ) { let weak = Arc::downgrade(&self.inner); tokio::spawn(async move { let mut ticker = tokio::time::interval(interval); @@ -499,6 +517,9 @@ impl SessionManager { // first tick fires immediately — skip it so we don't reap before // the caller has a chance to create anything. ticker.tick().await; + // Start "due" so the first live tick probes: a recovered daemon + // should not wait a full interval to show what blocks. + let mut since_probe = blocked_work_interval; loop { ticker.tick().await; let Some(inner) = weak.upgrade() else { @@ -508,6 +529,11 @@ impl SessionManager { // Same tick, no new timer (NAP-340): refresh activity for the // survivors and emit SessionUpdated on any transition. run_activity_pass(&inner).await; + since_probe += interval; + if since_probe >= blocked_work_interval { + since_probe = Duration::ZERO; + run_blocked_work_pass(&inner).await; + } } }); } @@ -813,6 +839,61 @@ async fn run_activity_pass(inner: &Arc) { } } +/// Refresh each running session's blocked-work indicator (#83): what its +/// worktree holds right now that would block an unforced kill. Rides the +/// liveness tick on its own (slower) cadence — see +/// [`DEFAULT_BLOCKED_WORK_INTERVAL`] — because every probe is three git +/// subprocesses per session. +/// +/// Sessions without a worktree are filtered out BEFORE any probe: they can +/// never block a kill, so their indicator must cost nothing at all. A failed +/// probe keeps the previous value rather than clearing it — a transient git +/// error must not read as "nothing blocks". Like the activity pass, this only +/// emits `SessionUpdated` when the value actually changed. +async fn run_blocked_work_pass(inner: &Arc) { + let targets: Vec<(Uuid, (PathBuf, PathBuf))> = { + let sessions = inner.sessions.lock().await; + sessions + .iter() + .filter(|(_, r)| r.session.status == SessionStatus::Running) + .filter_map(|(id, r)| { + let wt = r.worktree_base_repo.clone().zip(r.worktree_path.clone())?; + Some((*id, wt)) + }) + .collect() + }; + if targets.is_empty() { + return; + } + + for (id, wt) in targets { + // The kill guard's own probe (#79), so the indicator can never + // disagree with what a kill attempt would find. + let breakdown = match probe_worktree(&wt).await { + Ok(breakdown) => breakdown, + Err(e) => { + tracing::debug!(%id, error = ?e, "blocked-work: probe failed; keeping previous value"); + continue; + } + }; + let updated = { + let mut sessions = inner.sessions.lock().await; + let Some(rec) = sessions.get_mut(&id) else { + continue; + }; + if rec.session.blocked_work == Some(breakdown) { + None + } else { + rec.session.blocked_work = Some(breakdown); + Some(rec.session.clone()) + } + }; + if let Some(session) = updated { + let _ = inner.events.send(CatalogEvent::SessionUpdated { session }); + } + } +} + /// Mark a session whose tmux session has disappeared as [`SessionStatus::Dead`] /// and settle its worktree per the natural-exit rule (#82). /// @@ -856,10 +937,13 @@ async fn mark_session_dead(inner: &Arc, id: Uuid) { if let Some(wt) = &worktree { if settle_worktree_on_exit(inner, id, wt, branch).await { - session.worktree = None; + // The tree is settled (removed or retained); a stale "in the + // way" would describe something the session no longer holds. + session.blocked_work = None; let mut sessions = inner.sessions.lock().await; if let Some(record) = sessions.get_mut(&id) { record.session.worktree = None; + record.session.blocked_work = None; record.worktree_path = None; record.worktree_base_repo = None; record.worktree_branch = None; diff --git a/crates/maestro-daemon/tests/blocked_work_cost.rs b/crates/maestro-daemon/tests/blocked_work_cost.rs new file mode 100644 index 0000000..a04a923 --- /dev/null +++ b/crates/maestro-daemon/tests/blocked_work_cost.rs @@ -0,0 +1,116 @@ +//! The performance contract of #83: the blocked-work pass must cost NO git +//! probe at all for a session without a worktree, while a session with one is +//! still probed. +//! +//! `git` itself is shimmed through PATH with a wrapper that logs every +//! invocation before exec'ing the real binary, so the assertion counts actual +//! probes rather than inferring them from state. This lives in its own test +//! binary because it mutates the process environment. + +mod common; + +use std::collections::HashMap; +use std::process::Command; +use std::time::Duration; + +use common::{init_bare_repo, init_repo, push_all_to_origin, wait_for_async, DaemonFixture}; +use maestro_protocol::{CreateSessionRequest, InTheWay, WorktreeSpec}; +use tempfile::TempDir; + +fn request(cwd: &std::path::Path, repo: &std::path::Path, branch: &str) -> CreateSessionRequest { + CreateSessionRequest { + cwd: cwd.to_path_buf(), + command: "sleep 30".into(), + env: HashMap::new(), + metadata: HashMap::new(), + worktree: Some(WorktreeSpec { + base_repo: repo.to_path_buf(), + branch: branch.into(), + create_branch: true, + }), + } +} + +#[tokio::test] +async fn a_session_without_a_worktree_costs_no_probe() { + let fx = DaemonFixture::new(); + let log = TempDir::new().unwrap(); + let log_path = log.path().join("git-calls.log"); + + // Session A owns a real (pristine, pushed) worktree; session B has none. + let repo_a = TempDir::new().unwrap(); + init_repo(repo_a.path()); + init_bare_repo(&repo_a.path().join("origin.git")); + push_all_to_origin(repo_a.path(), &repo_a.path().join("origin.git")); + fx.manager + .create(request(repo_a.path(), repo_a.path(), "feat/probed")) + .await + .expect("create A"); + + let cwd_b = TempDir::new().unwrap(); + let mut req_b = request(cwd_b.path(), cwd_b.path(), "feat/unused"); + req_b.worktree = None; + fx.manager.create(req_b).await.expect("create B"); + + // Shim `git` AFTER setup so only watcher-driven probes get logged. + let real_git = String::from_utf8( + Command::new("sh") + .arg("-c") + .arg("command -v git") + .output() + .expect("locate git") + .stdout, + ) + .unwrap() + .trim() + .to_string(); + let shim_dir = TempDir::new().unwrap(); + let script = format!( + "#!/bin/sh\nprintf '%s\\n' \"$*\" >> {}\nexec {} \"$@\"\n", + log_path.display(), + real_git, + ); + std::fs::write(shim_dir.path().join("git"), script).unwrap(); + #[cfg(unix)] + { + use std::os::unix::fs::PermissionsExt; + std::fs::set_permissions( + shim_dir.path().join("git"), + std::fs::Permissions::from_mode(0o755), + ) + .unwrap(); + } + let old_path = std::env::var("PATH").unwrap_or_default(); + std::env::set_var( + "PATH", + format!("{}:{}", shim_dir.path().display(), old_path), + ); + + fx.manager.start_liveness_watcher_with_blocked_work( + Duration::from_millis(50), + Duration::from_millis(100), + ); + + // The pass demonstrably runs and probed A's tree... + wait_for_async(Duration::from_secs(5), || async { + fx.manager + .snapshot() + .await + .iter() + .any(|s| s.worktree.is_some() && s.blocked_work == Some(InTheWay::default())) + }) + .await; + + std::env::set_var("PATH", &old_path); + + let calls = std::fs::read_to_string(&log_path).unwrap_or_default(); + assert!( + !calls.is_empty(), + "the pass never probed anything — the shim saw no git call at all" + ); + // ...while B — no worktree, so nothing to probe — was never touched. + assert!( + !calls.contains(cwd_b.path().to_str().unwrap()), + "a worktree-less session was probed:\n{calls}" + ); +} diff --git a/crates/maestro-daemon/tests/session_manager.rs b/crates/maestro-daemon/tests/session_manager.rs index 12d5c5e..6b4f016 100644 --- a/crates/maestro-daemon/tests/session_manager.rs +++ b/crates/maestro-daemon/tests/session_manager.rs @@ -367,6 +367,67 @@ async fn liveness_watcher_marks_dead_sessions_and_keeps_them_until_dismissed() { assert!(!fx.fifo_dir().join(format!("{}.fifo", s.id)).exists()); } +/// #83: the watcher refreshes `Session.blocked_work` on its own cadence — +/// first "nothing blocks" for a pristine tree, then the breakdown once the +/// tree picks up unsaved work, so clients can warn before the kill attempt. +#[tokio::test] +async fn watcher_refreshes_the_blocked_work_indicator() { + let fx = DaemonFixture::new(); + let repo = TempDir::new().unwrap(); + init_repo(repo.path()); + let wt_path = repo.path().join("worktrees").join("feat").join("indicator"); + + fx.manager + .create(CreateSessionRequest { + cwd: repo.path().to_path_buf(), + command: "sleep 30".into(), + env: HashMap::new(), + metadata: HashMap::new(), + worktree: Some(WorktreeSpec { + base_repo: repo.path().to_path_buf(), + branch: "feat/indicator".into(), + create_branch: true, + }), + }) + .await + .expect("create"); + + // Pushed and clean: nothing in the way, and the indicator must say so. + let origin = TempDir::new().unwrap(); + init_bare_repo(origin.path()); + push_all_to_origin(repo.path(), origin.path()); + + fx.manager.start_liveness_watcher_with_blocked_work( + Duration::from_millis(50), + Duration::from_millis(100), + ); + + wait_for_async(Duration::from_secs(5), || async { + fx.manager.snapshot().await[0].blocked_work == Some(maestro_protocol::InTheWay::default()) + }) + .await; + + // Unsaved work appears; the next pass must surface it. + std::fs::write(wt_path.join("README"), b"unsaved edits\n").unwrap(); + std::fs::write(wt_path.join("notes.txt"), b"untracked\n").unwrap(); + wait_for_async(Duration::from_secs(5), || async { + matches!( + &fx.manager.snapshot().await[0].blocked_work, + Some(b) if b.uncommitted_files >= 1 && b.untracked_files >= 1 + ) + }) + .await; + + // A steady state must not re-emit every tick (#83 change-detection). + let mut events = fx.manager.events(); + while events.try_recv().is_ok() {} + tokio::time::sleep(Duration::from_millis(400)).await; + assert!( + !matches!(events.try_recv(), Ok(CatalogEvent::SessionUpdated { .. })), + "an unchanged worktree must not re-emit SessionUpdated" + ); +} + /// Cleanup on kill is unconditional once the guard passes (#79): even a spec /// that says "don't clean up" loses its worktree — the branch was pushed, so /// nothing in the tree is unique and removal goes ahead. @@ -721,7 +782,8 @@ async fn natural_exit_of_a_guarded_worktree_retains_it_in_the_store() { // settled, so dismissal is neither refused nor retained twice. // (`Dead` is observable a moment before the settle step finishes.) wait_for_async(Duration::from_secs(3), || async { - fx.manager.snapshot().await[0].worktree.is_none() + let s = &fx.manager.snapshot().await[0]; + s.worktree.is_none() && s.blocked_work.is_none() }) .await; let snap = fx.manager.snapshot().await;