[Infrastructure] Add OpenTelemetry tracing support for safety pipeline

README.md

@@ -540,13 +540,57 @@ All values can be set in `humane_proxy.yaml` (project root) or via `HUMANE_PROXY
 | `pipeline.stage2_ceiling` | `HUMANE_PROXY_STAGE2_CEILING` | `0.4` | Early exit after Stage 2 |
 | `stage3.provider` | `HUMANE_PROXY_STAGE3_PROVIDER` | `"auto"` | Stage 3 provider |
 | `stage3.timeout` | `HUMANE_PROXY_STAGE3_TIMEOUT` | `10` | Stage 3 timeout (s) |
+| `telemetry.enabled` | `HUMANE_PROXY_TELEMETRY_ENABLED` | `false` | Enable OpenTelemetry tracing (requires `humane-proxy[telemetry]`) |
+| `telemetry.otlp_endpoint` | `HUMANE_PROXY_OTLP_ENDPOINT` | `http://localhost:4318/v1/traces` | OTLP trace collector endpoint |
+| `telemetry.service_name` | `HUMANE_PROXY_SERVICE_NAME` | `"humane-proxy"` | Service name reported to backend |
 | `privacy.store_message_text` | — | `false` | Store raw text (vs SHA-256 hash) |
 | `escalation.rate_limit_max` | `HUMANE_PROXY_RATE_LIMIT_MAX` | `3` | Max alerts per session/window |
 | `storage.backend` | `HUMANE_PROXY_STORAGE_BACKEND` | `"sqlite"` | `"sqlite"`, `"redis"`, `"postgres"` |
 | `safety.categories.self_harm.response_mode` | — | `"block"` | `"block"` or `"forward"` |
 
 ---
 
+## OpenTelemetry Tracing
+
+HumaneProxy can emit OpenTelemetry traces for the safety pipeline and top-level proxy calls. This feature is optional and disabled by default for zero performance overhead.
+
+Install the telemetry extras:
+
+```bash
+pip install humane-proxy[telemetry]
+```
+
+Enable tracing in `humane_proxy.yaml`:
+
+```yaml
+telemetry:
+  enabled: true
+  otlp_endpoint: "http://localhost:4318/v1/traces"
+  service_name: "humane-proxy"
+```
+
+Or use environment variables:
+
+```bash
+export HUMANE_PROXY_TELEMETRY_ENABLED=true
+export HUMANE_PROXY_OTLP_ENDPOINT=http://localhost:4318/v1/traces
+export HUMANE_PROXY_SERVICE_NAME=humane-proxy
+```
+
+When enabled, spans include:
+
+- `humane_proxy.proxy.check` / `humane_proxy.proxy.check_async`
+- `humane_proxy.pipeline.classify`
+- `humane_proxy.stage1`
+- `humane_proxy.stage2`
+- `humane_proxy.stage3`
+- `humane_proxy.pipeline.finalize`
+
+Only safe telemetry attributes are emitted:
+`session_id` (hashed), `category`, `score`, `final_score`, `triggers_count`, `stage_reached`, and `message_hash`.
+
+---
+
 ## Privacy
 
 By default HumaneProxy **never stores raw message text**. Only a SHA-256 hash is persisted for correlation. The escalation DB stores:

humane_proxy.yaml

@@ -77,10 +77,22 @@ trajectory:
   window_size: 5
   spike_delta: 0.35
 
+# Optional OpenTelemetry tracing (disabled by default)
+telemetry:
+  enabled: false
+  service_name: "humane-proxy"
+  otlp_endpoint: "http://localhost:4318/v1/traces"
+
 escalation:
   rate_limit_max: 3
   rate_limit_window_hours: 1
   webhooks:
     slack_url: ""
     discord_url: ""
     pagerduty_routing_key: ""
+
+pipeline:
+  enabled_stages: [1, 2, 3]
+
+stage3:
+  provider: "openai_moderation"

humane_proxy/__init__.py

@@ -24,8 +24,15 @@
 # ---------------------------------------------------------------------------
 
 from pathlib import Path
+import hashlib
+from contextlib import nullcontext
 import yaml
 
+try:
+    from opentelemetry import trace as ot_trace
+except ImportError:
+    ot_trace = None
+
 _CONFIG_PATH = Path(__file__).resolve().parent / "config.yaml"
 
 
@@ -41,10 +48,6 @@ def load_config() -> dict:
 # ---------------------------------------------------------------------------
 # Plug-and-play public API
 # ---------------------------------------------------------------------------
-
-from humane_proxy.config import get_config as _get_config  # noqa: E402
-
-
 class HumaneProxy:
     """High-level, plug-and-play interface to the HumaneProxy safety pipeline.
 
@@ -65,20 +68,38 @@ class HumaneProxy:
 
     def __init__(self, config_path: str | None = None) -> None:
         import os
+
         if config_path:
             os.environ["HUMANE_PROXY_CONFIG"] = str(config_path)
 
         from humane_proxy.config import reload_config
+
         self._config = reload_config()
 
         # Ensure DB is initialised.
         from humane_proxy.escalation.local_db import init_db
+
         init_db()
 
+        # Initialize OpenTelemetry for direct library usage.
+        from humane_proxy.telemetry import setup_telemetry
+
+        setup_telemetry(self._config)
+
+        self._proxy_tracer = (
+            ot_trace.get_tracer("humane_proxy.proxy") if ot_trace is not None else None
+        )
+
         # Initialise the pipeline.
         from humane_proxy.classifiers.pipeline import SafetyPipeline
+
         self._pipeline = SafetyPipeline(self._config)
 
+    def _span(self, name: str):
+        if self._proxy_tracer is None:
+            return nullcontext()
+        return self._proxy_tracer.start_as_current_span(name)
+
     @property
     def config(self) -> dict:
         """Return the active merged configuration."""
@@ -98,7 +119,13 @@ def check(self, text: str, session_id: str = "programmatic") -> dict:
             ``{"safe": bool, "category": str, "score": float, "triggers": list,
                "stage_reached": int, ...}``
         """
-        result = self._pipeline.classify_sync(text, session_id)
+        with self._span("humane_proxy.proxy.check") as span:
+            if span is not None and span.is_recording(): 
+                span.set_attribute(
+                    "humane_proxy.session_id",
+                    hashlib.sha256(session_id.encode("utf-8")).hexdigest(),
+                )
+            result = self._pipeline.classify_sync(text, session_id)
         return result.to_dict()
 
     async def check_async(self, text: str, session_id: str = "programmatic") -> dict:
@@ -110,10 +137,17 @@ async def check_async(self, text: str, session_id: str = "programmatic") -> dict
             Same as :meth:`check`, but potentially enriched with Stage-3
             reasoning and higher accuracy.
         """
-        result = await self._pipeline.classify(text, session_id)
+        with self._span("humane_proxy.proxy.check_async") as span:
+            if span is not None and span.is_recording():
+                span.set_attribute(
+                    "humane_proxy.session_id",
+                    hashlib.sha256(session_id.encode("utf-8")).hexdigest(),
+                )
+            result = await self._pipeline.classify(text, session_id)
         return result.to_dict()
 
     def as_fastapi_app(self):
         """Return the configured FastAPI application instance."""
         from humane_proxy.middleware.interceptor import app
+
         return app