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/**
636 * Record a failed review attempt on the task's activity timeline and hand the outcome
637 * back to `classify_review_failure`, which returns it unchanged. Without this the only
638 * trace of a failed attempt was the final `last_error` written at escalation, so
639 * `orch task inspect` showed that a task gave up but never why each attempt failed.
640 *
641 * `external` marks a cause outside the review itself (agent rate limit, GitHub API
642 * limit, no review model). Those also clear `needs_review_refires`: the refire quota
643 * exists to catch a task the subscriber never picks up, and spending it on a cause the
644 * review agent cannot control escalated healthy tasks to Blocked during a rate-limit
645 * window. The retry itself is already handled by the `RateLimited` / `Reset` arms
646 **/
647async fn recorded(
648 store: &Arc<TaskStore>,
649 repo: &str,
650 task_id: &str,
651 reason: &str,
652 cause: &str,
653 external: bool,
654 outcome: ReviewOutcome,
655) -> ReviewOutcome {
656 let store_id = match store.resolve_task_id(repo, task_id).await {
657 Ok(Some(id)) => id,
658 Ok(None) => {
659 tracing::warn!(
660 task_id,
661 "review attempt not recorded — task not found in store"
662 );
663 return outcome;
664 }
665 Err(e) => {
666 tracing::warn!(task_id, err = %e, "review attempt not recorded — task id lookup failed");
667 return outcome;
668 }
669 };
670
671 if let Err(e) = store
672 .append_activity(
673 store_id,
674 "review_attempt_failed",
675 None,
676 None,
677 None,
678 None,
679 Some(&serde_json::json!({
680 "cause": cause,
681 "external": external,
682 "reason": reason,
683 })),
684 )
685 .await
686 {
687 tracing::warn!(task_id, err = %e, "failed to record review attempt failure");
688 }
689
690 if external {
691 if let Err(e) = store
692 .set_fields(store_id, &[("needs_review_refires", serde_json::json!(0))])
693 .await
694 {
695 tracing::warn!(task_id, err = %e, "failed to clear needs_review_refires after external review failure");
696 }
697 }
698
699 outcome
700}
701
702/// Classify a review failure and return the appropriate outcome.
703///
704/// Both the `Ok(ReviewDecision::Failed)` and `Err(e)` arms share the same
705/// logic; callers convert the error to a string before calling this helper.
706/// `context` is used in log messages to distinguish the two call sites
707/// ("review agent" vs "review_and_merge").
708async fn classify_review_failure(
709 store: &Arc<TaskStore>,
710 repo: &str,
711 task_id: &str,
712 reason: &str,
713 context: &str,
714) -> ReviewOutcome {
715 // Rate-limit and auth failures are not real review failures — don't count them
716 // toward MAX_REVIEW_AGENT_FAILURES. The cooldown system steers the next
717 // attempt to an available agent.
718 let lower_reason = reason.to_lowercase();
719
720 // Parse errors are orch's fault, not review feedback. Re-route for retry
721 // without counting toward the failure threshold, so parse errors don't
722 // incorrectly block tasks that have no actual review feedback.
723 if lower_reason.contains("parse error") {
724 tracing::warn!(
725 task_id,
726 reason,
727 "{context} hit parse error — re-routing for retry without counting as failure"
728 );
729 return recorded(
730 store,
731 repo,
732 task_id,
733 reason,
734 "parse_error",
735 false,
736 ReviewOutcome::Reset,
737 )
738 .await;
739 }
740
741 // Session exited mid-thought with no result envelope (issue #3734). Same
742 // treatment as a parse error: retry without counting toward the failure cap
743 if lower_reason.contains("incomplete:") {
744 tracing::warn!(
745 task_id,
746 reason,
747 "{context} session incomplete — re-routing for retry without counting as failure"
748 );
749 return recorded(
750 store,
751 repo,
752 task_id,
753 reason,
754 "incomplete_session",
755 false,
756 ReviewOutcome::Reset,
757 )
758 .await;
759 }
760
761 // The review output was cut off by the model's output/reasoning token
762 // budget, not a genuine parse/format failure. Same treatment as a parse
763 // error: re-route without counting toward MAX_REVIEW_AGENT_FAILURES.
764 if lower_reason.contains("truncated") {
765 tracing::warn!(
766 task_id,
767 reason,
768 "{context} output truncated by token budget — re-routing for retry without counting as failure"
769 );
770 return recorded(
771 store,
772 repo,
773 task_id,
774 reason,
775 "truncated_output",
776 false,
777 ReviewOutcome::Reset,
778 )
779 .await;
780 }
781
782 // Transient mergeability check: GitHub hasn't finished computing yet.
783 // This is a normal, expected transient state for freshly-pushed PRs.
784 // Return RetryMerge (not Reset) so the task stays in InReview — avoids
785 // spawning a redundant review agent that would just re-approve anyway.
786 if lower_reason.contains("not yet computed") {
787 tracing::warn!(
788 task_id,
789 reason,
790 "{context} PR mergeability check still pending — will retry merge without re-reviewing"
791 );
792 return recorded(
793 store,
794 repo,
795 task_id,
796 reason,
797 "mergeability_pending",
798 true,
799 ReviewOutcome::RetryMerge,
800 )
801 .await;
802 }
803
804 // Merge conflicts should be handled by handle_merge_conflict() in
805 // auto_merge.rs (which routes to the task agent). If a conflict error
806 // leaks here, route to the task agent instead of resetting to NeedsReview
807 // (which would re-spawn the review agent in a loop).
808 if lower_reason.contains("merge conflict")
809 || lower_reason.contains("not mergeable")
810 || lower_reason.contains("merge failed")
811 {
812 tracing::warn!(
813 task_id,
814 reason,
815 "{context} hit merge conflict — routing to task agent to resolve"
816 );
817 return recorded(
818 store,
819 repo,
820 task_id,
821 reason,
822 "merge_conflict",
823 false,
824 ReviewOutcome::Reroute,
825 )
826 .await;
827 }
828
829 // No configured review model for any available agent, or the selected agent
830 // has no review model. This is a configuration/cooldown gap, not a review
831 // failure. Defer without incrementing the failure counter so the task retries
832 // once a review model is available.
833 if lower_reason.contains("no review agent") || lower_reason.contains("review model missing") {
834 tracing::warn!(
835 task_id,
836 reason,
837 "{context}: no review model available — deferring retry until one is available"
838 );
839 return recorded(
840 store,
841 repo,
842 task_id,
843 reason,
844 "no_review_model",
845 true,
846 ReviewOutcome::RateLimited,
847 )
848 .await;
849 }
850
851 if lower_reason.contains("rate limit") || lower_reason.contains("auth error") {
852 tracing::warn!(
853 task_id,
854 reason,
855 "{context} hit rate limit or auth error — deferring retry until cooldown expires"
856 );
857 return recorded(
858 store,
859 repo,
860 task_id,
861 reason,
862 "rate_limit",
863 true,
864 ReviewOutcome::RateLimited,
865 )
866 .await;
867 }
868
869 if crate::engine::runner::git_ops::is_transient_github_api_error(reason) {
870 tracing::warn!(
871 task_id,
872 reason,
873 "{context} hit transient GitHub error — deferring retry without counting as failure"
874 );
875 return recorded(
876 store,
877 repo,
878 task_id,
879 reason,
880 "github_api_error",
881 true,
882 ReviewOutcome::Reset,
883 )
884 .await;
885 }
886
887 // Provider-side network errors (ConnectionRefused, timeouts, etc.) are
888 // transient infrastructure issues, not review feedback. Re-route without
889 // counting toward MAX_REVIEW_AGENT_FAILURES.
890 if crate::engine::runner::agents::patterns::detect_network_error(reason).is_some() {
891 tracing::warn!(
892 task_id,
893 reason,
894 "{context} hit transient provider network error — re-routing for retry without counting as failure"
895 );
896 return recorded(
897 store,
898 repo,
899 task_id,
900 reason,
901 "provider_network_error",
902 true,
903 ReviewOutcome::Reset,
904 )
905 .await;
906 }
907
908 let failures = match crate::store::store_increment(
909 &Some(store.clone()),
910 repo,
911 task_id,
912 "review_agent_failures",
913 )
914 .await
915 {
916 Ok(v) => v,
917 Err(e) => {
918 tracing::warn!(task_id, err = %e, "failed to increment review_agent_failures — skipping failure escalation this tick");
919 // Treat as 0 for now (no escalation)
920 0
921 }
922 };
923 let blocking = failures >= MAX_REVIEW_AGENT_FAILURES;
924 // Post failure comment to the PR so the history is visible.
925 post_review_failure_comment(store, repo, task_id, reason, failures, blocking).await;
926 if blocking {
927 tracing::error!(
928 task_id,
929 reason,
930 failures,
931 "{context} failed too many times — blocking task"
932 );
933 recorded(
934 store,
935 repo,
936 task_id,
937 reason,
938 "review_failure",
939 false,
940 ReviewOutcome::Block(format!("{context} failed {failures} times: {reason}")),
941 )
942 .await
943 } else {
944 tracing::error!(
945 task_id,
946 reason,
947 failures,
948 "{context} failed — resetting to NeedsReview for retry"
949 );
950 recorded(
951 store,
952 repo,
953 task_id,
954 reason,
955 "review_failure",
956 false,
957 ReviewOutcome::Reset,
958 )
959 .await
960 }
961}
962
963/// Post a comment on the PR explaining why the review agent failed.
964/// Best-effort — if the PR number is missing or the API call fails, we just log.
965async fn post_review_failure_comment(
966 store: &Arc<TaskStore>,
967 repo: &str,
968 task_id: &str,
969 reason: &str,
970 failures: u64,
971 blocking: bool,
972) {
973 let pr_number = match opt_store_get_task(&Some(store.clone()), repo, task_id).await {
974 Some(t) => match t.pr_number {
975 Some(n) => n,
976 None => return, // No PR to comment on
977 },
978 _ => return, // No PR to comment on
979 };
980
981 let status = if blocking {
982 "Task blocked for human review."
983 } else {
984 "Retrying with a different agent."
985 };
986
987 let comment = format!(
988 "⚠️ Review agent failed (attempt {}/{})\n\n**Reason:** {}\n\n{}{}",
989 failures,
990 MAX_REVIEW_AGENT_FAILURES,
991 reason,
992 status,
993 crate::engine::orch_footer()
994 );
995
996 let gh = match GhHttp::new() {
997 Ok(gh) => gh,
998 Err(_) => return,
999 };
1000 if let Err(e) = gh.add_comment(repo, &pr_number.to_string(), &comment).await {
1001 tracing::debug!(
1002 task_id,
1003 pr_number,
1004 error = %e,
1005 "failed to post review failure comment"
1006 );
1007 }
1008}
1009
1010#[cfg(test)]
1011mod tests {
1012 use super::*;
1013
1014 /// An incomplete mid-thought review (no result envelope) must reset for
1015 /// retry without counting toward MAX_REVIEW_AGENT_FAILURES (issue #3734).
1016 #[tokio::test]
1017 async fn classify_review_failure_incomplete_resets_without_counting() {
1018 let store = Arc::new(TaskStore::open_memory().await.unwrap());
1019 let repo = "owner/repo";
1020 let store_id = store
1021 .create_internal(
1022 repo,
1023 "Incomplete review test",
1024 "",
1025 "review",
1026 "review-3734",
1027 None,
1028 )
1029 .await
1030 .unwrap();
1031 let task_id = format!("internal:{store_id}");
1032
1033 for _ in 0..(MAX_REVIEW_AGENT_FAILURES + 1) {
1034 let outcome = classify_review_failure(
1035 &store,
1036 repo,
1037 &task_id,
1038 "incomplete: review session exited before producing a decision",
1039 "review agent",
1040 )
1041 .await;
1042 assert!(
1043 matches!(outcome, ReviewOutcome::Reset),
1044 "incomplete review failure must reset for retry, not block"
1045 );
1046 }
1047
1048 let task = opt_store_get_task(&Some(store.clone()), repo, &task_id)
1049 .await
1050 .expect("task should exist");
1051 assert_eq!(
1052 task.review_agent_failures, 0,
1053 "incomplete failures must not count toward review_agent_failures"
1054 );
1055 }
1056
1057 /// A truncated review output (opencode step_finish reason=length) must be
1058 /// treated like a parse error: reset for retry without counting toward
1059 /// MAX_REVIEW_AGENT_FAILURES, so a run of token-budget flukes never
1060 /// permanently blocks the task (issue #3473).
1061 #[tokio::test]
1062 async fn classify_review_failure_truncated_resets_without_counting() {
1063 let store = Arc::new(TaskStore::open_memory().await.unwrap());
1064 let repo = "owner/repo";
1065 let store_id = store
1066 .create_internal(
1067 repo,
1068 "Truncated review test",
1069 "",
1070 "review",
1071 "review-1",
1072 None,
1073 )
1074 .await
1075 .unwrap();
1076 let task_id = format!("internal:{store_id}");
1077
1078 for _ in 0..(MAX_REVIEW_AGENT_FAILURES + 1) {
1079 let outcome = classify_review_failure(
1080 &store,
1081 repo,
1082 &task_id,
1083 "truncated: review output was cut off before completion \
1084 (output/reasoning token budget exceeded)",
1085 "review agent",
1086 )
1087 .await;
1088 assert!(
1089 matches!(outcome, ReviewOutcome::Reset),
1090 "truncated review failure must reset for retry, not block"
1091 );
1092 }
1093
1094 let task = opt_store_get_task(&Some(store.clone()), repo, &task_id)
1095 .await
1096 .expect("task should exist");
1097 assert_eq!(
1098 task.review_agent_failures, 0,
1099 "truncated failures must not count toward review_agent_failures"
1100 );
1101 }
1102
1103 /// Transient provider network errors (ConnectionRefused, timeouts) must be
1104 /// treated like parse errors: reset for retry without counting toward
1105 /// MAX_REVIEW_AGENT_FAILURES (issue #3482).
1106 #[tokio::test]
1107 async fn classify_review_failure_network_error_resets_without_counting() {
1108 let store = Arc::new(TaskStore::open_memory().await.unwrap());
1109 let repo = "owner/repo";
1110 let store_id = store
1111 .create_internal(
1112 repo,
1113 "Network error review test",
1114 "",
1115 "review",
1116 "review-2",
1117 None,
1118 )
1119 .await
1120 .unwrap();
1121 let task_id = format!("internal:{store_id}");
1122
1123 let error_messages = [
1124 "API Error: Unable to connect to API (ConnectionRefused)",
1125 "connection refused",
1126 "ECONNREFUSED",
1127 "Upstream idle timeout exceeded",
1128 "connect ETIMEDOUT",
1129 "network request failed",
1130 ];
1131
1132 for msg in &error_messages {
1133 let outcome =
1134 classify_review_failure(&store, repo, &task_id, msg, "review agent").await;
1135 assert!(
1136 matches!(outcome, ReviewOutcome::Reset),
1137 "network error '{msg}' must reset for retry, not block"
1138 );
1139 }
1140
1141 let task = opt_store_get_task(&Some(store.clone()), repo, &task_id)
1142 .await
1143 .expect("task should exist");
1144 assert_eq!(
1145 task.review_agent_failures, 0,
1146 "network errors must not count toward review_agent_failures"
1147 );
1148 }
1149
1150 /// When no agent has a configured review model (or the fallback agent has none),
1151 /// the review must be deferred, not counted as a review failure. Counting it
1152 /// would eventually block tasks for a configuration/cooldown gap (issue #3584).
1153 #[tokio::test]
1154 async fn classify_review_failure_missing_review_model_is_rate_limited() {
1155 let store = Arc::new(TaskStore::open_memory().await.unwrap());
1156 let repo = "owner/repo";
1157 let store_id = store
1158 .create_internal(
1159 repo,
1160 "Missing review model test",
1161 "",
1162 "review",
1163 "review-3",
1164 None,
1165 )
1166 .await
1167 .unwrap();
1168 let task_id = format!("internal:{store_id}");
1169
1170 let error_messages = [
1171 "no review agent with a configured review model is available for task internal:123",
1172 "review model missing for task internal:123 after agent selection — refusing to dispatch",
1173 ];
1174
1175 for _ in 0..(MAX_REVIEW_AGENT_FAILURES + 1) {
1176 for msg in &error_messages {
1177 let outcome =
1178 classify_review_failure(&store, repo, &task_id, msg, "review_and_merge").await;
1179 assert!(
1180 matches!(outcome, ReviewOutcome::RateLimited),
1181 "missing review model error '{msg}' must defer, not block"
1182 );
1183 }
1184 }
1185
1186 let task = opt_store_get_task(&Some(store.clone()), repo, &task_id)
1187 .await
1188 .expect("task should exist");
1189 assert_eq!(
1190 task.review_agent_failures, 0,
1191 "missing review model errors must not count toward review_agent_failures"
1192 );
1193 }
1194
1195 /**
1196 * A review that fails on a rate limit is retried once the cooldown clears, and must
1197 * never spend the rebroadcast quota: `needs_review_refires` is cleared so the sync
1198 * sweep cannot escalate the task to Blocked for a cause outside the review (#3767)
1199 **/
1200 #[tokio::test]
1201 async fn classify_review_failure_rate_limit_clears_refire_quota() {
1202 let store = Arc::new(TaskStore::open_memory().await.unwrap());
1203 let repo = "owner/repo";
1204 let store_id = store
1205 .create_internal(
1206 repo,
1207 "Rate limited review",
1208 "",
1209 "review",
1210 "review-3767",
1211 None,
1212 )
1213 .await
1214 .unwrap();
1215 let task_id = format!("internal:{store_id}");
1216
1217 let error_messages = [
1218 "claude failed: rate limit reached for model sonnet",
1219 "GitHub API GET https://api.github.com/repos/owner/repo/pulls/65 failed (403): API rate limit exceeded",
1220 ];
1221
1222 // The sweep has already burned most of the quota before the rate-limit window
1223 store
1224 .set_fields(store_id, &[("needs_review_refires", serde_json::json!(5))])
1225 .await
1226 .unwrap();
1227
1228 for _ in 0..(MAX_REVIEW_AGENT_FAILURES + 1) {
1229 for msg in &error_messages {
1230 let outcome =
1231 classify_review_failure(&store, repo, &task_id, msg, "review agent").await;
1232 assert!(
1233 matches!(outcome, ReviewOutcome::RateLimited | ReviewOutcome::Reset),
1234 "rate limit '{msg}' must defer for retry, not block"
1235 );
1236 }
1237 }
1238
1239 let task = opt_store_get_task(&Some(store.clone()), repo, &task_id)
1240 .await
1241 .expect("task should exist");
1242 assert_eq!(
1243 task.needs_review_refires, 0,
1244 "a rate limit must not consume the rebroadcast quota"
1245 );
1246 assert_eq!(
1247 task.review_agent_failures, 0,
1248 "rate limits must not count toward review_agent_failures"
1249 );
1250 assert_ne!(
1251 task.status,
1252 crate::store::TaskStatus::Blocked,
1253 "a rate-limited review must leave the task retryable"
1254 );
1255
1256 let attempts = store
1257 .get_activity(store_id, None)
1258 .await
1259 .unwrap()
1260 .into_iter()
1261 .filter(|a| a.event_type == "review_attempt_failed")
1262 .collect::<Vec<_>>();
1263 assert_eq!(
1264 attempts.len(),
1265 (MAX_REVIEW_AGENT_FAILURES as usize + 1) * error_messages.len(),
1266 "every failed attempt must be recorded for `orch task inspect`"
1267 );
1268 assert!(
1269 attempts
1270 .iter()
1271 .all(|a| a.details.get("external").and_then(|v| v.as_bool()) == Some(true)),
1272 "rate limits must be recorded as an external cause"
1273 );
1274 assert!(
1275 attempts.iter().any(
1276 |a| a.details.get("reason").and_then(|v| v.as_str()) == Some(error_messages[0])
1277 ),
1278 "the recorded attempt must carry that attempt's error"
1279 );
1280 }
1281}