core.telemetry¶

class ConfidenceStats(BaseModel):
    """Aggregate statistics on prediction confidence."""

    mean: float
    median: float
    p5: float
    p95: float
    std: float

class TelemetrySnapshot(BaseModel):
    """Point-in-time quality snapshot of a prediction window."""

    timestamp: datetime
    user_id: str | None = None
    window_start: datetime | None = None
    window_end: datetime | None = None
    total_windows: int
    feature_missingness: dict[str, float] = Field(default_factory=dict)
    confidence_stats: ConfidenceStats | None = None
    reject_rate: float = 0.0
    mean_entropy: float = 0.0
    class_distribution: dict[str, float] = Field(default_factory=dict)
    schema_version: str = f"features_{LATEST_FEATURE_SCHEMA_VERSION}"
    suggestions_per_day: int = 0

@dataclass(eq=False)
class TelemetryStore:
    """Append-only JSONL store for telemetry snapshots."""

    store_dir: str | Path = DEFAULT_TELEMETRY_DIR
    _dir: Path = field(init=False)

    def __post_init__(self) -> None:
        self._dir = Path(self.store_dir)
        self._dir.mkdir(parents=True, exist_ok=True)

    def _path_for(self, user_id: str | None) -> Path:
        name = f"telemetry_{user_id}.jsonl" if user_id else "telemetry_global.jsonl"
        return self._dir / name

    def append(self, snapshot: TelemetrySnapshot) -> Path:
        """Append a snapshot to the appropriate JSONL file.

        Returns:
            Path of the file written to.
        """
        path = self._path_for(snapshot.user_id)
        line = snapshot.model_dump_json() + "\n"

        path.parent.mkdir(parents=True, exist_ok=True)
        fd, tmp_str = tempfile.mkstemp(dir=self._dir, suffix=".tmp")
        tmp = Path(tmp_str)
        try:
            existing = path.read_text() if path.exists() else ""
            os.write(fd, (existing + line).encode())
            os.close(fd)
            tmp.replace(path)
        except BaseException:
            os.close(fd)
            tmp.unlink(missing_ok=True)
            raise
        return path

    def read_recent(
        self,
        n: int = 10,
        *,
        user_id: str | None = None,
    ) -> list[TelemetrySnapshot]:
        """Read the last *n* snapshots.

        Args:
            n: Maximum number of snapshots to return.
            user_id: Scope to a specific user (``None`` = global).

        Returns:
            List of :class:`TelemetrySnapshot` (newest last).
        """
        path = self._path_for(user_id)
        if not path.exists():
            return []
        lines = [line for line in path.read_text().splitlines() if line.strip()]
        return [TelemetrySnapshot.model_validate_json(line) for line in lines[-n:]]

    def read_range(
        self,
        start: datetime,
        end: datetime,
        *,
        user_id: str | None = None,
    ) -> list[TelemetrySnapshot]:
        """Read snapshots whose timestamp falls within [*start*, *end*].

        Args:
            start: Inclusive lower bound.
            end: Inclusive upper bound.
            user_id: Scope to a specific user (``None`` = global).

        Returns:
            Matching snapshots ordered by timestamp.
        """
        all_snaps = self.read_recent(n=10_000, user_id=user_id)
        return [s for s in all_snaps if start <= s.timestamp <= end]

@dataclass
class SuggestionTracker:
    """In-memory counter of suggestion events grouped by calendar date.

    Used to enforce the decision-#4 guardrail: a loaded model must
    produce at least one suggestion per active day.  The tray or
    online loop calls :meth:`record` for each non-rejected prediction
    surfaced to the user.  At end-of-day (or on shutdown),
    :meth:`check_zero_suggestions` logs a warning if the count is zero.
    """

    _counts: dict[str, int] = field(default_factory=lambda: defaultdict(int))

    def record(self, ts: datetime) -> None:
        """Record a suggestion event at *ts*."""
        date_key = ts.strftime("%Y-%m-%d")
        self._counts[date_key] += 1

    def count_for_date(self, date_str: str) -> int:
        """Return the suggestion count for a given date string (``YYYY-MM-DD``)."""
        return self._counts.get(date_str, 0)

    def check_zero_suggestions(self, date_str: str, *, model_loaded: bool) -> None:
        """Log a warning if *model_loaded* and zero suggestions were recorded.

        Args:
            date_str: Calendar date to check (``YYYY-MM-DD``).
            model_loaded: Whether a model was loaded during the day.
        """
        if model_loaded and self.count_for_date(date_str) == 0:
            logger.warning(
                "Zero suggestions on %s with a loaded model — "
                "reject threshold may be too high",
                date_str,
            )