//! Hook receiver: agents report their own state (herdr's integration model). //! //! A unix socket accepts JSON lines from agent extensions: //! //! ```json //! {"method":"pane.report_agent","params":{"pane_id":"%3","agent":"pi","state":"working"}} //! ``` //! //! Reports are state-change *events*: the watcher applies every one in //! order, so a turn that starts and finishes between two polls is still a //! completion, and each report wakes the watcher immediately. //! //! `{"method":"poke"}` only wakes the watcher; `dingger focus` and tmux's //! focus-change hooks send it so attendance updates without waiting for the //! next poll. The last reported state stays authoritative for its pane //! until the next report or until the pane stops running the agent; screen //! detection is the fallback for panes that never report. use std::collections::HashMap; use std::io::{BufRead, BufReader}; use std::os::unix::net::{UnixListener, UnixStream}; use std::path::PathBuf; use std::sync::{Condvar, Mutex, MutexGuard, OnceLock}; use std::time::Duration; use detect_core::AgentState; use serde::Deserialize; use tracing::{debug, info, warn}; pub fn socket_path() -> PathBuf { if let Some(path) = std::env::var_os("DINGGER_AGENT_SOCKET") { return PathBuf::from(path); } let state = std::env::var_os("DINGGER_STATE_DIR") .map(PathBuf::from) .or_else(|| { std::env::var_os("HOME") .map(|home| PathBuf::from(home).join(".local").join("state").join("dingger")) }) .unwrap_or_else(|| PathBuf::from(".dingger")); state.join("agent.sock") } #[derive(Debug, Default)] struct PaneReports { /// Last reported state: authoritative until the next report. last: Option, /// Reports not yet applied by the watcher, oldest first. pending: Vec, } fn reports() -> MutexGuard<'static, HashMap> { static REPORTS: OnceLock>> = OnceLock::new(); REPORTS .get_or_init(|| Mutex::new(HashMap::new())) .lock() .unwrap_or_else(|poisoned| poisoned.into_inner()) } fn wake_signal() -> &'static (Mutex, Condvar) { static WAKE: OnceLock<(Mutex, Condvar)> = OnceLock::new(); WAKE.get_or_init(|| (Mutex::new(false), Condvar::new())) } fn wake() { let (flag, condvar) = wake_signal(); *flag.lock().unwrap_or_else(|poisoned| poisoned.into_inner()) = true; condvar.notify_all(); } /// Sleep until woken by a report or poke, or until `timeout` passes. pub fn wait_for_wake(timeout: Duration) { let (flag, condvar) = wake_signal(); let guard = flag.lock().unwrap_or_else(|poisoned| poisoned.into_inner()); let (mut guard, _) = condvar .wait_timeout_while(guard, timeout, |woken| !*woken) .unwrap_or_else(|poisoned| poisoned.into_inner()); *guard = false; } /// Wake a running watcher (used by `dingger focus` and tmux focus hooks). pub fn poke() -> Result<(), String> { use std::io::Write; let mut stream = UnixStream::connect(socket_path()) .map_err(|err| format!("dingger watch is not running: {err}"))?; stream .write_all(b"{\"method\":\"poke\"}\n") .map_err(|err| format!("poke failed: {err}")) } #[derive(Deserialize)] struct HookMessage { method: String, #[serde(default)] params: Option, } #[derive(Deserialize)] struct HookParams { pane_id: String, state: String, } fn handle_stream(stream: UnixStream) { for line in BufReader::new(stream).lines() { let Ok(line) = line else { return }; let Ok(message) = serde_json::from_str::(&line) else { warn!(%line, "invalid hook message"); continue; }; if message.method == "poke" { wake(); continue; } let (true, Some(params)) = (message.method == "pane.report_agent", message.params) else { continue; }; let state = match params.state.as_str() { "working" => AgentState::Working, "blocked" => AgentState::Blocked, "idle" => AgentState::Idle, other => { debug!(state = other, "unknown hook state"); continue; } }; info!(pane = %params.pane_id, ?state, "hook report"); { let mut reports = reports(); let pane = reports.entry(params.pane_id).or_default(); pane.last = Some(state); pane.pending.push(state); } wake(); } } /// Start listening; reports accumulate for the watcher to consume. pub fn spawn_listener() -> Result<(), String> { let path = socket_path(); if let Some(parent) = path.parent() { std::fs::create_dir_all(parent).map_err(|err| format!("mkdir failed: {err}"))?; } let _ = std::fs::remove_file(&path); let listener = UnixListener::bind(&path).map_err(|err| format!("bind {}: {err}", path.display()))?; info!(path = %path.display(), "hook receiver listening"); std::thread::spawn(move || { for stream in listener.incoming() { match stream { Ok(stream) => { std::thread::spawn(move || handle_stream(stream)); } Err(err) => warn!(%err, "hook listener accept failed"), } } }); Ok(()) } /// The pane's hook observations for this tick: every report since the last /// tick, in order; otherwise the last reported state; `None` if the pane /// never reported to this process. pub fn take_observations(pane_id: &str) -> Option> { let mut reports = reports(); let pane = reports.get_mut(pane_id)?; if pane.pending.is_empty() { return pane.last.map(|state| vec![state]); } Some(std::mem::take(&mut pane.pending)) } /// The last state the pane's extension reported, if any. pub fn last_state(pane_id: &str) -> Option { reports().get(pane_id).and_then(|pane| pane.last) } /// Drop a pane's reports (the agent exited). pub fn drop_reports(pane_id: &str) { reports().remove(pane_id); } /// Drop reports for panes that no longer exist. pub fn forget(pane_ids_to_keep: &[String]) { reports().retain(|pane_id, _| pane_ids_to_keep.contains(pane_id)); }