api/patch_tool.py

"""
patch_tool.py — Atomic file patching via unified diff (S366)

Pipeline: read → apply_patch → syntax_check → write_back (VFS)
Each stage is pure and independently testable. Write-back only happens
when patch applied cleanly AND syntax validation passes.

Design invariants:
- apply_unified_diff() is pure (no I/O, no side effects)
- check_syntax() uses stdlib only (ast, json) — zero extra deps
- patch_file_in_vfs() is atomic: partial failures leave VFS unchanged
- Never raises exceptions — returns error dict on all failure paths
- Compatible with both dict-of-str VFS and dict-of-dict VFS layouts
"""
from __future__ import annotations

import ast
import json
import re
from typing import Any

# ── Unified diff parser ───────────────────────────────────────────────────────
_HUNK_HEADER_RE = re.compile(
    r'^@@ -(\d+)(?:,(\d+))? \+(\d+)(?:,(\d+))? @@'
)


def apply_unified_diff(original: str, patch: str) -> tuple[str, list[str]]:
    """
    Apply a unified diff string to original text.
    Returns (patched_text, errors).
    errors is an empty list on clean apply.
    """
    if not patch.strip():
        return original, ["empty patch"]

    patch_lines = patch.splitlines()
    result: list[str] = original.splitlines(keepends=True)
    errors: list[str] = []

    # Collect hunk descriptors
    hunks: list[tuple[int, int, list[str]]] = []
    i = 0
    while i < len(patch_lines):
        m = _HUNK_HEADER_RE.match(patch_lines[i])
        if m:
            orig_start = int(m.group(1)) - 1   # convert to 0-indexed
            orig_count = int(m.group(2) or 1)
            body: list[str] = []
            i += 1
            while i < len(patch_lines) and not _HUNK_HEADER_RE.match(patch_lines[i]) \
                    and not (patch_lines[i].startswith("---") and i + 1 < len(patch_lines)
                             and patch_lines[i + 1].startswith("+++")):
                body.append(patch_lines[i])
                i += 1
            hunks.append((orig_start, orig_count, body))
        else:
            i += 1

    if not hunks:
        # No hunk headers: treat patch as full-file replacement
        return patch, []

    # Apply hunks from bottom to top (preserves line offsets for earlier hunks)
    offset = 0
    for (orig_start, orig_count, body) in hunks:
        pos = orig_start + offset
        removes: list[str] = []
        adds: list[str] = []
        for line in body:
            if line.startswith("-"):
                removes.append(line[1:])
            elif line.startswith("+"):
                adds.append(line[1:])
            # context lines ignored during apply

        # Soft context validation (log mismatch but apply anyway)
        for idx, (exp, got) in enumerate(zip(removes, result[pos:pos + len(removes)])):
            if exp.rstrip("\n") != got.rstrip("\n"):
                errors.append(
                    f"line {pos + idx + 1}: expected {exp.rstrip()!r:.40s}, "
                    f"got {got.rstrip()!r:.40s}"
                )

        # Apply: replace remove-lines with add-lines
        add_lines = [a if a.endswith("\n") else a + "\n" for a in adds]
        result[pos: pos + len(removes)] = add_lines
        offset += len(adds) - len(removes)

    return "".join(result), errors


# ── Syntax validation ─────────────────────────────────────────────────────────

def check_syntax(code: str, language: str) -> tuple[bool, str]:
    """
    Returns (is_ok, error_message).
    Python → ast.parse, JSON → json.loads, other languages → pass-through.
    """
    lang = language.lower().strip()
    if lang in ("python", "py"):
        try:
            ast.parse(code, mode="exec")
            return True, ""
        except SyntaxError as e:
            return False, f"SyntaxError line {e.lineno}: {e.msg}"
    elif lang == "json":
        try:
            json.loads(code)
            return True, ""
        except json.JSONDecodeError as e:
            return False, f"JSONDecodeError: {e}"
    # TypeScript / JavaScript / other: no static check available here
    return True, ""


# ── VFS-aware atomic patch ────────────────────────────────────────────────────

async def patch_file_in_vfs(
    path: str,
    patch_text: str,
    vfs: dict[str, Any],
    language: str = "python",
) -> dict:
    """
    Atomic read → apply_unified_diff → check_syntax → write_back.

    VFS layout support:
        vfs[path] = "content string"           (simple)
        vfs[path] = {"content": "...", ...}    (metadata dict)

    Returns:
        {success: bool, path: str, lines_changed: int, errors: list, error: str|None}
    Never raises.
    """
    try:
        # Read current content from VFS
        entry = vfs.get(path)
        if isinstance(entry, dict):
            original = entry.get("content", "")
        elif isinstance(entry, str):
            original = entry
        else:
            original = ""

        patched, diff_errors = apply_unified_diff(original, patch_text)

        # Hard-fail on context mismatches (data integrity risk)
        if diff_errors:
            return {
                "success":       False,
                "path":          path,
                "lines_changed": 0,
                "errors":        diff_errors[:5],
                "error":         f"Patch context errors: {'; '.join(diff_errors[:2])}",
            }

        # Syntax gate: only write back clean code
        syntax_ok, syntax_err = check_syntax(patched, language)
        if not syntax_ok:
            return {
                "success":       False,
                "path":          path,
                "lines_changed": 0,
                "errors":        [syntax_err],
                "error":         syntax_err,
            }

        # Atomic write-back
        lines_changed = abs(len(patched.splitlines()) - len(original.splitlines()))
        if isinstance(vfs.get(path), dict):
            vfs[path]["content"] = patched
        else:
            vfs[path] = patched

        return {
            "success":       True,
            "path":          path,
            "lines_changed": lines_changed,
            "errors":        [],
            "error":         None,
        }
    except Exception as e:
        return {
            "success":       False,
            "path":          path,
            "lines_changed": 0,
            "errors":        [str(e)[:300]],  # S588: 200→300
            "error":         str(e)[:300],   # S588: 200→300
        }