orch-cli 0.87.0

//! Reacts to NeedsReview events — spawns review agent immediately.

use crate::backends::{ExternalBackend, ExternalId, Status};
use crate::engine::cooldown::AGENT_COOLDOWN_SECS;
use crate::engine::dispatch_guard::DispatchGuard;
use crate::engine::events::TaskEvent;
use crate::engine::review::{review_and_merge, ReviewDecision, MAX_REVIEW_AGENT_FAILURES};
use crate::engine::router::Router;
use crate::engine::tasks::TaskManager;
use crate::github::http::GhHttp;
use crate::repo_context::REPO_CONTEXT;
use crate::store::{opt_store_get_task, TaskStore};
use crate::tmux::TmuxManager;
use dashmap::DashMap;
use std::sync::Arc;
use tokio::sync::broadcast;
use tokio::sync::{RwLock, Semaphore};

enum ReviewOutcome {
    Reset,
    RateLimited,
    /// PR was approved but mergeability not yet computed — retry the merge
    /// without resetting to NeedsReview (avoids spawning a redundant review agent).
    RetryMerge,
    /// Re-route to the task agent (e.g. merge conflicts that need the agent to fix).
    Reroute,
    Block(String),
    Ok,
}

/// Spawn a task that listens for NeedsReview events and triggers the review agent.
///
/// This is the sole trigger for NeedsReview → InReview transitions. The sync tick
/// no longer has a competing NeedsReview loop (removed in fix for issue #857 — both
/// paths firing simultaneously caused double failure-counter increments).
#[allow(clippy::too_many_arguments)]
pub fn spawn(
    mut rx: broadcast::Receiver<TaskEvent>,
    backend: Arc<dyn ExternalBackend>,
    tmux: Arc<TmuxManager>,
    semaphore: Arc<Semaphore>,
    task_manager: Arc<TaskManager>,
    router: Arc<RwLock<Router>>,
    dispatching: Arc<DashMap<String, String>>,
    store: Arc<TaskStore>,
    repo: String,
) {
    tokio::spawn(async move {
        loop {
            match rx.recv().await {
                Ok(event) if event.new_status == "needs_review" && event.repo == repo => {
                    // Check if review agent is enabled (read fresh each time).
                    let enable_review = crate::config::get("workflow.enable_review_agent")
                        .map(|v| v != "false")
                        .unwrap_or(true);
                    if !enable_review {
                        continue;
                    }

                    let task_id = &event.task_id;
                    let dispatch_key = format!("{}/{}", repo, task_id);

                    // Atomically claim the task before any async work so concurrent
                    // review events cannot double-spawn the same review flow.
                    // Using DashMap lets us distinguish same-task duplicates (expected) from
                    // key collisions involving a different task (should never happen).
                    {
                        use dashmap::mapref::entry::Entry;
                        match dispatching.entry(dispatch_key.clone()) {
                            Entry::Occupied(existing) => {
                                let existing_id = existing.get().clone();
                                drop(existing);
                                if existing_id == *task_id {
                                    tracing::debug!(
                                        task_id,
                                        "task locked by dispatch flow, skipping event-driven review"
                                    );
                                } else {
                                    tracing::warn!(
                                        task_id,
                                        existing_task_id = existing_id,
                                        dispatch_key,
                                        "dispatch key collision: unexpected task already holds this key"
                                    );
                                }
                                continue;
                            }
                            Entry::Vacant(slot) => {
                                slot.insert(task_id.clone());
                            }
                        }
                    }

                    // Create guard IMMEDIATELY after inserting the key — owns removal for all subsequent exit paths
                    let dispatch_guard =
                        DispatchGuard::new(dispatching.clone(), dispatch_key.clone());

                    tracing::info!(task_id, "event-driven review triggered");

                    // Look up the task from the store to get an ExternalTask.
                    let task = {
                        if crate::engine::tasks::is_internal_id(task_id)
                            || match store.has_external_tasks(&repo).await {
                                Ok(has_external) => has_external,
                                Err(e) => {
                                    tracing::warn!(
                                        repo = %repo,
                                        task_id,
                                        error = %e,
                                        "failed to check external task sentinel; keeping store-first review lookup"
                                    );
                                    true
                                }
                            }
                        {
                            // O(1) indexed lookup — resolve task_id to its SQLite row id,
                            // then fetch directly instead of scanning the whole needs_review set.
                            let result = if let Some(numeric_id) =
                                crate::engine::tasks::parse_internal_id(task_id)
                            {
                                store.get(numeric_id).await.map(Some)
                            } else {
                                store.get_by_external_id(&repo, task_id).await
                            };
                            match result {
                                Ok(Some(t))
                                    if t.status == crate::store::TaskStatus::NeedsReview =>
                                {
                                    Some(crate::engine::tasks::store_task_to_external(&t))
                                }
                                Ok(Some(t)) => {
                                    tracing::debug!(
                                        task_id,
                                        status = t.status.as_str(),
                                        "task found but not in needs_review status, skipping"
                                    );
                                    None
                                }
                                Ok(None) => None,
                                Err(e) => {
                                    tracing::warn!(task_id, err = %e, "store lookup failed for review");
                                    None
                                }
                            }
                        } else {
                            // Fall back to backend
                            match backend.list_by_status(Status::NeedsReview).await {
                                Ok(tasks) => tasks.into_iter().find(|t| t.id.0 == *task_id),
                                Err(e) => {
                                    tracing::warn!(task_id, err = %e, "backend lookup failed for review");
                                    None
                                }
                            }
                        }
                    };

                    let Some(task) = task else {
                        tracing::debug!(
                            task_id,
                            "task not found in needs_review list (may have already transitioned)"
                        );
                        continue;
                    };

                    if let Some(stored) =
                        opt_store_get_task(&Some(store.clone()), &repo, &task.id.0).await
                    {
                        let no_code_review = stored.worktree.trim().is_empty()
                            && stored.branch.trim().is_empty()
                            && stored.pr_number.is_none();
                        if no_code_review {
                            tracing::info!(
                                task_id,
                                "no worktree/branch/pr for needs_review task — skipping review dispatch and marking done"
                            );
                            crate::store::store_reset_failure_counters(
                                &Some(store.clone()),
                                &repo,
                                &task.id.0,
                            )
                            .await;
                            if let Err(e) = crate::store::store_set_result(
                                &Some(store.clone()),
                                &repo,
                                &task.id.0,
                                &[("needs_review_refires", serde_json::json!(0))],
                            )
                            .await
                            {
                                tracing::warn!(
                                    task_id,
                                    err = %e,
                                    "failed to reset needs_review_refires while skipping no-code review"
                                );
                            }
                            if let Err(e) = task_manager
                                .update_task_status(&task.id, Status::Done)
                                .await
                            {
                                tracing::error!(
                                    task_id,
                                    err = %e,
                                    "failed to mark no-code needs_review task as done"
                                );
                            }
                            continue;
                        }
                    }

                    // If no agents are currently routable (all cooled, degraded, or without
                    // an available review model), skip this event and let sync re-fire when
                    // an agent is available again. Sleeping inline here would block every
                    // subsequent NeedsReview event for this repo for the duration of the
                    // cooldown (potentially hours/days), causing sync's rebroadcast loop to
                    // escalate piled-up tasks to Blocked.
                    {
                        let router_guard = router.read().await;
                        let no_routable = !router_guard.available_agents.is_empty()
                            && router_guard.healthy_agent_count("review") == 0;
                        drop(router_guard);
                        if no_routable {
                            tracing::info!(
                                task_id,
                                "no routable review agents — leaving task in needs_review for sync to re-fire"
                            );
                            continue;
                        }
                    }

                    // Try to acquire a semaphore permit.
                    let permit = match semaphore.clone().try_acquire_owned() {
                        Ok(p) => p,
                        Err(_) => {
                            tracing::debug!(task_id, "all parallel slots busy; sync catch-up will re-fire when a slot opens");
                            continue;
                        }
                    };

                    // Transition to InReview — this IS the atomic guard against duplicates.
                    if let Err(e) = task_manager
                        .update_task_status(&task.id, Status::InReview)
                        .await
                    {
                        tracing::warn!(task_id, err = %e, "failed to transition to InReview");
                        drop(permit);
                        continue;
                    }

                    crate::store::set_review_session_expected(&store, &repo, &task.id.0, true)
                        .await;

                    let backend_c = backend.clone();
                    let task_manager_c = task_manager.clone();
                    let tmux_c = tmux.clone();
                    let router_c = router.clone();
                    let store_c = store.clone();
                    let repo_s = repo.clone();
                    tokio::spawn(REPO_CONTEXT.scope(repo_s.clone(), async move {
                        let _dispatch_guard = dispatch_guard; // released on drop (normal or panic)
                        let tid = task.id.0.clone();

                        // Re-check agent routability inside the spawned task. A concurrent
                        // review that ran between the outer dispatch check and now may have
                        // set a cooldown — bail early rather than burning a review run
                        // against a rate-limited agent.
                        let pre_check_all_cooled = {
                            let r = router_c.read().await;
                            !r.available_agents.is_empty() && r.healthy_agent_count("review") == 0
                        };
                        if pre_check_all_cooled {
                            tracing::info!(
                                task_id = tid,
                                "all agents cooled at review spawn time — resetting to needs_review for sync to re-fire"
                            );
                            crate::store::set_review_session_expected(
                                &store_c,
                                &repo_s,
                                &tid,
                                false,
                            )
                            .await;
                            if let Err(e) = task_manager_c
                                .update_task_status(
                                    &ExternalId(tid.clone()),
                                    Status::NeedsReview,
                                )
                                .await
                            {
                                tracing::error!(task_id = %tid, err = %e, "pre-check: update_task_status(NeedsReview) failed — task may be stuck in InReview");
                            }
                            drop(permit);
                            return;
                        }

                        let outcome = match review_and_merge(
                            &task,
                            &backend_c,
                            &tmux_c,
                            &repo_s,
                            &router_c,
                            &task_manager_c,
                            &store_c,
                        )
                        .await
                        {
                            Ok(ReviewDecision::Blocked(reason)) => {
                                tracing::error!(
                                    task_id = tid,
                                    reason,
                                    "review gate blocked after repeated failures — marking task blocked"
                                );
                                ReviewOutcome::Block(format!(
                                    "review gate blocked after repeated failures: {reason}"
                                ))
                            }
                            Ok(ReviewDecision::Failed(reason)) => {
                                classify_review_failure(
                                    &store_c, &repo_s, &tid, &reason, "review agent",
                                )
                                .await
                            }
                            Err(e) => {
                                let reason = format!("{e:#}");
                                classify_review_failure(
                                    &store_c, &repo_s, &tid, &reason, "review_and_merge",
                                )
                                .await
                            }
                            Ok(ReviewDecision::Approve) => {
                                // On approval we clear transient failure counters
                                // but preserve the `review_cycles` counter which
                                // tracks how many times the PR has requested
                                // changes and is used as a circuit-breaker.
                                crate::store::store_reset_failure_counters(
                                    &Some(store_c.clone()),
                                    &repo_s,
                                    &tid,
                                )
                                .await;
                                // Also reset needs_review_refires so the sync backoff doesn't persist across a successful review
                                if let Err(e) = crate::store::store_set_result(
                                    &Some(store_c.clone()),
                                    &repo_s,
                                    &tid,
                                    &[("needs_review_refires", serde_json::json!(0))],
                                )
                                .await
                                {
                                    tracing::warn!(task_id = %tid, err = %e, "failed to write needs_review_refires to store");
                                }

                                // Approval only means the code looks good — the PR
                                // merge is the actual completion signal.  Leave the
                                // task in InReview.  The review_open_prs polling loop
                                // detects `batch_data.merged == true` on the next tick
                                // and transitions the task to Done correctly.
                                tracing::info!(
                                    task_id = tid,
                                    "approval received — leaving task in_review until PR is merged"
                                );
                                ReviewOutcome::Ok
                            }
                            Ok(ReviewDecision::Rerouted) => {
                                // Task was re-routed because there was no PR and no commits.
                                // Do NOT reset failure counters — no_code_reroutes must accumulate
                                // so the max_reroute_attempts safety limit can be reached.
                                ReviewOutcome::Ok
                            }
                            Ok(ReviewDecision::Skipped) => {
                                // Skipped indicates there's nothing to review (no PR etc.).
                                // Clear transient failure counters but preserve review_cycles.
                                crate::store::store_reset_failure_counters(
                                    &Some(store_c.clone()),
                                    &repo_s,
                                    &tid,
                                )
                                .await;
                                if let Err(e) = crate::store::store_set_result(
                                    &Some(store_c.clone()),
                                    &repo_s,
                                    &tid,
                                    &[("needs_review_refires", serde_json::json!(0))],
                                )
                                .await
                                {
                                    tracing::warn!(task_id = %tid, err = %e, "failed to write needs_review_refires to store");
                                }
                                ReviewOutcome::Ok
                            }
                            Ok(ReviewDecision::RequestChanges { .. }) => {
                                // handle_review_changes already incremented review_cycles —
                                // only reset transient per-attempt counters.
                                crate::store::store_reset_failure_counters(
                                    &Some(store_c.clone()),
                                    &repo_s,
                                    &tid,
                                )
                                .await;
                                if let Err(e) = crate::store::store_set_result(
                                    &Some(store_c.clone()),
                                    &repo_s,
                                    &tid,
                                    &[("needs_review_refires", serde_json::json!(0))],
                                )
                                .await
                                {
                                    tracing::warn!(task_id = %tid, err = %e, "failed to write needs_review_refires to store");
                                }
                                ReviewOutcome::Ok
                            }
                        };
                        crate::store::set_review_session_expected(
                            &store_c,
                            &repo_s,
                            &tid,
                            false,
                        )
                        .await;

                        // Freshness guard: re-fetch the task's current status before applying
                        // any outcome transitions. A concurrent path (e.g. PR merge, manual
                        // close, or another review cycle) may have already moved the task out
                        // of `in_review`. If so, discard the stale outcome rather than
                        // overwriting the newer state.
                        //
                        // IMPORTANT: Distinguish between DB errors and legitimate status changes.
                        // A DB lookup failure should NOT be treated as "task moved out of in_review"
                        // — we should retain the review outcome rather than discarding it on
                        // transient errors.
                        let fetched: Result<Option<(crate::store::TaskStatus, Option<String>)>, anyhow::Error> = async {
                            let id = match store_c.resolve_task_id(&repo_s, &tid).await {
                                Ok(Some(id)) => id,
                                Ok(None) => return Ok(None), // Task doesn't exist — treat as missing
                                Err(e) => return Err(e),      // DB error — propagate, don't discard
                            };
                            let task = store_c.get(id).await?;
                            Ok(Some((task.status, task.block_reason)))
                        }
                        .await;

                        let (current_status, current_block_reason) = match fetched {
                            Ok(Some((status, block_reason))) => (Some(status), block_reason),
                            Ok(None) => {
                                // Task doesn't exist in store — treat as not in_review
                                (None, None)
                            }
                            Err(e) => {
                                // DB error — log and keep the review outcome instead of discarding
                                tracing::error!(
                                    task_id = tid,
                                    error = %e,
                                    "review outcome retained — DB lookup failed, not treating as stale"
                                );
                                // Skip the stale outcome check and proceed with the outcome
                                // by treating it as if the task is still InReview
                                (Some(crate::store::TaskStatus::InReview), None)
                            }
                        };

                        if !matches!(current_status, Some(crate::store::TaskStatus::InReview)) {
                            let ci_blocked = matches!(
                                current_status,
                                Some(crate::store::TaskStatus::Blocked)
                            ) && current_block_reason
                                .as_deref()
                                .is_some_and(|r| r.contains("CI failure limit"));
                            let expected = ci_blocked
                                || matches!(
                                    current_status,
                                    Some(crate::store::TaskStatus::Done)
                                        | Some(crate::store::TaskStatus::Routed)
                                        | Some(crate::store::TaskStatus::New)
                                );
                            if expected {
                                tracing::debug!(
                                    task_id = tid,
                                    current_status = current_status.as_ref().map(|s| s.as_str()).unwrap_or("unknown"),
                                    "review outcome discarded — task transitioned away from in_review (expected)"
                                );
                            } else {
                                tracing::warn!(
                                    task_id = tid,
                                    current_status = current_status.as_ref().map(|s| s.as_str()).unwrap_or("unknown"),
                                    "review outcome discarded — task is no longer in_review (stale review agent result)"
                                );
                            }

                            // Kill the orphaned review session
                            let stale_session = tmux_c.session_name(&repo_s, &format!("{}-review", tid));
                            if let Err(e) = tmux_c.kill_session(&stale_session).await {
                                tracing::debug!(task_id = tid, error = %e, "failed to kill stale review session");
                            }

                            drop(permit);
                            return;
                        }

                        match outcome {
                            ReviewOutcome::Reset => {
                                // Kill any stale tmux review session before resetting — the
                                // session may still be alive if the agent hit a rate limit,
                                // if two concurrent reviewers raced and the loser's spawn
                                // failed with "duplicate session", or if review_and_merge
                                // returned an error before reaching its own cleanup.
                                // Killing a non-existent session is a harmless no-op.
                                let stale_session =
                                    tmux_c.session_name(&repo_s, &format!("{}-review", tid));
                                tmux_c.kill_session(&stale_session).await.ok();
                                // Backoff before re-queuing to prevent rapid spin
                                // if the review agent keeps failing instantly.
                                tokio::time::sleep(std::time::Duration::from_secs(10)).await;
                                if let Err(e) = task_manager_c
                                    .update_task_status(
                                        &ExternalId(tid.clone()),
                                        Status::NeedsReview,
                                    )
                                    .await
                                {
                                    tracing::error!(task_id = %tid, err = %e, "update_task_status(NeedsReview) failed — task may be stuck in InReview");
                                }
                            }
                            ReviewOutcome::Reroute => {
                                // Merge conflict or similar — route back to the
                                // task agent so it can fix the issue.
                                let stale_session =
                                    tmux_c.session_name(&repo_s, &format!("{}-review", tid));
                                tmux_c.kill_session(&stale_session).await.ok();
                                tracing::info!(
                                    task_id = tid,
                                    "re-routing to task agent to resolve merge conflict"
                                );
                                if let Err(e) = task_manager_c
                                    .update_task_status(
                                        &ExternalId(tid.clone()),
                                        Status::Routed,
                                    )
                                    .await
                                {
                                    tracing::error!(task_id = %tid, err = %e, "update_task_status(Routed) failed");
                                }
                            }
                            ReviewOutcome::RetryMerge => {
                                // PR was already approved; GitHub just hasn't computed
                                // mergeability yet.  Keep the task in InReview and wait
                                // for review_poll to retry the merge — do NOT reset to
                                // NeedsReview, which would spawn a redundant review agent.
                                tracing::info!(
                                    task_id = tid,
                                    "mergeability pending — task stays in InReview, review_poll will retry merge"
                                );
                                // Brief pause so the next review_poll cycle fires soon.
                                tokio::time::sleep(std::time::Duration::from_secs(5)).await;
                            }
                            ReviewOutcome::RateLimited => {
                                // Kill stale tmux session — same as Reset.
                                let stale_session =
                                    tmux_c.session_name(&repo_s, &format!("{}-review", tid));
                                tmux_c.kill_session(&stale_session).await.ok();

                                // Compute wait duration: if any agent is available right now,
                                // use a short backoff (30 s). If every agent is cooled, sleep
                                // until the earliest cooldown expires so we don't spin.
                                let wait_secs = {
                                    let r = router_c.read().await;
                                    if r.healthy_agent_count("review") > 0 {
                                        30u64
                                    } else {
                                        let now = chrono::Utc::now().timestamp();
                                        r.earliest_cooldown_until(Some("review"))
                                            .map(|until| ((until - now).max(1)) as u64)
                                            .unwrap_or(AGENT_COOLDOWN_SECS as u64)
                                    }
                                };
                                tracing::info!(
                                    task_id = tid,
                                    wait_secs,
                                    "deferring review retry due to agent rate limit cooldown"
                                );
                                tokio::time::sleep(std::time::Duration::from_secs(wait_secs)).await;
                                if let Err(e) = task_manager_c
                                    .update_task_status(
                                        &ExternalId(tid.clone()),
                                        Status::NeedsReview,
                                    )
                                    .await
                                {
                                    tracing::error!(task_id = %tid, err = %e, "update_task_status(NeedsReview) failed after rate limit backoff — task may be stuck in InReview");
                                }
                            }
                             ReviewOutcome::Block(reason) => {
                                 // Persist block_reason BEFORE transitioning to Blocked to avoid
                                 // a race where auto_unblock sees a blocked task without a reason
                                 // and immediately unblocks it.
                                 let fields = [
                                     (
                                         "block_reason",
                                         serde_json::json!("review agent blocked — exceeded failure threshold"),
                                     ),
                                     ("last_error", serde_json::json!(reason)),
                                 ];
                                 if let Err(e) = task_manager_c
                                     .update_task_status_and_result(
                                         &ExternalId(tid.clone()),
                                         Status::Blocked,
                                         &fields,
                                     )
                                     .await
                                 {
                                     tracing::error!(task_id = %tid, err = %e, "update_task_status_and_result(Blocked) failed — task may be stuck in InReview");
                                 }
                             }
                            ReviewOutcome::Ok => {}
                        }

                        drop(permit);
                        // _dispatch_guard dropped here — releases the per-task lock.
                    }));

                    // Small delay between dispatches to allow cooldowns from
                    // early failures to propagate before subsequent reviews are
                    // dispatched. This prevents batch dispatch of N reviews to
                    // the same rate-limited agent when events arrive concurrently.
                    tokio::time::sleep(std::time::Duration::from_secs(2)).await;
                }
                Ok(_) => {} // Not a needs_review event or different repo
                Err(broadcast::error::RecvError::Lagged(n)) => {
                    tracing::debug!(n, "review subscriber lagged; sync catch-up will re-fire missed NeedsReview tasks");
                }
                Err(_) => break, // Channel closed
            }
        }
    });
}

/// Classify a review failure and return the appropriate outcome.
///
/// Both the `Ok(ReviewDecision::Failed)` and `Err(e)` arms share the same
/// logic; callers convert the error to a string before calling this helper.
/// `context` is used in log messages to distinguish the two call sites
/// ("review agent" vs "review_and_merge").
async fn classify_review_failure(
    store: &Arc<TaskStore>,
    repo: &str,
    task_id: &str,
    reason: &str,
    context: &str,
) -> ReviewOutcome {
    // Rate-limit and auth failures are not real review failures — don't count them
    // toward MAX_REVIEW_AGENT_FAILURES. The cooldown system steers the next
    // attempt to an available agent.
    let lower_reason = reason.to_lowercase();

    // Parse errors are orch's fault, not review feedback. Re-route for retry
    // without counting toward the failure threshold, so parse errors don't
    // incorrectly block tasks that have no actual review feedback.
    if lower_reason.contains("parse error") {
        tracing::warn!(
            task_id,
            reason,
            "{context} hit parse error — re-routing for retry without counting as failure"
        );
        return ReviewOutcome::Reset;
    }

    // The review output was cut off by the model's output/reasoning token
    // budget, not a genuine parse/format failure. Same treatment as a parse
    // error: re-route without counting toward MAX_REVIEW_AGENT_FAILURES.
    if lower_reason.contains("truncated") {
        tracing::warn!(
            task_id,
            reason,
            "{context} output truncated by token budget — re-routing for retry without counting as failure"
        );
        return ReviewOutcome::Reset;
    }

    // Transient mergeability check: GitHub hasn't finished computing yet.
    // This is a normal, expected transient state for freshly-pushed PRs.
    // Return RetryMerge (not Reset) so the task stays in InReview — avoids
    // spawning a redundant review agent that would just re-approve anyway.
    if lower_reason.contains("not yet computed") {
        tracing::warn!(
            task_id,
            reason,
            "{context} PR mergeability check still pending — will retry merge without re-reviewing"
        );
        return ReviewOutcome::RetryMerge;
    }

    // Merge conflicts should be handled by handle_merge_conflict() in
    // auto_merge.rs (which routes to the task agent). If a conflict error
    // leaks here, route to the task agent instead of resetting to NeedsReview
    // (which would re-spawn the review agent in a loop).
    if lower_reason.contains("merge conflict")
        || lower_reason.contains("not mergeable")
        || lower_reason.contains("merge failed")
    {
        tracing::warn!(
            task_id,
            reason,
            "{context} hit merge conflict — routing to task agent to resolve"
        );
        return ReviewOutcome::Reroute;
    }

    // No configured review model for any available agent, or the selected agent
    // has no review model. This is a configuration/cooldown gap, not a review
    // failure. Defer without incrementing the failure counter so the task retries
    // once a review model is available.
    if lower_reason.contains("no review agent") || lower_reason.contains("review model missing") {
        tracing::warn!(
            task_id,
            reason,
            "{context}: no review model available — deferring retry until one is available"
        );
        return ReviewOutcome::RateLimited;
    }

    if lower_reason.contains("rate limit") || lower_reason.contains("auth error") {
        tracing::warn!(
            task_id,
            reason,
            "{context} hit rate limit or auth error — deferring retry until cooldown expires"
        );
        return ReviewOutcome::RateLimited;
    }

    if crate::engine::runner::git_ops::is_transient_github_api_error(reason) {
        tracing::warn!(
            task_id,
            reason,
            "{context} hit transient GitHub error — deferring retry without counting as failure"
        );
        return ReviewOutcome::Reset;
    }

    // Provider-side network errors (ConnectionRefused, timeouts, etc.) are
    // transient infrastructure issues, not review feedback. Re-route without
    // counting toward MAX_REVIEW_AGENT_FAILURES.
    if crate::engine::runner::agents::patterns::detect_network_error(reason).is_some() {
        tracing::warn!(
            task_id,
            reason,
            "{context} hit transient provider network error — re-routing for retry without counting as failure"
        );
        return ReviewOutcome::Reset;
    }

    let failures = match crate::store::store_increment(
        &Some(store.clone()),
        repo,
        task_id,
        "review_agent_failures",
    )
    .await
    {
        Ok(v) => v,
        Err(e) => {
            tracing::warn!(task_id, err = %e, "failed to increment review_agent_failures — skipping failure escalation this tick");
            // Treat as 0 for now (no escalation)
            0
        }
    };
    let blocking = failures >= MAX_REVIEW_AGENT_FAILURES;
    // Post failure comment to the PR so the history is visible.
    post_review_failure_comment(store, repo, task_id, reason, failures, blocking).await;
    if blocking {
        tracing::error!(
            task_id,
            reason,
            failures,
            "{context} failed too many times — blocking task"
        );
        ReviewOutcome::Block(format!("{context} failed {failures} times: {reason}"))
    } else {
        tracing::error!(
            task_id,
            reason,
            failures,
            "{context} failed — resetting to NeedsReview for retry"
        );
        ReviewOutcome::Reset
    }
}

/// Post a comment on the PR explaining why the review agent failed.
/// Best-effort — if the PR number is missing or the API call fails, we just log.
async fn post_review_failure_comment(
    store: &Arc<TaskStore>,
    repo: &str,
    task_id: &str,
    reason: &str,
    failures: u64,
    blocking: bool,
) {
    let pr_number = match opt_store_get_task(&Some(store.clone()), repo, task_id).await {
        Some(t) => match t.pr_number {
            Some(n) => n,
            None => return, // No PR to comment on
        },
        _ => return, // No PR to comment on
    };

    let status = if blocking {
        "Task blocked for human review."
    } else {
        "Retrying with a different agent."
    };

    let comment = format!(
        "⚠️ Review agent failed (attempt {}/{})\n\n**Reason:** {}\n\n{}{}",
        failures,
        MAX_REVIEW_AGENT_FAILURES,
        reason,
        status,
        crate::engine::orch_footer()
    );

    let gh = match GhHttp::new() {
        Ok(gh) => gh,
        Err(_) => return,
    };
    if let Err(e) = gh.add_comment(repo, &pr_number.to_string(), &comment).await {
        tracing::debug!(
            task_id,
            pr_number,
            error = %e,
            "failed to post review failure comment"
        );
    }
}

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

    /// A truncated review output (opencode step_finish reason=length) must be
    /// treated like a parse error: reset for retry without counting toward
    /// MAX_REVIEW_AGENT_FAILURES, so a run of token-budget flukes never
    /// permanently blocks the task (issue #3473).
    #[tokio::test]
    async fn classify_review_failure_truncated_resets_without_counting() {
        let store = Arc::new(TaskStore::open_memory().await.unwrap());
        let repo = "owner/repo";
        let store_id = store
            .create_internal(
                repo,
                "Truncated review test",
                "",
                "review",
                "review-1",
                None,
            )
            .await
            .unwrap();
        let task_id = format!("internal:{store_id}");

        for _ in 0..(MAX_REVIEW_AGENT_FAILURES + 1) {
            let outcome = classify_review_failure(
                &store,
                repo,
                &task_id,
                "truncated: review output was cut off before completion \
                 (output/reasoning token budget exceeded)",
                "review agent",
            )
            .await;
            assert!(
                matches!(outcome, ReviewOutcome::Reset),
                "truncated review failure must reset for retry, not block"
            );
        }

        let task = opt_store_get_task(&Some(store.clone()), repo, &task_id)
            .await
            .expect("task should exist");
        assert_eq!(
            task.review_agent_failures, 0,
            "truncated failures must not count toward review_agent_failures"
        );
    }

    /// Transient provider network errors (ConnectionRefused, timeouts) must be
    /// treated like parse errors: reset for retry without counting toward
    /// MAX_REVIEW_AGENT_FAILURES (issue #3482).
    #[tokio::test]
    async fn classify_review_failure_network_error_resets_without_counting() {
        let store = Arc::new(TaskStore::open_memory().await.unwrap());
        let repo = "owner/repo";
        let store_id = store
            .create_internal(
                repo,
                "Network error review test",
                "",
                "review",
                "review-2",
                None,
            )
            .await
            .unwrap();
        let task_id = format!("internal:{store_id}");

        let error_messages = [
            "API Error: Unable to connect to API (ConnectionRefused)",
            "connection refused",
            "ECONNREFUSED",
            "Upstream idle timeout exceeded",
            "connect ETIMEDOUT",
            "network request failed",
        ];

        for msg in &error_messages {
            let outcome =
                classify_review_failure(&store, repo, &task_id, msg, "review agent").await;
            assert!(
                matches!(outcome, ReviewOutcome::Reset),
                "network error '{msg}' must reset for retry, not block"
            );
        }

        let task = opt_store_get_task(&Some(store.clone()), repo, &task_id)
            .await
            .expect("task should exist");
        assert_eq!(
            task.review_agent_failures, 0,
            "network errors must not count toward review_agent_failures"
        );
    }

    /// When no agent has a configured review model (or the fallback agent has none),
    /// the review must be deferred, not counted as a review failure. Counting it
    /// would eventually block tasks for a configuration/cooldown gap (issue #3584).
    #[tokio::test]
    async fn classify_review_failure_missing_review_model_is_rate_limited() {
        let store = Arc::new(TaskStore::open_memory().await.unwrap());
        let repo = "owner/repo";
        let store_id = store
            .create_internal(
                repo,
                "Missing review model test",
                "",
                "review",
                "review-3",
                None,
            )
            .await
            .unwrap();
        let task_id = format!("internal:{store_id}");

        let error_messages = [
            "no review agent with a configured review model is available for task internal:123",
            "review model missing for task internal:123 after agent selection — refusing to dispatch",
        ];

        for _ in 0..(MAX_REVIEW_AGENT_FAILURES + 1) {
            for msg in &error_messages {
                let outcome =
                    classify_review_failure(&store, repo, &task_id, msg, "review_and_merge").await;
                assert!(
                    matches!(outcome, ReviewOutcome::RateLimited),
                    "missing review model error '{msg}' must defer, not block"
                );
            }
        }

        let task = opt_store_get_task(&Some(store.clone()), repo, &task_id)
            .await
            .expect("task should exist");
        assert_eq!(
            task.review_agent_failures, 0,
            "missing review model errors must not count toward review_agent_failures"
        );
    }
}