feat(tracing): plug span-queue telemetry into merged next reliability

Wire OTel metrics into the post-merge span queue: enqueue/drop counters,
batch coalescing depth/lag, drain phase timing, and export failure
recording on permanent failure or exhausted retries. Preserves enqueued_at
across re-enqueue for lag histograms.

Co-authored-by: Cursor <cursoragent@cursor.com>

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.4 (2026-05-26)
 
 Full Changelog: [v0.11.3...v0.11.4](https://github.com/scaleapi/scale-agentex-python/compare/v0.11.3...v0.11.4)

src/agentex/lib/core/tracing/span_queue.py

@@ -1,12 +1,14 @@
 from __future__ import annotations
 
 import os
+import time
 import asyncio
 from enum import Enum
 from dataclasses import dataclass
 
 from agentex.types.span import Span
 from agentex.lib.utils.logging import make_logger
+from agentex.lib.core.observability import tracing_metrics_recording as _metrics
 from agentex.lib.core.tracing.processors.tracing_processor_interface import (
     AsyncTracingProcessor,
 )
@@ -67,6 +69,7 @@ class _SpanQueueItem:
     event_type: SpanEventType
     span: Span
     processors: list[AsyncTracingProcessor]
+    enqueued_at: float | None = None
     # Number of times this item has already been dispatched.  Used to bound
     # re-enqueue on transient failures.
     attempts: int = 0
@@ -134,6 +137,12 @@ def _record_drop(self, count: int, reason: str) -> None:
         if count <= 0:
             return
         self._dropped_spans += count
+        if "shutdown" in reason:
+            for _ in range(count):
+                _metrics.record_span_dropped("shutdown")
+        elif "queue full" in reason:
+            for _ in range(count):
+                _metrics.record_span_dropped("queue_full")
         # Warn on the first drop and then sparsely, so a drop storm is visible
         # without flooding the log.
         if self._dropped_spans == count or self._dropped_spans % 100 < count:
@@ -155,7 +164,15 @@ def enqueue(
             return
         self._ensure_drain_running()
         try:
-            self._queue.put_nowait(_SpanQueueItem(event_type=event_type, span=span, processors=processors))
+            self._queue.put_nowait(
+                _SpanQueueItem(
+                    event_type=event_type,
+                    span=span,
+                    processors=processors,
+                    enqueued_at=_metrics.monotonic_if_enabled(),
+                )
+            )
+            _metrics.record_span_enqueued(event_type.value)
         except asyncio.QueueFull:
             self._record_drop(1, "queue full")
 
@@ -197,16 +214,33 @@ async def _drain_loop(self) -> None:
                         break
 
             try:
+                _metrics.record_batch_coalesced(
+                    queue_depth=self._queue.qsize() + len(batch),
+                    batch_items=batch,
+                )
+
                 # Separate START and END events.  Processing all STARTs before
                 # ENDs ensures that on_span_start completes before on_span_end
                 # for any span whose both events land in the same batch.
                 starts = [i for i in batch if i.event_type == SpanEventType.START]
                 ends = [i for i in batch if i.event_type == SpanEventType.END]
 
                 if starts:
+                    phase_start = time.perf_counter()
                     await self._process_items(starts)
+                    _metrics.record_batch_phase(
+                        phase="start",
+                        size=len(starts),
+                        duration_ms=(time.perf_counter() - phase_start) * 1000.0,
+                    )
                 if ends:
+                    phase_start = time.perf_counter()
                     await self._process_items(ends)
+                    _metrics.record_batch_phase(
+                        phase="end",
+                        size=len(ends),
+                        duration_ms=(time.perf_counter() - phase_start) * 1000.0,
+                    )
             finally:
                 for _ in batch:
                     self._queue.task_done()
@@ -265,6 +299,12 @@ def _handle_failure(
             exhausted = len(items) - len(retriable)
             if exhausted:
                 self._record_drop(exhausted, f"{type(p).__name__} retries exhausted during {event_type.value}")
+                _metrics.record_export_failure(
+                    processor=p,
+                    event_type=event_type.value,
+                    span_count=exhausted,
+                    exc=exc,
+                )
             for item in retriable:
                 self._reenqueue(item, p)
             if retriable:
@@ -285,6 +325,12 @@ def _handle_failure(
             len(items),
             event_type.value,
         )
+        _metrics.record_export_failure(
+            processor=p,
+            event_type=event_type.value,
+            span_count=len(items),
+            exc=exc,
+        )
 
     def _reenqueue(self, item: _SpanQueueItem, p: AsyncTracingProcessor) -> None:
         """Put a single failed item back on the queue, scoped to the processor
@@ -295,6 +341,7 @@ def _reenqueue(self, item: _SpanQueueItem, p: AsyncTracingProcessor) -> None:
                     event_type=item.event_type,
                     span=item.span,
                     processors=[p],
+                    enqueued_at=item.enqueued_at,
                     attempts=item.attempts + 1,
                 )
             )
@@ -312,9 +359,11 @@ async def shutdown(self, timeout: float = 30.0) -> None:
         try:
             await asyncio.wait_for(self._queue.join(), timeout=timeout)
         except asyncio.TimeoutError:
+            remaining = self._queue.qsize()
             logger.warning(
-                "Span queue shutdown timed out after %.1fs with %d items remaining", timeout, self._queue.qsize()
+                "Span queue shutdown timed out after %.1fs with %d items remaining", timeout, remaining
             )
+            _metrics.record_shutdown_timeout(remaining_items=remaining)
         if self._drain_task is not None and not self._drain_task.done():
             self._drain_task.cancel()
             try:

tests/lib/core/tracing/test_span_queue.py

@@ -401,6 +401,7 @@ async def capture_starts(spans: list[Span]) -> None:
         assert sum(len(b) for b in received) == 7
 
 
+
 class _FakeHTTPError(Exception):
     """Mimics an SGP/httpx status error: carries a ``status_code`` attribute."""
 
@@ -630,3 +631,96 @@ async def capture_end(span: Span) -> None:
         # END should still carry output and end_time
         assert end_spans[0].output is not None
         assert end_spans[0].end_time is not None
+
+
+class TestAsyncSpanQueueMetrics:
+    async def test_batch_coalesced_records_depth_including_batch(self, monkeypatch):
+        monkeypatch.setenv("AGENTEX_TRACING_METRICS", "1")
+        import agentex.lib.core.observability.tracing_metrics_recording as recording
+
+        recording._metrics_enabled = None
+        proc = _make_processor()
+        queue = AsyncSpanQueue(linger_ms=0)
+        recorded_depths: list[int] = []
+
+        def capture_coalesced(*, queue_depth: int, batch_items: object) -> None:
+            recorded_depths.append(queue_depth)
+
+        with patch.object(recording, "record_batch_coalesced", side_effect=capture_coalesced):
+            for _ in range(3):
+                queue.enqueue(SpanEventType.START, _make_span(), [proc])
+            await asyncio.sleep(0.05)
+            await queue.shutdown()
+
+        assert recorded_depths, "expected at least one coalesced batch"
+        assert recorded_depths[0] >= 3, (
+            f"queue_depth should include batch items removed from queue, got {recorded_depths[0]}"
+        )
+
+    async def test_enqueue_records_enqueued_metric(self, monkeypatch):
+        monkeypatch.setenv("AGENTEX_TRACING_METRICS", "1")
+        import agentex.lib.core.observability.tracing_metrics_recording as recording
+
+        recording._metrics_enabled = None
+        recording._tracing = None
+        mock_metrics = MagicMock()
+        proc = _make_processor()
+        queue = AsyncSpanQueue()
+
+        with patch(
+            "agentex.lib.core.observability.tracing_metrics.get_tracing_metrics",
+            return_value=mock_metrics,
+        ):
+            queue.enqueue(SpanEventType.START, _make_span(), [proc])
+            await asyncio.sleep(0.05)
+            await queue.shutdown()
+
+        mock_metrics.span_events_enqueued.add.assert_any_call(1, {"event_type": "start"})
+
+    async def test_processor_failure_records_export_failure(self, monkeypatch):
+        monkeypatch.setenv("AGENTEX_TRACING_METRICS", "1")
+        import agentex.lib.core.observability.tracing_metrics_recording as recording
+
+        recording._metrics_enabled = None
+        recording._tracing = None
+        mock_metrics = MagicMock()
+
+        class ExportError(Exception):
+            pass
+
+        proc = AsyncMock()
+        proc.on_spans_start = AsyncMock(side_effect=ExportError("Error code: 401 - denied"))
+        proc.on_spans_end = AsyncMock()
+        queue = AsyncSpanQueue()
+
+        with patch(
+            "agentex.lib.core.observability.tracing_metrics.get_tracing_metrics",
+            return_value=mock_metrics,
+        ):
+            queue.enqueue(SpanEventType.START, _make_span(), [proc])
+            await asyncio.sleep(0.05)
+            await queue.shutdown()
+
+        mock_metrics.export_batch_failures.add.assert_called_once()
+        mock_metrics.export_span_failures.add.assert_called_once()
+
+    async def test_enqueue_overhead_with_metrics_disabled(self, monkeypatch):
+        monkeypatch.setenv("AGENTEX_TRACING_METRICS", "0")
+        import agentex.lib.core.observability.tracing_metrics_recording as recording
+
+        recording._metrics_enabled = None
+        recording._tracing = None
+        proc = _make_processor()
+        queue = AsyncSpanQueue()
+
+        with patch(
+            "agentex.lib.core.observability.tracing_metrics.get_tracing_metrics"
+        ) as mock_get:
+            start = time.monotonic()
+            for _ in range(200):
+                queue.enqueue(SpanEventType.START, _make_span(), [proc])
+            elapsed = time.monotonic() - start
+            await queue.shutdown()
+
+        assert elapsed < 0.05, f"disabled metrics enqueue too slow: {elapsed:.3f}s"
+        mock_get.assert_not_called()