Keep uvicorn records off the syslog app stream

CHANGELOG.md

@@ -28,6 +28,12 @@
   own, such as SQLAlchemy's with `echo=True` or an application module set to
   `DEBUG`, got records below `loglevel` onto every stream on the log server
   and onto the debug console. Lower `loglevel` to see those records again.
+- Records from the `uvicorn` and `uvicorn.error` loggers no longer reach the
+  log server on the app stream. They carried no event id, so the log server
+  could not place them, and uvicorn's "Exception in ASGI application"
+  duplicated the application's `SYS_UNHANDLED_EXCEPTION` record. They still
+  reach the log server on the debug stream when `syslog_streams` selects it,
+  and the console app stream still shows them.
 
 ## 0.5.0 - 2026-09-17
 

gfmodules/logging/config_builder.py

@@ -188,7 +188,7 @@ def _add_log_handlers(self, conf: dict[str, Any]) -> None:
         root_handlers = conf["root"]["handlers"]
 
         bindings: dict[str, tuple[str, str | None, list[list[str]]]] = {
-            "app": ("json_app", "app_filter", [app_logger_handlers, uvicorn_handlers, uvicorn_error_handlers]),
+            "app": ("json_app", "app_filter", [app_logger_handlers]),
             "siem": ("json_siem", "siem_filter", [app_logger_handlers]),
             "public_inspect": ("json_public_inspect", "public_inspect_filter", [app_logger_handlers]),
             "debug": (

tests/test_config_builder.py

@@ -119,12 +119,12 @@
             "propagate": False,
         },
         "uvicorn": {
-            "handlers": ["console_app", "syslog_app"],
+            "handlers": ["console_app"],
             "level": "INFO",
             "propagate": False,
         },
         "uvicorn.error": {
-            "handlers": ["console_app", "syslog_app"],
+            "handlers": ["console_app"],
             "level": "INFO",
             "propagate": False,
         },
@@ -331,6 +331,17 @@ def test_only_the_debug_stream_reaches_the_root_logger(self) -> None:
 
         assert conf["root"]["handlers"] == ["syslog_debug"]
 
+    @pytest.mark.parametrize("logger", ["uvicorn", "uvicorn.error"])
+    def test_server_records_reach_the_log_server_on_the_debug_stream_only(self, logger: str) -> None:
+        conf = build(ConfigLogging(syslog_path=SYSLOG, syslog_streams=EVERY_SYSLOG_STREAM))
+
+        assert [name for name in conf["loggers"][logger]["handlers"] if name.startswith("syslog_")] == ["syslog_debug"]
+
+    def test_server_records_still_reach_the_app_console(self) -> None:
+        conf = build(ConfigLogging(syslog_path=SYSLOG))
+
+        assert "console_app" in conf["loggers"]["uvicorn.error"]["handlers"]
+
     def test_siem_and_public_inspect_are_bound_to_the_app_logger_only(self) -> None:
         conf = build(ConfigLogging(syslog_path=SYSLOG))
 

tests/test_end_to_end.py

@@ -6,8 +6,11 @@
 
 import json
 import logging
+import re
+import socket
 from collections.abc import AsyncGenerator, Iterator
 from contextlib import asynccontextmanager
+from typing import Any
 
 import pytest
 from fastapi import FastAPI
@@ -234,3 +237,63 @@ def test_an_invalid_console_stream_is_not_reported_as_an_unreachable_log_server(
 
         assert "unknown console_streams" in str(raised.value)
         assert "could not reach the log server" not in str(raised.value)
+
+
+class TestTheLogServer:
+    @pytest.fixture
+    def log_server(self) -> Iterator[socket.socket]:
+        with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as server:
+            server.bind(("127.0.0.1", 0))
+            server.settimeout(0.2)
+            host, port = server.getsockname()
+            gflog.configure(
+                config=gflog.ConfigLogging(
+                    application_id="example-service",
+                    syslog_path=f"{host}:{port}",
+                    console_streams=[],
+                    syslog_streams=["app", "siem", "public_inspect", "debug"],
+                ),
+                loglevel="info",
+                catalogue=CompleteCatalogue,
+            )
+            try:
+                yield server
+            finally:
+                for name in ("app", "uvicorn", "uvicorn.error", ""):
+                    for handler in logging.getLogger(name or None).handlers:
+                        handler.close()
+
+    @staticmethod
+    def _received(server: socket.socket) -> list[dict[str, Any]]:
+        payloads = []
+        try:
+            while True:
+                datagram = server.recv(65535).decode().rstrip("\x00")
+                payloads.append(json.loads(re.sub(r"^<\d+>", "", datagram)))
+        except TimeoutError:
+            return payloads
+
+    @pytest.mark.parametrize(
+        "name,message",
+        [("uvicorn", "Started server process"), ("uvicorn.error", "Exception in ASGI application")],
+    )
+    def test_a_server_record_reaches_the_debug_stream_only(
+        self, log_server: socket.socket, name: str, message: str
+    ) -> None:
+        logging.getLogger(name).error(message)
+
+        assert [payload["stream_id"] for payload in self._received(log_server)] == ["debug"]
+
+    def test_an_unhandled_exception_reaches_the_app_stream_once(self, log_server: socket.socket) -> None:
+        with TestClient(build_app(), raise_server_exceptions=False) as client:
+            client.get("/boom")
+        logging.getLogger("uvicorn.error").error("Exception in ASGI application")
+
+        errors = [
+            payload
+            for payload in self._received(log_server)
+            if payload["stream_id"] == "app" and payload["level"] == "ERROR"
+        ]
+        assert [(payload["event_id"], payload["event_description"]) for payload in errors] == [
+            (CompleteCatalogue.SYS_UNHANDLED_EXCEPTION.event_id, "unhandled exception")
+        ]