6 files changed, 423 insertions, 14 deletions

diff --git a/crates/dinggerd/Cargo.toml b/crates/dinggerd/Cargo.toml
index 073b65e..8bd56b1 100644
--- a/crates/dinggerd/Cargo.toml
+++ b/crates/dinggerd/Cargo.toml
@@ -8,6 +8,7 @@ description = "macOS agent watcher for tmux panes: notifications and status with
 detect-core = { path = "../detect-core" }
 clap = { version = "4.5", default-features = false, features = ["std", "help", "usage", "error-context", "string", "derive"] }
 mac-notification-sys = "0.6"
+serde_json = { workspace = true }
 serde = { workspace = true }
 toml = { workspace = true }
 tracing = { workspace = true }
diff --git a/crates/dinggerd/assets/pi/pi-agent-state.ts b/crates/dinggerd/assets/pi/pi-agent-state.ts
new file mode 100644
index 0000000..49d4b9c
--- /dev/null
+++ b/crates/dinggerd/assets/pi/pi-agent-state.ts
@@ -0,0 +1,186 @@
+// installed by dingger
+// managed by dingger; reinstalling or updating the integration overwrites this file.
+// add custom hooks/plugins beside this file instead of editing it.
+
+import net from "node:net";
+import path from "node:path";
+import { execSync } from "node:child_process";
+
+// The dingger hook receiver listens here; override with DINGGER_AGENT_SOCKET.
+const socketPath =
+  process.env.DINGGER_AGENT_SOCKET ||
+  `${process.env.HOME}/.local/state/dingger/agent.sock`;
+
+// The extension runs inside a tmux pane, so it finds its own pane id.
+function discoverPaneId(): string | undefined {
+  try {
+    const out = execSync('tmux display-message -p "#{pane_id}"', {
+      encoding: "utf8",
+      stdio: ["ignore", "pipe", "ignore"],
+    });
+    const paneId = out.trim();
+    return paneId.startsWith("%") ? paneId : undefined;
+  } catch {
+    return undefined;
+  }
+}
+
+const paneId = process.env.DINGGER_PANE_ID || discoverPaneId();
+const source = "dingger:pi";
+
+function enabled() {
+  return !!paneId;
+}
+
+function sendRequestAttempt(request: unknown, timeoutMs: number): Promise<boolean> {
+  if (!enabled()) {
+    return Promise.resolve(true);
+  }
+
+  return new Promise((resolve) => {
+    let done = false;
+    let timeout: ReturnType<typeof setTimeout> | undefined;
+    const finish = (delivered: boolean) => {
+      if (done) return;
+      done = true;
+      if (timeout) {
+        clearTimeout(timeout);
+      }
+      socket.destroy();
+      resolve(delivered);
+    };
+
+    const socket = net.createConnection(socketPath);
+    socket.on("error", () => finish(false));
+    socket.on("connect", () => socket.write(`${JSON.stringify(request)}\n`));
+    socket.on("data", () => finish(true));
+    socket.on("end", () => finish(false));
+    timeout = setTimeout(() => finish(false), timeoutMs);
+    timeout.unref?.();
+  });
+}
+
+async function sendRequest(request: unknown): Promise<void> {
+  if (await sendRequestAttempt(request, 500)) {
+    return;
+  }
+  await sendRequestAttempt(request, 1500);
+}
+
+type AgentState = "working" | "blocked" | "idle";
+
+function report(state: AgentState, message?: string): Promise<void> {
+  return sendRequest({
+    id: `${source}:${Date.now()}:${Math.random().toString(36).slice(2)}`,
+    method: "pane.report_agent",
+    params: {
+      pane_id: paneId!,
+      source,
+      agent: "pi",
+      state,
+      message,
+      seq: Date.now() * 1000,
+    },
+  });
+}
+
+let sendInFlight = false;
+let queuedState: { state: AgentState; message?: string } | undefined;
+
+function queueState(state: AgentState, message?: string): void {
+  queuedState = { state, message };
+  if (!sendInFlight) {
+    void drainStateQueue();
+  }
+}
+
+async function drainStateQueue(): Promise<void> {
+  if (sendInFlight) {
+    return;
+  }
+
+  sendInFlight = true;
+  try {
+    while (queuedState) {
+      const next = queuedState;
+      queuedState = undefined;
+      await report(next.state, next.message);
+    }
+  } finally {
+    sendInFlight = false;
+    if (queuedState) {
+      void drainStateQueue();
+    }
+  }
+}
+
+export default function (pi) {
+  if (!enabled()) {
+    return;
+  }
+
+  let agentActive = false;
+  let blockedCount = 0;
+  let blockedMessage: string | undefined;
+  let lastState: AgentState | undefined;
+  let lastMessage: string | undefined;
+
+  function desiredState() {
+    if (blockedCount > 0) {
+      return { state: "blocked" as const, message: blockedMessage };
+    }
+    if (agentActive) {
+      return { state: "working" as const, message: undefined };
+    }
+    return { state: "idle" as const, message: undefined };
+  }
+
+  function publishState(force = false) {
+    const next = desiredState();
+    if (!force && next.state === lastState && next.message === lastMessage) {
+      return;
+    }
+    lastState = next.state;
+    lastMessage = next.message;
+    queueState(next.state, next.message);
+  }
+
+  pi.events.on("blocked", (data) => {
+    if (!data?.active) {
+      blockedCount = Math.max(0, blockedCount - 1);
+      if (blockedCount === 0) {
+        blockedMessage = undefined;
+      }
+      publishState();
+      return;
+    }
+
+    blockedCount += 1;
+    blockedMessage = data.label;
+    publishState();
+  });
+
+  pi.on("session_start", async (event, ctx) => {
+    // TUI only: RPC/JSON/print modes are headless.
+    if (ctx?.mode !== "tui") {
+      return;
+    }
+    // A reload can replace this extension mid-run without emitting another agent_start.
+    agentActive = ctx?.isIdle?.() === false;
+    publishState(true);
+  });
+
+  pi.on("agent_start", (_event, ctx) => {
+    agentActive = true;
+    publishState();
+  });
+
+  pi.on("agent_settled", (_event, ctx) => {
+    if (ctx?.isIdle?.() !== true) {
+      return;
+    }
+
+    agentActive = false;
+    publishState();
+  });
+}
diff --git a/crates/dinggerd/src/hooks.rs b/crates/dinggerd/src/hooks.rs
new file mode 100644
index 0000000..da1121e
--- /dev/null
+++ b/crates/dinggerd/src/hooks.rs
@@ -0,0 +1,142 @@
+//! 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 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.
+
+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 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);
+    }
+    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(SOCKET_FILENAME)
+}
+
+#[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,
+}
+
+fn reports() -> &'static Mutex<HashMap<String, HookReport>> {
+    static REPORTS: OnceLock<Mutex<HashMap<String, HookReport>>> = OnceLock::new();
+    REPORTS.get_or_init(|| Mutex::new(HashMap::new()))
+}
+
+#[derive(Deserialize)]
+struct HookMessage {
+    method: String,
+    params: 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() {
+        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" {
+            continue;
+        }
+        let state = match message.params.state.as_str() {
+            "working" => AgentState::Working,
+            "blocked" => AgentState::Blocked,
+            "idle" => AgentState::Idle,
+            other => {
+                debug!(state = other, "unknown hook state");
+                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(),
+            },
+        );
+    }
+}
+
+/// 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 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)
+}
+
+/// 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 reports for panes that no longer exist or no longer run the agent.
+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));
+}
diff --git a/crates/dinggerd/src/integrations.rs b/crates/dinggerd/src/integrations.rs
new file mode 100644
index 0000000..a6378d1
--- /dev/null
+++ b/crates/dinggerd/src/integrations.rs
@@ -0,0 +1,45 @@
+//! Agent integrations: hook extensions that let agents report their own
+//! state (herdr's integration model), so detection does not rely on screen
+//! scraping for the agents that support it.
+
+use std::path::PathBuf;
+
+use detect_core::Agent;
+
+/// The embedded pi extension, adapted from herdr's `herdr-agent-state.ts`
+/// (Apache-2.0, Copyright herdr contributors).
+const PI_EXTENSION: &str = include_str!("../assets/pi/pi-agent-state.ts");
+
+fn pi_extension_dir() -> Result<PathBuf, String> {
+    let base = std::env::var_os("PI_CODING_AGENT_DIR")
+        .map(PathBuf::from)
+        .or_else(|| std::env::var_os("HOME").map(|home| PathBuf::from(home).join(".pi").join("agent")))
+        .ok_or("cannot determine pi agent dir (no HOME)")?;
+    Ok(base.join("extensions"))
+}
+
+/// Install the integration for one agent. Only pi has an integration today.
+pub fn install(agent: &str) -> Result<(), String> {
+    let parsed = detect_core::parse_agent_label(agent)
+        .ok_or_else(|| format!("unknown agent: {agent}"))?;
+    match parsed {
+        Agent::Pi => install_pi(),
+        other => Err(format!(
+            "no hook integration for {} yet — screen detection is used",
+            detect_core::agent_label(other)
+        )),
+    }
+}
+
+fn install_pi() -> Result<(), String> {
+    let dir = pi_extension_dir()?;
+    std::fs::create_dir_all(&dir).map_err(|err| format!("mkdir failed: {err}"))?;
+    let path = dir.join("dingger-agent-state.ts");
+    std::fs::write(&path, PI_EXTENSION).map_err(|err| format!("write failed: {err}"))?;
+    tracing::info!(path = %path.display(), "installed pi extension");
+    println!(
+        "installed {} — new pi sessions report their state directly;\nrestart running pi agents (or reload them) to pick it up",
+        path.display()
+    );
+    Ok(())
+}
diff --git a/crates/dinggerd/src/main.rs b/crates/dinggerd/src/main.rs
index 4c8521c..61d9c15 100644
--- a/crates/dinggerd/src/main.rs
+++ b/crates/dinggerd/src/main.rs
@@ -1,5 +1,7 @@
 //! dingger: agent state watching for tmux panes.
 
+mod hooks;
+mod integrations;
 mod manifest_update;
 mod notify;
 mod tmux;
@@ -39,6 +41,11 @@ enum Command {
         /// Name of the tmux client that clicked, passed as #{client_name}.
         client: Option<String>,
     },
+    /// Install agent integrations (hook extensions that report state).
+    Install {
+        /// Agent to install the integration for.
+        agent: String,
+    },
 }
 
 #[derive(Subcommand)]
@@ -59,6 +66,10 @@ fn main() {
             interval_secs,
             no_sounds,
         } => {
+            if let Err(err) = hooks::spawn_listener() {
+                eprintln!("dingger: {err}");
+                std::process::exit(1);
+            }
             let mut watcher = watcher::Watcher::new(interval_secs, !no_sounds);
             if let Err(err) = watcher.run() {
                 eprintln!("dingger: {err}");
@@ -85,6 +96,12 @@ fn main() {
                 std::process::exit(1);
             }
         }
+        Command::Install { agent } => {
+            if let Err(err) = integrations::install(&agent) {
+                eprintln!("dingger: {err}");
+                std::process::exit(1);
+            }
+        }
     }
 }
 
diff --git a/crates/dinggerd/src/watcher.rs b/crates/dinggerd/src/watcher.rs
index 9fd999d..d352ab2 100644
--- a/crates/dinggerd/src/watcher.rs
+++ b/crates/dinggerd/src/watcher.rs
@@ -6,6 +6,7 @@ use std::collections::HashMap;
 use detect_core::{detect_agent_with_osc, AgentState};
 use tracing::{debug, info, warn};
 
+use crate::hooks;
 use crate::notify::{play_sound, send_notification, NotificationKind};
 use crate::tmux;
 
@@ -93,6 +94,12 @@ impl Watcher {
         // Forget panes that went away.
         self.panes
             .retain(|pane_id, _| panes.iter().any(|pane| pane.pane_id == *pane_id));
+        hooks::forget(
+            &panes
+                .iter()
+                .map(|pane| pane.pane_id.clone())
+                .collect::<Vec<_>>(),
+        );
 
         for pane in &panes {
             let prev_state = self
@@ -102,27 +109,34 @@ impl Watcher {
                 .unwrap_or(AgentState::Unknown);
 
             // The detection result is an observation of live agent state.
-            // Viewer screens (transcripts, pickers) return
+            // Agent-reported state (hook receiver) is authoritative while the
+            // pane runs the agent; screen detection is the fallback for panes
+            // that never report. Viewer screens (transcripts, pickers) return
             // `skip_state_update: true` — they say nothing about state and
             // leave the previous one standing (herdr's publishing contract).
             let observed = if pane.dead {
                 Some(AgentState::Idle)
             } else if let Some(agent) = pane.agent {
-                let screen = match tmux::capture_pane(&pane.pane_id) {
-                    Ok(screen) => screen,
-                    Err(err) => {
-                        warn!(target = %pane.target, %err, "capture failed");
-                        continue;
-                    }
-                };
-                // tmux pane_title carries the last OSC title the application
-                // set; tmux does not re-expose OSC progress sequences.
-                let detection =
-                    detect_agent_with_osc(Some(agent), &screen, &pane.pane_title, "");
-                (!detection.skip_state_update).then_some(detection.state)
+                if let Some(state) = hooks::hook_state(&pane.pane_id) {
+                    Some(state)
+                } else {
+                    let screen = match tmux::capture_pane(&pane.pane_id) {
+                        Ok(screen) => screen,
+                        Err(err) => {
+                            warn!(target = %pane.target, %err, "capture failed");
+                            continue;
+                        }
+                    };
+                    // tmux pane_title carries the last OSC title the application
+                    // set; tmux does not re-expose OSC progress sequences.
+                    let detection =
+                        detect_agent_with_osc(Some(agent), &screen, &pane.pane_title, "");
+                    (!detection.skip_state_update).then_some(detection.state)
+                }
             } else {
                 // A pane that stops running an agent (agent quit into a
-                // shell) is no longer an observation.
+                // shell) is no longer an observation; a stale hook report
+                // from the dead agent is not authoritative.
                 Some(AgentState::Unknown)
             };
 
@@ -130,6 +144,10 @@ impl Watcher {
                 debug!(target = %pane.target, "observation skipped (viewer screen)");
                 continue;
             };
+            if new_state == AgentState::Unknown {
+                // The agent is gone; its last hook report must not outlive it.
+                hooks::drop_report(&pane.pane_id);
+            }
 
             // Keep the pane's agent label current even when the state is
             // unchanged (e.g. an agent swap between two idle panes).