orch/engine/

review.rs

//! PR review pipeline — review agent orchestration.
//!
//! This module handles running the review agent on completed tasks, parsing its
//! decision, and posting automated review comments.
//!
//! Related modules:
//! - [`super::review_poll`] — polling open PRs for human review feedback
//! - [`super::auto_merge`] — merging approved PRs and handling change requests

/// Maximum consecutive review agent failures before the task is blocked
/// for human intervention. Exported so `tick` and `sync` use the same threshold.
pub(crate) const MAX_REVIEW_AGENT_FAILURES: u64 = 3;

/// Maximum consecutive PR-creation failures before blocking the task.
const MAX_PR_CREATE_FAILURES: u64 = 3;

/// Map an [`AgentError`] to the structured outcome string stored in `task_runs.outcome`.
///
/// Mirrors the logic in `runner::classify_run_outcome` so review runs carry the
/// same outcome fidelity as agent runs.
fn outcome_for_agent_error(err: &runner::agents::AgentError) -> &'static str {
    match err {
        runner::agents::AgentError::Timeout { .. } => "timeout",
        // Consult credit-exhaustion detection first — generic rate-limit patterns
        // (e.g. "usage limit") also match billing-cycle exhaustion messages, so
        // reviews on a billing-cycle-exhausted model must not be stored as a
        // plain "rate_limit"/"auth_error" outcome (issue #3529).
        runner::agents::AgentError::RateLimit { message } => {
            crate::engine::cooldown::detect_credit_exhaustion_outcome(message)
                .unwrap_or("rate_limit")
        }
        runner::agents::AgentError::Auth { message } => {
            crate::engine::cooldown::detect_credit_exhaustion_outcome(message)
                .unwrap_or("auth_error")
        }
        runner::agents::AgentError::InvalidResponse { .. } => "parse_error",
        _ => "failed",
    }
}

/// Format an [`AgentError`] as the compact string persisted to `task_runs.error`
/// and surfaced via [`ReviewDecision::Failed`].
///
/// Mirrors the per-variant formatting in `runner::fallback::handle_error` so review
/// runs never leak raw provider payloads (e.g. truncated `api_retry` NDJSON
/// fragments) into the audit trail. Without this, `err.to_string()` would expand
/// `AgentError::RateLimit { message }` via its `Display` impl and persist hundreds
/// of bytes of raw JSON into `task_runs.error` (issue #3149, regression of #2881).
fn format_review_error(agent: &str, err: &runner::agents::AgentError) -> String {
    match err {
        runner::agents::AgentError::RateLimit { message } => format!(
            "{agent} rate limit: {}",
            runner::fallback::summarize_rate_limit_error(message)
        ),
        runner::agents::AgentError::Auth { message } => format!("{agent} auth error: {message}"),
        runner::agents::AgentError::Timeout { elapsed } => {
            format!("{agent} timed out after {}s", elapsed.as_secs())
        }
        other => format!("{agent} agent error: {other}"),
    }
}

/// Parse a PR number from a GitHub PR URL.
///
/// Validates that the URL matches the expected GitHub PR URL format
/// (`https://github.com/<owner>/<repo>/pull/<number>`) before extracting the
/// number. Returns `Err` with a descriptive message for any URL that doesn't
/// conform to this structure, including URLs with wrong domains, missing
/// `/pull/` segments, or non-numeric trailing components.
///
/// The extracted segment is stripped of any query string (`?`) or hash
/// fragment (`#`) before parsing so that URLs like
/// `https://github.com/owner/repo/pull/42?tab=files` are handled correctly.
pub(crate) fn parse_pr_number_from_url(url: &str) -> Result<u64, String> {
    let url = url.trim();
    if url.is_empty() {
        return Err("URL is empty".to_string());
    }
    if !url.starts_with("https://github.com/") {
        return Err(format!(
            "URL does not start with 'https://github.com/': {url}"
        ));
    }
    if !url.contains("/pull/") {
        return Err(format!("URL missing '/pull/' segment: {url}"));
    }
    let last_segment = url
        .trim_end_matches('/')
        .rsplit('/')
        .next()
        .ok_or_else(|| format!("No segment after '/pull/' in URL: {url}"))?;
    let number_part = last_segment
        .split(['?', '#'])
        .next()
        .filter(|s| !s.is_empty())
        .ok_or_else(|| format!("PR number segment is empty in URL: {url}"))?;
    let pr_num: u64 = number_part.parse().map_err(|_| {
        format!("PR number is not a valid positive integer: '{number_part}' in URL: {url}")
    })?;
    if pr_num == 0 {
        return Err(format!("PR number cannot be zero in URL: {url}"));
    }
    Ok(pr_num)
}

fn review_started_comment(review_agent: &str, review_model: &str) -> String {
    let footer = attribution_footer("Review started", review_agent, Some(review_model));
    format!("🔍 Automated review started{}", footer)
}

fn should_skip_no_code_reroute_increment(last_error: &str) -> bool {
    crate::engine::runner::git_ops::is_transient_github_api_error(last_error)
}

use crate::backends::{ExternalBackend, ExternalTask, Status};
use crate::cmd::CommandErrorContext;
use crate::config;
use crate::engine::attribution_footer;
use crate::engine::auto_merge::{auto_merge_pr, handle_review_changes};
use crate::engine::runner;
use crate::engine::runner::worktree;
use crate::engine::tasks::TaskManager;
use crate::github::http::GhHttp;
use crate::store::store_log_activity;
use crate::store::TaskStore;
use crate::store::{opt_store_get_task, set_review_session_expected, store_increment};
use crate::store::{store_set, store_set_result};
use crate::store::{CompleteRun, RunTokenUsage, StartRun};
use crate::tmux::TmuxManager;
use anyhow::Context;
use std::collections::HashSet;
use std::sync::Arc;
use tokio::process::Command;
use tokio::sync::RwLock;

use super::router::Router;

/// Review agent decision result.
#[derive(Debug, Clone)]
pub(crate) enum ReviewDecision {
    /// Review approved, PR can be merged.
    Approve,
    /// Changes requested, PR needs fixes.
    RequestChanges {
        notes: String,
        issues: Vec<crate::engine::runner::response::ReviewIssue>,
    },
    /// Review agent failed or crashed (reason stored for logging).
    Failed(String),
    /// Unrecoverable failure — task should be blocked for human intervention.
    Blocked(String),
    /// No PR exists — nothing to review, task marked done directly.
    Skipped,
    /// No PR and no commits — task re-routed for retry (no_code_reroutes counter must not be reset).
    Rerouted,
}

/// Outcome of [`ensure_pr_exists`]: either a ready PR number or an early return decision.
enum EnsurePrResult {
    /// PR exists (found in store or via API, or just created). Proceed with review.
    Ready(u64),
    /// No PR and status already updated — return this decision to the caller.
    EarlyReturn(ReviewDecision),
}

enum ReviewPhase<T> {
    Ready(T),
    EarlyReturn(ReviewDecision),
}

struct ReviewContext {
    store_id: Option<i64>,
    had_prev_push_failure: bool,
    review_attempt: u32,
    worktree_path: std::path::PathBuf,
    branch_name: String,
    pr_number: u64,
    /// PR head SHA when the review started, the commit the decision applies to
    head_sha: Option<String>,
    default_branch: String,
    review_agent: String,
    review_model: Option<String>,
    review_task_id: String,
    review_attempt_dir: std::path::PathBuf,
    output_file: std::path::PathBuf,
    invocation: runner::agent::AgentInvocation,
    same_agent_fallback: bool,
}

impl ReviewContext {
    fn review_model_str(&self) -> &str {
        self.review_model.as_deref().unwrap_or("")
    }
}

struct ReviewRun {
    run_id: Option<i64>,
}

struct ParsedReview {
    run_id: Option<i64>,
    exit_code: i32,
    stderr: String,
    raw_output: String,
    text_for_review: String,
    token_usage: RunTokenUsage,
    review_notes_for_comment: String,
    decision: ReviewDecision,
}

/// KV key holding the commit an approving review was already written for
fn approved_sha_key(repo: &str, task_id: &str, pr_number: u64) -> String {
    format!("review_approved_sha:{repo}#{task_id}#{pr_number}")
}

/// Read `HEAD` of a worktree. `None` when the worktree is gone or not a repository
async fn worktree_head_sha(dir: &std::path::Path) -> Option<String> {
    let out = tokio::process::Command::new("git")
        .args(["rev-parse", "HEAD"])
        .current_dir(dir)
        .kill_on_drop(true)
        .output()
        .await
        .ok()?;
    if !out.status.success() {
        return None;
    }
    let sha = String::from_utf8_lossy(&out.stdout).trim().to_string();
    (!sha.is_empty()).then_some(sha)
}

/// Record that the commit just pushed to the PR carries an approving review, so a later
/// NeedsReview event on the same commit skips the review agent
async fn record_approved_sha(
    store: &Arc<TaskStore>,
    repo: &str,
    task_id: &str,
    pr_number: u64,
    dir: &std::path::Path,
) {
    let Some(sha) = worktree_head_sha(dir).await else {
        return;
    };
    if let Err(e) = store
        .kv_set(&approved_sha_key(repo, task_id, pr_number), &sha)
        .await
    {
        tracing::warn!(task_id, err = %e, "failed to record approved commit sha");
    }
}

/// Drop a recorded approval so the next NeedsReview event runs a real review
async fn clear_approved_sha(store: &Arc<TaskStore>, repo: &str, task_id: &str, pr_number: u64) {
    if let Err(e) = store
        .kv_delete(&approved_sha_key(repo, task_id, pr_number))
        .await
    {
        tracing::warn!(task_id, err = %e, "failed to clear approved commit sha");
    }
}

/// Records the earlier approval as this review round's decision. Without it the stale
/// in_review check sees a round with no decision and resets the task every minute
async fn record_carried_approval(
    store: &Arc<TaskStore>,
    repo: &str,
    task_id: &str,
    pr_number: u64,
) {
    store_log_activity(
        &Some(Arc::clone(store)),
        repo,
        task_id,
        "review_decision",
        None,
        None,
        None::<&str>,
        None::<&str>,
        Some(&serde_json::json!({
            "decision": "approve",
            "pr_number": pr_number,
            "reason": "head_already_approved",
        })),
    )
    .await;
}

/**
 * True when the PR's current head is the commit a previous review approved.
 *
 * Compared against the live PR head, not the worktree: a force-push from another clone
 * moves the head while the worktree stays on the approved commit, and skipping the
 * review there would merge code nobody reviewed. Any lookup failure returns false, so
 * the fallback is to review again (#3783)
 **/
async fn pr_head_already_approved(
    store: &Arc<TaskStore>,
    repo: &str,
    task_id: &str,
    pr_number: u64,
) -> bool {
    let approved = match store
        .kv_get(&approved_sha_key(repo, task_id, pr_number))
        .await
    {
        Ok(sha) => sha,
        Err(e) => {
            tracing::warn!(task_id, err = %e, "failed to read approved commit sha");
            return false;
        }
    };
    if approved.is_none() {
        return false;
    }
    let gh = match crate::github::http::GhHttp::new() {
        Ok(gh) => gh,
        Err(e) => {
            tracing::warn!(task_id, err = %e, "cannot check PR head against recorded approval");
            return false;
        }
    };
    let pr_head = match gh.get_pr(repo, pr_number).await {
        Ok(pr) => Some(pr.head.sha),
        Err(e) => {
            tracing::warn!(task_id, pr_number, err = %e, "failed to read PR head sha");
            None
        }
    };
    approval_covers_head(approved.as_deref(), pr_head.as_deref())
}

/// Whether a recorded approval still applies to the PR's current head commit
fn approval_covers_head(approved: Option<&str>, pr_head: Option<&str>) -> bool {
    match (approved, pr_head) {
        (Some(approved), Some(head)) => !approved.is_empty() && approved == head,
        _ => false,
    }
}

fn build_review_disallowed_tools(worktree_path: &std::path::Path) -> Vec<String> {
    let mut tools = vec![
        "Bash(rm *)".to_string(),
        "Bash(rm -*)".to_string(),
        "Bash(git push*)".to_string(),
        // Pushing and opening the PR are orch's steps, never the agent's (#3783)
        "Bash(gh pr create*)".to_string(),
    ];
    let orch_yml = worktree_path.join(".orch.yml");
    let orch_yml_str = orch_yml.to_string_lossy();
    tools.extend([
        format!("Write({orch_yml_str})"),
        format!("Edit({orch_yml_str})"),
    ]);
    if let Ok(orch_home) = crate::home::orch_home() {
        for path in [
            orch_home.join("config.yml"),
            orch_home.join("config.example.yml"),
        ] {
            let path_str = path.to_string_lossy();
            tools.extend([
                format!("Read({path_str})"),
                format!("Write({path_str})"),
                format!("Edit({path_str})"),
            ]);
        }
    }
    tools
}

async fn select_review_agent(
    task_id: &str,
    task_agent: &str,
    store_id: Option<i64>,
    router: &Arc<RwLock<Router>>,
    store: &Arc<TaskStore>,
) -> anyhow::Result<(String, Option<String>, bool)> {
    let mut exclude_set: HashSet<String> = HashSet::new();
    if !task_agent.is_empty() {
        exclude_set.insert(task_agent.to_string());
    }
    if let Some(store_id) = store_id {
        if let Ok(failed) = store.previous_review_agents(store_id).await {
            for agent in failed {
                exclude_set.insert(agent);
            }
        }
    }

    let mut r = router.write().await;
    let mut tried_agents: HashSet<String> = HashSet::new();
    let mut chosen_agent: Option<String> = None;
    let mut chosen_model: Option<String> = None;
    let available_count = r.available_agents.len().max(1);

    loop {
        let tmp_exclude_refs: Vec<&str> = exclude_set.iter().map(|s| s.as_str()).collect();
        let agent = match r.next_round_robin_agent(&tmp_exclude_refs, "review") {
            Some(a) => a,
            None => break,
        };

        if tried_agents.contains(&agent) {
            break;
        }
        tried_agents.insert(agent.clone());

        let model = r.config.model_for_complexity(&agent, "review", task_id);
        if model.is_some() {
            chosen_agent = Some(agent);
            chosen_model = model;
            break;
        }

        exclude_set.insert(agent);
        if exclude_set.len() >= available_count {
            break;
        }
    }

    if let Some(agent) = chosen_agent {
        let same = !task_agent.is_empty() && agent == task_agent;
        Ok((agent, chosen_model, same))
    } else {
        let final_exclude_refs: Vec<&str> = exclude_set.iter().map(|s| s.as_str()).collect();
        let fallback_agent = r
            .next_round_robin_agent(&final_exclude_refs, "review")
            .unwrap_or_else(|| "claude".to_string());
        let fallback_model = r
            .config
            .model_for_complexity(&fallback_agent, "review", task_id);
        if fallback_model.is_none() {
            anyhow::bail!(
                "no review agent with a configured review model is available for task {}",
                task_id
            );
        }
        let same = !task_agent.is_empty() && fallback_agent == task_agent;
        Ok((fallback_agent, fallback_model, same))
    }
}

#[allow(clippy::too_many_arguments)]
async fn build_review_context(
    task: &ExternalTask,
    repo: &str,
    router: &Arc<RwLock<Router>>,
    task_manager: &Arc<TaskManager>,
    store: &Arc<TaskStore>,
) -> anyhow::Result<ReviewPhase<ReviewContext>> {
    let stored_task = store
        .get_by_external_id(repo, &task.id.0)
        .await
        .with_context(|| format!("store lookup failed for task {}", task.id.0))?;

    let store_id: Option<i64> = stored_task.as_ref().map(|t| t.id);

    // A commit that already has an approval must not be reviewed again. Checked before
    // the review_invocations increment so a stuck merge cannot exhaust the review quota
    if let Some(pr_number) = stored_task.as_ref().and_then(|t| t.pr_number) {
        let pr_number = pr_number as u64;
        if pr_head_already_approved(store, repo, &task.id.0, pr_number).await {
            tracing::info!(
                task_id = task.id.0,
                pr_number,
                "PR head already approved — skipping review, merge retry continues"
            );
            record_carried_approval(store, repo, &task.id.0, pr_number).await;
            return Ok(ReviewPhase::EarlyReturn(ReviewDecision::Approve));
        }
    }

    let had_prev_push_failure = stored_task
        .as_ref()
        .is_some_and(|t| t.last_error.contains("push failed"));
    let review_attempt: u32 = match store_increment(
        &Some(Arc::clone(store)),
        repo,
        &task.id.0,
        "review_invocations",
    )
    .await
    {
        Ok(v) => v as u32,
        Err(e) => {
            tracing::warn!(task_id = task.id.0, err = %e, "failed to increment review_invocations — using fallback attempt=1");
            1
        }
    };

    let no_code_review = stored_task.as_ref().is_some_and(|t| {
        t.worktree.trim().is_empty() && t.branch.trim().is_empty() && t.pr_number.is_none()
    });
    if no_code_review {
        if crate::engine::tasks::is_internal_id(&task.id.0) {
            // Internal tasks legitimately reach needs_review with nothing to review.
            tracing::info!(
                task_id = task.id.0,
                "no worktree/branch/pr for internal needs_review task — skipping review and marking done"
            );
            if let Err(e) = task_manager
                .update_task_status(&task.id, Status::Done)
                .await
            {
                tracing::error!(
                    task_id = task.id.0,
                    err = %e,
                    "update_task_status(Done) failed for no-code review skip"
                );
            }
            return Ok(ReviewPhase::EarlyReturn(ReviewDecision::Skipped));
        }

        // External task with no code produced: re-route instead of marking done.
        // Shares the no_code_reroutes circuit breaker with the no-PR/no-commits path below.
        let max_reroutes: u32 = config::get("workflow.max_reroute_attempts")
            .ok()
            .and_then(|s| s.parse().ok())
            .or_else(|| {
                config::get("workflow.max_attempts")
                    .ok()
                    .and_then(|s| s.parse().ok())
            })
            .unwrap_or(3);

        let reroutes = match store_increment(
            &Some(Arc::clone(store)),
            repo,
            &task.id.0,
            "no_code_reroutes",
        )
        .await
        {
            Ok(v) => v,
            Err(e) => {
                tracing::warn!(task_id = task.id.0, err = %e, "failed to increment no_code_reroutes — skipping reroute/block decision this tick");
                return Ok(ReviewPhase::EarlyReturn(ReviewDecision::Failed(format!(
                    "transient store error: {e}"
                ))));
            }
        };

        if reroutes as u32 >= max_reroutes {
            tracing::error!(
                task_id = task.id.0,
                reroutes,
                max_reroutes,
                "reached max reroute attempts for no-worktree/branch/pr result — blocking for human review"
            );
            let msg = format!(
                "no worktree/branch/pr after {}/{} reroute attempts",
                reroutes, max_reroutes
            );
            let fields = [
                (
                    "block_reason",
                    serde_json::json!(format!(
                        "max reroute attempts ({}) reached — no worktree/branch/pr produced",
                        max_reroutes
                    )),
                ),
                ("last_error", serde_json::json!(msg)),
                ("agent", serde_json::json!(null)),
                ("model", serde_json::json!(null)),
            ];
            if let Err(e) = task_manager
                .update_task_status_and_result(&task.id, Status::Blocked, &fields)
                .await
            {
                tracing::error!(task_id = task.id.0, err = %e, "update_task_status_and_result(Blocked) failed — skipping block to avoid silent auto-unblock loop");
            }
            return Ok(ReviewPhase::EarlyReturn(ReviewDecision::Blocked(msg)));
        }

        tracing::warn!(
            task_id = task.id.0,
            reroutes,
            max_reroutes,
            "no worktree/branch/pr for external needs_review task — re-routing to a different agent"
        );
        if let Err(e) = store_set_result(
            &Some(Arc::clone(store)),
            repo,
            &task.id.0,
            &[
                ("agent", serde_json::json!(null)),
                ("model", serde_json::json!(null)),
                (
                    "last_error",
                    serde_json::json!("no worktree/branch/pr produced"),
                ),
            ],
        )
        .await
        {
            tracing::error!(task_id = task.id.0, err = %e, "failed to clear agent/model for reroute — skipping status transition");
        } else if let Err(e) = task_manager.update_task_status(&task.id, Status::New).await {
            tracing::error!(task_id = task.id.0, err = %e, "update_task_status(New) failed — task may be stuck in InReview");
        }
        return Ok(ReviewPhase::EarlyReturn(ReviewDecision::Rerouted));
    }

    let mut worktree_path = match stored_task.as_ref().map(|t| t.worktree.as_str()) {
        Some(w) if !w.is_empty() => std::path::PathBuf::from(w),
        _ => {
            tracing::warn!(task_id = task.id.0, "no worktree found for review");
            return Ok(ReviewPhase::EarlyReturn(ReviewDecision::Failed(
                "no worktree found".to_string(),
            )));
        }
    };

    let mut branch_name = match stored_task.as_ref().map(|t| t.branch.as_str()) {
        Some(b) if !b.is_empty() => b.to_string(),
        _ => {
            tracing::warn!(task_id = task.id.0, "no branch found for review");
            return Ok(ReviewPhase::EarlyReturn(ReviewDecision::Failed(
                "no branch found".to_string(),
            )));
        }
    };

    let mut missing_worktree_recovered = false;
    if !worktree_path.exists() {
        match recover_missing_review_worktree(task, repo, store).await {
            Ok(recovered) => {
                branch_name = recovered.branch;
                worktree_path = recovered.work_dir;
                missing_worktree_recovered = true;
            }
            Err(e) => {
                let reason = format!("missing review worktree recovery failed: {e}");
                tracing::error!(task_id = task.id.0, error = %reason, "cannot recover review worktree");
                store_set(
                    &Some(Arc::clone(store)),
                    repo,
                    &task.id.0,
                    &[("last_error", serde_json::json!(reason.clone()))],
                )
                .await;
                return Ok(ReviewPhase::EarlyReturn(ReviewDecision::Blocked(reason)));
            }
        }
    }

    if missing_worktree_recovered {
        store_set(
            &Some(Arc::clone(store)),
            repo,
            &task.id.0,
            &[("last_error", serde_json::json!(""))],
        )
        .await;
    }

    let agent_summary = stored_task
        .as_ref()
        .map(|t| t.summary.clone())
        .unwrap_or_default();
    let stored_pr_number = stored_task
        .as_ref()
        .and_then(|t| t.pr_number)
        .map(|n| n as u64)
        .filter(|&n| n > 0);

    let pr_number = match ensure_pr_exists(
        task,
        &branch_name,
        &worktree_path,
        repo,
        &agent_summary,
        store,
        task_manager,
        stored_pr_number,
    )
    .await?
    {
        EnsurePrResult::Ready(n) => n,
        EnsurePrResult::EarlyReturn(decision) => return Ok(ReviewPhase::EarlyReturn(decision)),
    };

    // Run `git fetch` with a timeout so a hung network call cannot stall the
    // review pipeline indefinitely. Treat a timeout or non-zero exit as a
    // non-fatal warning (stale remote refs are acceptable for the review).
    {
        let fetch_timeout = std::time::Duration::from_secs(60);
        let fetch_future = async {
            let mut cmd = Command::new("git");
            cmd.args(["fetch", "origin", "--prune"])
                .current_dir(&worktree_path);
            // Ensure the child is killed if the timeout elapses
            cmd.kill_on_drop(true);
            cmd.output_with_context().await
        };

        match tokio::time::timeout(fetch_timeout, fetch_future).await {
            Ok(Ok(o)) if o.status.success() => {
                // ok
            }
            Ok(Ok(o)) => {
                tracing::warn!(
                    task_id = task.id.0,
                    exit_code = ?o.status.code(),
                    stdout = %String::from_utf8_lossy(&o.stdout),
                    stderr = %String::from_utf8_lossy(&o.stderr),
                    "review: git fetch failed (non-zero exit) — diff/log may use stale remote refs"
                );
            }
            Ok(Err(e)) => {
                tracing::warn!(
                    task_id = task.id.0,
                    error = %e,
                    "review: git fetch failed — diff/log may use stale remote refs"
                );
            }
            Err(_) => {
                tracing::warn!(
                    task_id = task.id.0,
                    "review: git fetch timed out after 60s — diff/log may use stale remote refs"
                );
            }
        }
    }

    let default_branch = runner::worktree::detect_default_branch(&worktree_path).await;
    let git_diff = runner::context::build_git_diff(&worktree_path, &default_branch).await;
    let git_log = runner::context::build_git_log(&worktree_path, &default_branch).await;
    // An unknown project falls back to the global config, never to the worktree's .orch.yml,
    // which the task agent may have written
    let project_dir = crate::config::get_projects_with_paths()
        .unwrap_or_default()
        .into_iter()
        .find(|(candidate, _)| candidate == repo)
        .map(|(_, dir)| dir);
    let sandboxed = match &project_dir {
        Some(dir) => runner::sandbox::enabled(dir),
        None => runner::sandbox::enabled_globally(),
    };
    let main_project_dir = project_dir.unwrap_or_else(|| worktree_path.clone());
    let review_prompt = runner::agent::build_review_prompt(
        task,
        &agent_summary,
        &git_diff,
        &git_log,
        &default_branch,
        pr_number,
        runner::agent::Shipcheck::detect(&worktree_path)
            .await
            .for_prompt(sandboxed),
    );

    let task_agent = stored_task
        .as_ref()
        .and_then(|t| t.agent.clone())
        .unwrap_or_default();
    let (review_agent, review_model, same_agent_fallback) =
        select_review_agent(&task.id.0, &task_agent, store_id, router, store).await?;

    anyhow::ensure!(
        review_model.is_some(),
        "review model missing for task {} after agent selection — refusing to dispatch",
        task.id.0
    );

    if same_agent_fallback {
        tracing::warn!(
            task_id = task.id.0,
            agent = %review_agent,
            "no other agent available for review — falling back to the agent that wrote the code"
        );
        store_log_activity(
            &Some(Arc::clone(store)),
            repo,
            &task.id.0,
            "review_agent_fallback",
            None,
            None,
            Some(&review_agent),
            review_model.as_deref(),
            Some(&serde_json::json!({
                "reason": "all available agents excluded (author + previous reviewers)",
            })),
        )
        .await;
    }

    tracing::info!(
        task_id = task.id.0,
        agent = %review_agent,
        model = ?review_model,
        "spawning review agent"
    );

    let review_task_id = format!("{}-review", task.id.0);
    let review_attempt_dir =
        crate::home::task_attempt_dir_async(repo, &review_task_id, review_attempt).await?;
    let output_file = review_attempt_dir.join("output.json");

    let git_name =
        crate::config::get("git.name").unwrap_or_else(|_| format!("{review_agent}[bot]"));
    let git_email = crate::config::get("git.email")
        .unwrap_or_else(|_| format!("{review_agent}[bot]@users.noreply.github.com"));
    let review_timeout_secs: u64 = crate::config::get("workflow.review_timeout_seconds")
        .ok()
        .and_then(|s| s.parse().ok())
        .unwrap_or_else(|| {
            crate::config::get("workflow.timeout_seconds")
                .ok()
                .and_then(|s| s.parse().ok())
                .unwrap_or(1800)
        })
        .max(1800);

    let invocation = runner::agent::AgentInvocation {
        agent: review_agent.clone(),
        model: review_model.clone(),
        work_dir: worktree_path.clone(),
        system_prompt: runner::agent::review_system_prompt(),
        agent_message: review_prompt,
        task_id: review_task_id.clone(),
        disallowed_tools: build_review_disallowed_tools(&worktree_path),
        git_author_name: git_name,
        git_author_email: git_email,
        output_file: output_file.clone(),
        timeout_seconds: review_timeout_secs,
        repo: repo.to_string(),
        attempt: review_attempt,
        sandbox: sandboxed,
        main_project_dir,
        branch: branch_name.clone(),
    };

    let head_sha = match GhHttp::new() {
        Ok(gh) => gh.get_pr(repo, pr_number).await.ok().map(|pr| pr.head.sha),
        Err(_) => None,
    };

    Ok(ReviewPhase::Ready(ReviewContext {
        store_id,
        had_prev_push_failure,
        review_attempt,
        worktree_path,
        branch_name,
        pr_number,
        head_sha,
        default_branch,
        review_agent,
        review_model,
        review_task_id,
        review_attempt_dir,
        output_file,
        invocation,
        same_agent_fallback,
    }))
}

#[allow(clippy::too_many_arguments)]
async fn complete_review_run(
    store: &Arc<TaskStore>,
    run_id: Option<i64>,
    exit_code: Option<i32>,
    stdout: &str,
    stderr: &str,
    parsed: &str,
    outcome: &str,
    error: &str,
    tokens: RunTokenUsage,
) {
    if let Some(run_id) = run_id {
        if let Err(e) = store
            .complete_run(&CompleteRun {
                run_id,
                exit_code,
                stdout,
                stderr,
                parsed,
                outcome,
                error,
                tokens,
            })
            .await
        {
            tracing::warn!(
                run_id,
                error = %e,
                "failed to record review run completion in audit trail"
            );
        }
    }
}

async fn invoke_review_agent(
    task: &ExternalTask,
    tmux: &Arc<TmuxManager>,
    ctx: &ReviewContext,
    repo: &str,
    store: &Arc<TaskStore>,
) -> ReviewPhase<ReviewRun> {
    let run_id = if let Some(sid) = ctx.store_id {
        match store
            .start_run(&StartRun {
                task_id: sid,
                attempt: ctx.review_attempt as i32,
                run_type: "review",
                agent: &ctx.review_agent,
                model: ctx.review_model.as_deref().unwrap_or(""),
                command: "",
                prompt: "",
            })
            .await
        {
            Ok(run_id) => Some(run_id),
            Err(e) => {
                tracing::warn!(
                    task_id = task.id.0,
                    error = %e,
                    "failed to record review run start in audit trail"
                );
                None
            }
        }
    } else {
        None
    };

    let session = match runner::agent::spawn_in_tmux(tmux, &ctx.invocation).await {
        Ok(session) => session,
        Err(e) => {
            tracing::error!(task_id = task.id.0, error = %e, "failed to spawn review agent");
            complete_review_run(
                store,
                run_id,
                None,
                "",
                &e.to_string(),
                "",
                "failed",
                &format!("spawn failed: {e}"),
                RunTokenUsage::default(),
            )
            .await;
            return ReviewPhase::EarlyReturn(ReviewDecision::Failed(format!("spawn failed: {e}")));
        }
    };

    let start_comment = review_started_comment(&ctx.review_agent, ctx.review_model_str());
    match GhHttp::new() {
        Ok(gh) => {
            post_review_status(
                &gh,
                repo,
                ctx.head_sha.as_deref(),
                "pending",
                "Automated review running",
            )
            .await;
            if let Err(e) = gh
                .add_comment(repo, &ctx.pr_number.to_string(), &start_comment)
                .await
            {
                tracing::debug!(
                    task_id = task.id.0,
                    pr_number = ctx.pr_number,
                    error = %e,
                    "failed to post review-started comment"
                );
            }
        }
        Err(e) => {
            tracing::debug!(
                task_id = task.id.0,
                pr_number = ctx.pr_number,
                error = %e,
                "failed to create GitHub client for review-started comment"
            );
        }
    }

    let poll_interval = std::time::Duration::from_secs(5);
    let timeout_duration = std::time::Duration::from_secs(ctx.invocation.timeout_seconds);
    let wait_result = tokio::time::timeout(
        timeout_duration,
        tmux.wait_for_completion(&session, poll_interval),
    )
    .await;

    match wait_result {
        Ok(Ok(_)) => {
            tracing::info!(task_id = task.id.0, "review agent completed");
            let _ = tmux.kill_session(&session).await;
            ReviewPhase::Ready(ReviewRun { run_id })
        }
        Ok(Err(e)) => {
            tracing::warn!(task_id = task.id.0, error = %e, "review agent process exited with error");
            let _ = tmux.kill_session(&session).await;

            // Before recording failure, check if this is a "successful exit with non-zero
            // code" agent like kimi/minimax. Read the output file and check for a valid
            // result with is_error=false — if found, treat as success and proceed to
            // parse_review_output where the real result will be extracted.
            let exit_code: i32 = tokio::task::spawn_blocking({
                let review_attempt_dir = ctx.review_attempt_dir.clone();
                move || {
                    std::fs::read_to_string(review_attempt_dir.join("exit.txt"))
                        .ok()
                        .and_then(|s| s.trim().parse().ok())
                        .unwrap_or(-1)
                }
            })
            .await
            .unwrap_or(-1);

            if exit_code != 0 {
                // Try reading output via read_output_file which checks both the primary path
                // and all prior attempt dirs as a fallback. This handles agents like kimi/minimax
                // that exit non-zero even on success, as well as cases where the current attempt's
                // output.json was not written but a prior attempt's output contains a valid result.
                let raw_output =
                    runner::response::read_output_file(&ctx.review_task_id, &ctx.output_file, repo)
                        .await;
                let agent_result =
                    runner::agents::find_agent_result(&ctx.review_agent, &raw_output);
                // Treat as success if we found a valid result without is_error set
                if agent_result.as_ref().map(|r| !r.is_error).unwrap_or(false) {
                    tracing::info!(
                        task_id = task.id.0,
                        agent = %ctx.review_agent,
                        exit_code,
                        output_found = !raw_output.is_empty(),
                        "review agent succeeded despite non-zero exit — proceeding to parse output"
                    );
                    return ReviewPhase::Ready(ReviewRun { run_id });
                }
            }

            tracing::error!(task_id = task.id.0, error = %e, "review agent failed");
            complete_review_run(
                store,
                run_id,
                Some(exit_code),
                "",
                &e.to_string(),
                "",
                "failed",
                &format!("agent error: {e}"),
                RunTokenUsage::default(),
            )
            .await;
            ReviewPhase::EarlyReturn(ReviewDecision::Failed(format!("agent error: {e}")))
        }
        Err(_) => {
            tracing::error!(task_id = task.id.0, "review agent timed out");
            let _ = tmux.kill_session(&session).await;
            if let Some(model) = ctx.review_model.as_deref() {
                // Structural review failure: the model cannot finish review inside
                // the allowed time. Use persistent backoff (4h base → 7d max) so
                // a successful agent run on the same model does not reset the
                // failure counter and let this model be re-selected for review.
                crate::engine::cooldown::record_persistent_model_failure(&ctx.review_agent, model)
                    .await;
            }
            complete_review_run(
                store,
                run_id,
                Some(-1),
                "",
                "timeout",
                "",
                "timeout",
                "timeout",
                RunTokenUsage::default(),
            )
            .await;
            ReviewPhase::EarlyReturn(ReviewDecision::Failed("timeout".to_string()))
        }
    }
}

async fn record_review_agent_failure(
    task_id: &str,
    review_agent: &str,
    review_model: Option<&str>,
    err: &runner::agents::AgentError,
) {
    match err {
        runner::agents::AgentError::RateLimit { message }
        | runner::agents::AgentError::Auth { message } => {
            tracing::warn!(
                task_id,
                agent = %review_agent,
                "review agent hit error — adding to cooldown"
            );
            if let Some(reason) = crate::engine::cooldown::detect_credit_exhaustion(message) {
                if reason == crate::engine::cooldown::CreditExhaustionReason::BillingCycleExhausted
                {
                    if let Some(model) = review_model {
                        // Quota is model-scoped: don't penalize the whole agent.
                        crate::engine::cooldown::record_persistent_model_failure(
                            review_agent,
                            model,
                        )
                        .await;
                    } else {
                        crate::engine::cooldown::record_credit_exhaustion(review_agent, reason)
                            .await;
                    }
                } else {
                    crate::engine::cooldown::record_credit_exhaustion(review_agent, reason).await;
                }
            } else {
                runner::response::record_agent_failure_with_message(review_agent, &err.to_string())
                    .await;
            }
        }
        runner::agents::AgentError::InvalidResponse { .. } => {
            // Structural failure: model produced no parseable output at all.
            // Apply persistent backoff (4h base → 7d max) — same rationale as
            // silent exit-0 in the agent runner. A model that cannot produce
            // ANY output for a review is unlikely to self-heal on the next try.
            if let Some(model) = review_model {
                crate::engine::cooldown::record_persistent_model_failure(review_agent, model).await;
            }
        }
        runner::agents::AgentError::ModelUnavailable { .. } => {
            // Model not available (provider error, model not found, etc.)
            // Apply persistent model cooldown (4h base → 7d max) per settled
            // architecture (AGENTS.md:325-328). The agent itself is not
            // penalized — its other models still route normally.
            if let Some(model) = review_model {
                crate::engine::cooldown::record_persistent_model_failure(review_agent, model).await;
            }
        }
        _ => {
            // Generic failure (AgentFailed, NetworkError, etc.)
            // Apply standard model backoff (5min→4h max) to prevent immediate re-selection.
            if let Some(model) = review_model {
                crate::engine::cooldown::record_model_failure(review_agent, model).await;
            }
        }
    }
}

fn review_decision_from_response(
    review_response: runner::response::ReviewResponse,
) -> ReviewDecision {
    match review_response.decision.as_str() {
        "approve" => ReviewDecision::Approve,
        "request_changes" => ReviewDecision::RequestChanges {
            notes: review_response.notes,
            issues: review_response.issues,
        },
        _ => ReviewDecision::Failed(format!("unknown decision: {}", review_response.decision)),
    }
}

fn review_decision_name(decision: &ReviewDecision) -> &'static str {
    match decision {
        ReviewDecision::Approve => "approve",
        ReviewDecision::RequestChanges { .. } => "request_changes",
        ReviewDecision::Failed(_) => "failed",
        ReviewDecision::Blocked(_) => "blocked",
        ReviewDecision::Skipped => "skipped",
        ReviewDecision::Rerouted => "rerouted",
    }
}

async fn parse_review_output(
    task: &ExternalTask,
    repo: &str,
    ctx: &ReviewContext,
    run: ReviewRun,
    store: &Arc<TaskStore>,
) -> ReviewPhase<ParsedReview> {
    let (file_exists, file_size, exit_code, stderr) = {
        let output_file_clone = ctx.output_file.clone();
        let review_attempt_dir_clone = ctx.review_attempt_dir.clone();
        match tokio::task::spawn_blocking(move || {
            let file_exists = output_file_clone.exists();
            let file_size = std::fs::metadata(&output_file_clone)
                .map(|m| m.len())
                .unwrap_or(0);
            let exit_code = std::fs::read_to_string(review_attempt_dir_clone.join("exit.txt"))
                .ok()
                .and_then(|s| s.trim().parse::<i32>().ok())
                .unwrap_or(-1);
            let stderr = std::fs::read_to_string(review_attempt_dir_clone.join("stderr.txt"))
                .unwrap_or_default();
            (file_exists, file_size, exit_code, stderr)
        })
        .await
        {
            Ok(tuple) => tuple,
            Err(e) => {
                tracing::error!(
                    task_id = task.id.0,
                    error = %e,
                    "spawn_blocking panicked reading review output metadata"
                );
                (false, 0, -1, format!("spawn_blocking failed: {e}"))
            }
        }
    };

    tracing::info!(
        task_id = task.id.0,
        path = %ctx.output_file.display(),
        file_exists,
        file_size,
        "reading review agent output"
    );

    let raw_output =
        runner::response::read_output_file(&ctx.review_task_id, &ctx.output_file, repo).await;
    let raw_output = runner::agents::claude::strip_tty_control_lines(&raw_output);
    let has_result_envelope =
        runner::agents::claude::find_ndjson_result_line(&raw_output).is_some();
    let agent_result = runner::agents::find_agent_result(&ctx.review_agent, &raw_output);
    let token_usage = RunTokenUsage {
        input_tokens: agent_result
            .as_ref()
            .and_then(|r| r.input_tokens)
            .unwrap_or(0),
        output_tokens: agent_result
            .as_ref()
            .and_then(|r| r.output_tokens)
            .unwrap_or(0),
        total_cost_usd: agent_result
            .as_ref()
            .and_then(|r| r.cost_usd)
            .unwrap_or(0.0),
        duration_secs: 0.0,
    };

    let agent_result_is_error = agent_result.as_ref().map(|r| r.is_error).unwrap_or(false);
    let ndjson_completed_no_error = raw_output.contains("\"terminal_reason\":\"completed\"")
        && !raw_output.contains("\"is_error\":true");
    // Do not short-circuit completed NDJSON sessions before rescue parsing.
    let is_hard_failure =
        !ndjson_completed_no_error && (raw_output.is_empty() || agent_result_is_error);

    if is_hard_failure {
        tracing::warn!(
            task_id = task.id.0,
            exit_code,
            raw_output_len = raw_output.len(),
            stderr_len = stderr.len(),
            "review agent: entering error path"
        );
        let agent_runner = runner::agents::get_runner(&ctx.review_agent);
        let err = agent_runner.classify_error(exit_code, &raw_output, &stderr);
        record_review_agent_failure(
            &task.id.0,
            &ctx.review_agent,
            ctx.review_model.as_deref(),
            &err,
        )
        .await;
        tracing::error!(task_id = task.id.0, error = %err, "review agent failed");
        let stored_error = format_review_error(&ctx.review_agent, &err);
        complete_review_run(
            store,
            run.run_id,
            Some(exit_code),
            &raw_output,
            &stderr,
            "",
            outcome_for_agent_error(&err),
            &stored_error,
            token_usage,
        )
        .await;
        return ReviewPhase::EarlyReturn(ReviewDecision::Failed(stored_error));
    }

    let text_for_review = agent_result
        .as_ref()
        .map(|r| r.result_text.clone())
        .filter(|text| !text.is_empty())
        .unwrap_or_else(|| {
            tracing::debug!(
                task_id = task.id.0,
                agent = %ctx.review_agent,
                "review agent: per-agent extractor returned no text, using raw output"
            );
            raw_output.clone()
        });

    let review_response = match runner::response::parse_review_response(&text_for_review) {
        Ok(response) => response,
        Err(e) => {
            if let Some(response) = runner::response::infer_review_response(&text_for_review) {
                tracing::warn!(
                    task_id = task.id.0,
                    agent = %ctx.review_agent,
                    "review response parsed via keyword fallback"
                );
                response
            } else {
                // Guard: if the raw NDJSON signals a completed session
                // (terminal_reason:completed without is_error:true), do NOT run
                // generic pattern-based rate-limit detection over it. Token
                // counts in the telemetry blob can contain bare "429" values
                // that would otherwise be misclassified as HTTP 429 errors.
                let ndjson_completed = ndjson_completed_no_error;

                // Missing type:result (kimi/minimax) or terminal_reason:completed: recover
                // from assistant text. Never MiniMax parse_response, which synthesizes status=done
                if ndjson_completed || !has_result_envelope {
                    let assistant_text = runner::agents::collect_assistant_messages_text(
                        &ctx.review_agent,
                        &raw_output,
                    );
                    if !assistant_text.is_empty() {
                        if let Ok(response) =
                            runner::response::parse_review_response(&assistant_text)
                        {
                            tracing::debug!(
                                task_id = task.id.0,
                                agent = %ctx.review_agent,
                                "found review response JSON in assistant message (terminal_reason:completed)"
                            );
                            let review_notes_for_comment = response.notes.clone();
                            let decision = review_decision_from_response(response);
                            return ReviewPhase::Ready(ParsedReview {
                                run_id: run.run_id,
                                exit_code,
                                stderr,
                                raw_output,
                                text_for_review: assistant_text,
                                token_usage,
                                review_notes_for_comment,
                                decision,
                            });
                        }
                        if let Some(response) =
                            runner::response::infer_review_response(&assistant_text)
                        {
                            tracing::warn!(
                                task_id = task.id.0,
                                agent = %ctx.review_agent,
                                "inferred review response from assistant message (terminal_reason:completed)"
                            );
                            let review_notes_for_comment = response.notes.clone();
                            let decision = review_decision_from_response(response);
                            return ReviewPhase::Ready(ParsedReview {
                                run_id: run.run_id,
                                exit_code,
                                stderr,
                                raw_output,
                                text_for_review: assistant_text,
                                token_usage,
                                review_notes_for_comment,
                                decision,
                            });
                        }
                    }
                }

                // Scan only bounded head/tail windows, not the full review text.
                // The full transcript may contain agent work product (e.g. a
                // diff or markdown report quoting an `outcome`/`rate_limit`
                // table cell) that incidentally matches these patterns; real
                // CLI errors appear at the start or end of the output. See
                // `classify_from_text` in runner::agents for the same rationale.
                let scan_head = runner::agents::patterns::safe_head(
                    &text_for_review,
                    runner::agents::patterns::RATE_LIMIT_SCAN_HEAD_BYTES,
                );
                let scan_tail = runner::agents::patterns::safe_tail(
                    &text_for_review,
                    runner::agents::patterns::RATE_LIMIT_SCAN_TAIL_BYTES,
                );

                if let Some(err) = (!ndjson_completed)
                    .then(|| {
                        runner::agents::patterns::detect_rate_limit(scan_head)
                            .or_else(|| runner::agents::patterns::detect_rate_limit(scan_tail))
                    })
                    .flatten()
                {
                    let runner::agents::AgentError::RateLimit { message } = &err else {
                        unreachable!("detect_rate_limit only returns AgentError::RateLimit")
                    };
                    tracing::warn!(
                        task_id = task.id.0,
                        agent = %ctx.review_agent,
                        "review agent hit rate limit — adding to cooldown"
                    );
                    // Route through record_review_agent_failure/outcome_for_agent_error
                    // (not the generic record_agent_failure_with_message + hardcoded
                    // "rate_limit" string) so billing-cycle exhaustion messages caught by
                    // this inline text scan get the same model-scoped persistent cooldown
                    // and outcome classification as the AgentError-based failure path
                    // above (issue #3551, gap left by #3530).
                    let summary = runner::fallback::summarize_rate_limit_error(message);
                    let stored_error = format!("{} rate limit: {summary}", ctx.review_agent);
                    record_review_agent_failure(
                        &task.id.0,
                        &ctx.review_agent,
                        ctx.review_model.as_deref(),
                        &err,
                    )
                    .await;
                    complete_review_run(
                        store,
                        run.run_id,
                        Some(exit_code),
                        &raw_output,
                        &stderr,
                        &text_for_review,
                        outcome_for_agent_error(&err),
                        &stored_error,
                        token_usage,
                    )
                    .await;
                    return ReviewPhase::EarlyReturn(ReviewDecision::Failed(stored_error));
                }

                if let Some(err) = runner::agents::patterns::detect_auth_error(scan_tail) {
                    let runner::agents::AgentError::Auth { message } = &err else {
                        unreachable!("detect_auth_error only returns AgentError::Auth")
                    };
                    tracing::warn!(
                        task_id = task.id.0,
                        agent = %ctx.review_agent,
                        "review agent hit auth error — adding to cooldown"
                    );
                    let stored_error = format!("auth error: {message}");
                    record_review_agent_failure(
                        &task.id.0,
                        &ctx.review_agent,
                        ctx.review_model.as_deref(),
                        &err,
                    )
                    .await;
                    complete_review_run(
                        store,
                        run.run_id,
                        Some(exit_code),
                        &raw_output,
                        &stderr,
                        &text_for_review,
                        outcome_for_agent_error(&err),
                        &stored_error,
                        token_usage,
                    )
                    .await;
                    return ReviewPhase::EarlyReturn(ReviewDecision::Failed(stored_error));
                }

                // The model was cut off for exceeding its output/reasoning
                // token budget (opencode step_finish reason=length), not
                // because it produced malformed prose. Like a timeout, this is
                // a structural inability to finish review under the current
                // budget, so use the persistent 4h-7d cooldown rather than the
                // generic success-resettable model failure counter.
                if agent_result
                    .as_ref()
                    .map(|r| r.truncated_by_length)
                    .unwrap_or(false)
                {
                    tracing::warn!(
                        task_id = task.id.0,
                        agent = %ctx.review_agent,
                        model = ?ctx.review_model,
                        "review agent output truncated: step_finish reason=length \
                         (token budget exceeded before completion)"
                    );
                    if let Some(model) = ctx.review_model.as_deref() {
                        crate::engine::cooldown::record_persistent_model_failure(
                            &ctx.review_agent,
                            model,
                        )
                        .await;
                    }
                    let stored_error = "truncated: review output was cut off before completion \
                        (output/reasoning token budget exceeded)"
                        .to_string();
                    complete_review_run(
                        store,
                        run.run_id,
                        Some(exit_code),
                        &raw_output,
                        &stderr,
                        &text_for_review,
                        "truncated",
                        &stored_error,
                        token_usage,
                    )
                    .await;
                    return ReviewPhase::EarlyReturn(ReviewDecision::Failed(stored_error));
                }

                // No envelope and last content is thinking/tool_use: retry without
                // the persistent format-failure cooldown (issue #3734)
                if runner::agents::claude::review_session_exited_mid_thought(&raw_output) {
                    tracing::warn!(
                        task_id = task.id.0,
                        agent = %ctx.review_agent,
                        model = ?ctx.review_model,
                        "review session exited mid-thought with no result envelope \
                         — storing as incomplete and retrying without persistent cooldown"
                    );
                    let stored_error =
                        "incomplete: review session exited before producing a decision".to_string();
                    complete_review_run(
                        store,
                        run.run_id,
                        Some(exit_code),
                        &raw_output,
                        &stderr,
                        &text_for_review,
                        "incomplete",
                        &stored_error,
                        token_usage,
                    )
                    .await;
                    return ReviewPhase::EarlyReturn(ReviewDecision::Failed(stored_error));
                }

                tracing::error!(
                    task_id = task.id.0,
                    error = %e,
                    agent = %ctx.review_agent,
                    output = %text_for_review.chars().take(300).collect::<String>(),
                    raw_output = %raw_output.chars().take(1000).collect::<String>(),
                    "failed to parse review response"
                );
                if let Some(model) = ctx.review_model.as_deref() {
                    // Finished output that is not a review decision: persistent backoff
                    // (4h base → 7d max) so a later success cannot immediately re-select it
                    crate::engine::cooldown::record_persistent_model_failure(
                        &ctx.review_agent,
                        model,
                    )
                    .await;
                }
                complete_review_run(
                    store,
                    run.run_id,
                    Some(exit_code),
                    &raw_output,
                    &stderr,
                    &text_for_review,
                    "parse_error",
                    &format!("parse error: {e}"),
                    token_usage,
                )
                .await;
                return ReviewPhase::EarlyReturn(ReviewDecision::Failed(format!(
                    "parse error: {e}"
                )));
            }
        }
    };

    let review_notes_for_comment = review_response.notes.clone();
    let decision = review_decision_from_response(review_response);

    tracing::info!(
        task_id = task.id.0,
        pr_number = ctx.pr_number,
        decision = ?decision,
        "review agent decision received"
    );
    store_log_activity(
        &Some(Arc::clone(store)),
        repo,
        &task.id.0,
        "review_decision",
        None,
        None,
        Some(&ctx.review_agent),
        ctx.review_model.as_deref(),
        Some(&serde_json::json!({
            "decision": review_decision_name(&decision),
            "pr_number": ctx.pr_number,
        })),
    )
    .await;

    ReviewPhase::Ready(ParsedReview {
        run_id: run.run_id,
        exit_code,
        stderr,
        raw_output,
        text_for_review,
        token_usage,
        review_notes_for_comment,
        decision,
    })
}

async fn finalize_review_run(
    task: &ExternalTask,
    ctx: &ReviewContext,
    parsed: &ParsedReview,
    store: &Arc<TaskStore>,
) {
    let outcome = match &parsed.decision {
        ReviewDecision::Approve
        | ReviewDecision::RequestChanges { .. }
        | ReviewDecision::Skipped
        | ReviewDecision::Rerouted => "success",
        ReviewDecision::Failed(_) | ReviewDecision::Blocked(_) => "failed",
    };
    let error = match &parsed.decision {
        ReviewDecision::Failed(error) | ReviewDecision::Blocked(error) => error.as_str(),
        _ => "",
    };

    if outcome == "success" {
        // Mirror the runner's success path so review-only models don't ratchet
        // into multi-day cooldowns from isolated failures never offset by resets.
        crate::engine::cooldown::record_agent_success(&ctx.review_agent, ctx.review_model_str())
            .await;
    }

    complete_review_run(
        store,
        parsed.run_id,
        Some(parsed.exit_code),
        &parsed.raw_output,
        &parsed.stderr,
        &parsed.text_for_review,
        outcome,
        error,
        parsed.token_usage,
    )
    .await;

    if let Some(store_id) = ctx.store_id {
        if parsed.token_usage.input_tokens > 0 || parsed.token_usage.output_tokens > 0 {
            let model = if ctx.review_model_str().is_empty() {
                "unknown"
            } else {
                ctx.review_model_str()
            };
            if let Err(e) = store
                .store_tokens(
                    store_id,
                    parsed.token_usage.input_tokens,
                    parsed.token_usage.output_tokens,
                    model,
                )
                .await
            {
                tracing::warn!(
                    task_id = task.id.0,
                    ?e,
                    "failed to store review agent token usage"
                );
            }
        }
    }
}

async fn restore_review_config_if_needed(
    task_id: &str,
    worktree_path: &std::path::Path,
    default_branch: &str,
) {
    let _ = Command::new("git")
        .args(["checkout", "HEAD", "--", ".orch.yml"])
        .current_dir(worktree_path)
        .status()
        .await;

    let merge_base = Command::new("git")
        .args(["merge-base", "HEAD", default_branch])
        .current_dir(worktree_path)
        .output()
        .await
        .ok()
        .filter(|output| output.status.success())
        .and_then(|output| String::from_utf8(output.stdout).ok())
        .map(|s| s.trim().to_string())
        .unwrap_or_default();

    if merge_base.is_empty() {
        return;
    }

    let orch_yml_changed = Command::new("git")
        .args([
            "diff",
            "--name-only",
            &merge_base,
            "HEAD",
            "--",
            ".orch.yml",
        ])
        .current_dir(worktree_path)
        .output()
        .await
        .ok()
        .filter(|output| output.status.success())
        .map(|output| !output.stdout.is_empty())
        .unwrap_or(false);

    if !orch_yml_changed {
        return;
    }

    tracing::warn!(
        task_id,
        "review agent modified .orch.yml (forbidden); restoring to merge-base version"
    );
    let restore_ok = Command::new("git")
        .args(["checkout", &merge_base, "--", ".orch.yml"])
        .current_dir(worktree_path)
        .status()
        .await
        .map(|status| status.success())
        .unwrap_or(false);

    if !restore_ok {
        return;
    }

    let staged = Command::new("git")
        .args(["diff", "--cached", "--name-only", "--", ".orch.yml"])
        .current_dir(worktree_path)
        .output()
        .await
        .ok()
        .filter(|output| output.status.success())
        .map(|output| !output.stdout.is_empty())
        .unwrap_or(false);

    if staged {
        let _ = Command::new("git")
            .args([
                "-c",
                "commit.gpgsign=false",
                "commit",
                "-m",
                "revert: restore .orch.yml (review agent must not modify project config)",
            ])
            .current_dir(worktree_path)
            .status()
            .await;
    }
}

fn is_missing_path_error_message(err: &str) -> bool {
    err.to_ascii_lowercase()
        .contains("no such file or directory")
}

async fn push_review_branch(
    task: &ExternalTask,
    repo: &str,
    ctx: &ReviewContext,
    store: &Arc<TaskStore>,
) -> Result<(), String> {
    let mut worktree_path = ctx.worktree_path.clone();
    let mut branch_name = ctx.branch_name.clone();
    let mut retried_missing_worktree = false;

    let push_result = loop {
        match runner::git_ops::push_branch(&worktree_path, &branch_name, &ctx.default_branch).await
        {
            Ok(ok) => break Ok(ok),
            Err(e)
                if !retried_missing_worktree
                    && !worktree_path.exists()
                    && is_missing_path_error_message(&e.to_string()) =>
            {
                retried_missing_worktree = true;
                tracing::warn!(
                    task_id = task.id.0,
                    worktree = %worktree_path.display(),
                    branch = %branch_name,
                    "review push failed because worktree disappeared; attempting one-time recovery"
                );
                match recover_missing_review_worktree(task, repo, store).await {
                    Ok(recovered) => {
                        worktree_path = recovered.work_dir;
                        branch_name = recovered.branch;
                        continue;
                    }
                    Err(recover_err) => {
                        break Err(anyhow::anyhow!(
                            "missing review worktree recovery failed after push error: {recover_err}"
                        ));
                    }
                }
            }
            Err(e) => break Err(e),
        }
    };

    match push_result {
        Ok(_) => {
            store_log_activity(
                &Some(Arc::clone(store)),
                repo,
                &task.id.0,
                "push",
                None,
                None,
                Some(&ctx.review_agent),
                ctx.review_model.as_deref(),
                Some(&serde_json::json!({
                    "status": "ok",
                    "branch": branch_name.clone(),
                    "default_branch": ctx.default_branch.clone(),
                    "phase": "review",
                })),
            )
            .await;
            if ctx.had_prev_push_failure {
                if let Err(e) = store_set_result(
                    &Some(Arc::clone(store)),
                    repo,
                    &task.id.0,
                    &[
                        ("last_error", serde_json::json!("")),
                        ("push_failures", serde_json::json!(0)),
                    ],
                )
                .await
                {
                    tracing::warn!(task_id = task.id.0, err = %e, "failed to reset push_failures — counter may drift");
                }
            }
            Ok(())
        }
        Err(e) => {
            store_log_activity(
                &Some(Arc::clone(store)),
                repo,
                &task.id.0,
                "push",
                None,
                None,
                Some(&ctx.review_agent),
                ctx.review_model.as_deref(),
                Some(&serde_json::json!({
                    "status": "error",
                    "branch": branch_name.clone(),
                    "default_branch": ctx.default_branch.clone(),
                    "phase": "review",
                    "error": e.to_string(),
                })),
            )
            .await;
            tracing::error!(
                task_id = task.id.0,
                branch = %ctx.branch_name,
                error = %e,
                "review push failed"
            );
            store_set(
                &Some(Arc::clone(store)),
                repo,
                &task.id.0,
                &[("last_error", serde_json::json!(format!("push failed: {e}")))],
            )
            .await;
            Err(e.to_string())
        }
    }
}

fn build_pr_review_comment(decision: &ReviewDecision, review_notes_for_comment: &str) -> String {
    match decision {
        ReviewDecision::Approve => {
            format!(
                "## Automated Review \u{2014} Approve\n\n{}",
                review_notes_for_comment
            )
        }
        ReviewDecision::RequestChanges { notes, issues } => {
            let mut body = format!(
                "## Automated Review \u{2014} Changes Requested\n\n{}\n",
                notes
            );
            if !issues.is_empty() {
                body.push_str("\n**Issues Found:**\n");
                for issue in issues {
                    body.push_str(&format!(
                        "- `{}` line {}: {} [{}]\n",
                        issue.file,
                        issue
                            .line
                            .map(|line| line.to_string())
                            .unwrap_or_else(|| "?".to_string()),
                        issue.description,
                        issue.severity
                    ));
                }
            }
            body
        }
        _ => String::new(),
    }
}

const REVIEW_STATUS_CONTEXT: &str = "orch/review";

/// Commit status state and description for a review decision. Other decisions post nothing.
fn review_status(decision: &ReviewDecision) -> Option<(&'static str, &'static str)> {
    match decision {
        ReviewDecision::Approve => Some(("success", "Approved by automated review")),
        ReviewDecision::RequestChanges { .. } => {
            Some(("failure", "Automated review requested changes"))
        }
        _ => None,
    }
}

/// Post `orch/review` on the SHA the review started from. A new push moves the PR head, so an
/// old decision never covers a new commit.
async fn post_review_status(
    gh: &GhHttp,
    repo: &str,
    sha: Option<&str>,
    state: &str,
    description: &str,
) {
    let Some(sha) = sha else {
        return;
    };
    if let Err(e) = gh
        .post_commit_status(repo, sha, state, REVIEW_STATUS_CONTEXT, description)
        .await
    {
        tracing::warn!(sha, error = %e, "failed to post orch/review commit status");
    }
}

async fn post_review_comment(
    task: &ExternalTask,
    repo: &str,
    ctx: &ReviewContext,
    decision: &ReviewDecision,
    review_notes_for_comment: &str,
) -> anyhow::Result<()> {
    let gh = match GhHttp::new() {
        Ok(gh) => gh,
        Err(e) => {
            tracing::warn!(
                task_id = task.id.0,
                error = %e,
                "failed to create GH client for review comment"
            );
            return Ok(());
        }
    };
    let mut pr_comment = build_pr_review_comment(decision, review_notes_for_comment);
    if ctx.same_agent_fallback && !pr_comment.is_empty() {
        pr_comment.push_str(
            "\n> Note: no other agent was available, so this PR was reviewed by the same agent that wrote the code.\n",
        );
    }

    if let Some((state, description)) = review_status(decision) {
        post_review_status(&gh, repo, ctx.head_sha.as_deref(), state, description).await;
    }

    if !pr_comment.is_empty() {
        let footer =
            attribution_footer("Reviewed", &ctx.review_agent, Some(ctx.review_model_str()));
        let pr_comment_with_footer = format!("{}{}", pr_comment, footer);
        if let Err(e) = gh
            .add_comment(repo, &ctx.pr_number.to_string(), &pr_comment_with_footer)
            .await
        {
            tracing::warn!(
                task_id = task.id.0,
                pr_number = ctx.pr_number,
                error = %e,
                "failed to post automated review comment on PR"
            );
        }
    }

    Ok(())
}

#[allow(clippy::too_many_arguments)]
async fn apply_review_decision(
    task: &ExternalTask,
    backend: &Arc<dyn ExternalBackend>,
    repo: &str,
    task_manager: &Arc<TaskManager>,
    store: &Arc<TaskStore>,
    ctx: &ReviewContext,
    parsed: &ParsedReview,
    router_config: &crate::engine::router::RouterConfig,
) -> anyhow::Result<ReviewDecision> {
    restore_review_config_if_needed(&task.id.0, &ctx.worktree_path, &ctx.default_branch).await;

    if let Err(e) = push_review_branch(task, repo, ctx, store).await {
        return Ok(ReviewDecision::Failed(format!("push failed: {e}")));
    }

    post_review_comment(
        task,
        repo,
        ctx,
        &parsed.decision,
        &parsed.review_notes_for_comment,
    )
    .await?;

    match &parsed.decision {
        ReviewDecision::Approve => {
            record_approved_sha(store, repo, &task.id.0, ctx.pr_number, &ctx.worktree_path).await;
            // Always attempt auto-merge on approval. Tasks are only marked
            // Done when the PR is actually merged (detected by review_poll).
            // Never mark Done on approval alone — that closes the issue
            // before the code lands.
            if let Err(e) = auto_merge_pr(
                task,
                &ctx.branch_name,
                backend,
                repo,
                &ctx.review_agent,
                ctx.review_model_str(),
                task_manager,
                store,
            )
            .await
            {
                let error_msg = e.to_string();
                if error_msg.contains("not yet computed") {
                    tracing::warn!(
                        task_id = task.id.0,
                        pr_number = ctx.pr_number,
                        branch = %ctx.branch_name,
                        error = %e,
                        "auto-merge deferred — PR mergeability not yet computed"
                    );
                } else {
                    tracing::error!(
                        task_id = task.id.0,
                        pr_number = ctx.pr_number,
                        branch = %ctx.branch_name,
                        error = %e,
                        "auto-merge failed"
                    );
                }
                return Ok(ReviewDecision::Failed(format!("merge failed: {e}")));
            }
            Ok(ReviewDecision::Approve)
        }
        ReviewDecision::RequestChanges { notes, issues } => {
            clear_approved_sha(store, repo, &task.id.0, ctx.pr_number).await;
            handle_review_changes(
                task,
                notes,
                issues,
                backend,
                repo,
                ctx.pr_number,
                &ctx.review_agent,
                ctx.review_model_str(),
                task_manager,
                store,
                router_config,
            )
            .await?;
            Ok(parsed.decision.clone())
        }
        _ => Ok(parsed.decision.clone()),
    }
}

async fn recover_missing_review_worktree(
    task: &ExternalTask,
    repo: &str,
    store: &Arc<TaskStore>,
) -> anyhow::Result<worktree::WorktreeSetup> {
    let project_dir = crate::config::get_projects_with_paths()?
        .into_iter()
        .find(|(candidate_repo, _)| candidate_repo == repo)
        .map(|(_, project_dir)| project_dir)
        .ok_or_else(|| anyhow::anyhow!("no project directory configured for repo {repo}"))?;

    tracing::warn!(
        task_id = task.id.0,
        repo,
        project_dir = %project_dir.display(),
        "review worktree missing, recreating via normal setup"
    );

    let store_ref = Some(Arc::clone(store));
    let wt = worktree::setup_worktree(&task.id.0, &task.title, &project_dir, &store_ref, repo)
        .await
        .context("recreating missing review worktree")?;

    if !wt.work_dir.exists() {
        anyhow::bail!(
            "recreated review worktree at {} for task {} is still missing",
            wt.work_dir.display(),
            task.id.0
        );
    }

    Ok(wt)
}

/// Count commits on the worktree branch that are ahead of `origin/<default_branch>`.
///
/// Returns `Err` when the git command cannot determine the count (e.g., missing
/// remote ref, non-zero exit, unparseable output). Callers must NOT treat this
/// as zero — a failure means the ancestry relationship is unknown.
async fn count_ahead_commits(
    worktree_path: &std::path::Path,
    default_branch: &str,
) -> Result<u64, String> {
    let worktree_str = worktree_path.to_str().ok_or_else(|| {
        format!(
            "worktree path contains non-UTF-8 characters: {:?}",
            worktree_path
        )
    })?;

    let output = tokio::process::Command::new("git")
        .args([
            "-C",
            worktree_str,
            "rev-list",
            "--count",
            &format!("origin/{default_branch}..HEAD"),
        ])
        .output()
        .await
        .map_err(|e| format!("git rev-list command failed to start: {e}"))?;

    if !output.status.success() {
        let stderr = String::from_utf8_lossy(&output.stderr).trim().to_string();
        return Err(format!(
            "git rev-list exited with {:?}: {}",
            output.status.code(),
            if stderr.is_empty() {
                String::from_utf8_lossy(&output.stdout).trim().to_string()
            } else {
                stderr
            }
        ));
    }

    String::from_utf8_lossy(&output.stdout)
        .trim()
        .parse::<u64>()
        .map_err(|e| format!("could not parse git rev-list output as u64: {e}"))
}

/// Ensure an open PR exists for the task branch before running the review agent.
///
/// Checks the store first (avoids GitHub list-API cache race), then the API.
/// If no PR exists but the branch has commits, attempts to create one and
/// continues with the same review pass.
/// All error and no-op paths update task status and return [`EnsurePrResult::EarlyReturn`].
#[allow(clippy::too_many_arguments)]
async fn ensure_pr_exists(
    task: &ExternalTask,
    branch_name: &str,
    worktree_path: &std::path::Path,
    repo: &str,
    agent_summary: &str,
    store: &Arc<TaskStore>,
    task_manager: &Arc<TaskManager>,
    stored_pr_number: Option<u64>,
) -> anyhow::Result<EnsurePrResult> {
    let gh_check = GhHttp::new()?;

    if let Some(n) = stored_pr_number {
        tracing::info!(
            task_id = task.id.0,
            pr_number = n,
            branch = %branch_name,
            "open PR found in store, proceeding with review"
        );
        return Ok(EnsurePrResult::Ready(n));
    }

    match gh_check.get_pr_number(repo, branch_name).await {
        Ok(Some(n)) => {
            tracing::info!(
                task_id = task.id.0,
                pr_number = n,
                branch = %branch_name,
                "open PR found, proceeding with review"
            );
            Ok(EnsurePrResult::Ready(n))
        }
        Ok(None) => {
            // No open PR — check if branch has commits ahead of default branch.
            let default_branch = worktree::detect_default_branch(worktree_path).await;
            let has_commits = match count_ahead_commits(worktree_path, &default_branch).await {
                Ok(count) => count > 0,
                Err(e) => {
                    tracing::error!(
                        task_id = task.id.0,
                        branch = %branch_name,
                        default_branch = %default_branch,
                        err = %e,
                        "could not determine if branch has commits ahead of default"
                    );
                    return Ok(EnsurePrResult::EarlyReturn(ReviewDecision::Failed(
                        format!("git ancestry check failed: {e}"),
                    )));
                }
            };

            if has_commits {
                // Branch has unpushed or un-PR'd work — try to create a PR
                tracing::warn!(
                    task_id = task.id.0,
                    branch = %branch_name,
                    "no open PR but branch has commits — attempting to create PR"
                );
                let worktree_str = worktree_path.to_str().ok_or_else(|| {
                    anyhow::anyhow!(
                        "worktree path contains non-UTF-8 characters: {:?}",
                        worktree_path
                    )
                })?;
                // Push first in case agent forgot (use token auth like git_ops::push_branch)
                let auth_env = runner::git_ops::build_git_auth_env().await;

                // Run a credentialed push and log any errors so failures are
                // visible in logs instead of being silently discarded.
                // Wrap with a timeout to prevent indefinitely blocking a Tokio worker on a
                // hung network connection (corporate proxy, partial GitHub outage, TCP hang).
                let push_timeout = std::time::Duration::from_secs(120);
                match tokio::time::timeout(
                    push_timeout,
                    tokio::process::Command::new("git")
                        .args(["-C", worktree_str, "push", "-u", "origin", branch_name])
                        .envs(auth_env.iter().map(|(k, v)| (k.as_str(), v.as_str())))
                        .output(),
                )
                .await
                {
                    Err(_elapsed) => {
                        tracing::warn!(
                            task_id = task.id.0,
                            branch = %branch_name,
                            timeout_secs = push_timeout.as_secs(),
                            "review gate fallback push timed out"
                        );
                    }
                    Ok(Ok(o)) => {
                        if !o.status.success() {
                            tracing::warn!(
                                task_id = task.id.0,
                                branch = %branch_name,
                                exit_code = ?o.status.code(),
                                stdout = %String::from_utf8_lossy(&o.stdout),
                                stderr = %String::from_utf8_lossy(&o.stderr),
                                "review gate fallback push failed"
                            );
                        } else {
                            tracing::info!(task_id = task.id.0, branch = %branch_name, "review gate fallback push succeeded");
                        }
                    }
                    Ok(Err(e)) => {
                        tracing::error!(
                            task_id = task.id.0,
                            branch = %branch_name,
                            error = %e,
                            "review gate fallback push command failed to run"
                        );
                    }
                }

                let task_ref = runner::git_ops::format_task_ref(&task.id.0);
                let pr_body = format!(
                    "Resolves {task_ref}\n\nAuto-created by orch review gate (agent forgot to open PR)"
                );
                let gh = GhHttp::new()?;
                match gh
                    .create_pr(repo, &task.title, &pr_body, branch_name, &default_branch)
                    .await
                {
                    Ok(url) => match parse_pr_number_from_url(&url) {
                        Ok(pr_num) => {
                            store_set_result(
                                &Some(Arc::clone(store)),
                                repo,
                                &task.id.0,
                                &[("pr_number", serde_json::json!(pr_num as i64))],
                            )
                            .await?;
                            tracing::info!(
                                task_id = task.id.0,
                                branch = %branch_name,
                                pr_url = %url,
                                "created missing PR via GhHttp — continuing review"
                            );
                            Ok(EnsurePrResult::Ready(pr_num))
                        }
                        Err(e) => {
                            tracing::warn!(
                                task_id = task.id.0,
                                branch = %branch_name,
                                pr_url = %url,
                                error = %e,
                                "created missing PR via GhHttp, but failed to parse PR number"
                            );
                            Ok(EnsurePrResult::EarlyReturn(ReviewDecision::Failed(
                                format!("created missing PR, but could not parse PR number: {e}"),
                            )))
                        }
                    },
                    Err(e) => {
                        let e_str = format!("{e}");
                        if e_str.contains("already exists") {
                            // GitHub list-API cache may be stale immediately after a 422.
                            // Use two attempts with exponential backoff (2s → 4s) to avoid
                            // racing into a duplicate `gh pr create` call.
                            let backoff_delays = [2u64, 4u64];
                            let mut found_pr: Option<u64> = None;
                            for delay_secs in backoff_delays {
                                tokio::time::sleep(tokio::time::Duration::from_secs(delay_secs))
                                    .await;
                                if let Ok(Some(n)) = gh_check.get_pr_number(repo, branch_name).await
                                {
                                    found_pr = Some(n);
                                    break;
                                }
                            }
                            if let Some(n) = found_pr {
                                tracing::info!(
                                    task_id = task.id.0,
                                    pr_number = n,
                                    branch = %branch_name,
                                    "found existing PR after create_pr 422 — retrying review"
                                );
                                store_set_result(
                                    &Some(Arc::clone(store)),
                                    repo,
                                    &task.id.0,
                                    &[("pr_number", serde_json::json!(n as i64))],
                                )
                                .await?;
                                return Ok(EnsurePrResult::Ready(n));
                            }
                        }
                        tracing::warn!(
                            task_id = task.id.0,
                            branch = %branch_name,
                            error = %e,
                            "create_pr failed via GhHttp, falling back to CLI"
                        );
                        // Fall back to CLI. Wrap the `gh pr create` call in a timeout so
                        // a hung CLI/network does not block the review pipeline.
                        let pr_timeout = std::time::Duration::from_secs(120);
                        let mut gh_cmd = tokio::process::Command::new("gh");
                        gh_cmd.args([
                            "pr",
                            "create",
                            "--repo",
                            repo,
                            "--head",
                            branch_name,
                            "--title",
                            &task.title,
                            "--body",
                            &pr_body,
                        ]);
                        gh_cmd.current_dir(worktree_path);
                        gh_cmd.kill_on_drop(true);

                        let pr_result = tokio::time::timeout(pr_timeout, gh_cmd.output()).await;
                        match pr_result {
                            Ok(Ok(o)) if o.status.success() => {
                                let stdout = String::from_utf8_lossy(&o.stdout);
                                match parse_pr_number_from_url(stdout.trim()) {
                                    Ok(pr_num) => {
                                        store_set_result(
                                            &Some(Arc::clone(store)),
                                            repo,
                                            &task.id.0,
                                            &[("pr_number", serde_json::json!(pr_num as i64))],
                                        )
                                        .await?;
                                        tracing::info!(
                                            task_id = task.id.0,
                                            branch = %branch_name,
                                            "created missing PR via CLI — continuing review"
                                        );
                                        Ok(EnsurePrResult::Ready(pr_num))
                                    }
                                    Err(e) => {
                                        tracing::warn!(
                                            task_id = task.id.0,
                                            branch = %branch_name,
                                            stdout = %stdout,
                                            error = %e,
                                            "created missing PR via CLI, but failed to parse PR number"
                                        );
                                        Ok(EnsurePrResult::EarlyReturn(ReviewDecision::Failed(
                                            format!("created missing PR via CLI, but could not parse PR number: {e}")
                                        )))
                                    }
                                }
                            }
                            Ok(Ok(o)) => {
                                let stderr = String::from_utf8_lossy(&o.stderr);
                                if stderr.contains("already exists") {
                                    if let Some(pr_url) =
                                        stderr.lines().find(|l| l.trim().starts_with("https://"))
                                    {
                                        match parse_pr_number_from_url(pr_url.trim()) {
                                            Ok(pr_num) => {
                                                if let Err(e) = store_set_result(
                                                    &Some(Arc::clone(store)),
                                                    repo,
                                                    &task.id.0,
                                                    &[(
                                                        "pr_number",
                                                        serde_json::json!(pr_num as i64),
                                                    )],
                                                )
                                                .await
                                                {
                                                    tracing::warn!(
                                                        task_id = task.id.0,
                                                        pr_num,
                                                        err = %e,
                                                        "failed to persist pr_number from already-exists recovery — downstream may retry PR creation"
                                                    );
                                                }
                                                tracing::info!(
                                                    task_id = task.id.0,
                                                    branch = %branch_name,
                                                    pr_url = %pr_url,
                                                    "PR already exists (from CLI stderr) — retrying review"
                                                );
                                                return Ok(EnsurePrResult::Ready(pr_num));
                                            }
                                            Err(e) => {
                                                tracing::warn!(
                                                    task_id = task.id.0,
                                                    branch = %branch_name,
                                                    pr_url = %pr_url,
                                                    error = %e,
                                                    "found existing PR URL in CLI stderr, but failed to parse PR number"
                                                );
                                                // Continue to the error handling below
                                            }
                                        }
                                    }
                                }
                                tracing::error!(
                                    task_id = task.id.0,
                                    branch = %branch_name,
                                    stderr = %stderr,
                                    "failed to create missing PR — work may be stuck"
                                );

                                // If this was a transient GitHub 5xx or transport error,
                                // do NOT increment the persistent pr_create_failures counter
                                // (which would eventually block the task). Let the
                                // engine retry later. Detect via git_ops::is_transient_github_api_error
                                // so generic agent/model transport errors (e.g. broken pipe)
                                // are NOT mis-classified as GitHub API transients.
                                if crate::engine::runner::git_ops::is_transient_github_api_error(
                                    &stderr,
                                ) {
                                    tracing::warn!(
                                        task_id = task.id.0,
                                        branch = %branch_name,
                                        "transient GitHub error creating PR; will retry later without incrementing persistent failure counter"
                                    );
                                    return Ok(EnsurePrResult::EarlyReturn(
                                        ReviewDecision::Failed(format!(
                                            "transient github error creating PR: {stderr}"
                                        )),
                                    ));
                                }

                                match store_increment(
                                    &Some(Arc::clone(store)),
                                    repo,
                                    &task.id.0,
                                    "pr_create_failures",
                                )
                                .await
                                {
                                    Ok(failures) if failures >= MAX_PR_CREATE_FAILURES => {
                                        return Ok(EnsurePrResult::EarlyReturn(
                                            ReviewDecision::Blocked(format!(
                                                "no PR, create failed {failures} times: {stderr}"
                                            )),
                                        ));
                                    }
                                    Ok(_) => {}
                                    Err(e) => {
                                        tracing::warn!(task_id = task.id.0, err = %e, "failed to increment pr_create_failures — skipping blocking decision this tick");
                                    }
                                }
                                Ok(EnsurePrResult::EarlyReturn(ReviewDecision::Failed(
                                    format!("no PR, create failed: {stderr}"),
                                )))
                            }
                            Ok(Err(e)) => {
                                tracing::error!(
                                    task_id = task.id.0,
                                    error = %e,
                                    "failed to run gh pr create"
                                );
                                let e_str = format!("{e}");
                                if crate::engine::runner::git_ops::is_transient_github_api_error(
                                    &e_str,
                                ) {
                                    tracing::warn!(
                                        task_id = task.id.0,
                                        branch = %branch_name,
                                        "transient GitHub error from gh CLI fallback; will retry later without incrementing persistent failure counter"
                                    );
                                    return Ok(EnsurePrResult::EarlyReturn(
                                        ReviewDecision::Failed(format!("transient gh error: {e}")),
                                    ));
                                }

                                match store_increment(
                                    &Some(Arc::clone(store)),
                                    repo,
                                    &task.id.0,
                                    "pr_create_failures",
                                )
                                .await
                                {
                                    Ok(failures) if failures >= MAX_PR_CREATE_FAILURES => {
                                        return Ok(EnsurePrResult::EarlyReturn(
                                            ReviewDecision::Blocked(format!(
                                                "no PR, gh error {failures} times: {e}"
                                            )),
                                        ));
                                    }
                                    Ok(_) => {}
                                    Err(e) => {
                                        tracing::warn!(task_id = task.id.0, err = %e, "failed to increment pr_create_failures — skipping blocking decision this tick");
                                    }
                                }
                                Ok(EnsurePrResult::EarlyReturn(ReviewDecision::Failed(
                                    format!("no PR, gh error: {e}"),
                                )))
                            }
                            Err(_elapsed) => {
                                // Timeout from tokio::time::timeout
                                tracing::error!(
                                    task_id = task.id.0,
                                    timeout_secs = pr_timeout.as_secs(),
                                    "gh pr create CLI fallback timed out"
                                );
                                let e_str = format!(
                                    "gh pr create timed out after {}s",
                                    pr_timeout.as_secs()
                                );
                                if crate::engine::runner::git_ops::is_transient_github_api_error(
                                    &e_str,
                                ) {
                                    tracing::warn!(
                                        task_id = task.id.0,
                                        branch = %branch_name,
                                        "transient GitHub timeout from gh CLI fallback; will retry later without incrementing persistent failure counter"
                                    );
                                    return Ok(EnsurePrResult::EarlyReturn(
                                        ReviewDecision::Failed(format!(
                                            "transient gh error: {e_str}"
                                        )),
                                    ));
                                }

                                match store_increment(
                                    &Some(Arc::clone(store)),
                                    repo,
                                    &task.id.0,
                                    "pr_create_failures",
                                )
                                .await
                                {
                                    Ok(failures) if failures >= MAX_PR_CREATE_FAILURES => {
                                        return Ok(EnsurePrResult::EarlyReturn(
                                            ReviewDecision::Blocked(format!(
                                                "no PR, gh timeout {failures} times: {e_str}"
                                            )),
                                        ));
                                    }
                                    Ok(_) => {}
                                    Err(e) => {
                                        tracing::warn!(task_id = task.id.0, err = %e, "failed to increment pr_create_failures — skipping blocking decision this tick");
                                    }
                                }
                                Ok(EnsurePrResult::EarlyReturn(ReviewDecision::Failed(
                                    format!("no PR, gh timeout: {e_str}"),
                                )))
                            }
                        }
                    }
                }
            } else {
                // No PR and no commits — agent either failed or completed a read-only task.
                let merged = match gh_check.is_pr_merged(repo, branch_name).await {
                    Ok(v) => v,
                    Err(e) => {
                        tracing::warn!(task_id = task.id.0, branch = %branch_name, err = %e, "merge check failed, skipping task this tick");
                        return Ok(EnsurePrResult::EarlyReturn(ReviewDecision::Failed(
                            format!("merge check failed: {e}"),
                        )));
                    }
                };
                if merged {
                    tracing::info!(
                        task_id = task.id.0,
                        branch = %branch_name,
                        "PR already merged, marking done"
                    );
                    if let Err(e) = task_manager
                        .update_task_status(&task.id, Status::Done)
                        .await
                    {
                        tracing::error!(task_id = task.id.0, err = %e, "update_task_status(Done) failed — task may be stuck in InReview");
                    }
                    return Ok(EnsurePrResult::EarlyReturn(ReviewDecision::Skipped));
                }

                let last_error = opt_store_get_task(&Some(Arc::clone(store)), repo, &task.id.0)
                    .await
                    .map(|t| t.last_error)
                    .unwrap_or_default();
                let reason = if !agent_summary.is_empty() {
                    agent_summary.to_string()
                } else {
                    last_error.clone()
                };

                if last_error.contains("exceeded max attempts") {
                    tracing::warn!(
                        task_id = task.id.0,
                        branch = %branch_name,
                        "no PR and no commits after max attempts — marking blocked to stop loop"
                    );
                    // Write block_reason atomically with the status transition to prevent
                    // auto_unblock_blocked_tasks (block_reason.is_none() gate) from
                    // re-dispatching this task.
                    let fields = [
                        (
                            "block_reason",
                            serde_json::json!("max attempts exceeded — no PR or commits produced"),
                        ),
                        ("last_error", serde_json::json!(last_error)),
                    ];
                    if let Err(e) = task_manager
                        .update_task_status_and_result(&task.id, Status::Blocked, &fields)
                        .await
                    {
                        tracing::error!(task_id = task.id.0, err = %e, "update_task_status_and_result(Blocked) failed — skipping block to avoid silent auto-unblock loop");
                        return Ok(EnsurePrResult::EarlyReturn(ReviewDecision::Failed(
                            format!("failed to write block_reason: {e}"),
                        )));
                    }
                    return Ok(EnsurePrResult::EarlyReturn(ReviewDecision::Skipped));
                }

                if last_error.contains("422") && last_error.contains("head") {
                    tracing::info!(
                        task_id = task.id.0,
                        branch = %branch_name,
                        "no PR — 422/head-invalid means work already merged, marking done"
                    );
                    if let Err(e) = task_manager
                        .update_task_status(&task.id, Status::Done)
                        .await
                    {
                        tracing::error!(task_id = task.id.0, err = %e, "update_task_status(Done) failed");
                    }
                    return Ok(EnsurePrResult::EarlyReturn(ReviewDecision::Skipped));
                }

                if last_error.contains("no fallback agents")
                    || last_error.contains("all agents exhausted")
                {
                    tracing::warn!(
                        task_id = task.id.0,
                        branch = %branch_name,
                        last_error = %last_error,
                        "no PR and no commits after failover exhaustion — marking blocked to stop loop"
                    );
                    // Write block_reason atomically with the status transition to prevent
                    // auto_unblock_blocked_tasks (block_reason.is_none() gate) from
                    // re-dispatching this task.
                    let fields = [
                        (
                            "block_reason",
                            serde_json::json!("all agents exhausted — no PR or commits produced"),
                        ),
                        ("last_error", serde_json::json!(last_error)),
                    ];
                    if let Err(e) = task_manager
                        .update_task_status_and_result(&task.id, Status::Blocked, &fields)
                        .await
                    {
                        tracing::error!(task_id = task.id.0, err = %e, "update_task_status_and_result(Blocked) failed — skipping block to avoid silent auto-unblock loop");
                        return Ok(EnsurePrResult::EarlyReturn(ReviewDecision::Failed(
                            format!("failed to write block_reason: {e}"),
                        )));
                    }
                    return Ok(EnsurePrResult::EarlyReturn(ReviewDecision::Skipped));
                }

                // No PR and no commits — agent either failed or completed a read-only task.
                // Use a dedicated circuit-breaker counter persisted in the store so
                // repeated reroutes across separate runs are counted. This prevents
                // internal tasks from looping new→in_progress→needs_review→new indefinitely.
                // Prefer a dedicated config key `workflow.max_reroute_attempts` (fallback to
                // `workflow.max_attempts` for backwards compatibility).
                let max_reroutes: u32 = config::get("workflow.max_reroute_attempts")
                    .ok()
                    .and_then(|s| s.parse().ok())
                    .or_else(|| {
                        config::get("workflow.max_attempts")
                            .ok()
                            .and_then(|s| s.parse().ok())
                    })
                    .unwrap_or(3);

                tracing::warn!(
                    task_id = task.id.0,
                    branch = %branch_name,
                    reason = %reason,
                    "no PR and no commits — re-routing for retry"
                );

                // If the last_error indicates a transient GitHub 5xx/transport error,
                // do NOT increment the persistent reroute counter which would
                // eventually block the task. Instead, retry later.
                if should_skip_no_code_reroute_increment(&last_error) {
                    tracing::warn!(
                        task_id = task.id.0,
                        last_error = %last_error,
                        "transient GitHub error recorded; skipping persistent no_pr_reroutes increment"
                    );
                    return Ok(EnsurePrResult::EarlyReturn(ReviewDecision::Failed(
                        format!("transient github error: {last_error}"),
                    )));
                }

                // Atomically increment the persistent reroute counter and decide.
                let reroutes = match store_increment(
                    &Some(Arc::clone(store)),
                    repo,
                    &task.id.0,
                    "no_code_reroutes",
                )
                .await
                {
                    Ok(v) => v,
                    Err(e) => {
                        tracing::warn!(task_id = task.id.0, err = %e, "failed to increment no_code_reroutes — skipping reroute/block decision this tick");
                        // Skip action this tick
                        return Ok(EnsurePrResult::EarlyReturn(ReviewDecision::Failed(
                            format!("transient store error: {e}"),
                        )));
                    }
                };

                if reroutes as u32 >= max_reroutes {
                    tracing::error!(
                        task_id = task.id.0,
                        reroutes,
                        max_reroutes,
                        "reached max reroute attempts for no-pr-result — blocking for human review"
                    );
                    let msg = format!(
                        "no PR or code changes after {}/{} reroute attempts",
                        reroutes, max_reroutes
                    );
                    // Write block_reason atomically with the status transition to prevent
                    // auto_unblock_blocked_tasks (block_reason.is_none() gate) from
                    // re-dispatching this task.
                    let fields = [
                        (
                            "block_reason",
                            serde_json::json!(format!(
                                "max reroute attempts ({}) reached — no PR or code changes produced",
                                max_reroutes
                            )),
                        ),
                        ("last_error", serde_json::json!(msg)),
                        ("agent", serde_json::json!(null)),
                        ("model", serde_json::json!(null)),
                    ];
                    if let Err(e) = task_manager
                        .update_task_status_and_result(&task.id, Status::Blocked, &fields)
                        .await
                    {
                        tracing::error!(task_id = task.id.0, err = %e, "update_task_status_and_result(Blocked) failed — skipping block to avoid silent auto-unblock loop");
                    }
                    Ok(EnsurePrResult::EarlyReturn(ReviewDecision::Blocked(msg)))
                } else if task.id.0.starts_with("internal:") {
                    // Internal tasks with no PR or code changes are read-only/analysis
                    // tasks (e.g. morning briefing, evening retrospective) that
                    // legitimately produce no git-visible changes. Mark them done
                    // instead of re-routing, which would create a loop: needs_review
                    // → review agent → no PR/no commits → reroute to New → agent
                    // runs → needs_review → ...
                    tracing::info!(
                        task_id = task.id.0,
                        "no PR and no commits for internal task — marking done"
                    );
                    if let Err(e) = task_manager
                        .update_task_status(&task.id, Status::Done)
                        .await
                    {
                        tracing::error!(task_id = task.id.0, err = %e, "update_task_status(Done) failed — task may be stuck in InReview");
                    }
                    Ok(EnsurePrResult::EarlyReturn(ReviewDecision::Skipped))
                } else {
                    // Clear agent/model so router picks a different one and note the
                    // fact that this attempt produced no PR or code changes.
                    // Use store_set_result so we only transition status when the
                    // store write succeeds — partial failure would orphan the task.
                    if let Err(e) = store_set_result(
                        &Some(Arc::clone(store)),
                        repo,
                        &task.id.0,
                        &[
                            ("agent", serde_json::json!(null)),
                            ("model", serde_json::json!(null)),
                            (
                                "last_error",
                                serde_json::json!("no PR or code changes produced"),
                            ),
                        ],
                    )
                    .await
                    {
                        tracing::error!(task_id = task.id.0, err = %e, "failed to clear agent/model for reroute — skipping status transition");
                        // Don't update status — task stays in InReview and will be retried next tick
                    } else if let Err(e) =
                        task_manager.update_task_status(&task.id, Status::New).await
                    {
                        tracing::error!(task_id = task.id.0, err = %e, "update_task_status(New) failed — task may be stuck in InReview");
                    }
                    Ok(EnsurePrResult::EarlyReturn(ReviewDecision::Rerouted))
                }
            }
        }
        Err(e) => {
            tracing::warn!(
                task_id = task.id.0,
                branch = %branch_name,
                error = %e,
                "failed to check PR status"
            );
            Ok(EnsurePrResult::EarlyReturn(ReviewDecision::Failed(
                format!("PR check failed: {e}"),
            )))
        }
    }
}

/// Run the review agent on a completed task and handle the outcome.
///
/// Called after a task completes with status:done and a PR is created.
/// The review agent checks the changes and either approves (triggers auto-merge)
/// or requests changes (re-dispatches the original agent).
pub(crate) async fn review_and_merge(
    task: &ExternalTask,
    backend: &Arc<dyn ExternalBackend>,
    tmux: &Arc<TmuxManager>,
    repo: &str,
    router: &Arc<RwLock<Router>>,
    task_manager: &Arc<TaskManager>,
    store: &Arc<TaskStore>,
) -> anyhow::Result<ReviewDecision> {
    set_review_session_expected(store, repo, &task.id.0, true).await;

    let ctx = match build_review_context(task, repo, router, task_manager, store).await? {
        ReviewPhase::Ready(ctx) => ctx,
        ReviewPhase::EarlyReturn(decision) => return Ok(decision),
    };

    let run = match invoke_review_agent(task, tmux, &ctx, repo, store).await {
        ReviewPhase::Ready(run) => run,
        ReviewPhase::EarlyReturn(decision) => return Ok(decision),
    };

    let parsed = match parse_review_output(task, repo, &ctx, run, store).await {
        ReviewPhase::Ready(parsed) => parsed,
        ReviewPhase::EarlyReturn(decision) => return Ok(decision),
    };

    finalize_review_run(task, &ctx, &parsed, store).await;

    let router_config = router.read().await.config.clone();
    apply_review_decision(
        task,
        backend,
        repo,
        task_manager,
        store,
        &ctx,
        &parsed,
        &router_config,
    )
    .await
}

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

    use crate::backends::ExternalTask;
    use crate::engine::router::RouterConfig;
    use crate::github::types::{GitHubReview, GitHubReviewComment, GitHubUser, PullRequestReview};
    use crate::store::{TaskStatus, TaskStore};
    use tempfile::TempDir;

    // ── outcome_for_agent_error ─────────────────────────────────────────────

    #[test]
    fn outcome_for_agent_error_kimi_billing_cycle_is_not_rate_limit() {
        // Regression test for issue #3529: a review run hitting Kimi's billing-cycle
        // exhaustion (task 156734 in the bug report) must not be stored as a plain
        // "rate_limit" outcome.
        let err = runner::agents::AgentError::RateLimit {
            message: "Failed to authenticate. API Error: 403 You've reached your usage limit \
                      for this billing cycle. Your quota will be refreshed soon."
                .to_string(),
        };
        assert_eq!(outcome_for_agent_error(&err), "billing_cycle_exhausted");
    }

    #[test]
    fn outcome_for_agent_error_generic_rate_limit_unaffected() {
        let err = runner::agents::AgentError::RateLimit {
            message: "429 Too Many Requests".to_string(),
        };
        assert_eq!(outcome_for_agent_error(&err), "rate_limit");
    }

    #[test]
    fn outcome_for_agent_error_generic_auth_unaffected() {
        let err = runner::agents::AgentError::Auth {
            message: "401 Unauthorized".to_string(),
        };
        assert_eq!(outcome_for_agent_error(&err), "auth_error");
    }

    // ── parse_pr_number_from_url ────────────────────────────────────────────

    #[test]
    fn parse_pr_number_standard_url() {
        assert_eq!(
            parse_pr_number_from_url("https://github.com/owner/repo/pull/42"),
            Ok(42)
        );
    }

    #[test]
    fn parse_pr_number_trailing_slash() {
        assert_eq!(
            parse_pr_number_from_url("https://github.com/owner/repo/pull/42/"),
            Ok(42)
        );
    }

    #[test]
    fn parse_pr_number_with_query_string() {
        assert_eq!(
            parse_pr_number_from_url("https://github.com/owner/repo/pull/42?tab=files"),
            Ok(42)
        );
    }

    #[test]
    fn parse_pr_number_with_hash_fragment() {
        assert_eq!(
            parse_pr_number_from_url("https://github.com/owner/repo/pull/42#issuecomment-123"),
            Ok(42)
        );
    }

    #[test]
    fn parse_pr_number_with_query_and_hash() {
        assert_eq!(
            parse_pr_number_from_url("https://github.com/owner/repo/pull/42?quick_pull=1#top"),
            Ok(42)
        );
    }

    #[test]
    fn parse_pr_number_with_leading_whitespace() {
        assert_eq!(
            parse_pr_number_from_url("  https://github.com/owner/repo/pull/42\n"),
            Ok(42)
        );
    }

    #[test]
    fn parse_pr_number_wrong_domain() {
        let url = "https://gitlab.com/owner/repo/merge_requests/42";
        let result = parse_pr_number_from_url(url);
        assert!(result.is_err());
        assert!(result
            .unwrap_err()
            .contains("does not start with 'https://github.com/'"));
    }

    #[test]
    fn parse_pr_number_missing_pull_segment() {
        // Issue URL — not a PR URL
        let url = "https://github.com/owner/repo/issues/42";
        let result = parse_pr_number_from_url(url);
        assert!(result.is_err());
        assert!(result.unwrap_err().contains("missing '/pull/' segment"));
    }

    #[test]
    fn parse_pr_number_zero_is_rejected() {
        let url = "https://github.com/owner/repo/pull/0";
        let result = parse_pr_number_from_url(url);
        assert!(result.is_err());
        assert!(result.unwrap_err().contains("cannot be zero"));
    }

    #[test]
    fn parse_pr_number_non_numeric_segment() {
        let url = "https://github.com/owner/repo/pull/abc";
        let result = parse_pr_number_from_url(url);
        assert!(result.is_err());
        assert!(result.unwrap_err().contains("not a valid positive integer"));
    }

    #[test]
    fn parse_pr_number_empty_string() {
        let result = parse_pr_number_from_url("");
        assert!(result.is_err());
        assert!(result.unwrap_err().contains("empty"));
    }

    #[test]
    fn parse_pr_number_http_not_https() {
        // HTTP URLs are not accepted — only HTTPS
        let url = "http://github.com/owner/repo/pull/42";
        let result = parse_pr_number_from_url(url);
        assert!(result.is_err());
        assert!(result
            .unwrap_err()
            .contains("does not start with 'https://github.com/'"));
    }

    // ── format_review_error ─────────────────────────────────────────────────
    //
    // Regression coverage for issue #3149: review runs were persisting raw
    // `api_retry` NDJSON fragments into `task_runs.error` because the review
    // pipeline formatted rate-limit errors with `err.to_string()` instead of
    // routing the raw provider message through `summarize_rate_limit_error`.

    #[test]
    fn format_review_error_summarises_rate_limit_fragment_with_api_retry() {
        // Real-world fragment captured from a minimax review run (issue #3149).
        // The provider message contains a complete api_retry JSON line so the
        // summariser should produce a clean `status=… after … attempts` summary.
        let raw_fragment = "{\"attempt\":6,\"max_retries\":10,\"retry_delay_ms\":17266.789,\"error_status\":429,\"error\":\"rate_limit\",\"session_id\":\"5c7148ee\"}\n\
                            {\"type\":\"system\",\"subtype\":\"api_retry\",\"attempt\":7,\"max_retries\":10,\"retry_delay_ms\":35162.66,\"error_status\":429,\"error\":\"rate_limit\",\"session_id\":\"5c7148ee\"}";
        let err = runner::agents::AgentError::RateLimit {
            message: raw_fragment.to_string(),
        };

        let stored = format_review_error("minimax", &err);

        // Must not contain raw JSON braces or the api_retry marker — these are
        // exactly the leaks that filled `task_runs.error` with noisy payloads.
        assert!(
            !stored.contains("{\"type\""),
            "stored error must not contain raw NDJSON envelope, got: {stored}"
        );
        assert!(
            !stored.contains("subtype"),
            "stored error must not contain raw api_retry fragment, got: {stored}"
        );
        assert!(
            !stored.contains("session_id"),
            "stored error must not contain session_id from provider payload, got: {stored}"
        );
        assert_eq!(
            stored,
            "minimax rate limit: status=429 after 7 attempts (last delay 35s)"
        );
    }

    #[test]
    fn format_review_error_summarises_truncated_fragment_without_subtype_marker() {
        // Mid-object fragment that begins after the `subtype` key was cut off — this
        // is the shape that landed in `task_runs.error` for rows 13634/13542/9821
        // before the fix. The summariser's fallback path should emit a compact
        // hint rather than echoing the raw payload.
        let raw_fragment = ",\"attempt\":5,\"max_retries\":10,\"retry_delay_ms\":9034.81,\"error_status\":429,\"error\":\"rate_limit\",\"session_id\":\"992e17d3\"";
        let err = runner::agents::AgentError::RateLimit {
            message: raw_fragment.to_string(),
        };

        let stored = format_review_error("glm", &err);

        assert!(stored.starts_with("glm rate limit:"));
        assert!(
            !stored.contains("session_id"),
            "stored error must not leak session_id from truncated fragment, got: {stored}"
        );
        // The fragment first-line truncation path is bounded — verify the stored
        // string stays well under the 512-char cap suggested by the bug report.
        assert!(
            stored.len() <= 200,
            "stored error should be compact (<=200 chars), got {} chars: {stored}",
            stored.len()
        );
    }

    #[test]
    fn format_review_error_preserves_non_rate_limit_variants() {
        let err = runner::agents::AgentError::Auth {
            message: "invalid bearer token".to_string(),
        };
        assert_eq!(
            format_review_error("kimi", &err),
            "kimi auth error: invalid bearer token"
        );

        let err = runner::agents::AgentError::Timeout {
            elapsed: std::time::Duration::from_secs(1801),
        };
        assert_eq!(
            format_review_error("kimi", &err),
            "kimi timed out after 1801s"
        );
    }

    fn git(dir: &std::path::Path, args: &[&str]) {
        let mut cmd = std::process::Command::new("git");
        cmd.current_dir(dir);
        // Global commit.gpgsign calls the operator's signer (1Password)
        if args.first() == Some(&"commit") {
            cmd.args(["-c", "commit.gpgsign=false"]);
        }
        let status = cmd.args(args).status().unwrap();
        assert!(status.success(), "git {:?} failed", args);
    }

    #[test]
    fn test_pull_request_review_requests_changes() {
        let review = PullRequestReview {
            review: GitHubReview {
                id: 1,
                user: GitHubUser {
                    login: "reviewer".to_string(),
                },
                body: Some("Please fix".to_string()),
                state: "CHANGES_REQUESTED".to_string(),
                html_url: None,
                submitted_at: "2024-01-01T00:00:00Z".to_string(),
                commit_id: None,
            },
            comments: vec![],
        };
        assert!(review.requests_changes());
    }

    #[test]
    fn test_pull_request_review_does_not_request_changes() {
        let review = PullRequestReview {
            review: GitHubReview {
                id: 1,
                user: GitHubUser {
                    login: "reviewer".to_string(),
                },
                body: Some("LGTM".to_string()),
                state: "APPROVED".to_string(),
                html_url: None,
                submitted_at: "2024-01-01T00:00:00Z".to_string(),
                commit_id: None,
            },
            comments: vec![],
        };
        assert!(!review.requests_changes());
    }

    #[test]
    fn test_pull_request_review_actionable_comments_filters_empty_and_replies() {
        let review = PullRequestReview {
            review: GitHubReview {
                id: 1,
                user: GitHubUser {
                    login: "reviewer".to_string(),
                },
                body: None,
                state: "CHANGES_REQUESTED".to_string(),
                html_url: None,
                submitted_at: "2024-01-01T00:00:00Z".to_string(),
                commit_id: None,
            },
            comments: vec![
                GitHubReviewComment {
                    id: 1,
                    user: GitHubUser {
                        login: "reviewer".to_string(),
                    },
                    body: "Fix this issue".to_string(),
                    path: "src/main.rs".to_string(),
                    line: Some(10),
                    original_line: Some(10),
                    commit_id: "abc".to_string(),
                    original_commit_id: "abc".to_string(),
                    html_url: "url".to_string(),
                    created_at: "2024-01-01T00:00:00Z".to_string(),
                    updated_at: "2024-01-01T00:00:00Z".to_string(),
                    in_reply_to_id: None,
                    diff_hunk: Some(
                        "@@ -8,5 +8,5 @@ fn main() {\n-    let x = 1;\n+    let x = 2;".to_string(),
                    ),
                },
                GitHubReviewComment {
                    id: 2,
                    user: GitHubUser {
                        login: "reviewer".to_string(),
                    },
                    body: "".to_string(),
                    path: "src/lib.rs".to_string(),
                    line: Some(20),
                    original_line: Some(20),
                    commit_id: "abc".to_string(),
                    original_commit_id: "abc".to_string(),
                    html_url: "url".to_string(),
                    created_at: "2024-01-01T00:00:00Z".to_string(),
                    updated_at: "2024-01-01T00:00:00Z".to_string(),
                    in_reply_to_id: None,
                    diff_hunk: None,
                },
                GitHubReviewComment {
                    id: 3,
                    user: GitHubUser {
                        login: "reviewer".to_string(),
                    },
                    body: "Reply to this".to_string(),
                    path: "src/lib.rs".to_string(),
                    line: Some(30),
                    original_line: Some(30),
                    commit_id: "abc".to_string(),
                    original_commit_id: "abc".to_string(),
                    html_url: "url".to_string(),
                    created_at: "2024-01-01T00:00:00Z".to_string(),
                    updated_at: "2024-01-01T00:00:00Z".to_string(),
                    in_reply_to_id: Some(1),
                    diff_hunk: None,
                },
            ],
        };
        let actionable = review.actionable_comments();
        assert_eq!(actionable.len(), 1);
        assert_eq!(actionable[0].id, 1);
        assert_eq!(actionable[0].body, "Fix this issue");
        assert_eq!(actionable[0].path, "src/main.rs");
        assert_eq!(
            actionable[0].diff_hunk.as_ref().unwrap(),
            "@@ -8,5 +8,5 @@ fn main() {\n-    let x = 1;\n+    let x = 2;"
        );
    }

    #[test]
    fn test_router_config_model_for_complexity_returns_none_without_config() {
        // With no model_map configured, model_for_complexity must return None rather than
        // silently using a hardcoded model that may not exist for the given agent.
        let cfg = RouterConfig::default();
        assert!(cfg.model_for_complexity("claude", "simple", "").is_none());
        assert!(cfg.model_for_complexity("opencode", "review", "").is_none());
        assert!(cfg
            .model_for_complexity("unknown_agent_xyz", "simple", "")
            .is_none());
    }

    #[test]
    fn review_status_maps_decisions() {
        assert_eq!(
            review_status(&ReviewDecision::Approve).unwrap().0,
            "success"
        );
        let changes = ReviewDecision::RequestChanges {
            notes: String::new(),
            issues: vec![],
        };
        assert_eq!(review_status(&changes).unwrap().0, "failure");
        assert!(review_status(&ReviewDecision::Failed("x".into())).is_none());
    }

    #[test]
    fn review_started_comment_includes_agent_and_model() {
        let expected_footer = attribution_footer("Review started", "kimi", Some("opus"));
        let expected = format!("🔍 Automated review started{}", expected_footer);
        assert_eq!(review_started_comment("kimi", "opus"), expected);
    }

    // ── build_review_context: no worktree/branch/pr (issue #3623) ───────────

    fn no_code_review_task(id: &str) -> ExternalTask {
        ExternalTask {
            id: crate::backends::ExternalId(id.to_string()),
            title: "opencode discovery timeout".to_string(),
            body: "".to_string(),
            state: "open".to_string(),
            labels: vec![],
            author: "tester".to_string(),
            created_at: "2026-01-01T00:00:00Z".to_string(),
            updated_at: "2026-01-01T00:00:00Z".to_string(),
            url: "https://example.com/issues/1".to_string(),
        }
    }

    /// An approval only carries over while the PR head is still the commit that was
    /// approved. A force-push from another clone moves the head, and reviewing again is
    /// the safe answer in every other case (#3783)
    #[tokio::test]
    async fn carried_approval_counts_as_this_rounds_decision() {
        let store = Arc::new(TaskStore::open_memory().await.unwrap());
        let id = store
            .create_internal("owner/repo", "Approved head", "", "test", "carried", None)
            .await
            .unwrap();
        let task_id = format!("internal:{id}");
        store
            .update_status(id, crate::store::TaskStatus::InReview)
            .await
            .unwrap();
        assert_eq!(
            store.in_review_entry(id).await.unwrap().map(|e| e.1),
            Some(false)
        );

        record_carried_approval(&store, "owner/repo", &task_id, 7).await;

        assert_eq!(
            store.in_review_entry(id).await.unwrap().map(|e| e.1),
            Some(true),
            "the stale in_review check must see this round as decided"
        );
    }

    #[test]
    fn approval_covers_head_only_for_the_exact_commit() {
        assert!(approval_covers_head(Some("abc123"), Some("abc123")));
        assert!(
            !approval_covers_head(Some("abc123"), Some("def456")),
            "a moved PR head must be reviewed"
        );
        assert!(
            !approval_covers_head(None, Some("abc123")),
            "no recorded approval means review"
        );
        assert!(
            !approval_covers_head(Some("abc123"), None),
            "an unreadable PR head must not skip the review"
        );
        assert!(!approval_covers_head(Some(""), Some("")));
    }

    /**
     * The approval is recorded against the commit pushed to the PR and is dropped when a
     * review asks for changes, so the next NeedsReview event reviews the new work (#3783)
     **/
    #[tokio::test]
    async fn approved_sha_round_trips_through_the_kv_store() {
        let temp = TempDir::new().unwrap();
        let wt = temp.path();
        git(wt, &["init", "-b", "main"]);
        git(wt, &["config", "user.name", "Test"]);
        git(wt, &["config", "user.email", "test@test.com"]);
        std::fs::write(wt.join("fix.txt"), "the fix").unwrap();
        git(wt, &["add", "."]);
        git(wt, &["commit", "-m", "apply the fix"]);

        let store = Arc::new(TaskStore::open_memory().await.unwrap());
        let repo = "owner/repo";
        let task_id = "3783";

        record_approved_sha(&store, repo, task_id, 77, wt).await;
        let recorded = store
            .kv_get(&approved_sha_key(repo, task_id, 77))
            .await
            .unwrap()
            .expect("approval must be recorded");
        assert_eq!(
            recorded,
            worktree_head_sha(wt).await.unwrap(),
            "the recorded sha must be the commit that was pushed to the PR"
        );
        assert!(
            store
                .kv_get(&approved_sha_key(repo, task_id, 78))
                .await
                .unwrap()
                .is_none(),
            "an approval must not leak to another PR"
        );

        clear_approved_sha(&store, repo, task_id, 77).await;
        assert!(
            store
                .kv_get(&approved_sha_key(repo, task_id, 77))
                .await
                .unwrap()
                .is_none(),
            "requesting changes must drop the approval"
        );
    }

    #[tokio::test]
    async fn build_review_context_reroutes_external_task_with_no_code() {
        let store = Arc::new(TaskStore::open_memory().await.unwrap());
        let repo = "owner/repo";
        let ext = no_code_review_task("3617");
        store.ensure_external_task(repo, &ext).await.unwrap();

        let backend = Arc::new(crate::backends::test_helpers::NoopBackend);
        let task_manager = Arc::new(TaskManager::with_store(
            backend,
            Arc::clone(&store),
            repo.to_string(),
        ));
        let router = Arc::new(RwLock::new(Router::new(RouterConfig::default())));

        let phase = build_review_context(&ext, repo, &router, &task_manager, &store)
            .await
            .unwrap();

        assert!(
            matches!(phase, ReviewPhase::EarlyReturn(ReviewDecision::Rerouted)),
            "external task with no worktree/branch/pr must reroute, not skip-to-done"
        );

        let stored = store
            .get_by_external_id(repo, &ext.id.0)
            .await
            .unwrap()
            .expect("task must still exist");
        assert_eq!(stored.status, TaskStatus::New);
        assert!(stored.agent.is_none());
        assert!(stored.model.is_none());
    }

    #[tokio::test]
    async fn build_review_context_blocks_external_task_after_max_reroutes() {
        // Isolate from the real ~/.orch/config.yml (which may set a different
        // workflow.max_attempts) so the default of 3 reroutes is deterministic.
        let temp_home = TempDir::new().unwrap();
        let orch_dir = temp_home.path().join(".orch");
        std::fs::create_dir_all(&orch_dir).unwrap();
        let old_orch_home = std::env::var_os("ORCH_HOME");
        std::env::set_var("ORCH_HOME", &orch_dir);

        let store = Arc::new(TaskStore::open_memory().await.unwrap());
        let repo = "owner/repo";
        let ext = no_code_review_task("3618");
        store.ensure_external_task(repo, &ext).await.unwrap();

        let backend = Arc::new(crate::backends::test_helpers::NoopBackend);
        let task_manager = Arc::new(TaskManager::with_store(
            backend,
            Arc::clone(&store),
            repo.to_string(),
        ));
        let router = Arc::new(RwLock::new(Router::new(RouterConfig::default())));

        // Default max_reroute_attempts is 3 — exhaust it.
        for _ in 0..3 {
            let phase = build_review_context(&ext, repo, &router, &task_manager, &store)
                .await
                .unwrap();
            if let ReviewPhase::EarlyReturn(ReviewDecision::Blocked(_)) = phase {
                break;
            }
        }

        let stored = store
            .get_by_external_id(repo, &ext.id.0)
            .await
            .unwrap()
            .expect("task must still exist");
        assert_eq!(stored.status, TaskStatus::Blocked);

        if let Some(old) = old_orch_home {
            std::env::set_var("ORCH_HOME", old);
        } else {
            std::env::remove_var("ORCH_HOME");
        }
    }

    #[tokio::test]
    async fn build_review_context_marks_internal_task_done_with_no_code() {
        let store = Arc::new(TaskStore::open_memory().await.unwrap());
        let repo = "owner/repo";
        let internal_id = store
            .create_internal(repo, "Daily retrospective", "", "job", "job-1", None)
            .await
            .unwrap();
        let ext_id = format!("internal:{internal_id}");
        let ext = no_code_review_task(&ext_id);

        let backend = Arc::new(crate::backends::test_helpers::NoopBackend);
        let task_manager = Arc::new(TaskManager::with_store(
            backend,
            Arc::clone(&store),
            repo.to_string(),
        ));
        let router = Arc::new(RwLock::new(Router::new(RouterConfig::default())));

        let phase = build_review_context(&ext, repo, &router, &task_manager, &store)
            .await
            .unwrap();

        assert!(matches!(
            phase,
            ReviewPhase::EarlyReturn(ReviewDecision::Skipped)
        ));
        let stored = store.get(internal_id).await.unwrap();
        assert_eq!(stored.status, TaskStatus::Done);
    }

    #[test]
    fn no_code_reroute_skip_check_rejects_agent_stream_disconnects() {
        let last_error =
            "codex failed: Reconnecting... (stream disconnected before completion: Broken pipe)";
        assert!(!should_skip_no_code_reroute_increment(last_error));
    }

    // ── review fallback rate-limit/auth scan is bounded, not full-text ──────
    //
    // Regression test for a review transcript whose unparseable fallback text
    // is a long report/diff quoting a status table (`outcome` column values
    // like `rate_limit`, `failed`, `timeout`). Scanning the full text falsely
    // classified the review run as rate-limited even though the review agent
    // never actually hit a rate limit.

    #[test]
    fn review_fallback_scan_ignores_rate_limit_word_buried_in_long_report() {
        let padding = "x".repeat(10_000);
        let text_for_review = format!(
            "{padding}\n+| codex | gpt-5.4 | failed | 1 |\n+| codex | gpt-5.5 | failed / aborted / (empty) | 1 each |\n+| kimi | opus | rate_limit | 1 |\n{padding}"
        );

        let scan_head = runner::agents::patterns::safe_head(
            &text_for_review,
            runner::agents::patterns::RATE_LIMIT_SCAN_HEAD_BYTES,
        );
        let scan_tail = runner::agents::patterns::safe_tail(
            &text_for_review,
            runner::agents::patterns::RATE_LIMIT_SCAN_TAIL_BYTES,
        );

        assert!(runner::agents::patterns::detect_rate_limit(scan_head).is_none());
        assert!(runner::agents::patterns::detect_rate_limit(scan_tail).is_none());
    }

    #[test]
    fn review_fallback_scan_still_detects_real_rate_limit_at_tail() {
        let padding = "x".repeat(10_000);
        let text_for_review =
            format!("{padding}\nError: rate limit exceeded, please try again later");

        let scan_tail = runner::agents::patterns::safe_tail(
            &text_for_review,
            runner::agents::patterns::RATE_LIMIT_SCAN_TAIL_BYTES,
        );

        assert!(runner::agents::patterns::detect_rate_limit(scan_tail).is_some());
    }

    #[tokio::test]
    async fn recover_missing_review_worktree_recreates_worktree() {
        let temp_home = TempDir::new().unwrap();
        let orch_dir = temp_home.path().join(".orch");
        std::fs::create_dir_all(&orch_dir).unwrap();
        let old_orch_home = std::env::var_os("ORCH_HOME");
        std::env::set_var("ORCH_HOME", &orch_dir);

        let project_dir = temp_home.path().join("project");
        std::fs::create_dir_all(&project_dir).unwrap();
        git(&project_dir, &["init", "-b", "main"]);
        git(&project_dir, &["config", "user.name", "Test User"]);
        git(&project_dir, &["config", "user.email", "test@example.com"]);
        std::fs::write(project_dir.join("README.md"), "hello").unwrap();
        git(&project_dir, &["add", "README.md"]);
        git(&project_dir, &["commit", "-m", "init"]);

        std::fs::write(
            orch_dir.join("config.yml"),
            format!("projects:\n  - {}\n", project_dir.display()),
        )
        .unwrap();
        std::fs::write(project_dir.join(".orch.yml"), "gh:\n  repo: owner/repo\n").unwrap();

        let store = Arc::new(TaskStore::open_memory().await.unwrap());
        let repo = "owner/repo";
        store
            .create_internal(repo, "Fix review recovery", "", "review", "review-1", None)
            .await
            .unwrap();
        let ext = ExternalTask {
            id: crate::backends::ExternalId("1".to_string()),
            title: "Fix review recovery".to_string(),
            body: "".to_string(),
            state: "open".to_string(),
            labels: vec![],
            author: "tester".to_string(),
            created_at: "2026-01-01T00:00:00Z".to_string(),
            updated_at: "2026-01-01T00:00:00Z".to_string(),
            url: "https://example.com/issues/1".to_string(),
        };
        store.ensure_external_task(repo, &ext).await.unwrap();

        let recovered = recover_missing_review_worktree(&ext, repo, &store)
            .await
            .unwrap();

        assert!(recovered.work_dir.exists());
        assert!(recovered.branch.starts_with("gh-issue-1-"));

        if let Some(old) = old_orch_home {
            std::env::set_var("ORCH_HOME", old);
        } else {
            std::env::remove_var("ORCH_HOME");
        }
    }

    // ── count_ahead_commits ─────────────────────────────────────────────────

    #[tokio::test]
    async fn count_ahead_commits_returns_zero_when_no_ahead_commits() {
        let temp = TempDir::new().unwrap();
        let dir = temp.path();

        // Create a bare repo to act as the 'origin' remote
        let bare = temp.path().join("bare.git");
        std::fs::create_dir_all(&bare).unwrap();
        std::process::Command::new("git")
            .args(["init", "--bare"])
            .current_dir(&bare)
            .output()
            .unwrap();

        git(dir, &["init", "-b", "main"]);
        git(dir, &["config", "user.name", "Test"]);
        git(dir, &["config", "user.email", "test@test.com"]);
        std::fs::write(dir.join("f"), "a").unwrap();
        git(dir, &["add", "f"]);
        git(dir, &["commit", "-m", "init"]);
        std::process::Command::new("git")
            .args(["remote", "add", "origin", bare.to_str().unwrap()])
            .current_dir(dir)
            .output()
            .unwrap();
        std::process::Command::new("git")
            .args(["push", "origin", "main"])
            .current_dir(dir)
            .output()
            .unwrap();

        // origin/main now exists and HEAD is at the same commit — zero ahead
        let result = count_ahead_commits(dir, "main").await;
        assert!(
            result.is_ok(),
            "expected Ok when origin/main exists, got {result:?}"
        );
        assert_eq!(
            result.unwrap(),
            0,
            "expected 0 commits ahead when branch matches origin/main"
        );
    }

    #[tokio::test]
    async fn count_ahead_commits_returns_error_when_remote_ref_missing() {
        let temp = TempDir::new().unwrap();
        let dir = temp.path();
        git(dir, &["init", "-b", "main"]);
        git(dir, &["config", "user.name", "Test"]);
        git(dir, &["config", "user.email", "test@test.com"]);
        std::fs::write(dir.join("f"), "a").unwrap();
        git(dir, &["add", "f"]);
        git(dir, &["commit", "-m", "init"]);
        // No remote 'origin' configured — rev-list will fail
        let result = count_ahead_commits(dir, "main").await;
        assert!(
            result.is_err(),
            "expected Err when origin/main does not exist, got {result:?}"
        );
        let err = result.unwrap_err();
        assert!(
            err.contains("git rev-list exited"),
            "error should mention exit failure, got: {err}"
        );
    }

    #[tokio::test]
    async fn count_ahead_commits_returns_error_when_git_cannot_start() {
        // Use a path that is not a git repo — git will fail with non-zero exit
        let temp = TempDir::new().unwrap();
        let dir = temp.path();
        // Not a git repo at all
        let result = count_ahead_commits(dir, "main").await;
        assert!(
            result.is_err(),
            "expected Err when directory is not a git repo, got {result:?}"
        );
    }

    // ── review cooldown escalation (issue #3314) ────────────────────────────

    /// Review `InvalidResponse` (hard failure: empty/error output) must apply the
    /// persistent model backoff (~4h base) so the model isn't re-selected on the
    /// next tick after a brief standard cooldown expires.
    ///
    /// Regression test for issue #3314: `record_model_failure` (5-min base) was used
    /// instead of `record_persistent_model_failure` (4h base), allowing repeated
    /// re-selection of a structurally broken review model.
    #[serial_test::serial(cooldown_state)]
    #[tokio::test]
    async fn review_invalid_response_applies_persistent_model_cooldown() {
        crate::engine::cooldown::reset_global_state().await;
        let agent = "opencode-3314-review-invalid";
        let model = "opencode/north-mini-code-free";
        let key = format!("{agent}:{model}");

        assert!(
            !crate::engine::cooldown::is_model_in_cooldown(agent, model),
            "model should not be in cooldown before record_review_agent_failure"
        );

        let err = runner::agents::AgentError::InvalidResponse {
            raw: "empty review output".to_string(),
        };
        record_review_agent_failure("test-3314-a", agent, Some(model), &err).await;

        let now = chrono::Utc::now().timestamp();
        let until = crate::engine::cooldown::cooldown_until(&key)
            .expect("review InvalidResponse should set model cooldown");
        let remaining = until.saturating_sub(now);

        // Persistent backoff base is ~4h (14400s); standard model backoff base is 5min (300s).
        // Verify the persistent (4h) path was taken, not the short one.
        assert!(
            remaining >= crate::engine::cooldown::PERSISTENT_MODEL_BACKOFF_BASE_SECS - 30,
            "review InvalidResponse must apply ~4h persistent cooldown (issue #3314), \
             got {remaining}s — may have used 5-min standard backoff instead"
        );
    }

    /// Review `parse_error` (got output but couldn't parse it as a review response) must
    /// also apply the persistent model backoff so the model is cooled long enough to
    /// prevent re-selection after a brief cooldown or a success-reset from another task.
    ///
    /// Verifies the `record_review_agent_failure` path for `InvalidResponse`
    /// (which covers both the hard-failure and soft parse-error paths that were
    /// changed in the fix for issue #3314).
    #[serial_test::serial(cooldown_state)]
    #[tokio::test]
    async fn review_persistent_cooldown_survives_after_first_failure() {
        crate::engine::cooldown::reset_global_state().await;
        let agent = "opencode-3314-review-parse";
        let model = "opencode/north-mini-code-free";
        let key = format!("{agent}:{model}");

        // First failure: persistent backoff should apply immediately
        let err = runner::agents::AgentError::InvalidResponse {
            raw: "failed to parse review response".to_string(),
        };
        record_review_agent_failure("test-3314-b", agent, Some(model), &err).await;

        let now = chrono::Utc::now().timestamp();
        let until1 = crate::engine::cooldown::cooldown_until(&key)
            .expect("first failure should set cooldown");
        let remaining1 = until1.saturating_sub(now);

        // Simulate a success for the same agent+model (from a non-review task).
        // This resets failure_count:{agent}:{model} to 0.
        crate::engine::cooldown::record_agent_success(agent, model).await;

        // After success reset, cooldown timestamp is still set — the success only
        // resets the failure count, NOT the existing cooldown timer.
        // The next review failure will restart from count=1 (base backoff).
        // The key invariant: persistent base (4h) >> standard base (5min) so even
        // after a reset, the first-failure cooldown still lasts ~4h.
        assert!(
            remaining1 >= crate::engine::cooldown::PERSISTENT_MODEL_BACKOFF_BASE_SECS - 30,
            "after success reset, first-failure cooldown should still be ~4h, got {remaining1}s"
        );
    }

    /// Regression test for issue #3551: a billing-cycle-exhaustion message caught by
    /// the review module's *inline text scan* (`detect_rate_limit` on raw review
    /// output, used when the review process exits without a clean `AgentError`) must
    /// get the same treatment as the `AgentError`-based failure path fixed by #3530 —
    /// not a hardcoded `"rate_limit"` outcome plus a generic short agent cooldown.
    #[serial_test::serial(cooldown_state)]
    #[tokio::test]
    async fn review_inline_scan_billing_cycle_message_gets_model_scoped_cooldown() {
        crate::engine::cooldown::reset_global_state().await;
        let agent = "kimi-3551-review-inline";
        let model = "opus";
        let key = format!("{agent}:{model}");

        // Verbatim text from the observed kimi failure (issue #3551).
        let text_for_review = "kimi rate limit: Failed to authenticate. API Error: 403 \
            You've reached your usage limit for this billing cycle. Your quota will be \
            refreshed next cycle.";

        let scan_tail = runner::agents::patterns::safe_tail(
            text_for_review,
            runner::agents::patterns::RATE_LIMIT_SCAN_TAIL_BYTES,
        );
        let err = runner::agents::patterns::detect_rate_limit(scan_tail)
            .expect("billing-cycle message must still be caught by the rate-limit scan");

        assert_eq!(
            outcome_for_agent_error(&err),
            "billing_cycle_exhausted",
            "inline-scanned billing-cycle message must not be stored as plain rate_limit"
        );

        record_review_agent_failure("test-3551", agent, Some(model), &err).await;

        let now = chrono::Utc::now().timestamp();
        let until = crate::engine::cooldown::cooldown_until(&key)
            .expect("billing-cycle exhaustion must set a model-scoped cooldown");
        let remaining = until.saturating_sub(now);
        // Known model + BillingCycleExhausted routes to record_persistent_model_failure
        // (~4h base, 7d cap), not the agent-wide record_credit_exhaustion (~24h) path —
        // the quota is scoped to this model, so unrelated models on the same agent must
        // stay routable.
        assert!(
            remaining >= crate::engine::cooldown::PERSISTENT_MODEL_BACKOFF_BASE_SECS - 30,
            "billing-cycle exhaustion for a known model must apply the ~4h persistent \
             model cooldown, got {remaining}s — may have used the generic short agent \
             cooldown instead"
        );
    }

    // ── review success resets failure_count (issue #3478) ──────────────────

    fn test_review_task(id: &str) -> ExternalTask {
        ExternalTask {
            id: crate::backends::ExternalId(id.to_string()),
            title: "Test review task".to_string(),
            body: "".to_string(),
            state: "open".to_string(),
            labels: vec![],
            author: "tester".to_string(),
            created_at: "2026-01-01T00:00:00Z".to_string(),
            updated_at: "2026-01-01T00:00:00Z".to_string(),
            url: "https://example.com/issues/1".to_string(),
        }
    }

    fn test_review_context(agent: &str, model: &str, work_dir: &std::path::Path) -> ReviewContext {
        ReviewContext {
            store_id: None,
            had_prev_push_failure: false,
            review_attempt: 1,
            worktree_path: work_dir.to_path_buf(),
            branch_name: "gh-issue-1-test".to_string(),
            pr_number: 1,
            head_sha: None,
            default_branch: "main".to_string(),
            review_agent: agent.to_string(),
            review_model: Some(model.to_string()),
            review_task_id: "review-1".to_string(),
            review_attempt_dir: work_dir.to_path_buf(),
            output_file: work_dir.join("output.txt"),
            invocation: runner::agent::AgentInvocation {
                agent: agent.to_string(),
                model: Some(model.to_string()),
                work_dir: work_dir.to_path_buf(),
                system_prompt: String::new(),
                agent_message: String::new(),
                task_id: "review-1".to_string(),
                disallowed_tools: vec![],
                git_author_name: "Test".to_string(),
                git_author_email: "test@example.com".to_string(),
                output_file: work_dir.join("output.txt"),
                timeout_seconds: 0,
                repo: "owner/repo".to_string(),
                attempt: 1,
                sandbox: false,
                main_project_dir: work_dir.to_path_buf(),
                branch: String::new(),
            },
            same_agent_fallback: false,
        }
    }

    fn test_parsed_review(decision: ReviewDecision) -> ParsedReview {
        ParsedReview {
            run_id: None,
            exit_code: 0,
            stderr: String::new(),
            raw_output: String::new(),
            text_for_review: String::new(),
            token_usage: RunTokenUsage::default(),
            review_notes_for_comment: String::new(),
            decision,
        }
    }

    fn write_review_output(dir: &std::path::Path, stdout: &str, exit_code: i32) {
        std::fs::write(dir.join("output.txt"), stdout).unwrap();
        std::fs::write(dir.join("exit.txt"), exit_code.to_string()).unwrap();
        std::fs::write(dir.join("stderr.txt"), "").unwrap();
    }

    /// A kimi review that exits mid-thought (no result envelope, last content is a
    /// thinking block) must be stored as incomplete and must not apply the 4h
    /// persistent model cooldown (issue #3734).
    #[serial_test::serial(cooldown_state)]
    #[tokio::test]
    async fn parse_review_output_mid_thought_is_incomplete_without_persistent_cooldown() {
        crate::engine::cooldown::reset_global_state().await;
        let store = Arc::new(TaskStore::open_memory().await.unwrap());
        let work_dir = TempDir::new().unwrap();
        let agent = "kimi";
        let model = "opus-3734-mid-thought";
        let ctx = test_review_context(agent, model, work_dir.path());
        let redraw = "^D\u{0008}".repeat(20);
        let ndjson = format!(
            "{}\n{}\n{}\n{}\n{}",
            r#"{"type":"system","subtype":"thinking_tokens","tokens":1945}"#,
            r#"{"type":"assistant","message":{"content":[{"type":"text","text":"CI passes. Now verifying callers and the lease logic edge cases:"}],"stop_reason":null,"usage":{"input_tokens":10,"output_tokens":0}}}"#,
            r#"{"type":"assistant","message":{"content":[{"type":"thinking","thinking":"checking lease edge cases"}],"stop_reason":null,"usage":{"input_tokens":10,"output_tokens":0}}}"#,
            "\x1b[?25h",
            redraw,
        );
        write_review_output(work_dir.path(), &ndjson, 0);

        let task = test_review_task("3734");
        let phase = parse_review_output(
            &task,
            "owner/repo",
            &ctx,
            ReviewRun { run_id: None },
            &store,
        )
        .await;

        match phase {
            ReviewPhase::EarlyReturn(ReviewDecision::Failed(reason)) => {
                assert!(
                    reason.contains("incomplete:"),
                    "mid-thought review must be stored as incomplete, got {reason}"
                );
            }
            ReviewPhase::EarlyReturn(_) => panic!("expected Failed(incomplete)"),
            ReviewPhase::Ready(_) => panic!("expected Failed(incomplete), got Ready"),
        }
        assert!(
            !crate::engine::cooldown::is_model_in_cooldown(agent, model),
            "mid-thought incomplete must not apply persistent model cooldown"
        );
    }

    /// A tool_use as the last assistant content with no result envelope is the
    /// same incomplete shape as a thinking block (issue #3734).
    #[serial_test::serial(cooldown_state)]
    #[tokio::test]
    async fn parse_review_output_tool_use_without_envelope_is_incomplete() {
        crate::engine::cooldown::reset_global_state().await;
        let store = Arc::new(TaskStore::open_memory().await.unwrap());
        let work_dir = TempDir::new().unwrap();
        let agent = "kimi";
        let model = "opus-3734-tool-use";
        let ctx = test_review_context(agent, model, work_dir.path());
        let ndjson = r#"{"type":"assistant","message":{"content":[{"type":"tool_use","name":"Read","input":{"file_path":"src/lib.rs"}}]}}"#;
        write_review_output(work_dir.path(), ndjson, 0);

        let task = test_review_task("3734-tool");
        let phase = parse_review_output(
            &task,
            "owner/repo",
            &ctx,
            ReviewRun { run_id: None },
            &store,
        )
        .await;

        match phase {
            ReviewPhase::EarlyReturn(ReviewDecision::Failed(reason)) => {
                assert!(
                    reason.contains("incomplete:"),
                    "tool_use without envelope must be incomplete, got {reason}"
                );
            }
            ReviewPhase::EarlyReturn(_) => panic!("expected Failed(incomplete)"),
            ReviewPhase::Ready(_) => panic!("expected Failed(incomplete), got Ready"),
        }
        assert!(!crate::engine::cooldown::is_model_in_cooldown(agent, model));
    }

    /// A finished result envelope whose text is not a review decision still
    /// gets the persistent format-failure cooldown.
    #[serial_test::serial(cooldown_state)]
    #[tokio::test]
    async fn parse_review_output_unparseable_envelope_applies_persistent_cooldown() {
        crate::engine::cooldown::reset_global_state().await;
        let store = Arc::new(TaskStore::open_memory().await.unwrap());
        let work_dir = TempDir::new().unwrap();
        let agent = "kimi";
        let model = "opus-3734-format";
        let ctx = test_review_context(agent, model, work_dir.path());
        let ndjson = concat!(
            r#"{"type":"assistant","message":{"content":[{"type":"text","text":"I have finished looking at the code."}]}}"#,
            "\n",
            r#"{"type":"result","subtype":"success","is_error":false,"result":"I have finished looking at the code."}"#,
        );
        write_review_output(work_dir.path(), ndjson, 0);

        let task = test_review_task("3734-format");
        let phase = parse_review_output(
            &task,
            "owner/repo",
            &ctx,
            ReviewRun { run_id: None },
            &store,
        )
        .await;

        match phase {
            ReviewPhase::EarlyReturn(ReviewDecision::Failed(reason)) => {
                assert!(
                    reason.contains("parse error"),
                    "finished unparseable envelope must stay a parse error, got {reason}"
                );
            }
            ReviewPhase::EarlyReturn(_) => panic!("expected Failed(parse error)"),
            ReviewPhase::Ready(_) => panic!("expected Failed(parse error), got Ready"),
        }
        assert!(
            crate::engine::cooldown::is_model_in_cooldown(agent, model),
            "finished format failure must apply persistent model cooldown"
        );
    }

    /// Missing result envelope with a review decision in assistant text uses
    /// the assistant-text extractor, not MiniMax status=done synthesis.
    #[serial_test::serial(cooldown_state)]
    #[tokio::test]
    async fn parse_review_output_missing_envelope_uses_assistant_decision() {
        crate::engine::cooldown::reset_global_state().await;
        let store = Arc::new(TaskStore::open_memory().await.unwrap());
        let work_dir = TempDir::new().unwrap();
        let agent = "kimi";
        let model = "opus-3734-assistant";
        let ctx = test_review_context(agent, model, work_dir.path());
        let ndjson = r#"{"type":"assistant","message":{"content":[{"type":"text","text":"{\"decision\":\"approve\",\"notes\":\"LGTM\",\"issues\":[]}"}]}}"#;
        write_review_output(work_dir.path(), ndjson, 0);

        let task = test_review_task("3734-asst");
        let phase = parse_review_output(
            &task,
            "owner/repo",
            &ctx,
            ReviewRun { run_id: None },
            &store,
        )
        .await;

        match phase {
            ReviewPhase::Ready(parsed) => {
                assert!(matches!(parsed.decision, ReviewDecision::Approve));
            }
            ReviewPhase::EarlyReturn(_) => {
                panic!("expected Ready(Approve) from assistant text, got EarlyReturn")
            }
        }
        assert!(!crate::engine::cooldown::is_model_in_cooldown(agent, model));
    }

    /// MiniMax/Kimi parse_response synthesizes status=done from any assistant
    /// text. That task-path fallback must not be treated as a review approval.
    #[serial_test::serial(cooldown_state)]
    #[tokio::test]
    async fn parse_review_output_does_not_treat_task_done_synthesis_as_approval() {
        crate::engine::cooldown::reset_global_state().await;
        let store = Arc::new(TaskStore::open_memory().await.unwrap());
        let work_dir = TempDir::new().unwrap();
        let agent = "kimi";
        let model = "opus-3734-done";
        let ctx = test_review_context(agent, model, work_dir.path());
        let ndjson = r#"{"type":"assistant","message":{"content":[{"type":"text","text":"analyzing the diff"}]}}"#;
        write_review_output(work_dir.path(), ndjson, 0);

        let task = test_review_task("3734-done");
        let phase = parse_review_output(
            &task,
            "owner/repo",
            &ctx,
            ReviewRun { run_id: None },
            &store,
        )
        .await;

        match phase {
            ReviewPhase::EarlyReturn(ReviewDecision::Failed(reason)) => {
                assert!(
                    reason.contains("parse error"),
                    "unfinished prose without a decision must not approve, got {reason}"
                );
            }
            ReviewPhase::Ready(_) => {
                panic!("task-path status=done synthesis must not become a review approval")
            }
            ReviewPhase::EarlyReturn(_) => panic!("expected Failed(parse error)"),
        }
    }

    /// Regression test for issue #3478: a successful review run must reset
    /// `failure_count:{agent}:{model}`, mirroring the runner's task-agent
    /// success path, for generic/transient failures (not the persistent-model
    /// path — see issue #3571, which deliberately gives persistent failures
    /// their own counter that a success does NOT reset).
    #[serial_test::serial(cooldown_state)]
    #[tokio::test]
    async fn finalize_review_run_success_resets_generic_model_failure_count() {
        crate::engine::cooldown::reset_global_state().await;
        let store = Arc::new(TaskStore::open_memory().await.unwrap());
        crate::engine::cooldown::init_cooldown_store(store.clone()).await;

        let agent = "test-3478-agent";
        let model = "test-3478-model";
        let failure_key = format!("failure_count:{agent}:{model}");

        // AgentFailed is a generic/transient error, routed to record_model_failure
        // (not record_persistent_model_failure), so it should still be reset by
        // an intervening success.
        let err = runner::agents::AgentError::AgentFailed {
            message: "agent reported a generic failure".to_string(),
        };
        record_review_agent_failure("task-3478-a", agent, Some(model), &err).await;
        tokio::time::sleep(std::time::Duration::from_millis(50)).await;
        assert_eq!(
            store.kv_get(&failure_key).await.unwrap().as_deref(),
            Some("1"),
            "sanity check: failure should have incremented the persisted count"
        );

        let work_dir = TempDir::new().unwrap();
        let ctx = test_review_context(agent, model, work_dir.path());
        let parsed = test_parsed_review(ReviewDecision::Approve);
        let task = test_review_task("1");

        finalize_review_run(&task, &ctx, &parsed, &store).await;
        tokio::time::sleep(std::time::Duration::from_millis(50)).await;

        assert_eq!(
            store.kv_get(&failure_key).await.unwrap().as_deref(),
            Some("0"),
            "successful review must reset failure_count:{{agent}}:{{model}} (issue #3478)"
        );
    }

    /// Regression test for issue #3478: successful reviews interleaved between
    /// isolated generic failures must keep each failure's backoff at the base
    /// duration instead of escalating, since the previous success reset the
    /// counter. Uses AgentFailed (generic/transient), not InvalidResponse
    /// (persistent-model path — see issue #3571).
    #[serial_test::serial(cooldown_state)]
    #[tokio::test]
    async fn finalize_review_run_interleaved_success_keeps_generic_cooldown_at_base() {
        crate::engine::cooldown::reset_global_state().await;
        let store = Arc::new(TaskStore::open_memory().await.unwrap());
        crate::engine::cooldown::init_cooldown_store(store.clone()).await;

        let agent = "test-3478-interleave-agent";
        let model = "test-3478-interleave-model";
        let cooldown_key = format!("{agent}:{model}");

        let work_dir = TempDir::new().unwrap();
        let ctx = test_review_context(agent, model, work_dir.path());
        let task = test_review_task("2");

        for i in 0..4 {
            let err = runner::agents::AgentError::AgentFailed {
                message: format!("failure {i}"),
            };
            record_review_agent_failure("task-3478-b", agent, Some(model), &err).await;

            let now = chrono::Utc::now().timestamp();
            let until = crate::engine::cooldown::cooldown_until(&cooldown_key)
                .expect("failure should set cooldown");
            let remaining = until.saturating_sub(now);

            assert!(
                remaining <= crate::engine::cooldown::BACKOFF_BASE_SECS + 60,
                "failure #{} must stay at base backoff since the prior success reset the \
                 counter (issue #3478); got {remaining}s, base is {}s — an escalating value \
                 here means review successes are not resetting failure_count",
                i + 1,
                crate::engine::cooldown::BACKOFF_BASE_SECS
            );

            // A successful review interleaves between each failure, mirroring the
            // real-world pattern from issue #3478 (reviews for other tasks succeed
            // in between isolated failures for this model).
            let parsed = test_parsed_review(ReviewDecision::Approve);
            finalize_review_run(&task, &ctx, &parsed, &store).await;
            tokio::time::sleep(std::time::Duration::from_millis(20)).await;
        }
    }

    /// Regression test for issue #3571: unlike generic failures, persistent
    /// model failures (e.g. review parse errors, InvalidResponse) must NOT be
    /// reset by an intervening successful review. Otherwise a model that fails
    /// structurally every so often — but succeeds most of the time — never
    /// escalates past the 4h base cooldown and stays in rotation indefinitely.
    #[serial_test::serial(cooldown_state)]
    #[tokio::test]
    async fn finalize_review_run_success_does_not_reset_persistent_model_failure_count() {
        crate::engine::cooldown::reset_global_state().await;
        let store = Arc::new(TaskStore::open_memory().await.unwrap());
        crate::engine::cooldown::init_cooldown_store(store.clone()).await;

        let agent = "test-3571-agent";
        let model = "test-3571-model";
        let cooldown_key = format!("{agent}:{model}");

        let work_dir = TempDir::new().unwrap();
        let ctx = test_review_context(agent, model, work_dir.path());
        let task = test_review_task("3");

        // First structural (parse-error-equivalent) failure -> base 4h cooldown.
        let err = runner::agents::AgentError::InvalidResponse {
            raw: "empty review output".to_string(),
        };
        record_review_agent_failure("task-3571-a", agent, Some(model), &err).await;
        tokio::time::sleep(std::time::Duration::from_millis(20)).await;

        // A successful review for the same model interleaves, exactly like the
        // real-world pattern from issue #3571.
        let parsed = test_parsed_review(ReviewDecision::Approve);
        finalize_review_run(&task, &ctx, &parsed, &store).await;
        tokio::time::sleep(std::time::Duration::from_millis(20)).await;

        // Second structural failure must escalate from count 2 (~12h), not
        // reset back to the 4h base, despite the intervening success.
        let err2 = runner::agents::AgentError::InvalidResponse {
            raw: "empty review output again".to_string(),
        };
        record_review_agent_failure("task-3571-b", agent, Some(model), &err2).await;
        tokio::time::sleep(std::time::Duration::from_millis(20)).await;

        let now = chrono::Utc::now().timestamp();
        let until = crate::engine::cooldown::cooldown_until(&cooldown_key)
            .expect("second failure should set cooldown");
        let remaining = until.saturating_sub(now);

        assert!(
            remaining > crate::engine::cooldown::PERSISTENT_MODEL_BACKOFF_BASE_SECS,
            "second persistent model failure should escalate past the 4h base despite an \
             intervening success (issue #3571); got {remaining}s, base is {}s",
            crate::engine::cooldown::PERSISTENT_MODEL_BACKOFF_BASE_SECS
        );
    }
}