src/core/session.py - develop - dsboikov/aiBoardRoom

src/core/session.py
90 строк3 KB
Денис
Grok added
03 авг 2026, 16:31
03 авг 2026, 16:31255c2e0
"""Session lifecycle and status journal."""

from __future__ import annotations

from dataclasses import dataclass, field
from datetime import datetime, timezone
from enum import Enum
from typing import Any, Callable

from storage.sqlite import Database, get_db
from utils.logger import get_logger

logger = get_logger("session")


class SessionStatus(str, Enum):
    IDLE = "idle"
    RUNNING = "running"
    DONE = "done"
    ERROR = "error"
    CANCELLED = "cancelled"


@dataclass
class StatusEvent:
    kind: str
    message: str
    payload: dict[str, Any] = field(default_factory=dict)
    timestamp: str = field(
        default_factory=lambda: datetime.now(timezone.utc).replace(microsecond=0).isoformat()
    )


StatusCallback = Callable[[StatusEvent], None]


class SessionManager:
    """Track council session state and emit UI status events."""

    def __init__(self, db: Database | None = None) -> None:
        self.db = db or get_db()
        self.session_id: int | None = None
        self.room_id: int | None = None
        self.status = SessionStatus.IDLE
        self.events: list[StatusEvent] = []
        self._callbacks: list[StatusCallback] = []
        self.session_spend: dict[str, float] = {}  # provider -> USD
        self.session_tokens: dict[str, int] = {}

    def on_status(self, callback: StatusCallback) -> None:
        self._callbacks.append(callback)

    def emit(self, kind: str, message: str, **payload: Any) -> StatusEvent:
        event = StatusEvent(kind=kind, message=message, payload=payload)
        self.events.append(event)
        logger.info("[%s] %s", kind, message)
        for cb in list(self._callbacks):
            try:
                cb(event)
            except Exception as exc:  # noqa: BLE001
                logger.warning("Status callback error: %s", exc)
        return event

    def start(self, room_id: int, mode: str) -> int:
        self.room_id = room_id
        self.session_id = self.db.create_session(room_id, mode)
        self.status = SessionStatus.RUNNING
        self.events.clear()
        self.session_spend.clear()
        self.session_tokens.clear()
        self.emit("session_started", f"Сессия запущена (режим={mode})", mode=mode)
        return self.session_id

    def finish(self, status: SessionStatus = SessionStatus.DONE) -> None:
        self.status = status
        if self.session_id is not None:
            self.db.finish_session(self.session_id, status.value)
        self.emit("session_finished", f"Сессия: {status.value}", status=status.value)

    def record_usage(self, provider: str, tokens: int, cost_usd: float) -> None:
        self.session_tokens[provider] = self.session_tokens.get(provider, 0) + tokens
        self.session_spend[provider] = self.session_spend.get(provider, 0.0) + cost_usd

    def get_session_spend_summary(self) -> dict[str, Any]:
        return {
            "tokens": dict(self.session_tokens),
            "cost_usd": dict(self.session_spend),
            "total_cost": sum(self.session_spend.values()),
            "total_tokens": sum(self.session_tokens.values()),
        }