"""Deterministic virtual-time execution for timing-coupled async tests.

Voice tests drive the agent through fake STT/TTS/VAD/audio components in *real* wall-clock
time: the agent's endpointing, interruption timeouts and audio play-out clock all schedule
against the event loop and measure durations with `time.time()` / `time.perf_counter()`.
That makes the suite sensitive to scheduler jitter -- a shared event loop, or a slow/contended
CI runner, perturbs the relative timing and the count assertions race.

A test marked `@pytest.mark.virtual_time` runs instead on a :class:`_VirtualTimeLoop`, whose
clock only advances when the loop goes idle (it jumps straight to the next scheduled timer). No
real time passes, so a 30-second scripted conversation completes in microseconds and, crucially,
*deterministically* -- the order of timer firings is fixed by their scheduled times, not by who
won a CPU slice.

The loop only virtualizes `loop.time()` (what `asyncio.sleep`/`call_later` read). The
agent and fakes also read wall time directly, so :func:`_virtual_wall_clock` additionally points
`time.time()` / `time.perf_counter()` at the running loop's virtual clock *for the duration of
the marked test only* -- a scoped patch, so no global state leaks between tests.

Wiring (conftest.py re-exports these so pytest discovers them):
- :func:`event_loop_policy` is pytest-asyncio's override point for the loop a test runs on; it
  returns the virtual-time policy for marked tests and the default policy otherwise (so non-marked
  tests behave exactly as before).
- :func:`_virtual_wall_clock` is autouse; a no-op unless the test carries the marker. It patches
  the `time` module for inline reads and, for `Field(default_factory=time.time)` references
  captured at import (which a module patch cannot reach), rewrites the validators in place via
  :func:`_patch_model_factories` so `created_at` timestamps also run on virtual time.
"""

from __future__ import annotations

import asyncio
import contextlib
import selectors
import time
from collections.abc import Callable, Iterator, Mapping
from typing import Any
from unittest import mock

import pytest

# A fixed, realistic wall-clock base so patched `time.time()` still looks like a real epoch.
# Only differences are ever asserted under virtual time, so the exact value is irrelevant.
_WALL_CLOCK_EPOCH = 1_700_000_000.0

# The real clocks, captured at import (before any patch is installed) so the fallbacks below can
# hand real wall time to callers that aren't on the virtual loop, and as the signature for
# discovering which captured `default_factory` references to redirect.
_REAL_TIME = time.time
_REAL_PERF = time.perf_counter

# Smallest separation between two wall-clock reads at the same loop instant. The clock does not
# move between reads within one tick, but synchronously-created items still need distinct, ordered
# `created_at` values (ChatContext reconciliation orders by it), so reads are nudged to stay
# strictly increasing (see _VirtualClock.read). Tiny enough that accumulation across a test stays
# far below any timing assertion's tolerance.
_TICK_EPSILON = 1e-6


def _virtual_loop() -> _VirtualTimeLoop | None:
    """The virtual-time loop driving the current call, or None.

    Returns None when there is no running loop (e.g. a background thread such as the OpenTelemetry
    metrics exporter) or the running loop is an ordinary real one. The clock patch lives on the
    `time` module process-wide for the duration of a virtual-time test, so it is reachable from
    any thread; only callers actually on the virtual loop should see virtual time -- everyone else
    must get the real wall clock, otherwise a stray read freezes at the epoch (which, for the
    metrics exporter, stamps batches with a 2023 timestamp the gateway then rejects).
    """
    try:
        loop = asyncio.get_running_loop()
    except RuntimeError:
        return None
    return loop if isinstance(loop, _VirtualTimeLoop) else None


def _now() -> float:
    # Off the virtual loop (no running loop, or a real one) fall back to the real clock, so stray
    # reads from other threads see true wall time.
    loop = _virtual_loop()
    return _WALL_CLOCK_EPOCH + loop.monotonic_read() if loop is not None else _REAL_TIME()


def _perf() -> float:
    loop = _virtual_loop()
    return loop.monotonic_read() if loop is not None else _REAL_PERF()


# Maps a captured real clock to its virtual replacement. A model field built with
# `Field(default_factory=time.time)` captures the bare `time.time` at import; patching the
# `time` module later cannot reach that captured reference, so :func:`_patch_model_factories`
# rewrites it in the validator instead.
_VIRTUAL_FOR = {_REAL_TIME: _now, _REAL_PERF: _perf}


def is_virtual_time(node: pytest.Item) -> bool:
    # `--real-time` disables virtual time globally; the `real_time` marker opts out a single test.
    if node.config.getoption("--real-time", default=False):
        return False
    return (
        node.get_closest_marker("virtual_time") is not None
        and node.get_closest_marker("real_time") is None
    )


def _to_plain(obj: Any) -> Any:
    """Materialize a (possibly lazy `MockCoreSchema`) core schema into a fresh plain dict tree.

    pydantic exposes `__pydantic_core_schema__` as a `Mapping` proxy whose `deepcopy` stays
    a proxy, but `SchemaValidator` needs real dicts. Rebuilding every mapping/list also gives an
    independent copy we can edit without touching the class's schema.
    """
    if isinstance(obj, Mapping):
        return {k: _to_plain(v) for k, v in obj.items()}
    if isinstance(obj, list):
        return [_to_plain(v) for v in obj]
    return obj


def _model_node(schema: dict[str, Any]) -> dict[str, Any]:
    """Descend `definitions` / validator wrappers to the `model` node holding the fields."""
    node = schema
    while node.get("type") != "model":
        node = node["schema"]
    return node


# Built once, lazily: `model -> (patched_validator, original_validator)`. The patched validator
# is invariant across tests (its `default_factory` is `_now`, which resolves the running loop
# at call time), so we pay the schema-edit + validator construction a single time and then only
# swap references per test.
_PATCHED: dict[type, tuple[Any, Any]] = {}


def _build_patch_cache() -> None:
    """Construct, once, a virtual-clock validator for every model with a captured real clock."""
    if _PATCHED:
        return
    from pydantic_core import SchemaValidator

    for model, fields in _discover_factory_patches().items():
        schema = _to_plain(model.__pydantic_core_schema__)
        node = _model_node(schema)
        field_schemas = node["schema"]["fields"]
        for name, factory in fields.items():
            field_schemas[name]["schema"]["default_factory"] = factory
        # _use_prebuilt=False forces a build from our edited schema instead of reusing the
        # validator cached on the class.
        patched = SchemaValidator(schema, node.get("config"), _use_prebuilt=False)
        _PATCHED[model] = (patched, model.__pydantic_validator__)


@contextlib.contextmanager
def _patched_model_clocks() -> Iterator[None]:
    """Swap every captured-clock model onto its virtual-clock validator, then restore.

    All such models move together so `created_at`-ordered structures (e.g. `ChatContext`,
    which inserts by `created_at`) stay consistent -- a partial swap would interleave virtual
    and real timestamps and scramble item order. The per-test cost is just the reference swaps.
    """
    _build_patch_cache()
    try:
        for model, (patched, _original) in _PATCHED.items():
            model.__pydantic_validator__ = patched  # type: ignore[attr-defined]
        yield
    finally:
        for model, (_patched, original) in _PATCHED.items():
            model.__pydantic_validator__ = original  # type: ignore[attr-defined]


def _discover_factory_patches() -> dict[type, dict[str, Callable[[], Any]]]:
    """Find every pydantic model field whose `default_factory` is a real clock, grouped by model.

    Self-maintaining: any new event/chat-context model with a `default_factory=time.time` field
    is picked up automatically, no registry to update. Imported lazily so importing this module
    stays cheap and free of livekit import-order constraints.
    """
    from pydantic import BaseModel

    from livekit.agents.inference import interruption
    from livekit.agents.llm import chat_context
    from livekit.agents.voice import events

    patches: dict[type, dict[str, Callable[[], Any]]] = {}
    for module in (events, chat_context, interruption):
        for obj in vars(module).values():
            if not (isinstance(obj, type) and issubclass(obj, BaseModel)):
                continue
            fields = {
                name: _VIRTUAL_FOR[info.default_factory]  # type: ignore[index]
                for name, info in obj.model_fields.items()
                if info.default_factory in _VIRTUAL_FOR
            }
            if fields:
                patches[obj] = fields
    return patches


# Sub-tick ordering granularity. The autojumping clock jumps straight onto the next scheduled
# time, so two timers due at the same virtual instant fire in asyncio's heap order (which reflects
# heap structure, not scheduling order) -- and that can invert the order real time would produce
# for causally-ordered-but-simultaneous events. We give each timer a strictly increasing
# sub-microsecond nudge so same-instant timers fire in *scheduling* (FIFO) order, matching the
# causal order real wall-clock produces. The nudge needs a finer clock resolution than the default
# 1e-6 to be representable; accumulation across a test stays far below any timing tolerance.
_TIE_BREAK = 1e-9
_FINE_RESOLUTION = 1e-12


class _VirtualClock:
    """The loop's virtual clock.

    `time()` only moves when `advance()` is called (by the selector), and is what asyncio reads for
    scheduling -- it must stay raw so the sub-nanosecond timer tie-break (see _VirtualTimeLoop) is
    not swamped. `read()` is the wall-clock view: the same virtual time, but nudged to stay
    strictly increasing within a tick so synchronously-created items get distinct, ordered
    `created_at`. The high-water mark lives here (per loop), so it resets naturally with each test's
    fresh loop -- no global state.
    """

    __slots__ = ("_time", "_read")

    def __init__(self) -> None:
        self._time = 0.0
        self._read = 0.0

    def time(self) -> float:
        return self._time

    def advance(self, seconds: float) -> None:
        if seconds > 0:
            self._time += seconds

    def read(self) -> float:
        self._read = self._time if self._time > self._read else self._read + _TICK_EPSILON
        return self._read


class _AutojumpSelector(selectors.BaseSelector):
    """A selector that never really blocks: instead of waiting `timeout` seconds for the next
    timer, it jumps the virtual clock there and returns immediately.

    It delegates fd registration to a real selector and still polls it non-blockingly, so the
    loop's own self-pipe (and thus `call_soon_threadsafe` wakeups from executor threads) keep
    working. A `None` timeout means the loop has nothing scheduled and is genuinely waiting on
    such a wakeup, so we block on the real selector -- matching stock asyncio.
    """

    def __init__(self, clock: _VirtualClock) -> None:
        self._clock = clock
        self._real = selectors.DefaultSelector()

    def register(self, fileobj: Any, events: int, data: Any = None) -> selectors.SelectorKey:
        return self._real.register(fileobj, events, data)

    def unregister(self, fileobj: Any) -> selectors.SelectorKey:
        return self._real.unregister(fileobj)

    def modify(self, fileobj: Any, events: int, data: Any = None) -> selectors.SelectorKey:
        return self._real.modify(fileobj, events, data)

    def select(self, timeout: float | None = None) -> list[tuple[selectors.SelectorKey, int]]:
        # Hot path: only the loop's self-pipe is registered (no real I/O). A real poll could only
        # ever report that self-pipe, and a `call_soon_threadsafe` wakeup's callback is already
        # queued in the loop's `_ready` (so the next iteration runs it regardless) -- skipping the
        # poll therefore can't drop work. It just avoids an epoll syscall on every loop iteration,
        # which is what keeps virtual time as fast as a fully virtual selector.
        if len(self._real.get_map()) <= 1:
            if timeout is None:
                return self._real.select(None)  # genuinely idle: block for an off-loop wakeup
            if timeout > 0:
                self._clock.advance(timeout)
            return []
        # Real I/O is registered: poll without blocking and jump only if nothing is ready.
        ready = self._real.select(0)
        if ready or timeout == 0:
            return ready
        if timeout is None:
            return self._real.select(None)
        self._clock.advance(timeout)
        return []

    def get_map(self) -> Any:
        return self._real.get_map()

    def close(self) -> None:
        self._real.close()


class _VirtualTimeLoop(asyncio.SelectorEventLoop):
    """A `SelectorEventLoop` whose clock only advances when the loop would otherwise sleep.

    The selector jumps the virtual clock straight to the next scheduled callback (see
    :class:`_AutojumpSelector`), so timing-coupled tests run in ~0 wall time and deterministically.
    Same-instant timers are nudged into scheduling (FIFO) order so they fire in the causal order
    real wall time would produce.
    """

    def __init__(self) -> None:
        self._virtual_clock = _VirtualClock()
        super().__init__(selector=_AutojumpSelector(self._virtual_clock))
        # Resolve finely enough that the sub-nanosecond tie-break is representable and only the
        # exact-same-instant timer fires per jump.
        self._clock_resolution = _FINE_RESOLUTION
        self._tie_seq = 0

    def time(self) -> float:
        return self._virtual_clock.time()

    def monotonic_read(self) -> float:
        """Strictly-increasing wall-clock read off this loop's clock (see _VirtualClock.read)."""
        return self._virtual_clock.read()

    def call_at(self, when: float, callback: Any, *args: Any, **kwargs: Any) -> Any:
        self._tie_seq += 1
        return super().call_at(when + self._tie_seq * _TIE_BREAK, callback, *args, **kwargs)


class _VirtualTimePolicy(asyncio.DefaultEventLoopPolicy):
    def new_event_loop(self) -> asyncio.AbstractEventLoop:
        return _VirtualTimeLoop()


@pytest.fixture
def event_loop_policy(request: pytest.FixtureRequest) -> asyncio.AbstractEventLoopPolicy:
    """Loop policy pytest-asyncio builds the test loop from.

    Virtual-time tests get a :class:`_VirtualTimeLoop` (autojumping virtual clock); everything else
    gets the default policy, i.e. exactly what pytest-asyncio would use without this override, so
    non-marked tests are unaffected. `real_time` opts a test out even under a module-level mark.
    """
    if is_virtual_time(request.node):
        return _VirtualTimePolicy()
    return asyncio.DefaultEventLoopPolicy()


@pytest.fixture(autouse=True)
def _virtual_wall_clock(request: pytest.FixtureRequest) -> Iterator[None]:
    """Point wall-clock reads at the running loop's virtual clock for a `virtual_time` test.

    No-op unless the test is marked `virtual_time`. Two kinds of reads are covered:

    - inline `time.time()` / `time.perf_counter()` -- redirected by patching the `time`
      module (these re-read the attribute at call time, so a scoped patch reaches them);
    - `default_factory=time.time` captured on event/chat-context models at import -- redirected
      by :func:`_patched_model_clocks`, since a module patch cannot reach a captured reference.

    Both resolve the loop at call time, so reads inside the loop see virtual time while any stray
    read from outside it (no running loop) falls back to a stable constant. Scoped to the test.
    """
    if not is_virtual_time(request.node):
        yield
        return

    with (
        mock.patch.object(time, "time", _now),
        mock.patch.object(time, "perf_counter", _perf),
        _patched_model_clocks(),
    ):
        yield


def add_realtime_option(parser: pytest.Parser) -> None:
    group = parser.getgroup("virtual time")
    group.addoption(
        "--real-time",
        action="store_true",
        default=False,
        help="Disable virtual-time patching entirely; virtual_time tests run on the real clock.",
    )


def register_marker(config: pytest.Config) -> None:
    config.addinivalue_line(
        "markers",
        "virtual_time: run the test on a deterministic autojumping virtual-time event loop.",
    )
    config.addinivalue_line(
        "markers",
        "real_time: opt a test out of a module-level virtual_time mark (runs on the real loop).",
    )