fix(tasks): drop direct StatsD emission from stream metrics, OTel-only

These are new metrics and the target state routes OTel to Datadog through
the collector, so the parallel DogStatsD emission was a redundant second
copy of every point. Remove the StatsD path (and its os/datadog imports)
and update the tests to cover the OTel-only and no-op paths.

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

Author: cyntwang99

agentex/src/utils/stream_metrics.py

@@ -1,13 +1,17 @@
 """
 Metrics instrumentation for the task-event SSE stream lifecycle.
 
-Mirrors the dual-emit pattern in ``src/utils/cache_metrics.py``:
+Emission is OpenTelemetry-only:
 
 - When an OTLP endpoint is configured (``OTEL_EXPORTER_OTLP_ENDPOINT``), the
   instruments are recorded through the OpenTelemetry SDK.
-- When the Datadog Agent is reachable (``DD_AGENT_HOST``), the same events are
-  emitted as StatsD metrics.
-- When neither is configured, every function here is a cheap no-op.
+- When it is not configured, every function here is a cheap no-op.
+
+Unlike the older ``cache_metrics.py`` dual-emit path, there is no direct
+StatsD/DogStatsD emission here. These are new metrics, and the target state
+routes OTel to Datadog through the collector rather than emitting to the Datadog
+Agent directly — so a second, DogStatsD-native copy of every point would just be
+redundant.
 
 **Why this exists (see AGX1-616/AGX1-618):** HTTP request-level RED for
 ``GET /tasks/{task_id}/stream`` is already covered by auto-instrumentation — the
@@ -31,11 +35,8 @@
 
 from __future__ import annotations
 
-import os
 from typing import TYPE_CHECKING, Literal
 
-from datadog import statsd
-
 from src.utils.logging import make_logger
 from src.utils.otel_metrics import get_meter
 
@@ -44,9 +45,6 @@
 
 logger = make_logger(__name__)
 
-# StatsD is only emitted if the Datadog Agent host is configured.
-_STATSD_ENABLED = bool(os.environ.get("DD_AGENT_HOST"))
-
 # How a stream ended. "completed" = the generator returned normally;
 # "client_disconnect" = the client went away (asyncio.CancelledError);
 # "error" = the stream loop raised a fatal, non-recoverable exception. HTTP
@@ -74,7 +72,7 @@ def _ensure_instruments() -> None:
 
     meter = get_meter("agentex.task_stream")
     if meter is None:
-        # OTel not configured; OTel path stays disabled. StatsD may still emit.
+        # OTel not configured; every record_* call stays a no-op.
         return
 
     _opened_counter = meter.create_counter(
@@ -118,14 +116,6 @@ def record_stream_opened() -> None:
             _opened_counter.add(1)
         if _active_updown is not None:
             _active_updown.add(1)
-
-        if _STATSD_ENABLED:
-            statsd.increment("agentex.task_stream.opened")
-            # 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)
 
@@ -140,8 +130,8 @@ def record_stream_closed(outcome: StreamOutcome, duration_seconds: float) -> Non
         outcome: One of "completed", "client_disconnect", "error".
         duration_seconds: How long the stream was open.
 
-    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.
+    Never raises: emission failures (e.g. an OTel SDK fault) are swallowed so
+    instrumentation can never disrupt the SSE path.
     """
     try:
         _ensure_instruments()
@@ -152,17 +142,6 @@ def record_stream_closed(outcome: StreamOutcome, duration_seconds: float) -> Non
             _duration_histogram.record(duration_seconds, {"outcome": outcome})
         if _active_updown is not None:
             _active_updown.add(-1)
-
-        if _STATSD_ENABLED:
-            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,
-                tags=[f"outcome:{outcome}"],
-            )
-            # 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)
 
@@ -180,8 +159,5 @@ def record_stream_stall() -> None:
 
         if _stall_counter is not None:
             _stall_counter.add(1)
-
-        if _STATSD_ENABLED:
-            statsd.increment("agentex.task_stream.stall")
     except Exception:
         logger.debug("Failed to emit agentex.task_stream.stall metric", exc_info=True)

agentex/tests/unit/utils/test_stream_metrics.py

@@ -1,10 +1,10 @@
 """Tests for the task-stream lifecycle metrics emitter.
 
-Covers the two paths that matter operationally: the no-op path (neither OTel nor
-StatsD configured, which is the default in tests and local dev) must never
-raise, and the StatsD path must emit each metric with the expected name and
-tags. The SSE stream path must never be disrupted by an instrumentation fault,
-so emission errors must be swallowed.
+Emission is OpenTelemetry-only. Two paths matter operationally: the no-op path
+(OTel not configured, which is the default in tests and local dev) must never
+raise, and the OTel path must record each instrument with the expected value and
+bounded attributes. The SSE stream path must never be disrupted by an
+instrumentation fault, so emission errors must be swallowed.
 """
 
 from __future__ import annotations
@@ -17,9 +17,8 @@
 
 @pytest.mark.unit
 def test_record_functions_are_noop_when_unconfigured():
-    # With no OTLP endpoint and no DD_AGENT_HOST, every call must be harmless.
+    # With no OTLP endpoint the instruments stay None, so every call is harmless.
     with (
-        patch.object(stream_metrics, "_STATSD_ENABLED", False),
         patch.object(stream_metrics, "_opened_counter", None),
         patch.object(stream_metrics, "_closed_counter", None),
         patch.object(stream_metrics, "_duration_histogram", None),
@@ -34,82 +33,22 @@ def test_record_functions_are_noop_when_unconfigured():
 
 @pytest.mark.unit
 def test_record_functions_swallow_emission_errors():
-    # A failing backend must never propagate to the caller (live SSE path).
+    # A failing instrument must never propagate to the caller (live SSE path).
+    exploding = _ExplodingInstrument()
     with (
-        patch.object(stream_metrics, "_STATSD_ENABLED", True),
         patch.object(stream_metrics, "_instruments_initialized", True),
-        patch.object(stream_metrics, "_opened_counter", None),
-        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, "_stall_counter", None),
-        patch.object(stream_metrics, "statsd") as mock_statsd,
+        patch.object(stream_metrics, "_opened_counter", exploding),
+        patch.object(stream_metrics, "_closed_counter", exploding),
+        patch.object(stream_metrics, "_duration_histogram", exploding),
+        patch.object(stream_metrics, "_active_updown", exploding),
+        patch.object(stream_metrics, "_stall_counter", exploding),
     ):
-        mock_statsd.increment.side_effect = OSError("socket in a bad state")
-        mock_statsd.histogram.side_effect = OSError("socket in a bad state")
-
-        # None of these should raise despite the backend blowing up.
+        # None of these should raise despite the instrument blowing up.
         stream_metrics.record_stream_opened()
         stream_metrics.record_stream_closed("error", 2.0)
         stream_metrics.record_stream_stall()
 
 
-@pytest.mark.unit
-def test_record_stream_opened_emits_statsd_when_enabled():
-    with (
-        patch.object(stream_metrics, "_STATSD_ENABLED", True),
-        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, "statsd") as mock_statsd,
-    ):
-        stream_metrics.record_stream_opened()
-
-    # 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_not_called()
-
-
-@pytest.mark.unit
-def test_record_stream_closed_emits_statsd_when_enabled():
-    with (
-        patch.object(stream_metrics, "_STATSD_ENABLED", True),
-        patch.object(stream_metrics, "_instruments_initialized", True),
-        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, "statsd") as mock_statsd,
-    ):
-        stream_metrics.record_stream_closed("client_disconnect", 3.25)
-
-    mock_statsd.increment.assert_called_once_with(
-        "agentex.task_stream.closed",
-        tags=["outcome:client_disconnect"],
-    )
-    # Duration is reported to StatsD in milliseconds, tagged by outcome.
-    mock_statsd.histogram.assert_called_once_with(
-        "agentex.task_stream.duration",
-        3250.0,
-        tags=["outcome:client_disconnect"],
-    )
-    # Concurrency ("active") is OTel-only, so closing emits no StatsD gauge.
-    mock_statsd.gauge.assert_not_called()
-
-
-@pytest.mark.unit
-def test_record_stream_stall_emits_statsd_when_enabled():
-    with (
-        patch.object(stream_metrics, "_STATSD_ENABLED", True),
-        patch.object(stream_metrics, "_instruments_initialized", True),
-        patch.object(stream_metrics, "_stall_counter", None),
-        patch.object(stream_metrics, "statsd") as mock_statsd,
-    ):
-        stream_metrics.record_stream_stall()
-
-    mock_statsd.increment.assert_called_once_with("agentex.task_stream.stall")
-
-
 @pytest.mark.unit
 def test_closed_records_otel_instruments_with_outcome():
     # The OTel path records the outcome as a bounded attribute (not an id).
@@ -121,7 +60,6 @@ def test_closed_records_otel_instruments_with_outcome():
         _FakeInstrument(),
     )
     with (
-        patch.object(stream_metrics, "_STATSD_ENABLED", False),
         patch.object(stream_metrics, "_instruments_initialized", True),
         patch.object(stream_metrics, "_opened_counter", opened),
         patch.object(stream_metrics, "_closed_counter", closed),
@@ -139,6 +77,18 @@ def test_closed_records_otel_instruments_with_outcome():
     assert active.calls == [(1, None), (-1, None)]
 
 
+@pytest.mark.unit
+def test_stall_records_otel_counter():
+    stall = _FakeInstrument()
+    with (
+        patch.object(stream_metrics, "_instruments_initialized", True),
+        patch.object(stream_metrics, "_stall_counter", stall),
+    ):
+        stream_metrics.record_stream_stall()
+
+    assert stall.calls == [(1, None)]
+
+
 class _FakeInstrument:
     """Records (value, attributes) for add()/record() so tests can assert on them."""
 
@@ -150,3 +100,13 @@ def add(self, value, attributes=None):
 
     def record(self, value, attributes=None):
         self.calls.append((value, attributes))
+
+
+class _ExplodingInstrument:
+    """Raises on every emission to prove the record_* functions swallow faults."""
+
+    def add(self, value, attributes=None):
+        raise OSError("instrument in a bad state")
+
+    def record(self, value, attributes=None):
+        raise OSError("instrument in a bad state")