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

//! Poll-and-detect loop: tmux pane capture → detection engine → herdr's
//! transition policy → notifications and `@agent_state` write-back.

use std::collections::HashMap;

use detect_core::{detect_agent_with_osc, AgentState};
use tracing::{debug, info, warn};

use crate::notify::{play_sound, send_notification, NotificationKind};
use crate::tmux;

/// When a pane became ready for the user, as unix seconds. Ready means the
/// agent is waiting on the user: idle at its prompt or blocked on input.
fn is_ready(state: AgentState) -> bool {
    matches!(state, AgentState::Idle | AgentState::Blocked)
}

pub struct Watcher {
    /// Published state plus the unix second it became ready, per pane id.
    /// Published state plus the unix second it became ready, per pane id.
    /// `status` reads the same values back from tmux's `@agent_state`, so no
    /// other durable state is needed.
    panes: HashMap<String, PaneTrack>,
    /// Last (queue, counts) written to the global options; avoids rewriting
    /// them every tick.
    last_published: Option<(String, (String, String, String))>,
    interval_secs: u64,
    sounds: bool,
}

struct PaneTrack {
    state: AgentState,
    ready_since: Option<u64>,
}

/// herdr's notification policy (`notification_toast_for_state_change`).
pub fn notification_kind_for_transition(
    prev_state: AgentState,
    new_state: AgentState,
) -> Option<NotificationKind> {
    if new_state == prev_state {
        return None;
    }
    match new_state {
        AgentState::Blocked => Some(NotificationKind::NeedsAttention),
        AgentState::Idle
            if matches!(prev_state, AgentState::Working | AgentState::Blocked) =>
        {
            Some(NotificationKind::Finished)
        }
        _ => None,
    }
}

impl Watcher {
    pub fn new(interval_secs: u64, sounds: bool) -> Self {
        Self {
            panes: HashMap::new(),
            last_published: None,
            interval_secs: interval_secs.max(1),
            sounds,
        }
    }

    pub fn run(&mut self) -> Result<(), String> {
        info!(interval = self.interval_secs, "watching tmux panes");
        // dingger is a client of the tmux server, which may not exist yet —
        // it starts when the first terminal opens one. Wait for it, then
        // keep watching across server restarts.
        let mut had_server = false;
        loop {
            match self.tick() {
                Ok(()) => {
                    if !had_server {
                        info!("tmux server available");
                        had_server = true;
                    }
                }
                Err(err) => {
                    if had_server {
                        warn!(%err, "tmux server unavailable; waiting for it to return");
                    }
                    had_server = false;
                }
            }
            std::thread::sleep(std::time::Duration::from_secs(self.interval_secs));
        }
    }

    fn tick(&mut self) -> Result<(), String> {
        let panes = tmux::list_panes()?;

        // Forget panes that went away.
        self.panes
            .retain(|pane_id, _| panes.iter().any(|pane| pane.pane_id == *pane_id));

        for pane in &panes {
            let prev_state = self
                .panes
                .get(&pane.pane_id)
                .map(|track| track.state)
                .unwrap_or(AgentState::Unknown);

            // The detection result is an observation of live agent state.
            // 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)
            } else {
                // A pane that stops running an agent (agent quit into a
                // shell) is no longer an observation.
                Some(AgentState::Unknown)
            };

            let Some(new_state) = observed else {
                debug!(target = %pane.target, "observation skipped (viewer screen)");
                continue;
            };

            // Keep the pane's agent label current even when the state is
            // unchanged (e.g. an agent swap between two idle panes).
            let agent_label = pane
                .agent
                .map(detect_core::agent_label)
                .unwrap_or("")
                .to_string();
            if pane.written_agent != agent_label {
                if let Err(err) = tmux::set_agent(&pane.pane_id, pane.agent) {
                    warn!(target = %pane.target, %err, "failed to write @agent");
                }
            }

            if new_state == prev_state {
                continue;
            }
            let ready_since = is_ready(new_state).then(unix_now);
            self.panes.insert(
                pane.pane_id.clone(),
                PaneTrack {
                    state: new_state,
                    ready_since,
                },
            );
            debug!(target = %pane.target, ?prev_state, ?new_state, "state change");

            if let Err(err) = tmux::set_agent_state(&pane.pane_id, new_state) {
                warn!(target = %pane.target, %err, "failed to write @agent_state");
            }
            // A fresh transition re-enters the ready queue even if the user
            // already jumped to this pane in a previous ready episode.
            let _ = tmux::clear_ack(&pane.pane_id);


            if let Some(kind) = notification_kind_for_transition(prev_state, new_state) {
                if pane.suppresses_notifications() {
                    debug!(target = %pane.target, "suppressed (active pane)");
                    continue;
                }
                let context = format!("{} ({})", pane.target, agent_label);
                send_notification(&agent_label, &kind, &context);
                if self.sounds {
                    play_sound(&kind);
                }
            }
        }

        self.publish_ready_queue(&panes);
        Ok(())
    }

    /// Publish the FIFO ready queue (oldest first) as a server-global option;
    /// `dingger focus` pops its head, and the tmux status line renders counts.
    fn publish_ready_queue(&mut self, panes: &[tmux::Pane]) {
        let mut ready: Vec<(&str, String, u64)> = Vec::new(); // (pane_id, target, ready_since)
        let mut blocked = 0usize;
        let mut working = 0usize;
        for pane in panes {
            let Some(track) = self.panes.get(&pane.pane_id) else {
                continue;
            };
            match track.state {
                AgentState::Working => working += 1,
                AgentState::Idle | AgentState::Blocked => {
                    let acked = pane.written_ack == "1";
                    if !acked {
                        if let Some(since) = track.ready_since {
                            if track.state == AgentState::Blocked {
                                blocked += 1;
                            }
                            ready.push((pane.pane_id.as_str(), pane.target.clone(), since));
                        }
                    }
                }
                AgentState::Unknown => {}
            }
        }
        ready.sort_by_key(|(_, _, since)| *since);

        let queue = ready
            .iter()
            .map(|(pane_id, _, since)| format!("{pane_id}:{since}"))
            .collect::<Vec<_>>()
            .join(",");
        let counts = (
            ready.len().to_string(),
            blocked.to_string(),
            working.to_string(),
        );
        let next = (queue.clone(), counts.clone());
        if self.last_published.as_ref() != Some(&next) {
            self.last_published = Some(next);
            for (name, value) in [
                ("@dingger_queue", queue.as_str()),
                ("@dingger_ready", counts.0.as_str()),
                ("@dingger_blocked", counts.1.as_str()),
                ("@dingger_working", counts.2.as_str()),
            ] {
                if let Err(err) = tmux::set_global(name, value) {
                    warn!(%err, "failed to write {name}");
                }
            }
        }
    }
}

fn unix_now() -> u64 {
    std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .map(|d| d.as_secs())
        .unwrap_or(0)
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn policy_matches_herdr() {
        assert!(notification_kind_for_transition(AgentState::Working, AgentState::Blocked)
            .is_some_and(|kind| matches!(kind, NotificationKind::NeedsAttention)));
        assert!(notification_kind_for_transition(AgentState::Blocked, AgentState::Blocked).is_none());
        assert!(notification_kind_for_transition(AgentState::Unknown, AgentState::Blocked)
            .is_some_and(|kind| matches!(kind, NotificationKind::NeedsAttention)));
        assert!(notification_kind_for_transition(AgentState::Working, AgentState::Idle)
            .is_some_and(|kind| matches!(kind, NotificationKind::Finished)));
        assert!(notification_kind_for_transition(AgentState::Blocked, AgentState::Idle)
            .is_some_and(|kind| matches!(kind, NotificationKind::Finished)));
        // Unknown → Idle is a first observation, not a completion.
        assert!(notification_kind_for_transition(AgentState::Unknown, AgentState::Idle).is_none());
        assert!(notification_kind_for_transition(AgentState::Idle, AgentState::Working).is_none());
        assert!(notification_kind_for_transition(AgentState::Idle, AgentState::Unknown).is_none());
    }
}