STT-Assistant/src/stt_daemon/session.py

"""Pure conversation-session identity tracking.

:class:`SessionTracker` mints one id for the first observed utterance and rolls
over to a new id once the quiet gap between consecutive utterances exceeds the
configured threshold. It has no filesystem, config, or daemon dependencies: the
caller supplies the clock (``now``) and may inject the id factory, which keeps
the behavior deterministic and easy to test.
"""

import threading
from datetime import datetime
from typing import Callable


def _default_id_factory(now: float) -> str:
    """Format an epoch timestamp as a second-resolution session id."""
    return datetime.fromtimestamp(now).strftime("%Y-%m-%d-%H-%M-%S")


class SessionTracker:
    """Assigns a stable session id and rolls over after a quiet gap.

    Args:
        gap_s: Seconds of silence between consecutive observations that starts
            a new session. Values ``<= 0`` disable rollover after the first id.
        id_factory: Builds the id from an epoch timestamp. Defaults to the
            local-time ``YYYY-MM-DD-HH-MM-SS`` format; tests inject a counter.
    """

    def __init__(self, gap_s: float, id_factory: Callable[[float], str] | None = None):
        self._gap_s = gap_s
        self._id_factory = id_factory or _default_id_factory
        self._lock = threading.Lock()
        self._current_id: str | None = None
        self._last: float = 0.0

    def observe(self, now: float) -> str:
        """Return the session id for an observation at epoch ``now``.

        Mints on the first call, and mints a new id when ``now - last > gap_s``
        (only when ``gap_s > 0``); otherwise returns the current id. The
        observation time is remembered so the gap is measured between
        consecutive observations. Thread-safe.
        """
        with self._lock:
            if self._current_id is None or (
                self._gap_s > 0 and now - self._last > self._gap_s
            ):
                self._current_id = self._id_factory(now)
            self._last = now
            return self._current_id

    @property
    def current_id(self) -> str | None:
        """The current session id, or ``None`` before the first observation."""
        with self._lock:
            return self._current_id