diff --git a/CHANGELOG.md b/CHANGELOG.md index e387e8f..10313c6 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -14,6 +14,36 @@ named rather than smoothed. ## [Unreleased] ### Added +- `pause_voice_input` — the model can pause listening without ending the + call (#100). "Stop listening" or "mute the mic" is a tool call: the session + stays connected, playback keeps playing, background work keeps running and + its results are still announced; only the operator's speech stops reaching + the provider. The flag lives on the capture surface — `DuplexAudio` and + `DiscordAudio` grew `pause_input` / `resume_input` / `input_paused`, the + same one-interface pattern as `playback_pending` in #87 — so both rooms + honour it identically: blocks captured while paused are dropped (never + queued stale), the already-queued ones are discarded, and the Discord bridge + keeps the host's buffers drained and its inactivity timer armed so the bot + is not evicted from the channel. A paused microphone cannot hear the word + "resume", so the way back is the operator's own control — Enter in the + standalone `hermes talk` terminal (toggle; `p`/`r` explicit — a polling + watcher, never a blocking stdin read), `/talk pause` / `/talk resume` in + Discord (`/talk status` says when it is paused) — and the tool is offered + ONLY where that control is guaranteed: the pause decision is made once, + before the tool list is built, from the same predicate that starts the + keyboard watcher, and the registered control is what the receipt names. + No control, no pause: a piped or non-tty stdin gets no key and no tool; + `/talk` at the Hermes prompt shares its tty with prompt_toolkit, so that + lane never watches stdin and offers no pause; and a pause call that + arrives anyway is refused (`no_resume_path`) rather than armed, because a + pause nobody can undo would be a hang-up. On Windows an extended key + (arrows, Insert, F-keys) is consumed whole — before, Down-Arrow's scan + code read as `p` and paused the microphone. Both directions get a spoken + receipt: the model's tool result for its own flips, a contained + announcement for the operator's. The tool classifies read-only (it can + only narrow what a session does, and a pause is never a path to authority) + and refuses — never arms — when no session is attached. Ported idea from + bielcarpi/hermes-live-voice (MIT), idea only. - Run admission control on `delegate_task` (#101). The model may declare `execution_mode` (`exclusive`, the default, or `parallel_read_only`) and up to eight normalized `resource_keys` naming what a task touches — a repo diff --git a/README.md b/README.md index 1cbaf85..25de709 100644 --- a/README.md +++ b/README.md @@ -390,6 +390,29 @@ agent-loop-only tools (`memory`, `session_search`, `honcho_search`, says which of the two it answered from — a recollection can be stale in a way a verbatim line cannot, and nothing is on screen to check it against. +**Pause the microphone without hanging up.** Say "stop listening" (or "mute +the mic") and the model calls `pause_voice_input`: the call stays connected, +playback keeps playing, background work keeps running and its results are +still announced — only your speech stops reaching the provider. A paused +microphone cannot hear the word "resume", so the way back is your own control, +and **the pause is offered only where that control is guaranteed to exist**: + +- **`hermes talk` in a real terminal** — **Enter** toggles (`p` and `r` are + explicit; on Windows they are single keys, elsewhere type the word and + Enter). The connected line says `Enter to pause or resume the microphone` + when the key is live. With a piped or non-tty stdin (Git Bash's mintty + reports no tty to Python; launcher wrappers) there is no key, so no pause is + offered and a pause call is refused with a receipt that says why. +- **`/talk` typed at the Hermes prompt** — the prompt owns that terminal for + the whole call, so the session never watches it for a key, and offers no + pause either. Use `hermes talk` on its own when you want the control. +- **Discord** — `/talk pause` and `/talk resume`, typed. The model-side tool + is offered on the legacy provider-owned lane; on the `provider-host-tools` + lane the host supplies the tool list and the typed commands are the path. + +Both directions get a spoken receipt, the receipt names the control for the +room you are in, and Ctrl+C still hangs up. + **In Discord**, `/talk join` runs the call in the voice channel Hermes is already in — same conversation, same tools, same steering, in a room other people can hear. Talk now reports speaker transitions to the model using the @@ -397,7 +420,9 @@ member's immutable Discord user ID; display names are quoted as untrusted data, and an unknown SSRC stays unresolved and unauthorized. Configure immutable IDs with `TALK_DISCORD_OPERATOR_USER_IDS=[,...]`. Only those speakers may run `delegate_task`, `steer_agent`, `redirect_agent`, or `stop_work`; everyone -may still converse and use read-only tools. Unset, blank, or any malformed list +may still converse and use read-only tools — including `pause_voice_input`, +which can only narrow what the session does; `/talk resume` (text) brings +listening back. Unset, blank, or any malformed list authorizes nobody. Talk binds permission to the exact Discord PCM, VAD input item, and opaque Realtime response metadata — never a display name, SSRC, model argument, or whichever person spoke most recently. Mixed, missing, or diff --git a/__init__.py b/__init__.py index a26b05b..e2bb673 100644 --- a/__init__.py +++ b/__init__.py @@ -165,9 +165,9 @@ def _register_talk_command(ctx) -> None: "description": ( "Start provider-owned voice with canonical Hermes tools (join), or the " "canonical core voice lane (core join); " - "gateway also supports leave and status" + "gateway also supports pause, resume, leave and status" ), - "args_hint": "[join|core join|leave|status]", + "args_hint": "[join|core join|pause|resume|leave|status]", } if contextual: kwargs["invocation_context"] = True @@ -195,14 +195,20 @@ def _talk_command(raw_args: str = "", invocation=None) -> str: try: asyncio.get_running_loop() except RuntimeError: - if sub in {"join", "core join", "leave", "status"}: + if sub in {"join", "core join", "pause", "resume", "leave", "status"}: return ( "Those are for the gateway's Discord voice channel. Here in a " - "terminal, plain `/talk` starts the call." + "terminal, plain `/talk` starts the call; the standalone " + "`hermes talk` command adds Enter to pause and resume the " + "microphone." ) + # This prompt owns the terminal for the whole call (prompt_toolkit, + # raw mode, its own stdin reader), so the session must not watch + # stdin for the pause key — and without that key it offers no pause + # (hermes-talk#100). `hermes talk` on its own is the lane that does. return ( "Voice session ended." - if talk_cli.cli_entry() == 0 + if talk_cli.cli_entry(keyboard_control=False) == 0 else ("Voice session ended with errors — see stderr.") ) @@ -210,6 +216,12 @@ def _talk_command(raw_args: str = "", invocation=None) -> str: return talk_discord.stop_session() if sub == "status": return talk_discord.session_status() + # The room's microphone control (hermes-talk#100): text, because a paused + # session hears nobody and the way back cannot be spoken. + if sub in {"pause", "mute"}: + return talk_discord.pause_session() + if sub in {"resume", "unmute"}: + return talk_discord.resume_session() if sub == "core join": if not talk_core_realtime.core_provider_available(): return "Canonical core voice is unsupported by this Hermes host." diff --git a/dashboard/plugin_api.py b/dashboard/plugin_api.py index c2194c6..e68e679 100644 --- a/dashboard/plugin_api.py +++ b/dashboard/plugin_api.py @@ -237,6 +237,8 @@ def _mint(auth_token: str, voice: str, *, text_output: bool = False): text deltas back through the cascade relay to be spoken server-side. """ + # The browser owns this lane's microphone, so the pause tool is not + # offered here (default_talk_tools' pausable stays False). tools = talk_tools.default_talk_tools() return talk_wire.mint_ephemeral_session( auth_token=auth_token, diff --git a/docs/OPERATING.md b/docs/OPERATING.md index a6a3aa7..9908fe4 100644 --- a/docs/OPERATING.md +++ b/docs/OPERATING.md @@ -340,7 +340,9 @@ SSRC alone, model arguments, and "last speaker" state are never authority. Configured IDs may run the four state-changing tools: `delegate_task`, `steer_agent`, `redirect_agent`, and `stop_work`. Other speakers retain normal conversation and the read-only tools (`search_memory`, `search_vault`, -`check_work`, `list_agents`, `talk_status`, and `talk_capabilities`). Missing +`check_work`, `list_agents`, `talk_status`, `talk_capabilities`, and +`pause_voice_input` — a pause changes nothing outside the session and can only +narrow what it does; `/talk resume`, typed, brings listening back). Missing response correlation, an unresolved speaker, two speakers in one VAD turn, or a speaker outside the allowlist returns a non-sensitive spoken denial without running the handler. diff --git a/docs/VOICE-COMMANDS.md b/docs/VOICE-COMMANDS.md index 28c664d..a9f36fd 100644 --- a/docs/VOICE-COMMANDS.md +++ b/docs/VOICE-COMMANDS.md @@ -22,6 +22,7 @@ delivery; "landed" only ever follows a real delivery artifact. | "once" / "this session" / "no" (answering an approval question) | `resolve_approval` | "approved — just this once", "approved for the rest of the run", or "denied — the agent was told no" | voice can grant `once`, `session`, or `deny` — **never `always`** (narrowed in code, not in the prompt); an unanswered question denies itself on a timer, and interrupting the question denies it on the spot | | "what are you running on?" / "status report" | `talk_status` | version, model, voice, auth lane, agent lane, audio, identity sections | the verification command — field-by-field meaning in [OPERATING.md](OPERATING.md#2-talk_status--the-one-command) | | "what can you do right now?" / "which tools do you have?" | `talk_capabilities` | installed skills, resolved toolsets with their enabled/configured flags, gateway feature flags, live run counts | live evidence, not the prompt — read in-process off the attached agent, or over the api server when detached; a toolset listed `enabled: false` is reported as installed but NOT usable | +| "stop listening" / "mute the mic" / "hold on, I'm talking to someone" | `pause_voice_input` | "microphone paused — press Enter when you want me back" (standalone `hermes talk` in a real terminal) or "… say `/talk resume`" (Discord) | the call stays up: playback, background runs and their announcements continue; nothing you say reaches the provider until YOU resume it — a paused mic cannot hear "resume", so the way back is a key or a command, never speech, and the tool is offered only where that key or command exists (not for `/talk` at the Hermes prompt, not with a non-tty stdin). Resume gets its own spoken receipt | Things you'll hear without asking (v0.6+): a background agent finishing ("Background agent sa-… finished…"), a steering note landing ("the diff --git a/pyproject.toml b/pyproject.toml index 331d4f2..814beb3 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -45,6 +45,7 @@ py-modules = [ "talk_steer", "talk_lifecycle", "talk_operator_auth", + "talk_pause", "talk_progress", "talk_realtime", "talk_openai_realtime", @@ -111,6 +112,7 @@ known-first-party = [ "talk_openai_realtime", "talk_core_realtime", "talk_core_provider", + "talk_pause", "talk_progress", "talk_providers", "talk_realtime", diff --git a/talk_audio.py b/talk_audio.py index a931339..9a69610 100644 --- a/talk_audio.py +++ b/talk_audio.py @@ -269,6 +269,7 @@ def __init__(self) -> None: self._dropped_playback_bytes = 0 self._played_frames = 0 self._output_level = 0.0 + self._input_paused = False self._in_stream = None self._out_stream = None self._pulse_webrtc = _PulseWebRtcAudio() @@ -352,6 +353,13 @@ def _input_callback(self, indata, _frames, _time, _status) -> None: input_level = _pcm16_rms(pcm) with self._lock: output_level = self._output_level + paused = self._input_paused + if paused: + # Paused capture is DROPPED here, not queued and skipped later: a + # block that sits in the queue through a pause would be the first + # thing sent on resume, seconds stale. (One block can still race + # the flag; resume_input drains it.) + return if self._echo_gate_enabled and not self._pulse_webrtc.active: output_active = output_level > self._output_active_level echo_threshold = max( @@ -415,11 +423,56 @@ def _take_playback(self, wanted: int) -> bytes: def read_input_chunk(self) -> bytes | None: """One captured block, or ``None`` when the microphone has nothing yet.""" + if self.input_paused: + return None try: return self._input.get_nowait() except queue.Empty: return None + def _discard_queued_input(self) -> None: + while True: + try: + self._input.get_nowait() + except queue.Empty: + break + + def pause_input(self) -> None: + """Stop feeding captured audio to the session; playback is untouched. + + From the moment this returns nothing the microphone heard reaches + the wire: blocks captured from here on are dropped in the callback, + the ones already queued (captured before the flag, not yet drained + by the sender) are discarded, and the reader answers empty while the + flag is up. Playback, ``playback_pending`` and the barge-in boundary + are unaffected — a paused session still speaks, and still knows what + it has said. + """ + + with self._lock: + self._input_paused = True + self._discard_queued_input() + + def resume_input(self) -> None: + """Feed captured audio to the session again, from the next block on. + + Drains first: the callback's flag check and its queue write are not + one atomic step, so one block admitted just before the pause can + land after the pause's drain. It is the stalest audio there is, and + without this it would be the first thing sent on resume. + """ + + self._discard_queued_input() + with self._lock: + self._input_paused = False + + @property + def input_paused(self) -> bool: + """Whether capture is paused (hermes-talk#100).""" + + with self._lock: + return self._input_paused + def queue_playback(self, pcm: bytes, item_id: str | None = None) -> None: """Queue model audio for the speaker. Drops on overflow, never blocks.""" diff --git a/talk_cli.py b/talk_cli.py index ca1ecda..a3a93ba 100644 --- a/talk_cli.py +++ b/talk_cli.py @@ -27,11 +27,14 @@ import argparse import asyncio import json +import logging import os import re import sys +import threading import time import uuid +from collections.abc import Callable from contextlib import suppress from dataclasses import dataclass @@ -55,6 +58,7 @@ talk_lifecycle, talk_openai_realtime, talk_operator_auth, + talk_pause, talk_progress, talk_realtime, talk_runs, @@ -84,6 +88,7 @@ import talk_lifecycle import talk_openai_realtime import talk_operator_auth + import talk_pause import talk_progress import talk_realtime import talk_runs @@ -94,6 +99,8 @@ import talk_wire from talk_relay import RealtimeRelay +_log = logging.getLogger(__name__) + #: How long the sender waits when the microphone queue is empty. One tenth of #: a block: short enough that capture never falls behind, long enough that an #: idle call is not a spin loop. @@ -1036,6 +1043,174 @@ def approval_outcome_commands(event: dict) -> list[talk_realtime.RealtimeCommand return _announcement_commands(headline, "") +#: Which operator control flipped the microphone, in words the model can +#: repeat. The model's OWN flips (the pause_voice_input tool) are absent on +#: purpose: it speaks those as its tool result, and an announcement on top +#: would be the same receipt twice. +_PAUSE_CONTROLS = { + talk_pause.SOURCE_KEYBOARD: "the keyboard", + talk_pause.SOURCE_COMMAND: "a /talk command", +} + + +def input_pause_commands(paused: bool, source: str) -> list[talk_realtime.RealtimeCommand]: + """Contained speech for an OPERATOR-made microphone flip (hermes-talk#100). + + A control the operator pressed has no voice of its own, so the receipt + rides the same self-deleting, tools-off announcement shape as every + other out-of-band injection. Nothing untrusted is quoted: the headline + is plugin-owned words and a control name from a fixed table. + """ + + control = _PAUSE_CONTROLS.get(source) + if control is None: + return [] + if paused: + headline = ( + f"The operator just paused your microphone from {control}. You will " + "not hear them until they resume it; playback and background work " + "continue." + ) + else: + headline = ( + f"The operator just resumed your microphone from {control} — you can " + "hear them again." + ) + return _announcement_commands(headline, "") + + +#: How often the terminal control checks for a keypress. The reader never +#: blocks on stdin: a thread parked in ``readline()`` would outlive the +#: session and swallow the operator's NEXT line — the one meant for the +#: Hermes prompt `/talk` returns to. +KEYBOARD_POLL_S = 0.1 +_KEY_ACTIONS = { + "": "toggle", + "p": "pause", + "pause": "pause", + "mute": "pause", + "r": "resume", + "resume": "resume", + "unmute": "resume", + "listen": "resume", +} +#: ``msvcrt.getwch()`` returns one of these for an extended key, then the +#: key's scan code on the next call (`Python docs: msvcrt.getch`). +_WIN32_EXTENDED_KEY_PREFIXES = ("\x00", "\xe0") + + +def _read_control_key(stdin, stop: threading.Event) -> str | None: + """One bounded poll of the terminal: an action name, or None for nothing. + + Windows consoles have no ``select`` on stdin, so a waiting keypress is + read through ``msvcrt`` one character at a time — Enter toggles, ``p`` + and ``r`` pause and resume. An extended key (an arrow, Insert, an F-key) + arrives as a ``'\\x00'`` or ``'\\xe0'`` prefix and THEN its scan code on + the next read; the scan code is consumed here with the prefix, because + read on its own it is a letter — Down-Arrow's is ``'P'``, Insert's is + ``'R'`` — and would pause or resume the microphone. Elsewhere ``select`` + waits for a whole line in the terminal's own cooked mode (no tty state + is ever changed, so a crash cannot leave the shell raw): a bare Enter + toggles, and the words in ``_KEY_ACTIONS`` are explicit. EOF stops the + watcher for good. + """ + + if sys.platform == "win32": + import msvcrt + + if not msvcrt.kbhit(): + time.sleep(KEYBOARD_POLL_S) + return None + char = msvcrt.getwch() + if char in _WIN32_EXTENDED_KEY_PREFIXES: + msvcrt.getwch() # the scan code; documented to follow without blocking + return None + if char in ("\r", "\n"): + return "toggle" + if char.isascii() and char.isalpha(): + return _KEY_ACTIONS.get(char.lower()) + return None + import select + + try: + ready, _, _ = select.select([stdin], [], [], KEYBOARD_POLL_S) + except (OSError, ValueError): + stop.set() + return None + if not ready: + return None + line = stdin.readline() + if line == "": + stop.set() + return None + return _KEY_ACTIONS.get(line.strip().lower()) + + +def keyboard_pause_control_available(stdin=None) -> bool: + """Whether :func:`start_keyboard_pause_control` would start on ``stdin``. + + The ONE predicate both the advertisement and the watcher use: a pause + tool is offered on the terminal lane exactly when this returns True, so + the model can never be handed a pause the operator has no key to undo. + False for a piped or missing stdin, a closed file, or anything that is + not a tty (Git Bash's mintty reports ``isatty() == False`` to Python). + """ + + stdin = sys.stdin if stdin is None else stdin + try: + return stdin is not None and bool(stdin.isatty()) + except (AttributeError, ValueError): + return False + + +def start_keyboard_pause_control( + stdin=None, *, read_key=None +) -> Callable[[], None] | None: + """Watch the terminal for the operator's pause control (hermes-talk#100). + + Returns a stop callable, or ``None`` when there is no terminal to watch + — a piped stdin, a test, a gateway. The watcher is a daemon thread that + polls rather than blocks, so ``stop()`` is honoured within one poll and + nothing typed after the session ends is ever consumed here. Each key + goes through :func:`talk_pause.set_paused` exactly like the tool does; + the attached session's receipt callback owns what is said and printed. + + Callers decide WHETHER this terminal may be watched at all: the + standalone ``hermes talk`` command owns its tty, but ``/talk`` typed at + the Hermes prompt runs while prompt_toolkit holds that same tty in raw + mode with its own stdin reader, and a second reader would race it for + every byte (and, on POSIX, park in ``readline()`` waiting for a newline + raw mode never delivers). That lane passes ``keyboard_control=False`` + to :func:`run_talk_session` and never reaches here. + """ + + stdin = sys.stdin if stdin is None else stdin + if not keyboard_pause_control_available(stdin): + return None + stop = threading.Event() + read = read_key or _read_control_key + + def watch() -> None: + while not stop.is_set(): + try: + action = read(stdin, stop) + except Exception as exc: # noqa: BLE001 — a dead console ends the watcher, not the call + _log.debug("keyboard pause control stopped: %s: %s", type(exc).__name__, exc) + return + if action not in _KEY_ACTIONS.values() or stop.is_set(): + continue + # A toggle with nothing attached reads as "pause": the flip is + # refused downstream (NO_SESSION) rather than guessed here. + paused = not bool(talk_pause.is_paused()) if action == "toggle" else action == "pause" + try: + talk_pause.set_paused(paused, source=talk_pause.SOURCE_KEYBOARD) + except Exception as exc: # noqa: BLE001 — one bad key must not end the watcher + _log.debug("keyboard pause flip failed: %s: %s", type(exc).__name__, exc) + + threading.Thread(target=watch, name="talk-keyboard-pause", daemon=True).start() + return stop.set + + def _active_parent_session_id() -> str | None: """Snapshot the bound Hermes session id, or fail closed on older hosts.""" @@ -1270,6 +1445,7 @@ async def run_talk_session( host_execution_attachment=None, lane: str = "cli", on_refusal=None, + keyboard_control: bool = False, ) -> int: """Run one voice session. Returns a process exit code. @@ -1289,6 +1465,16 @@ async def run_talk_session( (the terminal) ignores it, and a lane that owes the operator a spoken receipt (Discord) uses it to say what actually refused instead of collapsing every startup failure into "exited unsuccessfully". + + ``keyboard_control`` says this session OWNS the terminal's stdin and may + watch it for the microphone pause key (hermes-talk#100). Only the + standalone ``hermes talk`` command passes True; ``/talk`` at the Hermes + prompt shares its tty with prompt_toolkit and must not. It is one input + to the pause decision — the key still has to exist (a tty) — and that + decision is made ONCE, before the tools are built: ``pause_voice_input`` + is advertised exactly when an operator resume control is guaranteed + (that key, or ``/talk resume`` in Discord), and the same control is + registered with :mod:`talk_pause`, which refuses to pause without one. """ def refuse(reason: str) -> int: @@ -1328,11 +1514,23 @@ def refuse(reason: str) -> int: host_execution_attachment.close() return refuse(STARTUP_REFUSAL_CONFIGURATION) + # The operator's way back from a pause (hermes-talk#100), decided HERE so + # the tool list below and the registry attach further down agree: the + # Discord room has `/talk resume`; the terminal has Enter only when this + # session owns a real tty. Anything else has no way back, so no pause is + # offered — Ctrl+C is the exit this feature exists to avoid. + if lane == "discord": + resume_control: str | None = talk_pause.RESUME_COMMAND + elif lane == "cli" and keyboard_control and keyboard_pause_control_available(): + resume_control = talk_pause.RESUME_KEYBOARD + else: + resume_control = None + try: tools = ( host_execution_attachment.tool_definitions() if host_execution_attachment is not None - else talk_tools.default_talk_tools() + else talk_tools.default_talk_tools(pausable=resume_control is not None) ) except Exception as exc: # noqa: BLE001 - host attachment startup boundary print(f"talk: host tool setup failed: {type(exc).__name__}", file=sys.stderr) @@ -1416,6 +1614,7 @@ def refuse(reason: str) -> int: watchers: list[asyncio.Task] = [] watched: set[int] = set() spoken_item: str | None = None + keyboard_stop: Callable[[], None] | None = None def on_barge_in() -> None: played = audio.played_ms @@ -1580,12 +1779,23 @@ async def send_outgoing(outgoing, *, is_announcement: bool = False) -> bool: start_watchers(commands) return True + # The operator's own microphone control (hermes-talk#100), terminal + # lane only: a gateway has no keyboard, and its room gets `/talk + # pause` instead. Started only when the pause decision above chose + # the keyboard (same predicate, so it starts iff the tool was + # advertised), and before the connected line so that line can say + # the key exists. + if resume_control == talk_pause.RESUME_KEYBOARD: + keyboard_stop = start_keyboard_pause_control() + controls = "Ctrl+C to hang up" + ( + ", Enter to pause or resume the microphone." if keyboard_stop else "." + ) print( f"talk: connected ({model}, voice {voice}, auth {auth.source}). " - "Ctrl+C to hang up.\n" + f"{controls}\n" if cascade_config is None else f"talk: connected ({model}, cascade voice {cascade_config[1]} " - f"via elevenlabs, auth {auth.source}). Ctrl+C to hang up.\n" + f"via elevenlabs, auth {auth.source}). {controls}\n" ) async def send_microphone() -> None: @@ -1835,6 +2045,25 @@ def on_note_landed(subagent_id: str) -> None: if commands: announce_queue.put_nowait(commands) + def on_pause_change(paused: bool, source: str) -> None: + """Receipt for a microphone flip (hermes-talk#100), from any thread. + + Printed for every flip. SPOKEN only for an operator control: the + model's own tool call already speaks its result, and an + announcement on top would say the same thing twice. + """ + + def deliver() -> None: + state = "paused" if paused else "listening again" + hint = " (Enter to resume)" if paused and keyboard_stop else "" + print(f"\ntalk: microphone {state}{hint}", flush=True) + commands = input_pause_commands(paused, source) + if commands: + announce_queue.put_nowait(commands) + + with suppress(RuntimeError): # loop closed while the flip was in flight + loop.call_soon_threadsafe(deliver) + loop = asyncio.get_running_loop() # Snapshot ownership once for this session. Older Hermes builds do not # expose the property; None suppresses announcements instead of guessing. @@ -1864,6 +2093,12 @@ def on_note_landed(subagent_id: str) -> None: operator=auth.source, profile=talk_profile, ) + # The microphone pause (hermes-talk#100) binds the SAME way, before + # any tool can run: the model's pause_voice_input and the operator's + # key or command all flip this one surface. The registered resume + # control is the one the tool list was built from; with none, the + # registry refuses every pause. + talk_pause.attach_session(audio, on_pause_change, resume_control=resume_control) # Results this session is OWED — accepted under a durable Hermes # session that is still ours, by this SAME operator/profile binding # (a ticket bound to a different binding is never adopted), finished @@ -1956,6 +2191,7 @@ def on_note_landed(subagent_id: str) -> None: # Unbound again: with no live connection there is no destination, # so further dispatch is refused rather than accepted into a void. talk_runs.detach_owner() + talk_pause.detach_session(audio) sender.cancel() pump.cancel() receiver.cancel() @@ -1999,6 +2235,9 @@ def on_note_landed(subagent_id: str) -> None: talk_lifecycle.detach_session() talk_progress.detach_session() talk_approvals.detach_session() + talk_pause.detach_session(audio) + if keyboard_stop is not None: + keyboard_stop() if authorization_ledger is not None: authorization_ledger.clear() if cascade is not None: @@ -2108,7 +2347,9 @@ def setup_cli(subparser: argparse.ArgumentParser) -> None: subparser.set_defaults(talk_command="session") -def cli_entry(args: argparse.Namespace | None = None) -> int: +def cli_entry( + args: argparse.Namespace | None = None, *, keyboard_control: bool | None = None +) -> int: """Synchronous entry point for ``hermes talk``. A failed session raises ``SystemExit`` rather than returning: Hermes's @@ -2116,6 +2357,13 @@ def cli_entry(args: argparse.Namespace | None = None) -> int: (``args.func(args)`` with no exit propagation), so a plain ``return 1`` would exit the process 0 on failure — scripts and CI would read a dead session as success. + + ``keyboard_control`` (hermes-talk#100): whether the session may watch + stdin for the pause key. Unset, it follows how we were called — the + standalone ``hermes talk`` subcommand arrives with argparse ``args`` and + owns the terminal; a bare call is the in-session ``/talk``, whose tty + belongs to the Hermes prompt (prompt_toolkit, raw mode) for the whole + call. ``/talk`` passes False explicitly as well. """ command = getattr(args, "talk_command", "session") if args is not None else "session" @@ -2155,8 +2403,10 @@ def cli_entry(args: argparse.Namespace | None = None) -> int: raise SystemExit(code) return 0 + if keyboard_control is None: + keyboard_control = args is not None try: - code = asyncio.run(run_talk_session()) + code = asyncio.run(run_talk_session(keyboard_control=keyboard_control)) except KeyboardInterrupt: print("\ntalk: hung up.") return 0 @@ -2169,6 +2419,7 @@ def cli_entry(args: argparse.Namespace | None = None) -> int: "ANNOUNCE_STARVATION_WARN_S", "CONNECT_TIMEOUT_S", "IDLE_POLL_S", + "KEYBOARD_POLL_S", "STARTUP_REFUSAL_AUDIO", "STARTUP_REFUSAL_AUTHORIZATION", "STARTUP_REFUSAL_CONFIGURATION", @@ -2183,6 +2434,8 @@ def cli_entry(args: argparse.Namespace | None = None) -> int: "SpeakerPacketLane", "build_session_update", "cli_entry", + "input_pause_commands", + "keyboard_pause_control_available", "landed_note_messages", "pump_announcements", "resolve_provider_lane", @@ -2190,6 +2443,7 @@ def cli_entry(args: argparse.Namespace | None = None) -> int: "run_phase_messages", "run_talk_session", "setup_cli", + "start_keyboard_pause_control", "started_run_ids", "subagent_phase_messages", "subagent_stop_messages", diff --git a/talk_discord.py b/talk_discord.py index 50fe6f1..45db900 100644 --- a/talk_discord.py +++ b/talk_discord.py @@ -1,11 +1,12 @@ -"""Discord voice as an audio device — the same eight methods, a different room. +"""Discord voice as an audio device — the same eleven methods, a different room. :class:`DiscordAudio` implements exactly the surface :class:`talk_audio.DuplexAudio` exposes (``start`` / ``stop`` / -``read_input_chunk`` / ``queue_playback`` / ``drain_playback`` / -``playback_pending`` / ``played_ms`` / ``reset_played_ms``), so the Realtime -session, its tool calls, the steering ledger, and the announcement pump all -run unchanged. +``read_input_chunk`` / ``pause_input`` / ``resume_input`` / ``input_paused`` +/ ``queue_playback`` / ``drain_playback`` / ``playback_pending`` / +``played_ms`` / ``reset_played_ms``), so the Realtime session, its tool +calls, the steering ledger, the announcement pump, and the microphone pause +all run unchanged. Only the room changes: instead of a microphone and a speaker, the frames come from and go to a Discord voice channel. @@ -44,9 +45,10 @@ from typing import Any try: - from . import talk_audio + from . import talk_audio, talk_pause except ImportError: # pragma: no cover - flat-module fallback (Hermes file-path load) import talk_audio + import talk_pause _log = logging.getLogger(__name__) @@ -406,6 +408,7 @@ def __init__(self, guild_id: int | None = None, *, capture_only: bool = False) - self._speaker_notifier = None self._speaker_notifier_generation = 0 self._last_speaker_key: Any = _UNSET + self._input_paused = False # -- lifecycle ------------------------------------------------------------ @@ -816,7 +819,19 @@ def _drain_receiver(self, receiver: Any) -> None: chunks = self._take_receiver_chunks(receiver) if chunks: self._touch_host_timer() + with self._lock: + paused = self._input_paused for ssrc, raw_user_id, chunk in chunks: + if paused: + # Paused (hermes-talk#100): the host's buffers were still + # taken — they must not grow — and its inactivity timer + # still re-armed, but nothing reaches the session. The + # per-speaker carry goes too, so a half-sample from + # before the pause cannot prefix the first frame after. + # (One chunk can still race the flag; resume_input + # drains it.) + self._capture_remainder.pop(ssrc, None) + continue # Per speaker: one speaker's partial sample group prepended # to another's audio would shift it and transpose L/R for # that chunk. @@ -1076,6 +1091,16 @@ def read_input_packet(self) -> InputAudioPacket | None: self._fail_if_bridge_lost() now = time.monotonic() + with self._lock: + if self._input_paused: + # Nothing flows while paused — not real audio, and not the + # synthesized silence either: a server that hears nothing + # detects no turn. The pacing clock keeps up with the wall + # clock so a resume starts at "now" instead of replaying the + # whole pause as a burst of catch-up silence frames. + if self._audio_clock > 0.0: + self._audio_clock = max(self._audio_clock, now) + return None try: packet = self._inbound.get_nowait() except queue.Empty: @@ -1108,6 +1133,49 @@ def read_input_chunk(self) -> bytes | None: packet = self.read_input_packet() return packet.pcm if packet is not None else None + def _discard_queued_input(self) -> None: + while True: + try: + self._inbound.get_nowait() + except queue.Empty: + break + + def pause_input(self) -> None: + """Stop feeding the channel's audio to the session (hermes-talk#100). + + The room keeps hearing the bot: playback, ``playback_pending`` and + the heard boundary are untouched. What stops is capture — the + host's buffers are still drained (and its inactivity timer still + re-armed, so it does not leave the channel), but nothing reaches the + session: frames decoded from here on are dropped, the ones already + queued are discarded, and the reader answers empty while the flag + is up. + """ + + with self._lock: + self._input_paused = True + self._discard_queued_input() + + def resume_input(self) -> None: + """Feed the channel's audio to the session again, from the next frame on. + + Drains first, for the same reason the terminal does: the drain + thread's flag check and its queue write are not one atomic step, so + one chunk admitted just before the pause can land after the pause's + drain — and it must not be the first thing sent on resume. + """ + + self._discard_queued_input() + with self._lock: + self._input_paused = False + + @property + def input_paused(self) -> bool: + """Whether capture is paused (hermes-talk#100).""" + + with self._lock: + return self._input_paused + # -- playback ------------------------------------------------------------- def queue_playback(self, pcm: bytes) -> None: @@ -1195,8 +1263,10 @@ def reset_played_ms(self) -> None: "InputAudioPacket", "TalkDiscordError", "discord_to_session", + "pause_session", "reset_for_tests", "resolve_voice_bridge", + "resume_session", "session_status", "session_to_discord", "start_core_session", @@ -1284,7 +1354,10 @@ def _startup_refusal_failure(reason: str | None) -> str | None: return _STARTUP_REFUSAL_FAILURES.get(reason) -JOIN_USAGE = "Say `talk join` once I'm in a voice channel, or `talk leave` to hand it back." +JOIN_USAGE = ( + "Say `talk join` once I'm in a voice channel, `talk pause` / `talk resume` to " + "mute or unmute my listening, or `talk leave` to hand it back." +) def session_status() -> str: @@ -1307,17 +1380,70 @@ def session_status() -> str: "say `talk leave` to stop." ) return f"Canonical core voice is starting on server {guild_id} — say `talk leave` to stop." + # The microphone pause (hermes-talk#100) is the one live fact that changes + # what "live" means: a paused session plays and announces but hears nobody. + paused = " The microphone is paused — say `talk resume` to be heard again." if ( + talk_pause.is_paused() + ) else "" if mode == "provider-host-tools": return ( f"Provider-owned Realtime voice with canonical Hermes tools is live on server " - f"{guild_id}. Say `talk leave` to stop." + f"{guild_id}. Say `talk leave` to stop.{paused}" ) return ( f"Limited legacy provider-owned voice is live on server {guild_id}. " "Use `/talk core join` for full canonical Hermes parity; say `talk leave` to stop." + f"{paused}" ) +#: The Discord room's own microphone control (hermes-talk#100). Text, on +#: purpose: a paused session hears nobody, so the way back cannot be spoken. +_PAUSE_COMMAND_RECEIPTS = { + talk_pause.PAUSED: ( + "Microphone paused — I'll keep talking and announcing, but I'm not " + "listening. Say `talk resume` when you want me to hear you again." + ), + talk_pause.ALREADY_PAUSED: "The microphone is already paused — `talk resume` unpauses it.", + talk_pause.RESUMED: "Microphone resumed — I'm listening again.", + talk_pause.ALREADY_LISTENING: "I'm already listening — the microphone wasn't paused.", + talk_pause.NO_SESSION: ( + "The voice session isn't listening yet — give it a moment, then try again." + ), + talk_pause.NO_RESUME_PATH: ( + "I didn't pause — this session registered no way to resume, and a pause " + "nobody can undo would be a hang-up. Say `talk leave` to stop instead." + ), + talk_pause.UNSUPPORTED: "This voice session can't pause its input.", +} + + +def _set_session_paused(paused: bool) -> str: + with _SESSION_LOCK: + task = _SESSION.get("task") + mode = _SESSION.get("mode", "legacy") + if task is None or task.done(): + return "I'm not in a voice session right now." + if mode == "core": + # The canonical core lane is input-only and host-owned: the host's + # orchestrator decides what it hears and says, not this plugin. + return "The canonical core lane doesn't pause — say `talk leave` to stop it." + outcome = talk_pause.set_paused(paused, source=talk_pause.SOURCE_COMMAND) + return _PAUSE_COMMAND_RECEIPTS[outcome] + + +def pause_session() -> str: + """Mute the live session's listening without leaving the channel.""" + + return _set_session_paused(True) + + +def resume_session() -> str: + """Unmute a paused session's listening.""" + + return _set_session_paused(False) + + async def _deliver_failure_receipt(adapter: Any, guild_id: int, receipt: str) -> bool: """Put a failed-closed receipt in the voice channel's linked text room.""" diff --git a/talk_operator_auth.py b/talk_operator_auth.py index f2cf3e6..931ef4f 100644 --- a/talk_operator_auth.py +++ b/talk_operator_auth.py @@ -42,6 +42,12 @@ #: (cleared from inbound events first), read only for the denial log line. _PERMIT_REFUSAL_EVENT_KEY = "_talk_permit_refusal" +#: ``pause_voice_input`` is read-only on purpose (hermes-talk#100): it changes +#: nothing outside this session and can only NARROW what the session does — +#: a paused microphone runs no tool at all, mutating or otherwise. Resume by +#: voice is unreachable once paused (nobody is heard), so a speaker who is +#: not the operator can at worst mute listening until the operator's own +#: control brings it back; the pause is never a path to authority. READ_ONLY_TALK_TOOLS = frozenset( { "search_memory", @@ -50,6 +56,7 @@ "list_agents", "talk_status", "talk_capabilities", + "pause_voice_input", } ) MUTATING_TALK_TOOLS = frozenset( diff --git a/talk_pause.py b/talk_pause.py new file mode 100644 index 0000000..869a7ba --- /dev/null +++ b/talk_pause.py @@ -0,0 +1,187 @@ +"""Voice-input pause — mute the microphone without ending the call (hermes-talk#100). + +One live capture surface per process. :func:`talk_cli.run_talk_session` +attaches the audio object it is pumping; the model's ``pause_voice_input`` +tool and the operator's own controls (Enter in the terminal, ``/talk pause`` +and ``/talk resume`` in Discord) flip it through here. Only capture stops: +the speaker keeps playing, every run watcher keeps polling, and +announcements keep landing. + +Same one-at-a-time contract as :func:`talk_runs.attach_owner` — last attach +wins, and while nothing is attached a pause is REFUSED rather than +remembered. A flag armed against a session that has not started yet would +silently mute the next one, and the dashboard tab (whose microphone lives in +the browser) must hear "there is nothing here to pause", not "paused". + +A pause is also refused when the attached session registered NO operator +control that can undo it (:func:`attach_session`'s ``resume_control``). A +paused microphone cannot hear the word "resume", so without a key or a +command the only way back would be Ctrl+C — the one exit this feature +exists to avoid. The session decides that control before it advertises the +tool; this gate is the execution-side half of the same decision, so a tool +call that arrives some other way (a relayed name, a stale schema) cannot +arm a pause nobody can end. + +Thread model: the tool runs on the relay's daemon pool, the terminal key on +its own reader thread, the Discord command on the gateway loop. Every entry +point takes the lock; the attaching session's ``on_change`` callback fires +OUTSIDE it and is the session's own business to marshal onto its loop. +""" + +from __future__ import annotations + +import logging +import threading +from collections.abc import Callable + +_log = logging.getLogger(__name__) + +#: Who flipped the flag. The session's receipt wording depends on it: a +#: pause the MODEL made is already spoken by the model as its tool result, +#: while an operator control has no voice of its own and is announced. +SOURCE_TOOL = "tool" +SOURCE_KEYBOARD = "keyboard" +SOURCE_COMMAND = "command" +SOURCES = frozenset({SOURCE_TOOL, SOURCE_KEYBOARD, SOURCE_COMMAND}) + +#: Outcomes of :func:`set_paused`. Callers compose the sentence; the state +#: change itself is decided here, once, under the lock. +PAUSED = "paused" +RESUMED = "resumed" +ALREADY_PAUSED = "already_paused" +ALREADY_LISTENING = "already_listening" +NO_SESSION = "no_session" +NO_RESUME_PATH = "no_resume_path" +UNSUPPORTED = "unsupported" + +#: The operator's way back from a pause, in the words the receipts use. A +#: session registers exactly one of these at attach time — or none, in which +#: case pausing is refused (see the module docstring). +RESUME_KEYBOARD = "Enter in the terminal" +RESUME_COMMAND = "/talk resume in Discord" + +_LOCK = threading.Lock() +_SURFACE: object | None = None +_ON_CHANGE: Callable[[bool, str], None] | None = None +_RESUME_CONTROL: str | None = None + + +def attach_session( + audio: object, + on_change: Callable[[bool, str], None] | None = None, + *, + resume_control: str | None = None, +) -> None: + """Bind the live session's capture surface (and its receipt callback). + + ``on_change(paused, source)`` is called after every ACTUAL flip — never + for a no-op — from whichever thread made it. ``resume_control`` names + the operator's own way back (:data:`RESUME_KEYBOARD`, + :data:`RESUME_COMMAND`); ``None`` means there is none, and every pause + is then refused with :data:`NO_RESUME_PATH`. + """ + + global _SURFACE, _ON_CHANGE, _RESUME_CONTROL + with _LOCK: + _SURFACE = audio + _ON_CHANGE = on_change + _RESUME_CONTROL = resume_control + + +def detach_session(audio: object | None = None) -> None: + """Drop the attached surface. + + With ``audio`` given, only if it is STILL the attached one: a session's + teardown must not undo the attach of the session that replaced it. + """ + + global _SURFACE, _ON_CHANGE, _RESUME_CONTROL + with _LOCK: + if audio is not None and _SURFACE is not audio: + return + _SURFACE = None + _ON_CHANGE = None + _RESUME_CONTROL = None + + +def resume_control() -> str | None: + """The attached session's operator resume control; ``None`` when there is none.""" + + with _LOCK: + return _RESUME_CONTROL if _SURFACE is not None else None + + +def is_paused() -> bool | None: + """Whether the attached surface is paused; ``None`` when nothing is attached.""" + + with _LOCK: + surface = _SURFACE + if surface is None: + return None + paused = getattr(surface, "input_paused", None) + return paused if isinstance(paused, bool) else None + + +def set_paused(paused: bool, *, source: str) -> str: + """Pause or resume capture on the attached surface. Returns an outcome. + + The read-modify-write is one critical section, so two controls racing + (the model and a keypress, say) resolve to one flip and one no-op instead + of two receipts for the same state. The surface's own ``pause_input`` / + ``resume_input`` are called inside it; they are queue flips, not device + calls, and hold nothing that calls back into this module. + """ + + if source not in SOURCES: + raise ValueError(f"unknown pause source: {source!r}") + with _LOCK: + surface, on_change, way_back = _SURFACE, _ON_CHANGE, _RESUME_CONTROL + if surface is None: + return NO_SESSION + pause = getattr(surface, "pause_input", None) + resume = getattr(surface, "resume_input", None) + current = getattr(surface, "input_paused", None) + if not callable(pause) or not callable(resume) or not isinstance(current, bool): + return UNSUPPORTED + if paused and current: + return ALREADY_PAUSED + if not paused and not current: + return ALREADY_LISTENING + if paused and way_back is None: + # Resuming is always allowed — it can only widen listening back + # to normal. Pausing needs a way back first. + return NO_RESUME_PATH + (pause if paused else resume)() + if on_change is not None: + try: + on_change(paused, source) + except Exception as exc: # noqa: BLE001 — a receipt must never undo the flip + _log.debug("pause change callback failed: %s: %s", type(exc).__name__, exc) + return PAUSED if paused else RESUMED + + +def reset_for_tests() -> None: + detach_session() + + +__all__ = [ + "ALREADY_LISTENING", + "ALREADY_PAUSED", + "NO_RESUME_PATH", + "NO_SESSION", + "PAUSED", + "RESUMED", + "RESUME_COMMAND", + "RESUME_KEYBOARD", + "SOURCES", + "SOURCE_COMMAND", + "SOURCE_KEYBOARD", + "SOURCE_TOOL", + "UNSUPPORTED", + "attach_session", + "detach_session", + "is_paused", + "reset_for_tests", + "resume_control", + "set_paused", +] diff --git a/talk_tools.py b/talk_tools.py index aa64b9e..1ae6237 100644 --- a/talk_tools.py +++ b/talk_tools.py @@ -37,6 +37,7 @@ talk_doctor, talk_host, talk_identity, + talk_pause, talk_runs, talk_steer, talk_vault, @@ -51,6 +52,7 @@ import talk_doctor import talk_host import talk_identity + import talk_pause import talk_runs import talk_steer import talk_vault @@ -387,6 +389,32 @@ } +_TOOL_PAUSE_VOICE_INPUT: dict = { + "type": "function", + "name": "pause_voice_input", + "description": ( + "Pause listening — mute your microphone WITHOUT ending the call. Use " + "when the operator says to stop listening, mute the mic, or hold on " + "while they talk to someone else. Playback, background work and its " + "announcements all continue; only their speech stops reaching you. " + "Once paused you cannot hear a spoken resume: the operator resumes " + "from their own control, which the tool result names — repeat it as " + "you confirm the pause. Pass paused=false to resume when a non-spoken " + "path asks you to." + ), + "parameters": { + "type": "object", + "properties": { + "paused": { + "type": "boolean", + "description": "true (default) pauses the microphone; false resumes it.", + }, + }, + "additionalProperties": False, + }, +} + + class TalkToolError(Exception): """Unknown tool name or otherwise malformed tool call.""" @@ -410,13 +438,20 @@ def plugin_version() -> str: return "unknown" -def default_talk_tools() -> list[dict]: +def default_talk_tools(*, pausable: bool = False) -> list[dict]: """The tool set advertised to a new Talk session (fresh copies per call). The base set is unconditional. ``search_vault`` is CONDITIONAL: it is advertised only when a memory provider is actually loadable in this process, because advertising a lookup that cannot be served is the same defect as the provider block this plugin stopped passing through. + ``pause_voice_input`` is conditional the same way, on ``pausable``: the + session passes True only when this process pumps the microphone AND the + operator has a guaranteed way to resume it (a keyboard the session owns, + or ``/talk resume``). Default False — the dashboard tab's microphone + lives in the browser, and a terminal whose stdin is not a tty has no key + to press — so a pause tool is never offered where the only way back + would be Ctrl+C. """ tools = [ @@ -431,6 +466,8 @@ def default_talk_tools() -> list[dict]: _TOOL_TALK_STATUS, _TOOL_TALK_CAPABILITIES, ] + if pausable: + tools.append(_TOOL_PAUSE_VOICE_INPUT) try: if talk_vault.available(): tools.insert(1, _TOOL_SEARCH_VAULT) @@ -625,6 +662,51 @@ def _handle_resolve_approval(arguments: dict) -> str: return talk_approvals.resolve(run_id, arguments.get("choice")) +#: What the model reads back after a pause flip. Spoken, so each one says +#: what is TRUE now and, for a pause, how the operator gets back — a paused +#: microphone cannot carry the word "resume". The PAUSED receipt names the +#: control THIS session registered (``{resume}``), never a key or a command +#: from another room. +PAUSE_RECEIPTS: dict[str, str] = { + talk_pause.PAUSED: ( + "Microphone paused — you are no longer hearing the operator. Playback, " + "background work and its announcements continue. Tell them how to " + "resume: {resume}." + ), + talk_pause.ALREADY_PAUSED: "The microphone was already paused.", + talk_pause.RESUMED: "Microphone resumed — you are hearing the operator again.", + talk_pause.ALREADY_LISTENING: "The microphone was not paused; you are already listening.", + talk_pause.NO_SESSION: ( + "There is no live voice session attached to this process, so there is " + "no microphone here to pause — in the dashboard tab the browser owns " + "the microphone, so use its own mute control." + ), + talk_pause.NO_RESUME_PATH: ( + "The microphone was not paused: this session has no control the " + "operator could resume it with, and a pause nobody can undo would end " + "the call in all but name. They can still hang up with Ctrl+C." + ), + talk_pause.UNSUPPORTED: "This session's audio device cannot pause its input.", +} + +_FALSE_WORDS = frozenset({"false", "no", "0", "off", "resume"}) + + +def _handle_pause_voice_input(arguments: dict) -> str: + raw = arguments.get("paused") + if isinstance(raw, str): + paused = raw.strip().lower() not in _FALSE_WORDS + else: + paused = True if raw is None else bool(raw) + outcome = talk_pause.set_paused(paused, source=talk_pause.SOURCE_TOOL) + receipt = PAUSE_RECEIPTS[outcome] + if outcome == talk_pause.PAUSED: + # The gate above guarantees a control was registered; the fallback + # only covers a detach racing this read. + receipt = receipt.format(resume=talk_pause.resume_control() or "their own control") + return receipt + + def _identity_summary() -> dict[str, int]: """Resolved identity sections as ``{NAME: char_count}``. Never content. @@ -798,12 +880,14 @@ def _handle_talk_capabilities(arguments: dict) -> str: "resolve_approval": _handle_resolve_approval, "talk_status": _handle_talk_status, "talk_capabilities": _handle_talk_capabilities, + "pause_voice_input": _handle_pause_voice_input, } __all__ = [ "MAX_CATALOG_ENTRIES", "MAX_OUTPUT_CHARS", + "PAUSE_RECEIPTS", "REGISTRATION_FAILURES", "REGISTRATION_RECEIPTS", "REGISTRATION_REQUIREMENTS", diff --git a/tests/test_cli.py b/tests/test_cli.py index b5cf26d..332fdff 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -2161,7 +2161,7 @@ def finish(self): def test_keyboard_interrupt_hangs_up_cleanly(monkeypatch, capsys): - async def interrupted(): + async def interrupted(**_kwargs): raise KeyboardInterrupt monkeypatch.setattr(talk_cli, "run_talk_session", interrupted) @@ -2178,7 +2178,7 @@ def test_setup_cli_adds_no_required_arguments(): def test_cli_entry_raises_systemexit_on_failure(monkeypatch): - async def failing(): + async def failing(**_kwargs): return 1 monkeypatch.setattr(talk_cli, "run_talk_session", failing) diff --git a/tests/test_discord.py b/tests/test_discord.py index d4d09e1..81aaf50 100644 --- a/tests/test_discord.py +++ b/tests/test_discord.py @@ -46,14 +46,19 @@ def _tone(samples: int, *, rate: int, freq: float = 440.0, stereo: bool = False) def test_bridge_wears_the_audio_device_surface(): - # The session calls exactly these eight. If DuplexAudio grows a ninth, + # The session calls exactly these eleven. If DuplexAudio grows a twelfth, # this test is what tells us the Discord room needs it too. # playback_pending joined in hermes-talk#50: the announcement gate needs - # to ask whether the room is still speaking WITHOUT draining it. + # to ask whether the room is still speaking WITHOUT draining it. The + # pause trio joined in hermes-talk#100: the model's tool and the + # operator's controls mute listening without leaving the room. surface = ( "start", "stop", "read_input_chunk", + "pause_input", + "resume_input", + "input_paused", "queue_playback", "drain_playback", "playback_pending", diff --git a/tests/test_pause.py b/tests/test_pause.py new file mode 100644 index 0000000..3b83a64 --- /dev/null +++ b/tests/test_pause.py @@ -0,0 +1,894 @@ +"""The microphone pause (hermes-talk#100). + +What is being proved: a paused capture surface feeds the session nothing +while playback, the barge-in boundary and the announcement gate carry on; +both rooms (terminal microphone, Discord channel) honour the same flag; +the model's tool, the operator's key and the ``/talk`` command all flip +the one attached surface and get the receipt they were promised; a +surface that is not attached refuses instead of arming a pause against +the next session; and a pause is never offered — nor armed — where the +operator has no key or command to undo it, so Ctrl+C is never the only +way back. +""" + +from __future__ import annotations + +import argparse +import asyncio +import json +import sys +import threading +import time +import types + +import pytest +from test_discord import _tone, _wired_host + +import talk_audio +import talk_cli +import talk_discord +import talk_operator_auth +import talk_pause +import talk_realtime +import talk_tools + + +@pytest.fixture(autouse=True) +def _clean(): + talk_pause.reset_for_tests() + talk_discord.reset_for_tests() + yield + talk_pause.reset_for_tests() + talk_discord.reset_for_tests() + + +def _block() -> bytes: + return b"\x10\x00" * talk_audio.BLOCKSIZE + + +# -- the terminal microphone -------------------------------------------------- + + +def test_pause_stops_terminal_capture_and_resume_restores_it(): + audio = talk_audio.DuplexAudio() + assert audio.input_paused is False + + audio._input_callback(_block(), talk_audio.BLOCKSIZE, None, None) + assert audio.read_input_chunk() is not None + + audio.pause_input() + assert audio.input_paused is True + audio._input_callback(_block(), talk_audio.BLOCKSIZE, None, None) + # Dropped in the CALLBACK, not queued and skipped by the reader: a queue + # that fills through a long pause overflows into dropped_input_blocks, + # which talk_status reports as capacity trouble. + assert audio._input.qsize() == 0, "paused capture was queued, not dropped" + assert audio.read_input_chunk() is None, "paused capture reached the reader" + # Dropped-while-paused is not an overflow: the counter reports capacity + # trouble, and a pause is the operator's choice. + assert audio.dropped_input_blocks == 0 + + audio.resume_input() + assert audio.input_paused is False + audio._input_callback(_block(), talk_audio.BLOCKSIZE, None, None) + assert audio.read_input_chunk() is not None + + +def test_pause_discards_what_was_queued_before_the_flag(): + """"Paused" means nothing the microphone heard reaches the wire — + including the blocks the sender had not drained yet.""" + + audio = talk_audio.DuplexAudio() + for _ in range(3): + audio._input_callback(_block(), talk_audio.BLOCKSIZE, None, None) + + audio.pause_input() + audio.resume_input() + + assert audio.read_input_chunk() is None + + +def test_resume_drains_the_block_that_raced_the_flag(): + """The callback's flag check and its queue write are not one step: a + block admitted just before the pause can land after the pause's drain. + It is the stalest audio there is and must not lead the resume.""" + + audio = talk_audio.DuplexAudio() + audio.pause_input() + audio._input.put_nowait(_block()) # landed after the drain, before the flag was seen + + assert audio.read_input_chunk() is None, "a paused reader must answer empty" + audio.resume_input() + assert audio.read_input_chunk() is None, "the raced block led the resume" + + +def test_pause_leaves_playback_and_the_announcement_gate_alone(): + """Only capture stops. The speaker, the drain signal the announcement + pump gates on (hermes-talk#50), and the heard boundary are untouched — + a paused session still speaks results and still knows what it said.""" + + audio = talk_audio.DuplexAudio() + audio.pause_input() + + audio.queue_playback(b"\x01\x02" * 480, "item-1") + assert audio.playback_pending is True + with audio._lock: + audio._take_playback(talk_audio.FRAME_BYTES * 240) + assert audio.played_ms == 10 + assert audio.drain_playback() == ("item-1", 10) + assert audio.playback_pending is False + assert audio.input_paused is True, "playback activity must not unpause" + + +# -- the Discord channel ------------------------------------------------------ + + +def test_pause_drops_channel_audio_but_keeps_the_host_drained_and_armed(monkeypatch): + """The host's buffers are still taken (they must not grow) and its + inactivity timer still re-armed (it must not leave the channel); what + stops is the audio reaching the session.""" + + connection, receiver, _vc, adapter = _wired_host(monkeypatch) + bridge = talk_discord.DiscordAudio(7) + inline = types.SimpleNamespace(call_soon_threadsafe=lambda fn, *a: fn(*a)) + bridge._loop = inline + bridge.start() + bridge._loop = inline + frame = _tone(1920, rate=48_000, stereo=True) + + connection.deliver(frame) + assert bridge.read_input_chunk() not in (None, talk_discord.SESSION_SILENCE) + + bridge.pause_input() + bridge._last_keepalive = 0.0 # past the keepalive throttle + connection.deliver(frame) + assert not receiver._buffers[1], "the host's buffer grew while paused" + assert adapter.__dict__.get("timer_resets") == [7, 7] + # Dropped in the drain loop, not queued for the reader to skip. + assert bridge._inbound.qsize() == 0, "paused channel audio was queued, not dropped" + assert bridge.read_input_chunk() is None + assert bridge.read_input_packet() is None, "silence must not be synthesized while paused" + + bridge.queue_playback(_tone(2400, rate=24_000)) + assert bridge.playback_pending is True, "a paused room must still play" + + bridge.resume_input() + connection.deliver(frame) + assert bridge.read_input_chunk() not in (None, talk_discord.SESSION_SILENCE) + bridge.stop() + + +def test_resume_does_not_replay_the_pause_as_a_burst_of_silence(monkeypatch): + """The pacing clock keeps up with the wall clock while paused. Otherwise + a resume would owe the server the whole pause in catch-up silence frames, + sent as fast as the pump can call — a minute of pause, three thousand + frames at once.""" + + _connection, _receiver, _vc, _adapter = _wired_host(monkeypatch) + bridge = talk_discord.DiscordAudio(7) + bridge.start() + bridge.pause_input() + bridge._audio_clock = time.monotonic() - 1.0 # a second of pause has passed + + assert bridge.read_input_packet() is None + bridge.resume_input() + + burst = 0 + while bridge.read_input_packet() is not None and burst < 100: + burst += 1 + assert burst <= 5, f"resume replayed {burst} catch-up frames" # ~50 without the fix + bridge.stop() + + +def test_a_bridge_pause_discards_the_frame_that_raced_the_flag(monkeypatch): + _connection, _receiver, _vc, _adapter = _wired_host(monkeypatch) + bridge = talk_discord.DiscordAudio(7) + bridge.start() + bridge.pause_input() + bridge._inbound.put_nowait( + talk_discord.InputAudioPacket(speaker=None, pcm=b"\x01\x02" * 480) + ) + + bridge.resume_input() + packet = bridge.read_input_packet() + assert packet is None or packet.pcm == talk_discord.SESSION_SILENCE + bridge.stop() + + +# -- the registry ------------------------------------------------------------- + + +class _Surface: + def __init__(self): + self.input_paused = False + self.calls: list[str] = [] + + def pause_input(self): + self.calls.append("pause") + self.input_paused = True + + def resume_input(self): + self.calls.append("resume") + self.input_paused = False + + +def test_nothing_attached_refuses_instead_of_arming_a_pause(): + assert talk_pause.is_paused() is None + assert talk_pause.resume_control() is None + assert talk_pause.set_paused(True, source=talk_pause.SOURCE_TOOL) == talk_pause.NO_SESSION + assert talk_pause.set_paused(False, source=talk_pause.SOURCE_TOOL) == talk_pause.NO_SESSION + + surface = _Surface() + talk_pause.attach_session(surface, resume_control=talk_pause.RESUME_KEYBOARD) + assert surface.input_paused is False, "a refused pause must not carry into the next attach" + + +def test_a_pause_needs_a_registered_way_back(): + """A paused microphone cannot hear "resume". A session that registered + no operator control is refused every pause — from any source — because + the only way back would be Ctrl+C, the exit this feature exists to + avoid. Resuming is always allowed: it only widens listening back.""" + + surface = _Surface() + changes: list = [] + talk_pause.attach_session(surface, lambda p, s: changes.append((p, s))) + assert talk_pause.resume_control() is None + + for source in talk_pause.SOURCES: + assert talk_pause.set_paused(True, source=source) == talk_pause.NO_RESUME_PATH + assert surface.input_paused is False and surface.calls == [] and changes == [] + + surface.input_paused = True # paused some other way; the way back is open + assert talk_pause.set_paused(False, source=talk_pause.SOURCE_TOOL) == talk_pause.RESUMED + assert surface.calls == ["resume"] + + talk_pause.attach_session(surface, resume_control=talk_pause.RESUME_COMMAND) + assert talk_pause.resume_control() == talk_pause.RESUME_COMMAND + assert talk_pause.set_paused(True, source=talk_pause.SOURCE_TOOL) == talk_pause.PAUSED + talk_pause.detach_session(surface) + assert talk_pause.resume_control() is None + + +def test_set_paused_flips_once_and_reports_no_ops(): + surface = _Surface() + changes: list[tuple[bool, str]] = [] + talk_pause.attach_session( + surface, lambda p, s: changes.append((p, s)), resume_control=talk_pause.RESUME_KEYBOARD + ) + + tool, key, command = ( + talk_pause.SOURCE_TOOL, + talk_pause.SOURCE_KEYBOARD, + talk_pause.SOURCE_COMMAND, + ) + assert talk_pause.set_paused(True, source=tool) == talk_pause.PAUSED + assert talk_pause.is_paused() is True + assert talk_pause.set_paused(True, source=key) == talk_pause.ALREADY_PAUSED + assert talk_pause.set_paused(False, source=command) == talk_pause.RESUMED + assert talk_pause.set_paused(False, source=tool) == talk_pause.ALREADY_LISTENING + + assert surface.calls == ["pause", "resume"] + # The receipt callback fires for the two ACTUAL flips only, tagged with + # who made them — never for a no-op. + assert changes == [(True, talk_pause.SOURCE_TOOL), (False, talk_pause.SOURCE_COMMAND)] + + +def test_a_raising_receipt_callback_never_undoes_the_flip(): + surface = _Surface() + + def boom(_paused, _source): + raise RuntimeError("loop is gone") + + talk_pause.attach_session(surface, boom, resume_control=talk_pause.RESUME_KEYBOARD) + assert talk_pause.set_paused(True, source=talk_pause.SOURCE_TOOL) == talk_pause.PAUSED + assert surface.input_paused is True + + +def test_a_surface_without_the_flag_is_unsupported_not_crashed(): + talk_pause.attach_session(object()) + assert talk_pause.is_paused() is None + assert talk_pause.set_paused(True, source=talk_pause.SOURCE_TOOL) == talk_pause.UNSUPPORTED + + +def test_an_unknown_source_is_a_caller_bug(): + talk_pause.attach_session(_Surface()) + with pytest.raises(ValueError, match="pause source"): + talk_pause.set_paused(True, source="webhook") + + +def test_detach_only_drops_the_surface_it_names(): + first, second = _Surface(), _Surface() + talk_pause.attach_session(first, resume_control=talk_pause.RESUME_KEYBOARD) + # A later session took the slot, with its own way back. + talk_pause.attach_session(second, resume_control=talk_pause.RESUME_COMMAND) + talk_pause.detach_session(first) # the earlier session tears down + assert talk_pause.set_paused(True, source=talk_pause.SOURCE_TOOL) == talk_pause.PAUSED + assert second.input_paused is True + + talk_pause.detach_session(second) + assert talk_pause.is_paused() is None + + +def test_concurrent_controls_resolve_to_one_flip(): + """The model and a keypress racing for the same state produce one flip + and one no-op, never two receipts for the same state.""" + + surface = _Surface() + changes: list = [] + talk_pause.attach_session( + surface, lambda p, s: changes.append(s), resume_control=talk_pause.RESUME_KEYBOARD + ) + outcomes: list[str] = [] + go = threading.Barrier(2) + + def press(source): + go.wait() + outcomes.append(talk_pause.set_paused(True, source=source)) + + threads = [ + threading.Thread(target=press, args=(talk_pause.SOURCE_TOOL,)), + threading.Thread(target=press, args=(talk_pause.SOURCE_KEYBOARD,)), + ] + for thread in threads: + thread.start() + for thread in threads: + thread.join() + + assert sorted(outcomes) == [talk_pause.ALREADY_PAUSED, talk_pause.PAUSED] + assert len(changes) == 1 + assert surface.calls == ["pause"] + + +# -- the tool ----------------------------------------------------------------- + + +def test_the_tool_is_advertised_handled_and_classified_read_only(): + tools = talk_tools.default_talk_tools(pausable=True) + schema = next(tool for tool in tools if tool["name"] == "pause_voice_input") + assert schema["parameters"]["properties"]["paused"]["type"] == "boolean" + assert "pause_voice_input" in talk_tools._HANDLERS + # Read-only on purpose: it changes nothing outside the session and can + # only NARROW what the session does. In a Discord room every speaker may + # mute listening; nobody gains authority by it, and resume is the + # operator's own control. + assert "pause_voice_input" in talk_operator_auth.READ_ONLY_TALK_TOOLS + assert "pause_voice_input" not in talk_operator_auth.MUTATING_TALK_TOOLS + + +def test_the_tool_is_offered_only_with_a_guaranteed_way_back(): + """``pausable`` is the session's word that an operator control exists. + Default off: the dashboard tab's microphone lives in the browser, and a + terminal without a tty has no key — a pause tool there would be a pause + only Ctrl+C could end.""" + + def names(**kwargs) -> list[str]: + return [tool["name"] for tool in talk_tools.default_talk_tools(**kwargs)] + + assert "pause_voice_input" in names(pausable=True) + assert "pause_voice_input" not in names(pausable=False) + assert "pause_voice_input" not in names() + assert names() == names(pausable=True)[:-1], "only the pause tool may differ" + + +def test_the_tool_refuses_when_no_microphone_is_attached(): + receipt = talk_tools.execute_talk_tool("pause_voice_input", {}) + assert receipt == talk_tools.PAUSE_RECEIPTS[talk_pause.NO_SESSION] + assert "browser owns the microphone" in receipt + + +def test_the_tool_pauses_says_how_to_resume_and_resumes(): + audio = talk_audio.DuplexAudio() + talk_pause.attach_session(audio, resume_control=talk_pause.RESUME_KEYBOARD) + + paused = talk_tools.execute_talk_tool("pause_voice_input", {}) + assert paused == talk_tools.PAUSE_RECEIPTS[talk_pause.PAUSED].format( + resume=talk_pause.RESUME_KEYBOARD + ) + # The receipt names THIS room's control, never the other room's. + assert paused.endswith("Tell them how to resume: Enter in the terminal.") + assert "/talk resume" not in paused + assert audio.input_paused is True + + again = talk_tools.execute_talk_tool("pause_voice_input", {"paused": True}) + assert again == talk_tools.PAUSE_RECEIPTS[talk_pause.ALREADY_PAUSED] + + resumed = talk_tools.execute_talk_tool("pause_voice_input", {"paused": False}) + assert resumed == talk_tools.PAUSE_RECEIPTS[talk_pause.RESUMED] + assert audio.input_paused is False + + listening = talk_tools.execute_talk_tool("pause_voice_input", {"paused": False}) + assert listening == talk_tools.PAUSE_RECEIPTS[talk_pause.ALREADY_LISTENING] + + +def test_the_tool_names_the_discord_control_in_a_discord_room(): + audio = talk_audio.DuplexAudio() + talk_pause.attach_session(audio, resume_control=talk_pause.RESUME_COMMAND) + + paused = talk_tools.execute_talk_tool("pause_voice_input", {}) + assert paused.endswith("Tell them how to resume: /talk resume in Discord.") + assert "Enter" not in paused + + +def test_the_tool_refuses_to_pause_a_session_with_no_way_back(): + """The execution-side half of the advertisement gate: a pause call that + arrives anyway — a relayed tool name, a stale schema — cannot arm a pause + nobody can undo.""" + + audio = talk_audio.DuplexAudio() + talk_pause.attach_session(audio) + + refused = talk_tools.execute_talk_tool("pause_voice_input", {}) + assert refused == talk_tools.PAUSE_RECEIPTS[talk_pause.NO_RESUME_PATH] + assert "was not paused" in refused and "Ctrl+C" in refused + assert audio.input_paused is False + + +@pytest.mark.parametrize("raw", ["false", "No", "0", "off", "resume"]) +def test_a_provider_that_serializes_the_flag_as_text_still_resumes(raw): + audio = talk_audio.DuplexAudio() + audio.pause_input() + talk_pause.attach_session(audio, resume_control=talk_pause.RESUME_KEYBOARD) + + assert talk_tools.execute_talk_tool("pause_voice_input", {"paused": raw}) == ( + talk_tools.PAUSE_RECEIPTS[talk_pause.RESUMED] + ) + assert audio.input_paused is False + + +def test_every_outcome_has_a_receipt(): + outcomes = { + talk_pause.PAUSED, + talk_pause.RESUMED, + talk_pause.ALREADY_PAUSED, + talk_pause.ALREADY_LISTENING, + talk_pause.NO_SESSION, + talk_pause.NO_RESUME_PATH, + talk_pause.UNSUPPORTED, + } + assert set(talk_tools.PAUSE_RECEIPTS) == outcomes + assert set(talk_discord._PAUSE_COMMAND_RECEIPTS) == outcomes + + +# -- the operator's controls -------------------------------------------------- + + +def test_operator_flips_are_announced_in_the_contained_shape_and_tool_flips_are_not(): + for source, control in ( + (talk_pause.SOURCE_KEYBOARD, "the keyboard"), + (talk_pause.SOURCE_COMMAND, "a /talk command"), + ): + commands = talk_cli.input_pause_commands(True, source) + assert [type(c) for c in commands] == [ + talk_realtime.AddContext, + talk_realtime.StartResponse, + talk_realtime.RemoveContext, + ] + assert f"paused your microphone from {control}" in commands[0].text + assert commands[1].allow_tools is False + assert commands[2].item_id == commands[0].item_id + resumed = talk_cli.input_pause_commands(False, source) + assert f"resumed your microphone from {control}" in resumed[0].text + # The model speaks its own tool result; an announcement on top would be + # the same receipt twice. + assert talk_cli.input_pause_commands(True, talk_pause.SOURCE_TOOL) == [] + assert talk_cli.input_pause_commands(True, "webhook") == [] + + +def test_the_keyboard_watcher_needs_a_terminal(): + piped = types.SimpleNamespace(isatty=lambda: False) + assert talk_cli.start_keyboard_pause_control(piped) is None + assert talk_cli.start_keyboard_pause_control(None) is None + + +def _raising_isatty(): + raise ValueError("I/O operation on closed file") + + +@pytest.mark.parametrize( + "stdin", + [ + types.SimpleNamespace(isatty=lambda: True), + types.SimpleNamespace(isatty=lambda: False), + types.SimpleNamespace(isatty=_raising_isatty), + object(), + None, + ], +) +def test_the_advertisement_predicate_is_the_watchers_own(stdin): + """One predicate decides both whether the pause tool is offered on the + terminal lane and whether the watcher starts — so the tool can never be + offered on a terminal where Enter would not be read.""" + + available = talk_cli.keyboard_pause_control_available(stdin) + stop = talk_cli.start_keyboard_pause_control(stdin, read_key=lambda _s, e: e.wait(0.01)) + assert available is (stop is not None) + if stop is not None: + stop() + + +def test_windows_extended_keys_are_consumed_whole_and_never_read_as_letters(monkeypatch): + """``msvcrt.getwch()`` hands an arrow, Insert or an F-key as a prefix + ('\\xe0' or '\\x00') and THEN the scan code. Read alone, Down-Arrow's scan + code is 'P' and Insert's is 'R' — a stray arrow key paused the microphone + and Insert resumed it. Both bytes go together now.""" + + down_arrow, insert, f1 = "\xe0P", "\xe0R", "\x00;" + chars = list(down_arrow + insert + f1 + "x" + "\xe9" + "\r" + "p" + "R") + fake_msvcrt = types.SimpleNamespace(kbhit=lambda: bool(chars), getwch=lambda: chars.pop(0)) + monkeypatch.setitem(sys.modules, "msvcrt", fake_msvcrt) + monkeypatch.setattr(sys, "platform", "win32") + + actions = [] + while chars: + actions.append(talk_cli._read_control_key(None, threading.Event())) + + # Three extended keys → three Nones, each eating both bytes; 'x' unknown; + # 'é' alphabetic but not a control; then Enter, p, R do their jobs. + assert actions == [None, None, None, None, None, "toggle", "pause", "resume"] + + +def test_the_keyboard_watcher_flips_the_attached_surface_and_stops_on_request(): + surface = _Surface() + sources: list[str] = [] + talk_pause.attach_session( + surface, lambda _p, s: sources.append(s), resume_control=talk_pause.RESUME_KEYBOARD + ) + keys = iter(["toggle", "toggle", "pause", "resume", "resume", "x", "pause"]) + seen = threading.Event() + + def read_key(_stdin, stop): + try: + key = next(keys) + except StopIteration: + seen.set() + stop.wait(0.01) + return None + return key + + stop = talk_cli.start_keyboard_pause_control( + types.SimpleNamespace(isatty=lambda: True), read_key=read_key + ) + assert stop is not None + assert seen.wait(5.0) + stop() + + # toggle, toggle, pause, resume, resume (a no-op), x (unknown: ignored), + # pause → the flips: + assert surface.calls == ["pause", "resume", "pause", "resume", "pause"] + assert set(sources) == {talk_pause.SOURCE_KEYBOARD} + + +def test_the_keyboard_watcher_survives_a_dead_console(): + def read_key(_stdin, _stop): + raise OSError("console closed") + + stop = talk_cli.start_keyboard_pause_control( + types.SimpleNamespace(isatty=lambda: True), read_key=read_key + ) + assert stop is not None + stop() + + +def test_discord_commands_route_to_the_live_session_and_refuse_without_one(): + assert talk_discord.pause_session() == "I'm not in a voice session right now." + assert talk_discord.resume_session() == "I'm not in a voice session right now." + + class _LiveTask: + def done(self): + return False + + with talk_discord._SESSION_LOCK: + talk_discord._SESSION.update({"task": _LiveTask(), "guild_id": 7, "mode": "legacy"}) + # The session is claimed but has not attached its surface yet. + assert "isn't listening yet" in talk_discord.pause_session() + + audio = talk_audio.DuplexAudio() + sources: list[str] = [] + talk_pause.attach_session( + audio, lambda _p, s: sources.append(s), resume_control=talk_pause.RESUME_COMMAND + ) + assert talk_discord.pause_session() == talk_discord._PAUSE_COMMAND_RECEIPTS[talk_pause.PAUSED] + assert audio.input_paused is True + assert "microphone is paused" in talk_discord.session_status() + assert talk_discord.pause_session() == ( + talk_discord._PAUSE_COMMAND_RECEIPTS[talk_pause.ALREADY_PAUSED] + ) + assert talk_discord.resume_session() == talk_discord._PAUSE_COMMAND_RECEIPTS[talk_pause.RESUMED] + assert "microphone is paused" not in talk_discord.session_status() + assert sources == [talk_pause.SOURCE_COMMAND, talk_pause.SOURCE_COMMAND] + + with talk_discord._SESSION_LOCK: + talk_discord._SESSION["mode"] = "core" + assert "doesn't pause" in talk_discord.pause_session() + + +# -- the live session --------------------------------------------------------- + + +class _PausableAudio: + """A microphone that streams while listening and honours the flag.""" + + played_ms = 0 + playback_pending = False + + def __init__(self): + self.input_paused = False + self.reads = 0 + + def start(self): + pass + + def stop(self): + pass + + def pause_input(self): + self.input_paused = True + + def resume_input(self): + self.input_paused = False + + def read_input_chunk(self): + # Mostly empty, like a real device: a chunk on EVERY read would keep + # the sender from ever sleeping, and the fake wire's send never + # yields, so the receiver would starve. + self.reads += 1 + if self.input_paused or self.reads % 4: + return None + return b"\x00\x00" * 240 + + def queue_playback(self, _pcm): + pass + + def drain_playback(self): + pass + + def reset_played_ms(self): + pass + + +def _run_session(monkeypatch, *, keyboard_control: bool, tty: bool, lane: str = "cli", probe=None): + """One fake session on ``lane``. After the microphone has streamed once, + ``probe`` runs on the wire's side and its return value is kept; the wire + then waits briefly for an operator-flip announcement and hangs up.""" + + sent: list[dict] = [] + marks: list[str] = [] + started: list[bool] = [] + stopped: list[bool] = [] + minted: dict = {} + probed: dict = {} + + class _Message: + type = "text" + + def __init__(self, event): + self.data = json.dumps(event) + + class _WS: + def __init__(self): + self.step = 0 + + async def __aenter__(self): + return self + + async def __aexit__(self, *_args): + return None + + def __aiter__(self): + return self + + async def __anext__(self): + self.step += 1 + if self.step == 1: + return _Message({"type": "response.created", "response": {"id": "r1"}}) + if self.step == 2: + return _Message({"type": "response.done", "response": {"id": "r1"}}) + if self.step == 3: + # Let the microphone stream first, then run the probe while + # the wire is idle. + for _ in range(300): + if "append" in marks: + break + await asyncio.sleep(0.01) + assert talk_pause.is_paused() is False, "the session never attached its audio" + marks.append("PROBE") + if probe is not None: + probed["result"] = probe() + for _ in range(50): + await asyncio.sleep(0.01) + if any("your microphone" in json.dumps(m) for m in sent): + break + await asyncio.sleep(0.05) + raise StopAsyncIteration + raise StopAsyncIteration + + async def send_json(self, message): + sent.append(message) + if message.get("type") == "input_audio_buffer.append": + marks.append("append") + + class _ClientSession: + async def __aenter__(self): + return self + + async def __aexit__(self, *_args): + return None + + def ws_connect(self, *_args, **_kwargs): + return _WS() + + def fake_mint(*_args, **kwargs): + minted.update(kwargs) + return types.SimpleNamespace(client_secret="x") + + def fake_start(*_args, **_kwargs): + started.append(True) + return lambda: stopped.append(True) + + host = types.SimpleNamespace( + resolve_auth=lambda: types.SimpleNamespace(token="token", source="test"), + identity_sections=lambda: {}, + ) + monkeypatch.setattr(talk_cli.talk_host, "host", lambda: host) + monkeypatch.setattr(talk_cli, "_mint_session", fake_mint) + monkeypatch.setattr( + talk_cli, + "_import_aiohttp", + lambda: types.SimpleNamespace( + ClientSession=_ClientSession, + WSMsgType=types.SimpleNamespace(TEXT="text"), + ), + ) + # The real predicate reads sys.stdin; the session must consult it (and + # only it) to decide whether a key exists on this terminal. + monkeypatch.setattr(talk_cli, "keyboard_pause_control_available", lambda *a, **k: tty) + monkeypatch.setattr(talk_cli, "start_keyboard_pause_control", fake_start) + audio = _PausableAudio() + code = asyncio.run( + talk_cli.run_talk_session(audio=audio, lane=lane, keyboard_control=keyboard_control) + ) + assert code == 0 + return types.SimpleNamespace( + sent=sent, + marks=marks, + started=started, + stopped=stopped, + audio=audio, + tools=[tool["name"] for tool in minted["tools"]], + probe=probed.get("result"), + ) + + +def _pause_by_tool(): + return talk_tools.execute_talk_tool("pause_voice_input", {}) + + +def test_the_standalone_terminal_offers_the_pause_and_watches_its_own_keyboard( + monkeypatch, capsys +): + """``hermes talk`` on a real tty: the key exists, so the tool is offered, + the watcher runs, the connected line says so, and an operator flip from + the keyboard is announced in the contained shape and printed with the + way back.""" + + run = _run_session( + monkeypatch, + keyboard_control=True, + tty=True, + probe=lambda: talk_pause.set_paused(True, source=talk_pause.SOURCE_KEYBOARD), + ) + + assert "pause_voice_input" in run.tools + assert run.started == [True] and run.stopped == [True], "watcher not started+stopped once" + assert run.probe == talk_pause.PAUSED + assert "append" in run.marks[: run.marks.index("PROBE")], "the microphone never streamed" + assert "append" not in run.marks[run.marks.index("PROBE") + 1 :], "audio flowed after the pause" + assert run.audio.input_paused is True + + announcement = next( + m for m in run.sent if "paused your microphone from the keyboard" in json.dumps(m) + ) + assert announcement["item"]["role"] == "system" + assert "playback and background work continue" in announcement["item"]["content"][0]["text"] + # Detached at teardown: a pause must never be armed against the next call. + assert talk_pause.is_paused() is None + + out = capsys.readouterr().out + assert "Ctrl+C to hang up, Enter to pause or resume the microphone." in out + assert "talk: microphone paused (Enter to resume)" in out + + +@pytest.mark.parametrize( + ("keyboard_control", "tty", "why"), + [ + (True, False, "hermes talk with a piped or non-tty stdin (mintty, a launcher wrapper)"), + (False, True, "/talk at the Hermes prompt — prompt_toolkit owns the tty"), + (False, False, "neither"), + ], +) +def test_a_terminal_with_no_way_back_offers_no_pause_and_refuses_one( + monkeypatch, capsys, keyboard_control, tty, why +): + """Must-fix from the #105 review: the tool used to be advertised before + the session knew whether a key existed, so a non-tty stdin — or the + Hermes prompt's own terminal — got a pause only Ctrl+C could end. Now + the decision is made once, before the tools are built, and the registry + refuses the pause even if the call arrives anyway.""" + + run = _run_session( + monkeypatch, keyboard_control=keyboard_control, tty=tty, probe=_pause_by_tool + ) + + assert "pause_voice_input" not in run.tools, why + assert run.started == [], f"the watcher must not start: {why}" + assert run.probe == talk_tools.PAUSE_RECEIPTS[talk_pause.NO_RESUME_PATH] + assert run.audio.input_paused is False + # The microphone kept streaming after the refused pause. + assert "append" in run.marks[run.marks.index("PROBE") + 1 :] + assert not any("your microphone" in json.dumps(m) for m in run.sent) + + out = capsys.readouterr().out + assert "Ctrl+C to hang up." in out and "Enter to pause" not in out + assert "microphone paused" not in out + + +def test_the_discord_room_offers_the_pause_with_the_command_as_the_way_back(monkeypatch, capsys): + """The Discord lane has no keyboard and needs none: `/talk resume` is + text, typed, and always there. The tool is offered, the receipt names + that command, and a command flip is announced as one.""" + + run = _run_session( + monkeypatch, + keyboard_control=False, + tty=False, + lane="discord", + # The model pauses; the operator types `/talk resume`. + probe=lambda: ( + _pause_by_tool(), + talk_pause.set_paused(False, source=talk_pause.SOURCE_COMMAND), + ), + ) + + assert "pause_voice_input" in run.tools + assert run.started == [], "a gateway has no keyboard to watch" + tool_receipt, resumed = run.probe + assert tool_receipt.endswith("Tell them how to resume: /talk resume in Discord.") + assert "Enter" not in tool_receipt + assert resumed == talk_pause.RESUMED + assert run.audio.input_paused is False + # The model's own flip is not announced (it speaks its tool result); the + # operator's command is. + assert not any("paused your microphone" in json.dumps(m) for m in run.sent) + assert any("resumed your microphone from a /talk command" in json.dumps(m) for m in run.sent) + + out = capsys.readouterr().out + assert "Enter to pause" not in out + assert "talk: microphone paused\n" in out, "no key hint where there is no key" + assert "talk: microphone listening again" in out + + +def test_cli_entry_grants_the_keyboard_only_to_the_standalone_command(monkeypatch): + """``hermes talk`` arrives through argparse and owns its tty; the bare + ``cli_entry()`` the in-session ``/talk`` makes does not — that terminal + belongs to the Hermes prompt for the whole call.""" + + seen: list[dict] = [] + + async def fake_session(**kwargs): + seen.append(kwargs) + return 0 + + monkeypatch.setattr(talk_cli, "run_talk_session", fake_session) + + assert talk_cli.cli_entry(argparse.Namespace(talk_command="session")) == 0 + assert talk_cli.cli_entry() == 0 + assert talk_cli.cli_entry(keyboard_control=False) == 0 + standalone = argparse.Namespace(talk_command="session") + assert talk_cli.cli_entry(standalone, keyboard_control=False) == 0 + assert [k["keyboard_control"] for k in seen] == [True, False, False, False] + + +def test_the_cli_lane_watches_the_real_stdin_only_when_it_is_a_terminal(monkeypatch): + monkeypatch.setattr(sys, "stdin", types.SimpleNamespace(isatty=lambda: False)) + assert talk_cli.keyboard_pause_control_available() is False + assert talk_cli.start_keyboard_pause_control() is None diff --git a/tests/test_register.py b/tests/test_register.py index 4da8705..177c087 100644 --- a/tests/test_register.py +++ b/tests/test_register.py @@ -497,9 +497,41 @@ def test_slash_command_subcommands_are_gateway_only(plugin): def test_slash_command_runs_the_session_outside_a_loop(plugin, monkeypatch): ctx = StubCtx() plugin.register(ctx) - monkeypatch.setattr(plugin.talk_cli, "cli_entry", lambda *a, **k: 0) + calls: list[dict] = [] + monkeypatch.setattr(plugin.talk_cli, "cli_entry", lambda *a, **k: calls.append(k) or 0) assert ctx.commands["talk"]["handler"]("") == "Voice session ended." + # The Hermes prompt owns this terminal for the whole call (prompt_toolkit, + # raw mode, its own stdin reader): the session must never watch stdin for + # the pause key here (hermes-talk#100). + assert calls == [{"keyboard_control": False}] + + +def test_slash_command_reaches_the_rooms_pause_and_resume(plugin, monkeypatch): + """`/talk pause` / `/talk resume` (hermes-talk#100) are text on purpose: + a paused session hears nobody, so the way back cannot be spoken.""" + + ctx = StubCtx() + plugin.register(ctx) + handler = ctx.commands["talk"]["handler"] + assert "pause" in ctx.commands["talk"]["args_hint"] + calls: list[str] = [] + monkeypatch.setattr( + plugin.talk_discord, "pause_session", lambda: calls.append("pause") or "p", raising=False + ) + monkeypatch.setattr( + plugin.talk_discord, "resume_session", lambda: calls.append("resume") or "r", raising=False + ) + + async def call_from_a_loop(): + return [handler(word) for word in ("pause", "mute", "resume", "unmute")] + + assert asyncio.run(call_from_a_loop()) == ["p", "p", "r", "r"] + assert calls == ["pause", "pause", "resume", "resume"] + # In a terminal the words name the room that isn't there — and the key + # exists only on the standalone command, whose session owns its tty. + reply = handler("pause") + assert "terminal" in reply.lower() and "`hermes talk` command adds Enter" in reply def test_core_absent_process_keeps_legacy_imports_and_reports_optional(): diff --git a/tests/test_tools.py b/tests/test_tools.py index 31b6479..1507ca2 100644 --- a/tests/test_tools.py +++ b/tests/test_tools.py @@ -78,6 +78,18 @@ def test_default_tools_are_fresh_copies(): assert [tool["name"] for tool in talk_tools.default_talk_tools()] == _BASE_TOOLS +def test_the_pause_tool_is_advertised_only_when_the_session_asks_for_it(): + """pause_voice_input (hermes-talk#100) rides on ``pausable``, which the + session sets only when the operator has a guaranteed way to resume; the + default is the safe direction — no pause a key or command cannot undo.""" + + assert [tool["name"] for tool in talk_tools.default_talk_tools(pausable=True)] == [ + *_BASE_TOOLS, + "pause_voice_input", + ] + assert [tool["name"] for tool in talk_tools.default_talk_tools(pausable=False)] == _BASE_TOOLS + + def test_the_vault_tool_is_advertised_only_when_it_can_be_served(monkeypatch): """Advertising a lookup that cannot run is the same defect as the provider block this plugin stopped passing through — the model calls it, @@ -108,10 +120,11 @@ def test_every_advertised_tool_has_a_handler(monkeypatch): middle of a live call.""" monkeypatch.setattr(talk_vault, "available", lambda: True) - names = {tool["name"] for tool in talk_tools.default_talk_tools()} + names = {tool["name"] for tool in talk_tools.default_talk_tools(pausable=True)} assert "search_vault" in names + assert "pause_voice_input" in names - for tool in talk_tools.default_talk_tools(): + for tool in talk_tools.default_talk_tools(pausable=True): assert tool["name"] in talk_tools._HANDLERS assert tool["type"] == "function" assert tool["parameters"]["type"] == "object" @@ -123,6 +136,7 @@ def test_no_handler_is_orphaned(): advertised = {tool["name"] for tool in talk_tools.default_talk_tools()} advertised.add("search_vault") # conditional, absent when unservable + advertised.add("pause_voice_input") # conditional, absent without a way to resume assert set(talk_tools._HANDLERS) == advertised