all repos — dingger @ de1a3ce438d31df262a026bfb104ca07e656ee30

Watches the AI coding agents in your tmux panes on macOS and tells you which need you, from the menu bar, a floating panel, the status line and notifications; jumps to them, and restores tmux sessions and agents after a crash

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