fix(monitoring): enrich gateway diagnostic scope

agent/monitoring/events.py

@@ -57,6 +57,7 @@ class GatewayDiagnosticEvent:
     version: Optional[str] = None
     severity: str = "warning"
     ts_ns: int = field(default_factory=_now_ns)
+    source_logger: Optional[str] = None
 
     def to_dict(self) -> Dict[str, Any]:
         return {"event": "gateway_diagnostic", **asdict(self)}

agent/monitoring/gateway_health.py

@@ -10,6 +10,7 @@ from __future__ import annotations
 
 import hashlib
 import logging
+import re
 from dataclasses import dataclass
 from typing import Any, Dict, List, Optional
 
@@ -38,12 +39,19 @@ _KNOWN_PLATFORM_STATES = _RUNNING_PLATFORM_STATES | _FATAL_PLATFORM_STATES | {
     "connecting", "disconnected", "disabled", "paused", "retrying", "unknown"
 }
 _SUPERVISION_MODES = {"systemd", "s6", "container", "launchd", "manual", "unknown"}
+_SOURCE_LOGGER_RE = re.compile(r"^gateway(?:\.[A-Za-z_][A-Za-z0-9_]*)*$")
 
 
 def _allowed_logger(name: str) -> bool:
     return name == "gateway" or name.startswith("gateway.")
 
 
+def source_logger_for_export(name: Any) -> Optional[str]:
+    """Return a bounded source-controlled gateway logger name for OTLP scope."""
+    value = str(name or "")
+    return value if len(value) <= 128 and _SOURCE_LOGGER_RE.fullmatch(value) else None
+
+
 def redact_gateway_message(message: Any) -> str:
     """Redact gateway diagnostic free text for operator-owned export.
 
@@ -67,7 +75,20 @@ def classify_gateway_error(raw: Any) -> str:
         return "rate_limited"
     if "timeout" in s or "timed out" in s:
         return "timeout"
-    if any(k in s for k in ("network", "connection", "dns", "socket")):
+    if any(
+        k in s
+        for k in (
+            "network",
+            "connection",
+            "dns",
+            "socket",
+            "connect call failed",
+            "failed to connect",
+            "cannot connect",
+            "unreachable",
+            "name resolution",
+        )
+    ):
         return "network_error"
     if any(k in s for k in ("config", "missing", "invalid")):
         return "invalid_config"
@@ -119,6 +140,8 @@ def _safe_instance_id(raw: Any) -> str:
 
 
 def subsystem_for_logger(logger_name: str) -> str:
+    if logger_name == "gateway.relay" or logger_name.startswith("gateway.relay."):
+        return "platform.relay"
     if logger_name.startswith("gateway.platforms."):
         parts = logger_name.split(".")
         if len(parts) >= 3 and parts[2]:
@@ -417,11 +440,14 @@ class GatewayDiagnosticLogHandler(logging.Handler):
                 return
             subsystem = subsystem_for_logger(record.name)
             message = record.getMessage()
+            error_class = classify_gateway_error(message)
             event = GatewayDiagnosticEvent(
                 name=f"gateway.log.{record.levelname.lower()}",
                 subsystem=subsystem,
+                source_logger=source_logger_for_export(record.name),
                 platform=platform_for_subsystem(subsystem),
-                error_class=classify_gateway_error(message),
+                error_class=error_class,
+                error_code=error_class,
                 profile=self.profile,
                 version=self.version,
                 severity=record.levelname.lower(),
@@ -438,5 +464,6 @@ __all__ = [
     "GatewayDiagnosticLogHandler",
     "build_gateway_health_snapshot",
     "classify_gateway_error",
+    "source_logger_for_export",
     "redact_gateway_message",
 ]

agent/monitoring/gateway_health_export.py

@@ -17,6 +17,8 @@ from typing import Any, Dict, Optional
 
 logger = logging.getLogger(__name__)
 
+_DEFAULT_DIAGNOSTIC_SCOPE = "hermes.gateway.diagnostics"
+
 _RESOURCE_ATTRIBUTE_KEYS = frozenset({
     "service.name",
     "service.namespace",
@@ -381,19 +383,31 @@ class GatewayDiagnosticLogStreamer:
             sdk["OTLPLogExporter"](endpoint=endpoint, headers=headers or None)
         )
         self._provider.add_log_record_processor(self._processor)
-        self._logger = self._provider.get_logger("hermes.gateway.diagnostics")
+        self._logger = self._provider.get_logger(_DEFAULT_DIAGNOSTIC_SCOPE)
         self._LogRecord = sdk["LogRecord"]
         self._sdk = sdk
         self.exported = 0
 
     def __call__(self, batch: list[Dict[str, Any]]) -> None:
+        from agent.monitoring.gateway_health import source_logger_for_export
+
         for ev in batch:
             if ev.get("event") != "gateway_diagnostic":
                 continue
             attrs = _diagnostic_log_attributes(ev)
-            # Rendered Python log messages may contain arbitrary user IDs, names,
-            # paths, or configured strings. Keep the OTLP body content-free and
-            # carry only the structured, allowlisted attributes above.
+            # Preserve the source-controlled Python logger as the OTel
+            # instrumentation scope. This adds precise code attribution without
+            # turning a fluid module layout into a maintained subsystem enum.
+            # Rendered messages stay out because they may contain arbitrary IDs,
+            # names, paths, or configured strings. A future, separately gated
+            # ``diagnostic_detail: redacted_message`` mode may add best-effort
+            # free text when an observability plane defines that privacy policy.
+            source_logger = source_logger_for_export(ev.get("source_logger"))
+            otel_logger = (
+                self._provider.get_logger(source_logger)
+                if source_logger is not None
+                else self._logger
+            )
             body = "gateway diagnostic"
             record = self._LogRecord(
                 timestamp=ev.get("ts_ns"),
@@ -405,7 +419,7 @@ class GatewayDiagnosticLogStreamer:
                 body=_redact_string(body),
                 attributes=attrs,
             )
-            self._logger.emit(record)
+            otel_logger.emit(record)
             self.exported += 1
 
     def shutdown(self) -> None:

agent/monitoring/redaction.py

@@ -10,8 +10,9 @@ process passes through ``redact_for_export``:
     are rewritten to ``[email]`` / ``[phone]`` / ``[id]``.
 
 There is deliberately no setting to weaken this. The monitoring plane is
-content-free by design, so the only free text it carries (log-derived
-diagnostic messages) is always fully scrubbed.
+content-free by design: rendered log messages are not exported, and bounded
+structured strings are still scrubbed as defense-in-depth. This redactor also
+remains available for a future, explicitly gated redacted-message detail mode.
 """
 
 from __future__ import annotations

tests/monitoring/test_gateway_health_export.py

@@ -2,6 +2,17 @@ from __future__ import annotations
 
 import logging
 
+import pytest
+
+
+def test_gateway_diagnostic_event_preserves_positional_error_class():
+    from agent.monitoring.events import GatewayDiagnosticEvent
+
+    event = GatewayDiagnosticEvent("gateway.log.warning", "gateway", "auth_failed")
+
+    assert event.error_class == "auth_failed"
+    assert event.source_logger is None
+
 
 def test_default_config_keeps_gateway_health_export_disabled():
     from hermes_cli.config import DEFAULT_CONFIG
@@ -180,12 +191,64 @@ def test_gateway_diagnostic_log_handler_never_carries_rendered_message(caplog):
     assert event["event"] == "gateway_diagnostic"
     assert event["name"] == "gateway.log.warning"
     assert event["subsystem"] == "platform.slack"
+    assert event["source_logger"] == "gateway.platforms.slack"
     assert event["error_class"] == "auth_failed"
     assert "redacted_message" not in event
     assert "acct_7f3a" not in str(event)
     assert "Alice Smith" not in str(event)
 
 
+@pytest.mark.parametrize(
+    "message",
+    [
+        "Connect call failed ('127.0.0.1', 9)",
+        "failed to connect to relay",
+        "connection refused",
+        "network is unreachable",
+        "temporary failure in name resolution",
+    ],
+)
+def test_gateway_error_classifier_recognizes_bounded_network_failures(message):
+    from agent.monitoring.gateway_health import classify_gateway_error
+
+    assert classify_gateway_error(message) == "network_error"
+
+
+def test_gateway_diagnostic_log_handler_enriches_relay_scope_without_message_content(
+    monkeypatch,
+):
+    from agent.monitoring import emitter
+    from agent.monitoring.gateway_health import GatewayDiagnosticLogHandler
+
+    captured = []
+
+    class DummyEmitter:
+        def emit(self, event):
+            captured.append(event.to_dict())
+
+    monkeypatch.setattr(emitter, "get_emitter", lambda: DummyEmitter())
+    handler = GatewayDiagnosticLogHandler(profile="default", version="v-test")
+    logger = logging.getLogger("gateway.relay.adapter")
+    logger.addHandler(handler)
+    try:
+        logger.warning(
+            "Connect call failed for ws://alice@example.com/private; "
+            "credential=«redacted:sk-…»"
+        )
+    finally:
+        logger.removeHandler(handler)
+
+    assert len(captured) == 1
+    event = captured[0]
+    assert event["subsystem"] == "platform.relay"
+    assert event["platform"] == "relay"
+    assert event["source_logger"] == "gateway.relay.adapter"
+    assert event["error_class"] == "network_error"
+    assert event["error_code"] == "network_error"
+    assert "alice@example.com" not in str(event)
+    assert "private" not in str(event)
+
+
 def test_runtime_status_transition_emits_lifecycle_and_platform_events(monkeypatch):
     from agent.monitoring import emitter
     from agent.monitoring.gateway_health import emit_runtime_status_transition
@@ -374,6 +437,69 @@ def test_diagnostic_log_attributes_are_allowlisted_redacted_and_profile_free():
     assert "top-secret-token" not in str(attrs)
 
 
+def test_diagnostic_log_streamer_uses_validated_source_as_otel_scope():
+    from types import SimpleNamespace
+
+    from agent.monitoring.gateway_health_export import GatewayDiagnosticLogStreamer
+
+    class FakeLogger:
+        def __init__(self):
+            self.records = []
+
+        def emit(self, record):
+            self.records.append(record)
+
+    class FakeProvider:
+        def __init__(self):
+            self.loggers = {}
+
+        def get_logger(self, name):
+            return self.loggers.setdefault(name, FakeLogger())
+
+    provider = FakeProvider()
+    streamer = object.__new__(GatewayDiagnosticLogStreamer)
+    streamer._provider = provider
+    streamer._logger = provider.get_logger("hermes.gateway.diagnostics")
+    streamer._LogRecord = lambda **kwargs: SimpleNamespace(**kwargs)
+    streamer._sdk = {
+        "INVALID_TRACE_ID": 0,
+        "INVALID_SPAN_ID": 0,
+        "TraceFlags": SimpleNamespace(DEFAULT=0),
+        "SeverityNumber": SimpleNamespace(
+            FATAL="fatal", ERROR="error", WARN="warn", INFO="info", DEBUG="debug"
+        ),
+    }
+    streamer.exported = 0
+
+    streamer([
+        {
+            "event": "gateway_diagnostic",
+            "name": "gateway.log.warning",
+            "subsystem": "platform.relay",
+            "platform": "relay",
+            "source_logger": "gateway.relay.adapter",
+            "error_class": "network_error",
+            "severity": "warning",
+        },
+        {
+            "event": "gateway_diagnostic",
+            "name": "gateway.log.warning",
+            "subsystem": "gateway",
+            "source_logger": "gateway.relay.adapter\nalice@example.com",
+            "error_class": "unknown",
+            "severity": "warning",
+        },
+    ])
+
+    precise = provider.loggers["gateway.relay.adapter"].records
+    fallback = provider.loggers["hermes.gateway.diagnostics"].records
+    assert len(precise) == 1
+    assert len(fallback) == 1
+    assert precise[0].body == "gateway diagnostic"
+    assert "source_logger" not in precise[0].attributes
+    assert "alice@example.com" not in str(provider.loggers)
+
+
 def test_gateway_health_export_start_is_fail_open_when_otlp_missing(monkeypatch):
     from agent.monitoring import gateway_health_export
     from agent.monitoring.gateway_health_export import GatewayHealthExportRuntime