fix(runtime): expose terminal output observations without inferring working state

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::<chrono::Utc>::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),
     ])
 }

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<String, Value>) -> 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<String, Value> {
+        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<String, Value>) {
+        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:?}");
+    }
+}

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<Option<OutputObservation>, 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,

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::{

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<Option<OutputObservation>, 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<bool, RuntimeError> {
         self.state
             .lock()