orch/engine/subscribers/review.rs
1//! Reacts to NeedsReview events — spawns review agent immediately.
2
3use crate::backends::{ExternalBackend, ExternalId, Status};
4use crate::engine::cooldown::AGENT_COOLDOWN_SECS;
5use crate::engine::dispatch_guard::DispatchGuard;
6use crate::engine::events::TaskEvent;
7use crate::engine::review::{review_and_merge, ReviewDecision, MAX_REVIEW_AGENT_FAILURES};
8use crate::engine::router::Router;
9use crate::engine::tasks::TaskManager;
10use crate::github::http::GhHttp;
11use crate::repo_context::REPO_CONTEXT;
12use crate::store::{opt_store_get_task, TaskStore};
13use crate::tmux::TmuxManager;
14use dashmap::DashMap;
15use std::sync::Arc;
16use tokio::sync::broadcast;
17use tokio::sync::{RwLock, Semaphore};
18
19enum ReviewOutcome {
20 Reset,
21 RateLimited,
22 /// PR was approved but mergeability not yet computed — retry the merge
23 /// without resetting to NeedsReview (avoids spawning a redundant review agent).
24 RetryMerge,
25 /// Re-route to the task agent (e.g. merge conflicts that need the agent to fix).
26 Reroute,
27 Block(String),
28 Ok,
29}
30
31/// Spawn a task that listens for NeedsReview events and triggers the review agent.
32///
33/// This is the sole trigger for NeedsReview → InReview transitions. The sync tick
34/// no longer has a competing NeedsReview loop (removed in fix for issue #857 — both
35/// paths firing simultaneously caused double failure-counter increments).
36#[allow(clippy::too_many_arguments)]
37pub fn spawn(
38 mut rx: broadcast::Receiver<TaskEvent>,
39 backend: Arc<dyn ExternalBackend>,
40 tmux: Arc<TmuxManager>,
41 semaphore: Arc<Semaphore>,
42 task_manager: Arc<TaskManager>,
43 router: Arc<RwLock<Router>>,
44 dispatching: Arc<DashMap<String, String>>,
45 store: Arc<TaskStore>,
46 repo: String,
47) {
48 tokio::spawn(async move {
49 loop {
50 match rx.recv().await {
51 Ok(event) if event.new_status == "needs_review" && event.repo == repo => {
52 // Check if review agent is enabled (read fresh each time).
53 let enable_review = crate::config::get("workflow.enable_review_agent")
54 .map(|v| v != "false")
55 .unwrap_or(true);
56 if !enable_review {
57 continue;
58 }
59
60 let task_id = &event.task_id;
61 let dispatch_key = format!("{}/{}", repo, task_id);
62
63 // Atomically claim the task before any async work so concurrent
64 // review events cannot double-spawn the same review flow.
65 // Using DashMap lets us distinguish same-task duplicates (expected) from
66 // key collisions involving a different task (should never happen).
67 {
68 use dashmap::mapref::entry::Entry;
69 match dispatching.entry(dispatch_key.clone()) {
70 Entry::Occupied(existing) => {
71 let existing_id = existing.get().clone();
72 drop(existing);
73 if existing_id == *task_id {
74 tracing::debug!(
75 task_id,
76 "task locked by dispatch flow, skipping event-driven review"
77 );
78 } else {
79 tracing::warn!(
80 task_id,
81 existing_task_id = existing_id,
82 dispatch_key,
83 "dispatch key collision: unexpected task already holds this key"
84 );
85 }
86 continue;
87 }
88 Entry::Vacant(slot) => {
89 slot.insert(task_id.clone());
90 }
91 }
92 }
93
94 // Create guard IMMEDIATELY after inserting the key — owns removal for all subsequent exit paths
95 let dispatch_guard =
96 DispatchGuard::new(dispatching.clone(), dispatch_key.clone());
97
98 tracing::info!(task_id, "event-driven review triggered");
99
100 // Look up the task from the store to get an ExternalTask.
101 let task = {
102 if crate::engine::tasks::is_internal_id(task_id)
103 || match store.has_external_tasks(&repo).await {
104 Ok(has_external) => has_external,
105 Err(e) => {
106 tracing::warn!(
107 repo = %repo,
108 task_id,
109 error = %e,
110 "failed to check external task sentinel; keeping store-first review lookup"
111 );
112 true
113 }
114 }
115 {
116 // O(1) indexed lookup — resolve task_id to its SQLite row id,
117 // then fetch directly instead of scanning the whole needs_review set.
118 let result = if let Some(numeric_id) =
119 crate::engine::tasks::parse_internal_id(task_id)
120 {
121 store.get(numeric_id).await.map(Some)
122 } else {
123 store.get_by_external_id(&repo, task_id).await
124 };
125 match result {
126 Ok(Some(t))
127 if t.status == crate::store::TaskStatus::NeedsReview =>
128 {
129 Some(crate::engine::tasks::store_task_to_external(&t))
130 }
131 Ok(Some(t)) => {
132 tracing::debug!(
133 task_id,
134 status = t.status.as_str(),
135 "task found but not in needs_review status, skipping"
136 );
137 None
138 }
139 Ok(None) => None,
140 Err(e) => {
141 tracing::warn!(task_id, err = %e, "store lookup failed for review");
142 None
143 }
144 }
145 } else {
146 // Fall back to backend
147 match backend.list_by_status(Status::NeedsReview).await {
148 Ok(tasks) => tasks.into_iter().find(|t| t.id.0 == *task_id),
149 Err(e) => {
150 tracing::warn!(task_id, err = %e, "backend lookup failed for review");
151 None
152 }
153 }
154 }
155 };
156
157 let Some(task) = task else {
158 tracing::debug!(
159 task_id,
160 "task not found in needs_review list (may have already transitioned)"
161 );
162 continue;
163 };
164
165 if let Some(stored) =
166 opt_store_get_task(&Some(store.clone()), &repo, &task.id.0).await
167 {
168 let no_code_review = stored.worktree.trim().is_empty()
169 && stored.branch.trim().is_empty()
170 && stored.pr_number.is_none();
171 if no_code_review {
172 tracing::info!(
173 task_id,
174 "no worktree/branch/pr for needs_review task — skipping review dispatch and marking done"
175 );
176 crate::store::store_reset_failure_counters(
177 &Some(store.clone()),
178 &repo,
179 &task.id.0,
180 )
181 .await;
182 if let Err(e) = crate::store::store_set_result(
183 &Some(store.clone()),
184 &repo,
185 &task.id.0,
186 &[("needs_review_refires", serde_json::json!(0))],
187 )
188 .await
189 {
190 tracing::warn!(
191 task_id,
192 err = %e,
193 "failed to reset needs_review_refires while skipping no-code review"
194 );
195 }
196 if let Err(e) = task_manager
197 .update_task_status(&task.id, Status::Done)
198 .await
199 {
200 tracing::error!(
201 task_id,
202 err = %e,
203 "failed to mark no-code needs_review task as done"
204 );
205 }
206 continue;
207 }
208 }
209
210 // If no agents are currently routable (all cooled, degraded, or without
211 // an available review model), skip this event and let sync re-fire when
212 // an agent is available again. Sleeping inline here would block every
213 // subsequent NeedsReview event for this repo for the duration of the
214 // cooldown (potentially hours/days), causing sync's rebroadcast loop to
215 // escalate piled-up tasks to Blocked.
216 {
217 let router_guard = router.read().await;
218 let no_routable = !router_guard.available_agents.is_empty()
219 && router_guard.healthy_agent_count("review") == 0;
220 drop(router_guard);
221 if no_routable {
222 tracing::info!(
223 task_id,
224 "no routable review agents — leaving task in needs_review for sync to re-fire"
225 );
226 continue;
227 }
228 }
229
230 // Try to acquire a semaphore permit.
231 let permit = match semaphore.clone().try_acquire_owned() {
232 Ok(p) => p,
233 Err(_) => {
234 tracing::debug!(task_id, "all parallel slots busy; sync catch-up will re-fire when a slot opens");
235 continue;
236 }
237 };
238
239 // Transition to InReview — this IS the atomic guard against duplicates.
240 if let Err(e) = task_manager
241 .update_task_status(&task.id, Status::InReview)
242 .await
243 {
244 tracing::warn!(task_id, err = %e, "failed to transition to InReview");
245 drop(permit);
246 continue;
247 }
248
249 crate::store::set_review_session_expected(&store, &repo, &task.id.0, true)
250 .await;
251
252 let backend_c = backend.clone();
253 let task_manager_c = task_manager.clone();
254 let tmux_c = tmux.clone();
255 let router_c = router.clone();
256 let store_c = store.clone();
257 let repo_s = repo.clone();
258 tokio::spawn(REPO_CONTEXT.scope(repo_s.clone(), async move {
259 let _dispatch_guard = dispatch_guard; // released on drop (normal or panic)
260 let tid = task.id.0.clone();
261
262 // Re-check agent routability inside the spawned task. A concurrent
263 // review that ran between the outer dispatch check and now may have
264 // set a cooldown — bail early rather than burning a review run
265 // against a rate-limited agent.
266 let pre_check_all_cooled = {
267 let r = router_c.read().await;
268 !r.available_agents.is_empty() && r.healthy_agent_count("review") == 0
269 };
270 if pre_check_all_cooled {
271 tracing::info!(
272 task_id = tid,
273 "all agents cooled at review spawn time — resetting to needs_review for sync to re-fire"
274 );
275 crate::store::set_review_session_expected(
276 &store_c,
277 &repo_s,
278 &tid,
279 false,
280 )
281 .await;
282 if let Err(e) = task_manager_c
283 .update_task_status(
284 &ExternalId(tid.clone()),
285 Status::NeedsReview,
286 )
287 .await
288 {
289 tracing::error!(task_id = %tid, err = %e, "pre-check: update_task_status(NeedsReview) failed — task may be stuck in InReview");
290 }
291 drop(permit);
292 return;
293 }
294
295 let outcome = match review_and_merge(
296 &task,
297 &backend_c,
298 &tmux_c,
299 &repo_s,
300 &router_c,
301 &task_manager_c,
302 &store_c,
303 )
304 .await
305 {
306 Ok(ReviewDecision::Blocked(reason)) => {
307 tracing::error!(
308 task_id = tid,
309 reason,
310 "review gate blocked after repeated failures — marking task blocked"
311 );
312 ReviewOutcome::Block(format!(
313 "review gate blocked after repeated failures: {reason}"
314 ))
315 }
316 Ok(ReviewDecision::Failed(reason)) => {
317 classify_review_failure(
318 &store_c, &repo_s, &tid, &reason, "review agent",
319 )
320 .await
321 }
322 Err(e) => {
323 let reason = format!("{e:#}");
324 classify_review_failure(
325 &store_c, &repo_s, &tid, &reason, "review_and_merge",
326 )
327 .await
328 }
329 Ok(ReviewDecision::Approve) => {
330 // On approval we clear transient failure counters
331 // but preserve the `review_cycles` counter which
332 // tracks how many times the PR has requested
333 // changes and is used as a circuit-breaker.
334 crate::store::store_reset_failure_counters(
335 &Some(store_c.clone()),
336 &repo_s,
337 &tid,
338 )
339 .await;
340 // Also reset needs_review_refires so the sync backoff doesn't persist across a successful review
341 if let Err(e) = crate::store::store_set_result(
342 &Some(store_c.clone()),
343 &repo_s,
344 &tid,
345 &[("needs_review_refires", serde_json::json!(0))],
346 )
347 .await
348 {
349 tracing::warn!(task_id = %tid, err = %e, "failed to write needs_review_refires to store");
350 }
351
352 // Approval only means the code looks good — the PR
353 // merge is the actual completion signal. Leave the
354 // task in InReview. The review_open_prs polling loop
355 // detects `batch_data.merged == true` on the next tick
356 // and transitions the task to Done correctly.
357 tracing::info!(
358 task_id = tid,
359 "approval received — leaving task in_review until PR is merged"
360 );
361 ReviewOutcome::Ok
362 }
363 Ok(ReviewDecision::Rerouted) => {
364 // Task was re-routed because there was no PR and no commits.
365 // Do NOT reset failure counters — no_code_reroutes must accumulate
366 // so the max_reroute_attempts safety limit can be reached.
367 ReviewOutcome::Ok
368 }
369 Ok(ReviewDecision::Skipped) => {
370 // Skipped indicates there's nothing to review (no PR etc.).
371 // Clear transient failure counters but preserve review_cycles.
372 crate::store::store_reset_failure_counters(
373 &Some(store_c.clone()),
374 &repo_s,
375 &tid,
376 )
377 .await;
378 if let Err(e) = crate::store::store_set_result(
379 &Some(store_c.clone()),
380 &repo_s,
381 &tid,
382 &[("needs_review_refires", serde_json::json!(0))],
383 )
384 .await
385 {
386 tracing::warn!(task_id = %tid, err = %e, "failed to write needs_review_refires to store");
387 }
388 ReviewOutcome::Ok
389 }
390 Ok(ReviewDecision::RequestChanges { .. }) => {
391 // handle_review_changes already incremented review_cycles —
392 // only reset transient per-attempt counters.
393 crate::store::store_reset_failure_counters(
394 &Some(store_c.clone()),
395 &repo_s,
396 &tid,
397 )
398 .await;
399 if let Err(e) = crate::store::store_set_result(
400 &Some(store_c.clone()),
401 &repo_s,
402 &tid,
403 &[("needs_review_refires", serde_json::json!(0))],
404 )
405 .await
406 {
407 tracing::warn!(task_id = %tid, err = %e, "failed to write needs_review_refires to store");
408 }
409 ReviewOutcome::Ok
410 }
411 };
412 crate::store::set_review_session_expected(
413 &store_c,
414 &repo_s,
415 &tid,
416 false,
417 )
418 .await;
419
420 // Freshness guard: re-fetch the task's current status before applying
421 // any outcome transitions. A concurrent path (e.g. PR merge, manual
422 // close, or another review cycle) may have already moved the task out
423 // of `in_review`. If so, discard the stale outcome rather than
424 // overwriting the newer state.
425 //
426 // IMPORTANT: Distinguish between DB errors and legitimate status changes.
427 // A DB lookup failure should NOT be treated as "task moved out of in_review"
428 // — we should retain the review outcome rather than discarding it on
429 // transient errors.
430 let fetched: Result<Option<(crate::store::TaskStatus, Option<String>)>, anyhow::Error> = async {
431 let id = match store_c.resolve_task_id(&repo_s, &tid).await {
432 Ok(Some(id)) => id,
433 Ok(None) => return Ok(None), // Task doesn't exist — treat as missing
434 Err(e) => return Err(e), // DB error — propagate, don't discard
435 };
436 let task = store_c.get(id).await?;
437 Ok(Some((task.status, task.block_reason)))
438 }
439 .await;
440
441 let (current_status, current_block_reason) = match fetched {
442 Ok(Some((status, block_reason))) => (Some(status), block_reason),
443 Ok(None) => {
444 // Task doesn't exist in store — treat as not in_review
445 (None, None)
446 }
447 Err(e) => {
448 // DB error — log and keep the review outcome instead of discarding
449 tracing::error!(
450 task_id = tid,
451 error = %e,
452 "review outcome retained — DB lookup failed, not treating as stale"
453 );
454 // Skip the stale outcome check and proceed with the outcome
455 // by treating it as if the task is still InReview
456 (Some(crate::store::TaskStatus::InReview), None)
457 }
458 };
459
460 if !matches!(current_status, Some(crate::store::TaskStatus::InReview)) {
461 let ci_blocked = matches!(
462 current_status,
463 Some(crate::store::TaskStatus::Blocked)
464 ) && current_block_reason
465 .as_deref()
466 .is_some_and(|r| r.contains("CI failure limit"));
467 let expected = ci_blocked
468 || matches!(
469 current_status,
470 Some(crate::store::TaskStatus::Done)
471 | Some(crate::store::TaskStatus::Routed)
472 | Some(crate::store::TaskStatus::New)
473 );
474 if expected {
475 tracing::debug!(
476 task_id = tid,
477 current_status = current_status.as_ref().map(|s| s.as_str()).unwrap_or("unknown"),
478 "review outcome discarded — task transitioned away from in_review (expected)"
479 );
480 } else {
481 tracing::warn!(
482 task_id = tid,
483 current_status = current_status.as_ref().map(|s| s.as_str()).unwrap_or("unknown"),
484 "review outcome discarded — task is no longer in_review (stale review agent result)"
485 );
486 }
487
488 // Kill the orphaned review session
489 let stale_session = tmux_c.session_name(&repo_s, &format!("{}-review", tid));
490 if let Err(e) = tmux_c.kill_session(&stale_session).await {
491 tracing::debug!(task_id = tid, error = %e, "failed to kill stale review session");
492 }
493
494 drop(permit);
495 return;
496 }
497
498 match outcome {
499 ReviewOutcome::Reset => {
500 // Kill any stale tmux review session before resetting — the
501 // session may still be alive if the agent hit a rate limit,
502 // if two concurrent reviewers raced and the loser's spawn
503 // failed with "duplicate session", or if review_and_merge
504 // returned an error before reaching its own cleanup.
505 // Killing a non-existent session is a harmless no-op.
506 let stale_session =
507 tmux_c.session_name(&repo_s, &format!("{}-review", tid));
508 tmux_c.kill_session(&stale_session).await.ok();
509 // Backoff before re-queuing to prevent rapid spin
510 // if the review agent keeps failing instantly.
511 tokio::time::sleep(std::time::Duration::from_secs(10)).await;
512 if let Err(e) = task_manager_c
513 .update_task_status(
514 &ExternalId(tid.clone()),
515 Status::NeedsReview,
516 )
517 .await
518 {
519 tracing::error!(task_id = %tid, err = %e, "update_task_status(NeedsReview) failed — task may be stuck in InReview");
520 }
521 }
522 ReviewOutcome::Reroute => {
523 // Merge conflict or similar — route back to the
524 // task agent so it can fix the issue.
525 let stale_session =
526 tmux_c.session_name(&repo_s, &format!("{}-review", tid));
527 tmux_c.kill_session(&stale_session).await.ok();
528 tracing::info!(
529 task_id = tid,
530 "re-routing to task agent to resolve merge conflict"
531 );
532 if let Err(e) = task_manager_c
533 .update_task_status(
534 &ExternalId(tid.clone()),
535 Status::Routed,
536 )
537 .await
538 {
539 tracing::error!(task_id = %tid, err = %e, "update_task_status(Routed) failed");
540 }
541 }
542 ReviewOutcome::RetryMerge => {
543 // PR was already approved; GitHub just hasn't computed
544 // mergeability yet. Keep the task in InReview and wait
545 // for review_poll to retry the merge — do NOT reset to
546 // NeedsReview, which would spawn a redundant review agent.
547 tracing::info!(
548 task_id = tid,
549 "mergeability pending — task stays in InReview, review_poll will retry merge"
550 );
551 // Brief pause so the next review_poll cycle fires soon.
552 tokio::time::sleep(std::time::Duration::from_secs(5)).await;
553 }
554 ReviewOutcome::RateLimited => {
555 // Kill stale tmux session — same as Reset.
556 let stale_session =
557 tmux_c.session_name(&repo_s, &format!("{}-review", tid));
558 tmux_c.kill_session(&stale_session).await.ok();
559
560 // Compute wait duration: if any agent is available right now,
561 // use a short backoff (30 s). If every agent is cooled, sleep
562 // until the earliest cooldown expires so we don't spin.
563 let wait_secs = {
564 let r = router_c.read().await;
565 if r.healthy_agent_count("review") > 0 {
566 30u64
567 } else {
568 let now = chrono::Utc::now().timestamp();
569 r.earliest_cooldown_until(Some("review"))
570 .map(|until| ((until - now).max(1)) as u64)
571 .unwrap_or(AGENT_COOLDOWN_SECS as u64)
572 }
573 };
574 tracing::info!(
575 task_id = tid,
576 wait_secs,
577 "deferring review retry due to agent rate limit cooldown"
578 );
579 tokio::time::sleep(std::time::Duration::from_secs(wait_secs)).await;
580 if let Err(e) = task_manager_c
581 .update_task_status(
582 &ExternalId(tid.clone()),
583 Status::NeedsReview,
584 )
585 .await
586 {
587 tracing::error!(task_id = %tid, err = %e, "update_task_status(NeedsReview) failed after rate limit backoff — task may be stuck in InReview");
588 }
589 }
590 ReviewOutcome::Block(reason) => {
591 // Persist block_reason BEFORE transitioning to Blocked to avoid
592 // a race where auto_unblock sees a blocked task without a reason
593 // and immediately unblocks it.
594 let fields = [
595 (
596 "block_reason",
597 serde_json::json!("review agent blocked — exceeded failure threshold"),
598 ),
599 ("last_error", serde_json::json!(reason)),
600 ];
601 if let Err(e) = task_manager_c
602 .update_task_status_and_result(
603 &ExternalId(tid.clone()),
604 Status::Blocked,
605 &fields,
606 )
607 .await
608 {
609 tracing::error!(task_id = %tid, err = %e, "update_task_status_and_result(Blocked) failed — task may be stuck in InReview");
610 }
611 }
612 ReviewOutcome::Ok => {}
613 }
614
615 drop(permit);
616 // _dispatch_guard dropped here — releases the per-task lock.
617 }));
618
619 // Small delay between dispatches to allow cooldowns from
620 // early failures to propagate before subsequent reviews are
621 // dispatched. This prevents batch dispatch of N reviews to
622 // the same rate-limited agent when events arrive concurrently.
623 tokio::time::sleep(std::time::Duration::from_secs(2)).await;
624 }
625 Ok(_) => {} // Not a needs_review event or different repo
626 Err(broadcast::error::RecvError::Lagged(n)) => {
627 tracing::debug!(n, "review subscriber lagged; sync catch-up will re-fire missed NeedsReview tasks");
628 }
629 Err(_) => break, // Channel closed
630 }
631 }
632 });
633}
634
635/// Classify a review failure and return the appropriate outcome.
636///
637/// Both the `Ok(ReviewDecision::Failed)` and `Err(e)` arms share the same
638/// logic; callers convert the error to a string before calling this helper.
639/// `context` is used in log messages to distinguish the two call sites
640/// ("review agent" vs "review_and_merge").
641async fn classify_review_failure(
642 store: &Arc<TaskStore>,
643 repo: &str,
644 task_id: &str,
645 reason: &str,
646 context: &str,
647) -> ReviewOutcome {
648 // Rate-limit and auth failures are not real review failures — don't count them
649 // toward MAX_REVIEW_AGENT_FAILURES. The cooldown system steers the next
650 // attempt to an available agent.
651 let lower_reason = reason.to_lowercase();
652
653 // Parse errors are orch's fault, not review feedback. Re-route for retry
654 // without counting toward the failure threshold, so parse errors don't
655 // incorrectly block tasks that have no actual review feedback.
656 if lower_reason.contains("parse error") {
657 tracing::warn!(
658 task_id,
659 reason,
660 "{context} hit parse error — re-routing for retry without counting as failure"
661 );
662 return ReviewOutcome::Reset;
663 }
664
665 // Session exited mid-thought with no result envelope (issue #3734). Same
666 // treatment as a parse error: retry without counting toward the failure cap
667 if lower_reason.contains("incomplete:") {
668 tracing::warn!(
669 task_id,
670 reason,
671 "{context} session incomplete — re-routing for retry without counting as failure"
672 );
673 return ReviewOutcome::Reset;
674 }
675
676 // The review output was cut off by the model's output/reasoning token
677 // budget, not a genuine parse/format failure. Same treatment as a parse
678 // error: re-route without counting toward MAX_REVIEW_AGENT_FAILURES.
679 if lower_reason.contains("truncated") {
680 tracing::warn!(
681 task_id,
682 reason,
683 "{context} output truncated by token budget — re-routing for retry without counting as failure"
684 );
685 return ReviewOutcome::Reset;
686 }
687
688 // Transient mergeability check: GitHub hasn't finished computing yet.
689 // This is a normal, expected transient state for freshly-pushed PRs.
690 // Return RetryMerge (not Reset) so the task stays in InReview — avoids
691 // spawning a redundant review agent that would just re-approve anyway.
692 if lower_reason.contains("not yet computed") {
693 tracing::warn!(
694 task_id,
695 reason,
696 "{context} PR mergeability check still pending — will retry merge without re-reviewing"
697 );
698 return ReviewOutcome::RetryMerge;
699 }
700
701 // Merge conflicts should be handled by handle_merge_conflict() in
702 // auto_merge.rs (which routes to the task agent). If a conflict error
703 // leaks here, route to the task agent instead of resetting to NeedsReview
704 // (which would re-spawn the review agent in a loop).
705 if lower_reason.contains("merge conflict")
706 || lower_reason.contains("not mergeable")
707 || lower_reason.contains("merge failed")
708 {
709 tracing::warn!(
710 task_id,
711 reason,
712 "{context} hit merge conflict — routing to task agent to resolve"
713 );
714 return ReviewOutcome::Reroute;
715 }
716
717 // No configured review model for any available agent, or the selected agent
718 // has no review model. This is a configuration/cooldown gap, not a review
719 // failure. Defer without incrementing the failure counter so the task retries
720 // once a review model is available.
721 if lower_reason.contains("no review agent") || lower_reason.contains("review model missing") {
722 tracing::warn!(
723 task_id,
724 reason,
725 "{context}: no review model available — deferring retry until one is available"
726 );
727 return ReviewOutcome::RateLimited;
728 }
729
730 if lower_reason.contains("rate limit") || lower_reason.contains("auth error") {
731 tracing::warn!(
732 task_id,
733 reason,
734 "{context} hit rate limit or auth error — deferring retry until cooldown expires"
735 );
736 return ReviewOutcome::RateLimited;
737 }
738
739 if crate::engine::runner::git_ops::is_transient_github_api_error(reason) {
740 tracing::warn!(
741 task_id,
742 reason,
743 "{context} hit transient GitHub error — deferring retry without counting as failure"
744 );
745 return ReviewOutcome::Reset;
746 }
747
748 // Provider-side network errors (ConnectionRefused, timeouts, etc.) are
749 // transient infrastructure issues, not review feedback. Re-route without
750 // counting toward MAX_REVIEW_AGENT_FAILURES.
751 if crate::engine::runner::agents::patterns::detect_network_error(reason).is_some() {
752 tracing::warn!(
753 task_id,
754 reason,
755 "{context} hit transient provider network error — re-routing for retry without counting as failure"
756 );
757 return ReviewOutcome::Reset;
758 }
759
760 let failures = match crate::store::store_increment(
761 &Some(store.clone()),
762 repo,
763 task_id,
764 "review_agent_failures",
765 )
766 .await
767 {
768 Ok(v) => v,
769 Err(e) => {
770 tracing::warn!(task_id, err = %e, "failed to increment review_agent_failures — skipping failure escalation this tick");
771 // Treat as 0 for now (no escalation)
772 0
773 }
774 };
775 let blocking = failures >= MAX_REVIEW_AGENT_FAILURES;
776 // Post failure comment to the PR so the history is visible.
777 post_review_failure_comment(store, repo, task_id, reason, failures, blocking).await;
778 if blocking {
779 tracing::error!(
780 task_id,
781 reason,
782 failures,
783 "{context} failed too many times — blocking task"
784 );
785 ReviewOutcome::Block(format!("{context} failed {failures} times: {reason}"))
786 } else {
787 tracing::error!(
788 task_id,
789 reason,
790 failures,
791 "{context} failed — resetting to NeedsReview for retry"
792 );
793 ReviewOutcome::Reset
794 }
795}
796
797/// Post a comment on the PR explaining why the review agent failed.
798/// Best-effort — if the PR number is missing or the API call fails, we just log.
799async fn post_review_failure_comment(
800 store: &Arc<TaskStore>,
801 repo: &str,
802 task_id: &str,
803 reason: &str,
804 failures: u64,
805 blocking: bool,
806) {
807 let pr_number = match opt_store_get_task(&Some(store.clone()), repo, task_id).await {
808 Some(t) => match t.pr_number {
809 Some(n) => n,
810 None => return, // No PR to comment on
811 },
812 _ => return, // No PR to comment on
813 };
814
815 let status = if blocking {
816 "Task blocked for human review."
817 } else {
818 "Retrying with a different agent."
819 };
820
821 let comment = format!(
822 "⚠️ Review agent failed (attempt {}/{})\n\n**Reason:** {}\n\n{}{}",
823 failures,
824 MAX_REVIEW_AGENT_FAILURES,
825 reason,
826 status,
827 crate::engine::orch_footer()
828 );
829
830 let gh = match GhHttp::new() {
831 Ok(gh) => gh,
832 Err(_) => return,
833 };
834 if let Err(e) = gh.add_comment(repo, &pr_number.to_string(), &comment).await {
835 tracing::debug!(
836 task_id,
837 pr_number,
838 error = %e,
839 "failed to post review failure comment"
840 );
841 }
842}
843
844#[cfg(test)]
845mod tests {
846 use super::*;
847
848 /// An incomplete mid-thought review (no result envelope) must reset for
849 /// retry without counting toward MAX_REVIEW_AGENT_FAILURES (issue #3734).
850 #[tokio::test]
851 async fn classify_review_failure_incomplete_resets_without_counting() {
852 let store = Arc::new(TaskStore::open_memory().await.unwrap());
853 let repo = "owner/repo";
854 let store_id = store
855 .create_internal(
856 repo,
857 "Incomplete review test",
858 "",
859 "review",
860 "review-3734",
861 None,
862 )
863 .await
864 .unwrap();
865 let task_id = format!("internal:{store_id}");
866
867 for _ in 0..(MAX_REVIEW_AGENT_FAILURES + 1) {
868 let outcome = classify_review_failure(
869 &store,
870 repo,
871 &task_id,
872 "incomplete: review session exited before producing a decision",
873 "review agent",
874 )
875 .await;
876 assert!(
877 matches!(outcome, ReviewOutcome::Reset),
878 "incomplete review failure must reset for retry, not block"
879 );
880 }
881
882 let task = opt_store_get_task(&Some(store.clone()), repo, &task_id)
883 .await
884 .expect("task should exist");
885 assert_eq!(
886 task.review_agent_failures, 0,
887 "incomplete failures must not count toward review_agent_failures"
888 );
889 }
890
891 /// A truncated review output (opencode step_finish reason=length) must be
892 /// treated like a parse error: reset for retry without counting toward
893 /// MAX_REVIEW_AGENT_FAILURES, so a run of token-budget flukes never
894 /// permanently blocks the task (issue #3473).
895 #[tokio::test]
896 async fn classify_review_failure_truncated_resets_without_counting() {
897 let store = Arc::new(TaskStore::open_memory().await.unwrap());
898 let repo = "owner/repo";
899 let store_id = store
900 .create_internal(
901 repo,
902 "Truncated review test",
903 "",
904 "review",
905 "review-1",
906 None,
907 )
908 .await
909 .unwrap();
910 let task_id = format!("internal:{store_id}");
911
912 for _ in 0..(MAX_REVIEW_AGENT_FAILURES + 1) {
913 let outcome = classify_review_failure(
914 &store,
915 repo,
916 &task_id,
917 "truncated: review output was cut off before completion \
918 (output/reasoning token budget exceeded)",
919 "review agent",
920 )
921 .await;
922 assert!(
923 matches!(outcome, ReviewOutcome::Reset),
924 "truncated review failure must reset for retry, not block"
925 );
926 }
927
928 let task = opt_store_get_task(&Some(store.clone()), repo, &task_id)
929 .await
930 .expect("task should exist");
931 assert_eq!(
932 task.review_agent_failures, 0,
933 "truncated failures must not count toward review_agent_failures"
934 );
935 }
936
937 /// Transient provider network errors (ConnectionRefused, timeouts) must be
938 /// treated like parse errors: reset for retry without counting toward
939 /// MAX_REVIEW_AGENT_FAILURES (issue #3482).
940 #[tokio::test]
941 async fn classify_review_failure_network_error_resets_without_counting() {
942 let store = Arc::new(TaskStore::open_memory().await.unwrap());
943 let repo = "owner/repo";
944 let store_id = store
945 .create_internal(
946 repo,
947 "Network error review test",
948 "",
949 "review",
950 "review-2",
951 None,
952 )
953 .await
954 .unwrap();
955 let task_id = format!("internal:{store_id}");
956
957 let error_messages = [
958 "API Error: Unable to connect to API (ConnectionRefused)",
959 "connection refused",
960 "ECONNREFUSED",
961 "Upstream idle timeout exceeded",
962 "connect ETIMEDOUT",
963 "network request failed",
964 ];
965
966 for msg in &error_messages {
967 let outcome =
968 classify_review_failure(&store, repo, &task_id, msg, "review agent").await;
969 assert!(
970 matches!(outcome, ReviewOutcome::Reset),
971 "network error '{msg}' must reset for retry, not block"
972 );
973 }
974
975 let task = opt_store_get_task(&Some(store.clone()), repo, &task_id)
976 .await
977 .expect("task should exist");
978 assert_eq!(
979 task.review_agent_failures, 0,
980 "network errors must not count toward review_agent_failures"
981 );
982 }
983
984 /// When no agent has a configured review model (or the fallback agent has none),
985 /// the review must be deferred, not counted as a review failure. Counting it
986 /// would eventually block tasks for a configuration/cooldown gap (issue #3584).
987 #[tokio::test]
988 async fn classify_review_failure_missing_review_model_is_rate_limited() {
989 let store = Arc::new(TaskStore::open_memory().await.unwrap());
990 let repo = "owner/repo";
991 let store_id = store
992 .create_internal(
993 repo,
994 "Missing review model test",
995 "",
996 "review",
997 "review-3",
998 None,
999 )
1000 .await
1001 .unwrap();
1002 let task_id = format!("internal:{store_id}");
1003
1004 let error_messages = [
1005 "no review agent with a configured review model is available for task internal:123",
1006 "review model missing for task internal:123 after agent selection — refusing to dispatch",
1007 ];
1008
1009 for _ in 0..(MAX_REVIEW_AGENT_FAILURES + 1) {
1010 for msg in &error_messages {
1011 let outcome =
1012 classify_review_failure(&store, repo, &task_id, msg, "review_and_merge").await;
1013 assert!(
1014 matches!(outcome, ReviewOutcome::RateLimited),
1015 "missing review model error '{msg}' must defer, not block"
1016 );
1017 }
1018 }
1019
1020 let task = opt_store_get_task(&Some(store.clone()), repo, &task_id)
1021 .await
1022 .expect("task should exist");
1023 assert_eq!(
1024 task.review_agent_failures, 0,
1025 "missing review model errors must not count toward review_agent_failures"
1026 );
1027 }
1028}