"""Event-loop lag gauge — the per-pod saturation signal for autoscaling load tests.

The web pod runs a single uvicorn worker (one asyncio event loop, see
``scripts/run_web.sh``). When that loop is saturated it wakes *late* from
``asyncio.sleep``: a task that asked to sleep 100 ms actually resumes at
100 + X ms. That overshoot X is the cleanest provider-independent measure of how
close the pod is to its real ceiling — unlike CPU%, which is measured against the
2-core limit and reads ~50% at true saturation, and unlike turn latency, which is
dominated by external STT/LLM/TTS round-trips.

An autoscaling load test ramps concurrent calls against one pod and reads this
gauge to find the knee — the ``active_calls`` count where p95 lag climbs — from
which the KEDA callsPerPod (K) value is derived.

The gauge is a module global updated by one background task and read (peek) off
``GET /api/v1/health/active-calls``. Single event loop, so no lock is needed.
"""

import asyncio
import math

# Single in-process gauge — exactly the unit (one event loop) we're sizing. A
# window of recent lag samples is enough for a p95; no metrics library.
_INTERVAL = 0.1  # seconds between probes
_WINDOW = 600  # ~60s of samples at 0.1s cadence
_samples: list[float] = []
# Strong ref to the monitor task: asyncio keeps only a weak reference, so
# without this the task can be garbage-collected mid-run and the gauge dies.
_task: "asyncio.Task | None" = None


async def _monitor() -> None:
    loop = asyncio.get_running_loop()
    while True:
        t0 = loop.time()
        await asyncio.sleep(_INTERVAL)
        lag_ms = (loop.time() - t0 - _INTERVAL) * 1000
        if lag_ms < 0:  # clock jitter; floor at 0
            lag_ms = 0.0
        _samples.append(lag_ms)
        if len(_samples) > _WINDOW:
            del _samples[: len(_samples) - _WINDOW]


def start() -> asyncio.Task:
    """Start the lag monitor on the running loop. Idempotent; call from lifespan."""
    global _task
    if _task is None or _task.done():
        _task = asyncio.get_running_loop().create_task(_monitor())
    return _task


async def stop() -> None:
    """Cancel the monitor on shutdown so the loop doesn't warn about a pending
    task being destroyed. Safe to call when never started."""
    global _task
    task, _task = _task, None
    if task is not None:
        task.cancel()
        try:
            await task
        except asyncio.CancelledError:
            pass


def _percentile(values: list[float], pct: float) -> float:
    if not values:
        return 0.0
    ordered = sorted(values)
    # nearest-rank: rank = ceil(pct/100 * n), 1-indexed -> subtract 1. Using
    # int()/floor here would land one rank high (p95 of 20 samples = the max).
    idx = min(len(ordered) - 1, max(0, math.ceil(pct / 100 * len(ordered)) - 1))
    return ordered[idx]


def stats() -> dict[str, float]:
    """Current lag over the recent window, in milliseconds. Read-only peek."""
    snapshot = list(_samples)
    return {
        "p95_ms": round(_percentile(snapshot, 95), 2),
        "max_ms": round(max(snapshot), 2) if snapshot else 0.0,
        "samples": len(snapshot),
    }


def demo() -> None:
    """Self-check: lag is ~0 when the loop is idle and spikes under a busy task."""

    async def _run():
        start()
        await asyncio.sleep(0.5)
        idle = stats()
        assert idle["p95_ms"] < 5, f"idle lag too high: {idle}"

        # Block the loop synchronously for ~150ms across several probe intervals.
        deadline = asyncio.get_running_loop().time() + 0.4
        while asyncio.get_running_loop().time() < deadline:
            end = asyncio.get_running_loop().time() + 0.15
            while asyncio.get_running_loop().time() < end:
                pass  # busy-spin, starves the loop
            await asyncio.sleep(0)
        busy = stats()
        assert busy["max_ms"] > 50, f"busy lag not detected: {busy}"
        print(f"OK  idle={idle}  busy={busy}")

    asyncio.run(_run())


if __name__ == "__main__":
    demo()