crates/dinggerd/src/hooks.rs (view raw)
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 |
//! 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<AgentState>,
/// Reports not yet applied by the watcher, oldest first.
pending: Vec<AgentState>,
}
fn reports() -> MutexGuard<'static, HashMap<String, PaneReports>> {
static REPORTS: OnceLock<Mutex<HashMap<String, PaneReports>>> = OnceLock::new();
REPORTS
.get_or_init(|| Mutex::new(HashMap::new()))
.lock()
.unwrap_or_else(|poisoned| poisoned.into_inner())
}
fn wake_signal() -> &'static (Mutex<bool>, Condvar) {
static WAKE: OnceLock<(Mutex<bool>, 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<HookParams>,
}
#[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::<HookMessage>(&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<Vec<AgentState>> {
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<AgentState> {
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));
}
|