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 /// The PR was approved but the merge did not land for a reason outside the
23 /// review (mergeability not computed yet, rate limit, transient GitHub error).
24 /// Keep the task in InReview so `review_open_prs` retries the merge, instead of
25 /// resetting to NeedsReview (which spawns a redundant review agent)
26 RetryMerge,
27 /// Re-route to the task agent (e.g. merge conflicts that need the agent to fix).
28 Reroute,
29 Block(String),
30 Ok,
31}
32
33/// Spawn a task that listens for NeedsReview events and triggers the review agent.
34///
35/// This is the sole trigger for NeedsReview → InReview transitions. The sync tick
36/// no longer has a competing NeedsReview loop (removed in fix for issue #857 — both
37/// paths firing simultaneously caused double failure-counter increments).
38#[allow(clippy::too_many_arguments)]
39pub fn spawn(
40 mut rx: broadcast::Receiver<TaskEvent>,
41 backend: Arc<dyn ExternalBackend>,
42 tmux: Arc<TmuxManager>,
43 semaphore: Arc<Semaphore>,
44 task_manager: Arc<TaskManager>,
45 router: Arc<RwLock<Router>>,
46 dispatching: Arc<DashMap<String, String>>,
47 store: Arc<TaskStore>,
48 repo: String,
49) {
50 tokio::spawn(async move {
51 loop {
52 match rx.recv().await {
53 Ok(event) if event.new_status == "needs_review" && event.repo == repo => {
54 // Check if review agent is enabled (read fresh each time).
55 let enable_review = crate::config::get("workflow.enable_review_agent")
56 .map(|v| v != "false")
57 .unwrap_or(true);
58 if !enable_review {
59 continue;
60 }
61
62 let task_id = &event.task_id;
63 let dispatch_key = format!("{}/{}", repo, task_id);
64
65 // Atomically claim the task before any async work so concurrent
66 // review events cannot double-spawn the same review flow.
67 // Using DashMap lets us distinguish same-task duplicates (expected) from
68 // key collisions involving a different task (should never happen).
69 {
70 use dashmap::mapref::entry::Entry;
71 match dispatching.entry(dispatch_key.clone()) {
72 Entry::Occupied(existing) => {
73 let existing_id = existing.get().clone();
74 drop(existing);
75 if existing_id == *task_id {
76 tracing::debug!(
77 task_id,
78 "task locked by dispatch flow, skipping event-driven review"
79 );
80 } else {
81 tracing::warn!(
82 task_id,
83 existing_task_id = existing_id,
84 dispatch_key,
85 "dispatch key collision: unexpected task already holds this key"
86 );
87 }
88 continue;
89 }
90 Entry::Vacant(slot) => {
91 slot.insert(task_id.clone());
92 }
93 }
94 }
95
96 // Create guard IMMEDIATELY after inserting the key — owns removal for all subsequent exit paths
97 let dispatch_guard =
98 DispatchGuard::new(dispatching.clone(), dispatch_key.clone());
99
100 tracing::info!(task_id, "event-driven review triggered");
101
102 // Look up the task from the store to get an ExternalTask.
103 let task = {
104 if crate::engine::tasks::is_internal_id(task_id)
105 || match store.has_external_tasks(&repo).await {
106 Ok(has_external) => has_external,
107 Err(e) => {
108 tracing::warn!(
109 repo = %repo,
110 task_id,
111 error = %e,
112 "failed to check external task sentinel; keeping store-first review lookup"
113 );
114 true
115 }
116 }
117 {
118 // O(1) indexed lookup — resolve task_id to its SQLite row id,
119 // then fetch directly instead of scanning the whole needs_review set.
120 let result = if let Some(numeric_id) =
121 crate::engine::tasks::parse_internal_id(task_id)
122 {
123 store.get(numeric_id).await.map(Some)
124 } else {
125 store.get_by_external_id(&repo, task_id).await
126 };
127 match result {
128 Ok(Some(t))
129 if t.status == crate::store::TaskStatus::NeedsReview =>
130 {
131 Some(crate::engine::tasks::store_task_to_external(&t))
132 }
133 Ok(Some(t)) => {
134 tracing::debug!(
135 task_id,
136 status = t.status.as_str(),
137 "task found but not in needs_review status, skipping"
138 );
139 None
140 }
141 Ok(None) => None,
142 Err(e) => {
143 tracing::warn!(task_id, err = %e, "store lookup failed for review");
144 None
145 }
146 }
147 } else {
148 // Fall back to backend
149 match backend.list_by_status(Status::NeedsReview).await {
150 Ok(tasks) => tasks.into_iter().find(|t| t.id.0 == *task_id),
151 Err(e) => {
152 tracing::warn!(task_id, err = %e, "backend lookup failed for review");
153 None
154 }
155 }
156 }
157 };
158
159 let Some(task) = task else {
160 tracing::debug!(
161 task_id,
162 "task not found in needs_review list (may have already transitioned)"
163 );
164 continue;
165 };
166
167 if let Some(stored) =
168 opt_store_get_task(&Some(store.clone()), &repo, &task.id.0).await
169 {
170 let no_code_review = stored.worktree.trim().is_empty()
171 && stored.branch.trim().is_empty()
172 && stored.pr_number.is_none();
173 if no_code_review {
174 tracing::info!(
175 task_id,
176 "no worktree/branch/pr for needs_review task — skipping review dispatch and marking done"
177 );
178 crate::store::store_reset_failure_counters(
179 &Some(store.clone()),
180 &repo,
181 &task.id.0,
182 )
183 .await;
184 if let Err(e) = crate::store::store_set_result(
185 &Some(store.clone()),
186 &repo,
187 &task.id.0,
188 &[("needs_review_refires", serde_json::json!(0))],
189 )
190 .await
191 {
192 tracing::warn!(
193 task_id,
194 err = %e,
195 "failed to reset needs_review_refires while skipping no-code review"
196 );
197 }
198 if let Err(e) = task_manager
199 .update_task_status(&task.id, Status::Done)
200 .await
201 {
202 tracing::error!(
203 task_id,
204 err = %e,
205 "failed to mark no-code needs_review task as done"
206 );
207 }
208 continue;
209 }
210 }
211
212 // If no agents are currently routable (all cooled, degraded, or without
213 // an available review model), skip this event and let sync re-fire when
214 // an agent is available again. Sleeping inline here would block every
215 // subsequent NeedsReview event for this repo for the duration of the
216 // cooldown (potentially hours/days), causing sync's rebroadcast loop to
217 // escalate piled-up tasks to Blocked.
218 {
219 let router_guard = router.read().await;
220 let no_routable = !router_guard.available_agents.is_empty()
221 && router_guard.healthy_agent_count("review") == 0;
222 drop(router_guard);
223 if no_routable {
224 tracing::info!(
225 task_id,
226 "no routable review agents — leaving task in needs_review for sync to re-fire"
227 );
228 continue;
229 }
230 }
231
232 // Try to acquire a semaphore permit.
233 let permit = match semaphore.clone().try_acquire_owned() {
234 Ok(p) => p,
235 Err(_) => {
236 tracing::debug!(task_id, "all parallel slots busy; sync catch-up will re-fire when a slot opens");
237 continue;
238 }
239 };
240
241 // Transition to InReview — this IS the atomic guard against duplicates.
242 if let Err(e) = task_manager
243 .update_task_status(&task.id, Status::InReview)
244 .await
245 {
246 tracing::warn!(task_id, err = %e, "failed to transition to InReview");
247 drop(permit);
248 continue;
249 }
250
251 crate::store::set_review_session_expected(&store, &repo, &task.id.0, true)
252 .await;
253
254 let backend_c = backend.clone();
255 let task_manager_c = task_manager.clone();
256 let tmux_c = tmux.clone();
257 let router_c = router.clone();
258 let store_c = store.clone();
259 let repo_s = repo.clone();
260 tokio::spawn(REPO_CONTEXT.scope(repo_s.clone(), async move {
261 let _dispatch_guard = dispatch_guard; // released on drop (normal or panic)
262 let tid = task.id.0.clone();
263
264 // Re-check agent routability inside the spawned task. A concurrent
265 // review that ran between the outer dispatch check and now may have
266 // set a cooldown — bail early rather than burning a review run
267 // against a rate-limited agent.
268 let pre_check_all_cooled = {
269 let r = router_c.read().await;
270 !r.available_agents.is_empty() && r.healthy_agent_count("review") == 0
271 };
272 if pre_check_all_cooled {
273 tracing::info!(
274 task_id = tid,
275 "all agents cooled at review spawn time — resetting to needs_review for sync to re-fire"
276 );
277 crate::store::set_review_session_expected(
278 &store_c,
279 &repo_s,
280 &tid,
281 false,
282 )
283 .await;
284 if let Err(e) = task_manager_c
285 .update_task_status(
286 &ExternalId(tid.clone()),
287 Status::NeedsReview,
288 )
289 .await
290 {
291 tracing::error!(task_id = %tid, err = %e, "pre-check: update_task_status(NeedsReview) failed — task may be stuck in InReview");
292 }
293 drop(permit);
294 return;
295 }
296
297 let outcome = match review_and_merge(
298 &task,
299 &backend_c,
300 &tmux_c,
301 &repo_s,
302 &router_c,
303 &task_manager_c,
304 &store_c,
305 )
306 .await
307 {
308 Ok(ReviewDecision::Blocked(reason)) => {
309 tracing::error!(
310 task_id = tid,
311 reason,
312 "review gate blocked after repeated failures — marking task blocked"
313 );
314 ReviewOutcome::Block(format!(
315 "review gate blocked after repeated failures: {reason}"
316 ))
317 }
318 Ok(ReviewDecision::Failed(reason)) => {
319 classify_review_failure(
320 &store_c, &repo_s, &tid, &reason, "review agent",
321 )
322 .await
323 }
324 Err(e) => {
325 let reason = format!("{e:#}");
326 classify_review_failure(
327 &store_c, &repo_s, &tid, &reason, "review_and_merge",
328 )
329 .await
330 }
331 Ok(ReviewDecision::Approve) => {
332 // On approval we clear transient failure counters
333 // but preserve the `review_cycles` counter which
334 // tracks how many times the PR has requested
335 // changes and is used as a circuit-breaker.
336 crate::store::store_reset_failure_counters(
337 &Some(store_c.clone()),
338 &repo_s,
339 &tid,
340 )
341 .await;
342 // Also reset needs_review_refires so the sync backoff doesn't persist across a successful review
343 if let Err(e) = crate::store::store_set_result(
344 &Some(store_c.clone()),
345 &repo_s,
346 &tid,
347 &[("needs_review_refires", serde_json::json!(0))],
348 )
349 .await
350 {
351 tracing::warn!(task_id = %tid, err = %e, "failed to write needs_review_refires to store");
352 }
353
354 // Approval only means the code looks good — the PR
355 // merge is the actual completion signal. Leave the
356 // task in InReview. The review_open_prs polling loop
357 // detects `batch_data.merged == true` on the next tick
358 // and transitions the task to Done correctly.
359 tracing::info!(
360 task_id = tid,
361 "approval received — leaving task in_review until PR is merged"
362 );
363 ReviewOutcome::Ok
364 }
365 Ok(ReviewDecision::Rerouted) => {
366 // Task was re-routed because there was no PR and no commits.
367 // Do NOT reset failure counters — no_code_reroutes must accumulate
368 // so the max_reroute_attempts safety limit can be reached.
369 ReviewOutcome::Ok
370 }
371 Ok(ReviewDecision::Skipped) => {
372 // Skipped indicates there's nothing to review (no PR etc.).
373 // Clear transient failure counters but preserve review_cycles.
374 crate::store::store_reset_failure_counters(
375 &Some(store_c.clone()),
376 &repo_s,
377 &tid,
378 )
379 .await;
380 if let Err(e) = crate::store::store_set_result(
381 &Some(store_c.clone()),
382 &repo_s,
383 &tid,
384 &[("needs_review_refires", serde_json::json!(0))],
385 )
386 .await
387 {
388 tracing::warn!(task_id = %tid, err = %e, "failed to write needs_review_refires to store");
389 }
390 ReviewOutcome::Ok
391 }
392 Ok(ReviewDecision::RequestChanges { .. }) => {
393 // handle_review_changes already incremented review_cycles —
394 // only reset transient per-attempt counters.
395 crate::store::store_reset_failure_counters(
396 &Some(store_c.clone()),
397 &repo_s,
398 &tid,
399 )
400 .await;
401 if let Err(e) = crate::store::store_set_result(
402 &Some(store_c.clone()),
403 &repo_s,
404 &tid,
405 &[("needs_review_refires", serde_json::json!(0))],
406 )
407 .await
408 {
409 tracing::warn!(task_id = %tid, err = %e, "failed to write needs_review_refires to store");
410 }
411 ReviewOutcome::Ok
412 }
413 };
414 crate::store::set_review_session_expected(
415 &store_c,
416 &repo_s,
417 &tid,
418 false,
419 )
420 .await;
421
422 // Freshness guard: re-fetch the task's current status before applying
423 // any outcome transitions. A concurrent path (e.g. PR merge, manual
424 // close, or another review cycle) may have already moved the task out
425 // of `in_review`. If so, discard the stale outcome rather than
426 // overwriting the newer state.
427 //
428 // IMPORTANT: Distinguish between DB errors and legitimate status changes.
429 // A DB lookup failure should NOT be treated as "task moved out of in_review"
430 // — we should retain the review outcome rather than discarding it on
431 // transient errors.
432 let fetched: Result<Option<(crate::store::TaskStatus, Option<String>)>, anyhow::Error> = async {
433 let id = match store_c.resolve_task_id(&repo_s, &tid).await {
434 Ok(Some(id)) => id,
435 Ok(None) => return Ok(None), // Task doesn't exist — treat as missing
436 Err(e) => return Err(e), // DB error — propagate, don't discard
437 };
438 let task = store_c.get(id).await?;
439 Ok(Some((task.status, task.block_reason)))
440 }
441 .await;
442
443 let (current_status, current_block_reason) = match fetched {
444 Ok(Some((status, block_reason))) => (Some(status), block_reason),
445 Ok(None) => {
446 // Task doesn't exist in store — treat as not in_review
447 (None, None)
448 }
449 Err(e) => {
450 // DB error — log and keep the review outcome instead of discarding
451 tracing::error!(
452 task_id = tid,
453 error = %e,
454 "review outcome retained — DB lookup failed, not treating as stale"
455 );
456 // Skip the stale outcome check and proceed with the outcome
457 // by treating it as if the task is still InReview
458 (Some(crate::store::TaskStatus::InReview), None)
459 }
460 };
461
462 if !matches!(current_status, Some(crate::store::TaskStatus::InReview)) {
463 let ci_blocked = matches!(
464 current_status,
465 Some(crate::store::TaskStatus::Blocked)
466 ) && current_block_reason
467 .as_deref()
468 .is_some_and(|r| r.contains("CI failure limit"));
469 let expected = ci_blocked
470 || matches!(
471 current_status,
472 Some(crate::store::TaskStatus::Done)
473 | Some(crate::store::TaskStatus::Routed)
474 | Some(crate::store::TaskStatus::New)
475 );
476 if expected {
477 tracing::debug!(
478 task_id = tid,
479 current_status = current_status.as_ref().map(|s| s.as_str()).unwrap_or("unknown"),
480 "review outcome discarded — task transitioned away from in_review (expected)"
481 );
482 } else {
483 tracing::warn!(
484 task_id = tid,
485 current_status = current_status.as_ref().map(|s| s.as_str()).unwrap_or("unknown"),
486 "review outcome discarded — task is no longer in_review (stale review agent result)"
487 );
488 }
489
490 // Kill the orphaned review session
491 let stale_session = tmux_c.session_name(&repo_s, &format!("{}-review", tid));
492 if let Err(e) = tmux_c.kill_session(&stale_session).await {
493 tracing::debug!(task_id = tid, error = %e, "failed to kill stale review session");
494 }
495
496 drop(permit);
497 return;
498 }
499
500 match outcome {
501 ReviewOutcome::Reset => {
502 // Kill any stale tmux review session before resetting — the
503 // session may still be alive if the agent hit a rate limit,
504 // if two concurrent reviewers raced and the loser's spawn
505 // failed with "duplicate session", or if review_and_merge
506 // returned an error before reaching its own cleanup.
507 // Killing a non-existent session is a harmless no-op.
508 let stale_session =
509 tmux_c.session_name(&repo_s, &format!("{}-review", tid));
510 tmux_c.kill_session(&stale_session).await.ok();
511 // Backoff before re-queuing to prevent rapid spin
512 // if the review agent keeps failing instantly.
513 tokio::time::sleep(std::time::Duration::from_secs(10)).await;
514 if let Err(e) = task_manager_c
515 .update_task_status(
516 &ExternalId(tid.clone()),
517 Status::NeedsReview,
518 )
519 .await
520 {
521 tracing::error!(task_id = %tid, err = %e, "update_task_status(NeedsReview) failed — task may be stuck in InReview");
522 }
523 }
524 ReviewOutcome::Reroute => {
525 // Merge conflict or similar — route back to the
526 // task agent so it can fix the issue.
527 let stale_session =
528 tmux_c.session_name(&repo_s, &format!("{}-review", tid));
529 tmux_c.kill_session(&stale_session).await.ok();
530 tracing::info!(
531 task_id = tid,
532 "re-routing to task agent to resolve merge conflict"
533 );
534 if let Err(e) = task_manager_c
535 .update_task_status(
536 &ExternalId(tid.clone()),
537 Status::Routed,
538 )
539 .await
540 {
541 tracing::error!(task_id = %tid, err = %e, "update_task_status(Routed) failed");
542 }
543 }
544 ReviewOutcome::RetryMerge => {
545 // The PR is already approved and the merge is the only
546 // step left. Keep the task in InReview and wait for
547 // review_poll to retry the merge — do NOT reset to
548 // NeedsReview, which would spawn a redundant review agent.
549 tracing::info!(
550 task_id = tid,
551 "merge pending — task stays in InReview, review_poll will retry merge"
552 );
553 // Brief pause so the next review_poll cycle fires soon.
554 tokio::time::sleep(std::time::Duration::from_secs(5)).await;
555 }
556 ReviewOutcome::RateLimited => {
557 // Kill stale tmux session — same as Reset.
558 let stale_session =
559 tmux_c.session_name(&repo_s, &format!("{}-review", tid));
560 tmux_c.kill_session(&stale_session).await.ok();
561
562 // Compute wait duration: if any agent is available right now,
563 // use a short backoff (30 s). If every agent is cooled, sleep
564 // until the earliest cooldown expires so we don't spin.
565 let wait_secs = {
566 let r = router_c.read().await;
567 if r.healthy_agent_count("review") > 0 {
568 30u64
569 } else {
570 let now = chrono::Utc::now().timestamp();
571 r.earliest_cooldown_until(Some("review"))
572 .map(|until| ((until - now).max(1)) as u64)
573 .unwrap_or(AGENT_COOLDOWN_SECS as u64)
574 }
575 };
576 tracing::info!(
577 task_id = tid,
578 wait_secs,
579 "deferring review retry due to agent rate limit cooldown"
580 );
581 tokio::time::sleep(std::time::Duration::from_secs(wait_secs)).await;
582 if let Err(e) = task_manager_c
583 .update_task_status(
584 &ExternalId(tid.clone()),
585 Status::NeedsReview,
586 )
587 .await
588 {
589 tracing::error!(task_id = %tid, err = %e, "update_task_status(NeedsReview) failed after rate limit backoff — task may be stuck in InReview");
590 }
591 }
592 ReviewOutcome::Block(reason) => {
593 // Persist block_reason BEFORE transitioning to Blocked to avoid
594 // a race where auto_unblock sees a blocked task without a reason
595 // and immediately unblocks it.
596 let fields = [
597 (
598 "block_reason",
599 serde_json::json!("review agent blocked — exceeded failure threshold"),
600 ),
601 ("last_error", serde_json::json!(reason)),
602 ];
603 if let Err(e) = task_manager_c
604 .update_task_status_and_result(
605 &ExternalId(tid.clone()),
606 Status::Blocked,
607 &fields,
608 )
609 .await
610 {
611 tracing::error!(task_id = %tid, err = %e, "update_task_status_and_result(Blocked) failed — task may be stuck in InReview");
612 }
613 }
614 ReviewOutcome::Ok => {}
615 }
616
617 drop(permit);
618 // _dispatch_guard dropped here — releases the per-task lock.
619 }));
620
621 // Small delay between dispatches to allow cooldowns from
622 // early failures to propagate before subsequent reviews are
623 // dispatched. This prevents batch dispatch of N reviews to
624 // the same rate-limited agent when events arrive concurrently.
625 tokio::time::sleep(std::time::Duration::from_secs(2)).await;
626 }
627 Ok(_) => {} // Not a needs_review event or different repo
628 Err(broadcast::error::RecvError::Lagged(n)) => {
629 tracing::debug!(n, "review subscriber lagged; sync catch-up will re-fire missed NeedsReview tasks");
630 }
631 Err(_) => break, // Channel closed
632 }
633 }
634 });
635}
636
637/**
638 * Record a failed review attempt on the task's activity timeline and hand the outcome
639 * back to `classify_review_failure`, which returns it unchanged. Without this the only
640 * trace of a failed attempt was the final `last_error` written at escalation, so
641 * `orch task inspect` showed that a task gave up but never why each attempt failed.
642 *
643 * `external` marks a cause outside the review itself (agent rate limit, GitHub API
644 * limit, no review model). Those also clear `needs_review_refires`: the refire quota
645 * exists to catch a task the subscriber never picks up, and spending it on a cause the
646 * review agent cannot control escalated healthy tasks to Blocked during a rate-limit
647 * window. The retry itself is already handled by the `RateLimited` / `Reset` arms
648 **/
649async fn recorded(
650 store: &Arc<TaskStore>,
651 repo: &str,
652 task_id: &str,
653 reason: &str,
654 cause: &str,
655 external: bool,
656 outcome: ReviewOutcome,
657) -> ReviewOutcome {
658 let store_id = match store.resolve_task_id(repo, task_id).await {
659 Ok(Some(id)) => id,
660 Ok(None) => {
661 tracing::warn!(
662 task_id,
663 "review attempt not recorded — task not found in store"
664 );
665 return outcome;
666 }
667 Err(e) => {
668 tracing::warn!(task_id, err = %e, "review attempt not recorded — task id lookup failed");
669 return outcome;
670 }
671 };
672
673 if let Err(e) = store
674 .append_activity(
675 store_id,
676 "review_attempt_failed",
677 None,
678 None,
679 None,
680 None,
681 Some(&serde_json::json!({
682 "cause": cause,
683 "external": external,
684 "reason": reason,
685 })),
686 )
687 .await
688 {
689 tracing::warn!(task_id, err = %e, "failed to record review attempt failure");
690 }
691
692 if external {
693 if let Err(e) = store
694 .set_fields(store_id, &[("needs_review_refires", serde_json::json!(0))])
695 .await
696 {
697 tracing::warn!(task_id, err = %e, "failed to clear needs_review_refires after external review failure");
698 }
699 }
700
701 outcome
702}
703
704/// Classify a review failure and return the appropriate outcome.
705///
706/// Both the `Ok(ReviewDecision::Failed)` and `Err(e)` arms share the same
707/// logic; callers convert the error to a string before calling this helper.
708/// `context` is used in log messages to distinguish the two call sites
709/// ("review agent" vs "review_and_merge").
710async fn classify_review_failure(
711 store: &Arc<TaskStore>,
712 repo: &str,
713 task_id: &str,
714 reason: &str,
715 context: &str,
716) -> ReviewOutcome {
717 // Rate-limit and auth failures are not real review failures — don't count them
718 // toward MAX_REVIEW_AGENT_FAILURES. The cooldown system steers the next
719 // attempt to an available agent.
720 let lower_reason = reason.to_lowercase();
721
722 // Parse errors are orch's fault, not review feedback. Re-route for retry
723 // without counting toward the failure threshold, so parse errors don't
724 // incorrectly block tasks that have no actual review feedback.
725 if lower_reason.contains("parse error") {
726 tracing::warn!(
727 task_id,
728 reason,
729 "{context} hit parse error — re-routing for retry without counting as failure"
730 );
731 return recorded(
732 store,
733 repo,
734 task_id,
735 reason,
736 "parse_error",
737 false,
738 ReviewOutcome::Reset,
739 )
740 .await;
741 }
742
743 // Session exited mid-thought with no result envelope (issue #3734). Same
744 // treatment as a parse error: retry without counting toward the failure cap
745 if lower_reason.contains("incomplete:") {
746 tracing::warn!(
747 task_id,
748 reason,
749 "{context} session incomplete — re-routing for retry without counting as failure"
750 );
751 return recorded(
752 store,
753 repo,
754 task_id,
755 reason,
756 "incomplete_session",
757 false,
758 ReviewOutcome::Reset,
759 )
760 .await;
761 }
762
763 // The review output was cut off by the model's output/reasoning token
764 // budget, not a genuine parse/format failure. Same treatment as a parse
765 // error: re-route without counting toward MAX_REVIEW_AGENT_FAILURES.
766 if lower_reason.contains("truncated") {
767 tracing::warn!(
768 task_id,
769 reason,
770 "{context} output truncated by token budget — re-routing for retry without counting as failure"
771 );
772 return recorded(
773 store,
774 repo,
775 task_id,
776 reason,
777 "truncated_output",
778 false,
779 ReviewOutcome::Reset,
780 )
781 .await;
782 }
783
784 // Transient mergeability check: GitHub hasn't finished computing yet.
785 // This is a normal, expected transient state for freshly-pushed PRs.
786 // Return RetryMerge (not Reset) so the task stays in InReview — avoids
787 // spawning a redundant review agent that would just re-approve anyway.
788 if lower_reason.contains("not yet computed") {
789 tracing::warn!(
790 task_id,
791 reason,
792 "{context} PR mergeability check still pending — will retry merge without re-reviewing"
793 );
794 return recorded(
795 store,
796 repo,
797 task_id,
798 reason,
799 "mergeability_pending",
800 true,
801 ReviewOutcome::RetryMerge,
802 )
803 .await;
804 }
805
806 // Merge conflicts should be handled by handle_merge_conflict() in
807 // auto_merge.rs (which routes to the task agent). If a conflict error
808 // leaks here, route to the task agent instead of resetting to NeedsReview
809 // (which would re-spawn the review agent in a loop).
810 if lower_reason.contains("merge conflict") || lower_reason.contains("not mergeable") {
811 tracing::warn!(
812 task_id,
813 reason,
814 "{context} hit merge conflict — routing to task agent to resolve"
815 );
816 return recorded(
817 store,
818 repo,
819 task_id,
820 reason,
821 "merge_conflict",
822 false,
823 ReviewOutcome::Reroute,
824 )
825 .await;
826 }
827
828 // Any other merge failure on approved work (rate limit, 5xx, stale lease, a check
829 // still pending) is orch's problem, not the agent's. Keep the task in InReview so
830 // review_open_prs retries the merge once the cause clears, and never re-dispatch the
831 // agent — that burned attempts until the task blocked on work that was already done
832 if lower_reason.contains("merge failed") {
833 tracing::warn!(
834 task_id,
835 reason,
836 "{context} merge failed without a conflict — retrying the merge, not re-dispatching the agent"
837 );
838 return recorded(
839 store,
840 repo,
841 task_id,
842 reason,
843 "merge_retry",
844 true,
845 ReviewOutcome::RetryMerge,
846 )
847 .await;
848 }
849
850 // No configured review model for any available agent, or the selected agent
851 // has no review model. This is a configuration/cooldown gap, not a review
852 // failure. Defer without incrementing the failure counter so the task retries
853 // once a review model is available.
854 if lower_reason.contains("no review agent") || lower_reason.contains("review model missing") {
855 tracing::warn!(
856 task_id,
857 reason,
858 "{context}: no review model available — deferring retry until one is available"
859 );
860 return recorded(
861 store,
862 repo,
863 task_id,
864 reason,
865 "no_review_model",
866 true,
867 ReviewOutcome::RateLimited,
868 )
869 .await;
870 }
871
872 // GhHttp's own throttle says "rate-limited", and GitHub answers 429 or abuse detection
873 // without the words "rate limit"
874 if lower_reason.contains("rate limit")
875 || lower_reason.contains("rate-limited")
876 || lower_reason.contains("rate limited")
877 || lower_reason.contains("too many requests")
878 || lower_reason.contains("abuse detection")
879 || lower_reason.contains("failed (429)")
880 || lower_reason.contains("auth error")
881 {
882 tracing::warn!(
883 task_id,
884 reason,
885 "{context} hit rate limit or auth error — deferring retry until cooldown expires"
886 );
887 return recorded(
888 store,
889 repo,
890 task_id,
891 reason,
892 "rate_limit",
893 true,
894 ReviewOutcome::RateLimited,
895 )
896 .await;
897 }
898
899 if crate::engine::runner::git_ops::is_transient_github_api_error(reason) {
900 tracing::warn!(
901 task_id,
902 reason,
903 "{context} hit transient GitHub error — deferring retry without counting as failure"
904 );
905 return recorded(
906 store,
907 repo,
908 task_id,
909 reason,
910 "github_api_error",
911 true,
912 ReviewOutcome::Reset,
913 )
914 .await;
915 }
916
917 // Provider-side network errors (ConnectionRefused, timeouts, etc.) are
918 // transient infrastructure issues, not review feedback. Re-route without
919 // counting toward MAX_REVIEW_AGENT_FAILURES.
920 if crate::engine::runner::agents::patterns::detect_network_error(reason).is_some() {
921 tracing::warn!(
922 task_id,
923 reason,
924 "{context} hit transient provider network error — re-routing for retry without counting as failure"
925 );
926 return recorded(
927 store,
928 repo,
929 task_id,
930 reason,
931 "provider_network_error",
932 true,
933 ReviewOutcome::Reset,
934 )
935 .await;
936 }
937
938 let failures = match crate::store::store_increment(
939 &Some(store.clone()),
940 repo,
941 task_id,
942 "review_agent_failures",
943 )
944 .await
945 {
946 Ok(v) => v,
947 Err(e) => {
948 tracing::warn!(task_id, err = %e, "failed to increment review_agent_failures — skipping failure escalation this tick");
949 // Treat as 0 for now (no escalation)
950 0
951 }
952 };
953 let blocking = failures >= MAX_REVIEW_AGENT_FAILURES;
954 // Post failure comment to the PR so the history is visible.
955 post_review_failure_comment(store, repo, task_id, reason, failures, blocking).await;
956 if blocking {
957 tracing::error!(
958 task_id,
959 reason,
960 failures,
961 "{context} failed too many times — blocking task"
962 );
963 recorded(
964 store,
965 repo,
966 task_id,
967 reason,
968 "review_failure",
969 false,
970 ReviewOutcome::Block(format!("{context} failed {failures} times: {reason}")),
971 )
972 .await
973 } else {
974 tracing::error!(
975 task_id,
976 reason,
977 failures,
978 "{context} failed — resetting to NeedsReview for retry"
979 );
980 recorded(
981 store,
982 repo,
983 task_id,
984 reason,
985 "review_failure",
986 false,
987 ReviewOutcome::Reset,
988 )
989 .await
990 }
991}
992
993/// Post a comment on the PR explaining why the review agent failed.
994/// Best-effort — if the PR number is missing or the API call fails, we just log.
995async fn post_review_failure_comment(
996 store: &Arc<TaskStore>,
997 repo: &str,
998 task_id: &str,
999 reason: &str,
1000 failures: u64,
1001 blocking: bool,
1002) {
1003 let pr_number = match opt_store_get_task(&Some(store.clone()), repo, task_id).await {
1004 Some(t) => match t.pr_number {
1005 Some(n) => n,
1006 None => return, // No PR to comment on
1007 },
1008 _ => return, // No PR to comment on
1009 };
1010
1011 let status = if blocking {
1012 "Task blocked for human review."
1013 } else {
1014 "Retrying with a different agent."
1015 };
1016
1017 let comment = format!(
1018 "⚠️ Review agent failed (attempt {}/{})\n\n**Reason:** {}\n\n{}{}",
1019 failures,
1020 MAX_REVIEW_AGENT_FAILURES,
1021 reason,
1022 status,
1023 crate::engine::orch_footer()
1024 );
1025
1026 let gh = match GhHttp::new() {
1027 Ok(gh) => gh,
1028 Err(_) => return,
1029 };
1030 if let Err(e) = gh.add_comment(repo, &pr_number.to_string(), &comment).await {
1031 tracing::debug!(
1032 task_id,
1033 pr_number,
1034 error = %e,
1035 "failed to post review failure comment"
1036 );
1037 }
1038}
1039
1040#[cfg(test)]
1041mod tests {
1042 use super::*;
1043
1044 /// An incomplete mid-thought review (no result envelope) must reset for
1045 /// retry without counting toward MAX_REVIEW_AGENT_FAILURES (issue #3734).
1046 #[tokio::test]
1047 async fn classify_review_failure_incomplete_resets_without_counting() {
1048 let store = Arc::new(TaskStore::open_memory().await.unwrap());
1049 let repo = "owner/repo";
1050 let store_id = store
1051 .create_internal(
1052 repo,
1053 "Incomplete review test",
1054 "",
1055 "review",
1056 "review-3734",
1057 None,
1058 )
1059 .await
1060 .unwrap();
1061 let task_id = format!("internal:{store_id}");
1062
1063 for _ in 0..(MAX_REVIEW_AGENT_FAILURES + 1) {
1064 let outcome = classify_review_failure(
1065 &store,
1066 repo,
1067 &task_id,
1068 "incomplete: review session exited before producing a decision",
1069 "review agent",
1070 )
1071 .await;
1072 assert!(
1073 matches!(outcome, ReviewOutcome::Reset),
1074 "incomplete review failure must reset for retry, not block"
1075 );
1076 }
1077
1078 let task = opt_store_get_task(&Some(store.clone()), repo, &task_id)
1079 .await
1080 .expect("task should exist");
1081 assert_eq!(
1082 task.review_agent_failures, 0,
1083 "incomplete failures must not count toward review_agent_failures"
1084 );
1085 }
1086
1087 /// A truncated review output (opencode step_finish reason=length) must be
1088 /// treated like a parse error: reset for retry without counting toward
1089 /// MAX_REVIEW_AGENT_FAILURES, so a run of token-budget flukes never
1090 /// permanently blocks the task (issue #3473).
1091 #[tokio::test]
1092 async fn classify_review_failure_truncated_resets_without_counting() {
1093 let store = Arc::new(TaskStore::open_memory().await.unwrap());
1094 let repo = "owner/repo";
1095 let store_id = store
1096 .create_internal(
1097 repo,
1098 "Truncated review test",
1099 "",
1100 "review",
1101 "review-1",
1102 None,
1103 )
1104 .await
1105 .unwrap();
1106 let task_id = format!("internal:{store_id}");
1107
1108 for _ in 0..(MAX_REVIEW_AGENT_FAILURES + 1) {
1109 let outcome = classify_review_failure(
1110 &store,
1111 repo,
1112 &task_id,
1113 "truncated: review output was cut off before completion \
1114 (output/reasoning token budget exceeded)",
1115 "review agent",
1116 )
1117 .await;
1118 assert!(
1119 matches!(outcome, ReviewOutcome::Reset),
1120 "truncated review failure must reset for retry, not block"
1121 );
1122 }
1123
1124 let task = opt_store_get_task(&Some(store.clone()), repo, &task_id)
1125 .await
1126 .expect("task should exist");
1127 assert_eq!(
1128 task.review_agent_failures, 0,
1129 "truncated failures must not count toward review_agent_failures"
1130 );
1131 }
1132
1133 /**
1134 * After an approved review, a merge that fails for anything other than a content
1135 * conflict must retry the merge with the task still in InReview. Re-dispatching the
1136 * agent re-ran finished work until the task hit the attempt limit (#3783)
1137 **/
1138 #[tokio::test]
1139 async fn classify_review_failure_non_conflict_merge_failure_retries_merge() {
1140 let store = Arc::new(TaskStore::open_memory().await.unwrap());
1141 let repo = "owner/repo";
1142 let store_id = store
1143 .create_internal(repo, "Merge retry", "", "review", "review-3783", None)
1144 .await
1145 .unwrap();
1146 let task_id = format!("internal:{store_id}");
1147
1148 let error_messages = [
1149 "merge failed: GitHub REST API rate-limited, skipping call for 187s",
1150 "merge failed: GitHub PR merge failed (502 Bad Gateway)",
1151 "merge failed: required status check still pending",
1152 ];
1153
1154 for _ in 0..(MAX_REVIEW_AGENT_FAILURES + 1) {
1155 for msg in &error_messages {
1156 let outcome =
1157 classify_review_failure(&store, repo, &task_id, msg, "review agent").await;
1158 assert!(
1159 matches!(outcome, ReviewOutcome::RetryMerge),
1160 "non-conflict merge failure '{msg}' must retry the merge, not re-dispatch the agent"
1161 );
1162 }
1163 }
1164
1165 let task = opt_store_get_task(&Some(store.clone()), repo, &task_id)
1166 .await
1167 .expect("task should exist");
1168 assert_eq!(
1169 task.review_agent_failures, 0,
1170 "a stuck merge is not a review failure"
1171 );
1172 }
1173
1174 /// A genuine content conflict is the one merge failure the agent has to fix
1175 #[tokio::test]
1176 async fn classify_review_failure_merge_conflict_still_reroutes() {
1177 let store = Arc::new(TaskStore::open_memory().await.unwrap());
1178 let repo = "owner/repo";
1179 let store_id = store
1180 .create_internal(repo, "Merge conflict", "", "review", "review-3783b", None)
1181 .await
1182 .unwrap();
1183 let task_id = format!("internal:{store_id}");
1184
1185 for msg in [
1186 "merge failed: PR is not mergeable",
1187 "merge conflict detected on PR #12",
1188 ] {
1189 let outcome =
1190 classify_review_failure(&store, repo, &task_id, msg, "review agent").await;
1191 assert!(
1192 matches!(outcome, ReviewOutcome::Reroute),
1193 "content conflict '{msg}' must go back to the task agent"
1194 );
1195 }
1196 }
1197
1198 /// Transient provider network errors (ConnectionRefused, timeouts) must be
1199 /// treated like parse errors: reset for retry without counting toward
1200 /// MAX_REVIEW_AGENT_FAILURES (issue #3482).
1201 #[tokio::test]
1202 async fn classify_review_failure_network_error_resets_without_counting() {
1203 let store = Arc::new(TaskStore::open_memory().await.unwrap());
1204 let repo = "owner/repo";
1205 let store_id = store
1206 .create_internal(
1207 repo,
1208 "Network error review test",
1209 "",
1210 "review",
1211 "review-2",
1212 None,
1213 )
1214 .await
1215 .unwrap();
1216 let task_id = format!("internal:{store_id}");
1217
1218 let error_messages = [
1219 "API Error: Unable to connect to API (ConnectionRefused)",
1220 "connection refused",
1221 "ECONNREFUSED",
1222 "Upstream idle timeout exceeded",
1223 "connect ETIMEDOUT",
1224 "network request failed",
1225 ];
1226
1227 for msg in &error_messages {
1228 let outcome =
1229 classify_review_failure(&store, repo, &task_id, msg, "review agent").await;
1230 assert!(
1231 matches!(outcome, ReviewOutcome::Reset),
1232 "network error '{msg}' must reset for retry, not block"
1233 );
1234 }
1235
1236 let task = opt_store_get_task(&Some(store.clone()), repo, &task_id)
1237 .await
1238 .expect("task should exist");
1239 assert_eq!(
1240 task.review_agent_failures, 0,
1241 "network errors must not count toward review_agent_failures"
1242 );
1243 }
1244
1245 /// When no agent has a configured review model (or the fallback agent has none),
1246 /// the review must be deferred, not counted as a review failure. Counting it
1247 /// would eventually block tasks for a configuration/cooldown gap (issue #3584).
1248 #[tokio::test]
1249 async fn classify_review_failure_missing_review_model_is_rate_limited() {
1250 let store = Arc::new(TaskStore::open_memory().await.unwrap());
1251 let repo = "owner/repo";
1252 let store_id = store
1253 .create_internal(
1254 repo,
1255 "Missing review model test",
1256 "",
1257 "review",
1258 "review-3",
1259 None,
1260 )
1261 .await
1262 .unwrap();
1263 let task_id = format!("internal:{store_id}");
1264
1265 let error_messages = [
1266 "no review agent with a configured review model is available for task internal:123",
1267 "review model missing for task internal:123 after agent selection — refusing to dispatch",
1268 ];
1269
1270 for _ in 0..(MAX_REVIEW_AGENT_FAILURES + 1) {
1271 for msg in &error_messages {
1272 let outcome =
1273 classify_review_failure(&store, repo, &task_id, msg, "review_and_merge").await;
1274 assert!(
1275 matches!(outcome, ReviewOutcome::RateLimited),
1276 "missing review model error '{msg}' must defer, not block"
1277 );
1278 }
1279 }
1280
1281 let task = opt_store_get_task(&Some(store.clone()), repo, &task_id)
1282 .await
1283 .expect("task should exist");
1284 assert_eq!(
1285 task.review_agent_failures, 0,
1286 "missing review model errors must not count toward review_agent_failures"
1287 );
1288 }
1289
1290 /**
1291 * A review that fails on a rate limit is retried once the cooldown clears, and must
1292 * never spend the rebroadcast quota: `needs_review_refires` is cleared so the sync
1293 * sweep cannot escalate the task to Blocked for a cause outside the review (#3767)
1294 **/
1295 #[tokio::test]
1296 async fn classify_review_failure_rate_limit_clears_refire_quota() {
1297 let store = Arc::new(TaskStore::open_memory().await.unwrap());
1298 let repo = "owner/repo";
1299 let store_id = store
1300 .create_internal(
1301 repo,
1302 "Rate limited review",
1303 "",
1304 "review",
1305 "review-3767",
1306 None,
1307 )
1308 .await
1309 .unwrap();
1310 let task_id = format!("internal:{store_id}");
1311
1312 let error_messages = [
1313 "claude failed: rate limit reached for model sonnet",
1314 "GitHub API GET https://api.github.com/repos/owner/repo/pulls/65 failed (403): API rate limit exceeded",
1315 "GitHub core API rate-limited, skipping call for 1800s",
1316 "GitHub API GET https://api.github.com/repos/owner/repo/pulls/65 failed (429): Too Many Requests",
1317 ];
1318
1319 // The sweep has already burned most of the quota before the rate-limit window
1320 store
1321 .set_fields(store_id, &[("needs_review_refires", serde_json::json!(5))])
1322 .await
1323 .unwrap();
1324
1325 for _ in 0..(MAX_REVIEW_AGENT_FAILURES + 1) {
1326 for msg in &error_messages {
1327 let outcome =
1328 classify_review_failure(&store, repo, &task_id, msg, "review agent").await;
1329 assert!(
1330 matches!(outcome, ReviewOutcome::RateLimited | ReviewOutcome::Reset),
1331 "rate limit '{msg}' must defer for retry, not block"
1332 );
1333 }
1334 }
1335
1336 let task = opt_store_get_task(&Some(store.clone()), repo, &task_id)
1337 .await
1338 .expect("task should exist");
1339 assert_eq!(
1340 task.needs_review_refires, 0,
1341 "a rate limit must not consume the rebroadcast quota"
1342 );
1343 assert_eq!(
1344 task.review_agent_failures, 0,
1345 "rate limits must not count toward review_agent_failures"
1346 );
1347 assert_ne!(
1348 task.status,
1349 crate::store::TaskStatus::Blocked,
1350 "a rate-limited review must leave the task retryable"
1351 );
1352
1353 let attempts = store
1354 .get_activity(store_id, None)
1355 .await
1356 .unwrap()
1357 .into_iter()
1358 .filter(|a| a.event_type == "review_attempt_failed")
1359 .collect::<Vec<_>>();
1360 assert_eq!(
1361 attempts.len(),
1362 (MAX_REVIEW_AGENT_FAILURES as usize + 1) * error_messages.len(),
1363 "every failed attempt must be recorded for `orch task inspect`"
1364 );
1365 assert!(
1366 attempts
1367 .iter()
1368 .all(|a| a.details.get("external").and_then(|v| v.as_bool()) == Some(true)),
1369 "rate limits must be recorded as an external cause"
1370 );
1371 assert!(
1372 attempts.iter().any(
1373 |a| a.details.get("reason").and_then(|v| v.as_str()) == Some(error_messages[0])
1374 ),
1375 "the recorded attempt must carry that attempt's error"
1376 );
1377 }
1378}