"""HuxClient transport contract and the pure helpers of the agent hook.

Security obligations exercised: SO-04 (the client emits exactly the identity
header vocabulary ``hux.identity`` accepts, with the worker key), SO-07 (errors
carry status, code and message only; request bodies are never embedded),
SO-27 and SO-28 (memory writes need the privacy card and are refused in a
private conversation), SO-37 (the argument hash is stable across key order
and whitespace), SO-50 (capabilities are read from the resolved flag chain).
"""

from __future__ import annotations

import ast
import http.client
import io
import sys
import threading
import urllib.error
import urllib.request
from pathlib import Path

import pytest

ROOT = Path(__file__).resolve().parents[2]
FOUNDATION = ROOT / "dockerfiles" / "hermes-hux-foundation"
HOOK_ROOT = ROOT / "dockerfiles" / "hermes-worker-hux"
for entry in (FOUNDATION, HOOK_ROOT):
    if str(entry) not in sys.path:
        sys.path.insert(0, str(entry))

from hux import contracts, identity  # noqa: E402
from hux.http import serve  # noqa: E402
from hux.server import build_router  # noqa: E402
from hux_hook import HuxClient, HuxServiceError, HuxUnavailable, canonical_argument_hash, emit, memory_gate  # noqa: E402
from hux_hook import client as client_mod  # noqa: E402
from hux_hook import hooks  # noqa: E402

SCHEMAS = contracts.load_all()
ALL_ON = ",".join(card["flag"] for card in contracts.load_flags()["cards"])
SUBJECT = "usr_0123456789abcdef"
WORKER = {"tenant_slot": "slot-3", "subject": SUBJECT, "surface": "worker", "trust": "worker"}
HUMAN = {"tenant_slot": "slot-3", "subject": SUBJECT, "surface": "chat", "trust": "router"}
OTHER = {**HUMAN, "subject": "usr_fedcba9876543210"}
CANARY = "CANARY-9c1d-SECRET"


def start(tmp_path: Path, flags: str = ALL_ON):
    router = build_router(tmp_path, {"HUX_FLAGS": flags, "HUX_WORKER_KEY": "wk"})
    server = serve(router, "127.0.0.1", 0)
    threading.Thread(target=server.serve_forever, daemon=True).start()
    return f"http://127.0.0.1:{server.server_address[1]}", server


@pytest.fixture
def live(tmp_path):
    base, server = start(tmp_path)
    yield base, tmp_path
    server.shutdown()


# --- headers and identity ---------------------------------------------------------

def test_headers_match_the_service_vocabulary():
    """SO-04: the exact header names identity.resolve reads, key only when present, extras only when non-empty."""
    worker = HuxClient("http://127.0.0.1:1/", WORKER, key="wk")
    sent = worker.headers({"Idempotency-Key": "run:approval:1", "If-Match": ""})
    assert identity.resolve(sent, {"HUX_WORKER_KEY": "wk"}) == identity.Identity("slot-3", SUBJECT, "worker", "worker")
    assert sent["Idempotency-Key"] == "run:approval:1" and "If-Match" not in sent
    assert worker.base_url == "http://127.0.0.1:1"
    plain = HuxClient(identity={"tenant_slot": "slot-3", "subject": SUBJECT}).headers()
    assert client_mod.HEADER_KEY not in plain and plain[client_mod.HEADER_TRUST] == "worker"
    assert HuxClient().identity == {"tenant_slot": "", "subject": "", "surface": "worker", "trust": "worker"}


def test_error_mapping_and_no_body_leak(live):
    """SO-07: a hux.error.v1 answer becomes HuxServiceError(status, code, message) and the body never appears in it."""
    base, _ = live
    human = HuxClient(base, HUMAN)
    with pytest.raises(HuxServiceError) as bad:
        human.post("/hux/v1/approvals", {"conversation_id": "conv_0001abcd", "capability": "nope", "secret": CANARY})
    assert (bad.value.status, bad.value.code) == (400, "invalid") and CANARY not in str(bad.value)
    with pytest.raises(HuxServiceError) as unauth:
        HuxClient(base, WORKER, key="wrong").get("/hux/v1/capabilities")
    assert unauth.value.code == "unauthorized"
    with pytest.raises(HuxServiceError) as missing:
        human.get("/hux/v1/no/such/route")
    assert (missing.value.status, missing.value.code) == (404, "not_found")
    assert HuxUnavailable().code == "unavailable" and HuxUnavailable().status == 0
    assert HuxServiceError(500, "x", "m" * 400).message == "m" * 280


def test_transport_edge_cases(monkeypatch):
    """Non-JSON failures, a 4xx delivered without an exception and socket errors map to typed errors."""
    client = HuxClient("http://127.0.0.1:1", WORKER, key="wk")

    def raise_http(*args, **kwargs):
        raise urllib.error.HTTPError("u", 502, "bad gateway", {}, io.BytesIO(b"<html>"))

    monkeypatch.setattr(urllib.request, "urlopen", raise_http)
    with pytest.raises(HuxServiceError) as html:
        client.get("/hux/v1/capabilities")
    assert (html.value.status, html.value.code) == (502, "invalid")

    class Raw:
        status = 418
        headers = {"X-Test": "1"}

        def read(self):
            return b"{bad json"

        def __enter__(self):
            return self

        def __exit__(self, *exc):
            return False

    monkeypatch.setattr(urllib.request, "urlopen", lambda *a, **k: Raw())
    with pytest.raises(HuxServiceError) as teapot:
        client.put("/hux/v1/policy", {"x": 1}, if_match=3)
    assert teapot.value.status == 418

    def raise_socket(*args, **kwargs):
        raise TimeoutError("slow")

    monkeypatch.setattr(urllib.request, "urlopen", raise_socket)
    with pytest.raises(HuxUnavailable):
        client.get("/hux/v1/capabilities", query={"a": "b c"})

    def raise_half_closed(*args, **kwargs):
        raise http.client.BadStatusLine("gone")

    monkeypatch.setattr(urllib.request, "urlopen", raise_half_closed)
    with pytest.raises(HuxUnavailable):
        client.get("/hux/v1/capabilities")


def test_put_with_if_match_and_get_with_query(live):
    """Revisioned writes send If-Match; a stale revision is a conflict; queries reach the service."""
    base, _ = live
    human = HuxClient(base, HUMAN)
    first = human.put("/hux/v1/policy", {"scope": {"level": "global"}, "autonomy": "safe"})
    assert first.header("ETag") == "1" and first.header("Missing") == ""
    second = human.put("/hux/v1/policy", {"scope": {"level": "global"}, "autonomy": "autonomous"}, if_match=1)
    assert second.body["revision"] == 2
    with pytest.raises(HuxServiceError) as stale:
        human.put("/hux/v1/policy", {"scope": {"level": "global"}, "autonomy": "safe"}, if_match=1)
    assert stale.value.code == "conflict"
    assert human.get("/hux/v1/policy", {"scope": "global"}).body["autonomy"] == "autonomous"


# --- capabilities ---------------------------------------------------------------------

def test_capabilities_cached_per_process(live, monkeypatch):
    """SO-50: one capabilities read per process; refresh and forget re-read; failures are not cached."""
    base, _ = live
    worker = HuxClient(base, WORKER, key="wk")
    calls = []
    real_get = worker.get
    monkeypatch.setattr(worker, "get", lambda path, query=None: calls.append(path) or real_get(path, query))
    assert worker.card_enabled("HUX-05") and worker.card_enabled("HUX-01")
    assert worker.capabilities()["contract_version"] == contracts.load_flags()["contract_version"]
    assert len(calls) == 1
    worker.capabilities(refresh=True)
    worker.forget_capabilities()
    worker.capabilities()
    assert len(calls) == 3
    monkeypatch.setattr(worker, "get", lambda path, query=None: (_ for _ in ()).throw(HuxServiceError(404, "flag_off", "off")))
    worker.forget_capabilities()
    assert worker.capabilities() == {"reachable": True, "cards": {}, "contract_version": ""}
    dead = HuxClient("http://127.0.0.1:1", WORKER, key="wk", timeout=1)
    assert dead.capabilities()["reachable"] is False and not dead.card_enabled("HUX-11")


# --- events ---------------------------------------------------------------------------

def test_emit_is_best_effort_and_redaction_safe(live):
    """SO-11: detail outside the allowlist is dropped by the service; failures return None and never raise."""
    base, root = live
    human = HuxClient(base, HUMAN)
    conv = human.post("/hux/v1/conversations", {"title": "t"}).body["id"]
    worker = HuxClient(base, WORKER, key="wk")
    record = emit(worker, conv, "tool.call", "shell call", {"tool": "shell", "arguments": {"cmd": CANARY}, "argument_bytes": 7},
                  evidence=[{"kind": "run", "id": "run_1"}], run_id="run_1", turn=2, correlation_id="corr-1", idempotency_key="run_1:call:0001")
    assert contracts.validate_record(record, SCHEMAS) == [] and record["detail"] == {"tool": "shell", "argument_bytes": 7}
    assert record["provenance"]["actor"] == {"type": "system", "id": "hux-worker"} and record["correlation_id"] == "corr-1"
    assert emit(worker, conv, "tool.call", "again", idempotency_key="run_1:call:0001")["id"] == record["id"]
    assert emit(worker, conv, "not.a.kind", "x") is None
    assert emit(worker, "conv_unknown0001", "tool.call", "x") is None
    assert emit(worker, conv, "tool.call", "x", {"tool": object()}) is None
    assert emit(HuxClient(base, OTHER), conv, "tool.call", "cross tenant") is None
    private = human.post("/hux/v1/conversations", {"title": "p", "mode": "private"}).body["id"]
    assert emit(worker, private, "tool.call", "private mode writes nothing") is None
    assert CANARY not in "\n".join(p.read_text(errors="ignore") for p in root.rglob("*") if p.is_file())


# --- memory gate ------------------------------------------------------------------------

def test_memory_gate(live, tmp_path):
    """SO-27, SO-28: allowed only when the privacy card answers and the conversation is not private."""
    base, _ = live
    human = HuxClient(base, HUMAN)
    worker = HuxClient(base, WORKER, key="wk")
    normal = human.post("/hux/v1/conversations", {"title": "n", "mode": "thoughtful"}).body["id"]
    private = human.post("/hux/v1/conversations", {"title": "p", "mode": "private"}).body["id"]
    assert memory_gate(worker, normal) is True
    assert memory_gate(worker, private) is False
    assert memory_gate(worker, "conv_notknown01") is True
    assert memory_gate(worker, "bad id") is False
    human.post(f"/hux/v1/conversations/{normal}/forget", {})
    assert memory_gate(worker, normal) is False
    assert memory_gate(HuxClient("http://127.0.0.1:1", WORKER, key="wk", timeout=1), normal) is False
    off_base, off_server = start(tmp_path / "off", "hux.foundation,hux.projects")
    try:
        assert memory_gate(HuxClient(off_base, WORKER, key="wk"), normal) is False
    finally:
        off_server.shutdown()


# --- pure helpers -------------------------------------------------------------------------

def test_hash_is_canonical():
    """SO-37: key order, whitespace and nesting order of dict keys do not change the hash; values do."""
    a = canonical_argument_hash("write", {"path": "n.md", "opts": {"b": 1, "a": [1, 2]}})
    b = canonical_argument_hash("write", {"opts": {"a": [1, 2], "b": 1}, "path": "n.md"})
    assert a == b and a.startswith("sha256:") and len(a) == 71
    assert canonical_argument_hash("write", {"path": "n.md ", "opts": {"b": 1, "a": [1, 2]}}) != a
    assert canonical_argument_hash("other", {"path": "n.md", "opts": {"b": 1, "a": [1, 2]}}) != a
    assert canonical_argument_hash("w", "raw string") == canonical_argument_hash("w", "raw string")
    assert hooks.canonical_json({"z": "é", "a": None}) == b'{"a":null,"z":"\\u00e9"}'


def test_key_and_ref_helpers():
    """Idempotency keys always satisfy the contract pattern; call refs never contain arguments."""
    assert hooks.idempotency_key("r", "a", "") == "r:a:.pad"
    assert hooks.idempotency_key("r", "a", "x") == "r:a:x.pad"
    key = hooks.idempotency_key("run id/with spaces", "approval", "f" * 200)
    assert len(key) == 120 and " " not in key and "/" not in key
    assert hooks.call_ref("my tool/x", "sha256:" + "ab" * 32) == "my-tool-x:abababababababab"
    assert hooks._failure_reason(HuxServiceError(404, "flag_off", "")) == "flag_off"
    assert hooks._failure_reason(HuxServiceError(0, "unavailable", "")) == "hux_unavailable"
    assert hooks._failure_reason(HuxServiceError(409, "conflict", "")) == "service_error:conflict"


def test_library_is_stdlib_only_small_and_documented():
    """Every module ≤ 500 lines, every public function and module documented, no third-party imports."""
    for path in sorted((HOOK_ROOT / "hux_hook").glob("*.py")):
        source = path.read_text()
        assert len(source.splitlines()) <= 500, path
        tree = ast.parse(source)
        assert ast.get_docstring(tree), path
        for node in ast.walk(tree):
            if isinstance(node, ast.FunctionDef | ast.ClassDef) and not node.name.startswith("_"):
                assert ast.get_docstring(node), f"{path.name}:{node.name}"
            if isinstance(node, ast.Import | ast.ImportFrom):
                root = (node.names[0].name if isinstance(node, ast.Import) else node.module or "").split(".")[0]
                assert root in {"http", "json", "threading", "urllib", "collections", "typing", "hashlib", "re", "dataclasses", "hux_hook", "__future__"}, root