feat: persist PTY output activity evidence

README.md

@@ -23,6 +23,11 @@ as Fractal. The Node implementation is the behavioral reference while the port
 converges. This README does not claim that the current experiment has reached
 full parity.
 
+The durable constraints and implementation contract live in
+[`docs/vrs/requirements.md`](docs/vrs/requirements.md) and
+[`docs/vrs/spec.md`](docs/vrs/spec.md). Issues remain the work/discussion
+surface; they do not replace the VRS.
+
 Compatibility means that the same user-visible operations and wire messages have
 the same result. It does not require identical source code or internal design.
 Rust, `portable-pty`, and `libghostty` can require a different implementation.
@@ -129,7 +134,7 @@ drains into the terminal on demand.
 
 ## Build requirements
 
-- **Rust** (edition 2021; built with 1.97).
+- **Rust** (edition 2024; built with 1.97).
 - **Zig 0.15.2** on `PATH`. The `libghostty-vt-sys` build script fetches the
   Ghostty source and compiles the VT core with `zig build`, so a matching Zig
   toolchain must be installed. Install it with:

docs/vrs/requirements.md

@@ -0,0 +1,27 @@
+# pty-rust requirements
+
+## Context
+
+pty-rust is a Rust implementation of the compoundingtech/pty session protocol and registry contract. It owns a per-session daemon, PTY child, libghostty terminal state, and Unix-socket clients. These requirements define the durable constraints needed for Node/Rust compatibility and launcher-agnostic session activity evidence.
+
+## Assumptions
+
+- **A01 Unix runtime:** Supported hosts provide Unix PTYs, sockets, signals, process liveness probes, and atomic same-filesystem rename.
+- **A02 Trusted user registry:** One registry belongs to one trusted OS user; filesystem permissions are the access boundary.
+- **A03 Actor ownership:** The daemon actor is the single owner of terminal state. Reader/client threads communicate with it through typed messages.
+- **A04 Universal output evidence:** Every successful PTY reader chunk reaches `DaemonMsg::PtyData` before terminal parsing and client broadcast. Recording when that happens adds no observer and carries no launcher or harness semantics.
+
+## Acceptable tradeoffs
+
+- **T01 Per-session daemon:** Each session pays for one independent actor/daemon in exchange for client-independent lifetime and failure isolation.
+- **T02 Coalesced metadata:** Live output evidence may lag the newest chunk by at most one second to bound metadata writes; retained exit metadata must carry the final in-memory value.
+- **T03 Pre-1.0 storage evolution:** Optional additive metadata fields may appear, while older records without them remain readable.
+
+## Requirements
+
+- **R01 Node-compatible session registry:** Stable ids own metadata, socket, pid, and retained-screen artifacts under `PTY_ROOT`; field names and omission behavior remain compatible with compoundingtech/pty where the surface is implemented.
+- **R02 Ordered terminal ownership:** The daemon actor applies PTY output to libghostty before exposing derived screen state, and preserves output ordering for streaming clients.
+- **R03 Atomic durable metadata:** Metadata writes serialize one complete JSON object to a sibling temporary file and atomically rename it over the session record. Optional unknown/additive fields survive supported read-modify-write paths.
+- **R04 Durable output-activity evidence:** Session metadata exposes optional unix-millisecond `lastOutputAtMs`. The actor stamps every nonempty `PtyData` message, persists the newest stamp on a trailing-edge one-second deadline, and carries the final in-memory stamp into retained exit metadata before teardown. A new silent session omits the field. The field is evidence only: pty-rust does not classify active/idle, infer liveness, or authorize lifecycle/delivery behavior.
+- **R05 Bounded write amplification:** A continuously chatty session performs at most one live activity metadata persist per one-second window. No filesystem work occurs per output chunk beyond updating actor-owned memory.
+- **R06 Behavioral proof:** Real-process tests cover absence before output, a recent stamp after output, monotonic advancement after a later burst, and immediate output followed by exit retaining the final stamp. Build/test tooling must compile the daemon actor and registry on the pinned Rust/libghostty toolchain.

docs/vrs/spec.md

@@ -0,0 +1,71 @@
+# pty-rust specification
+
+## Status
+
+Implemented for the session-activity slice. Broader Node parity remains tracked by repository issues.
+
+## Scope
+
+This specification defines the per-session actor, compatible registry metadata, and durable output-activity evidence. It does not define harness semantics, active/idle thresholds, agent orchestration, delivery policy, or authorization.
+
+## Architecture
+
+```text
+portable-pty reader thread
+          |
+          v
+DaemonMsg::PtyData(Vec<u8>)
+          |
+          v
+single daemon actor ──> libghostty terminal ──> streaming clients
+          |
+          +── actor-owned last_output_at_ms
+                  |
+                  +── trailing-edge deadline (1s)
+                  v
+          <PTY_ROOT>/<name>.json
+```
+
+The actor owns libghostty's non-`Send` terminal, the PTY writer, client registry, and activity timestamp (A03, R02). Reader/client threads only send `DaemonMsg` values.
+
+## Registry wire
+
+`registry::SessionMetadata` is serialized with camelCase field names. Output evidence is additive:
+
+```json
+{
+  "lastOutputAtMs": 1787761801896
+}
+```
+
+The field is absent until nonempty output is observed and absent in older records. Its value is unix milliseconds. `registry::write_metadata` writes pretty JSON to `<name>.json.tmp` and atomically renames it over `<name>.json` (R01, R03, R04).
+
+## Activity persistence
+
+The actor loop maintains:
+
+```rust
+last_output_at_ms: Option<u64>
+activity_persist_deadline: Option<Instant>
+```
+
+For each nonempty `DaemonMsg::PtyData(bytes)`:
+
+1. set `last_output_at_ms = now_epoch_ms()`;
+2. if no activity deadline exists, schedule `Instant::now() + 1s`;
+3. process the same bytes through libghostty and client broadcast.
+
+The actor receives with `recv_timeout` while a deadline exists. On timeout it reads current metadata, changes only `last_output_at_ms`, writes atomically, clears the deadline, and returns to the loop. Further chunks inside the window update memory but do not schedule or write again (R04, R05).
+
+On `DaemonMsg::PtyExited(code)`, retained-session finalization captures the screen/tail and writes exit fields plus the latest in-memory `last_output_at_ms` in the same metadata replacement. Reaped sessions remove metadata and therefore retain no evidence. The pending timer cannot lose the final retained stamp because the actor exits after that synchronous write (R04).
+
+pty-rust reports evidence only. Consumers own activity windows and composition with richer observations.
+
+## Validation
+
+| Requirement | Owning source | Executable evidence |
+| --- | --- | --- |
+| R01, R03 | `src/registry.rs` | `tests/registry_liveness.rs`, `tests/cli_e2e.rs` |
+| R02 | `src/daemon.rs` | existing terminal/stream parity tests |
+| R04, R05 | `src/daemon.rs`, `src/registry.rs` | `output_activity_stamp_appears_and_advances`, `post_exit_peek_returns_final_screen` |
+| R06 | crate/test harness | `cargo build`, `cargo test --test cli_e2e`, `cargo test --test registry_liveness` |

src/daemon.rs

@@ -14,8 +14,9 @@ use std::os::unix::net::{UnixListener, UnixStream};
 use std::path::PathBuf;
 use std::rc::Rc;
 use std::sync::atomic::{AtomicI32, AtomicU64, Ordering};
-use std::sync::mpsc::{self, Receiver, Sender};
+use std::sync::mpsc::{self, Receiver, RecvTimeoutError, Sender};
 use std::sync::Arc;
+use std::time::{Duration, Instant};
 
 use libghostty_vt::terminal::{Mode, Options, Terminal};
 use portable_pty::{native_pty_system, CommandBuilder, PtySize};
@@ -204,6 +205,7 @@ pub fn run(cfg: DaemonConfig) -> std::io::Result<i32> {
         },
         display_name: cfg.display_name.clone(),
         last_attach_at: None,
+        last_output_at_ms: None,
     };
     registry::write_metadata(&cfg.name, &meta)?;
     // Record the DAEMON pid (matching node: `<name>.pid` holds the server
@@ -276,6 +278,20 @@ pub fn run(cfg: DaemonConfig) -> std::io::Result<i32> {
     let mut cur_rows = cfg.rows;
     let mut cur_cols = cfg.cols;
 
+    let mut last_output_at_ms: Option<u64> = None;
+    let mut activity_persist_deadline: Option<Instant> = None;
+
+    let persist_activity = |timestamp: Option<u64>| {
+        let Some(timestamp) = timestamp else {
+            return;
+        };
+        if let Some(mut metadata) = registry::read_metadata(&cfg.name)
+            && metadata.last_output_at_ms != Some(timestamp)
+        {
+            metadata.last_output_at_ms = Some(timestamp);
+            let _ = registry::write_metadata(&cfg.name, &metadata);
+        }
+    };
     let flush_pending = |writer: &mut Box<dyn Write + Send>, pending: &Rc<RefCell<Vec<u8>>>| {
         let out = std::mem::take(&mut *pending.borrow_mut());
         if !out.is_empty() {
@@ -286,15 +302,44 @@ pub fn run(cfg: DaemonConfig) -> std::io::Result<i32> {
 
     let exit_code_final;
     loop {
-        let msg = match rx.recv() {
-            Ok(m) => m,
-            Err(_) => {
-                exit_code_final = -1;
-                break;
+        // A permanently nonempty PTY queue must not starve the deadline:
+        // `recv_timeout(Duration::ZERO)` may keep returning queued messages.
+        // Settle the persist before dequeuing once the absolute deadline passed.
+        if activity_persist_deadline.is_some_and(|deadline| Instant::now() >= deadline) {
+            persist_activity(last_output_at_ms);
+            activity_persist_deadline = None;
+            continue;
+        }
+        let msg = if let Some(deadline) = activity_persist_deadline {
+            match rx.recv_timeout(deadline.saturating_duration_since(Instant::now())) {
+                Ok(message) => message,
+                Err(RecvTimeoutError::Timeout) => {
+                    persist_activity(last_output_at_ms);
+                    activity_persist_deadline = None;
+                    continue;
+                }
+                Err(RecvTimeoutError::Disconnected) => {
+                    exit_code_final = -1;
+                    break;
+                }
+            }
+        } else {
+            match rx.recv() {
+                Ok(message) => message,
+                Err(_) => {
+                    exit_code_final = -1;
+                    break;
+                }
             }
         };
         match msg {
             DaemonMsg::PtyData(bytes) => {
+                if !bytes.is_empty() {
+                    last_output_at_ms = Some(now_epoch_ms());
+                    if activity_persist_deadline.is_none() {
+                        activity_persist_deadline = Some(Instant::now() + Duration::from_secs(1));
+                    }
+                }
                 terminal.vt_write(&bytes);
                 flush_pending(&mut writer, &pending);
                 // Broadcast to streaming clients; drop dead ones.
@@ -458,6 +503,7 @@ pub fn run(cfg: DaemonConfig) -> std::io::Result<i32> {
                         m.exit_code = Some(code);
                         m.exited_at = Some(now_iso8601());
                         m.last_lines = Some(tail);
+                        m.last_output_at_ms = last_output_at_ms;
                         let _ = registry::write_metadata(&cfg.name, &m);
                     }
                 }
@@ -544,6 +590,15 @@ fn exit_code(status: portable_pty::ExitStatus) -> Option<i32> {
     Some(if status.success() { 0 } else { status.exit_code() as i32 })
 }
 
+/// Current epoch time in unix milliseconds, saturating at `u64::MAX`.
+pub fn now_epoch_ms() -> u64 {
+    use std::time::{SystemTime, UNIX_EPOCH};
+    SystemTime::now()
+        .duration_since(UNIX_EPOCH)
+        .map(|duration| duration.as_millis().min(u128::from(u64::MAX)) as u64)
+        .unwrap_or_default()
+}
+
 /// Current epoch time in seconds (fractional), for uptime.
 pub fn now_epoch_f64() -> f64 {
     use std::time::{SystemTime, UNIX_EPOCH};

src/registry.rs

@@ -33,6 +33,10 @@ pub struct SessionMetadata {
     pub display_name: Option<String>,
     #[serde(skip_serializing_if = "Option::is_none", default)]
     pub last_attach_at: Option<String>,
+    /// Unix-millisecond timestamp of the last PTY output chunk processed by
+    /// the daemon. Evidence only: consumers classify activity; PTY does not.
+    #[serde(skip_serializing_if = "Option::is_none", default)]
+    pub last_output_at_ms: Option<u64>,
 }
 
 /// Resolve the session registry directory: `$PTY_ROOT`, else the deprecated

tests/cli_e2e.rs

@@ -75,6 +75,67 @@ fn ok_pty(root: &PathBuf, args: &[&str]) -> String {
     out
 }
 
+fn metadata_json(root: &PathBuf, name: &str) -> serde_json::Value {
+    let raw = std::fs::read_to_string(root.join(format!("{name}.json")))
+        .expect("read session metadata");
+    serde_json::from_str(&raw).expect("parse session metadata")
+}
+
+#[test]
+fn output_activity_stamp_appears_and_advances() {
+    let _serial = serial();
+    let root = unique_root();
+    let (_name, err, code) = run_pty(&root, &["run", "--id", "oa", "--", "cat"]);
+    assert_eq!(code, 0, "run failed: {err}");
+
+    let start = Instant::now();
+    while start.elapsed() < Duration::from_secs(5) && !root.join("oa.json").exists() {
+        std::thread::sleep(Duration::from_millis(50));
+    }
+    assert!(
+        metadata_json(&root, "oa").get("lastOutputAtMs").is_none(),
+        "silent session must not fabricate activity"
+    );
+
+    let before = now_ms();
+    ok_pty(&root, &["send", "oa", "--seq", "first", "--seq", "key:return"]);
+    let first = wait_last_output_ms(&root, "oa", None);
+    assert!(first >= before.saturating_sub(1_000));
+
+    // The actor's trailing-edge persist window lives in the daemon process;
+    // wait it out before requiring a later output burst to advance the stamp.
+    std::thread::sleep(Duration::from_millis(1_200));
+    ok_pty(&root, &["send", "oa", "--seq", "second", "--seq", "key:return"]);
+    let second = wait_last_output_ms(&root, "oa", Some(first));
+    assert!(second > first);
+
+    let _ = run_pty(&root, &["kill", "oa"]);
+    let _ = run_pty(&root, &["rm", "oa"]);
+    let _ = std::fs::remove_dir_all(&root);
+}
+
+fn wait_last_output_ms(root: &PathBuf, name: &str, after: Option<u64>) -> u64 {
+    let start = Instant::now();
+    while start.elapsed() < Duration::from_secs(5) {
+        if let Some(value) = metadata_json(root, name)
+            .get("lastOutputAtMs")
+            .and_then(serde_json::Value::as_u64)
+            && after.is_none_or(|previous| value > previous)
+        {
+            return value;
+        }
+        std::thread::sleep(Duration::from_millis(50));
+    }
+    panic!("lastOutputAtMs did not satisfy expected bound");
+}
+
+fn now_ms() -> u64 {
+    std::time::SystemTime::now()
+        .duration_since(std::time::UNIX_EPOCH)
+        .map(|duration| duration.as_millis().min(u128::from(u64::MAX)) as u64)
+        .unwrap_or_default()
+}
+
 #[test]
 fn run_ls_peek_send_kill_lifecycle() {
     let _serial = serial();
@@ -520,6 +581,13 @@ fn post_exit_peek_returns_final_screen() {
         std::thread::sleep(Duration::from_millis(100));
     }
     assert!(exited, "session never recorded exit:7");
+    let metadata = metadata_json(&root, "px");
+    assert_eq!(metadata["exitCode"], 7);
+    assert!(
+        metadata["lastOutputAtMs"].as_u64().is_some(),
+        "exit metadata must carry the final output stamp: {metadata}"
+    );
+
 
     // peek --plain must succeed AND contain the final output (not ENOENT).
     let (screen, err, code) = run_pty(&root, &["peek", "--plain", "px"]);

tests/registry_liveness.rs

@@ -34,6 +34,7 @@ fn meta(exit_code: Option<i32>) -> SessionMetadata {
         tags: None,
         display_name: None,
         last_attach_at: None,
+        last_output_at_ms: None,
     }
 }