From 8370ddf066fd81cada556d320d9beca7d0d1c9a5 Mon Sep 17 00:00:00 2001 From: Miguel Palhas Date: Mon, 24 Aug 2026 21:36:56 +0100 Subject: [PATCH] feat(maestro): hold agent-reported activity above inference for a TTL --- crates/maestro-daemon/src/activity.rs | 59 +++++++ crates/maestro-daemon/src/lib.rs | 1 + crates/maestro-daemon/src/session.rs | 78 ++++++++- .../maestro-daemon/tests/activity_report.rs | 164 ++++++++++++++++++ 4 files changed, 301 insertions(+), 1 deletion(-) create mode 100644 crates/maestro-daemon/tests/activity_report.rs diff --git a/crates/maestro-daemon/src/activity.rs b/crates/maestro-daemon/src/activity.rs index 7405bb5..5ece6ed 100644 --- a/crates/maestro-daemon/src/activity.rs +++ b/crates/maestro-daemon/src/activity.rs @@ -15,6 +15,44 @@ use maestro_protocol::ActivityState; /// Active↔Idle. This window *is* the debounce the issue calls for. pub const ACTIVE_WINDOW: Duration = Duration::from_secs(3); +/// How long an agent-reported state (#105) stays in force before inference +/// resumes. Long enough for a working agent's heartbeat to hold between +/// reports, short enough that an agent dying mid-run cannot pin its session +/// at `Error` forever. +pub const REPORT_TTL: Duration = Duration::from_secs(300); + +/// Server-side cap (#105) on a reported `detail`, in characters. The meta +/// line renders it inline, so an unbounded string would own the whole card. +pub const MAX_REPORT_DETAIL_CHARS: usize = 256; + +/// An agent-reported activity state (#105) still within its TTL. While one +/// is [`Self::in_force`], it outranks [`decide`]'s inference; when it +/// lapses, inference resumes and any reported detail is cleared with it. +#[derive(Debug, Clone)] +pub struct ReportedActivity { + pub state: ActivityState, + pub expires_at: std::time::Instant, +} + +impl ReportedActivity { + pub fn new(state: ActivityState, ttl: Duration) -> Self { + Self { + state, + expires_at: std::time::Instant::now() + ttl, + } + } + + pub fn in_force(&self, now: std::time::Instant) -> bool { + now < self.expires_at + } +} + +/// Cap a reported detail to [`MAX_REPORT_DETAIL_CHARS`] characters, +/// char-boundary safe. +pub fn truncate_detail(detail: Option) -> Option { + detail.map(|d| d.chars().take(MAX_REPORT_DETAIL_CHARS).collect()) +} + /// Combine the observed signals into an [`ActivityState`]: /// recent output ⇒ `Active`; else blocked on a terminal read ⇒ `AwaitingInput`; /// else `Idle`. @@ -59,4 +97,25 @@ mod tests { ActivityState::Idle ); } + + #[test] + fn report_is_in_force_until_exactly_its_expiry() { + let ttl = Duration::from_secs(300); + let report = ReportedActivity::new(ActivityState::Error, ttl); + assert!(report.in_force(report.expires_at - Duration::from_millis(1))); + assert!(!report.in_force(report.expires_at)); + } + + #[test] + fn detail_truncates_to_the_cap_without_panic_on_multibyte_input() { + let long = "é".repeat(MAX_REPORT_DETAIL_CHARS + 10); + let capped = truncate_detail(Some(long)).unwrap(); + assert_eq!(capped.chars().count(), MAX_REPORT_DETAIL_CHARS); + + assert_eq!(truncate_detail(None), None); + assert_eq!( + truncate_detail(Some("short".into())).as_deref(), + Some("short") + ); + } } diff --git a/crates/maestro-daemon/src/lib.rs b/crates/maestro-daemon/src/lib.rs index 8f85980..b92689e 100644 --- a/crates/maestro-daemon/src/lib.rs +++ b/crates/maestro-daemon/src/lib.rs @@ -27,5 +27,6 @@ pub mod tmux; pub mod worktree; pub mod worktree_probe; +pub use activity::MAX_REPORT_DETAIL_CHARS; pub use config::{default_config_path, default_state_dir, load_config, DaemonConfig, EnvOverrides}; pub use session::{SessionError, SessionManager}; diff --git a/crates/maestro-daemon/src/session.rs b/crates/maestro-daemon/src/session.rs index 82c1a94..0575760 100644 --- a/crates/maestro-daemon/src/session.rs +++ b/crates/maestro-daemon/src/session.rs @@ -57,6 +57,11 @@ pub struct SessionRecord { /// fall back to a probe. pub worktree_branch: Option, pub bridge: PtyBridge, + /// Agent-reported activity (#105) still within its TTL. While in force it + /// outranks the activity pass's own inference; when it lapses, inference + /// resumes and [`Session::activity_detail`] is cleared with it. This + /// field is daemon-local bookkeeping only — it never rides the wire. + pub reported_activity: Option, } #[derive(Debug, thiserror::Error)] @@ -136,6 +141,9 @@ struct Inner { /// the variable. A mutex because the runner sets it after construction, /// possibly while clones are already serving. daemon_url: std::sync::Mutex>, + /// TTL an agent-reported activity state (#105) stays in force. A const in + /// production; overridable for tests that watch the lapse happen. + report_ttl: Duration, sessions: Mutex>, events: broadcast::Sender, } @@ -176,6 +184,7 @@ impl SessionManager { spawn_history: None, retained: None, daemon_url: std::sync::Mutex::new(None), + report_ttl: crate::activity::REPORT_TTL, sessions: Mutex::new(HashMap::new()), events, }), @@ -208,6 +217,15 @@ impl SessionManager { self } + /// Override the agent-report TTL (#105). Tests pass a short one to watch + /// inference resume without waiting out [`crate::activity::REPORT_TTL`]. + pub fn with_activity_report_ttl(mut self, ttl: Duration) -> Self { + let inner = Arc::get_mut(&mut self.inner) + .expect("with_activity_report_ttl must be called before the SessionManager is cloned"); + inner.report_ttl = ttl; + self + } + /// Set the daemon's own HTTP base URL (#103), exported to every session /// spawned afterwards as `MAESTRO_URL`. Unlike the `with_*` builders this /// is a plain setter: the daemon runner learns its bound address after @@ -410,6 +428,7 @@ impl SessionManager { worktree_base_repo, worktree_branch: req.worktree.as_ref().map(|w| w.branch.clone()), bridge, + reported_activity: None, }, ); } @@ -637,6 +656,7 @@ impl SessionManager { // recovered session probes git for it at retain time. worktree_branch: None, bridge, + reported_activity: None, }, ); let _ = self @@ -677,6 +697,48 @@ impl SessionManager { .await .map_err(SessionError::Tmux) } + + /// Accept an agent-reported activity state (#105), the only path that can + /// produce `ActivityState::Error`. The report outranks the activity pass's + /// own inference until its TTL lapses — so an agent that dies mid-run + /// cannot pin its session at `Error` forever. An unknown session is + /// [`SessionError::NotFound`]; the reported `detail` is capped + /// server-side. + /// + /// A report that changes the effective state restarts `activity_since` + /// (#104's clock); re-reporting an unchanged state only extends the TTL + /// and stays silent on the event stream at steady state. + pub async fn report_activity( + &self, + id: Uuid, + state: ActivityState, + detail: Option, + ) -> Result<(), SessionError> { + let now = std::time::Instant::now(); + let updated = { + let mut sessions = self.inner.sessions.lock().await; + let rec = sessions.get_mut(&id).ok_or(SessionError::NotFound(id))?; + let detail = crate::activity::truncate_detail(detail); + let changed = rec.session.activity != state || rec.session.activity_detail != detail; + if rec.session.activity != state { + rec.session.activity_since = Utc::now(); + } + rec.session.activity = state; + rec.session.activity_detail = detail; + rec.reported_activity = Some(crate::activity::ReportedActivity::new( + state, + self.inner.report_ttl, + )); + changed.then(|| rec.session.clone()) + }; + if let Some(session) = updated { + let _ = self + .inner + .events + .send(CatalogEvent::SessionUpdated { session }); + } + Ok(()) + } } /// Run a blocking `tmux` subprocess call on the blocking thread pool so it @@ -847,8 +909,22 @@ async fn run_activity_pass(inner: &Arc) { let Some(rec) = sessions.get_mut(&id) else { continue; }; + // #105: an agent report lapses at its TTL — drop it and its + // detail so this tick's inference takes over again. + if let Some(report) = &rec.reported_activity { + if !report.in_force(std::time::Instant::now()) { + rec.reported_activity = None; + rec.session.activity_detail = None; + } + } let idle = rec.bridge.idle_for(); - let activity = crate::activity::decide(idle, awaiting); + // #105: while a report is in force it supplies the state and + // nothing here may overwrite it; only an unreported session is + // decided by inference. + let activity = match &rec.reported_activity { + Some(report) => report.state, + None => crate::activity::decide(idle, awaiting), + }; let foreground = Some(pane.command.clone()); // The pane's live cwd, drives the dir-based session tree; empty means // tmux reported none. diff --git a/crates/maestro-daemon/tests/activity_report.rs b/crates/maestro-daemon/tests/activity_report.rs new file mode 100644 index 0000000..e494856 --- /dev/null +++ b/crates/maestro-daemon/tests/activity_report.rs @@ -0,0 +1,164 @@ +//! Agent-reported activity (#105): `POST /api/sessions/:id/activity` +//! semantics at the SessionManager level — precedence over the daemon's own +//! inference, detail handling, and lapse back to inference at the TTL. + +mod common; + +use std::sync::Arc; +use std::time::Duration; + +use common::TmuxFixture; +use maestro_daemon::sidecar::SidecarStore; +use maestro_daemon::worktree::WorktreeManager; +use maestro_daemon::SessionManager; +use maestro_protocol::{ActivityState, CreateSessionRequest}; +use tempfile::TempDir; +use tokio::time::sleep; +use uuid::Uuid; + +struct Fixture { + manager: SessionManager, + _tmux: TmuxFixture, + _state_dir: TempDir, +} + +async fn fixture(ttl: Duration) -> Fixture { + let tmux = TmuxFixture::new(); + let state_dir = TempDir::new().unwrap(); + let manager = SessionManager::new( + Arc::new(tmux.tmux.clone()), + SidecarStore::new(state_dir.path().join("sessions")), + WorktreeManager::new(None), + state_dir.path().join("fifos"), + ) + .with_activity_report_ttl(ttl); + Fixture { + manager, + _tmux: tmux, + _state_dir: state_dir, + } +} + +async fn spawn_sleeping(manager: &SessionManager) -> Uuid { + manager + .create(CreateSessionRequest { + cwd: "/tmp".into(), + command: "sleep 60".into(), + env: Default::default(), + metadata: Default::default(), + worktree: None, + }) + .await + .unwrap() + .id +} + +async fn observed(manager: &SessionManager, id: Uuid) -> maestro_protocol::Session { + manager + .snapshot() + .await + .into_iter() + .find(|s| s.id == id) + .expect("session still tracked") +} + +#[tokio::test] +async fn report_on_unknown_session_is_not_found() { + let fx = fixture(Duration::from_secs(300)).await; + let err = fx + .manager + .report_activity(Uuid::new_v4(), ActivityState::Idle, None) + .await + .unwrap_err(); + assert!(matches!(err, maestro_daemon::SessionError::NotFound(_))); +} + +#[tokio::test] +async fn report_sets_error_state_and_detail() { + let fx = fixture(Duration::from_secs(300)).await; + let id = spawn_sleeping(&fx.manager).await; + + fx.manager + .report_activity( + id, + ActivityState::Error, + Some("cargo build failed: link error".into()), + ) + .await + .unwrap(); + + let s = observed(&fx.manager, id).await; + assert_eq!(s.activity, ActivityState::Error); + assert_eq!( + s.activity_detail.as_deref(), + Some("cargo build failed: link error") + ); +} + +#[tokio::test] +async fn replacing_a_report_replaces_both_state_and_detail() { + let fx = fixture(Duration::from_secs(300)).await; + let id = spawn_sleeping(&fx.manager).await; + + fx.manager + .report_activity(id, ActivityState::Error, Some("first failure".into())) + .await + .unwrap(); + fx.manager + .report_activity( + id, + ActivityState::AwaitingInput, + Some("needs approval".into()), + ) + .await + .unwrap(); + + let s = observed(&fx.manager, id).await; + assert_eq!(s.activity, ActivityState::AwaitingInput); + assert_eq!(s.activity_detail.as_deref(), Some("needs approval")); +} + +#[tokio::test] +async fn detail_is_capped_server_side() { + let fx = fixture(Duration::from_secs(300)).await; + let id = spawn_sleeping(&fx.manager).await; + + let long = "x".repeat(10_000); + fx.manager + .report_activity(id, ActivityState::Error, Some(long)) + .await + .unwrap(); + + let s = observed(&fx.manager, id).await; + let detail = s.activity_detail.expect("detail present"); + assert_eq!( + detail.chars().count(), + maestro_daemon::MAX_REPORT_DETAIL_CHARS + ); +} + +#[tokio::test] +async fn reported_state_survives_watcher_ticks_until_the_ttl_then_inference_resumes() { + let fx = fixture(Duration::from_millis(150)).await; + let id = spawn_sleeping(&fx.manager).await; + fx.manager.start_liveness_watcher(Duration::from_millis(50)); + + fx.manager + .report_activity(id, ActivityState::Error, Some("agent died mid-run".into())) + .await + .unwrap(); + + // Several watcher ticks inside the TTL: nothing infers `Error`, so the + // report must hold above inference. + sleep(Duration::from_millis(80)).await; + let s = observed(&fx.manager, id).await; + assert_eq!(s.activity, ActivityState::Error); + assert_eq!(s.activity_detail.as_deref(), Some("agent died mid-run")); + + // After the TTL lapses, inference resumes and the detail goes with it. + common::wait_for_async(Duration::from_secs(5), || async { + let s = observed(&fx.manager, id).await; + s.activity != ActivityState::Error && s.activity_detail.is_none() + }) + .await; +}