"""``calculus-limit@v1`` π* — symbolic-limit canonicalizer.

Domain: ``calculus``. Computes ``lim_{x → point} f(x)``, optionally
one-sided. Wraps ``sp.limit`` with a soft timeout so pathological
inputs cannot hang the caller indefinitely.

Why this matters
----------------
Ticket #000030 §2.2 Phase 4. Limit is the third calculus primitive
after derivative (Phase 2) and integral (Phase 3). Same JSON-shaped
input contract; canonicalizes through ``algebra-symbolic@v1`` for
finite-result cases. Infinite-result cases pin a stable spelling
(``+oo`` / ``-oo`` / ``zoo``) so two LLM-spellings of "infinity"
collapse.

Input shape (JSON, UTF-8)
-------------------------
::

    {"f": "<expression-text>",
     "x": "<variable-name>",
     "point": "<sympify-able-point>",
     "dir"?: "+" | "-" | "+-",
     "timeout_seconds"?: <float>}

- ``f`` — the function (parsed via :func:`sympy.sympify`).
- ``x`` — the variable name.
- ``point`` — the limit point. Sympify-friendly: ``"0"``, ``"oo"``,
  ``"-oo"``, ``"pi"`` all work.
- ``dir`` — optional one-sided limit direction. Default is the
  two-sided limit (``"+-"`` in SymPy parlance, or omitted from the
  call). ``"+"`` for right-side only, ``"-"`` for left-side.
- ``timeout_seconds`` — optional, default :data:`DEFAULT_TIMEOUT`.

Output canonical bytes
----------------------
Three paths:

1. **Finite result** — ``sp.srepr(sp.expand(limit_result))`` UTF-8.
   Same recipe as ``algebra-symbolic@v1`` so numerical limits
   (``lim sin(x)/x → 1``) compose with the rest of the math
   substrate.
2. **Infinite result** — pinned spelling: ``b"+oo"`` for ``+∞``,
   ``b"-oo"`` for ``-∞``, ``b"zoo"`` for SymPy's ``ComplexInfinity``.
   These don't go through ``sp.expand`` since ``Infinity`` is not
   an algebraic object.
3. **No closed form / timeout** — bytes are
   ``b"unevaluated:" + sp.srepr(<Limit>)``. Mirrors
   ``calculus-integral@v1``'s sentinel discipline.

Optional dependency
-------------------
Same as ``algebra-symbolic@v1``: SymPy via the ``[math]`` extras.

Source: ticket #000030 §2.2 Phase 4 — landed 2026-05-09.
"""

from __future__ import annotations

import concurrent.futures
import json
from dataclasses import dataclass

from arborist.pi_star.protocol import PiStarError
from arborist.pi_star.registry import register


try:
    import sympy as sp  # type: ignore[import-not-found]
except ImportError:  # pragma: no cover — optional extra
    sp = None  # type: ignore[assignment]


DEFAULT_TIMEOUT = 30.0


_SYMPY_REQUIRED_MSG = (
    "calculus-limit@v1 requires sympy; install with "
    "`pip install 'arborist[math]'`"
)


_VALID_DIRS = ("+", "-", "+-")


def _serialize_infinity(result) -> bytes | None:
    """Map SymPy infinities to pinned UTF-8 byte spellings, or None
    if the result isn't an infinity case."""
    if result is sp.oo or result == sp.oo:
        return b"+oo"
    if result is -sp.oo or result == -sp.oo:
        return b"-oo"
    if result is sp.zoo or result == sp.zoo:
        return b"zoo"
    return None


def _is_unevaluated(result) -> bool:
    """True when ``sp.limit`` returned an unevaluated form."""
    return isinstance(result, sp.Limit)


def _sentinel(unevaluated) -> bytes:
    return b"unevaluated:" + sp.srepr(unevaluated).encode("utf-8")


def _build_unevaluated(f_expr, x_sym, point_expr, dir_):
    if dir_ in (None, "+-"):
        return sp.Limit(f_expr, x_sym, point_expr)
    return sp.Limit(f_expr, x_sym, point_expr, dir_)


@dataclass
class CalculusLimitV1:
    name: str = "calculus-limit"
    version: str = "v1"
    domain: str = "calculus"

    def canonicalize(self, raw: bytes) -> bytes:
        if sp is None:
            raise PiStarError(_SYMPY_REQUIRED_MSG)
        if not isinstance(raw, (bytes, bytearray)):
            raise PiStarError(
                "calculus-limit@v1 expects bytes; got "
                f"{type(raw).__name__}"
            )
        try:
            text = raw.decode("utf-8", errors="surrogatepass")
        except UnicodeDecodeError as exc:  # pragma: no cover
            raise PiStarError(f"input not valid UTF-8: {exc}") from exc

        try:
            obj = json.loads(text)
        except json.JSONDecodeError as exc:
            raise PiStarError(
                f"calculus-limit@v1 input is not valid JSON: {exc}"
            ) from exc
        if not isinstance(obj, dict):
            raise PiStarError(
                "calculus-limit@v1 input must be a JSON object"
            )
        for required in ("f", "x", "point"):
            if required not in obj:
                raise PiStarError(
                    f"calculus-limit@v1 missing required field {required!r}"
                )

        f_text = obj["f"]
        x_name = obj["x"]
        point_text = obj["point"]
        dir_ = obj.get("dir")
        if dir_ is not None and dir_ not in _VALID_DIRS:
            raise PiStarError(
                f"calculus-limit@v1 dir must be one of {_VALID_DIRS}; "
                f"got {dir_!r}"
            )
        timeout_s = float(obj.get("timeout_seconds", DEFAULT_TIMEOUT))

        if not isinstance(f_text, str) or not f_text.strip():
            raise PiStarError("calculus-limit@v1 f must be a non-empty string")
        if not isinstance(x_name, str) or not x_name.strip():
            raise PiStarError("calculus-limit@v1 x must be a non-empty string")

        try:
            x_sym = sp.Symbol(x_name)
            f_expr = sp.sympify(f_text, locals={x_name: x_sym})
            # ``point`` may be "oo", "-oo", "pi", "0", etc.
            point_expr = sp.sympify(point_text)
        except (sp.SympifyError, SyntaxError, TypeError) as exc:
            raise PiStarError(
                f"calculus-limit@v1 cannot parse: {exc}"
            ) from exc

        # SymPy's limit() takes the dir as a string; default
        # two-sided. Use a thread + timeout for the same reason
        # calculus-integral does.
        def _compute():
            if dir_ in (None, "+-"):
                return sp.limit(f_expr, x_sym, point_expr)
            return sp.limit(f_expr, x_sym, point_expr, dir_)

        try:
            with concurrent.futures.ThreadPoolExecutor(max_workers=1) as ex:
                future = ex.submit(_compute)
                try:
                    result = future.result(timeout=timeout_s)
                except concurrent.futures.TimeoutError:
                    return _sentinel(
                        _build_unevaluated(f_expr, x_sym, point_expr, dir_)
                    )
        except (sp.SympifyError, ValueError, TypeError) as exc:
            raise PiStarError(
                f"calculus-limit@v1 evaluation failed: {exc}"
            ) from exc

        # Infinity / complex-infinity: pinned spelling.
        inf_bytes = _serialize_infinity(result)
        if inf_bytes is not None:
            return inf_bytes

        # SymPy may also return an unevaluated Limit object on weird
        # inputs even without timeout firing — fall back to sentinel.
        if _is_unevaluated(result):
            return _sentinel(result)

        # Finite result — canonicalize through algebra-symbolic@v1's
        # recipe (sp.expand + sp.srepr).
        try:
            expanded = sp.expand(result)
        except Exception as exc:  # pragma: no cover — sympy edge cases
            raise PiStarError(
                f"calculus-limit@v1 cannot expand result: {exc}"
            ) from exc
        return sp.srepr(expanded).encode("utf-8")


if sp is not None:
    register(CalculusLimitV1())
