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