"""Per-session turn lease — serializes the [load history -> run -> flush] region.

Busy guards are keyed by ROUTING KEY but the transcript is owned by SESSION_ID, and
``switch_session()`` makes key->id many-to-one (/resume from a second chat, CLI-continuity,
delegation pinning, topic tip-walks), so two keys could interleave flushes on one transcript
(``user;user`` wedge). The lease serializes per RESOLVED session_id: acquired right before the
transcript load, released in the dispatch layer's ``finally``; identity-checked release; a
timed-out waiter fails CLOSED (:class:`TurnLeaseTimeoutError`); only idle entries evict. Limits:
CLI-continuity processes are outside this lock; mid-turn rotation alias is closed by ``rebind``.
"""

import asyncio
import logging
import time
from typing import Dict, Optional

logger = logging.getLogger(__name__)

# Cap on tracked leases. Idle entries evict oldest-first; live leases never do, so a burst of
# distinct sessions may transiently exceed the cap rather than break serialization.
DEFAULT_MAX_LEASES = 512
# Fallback wait (seconds) when the caller passes no positive timeout (bridged via
# HERMES_TURN_LEASE_TIMEOUT — lease contention is not agent inactivity). Fail-closed but short:
# never pin a sequential platform updater for minutes.
DEFAULT_LEASE_WAIT = 5.0


def _holder_desc(holder: Optional["TurnLeaseToken"]) -> tuple:
    return (holder.owner_key, holder.generation) if holder else ("?", "?")


class TurnLeaseTimeoutError(TimeoutError):
    """Lease held for the full wait budget; fail-closed: caller must not enter the turn region."""

    def __init__(self, session_id: str, *, owner_key: str, generation: int, wait_seconds: float):
        self.session_id, self.owner_key = session_id, owner_key
        self.generation, self.wait_seconds = generation, wait_seconds
        super().__init__(f"turn lease wait timed out after {wait_seconds:.0f}s on session "
                         f"{session_id} for routing key {owner_key} (gen {generation})")


class TurnLeaseToken:
    """Held-lease handle from :meth:`SessionTurnLeaseRegistry.acquire`; ``released`` makes
    release idempotent."""

    __slots__ = ("session_id", "owner_key", "generation", "released", "lease")

    def __init__(self, session_id: str, owner_key: str, generation: int, lease: "_SessionLease") -> None:
        self.session_id, self.owner_key, self.generation = session_id, owner_key, generation
        self.released = False
        # The concrete lease, so release resolves by identity even after a rotation re-aliases
        # ``session_id`` (both ids map to this same lease).
        self.lease = lease

    def __repr__(self) -> str:  # pragma: no cover - debug aid
        return (f"TurnLeaseToken(session_id={self.session_id!r}, owner_key={self.owner_key!r}, "
                f"generation={self.generation}, released={self.released})")


class _SessionLease:
    __slots__ = ("lock", "holder", "acquired_at", "last_used", "pending_acquires")

    def __init__(self) -> None:
        self.lock = asyncio.Lock()
        self.holder: Optional[TurnLeaseToken] = None
        self.acquired_at, self.last_used, self.pending_acquires = 0.0, time.time(), 0

    @property
    def idle(self) -> bool:
        """True when evictable: nobody holds or awaits it."""
        return self.holder is None and not self.lock.locked() and self.pending_acquires == 0


class SessionTurnLeaseRegistry:
    """Asyncio lease per resolved session_id. Process-local, single-event-loop by design (same
    visibility scope as the routing-key guards it extends); call only from the gateway loop."""

    def __init__(self, max_entries: int = DEFAULT_MAX_LEASES) -> None:
        self._leases: Dict[str, _SessionLease] = {}
        self._max_entries = max(1, int(max_entries))

    def _get_or_create(self, session_id: str) -> _SessionLease:
        if (lease := self._leases.get(session_id)) is None:
            self._evict_idle()
            lease = self._leases[session_id] = _SessionLease()
        lease.last_used = time.time()
        return lease

    def _evict_idle(self) -> None:
        """Drop oldest idle entries to fit a new lease under the cap; never a held/contended one."""
        if (overflow := len(self._leases) - self._max_entries + 1) <= 0:
            return
        idle = sorted((sid for sid, l in self._leases.items() if l.idle),
                      key=lambda sid: self._leases[sid].last_used)
        for sid in idle[:overflow]:
            self._leases.pop(sid, None)

    async def acquire(
        self, session_id: str, *, owner_key: str, generation: int, timeout: Optional[float] = None
    ) -> Optional[TurnLeaseToken]:
        """Acquire the lease for ``session_id``, waiting if held. Raises
        :class:`TurnLeaseTimeoutError` when the wait budget expires; None for a falsy id."""
        if not session_id:
            return None
        wait = float(timeout) if timeout and timeout > 0 else DEFAULT_LEASE_WAIT
        lease = self._get_or_create(session_id)
        token = TurnLeaseToken(session_id, owner_key, int(generation), lease=lease)
        if lease.lock.locked():
            logger.warning(
                "turn lease contention on session %s: routing key %s (gen %s) waiting behind "
                "in-flight turn held by routing key %s (gen %s, held %.0fs) — two routing keys "
                "are mapped to one session_id (#64934); serializing this turn behind the previous "
                "turn's flush",
                session_id, owner_key, generation, *_holder_desc(lease.holder),
                time.time() - lease.acquired_at if lease.acquired_at else -1.0)
        # Lock.release() wakes a waiter while leaving the lock momentarily unlocked. Count every
        # in-progress acquire across that handoff (even apparently-uncontended ones — wait_for()
        # may schedule them before the lock coroutine runs) so eviction cannot orphan the old
        # lock and create a second lock for the same session.
        lease.pending_acquires += 1
        try:
            await asyncio.wait_for(lease.lock.acquire(), timeout=wait)
        except asyncio.TimeoutError:
            logger.error(
                "turn lease wait timed out after %.0fs on session %s (waiter: routing key %s gen "
                "%s; holder: routing key %s gen %s) — failing closed: refusing to run this turn "
                "UNSERIALIZED against the still-held lease",
                wait, session_id, owner_key, generation, *_holder_desc(lease.holder))
            raise TurnLeaseTimeoutError(
                session_id, owner_key=owner_key, generation=generation, wait_seconds=wait) from None
        finally:
            lease.pending_acquires -= 1
        # Lock held and no await before holder publication, so the lease cannot become
        # evictable after the pending count is cleared.
        lease.holder = token
        lease.acquired_at = lease.last_used = time.time()
        return token

    def rebind(self, token: Optional[TurnLeaseToken], new_session_id: str) -> bool:
        """Alias a HELD lease onto ``new_session_id`` after mid-turn rotation (compression) so the
        flush target stays serialized: the SAME ``_SessionLease`` is registered under the new id
        (old mapping idle-evicts later), only the holder may rebind, the token follows. A live
        lease on the new id: log loudly, keep the old id (fail-open)."""
        if (token is None or token.released or not new_session_id
                or new_session_id == token.session_id):
            return False
        lease = token.lease
        if lease.holder is not token:
            return False
        existing = self._leases.get(new_session_id)
        if existing is not None and existing is not lease and not existing.idle:
            logger.warning(
                "turn lease rebind blocked: session %s rotated to %s mid-turn (holder: routing key "
                "%s gen %s) but the target session's lease is already live (holder: routing key %s "
                "gen %s) — keeping the lease on the old id; transcript writes on %s may "
                "interleave (#64934 rotation-alias edge)",
                token.session_id, new_session_id, token.owner_key, token.generation,
                *_holder_desc(existing.holder), new_session_id)
            return False
        self._leases[new_session_id] = lease
        lease.last_used = time.time()
        token.session_id = new_session_id
        return True

    def release(self, token: Optional[TurnLeaseToken]) -> bool:
        """Release ``token``'s lease. Idempotent; True only when this exact token was the current
        holder (a re-release or a stale token whose slot went to a newer turn is a safe no-op)."""
        if token is None or token.released:
            return False
        token.released = True
        lease = token.lease
        if lease.holder is not token:
            logger.debug("turn lease release skipped on session %s: token (key %s gen %s) is not "
                         "the current holder", token.session_id, token.owner_key, token.generation)
            return False
        lease.holder, lease.acquired_at, lease.last_used = None, 0.0, time.time()
        if lease.lock.locked():
            lease.lock.release()
        return True