"""Adversarial tests for the Hermes HUX execution middleware."""

from __future__ import annotations

import hashlib
import hmac
import importlib
import importlib.util
import json
import os
import sys
import threading
from pathlib import Path
from types import SimpleNamespace

import pytest

ROOT = Path(__file__).resolve().parents[2]
PLUGIN = ROOT / "services" / "hermes" / "plugins" / "hux-runtime"
HOOK_ROOT = ROOT / "dockerfiles" / "hermes-worker-hux"
sys.path.insert(0, str(HOOK_ROOT))
SPEC = importlib.util.spec_from_file_location(
    "hermes_hux_runtime",
    PLUGIN / "__init__.py",
    submodule_search_locations=[str(PLUGIN)],
)
assert SPEC and SPEC.loader
PACKAGE = importlib.util.module_from_spec(SPEC)
sys.modules[SPEC.name] = PACKAGE
SPEC.loader.exec_module(PACKAGE)
context_ids = importlib.import_module("hermes_hux_runtime.context_ids")
runtime_module = importlib.import_module("hermes_hux_runtime.runtime")
tool_policy = importlib.import_module("hermes_hux_runtime.tool_policy")

SUBJECT = "usr_" + "a" * 64
SLOT = "slot-2"
KEY = bytes(range(32))
ARGS = {"path": "notes.txt", "content": "CANARY-secret-value"}


class FakeIds:
    """Stable ids without touching a key file in middleware-only tests."""

    def conversation(self, raw: str) -> str:
        if not raw:
            raise context_ids.ContextUnavailable("missing")
        return "conv_" + "c" * 32

    def session(self, raw: str) -> str:
        if not raw:
            raise context_ids.ContextUnavailable("missing")
        return "ses_" + "b" * 32

    def project(self, raw: str) -> str:
        if not raw:
            raise context_ids.ContextUnavailable("missing")
        return "prj_" + "a" * 32

    def run(self, raw: str) -> str:
        if not raw:
            raise context_ids.ContextUnavailable("missing")
        return "run_" + "d" * 32


def middleware_metadata() -> dict[str, str]:
    return {"session_id": "session-1", "turn_id": "turn-1", "tool_call_id": "call-1"}


class FakeClient:
    """Capture only the mandatory context bootstrap request."""

    def __init__(self):
        self.posts = []

    def post(self, path, body, idempotency_key=None):
        self.posts.append((path, body, idempotency_key))
        return SimpleNamespace(body={})


def make_runtime(enforce: bool = True):
    return runtime_module.Runtime(FakeClient(), FakeIds(), enforce)


def test_context_id_matches_frozen_webui_contract():
    raw = "session:one+two"
    message = "\0".join(
        ("hux.context.id.v1", "conversation", SLOT, SUBJECT, raw)
    ).encode()
    expected = "conv_" + hmac.new(KEY, message, hashlib.sha256).hexdigest()[:32]
    assert context_ids.derive_hux_id(KEY, "conv", "conversation", SLOT, SUBJECT, raw) == expected
    run = context_ids.derive_hux_id(KEY, "run", "run", SLOT, SUBJECT, "turn-9")
    assert run.startswith("run_") and len(run) == 36


@pytest.mark.parametrize(
    ("prefix", "purpose", "slot", "subject", "raw", "key"),
    [
        ("bad", "conversation", SLOT, SUBJECT, "s", KEY),
        ("conv", "conversation", "tenant-2", SUBJECT, "s", KEY),
        ("conv", "conversation", SLOT, "usr_short", "s", KEY),
        ("conv", "conversation", SLOT, SUBJECT, "slash/not-allowed", KEY),
        ("conv", "conversation", SLOT, SUBJECT, "s", b"short"),
    ],
)
def test_context_id_rejects_noncanonical_inputs(prefix, purpose, slot, subject, raw, key):
    with pytest.raises(context_ids.ContextUnavailable):
        context_ids.derive_hux_id(key, prefix, purpose, slot, subject, raw)


def test_context_key_requires_owner_only_regular_single_link_file(tmp_path: Path):
    path = tmp_path / "context-key"
    path.write_bytes(KEY)
    path.chmod(0o600)
    ids = context_ids.ContextIds(path, SLOT, SUBJECT)
    assert ids.conversation("session-1").startswith("conv_")
    assert ids.session("session-1").startswith("ses_")
    assert ids.project("profile:default").startswith("prj_")
    assert ids.run("turn-1").startswith("run_")
    path.chmod(0o640)
    with pytest.raises(context_ids.ContextUnavailable, match="unsafe"):
        context_ids.ContextIds(path, SLOT, SUBJECT)
    path.chmod(0o600)
    linked = tmp_path / "linked"
    os.link(path, linked)
    with pytest.raises(context_ids.ContextUnavailable, match="unsafe"):
        context_ids.ContextIds(path, SLOT, SUBJECT)
    linked.unlink()
    path.write_bytes(b"short")
    with pytest.raises(context_ids.ContextUnavailable):
        context_ids.ContextIds(path, SLOT, SUBJECT)


def test_context_key_rejects_missing_and_symlink(tmp_path: Path):
    with pytest.raises(context_ids.ContextUnavailable, match="unavailable"):
        context_ids.ContextIds(tmp_path / "missing", SLOT, SUBJECT)
    real = tmp_path / "real"
    real.write_bytes(KEY)
    real.chmod(0o600)
    link = tmp_path / "link"
    link.symlink_to(real)
    with pytest.raises(context_ids.ContextUnavailable):
        context_ids.ContextIds(link, SLOT, SUBJECT)
    with pytest.raises(context_ids.ContextUnavailable, match="identity"):
        context_ids.ContextIds(real, SLOT, "usr_" + "A" * 64)


def test_context_key_handles_short_read(monkeypatch, tmp_path: Path):
    path = tmp_path / "key"
    path.write_bytes(KEY)
    path.chmod(0o600)
    monkeypatch.setattr(context_ids.os, "read", lambda *_args: b"short")
    with pytest.raises(context_ids.ContextUnavailable, match="invalid"):
        context_ids.ContextIds(path, SLOT, SUBJECT)


@pytest.mark.parametrize(
    ("name", "capability", "risk", "external"),
    [
        ("read_file", "read_files", "low", False),
        ("write_file", "write_files", "medium", False),
        ("memory", "memory_write", "medium", False),
        ("delegate_task", "delegate", "high", False),
        ("terminal", "shell", "high", True),
        ("web_search", "network", "high", True),
        ("send_email", "send_message", "high", True),
        ("kubectl_apply", "deploy", "high", True),
        ("image_generate", "external_side_effect", "high", True),
        ("artifact_create", "artifact_write", "medium", False),
        ("memory_forget", "memory_write", "medium", False),
        ("subagent_spawn", "delegate", "high", False),
        ("edit_document", "write_files", "medium", False),
        ("new_plugin_tool", "external_side_effect", "high", True),
        (None, "external_side_effect", "high", True),
    ],
)
def test_tool_mapping_is_conservative(name, capability, risk, external):
    assert tool_policy.classify(name) == tool_policy.ToolPolicy(capability, risk, external)


def test_default_off_registers_nothing(monkeypatch):
    hooks, middleware = {}, {}
    ctx = SimpleNamespace(
        register_hook=lambda name, callback: hooks.setdefault(name, callback),
        register_middleware=lambda name, callback: middleware.setdefault(name, callback),
    )
    monkeypatch.delenv("HUX_RUNTIME_ENABLED", raising=False)
    PACKAGE.register(ctx)
    assert not hooks and not middleware


@pytest.mark.parametrize(("enforce", "executes"), [("1", False), ("0", True)])
def test_misconfigured_opt_in_closes_only_enforcement(monkeypatch, enforce, executes):
    hooks, middleware = {}, {}
    ctx = SimpleNamespace(
        register_hook=lambda name, callback: hooks.setdefault(name, callback),
        register_middleware=lambda name, callback: middleware.setdefault(name, callback),
    )
    monkeypatch.setenv("HUX_RUNTIME_ENABLED", "true")
    monkeypatch.setenv("HUX_TOOL_ENFORCEMENT", enforce)
    monkeypatch.setattr(PACKAGE.Runtime, "from_env", classmethod(lambda cls: (_ for _ in ()).throw(ValueError())))
    PACKAGE.register(ctx)
    called = []
    result = middleware["tool_execution"](args=ARGS, next_call=lambda value: called.append(value) or "ok")
    assert bool(called) is executes
    assert (result == "ok") is executes
    assert set(hooks) == {"on_session_end"}


def test_from_env_reads_subject_only_from_file(tmp_path: Path):
    subject = tmp_path / "subject"
    subject.write_text(SUBJECT + "\n")
    subject.chmod(0o400)
    worker = tmp_path / "worker"
    worker.write_text("worker-key")
    worker.chmod(0o400)
    context = tmp_path / "context"
    context.write_bytes(KEY)
    context.chmod(0o600)
    env = {
        "HUX_TENANT_SLOT": SLOT,
        "HUX_SUBJECT_FILE": str(subject),
        "HUX_WORKER_KEY_FILE": str(worker),
        "HUX_CONTEXT_KEY_FILE": str(context),
        "HUX_SUBJECT": "usr_" + "b" * 64,
        "HUX_TOOL_ENFORCEMENT": "yes",
        "HUX_TIMEOUT_SECONDS": "0.2",
    }
    instance = runtime_module.Runtime.from_env(env)
    assert instance.client.identity == {
        "tenant_slot": SLOT,
        "subject": SUBJECT,
        "surface": "worker",
        "trust": "worker",
    }
    assert instance.enforce is True
    assert instance.project_source == "profile:default"


def test_from_env_rejects_missing_file_contract():
    with pytest.raises(context_ids.ContextUnavailable):
        runtime_module.Runtime.from_env({"HUX_TENANT_SLOT": SLOT})


@pytest.mark.parametrize("project_source", [None, "", "x" * 201])
def test_runtime_rejects_malformed_project_source(project_source):
    with pytest.raises(context_ids.ContextUnavailable, match="project source"):
        runtime_module.Runtime(FakeClient(), FakeIds(), False, project_source)


@pytest.mark.parametrize(
    "metadata",
    [
        {"session_id": None, "turn_id": "turn-1"},
        {"session_id": "session-1", "turn_id": None},
    ],
)
def test_scope_requires_host_owned_string_ids(metadata):
    with pytest.raises(context_ids.ContextUnavailable, match="session context"):
        make_runtime(False)._scope(metadata)


def test_denial_never_executes_and_exposes_no_arguments(monkeypatch):
    instance = make_runtime()
    calls, telemetry = [], []
    monkeypatch.setattr(runtime_module, "emit", lambda *args, **kwargs: telemetry.append((args[1:], kwargs)))

    def deny(_client, run_id, conversation_id, name, arguments, capability, **policy):
        calls.append((run_id, conversation_id, name, arguments, capability, policy))
        return SimpleNamespace(proceed=False, reason="approval_required", approval_id="apr_safe")

    monkeypatch.setattr(runtime_module, "before_tool", deny)
    downstream = []
    result = instance.tool_execution(
        tool_name="write_file",
        args=ARGS,
        next_call=lambda args: downstream.append(args),
        **middleware_metadata(),
    )
    assert not downstream and calls[0][3] is ARGS
    assert calls[0][-1] == {"external": False, "risk": "medium"}
    assert json.loads(result) == {
        "approval_id": "apr_safe",
        "error": "HUX policy did not release this tool call",
        "reason": "approval_required",
        "schema": "hux.tool_block.v1",
        "status": "blocked",
    }
    assert ARGS["content"] not in json.dumps(telemetry)


def test_bootstrap_uses_exact_ids_and_replays_in_process(monkeypatch):
    instance = make_runtime(False)
    monkeypatch.setattr(runtime_module, "emit", lambda *args, **kwargs: {})
    monkeypatch.setattr(runtime_module, "after_tool", lambda *args, **kwargs: None)
    monkeypatch.setattr(runtime_module, "record_spend", lambda *args, **kwargs: None)
    for call_id in ("call-1", "call-2"):
        instance.tool_execution(
            tool_name="read_file",
            args={},
            next_call=lambda _args: "ok",
            session_id="session-1",
            turn_id="turn-1",
            tool_call_id=call_id,
        )
    assert instance.client.posts == [
        (
            "/hux/v1/context/bootstrap",
            {
                "raw_session_id": "session-1",
                "project_source": "profile:default",
                "session_id": "ses_" + "b" * 32,
                "conversation_id": "conv_" + "c" * 32,
                "project_id": "prj_" + "a" * 32,
            },
            "context:conv_" + "c" * 32,
        )
    ]


def test_bootstrap_failure_blocks_enforcement_and_opens_telemetry(monkeypatch):
    monkeypatch.setattr(runtime_module, "emit", lambda *args, **kwargs: {})
    for enforce in (True, False):
        instance = make_runtime(enforce)
        instance.client.post = lambda *args, **kwargs: (_ for _ in ()).throw(OSError("offline"))
        called = []
        result = instance.tool_execution(
            tool_name="read_file",
            args={},
            next_call=lambda value: called.append(value) or "ok",
            **middleware_metadata(),
        )
        assert bool(called) is (not enforce)
        assert (result == "ok") is (not enforce)


def test_enforcement_failure_is_closed_but_telemetry_failure_is_open(monkeypatch):
    for enforce in (True, False):
        instance = make_runtime(enforce)
        monkeypatch.setattr(runtime_module, "canonical_argument_hash", lambda *_: (_ for _ in ()).throw(TypeError()))
        called = []
        result = instance.tool_execution(
            tool_name="write_file", args=ARGS, next_call=lambda args: called.append(args) or "done", **middleware_metadata()
        )
        assert bool(called) is (not enforce)
        assert (result == "done") is (not enforce)


def test_released_call_executes_exact_object_once_and_reports_status_only(monkeypatch):
    instance = make_runtime()
    events, after, spend = [], [], []
    monkeypatch.setattr(runtime_module, "emit", lambda *args, **kwargs: events.append((args[1:], kwargs)))
    monkeypatch.setattr(
        runtime_module,
        "before_tool",
        lambda *args, **kwargs: SimpleNamespace(proceed=True, reason="released", approval_id="apr_1"),
    )
    monkeypatch.setattr(runtime_module, "after_tool", lambda *args, **kwargs: after.append((args[1:], kwargs)))
    monkeypatch.setattr(runtime_module, "record_spend", lambda *args, **kwargs: spend.append((args, kwargs)))
    downstream = []

    def execute(value):
        downstream.append(value)
        return {"ok": True, "payload": "RAW-RESULT-SECRET"}

    result = instance.tool_execution(
        tool_name="write_file", args=ARGS, next_call=execute, **middleware_metadata()
    )
    assert downstream == [ARGS] and downstream[0] is ARGS
    assert result["payload"] == "RAW-RESULT-SECRET"
    assert after[0][0][3] is True and after[0][0][4] > 0
    assert spend[0][1]["tool_calls"] == 1
    assert ARGS["content"] not in json.dumps(events)
    assert "RAW-RESULT-SECRET" not in json.dumps(after)


def test_downstream_exception_is_reported_then_re_raised(monkeypatch):
    instance = make_runtime()
    after = []
    monkeypatch.setattr(runtime_module, "emit", lambda *args, **kwargs: {})
    monkeypatch.setattr(
        runtime_module,
        "before_tool",
        lambda *args, **kwargs: SimpleNamespace(proceed=True, reason="released", approval_id="apr_1"),
    )
    monkeypatch.setattr(runtime_module, "after_tool", lambda *args, **kwargs: after.append(args))
    monkeypatch.setattr(runtime_module, "record_spend", lambda *args, **kwargs: None)

    def explode(_args):
        raise RuntimeError("RAW-EXECUTION-SECRET")

    with pytest.raises(RuntimeError, match="RAW-EXECUTION-SECRET"):
        instance.tool_execution(tool_name="terminal", args=ARGS, next_call=explode, **middleware_metadata())
    assert after[0][4] is False and after[0][5] == 0


def test_after_telemetry_failure_never_changes_tool_result(monkeypatch):
    instance = make_runtime(False)
    monkeypatch.setattr(runtime_module, "emit", lambda *args, **kwargs: {})
    monkeypatch.setattr(runtime_module, "after_tool", lambda *args, **kwargs: (_ for _ in ()).throw(RuntimeError()))
    assert instance.tool_execution(
        tool_name="read_file", args={}, next_call=lambda _args: "result", **middleware_metadata()
    ) == "result"


def test_run_start_is_single_fire_under_concurrency(monkeypatch):
    instance = make_runtime(False)
    events = []
    monkeypatch.setattr(runtime_module, "emit", lambda *args, **kwargs: events.append(args[2]))
    call = runtime_module.CallContext(
        "conv_" + "c" * 32,
        "run_" + "d" * 32,
        "call",
        "session-1",
        "ses_" + "b" * 32,
        "prj_" + "a" * 32,
    )
    threads = [threading.Thread(target=instance._start_once, args=(call,)) for _ in range(20)]
    for thread in threads:
        thread.start()
    for thread in threads:
        thread.join()
    assert events == ["run.started"]


@pytest.mark.parametrize(
    ("result", "ok"),
    [
        ("plain", True),
        ("{not json", True),
        ({"status": "ok"}, True),
        ({"error": "no"}, False),
        (json.dumps({"status": "cancelled"}), False),
        ({"status": "failed"}, False),
    ],
)
def test_result_status_detection(result, ok):
    assert runtime_module._result_ok(result) is ok


def test_result_size_is_bounded_and_tolerates_bad_values(monkeypatch):
    class Bad:
        def __str__(self):
            raise ValueError

    assert runtime_module._result_size(b"abc") == 3
    assert runtime_module._result_size("é") == 2
    assert runtime_module._result_size(Bad()) == 0
    monkeypatch.setattr(runtime_module, "MAX_REPORTED_BYTES", 2)
    assert runtime_module._result_size(b"xxx") == 2


@pytest.mark.parametrize(
    ("completed", "interrupted", "kind"),
    [(True, False, "run.completed"), (True, True, "run.failed"), (False, False, "run.failed")],
)
def test_session_end_never_claims_unverified_cancellation(monkeypatch, completed, interrupted, kind):
    instance = make_runtime(False)
    events = []
    monkeypatch.setattr(runtime_module, "emit", lambda *args, **kwargs: events.append((args, kwargs)))
    instance.session_end(completed=completed, interrupted=interrupted, **middleware_metadata())
    assert events[0][0][2] == kind
    assert "cancel" not in events[0][0][3].lower() or "without a verified cancellation" in events[0][0][3]


def test_session_end_and_after_ignore_telemetry_errors(monkeypatch):
    instance = make_runtime(False)
    monkeypatch.setattr(runtime_module, "emit", lambda *args, **kwargs: (_ for _ in ()).throw(RuntimeError()))
    instance.session_end(session_id="bad session", turn_id="bad turn")
    monkeypatch.setattr(runtime_module, "after_tool", lambda *args, **kwargs: (_ for _ in ()).throw(RuntimeError()))
    call = runtime_module.CallContext(
        "conv_" + "c" * 32,
        "run_" + "d" * 32,
        "call",
        "session-1",
        "ses_" + "b" * 32,
        "prj_" + "a" * 32,
    )
    instance._after(call, "read_file", "sha256:" + "e" * 64, True, 1, 0.0)


def test_plugin_files_stay_small_and_stop_bridge_is_honest():
    for path in PLUGIN.glob("*.py"):
        assert len(path.read_text().splitlines()) <= 500
    source = (PLUGIN / "runtime.py").read_text()
    notes = (PLUGIN / "NOTES.md").read_text()
    assert "on_stop(" not in source
    assert "process_registry_empty" in notes
    assert "HUX_SUBJECT" not in source.replace("HUX_SUBJECT_FILE", "")