Feature/mvp phase2 verification

CLAUDE.md

@@ -82,13 +82,25 @@ refactron
 | 3 | `--dry-run` flag for `refactron autofix`; `generate_diff()` in `autofix/file_ops.py`; `AutoFixEngine.fix_file()` | ✅ Done |
 | 4 | Test fixture files in `tests/fixtures/` (6 files, 8 fixture validation tests) | ✅ Done |
 | 5 | Phase 1 gate: 758 tests green, self-analysis 96 files/0 crashes, added `--no-cache` flag to `analyze` | ✅ Done |
-| 6–15 | Verification Engine (`refactron/verification/`) | Pending |
+| 6-7 | VerificationResult/CheckResult data contracts, BaseCheck ABC, VerificationEngine skeleton | ✅ Done |
+| 8 | SyntaxVerifier (Check 1): ast.parse, CST roundtrip, dangerous calls, import count | ✅ Done |
+| 9 | ImportIntegrityVerifier (Check 2): removed imports, dotted access, new import resolution | ✅ Done |
+| 10 | TestSuiteGate (Check 3): reverse import graph, swap-and-restore, subprocess pytest | ✅ Done |
+| 11-12 | Wire into AutoFixEngine.fix_file(verify=True), --verify CLI flag, report.py | ✅ Done |
+| 13-15 | Full fixture validation, Phase 2 gate check (827 tests green) | ✅ Done |
 
 **Key new APIs (Day 1–3):**
 - `AnalysisSkipWarning` dataclass in `core/models.py` — surfaced on `AnalysisResult.semantic_skip_warnings`
 - `BackupManager.validate_backup_integrity(session_id)` → `(valid_paths, corrupt_paths)`
 - `generate_diff(original, modified, filename)` in `autofix/file_ops.py` — returns unified diff string
-- `AutoFixEngine.fix_file(file_path, issues, dry_run=True)` → `(fixed_code, diff_or_None)`
+- `AutoFixEngine.fix_file(file_path, issues, dry_run=True, verify=False)` → `(fixed_code, diff_or_None)`
+
+**Key new APIs (Phase 2 — Verification Engine):**
+- `VerificationEngine(project_root).verify(original, transformed, file_path)` → `VerificationResult`
+- `CheckResult` / `VerificationResult` — frozen dataclasses in `verification/result.py`
+- `BaseCheck` ABC in `verification/engine.py` — 3 implementations: `SyntaxVerifier`, `ImportIntegrityVerifier`, `TestSuiteGate`
+- `format_verification_result(result, console)` — Rich CLI output in `verification/report.py`
+- `--verify` flag on `refactron autofix` — runs verification pipeline before applying fixes
 
 **CLI output overhaul:**
 - `refactron analyze` now shows an **interactive issue viewer** (TTY) with severity-grouped navigation (`[1-4]` to drill in, `[n/p/b/q]` to navigate)

refactron/autofix/engine.py

@@ -145,6 +145,7 @@ def fix_file(
         file_path: Path,
         issues: List[CodeIssue],
         dry_run: bool = True,
+        verify: bool = False,
     ) -> Tuple[str, Optional[str]]:
         """
         Apply all fixable issues to a file.
@@ -153,10 +154,15 @@ def fix_file(
         unified diff are returned for display only.  In dry_run=False mode the
         fixed content is written atomically (temp-file → os.replace).
 
+        When verify=True the VerificationEngine is invoked after generating
+        the diff.  If verification blocks the transform, (original_code, None)
+        is returned and no bytes are written.
+
         Args:
             file_path: Path to the Python file to fix.
             issues: List of CodeIssue objects to attempt to fix.
             dry_run: When True, no bytes are written to disk.
+            verify: When True, run VerificationEngine before writing.
 
         Returns:
             Tuple of (fixed_code, diff).  diff is None/empty when no changes
@@ -176,6 +182,19 @@ def fix_file(
 
         diff = generate_diff(code, current_code, file_path.name)
 
+        # Verification gate
+        if verify and current_code != code:
+            import logging
+
+            from refactron.verification import VerificationEngine
+
+            logger = logging.getLogger(__name__)
+            ve = VerificationEngine(project_root=file_path.parent)
+            vr = ve.verify(code, current_code, file_path)
+            if not vr.safe_to_apply:
+                logger.warning("Verification blocked %s: %s", file_path, vr.blocking_reason)
+                return code, None
+
         if not dry_run and current_code != code:
             # Atomic write: temp file in same directory → os.replace
             tmp_fd, tmp_path = tempfile.mkstemp(

refactron/cli/refactor.py

@@ -255,6 +255,12 @@ def refactor(
     default="safe",
     help="Maximum risk level for automatic fixes",
 )
+@click.option(
+    "--verify",
+    is_flag=True,
+    default=False,
+    help="Run verification checks (syntax, imports, tests) before applying fixes",
+)
 def autofix(
     target: str,
     config: Optional[str],
@@ -263,6 +269,7 @@ def autofix(
     preview: bool,
     dry_run: bool,
     safety_level: str,
+    verify: bool,
 ) -> None:
     """
     Automatically fix code issues (Phase 3 feature).

refactron/verification/__init__.py

@@ -0,0 +1,15 @@
+"""Verification Engine — proves code transforms are safe before writing."""
+
+from refactron.verification.checks import ImportIntegrityVerifier, SyntaxVerifier, TestSuiteGate
+from refactron.verification.engine import BaseCheck, VerificationEngine
+from refactron.verification.result import CheckResult, VerificationResult
+
+__all__ = [
+    "BaseCheck",
+    "CheckResult",
+    "ImportIntegrityVerifier",
+    "SyntaxVerifier",
+    "TestSuiteGate",
+    "VerificationEngine",
+    "VerificationResult",
+]

refactron/verification/checks/__init__.py

@@ -0,0 +1,7 @@
+"""Verification checks package."""
+
+from refactron.verification.checks.imports import ImportIntegrityVerifier
+from refactron.verification.checks.syntax import SyntaxVerifier
+from refactron.verification.checks.test_gate import TestSuiteGate
+
+__all__ = ["ImportIntegrityVerifier", "SyntaxVerifier", "TestSuiteGate"]

refactron/verification/checks/imports.py

@@ -0,0 +1,128 @@
+"""ImportIntegrityVerifier — Check 2: import removal, resolution, cycle detection."""
+
+import ast
+import importlib.util
+import time
+from pathlib import Path
+from typing import Any, Dict, Set
+
+from refactron.verification.engine import BaseCheck
+from refactron.verification.result import CheckResult
+
+
+class ImportIntegrityVerifier(BaseCheck):
+    """Validates import integrity after a transform."""
+
+    name = "import_integrity"
+
+    def verify(self, original: str, transformed: str, file_path: Path) -> CheckResult:
+        start = time.monotonic()
+        details: Dict[str, Any] = {}
+
+        orig_imports = self._extract_import_names(original)
+        trans_imports = self._extract_import_names(transformed)
+
+        # Step 1-2: removed imports still referenced
+        removed = orig_imports - trans_imports
+        if removed:
+            still_used = self._find_references(transformed, removed)
+            if still_used:
+                names = ", ".join(sorted(still_used))
+                details["removed_but_used"] = sorted(still_used)
+                return self._fail(
+                    f"Import(s) removed but still used: {names}",
+                    start,
+                    details,
+                )
+
+        # Step 3: new imports resolvable
+        # TODO: Step 4 (cycle detection) deferred — requires project-wide import graph
+        #       which is expensive for MVP. Will add in Phase 3 if needed.
+        added = trans_imports - orig_imports
+        if added:
+            orig_modules = self._extract_module_names(original)
+            trans_modules = self._extract_module_names(transformed)
+            new_modules = trans_modules - orig_modules
+            for mod in new_modules:
+                top_level = mod.split(".")[0]
+                if importlib.util.find_spec(top_level) is None:
+                    details["unresolvable_import"] = mod
+                    return self._fail(
+                        f"New import '{mod}' cannot be resolved",
+                        start,
+                        details,
+                    )
+
+        elapsed = int((time.monotonic() - start) * 1000)
+        return CheckResult(
+            check_name=self.name,
+            passed=True,
+            blocking_reason="",
+            confidence=1.0,
+            duration_ms=elapsed,
+            details=details,
+        )
+
+    def _fail(self, reason: str, start: float, details: Dict[str, Any]) -> CheckResult:
+        elapsed = int((time.monotonic() - start) * 1000)
+        return CheckResult(
+            check_name=self.name,
+            passed=False,
+            blocking_reason=reason,
+            confidence=0.0,
+            duration_ms=elapsed,
+            details=details,
+        )
+
+    @staticmethod
+    def _extract_import_names(code: str) -> Set[str]:
+        """Extract all locally-bound names from import statements."""
+        names: Set[str] = set()
+        try:
+            tree = ast.parse(code)
+        except SyntaxError:
+            return names
+        for node in ast.walk(tree):
+            if isinstance(node, ast.Import):
+                for alias in node.names:
+                    names.add(alias.asname or alias.name)
+            elif isinstance(node, ast.ImportFrom):
+                for alias in node.names:
+                    names.add(alias.asname or alias.name)
+        return names
+
+    @staticmethod
+    def _extract_module_names(code: str) -> Set[str]:
+        """Extract actual module paths from import statements."""
+        modules: Set[str] = set()
+        try:
+            tree = ast.parse(code)
+        except SyntaxError:
+            return modules
+        for node in ast.walk(tree):
+            if isinstance(node, ast.Import):
+                for alias in node.names:
+                    modules.add(alias.name)
+            elif isinstance(node, ast.ImportFrom) and node.module:
+                modules.add(node.module)
+        return modules
+
+    @staticmethod
+    def _find_references(code: str, names: Set[str]) -> Set[str]:
+        """Find which of the given names are still referenced in code body."""
+        referenced: Set[str] = set()
+        try:
+            tree = ast.parse(code)
+        except SyntaxError:
+            return referenced
+
+        for node in ast.walk(tree):
+            # Skip import nodes themselves
+            if isinstance(node, (ast.Import, ast.ImportFrom)):
+                continue
+            if isinstance(node, ast.Name) and node.id in names:
+                referenced.add(node.id)
+            elif isinstance(node, ast.Attribute) and isinstance(node.value, ast.Name):
+                if node.value.id in names:
+                    referenced.add(node.value.id)
+        return referenced

refactron/verification/checks/syntax.py

@@ -0,0 +1,117 @@
+"""SyntaxVerifier — Check 1: syntax validation, CST roundtrip, dangerous calls."""
+
+import ast
+import time
+from pathlib import Path
+from typing import Any, Dict, Set
+
+from refactron.verification.engine import BaseCheck
+from refactron.verification.result import CheckResult
+
+
+class SyntaxVerifier(BaseCheck):
+    """Validates that transformed code has valid syntax and no new dangerous calls."""
+
+    name = "syntax"
+
+    def verify(self, original: str, transformed: str, file_path: Path) -> CheckResult:
+        start = time.monotonic()
+        details: Dict[str, Any] = {}
+
+        # Step 1: ast.parse
+        try:
+            ast.parse(transformed, filename=str(file_path))
+        except SyntaxError as e:
+            return self._fail(
+                f"SyntaxError: {e.msg} (line {e.lineno})",
+                start,
+                details,
+            )
+
+        # Step 2: libcst roundtrip
+        try:
+            import libcst
+
+            tree = libcst.parse_module(transformed)
+            roundtripped = tree.code
+            libcst.parse_module(roundtripped)
+        except Exception as e:
+            return self._fail(
+                f"CST round-trip failed: {e}",
+                start,
+                details,
+            )
+
+        # Step 3: new dangerous calls
+        original_calls = self._find_dangerous_calls(original)
+        transformed_calls = self._find_dangerous_calls(transformed)
+        new_calls = transformed_calls - original_calls
+        if new_calls:
+            details["new_dangerous_calls"] = sorted(new_calls)
+            return self._fail(
+                f"New dangerous call(s) introduced: {', '.join(sorted(new_calls))}",
+                start,
+                details,
+            )
+
+        # Step 4: import count comparison (full reference check is in ImportIntegrityVerifier)
+        orig_import_count = self._count_imports(original)
+        trans_import_count = self._count_imports(transformed)
+        if trans_import_count < orig_import_count:
+            details["imports_removed"] = orig_import_count - trans_import_count
+
+        elapsed = int((time.monotonic() - start) * 1000)
+        return CheckResult(
+            check_name=self.name,
+            passed=True,
+            blocking_reason="",
+            confidence=1.0,
+            duration_ms=elapsed,
+            details=details,
+        )
+
+    def _fail(self, reason: str, start: float, details: Dict[str, Any]) -> CheckResult:
+        elapsed = int((time.monotonic() - start) * 1000)
+        return CheckResult(
+            check_name=self.name,
+            passed=False,
+            blocking_reason=reason,
+            confidence=0.0,
+            duration_ms=elapsed,
+            details=details,
+        )
+
+    @staticmethod
+    def _find_dangerous_calls(code: str) -> Set[str]:
+        """Find all calls to eval/exec/os.system in the code."""
+        dangerous: Set[str] = set()
+        try:
+            tree = ast.parse(code)
+        except SyntaxError:
+            return dangerous
+
+        for node in ast.walk(tree):
+            if isinstance(node, ast.Call):
+                if isinstance(node.func, ast.Name) and node.func.id in ("eval", "exec"):
+                    dangerous.add(node.func.id)
+                elif isinstance(node.func, ast.Attribute):
+                    if (
+                        isinstance(node.func.value, ast.Name)
+                        and node.func.value.id == "os"
+                        and node.func.attr == "system"
+                    ):
+                        dangerous.add("os.system")
+        return dangerous
+
+    @staticmethod
+    def _count_imports(code: str) -> int:
+        """Count import statements in code."""
+        count = 0
+        try:
+            tree = ast.parse(code)
+        except SyntaxError:
+            return 0
+        for node in ast.walk(tree):
+            if isinstance(node, (ast.Import, ast.ImportFrom)):
+                count += 1
+        return count