"""Concurrency family: N clients, shared checkouts, stale lock owners.

Scenarios (per subject):

- same_branch   N clones push to ONE branch (commit, push; on rejection pull then

                retry). The push history is checked for linearizability against

                the fast-forward register model and for lost acknowledged writes;

                every acknowledged file must be in the final server tree.

- distinct      N clones each own a branch; every acknowledged head must be the

                server's head for that branch.

- shared_commit N processes commit concurrently in ONE checkout (workdir lock):

                each exits 0 or with an actionable (lock) error, nothing hangs,

                the store opens, and a final serial commit captures every file.

- shared_push   N processes push concurrently from ONE checkout (publication lock).

- stale_owner   a commit is SIGKILLed at its first durability boundary while

                holding the workdir lock; N contenders must not hang (stale-owner

                reaping) and at least one commit must land.

"""

from __future__ import annotations

import json

import math

import shutil

import threading

import time

from pathlib import Path

from typing import Optional

from . import FAULT_NOT_FIRED, HARNESS_ERROR

from .adapters import REPO_ID, GitAdapter, OakAdapter, Server, payload

from .engine import actionable, classify

from .lincheck import OK, REJECTED, UNKNOWN, Op, ancestry_from_parents, check_partitioned, lost_acknowledged

from .procs import free_port, run

from .readers import OakGraph, oak_commit_tree

PUSH_ATTEMPTS = 10

OP_TIMEOUT = 120.0

def _oak_parents(db: Path) -> dict[str, tuple]:

    with OakGraph(db) as g:

        return {h: (p, m) for h, p, m in g.conn.execute("SELECT hash, parent_hash, merge_parent_hash FROM commits")}

def _git_parents(adapter: GitAdapter, repo: Path) -> dict[str, tuple]:

    res = run([adapter.bin, "rev-list", "--all", "--parents"], repo, adapter.env(), timeout=60)

    parents = {}

    for line in res.stdout.splitlines():

        parts = line.split()

        if parts:

            parents[parts[0]] = tuple(parts[1:])

    return parents

class World:

    """One subject's remote + N worker checkouts."""

    def __init__(self, kind: str, binary: str, root: Path, n: int):

        self.kind, self.bin, self.root, self.n = kind, binary, root, n

        if root.exists():

            shutil.rmtree(root)

        root.mkdir(parents=True)

        self.server: Optional[Server] = None

        self.seed = root / "seed"

        self.seed.mkdir()

        if kind == "oak":

            self.adapter = OakAdapter(binary, root / "home")

            self.server = Server(binary, root / "srv", free_port(), root / "serve.log").start()

            self.adapter.must([binary, "init", "."], self.seed)

            payload(self.seed, "base", count=20)

            self.adapter.must([binary, "commit"], self.seed)

            self.adapter.must([binary, "push", "--repo", REPO_ID, "-r", self.server.url], self.seed)

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

                self.branch = g.metadata("current_branch")

        else:

            self.adapter = GitAdapter(binary, root / "home")

            (root / "srv").mkdir()

            g = binary

            self.adapter.must([g, "init", "-q", "--bare", "-b", "main", "remote.git"], root / "srv")

            self.adapter.must([g, "init", "-q", "-b", "main", "."], self.seed)

            payload(self.seed, "base", count=20)

            self.adapter.must([g, "add", "-A"], self.seed)

            self.adapter.must([g, "commit", "-q", "-m", "base"], self.seed)

            self.adapter.must([g, "remote", "add", "origin", "../srv/remote.git"], self.seed)

            self.adapter.must([g, "push", "-q", "-u", "origin", "main"], self.seed)

            self.branch = "main"

        self.workers = []

        for i in range(n):

            dest = root / f"w{i:02d}"

            shutil.copytree(self.seed, dest, symlinks=True)  # "second machine" copy (jj bad-locking pattern)

            self.workers.append(dest)

    def stop(self) -> None:

        if self.server:

            self.server.stop()

    def head(self, repo: Path) -> Optional[str]:

        return self.adapter.head(repo)

    def server_heads(self) -> dict[str, str]:

        if self.kind == "oak":

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

                return g.branch_heads()

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

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

    def parents(self) -> dict[str, tuple]:

        merged: dict[str, tuple] = {}

        if self.kind == "oak":

            for db in [self.server.db()] + [w / ".oak" / "oak.db" for w in self.workers]:

                if db.exists():

                    merged.update(_oak_parents(db))

        else:

            for repo in [self.root / "srv" / "remote.git"] + self.workers:

                merged.update(_git_parents(self.adapter, repo))

        return merged

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

        if self.kind == "oak":

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

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

    def commit_cmds(self) -> list[list[str]]:

        if self.kind == "oak":

            return [[self.bin, "commit"]]

        return [[self.bin, "add", "-A"], [self.bin, "commit", "-q", "--no-verify", "-m", "w"]]

    def push_cmd(self, branch: Optional[str] = None) -> list[str]:

        if self.kind == "oak":

            return [self.bin, "push", "--json"]

        return [self.bin, "push", "-q", "origin", f"HEAD:refs/heads/{branch or self.branch}"]

    def pull_cmd(self) -> list[str]:

        if self.kind == "oak":

            return [self.bin, "pull"]

        return [self.bin, "pull", "-q", "--no-rebase", "--no-edit", "origin", self.branch]

def reported_pushed_head(res) -> Optional[str]:

    """The head the subject says it published (Oak ``push --json`` pushed_head); None if not reported."""

    try:

        data = json.loads(res.stdout)

    except (ValueError, TypeError):

        return None

    head = data.get("pushed_head") if isinstance(data, dict) else None

    return head if isinstance(head, str) and head else None

def _outcome(res) -> str:

    if res.returncode == 0:

        return OK

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

        return UNKNOWN

    text = res.text().lower()

    if "unknown" in text or "not received" in text or "inspect remote state" in text:

        return UNKNOWN

    return REJECTED

def same_branch(kind: str, binary: str, root: Path, n: int, rounds: int = 2, distinct: bool = False) -> dict:

    world = World(kind, binary, root, n)

    t0 = time.monotonic()

    ops: list[Op] = []

    acked_files: list[str] = []

    errors: list[str] = []

    unactionable: list[dict] = []

    lock = threading.Lock()

    branches: dict[int, str] = {}

    pulls: list = []

    head_mismatches: list[dict] = []

    def worker(i: int) -> None:

        repo = world.workers[i]

        env = world.adapter.env()

        branch = world.branch

        try:

            if distinct:

                branch = f"p4w{i:02d}"

                if kind == "oak":

                    world.adapter.must([binary, "switch", "-c", branch], repo)

                else:

                    world.adapter.must([binary, "switch", "-q", "-c", branch], repo)

            branches[i] = branch

            for r in range(rounds):

                name = f"conc/w{i:02d}-r{r}.txt"

                (repo / "conc").mkdir(exist_ok=True)

                (repo / name).write_text(f"worker {i} round {r}\n")

                for argv in world.commit_cmds():

                    res = run(argv, repo, env, timeout=OP_TIMEOUT)

                    if res.returncode != 0:

                        raise RuntimeError(f"commit failed: {res.stderr[-200:]}")

                for attempt in range(PUSH_ATTEMPTS):

                    value = world.head(repo)

                    call = time.monotonic() - t0

                    res = run(world.push_cmd(branch), repo, env, timeout=OP_TIMEOUT)

                    ret = time.monotonic() - t0

                    outcome = _outcome(res)

                    reported = reported_pushed_head(res) if outcome == OK else None

                    with lock:

                        if reported is not None and reported != value:

                            head_mismatches.append({"worker": i, "local_head": value, "pushed_head": reported})

                        ops.append(Op(f"w{i}", "push", value, call, ret if outcome != UNKNOWN else math.inf,

                                      outcome, branch, meta={"rc": res.returncode, "attempt": attempt,

                                                             "pushed_head_reported": reported is not None}))

                        if outcome == REJECTED and not actionable(res):

                            unactionable.append(res.brief(200))

                    if outcome == OK:

                        with lock:

                            acked_files.append(name)

                        break

                    pull = run(world.pull_cmd(), repo, env, timeout=OP_TIMEOUT)

                    with lock:

                        pulls.append(pull.returncode)

                    if pull.returncode != 0 and kind == "git":

                        # a conflicting merge cannot happen (disjoint files); record and stop

                        raise RuntimeError(f"pull failed: {pull.stderr[-200:]}")

        except Exception as exc:  # noqa: BLE001

            with lock:

                errors.append(f"w{i}: {exc}")

    threads = [threading.Thread(target=worker, args=(i,)) for i in range(n)]

    for thread in threads:

        thread.start()

    for thread in threads:

        thread.join(timeout=OP_TIMEOUT * PUSH_ATTEMPTS * rounds)

    hung = any(thread.is_alive() for thread in threads)

    try:

        final = world.server_heads()

        parents = world.parents()

        is_anc = ancestry_from_parents(parents)

        # the final server observation is a read at +end for each branch

        end = time.monotonic() - t0 + 1

        ops_with_reads = ops + [Op("final", "read", final.get(b), end, end, OK, b) for b in set(o.branch for o in ops)]

        results = check_partitioned(ops_with_reads, is_anc)

        lost = lost_acknowledged(ops, final, is_anc)

        missing_files: list[str] = []

        tree_status = "skipped"

        if not distinct and final.get(world.branch):

            tree, tree_status = world.server_tree(final[world.branch])

            if tree is not None:

                missing_files = [f for f in acked_files if f not in tree]

        integrity = "ok"

        if kind == "oak":

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

                integrity = g.integrity()

        else:

            integrity = world.adapter.oracle.fsck(world.root / "srv" / "remote.git")

        lin = [r.linearizable for r in results.values()]

        inv = {

            "no_hang": not hung,

            "no_lost_acknowledged_write": not lost,

            "linearizable": (None if any(v is None for v in lin) else all(lin)) if lin else None,

            "acked_content_on_server": (not missing_files) if tree_status == "ok" else None,

            "server_opens": integrity == "ok",

            "rejections_actionable": not unactionable,

            # an acknowledged push must report the commit the client had (Oak push --json pushed_head)

            "acked_head_matches_reported": (not head_mismatches) if any(o.meta.get("pushed_head_reported")

                                                                        for o in ops) else None,

        }

        obs = {

            "pushes": len(ops),

            "acknowledged": sum(1 for o in ops if o.outcome == OK),

            "rejected": sum(1 for o in ops if o.outcome == REJECTED),

            "unknown": sum(1 for o in ops if o.outcome == UNKNOWN),

            "pull_returncodes": {str(rc): pulls.count(rc) for rc in set(pulls)},

            "lost": [o.value for o in lost][:5],

            "missing_files": missing_files[:5],

            "lincheck": {b: {"linearizable": r.linearizable, "explored": r.explored, "reason": r.reason}

                         for b, r in results.items()},

            "history": [[o.client, o.outcome, (o.value or "")[:12], round(o.call, 3),

                         None if o.ret == math.inf else round(o.ret, 3), o.meta.get("rc")]

                        for o in sorted(ops, key=lambda o: o.call)][:400],

            "final_heads": {b: (h or "")[:12] for b, h in final.items()},

            "worker_errors": errors[:5],

            "unactionable_rejections": unactionable[:3],

            "pushed_head_mismatches": head_mismatches[:5],

            "tree_reader": tree_status,

        }

        classification = classify(inv, harness_error="; ".join(errors[:3]) if errors else None)

    finally:

        world.stop()

    return {"invariants": inv, "observations": obs, "classification": classification,

            "elapsed_ms": round((time.monotonic() - t0) * 1000.0, 3),

            **({"harness_error": "; ".join(errors[:3])} if errors else {})}

def shared_checkout(kind: str, binary: str, root: Path, n: int, what: str) -> dict:

    """what: commit | push | stale_owner"""

    world = World(kind, binary, root, 1)

    repo = world.workers[0]

    env = world.adapter.env()

    t0 = time.monotonic()

    results = []

    lock = threading.Lock()

    obs: dict = {}

    try:

        stale = None

        stale_fired: Optional[bool] = None

        if what == "stale_owner":

            if kind != "oak":

                raise NotImplementedError("stale_owner is Oak-only (git leaves index.lock by design)")

            from . import crashpoints

            inst, why = crashpoints.build(root / "inst", binary, None)

            why = crashpoints.unavailable_reason(inst, why, "oak")

            if why:

                raise NotImplementedError(why)

            (repo / "stale.txt").write_text("stale owner\n")

            stale = run([str(inst.oak_bin), "commit"], repo, world.adapter.env(inst.env(1)), timeout=30)

            obs["stale_owner_rc"] = stale.returncode

            stale_fired = stale.returncode == -9

        if what == "push":

            (repo / "pub.txt").write_text("publication\n")

            for argv in world.commit_cmds():

                world.adapter.must(argv, repo)

        for i in range(n):

            (repo / f"shared-{i:02d}.txt").write_text(f"shared {i}\n")

        def contender(i: int) -> None:

            if what == "push":

                cmds = [world.push_cmd()]

            else:

                cmds = world.commit_cmds()

            for argv in cmds:

                res = run(argv, repo, env, timeout=OP_TIMEOUT)

                with lock:

                    results.append(res)

                if res.returncode != 0:

                    break

        threads = [threading.Thread(target=contender, args=(i,)) for i in range(n)]

        for thread in threads:

            thread.start()

        for thread in threads:

            thread.join(timeout=OP_TIMEOUT * 3)

        hung = any(thread.is_alive() for thread in threads) or any(r.timed_out for r in results)

        failures = [r for r in results if r.returncode != 0]

        final_ok = True

        missing: list[str] = []

        if what in ("commit", "stale_owner"):

            for argv in world.commit_cmds():

                res = run(argv, repo, env, timeout=OP_TIMEOUT)

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

            obs["tree_reader"] = status

            if tree is not None:

                missing = [f"shared-{i:02d}.txt" for i in range(n) if f"shared-{i:02d}.txt" not in tree]

            final_ok = None if tree is None else not missing

        else:

            final_ok = world.server_heads().get(world.branch) == world.head(repo)

        if kind == "oak":

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

                opens = g.integrity() == "ok"

        else:

            opens = world.adapter.oracle.fsck(repo) == "ok"

        successes = sum(1 for r in results if r.returncode == 0)

        inv = {

            "no_hang": not hung,

            "failures_actionable": all(actionable(r) for r in failures),

            "repo_opens": opens,

            "final_state_complete": final_ok,

        }

        if what == "stale_owner":

            inv["stale_owner_reaped"] = successes > 0

        obs.update(contenders=n, successes=successes, failures=len(failures), missing=missing[:5],

                   failure_samples=[r.brief(200) for r in failures if not actionable(r)][:2],

                   max_contender_ms=round(max((r.elapsed_ms for r in results), default=0.0), 1))

        classification = classify(inv, fault_fired=stale_fired)

        error = None

    except NotImplementedError as exc:

        return {"unsupported": str(exc)}

    except Exception as exc:  # noqa: BLE001

        inv, classification, error = {}, HARNESS_ERROR, f"{type(exc).__name__}: {exc}"

    finally:

        world.stop()

    row = {"invariants": inv, "observations": obs, "classification": classification,

           "elapsed_ms": round((time.monotonic() - t0) * 1000.0, 3)}

    if what == "stale_owner":

        row["fault_fired"] = stale_fired

        if classification == FAULT_NOT_FIRED:

            row["skip_reason"] = (f"lock owner was not killed at its first durability boundary "

                                  f"(rc={obs.get('stale_owner_rc')})")

    if error:

        row["harness_error"] = error

    return row