fix(tasks): emit task_stream.active as an absolute StatsD gauge

statsd increment/decrement submit a COUNT, so the active metric rendered as a
rate of change rather than the live concurrency level. DogStatsD does not honor
gauge deltas, so track a process-local running total and report it via
statsd.gauge as an absolute value. The OTel UpDownCounter path already handled
this natively and is unchanged.

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

Author: cyntwang99

agentex/src/utils/stream_metrics.py

@@ -62,6 +62,14 @@
 _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."""
@@ -111,6 +119,7 @@ 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()
 
@@ -121,15 +130,17 @@ def record_stream_opened() -> None:
 
         if _STATSD_ENABLED:
             statsd.increment("agentex.task_stream.opened")
-            statsd.increment("agentex.task_stream.active")
+            # Report the concurrency level as an absolute gauge, not a delta.
+            _active_stream_count += 1
+            statsd.gauge("agentex.task_stream.active", _active_stream_count)
     except Exception:
         logger.debug("Failed to emit agentex.task_stream.opened metric", exc_info=True)
 
 
 def record_stream_closed(outcome: StreamOutcome, duration_seconds: float) -> None:
     """
     Record a task-event stream closing: bumps the closed counter (tagged by
-    outcome), records the stream lifetime, and decrements the active gauge.
+    outcome), records the stream lifetime, and lowers the active gauge.
 
     Args:
         outcome: One of "completed", "client_disconnect", "error".
@@ -138,6 +149,7 @@ 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()
 
@@ -152,7 +164,9 @@ 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)
-            statsd.decrement("agentex.task_stream.active")
+            # Report the concurrency level as an absolute gauge, not a delta.
+            _active_stream_count -= 1
+            statsd.gauge("agentex.task_stream.active", _active_stream_count)
     except Exception:
         logger.debug("Failed to emit agentex.task_stream.closed metric", exc_info=True)
 

agentex/tests/unit/utils/test_stream_metrics.py

@@ -46,8 +46,8 @@ def test_record_functions_swallow_emission_errors():
         patch.object(stream_metrics, "statsd") as mock_statsd,
     ):
         mock_statsd.increment.side_effect = OSError("socket in a bad state")
-        mock_statsd.decrement.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,13 +62,15 @@ 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 both the opened counter and the active gauge.
-    mock_statsd.increment.assert_any_call("agentex.task_stream.opened")
-    mock_statsd.increment.assert_any_call("agentex.task_stream.active")
+    # Opening bumps the opened counter and reports the active concurrency level
+    # as an absolute gauge (DogStatsD gauges are last-value-wins, not deltas).
+    mock_statsd.increment.assert_called_once_with("agentex.task_stream.opened")
+    mock_statsd.gauge.assert_called_once_with("agentex.task_stream.active", 1)
 
 
 @pytest.mark.unit
@@ -79,6 +81,7 @@ 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)
@@ -91,8 +94,8 @@ def test_record_stream_closed_emits_statsd_when_enabled():
     mock_statsd.histogram.assert_called_once_with(
         "agentex.task_stream.duration", 3250.0
     )
-    # Closing decrements the active gauge.
-    mock_statsd.decrement.assert_called_once_with("agentex.task_stream.active")
+    # Closing lowers the active gauge to the new absolute level (1 -> 0).
+    mock_statsd.gauge.assert_called_once_with("agentex.task_stream.active", 0)
 
 
 @pytest.mark.unit