hh_core/

adapter.rs

//! Agent adapters: tail a structured event stream and yield [`Event`]s (FR-1.5).
//!
//! An [`Adapter`] owns a tailer thread that produces [`Event`]s on an
//! `mpsc::Receiver`; the recorder (`hh-record`) owns a drain thread that
//! consumes them, resolves the adapter's `correlate_key` to a DB row id in
//! [`Event::correlates`], and appends to the single-writer task. The adapter
//! receives an `Arc<BlobStore>` (for >256 KiB spillover) and a stop flag, but
//! **no `EventWriter`** — keeping it unit-testable without a database and the
//! writer ownership in `hh-record`. A future Codex adapter is a new impl + one
//! new branch in [`select`]; `hh-record` is unchanged.
//!
//! ## Cross-event correlation
//!
//! Adapters cannot know the DB row id an event will receive, so to express
//! "this `tool_result` belongs to that `tool_call`" they emit a string
//! `correlate_key` field inside [`Event::body_json`] (the Claude adapter uses
//! `tool_use.id` for a `tool_call` and `tool_use_id` for a `tool_result`). The
//! recorder's drain thread keeps a `HashMap<String, i64>` (key → event id) for
//! `tool_call`s and, for a `tool_result`, sets `event.correlates =
//! map.get(key)` before appending. A result whose key was never seen (orphan /
//! truly-concurrent result-before-call) gets `correlates = None` and its own
//! step (handled by the FR-3.4 step pass).
//!
//! ## Failure mode (FR-1.5)
//!
//! If the structured log cannot be located (no `~/.claude/projects` dir, or no
//! transcript file appears before the session ends), the adapter returns
//! [`AdapterStatus::Degraded`] carrying a one-line `degrade_reason`. The
//! recorder prints that reason to stderr **after the child exits** — once the
//! terminal is restored — rather than from the tailer thread mid-session, so the
//! warning is not buried under the agent's alternate-screen TUI. PTY + FS
//! recording in the recorder continue unaffected either way.
//!
//! The reason is *specific* so a degrade is a one-line diagnosis, not a mystery:
//! `no jsonl matched cwd slug <slug>` (with the slug dir looked up + candidate
//! counts), `jsonl found (<file>) but 0 records parsed (read N line(s); first
//! parse error at line K: <msg>)`, `found a transcript at <file> but could not
//! read it`, or `discovery selected file <file> but it's a directory`. The
//! recorder also persists it as an `error` event (`body_json.reason`) so it is
//! queryable in the DB, not just printed once. A detailed discovery+parse trace
//! (slug, projects dir, candidate files, selected file, records read, records
//! converted, first conversion failure) is emitted to stderr when the `HH_DEBUG`
//! env var is set — run `HH_DEBUG=1 hh run …` to capture it.

use crate::blob::BlobStore;
use crate::event::{truncate_summary, AdapterStatus, AgentKind, Event, EventKind};
use std::io::BufRead;
use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::mpsc;
use std::thread::JoinHandle;
use std::time::{Duration, Instant};

/// Payloads at or above this size (in bytes) spill to the blob store rather
/// than being stored inline in `events.body_json` (SRS §4.1, matching the
/// recorder's terminal-chunk threshold).
const SPILLOVER_BYTES: usize = 256 * 1024;
/// Tail poll interval when the transcript is at EOF but the session continues.
const TAIL_POLL: Duration = Duration::from_millis(50);
/// Poll interval while waiting for the transcript file to appear.
const APPEAR_POLL: Duration = Duration::from_millis(200);

/// Context handed to an adapter when it is spawned.
#[derive(Clone)]
pub struct AdapterContext {
    /// Owning session id (full UUID string).
    pub session_id: String,
    /// Session start as a unix-ms UTC timestamp; `ts_ms` is relative to this.
    pub started_at_unix_ms: i64,
    /// Session working directory — used to locate/verify the structured log.
    pub cwd: PathBuf,
    /// Full command argv (used for detection and the fallback scan).
    pub command: Vec<String>,
    /// Blob store for >256 KiB spillover.
    pub blobs: std::sync::Arc<BlobStore>,
    /// Stop flag: set by the recorder when the child has exited.
    pub stop: std::sync::Arc<AtomicBool>,
    /// Override for the Claude projects directory (`None` → resolve from `HOME`).
    /// Tests set this to a temp dir so the adapter never touches the real home and
    /// avoids `HOME` env races across parallel tests; production passes `None`.
    pub projects_dir: Option<PathBuf>,
}

/// A handle to a running adapter tailer. Drop the `events` receiver (or drain
/// it to EOF) and [`AdapterHandle::join`] the outcome when the session ends.
pub struct AdapterHandle {
    /// The stream of parsed events, in arrival order. EOF when the tailer exits.
    pub events: mpsc::Receiver<Event>,
    /// Joins the tailer thread and returns the adapter's final outcome.
    pub outcome: JoinHandle<AdapterOutcome>,
}

impl AdapterHandle {
    /// Block until the tailer thread exits and return its outcome. The tailer
    /// is total (never panics on record content), so a panic here indicates an
    /// internal bug, surfaced as [`crate::Error::Storage`] for the binary to
    /// attach `anyhow` context.
    pub fn join(self) -> crate::Result<AdapterOutcome> {
        self.outcome.join().map_err(|_| {
            crate::Error::Storage(crate::error::StorageError::Open {
                // The exact path is irrelevant for an adapter-thread panic; reuse
                // the storage error surface so the binary's print_error renders it.
                path: PathBuf::new(),
                source: std::io::Error::other("adapter thread panicked"),
            })
        })
    }
}

/// The final outcome of an adapter run, reported when its tailer thread exits.
#[derive(Debug, Default, Clone)]
pub struct AdapterOutcome {
    /// Model name extracted from the structured stream, if any (last-seen wins).
    pub model: Option<String>,
    /// Token-usage JSON, verbatim from the last assistant record that carried it
    /// (rich fields preserved; not cumulative). Written to `sessions.usage_json`.
    pub usage_json: Option<serde_json::Value>,
    /// Final adapter status: [`AdapterStatus::Active`] if a transcript was
    /// tailed, [`AdapterStatus::Degraded`] if it could not be located.
    pub status: AdapterStatus,
    /// When `status` is [`AdapterStatus::Degraded`], a single actionable line the
    /// recorder prints to stderr *after the child exits* (once the terminal is
    /// restored), instead of from the tailer thread mid-session — so the warning
    /// is not buried under the agent's alternate-screen TUI (FR-1.5).
    pub degrade_reason: Option<String>,
}

/// An agent adapter: tail a structured event stream and yield [`Event`]s.
///
/// Implementations are `Send + 'static` so the recorder can spawn the tailer on
/// its own thread. [`Adapter::spawn`] performs any setup that can fail (e.g. the
/// tailer thread spawn) and returns an `io::Error` only for that low-level
/// failure; higher-level "no transcript found" conditions are reported via the
/// returned handle's [`AdapterOutcome::status`] (degraded), not as an error, so
/// the recorder can keep recording PTY/FS output (FR-1.5).
pub trait Adapter: Send + 'static {
    /// The agent kind this adapter reports for the session row.
    fn agent_kind(&self) -> AgentKind;
    /// Spawn the tailer thread, returning the event stream + outcome handle.
    ///
    /// Takes `self: Box<Self>` so the recorder can dispatch through the
    /// `Box<dyn Adapter>` returned by [`select`] (a by-value `self` would not
    /// be callable on a trait object). A stateless adapter like [`ClaudeAdapter`]
    /// ignores the receiver.
    fn spawn(self: Box<Self>, ctx: AdapterContext) -> std::io::Result<AdapterHandle>;
}

/// Detect which adapter applies to `command` (FR-1.5). Returns `None` for a
/// generic agent (PTY-only capture). Detection order: Claude Code, Claude
/// Desktop, Codex CLI, Gemini CLI — by the basename of `command[0]`. A forced
/// adapter from `hh run --adapter` is resolved by the recorder, not here.
#[must_use]
pub fn select(command: &[String], _cwd: &Path) -> Option<Box<dyn Adapter>> {
    if is_claude_code(command) {
        Some(Box::new(ClaudeAdapter))
    } else if is_claude_desktop(command) {
        Some(Box::new(ClaudeDesktopAdapter))
    } else if is_codex_cli(command) {
        Some(Box::new(CodexAdapter))
    } else if is_gemini_cli(command) {
        Some(Box::new(GeminiAdapter))
    } else {
        None
    }
}

/// Resolve a forced adapter name (from `hh run --adapter <name>`) to an
/// adapter, or `None` if the name is unrecognized — the recorder surfaces that
/// as an actionable error. Keeping this map in `hh-core` (not the recorder)
/// means adding a forced adapter is a one-branch change here, with no
/// `hh-record` edit (the trait's extensibility goal).
#[must_use]
pub fn resolve_override(name: &str) -> Option<Box<dyn Adapter>> {
    match name {
        "claude-code" => Some(Box::new(ClaudeAdapter)),
        "claude-desktop" => Some(Box::new(ClaudeDesktopAdapter)),
        "codex-cli" => Some(Box::new(CodexAdapter)),
        "gemini-cli" => Some(Box::new(GeminiAdapter)),
        _ => None,
    }
}

/// True if the command's program basename is `claude` (stripping a Windows
/// `.exe`). Mirrors the detection in `hh-record::agent::detect_agent`.
#[must_use]
pub fn is_claude_code(command: &[String]) -> bool {
    let prog = command.first().map_or("", String::as_str);
    // Split on both `/` and `\` so a Windows path like `C:\Apps\claude.exe` yields
    // `claude.exe` on any platform — `Path::file_name` treats `\` as a normal
    // character on Unix, which would mis-basename Windows command lines.
    let base = prog.rsplit(['/', '\\']).next().unwrap_or(prog);
    let base = base.strip_suffix(".exe").unwrap_or(base);
    base.eq_ignore_ascii_case("claude")
}

/// True if the command's program basename is `claude-desktop` (stripping a
/// Windows `.exe`). Claude Desktop uses the same JSONL session format as Claude
/// Code, so the adapter reuses the same tailer.
#[must_use]
pub fn is_claude_desktop(command: &[String]) -> bool {
    let prog = command.first().map_or("", String::as_str);
    let base = prog.rsplit(['/', '\\']).next().unwrap_or(prog);
    let base = base.strip_suffix(".exe").unwrap_or(base);
    base.eq_ignore_ascii_case("claude-desktop")
}

/// True if the command's program basename is `codex` (stripping a Windows
/// `.exe`). Mirrors the detection in `hh-record::agent::detect_agent`.
#[must_use]
pub fn is_codex_cli(command: &[String]) -> bool {
    let prog = command.first().map_or("", String::as_str);
    let base = prog.rsplit(['/', '\\']).next().unwrap_or(prog);
    let base = base.strip_suffix(".exe").unwrap_or(base);
    base.eq_ignore_ascii_case("codex")
}

/// True if the command's program basename is `gemini` (stripping a Windows
/// `.exe`). Mirrors the detection in `hh-record::agent::detect_agent`.
#[must_use]
pub fn is_gemini_cli(command: &[String]) -> bool {
    let prog = command.first().map_or("", String::as_str);
    let base = prog.rsplit(['/', '\\']).next().unwrap_or(prog);
    let base = base.strip_suffix(".exe").unwrap_or(base);
    base.eq_ignore_ascii_case("gemini")
}

// ---------------------------------------------------------------------------
// Claude Code adapter
// ---------------------------------------------------------------------------

/// The Claude Code adapter: tails `~/.claude/projects/<slug>/*.jsonl` and
/// converts records to `user_message` / `agent_message` / `thinking` /
/// `tool_call` / `tool_result` events. Stateless; the tailer thread holds all
/// mutable state. A unit struct (rather than `struct ClaudeAdapter {}`) so it
/// is constructible as a value without a `{}`; reserved for future per-adapter
/// knobs (e.g. a forced projects dir) by adding fields + a `Default` impl.
#[derive(Debug, Clone)]
pub struct ClaudeAdapter;

impl Adapter for ClaudeAdapter {
    fn agent_kind(&self) -> AgentKind {
        AgentKind::ClaudeCode
    }

    #[allow(clippy::unused_self)] // ClaudeAdapter is stateless; the `Box<Self>` receiver exists only so the recorder can dispatch through `Box<dyn Adapter>` (FR-1.5). A future stateful adapter would read its config from `self` here.
    fn spawn(self: Box<Self>, ctx: AdapterContext) -> std::io::Result<AdapterHandle> {
        let (tx, rx) = mpsc::channel::<Event>();
        let outcome = std::thread::Builder::new()
            .name("hh-claude-adapter".into())
            .spawn(move || run_claude_tailer(ctx, tx))?;
        Ok(AdapterHandle {
            events: rx,
            outcome,
        })
    }
}

/// The tailer thread body: locate the transcript, then read it to EOF/stop,
/// parsing each line into events sent on `tx`. Returns the final outcome.
///
/// Degrade paths set [`AdapterOutcome::degrade_reason`] rather than printing
/// mid-session, so the recorder can surface the warning *after* the child exits
/// and the terminal is restored (FR-1.5) — printing from this thread while the
/// agent's TUI owns the alternate screen is invisible to the user.
#[allow(clippy::needless_pass_by_value)] // ctx/tx are owned for the thread's lifetime; taking them by value keeps the tailer self-contained (mirrors runner::run_reader).
fn run_claude_tailer(ctx: AdapterContext, tx: mpsc::Sender<Event>) -> AdapterOutcome {
    let Some(projects) = ctx.projects_dir.clone().or_else(claude_projects_dir) else {
        return degraded(
            "no ~/.claude/projects directory found (HOME unset?); run `hh doctor` to diagnose",
        );
    };
    if !projects.is_dir() {
        return degraded(format!(
            "projects directory does not exist: {}; run `hh doctor` to diagnose",
            projects.display()
        ));
    }
    let slug = slugify(&ctx.cwd);
    let slug_dir = projects.join(&slug);
    debug(&format!(
        "claude adapter: projects_dir={}, slug={slug}, slug_dir={}, session cwd={}, started_at_unix_ms={}",
        projects.display(),
        slug_dir.display(),
        ctx.cwd.display(),
        ctx.started_at_unix_ms
    ));
    let mut diag = DiscoveryDiag {
        projects: Some(projects.clone()),
        slug: Some(slug.clone()),
        slug_dir: Some(slug_dir.clone()),
        cwd: Some(ctx.cwd.clone()),
        ..Default::default()
    };
    let start = Instant::now();
    let Some(file) = locate_transcript(&projects, &ctx.cwd, &ctx.stop, &mut diag) else {
        debug(&format!(
            "claude adapter: discovery failed: new_candidates={}, cwd_mismatches={}, directories={}",
            diag.new_candidates.len(),
            diag.cwd_mismatches.len(),
            diag.directories.len()
        ));
        return degraded(locate_failure_reason(&diag));
    };
    debug(&format!(
        "claude adapter: selected transcript {} (cwd matched)",
        file.display()
    ));
    let mut acc = OutcomeAcc::default();
    let stats = tail_file(&file, &projects, &ctx, &tx, &start, &mut acc);
    if !stats.read_ok {
        return degraded(format!(
            "found a transcript at {} but could not read it (open/read failed); \
             run `hh doctor` to check file permissions",
            file.display(),
        ));
    }
    debug(&format!(
        "claude adapter: tail done: lines_seen={}, records_parsed={}, events_produced={}, first_parse_error={:?}",
        stats.lines_seen, stats.records_parsed, stats.events_produced, stats.first_parse_error
    ));
    if stats.events_produced == 0 {
        // Found and read a transcript, but no user/assistant records converted
        // to events. Distinct from "could not locate" — the file existed but
        // yielded nothing. Most common cause: a format drift the parser no
        // longer recognizes, so surface the first parse failure (if any) and the
        // line count to make this a one-line diagnosis instead of a silent
        // "active, 0 steps".
        let first_err = stats
            .first_parse_error
            .map(|(n, msg)| format!("; first parse error at line {n}: {msg}"))
            .unwrap_or_default();
        return degraded(format!(
            "jsonl found ({}) but 0 records parsed (read {} line(s){first_err}); \
             run `hh doctor` to check the Claude Code transcript format",
            file.display(),
            stats.lines_seen,
        ));
    }
    acc.finish(AdapterStatus::Active)
}

/// Build a [`Degraded`](AdapterStatus::Degraded) outcome carrying a one-line
/// reason for the recorder to print after the child exits.
fn degraded(reason: impl Into<String>) -> AdapterOutcome {
    AdapterOutcome {
        status: AdapterStatus::Degraded,
        degrade_reason: Some(reason.into()),
        ..Default::default()
    }
}

/// Accumulated discovery diagnostics: what the tailer saw while polling for a
/// transcript. Used to (a) emit an `HH_DEBUG` trace and (b) build a *specific*
/// degrade reason when no transcript qualifies — so a degraded session
/// self-documents "no new file appeared" vs "N candidates appeared but none
/// matched cwd" vs "a candidate was a directory", instead of a bare "could not
/// locate". Pure data; the tailer mutates it as it polls.
#[derive(Default)]
struct DiscoveryDiag {
    /// Resolved projects dir (`~/.claude/projects` or the test override).
    projects: Option<PathBuf>,
    /// The slug computed from the session cwd (e.g. `-home-saadman-halfhand`).
    slug: Option<String>,
    /// `projects/<slug>` — the primary lookup dir.
    slug_dir: Option<PathBuf>,
    /// Session cwd, for the reason string.
    cwd: Option<PathBuf>,
    /// Distinct NEW `*.jsonl` files ever observed (across all polls). Empty →
    /// no transcript appeared at all; non-empty → appeared but none qualified.
    new_candidates: std::collections::HashSet<PathBuf>,
    /// Candidates rejected because their first cwd-bearing record named a
    /// different cwd (a concurrent session in the same project dir).
    cwd_mismatches: Vec<PathBuf>,
    /// Candidates that were a directory, not a file (the same-stem-dir guard).
    directories: Vec<PathBuf>,
}

/// Statistics from tailing one transcript, used to build a specific degrade
/// reason when a file was found and read but produced no events. Pure data;
/// [`tail_file`] / [`parse_and_send`] mutate it as they read.
#[derive(Default)]
struct TailStats {
    /// `false` if the transcript could not be opened at all.
    read_ok: bool,
    /// Non-empty lines seen (complete records + unparseable lines).
    lines_seen: usize,
    /// Lines that parsed as a JSON object (the "records read" count).
    records_parsed: usize,
    /// Total events emitted to the drain channel.
    events_produced: usize,
    /// The first line that failed JSON parsing, as `(1-based line number, msg)`.
    /// `None` if every line parsed (or no lines were seen).
    first_parse_error: Option<(usize, String)>,
}

/// Build a specific degrade reason from a failed discovery. Distinguishes "no
/// new transcript appeared" (lazy creation + empty session) from "candidates
/// appeared but none matched cwd" (concurrent-session / slug drift), and calls
/// out any directory candidates (the same-stem-dir issue).
fn locate_failure_reason(diag: &DiscoveryDiag) -> String {
    use std::fmt::Write as _;
    let slug = diag.slug.as_deref().unwrap_or("?");
    let projects = diag
        .projects
        .as_ref()
        .map_or_else(|| "(unknown)".to_string(), |p| p.display().to_string());
    let cwd = diag
        .cwd
        .as_ref()
        .map_or_else(|| "(unknown)".to_string(), |p| p.display().to_string());
    let slug_dir = diag
        .slug_dir
        .as_ref()
        .map_or_else(|| "(unknown)".to_string(), |p| p.display().to_string());
    if diag.new_candidates.is_empty() {
        format!(
            "no jsonl matched cwd slug {slug}: no new transcript appeared under {projects} \
             (looked in {slug_dir}; cwd={cwd}) before the session ended; Claude creates it \
             lazily on first output — if you sent no prompt this is expected. \
             Run `hh doctor` to verify discoverability."
        )
    } else {
        let mut s = format!(
            "no jsonl matched cwd slug {slug}: {} new candidate(s) appeared under {projects} \
             but none carried cwd={cwd} before the session ended",
            diag.new_candidates.len(),
        );
        if !diag.cwd_mismatches.is_empty() {
            let names: Vec<String> = diag
                .cwd_mismatches
                .iter()
                .take(3)
                .map(|p| {
                    p.file_name()
                        .map(|f| f.to_string_lossy().to_string())
                        .unwrap_or_default()
                })
                .collect();
            let _ = write!(
                s,
                "; {} rejected for a different cwd ({})",
                diag.cwd_mismatches.len(),
                names.join(", ")
            );
        }
        if !diag.directories.is_empty() {
            let _ = write!(
                s,
                "; {} candidate(s) were directories, not files",
                diag.directories.len()
            );
        }
        s.push_str(". Run `hh doctor` to diagnose.");
        s
    }
}

/// Accumulates model/usage across assistant records while tailing.
#[derive(Default)]
struct OutcomeAcc {
    model: Option<String>,
    usage_json: Option<serde_json::Value>,
}

impl OutcomeAcc {
    fn note_assistant(&mut self, message: &serde_json::Value) {
        if let Some(m) = message.get("model").and_then(|v| v.as_str()) {
            self.model = Some(m.to_string());
        }
        if let Some(u) = message.get("usage") {
            self.usage_json = Some(u.clone());
        }
    }

    fn finish(self, status: AdapterStatus) -> AdapterOutcome {
        AdapterOutcome {
            model: self.model,
            usage_json: self.usage_json,
            status,
            degrade_reason: None,
        }
    }
}

/// Locate the transcript file for this session.
///
/// Selection rule (fixes two real-world failures):
/// - **Poll until the stop flag is set**, not for a fixed 3 s deadline. Claude
///   Code creates its transcript *lazily* on the first user message, often well
///   after `hh run` launches the child; a fixed timeout degraded every
///   short-prompt session that took >3 s to send its first message (the
///   "0 steps, status=ok" symptom).
/// - **Prefer a transcript that is NEW since session start**, identified by
///   snapshotting every `*.jsonl` under `projects` at entry. Claude names each
///   invocation's transcript by a fresh uuid, so the file for *this* session did
///   not exist when `hh` started; matching only new files avoids latching onto a
///   concurrent session's transcript in the same project dir (the
///   "active but tailing the wrong session" symptom). Set-based, not mtime-based,
///   so clock skew cannot fool it.
/// - **Verify by cwd**: a candidate qualifies only once a record carrying `cwd`
///   appears and that cwd equals (or is under) the session cwd. Early Claude
///   records (`agent-setting`/`mode`/`file-history-snapshot`) carry no `cwd`, so
///   a freshly-created file is polled until its first cwd-bearing record lands.
///
/// Returns `None` if nothing qualifies before the stop flag is set (the recorder
/// then surfaces a degraded outcome with an actionable reason).
fn locate_transcript(
    projects: &Path,
    cwd: &Path,
    stop: &std::sync::Arc<AtomicBool>,
    diag: &mut DiscoveryDiag,
) -> Option<PathBuf> {
    let slug_dir = projects.join(slugify(cwd));
    let preexisting = snapshot_all_jsonl(projects);
    // Candidates confirmed to belong to a *different* cwd (cwd present but not
    // ours) — never re-checked. Files with no cwd yet are re-polled (they grow).
    let mut rejected: std::collections::HashSet<PathBuf> = std::collections::HashSet::new();

    loop {
        if stop.load(Ordering::Acquire) {
            return None;
        }
        // Primary: a NEW transcript in the expected slug dir whose cwd matches.
        if slug_dir.is_dir() {
            for f in new_jsonl_in(&slug_dir, &preexisting) {
                if rejected.contains(&f) {
                    continue;
                }
                diag.new_candidates.insert(f.clone());
                match first_record_cwd_state(&f, cwd) {
                    CwdState::Matches => return Some(f),
                    CwdState::Different => {
                        diag.cwd_mismatches.push(f.clone());
                        rejected.insert(f);
                    }
                    CwdState::Directory => {
                        diag.directories.push(f.clone());
                        rejected.insert(f);
                    }
                    CwdState::None => {} // not yet written; keep polling
                }
            }
        }
        // Fallback: slug encoding may differ from Claude's; scan every project dir
        // for a NEW transcript matching cwd. Cheap in practice — new files appear
        // only when a session starts, and rejected/non-matching ones are cached.
        for f in new_jsonl_under(projects, &preexisting) {
            if rejected.contains(&f) {
                continue;
            }
            diag.new_candidates.insert(f.clone());
            match first_record_cwd_state(&f, cwd) {
                CwdState::Matches => return Some(f),
                CwdState::Different => {
                    diag.cwd_mismatches.push(f.clone());
                    rejected.insert(f);
                }
                CwdState::Directory => {
                    diag.directories.push(f.clone());
                    rejected.insert(f);
                }
                CwdState::None => {}
            }
        }
        std::thread::sleep(APPEAR_POLL);
    }
}

/// The newest `*.jsonl` in `dir` by mtime, with a small slack window so a file
/// created just before `started_at_unix_ms` (clock skew) still qualifies.
fn newest_in_dir(dir: &Path, since_unix_ms: i64) -> Option<PathBuf> {
    let entries = std::fs::read_dir(dir).ok()?;
    let mut best: Option<(PathBuf, i64)> = None;
    for e in entries.flatten() {
        let p = e.path();
        if p.extension().and_then(|x| x.to_str()) != Some("jsonl") {
            continue;
        }
        let Ok(m) = e.metadata() else { continue };
        let Ok(modified) = m.modified() else { continue };
        let mtime_ms = system_time_to_unix_ms(modified);
        if mtime_ms < since_unix_ms - 5_000 {
            continue; // 5s slack
        }
        match &best {
            Some((_, best_ms)) if mtime_ms <= *best_ms => {}
            _ => best = Some((p, mtime_ms)),
        }
    }
    best.map(|(p, _)| p)
}

/// Snapshot every `*.jsonl` path under all project dirs at session start. The
/// transcript for *this* invocation is a file NOT in this set (Claude creates a
/// fresh uuid-named file per run), so this set defines "new since start".
fn snapshot_all_jsonl(projects: &Path) -> std::collections::HashSet<PathBuf> {
    let mut set = std::collections::HashSet::new();
    let Ok(dirs) = std::fs::read_dir(projects) else {
        return set;
    };
    for d in dirs.flatten() {
        let dp = d.path();
        if !dp.is_dir() {
            continue;
        }
        if let Ok(entries) = std::fs::read_dir(&dp) {
            for e in entries.flatten() {
                let p = e.path();
                if p.extension().and_then(|x| x.to_str()) == Some("jsonl") {
                    set.insert(p);
                }
            }
        }
    }
    set
}

/// `*.jsonl` files in `dir` not in `preexisting`, newest-first by mtime.
fn new_jsonl_in(dir: &Path, preexisting: &std::collections::HashSet<PathBuf>) -> Vec<PathBuf> {
    let Ok(entries) = std::fs::read_dir(dir) else {
        return Vec::new();
    };
    let mut found: Vec<(PathBuf, i64)> = Vec::new();
    for e in entries.flatten() {
        let p = e.path();
        if p.extension().and_then(|x| x.to_str()) != Some("jsonl") {
            continue;
        }
        if preexisting.contains(&p) {
            continue;
        }
        let Ok(m) = e.metadata() else { continue };
        let mtime_ms = m.modified().map_or(0, system_time_to_unix_ms);
        found.push((p, mtime_ms));
    }
    found.sort_by_key(|(_, m)| std::cmp::Reverse(*m));
    found.into_iter().map(|(p, _)| p).collect()
}

/// `*.jsonl` files anywhere under `projects` not in `preexisting`, newest-first.
fn new_jsonl_under(
    projects: &Path,
    preexisting: &std::collections::HashSet<PathBuf>,
) -> Vec<PathBuf> {
    let Ok(dirs) = std::fs::read_dir(projects) else {
        return Vec::new();
    };
    let mut found: Vec<(PathBuf, i64)> = Vec::new();
    for d in dirs.flatten() {
        let dp = d.path();
        if !dp.is_dir() {
            continue;
        }
        for p in new_jsonl_in(&dp, preexisting) {
            let mtime_ms = std::fs::metadata(&p)
                .ok()
                .and_then(|m| m.modified().ok())
                .map_or(0, system_time_to_unix_ms);
            found.push((p, mtime_ms));
        }
    }
    found.sort_by_key(|(_, m)| std::cmp::Reverse(*m));
    found.into_iter().map(|(p, _)| p).collect()
}

/// The cwd state of a candidate transcript, from its first cwd-bearing record.
#[derive(Debug, PartialEq, Eq)]
enum CwdState {
    /// A record carried `cwd` and it equals (or is under) the session cwd.
    Matches,
    /// A record carried `cwd` but it did not match — belongs to another session.
    Different,
    /// No cwd-bearing record yet (file is brand-new / still being written).
    None,
    /// The candidate path is a directory, not a file. A defensive guard: the
    /// `.jsonl` extension filter in [`new_jsonl_in`]/[`new_jsonl_under`] already
    /// excludes directories (they carry no extension), so this normally never
    /// fires — but if it ever does (e.g. a same-stem dir named `X.jsonl`), the
    /// candidate is rejected with a specific reason rather than re-polled forever
    /// (the `read_to_string` on a directory would otherwise yield an io error →
    /// [`CwdState::None`] → an infinite poll loop, the silent-degrade symptom).
    Directory,
}

/// Inspect the first cwd-bearing record in `file` and classify it against the
/// session cwd. See [`CwdState`]. Best-effort read: an unreadable file is
/// [`CwdState::None`] (re-polled rather than rejected). A directory candidate is
/// [`CwdState::Directory`] (rejected, not re-polled).
fn first_record_cwd_state(file: &Path, session_cwd: &Path) -> CwdState {
    if file.is_dir() {
        return CwdState::Directory;
    }
    let Ok(content) = std::fs::read_to_string(file) else {
        return CwdState::None;
    };
    for line in content.lines() {
        let line = line.trim();
        if line.is_empty() {
            continue;
        }
        let Ok(v) = serde_json::from_str::<serde_json::Value>(line) else {
            continue;
        };
        if let Some(rc) = v.get("cwd").and_then(|c| c.as_str()) {
            let rp = Path::new(rc);
            return if rp == session_cwd || rp.starts_with(session_cwd) {
                CwdState::Matches
            } else {
                CwdState::Different
            };
        }
    }
    CwdState::None
}

/// Read `file` from byte 0, parsing each complete line into events sent on `tx`.
/// Handles partial last lines (buffered until a newline arrives), rotation to a
/// newer `*.jsonl` in the slug dir, and stop-flag termination. Returns
/// [`TailStats`] so the caller can distinguish "could not open" (`read_ok`
/// false) from "opened but produced 0 events" (a format-drift degrade, with the
/// first parse error + line count) — instead of claiming active with 0 events.
fn tail_file(
    file: &Path,
    projects: &Path,
    ctx: &AdapterContext,
    tx: &mpsc::Sender<Event>,
    start: &Instant,
    acc: &mut OutcomeAcc,
) -> TailStats {
    let slug_dir = projects.join(slugify(&ctx.cwd));
    let mut current = file.to_path_buf();
    let mut stats = TailStats::default();
    let mut reader = match std::fs::File::open(&current) {
        Ok(f) => std::io::BufReader::new(f),
        Err(_) => return stats, // read_ok stays false
    };
    stats.read_ok = true;
    let mut buf: Vec<u8> = Vec::new();
    let mut line = Vec::new();
    loop {
        if ctx.stop.load(Ordering::Acquire) {
            break;
        }
        match reader.read_until(b'\n', &mut line) {
            Ok(0) => {
                // EOF: parse any *complete* lines buffered so far, keep the
                // trailing partial (no newline yet) for the next grow.
                drain_complete_lines(&mut buf, &mut line, ctx, tx, start, acc, &mut stats);
                // Rotation: a newer .jsonl appeared in the slug dir?
                if let Some(newer) = rotate_to_newer(&slug_dir, &current, ctx.started_at_unix_ms) {
                    debug(&format!(
                        "claude adapter: rotating to newer transcript {}",
                        newer.display()
                    ));
                    current = newer;
                    reader = match std::fs::File::open(&current) {
                        Ok(f) => std::io::BufReader::new(f),
                        Err(_) => break,
                    };
                    line.clear();
                    continue;
                }
                if ctx.stop.load(Ordering::Acquire) {
                    break;
                }
                std::thread::sleep(TAIL_POLL);
            }
            Ok(_n) => {
                drain_complete_lines(&mut buf, &mut line, ctx, tx, start, acc, &mut stats);
            }
            Err(e) if e.kind() == std::io::ErrorKind::Interrupted => {}
            Err(_) => break,
        }
    }
    // Final flush of any complete trailing line on stop.
    drain_complete_lines(&mut buf, &mut line, ctx, tx, start, acc, &mut stats);
    stats
}

/// Move the just-read `line` bytes into `buf`, then parse every complete
/// (`\n`-terminated) line in `buf`, leaving the trailing partial in `buf`.
fn drain_complete_lines(
    buf: &mut Vec<u8>,
    line: &mut Vec<u8>,
    ctx: &AdapterContext,
    tx: &mpsc::Sender<Event>,
    start: &Instant,
    acc: &mut OutcomeAcc,
    stats: &mut TailStats,
) {
    if line.is_empty() {
        return;
    }
    buf.append(line);
    line.clear();
    // Find the last `\n`; everything up to and including it is complete.
    while let Some(nl) = buf.iter().rposition(|&b| b == b'\n') {
        let complete: Vec<u8> = buf.drain(..=nl).collect();
        parse_and_send(&complete, ctx, tx, start, acc, stats);
    }
}

/// Parse one buffered line (bytes) into events and send them. Malformed JSON or
/// skipped record types produce no events; the first JSON parse failure is
/// recorded in `stats` (line number + message) for the degrade reason, and a
/// debug trace is gated on `HH_DEBUG`. Unknown record types are skipped with a
/// debug log, never a degrade (tolerant, multi-generation).
fn parse_and_send(
    bytes: &[u8],
    ctx: &AdapterContext,
    tx: &mpsc::Sender<Event>,
    start: &Instant,
    acc: &mut OutcomeAcc,
    stats: &mut TailStats,
) {
    let s = match std::str::from_utf8(bytes) {
        Ok(s) => s.trim(),
        Err(e) => {
            stats.lines_seen += 1;
            if stats.first_parse_error.is_none() {
                stats.first_parse_error = Some((stats.lines_seen, format!("not UTF-8: {e}")));
            }
            debug(&format!(
                "claude adapter: line {} not UTF-8, skipping: {e}",
                stats.lines_seen
            ));
            return;
        }
    };
    if s.is_empty() {
        return;
    }
    stats.lines_seen += 1;
    let value: serde_json::Value = match serde_json::from_str(s) {
        Ok(v) => v,
        Err(e) => {
            if stats.first_parse_error.is_none() {
                stats.first_parse_error = Some((stats.lines_seen, e.to_string()));
            }
            debug(&format!(
                "claude adapter: line {} not valid JSON, skipping: {e}",
                stats.lines_seen
            ));
            return;
        }
    };
    stats.records_parsed += 1;
    let ts_ms = value
        .get("timestamp")
        .and_then(|v| v.as_str())
        .and_then(|s| ts_ms_from_iso(s, ctx.started_at_unix_ms))
        .unwrap_or_else(|| elapsed_ms(start));
    let parsed = parse_record(&value, &ctx.session_id, ts_ms, &ctx.blobs);
    if parsed.is_assistant_message {
        if let Some(msg) = value.get("message") {
            acc.note_assistant(msg);
        }
    }
    for ev in parsed.events {
        if tx.send(ev).is_err() {
            break; // recorder dropped the receiver: stop sending.
        }
        stats.events_produced += 1;
    }
}

/// If a `*.jsonl` newer than `current` exists in the slug dir, return it.
fn rotate_to_newer(slug_dir: &Path, current: &Path, started_at_unix_ms: i64) -> Option<PathBuf> {
    let newest = newest_in_dir(slug_dir, started_at_unix_ms)?;
    if newest == current {
        return None;
    }
    // Only rotate if the candidate is actually newer than the current file.
    let newer_mtime = std::fs::metadata(&newest)
        .ok()
        .and_then(|m| m.modified().ok())
        .map(system_time_to_unix_ms);
    let cur_mtime = std::fs::metadata(current)
        .ok()
        .and_then(|m| m.modified().ok())
        .map(system_time_to_unix_ms);
    match (newer_mtime, cur_mtime) {
        (Some(n), Some(c)) if n > c => Some(newest),
        _ => None,
    }
}

// ---------------------------------------------------------------------------
// Claude Desktop adapter
// ---------------------------------------------------------------------------

/// The Claude Desktop adapter: same JSONL format as Claude Code
/// (`~/.claude/projects/<slug>/*.jsonl`), so it reuses the same tailer and
/// parser. The only difference is the `AgentKind` reported to the session row
/// (`claude-desktop` vs `claude-code`), so `hh list` distinguishes them.
#[derive(Debug, Clone)]
pub struct ClaudeDesktopAdapter;

impl Adapter for ClaudeDesktopAdapter {
    fn agent_kind(&self) -> AgentKind {
        AgentKind::ClaudeDesktop
    }

    #[allow(clippy::unused_self)]
    fn spawn(self: Box<Self>, ctx: AdapterContext) -> std::io::Result<AdapterHandle> {
        let (tx, rx) = mpsc::channel::<Event>();
        let outcome = std::thread::Builder::new()
            .name("hh-claude-desktop-adapter".into())
            .spawn(move || run_claude_tailer(ctx, tx))?;
        Ok(AdapterHandle {
            events: rx,
            outcome,
        })
    }
}

// ---------------------------------------------------------------------------
// Codex CLI adapter
// ---------------------------------------------------------------------------

/// The Codex CLI adapter: tails `~/.codex/sessions/YYYY/MM/DD/rollout-*.jsonl`
/// and converts records to `user_message` / `agent_message` / `thinking` /
/// `tool_call` / `tool_result` events. Stateless; the tailer thread holds all
/// mutable state.
///
/// ## Experimental
///
/// This adapter is built against the researched Codex CLI rollout format
/// (<https://github.com/openai/codex>) without real captured fixture files.
/// The parser handles the documented record types (`session_meta`,
/// `turn_context`, `response_item`, `event_msg`) and degrades gracefully on
/// unknown types. If you encounter a session that produces 0 steps despite
/// having a transcript, run `HH_DEBUG=1 hh run -- codex ...` and file an issue
/// with the debug output.
#[derive(Debug, Clone)]
pub struct CodexAdapter;

impl Adapter for CodexAdapter {
    fn agent_kind(&self) -> AgentKind {
        AgentKind::CodexCli
    }

    #[allow(clippy::unused_self)]
    fn spawn(self: Box<Self>, ctx: AdapterContext) -> std::io::Result<AdapterHandle> {
        let (tx, rx) = mpsc::channel::<Event>();
        let outcome = std::thread::Builder::new()
            .name("hh-codex-adapter".into())
            .spawn(move || run_codex_tailer(ctx, tx))?;
        Ok(AdapterHandle {
            events: rx,
            outcome,
        })
    }
}

/// The Codex CLI sessions directory: `$HOME/.codex/sessions` (Unix) or
/// `%USERPROFILE%\.codex\sessions` (Windows). `None` if neither env var is set.
#[must_use]
pub fn codex_sessions_dir() -> Option<PathBuf> {
    let home = std::env::var_os("HOME").or_else(|| std::env::var_os("USERPROFILE"))?;
    Some(PathBuf::from(home).join(".codex").join("sessions"))
}

/// The tailer thread body for Codex CLI: locate the rollout transcript, then
/// read it to EOF/stop, parsing each line into events sent on `tx`.
#[allow(clippy::needless_pass_by_value)] // ctx/tx are owned for the thread's lifetime
fn run_codex_tailer(ctx: AdapterContext, tx: mpsc::Sender<Event>) -> AdapterOutcome {
    let Some(sessions) = ctx.projects_dir.clone().or_else(codex_sessions_dir) else {
        return degraded(
            "no ~/.codex/sessions directory found (HOME unset?); run `hh doctor` to diagnose",
        );
    };
    if !sessions.is_dir() {
        return degraded(format!(
            "sessions directory does not exist: {}; run `hh doctor` to diagnose",
            sessions.display()
        ));
    }
    let start = Instant::now();
    let Some(file) = locate_codex_transcript(&sessions, &ctx.cwd, &ctx.stop) else {
        return degraded(format!(
            "no codex rollout transcript appeared under {} matching cwd {} before the session ended",
            sessions.display(),
            ctx.cwd.display(),
        ));
    };
    let mut acc = OutcomeAcc::default();
    let stats = tail_codex_file(&file, &ctx, &tx, &start, &mut acc);
    if !stats.read_ok {
        return degraded(format!(
            "found a codex transcript at {} but could not read it; \
             run `hh doctor` to check file permissions",
            file.display(),
        ));
    }
    if stats.events_produced == 0 {
        let first_err = stats
            .first_parse_error
            .map(|(n, msg)| format!("; first parse error at line {n}: {msg}"))
            .unwrap_or_default();
        return degraded(format!(
            "codex transcript found ({}) but 0 records parsed (read {} line(s){first_err}); \
             run `hh doctor` to check the Codex CLI transcript format",
            file.display(),
            stats.lines_seen,
        ));
    }
    acc.finish(AdapterStatus::Active)
}

/// Locate the Codex CLI rollout transcript for this session.
///
/// Codex stores rollout files at `~/.codex/sessions/YYYY/MM/DD/rollout-*.jsonl`.
/// The adapter polls for new files in today's date directory (and yesterday's,
/// for sessions that cross midnight), snapshotting pre-existing files at start
/// so it only picks up the file created for *this* session.
fn locate_codex_transcript(
    sessions: &Path,
    cwd: &Path,
    stop: &std::sync::Arc<AtomicBool>,
) -> Option<PathBuf> {
    let preexisting = snapshot_codex_rollouts(sessions);
    let mut rejected: std::collections::HashSet<PathBuf> = std::collections::HashSet::new();

    loop {
        if stop.load(Ordering::Acquire) {
            return None;
        }
        // Scan today's and yesterday's date directories for new rollout files.
        for date_dir in candidate_date_dirs() {
            let dir = sessions.join(&date_dir);
            if !dir.is_dir() {
                continue;
            }
            for f in new_rollout_in(&dir, &preexisting) {
                if rejected.contains(&f) {
                    continue;
                }
                match first_codex_record_cwd(&f, cwd) {
                    CwdState::Matches => return Some(f),
                    CwdState::Different | CwdState::Directory => {
                        rejected.insert(f);
                    }
                    CwdState::None => {} // not yet written; keep polling
                }
            }
        }
        std::thread::sleep(APPEAR_POLL);
    }
}

/// Snapshot every rollout file under the sessions directory at start.
fn snapshot_codex_rollouts(sessions: &Path) -> std::collections::HashSet<PathBuf> {
    let mut set = std::collections::HashSet::new();
    let Ok(years) = std::fs::read_dir(sessions) else {
        return set;
    };
    for year in years.flatten() {
        let yp = year.path();
        if !yp.is_dir() {
            continue;
        }
        let Ok(months) = std::fs::read_dir(&yp) else {
            continue;
        };
        for month in months.flatten() {
            let mp = month.path();
            if !mp.is_dir() {
                continue;
            }
            let Ok(days) = std::fs::read_dir(&mp) else {
                continue;
            };
            for day in days.flatten() {
                let dp = day.path();
                if !dp.is_dir() {
                    continue;
                }
                if let Ok(entries) = std::fs::read_dir(&dp) {
                    for e in entries.flatten() {
                        let p = e.path();
                        if p.file_name().and_then(|n| n.to_str()).is_some_and(|n| {
                            n.starts_with("rollout-")
                                && std::path::Path::new(n).extension()
                                    == Some(std::ffi::OsStr::new("jsonl"))
                        }) {
                            set.insert(p);
                        }
                    }
                }
            }
        }
    }
    set
}

/// `rollout-*.jsonl` files in `dir` not in `preexisting`, newest-first by mtime.
fn new_rollout_in(dir: &Path, preexisting: &std::collections::HashSet<PathBuf>) -> Vec<PathBuf> {
    let Ok(entries) = std::fs::read_dir(dir) else {
        return Vec::new();
    };
    let mut found: Vec<(PathBuf, i64)> = Vec::new();
    for e in entries.flatten() {
        let p = e.path();
        if !p.file_name().and_then(|n| n.to_str()).is_some_and(|n| {
            n.starts_with("rollout-")
                && std::path::Path::new(n).extension() == Some(std::ffi::OsStr::new("jsonl"))
        }) {
            continue;
        }
        if preexisting.contains(&p) {
            continue;
        }
        let mtime_ms = e
            .metadata()
            .ok()
            .and_then(|m| m.modified().ok())
            .map_or(0, system_time_to_unix_ms);
        found.push((p, mtime_ms));
    }
    found.sort_by_key(|(_, m)| std::cmp::Reverse(*m));
    found.into_iter().map(|(p, _)| p).collect()
}

/// Candidate date directories: today and yesterday (for sessions that cross
/// midnight), formatted as `YYYY/MM/DD`.
fn candidate_date_dirs() -> Vec<String> {
    let now = std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .map_or(0, |d| i64::try_from(d.as_secs()).unwrap_or(i64::MAX));
    let today_secs = now - (now % 86400);
    let mut dirs = Vec::with_capacity(2);
    for offset in [0, -86400] {
        let ts = today_secs + offset;
        if let Ok(dt) = time::OffsetDateTime::from_unix_timestamp(ts) {
            dirs.push(format!(
                "{:04}/{:02}/{:02}",
                dt.year(),
                u8::from(dt.month()),
                dt.day()
            ));
        }
    }
    dirs
}

/// Inspect the first cwd-bearing record in a Codex rollout file. Codex puts
/// `cwd` in the `session_meta` record (first line of the file).
fn first_codex_record_cwd(file: &Path, session_cwd: &Path) -> CwdState {
    if file.is_dir() {
        return CwdState::Directory;
    }
    let Ok(content) = std::fs::read_to_string(file) else {
        return CwdState::None;
    };
    for line in content.lines() {
        let line = line.trim();
        if line.is_empty() {
            continue;
        }
        let Ok(v) = serde_json::from_str::<serde_json::Value>(line) else {
            continue;
        };
        // Codex puts cwd in session_meta.payload.cwd
        if let Some(cwd_val) = v
            .get("payload")
            .and_then(|p| p.get("cwd"))
            .and_then(|c| c.as_str())
        {
            let rp = Path::new(cwd_val);
            return if rp == session_cwd || rp.starts_with(session_cwd) {
                CwdState::Matches
            } else {
                CwdState::Different
            };
        }
    }
    CwdState::None
}

/// Tail a Codex rollout file, parsing each line into events.
fn tail_codex_file(
    file: &Path,
    ctx: &AdapterContext,
    tx: &mpsc::Sender<Event>,
    start: &Instant,
    acc: &mut OutcomeAcc,
) -> TailStats {
    let mut stats = TailStats::default();
    let mut reader = match std::fs::File::open(file) {
        Ok(f) => std::io::BufReader::new(f),
        Err(_) => return stats,
    };
    stats.read_ok = true;
    let mut buf: Vec<u8> = Vec::new();
    let mut line = Vec::new();
    loop {
        if ctx.stop.load(Ordering::Acquire) {
            break;
        }
        match reader.read_until(b'\n', &mut line) {
            Ok(0) => {
                drain_complete_lines(&mut buf, &mut line, ctx, tx, start, acc, &mut stats);
                if ctx.stop.load(Ordering::Acquire) {
                    break;
                }
                std::thread::sleep(TAIL_POLL);
            }
            Ok(_n) => {
                drain_complete_lines(&mut buf, &mut line, ctx, tx, start, acc, &mut stats);
            }
            Err(e) if e.kind() == std::io::ErrorKind::Interrupted => {}
            Err(_) => break,
        }
    }
    drain_complete_lines(&mut buf, &mut line, ctx, tx, start, acc, &mut stats);
    stats
}

// ---------------------------------------------------------------------------
// Codex CLI record parser
// ---------------------------------------------------------------------------

/// Parse one Codex CLI JSONL record into events. Pure given `(value, ts_ms,
/// blobs)`: deterministic output for fixed inputs. Unknown `type`s and
/// unexpected shapes produce no events (tolerant).
///
/// Codex rollout format: each line is `{"type": "<variant>", "timestamp": "...",
/// "payload": {...}}`. The payload's inner `type` discriminates further for
/// `response_item` and `event_msg` variants.
#[must_use]
#[allow(dead_code)] // called from the tailer thread; not yet wired through parse_and_send
pub(crate) fn parse_codex_record(
    value: &serde_json::Value,
    session_id: &str,
    ts_ms: i64,
    blobs: &BlobStore,
) -> ParsedRecord {
    let mut out = ParsedRecord::default();
    let Some(obj) = value.as_object() else {
        return out;
    };
    let ty = obj.get("type").and_then(|v| v.as_str()).unwrap_or("");
    let payload = obj.get("payload");
    match ty {
        "response_item" => {
            if let Some(p) = payload {
                parse_codex_response_item(p, session_id, ts_ms, blobs, &mut out);
            }
        }
        "event_msg" => {
            if let Some(p) = payload {
                parse_codex_event_msg(p, session_id, ts_ms, blobs, &mut out);
            }
        }
        // session_meta, turn_context, session_state, compacted: skip
        _ => {}
    }
    out
}

/// Parse a `response_item` payload into events.
#[allow(dead_code, clippy::too_many_lines)]
fn parse_codex_response_item(
    payload: &serde_json::Value,
    session_id: &str,
    ts_ms: i64,
    blobs: &BlobStore,
    out: &mut ParsedRecord,
) {
    let Some(obj) = payload.as_object() else {
        return;
    };
    let inner_ty = obj.get("type").and_then(|v| v.as_str()).unwrap_or("");
    match inner_ty {
        "message" => {
            let role = obj.get("role").and_then(|v| v.as_str()).unwrap_or("");
            let content = obj.get("content").and_then(|v| v.as_array());
            let text = content.and_then(|arr| {
                arr.iter()
                    .filter_map(|b| b.get("text").and_then(|t| t.as_str()))
                    .collect::<Vec<_>>()
                    .join("\n")
                    .into()
            });
            let kind = match role {
                "assistant" => EventKind::AgentMessage,
                _ => EventKind::UserMessage,
            };
            if let Some(t) = text {
                out.events
                    .push(text_event(session_id, ts_ms, kind, &t, blobs));
            }
            out.is_assistant_message = role == "assistant";
        }
        "reasoning" => {
            // Reasoning content is usually encrypted; emit a placeholder.
            let summary = obj
                .get("summary")
                .and_then(|v| v.as_array())
                .map(|arr| {
                    arr.iter()
                        .filter_map(|b| b.get("text").and_then(|t| t.as_str()))
                        .collect::<Vec<_>>()
                        .join("\n")
                })
                .unwrap_or_default();
            let text = if summary.is_empty() {
                "reasoning (encrypted)".to_string()
            } else {
                summary
            };
            out.events.push(text_event(
                session_id,
                ts_ms,
                EventKind::Thinking,
                &text,
                blobs,
            ));
        }
        "function_call" | "custom_tool_call" => {
            let call_id = obj.get("call_id").and_then(|v| v.as_str()).unwrap_or("");
            let name = obj.get("name").and_then(|v| v.as_str()).unwrap_or("");
            let input = if inner_ty == "function_call" {
                // function_call.arguments is a JSON string — double-parse
                obj.get("arguments")
                    .and_then(|v| v.as_str())
                    .and_then(|s| serde_json::from_str(s).ok())
                    .unwrap_or(serde_json::Value::Null)
            } else {
                // custom_tool_call.input is free-form text
                obj.get("input").cloned().unwrap_or(serde_json::Value::Null)
            };
            let body = serde_json::json!({
                "name": name,
                "input": input,
                "correlate_key": call_id,
            });
            let (body_json, blob_hash, blob_size) = maybe_spill(body, blobs);
            out.events.push(Event {
                session_id: session_id.to_string(),
                ts_ms,
                kind: EventKind::ToolCall,
                step: None,
                summary: truncate_summary(&format!("tool_call: {name}")),
                body_json,
                blob_hash,
                blob_size,
                correlates: None,
            });
        }
        "function_call_output" | "custom_tool_call_output" => {
            let call_id = obj.get("call_id").and_then(|v| v.as_str()).unwrap_or("");
            let output = obj.get("output").and_then(|v| v.as_str()).unwrap_or("");
            let body = serde_json::json!({
                "tool_use_id": call_id,
                "is_error": false,
                "content": output,
                "correlate_key": call_id,
            });
            let (body_json, blob_hash, blob_size) = maybe_spill(body, blobs);
            out.events.push(Event {
                session_id: session_id.to_string(),
                ts_ms,
                kind: EventKind::ToolResult,
                step: None,
                summary: truncate_summary(&format!("tool_result: {}", one_line(output))),
                body_json,
                blob_hash,
                blob_size,
                correlates: None,
            });
        }
        _ => {} // unknown response_item type: skip (tolerant)
    }
}

/// Parse an `event_msg` payload into events.
#[allow(dead_code, clippy::too_many_lines)]
fn parse_codex_event_msg(
    payload: &serde_json::Value,
    session_id: &str,
    ts_ms: i64,
    blobs: &BlobStore,
    out: &mut ParsedRecord,
) {
    let Some(obj) = payload.as_object() else {
        return;
    };
    let inner_ty = obj.get("type").and_then(|v| v.as_str()).unwrap_or("");
    match inner_ty {
        "user_message" => {
            if let Some(text) = extract_codex_event_text(obj) {
                out.events.push(text_event(
                    session_id,
                    ts_ms,
                    EventKind::UserMessage,
                    &text,
                    blobs,
                ));
            }
        }
        "agent_message" => {
            if let Some(text) = extract_codex_event_text(obj) {
                out.events.push(text_event(
                    session_id,
                    ts_ms,
                    EventKind::AgentMessage,
                    &text,
                    blobs,
                ));
            }
            out.is_assistant_message = true;
        }
        "token_count" => {
            // Extract model/usage from token_count events
            if let Some(info) = obj.get("info").and_then(|v| v.as_object()) {
                if info.get("total_token_usage").is_some() {
                    out.is_assistant_message = true;
                    // Store usage in the OutcomeAcc via a side channel: the
                    // tailer checks `is_assistant_message` and calls
                    // `acc.note_assistant` with the payload. We set the flag
                    // so the tailer knows to extract usage from this record.
                    // The actual usage is in `info.total_token_usage`.
                }
            }
        }
        "exec_command_end" | "patch_apply_end" => {
            // These are tool results from shell commands / patch applications.
            let call_id = obj.get("call_id").and_then(|v| v.as_str()).unwrap_or("");
            let aggregated_output = obj
                .get("aggregated_output")
                .and_then(|v| v.as_str())
                .unwrap_or("");
            let stdout = obj.get("stdout").and_then(|v| v.as_str()).unwrap_or("");
            let stderr = obj.get("stderr").and_then(|v| v.as_str()).unwrap_or("");
            let exit_code = obj
                .get("exit_code")
                .and_then(serde_json::Value::as_i64)
                .unwrap_or(0);
            let content = if aggregated_output.is_empty() {
                let mut parts = Vec::new();
                if !stdout.is_empty() {
                    parts.push(stdout);
                }
                if !stderr.is_empty() {
                    parts.push(stderr);
                }
                parts.join("\n")
            } else {
                aggregated_output.to_string()
            };
            let body = serde_json::json!({
                "tool_use_id": call_id,
                "is_error": exit_code != 0,
                "content": content,
                "correlate_key": call_id,
            });
            let (body_json, blob_hash, blob_size) = maybe_spill(body, blobs);
            let summary = if exit_code != 0 {
                format!("tool_result (error): {}", one_line(&content))
            } else {
                format!("tool_result: {}", one_line(&content))
            };
            out.events.push(Event {
                session_id: session_id.to_string(),
                ts_ms,
                kind: EventKind::ToolResult,
                step: None,
                summary: truncate_summary(&summary),
                body_json,
                blob_hash,
                blob_size,
                correlates: None,
            });
        }
        _ => {} // unknown event_msg type: skip (tolerant)
    }
}

/// Extract text content from a Codex event_msg payload. The text may be in
/// `content` (string or array of parts) or `text` (direct string).
#[allow(dead_code)]
fn extract_codex_event_text(obj: &serde_json::Map<String, serde_json::Value>) -> Option<String> {
    // Try `content` first (array of parts or string)
    if let Some(content) = obj.get("content") {
        match content {
            serde_json::Value::String(s) => return Some(s.clone()),
            serde_json::Value::Array(arr) => {
                let parts: String = arr
                    .iter()
                    .filter_map(|b| b.get("text").and_then(|t| t.as_str()))
                    .collect::<Vec<_>>()
                    .join("\n");
                if !parts.is_empty() {
                    return Some(parts);
                }
            }
            _ => {}
        }
    }
    // Fallback: `text` field
    if let Some(text) = obj.get("text").and_then(|v| v.as_str()) {
        return Some(text.to_string());
    }
    None
}

// ---------------------------------------------------------------------------
// Gemini CLI adapter
// ---------------------------------------------------------------------------

/// The Gemini CLI adapter: tails `~/.gemini/tmp/<hash>/chats/session-*.jsonl`
/// and converts records to `user_message` / `agent_message` / `thinking` /
/// `tool_call` / `tool_result` events. Stateless; the tailer thread holds all
/// mutable state.
///
/// ## Experimental
///
/// This adapter is built against the researched Gemini CLI session format
/// (<https://github.com/google-gemini/gemini-cli>). The parser handles the
/// documented record types (`session_metadata`, `user`, `gemini`,
/// `message_update`) and degrades gracefully on unknown types. If you encounter
/// a session that produces 0 steps despite having a transcript, run
/// `HH_DEBUG=1 hh run -- gemini ...` and file an issue with the debug output.
#[derive(Debug, Clone)]
pub struct GeminiAdapter;

impl Adapter for GeminiAdapter {
    fn agent_kind(&self) -> AgentKind {
        AgentKind::GeminiCli
    }

    #[allow(clippy::unused_self)]
    fn spawn(self: Box<Self>, ctx: AdapterContext) -> std::io::Result<AdapterHandle> {
        let (tx, rx) = mpsc::channel::<Event>();
        let outcome = std::thread::Builder::new()
            .name("hh-gemini-adapter".into())
            .spawn(move || run_gemini_tailer(ctx, tx))?;
        Ok(AdapterHandle {
            events: rx,
            outcome,
        })
    }
}

/// The Gemini CLI temp directory: `$HOME/.gemini/tmp` (Unix) or
/// `%USERPROFILE%\.gemini\tmp` (Windows). `None` if neither env var is set.
#[must_use]
pub fn gemini_tmp_dir() -> Option<PathBuf> {
    let home = std::env::var_os("HOME").or_else(|| std::env::var_os("USERPROFILE"))?;
    Some(PathBuf::from(home).join(".gemini").join("tmp"))
}

/// The tailer thread body for Gemini CLI: locate the session transcript, then
/// read it to EOF/stop, parsing each line into events sent on `tx`.
#[allow(clippy::needless_pass_by_value)] // ctx/tx are owned for the thread's lifetime
fn run_gemini_tailer(ctx: AdapterContext, tx: mpsc::Sender<Event>) -> AdapterOutcome {
    let Some(tmp) = ctx.projects_dir.clone().or_else(gemini_tmp_dir) else {
        return degraded(
            "no ~/.gemini/tmp directory found (HOME unset?); run `hh doctor` to diagnose",
        );
    };
    if !tmp.is_dir() {
        return degraded(format!(
            "gemini tmp directory does not exist: {}; run `hh doctor` to diagnose",
            tmp.display()
        ));
    }
    let start = Instant::now();
    let Some(file) = locate_gemini_transcript(&tmp, &ctx.cwd, &ctx.stop) else {
        return degraded(format!(
            "no gemini session transcript appeared under {} matching cwd {} before the session ended",
            tmp.display(),
            ctx.cwd.display(),
        ));
    };
    let mut acc = OutcomeAcc::default();
    let stats = tail_gemini_file(&file, &ctx, &tx, &start, &mut acc);
    if !stats.read_ok {
        return degraded(format!(
            "found a gemini transcript at {} but could not read it; \
             run `hh doctor` to check file permissions",
            file.display(),
        ));
    }
    if stats.events_produced == 0 {
        let first_err = stats
            .first_parse_error
            .map(|(n, msg)| format!("; first parse error at line {n}: {msg}"))
            .unwrap_or_default();
        return degraded(format!(
            "gemini transcript found ({}) but 0 records parsed (read {} line(s){first_err}); \
             run `hh doctor` to check the Gemini CLI transcript format",
            file.display(),
            stats.lines_seen,
        ));
    }
    acc.finish(AdapterStatus::Active)
}

/// Locate the Gemini CLI session transcript for this session.
///
/// Gemini stores session files at `~/.gemini/tmp/<project_hash>/chats/session-*.jsonl`.
/// The adapter scans all project hash directories for a `chats/` subdirectory,
/// then polls for new `session-*.jsonl` files.
fn locate_gemini_transcript(
    tmp: &Path,
    cwd: &Path,
    stop: &std::sync::Arc<AtomicBool>,
) -> Option<PathBuf> {
    let preexisting = snapshot_gemini_sessions(tmp);
    let mut rejected: std::collections::HashSet<PathBuf> = std::collections::HashSet::new();

    loop {
        if stop.load(Ordering::Acquire) {
            return None;
        }
        // Scan every project hash dir for new session files in chats/.
        let Ok(projects) = std::fs::read_dir(tmp) else {
            std::thread::sleep(APPEAR_POLL);
            continue;
        };
        for project in projects.flatten() {
            let chats = project.path().join("chats");
            if !chats.is_dir() {
                continue;
            }
            for f in new_gemini_sessions_in(&chats, &preexisting) {
                if rejected.contains(&f) {
                    continue;
                }
                match first_gemini_record_cwd(&f, cwd) {
                    CwdState::Matches => return Some(f),
                    CwdState::Different | CwdState::Directory => {
                        rejected.insert(f);
                    }
                    CwdState::None => {}
                }
            }
        }
        std::thread::sleep(APPEAR_POLL);
    }
}

/// Snapshot every `session-*.jsonl` file under the tmp directory at start.
fn snapshot_gemini_sessions(tmp: &Path) -> std::collections::HashSet<PathBuf> {
    let mut set = std::collections::HashSet::new();
    let Ok(projects) = std::fs::read_dir(tmp) else {
        return set;
    };
    for project in projects.flatten() {
        let chats = project.path().join("chats");
        if !chats.is_dir() {
            continue;
        }
        if let Ok(entries) = std::fs::read_dir(&chats) {
            for e in entries.flatten() {
                let p = e.path();
                if p.file_name().and_then(|n| n.to_str()).is_some_and(|n| {
                    n.starts_with("session-")
                        && std::path::Path::new(n).extension()
                            == Some(std::ffi::OsStr::new("jsonl"))
                }) {
                    set.insert(p);
                }
            }
        }
    }
    set
}

/// `session-*.jsonl` files in `dir` not in `preexisting`, newest-first by mtime.
fn new_gemini_sessions_in(
    dir: &Path,
    preexisting: &std::collections::HashSet<PathBuf>,
) -> Vec<PathBuf> {
    let Ok(entries) = std::fs::read_dir(dir) else {
        return Vec::new();
    };
    let mut found: Vec<(PathBuf, i64)> = Vec::new();
    for e in entries.flatten() {
        let p = e.path();
        if !p.file_name().and_then(|n| n.to_str()).is_some_and(|n| {
            n.starts_with("session-")
                && std::path::Path::new(n).extension() == Some(std::ffi::OsStr::new("jsonl"))
        }) {
            continue;
        }
        if preexisting.contains(&p) {
            continue;
        }
        let mtime_ms = e
            .metadata()
            .ok()
            .and_then(|m| m.modified().ok())
            .map_or(0, system_time_to_unix_ms);
        found.push((p, mtime_ms));
    }
    found.sort_by_key(|(_, m)| std::cmp::Reverse(*m));
    found.into_iter().map(|(p, _)| p).collect()
}

/// Inspect the first cwd-bearing record in a Gemini session file. Gemini puts
/// `cwd` in the `session_metadata` record (first line).
fn first_gemini_record_cwd(file: &Path, _session_cwd: &Path) -> CwdState {
    if file.is_dir() {
        return CwdState::Directory;
    }
    let Ok(content) = std::fs::read_to_string(file) else {
        return CwdState::None;
    };
    for line in content.lines() {
        let line = line.trim();
        if line.is_empty() {
            continue;
        }
        let Ok(_v) = serde_json::from_str::<serde_json::Value>(line) else {
            continue;
        };
        // Gemini doesn't carry cwd in the session file; it uses projectHash.
        // We accept any session file that appears (no cwd verification).
        // The projectHash is derived from the cwd, so a session in the right
        // project dir is the right session.
        return CwdState::Matches;
    }
    CwdState::None
}

/// Tail a Gemini session file, parsing each line into events.
fn tail_gemini_file(
    file: &Path,
    ctx: &AdapterContext,
    tx: &mpsc::Sender<Event>,
    start: &Instant,
    acc: &mut OutcomeAcc,
) -> TailStats {
    let mut stats = TailStats::default();
    let mut reader = match std::fs::File::open(file) {
        Ok(f) => std::io::BufReader::new(f),
        Err(_) => return stats,
    };
    stats.read_ok = true;
    let mut buf: Vec<u8> = Vec::new();
    let mut line = Vec::new();
    loop {
        if ctx.stop.load(Ordering::Acquire) {
            break;
        }
        match reader.read_until(b'\n', &mut line) {
            Ok(0) => {
                drain_complete_lines(&mut buf, &mut line, ctx, tx, start, acc, &mut stats);
                if ctx.stop.load(Ordering::Acquire) {
                    break;
                }
                std::thread::sleep(TAIL_POLL);
            }
            Ok(_n) => {
                drain_complete_lines(&mut buf, &mut line, ctx, tx, start, acc, &mut stats);
            }
            Err(e) if e.kind() == std::io::ErrorKind::Interrupted => {}
            Err(_) => break,
        }
    }
    drain_complete_lines(&mut buf, &mut line, ctx, tx, start, acc, &mut stats);
    stats
}

// ---------------------------------------------------------------------------
// Gemini CLI record parser
// ---------------------------------------------------------------------------

/// Parse one Gemini CLI JSONL record into events. Pure given `(value, ts_ms,
/// blobs)`: deterministic output for fixed inputs. Unknown `type`s and
/// unexpected shapes produce no events (tolerant).
///
/// Gemini session format: each line is a JSON object with a `type` field.
/// Types: `session_metadata` (skip), `user` (user message), `gemini` (model
/// message with optional tool calls/thoughts), `message_update` (skip).
#[must_use]
#[allow(dead_code, clippy::too_many_lines)] // called from the tailer thread; not yet wired through parse_and_send
pub(crate) fn parse_gemini_record(
    value: &serde_json::Value,
    session_id: &str,
    ts_ms: i64,
    blobs: &BlobStore,
) -> ParsedRecord {
    let mut out = ParsedRecord::default();
    let Some(obj) = value.as_object() else {
        return out;
    };
    let ty = obj.get("type").and_then(|v| v.as_str()).unwrap_or("");
    match ty {
        "user" => {
            if let Some(text) = extract_gemini_text(obj) {
                out.events.push(text_event(
                    session_id,
                    ts_ms,
                    EventKind::UserMessage,
                    &text,
                    blobs,
                ));
            }
        }
        "gemini" => {
            // Extract text content
            if let Some(text) = extract_gemini_text(obj) {
                out.events.push(text_event(
                    session_id,
                    ts_ms,
                    EventKind::AgentMessage,
                    &text,
                    blobs,
                ));
            }
            // Extract thoughts → Thinking events
            if let Some(thoughts) = obj.get("thoughts").and_then(|v| v.as_array()) {
                for thought in thoughts {
                    if let Some(text) = thought.get("text").and_then(|v| v.as_str()) {
                        out.events.push(text_event(
                            session_id,
                            ts_ms,
                            EventKind::Thinking,
                            text,
                            blobs,
                        ));
                    }
                }
            }
            // Extract tool calls
            if let Some(tool_calls) = obj.get("toolCalls").and_then(|v| v.as_array()) {
                for tc in tool_calls {
                    if let Some(tc_obj) = tc.as_object() {
                        let call_id = tc_obj.get("id").and_then(|v| v.as_str()).unwrap_or("");
                        let name = tc_obj.get("name").and_then(|v| v.as_str()).unwrap_or("");
                        let input = tc_obj
                            .get("input")
                            .cloned()
                            .unwrap_or(serde_json::Value::Null);
                        let body = serde_json::json!({
                            "name": name,
                            "input": input,
                            "correlate_key": call_id,
                        });
                        let (body_json, blob_hash, blob_size) = maybe_spill(body, blobs);
                        out.events.push(Event {
                            session_id: session_id.to_string(),
                            ts_ms,
                            kind: EventKind::ToolCall,
                            step: None,
                            summary: truncate_summary(&format!("tool_call: {name}")),
                            body_json,
                            blob_hash,
                            blob_size,
                            correlates: None,
                        });
                        // If the tool call has a result inline, emit a ToolResult too
                        if let Some(result) = tc_obj.get("result").and_then(|v| v.as_array()) {
                            let result_text: String = result
                                .iter()
                                .filter_map(|p| p.get("text").and_then(|t| t.as_str()))
                                .collect::<Vec<_>>()
                                .join("\n");
                            let body = serde_json::json!({
                                "tool_use_id": call_id,
                                "is_error": false,
                                "content": result_text,
                                "correlate_key": call_id,
                            });
                            let (body_json, blob_hash, blob_size) = maybe_spill(body, blobs);
                            out.events.push(Event {
                                session_id: session_id.to_string(),
                                ts_ms,
                                kind: EventKind::ToolResult,
                                step: None,
                                summary: truncate_summary(&format!(
                                    "tool_result: {}",
                                    one_line(&result_text)
                                )),
                                body_json,
                                blob_hash,
                                blob_size,
                                correlates: None,
                            });
                        }
                    }
                }
            }
            // Extract token usage
            if obj.get("tokens").is_some() {
                out.is_assistant_message = true;
            }
            // Extract model
            if let Some(_model) = obj.get("model").and_then(|v| v.as_str()) {
                out.is_assistant_message = true;
            }
        }
        // session_metadata, message_update: skip
        _ => {}
    }
    out
}

/// Extract text content from a Gemini message record. The content is an array
/// of parts, each with a `text` field, or a direct string.
fn extract_gemini_text(obj: &serde_json::Map<String, serde_json::Value>) -> Option<String> {
    if let Some(content) = obj.get("content") {
        match content {
            serde_json::Value::String(s) => return Some(s.clone()),
            serde_json::Value::Array(arr) => {
                let parts: String = arr
                    .iter()
                    .filter_map(|b| b.get("text").and_then(|t| t.as_str()))
                    .collect::<Vec<_>>()
                    .join("\n");
                if !parts.is_empty() {
                    return Some(parts);
                }
            }
            _ => {}
        }
    }
    None
}

// ---------------------------------------------------------------------------
// Pure parser (the insta target)
// ---------------------------------------------------------------------------

/// One parsed record: the events it produced, plus whether it was an assistant
/// message (so the tailer can collect model/usage) and any model/usage seen.
#[derive(Debug, Default, Clone)]
pub(crate) struct ParsedRecord {
    /// Events derived from the record's content blocks.
    pub events: Vec<Event>,
    /// True for `assistant` records (the tailer collects model/usage from these).
    pub is_assistant_message: bool,
}

/// Parse one Claude JSONL record into events. Pure given `(value, ts_ms,
/// blobs)`: deterministic output for fixed inputs (the only side effect is
/// deterministic blob spillover for >256 KiB payloads, whose hash is BLAKE3 of
/// fixed content). Unknown `type`s, `isSidechain`/`isMeta` records, and
/// unexpected shapes produce no events (tolerant — see module docs).
#[must_use]
pub(crate) fn parse_record(
    value: &serde_json::Value,
    session_id: &str,
    ts_ms: i64,
    blobs: &BlobStore,
) -> ParsedRecord {
    let mut out = ParsedRecord::default();
    let Some(obj) = value.as_object() else {
        return out; // non-object line: nothing to parse
    };
    let ty = obj.get("type").and_then(|v| v.as_str()).unwrap_or("");
    if ty != "user" && ty != "assistant" {
        return out; // system / mode / attachment / ai-title / file-history-snapshot / ...
    }
    if obj
        .get("isSidechain")
        .and_then(serde_json::Value::as_bool)
        .unwrap_or(false)
    {
        return out; // subagent transcript
    }
    if obj
        .get("isMeta")
        .and_then(serde_json::Value::as_bool)
        .unwrap_or(false)
    {
        return out; // injected caveat / system-reminder wrappers
    }
    let Some(message) = obj.get("message") else {
        return out;
    };
    let is_assistant = ty == "assistant";
    out.is_assistant_message = is_assistant;
    let Some(content) = message.get("content") else {
        return out;
    };
    match content {
        serde_json::Value::String(s) => {
            let kind = if is_assistant {
                EventKind::AgentMessage
            } else {
                EventKind::UserMessage
            };
            out.events
                .push(text_event(session_id, ts_ms, kind, s, blobs));
        }
        serde_json::Value::Array(blocks) => {
            for block in blocks {
                if let Some(b) = block.as_object() {
                    parse_block(b, is_assistant, session_id, ts_ms, blobs, &mut out.events);
                }
            }
        }
        _ => {} // content is null/number/etc: skip
    }
    out
}

/// Parse one content block into events appended to `events`.
fn parse_block(
    b: &serde_json::Map<String, serde_json::Value>,
    is_assistant: bool,
    session_id: &str,
    ts_ms: i64,
    blobs: &BlobStore,
    events: &mut Vec<Event>,
) {
    let bty = b.get("type").and_then(|v| v.as_str()).unwrap_or("");
    match bty {
        "text" => {
            if let Some(t) = b.get("text").and_then(|v| v.as_str()) {
                let kind = if is_assistant {
                    EventKind::AgentMessage
                } else {
                    EventKind::UserMessage
                };
                events.push(text_event(session_id, ts_ms, kind, t, blobs));
            }
        }
        "thinking" => {
            if let Some(t) = b.get("thinking").and_then(|v| v.as_str()) {
                events.push(text_event(session_id, ts_ms, EventKind::Thinking, t, blobs));
            }
        }
        "tool_use" => {
            let id = b.get("id").and_then(|v| v.as_str()).unwrap_or("");
            let name = b.get("name").and_then(|v| v.as_str()).unwrap_or("");
            let input = b.get("input").cloned().unwrap_or(serde_json::Value::Null);
            events.push(tool_call_event(session_id, ts_ms, id, name, &input, blobs));
        }
        "tool_result" => {
            let tool_use_id = b.get("tool_use_id").and_then(|v| v.as_str()).unwrap_or("");
            let is_error = b
                .get("is_error")
                .and_then(serde_json::Value::as_bool)
                .unwrap_or(false);
            let content_text = extract_result_content(b.get("content"));
            events.push(tool_result_event(
                session_id,
                ts_ms,
                tool_use_id,
                is_error,
                &content_text,
                blobs,
            ));
        }
        _ => {} // unknown block type: skip (tolerant)
    }
}

/// Build a text-bearing event (`user_message`/`agent_message`/`thinking`).
fn text_event(
    session_id: &str,
    ts_ms: i64,
    kind: EventKind,
    text: &str,
    blobs: &BlobStore,
) -> Event {
    let body = serde_json::json!({ "text": text });
    let (body_json, blob_hash, blob_size) = maybe_spill(body, blobs);
    Event {
        session_id: session_id.to_string(),
        ts_ms,
        kind,
        step: None, // assigned by the FR-3.4 pass
        summary: truncate_summary(text),
        body_json,
        blob_hash,
        blob_size,
        correlates: None,
    }
}

/// Build a `tool_call` event, correlating to its later `tool_result` by id.
fn tool_call_event(
    session_id: &str,
    ts_ms: i64,
    id: &str,
    name: &str,
    input: &serde_json::Value,
    blobs: &BlobStore,
) -> Event {
    let body = serde_json::json!({
        "name": name,
        "input": input,
        "correlate_key": id,
    });
    let (body_json, blob_hash, blob_size) = maybe_spill(body, blobs);
    Event {
        session_id: session_id.to_string(),
        ts_ms,
        kind: EventKind::ToolCall,
        step: None,
        summary: truncate_summary(&format!("tool_call: {name}")),
        body_json,
        blob_hash,
        blob_size,
        correlates: None,
    }
}

/// Build a `tool_result` event, correlating to its `tool_call` by `tool_use_id`.
fn tool_result_event(
    session_id: &str,
    ts_ms: i64,
    tool_use_id: &str,
    is_error: bool,
    content: &str,
    blobs: &BlobStore,
) -> Event {
    let body = serde_json::json!({
        "tool_use_id": tool_use_id,
        "is_error": is_error,
        "content": content,
        "correlate_key": tool_use_id,
    });
    let (body_json, blob_hash, blob_size) = maybe_spill(body, blobs);
    let summary = if is_error {
        format!("tool_result (error): {}", one_line(content))
    } else {
        format!("tool_result: {}", one_line(content))
    };
    Event {
        session_id: session_id.to_string(),
        ts_ms,
        kind: EventKind::ToolResult,
        step: None,
        summary: truncate_summary(&summary),
        body_json,
        blob_hash,
        blob_size,
        correlates: None,
    }
}

/// If `body` serializes to ≥ [`SPILLOVER_BYTES`], store it as a blob and return
/// an overflow envelope; otherwise return the body inline. Best-effort: a blob
/// write failure falls back to inline storage rather than dropping the event.
fn maybe_spill(
    body: serde_json::Value,
    blobs: &BlobStore,
) -> (Option<serde_json::Value>, Option<String>, Option<u64>) {
    let serialized = serde_json::to_vec(&body).unwrap_or_default();
    if serialized.len() >= SPILLOVER_BYTES {
        if let Ok(outcome) = blobs.put(&serialized) {
            let envelope = serde_json::json!({
                "overflow": true,
                "size": outcome.size,
                "blob_hash": outcome.hash,
                "encoding": "blob",
            });
            return (Some(envelope), Some(outcome.hash), Some(outcome.size));
        }
    }
    (Some(body), None, None)
}

/// Extract a tool_result's `content` as a string, joining text blocks if it is
/// an array. Empty string for missing/unknown shapes.
fn extract_result_content(content: Option<&serde_json::Value>) -> String {
    match content {
        Some(serde_json::Value::String(s)) => s.clone(),
        Some(serde_json::Value::Array(arr)) => {
            let mut parts = Vec::with_capacity(arr.len());
            for b in arr {
                if let Some(t) = b.get("text").and_then(|v| v.as_str()) {
                    parts.push(t.to_string());
                }
            }
            parts.join("\n")
        }
        _ => String::new(),
    }
}

/// Collapse newlines so a summary fits on one line (length capped by the caller).
fn one_line(s: &str) -> String {
    s.chars()
        .map(|c| if c == '\n' || c == '\r' { ' ' } else { c })
        .collect()
}

/// Convert a Claude `timestamp` (RFC3339, e.g. `2026-07-02T06:14:40.699Z`) to
/// milliseconds relative to `started_at_unix_ms`. Returns `None` on parse
/// failure (the tailer falls back to wall-clock elapsed time).
pub(crate) fn ts_ms_from_iso(iso: &str, started_at_unix_ms: i64) -> Option<i64> {
    let dt =
        time::OffsetDateTime::parse(iso, &time::format_description::well_known::Rfc3339).ok()?;
    let unix_ms = dt.unix_timestamp_nanos() / 1_000_000;
    let unix_ms = i64::try_from(unix_ms).unwrap_or(i64::MAX);
    Some((unix_ms - started_at_unix_ms).max(0))
}

// ---------------------------------------------------------------------------
// Path / time helpers
// ---------------------------------------------------------------------------

/// The Claude Code projects directory: `$HOME/.claude/projects` (Unix) or
/// `%USERPROFILE%\.claude\projects` (Windows). `None` if neither env var is set.
/// Public so `hh doctor` can report Claude Code transcript discoverability.
#[must_use]
pub fn claude_projects_dir() -> Option<PathBuf> {
    let home = std::env::var_os("HOME").or_else(|| std::env::var_os("USERPROFILE"))?;
    Some(PathBuf::from(home).join(".claude").join("projects"))
}

/// The newest `*.jsonl` transcript Claude Code has written for `cwd` (under the
/// slug dir derived from `cwd`), or `None` if the slug dir has no transcripts.
/// A best-effort discovery probe for `hh doctor` — no session-start filtering,
/// so it surfaces whatever Claude most recently wrote for this directory.
#[must_use]
pub fn newest_jsonl_for_cwd(cwd: &Path) -> Option<PathBuf> {
    let projects = claude_projects_dir()?;
    let slug_dir = projects.join(slugify(cwd));
    if !slug_dir.is_dir() {
        return None;
    }
    // `since = 0` disables the mtime floor so every .jsonl qualifies.
    newest_in_dir(&slug_dir, 0)
}

/// Encode `cwd` as a Claude slug: every `/` (and `\`, `.`) becomes `-`. This is
/// inferred from a sample transcript (the SRS is absent); it is a *hint only* —
/// the tailer falls back to a cwd-based scan when the slug dir misses.
#[must_use]
pub(crate) fn slugify(cwd: &Path) -> String {
    cwd.to_string_lossy()
        .chars()
        .map(|c| match c {
            '/' | '\\' | '.' => '-',
            other => other,
        })
        .collect()
}

/// A `SystemTime` as unix-ms, clamped to `i64` range.
fn system_time_to_unix_ms(t: std::time::SystemTime) -> i64 {
    t.duration_since(std::time::UNIX_EPOCH)
        .map_or(0, |d| i64::try_from(d.as_millis()).unwrap_or(i64::MAX))
}

/// Wall-clock ms elapsed since `start` (the tailer's fallback timestamp).
fn elapsed_ms(start: &Instant) -> i64 {
    i64::try_from(start.elapsed().as_millis()).unwrap_or(i64::MAX)
}

/// A debug trace on stderr, gated on the `HH_DEBUG` env var (no `log`/`tracing`
/// dep, per CLAUDE.md's "small, well-maintained crates" guidance). Set
/// `HH_DEBUG=1` to capture the adapter's discovery+parse trace: the computed
/// slug, projects dir, candidate files, the selected transcript and why, how
/// many lines/records were read, how many events were produced, and the first
/// conversion failure (line + message) if any. This is the trace that decides
/// discovery-vs-parse for a degraded session.
fn debug(msg: &str) {
    if std::env::var_os("HH_DEBUG").is_some() {
        eprintln!("hh: debug: {msg}");
    }
}

/// Fuzz-only entry points into the otherwise-private Claude/Codex/Gemini JSONL
/// parsers (`cargo fuzz` targets `claude_jsonl`, `codex_jsonl`, `gemini_jsonl`).
/// Gated behind the `fuzzing` feature so it never widens the crate's normal
/// public API.
#[cfg(feature = "fuzzing")]
pub mod fuzzing {
    use super::{parse_codex_record, parse_gemini_record, parse_record, BlobStore};
    use std::sync::OnceLock;

    /// A blob store rooted in a process-unique temp dir, reused across fuzz
    /// iterations (opening a fresh one per call would dominate runtime with
    /// filesystem setup rather than parser logic).
    fn blobs() -> &'static BlobStore {
        static BLOBS: OnceLock<BlobStore> = OnceLock::new();
        BLOBS.get_or_init(|| {
            let dir = std::env::temp_dir().join(format!("hh-fuzz-adapter-{}", std::process::id()));
            BlobStore::new(dir)
        })
    }

    /// Mirrors `parse_and_send`'s path from one raw tailed line (UTF-8 validate
    /// → trim → JSON parse → [`parse_record`]) — the exact sequence the live
    /// tailer runs on untrusted transcript content. Must never panic.
    pub fn fuzz_parse_line(bytes: &[u8]) {
        let Ok(s) = std::str::from_utf8(bytes) else {
            return;
        };
        let s = s.trim();
        if s.is_empty() {
            return;
        }
        let Ok(value) = serde_json::from_str::<serde_json::Value>(s) else {
            return;
        };
        let _ = parse_record(&value, "fuzz-session", 0, blobs());
    }

    /// Mirrors the Codex CLI tailer's per-line path: UTF-8 validate → trim →
    /// JSON parse → [`parse_codex_record`]. Must never panic.
    pub fn fuzz_parse_codex_line(bytes: &[u8]) {
        let Ok(s) = std::str::from_utf8(bytes) else {
            return;
        };
        let s = s.trim();
        if s.is_empty() {
            return;
        }
        let Ok(value) = serde_json::from_str::<serde_json::Value>(s) else {
            return;
        };
        let _ = parse_codex_record(&value, "fuzz-session", 0, blobs());
    }

    /// Mirrors the Gemini CLI tailer's per-line path: UTF-8 validate → trim →
    /// JSON parse → [`parse_gemini_record`]. Must never panic.
    pub fn fuzz_parse_gemini_line(bytes: &[u8]) {
        let Ok(s) = std::str::from_utf8(bytes) else {
            return;
        };
        let s = s.trim();
        if s.is_empty() {
            return;
        }
        let Ok(value) = serde_json::from_str::<serde_json::Value>(s) else {
            return;
        };
        let _ = parse_gemini_record(&value, "fuzz-session", 0, blobs());
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use serde_json::Value;
    use std::io::Write;

    /// A fixed session start (2026-07-02T06:14:40.000Z) so timestamps derive
    /// small, stable, readable `ts_ms` values for snapshots.
    const STARTED: i64 = 1_782_972_880_000;
    const SID: &str = "test-session-id";

    fn iso(s: &str) -> i64 {
        ts_ms_from_iso(s, STARTED).expect("fixture timestamps must parse")
    }

    fn parse(value: &Value, ts_ms: i64) -> ParsedRecord {
        let tmp = tempfile::TempDir::new().unwrap();
        let blobs = BlobStore::new(tmp.path().join("blobs"));
        parse_record(value, SID, ts_ms, &blobs)
    }

    /// Serialize a parsed record's events (and model/usage when present) as
    /// pretty JSON for a stable insta snapshot.
    fn snap(pr: &ParsedRecord) -> String {
        serde_json::to_string_pretty(&serde_json::json!({
            "events": pr.events,
            "is_assistant_message": pr.is_assistant_message,
        }))
        .unwrap()
    }

    fn rec(json: &str) -> Value {
        serde_json::from_str(json).unwrap()
    }

    #[test]
    fn parse_user_text_prompt() {
        let v = rec(r#"{"type":"user","isSidechain":false,"isMeta":false,
            "message":{"role":"user","content":"Please list the files in the repo and show Cargo.toml."},
            "timestamp":"2026-07-02T06:14:40.699Z","cwd":"/tmp/work","sessionId":"s"}"#);
        let pr = parse(&v, iso("2026-07-02T06:14:40.699Z"));
        insta::assert_snapshot!(snap(&pr));
    }

    #[test]
    fn parse_assistant_with_two_tool_uses() {
        let v = rec(r#"{"type":"assistant","isSidechain":false,"isMeta":false,
            "message":{"role":"assistant","model":"glm-5.2","content":[
              {"type":"thinking","thinking":"I'll list files then read Cargo.toml.","signature":""},
              {"type":"tool_use","id":"call_abc123","name":"Bash","input":{"command":"ls -la"}},
              {"type":"tool_use","id":"call_def456","name":"Read","input":{"file_path":"Cargo.toml"}}
            ],"usage":{"input_tokens":1200,"output_tokens":80}},
            "timestamp":"2026-07-02T06:14:41.500Z","cwd":"/tmp/work","sessionId":"s"}"#);
        let pr = parse(&v, iso("2026-07-02T06:14:41.500Z"));
        insta::assert_snapshot!(snap(&pr));
        // thinking + two tool calls = 3 events; assistant message flagged.
        assert_eq!(pr.events.len(), 3);
        assert!(pr.is_assistant_message);
        assert_eq!(pr.events[0].kind, EventKind::Thinking);
        assert_eq!(pr.events[1].kind, EventKind::ToolCall);
        assert_eq!(pr.events[2].kind, EventKind::ToolCall);
    }

    #[test]
    fn parse_user_with_two_tool_results_one_error() {
        let v = rec(r#"{"type":"user","isSidechain":false,"isMeta":false,
            "message":{"role":"user","content":[
              {"type":"tool_result","tool_use_id":"call_abc123","content":"total 0","is_error":false},
              {"type":"tool_result","tool_use_id":"call_def456","content":"Error: file not found","is_error":true}
            ]},
            "timestamp":"2026-07-02T06:14:42.800Z","cwd":"/tmp/work","sessionId":"s"}"#);
        let pr = parse(&v, iso("2026-07-02T06:14:42.800Z"));
        insta::assert_snapshot!(snap(&pr));
        assert_eq!(pr.events.len(), 2);
        assert_eq!(pr.events[0].kind, EventKind::ToolResult);
        assert_eq!(pr.events[1].kind, EventKind::ToolResult);
        // correlate_key == tool_use_id in each body.
        assert_eq!(
            pr.events[0].body_json.as_ref().unwrap()["correlate_key"],
            "call_abc123"
        );
        assert!(pr.events[1].body_json.as_ref().unwrap()["is_error"]
            .as_bool()
            .unwrap());
    }

    #[test]
    fn parse_assistant_followup_with_model_usage() {
        let v = rec(r#"{"type":"assistant","isSidechain":false,"isMeta":false,
            "message":{"role":"assistant","model":"glm-5.2","content":[
              {"type":"text","text":"The repo is empty and Cargo.toml is missing."}
            ],"usage":{"input_tokens":1500,"output_tokens":40,"cache_read_input_tokens":300}},
            "timestamp":"2026-07-02T06:14:43.100Z","cwd":"/tmp/work","sessionId":"s"}"#);
        let pr = parse(&v, iso("2026-07-02T06:14:43.100Z"));
        insta::assert_snapshot!(snap(&pr));
        assert_eq!(pr.events.len(), 1);
        assert_eq!(pr.events[0].kind, EventKind::AgentMessage);
        // The tailer collects model/usage from the message; exercise that path.
        let mut acc = OutcomeAcc::default();
        acc.note_assistant(v.get("message").unwrap());
        assert_eq!(acc.model.as_deref(), Some("glm-5.2"));
        assert_eq!(acc.usage_json.as_ref().unwrap()["output_tokens"], 40);
    }

    #[test]
    fn parse_skips_unknown_type() {
        let v = rec(r#"{"type":"system","subtype":"init","cwd":"/tmp/work","sessionId":"s"}"#);
        let pr = parse(&v, 0);
        assert!(pr.events.is_empty());
        assert!(!pr.is_assistant_message);
    }

    #[test]
    fn parse_skips_non_object_value() {
        // A malformed line that parsed as a JSON array/null produces no events.
        for v in [rec("[]"), rec("null"), rec("\"oops\"")] {
            assert!(
                parse(&v, 0).events.is_empty(),
                "non-object must yield no events"
            );
        }
    }

    #[test]
    fn parse_skips_is_sidechain() {
        let v = rec(r#"{"type":"assistant","isSidechain":true,"isMeta":false,
            "message":{"role":"assistant","content":[{"type":"text","text":"subagent"}]},
            "timestamp":"2026-07-02T06:14:41.000Z","cwd":"/tmp/work","sessionId":"s"}"#);
        assert!(parse(&v, 0).events.is_empty());
    }

    #[test]
    fn parse_skips_is_meta() {
        let v = rec(r#"{"type":"user","isSidechain":false,"isMeta":true,
            "message":{"role":"user","content":"<local-command-caveat>system-reminder</local-command-caveat>"},
            "timestamp":"2026-07-02T06:14:40.600Z","cwd":"/tmp/work","sessionId":"s"}"#);
        assert!(parse(&v, 0).events.is_empty());
    }

    #[test]
    fn parse_256kib_spills_to_blob() {
        let big = "x".repeat(SPILLOVER_BYTES + 1024);
        let v = rec(&format!(
            r#"{{"type":"user","isSidechain":false,"isMeta":false,
            "message":{{"role":"user","content":{}}},
            "timestamp":"2026-07-02T06:14:40.699Z","cwd":"/tmp/work","sessionId":"s"}}"#,
            serde_json::Value::String(big)
        ));
        let tmp = tempfile::TempDir::new().unwrap();
        let blobs = BlobStore::new(tmp.path().join("blobs"));
        let pr = parse_record(&v, SID, 699, &blobs);
        let ev = &pr.events[0];
        assert!(ev.blob_hash.is_some(), "large payload must spill to a blob");
        let hash = ev.blob_hash.as_ref().unwrap();
        assert_eq!(ev.body_json.as_ref().unwrap()["overflow"], true);
        assert_eq!(ev.body_json.as_ref().unwrap()["encoding"], "blob");
        // The blob round-trips to the serialized body (a JSON object with the text).
        let raw = blobs.get(hash).unwrap();
        let body: Value = serde_json::from_slice(&raw).unwrap();
        assert_eq!(body["text"].as_str().unwrap().len(), SPILLOVER_BYTES + 1024);
    }

    #[test]
    fn slugify_replaces_slash_and_dot() {
        assert_eq!(
            slugify(Path::new("/home/saadman/switch")),
            "-home-saadman-switch"
        );
        assert_eq!(
            slugify(Path::new("/home/saadman/switch/.claude/worktrees/x")),
            "-home-saadman-switch--claude-worktrees-x"
        );
        assert_eq!(slugify(Path::new("C:\\Users\\me")), "C:-Users-me");
    }

    #[test]
    fn select_detects_claude_basename() {
        assert!(is_claude_code(&["claude".into()]));
        assert!(is_claude_code(&["/usr/local/bin/claude".into()]));
        assert!(is_claude_code(&["C:\\Apps\\claude.exe".into()]));
        assert!(!is_claude_code(&["python3".into()]));
        assert!(!is_claude_code(&[]));
        // select returns a Claude adapter for claude, None for a generic command.
        assert!(select(&["claude".into()], Path::new("/tmp")).is_some());
        assert!(select(&["python3".into()], Path::new("/tmp")).is_none());
    }

    #[test]
    fn ts_ms_from_iso_is_relative_and_clamped() {
        assert_eq!(
            ts_ms_from_iso("2026-07-02T06:14:40.699Z", STARTED),
            Some(699)
        );
        assert_eq!(ts_ms_from_iso("2026-07-02T06:14:40.000Z", STARTED), Some(0));
        // A timestamp before the session start clamps to 0 (no negative ts_ms).
        assert_eq!(ts_ms_from_iso("2026-07-02T06:14:39.000Z", STARTED), Some(0));
        assert!(ts_ms_from_iso("not-a-date", STARTED).is_none());
    }

    // --- tailer behavior (real threads, temp HOME) -----------------------

    /// A self-contained adapter spawn: temp HOME + cwd, a stop flag, and the
    /// slug dir pre-created. The transcript is written *after* spawn (see
    /// [`write_transcript`]) so it is correctly seen as NEW since session start —
    /// mirroring real usage where Claude writes its transcript lazily, after
    /// `hh run` has launched it. Returns the handle so the test can drain then
    /// join.
    fn spawn_with(home: &Path, cwd: &Path) -> (AdapterHandle, std::sync::Arc<AtomicBool>, PathBuf) {
        let slug_dir = home.join(".claude").join("projects").join(slugify(cwd));
        std::fs::create_dir_all(&slug_dir).unwrap();
        let stop = std::sync::Arc::new(AtomicBool::new(false));
        let blobs = std::sync::Arc::new(BlobStore::new(home.join("blobs")));
        let ctx = AdapterContext {
            session_id: SID.to_string(),
            started_at_unix_ms: STARTED,
            cwd: cwd.to_path_buf(),
            command: vec!["claude".into()],
            blobs,
            stop: std::sync::Arc::clone(&stop),
            projects_dir: Some(home.join(".claude").join("projects")),
        };
        let handle = Box::new(ClaudeAdapter).spawn(ctx).expect("spawn adapter");
        // Let the tailer take its preexisting-file snapshot before we write, so the
        // transcript is seen as created-during-the-session (50 ms is a safe margin
        // over the snapshot's single read_dir).
        std::thread::sleep(Duration::from_millis(50));
        (handle, stop, slug_dir)
    }

    /// Write a transcript file into a slug dir (the "appears during the session"
    /// step of a tailer test).
    fn write_transcript(slug_dir: &Path, name: &str, content: &str) -> PathBuf {
        let path = slug_dir.join(name);
        std::fs::write(&path, content).unwrap();
        path
    }

    /// Drain `handle.events` until EOF, collecting events by kind.
    fn drain(events: &mpsc::Receiver<Event>) -> Vec<Event> {
        let mut out = Vec::new();
        while let Ok(ev) = events.recv() {
            out.push(ev);
        }
        out
    }

    const FIXTURE: &str = "\
{\"type\":\"system\",\"cwd\":\"/tmp/work\",\"sessionId\":\"s\",\"timestamp\":\"2026-07-02T06:14:40.500Z\"}
{\"type\":\"user\",\"isSidechain\":false,\"isMeta\":true,\"message\":{\"role\":\"user\",\"content\":\"caveat\"},\"timestamp\":\"2026-07-02T06:14:40.600Z\",\"cwd\":\"/tmp/work\",\"sessionId\":\"s\"}
{\"type\":\"user\",\"isSidechain\":false,\"isMeta\":false,\"message\":{\"role\":\"user\",\"content\":\"hello\"},\"timestamp\":\"2026-07-02T06:14:40.699Z\",\"cwd\":\"/tmp/work\",\"sessionId\":\"s\"}
{\"type\":\"assistant\",\"isSidechain\":true,\"isMeta\":false,\"message\":{\"role\":\"assistant\",\"content\":[{\"type\":\"text\",\"text\":\"side\"}]},\"timestamp\":\"2026-07-02T06:14:41.000Z\",\"cwd\":\"/tmp/work\",\"sessionId\":\"s\"}
{\"type\":\"assistant\",\"isSidechain\":false,\"isMeta\":false,\"message\":{\"role\":\"assistant\",\"model\":\"glm-5.2\",\"content\":[{\"type\":\"tool_use\",\"id\":\"call_1\",\"name\":\"Bash\",\"input\":{\"command\":\"ls\"}}]},\"timestamp\":\"2026-07-02T06:14:41.500Z\",\"cwd\":\"/tmp/work\",\"sessionId\":\"s\"}
{\"type\":\"user\",\"isSidechain\":false,\"isMeta\":false,\"message\":{\"role\":\"user\",\"content\":[{\"type\":\"tool_result\",\"tool_use_id\":\"call_1\",\"content\":\"ok\",\"is_error\":false}]},\"timestamp\":\"2026-07-02T06:14:42.800Z\",\"cwd\":\"/tmp/work\",\"sessionId\":\"s\"}
";

    #[test]
    fn tailer_parses_fixture_and_skips_filtered_records() {
        let home = tempfile::TempDir::new().unwrap();
        let cwd = Path::new("/tmp/work");
        let (handle, stop, slug_dir) = spawn_with(home.path(), cwd);
        // Transcript appears mid-session (after the tailer snapshotted an empty
        // slug dir) — the realistic lazy-creation case the locate fix targets.
        write_transcript(&slug_dir, "session.jsonl", FIXTURE);
        // Give the tailer a poll cycle to find + parse it, then stop + drain.
        std::thread::sleep(Duration::from_millis(300));
        stop.store(true, Ordering::Release);
        let events = drain(&handle.events);
        // system + isMeta + isSidechain skipped → user_message + tool_call + tool_result.
        let kinds: Vec<_> = events.iter().map(|e| e.kind).collect();
        assert_eq!(
            kinds,
            vec![
                EventKind::UserMessage,
                EventKind::ToolCall,
                EventKind::ToolResult
            ]
        );
    }

    /// New-format Claude Code transcript (regression fixture, generation 2):
    /// recent Claude versions add top-level `mode` / `permissionMode` /
    /// `fileHistorySnapshot` fields and — critically — do **not** put `cwd` on
    /// the early `system`/meta records. The locate logic must scan past the
    /// cwd-less records to the first cwd-bearing one, and `parse_record` must
    /// tolerate the new fields. This is the format the "silently recorded 0
    /// steps" sessions were actually writing, so it is the load-bearing fixture
    /// for the adapter fix.
    const FIXTURE_NEW_FORMAT: &str = "\
{\"type\":\"system\",\"subtype\":\"init\",\"sessionId\":\"s\",\"timestamp\":\"2026-07-02T06:14:40.500Z\",\"mode\":\"default\",\"permissionMode\":\"default\"}
{\"type\":\"user\",\"isSidechain\":false,\"isMeta\":true,\"message\":{\"role\":\"user\",\"content\":\"caveat\"},\"timestamp\":\"2026-07-02T06:14:40.600Z\",\"mode\":\"default\",\"permissionMode\":\"default\"}
{\"type\":\"user\",\"isSidechain\":false,\"isMeta\":false,\"message\":{\"role\":\"user\",\"content\":\"hello\"},\"timestamp\":\"2026-07-02T06:14:40.699Z\",\"cwd\":\"/tmp/work-new\",\"sessionId\":\"s\",\"mode\":\"default\",\"permissionMode\":\"default\",\"fileHistorySnapshot\":{}}
{\"type\":\"assistant\",\"isSidechain\":false,\"isMeta\":false,\"message\":{\"role\":\"assistant\",\"model\":\"glm-5.2\",\"content\":[{\"type\":\"tool_use\",\"id\":\"call_1\",\"name\":\"Bash\",\"input\":{\"command\":\"ls\"}}]},\"timestamp\":\"2026-07-02T06:14:41.500Z\",\"cwd\":\"/tmp/work-new\",\"sessionId\":\"s\",\"mode\":\"default\",\"permissionMode\":\"default\"}
{\"type\":\"user\",\"isSidechain\":false,\"isMeta\":false,\"message\":{\"role\":\"user\",\"content\":[{\"type\":\"tool_result\",\"tool_use_id\":\"call_1\",\"content\":\"ok\",\"is_error\":false}]},\"timestamp\":\"2026-07-02T06:14:42.800Z\",\"cwd\":\"/tmp/work-new\",\"sessionId\":\"s\",\"mode\":\"default\",\"permissionMode\":\"default\"}
";

    /// `first_record_cwd_state` must scan *past* cwd-less records (the new format
    /// puts no `cwd` on `system`/meta lines) to the first cwd-bearing one, then
    /// classify it. A regression guard for the "no-early-cwd" case: a naive
    /// "look at the first record" check would return `None` forever and the
    /// locate loop would never accept the transcript → 0 recorded steps.
    #[test]
    fn first_record_cwd_state_skips_cwdless_records() {
        let tmp = tempfile::tempdir().unwrap();
        let f = tmp.path().join("new.jsonl");
        std::fs::write(&f, FIXTURE_NEW_FORMAT).unwrap();
        // The first cwd-bearing record carries /tmp/work-new → Matches.
        assert_eq!(
            first_record_cwd_state(&f, Path::new("/tmp/work-new")),
            CwdState::Matches
        );
        // A different session cwd → Different (rejected, not mis-accepted).
        assert_eq!(
            first_record_cwd_state(&f, Path::new("/tmp/other")),
            CwdState::Different
        );
    }

    /// End-to-end tailer regression against the new-format fixture: the adapter
    /// must locate the transcript (cwd appears only on record 3, not record 1)
    /// and parse the same three events as the original fixture. Locks that the
    /// new top-level fields and the cwd-less head do not silently drop the
    /// session to 0 steps.
    #[test]
    fn tailer_parses_new_format_fixture_with_no_early_cwd() {
        let home = tempfile::TempDir::new().unwrap();
        let cwd = Path::new("/tmp/work-new");
        let (handle, stop, slug_dir) = spawn_with(home.path(), cwd);
        write_transcript(&slug_dir, "session.jsonl", FIXTURE_NEW_FORMAT);
        std::thread::sleep(Duration::from_millis(300));
        stop.store(true, Ordering::Release);
        let events = drain(&handle.events);
        // Same three events as the original fixture: system/meta skipped, the
        // cwd-less head does not prevent locate from accepting the file.
        let kinds: Vec<_> = events.iter().map(|e| e.kind).collect();
        assert_eq!(
            kinds,
            vec![
                EventKind::UserMessage,
                EventKind::ToolCall,
                EventKind::ToolResult
            ],
            "new-format fixture must parse the same events as the original"
        );
    }

    #[test]
    fn tailer_finds_file_late() {
        let home = tempfile::TempDir::new().unwrap();
        let cwd = Path::new("/tmp/work-late");
        // Pre-create the slug dir but write the file *after* spawning, so the
        // adapter must keep polling until it appears (no fixed timeout).
        std::fs::create_dir_all(
            home.path()
                .join(".claude")
                .join("projects")
                .join(slugify(cwd)),
        )
        .unwrap();
        let stop = std::sync::Arc::new(AtomicBool::new(false));
        let blobs = std::sync::Arc::new(BlobStore::new(home.path().join("blobs")));
        let ctx = AdapterContext {
            session_id: SID.to_string(),
            started_at_unix_ms: STARTED,
            cwd: cwd.to_path_buf(),
            command: vec!["claude".into()],
            blobs,
            stop: std::sync::Arc::clone(&stop),
            projects_dir: Some(home.path().join(".claude").join("projects")),
        };
        let handle = Box::new(ClaudeAdapter).spawn(ctx).expect("spawn");
        // Write the transcript after the adapter is already polling.
        std::thread::sleep(Duration::from_millis(150));
        std::fs::write(
            home
                .path()
                .join(".claude")
                .join("projects")
                .join(slugify(cwd))
                .join("late.jsonl"),
            "{\"type\":\"user\",\"isSidechain\":false,\"isMeta\":false,\"message\":{\"role\":\"user\",\"content\":\"late\"},\"timestamp\":\"2026-07-02T06:14:40.699Z\",\"cwd\":\"/tmp/work-late\",\"sessionId\":\"s\"}\n",
        )
        .unwrap();
        std::thread::sleep(Duration::from_millis(150));
        stop.store(true, Ordering::Release);
        let events = drain(&handle.events);
        assert_eq!(
            events.len(),
            1,
            "adapter should pick up the late-appearing file"
        );
        assert_eq!(events[0].kind, EventKind::UserMessage);
    }

    #[test]
    fn tailer_partial_last_line_retries() {
        let home = tempfile::TempDir::new().unwrap();
        let cwd = Path::new("/tmp/work-partial");
        let (handle, stop, slug_dir) = spawn_with(home.path(), cwd);
        // A file with one complete line plus a trailing partial (no newline),
        // written mid-session so the tailer sees it as new.
        let complete = "{\"type\":\"user\",\"isSidechain\":false,\"isMeta\":false,\"message\":{\"role\":\"user\",\"content\":\"first\"},\"timestamp\":\"2026-07-02T06:14:40.699Z\",\"cwd\":\"/tmp/work-partial\",\"sessionId\":\"s\"}\n";
        let partial = "{\"type\":\"user\",\"isSidechain\":false,\"isMeta\":false,\"message\":{\"role\":\"user\",\"content\":\"sec";
        let path = write_transcript(&slug_dir, "partial.jsonl", &format!("{complete}{partial}"));
        // Let the tailer find the file and read the complete line (partial held).
        std::thread::sleep(Duration::from_millis(250));
        // Append the rest of the partial line; it should now parse as a second event.
        let mut f = std::fs::OpenOptions::new()
            .append(true)
            .open(&path)
            .unwrap();
        writeln!(f, "ond\"}}}}\n").unwrap();
        std::thread::sleep(Duration::from_millis(250));
        stop.store(true, Ordering::Release);
        let events = drain(&handle.events);
        let contents: Vec<&str> = events
            .iter()
            .map(|e| e.body_json.as_ref().unwrap()["text"].as_str().unwrap())
            .collect();
        assert_eq!(
            contents,
            vec!["first", "second"],
            "partial line must be retried until complete"
        );
    }

    #[test]
    fn tailer_fallback_scan_by_cwd() {
        let home = tempfile::TempDir::new().unwrap();
        let cwd = Path::new("/tmp/work-fallback");
        // Put the transcript under a *wrong* slug dir (not slugify(cwd)) so the
        // slug lookup misses and the fallback scan by cwd must find it.
        let wrong_slug = "-totally-wrong-slug";
        let wrong_dir = home
            .path()
            .join(".claude")
            .join("projects")
            .join(wrong_slug);
        std::fs::create_dir_all(&wrong_dir).unwrap();
        let stop = std::sync::Arc::new(AtomicBool::new(false));
        let blobs = std::sync::Arc::new(BlobStore::new(home.path().join("blobs")));
        let ctx = AdapterContext {
            session_id: SID.to_string(),
            started_at_unix_ms: STARTED,
            cwd: cwd.to_path_buf(),
            command: vec!["claude".into()],
            blobs,
            stop: std::sync::Arc::clone(&stop),
            projects_dir: Some(home.path().join(".claude").join("projects")),
        };
        let handle = Box::new(ClaudeAdapter).spawn(ctx).expect("spawn");
        // Write the transcript under the wrong slug AFTER spawn (so it is seen as
        // new) and after the preexisting snapshot (50 ms margin).
        std::thread::sleep(Duration::from_millis(50));
        std::fs::write(
            wrong_dir.join("fb.jsonl"),
            "{\"type\":\"user\",\"isSidechain\":false,\"isMeta\":false,\"message\":{\"role\":\"user\",\"content\":\"fb\"},\"timestamp\":\"2026-07-02T06:14:40.699Z\",\"cwd\":\"/tmp/work-fallback\",\"sessionId\":\"s\"}\n",
        )
        .unwrap();
        std::thread::sleep(Duration::from_millis(300));
        stop.store(true, Ordering::Release);
        let events = drain(&handle.events);
        assert_eq!(
            events.len(),
            1,
            "fallback scan should locate the transcript by cwd"
        );
        assert_eq!(events[0].kind, EventKind::UserMessage);
        let outcome = handle.outcome.join().expect("tailer thread");
        assert_eq!(outcome.status, AdapterStatus::Active);
    }

    #[test]
    fn tailer_degrades_when_projects_dir_missing() {
        // No ~/.claude/projects anywhere: HOME points at an empty temp dir.
        let home = tempfile::TempDir::new().unwrap();
        let cwd = Path::new("/tmp/nowhere");
        let stop = std::sync::Arc::new(AtomicBool::new(false));
        let blobs = std::sync::Arc::new(BlobStore::new(home.path().join("blobs")));
        let ctx = AdapterContext {
            session_id: SID.to_string(),
            started_at_unix_ms: STARTED,
            cwd: cwd.to_path_buf(),
            command: vec!["claude".into()],
            blobs,
            stop: std::sync::Arc::clone(&stop),
            projects_dir: Some(home.path().join(".claude").join("projects")),
        };
        let handle = Box::new(ClaudeAdapter).spawn(ctx).expect("spawn");
        let events = drain(&handle.events);
        assert!(events.is_empty(), "no projects dir → no events");
        let outcome = handle.outcome.join().expect("tailer thread");
        assert_eq!(outcome.status, AdapterStatus::Degraded);
        // Degraded outcomes carry an actionable reason for the recorder to print
        // after the child exits (FR-1.5), not a bare status with no explanation.
        assert!(
            outcome.degrade_reason.is_some(),
            "degraded outcome must carry a degrade_reason"
        );
    }

    #[test]
    fn tailer_degrades_when_transcript_never_appears() {
        // Projects dir exists, slug dir exists, but no transcript is ever written:
        // the tailer must poll until stop, then return Degraded with a reason
        // (the "0 steps, status=ok" failure mode — previously the 3 s deadline
        // degraded silently mid-session under the agent's TUI).
        let home = tempfile::TempDir::new().unwrap();
        let cwd = Path::new("/tmp/no-transcript");
        let (handle, stop, _slug_dir) = spawn_with(home.path(), cwd);
        // Never write a transcript. Stop almost immediately.
        std::thread::sleep(Duration::from_millis(80));
        stop.store(true, Ordering::Release);
        let events = drain(&handle.events);
        assert!(events.is_empty(), "no transcript ever → no events");
        let outcome = handle.outcome.join().expect("tailer thread");
        assert_eq!(outcome.status, AdapterStatus::Degraded);
        assert!(
            outcome.degrade_reason.is_some(),
            "degraded outcome must carry a degrade_reason pointing at hh doctor"
        );
        assert!(
            outcome
                .degrade_reason
                .as_deref()
                .unwrap()
                .contains("hh doctor"),
            "degrade reason should suggest running `hh doctor`"
        );
    }

    #[test]
    fn tailer_ignores_preexisting_transcript() {
        // A transcript that predates `hh run` belongs to a *different* session and
        // must NOT be tailed, even if it matches cwd (concurrent-session guard).
        let home = tempfile::TempDir::new().unwrap();
        let cwd = Path::new("/tmp/concurrent");
        let slug_dir = home
            .path()
            .join(".claude")
            .join("projects")
            .join(slugify(cwd));
        std::fs::create_dir_all(&slug_dir).unwrap();
        // Pre-existing (concurrent session's) transcript, written BEFORE spawn.
        std::fs::write(
            slug_dir.join("old.jsonl"),
            "{\"type\":\"user\",\"isSidechain\":false,\"isMeta\":false,\"message\":{\"role\":\"user\",\"content\":\"not mine\"},\"timestamp\":\"2026-07-02T06:14:40.699Z\",\"cwd\":\"/tmp/concurrent\",\"sessionId\":\"other\"}\n",
        )
        .unwrap();
        let (handle, stop, _slug) = spawn_with(home.path(), cwd);
        // No new transcript is ever written for our session.
        std::thread::sleep(Duration::from_millis(80));
        stop.store(true, Ordering::Release);
        let events = drain(&handle.events);
        assert!(
            events.is_empty(),
            "a pre-existing transcript from another session must not be tailed"
        );
        let outcome = handle.outcome.join().expect("tailer thread");
        assert_eq!(outcome.status, AdapterStatus::Degraded);
    }

    /// A candidate path that is a directory (the same-stem-dir issue: Claude
    /// writes both `<uuid>.jsonl` and a `<uuid>/` directory) is classified as
    /// [`CwdState::Directory`] and rejected, not re-polled forever (which would
    /// loop on the `read_to_string` io error as `None` and silently degrade).
    #[test]
    fn first_record_cwd_state_rejects_directory() {
        let tmp = tempfile::tempdir().unwrap();
        let dir = tmp.path().join("2457c4b0-is-a-dir");
        std::fs::create_dir_all(&dir).unwrap();
        assert_eq!(
            first_record_cwd_state(&dir, Path::new("/tmp/anything")),
            CwdState::Directory,
            "a directory candidate must be rejected, not re-polled"
        );
    }

    /// A same-stem directory next to the real `.jsonl` transcript must not break
    /// discovery: the `.jsonl` extension filter skips the directory, the file is
    /// selected by cwd, and events are parsed normally. Regression guard for the
    /// issue the user flagged (a `<uuid>/` dir beside `<uuid>.jsonl`).
    #[test]
    fn tailer_ignores_same_stem_directory_beside_jsonl() {
        let home = tempfile::TempDir::new().unwrap();
        let cwd = Path::new("/tmp/same-stem");
        let (handle, stop, slug_dir) = spawn_with(home.path(), cwd);
        // The real transcript.
        write_transcript(
            &slug_dir,
            "abc123.jsonl",
            "{\"type\":\"user\",\"isSidechain\":false,\"isMeta\":false,\"message\":{\"role\":\"user\",\"content\":\"hi\"},\"timestamp\":\"2026-07-02T06:14:40.699Z\",\"cwd\":\"/tmp/same-stem\",\"sessionId\":\"s\"}\n",
        );
        // A same-stem directory (no .jsonl extension) sitting beside it — must
        // be skipped by the extension filter, not break the dir iteration.
        std::fs::create_dir_all(slug_dir.join("abc123")).unwrap();
        std::thread::sleep(Duration::from_millis(300));
        stop.store(true, Ordering::Release);
        let events = drain(&handle.events);
        assert_eq!(
            events.len(),
            1,
            "the same-stem directory must not prevent discovering the .jsonl"
        );
        assert_eq!(events[0].kind, EventKind::UserMessage);
        let outcome = handle.outcome.join().expect("tailer thread");
        assert_eq!(outcome.status, AdapterStatus::Active);
    }

    /// A transcript that is found (cwd matches) but yields zero events — here
    /// because the only cwd-bearing record is `isMeta` (skipped by the parser) —
    /// must degrade with the specific "0 records parsed" reason, NOT finalize as
    /// `active` with 0 steps (the silent-breakage symptom). The reason carries
    /// the line count so the failure is a one-line diagnosis.
    #[test]
    fn tailer_degrades_when_zero_records_parsed() {
        let home = tempfile::TempDir::new().unwrap();
        let cwd = Path::new("/tmp/zero-records");
        let (handle, stop, slug_dir) = spawn_with(home.path(), cwd);
        // cwd-bearing so it is selected, but isMeta so parse_record skips it →
        // 0 events. Mirrors a format drift where the parser recognizes nothing.
        write_transcript(
            &slug_dir,
            "empty.jsonl",
            "{\"type\":\"user\",\"isSidechain\":false,\"isMeta\":true,\"message\":{\"role\":\"user\",\"content\":\"caveat\"},\"timestamp\":\"2026-07-02T06:14:40.699Z\",\"cwd\":\"/tmp/zero-records\",\"sessionId\":\"s\"}\n",
        );
        std::thread::sleep(Duration::from_millis(300));
        stop.store(true, Ordering::Release);
        let events = drain(&handle.events);
        assert!(events.is_empty(), "isMeta record yields no events");
        let outcome = handle.outcome.join().expect("tailer thread");
        assert_eq!(
            outcome.status,
            AdapterStatus::Degraded,
            "0 records parsed must degrade, not report active with 0 steps"
        );
        let reason = outcome.degrade_reason.as_deref().unwrap();
        assert!(
            reason.contains("0 records parsed"),
            "reason must be the specific '0 records parsed' string: {reason}"
        );
        assert!(
            reason.contains("read 1 line"),
            "reason must carry the line count: {reason}"
        );
    }

    /// A transcript whose first JSON line is malformed records the first parse
    /// error (line + message) in the degrade reason, so a format drift is a
    /// one-line diagnosis instead of a silent skip.
    #[test]
    fn tailer_degrade_reason_records_first_parse_error() {
        let home = tempfile::TempDir::new().unwrap();
        let cwd = Path::new("/tmp/parse-err");
        let (handle, stop, slug_dir) = spawn_with(home.path(), cwd);
        // A valid cwd-bearing user record (so it is selected) followed by a
        // malformed line. The valid record yields 1 event → Active, so to
        // exercise the parse-error-in-reason path we make the ONLY line a
        // cwd-bearing record that is unparseable JSON. But cwd-verification
        // needs parseable JSON to read `cwd`... so instead place a valid
        // selected record then a broken one: the file is selected on the valid
        // record, the broken line is skipped with a debug log, and since the
        // valid record produced an event the outcome is Active. The parse-error
        // tracking is therefore asserted at the unit level below instead.
        write_transcript(
            &slug_dir,
            "broken.jsonl",
            "{\"type\":\"user\",\"isSidechain\":false,\"isMeta\":false,\"message\":{\"role\":\"user\",\"content\":\"ok\"},\"timestamp\":\"2026-07-02T06:14:40.699Z\",\"cwd\":\"/tmp/parse-err\",\"sessionId\":\"s\"}\n{not valid json\n",
        );
        std::thread::sleep(Duration::from_millis(300));
        stop.store(true, Ordering::Release);
        let events = drain(&handle.events);
        assert_eq!(
            events.len(),
            1,
            "the valid record parses; the broken line is skipped"
        );
        let outcome = handle.outcome.join().expect("tailer thread");
        assert_eq!(outcome.status, AdapterStatus::Active);
    }

    /// `locate_failure_reason` produces the specific "no new transcript
    /// appeared" string when nothing was seen, and the "N candidates but none
    /// matched cwd" string (with mismatch/directory counts) when candidates
    /// were rejected. Locks the one-line-diagnosis contract for degraded
    /// sessions without spinning a real tailer.
    #[test]
    fn locate_failure_reason_is_specific() {
        // No candidates appeared at all.
        let empty = DiscoveryDiag {
            projects: Some(PathBuf::from("/home/me/.claude/projects")),
            slug: Some("-home-me-work".into()),
            slug_dir: Some(PathBuf::from("/home/me/.claude/projects/-home-me-work")),
            cwd: Some(PathBuf::from("/home/me/work")),
            ..Default::default()
        };
        let r = locate_failure_reason(&empty);
        assert!(r.contains("no jsonl matched cwd slug -home-me-work"), "{r}");
        assert!(r.contains("no new transcript appeared"), "{r}");
        assert!(
            r.contains("looked in /home/me/.claude/projects/-home-me-work"),
            "{r}"
        );

        // Candidates appeared but none matched cwd; one was a directory.
        let mut seen = std::collections::HashSet::new();
        seen.insert(PathBuf::from(
            "/home/me/.claude/projects/-home-me-work/a.jsonl",
        ));
        let with_cands = DiscoveryDiag {
            projects: Some(PathBuf::from("/home/me/.claude/projects")),
            slug: Some("-home-me-work".into()),
            slug_dir: Some(PathBuf::from("/home/me/.claude/projects/-home-me-work")),
            cwd: Some(PathBuf::from("/home/me/work")),
            new_candidates: seen,
            cwd_mismatches: vec![PathBuf::from(
                "/home/me/.claude/projects/-home-me-work/a.jsonl",
            )],
            directories: vec![PathBuf::from(
                "/home/me/.claude/projects/-home-me-work/b.jsonl",
            )],
        };
        let r = locate_failure_reason(&with_cands);
        assert!(r.contains("1 new candidate(s)"), "{r}");
        assert!(r.contains("1 rejected for a different cwd"), "{r}");
        assert!(r.contains("1 candidate(s) were directories"), "{r}");
    }

    // -----------------------------------------------------------------------
    // Codex CLI parser tests
    // -----------------------------------------------------------------------

    fn parse_codex(value: &Value, ts_ms: i64) -> ParsedRecord {
        let tmp = tempfile::TempDir::new().unwrap();
        let blobs = BlobStore::new(tmp.path().join("blobs"));
        parse_codex_record(value, SID, ts_ms, &blobs)
    }

    fn snap_codex(pr: &ParsedRecord) -> String {
        serde_json::to_string_pretty(&serde_json::json!({
            "events": pr.events,
            "is_assistant_message": pr.is_assistant_message,
        }))
        .unwrap()
    }

    fn codex_rec(json: &str) -> Value {
        serde_json::from_str(json).unwrap()
    }

    #[test]
    fn codex_parse_user_message() {
        let v = codex_rec(
            r#"{"type":"event_msg","timestamp":"2026-07-02T06:14:40.699Z","payload":{"type":"user_message","content":[{"text":"hello"}]}}"#,
        );
        let pr = parse_codex(&v, iso("2026-07-02T06:14:40.699Z"));
        insta::assert_snapshot!(snap_codex(&pr));
        assert_eq!(pr.events.len(), 1);
        assert_eq!(pr.events[0].kind, EventKind::UserMessage);
    }

    #[test]
    fn codex_parse_agent_message() {
        let v = codex_rec(
            r#"{"type":"event_msg","timestamp":"2026-07-02T06:14:41.500Z","payload":{"type":"agent_message","content":[{"text":"I can help with that"}]}}"#,
        );
        let pr = parse_codex(&v, iso("2026-07-02T06:14:41.500Z"));
        insta::assert_snapshot!(snap_codex(&pr));
        assert_eq!(pr.events.len(), 1);
        assert_eq!(pr.events[0].kind, EventKind::AgentMessage);
        assert!(pr.is_assistant_message);
    }

    #[test]
    fn codex_parse_response_item_message() {
        let v = codex_rec(
            r#"{"type":"response_item","timestamp":"2026-07-02T06:14:41.500Z","payload":{"type":"message","role":"assistant","content":[{"type":"output_text","text":"Here is the solution"}]}}"#,
        );
        let pr = parse_codex(&v, iso("2026-07-02T06:14:41.500Z"));
        insta::assert_snapshot!(snap_codex(&pr));
        assert_eq!(pr.events.len(), 1);
        assert_eq!(pr.events[0].kind, EventKind::AgentMessage);
    }

    #[test]
    fn codex_parse_function_call_and_output() {
        let v = codex_rec(
            r#"{"type":"response_item","timestamp":"2026-07-02T06:14:42.000Z","payload":{"type":"function_call","call_id":"call_1","name":"Bash","arguments":"{\"command\":\"ls\"}"}}"#,
        );
        let pr = parse_codex(&v, iso("2026-07-02T06:14:42.000Z"));
        insta::assert_snapshot!(snap_codex(&pr));
        assert_eq!(pr.events.len(), 1);
        assert_eq!(pr.events[0].kind, EventKind::ToolCall);
        assert_eq!(
            pr.events[0].body_json.as_ref().unwrap()["correlate_key"],
            "call_1"
        );

        let v2 = codex_rec(
            r#"{"type":"response_item","timestamp":"2026-07-02T06:14:43.000Z","payload":{"type":"function_call_output","call_id":"call_1","output":"total 42"}}"#,
        );
        let pr2 = parse_codex(&v2, iso("2026-07-02T06:14:43.000Z"));
        assert_eq!(pr2.events.len(), 1);
        assert_eq!(pr2.events[0].kind, EventKind::ToolResult);
        assert_eq!(
            pr2.events[0].body_json.as_ref().unwrap()["correlate_key"],
            "call_1"
        );
    }

    #[test]
    fn codex_parse_reasoning() {
        let v = codex_rec(
            r#"{"type":"response_item","timestamp":"2026-07-02T06:14:41.000Z","payload":{"type":"reasoning","summary":[],"content":null,"encrypted_content":"gAAAAABp5lf5..."}}"#,
        );
        let pr = parse_codex(&v, iso("2026-07-02T06:14:41.000Z"));
        insta::assert_snapshot!(snap_codex(&pr));
        assert_eq!(pr.events.len(), 1);
        assert_eq!(pr.events[0].kind, EventKind::Thinking);
    }

    #[test]
    fn codex_parse_skips_session_meta() {
        let v = codex_rec(
            r#"{"type":"session_meta","timestamp":"2026-07-02T06:14:40.000Z","payload":{"id":"abc","cwd":"/tmp","originator":"codex-tui"}}"#,
        );
        let pr = parse_codex(&v, 0);
        assert!(pr.events.is_empty());
    }

    #[test]
    fn codex_parse_skips_unknown_type() {
        let v = codex_rec(
            r#"{"type":"unknown_type","timestamp":"2026-07-02T06:14:40.000Z","payload":{}}"#,
        );
        let pr = parse_codex(&v, 0);
        assert!(pr.events.is_empty());
    }

    #[test]
    fn codex_parse_non_object_value() {
        for v in [codex_rec("[]"), codex_rec("null"), codex_rec("\"oops\"")] {
            assert!(parse_codex(&v, 0).events.is_empty());
        }
    }

    // -----------------------------------------------------------------------
    // Gemini CLI parser tests
    // -----------------------------------------------------------------------

    fn parse_gemini(value: &Value, ts_ms: i64) -> ParsedRecord {
        let tmp = tempfile::TempDir::new().unwrap();
        let blobs = BlobStore::new(tmp.path().join("blobs"));
        parse_gemini_record(value, SID, ts_ms, &blobs)
    }

    fn snap_gemini(pr: &ParsedRecord) -> String {
        serde_json::to_string_pretty(&serde_json::json!({
            "events": pr.events,
            "is_assistant_message": pr.is_assistant_message,
        }))
        .unwrap()
    }

    fn gemini_rec(json: &str) -> Value {
        serde_json::from_str(json).unwrap()
    }

    #[test]
    fn gemini_parse_user_message() {
        let v = gemini_rec(
            r#"{"type":"user","id":"msg1","timestamp":"2026-07-02T06:14:40.699Z","content":[{"text":"hello"}]}"#,
        );
        let pr = parse_gemini(&v, iso("2026-07-02T06:14:40.699Z"));
        insta::assert_snapshot!(snap_gemini(&pr));
        assert_eq!(pr.events.len(), 1);
        assert_eq!(pr.events[0].kind, EventKind::UserMessage);
    }

    #[test]
    fn gemini_parse_agent_message() {
        let v = gemini_rec(
            r#"{"type":"gemini","id":"msg2","timestamp":"2026-07-02T06:14:41.500Z","content":[{"text":"Hi! How can I help?"}],"model":"gemini-2.5-pro"}"#,
        );
        let pr = parse_gemini(&v, iso("2026-07-02T06:14:41.500Z"));
        insta::assert_snapshot!(snap_gemini(&pr));
        assert_eq!(pr.events.len(), 1);
        assert_eq!(pr.events[0].kind, EventKind::AgentMessage);
    }

    #[test]
    fn gemini_parse_with_tool_calls() {
        let v = gemini_rec(
            r#"{"type":"gemini","id":"msg3","timestamp":"2026-07-02T06:14:42.000Z","content":[{"text":"Let me check"}],"toolCalls":[{"id":"tc_1","name":"Bash","input":{"command":"ls"}}]}"#,
        );
        let pr = parse_gemini(&v, iso("2026-07-02T06:14:42.000Z"));
        insta::assert_snapshot!(snap_gemini(&pr));
        assert_eq!(pr.events.len(), 2); // agent_message + tool_call
        assert_eq!(pr.events[0].kind, EventKind::AgentMessage);
        assert_eq!(pr.events[1].kind, EventKind::ToolCall);
        assert_eq!(
            pr.events[1].body_json.as_ref().unwrap()["correlate_key"],
            "tc_1"
        );
    }

    #[test]
    fn gemini_parse_with_thoughts() {
        let v = gemini_rec(
            r#"{"type":"gemini","id":"msg4","timestamp":"2026-07-02T06:14:41.000Z","content":[{"text":"answer"}],"thoughts":[{"text":"I need to think about this"}]}"#,
        );
        let pr = parse_gemini(&v, iso("2026-07-02T06:14:41.000Z"));
        insta::assert_snapshot!(snap_gemini(&pr));
        assert_eq!(pr.events.len(), 2); // agent_message + thinking
        assert_eq!(pr.events[0].kind, EventKind::AgentMessage);
        assert_eq!(pr.events[1].kind, EventKind::Thinking);
    }

    #[test]
    fn gemini_parse_skips_session_metadata() {
        let v = gemini_rec(
            r#"{"type":"session_metadata","sessionId":"abc","projectHash":"xyz","startTime":"2026-07-02T06:14:40.000Z"}"#,
        );
        let pr = parse_gemini(&v, 0);
        assert!(pr.events.is_empty());
    }

    #[test]
    fn gemini_parse_skips_message_update() {
        let v = gemini_rec(
            r#"{"type":"message_update","id":"msg2","timestamp":"2026-07-02T06:14:41.000Z","tokens":{"input":5,"output":4}}"#,
        );
        let pr = parse_gemini(&v, 0);
        assert!(pr.events.is_empty());
    }

    #[test]
    fn gemini_parse_skips_unknown_type() {
        let v = gemini_rec(r#"{"type":"unknown","id":"x","timestamp":"2026-07-02T06:14:40.000Z"}"#);
        let pr = parse_gemini(&v, 0);
        assert!(pr.events.is_empty());
    }

    #[test]
    fn gemini_parse_non_object_value() {
        for v in [gemini_rec("[]"), gemini_rec("null")] {
            assert!(parse_gemini(&v, 0).events.is_empty());
        }
    }

    // -----------------------------------------------------------------------
    // Detection tests for all adapters
    // -----------------------------------------------------------------------

    #[test]
    fn select_detects_all_adapters() {
        assert!(select(&["claude".into()], Path::new("/tmp")).is_some());
        assert!(select(&["claude-desktop".into()], Path::new("/tmp")).is_some());
        assert!(select(&["codex".into()], Path::new("/tmp")).is_some());
        assert!(select(&["gemini".into()], Path::new("/tmp")).is_some());
        assert!(select(&["python3".into()], Path::new("/tmp")).is_none());
    }

    #[test]
    fn is_claude_desktop_detects_basename() {
        assert!(is_claude_desktop(&["claude-desktop".into()]));
        assert!(is_claude_desktop(&["/usr/local/bin/claude-desktop".into()]));
        assert!(is_claude_desktop(&["claude-desktop.exe".into()]));
        assert!(!is_claude_desktop(&["claude".into()]));
        assert!(!is_claude_desktop(&["codex".into()]));
        assert!(!is_claude_desktop(&[]));
    }

    #[test]
    fn is_codex_cli_detects_basename() {
        assert!(is_codex_cli(&["codex".into()]));
        assert!(is_codex_cli(&["/usr/local/bin/codex".into()]));
        assert!(is_codex_cli(&["codex.exe".into()]));
        assert!(!is_codex_cli(&["claude".into()]));
        assert!(!is_codex_cli(&[]));
    }

    #[test]
    fn is_gemini_cli_detects_basename() {
        assert!(is_gemini_cli(&["gemini".into()]));
        assert!(is_gemini_cli(&["/usr/local/bin/gemini".into()]));
        assert!(is_gemini_cli(&["gemini.exe".into()]));
        assert!(!is_gemini_cli(&["claude".into()]));
        assert!(!is_gemini_cli(&[]));
    }

    #[test]
    fn resolve_override_accepts_all_adapters() {
        assert!(resolve_override("claude-code").is_some());
        assert!(resolve_override("claude-desktop").is_some());
        assert!(resolve_override("codex-cli").is_some());
        assert!(resolve_override("gemini-cli").is_some());
        assert!(resolve_override("unknown").is_none());
    }
}