cmux-host: report the machine's workload so a Cloud idle pause never freezes a dev server, build or cron job

cmux-tui/crates/cmux-host/src/cloud/mod.rs

@@ -15,10 +15,14 @@
 //!   (`acp_activity` socket lines, held 90 s unless renewed), `activity`
 //!   advertised only while the stream is live and `activity-turns` with it
 //!   only while acpmux's lines arrive;
+//! - workload: every 30 s a probe of the machine's processes using CPU,
+//!   listening sockets and timers that are not the system's ([`workload`]),
+//!   reported as `activity.workload` with `activity-workload` while it reads;
+//!   the Cloud reads a machine as idle only when it counts none;
 //! - resume: the supervisor's `Resumed` event (clock-set wake, id
 //!   confirmed) sends one report named `resume`.
 //!
-//! Modules: [`wire`] (contracts, pure), [`sender`] (the two senders, sans
+//! Modules: [`wire`] (contracts, pure), [`workload`] (the workload probe), [`sender`] (the two senders, sans
 //! I/O), [`session`] (inputs to requests, sans I/O), [`client`] (bind and
 //! auth over the [`client::Http`] and [`client::Store`] traits), [`github`]
 //! (GitHub tokens for the machine's git and `gh` over the agent socket),
@@ -61,6 +65,7 @@ pub mod trajectory;
 #[cfg(unix)]
 pub mod trajectory_thread;
 pub mod wire;
+pub mod workload;
 
 #[cfg(all(test, unix))]
 mod github_tests;

cmux-tui/crates/cmux-host/src/cloud/role.rs

@@ -492,6 +492,9 @@ struct Worker {
     relay: Option<Relay>,
     bind_retry_at: Option<u64>,
     bind_backoff: Backoff,
+    /// The workload probe (`workload.rs`): its memory and when it runs next (while bound).
+    workload: super::workload::Tracker,
+    workload_at: Option<u64>,
     rng: SystemRandom,
     epoch: Instant,
 }
@@ -541,6 +544,8 @@ impl Worker {
             relay: None,
             bind_retry_at: None,
             bind_backoff: Backoff::new(2_000, 600_000),
+            workload: super::workload::Tracker::default(),
+            workload_at: None,
             rng: SystemRandom::new(),
             epoch: Instant::now(),
         })
@@ -567,6 +572,7 @@ impl Worker {
                 self.bind_retry_at,
                 self.activity.retry_at,
                 self.bootstrap_retry_at,
+                self.workload_at,
             ]
             .into_iter()
             .flatten()
@@ -963,6 +969,9 @@ impl Worker {
         self.activity = self.activity_reset();
         self.connect_activity();
         self.drive(reqs);
+        // A new session knows no workload yet: the first probe now, the first result one interval on.
+        self.workload = super::workload::Tracker::default();
+        self.workload_at = Some(self.now());
     }
 
     fn activity_reset(&self) -> Activity {
@@ -1068,13 +1077,37 @@ impl Worker {
         if self.activity.retry_at.is_some_and(|at| at <= now) {
             self.connect_activity();
         }
+        if self.workload_at.is_some_and(|at| at <= self.now()) {
+            self.probe_workload();
+        }
         let now = self.now();
         if let Some(live) = self.live.as_mut() {
             let reqs = live.session.fire(now);
             self.drive(reqs);
         }
     }
 
+    /// One workload probe (`workload.rs`); the next one an interval later. A probe that does not
+    /// read makes the workload unknown, and the report then drops `activity-workload`.
+    fn probe_workload(&mut self) {
+        let now = self.now();
+        self.workload_at = Some(now + super::workload::PROBE_INTERVAL_MS);
+        let Some(live) = self.live.as_mut() else { return };
+        let seen = match super::workload::read_snapshot(self.paths.root(), now) {
+            // The first probe has no CPU baseline yet: it changes nothing.
+            Some(snapshot) => match self.workload.observe(&snapshot, wall_ms()) {
+                Some(workload) => Some(workload),
+                None => return,
+            },
+            None => {
+                self.workload = super::workload::Tracker::default();
+                None
+            }
+        };
+        let reqs = live.session.workload(seen, now);
+        self.drive(reqs);
+    }
+
     /// Sends each request and feeds its answer back until nothing is due.
     fn drive(&mut self, mut queue: Vec<OpRequest>) {
         while let Some(req) = queue.pop() {

cmux-tui/crates/cmux-host/src/cloud/session.rs

@@ -12,6 +12,7 @@ use super::wire::{
     ACP_ACTIVITY_TTL_MS, ATTENTION_EMIT_OP, Activity, ActivityChange, DaemonInfo, EVENT_EMIT_OP,
     activity_from_daemon,
 };
+use super::workload::Workload;
 use serde_json::json;
 
 pub struct Session {
@@ -27,6 +28,8 @@ pub struct Session {
     acp: AcpActivity,
     /// The daemon's activity stream is live.
     stream_live: bool,
+    /// The workload probe's last reported result (`workload.rs`); None while it is unknown.
+    workload: Option<Workload>,
 }
 
 /// What the machine's acpmux daemons last said: each one's sessions at work and until when it
@@ -149,6 +152,7 @@ impl Session {
             own: Activity::default(),
             acp: AcpActivity::default(),
             stream_live: false,
+            workload: None,
         }
     }
 
@@ -195,11 +199,7 @@ impl Session {
     /// the activity change together, so the report that goes carries both (cmux#33331: an
     /// expiry's report never still says the count is known).
     fn acp_refresh(&mut self, now: u64) -> Vec<OpRequest> {
-        let daemon = self.base.with_activity(
-            self.stream_live,
-            self.acp.known(),
-            self.acp.phases().is_some(),
-        );
+        let daemon = self.daemon_now();
         let block_changed = &daemon != self.reporter.daemon();
         self.reporter.replace_daemon(daemon);
         let mut out = self.acp_changed(now);
@@ -225,14 +225,43 @@ impl Session {
         self.reporter.set_activity(activity, report, now).into_iter().collect()
     }
 
-    /// The activity a report carries: the daemon's plus acpmux's.
+    /// The activity a report carries: the daemon's plus acpmux's, and the workload.
     fn activity(&self) -> Activity {
-        self.own.with_acp(
+        let mut activity = self.own.with_acp(
             self.acp.sessions(),
             self.acp.last_action_at,
             self.acp.phases(),
             self.acp.last_person_prompt_at,
-        )
+        );
+        activity.workload = self.workload.as_ref().map(Workload::to_json);
+        activity
+    }
+
+    /// The daemon block as it stands: `activity` while the stream is live, `activity-turns` with
+    /// it while acpmux's count is known, `activity-phases` with that while its phases are, and
+    /// `activity-workload` with `activity` while the workload probe reads.
+    fn daemon_now(&self) -> DaemonInfo {
+        self.base
+            .with_activity(self.stream_live, self.acp.known(), self.acp.phases().is_some())
+            .with_workload(self.workload.is_some())
+    }
+
+    /// One workload probe (`workload.rs`; None: it did not read). A report goes when the machine
+    /// turns busy or quiet, when its listening ports or timers change, when a quiet period starts
+    /// anew, and when the probe starts or stops reading. Which processes use CPU while it is busy
+    /// rides along with the next report.
+    pub fn workload(&mut self, seen: Option<Workload>, now: u64) -> Vec<OpRequest> {
+        let next = match (seen, &self.workload) {
+            (Some(seen), Some(cur)) if seen.key() == cur.key() => return Vec::new(),
+            (seen, _) => seen,
+        };
+        if next.is_none() && self.workload.is_none() {
+            return Vec::new();
+        }
+        self.workload = next;
+        self.reporter.replace_daemon(self.daemon_now());
+        let activity = self.activity();
+        self.reporter.set_activity(activity, true, now).into_iter().collect()
     }
 
     /// The activity stream connected or dropped: `activity` is advertised
@@ -245,11 +274,7 @@ impl Session {
     /// The daemon block as it stands: `activity` while the stream is live, `activity-turns` with
     /// it while acpmux's count is known.
     fn daemon_block(&mut self, now: u64) -> Vec<OpRequest> {
-        let daemon = self.base.with_activity(
-            self.stream_live,
-            self.acp.known(),
-            self.acp.phases().is_some(),
-        );
+        let daemon = self.daemon_now();
         if &daemon == self.reporter.daemon() {
             return Vec::new();
         }
@@ -737,6 +762,70 @@ mod tests {
         assert!(read_phases(&json!({"turns": 0}), 0).is_none(), "no block");
     }
 
+    /// The workload probe's result reaches the report as `activity.workload` with
+    /// `activity-workload` (only with `activity`); a report goes when the machine turns busy or
+    /// quiet, not when only which process uses CPU changes; a probe that stops reading withdraws
+    /// the capability, so the Cloud reads the machine as busy.
+    #[test]
+    fn the_workload_reaches_the_report_and_only_a_change_of_state_is_reported() {
+        use super::super::wire::ACTIVITY_WORKLOAD_CAPABILITY;
+        let mut s = session();
+        let send = |s: &mut Session, mut r: Vec<OpRequest>, now: u64| {
+            r.extend(s.fire(now));
+            for req in &r {
+                s.answered(req, &ok(), now, 0.5);
+            }
+            r
+        };
+        let start = s.start(0);
+        send(&mut s, start, 0);
+        let cap = |r: &OpRequest| {
+            r.params["daemon"]["capabilities"]
+                .as_array()
+                .unwrap()
+                .contains(&json!(ACTIVITY_WORKLOAD_CAPABILITY))
+        };
+        let busy = |comm: &str, at: u64| Workload {
+            busy_processes: vec![comm.into()],
+            last_busy_at: Some(at),
+            ..Workload::default()
+        };
+        // Without the activity stream the block rides along, never the capability.
+        let r = s.workload(Some(busy("node", 1)), 20_000);
+        let r = send(&mut s, r, 20_000);
+        assert_eq!(r.len(), 1);
+        assert!(!cap(&r[0]));
+        assert_eq!(r[0].params["activity"]["workload"]["busy_processes"], 1);
+        let r = s.activity_stream(true, 40_000);
+        let r = send(&mut s, r, 40_000);
+        assert!(cap(&r[0]));
+        assert_eq!(r[0].params["activity"]["workload"]["process_names"], json!(["node"]));
+
+        // Another process now uses CPU: still busy, no report.
+        assert!(s.workload(Some(busy("cargo", 2)), 60_000).is_empty());
+        // Quiet: a report, with the last busy time.
+        let quiet = Workload { last_busy_at: Some(2), ..Workload::default() };
+        let r = s.workload(Some(quiet.clone()), 80_000);
+        let r = send(&mut s, r, 80_000);
+        assert_eq!(r.len(), 1);
+        assert_eq!(r[0].params["activity"]["workload"]["busy_processes"], 0);
+        assert_eq!(r[0].params["activity"]["workload"]["last_busy_at"], 2);
+        assert!(s.workload(Some(quiet), 100_000).is_empty(), "nothing changed");
+        // A port opens: a report.
+        let port =
+            Workload { listening_ports: vec![3000], last_busy_at: Some(3), ..Workload::default() };
+        let r = s.workload(Some(port), 120_000);
+        let r = send(&mut s, r, 120_000);
+        assert_eq!(r[0].params["activity"]["workload"]["ports"], json!([3000]));
+        // The probe stops reading: the capability and the block go.
+        let r = s.workload(None, 140_000);
+        let r = send(&mut s, r, 140_000);
+        assert_eq!(r.len(), 1);
+        assert!(!cap(&r[0]));
+        assert!(r[0].params["activity"].get("workload").is_none());
+        assert!(s.workload(None, 160_000).is_empty());
+    }
+
     /// Relayed only when every daemon says so; turns from every daemon.
     #[test]
     fn the_machine_phases_join_every_daemon() {

cmux-tui/crates/cmux-host/src/cloud/wire.rs

@@ -50,6 +50,12 @@ pub const ACTIVITY_TURNS_CAPABILITY: &str = "activity-turns";
 /// work; acpmux `daemon/cloud_activity.rs` names the phases). The Cloud sleeps a machine inside
 /// its turns only from a report that advertises this (cmux#34298).
 pub const ACTIVITY_PHASES_CAPABILITY: &str = "activity-phases";
+/// Advertised with `activity` while the agent's workload probe reads (`workload.rs`): the
+/// report's `activity.workload` counts the processes using CPU, the listening sockets and the
+/// timers that are not the system's. The Cloud reads a machine as idle only from a report that
+/// advertises this and counts none of them, so a dev server, a build or a cron job an agent left
+/// running is never paused.
+pub const ACTIVITY_WORKLOAD_CAPABILITY: &str = "activity-workload";
 /// How long acpmux's count of sessions at work holds without a new `acp_activity` line
 /// (acpmux renews it every 20 s, `daemon/cloud_activity.rs`): a
 /// daemon that died mid-turn does not keep the machine awake for good.
@@ -212,13 +218,31 @@ impl DaemonInfo {
         DaemonInfo { version: self.version.clone(), capabilities }
     }
 
+    /// The same block with `activity-workload` on (the workload is known and `activity` is on)
+    /// or off. [`Self::with_activity`] drops it: apply this after.
+    pub fn with_workload(&self, on: bool) -> DaemonInfo {
+        let mut capabilities: Vec<String> = self
+            .capabilities
+            .iter()
+            .filter(|c| *c != ACTIVITY_WORKLOAD_CAPABILITY)
+            .cloned()
+            .collect();
+        if on && capabilities.iter().any(|c| c == ACTIVITY_CAPABILITY) {
+            capabilities.push(ACTIVITY_WORKLOAD_CAPABILITY.to_owned());
+        }
+        DaemonInfo { version: self.version.clone(), capabilities }
+    }
+
     pub fn to_json(&self) -> Value {
         json!({ "version": self.version, "capabilities": self.capabilities })
     }
 }
 
 fn is_activity_capability(c: &str) -> bool {
-    c == ACTIVITY_CAPABILITY || c == ACTIVITY_TURNS_CAPABILITY || c == ACTIVITY_PHASES_CAPABILITY
+    c == ACTIVITY_CAPABILITY
+        || c == ACTIVITY_TURNS_CAPABILITY
+        || c == ACTIVITY_PHASES_CAPABILITY
+        || c == ACTIVITY_WORKLOAD_CAPABILITY
 }
 
 /// The daemon block from one `identify` answer's `data`: version plus the
@@ -263,6 +287,9 @@ pub struct Activity {
     /// the Cloud keeps the machine running for a while after it (cmux-platform `IdlePolicy.ts`,
     /// the post-chat warm window).
     pub last_person_prompt_at: Option<u64>,
+    /// The agent's workload probe (`workload.rs`), while it reads
+    /// ([`ACTIVITY_WORKLOAD_CAPABILITY`]).
+    pub workload: Option<Value>,
 }
 
 impl Activity {
@@ -281,6 +308,9 @@ impl Activity {
         if let Some(at) = self.last_person_prompt_at {
             out.insert("last_person_prompt_at".to_owned(), json!(at));
         }
+        if let Some(workload) = &self.workload {
+            out.insert("workload".to_owned(), workload.clone());
+        }
         Value::Object(out)
     }
 }
@@ -302,6 +332,7 @@ impl Activity {
             last_agent_action_at: self.last_agent_action_at.max(acp_action_at),
             phases: acp_phases,
             last_person_prompt_at: acp_prompt_at,
+            workload: self.workload.clone(),
         }
     }
 }