fix(tasks): drop StatsD active gauge, keep concurrency OTel-only

A process-local gauge is wrong under multiple workers: DogStatsD gauges are
last-write-wins per flush, so N workers emitting their own counts make the
metric flap between workers instead of summing to true concurrency. Remove the
StatsD active gauge and its bookkeeping; the OTel UpDownCounter already sums
correctly across per-instance series and remains the sole concurrency source.
The additive StatsD emits (opened/closed counters, duration, stall) aggregate
correctly across workers and are unchanged.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

Author: cyntwang99

agentex/src/utils/stream_metrics.py

@@ -62,14 +62,6 @@
 _stall_counter: Counter | None = None
 _instruments_initialized = False
 
-# Process-local count of currently-open streams, used solely to emit the StatsD
-# ``active`` gauge as an absolute value. DogStatsD gauges are last-value-wins and
-# do not honor deltas (unlike Etsy StatsD); increment/decrement would instead
-# submit a COUNT — the rate of change — never the concurrency level. So the
-# StatsD path keeps the running total here and reports it via ``statsd.gauge``.
-# The OTel ``UpDownCounter`` expresses this natively and needs no bookkeeping.
-_active_stream_count = 0
-
 
 def _ensure_instruments() -> None:
     """Create the OTel instruments on first use. No-op if OTel is not configured."""
@@ -119,7 +111,6 @@ def record_stream_opened() -> None:
     Pair every call with exactly one ``record_stream_closed`` so the active gauge
     stays balanced. Never raises: see ``record_stream_closed``.
     """
-    global _active_stream_count
     try:
         _ensure_instruments()
 
@@ -130,9 +121,11 @@ def record_stream_opened() -> None:
 
         if _STATSD_ENABLED:
             statsd.increment("agentex.task_stream.opened")
-            # Report the concurrency level as an absolute gauge, not a delta.
-            _active_stream_count += 1
-            statsd.gauge("agentex.task_stream.active", _active_stream_count)
+            # No StatsD "active" gauge: concurrency is OTel-only. DogStatsD
+            # gauges are last-write-wins per flush, so N workers each emitting a
+            # process-local count would make the gauge flap between workers
+            # rather than sum to the true concurrency. The OTel UpDownCounter
+            # sums correctly across per-instance series at query time.
     except Exception:
         logger.debug("Failed to emit agentex.task_stream.opened metric", exc_info=True)
 
@@ -149,7 +142,6 @@ def record_stream_closed(outcome: StreamOutcome, duration_seconds: float) -> Non
     Never raises: emission failures (e.g. a StatsD UDP socket error or an OTel
     SDK fault) are swallowed so instrumentation can never disrupt the SSE path.
     """
-    global _active_stream_count
     try:
         _ensure_instruments()
 
@@ -164,9 +156,8 @@ def record_stream_closed(outcome: StreamOutcome, duration_seconds: float) -> Non
             statsd.increment("agentex.task_stream.closed", tags=[f"outcome:{outcome}"])
             # Datadog histograms conventionally take milliseconds for durations.
             statsd.histogram("agentex.task_stream.duration", duration_seconds * 1000)
-            # Report the concurrency level as an absolute gauge, not a delta.
-            _active_stream_count -= 1
-            statsd.gauge("agentex.task_stream.active", _active_stream_count)
+            # No StatsD "active" gauge — concurrency is OTel-only; a per-worker
+            # DogStatsD gauge would flap rather than sum (see record_stream_opened).
     except Exception:
         logger.debug("Failed to emit agentex.task_stream.closed metric", exc_info=True)
 

agentex/tests/unit/utils/test_stream_metrics.py

@@ -47,7 +47,6 @@ def test_record_functions_swallow_emission_errors():
     ):
         mock_statsd.increment.side_effect = OSError("socket in a bad state")
         mock_statsd.histogram.side_effect = OSError("socket in a bad state")
-        mock_statsd.gauge.side_effect = OSError("socket in a bad state")
 
         # None of these should raise despite the backend blowing up.
         stream_metrics.record_stream_opened()
@@ -62,15 +61,14 @@ def test_record_stream_opened_emits_statsd_when_enabled():
         patch.object(stream_metrics, "_instruments_initialized", True),
         patch.object(stream_metrics, "_opened_counter", None),
         patch.object(stream_metrics, "_active_updown", None),
-        patch.object(stream_metrics, "_active_stream_count", 0),
         patch.object(stream_metrics, "statsd") as mock_statsd,
     ):
         stream_metrics.record_stream_opened()
 
-    # Opening bumps the opened counter and reports the active concurrency level
-    # as an absolute gauge (DogStatsD gauges are last-value-wins, not deltas).
+    # Opening bumps the opened counter. Concurrency ("active") is deliberately
+    # OTel-only, so the StatsD path emits no gauge (see module rationale).
     mock_statsd.increment.assert_called_once_with("agentex.task_stream.opened")
-    mock_statsd.gauge.assert_called_once_with("agentex.task_stream.active", 1)
+    mock_statsd.gauge.assert_not_called()
 
 
 @pytest.mark.unit
@@ -81,7 +79,6 @@ def test_record_stream_closed_emits_statsd_when_enabled():
         patch.object(stream_metrics, "_closed_counter", None),
         patch.object(stream_metrics, "_duration_histogram", None),
         patch.object(stream_metrics, "_active_updown", None),
-        patch.object(stream_metrics, "_active_stream_count", 1),
         patch.object(stream_metrics, "statsd") as mock_statsd,
     ):
         stream_metrics.record_stream_closed("client_disconnect", 3.25)
@@ -94,8 +91,8 @@ def test_record_stream_closed_emits_statsd_when_enabled():
     mock_statsd.histogram.assert_called_once_with(
         "agentex.task_stream.duration", 3250.0
     )
-    # Closing lowers the active gauge to the new absolute level (1 -> 0).
-    mock_statsd.gauge.assert_called_once_with("agentex.task_stream.active", 0)
+    # Concurrency ("active") is OTel-only, so closing emits no StatsD gauge.
+    mock_statsd.gauge.assert_not_called()
 
 
 @pytest.mark.unit