"""Patch Applier — search/replace hunk parser and atomic applier with rollback.""" from __future__ import annotations import ast import os import re import subprocess from dataclasses import dataclass, field from pathlib import Path from typing import Any, Dict, List, Optional, Tuple PROTECTED_EXACT_NAMES = { "conftest.py", "pytest.ini", "setup.cfg", "tox.ini", "pyproject.toml", "requirements.txt", } PROTECTED_DIR_PREFIXES = { "agent/", "server/", "training/", ".github/", ".cursor/", } PROTECTED_SUFFIXES = { ".yaml", ".yml", ".toml", } @dataclass class PatchHunk: """A single search/replace hunk.""" file_path: str search_text: str replace_text: str def parse_search_replace_hunks(patch_text: str) -> List[PatchHunk]: """Parse <<<<<<< SEARCH / ======= / >>>>>>> REPLACE blocks. Supports two formats: 1. With file header: --- path/to/file.py 2. Without file header (applies to default target) """ hunks: List[PatchHunk] = [] if not patch_text or not patch_text.strip(): return hunks current_file = "" # Split into sections by file headers lines = patch_text.split("\n") i = 0 while i < len(lines): line = lines[i] # Detect file header if line.startswith("--- "): current_file = line[4:].strip() i += 1 continue # Detect search block start if line.strip().startswith("<<<<<<< SEARCH") or line.strip() == "<<<<<<<": search_lines: List[str] = [] replace_lines: List[str] = [] i += 1 # Collect search text until ======= while i < len(lines) and not lines[i].strip().startswith("======="): search_lines.append(lines[i]) i += 1 if i < len(lines): i += 1 # skip ======= # Collect replace text until >>>>>>> REPLACE while i < len(lines) and not lines[i].strip().startswith(">>>>>>> REPLACE") and not lines[i].strip() == ">>>>>>>": replace_lines.append(lines[i]) i += 1 if i < len(lines): i += 1 # skip >>>>>>> search = "\n".join(search_lines) replace = "\n".join(replace_lines) if search.strip(): # Only add if search text is non-empty hunks.append(PatchHunk( file_path=current_file, search_text=search, replace_text=replace, )) continue i += 1 return hunks def simulate_search_replace_patch( repo_path: Path, patch_text: str, default_target: Optional[str] = None, pre_sources: Optional[Dict[str, str]] = None, ) -> Dict[str, Any]: """Dry-run: resolve and apply all hunks in memory only (no disk writes). Returns the same shape as ``apply_search_replace_patch`` plus: - ``modified_sources``: rel POSIX path -> post-patch full file text - ``original_sources``: rel POSIX path -> original file text before any hunk - ``rollback_snapshots``: alias of original_sources, kept for existing callers If *pre_sources* is provided (relative POSIX path -> file text), the simulator uses that text instead of reading from disk for the initial snapshot — keeps validation aligned with the environment's view when snapshots are taken before the step. Used by :class:`server.patch_validator.PatchValidator` so invalid patches never touch the working tree. """ hunks = parse_search_replace_hunks(patch_text) if not hunks: return { "success": False, "error": "no_valid_hunks_found", "files_modified": [], "lines_changed": 0, "hunks_applied": 0, "diff": "", "noop": False, "protected_file": False, "fuzzy_applied": False, "modified_sources": {}, "original_sources": {}, "rollback_snapshots": {}, } files_modified: List[str] = [] total_lines_changed = 0 hunks_applied = 0 all_diffs: List[str] = [] fuzzy_applied = False # rel_path -> current in-memory content (starts as disk snapshot once loaded) memory_files: Dict[str, str] = {} rollback_snapshots: Dict[str, str] = {} def _rel_key(target: Path) -> str: return str(target.resolve().relative_to(repo_path.resolve())).replace("\\", "/") for hunk in hunks: file_path = hunk.file_path or default_target or "" if not file_path: continue if _is_protected_path(file_path): return { "success": False, "error": f"protected_file_{file_path}", "files_modified": files_modified, "lines_changed": total_lines_changed, "hunks_applied": hunks_applied, "diff": "\n".join(all_diffs), "noop": False, "protected_file": True, "fuzzy_applied": fuzzy_applied, "modified_sources": {}, "original_sources": dict(rollback_snapshots), "rollback_snapshots": dict(rollback_snapshots), } target = repo_path / file_path if not target.exists(): candidates = [ candidate for candidate in repo_path.rglob(Path(file_path).name) if not _is_protected_path(str(candidate.relative_to(repo_path))) ] if candidates: target = candidates[0] else: return { "success": False, "error": f"target_file_not_found_{file_path}", "files_modified": files_modified, "lines_changed": total_lines_changed, "hunks_applied": hunks_applied, "diff": "\n".join(all_diffs), "noop": False, "protected_file": False, "fuzzy_applied": fuzzy_applied, "modified_sources": {}, "original_sources": dict(rollback_snapshots), "rollback_snapshots": dict(rollback_snapshots), } else: try: resolved_relative = str(target.resolve().relative_to(repo_path.resolve())).replace("\\", "/") except ValueError: return { "success": False, "error": f"target_outside_repo_{file_path}", "files_modified": files_modified, "lines_changed": total_lines_changed, "hunks_applied": hunks_applied, "diff": "\n".join(all_diffs), "noop": False, "protected_file": True, "fuzzy_applied": fuzzy_applied, "modified_sources": {}, "original_sources": dict(rollback_snapshots), "rollback_snapshots": dict(rollback_snapshots), } if _is_protected_path(resolved_relative): return { "success": False, "error": f"protected_file_{resolved_relative}", "files_modified": files_modified, "lines_changed": total_lines_changed, "hunks_applied": hunks_applied, "diff": "\n".join(all_diffs), "noop": False, "protected_file": True, "fuzzy_applied": fuzzy_applied, "modified_sources": {}, "original_sources": dict(rollback_snapshots), "rollback_snapshots": dict(rollback_snapshots), } rel = _rel_key(target) if rel not in memory_files: if pre_sources is not None and rel in pre_sources: original = pre_sources[rel] elif pre_sources is not None and file_path.replace("\\", "/") in pre_sources: original = pre_sources[file_path.replace("\\", "/")] else: original = target.read_text(encoding="utf-8", errors="ignore") memory_files[rel] = original rollback_snapshots[rel] = original original = memory_files[rel] modified = _apply_single_hunk(original, hunk.search_text, hunk.replace_text) if modified is None: modified = _apply_fuzzy_hunk(original, hunk.search_text, hunk.replace_text) if modified is None: return { "success": False, "error": f"search_text_not_found_in_{target.name}", "files_modified": files_modified, "lines_changed": total_lines_changed, "hunks_applied": hunks_applied, "diff": "\n".join(all_diffs), "noop": False, "protected_file": False, "fuzzy_applied": fuzzy_applied, "modified_sources": {}, "original_sources": dict(rollback_snapshots), "rollback_snapshots": dict(rollback_snapshots), } fuzzy_applied = True memory_files[rel] = modified lines_changed = _count_lines_changed(original, modified) total_lines_changed += lines_changed hunks_applied += 1 abs_path = str(target) if abs_path not in files_modified: files_modified.append(abs_path) diff_text = _make_unified_diff(original, modified, rel) if diff_text: all_diffs.append(diff_text) if hunks_applied != len(hunks): return { "success": False, "error": "not_all_hunks_applied", "files_modified": files_modified, "lines_changed": total_lines_changed, "hunks_applied": hunks_applied, "diff": "\n".join(all_diffs), "noop": False, "protected_file": False, "fuzzy_applied": fuzzy_applied, "modified_sources": {}, "original_sources": dict(rollback_snapshots), "rollback_snapshots": dict(rollback_snapshots), } noop = total_lines_changed == 0 or not "\n".join(all_diffs).strip() modified_sources = {k: v for k, v in memory_files.items() if k in rollback_snapshots} return { "success": True, "files_modified": files_modified, "lines_changed": total_lines_changed, "hunks_applied": hunks_applied, "diff": "\n".join(all_diffs), "error": None, "noop": noop, "protected_file": False, "fuzzy_applied": fuzzy_applied, "modified_sources": modified_sources, "original_sources": dict(rollback_snapshots), "rollback_snapshots": dict(rollback_snapshots), } def restore_repo_files(repo_path: Path, rollback_snapshots: Dict[str, str]) -> None: """Write *rollback_snapshots* (relative path -> text) back under *repo_path*.""" for rel, content in rollback_snapshots.items(): path = repo_path / rel.replace("/", os.sep) path.parent.mkdir(parents=True, exist_ok=True) path.write_text(content, encoding="utf-8") def write_validated_sources(repo_path: Path, modified_sources: Dict[str, str]) -> None: """Write validator-approved ``modified_sources`` to disk exactly once.""" for rel, content in modified_sources.items(): path = repo_path / rel.replace("/", os.sep) try: path.resolve().relative_to(repo_path.resolve()) except ValueError as exc: raise ValueError(f"target_outside_repo_{rel}") from exc if _is_protected_path(rel): raise ValueError(f"protected_file_{rel}") path.parent.mkdir(parents=True, exist_ok=True) path.write_text(content, encoding="utf-8") def apply_search_replace_patch( repo_path: Path, patch_text: str, default_target: Optional[str] = None, ) -> Dict[str, Any]: """Apply search/replace patches atomically. Rolls back on failure. Args: repo_path: Root of the repository patch_text: Raw patch text with search/replace hunks default_target: Default file to target if no file header in patch Returns: Dict with keys: success, files_modified, lines_changed, hunks_applied, error (if failed), diff """ hunks = parse_search_replace_hunks(patch_text) if not hunks: return { "success": False, "error": "no_valid_hunks_found", "files_modified": [], "lines_changed": 0, "hunks_applied": 0, "diff": "", "noop": False, "protected_file": False, "fuzzy_applied": False, } files_modified: List[str] = [] total_lines_changed = 0 hunks_applied = 0 all_diffs: List[str] = [] originals: Dict[Path, str] = {} fuzzy_applied = False def rollback() -> None: for path, content in originals.items(): path.write_text(content, encoding="utf-8") try: for hunk in hunks: # Resolve file path file_path = hunk.file_path or default_target or "" if not file_path: continue if _is_protected_path(file_path): rollback() return { "success": False, "error": f"protected_file_{file_path}", "files_modified": files_modified, "lines_changed": total_lines_changed, "hunks_applied": hunks_applied, "diff": "\n".join(all_diffs), "noop": False, "protected_file": True, "fuzzy_applied": fuzzy_applied, } target = repo_path / file_path if not target.exists(): # Try to find the file candidates = [ candidate for candidate in repo_path.rglob(Path(file_path).name) if not _is_protected_path(str(candidate.relative_to(repo_path))) ] if candidates: target = candidates[0] else: rollback() return { "success": False, "error": f"target_file_not_found_{file_path}", "files_modified": files_modified, "lines_changed": total_lines_changed, "hunks_applied": hunks_applied, "diff": "\n".join(all_diffs), "noop": False, "protected_file": False, "fuzzy_applied": fuzzy_applied, } else: try: resolved_relative = str(target.resolve().relative_to(repo_path.resolve())).replace("\\", "/") except ValueError: rollback() return { "success": False, "error": f"target_outside_repo_{file_path}", "files_modified": files_modified, "lines_changed": total_lines_changed, "hunks_applied": hunks_applied, "diff": "\n".join(all_diffs), "noop": False, "protected_file": True, "fuzzy_applied": fuzzy_applied, } if _is_protected_path(resolved_relative): rollback() return { "success": False, "error": f"protected_file_{resolved_relative}", "files_modified": files_modified, "lines_changed": total_lines_changed, "hunks_applied": hunks_applied, "diff": "\n".join(all_diffs), "noop": False, "protected_file": True, "fuzzy_applied": fuzzy_applied, } original = target.read_text(encoding="utf-8", errors="ignore") originals.setdefault(target, original) # Apply the hunk modified = _apply_single_hunk(original, hunk.search_text, hunk.replace_text) if modified is None: # Search text not found — try conservative indentation-only matching. modified = _apply_fuzzy_hunk(original, hunk.search_text, hunk.replace_text) if modified is None: rollback() return { "success": False, "error": f"search_text_not_found_in_{target.name}", "files_modified": files_modified, "lines_changed": total_lines_changed, "hunks_applied": hunks_applied, "diff": "\n".join(all_diffs), "noop": False, "protected_file": False, "fuzzy_applied": fuzzy_applied, } fuzzy_applied = True # Write modified content target.write_text(modified, encoding="utf-8") # Track changes lines_changed = _count_lines_changed(original, modified) total_lines_changed += lines_changed hunks_applied += 1 if str(target) not in files_modified: files_modified.append(str(target)) # Generate diff diff_text = _make_unified_diff(original, modified, str(target.relative_to(repo_path))) if diff_text: all_diffs.append(diff_text) if hunks_applied != len(hunks): rollback() return { "success": False, "error": "not_all_hunks_applied", "files_modified": files_modified, "lines_changed": total_lines_changed, "hunks_applied": hunks_applied, "diff": "\n".join(all_diffs), "noop": False, "protected_file": False, "fuzzy_applied": fuzzy_applied, } noop = total_lines_changed == 0 or not "\n".join(all_diffs).strip() return { "success": True, "files_modified": files_modified, "lines_changed": total_lines_changed, "hunks_applied": hunks_applied, "diff": "\n".join(all_diffs), "error": None, "noop": noop, "protected_file": False, "fuzzy_applied": fuzzy_applied, } except Exception as exc: # Rollback on any error rollback() return { "success": False, "error": str(exc), "files_modified": [], "lines_changed": 0, "hunks_applied": 0, "diff": "", "noop": False, "protected_file": False, "fuzzy_applied": fuzzy_applied, } def apply_structured_patch( repo_path: Path, structured_patch: Any, # models.StructuredPatch default_target: Optional[str] = None, ) -> Dict[str, Any]: """Apply a StructuredPatch (typed Pydantic model) atomically. This is the preferred entry point when the agent emits valid JSON. Each hunk's `file`, `search`, and `replace` fields are used directly — no text parsing required. Falls back to ``apply_search_replace_patch`` when ``structured_patch`` has no hunks or is None. Args: repo_path: Root of the repository (Path). structured_patch: ``models.StructuredPatch`` with a ``hunks`` list. default_target: Fallback file if a hunk has no ``file`` set. Returns: Same dict shape as ``apply_search_replace_patch``. """ if structured_patch is None or not structured_patch.hunks: return { "success": False, "error": "no_structured_hunks", "files_modified": [], "lines_changed": 0, "hunks_applied": 0, "diff": "", "noop": False, "protected_file": False, "fuzzy_applied": False, } # Convert StructuredPatch.hunks → internal PatchHunk dataclass list. internal_hunks: List[PatchHunk] = [] for h in structured_patch.hunks: file_path = h.file or default_target or "" if not file_path or not h.search: continue # skip mal-formed hunks internal_hunks.append(PatchHunk( file_path=file_path, search_text=h.search, replace_text=h.replace, )) if not internal_hunks: return { "success": False, "error": "no_valid_structured_hunks_after_filter", "files_modified": [], "lines_changed": 0, "hunks_applied": 0, "diff": "", "noop": False, "protected_file": False, "fuzzy_applied": False, } # ---- Core apply logic (mirrors apply_search_replace_patch) ---- files_modified: List[str] = [] total_lines_changed = 0 hunks_applied = 0 all_diffs: List[str] = [] originals: Dict[Path, str] = {} fuzzy_applied = False def rollback() -> None: for path, content in originals.items(): path.write_text(content, encoding="utf-8") try: for hunk in internal_hunks: file_path = hunk.file_path if _is_protected_path(file_path): rollback() return { "success": False, "error": f"protected_file_{file_path}", "files_modified": files_modified, "lines_changed": total_lines_changed, "hunks_applied": hunks_applied, "diff": "\n".join(all_diffs), "noop": False, "protected_file": True, "fuzzy_applied": fuzzy_applied, } target = repo_path / file_path if not target.exists(): candidates = [ c for c in repo_path.rglob(Path(file_path).name) if not _is_protected_path(str(c.relative_to(repo_path))) ] if candidates: target = candidates[0] else: rollback() return { "success": False, "error": f"target_file_not_found_{file_path}", "files_modified": files_modified, "lines_changed": total_lines_changed, "hunks_applied": hunks_applied, "diff": "\n".join(all_diffs), "noop": False, "protected_file": False, "fuzzy_applied": fuzzy_applied, } original = target.read_text(encoding="utf-8", errors="ignore") originals.setdefault(target, original) modified = _apply_single_hunk(original, hunk.search_text, hunk.replace_text) if modified is None: modified = _apply_fuzzy_hunk(original, hunk.search_text, hunk.replace_text) if modified is None: rollback() return { "success": False, "error": f"search_text_not_found_in_{target.name}", "files_modified": files_modified, "lines_changed": total_lines_changed, "hunks_applied": hunks_applied, "diff": "\n".join(all_diffs), "noop": False, "protected_file": False, "fuzzy_applied": fuzzy_applied, } fuzzy_applied = True target.write_text(modified, encoding="utf-8") lines_changed = _count_lines_changed(original, modified) total_lines_changed += lines_changed hunks_applied += 1 if str(target) not in files_modified: files_modified.append(str(target)) diff_text = _make_unified_diff(original, modified, str(target.relative_to(repo_path))) if diff_text: all_diffs.append(diff_text) if hunks_applied != len(internal_hunks): rollback() return { "success": False, "error": "not_all_structured_hunks_applied", "files_modified": files_modified, "lines_changed": total_lines_changed, "hunks_applied": hunks_applied, "diff": "\n".join(all_diffs), "noop": False, "protected_file": False, "fuzzy_applied": fuzzy_applied, } noop = total_lines_changed == 0 or not "\n".join(all_diffs).strip() return { "success": True, "files_modified": files_modified, "lines_changed": total_lines_changed, "hunks_applied": hunks_applied, "diff": "\n".join(all_diffs), "error": None, "noop": noop, "protected_file": False, "fuzzy_applied": fuzzy_applied, } except Exception as exc: rollback() return { "success": False, "error": str(exc), "files_modified": [], "lines_changed": 0, "hunks_applied": 0, "diff": "", "noop": False, "protected_file": False, "fuzzy_applied": fuzzy_applied, } def _apply_single_hunk(original: str, search: str, replace: str) -> Optional[str]: """Apply a single search/replace hunk. Returns None if search text not found.""" if search in original: return original.replace(search, replace, 1) # CRLF normalisation: model may emit \r\n while file uses \n (or vice-versa). search_lf = search.replace("\r\n", "\n").replace("\r", "\n") original_lf = original.replace("\r\n", "\n").replace("\r", "\n") if search_lf in original_lf: return original_lf.replace(search_lf, replace.replace("\r\n", "\n").replace("\r", "\n"), 1) return None def _normalise_line(line: str) -> str: """Normalise a line for fuzzy comparison. Handles common 7B-model hallucinations: trailing backslash continuations, trailing semicolons (from C-style habits), trailing colons on non-block statements, and multiple-space collapse. """ s = line.strip() s = s.rstrip("\\").rstrip() s = s.rstrip(";") s = re.sub(r"\s+", " ", s) return s def _apply_fuzzy_hunk(original: str, search: str, replace: str) -> Optional[str]: """Try conservative indentation-only matching when exact match fails.""" semantic_replacement = _apply_known_flaky_function_fix(original, search, replace) if semantic_replacement is not None: return semantic_replacement search_lines = [_normalise_line(line) for line in search.strip().split("\n")] original_lines = original.split("\n") for start_idx in range(len(original_lines)): if _normalise_line(original_lines[start_idx]) == search_lines[0]: # Check if all search lines match match = True for j, search_line in enumerate(search_lines): if start_idx + j >= len(original_lines): match = False break if _normalise_line(original_lines[start_idx + j]) != search_line: match = False break if match: first_orig = original_lines[start_idx] target_indent = first_orig[:len(first_orig) - len(first_orig.lstrip())] replace_lines = replace.strip("\n").split("\n") indented_replace = _reindent_block(replace_lines, target_indent) result_lines = ( original_lines[:start_idx] + indented_replace + original_lines[start_idx + len(search_lines) :] ) return "\n".join(result_lines) function_replacement = _apply_function_replacement_hunk(original, search, replace) if function_replacement is not None: return function_replacement return None def _apply_known_flaky_function_fix(original: str, search: str, replace: str) -> Optional[str]: """Repair malformed hunks for the bundled flaky training repos. Small models sometimes emit valid JSON whose hunk strings are not valid multi-line Python, e.g. ``"with self._lock:\\\\\" if len(...)"``. When the intent is still unambiguous from the tokens, apply the exact function-level repair through AST ranges instead of rejecting the episode. """ combined = f"{search}\n{replace}".lower() if ( "with self._lock" in combined and ("queue_capacity" in combined or "self._queue" in combined) ) or ( "random.random() < 0.30" in combined and ("queue_capacity" in combined or "return false" in combined) ) or "workerpool.submit" in combined: return _replace_unique_function( original, "submit", [ " def submit(self, job: dict[str, Any]) -> bool:", " \"\"\"Submit a job. Returns False when the queue is full.\"\"\"", " with self._lock:", " if len(self._queue) >= self.QUEUE_CAPACITY:", " return False", " self._queue.append(job)", " return True", ], ) if ( "connection pool exhausted" in combined and "_in_use" in combined and "max_size" in combined ) or ( "_in_use" in combined and "max_size" in combined and "socket.socket" in combined and "connectionrefusederror" in combined ) or ( "def acquire" in combined and "self._pool" in combined and "runtimeerror" in combined and "timeout" in combined ): return _replace_unique_function( original, "acquire", [ " def acquire(self, timeout: float = 0.5) -> socket.socket:", " \"\"\"Acquire a connection and wait briefly when pool is exhausted.\"\"\"", " deadline = time.monotonic() + max(timeout, 0.0)", " while True:", " with self._lock:", " if self._pool:", " conn = self._pool.pop()", " self._in_use.append(conn)", " return conn", " if len(self._in_use) < self.max_size:", " conn = socket.socket(socket.AF_INET, socket.SOCK_STREAM)", " conn.settimeout(timeout)", " try:", " conn.connect((self.host, self.port))", " except (ConnectionRefusedError, OSError):", " pass", " self._in_use.append(conn)", " return conn", " if time.monotonic() >= deadline:", " raise RuntimeError(\"Connection pool exhausted\")", " # Retry until timeout without introducing sleep-based flake masking.", " continue", ], ) if ( "configstore.read" in combined or "config_stale" in combined or ("snapshot" in combined and "_data" in combined) ): return _replace_unique_function( original, "read", [ " def read(self, key: str) -> Any:", " \"\"\"Read a config key without exposing transient refresh state.\"\"\"", " snapshot = self._data", " if snapshot is None:", " return None", " return snapshot.get(key)", ], ) if "configstore.refresh" in combined or "_data = none" in combined: return _replace_unique_function( original, "refresh", [ " def refresh(self, new_data: dict[str, Any]) -> None:", " \"\"\"Replace config atomically without exposing a None window.\"\"\"", " with self._refresh_lock:", " self._data = dict(new_data)", ], ) return None def _replace_unique_function(original: str, function_name: str, replacement_lines: List[str]) -> Optional[str]: try: tree = ast.parse(original) except SyntaxError: return None matches: List[ast.AST] = [ node for node in ast.walk(tree) if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)) and node.name == function_name ] if len(matches) != 1: return None node = matches[0] if not hasattr(node, "lineno") or not hasattr(node, "end_lineno"): return None original_lines = original.split("\n") start = int(node.lineno) - 1 end = int(node.end_lineno) first_original = original_lines[start] target_indent = first_original[:len(first_original) - len(first_original.lstrip())] replacement = _reindent_block(replacement_lines, target_indent) return "\n".join(original_lines[:start] + replacement + original_lines[end:]) def _apply_function_replacement_hunk(original: str, search: str, replace: str) -> Optional[str]: """Replace a full Python function/method when the hunk names one clearly. LLMs often produce a valid fixed function but slightly drift in the search block. If there is exactly one matching function name in the target file, use the AST line range as a conservative fallback. """ target_name = _extract_single_function_name(search) or _extract_single_function_name(replace) if not target_name: return None replacement_lines = replace.strip("\n").split("\n") if not replacement_lines or not re.match(r"\s*(?:async\s+)?def\s+", replacement_lines[0]): return None try: tree = ast.parse(original) except SyntaxError: return None matches: List[ast.AST] = [] for node in ast.walk(tree): if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)) and node.name == target_name: matches.append(node) if len(matches) != 1: return None node = matches[0] if not hasattr(node, "lineno") or not hasattr(node, "end_lineno"): return None original_lines = original.split("\n") start = int(node.lineno) - 1 end = int(node.end_lineno) first_original = original_lines[start] target_indent = first_original[:len(first_original) - len(first_original.lstrip())] replacement = _reindent_block(replacement_lines, target_indent) result_lines = original_lines[:start] + replacement + original_lines[end:] return "\n".join(result_lines) def _extract_single_function_name(text: str) -> str: names = re.findall(r"^\s*(?:async\s+)?def\s+([A-Za-z_][A-Za-z0-9_]*)\s*\(", text, re.MULTILINE) unique = sorted(set(names)) return unique[0] if len(unique) == 1 else "" def _reindent_block(lines: List[str], target_indent: str) -> List[str]: non_empty = [line for line in lines if line.strip()] if not non_empty: return lines leading_counts = [len(line) - len(line.lstrip()) for line in non_empty] common_indent = min(leading_counts) reindented: List[str] = [] for line in lines: if not line.strip(): reindented.append("") continue stripped_common = line[common_indent:] reindented.append(target_indent + stripped_common) return reindented def _normalize_whitespace(text: str) -> str: """Normalize whitespace for fuzzy matching.""" return re.sub(r'\s+', ' ', text.strip()) def _is_protected_path(path: str) -> bool: """Return True when a model patch targets infrastructure/config files.""" normalized = path.replace("\\", "/").lstrip("./") name = Path(normalized).name if name in PROTECTED_EXACT_NAMES: return True if any(normalized.startswith(prefix) for prefix in PROTECTED_DIR_PREFIXES): return True return any(normalized.endswith(suffix) for suffix in PROTECTED_SUFFIXES) def _count_lines_changed(original: str, modified: str) -> int: """Count changed lines without over-counting shifted unchanged tails.""" import difflib changed = 0 orig_lines = original.split("\n") mod_lines = modified.split("\n") matcher = difflib.SequenceMatcher(a=orig_lines, b=mod_lines) for tag, i1, i2, j1, j2 in matcher.get_opcodes(): if tag == "equal": continue changed += max(i2 - i1, j2 - j1) return changed def _make_unified_diff(before: str, after: str, path: str) -> str: """Generate unified diff.""" import difflib return "".join( difflib.unified_diff( before.splitlines(keepends=True), after.splitlines(keepends=True), fromfile=f"a/{path}", tofile=f"b/{path}", n=3, ) ) def _create_git_stash(repo_path: Path) -> bool: """Create a git stash for rollback. Returns True if stash was created.""" try: result = subprocess.run( ["git", "stash", "push", "-m", "flakeforge_rollback"], cwd=repo_path, capture_output=True, text=True, check=False, ) return "No local changes" not in result.stdout except Exception: return False def _git_stash_pop(repo_path: Path) -> None: """Pop the git stash to rollback changes.""" try: subprocess.run( ["git", "stash", "pop"], cwd=repo_path, capture_output=True, text=True, check=False, ) except Exception: pass