feat(core): verification gate wrapper with quick fixes and escalation

codeframe/core/adapters/verification_wrapper.py

@@ -1,32 +1,53 @@
-"""Verification gate wrapper for agent adapters."""
+"""Verification gate wrapper for agent adapters.
+
+Wraps any AgentAdapter with post-execution verification gates, quick fixes,
+fix attempt tracking, and escalation to blockers. This gives all execution
+engines (built-in and external) the same self-correction capabilities that
+ReactAgent has internally.
+"""
 
 from __future__ import annotations
 
+import logging
 from pathlib import Path
 from typing import Callable, Optional
 
 from codeframe.core.adapters.agent_adapter import AgentAdapter, AgentEvent, AgentResult
+from codeframe.core import blockers
+from codeframe.core.fix_tracker import (
+    EscalationDecision,
+    FixAttemptTracker,
+    FixOutcome,
+    build_escalation_question,
+)
 from codeframe.core.gates import GateStatus
 from codeframe.core.gates import run as run_gates
+from codeframe.core.quick_fixes import apply_quick_fix, find_quick_fix
 from codeframe.core.workspace import Workspace
 
+logger = logging.getLogger(__name__)
+
 
 class VerificationWrapper:
     """Wraps any AgentAdapter with post-execution verification gates.
 
     After the inner adapter completes, runs verification gates (pytest, ruff, etc.).
-    If gates fail, re-invokes the adapter with error context for self-correction,
-    up to max_correction_rounds times.
+    If gates fail:
+    1. Try a pattern-based quick fix (no LLM needed)
+    2. If quick fix applied, re-run gates immediately
+    3. If no quick fix, re-invoke adapter with error context for self-correction
+    4. Track all fix attempts to detect loops
+    5. Escalate to blocker when fix tracker recommends it or retries exhausted
 
-    This is the same self-correction loop that ReactAgent._run_final_verification()
+    This is the same self-correction pattern that ReactAgent._run_final_verification()
     uses, but decoupled from any specific engine so it wraps any adapter.
     """
 
     def __init__(
         self,
         inner: AgentAdapter,
         workspace: Workspace,
-        max_correction_rounds: int = 3,
+        max_correction_rounds: int = 5,
         gate_names: Optional[list[str]] = None,
         verbose: bool = False,
     ) -> None:
@@ -35,6 +56,7 @@ def __init__(
         self._max_correction_rounds = max_correction_rounds
         self._gate_names = gate_names  # None = use default gates
         self._verbose = verbose
+        self.fix_tracker = FixAttemptTracker()
 
     @property
     def name(self) -> str:
@@ -78,7 +100,12 @@ def run(
                     on_event(AgentEvent(type="verification_passed", data={}))
                 return result
 
-            # Gates failed -- build correction prompt and re-invoke
+            # Gates failed — get structured error summary
+            error_summary = (
+                gate_result.get_error_summary() or
+                self._format_gate_errors(gate_result)
+            )
+
             if on_event:
                 on_event(AgentEvent(
                     type="verification_failed",
@@ -88,12 +115,40 @@ def run(
                     },
                 ))
 
-            error_summary = self._format_gate_errors(gate_result)
+            # 1. Record the attempt (outcome deferred until we know if fix works)
+            self.fix_tracker.record_attempt(error_summary, "verification_gate")
+
+            # 2. Check escalation based on prior history
+            escalation = self.fix_tracker.should_escalate(error_summary)
+            if escalation.should_escalate:
+                self.fix_tracker.record_outcome(
+                    error_summary, "verification_gate", FixOutcome.FAILED,
+                )
+                return self._create_escalation_blocker(
+                    task_id, error_summary, escalation,
+                    last_output=result.output,
+                )
+
+            # 3. Try quick fix first (no adapter re-invocation needed)
+            if self._try_quick_fix(error_summary):
+                self.fix_tracker.record_outcome(
+                    error_summary, "verification_gate", FixOutcome.SUCCESS,
+                )
+                self._verbose_print(
+                    f"[VerificationWrapper] Quick fix applied (round {round_num + 1})"
+                )
+                continue  # Re-run gates without re-invoking adapter
+
+            # 4. No quick fix — record failure and re-invoke adapter with error context
+            self.fix_tracker.record_outcome(
+                error_summary, "verification_gate", FixOutcome.FAILED,
+            )
+            formatted_errors = self._format_gate_errors(gate_result)
             correction_prompt = (
                 f"{prompt}\n\n"
                 f"## Verification Gate Failures (Correction Round {round_num + 1})\n\n"
                 f"Your previous changes failed the following verification gates. "
-                f"Fix these issues:\n\n{error_summary}"
+                f"Fix these issues:\n\n{formatted_errors}"
             )
 
             result = self._inner.run(
@@ -113,17 +168,91 @@ def run(
         if gate_result.passed:
             return result
 
-        # All rounds exhausted, gates still failing
+        # All rounds exhausted — create blocker
         error_summary = self._format_gate_errors(gate_result)
+        return self._create_exhaustion_blocker(
+            task_id, error_summary, last_output=result.output,
+        )
+
+    def _try_quick_fix(self, error_summary: str) -> bool:
+        """Attempt a pattern-based quick fix for the gate error.
+
+        Returns True if a fix was successfully applied.
+        """
+        fix = find_quick_fix(error_summary, repo_path=self._workspace.repo_path)
+        if fix is None:
+            return False
+
+        success, msg = apply_quick_fix(fix, self._workspace.repo_path)
+        if success:
+            self._verbose_print(f"[VerificationWrapper] Quick fix: {msg}")
+        return success
+
+    def _create_escalation_blocker(
+        self,
+        task_id: str,
+        error: str,
+        escalation: EscalationDecision,
+        last_output: str = "",
+    ) -> AgentResult:
+        """Create a blocker when fix tracker recommends escalation."""
+        question = build_escalation_question(
+            error, escalation.reason, self.fix_tracker,
+        )
+
+        try:
+            blockers.create(
+                workspace=self._workspace,
+                question=question,
+                task_id=task_id,
+            )
+        except Exception:
+            logger.warning("Failed to create escalation blocker", exc_info=True)
+
         return AgentResult(
-            status="failed",
-            output=result.output,
+            status="blocked",
+            output=last_output,
+            blocker_question=question,
+            error=f"Escalated to blocker: {escalation.reason}",
+        )
+
+    def _create_exhaustion_blocker(
+        self,
+        task_id: str,
+        error_summary: str,
+        last_output: str = "",
+    ) -> AgentResult:
+        """Create a blocker when all correction rounds are exhausted."""
+        question = (
+            f"Verification gates still failing after "
+            f"{self._max_correction_rounds} correction rounds.\n\n"
+            f"Errors:\n{error_summary[:500]}\n\n"
+            f"Please investigate and provide guidance."
+        )
+
+        try:
+            blockers.create(
+                workspace=self._workspace,
+                question=question,
+                task_id=task_id,
+            )
+        except Exception:
+            logger.warning("Failed to create exhaustion blocker", exc_info=True)
+
+        return AgentResult(
+            status="blocked",
+            output=last_output,
+            blocker_question=question,
             error=(
                 f"Verification gates still failing after "
                 f"{self._max_correction_rounds} correction rounds:\n{error_summary}"
             ),
         )
 
+    def _verbose_print(self, msg: str) -> None:
+        if self._verbose:
+            print(msg)
+
     @staticmethod
     def _format_gate_errors(gate_result) -> str:
         """Format gate check failures into a readable summary."""

codeframe/core/fix_tracker.py

@@ -453,3 +453,28 @@ def from_dict(cls, data: dict) -> "FixAttemptTracker":
         tracker._file_counts = dict(data.get("file_counts", {}))
 
         return tracker
+
+
+def build_escalation_question(
+    error: str,
+    escalation_reason: str,
+    fix_tracker: FixAttemptTracker,
+) -> str:
+    """Build a human-readable blocker question for escalation.
+
+    Shared by VerificationWrapper and ReactAgent to produce consistent
+    escalation blocker messages.
+    """
+    context = fix_tracker.get_blocker_context(error)
+    attempted = context.get("attempted_fixes", [])
+    attempted_str = (
+        "\n".join(f"  - {f}" for f in attempted) if attempted else "  (none)"
+    )
+    return (
+        f"Verification keeps failing and automated fixes are not working.\n\n"
+        f"Error: {error[:300]}\n\n"
+        f"Reason for escalation: {escalation_reason}\n\n"
+        f"Fixes already attempted:\n{attempted_str}\n\n"
+        f"Total failures in this run: {context.get('total_run_failures', 0)}\n\n"
+        f"Please investigate and provide guidance."
+    )

codeframe/core/react_agent.py

@@ -22,7 +22,12 @@
 from codeframe.core.blocker_detection import classify_error_for_blocker
 from codeframe.core.context import ContextLoader, TaskContext
 from codeframe.core.events import EventType
-from codeframe.core.fix_tracker import EscalationDecision, FixAttemptTracker, FixOutcome
+from codeframe.core.fix_tracker import (
+    EscalationDecision,
+    FixAttemptTracker,
+    FixOutcome,
+    build_escalation_question,
+)
 from codeframe.core.stall_detector import StallAction, StallDetectedError
 from codeframe.core.stall_monitor import StallEvent, StallMonitor
 from codeframe.core.models import AgentPhase, CompletionEvent, ErrorEvent, ProgressEvent
@@ -1029,17 +1034,8 @@ def _create_escalation_blocker(
         the run record to the blocker.  If creation fails the exception
         propagates — callers in ``run()`` catch it and return FAILED.
         """
-        context = self.fix_tracker.get_blocker_context(error)
-        attempted = context.get("attempted_fixes", [])
-        attempted_str = "\n".join(f"  - {f}" for f in attempted) if attempted else "  (none)"
-
-        question = (
-            f"Verification keeps failing and automated fixes are not working.\n\n"
-            f"Error: {error[:300]}\n\n"
-            f"Reason for escalation: {escalation.reason}\n\n"
-            f"Fixes already attempted:\n{attempted_str}\n\n"
-            f"Total failures in this run: {context.get('total_run_failures', 0)}\n\n"
-            f"Please investigate and provide guidance."
+        question = build_escalation_question(
+            error, escalation.reason, self.fix_tracker,
         )
         blocker = blockers.create(
             workspace=self.workspace,

codeframe/core/runtime.py

@@ -692,7 +692,7 @@ def on_adapter_event(event: AdapterEvent) -> None:
 
             adapter = get_external_adapter(engine)
             wrapper = VerificationWrapper(
-                adapter, workspace, max_correction_rounds=3, verbose=verbose,
+                adapter, workspace, max_correction_rounds=5, verbose=verbose,
             )
 
             result = wrapper.run(