fix: disable Windows console QuickEdit mode on startup

src/croupier/main.py

@@ -1,7 +1,14 @@
+import atexit
+import ctypes
 import logging
+import sys
+from contextlib import contextmanager
 from logging.handlers import RotatingFileHandler
 from pathlib import Path
+from typing import TYPE_CHECKING
+from typing import Any
 from typing import ClassVar
+from typing import NoReturn
 
 import uvicorn
 from escpos.printer import Network
@@ -21,6 +28,9 @@
 from starlette.status import HTTP_204_NO_CONTENT
 from starlette.status import HTTP_500_INTERNAL_SERVER_ERROR
 
+if TYPE_CHECKING:
+    from collections.abc import Iterator
+
 
 class Settings(BaseSettings):
     model_config: ClassVar[SettingsConfigDict] = SettingsConfigDict(
@@ -76,6 +86,129 @@ async def health(request: Request) -> Response:
     return Response(status_code=HTTP_500_INTERNAL_SERVER_ERROR)
 
 
+logger = logging.getLogger(__name__)
+
+INVALID_HANDLE_VALUE = ctypes.c_void_p(-1).value
+GENERIC_READ = 0x80000000
+GENERIC_WRITE = 0x40000000
+FILE_SHARE_READ = 0x1
+FILE_SHARE_WRITE = 0x2
+OPEN_EXISTING = 3
+ERROR_FILE_NOT_FOUND = 2
+ERROR_INVALID_HANDLE = 6
+# CreateFileW("CONIN$") fails with one of these when no console is attached.
+NO_CONSOLE_ERRORS = frozenset({ERROR_FILE_NOT_FOUND, ERROR_INVALID_HANDLE})
+ENABLE_QUICK_EDIT_MODE = 0x0040
+ENABLE_EXTENDED_FLAGS = 0x0080
+
+
+def disable_console_quick_edit() -> None:
+    """Disable Windows console QuickEdit mode for the life of the process.
+
+    With QuickEdit on, a click inside the console starts a text selection,
+    which blocks writes to stdout and freezes the service until a key is
+    pressed.
+
+    The console input mode is shared with the parent shell, so the original
+    mode is restored at exit. This is a conhost setting; it may have no effect
+    under Windows Terminal. Failures are logged and never prevent startup.
+    """
+    if sys.platform == "win32":
+        try:
+            _clear_quick_edit_flag()
+        except OSError:
+            logger.warning("Could not disable console QuickEdit mode", exc_info=True)
+
+
+def _kernel32() -> Any:  # noqa: ANN401
+    kernel32 = ctypes.WinDLL(  # type: ignore[attr-defined]  # pyrefly: ignore[missing-attribute]
+        "kernel32", use_last_error=True
+    )
+    kernel32.CreateFileW.restype = ctypes.c_void_p
+    kernel32.CreateFileW.argtypes = [
+        ctypes.c_wchar_p,
+        ctypes.c_uint32,
+        ctypes.c_uint32,
+        ctypes.c_void_p,
+        ctypes.c_uint32,
+        ctypes.c_uint32,
+        ctypes.c_void_p,
+    ]
+    kernel32.GetConsoleMode.argtypes = [
+        ctypes.c_void_p,
+        ctypes.POINTER(ctypes.c_uint32),
+    ]
+    kernel32.SetConsoleMode.argtypes = [ctypes.c_void_p, ctypes.c_uint32]
+    kernel32.CloseHandle.argtypes = [ctypes.c_void_p]
+    return kernel32
+
+
+def _raise_last_error(call: str) -> NoReturn:
+    error = ctypes.get_last_error()  # type: ignore[attr-defined]  # pyrefly: ignore[missing-attribute]
+    msg = f"{call} failed (error {error})"
+    raise OSError(msg)
+
+
+@contextmanager
+def _console_input(kernel32: Any) -> Iterator[int | None]:  # noqa: ANN401
+    """Yield a handle to the console input buffer, or None without a console.
+
+    ``CONIN$`` is opened explicitly because the standard input handle points
+    at a file or pipe when stdin is redirected. Only the errors Windows
+    reports for a missing console mean "no console"; any other failure raises.
+    """
+    handle = kernel32.CreateFileW(
+        "CONIN$",
+        GENERIC_READ | GENERIC_WRITE,
+        FILE_SHARE_READ | FILE_SHARE_WRITE,
+        None,
+        OPEN_EXISTING,
+        0,
+        None,
+    )
+    if handle in {None, INVALID_HANDLE_VALUE}:
+        if ctypes.get_last_error() in NO_CONSOLE_ERRORS:  # type: ignore[attr-defined]  # pyrefly: ignore[missing-attribute]
+            yield None
+            return
+        _raise_last_error("CreateFileW(CONIN$)")
+    try:
+        yield handle
+    finally:
+        if not kernel32.CloseHandle(handle):
+            # Cleanup only: raising here would mask the real outcome.
+            logger.warning(
+                "CloseHandle failed (error %s)",
+                ctypes.get_last_error(),  # type: ignore[attr-defined]  # pyrefly: ignore[missing-attribute]
+            )
+
+
+def _clear_quick_edit_flag() -> None:
+    kernel32 = _kernel32()
+    with _console_input(kernel32) as handle:
+        if handle is None:
+            return  # Not attached to a console (e.g. windowless).
+        mode = ctypes.c_uint32()
+        if not kernel32.GetConsoleMode(handle, ctypes.byref(mode)):
+            _raise_last_error("GetConsoleMode")
+        original_mode = mode.value
+        new_mode = (original_mode & ~ENABLE_QUICK_EDIT_MODE) | ENABLE_EXTENDED_FLAGS
+        if not kernel32.SetConsoleMode(handle, new_mode):
+            _raise_last_error("SetConsoleMode")
+    atexit.register(restore_console_mode, original_mode)
+
+
+def restore_console_mode(original_mode: int) -> None:
+    try:
+        kernel32 = _kernel32()
+        with _console_input(kernel32) as handle:
+            if handle is not None and not kernel32.SetConsoleMode(
+                handle, original_mode
+            ):
+                _raise_last_error("SetConsoleMode")
+    except OSError:
+        logger.warning("Could not restore console mode", exc_info=True)
+
+
 def main() -> None:
     file_handler = RotatingFileHandler(
         filename=Path.home() / ".croupier.log",
@@ -85,9 +218,11 @@ def main() -> None:
     file_handler.setFormatter(
         logging.Formatter("%(asctime)s %(name)s %(levelname)s %(message)s"),
     )
-    for name in ("faststream", "faststream.access.rabbit"):
+    for name in ("faststream", "faststream.access.rabbit", __name__):
         logging.getLogger(name).addHandler(file_handler)
 
+    disable_console_quick_edit()
+
     app = FastAPI(debug=True)
     app.include_router(router)
     app.add_middleware(

tests/test_main.py

@@ -5,9 +5,15 @@
 import pytest
 from faststream.rabbit import TestRabbitBroker
 
+from croupier.main import ENABLE_EXTENDED_FLAGS
+from croupier.main import ENABLE_QUICK_EDIT_MODE
+from croupier.main import ERROR_INVALID_HANDLE
+from croupier.main import INVALID_HANDLE_VALUE
 from croupier.main import Message
+from croupier.main import disable_console_quick_edit
 from croupier.main import handle_message
 from croupier.main import health
+from croupier.main import restore_console_mode
 from croupier.main import router
 from croupier.main import settings
 
@@ -88,3 +94,146 @@ async def test_returns_500_when_broker_is_unhealthy(self) -> None:
 
         response = await health(mock_request)
         assert response.status_code == 500
+
+
+class TestDisableConsoleQuickEdit:
+    """Tests for disable_console_quick_edit."""
+
+    @staticmethod
+    def _kernel32(*, handle: int | None = 1234, mode: int = 0x00F7) -> MagicMock:
+        kernel32 = MagicMock()
+        kernel32.CreateFileW.return_value = handle
+
+        def fake_get_console_mode(_handle: int, mode_ref: MagicMock) -> int:
+            mode_ref.value = mode
+            return 1
+
+        kernel32.GetConsoleMode.side_effect = fake_get_console_mode
+        kernel32.SetConsoleMode.return_value = 1
+        kernel32.CloseHandle.return_value = 1
+        return kernel32
+
+    def _run(
+        self, kernel32: MagicMock, *, last_error: int = 5
+    ) -> tuple[MagicMock, MagicMock]:
+        with (
+            patch("croupier.main.sys.platform", "win32"),
+            patch("croupier.main.ctypes") as mock_ctypes,
+            patch("croupier.main.atexit") as mock_atexit,
+            patch("croupier.main.logger") as mock_logger,
+        ):
+            mock_ctypes.WinDLL.return_value = kernel32
+            mock_ctypes.c_uint32.return_value = MagicMock()
+            mock_ctypes.byref.side_effect = lambda x: x
+            mock_ctypes.get_last_error.return_value = last_error
+            disable_console_quick_edit()
+        return mock_atexit, mock_logger
+
+    def test_is_noop_on_non_windows(self) -> None:
+        with (
+            patch("croupier.main.sys.platform", "linux"),
+            patch("croupier.main.ctypes") as mock_ctypes,
+        ):
+            disable_console_quick_edit()
+        mock_ctypes.WinDLL.assert_not_called()
+
+    def test_clears_quick_edit_flag_on_windows(self) -> None:
+        kernel32 = self._kernel32()
+        self._run(kernel32)
+
+        new_mode = kernel32.SetConsoleMode.call_args.args[1]
+        assert not new_mode & ENABLE_QUICK_EDIT_MODE
+        assert new_mode & ENABLE_EXTENDED_FLAGS
+
+    def test_opens_conin_instead_of_std_handle(self) -> None:
+        kernel32 = self._kernel32()
+        self._run(kernel32)
+
+        assert kernel32.CreateFileW.call_args.args[0] == "CONIN$"
+        kernel32.CloseHandle.assert_called_once_with(1234)
+
+    def test_registers_restore_of_original_mode(self) -> None:
+        kernel32 = self._kernel32(mode=0x00F7)
+        mock_atexit, _ = self._run(kernel32)
+
+        mock_atexit.register.assert_called_once_with(restore_console_mode, 0x00F7)
+
+    def test_restore_resets_original_mode(self) -> None:
+        kernel32 = self._kernel32()
+        with (
+            patch("croupier.main.ctypes") as mock_ctypes,
+        ):
+            mock_ctypes.WinDLL.return_value = kernel32
+            restore_console_mode(0x00F7)
+
+        assert kernel32.SetConsoleMode.call_args.args[1] == 0x00F7
+
+    def test_skips_when_no_console(self) -> None:
+        kernel32 = self._kernel32(handle=INVALID_HANDLE_VALUE)
+        mock_atexit, mock_logger = self._run(kernel32, last_error=ERROR_INVALID_HANDLE)
+
+        mock_logger.warning.assert_not_called()
+
+        kernel32.GetConsoleMode.assert_not_called()
+        kernel32.SetConsoleMode.assert_not_called()
+        kernel32.CloseHandle.assert_not_called()
+        mock_atexit.register.assert_not_called()
+
+    def test_logs_when_conin_open_fails_for_other_reason(self) -> None:
+        kernel32 = self._kernel32(handle=INVALID_HANDLE_VALUE)
+        mock_atexit, mock_logger = self._run(kernel32, last_error=5)
+
+        mock_logger.warning.assert_called_once()
+        kernel32.GetConsoleMode.assert_not_called()
+        mock_atexit.register.assert_not_called()
+
+    def test_logs_when_close_handle_fails(self) -> None:
+        kernel32 = self._kernel32()
+        kernel32.CloseHandle.return_value = 0
+        mock_atexit, mock_logger = self._run(kernel32)
+
+        mock_logger.warning.assert_called_once()
+        mock_atexit.register.assert_called_once()
+
+    def test_restore_logs_when_set_console_mode_fails(self) -> None:
+        kernel32 = self._kernel32()
+        kernel32.SetConsoleMode.return_value = 0
+        with (
+            patch("croupier.main.ctypes") as mock_ctypes,
+            patch("croupier.main.logger") as mock_logger,
+        ):
+            mock_ctypes.WinDLL.return_value = kernel32
+            mock_ctypes.get_last_error.return_value = 5
+            restore_console_mode(0x00F7)
+
+        mock_logger.warning.assert_called_once()
+
+    def test_logs_and_continues_when_win32_call_fails(self) -> None:
+        with (
+            patch("croupier.main.sys.platform", "win32"),
+            patch("croupier.main.ctypes") as mock_ctypes,
+            patch("croupier.main.logger") as mock_logger,
+        ):
+            mock_ctypes.WinDLL.side_effect = OSError("boom")
+            disable_console_quick_edit()
+        mock_logger.warning.assert_called_once()
+
+    def test_logs_when_get_console_mode_fails(self) -> None:
+        kernel32 = self._kernel32()
+        kernel32.GetConsoleMode.side_effect = None
+        kernel32.GetConsoleMode.return_value = 0
+        mock_atexit, mock_logger = self._run(kernel32)
+
+        mock_logger.warning.assert_called_once()
+        kernel32.SetConsoleMode.assert_not_called()
+        mock_atexit.register.assert_not_called()
+        kernel32.CloseHandle.assert_called_once_with(1234)
+
+    def test_logs_when_set_console_mode_returns_zero(self) -> None:
+        kernel32 = self._kernel32()
+        kernel32.SetConsoleMode.return_value = 0
+        mock_atexit, mock_logger = self._run(kernel32)
+
+        mock_logger.warning.assert_called_once()
+        mock_atexit.register.assert_not_called()
+        kernel32.CloseHandle.assert_called_once_with(1234)