feat(maestro): hold agent-reported activity above inference for a TTL

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<String>) -> Option<String> {
+    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")
+        );
+    }
 }

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};

crates/maestro-daemon/src/session.rs

@@ -57,6 +57,11 @@ pub struct SessionRecord {
     /// fall back to a probe.
     pub worktree_branch: Option<String>,
     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<crate::activity::ReportedActivity>,
 }
 
 #[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<Option<String>>,
+    /// 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<HashMap<Uuid, SessionRecord>>,
     events: broadcast::Sender<CatalogEvent>,
 }
@@ -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<String>,
+    ) -> 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<Inner>) {
             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.

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;
+}