"""Per-turn state for a voice session — a shared core type.

Each turn (user speaks → agent responds → bot speaks) gets its own
``TurnContext``.  Session creates one at turn start and discards it at
turn end, replacing the 15+ per-turn instance variables that previously
lived on Session.

Per-turn STT futures (``pending_stt_segment_futures``) also live here so
that a stale callback from the previous turn cannot resolve a future on
the next turn — the futures naturally die when the ``TurnContext`` is
replaced.

This lives at the package root (a leaf, depending only on the package-root
Epoch and cancellation primitives) rather than under ``session/`` so both the
``session`` layer *and* the lower ``stages`` layer can import it
*downward*.  Stages receive a ``TurnContext`` per ``execute`` call but
must not depend on the ``session`` package; keeping the type here
preserves the documented ``Session → Stages → Providers`` direction and
avoids the import cycle that previously forced a lazy ``__getattr__`` in
``session/__init__.py``.
"""

from __future__ import annotations

import asyncio
import time
from collections import deque
from typing import Protocol, runtime_checkable

from easycat._epoch import Lease
from easycat.cancel import CancelToken

TURN_AUDIO_LOG_MAXLEN = 10_000


@runtime_checkable
class TurnHandle(Protocol):
    """The contract between Session and turn-running collaborators.

    Session is the single authority on the active turn pointer and
    turn generation.  Collaborators read and write through this handle
    rather than holding ``Session`` refs.
    """

    @property
    def current(self) -> TurnContext | None: ...

    @property
    def generation(self) -> int: ...

    @property
    def no_turn(self) -> TurnContext: ...

    def begin(self, turn_id: str, cancel_token: CancelToken | None = None) -> TurnContext: ...

    def capture_identity(self) -> Lease[TurnContext | None]: ...

    def set(self, turn: TurnContext | None) -> None: ...


class TurnContext:
    """Holds all mutable state scoped to a single conversational turn."""

    _generation_counter: int = 0

    __slots__ = (
        "audio_bytes_sent",
        "audio_send_log",
        "audio_send_log_base_bytes",
        "audio_send_log_base_playout_cursor",
        "audio_send_log_base_playout_start",
        "bytes_since_last_mark",
        "cancel_token",
        "end_time",
        "first_agent_time",
        "first_tts_audio_time",
        "generation",
        "id",
        "last_barge_in_time",
        "pending_stt_segment_futures",
        "playback_ack_log",
        "playback_mark_to_bytes",
        "stt_final_time",
        "stt_has_uncommitted_audio",
        "stt_segments",
        "stt_track",
    )

    def __init__(self, turn_id: str, cancel_token: CancelToken) -> None:
        TurnContext._generation_counter += 1
        self.id = turn_id
        self.generation = TurnContext._generation_counter
        self.cancel_token = cancel_token

        # Timing markers (set as the turn progresses)
        self.end_time: float | None = None
        self.stt_final_time: float | None = None
        self.stt_segments: list[str] = []
        self.stt_track: str | None = None
        self.stt_has_uncommitted_audio = False
        self.first_agent_time: float | None = None
        self.first_tts_audio_time: float | None = None

        # Audio bytes sent to the transport during this turn.
        # Used to estimate what the user heard before a barge-in.
        self.audio_bytes_sent: int = 0
        self.audio_send_log: deque[tuple[float, int, float]] = deque(maxlen=TURN_AUDIO_LOG_MAXLEN)
        self.audio_send_log_base_bytes: int = 0
        self.audio_send_log_base_playout_start: float | None = None
        self.audio_send_log_base_playout_cursor: float | None = None

        # Playback mark tracking (maps mark names to cumulative byte positions)
        self.playback_mark_to_bytes: dict[str, int] = {}
        self.playback_ack_log: deque[tuple[float, int]] = deque(maxlen=TURN_AUDIO_LOG_MAXLEN)
        self.bytes_since_last_mark: int = 0

        self.last_barge_in_time: float | None = None

        # STT future plumbing — per-turn so a stale callback cannot resolve
        # a future on the next turn (the futures die with the TurnContext).
        self.pending_stt_segment_futures: list[asyncio.Future[str]] = []

    def record_barge_in(self) -> None:
        self.last_barge_in_time = time.monotonic()

    def append_stt_segment(self, text: str, *, track: str | None = None) -> None:
        normalized = " ".join(text.split())
        if normalized:
            self.stt_segments.append(normalized)
        if track is not None:
            self.stt_track = track

    @property
    def transcript_text(self) -> str:
        return " ".join(self.stt_segments).strip()

    def record_audio_sent(self, size: int, duration_ms: float) -> None:
        """Record that audio was sent to the transport."""
        self.audio_bytes_sent += size
        send_time = time.monotonic()
        if self.audio_send_log.maxlen and len(self.audio_send_log) == self.audio_send_log.maxlen:
            self._record_audio_send_log_eviction(self.audio_send_log[0])
        self.audio_send_log.append((send_time, size, duration_ms))
        self.bytes_since_last_mark += size

    def _record_audio_send_log_eviction(self, entry: tuple[float, int, float]) -> None:
        """Fold an evicted send-log entry into cumulative playback accounting."""
        send_time, size, duration_ms = entry
        self.audio_send_log_base_bytes += max(size, 0)
        if duration_ms <= 0:
            return
        start_time = send_time
        if (
            self.audio_send_log_base_playout_cursor is not None
            and self.audio_send_log_base_playout_cursor > start_time
        ):
            start_time = self.audio_send_log_base_playout_cursor
        if self.audio_send_log_base_playout_start is None:
            self.audio_send_log_base_playout_start = start_time
        self.audio_send_log_base_playout_cursor = start_time + (duration_ms / 1000.0)
