"""Isolated mediator, SCM gate, and model-facing API contracts."""

from __future__ import annotations

import json
import subprocess
import sys
import threading
import urllib.error
import urllib.request
from pathlib import Path

import pytest


ROOT = Path(__file__).parents[2]
SCRIPTS = ROOT / "services/hermes/scripts"
SCM_SCRIPTS = ROOT / "services/hermes/scm-common/scripts"
sys.path[:0] = [str(SCRIPTS), str(SCM_SCRIPTS)]

import execution_pool_client as client  # noqa: E402
import execution_pool_protocol as protocol  # noqa: E402
import execution_pool_scm as scm  # noqa: E402


KEY = b"d" * 64
RESULT = {
    "status": "completed",
    "summary": "Completed safely.",
    "changed_files": ["safe.py"],
    "tests_run": ["pytest"],
    "artifacts": [],
    "findings": [],
    "blockers": [],
}


def binding(**changes):
    value = {
        "board": "metis", "task_id": "t_deadbeef", "run_id": "42",
        "worker_ordinal": 0, "attempt": 1,
    }
    value.update(changes)
    return value


def payload(**changes):
    value = {
        "context": "safe objective",
        "repo_url": "https://scm.bstein.dev/titan/metis.git",
        "branch": "wt/t_deadbeef",
        "base_branch": "main",
    }
    value.update(changes)
    return value


def assignment(**payload_changes):
    return protocol.sign_envelope(
        KEY, "assignment", binding(), payload(**payload_changes)
    )


class Response:
    def __init__(self, body):
        self.body = body

    def __enter__(self):
        return self

    def __exit__(self, *_args):
        return False

    def read(self, _size=-1):
        return self.body


class FakeSCM:
    def __init__(self):
        self.submitted = []

    def checkout(self, _envelope):
        return {"workspace": "/workspace/run", "baseline_sha": "a" * 40}

    def submit(self, envelope, request):
        self.submitted.append((envelope, request))
        return {"pull_request": "https://scm.bstein.dev/titan/metis/pulls/7"}


def test_result_schema_validation_rejects_every_unsafe_shape():
    invalid = [
        None,
        {},
        {"structured": []},
        {"structured": {**RESULT, "extra": True}},
        {"structured": {**RESULT, "status": "unknown"}},
        {"structured": {**RESULT, "summary": ""}},
        {"structured": {**RESULT, "tests_run": "pytest"}},
        {"structured": {**RESULT, "tests_run": [7]}},
    ]
    for value in invalid:
        with pytest.raises(protocol.ProtocolError):
            client._validate_result(value)
    assert client._validate_result({"structured": dict(RESULT)})["structured"] == RESULT


def test_client_post_verifies_response_and_bounds_body(monkeypatch):
    boundary = client.ClientBoundary(KEY, FakeSCM())
    ack = protocol.sign_envelope(KEY, "ack", binding(), {"accepted": True})
    monkeypatch.setattr(client.urllib.request, "urlopen", lambda *_a, **_k: Response(protocol.canonical_json(ack)))
    assert boundary._post("/v1/result", ack)["kind"] == "ack"
    monkeypatch.setattr(
        client.urllib.request,
        "urlopen",
        lambda *_a, **_k: Response(b"x" * (protocol.MAX_WIRE_BYTES + 1)),
    )
    with pytest.raises(protocol.ProtocolError, match="exceeds"):
        boundary._post("/v1/result", ack)


def test_poll_materializes_only_ordinal_owned_assignment(monkeypatch):
    monkeypatch.setattr(client, "ORDINAL", 0)
    boundary = client.ClientBoundary(KEY, FakeSCM())
    boundary._post = lambda *_a: protocol.sign_envelope(
        KEY,
        "ack",
        binding(board="", task_id="", run_id="", attempt=0),
        {"assignment": None},
    )
    assert boundary.poll() == {"assignment": None}

    boundary._post = lambda *_a: assignment()
    result = boundary.poll()["assignment"]
    assert result["workspace"] == "/workspace/run"
    assert result["protocol_version"] == 2
    assert "signature" not in result

    boundary._post = lambda *_a: protocol.sign_envelope(
        KEY, "assignment", binding(worker_ordinal=1), payload()
    )
    with pytest.raises(protocol.ProtocolError, match="foreign"):
        boundary.poll()


def test_heartbeat_and_finish_require_exact_current_binding(monkeypatch):
    fake_scm = FakeSCM()
    boundary = client.ClientBoundary(KEY, fake_scm)
    current = assignment()
    boundary.current = current

    def post(path, envelope):
        kind = "heartbeat" if path.endswith("heartbeat") else "result"
        protocol.verify_envelope(KEY, envelope, expected_kind=kind)
        return protocol.sign_envelope(
            KEY, "ack", binding(), {"accepted": True, "duplicate": False}
        )

    boundary._post = post
    assert boundary.heartbeat(
        {"binding": binding(), "payload": {"note": "active"}}
    )["ack"]["accepted"]
    request = {
        "binding": binding(),
        "payload": {"structured": dict(RESULT), "returncode": 0},
        "title": "Safe change",
        "body": "Evidence",
    }
    finished = boundary.finish(request)
    assert finished["ack"]["accepted"] and boundary.current is None
    assert fake_scm.submitted
    assert finished["structured"]["artifacts"] == [
        "https://scm.bstein.dev/titan/metis/pulls/7"
    ]

    boundary.current = current
    with pytest.raises(protocol.ProtocolError, match="heartbeat payload"):
        boundary.heartbeat({"binding": binding(), "payload": "bad"})
    with pytest.raises(protocol.ProtocolError, match="object"):
        boundary._current_for(None)
    with pytest.raises(protocol.ProtocolError, match="does not own"):
        boundary._current_for(binding(attempt=2))


def test_noncompleted_finish_never_invokes_scm_and_bad_ack_is_rejected():
    fake_scm = FakeSCM()
    boundary = client.ClientBoundary(KEY, fake_scm)
    boundary.current = assignment()
    bad = protocol.sign_envelope(KEY, "ack", binding(attempt=2), {"accepted": True})
    boundary._post = lambda *_a: bad
    blocked = {**RESULT, "status": "blocked", "blockers": ["capacity"]}
    with pytest.raises(protocol.ProtocolError, match="binding changed"):
        boundary.finish(
            {"binding": binding(), "payload": {"structured": blocked, "returncode": 1}}
        )
    assert fake_scm.submitted == []

    boundary._post = lambda *_a: protocol.sign_envelope(
        KEY, "result", binding(), {"accepted": True}
    )
    with pytest.raises(protocol.ProtocolError, match="binding changed"):
        boundary.heartbeat({"binding": binding(), "payload": {}})


def http_request(handler, path, *, body=None):
    server = protocol.BoundedHTTPServer(("127.0.0.1", 0), handler, max_workers=2)
    thread = threading.Thread(target=server.handle_request)
    thread.start()
    request = urllib.request.Request(
        f"http://127.0.0.1:{server.server_port}{path}",
        data=body,
        method="POST" if body is not None else "GET",
        headers={"Content-Type": "application/json"},
    )
    try:
        with urllib.request.urlopen(request, timeout=3) as response:
            return response.status, json.loads(response.read())
    except urllib.error.HTTPError as error:
        return error.code, json.loads(error.read())
    finally:
        thread.join(timeout=3)
        server.server_close()


def test_model_api_exposes_only_gated_state_machine_operations():
    class Boundary:
        poll = staticmethod(lambda: {"assignment": None})
        heartbeat = staticmethod(lambda request: {"heartbeat": request["payload"]})
        finish = staticmethod(lambda request: {"finish": request["payload"]})

    handler = client.handler_factory(Boundary())
    assert http_request(handler, "/ready") == (
        200, {"protocol_version": 2, "ready": True}
    )
    assert http_request(handler, "/missing")[0] == 404
    for operation in ("poll", "heartbeat", "finish"):
        body = protocol.canonical_json(
            {"operation": operation, "binding": binding(), "payload": {}}
        )
        assert http_request(handler, "/v1/client", body=body)[0] == 200
    for value in (
        {"operation": "bypass"},
        {"operation": "poll", "authority": "steal"},
    ):
        status, response = http_request(
            handler, "/v1/client", body=protocol.canonical_json(value)
        )
        assert status == 409 and response.get("error")


def test_client_main_validates_ordinal_and_starts_bounded_server(monkeypatch):
    monkeypatch.setattr(client, "ORDINAL", -1)
    with pytest.raises(SystemExit, match="ORDINAL"):
        client.main()
    started = []

    class Server:
        def __init__(self, address, _handler, max_workers):
            started.append((address, max_workers))

        def serve_forever(self):
            return

    monkeypatch.setattr(client, "ORDINAL", 2)
    monkeypatch.setattr(client, "read_key", lambda _path: KEY)
    monkeypatch.setattr(client, "BoundedHTTPServer", Server)
    monkeypatch.setattr(client, "SCMBoundary", lambda _key: FakeSCM())
    assert client.main() == 0
    assert started == [(("0.0.0.0", client.PORT), 4)]


def init_checkout(path, branch="wt/t_deadbeef"):
    subprocess.run(["git", "init", "-q", str(path)], check=True)
    commands = [
        ("remote", "add", "origin", "https://scm.bstein.dev/titan/metis.git"),
        (
            "remote",
            "add",
            "hermes-broker",
            "http://hermes-scm-broker.hermes-scm.svc.cluster.local:9081/"
            "git/atlas/metis.git",
        ),
        ("checkout", "-qb", branch),
        ("config", "user.email", "test@example.com"),
        ("config", "user.name", "Test"),
    ]
    for command in commands:
        subprocess.run(["git", "-C", str(path), *command], check=True)
    (path / "tracked").write_text("safe\n")
    subprocess.run(["git", "-C", str(path), "add", "tracked"], check=True)
    subprocess.run(["git", "-C", str(path), "commit", "-qm", "initial"], check=True)
    return subprocess.check_output(
        ["git", "-C", str(path), "rev-parse", "HEAD"], text=True
    ).strip()


def test_scm_run_and_binding_validation(tmp_path, monkeypatch):
    checkout = tmp_path / "checkout"
    head = init_checkout(checkout)
    assert scm._run("rev-parse", "HEAD", cwd=checkout) == head
    with pytest.raises(RuntimeError):
        scm._run("rev-parse", "missing", cwd=checkout)
    monkeypatch.setattr(scm, "MAX_STATUS_BYTES", 1)
    with pytest.raises(protocol.ProtocolError, match="output"):
        scm._run("rev-parse", "HEAD", cwd=checkout)

    monkeypatch.setattr(scm, "ORDINAL", 0)
    assert scm._binding(assignment())[1:] == ("metis", "wt/t_deadbeef", "main")
    for envelope in (
        {**assignment(), "kind": "result"},
        {**assignment(), "payload": []},
        assignment(repo_url="https://evil.example/metis.git"),
        assignment(branch="main"),
        assignment(base_branch="../main"),
        protocol.sign_envelope(KEY, "assignment", binding(worker_ordinal=1), payload()),
    ):
        with pytest.raises(protocol.ProtocolError):
            scm._binding(envelope)


def test_scm_private_paths_and_text_are_bounded(tmp_path, monkeypatch):
    monkeypatch.setattr(scm, "WORKSPACE_ROOT", tmp_path / "workspace")
    monkeypatch.setattr(scm, "SCM_ROOT", tmp_path / "state")
    target = scm.workspace_path(assignment())
    assert target.name == "42"
    assert scm._state_path(assignment()).name == "metis-t_deadbeef-42.json"
    bad = {**assignment(), "task_id": "../bad"}
    with pytest.raises(protocol.ProtocolError, match="binding"):
        scm.workspace_path(bad)
    state = tmp_path / "regular"
    state.write_text("safe")
    assert scm._regular_text(state, 10) == "safe"
    with pytest.raises(protocol.ProtocolError, match="invalid"):
        scm._regular_text(state, 1)
    binary = tmp_path / "binary"
    binary.write_bytes(b"\xff")
    with pytest.raises(protocol.ProtocolError, match="malformed"):
        scm._regular_text(binary, 10)
    linked_state = tmp_path / "linked-state"
    linked_state.mkdir()
    (tmp_path / "state-link").symlink_to(linked_state, target_is_directory=True)
    monkeypatch.setattr(scm, "SCM_ROOT", tmp_path / "state-link")
    with pytest.raises(protocol.ProtocolError):
        scm._state_path(assignment())


def test_boundary_verification_existing_checkout_and_missing_baseline(tmp_path, monkeypatch):
    workspace = tmp_path / "workspace"
    state_root = tmp_path / "state"
    monkeypatch.setattr(scm, "WORKSPACE_ROOT", workspace)
    monkeypatch.setattr(scm, "SCM_ROOT", state_root)
    monkeypatch.setattr(scm, "ORDINAL", 0)
    boundary = scm.Boundary(KEY)
    assert boundary.verify(assignment())["kind"] == "assignment"
    destination = scm.workspace_path(assignment())
    baseline = init_checkout(destination)
    protocol.atomic_json(
        scm._state_path(assignment()),
        {"baseline_sha": baseline, "repo": "metis", "branch": "wt/t_deadbeef"},
    )
    assert boundary.checkout(assignment())["baseline_sha"] == baseline
    scm._state_path(assignment()).write_text("{}")
    with pytest.raises(protocol.ProtocolError, match="baseline"):
        boundary.checkout(assignment())


@pytest.mark.parametrize("feature_exists", [True, False])
def test_new_checkout_uses_broker_and_safe_base_fallback(
    tmp_path, monkeypatch, feature_exists
):
    monkeypatch.setattr(scm, "WORKSPACE_ROOT", tmp_path / "workspace")
    monkeypatch.setattr(scm, "SCM_ROOT", tmp_path / "state")
    monkeypatch.setattr(scm, "ORDINAL", 0)
    calls = []

    def run(*arguments, cwd=None, timeout=300):
        calls.append((arguments, cwd, timeout))
        if arguments[0] == "clone":
            branch = arguments[arguments.index("--branch") + 1]
            destination = Path(arguments[-1])
            if branch == "wt/t_deadbeef" and not feature_exists:
                raise RuntimeError("missing branch")
            (destination / ".git").mkdir(parents=True)
        return ""

    monkeypatch.setattr(scm, "_run", run)
    monkeypatch.setattr(scm, "_workspace_identity", lambda *_a: "a" * 40)
    result = scm.Boundary(KEY).checkout(assignment())
    assert result["baseline_sha"] == "a" * 40
    assert any("hermes-scm-broker" in str(call) for call in calls)
    if not feature_exists:
        assert any(call[0][0] == "checkout" for call in calls)


def test_failed_clone_never_deletes_unmanaged_state(tmp_path, monkeypatch):
    monkeypatch.setattr(scm, "WORKSPACE_ROOT", tmp_path / "workspace")
    monkeypatch.setattr(scm, "SCM_ROOT", tmp_path / "state")
    monkeypatch.setattr(scm, "ORDINAL", 0)

    def fail(*arguments, **_kwargs):
        destination = Path(arguments[-1])
        destination.mkdir(parents=True, exist_ok=True)
        (destination / "preserved").write_text("owner data")
        raise RuntimeError("clone failed")

    monkeypatch.setattr(scm, "_run", fail)
    with pytest.raises(protocol.ProtocolError, match="unmanaged"):
        scm.Boundary(KEY).checkout(assignment())
    assert (scm.workspace_path(assignment()) / "preserved").read_text() == "owner data"


def test_draft_reuse_create_and_submit_gates(tmp_path, monkeypatch):
    existing = json.dumps([{"number": 1, "html_url": "https://scm/pulls/1", "head": {"ref": "wt/task"}, "base": {"ref": "main"}}]).encode()
    monkeypatch.setattr(scm.scm_broker_client, "read", lambda _path: existing)
    monkeypatch.setattr(scm.scm_broker_client, "update_draft", lambda *_a, **_k: json.dumps({"html_url": "https://scm/pulls/1"}).encode())
    assert scm.Boundary._draft("metis", "wt/task", "main", "a" * 40, "t", "b", "grant") == "https://scm/pulls/1"
    monkeypatch.setattr(scm.scm_broker_client, "read", lambda _path: b"[]")
    monkeypatch.setattr(
        scm.scm_broker_client, "create_draft",
        lambda *_a, **_k: json.dumps({"html_url": "https://scm/pulls/2"}).encode(),
    )
    assert scm.Boundary._draft("metis", "wt/task", "main", "a" * 40, "t", "b") == "https://scm/pulls/2"

    monkeypatch.setattr(scm, "WORKSPACE_ROOT", tmp_path / "workspace")
    monkeypatch.setattr(scm, "SCM_ROOT", tmp_path / "state")
    monkeypatch.setattr(scm, "ORDINAL", 0)
    destination = scm.workspace_path(assignment())
    destination.mkdir(parents=True)
    state = scm._state_path(assignment())
    protocol.atomic_json(state, {"baseline_sha": "a" * 40})
    boundary = scm.Boundary(KEY)
    monkeypatch.setattr(scm, "_workspace_identity", lambda *_a: "b" * 40)
    outputs = {"status": "", "rev-list": "1", "push": ""}

    def run(*arguments, **_kwargs):
        return outputs.get(arguments[0], "")

    monkeypatch.setattr(scm, "_run", run)
    monkeypatch.setattr(boundary, "_draft", lambda *_a: "https://scm/pulls/3")
    monkeypatch.setattr(boundary, "_grant", lambda *_a: "signed-grant")
    monkeypatch.setattr(scm.scm_broker_client, "register_task", lambda _grant: b'{"registered":true}')
    result = boundary.submit(assignment(), {"title": "safe", "body": "evidence"})
    assert result["pull_request"] == "https://scm/pulls/3"
    outputs["status"] = "?? untracked"
    with pytest.raises(protocol.ProtocolError, match="uncommitted"):
        boundary.submit(assignment(), {})
    outputs["status"] = ""
    outputs["rev-list"] = "0"
    assert boundary.submit(assignment(), {})["pull_request"] == ""
    with pytest.raises(protocol.ProtocolError, match="metadata"):
        boundary.submit(assignment(), {"title": "x" * 513, "body": "x"})