refactor(telemetry): keep only the unraisable hook, no docstrings

strix/telemetry/logging.py

@@ -6,7 +6,6 @@ import contextlib
 import logging
 import os
 import sys
-import threading
 import warnings
 from contextvars import ContextVar
 from pathlib import Path  # noqa: TC003  used at runtime by ``setup_scan_logging``
@@ -89,34 +88,20 @@ def configure_dependency_logging() -> None:
     logging.getLogger("asyncio").setLevel(logging.CRITICAL)
     logging.getLogger("asyncio").propagate = False
     warnings.filterwarnings("ignore", category=RuntimeWarning, module="asyncio")
-    _route_hook_exceptions_to_log()
+    _route_unraisable_to_log()
 
 
-def _route_hook_exceptions_to_log() -> None:
-    """Python prints finalizer and thread tracebacks to stderr; send them to strix.log instead.
-
-    Dropped when the strix loggers have no handler (after scan teardown)
-    rather than left to ``logging.lastResort``, and any failure inside the
-    hook is swallowed: the interpreter may be shutting down.
-    """
-
-    def log(message: str, exc_info: tuple[object, object, object]) -> None:
+def _route_unraisable_to_log() -> None:
+    def hook(u: sys.UnraisableHookArgs) -> None:
         with contextlib.suppress(BaseException):
             logger = logging.getLogger("strix.telemetry")
             if logger.hasHandlers():
-                logger.warning(message, exc_info=exc_info)  # type: ignore[arg-type]
+                logger.warning(
+                    u.err_msg or f"Exception ignored in {u.object!r}",
+                    exc_info=(u.exc_type, u.exc_value, u.exc_traceback),  # type: ignore[arg-type]
+                )
 
-    def unraisable_hook(u: sys.UnraisableHookArgs) -> None:
-        log(
-            u.err_msg or f"Exception ignored in {u.object!r}",
-            (u.exc_type, u.exc_value, u.exc_traceback),
-        )
-
-    def thread_hook(a: threading.ExceptHookArgs) -> None:
-        log(f"Exception in thread {a.thread}", (a.exc_type, a.exc_value, a.exc_traceback))
-
-    sys.unraisablehook = unraisable_hook
-    threading.excepthook = thread_hook
+    sys.unraisablehook = hook
 
 
 class _CurrentStderrHandler(logging.StreamHandler):  # type: ignore[type-arg]

tests/test_unraisable_hook.py

@@ -1,18 +1,9 @@
-"""Nothing from sys.unraisablehook or threading.excepthook reaches the terminal.
-
-Python prints "Exception ignored in ..." reports (finalizers, ``__del__``, GC
-and weakref callbacks, files closed at interpreter exit) and uncaught thread
-exceptions straight to stderr. Strix routes both to its own logger instead,
-so they end up in strix.log (and on stderr only under STRIX_DEBUG=1).
-"""
-
 from __future__ import annotations
 
 import logging
 import os
 import subprocess
 import sys
-import threading
 from pathlib import Path
 from types import SimpleNamespace
 from typing import TYPE_CHECKING
@@ -54,14 +45,11 @@ def strix_records() -> Iterator[list[logging.LogRecord]]:
 
 
 @pytest.fixture
-def hooks(monkeypatch: pytest.MonkeyPatch) -> tuple[list[object], list[object]]:
-    """Install strix's hooks over recording stand-ins for Python's defaults."""
-    unraisable_calls: list[object] = []
-    thread_calls: list[object] = []
-    monkeypatch.setattr(sys, "unraisablehook", unraisable_calls.append)
-    monkeypatch.setattr(threading, "excepthook", thread_calls.append)
+def default_hook_calls(monkeypatch: pytest.MonkeyPatch) -> list[object]:
+    calls: list[object] = []
+    monkeypatch.setattr(sys, "unraisablehook", calls.append)
     tlog.configure_dependency_logging()
-    return unraisable_calls, thread_calls
+    return calls
 
 
 @pytest.mark.parametrize(
@@ -82,15 +70,13 @@ def hooks(monkeypatch: pytest.MonkeyPatch) -> tuple[list[object], list[object]]:
 )
 def test_every_unraisable_is_logged_not_printed(
     args: object,
-    hooks: tuple[list[object], list[object]],
+    default_hook_calls: list[object],
     strix_records: list[logging.LogRecord],
     capsys: pytest.CaptureFixture[str],
 ) -> None:
-    unraisable_calls, _ = hooks
-
     sys.unraisablehook(args)  # type: ignore[arg-type]
 
-    assert unraisable_calls == []
+    assert default_hook_calls == []
     assert capsys.readouterr().err == ""
     [record] = strix_records
     assert record.levelno == logging.WARNING
@@ -103,12 +89,10 @@ def test_every_unraisable_is_logged_not_printed(
     assert str(args.exc_value) in message  # type: ignore[attr-defined]
 
 
-@pytest.mark.usefixtures("hooks")
+@pytest.mark.usefixtures("default_hook_calls")
 def test_unraisable_is_dropped_when_strix_has_no_handlers(
     capsys: pytest.CaptureFixture[str],
 ) -> None:
-    """After scan teardown the strix logger tree has no handlers; the report
-    must be dropped rather than handed to logging.lastResort (stderr)."""
     root = logging.getLogger("strix")
     saved, root.handlers = root.handlers, []
     try:
@@ -118,31 +102,8 @@ def test_unraisable_is_dropped_when_strix_has_no_handlers(
     assert capsys.readouterr().err == ""
 
 
-def test_thread_exceptions_are_logged_not_printed(
-    hooks: tuple[list[object], list[object]],
-    strix_records: list[logging.LogRecord],
-    capsys: pytest.CaptureFixture[str],
-) -> None:
-    _, thread_calls = hooks
-
-    def _boom() -> None:
-        raise RuntimeError("worker failed")
-
-    thread = threading.Thread(target=_boom, name="strix-worker")
-    thread.start()
-    thread.join()
-
-    assert thread_calls == []
-    assert capsys.readouterr().err == ""
-    [record] = strix_records
-    assert record.levelno == logging.WARNING
-    message = logging.Formatter().format(record)
-    assert "Exception in thread <Thread(strix-worker" in message
-    assert "RuntimeError: worker failed" in message
-
-
 _EXIT_SCRIPT = r"""
-import logging, sys, threading
+import io, sys
 from strix.telemetry.logging import setup_console_logging
 setup_console_logging()
 
@@ -150,40 +111,30 @@ class Leaky:
     def __del__(self):
         raise ValueError("I/O operation on closed file.")
 
-Leaky()                      # collected right away
-keep = Leaky()               # collected at interpreter shutdown
+Leaky()
+keep = Leaky()
 
-def worker():
-    raise RuntimeError("worker failed")
-t = threading.Thread(target=worker); t.start(); t.join()
-
-import urllib3.response, io
+import urllib3.response
 resp = urllib3.response.HTTPResponse(body=io.BytesIO(b""), preload_content=False)
-resp._fp.close()             # socket file closed before the response: the 3.14 exit noise
+resp._fp.close()
 del resp
 print("done")
 """
 
 
 def test_process_exit_prints_nothing_to_stderr() -> None:
-    """End to end in a fresh interpreter: finalizer errors (including one at
-    interpreter shutdown), a thread traceback and the closed-file response
-    case produce no stderr output, with the default (non-debug) handlers."""
-    env = {"PYTHONPATH": str(Path(__file__).resolve().parents[1])}
+    env = {
+        k: v for k, v in os.environ.items() if k in {"PATH", "HOME", "SYSTEMROOT", "TEMP", "TMP"}
+    }
+    env |= {"PYTHONPATH": str(Path(__file__).resolve().parents[1]), "STRIX_DEBUG": ""}
     proc = subprocess.run(  # noqa: S603
         [sys.executable, "-c", _EXIT_SCRIPT],
         capture_output=True,
         text=True,
         timeout=60,
-        env={**_safe_env(), **env, "STRIX_DEBUG": ""},
+        env=env,
         check=False,
     )
     assert proc.returncode == 0, proc.stderr
     assert proc.stdout.strip() == "done"
     assert proc.stderr == ""
-
-
-def _safe_env() -> dict[str, str]:
-    return {
-        k: v for k, v in os.environ.items() if k in {"PATH", "HOME", "SYSTEMROOT", "TEMP", "TMP"}
-    }