"""Gate 6B sandboxed patch candidate execution.

Patch candidates still come from deterministic Gate 6A generation. Gate 6B may
apply an approved candidate only inside a disposable sandbox copy, then it runs
the candidate verifier plan and rolls back failed verifiers to a pre-patch
snapshot.
"""

from __future__ import annotations

import base64
from dataclasses import dataclass, field
import difflib
import os
from pathlib import Path
import shutil
import subprocess
import sys
from typing import Any

from .ledger import stable_hash
from .patch_candidates import ALLOWED_VERIFIER_PLAN, PatchCandidate, validate_patch_candidate
from .worktree_runner import MainCheckoutMutationSnapshot, main_checkout_mutation_count


_IGNORED_DIFF_PARTS = {".git", ".pytest_cache", "__pycache__"}
_IGNORED_SUFFIXES = {".pyc", ".pyo"}
_GENERATED_VERIFIER_CACHE_DIRS = {".pytest_cache", "__pycache__"}


@dataclass(slots=True)
class PatchSandboxSnapshot:
    sandbox_root: str
    files: dict[str, str]
    content_hash: str

    @classmethod
    def capture(cls, sandbox_root: str | Path) -> "PatchSandboxSnapshot":
        root = Path(sandbox_root).resolve()
        root.mkdir(parents=True, exist_ok=True)
        files: dict[str, str] = {}
        for item in sorted(root.rglob("*")):
            if item.is_symlink():
                raise ValueError(f"symlink is not allowed in patch sandbox: {item}")
            if not item.is_file():
                continue
            relative = item.relative_to(root).as_posix()
            if _ignored_path(relative):
                continue
            files[relative] = base64.b64encode(item.read_bytes()).decode("ascii")
        content_hash = stable_hash({"files": files})
        return cls(sandbox_root=str(root), files=files, content_hash=content_hash)

    def restore(self) -> "PatchSandboxSnapshot":
        root = Path(self.sandbox_root).resolve()
        expected = set(self.files)
        for item in sorted(root.rglob("*"), key=lambda path: len(path.parts), reverse=True):
            if item.is_file() and not item.is_symlink():
                relative = item.relative_to(root).as_posix()
                if relative not in expected or _ignored_path(relative):
                    item.unlink()
        for relative, encoded in self.files.items():
            target = (root / relative).resolve()
            if not target.is_relative_to(root):
                raise ValueError(f"snapshot restore path escapes sandbox: {relative}")
            target.parent.mkdir(parents=True, exist_ok=True)
            target.write_bytes(base64.b64decode(encoded.encode("ascii")))
        for item in sorted(root.rglob("*"), key=lambda path: len(path.parts), reverse=True):
            if item.is_dir() and item != root and not any(item.iterdir()):
                item.rmdir()
        return self.capture(root)

    def verify_restored(self) -> bool:
        return self.capture(self.sandbox_root).content_hash == self.content_hash

    def to_dict(self) -> dict[str, Any]:
        return {
            "sandbox_root": self.sandbox_root,
            "files": dict(self.files),
            "content_hash": self.content_hash,
        }

    @classmethod
    def from_dict(cls, data: dict[str, Any]) -> "PatchSandboxSnapshot":
        return cls(
            sandbox_root=str(data["sandbox_root"]),
            files={str(key): str(value) for key, value in dict(data.get("files") or {}).items()},
            content_hash=str(data["content_hash"]),
        )


@dataclass(slots=True)
class PatchDiffMinimizationAudit:
    candidate_id: str
    expected_target_path: str
    changed_files: list[str]
    changed_file_count: int
    diff_line_count: int
    diff_hash: str
    matches_candidate_scope: bool
    minimized: bool
    unsafe_changed_file_count: int
    diff_text: str = ""

    @classmethod
    def capture(
        cls,
        before_root: str | Path,
        after_root: str | Path,
        candidate: PatchCandidate,
    ) -> "PatchDiffMinimizationAudit":
        before_files = _text_files(Path(before_root))
        after_files = _text_files(Path(after_root))
        changed_files = sorted(
            relative for relative in set(before_files) | set(after_files) if before_files.get(relative) != after_files.get(relative)
        )
        chunks: list[str] = []
        for relative in changed_files:
            before_lines = before_files.get(relative, "").splitlines()
            after_lines = after_files.get(relative, "").splitlines()
            chunks.extend(
                difflib.unified_diff(
                    before_lines,
                    after_lines,
                    fromfile=f"a/{relative}",
                    tofile=f"b/{relative}",
                    lineterm="",
                )
            )
        diff_text = "\n".join(chunks)
        if diff_text and not diff_text.endswith("\n"):
            diff_text += "\n"
        expected_targets = [str(item) for item in candidate.diff_scope.get("target_paths", [candidate.target_path])]
        expected_line_limit = int(candidate.diff_scope.get("diff_line_count", 0))
        unsafe_changed_file_count = sum(1 for relative in changed_files if not _safe_relative_path(relative) or relative != candidate.target_path)
        matches_scope = (
            changed_files == expected_targets
            and len(changed_files) == int(candidate.diff_scope.get("changed_file_count", 0))
            and len(diff_text.splitlines()) <= expected_line_limit
        )
        return cls(
            candidate_id=candidate.candidate_id,
            expected_target_path=candidate.target_path,
            changed_files=changed_files,
            changed_file_count=len(changed_files),
            diff_line_count=len(diff_text.splitlines()),
            diff_hash=stable_hash({"changed_files": changed_files, "diff_text": diff_text}),
            matches_candidate_scope=matches_scope,
            minimized=matches_scope and unsafe_changed_file_count == 0,
            unsafe_changed_file_count=unsafe_changed_file_count,
            diff_text=diff_text,
        )

    def to_dict(self) -> dict[str, Any]:
        return {
            "candidate_id": self.candidate_id,
            "expected_target_path": self.expected_target_path,
            "changed_files": list(self.changed_files),
            "changed_file_count": self.changed_file_count,
            "diff_line_count": self.diff_line_count,
            "diff_hash": self.diff_hash,
            "matches_candidate_scope": self.matches_candidate_scope,
            "minimized": self.minimized,
            "unsafe_changed_file_count": self.unsafe_changed_file_count,
            "diff_text": self.diff_text,
        }

    @classmethod
    def from_dict(cls, data: dict[str, Any]) -> "PatchDiffMinimizationAudit":
        return cls(
            candidate_id=str(data["candidate_id"]),
            expected_target_path=str(data["expected_target_path"]),
            changed_files=[str(item) for item in data.get("changed_files", [])],
            changed_file_count=int(data.get("changed_file_count", 0)),
            diff_line_count=int(data.get("diff_line_count", 0)),
            diff_hash=str(data["diff_hash"]),
            matches_candidate_scope=bool(data.get("matches_candidate_scope", False)),
            minimized=bool(data.get("minimized", False)),
            unsafe_changed_file_count=int(data.get("unsafe_changed_file_count", 0)),
            diff_text=str(data.get("diff_text", "")),
        )


@dataclass(slots=True)
class SandboxPatchExecutionResult:
    execution_id: str
    candidate_id: str
    sandbox_root: str
    snapshot_root: str
    approved: bool
    executed: bool
    success: bool
    reason_code: str
    patch_apply_success: bool
    verifier_required_after_patch: bool
    verifier_after_patch: bool
    verifier_passed: bool
    verifier_override_count: int
    rollback_performed: bool
    rollback_success: bool
    snapshot_before_hash: str
    snapshot_after_patch_hash: str
    snapshot_after_rollback_hash: str
    diff_minimized: bool
    patch_escape_count: int
    real_repo_mutation_count: int
    learned_raw_patch_generation_count: int
    regression_count: int
    diff_minimization_audit: dict[str, Any] = field(default_factory=dict)
    verifier_summary: dict[str, Any] = field(default_factory=dict)
    patch_apply_summary: dict[str, Any] = field(default_factory=dict)
    mutation_proof: dict[str, Any] = field(default_factory=dict)

    def to_dict(self) -> dict[str, Any]:
        return {
            "execution_id": self.execution_id,
            "candidate_id": self.candidate_id,
            "sandbox_root": self.sandbox_root,
            "snapshot_root": self.snapshot_root,
            "approved": self.approved,
            "executed": self.executed,
            "success": self.success,
            "reason_code": self.reason_code,
            "patch_apply_success": self.patch_apply_success,
            "verifier_required_after_patch": self.verifier_required_after_patch,
            "verifier_after_patch": self.verifier_after_patch,
            "verifier_passed": self.verifier_passed,
            "verifier_override_count": self.verifier_override_count,
            "rollback_performed": self.rollback_performed,
            "rollback_success": self.rollback_success,
            "snapshot_before_hash": self.snapshot_before_hash,
            "snapshot_after_patch_hash": self.snapshot_after_patch_hash,
            "snapshot_after_rollback_hash": self.snapshot_after_rollback_hash,
            "diff_minimized": self.diff_minimized,
            "patch_escape_count": self.patch_escape_count,
            "real_repo_mutation_count": self.real_repo_mutation_count,
            "learned_raw_patch_generation_count": self.learned_raw_patch_generation_count,
            "regression_count": self.regression_count,
            "diff_minimization_audit": dict(self.diff_minimization_audit),
            "verifier_summary": dict(self.verifier_summary),
            "patch_apply_summary": dict(self.patch_apply_summary),
            "mutation_proof": dict(self.mutation_proof),
        }

    @classmethod
    def from_dict(cls, data: dict[str, Any]) -> "SandboxPatchExecutionResult":
        return cls(
            execution_id=str(data["execution_id"]),
            candidate_id=str(data["candidate_id"]),
            sandbox_root=str(data["sandbox_root"]),
            snapshot_root=str(data["snapshot_root"]),
            approved=bool(data["approved"]),
            executed=bool(data["executed"]),
            success=bool(data["success"]),
            reason_code=str(data["reason_code"]),
            patch_apply_success=bool(data["patch_apply_success"]),
            verifier_required_after_patch=bool(data["verifier_required_after_patch"]),
            verifier_after_patch=bool(data["verifier_after_patch"]),
            verifier_passed=bool(data["verifier_passed"]),
            verifier_override_count=int(data.get("verifier_override_count", 0)),
            rollback_performed=bool(data.get("rollback_performed", False)),
            rollback_success=bool(data.get("rollback_success", False)),
            snapshot_before_hash=str(data.get("snapshot_before_hash", "")),
            snapshot_after_patch_hash=str(data.get("snapshot_after_patch_hash", "")),
            snapshot_after_rollback_hash=str(data.get("snapshot_after_rollback_hash", "")),
            diff_minimized=bool(data.get("diff_minimized", False)),
            patch_escape_count=int(data.get("patch_escape_count", 0)),
            real_repo_mutation_count=int(data.get("real_repo_mutation_count", 0)),
            learned_raw_patch_generation_count=int(data.get("learned_raw_patch_generation_count", 0)),
            regression_count=int(data.get("regression_count", 0)),
            diff_minimization_audit=dict(data.get("diff_minimization_audit") or {}),
            verifier_summary=dict(data.get("verifier_summary") or {}),
            patch_apply_summary=dict(data.get("patch_apply_summary") or {}),
            mutation_proof=dict(data.get("mutation_proof") or {}),
        )


class SandboxPatchExecutor:
    def __init__(
        self,
        state_dir: str | Path,
        *,
        main_checkout_root: str | Path | None = None,
        timeout_seconds: float = 120.0,
    ) -> None:
        self.state_dir = Path(state_dir)
        self.state_dir.mkdir(parents=True, exist_ok=True)
        self.main_checkout_root = Path(main_checkout_root).resolve() if main_checkout_root is not None else None
        self.timeout_seconds = timeout_seconds

    def execute_candidate(
        self,
        *,
        execution_id: str,
        source_root: str | Path,
        candidate: PatchCandidate,
        expected_verifier_success: bool,
    ) -> SandboxPatchExecutionResult:
        isolation_root = self.state_dir / _safe_name(execution_id)
        sandbox_root = isolation_root / "sandbox"
        snapshot_root = isolation_root / "snapshot_before_patch"
        if isolation_root.exists():
            shutil.rmtree(isolation_root)
        isolation_root.mkdir(parents=True, exist_ok=True)
        shutil.copytree(source_root, sandbox_root, ignore=_copy_ignore)
        shutil.copytree(source_root, snapshot_root, ignore=_copy_ignore)

        main_before = _capture_main_checkout(self.main_checkout_root)
        valid, reason = validate_patch_candidate(candidate)
        snapshot = PatchSandboxSnapshot.capture(sandbox_root)
        if not valid:
            main_after = _capture_main_checkout(self.main_checkout_root)
            return self._not_executed(
                execution_id,
                candidate,
                sandbox_root,
                snapshot_root,
                snapshot,
                reason,
                main_before,
                main_after,
            )
        if self.main_checkout_root is not None and sandbox_root.resolve().is_relative_to(self.main_checkout_root):
            main_after = _capture_main_checkout(self.main_checkout_root)
            return self._not_executed(
                execution_id,
                candidate,
                sandbox_root,
                snapshot_root,
                snapshot,
                "sandbox_inside_real_repo_forbidden",
                main_before,
                main_after,
            )

        patch_summary = _apply_patch_candidate(sandbox_root, candidate)
        if not bool(patch_summary["success"]):
            after_patch = PatchSandboxSnapshot.capture(sandbox_root)
            restored = snapshot.restore()
            main_after = _capture_main_checkout(self.main_checkout_root)
            mutation_count = _main_mutation_count(main_before, main_after)
            return SandboxPatchExecutionResult(
                execution_id=execution_id,
                candidate_id=candidate.candidate_id,
                sandbox_root=str(sandbox_root.resolve()),
                snapshot_root=str(snapshot_root.resolve()),
                approved=True,
                executed=False,
                success=False,
                reason_code=str(patch_summary["reason"]),
                patch_apply_success=False,
                verifier_required_after_patch=True,
                verifier_after_patch=False,
                verifier_passed=False,
                verifier_override_count=0,
                rollback_performed=True,
                rollback_success=restored.content_hash == snapshot.content_hash,
                snapshot_before_hash=snapshot.content_hash,
                snapshot_after_patch_hash=after_patch.content_hash,
                snapshot_after_rollback_hash=restored.content_hash,
                diff_minimized=False,
                patch_escape_count=0,
                real_repo_mutation_count=mutation_count,
                learned_raw_patch_generation_count=0,
                regression_count=int(expected_verifier_success),
                patch_apply_summary=patch_summary,
                mutation_proof=_mutation_proof(main_before, main_after, mutation_count),
            )

        verifier_summary = _run_verifier_commands(sandbox_root, candidate.verifier_plan, timeout_seconds=self.timeout_seconds)
        _remove_generated_verifier_caches(sandbox_root)
        after_patch = PatchSandboxSnapshot.capture(sandbox_root)
        audit = PatchDiffMinimizationAudit.capture(snapshot_root, sandbox_root, candidate)
        patch_escape_count = int(not audit.minimized) + int(audit.unsafe_changed_file_count)
        verifier_passed = bool(verifier_summary.get("passed", False))
        rollback_performed = not verifier_passed
        rollback_success = False
        restored_hash = ""
        if rollback_performed:
            restored = snapshot.restore()
            _remove_generated_verifier_caches(sandbox_root)
            restored = PatchSandboxSnapshot.capture(sandbox_root)
            restored_hash = restored.content_hash
            rollback_success = restored_hash == snapshot.content_hash

        main_after = _capture_main_checkout(self.main_checkout_root)
        mutation_count = _main_mutation_count(main_before, main_after)
        success = verifier_passed and audit.minimized and mutation_count == 0 and patch_escape_count == 0
        regression_count = int(expected_verifier_success and not success)
        reason_code = "verifier_passed" if success else "verifier_failed_and_rolled_back"
        if patch_escape_count:
            reason_code = "patch_escape_detected"
        elif mutation_count:
            reason_code = "real_repo_mutation_detected"
        return SandboxPatchExecutionResult(
            execution_id=execution_id,
            candidate_id=candidate.candidate_id,
            sandbox_root=str(sandbox_root.resolve()),
            snapshot_root=str(snapshot_root.resolve()),
            approved=True,
            executed=True,
            success=success,
            reason_code=reason_code,
            patch_apply_success=True,
            verifier_required_after_patch=True,
            verifier_after_patch=True,
            verifier_passed=verifier_passed,
            verifier_override_count=0,
            rollback_performed=rollback_performed,
            rollback_success=rollback_success,
            snapshot_before_hash=snapshot.content_hash,
            snapshot_after_patch_hash=after_patch.content_hash,
            snapshot_after_rollback_hash=restored_hash,
            diff_minimized=audit.minimized,
            patch_escape_count=patch_escape_count,
            real_repo_mutation_count=mutation_count,
            learned_raw_patch_generation_count=0,
            regression_count=regression_count,
            diff_minimization_audit=audit.to_dict(),
            verifier_summary=verifier_summary,
            patch_apply_summary=patch_summary,
            mutation_proof=_mutation_proof(main_before, main_after, mutation_count),
        )

    def _not_executed(
        self,
        execution_id: str,
        candidate: PatchCandidate,
        sandbox_root: Path,
        snapshot_root: Path,
        snapshot: PatchSandboxSnapshot,
        reason: str,
        main_before: MainCheckoutMutationSnapshot | None,
        main_after: MainCheckoutMutationSnapshot | None,
    ) -> SandboxPatchExecutionResult:
        mutation_count = _main_mutation_count(main_before, main_after)
        return SandboxPatchExecutionResult(
            execution_id=execution_id,
            candidate_id=candidate.candidate_id,
            sandbox_root=str(sandbox_root.resolve()),
            snapshot_root=str(snapshot_root.resolve()),
            approved=False,
            executed=False,
            success=False,
            reason_code=reason,
            patch_apply_success=False,
            verifier_required_after_patch=True,
            verifier_after_patch=False,
            verifier_passed=False,
            verifier_override_count=0,
            rollback_performed=False,
            rollback_success=False,
            snapshot_before_hash=snapshot.content_hash,
            snapshot_after_patch_hash=snapshot.content_hash,
            snapshot_after_rollback_hash="",
            diff_minimized=False,
            patch_escape_count=0,
            real_repo_mutation_count=mutation_count,
            learned_raw_patch_generation_count=0,
            regression_count=0,
            verifier_summary={"passed": False, "command_count": 0, "records": []},
            patch_apply_summary={"success": False, "reason": reason},
            mutation_proof=_mutation_proof(main_before, main_after, mutation_count),
        )


def _apply_patch_candidate(root: Path, candidate: PatchCandidate) -> dict[str, Any]:
    target = (root / candidate.target_path).resolve()
    if not target.is_relative_to(root):
        return {"success": False, "reason": "target_path_escapes_sandbox", "target_path": candidate.target_path}
    if not target.exists():
        return {"success": False, "reason": "target_file_missing", "target_path": candidate.target_path}
    before = target.read_text(encoding="utf-8")
    patched, reason = _apply_unified_diff(before, candidate.diff_text)
    if patched is None:
        return {"success": False, "reason": reason, "target_path": candidate.target_path}
    target.write_text(patched, encoding="utf-8")
    return {
        "success": True,
        "reason": "applied_from_deterministic_patch_candidate",
        "target_path": candidate.target_path,
        "diff_hash": stable_hash({"diff_text": candidate.diff_text}),
        "deterministic_candidate_id": candidate.candidate_id,
    }


def _apply_unified_diff(content: str, diff_text: str) -> tuple[str | None, str]:
    old_lines, new_lines = _parse_old_new_blocks(diff_text)
    if not old_lines:
        return None, "diff_has_no_removal_or_context"
    content_lines = content.splitlines()
    start = _find_subsequence(content_lines, old_lines)
    if start < 0:
        return None, "diff_context_not_found"
    patched_lines = content_lines[:start] + new_lines + content_lines[start + len(old_lines) :]
    trailing_newline = "\n" if content.endswith("\n") else ""
    return "\n".join(patched_lines) + trailing_newline, "ok"


def _parse_old_new_blocks(diff_text: str) -> tuple[list[str], list[str]]:
    old_lines: list[str] = []
    new_lines: list[str] = []
    in_hunk = False
    for line in diff_text.splitlines():
        if line.startswith("--- ") or line.startswith("+++ "):
            continue
        if line.startswith("@@"):
            in_hunk = True
            continue
        if not in_hunk or not line:
            continue
        prefix, value = line[0], line[1:]
        if prefix == " ":
            old_lines.append(value)
            new_lines.append(value)
        elif prefix == "-":
            old_lines.append(value)
        elif prefix == "+":
            new_lines.append(value)
    return old_lines, new_lines


def _find_subsequence(lines: list[str], needle: list[str]) -> int:
    if len(needle) > len(lines):
        return -1
    for index in range(len(lines) - len(needle) + 1):
        if lines[index : index + len(needle)] == needle:
            return index
    return -1


def _run_verifier_commands(root: Path, verifier_commands: list[list[str]], *, timeout_seconds: float) -> dict[str, Any]:
    records: list[dict[str, Any]] = []
    env = {
        **os.environ,
        "PYTHONPATH": str(root / "src"),
        "PYTHONDONTWRITEBYTECODE": "1",
        "PYTEST_DISABLE_PLUGIN_AUTOLOAD": "1",
        "PYTEST_ADDOPTS": "",
        "PYTHONHASHSEED": "0",
    }
    env.pop("PYTEST_CURRENT_TEST", None)
    for command in verifier_commands:
        command_tuple = tuple(str(item) for item in command)
        if command_tuple not in ALLOWED_VERIFIER_PLAN:
            records.append({"command": list(command_tuple), "success": False, "reason": "verifier_command_not_allowed"})
            continue
        try:
            executable_command = [sys.executable, *command_tuple[1:]] if command_tuple and command_tuple[0] == "python" else list(command_tuple)
            completed = subprocess.run(
                executable_command,
                cwd=root,
                check=False,
                capture_output=True,
                text=True,
                timeout=timeout_seconds,
                env=env,
            )
        except subprocess.TimeoutExpired as exc:
            records.append(
                {
                    "command": list(command_tuple),
                    "success": False,
                    "reason": "verifier_timeout",
                    "stdout_tail": str(exc.stdout or "")[-500:],
                    "stderr_tail": str(exc.stderr or "")[-500:],
                }
            )
            continue
        records.append(
            {
                "command": list(command_tuple),
                "success": completed.returncode == 0,
                "returncode": completed.returncode,
                "stdout_tail": completed.stdout[-500:],
                "stderr_tail": completed.stderr[-500:],
            }
        )
    passed = bool(records) and all(bool(record.get("success", False)) for record in records)
    return {
        "passed": passed,
        "command_count": len(records),
        "pass_count": sum(1 for record in records if bool(record.get("success", False))),
        "records": records,
    }


def _text_files(root: Path) -> dict[str, str]:
    files: dict[str, str] = {}
    for item in sorted(root.rglob("*")):
        if item.is_dir() or item.is_symlink():
            continue
        relative = item.relative_to(root).as_posix()
        if _ignored_path(relative):
            continue
        try:
            files[relative] = item.read_text(encoding="utf-8")
        except UnicodeDecodeError:
            continue
    return files


def _copy_ignore(_directory: str, names: list[str]) -> set[str]:
    return {name for name in names if name in _IGNORED_DIFF_PARTS or Path(name).suffix in _IGNORED_SUFFIXES}


def _ignored_path(relative: str) -> bool:
    path = Path(relative)
    return any(part in _IGNORED_DIFF_PARTS for part in path.parts) or path.suffix in _IGNORED_SUFFIXES


def _safe_relative_path(value: str) -> bool:
    path = Path(value)
    return bool(value) and not path.is_absolute() and ".." not in path.parts and not _ignored_path(value)


def _remove_generated_verifier_caches(root: Path) -> None:
    for item in sorted(root.rglob("*"), key=lambda path: len(path.parts), reverse=True):
        if item.is_dir() and item.name in _GENERATED_VERIFIER_CACHE_DIRS:
            shutil.rmtree(item, ignore_errors=True)
        elif item.is_file() and item.suffix in _IGNORED_SUFFIXES:
            try:
                item.unlink()
            except OSError:
                pass


def _safe_name(value: str) -> str:
    return "".join(ch if ch.isalnum() or ch in {"-", "_"} else "_" for ch in value)[:120]


def _capture_main_checkout(root: Path | None) -> MainCheckoutMutationSnapshot | None:
    if root is None:
        return None
    return MainCheckoutMutationSnapshot.capture(root)


def _main_mutation_count(
    before: MainCheckoutMutationSnapshot | None,
    after: MainCheckoutMutationSnapshot | None,
) -> int:
    if before is None or after is None:
        return 0
    return main_checkout_mutation_count(before, after)


def _mutation_proof(
    before: MainCheckoutMutationSnapshot | None,
    after: MainCheckoutMutationSnapshot | None,
    mutation_count: int,
) -> dict[str, Any]:
    return {
        "main_checkout_checked": before is not None and after is not None,
        "real_repo_mutation_count": mutation_count,
        "before": before.to_dict() if before is not None else None,
        "after": after.to_dict() if after is not None else None,
    }
