"""Adversarial tests for the authoritative runtime dispatch boundary."""

from __future__ import annotations

import asyncio
from pathlib import Path

import pytest

from trace2mcp.codegen import generate_project, generate_server_py
from trace2mcp.models import PolicyViolationError, Trace, TraceEvent, Workflow, WorkflowNode
from trace2mcp.policy import ToolContract
from trace2mcp.runtime import (
    SchemaValidationError,
    ToolNotFoundError,
    ToolRegistry,
    WorkflowExecutionError,
    execute_workflow,
    invoke_capability,
)


def reviewed(name: str, side_effect: str = "read", **kwargs: object) -> ToolContract:
    return ToolContract(name, side_effect=side_effect, reviewed=True, **kwargs)


def strict_registry(
    handlers: dict[str, object], contracts: dict[str, ToolContract]
) -> ToolRegistry:
    return ToolRegistry(
        handlers,  # type: ignore[arg-type]
        contracts=contracts,
        require_contracts=True,
        require_reviewed_contracts=True,
    )


def test_direct_destructive_operation_is_denied_before_handler() -> None:
    calls = 0

    def destroy() -> None:
        nonlocal calls
        calls += 1

    workflow = Workflow("deny-delete", {}, [WorkflowNode("delete", "files.delete", {})])
    registry = strict_registry(
        {"files.delete": destroy},
        {"files.delete": reviewed("files.delete", "delete")},
    )

    with pytest.raises(PolicyViolationError, match="denied"):
        asyncio.run(execute_workflow(workflow, registry))
    assert calls == 0


def test_reviewed_contract_defeats_trace_metadata_confusion() -> None:
    calls = 0

    def send() -> str:
        nonlocal calls
        calls += 1
        return "sent"

    workflow = Workflow(
        "metadata-confusion",
        {},
        [WorkflowNode("send", "gmail.send", {}, side_effect="read")],
    )
    registry = strict_registry(
        {"gmail.send": send},
        {"gmail.send": reviewed("gmail.send", "external_side_effect")},
    )

    with pytest.raises(PolicyViolationError, match="approval"):
        asyncio.run(execute_workflow(workflow, registry))
    assert calls == 0

    result = asyncio.run(execute_workflow(workflow, registry, approvals={"send"}))
    assert result.output == "sent"
    assert calls == 1


def test_alias_and_registry_proxy_cannot_bypass_policy() -> None:
    dangerous_calls = 0

    def dangerous() -> None:
        nonlocal dangerous_calls
        dangerous_calls += 1

    alias_workflow = Workflow(
        "alias",
        {},
        [WorkflowNode("alias", "safe-looking-alias", {}, side_effect="read")],
    )
    alias_registry = strict_registry(
        {"safe-looking-alias": dangerous},
        {"safe-looking-alias": reviewed("safe-looking-alias", "delete")},
    )
    with pytest.raises(PolicyViolationError, match="denied"):
        asyncio.run(execute_workflow(alias_workflow, alias_registry))
    assert dangerous_calls == 0

    registry = ToolRegistry(
        {"dangerous.delete": dangerous},
        contracts={"dangerous.delete": reviewed("dangerous.delete", "delete")},
    )
    mediated = registry.get("dangerous.delete")
    with pytest.raises(PolicyViolationError, match="active Trace2MCP runtime"):
        asyncio.run(mediated())
    assert dangerous_calls == 0


def test_unknown_and_unreviewed_capabilities_fail_closed() -> None:
    unknown = Workflow("unknown", {}, [WorkflowNode("missing", "missing.tool", {})])
    with pytest.raises(ToolNotFoundError):
        asyncio.run(execute_workflow(unknown, ToolRegistry()))

    calls = 0

    def unreviewed() -> None:
        nonlocal calls
        calls += 1

    workflow = Workflow("unreviewed", {}, [WorkflowNode("tool", "tool", {})])
    registry = ToolRegistry(
        {"tool": unreviewed},
        contracts={"tool": ToolContract("tool", side_effect="read", reviewed=False)},
        require_contracts=True,
        require_reviewed_contracts=True,
    )
    with pytest.raises(PolicyViolationError, match="reviewed contract"):
        asyncio.run(execute_workflow(workflow, registry))
    assert calls == 0


def test_approval_is_necessary_but_does_not_override_other_denials() -> None:
    calls = 0

    def send() -> str:
        nonlocal calls
        calls += 1
        return "sent"

    workflow = Workflow("approval", {}, [WorkflowNode("send", "send", {})])
    registry = strict_registry(
        {"send": send},
        {"send": reviewed("send", "external_side_effect")},
    )
    with pytest.raises(PolicyViolationError, match="approval"):
        asyncio.run(execute_workflow(workflow, registry))
    assert calls == 0
    assert asyncio.run(execute_workflow(workflow, registry, approvals={"send"})).output == "sent"

    denied_registry = strict_registry(
        {"send": send},
        {"send": reviewed("send", "delete")},
    )
    with pytest.raises(PolicyViolationError, match="denied"):
        asyncio.run(execute_workflow(workflow, denied_registry, approvals={"send"}))
    assert calls == 1


def test_nested_capability_escalation_returns_to_same_gate() -> None:
    dangerous_calls = 0
    safe_calls = 0
    registry: ToolRegistry

    async def parent() -> str:
        dynamic_name = ".".join(("dangerous", "delete"))
        return await registry.invoke(dynamic_name)

    def dangerous() -> str:
        nonlocal dangerous_calls
        dangerous_calls += 1
        return "destroyed"

    def safe_child(value: int) -> int:
        nonlocal safe_calls
        safe_calls += 1
        return value + 1

    registry = strict_registry(
        {
            "allowed.read": parent,
            "dangerous.delete": dangerous,
            "safe.child": safe_child,
        },
        {
            "allowed.read": reviewed("allowed.read"),
            "dangerous.delete": reviewed("dangerous.delete", "delete"),
            "safe.child": reviewed("safe.child"),
        },
    )
    workflow = Workflow("nested-deny", {}, [WorkflowNode("parent", "allowed.read", {})])
    with pytest.raises(WorkflowExecutionError) as captured:
        asyncio.run(execute_workflow(workflow, registry))
    assert isinstance(captured.value.cause, PolicyViolationError)
    assert dangerous_calls == 0

    async def safe_parent() -> int:
        return await invoke_capability("safe.child", value=4)

    registry.register("allowed.read", safe_parent)
    assert asyncio.run(execute_workflow(workflow, registry)).output == 5
    assert safe_calls == 1


def test_generated_server_has_no_direct_handler_route(tmp_path: Path) -> None:
    workflow = Workflow("generated", {}, [WorkflowNode("read", "records.read", {})])
    server = generate_server_py(workflow)
    assert "result = await execute_workflow(" in server
    assert "WORKFLOW," in server
    assert "REGISTRY," in server
    assert "runtime_guard=RUNTIME_GUARD" in server
    assert "HANDLERS[" not in server
    assert "require_contracts=True" in server
    assert "require_reviewed_contracts=True" in server

    generated = generate_project(workflow, tmp_path)
    assert "contracts" in generated
    contracts = (tmp_path / "contracts.py").read_text(encoding="utf-8")
    assert "reviewed=False" in contracts
    assert "Set reviewed=True only after checking" in contracts


def test_registry_swap_between_preflight_and_invoke_fails_closed() -> None:
    old_calls = 0
    new_calls = 0
    registry: ToolRegistry

    def old_target() -> str:
        nonlocal old_calls
        old_calls += 1
        return "old"

    def new_target() -> str:
        nonlocal new_calls
        new_calls += 1
        return "new"

    def mutate_registry() -> str:
        registry.register("target", new_target)
        return "mutated"

    workflow = Workflow(
        "toctou",
        {},
        [
            WorkflowNode("mutate", "mutate", {}),
            WorkflowNode("target", "target", {}, depends_on=("mutate",)),
        ],
    )
    registry = strict_registry(
        {"mutate": mutate_registry, "target": old_target},
        {"mutate": reviewed("mutate"), "target": reviewed("target")},
    )
    with pytest.raises(WorkflowExecutionError) as captured:
        asyncio.run(execute_workflow(workflow, registry))
    assert isinstance(captured.value.cause, PolicyViolationError)
    assert "changed after authorization" in str(captured.value.cause)
    assert old_calls == 0
    assert new_calls == 0


def test_contract_schema_is_enforced_before_and_after_handler() -> None:
    calls = 0

    def handler(value: object) -> object:
        nonlocal calls
        calls += 1
        return value

    contract = reviewed(
        "typed",
        input_schema={
            "type": "object",
            "properties": {"value": {"type": "integer"}},
            "required": ["value"],
            "additionalProperties": False,
        },
        output_schema={"type": "string"},
    )
    registry = strict_registry({"typed": handler}, {"typed": contract})

    invalid_input = Workflow(
        "invalid-input", {}, [WorkflowNode("typed", "typed", {"value": "not-int"})]
    )
    with pytest.raises(WorkflowExecutionError) as captured:
        asyncio.run(execute_workflow(invalid_input, registry))
    assert isinstance(captured.value.cause, SchemaValidationError)
    assert calls == 0

    invalid_output = Workflow("invalid-output", {}, [WorkflowNode("typed", "typed", {"value": 1})])
    with pytest.raises(WorkflowExecutionError) as captured:
        asyncio.run(execute_workflow(invalid_output, registry))
    assert isinstance(captured.value.cause, SchemaValidationError)
    assert calls == 1


def test_handler_exception_stops_dependents_and_approval_does_not_leak() -> None:
    send_calls = 0

    def fail() -> None:
        raise RuntimeError("boom")

    def send() -> None:
        nonlocal send_calls
        send_calls += 1

    workflow = Workflow(
        "exception",
        {},
        [
            WorkflowNode("fail", "fail", {}),
            WorkflowNode("send", "send", {}, depends_on=("fail",)),
        ],
    )
    registry = strict_registry(
        {"fail": fail, "send": send},
        {"fail": reviewed("fail"), "send": reviewed("send", "external_side_effect")},
    )
    with pytest.raises(WorkflowExecutionError):
        asyncio.run(execute_workflow(workflow, registry, approvals={"send"}))
    assert send_calls == 0

    send_only = Workflow("send-only", {}, [WorkflowNode("send", "send", {})])
    with pytest.raises(PolicyViolationError, match="approval"):
        asyncio.run(execute_workflow(send_only, registry))
    assert send_calls == 0


def test_trace_compiled_server_still_uses_authoritative_runtime() -> None:
    workflow = Workflow(
        "compiled",
        {},
        [WorkflowNode("read", "read", {}, side_effect="read")],
    )
    source = generate_server_py(workflow)
    assert source.count("execute_workflow(") == 1
    assert "result = await execute_workflow" in source

    trace = Trace(
        inputs={},
        events=[TraceEvent("read", "read", {}, "ok", side_effect="read")],
    )
    assert trace.events[0].tool == "read"