dv-review 0.95.18

//! Batch reviews from `dv list`: Claude reviews each marked PR in the background, in the PR's
//! worktree (`<repo>-N`, as `dv N -w` makes it), a couple at a time. Each gets the PR's own
//! session and review file, so opening the PR in dv later picks up where it left off.
use crate::claude::Subject;
use std::collections::VecDeque;
use std::path::{Path, PathBuf};
use std::sync::mpsc::{self, Receiver, Sender};
use std::sync::{Arc, Mutex};

/// How many PRs Claude reviews at once.
pub const AT_ONCE: usize = 2;

#[derive(Clone, Debug, PartialEq)]
pub enum Status {
    Queued,
    Preparing,
    Reviewing,
    /// What it took ("5 turns, $0.22").
    Done(String),
    /// Why, and when (a review of it written since ends it: `List::watch_reviews`).
    Failed(String, std::time::SystemTime),
    /// Not run, and why (Claude at it already, from another run, `claude::claim`'s holder; in
    /// the checkout it'd work in, that one): not a failure, asked again once that's over.
    Busy(String, Option<PathBuf>),
    /// Not run: the PR isn't open now (its state: merged or closed meanwhile), nothing to do.
    Gone(String),
}

impl Status {
    pub fn failed(why: impl Into<String>) -> Status {
        Status::Failed(why.into(), std::time::SystemTime::now())
    }
}

/// Whether what held run `n` back (`Busy`) is still there: Claude on it (its claim), or in the
/// checkout it'd work in (`dir`). The one test, for the run and for asking it again.
pub fn held(n: u64, dir: Option<&Path>) -> bool {
    crate::claude::at_work(&crate::claude::pr_key(n), None, None).is_some() || dir.is_some_and(|d| crate::claude::at_work_in(d, None).is_some())
}

/// What Claude does on a PR: review it (or update the review it has: the prompt says to), update
/// the review from the commit it reviewed, or answer your replies in it (`theirs`: someone else's
/// PR, so suggestions, not edits).
#[derive(Clone, Debug, PartialEq)]
pub enum Task {
    Review,
    Update(String),
    Answer { theirs: bool },
    /// Your PR's CI fails: find out why, fix it in the checkout (you push).
    FixChecks,
    /// Claude's review would close it: draft the comment it closes with (you close it, in dv).
    Close,
    /// Your PR changed since this commit (the first) to its head now (the second, noted as
    /// checked once it's done): propose a new description if it no longer fits (you approve
    /// it, in dv).
    Describe(String, String),
}

impl Task {
    /// Its `Work` (its model: `dv.model.<key>`), as the PR view's same asks have.
    pub fn work(&self) -> crate::claude::Work {
        use crate::claude::Work;
        match self {
            Task::Review | Task::Update(_) => Work::Review,
            Task::Answer { .. } => Work::Answer,
            Task::FixChecks => Work::FixCi,
            Task::Close => Work::Close,
            Task::Describe(..) => Work::Describe,
        }
    }

    /// For the cards and messages: "reviewing", "updating its review", "answering".
    pub fn doing(&self) -> &'static str {
        match self {
            Task::Review => "reviewing",
            Task::Update(_) => "updating its review",
            Task::Answer { .. } => "answering",
            Task::FixChecks => "fixing CI",
            Task::Close => "drafting a closing comment",
            Task::Describe(..) => "checking its description",
        }
    }

    /// For a card, before it's asked: "could review", "could update its review", "could answer you".
    pub fn could(&self) -> &'static str {
        match self {
            Task::Review => "could review",
            Task::Update(_) => "could update its review",
            Task::Answer { .. } => "could answer you",
            Task::FixChecks => "could fix CI",
            Task::Close => "could draft a closing comment",
            Task::Describe(..) => "could check its description",
        }
    }

    fn prompt(&self, s: crate::claude::Subject) -> String {
        let (n, what) = (s.n(), s.what());
        match self {
            Task::Review => crate::claude::review_prompt(&what),
            Task::Update(old) => crate::claude::update_prompt(&what, old),
            Task::Answer { theirs } => crate::claude::answer_prompt(&what, *theirs),
            Task::FixChecks => crate::claude::fix_checks_prompt(n),
            Task::Close => format!(
                "Use the dv-review skill to draft a comment closing {what}, in the review file's top-level \"closing\" \
                 field (see \"Draft a closing comment\"): your review (or triage) says it should be closed. Don't close it \
                 or comment on GitHub: the user closes it in dv."
            ),
            Task::Describe(old, _) => format!(
                "{what} has changed since {old} (`git diff {old} HEAD`; that commit gone, a force-push: the whole PR, \
                 `gh pr diff {n}`). Read its description (`gh pr view {n} --json body`). \
                 If it no longer describes the PR (what it changes now, why, how it's tested, anything in it now wrong \
                 or missing), use the dv-review skill to propose a new one in the review file's top-level \
                 \"description\" field (see \"Propose a description\"), changing only what needs to, keeping the rest. \
                 If it still fits, change nothing. Don't edit the PR on GitHub: the user approves it in dv."
            ),
        }
    }
}

impl Status {
    pub fn running(&self) -> bool {
        matches!(self, Status::Queued | Status::Preparing | Status::Reviewing)
    }
}

/// Claude does `task` on each of `prs` (PR number, whether its worktree is on the PR's branch,
/// the task) from the repo at `root`, with the user's `extra` instructions (if any); progress
/// comes back on the receiver.
pub fn start(prs: Vec<(u64, bool, Task)>, root: PathBuf, extra: String) -> Receiver<(u64, Status)> {
    pool(prs, move |(n, branch, task), tx| (*n, run(Subject::Pr(*n), *branch, task, &root, &extra, tx).unwrap_or_else(|e| Status::failed(format!("{e:#}")))))
}

/// Claude does `task` on issue `n` (its closing comment, say), in the checkout at `root`: as
/// on a PR, its review file the issue's.
pub fn start_issue(n: u64, task: Task, root: PathBuf) -> Receiver<(u64, Status)> {
    pool(vec![n], move |n, tx| (*n, run(Subject::Issue(*n), false, &task, &root, "", tx).unwrap_or_else(|e| Status::failed(format!("{e:#}")))))
}

/// Something to do (a PR to review, an issue to triage): its number first.
pub trait Numbered {
    fn number(&self) -> u64;
}

impl Numbered for u64 {
    fn number(&self) -> u64 {
        *self
    }
}

impl Numbered for (u64, String) {
    fn number(&self) -> u64 {
        self.0
    }
}

impl Numbered for (u64, bool, Task) {
    fn number(&self) -> u64 {
        self.0
    }
}

/// One of the `AT_ONCE` Claude runs this dv may have going from its lists, across all its
/// pools, drafts and groupings (one asked while others run waits for them): held while a run's at work, given back as it ends.
pub(crate) struct Slot;

static SLOTS: (Mutex<usize>, std::sync::Condvar) = (Mutex::new(0), std::sync::Condvar::new());

impl Slot {
    pub(crate) fn take() -> Slot {
        let mut n = SLOTS.0.lock().unwrap_or_else(|e| e.into_inner());
        while *n >= AT_ONCE {
            n = SLOTS.1.wait(n).unwrap_or_else(|e| e.into_inner());
        }
        *n += 1;
        Slot
    }
}

impl Drop for Slot {
    fn drop(&mut self) {
        *SLOTS.0.lock().unwrap_or_else(|e| e.into_inner()) -= 1;
        SLOTS.1.notify_one();
    }
}

/// `items` done by `work` (which may say how it's going on the sender), `AT_ONCE` at a time,
/// in the background: each queued first, then its news and how it ended, by number.
pub fn pool<T: Numbered + Send + 'static>(items: Vec<T>, work: impl Fn(&T, &Sender<(u64, Status)>) -> (u64, Status) + Send + Sync + 'static) -> Receiver<(u64, Status)> {
    let (tx, rx) = mpsc::channel();
    items.iter().for_each(|i| _ = tx.send((i.number(), Status::Queued)));
    let (queue, work) = (Arc::new(Mutex::new(VecDeque::from(items))), Arc::new(work));
    for _ in 0..AT_ONCE {
        let (queue, tx, work) = (queue.clone(), tx.clone(), work.clone());
        std::thread::spawn(move || {
            while let Some(item) = queue.lock().ok().and_then(|mut q| q.pop_front()) {
                let _slot = Slot::take();
                // (A panic in one: it failed, and the worker goes on to the next.)
                let st = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| work(&item, &tx)));
                _ = tx.send(st.unwrap_or_else(|p| (item.number(), Status::failed(stopped(&p)))));
            }
        });
    }
    rx
}

fn run(s: Subject, branch: bool, task: &Task, root: &Path, extra: &str, tx: &Sender<(u64, Status)>) -> anyhow::Result<Status> {
    use anyhow::Context;
    let n = s.n();
    // Its hold on it: one run on it at a time, from any dv (an issue's view, a PR's); let go as
    // this ends, however.
    let mut claim = match crate::claude::claim(&crate::claude::pr_key(n), task.doing(), None)? {
        Ok(c) => c,
        Err(w) => return Ok(Status::Busy(format!("Claude is {w} on it"), None)),
    };
    // A PR merged or closed while it waited its turn (queued): nothing to do on it now.
    if let Subject::Pr(_) = s
        && let Ok(Some(state)) = crate::gh::pr_state(n)
        && state != "OPEN"
    {
        return Ok(Status::Gone(state));
    }
    _ = tx.send((n, Status::Preparing));
    // The worktree, checked out at the PR (by a dv of its own: that changes directory); an
    // issue's, here (no code of its own).
    let (dir, note) = match s {
        Subject::Issue(_) => (root.to_path_buf(), String::new()),
        Subject::Pr(_) => {
            let mut prep = std::process::Command::new(crate::self_exe());
            prep.args(["prepare-worktree", &n.to_string()]).current_dir(root);
            if branch {
                prep.arg("-b");
            }
            let out = prep.stdin(std::process::Stdio::null()).output()?;
            if !out.status.success() {
                let err = String::from_utf8_lossy(&out.stderr);
                anyhow::bail!("{}", err.lines().rev().find(|l| !l.trim().is_empty()).unwrap_or("preparing the worktree failed").trim());
            }
            let out = String::from_utf8_lossy(&out.stdout);
            let mut said = out.lines();
            (PathBuf::from(said.next().unwrap_or("").trim()), said.next().unwrap_or("").trim().to_owned())
        }
    };
    let note = note.as_str();
    // A PR's checkout (the one it's prepared in: its worktree, or one on its branch already):
    // not where Claude's at work otherwise (a pane you started there), not a second; then marked
    // its. An issue's run has no checkout of its own: none said.
    if let Subject::Pr(_) = s {
        if let Some(w) = crate::claude::at_work_in(&dir, None) {
            return Ok(Status::Busy(format!("Claude is {w} in {}", dir.display()), Some(dir)));
        }
        claim.set_dir(&dir);
    }
    // An update reviews what's new over the commit reviewed, as checked out here. The checkout
    // still there (kept back for work in it): nothing to review. Rewritten (force-pushed,
    // rebased): all of it is new.
    let task = &match task {
        Task::Update(old) => {
            let head = crate::diff::git_in(&dir, &["rev-parse", "HEAD"])?;
            // (At it, or behind it: the review came from a checkout further on.)
            if head.trim().starts_with(old.as_str()) || crate::diff::git_in(&dir, &["merge-base", "--is-ancestor", "HEAD", old]).is_ok() {
                let why = if note.is_empty() { "it isn't at the PR's head" } else { note };
                anyhow::bail!("not updating Claude's review: the checkout isn't past the commit it reviewed ({why})");
            }
            match crate::diff::git_in(&dir, &["merge-base", "--is-ancestor", old, "HEAD"]) {
                Ok(_) => task.clone(),
                Err(_) => Task::Review,
            }
        }
        _ => task.clone(),
    };
    // The PR's session in that checkout, as dv there would resume it (a new one, if it has
    // none, or one from another version of the skill).
    let review = s.path()?;
    let session = crate::claude::session_args(&review, &dir)?;
    _ = tx.send((n, Status::Reviewing));
    let prompt = format!("{} (It's {} now.)", task.prompt(s), crate::claude::now());
    let prompt = crate::claude::with_instructions(&prompt, extra);
    let log = std::env::temp_dir().join(format!("dv-claude-batch-{}-{n}.log", std::process::id()));
    let more = crate::claude::tools_for(matches!(task, Task::FixChecks).then_some(n), matches!(task, Task::Answer { .. }));
    let mut child = crate::claude::claude_command(task.work(), &prompt, &dir, &session, &more.iter().map(String::as_str).collect::<Vec<_>>(), None)
        .envs([s.env()])
        .stdin(std::process::Stdio::null())
        .stdout(std::fs::File::create(&log)?)
        .stderr(std::fs::File::create(log.with_extension("err"))?)
        .spawn()
        .context("is Claude Code (`claude`) installed?")?;
    // (Its claim by its own process from here: what tells any dv it's at it, this one gone even.)
    claim.set_pid(child.id());
    let st = child.wait()?;
    let v: serde_json::Value = std::fs::read_to_string(&log).ok().and_then(|t| serde_json::from_str(&t).ok()).unwrap_or_default();
    if !st.success() || v["is_error"] != false {
        _ = crate::claude::save_session(&review, &dir, None); // start afresh next time
        anyhow::bail!("Claude stopped ({st}): see {}", log.display());
    }
    Ok(Status::Done(format!(
        "{} turns, ${:.2}",
        v["num_turns"].as_u64().unwrap_or(0),
        v["total_cost_usd"].as_f64().unwrap_or(0.0)
    )))
}

/// `f`'s result, or a failure if it panics: a background run always says how it ended (its
/// view counts it as under way till then: dv would ask before quitting forever).
pub fn guarded<T>(f: impl FnOnce() -> anyhow::Result<T>) -> anyhow::Result<T> {
    std::panic::catch_unwind(std::panic::AssertUnwindSafe(f)).unwrap_or_else(|p| Err(anyhow::anyhow!(stopped(&p))))
}

/// What a panic said: "dv hit a bug: …".
pub(crate) fn panicked(p: &Box<dyn std::any::Any + Send>) -> String {
    let said = p.downcast_ref::<&str>().map(|s| s.to_string()).or_else(|| p.downcast_ref::<String>().cloned()).unwrap_or_default();
    format!("dv hit a bug: {said}")
}

/// A background run's failure from its panic.
fn stopped(p: &Box<dyn std::any::Any + Send>) -> String {
    format!("its run stopped ({})", panicked(p))
}

/// Claude drafts one replacement for `prs` (see `replace.rs`) in the background: its news
/// comes back on the receiver, under the replacement's number (the first PR's).
pub fn replace(prs: Vec<u64>, root: PathBuf, extra: String) -> Receiver<(u64, Status)> {
    let (tx, rx) = mpsc::channel();
    let id = prs[0];
    _ = tx.send((id, Status::Preparing));
    std::thread::spawn(move || {
        let ready = tx.clone();
        // (Its run guarded inside, the record's mark coming off; and all of it, here.)
        let st = guarded(|| crate::replace::draft(&prs, &root, &extra, move || _ = ready.send((id, Status::Reviewing))));
        _ = tx.send((id, st.map_or_else(|e| Status::failed(format!("{e:#}")), Status::Done)));
    });
    rx
}

/// A replacement each: the PRs it replaces, and why they go together (Claude's word).
pub type Groups = Vec<(Vec<u64>, String)>;

/// Claude suggests which of `prs` (number, title, author, base) one replacement should cover
/// together (attempts at the same change) and which stand alone, in the background. It only
/// reads them.
pub fn group(prs: Vec<(u64, String, String, String)>, root: PathBuf) -> Receiver<anyhow::Result<Groups>> {
    let (tx, rx) = mpsc::channel();
    std::thread::spawn(move || {
        let _slot = Slot::take();
        _ = tx.send(guarded(|| suggest_groups(&prs, &root)));
    });
    rx
}

fn suggest_groups(prs: &[(u64, String, String, String)], root: &Path) -> anyhow::Result<Groups> {
    let list: Vec<String> = prs.iter().map(|(n, t, a, b)| format!("- #{n} {t} (@{a}, into {b})")).collect();
    let prompt = format!(
        "These PRs are to be replaced by better ones:\n{}\n\nRead each (`gh pr view N`, `gh pr diff N`) and group \
         them: PRs that try to make the same change (the same issue, the same code) go in one group, to be replaced \
         by one PR; the others each alone. PRs into different base branches never share a group. Change nothing. \
         Reply with only a JSON array, no other text: [{{\"prs\": [12, 15], \"why\": \"one short line\"}}], \
         every PR in exactly one group.",
        list.join("\n")
    );
    let tools = ["Bash(gh pr view:*)", "Bash(gh pr diff:*)"];
    let out = crate::claude::claude_reader(crate::claude::Work::Group, &prompt, root, &tools).stdin(std::process::Stdio::null()).output()?;
    let v: serde_json::Value = serde_json::from_slice(&out.stdout).unwrap_or_default();
    anyhow::ensure!(v["is_error"] == false, "Claude couldn't group them ({})", out.status);
    Ok(complete(parse_groups(v["result"].as_str().unwrap_or(""))?, prs.iter().map(|p| p.0)))
}

/// Claude's `groups` of the PRs `asked`, each once and no other: one it named again or didn't
/// ask about dropped, one it left out alone (said so), for you to see in your editor.
fn complete(groups: Groups, asked: impl Iterator<Item = u64>) -> Groups {
    let asked: Vec<u64> = asked.collect();
    let mut seen: Vec<u64> = vec![];
    let mut out: Groups = vec![];
    for (g, why) in groups {
        let mut kept = vec![];
        for n in g {
            if asked.contains(&n) && !seen.contains(&n) {
                seen.push(n);
                kept.push(n);
            }
        }
        if !kept.is_empty() {
            out.push((kept, why));
        }
    }
    out.extend(asked.into_iter().filter(|n| !seen.contains(n)).map(|n| (vec![n], "left out by Claude".to_owned())));
    out
}

#[test]
fn groups_completed() {
    let g = |v: &[u64]| (v.to_vec(), String::new());
    assert_eq!(complete(vec![g(&[12, 15]), g(&[15, 99])], [12, 15, 20].into_iter()), vec![g(&[12, 15]), (vec![20], "left out by Claude".into())]);
}

/// The groups in Claude's answer (a JSON array, maybe with words around it).
fn parse_groups(answer: &str) -> anyhow::Result<Groups> {
    use anyhow::Context;
    let v = crate::claude::json_in(answer, '[').filter(serde_json::Value::is_array).context("no groups in Claude's answer")?;
    Ok(v.as_array()
        .context("no groups")?
        .iter()
        .map(|g| (g["prs"].as_array().into_iter().flatten().filter_map(serde_json::Value::as_u64).collect(), g["why"].as_str().unwrap_or("").to_owned()))
        .filter(|g: &(Vec<u64>, String)| !g.0.is_empty())
        .collect())
}

/// Groups as you edit them: a line each, the PRs' numbers, `#` and after it a note.
pub fn groups_text(groups: &Groups) -> String {
    groups.iter().map(|(prs, why)| {
        let nums: Vec<String> = prs.iter().map(u64::to_string).collect();
        format!("{:<14} # {why}\n", nums.join(" "))
    }).collect()
}

pub fn parse_groups_text(text: &str) -> Vec<Vec<u64>> {
    text.lines()
        // (Its numbers, `#12` or `12`, up to the first word that isn't one: a note.)
        .map(|l| l.split_whitespace().map_while(|t| t.trim_start_matches('#').parse().ok()).collect::<Vec<u64>>())
        .filter(|g| !g.is_empty())
        .collect()
}

#[test]
fn groups() {
    let g = parse_groups("Here: [{\"prs\": [12, 15], \"why\": \"same fix\"}, {\"prs\": [20], \"why\": \"alone\"}]").unwrap();
    assert_eq!(g, vec![(vec![12, 15], "same fix".to_owned()), (vec![20], "alone".to_owned())]);
    let text = groups_text(&g);
    assert_eq!(text, "12 15          # same fix\n20             # alone\n");
    assert_eq!(parse_groups_text(&format!("{text}\n21 22 # mine\n#23 #24")), [vec![12, 15], vec![20], vec![21, 22], vec![23, 24]]);
}

/// The PRs that have a review from Claude (a review file in dv's directory).
pub fn reviewed() -> std::collections::HashSet<u64> {
    let Ok(dir) = crate::claude::dv_dir() else { return Default::default() };
    reviewed_in(&dir.join("reviews"))
}

/// The PRs with a review in `dir`.
pub fn reviewed_in(dir: &std::path::Path) -> std::collections::HashSet<u64> {
    let Ok(entries) = std::fs::read_dir(dir) else { return Default::default() };
    entries
        .flatten()
        .filter_map(|e| e.file_name().to_str()?.strip_prefix("pr-")?.strip_suffix(".json")?.parse().ok())
        .collect()
}

#[test]
fn a_run_that_panics_says_so() {
    assert_eq!(guarded(|| Ok(1)).unwrap(), 1);
    let e = guarded::<()>(|| panic!("boom")).unwrap_err();
    assert!(format!("{e}").contains("its run stopped (dv hit a bug: boom)"));
    // A worker's panic: that one failed, the next still runs.
    let rx = pool(vec![1u64, 2], |&n, _| if n == 1 { panic!("one") } else { (n, Status::Done("ok".into())) });
    let ends: Vec<(u64, Status)> = rx.iter().filter(|(_, s)| !matches!(s, Status::Queued)).collect();
    assert!(ends.iter().any(|(n, s)| *n == 1 && matches!(s, Status::Failed(e, _) if e.contains("one"))));
    assert!(ends.iter().any(|(n, s)| *n == 2 && matches!(s, Status::Done(_))));
}