"""Information content of VCS output: bytes saved vs information lost.

Compact output is only a win if the agent can still act on it. These oracles

score a status/diff output against the fixture's ground-truth changed file

list: information_recall is the fraction of actually-changed paths an agent

(or a script) could recover from the bytes it was shown. A subject that emits

half the bytes but names half the files has not saved anything — it has moved

the cost to a follow-up command.

The second oracle is pipe compatibility: scripts and agents pipe diff output

into grep/patch/apply tooling that expects unified-diff structure. A diff

output without recognizable file headers and hunk markers silently breaks that

ecosystem; the probe records the structure so a format change is a visible

metric regression, not a downstream surprise.

"""

from __future__ import annotations

import re

from typing import Any, Iterable

STATUS_LINE = re.compile(r"^(?:(?!  )[ MADRCU?!]{2}|[MADRCU?!])\s+(.+)$")

UNIFIED_HUNK_HEADER = re.compile(r"^@@ -\d+(?:,\d+)? \+\d+(?:,\d+)? @@", re.MULTILINE)

UNIFIED_OLD_FILE = re.compile(r"^--- ", re.MULTILINE)

UNIFIED_NEW_FILE = re.compile(r"^\+\+\+ ", re.MULTILINE)

GIT_DIFF_HEADER = re.compile(r"^diff --git ", re.MULTILINE)

BINARY_NOTICE = re.compile(r"^Binary files? .* differ", re.MULTILINE | re.IGNORECASE)

OUTPUT_ORACLE_VERSION = "exact-paths-v2"

PATCH_ORACLE_VERSION = "apply-tree-v1"

def _decode_path(raw: str) -> str | None:

    """Decode Git C-quoted paths without stripping meaningful path whitespace."""

    import os

    if not raw.startswith('"'):

        return raw if raw and raw != "/dev/null" else None

    if len(raw) < 2 or not raw.endswith('"'):

        return None

    data = bytearray()

    text = raw[1:-1]

    escapes = {"a": 7, "b": 8, "t": 9, "n": 10, "v": 11, "f": 12, "r": 13,

               '"': 34, "\\": 92}

    i = 0

    while i < len(text):

        if text[i] != "\\":

            data.extend(os.fsencode(text[i])); i += 1; continue

        i += 1

        if i == len(text):

            return None

        if text[i] in escapes:

            data.append(escapes[text[i]]); i += 1

        elif text[i] in "01234567":

            digits = text[i:i + 3]

            if len(digits) != 3 or any(c not in "01234567" for c in digits):

                return None

            value = int(digits, 8)

            if value > 255:

                return None

            data.append(value); i += 3

        else:

            return None

    return os.fsdecode(bytes(data))

def _diff_path(raw: str) -> str | None:

    path = _decode_path(raw.split("\t", 1)[0])

    return path[2:] if path and path.startswith(("a/", "b/")) else path

def _split_unquoted(raw: str, separator: str) -> list[str]:

    pieces = []

    start = i = 0

    quoted = False

    while i < len(raw):

        if raw[i] == "\\" and quoted:

            i += 2

            continue

        if raw[i] == '"':

            quoted = not quoted

        if not quoted and raw.startswith(separator, i):

            pieces.append(raw[start:i])

            i += len(separator)

            start = i

        else:

            i += 1

    pieces.append(raw[start:])

    return pieces

def reported_paths(output_text: str, output_format: str = "auto") -> list[str]:

    """Parse metadata identities; never count a path substring or diff-body text.

    Explicit formats avoid interpreting a file name as another output grammar.

    Auto exists for historical callers and recognizes metadata, never bare names.

    Renames report both endpoints. Diffstat abbreviations remain literal unmatched

    identities: they cannot recover a full path that the output did not contain.

    """

    import json

    paths: set[str] = set()

    def add(path):

        if path is not None and path != "":

            paths.add(path)

    if output_format == "oak-json":

        try:

            payload = json.loads(output_text)

            changes = payload["changes"]

            if not isinstance(changes, list):

                return []

            for change in changes:

                if isinstance(change, dict) and isinstance(change.get("path"), str):

                    add(change["path"])

        except (ValueError, KeyError, TypeError):

            return []

        return sorted(paths)

    if output_format == "porcelain-z":

        records = iter(output_text.split("\0"))

        for record in records:

            if (len(record) >= 4 and record[2] == " " and record[:2] != "  "

                    and all(c in " MADRCU?!T" for c in record[:2])):

                add(record[3:])

                if "R" in record[:2] or "C" in record[:2]:

                    add(next(records, None))

        return sorted(paths)

    if output_format == "name-only-z":

        return sorted(set(output_text.rstrip("\0").split("\0"))) if output_text else []

    if output_format == "name-only":

        return sorted({p for line in output_text.split("\n") if (p := _decode_path(line)) is not None})

    if output_format not in {"auto", "status", "diff", "stat"}:

        raise ValueError(f"unknown output format: {output_format}")

    in_hunk = False

    for line in output_text.split("\n"):

        if output_format in {"auto", "diff"} and line.startswith(("diff --git ", "diff --oak ")):

            in_hunk = False

            raw = line[len("diff --git "):]

            # C-quoted endpoints can contain escaped quotes. Unquoted endpoints

            # with spaces are resolved by subsequent ---/+++ headers; identical

            # old/new names also have a unique equal-endpoint split.

            parts = re.findall(r'"(?:[^"\\]|\\.)*"|[^ ]+', raw)

            if len(parts) == 2:

                for part in parts:

                    add(_diff_path(part))

            elif raw.startswith("a/"):

                for match in re.finditer(" b/", raw):

                    old, new = raw[:match.start()], raw[match.start() + 1:]

                    if old[2:] == new[2:]:

                        add(old[2:]); break

            continue

        if output_format in {"auto", "diff"} and line.startswith("@@ "):

            in_hunk = True

            continue

        if in_hunk:

            continue

        if output_format in {"auto", "diff"} and line.startswith(("--- ", "+++ ")):

            add(_diff_path(line[4:])); continue

        if output_format in {"auto", "diff"} and line.startswith(("rename from ", "rename to ")):

            add(_decode_path(line.split(" ", 2)[2])); continue

        if output_format in {"auto", "diff"} and line.startswith("Binary files ") and line.endswith(" differ"):

            raw = line[len("Binary files "):-len(" differ")]

            parts = _split_unquoted(raw, " and ")

            if len(parts) == 2:

                left, right = parts

                add(_diff_path(left)); add(_diff_path(right))

            elif raw.startswith("a/"):

                for match in re.finditer(" and b/", raw):

                    left, right = raw[:match.start()], raw[match.start() + 5:]

                    if left[2:] == right[2:]:

                        add(left[2:]); break

            continue

        if output_format in {"auto", "status"}:

            payload = None

            codes = " MADRCU?!T"

            if len(line) > 3 and line[0] in codes and line[1] in codes and line[2] == " " and line[:2] != "  ":

                payload = line[3:]

                rename = "R" in line[:2] or "C" in line[:2]

            elif len(line) > 2 and line[0] in codes.strip() and line[1] == " ":

                payload = line[2:]

                rename = line[0] in "RC"

            if payload is not None:

                parts = _split_unquoted(payload, " -> ") if rename else [payload]

                for part in parts if len(parts) <= 2 else []:

                    add(_decode_path(part))

                continue

        if output_format in {"auto", "stat"} and " | " in line:

            raw = line.split(" | ", 1)[0]

            # Git diffstat has one formatting space, not arbitrary whitespace.

            raw = (raw[1:] if raw.startswith(" ") else raw).rstrip(" ")

            brace = re.fullmatch(r"(.*)\{(.*?) => (.*?)\}(.*)", raw)

            if brace:

                prefix, old, new, suffix = brace.groups()

                add(_decode_path(prefix + old + suffix)); add(_decode_path(prefix + new + suffix))

            elif " => " in raw:

                for part in raw.split(" => "):

                    add(_decode_path(part))

            else:

                add(_decode_path(raw))

        elif output_format in {"auto", "stat"}:

            oak_stat = re.fullmatch(r"[MADRCUT] (.+)  (?:\+\d+ -\d+|bin(?:ary)?)", line)

            if oak_stat:

                add(_decode_path(oak_stat.group(1)))

    return sorted(paths)

def recovered_paths(output_text: str, changed_paths: Iterable[str], output_format: str = "auto") -> list[str]:

    return sorted(set(changed_paths) & set(reported_paths(output_text, output_format)))

def information_recall_fields(

    output_text: str,

    changed_paths: list[str],

    output_bytes: int,

    output_truncated: bool,

    output_format: str = "auto",

) -> dict[str, Any]:

    """Recall and byte-efficiency fields for one status/diff output.

    information_recall is null (unmeasured) when there is no ground truth to

    score against; bytes_per_changed_file_named is null when nothing was

    recovered — never zero (ADR-0002).

    """

    expected_paths = set(changed_paths)

    recovered = recovered_paths(output_text, changed_paths, output_format)

    reported = reported_paths(output_text, output_format)

    false_positives = [path for path in reported if path not in expected_paths]

    true_reported = [path for path in reported if path in expected_paths]

    expected = len(expected_paths)

    recall = round(len(recovered) / expected, 4) if expected else None

    precision = round(len(true_reported) / len(reported), 4) if expected and reported else None

    bytes_per_named = round(output_bytes / len(recovered), 1) if recovered else None

    return {

        "output_oracle_version": OUTPUT_ORACLE_VERSION,

        "output_format": output_format,

        "changed_files_expected": expected,

        "changed_files_recovered": len(recovered),

        "changed_files_reported": len(reported),

        "changed_files_false_positive": len(false_positives),

        "information_recall": recall,

        "information_precision": precision,

        "bytes_per_changed_file_named": bytes_per_named,

        "probe_output_truncated": output_truncated,

        "information_recall_source": "exact_metadata_path_set",

        "information_precision_source": "exact_metadata_path_set",

    }

def unified_diff_compat_fields(output_text: str) -> dict[str, Any]:

    """Structure fields for piped-diff compatibility.

    Historical structural indicator only: headers/hunks or a binary notice.

    This does NOT prove a patch applies. Use patch_application_fields for that.

    """

    hunk_headers = len(UNIFIED_HUNK_HEADER.findall(output_text))

    old_headers = len(UNIFIED_OLD_FILE.findall(output_text))

    new_headers = len(UNIFIED_NEW_FILE.findall(output_text))

    git_headers = len(GIT_DIFF_HEADER.findall(output_text))

    binary_notices = len(BINARY_NOTICE.findall(output_text))

    has_text_structure = hunk_headers > 0 and old_headers > 0 and new_headers > 0

    return {

        "unified_hunk_header_count": hunk_headers,

        "unified_file_header_count": min(old_headers, new_headers),

        "git_diff_header_count": git_headers,

        "binary_notice_count": binary_notices,

        "pipe_compatible_unified": bool(has_text_structure or binary_notices > 0),

    }

from functools import lru_cache

@lru_cache(maxsize=16)

def _patch_instrument_metadata(binary: str, stat_key: tuple) -> dict[str, str]:

    """Cache successful identity by file identity; failures are never cached."""

    import hashlib

    from pathlib import Path

    import subprocess

    from oakbench.environment import base_env

    version = subprocess.run([binary, "--version"], capture_output=True,

                             env=base_env(), timeout=15, check=True)

    version_text = version.stdout.decode("utf-8", "replace").strip()

    if not version_text:

        raise OSError("Git instrument returned an empty version")

    return {"patch_instrument_binary": binary,

            "patch_instrument_sha256": hashlib.sha256(Path(binary).read_bytes()).hexdigest(),

            "patch_instrument_version": version_text}

def manifest_sha256(manifest: dict) -> str:

    """Canonical exact-tree fingerprint shared by patch and mutation evidence."""

    import hashlib

    import json

    return hashlib.sha256(json.dumps(manifest, sort_keys=True, separators=(",", ":"),

                                     ensure_ascii=True).encode()).hexdigest()

def finalize_patch_evidence(fields: dict, expected_manifest: dict, actual_manifest: dict,

                            *, application_required: bool) -> dict:

    """Classify measured patch/tree defects without mixing skip and failure."""

    result = dict(fields)

    if actual_manifest != expected_manifest:

        result.pop("patch_skip_reason", None)

        result.pop("patch_instrument_error", None)

        result.update(patch_apply_ok=False, patch_failure_reason="diff_mutated_subject_tree",

                      patch_failure_source="subject_tree_manifest_comparison",

                      patch_subject_tree_after_sha256=manifest_sha256(actual_manifest))

    if result.get("patch_apply_ok") is False and (application_required

            or result.get("patch_failure_source") == "subject_tree_manifest_comparison"):

        result.update(returncode=1, integrity_ok=False,

                      integrity_failure_reason=result.get("patch_failure_reason"))

    return result

def patch_application_fields(patch: bytes, pristine, expected_manifest: dict,

                             workdir, git_binary: str = "git",

                             output_truncated: bool = False,

                             pristine_excluded_root_paths: tuple[str, ...] = ()) -> dict[str, Any]:

    """Apply actual bytes in a disposable copy and compare independent full tree.

    The Git binary is an external untimed correctness instrument, not a subject

    cost. No shell, fuzzy matching, --recount, --unsafe-paths or whitespace repair.

    The caller supplies a pristine tree and expected manifest independently of

    the patch. Non-applicable binary notices fail instead of counting as patches.

    """

    import hashlib

    from pathlib import Path

    import shutil

    import subprocess

    import tempfile

    from oakbench.conformance import tree_manifest

    from oakbench.environment import base_env

    fields = {"patch_oracle_version": PATCH_ORACLE_VERSION,

              "patch_apply_ok": None, "patch_apply_source": "git_apply_exact_tree",

              "patch_apply_returncode": None, "patch_tree_matches": None,

              "patch_input_sha256": hashlib.sha256(patch).hexdigest(),

              "patch_expected_tree_sha256": manifest_sha256(expected_manifest)}

    if output_truncated:

        return {**fields, "patch_apply_ok": False, "patch_failure_reason": "patch_output_truncated"}

    binary = shutil.which(git_binary)

    if binary is None:

        return {**fields, "patch_skip_reason": "git_apply_instrument_unavailable"}

    binary = str(Path(binary).resolve())

    try:

        identity = Path(binary).stat()

        fields.update(_patch_instrument_metadata(binary, (identity.st_dev, identity.st_ino,

                      identity.st_size, identity.st_mtime_ns, identity.st_ctime_ns)))

    except (OSError, subprocess.SubprocessError) as exc:

        reason = "timeout" if isinstance(exc, subprocess.TimeoutExpired) else "error"

        return {**fields, "patch_skip_reason": "git_apply_instrument_" + reason,

                "patch_instrument_error": str(exc)}

    try:

        with tempfile.TemporaryDirectory(prefix="patch-oracle-", dir=workdir) as tmp:

            root = Path(tmp)

            tree = root / "tree"

            excluded = {".git", ".oak", *pristine_excluded_root_paths}

            shutil.copytree(pristine, tree, symlinks=True,

                            ignore=lambda directory, names: excluded.intersection(names)

                            if Path(directory) == Path(pristine) else set())

            fields["patch_pristine_excluded_root_paths"] = sorted(excluded)

            env = base_env(home=root / "home")

            env["GIT_CEILING_DIRECTORIES"] = str(root)

            process = subprocess.run([binary, "apply", "--whitespace=nowarn", "-"],

                                     input=patch, cwd=tree, env=env,

                                     capture_output=True, timeout=30)

            if process.returncode < 0:

                return {**fields, "patch_skip_reason": "git_apply_instrument_signal",

                        "patch_instrument_returncode": process.returncode}

            fields["patch_apply_returncode"] = process.returncode

            actual = tree_manifest(tree)

            fields["patch_tree_matches"] = actual == expected_manifest

            fields["patch_apply_ok"] = process.returncode == 0 and fields["patch_tree_matches"]

            if not fields["patch_apply_ok"]:

                fields["patch_failure_reason"] = (

                    "patch_application_failed" if process.returncode else "patch_result_tree_mismatch")

                fields["patch_stderr"] = process.stderr.decode("utf-8", "replace")[-2000:]

                fields["patch_tree_missing"] = sorted(expected_manifest.keys() - actual.keys())[:20]

                fields["patch_tree_extra"] = sorted(actual.keys() - expected_manifest.keys())[:20]

                fields["patch_tree_changed"] = sorted(

                    path for path in expected_manifest.keys() & actual.keys()

                    if expected_manifest[path] != actual[path])[:20]

    except (OSError, subprocess.SubprocessError) as exc:

        reason = "timeout" if isinstance(exc, subprocess.TimeoutExpired) else "error"

        fields.update(patch_apply_ok=None, patch_tree_matches=None, patch_apply_returncode=None,

                      patch_skip_reason="git_apply_instrument_" + reason,

                      patch_instrument_error=str(exc))

    return fields