Updated session control

README.md

@@ -29,7 +29,7 @@ Queue (segment_queue)
     ↓
 Faster-Whisper (transcription)
     ↓
-TranscriptSink (stdout + transcripts/YYYY-MM-DD.jsonl) + EventPublisher
+TranscriptSink (stdout + <session_dir>/<id>/transcripts.jsonl) + EventPublisher
     ↓  publish-only UNIX socket ($XDG_RUNTIME_DIR/stt-daemon.sock)
 stt-assistant  (subscribes; owns modes, command buffer, chat, LLM, tools)
     ↓  control socket ($XDG_RUNTIME_DIR/stt-assistant.sock)

config.example.toml

@@ -39,11 +39,11 @@ loopback_latency_msec = 1
 monitor_output        = false
 
 [output]
-stdout     = true
-jsonl      = false
-jsonl_dir  = "transcripts"
-save_audio = false
-audio_dir  = "audio"
+stdout          = true
+jsonl           = false
+session_dir     = "transcripts"
+save_audio      = false
+session_gap_min = 5.0
 
 [daemon]
 queue_size      = 50
@@ -133,9 +133,7 @@ store_dir         = "speakers"
 join_timeout_s    = 2.0
 
 [brief]
-write    = false
-dir      = "brief"
-filename = "brief.md"
+write = false
 
 [debug]
 enabled           = false

docs/ASSISTANT.md

@@ -187,8 +187,10 @@ LLM lane up to `compaction_timeout_s`. The deployment thresholds (`escalate_thre
 The panel renders the last-triage clock (transcript time format/color) and both stages; it is
 display-only and triggers no modes, commands, or tools. Leaving chat mode does not clear the panel or
 its buffer; an explicit `clear` does (it also truncates the durable brief file). With
-`[brief].write = true` the latest brief is snapshotted to `dir/filename` (default `brief/brief.md`,
-gitignored); with the default `write = false` it stays in the TUI only. With `dump_io = true` each
+`[brief].write = true` the latest brief is snapshotted to `<session_dir>/<id>/brief.md` inside the
+active conversation folder (gitignored, falling back to `brief/brief.md` when no session is active);
+with the default `write = false` it stays in the TUI only. A new conversation resets the brief, so
+each session folder gets a fresh one. With `dump_io = true` each
 triage also writes its I/O under `triage/` (a coalesced update writes two files; see
 [TRIAGE.md](TRIAGE.md)).
 

docs/CLI.md

@@ -122,8 +122,9 @@ from a live voice. `--name` is required; pass `--wav` once per clip, and each fi
 contributes one prototype (capped at `max_prototypes`, oldest evicted). On success the
 command prints something like `enrolled 'Efren' as spk_0002 (2 clip(s), 7.4s)`.
 
-Each WAV must be 16 kHz mono 16-bit PCM; the clips the daemon saves under `audio/` already
-are, so they need no conversion, and each must be at least `[speaker].min_enroll_s` seconds.
+Each WAV must be 16 kHz mono 16-bit PCM; the clips the daemon saves under
+`<session_dir>/<id>/audio/` are already in that format, so they need no conversion, and each
+must be at least `[speaker].min_enroll_s` seconds.
 The store is single-writer and owned by the daemon, so stop it before enrolling, or pass
 `--force` to bypass the refusal. Like every subcommand, `speaker-enroll` also accepts
 `--config`.
@@ -148,7 +149,8 @@ uv run stt-daemon speaker-enroll-seed --name "Efren Yevale Varela" --seed audio/
 ```
 
 Enrolls from ONE clip you are certain is the speaker. The command scans `--audio-dir` (default
-`[output].audio_dir`), embeds each `*.wav`, ranks them by cosine similarity to the seed, and offers
+`[output].session_dir`, the session root `transcripts/`, searched recursively for `*.wav`), embeds
+each `*.wav`, ranks them by cosine similarity to the seed, and offers
 the ones at or above `--min-score` (default `0.55`) and at least `[speaker].min_enroll_s` seconds
 for review. `--name` and `--seed` are required; the seed must be 16 kHz mono 16-bit PCM.
 

docs/CONFIGURATION.md

@@ -63,11 +63,11 @@ loopback_latency_msec = 1                      # Module latency (ms)
 monitor_output        = false                  # Loop sink.monitor → default sink (OFF = no feedback risk)
 
 [output]
-stdout     = true                     # Print transcripts to stdout
-jsonl      = false                    # Persist transcripts to JSONL files (off by default for agent-only use)
-jsonl_dir  = "transcripts"            # JSONL destination (one file per day); only used if jsonl = true
-save_audio = false                    # Also write each transcribed utterance as a 16 kHz mono WAV
-audio_dir  = "audio"                  # WAV destination (YYYY-MM-DD subfolders); only used if save_audio
+stdout          = true               # Print transcripts to stdout
+jsonl           = false              # Persist transcripts to JSONL files (off by default for agent-only use)
+session_dir     = "transcripts"      # Conversation root: one folder per conversation, named by the first transcription's YYYY-MM-DD-HH-MM-SS
+save_audio      = false              # Also write each transcribed utterance as a 16 kHz mono WAV
+session_gap_min = 5.0                # Quiet gap in minutes before a new conversation starts; <= 0 disables rollover
 
 [daemon]
 queue_size      = 50                        # Max utterances queued for transcription (oldest dropped when full)
@@ -157,9 +157,7 @@ store_dir         = "speakers"               # Store root (cwd-relative); gitign
 join_timeout_s    = 2.0                      # Embedding-lane join timeout before emitting the transcript
 
 [brief]                                      # Durable brief artifact (plain-text markdown snapshot; off by default)
-write    = false                            # false = brief shown in the TUI only; true = also persist to disk
-dir      = "brief"                          # Destination directory (cwd-relative); gitignored
-filename = "brief.md"                       # Snapshot filename inside dir
+write = false                                # false = brief shown in the TUI only; true = also persist under the conversation folder
 
 [debug]                                     # Structured health-signal log (no transcript text; off by default)
 enabled          = false                    # false = no debug lines, zero overhead (DebugSink is None)
@@ -246,8 +244,12 @@ more), and the server rejects an out-of-range slot.
 
 `[brief]` controls persistence of that living brief. With `write = false` (the default) it stays
 display-only: the TUI renders it and nothing is written to disk. With `write = true` the
-assistant snapshots the current brief as plain text to `dir/filename` (default
-`brief/brief.md`, cwd-relative, gitignored); clearing the brief truncates the file.
+assistant snapshots the current brief as plain text to
+`<session_dir>/<YYYY-MM-DD-HH-MM-SS>/brief.md` (cwd-relative, gitignored): one folder per
+conversation, named by the first transcription's `YYYY-MM-DD-HH-MM-SS`, holding that
+conversation's transcript JSONL, audio, brief, and screenshots, while audit/debug/triage
+remain global. Starting a new conversation resets the brief; clearing the brief truncates
+the file.
 
 `[debug]` persists per-component health signals to `debug/YYYY-MM-DD.jsonl` (one JSON
 object per line, `ensure_ascii=False`) so the next silent freeze is diagnosable from disk.

docs/OUTPUT.md

@@ -2,6 +2,11 @@
 
 Transcripts reach you three ways: human-readable stdout, machine-readable JSONL, and optional WAV clips.
 
+Each conversation writes to its own folder under `session_dir` (default `transcripts/`), named
+`<YYYY-MM-DD-HH-MM-SS>` after its first transcription. The transcript log, saved WAVs, the optional
+brief, and TUI screenshots all live inside it. A new folder starts once `session_gap_min` minutes
+pass without speech, or on a fresh process start.
+
 **Stdout** (human-readable):
 ```
 [14:23:45] (es) Hola, cómo estás?  [rms=0.041 end_of_turn]
@@ -9,12 +14,17 @@ Transcripts reach you three ways: human-readable stdout, machine-readable JSONL,
 [14:24:10] (es) Gracias por tu ayuda.  [rms=0.052 pause]
 ```
 
-**JSONL** (one line per utterance; only written when `jsonl = true`):
+**JSONL** (one line per utterance; only written when `jsonl = true`, at
+`<session_dir>/<id>/transcripts.jsonl`):
 ```json
-{"ts": "2025-09-20T14:23:45.123456", "lang": "es", "lang_prob": 0.95, "duration_s": 2.5, "rms": 0.04123, "end_reason": "end_of_turn", "text": "Hola, cómo estás?", "transcribe_s": 0.42}
-{"ts": "2025-09-20T14:23:52.654321", "lang": "en", "lang_prob": 0.98, "duration_s": 1.8, "rms": 0.03811, "end_reason": "pause", "text": "What time is it?", "transcribe_s": 0.31}
+{"ts": "2025-09-20T14:23:45.123456", "session_id": "2025-09-20-14-23-45", "lang": "es", "lang_prob": 0.95, "duration_s": 2.5, "rms": 0.04123, "end_reason": "end_of_turn", "text": "Hola, cómo estás?", "transcribe_s": 0.42}
+{"ts": "2025-09-20T14:23:52.654321", "session_id": "2025-09-20-14-23-45", "lang": "en", "lang_prob": 0.98, "duration_s": 1.8, "rms": 0.03811, "end_reason": "pause", "text": "What time is it?", "transcribe_s": 0.31}
 ```
 
+`session_id` names the conversation folder (`<session_dir>/<session_id>/`) so records can be
+correlated back to their transcript log and audio. Records from one conversation all carry the same
+value; a new folder gets a new one.
+
 `duration_s` is the length of the utterance audio; `transcribe_s` is the wall-clock
 time the worker spent transcribing it (inference latency), rounded to milliseconds.
 Their ratio is the real-time factor. The field is omitted if the timing was not
@@ -45,10 +55,10 @@ speaker fields (always present; `null` when unknown):
 Existing tooling that ignores unknown JSON keys keeps working unchanged.
 
 **Audio files** — when `save_audio = true`, each transcribed utterance is also
-written as a 16 kHz mono WAV under `audio_dir/YYYY-MM-DD/`, and the record gains
+written as a 16 kHz mono WAV under `<session_dir>/<id>/audio/`, and the record gains
 an `"audio"` field with its path:
 ```json
-{..., "end_reason": "pause", "text": "Hola", "audio": "audio/2025-09-20/142345-123-pause.wav"}
+{..., "end_reason": "pause", "text": "Hola", "audio": "transcripts/2025-09-20-14-23-45/audio/142345-123-pause.wav"}
 ```
 Saving is best-effort: if a write fails (e.g. disk full), the transcript is still
 emitted and a warning is logged.
@@ -57,7 +67,7 @@ emitted and a warning is logged.
 > At 16 kHz mono 16-bit it costs 32 KB/s, i.e. **~1.9 MB/min, ~115 MB/hour,
 > ~2.8 GB/day** of *continuous speech*. Silence is not written, so real usage is
 > lower — but an always-on daemon in a talkative room still accumulates quickly.
-> Point `audio_dir` at a volume with headroom and prune it yourself (e.g. a cron
+> Point `session_dir` at a volume with headroom and prune it yourself (e.g. a cron
 > job removing files older than N days), and leave `save_audio = false` unless you
 > actually need the clips for auditing.
 

docs/SPEAKERS.md

@@ -167,7 +167,7 @@ is the same global override as every other subcommand.
 
 Each WAV must be **16 kHz mono 16-bit PCM**. No resampling is performed: stereo, a different
 sample rate, or a non-WAV file is rejected with a clear error. The clips the daemon saves
-under `audio/YYYY-MM-DD/` when `[output].save_audio = true` are already in that format, so
+under `<session_dir>/<YYYY-MM-DD-HH-MM-SS>/audio/` when `[output].save_audio = true` are already in that format, so
 they need no conversion. Each clip must also be at least `[speaker].min_enroll_s` seconds
 (default `2.0`); shorter clips are rejected. This is the "is it enough audio?" check.
 
@@ -205,7 +205,7 @@ Flags:
 - `--seed` (required) — a 16 kHz mono 16-bit WAV known to be that speaker.
 - `--count N` (default `5`) — how many reviewed clips to accept.
 - `--min-score F` (default `0.55`) — minimum cosine to the seed for a clip to be offered.
-- `--audio-dir DIR` (default `[output].audio_dir`) — where to scan for `*.wav`.
+- `--audio-dir DIR` (default `[output].session_dir`, scanned recursively) — where to scan for `*.wav`.
 - `--max-scan N` (default `0` = all) — cap the number of files embedded, useful on large dirs.
 - `--dedupe-threshold F` (default `0.98`) — cosine at or above which a candidate counts as already
   covered by one of that speaker's prototypes.

docs/TRIAGE.md

@@ -98,7 +98,9 @@ settled turn (any mode)
    buffer, drops any pending lane request and any queued `brief-update` job, truncates the
    optional durable brief file, and emits the `cleared` stage. An in-flight result that
    lands after the bump is suppressed, so a stale brief writes neither the pane nor the
-   file. Leaving chat mode does not reset the panel or its buffer.
+   file. Leaving chat mode does not reset the panel or its buffer. A new conversation also
+   resets the brief, which starts fresh in the new session folder, without clearing chat or
+   command state.
 
 ## Single-outstanding job and coalescing
 
@@ -297,14 +299,16 @@ own history is compacted.
 
 `[brief]` is a separate section that controls durability of the living brief. `write`
 (default `false`) gates disk persistence only: with it false the brief stays in the TUI, and
-with `write = true` each accepted brief is snapshotted to `dir/filename` (default
-`brief/brief.md`, cwd-relative and gitignored) and truncated on `clear`.
+with `write = true` each accepted brief is snapshotted to `<session_dir>/<id>/brief.md` inside
+the active conversation folder (falling back to `brief/brief.md`, cwd-relative and gitignored,
+when no session is active) and truncated on `clear`. A new conversation resets the brief, so
+each session folder gets a fresh one.
 
 ## Opt-in I/O dump (`dump_io`)
 
 With `[triage].dump_io = true` (default `false`) each triage writes a self-contained JSON
 file under `triage/YYYY-MM-DD/HHMMSS-mmm.json` — local time with millisecond precision, the
-same naming scheme as the audio files (`audio/YYYY-MM-DD/HHMMSS-mmm-<reason>.wav`). A file
+same naming scheme as the audio files (`<session_dir>/<id>/audio/HHMMSS-mmm-<reason>.wav`). A file
 holds the Laya request, the Laya response, the LLM request (endpoint URL plus the full
 chat-completion body), and the LLM response (raw envelope), keyed by the `ts` carried on the
 triage events:
@@ -329,7 +333,7 @@ record overwrites another.
 When the brief does not run (the master switch is off, the gate fails, or no brief function
 is wired) the file still gets written with only the `laya` half (`llm` is `null`); on a Laya
 error the file records `laya.response = null` with `laya.error = "predict failed"`. The
-directory is fixed as `triage/` (cwd-relative, like `transcripts/`/`audio/`/`audit/`) —
+directory is fixed as `triage/` (cwd-relative, like the session folders and `audit/`) —
 there is no second config key for it. Files contain conversation text, so they are
 gitignored private debug output. Only the triage Laya predict and the triage brief call are
 dumped (never the loader warmup predict or non-triage LLM calls), and dumping is best-effort:

src/stt_daemon/assistant/brief_file.py

@@ -1,19 +1,25 @@
 """Best-effort atomic writer for the durable brief markdown snapshot.
 
 Enabled via `[brief].write = true` (off by default). When enabled, each brief
-update replaces `[brief].dir`/`[brief].filename` (default `brief/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.
+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 `brief/` directory is cwd-relative user state and gitignored. Best-effort
+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__)
 
@@ -21,23 +27,34 @@ 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, filename: str = "brief.md") -> None:
+    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
-        path = os.path.join(self._directory, self._filename)
+        with self._lock:
+            directory = self._directory
+            filename = self._filename
+        path = os.path.join(directory, filename)
         try:
-            os.makedirs(self._directory, exist_ok=True)
-            fd, tmp = tempfile.mkstemp(dir=self._directory, suffix=".tmp")
+            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)
@@ -55,7 +72,10 @@ class BriefFile:
         """Truncate the brief file to empty (no-op when disabled; never raises)."""
         if not self._write_enabled:
             return
-        path = os.path.join(self._directory, self._filename)
+        with self._lock:
+            directory = self._directory
+            filename = self._filename
+        path = os.path.join(directory, filename)
         try:
             with open(path, "w", encoding="utf-8"):
                 pass

src/stt_daemon/assistant/runner.py

@@ -95,7 +95,7 @@ def main() -> int:
         client if cfg.llm.enabled else None,
         on_result=lambda gen, payload: holder["svc"]._on_triage_result(gen, payload),
     )
-    brief_file = BriefFile(cfg.brief.write, cfg.brief.dir, cfg.brief.filename)
+    brief_file = BriefFile(cfg.brief.write)
     _debug_factory = None
     debug_cfg = getattr(cfg, "debug", None)
     if debug_cfg is not None and bool(getattr(debug_cfg, "enabled", False)):

src/stt_daemon/assistant/service.py

@@ -43,7 +43,13 @@ HEARTBEAT_EVENT = "heartbeat"
 CAPTURING = "capturing"
 READY = "ready"
 EXECUTING = "executing"
-RELAYED_TYPES = (events.TRANSCRIPT, events.SPEECH_START, events.SPEECH_STOP, events.SPEAKERS)
+RELAYED_TYPES = (
+    events.TRANSCRIPT,
+    events.SPEECH_START,
+    events.SPEECH_STOP,
+    events.SPEAKERS,
+    events.SESSION,
+)
 
 _TRIAGE_HISTORY_MAX = 200
 
@@ -119,6 +125,15 @@ class AssistantService:
         # Speaker roster cache (display-name join only; the daemon owns identity).
         self._speakers: dict[str, dict] = {}
 
+        # Conversation session directory, adopted verbatim from SESSION/TRANSCRIPT
+        # events (never recomputed locally). The bus seeds every connection
+        # `status -> ring -> ring_end`, so STATUS opens an explicit replay window
+        # and RING_END commits exactly the newest session observed inside it.
+        self._session_dir: str | None = None
+        self._replaying = False
+        self._replay_first: str | None = None
+        self._replay_newest: str | None = None
+
         self._triage = triage
         self._laya_lane = laya_lane
         self._brief_fn = brief_fn
@@ -294,6 +309,7 @@ class AssistantService:
             "command_text": self.command_text,
             "speech_open": self.speech_open,
             "awaited_seg_id": self.awaited_seg_id,
+            "session_dir": self._session_dir,
             "chat_turns": len(self.chat_history),
             "inference_depth": self._inference.depth(),
             "inference_in_flight": self._inference.in_flight(),
@@ -584,7 +600,12 @@ class AssistantService:
             self._relay(event)
 
         if etype == events.STATUS:
+            self._replaying = True
+            self._replay_first = None
+            self._replay_newest = None
             self._on_status(now)
+        elif etype == events.RING_END:
+            self._commit_replay()
         elif etype == events.SPEECH_START:
             self.speech_open = True
             self._last_activity_ts = now
@@ -597,7 +618,10 @@ class AssistantService:
                 self.awaited_seg_id = None
             self._last_activity_ts = now
         elif etype == events.TRANSCRIPT:
+            self._observe_session(event)
             self._on_transcript(event, now)
+        elif etype == events.SESSION:
+            self._observe_session(event)
         elif etype == events.HEARTBEAT:
             self._last_heartbeat_ts = now
             self._worker_stall_s = float(event.get("worker_stall_s", 0.0) or 0.0)
@@ -607,6 +631,84 @@ class AssistantService:
         self._check_timeouts(now)
         self._emit_command_text()
 
+    @staticmethod
+    def _session_value(event: dict) -> str | None:
+        value = event.get("session_dir")
+        if isinstance(value, str) and value:
+            return value
+        return None
+
+    def _observe_session(self, event: dict) -> None:
+        """Adopt a live session, or buffer one replayed inside the ring window."""
+        value = self._session_value(event)
+        if not value:
+            return
+        if self._replaying:
+            if self._replay_first is None:
+                self._replay_first = value
+            self._replay_newest = value
+        else:
+            self._apply_session(value)
+
+    def _commit_replay(self) -> None:
+        """Close the replay window and adopt its newest session, if any."""
+        if not self._replaying:
+            return
+        first = self._replay_first
+        newest = self._replay_newest
+        self._replaying = False
+        self._replay_first = None
+        self._replay_newest = None
+        if newest is not None:
+            window_changed = first is not None and first != newest
+            self._apply_session(newest, window_changed=window_changed)
+
+    def _apply_session(self, value: str, window_changed: bool = False) -> None:
+        """Adopt `value` as the conversation directory, resetting when it changes.
+
+        A first-ever adoption (`previous is None`) never resets. At a RING_END
+        commit whose ring spanned an older session, `window_changed` forces the
+        reset even when `value` equals the current directory, so a reconnect
+        cannot leave the older session's replayed turns buffered. Never raises:
+        any failure is logged and swallowed.
+        """
+        if not value:
+            return
+        if value == self._session_dir and not window_changed:
+            return
+        try:
+            previous = self._session_dir
+            if value != self._session_dir:
+                self._session_dir = value
+                if self._brief_file is not None:
+                    self._brief_file.set_directory(value)
+            if (previous is not None and previous != value) or window_changed:
+                self._reset_for_session()
+        except Exception as e:  # noqa: BLE001 - nothing here may break the reader loop
+            logger.warning("session adoption failed: %s", e)
+
+    def _reset_for_session(self) -> None:
+        """Reset per-conversation triage/brief accumulation for a new session.
+
+        Mirrors `clear()` without the chat/command state and without truncating
+        the durable brief, so the just-started session's fresh `<dir>/brief.md`
+        is preserved. Emits one triage `cleared` only when triage is active.
+        """
+        with self._lock:
+            self._reset_triage()
+            self._pending_brief = []
+            self._triage_dump.clear()
+        if self._laya_lane is not None:
+            self._laya_lane.drop_pending()
+        self._inference.drop(lambda j: j.name == "brief-update")
+        if self._brief_reset is not None:
+            try:
+                self._brief_reset()
+            except Exception as e:  # noqa: BLE001 - clearing brief must not fail
+                logger.warning("brief reset failed: %s", e)
+        if self._triage is not None or self._laya_lane is not None:
+            self._emit(TRIAGE_EVENT, stage="cleared", triage=None)
+
     def tick(self) -> None:
         """Evaluate timeouts, dispatch pending chat, and emit a heartbeat."""
         now = self._now()

src/stt_daemon/assistant/triage_dump.py

@@ -3,10 +3,11 @@
 Enabled via `[triage].dump_io = true` (off by default). When enabled, each
 triage handled in `_on_triage_result` produces exactly one file under
 `triage/YYYY-MM-DD/HHMMSS-mmm.json` (local time, millisecond precision — the
-same naming scheme as `audio/YYYY-MM-DD/HHMMSS-mmm-<reason>.wav`).
+same naming scheme as the session audio
+`<session_dir>/<id>/audio/HHMMSS-mmm-<reason>.wav`).
 
-The directory `triage/` is fixed (cwd-relative, like `transcripts/`/`audio/`/
-`audit/`) and gitignored. Files contain conversation text (transcript state,
+The directory `triage/` is fixed (cwd-relative, like `transcripts/`/`audit/`)
+and gitignored. Files contain conversation text (transcript state,
 Laya questions/answers, LLM prompts), so treat them as private debug output.
 
 Best-effort by design: a dump failure is logged and swallowed, never raised,

src/stt_daemon/cli.py

@@ -230,7 +230,7 @@ def main() -> int:
         "--audio-dir",
         type=str,
         default=None,
-        help="Directory to scan for WAV clips (default: [output] audio_dir)",
+        help="Directory to scan for WAV clips (default: [output] session_dir, scanned recursively)",
     )
     seed_parser.add_argument(
         "--max-scan",

src/stt_daemon/config.py

@@ -74,9 +74,9 @@ class OutputConfig:
     """Output configuration."""
     stdout: bool = True
     jsonl: bool = False
-    jsonl_dir: str | None = "transcripts"
+    session_dir: str = "transcripts"
     save_audio: bool = False
-    audio_dir: str = "audio"
+    session_gap_min: float = 5.0
 
 
 @dataclass
@@ -220,8 +220,6 @@ class SpeakerConfig:
 class BriefConfig:
     """Durable brief document configuration (markdown snapshot written to disk)."""
     write: bool = False
-    dir: str = "brief"
-    filename: str = "brief.md"
 
 
 @dataclass
@@ -379,11 +377,13 @@ def load_config(path: str | None = None) -> Config:
         if "output" in data:
             output_data = data["output"]
             cfg.output = OutputConfig(
-        stdout=output_data.get("stdout", cfg.output.stdout),
-        jsonl=output_data.get("jsonl", cfg.output.jsonl),
-        jsonl_dir=output_data.get("jsonl_dir", cfg.output.jsonl_dir),
+                stdout=output_data.get("stdout", cfg.output.stdout),
+                jsonl=output_data.get("jsonl", cfg.output.jsonl),
+                session_dir=output_data.get("session_dir", cfg.output.session_dir),
                 save_audio=output_data.get("save_audio", cfg.output.save_audio),
-                audio_dir=output_data.get("audio_dir", cfg.output.audio_dir),
+                session_gap_min=output_data.get(
+                    "session_gap_min", cfg.output.session_gap_min
+                ),
             )
 
         # Sink section
@@ -612,8 +612,6 @@ def load_config(path: str | None = None) -> Config:
             brief_data = data["brief"]
             cfg.brief = BriefConfig(
                 write=brief_data.get("write", cfg.brief.write),
-                dir=brief_data.get("dir", cfg.brief.dir),
-                filename=brief_data.get("filename", cfg.brief.filename),
             )
 
         # Debug section

src/stt_daemon/daemon.py

@@ -23,6 +23,7 @@ from .bus import EventPublisher
 from .config import Config
 from .output import TranscriptSink
 from .segmenter import FrameSegmenter, Utterance, make_vad
+from .session import SessionTracker
 from .speakers import SpeakerStore
 from .transcriber import TranscriberLike, build_transcriber
 
@@ -292,6 +293,10 @@ class Daemon:
         self._last_segment_done_ts = 0.0
         self._last_worker_activity_ts = 0.0
         self._last_transcript_ts = 0.0
+        # Session identity is minted here (in __init__, so it exists even when
+        # tests inject sink_factory and bypass _initialize) and advanced only
+        # in _process once a non-empty transcript is observed.
+        self._session_tracker = SessionTracker(self.cfg.output.session_gap_min * 60.0)
         self._in_flight = False
         self._frames_since_heartbeat = 0
         self._started_at = 0.0
@@ -317,10 +322,8 @@ class Daemon:
     def _build_sink(self) -> TranscriptSink:
         return TranscriptSink(
             stdout=self.cfg.output.stdout,
-            jsonl_dir=self.cfg.output.jsonl_dir,
             jsonl=self.cfg.output.jsonl,
             save_audio=self.cfg.output.save_audio,
-            audio_dir=self.cfg.output.audio_dir,
             sample_rate=self.cfg.audio.sample_rate,
         )
 
@@ -748,6 +751,21 @@ class Daemon:
                     speaker_id, speaker_status, speaker, speaker_score = (
                         self._resolve_speaker(speaker_future)
                     )
+                sid: str | None = None
+                session_dir: Path | None = None
+                now_epoch = 0.0
+                if result.text:
+                    prev = self._session_tracker.current_id
+                    now_epoch = time.time()
+                    sid = self._session_tracker.observe(now_epoch)
+                    session_dir = Path(self.cfg.output.session_dir) / sid
+                    if sid != prev:
+                        self._publish(
+                            events.SESSION,
+                            session_id=sid,
+                            session_dir=str(session_dir),
+                            ts=now_epoch,
+                        )
                 try:
                     audio_path = self.sink.emit(
                         result,
@@ -761,6 +779,7 @@ class Daemon:
                         speaker_status=speaker_status,
                         speaker=speaker,
                         speaker_score=speaker_score,
+                        session_dir=session_dir,
                     )
                 except Exception as e:
                     # A sink failure must not drop the transcript from the bus.
@@ -768,7 +787,7 @@ class Daemon:
                     audio_path = None
                 if result.text:
                     emitted = True
-                    self._last_transcript_ts = time.time()
+                    self._last_transcript_ts = now_epoch
                     self._publish(
                         events.TRANSCRIPT,
                         seg_id=seg_id,
@@ -783,6 +802,8 @@ class Daemon:
                         speaker_status=speaker_status,
                         speaker=speaker,
                         speaker_score=speaker_score,
+                        session_id=sid,
+                        session_dir=str(session_dir),
                     )
         except Exception as e:
             self.logger.error(f"Transcription error: {e}")

src/stt_daemon/enroll_assist.py

@@ -403,7 +403,7 @@ def seed_enroll(cfg, embedder, store, name, seed_path, *, count=5,
     if not seed_ok:
         print(f"warning: seed {seed_path.name} is {seed_duration_s:.2f}s "
               f"< {min_enroll_s:.2f}s; used as query only, not enrolled")
-    base = Path(audio_dir) if audio_dir is not None else Path(cfg.output.audio_dir)
+    base = Path(audio_dir) if audio_dir is not None else Path(cfg.output.session_dir)
     scanned = sorted(p for p in base.rglob("*.wav") if p.is_file()) if base.exists() else []
     seed_resolved = _resolve(seed_path)
     files = [p for p in scanned if _resolve(p) != seed_resolved]

src/stt_daemon/events.py

@@ -13,6 +13,7 @@ HEARTBEAT = "heartbeat"
 STATUS = "status"
 RING_END = "ring_end"
 SPEAKERS = "speakers"
+SESSION = "session"
 
 ALL_TYPES = frozenset(
     {
@@ -24,6 +25,7 @@ ALL_TYPES = frozenset(
         STATUS,
         RING_END,
         SPEAKERS,
+        SESSION,
     }
 )
 

src/stt_daemon/output.py

@@ -29,29 +29,27 @@ class TranscriptSink:
     def __init__(
         self,
         stdout: bool = True,
-        jsonl_dir: str | None = None,
         jsonl: bool = False,
         save_audio: bool = False,
-        audio_dir: str = "audio",
         sample_rate: int = 16000,
     ):
         """Initialize output sinks.
 
         Args:
             stdout: If True, emit to stdout.
-            jsonl_dir: Directory for daily JSONL files.
-            jsonl: If True (and jsonl_dir is set), append JSONL lines to daily
-                files in that directory. Defaults off so agent-only runs do not
-                persist transcripts to disk.
-            save_audio: If True, write each transcript's audio as a WAV file.
-            audio_dir: Directory for WAV files (dated subfolders).
+            jsonl: If True, append JSONL lines to `transcripts.jsonl` under the
+                session directory supplied per emit(). Defaults off so
+                agent-only runs do not persist transcripts to disk.
+            save_audio: If True, write each transcript's audio as a WAV file
+                under the session directory supplied per emit().
             sample_rate: Audio sample rate, written into the WAV header.
+
+        The session directory is resolved by the caller and passed per emit();
+        the sink owns no root or day path of its own.
         """
         self.stdout = stdout
         self.jsonl = jsonl
-        self.jsonl_dir = Path(jsonl_dir) if (jsonl and jsonl_dir) else None
         self.save_audio = save_audio
-        self.audio_dir = Path(audio_dir)
         self.sample_rate = sample_rate
 
     def emit(
@@ -63,6 +61,7 @@ class TranscriptSink:
         audio: np.ndarray | None = None,
         transcribe_s: float | None = None,
         *,
+        session_dir: Path | None = None,
         seg_id: int | None = None,
         speaker_id: str | None = None,
         speaker_status: str = "none",
@@ -79,6 +78,8 @@ class TranscriptSink:
             audio: Utterance samples (float32, mono); saved when save_audio is enabled.
             transcribe_s: Wall-clock seconds spent transcribing this utterance.
                 None omits the field from the record.
+            session_dir: Resolved per-conversation directory. Required for JSONL
+                and WAV output; when None those writes are skipped.
             seg_id: Stable correlation key for the segment.
             speaker_id: Stable cluster key for cross-reference (None when unknown).
             speaker_status: matched | uncertain | unknown | none.
@@ -95,10 +96,12 @@ class TranscriptSink:
         # WAV path must agree even when emit() straddles midnight.
         now = datetime.now()
 
+        base = Path(session_dir) if session_dir is not None else None
+
         audio_path = None
-        if self.save_audio and audio is not None:
+        if self.save_audio and audio is not None and base is not None:
             try:
-                audio_path = self._save_audio(audio, reason, now)
+                audio_path = self._save_audio(audio, reason, now, base)
             except Exception as e:
                 # Best-effort side output: a failed save must not drop the transcript.
                 logger.warning("Failed to save audio for this utterance: %s", e)
@@ -109,44 +112,53 @@ class TranscriptSink:
             line = f"[{timestamp}] ({result.language}) {result.text}  [rms={rms:.3f} {reason}]"
             print(line, flush=True)
 
-        # JSONL file per day
-        if self.jsonl_dir:
-            self.jsonl_dir.mkdir(parents=True, exist_ok=True)
-            today = now.strftime("%Y-%m-%d")
-            jsonl_path = self.jsonl_dir / f"{today}.jsonl"
-
-            record = {
-                "ts": now.isoformat(),
-                "lang": result.language,
-                "lang_prob": result.language_probability,
-                "duration_s": duration_s,
-                "rms": round(rms, 5),
-                "end_reason": reason,
-                "text": result.text,
-                "seg_id": seg_id,
-                "speaker_id": speaker_id,
-                "speaker_status": speaker_status,
-                "speaker": speaker,
-                "speaker_score": round(speaker_score, 4) if speaker_score is not None else None,
-            }
-            if audio_path is not None:
-                record["audio"] = str(audio_path)
-            if transcribe_s is not None:
-                record["transcribe_s"] = round(transcribe_s, 3)
-
-            # Serialize before opening: a failure here must not leave a partial
-            # line in the JSONL file for the next record to append onto.
-            line = json.dumps(record, ensure_ascii=False)
-            with open(jsonl_path, "a", encoding="utf-8") as f:
-                f.write(line + "\n")
+        # JSONL file per conversation session
+        if self.jsonl:
+            if base is None:
+                logger.warning(
+                    "jsonl output enabled but no session_dir was provided; skipping"
+                )
+            else:
+                base.mkdir(parents=True, exist_ok=True)
+                jsonl_path = base / "transcripts.jsonl"
+
+                record = {
+                    "ts": now.isoformat(),
+                    "session_id": base.name,
+                    "lang": result.language,
+                    "lang_prob": result.language_probability,
+                    "duration_s": duration_s,
+                    "rms": round(rms, 5),
+                    "end_reason": reason,
+                    "text": result.text,
+                    "seg_id": seg_id,
+                    "speaker_id": speaker_id,
+                    "speaker_status": speaker_status,
+                    "speaker": speaker,
+                    "speaker_score": (
+                        round(speaker_score, 4) if speaker_score is not None else None
+                    ),
+                }
+                if audio_path is not None:
+                    record["audio"] = str(audio_path)
+                if transcribe_s is not None:
+                    record["transcribe_s"] = round(transcribe_s, 3)
+
+                # Serialize before opening: a failure here must not leave a partial
+                # line in the JSONL file for the next record to append onto.
+                line = json.dumps(record, ensure_ascii=False)
+                with open(jsonl_path, "a", encoding="utf-8") as f:
+                    f.write(line + "\n")
 
         return str(audio_path) if audio_path is not None else None
 
-    def _save_audio(self, audio: np.ndarray, reason: str, now: datetime) -> Path:
-        day_dir = self.audio_dir / now.strftime("%Y-%m-%d")
-        day_dir.mkdir(parents=True, exist_ok=True)
+    def _save_audio(
+        self, audio: np.ndarray, reason: str, now: datetime, base: Path
+    ) -> Path:
+        audio_dir = base / "audio"
+        audio_dir.mkdir(parents=True, exist_ok=True)
         name = f"{now.strftime('%H%M%S')}-{now.strftime('%f')[:3]}-{reason}.wav"
-        path = day_dir / name
+        path = audio_dir / name
         write_wav(path, audio, self.sample_rate)
         return path
 

src/stt_daemon/session.py

@@ -0,0 +1,57 @@
+"""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

src/stt_daemon/tui/app.py

@@ -315,7 +315,11 @@ class AssistantTUI(App):
 
     def action_screenshot(self) -> None:
         name = "stt-tui-%s.svg" % datetime.now().strftime("%Y%m%d-%H%M%S")
-        out_dir = "screenshots"
+        session_dir = getattr(self.state, "session_dir", None)
+        if session_dir:
+            out_dir = os.path.join(session_dir, "screenshots")
+        else:
+            out_dir = "screenshots"
         try:
             os.makedirs(out_dir, exist_ok=True)
             save = getattr(self, "save_screenshot", None)

src/stt_daemon/tui/state.py

@@ -106,6 +106,7 @@ class ViewState:
     questions: dict | None = None
     questions_epoch: int = 0
     speakers: list[dict] = field(default_factory=list)
+    session_dir: str | None = None
 
     def ingest(self, event: dict) -> None:
         """Fold one event into the view state."""
@@ -137,6 +138,9 @@ class ViewState:
                 # Static config from the analyzer: independent of the triage epoch.
                 self.questions = event["questions"]
                 self.questions_epoch += 1
+            self.session_dir = event.get("session_dir") or self.session_dir
+        elif etype == "session":
+            self.session_dir = event.get("session_dir") or self.session_dir
         elif etype == "triage":
             stage = event.get("stage")
             if stage == "cleared":
@@ -174,6 +178,7 @@ class ViewState:
         elif etype == "speakers":
             self.speakers = list(event.get("speakers") or [])
         elif etype == "transcript":
+            self.session_dir = event.get("session_dir") or self.session_dir
             self.transcripts.append(event)
             del self.transcripts[:-MAX_TRANSCRIPTS]
             self.transcripts_total += 1

tests/test_assistant_integration.py

@@ -415,8 +415,6 @@ def test_runner_brief_file_writes_when_enabled(monkeypatch, tmp_path):
 
     cfg = _enabled_triage_cfg(tmp_path)
     cfg.brief.write = True
-    cfg.brief.dir = str(tmp_path / "brief")
-    cfg.brief.filename = "b.md"
 
     rc, _, _, made_services = _run_runner_main(monkeypatch, cfg)
 
@@ -424,8 +422,10 @@ def test_runner_brief_file_writes_when_enabled(monkeypatch, tmp_path):
     brief_file = made_services[0].kwargs["brief_file"]
     assert isinstance(brief_file, BriefFile)
     assert brief_file.write_enabled is True
+    session_dir = str(tmp_path / "transcripts" / "s1")
+    brief_file.set_directory(session_dir)
     brief_file.write("latest brief")
-    path = os.path.join(cfg.brief.dir, "b.md")
+    path = os.path.join(session_dir, "brief.md")
     with open(path, encoding="utf-8") as handle:
         assert handle.read() == "latest brief"
 
@@ -436,7 +436,6 @@ def test_runner_brief_file_disabled_writes_nothing(monkeypatch, tmp_path):
 
     cfg = _enabled_triage_cfg(tmp_path)
     cfg.brief.write = False
-    cfg.brief.dir = str(tmp_path / "brief")
 
     rc, _, _, made_services = _run_runner_main(monkeypatch, cfg)
 
@@ -444,8 +443,10 @@ def test_runner_brief_file_disabled_writes_nothing(monkeypatch, tmp_path):
     brief_file = made_services[0].kwargs["brief_file"]
     assert isinstance(brief_file, BriefFile)
     assert brief_file.write_enabled is False
+    session_dir = str(tmp_path / "transcripts" / "s1")
+    brief_file.set_directory(session_dir)
     brief_file.write("nope")
-    assert not os.path.exists(os.path.join(cfg.brief.dir, cfg.brief.filename))
+    assert not os.path.exists(os.path.join(session_dir, "brief.md"))
 
 
 # ---------------------------------------------------------------------------
@@ -545,8 +546,6 @@ def _engine_service(
     cfg.triage.brief_slot_id = brief_slot
     cfg.llm.chat_slot_id = chat_slot
     cfg.brief.write = brief_write
-    cfg.brief.dir = str(tmp_path / "brief")
-    cfg.brief.filename = "brief.md"
     client.brief_system = cfg.llm.triage_system_prompt
 
     analyzer, lane, brief_fn, brief_reset = build_triage(cfg, client)
@@ -558,7 +557,8 @@ def _engine_service(
     )
     emitted: list[dict] = []
     clock = _Clock()
-    brief_file = BriefFile(cfg.brief.write, cfg.brief.dir, cfg.brief.filename)
+    session_dir = str(tmp_path / "transcripts" / "s1")
+    brief_file = BriefFile(cfg.brief.write)
     service = AssistantService(
         cfg,
         _SilentSubscriber(),
@@ -580,13 +580,17 @@ def _engine_service(
         lane=lane,
         chat_engine=chat_engine,
         brief_file=brief_file,
+        session_dir=session_dir,
         cfg=cfg,
         client=client,
     )
 
 
-def _transcript(clock, text):
-    return {"v": 1, "type": events.TRANSCRIPT, "seq": 1, "ts": clock(), "text": text}
+def _transcript(clock, text, session_dir=None):
+    event = {"v": 1, "type": events.TRANSCRIPT, "seq": 1, "ts": clock(), "text": text}
+    if session_dir is not None:
+        event["session_dir"] = session_dir
+    return event
 
 
 def _triage_events(emitted):
@@ -611,8 +615,8 @@ def _run_queued(service, job):
     job.run()
 
 
-def _settle(service, clock, text):
-    service.handle_event(_transcript(clock, text))
+def _settle(service, clock, text, session_dir=None):
+    service.handle_event(_transcript(clock, text, session_dir))
     clock.advance(service.cfg.assistant.turn_gap_s + 0.1)
 
 
@@ -826,16 +830,16 @@ def test_durable_brief_file_holds_latest_and_clears(tmp_path):
     client = _EngineClient(brief_reply="FIRST BRIEF")
     h = _engine_service(tmp_path, client, brief_write=True)
     service = h.service
-    path = os.path.join(h.cfg.brief.dir, h.cfg.brief.filename)
+    path = os.path.join(h.session_dir, "brief.md")
 
-    _settle(service, h.clock, "first thought")
+    _settle(service, h.clock, "first thought", h.session_dir)
     _deliver(service, _laya_payload())
     _run_queued(service, _brief_jobs(service)[0])
     with open(path, encoding="utf-8") as handle:
         assert handle.read() == "FIRST BRIEF"
 
     client.brief_reply = "LATEST BRIEF"
-    _settle(service, h.clock, "second thought")
+    _settle(service, h.clock, "second thought", h.session_dir)
     _deliver(service, _laya_payload())
     _run_queued(service, _brief_jobs(service)[0])
     with open(path, encoding="utf-8") as handle:
@@ -854,9 +858,9 @@ def test_disabled_brief_file_absent_but_panel_updates(tmp_path):
     client = _EngineClient(brief_reply="PANEL BRIEF")
     h = _engine_service(tmp_path, client, brief_write=False)
     service = h.service
-    path = os.path.join(h.cfg.brief.dir, h.cfg.brief.filename)
+    path = os.path.join(h.session_dir, "brief.md")
 
-    _settle(service, h.clock, "first thought")
+    _settle(service, h.clock, "first thought", h.session_dir)
     _deliver(service, _laya_payload())
     _run_queued(service, _brief_jobs(service)[0])
 

tests/test_assistant_service.py

@@ -53,14 +53,17 @@ def _ev(type_, clock, **fields):
     return event
 
 
-def _transcript(clock, text, ts=None):
-    return {
+def _transcript(clock, text, ts=None, session_dir=None):
+    event = {
         "v": 1,
         "type": events.TRANSCRIPT,
         "seq": next(_sequence),
         "ts": clock() if ts is None else ts,
         "text": text,
     }
+    if session_dir is not None:
+        event["session_dir"] = session_dir
+    return event
 
 
 def test_ready_requires_matching_segment_done():
@@ -606,11 +609,15 @@ class FakeTriage:
 
 
 class FakeBriefFile:
-    """Recording stand-in for BriefFile (write/clear never raise)."""
+    """Recording stand-in for BriefFile (write/clear/set_directory never raise)."""
 
     def __init__(self):
         self.writes: list[str] = []
         self.clears = 0
+        self.directories: list[str] = []
+
+    def set_directory(self, directory):
+        self.directories.append(directory)
 
     def write(self, text):
         self.writes.append(text)
@@ -2196,3 +2203,277 @@ def test_chat_client_failure_emits_llm_error_and_service_survives():
     assert errors and "boom" in errors[-1]["message"]
     assert service.toggle_chat() is True
     assert service.modes.mode.value == "idle"
+
+
+# ---------------------------------------------------------------------------
+# Conversation session adoption (SESSION/TRANSCRIPT session_dir).
+# ---------------------------------------------------------------------------
+
+
+def _session_event(clock, session_dir):
+    event = _ev(events.SESSION, clock)
+    event["session_id"] = session_dir.rsplit("/", 1)[-1]
+    event["session_dir"] = session_dir
+    return event
+
+
+def _ring(service, clock, *items):
+    """Feed a daemon-seeded replay window: status -> items -> ring_end."""
+    service.handle_event(_ev(events.STATUS, clock))
+    for item in items:
+        service.handle_event(item)
+    service.handle_event(_ev(events.RING_END, clock))
+
+
+def _cleared(emitted):
+    return [e for e in _triage_events(emitted) if e.get("stage") == "cleared"]
+
+
+def test_status_includes_session_dir_without_brief_file():
+    service, _, clock = _service()
+    assert service.status()["session_dir"] is None
+
+    service.handle_event(_session_event(clock, "transcripts/2026-09-30-12-47-29"))
+
+    assert service.status()["session_dir"] == "transcripts/2026-09-30-12-47-29"
+
+
+def test_live_session_event_sets_session_and_records_directory():
+    brief_file = FakeBriefFile()
+    service, _, clock, _, _ = _triage_service(brief_file=brief_file)
+    session = "transcripts/2026-09-30-12-47-29"
+
+    service.handle_event(_session_event(clock, session))
+
+    assert service.status()["session_dir"] == session
+    assert brief_file.directories == [session]
+    assert service._triage_generation == 0
+
+
+def test_live_transcript_session_sets_session_without_wiping_utterance():
+    brief_file = FakeBriefFile()
+    service, _, clock, _, _ = _triage_service(brief_file=brief_file)
+    session = "transcripts/s1"
+
+    service.handle_event(_transcript(clock, "hello", session_dir=session))
+
+    assert service.status()["session_dir"] == session
+    assert brief_file.directories == [session]
+    assert service.triage_history == ["hello"]
+    assert service._pending_brief == ["Unknown: hello"]
+    assert service._triage_generation == 0
+
+
+def test_brief_write_lands_under_session_dir(tmp_path):
+    from stt_daemon.assistant.brief_file import BriefFile
+
+    brief_file = BriefFile(True)
+    service, _, clock, _, _ = _triage_service(brief_file=brief_file)
+    session = str(tmp_path / "transcripts" / "s1")
+
+    service.handle_event(_transcript(clock, "hello", session_dir=session))
+    brief_file.write("the brief")
+
+    path = tmp_path / "transcripts" / "s1" / "brief.md"
+    assert path.read_text(encoding="utf-8") == "the brief"
+
+
+def test_live_session_change_resets_triage_and_keeps_chat():
+    brief_file = FakeBriefFile()
+    resets: list[int] = []
+    service, emitted, clock, _, lane = _triage_service(
+        brief_file=brief_file, brief_reset=lambda: resets.append(1)
+    )
+    service.toggle_chat()
+    service.handle_event(_transcript(clock, "hello"))
+    service.handle_event(_session_event(clock, "transcripts/a"))
+    service.last_triage = {"signals": 1}
+    service._pending_brief = ["stale"]
+    emitted.clear()
+
+    service.handle_event(_session_event(clock, "transcripts/b"))
+
+    assert service.status()["session_dir"] == "transcripts/b"
+    assert service.last_triage is None
+    assert service._pending_brief == []
+    assert service._triage_generation == 1
+    assert service.chat_history == ["Unknown: hello"]
+    assert resets == [1]
+    assert lane.drops == 1
+    assert brief_file.directories == ["transcripts/a", "transcripts/b"]
+    assert len(_cleared(emitted)) == 1
+
+
+def test_sessionless_transcript_does_not_reset():
+    service, _, clock, _, _ = _triage_service()
+    service.handle_event(_session_event(clock, "transcripts/a"))
+    service.last_triage = {"signals": 1}
+
+    service.handle_event(_transcript(clock, "hello"))
+
+    assert service._triage_generation == 0
+    assert service.last_triage == {"signals": 1}
+    assert service.status()["session_dir"] == "transcripts/a"
+
+
+def test_same_dir_readoption_does_not_reset():
+    brief_file = FakeBriefFile()
+    service, _, clock, _, _ = _triage_service(brief_file=brief_file)
+    session = "transcripts/a"
+    service.handle_event(_session_event(clock, session))
+    service.last_triage = {"signals": 1}
+
+    service.handle_event(_transcript(clock, "again", session_dir=session))
+
+    assert service._triage_generation == 0
+    assert service.last_triage == {"signals": 1}
+    assert brief_file.directories == [session]
+
+
+def test_session_change_without_brief_file_does_not_raise():
+    service, _, clock, _, _ = _triage_service(brief_file=None)
+
+    service.handle_event(_session_event(clock, "transcripts/a"))
+    service.handle_event(_session_event(clock, "transcripts/b"))
+
+    assert service.status()["session_dir"] == "transcripts/b"
+    assert service._triage_generation == 1
+
+
+def test_replay_single_session_adopts_without_reset():
+    brief_file = FakeBriefFile()
+    service, emitted, clock, _, _ = _triage_service(brief_file=brief_file)
+    session = "transcripts/a"
+
+    _ring(service, clock, _transcript(clock, "hello", session_dir=session))
+
+    assert service.status()["session_dir"] == session
+    assert brief_file.directories == [session]
+    assert service._triage_generation == 0
+    assert _cleared(emitted) == []
+    assert service.triage_history == ["hello"]
+
+
+def test_replay_two_sessions_commits_newest_and_resets_once():
+    brief_file = FakeBriefFile()
+    service, emitted, clock, _, _ = _triage_service(brief_file=brief_file)
+    a, b = "transcripts/a", "transcripts/b"
+
+    _ring(
+        service,
+        clock,
+        _session_event(clock, a),
+        _transcript(clock, "old turn", session_dir=a),
+        _session_event(clock, b),
+        _transcript(clock, "new turn", session_dir=b),
+    )
+
+    assert service.status()["session_dir"] == b
+    assert brief_file.directories == [b]
+    assert service._triage_generation == 1
+    assert service.last_triage is None
+    assert service._pending_brief == []
+    assert len(_cleared(emitted)) == 1
+
+
+def test_reconnect_tailing_current_session_is_noop():
+    brief_file = FakeBriefFile()
+    service, emitted, clock, _, _ = _triage_service(brief_file=brief_file)
+    session = "transcripts/a"
+    service.handle_event(_session_event(clock, session))
+    service.last_triage = {"signals": 1}
+    emitted.clear()
+
+    _ring(service, clock, _transcript(clock, "replayed", session_dir=session))
+
+    assert service.status()["session_dir"] == session
+    assert brief_file.directories == [session]
+    assert service._triage_generation == 0
+    assert service.last_triage == {"signals": 1}
+    assert _cleared(emitted) == []
+
+
+def test_reconnect_older_to_current_resets_once_without_rewind():
+    brief_file = FakeBriefFile()
+    service, emitted, clock, _, _ = _triage_service(brief_file=brief_file)
+    a, b = "transcripts/a", "transcripts/b"
+    service.handle_event(_session_event(clock, a))
+    service.last_triage = {"signals": 1}
+    emitted.clear()
+
+    _ring(
+        service,
+        clock,
+        _session_event(clock, a),
+        _transcript(clock, "old", session_dir=a),
+        _session_event(clock, b),
+        _transcript(clock, "new", session_dir=b),
+    )
+
+    assert service.status()["session_dir"] == b
+    assert brief_file.directories == [a, b]
+    assert service._triage_generation == 1
+    assert service.last_triage is None
+    assert len(_cleared(emitted)) == 1
+
+
+def test_reconnect_only_newer_session_resets_once():
+    brief_file = FakeBriefFile()
+    service, _, clock, _, _ = _triage_service(brief_file=brief_file)
+    a, b = "transcripts/a", "transcripts/b"
+    service.handle_event(_session_event(clock, a))
+    service.last_triage = {"signals": 1}
+
+    _ring(service, clock, _transcript(clock, "new", session_dir=b))
+
+    assert service.status()["session_dir"] == b
+    assert service._triage_generation == 1
+    assert service.last_triage is None
+    assert brief_file.directories == [a, b]
+
+
+def test_reconnect_span_older_current_clears_buffered_older_turns():
+    """A RING_END whose ring spanned an older session resets even when the
+    newest session already matches the adopted directory (no re-set_directory)."""
+    brief_file = FakeBriefFile()
+    service, emitted, clock, _, _ = _triage_service(brief_file=brief_file)
+    a, b = "transcripts/a", "transcripts/b"
+    _ring(service, clock, _session_event(clock, a))
+    assert service._triage_generation == 0
+    service.handle_event(_transcript(clock, "old turn", session_dir=a))
+    assert service._pending_brief == ["Unknown: old turn"]
+
+    service.handle_event(_session_event(clock, b))
+    assert service._triage_generation == 1
+    service.handle_event(_transcript(clock, "new turn", session_dir=b))
+    assert service._pending_brief == ["Unknown: new turn"]
+    emitted.clear()
+
+    _ring(
+        service,
+        clock,
+        _session_event(clock, a),
+        _transcript(clock, "old turn", session_dir=a),
+        _session_event(clock, b),
+        _transcript(clock, "new turn", session_dir=b),
+    )
+
+    assert service.status()["session_dir"] == b
+    assert brief_file.directories == [a, b]
+    assert service._pending_brief == []
+    assert service._triage_generation == 2
+    assert len(_cleared(emitted)) == 1
+
+
+def test_daemon_restart_new_session_resets_once():
+    brief_file = FakeBriefFile()
+    service, _, clock, _, _ = _triage_service(brief_file=brief_file)
+    service.handle_event(_session_event(clock, "transcripts/a"))
+    service.last_triage = {"signals": 1}
+
+    service.handle_event(_session_event(clock, "transcripts/b"))
+
+    assert service.status()["session_dir"] == "transcripts/b"
+    assert service._triage_generation == 1
+    assert service.last_triage is None
+    assert brief_file.directories == ["transcripts/a", "transcripts/b"]

tests/test_brief_file.py

@@ -1,6 +1,7 @@
 """Tests for the atomic best-effort durable brief writer (BriefFile)."""
 
 import os
+import threading
 
 from stt_daemon.assistant.brief_file import BriefFile
 
@@ -65,3 +66,59 @@ def test_write_failure_path_swallowed_no_raise(tmp_path):
 def test_clear_failure_path_swallowed_no_raise(tmp_path):
     brief = BriefFile(True, str(tmp_path / "missing"))
     brief.clear()  # missing dir -> open fails -> swallowed, no raise
+
+
+def test_default_directory_is_brief(tmp_path, monkeypatch):
+    # constructor default: `brief/brief.md` relative to the cwd.
+    monkeypatch.chdir(tmp_path)
+    brief = BriefFile(True)
+    brief.write("x")
+    assert (tmp_path / "brief" / "brief.md").read_text(encoding="utf-8") == "x"
+
+
+def test_set_directory_redirects_writes(tmp_path):
+    brief = BriefFile(True)
+    brief.set_directory(str(tmp_path))
+    brief.write("x")
+    assert (tmp_path / "brief.md").read_text(encoding="utf-8") == "x"
+
+
+def test_disabled_after_set_directory_writes_nothing(tmp_path):
+    brief = BriefFile(False)
+    brief.set_directory(str(tmp_path))
+    brief.write("x")
+    brief.clear()
+    assert not any(tmp_path.iterdir())
+
+
+def test_concurrent_set_directory_and_write_stay_within_dirs(tmp_path):
+    # race probe: a writer flipping between two dirs must never land elsewhere.
+    dir_a = tmp_path / "a"
+    dir_b = tmp_path / "b"
+    brief = BriefFile(True)
+    brief.set_directory(str(dir_a))
+    stop = threading.Event()
+
+    def swapper():
+        while not stop.is_set():
+            brief.set_directory(str(dir_a))
+            brief.set_directory(str(dir_b))
+
+    def writer():
+        for _ in range(500):
+            brief.write("x")
+        stop.set()
+
+    threads = [threading.Thread(target=swapper), threading.Thread(target=writer)]
+    for thread in threads:
+        thread.start()
+    for thread in threads:
+        thread.join()
+
+    allowed = {dir_a.resolve(), dir_b.resolve()}
+    produced = [p for p in tmp_path.rglob("*") if p.is_file()]
+    assert produced, "writer should have produced at least one file"
+    for path in produced:
+        assert path.resolve().parent in allowed, f"write escaped to {path}"
+    assert not (tmp_path / "brief").exists()
+    assert not (tmp_path / "brief.md").exists()

tests/test_config.py

@@ -52,7 +52,7 @@ language = "es"
 [output]
 stdout = false
 jsonl = true
-jsonl_dir = "logs"
+session_dir = "logs"
 """
     with tempfile.TemporaryDirectory() as tmpdir:
         config_path = Path(tmpdir) / "config.toml"
@@ -70,7 +70,7 @@ jsonl_dir = "logs"
         assert cfg.whisper.language == "es"
         assert cfg.output.stdout is False
         assert cfg.output.jsonl is True
-        assert cfg.output.jsonl_dir == "logs"
+        assert cfg.output.session_dir == "logs"
 
 
 def test_language_normalization_auto():
@@ -207,14 +207,18 @@ def test_output_defaults_disable_audio_saving():
             os.chdir(old_cwd)
 
     assert cfg.output.save_audio is False
-    assert cfg.output.audio_dir == "audio"
+    assert cfg.output.session_dir == "transcripts"
+    assert cfg.output.session_gap_min == 5.0
+    assert not hasattr(cfg.output, "jsonl_dir")
+    assert not hasattr(cfg.output, "audio_dir")
 
 
 def test_output_save_audio_fields():
     config_toml = """
 [output]
 save_audio = true
-audio_dir = "clips"
+session_dir = "clips"
+session_gap_min = 12.0
 """
     with tempfile.TemporaryDirectory() as tmpdir:
         config_path = Path(tmpdir) / "config.toml"
@@ -223,7 +227,34 @@ audio_dir = "clips"
         cfg = load_config(str(config_path))
 
     assert cfg.output.save_audio is True
-    assert cfg.output.audio_dir == "clips"
+    assert cfg.output.session_dir == "clips"
+    assert cfg.output.session_gap_min == 12.0
+
+
+def test_legacy_output_and_brief_keys_ignored():
+    """Retired keys in a legacy TOML load without error and keep the new defaults."""
+    config_toml = """
+[output]
+jsonl_dir = "logs"
+audio_dir = "clips"
+
+[brief]
+write = false
+dir = "briefing"
+"""
+    with tempfile.TemporaryDirectory() as tmpdir:
+        config_path = Path(tmpdir) / "config.toml"
+        config_path.write_text(config_toml)
+
+        cfg = load_config(str(config_path))
+
+    assert cfg.output.session_dir == "transcripts"
+    assert cfg.output.session_gap_min == 5.0
+    assert cfg.brief.write is False
+    assert not hasattr(cfg.output, "jsonl_dir")
+    assert not hasattr(cfg.output, "audio_dir")
+    assert not hasattr(cfg.brief, "dir")
+    assert not hasattr(cfg.brief, "filename")
 
 
 def test_daemon_defaults():
@@ -711,17 +742,15 @@ def test_brief_defaults():
             os.chdir(old_cwd)
 
     assert cfg.brief.write is False
-    assert cfg.brief.dir == "brief"
-    assert cfg.brief.filename == "brief.md"
+    assert not hasattr(cfg.brief, "dir")
+    assert not hasattr(cfg.brief, "filename")
 
 
 def test_brief_config_fields_are_parsed():
-    """Every [brief] key is applied from TOML."""
+    """The [brief] write key is applied from TOML."""
     config_toml = """
 [brief]
 write = true
-dir = "briefing"
-filename = "live.md"
 """
     with tempfile.TemporaryDirectory() as tmpdir:
         config_path = Path(tmpdir) / "config.toml"
@@ -730,8 +759,8 @@ filename = "live.md"
         cfg = load_config(str(config_path))
 
     assert cfg.brief.write is True
-    assert cfg.brief.dir == "briefing"
-    assert cfg.brief.filename == "live.md"
+    assert not hasattr(cfg.brief, "dir")
+    assert not hasattr(cfg.brief, "filename")
 
 
 def test_brief_write_false_is_parsed(caplog):

tests/test_daemon.py

@@ -1,5 +1,6 @@
 """Tests for the daemon's worker processing and transcription backlog warning."""
 
+import json
 import logging
 
 import numpy as np
@@ -9,6 +10,7 @@ from stt_daemon import events
 from stt_daemon import transcriber as transcriber_mod
 from stt_daemon.config import Config
 from stt_daemon.daemon import Daemon
+from stt_daemon.output import TranscriptSink
 from stt_daemon.segmenter import Utterance
 from stt_daemon.transcriber import TranscriptResult
 
@@ -142,6 +144,10 @@ def test_sink_failure_still_publishes_transcript(caplog):
     assert transcripts[0]["lang_prob"] == 0.9
     assert transcripts[0]["audio"] is None
     assert "Output error" in caplog.text
+    sid = daemon._session_tracker.current_id
+    assert transcripts[0]["session_id"] == sid
+    assert sid
+    assert transcripts[0]["session_dir"].endswith(sid)
 
 
 def test_sink_failure_reports_segment_done_as_emitted():
@@ -152,3 +158,113 @@ def test_sink_failure_reports_segment_done_as_emitted():
     done = [fields for type_, fields in published if type_ == events.SEGMENT_DONE]
     assert len(done) == 1
     assert done[0]["emitted"] is True
+
+
+# --- session identity -------------------------------------------------------
+
+
+class _RecordingSink:
+    """Minimal sink capturing the per-emit session_dir kwarg."""
+
+    def __init__(self):
+        self.session_dirs = []
+
+    def emit(self, *args, **kwargs):
+        self.session_dirs.append(kwargs.get("session_dir"))
+        return None
+
+    def close(self):
+        pass
+
+
+def _session_daemon(monkeypatch, clock):
+    daemon = _daemon()
+    daemon.transcriber = _FakeTranscriber(_transcript_result())
+    sink = _RecordingSink()
+    daemon.sink = sink
+    published = []
+    daemon._publish = lambda type_, **fields: published.append((type_, fields))
+    monkeypatch.setattr("stt_daemon.daemon.time.time", lambda: clock["now"])
+    return daemon, sink, published
+
+
+def test_process_stamps_session_id_matching_jsonl(tmp_path):
+    daemon = _daemon()
+    daemon.transcriber = _FakeTranscriber(_transcript_result())
+    daemon.cfg.output.stdout = False
+    daemon.cfg.output.jsonl = True
+    daemon.cfg.output.save_audio = False
+    daemon.cfg.output.session_dir = str(tmp_path)
+    daemon.sink = TranscriptSink(stdout=False, jsonl=True, save_audio=False, sample_rate=16000)
+    published = []
+    daemon._publish = lambda type_, **fields: published.append((type_, fields))
+
+    daemon._process(_utterance(1))
+
+    sid = daemon._session_tracker.current_id
+    assert isinstance(sid, str) and sid
+    transcripts = [f for t, f in published if t == events.TRANSCRIPT]
+    assert len(transcripts) == 1
+    assert transcripts[0]["session_id"] == sid
+    assert transcripts[0]["session_dir"] == str(tmp_path / sid)
+    sessions = [f for t, f in published if t == events.SESSION]
+    assert len(sessions) == 1
+    assert sessions[0]["session_id"] == sid
+
+    record = json.loads((tmp_path / sid / "transcripts.jsonl").read_text())
+    assert record["session_id"] == sid
+
+
+def test_within_gap_reuses_one_session(monkeypatch):
+    clock = {"now": 1000.0}
+    daemon, sink, published = _session_daemon(monkeypatch, clock)
+
+    daemon._process(_utterance(1))
+    clock["now"] = 1100.0
+    daemon._process(_utterance(2))
+
+    sessions = [f for t, f in published if t == events.SESSION]
+    assert len(sessions) == 1
+    transcripts = [f for t, f in published if t == events.TRANSCRIPT]
+    assert len(transcripts) == 2
+    assert transcripts[0]["session_id"] == transcripts[1]["session_id"]
+    assert transcripts[0]["session_id"] == sessions[0]["session_id"]
+    assert len(sink.session_dirs) == 2
+    assert all(
+        d is not None and d.name == sessions[0]["session_id"] for d in sink.session_dirs
+    )
+
+
+def test_beyond_gap_mints_new_session(monkeypatch):
+    clock = {"now": 1000.0}
+    daemon, _sink, published = _session_daemon(monkeypatch, clock)
+
+    daemon._process(_utterance(1))
+    clock["now"] = 1100.0  # inside the 300s default gap
+    daemon._process(_utterance(2))
+    clock["now"] = 1500.0  # 400s > 300s: rollover
+    daemon._process(_utterance(3))
+
+    sessions = [f for t, f in published if t == events.SESSION]
+    assert len(sessions) == 2
+    assert sessions[0]["session_id"] != sessions[1]["session_id"]
+    transcripts = [f for t, f in published if t == events.TRANSCRIPT]
+    assert transcripts[0]["session_id"] == sessions[0]["session_id"]
+    assert transcripts[1]["session_id"] == sessions[0]["session_id"]
+    assert transcripts[2]["session_id"] == sessions[1]["session_id"]
+    assert daemon._session_tracker.current_id == sessions[1]["session_id"]
+
+
+def test_session_advances_with_file_writes_off(monkeypatch):
+    clock = {"now": 1000.0}
+    daemon, _sink, published = _session_daemon(monkeypatch, clock)
+    daemon.cfg.output.jsonl = False
+    daemon.cfg.output.save_audio = False
+
+    daemon._process(_utterance(1))
+    clock["now"] = 1500.0
+    daemon._process(_utterance(2))
+
+    sessions = [f for t, f in published if t == events.SESSION]
+    assert len(sessions) == 2
+    assert sessions[1]["session_id"] != sessions[0]["session_id"]

tests/test_daemon_events.py

@@ -164,7 +164,7 @@ def _base_config(sock_path):
     cfg.vad.rms_floor = 0.0
     cfg.daemon.queue_size = 50
     cfg.output.stdout = False
-    cfg.output.jsonl_dir = None
+    cfg.output.jsonl = False
     cfg.output.save_audio = False
     return cfg
 

tests/test_daemon_speaker.py

@@ -108,7 +108,7 @@ def _config(tmp_path, **overrides):
     cfg = Config()
     cfg.events.socket = str(tmp_path / "events.sock")
     cfg.output.stdout = False
-    cfg.output.jsonl_dir = None
+    cfg.output.jsonl = False
     cfg.output.save_audio = False
     cfg.speaker.enabled = True
     cfg.speaker.model_path = str(tmp_path / "model.onnx")

tests/test_enroll_assist.py

@@ -120,8 +120,9 @@ def write_cfg(tmp_path: Path, *, min_enroll_s: float = 2.0,
     model = tmp_path / "model.onnx"
     if model_ok:
         model.write_bytes(b"fake-model")
-    audio_dir = tmp_path / "audio"
-    audio_dir.mkdir(exist_ok=True)
+    session_dir = tmp_path / "sessions"
+    audio_dir = session_dir / "2026-09-30-12-00-00" / "audio"
+    audio_dir.mkdir(parents=True, exist_ok=True)
     store_dir = tmp_path / "speakers"
     cfg_path = tmp_path / "config.toml"
     cfg_path.write_text(
@@ -131,7 +132,7 @@ def write_cfg(tmp_path: Path, *, min_enroll_s: float = 2.0,
         f"max_prototypes = {max_prototypes}\n"
         f'store_dir = "{store_dir}"\n'
         "[output]\n"
-        f'audio_dir = "{audio_dir}"\n'
+        f'session_dir = "{session_dir}"\n'
         "[events]\n"
         f'socket = "{tmp_path / "bus.sock"}"\n',
         encoding="utf-8",
@@ -392,6 +393,20 @@ def test_seed_enroll_empty_dir_writes_nothing(tmp_path, capsys):
     assert "nothing enrolled" in capsys.readouterr().out
 
 
+def test_seed_enroll_missing_session_root_writes_nothing(tmp_path, capsys):
+    cfg, _, store_dir = write_cfg(tmp_path)
+    missing = tmp_path / "sessions" / "no-such-session"
+    cfg.output.session_dir = str(missing)
+    seed = write_const_wav(tmp_path / "seed.wav", 2.5, A)
+
+    accepted = seed_enroll(cfg, ConstEmbedder(), make_store(cfg), "Efren",
+                           seed, count=3, auto=True)
+
+    assert accepted == []
+    assert not store_files(store_dir)
+    assert "nothing enrolled" in capsys.readouterr().out
+
+
 def test_seed_enroll_quit_aborts_without_writing(tmp_path, capsys):
     cfg, audio_dir, store_dir = write_cfg(tmp_path)
     seed, _ = seed_and_clips(audio_dir)

tests/test_events.py

@@ -74,9 +74,17 @@ def test_all_types_is_complete():
         events.STATUS,
         events.RING_END,
         events.SPEAKERS,
+        events.SESSION,
     }
 
 
 def test_speakers_event_constant():
     assert events.SPEAKERS == "speakers"
     assert events.SPEAKERS in events.ALL_TYPES
+
+
+def test_session_event_round_trip():
+    event = events.make_event(events.SESSION, 1, session_id="x")
+    assert event["type"] == "session"
+    assert event["session_id"] == "x"
+    assert events.decode(events.encode(event)) == event

tests/test_output.py

@@ -1,6 +1,7 @@
 """Tests for transcript output sinks."""
 
 import json
+import re
 import wave
 
 import numpy as np
@@ -15,13 +16,11 @@ def _result(text="hello world"):
 
 
 def test_jsonl_includes_level_and_reason(tmp_path):
-    sink = TranscriptSink(stdout=False, jsonl_dir=str(tmp_path), jsonl=True)
-    sink.emit(_result(), 2.5, rms=0.041234, reason="pause")
+    session = tmp_path / "2026-09-30-12-47-29"
+    sink = TranscriptSink(stdout=False, jsonl=True)
+    sink.emit(_result(), 2.5, rms=0.041234, reason="pause", session_dir=session)
 
-    files = list(tmp_path.glob("*.jsonl"))
-    assert len(files) == 1
-
-    record = json.loads(files[0].read_text().splitlines()[0])
+    record = json.loads((session / "transcripts.jsonl").read_text().splitlines()[0])
     assert record["rms"] == round(0.041234, 5)
     assert record["end_reason"] == "pause"
     assert record["duration_s"] == 2.5
@@ -30,32 +29,34 @@ def test_jsonl_includes_level_and_reason(tmp_path):
 
 
 def test_jsonl_includes_transcribe_seconds(tmp_path):
-    sink = TranscriptSink(stdout=False, jsonl_dir=str(tmp_path), jsonl=True)
-    sink.emit(_result(), 2.5, rms=0.041234, reason="pause", transcribe_s=0.81234)
+    session = tmp_path / "sess"
+    sink = TranscriptSink(stdout=False, jsonl=True)
+    sink.emit(_result(), 2.5, session_dir=session, rms=0.041234, transcribe_s=0.81234)
 
-    record = json.loads(next(tmp_path.glob("*.jsonl")).read_text().splitlines()[0])
+    record = json.loads((session / "transcripts.jsonl").read_text().splitlines()[0])
     assert record["transcribe_s"] == round(0.81234, 3)
 
 
 def test_transcribe_seconds_omitted_when_unknown(tmp_path):
-    sink = TranscriptSink(stdout=False, jsonl_dir=str(tmp_path), jsonl=True)
-    sink.emit(_result(), 2.5, rms=0.041234, reason="pause")
+    session = tmp_path / "sess"
+    sink = TranscriptSink(stdout=False, jsonl=True)
+    sink.emit(_result(), 2.5, session_dir=session, rms=0.041234, reason="pause")
 
-    record = json.loads(next(tmp_path.glob("*.jsonl")).read_text().splitlines()[0])
+    record = json.loads((session / "transcripts.jsonl").read_text().splitlines()[0])
     assert "transcribe_s" not in record
 
 
 def test_jsonl_disabled_writes_nothing(tmp_path, capsys):
-    """jsonl=False keeps transcripts off disk even when jsonl_dir is set."""
-    sink = TranscriptSink(stdout=True, jsonl_dir=str(tmp_path), jsonl=False)
-    sink.emit(_result(), 1.0, rms=0.02, reason="pause")
+    """jsonl=False keeps transcripts off disk even when a session dir is given."""
+    sink = TranscriptSink(stdout=True, jsonl=False)
+    sink.emit(_result(), 1.0, session_dir=tmp_path / "sess", rms=0.02, reason="pause")
 
     assert "hello world" in capsys.readouterr().out
-    assert list(tmp_path.glob("*.jsonl")) == []
+    assert not (tmp_path / "sess" / "transcripts.jsonl").exists()
 
 
 def test_stdout_includes_level_and_reason(capsys):
-    sink = TranscriptSink(stdout=True, jsonl_dir=None)
+    sink = TranscriptSink(stdout=True)
     sink.emit(_result(), 1.0, rms=0.02, reason="end_of_turn")
 
     out = capsys.readouterr().out
@@ -65,29 +66,41 @@ def test_stdout_includes_level_and_reason(capsys):
 
 
 def test_empty_text_is_skipped(tmp_path, capsys):
-    sink = TranscriptSink(stdout=True, jsonl_dir=str(tmp_path), jsonl=True)
-    sink.emit(_result(text=""), 1.0, rms=0.02, reason="pause")
+    session = tmp_path / "sess"
+    sink = TranscriptSink(stdout=True, jsonl=True)
+    sink.emit(_result(text=""), 1.0, session_dir=session, rms=0.02, reason="pause")
 
     assert capsys.readouterr().out == ""
-    assert list(tmp_path.glob("*.jsonl")) == []
+    assert not (session / "transcripts.jsonl").exists()
+
+
+def test_jsonl_without_session_dir_logs_warning(tmp_path, caplog):
+    sink = TranscriptSink(stdout=False, jsonl=True)
+    with caplog.at_level("WARNING"):
+        sink.emit(_result(), 1.0, rms=0.02, reason="pause", session_dir=None)
+
+    assert list(tmp_path.rglob("transcripts.jsonl")) == []
+    assert "no session_dir" in caplog.text
+
+
+def test_jsonl_records_session_id(tmp_path):
+    session = tmp_path / "2026-09-30-12-47-29"
+    sink = TranscriptSink(stdout=False, jsonl=True)
+    sink.emit(_result(), 1.0, session_dir=session, rms=0.02, reason="pause")
+
+    record = json.loads((session / "transcripts.jsonl").read_text().splitlines()[0])
+    assert record["session_id"] == "2026-09-30-12-47-29"
 
 
 def test_saves_wav_when_enabled(tmp_path):
     audio = np.zeros(16000, dtype=np.float32)
     audio[:8000] = 0.5
-    logs = tmp_path / "logs"
-
-    sink = TranscriptSink(
-        stdout=False,
-        jsonl_dir=str(logs),
-        jsonl=True,
-        save_audio=True,
-        audio_dir=str(tmp_path / "audio"),
-        sample_rate=16000,
-    )
-    sink.emit(_result(), 1.0, rms=0.3, reason="pause", audio=audio)
+    session = tmp_path / "2026-09-30-12-47-29"
 
-    wavs = list((tmp_path / "audio").rglob("*.wav"))
+    sink = TranscriptSink(stdout=False, jsonl=True, save_audio=True, sample_rate=16000)
+    sink.emit(_result(), 1.0, session_dir=session, rms=0.3, reason="pause", audio=audio)
+
+    wavs = list((session / "audio").glob("*.wav"))
     assert len(wavs) == 1
 
     with wave.open(str(wavs[0]), "rb") as wf:
@@ -96,10 +109,23 @@ def test_saves_wav_when_enabled(tmp_path):
         assert wf.getframerate() == 16000
         assert wf.getnframes() == 16000
 
-    record = json.loads(next(logs.glob("*.jsonl")).read_text().splitlines()[0])
+    record = json.loads((session / "transcripts.jsonl").read_text().splitlines()[0])
     assert record["audio"] == str(wavs[0])
 
 
+def test_session_wav_name_and_location(tmp_path):
+    session = tmp_path / "2026-09-30-12-47-29"
+    audio = np.zeros(160, dtype=np.float32)
+
+    sink = TranscriptSink(stdout=False, save_audio=True, sample_rate=16000)
+    sink.emit(_result(), 1.0, session_dir=session, rms=0.3, reason="pause", audio=audio)
+
+    wavs = list((session / "audio").glob("*.wav"))
+    assert len(wavs) == 1
+    assert re.fullmatch(r"\d{6}-\d{3}-pause\.wav", wavs[0].name)
+    assert wavs[0].parent == session / "audio"
+
+
 def test_single_timestamp_per_emit(tmp_path, monkeypatch):
     """All artifacts of one emit() share one timestamp, even across midnight."""
     import stt_daemon.output as output
@@ -117,50 +143,41 @@ def test_single_timestamp_per_emit(tmp_path, monkeypatch):
 
     monkeypatch.setattr(output, "datetime", FakeDatetime)
 
-    logs = tmp_path / "logs"
-    sink = TranscriptSink(
-        stdout=False,
-        jsonl_dir=str(logs),
-        jsonl=True,
-        save_audio=True,
-        audio_dir=str(tmp_path / "audio"),
-        sample_rate=16000,
+    session = tmp_path / "sess"
+    sink = TranscriptSink(stdout=False, jsonl=True, save_audio=True, sample_rate=16000)
+    sink.emit(
+        _result(),
+        1.0,
+        session_dir=session,
+        rms=0.1,
+        reason="pause",
+        audio=np.zeros(160, dtype=np.float32),
     )
-    sink.emit(_result(), 1.0, rms=0.1, reason="pause", audio=np.zeros(160, dtype=np.float32))
 
     assert len(calls) == 1
-    jsonl = next(logs.glob("*.jsonl"))
-    assert jsonl.name == "2026-09-20.jsonl"
-    record = json.loads(jsonl.read_text().splitlines()[0])
+    record = json.loads((session / "transcripts.jsonl").read_text().splitlines()[0])
     assert record["ts"].startswith("2026-09-20")
-    wav = next((tmp_path / "audio").rglob("*.wav"))
-    assert wav.parent.name == "2026-09-20"
+    wav = next((session / "audio").glob("*.wav"))
+    assert wav.name.startswith("235959-")
 
 
 def test_no_wav_when_disabled(tmp_path):
-    sink = TranscriptSink(
-        stdout=False,
-        jsonl_dir=None,
-        save_audio=False,
-        audio_dir=str(tmp_path / "audio"),
-        sample_rate=16000,
+    session = tmp_path / "sess"
+    sink = TranscriptSink(stdout=False, save_audio=False, sample_rate=16000)
+    sink.emit(
+        _result(), 1.0, session_dir=session, rms=0.3, reason="pause",
+        audio=np.zeros(1600, dtype=np.float32),
     )
-    sink.emit(_result(), 1.0, rms=0.3, reason="pause", audio=np.zeros(1600, dtype=np.float32))
 
-    assert not (tmp_path / "audio").exists()
+    assert not (session / "audio").exists()
 
 
 def test_save_audio_without_samples_is_safe(tmp_path):
-    sink = TranscriptSink(
-        stdout=False,
-        jsonl_dir=None,
-        save_audio=True,
-        audio_dir=str(tmp_path / "audio"),
-        sample_rate=16000,
-    )
-    sink.emit(_result(), 1.0, rms=0.3, reason="pause", audio=None)
+    session = tmp_path / "sess"
+    sink = TranscriptSink(stdout=False, save_audio=True, sample_rate=16000)
+    sink.emit(_result(), 1.0, session_dir=session, rms=0.3, reason="pause", audio=None)
 
-    assert not (tmp_path / "audio").exists()
+    assert not (session / "audio").exists()
 
 
 def test_failed_audio_save_is_non_fatal(tmp_path, monkeypatch, capsys, caplog):
@@ -171,22 +188,16 @@ def test_failed_audio_save_is_non_fatal(tmp_path, monkeypatch, capsys, caplog):
 
     monkeypatch.setattr(output, "write_wav", boom)
 
-    logs = tmp_path / "logs"
-    sink = TranscriptSink(
-        stdout=True,
-        jsonl_dir=str(logs),
-        jsonl=True,
-        save_audio=True,
-        audio_dir=str(tmp_path / "audio"),
-        sample_rate=16000,
-    )
+    session = tmp_path / "sess"
+    sink = TranscriptSink(stdout=True, jsonl=True, save_audio=True, sample_rate=16000)
     with caplog.at_level("WARNING"):
         sink.emit(
-            _result(), 1.0, rms=0.3, reason="pause", audio=np.zeros(1600, dtype=np.float32)
+            _result(), 1.0, session_dir=session, rms=0.3, reason="pause",
+            audio=np.zeros(1600, dtype=np.float32),
         )
 
     assert "hello world" in capsys.readouterr().out
-    record = json.loads(next(logs.glob("*.jsonl")).read_text().splitlines()[0])
+    record = json.loads((session / "transcripts.jsonl").read_text().splitlines()[0])
     assert record["text"] == "hello world"
     assert "audio" not in record
     assert "Failed to save audio" in caplog.text
@@ -194,23 +205,26 @@ def test_failed_audio_save_is_non_fatal(tmp_path, monkeypatch, capsys, caplog):
 
 def test_failed_serialization_leaves_jsonl_untouched(tmp_path):
     """A record that cannot be serialized must not leave a partial line behind."""
-    sink = TranscriptSink(stdout=False, jsonl_dir=str(tmp_path), jsonl=True)
-    sink.emit(_result(), 2.5, rms=0.04, reason="pause")
-    path = next(tmp_path.glob("*.jsonl"))
+    session = tmp_path / "sess"
+    sink = TranscriptSink(stdout=False, jsonl=True)
+    sink.emit(_result(), 2.5, session_dir=session, rms=0.04, reason="pause")
+    path = session / "transcripts.jsonl"
     before = path.read_text()
 
     bad = TranscriptResult(language="en", language_probability=np.float32(0.98), text="boom")
     with pytest.raises(TypeError):
-        sink.emit(bad, 2.5, rms=0.04, reason="pause")
+        sink.emit(bad, 2.5, session_dir=session, rms=0.04, reason="pause")
 
     assert path.read_text() == before
 
 
 def test_jsonl_includes_speaker_fields(tmp_path):
-    sink = TranscriptSink(stdout=False, jsonl_dir=str(tmp_path), jsonl=True)
+    session = tmp_path / "sess"
+    sink = TranscriptSink(stdout=False, jsonl=True)
     sink.emit(
         _result(),
         3.2,
+        session_dir=session,
         rms=0.031,
         reason="pause",
         seg_id=142,
@@ -220,7 +234,7 @@ def test_jsonl_includes_speaker_fields(tmp_path):
         speaker_score=0.712345,
     )
 
-    record = json.loads(next(tmp_path.glob("*.jsonl")).read_text().splitlines()[0])
+    record = json.loads((session / "transcripts.jsonl").read_text().splitlines()[0])
     assert record["seg_id"] == 142
     assert record["speaker_id"] == "spk_0001"
     assert record["speaker_status"] == "matched"
@@ -229,10 +243,11 @@ def test_jsonl_includes_speaker_fields(tmp_path):
 
 
 def test_jsonl_speaker_fields_default_to_nulls(tmp_path):
-    sink = TranscriptSink(stdout=False, jsonl_dir=str(tmp_path), jsonl=True)
-    sink.emit(_result(), 2.5, rms=0.04, reason="pause")
+    session = tmp_path / "sess"
+    sink = TranscriptSink(stdout=False, jsonl=True)
+    sink.emit(_result(), 2.5, session_dir=session, rms=0.04, reason="pause")
 
-    record = json.loads(next(tmp_path.glob("*.jsonl")).read_text().splitlines()[0])
+    record = json.loads((session / "transcripts.jsonl").read_text().splitlines()[0])
     assert record["seg_id"] is None
     assert record["speaker_id"] is None
     assert record["speaker_status"] == "none"
@@ -241,10 +256,9 @@ def test_jsonl_speaker_fields_default_to_nulls(tmp_path):
 
 
 def test_jsonl_speaker_score_rounded_to_4dp(tmp_path):
-    sink = TranscriptSink(stdout=False, jsonl_dir=str(tmp_path), jsonl=True)
-    sink.emit(_result(), 1.0, rms=0.02, reason="pause", speaker_score=0.123456)
+    session = tmp_path / "sess"
+    sink = TranscriptSink(stdout=False, jsonl=True)
+    sink.emit(_result(), 1.0, session_dir=session, rms=0.02, reason="pause", speaker_score=0.123456)
 
-    record = json.loads(next(tmp_path.glob("*.jsonl")).read_text().splitlines()[0])
+    record = json.loads((session / "transcripts.jsonl").read_text().splitlines()[0])
     assert record["speaker_score"] == round(0.123456, 4)
-
-

tests/test_session.py

@@ -0,0 +1,106 @@
+"""Tests for the pure SessionTracker module (no fakes, no I/O)."""
+
+from stt_daemon.session import SessionTracker
+
+
+def _counter():
+    """Deterministic id factory: returns id-1, id-2, ... and records call times."""
+    calls = []
+
+    def factory(now):
+        calls.append(now)
+        return f"id-{len(calls)}"
+
+    return factory, calls
+
+
+def test_current_id_is_none_before_first_observe():
+    factory, _ = _counter()
+    tracker = SessionTracker(300.0, factory)
+
+    assert tracker.current_id is None
+
+
+def test_first_observe_mints_the_factory_id_for_now():
+    def factory(now):
+        return f"id-{now}"
+
+    tracker = SessionTracker(300.0, factory)
+
+    assert tracker.observe(100.0) == factory(100.0)
+
+
+def test_same_id_within_gap_then_new_id_beyond_gap():
+    gap = 1.0
+    factory, calls = _counter()
+    tracker = SessionTracker(gap, factory)
+
+    first = tracker.observe(100.0)
+    # `gap - 1` after the first observation: still inside the gap, same id.
+    assert tracker.observe(100.0 + gap - 1) == first
+    # Beyond the gap: a new id, and the factory minted exactly twice.
+    assert tracker.observe(100.0 + gap + 1) != first
+    assert len(calls) == 2
+
+
+def test_exact_gap_boundary_does_not_mint():
+    gap = 300.0
+    factory, calls = _counter()
+    tracker = SessionTracker(gap, factory)
+
+    first = tracker.observe(100.0)
+    # now - last == gap is NOT > gap, so the boundary must not roll over.
+    assert tracker.observe(100.0 + gap) == first
+    assert len(calls) == 1
+
+
+def test_just_beyond_gap_mints_a_new_id():
+    gap = 300.0
+    factory, calls = _counter()
+    tracker = SessionTracker(gap, factory)
+
+    first = tracker.observe(100.0)
+    assert tracker.observe(100.0 + gap + 1) != first
+    assert len(calls) == 2
+
+
+def test_zero_gap_returns_one_id_forever():
+    factory, calls = _counter()
+    tracker = SessionTracker(0.0, factory)
+
+    first = tracker.observe(0.0)
+    assert tracker.observe(1_000_000.0) == first
+    assert tracker.observe(1_000_000_000.0) == first
+    assert len(calls) == 1
+
+
+def test_negative_gap_also_disables_rollover():
+    factory, calls = _counter()
+    tracker = SessionTracker(-5.0, factory)
+
+    first = tracker.observe(100.0)
+    assert tracker.observe(100.0 + 86_400.0) == first
+    assert len(calls) == 1
+
+
+def test_large_jump_yields_exactly_one_new_id():
+    gap = 300.0
+    factory, calls = _counter()
+    tracker = SessionTracker(gap, factory)
+
+    first = tracker.observe(100.0)
+    after = tracker.observe(100.0 + 86_400.0)
+
+    assert after != first
+    assert len(calls) == 2
+
+
+def test_current_id_tracks_the_latest_minted_id():
+    factory, _ = _counter()
+    tracker = SessionTracker(300.0, factory)
+
+    assert tracker.current_id is None
+    first = tracker.observe(100.0)
+    assert tracker.current_id == first
+    rolled = tracker.observe(100.0 + 301)
+    assert tracker.current_id == rolled

tests/test_tui_app.py

@@ -1326,6 +1326,34 @@ def test_screenshot_saves_svg(server, tmp_path, monkeypatch):
         "screenshot leaked into %s instead of screenshots/" % tmp_path
 
 
+def test_screenshot_saves_into_session_dir(server, tmp_path, monkeypatch):
+    pytest.importorskip("textual")
+    import glob
+
+    from stt_daemon.tui.app import AssistantTUI
+
+    monkeypatch.chdir(tmp_path)
+    path, _service = server
+    app = AssistantTUI(path)
+    app.state.session_dir = "transcripts/x"
+
+    async def scenario():
+        async with app.run_test() as pilot:
+            await pilot.pause()
+            await pilot.press("ctrl+s")
+            await pilot.pause()
+            await pilot.pause()
+
+    asyncio.run(scenario())
+    out_dir = tmp_path / "transcripts" / "x" / "screenshots"
+    assert out_dir.is_dir(), "session screenshots/ not created in %s" % out_dir
+    files = glob.glob(str(out_dir / "stt-tui-*.svg"))
+    assert files, "no stt-tui-*.svg found in %s" % out_dir
+    assert os.path.getsize(files[0]) > 0
+    assert glob.glob(str(tmp_path / "screenshots" / "stt-tui-*.svg")) == [], \
+        "screenshot leaked into screenshots/ instead of the session dir"
+
+
 def test_chat_hidden_by_default_triage_fills_main(server):
     """Fresh state: #chat hidden, #triage fills the whole #main width."""
     pytest.importorskip("textual")

tests/test_tui_state.py

@@ -949,3 +949,51 @@ def test_clear_output_keeps_questions():
     assert state.questions == _questions()
     assert state.questions_epoch == 0
     assert state.questions_lines()[0] == "topic  [choice]"
+
+
+def test_status_event_carrying_session_dir_sets_it():
+    state = ViewState()
+    assert state.session_dir is None
+    state.ingest({"type": "status", "session_dir": "transcripts/2026-09-30-12-47-29"})
+    assert state.session_dir == "transcripts/2026-09-30-12-47-29"
+
+
+def test_status_without_session_dir_preserves_prior_value():
+    state = ViewState()
+    state.ingest({"type": "status", "session_dir": "transcripts/2026-09-30-12-47-29"})
+    state.ingest({"type": "status", "mode": "idle", "ready": True})
+    assert state.session_dir == "transcripts/2026-09-30-12-47-29"
+
+
+def test_session_event_sets_session_dir():
+    state = ViewState()
+    state.ingest(
+        {
+            "type": "session",
+            "session_id": "2026-09-30-12-47-29",
+            "session_dir": "transcripts/2026-09-30-12-47-29",
+        }
+    )
+    assert state.session_dir == "transcripts/2026-09-30-12-47-29"
+
+
+def test_transcript_carrying_session_dir_sets_it():
+    state = ViewState()
+    state.ingest(
+        {
+            "type": "transcript",
+            "text": "hola",
+            "lang": "es",
+            "session_dir": "transcripts/2026-09-30-12-47-29",
+        }
+    )
+    assert state.session_dir == "transcripts/2026-09-30-12-47-29"
+
+
+def test_cleared_keeps_session_dir():
+    state = ViewState()
+    state.ingest({"type": "status", "session_dir": "transcripts/2026-09-30-12-47-29"})
+    state.ingest({"type": "chat_message", "role": "user", "text": "hello"})
+    state.ingest({"type": "cleared"})
+    assert state.session_dir == "transcripts/2026-09-30-12-47-29"
+    assert state.chat_lines() == []