crash_handler.py

"""
crash_handler.py — Hard-Crash auf unbehandelte Exceptions
=========================================================
sys/threading/asyncio-Hooks + faulthandler. ENV-Flag PX_HARD_CRASH=0
deaktiviert (default: 1, hard crash).

Run-local.sh teet stderr nach local_debug.log — Tracebacks landen persistent
und sind tail-bar. Ohne diese Hooks schlucken server.py + generators.py +
chat_tab.py die Exceptions an mehreren Stellen und verwandeln sie in stille
500/HTML/Partial-Responses (siehe Plan 7.1 für die exakte Architektur).

Hard Rule: KEIN motor-touch. Diese Hooks sitzen ausschließlich am
Process-Boundary — keine px_patches, keine generators.py-Logik.

Usage (in app.py):
    import crash_handler
    crash_handler.install()                      # sys + threading + faulthandler
    # ... uvicorn.Server statt uvicorn.run() ...
    crash_handler.install_asyncio(server.loop)   # asyncio
"""
import asyncio
import faulthandler
import os
import sys
import threading
import traceback


_BANNER = (
    "\n" + "=" * 72 + "\n"
    "[CRASH_HANDLER] Unhandled exception — terminating process\n"
    + "=" * 72 + "\n"
)


def _log_and_die(prefix, exc_type, exc_value, exc_tb):
    """Print full traceback to stderr, then os._exit(1).

    os._exit (nicht sys.exit) umgeht alle atexit/shutdown-Handler, damit
    ein Bug nicht durch partielle Cleanup-Logik verschleiert wird.
    """
    sys.stderr.write(_BANNER)
    sys.stderr.write(f"[CRASH_HANDLER] hook={prefix}\n")
    traceback.print_exception(exc_type, exc_value, exc_tb, file=sys.stderr)
    sys.stderr.flush()
    os._exit(1)


def _sys_hook(exc_type, exc_value, exc_tb):
    """sys.excepthook — Main-Thread unbehandelte Exceptions."""
    # KeyboardInterrupt soll weiterhin graceful Ctrl-C bleiben (nicht crashen).
    if issubclass(exc_type, KeyboardInterrupt):
        return sys.__excepthook__(exc_type, exc_value, exc_tb)
    _log_and_die("sys.excepthook", exc_type, exc_value, exc_tb)


def _threading_hook(args):
    """threading.excepthook — Unhandled exceptions in Thread(target=...).

    Fängt z.B. generators.py:554, 604, 723 + chat_tab.py:496 (alle unguarded,
    Python-Thread-Semantik würde sie ohne unseren Hook einfach verschlucken).
    """
    _log_and_die(
        f"threading.excepthook thread={args.thread.name!r}",
        args.exc_type, args.exc_value, args.exc_traceback,
    )


def _asyncio_hook(loop, context):
    """asyncio loop.set_exception_handler — uvicorn-Task-Fehler."""
    exc = context.get("exception")
    if exc is None:
        # Non-exception async errors (z.B. "Task was destroyed but it is
        # pending"). Trotzdem fatal — meist Logic-Bug.
        sys.stderr.write(_BANNER)
        sys.stderr.write(f"[CRASH_HANDLER] asyncio non-exc: {context}\n")
        sys.stderr.flush()
        os._exit(1)
    _log_and_die("asyncio", type(exc), exc, exc.__traceback__)


def install():
    """Install sys + threading + faulthandler hooks. Idempotent.

    Kill-Switch via ENV: PX_HARD_CRASH=0 macht install() zum no-op.
    """
    if os.environ.get("PX_HARD_CRASH", "1") == "0":
        return
    sys.excepthook = _sys_hook
    threading.excepthook = _threading_hook
    faulthandler.enable()  # SIGSEGV / native C++ crashes → stderr


def install_asyncio(loop):
    """Attach asyncio exception handler to uvicorn's event loop.

    MUSS nach Loop-Creation aufgerufen werden — uvicorn erstellt den Loop
    intern in Server.run(), daher der uvicorn.Server.startup()-Patch in
    app.py statt direkter Aufruf vor uvicorn.run().
    """
    if os.environ.get("PX_HARD_CRASH", "1") == "0":
        return
    loop.set_exception_handler(_asyncio_hook)