"""Drive Oak's client merge engine on a scenario and report what it produced.

Exercised Oak path (phase 1, "pull-sync"):

    oak clone <git.git>   (git-history import: theirs lineage, then the fork)

    oak push              (theirs clone bootstraps `main` on a local `oak serve`)

    oak commit            (ours: one checkpoint per scripted tree state)

    oak push; oak fetch; oak merge --dry-run --json   (local merge prediction)

    oak pull --json       (parent sync: the CLI merges `main` INTO the branch)

`oak pull`'s parent sync is the Oak client's 3-way merge engine (merge base

resolution over the local+hydrated graph, `three_way_merge_manifests`, diffy

text merge, conflict markers). It is the same engine `oak merge` uses for

non-main parents. The hosted squash-merge onto `main` (oakspace) is NOT

exercised; see HostedMergeAdapter for the phase-2 plug point.

`oak serve` 0.105 does not implement `GET /api/:o/:r/raw/:commit/:path`, which

the client uses to hydrate parent blobs. RawShim fronts the real server and

answers only that route, from the publishing (theirs) clone via

`oak file inspect --at <commit> --output` (hash-verified by Oak itself); every

other request is forwarded byte-for-byte. The shim can also pin the `head`

the server reports for `main` to an earlier main commit ("staging"), which

lets multi-step scenarios move `main` forward between client pulls without a

hosted merge endpoint. No merge logic lives in the shim.

"""

from __future__ import annotations

import http.server

import json

import os

from pathlib import Path

import shutil

import socket

import sqlite3

import subprocess

import tempfile

import threading

import time

import urllib.error

import urllib.parse

import urllib.request

from typing import Dict, List, Optional

from .gitops import Git, worktree_manifest

PATH_EXERCISED = "oak-pull-parent-sync"

PATH_DESCRIPTION = ("oak pull parent-sync (client merge engine: merge base over local+hydrated "

                    "graph, three_way_merge_manifests, diffy text merge) against a local oak serve; "

                    "prediction via oak merge --dry-run --json")

class HarnessError(Exception):

    """The instrument (not Oak) failed; the case is classified harness-error."""

class StepFailed(Exception):

    """An Oak command in scenario setup failed; carries the step record."""

    def __init__(self, message: str, step: dict):

        super().__init__(message)

        self.step = step

def _free_port() -> int:

    with socket.socket() as s:

        s.bind(("127.0.0.1", 0))

        return s.getsockname()[1]

class RawShim:

    """Forwarding proxy that adds the missing /raw route and optional main-head pinning."""

    def __init__(self, upstream: str, oak_bin: str, env: dict):

        self.upstream, self.oak_bin, self.env = upstream, oak_bin, env

        self.repos: Dict[str, dict] = {}  # "owner/name" -> {"source": Path, "stage_head": str|None}

        self.lock = threading.Lock()

        self.log: List[str] = []

        shim = self

        class Handler(http.server.BaseHTTPRequestHandler):

            protocol_version = "HTTP/1.1"

            def log_message(self, *args):

                pass

            def _send(self, code, data, headers=()):

                self.send_response(code)

                for k, v in headers:

                    if k.lower() not in ("transfer-encoding", "content-length", "connection"):

                        self.send_header(k, v)

                self.send_header("Content-Length", str(len(data)))

                self.end_headers()

                self.wfile.write(data)

            def _handle(self):

                parsed = urllib.parse.urlsplit(self.path)

                parts = parsed.path.split("/")

                length = int(self.headers.get("Content-Length") or 0)

                body = self.rfile.read(length) if length else None

                if self.command == "GET" and len(parts) > 6 and parts[1] == "api" and parts[4] == "raw":

                    key = f"{parts[2]}/{parts[3]}"

                    code, data = shim.raw(key, parts[5], urllib.parse.unquote("/".join(parts[6:])))

                    return self._send(code, data)

                req = urllib.request.Request(

                    shim.upstream + self.path, data=body, method=self.command,

                    headers={k: v for k, v in self.headers.items()

                             if k.lower() not in ("host", "content-length", "connection")})

                try:

                    with urllib.request.urlopen(req, timeout=120) as r:

                        code, data, hdrs = r.status, r.read(), list(r.headers.items())

                except urllib.error.HTTPError as e:

                    code, data, hdrs = e.code, e.read(), list(e.headers.items())

                if self.command == "GET" and code == 200 and len(parts) in (4, 6) and parts[1] == "api":

                    data = shim.pin_head(f"{parts[2]}/{parts[3]}", parts, data)

                return self._send(code, data, hdrs)

            do_GET = do_POST = do_PUT = do_DELETE = do_PATCH = _handle

        class Server(http.server.ThreadingHTTPServer):

            def handle_error(self, request, client_address):

                shim.log.append("shim connection error (client disconnected)")

        self.server = Server(("127.0.0.1", 0), Handler)

        self.server.daemon_threads = True

        self.url = f"http://127.0.0.1:{self.server.server_address[1]}"

        self.thread = threading.Thread(target=self.server.serve_forever, daemon=True)

        self.thread.start()

    def register(self, key: str, source: Path) -> None:

        with self.lock:

            self.repos[key] = {"source": Path(source), "stage_head": None}

    def stage(self, key: str, head: Optional[str]) -> None:

        with self.lock:

            self.repos[key]["stage_head"] = head

    def pin_head(self, key: str, parts: List[str], data: bytes) -> bytes:

        with self.lock:

            entry = self.repos.get(key)

            head = entry and entry["stage_head"]

        if not head:

            return data

        is_repo = len(parts) == 4

        is_main = len(parts) == 6 and parts[4] == "branches" and parts[5] == "main"

        if not (is_repo or is_main):

            return data

        try:

            doc = json.loads(data)

        except ValueError:

            return data

        for field in ("head", "head_hash"):

            if isinstance(doc, dict) and field in doc:

                doc[field] = head

        return json.dumps(doc).encode()

    def raw(self, key: str, commit: str, path: str):

        with self.lock:

            entry = self.repos.get(key)

        if entry is None:

            self.log.append(f"raw for unregistered repo {key}")

            return 404, b""

        with tempfile.TemporaryDirectory(prefix="rawshim-") as tmp:

            out = Path(tmp) / "blob"

            p = subprocess.run([self.oak_bin, "file", "inspect", "--at", commit, "--json", path,

                                "--output", str(out)], cwd=entry["source"], env=self.env,

                               capture_output=True, timeout=60)

            if p.returncode or not out.exists():

                self.log.append(f"raw miss {key} {commit[:12]} {path!r} rc={p.returncode} "

                                f"{p.stderr.decode('utf-8', 'replace')[:200]}")

                return 404, b""

            return 200, out.read_bytes()

    def close(self) -> None:

        self.server.shutdown()

        self.server.server_close()

class LocalOakServer:

    """`oak serve` on a private data dir + RawShim in front of it."""

    def __init__(self, workdir: Path, oak_bin: str, env: dict):

        self.data = workdir / "oak-serve-data"

        self.data.mkdir(parents=True, exist_ok=True)

        self.port = _free_port()

        self.log_path = workdir / "oak-serve.log"

        self._log = open(self.log_path, "wb")

        self.proc = subprocess.Popen([oak_bin, "serve", "-d", str(self.data), "-p", str(self.port)],

                                     stdout=self._log, stderr=subprocess.STDOUT, env=env,

                                     stdin=subprocess.DEVNULL)

        upstream = f"http://127.0.0.1:{self.port}"

        deadline = time.monotonic() + 20

        while True:

            try:

                urllib.request.urlopen(upstream + "/api/capabilities", timeout=2).read()

                break

            except (urllib.error.URLError, ConnectionError, OSError):

                if self.proc.poll() is not None or time.monotonic() > deadline:

                    raise HarnessError(f"oak serve did not start (see {self.log_path})")

                time.sleep(0.1)

        self.shim = RawShim(upstream, oak_bin, env)

        self.url = self.shim.url

    def close(self) -> None:

        self.shim.close()

        if self.proc.poll() is None:

            self.proc.terminate()  # our own child only; never pattern-kill

            try:

                self.proc.wait(timeout=10)

            except subprocess.TimeoutExpired:

                self.proc.kill()

                self.proc.wait(timeout=10)

        self._log.close()

def oak_env(home: Path, git_bin: str) -> dict:

    env = {k: v for k, v in os.environ.items()

           if not k.startswith(("OAK_", "GIT_")) and k not in ("HOME",)}

    git_dir = str(Path(git_bin).parent)

    env.update({

        "HOME": str(home), "OAK_NO_UPDATE_CHECK": "1", "NO_COLOR": "1", "CI": "1",

        "OAK_PROGRESS": "never", "OAK_AUTHOR": "merge-suite",

        # `oak clone <x.git>` shells out to `git clone`; make it the resolved real git.

        "PATH": git_dir + os.pathsep + env.get("PATH", ""),

        "GIT_CONFIG_NOSYSTEM": "1", "GIT_CONFIG_GLOBAL": str(home / "gitconfig"),

    })

    return env

class OakRun:

    """One directional scenario execution against Oak."""

    def __init__(self, server: LocalOakServer, oak_bin: str, env: dict, case_dir: Path, repo_key: str):

        self.server, self.oak_bin, self.env = server, oak_bin, env

        self.case_dir, self.repo_key = case_dir, repo_key

        self.steps: List[dict] = []

    def oak(self, cwd: Path, *args: str, ok=(0,), label: str = "") -> dict:

        start = time.monotonic()

        p = subprocess.run([self.oak_bin, *args], cwd=cwd, env=self.env, capture_output=True,

                           stdin=subprocess.DEVNULL, timeout=180)

        step = {"label": label or args[0], "argv": ["oak", *args], "cwd": cwd.name,

                "returncode": p.returncode, "ms": round((time.monotonic() - start) * 1000),

                "stdout": p.stdout.decode("utf-8", "replace")[-4000:],

                "stderr": p.stderr.decode("utf-8", "replace")[-2000:]}

        self.steps.append(step)

        if ok is not None and p.returncode not in ok:

            raise StepFailed(f"oak {' '.join(args)} exited {p.returncode}", step)

        return step

def _json_tail(text: str) -> Optional[dict]:

    """Last JSON object printed on stdout (receipts are single-line JSON)."""

    for line in reversed(text.strip().splitlines()):

        line = line.strip()

        if line.startswith("{"):

            try:

                return json.loads(line)

            except ValueError:

                continue

    return None

def oak_commit_map(repo: Path) -> Dict[str, str]:

    """Read-only map commit message -> oak hash from a clone's local store.

    Plumbing only (locates staged main heads by their unique scenario

    messages); it is never used to judge merge results."""

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

    with tempfile.TemporaryDirectory(prefix="oakmap-") as tmp:

        # Copy (db + WAL) so we never take locks or write -shm files in Oak's store.

        for suffix in ("", "-wal"):

            src = Path(str(db) + suffix)

            if src.exists():

                shutil.copyfile(src, Path(tmp) / ("oak.db" + suffix))

        con = sqlite3.connect(str(Path(tmp) / "oak.db"))

        try:

            rows = con.execute("SELECT hash, message FROM commits").fetchall()

        finally:

            con.close()

    out: Dict[str, str] = {}

    for h, msg in rows:

        if msg:

            out.setdefault(msg.strip(), h)

    return out

def _clear_worktree(root: Path) -> None:

    for child in root.iterdir():

        if child.name == ".oak":

            continue

        if child.is_dir() and not child.is_symlink():

            shutil.rmtree(child)

        else:

            child.unlink()

def run_pull_sync(server: LocalOakServer, oak_bin: str, env: dict, git: Git, bare: Path,

                  case_dir: Path, repo_key: str, plan: dict) -> dict:

    """Execute a directional plan and return Oak's observed result.

    plan = {

      "main": git commit that becomes server `main` (theirs tip / final main),

      "fork": git commit the ours clone is imported at (must be on main's history),

      "steps": [{"op": "commit", "tree": <git commit whose tree to check in>}

                | {"op": "pull", "stage": <git commit on main>, "resolve": {path: bytes|None}}],

      "final_stage": optional git commit to pin as main for the measured pull,

    }

    """

    run = OakRun(server, oak_bin, env, case_dir, repo_key)

    result = {"path_exercised": PATH_EXERCISED, "steps": run.steps, "error": None,

              "error_kind": None}

    theirs_dir, ours_dir = case_dir / "theirs", case_dir / "ours"

    owner, name = repo_key.split("/")

    try:

        # 1. theirs clone imports main's lineage and bootstraps server main.

        _point_head(git, bare, plan["main"])

        run.oak(case_dir, "clone", str(bare), "theirs", label="import-main")

        run.oak(theirs_dir, "push", "-r", server.url, "--repo", repo_key, label="bootstrap-main")

        server.shim.register(repo_key, theirs_dir)

        messages = oak_commit_map(theirs_dir)

        commit_msgs = _git_messages(git, bare)

        def oak_hash(git_commit: str) -> str:

            msg = commit_msgs.get(git_commit)

            if msg is None or msg not in messages:

                raise HarnessError(f"cannot map git commit {git_commit[:12]} to an oak hash")

            return messages[msg]

        # 2. ours clone imports the fork point; scripted steps follow.

        _point_head(git, bare, plan["fork"])

        run.oak(case_dir, "clone", str(bare), "ours", label="import-fork")

        shutil.rmtree(ours_dir / ".git", ignore_errors=True)

        linked = False

        for step in plan["steps"]:

            if step["op"] == "commit":

                _clear_worktree(ours_dir)

                git.materialize(step["tree"], ours_dir)

                st = run.oak(ours_dir, "status", "--json", label="status")

                if (_json_tail(st["stdout"]) or {}).get("changes"):

                    run.oak(ours_dir, "commit", "--json", label="commit")

            elif step["op"] == "pull":

                server.shim.stage(repo_key, oak_hash(step["stage"]))

                run.oak(ours_dir, "push", "-r", server.url, "--repo", repo_key, label="push")

                linked = True

                pull = run.oak(ours_dir, "pull", "--json", ok=(0, 5), label="pull-step")

                step_receipt = _json_tail(pull["stdout"]) or {}

                if pull["returncode"] == 5 and step_receipt.get("parent_sync") != "conflict":

                    raise StepFailed("setup pull failed: " + str(step_receipt.get("parent_sync_error")), pull)

                if pull["returncode"] == 5:

                    if step.get("resolve") is None:

                        raise StepFailed("setup pull conflicted without a scripted resolution", pull)

                    for path, content in step["resolve"].items():

                        target = ours_dir / path

                        if content is None:

                            target.unlink(missing_ok=True)

                        else:

                            target.parent.mkdir(parents=True, exist_ok=True)

                            target.write_bytes(content)

                    run.oak(ours_dir, "pull", "--continue", label="pull-continue")

            else:

                raise HarnessError("unknown step op " + step["op"])

        # 3. measured merge: pin final main, predict, then pull.

        final = plan.get("final_stage")

        server.shim.stage(repo_key, oak_hash(final) if final else None)

        if not linked or plan["steps"]:

            run.oak(ours_dir, "push", "-r", server.url, "--repo", repo_key, label="push")

        run.oak(ours_dir, "fetch", ok=None, label="fetch")

        pred = run.oak(ours_dir, "merge", "--dry-run", "--json", ok=None, label="predict")

        result["prediction"] = summarize_prediction(_json_tail(pred["stdout"]), pred["returncode"])

        pre = _json_tail(run.oak(ours_dir, "status", "--json", label="pre-pull-status")["stdout"]) or {}

        pull = run.oak(ours_dir, "pull", "--json", ok=None, label="measured-pull")

        receipt = _json_tail(pull["stdout"]) or {}

        result["pull_returncode"] = pull["returncode"]

        result["pull_receipt"] = {k: receipt.get(k) for k in (

            "parent_sync", "convergence", "parent_head_before", "parent_head_after",

            "head_before", "head_after", "parent_sync_error", "parent_sync_error_code")}

        result["tree"] = worktree_manifest(ours_dir)

        result["result_source"] = "worktree-after-pull"

        if pull["returncode"] == 5 and receipt.get("parent_sync") == "conflict":

            result["clean"] = False

            show = run.oak(ours_dir, "conflict", "show", "--json", ok=None, label="conflict-show")

            result["conflicts"] = parse_conflict_show(_json_tail(show["stdout"]), receipt)

            # The conflicted worktree is a working view, not the merge result: Oak writes the

            # merged manifest's entries but leaves paths the merge deletes in place until

            # `--continue`. Measure the committed result: overwrite each conflicted path with a

            # sentinel resolution, continue, and read the (then clean) worktree.

            result["conflict_worktree"] = result["tree"]

            for path in result["conflicts"]:

                target = ours_dir / path

                if target.is_file() and not target.is_symlink():

                    target.write_bytes(b"mergesuite sentinel resolution\n")

            cont = run.oak(ours_dir, "pull", "--continue", ok=None, label="continue-sentinel")

            st = _json_tail(run.oak(ours_dir, "status", "--json", label="status")["stdout"]) or {}

            if cont["returncode"] == 0 and not st.get("changes"):

                result["tree"] = worktree_manifest(ours_dir)

                result["result_source"] = "committed-after-sentinel-continue"

                result["conflict_worktree_stale_paths"] = sorted(

                    p for p in set(result["conflict_worktree"]) - set(result["tree"])

                    if p not in result["conflicts"])

            else:

                result["result_source"] = "conflict-worktree (continue failed)"

        elif pull["returncode"] == 0 and receipt.get("parent_sync") in ("merged", "up_to_date",

                                                                       "fast_forward", "fast_forwarded"):

            result["clean"] = True

            result["conflicts"] = {}

            st = _json_tail(run.oak(ours_dir, "status", "--json", label="status")["stdout"]) or {}

            result["worktree_matches_head"] = not st.get("changes")

            again = run.oak(ours_dir, "pull", "--json", ok=None, label="idempotence-pull")

            again_receipt = _json_tail(again["stdout"]) or {}

            result["idempotence"] = {

                "returncode": again["returncode"],

                "parent_sync": again_receipt.get("parent_sync"),

                "head_unchanged": again_receipt.get("head_after") == receipt.get("head_after"),

                "tree_unchanged": worktree_manifest(ours_dir) == result["tree"],

            }

        else:

            result["clean"] = None

            result["error"] = (receipt.get("parent_sync_error") or pull["stderr"][-500:]

                               or f"oak pull exited {pull['returncode']}")

            result["error_kind"] = "oak"

            result["post_failure"] = post_failure_invariants(run, ours_dir, pre, result["tree"])

    except StepFailed as exc:

        result["error"] = f"{exc}: {exc.step['stderr'][-400:]}"

        result["error_kind"] = "oak-setup"

    except (HarnessError, ValueError, RuntimeError, OSError, subprocess.TimeoutExpired) as exc:

        result["error"] = f"{type(exc).__name__}: {exc}"

        result["error_kind"] = "harness"

    result["shim_log"] = [line for line in server.shim.log if repo_key.split("/")[1] in line][-5:]

    return result

def post_failure_invariants(run: "OakRun", ours_dir: Path, pre: dict, tree_after: dict) -> dict:

    """After a failed (non-conflict) pull, check HEAD vs worktree vs what a retry claims.

    A failed sync must leave the branch where it was (HEAD unchanged, worktree

    matching HEAD). The hazard: HEAD advanced to a merge commit while the

    worktree still holds the pre-merge state, so the next pull says

    up_to_date and the next commit silently reverts main's changes."""

    st = _json_tail(run.oak(ours_dir, "status", "--json", ok=None, label="post-failure-status")["stdout"]) or {}

    retry = run.oak(ours_dir, "pull", "--json", ok=None, label="post-failure-repull")

    retry_receipt = _json_tail(retry["stdout"]) or {}

    head_moved = bool(pre.get("head")) and st.get("head") != pre.get("head")

    changes = st.get("changes") or []

    return {

        "head_before": pre.get("head"), "head_after": st.get("head"), "head_moved": head_moved,

        "worktree_changes_vs_head": len(changes),

        "worktree_changes_sample": [c.get("path") for c in changes[:8]],

        "sync_in_progress": st.get("sync_in_progress"),

        "repull_returncode": retry["returncode"], "repull_parent_sync": retry_receipt.get("parent_sync"),

        "pre_pull_worktree_changes": len(pre.get("changes") or []),

        "hazard_head_moved_worktree_stale": head_moved and bool(changes),

        # HEAD stayed put but the failed sync half-applied the merge to a worktree that

        # was clean before the pull: the invariant "failed sync => worktree matches HEAD".

        "hazard_worktree_dirty_head_unmoved": (not head_moved) and bool(changes)

        and not (pre.get("changes") or []),

    }

def _point_head(git: Git, bare: Path, commit: str) -> None:

    g = Git(bare, git.bin, git.home)

    g.run("update-ref", "refs/heads/oak-import", commit)

    g.run("symbolic-ref", "HEAD", "refs/heads/oak-import")

def _git_messages(git: Git, bare: Path) -> Dict[str, str]:

    g = Git(bare, git.bin, git.home)

    raw = g.run("log", "--all", "--format=%H%x00%B%x01").stdout.decode("utf-8", "replace")

    out = {}

    for rec in raw.split("\x01"):

        rec = rec.strip("\n")

        if "\x00" in rec:

            h, msg = rec.split("\x00", 1)

            out[h.strip()] = msg.strip()

    return out

def summarize_prediction(doc: Optional[dict], returncode: int) -> dict:

    if not doc:

        return {"available": False, "returncode": returncode}

    return {

        "available": bool(doc.get("prediction_available")),

        "returncode": returncode,

        "method": doc.get("prediction_method"),

        "clean": doc.get("clean"),

        "conflict_files": sorted(_path_of(x) for x in doc.get("conflict_files") or []),

        "certified": (doc.get("merge_safety") or {}).get("certified"),

        "fork_base": (doc.get("merge_safety") or {}).get("fork_base_commit"),

    }

def _path_of(item) -> str:

    return item.get("path") if isinstance(item, dict) else str(item)

def parse_conflict_show(doc: Optional[dict], receipt: dict) -> Dict[str, List[str]]:

    """{path: [oak-reported kinds]} from `oak conflict show --json` (fallback: receipt)."""

    paths: Dict[str, List[str]] = {}

    if doc:

        for key in ("paths", "conflicts", "files", "conflict_paths"):

            items = doc.get(key)

            if isinstance(items, list):

                for item in items:

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

                        kind = item.get("kind") or item.get("conflict_kind") or item.get("type")

                        facts = [str(kind)] if kind else []

                        if item.get("has_conflict_markers") is False:

                            facts.append("no-markers")

                        if item.get("exists") is False:

                            facts.append("absent-in-worktree")

                        paths[item["path"]] = facts

                    elif isinstance(item, str):

                        paths[item] = []

                if paths:

                    break

    if not paths:

        for p in receipt.get("conflict_paths") or []:

            paths[p] = []

    return dict(sorted(paths.items()))

class HostedMergeAdapter:

    """Phase-2 plug point: server-side (oakspace) merge of branch into main.

    Contract: given the same plan, return the same result dict shape as

    run_pull_sync with path_exercised="oakspace-server-merge". Needs a local

    oakspace (Postgres + oak-web) exposing POST /api/:o/:r/branches/:b/merge.

    Not implemented in phase 1.

    """

    path_exercised = "oakspace-server-merge"

    def run(self, *args, **kwargs):  # pragma: no cover - phase 2

        raise NotImplementedError("hosted merge adapter is phase 2")