src/reachy_mini_receptionist/session_manager.py

"""Session manager for the receptionist state machine.

Single source of truth for the active visitor's session. The face worker
(background thread), conversation controller, and tools (asyncio) all mutate
state through this manager. All mutations take a lock; reads return snapshots.

State changes are dispatched to a subscriber callback so the realtime handler
can push them to the LLM as context events (mirroring how face events are
pushed today via FaceRecognitionWorker.set_face_event_callback).
"""
from __future__ import annotations

import logging
import threading
import time
from dataclasses import dataclass, field, replace
from datetime import datetime
from typing import Any, Callable, Optional

from reachy_mini_receptionist.receptionist_state import ReceptionState

logger = logging.getLogger(__name__)


@dataclass
class VisitorSession:
    """Snapshot of the current visitor's session state."""

    current_state: ReceptionState = ReceptionState.IDLE
    visitor_name: Optional[str] = None
    recognized_face_name: Optional[str] = None
    employee_name: Optional[str] = None
    purpose: Optional[str] = None
    matched_appointment: Optional[dict[str, Any]] = None
    email_sent_to: Optional[str] = None
    error_message: Optional[str] = None
    started_at: float = field(default_factory=time.time)
    last_event_at: float = field(default_factory=time.time)
    # Last completed visitor transcript — used by register_guest's
    # confirmation enforcement so the backend can verify the visitor
    # actually said "yes" before saving a face under a name. The realtime
    # handler populates this on every input_audio_transcription.completed.
    last_user_transcript: Optional[str] = None
    # How many times register_guest has refused this session (no
    # confirmation, mistranscribed name, etc.). After 2 refusals the
    # bot should hand off to an operator instead of looping forever.
    register_guest_rejections: int = 0
    # Last name proposed via a rejected register_guest(confirmed=false)
    # call. Used by send_email's "no visitor identity" block message to
    # give a directive error ("call register_guest(name=<this>,
    # confirmed=true) now") instead of telling the model to ask the
    # visitor again — which it already did 5 seconds ago.
    last_register_guest_name: Optional[str] = None

    def to_dict(self) -> dict[str, Any]:
        """Serialize for /api/session and LLM context messages."""
        return {
            "current_state": self.current_state.value,
            "visitor_name": self.visitor_name,
            "recognized_face_name": self.recognized_face_name,
            "employee_name": self.employee_name,
            "purpose": self.purpose,
            "matched_appointment": self.matched_appointment,
            "email_sent_to": self.email_sent_to,
            "error_message": self.error_message,
            "started_at_iso": datetime.fromtimestamp(self.started_at).isoformat(timespec="seconds"),
            "last_event_at_iso": datetime.fromtimestamp(self.last_event_at).isoformat(timespec="seconds"),
        }


StateChangeCallback = Callable[[ReceptionState, ReceptionState, VisitorSession], None]
"""Called as ``callback(previous_state, new_state, snapshot)`` on every transition."""


# How long a non-IDLE session can sit without any transition before we
# consider it stale and reset. Visitors who walk up but never speak should
# not hold state forever.
DEFAULT_IDLE_TIMEOUT_SECONDS: float = 60.0


class SessionManager:
    """Thread-safe holder for the active visitor session."""

    def __init__(
        self,
        idle_timeout_seconds: float = DEFAULT_IDLE_TIMEOUT_SECONDS,
        visitor_log: Any | None = None,
    ) -> None:
        self._lock = threading.Lock()
        self._session = VisitorSession()
        self._callback: Optional[StateChangeCallback] = None
        self._idle_timeout_seconds = float(idle_timeout_seconds)
        # Optional sink: when reset() is called, persist the pre-reset
        # session here so both the face-loss path AND the timeout path
        # produce log entries.
        self._visitor_log = visitor_log

    @property
    def session(self) -> VisitorSession:
        """Return a snapshot of the current session. Safe from any thread."""
        with self._lock:
            return replace(self._session)

    @property
    def current_state(self) -> ReceptionState:
        with self._lock:
            return self._session.current_state

    def subscribe(self, callback: Optional[StateChangeCallback]) -> None:
        """Register (or clear, with ``None``) the state-change callback.

        Only one subscriber, matching the face worker's pattern.
        """
        with self._lock:
            self._callback = callback

    def transition(self, new_state: ReceptionState, **updates: Any) -> VisitorSession:
        """Move to a new state and optionally update session fields atomically.

        Returns a snapshot. The callback is dispatched OUTSIDE the lock so
        subscribers can safely call back in without deadlocking.
        """
        with self._lock:
            self._validate_updates(updates)
            previous_state = self._session.current_state
            self._session = replace(
                self._session,
                current_state=new_state,
                last_event_at=time.time(),
                **updates,
            )
            snapshot = replace(self._session)
            callback = self._callback

        logger.info(
            "Session: %s -> %s (visitor=%r employee=%r)",
            previous_state.value,
            new_state.value,
            snapshot.visitor_name,
            snapshot.employee_name,
        )

        if previous_state != new_state and callback is not None:
            try:
                callback(previous_state, new_state, snapshot)
            except Exception as e:
                logger.warning(
                    "SessionManager: callback raised %s: %s",
                    type(e).__name__,
                    e,
                )

        return snapshot

    def update(self, **updates: Any) -> VisitorSession:
        """Update session fields without changing state.

        For stashing data (e.g. the recognized face name) where no transition
        is appropriate yet.
        """
        with self._lock:
            self._validate_updates(updates)
            self._session = replace(
                self._session, **updates, last_event_at=time.time()
            )
            return replace(self._session)

    # ------------------------------------------------------------------
    # Atomic claim helpers (used by send_email to prevent races where two
    # concurrent tool calls both pass a read-then-check dedupe and end up
    # delivering the same notification twice).
    # ------------------------------------------------------------------

    def try_claim_email(self, to: str) -> bool:
        """Atomically claim the right to send an email to ``to``.

        Returns True iff the caller wins the race; False if another
        concurrent caller already claimed (or completed) a send to the
        same recipient within this session. Callers that win MUST call
        ``release_email_claim(to)`` on failure so a retry can succeed.

        The mutation here intentionally does NOT fire the state-change
        callback — this is an in-flight reservation, not a transition.
        The real NOTIFIED transition (which pushes context to the LLM)
        still happens via the controller's on_tool_completed path after
        Resend confirms delivery.
        """
        norm = (to or "").strip().lower()
        if not norm:
            return False
        with self._lock:
            existing = (self._session.email_sent_to or "").strip().lower()
            if existing == norm:
                return False
            self._session = replace(
                self._session,
                email_sent_to=to,
                last_event_at=time.time(),
            )
            return True

    def release_email_claim(self, to: str) -> None:
        """Undo a previous ``try_claim_email`` after a delivery failure."""
        norm = (to or "").strip().lower()
        if not norm:
            return
        with self._lock:
            existing = (self._session.email_sent_to or "").strip().lower()
            if existing == norm:
                self._session = replace(
                    self._session,
                    email_sent_to=None,
                    last_event_at=time.time(),
                )

    def reset(self) -> VisitorSession:
        """Drop the current visitor and return to IDLE.

        If a ``visitor_log`` was provided at construction, the pre-reset
        snapshot is persisted there before the fields are cleared. This
        means BOTH the controller's face-loss path AND the timeout path
        produce a log entry — without each call site needing to remember
        to capture the snapshot.
        """
        # Capture pre-reset state for logging. We do this outside the lock
        # (the manager handles its own concurrency on read).
        pre_snapshot = self.session
        pre_state = pre_snapshot.current_state

        if (
            self._visitor_log is not None
            and pre_state != ReceptionState.IDLE
        ):
            try:
                self._visitor_log.record_visit(pre_snapshot, pre_state)
            except Exception as e:
                logger.warning(
                    "SessionManager: visitor_log.record_visit raised %s: %s",
                    type(e).__name__, e,
                )

        return self.transition(
            ReceptionState.IDLE,
            visitor_name=None,
            recognized_face_name=None,
            employee_name=None,
            purpose=None,
            matched_appointment=None,
            email_sent_to=None,
            error_message=None,
            started_at=time.time(),
            last_user_transcript=None,
            register_guest_rejections=0,
        )

    # ------------------------------------------------------------------
    # Idle handling
    # ------------------------------------------------------------------

    def bump_register_guest_rejection(self) -> int:
        """Increment the per-session register_guest rejection counter.

        Returns the new count. Used by register_guest to detect when the
        visitor's name cannot be captured reliably (after 2+ rejections,
        the bot should hand off to an operator rather than looping).
        """
        with self._lock:
            self._session = replace(
                self._session,
                register_guest_rejections=self._session.register_guest_rejections + 1,
                last_event_at=time.time(),
            )
            return self._session.register_guest_rejections

    def record_user_transcript(self, transcript: str) -> None:
        """Stash the most recent visitor utterance for backend introspection.

        Used by register_guest's confirmation guard so we can verify that
        the visitor actually said an affirmation before the LLM tries to
        save a face under that name. Does NOT fire the state-change
        callback — purely a side-channel record.
        """
        with self._lock:
            self._session = replace(
                self._session,
                last_user_transcript=transcript,
                last_event_at=time.time(),
            )

    def record_register_guest_attempt(self, name: str) -> None:
        """Stash the name proposed via a rejected register_guest(confirmed=false).

        Used by send_email's no-visitor-identity block path so we can tell
        the LLM exactly which name to re-call register_guest with, instead
        of asking the visitor again. Does NOT fire the state-change callback.
        """
        with self._lock:
            self._session = replace(
                self._session,
                last_register_guest_name=(name or "").strip() or None,
                last_event_at=time.time(),
            )

    def touch(self) -> None:
        """Bump ``last_event_at`` without changing any other session field.

        Used by the realtime handler when the visitor starts speaking or
        a transcript completes — those events prove the session is still
        active even though no state transition fires. Without this, a
        flow like 'I heard Henry — is that right?' followed by the
        visitor thinking for 65s would get auto-reset out from under the
        bot, and the visitor's eventual 'yes' would arrive into an empty
        IDLE session.

        Does NOT fire the state-change callback — this is a pure
        liveness ping, not a transition.
        """
        with self._lock:
            self._session = replace(
                self._session, last_event_at=time.time()
            )

    def is_stale(self, now: Optional[float] = None) -> bool:
        """Return True if a non-IDLE session has had no transition for the timeout.

        Does NOT mutate. Callers can decide whether to reset.
        """
        ref = now if now is not None else time.time()
        with self._lock:
            if self._session.current_state == ReceptionState.IDLE:
                return False
            return (ref - self._session.last_event_at) > self._idle_timeout_seconds

    def maybe_reset_if_stale(self) -> bool:
        """Reset the session if it has gone stale. Returns True if reset happened.

        Safe to call from any thread on any tick. Designed to be invoked
        from the realtime handler's existing periodic ``emit()`` loop so
        we don't need another timer thread.
        """
        if not self.is_stale():
            return False
        logger.info(
            "Session idle for > %.0fs — auto-resetting",
            self._idle_timeout_seconds,
        )
        self.reset()
        return True

    @staticmethod
    def _validate_updates(updates: dict[str, Any]) -> None:
        unknown = set(updates) - set(VisitorSession.__dataclass_fields__)
        if unknown:
            raise ValueError(f"Unknown session fields: {sorted(unknown)!r}")