fix(cli): suppress OSError EIO on interrupt shutdown

When the user interrupts a long-running task, prompt_toolkit tries to
flush stdout during emergency shutdown.  If stdout is in a broken state
(redirected to /dev/null, pipe closed, terminal gone), the flush raises
`OSError: [Errno 5] Input/output error` which propagates unhandled and
crashes the CLI.

Two defense layers:

1. `_suppress_closed_loop_errors`: add `OSError` with `errno.EIO` to
   the asyncio exception handler, matching the existing pattern for
   `RuntimeError("Event loop is closed")` and `KeyError("is not
   registered")`.

2. Outer `except (KeyError, OSError)` block: add `errno.EIO` check
   before the existing string-match guards, silently suppressing the
   error instead of printing a misleading stdin-related message.

Fixes #13710.
This commit is contained in:
voidborne-d 2026-04-21 23:08:46 +00:00 committed by Teknium
parent 4d170134ef
commit 45e1228a8a
2 changed files with 123 additions and 3 deletions

11
cli.py
View file

@ -22,6 +22,7 @@ import re
import concurrent.futures
import base64
import atexit
import errno
import tempfile
import time
import uuid
@ -10729,6 +10730,8 @@ class HermesCLI:
return # silently suppress
if isinstance(exc, KeyError) and "is not registered" in str(exc):
return # suppress selector registration failures (#6393)
if isinstance(exc, OSError) and getattr(exc, "errno", None) == errno.EIO:
return # suppress I/O errors from broken stdout on interrupt (#13710)
# Fall back to default handler for everything else
loop.default_exception_handler(context)
@ -10761,9 +10764,11 @@ class HermesCLI:
except (EOFError, KeyboardInterrupt, BrokenPipeError):
pass
except (KeyError, OSError) as _stdin_err:
# Catch selector registration failures from broken stdin (#6393).
# This is the fallback for cases that slip past the fstat() guard.
if "is not registered" in str(_stdin_err) or "Bad file descriptor" in str(_stdin_err):
# Catch selector registration failures from broken stdin (#6393)
# and I/O errors from broken stdout during interrupt (#13710).
if isinstance(_stdin_err, OSError) and getattr(_stdin_err, "errno", None) == errno.EIO:
pass # suppress broken-stdout I/O errors on interrupt (#13710)
elif "is not registered" in str(_stdin_err) or "Bad file descriptor" in str(_stdin_err):
print(
f"\nError: stdin is not usable ({_stdin_err}).\n"
"This can happen with certain Python installations (e.g. uv-managed cPython on macOS).\n"

View file

@ -0,0 +1,115 @@
"""Tests for OSError EIO suppression during interrupt shutdown (#13710).
When the user interrupts a running task, prompt_toolkit tries to flush
stdout during emergency shutdown. If stdout is already in a broken state
(redirected to /dev/null, pipe closed, etc.), the flush raises
``OSError: [Errno 5] Input/output error``.
The ``_suppress_closed_loop_errors`` asyncio exception handler and the
outer ``except (KeyError, OSError)`` block must both suppress this error
to prevent a hard crash.
"""
from __future__ import annotations
import errno
import os
from unittest.mock import MagicMock
import pytest
# ---------------------------------------------------------------------------
# _suppress_closed_loop_errors asyncio exception handler
# ---------------------------------------------------------------------------
def _make_suppress_fn():
"""Build a standalone copy of ``_suppress_closed_loop_errors``.
The real function is defined as a closure inside
``CLI._run_interactive``; we reconstruct an equivalent here so the
unit tests don't need a full CLI instance.
"""
def _suppress_closed_loop_errors(loop, context):
exc = context.get("exception")
if isinstance(exc, RuntimeError) and "Event loop is closed" in str(exc):
return
if isinstance(exc, KeyError) and "is not registered" in str(exc):
return
if isinstance(exc, OSError) and getattr(exc, "errno", None) == errno.EIO:
return
loop.default_exception_handler(context)
return _suppress_closed_loop_errors
class TestSuppressClosedLoopErrors:
"""Verify the asyncio exception handler suppresses expected errors."""
def test_suppresses_event_loop_closed(self):
handler = _make_suppress_fn()
loop = MagicMock()
handler(loop, {"exception": RuntimeError("Event loop is closed")})
loop.default_exception_handler.assert_not_called()
def test_suppresses_key_not_registered(self):
handler = _make_suppress_fn()
loop = MagicMock()
handler(loop, {"exception": KeyError("0 is not registered")})
loop.default_exception_handler.assert_not_called()
def test_suppresses_oserror_eio(self):
"""OSError with errno.EIO must be suppressed (#13710)."""
handler = _make_suppress_fn()
loop = MagicMock()
exc = OSError(errno.EIO, "Input/output error")
handler(loop, {"exception": exc})
loop.default_exception_handler.assert_not_called()
def test_does_not_suppress_oserror_other_errno(self):
"""OSError with a different errno must still propagate."""
handler = _make_suppress_fn()
loop = MagicMock()
exc = OSError(errno.EACCES, "Permission denied")
handler(loop, {"exception": exc})
loop.default_exception_handler.assert_called_once()
def test_does_not_suppress_unrelated_exception(self):
"""Unrelated exceptions must still propagate."""
handler = _make_suppress_fn()
loop = MagicMock()
handler(loop, {"exception": ValueError("something else")})
loop.default_exception_handler.assert_called_once()
def test_no_exception_key(self):
"""Context without 'exception' must propagate to default handler."""
handler = _make_suppress_fn()
loop = MagicMock()
handler(loop, {"message": "some log"})
loop.default_exception_handler.assert_called_once()
# ---------------------------------------------------------------------------
# Outer except block EIO handling
# ---------------------------------------------------------------------------
class TestOuterExceptEIO:
"""Verify the outer ``except (KeyError, OSError)`` block logic."""
def test_eio_does_not_reraise(self):
"""OSError with errno.EIO should be silently suppressed."""
exc = OSError(errno.EIO, "Input/output error")
# Simulate the condition check from the outer except block:
assert isinstance(exc, OSError)
assert getattr(exc, "errno", None) == errno.EIO
def test_bad_file_descriptor_matches(self):
"""'Bad file descriptor' string should be caught."""
exc = OSError(errno.EBADF, "Bad file descriptor")
assert "Bad file descriptor" in str(exc)
def test_other_oserror_reraises(self):
"""Other OSError variants must not match the EIO guard."""
exc = OSError(errno.EACCES, "Permission denied")
assert not (getattr(exc, "errno", None) == errno.EIO)
assert "is not registered" not in str(exc)
assert "Bad file descriptor" not in str(exc)