Stream utterances through the speech queue

Replace the per-chunk queue model with whole streaming utterances so
concurrent projects never interleave. Each speak() reserves one ordering
slot up front and streams its synthesized chunks through an internal
channel; the consumer stays locked to that utterance from entry tone to
exit tone, preserving synth/playback pipelining without letting another
project's audio wedge between sentences.

Also folded in:

- Reap synthesized WAVs after playback (and on cancel/shutdown/queue-full),
  fixing an unbounded /tmp/mcspeak leak — one file per spoken sentence was
  never deleted.
- Secretary announcements: announce a project by name on speaker-change or
  after a lull, synthesized in a reserved voice (Kokoro) that is excluded
  from the project auto-assignment pool. Decision made at play time so
  urgent reordering is respected.
- Call-waiting: a brief blip mixed over the current message when a different
  project queues up, gated to once per playing turn so it never spams.
- Unify _speak_single/_speak_chunked into one _speak path.
- Status-aware progress: a queued item now reports "waiting in line" instead
  of racing to a false "playing 99%".
- Synchronous abort flag closes a cancel-race where a chunk synthesized
  during the cancel-tone window could leak past the reap.

src/mcspeak/queue.py

@@ -1,12 +1,20 @@
-"""Speech queue — producer-consumer pattern with priority and cancellation.
+"""Speech queue — producer-consumer with priority, streaming utterances, cancellation.
 
-Replaces the original asyncio.Lock approach. A dedicated consumer coroutine
+A dedicated consumer coroutine pulls whole *utterances* from a bounded
-pulls work items from a bounded PriorityQueue and plays them one at a time.
+PriorityQueue and plays them one at a time. Each utterance streams its
-Producers call enqueue() which returns immediately with a speech_id.
+synthesized chunks in through an internal channel, so playback of chunk N
-Use get_status() to check playback outcomes asynchronously.
+overlaps synthesis of chunk N+1 (pipelining) WITHOUT letting a second
+project's audio interleave: the consumer stays locked to a single utterance
+from its entry tone through its exit tone.
+
+This is what makes the queue behave like a secretary when several projects
+speak at once — each message is played whole, in order, one at a time, and
+a project is announced by name when the speaker changes (or returns after a
+lull).
 
 Priority tiers:
-    0 = urgent (preempts normal items in the queue)
+    0 = urgent (jumps ahead of *queued* normal utterances; never chops the
+        one already playing)
     1 = normal (default)
 """
 
@@ -31,30 +39,99 @@ class Priority(IntEnum):
 MAX_QUEUE_DEPTH = 20
 MAX_OUTCOMES = 100
 
+# Sentinel pushed onto an utterance's chunk channel to mark "no more chunks".
+_END = object()
+
+
+def _unlink_quietly(path: Path | str | None) -> None:
+    """Best-effort delete of a synthesized WAV. Swallows all OS errors.
+
+    Speech WAVs (chunks + the project preamble) are ephemeral — once pw-play
+    has read them off disk they're never needed again. Deleting them here is
+    what stops /tmp/mcspeak from growing one file per spoken sentence for the
+    container's whole lifetime. Shared tone WAVs (_tone-*.wav) are never
+    passed here, so they survive.
+    """
+    if path is None:
+        return
+    try:
+        Path(path).unlink(missing_ok=True)
+    except OSError:
+        pass
+
 
 @dataclass(order=True)
-class _WorkItem:
+class _Utterance:
-    """A prioritized playback request."""
+    """One speak() call — an ordered stream of synthesized chunks played as a unit.
+
+    Ordering (priority, sequence) is fixed the moment the utterance is created,
+    so all of its chunks travel together: nothing another project enqueues can
+    sort between them. Chunks arrive lazily through `chunks` (a channel closed
+    by an `_END` sentinel), which is what preserves synth/playback pipelining
+    without giving up per-message coherence.
+    """
 
     priority: int
-    sequence: int  # tiebreaker for same-priority items (FIFO)
+    sequence: int
-    result: TTSResult = field(compare=False)
-    future: asyncio.Future = field(compare=False)
     speech_id: str = field(default="", compare=False)
-    enqueued_at: float = field(default_factory=time.time, compare=False)
+    project: str | None = field(default=None, compare=False)
-    suppress_exit_tone: bool = field(default=False, compare=False)
+    engine: str = field(default="", compare=False)
-    # Entry tone deferred from speak() handler when the queue was busy at
+    voice: str = field(default="", compare=False)
-    # call time. Consumer plays this right before the item's audio so the
+    # Entry tone deferred from the speak() handler when the queue was busy at
-    # tone doesn't overlap a previously-playing item. None = no entry tone
+    # call time. Consumer plays it right before this utterance so it doesn't
-    # (either no tone configured, or already played immediately by the handler).
+    # overlap a previously-playing item. Shared tone WAV — never unlinked.
     entry_tone: Path | None = field(default=None, compare=False)
+    # Short "<project>." preamble in this utterance's own voice. The consumer
+    # decides at play time whether to actually play it (see _should_announce).
+    # Per-message WAV — unlinked whether or not it ends up played.
+    announce_audio: Path | None = field(default=None, compare=False)
+    future: asyncio.Future | None = field(default=None, compare=False)
+    enqueued_at: float = field(default_factory=time.time, compare=False)
+    chunks: asyncio.Queue = field(default_factory=asyncio.Queue, compare=False)
+    # Running total of chunk durations — grows as chunks stream in. Used for
+    # progress/shutdown ETA only, never for control flow.
+    duration_estimate: float = field(default=0.0, compare=False)
+    # Set synchronously the instant this utterance is cancelled/abandoned, so a
+    # producer mid-synthesis sees it BEFORE the consumer has finished resolving
+    # the future (which can lag behind a ~0.3s cancel tone). Without this, a
+    # chunk synthesized during that window would slip into an already-reaped
+    # queue and leak. Distinct from future.done() for exactly that timing reason.
+    _aborted: bool = field(default=False, compare=False)
+
+    def abort(self) -> None:
+        """Signal — synchronously — that nothing more should be synthesized."""
+        self._aborted = True
+
+    async def add_chunk(self, result: TTSResult) -> None:
+        """Producer: hand a synthesized chunk to the consumer.
+
+        If we've been aborted, drop and reap the chunk instead of enqueuing it —
+        the consumer has already moved on and won't reap anything added late.
+        """
+        if self.aborted:
+            _unlink_quietly(result.audio_path)
+            return
+        self.duration_estimate += result.duration_seconds
+        await self.chunks.put(result)
+
+    async def close(self) -> None:
+        """Producer: signal no more chunks. MUST be called (even on error)."""
+        await self.chunks.put(_END)
+
+    @property
+    def aborted(self) -> bool:
+        """True once cancelled/abandoned — producers poll this between chunks.
+
+        Either the explicit abort flag (set the moment cancel fires) or the
+        future resolving early trips it.
+        """
+        return self._aborted or (self.future is not None and self.future.done())
 
 
 class SpeechQueue:
     """Bounded priority queue with a dedicated playback consumer.
 
-    Call start() to launch the consumer, stop() to cancel it and
+    Call start() to launch the consumer, stop() to gracefully drain it.
-    drain pending items.
     """
 
     def __init__(
@@ -65,12 +142,15 @@ class SpeechQueue:
         cancel_tone: Path | None = None,
         shutdown_timeout: float = 30.0,
         ducker: MediaDucker | None = None,
+        announce_mode: str = "secretary",
+        reintroduce_after: float = 120.0,
+        call_waiting_tone: Path | None = None,
     ) -> None:
-        self._queue: asyncio.PriorityQueue[_WorkItem] = asyncio.PriorityQueue(
+        self._queue: asyncio.PriorityQueue[_Utterance] = asyncio.PriorityQueue(
             maxsize=max_depth
         )
         self._consumer_task: asyncio.Task | None = None
-        self._current: _WorkItem | None = None
+        self._current: _Utterance | None = None
         self._item_cancelled = False
         self._counter = 0
         self._max_depth = max_depth
@@ -82,6 +162,17 @@ class SpeechQueue:
         self._ducker = ducker
         self._outcomes: OrderedDict[str, dict] = OrderedDict()
         self._futures: dict[str, asyncio.Future] = {}
+        # Secretary state — who spoke last and when, for announcement decisions.
+        self._announce_mode = announce_mode
+        self._reintroduce_after = reintroduce_after
+        self._last_project: str | None = None
+        self._last_spoke_at: float = 0.0
+        # Call-waiting: a brief blip mixed over the current message when another
+        # project queues up. _call_waiting_fired gates it to once per playing
+        # turn (reset when a new utterance starts playing) so it never spams.
+        self._call_waiting_tone = call_waiting_tone
+        self._call_waiting_fired = False
+        self._bg_tasks: set[asyncio.Task] = set()
 
     def _next_speech_id(self) -> str:
         self._counter += 1
@@ -109,11 +200,37 @@ class SpeechQueue:
         """
         return self._current is None and self._queue.empty()
 
+    def should_prepare_announcement(self, project: str | None) -> bool:
+        """Cheap producer-side hint: is it worth synthesizing a preamble?
+
+        The real decision is made at play time in _should_announce (order isn't
+        final until then). This just lets the common case — one project talking
+        to itself with an empty queue — skip the wasted preamble synthesis.
+        Errs toward preparing whenever anything could reorder or the speaker
+        might have changed.
+        """
+        if project is None or self._announce_mode == "off":
+            return False
+        if self._announce_mode == "always":
+            return True
+        # secretary: only skip if we're confident the same project is
+        # continuing and nothing queued/playing could slip in front.
+        if (
+            self._last_project == project
+            and (time.time() - self._last_spoke_at) <= self._reintroduce_after
+            and self._current is None
+            and self._queue.empty()
+        ):
+            return False
+        return True
+
     def status(self) -> dict:
         return {
             "current_speaker": self.current_speaker,
+            "current_project": self._current.project if self._current else None,
             "queue_depth": self._queue.qsize(),
             "max_depth": self._max_depth,
+            "last_project": self._last_project,
         }
 
     def start(self) -> None:
@@ -125,21 +242,39 @@ class SpeechQueue:
             )
 
     def _drain_remaining(self) -> None:
-        """Drain pending items, recording shutdown outcomes. Sync — safe for finally."""
+        """Drain pending utterances, recording shutdown outcomes. Sync — safe for finally."""
         while not self._queue.empty():
             try:
-                item = self._queue.get_nowait()
+                utt = self._queue.get_nowait()
-                outcome = {
-                    "played": False,
-                    "error": "Queue shut down",
-                    "engine": item.result.engine,
-                    "voice": item.result.voice,
-                }
-                if not item.future.done():
-                    item.future.set_result(outcome)
-                self._record_outcome(item.speech_id, outcome)
             except asyncio.QueueEmpty:
                 break
+            outcome = {
+                "played": False,
+                "error": "Queue shut down",
+                "engine": utt.engine,
+                "voice": utt.voice,
+            }
+            if utt.future is not None and not utt.future.done():
+                utt.future.set_result(outcome)
+            self._record_outcome(utt.speech_id, outcome)
+            # Never played, so the consumer's cleanup never ran — reap files.
+            utt.abort()
+            self._reap_utterance_files(utt)
+
+    def _reap_utterance_files(self, utt: _Utterance) -> None:
+        """Delete an utterance's per-message WAVs that never got played.
+
+        The preamble plus any chunks already sitting in the channel. Best-effort
+        and synchronous — the entry tone is a shared WAV and is left alone.
+        """
+        _unlink_quietly(utt.announce_audio)
+        while True:
+            try:
+                chunk = utt.chunks.get_nowait()
+            except asyncio.QueueEmpty:
+                break
+            if chunk is not _END:
+                _unlink_quietly(chunk.audio_path)
 
     def _force_cancel_consumer(self) -> None:
         """Cancel consumer task if still running."""
@@ -149,17 +284,16 @@ class SpeechQueue:
     async def stop(self) -> None:
         """Gracefully stop the consumer — let current speech finish, then drain.
 
-        Waits up to shutdown_timeout seconds for the currently-playing item to
+        Waits up to shutdown_timeout seconds for the currently-playing utterance
-        complete. If it doesn't finish in time, falls back to hard cancel.
+        to complete. If it doesn't finish in time, falls back to hard cancel.
         Drain always runs, even if stop() itself is cancelled.
         """
         self._stopped = True
         try:
             if self._consumer_task and not self._consumer_task.done():
-                # Log what we're waiting for
                 if self._current:
                     elapsed = time.time() - self._current.enqueued_at
-                    remaining = max(0, self._current.result.duration_seconds - elapsed)
+                    remaining = max(0, self._current.duration_estimate - elapsed)
                     print(
                         f"Speech queue: waiting for current audio "
                         f"(~{remaining:.0f}s remaining, budget {self._shutdown_timeout:.0f}s)",
@@ -168,17 +302,14 @@ class SpeechQueue:
                 else:
                     print("Speech queue: consumer idle, shutting down", file=sys.stderr)
 
-                # Consumer polls _stopped every 0.5s between items, so it will
+                # Consumer polls _stopped every 0.5s between utterances, so it
-                # exit the while loop after finishing the current item.
+                # exits the while loop after finishing the current one.
                 try:
                     await asyncio.wait_for(
                         asyncio.shield(self._consumer_task),
                         timeout=self._shutdown_timeout,
                     )
-                    print(
+                    print("Speech queue: graceful shutdown complete", file=sys.stderr)
-                        "Speech queue: graceful shutdown complete",
-                        file=sys.stderr,
-                    )
                 except asyncio.TimeoutError:
                     print(
                         f"Speech queue: shutdown timeout ({self._shutdown_timeout:.0f}s) "
@@ -191,11 +322,9 @@ class SpeechQueue:
                     except asyncio.CancelledError:
                         pass
         except asyncio.CancelledError:
-            # stop() itself was cancelled — force-kill consumer before draining
             self._force_cancel_consumer()
             raise
         finally:
-            # Always restore media volumes on shutdown
             if self._ducker:
                 try:
                     await self._ducker.unduck()
@@ -203,186 +332,269 @@ class SpeechQueue:
                     pass  # Non-fatal
             self._drain_remaining()
 
-    def _resolve_outcome(self, item: _WorkItem, outcome: dict) -> None:
+    def _resolve_outcome(self, utt: _Utterance, outcome: dict) -> None:
         """Set the future result and record the outcome."""
-        if not item.future.done():
+        if utt.future is not None and not utt.future.done():
-            item.future.set_result(outcome)
+            utt.future.set_result(outcome)
-        self._record_outcome(item.speech_id, outcome)
+        self._record_outcome(utt.speech_id, outcome)
+
+    def _should_announce(self, utt: _Utterance) -> bool:
+        """Play-time decision: introduce this utterance's project by name?
+
+        Made here (not at enqueue) because urgent reordering means the real
+        speaker sequence isn't known until we pull the next utterance.
+        """
+        if utt.announce_audio is None or utt.project is None:
+            return False
+        if self._announce_mode == "off":
+            return False
+        if self._announce_mode == "always":
+            return True
+        # secretary: announce on speaker change, or same speaker after a lull.
+        if self._last_project != utt.project:
+            return True
+        return (time.time() - self._last_spoke_at) > self._reintroduce_after
+
+    async def _play_utterance(self, utt: _Utterance) -> None:
+        """Play one utterance whole: entry tone, announcement, chunks, exit tone.
+
+        Blocks on the chunk channel between chunks — that's the pipelining seam
+        (consumer plays chunk N while the producer synthesizes chunk N+1). No
+        other utterance can be pulled until this returns, so nothing interleaves.
+        """
+        # Deferred entry tone (queue was busy when speak() was called).
+        if utt.entry_tone:
+            try:
+                await play_audio(utt.entry_tone, expected_seconds=0.3)
+            except PlaybackError:
+                pass  # Non-fatal — shared tone WAV, don't unlink
+
+        # Secretary announcement — decided now that play order is final.
+        if utt.announce_audio:
+            if self._should_announce(utt):
+                try:
+                    await play_audio(utt.announce_audio, expected_seconds=0.6)
+                except PlaybackError:
+                    pass
+            _unlink_quietly(utt.announce_audio)  # reap whether played or skipped
+
+        # Stream chunks until the producer closes the channel.
+        while True:
+            chunk = await utt.chunks.get()
+            if chunk is _END:
+                break
+            try:
+                await play_audio(
+                    chunk.audio_path, expected_seconds=chunk.duration_seconds
+                )
+            finally:
+                _unlink_quietly(chunk.audio_path)
+
+        # Exit tone once, at the very end — "roger" if the queue is now empty,
+        # "standby" if more utterances are waiting.
+        exit_t = (
+            self._exit_tone if self._queue.qsize() == 0 else self._standby_tone
+        )
+        if exit_t:
+            try:
+                await play_audio(exit_t, expected_seconds=0.3)
+            except PlaybackError:
+                pass
+
+        # Remember who just spoke — drives the next utterance's announcement.
+        if utt.project is not None:
+            self._last_project = utt.project
+        self._last_spoke_at = time.time()
+
+        self._resolve_outcome(utt, {
+            "played": True,
+            "engine": utt.engine,
+            "voice": utt.voice,
+            "project": utt.project,
+            "duration_seconds": utt.duration_estimate,
+        })
 
     async def _consumer(self) -> None:
-        """Pull work items and play them sequentially.
+        """Pull utterances and play them sequentially, one whole message at a time.
 
         The _item_cancelled flag distinguishes cancel(speech_id) (item-level,
         continue loop) from shutdown cancellation (exit loop).
         """
         while not self._stopped:
             try:
-                item = await asyncio.wait_for(self._queue.get(), timeout=0.5)
+                utt = await asyncio.wait_for(self._queue.get(), timeout=0.5)
             except asyncio.TimeoutError:
                 continue  # Re-check _stopped flag
             except asyncio.CancelledError:
                 break
 
-            if item.future.cancelled():
+            if utt.future is not None and utt.future.cancelled():
+                # Cancelled while queued — skip it, reap its files.
                 self._queue.task_done()
+                utt.abort()
+                self._reap_utterance_files(utt)
                 continue
 
-            self._current = item
+            self._current = utt
             self._item_cancelled = False
+            # New turn — allow one call-waiting blip if others queue during it.
+            self._call_waiting_fired = False
 
             try:
-                # Deferred entry tone — speak() handler deferred this when the
+                await self._play_utterance(utt)
-                # queue was busy at call time. Play it now so it lands right
-                # before this item's audio (no overlap with prior playback).
-                if item.entry_tone:
-                    try:
-                        await play_audio(item.entry_tone, expected_seconds=0.3)
-                    except PlaybackError:
-                        pass  # Non-fatal
-
-                await play_audio(
-                    item.result.audio_path,
-                    expected_seconds=item.result.duration_seconds,
-                )
-
-                # Play exit tone — "standby" if more queued, "roger" if done
-                # suppress_exit_tone skips tones between chunks of the same message
-                if not item.suppress_exit_tone:
-                    exit_t = (
-                        self._exit_tone if self._queue.qsize() == 0
-                        else self._standby_tone
-                    )
-                    if exit_t:
-                        try:
-                            await play_audio(exit_t, expected_seconds=0.3)
-                        except PlaybackError:
-                            pass  # Non-fatal
-
-                self._resolve_outcome(item, {
-                    "played": True,
-                    "file": str(item.result.audio_path),
-                    "duration_seconds": item.result.duration_seconds,
-                    "engine": item.result.engine,
-                    "voice": item.result.voice,
-                })
-
-            except PlaybackError as e:
-                self._resolve_outcome(item, {
-                    "played": False,
-                    "error": str(e),
-                    "file": str(item.result.audio_path),
-                    "engine": item.result.engine,
-                    "voice": item.result.voice,
-                })
 
             except asyncio.CancelledError:
                 if self._item_cancelled:
-                    # Item-level cancel via cancel() — play cancel tone, continue
+                    # Item-level cancel via cancel() — cancel tone, then continue.
                     self._item_cancelled = False
+                    utt.abort()
+                    self._reap_utterance_files(utt)
                     if self._cancel_tone:
                         try:
                             await play_audio(self._cancel_tone, expected_seconds=0.3)
                         except (PlaybackError, asyncio.CancelledError):
                             pass
-                    self._resolve_outcome(item, {
+                    self._resolve_outcome(utt, {
                         "played": False,
                         "error": "Cancelled by client",
-                        "engine": item.result.engine,
+                        "engine": utt.engine,
-                        "voice": item.result.voice,
+                        "voice": utt.voice,
                     })
-                    # Don't break — continue to next item
+                    # Fall through to finally, keep looping.
                 else:
-                    # Shutdown — record and exit loop
+                    # Shutdown — record and exit loop.
-                    self._resolve_outcome(item, {
+                    utt.abort()
+                    self._reap_utterance_files(utt)
+                    self._resolve_outcome(utt, {
                         "played": False,
                         "error": "Playback cancelled",
-                        "engine": item.result.engine,
+                        "engine": utt.engine,
-                        "voice": item.result.voice,
+                        "voice": utt.voice,
                     })
                     break
 
             except Exception as e:
-                self._resolve_outcome(item, {
+                utt.abort()
+                self._reap_utterance_files(utt)
+                self._resolve_outcome(utt, {
                     "played": False,
                     "error": f"Unexpected error: {e}",
-                    "engine": item.result.engine,
+                    "engine": utt.engine,
-                    "voice": item.result.voice,
+                    "voice": utt.voice,
                 })
 
             finally:
                 self._current = None
                 self._queue.task_done()
-                # Unduck media when queue is empty — covers ALL exit paths
+                # Unduck media when queue is empty — covers ALL exit paths.
-                # (normal completion, PlaybackError, cancel, generic exception)
                 if self._queue.qsize() == 0 and self._ducker and not self._stopped:
                     try:
                         await self._ducker.unduck()
                     except Exception:
                         pass  # Non-fatal
 
-    async def enqueue(
+    def create_utterance(
         self,
-        result: TTSResult,
         priority: Priority = Priority.NORMAL,
-        suppress_exit_tone: bool = False,
+        project: str | None = None,
+        engine: str = "",
+        voice: str = "",
         entry_tone: Path | None = None,
-    ) -> dict:
+        announce_audio: Path | None = None,
-        """Enqueue a TTSResult for playback. Returns immediately with enqueue metadata.
+    ) -> tuple[_Utterance | None, dict]:
+        """Reserve a queue slot for a new utterance and register it for playback.
 
-        Does NOT block until playback finishes — use get_status(speech_id) to
+        Returns (utterance, meta). On success `utterance` is the _Utterance to
-        check the outcome later.
+        feed chunks into (add_chunk/close) and `meta` carries the speech_id and
+        queue position. On failure (shut down / full) `utterance` is None and
+        `meta` is an error dict — the caller should reap any announce_audio it
+        pre-synthesized.
 
-        If entry_tone is given, the consumer plays it immediately before this
+        The utterance goes into the priority queue immediately (before its
-        item's audio (used when speak() handler deferred the tone because the
+        chunks exist), which fixes its ordering vs. concurrent producers. The
-        queue was busy at call time).
+        consumer that pulls it then blocks on the chunk channel until chunks
-
+        arrive, so a slow synth just delays that one message — it never lets a
-        Raises QueueFull if the queue is at max capacity (backpressure).
+        faster producer's audio interleave.
         """
         if self._stopped:
-            return {
+            _unlink_quietly(announce_audio)
+            return None, {
                 "queued": False,
                 "error": "Queue is shut down",
-                "engine": result.engine,
+                "engine": engine,
-                "voice": result.voice,
+                "voice": voice,
             }
 
         speech_id = self._next_speech_id()
         loop = asyncio.get_running_loop()
         future: asyncio.Future = loop.create_future()
 
-        item = _WorkItem(
+        utt = _Utterance(
             priority=priority,
             sequence=self._counter,
-            result=result,
-            future=future,
             speech_id=speech_id,
-            suppress_exit_tone=suppress_exit_tone,
+            project=project,
+            engine=engine,
+            voice=voice,
             entry_tone=entry_tone,
+            announce_audio=announce_audio,
+            future=future,
         )
 
         try:
-            self._queue.put_nowait(item)
+            self._queue.put_nowait(utt)
         except asyncio.QueueFull:
-            return {
+            _unlink_quietly(announce_audio)
+            return None, {
                 "queued": False,
                 "error": f"Queue full ({self._max_depth} items). Try again later.",
                 "speech_id": speech_id,
-                "engine": result.engine,
+                "engine": engine,
-                "voice": result.voice,
+                "voice": voice,
             }
 
         self._futures[speech_id] = future
-
+        self._maybe_fire_call_waiting(utt)
-        return {
+        return utt, {
             "speech_id": speech_id,
             "queued": True,
             "position": self._queue.qsize(),
-            "file": str(result.audio_path),
+            "engine": engine,
-            "duration_seconds": result.duration_seconds,
+            "voice": voice,
-            "engine": result.engine,
-            "voice": result.voice,
         }
 
+    def _maybe_fire_call_waiting(self, new_utt: _Utterance) -> None:
+        """Blip once over the current message when a DIFFERENT project queues up.
+
+        Gated to once per playing turn (_call_waiting_fired, reset when a new
+        utterance starts playing) so a burst of queued messages doesn't
+        machine-gun the listener. Fires as a background task — a second pw-play
+        stream that PipeWire mixes over the currently-playing audio.
+        """
+        cur = self._current
+        if (
+            self._call_waiting_tone is None
+            or self._call_waiting_fired
+            or cur is None
+            or new_utt.project is None
+            or cur.project is None
+            or new_utt.project == cur.project
+        ):
+            return
+        self._call_waiting_fired = True
+        task = asyncio.create_task(self._play_call_waiting())
+        self._bg_tasks.add(task)
+        task.add_done_callback(self._bg_tasks.discard)
+
+    async def _play_call_waiting(self) -> None:
+        """Play the call-waiting blip over the current audio. Non-fatal."""
+        if self._call_waiting_tone is None:
+            return
+        try:
+            await play_audio(self._call_waiting_tone, expected_seconds=0.2)
+        except Exception:
+            pass
+
     async def wait_for_completion(self, speech_id: str) -> dict:
         """Await playback completion for a specific speech_id.
 
@@ -391,25 +603,28 @@ class SpeechQueue:
         """
         future = self._futures.get(speech_id)
         if future is None:
-            # Already completed and cleaned up, or unknown
             return self._outcomes.get(speech_id, {"status": "unknown", "speech_id": speech_id})
         return await future
 
     def cancel(self, speech_id: str) -> dict:
-        """Cancel a queued or playing speech item.
+        """Cancel a queued or playing utterance.
 
-        If the item is currently playing, kills pw-play and the consumer plays
+        If it's currently playing, kills pw-play and the consumer plays the
-        the cancel tone. If queued, marks the future as cancelled so the
+        cancel tone. If queued, marks the future cancelled so the consumer skips
-        consumer skips it. Returns cancellation status.
+        it (and the producer stops synthesizing once it sees `aborted`).
         """
-        # Currently playing — cancel via the consumer task
+        # Currently playing — cancel via the consumer task.
         if self._current and self._current.speech_id == speech_id:
             if self._consumer_task and not self._consumer_task.done():
+                # Flag abort NOW (synchronously) so a producer still synthesizing
+                # stops before the consumer's cancel-tone delay lets a late chunk
+                # leak past the reap.
+                self._current.abort()
                 self._item_cancelled = True
                 self._consumer_task.cancel()
                 return {"cancelled": True, "was": "playing", "speech_id": speech_id}
 
-        # Queued — cancel the future (consumer skips cancelled futures)
+        # Queued — cancel the future (consumer skips cancelled utterances).
         future = self._futures.get(speech_id)
         if future and not future.done():
             future.cancel()
@@ -427,36 +642,35 @@ class SpeechQueue:
         }
 
     def get_status(self, speech_id: str) -> dict:
-        """Check the status of a speech item by its ID.
+        """Check the status of an utterance by its ID.
 
-        Returns a dict with 'status' key: 'completed', 'playing', 'queued', or 'unknown'.
+        Returns a dict with 'status': 'completed', 'playing', 'queued', or 'unknown'.
         """
-        # Check completed outcomes
         if speech_id in self._outcomes:
             outcome = self._outcomes[speech_id]
             return {"status": "completed", **outcome}
 
-        # Check currently playing
         if self._current and self._current.speech_id == speech_id:
             return {
                 "status": "playing",
                 "speech_id": speech_id,
-                "engine": self._current.result.engine,
+                "project": self._current.project,
-                "voice": self._current.result.voice,
+                "engine": self._current.engine,
-                "file": str(self._current.result.audio_path),
+                "voice": self._current.voice,
-                "duration_seconds": self._current.result.duration_seconds,
+                "duration_seconds": self._current.duration_estimate,
             }
 
-        # Scan queue (PriorityQueue._queue is the underlying heap list)
+        # Scan queue (PriorityQueue._queue is the underlying heap list).
         try:
-            for pos, item in enumerate(self._queue._queue, start=1):
+            for pos, utt in enumerate(self._queue._queue, start=1):
-                if item.speech_id == speech_id:
+                if utt.speech_id == speech_id:
                     return {
                         "status": "queued",
                         "speech_id": speech_id,
                         "position": pos,
-                        "engine": item.result.engine,
+                        "project": utt.project,
-                        "voice": item.result.voice,
+                        "engine": utt.engine,
+                        "voice": utt.voice,
                     }
         except AttributeError:
             pass  # Internal API — degrade gracefully

src/mcspeak/server.py

@@ -13,9 +13,12 @@ from fastmcp.server.dependencies import CurrentContext
 from mcp.types import ToolAnnotations
 
 from .audio import (
-    ConversionError, SUPPORTED_FORMATS, convert_audio, play_audio, record_audio,
+    ConversionError,
+    convert_audio,
+    play_audio,
+    record_audio,
+    record_audio_until_silence,
 )
-from .transcribe import TranscriptionError, transcribe_audio
 from .engines.base import TTSEngine
 from .engines.kokoro import KokoroEngine
 from .engines.orpheus import OrpheusEngine
@@ -24,6 +27,7 @@ from .media_duck import MediaDucker
 from .queue import Priority, SpeechQueue
 from .settings import settings
 from .tones import generate_tones, resolve_tone
+from .transcribe import TranscriptionError, transcribe_audio
 from .voice_identity import VoiceIdentityCache, get_project_name, resolve_voice
 
 ENGINE_NAMES = Literal["piper", "kokoro", "orpheus"]
@@ -145,10 +149,22 @@ async def app_lifespan(server: FastMCP):
 
     # Generate tones and resolve which ones to use
     tone_paths = generate_tones(settings.output_dir)
-    entry_tone = resolve_tone(settings.entry_tone, tone_paths, "entry_tone", default="chirp")
+    entry_tone = resolve_tone(settings.entry_tone, tone_paths, "entry_tone", default="heartbeat")
-    exit_tone = resolve_tone(settings.exit_tone, tone_paths, "exit_tone", default="roger")
+    exit_tone = resolve_tone(settings.exit_tone, tone_paths, "exit_tone", default="soft-chord")
     cancel_tone = resolve_tone(settings.cancel_tone, tone_paths, "cancel_tone", default="scratch")
     standby_tone = tone_paths.get("standby")
+    # Call-waiting blip mixed over the current message when another project queues.
+    call_waiting_tone = resolve_tone(
+        settings.call_waiting_tone, tone_paths, "call_waiting_tone", default="call-waiting",
+    )
+    # listen() bookends — distinct from speak's chirp/roger so the user can
+    # tell capture mode from playback mode by ear alone.
+    listen_start_tone = resolve_tone(
+        settings.listen_start_tone, tone_paths, "listen_start_tone", default="mf-dial",
+    )
+    listen_end_tone = resolve_tone(
+        settings.listen_end_tone, tone_paths, "listen_end_tone", default="machine",
+    )
 
     # --- Media ducker ---
     ducker: MediaDucker | None = None
@@ -158,12 +174,18 @@ async def app_lifespan(server: FastMCP):
             fade_in_ms=settings.duck_fade_in_ms,
         )
 
+    # Legacy announce_project=True maps to the "always" mode; otherwise use
+    # announce_mode (default "secretary": announce on speaker change / after a lull).
+    announce_mode = "always" if settings.announce_project else settings.announce_mode
     queue = SpeechQueue(
         exit_tone=exit_tone,
         standby_tone=standby_tone,
         cancel_tone=cancel_tone,
         shutdown_timeout=settings.shutdown_timeout,
         ducker=ducker,
+        announce_mode=announce_mode,
+        reintroduce_after=settings.reintroduce_after_seconds,
+        call_waiting_tone=call_waiting_tone,
     )
     queue.start()
 
@@ -181,7 +203,7 @@ async def app_lifespan(server: FastMCP):
     voice_cache = VoiceIdentityCache(persist_path=settings.voice_identity_file)
 
     identity_label = "on" if settings.voice_identity else "off"
-    announce_label = "+announce" if settings.announce_project else ""
+    announce_label = f"+announce:{announce_mode}" if announce_mode != "off" else ""
     print(
         f"  Voice identity: {identity_label}{announce_label}",
         file=sys.stderr,
@@ -199,6 +221,8 @@ async def app_lifespan(server: FastMCP):
             "voice_cache": voice_cache,
             "entry_tone": entry_tone,
             "ducker": ducker,
+            "listen_start_tone": listen_start_tone,
+            "listen_end_tone": listen_end_tone,
         }
     finally:
         print("McSpeak shutting down", file=sys.stderr)
@@ -227,6 +251,14 @@ mcp = FastMCP(
         "~0.5s during playback so clients can show a progress indicator.\n\n"
         "If you need to check whether speech finished, use speech_status(speech_id) "
         "after the fact. Use generate_audio to synthesize without playing.\n\n"
+        "TALKING WITH THE PERSON (voice conversation): when you speak something "
+        "that expects a spoken reply — a question, a confirmation prompt, 'anything "
+        "else?' — follow the speak() with a listen() call to capture their answer. "
+        "Do this SEQUENTIALLY, not in parallel: let speak() finish (it blocks until "
+        "playback is done), THEN call listen(), so the mic doesn't record your own "
+        "voice. Prefer listen(wait_for_silence=True) so recording stops when they "
+        "stop talking. Rule of thumb: if you'd expect a human to answer out loud, "
+        "listen for it — don't just speak and move on.\n\n"
         "Engines: kokoro (fast ONNX, ~50 voices), piper (Wyoming/Docker), "
         "orpheus (LLM via llama-server, supports <laugh> etc.). "
         "Always pass project= with the current project directory name (last path component) "
@@ -300,59 +332,74 @@ async def _resolve_project_voice(
     engine_name: str,
     eng: TTSEngine,
     voice: str | None,
-    text: str,
     voice_cache: VoiceIdentityCache,
     project: str | None = None,
-) -> tuple[str | None, str]:
+) -> tuple[str | None, str | None]:
-    """Auto-assign voice from project hint or MCP roots, optionally prefix text."""
+    """Resolve (voice, project_name) for this call.
-    if voice is not None or not settings.voice_identity:
-        return voice, text
 
-    # Explicit project param takes priority, then try MCP roots
+    The project name is resolved regardless of whether a voice was passed
-    project_name = project or await get_project_name(ctx)
+    explicitly, because it drives the consumer's spoken announcements. A voice
-    if not project_name:
+    is auto-assigned only when none was given and voice identity is enabled.
-        return voice, text
+    Announcements are handled downstream as a synthesized preamble (not by
+    prefixing the text), so this no longer mutates `text`.
+    """
+    # Resolve the project name: explicit param wins, else try MCP roots.
+    project_name = project
+    if project_name is None and settings.voice_identity:
+        project_name = await get_project_name(ctx)
 
-    voice = await resolve_voice(engine_name, eng, project_name, voice_cache)
+    # Auto-assign a voice only if the caller didn't pin one.
-    if settings.announce_project and voice is not None:
+    if voice is None and settings.voice_identity and project_name:
-        text = f"{project_name}. {text}"
+        voice = await resolve_voice(engine_name, eng, project_name, voice_cache)
 
-    return voice, text
+    return voice, project_name
 
 
 # ---------------------------------------------------------------------------
 # Tools
 # ---------------------------------------------------------------------------
 
-async def _speak_single(
+async def _speak(
     eng: TTSEngine,
     engine: str,
     text: str,
     voice: str | None,
+    project: str | None,
     queue: SpeechQueue,
     entry_tone: Path | None,
     priority: Priority,
     ctx: Context,
     ducker: MediaDucker | None = None,
+    announce_eng: TTSEngine | None = None,
+    secretary_voice: str = "",
 ) -> dict:
-    """Single-shot speak path — synthesize full text, then enqueue.
+    """Stream one utterance's chunks into the queue as a single coherent unit.
 
-    Used for short texts (< 20 words) and as the fallback when text has
+    Short texts are one chunk; long texts split into sentence chunks that
-    no sentence boundaries.
+    pipeline (synth of chunk N+1 overlaps playback of chunk N). Either way the
+    whole message is ONE queue utterance, so when several projects speak at
+    once nothing interleaves: each message plays whole, in order.
+
+    A short "<project>." preamble is synthesized in this utterance's own voice
+    and handed to the consumer, which decides at play time whether to actually
+    announce (speaker changed, or same speaker after a lull).
     """
-    # Entry tone gating — idle queue plays tone now (hides synth latency);
+    chunks = split_text(text)
-    # busy queue defers tone to the consumer so it doesn't overlap whatever's
+    n = len(chunks)
-    # already playing. Capture idle state before duck/synth changes it.
+
+    # Entry tone gating: idle queue plays the tone now (hides synth latency);
+    # busy queue defers it to the consumer so it lands right before this
+    # utterance instead of over whatever's already playing.
     queue_was_idle = queue.is_idle()
     deferred_entry_tone: Path | None = (
         None if queue_was_idle or entry_tone is None else entry_tone
     )
 
     try:
-        # Duck external media with crossfade into entry tone
+        # Duck external media with crossfade into the entry tone.
         if ducker:
             duck_task = asyncio.create_task(ducker.duck())
-            await asyncio.sleep(0.15)  # Let fade start — media dips to ~70%
+            await asyncio.sleep(0.15)  # Let fade start, media dips to ~70%
             if entry_tone and queue_was_idle:
                 await _play_tone(entry_tone)  # Tone crossfades over fading media
             await duck_task  # Ensure duck completes
@@ -360,155 +407,80 @@ async def _speak_single(
             await _play_tone(entry_tone)
         await ctx.report_progress(progress=5, total=100, message="Entry tone")
 
-        await ctx.info(f"Synthesizing with {engine}...")
+        # Secretary preamble: a short "<project>." spoken in the reserved
-        result = await eng.synthesize(text, voice)
+        # secretary voice (always via Kokoro, so it sounds identical no matter
-        await ctx.report_progress(progress=30, total=100, message="Synthesizing...")
+        # which engine the project uses). Synthesized here, played (or skipped)
+        # by the consumer. Skip the synthesis entirely when we're confident it
+        # won't be needed (same project talking to itself, empty queue).
+        announce_audio: Path | None = None
+        if project and secretary_voice and queue.should_prepare_announcement(project):
+            try:
+                display = project.replace("-", " ").replace("_", " ")
+                announce_res = await (announce_eng or eng).synthesize(
+                    f"{display}.", secretary_voice,
+                )
+                announce_audio = announce_res.audio_path
+            except Exception as e:
+                await ctx.info(f"Announcement synth failed (non-fatal): {e}")
 
-        await ctx.info(f"Synthesized {result.duration_seconds:.1f}s audio, enqueueing...")
+        # Reserve the queue slot NOW: this fixes ordering vs. concurrent
-        enqueue_result = await queue.enqueue(
+        # producers before any chunk exists. The consumer that pulls it blocks
-            result, priority=priority, entry_tone=deferred_entry_tone,
+        # on the chunk channel until chunks arrive, so a slow synth only delays
+        # this one message; it never lets another project's audio slip in.
+        utt, meta = queue.create_utterance(
+            priority=priority,
+            project=project,
+            engine=engine,
+            voice=voice or eng.default_voice,
+            entry_tone=deferred_entry_tone,
+            announce_audio=announce_audio,
         )
+        if utt is None:
+            # Queue full / shut down: create_utterance reaped the preamble.
+            return meta
 
-        if not enqueue_result.get("queued"):
+        # Stream chunks: synthesize each and hand it straight to the consumer.
-            return enqueue_result
+        try:
+            for i, chunk in enumerate(chunks):
+                if utt.aborted:
+                    break  # cancelled or shut down mid-synthesis, stop early
+                pct_synth = 5 + int(25 * (i + 1) / n)  # 5-30% across all chunks
+                label = (
+                    f"Synthesizing with {engine}..." if n == 1
+                    else f"Synthesizing chunk {i + 1}/{n} with {engine}..."
+                )
+                await ctx.info(label)
+                result = await eng.synthesize(chunk, voice)
+                await utt.add_chunk(result)
+                await ctx.report_progress(
+                    progress=pct_synth, total=100,
+                    message=("Synthesizing..." if n == 1
+                             else f"Synthesizing chunk {i + 1}/{n}..."),
+                )
+        finally:
+            # ALWAYS close the channel, otherwise the consumer blocks forever
+            # waiting for a next chunk that never comes (synth error, or this
+            # handler was cancelled mid-loop).
+            await utt.close()
 
-        speech_id = enqueue_result["speech_id"]
+        await ctx.info("Enqueued for playback")
-        duration = result.duration_seconds
+        # NOTE: Do NOT use asyncio.wait_for() - see CLAUDE.md Python 3.13 pitfall.
-        await ctx.report_progress(
+        # _await_with_progress owns 35→100 and distinguishes "queued/waiting"
-            progress=35, total=100,
+        # from "actually playing" so a queued item doesn't falsely report 99%.
-            message="Playing audio \u2014 you can continue working",
+        outcome = await _await_with_progress(
+            queue, utt.speech_id, utt.duration_estimate, ctx, 35,
         )
-
-        await ctx.info("Playing...")
-        # NOTE: Do NOT use asyncio.wait_for() — see CLAUDE.md Python 3.13 pitfall
-        outcome = await _await_with_progress(queue, speech_id, duration, ctx, 35)
-
         await ctx.report_progress(progress=100, total=100, message="Playback complete")
+        if n > 1:
+            outcome["chunks"] = n
         return outcome
 
     except asyncio.CancelledError:
         # speak() was cancelled (server shutdown or MCP client disconnect).
-        # Do NOT cancel the consumer — let it finish the current audio.
+        # The finally above already closed the utterance, so the consumer drains
-        # Shutdown: queue.stop() in the lifespan finalizer handles graceful drain.
+        # the chunks it has and finishes cleanly. Do NOT cancel the consumer:
-        # Client cancel: use cancel_speech(speech_id) for explicit mid-playback stop.
+        # already-synthesized audio should play through. For explicit
-        raise
+        # mid-playback stop, use cancel_speech(speech_id).
-
-
-async def _speak_chunked(
-    eng: TTSEngine,
-    engine: str,
-    chunks: list[str],
-    voice: str | None,
-    queue: SpeechQueue,
-    entry_tone: Path | None,
-    priority: Priority,
-    ctx: Context,
-    ducker: MediaDucker | None = None,
-) -> dict:
-    """Chunked speak path — pipeline synthesis with playback.
-
-    Synthesizes sentence chunks one at a time and enqueues each immediately.
-    Playback of chunk N overlaps synthesis of chunk N+1. Non-final chunks
-    suppress exit/standby tones for seamless playback.
-    """
-    n_chunks = len(chunks)
-    total_duration = 0.0
-    all_speech_ids: list[str] = []
-    first_enqueue_time: float | None = None
-
-    # Entry tone gating (see _speak_single for full rationale). Only the FIRST
-    # chunk carries the deferred tone — subsequent chunks are continuations of
-    # the same message and don't get their own tone.
-    queue_was_idle = queue.is_idle()
-    deferred_entry_tone: Path | None = (
-        None if queue_was_idle or entry_tone is None else entry_tone
-    )
-
-    try:
-        # Duck external media with crossfade into entry tone
-        if ducker:
-            duck_task = asyncio.create_task(ducker.duck())
-            await asyncio.sleep(0.15)  # Let fade start — media dips to ~70%
-            if entry_tone and queue_was_idle:
-                await _play_tone(entry_tone)  # Tone crossfades over fading media
-            await duck_task  # Ensure duck completes
-        elif entry_tone and queue_was_idle:
-            await _play_tone(entry_tone)
-        await ctx.report_progress(progress=5, total=100, message="Entry tone")
-
-        for i, chunk in enumerate(chunks):
-            is_last = i == n_chunks - 1
-            pct_synth = 5 + int(25 * (i + 1) / n_chunks)  # 5-30% across all chunks
-
-            await ctx.info(f"Synthesizing chunk {i + 1}/{n_chunks} with {engine}...")
-            result = await eng.synthesize(chunk, voice)
-            total_duration += result.duration_seconds
-            await ctx.report_progress(
-                progress=pct_synth, total=100,
-                message=f"Synthesizing chunk {i + 1}/{n_chunks}...",
-            )
-
-            enqueue_result = await queue.enqueue(
-                result,
-                priority=priority,
-                suppress_exit_tone=not is_last,
-                entry_tone=deferred_entry_tone if i == 0 else None,
-            )
-
-            if not enqueue_result.get("queued"):
-                # Queue full mid-message — already-enqueued chunks play through
-                if all_speech_ids:
-                    msg = f"Queue full at chunk {i + 1}/{n_chunks}, waiting for enqueued chunks..."
-                    await ctx.info(msg)
-                    elapsed = time.time() - first_enqueue_time if first_enqueue_time else 0
-                    remaining = max(total_duration - elapsed, 0.1)
-                    outcome = await _await_with_progress(
-                        queue, all_speech_ids[-1], remaining, ctx, 35
-                    )
-                    outcome["chunks_enqueued"] = len(all_speech_ids)
-                    outcome["chunks_total"] = n_chunks
-                    return outcome
-                return enqueue_result
-
-            all_speech_ids.append(enqueue_result["speech_id"])
-            if first_enqueue_time is None:
-                first_enqueue_time = time.time()
-
-        # All chunks enqueued — wait for final chunk to finish playing
-        if not all_speech_ids:
-            return {"error": "No chunks were successfully enqueued", "chunks_total": n_chunks}
-
-        await ctx.report_progress(
-            progress=35, total=100,
-            message="Playing audio \u2014 you can continue working",
-        )
-        await ctx.info(f"Playing {n_chunks} chunks ({total_duration:.1f}s total)...")
-
-        # Subtract time already elapsed since first chunk started playing
-        elapsed = time.time() - first_enqueue_time if first_enqueue_time else 0
-        remaining = max(total_duration - elapsed, 0.1)
-        outcome = await _await_with_progress(queue, all_speech_ids[-1], remaining, ctx, 35)
-        await ctx.report_progress(progress=100, total=100, message="Playback complete")
-
-        # Check for partial failures in earlier chunks
-        failed_chunks = []
-        for idx, sid in enumerate(all_speech_ids[:-1]):
-            chunk_status = queue.get_status(sid)
-            if chunk_status.get("status") == "completed" and not chunk_status.get("played", True):
-                failed_chunks.append(idx + 1)
-
-        outcome["chunks"] = n_chunks
-        outcome["total_duration_seconds"] = total_duration
-        if failed_chunks:
-            outcome["partial_failure"] = True
-            outcome["failed_chunks"] = failed_chunks
-        return outcome
-
-    except asyncio.CancelledError:
-        # speak() was cancelled (server shutdown or MCP client disconnect).
-        # Do NOT cancel the consumer — let already-enqueued chunks play through.
-        # Shutdown: queue.stop() in the lifespan finalizer handles graceful drain.
-        # Client cancel: use cancel_speech(speech_id) for explicit mid-playback stop.
         raise
 
 
@@ -519,21 +491,45 @@ async def _await_with_progress(
     ctx: Context,
     base_pct: int,
 ) -> dict:
-    """Await playback completion with a progress ticker.
+    """Await playback completion with a status-aware progress ticker.
 
-    Uses a background task for progress — NOT asyncio.wait_for polling
+    Uses a background task for progress \u2014 NOT asyncio.wait_for polling
     (see CLAUDE.md for the Python 3.13 pitfall).
+
+    The ticker distinguishes a WAITING item from a PLAYING one by polling
+    queue.get_status(): while the item is still queued behind others it holds
+    at base_pct with a "waiting in line" message; the playing percentage only
+    advances from the moment it actually becomes the current speaker. Without
+    this, an item queued behind a long message raced to "Playing audio 99%"
+    while it was silent and just waiting its turn.
     """
-    async def _progress_ticker():
+    async def _report(play_started):
-        t0 = time.time()
+        status = queue.get_status(speech_id)
-        while True:
+        st = status.get("status")
-            await asyncio.sleep(0.5)
+        if st == "queued":
-            elapsed = time.time() - t0
+            pos = status.get("position")
+            msg = (
+                f"Waiting in the queue (about {pos} ahead) \u2014 playback hasn't started"
+                if pos else "Waiting in the queue \u2014 playback hasn't started"
+            )
+            await ctx.report_progress(progress=base_pct, total=100, message=msg)
+        elif st == "playing":
+            if play_started is None:
+                play_started = time.time()  # clock starts when audio actually starts
+            elapsed = time.time() - play_started
             pct = min(base_pct + int((99 - base_pct) * elapsed / max(duration, 0.1)), 99)
             await ctx.report_progress(
                 progress=pct, total=100,
                 message="Playing audio \u2014 you can continue working",
             )
+        # completed/unknown: nothing to report; wait_for_completion resolves it.
+        return play_started
+
+    async def _progress_ticker():
+        play_started = await _report(None)  # immediate first report
+        while True:
+            await asyncio.sleep(0.5)
+            play_started = await _report(play_started)
 
     ticker = asyncio.create_task(_progress_ticker())
     try:
@@ -562,11 +558,22 @@ async def speak(
 ) -> dict:
     """Synthesize text and play it through the host speakers.
 
-    CALL THIS IN PARALLEL with other tools — don't block on it alone.
+    Two ways to use this:
-    Audio plays through physical speakers, not into your context. The return
+
-    value (speech_id, duration) is informational only and never needed for
+    1. MILESTONE PING (default) — a status update the person overhears ("build
-    subsequent reasoning. Call speak alongside your next tool in the same
+       done", "tests green"). CALL THIS IN PARALLEL with your next tool; don't
-    message and let progress notifications track playback.
+       block on it alone. The return value (speech_id, duration) is informational
+       and never needed for subsequent reasoning.
+
+    2. VOICE CONVERSATION — you asked the person something and want their spoken
+       answer. Then speak() and listen() form a turn: let speak() finish, THEN
+       call listen() to capture the reply. Do NOT run them in parallel here — the
+       mic must open after playback ends or it records your own voice. If your
+       text ends in a question or otherwise invites a response, reach for
+       listen(wait_for_silence=True) right after this returns.
+
+    Audio plays through physical speakers, not into your context. Progress
+    notifications track playback either way.
 
     Progress notifications arrive every ~0.5s during playback:
     - 5%: Entry tone played (audible acknowledgement)
@@ -597,19 +604,20 @@ async def speak(
         return {"error": f"Unknown engine: {engine}. Available: {list(engines.keys())}"}
 
     eng = engines[engine]
-    voice, text = await _resolve_project_voice(
+    voice, project_name = await _resolve_project_voice(
-        ctx, engine, eng, voice, text, voice_cache, project,
+        ctx, engine, eng, voice, voice_cache, project,
     )
 
     priority = Priority.URGENT if urgent else Priority.NORMAL
 
-    chunks = split_text(text)
+    # Announcements always go through Kokoro so the secretary sounds identical
-    if len(chunks) <= 1:
+    # regardless of the project's engine; fall back to the speaking engine if
-        return await _speak_single(
+    # Kokoro somehow isn't loaded.
-            eng, engine, text, voice, queue, entry_tone, priority, ctx, ducker,
+    announce_eng = engines.get("kokoro", eng)
-        )
+
-    return await _speak_chunked(
+    return await _speak(
-        eng, engine, chunks, voice, queue, entry_tone, priority, ctx, ducker,
+        eng, engine, text, voice, project_name, queue, entry_tone, priority, ctx, ducker,
+        announce_eng=announce_eng, secretary_voice=settings.secretary_voice,
     )
 
 
@@ -692,8 +700,10 @@ async def generate_audio(
         return {"error": f"Unknown engine: {engine}. Available: {list(engines.keys())}"}
 
     eng = engines[engine]
-    voice, text = await _resolve_project_voice(
+    # generate_audio bypasses the queue, so it never announces — it only needs
-        ctx, engine, eng, voice, text, voice_cache, project,
+    # the resolved voice, not the project name.
+    voice, _project_name = await _resolve_project_voice(
+        ctx, engine, eng, voice, voice_cache, project,
     )
 
     await ctx.info(f"Generating audio with {engine}...")
@@ -788,7 +798,10 @@ async def transcribe(
 
 @mcp.tool(annotations=ToolAnnotations(readOnlyHint=False, openWorldHint=True))
 async def listen(
-    duration_seconds: float = 5.0,
+    duration_seconds: float = 30.0,
+    wait_for_silence: bool = True,
+    silence_threshold_ms: int = 2200,
+    vad_aggressiveness: int = 3,
     response_format: Literal["json", "text", "verbose_json"] = "json",
     source: str | None = None,
     save_path: str | None = None,
@@ -800,7 +813,14 @@ async def listen(
     min_confidence: float | None = None,
     ctx: Context = CurrentContext(),
 ) -> dict:
-    """Capture audio from the host mic for N seconds, transcribe via Parakeet.
+    """Capture audio from the host mic, transcribe via Parakeet, return the text.
+
+    This is how you HEAR THE PERSON BACK. Pair it with speak() to hold a
+    voice conversation: speak your question (let it finish), then call listen()
+    to capture their spoken answer. The transcribed text comes back to you in
+    the result, so unlike speak() you DO use the return value. For a natural
+    turn, set wait_for_silence=True so recording ends when they stop talking
+    instead of running the full duration.
 
     Records 16 kHz mono WAV via pw-record using the container's PipeWire
     socket bind mount (same socket play_audio uses for output). Default
@@ -808,9 +828,20 @@ async def listen(
     same transcribe_audio() machinery as transcribe(), so all forward-compat
     params (diarize, timestamp_granularities, etc.) work identically.
 
+    A "ready to talk" tone plays before recording opens and a "got it" tone
+    after it closes, so the person knows exactly when to speak.
+
     Args:
-        duration_seconds: How long to listen. pw-record runs for this
+        duration_seconds: Max seconds to listen. With wait_for_silence this is
-            wall-clock duration then receives SIGTERM to close the WAV.
+            an upper bound; otherwise pw-record runs the full duration then
+            receives SIGTERM to close the WAV.
+        wait_for_silence: Stop as soon as the person stops talking (voice-
+            activity detection) instead of recording the whole duration. Best
+            for conversational turns — recommended when capturing a reply.
+        silence_threshold_ms: With wait_for_silence, how long a pause counts as
+            "they're done" (default 2200ms).
+        vad_aggressiveness: webrtcvad mode 0-3 (0 lax, 3 strict). 2 balances
+            rejecting background noise against catching soft speech.
         response_format: 'json' (default), 'text', or 'verbose_json'.
         source: PipeWire source name (e.g. "alsa_input.usb-..." or
             "bluez_input.XX:XX:XX..."). None = system default source.
@@ -828,11 +859,52 @@ async def listen(
     rec_path = Path("/tmp/mcspeak") / f"listen-{ts}.wav"
     rec_path.parent.mkdir(parents=True, exist_ok=True)
 
-    await ctx.info(f"Listening for {duration_seconds:.1f}s (source={source or 'default'})...")
+    state = ctx.lifespan_context
-    try:
+    start_tone = state.get("listen_start_tone")
-        await record_audio(rec_path, duration_seconds, source=source)
+    end_tone = state.get("listen_end_tone")
-    except Exception as e:
+    warmup_ms = settings.listen_warmup_ms
-        return {"error": f"Recording failed: {e}"}
+
+    # The "ready to talk" tone is handed to the recorder rather than played
+    # here: it fires AFTER the mic goes live (post-warmup) so the person's first
+    # word isn't clipped by pw-record's stream startup. See audio.py for the
+    # warmup/no-clip rationale.
+    vad_info: dict | None = None
+    if wait_for_silence:
+        await ctx.info(
+            f"Listening (VAD; up to {duration_seconds:.1f}s, stop after "
+            f"{silence_threshold_ms}ms silence)..."
+        )
+        try:
+            vad_info = await record_audio_until_silence(
+                rec_path,
+                silence_threshold_ms=silence_threshold_ms,
+                max_seconds=duration_seconds,
+                aggressiveness=vad_aggressiveness,
+                source=source,
+                ready_tone=start_tone,
+                warmup_ms=warmup_ms,
+            )
+        except Exception as e:
+            return {"error": f"VAD recording failed: {e}"}
+    else:
+        await ctx.info(f"Listening for {duration_seconds:.1f}s (source={source or 'default'})...")
+        try:
+            await record_audio(
+                rec_path, duration_seconds, source=source,
+                ready_tone=start_tone, warmup_ms=warmup_ms,
+            )
+        except Exception as e:
+            return {"error": f"Recording failed: {e}"}
+
+    # "Capture done" tone AFTER pw-record stops. Audible "over and out"
+    # so the user knows they can stop talking and the system is now
+    # working on transcription.
+    if end_tone:
+        try:
+            await asyncio.sleep(0.05)  # let record buffers flush first
+            await _play_tone(end_tone)
+        except Exception:
+            pass
 
     # Optional persistence to /output/. Reuse _resolve_output_path so the
     # scoping rules + extension correction are consistent with generate_audio.
@@ -871,7 +943,12 @@ async def listen(
     result["recorded"] = str(rec_path)
     if saved_to:
         result["saved_to"] = saved_to
-    result["duration_recorded"] = duration_seconds
+    if vad_info:
+        # VAD path: real recorded duration came from VAD, not the cap.
+        result["duration_recorded"] = vad_info["duration_ms"] / 1000.0
+        result["vad"] = vad_info
+    else:
+        result["duration_recorded"] = duration_seconds
     return result
 
 

src/mcspeak/voice_identity.py

@@ -51,6 +51,10 @@ def _build_identity_pool(voices: list[str]) -> list[str]:
         for v in settings.voice_identity_exclude.split(",")
         if v.strip()
     }
+    # The secretary's voice is reserved — never auto-assign it to a project, so
+    # no project can be confused for the receptionist. (Explicit voice= still works.)
+    if settings.secretary_voice:
+        excludes.add(settings.secretary_voice.strip().lower())
     eligible = [
         v for v in voices
         if v.startswith(prefixes) and v.lower() not in excludes