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.
This commit is contained in:
Miguel Palhas
2026-08-24 13:04:30 +01:00
parent 003e3839bc
commit 187bcb01d6
3 changed files with 264 additions and 2 deletions
+85 -1
View File
@@ -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<Inner>) {
}
}
/// 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<Inner>) {
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<Inner>, 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;
@@ -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}"
);
}
+63 -1
View File
@@ -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;