IdeaSDK/crates/infrastructure/src/inspector/claude_turn_watcher.rs

//! [`ClaudeTranscriptTurnWatcher`] — the [`TurnWatcher`] adapter for the **Claude Code**
//! CLI (rendez-vous no-reply backstop).
//!
//! It replaces the dead PTY prompt-ready sniff as the **end-of-turn** authority. Claude
//! records each completed turn as a `turn_duration` entry in its JSONL transcript; this
//! adapter **tail-polls** the agent's per-run-dir transcript folder
//! (`<home>/.claude/projects/<encoded-run-dir>/`) every [`POLL_INTERVAL`] and fires the
//! [`OnTurnEnd`] callback each time the **aggregate** count of `turn_duration` records
//! grows above the **baseline** captured at arm time.
//!
//! ## Why the whole folder, not one file
//!
//! The transcript file is named by Claude's **engine session id** (the resumable), which
//! is unknown at cold start (the file does not exist yet) and is **not** the IdeA pair
//! `conversation_id`. Since every IdeA agent runs in its **own isolated run dir**
//! (`.ideai/run/<agent>/`), its `<encoded-run-dir>` folder belongs to exactly **one**
//! agent: aggregating `turn_duration` across the `.jsonl` files in that folder is a sound
//! per-agent turn-end signal that needs neither the engine id nor file mtimes (the
//! [`domain::ports::FileSystem`] port exposes neither).
//!
//! ## Robustness (cadrage pièges)
//!
//! - **Baseline at arm** ⇒ a pre-existing transcript (relaunch / background wake) never
//!   fires a phantom turn end (cf. `mcp-e2e-findings-reply-wedge-phantom-busy`).
//! - **Cold start**: the folder/file may not exist yet ⇒ treated as count `0`, the watch
//!   simply waits for it to appear (the first turn is also covered by the MCP-initialize
//!   cold-start release).
//! - **Rotation / truncation**: if the aggregate count ever **drops**, the baseline is
//!   reset to the new (lower) value — best-effort, never fires spuriously.
//! - **Poll, not inotify** ⇒ portable inside the AppImage.

use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::Arc;
use std::time::Duration;

use domain::ids::AgentId;
use domain::ports::{FileSystem, OnTurnEnd, RemotePath, TurnWatchHandle, TurnWatcher};
use domain::profile::AgentProfile;
use domain::project::ProjectPath;

use super::claude_paths::{claude_project_dir, supports_claude};

/// Tail-poll interval (cadrage : 250-500 ms). 300 ms balances latency vs. cost.
const POLL_INTERVAL: Duration = Duration::from_millis(300);

/// Substring marking a completed turn in a Claude transcript line.
const TURN_END_MARKER: &str = "turn_duration";

/// [`TurnWatcher`] reading Claude Code transcripts. Composes a [`FileSystem`] port (so it
/// stays OS- and Tauri-agnostic and trivially testable) plus the base directory standing
/// in for the user's home (`<home>/.claude/projects/`).
#[derive(Clone)]
pub struct ClaudeTranscriptTurnWatcher {
    fs: Arc<dyn FileSystem>,
    home_dir: String,
}

impl ClaudeTranscriptTurnWatcher {
    /// Builds the watcher from an injected [`FileSystem`] and the user's home directory
    /// (resolved by the composition root, e.g. from `$HOME`).
    #[must_use]
    pub fn new(fs: Arc<dyn FileSystem>, home_dir: impl Into<String>) -> Self {
        Self {
            fs,
            home_dir: home_dir.into(),
        }
    }
}

/// Handle whose [`Drop`] stops the polling task **effectively and idempotently**: it
/// flips the shared stop flag (observed at the top *and* bottom of each loop iteration)
/// **and** aborts the task (so an in-flight `await` cannot fire after the drop).
struct ClaudeTurnWatchHandle {
    stop: Arc<AtomicBool>,
    task: tokio::task::JoinHandle<()>,
}

impl TurnWatchHandle for ClaudeTurnWatchHandle {}

impl Drop for ClaudeTurnWatchHandle {
    fn drop(&mut self) {
        self.stop.store(true, Ordering::SeqCst);
        self.task.abort();
    }
}

impl TurnWatcher for ClaudeTranscriptTurnWatcher {
    fn supports(&self, profile: &AgentProfile) -> bool {
        supports_claude(profile)
    }

    fn watch(
        &self,
        agent: AgentId,
        conversation_id: Option<String>,
        cwd: ProjectPath,
        on_turn_end: OnTurnEnd,
    ) -> Box<dyn TurnWatchHandle> {
        let fs = Arc::clone(&self.fs);
        let dir = claude_project_dir(&self.home_dir, &cwd);
        let stop = Arc::new(AtomicBool::new(false));
        let stop_task = Arc::clone(&stop);
        application::diag!(
            "[turn-watcher] arm agent={agent} dir={} conversation={}",
            dir.0,
            conversation_id.as_deref().unwrap_or("<none>"),
        );
        let task = tokio::spawn(async move {
            // Baseline = turn ends already present at arm time (never re-fired).
            let mut last = count_turn_ends(fs.as_ref(), &dir).await;
            application::diag!("[turn-watcher] baseline agent={agent} count={last}");
            loop {
                if stop_task.load(Ordering::SeqCst) {
                    break;
                }
                tokio::time::sleep(POLL_INTERVAL).await;
                if stop_task.load(Ordering::SeqCst) {
                    break;
                }
                let current = count_turn_ends(fs.as_ref(), &dir).await;
                if current < last {
                    // Rotation / truncation : re-baseline (best-effort, no spurious fire).
                    last = current;
                    continue;
                }
                if current > last {
                    application::diag!(
                        "[turn-watcher] turn_duration detected agent={agent} count {last}->{current} -> turn_ended"
                    );
                    last = current;
                    // Callback invoked from the polling task, holding no lock (it ends up
                    // in `InputMediator::turn_ended`). One fire per detected increase: a
                    // burst still ends the currently-busy turn; an idle agent is a no-op.
                    on_turn_end(agent);
                }
            }
        });
        Box::new(ClaudeTurnWatchHandle { stop, task })
    }
}

/// Aggregate count of `turn_duration` records across **all** `.jsonl` transcripts in the
/// agent's per-run-dir folder. A missing folder/file (cold start) ⇒ `0`. Unreadable
/// files are skipped (best-effort, never fatal).
async fn count_turn_ends(fs: &dyn FileSystem, dir: &RemotePath) -> usize {
    let Ok(entries) = fs.list(dir).await else {
        return 0; // folder not created yet (cold start) or transient error.
    };
    let mut total = 0usize;
    for entry in entries {
        if entry.is_dir || !entry.name.ends_with(".jsonl") {
            continue;
        }
        let path = RemotePath::new(format!("{}/{}", dir.0, entry.name));
        let Ok(bytes) = fs.read(&path).await else {
            continue;
        };
        total += count_marker_lines(&bytes);
    }
    total
}

/// Counts transcript **lines** containing the [`TURN_END_MARKER`] (one logical turn end
/// per `turn_duration` record). Substring match on the raw bytes — cheap and tolerant of
/// partial/unparseable lines.
fn count_marker_lines(body: &[u8]) -> usize {
    String::from_utf8_lossy(body)
        .lines()
        .filter(|line| line.contains(TURN_END_MARKER))
        .count()
}

#[cfg(test)]
mod tests {
    use super::*;
    use async_trait::async_trait;
    use domain::ports::{DirEntry, FsError};
    use std::collections::HashMap;
    use std::sync::Mutex;

    /// In-memory [`FileSystem`] double: a flat map of path → bytes; `list` returns the
    /// basenames whose parent is the queried dir. Only `read`/`list` are exercised.
    #[derive(Default)]
    struct FakeFs {
        files: Mutex<HashMap<String, Vec<u8>>>,
    }
    impl FakeFs {
        fn set(&self, path: &str, body: &str) {
            self.files
                .lock()
                .unwrap()
                .insert(path.to_owned(), body.as_bytes().to_vec());
        }
    }
    #[async_trait]
    impl FileSystem for FakeFs {
        async fn read(&self, path: &RemotePath) -> Result<Vec<u8>, FsError> {
            self.files
                .lock()
                .unwrap()
                .get(&path.0)
                .cloned()
                .ok_or_else(|| FsError::NotFound(path.0.clone()))
        }
        async fn write(&self, path: &RemotePath, data: &[u8]) -> Result<(), FsError> {
            self.files.lock().unwrap().insert(path.0.clone(), data.to_vec());
            Ok(())
        }
        async fn exists(&self, path: &RemotePath) -> Result<bool, FsError> {
            Ok(self.files.lock().unwrap().contains_key(&path.0))
        }
        async fn create_dir_all(&self, _path: &RemotePath) -> Result<(), FsError> {
            Ok(())
        }
        async fn list(&self, path: &RemotePath) -> Result<Vec<DirEntry>, FsError> {
            let prefix = format!("{}/", path.0);
            let files = self.files.lock().unwrap();
            let mut out = Vec::new();
            for key in files.keys() {
                if let Some(rest) = key.strip_prefix(&prefix) {
                    if !rest.contains('/') {
                        out.push(DirEntry {
                            name: rest.to_owned(),
                            is_dir: false,
                        });
                    }
                }
            }
            if out.is_empty() && !files.keys().any(|k| k.starts_with(&prefix)) {
                return Err(FsError::NotFound(path.0.clone()));
            }
            Ok(out)
        }
        async fn symlink(&self, _src: &RemotePath, _dst: &RemotePath) -> Result<(), FsError> {
            Ok(())
        }
    }

    fn agent(n: u128) -> AgentId {
        AgentId::from_uuid(uuid::Uuid::from_u128(n))
    }

    fn line(extra: &str) -> String {
        format!("{{\"type\":\"result\",\"{extra}\":1}}\n")
    }

    /// The run dir `/run/a` encodes to `-run-a` ⇒ folder `<home>/.claude/projects/-run-a`.
    fn transcript_path(name: &str) -> String {
        format!("/home/me/.claude/projects/-run-a/{name}")
    }

    fn count_calls() -> (Arc<Mutex<Vec<AgentId>>>, OnTurnEnd) {
        let log: Arc<Mutex<Vec<AgentId>>> = Arc::new(Mutex::new(Vec::new()));
        let log2 = Arc::clone(&log);
        let cb: OnTurnEnd = Arc::new(move |a: AgentId| log2.lock().unwrap().push(a));
        (log, cb)
    }

    async fn wait_until(mut cond: impl FnMut() -> bool) -> bool {
        for _ in 0..100 {
            if cond() {
                return true;
            }
            tokio::time::sleep(Duration::from_millis(20)).await;
        }
        cond()
    }

    #[test]
    fn count_marker_lines_counts_only_turn_duration_lines() {
        let body = concat!(
            "{\"type\":\"user\"}\n",
            "{\"type\":\"result\",\"turn_duration\":12}\n",
            "garbage line\n",
            "{\"type\":\"result\",\"turn_duration\":7}\n",
        );
        assert_eq!(count_marker_lines(body.as_bytes()), 2);
    }

    #[tokio::test]
    async fn fires_on_new_turn_duration_above_baseline() {
        let fs = Arc::new(FakeFs::default());
        // Baseline: one pre-existing turn end ⇒ must NOT fire for it.
        fs.set(&transcript_path("engine-1.jsonl"), &line("turn_duration"));
        let watcher = ClaudeTranscriptTurnWatcher::new(Arc::clone(&fs) as Arc<dyn FileSystem>, "/home/me");
        let (log, cb) = count_calls();
        let cwd = ProjectPath::new("/run/a").expect("path");
        let _h = watcher.watch(agent(1), None, cwd, cb);

        // Append a second turn end ⇒ exactly one fire.
        tokio::time::sleep(Duration::from_millis(50)).await;
        fs.set(
            &transcript_path("engine-1.jsonl"),
            &format!("{}{}", line("turn_duration"), line("turn_duration")),
        );
        assert!(
            wait_until(|| !log.lock().unwrap().is_empty()).await,
            "a new turn_duration above baseline fires turn_ended"
        );
        assert_eq!(log.lock().unwrap().as_slice(), &[agent(1)]);
    }

    #[tokio::test]
    async fn pre_existing_transcript_never_fires_phantom() {
        let fs = Arc::new(FakeFs::default());
        fs.set(
            &transcript_path("engine-1.jsonl"),
            &format!("{}{}", line("turn_duration"), line("turn_duration")),
        );
        let watcher = ClaudeTranscriptTurnWatcher::new(Arc::clone(&fs) as Arc<dyn FileSystem>, "/home/me");
        let (log, cb) = count_calls();
        let cwd = ProjectPath::new("/run/a").expect("path");
        let _h = watcher.watch(agent(1), None, cwd, cb);

        // No new turn end appended ⇒ no fire, ever (baseline absorbs the existing ones).
        tokio::time::sleep(Duration::from_millis(120)).await;
        assert!(log.lock().unwrap().is_empty(), "baseline must absorb pre-existing turn ends");
    }

    #[tokio::test]
    async fn cold_start_missing_folder_then_first_turn_fires() {
        let fs = Arc::new(FakeFs::default());
        let watcher = ClaudeTranscriptTurnWatcher::new(Arc::clone(&fs) as Arc<dyn FileSystem>, "/home/me");
        let (log, cb) = count_calls();
        let cwd = ProjectPath::new("/run/a").expect("path");
        let _h = watcher.watch(agent(1), None, cwd, cb);

        // Folder/file appear later (cold start), then a first turn end is written.
        tokio::time::sleep(Duration::from_millis(40)).await;
        fs.set(&transcript_path("engine-1.jsonl"), &line("turn_duration"));
        assert!(
            wait_until(|| !log.lock().unwrap().is_empty()).await,
            "a first turn end after a cold start fires (baseline was 0)"
        );
    }

    #[tokio::test]
    async fn dropping_handle_stops_polling() {
        let fs = Arc::new(FakeFs::default());
        let watcher = ClaudeTranscriptTurnWatcher::new(Arc::clone(&fs) as Arc<dyn FileSystem>, "/home/me");
        let (log, cb) = count_calls();
        let cwd = ProjectPath::new("/run/a").expect("path");
        let h = watcher.watch(agent(1), None, cwd, cb);
        drop(h); // stop the watch.

        // A turn end written after the drop must not fire.
        tokio::time::sleep(Duration::from_millis(40)).await;
        fs.set(&transcript_path("engine-1.jsonl"), &line("turn_duration"));
        tokio::time::sleep(Duration::from_millis(120)).await;
        assert!(log.lock().unwrap().is_empty(), "a dropped handle stops firing");
    }
}