fix(tracing): address Greptile review on queue depth and labels

Include batch size in queue_depth sampling, clamp out-of-range HTTP codes
to a bounded label, and parameterize export success processor.

Author: james-cardenas

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

@@ -33,6 +33,13 @@ class APIError(Exception):
         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

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

@@ -122,3 +122,20 @@ def test_record_export_success(self, monkeypatch):
             12,
             {"processor": "sgp", "event_type": "end", "outcome": "success"},
         )
+
+    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", "outcome": "success"},
+        )

src/agentex/lib/core/observability/tracing_metrics.py

@@ -148,7 +148,7 @@ def classify_export_error(exc: BaseException) -> tuple[str, str]:
             return "client_error", "4xx"
         if 500 <= code < 600:
             return "server_error", "5xx"
-        return "other_error", str(code)
+        return "other_error", "other"
 
     if any(s in name for s in ("Authentication", "Permission")):
         return "authentication", "unknown"

src/agentex/lib/core/observability/tracing_metrics_recording.py

@@ -97,13 +97,13 @@ def record_batch_phase(*, phase: str, size: int, duration_ms: float) -> None:
         pass
 
 
-def record_export_success(*, event_type: str, span_count: int) -> None:
+def record_export_success(*, event_type: str, span_count: int, processor: str = "sgp") -> None:
     if not is_metrics_enabled():
         return
     try:
         from agentex.lib.core.observability.tracing_metrics import get_tracing_metrics
 
-        attrs = {"processor": "sgp", "event_type": event_type, "outcome": "success"}
+        attrs = {"processor": processor, "event_type": event_type, "outcome": "success"}
         metrics = get_tracing_metrics()
         metrics.export_batches.add(1, attrs)
         metrics.export_spans.add(span_count, attrs)

src/agentex/lib/core/tracing/processors/sgp_tracing_processor.py

@@ -160,7 +160,9 @@ async def on_spans_start(self, spans: list[Span]) -> None:
 
         sgp_spans = [_build_sgp_span(span, self.env_vars) for span in spans]
         await client.spans.upsert_batch(items=[s.to_request_params() for s in sgp_spans])
-        _metrics.record_export_success(event_type="start", span_count=len(spans))
+        _metrics.record_export_success(
+            event_type="start", span_count=len(spans), processor="sgp"
+        )
 
     @override
     async def on_spans_end(self, spans: list[Span]) -> None:
@@ -177,7 +179,9 @@ async def on_spans_end(self, spans: list[Span]) -> None:
             sgp_span.end_time = span.end_time.isoformat()  # type: ignore[union-attr]
             sgp_spans.append(sgp_span)
         await client.spans.upsert_batch(items=[s.to_request_params() for s in sgp_spans])
-        _metrics.record_export_success(event_type="end", span_count=len(sgp_spans))
+        _metrics.record_export_success(
+            event_type="end", span_count=len(sgp_spans), processor="sgp"
+        )
 
     @override
     async def shutdown(self) -> None:

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

@@ -138,7 +138,7 @@ async def _drain_loop(self) -> None:
 
             try:
                 _metrics.record_batch_coalesced(
-                    queue_depth=self._queue.qsize(),
+                    queue_depth=self._queue.qsize() + len(batch),
                     batch_items=batch,
                 )
 

tests/lib/core/tracing/test_span_queue.py

@@ -503,6 +503,29 @@ async def capture_end(span: Span) -> 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