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 // GhHttp's own throttle says "rate-limited", and GitHub answers 429 or abuse detection
852 // without the words "rate limit"
853 if lower_reason.contains("rate limit")
854 || lower_reason.contains("rate-limited")
855 || lower_reason.contains("rate limited")
856 || lower_reason.contains("too many requests")
857 || lower_reason.contains("abuse detection")
858 || lower_reason.contains("failed (429)")
859 || lower_reason.contains("auth error")
860 {
861 tracing::warn!(
862 task_id,
863 reason,
864 "{context} hit rate limit or auth error — deferring retry until cooldown expires"
865 );
866 return recorded(
867 store,
868 repo,
869 task_id,
870 reason,
871 "rate_limit",
872 true,
873 ReviewOutcome::RateLimited,
874 )
875 .await;
876 }
877
878 if crate::engine::runner::git_ops::is_transient_github_api_error(reason) {
879 tracing::warn!(
880 task_id,
881 reason,
882 "{context} hit transient GitHub error — deferring retry without counting as failure"
883 );
884 return recorded(
885 store,
886 repo,
887 task_id,
888 reason,
889 "github_api_error",
890 true,
891 ReviewOutcome::Reset,
892 )
893 .await;
894 }
895
896 // Provider-side network errors (ConnectionRefused, timeouts, etc.) are
897 // transient infrastructure issues, not review feedback. Re-route without
898 // counting toward MAX_REVIEW_AGENT_FAILURES.
899 if crate::engine::runner::agents::patterns::detect_network_error(reason).is_some() {
900 tracing::warn!(
901 task_id,
902 reason,
903 "{context} hit transient provider network error — re-routing for retry without counting as failure"
904 );
905 return recorded(
906 store,
907 repo,
908 task_id,
909 reason,
910 "provider_network_error",
911 true,
912 ReviewOutcome::Reset,
913 )
914 .await;
915 }
916
917 let failures = match crate::store::store_increment(
918 &Some(store.clone()),
919 repo,
920 task_id,
921 "review_agent_failures",
922 )
923 .await
924 {
925 Ok(v) => v,
926 Err(e) => {
927 tracing::warn!(task_id, err = %e, "failed to increment review_agent_failures — skipping failure escalation this tick");
928 // Treat as 0 for now (no escalation)
929 0
930 }
931 };
932 let blocking = failures >= MAX_REVIEW_AGENT_FAILURES;
933 // Post failure comment to the PR so the history is visible.
934 post_review_failure_comment(store, repo, task_id, reason, failures, blocking).await;
935 if blocking {
936 tracing::error!(
937 task_id,
938 reason,
939 failures,
940 "{context} failed too many times — blocking task"
941 );
942 recorded(
943 store,
944 repo,
945 task_id,
946 reason,
947 "review_failure",
948 false,
949 ReviewOutcome::Block(format!("{context} failed {failures} times: {reason}")),
950 )
951 .await
952 } else {
953 tracing::error!(
954 task_id,
955 reason,
956 failures,
957 "{context} failed — resetting to NeedsReview for retry"
958 );
959 recorded(
960 store,
961 repo,
962 task_id,
963 reason,
964 "review_failure",
965 false,
966 ReviewOutcome::Reset,
967 )
968 .await
969 }
970}
971
972/// Post a comment on the PR explaining why the review agent failed.
973/// Best-effort — if the PR number is missing or the API call fails, we just log.
974async fn post_review_failure_comment(
975 store: &Arc<TaskStore>,
976 repo: &str,
977 task_id: &str,
978 reason: &str,
979 failures: u64,
980 blocking: bool,
981) {
982 let pr_number = match opt_store_get_task(&Some(store.clone()), repo, task_id).await {
983 Some(t) => match t.pr_number {
984 Some(n) => n,
985 None => return, // No PR to comment on
986 },
987 _ => return, // No PR to comment on
988 };
989
990 let status = if blocking {
991 "Task blocked for human review."
992 } else {
993 "Retrying with a different agent."
994 };
995
996 let comment = format!(
997 "⚠️ Review agent failed (attempt {}/{})\n\n**Reason:** {}\n\n{}{}",
998 failures,
999 MAX_REVIEW_AGENT_FAILURES,
1000 reason,
1001 status,
1002 crate::engine::orch_footer()
1003 );
1004
1005 let gh = match GhHttp::new() {
1006 Ok(gh) => gh,
1007 Err(_) => return,
1008 };
1009 if let Err(e) = gh.add_comment(repo, &pr_number.to_string(), &comment).await {
1010 tracing::debug!(
1011 task_id,
1012 pr_number,
1013 error = %e,
1014 "failed to post review failure comment"
1015 );
1016 }
1017}
1018
1019#[cfg(test)]
1020mod tests {
1021 use super::*;
1022
1023 /// An incomplete mid-thought review (no result envelope) must reset for
1024 /// retry without counting toward MAX_REVIEW_AGENT_FAILURES (issue #3734).
1025 #[tokio::test]
1026 async fn classify_review_failure_incomplete_resets_without_counting() {
1027 let store = Arc::new(TaskStore::open_memory().await.unwrap());
1028 let repo = "owner/repo";
1029 let store_id = store
1030 .create_internal(
1031 repo,
1032 "Incomplete review test",
1033 "",
1034 "review",
1035 "review-3734",
1036 None,
1037 )
1038 .await
1039 .unwrap();
1040 let task_id = format!("internal:{store_id}");
1041
1042 for _ in 0..(MAX_REVIEW_AGENT_FAILURES + 1) {
1043 let outcome = classify_review_failure(
1044 &store,
1045 repo,
1046 &task_id,
1047 "incomplete: review session exited before producing a decision",
1048 "review agent",
1049 )
1050 .await;
1051 assert!(
1052 matches!(outcome, ReviewOutcome::Reset),
1053 "incomplete review failure must reset for retry, not block"
1054 );
1055 }
1056
1057 let task = opt_store_get_task(&Some(store.clone()), repo, &task_id)
1058 .await
1059 .expect("task should exist");
1060 assert_eq!(
1061 task.review_agent_failures, 0,
1062 "incomplete failures must not count toward review_agent_failures"
1063 );
1064 }
1065
1066 /// A truncated review output (opencode step_finish reason=length) must be
1067 /// treated like a parse error: reset for retry without counting toward
1068 /// MAX_REVIEW_AGENT_FAILURES, so a run of token-budget flukes never
1069 /// permanently blocks the task (issue #3473).
1070 #[tokio::test]
1071 async fn classify_review_failure_truncated_resets_without_counting() {
1072 let store = Arc::new(TaskStore::open_memory().await.unwrap());
1073 let repo = "owner/repo";
1074 let store_id = store
1075 .create_internal(
1076 repo,
1077 "Truncated review test",
1078 "",
1079 "review",
1080 "review-1",
1081 None,
1082 )
1083 .await
1084 .unwrap();
1085 let task_id = format!("internal:{store_id}");
1086
1087 for _ in 0..(MAX_REVIEW_AGENT_FAILURES + 1) {
1088 let outcome = classify_review_failure(
1089 &store,
1090 repo,
1091 &task_id,
1092 "truncated: review output was cut off before completion \
1093 (output/reasoning token budget exceeded)",
1094 "review agent",
1095 )
1096 .await;
1097 assert!(
1098 matches!(outcome, ReviewOutcome::Reset),
1099 "truncated review failure must reset for retry, not block"
1100 );
1101 }
1102
1103 let task = opt_store_get_task(&Some(store.clone()), repo, &task_id)
1104 .await
1105 .expect("task should exist");
1106 assert_eq!(
1107 task.review_agent_failures, 0,
1108 "truncated failures must not count toward review_agent_failures"
1109 );
1110 }
1111
1112 /// Transient provider network errors (ConnectionRefused, timeouts) must be
1113 /// treated like parse errors: reset for retry without counting toward
1114 /// MAX_REVIEW_AGENT_FAILURES (issue #3482).
1115 #[tokio::test]
1116 async fn classify_review_failure_network_error_resets_without_counting() {
1117 let store = Arc::new(TaskStore::open_memory().await.unwrap());
1118 let repo = "owner/repo";
1119 let store_id = store
1120 .create_internal(
1121 repo,
1122 "Network error review test",
1123 "",
1124 "review",
1125 "review-2",
1126 None,
1127 )
1128 .await
1129 .unwrap();
1130 let task_id = format!("internal:{store_id}");
1131
1132 let error_messages = [
1133 "API Error: Unable to connect to API (ConnectionRefused)",
1134 "connection refused",
1135 "ECONNREFUSED",
1136 "Upstream idle timeout exceeded",
1137 "connect ETIMEDOUT",
1138 "network request failed",
1139 ];
1140
1141 for msg in &error_messages {
1142 let outcome =
1143 classify_review_failure(&store, repo, &task_id, msg, "review agent").await;
1144 assert!(
1145 matches!(outcome, ReviewOutcome::Reset),
1146 "network error '{msg}' must reset for retry, not block"
1147 );
1148 }
1149
1150 let task = opt_store_get_task(&Some(store.clone()), repo, &task_id)
1151 .await
1152 .expect("task should exist");
1153 assert_eq!(
1154 task.review_agent_failures, 0,
1155 "network errors must not count toward review_agent_failures"
1156 );
1157 }
1158
1159 /// When no agent has a configured review model (or the fallback agent has none),
1160 /// the review must be deferred, not counted as a review failure. Counting it
1161 /// would eventually block tasks for a configuration/cooldown gap (issue #3584).
1162 #[tokio::test]
1163 async fn classify_review_failure_missing_review_model_is_rate_limited() {
1164 let store = Arc::new(TaskStore::open_memory().await.unwrap());
1165 let repo = "owner/repo";
1166 let store_id = store
1167 .create_internal(
1168 repo,
1169 "Missing review model test",
1170 "",
1171 "review",
1172 "review-3",
1173 None,
1174 )
1175 .await
1176 .unwrap();
1177 let task_id = format!("internal:{store_id}");
1178
1179 let error_messages = [
1180 "no review agent with a configured review model is available for task internal:123",
1181 "review model missing for task internal:123 after agent selection — refusing to dispatch",
1182 ];
1183
1184 for _ in 0..(MAX_REVIEW_AGENT_FAILURES + 1) {
1185 for msg in &error_messages {
1186 let outcome =
1187 classify_review_failure(&store, repo, &task_id, msg, "review_and_merge").await;
1188 assert!(
1189 matches!(outcome, ReviewOutcome::RateLimited),
1190 "missing review model error '{msg}' must defer, not block"
1191 );
1192 }
1193 }
1194
1195 let task = opt_store_get_task(&Some(store.clone()), repo, &task_id)
1196 .await
1197 .expect("task should exist");
1198 assert_eq!(
1199 task.review_agent_failures, 0,
1200 "missing review model errors must not count toward review_agent_failures"
1201 );
1202 }
1203
1204 /**
1205 * A review that fails on a rate limit is retried once the cooldown clears, and must
1206 * never spend the rebroadcast quota: `needs_review_refires` is cleared so the sync
1207 * sweep cannot escalate the task to Blocked for a cause outside the review (#3767)
1208 **/
1209 #[tokio::test]
1210 async fn classify_review_failure_rate_limit_clears_refire_quota() {
1211 let store = Arc::new(TaskStore::open_memory().await.unwrap());
1212 let repo = "owner/repo";
1213 let store_id = store
1214 .create_internal(
1215 repo,
1216 "Rate limited review",
1217 "",
1218 "review",
1219 "review-3767",
1220 None,
1221 )
1222 .await
1223 .unwrap();
1224 let task_id = format!("internal:{store_id}");
1225
1226 let error_messages = [
1227 "claude failed: rate limit reached for model sonnet",
1228 "GitHub API GET https://api.github.com/repos/owner/repo/pulls/65 failed (403): API rate limit exceeded",
1229 "GitHub core API rate-limited, skipping call for 1800s",
1230 "GitHub API GET https://api.github.com/repos/owner/repo/pulls/65 failed (429): Too Many Requests",
1231 ];
1232
1233 // The sweep has already burned most of the quota before the rate-limit window
1234 store
1235 .set_fields(store_id, &[("needs_review_refires", serde_json::json!(5))])
1236 .await
1237 .unwrap();
1238
1239 for _ in 0..(MAX_REVIEW_AGENT_FAILURES + 1) {
1240 for msg in &error_messages {
1241 let outcome =
1242 classify_review_failure(&store, repo, &task_id, msg, "review agent").await;
1243 assert!(
1244 matches!(outcome, ReviewOutcome::RateLimited | ReviewOutcome::Reset),
1245 "rate limit '{msg}' must defer for retry, not block"
1246 );
1247 }
1248 }
1249
1250 let task = opt_store_get_task(&Some(store.clone()), repo, &task_id)
1251 .await
1252 .expect("task should exist");
1253 assert_eq!(
1254 task.needs_review_refires, 0,
1255 "a rate limit must not consume the rebroadcast quota"
1256 );
1257 assert_eq!(
1258 task.review_agent_failures, 0,
1259 "rate limits must not count toward review_agent_failures"
1260 );
1261 assert_ne!(
1262 task.status,
1263 crate::store::TaskStatus::Blocked,
1264 "a rate-limited review must leave the task retryable"
1265 );
1266
1267 let attempts = store
1268 .get_activity(store_id, None)
1269 .await
1270 .unwrap()
1271 .into_iter()
1272 .filter(|a| a.event_type == "review_attempt_failed")
1273 .collect::<Vec<_>>();
1274 assert_eq!(
1275 attempts.len(),
1276 (MAX_REVIEW_AGENT_FAILURES as usize + 1) * error_messages.len(),
1277 "every failed attempt must be recorded for `orch task inspect`"
1278 );
1279 assert!(
1280 attempts
1281 .iter()
1282 .all(|a| a.details.get("external").and_then(|v| v.as_bool()) == Some(true)),
1283 "rate limits must be recorded as an external cause"
1284 );
1285 assert!(
1286 attempts.iter().any(
1287 |a| a.details.get("reason").and_then(|v| v.as_str()) == Some(error_messages[0])
1288 ),
1289 "the recorded attempt must carry that attempt's error"
1290 );
1291 }
1292}