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