1 files changed, 94 insertions, 50 deletions

diff --git a/crates/dinggerd/src/hooks.rs b/crates/dinggerd/src/hooks.rs
index da1121e..40f0f20 100644
--- a/crates/dinggerd/src/hooks.rs
+++ b/crates/dinggerd/src/hooks.rs
@@ -6,23 +6,27 @@
 //! {"method":"pane.report_agent","params":{"pane_id":"%3","agent":"pi","state":"working"}}
 //! ```
 //!
-//! Reports are authoritative for their pane until the next report or until
-//! the pane stops running the agent; screen detection is the fallback for
-//! panes that never report.
+//! 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::{Mutex, OnceLock};
-use std::time::Instant;
+use std::sync::{Condvar, Mutex, MutexGuard, OnceLock};
+use std::time::Duration;
 
 use detect_core::AgentState;
 use serde::Deserialize;
 use tracing::{debug, info, warn};
 
-pub const SOCKET_FILENAME: &str = "agent.sock";
-
 pub fn socket_path() -> PathBuf {
     if let Some(path) = std::env::var_os("DINGGER_AGENT_SOCKET") {
         return PathBuf::from(path);
@@ -30,54 +34,88 @@ pub fn socket_path() -> PathBuf {
     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")
-            })
+            std::env::var_os("HOME")
+                .map(|home| PathBuf::from(home).join(".local").join("state").join("dingger"))
         })
         .unwrap_or_else(|| PathBuf::from(".dingger"));
-    state.join(SOCKET_FILENAME)
+    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();
 }
 
-#[derive(Debug, Clone)]
-struct HookReport {
-    state: AgentState,
-    #[allow(dead_code)] // kept for future hook diagnostics
-    agent: Option<String>,
-    #[allow(dead_code)] // kept for future staleness policy
-    seen: Instant,
+/// 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;
 }
 
-fn reports() -> &'static Mutex<HashMap<String, HookReport>> {
-    static REPORTS: OnceLock<Mutex<HashMap<String, HookReport>>> = OnceLock::new();
-    REPORTS.get_or_init(|| Mutex::new(HashMap::new()))
+/// 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,
-    params: HookParams,
+    #[serde(default)]
+    params: Option<HookParams>,
 }
 
 #[derive(Deserialize)]
 struct HookParams {
     pane_id: String,
     state: String,
-    #[serde(default)]
-    agent: Option<String>,
 }
 
 fn handle_stream(stream: UnixStream) {
-    let reader = BufReader::new(stream);
-    for line in reader.lines() {
+    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 != "pane.report_agent" {
+        if message.method == "poke" {
+            wake();
             continue;
         }
-        let state = match message.params.state.as_str() {
+        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,
@@ -86,17 +124,14 @@ fn handle_stream(stream: UnixStream) {
                 continue;
             }
         };
-        let pane_id = message.params.pane_id;
-        debug!(pane = %pane_id, ?state, "hook report");
-        let mut reports = reports().lock().unwrap_or_else(|poisoned| poisoned.into_inner());
-        reports.insert(
-            pane_id,
-            HookReport {
-                state,
-                agent: message.params.agent,
-                seen: Instant::now(),
-            },
-        );
+        debug!(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();
     }
 }
 
@@ -123,20 +158,29 @@ pub fn spawn_listener() -> Result<(), String> {
     Ok(())
 }
 
-/// The authoritative state for a pane, if an extension reported one.
-pub fn hook_state(pane_id: &str) -> Option<AgentState> {
-    let reports = reports().lock().unwrap_or_else(|poisoned| poisoned.into_inner());
-    reports.get(pane_id).map(|report| report.state)
+/// 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 single pane's report (the agent exited).
-pub fn drop_report(pane_id: &str) {
-    let mut reports = reports().lock().unwrap_or_else(|poisoned| poisoned.into_inner());
-    reports.remove(pane_id);
+/// 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 or no longer run the agent.
+/// Drop reports for panes that no longer exist.
 pub fn forget(pane_ids_to_keep: &[String]) {
-    let mut reports = reports().lock().unwrap_or_else(|poisoned| poisoned.into_inner());
-    reports.retain(|pane_id, _| pane_ids_to_keep.contains(pane_id));
+    reports().retain(|pane_id, _| pane_ids_to_keep.contains(pane_id));
 }