reload pi agents on an old extension: versioned extension, control socket, menu item and reload-agents

1 files changed, 131 insertions, 10 deletions

diff --git a/crates/dinggerd/src/hooks.rs b/crates/dinggerd/src/hooks.rs
index b2f9e97..0074544 100644
--- a/crates/dinggerd/src/hooks.rs
+++ b/crates/dinggerd/src/hooks.rs
@@ -21,6 +21,12 @@
 //! `{"method":"pane.mark_unread","params":{"pane_id":"%3"}}` marks an idle
 //! agent unread (`dingger unread`, pi's /unread): it shows as done again.
 //!
+//! A pi report also carries the extension's `extension` version and its
+//! `control` socket, where the daemon asks it to reload (`{"method":
+//! "reload"}`; it answers whether it does): `dingger reload-agents` sends
+//! `{"method":"agents.reload_outdated"}`, and the watcher reloads each agent
+//! on an old extension once it is idle and off screen.
+//!
 //! Each line is answered with `{"ok":true}`.
 //!
 //! `{"method":"poke"}` only wakes the watcher; `dingger focus` and tmux's
@@ -33,6 +39,7 @@ use std::collections::{HashMap, HashSet};
 use std::io::{BufRead, BufReader};
 use std::os::unix::net::{UnixListener, UnixStream};
 use std::path::PathBuf;
+use std::sync::atomic::{AtomicBool, Ordering};
 use std::sync::{Condvar, Mutex, MutexGuard, OnceLock};
 use std::time::{Duration, Instant};
 
@@ -59,6 +66,9 @@ struct PaneReports {
     session: Option<String>,
     /// The agent's importance, if reported.
     importance: Option<Importance>,
+    /// The version of the extension that reported, and its control socket.
+    extension: Option<String>,
+    control: Option<String>,
     /// When the last report arrived.
     at: Option<Instant>,
 }
@@ -85,6 +95,18 @@ pub fn take_unread(pane_id: &str) -> bool {
     unread_marks().remove(pane_id)
 }
 
+/// Whether agents on an old extension should be reloaded (`dingger
+/// reload-agents`); the watcher clears it when none is left.
+static RELOAD_WANTED: AtomicBool = AtomicBool::new(false);
+
+pub fn reload_wanted() -> bool {
+    RELOAD_WANTED.load(Ordering::Relaxed)
+}
+
+pub fn reload_done() {
+    RELOAD_WANTED.store(false, Ordering::Relaxed);
+}
+
 fn wake_signal() -> &'static (Mutex<bool>, Condvar) {
     static WAKE: OnceLock<(Mutex<bool>, Condvar)> = OnceLock::new();
     WAKE.get_or_init(|| (Mutex::new(false), Condvar::new()))
@@ -139,6 +161,10 @@ struct HookParams {
     agent_session_id: Option<String>,
     #[serde(default)]
     importance: Option<String>,
+    #[serde(default)]
+    extension: Option<String>,
+    #[serde(default)]
+    control: Option<String>,
 }
 
 fn handle_stream(stream: UnixStream) {
@@ -162,6 +188,12 @@ fn handle_stream(stream: UnixStream) {
             wake();
             continue;
         }
+        if message.method == "agents.reload_outdated" {
+            info!("reload agents on an old extension: requested");
+            RELOAD_WANTED.store(true, Ordering::Relaxed);
+            wake();
+            continue;
+        }
         if message.method == "pane.mark_unread" {
             if let Some(params) = message.params {
                 info!(pane = %params.pane_id, "marked unread");
@@ -194,12 +226,32 @@ fn handle_stream(stream: UnixStream) {
             Some(state) => info!(pane = %params.pane_id, ?state, ?importance, "hook report"),
             None => info!(pane = %params.pane_id, ?importance, "hook report without state"),
         }
-        record(params.pane_id, state, params.agent_session_path.or(params.agent_session_id), importance);
+        let reported = Reported {
+            session: params.agent_session_path.or(params.agent_session_id),
+            importance,
+            extension: params.extension,
+            control: params.control,
+        };
+        record(params.pane_id, state, reported);
         wake();
     }
 }
 
-fn record(pane_id: String, state: Option<AgentState>, session: Option<String>, importance: Option<Importance>) {
+/// What a report carries besides its state; each kept until reported again.
+#[derive(Debug, Default)]
+struct Reported {
+    session: Option<String>,
+    importance: Option<Importance>,
+    extension: Option<String>,
+    control: Option<String>,
+}
+
+/// A value fit for a tmux option: not empty, no control characters.
+fn option_value(value: Option<String>) -> Option<String> {
+    value.filter(|v| !v.is_empty() && !v.chars().any(char::is_control))
+}
+
+fn record(pane_id: String, state: Option<AgentState>, reported: Reported) {
     let mut reports = reports();
     let pane = reports.entry(pane_id).or_default();
     pane.at = Some(Instant::now());
@@ -207,11 +259,17 @@ fn record(pane_id: String, state: Option<AgentState>, session: Option<String>, i
         pane.last = Some(state);
         pane.pending.push(state);
     }
-    if let Some(session) = session.filter(|s| !s.is_empty() && !s.chars().any(char::is_control)) {
+    if let Some(session) = option_value(reported.session) {
         pane.session = Some(session);
     }
-    if importance.is_some() {
-        pane.importance = importance;
+    if reported.importance.is_some() {
+        pane.importance = reported.importance;
+    }
+    if let Some(extension) = option_value(reported.extension) {
+        pane.extension = Some(extension);
+    }
+    if let Some(control) = option_value(reported.control) {
+        pane.control = Some(control);
     }
 }
 
@@ -265,6 +323,33 @@ pub fn importance(pane_id: &str) -> Option<Importance> {
     reports().get(pane_id).and_then(|pane| pane.importance)
 }
 
+/// The version of the pane's extension and its control socket, if reported.
+pub fn extension(pane_id: &str) -> (Option<String>, Option<String>) {
+    reports().get(pane_id).map(|pane| (pane.extension.clone(), pane.control.clone())).unwrap_or_default()
+}
+
+/// Ask the extension behind `control` to reload; its answer: `Ok(())` when
+/// it does, else why not (busy, a draft, or no answer).
+pub fn request_reload(control: &str) -> Result<(), String> {
+    use std::io::Write;
+    let mut stream = UnixStream::connect(control).map_err(|err| format!("connect: {err}"))?;
+    stream.set_read_timeout(Some(Duration::from_secs(2))).map_err(|err| err.to_string())?;
+    stream.write_all(b"{\"method\":\"reload\"}\n").map_err(|err| format!("send: {err}"))?;
+    let mut answer = String::new();
+    BufReader::new(stream).read_line(&mut answer).map_err(|err| format!("no answer: {err}"))?;
+    #[derive(Deserialize)]
+    struct Answer {
+        ok: bool,
+        #[serde(default)]
+        reason: Option<String>,
+    }
+    match serde_json::from_str::<Answer>(&answer) {
+        Ok(Answer { ok: true, .. }) => Ok(()),
+        Ok(Answer { reason, .. }) => Err(reason.unwrap_or_else(|| "refused".into())),
+        Err(_) => Err(format!("unexpected answer: {}", answer.trim())),
+    }
+}
+
 /// Drop a pane's reports: the agent was not running when the panes were
 /// listed at `listed`. A report since then is kept: the agent started
 /// after the listing (Claude reports its session once, as it starts).
@@ -286,7 +371,7 @@ mod tests {
 
     #[test]
     fn a_session_report_records_only_the_session() {
-        record("%t1".into(), None, Some("/p/s.jsonl".into()), None);
+        record("%t1".into(), None, Reported { session: Some("/p/s.jsonl".into()), ..Default::default() });
         assert_eq!(session("%t1").as_deref(), Some("/p/s.jsonl"));
         assert_eq!(take_observations("%t1"), None, "state stays with the screen");
         assert_eq!(last_state("%t1"), None);
@@ -295,7 +380,7 @@ mod tests {
     #[test]
     fn a_report_after_the_pane_listing_is_not_dropped() {
         let listed = Instant::now();
-        record("%t2".into(), None, Some("/p/s.jsonl".into()), None);
+        record("%t2".into(), None, Reported { session: Some("/p/s.jsonl".into()), ..Default::default() });
         drop_reports("%t2", listed);
         assert!(session("%t2").is_some(), "the agent started after the listing");
         drop_reports("%t2", Instant::now());
@@ -304,11 +389,47 @@ mod tests {
 
     #[test]
     fn importance_is_kept_until_reported_again() {
-        record("%t3".into(), Some(AgentState::Working), None, Some(Importance::High));
-        record("%t3".into(), Some(AgentState::Idle), None, None);
+        record("%t3".into(), Some(AgentState::Working), Reported { importance: Some(Importance::High), ..Default::default() });
+        record("%t3".into(), Some(AgentState::Idle), Reported::default());
         assert_eq!(importance("%t3"), Some(Importance::High), "a report without it keeps it");
-        record("%t3".into(), None, None, Some(Importance::Normal));
+        record("%t3".into(), None, Reported { importance: Some(Importance::Normal), ..Default::default() });
         assert_eq!(importance("%t3"), Some(Importance::Normal));
         assert_eq!(last_state("%t3"), Some(AgentState::Idle), "a report without state keeps the state");
     }
+
+    #[test]
+    fn the_extension_and_its_control_socket_are_kept_until_reported_again() {
+        let reported = || Reported { extension: Some("v1".into()), control: Some("/s/pi-1.sock".into()), ..Default::default() };
+        record("%t4".into(), Some(AgentState::Idle), reported());
+        record("%t4".into(), None, Reported { importance: Some(Importance::Low), ..Default::default() });
+        assert_eq!(extension("%t4"), (Some("v1".into()), Some("/s/pi-1.sock".into())));
+        record("%t4".into(), None, Reported { extension: Some("v2".into()), ..Default::default() });
+        assert_eq!(extension("%t4").0.as_deref(), Some("v2"));
+    }
+
+    #[test]
+    fn a_reload_answer_says_whether_it_reloads() {
+        let dir = std::env::temp_dir().join(format!("dingger-reload-{}", std::process::id()));
+        std::fs::create_dir_all(&dir).unwrap();
+        for (answer, want) in [
+            ("{\"ok\":true}", Ok(())),
+            ("{\"ok\":false,\"reason\":\"draft\"}", Err("draft".to_string())),
+        ] {
+            let path = dir.join("pi.sock");
+            let _ = std::fs::remove_file(&path);
+            let listener = UnixListener::bind(&path).unwrap();
+            let server = std::thread::spawn(move || {
+                use std::io::Write;
+                let (mut stream, _) = listener.accept().unwrap();
+                let mut line = String::new();
+                BufReader::new(stream.try_clone().unwrap()).read_line(&mut line).unwrap();
+                assert_eq!(line, "{\"method\":\"reload\"}\n");
+                stream.write_all(format!("{answer}\n").as_bytes()).unwrap();
+            });
+            assert_eq!(request_reload(path.to_str().unwrap()), want);
+            server.join().unwrap();
+        }
+        assert!(request_reload(dir.join("none.sock").to_str().unwrap()).is_err(), "no extension there");
+        let _ = std::fs::remove_dir_all(&dir);
+    }
 }