Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
40 changes: 40 additions & 0 deletions aai_cli/core/microphone.py
Original file line number Diff line number Diff line change
@@ -1,5 +1,8 @@
from __future__ import annotations

import atexit
import contextlib
import signal
import warnings
from abc import abstractmethod
from collections.abc import Callable, Iterable, Iterator, Mapping
Expand Down Expand Up @@ -54,6 +57,42 @@ def audio_missing_error() -> CLIError:
)


# Process-global once-latch. The default is only observable on the very first install
# in a fresh process; the suite mutates this flag across tests, so the load-time value
# can't be asserted in isolation — the check/set in _install_… are what the tests pin.
_shutdown_interrupt_guard_installed = False # pragma: no mutate


def _ignore_interrupt_during_shutdown() -> None:
"""Drop SIGINT for the remainder of interpreter shutdown.

sounddevice registers its own atexit handler that calls ``Pa_Terminate`` to tear
down PortAudio. A second Ctrl-C while that runs raises ``KeyboardInterrupt``
*inside* the atexit callback, which Python reports as a noisy "Exception ignored in
atexit callback" traceback — even though the first Ctrl-C already stopped the
session cleanly. There is nothing left to cancel once we're exiting, so ignore the
late interrupt.
"""
# signal.signal only works on the main thread; atexit runs there, but a ValueError
# is still possible in odd embeddings, so guard it rather than crash the teardown.
with contextlib.suppress(ValueError):
signal.signal(signal.SIGINT, signal.SIG_IGN)


def _install_shutdown_interrupt_guard() -> None:
"""Register ``_ignore_interrupt_during_shutdown`` with atexit exactly once.

Registered *after* sounddevice imports so atexit's LIFO order runs our guard
before sounddevice's PortAudio teardown, neutralizing a second Ctrl-C that would
otherwise raise inside that atexit callback.
"""
global _shutdown_interrupt_guard_installed
if _shutdown_interrupt_guard_installed:
return
atexit.register(_ignore_interrupt_during_shutdown)
_shutdown_interrupt_guard_installed = True


def import_sounddevice() -> ModuleType:
"""Import sounddevice lazily, mapping an ImportError to ``audio_missing_error``.

Expand All @@ -65,6 +104,7 @@ def import_sounddevice() -> ModuleType:
import sounddevice
except ImportError as exc:
raise audio_missing_error() from exc
_install_shutdown_interrupt_guard()
module: ModuleType = sounddevice
return module

Expand Down
2 changes: 2 additions & 0 deletions pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -420,6 +420,8 @@ max-statements = 40
"aai_cli/core/environments.py" = ["PLW0603"]
# Verbosity is process-global startup state by design (mirrors environments.py).
"aai_cli/core/debuglog.py" = ["PLW0603"]
# The "shutdown SIGINT guard installed" latch is process-global once-only state.
"aai_cli/core/microphone.py" = ["PLW0603"]
# BaseHTTPRequestHandler.log_message requires a parameter named `format`.
"aai_cli/auth/loopback.py" = ["A002"]
# Template constants include URL path names such as TOKEN_PATH, not credentials.
Expand Down
51 changes: 51 additions & 0 deletions tests/test_microphone.py
Original file line number Diff line number Diff line change
@@ -1,3 +1,4 @@
import signal
import sys
import types
from typing import Any
Expand All @@ -11,7 +12,10 @@
MicrophoneSource,
_default_mic_stream,
_device_default_rate,
_ignore_interrupt_during_shutdown,
_install_shutdown_interrupt_guard,
_SoundDeviceMic,
import_sounddevice,
resample_pcm16,
)

Expand Down Expand Up @@ -286,3 +290,50 @@ def test_default_mic_stream_missing_sounddevice_raises_mic_missing(monkeypatch):
_default_mic_stream(sample_rate=16000, device=None)
assert exc.value.error_type == "mic_missing"
assert exc.value.exit_code == 2


def test_ignore_interrupt_during_shutdown_sets_sig_ign():
# The guard drops a second Ctrl-C during teardown so it can't raise inside
# sounddevice's atexit PortAudio terminate. Save/restore the global disposition.
before = signal.getsignal(signal.SIGINT)
try:
_ignore_interrupt_during_shutdown()
assert signal.getsignal(signal.SIGINT) is signal.SIG_IGN
finally:
signal.signal(signal.SIGINT, before)


def test_install_shutdown_interrupt_guard_registers_once(monkeypatch):
registered = []
monkeypatch.setattr(microphone, "_shutdown_interrupt_guard_installed", False)
monkeypatch.setattr(microphone.atexit, "register", lambda fn: registered.append(fn))

_install_shutdown_interrupt_guard()
_install_shutdown_interrupt_guard() # idempotent: the flag short-circuits the second call

assert registered == [_ignore_interrupt_during_shutdown]


def test_import_sounddevice_installs_shutdown_guard(monkeypatch):
registered = []
monkeypatch.setattr(microphone, "_shutdown_interrupt_guard_installed", False)
monkeypatch.setattr(microphone.atexit, "register", lambda fn: registered.append(fn))
monkeypatch.setitem(sys.modules, "sounddevice", types.ModuleType("sounddevice"))

import_sounddevice()

assert registered == [_ignore_interrupt_during_shutdown]


def test_import_sounddevice_missing_does_not_register_guard(monkeypatch):
# A broken install raises before the guard is reached, so nothing is registered.
registered = []
monkeypatch.setattr(microphone, "_shutdown_interrupt_guard_installed", False)
monkeypatch.setattr(microphone.atexit, "register", lambda fn: registered.append(fn))
monkeypatch.setitem(sys.modules, "sounddevice", None) # import -> ImportError

with pytest.raises(CLIError) as exc:
import_sounddevice()

assert exc.value.error_type == "mic_missing"
assert registered == []
Loading