--- title: session_logger ---

Per-session JSONL event logger.

Each session gets its own JSONL file under the configured log root so post-hoc analytics (enhancer latency, GPU assignment, segment timings) can be recovered without a tracing backend. The internal UI uses this format; keeping the same shape makes log tooling portable.

fastvideo.entrypoints.streaming.session_logger.SessionLogEvent

SessionLogEvent(session_id: str, event: str, payload: dict[str, Any] = dict(), ts: float = time.time())

One line in the session JSONL file.

Source
@dataclass
class SessionLogEvent:
    """One line in the session JSONL file."""

    session_id: str
    event: str
    payload: dict[str, Any] = field(default_factory=dict)
    ts: float = field(default_factory=time.time)

fastvideo.entrypoints.streaming.session_logger.SessionLogger

SessionLogger(log_dir: str | None)

Append-only JSONL logger keyed by session id.

Thread-safe; the server may be writing from multiple asyncio tasks (fMP4 encoder thread + control-frame handler) for the same session.

Source
class SessionLogger:
    """Append-only JSONL logger keyed by session id.

    Thread-safe; the server may be writing from multiple asyncio tasks
    (fMP4 encoder thread + control-frame handler) for the same session.
    """

    def __init__(self, log_dir: str | None) -> None:
        self._log_dir = log_dir
        self._files: dict[str, TextIO] = {}
        self._locks: dict[str, threading.Lock] = {}
        self._registry_lock = threading.Lock()
        self._ensure_dir()

    def log(self, event: SessionLogEvent) -> None:
        if self._log_dir is None:
            return
        opened = self._get_file(event.session_id)
        if opened is None:
            return
        handle, lock = opened
        line = json.dumps({
            "session_id": event.session_id,
            "event": event.event,
            "ts": event.ts,
            "payload": event.payload,
        })
        with lock, contextlib.suppress(ValueError):
            handle.write(line + "\n")
            handle.flush()

    def close(self, session_id: str) -> None:
        with self._registry_lock:
            handle = self._files.pop(session_id, None)
            lock = self._locks.pop(session_id, None)
        if handle is None or lock is None:
            return
        with lock, contextlib.suppress(Exception):
            handle.close()

    def close_all(self) -> None:
        with self._registry_lock:
            sids = list(self._files)
        for sid in sids:
            self.close(sid)

    def _ensure_dir(self) -> None:
        if self._log_dir is None:
            return
        os.makedirs(self._log_dir, exist_ok=True)

    def _get_file(self, session_id: str) -> tuple[TextIO, threading.Lock] | None:
        if self._log_dir is None:
            return None
        with self._registry_lock:
            handle = self._files.get(session_id)
            lock = self._locks.get(session_id)
            if handle is not None and lock is not None:
                return handle, lock
            # Defense-in-depth: session_id is server-generated UUID today,
            # but sanitize against path traversal in case future code paths
            # allow client-supplied ids.
            safe_id = _FILENAME_SANITIZE_RE.sub("_", session_id) or "unknown"
            path = os.path.join(
                self._log_dir,
                f"session-{safe_id}.jsonl",
            )
            try:
                handle = open(path, "a", encoding="utf-8")  # noqa: SIM115
            except OSError:
                return None
            lock = threading.Lock()
            self._files[session_id] = handle
            self._locks[session_id] = lock
            return handle, lock

fastvideo.entrypoints.streaming.session_logger.SessionLogger.log

log(event: SessionLogEvent) -> None
Source
def log(self, event: SessionLogEvent) -> None:
    if self._log_dir is None:
        return
    opened = self._get_file(event.session_id)
    if opened is None:
        return
    handle, lock = opened
    line = json.dumps({
        "session_id": event.session_id,
        "event": event.event,
        "ts": event.ts,
        "payload": event.payload,
    })
    with lock, contextlib.suppress(ValueError):
        handle.write(line + "\n")
        handle.flush()

fastvideo.entrypoints.streaming.session_logger.SessionLogger.close

close(session_id: str) -> None
Source
def close(self, session_id: str) -> None:
    with self._registry_lock:
        handle = self._files.pop(session_id, None)
        lock = self._locks.pop(session_id, None)
    if handle is None or lock is None:
        return
    with lock, contextlib.suppress(Exception):
        handle.close()

fastvideo.entrypoints.streaming.session_logger.SessionLogger.close_all

close_all() -> None
Source
def close_all(self) -> None:
    with self._registry_lock:
        sids = list(self._files)
    for sid in sids:
        self.close(sid)