diff --git a/src/mcspeak/queue.py b/src/mcspeak/queue.py index e6b6b35..aec38dc 100644 --- a/src/mcspeak/queue.py +++ b/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 -pulls work items from a bounded PriorityQueue and plays them one at a time. -Producers call enqueue() which returns immediately with a speech_id. -Use get_status() to check playback outcomes asynchronously. +A dedicated consumer coroutine pulls whole *utterances* from a bounded +PriorityQueue and plays them one at a time. Each utterance streams its +synthesized chunks in through an internal channel, so playback of chunk N +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: - """A prioritized playback request.""" +class _Utterance: + """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) - result: TTSResult = field(compare=False) - future: asyncio.Future = field(compare=False) + sequence: int speech_id: str = field(default="", compare=False) - enqueued_at: float = field(default_factory=time.time, compare=False) - suppress_exit_tone: bool = field(default=False, compare=False) - # Entry tone deferred from speak() handler when the queue was busy at - # call time. Consumer plays this right before the item's audio so the - # tone doesn't overlap a previously-playing item. None = no entry tone - # (either no tone configured, or already played immediately by the handler). + project: str | None = field(default=None, compare=False) + engine: str = field(default="", compare=False) + voice: str = field(default="", compare=False) + # Entry tone deferred from the speak() handler when the queue was busy at + # call time. Consumer plays it right before this utterance so it doesn't + # overlap a previously-playing item. Shared tone WAV — never unlinked. entry_tone: Path | None = field(default=None, compare=False) + # Short "." 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 - drain pending items. + Call start() to launch the consumer, stop() to gracefully drain it. """ 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() - 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) + utt = self._queue.get_nowait() 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 - complete. If it doesn't finish in time, falls back to hard cancel. + Waits up to shutdown_timeout seconds for the currently-playing utterance + 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 - # exit the while loop after finishing the current item. + # Consumer polls _stopped every 0.5s between utterances, so it + # exits the while loop after finishing the current one. try: await asyncio.wait_for( asyncio.shield(self._consumer_task), timeout=self._shutdown_timeout, ) - print( - "Speech queue: graceful shutdown complete", - file=sys.stderr, - ) + print("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(): - item.future.set_result(outcome) - self._record_outcome(item.speech_id, outcome) + if utt.future is not None and not utt.future.done(): + utt.future.set_result(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 - # 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, - }) + await self._play_utterance(utt) 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, - "voice": item.result.voice, + "engine": utt.engine, + "voice": utt.voice, }) - # Don't break — continue to next item + # Fall through to finally, keep looping. else: - # Shutdown — record and exit loop - self._resolve_outcome(item, { + # Shutdown — record and exit loop. + utt.abort() + self._reap_utterance_files(utt) + self._resolve_outcome(utt, { "played": False, "error": "Playback cancelled", - "engine": item.result.engine, - "voice": item.result.voice, + "engine": utt.engine, + "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, - "voice": item.result.voice, + "engine": utt.engine, + "voice": utt.voice, }) finally: self._current = None self._queue.task_done() - # Unduck media when queue is empty — covers ALL exit paths - # (normal completion, PlaybackError, cancel, generic exception) + # Unduck media when queue is empty — covers ALL exit paths. 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: - """Enqueue a TTSResult for playback. Returns immediately with enqueue metadata. + announce_audio: Path | None = None, + ) -> 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 - check the outcome later. + Returns (utterance, meta). On success `utterance` is the _Utterance to + 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 - item's audio (used when speak() handler deferred the tone because the - queue was busy at call time). - - Raises QueueFull if the queue is at max capacity (backpressure). + The utterance goes into the priority queue immediately (before its + chunks exist), which fixes its ordering vs. concurrent producers. The + 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 + 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, - "voice": result.voice, + "engine": engine, + "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, - "voice": result.voice, + "engine": engine, + "voice": voice, } self._futures[speech_id] = future - - return { + self._maybe_fire_call_waiting(utt) + return utt, { "speech_id": speech_id, "queued": True, "position": self._queue.qsize(), - "file": str(result.audio_path), - "duration_seconds": result.duration_seconds, - "engine": result.engine, - "voice": result.voice, + "engine": engine, + "voice": 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 - the cancel tone. If queued, marks the future as cancelled so the - consumer skips it. Returns cancellation status. + If it's currently playing, kills pw-play and the consumer plays the + cancel tone. If queued, marks the future cancelled so the consumer skips + 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, - "voice": self._current.result.voice, - "file": str(self._current.result.audio_path), - "duration_seconds": self._current.result.duration_seconds, + "project": self._current.project, + "engine": self._current.engine, + "voice": self._current.voice, + "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): - if item.speech_id == speech_id: + for pos, utt in enumerate(self._queue._queue, start=1): + if utt.speech_id == speech_id: return { "status": "queued", "speech_id": speech_id, "position": pos, - "engine": item.result.engine, - "voice": item.result.voice, + "project": utt.project, + "engine": utt.engine, + "voice": utt.voice, } except AttributeError: pass # Internal API — degrade gracefully diff --git a/src/mcspeak/server.py b/src/mcspeak/server.py index 91d307e..3342177 100644 --- a/src/mcspeak/server.py +++ b/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") - exit_tone = resolve_tone(settings.exit_tone, tone_paths, "exit_tone", default="roger") + 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="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 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]: - """Auto-assign voice from project hint or MCP roots, optionally prefix text.""" - if voice is not None or not settings.voice_identity: - return voice, text +) -> tuple[str | None, str | None]: + """Resolve (voice, project_name) for this call. - # Explicit project param takes priority, then try MCP roots - project_name = project or await get_project_name(ctx) - if not project_name: - return voice, text + The project name is resolved regardless of whether a voice was passed + explicitly, because it drives the consumer's spoken announcements. A voice + is auto-assigned only when none was given and voice identity is enabled. + 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) - if settings.announce_project and voice is not None: - text = f"{project_name}. {text}" + # Auto-assign a voice only if the caller didn't pin one. + if voice is None and settings.voice_identity and project_name: + 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 - no sentence boundaries. + Short texts are one chunk; long texts split into sentence chunks that + 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 "." 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); - # busy queue defers tone to the consumer so it doesn't overlap whatever's - # already playing. Capture idle state before duck/synth changes it. + chunks = split_text(text) + n = len(chunks) + + # 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}...") - result = await eng.synthesize(text, voice) - await ctx.report_progress(progress=30, total=100, message="Synthesizing...") + # Secretary preamble: a short "." spoken in the reserved + # secretary voice (always via Kokoro, so it sounds identical no matter + # 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...") - enqueue_result = await queue.enqueue( - result, priority=priority, entry_tone=deferred_entry_tone, + # Reserve the queue slot NOW: this fixes ordering vs. concurrent + # producers before any chunk exists. The consumer that pulls it blocks + # 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"): - return enqueue_result + # Stream chunks: synthesize each and hand it straight to the consumer. + 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"] - duration = result.duration_seconds - await ctx.report_progress( - progress=35, total=100, - message="Playing audio \u2014 you can continue working", + await ctx.info("Enqueued for playback") + # NOTE: Do NOT use asyncio.wait_for() - see CLAUDE.md Python 3.13 pitfall. + # _await_with_progress owns 35→100 and distinguishes "queued/waiting" + # from "actually playing" so a queued item doesn't falsely report 99%. + 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. - # Shutdown: queue.stop() in the lifespan finalizer handles graceful drain. - # Client cancel: use cancel_speech(speech_id) for explicit mid-playback stop. - raise - - -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. + # The finally above already closed the utterance, so the consumer drains + # the chunks it has and finishes cleanly. Do NOT cancel the consumer: + # already-synthesized audio should play through. For explicit + # mid-playback stop, use cancel_speech(speech_id). 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(): - t0 = time.time() - while True: - await asyncio.sleep(0.5) - elapsed = time.time() - t0 + async def _report(play_started): + status = queue.get_status(speech_id) + st = status.get("status") + if st == "queued": + 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. - Audio plays through physical speakers, not into your context. The return - value (speech_id, duration) is informational only and never needed for - subsequent reasoning. Call speak alongside your next tool in the same - message and let progress notifications track playback. + Two ways to use this: + + 1. MILESTONE PING (default) — a status update the person overhears ("build + done", "tests green"). CALL THIS IN PARALLEL with your next tool; don't + 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( - ctx, engine, eng, voice, text, voice_cache, project, + voice, project_name = await _resolve_project_voice( + ctx, engine, eng, voice, voice_cache, project, ) priority = Priority.URGENT if urgent else Priority.NORMAL - chunks = split_text(text) - if len(chunks) <= 1: - return await _speak_single( - eng, engine, text, voice, queue, entry_tone, priority, ctx, ducker, - ) - return await _speak_chunked( - eng, engine, chunks, voice, queue, entry_tone, priority, ctx, ducker, + # Announcements always go through Kokoro so the secretary sounds identical + # regardless of the project's engine; fall back to the speaking engine if + # Kokoro somehow isn't loaded. + announce_eng = engines.get("kokoro", eng) + + return await _speak( + 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( - ctx, engine, eng, voice, text, voice_cache, project, + # generate_audio bypasses the queue, so it never announces — it only needs + # 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 - wall-clock duration then receives SIGTERM to close the WAV. + duration_seconds: Max seconds to listen. With wait_for_silence this is + 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'})...") - try: - await record_audio(rec_path, duration_seconds, source=source) - except Exception as e: - return {"error": f"Recording failed: {e}"} + state = ctx.lifespan_context + start_tone = state.get("listen_start_tone") + end_tone = state.get("listen_end_tone") + warmup_ms = settings.listen_warmup_ms + + # 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 diff --git a/src/mcspeak/voice_identity.py b/src/mcspeak/voice_identity.py index f35f8c8..c62fd9b 100644 --- a/src/mcspeak/voice_identity.py +++ b/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