STT-Assistant/src/stt_daemon/assistant/brief_file.py

"""Best-effort atomic writer for the durable brief markdown snapshot.

Enabled via `[brief].write = true` (off by default). When enabled, each brief
update replaces `<session_dir>/brief.md` with the exact text supplied by the
assistant, using the same `tempfile.mkstemp` + `os.replace` atomic pattern as
the rest of the codebase. `set_directory()` points the writer at the current
conversation's session folder; until then it defaults to `brief/brief.md`.

The target directory is cwd-relative user state and gitignored. Best-effort
by design: any `OSError` is logged and swallowed so the brief writer can never
disturb the assistant flow; `clear()` truncates to empty and likewise never
raises.

Thread-safe: the reader thread may call `set_directory()` while the inference
lane calls `write()`/`clear()`, so both are snapshotted under a lock and the
I/O then runs against the snapshotted locals only.
"""

import logging
import os
import tempfile
import threading

logger = logging.getLogger(__name__)


class BriefFile:
    """Best-effort, atomic, opt-in writer for the durable brief document."""

    def __init__(
        self, write_enabled: bool, directory: str = "brief", filename: str = "brief.md"
    ) -> None:
        self._write_enabled = bool(write_enabled)
        self._directory = str(directory)
        self._filename = str(filename)
        self._lock = threading.Lock()

    @property
    def write_enabled(self) -> bool:
        return self._write_enabled

    def set_directory(self, directory: str) -> None:
        """Point the writer at `directory` for subsequent writes (no I/O)."""
        with self._lock:
            self._directory = str(directory)

    def write(self, text: str) -> None:
        """Atomically replace the brief file with `text` (no-op when disabled)."""
        if not self._write_enabled:
            return
        with self._lock:
            directory = self._directory
            filename = self._filename
        path = os.path.join(directory, filename)
        try:
            os.makedirs(directory, exist_ok=True)
            fd, tmp = tempfile.mkstemp(dir=directory, suffix=".tmp")
            try:
                with os.fdopen(fd, "w", encoding="utf-8") as f:
                    f.write(text)
                os.replace(tmp, path)
            except OSError:
                try:
                    os.unlink(tmp)
                except OSError:
                    pass
                raise
        except OSError as e:  # noqa: BLE001 - brief write is best-effort, never raises
            logger.warning("brief file write failed: %s", e)

    def clear(self) -> None:
        """Truncate the brief file to empty (no-op when disabled; never raises)."""
        if not self._write_enabled:
            return
        with self._lock:
            directory = self._directory
            filename = self._filename
        path = os.path.join(directory, filename)
        try:
            with open(path, "w", encoding="utf-8"):
                pass
        except OSError as e:  # noqa: BLE001 - brief clear is best-effort, never raises
            logger.warning("brief file clear failed: %s", e)