"""The governed ToolUniverse client — the ONLY module that imports ``tooluniverse``.

ADR-0001 §7, enforced in code rather than by convention:

- **Allow-list by construction.** ToolUniverse is built with ``keep_default_tools=False`` and
  ``load_workspace=False``, then ``load_tools(include_tools=<the endpoints' set>)``, and startup
  is REFUSED unless the loaded set equals that set exactly. With these arguments 1.5.6 loads the
  named tools and nothing else: no special tools, no tool finder, no agentic tools, no remote MCP
  servers (measured 2026-10-20, with outbound sockets blocked, so loading opens no connection).
- **No workspace.** ``load_workspace=False`` is not optional. With a workspace directory present
  (``TOOLUNIVERSE_HOME``, or ``./.tooluniverse``), 1.5.6's constructor writes a default
  ``profile.yaml`` into it and applies it, which loads ALL 2,691 tools before ``load_tools`` runs;
  it also reads that workspace's ``.env`` and ``~/.tooluniverse/.env`` into the environment, which
  could bring a model key in after the check below. Found by this repo's own tests on 2026-10-20,
  when an earlier draft pointed ``TOOLUNIVERSE_HOME`` at an empty temp dir.
- **No model provider keys.** ToolUniverse's agentic tools and tool finder can reach OpenAI, Azure
  OpenAI, Gemini, OpenRouter or a vLLM server when keys are set. None of those tools can load
  here, and startup still refuses if any key is present: the Space makes no model call
  (ADR-0001 §2). ``HF_TOKEN`` is allowed — the machine gate needs it — and in 1.5.6 only
  Hugging Face inference and embedding tools read it, none of which is on the allow-list.
- **Exactly one cache.** ToolUniverse's own result cache is disabled through its environment
  switches and every call passes ``use_cache=False``; :mod:`src.cache` is the only cache, because
  it is the one that keeps ``retrieved_at`` honest.
- **Pinned.** The installed version must equal ``config/tools_allowlist.yaml``'s
  ``tooluniverse_version``. A bump is its own lane (doctor, re-recorded fixtures, schema tests,
  security scan) and never changes an endpoint's name or payload shape.

:class:`ToolClient` is what the endpoints call. It returns a :class:`SourceResult` or raises a
:class:`SourceError` whose ``reason`` is one of the three payload source-error reasons, so a
timeout, a limit or a drifted response is always structured, never a guess.
"""

from __future__ import annotations

import ast
import concurrent.futures
import inspect
import os
import re
import threading
import time
from collections.abc import Callable, Mapping
from dataclasses import dataclass
from typing import Any

from src import config, schemas
from src.cache import ResponseCache
from src.ratelimit import RateLimiters

# Every variable through which ToolUniverse 1.5.6 (or a library it can load) reaches a model.
LLM_KEY_VARS = (
    "OPENAI_API_KEY",
    "OPENAI_BASE_URL",
    "AZURE_OPENAI_API_KEY",
    "AZURE_OPENAI_ENDPOINT",
    "GEMINI_API_KEY",
    "GOOGLE_API_KEY",
    "OPENROUTER_API_KEY",
    "ANTHROPIC_API_KEY",
    "VLLM_SERVER_URL",
)

# Tool types that must never load, whatever the allow-list says: LLM-backed agents, the tool
# finders (they can call a model or download an embedding index), remote MCP auto-loaders and
# web-search backends. A tool whose config names `required_api_keys` is key-gated and refused too.
FORBIDDEN_TOOL_TYPES = frozenset(
    {
        "AgenticTool",
        "SmolAgentTool",
        "MCPAutoLoaderTool",
        "MCPClientTool",
        "ToolFinderEmbedding",
        "ToolFinderLLM",
        "ToolFinderKeyword",
        "ToolFinderJev",
        "WebSearchTool",
        "WebAPIDocumentationSearchTool",
    }
)

# ToolUniverse's own switches: its result cache off. The workspace and profile variables are
# REMOVED rather than set, because any workspace widens what loads (see the module docstring).
_TU_ENV = {
    "TOOLUNIVERSE_CACHE_ENABLED": "false",
    "TOOLUNIVERSE_CACHE_PERSIST": "false",
    "TOOLUNIVERSE_QUIET": "1",
}
_TU_ENV_REMOVED = ("TOOLUNIVERSE_PROFILE", "TOOLUNIVERSE_HOME")

_HOST = re.compile(r"https?://([A-Za-z0-9.-]+)")
_build_lock = threading.Lock()


class StartupRefused(RuntimeError):
    """The Space must not start: a key is set, the version drifted, or the tool set is wrong."""


class SourceError(Exception):
    """A structured failure from one source call (``reason`` ∈ payload SOURCE_ERROR_REASONS)."""

    def __init__(self, source: str, reason: str, detail: str):
        super().__init__(f"{source}: {reason}: {detail}")
        self.source = source
        self.reason = reason
        self.detail = detail

    def as_payload(self) -> dict[str, str]:
        return {"source": self.source, "reason": self.reason, "detail": self.detail}


@dataclass(frozen=True)
class SourceResult:
    tool: str
    source: str
    args: dict[str, Any]
    data: Any
    retrieved_at: float  # epoch seconds when the SOURCE answered
    cache: str  # "hit" | "miss"
    tooluniverse_version: str


def assert_no_llm_keys(environ: Mapping[str, str] | None = None) -> None:
    env = os.environ if environ is None else environ
    present = [var for var in LLM_KEY_VARS if (env.get(var) or "").strip()]
    if present:
        raise StartupRefused(
            f"Model-provider settings are present ({', '.join(present)}). This Space makes no "
            "model call and must run without them (ADR-0001 §2); remove them from the Space's "
            "secrets and variables."
        )


def configure_tooluniverse_env() -> None:
    """Set ToolUniverse's environment before an instance is built (idempotent)."""
    for var, value in _TU_ENV.items():
        os.environ[var] = value
    for var in _TU_ENV_REMOVED:
        os.environ.pop(var, None)


def installed_tooluniverse_version() -> str:
    import tooluniverse

    return str(getattr(tooluniverse, "__version__", ""))


def build_tooluniverse(include: frozenset[str] | None = None):
    """A ToolUniverse holding exactly ``include`` (default: the endpoints' set), or refuse."""
    include = config.endpoint_tool_names() if include is None else include
    if not include <= config.allowed_tool_names():
        raise StartupRefused(
            f"not on the allow-list: {sorted(include - config.allowed_tool_names())}"
        )
    assert_no_llm_keys()
    configure_tooluniverse_env()
    version = installed_tooluniverse_version()
    if version != config.pinned_tooluniverse_version():
        raise StartupRefused(
            f"ToolUniverse {version!r} is installed; config pins "
            f"{config.pinned_tooluniverse_version()!r}. A version bump is its own reviewed lane."
        )
    from tooluniverse import ToolUniverse

    tu = ToolUniverse(keep_default_tools=False, load_workspace=False)
    if tu.all_tool_dict:
        raise StartupRefused(
            f"ToolUniverse loaded {len(tu.all_tool_dict)} tools before the allow-list was applied"
        )
    tu.load_tools(include_tools=sorted(include), quiet=True)
    assert_no_llm_keys()  # again: nothing during construction may have brought one in
    loaded = frozenset(tu.all_tool_dict)
    if loaded != include:
        raise StartupRefused(
            f"loaded tool set differs from the allow-listed set: extra {sorted(loaded - include)}, "
            f"missing {sorted(include - loaded)}"
        )
    unsafe = sorted(
        name
        for name, cfg in tu.all_tool_dict.items()
        if cfg.get("type") in FORBIDDEN_TOOL_TYPES or cfg.get("required_api_keys")
    )
    if unsafe:
        raise StartupRefused(f"agentic, finder, remote-MCP, web-search or key-gated: {unsafe}")
    cache_manager = getattr(tu, "cache_manager", None)
    if cache_manager is not None and getattr(cache_manager, "enabled", False):
        raise StartupRefused("ToolUniverse's own result cache is enabled; it must be off")
    return tu


def tool_hosts(tu, tool: str) -> set[str]:
    """Every host a loaded tool can reach: URLs in its config plus its class's code.

    Read offline from the installed package: the tool's config dict, the source of its class and
    ToolUniverse base classes, and any module-level URL constant those classes name (HGNC keeps
    its base URL in one). So the egress statement in config/sources.yaml can be checked without a
    network call (tests/test_sources_hosts.py).
    """
    from tooluniverse.tool_registry import get_tool_class_lazy

    cfg = tu.all_tool_dict[tool]
    hosts = set(_HOST.findall(repr(cfg)))
    # The lazy lookup imports the class's module (offline); the plain registry is empty until a
    # tool instance has been created, which would make this check silently find nothing.
    cls = get_tool_class_lazy(cfg["type"])
    if cls is None:
        raise StartupRefused(f"{tool}: tool type {cfg['type']!r} has no class to inspect")
    for klass in inspect.getmro(cls):
        if not klass.__module__.startswith("tooluniverse"):
            continue
        try:
            class_source = inspect.getsource(klass)
            module_source = inspect.getsource(inspect.getmodule(klass))
        except (OSError, TypeError):
            continue
        hosts.update(_HOST.findall(class_source))
        for name, value in _module_url_constants(module_source).items():
            if re.search(rf"\b{re.escape(name)}\b", class_source):
                hosts.update(_HOST.findall(value))
    return hosts


def _module_url_constants(module_source: str) -> dict[str, str]:
    """Top-level ``NAME = "https://…"`` assignments in a module."""
    found: dict[str, str] = {}
    for node in ast.parse(module_source).body:
        if isinstance(node, ast.Assign) and isinstance(node.value, ast.Constant):
            value = node.value.value
            if isinstance(value, str) and _HOST.search(value):
                for target in node.targets:
                    if isinstance(target, ast.Name):
                        found[target.id] = value
    return found


def _unwrap(tool: str, source: str, raw: Any) -> Any:
    """ToolUniverse's envelope → the source's data, or a structured error."""
    if raw is None:
        raise SourceError(source, "source_unavailable", f"{tool} returned nothing")
    if isinstance(raw, dict):
        status = raw.get("status")
        if status == "success" and "data" in raw:
            return raw["data"]
        if status == "error" or ("error" in raw and raw.get("error")):
            detail = str(raw.get("error") or raw.get("message") or "error")[:200]
            raise SourceError(source, "source_unavailable", f"{tool}: {detail}")
    if isinstance(raw, str) and raw.lower().startswith("error"):
        raise SourceError(source, "source_unavailable", f"{tool}: {raw[:200]}")
    return raw


class ToolClient:
    """Cache → rate limit → call with a deadline → schema check → cache.

    ``runner(tool, args)`` performs the actual call. By default it is ToolUniverse's
    ``run_one_function`` on a lazily built, allow-listed instance; tests inject a fake runner
    with synthetic or recorded responses, so the suite never reaches the network.
    """

    def __init__(
        self,
        *,
        runner: Callable[[str, dict[str, Any]], Any] | None = None,
        cache: ResponseCache | None = None,
        limiters: RateLimiters | None = None,
        clock: Callable[[], float] = time.time,
        tooluniverse_version: str | None = None,
        tooluniverse: Any = None,
    ):
        self._runner = runner
        # gradio_ui passes an instance built at import, so a Space that must not start (a model
        # key, a version drift, a widened tool set) fails at startup, not on its first call.
        self._tu = tooluniverse
        self.cache = cache if cache is not None else ResponseCache()
        self.limiters = limiters if limiters is not None else RateLimiters()
        self._clock = clock
        self._version = tooluniverse_version
        self._pool = concurrent.futures.ThreadPoolExecutor(
            max_workers=8, thread_name_prefix="tu-call"
        )

    @property
    def tooluniverse_version(self) -> str:
        if self._version is None:
            self._version = installed_tooluniverse_version()
        return self._version

    def _default_runner(self, tool: str, args: dict[str, Any]) -> Any:
        with _build_lock:
            if self._tu is None:
                self._tu = build_tooluniverse()
        return self._tu.run_one_function({"name": tool, "arguments": args}, use_cache=False)

    def call(
        self,
        tool: str,
        args: dict[str, Any],
        *,
        max_wait: float = 10.0,
        deadline: float | None = None,
    ) -> SourceResult:
        """``deadline`` can shorten the source's ``timeout_seconds`` for this call, never extend it."""
        if tool not in config.endpoint_tool_names():
            # A programming error, not a user input: endpoints only ever name loaded tools.
            raise ValueError(f"{tool!r} is not an endpoint tool")
        source = config.source_of(tool)
        cfg = config.source(source)

        hit = self.cache.get(source, tool, args)
        if hit is not None:
            data, retrieved_at = hit
            return SourceResult(
                tool, source, dict(args), data, retrieved_at, "hit", self.tooluniverse_version
            )

        if not self.limiters.acquire(source, max_wait):
            raise SourceError(
                source, "rate_limited", f"{cfg['name']} limit of {cfg['rate_per_second']}/s reached"
            )

        timeout = cfg["timeout_seconds"]
        if deadline is not None:
            timeout = max(0.0, min(timeout, deadline))
        runner = self._runner or self._default_runner
        future = self._pool.submit(runner, tool, dict(args))
        # `wait`, not `result(timeout=…)`: since Python 3.11 concurrent.futures.TimeoutError IS the
        # builtin TimeoutError, so a source's own read timeout raised inside the call would be
        # indistinguishable from our deadline passing (caught by a test on 2026-10-20).
        done, _ = concurrent.futures.wait([future], timeout=timeout)
        if not done:
            future.cancel()
            raise SourceError(
                source,
                "source_unavailable",
                f"{cfg['name']} did not answer within {round(timeout, 1):g} s",
            )
        try:
            raw = future.result()
        except SourceError:
            raise
        except Exception as e:  # any failure inside a source call is structured, never raised
            raise SourceError(source, "source_unavailable", f"{tool}: {type(e).__name__}") from e
        retrieved_at = self._clock()

        data = _unwrap(tool, source, raw)
        try:
            schemas.validate(tool, data)
        except schemas.SchemaDrift as drift:
            raise SourceError(source, "upstream_schema_drift", drift.detail) from drift

        self.cache.put(source, tool, args, data, retrieved_at)
        return SourceResult(
            tool, source, dict(args), data, retrieved_at, "miss", self.tooluniverse_version
        )