Fix/resilience qol

app/commands/move/move_apply.py

@@ -48,16 +48,29 @@ def apply_file_move(
 
     Path(dest_abs).parent.mkdir(parents=True, exist_ok=True)
     written_files: dict[str, str] = {}
+
+    # 1. Snapshot all original file contents explicitly BEFORE the filesystem mutates
+    # We do NOT snapshot source_abs into written_files because the rollback
+    # strategy is handled separately by `_rollback_move_target`. We only snapshot
+    # files whose contents we are going to modify in place.
     try:
+        for filepath in importer_changes:
+            if filepath in new_contents and Path(filepath).exists():
+                written_files[filepath] = Path(filepath).read_text()
+
+        # 2. Execute the filesystem move
         shutil.move(source_abs, dest_abs)
 
+        # 3. Apply memory modifications to the new destination
         if dest_abs in new_contents:
-            written_files[dest_abs] = Path(dest_abs).read_text()
+            # We don't snapshot dest_abs before this write because if a rollback happens, 
+            # `dest_abs` ceases to exist (it gets moved back to source_abs by `_rollback_move_target`).
+            # In fact, writing to dest_abs during rollback would recreate an orphaned file.
             safe_write_text(dest_abs, new_contents[dest_abs])
 
+        # 4. Apply rest of imports
         for filepath in importer_changes:
             if filepath in new_contents:
-                written_files[filepath] = Path(filepath).read_text()
                 safe_write_text(filepath, new_contents[filepath])
 
     except (OSError, UnicodeDecodeError, shutil.Error) as ex:
@@ -78,25 +91,33 @@ def apply_directory_move(
     """Move a directory and apply external/internal import replacements."""
     Path(dest_abs).parent.mkdir(parents=True, exist_ok=True)
     written_files: dict[str, str] = {}
+
     try:
+        # Snapshot content files prior to mutating
+        for filepath, replacements in external_changes.items():
+            if Path(filepath).exists():
+                written_files[filepath] = Path(filepath).read_text()
+
         shutil.move(source_abs, dest_abs)
 
         for src_file, changes in internal_changes.items():
             rel_in_dir = Path(src_file).relative_to(source_path)
             dest_file = Path(dest_abs) / rel_in_dir
+            if not dest_file.exists():
+                continue
             original = dest_file.read_text()
             content = original
             for old_str, new_str in changes:
                 content = content.replace(old_str, new_str)
-            written_files[str(dest_file)] = original
             safe_write_text(dest_file, content)
 
         for filepath, replacements in external_changes.items():
-            original = Path(filepath).read_text()
+            original = written_files.get(filepath)
+            if original is None:
+                continue
             content = original
             for old_str, new_str in replacements:
                 content = content.replace(old_str, new_str)
-            written_files[filepath] = original
             safe_write_text(filepath, content)
 
     except (OSError, UnicodeDecodeError, shutil.Error) as ex:

app/commands/plan_cmd.py

@@ -11,11 +11,11 @@
 from app.commands.helpers.rendering import print_agent_plan
 from app.commands.helpers.runtime import command_runtime
 from app.commands.helpers.state import require_completed_scan
-from core.fallbacks import warn_best_effort
-from engine import planning as planning_mod
 from core.discovery_api import safe_write_text
+from core.fallbacks import warn_best_effort
 from core.output_api import colorize
 from core.tooling import check_config_staleness
+from engine import planning as planning_mod
 
 
 def cmd_plan_output(args: argparse.Namespace) -> None:

app/commands/scan/scan_workflow.py

@@ -487,8 +487,9 @@ def merge_scan_results(
 
         if dirty:
             save_plan(plan, plan_path)
-    except Exception:
-        pass  # Plan reconciliation is best-effort
+    except Exception as exc:
+        import logging
+        logging.getLogger(__name__).warning("Plan reconciliation failed: %s", exc)
 
     return ScanMergeResult(
         diff=diff,

app/commands/show/payload.py

@@ -27,9 +27,9 @@ def build_show_payload(
     by_detector: dict[str, int] = defaultdict(int)
     by_tier: dict[int, int] = defaultdict(int)
     for finding in matches:
-        by_file[finding["file"]].append(finding)
-        by_detector[finding["detector"]] += 1
-        by_tier[finding["tier"]] += 1
+        by_file[finding.get("file", "unknown")].append(finding)
+        by_detector[finding.get("detector", "unknown")] += 1
+        by_tier[finding.get("tier", 5)] += 1
 
     payload = {
         "query": pattern,
@@ -43,10 +43,10 @@ def build_show_payload(
         "by_file": {
             fp: [
                 {
-                    "id": f["id"],
-                    "tier": f["tier"],
-                    "confidence": f["confidence"],
-                    "summary": f["summary"],
+                    "id": f.get("id", "tmp-id"),
+                    "tier": f.get("tier", 5),
+                    "confidence": f.get("confidence", "unknown"),
+                    "summary": f.get("summary", "No summary provided"),
                     "detail": f.get("detail", {}),
                 }
                 for f in fs

core/file_paths.py

@@ -11,27 +11,25 @@
 
 def matches_exclusion(rel_path: str, exclusion: str) -> bool:
     """Check if a relative path matches an exclusion pattern."""
-    normalized_path = rel_path.lstrip("./")
-    parts = Path(normalized_path).parts
-    if exclusion in parts:
+    import fnmatch
+
+    normalized_path = rel_path.lstrip("./").replace("\\", "/")
+    normalized_excl = exclusion.lstrip("./").replace("\\", "/")
+    parts = tuple(part for part in normalized_path.split("/") if part)
+    if normalized_excl in parts:
         return True
-    if "*" in exclusion:
-        import fnmatch
 
-        if any(fnmatch.fnmatch(part, exclusion) for part in parts):
+    if "*" in normalized_excl:
+        if any(fnmatch.fnmatch(part, normalized_excl) for part in parts):
             return True
-        # Full-path glob match for patterns with directory separators
-        # (e.g. "Wan2GP/**" should match "Wan2GP/models/rf.py").
-        if "/" in exclusion or os.sep in exclusion:
-            if fnmatch.fnmatch(normalized_path, exclusion):
-                return True
-    if "/" in exclusion or os.sep in exclusion:
-        normalized = exclusion.rstrip("/").rstrip(os.sep)
-        return normalized_path == normalized or normalized_path.startswith(
-            normalized + "/"
-        ) or normalized_path.startswith(
-            normalized + os.sep
-        )
+        if "/" in normalized_excl and fnmatch.fnmatch(normalized_path, normalized_excl):
+            return True
+
+    if "/" in normalized_excl:
+        clean_excl = normalized_excl.rstrip("/")
+        if normalized_path == clean_excl or normalized_path.startswith(clean_excl + "/"):
+            return True
+
     return False
 
 

core/grep.py

@@ -15,7 +15,10 @@ def grep_files(
     pattern: str, file_list: list[str], *, flags: int = 0
 ) -> list[tuple[str, int, str]]:
     """Search files for a regex pattern. Returns (filepath, lineno, line_text)."""
-    compiled = re.compile(pattern, flags)
+    try:
+        compiled = re.compile(pattern, flags)
+    except re.error:
+        return []
     results: list[tuple[str, int, str]] = []
     for filepath in file_list:
         abs_path = filepath if os.path.isabs(filepath) else str(_get_project_root() / filepath)
@@ -35,22 +38,29 @@ def grep_files_containing(
     if not names:
         return {}
     names_by_length = sorted(names, key=len, reverse=True)
-    if word_boundary:
-        combined = re.compile(
-            r"\b(?:" + "|".join(re.escape(n) for n in names_by_length) + r")\b"
-        )
-    else:
-        combined = re.compile("|".join(re.escape(n) for n in names_by_length))
-
+
     name_to_files: dict[str, set[str]] = {}
-    for filepath in file_list:
-        abs_path = filepath if os.path.isabs(filepath) else str(_get_project_root() / filepath)
-        content = _read_file_text(abs_path)
-        if content is None:
+    chunk_size = 500
+    for i in range(0, len(names_by_length), chunk_size):
+        chunk = names_by_length[i:i + chunk_size]
+        try:
+            if word_boundary:
+                combined = re.compile(
+                    r"\b(?:" + "|".join(re.escape(n) for n in chunk) + r")\b"
+                )
+            else:
+                combined = re.compile("|".join(re.escape(n) for n in chunk))
+        except re.error:
             continue
-        found = set(combined.findall(content))
-        for name in found & names:
-            name_to_files.setdefault(name, set()).add(filepath)
+
+        for filepath in file_list:
+            abs_path = filepath if os.path.isabs(filepath) else str(_get_project_root() / filepath)
+            content = _read_file_text(abs_path)
+            if content is None:
+                continue
+            found = set(combined.findall(content))
+            for name in found & set(chunk):
+                name_to_files.setdefault(name, set()).add(filepath)
     return name_to_files
 
 

engine/detectors/coverage/mapping.py

@@ -40,23 +40,13 @@ def _infer_lang_name(test_files: set[str], production_files: set[str]) -> str |
             continue
         counts[lang_name] = counts.get(lang_name, 0) + 1
     if counts:
-        return max(counts.items(), key=lambda item: item[1])[0]
+        # Sort by count descending, then by language name alphabetically to ensure deterministic ties
+        return sorted(counts.items(), key=lambda item: (-item[1], item[0]))[0][0]
     return None
 
 
-def _import_based_mapping(
-    graph: dict,
-    test_files: set[str],
-    production_files: set[str],
-    lang_name: str | None = None,
-) -> set[str]:
-    """Map test files to production files via import edges."""
-    lang_name = lang_name or _infer_lang_name(test_files, production_files)
-    mod = _load_lang_test_coverage_module(lang_name)
-
-    tested = set()
-
-    # Build module-name->path index for resolving test imports.
+def _build_prod_by_module_index(production_files: set[str]) -> dict[str, str]:
+    """Build a module-name -> file-path mapping for imports."""
     prod_by_module: dict[str, str] = {}
     root_str = str(PROJECT_ROOT) + os.sep
     for pf in production_files:
@@ -66,13 +56,32 @@ def _import_based_mapping(
             module_name = module_name.rsplit(".", 1)[0]
         prod_by_module[module_name] = pf
 
-        # __init__.py: also map package path (e.g. "foo.bar" -> __init__.py).
         if module_name.endswith(".__init__"):
             prod_by_module[module_name[: -len(".__init__")]] = pf
 
         parts = module_name.split(".")
-        if parts:
+        if parts and parts[-1] not in prod_by_module:
             prod_by_module[parts[-1]] = pf
+
+    return prod_by_module
+
+
+def _import_based_mapping(
+    graph: dict,
+    test_files: set[str],
+    production_files: set[str],
+    lang_name: str | None = None,
+    parsed_imports_by_test: dict[str, set[str]] | None = None,
+) -> set[str]:
+    """Map test files to production files via import edges."""
+    lang_name = lang_name or _infer_lang_name(test_files, production_files)
+    mod = _load_lang_test_coverage_module(lang_name)
+    parsed_cache = parsed_imports_by_test or {}
+
+    tested = set()
+
+    # Build module-name->path index for resolving test imports.
+    prod_by_module = _build_prod_by_module_index(production_files)
 
     for tf in test_files:
         entry = graph.get(tf)
@@ -87,9 +96,12 @@ def _import_based_mapping(
         # or always for TypeScript where dynamic import('...') is common in
         # coverage smoke tests and may be missed by static graph building.
         if not graph_mapped or lang_name == "typescript":
-            tested |= _parse_test_imports(
-                tf, production_files, prod_by_module, lang_name
-            )
+            if tf in parsed_cache:
+                tested |= parsed_cache[tf]
+            else:
+                parsed = _parse_test_imports(tf, production_files, prod_by_module, lang_name)
+                parsed_cache[tf] = parsed
+                tested |= parsed
 
     barrel_basenames = getattr(mod, "BARREL_BASENAMES", set())
     if barrel_basenames:
@@ -370,7 +382,7 @@ def _get_test_files_for_prod(
         module_name = module_name.rsplit(".", 1)[0]
     prod_by_module: dict[str, str] = {module_name: prod_file}
     parts = module_name.split(".")
-    if parts:
+    if parts and parts[-1] not in prod_by_module:
         prod_by_module[parts[-1]] = prod_file
 
     result = []
@@ -396,19 +408,7 @@ def _build_test_import_index(
     lang_name: str,
 ) -> dict[str, set[str]]:
     """Parse test import sources once, producing a test->production import index."""
-    root_str = str(PROJECT_ROOT) + os.sep
-    prod_by_module: dict[str, str] = {}
-    for pf in production_files:
-        rel_pf = pf[len(root_str):] if pf.startswith(root_str) else pf
-        module_name = rel_pf.replace("/", ".").replace("\\", ".")
-        if "." in module_name:
-            module_name = module_name.rsplit(".", 1)[0]
-        prod_by_module[module_name] = pf
-        if module_name.endswith(".__init__"):
-            prod_by_module[module_name[: -len(".__init__")]] = pf
-        parts = module_name.split(".")
-        if parts:
-            prod_by_module[parts[-1]] = pf
+    prod_by_module = _build_prod_by_module_index(production_files)
 
     index: dict[str, set[str]] = {}
     for tf in test_files:

engine/detectors/patterns/security.py

@@ -174,4 +174,4 @@ def is_placeholder(value: str) -> bool:
         return True
     if any(lower.startswith(prefix) for prefix in PLACEHOLDER_PREFIXES):
         return True
-    return len(value) < 8
+    return len(value) < 5

engine/detectors/test_coverage/discovery.py

@@ -32,6 +32,7 @@ def _to_rel(path: str) -> str:
         norm_graph[rel_key] = {
             **value,
             "imports": {_to_rel(imp) for imp in value.get("imports", set())},
+            "importers": {_to_rel(imp) for imp in value.get("importers", set())},
         }
     return norm_graph
 

hook_registry.py

@@ -39,15 +39,15 @@ def get_lang_hook(lang_name: str | None, hook_name: str) -> object | None:
         try:
             importlib.import_module(module_name)
         except (ImportError, ValueError, TypeError, RuntimeError, OSError) as exc:
-            _LOGGER.debug(
+            _LOGGER.warning(
                 "Unable to import language hook package %s: %s", lang_name, exc
             )
             return None
     elif lang_name not in _LANG_HOOKS:
         try:
             importlib.reload(module)
         except (ImportError, ValueError, TypeError, RuntimeError, OSError) as exc:
-            _LOGGER.debug(
+            _LOGGER.warning(
                 "Unable to reload language hook package %s: %s", lang_name, exc
             )
             return None