From eaf16c681752764c73792bad9d0278bdef0e484f Mon Sep 17 00:00:00 2001 From: Ahmed Allam Date: Sun, 4 Oct 2026 15:17:41 +0000 Subject: [PATCH] refactor(telemetry): collapse the exception hooks into one plain function No install-once global, no hand-rolled traceback formatting, no SystemExit special case: one helper logs the report with exc_info on strix.telemetry when a handler exists, and both hooks call it. --- strix/telemetry/logging.py | 76 ++++++---------------------- tests/test_hook_exceptions_logged.py | 23 ++------- 2 files changed, 19 insertions(+), 80 deletions(-) diff --git a/strix/telemetry/logging.py b/strix/telemetry/logging.py index 0e4dc7f2..5c152cc9 100644 --- a/strix/telemetry/logging.py +++ b/strix/telemetry/logging.py @@ -7,7 +7,6 @@ import logging import os import sys import threading -import traceback import warnings from contextvars import ContextVar from pathlib import Path # noqa: TC003 used at runtime by ``setup_scan_logging`` @@ -16,7 +15,6 @@ from typing import TYPE_CHECKING if TYPE_CHECKING: from collections.abc import Callable - from types import TracebackType _SCAN_ID: ContextVar[str | None] = ContextVar("strix_scan_id", default=None) @@ -94,70 +92,28 @@ def configure_dependency_logging() -> None: _route_hook_exceptions_to_log() -_hooks_installed = False - - -def _format_exception( - exc_type: type[BaseException] | None, - exc_value: BaseException | None, - exc_traceback: TracebackType | None, -) -> str: - if exc_value is None and exc_type is None: - return "-" - return "".join(traceback.format_exception(exc_type, exc_value, exc_traceback)).rstrip() - - -def _log_quietly(message: str, *args: object) -> None: - """Write a WARNING to the strix log; never let it reach the terminal. - - Dropped (not printed) when the strix logger tree has no handler, which - is the case after scan teardown: ``logging.lastResort`` would otherwise - print it to stderr. The interpreter may be finalizing, so any failure - inside logging itself is swallowed too. - """ - with contextlib.suppress(BaseException): - logger = logging.getLogger("strix.telemetry") - if logger.hasHandlers(): - logger.warning(message, *args) - - def _route_hook_exceptions_to_log() -> None: - """Keep "Exception ignored in ..." reports and thread tracebacks off the terminal. + """Python prints finalizer and thread tracebacks to stderr; send them to strix.log instead. - Python's default ``sys.unraisablehook`` (finalizers, ``__del__``, GC and - weakref callbacks, buffered-file close at exit) and - ``threading.excepthook`` (uncaught exceptions in threads) print a - traceback to stderr, which lands in the terminal after the scan summary. - Both are replaced, for the life of the process, with hooks that log the - report at WARNING on ``strix.telemetry`` instead: it goes to strix.log, - and to stderr only when the stream handler runs at DEBUG (STRIX_DEBUG=1). + 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. """ - global _hooks_installed # noqa: PLW0603 - if _hooks_installed: - return - _hooks_installed = True - def unraisable_hook(unraisable: sys.UnraisableHookArgs) -> None: + def log(message: str, exc_info: tuple[object, object, object]) -> None: with contextlib.suppress(BaseException): - subject = unraisable.err_msg or f"Exception ignored in {unraisable.object!r}" - _log_quietly( - "%s\n%s", - subject, - _format_exception( - unraisable.exc_type, unraisable.exc_value, unraisable.exc_traceback - ), - ) + logger = logging.getLogger("strix.telemetry") + if logger.hasHandlers(): + logger.warning(message, exc_info=exc_info) # type: ignore[arg-type] - def thread_hook(args: threading.ExceptHookArgs) -> None: - if args.exc_type is SystemExit: - return - with contextlib.suppress(BaseException): - name = args.thread.name if args.thread is not None else "-" - _log_quietly( - "Exception in thread %s\n%s", - name, - _format_exception(args.exc_type, args.exc_value, args.exc_traceback), - ) + 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 diff --git a/tests/test_hook_exceptions_logged.py b/tests/test_hook_exceptions_logged.py index 06610451..41fbcd32 100644 --- a/tests/test_hook_exceptions_logged.py +++ b/tests/test_hook_exceptions_logged.py @@ -60,7 +60,6 @@ def hooks(monkeypatch: pytest.MonkeyPatch) -> tuple[list[object], list[object]]: thread_calls: list[object] = [] monkeypatch.setattr(sys, "unraisablehook", unraisable_calls.append) monkeypatch.setattr(threading, "excepthook", thread_calls.append) - monkeypatch.setattr(tlog, "_hooks_installed", False) tlog.configure_dependency_logging() return unraisable_calls, thread_calls @@ -96,7 +95,7 @@ def test_every_unraisable_is_logged_not_printed( [record] = strix_records assert record.levelno == logging.WARNING assert record.name == "strix.telemetry" - message = record.getMessage() + message = logging.Formatter().format(record) if args.err_msg: # type: ignore[attr-defined] assert message.startswith(args.err_msg) # type: ignore[attr-defined] else: @@ -137,27 +136,11 @@ def test_thread_exceptions_are_logged_not_printed( assert capsys.readouterr().err == "" [record] = strix_records assert record.levelno == logging.WARNING - message = record.getMessage() - assert message.startswith("Exception in thread strix-worker") + message = logging.Formatter().format(record) + assert "Exception in thread None: - thread = threading.Thread(target=sys.exit, args=(3,)) - thread.start() - thread.join() - assert strix_records == [] - - -def test_hooks_install_once(monkeypatch: pytest.MonkeyPatch) -> None: - monkeypatch.setattr(tlog, "_hooks_installed", False) - tlog.configure_dependency_logging() - installed = (sys.unraisablehook, threading.excepthook) - tlog.configure_dependency_logging() - assert (sys.unraisablehook, threading.excepthook) == installed - - _EXIT_SCRIPT = r""" import logging, sys, threading from strix.telemetry.logging import setup_console_logging