diff --git a/crates/tty7-core/src/client/pane.rs b/crates/tty7-core/src/client/pane.rs index daa46479..01208be6 100644 --- a/crates/tty7-core/src/client/pane.rs +++ b/crates/tty7-core/src/client/pane.rs @@ -11,6 +11,10 @@ use crate::daemon::transport; const OPEN_REPLY_WAIT: Duration = Duration::from_secs(15); +/// How long a hook waits on the daemon before it falls back to the tty. The +/// agent waits on the hook, so this is time the agent stands still. +const AGENT_EVENT_REPLY_WAIT: Duration = Duration::from_secs(2); + #[derive(Clone, Debug, Default)] enum PaneEndpoint { #[default] @@ -128,6 +132,27 @@ impl PaneClient { } } + /// Hand the daemon an agent hook's report about `pane_id` (see + /// `ClientMsg::AgentEvent`). An error — the pane is not this daemon's, + /// `pid` does not run in it, the daemon predates the message — means it + /// was not applied. + pub fn report_agent_event(&self, pane_id: u64, pid: u32, event: &str) -> io::Result<()> { + let mut stream = self.open()?; + let _ = stream.set_read_timeout(Some(AGENT_EVENT_REPLY_WAIT)); + let _ = stream.set_write_timeout(Some(AGENT_EVENT_REPLY_WAIT)); + ClientMsg::AgentEvent { + pane_id, + pid, + event: event.to_string(), + } + .encode(&mut stream)?; + match DaemonMsg::read(&mut stream)? { + DaemonMsg::InputAck { .. } => Ok(()), + DaemonMsg::Error(message) => Err(io::Error::other(message)), + other => Err(unexpected_reply("AgentEvent", &other)), + } + } + pub fn procs(&self, pane_id: u64) -> io::Result { let mut stream = self.open()?; ClientMsg::QueryProcs { pane_id }.encode(&mut stream)?; diff --git a/crates/tty7-core/src/core/agent_hooks.rs b/crates/tty7-core/src/core/agent_hooks.rs index 5863e784..23c3278f 100644 --- a/crates/tty7-core/src/core/agent_hooks.rs +++ b/crates/tty7-core/src/core/agent_hooks.rs @@ -24,7 +24,34 @@ pub fn run_agent_hook(agent: &str, event: &str) { let Some(event) = effective_event(agent, event, &input) else { return; }; - write_to_controlling_tty(&build_hook_sequence(agent, event, &input)); + let body = build_hook_body(agent, event, &input); + if report_to_daemon(&body) { + return; + } + write_to_controlling_tty(&hook_sequence(&body)); +} + +/// Hand the report to the daemon that owns this pane, over its socket. +/// +/// The tty is the channel of last resort. The agent draws on it too, and a +/// tty write is not atomic — on macOS the kernel parks a writer mid-write +/// whenever the output queue is full, which is exactly when an agent is busy +/// redrawing — so the agent's output can land inside the sequence, break it, +/// and leave the rest of it on screen as text. The socket has no such +/// neighbour. It only works where the daemon runs, though: a hook the +/// environment points at no daemon, at a daemon that predates this message, +/// or at a pane it does not run in (see `DaemonPane::report_agent_event`) +/// falls back to the tty. +fn report_to_daemon(body: &str) -> bool { + let Some(pane) = std::env::var(crate::daemon::pane::TTY7_PANE_ENV) + .ok() + .and_then(|p| p.trim().parse::().ok()) + else { + return false; + }; + crate::client::PaneClient::local() + .report_agent_event(pane, std::process::id(), body) + .is_ok() } #[cfg(not(unix))] @@ -97,7 +124,19 @@ fn effective_event<'a>(agent: &str, event: &'a str, stdin_json: &str) -> Option< Some(event) } +#[cfg(test)] fn build_hook_sequence(agent: &str, event: &str, stdin_json: &str) -> Vec { + hook_sequence(&build_hook_body(agent, event, stdin_json)) +} + +/// The report as it goes out on the tty: the body wrapped in the OSC 777 +/// notification the daemon's reader looks for. +fn hook_sequence(body: &str) -> Vec { + format!("\x1b]777;notify;{AGENT_EVENT_SENTINEL};{body}\x07").into_bytes() +} + +/// The report itself, as JSON — see `cli_agent::parse_agent_event_body`. +fn build_hook_body(agent: &str, event: &str, stdin_json: &str) -> String { let payload: serde_json::Value = serde_json::from_str(stdin_json).unwrap_or(serde_json::json!({})); let mut body = serde_json::json!({ @@ -158,7 +197,7 @@ fn build_hook_sequence(agent: &str, event: &str, stdin_json: &str) -> Vec { if agent == "claude" { claude_readout(event, &payload, &mut body); } - format!("\x1b]777;notify;{AGENT_EVENT_SENTINEL};{body}\x07").into_bytes() + body.to_string() } /// How much of the end of a transcript is read for the latest reply. One diff --git a/crates/tty7-core/src/core/cli_agent.rs b/crates/tty7-core/src/core/cli_agent.rs index d81c2abe..b40d7917 100644 --- a/crates/tty7-core/src/core/cli_agent.rs +++ b/crates/tty7-core/src/core/cli_agent.rs @@ -1199,8 +1199,12 @@ pub struct AgentEvent { pub fn parse_agent_event(payload: &[u8]) -> Option { let rest = payload.strip_prefix(b"777;notify;")?; let rest = rest.strip_prefix(AGENT_EVENT_SENTINEL.as_bytes())?; - let json = rest.strip_prefix(b";")?; + parse_agent_event_body(rest.strip_prefix(b";")?) +} +/// The JSON body of an agent event, without the OSC around it — what a hook +/// hands the daemon over its socket. +pub fn parse_agent_event_body(json: &[u8]) -> Option { #[derive(Deserialize)] struct Wire { #[serde(default)] diff --git a/crates/tty7-core/src/daemon/pane.rs b/crates/tty7-core/src/daemon/pane.rs index 1ca57664..06996723 100644 --- a/crates/tty7-core/src/daemon/pane.rs +++ b/crates/tty7-core/src/daemon/pane.rs @@ -387,7 +387,7 @@ const TERM_PROGRAM_NAME: &str = "tty7"; /// Publishing the directory instead means a CLI in this shell resolves both /// endpoints with the very same functions the server used to open them. const TTY7_CONFIG_DIR_ENV: &str = "TTY7_CONFIG_DIR"; -const TTY7_PANE_ENV: &str = "TTY7_PANE"; +pub(crate) const TTY7_PANE_ENV: &str = "TTY7_PANE"; const TTY7_WS_ENV: &str = "TTY7_WS"; fn config_dir_env() -> Option { @@ -2371,21 +2371,7 @@ impl DaemonPane { && let (Some(before), Some(after)) = (facts_before, facts_after) && facts_changed(&before, &after) { - crate::core::machine::observe_pane(pane, |p| { - if after.cwd.is_some() { - p.cwd = after.cwd; - } - // Unlike the others this one is also cleared - // by a reset, so it is assigned either way. - p.osc_title = after.osc_title; - p.agent = after.agent; - if after.shell.is_some() { - p.shell = after.shell; - } - if alive { - p.live = true; - } - }); + publish_facts(pane, alive, after); } } Err(err) if err.kind() == std::io::ErrorKind::Interrupted => continue, @@ -2513,6 +2499,36 @@ impl DaemonPane { expire_stale_agent(&mut self.state.lock().unwrap(), std::time::Instant::now()); } + /// Apply an agent hook's report that came over the socket rather than + /// through the pane's output (`ClientMsg::AgentEvent`), exactly as if the + /// reader had found it there — once `pid`, the hook, is shown to run + /// under this pane's shell. The pane id the hook named came from its + /// environment, and a detached `tmux` server or the like carries that + /// into processes this pane does not own. + pub fn report_agent_event(&self, pid: u32, body: &str) -> Result<(), String> { + let shell = self + .pty() + .and_then(|pty| pty.shell_pid) + .ok_or_else(|| format!("pane {} runs no local process", self.id))?; + if !crate::daemon::procinfo::descends_from(pid, shell) { + return Err(format!("process {pid} does not run in pane {}", self.id)); + } + let event = crate::core::cli_agent::parse_agent_event_body(body.as_bytes()) + .ok_or_else(|| "not an agent event".to_string())?; + let mut st = self.state.lock().unwrap(); + if !st.alive { + return Err(format!("pane {} is not running", self.id)); + } + let before = observed_facts(&st); + apply_agent_signals(&mut st, vec![event], None); + let after = observed_facts(&st); + drop(st); + if !self.shutting_down.load(Ordering::SeqCst) && facts_changed(&before, &after) { + publish_facts(self.id, true, after); + } + Ok(()) + } + pub fn gate(&self) -> Arc { self.gate.clone() } @@ -3287,6 +3303,24 @@ fn observed_facts(st: &PaneState) -> ObservedFacts { } } +fn publish_facts(pane: u64, alive: bool, after: ObservedFacts) { + crate::core::machine::observe_pane(pane, |p| { + if after.cwd.is_some() { + p.cwd = after.cwd; + } + // Unlike the others this one is also cleared by a reset, so it is + // assigned either way. + p.osc_title = after.osc_title; + p.agent = after.agent; + if after.shell.is_some() { + p.shell = after.shell; + } + if alive { + p.live = true; + } + }); +} + fn facts_changed(before: &ObservedFacts, after: &ObservedFacts) -> bool { before.cwd != after.cwd || before.osc_title != after.osc_title @@ -4054,6 +4088,54 @@ mod tests { let _ = std::fs::remove_file(&file); } + /// A hook's report that arrives over the socket lands like one read off + /// the output — but only from a process that runs in the pane. + #[cfg(any(target_os = "macos", target_os = "linux"))] + #[test] + fn a_hook_report_counts_only_from_inside_the_pane() { + let (tx, rx) = mpsc::channel(); + let pane = DaemonPane::spawn( + 1, + Some(PathBuf::from("/")), + ws(80, 24), + Some(ShellSpec { + program: "sh".into(), + args: vec!["-c".into(), "exec cat".into()], + args_are_tty7_defaults: false, + }), + None, + None, + None, + false, + || {}, + ) + .expect("spawn pane"); + pane.attach(tx); + let shell = pane.pty().and_then(|p| p.shell_pid).expect("shell pid"); + let body = r#"{"v":1,"agent":"claude","event":"prompt-submit","session_id":"s-1"}"#; + + let outsider = pane.report_agent_event(std::process::id(), body); + let garbled = pane.report_agent_event(shell, "{not json"); + let applied = pane.report_agent_event(shell, body); + let mut status = None; + while let Ok(msg) = rx.recv_timeout(std::time::Duration::from_millis(200)) { + if let DaemonMsg::AgentStatus(s) = msg { + status = s; + } + } + pane.kill(); + + assert!( + outsider.is_err(), + "the test runner does not run in the pane" + ); + assert!(garbled.is_err()); + assert_eq!(applied, Ok(())); + let status = status.expect("the report reached the pane's subscribers"); + assert_eq!(status.status, crate::core::cli_agent::AgentStatus::Working); + assert_eq!(status.session_id.as_deref(), Some("s-1")); + } + #[cfg(any(target_os = "macos", target_os = "linux"))] #[test] fn live_pane_reports_an_uninstrumented_shells_cwd() { diff --git a/crates/tty7-core/src/daemon/procinfo.rs b/crates/tty7-core/src/daemon/procinfo.rs index 2e7a29c3..1b5db5f2 100644 --- a/crates/tty7-core/src/daemon/procinfo.rs +++ b/crates/tty7-core/src/daemon/procinfo.rs @@ -392,6 +392,63 @@ fn process_table() -> HashMap { HashMap::new() } +/// Whether `pid` runs under `ancestor`: is it, or is one of its parents. +/// +/// What an agent hook's report is checked against before it is applied to a +/// pane: the pane id it names comes from the environment, which a detached +/// `tmux` server or a shell reached some other way carries into places that +/// are not that pane any more. +pub fn descends_from(pid: u32, ancestor: u32) -> bool { + runs_under(pid, ancestor, parent_pid) +} + +/// How far up a chain [`descends_from`] climbs before giving up. A hook sits a +/// handful of levels under the pane's shell: shell, agent, the agent's +/// `sh -c`, the hook. +const MAX_ANCESTRY: usize = 64; + +fn runs_under(pid: u32, ancestor: u32, parent_of: impl Fn(u32) -> Option) -> bool { + let mut cur = pid; + for _ in 0..MAX_ANCESTRY { + if cur == ancestor { + return true; + } + match parent_of(cur) { + Some(parent) if parent > 1 && parent != cur => cur = parent, + _ => return false, + } + } + false +} + +#[cfg(target_os = "macos")] +fn parent_pid(pid: u32) -> Option { + let mut info: libc::proc_bsdinfo = unsafe { std::mem::zeroed() }; + let size = std::mem::size_of::() as libc::c_int; + let ret = unsafe { + libc::proc_pidinfo( + pid as libc::c_int, + libc::PROC_PIDTBSDINFO, + 0, + &mut info as *mut _ as *mut libc::c_void, + size, + ) + }; + (ret == size).then_some(info.pbi_ppid) +} + +#[cfg(target_os = "linux")] +fn parent_pid(pid: u32) -> Option { + let stat = std::fs::read_to_string(format!("/proc/{pid}/stat")).ok()?; + let close = stat.rfind(')')?; + stat[close + 1..].split_whitespace().nth(1)?.parse().ok() +} + +#[cfg(not(any(target_os = "macos", target_os = "linux")))] +fn parent_pid(pid: u32) -> Option { + process_table().get(&pid).map(|row| row.ppid) +} + /// The executable name behind a pid. /// /// One copy, shared with `pane.rs`. There used to be two, and each carried a @@ -893,6 +950,46 @@ fn parse_listen_addr(name: &str) -> Option<(&str, u16)> { mod tests { use super::*; + #[test] + fn a_process_runs_under_its_ancestors_only() { + // 1 ─ 10 (pane shell) ─ 20 (agent) ─ 30 (hook); 1 ─ 40 (tmux server) ─ 50 + let parent_of = |pid: u32| match pid { + 10 | 40 => Some(1), + 20 => Some(10), + 30 => Some(20), + 50 => Some(40), + _ => None, + }; + assert!(runs_under(30, 10, parent_of)); + assert!(runs_under(10, 10, parent_of)); + assert!( + !runs_under(50, 10, parent_of), + "a detached server is not the pane's" + ); + assert!( + !runs_under(10, 30, parent_of), + "a parent is not under its child" + ); + assert!( + !runs_under(99, 10, parent_of), + "a pid nobody knows is under nothing" + ); + assert!(!runs_under(7, 10, |_| Some(7)), "a loop ends"); + } + + #[cfg(any(target_os = "macos", target_os = "linux"))] + #[test] + fn this_process_descends_from_its_parent() { + let me = std::process::id(); + let parent = unsafe { libc::getppid() } as u32; + assert_eq!(parent_pid(me), Some(parent)); + assert!(descends_from(me, me)); + if parent > 1 { + assert!(descends_from(me, parent)); + assert!(!descends_from(parent, me)); + } + } + /// The uid every fabricated row belongs to unless a test says otherwise. const ME: u32 = 501; diff --git a/crates/tty7-core/src/daemon/protocol.rs b/crates/tty7-core/src/daemon/protocol.rs index 634ff5d9..c574d82b 100644 --- a/crates/tty7-core/src/daemon/protocol.rs +++ b/crates/tty7-core/src/daemon/protocol.rs @@ -896,6 +896,18 @@ pub enum ClientMsg { pane_id: u64, bytes: Vec, }, + /// An agent hook's report about the pane it runs in — the JSON body of + /// the `tty7://cli-agent` OSC 777 sequence — handed straight to the + /// daemon rather than written to the pane's tty, where the agent's own + /// output can cut into it. `pid` is the hook's process, which has to be + /// running under the pane for the report to count. Answered with + /// `InputAck` when applied and `Error` when not, so the hook knows to + /// fall back to the tty. + AgentEvent { + pane_id: u64, + pid: u32, + event: String, + }, Resize(WinSize), Detach, Kill { @@ -1078,6 +1090,7 @@ mod kind { pub const SEND_INPUT: u8 = 55; pub const HANDOFF: u8 = 56; pub const LEASE: u8 = 57; + pub const AGENT_EVENT: u8 = 58; pub const SPAWNED: u8 = 1; pub const SNAPSHOT: u8 = 2; @@ -1302,6 +1315,11 @@ impl ClientMsg { ClientMsg::SendInput { pane_id, bytes } => { write_frame(w, kind::SEND_INPUT, &to_json(&(pane_id, bytes))?) } + ClientMsg::AgentEvent { + pane_id, + pid, + event, + } => write_frame(w, kind::AGENT_EVENT, &to_json(&(pane_id, pid, event))?), ClientMsg::Resize(size) => write_frame(w, kind::RESIZE, &to_json(size)?), ClientMsg::Detach => write_frame(w, kind::DETACH, &[]), ClientMsg::Kill { pane_id } => write_frame(w, kind::KILL, &to_json(pane_id)?), @@ -1429,6 +1447,14 @@ impl ClientMsg { let (pane_id, bytes) = from_json(&payload)?; ClientMsg::SendInput { pane_id, bytes } } + kind::AGENT_EVENT => { + let (pane_id, pid, event) = from_json(&payload)?; + ClientMsg::AgentEvent { + pane_id, + pid, + event, + } + } kind::RESIZE => ClientMsg::Resize(from_json(&payload)?), kind::DETACH => ClientMsg::Detach, kind::KILL => ClientMsg::Kill { @@ -1789,6 +1815,11 @@ mod tests { pane_id: 7, bytes: Vec::new(), }, + ClientMsg::AgentEvent { + pane_id: 7, + pid: 4242, + event: r#"{"v":1,"agent":"claude","event":"stop"}"#.into(), + }, ClientMsg::Resize(SIZE), ClientMsg::Detach, ClientMsg::Kill { pane_id: 7 }, diff --git a/crates/tty7-core/src/daemon/server.rs b/crates/tty7-core/src/daemon/server.rs index 18e3457e..9051d0f2 100644 --- a/crates/tty7-core/src/daemon/server.rs +++ b/crates/tty7-core/src/daemon/server.rs @@ -1152,6 +1152,23 @@ fn handle_conn(stream: Stream, registry: Arc) -> anyhow::Result<()> { Ok(()) } + ClientMsg::AgentEvent { + pane_id, + pid, + event, + } => { + let mut w = write_stream; + let applied = match registry.get(pane_id) { + Some(pane) => pane.report_agent_event(pid, &event), + None => Err(format!("no such pane {pane_id}")), + }; + match applied { + Ok(()) => DaemonMsg::InputAck { pane_id }.encode(&mut w)?, + Err(message) => DaemonMsg::Error(message).encode(&mut w)?, + } + Ok(()) + } + ClientMsg::ListForwards { pane_id } => { let mut w = write_stream; let list = crate::daemon::ssh::SshManager::global().list_forwards(pane_id); diff --git a/crates/tty7-server/tests/agent_hook_report.rs b/crates/tty7-server/tests/agent_hook_report.rs new file mode 100644 index 00000000..181e59e5 --- /dev/null +++ b/crates/tty7-server/tests/agent_hook_report.rs @@ -0,0 +1,134 @@ +//! An agent hook run inside a pane reports to the daemon over its socket, not +//! through the pane's tty — where the agent's own output can cut into it. +#![cfg(unix)] + +use std::path::PathBuf; +use std::process::{Child, Command, Stdio}; +use std::time::{Duration, Instant}; + +use tty7_core::client::PaneClient; +use tty7_core::core::cli_agent::AgentStatus; +use tty7_core::daemon::protocol::{DaemonMsg, ShellSpec, WinSize}; + +const READY_WITHIN: Duration = Duration::from_secs(30); +const STREAM_WITHIN: Duration = Duration::from_secs(30); + +struct Daemon { + child: Child, + dir: tempfile::TempDir, +} + +impl Daemon { + fn start() -> Daemon { + let dir = tempfile::TempDir::new().unwrap(); + let child = Command::new(env!("CARGO_BIN_EXE_tty7-server")) + .arg("--daemon") + .arg("--config-dir") + .arg(dir.path()) + .env("TTY7_DATA_DIR", dir.path()) + .env("TTY7_CONTROL_SOCK", dir.path().join("control.sock")) + .stdin(Stdio::null()) + .stdout(Stdio::null()) + .stderr(Stdio::null()) + .spawn() + .expect("start tty7-server --daemon"); + let daemon = Daemon { child, dir }; + let deadline = Instant::now() + READY_WITHIN; + while daemon.panes().version().is_err() { + assert!(Instant::now() < deadline, "the daemon never came up"); + std::thread::sleep(Duration::from_millis(50)); + } + daemon + } + + fn panes(&self) -> PaneClient { + PaneClient::at(self.dir.path().join("daemon.sock")) + } +} + +impl Drop for Daemon { + fn drop(&mut self) { + let _ = self.child.kill(); + let _ = self.child.wait(); + } +} + +fn hook_shell(pane_env: &str) -> ShellSpec { + let hook = PathBuf::from(env!("CARGO_BIN_EXE_tty7-server")); + let command = format!( + "printf '%s' '{{\"session_id\":\"s-1\",\"permission_mode\":\"plan\"}}' \ + | {pane_env} '{}' agent-hook claude prompt-submit; echo hook_ran; exec cat", + hook.display() + ); + ShellSpec { + program: "/bin/sh".into(), + args: vec!["-c".into(), command], + args_are_tty7_defaults: false, + } +} + +/// Runs the hook in a fresh pane and returns what the pane printed up to the +/// hook finishing, and the agent status the daemon reported meanwhile. +fn run_hook(pane_env: &str) -> (Vec, Option) { + let daemon = Daemon::start(); + let size = WinSize { + cols: 100, + rows: 30, + cell_w: 8, + cell_h: 16, + }; + let mut session = daemon + .panes() + .spawn(None, size, Some(hook_shell(pane_env)), None, None) + .expect("spawn the pane"); + session + .set_recv_timeout(Some(STREAM_WITHIN)) + .expect("bound the stream reads"); + let mut seen = Vec::new(); + let mut status = None; + while !seen.windows(8).any(|w| w == b"hook_ran") { + match session.recv().expect("the pane keeps streaming") { + DaemonMsg::Output(bytes) | DaemonMsg::Snapshot(bytes) => seen.extend(bytes), + DaemonMsg::AgentStatus(Some(s)) => status = Some(s.status), + _ => {} + } + } + // A report over the tty is parsed off the same output, just before it. + while status.is_none() { + match session.recv() { + Ok(DaemonMsg::AgentStatus(Some(s))) => status = Some(s.status), + Ok(_) => {} + Err(_) => break, + } + } + let _ = session.kill(); + (seen, status) +} + +fn carries_the_sequence(output: &[u8]) -> bool { + output.windows(16).any(|w| w == b"tty7://cli-agent") +} + +#[test] +fn a_hook_reports_over_the_socket_and_leaves_the_tty_alone() { + let (output, status) = run_hook(""); + assert_eq!(status, Some(AgentStatus::Working)); + assert!( + !carries_the_sequence(&output), + "the report went through the tty: {:?}", + String::from_utf8_lossy(&output) + ); +} + +#[test] +fn a_hook_the_daemon_turns_away_falls_back_to_the_tty() { + // A pane id this daemon never handed out: the report is refused, and the + // hook writes the sequence to its tty as before. + let (output, status) = run_hook("TTY7_PANE=999999"); + assert!( + carries_the_sequence(&output), + "{:?} {status:?}", + String::from_utf8_lossy(&output) + ); + assert_eq!(status, Some(AgentStatus::Working)); +}