feat(tracing): OTel span queue and export telemetry (SGPINF-1863) (#373)

Add OpenTelemetry metrics for async span queue processing and SGP export:
queue depth, batch lag, drain duration, shutdown flush timing, and
export success/failure counters with bounded HTTP status labels.

Introduce AGENTEX_TRACING_METRICS=0|false|no|off kill switch to disable
SDK-side recording without code changes.

Linear: SGPINF-1863

Author: james-cardenas

CHANGELOG.md

@@ -1,5 +1,11 @@
 # Changelog
 
+## Unreleased
+
+### Features
+
+* **tracing:** emit OTel metrics for async span queue depth, batch drain, and SGP export success/failure (HTTP status labels). Disable SDK-side recording with ``AGENTEX_TRACING_METRICS=0``.
+
 ## 0.11.6 (2026-05-29)
 
 Full Changelog: [v0.11.5...v0.11.6](https://github.com/scaleapi/scale-agentex-python/compare/v0.11.5...v0.11.6)

src/agentex/lib/core/observability/tests/test_tracing_metrics.py

@@ -0,0 +1,100 @@
+"""Tests for ``agentex.lib.core.observability.tracing_metrics``."""
+
+from __future__ import annotations
+
+import agentex.lib.core.observability.tracing_metrics as tracing_metrics
+from agentex.lib.core.observability.tracing_metrics import (
+    TracingMetrics,
+    processor_label,
+    get_tracing_metrics,
+    classify_export_error,
+)
+
+
+class TestClassifyExportError:
+    def test_scale_gp_authentication_error(self):
+        class AuthenticationError(Exception):
+            pass
+
+        exc = AuthenticationError("Error code: 401 - {'message': 'Not authorized to access Account'}")
+        assert classify_export_error(exc) == ("authentication", "401")
+
+    def test_rate_limit_code(self):
+        class APIError(Exception):
+            pass
+
+        exc = APIError("Error code: 429 - rate limited")
+        assert classify_export_error(exc) == ("rate_limit", "429")
+
+    def test_server_error_code(self):
+        class APIError(Exception):
+            pass
+
+        exc = APIError("Error code: 503 - unavailable")
+        assert classify_export_error(exc) == ("server_error", "5xx")
+
+    def test_out_of_range_code_uses_bounded_label(self):
+        class APIError(Exception):
+            pass
+
+        exc = APIError("Error code: 100 - continue")
+        assert classify_export_error(exc) == ("other_error", "other")
+
+    def test_timeout_by_name(self):
+        class APITimeoutError(Exception):
+            pass
+
+        assert classify_export_error(APITimeoutError("slow")) == ("timeout", "timeout")
+
+    def test_unknown_error(self):
+        class WeirdError(Exception):
+            pass
+
+        assert classify_export_error(WeirdError("boom")) == ("other_error", "unknown")
+
+
+class TestProcessorLabel:
+    def test_sgp_async_processor(self):
+        class SGPAsyncTracingProcessor:
+            pass
+
+        assert processor_label(SGPAsyncTracingProcessor()) == "sgp"
+
+    def test_other_processor(self):
+        class AgentexAsyncTracingProcessor:
+            pass
+
+        assert processor_label(AgentexAsyncTracingProcessor()) == "other"
+
+
+class TestGetTracingMetrics:
+    def test_returns_tracing_metrics_instance(self, monkeypatch):
+        monkeypatch.setattr(tracing_metrics, "_tracing_metrics", None)
+        m = get_tracing_metrics()
+        assert isinstance(m, TracingMetrics)
+
+    def test_singleton_returns_same_instance(self, monkeypatch):
+        monkeypatch.setattr(tracing_metrics, "_tracing_metrics", None)
+        first = get_tracing_metrics()
+        second = get_tracing_metrics()
+        assert first is second
+
+    def test_instruments_exist(self, monkeypatch):
+        monkeypatch.setattr(tracing_metrics, "_tracing_metrics", None)
+        m = get_tracing_metrics()
+        for name in (
+            "span_events_enqueued",
+            "span_events_dropped",
+            "queue_depth",
+            "queue_lag",
+            "batch_items",
+            "batch_size",
+            "batch_drain_duration",
+            "export_batches",
+            "export_spans",
+            "export_batch_failures",
+            "export_span_failures",
+            "shutdown_timeouts",
+            "shutdown_remaining_items",
+        ):
+            assert hasattr(m, name), f"missing instrument: {name}"

src/agentex/lib/core/observability/tests/test_tracing_metrics_recording.py

@@ -0,0 +1,143 @@
+"""Tests for ``agentex.lib.core.observability.tracing_metrics_recording``."""
+
+from __future__ import annotations
+
+from unittest.mock import MagicMock, patch
+
+import agentex.lib.core.observability.tracing_metrics_recording as recording
+
+
+class _Item:
+    def __init__(self, enqueued_at: float | None) -> None:
+        self.enqueued_at = enqueued_at
+
+
+class TestIsMetricsEnabled:
+    def setup_method(self) -> None:
+        recording._metrics_enabled = None
+        recording._tracing = None
+
+    def test_enabled_by_default(self, monkeypatch):
+        monkeypatch.delenv("AGENTEX_TRACING_METRICS", raising=False)
+        assert recording.is_metrics_enabled() is True
+
+    def test_disabled_by_zero(self, monkeypatch):
+        monkeypatch.setenv("AGENTEX_TRACING_METRICS", "0")
+        recording._metrics_enabled = None
+        assert recording.is_metrics_enabled() is False
+
+
+class TestRecordingHelpers:
+    def setup_method(self) -> None:
+        recording._metrics_enabled = None
+        recording._tracing = None
+
+    def test_record_span_enqueued_when_disabled_does_not_load_metrics(self, monkeypatch):
+        monkeypatch.setenv("AGENTEX_TRACING_METRICS", "0")
+        recording._metrics_enabled = None
+        with patch(
+            "agentex.lib.core.observability.tracing_metrics.get_tracing_metrics"
+        ) as mock_get:
+            recording.record_span_enqueued("start")
+            mock_get.assert_not_called()
+
+    def test_record_span_enqueued_when_enabled(self, monkeypatch):
+        monkeypatch.setenv("AGENTEX_TRACING_METRICS", "1")
+        recording._metrics_enabled = None
+        mock_metrics = MagicMock()
+        with patch(
+            "agentex.lib.core.observability.tracing_metrics.get_tracing_metrics",
+            return_value=mock_metrics,
+        ):
+            recording.record_span_enqueued("end")
+        mock_metrics.span_events_enqueued.add.assert_called_once_with(1, {"event_type": "end"})
+
+    def test_monotonic_if_enabled_respects_kill_switch(self, monkeypatch):
+        monkeypatch.setenv("AGENTEX_TRACING_METRICS", "0")
+        recording._metrics_enabled = None
+        assert recording.monotonic_if_enabled() is None
+
+    def test_record_batch_coalesced_records_lag(self, monkeypatch):
+        monkeypatch.setenv("AGENTEX_TRACING_METRICS", "1")
+        recording._metrics_enabled = None
+        mock_metrics = MagicMock()
+        with patch(
+            "agentex.lib.core.observability.tracing_metrics.get_tracing_metrics",
+            return_value=mock_metrics,
+        ), patch("agentex.lib.core.observability.tracing_metrics_recording.time.monotonic", return_value=10.0):
+            recording.record_batch_coalesced(
+                queue_depth=3,
+                batch_items=[_Item(9.5), _Item(9.0)],
+            )
+        mock_metrics.queue_depth.record.assert_called_once_with(3)
+        mock_metrics.batch_items.record.assert_called_once_with(2)
+        mock_metrics.queue_lag.record.assert_called_once_with(1000.0)
+
+    def test_record_export_failure(self, monkeypatch):
+        monkeypatch.setenv("AGENTEX_TRACING_METRICS", "1")
+        recording._metrics_enabled = None
+        mock_metrics = MagicMock()
+
+        class AuthenticationError(Exception):
+            pass
+
+        exc = AuthenticationError("Error code: 401 - denied")
+        processor = type("SGPAsyncTracingProcessor", (), {})()
+
+        with patch(
+            "agentex.lib.core.observability.tracing_metrics.get_tracing_metrics",
+            return_value=mock_metrics,
+        ):
+            recording.record_export_failure(
+                processor=processor,
+                event_type="start",
+                span_count=5,
+                exc=exc,
+            )
+
+        mock_metrics.export_batch_failures.add.assert_called_once()
+        mock_metrics.export_span_failures.add.assert_called_once_with(
+            5,
+            {
+                "processor": "sgp",
+                "event_type": "start",
+                "http_code": "401",
+                "error_class": "authentication",
+            },
+        )
+
+    def test_record_export_success(self, monkeypatch):
+        monkeypatch.setenv("AGENTEX_TRACING_METRICS", "1")
+        recording._metrics_enabled = None
+        mock_metrics = MagicMock()
+        with patch(
+            "agentex.lib.core.observability.tracing_metrics.get_tracing_metrics",
+            return_value=mock_metrics,
+        ):
+            recording.record_export_success(event_type="end", span_count=12, processor="sgp")
+
+        mock_metrics.export_batches.add.assert_called_once_with(
+            1,
+            {"processor": "sgp", "event_type": "end"},
+        )
+        mock_metrics.export_spans.add.assert_called_once_with(
+            12,
+            {"processor": "sgp", "event_type": "end"},
+        )
+
+    def test_record_export_success_accepts_processor_label(self, monkeypatch):
+        monkeypatch.setenv("AGENTEX_TRACING_METRICS", "1")
+        recording._metrics_enabled = None
+        mock_metrics = MagicMock()
+        with patch(
+            "agentex.lib.core.observability.tracing_metrics.get_tracing_metrics",
+            return_value=mock_metrics,
+        ):
+            recording.record_export_success(
+                event_type="start", span_count=3, processor="other"
+            )
+
+        mock_metrics.export_batches.add.assert_called_once_with(
+            1,
+            {"processor": "other", "event_type": "start"},
+        )