feat(maestro): infer session activity state + foreground (Active/Idle/AwaitingInput) (#18)
ci / rust (push) Failing after 2m15s
ci / rust (push) Failing after 2m15s
Add an orthogonal activity dimension to sessions, inferred entirely from
structured tmux/`/proc` state — no PTY byte-stream parsing, no work on the
attach hot path.
Protocol: `ActivityState` enum (Active/Idle/AwaitingInput) + three additive,
serde-defaulted `Session` fields (`foreground`, `activity`, `last_activity_at`)
so the snapshot/diff flow stays forward-compatible with older daemons.
Detection (piggybacks the existing ~2s liveness tick, no new timer):
- foreground command via `tmux list-panes -F '#{pane_pid} #{pane_current_command}
#{pane_tty}'`;
- quiescence from a per-`PtyBridge` last-read `Instant`, bumped in `run_reader`;
- AwaitingInput via best-effort `/proc` walk to the deepest descendant: state
`S` + a non-timer `wchan` + an stdio fd on the pane tty. (`/proc/pid/syscall`
is ptrace-gated and unreadable for own sessions, so `wchan` distinguishes a
blocking read from a plain `sleep`.)
- combine: recent output → Active; else input-blocked → AwaitingInput; else
Idle. Emit `SessionUpdated` only on (activity, foreground) change — no diff
storms at steady state.
`maestro list` gains an ACTIVITY column and shows the live foreground. Pure
combine/parse logic is unit-tested; an e2e asserts `cat` → AwaitingInput and
`sleep` → Idle with correct foreground against real tmux.
Closes #18
Closes #8
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -4,7 +4,8 @@ use std::collections::HashMap;
|
||||
use std::io::Write;
|
||||
|
||||
use maestro_protocol::{
|
||||
ConnectInfo, CreateSessionRequest, DaemonInfo, Session, SessionStatus, WorktreeSpec,
|
||||
ActivityState, ConnectInfo, CreateSessionRequest, DaemonInfo, Session, SessionStatus,
|
||||
WorktreeSpec,
|
||||
};
|
||||
use serde_json::Value;
|
||||
use uuid::Uuid;
|
||||
@@ -126,8 +127,8 @@ fn print_session_table(out: &mut impl Write, sessions: &[Session], daemons: &[Da
|
||||
writeln!(out, "\n{daemon_id} ({hostname})").ok();
|
||||
writeln!(
|
||||
out,
|
||||
" {:<36} {:<8} {:<20} {}",
|
||||
"ID", "STATUS", "LABEL", "COMMAND"
|
||||
" {:<36} {:<8} {:<9} {:<20} {}",
|
||||
"ID", "STATUS", "ACTIVITY", "LABEL", "COMMAND"
|
||||
)
|
||||
.ok();
|
||||
for s in group {
|
||||
@@ -140,10 +141,21 @@ fn print_session_table(out: &mut impl Write, sessions: &[Session], daemons: &[Da
|
||||
SessionStatus::Running => "running",
|
||||
SessionStatus::Dead => "dead",
|
||||
};
|
||||
let activity = match s.activity {
|
||||
ActivityState::Active => "active",
|
||||
ActivityState::Idle => "idle",
|
||||
ActivityState::AwaitingInput => "awaiting",
|
||||
};
|
||||
// Show the live foreground command alongside the spawn command when
|
||||
// the daemon has observed one that differs (NAP-340).
|
||||
let command = match &s.foreground {
|
||||
Some(fg) if fg != &s.command => format!("{} [{}]", s.command, fg),
|
||||
_ => s.command.clone(),
|
||||
};
|
||||
writeln!(
|
||||
out,
|
||||
" {:<36} {:<8} {:<20} {}",
|
||||
s.id, status, label, s.command
|
||||
" {:<36} {:<8} {:<9} {:<20} {command}",
|
||||
s.id, status, activity, label
|
||||
)
|
||||
.ok();
|
||||
}
|
||||
@@ -156,7 +168,7 @@ mod tests {
|
||||
use std::path::PathBuf;
|
||||
|
||||
use chrono::{TimeZone, Utc};
|
||||
use maestro_protocol::{DaemonInfo, Session, SessionStatus};
|
||||
use maestro_protocol::{ActivityState, DaemonInfo, Session, SessionStatus};
|
||||
use serde_json::Value;
|
||||
use uuid::Uuid;
|
||||
|
||||
@@ -172,6 +184,9 @@ mod tests {
|
||||
metadata: meta,
|
||||
status: SessionStatus::Running,
|
||||
created_at: Utc.with_ymd_and_hms(2026, 5, 25, 12, 0, 0).unwrap(),
|
||||
foreground: None,
|
||||
activity: Default::default(),
|
||||
last_activity_at: chrono::Utc::now(),
|
||||
}
|
||||
}
|
||||
|
||||
@@ -205,6 +220,28 @@ mod tests {
|
||||
assert!(s.contains("vm1 (vm1.lan)"), "{s}");
|
||||
assert!(s.contains("personal/maestro"), "{s}");
|
||||
assert!(s.contains("running"), "{s}");
|
||||
assert!(s.contains("ACTIVITY"), "{s}");
|
||||
assert!(s.contains("idle"), "{s}");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn shows_activity_and_foreground() {
|
||||
let mut s1 = session("laptop", "work", "bash", 1);
|
||||
s1.activity = ActivityState::AwaitingInput;
|
||||
s1.foreground = Some("claude".into()); // differs from spawn command
|
||||
let daemons = vec![DaemonInfo {
|
||||
daemon_id: "laptop".into(),
|
||||
hostname: "h".into(),
|
||||
version: "0.1.0".into(),
|
||||
daemon_url: "ws://x".into(),
|
||||
home_dir: String::new(),
|
||||
}];
|
||||
let mut buf = Vec::new();
|
||||
super::print_session_table(&mut buf, &[s1], &daemons);
|
||||
let out = String::from_utf8(buf).unwrap();
|
||||
assert!(out.contains("awaiting"), "{out}");
|
||||
// Live foreground surfaces next to the spawn command.
|
||||
assert!(out.contains("bash [claude]"), "{out}");
|
||||
}
|
||||
|
||||
#[test]
|
||||
|
||||
@@ -0,0 +1,56 @@
|
||||
//! Activity-state inference combine logic (NAP-340).
|
||||
//!
|
||||
//! Pure decision from three structured signals — PTY quiescence, foreground
|
||||
//! process, and `/proc` block-state — so it's unit-testable without tmux or a
|
||||
//! live session. The orchestration (gathering the signals each liveness tick
|
||||
//! and emitting `SessionUpdated` on change) lives in [`crate::session`].
|
||||
|
||||
use std::time::Duration;
|
||||
|
||||
use maestro_protocol::ActivityState;
|
||||
|
||||
/// PTY output more recent than this keeps a session `Active`. It must exceed the
|
||||
/// liveness tick (~2s, [`crate::session::DEFAULT_LIVENESS_INTERVAL`]) so a
|
||||
/// session emitting a chunk every tick stays `Active` instead of flapping
|
||||
/// Active↔Idle. This window *is* the debounce the issue calls for.
|
||||
pub const ACTIVE_WINDOW: Duration = Duration::from_secs(3);
|
||||
|
||||
/// Combine the observed signals into an [`ActivityState`]:
|
||||
/// recent output ⇒ `Active`; else blocked on a terminal read ⇒ `AwaitingInput`;
|
||||
/// else `Idle`.
|
||||
pub fn decide(idle_for: Duration, awaiting_input: bool) -> ActivityState {
|
||||
if idle_for < ACTIVE_WINDOW {
|
||||
ActivityState::Active
|
||||
} else if awaiting_input {
|
||||
ActivityState::AwaitingInput
|
||||
} else {
|
||||
ActivityState::Idle
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
#[test]
|
||||
fn recent_output_is_active_regardless_of_block_state() {
|
||||
assert_eq!(decide(Duration::from_millis(100), false), ActivityState::Active);
|
||||
assert_eq!(decide(Duration::from_millis(100), true), ActivityState::Active);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn quiescent_and_blocked_is_awaiting_input() {
|
||||
assert_eq!(
|
||||
decide(ACTIVE_WINDOW + Duration::from_secs(1), true),
|
||||
ActivityState::AwaitingInput
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn quiescent_and_not_blocked_is_idle() {
|
||||
assert_eq!(
|
||||
decide(ACTIVE_WINDOW + Duration::from_secs(1), false),
|
||||
ActivityState::Idle
|
||||
);
|
||||
}
|
||||
}
|
||||
@@ -571,6 +571,9 @@ mod tests {
|
||||
metadata: Default::default(),
|
||||
status: maestro_protocol::SessionStatus::Running,
|
||||
created_at: chrono::Utc::now(),
|
||||
foreground: None,
|
||||
activity: Default::default(),
|
||||
last_activity_at: chrono::Utc::now(),
|
||||
};
|
||||
handle
|
||||
.complete_forward_create(
|
||||
|
||||
@@ -122,6 +122,9 @@ mod tests {
|
||||
metadata: StdHashMap::new(),
|
||||
status: SessionStatus::Running,
|
||||
created_at: Utc::now(),
|
||||
foreground: None,
|
||||
activity: Default::default(),
|
||||
last_activity_at: chrono::Utc::now(),
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -6,6 +6,7 @@
|
||||
|
||||
#![deny(missing_debug_implementations)]
|
||||
|
||||
pub mod activity;
|
||||
pub mod attach;
|
||||
pub mod broker;
|
||||
pub mod catalog_mirror;
|
||||
@@ -17,6 +18,7 @@ pub mod gitroot;
|
||||
pub mod jwt;
|
||||
pub mod local_api;
|
||||
pub mod local_events;
|
||||
pub mod proc;
|
||||
pub mod pty;
|
||||
pub mod registry_client;
|
||||
pub mod session;
|
||||
|
||||
@@ -0,0 +1,129 @@
|
||||
//! Best-effort `/proc` inspection for the activity watcher (NAP-340, Linux).
|
||||
//!
|
||||
//! Decides whether a session is **blocked waiting on the user**: walk from the
|
||||
//! pane's process down to the deepest descendant (the agent/shell actually on
|
||||
//! the terminal) and check it's parked waiting on terminal input.
|
||||
//!
|
||||
//! `/proc/<pid>/syscall` would say so exactly, but it's ptrace-gated and
|
||||
//! unreadable for our own sessions on a default kernel, so we combine three
|
||||
//! *unprivileged* signals instead:
|
||||
//!
|
||||
//! 1. process state `S` (interruptible sleep — where a blocking `read()` parks),
|
||||
//! 2. its `wchan` is not a pure timer sleep (so `sleep 30`, which also sits in
|
||||
//! `S` with the terminal inherited, is excluded), and
|
||||
//! 3. one of its stdio fds is the pane's tty.
|
||||
//!
|
||||
//! Every step degrades to `false` on a read failure, so a non-Linux host or a
|
||||
//! denied `/proc` read falls back to `Idle`.
|
||||
|
||||
/// Whether the process tree under `pane_pid` looks blocked on terminal input
|
||||
/// from `tty` (e.g. `/dev/pts/3`) — the signal for `AwaitingInput`.
|
||||
pub fn awaiting_input(pane_pid: i32, tty: &str) -> bool {
|
||||
let leaf = deepest_descendant(pane_pid);
|
||||
// Interruptible sleep is where a blocking read parks; anything else
|
||||
// (running, uninterruptible) is busy, not waiting on us.
|
||||
if read_state(leaf) != Some('S') {
|
||||
return false;
|
||||
}
|
||||
// A pure timer sleep (`sleep`, `usleep`) also sits in 'S' with the terminal
|
||||
// inherited — exclude it so it reads as Idle, not AwaitingInput.
|
||||
if is_timer_sleep(leaf) {
|
||||
return false;
|
||||
}
|
||||
fd_points_at_tty(leaf, tty)
|
||||
}
|
||||
|
||||
/// Follow the (last) child link down the process tree to the deepest
|
||||
/// descendant. Bounded so a pathological/cyclic tree can't loop.
|
||||
fn deepest_descendant(pid: i32) -> i32 {
|
||||
let mut cur = pid;
|
||||
for _ in 0..64 {
|
||||
match last_child(cur) {
|
||||
Some(child) if child != cur => cur = child,
|
||||
_ => break,
|
||||
}
|
||||
}
|
||||
cur
|
||||
}
|
||||
|
||||
/// The most-recently-created child of `pid`, via
|
||||
/// `/proc/<pid>/task/<pid>/children` (a space-separated pid list). The last
|
||||
/// entry is the newest child — typically the foreground one an agent spawned.
|
||||
fn last_child(pid: i32) -> Option<i32> {
|
||||
let contents = std::fs::read_to_string(format!("/proc/{pid}/task/{pid}/children")).ok()?;
|
||||
contents.split_whitespace().last()?.parse().ok()
|
||||
}
|
||||
|
||||
/// Read the process state char from `/proc/<pid>/stat`.
|
||||
fn read_state(pid: i32) -> Option<char> {
|
||||
parse_stat_state(&std::fs::read_to_string(format!("/proc/{pid}/stat")).ok()?)
|
||||
}
|
||||
|
||||
/// Extract the state field from a `/proc/<pid>/stat` line. The format is
|
||||
/// `pid (comm) STATE ...`; `comm` can itself contain spaces and parens, so the
|
||||
/// state is the first non-space char *after the final `)`*.
|
||||
fn parse_stat_state(stat: &str) -> Option<char> {
|
||||
let close = stat.rfind(')')?;
|
||||
stat[close + 1..].trim_start().chars().next()
|
||||
}
|
||||
|
||||
/// Whether `pid`'s `wchan` (the kernel symbol it's blocked in) is a timer
|
||||
/// sleep — `hrtimer_nanosleep`, `do_nanosleep`, etc. Those are quiescent but
|
||||
/// *not* waiting on the user.
|
||||
fn is_timer_sleep(pid: i32) -> bool {
|
||||
std::fs::read_to_string(format!("/proc/{pid}/wchan"))
|
||||
.map(|w| is_timer_wchan(&w))
|
||||
.unwrap_or(false)
|
||||
}
|
||||
|
||||
/// Classify a `wchan` symbol as a pure timer sleep. Kept pure for testing.
|
||||
fn is_timer_wchan(wchan: &str) -> bool {
|
||||
wchan.contains("nanosleep")
|
||||
}
|
||||
|
||||
/// Whether any of `pid`'s stdio fds (0/1/2) is the terminal `tty`.
|
||||
fn fd_points_at_tty(pid: i32, tty: &str) -> bool {
|
||||
(0..3).any(|fd| {
|
||||
std::fs::read_link(format!("/proc/{pid}/fd/{fd}"))
|
||||
.map(|target| target.to_string_lossy() == tty)
|
||||
.unwrap_or(false)
|
||||
})
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
#[test]
|
||||
fn parse_stat_state_handles_simple_and_tricky_comm() {
|
||||
assert_eq!(parse_stat_state("1234 (bash) S 1 1234 ..."), Some('S'));
|
||||
assert_eq!(parse_stat_state("1234 (sleep) S 1 ..."), Some('S'));
|
||||
assert_eq!(parse_stat_state("99 (cat) R 1 ..."), Some('R'));
|
||||
// comm with spaces and a paren — only the final ')' delimits it.
|
||||
assert_eq!(parse_stat_state("42 (weird (name)) D 1 42 ..."), Some('D'));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn parse_stat_state_rejects_garbage() {
|
||||
assert_eq!(parse_stat_state("not a stat line"), None);
|
||||
assert_eq!(parse_stat_state("123 (comm)"), None);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn timer_wchan_excludes_sleep_but_not_a_tty_wait() {
|
||||
// `sleep` blocks here — excluded.
|
||||
assert!(is_timer_wchan("hrtimer_nanosleep"));
|
||||
assert!(is_timer_wchan("do_nanosleep"));
|
||||
// A terminal read / poll parks here — not a timer.
|
||||
assert!(!is_timer_wchan("wait_woken"));
|
||||
assert!(!is_timer_wchan("n_tty_read"));
|
||||
assert!(!is_timer_wchan("do_select"));
|
||||
assert!(!is_timer_wchan("0"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn awaiting_input_is_false_for_nonexistent_pid() {
|
||||
// A pid that can't exist → every read fails → false (fall back to Idle).
|
||||
assert!(!awaiting_input(i32::MAX, "/dev/pts/0"));
|
||||
}
|
||||
}
|
||||
@@ -30,6 +30,7 @@ use std::os::unix::ffi::OsStrExt;
|
||||
use std::os::unix::fs::OpenOptionsExt;
|
||||
use std::path::{Path, PathBuf};
|
||||
use std::sync::{Arc, Mutex};
|
||||
use std::time::Instant;
|
||||
|
||||
use tokio::io::unix::AsyncFd;
|
||||
use tokio::io::Interest;
|
||||
@@ -59,6 +60,10 @@ pub struct PtyBridge {
|
||||
fifo_path: PathBuf,
|
||||
sender: broadcast::Sender<Vec<u8>>,
|
||||
replay: ReplayBuffer,
|
||||
/// Monotonic timestamp of the last non-empty read off the FIFO, bumped by
|
||||
/// `run_reader`. The activity watcher reads it to gauge PTY quiescence
|
||||
/// without touching the byte stream (NAP-340).
|
||||
last_activity: Arc<Mutex<Instant>>,
|
||||
task: JoinHandle<()>,
|
||||
}
|
||||
|
||||
@@ -88,17 +93,29 @@ impl PtyBridge {
|
||||
let replay = ReplayBuffer::new(replay_bytes);
|
||||
let tx = sender.clone();
|
||||
let replay_for_task = replay.clone();
|
||||
let last_activity = Arc::new(Mutex::new(Instant::now()));
|
||||
let last_activity_for_task = last_activity.clone();
|
||||
let task = tokio::spawn(async move {
|
||||
run_reader(async_fd, tx, replay_for_task).await;
|
||||
run_reader(async_fd, tx, replay_for_task, last_activity_for_task).await;
|
||||
});
|
||||
Ok(Self {
|
||||
fifo_path,
|
||||
sender,
|
||||
replay,
|
||||
last_activity,
|
||||
task,
|
||||
})
|
||||
}
|
||||
|
||||
/// How long since the session last produced PTY output. The activity
|
||||
/// watcher treats a short value as `Active` (NAP-340).
|
||||
pub fn idle_for(&self) -> std::time::Duration {
|
||||
self.last_activity
|
||||
.lock()
|
||||
.expect("last_activity poisoned")
|
||||
.elapsed()
|
||||
}
|
||||
|
||||
pub fn fifo_path(&self) -> &Path {
|
||||
&self.fifo_path
|
||||
}
|
||||
@@ -197,7 +214,12 @@ impl ReplayGuard<'_> {
|
||||
}
|
||||
}
|
||||
|
||||
async fn run_reader(async_fd: AsyncFd<File>, tx: broadcast::Sender<Vec<u8>>, replay: ReplayBuffer) {
|
||||
async fn run_reader(
|
||||
async_fd: AsyncFd<File>,
|
||||
tx: broadcast::Sender<Vec<u8>>,
|
||||
replay: ReplayBuffer,
|
||||
last_activity: Arc<Mutex<Instant>>,
|
||||
) {
|
||||
let mut buf = [0u8; READ_BUF_BYTES];
|
||||
loop {
|
||||
let mut guard = match async_fd.readable().await {
|
||||
@@ -219,6 +241,9 @@ async fn run_reader(async_fd: AsyncFd<File>, tx: broadcast::Sender<Vec<u8>>, rep
|
||||
tokio::task::yield_now().await;
|
||||
}
|
||||
Ok(Ok(n)) => {
|
||||
// Mark the session as having just produced output, for the
|
||||
// activity watcher's quiescence check (NAP-340).
|
||||
*last_activity.lock().expect("last_activity poisoned") = Instant::now();
|
||||
// Hold the replay lock across both the buffer push and the
|
||||
// broadcast send so `subscribe_with_replay` either sees this
|
||||
// chunk in its snapshot *or* receives it on the broadcast —
|
||||
|
||||
@@ -558,6 +558,9 @@ mod tests {
|
||||
metadata: Default::default(),
|
||||
status: maestro_protocol::SessionStatus::Running,
|
||||
created_at: chrono::Utc::now(),
|
||||
foreground: None,
|
||||
activity: Default::default(),
|
||||
last_activity_at: chrono::Utc::now(),
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -11,7 +11,9 @@ use std::sync::Arc;
|
||||
use std::time::Duration;
|
||||
|
||||
use chrono::Utc;
|
||||
use maestro_protocol::{CreateSessionRequest, ErrorInfo, Session, SessionStatus, WorktreeSpec};
|
||||
use maestro_protocol::{
|
||||
ActivityState, CreateSessionRequest, ErrorInfo, Session, SessionStatus, WorktreeSpec,
|
||||
};
|
||||
use tokio::sync::{broadcast, Mutex};
|
||||
use uuid::Uuid;
|
||||
|
||||
@@ -290,6 +292,11 @@ impl SessionManager {
|
||||
metadata: req.metadata,
|
||||
status: SessionStatus::Running,
|
||||
created_at: Utc::now(),
|
||||
// A freshly-spawned session is producing its initial output; the
|
||||
// activity watcher refines `foreground`/`activity` on its next tick.
|
||||
foreground: None,
|
||||
activity: ActivityState::Active,
|
||||
last_activity_at: Utc::now(),
|
||||
};
|
||||
|
||||
if let Err(e) = self.inner.sidecar.write(&session) {
|
||||
@@ -384,6 +391,9 @@ impl SessionManager {
|
||||
return;
|
||||
};
|
||||
run_liveness_pass(&inner).await;
|
||||
// Same tick, no new timer (NAP-340): refresh activity for the
|
||||
// survivors and emit SessionUpdated on any transition.
|
||||
run_activity_pass(&inner).await;
|
||||
}
|
||||
});
|
||||
}
|
||||
@@ -414,6 +424,10 @@ impl SessionManager {
|
||||
metadata: Default::default(),
|
||||
status: SessionStatus::Running,
|
||||
created_at: Utc::now(),
|
||||
// Activity is unknown on adoption; the watcher fills it in.
|
||||
foreground: None,
|
||||
activity: ActivityState::Idle,
|
||||
last_activity_at: Utc::now(),
|
||||
},
|
||||
};
|
||||
// The sidecar persists whatever daemon_id was current when the
|
||||
@@ -529,6 +543,63 @@ async fn run_liveness_pass(inner: &Arc<Inner>) {
|
||||
}
|
||||
}
|
||||
|
||||
/// Refresh each live session's inferred activity (NAP-340) and emit
|
||||
/// `SessionUpdated` only when `activity` or `foreground` actually changes — so a
|
||||
/// steady state produces no diffs. Runs on the existing liveness tick.
|
||||
async fn run_activity_pass(inner: &Arc<Inner>) {
|
||||
let ids: Vec<Uuid> = {
|
||||
let sessions = inner.sessions.lock().await;
|
||||
if sessions.is_empty() {
|
||||
return;
|
||||
}
|
||||
sessions.keys().copied().collect()
|
||||
};
|
||||
|
||||
for id in ids {
|
||||
// The tmux pane lookup is a subprocess — keep it off the runtime worker
|
||||
// and outside the sessions lock.
|
||||
let pane = {
|
||||
let tmux = inner.tmux.clone();
|
||||
match tmux_blocking(move || tmux.pane_info(id)).await {
|
||||
Ok(Some(p)) => p,
|
||||
// Session/pane vanished — the liveness pass reaps it; skip.
|
||||
Ok(None) => continue,
|
||||
Err(e) => {
|
||||
tracing::debug!(%id, error = ?e, "activity: pane_info failed");
|
||||
continue;
|
||||
}
|
||||
}
|
||||
};
|
||||
// `/proc` inspection is cheap, but still done off the async lock.
|
||||
let awaiting = crate::proc::awaiting_input(pane.pane_pid, &pane.tty);
|
||||
|
||||
let updated = {
|
||||
let mut sessions = inner.sessions.lock().await;
|
||||
let Some(rec) = sessions.get_mut(&id) else {
|
||||
continue;
|
||||
};
|
||||
let idle = rec.bridge.idle_for();
|
||||
let activity = crate::activity::decide(idle, awaiting);
|
||||
let foreground = Some(pane.command.clone());
|
||||
// Change-detection on (activity, foreground) is the debounce: nothing
|
||||
// emitted at steady state.
|
||||
if rec.session.activity == activity && rec.session.foreground == foreground {
|
||||
None
|
||||
} else {
|
||||
rec.session.activity = activity;
|
||||
rec.session.foreground = foreground;
|
||||
// Wall-clock of the last output ≈ now − idle.
|
||||
rec.session.last_activity_at = Utc::now()
|
||||
- chrono::Duration::from_std(idle).unwrap_or_else(|_| chrono::Duration::zero());
|
||||
Some(rec.session.clone())
|
||||
}
|
||||
};
|
||||
if let Some(session) = updated {
|
||||
let _ = inner.events.send(SessionEvent::Updated(session));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Remove a session whose tmux session is already gone: drop the in-memory
|
||||
/// record, delete the sidecar + FIFO, clean up its worktree if requested, and
|
||||
/// fire `SessionEvent::Removed`. No tmux kill — there is nothing left to kill.
|
||||
|
||||
@@ -155,6 +155,39 @@ impl Tmux {
|
||||
Ok(ids)
|
||||
}
|
||||
|
||||
/// `tmux list-panes -t maestro-<id> -F '#{pane_pid} #{pane_current_command}
|
||||
/// #{pane_tty}'` for the session's (single) pane. Used by the activity
|
||||
/// watcher (NAP-340) to learn the foreground command, the pane's process,
|
||||
/// and its tty. Returns `Ok(None)` when the session/pane is gone or tmux
|
||||
/// isn't running, so the watcher degrades to "unknown" rather than erroring.
|
||||
pub fn pane_info(&self, id: Uuid) -> Result<Option<PaneInfo>, TmuxError> {
|
||||
let name = session_name(id);
|
||||
let out = self
|
||||
.cmd()
|
||||
.args([
|
||||
"list-panes",
|
||||
"-t",
|
||||
&name,
|
||||
"-F",
|
||||
"#{pane_pid} #{pane_current_command} #{pane_tty}",
|
||||
])
|
||||
.output()?;
|
||||
if !out.status.success() {
|
||||
let stderr = String::from_utf8_lossy(&out.stderr);
|
||||
// No server, or the session/pane vanished between the liveness check
|
||||
// and now — both are "no info", not a hard error.
|
||||
if is_no_server_error(&stderr) || stderr.contains("can't find") {
|
||||
return Ok(None);
|
||||
}
|
||||
return Err(TmuxError::NonZero {
|
||||
code: out.status.code().unwrap_or(-1),
|
||||
stderr: stderr.into_owned(),
|
||||
});
|
||||
}
|
||||
let stdout = String::from_utf8_lossy(&out.stdout);
|
||||
Ok(stdout.lines().next().and_then(parse_pane_line))
|
||||
}
|
||||
|
||||
/// `tmux send-keys -t maestro-<id> -l <bytes>`. The `-l` flag tells tmux to
|
||||
/// treat the payload as literal input (no key-name translation), which is
|
||||
/// what we want for forwarded PTY input.
|
||||
@@ -229,6 +262,34 @@ fn is_no_server_error(stderr: &str) -> bool {
|
||||
|| stderr.contains("error connecting")
|
||||
}
|
||||
|
||||
/// A session pane's process id, foreground command, and controlling tty, as
|
||||
/// reported by `tmux list-panes` (NAP-340).
|
||||
#[derive(Debug, Clone, PartialEq, Eq)]
|
||||
pub struct PaneInfo {
|
||||
pub pane_pid: i32,
|
||||
pub command: String,
|
||||
pub tty: String,
|
||||
}
|
||||
|
||||
/// Parse one `#{pane_pid} #{pane_current_command} #{pane_tty}` line. `pane_pid`
|
||||
/// is the leading integer and `pane_tty` the trailing token; everything between
|
||||
/// is the command (which is normally a single word, but is rejoined defensively
|
||||
/// in case tmux ever emits a spaced value).
|
||||
fn parse_pane_line(line: &str) -> Option<PaneInfo> {
|
||||
let mut parts = line.split_whitespace();
|
||||
let pane_pid: i32 = parts.next()?.parse().ok()?;
|
||||
let rest: Vec<&str> = parts.collect();
|
||||
let (tty, command) = rest.split_last()?;
|
||||
if command.is_empty() {
|
||||
return None;
|
||||
}
|
||||
Some(PaneInfo {
|
||||
pane_pid,
|
||||
command: command.join(" "),
|
||||
tty: (*tty).to_string(),
|
||||
})
|
||||
}
|
||||
|
||||
pub fn session_name(id: Uuid) -> String {
|
||||
format!("{SESSION_PREFIX}{id}")
|
||||
}
|
||||
@@ -289,6 +350,22 @@ fn check(out: Output) -> Result<(), TmuxError> {
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
#[test]
|
||||
fn parse_pane_line_splits_pid_command_tty() {
|
||||
let p = parse_pane_line("12345 claude /dev/pts/3").unwrap();
|
||||
assert_eq!(p.pane_pid, 12345);
|
||||
assert_eq!(p.command, "claude");
|
||||
assert_eq!(p.tty, "/dev/pts/3");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn parse_pane_line_rejects_malformed() {
|
||||
assert!(parse_pane_line("").is_none());
|
||||
assert!(parse_pane_line("notapid bash /dev/pts/0").is_none());
|
||||
// pid + tty but no command between them.
|
||||
assert!(parse_pane_line("123 /dev/pts/0").is_none());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn new_session_args_pass_256color_env_and_default_terminal() {
|
||||
let args = new_session_args("maestro-abc", Path::new("/tmp/proj"), "sleep 30", None);
|
||||
|
||||
@@ -24,6 +24,9 @@ fn sample(id: Uuid, daemon_id: &str) -> Session {
|
||||
created_at: DateTime::parse_from_rfc3339("2026-05-24T12:00:00Z")
|
||||
.unwrap()
|
||||
.with_timezone(&Utc),
|
||||
foreground: None,
|
||||
activity: Default::default(),
|
||||
last_activity_at: chrono::Utc::now(),
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -522,6 +522,65 @@ async fn spawn_records_history_and_infers_gitroot_for_form() {
|
||||
client.kill_session(s.id).await.expect("kill");
|
||||
}
|
||||
|
||||
#[tokio::test(flavor = "multi_thread")]
|
||||
async fn activity_reflects_foreground_and_input_blocking() {
|
||||
// NAP-340 (#18): the watcher infers each session's foreground command and
|
||||
// activity from structured tmux/proc state. `cat` parks in read() on the
|
||||
// pty → AwaitingInput; `sleep` is quiescent but not reading the tty → Idle
|
||||
// (the distinction the /proc/<pid>/syscall check buys us). Both report the
|
||||
// right foreground. Exercised against real tmux.
|
||||
use maestro_protocol::ActivityState;
|
||||
let (_reg, _daemons, client) = bring_up(&["host"]).await;
|
||||
let cwd = TempDir::new().unwrap();
|
||||
|
||||
let cat = client
|
||||
.spawn_session(&create_req("host", cwd.path(), "cat"))
|
||||
.await
|
||||
.expect("spawn cat");
|
||||
let slp = client
|
||||
.spawn_session(&create_req("host", cwd.path(), "sleep 300"))
|
||||
.await
|
||||
.expect("spawn sleep");
|
||||
|
||||
// Initial state is Active; the watcher needs a couple ticks past the
|
||||
// quiescence window (3s) to settle into the steady states.
|
||||
wait_for_async(
|
||||
"cat → AwaitingInput, sleep → Idle, with correct foreground",
|
||||
Duration::from_secs(20),
|
||||
|| {
|
||||
let client = client.clone();
|
||||
let (cat_id, sleep_id) = (cat.id, slp.id);
|
||||
async move {
|
||||
let sessions = match client.list_sessions().await {
|
||||
Ok(s) => s,
|
||||
Err(_) => return false,
|
||||
};
|
||||
let cat_ok = sessions
|
||||
.iter()
|
||||
.find(|s| s.id == cat_id)
|
||||
.map(|s| {
|
||||
s.activity == ActivityState::AwaitingInput
|
||||
&& s.foreground.as_deref() == Some("cat")
|
||||
})
|
||||
.unwrap_or(false);
|
||||
let sleep_ok = sessions
|
||||
.iter()
|
||||
.find(|s| s.id == sleep_id)
|
||||
.map(|s| {
|
||||
s.activity == ActivityState::Idle
|
||||
&& s.foreground.as_deref() == Some("sleep")
|
||||
})
|
||||
.unwrap_or(false);
|
||||
cat_ok && sleep_ok
|
||||
}
|
||||
},
|
||||
)
|
||||
.await;
|
||||
|
||||
client.kill_session(cat.id).await.expect("kill cat");
|
||||
client.kill_session(slp.id).await.expect("kill sleep");
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Helpers
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
@@ -34,6 +34,28 @@ pub struct Session {
|
||||
pub metadata: HashMap<String, Value>,
|
||||
pub status: SessionStatus,
|
||||
pub created_at: DateTime<Utc>,
|
||||
/// Current foreground command in the session's pane (e.g. `claude`,
|
||||
/// `bash`), as tmux reports it — distinct from the `command` it was spawned
|
||||
/// with. Daemon-derived (NAP-340); `None` until the daemon's activity
|
||||
/// watcher has observed it. All three activity fields default on the wire so
|
||||
/// older daemons/payloads stay forward-compatible.
|
||||
#[serde(default)]
|
||||
pub foreground: Option<String>,
|
||||
/// Inferred activity dimension, orthogonal to [`SessionStatus`]
|
||||
/// (NAP-340). Derived from structured tmux/`/proc` state, never from the
|
||||
/// byte stream.
|
||||
#[serde(default)]
|
||||
pub activity: ActivityState,
|
||||
/// Wall-clock time of the session's most recent observed PTY output.
|
||||
/// Drives "idle for Ns" displays. Defaults to the Unix epoch on payloads
|
||||
/// from daemons that predate the field.
|
||||
#[serde(default = "epoch")]
|
||||
pub last_activity_at: DateTime<Utc>,
|
||||
}
|
||||
|
||||
/// Default for [`Session::last_activity_at`] when an older payload omits it.
|
||||
fn epoch() -> DateTime<Utc> {
|
||||
DateTime::from_timestamp(0, 0).expect("unix epoch is a valid timestamp")
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
|
||||
@@ -42,6 +64,20 @@ pub enum SessionStatus {
|
||||
Dead,
|
||||
}
|
||||
|
||||
/// Inferred session activity, orthogonal to [`SessionStatus`] (NAP-340). The
|
||||
/// daemon derives this from tmux pane state, PTY quiescence, and `/proc`
|
||||
/// inspection — never by parsing the terminal byte stream.
|
||||
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, Default)]
|
||||
pub enum ActivityState {
|
||||
/// Producing output, or the foreground process is busy.
|
||||
Active,
|
||||
/// Foreground process alive but quiescent. The forward-compatible default.
|
||||
#[default]
|
||||
Idle,
|
||||
/// Blocked on a `read()` from the pty — waiting on the user.
|
||||
AwaitingInput,
|
||||
}
|
||||
|
||||
/// Optional worktree request attached to a `CreateSessionRequest`.
|
||||
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
|
||||
pub struct WorktreeSpec {
|
||||
@@ -632,9 +668,41 @@ mod tests {
|
||||
created_at: DateTime::parse_from_rfc3339("2026-05-24T12:00:00Z")
|
||||
.unwrap()
|
||||
.with_timezone(&Utc),
|
||||
foreground: Some("claude".into()),
|
||||
activity: ActivityState::Active,
|
||||
last_activity_at: DateTime::parse_from_rfc3339("2026-05-24T12:05:00Z")
|
||||
.unwrap()
|
||||
.with_timezone(&Utc),
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn session_activity_fields_default_on_legacy_payload() {
|
||||
// A pre-NAP-340 daemon omits the activity fields; they must deserialize
|
||||
// to the forward-compat defaults rather than failing.
|
||||
let json = serde_json::json!({
|
||||
"id": Uuid::from_u128(1),
|
||||
"daemon_id": "laptop",
|
||||
"cwd": "/tmp",
|
||||
"command": "bash",
|
||||
"status": "Running",
|
||||
"created_at": "2026-05-24T12:00:00Z"
|
||||
});
|
||||
let s: Session = serde_json::from_value(json).unwrap();
|
||||
assert!(s.foreground.is_none());
|
||||
assert_eq!(s.activity, ActivityState::Idle);
|
||||
assert_eq!(s.last_activity_at, epoch());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn activity_state_serializes_pascal_case_and_defaults_idle() {
|
||||
assert_eq!(
|
||||
serde_json::to_value(ActivityState::AwaitingInput).unwrap(),
|
||||
serde_json::json!("AwaitingInput")
|
||||
);
|
||||
assert_eq!(ActivityState::default(), ActivityState::Idle);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn session_round_trips() {
|
||||
let s = sample_session();
|
||||
|
||||
@@ -929,6 +929,9 @@ mod tests {
|
||||
metadata: HashMap::new(),
|
||||
status: SessionStatus::Running,
|
||||
created_at: Utc::now(),
|
||||
foreground: None,
|
||||
activity: Default::default(),
|
||||
last_activity_at: chrono::Utc::now(),
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -105,6 +105,9 @@ fn sample_session(id: Uuid, daemon_id: &str) -> Session {
|
||||
created_at: DateTime::parse_from_rfc3339("2026-05-24T12:00:00Z")
|
||||
.unwrap()
|
||||
.with_timezone(&Utc),
|
||||
foreground: None,
|
||||
activity: Default::default(),
|
||||
last_activity_at: chrono::Utc::now(),
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -219,6 +219,9 @@ fn mock_session(daemon_id: &str, request: &CreateSessionRequest) -> Session {
|
||||
metadata: request.metadata.clone(),
|
||||
status: SessionStatus::Running,
|
||||
created_at: Utc::now(),
|
||||
foreground: None,
|
||||
activity: Default::default(),
|
||||
last_activity_at: chrono::Utc::now(),
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -549,6 +549,9 @@ mod tests {
|
||||
metadata: Default::default(),
|
||||
status: SessionStatus::Running,
|
||||
created_at: Utc::now(),
|
||||
foreground: None,
|
||||
activity: Default::default(),
|
||||
last_activity_at: chrono::Utc::now(),
|
||||
}],
|
||||
});
|
||||
app.dispatch(Action::KillHighlighted).await;
|
||||
|
||||
@@ -219,6 +219,9 @@ mod tests {
|
||||
metadata: Default::default(),
|
||||
status: SessionStatus::Running,
|
||||
created_at: Utc::now(),
|
||||
foreground: None,
|
||||
activity: Default::default(),
|
||||
last_activity_at: chrono::Utc::now(),
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -109,6 +109,9 @@ mod tests {
|
||||
metadata,
|
||||
status: SessionStatus::Running,
|
||||
created_at: Utc::now(),
|
||||
foreground: None,
|
||||
activity: Default::default(),
|
||||
last_activity_at: chrono::Utc::now(),
|
||||
}
|
||||
}
|
||||
|
||||
@@ -203,6 +206,9 @@ mod tests {
|
||||
metadata: HashMap::new(),
|
||||
status: SessionStatus::Running,
|
||||
created_at: Utc::now(),
|
||||
foreground: None,
|
||||
activity: Default::default(),
|
||||
last_activity_at: chrono::Utc::now(),
|
||||
};
|
||||
let sb = Sidebar {
|
||||
daemons: vec![daemon("konishi")],
|
||||
|
||||
@@ -33,6 +33,9 @@ fn session(id: u128, daemon_id: &str) -> Session {
|
||||
metadata: Default::default(),
|
||||
status: SessionStatus::Running,
|
||||
created_at: Utc::now(),
|
||||
foreground: None,
|
||||
activity: Default::default(),
|
||||
last_activity_at: chrono::Utc::now(),
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -6,6 +6,10 @@ export type IsoTimestamp = string;
|
||||
|
||||
export type SessionStatus = "Running" | "Dead";
|
||||
|
||||
/** Daemon-inferred activity dimension, orthogonal to `SessionStatus` (NAP-340).
|
||||
* Absent on payloads from daemons that predate the field. */
|
||||
export type ActivityState = "Active" | "Idle" | "AwaitingInput";
|
||||
|
||||
export interface Session {
|
||||
id: Uuid;
|
||||
daemon_id: string;
|
||||
@@ -15,6 +19,13 @@ export interface Session {
|
||||
metadata: Record<string, unknown>;
|
||||
status: SessionStatus;
|
||||
created_at: IsoTimestamp;
|
||||
/** Current foreground command in the pane (e.g. `claude`, `bash`), as the
|
||||
* daemon's activity watcher observed it. Absent until observed (NAP-340). */
|
||||
foreground?: string | null;
|
||||
/** Inferred activity. Defaults to `Idle` on the wire when omitted. */
|
||||
activity?: ActivityState;
|
||||
/** Wall-clock of the most recent observed PTY output (NAP-340). */
|
||||
last_activity_at?: IsoTimestamp;
|
||||
}
|
||||
|
||||
export interface DaemonInfo {
|
||||
|
||||
Reference in New Issue
Block a user