"""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_ROUTER_KEY": "rk", "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_client_loads_only_a_bounded_0400_key_file(tmp_path):
    """The hook can consume its projected worker key without an inline environment secret."""
    key_file = tmp_path / "worker-key"
    key_file.write_text("wk\n")
    key_file.chmod(0o400)
    client = HuxClient("http://127.0.0.1:1", WORKER, key_file=key_file)
    assert client.headers()[client_mod.HEADER_KEY] == "wk"
    with pytest.raises(ValueError):
        HuxClient("http://127.0.0.1:1", WORKER, key="wk", key_file=key_file)
    key_file.chmod(0o444)
    with pytest.raises(ValueError):
        HuxClient("http://127.0.0.1:1", WORKER, key_file=key_file)


def test_client_loads_router_bound_subject_from_file(tmp_path, monkeypatch):
    """The worker gets its subject only from the shared router binding and rejects conflicts."""
    subject_file = tmp_path / "subject"
    subject_file.write_text(SUBJECT + "\n")
    subject_file.chmod(0o440)
    unbound_worker = {**WORKER, "subject": ""}
    assert HuxClient(identity=unbound_worker, subject_file=subject_file).identity["subject"] == SUBJECT
    assert HuxClient(identity=WORKER, subject_file=subject_file).identity["subject"] == SUBJECT
    with pytest.raises(ValueError, match="conflicts"):
        HuxClient(identity={**WORKER, "subject": OTHER["subject"]}, subject_file=subject_file)
    monkeypatch.setenv("HUX_SUBJECT_FILE", str(subject_file))
    assert HuxClient(identity=unbound_worker).identity["subject"] == SUBJECT


def test_client_subject_file_failures_are_closed(tmp_path):
    """Unavailable, linked, weak, malformed, oversized, and non-UTF-8 subject files are rejected."""
    with pytest.raises(ValueError, match="unavailable"):
        HuxClient(identity=WORKER, subject_file=tmp_path / "missing")
    for name, payload, mode, message in (
        ("empty", b"", 0o440, "empty or oversized"),
        ("oversized", b"x" * (client_mod.MAX_SUBJECT_BYTES + 1), 0o440, "empty or oversized"),
        ("unicode", b"\xff", 0o440, "not UTF-8"),
        ("malformed", b"brad@example.test", 0o440, "malformed"),
        ("weak", SUBJECT.encode(), 0o444, "0400 or 0440"),
    ):
        subject_file = tmp_path / name
        subject_file.write_bytes(payload)
        subject_file.chmod(mode)
        with pytest.raises(ValueError, match=message):
            HuxClient(identity=WORKER, subject_file=subject_file)
    valid = tmp_path / "valid"
    valid.write_text(SUBJECT)
    valid.chmod(0o400)
    linked = tmp_path / "linked"
    linked.symlink_to(valid)
    with pytest.raises(ValueError, match="unavailable"):
        HuxClient(identity=WORKER, subject_file=linked)


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, key="rk")
    worker = HuxClient(base, WORKER, key="wk")
    with pytest.raises(HuxServiceError) as bad:
        worker.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(client._opener, "open", 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, *_args):
            return b"{bad json"

        def __enter__(self):
            return self

        def __exit__(self, *exc):
            return False

    monkeypatch.setattr(client._opener, "open", 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(client._opener, "open", 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(client._opener, "open", raise_half_closed)
    with pytest.raises(HuxUnavailable):
        client.get("/hux/v1/capabilities")


@pytest.mark.parametrize("base", [
    "https://127.0.0.1:8790", "http://localhost:8790", "http://10.0.0.1:8790",
    "http://user:secret@127.0.0.1:8790", "http://127.0.0.1:8790/path",
    "http://127.0.0.1:8790?next=x", "http://127.0.0.1",
])
def test_client_accepts_only_literal_loopback_origin(base):
    """The worker client refuses non-loopback, credentialed, ambiguous and TLS origins."""
    with pytest.raises(ValueError):
        HuxClient(base, WORKER)
    assert HuxClient("http://[::1]:8790", WORKER).base_url == "http://[::1]:8790"
    with pytest.raises(ValueError):
        HuxClient("http://127.0.0.1:8790", WORKER, timeout=0)


@pytest.mark.parametrize("path", ["https://evil.invalid/hux/v1/x", "//evil.invalid/x", "/hux/v1/../x", "/hux/v1/%2e%2e/x", "/hux/v1/x?y=1", "/healthz"])
def test_client_rejects_noncanonical_paths_and_redirects(path, monkeypatch):
    """Paths stay same-origin and the opener never creates a redirected credential-bearing request."""
    monkeypatch.setattr(urllib.request, "getproxies", lambda: {"http": "http://proxy.invalid:8080"})
    client = HuxClient("http://127.0.0.1:8790", WORKER, key=CANARY)
    with pytest.raises(HuxServiceError) as invalid:
        client.get(path)
    assert invalid.value.status == 400
    assert client_mod._RejectRedirect().redirect_request(None, None, 302, "moved", {}, "https://evil.invalid") is None
    proxy_handlers = [handler for handler in client._opener.handlers if isinstance(handler, urllib.request.ProxyHandler)]
    assert proxy_handlers == [], "loopback credentials never enter an environment proxy"


def test_client_bounds_sidecar_responses_and_non_string_paths():
    """A compromised local sidecar cannot allocate an unbounded response or smuggle a non-string URL."""
    class Oversized:
        def read(self, amount):
            return b"x" * amount

    with pytest.raises(HuxUnavailable, match="oversized response"):
        client_mod._bounded_read(Oversized())
    with pytest.raises(HuxServiceError) as malformed:
        HuxClient("http://127.0.0.1:8790", WORKER).get(None)  # type: ignore[arg-type]
    assert malformed.value.status == 400


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, key="rk")
    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, key="rk")
    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, key="rk"), 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, key="rk")
    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 False
    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", "hmac", "json", "os", "stat", "threading", "urllib", "collections", "pathlib", "typing", "hashlib", "re", "dataclasses", "hux_hook", "__future__"}, root