fix: use write_json_atomic for implementation_plan.json in recovery

Replace raw json.dump with write_json_atomic when writing
implementation_plan.json in mark_subtask_stuck() to prevent file
corruption, consistent with 8+ other call sites in the codebase.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

apps/backend/cli/qa_commands.py

@@ -101,8 +101,12 @@ def handle_qa_command(
             print("\n✅ Build already approved by QA.")
         else:
             completed, total = count_subtasks(spec_dir)
-            print(f"\n❌ Build not complete ({completed}/{total} subtasks).")
-            print("Complete all subtasks before running QA validation.")
+            print(
+                f"\n❌ Build not ready for QA ({completed}/{total} subtasks completed)."
+            )
+            print(
+                "All subtasks must reach a terminal state (completed, failed, or stuck) before running QA."
+            )
         return
 
     if has_human_feedback:

apps/backend/core/progress.py

@@ -115,6 +115,65 @@ def is_build_complete(spec_dir: Path) -> bool:
     return total > 0 and completed == total
 
 
+def _load_stuck_subtask_ids(spec_dir: Path) -> set[str]:
+    """Load IDs of subtasks marked as stuck from attempt_history.json."""
+    stuck_subtask_ids: set[str] = set()
+    attempt_history_file = spec_dir / "memory" / "attempt_history.json"
+    if attempt_history_file.exists():
+        try:
+            with open(attempt_history_file, encoding="utf-8") as f:
+                attempt_history = json.load(f)
+            for entry in attempt_history.get("stuck_subtasks", []):
+                if "subtask_id" in entry:
+                    stuck_subtask_ids.add(entry["subtask_id"])
+        except (OSError, json.JSONDecodeError, UnicodeDecodeError):
+            # Corrupted attempt history is non-fatal; skip stuck-subtask filtering
+            pass
+    return stuck_subtask_ids
+
+
+def is_build_ready_for_qa(spec_dir: Path) -> bool:
+    """
+    Check if the build is ready for QA validation.
+
+    Unlike is_build_complete() which requires all subtasks to be "completed",
+    this function considers the build ready when all subtasks have reached
+    a terminal state: completed, failed, or stuck (exhausted retries in attempt_history.json).
+
+    Args:
+        spec_dir: Directory containing implementation_plan.json
+
+    Returns:
+        True if all subtasks are in a terminal state, False otherwise
+    """
+    plan_file = spec_dir / "implementation_plan.json"
+    if not plan_file.exists():
+        return False
+
+    stuck_subtask_ids = _load_stuck_subtask_ids(spec_dir)
+
+    try:
+        with open(plan_file, encoding="utf-8") as f:
+            plan = json.load(f)
+
+        total = 0
+        terminal = 0
+
+        for phase in plan.get("phases", []):
+            for subtask in phase.get("subtasks", []):
+                total += 1
+                status = subtask.get("status", "pending")
+                subtask_id = subtask.get("id")
+
+                if status in ("completed", "failed") or subtask_id in stuck_subtask_ids:
+                    terminal += 1
+
+        return total > 0 and terminal == total
+
+    except (OSError, json.JSONDecodeError, UnicodeDecodeError):
+        return False
+
+
 def get_progress_percentage(spec_dir: Path) -> float:
     """
     Get the progress as a percentage.
@@ -420,22 +479,7 @@ def get_next_subtask(spec_dir: Path) -> dict | None:
     if not plan_file.exists():
         return None
 
-    # Load stuck subtasks from recovery manager's attempt history
-    stuck_subtask_ids = set()
-    attempt_history_file = spec_dir / "memory" / "attempt_history.json"
-    if attempt_history_file.exists():
-        try:
-            with open(attempt_history_file, encoding="utf-8") as f:
-                attempt_history = json.load(f)
-            # Collect IDs of subtasks marked as stuck
-            stuck_subtask_ids = {
-                entry["subtask_id"]
-                for entry in attempt_history.get("stuck_subtasks", [])
-                if "subtask_id" in entry
-            }
-        except (OSError, json.JSONDecodeError, UnicodeDecodeError):
-            # If we can't read the file, continue without stuck checking
-            pass
+    stuck_subtask_ids = _load_stuck_subtask_ids(spec_dir)
 
     try:
         with open(plan_file, encoding="utf-8") as f:

apps/backend/progress.py

@@ -14,6 +14,7 @@ from core.progress import (
     get_plan_summary,
     get_progress_percentage,
     is_build_complete,
+    is_build_ready_for_qa,
     print_build_complete_banner,
     print_paused_banner,
     print_progress_summary,
@@ -29,6 +30,7 @@ __all__ = [
     "get_plan_summary",
     "get_progress_percentage",
     "is_build_complete",
+    "is_build_ready_for_qa",
     "print_build_complete_banner",
     "print_paused_banner",
     "print_progress_summary",

apps/backend/qa/criteria.py

@@ -8,7 +8,7 @@ Manages acceptance criteria validation and status tracking.
 import json
 from pathlib import Path
 
-from progress import is_build_complete
+from progress import is_build_ready_for_qa
 
 # =============================================================================
 # IMPLEMENTATION PLAN I/O
@@ -95,10 +95,10 @@ def should_run_qa(spec_dir: Path) -> bool:
     Determine if QA validation should run.
 
     QA should run when:
-    - All subtasks are completed
+    - All subtasks have reached a terminal state (completed, failed, or stuck)
     - QA has not yet approved
     """
-    if not is_build_complete(spec_dir):
+    if not is_build_ready_for_qa(spec_dir):
         return False
 
     if is_qa_approved(spec_dir):

apps/backend/qa/loop.py

@@ -28,7 +28,7 @@ from phase_config import (
     get_phase_model_betas,
 )
 from phase_event import ExecutionPhase, emit_phase
-from progress import count_subtasks, is_build_complete
+from progress import count_subtasks, is_build_ready_for_qa
 from security.constants import PROJECT_DIR_ENV_VAR
 from task_logger import (
     LogPhase,
@@ -114,14 +114,25 @@ async def run_qa_validation_loop(
     # Initialize task logger for the validation phase
     task_logger = get_task_logger(spec_dir)
 
-    # Verify build is complete
-    if not is_build_complete(spec_dir):
-        debug_warning("qa_loop", "Build is not complete, cannot run QA")
-        print("\n❌ Build is not complete. Cannot run QA validation.")
-        completed, total = count_subtasks(spec_dir)
-        debug("qa_loop", "Build progress", completed=completed, total=total)
-        print(f"   Progress: {completed}/{total} subtasks completed")
-        return False
+    # Check if there's pending human feedback that needs to be processed
+    fix_request_file = spec_dir / "QA_FIX_REQUEST.md"
+    has_human_feedback = fix_request_file.exists()
+
+    # Human feedback takes priority — if the user explicitly asked to proceed,
+    # skip the build completeness gate entirely
+    if not has_human_feedback:
+        # Verify build is ready for QA (all subtasks in terminal state)
+        if not is_build_ready_for_qa(spec_dir):
+            debug_warning(
+                "qa_loop", "Build is not ready for QA - subtasks still in progress"
+            )
+            print("\n❌ Build is not ready for QA validation.")
+            completed, total = count_subtasks(spec_dir)
+            debug("qa_loop", "Build progress", completed=completed, total=total)
+            print(
+                f"   Progress: {completed}/{total} subtasks in terminal state (completed/failed/stuck)"
+            )
+            return False
 
     # Emit phase event at start of QA validation (before any early returns)
     emit_phase(ExecutionPhase.QA_REVIEW, "Starting QA validation")
@@ -136,10 +147,6 @@ async def run_qa_validation_loop(
         f"[Fast Mode] {'ENABLED' if fast_mode else 'disabled'} for QA validation",
     )
 
-    # Check if there's pending human feedback that needs to be processed
-    fix_request_file = spec_dir / "QA_FIX_REQUEST.md"
-    has_human_feedback = fix_request_file.exists()
-
     # Check if already approved - but if there's human feedback, we need to process it first
     if is_qa_approved(spec_dir) and not has_human_feedback:
         debug_success("qa_loop", "Build already approved by QA")

apps/backend/runners/github/services/followup_reviewer.py

@@ -886,7 +886,8 @@ Analyze this follow-up review context and provide your structured response.
         """Attempt a short SDK call with minimal schema to recover review data.
 
         This is the extraction recovery step when full structured output validation fails.
-        Uses FollowupExtractionResponse (~6 flat fields) which has near-100% success rate.
+        Uses FollowupExtractionResponse (small schema with ExtractedFindingSummary nesting)
+        which has near-100% success rate.
 
         Uses create_client() + process_sdk_stream() for proper OAuth handling,
         matching the pattern in parallel_followup_reviewer.py.
@@ -900,7 +901,8 @@ Analyze this follow-up review context and provide your structured response.
             extraction_prompt = (
                 "Extract the key review data from the following AI analysis output. "
                 "Return the verdict, reasoning, resolved finding IDs, unresolved finding IDs, "
-                "one-line summaries of any new findings, and counts of confirmed/dismissed findings.\n\n"
+                "structured summaries of any new findings (including severity, description, file path, and line number), "
+                "and counts of confirmed/dismissed findings.\n\n"
                 f"--- AI ANALYSIS OUTPUT ---\n{text[:8000]}\n--- END ---"
             )
 
@@ -946,9 +948,16 @@ Analyze this follow-up review context and provide your structured response.
 
             # Convert extraction to internal format with reconstructed findings
             new_findings = []
-            for i, summary in enumerate(extracted.new_finding_summaries):
+            for i, summary_obj in enumerate(extracted.new_finding_summaries):
                 new_findings.append(
-                    create_finding_from_summary(summary, i, id_prefix="FR")
+                    create_finding_from_summary(
+                        summary=summary_obj.description,
+                        index=i,
+                        id_prefix="FR",
+                        severity_override=summary_obj.severity,
+                        file=summary_obj.file,
+                        line=summary_obj.line,
+                    )
                 )
 
             # Build finding_resolutions from extraction data for _apply_ai_resolutions

apps/backend/runners/github/services/parallel_followup_reviewer.py

@@ -1129,7 +1129,8 @@ The SDK will run invoked agents in parallel automatically.
         """Attempt a short SDK call with a minimal schema to recover review data.
 
         This is the Tier 2 recovery step when full structured output validation fails.
-        Uses FollowupExtractionResponse (~6 flat fields) which has near-100% success rate.
+        Uses FollowupExtractionResponse (small schema with ExtractedFindingSummary nesting)
+        which has near-100% success rate.
 
         Returns parsed result dict on success, None on failure.
         """
@@ -1146,7 +1147,8 @@ The SDK will run invoked agents in parallel automatically.
             extraction_prompt = (
                 "Extract the key review data from the following AI analysis output. "
                 "Return the verdict, reasoning, resolved finding IDs, unresolved finding IDs, "
-                "one-line summaries of any new findings, and counts of confirmed/dismissed findings.\n\n"
+                "structured summaries of any new findings (including severity, description, file path, and line number), "
+                "and counts of confirmed/dismissed findings.\n\n"
                 f"--- AI ANALYSIS OUTPUT ---\n{text[:8000]}\n--- END ---"
             )
 
@@ -1205,10 +1207,17 @@ The SDK will run invoked agents in parallel automatically.
             findings = []
             new_finding_ids = []
 
-            # 1. Convert new_finding_summaries to minimal PRReviewFinding objects
-            # Uses shared helper for "SEVERITY: description" parsing and ID generation
-            for i, summary in enumerate(extracted.new_finding_summaries):
-                finding = create_finding_from_summary(summary, i, id_prefix="FU")
+            # 1. Convert new_finding_summaries to PRReviewFinding objects
+            # ExtractedFindingSummary objects carry file/line from extraction
+            for i, summary_obj in enumerate(extracted.new_finding_summaries):
+                finding = create_finding_from_summary(
+                    summary=summary_obj.description,
+                    index=i,
+                    id_prefix="FU",
+                    severity_override=summary_obj.severity,
+                    file=summary_obj.file,
+                    line=summary_obj.line,
+                )
                 new_finding_ids.append(finding.id)
                 findings.append(finding)
 

apps/backend/runners/github/services/parallel_orchestrator_reviewer.py

@@ -1289,12 +1289,30 @@ The SDK will run invoked agents in parallel automatically.
                 f"{len(filtered_findings)} filtered"
             )
 
-            # No confidence routing - validation is binary via finding-validator
-            unique_findings = validated_findings
-            logger.info(f"[PRReview] Final findings: {len(unique_findings)} validated")
+            # Separate active findings (drive verdict) from dismissed (shown in UI only)
+            active_findings = []
+            dismissed_findings = []
+            for f in validated_findings:
+                if f.validation_status == "dismissed_false_positive":
+                    dismissed_findings.append(f)
+                else:
+                    active_findings.append(f)
 
+            safe_print(
+                f"[ParallelOrchestrator] Final: {len(active_findings)} active, "
+                f"{len(dismissed_findings)} disputed by validator",
+                flush=True,
+            )
             logger.info(
-                f"[ParallelOrchestrator] Review complete: {len(unique_findings)} findings"
+                f"[PRReview] Final findings: {len(active_findings)} active, "
+                f"{len(dismissed_findings)} disputed"
+            )
+
+            # All findings (active + dismissed) go in the result for UI display
+            all_review_findings = validated_findings
+            logger.info(
+                f"[ParallelOrchestrator] Review complete: {len(all_review_findings)} findings "
+                f"({len(active_findings)} active, {len(dismissed_findings)} disputed)"
             )
 
             # Fetch CI status for verdict consideration
@@ -1304,9 +1322,9 @@ The SDK will run invoked agents in parallel automatically.
                 f"{ci_status.get('failing', 0)} failing, {ci_status.get('pending', 0)} pending"
             )
 
-            # Generate verdict (includes merge conflict check, branch-behind check, and CI status)
+            # Generate verdict from ACTIVE findings only (dismissed don't affect verdict)
             verdict, verdict_reasoning, blockers = self._generate_verdict(
-                unique_findings,
+                active_findings,
                 has_merge_conflicts=context.has_merge_conflicts,
                 merge_state_status=context.merge_state_status,
                 ci_status=ci_status,
@@ -1317,7 +1335,7 @@ The SDK will run invoked agents in parallel automatically.
                 verdict=verdict,
                 verdict_reasoning=verdict_reasoning,
                 blockers=blockers,
-                findings=unique_findings,
+                findings=all_review_findings,
                 agents_invoked=agents_invoked,
             )
 
@@ -1362,7 +1380,7 @@ The SDK will run invoked agents in parallel automatically.
                 pr_number=context.pr_number,
                 repo=self.config.repo,
                 success=True,
-                findings=unique_findings,
+                findings=all_review_findings,
                 summary=summary,
                 overall_status=overall_status,
                 verdict=verdict,
@@ -1937,12 +1955,38 @@ For EACH finding above:
                 validated_findings.append(finding)
 
             elif validation.validation_status == "dismissed_false_positive":
-                # Dismiss - do not include
-                dismissed_count += 1
-                logger.info(
-                    f"[PRReview] Dismissed {finding.id} as false positive: "
-                    f"{validation.explanation[:100]}"
-                )
+                # Protect cross-validated findings from dismissal —
+                # if multiple specialists independently found the same issue,
+                # a single validator should not override that consensus
+                if finding.cross_validated:
+                    finding.validation_status = "confirmed_valid"
+                    finding.validation_evidence = validation.code_evidence
+                    finding.validation_explanation = (
+                        f"[Auto-kept: cross-validated by {len(finding.source_agents)} agents] "
+                        f"{validation.explanation}"
+                    )
+                    validated_findings.append(finding)
+                    safe_print(
+                        f"[FindingValidator] Kept cross-validated finding '{finding.title}' "
+                        f"despite dismissal (agents={finding.source_agents})",
+                        flush=True,
+                    )
+                else:
+                    # Keep finding but mark as dismissed (user can see it in UI)
+                    finding.validation_status = "dismissed_false_positive"
+                    finding.validation_evidence = validation.code_evidence
+                    finding.validation_explanation = validation.explanation
+                    validated_findings.append(finding)
+                    dismissed_count += 1
+                    safe_print(
+                        f"[FindingValidator] Disputed '{finding.title}': "
+                        f"{validation.explanation} (file={finding.file}:{finding.line})",
+                        flush=True,
+                    )
+                    logger.info(
+                        f"[PRReview] Disputed {finding.id}: "
+                        f"{validation.explanation[:200]}"
+                    )
 
             elif validation.validation_status == "needs_human_review":
                 # Keep but flag
@@ -2127,11 +2171,16 @@ For EACH finding above:
                 sev = f.severity.value
                 emoji = severity_emoji.get(sev, "⚪")
 
+                is_disputed = f.validation_status == "dismissed_false_positive"
+
                 # Finding header with location
                 line_range = f"L{f.line}"
                 if f.end_line and f.end_line != f.line:
                     line_range = f"L{f.line}-L{f.end_line}"
-                lines.append(f"#### {emoji} [{sev.upper()}] {f.title}")
+                if is_disputed:
+                    lines.append(f"#### ⚪ [DISPUTED] ~~{f.title}~~")
+                else:
+                    lines.append(f"#### {emoji} [{sev.upper()}] {f.title}")
                 lines.append(f"**File:** `{f.file}` ({line_range})")
 
                 # Cross-validation badge
@@ -2161,6 +2210,7 @@ For EACH finding above:
                     status_label = {
                         "confirmed_valid": "Confirmed",
                         "needs_human_review": "Needs human review",
+                        "dismissed_false_positive": "Disputed by validator",
                     }.get(f.validation_status, f.validation_status)
                     lines.append("")
                     lines.append(f"**Validation:** {status_label}")
@@ -2182,18 +2232,27 @@ For EACH finding above:
 
                 lines.append("")
 
-            # Findings count summary
+            # Findings count summary (exclude dismissed from active count)
+            active_count = 0
+            dismissed_count = 0
             by_severity: dict[str, int] = {}
             for f in findings:
+                if f.validation_status == "dismissed_false_positive":
+                    dismissed_count += 1
+                    continue
+                active_count += 1
                 sev = f.severity.value
                 by_severity[sev] = by_severity.get(sev, 0) + 1
             summary_parts = []
             for sev in ["critical", "high", "medium", "low"]:
                 if sev in by_severity:
                     summary_parts.append(f"{by_severity[sev]} {sev}")
-            lines.append(
-                f"**Total:** {len(findings)} finding(s) ({', '.join(summary_parts)})"
+            count_text = (
+                f"**Total:** {active_count} finding(s) ({', '.join(summary_parts)})"
             )
+            if dismissed_count > 0:
+                count_text += f" + {dismissed_count} disputed"
+            lines.append(count_text)
             lines.append("")
 
         lines.append("---")

apps/backend/runners/github/services/pydantic_models.py

@@ -533,10 +533,26 @@ class FindingValidationResponse(BaseModel):
 # =============================================================================
 
 
+class ExtractedFindingSummary(BaseModel):
+    """Per-finding summary with file location for extraction recovery."""
+
+    severity: str = Field(description="Severity level: LOW, MEDIUM, HIGH, or CRITICAL")
+    description: str = Field(description="One-line description of the finding")
+    file: str = Field(
+        default="unknown", description="File path where the issue was found"
+    )
+    line: int = Field(default=0, description="Line number in the file (0 if unknown)")
+
+    @field_validator("severity", mode="before")
+    @classmethod
+    def _normalize_severity(cls, v: str) -> str:
+        return _normalize_severity(v)
+
+
 class FollowupExtractionResponse(BaseModel):
     """Minimal extraction schema for recovering data when full structured output fails.
 
-    Deliberately kept small (~6 fields, no nesting) for near-100% validation success.
+    Uses ExtractedFindingSummary for new findings to preserve file/line information.
     Used as an intermediate recovery step before falling back to raw text parsing.
     """
 
@@ -552,9 +568,9 @@ class FollowupExtractionResponse(BaseModel):
         default_factory=list,
         description="IDs of previous findings that remain unresolved",
     )
-    new_finding_summaries: list[str] = Field(
+    new_finding_summaries: list[ExtractedFindingSummary] = Field(
         default_factory=list,
-        description="One-line summary of each new finding (e.g. 'HIGH: cleanup deletes QA-rejected specs in batch_commands.py')",
+        description="Structured summary of each new finding with file location",
     )
     confirmed_finding_count: int = Field(
         0, description="Number of findings confirmed as valid"

apps/backend/runners/github/services/recovery_utils.py

@@ -80,6 +80,9 @@ def create_finding_from_summary(
     summary: str,
     index: int,
     id_prefix: str = "FR",
+    severity_override: str | None = None,
+    file: str = "unknown",
+    line: int = 0,
 ) -> PRReviewFinding:
     """Create a PRReviewFinding from an extraction summary string.
 
@@ -90,11 +93,20 @@ def create_finding_from_summary(
         summary: Raw summary string, e.g. "HIGH: Missing null check in parser.py"
         index: The index of the finding in the extraction list.
         id_prefix: ID prefix for traceability. Default "FR" (Followup Recovery).
+        severity_override: If provided, use this severity instead of parsing from summary.
+        file: File path where the issue was found (default "unknown").
+        line: Line number in the file (default 0).
 
     Returns:
         A PRReviewFinding with parsed severity, generated ID, and description.
     """
     severity, description = parse_severity_from_summary(summary)
+
+    # Use severity_override if provided
+    if severity_override is not None:
+        severity_map = {k.rstrip(":"): v for k, v in _EXTRACTION_SEVERITY_MAP}
+        severity = severity_map.get(severity_override.upper(), severity)
+
     finding_id = generate_recovery_finding_id(index, description, prefix=id_prefix)
 
     return PRReviewFinding(
@@ -103,6 +115,6 @@ def create_finding_from_summary(
         category=ReviewCategory.QUALITY,
         title=description[:80],
         description=f"[Recovered via extraction] {description}",
-        file="unknown",
-        line=0,
+        file=file,
+        line=line,
     )

apps/backend/services/recovery.py

@@ -21,6 +21,8 @@ from datetime import datetime, timedelta, timezone
 from enum import Enum
 from pathlib import Path
 
+from core.file_utils import write_json_atomic
+
 # Recovery manager configuration
 ATTEMPT_WINDOW_SECONDS = 7200  # Only count attempts within last 2 hours
 MAX_ATTEMPT_HISTORY_PER_SUBTASK = 50  # Cap stored attempts per subtask
@@ -514,6 +516,36 @@ class RecoveryManager:
 
         self._save_attempt_history(history)
 
+        # Also update the subtask status in implementation_plan.json
+        # so that other callers (like is_build_ready_for_qa) see accurate status
+        try:
+            plan_file = self.spec_dir / "implementation_plan.json"
+            if plan_file.exists():
+                with open(plan_file, encoding="utf-8") as f:
+                    plan = json.load(f)
+
+                updated = False
+                for phase in plan.get("phases", []):
+                    for subtask in phase.get("subtasks", []):
+                        if subtask.get("id") == subtask_id:
+                            subtask["status"] = "failed"
+                            stuck_note = f"Marked as stuck: {reason}"
+                            existing = subtask.get("actual_output", "")
+                            subtask["actual_output"] = (
+                                f"{stuck_note}\n{existing}" if existing else stuck_note
+                            )
+                            updated = True
+                            break
+                    if updated:
+                        break
+
+                if updated:
+                    write_json_atomic(plan_file, plan, indent=2)
+        except (OSError, json.JSONDecodeError, UnicodeDecodeError) as e:
+            logger.warning(
+                f"Failed to update implementation_plan.json for stuck subtask {subtask_id}: {e}"
+            )
+
     def get_stuck_subtasks(self) -> list[dict]:
         """
         Get all subtasks marked as stuck.

apps/frontend/src/main/ipc-handlers/github/pr-handlers.ts

@@ -268,6 +268,10 @@ export interface PRReviewFinding {
   endLine?: number;
   suggestedFix?: string;
   fixable: boolean;
+  validationStatus?: "confirmed_valid" | "dismissed_false_positive" | "needs_human_review" | null;
+  validationExplanation?: string;
+  sourceAgents?: string[];
+  crossValidated?: boolean;
 }
 
 /**
@@ -1341,6 +1345,10 @@ function getReviewResult(project: Project, prNumber: number): PRReviewResult | n
           endLine: f.end_line,
           suggestedFix: f.suggested_fix,
           fixable: f.fixable ?? false,
+          validationStatus: f.validation_status ?? null,
+          validationExplanation: f.validation_explanation ?? undefined,
+          sourceAgents: f.source_agents ?? [],
+          crossValidated: f.cross_validated ?? false,
         })) ?? [],
       summary: data.summary ?? "",
       overallStatus: data.overall_status ?? "comment",

apps/frontend/src/preload/api/modules/github-api.ts

@@ -376,6 +376,10 @@ export interface PRReviewFinding {
   endLine?: number;
   suggestedFix?: string;
   fixable: boolean;
+  validationStatus?: 'confirmed_valid' | 'dismissed_false_positive' | 'needs_human_review' | null;
+  validationExplanation?: string;
+  sourceAgents?: string[];
+  crossValidated?: boolean;
 }
 
 /**

apps/frontend/src/renderer/components/github-prs/components/FindingItem.tsx

@@ -14,6 +14,7 @@ interface FindingItemProps {
   finding: PRReviewFinding;
   selected: boolean;
   posted?: boolean;
+  disputed?: boolean;
   onToggle: () => void;
 }
 
@@ -33,7 +34,7 @@ function getCategoryTranslationKey(category: string): string {
   return categoryMap[category.toLowerCase()] || category;
 }
 
-export function FindingItem({ finding, selected, posted = false, onToggle }: FindingItemProps) {
+export function FindingItem({ finding, selected, posted = false, disputed = false, onToggle }: FindingItemProps) {
   const { t } = useTranslation('common');
   const CategoryIcon = getCategoryIcon(finding.category);
 
@@ -45,8 +46,9 @@ export function FindingItem({ finding, selected, posted = false, onToggle }: Fin
     <div
       className={cn(
         "rounded-lg border bg-background p-3 space-y-2 transition-colors",
-        selected && !posted && "ring-2 ring-primary/50",
-        posted && "opacity-60"
+        selected && !posted && !disputed && "ring-2 ring-primary/50",
+        selected && disputed && "ring-2 ring-purple-500/50",
+        (posted || (disputed && !selected)) && "opacity-60"
       )}
     >
       {/* Finding Header */}
@@ -72,6 +74,16 @@ export function FindingItem({ finding, selected, posted = false, onToggle }: Fin
                 {t('prReview.posted')}
               </Badge>
             )}
+            {disputed && (
+              <Badge variant="outline" className="text-xs shrink-0 bg-purple-500/10 text-purple-500 border-purple-500/30">
+                {t('prReview.disputed')}
+              </Badge>
+            )}
+            {finding.crossValidated && finding.sourceAgents && finding.sourceAgents.length > 1 && (
+              <Badge variant="outline" className="text-xs shrink-0 bg-green-500/10 text-green-500 border-green-500/30">
+                {t('prReview.crossValidatedBy', { count: finding.sourceAgents.length })}
+              </Badge>
+            )}
             <span className="font-medium text-sm break-words">
               {finding.title}
             </span>
@@ -79,6 +91,11 @@ export function FindingItem({ finding, selected, posted = false, onToggle }: Fin
           <p className="text-sm text-muted-foreground break-words">
             {finding.description}
           </p>
+          {disputed && finding.validationExplanation && (
+            <p className="text-xs text-purple-500/80 italic break-words">
+              {finding.validationExplanation}
+            </p>
+          )}
           <div className="text-xs text-muted-foreground">
             <code className="bg-muted px-1 py-0.5 rounded break-all">
               {finding.file}:{finding.line}

apps/frontend/src/renderer/components/github-prs/components/FindingsSummary.tsx

@@ -9,9 +9,10 @@ import type { PRReviewFinding } from '../hooks/useGitHubPRs';
 interface FindingsSummaryProps {
   findings: PRReviewFinding[];
   selectedCount: number;
+  disputedCount?: number;
 }
 
-export function FindingsSummary({ findings, selectedCount }: FindingsSummaryProps) {
+export function FindingsSummary({ findings, selectedCount, disputedCount = 0 }: FindingsSummaryProps) {
   const { t } = useTranslation('common');
 
   // Count findings by severity
@@ -46,6 +47,11 @@ export function FindingsSummary({ findings, selectedCount }: FindingsSummaryProp
             {counts.low} {t('prReview.severity.low')}
           </Badge>
         )}
+        {disputedCount > 0 && (
+          <Badge variant="outline" className="bg-purple-500/10 text-purple-500 border-purple-500/30">
+            {disputedCount} {t('prReview.disputed')}
+          </Badge>
+        )}
       </div>
       <span className="text-xs text-muted-foreground">
         {t('prReview.selectedOfTotal', { selected: selectedCount, total: counts.total })}

apps/frontend/src/renderer/components/github-prs/components/ReviewFindings.tsx

@@ -7,6 +7,7 @@
  * - Quick select actions (Critical/High, All, None)
  * - Collapsible sections for less important findings
  * - Visual summary of finding counts
+ * - Disputed findings shown in a separate collapsible section
  */
 
 import { useState, useMemo } from 'react';
@@ -16,6 +17,9 @@ import {
   CheckSquare,
   Square,
   Send,
+  ChevronDown,
+  ChevronRight,
+  ShieldQuestion,
 } from 'lucide-react';
 import { useTranslation } from 'react-i18next';
 import { Button } from '../../ui/button';
@@ -47,6 +51,7 @@ export function ReviewFindings({
   const [expandedSections, setExpandedSections] = useState<Set<SeverityGroup>>(
     new Set<SeverityGroup>(['critical', 'high']) // Critical and High expanded by default
   );
+  const [disputedExpanded, setDisputedExpanded] = useState(false);
 
   // Filter out posted findings - only show unposted findings for selection
   const unpostedFindings = useMemo(() =>
@@ -54,10 +59,24 @@ export function ReviewFindings({
     [findings, postedIds]
   );
 
+  // Split unposted findings into active vs disputed (single pass)
+  const { activeFindings, disputedFindings } = useMemo(() => {
+    const active: PRReviewFinding[] = [];
+    const disputed: PRReviewFinding[] = [];
+    for (const finding of unpostedFindings) {
+      if (finding.validationStatus === 'dismissed_false_positive') {
+        disputed.push(finding);
+      } else {
+        active.push(finding);
+      }
+    }
+    return { activeFindings: active, disputedFindings: disputed };
+  }, [unpostedFindings]);
+
   // Check if all findings are posted
   const allFindingsPosted = findings.length > 0 && unpostedFindings.length === 0;
 
-  // Group unposted findings by severity (only show findings that haven't been posted)
+  // Group ACTIVE unposted findings by severity (disputed go in their own section)
   const groupedFindings = useMemo(() => {
     const groups: Record<SeverityGroup, PRReviewFinding[]> = {
       critical: [],
@@ -66,7 +85,7 @@ export function ReviewFindings({
       low: [],
     };
 
-    for (const finding of unpostedFindings) {
+    for (const finding of activeFindings) {
       const severity = finding.severity as SeverityGroup;
       if (groups[severity]) {
         groups[severity].push(finding);
@@ -74,20 +93,20 @@ export function ReviewFindings({
     }
 
     return groups;
-  }, [unpostedFindings]);
+  }, [activeFindings]);
 
-  // Count by severity (unposted findings only)
+  // Count by severity (active findings only)
   const counts = useMemo(() => ({
     critical: groupedFindings.critical.length,
     high: groupedFindings.high.length,
     medium: groupedFindings.medium.length,
     low: groupedFindings.low.length,
-    total: unpostedFindings.length,
+    total: activeFindings.length,
     important: groupedFindings.critical.length + groupedFindings.high.length,
     posted: postedIds.size,
-  }), [groupedFindings, unpostedFindings.length, postedIds.size]);
+  }), [groupedFindings, activeFindings.length, postedIds.size]);
 
-  // Selection hooks - use unposted findings only
+  // Selection hooks - use ACTIVE unposted findings only (Select All excludes disputed)
   const {
     toggleFinding,
     selectAll,
@@ -95,7 +114,7 @@ export function ReviewFindings({
     selectImportant,
     toggleSeverityGroup,
   } = useFindingSelection({
-    findings: unpostedFindings,
+    findings: activeFindings,
     selectedIds,
     onSelectionChange,
     groupedFindings,
@@ -114,6 +133,12 @@ export function ReviewFindings({
     });
   };
 
+  // Count only active findings that are selected (excludes disputed from count)
+  const selectedActiveCount = useMemo(
+    () => activeFindings.filter(f => selectedIds.has(f.id)).length,
+    [activeFindings, selectedIds]
+  );
+
   // When all findings have been posted, show a success message instead of the selection UI
   if (allFindingsPosted) {
     return (
@@ -131,10 +156,11 @@ export function ReviewFindings({
 
   return (
     <div className="space-y-4">
-      {/* Summary Stats Bar - show unposted findings only */}
+      {/* Summary Stats Bar - show active findings + disputed count */}
       <FindingsSummary
-        findings={unpostedFindings}
-        selectedCount={selectedIds.size}
+        findings={activeFindings}
+        selectedCount={selectedActiveCount}
+        disputedCount={disputedFindings.length}
       />
 
       {/* Quick Select Actions */}
@@ -170,7 +196,7 @@ export function ReviewFindings({
         </Button>
       </div>
 
-      {/* Grouped Findings (unposted only) */}
+      {/* Grouped Findings (active only) */}
       <div className="space-y-3">
         {SEVERITY_ORDER.map((severity) => {
           const group = groupedFindings[severity];
@@ -220,6 +246,48 @@ export function ReviewFindings({
         })}
       </div>
 
+      {/* Disputed Findings Section */}
+      {disputedFindings.length > 0 && (
+        <div className="rounded-lg border border-purple-500/20 bg-purple-500/5">
+          {/* Disputed Header */}
+          <button
+            type="button"
+            onClick={() => setDisputedExpanded(!disputedExpanded)}
+            aria-expanded={disputedExpanded}
+            className="w-full flex items-center gap-2 p-3 text-left hover:bg-purple-500/10 transition-colors rounded-t-lg"
+          >
+            {disputedExpanded ? (
+              <ChevronDown className="h-4 w-4 text-purple-500 shrink-0" />
+            ) : (
+              <ChevronRight className="h-4 w-4 text-purple-500 shrink-0" />
+            )}
+            <ShieldQuestion className="h-4 w-4 text-purple-500 shrink-0" />
+            <span className="text-sm font-medium text-purple-500">
+              {t('prReview.disputedByValidator', { count: disputedFindings.length })}
+            </span>
+          </button>
+
+          {/* Disputed Content */}
+          {disputedExpanded && (
+            <div className="p-3 pt-0 space-y-2">
+              <p className="text-xs text-muted-foreground italic mb-2">
+                {t('prReview.disputedSectionHint')}
+              </p>
+              {disputedFindings.map((finding) => (
+                <FindingItem
+                  key={finding.id}
+                  finding={finding}
+                  selected={selectedIds.has(finding.id)}
+                  posted={false}
+                  disputed
+                  onToggle={() => toggleFinding(finding.id)}
+                />
+              ))}
+            </div>
+          )}
+        </div>
+      )}
+
       {/* Empty State - no findings at all */}
       {findings.length === 0 && (
         <div className="text-center py-8 text-muted-foreground">

apps/frontend/src/renderer/components/github-prs/hooks/useFindingSelection.ts

@@ -30,21 +30,31 @@ export function useFindingSelection({
     onSelectionChange(next);
   }, [selectedIds, onSelectionChange]);
 
-  // Select all findings
+  // Select all findings (preserving any disputed selections not in active findings)
   const selectAll = useCallback(() => {
-    onSelectionChange(new Set(findings.map(f => f.id)));
-  }, [findings, onSelectionChange]);
+    const activeIds = new Set(findings.map(f => f.id));
+    // Preserve selections for disputed findings (IDs not in active findings list)
+    for (const id of selectedIds) {
+      if (!findings.some(f => f.id === id)) activeIds.add(id);
+    }
+    onSelectionChange(activeIds);
+  }, [findings, selectedIds, onSelectionChange]);
 
   // Clear all selections
   const selectNone = useCallback(() => {
     onSelectionChange(new Set());
   }, [onSelectionChange]);
 
-  // Select only critical and high severity findings
+  // Select only critical and high severity findings (preserving disputed selections)
   const selectImportant = useCallback(() => {
     const important = [...groupedFindings.critical, ...groupedFindings.high];
-    onSelectionChange(new Set(important.map(f => f.id)));
-  }, [groupedFindings, onSelectionChange]);
+    const importantIds = new Set(important.map(f => f.id));
+    // Preserve selections for disputed findings (IDs not in active findings list)
+    for (const id of selectedIds) {
+      if (!findings.some(f => f.id === id)) importantIds.add(id);
+    }
+    onSelectionChange(importantIds);
+  }, [groupedFindings, findings, selectedIds, onSelectionChange]);
 
   // Toggle entire severity group selection
   const toggleSeverityGroup = useCallback((severity: SeverityGroup) => {

apps/frontend/src/shared/i18n/locales/en/common.json

@@ -402,6 +402,10 @@
     "verifyChanges": "Verify Changes",
     "verifyAnyway": "Verify",
     "runFollowupAnyway": "Run follow-up verification even though no files overlap",
+    "disputed": "Disputed",
+    "disputedByValidator": "Disputed by Validator ({{count}})",
+    "crossValidatedBy": "Confirmed by {{count}} agents",
+    "disputedSectionHint": "These findings were reported by specialists but disputed by the validator. You can still select and post them.",
     "logs": {
       "agentActivity": "Agent Activity",
       "showMore": "Show {{count}} more",

apps/frontend/src/shared/i18n/locales/fr/common.json

@@ -402,6 +402,10 @@
     "blockedStatusMessageTitle": "## 🤖 Auto Claude PR Review",
     "blockedStatusMessageFooter": "*This review identified blockers that must be resolved before merge. Generated by Auto Claude.*",
     "failedPostBlockedStatus": "Échec de la publication du statut",
+    "disputed": "Contesté",
+    "disputedByValidator": "Contesté par le validateur ({{count}})",
+    "crossValidatedBy": "Confirmé par {{count}} agents",
+    "disputedSectionHint": "Ces résultats ont été signalés par les spécialistes mais contestés par le validateur. Vous pouvez toujours les sélectionner et les publier.",
     "logs": {
       "agentActivity": "Activité des agents",
       "showMore": "Afficher {{count}} de plus",

tests/agents/test_agent_flow.py

@@ -922,8 +922,8 @@ class TestQALoopStateTransitions:
     def test_qa_not_required_when_build_incomplete(self, test_env):
         """QA should not run when build is incomplete."""
         from qa_loop import save_implementation_plan
-        # Import the real is_build_complete to patch at the right level
-        from core.progress import is_build_complete as real_is_build_complete
+        # Import the real is_build_ready_for_qa to patch at the right level
+        from core.progress import is_build_ready_for_qa as real_is_build_ready_for_qa
 
         temp_dir, spec_dir, project_dir = test_env
 
@@ -943,16 +943,16 @@ class TestQALoopStateTransitions:
         }
         save_implementation_plan(spec_dir, plan)
 
-        # Patch is_build_complete where it's used (qa.criteria) to use real implementation
+        # Patch is_build_ready_for_qa where it's used (qa.criteria) to use real implementation
         # This is needed because test_qa_criteria.py module-level mocks may pollute
-        with patch('qa.criteria.is_build_complete', side_effect=real_is_build_complete):
+        with patch('qa.criteria.is_build_ready_for_qa', side_effect=real_is_build_ready_for_qa):
             from qa.criteria import should_run_qa
             assert should_run_qa(spec_dir) is False, "QA should not run with pending subtasks"
 
     def test_qa_required_when_build_complete(self, test_env):
         """QA should run when build is complete and not yet approved."""
         from qa_loop import save_implementation_plan
-        from core.progress import is_build_complete as real_is_build_complete
+        from core.progress import is_build_ready_for_qa as real_is_build_ready_for_qa
 
         temp_dir, spec_dir, project_dir = test_env
 
@@ -972,15 +972,15 @@ class TestQALoopStateTransitions:
         }
         save_implementation_plan(spec_dir, plan)
 
-        # Patch is_build_complete where it's used (qa.criteria) to use real implementation
-        with patch('qa.criteria.is_build_complete', side_effect=real_is_build_complete):
+        # Patch is_build_ready_for_qa where it's used (qa.criteria) to use real implementation
+        with patch('qa.criteria.is_build_ready_for_qa', side_effect=real_is_build_ready_for_qa):
             from qa.criteria import should_run_qa
             assert should_run_qa(spec_dir) is True, "QA should run when build complete"
 
     def test_qa_not_required_when_already_approved(self, test_env):
         """QA should not run when build is already approved."""
         from qa_loop import save_implementation_plan
-        from core.progress import is_build_complete as real_is_build_complete
+        from core.progress import is_build_ready_for_qa as real_is_build_ready_for_qa
 
         temp_dir, spec_dir, project_dir = test_env
 
@@ -1003,8 +1003,8 @@ class TestQALoopStateTransitions:
         }
         save_implementation_plan(spec_dir, plan)
 
-        # Patch is_build_complete where it's used (qa.criteria) to use real implementation
-        with patch('qa.criteria.is_build_complete', side_effect=real_is_build_complete):
+        # Patch is_build_ready_for_qa where it's used (qa.criteria) to use real implementation
+        with patch('qa.criteria.is_build_ready_for_qa', side_effect=real_is_build_ready_for_qa):
             from qa.criteria import should_run_qa
             assert should_run_qa(spec_dir) is False, "QA should not run when already approved"
 

tests/test_cli_qa_commands.py

@@ -284,7 +284,7 @@ class TestHandleQaCommand:
             )
 
         captured = capsys.readouterr()
-        assert "Build not complete" in captured.out
+        assert "Build not ready for QA" in captured.out
         assert "1/2" in captured.out
 
     def test_processes_human_feedback(

tests/test_progress_qa_readiness.py

@@ -0,0 +1,418 @@
+#!/usr/bin/env python3
+"""
+Tests for Progress Module - QA Readiness Check
+===============================================
+
+Tests the core/progress.py is_build_ready_for_qa() function which determines
+if a build has reached a terminal state (all subtasks completed, failed, or stuck).
+
+This function differs from is_build_complete() in that it considers builds with
+failed/stuck subtasks as ready for QA validation.
+"""
+
+import json
+import sys
+from pathlib import Path
+
+import pytest
+
+# Add parent directory to path for imports
+sys.path.insert(0, str(Path(__file__).parent.parent / "apps" / "backend"))
+
+from core.progress import is_build_ready_for_qa
+
+
+@pytest.fixture
+def spec_dir(tmp_path):
+    """Create a spec directory for testing."""
+    spec = tmp_path / "spec"
+    spec.mkdir()
+    return spec
+
+
+@pytest.fixture
+def memory_dir(spec_dir):
+    """Create a memory directory for attempt_history.json."""
+    memory = spec_dir / "memory"
+    memory.mkdir()
+    return memory
+
+
+class TestIsBuildReadyForQA:
+    """Tests for is_build_ready_for_qa function."""
+
+    def test_all_subtasks_completed(self, spec_dir: Path):
+        """Returns True when all subtasks are completed."""
+        plan = {
+            "feature": "Test Feature",
+            "phases": [
+                {
+                    "phase": 1,
+                    "name": "Phase 1",
+                    "subtasks": [
+                        {"id": "subtask-1-1", "status": "completed"},
+                        {"id": "subtask-1-2", "status": "completed"},
+                    ],
+                },
+                {
+                    "phase": 2,
+                    "name": "Phase 2",
+                    "subtasks": [
+                        {"id": "subtask-2-1", "status": "completed"},
+                    ],
+                },
+            ],
+        }
+        plan_file = spec_dir / "implementation_plan.json"
+        plan_file.write_text(json.dumps(plan))
+
+        result = is_build_ready_for_qa(spec_dir)
+        assert result is True
+
+    def test_mix_completed_and_pending(self, spec_dir: Path):
+        """Returns False when some subtasks are still pending."""
+        plan = {
+            "feature": "Test Feature",
+            "phases": [
+                {
+                    "phase": 1,
+                    "name": "Phase 1",
+                    "subtasks": [
+                        {"id": "subtask-1-1", "status": "completed"},
+                        {"id": "subtask-1-2", "status": "pending"},
+                    ],
+                },
+            ],
+        }
+        plan_file = spec_dir / "implementation_plan.json"
+        plan_file.write_text(json.dumps(plan))
+
+        result = is_build_ready_for_qa(spec_dir)
+        assert result is False
+
+    def test_mix_completed_and_failed(self, spec_dir: Path):
+        """Returns True when all subtasks are terminal (completed + failed)."""
+        plan = {
+            "feature": "Test Feature",
+            "phases": [
+                {
+                    "phase": 1,
+                    "name": "Phase 1",
+                    "subtasks": [
+                        {"id": "subtask-1-1", "status": "completed"},
+                        {"id": "subtask-1-2", "status": "failed"},
+                    ],
+                },
+                {
+                    "phase": 2,
+                    "name": "Phase 2",
+                    "subtasks": [
+                        {"id": "subtask-2-1", "status": "completed"},
+                        {"id": "subtask-2-2", "status": "failed"},
+                    ],
+                },
+            ],
+        }
+        plan_file = spec_dir / "implementation_plan.json"
+        plan_file.write_text(json.dumps(plan))
+
+        result = is_build_ready_for_qa(spec_dir)
+        assert result is True
+
+    def test_subtask_stuck_in_attempt_history(self, spec_dir: Path, memory_dir: Path):
+        """Returns True when subtask is marked stuck in attempt_history even if plan shows pending."""
+        plan = {
+            "feature": "Test Feature",
+            "phases": [
+                {
+                    "phase": 1,
+                    "name": "Phase 1",
+                    "subtasks": [
+                        {"id": "subtask-1-1", "status": "completed"},
+                        {"id": "subtask-1-2", "status": "pending"},  # Stuck but plan not updated
+                    ],
+                },
+            ],
+        }
+        plan_file = spec_dir / "implementation_plan.json"
+        plan_file.write_text(json.dumps(plan))
+
+        # Create attempt_history with stuck subtask
+        attempt_history = {
+            "stuck_subtasks": [
+                {
+                    "subtask_id": "subtask-1-2",
+                    "reason": "Circular fix after 3 attempts",
+                    "escalated_at": "2024-01-01T12:00:00Z",
+                    "attempt_count": 3,
+                }
+            ],
+            "subtasks": {},
+        }
+        history_file = memory_dir / "attempt_history.json"
+        history_file.write_text(json.dumps(attempt_history))
+
+        result = is_build_ready_for_qa(spec_dir)
+        assert result is True
+
+    def test_no_plan_file(self, spec_dir: Path):
+        """Returns False when implementation_plan.json doesn't exist."""
+        result = is_build_ready_for_qa(spec_dir)
+        assert result is False
+
+    def test_empty_phases(self, spec_dir: Path):
+        """Returns False when plan has no subtasks (total=0)."""
+        plan = {
+            "feature": "Test Feature",
+            "phases": [],
+        }
+        plan_file = spec_dir / "implementation_plan.json"
+        plan_file.write_text(json.dumps(plan))
+
+        result = is_build_ready_for_qa(spec_dir)
+        assert result is False
+
+    def test_phases_with_no_subtasks(self, spec_dir: Path):
+        """Returns False when phases exist but contain no subtasks."""
+        plan = {
+            "feature": "Test Feature",
+            "phases": [
+                {
+                    "phase": 1,
+                    "name": "Phase 1",
+                    "subtasks": [],
+                },
+            ],
+        }
+        plan_file = spec_dir / "implementation_plan.json"
+        plan_file.write_text(json.dumps(plan))
+
+        result = is_build_ready_for_qa(spec_dir)
+        assert result is False
+
+    def test_no_attempt_history_file(self, spec_dir: Path):
+        """Returns True based on plan file alone when attempt_history.json doesn't exist."""
+        plan = {
+            "feature": "Test Feature",
+            "phases": [
+                {
+                    "phase": 1,
+                    "name": "Phase 1",
+                    "subtasks": [
+                        {"id": "subtask-1-1", "status": "completed"},
+                        {"id": "subtask-1-2", "status": "failed"},
+                    ],
+                },
+            ],
+        }
+        plan_file = spec_dir / "implementation_plan.json"
+        plan_file.write_text(json.dumps(plan))
+
+        # No attempt_history.json created
+        result = is_build_ready_for_qa(spec_dir)
+        assert result is True
+
+    def test_invalid_json_in_attempt_history(self, spec_dir: Path, memory_dir: Path):
+        """Gracefully handles invalid JSON in attempt_history and falls back to plan-only check."""
+        plan = {
+            "feature": "Test Feature",
+            "phases": [
+                {
+                    "phase": 1,
+                    "name": "Phase 1",
+                    "subtasks": [
+                        {"id": "subtask-1-1", "status": "completed"},
+                    ],
+                },
+            ],
+        }
+        plan_file = spec_dir / "implementation_plan.json"
+        plan_file.write_text(json.dumps(plan))
+
+        # Create invalid JSON in attempt_history
+        history_file = memory_dir / "attempt_history.json"
+        history_file.write_text("{ invalid json }")
+
+        # Should fallback to plan-only check and return True
+        result = is_build_ready_for_qa(spec_dir)
+        assert result is True
+
+    def test_invalid_json_in_plan(self, spec_dir: Path):
+        """Returns False when implementation_plan.json contains invalid JSON."""
+        plan_file = spec_dir / "implementation_plan.json"
+        plan_file.write_text("{ invalid json }")
+
+        result = is_build_ready_for_qa(spec_dir)
+        assert result is False
+
+    def test_empty_plan_file(self, spec_dir: Path):
+        """Returns False when implementation_plan.json is empty."""
+        plan_file = spec_dir / "implementation_plan.json"
+        plan_file.write_text("")
+
+        result = is_build_ready_for_qa(spec_dir)
+        assert result is False
+
+    def test_multiple_stuck_subtasks(self, spec_dir: Path, memory_dir: Path):
+        """Returns True when multiple subtasks are stuck in attempt_history."""
+        plan = {
+            "feature": "Test Feature",
+            "phases": [
+                {
+                    "phase": 1,
+                    "name": "Phase 1",
+                    "subtasks": [
+                        {"id": "subtask-1-1", "status": "pending"},
+                        {"id": "subtask-1-2", "status": "pending"},
+                        {"id": "subtask-1-3", "status": "completed"},
+                    ],
+                },
+            ],
+        }
+        plan_file = spec_dir / "implementation_plan.json"
+        plan_file.write_text(json.dumps(plan))
+
+        # Mark two subtasks as stuck
+        attempt_history = {
+            "stuck_subtasks": [
+                {"subtask_id": "subtask-1-1", "reason": "Error 1"},
+                {"subtask_id": "subtask-1-2", "reason": "Error 2"},
+            ],
+            "subtasks": {},
+        }
+        history_file = memory_dir / "attempt_history.json"
+        history_file.write_text(json.dumps(attempt_history))
+
+        result = is_build_ready_for_qa(spec_dir)
+        assert result is True
+
+    def test_mix_of_all_terminal_states(self, spec_dir: Path, memory_dir: Path):
+        """Returns True with completed, failed, and stuck subtasks."""
+        plan = {
+            "feature": "Test Feature",
+            "phases": [
+                {
+                    "phase": 1,
+                    "name": "Phase 1",
+                    "subtasks": [
+                        {"id": "subtask-1-1", "status": "completed"},
+                        {"id": "subtask-1-2", "status": "failed"},
+                        {"id": "subtask-1-3", "status": "pending"},  # Will be stuck
+                    ],
+                },
+            ],
+        }
+        plan_file = spec_dir / "implementation_plan.json"
+        plan_file.write_text(json.dumps(plan))
+
+        attempt_history = {
+            "stuck_subtasks": [
+                {"subtask_id": "subtask-1-3", "reason": "Stuck"},
+            ],
+            "subtasks": {},
+        }
+        history_file = memory_dir / "attempt_history.json"
+        history_file.write_text(json.dumps(attempt_history))
+
+        result = is_build_ready_for_qa(spec_dir)
+        assert result is True
+
+    def test_in_progress_status(self, spec_dir: Path):
+        """Returns False when subtasks are in_progress."""
+        plan = {
+            "feature": "Test Feature",
+            "phases": [
+                {
+                    "phase": 1,
+                    "name": "Phase 1",
+                    "subtasks": [
+                        {"id": "subtask-1-1", "status": "completed"},
+                        {"id": "subtask-1-2", "status": "in_progress"},
+                    ],
+                },
+            ],
+        }
+        plan_file = spec_dir / "implementation_plan.json"
+        plan_file.write_text(json.dumps(plan))
+
+        result = is_build_ready_for_qa(spec_dir)
+        assert result is False
+
+    def test_missing_status_field(self, spec_dir: Path):
+        """Returns False when subtask has no status field (defaults to pending)."""
+        plan = {
+            "feature": "Test Feature",
+            "phases": [
+                {
+                    "phase": 1,
+                    "name": "Phase 1",
+                    "subtasks": [
+                        {"id": "subtask-1-1", "status": "completed"},
+                        {"id": "subtask-1-2"},  # No status field
+                    ],
+                },
+            ],
+        }
+        plan_file = spec_dir / "implementation_plan.json"
+        plan_file.write_text(json.dumps(plan))
+
+        result = is_build_ready_for_qa(spec_dir)
+        assert result is False
+
+    def test_stuck_subtask_without_id_field(self, spec_dir: Path, memory_dir: Path):
+        """Ignores stuck subtasks without subtask_id field in attempt_history."""
+        plan = {
+            "feature": "Test Feature",
+            "phases": [
+                {
+                    "phase": 1,
+                    "name": "Phase 1",
+                    "subtasks": [
+                        {"id": "subtask-1-1", "status": "pending"},
+                    ],
+                },
+            ],
+        }
+        plan_file = spec_dir / "implementation_plan.json"
+        plan_file.write_text(json.dumps(plan))
+
+        # Malformed stuck subtask entry without subtask_id
+        attempt_history = {
+            "stuck_subtasks": [
+                {"reason": "Error", "escalated_at": "2024-01-01T12:00:00Z"}
+            ],
+            "subtasks": {},
+        }
+        history_file = memory_dir / "attempt_history.json"
+        history_file.write_text(json.dumps(attempt_history))
+
+        # Should return False since subtask-1-1 is still pending
+        result = is_build_ready_for_qa(spec_dir)
+        assert result is False
+
+    def test_unicode_encoding_in_files(self, spec_dir: Path, memory_dir: Path):
+        """Handles UTF-8 encoded content correctly."""
+        plan = {
+            "feature": "Test Feature 测试功能",
+            "phases": [
+                {
+                    "phase": 1,
+                    "name": "Phase 1",
+                    "subtasks": [
+                        {"id": "subtask-1-1", "status": "completed", "notes": "完成"},
+                    ],
+                },
+            ],
+        }
+        plan_file = spec_dir / "implementation_plan.json"
+        plan_file.write_text(json.dumps(plan, ensure_ascii=False), encoding="utf-8")
+
+        attempt_history = {
+            "stuck_subtasks": [],
+            "subtasks": {},
+        }
+        history_file = memory_dir / "attempt_history.json"
+        history_file.write_text(json.dumps(attempt_history, ensure_ascii=False), encoding="utf-8")
+
+        result = is_build_ready_for_qa(spec_dir)
+        assert result is True

tests/test_qa_criteria.py

@@ -529,7 +529,7 @@ class TestShouldRunQA:
         plan = {"feature": "Test", "phases": []}
         save_implementation_plan(spec_dir, plan)
 
-        with patch('qa.criteria.is_build_complete', return_value=False):
+        with patch('qa.criteria.is_build_ready_for_qa', return_value=False):
             result = should_run_qa(spec_dir)
             assert result is False
 
@@ -540,15 +540,15 @@ class TestShouldRunQA:
         plan = {"feature": "Test", "qa_signoff": qa_signoff_approved}
         save_implementation_plan(spec_dir, plan)
 
-        with patch('qa.criteria.is_build_complete', return_value=True):
+        with patch('qa.criteria.is_build_ready_for_qa', return_value=True):
             result = should_run_qa(spec_dir)
             assert result is False
 
     def test_should_run_qa_build_complete_not_approved(self, spec_dir: Path):
         """Returns True when build complete but not approved."""
-        # Explicitly patch is_build_complete to return True
+        # Explicitly patch is_build_ready_for_qa to return True
         from unittest.mock import patch
-        with patch('qa.criteria.is_build_complete', return_value=True):
+        with patch('qa.criteria.is_build_ready_for_qa', return_value=True):
             plan = {"feature": "Test", "phases": []}
             save_implementation_plan(spec_dir, plan)
 
@@ -563,15 +563,15 @@ class TestShouldRunQA:
         plan = {"feature": "Test", "qa_signoff": qa_signoff_rejected}
         save_implementation_plan(spec_dir, plan)
 
-        with patch('qa.criteria.is_build_complete', return_value=True):
+        with patch('qa.criteria.is_build_ready_for_qa', return_value=True):
             result = should_run_qa(spec_dir)
             assert result is True
 
     def test_should_run_qa_no_plan(self, spec_dir: Path):
-        """Returns False when no plan exists (build not complete)."""
+        """Returns False when no plan exists (build not ready)."""
         from unittest.mock import patch
 
-        with patch('qa.criteria.is_build_complete', return_value=False):
+        with patch('qa.criteria.is_build_ready_for_qa', return_value=False):
             result = should_run_qa(spec_dir)
             assert result is False
 
@@ -905,7 +905,7 @@ class TestQAIntegration:
         save_implementation_plan(spec_dir, plan)
 
         # Should run QA
-        with patch('qa.criteria.is_build_complete', return_value=True):
+        with patch('qa.criteria.is_build_ready_for_qa', return_value=True):
             assert should_run_qa(spec_dir) is True
 
         # QA approves
@@ -917,7 +917,7 @@ class TestQAIntegration:
         save_implementation_plan(spec_dir, plan)
 
         # Should not run QA again or fixes
-        with patch('qa.criteria.is_build_complete', return_value=True):
+        with patch('qa.criteria.is_build_ready_for_qa', return_value=True):
             assert should_run_qa(spec_dir) is False
         assert should_run_fixes(spec_dir) is False
         assert is_qa_approved(spec_dir) is True
@@ -931,7 +931,7 @@ class TestQAIntegration:
         save_implementation_plan(spec_dir, plan)
 
         # Should run QA
-        with patch('qa.criteria.is_build_complete', return_value=True):
+        with patch('qa.criteria.is_build_ready_for_qa', return_value=True):
             assert should_run_qa(spec_dir) is True
 
         # QA rejects
@@ -979,5 +979,5 @@ class TestQAIntegration:
         # Should not run more fixes after max iterations
         assert should_run_fixes(spec_dir) is False
         # But QA can still be run (to re-check)
-        with patch('qa.criteria.is_build_complete', return_value=True):
+        with patch('qa.criteria.is_build_ready_for_qa', return_value=True):
             assert should_run_qa(spec_dir) is True

tests/test_recovery.py

@@ -250,6 +250,119 @@ def test_mark_subtask_stuck(test_env):
     assert history["status"] == "stuck", "Chunk status not updated to stuck"
 
 
+def test_mark_subtask_stuck_updates_plan(test_env):
+    """Test that mark_subtask_stuck updates implementation_plan.json status."""
+    temp_dir, spec_dir, project_dir = test_env
+
+    # Create implementation_plan.json with subtask in_progress
+    plan = {
+        "feature": "Test Feature",
+        "phases": [
+            {
+                "phase": 1,
+                "name": "Phase 1",
+                "subtasks": [
+                    {
+                        "id": "subtask-1-1",
+                        "description": "Implement feature A",
+                        "status": "in_progress",
+                    },
+                    {
+                        "id": "subtask-1-2",
+                        "description": "Implement feature B",
+                        "status": "completed",
+                    },
+                ],
+            },
+        ],
+    }
+    plan_file = spec_dir / "implementation_plan.json"
+    plan_file.write_text(json.dumps(plan, indent=2))
+
+    manager = RecoveryManager(spec_dir, project_dir)
+
+    # Record some attempts for subtask-1-1
+    manager.record_attempt("subtask-1-1", 1, False, "Try 1", "Error 1")
+    manager.record_attempt("subtask-1-1", 2, False, "Try 2", "Error 2")
+    manager.record_attempt("subtask-1-1", 3, False, "Try 3", "Error 3")
+
+    # Mark subtask-1-1 as stuck
+    reason = "Circular fix after 3 attempts"
+    manager.mark_subtask_stuck("subtask-1-1", reason)
+
+    # Verify plan file was updated
+    with open(plan_file, encoding="utf-8") as f:
+        updated_plan = json.load(f)
+
+    # Find the stuck subtask
+    subtask_1_1 = updated_plan["phases"][0]["subtasks"][0]
+    assert subtask_1_1["id"] == "subtask-1-1"
+    assert subtask_1_1["status"] == "failed", "Stuck subtask status should be 'failed'"
+    assert "actual_output" in subtask_1_1, "actual_output field should be added"
+    assert "Marked as stuck" in subtask_1_1["actual_output"], "actual_output should mention stuck status"
+    assert reason in subtask_1_1["actual_output"], "actual_output should include the reason"
+
+    # Verify other subtask was not affected
+    subtask_1_2 = updated_plan["phases"][0]["subtasks"][1]
+    assert subtask_1_2["id"] == "subtask-1-2"
+    assert subtask_1_2["status"] == "completed", "Other subtask status should be unchanged"
+
+
+def test_mark_subtask_stuck_plan_missing_subtask(test_env):
+    """Test mark_subtask_stuck when subtask doesn't exist in plan."""
+    temp_dir, spec_dir, project_dir = test_env
+
+    # Create plan without the subtask we'll mark as stuck
+    plan = {
+        "feature": "Test Feature",
+        "phases": [
+            {
+                "phase": 1,
+                "name": "Phase 1",
+                "subtasks": [
+                    {
+                        "id": "subtask-1-1",
+                        "description": "Implement feature A",
+                        "status": "completed",
+                    },
+                ],
+            },
+        ],
+    }
+    plan_file = spec_dir / "implementation_plan.json"
+    plan_file.write_text(json.dumps(plan, indent=2))
+
+    manager = RecoveryManager(spec_dir, project_dir)
+
+    # Mark a non-existent subtask as stuck
+    manager.mark_subtask_stuck("subtask-2-1", "Some error")
+
+    # Verify plan file was not corrupted
+    with open(plan_file, encoding="utf-8") as f:
+        updated_plan = json.load(f)
+
+    # Plan should remain unchanged
+    assert len(updated_plan["phases"]) == 1
+    assert len(updated_plan["phases"][0]["subtasks"]) == 1
+    assert updated_plan["phases"][0]["subtasks"][0]["status"] == "completed"
+
+
+def test_mark_subtask_stuck_plan_missing_file(test_env):
+    """Test mark_subtask_stuck when implementation_plan.json doesn't exist."""
+    temp_dir, spec_dir, project_dir = test_env
+
+    manager = RecoveryManager(spec_dir, project_dir)
+
+    # Record attempts and mark as stuck (should not crash)
+    manager.record_attempt("subtask-1", 1, False, "Try 1", "Error 1")
+    manager.mark_subtask_stuck("subtask-1", "Some error")
+
+    # Verify stuck status in attempt_history
+    stuck_subtasks = manager.get_stuck_subtasks()
+    assert len(stuck_subtasks) == 1
+    assert stuck_subtasks[0]["subtask_id"] == "subtask-1"
+
+
 def test_recovery_hints(test_env):
     """Test recovery hints generation."""
     temp_dir, spec_dir, project_dir = test_env

tests/test_structured_output_recovery.py

@@ -18,17 +18,22 @@ import pytest
 # services/ package at both apps/backend/services/ and runners/github/services/.
 # To avoid collision, add the github services dir directly and import bare module names.
 _backend_dir = Path(__file__).parent.parent / "apps" / "backend"
-_github_services_dir = _backend_dir / "runners" / "github" / "services"
+_github_runner_dir = _backend_dir / "runners" / "github"
+_github_services_dir = _github_runner_dir / "services"
 if str(_backend_dir) not in sys.path:
     sys.path.insert(0, str(_backend_dir))
+if str(_github_runner_dir) not in sys.path:
+    sys.path.insert(0, str(_github_runner_dir))
 if str(_github_services_dir) not in sys.path:
     sys.path.insert(0, str(_github_services_dir))
 
 from agents.tools_pkg.models import AGENT_CONFIGS
 from pydantic_models import (
+    ExtractedFindingSummary,
     FollowupExtractionResponse,
     ParallelFollowupResponse,
 )
+from recovery_utils import create_finding_from_summary
 from sdk_utils import RECOVERABLE_ERRORS
 
 
@@ -53,20 +58,38 @@ class TestFollowupExtractionResponse:
         assert resp.dismissed_finding_count == 0
 
     def test_full_valid_response(self):
-        """Accepts fully populated response."""
+        """Accepts fully populated response with ExtractedFindingSummary objects."""
         resp = FollowupExtractionResponse(
             verdict="READY_TO_MERGE",
             verdict_reasoning="All findings resolved",
             resolved_finding_ids=["NCR-001", "NCR-002"],
             unresolved_finding_ids=[],
-            new_finding_summaries=["HIGH: potential cleanup issue in batch_commands.py"],
+            new_finding_summaries=[
+                ExtractedFindingSummary(
+                    severity="HIGH",
+                    description="potential cleanup issue in batch_commands.py",
+                    file="apps/backend/cli/batch_commands.py",
+                    line=42,
+                )
+            ],
             confirmed_finding_count=1,
             dismissed_finding_count=1,
         )
         assert len(resp.resolved_finding_ids) == 2
         assert len(resp.new_finding_summaries) == 1
+        assert resp.new_finding_summaries[0].file == "apps/backend/cli/batch_commands.py"
+        assert resp.new_finding_summaries[0].line == 42
         assert resp.confirmed_finding_count == 1
 
+    def test_finding_summary_defaults(self):
+        """ExtractedFindingSummary defaults file='unknown' and line=0."""
+        summary = ExtractedFindingSummary(
+            severity="MEDIUM",
+            description="Some issue without location",
+        )
+        assert summary.file == "unknown"
+        assert summary.line == 0
+
     def test_schema_is_small(self):
         """Schema should be significantly smaller than ParallelFollowupResponse."""
         extraction_schema = json.dumps(
@@ -75,10 +98,11 @@ class TestFollowupExtractionResponse:
         followup_schema = json.dumps(
             ParallelFollowupResponse.model_json_schema()
         )
-        # Extraction schema should be less than half the size of the full schema
-        assert len(extraction_schema) < len(followup_schema) / 2, (
+        # Actual ratio is ~50.7% after adding ExtractedFindingSummary nesting.
+        # Threshold at 55% gives headroom while still guarding against schema bloat.
+        assert len(extraction_schema) < len(followup_schema) * 0.55, (
             f"Extraction schema ({len(extraction_schema)} chars) should be "
-            f"less than half of full schema ({len(followup_schema)} chars)"
+            f"less than 55% of full schema ({len(followup_schema)} chars)"
         )
 
     def test_all_verdict_values_accepted(self):
@@ -143,3 +167,81 @@ class TestAgentConfigRegistration:
         """Extraction agent should use low thinking (lightweight call)."""
         config = AGENT_CONFIGS["pr_followup_extraction"]
         assert config["thinking_default"] == "low"
+
+
+# ============================================================================
+# Test create_finding_from_summary with file/line params
+# ============================================================================
+
+
+class TestCreateFindingFromSummary:
+    """Tests for create_finding_from_summary with file/line support."""
+
+    def test_backward_compatible_defaults(self):
+        """Calling without file/line still produces file='unknown', line=0."""
+        finding = create_finding_from_summary("HIGH: some issue", 0)
+        assert finding.file == "unknown"
+        assert finding.line == 0
+        assert finding.severity.value == "high"
+
+    def test_file_and_line_passed_through(self):
+        """File and line params are used in the resulting finding."""
+        finding = create_finding_from_summary(
+            summary="Missing null check",
+            index=0,
+            file="src/parser.py",
+            line=42,
+        )
+        assert finding.file == "src/parser.py"
+        assert finding.line == 42
+
+    def test_severity_override(self):
+        """severity_override takes precedence over parsed severity."""
+        finding = create_finding_from_summary(
+            summary="HIGH: some issue",
+            index=0,
+            severity_override="CRITICAL",
+        )
+        assert finding.severity.value == "critical"
+
+    def test_severity_override_case_insensitive(self):
+        """severity_override works regardless of case."""
+        finding = create_finding_from_summary(
+            summary="some issue",
+            index=0,
+            severity_override="high",
+        )
+        assert finding.severity.value == "high"
+
+    def test_severity_override_invalid_falls_back(self):
+        """Invalid severity_override falls back to parsed severity."""
+        finding = create_finding_from_summary(
+            summary="LOW: minor issue",
+            index=0,
+            severity_override="UNKNOWN",
+        )
+        # Falls back to parsed "LOW" from summary
+        assert finding.severity.value == "low"
+
+    def test_id_prefix(self):
+        """Custom id_prefix is used in the finding ID."""
+        finding = create_finding_from_summary(
+            summary="some issue", index=0, id_prefix="FU"
+        )
+        assert finding.id.startswith("FU-")
+
+    def test_all_params_together(self):
+        """All new params work together correctly."""
+        finding = create_finding_from_summary(
+            summary="Regex issue in subtask title truncation",
+            index=3,
+            id_prefix="FU",
+            severity_override="MEDIUM",
+            file="apps/backend/agents/planner.py",
+            line=187,
+        )
+        assert finding.id.startswith("FU-")
+        assert finding.severity.value == "medium"
+        assert finding.file == "apps/backend/agents/planner.py"
+        assert finding.line == 187
+        assert "Regex issue" in finding.title