diff --git a/app/commands/move/move_apply.py b/app/commands/move/move_apply.py index 7f26d7a..c676db6 100644 --- a/app/commands/move/move_apply.py +++ b/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: diff --git a/app/commands/plan_cmd.py b/app/commands/plan_cmd.py index c850336..eaf492b 100644 --- a/app/commands/plan_cmd.py +++ b/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: diff --git a/app/commands/scan/scan_workflow.py b/app/commands/scan/scan_workflow.py index 9924b3c..2e34fbd 100644 --- a/app/commands/scan/scan_workflow.py +++ b/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, diff --git a/app/commands/show/payload.py b/app/commands/show/payload.py index 772239a..067519a 100644 --- a/app/commands/show/payload.py +++ b/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 diff --git a/core/file_paths.py b/core/file_paths.py index 7762105..972cff8 100644 --- a/core/file_paths.py +++ b/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 diff --git a/core/grep.py b/core/grep.py index 5411ac0..afd0011 100644 --- a/core/grep.py +++ b/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 diff --git a/engine/detectors/coverage/mapping.py b/engine/detectors/coverage/mapping.py index 6d00efa..ac4c3c6 100644 --- a/engine/detectors/coverage/mapping.py +++ b/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: diff --git a/engine/detectors/patterns/security.py b/engine/detectors/patterns/security.py index b3947bb..ab86bb8 100644 --- a/engine/detectors/patterns/security.py +++ b/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 diff --git a/engine/detectors/test_coverage/discovery.py b/engine/detectors/test_coverage/discovery.py index bf99600..a4c5b6e 100644 --- a/engine/detectors/test_coverage/discovery.py +++ b/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 diff --git a/hook_registry.py b/hook_registry.py index ff14e01..104b704 100644 --- a/hook_registry.py +++ b/hook_registry.py @@ -39,7 +39,7 @@ 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 @@ -47,7 +47,7 @@ def get_lang_hook(lang_name: str | None, hook_name: str) -> object | None: 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