"""Bisect engine — binary search through commit history to locate regressions.

``muse bisect`` is a power-tool for both human engineers and autonomous agents.
Given a known-bad commit and a known-good commit, it performs a binary search
through the commits between them, asking at each midpoint "is this good or
bad?" until the first bad commit is isolated.

Agent-safety
------------
The ``run`` subcommand accepts an arbitrary shell command.  The command is
executed in a subprocess; the exit code determines the verdict:

    0         → good (the bug is NOT present)
    125       → skip (cannot test this commit — e.g. build fails)
    any other → bad  (the bug IS present)

This mirrors Git's bisect protocol so any existing test harness works without
modification.

Symbol-scoped bisect
--------------------
Pass ``symbol_filter`` to ``start_bisect`` to restrict the candidate commit
list to only those commits whose ``structured_delta`` contains an op for the
given symbol address (or any symbol whose address starts with that prefix,
enabling class-level filtering like ``billing.py::Invoice``).

This typically reduces the search space from hundreds of commits to fewer
than ten — fewer binary-search steps, and each displayed midpoint shows
exactly which ops on that symbol occurred.

State file
----------
Bisect state is stored at ``.muse/BISECT_STATE.toml``::

    bad_id = "<sha256>"
    good_ids = ["<sha256>", …]
    skipped_ids = ["<sha256>", …]
    remaining = ["<sha256>", …]        # sorted oldest-first
    log = ["<sha256> <verdict> <ts>", …]
    symbol_filter = "billing.py::Invoice.compute_total"   # optional
    branch = "feat/my-feature"                             # optional

The remaining list is rebuilt at every step so it tolerates interruptions.

Security model
--------------
- **TOML injection**: All user-controlled string fields written to the state
  file (``branch``, ``symbol_filter``) are passed through ``_toml_escape``
  before serialisation.  Commit IDs are always SHA-256 hex strings and need
  no escaping.
- **Symlink guard**: ``_load_state`` rejects a symlink at the state file path
  to prevent path-traversal attacks that could redirect reads or writes to
  files outside the repository.
- **Size cap**: ``_load_state`` refuses to read a state file larger than
  ``_MAX_STATE_BYTES`` (4 MiB) to prevent memory exhaustion from a tampered
  or corrupt state.
"""

import datetime
import logging
import pathlib
import subprocess
from dataclasses import dataclass, field
from typing import TypedDict

from muse.core.types import MUSE_DIR, now_utc_iso
from muse.core.paths import muse_dir as _muse_dir

logger = logging.getLogger(__name__)

_SKIP_EXIT_CODE = 125

# Guard against pathologically large or tampered state files consuming memory.
_MAX_STATE_BYTES: int = 4 * 1024 * 1024  # 4 MiB

# ---------------------------------------------------------------------------
# State persistence
# ---------------------------------------------------------------------------

class BisectStateDict(TypedDict, total=False):
    """On-disk shape of the bisect state."""

    bad_id: str
    good_ids: list[str]
    skipped_ids: list[str]
    remaining: list[str]
    log: list[str]
    branch: str
    symbol_filter: str

class BisectStatusDict(TypedDict, total=False):
    """Public session summary returned by :func:`get_bisect_status`.

    ``active`` is always present.  All other keys are only present when
    ``active`` is ``True``.
    """

    active: bool
    bad_id: str
    good_ids: list[str]
    symbol_filter: str
    remaining_count: int
    steps_remaining: int
    skipped_count: int
    branch: str

def _state_path(repo_root: pathlib.Path) -> pathlib.Path:
    return repo_root / MUSE_DIR / "BISECT_STATE.toml"

def _load_state(repo_root: pathlib.Path) -> BisectStateDict | None:
    """Return the bisect state if a session is active, else None.

    Safety guards applied before any read:

    - Symlink check: a symlink at the state path could redirect writes to
      arbitrary locations or reads to sensitive files outside the repo.
    - Size cap (``_MAX_STATE_BYTES``): a tampered or corrupt state file cannot
      be used to exhaust memory.
    """
    import tomllib

    path = _state_path(repo_root)
    if not path.exists():
        return None
    # Reject symlinks before any read/write to prevent path-traversal attacks.
    if path.is_symlink():
        logger.warning("⚠️ Bisect state file is a symlink — ignoring to prevent path traversal")
        return None
    try:
        size = path.stat().st_size
        if size > _MAX_STATE_BYTES:
            logger.warning(
                "⚠️ Bisect state file is %.1f MiB — exceeds cap of %d MiB; ignoring",
                size / (1024 * 1024),
                _MAX_STATE_BYTES // (1024 * 1024),
            )
            return None
        raw = tomllib.loads(path.read_text(encoding="utf-8"))
    except Exception as exc:
        logger.warning("⚠️ Could not read bisect state: %s", exc)
        return None
    state: BisectStateDict = {}
    if "bad_id" in raw and isinstance(raw["bad_id"], str):
        state["bad_id"] = raw["bad_id"]
    if "branch" in raw and isinstance(raw["branch"], str):
        state["branch"] = raw["branch"]
    if "symbol_filter" in raw and isinstance(raw["symbol_filter"], str):
        state["symbol_filter"] = raw["symbol_filter"]
    for key in ("good_ids", "skipped_ids", "remaining", "log"):
        val = raw.get(key)
        if isinstance(val, list) and all(isinstance(x, str) for x in val):
            str_list: list[str] = list(val)
            if key == "good_ids":
                state["good_ids"] = str_list
            elif key == "skipped_ids":
                state["skipped_ids"] = str_list
            elif key == "remaining":
                state["remaining"] = str_list
            elif key == "log":
                state["log"] = str_list
    return state

def _toml_escape(value: str) -> str:
    """Escape *value* for safe embedding inside a TOML double-quoted string.

    Only backslash and double-quote need escaping in TOML basic strings.
    Applied to all free-form string fields (branch name, symbol_filter) before
    serialisation to prevent TOML injection via crafted branch names.
    """
    return value.replace("\\", "\\\\").replace('"', '\\"')

def _save_state(repo_root: pathlib.Path, state: BisectStateDict) -> None:
    """Write the bisect state to disk (TOML, atomic).

    All string values that may contain user-controlled content (branch name,
    symbol_filter) are escaped with ``_toml_escape`` before serialisation.
    Commit IDs are always 64-char hex strings and need no escaping.
    """
    path = _state_path(repo_root)
    lines = [f'bad_id = "{state.get("bad_id", "")}"']
    if "branch" in state:
        # Branch names can contain backslashes and quotes — escape both.
        lines.append(f'branch = "{_toml_escape(state["branch"])}"')
    if "symbol_filter" in state and state["symbol_filter"]:
        lines.append(f'symbol_filter = "{_toml_escape(state["symbol_filter"])}"')
    for key, items in (
        ("good_ids", state.get("good_ids", [])),
        ("skipped_ids", state.get("skipped_ids", [])),
        ("remaining", state.get("remaining", [])),
        ("log", state.get("log", [])),
    ):
        formatted = ", ".join(f'"{v}"' for v in items)
        lines.append(f"{key} = [{formatted}]")
    tmp = path.with_suffix(".tmp")
    tmp.write_text("\n".join(lines) + "\n", encoding="utf-8")
    tmp.replace(path)

# ---------------------------------------------------------------------------
# Commit graph helpers
# ---------------------------------------------------------------------------

def _ancestors(
    repo_root: pathlib.Path,
    start_id: str,
) -> list[str]:
    """Return commit IDs from start_id back to root, oldest first."""
    from muse.core.graph import iter_ancestors
    return [c.commit_id for c in reversed(list(iter_ancestors(repo_root, start_id)))]

def _reachable_from_good(
    repo_root: pathlib.Path,
    good_ids: list[str],
) -> set[str]:
    """Return all commits reachable (inclusive) from any good commit."""
    from muse.core.graph import ancestor_ids
    return ancestor_ids(repo_root, good_ids)

def _build_remaining(
    repo_root: pathlib.Path,
    bad_id: str,
    good_ids: list[str],
    skipped_ids: list[str],
) -> list[str]:
    """Return commits strictly between good and bad (exclusive of both endpoints).

    Excludes ``bad_id`` itself — it is already the known-bad boundary.
    Excludes good-reachable commits (already confirmed good).
    Excludes skipped commits.

    When this list has length 0, ``bad_id`` is the first bad commit.
    When it has length 1, there is exactly one candidate to test — if it is
    bad, it becomes the new ``bad_id`` and remaining collapses to 0; if it is
    good, ``bad_id`` is confirmed as first-bad.
    """
    ancestors_of_bad = _ancestors(repo_root, bad_id)
    good_reachable = _reachable_from_good(repo_root, good_ids)
    skipped = set(skipped_ids)
    return [
        cid for cid in ancestors_of_bad
        if cid not in good_reachable and cid not in skipped and cid != bad_id
    ]

def _midpoint(remaining: list[str]) -> str | None:
    if not remaining:
        return None
    return remaining[len(remaining) // 2]

def _addr_matches(addr: str, symbol_filter: str) -> bool:
    """Return True if *addr* is the filtered symbol or a child of it.

    Exact match covers ``file.py::func``.
    Prefix match (``addr.startswith(symbol_filter + ".")``) covers class
    methods when the filter is ``file.py::ClassName``.
    """
    return addr == symbol_filter or addr.startswith(f"{symbol_filter}.")

def _commits_touching_symbol(
    repo_root: pathlib.Path,
    commit_ids: list[str],
    symbol_filter: str,
) -> list[str]:
    """Return the subset of *commit_ids* whose structured_delta touched *symbol_filter*.

    Preserves the original order of *commit_ids*.  Commits with no
    ``structured_delta`` (genesis commits) are excluded.
    """
    from muse.core.commits import read_commit
    from muse.plugins.code._query import flat_symbol_ops

    result: list[str] = []
    for cid in commit_ids:
        commit = read_commit(repo_root, cid)
        if commit is None or commit.structured_delta is None or commit.parent_commit_id is None:
            continue
        ops_list = commit.structured_delta.get("ops", [])
        for op in flat_symbol_ops(ops_list):
            if _addr_matches(op.get("address", ""), symbol_filter):
                result.append(cid)
                break
    return result

def _symbol_ops_in_commit(
    repo_root: pathlib.Path,
    commit_id: str,
    symbol_filter: str,
) -> list[str]:
    """Return human-readable descriptions of ops for *symbol_filter* in *commit_id*.

    Used to display what changed in the midpoint commit before the user tests it.
    Returns an empty list when the commit has no relevant delta.
    """
    from muse.core.commits import read_commit
    from muse.plugins.code._query import flat_symbol_ops

    commit = read_commit(repo_root, commit_id)
    if commit is None or commit.structured_delta is None:
        return []

    descriptions: list[str] = []
    ops_list = commit.structured_delta.get("ops", [])
    for op in flat_symbol_ops(ops_list):
        addr = op.get("address", "")
        if not _addr_matches(addr, symbol_filter):
            continue
        op_type = op.get("op", "?")
        old = op.get("old_summary", "")
        new = op.get("new_summary", op.get("content_summary", ""))
        if op_type == "replace":
            descriptions.append(f"  {addr}  →  {old!r} → {new!r}")
        elif op_type == "insert":
            descriptions.append(f"  {addr}  added  {new!r}")
        elif op_type == "delete":
            descriptions.append(f"  {addr}  removed  {old!r}")
        else:
            descriptions.append(f"  {addr}  {op_type}")
    return descriptions

# ---------------------------------------------------------------------------
# Result type
# ---------------------------------------------------------------------------

@dataclass
class BisectResult:
    """Result of a single bisect step."""

    done: bool = False
    """True when we have isolated the first bad commit."""

    first_bad: str | None = None
    """Commit ID of the isolated first-bad commit."""

    next_to_test: str | None = None
    """Commit ID to test next (None when done)."""

    remaining_count: int = 0
    """How many commits remain to test."""

    steps_remaining: int = 0
    """Approximate remaining binary-search steps."""

    verdict: str = ""
    """The verdict just applied: 'good', 'bad', 'skip'."""

    symbol_changes: list[str] = field(default_factory=list)
    """Human-readable descriptions of symbol ops in *next_to_test*.

    Only populated when a symbol_filter is active.  Empty list otherwise.
    """

# ---------------------------------------------------------------------------
# Public API
# ---------------------------------------------------------------------------

def start_bisect(
    repo_root: pathlib.Path,
    bad_id: str,
    good_ids: list[str],
    branch: str = "",
    symbol_filter: str = "",
) -> BisectResult:
    """Start a new bisect session.

    Args:
        repo_root:      Repository root.
        bad_id:         Known-bad commit.
        good_ids:       One or more known-good commits.
        branch:         Current branch name (for display only).
        symbol_filter:  Optional symbol address (e.g. ``billing.py::Invoice.compute_total``).
                        When set, only commits whose structured_delta touched this
                        symbol are included in the search space.
    """
    import math

    remaining = _build_remaining(repo_root, bad_id, good_ids, skipped_ids=[])

    # Narrow the search space to commits that touched the requested symbol.
    if symbol_filter:
        remaining = _commits_touching_symbol(repo_root, remaining, symbol_filter)

    state: BisectStateDict = {
        "bad_id": bad_id,
        "good_ids": good_ids,
        "skipped_ids": [],
        "remaining": remaining,
        "log": [
            f"{bad_id} bad {_ts()}",
            *[f"{g} good {_ts()}" for g in good_ids],
        ],
    }
    if branch:
        state["branch"] = branch
    if symbol_filter:
        state["symbol_filter"] = symbol_filter
    _save_state(repo_root, state)

    if not remaining:
        # bad_id is immediately the first bad commit (no unknowns).
        return BisectResult(
            done=True,
            first_bad=bad_id,
            next_to_test=None,
            remaining_count=0,
            steps_remaining=0,
            verdict="started",
        )
    next_id = _midpoint(remaining)
    steps = int(math.log2(len(remaining) + 1))
    changes: list[str] = []
    if symbol_filter and next_id:
        changes = _symbol_ops_in_commit(repo_root, next_id, symbol_filter)
    return BisectResult(
        done=False,
        first_bad=None,
        next_to_test=next_id,
        remaining_count=len(remaining),
        steps_remaining=steps,
        verdict="started",
        symbol_changes=changes,
    )

def _ts() -> str:
    return now_utc_iso()

def _apply_verdict(
    repo_root: pathlib.Path,
    commit_id: str,
    verdict: str,
) -> BisectResult:
    """Apply a 'good', 'bad', or 'skip' verdict and return the next step."""
    import math

    state = _load_state(repo_root)
    if state is None:
        raise RuntimeError("No bisect session in progress. Run 'muse bisect start' first.")

    bad_id = state.get("bad_id", "")
    good_ids = list(state.get("good_ids", []))
    skipped_ids = list(state.get("skipped_ids", []))
    log = list(state.get("log", []))
    symbol_filter = state.get("symbol_filter", "")

    if verdict == "good":
        good_ids.append(commit_id)
    elif verdict == "bad":
        bad_id = commit_id
    else:
        skipped_ids.append(commit_id)

    log.append(f"{commit_id} {verdict} {_ts()}")

    remaining = _build_remaining(repo_root, bad_id, good_ids, skipped_ids)

    # Re-apply the symbol filter so newly included commits are also filtered.
    if symbol_filter:
        remaining = _commits_touching_symbol(repo_root, remaining, symbol_filter)

    new_state: BisectStateDict = {
        "bad_id": bad_id,
        "good_ids": good_ids,
        "skipped_ids": skipped_ids,
        "remaining": remaining,
        "log": log,
    }
    if "branch" in state:
        new_state["branch"] = state["branch"]
    if symbol_filter:
        new_state["symbol_filter"] = symbol_filter
    _save_state(repo_root, new_state)

    if len(remaining) == 0:
        # No unknowns remain — bad_id is confirmed as the first bad commit.
        return BisectResult(
            done=True,
            first_bad=bad_id,
            next_to_test=None,
            remaining_count=0,
            steps_remaining=0,
            verdict=verdict,
        )

    next_id = _midpoint(remaining)
    steps = int(math.log2(len(remaining) + 1))
    changes: list[str] = []
    if symbol_filter and next_id:
        changes = _symbol_ops_in_commit(repo_root, next_id, symbol_filter)
    return BisectResult(
        done=False,
        first_bad=None,
        next_to_test=next_id,
        remaining_count=len(remaining),
        steps_remaining=steps,
        verdict=verdict,
        symbol_changes=changes,
    )

def mark_good(repo_root: pathlib.Path, commit_id: str) -> BisectResult:
    """Mark *commit_id* as good and advance the bisect."""
    return _apply_verdict(repo_root, commit_id, "good")

def mark_bad(repo_root: pathlib.Path, commit_id: str) -> BisectResult:
    """Mark *commit_id* as bad and advance the bisect."""
    return _apply_verdict(repo_root, commit_id, "bad")

def skip_commit(repo_root: pathlib.Path, commit_id: str) -> BisectResult:
    """Skip *commit_id* (e.g. build fails) and advance the bisect."""
    return _apply_verdict(repo_root, commit_id, "skip")

def reset_bisect(repo_root: pathlib.Path) -> None:
    """End the bisect session and remove state."""
    path = _state_path(repo_root)
    if path.exists():
        path.unlink()

def get_bisect_log(repo_root: pathlib.Path) -> list[str]:
    """Return the bisect log entries, oldest first."""
    state = _load_state(repo_root)
    if state is None:
        return []
    return list(state.get("log", []))

def get_bisect_next(
    repo_root: pathlib.Path,
) -> tuple[str | None, str]:
    """Return ``(next_to_test, symbol_filter)`` from the current bisect state.

    This is the public alternative to importing the private ``_load_state``
    directly from the CLI layer.  Returns ``(None, "")`` when no session is
    active or remaining is empty.
    """
    state = _load_state(repo_root)
    if state is None:
        return None, ""
    remaining = state.get("remaining", [])
    symbol_filter = state.get("symbol_filter", "")
    if not remaining:
        return None, symbol_filter
    return _midpoint(remaining), symbol_filter

def is_bisect_active(repo_root: pathlib.Path) -> bool:
    """Return True if a bisect session is in progress."""
    return _state_path(repo_root).exists()

def get_bisect_status(repo_root: pathlib.Path) -> BisectStatusDict:
    """Return a structured read-only summary of the current bisect session.

    Always returns a dict with ``active`` set.  When ``active`` is ``False``
    all other fields are absent.  Agents should call this to inspect session
    state without advancing or modifying anything.

    Returns:
        ``{"active": False}`` when no session exists, or::

            {
                "active":          True,
                "bad_id":          "<commit-id>",
                "good_ids":        ["<commit-id>", ...],
                "symbol_filter":   "",
                "remaining_count": <int>,
                "steps_remaining": <int>,
                "skipped_count":   <int>,
                "branch":          "<branch-name>",
            }
    """
    import math

    state = _load_state(repo_root)
    if state is None:
        return BisectStatusDict(active=False)
    remaining = state.get("remaining", [])
    steps = int(math.log2(len(remaining) + 1)) if remaining else 0
    status = BisectStatusDict(
        active=True,
        bad_id=state.get("bad_id", ""),
        good_ids=list(state.get("good_ids", [])),
        symbol_filter=state.get("symbol_filter", ""),
        remaining_count=len(remaining),
        steps_remaining=steps,
        skipped_count=len(state.get("skipped_ids", [])),
    )
    if "branch" in state:
        status["branch"] = state["branch"]
    return status

def run_bisect_command(
    repo_root: pathlib.Path,
    command: str,
    current_commit_id: str,
    timeout: int | None = None,
) -> BisectResult:
    """Run *command* in a shell, interpret exit code, and apply verdict.

    Exit codes::

        0    → good
        125  → skip (also triggered when *timeout* is exceeded)
        1-124, 126-255 → bad

    The command is executed with the repository root as the working directory.
    When *timeout* is set and the command exceeds it, the process is killed and
    the commit is treated as untestable (skip), identical to exit code 125.
    """
    try:
        proc = subprocess.run(
            command,
            shell=True,
            cwd=str(repo_root),
            timeout=timeout,
        )
        code = proc.returncode
    except subprocess.TimeoutExpired:
        code = _SKIP_EXIT_CODE  # timed-out commits are untestable → skip
    if code == 0:
        verdict = "good"
    elif code == _SKIP_EXIT_CODE:
        verdict = "skip"
    else:
        verdict = "bad"
    return _apply_verdict(repo_root, current_commit_id, verdict)