"""Trial engine: copy a pristine fixture, inject one fault, judge invariants.

Invariants (each True / False / None=unmeasured, ADR-0002):

- ``op_no_hang``            the faulted operation ended (killed or exited) within its timeout

- ``repo_opens``            independent open: SQLite integrity_check (Oak) / git fsck (Git)

- ``acked_local_reachable`` every commit a local ref named before the op is still reachable

                            from a local ref (branch head / HEAD / tag), per the independent reader

- ``server_intact``         remote store opens, its pre-op heads stay reachable, and the branch

                            head is either the pre-op head or the operation's intended head

- ``worktree_consistent``   the worktree equals the pre- or post-op state, or a committed tree,

                            or the subject's status reports it dirty/in-progress (never "clean"

                            while differing from HEAD)

- ``next_op_ok_or_actionable`` rerunning the operation succeeds, or fails with a typed/actionable

                            message (not a hang, panic, or signal)

- ``no_false_success``      a reported success (exit 0) leaves exactly the calibrated outcome

- ``acked_after_next``      pre-op local commits are still reachable after the next operation

"""

from __future__ import annotations

import json

import re

import shutil

import time

from dataclasses import dataclass, field

from pathlib import Path

from typing import Any, Callable, Optional

from . import FAULT_NOT_FIRED, HARNESS_ERROR, HELD, VIOLATED

from .adapters import GitAdapter, OakAdapter, Server

from .fixtures import Fixture

from .procs import Result, copy_tree

from .readers import OakGraph

# A failure is actionable only if it tells the user what to do next:

# a concrete recovery command (a real subcommand of the subject, or a recovery

# flag), an explicit manual step, a documented retryable condition, or a typed,

# specific JSON error code / recommended command list. Mentioning "oak" or

# "git" is not enough ("failed to open oak database", "not a git repository").

RECOVERY_SUBCOMMANDS = (

    "pull", "push", "commit", "fetch", "clone", "switch", "checkout", "merge", "rebase", "reset", "restore",

    "stash", "add", "rm", "status", "init", "close", "am", "cherry-pick", "revert", "gc", "fsck", "prune",

    "doctor", "sparse", "conflict", "login", "finish", "desc")

RECOVERY_COMMAND_RE = re.compile(

    r"\b(?:oak|git)\s+(?:" + "|".join(re.escape(c) for c in RECOVERY_SUBCOMMANDS) + r")\b"

    r"|(?:^|\s)--(?:continue|abort|skip|force)\b", re.I)

RECOVERY_STEP_RE = re.compile(

    r"remove the file manually|remove this directory|move or remove|delete (?:it|this directory|the directory)"

    r"|commit your changes|stash them|\bpull (?:changes )?first\b|\((?:fetch first|non-fast-forward)\)", re.I)

RETRYABLE_RE = re.compile(r"\btry again\b|\bretry\b|\bre-?run\b|repository (?:is )?locked|lock(?:ed)? by (?:another|pid)",

                          re.I)

# Git's normal outcome messages for a conflicted merge / refused fast-forward. They

# name the next step (resolve and commit; merge or rebase instead of --ff-only). They

# are accepted only from the git subject and only when EVERY output line is one of

# git's ordinary merge/pull lines (an allowlist, not a denylist of corruption words):

# any other line - "fatal: bad object HEAD", "unable to read sha1 file", a segfault,

# "unable to open database file" - means something else went wrong, so the hint

# cannot launder it.

GIT_OUTCOME_HINT_RE = re.compile(

    r"^(?:fatal: )?Not possible to fast-forward, aborting\.$"

    r"|^Automatic merge failed; fix conflicts and then commit the result\.$", re.M)

GIT_OUTCOME_LINE_RE = re.compile(

    r"Automatic merge failed; fix conflicts and then commit the result\."

    r"|(?:fatal: )?Not possible to fast-forward, aborting\."

    r"|hint:.*"

    r"|Auto-merging [^\n]+"

    r"|CONFLICT \([a-z/ -]+\): [^\n]+"

    r"|From [^\s]+"

    r"|\s*\* branch\s+\S+\s+-> FETCH_HEAD"

    r"|\s*(?:\* \[new branch\]|[0-9a-f]{7,40}\.\.[0-9a-f]{7,40}|\+ [0-9a-f]{7,40}\.\.\.[0-9a-f]{7,40})"

    r"\s+\S+\s+-> \S+(?: \(forced update\))?"

    r"|Updating [0-9a-f]{7,40}\.\.[0-9a-f]{7,40}")

def git_outcome_hint(res: Result) -> bool:

    """git's conflict / refused-fast-forward outcome with nothing else in the output."""

    if not res.argv or Path(res.argv[0]).name != "git":

        return False

    text = res.text()

    if not GIT_OUTCOME_HINT_RE.search(text):

        return False

    return all(GIT_OUTCOME_LINE_RE.fullmatch(line.rstrip()) for line in text.splitlines() if line.strip())

# Corruption / internal-failure wording: never actionable on retry wording or a generic code alone.

NON_ACTIONABLE_RE = re.compile(

    r"panicked at|RUST_BACKTRACE|not a database|malformed|corrupt|internal (?:server )?error|\binternal\b"

    r"|integrity check failed|unexpected (?:eof|end)|not a git repository|HTTP 500\b", re.I)

GENERIC_ERROR_CODES = {"error", "internal", "internal_error", "unknown", "unknown_error", "io", "io_error", "other",

                       "panic", "database", "database_error", "sqlite", "unexpected", "server_error", "http_error"}

def _json_error(stdout: str) -> Optional[dict]:

    try:

        data = json.loads(stdout)

    except (ValueError, TypeError):

        return None

    if isinstance(data, dict) and isinstance(data.get("error"), dict):

        return data["error"]

    return None

def actionable(res: Result) -> bool:

    """True when a failed command tells the user a concrete next step."""

    text = res.text()

    if res.returncode is None or res.returncode <= 0:

        return False

    if "panicked at" in text or "RUST_BACKTRACE" in text:

        return False

    if RECOVERY_COMMAND_RE.search(text) or RECOVERY_STEP_RE.search(text):

        return True

    broken = bool(NON_ACTIONABLE_RE.search(text))

    if not broken and git_outcome_hint(res):

        return True

    error = _json_error(res.stdout)

    if error is not None and not broken:

        code = str(error.get("code") or "").strip().lower()

        if error.get("recommended_next_commands"):

            return True

        if code and code not in GENERIC_ERROR_CODES and not code.startswith(("internal", "unknown", "io_")):

            return True

    return not broken and bool(RETRYABLE_RE.search(text))

@dataclass

class Calibration:

    elapsed_ms: float

    post_tree: Optional[dict]

    post_worktree: Optional[dict]

    post_head: Optional[str]

    post_server_head: Optional[str]

    boundaries: int = 0

    boundary_kinds: list = field(default_factory=list)

    requests: list = field(default_factory=list)

    ok: bool = True

    detail: str = ""

class Trial:

    """One copied fixture + (for Oak) its loopback server, ready for one faulted op."""

    def __init__(self, fixture: Fixture, adapter_bin: str, work: Path, name: str):

        self.fixture = fixture

        self.dir = work / name

        copy_tree(fixture.pristine, self.dir)

        self.home = self.dir / "home"

        self.kind = fixture.kind

        if self.kind == "oak":

            self.adapter = OakAdapter(adapter_bin, self.home)

            self.server: Optional[Server] = Server(adapter_bin, self.dir / "srv", fixture.port or 0,

                                                   self.dir / "serve.log")

        else:

            self.adapter = GitAdapter(adapter_bin, self.home)

            self.server = None

        self.client = self.dir / "client"

    # op plumbing ---------------------------------------------------------------

    def op_cwd(self) -> Path:

        return self.dir if self.fixture.op == "clone" else self.client

    def op_argv(self, bin_override: Optional[str] = None, dest: str = "cloned", url: Optional[str] = None) -> list[str]:

        kw: dict[str, Any] = {"bin": bin_override}

        if self.fixture.op == "clone":

            kw.update(dest=dest, branch=self.fixture.branch, url=url or self.fixture.meta["url"])

        if self.fixture.op == "switch":

            kw["target"] = self.fixture.meta["target"]

        return self.adapter.argv(self.fixture.op, **kw)

    def pre(self) -> None:

        for argv in self.adapter.pre_argv(self.fixture.op):

            self.adapter.must(argv, self.client)

    def start_server(self, port: Optional[int] = None) -> None:

        if self.server is not None:

            self.server.start(port)

    def stop_server(self) -> None:

        if self.server is not None:

            self.server.stop()

    def settle(self) -> None:

        """Let orphaned server-side children (Git local receive-pack) finish."""

        if self.kind == "git":

            remote = self.dir / "srv" / "remote.git"

            deadline = time.monotonic() + 10

            while time.monotonic() < deadline and any(remote.rglob("*.lock")):

                time.sleep(0.05)

            time.sleep(0.1)

    def target_repo(self, dest: str = "cloned") -> Path:

        return self.dir / dest if self.fixture.op == "clone" else self.client

    def cleanup(self) -> None:

        self.stop_server()

        shutil.rmtree(self.dir, ignore_errors=True)

    # independent observations ---------------------------------------------------

    def server_head(self) -> Optional[str]:

        if self.kind == "oak":

            db = self.server.db() if self.server else None

            if not db or not db.exists():

                return None

            with OakGraph(db) as g:

                return g.branch_heads().get(self.fixture.branch)

        refs = self.adapter.oracle.refs(self.dir / "srv" / "remote.git")

        return refs.get(f"refs/heads/{self.fixture.branch}")

    def server_tree(self, commit: str) -> tuple[Optional[dict], str]:

        if self.kind == "oak":

            from .readers import oak_commit_tree

            return oak_commit_tree(self.server.db(), commit)

        return self.adapter.oracle.commit_tree(self.dir / "srv" / "remote.git", commit)

    def repo_opens(self, repo: Path) -> Optional[bool]:

        if self.kind == "oak":

            db = repo / ".oak" / "oak.db"

            if not db.exists():

                return False

            try:

                with OakGraph(db) as g:

                    return g.integrity() == "ok" and g.metadata("current_branch") is not None

            except Exception:  # noqa: BLE001 - any failure to open is the finding

                return False

        if not (repo / ".git").exists():

            return False

        return self.adapter.oracle.fsck(repo) == "ok"

    def acked_reachable(self, repo: Path, heads: list[str]) -> tuple[Optional[bool], list[str]]:

        if not heads:

            return None, []

        try:

            if self.kind == "oak":

                with OakGraph(repo / ".oak" / "oak.db") as g:

                    reach = g.reachable()

                lost = [h for h in heads if h not in reach]

            else:

                lost = [h for h in heads if not self.adapter.oracle.reachable_from_refs(repo, h)]

        except Exception as exc:  # noqa: BLE001 - reader failure is unmeasured, never a subject violation

            return None, [f"reader error: {type(exc).__name__}: {exc}"]

        return not lost, lost

    def server_intact(self, intended: Optional[str]) -> tuple[Optional[bool], dict]:

        pre = self.fixture.meta.get("pre_server_heads", {})

        detail: dict = {}

        try:

            if self.kind == "oak":

                with OakGraph(self.server.db()) as g:

                    detail["integrity"] = g.integrity()

                    heads = g.branch_heads()

                    reach = g.ancestors(heads.values())

                    missing = g.missing_parents([h for h in heads.values()])

            else:

                remote = self.dir / "srv" / "remote.git"

                detail["integrity"] = self.adapter.oracle.fsck(remote)

                refs = self.adapter.oracle.refs(remote)

                heads = {k.removeprefix("refs/heads/"): v for k, v in refs.items() if k.startswith("refs/heads/")}

                reach = set()

                for v in pre.values():

                    if any(self.adapter.oracle.is_ancestor(remote, v, h) for h in heads.values()):

                        reach.add(v)

                missing = []

        except Exception as exc:  # noqa: BLE001 - reader failure is unmeasured, never a subject violation

            return None, {"reader_error": f"{type(exc).__name__}: {exc}"}

        head = heads.get(self.fixture.branch)

        detail.update(server_head=head, lost_pre_heads=[v for v in pre.values() if v not in reach],

                      torn_graph=missing[:5])

        allowed = {pre.get(self.fixture.branch), intended}

        detail["head_in_allowed"] = head in allowed

        ok = detail["integrity"] == "ok" and not detail["lost_pre_heads"] and not missing and head in allowed

        return ok, detail

def worktree_verdict(trial: Trial, repo: Path, allowed: list[Optional[dict]]) -> tuple[Optional[bool], dict]:

    from .readers import worktree_manifest

    wt = worktree_manifest(repo)

    if wt is None:

        return None, {"worktree": "unreadable"}

    if any(a is not None and wt == a for a in allowed):

        return True, {"worktree": "pre_or_post"}

    head_tree, status = trial.adapter.tree(repo)

    if head_tree is not None and wt == head_tree:

        return True, {"worktree": "matches_head_tree"}

    dirty, res = trial.adapter.status_dirty(repo)

    if dirty is None:

        # status refused with an error: acceptable only if it is actionable

        ok = actionable(res)

        return ok, {"worktree": "differs; status errored", "status_actionable": ok, "status": res.brief(300),

                    "head_tree_reader": status}

    if dirty:

        return True, {"worktree": "differs; reported dirty", "head_tree_reader": status}

    if head_tree is None:

        return None, {"worktree": "differs; status clean; head tree unreadable", "head_tree_reader": status}

    return False, {"worktree": "SILENT_TORN: differs from HEAD tree while status reports clean"}

def torn_worktree(wt: Optional[dict], pre: Optional[dict], post: Optional[dict]) -> Optional[dict]:

    """Split a crashed worktree against the interrupted operation's two sides.

    Returns None when unmeasurable, else lists of paths the op changes that

    already hold the post-op (target) side (``op_output``) or still hold the

    pre-op side (``pre_side``), plus op artifacts in neither side (``stray``,

    e.g. temp files).

    """

    if wt is None or pre is None or post is None:

        return None

    changed_by_op = {p for p in set(pre) | set(post) if pre.get(p) != post.get(p)}

    op_output, pre_side, stray = [], [], []

    for path in sorted(set(wt) | changed_by_op):

        have = wt.get(path)

        if path in changed_by_op:

            if have == post.get(path):

                op_output.append(path)

            elif have == pre.get(path):

                pre_side.append(path)

        elif path not in pre and path not in post and have is not None:

            stray.append(path)

    return {"op_output": op_output, "pre_side": pre_side, "stray": stray}

def torn_state(split: dict, ref_side: Optional[str]) -> bool:

    """Is the worktree out of step with where the ref says the operation is?

    ref_side "pre": any of the op's output (or its temp files) is present.

    ref_side "post": any changed path still on the pre side (or temp files).

    unknown ref: the worktree mixes both sides, or has temp files.

    """

    if split["stray"]:

        return True

    if ref_side == "pre":

        return bool(split["op_output"])

    if ref_side == "post":

        return bool(split["pre_side"])

    return bool(split["op_output"]) and bool(split["pre_side"])

def torn_verdict(trial: "Trial", repo: Path, pre: Optional[dict], post: Optional[dict],

                 pre_head: Optional[str], op_res: Optional[Result] = None) -> tuple[Optional[bool], dict]:

    """Invariant ``worktree_not_torn``.

    Fails when, after the crash, the worktree holds the interrupted operation's

    own output out of step with the ref (e.g. files already on the target side

    while HEAD still names the old side, or the op's temp files) and the

    subject presents it as ordinary uncommitted user changes: status lists

    changes, reports no operation in progress, and does not refuse. Committing

    that state would record the op's partial output as the user's own work.

    Passing: worktree in step with the ref, an in-progress flag (MERGE_HEAD /

    progress_state / a stale git lock - index.lock, HEAD.lock or the current

    branch's ref lock - that makes the next commit refuse), a status

    refusal (judged by ``worktree_consistent``), or the failing op itself naming

    a recovery command (e.g. git's "checkout failed ... retry with git restore").

    A killed op prints nothing, so crash rows rely on the persistent flags only.

    """

    from .readers import worktree_manifest

    wt = worktree_manifest(repo)

    split = torn_worktree(wt, pre, post)

    if split is None:

        return None, {"torn": "unmeasurable"}

    head = trial.adapter.head(repo)

    # The op either leaves HEAD where it was or moves it to its result (a merge /

    # re-parent commit's hash differs per run, so "moved" is the post side).

    ref_side = None if head is None else ("pre" if head == pre_head else "post")

    detail = {"ref_side": ref_side, "op_output_files": len(split["op_output"]), "stray_files": len(split["stray"]),

              "pre_side_files": len(split["pre_side"]), "sample": (split["stray"] + split["op_output"])[:5]}

    if not torn_state(split, ref_side):

        return True, {**detail, "torn": "no"}

    if op_res is not None and (op_res.returncode or 0) > 0 and RECOVERY_COMMAND_RE.search(op_res.text()):

        return True, {**detail, "torn": "failing op named a recovery command"}

    changes, in_progress, res = trial.adapter.status_state(repo)

    detail["status_rc"] = res.returncode

    if changes is None:

        return True, {**detail, "torn": "status refused (judged by worktree_consistent)"}

    if in_progress:

        return True, {**detail, "torn": "flagged in progress"}

    if changes:

        return False, {**detail, "torn": "PARTIAL_OP_OUTPUT_AS_USER_CHANGES"}

    return True, {**detail, "torn": "status clean (judged by worktree_consistent)"}

def classify(invariants: dict[str, Optional[bool]], harness_error: Optional[str] = None,

             fault_fired: Optional[bool] = None, reader_errors: Optional[list] = None) -> str:

    """held / violated / harness_error / fault_not_fired.

    ``fault_fired=False`` (the fault demonstrably never took effect) is never

    ``held``: the row is a clean run. ``None`` means the family has no single

    fault to fire (concurrency). Reader exceptions are harness errors unless a

    subject-level invariant already failed independently of the reader.

    """

    if harness_error:

        return HARNESS_ERROR

    if fault_fired is False:

        return FAULT_NOT_FIRED

    if any(v is False for v in invariants.values()):

        return VIOLATED

    if reader_errors:

        return HARNESS_ERROR

    return HELD

def converge_check(trial: Trial, calib: Calibration, repo: Path) -> tuple[Optional[bool], dict]:

    """After a successful next op, is the state exactly the calibrated outcome?"""

    from .readers import worktree_manifest

    op = trial.fixture.op

    detail: dict = {}

    if op == "push":

        local = trial.adapter.head(repo)

        remote = trial.server_head()

        detail.update(local_head=local, server_head=remote)

        if remote is None or local is None or remote != local:

            return False, detail

        tree, status = trial.server_tree(remote)

        detail["server_tree_reader"] = status

        if tree is None:

            return None, detail

        return tree == calib.post_tree, detail

    if op == "switch":

        branch = trial.adapter.current_branch(repo)

        detail["branch"] = branch

        return branch == trial.fixture.meta["target"] and worktree_manifest(repo) == calib.post_worktree, detail

    tree, status = trial.adapter.tree(repo)

    detail["head_tree_reader"] = status

    if tree is None:

        return (False if status.startswith("corrupt") else None), detail

    same_tree = tree == calib.post_tree

    same_wt = worktree_manifest(repo) == calib.post_worktree

    detail.update(head_tree_matches=same_tree, worktree_matches=same_wt)

    if op == "clone":

        detail["head"] = trial.adapter.head(repo)

    return same_tree and same_wt, detail