feat(tasks): add task-stream lifecycle metrics

agentex/src/config/environment_variables.py

@@ -55,6 +55,7 @@ class EnvVarKeys(str, Enum):
     HTTPX_POOL_TIMEOUT = "HTTPX_POOL_TIMEOUT"
     HTTPX_STREAMING_READ_TIMEOUT = "HTTPX_STREAMING_READ_TIMEOUT"
     SSE_KEEPALIVE_PING_INTERVAL = "SSE_KEEPALIVE_PING_INTERVAL"
+    SSE_STREAM_STALL_THRESHOLD_SECONDS = "SSE_STREAM_STALL_THRESHOLD_SECONDS"
     AGENTEX_SERVER_TASK_QUEUE = "AGENTEX_SERVER_TASK_QUEUE"
     ENABLE_HEALTH_CHECK_WORKFLOW = "ENABLE_HEALTH_CHECK_WORKFLOW"
     ENABLE_AGENT_RUN_SCHEDULES = "ENABLE_AGENT_RUN_SCHEDULES"
@@ -150,6 +151,10 @@ class EnvironmentVariables(BaseModel):
         300.0  # HTTPX streaming read timeout in seconds (5 minutes)
     )
     SSE_KEEPALIVE_PING_INTERVAL: int = 15  # SSE keepalive ping interval in seconds
+    # An open stream is counted as stalled once this many seconds pass with no
+    # data event pushed to the client. Kept above the keepalive interval so that
+    # keepalive pings (which are not data events) don't mask a real stall.
+    SSE_STREAM_STALL_THRESHOLD_SECONDS: int = 30
     AGENTEX_SERVER_TASK_QUEUE: str | None = None
     ENABLE_HEALTH_CHECK_WORKFLOW: bool = False
     # Gates the agent run schedules API. Off by default; enabled in development.
@@ -245,6 +250,9 @@ def refresh(cls, force_refresh: bool = False) -> EnvironmentVariables | None:
             SSE_KEEPALIVE_PING_INTERVAL=int(
                 os.environ.get(EnvVarKeys.SSE_KEEPALIVE_PING_INTERVAL, "15")
             ),
+            SSE_STREAM_STALL_THRESHOLD_SECONDS=int(
+                os.environ.get(EnvVarKeys.SSE_STREAM_STALL_THRESHOLD_SECONDS, "30")
+            ),
             AGENTEX_SERVER_TASK_QUEUE=os.environ.get(
                 EnvVarKeys.AGENTEX_SERVER_TASK_QUEUE
             ),

agentex/src/domain/use_cases/streams_use_case.py

@@ -19,6 +19,12 @@
 from src.domain.entities.tasks import TERMINAL_TASK_STATUSES
 from src.domain.services.task_service import DAgentTaskService
 from src.utils.logging import make_logger
+from src.utils.stream_metrics import (
+    StreamOutcome,
+    record_stream_closed,
+    record_stream_opened,
+    record_stream_stall,
+)
 from src.utils.stream_topics import get_task_event_stream_topic
 
 logger = make_logger(__name__)
@@ -104,32 +110,67 @@ async def stream_task_events(
             task_id = task.id
 
         stream_topic = get_task_event_stream_topic(task_id=task_id)
-        # Cursor before status read: catches a racing terminal event.
-        last_id = await self.stream_repository.get_stream_tail_id(stream_topic)
-        task = await self.task_service.get_task(id=task_id)
-        # Send initial connection data
-        yield f"data: {TaskStreamConnectedEventEntity(type='connected', taskId=task_id).model_dump_json()}\n\n"
-        # Already terminal: replay buffered events and end (late connect).
-        if task.status in TERMINAL_TASK_STATUSES:
-            async for _id, data in self.read_messages(topic=stream_topic, last_id="0"):
-                yield f"data: {data.model_dump_json()}\n\n"
-                await asyncio.sleep(0.02)
-            logger.info(
-                f"Ending SSE stream for task {task_id}: already terminal at connect"
-            )
-            return
-
-        last_message_time = asyncio.get_running_loop().time()
-        ping_interval = float(
-            self.environment_variables.SSE_KEEPALIVE_PING_INTERVAL
-        )  # Configurable keepalive ping interval
-        # Track consecutive read failures so we can back off and avoid a
-        # tight error loop. When the Redis pool is exhausted, every connected
-        # client's read fails on each cycle; without backoff this turns into a
-        # log-ingestion firehose (one failure per client per cycle, ~once/sec).
-        consecutive_errors = 0
-        last_status_check = last_message_time
+
+        # Capture the timing/outcome state the finally needs *before* marking the
+        # stream open, then flip ``opened`` as the first statement inside the try.
+        # This keeps record_stream_opened paired with exactly one
+        # record_stream_closed: the open lives inside the try, so any failure
+        # after it still routes through the finally and rebalances the active
+        # gauge — closing the narrow window that opening before the try left
+        # exposed. Placed after task resolution so a bad task_name never counts
+        # as an opened stream.
+        stream_start_time = asyncio.get_running_loop().time()
+        outcome: StreamOutcome = "completed"
+        opened = False
         try:
+            record_stream_opened()
+            opened = True
+            # Snapshot the read cursor BEFORE yielding "connected". "connected"
+            # is the client's cue to send its message, which makes the agent
+            # start XADD-ing deltas. Snapshotting after the yield lets a
+            # congested relay fall behind far enough that those deltas land
+            # before the snapshot and are never read. Snapshotting first resolves
+            # to "0-0" (stream is empty until the client sends), so we read from
+            # the beginning. Reading the cursor before the status check also
+            # catches a task that goes terminal while we connect.
+            last_id = await self.stream_repository.get_stream_tail_id(stream_topic)
+            task = await self.task_service.get_task(id=task_id)
+            # Send initial connection data
+            yield f"data: {TaskStreamConnectedEventEntity(type='connected', taskId=task_id).model_dump_json()}\n\n"
+            # Already terminal: replay buffered events and end (late connect).
+            if task.status in TERMINAL_TASK_STATUSES:
+                async for _id, data in self.read_messages(
+                    topic=stream_topic, last_id="0"
+                ):
+                    yield f"data: {data.model_dump_json()}\n\n"
+                    await asyncio.sleep(0.02)
+                logger.info(
+                    f"Ending SSE stream for task {task_id}: already terminal at connect"
+                )
+                return
+            now = asyncio.get_running_loop().time()
+            # last_message_time drives the keepalive ping and is reset by pings;
+            # last_event_time tracks only real data pushes (a ping is not a data
+            # event) and drives stall detection, so a persistently quiet stream
+            # still trips the stall signal even while keepalive pings flow.
+            last_message_time = now
+            last_event_time = now
+            ping_interval = float(
+                self.environment_variables.SSE_KEEPALIVE_PING_INTERVAL
+            )  # Configurable keepalive ping interval
+            stall_threshold = float(
+                self.environment_variables.SSE_STREAM_STALL_THRESHOLD_SECONDS
+            )
+            # Whether the current quiet spell has already been counted, so one
+            # stall episode increments the counter once (measuring stall onsets)
+            # rather than once per idle cycle. Reset when a data event flows.
+            stalled = False
+            # Track consecutive read failures so we can back off and avoid a
+            # tight error loop. When the Redis pool is exhausted, every connected
+            # client's read fails on each cycle; without backoff this turns into a
+            # log-ingestion firehose (one failure per client per cycle, ~once/sec).
+            consecutive_errors = 0
+            last_status_check = last_message_time
             # Application-level control loop
             while True:
                 try:
@@ -168,7 +209,10 @@ async def stream_task_events(
                         # Send the data to the client
                         data_str = f"data: {data.model_dump_json()}\n\n"
                         yield data_str
-                        last_message_time = asyncio.get_running_loop().time()
+                        now = asyncio.get_running_loop().time()
+                        last_message_time = now
+                        last_event_time = now
+                        stalled = False
                         # Terminal event is the last one — end here.
                         if (
                             isinstance(data, TaskStreamTaskUpdatedEventEntity)
@@ -191,6 +235,13 @@ async def stream_task_events(
                     # the loop-top current_time would be stale for ping timing.
                     if message_count == 0:
                         now = asyncio.get_running_loop().time()
+                        # No data event pushed for the stall window: count the
+                        # onset once. Keepalive pings deliberately do not reset
+                        # last_event_time, so a persistently quiet stream is
+                        # visible even though the connection stays alive.
+                        if now - last_event_time >= stall_threshold and not stalled:
+                            stalled = True
+                            record_stream_stall()
                         if now - last_message_time >= ping_interval:
                             yield ":ping\n\n"
                             last_message_time = now
@@ -227,16 +278,24 @@ async def stream_task_events(
 
         except asyncio.CancelledError:
             # Just exit the generator on cancellation
+            outcome = "client_disconnect"
             logger.info(f"Client disconnected from SSE stream for task {task_id}")
-            pass
         except Exception as e:
+            outcome = "error"
             logger.error(
                 f"Fatal error in SSE stream for task {task_id}: {e}", exc_info=True
             )
             yield f"data: {TaskStreamErrorEventEntity(type='error', message=str(e)).model_dump_json()}\n\n"
         finally:
             # Don't delete the shared topic; the TTL reclaims it.
             logger.info(f"SSE stream for task {task_id} has ended")
+            # Only close what we actually opened, so the active gauge is never
+            # decremented for a stream that never incremented it.
+            if opened:
+                record_stream_closed(
+                    outcome,
+                    asyncio.get_running_loop().time() - stream_start_time,
+                )
 
 
 DStreamsUseCase = Annotated[StreamsUseCase, Depends(StreamsUseCase)]

agentex/src/utils/stream_metrics.py

@@ -0,0 +1,163 @@
+"""
+Metrics instrumentation for the task-event SSE stream lifecycle.
+
+Emission is OpenTelemetry-only:
+
+- When an OTLP endpoint is configured (``OTEL_EXPORTER_OTLP_ENDPOINT``), the
+  instruments are recorded through the OpenTelemetry SDK.
+- 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
+route records into ``http_server_request_duration_seconds`` at connection close.
+But a single SSE request produces exactly one duration sample, which says nothing
+about what happened *during* the stream's life: how many were concurrently open,
+whether a stream ended normally vs. by client disconnect vs. by error, or whether
+a stream went quiet (no events pushed) for a stretch. Those are the stream-
+lifecycle signals request-level instrumentation cannot express, and they are what
+this module adds.
+
+Instrument names are deliberately dotted (``agentex.task_stream.*``); the OTLP →
+Prometheus translation flattens dots to underscores, appends ``_total`` to
+monotonic counters, and appends the unit (``s`` → ``_seconds``) to the histogram,
+so the series land in Mimir as ``agentex_task_stream_opened_total``,
+``agentex_task_stream_closed_total``, ``agentex_task_stream_duration_seconds``,
+``agentex_task_stream_active`` and ``agentex_task_stream_stall_total``. These
+hand-instrumented series carry only bounded attributes (``outcome``) and no
+``http_route`` label — group by ``outcome``, not ``http_route``.
+"""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING, Literal
+
+from src.utils.logging import make_logger
+from src.utils.otel_metrics import get_meter
+
+if TYPE_CHECKING:
+    from opentelemetry.metrics import Counter, Histogram, UpDownCounter
+
+logger = make_logger(__name__)
+
+# 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
+# status 200 at connection close cannot tell these three apart, which is why the
+# distinction lives here as a bounded label rather than on the request metric.
+StreamOutcome = Literal["completed", "client_disconnect", "error"]
+
+# Lazily-created OTel instruments (created once, on first use).
+_opened_counter: Counter | None = None
+_closed_counter: Counter | None = None
+_duration_histogram: Histogram | None = None
+_active_updown: UpDownCounter | None = None
+_stall_counter: Counter | None = None
+_instruments_initialized = False
+
+
+def _ensure_instruments() -> None:
+    """Create the OTel instruments on first use. No-op if OTel is not configured."""
+    global _opened_counter, _closed_counter, _duration_histogram
+    global _active_updown, _stall_counter, _instruments_initialized
+
+    if _instruments_initialized:
+        return
+    _instruments_initialized = True
+
+    meter = get_meter("agentex.task_stream")
+    if meter is None:
+        # OTel not configured; every record_* call stays a no-op.
+        return
+
+    _opened_counter = meter.create_counter(
+        name="agentex.task_stream.opened",
+        description="Task-event SSE streams opened",
+        unit="{stream}",
+    )
+    _closed_counter = meter.create_counter(
+        name="agentex.task_stream.closed",
+        description="Task-event SSE streams closed, tagged by outcome",
+        unit="{stream}",
+    )
+    _duration_histogram = meter.create_histogram(
+        name="agentex.task_stream.duration",
+        description="Task-event SSE stream lifetime",
+        unit="s",
+    )
+    _active_updown = meter.create_up_down_counter(
+        name="agentex.task_stream.active",
+        description="Currently open task-event SSE streams",
+        unit="{stream}",
+    )
+    _stall_counter = meter.create_counter(
+        name="agentex.task_stream.stall",
+        description="Task-event SSE streams that went idle (no event pushed) past the stall threshold",
+        unit="{stream}",
+    )
+
+
+def record_stream_opened() -> None:
+    """
+    Record a task-event stream opening: bumps the opened counter and the active gauge.
+
+    Pair every call with exactly one ``record_stream_closed`` so the active gauge
+    stays balanced. Never raises: see ``record_stream_closed``.
+    """
+    try:
+        _ensure_instruments()
+
+        if _opened_counter is not None:
+            _opened_counter.add(1)
+        if _active_updown is not None:
+            _active_updown.add(1)
+    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 (also tagged by outcome), and lowers
+    the active gauge.
+
+    Args:
+        outcome: One of "completed", "client_disconnect", "error".
+        duration_seconds: How long the stream was open.
+
+    Never raises: emission failures (e.g. an OTel SDK fault) are swallowed so
+    instrumentation can never disrupt the SSE path.
+    """
+    try:
+        _ensure_instruments()
+
+        if _closed_counter is not None:
+            _closed_counter.add(1, {"outcome": outcome})
+        if _duration_histogram is not None:
+            _duration_histogram.record(duration_seconds, {"outcome": outcome})
+        if _active_updown is not None:
+            _active_updown.add(-1)
+    except Exception:
+        logger.debug("Failed to emit agentex.task_stream.closed metric", exc_info=True)
+
+
+def record_stream_stall() -> None:
+    """
+    Record that an open stream went idle — no event pushed to the client for
+    longer than the configured stall threshold. Emit once per stall episode
+    (the caller de-dupes) so the counter measures stall onsets, not idle cycles.
+
+    Never raises: see ``record_stream_closed``.
+    """
+    try:
+        _ensure_instruments()
+
+        if _stall_counter is not None:
+            _stall_counter.add(1)
+    except Exception:
+        logger.debug("Failed to emit agentex.task_stream.stall metric", exc_info=True)