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 localhost:8889/metrics | grep -E '^agentex_task_stream_'
agentex_task_stream_active{instance="agentex-api.unknown.21635",job="agentex-api"} 0 1785785023147
agentex_task_stream_closed_total{instance="agentex-api.unknown.21635",job="agentex-api",outcome="client_disconnect"} 1 1785785023147
agentex_task_stream_closed_total{instance="agentex-api.unknown.21635",job="agentex-api",outcome="error"} 1 1785785023147
agentex_task_stream_duration_seconds_bucket{instance="agentex-api.unknown.21635",job="agentex-api",outcome="client_disconnect",le="1"} 0 1785785023147
agentex_task_stream_duration_seconds_bucket{instance="agentex-api.unknown.21635",job="agentex-api",outcome="client_disconnect",le="5"} 0 1785785023147
agentex_task_stream_duration_seconds_bucket{instance="agentex-api.unknown.21635",job="agentex-api",outcome="client_disconnect",le="15"} 0 1785785023147
agentex_task_stream_duration_seconds_bucket{instance="agentex-api.unknown.21635",job="agentex-api",outcome="client_disconnect",le="30"} 0 1785785023147
agentex_task_stream_duration_seconds_bucket{instance="agentex-api.unknown.21635",job="agentex-api",outcome="client_disconnect",le="60"} 1 1785785023147
agentex_task_stream_duration_seconds_bucket{instance="agentex-api.unknown.21635",job="agentex-api",outcome="client_disconnect",le="120"} 1 1785785023147
agentex_task_stream_duration_seconds_bucket{instance="agentex-api.unknown.21635",job="agentex-api",outcome="client_disconnect",le="300"} 1 1785785023147
agentex_task_stream_duration_seconds_bucket{instance="agentex-api.unknown.21635",job="agentex-api",outcome="client_disconnect",le="600"} 1 1785785023147
agentex_task_stream_duration_seconds_bucket{instance="agentex-api.unknown.21635",job="agentex-api",outcome="client_disconnect",le="1800"} 1 1785785023147
agentex_task_stream_duration_seconds_bucket{instance="agentex-api.unknown.21635",job="agentex-api",outcome="client_disconnect",le="3600"} 1 1785785023147
agentex_task_stream_duration_seconds_bucket{instance="agentex-api.unknown.21635",job="agentex-api",outcome="client_disconnect",le="7200"} 1 1785785023147
agentex_task_stream_duration_seconds_bucket{instance="agentex-api.unknown.21635",job="agentex-api",outcome="client_disconnect",le="14400"} 1 1785785023147
agentex_task_stream_duration_seconds_bucket{instance="agentex-api.unknown.21635",job="agentex-api",outcome="client_disconnect",le="+Inf"} 1 1785785023147
agentex_task_stream_duration_seconds_sum{instance="agentex-api.unknown.21635",job="agentex-api",outcome="client_disconnect"} 39.602999999944586 1785785023147
agentex_task_stream_duration_seconds_count{instance="agentex-api.unknown.21635",job="agentex-api",outcome="client_disconnect"} 1 1785785023147
agentex_task_stream_duration_seconds_bucket{instance="agentex-api.unknown.21635",job="agentex-api",outcome="error",le="1"} 1 1785785023147
agentex_task_stream_duration_seconds_bucket{instance="agentex-api.unknown.21635",job="agentex-api",outcome="error",le="5"} 1 1785785023147
agentex_task_stream_duration_seconds_bucket{instance="agentex-api.unknown.21635",job="agentex-api",outcome="error",le="15"} 1 1785785023147
agentex_task_stream_duration_seconds_bucket{instance="agentex-api.unknown.21635",job="agentex-api",outcome="error",le="30"} 1 1785785023147
agentex_task_stream_duration_seconds_bucket{instance="agentex-api.unknown.21635",job="agentex-api",outcome="error",le="60"} 1 1785785023147
agentex_task_stream_duration_seconds_bucket{instance="agentex-api.unknown.21635",job="agentex-api",outcome="error",le="120"} 1 1785785023147
agentex_task_stream_duration_seconds_bucket{instance="agentex-api.unknown.21635",job="agentex-api",outcome="error",le="300"} 1 1785785023147
agentex_task_stream_duration_seconds_bucket{instance="agentex-api.unknown.21635",job="agentex-api",outcome="error",le="600"} 1 1785785023147
agentex_task_stream_duration_seconds_bucket{instance="agentex-api.unknown.21635",job="agentex-api",outcome="error",le="1800"} 1 1785785023147
agentex_task_stream_duration_seconds_bucket{instance="agentex-api.unknown.21635",job="agentex-api",outcome="error",le="3600"} 1 1785785023147
agentex_task_stream_duration_seconds_bucket{instance="agentex-api.unknown.21635",job="agentex-api",outcome="error",le="7200"} 1 1785785023147
agentex_task_stream_duration_seconds_bucket{instance="agentex-api.unknown.21635",job="agentex-api",outcome="error",le="14400"} 1 1785785023147
agentex_task_stream_duration_seconds_bucket{instance="agentex-api.unknown.21635",job="agentex-api",outcome="error",le="+Inf"} 1 1785785023147
agentex_task_stream_duration_seconds_sum{instance="agentex-api.unknown.21635",job="agentex-api",outcome="error"} 0.03399999998509884 1785785023147
agentex_task_stream_duration_seconds_count{instance="agentex-api.unknown.21635",job="agentex-api",outcome="error"} 1 1785785023147
agentex_task_stream_opened_total{instance="agentex-api.unknown.21635",job="agentex-api"} 2 1785785023147
agentex_task_stream_stall_total{instance="agentex-api.unknown.21635",job="agentex-api"} 1 1785785023147
cynthia.wang@SCMJMH9C4MX2W agentex % 

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 series that fill the observability gap left by auto-instrumented HTTP RED metrics on the long-lived GET /tasks/{task_id}/stream SSE endpoint. Each series is scoped to a bounded outcome attribute, so cardinality stays flat across any deployment scale.

• stream_metrics.py: New OTel-only module with lazily-created instruments, exception-swallowing emit functions, and clear separation from the older StatsD dual-emit path used by cache_metrics.py.
• streams_use_case.py: Instruments the generator using an opened flag inside a try/finally so the active gauge is always balanced; stall detection uses a separate last_event_time clock so keepalive pings don't mask real idle periods, and a stalled boolean ensures each stall episode increments the counter exactly once.
• environment_variables.py: Adds SSE_STREAM_STALL_THRESHOLD_SECONDS (default 30 s) with consistent enum key, model field, and from_env wiring.

Confidence Score: 5/5

Safe to merge — instrumentation is additive, all emit functions are exception-safe, and the active gauge is correctly balanced via the opened flag in try/finally.

All five metric series are wired correctly, the outcome attribute stays bounded to three values, the opened guard prevents double-decrementing the active gauge, and stall detection uses a separate clock so keepalive pings do not mask real idle periods. Unit tests cover the no-op, error-swallowing, and emit paths. No metric cardinality violations found.

Files Needing Attention: No files require special attention.

Important Files Changed

Filename	Overview
agentex/src/utils/stream_metrics.py	New OTel-only metrics module for SSE stream lifecycle; lazy instrument init, all emit paths swallow exceptions, bounded outcome attribute only — no cardinality issues.
agentex/src/domain/use_cases/streams_use_case.py	Adds stream-open/close instrumentation inside try/finally with opened flag to avoid over-decrementing the active gauge; correctly maps all four exit paths to the right StreamOutcome values.
agentex/src/config/environment_variables.py	Adds SSE_STREAM_STALL_THRESHOLD_SECONDS env var (default 30) alongside the existing keepalive interval; enum key, model field, and from_env construction are all consistent.
agentex/tests/unit/utils/test_stream_metrics.py	Unit tests cover the three key behaviours: no-op when unconfigured, exception-swallowing, and correct OTel instrument recording for open/close/stall paths.

Sequence Diagram

sequenceDiagram
    participant Client
    participant StreamsUseCase
    participant stream_metrics
    participant OTel

    Client->>StreamsUseCase: "GET /tasks/{id}/stream"
    Note over StreamsUseCase: stream_start_time = loop.time()
    StreamsUseCase->>stream_metrics: record_stream_opened()
    stream_metrics->>OTel: opened_counter +1, active_updown +1
    StreamsUseCase-->>Client: data connected
    loop Live polling
        StreamsUseCase->>StreamsUseCase: read_messages()
        alt Messages received
            StreamsUseCase-->>Client: data event
            Note over StreamsUseCase: last_event_time=now, stalled=False
        else Idle
            alt stall threshold exceeded
                StreamsUseCase->>stream_metrics: record_stream_stall()
                stream_metrics->>OTel: stall_counter +1
            end
            StreamsUseCase-->>Client: :ping
        end
    end
    alt Normal completion
        StreamsUseCase->>stream_metrics: record_stream_closed(completed, duration)
    else Client disconnect
        StreamsUseCase->>stream_metrics: record_stream_closed(client_disconnect, duration)
    else Error
        StreamsUseCase->>stream_metrics: record_stream_closed(error, duration)
    end
    stream_metrics->>OTel: closed_counter, duration_histogram, active_updown -1

Loading

_{Reviews (8): Last reviewed commit: "fix(tasks): set seconds-scale buckets on..." | 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.

[harvhan]

nit: this repo is public and we keep internal ticket IDs out of anything pushed, including docstrings. Suggest dropping the parenthetical ticket reference, the rationale text stands on its own.

[harvhan]

The duration histogram doesn't pass explicit_bucket_boundaries_advisory, so the SDK falls back to its default boundaries (0, 5, 10, 25, ... 10000), which are tuned for milliseconds. Your local test output shows it: the 81 second stream landed in the le=100 bucket, and anything past roughly 2.8 hours goes to +Inf. rpc_metrics.py and db_metrics.py hit the same issue and pass explicit boundaries in seconds. Since this is now the only duration signal, suggest a seconds scale advisory suited to stream lifetimes, for example (1, 5, 15, 30, 60, 120, 300, 600, 1800, 3600, 7200, 14400). The exact values matter less than being seconds scale.

[cyntwang99]

@greptileai review