docs/USER_GUIDE.md

@@ -44,7 +44,10 @@ The clipboard copy behind these pastes is temporary: your previous clipboard com
 
 Open **Settings → Audio → Input Device** to select a specific microphone or choose
 **System Default**. A valid explicit microphone selection uses that device and
-bypasses the System Default preference below.
+bypasses the System Default preference below. If the selected microphone is
+unplugged, renamed, or is a virtual device, dictation does not switch to another
+input: Voculinux shows an "Audio Input Unavailable" notification and records
+nothing. Choose a microphone again, or System Default, in Settings.
 
 With **System Default**, Voculinux leaves valid default or virtual inputs under
 the audio system's control rather than pinning a hardware device.

src/vocalinux/audio/capture.py

@@ -17,7 +17,7 @@ import subprocess
 import time
 from typing import TYPE_CHECKING, Any, Callable, Optional, cast
 
-from .errors import CaptureStalledError
+from .errors import AudioDeviceUnavailableError, CaptureStalledError
 from .pipewire import (
     PIPEWIRE_INDEX_BASE,
     _test_pipewire_input,
@@ -427,7 +427,7 @@ def _resolve_valid_input_device(audio: Any, preferred_index: Optional[int] = Non
             device_name = "unknown"
         logger.warning(
             "Configured audio device [%s] (%s) is not a safe input device. "
-            "Falling back to a valid input device.",
+            "Choosing another input device.",
             preferred_index,
             device_name,
         )
@@ -438,6 +438,86 @@ def _resolve_valid_input_device(audio: Any, preferred_index: Optional[int] = Non
     return input_device_indices[0]
 
 
+def _selects_system_default(device_index: Optional[int], device_name: Optional[str]) -> bool:
+    """Whether a stored selection means System Default.
+
+    Settings stores System Default as -1 and clears both fields; an unnamed
+    index of -1 (PortAudio's own default sentinel) is the same choice.
+    """
+    return not device_name and (device_index is None or device_index == -1)
+
+
+def _configured_label(device_name: Optional[str], device_index: Optional[int]) -> str:
+    """Name a configured selection in user-facing messages."""
+    label = device_name or f"index {device_index}"
+    return f"Configured audio device '{label}'"
+
+
+def _require_usable_configured_input(
+    audio: Any, device_name: Optional[str], device_index: Optional[int]
+) -> None:
+    """Refuse an explicit microphone that cannot be used exactly as configured.
+
+    The device must be found by name (or by index when no name was stored) and
+    must be a real input. Matching follows :func:`_resolve_device_by_name`, so
+    a device accepted here is the one it returns. Returns quietly when device
+    enumeration itself fails, leaving the stored index to PortAudio as before.
+
+    Raises:
+        AudioDeviceUnavailableError: the device is missing, renamed, virtual or
+            output-only. Nothing is opened, so no other input is recorded.
+    """
+    label = _configured_label(device_name, device_index)
+    try:
+        device_count = int(audio.get_device_count())
+    except (IOError, OSError, TypeError, ValueError, AttributeError):
+        return
+
+    match: Optional[tuple[int, Any]] = None
+    for index in range(max(device_count, 0)):
+        try:
+            info = audio.get_device_info_by_index(index)
+        except (IOError, OSError, TypeError, ValueError, AttributeError):
+            continue
+        if device_name:
+            if isinstance(info, dict) and info.get("name", "") == device_name:
+                match = (index, info)
+                break
+        elif index == device_index:
+            match = (index, info)
+            break
+
+    if match is None:
+        raise AudioDeviceUnavailableError(
+            f"{label} is not connected or was renamed; nothing was recorded from another input."
+        )
+    _, info = match
+    if not isinstance(info, dict):
+        return  # Cannot classify it; keep the same leniency as _resolve_valid_input_device.
+    name = str(info.get("name", ""))
+    channels = info.get("maxInputChannels", 0)
+    has_input = isinstance(channels, (int, float)) and channels > 0
+    if _is_virtual_device(name) or not has_input:
+        raise AudioDeviceUnavailableError(
+            f"{label} is not a usable microphone (virtual or output-only); "
+            "nothing was recorded from another input."
+        )
+
+
+def _resolve_configured_input(
+    audio: Any, device_name: Optional[str], device_index: Optional[int]
+) -> int:
+    """Resolve an explicit microphone selection, or raise. Never substitutes a device."""
+    _require_usable_configured_input(audio, device_name, device_index)
+    resolved = _resolve_device_by_name(audio, device_name, device_index)
+    if resolved is None:
+        raise AudioDeviceUnavailableError(
+            f"{_configured_label(device_name, device_index)} "
+            "could not be opened; nothing was recorded from another input."
+        )
+    return resolved
+
+
 def _resolve_system_default(audio: Any) -> Optional[int]:
     """Prefer ALSA's sound-server PCM when PortAudio substituted hardware.
 
@@ -971,25 +1051,16 @@ class PortAudioCaptureSource:
         # PulseAudio source opens the default device with PULSE_SOURCE set.
         pulse_selected, pulse_source = _resolve_pulse_selection(self.device_index, self.device_name)
         pulse_name = pulse_source.name if pulse_source is not None else None
-        use_system_default = pulse_selected or (
-            self.device_index is None and self.device_name is None
+        use_system_default = pulse_selected or _selects_system_default(
+            self.device_index, self.device_name
         )
         if use_system_default:
             resolved_device_index = _resolve_system_default(audio)
         else:
-            resolved_device_index = _resolve_device_by_name(
+            # An explicit selection is used exactly or refused (issue #42).
+            resolved_device_index = _resolve_configured_input(
                 audio, self.device_name, self.device_index
             )
-            if resolved_device_index is None:
-                resolved_device_index = _resolve_valid_input_device(audio, None)
-        if resolved_device_index is None and not use_system_default:
-            # No safe enumerated mic left (e.g. only PipeWire pseudo devices).
-            # Fall back to PortAudio system default instead of aborting.
-            logger.warning(
-                "No safe audio input devices enumerated; " "falling back to system default capture."
-            )
-            resolved_device_index = _resolve_system_default(audio)
-            use_system_default = True
 
         # Log available devices for debugging (skip virtual devices)
         logger.debug("Available audio input devices:")
@@ -1117,23 +1188,16 @@ class PortAudioCaptureSource:
                 self.device_index, self.device_name
             )
             pulse_name = pulse_source.name if pulse_source is not None else None
-            use_system_default = pulse_selected or (
-                self.device_index is None and self.device_name is None
+            use_system_default = pulse_selected or _selects_system_default(
+                self.device_index, self.device_name
             )
             if use_system_default:
                 resolved_device_index = _resolve_system_default(audio_instance)
             else:
-                resolved_device_index = _resolve_device_by_name(
+                # An explicit selection is used exactly or refused (issue #42).
+                resolved_device_index = _resolve_configured_input(
                     audio_instance, self.device_name, self.device_index
                 )
-                if resolved_device_index is None:
-                    resolved_device_index = _resolve_valid_input_device(audio_instance, None)
-            if resolved_device_index is None and not use_system_default:
-                logger.warning(
-                    "Reconnection: no safe input devices enumerated; "
-                    "falling back to system default capture."
-                )
-                resolved_device_index = _resolve_system_default(audio_instance)
 
             # Negotiate the format and open the capture stream in ONE
             # PortAudio open — Bluetooth SCO devices abort with heap

src/vocalinux/audio/errors.py

@@ -3,3 +3,12 @@
 
 class CaptureStalledError(IOError):
     """The capture device stopped delivering audio frames."""
+
+
+class AudioDeviceUnavailableError(IOError):
+    """The configured capture device cannot be used as configured.
+
+    Raised instead of recording from another input: a configured device that
+    is missing, renamed, virtual or output-only must fail visibly (issue #42).
+    The message is user-facing and names the configured device.
+    """

src/vocalinux/audio/pipewire.py

@@ -25,10 +25,10 @@ import shutil
 import subprocess
 import time
 from dataclasses import dataclass
-from typing import Any, Callable, List, Optional, Tuple
+from typing import Any, Callable, Dict, List, Mapping, Optional, Tuple
 
 from ..utils.host_process import host_env
-from .errors import CaptureStalledError
+from .errors import AudioDeviceUnavailableError, CaptureStalledError
 
 logger = logging.getLogger(__name__)
 
@@ -52,6 +52,28 @@ _MONITOR_CLASSES = frozenset({"Audio/Sink", "Audio/Sink/Virtual"})
 #: Metadata keys carrying the effective and configured default sink.
 _DEFAULT_SINK_KEYS = ("default.audio.sink", "default.configured.audio.sink")
 
+#: Stream property that makes ``--target <sink>`` record the sink's monitor.
+#: Without it PipeWire also links the default microphone to the stream (issue
+#: #41), so system-audio capture must always set it.
+_CAPTURE_SINK_PROPERTY = "stream.capture.sink = true"
+
+
+def _capture_sink_env(base: Mapping[str, str]) -> Dict[str, str]:
+    """Return ``base`` with ``PIPEWIRE_PROPS`` pinning capture to the sink monitor.
+
+    This is the form verified on the real machine (PipeWire 1.6.10): the stream
+    linked the sink's monitor and no microphone. A user's existing
+    ``PIPEWIRE_PROPS`` is kept, and only this child process sees the change.
+    """
+    env = dict(base)
+    existing = env.get("PIPEWIRE_PROPS", "").strip()
+    if existing.startswith("{") and existing.endswith("}"):
+        existing = existing[1:-1].strip()
+    parts = [part for part in (existing, _CAPTURE_SINK_PROPERTY) if part]
+    env["PIPEWIRE_PROPS"] = "{ " + " ".join(parts) + " }"
+    return env
+
+
 _COMMAND_TIMEOUT_SECONDS = 3.0
 
 CommandResult = Tuple[int, str, str]
@@ -346,7 +368,8 @@ class PipeWireCaptureSource:
 
         Raises:
             FileNotFoundError: when ``pw-record`` is not installed.
-            IOError: when the stored selection no longer resolves to a source.
+            AudioDeviceUnavailableError: when the stored selection no longer
+                resolves to a source (an IOError subclass).
         """
         if self.stream is not None:
             return
@@ -356,7 +379,11 @@ class PipeWireCaptureSource:
         if source is None:
             # Deliberately no PortAudio fallback: silently switching to a
             # microphone would capture the wrong audio.
-            raise IOError("PipeWire source is no longer available")
+            label = self.device_name or f"index {self.device_index}"
+            raise AudioDeviceUnavailableError(
+                f"System audio source '{label}' is no longer available; "
+                "nothing was recorded from another input."
+            )
         self.source = source
         self.stream = self._popen(
             [
@@ -374,7 +401,7 @@ class PipeWireCaptureSource:
             ],
             stdout=subprocess.PIPE,
             stderr=subprocess.DEVNULL,
-            env=host_env(),
+            env=host_env(_capture_sink_env(os.environ)),
         )
 
     def reopen(

src/vocalinux/speech_recognition/recognition_manager.py

@@ -30,6 +30,7 @@ if TYPE_CHECKING:
 from ..app_identity import APP_NAME, RESOURCE_DIR_NAME, user_agent
 from ..audio.capture import (  # noqa: F401  (re-exported for compatibility)
     _STICKY_LOCK_MIN_MEAN_SQUARE,
+    AudioDeviceUnavailableError,
     CaptureStalledError,
     PortAudioCaptureSource,
     _alsa_handler,
@@ -112,6 +113,9 @@ class _CaptureSession:
         self.stopping = False
         self.stall_recovery_attempted = False
         self.stall_error: Optional[OSError] = None
+        # A configured device that cannot be used (issue #42), kept for the
+        # failure notice when the read error that ended the session is not it.
+        self.unavailable_error: Optional[AudioDeviceUnavailableError] = None
         self.device_name = "System Default"
         self.segments: queue.Queue = queue.Queue(maxsize=32)
         # A dispatched callback can outlive the bounded recognition join.
@@ -3955,6 +3959,10 @@ class SpeechRecognitionManager:
         def recover(error: OSError) -> bool:
             if not recording():
                 return False
+            if isinstance(error, AudioDeviceUnavailableError):
+                # A configured device that cannot be used does not recover by retrying.
+                session.unavailable_error = session.unavailable_error or error
+                return False
             if session.stall_error is None:
                 session.device_name = self._capture_device_name(source)
             if isinstance(error, CaptureStalledError):
@@ -3965,6 +3973,8 @@ class SpeechRecognitionManager:
                 session.stall_recovery_attempted = True
             recovered = self._attempt_audio_reconnection(audio, should_continue=recording)
             validation_error = getattr(self._capture_source, "last_error", None)
+            if isinstance(validation_error, AudioDeviceUnavailableError):
+                session.unavailable_error = session.unavailable_error or validation_error
             if isinstance(validation_error, CaptureStalledError):
                 session.stall_error = session.stall_error or validation_error
                 session.stall_recovery_attempted = True
@@ -4324,7 +4334,17 @@ class SpeechRecognitionManager:
         play_error_sound()
         if isinstance(error, CaptureStalledError):
             session.stall_error = session.stall_error or error
-        if session.stall_error is not None:
+        unavailable = session.unavailable_error
+        if unavailable is None and isinstance(error, AudioDeviceUnavailableError):
+            unavailable = error
+        if unavailable is not None:
+            logger.error("Audio input unavailable: %s", unavailable)
+            _show_notification(
+                "Audio Input Unavailable",
+                f"{unavailable} Choose a microphone in Settings, then try dictation again.",
+                "dialog-warning",
+            )
+        elif session.stall_error is not None:
             logger.error("Microphone '%s' stalled: %s", session.device_name, session.stall_error)
             _show_notification(
                 "Microphone Stalled",

tests/test_capture_recovery.py

@@ -385,3 +385,79 @@ def test_successful_reopen_updates_identity_for_a_later_stall(
     for stream in streams:
         stream.close.assert_called_once()
     audio.terminate.assert_called_once()
+
+
+def test_unusable_configured_device_is_reported_without_retrying(
+    manager: rm.SpeechRecognitionManager,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """A configured virtual mic fails visibly; it is not reopened or replaced (#42)."""
+    source = capture.PortAudioCaptureSource(device_index=0, device_name="pipewire")
+    devices = [
+        {"index": 0, "name": "pipewire", "maxInputChannels": 32},
+        {"index": 1, "name": "Built-in Mic", "maxInputChannels": 2},
+    ]
+    audio = Mock()
+    audio.get_device_count.return_value = 2
+    audio.get_device_info_by_index.side_effect = lambda i: devices[i]
+    audio.get_default_input_device_info.return_value = {"index": 1, "name": "Built-in Mic"}
+    monkeypatch.setitem(sys.modules, "pyaudio", Mock(PyAudio=lambda: audio, paInt16=8))
+    monkeypatch.setattr(manager, "_new_capture_source", lambda: source)
+    opener = Mock(return_value=(1, 16000, Mock()))
+    monkeypatch.setattr(capture, "_open_capture_stream", opener)
+    reconnect = Mock(return_value=False)
+    monkeypatch.setattr(manager, "_attempt_audio_reconnection", reconnect)
+    manager.should_record = True
+    manager.state = RecognitionState.LISTENING
+
+    manager._record_audio()
+
+    opener.assert_not_called()
+    audio.open.assert_not_called()
+    reconnect.assert_not_called()
+    title, message = rm._show_notification.call_args.args[:2]
+    assert title == "Audio Input Unavailable"
+    assert "'pipewire'" in message
+    assert "Choose a microphone in Settings" in message
+    rm.play_error_sound.assert_called()
+    assert manager.state == RecognitionState.IDLE
+
+
+def test_configured_device_lost_mid_dictation_is_reported(
+    manager: rm.SpeechRecognitionManager,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """A read error followed by a refused reconnect still reports the cause (#42)."""
+    source = capture.PortAudioCaptureSource(device_index=0, device_name="USB Microphone")
+    devices = {
+        0: {"index": 0, "name": "USB Microphone", "maxInputChannels": 1},
+        1: {"index": 1, "name": "Built-in Mic", "maxInputChannels": 2},
+    }
+    audio = Mock()
+    audio.get_device_count.return_value = 2
+    audio.get_device_info_by_index.side_effect = lambda i: devices[i]
+    audio.get_default_input_device_info.return_value = {"index": 1, "name": "Built-in Mic"}
+    monkeypatch.setitem(sys.modules, "pyaudio", Mock(PyAudio=lambda: audio, paInt16=8))
+    monkeypatch.setattr(manager, "_new_capture_source", lambda: source)
+    opener = Mock(return_value=(1, 16000, Mock()))
+    monkeypatch.setattr(capture, "_open_capture_stream", opener)
+
+    def read(**_kwargs: object) -> bytes:
+        # The microphone is renamed while dictating: the reconnect must refuse it.
+        devices[0] = {"index": 0, "name": "Renamed USB", "maxInputChannels": 1}
+        raise OSError("device lost")
+
+    monkeypatch.setattr(source, "read_chunk", read)
+    # Reconnection rebuilds the source from the manager's configured device.
+    manager.audio_device_index = 0
+    manager.audio_device_name = "USB Microphone"
+    manager.should_record = True
+    manager.state = RecognitionState.LISTENING
+
+    manager._record_audio()
+
+    opener.assert_called_once()
+    title, message = rm._show_notification.call_args.args[:2]
+    assert title == "Audio Input Unavailable"
+    assert "'USB Microphone'" in message
+    assert manager.state == RecognitionState.IDLE

tests/test_pipewire_capture.py

@@ -11,6 +11,7 @@ import unittest
 from typing import Any, Callable, Dict, List, Optional, Tuple
 from unittest.mock import MagicMock, patch
 
+from vocalinux.audio.errors import AudioDeviceUnavailableError
 from vocalinux.audio.pipewire import (
     PIPEWIRE_INDEX_BASE,
     SYSTEM_AUDIO_PREFIX,
@@ -347,6 +348,43 @@ class TestPipeWireCaptureSource(unittest.TestCase):
         assert "env" in kwargs  # host binaries get the de-bundled environment
         assert capture.stream is process
 
+    def test_open_pins_capture_to_the_sink_monitor(self) -> None:
+        # Without stream.capture.sink, pw-record --target <sink> also links the
+        # default microphone (issue #41). Set it through PIPEWIRE_PROPS, the
+        # form verified on the real machine.
+        calls: List[Tuple[List[str], Dict[str, Any]]] = []
+        with (
+            patch("vocalinux.audio.pipewire.resolve_pipewire_source", return_value=_source()),
+            patch.dict("os.environ", {"PATH": "/usr/bin"}, clear=True),
+        ):
+            capture = PipeWireCaptureSource(
+                device_index=PIPEWIRE_INDEX_BASE,
+                popen=_popen_factory(_FakeProcess(), calls),
+                which=lambda n: "/usr/bin/pw-record",
+            )
+            capture.open()
+        args, kwargs = calls[0]
+        assert args[args.index("--target") + 1] == "alsa_output.dac"
+        assert args[-1] == "-"
+        assert "stream.capture.sink = true" in kwargs["env"]["PIPEWIRE_PROPS"]
+        assert kwargs["env"]["PATH"] == "/usr/bin"
+
+    def test_open_keeps_user_pipewire_props_when_pinning_the_monitor(self) -> None:
+        calls: List[Tuple[List[str], Dict[str, Any]]] = []
+        user_props = "{ node.latency = 1024/48000 }"
+        with (
+            patch("vocalinux.audio.pipewire.resolve_pipewire_source", return_value=_source()),
+            patch.dict("os.environ", {"PIPEWIRE_PROPS": user_props}, clear=True),
+        ):
+            capture = PipeWireCaptureSource(
+                device_index=PIPEWIRE_INDEX_BASE,
+                popen=_popen_factory(_FakeProcess(), calls),
+                which=lambda n: "/usr/bin/pw-record",
+            )
+            capture.open()
+        props = calls[0][1]["env"]["PIPEWIRE_PROPS"]
+        assert props == "{ node.latency = 1024/48000 stream.capture.sink = true }"
+
     def test_open_requires_pw_record(self) -> None:
         capture = PipeWireCaptureSource(device_index=PIPEWIRE_INDEX_BASE, which=lambda name: None)
         with self.assertRaises(FileNotFoundError):
@@ -358,6 +396,14 @@ class TestPipeWireCaptureSource(unittest.TestCase):
             with self.assertRaises(IOError):
                 capture.open()
 
+    def test_open_reports_a_gone_source_as_unavailable(self) -> None:
+        # No PortAudio fallback: the typed error lets the manager tell the user.
+        with patch("vocalinux.audio.pipewire.resolve_pipewire_source", return_value=None):
+            capture = PipeWireCaptureSource(device_index=PIPEWIRE_INDEX_BASE, which=lambda n: "x")
+            with self.assertRaises(AudioDeviceUnavailableError) as ctx:
+                capture.open()
+        assert "no longer available" in str(ctx.exception)
+
     def test_read_chunk_returns_exact_frame_bytes(self) -> None:
         payload = b"\x01\x02" * 4096
         process = _FakeProcess(payload)

tests/test_pulse_sources.py

@@ -389,13 +389,13 @@ class TestPulseSourceEnv:
         assert leaked is None
 
 
-def _pyaudio_with_one_mic() -> Tuple[MagicMock, MagicMock]:
+def _pyaudio_with_one_mic(name: str = "default") -> Tuple[MagicMock, MagicMock]:
     mock_audio = MagicMock()
     mock_audio.get_device_count.return_value = 1
-    mock_audio.get_default_input_device_info.return_value = {"index": 0, "name": "default"}
+    mock_audio.get_default_input_device_info.return_value = {"index": 0, "name": name}
     mock_audio.get_device_info_by_index.return_value = {
         "index": 0,
-        "name": "default",
+        "name": name,
         "maxInputChannels": 32,
     }
     return mock_audio, MagicMock(PyAudio=MagicMock(return_value=mock_audio))
@@ -434,9 +434,13 @@ class TestCaptureIntegration:
         assert devices == [(PULSE_INDEX_BASE, "Yeti", False)]
 
     def _open_recording_env(
-        self, device_index: int, device_name: str, sources: List[PulseSource]
+        self,
+        device_index: int,
+        device_name: str,
+        sources: List[PulseSource],
+        mic_name: str = "default",
     ) -> Tuple[Dict[str, object], Optional[str]]:
-        mock_audio, mock_pyaudio = _pyaudio_with_one_mic()
+        mock_audio, mock_pyaudio = _pyaudio_with_one_mic(mic_name)
         seen: Dict[str, object] = {}
 
         def fake_open(audio: Any, index: Optional[int]) -> Tuple[int, int, MagicMock]:
@@ -467,7 +471,10 @@ class TestCaptureIntegration:
         assert leaked is None
 
     def test_portaudio_selection_is_untouched(self) -> None:
-        seen, _leaked = self._open_recording_env(0, "default", self.sources)
+        # A real configured microphone; an explicit virtual "default" is now refused (#42).
+        seen, _leaked = self._open_recording_env(
+            0, "USB Microphone", self.sources, mic_name="USB Microphone"
+        )
         assert seen["env"] is None
 
     def test_audio_test_uses_the_chosen_source(self) -> None:

tests/test_recognition_manager_device_config.py

@@ -160,8 +160,8 @@ class TestAudioDeviceDetection(unittest.TestCase):
 
         mock_open.assert_called_once_with(mock_audio, None)
 
-    def test_record_audio_falls_back_to_system_default_when_no_safe_device(self):
-        """Stale/unsafe saved devices should reopen via PortAudio system default."""
+    def test_record_audio_refuses_system_default_for_unusable_configured_device(self):
+        """An unusable configured mic must not reopen the system default (issue #42)."""
         manager = _make_manager(
             engine="whisper_cpp",
             audio_device_index=0,
@@ -172,14 +172,6 @@ class TestAudioDeviceDetection(unittest.TestCase):
         manager.silence_timeout = 0.05
         manager._silero_vad = None
 
-        mock_stream = MagicMock()
-
-        def _read_once(*_args, **_kwargs):
-            manager.should_record = False
-            return b"\x00" * 2048
-
-        mock_stream.get_read_available.return_value = 1024
-        mock_stream.read.side_effect = _read_once
         mock_audio = MagicMock()
         mock_audio.get_device_count.return_value = 1
         mock_audio.get_device_info_by_index.return_value = {
@@ -196,17 +188,9 @@ class TestAudioDeviceDetection(unittest.TestCase):
 
         with (
             patch.dict("sys.modules", {"pyaudio": mock_pyaudio}),
-            patch(
-                "vocalinux.audio.capture._resolve_device_by_name",
-                return_value=None,
-            ),
-            patch(
-                "vocalinux.audio.capture._resolve_valid_input_device",
-                return_value=None,
-            ),
             patch(
                 "vocalinux.audio.capture._open_capture_stream",
-                return_value=(1, 16000, mock_stream),
+                return_value=(1, 16000, MagicMock()),
             ) as mock_open,
             patch("vocalinux.speech_recognition.recognition_manager.play_error_sound"),
         ):
@@ -214,8 +198,9 @@ class TestAudioDeviceDetection(unittest.TestCase):
                 del sys.modules["numpy"]
             manager._record_audio()
 
-        mock_open.assert_called_once_with(mock_audio, None)
-        mock_audio.get_default_input_device_info.assert_called()
+        mock_open.assert_not_called()
+        mock_audio.open.assert_not_called()
+        assert manager._buffered_capture_failed
 
     def test_test_audio_input_system_default_fallback_open_omits_device_index(self):
         """Fallback mic-test open for System Default must omit input_device_index."""
@@ -946,6 +931,120 @@ class TestDownmixToMono(unittest.TestCase):
         assert sticky3 == 2
 
 
+class TestExplicitMicrophoneSelection(unittest.TestCase):
+    """A configured microphone is recorded exactly, or not at all (issue #42)."""
+
+    def _audio(self, devices):
+        audio = MagicMock()
+        audio.get_device_count.return_value = len(devices)
+        audio.get_device_info_by_index.side_effect = lambda i: devices[i]
+        audio.get_default_input_device_info.return_value = {"index": 2, "name": "Built-in Mic"}
+        return audio
+
+    def _pyaudio_module(self):
+        module = MagicMock(paInt16=8)
+        module.PyAudio.return_value = MagicMock()
+        return module
+
+    def _devices(self):
+        return [
+            {"index": 0, "name": "pipewire", "maxInputChannels": 32},
+            {"index": 1, "name": "USB Microphone", "maxInputChannels": 1},
+            {"index": 2, "name": "Built-in Mic", "maxInputChannels": 2},
+        ]
+
+    def test_virtual_configured_device_is_refused_not_replaced(self):
+        """A configured virtual device must not open the hardware default instead."""
+        from vocalinux.audio.capture import PortAudioCaptureSource
+        from vocalinux.audio.errors import AudioDeviceUnavailableError
+
+        audio = self._audio(self._devices())
+        source = PortAudioCaptureSource(device_index=0, device_name="pipewire")
+        with (
+            patch.dict("sys.modules", {"pyaudio": self._pyaudio_module()}),
+            patch(
+                "vocalinux.audio.capture._open_capture_stream",
+                return_value=(1, 16000, MagicMock()),
+            ) as opener,
+        ):
+            with self.assertRaises(AudioDeviceUnavailableError) as ctx:
+                source.open(audio)
+        opener.assert_not_called()
+        audio.open.assert_not_called()
+        assert "'pipewire'" in str(ctx.exception)
+
+    def test_configured_device_that_disappeared_is_not_switched(self):
+        """A missing configured mic must not reuse the stored index for another input."""
+        from vocalinux.audio.capture import PortAudioCaptureSource
+        from vocalinux.audio.errors import AudioDeviceUnavailableError
+
+        devices = self._devices()
+        devices[1] = {"index": 1, "name": "Built-in Mic", "maxInputChannels": 2}
+        devices[2] = {"index": 2, "name": "Other Mic", "maxInputChannels": 1}
+        audio = self._audio(devices)
+        source = PortAudioCaptureSource(device_index=1, device_name="USB Microphone")
+        with (
+            patch.dict("sys.modules", {"pyaudio": self._pyaudio_module()}),
+            patch(
+                "vocalinux.audio.capture._open_capture_stream",
+                return_value=(1, 16000, MagicMock()),
+            ) as opener,
+        ):
+            with self.assertRaises(AudioDeviceUnavailableError):
+                source.open(audio)
+        opener.assert_not_called()
+
+    def test_configured_usable_device_opens_by_name(self):
+        """A configured microphone that is present still opens at its index."""
+        from vocalinux.audio.capture import PortAudioCaptureSource
+
+        audio = self._audio(self._devices())
+        source = PortAudioCaptureSource(device_index=1, device_name="USB Microphone")
+        with (
+            patch.dict("sys.modules", {"pyaudio": self._pyaudio_module()}),
+            patch(
+                "vocalinux.audio.capture._open_capture_stream",
+                return_value=(1, 16000, MagicMock()),
+            ) as opener,
+        ):
+            source.open(audio)
+        opener.assert_called_once_with(audio, 1)
+
+    def test_system_default_sentinel_still_opens_the_default_device(self):
+        """A stored -1 is PortAudio's System Default, not a device to look up."""
+        from vocalinux.audio.capture import PortAudioCaptureSource
+
+        audio = self._audio(self._devices())
+        source = PortAudioCaptureSource(device_index=-1, device_name=None)
+        with (
+            patch.dict("sys.modules", {"pyaudio": self._pyaudio_module()}),
+            patch(
+                "vocalinux.audio.capture._open_capture_stream",
+                return_value=(1, 16000, MagicMock()),
+            ) as opener,
+        ):
+            source.open(audio)
+        opener.assert_called_once_with(audio, None)
+
+    def test_reopen_refuses_a_virtual_configured_device(self):
+        """Reconnection reports the refusal through last_error and opens nothing."""
+        from vocalinux.audio.capture import PortAudioCaptureSource
+        from vocalinux.audio.errors import AudioDeviceUnavailableError
+
+        audio = self._audio(self._devices())
+        source = PortAudioCaptureSource(device_index=0, device_name="pipewire")
+        with (
+            patch.dict("sys.modules", {"pyaudio": self._pyaudio_module()}),
+            patch(
+                "vocalinux.audio.capture._open_capture_stream",
+                return_value=(1, 16000, MagicMock()),
+            ) as opener,
+        ):
+            assert source.reopen(audio) is False
+        opener.assert_not_called()
+        assert isinstance(source.last_error, AudioDeviceUnavailableError)
+
+
 class TestRecordAudioNegotiationFallback(unittest.TestCase):
     """Cover _record_audio fallback when stream negotiation returns None."""
 

tests/test_recognition_manager_downloads.py

@@ -1295,76 +1295,62 @@ class TestAudioReconnection:
         assert result is True
         assert manager._audio_stream == mock_stream
 
-    def test_attempt_audio_reconnection_falls_back_to_default_resolver(self):
-        """Test reconnection falls back when saved device name/index cannot resolve."""
+    def test_attempt_audio_reconnection_refuses_missing_configured_device(self):
+        """A configured mic that is gone must fail visibly, not reopen the default (#42)."""
         manager = _make_manager(engine="whisper_cpp", audio_device_name="Missing Mic")
-
         mock_pyaudio_mod = MagicMock()
         mock_pyaudio_mod.paInt16 = 8
-        mock_stream = MagicMock()
-        mock_stream.get_read_available.return_value = 1024
-        mock_stream.read.return_value = b"\x00" * 1024
         mock_audio_instance = MagicMock()
-        mock_audio_instance.get_default_input_device_info.return_value = {"index": 0}
-        mock_audio_instance.open.return_value = mock_stream
+        mock_audio_instance.get_device_count.return_value = 1
+        mock_audio_instance.get_device_info_by_index.return_value = {
+            "index": 0,
+            "name": "Built-in Mic",
+            "maxInputChannels": 2,
+        }
+        mock_audio_instance.get_default_input_device_info.return_value = {
+            "index": 0,
+            "name": "Built-in Mic",
+        }
 
         with (
             patch.dict("sys.modules", {"pyaudio": mock_pyaudio_mod}),
             patch("time.sleep"),
-            patch(
-                "vocalinux.audio.capture._resolve_device_by_name",
-                return_value=None,
-            ) as mock_resolve_name,
-            patch(
-                "vocalinux.audio.capture._resolve_valid_input_device",
-                return_value=1,
-            ) as mock_resolve_default,
-            patch(
-                "vocalinux.audio.capture._get_supported_channels",
-                return_value=1,
-            ),
-            patch(
-                "vocalinux.audio.capture._get_supported_sample_rate",
-                return_value=16000,
-            ),
-        ):
-            result = manager._attempt_audio_reconnection(mock_audio_instance)
-
-        assert result is True
-        assert manager._audio_stream == mock_stream
-        mock_resolve_name.assert_called_once_with(mock_audio_instance, "Missing Mic", None)
-        mock_resolve_default.assert_called_once_with(mock_audio_instance, None)
-
-    def test_attempt_audio_reconnection_no_resolved_device(self):
-        """When no safe device is enumerated, reconnect via system default."""
-        manager = _make_manager(engine="whisper_cpp", audio_device_name="Missing Mic")
-
-        mock_pyaudio_mod = MagicMock()
-        mock_pyaudio_mod.paInt16 = 8
-        mock_audio_instance = MagicMock()
-        mock_stream = MagicMock()
-        mock_stream.get_read_available.return_value = 1024
-
-        with (
-            patch.dict("sys.modules", {"pyaudio": mock_pyaudio_mod}),
-            patch("time.sleep"),
-            patch(
-                "vocalinux.audio.capture._resolve_device_by_name",
-                return_value=None,
-            ),
-            patch(
-                "vocalinux.audio.capture._resolve_valid_input_device",
-                return_value=None,
-            ),
             patch(
                 "vocalinux.audio.capture._open_capture_stream",
-                return_value=(1, 16000, mock_stream),
+                return_value=(1, 16000, MagicMock()),
             ) as mock_open,
         ):
             result = manager._attempt_audio_reconnection(mock_audio_instance)
 
-        assert result is True
-        mock_open.assert_called_once_with(mock_audio_instance, None)
+        assert result is False
+        mock_open.assert_not_called()
+        mock_audio_instance.open.assert_not_called()
+
+    def test_attempt_audio_reconnection_refuses_when_no_safe_input_exists(self):
+        """With no usable input enumerated, reconnection refuses rather than using the default."""
+        manager = _make_manager(engine="whisper_cpp", audio_device_name="Missing Mic")
+        mock_pyaudio_mod = MagicMock()
+        mock_pyaudio_mod.paInt16 = 8
+        mock_audio_instance = MagicMock()
+        mock_audio_instance.get_device_count.return_value = 1
+        mock_audio_instance.get_device_info_by_index.return_value = {
+            "index": 0,
+            "name": "pipewire",
+            "maxInputChannels": 32,
+        }
+
+        with (
+            patch.dict("sys.modules", {"pyaudio": mock_pyaudio_mod}),
+            patch("time.sleep"),
+            patch(
+                "vocalinux.audio.capture._open_capture_stream",
+                return_value=(1, 16000, MagicMock()),
+            ) as mock_open,
+        ):
+            result = manager._attempt_audio_reconnection(mock_audio_instance)
+
+        assert result is False
+        mock_open.assert_not_called()
 
     def test_attempt_audio_reconnection_max_attempts(self):
         """Test reconnection stops after max attempts."""