diff --git a/server/digi_server/logger.py b/server/digi_server/logger.py
index 3c076c5c..23c71afa 100644
--- a/server/digi_server/logger.py
+++ b/server/digi_server/logger.py
@@ -1,7 +1,11 @@
 import logging
+import sys
+import threading
+import weakref
 from logging.handlers import RotatingFileHandler
 from typing import Optional
 
+from prometheus_client import Counter
 from tornado.log import LogFormatter
 
 from utils.log_buffer import (
@@ -16,9 +20,182 @@
     logging.getLogger("tornado.access"),
     logging.getLogger("tornado.application"),
     logging.getLogger("tornado.general"),
+    # Stdlib/third-party loggers whose output would otherwise only reach stderr
+    # (e.g. "Exception in callback ..." from fire-and-forget loop callbacks).
+    logging.getLogger("asyncio"),
+    logging.getLogger("alembic"),
+    logging.getLogger("py.warnings"),
     logger,
 ]
 
+UNHANDLED_ERRORS = Counter(
+    "digiscript_unhandled_errors_total",
+    "Unhandled exceptions caught by the global crash hooks",
+    ["source"],
+)
+UNHANDLED_ERROR_SOURCES = ("sys_excepthook", "asyncio", "thread")
+# Pre-register every label series at 0 so Prometheus increase() sees the very
+# first error for each source.
+for _source in UNHANDLED_ERROR_SOURCES:
+    UNHANDLED_ERRORS.labels(source=_source)
+
+# Guards install/uninstall; the state below is only touched while holding it.
+_hooks_lock = threading.RLock()
+# Previous hooks / warning state, saved at install; empty while not installed.
+_saved_hooks: dict = {}
+# Event loops we attached a handler to, mapped to the handler they had before.
+_saved_loop_handlers: "weakref.WeakKeyDictionary" = weakref.WeakKeyDictionary()
+
+
+def _count_unhandled(source: str) -> None:
+    UNHANDLED_ERRORS.labels(source=source).inc()
+
+
+def _chain(previous, default, *args) -> None:
+    """Call a pre-existing hook, skipping the interpreter default.
+
+    The defaults only print to stderr, which would duplicate the log record.
+    """
+    if previous is None or previous is default:
+        return
+    try:
+        previous(*args)
+    except Exception:  # noqa: BLE001 - a broken foreign hook must not mask the crash
+        logger.exception("Chained exception hook failed")
+
+
+def _sys_excepthook(exc_type, exc_value, exc_tb):
+    previous = _saved_hooks.get("sys")
+    if issubclass(exc_type, (KeyboardInterrupt, SystemExit)):
+        (previous or sys.__excepthook__)(exc_type, exc_value, exc_tb)
+        return
+    _count_unhandled("sys_excepthook")
+    try:
+        logger.critical("Uncaught exception", exc_info=(exc_type, exc_value, exc_tb))
+    finally:
+        _chain(previous, sys.__excepthook__, exc_type, exc_value, exc_tb)
+
+
+def _threading_excepthook(args):
+    previous = _saved_hooks.get("thread")
+    if issubclass(args.exc_type, SystemExit):
+        _chain(previous, threading.__excepthook__, args)
+        return
+    _count_unhandled("thread")
+    try:
+        thread_name = args.thread.name if args.thread else "<unknown>"
+        logger.error(
+            "Unhandled exception in thread %s",
+            thread_name,
+            exc_info=(args.exc_type, args.exc_value, args.exc_traceback),
+        )
+    finally:
+        _chain(previous, threading.__excepthook__, args)
+
+
+def _safe_repr(value) -> str:
+    try:
+        return repr(value)
+    except Exception:  # noqa: BLE001 - arbitrary __repr__ must not break logging
+        return f"<unrepresentable {type(value).__name__}>"
+
+
+def _asyncio_exception_handler(loop, context):
+    exc = context.get("exception")
+    previous = _saved_loop_handlers.get(loop)
+    if isinstance(exc, (KeyboardInterrupt, SystemExit)):
+        if previous is not None:
+            previous(loop, context)
+        else:
+            loop.default_exception_handler(context)
+        return
+    try:
+        message = context.get("message", "Unhandled exception in event loop")
+        extras = "".join(
+            f"\n{key}: {_safe_repr(context[key])}"
+            for key in ("future", "task", "handle", "protocol", "transport")
+            if key in context
+        )
+        if exc is None:
+            # Message-only context (e.g. slow callback): a diagnostic, not a crash.
+            logger.warning("%s%s", message, extras)
+        else:
+            _count_unhandled("asyncio")
+            logger.error(
+                "%s%s",
+                message,
+                extras,
+                exc_info=(type(exc), exc, exc.__traceback__),
+            )
+    except Exception:  # noqa: BLE001 - the handler itself must never raise
+        try:
+            loop.default_exception_handler(context)
+        except Exception:  # noqa: BLE001, S110
+            pass
+    if previous is not None:
+        _chain(previous, None, loop, context)
+
+
+def install_exception_hooks(loop=None) -> None:
+    """Route otherwise-invisible crashes into the DigiScript log.
+
+    Installs ``sys.excepthook`` and ``threading.excepthook`` and the given
+    loop's exception handler, and enables ``logging.captureWarnings`` so
+    ``warnings.warn`` reaches the ``py.warnings`` logger. Each unhandled error
+    increments ``digiscript_unhandled_errors_total{source}``. Pre-existing hooks
+    are chained to after logging (interpreter defaults are skipped to avoid
+    duplicate stderr output); ``KeyboardInterrupt`` and ``SystemExit`` keep
+    their default behaviour. Idempotent and thread-safe.
+
+    :param loop: Event loop to attach the exception handler to, or ``None`` to
+        skip the loop handler.
+    """
+    with _hooks_lock:
+        if not _saved_hooks:
+            _saved_hooks["warnings"] = logging._warnings_showwarning is not None
+            if sys.excepthook is not _sys_excepthook:
+                _saved_hooks["sys"] = sys.excepthook
+            else:
+                _saved_hooks["sys"] = sys.__excepthook__
+            if threading.excepthook is not _threading_excepthook:
+                _saved_hooks["thread"] = threading.excepthook
+            else:
+                _saved_hooks["thread"] = threading.__excepthook__
+            sys.excepthook = _sys_excepthook
+            threading.excepthook = _threading_excepthook
+            logging.captureWarnings(True)
+        if loop is not None and loop not in _saved_loop_handlers:
+            current = loop.get_exception_handler()
+            if current is not _asyncio_exception_handler:
+                _saved_loop_handlers[loop] = current
+                loop.set_exception_handler(_asyncio_exception_handler)
+
+
+def uninstall_exception_hooks() -> None:
+    """Undo :func:`install_exception_hooks`.
+
+    Restores ``sys.excepthook``, ``threading.excepthook``, the previous
+    handler of every loop we attached to (unless closed) and the prior
+    ``captureWarnings`` state. Idempotent and thread-safe.
+    """
+    with _hooks_lock:
+        if not _saved_hooks:
+            return
+        # Only restore hooks that are still ours; never clobber a wrapper
+        # installed after us.
+        if sys.excepthook is _sys_excepthook:
+            sys.excepthook = _saved_hooks["sys"]
+        if threading.excepthook is _threading_excepthook:
+            threading.excepthook = _saved_hooks["thread"]
+        logging.captureWarnings(_saved_hooks["warnings"])
+        for loop, previous in _saved_loop_handlers.items():
+            if not loop.is_closed() and (
+                loop.get_exception_handler() is _asyncio_exception_handler
+            ):
+                loop.set_exception_handler(previous)
+        _saved_loop_handlers.clear()
+        _saved_hooks.clear()
+
 
 def get_logger(name: Optional[str] = None):
     if name:
diff --git a/server/main.py b/server/main.py
index ccbf5b02..5b3fffb4 100755
--- a/server/main.py
+++ b/server/main.py
@@ -32,7 +32,11 @@ def ensure_writable_db_path(path):
 
 
 from digi_server.app_server import DigiScriptServer
-from digi_server.logger import add_logging_level, get_logger
+from digi_server.logger import (
+    add_logging_level,
+    get_logger,
+    install_exception_hooks,
+)
 
 
 add_logging_level("TRACE", logging.DEBUG - 5)
@@ -51,6 +55,12 @@ def ensure_writable_db_path(path):
 
 
 async def main():
+    # Installed first so crashes during startup are counted and reach stderr
+    # (logging's last-resort handler). The durable file log and in-app buffer
+    # handlers are attached later by DigiScriptServer.configure(), so errors in
+    # this early window are not written to the log file.
+    install_exception_hooks(asyncio.get_running_loop())
+
     # Parse command-line options
     try:
         parse_command_line()
diff --git a/server/test/digi_server/test_logger.py b/server/test/digi_server/test_logger.py
index 617be374..1462a27a 100644
--- a/server/test/digi_server/test_logger.py
+++ b/server/test/digi_server/test_logger.py
@@ -1,7 +1,26 @@
+import asyncio
+import gc
 import logging
+import sys
+import threading
+import time
 import unittest
+import warnings
+from unittest import mock
 
-from digi_server.logger import CLIENT_LEVEL_MAP, map_client_level
+from tornado.ioloop import IOLoop
+
+import digi_server.logger as logger_module
+from digi_server.logger import (
+    ALL_LOGGERS,
+    CLIENT_LEVEL_MAP,
+    UNHANDLED_ERROR_SOURCES,
+    UNHANDLED_ERRORS,
+    install_exception_hooks,
+    logger,
+    map_client_level,
+    uninstall_exception_hooks,
+)
 
 
 class TestMapClientLevel(unittest.TestCase):
@@ -38,3 +57,330 @@ def test_case_insensitive(self):
     def test_client_level_map_contains_all_expected_keys(self):
         expected = {"TRACE", "DEBUG", "INFO", "WARN", "ERROR", "SILENT"}
         self.assertEqual(expected, set(CLIENT_LEVEL_MAP.keys()))
+
+
+class _ListHandler(logging.Handler):
+    def __init__(self):
+        super().__init__(logging.DEBUG)
+        self.records = []
+
+    def emit(self, record):
+        self.records.append(record)
+
+
+class TestExceptionHooks(unittest.IsolatedAsyncioTestCase):
+    def setUp(self):
+        self._excepthook = sys.excepthook
+        self._thread_hook = threading.excepthook
+        self._warnings_captured = logging._warnings_showwarning is not None
+        self._levels = {lg.name: lg.level for lg in ALL_LOGGERS}
+        self.handler = _ListHandler()
+        for lg in ALL_LOGGERS:
+            lg.addHandler(self.handler)
+        logger.setLevel(logging.DEBUG)
+        uninstall_exception_hooks()
+
+    def tearDown(self):
+        uninstall_exception_hooks()
+        sys.excepthook = self._excepthook
+        threading.excepthook = self._thread_hook
+        logging.captureWarnings(self._warnings_captured)
+        for lg in ALL_LOGGERS:
+            lg.removeHandler(self.handler)
+            lg.setLevel(self._levels[lg.name])
+
+    @staticmethod
+    def _count(source):
+        return UNHANDLED_ERRORS.labels(source=source)._value.get()
+
+    def test_label_series_preregistered_at_zero(self):
+        samples = {
+            s.labels["source"]: s.value
+            for m in UNHANDLED_ERRORS.collect()
+            for s in m.samples
+            if s.name == "digiscript_unhandled_errors_total"
+        }
+        self.assertEqual(
+            {"sys_excepthook", "asyncio", "thread"}, set(UNHANDLED_ERROR_SOURCES)
+        )
+        for source in UNHANDLED_ERROR_SOURCES:
+            self.assertIn(source, samples)
+
+    def test_all_loggers_cover_stdlib_sources(self):
+        names = {lg.name for lg in ALL_LOGGERS}
+        self.assertLessEqual({"asyncio", "alembic", "py.warnings", "DigiScript"}, names)
+        self.assertNotIn("sqlalchemy.engine", names)
+        self.assertNotIn("root", names)
+
+    def test_install_is_idempotent_and_restores_originals(self):
+        install_exception_hooks()
+        first = sys.excepthook
+        install_exception_hooks()
+        self.assertIs(first, sys.excepthook)
+        self.assertIsNot(self._excepthook, sys.excepthook)
+        uninstall_exception_hooks()
+        self.assertIs(self._excepthook, sys.excepthook)
+        self.assertIs(self._thread_hook, threading.excepthook)
+
+    def test_sys_excepthook_logs_critical_and_counts(self):
+        install_exception_hooks()
+        before = self._count("sys_excepthook")
+        try:
+            raise ValueError("boom-sys")
+        except ValueError:
+            sys.excepthook(*sys.exc_info())
+        self.assertEqual(before + 1, self._count("sys_excepthook"))
+        records = [r for r in self.handler.records if r.name == "DigiScript"]
+        self.assertEqual(1, len(records))
+        self.assertEqual(logging.CRITICAL, records[0].levelno)
+        self.assertIn("boom-sys", repr(records[0].exc_info[1]))
+
+    def test_sys_excepthook_keeps_default_for_keyboard_interrupt(self):
+        calls = []
+        sys.excepthook = lambda *a: calls.append(a)
+        install_exception_hooks()
+        before = self._count("sys_excepthook")
+        sys.excepthook(KeyboardInterrupt, KeyboardInterrupt(), None)
+        self.assertEqual(1, len(calls))
+        self.assertEqual(before, self._count("sys_excepthook"))
+        self.assertEqual([], self.handler.records)
+
+    def test_thread_excepthook_logs_and_counts(self):
+        # Replace pytest's own thread hook so it doesn't flag the intended crash.
+        threading.excepthook = lambda _args: None
+        install_exception_hooks()
+        before = self._count("thread")
+
+        def target():
+            raise RuntimeError("boom-thread")
+
+        t = threading.Thread(target=target, name="obs-test-thread")
+        t.start()
+        t.join()
+        self.assertEqual(before + 1, self._count("thread"))
+        records = [r for r in self.handler.records if r.name == "DigiScript"]
+        self.assertEqual(1, len(records))
+        self.assertEqual(logging.ERROR, records[0].levelno)
+        self.assertIn("obs-test-thread", records[0].getMessage())
+        self.assertIn("boom-thread", repr(records[0].exc_info[1]))
+
+    async def test_loop_callback_exception_logged_with_traceback(self):
+        loop = asyncio.get_running_loop()
+        install_exception_hooks(loop)
+        before = self._count("asyncio")
+        done = asyncio.Event()
+
+        def bad_callback():
+            loop.call_soon(done.set)
+            raise RuntimeError("boom-callback")
+
+        # Raw asyncio callbacks (call_soon/call_later) reach the loop handler.
+        # IOLoop.add_callback catches exceptions itself and logs them to
+        # tornado.application, which is already attached to the same handlers.
+        loop.call_soon(bad_callback)
+        await asyncio.wait_for(done.wait(), 5)
+        await asyncio.sleep(0.05)
+        self.assertEqual(before + 1, self._count("asyncio"))
+        records = [
+            r for r in self.handler.records if "boom-callback" in repr(r.exc_info)
+        ]
+        # Exactly one record carries the traceback (tornado's own logger may
+        # also report the callback; filter to ours and the asyncio logger).
+        ours = [r for r in records if r.name == "DigiScript"]
+        self.assertEqual(1, len(ours))
+        self.assertEqual(logging.ERROR, ours[0].levelno)
+        self.assertIsNotNone(ours[0].exc_info[2])
+        self.assertEqual([], [r for r in records if r.name == "asyncio"])
+
+    async def test_ioloop_add_callback_exception_logged_once(self):
+        loop = asyncio.get_running_loop()
+        install_exception_hooks(loop)
+        before = self._count("asyncio")
+        done = asyncio.Event()
+
+        def bad_callback():
+            loop.call_soon(done.set)
+            raise RuntimeError("boom-add-callback")
+
+        IOLoop.current().add_callback(bad_callback)
+        await asyncio.wait_for(done.wait(), 5)
+        await asyncio.sleep(0.05)
+        records = [
+            r for r in self.handler.records if "boom-add-callback" in repr(r.exc_info)
+        ]
+        self.assertEqual(1, len(records))
+        self.assertEqual("tornado.application", records[0].name)
+        self.assertEqual(before, self._count("asyncio"))
+
+    async def test_unretrieved_task_exception_logged_once(self):
+        loop = asyncio.get_running_loop()
+        install_exception_hooks(loop)
+        before = self._count("asyncio")
+
+        async def bad():
+            raise RuntimeError("boom-task")
+
+        task = asyncio.create_task(bad())
+        await asyncio.sleep(0.05)
+        del task
+        gc.collect()
+        await asyncio.sleep(0.05)
+        self.assertEqual(before + 1, self._count("asyncio"))
+        records = [r for r in self.handler.records if "boom-task" in repr(r.exc_info)]
+        self.assertEqual(1, len(records))
+        self.assertIn("Task exception was never retrieved", records[0].getMessage())
+        self.assertIn("future: <Task", records[0].getMessage())
+
+    def test_warnings_reach_py_warnings_logger_once(self):
+        with warnings.catch_warnings():
+            warnings.simplefilter("always")
+            install_exception_hooks()
+            warnings.warn("obs-test-warning", UserWarning)
+        records = [r for r in self.handler.records if r.name == "py.warnings"]
+        self.assertEqual(1, len(records))
+        self.assertIn("obs-test-warning", records[0].getMessage())
+
+    def test_hooks_chain_to_preexisting_hooks(self):
+        sys_calls, thread_calls = [], []
+        sys.excepthook = lambda *a: sys_calls.append(a)
+        threading.excepthook = thread_calls.append
+        install_exception_hooks()
+        try:
+            raise ValueError("chain-sys")
+        except ValueError:
+            sys.excepthook(*sys.exc_info())
+        t = threading.Thread(target=lambda: 1 / 0)
+        t.start()
+        t.join()
+        self.assertEqual(1, len(sys_calls))
+        self.assertEqual(1, len(thread_calls))
+        self.assertEqual(
+            2, len([r for r in self.handler.records if r.name == "DigiScript"])
+        )
+
+    def test_reinstall_cycles_restore_originals(self):
+        install_exception_hooks()
+        uninstall_exception_hooks()
+        install_exception_hooks()
+        install_exception_hooks()
+        uninstall_exception_hooks()
+        self.assertIs(self._excepthook, sys.excepthook)
+        self.assertIs(self._thread_hook, threading.excepthook)
+
+    def test_concurrent_install_is_race_free(self):
+        """Force the check-then-act interleaving: every thread passes the
+        "is anything saved yet" check before any of them writes."""
+
+        class SlowCheckDict(dict):
+            def __len__(self):
+                size = super().__len__()
+                time.sleep(0.05)  # widen the window after the emptiness check
+                return size
+
+        sentinel_sys = lambda *_a: None  # noqa: E731
+        sentinel_thread = lambda _a: None  # noqa: E731
+        sys.excepthook = sentinel_sys
+        threading.excepthook = sentinel_thread
+        barrier = threading.Barrier(4)
+
+        def run():
+            barrier.wait()
+            install_exception_hooks()
+
+        with mock.patch.object(logger_module, "_saved_hooks", SlowCheckDict()):
+            threads = [threading.Thread(target=run) for _ in range(4)]
+            for t in threads:
+                t.start()
+            for t in threads:
+                t.join()
+            uninstall_exception_hooks()
+        # Without the lock a late thread saves one of our own hooks / defaults
+        # as the "original", so the sentinels would not be restored.
+        self.assertIs(sentinel_sys, sys.excepthook)
+        self.assertIs(sentinel_thread, threading.excepthook)
+
+    def test_uninstall_does_not_clobber_later_hooks(self):
+        install_exception_hooks()
+
+        def later_sys(*_a):
+            pass
+
+        def later_thread(_a):
+            pass
+
+        sys.excepthook = later_sys
+        threading.excepthook = later_thread
+        uninstall_exception_hooks()
+        self.assertIs(later_sys, sys.excepthook)
+        self.assertIs(later_thread, threading.excepthook)
+
+    async def test_keyboard_interrupt_in_loop_uses_previous_handler(self):
+        loop = asyncio.get_running_loop()
+        calls = []
+        loop.set_exception_handler(lambda lp, ctx: calls.append(ctx))
+        install_exception_hooks(loop)
+        before = self._count("asyncio")
+        context = {"message": "ki", "exception": KeyboardInterrupt()}
+        loop.call_exception_handler(context)
+        self.assertEqual([context], calls)
+        self.assertEqual(before, self._count("asyncio"))
+
+    def test_keyboard_interrupt_does_not_recurse_after_reinstall(self):
+        calls = []
+        sys.excepthook = lambda *a: calls.append(a)
+        install_exception_hooks()
+        install_exception_hooks()
+        sys.excepthook(KeyboardInterrupt, KeyboardInterrupt(), None)
+        self.assertEqual(1, len(calls))
+
+    def test_uninstall_restores_warning_capture_state(self):
+        logging.captureWarnings(True)
+        install_exception_hooks()
+        uninstall_exception_hooks()
+        self.assertIsNotNone(logging._warnings_showwarning)
+        logging.captureWarnings(False)
+        install_exception_hooks()
+        uninstall_exception_hooks()
+        self.assertIsNone(logging._warnings_showwarning)
+
+    async def test_uninstall_restores_loop_handler(self):
+        loop = asyncio.get_running_loop()
+
+        def previous(_loop, _ctx):
+            pass
+
+        loop.set_exception_handler(previous)
+        install_exception_hooks(loop)
+        self.assertIsNot(previous, loop.get_exception_handler())
+        uninstall_exception_hooks()
+        self.assertIs(previous, loop.get_exception_handler())
+
+    async def test_message_only_context_not_counted_as_error(self):
+        loop = asyncio.get_running_loop()
+        install_exception_hooks(loop)
+        before = self._count("asyncio")
+        loop.call_exception_handler({"message": "Executing <Handle> took 1.0s"})
+        self.assertEqual(before, self._count("asyncio"))
+        records = [r for r in self.handler.records if r.name == "DigiScript"]
+        self.assertEqual(1, len(records))
+        self.assertEqual(logging.WARNING, records[0].levelno)
+
+    async def test_asyncio_handler_survives_failing_repr(self):
+        class BadRepr:
+            def __repr__(self):
+                raise RuntimeError("repr failed")
+
+        loop = asyncio.get_running_loop()
+        install_exception_hooks(loop)
+        before = self._count("asyncio")
+        loop.call_exception_handler(
+            {
+                "message": "boom",
+                "exception": RuntimeError("boom-repr"),
+                "future": BadRepr(),
+            }
+        )
+        self.assertEqual(before + 1, self._count("asyncio"))
+        records = [r for r in self.handler.records if "boom-repr" in repr(r.exc_info)]
+        self.assertEqual(1, len(records))
+        self.assertIn("unrepresentable BadRepr", records[0].getMessage())