fix: resolve QA validation deadlock when subtasks are stuck or failed

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 @@
     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 @@
     "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 @@
 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 @@
     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/services/recovery.py

@@ -21,6 +21,8 @@
 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 @@ def mark_subtask_stuck(self, subtask_id: str, reason: str) -> None:
 
         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.

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 @@ def test_qa_not_required_when_build_incomplete(self, test_env):
         }
         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 @@ def test_qa_required_when_build_complete(self, test_env):
         }
         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 @@ def test_qa_not_required_when_already_approved(self, test_env):
         }
         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 @@ def test_incomplete_build_message(
             )
 
         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(