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

Domain: ``calculus``. Computes the indefinite or definite integral
of a symbolic expression with respect to a named variable. Wraps
``sp.integrate`` with a soft timeout so pathological inputs cannot
hang the caller indefinitely.

Why this matters
----------------
Ticket #000030 §2.2 Phase 3. Integration is the natural counterpart
to the Phase-2 derivative: same JSON-shaped input contract, same
``algebra-symbolic@v1``-style canonical bytes for the
closed-form-result case, with one extra rule for unevaluated
results.

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

    {"f": "<expression-text>", "x": "<variable-name>",
     "limits"?: [<lower>, <upper>],
     "timeout_seconds"?: <float>}

- ``f`` — the integrand (parsed via :func:`sympy.sympify`).
- ``x`` — the variable name.
- ``limits`` — optional ``[a, b]``. Each endpoint is sympified, so
  ``"oo"`` (positive infinity), ``"-oo"``, ``"pi"``, etc. all work.
- ``timeout_seconds`` — optional, default :data:`DEFAULT_TIMEOUT`.

Output canonical bytes
----------------------
Two paths:

1. **Closed-form result:**
   ``sp.srepr(sp.expand(integrate_result))`` UTF-8 — same recipe as
   ``algebra-symbolic@v1``. Means an integral that evaluates is a
   first-class composition target.
2. **Unevaluated result** (SymPy returns an :class:`sympy.Integral`
   or :class:`sympy.NonElementaryIntegral` object, OR the timeout
   fires): bytes are ``b"unevaluated:" + sp.srepr(<Integral>)``.
   The ``unevaluated:`` prefix lets callers distinguish "no closed
   form" from "input invalid" without parsing the canonical form.

Timeout discipline
------------------
``sp.integrate`` can hang on pathological inputs. We launch the
work on a :class:`ThreadPoolExecutor` and ``future.result(timeout=…)``;
on :class:`TimeoutError`, we synthesize an unevaluated sentinel
(``Integral`` / ``Integral(<expr>, (<x>, a, b))``). A timed-out
thread continues running in the background — Python threads cannot
be killed cleanly. Operators tuning a hot path should size the
worker pool accordingly. The default 30s timeout is conservative;
per-call override via ``timeout_seconds`` in the input JSON.

Round-trip property
-------------------
Closed-form bytes round-trip through the algebra-symbolic@v1
recipe. Unevaluated-sentinel bytes do NOT round-trip through this
π* (the input contract is JSON, not srepr); their round-trip is at
the equivalence-class layer (same input ⇒ same sentinel).

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

Source: ticket #000030 §2.2 Phase 3 — 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  # seconds


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


def _is_unevaluated(result) -> bool:
    """True when ``sp.integrate`` returned an unevaluated form.

    Both :class:`sympy.Integral` and the more specific
    :class:`sympy.integrals.integrals.NonElementaryIntegral` count
    — the latter is a subclass.
    """
    return isinstance(result, sp.Integral)


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


def _build_unevaluated(f_expr, x_sym, limits_pair):
    """Construct an unevaluated Integral object representing the
    requested integration. Used both for the timeout-fallback path
    and as the test target for sentinel parsing."""
    if limits_pair is None:
        return sp.Integral(f_expr, x_sym)
    a, b = limits_pair
    return sp.Integral(f_expr, (x_sym, a, b))


@dataclass
class CalculusIntegralV1:
    name: str = "calculus-integral"
    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-integral@v1 expects bytes; got "
                f"{type(raw).__name__}"
            )
        try:
            text = raw.decode("utf-8", errors="surrogatepass")
        except UnicodeDecodeError as exc:  # pragma: no cover — defensive
            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-integral@v1 input must be JSON object: {exc}"
            ) from exc
        if not isinstance(obj, dict):
            raise PiStarError(
                f"calculus-integral@v1 input must be JSON object; got {type(obj).__name__}"
            )

        for k in ("f", "x"):
            if k not in obj:
                raise PiStarError(f"calculus-integral@v1 missing field {k!r}")
            if not isinstance(obj[k], str) or not obj[k].strip():
                raise PiStarError(
                    f"calculus-integral@v1 field {k!r} must be a non-empty string"
                )

        timeout = obj.get("timeout_seconds", DEFAULT_TIMEOUT)
        if not isinstance(timeout, (int, float)) or isinstance(timeout, bool) or timeout <= 0:
            raise PiStarError(
                "calculus-integral@v1 'timeout_seconds' must be a positive number"
            )

        x_name = obj["x"].strip()
        x_sym = sp.Symbol(x_name)
        try:
            f_expr = sp.sympify(obj["f"], locals={x_name: x_sym})
        except (sp.SympifyError, SyntaxError, TypeError) as exc:
            raise PiStarError(
                f"calculus-integral@v1 cannot parse f={obj['f']!r}: {exc}"
            ) from exc
        if not isinstance(f_expr, sp.Expr):
            raise PiStarError(
                f"calculus-integral@v1 rejects boolean/relational f={obj['f']!r}"
            )

        limits_pair = None
        if "limits" in obj:
            lim = obj["limits"]
            if not (isinstance(lim, list) and len(lim) == 2):
                raise PiStarError(
                    "calculus-integral@v1 'limits' must be a 2-element list [a, b]"
                )
            try:
                a = sp.sympify(lim[0])
                b = sp.sympify(lim[1])
            except (sp.SympifyError, SyntaxError, TypeError) as exc:
                raise PiStarError(
                    f"calculus-integral@v1 cannot parse limits {lim!r}: {exc}"
                ) from exc
            limits_pair = (a, b)

        # Run sp.integrate in a worker thread with a timeout. If the
        # work hangs, the thread leaks (Python threads aren't killable);
        # the canonical result for the timeout case is an unevaluated
        # Integral sentinel so the equivalence-class behavior stays
        # deterministic.
        def _work():
            if limits_pair is None:
                return sp.integrate(f_expr, x_sym)
            return sp.integrate(f_expr, (x_sym, limits_pair[0], limits_pair[1]))

        with concurrent.futures.ThreadPoolExecutor(max_workers=1) as pool:
            future = pool.submit(_work)
            try:
                result = future.result(timeout=float(timeout))
            except concurrent.futures.TimeoutError:
                # Synthesize the unevaluated sentinel — same content
                # we'd emit if SymPy itself returned an unevaluated
                # Integral. Avoids two distinct "no closed form" paths
                # producing different bytes.
                return _sentinel(_build_unevaluated(f_expr, x_sym, limits_pair))
            except (TypeError, ValueError, sp.SympifyError) as exc:
                raise PiStarError(
                    f"calculus-integral@v1 integrate failed: {exc}"
                ) from exc

        if _is_unevaluated(result):
            return _sentinel(result)

        try:
            canonical = sp.expand(result)
        except (TypeError, ValueError) as exc:
            raise PiStarError(
                f"calculus-integral@v1 expand failed: {exc}"
            ) from exc
        return sp.srepr(canonical).encode("utf-8")


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