path: root/crates/dinggerd/src/hooks.rs

//! 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"}}
//! ```
//!
//! A report may carry the agent's session reference (`agent_session_path`
//! or `agent_session_id`, herdr's names), which restore uses to resume it.
//!
//! 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);
    }
    crate::snapshot::state_dir().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>,
    /// The agent's latest session reference.
    session: Option<String>,
}

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,
    #[serde(default)]
    agent_session_path: Option<String>,
    #[serde(default)]
    agent_session_id: Option<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);
            let session = params.agent_session_path.or(params.agent_session_id);
            if let Some(session) = session.filter(|s| !s.is_empty() && !s.chars().any(char::is_control)) {
                pane.session = Some(session);
            }
        }
        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)
}

/// The pane's agent session reference, if its extension reported one.
pub fn session(pane_id: &str) -> Option<String> {
    reports().get(pane_id).and_then(|pane| pane.session.clone())
}

/// 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));
}