fix(tracing): keep eval span id captures apart from plain span pins so a span flood cannot evict them

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>

Author: mohammadatallah-scale

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

@@ -36,6 +36,8 @@
 _attrs_by_task: OrderedDict[str, dict[str, Any]] = OrderedDict()
 # Attrs captured when a span starts, so a later registry change cannot alter a span still queued for export.
 _attrs_by_span: OrderedDict[str, dict[str, Any]] = OrderedDict()
+# Spans pinned with no attrs live apart so a flood of plain spans cannot evict an eval span's pinned ids.
+_plain_spans: OrderedDict[str, None] = OrderedDict()
 _lock = threading.Lock()
 
 
@@ -83,24 +85,29 @@ def _lookup(span: Span) -> dict[str, Any]:
 def capture_for_span(span: Span) -> None:
     """Pin the span's attrs at start, empty included, so an eval task registered later cannot claim it."""
     with _lock:
-        _attrs_by_span[span.id] = _lookup(span)
-        while len(_attrs_by_span) > _MAX_TASKS:
-            _attrs_by_span.popitem(last=False)
+        attrs = _lookup(span)
+        store: OrderedDict[str, Any] = _attrs_by_span if attrs else _plain_spans
+        store[span.id] = attrs or None
+        while len(store) > _MAX_TASKS:
+            store.popitem(last=False)
 
 
 def release_span(span_id: str) -> None:
-    if _attrs_by_span:
+    if _attrs_by_span or _plain_spans:
         with _lock:
             _attrs_by_span.pop(span_id, None)
+            _plain_spans.pop(span_id, None)
 
 
 def attrs_for_span(span: Span) -> dict[str, Any]:
     """Attrs captured at span start, else those of the task found by ``span.task_id`` then ``span.trace_id``."""
-    if not _attrs_by_task and not _attrs_by_span:
+    if not _attrs_by_task and not _attrs_by_span and not _plain_spans:
         return {}
     with _lock:
         if span.id in _attrs_by_span:
             return dict(_attrs_by_span[span.id])
+        if span.id in _plain_spans:
+            return {}
         return _lookup(span)
 
 
@@ -109,3 +116,4 @@ def clear() -> None:
     with _lock:
         _attrs_by_task.clear()
         _attrs_by_span.clear()
+        _plain_spans.clear()

tests/lib/core/tracing/test_sgp_evals.py

@@ -201,6 +201,17 @@ def test_captures_are_bounded(self) -> None:
         assert sgp_evals.attrs_for_span(spans[0]) == {}
         assert sgp_evals.attrs_for_span(spans[2]) == EXPECTED_ATTRS
 
+    def test_plain_spans_cannot_evict_an_eval_span_capture(self) -> None:
+        sgp_evals.register_task_metadata("task-1", EVAL_METADATA)
+        eval_span = _span(data={})
+        sgp_evals.capture_for_span(eval_span)
+        sgp_evals.unregister_task("task-1")
+        with patch.object(sgp_evals, "_MAX_TASKS", 2):
+            for _ in range(5):
+                sgp_evals.capture_for_span(_span(data={}))
+
+        assert sgp_evals.attrs_for_span(eval_span) == EXPECTED_ATTRS
+
     def test_sync_trace_start_span_captures_ids(self) -> None:
         sgp_evals.register_task_metadata("task-1", EVAL_METADATA)
         span = Trace(processors=[], client=Mock(), trace_id="task-1").start_span(name="s")