diff --git a/crates/cccc-daemon/src/ops/working_state.rs b/crates/cccc-daemon/src/ops/working_state.rs index f04590470..dd40e8795 100644 --- a/crates/cccc-daemon/src/ops/working_state.rs +++ b/crates/cccc-daemon/src/ops/working_state.rs @@ -2,6 +2,20 @@ use cccc_contracts::{Actor, ActorRuntime, RunnerKind}; use cccc_core::HomeLayout; use serde_json::{Map, Value, json}; +/// When a live terminal session last wrote output, for display only. Output +/// does not establish whether the process is working: a model can compute +/// silently, and a spinner or a waiting prompt can keep printing. The working +/// state therefore never reads these values. +fn pty_output_observation(group_id: &str, actor_id: &str) -> (Value, Value) { + let Ok(Some(observed)) = cccc_runtime::last_output(group_id, actor_id) else { + return (Value::Null, Value::Null); + }; + let last_output_at = chrono::DateTime::::from(observed.last_output_at) + .to_rfc3339_opts(chrono::SecondsFormat::Micros, true); + let quiet_seconds = observed.quiet_for.as_secs(); + (Value::String(last_output_at), json!(quiet_seconds)) +} + pub fn runtime_actor_fields( home: &HomeLayout, actor: &Actor, @@ -98,6 +112,11 @@ pub(super) fn fields( None, ) }; + let (pty_last_output_at, pty_quiet_seconds) = if running && runner_effective == "pty" { + pty_output_observation(group_id, &actor.id) + } else { + (Value::Null, Value::Null) + }; Map::from_iter([ ("idle_seconds".into(), Value::Null), @@ -106,5 +125,7 @@ pub(super) fn fields( ("effective_working_reason".into(), json!(reason)), ("effective_working_updated_at".into(), json!(updated_at)), ("effective_active_task_id".into(), json!(active_task_id)), + ("pty_last_output_at".into(), pty_last_output_at), + ("pty_quiet_seconds".into(), pty_quiet_seconds), ]) } diff --git a/crates/cccc-daemon/src/ops/working_state_tests.rs b/crates/cccc-daemon/src/ops/working_state_tests.rs index d721262a1..a57d6907b 100644 --- a/crates/cccc-daemon/src/ops/working_state_tests.rs +++ b/crates/cccc-daemon/src/ops/working_state_tests.rs @@ -101,3 +101,211 @@ fn pending_managed_session_and_structured_runtime_have_distinct_states() { assert_eq!(state["effective_working_state"], "idle"); assert_eq!(state["effective_working_reason"], "headless_running"); } + +/// Terminal output is an observation for display. It never decides the working +/// state, because output does not establish whether a process is working. +#[cfg(unix)] +mod terminal_output { + use super::*; + use serde_json::Value; + use std::path::Path; + use std::time::{Duration, Instant}; + + const ACTOR_ID: &str = "peer1"; + /// Prints once, then computes without writing anything. + const QUIET_WORK: &str = "printf 'computing\\n'; while :; do :; done"; + /// A prompt that keeps redrawing while it waits for an answer. + const NOISY_PROMPT: &str = "while :; do printf 'Continue? [y/N] \\r'; sleep 0.2; done"; + const PRINT_THEN_SLEEP: &str = "printf 'done\\n'; exec sleep 30"; + + struct StopOnDrop(&'static str); + + impl Drop for StopOnDrop { + fn drop(&mut self) { + let _ = cccc_runtime::stop(self.0, ACTOR_ID); + } + } + + fn start_terminal(group_id: &'static str, cwd: &Path, script: &str) -> StopOnDrop { + cccc_runtime::start(cccc_runtime::LaunchSpec { + group_id: group_id.into(), + actor_id: ACTOR_ID.into(), + runner: RunnerKind::Pty, + command: vec!["sh".into(), "-c".into(), script.into()], + cwd: cwd.into(), + env: Default::default(), + cols: 80, + rows: 24, + }) + .expect("start terminal process"); + StopOnDrop(group_id) + } + + fn custom_actor() -> Actor { + let mut actor = Actor::new(ACTOR_ID); + actor.runtime = ActorRuntime::Custom; + actor + } + + fn still_running(group_id: &str) -> bool { + cccc_runtime::status(group_id, ACTOR_ID).is_ok_and(|status| status.running) + } + + fn quiet_seconds(state: &Map) -> u64 { + state["pty_quiet_seconds"].as_u64().unwrap_or(0) + } + + /// Polls until the daemon has seen terminal output, or until the working + /// state leaves "unknown" so that the caller's assertion reports it. + fn observe_output(home: &HomeLayout, actor: &Actor, group_id: &str) -> Map { + let deadline = Instant::now() + Duration::from_secs(10); + loop { + let state = fields(home, actor, group_id, true, "pty"); + let last_output = state.get("pty_last_output_at"); + if last_output.is_some_and(Value::is_string) + || state["effective_working_reason"] != "pty_running_state_unknown" + || Instant::now() >= deadline + { + return state; + } + std::thread::sleep(Duration::from_millis(20)); + } + } + + fn assert_unknown_state(state: &Map) { + assert_eq!(state["effective_working_state"], "waiting", "{state:?}"); + assert_eq!( + state["effective_working_reason"], "pty_running_state_unknown", + "{state:?}" + ); + assert!(state["effective_working_updated_at"].is_null()); + assert!(state["idle_seconds"].is_null()); + } + + #[test] + fn quiet_ongoing_work_keeps_the_working_state_unknown() { + let temp = tempfile::tempdir().expect("tempdir"); + let home = HomeLayout::from_path(temp.path()).expect("home"); + let group_id = "g_pty_quiet_work"; + let _stop = start_terminal(group_id, temp.path(), QUIET_WORK); + let actor = custom_actor(); + + let observed = observe_output(&home, &actor, group_id); + assert_unknown_state(&observed); + assert!(observed["pty_last_output_at"].is_string(), "{observed:?}"); + + // The process keeps computing but writes nothing more. + let deadline = Instant::now() + Duration::from_secs(10); + let quiet = loop { + let state = fields(&home, &actor, group_id, true, "pty"); + assert_unknown_state(&state); + if quiet_seconds(&state) >= 1 || Instant::now() >= deadline { + break state; + } + std::thread::sleep(Duration::from_millis(50)); + }; + assert!(quiet_seconds(&quiet) >= 1, "{quiet:?}"); + assert!(still_running(group_id)); + } + + #[test] + fn noisy_input_waiting_keeps_the_working_state_unknown() { + let temp = tempfile::tempdir().expect("tempdir"); + let home = HomeLayout::from_path(temp.path()).expect("home"); + let group_id = "g_pty_noisy_prompt"; + let _stop = start_terminal(group_id, temp.path(), NOISY_PROMPT); + let actor = custom_actor(); + + let observed = observe_output(&home, &actor, group_id); + assert_unknown_state(&observed); + let first_output = observed["pty_last_output_at"].clone(); + assert!(first_output.is_string(), "{observed:?}"); + + // The prompt keeps printing while it waits; the state stays unknown. + let deadline = Instant::now() + Duration::from_secs(10); + let later = loop { + let state = fields(&home, &actor, group_id, true, "pty"); + assert_unknown_state(&state); + if state["pty_last_output_at"] != first_output || Instant::now() >= deadline { + break state; + } + std::thread::sleep(Duration::from_millis(50)); + }; + assert_ne!(later["pty_last_output_at"], first_output, "{later:?}"); + assert!(still_running(group_id)); + } + + #[test] + fn print_then_sleep_reports_observations_only() { + let temp = tempfile::tempdir().expect("tempdir"); + let home = HomeLayout::from_path(temp.path()).expect("home"); + let group_id = "g_pty_print_then_sleep"; + let _stop = start_terminal(group_id, temp.path(), PRINT_THEN_SLEEP); + let actor = custom_actor(); + + let observed = observe_output(&home, &actor, group_id); + assert_unknown_state(&observed); + let last_output_at = observed["pty_last_output_at"].clone(); + assert!(last_output_at.is_string(), "{observed:?}"); + assert!(observed["pty_quiet_seconds"].is_u64(), "{observed:?}"); + + let deadline = Instant::now() + Duration::from_secs(10); + let quiet = loop { + let state = fields(&home, &actor, group_id, true, "pty"); + if quiet_seconds(&state) >= 1 || Instant::now() >= deadline { + break state; + } + std::thread::sleep(Duration::from_millis(50)); + }; + assert_unknown_state(&quiet); + assert!(quiet_seconds(&quiet) >= 1, "{quiet:?}"); + assert_eq!(quiet["pty_last_output_at"], last_output_at); + assert!(still_running(group_id)); + } + + #[test] + fn a_new_terminal_session_starts_without_observations() { + let temp = tempfile::tempdir().expect("tempdir"); + let home = HomeLayout::from_path(temp.path()).expect("home"); + let group_id = "g_pty_session_reset"; + let actor = custom_actor(); + let first = start_terminal(group_id, temp.path(), PRINT_THEN_SLEEP); + let observed = observe_output(&home, &actor, group_id); + assert_unknown_state(&observed); + assert!(observed["pty_last_output_at"].is_string(), "{observed:?}"); + drop(first); + + let _stop = start_terminal(group_id, temp.path(), "exec sleep 30"); + let fresh = fields(&home, &actor, group_id, true, "pty"); + assert_unknown_state(&fresh); + assert!(fresh["pty_last_output_at"].is_null(), "{fresh:?}"); + assert!(fresh["pty_quiet_seconds"].is_null(), "{fresh:?}"); + } + + #[test] + fn a_managed_session_keeps_precedence_over_terminal_output() { + let temp = tempfile::tempdir().expect("tempdir"); + let home = HomeLayout::from_path(temp.path()).expect("home"); + let group_id = "g_pty_managed_session"; + let mut actor = Actor::new(ACTOR_ID); + actor.runtime = ActorRuntime::Claude; + actor.runtime_state_source = RuntimeStateSource::ManagedSession; + let _stop = start_terminal(group_id, temp.path(), PRINT_THEN_SLEEP); + + let deadline = Instant::now() + Duration::from_secs(10); + let state = loop { + let state = fields(&home, &actor, group_id, true, "pty"); + let last_output = state.get("pty_last_output_at"); + if last_output.is_some_and(Value::is_string) || Instant::now() >= deadline { + break state; + } + std::thread::sleep(Duration::from_millis(20)); + }; + assert_eq!(state["effective_working_state"], "waiting", "{state:?}"); + assert_eq!( + state["effective_working_reason"], + "managed_agent_session_pending" + ); + assert!(state["pty_last_output_at"].is_string(), "{state:?}"); + } +} diff --git a/crates/cccc-runtime/src/history_access.rs b/crates/cccc-runtime/src/history_access.rs index af1d904ac..ea06d9c66 100644 --- a/crates/cccc-runtime/src/history_access.rs +++ b/crates/cccc-runtime/src/history_access.rs @@ -1,6 +1,7 @@ use crate::RuntimeError; use crate::output::HistoryPage; use crate::registry::{completed_history, lookup, with_session}; +use crate::session_history::OutputObservation; pub fn history( group_id: &str, @@ -65,6 +66,19 @@ pub fn active_history_since( .active_page_since(after, limit) } +/// When the live session's process last wrote to its terminal, or `None` before +/// its first output. A new session starts without an observation. +pub fn last_output( + group_id: &str, + actor_id: &str, +) -> Result, RuntimeError> { + lookup(group_id, actor_id)? + .lock() + .map_err(|_| RuntimeError::Poisoned)? + .history_handle() + .last_output() +} + pub fn active_history_replay( group_id: &str, actor_id: &str, diff --git a/crates/cccc-runtime/src/lib.rs b/crates/cccc-runtime/src/lib.rs index 78475248b..a7b43ced7 100644 --- a/crates/cccc-runtime/src/lib.rs +++ b/crates/cccc-runtime/src/lib.rs @@ -53,7 +53,7 @@ pub use command_session::CommandSession; pub use executable::{prepare_pty_command, resolve_command_executable, resolve_executable_in_path}; pub use history_access::{ active_history_replay, active_history_since, bracketed_paste_enabled, clear, history, - history_since, retained_history, retained_history_tail, + history_since, last_output, retained_history, retained_history_tail, }; pub use manager::{ reap, resize, start, start_with_history, status, stop, stop_all, stop_if_started_at, submit, @@ -64,6 +64,7 @@ pub use output::HistoryPage; pub use process_tree::protect_owned_process_groups; pub use process_tree::{OwnedProcessTree, force_terminate_owned, owned_process_recovery_pending}; pub use session::{LaunchSpec, SessionStatus}; +pub use session_history::OutputObservation; pub use terminal_attach::{TerminalAttachMode, TerminalAttachment, TerminalInput, TerminalOutput}; pub use terminal_initial_output::{TerminalInitialOutput, TerminalInitialOutputKind}; pub use terminal_manager::{ diff --git a/crates/cccc-runtime/src/session_history.rs b/crates/cccc-runtime/src/session_history.rs index 776f3ff59..7de71ea16 100644 --- a/crates/cccc-runtime/src/session_history.rs +++ b/crates/cccc-runtime/src/session_history.rs @@ -3,6 +3,7 @@ use crate::output::{HistoryPage, OutputBuffer}; use crate::terminal_snapshot::TerminalStateMirror; use crate::transcript_archive::{HistoryConfig, TranscriptArchive}; use std::sync::{Arc, Mutex}; +use std::time::{Duration, Instant, SystemTime}; use tokio::sync::watch; mod subscription; @@ -14,6 +15,18 @@ struct SessionHistoryState { archive_writable: bool, accepting_output: bool, terminal: TerminalStateMirror, + last_output: Option<(Instant, SystemTime)>, +} + +/// When a live terminal session's process last wrote output. This is an +/// observation only: output neither proves nor disproves that the process is +/// working. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct OutputObservation { + /// Wall-clock time of the last output, for display. + pub last_output_at: SystemTime, + /// Time since the last output, measured on the monotonic clock. + pub quiet_for: Duration, } #[derive(Clone)] @@ -77,6 +90,7 @@ impl SessionHistory { archive_writable, accepting_output: true, terminal: TerminalStateMirror::new(cols, rows), + last_output: None, })), changes, }) @@ -109,6 +123,9 @@ impl SessionHistory { archive_error: None, }); } + if !data.is_empty() { + state.last_output = Some((Instant::now(), SystemTime::now())); + } let responses = state.terminal.process(data); state.output.push(data); let end_cursor = state.output.end_cursor(); @@ -257,6 +274,20 @@ impl SessionHistory { Ok(()) } + /// When the process last wrote to its terminal, or `None` before its first + /// output in this session. + pub(crate) fn last_output(&self) -> Result, RuntimeError> { + self.state + .lock() + .map_err(|_| RuntimeError::Poisoned) + .map(|state| { + state.last_output.map(|(at, wall)| OutputObservation { + last_output_at: wall, + quiet_for: at.elapsed(), + }) + }) + } + pub(crate) fn bracketed_paste_enabled(&self) -> Result { self.state .lock()