feat(tasks): add task-stream lifecycle metrics

[cyntwang99]

Summary

Adds stream-lifecycle instrumentation for the task-event SSE stream (GET /tasks/{task_id}/stream). HTTP request-level RED is already covered by auto-instrumentation — but a single long-lived SSE request emits exactly one duration sample at connection close, which says nothing about what happened during the stream's life. This fills that gap with five hand-instrumented series:

• agentex_task_stream_opened_total — streams opened
• agentex_task_stream_closed_total{outcome} — streams closed, tagged completed / client_disconnect / error
• agentex_task_stream_duration_seconds — stream lifetime histogram
• agentex_task_stream_active — currently-open streams (gauge)
• agentex_task_stream_stall_total — streams that went idle (no event pushed) past a configurable threshold

Local Testing:

• make dev -> kicks off local OpenTelemetry collector fired an SSE stream against the endpoint:
• ran fake tasks/{id}/stream API locally: curl -N http://localhost:5003/tasks/blah/stream
  • This returned the initial data: {"type":"connected","taskId":"blah"}, then :ping, then disconnected, and checcked the metrics via curl -s http://localhost:8889/metrics | grep agentex_task_stream, which printed out stuff that included all 5 custom metrics:

cynthia.wang@SCMJMH9C4MX2W agentex % curl -s http://localhost:8889/metrics | grep agentex_task_stream      
# HELP agentex_task_stream_active Currently open task-event SSE streams
# TYPE agentex_task_stream_active gauge
agentex_task_stream_active{instance="agentex-api.unknown.63",job="agentex-api"} 1 1785454302581
# HELP agentex_task_stream_closed_total Task-event SSE streams closed, tagged by outcome
# TYPE agentex_task_stream_closed_total counter
agentex_task_stream_closed_total{instance="agentex-api.unknown.63",job="agentex-api",outcome="client_disconnect"} 1 1785454302581
# HELP agentex_task_stream_duration_seconds Task-event SSE stream lifetime
# TYPE agentex_task_stream_duration_seconds histogram
agentex_task_stream_duration_seconds_bucket{instance="agentex-api.unknown.63",job="agentex-api",outcome="client_disconnect",le="0"} 0 1785454302581
agentex_task_stream_duration_seconds_bucket{instance="agentex-api.unknown.63",job="agentex-api",outcome="client_disconnect",le="5"} 0 1785454302581
agentex_task_stream_duration_seconds_bucket{instance="agentex-api.unknown.63",job="agentex-api",outcome="client_disconnect",le="10"} 0 1785454302581
agentex_task_stream_duration_seconds_bucket{instance="agentex-api.unknown.63",job="agentex-api",outcome="client_disconnect",le="25"} 0 1785454302581
agentex_task_stream_duration_seconds_bucket{instance="agentex-api.unknown.63",job="agentex-api",outcome="client_disconnect",le="50"} 0 1785454302581
agentex_task_stream_duration_seconds_bucket{instance="agentex-api.unknown.63",job="agentex-api",outcome="client_disconnect",le="75"} 0 1785454302581
agentex_task_stream_duration_seconds_bucket{instance="agentex-api.unknown.63",job="agentex-api",outcome="client_disconnect",le="100"} 1 1785454302581
agentex_task_stream_duration_seconds_bucket{instance="agentex-api.unknown.63",job="agentex-api",outcome="client_disconnect",le="250"} 1 1785454302581
agentex_task_stream_duration_seconds_bucket{instance="agentex-api.unknown.63",job="agentex-api",outcome="client_disconnect",le="500"} 1 1785454302581
agentex_task_stream_duration_seconds_bucket{instance="agentex-api.unknown.63",job="agentex-api",outcome="client_disconnect",le="750"} 1 1785454302581
agentex_task_stream_duration_seconds_bucket{instance="agentex-api.unknown.63",job="agentex-api",outcome="client_disconnect",le="1000"} 1 1785454302581
agentex_task_stream_duration_seconds_bucket{instance="agentex-api.unknown.63",job="agentex-api",outcome="client_disconnect",le="2500"} 1 1785454302581
agentex_task_stream_duration_seconds_bucket{instance="agentex-api.unknown.63",job="agentex-api",outcome="client_disconnect",le="5000"} 1 1785454302581
agentex_task_stream_duration_seconds_bucket{instance="agentex-api.unknown.63",job="agentex-api",outcome="client_disconnect",le="7500"} 1 1785454302581
agentex_task_stream_duration_seconds_bucket{instance="agentex-api.unknown.63",job="agentex-api",outcome="client_disconnect",le="10000"} 1 1785454302581
agentex_task_stream_duration_seconds_bucket{instance="agentex-api.unknown.63",job="agentex-api",outcome="client_disconnect",le="+Inf"} 1 1785454302581
agentex_task_stream_duration_seconds_sum{instance="agentex-api.unknown.63",job="agentex-api",outcome="client_disconnect"} 80.97900000005029 1785454302581
agentex_task_stream_duration_seconds_count{instance="agentex-api.unknown.63",job="agentex-api",outcome="client_disconnect"} 1 1785454302581
# HELP agentex_task_stream_opened_total Task-event SSE streams opened
# TYPE agentex_task_stream_opened_total counter
agentex_task_stream_opened_total{instance="agentex-api.unknown.63",job="agentex-api"} 2 1785454302581
# HELP agentex_task_stream_stall_total Task-event SSE streams that went idle (no event pushed) past the stall threshold
# TYPE agentex_task_stream_stall_total counter
agentex_task_stream_stall_total{instance="agentex-api.unknown.63",job="agentex-api"} 2 1785454302581

Details

• New src/utils/stream_metrics.py mirrors the dual-emit pattern in cache_metrics.py: records through the OpenTelemetry SDK when an OTLP endpoint is configured, emits StatsD when the Datadog Agent host is set, and is a cheap no-op otherwise. Every emit path is wrapped so an instrumentation fault can never disrupt the live SSE path.
• Instruments carry only the bounded outcome label (no ids, no http_route), so cardinality stays flat.
• Open/close are paired via try/finally in stream_task_events so the active gauge stays balanced even if setup raises. Stall detection tracks last data event separately from keepalive pings, so a persistently quiet stream still trips the stall signal while the connection stays alive. One stall episode increments the counter once (onset), not once per idle cycle.
• New env var SSE_STREAM_STALL_THRESHOLD_SECONDS (default 30) controls the stall window.

Test plan

• make test FILE=tests/unit/utils/test_stream_metrics.py — no-op-when-unconfigured, error-swallowing, and StatsD/OTel emission assertions
• Confirm the five series land in Mimir with the expected flattened names once deployed to an OTLP-enabled environment
• Verify outcome distribution (completed / client_disconnect / error) on a real stream close

🤖 Generated with Claude Code

Greptile Summary

This PR adds five hand-instrumented OTel metrics for the task-event SSE stream lifecycle (GET /tasks/{task_id}/stream), filling the gap left by HTTP-level auto-instrumentation which only emits a single duration sample at connection close. All emission paths are wrapped in try/except so instrumentation faults can never disrupt the live SSE path.

• stream_metrics.py: New OTel-only module exposing record_stream_opened, record_stream_closed, and record_stream_stall; instruments are lazily initialized and guarded behind null checks; only the bounded outcome label is used (no entity IDs).
• streams_use_case.py: Integrates the metrics into stream_task_events with record_stream_opened as the first statement inside the try block and record_stream_closed in the finally, keeping the active gauge balanced across all exit paths; stall detection tracks last_event_time separately from last_message_time so keepalive pings don't mask quiet streams.
• environment_variables.py: Adds SSE_STREAM_STALL_THRESHOLD_SECONDS (default 30 s) above the 15 s ping interval.

Confidence Score: 5/5

Safe to merge; instrumentation is fully isolated from the SSE path and all emit calls are wrapped in error-swallowing guards.

The change adds observability-only code that cannot affect task streaming correctness: every record_* function swallows its own exceptions, the open/close gauge is balanced by a try/finally, and the only label used (outcome) is bounded to three values. No data path logic was altered.

Files Needing Attention: No files require special attention.

Important Files Changed

Filename	Overview
agentex/src/utils/stream_metrics.py	New OTel-only metrics module; instruments are cleanly gated behind null checks and errors are swallowed. Default histogram bucket boundaries may not be ideal for SSE stream durations.
agentex/src/domain/use_cases/streams_use_case.py	Instruments are correctly paired inside try/finally; stall detection and outcome tagging logic look correct; asyncio.get_running_loop().time() in the finally block is unprotected but safe in all realistic asyncio execution paths.
agentex/src/config/environment_variables.py	Adds SSE_STREAM_STALL_THRESHOLD_SECONDS with a sensible default (30 s, above the 15 s ping interval). Enum key and model field are consistent.
agentex/tests/unit/utils/test_stream_metrics.py	Tests cover no-op path, error-swallowing, and attribute assertions; _FakeInstrument / _ExplodingInstrument helpers are clean and purpose-built.

Sequence Diagram

sequenceDiagram
    participant Client
    participant StreamsUseCase
    participant stream_metrics
    participant OTel

    Client->>StreamsUseCase: "GET /tasks/{id}/stream"
    StreamsUseCase->>stream_metrics: record_stream_opened()
    stream_metrics->>OTel: opened_counter.add(1), active_updown.add(+1)

    StreamsUseCase->>Client: "data: {type: connected}"

    loop While task is active
        StreamsUseCase->>StreamsUseCase: read_messages (Redis XREAD)
        alt Data event received
            StreamsUseCase->>Client: "data: {event}"
            Note over StreamsUseCase: last_event_time = now, stalled = False
        else Empty read cycle
            Note over StreamsUseCase: now - last_event_time >= stall_threshold?
            opt Stall onset
                StreamsUseCase->>stream_metrics: record_stream_stall()
                stream_metrics->>OTel: stall_counter.add(1)
            end
            opt Ping due
                StreamsUseCase->>Client: :ping
            end
        end
    end

    alt Normal completion
        Note over StreamsUseCase: outcome=completed
    else Client disconnect
        Note over StreamsUseCase: outcome=client_disconnect
    else Fatal error
        Note over StreamsUseCase: outcome=error
    end

    StreamsUseCase->>stream_metrics: record_stream_closed(outcome, duration_s)
    stream_metrics->>OTel: closed_counter, duration_histogram, active_updown.add(-1)

Loading

_{Reviews (6): Last reviewed commit: "fix(tasks): drop direct StatsD emission ..." | Re-trigger Greptile}

[stephen-wang24]

Given that these are new metrics, and in the final state we want otel to emit to datadog instead of emitting to dd directly, i think we can remove the double emission here.