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 // The review output was cut off by the model's output/reasoning token
666 // budget, not a genuine parse/format failure. Same treatment as a parse
667 // error: re-route without counting toward MAX_REVIEW_AGENT_FAILURES.
668 if lower_reason.contains("truncated") {
669 tracing::warn!(
670 task_id,
671 reason,
672 "{context} output truncated by token budget — re-routing for retry without counting as failure"
673 );
674 return ReviewOutcome::Reset;
675 }
676
677 // Transient mergeability check: GitHub hasn't finished computing yet.
678 // This is a normal, expected transient state for freshly-pushed PRs.
679 // Return RetryMerge (not Reset) so the task stays in InReview — avoids
680 // spawning a redundant review agent that would just re-approve anyway.
681 if lower_reason.contains("not yet computed") {
682 tracing::warn!(
683 task_id,
684 reason,
685 "{context} PR mergeability check still pending — will retry merge without re-reviewing"
686 );
687 return ReviewOutcome::RetryMerge;
688 }
689
690 // Merge conflicts should be handled by handle_merge_conflict() in
691 // auto_merge.rs (which routes to the task agent). If a conflict error
692 // leaks here, route to the task agent instead of resetting to NeedsReview
693 // (which would re-spawn the review agent in a loop).
694 if lower_reason.contains("merge conflict")
695 || lower_reason.contains("not mergeable")
696 || lower_reason.contains("merge failed")
697 {
698 tracing::warn!(
699 task_id,
700 reason,
701 "{context} hit merge conflict — routing to task agent to resolve"
702 );
703 return ReviewOutcome::Reroute;
704 }
705
706 // No configured review model for any available agent, or the selected agent
707 // has no review model. This is a configuration/cooldown gap, not a review
708 // failure. Defer without incrementing the failure counter so the task retries
709 // once a review model is available.
710 if lower_reason.contains("no review agent") || lower_reason.contains("review model missing") {
711 tracing::warn!(
712 task_id,
713 reason,
714 "{context}: no review model available — deferring retry until one is available"
715 );
716 return ReviewOutcome::RateLimited;
717 }
718
719 if lower_reason.contains("rate limit") || lower_reason.contains("auth error") {
720 tracing::warn!(
721 task_id,
722 reason,
723 "{context} hit rate limit or auth error — deferring retry until cooldown expires"
724 );
725 return ReviewOutcome::RateLimited;
726 }
727
728 if crate::engine::runner::git_ops::is_transient_github_api_error(reason) {
729 tracing::warn!(
730 task_id,
731 reason,
732 "{context} hit transient GitHub error — deferring retry without counting as failure"
733 );
734 return ReviewOutcome::Reset;
735 }
736
737 // Provider-side network errors (ConnectionRefused, timeouts, etc.) are
738 // transient infrastructure issues, not review feedback. Re-route without
739 // counting toward MAX_REVIEW_AGENT_FAILURES.
740 if crate::engine::runner::agents::patterns::detect_network_error(reason).is_some() {
741 tracing::warn!(
742 task_id,
743 reason,
744 "{context} hit transient provider network error — re-routing for retry without counting as failure"
745 );
746 return ReviewOutcome::Reset;
747 }
748
749 let failures = match crate::store::store_increment(
750 &Some(store.clone()),
751 repo,
752 task_id,
753 "review_agent_failures",
754 )
755 .await
756 {
757 Ok(v) => v,
758 Err(e) => {
759 tracing::warn!(task_id, err = %e, "failed to increment review_agent_failures — skipping failure escalation this tick");
760 // Treat as 0 for now (no escalation)
761 0
762 }
763 };
764 let blocking = failures >= MAX_REVIEW_AGENT_FAILURES;
765 // Post failure comment to the PR so the history is visible.
766 post_review_failure_comment(store, repo, task_id, reason, failures, blocking).await;
767 if blocking {
768 tracing::error!(
769 task_id,
770 reason,
771 failures,
772 "{context} failed too many times — blocking task"
773 );
774 ReviewOutcome::Block(format!("{context} failed {failures} times: {reason}"))
775 } else {
776 tracing::error!(
777 task_id,
778 reason,
779 failures,
780 "{context} failed — resetting to NeedsReview for retry"
781 );
782 ReviewOutcome::Reset
783 }
784}
785
786/// Post a comment on the PR explaining why the review agent failed.
787/// Best-effort — if the PR number is missing or the API call fails, we just log.
788async fn post_review_failure_comment(
789 store: &Arc<TaskStore>,
790 repo: &str,
791 task_id: &str,
792 reason: &str,
793 failures: u64,
794 blocking: bool,
795) {
796 let pr_number = match opt_store_get_task(&Some(store.clone()), repo, task_id).await {
797 Some(t) => match t.pr_number {
798 Some(n) => n,
799 None => return, // No PR to comment on
800 },
801 _ => return, // No PR to comment on
802 };
803
804 let status = if blocking {
805 "Task blocked for human review."
806 } else {
807 "Retrying with a different agent."
808 };
809
810 let comment = format!(
811 "⚠️ Review agent failed (attempt {}/{})\n\n**Reason:** {}\n\n{}{}",
812 failures,
813 MAX_REVIEW_AGENT_FAILURES,
814 reason,
815 status,
816 crate::engine::orch_footer()
817 );
818
819 let gh = match GhHttp::new() {
820 Ok(gh) => gh,
821 Err(_) => return,
822 };
823 if let Err(e) = gh.add_comment(repo, &pr_number.to_string(), &comment).await {
824 tracing::debug!(
825 task_id,
826 pr_number,
827 error = %e,
828 "failed to post review failure comment"
829 );
830 }
831}
832
833#[cfg(test)]
834mod tests {
835 use super::*;
836
837 /// A truncated review output (opencode step_finish reason=length) must be
838 /// treated like a parse error: reset for retry without counting toward
839 /// MAX_REVIEW_AGENT_FAILURES, so a run of token-budget flukes never
840 /// permanently blocks the task (issue #3473).
841 #[tokio::test]
842 async fn classify_review_failure_truncated_resets_without_counting() {
843 let store = Arc::new(TaskStore::open_memory().await.unwrap());
844 let repo = "owner/repo";
845 let store_id = store
846 .create_internal(
847 repo,
848 "Truncated review test",
849 "",
850 "review",
851 "review-1",
852 None,
853 )
854 .await
855 .unwrap();
856 let task_id = format!("internal:{store_id}");
857
858 for _ in 0..(MAX_REVIEW_AGENT_FAILURES + 1) {
859 let outcome = classify_review_failure(
860 &store,
861 repo,
862 &task_id,
863 "truncated: review output was cut off before completion \
864 (output/reasoning token budget exceeded)",
865 "review agent",
866 )
867 .await;
868 assert!(
869 matches!(outcome, ReviewOutcome::Reset),
870 "truncated review failure must reset for retry, not block"
871 );
872 }
873
874 let task = opt_store_get_task(&Some(store.clone()), repo, &task_id)
875 .await
876 .expect("task should exist");
877 assert_eq!(
878 task.review_agent_failures, 0,
879 "truncated failures must not count toward review_agent_failures"
880 );
881 }
882
883 /// Transient provider network errors (ConnectionRefused, timeouts) must be
884 /// treated like parse errors: reset for retry without counting toward
885 /// MAX_REVIEW_AGENT_FAILURES (issue #3482).
886 #[tokio::test]
887 async fn classify_review_failure_network_error_resets_without_counting() {
888 let store = Arc::new(TaskStore::open_memory().await.unwrap());
889 let repo = "owner/repo";
890 let store_id = store
891 .create_internal(
892 repo,
893 "Network error review test",
894 "",
895 "review",
896 "review-2",
897 None,
898 )
899 .await
900 .unwrap();
901 let task_id = format!("internal:{store_id}");
902
903 let error_messages = [
904 "API Error: Unable to connect to API (ConnectionRefused)",
905 "connection refused",
906 "ECONNREFUSED",
907 "Upstream idle timeout exceeded",
908 "connect ETIMEDOUT",
909 "network request failed",
910 ];
911
912 for msg in &error_messages {
913 let outcome =
914 classify_review_failure(&store, repo, &task_id, msg, "review agent").await;
915 assert!(
916 matches!(outcome, ReviewOutcome::Reset),
917 "network error '{msg}' must reset for retry, not block"
918 );
919 }
920
921 let task = opt_store_get_task(&Some(store.clone()), repo, &task_id)
922 .await
923 .expect("task should exist");
924 assert_eq!(
925 task.review_agent_failures, 0,
926 "network errors must not count toward review_agent_failures"
927 );
928 }
929
930 /// When no agent has a configured review model (or the fallback agent has none),
931 /// the review must be deferred, not counted as a review failure. Counting it
932 /// would eventually block tasks for a configuration/cooldown gap (issue #3584).
933 #[tokio::test]
934 async fn classify_review_failure_missing_review_model_is_rate_limited() {
935 let store = Arc::new(TaskStore::open_memory().await.unwrap());
936 let repo = "owner/repo";
937 let store_id = store
938 .create_internal(
939 repo,
940 "Missing review model test",
941 "",
942 "review",
943 "review-3",
944 None,
945 )
946 .await
947 .unwrap();
948 let task_id = format!("internal:{store_id}");
949
950 let error_messages = [
951 "no review agent with a configured review model is available for task internal:123",
952 "review model missing for task internal:123 after agent selection — refusing to dispatch",
953 ];
954
955 for _ in 0..(MAX_REVIEW_AGENT_FAILURES + 1) {
956 for msg in &error_messages {
957 let outcome =
958 classify_review_failure(&store, repo, &task_id, msg, "review_and_merge").await;
959 assert!(
960 matches!(outcome, ReviewOutcome::RateLimited),
961 "missing review model error '{msg}' must defer, not block"
962 );
963 }
964 }
965
966 let task = opt_store_get_task(&Some(store.clone()), repo, &task_id)
967 .await
968 .expect("task should exist");
969 assert_eq!(
970 task.review_agent_failures, 0,
971 "missing review model errors must not count toward review_agent_failures"
972 );
973 }
974}