Trixy7/trixy_core/conversation/session.py

# -*- coding: utf-8 -*-
"""
Conversation Session - Einzelne Dialog-Session.
"""

from dataclasses import dataclass, field
from datetime import datetime, timedelta
from enum import Enum
from typing import Any
import uuid


class SessionState(Enum):
    """Zustand einer Conversation-Session."""
    CREATED = "created"           # Session erstellt
    AWAITING_INPUT = "awaiting"   # Warte auf User-Input
    PROCESSING = "processing"     # Verarbeite Input
    RESPONDING = "responding"     # Antwort wird generiert
    FOLLOW_UP = "follow_up"       # Rückfrage gestellt
    COMPLETED = "completed"       # Session abgeschlossen
    TIMEOUT = "timeout"           # Session durch Timeout beendet
    CANCELLED = "cancelled"       # Session abgebrochen


class TurnType(Enum):
    """Typ eines Dialog-Turns."""
    USER_SPEECH = "user_speech"       # User-Spracheingabe
    USER_TEXT = "user_text"           # User-Texteingabe
    ASSISTANT_SPEECH = "assistant_speech"  # Assistenten-Sprachausgabe
    ASSISTANT_TEXT = "assistant_text"      # Assistenten-Text
    FOLLOW_UP = "follow_up"           # Rückfrage
    SYSTEM = "system"                 # System-Nachricht


@dataclass
class SessionTurn:
    """Ein Turn (Gesprächsbeitrag) in der Session."""

    turn_id: str
    turn_type: TurnType
    timestamp: datetime
    content: str  # Text-Inhalt (transkribiert oder generiert)

    # Audio-Daten (optional)
    audio_data: bytes | None = None
    audio_duration_seconds: float = 0.0

    # Metadaten
    metadata: dict[str, Any] = field(default_factory=dict)

    # Intent/Entity-Erkennung
    intent: str | None = None
    entities: dict[str, Any] = field(default_factory=dict)
    confidence: float = 0.0

    # Für Rückfragen
    requires_response: bool = False
    response_turn_id: str | None = None

    def to_dict(self) -> dict:
        """Konvertiert zu Dictionary."""
        return {
            "turn_id": self.turn_id,
            "turn_type": self.turn_type.value,
            "timestamp": self.timestamp.isoformat(),
            "content": self.content,
            "audio_duration": self.audio_duration_seconds,
            "intent": self.intent,
            "entities": self.entities,
            "confidence": self.confidence,
            "requires_response": self.requires_response,
            "metadata": self.metadata,
        }


@dataclass
class SessionConfig:
    """Konfiguration für Conversation-Sessions."""

    # Timeouts
    input_timeout_seconds: float = 60.0    # Timeout für User-Input
    follow_up_timeout_seconds: float = 60.0  # Timeout für Rückfrage-Antwort
    session_timeout_seconds: float = 300.0   # Gesamt-Session-Timeout

    # Limits
    max_turns: int = 20                     # Maximale Turns pro Session
    max_follow_ups: int = 5                 # Maximale Rückfragen

    # Verhalten
    auto_complete_on_timeout: bool = True   # Bei Timeout automatisch abschließen


class ConversationSession:
    """
    Eine Conversation-Session.

    Repräsentiert einen kompletten Dialog von Wakeword bis Abschluss.
    Trackt alle Turns (User-Eingaben, Assistenten-Antworten, Rückfragen).
    """

    def __init__(
        self,
        session_id: str | None = None,
        satellite_id: str | None = None,
        wakeword_type: str | None = None,
        config: SessionConfig | None = None,
    ):
        """
        Initialisiert die Session.

        Args:
            session_id: Session-ID (wird generiert wenn None)
            satellite_id: ID des auslösenden Satellites
            wakeword_type: Typ des erkannten Wakewords
            config: Session-Konfiguration
        """
        self._session_id = session_id or f"conv-{uuid.uuid4().hex[:12]}"
        self._satellite_id = satellite_id
        self._wakeword_type = wakeword_type
        self._config = config or SessionConfig()

        # Timestamps
        self._created_at = datetime.now()
        self._started_at: datetime | None = None
        self._completed_at: datetime | None = None
        self._last_activity = datetime.now()

        # State
        self._state = SessionState.CREATED
        self._turns: list[SessionTurn] = []
        self._follow_up_count = 0

        # Kontext für diese Session
        self._context: dict[str, Any] = {}

        # Pending Follow-Up
        self._pending_follow_up: SessionTurn | None = None

    # === Properties ===

    @property
    def session_id(self) -> str:
        """Session-ID."""
        return self._session_id

    @property
    def satellite_id(self) -> str | None:
        """Satellite-ID."""
        return self._satellite_id

    @property
    def wakeword_type(self) -> str | None:
        """Wakeword-Typ."""
        return self._wakeword_type

    @property
    def config(self) -> SessionConfig:
        """Konfiguration."""
        return self._config

    @property
    def state(self) -> SessionState:
        """Aktueller Zustand."""
        return self._state

    @property
    def is_active(self) -> bool:
        """Prüft ob Session aktiv ist."""
        return self._state in (
            SessionState.CREATED,
            SessionState.AWAITING_INPUT,
            SessionState.PROCESSING,
            SessionState.RESPONDING,
            SessionState.FOLLOW_UP,
        )

    @property
    def is_completed(self) -> bool:
        """Prüft ob Session abgeschlossen ist."""
        return self._state in (
            SessionState.COMPLETED,
            SessionState.TIMEOUT,
            SessionState.CANCELLED,
        )

    @property
    def created_at(self) -> datetime:
        """Erstellungszeit."""
        return self._created_at

    @property
    def started_at(self) -> datetime | None:
        """Startzeit (erster User-Turn)."""
        return self._started_at

    @property
    def completed_at(self) -> datetime | None:
        """Abschlusszeit."""
        return self._completed_at

    @property
    def last_activity(self) -> datetime:
        """Letzte Aktivität."""
        return self._last_activity

    @property
    def duration_seconds(self) -> float:
        """Session-Dauer in Sekunden."""
        end = self._completed_at or datetime.now()
        return (end - self._created_at).total_seconds()

    @property
    def turns(self) -> list[SessionTurn]:
        """Alle Turns."""
        return list(self._turns)

    @property
    def turn_count(self) -> int:
        """Anzahl Turns."""
        return len(self._turns)

    @property
    def user_turns(self) -> list[SessionTurn]:
        """Nur User-Turns."""
        return [t for t in self._turns if t.turn_type in (
            TurnType.USER_SPEECH, TurnType.USER_TEXT
        )]

    @property
    def assistant_turns(self) -> list[SessionTurn]:
        """Nur Assistenten-Turns."""
        return [t for t in self._turns if t.turn_type in (
            TurnType.ASSISTANT_SPEECH, TurnType.ASSISTANT_TEXT, TurnType.FOLLOW_UP
        )]

    @property
    def follow_up_count(self) -> int:
        """Anzahl Rückfragen."""
        return self._follow_up_count

    @property
    def has_pending_follow_up(self) -> bool:
        """Prüft ob Rückfrage aussteht."""
        return self._pending_follow_up is not None

    @property
    def context(self) -> dict[str, Any]:
        """Session-Kontext."""
        return self._context

    # === Turn Management ===

    def add_user_turn(
        self,
        content: str,
        audio_data: bytes | None = None,
        audio_duration: float = 0.0,
        intent: str | None = None,
        entities: dict | None = None,
        confidence: float = 0.0,
        metadata: dict | None = None,
    ) -> SessionTurn:
        """
        Fügt User-Turn hinzu.

        Args:
            content: Transkribierter Text
            audio_data: Original-Audio (optional)
            audio_duration: Audio-Dauer
            intent: Erkannter Intent
            entities: Extrahierte Entities
            confidence: Erkennungs-Confidence
            metadata: Zusätzliche Metadaten

        Returns:
            Erstellter Turn
        """
        self._check_can_add_turn()

        turn = SessionTurn(
            turn_id=self._generate_turn_id(),
            turn_type=TurnType.USER_SPEECH if audio_data else TurnType.USER_TEXT,
            timestamp=datetime.now(),
            content=content,
            audio_data=audio_data,
            audio_duration_seconds=audio_duration,
            intent=intent,
            entities=entities or {},
            confidence=confidence,
            metadata=metadata or {},
        )

        self._add_turn(turn)

        # Erster User-Turn startet die Session
        if self._started_at is None:
            self._started_at = turn.timestamp

        # Wenn Follow-Up beantwortet wurde
        if self._pending_follow_up:
            self._pending_follow_up.response_turn_id = turn.turn_id
            self._pending_follow_up = None

        self._state = SessionState.PROCESSING

        return turn

    def add_assistant_turn(
        self,
        content: str,
        audio_data: bytes | None = None,
        audio_duration: float = 0.0,
        is_follow_up: bool = False,
        metadata: dict | None = None,
    ) -> SessionTurn:
        """
        Fügt Assistenten-Turn hinzu.

        Args:
            content: Antwort-Text
            audio_data: Generiertes Audio (optional)
            audio_duration: Audio-Dauer
            is_follow_up: Ist dies eine Rückfrage?
            metadata: Zusätzliche Metadaten

        Returns:
            Erstellter Turn
        """
        turn_type = TurnType.FOLLOW_UP if is_follow_up else TurnType.ASSISTANT_SPEECH

        turn = SessionTurn(
            turn_id=self._generate_turn_id(),
            turn_type=turn_type,
            timestamp=datetime.now(),
            content=content,
            audio_data=audio_data,
            audio_duration_seconds=audio_duration,
            metadata=metadata or {},
            requires_response=is_follow_up,
        )

        self._add_turn(turn)

        if is_follow_up:
            self._follow_up_count += 1
            self._pending_follow_up = turn
            self._state = SessionState.FOLLOW_UP
        else:
            self._state = SessionState.RESPONDING

        return turn

    def add_system_turn(
        self,
        content: str,
        metadata: dict | None = None,
    ) -> SessionTurn:
        """
        Fügt System-Turn hinzu.

        Args:
            content: System-Nachricht
            metadata: Zusätzliche Metadaten

        Returns:
            Erstellter Turn
        """
        turn = SessionTurn(
            turn_id=self._generate_turn_id(),
            turn_type=TurnType.SYSTEM,
            timestamp=datetime.now(),
            content=content,
            metadata=metadata or {},
        )

        self._add_turn(turn)
        return turn

    def _add_turn(self, turn: SessionTurn) -> None:
        """Interne Methode zum Hinzufügen eines Turns."""
        self._turns.append(turn)
        self._last_activity = turn.timestamp

    def _check_can_add_turn(self) -> None:
        """Prüft ob Turn hinzugefügt werden kann."""
        if self.is_completed:
            raise RuntimeError(f"Session {self._session_id} ist bereits abgeschlossen")

        if len(self._turns) >= self._config.max_turns:
            raise RuntimeError(f"Maximale Turns ({self._config.max_turns}) erreicht")

    def _generate_turn_id(self) -> str:
        """Generiert Turn-ID."""
        return f"{self._session_id}-t{len(self._turns) + 1}"

    # === Lifecycle ===

    def start(self) -> None:
        """Startet die Session."""
        if self._state != SessionState.CREATED:
            return

        self._started_at = datetime.now()
        self._state = SessionState.AWAITING_INPUT

    def complete(self, reason: str | None = None) -> None:
        """
        Schließt die Session ab.

        Args:
            reason: Abschlussgrund (optional)
        """
        if self.is_completed:
            return

        self._state = SessionState.COMPLETED
        self._completed_at = datetime.now()

        if reason:
            self.add_system_turn(f"Session beendet: {reason}")

    def timeout(self) -> None:
        """Beendet Session wegen Timeout."""
        if self.is_completed:
            return

        self._state = SessionState.TIMEOUT
        self._completed_at = datetime.now()
        self.add_system_turn("Session-Timeout")

    def cancel(self, reason: str | None = None) -> None:
        """
        Bricht Session ab.

        Args:
            reason: Abbruchgrund
        """
        if self.is_completed:
            return

        self._state = SessionState.CANCELLED
        self._completed_at = datetime.now()

        if reason:
            self.add_system_turn(f"Session abgebrochen: {reason}")

    def check_timeout(self) -> bool:
        """
        Prüft ob Session getimeoutet ist.

        Returns:
            True wenn Timeout erreicht
        """
        now = datetime.now()

        # Gesamt-Timeout
        if self.duration_seconds >= self._config.session_timeout_seconds:
            if self._config.auto_complete_on_timeout:
                self.timeout()
            return True

        # Input-Timeout (wenn auf Input gewartet wird)
        if self._state == SessionState.AWAITING_INPUT:
            idle_seconds = (now - self._last_activity).total_seconds()
            if idle_seconds >= self._config.input_timeout_seconds:
                if self._config.auto_complete_on_timeout:
                    self.timeout()
                return True

        # Follow-Up-Timeout
        if self._state == SessionState.FOLLOW_UP:
            idle_seconds = (now - self._last_activity).total_seconds()
            if idle_seconds >= self._config.follow_up_timeout_seconds:
                if self._config.auto_complete_on_timeout:
                    self.timeout()
                return True

        return False

    # === Context ===

    def set_context(self, key: str, value: Any) -> None:
        """Setzt Kontext-Variable."""
        self._context[key] = value

    def get_context(self, key: str, default: Any = None) -> Any:
        """Gibt Kontext-Variable zurück."""
        return self._context.get(key, default)

    def clear_context(self) -> None:
        """Löscht Kontext."""
        self._context.clear()

    # === Serialization ===

    def to_dict(self) -> dict:
        """Konvertiert zu Dictionary."""
        return {
            "session_id": self._session_id,
            "satellite_id": self._satellite_id,
            "wakeword_type": self._wakeword_type,
            "state": self._state.value,
            "created_at": self._created_at.isoformat(),
            "started_at": self._started_at.isoformat() if self._started_at else None,
            "completed_at": self._completed_at.isoformat() if self._completed_at else None,
            "duration_seconds": self.duration_seconds,
            "turn_count": len(self._turns),
            "follow_up_count": self._follow_up_count,
            "turns": [t.to_dict() for t in self._turns],
            "context": self._context,
        }

    def get_transcript(self) -> str:
        """
        Gibt Transkript der Session zurück.

        Returns:
            Formatiertes Transkript
        """
        lines = []

        for turn in self._turns:
            if turn.turn_type == TurnType.SYSTEM:
                lines.append(f"[System] {turn.content}")
            elif turn.turn_type in (TurnType.USER_SPEECH, TurnType.USER_TEXT):
                lines.append(f"User: {turn.content}")
            elif turn.turn_type == TurnType.FOLLOW_UP:
                lines.append(f"Assistent (Rückfrage): {turn.content}")
            else:
                lines.append(f"Assistent: {turn.content}")

        return "\n".join(lines)