fix(observability): isolate SSE stream spans and continue ingress traceparent

[cyntwang99]

Summary

The SSE task-event stream reused the ambient OpenTelemetry context across requests. Because the SSE body is pumped by the ASGI server long after the request handler returns, the ambient context could still carry a previous request's span — so a stream's logs (otelTraceID) and any child spans resolved to an unrelated, long-lived trace (cross-request context bleed).

This change makes each stream self-contained:

• Isolated per-stream span. Every stream_task_events call runs under its own span, parented only on the inbound W3C traceparent (via extract(carrier, context=Context())) or a fresh root when absent — never the ambient context. The empty Context() base is the isolation mechanism.
• Ingress traceparent continuation. The stream routes now pass request headers, so when a caller supplies a traceparent the stream span continues that trace as a child instead of starting a new root.
• Lifecycle span. The span carries open / first-event / close events and task.id / stream.outcome / disconnect.reason attributes, and stays attached for the generator's lifetime so log lines correlate to the correct trace. A finally block classifies the in-flight exception (GeneratorExit / CancelledError → client disconnect) so the ASGI disconnect path isn't mislabeled completed.

Existing SSE behavior (keepalive pings, backoff, terminal detection, shared-topic non-deletion) is unchanged.

Changes

• src/domain/use_cases/streams_use_case.py — module tracer + span lifecycle around the stream generator; carrier param.
• src/api/routes/tasks.py — both stream routes inject Request and pass carrier=dict(request.headers).
• tests/unit/use_cases/test_streams_use_case_tracing.py (new) — 7 tests covering isolation-from-ambient, traceparent continuation, ingress-wins-over-ambient, lifecycle event order/attributes, and clean context detach.

Test plan

• New tracing unit tests pass (7 passed)
• Full unit suite passes (433 passed) — no regression
• Ruff lint + format clean (pre-commit hooks pass)
• Integration suite (test_task_stream.py) — could not run locally (testcontainers/Docker env issue); calls are signature-compatible with the new carrier=None default
• E2E in a collector: confirm a /stream trace is a self-contained root without a traceparent header, and a child of the caller when one is supplied

Note: real SDK clients won't inject traceparent until the SDK-side ingress injection lands; until then the continuation half is verified with an explicit traceparent header.

🤖 Generated with Claude Code

Greptile Summary

Introduces isolated lifecycle tracing for task-event SSE streams.

• Creates a per-stream server span rooted independently or continued from an inbound W3C trace context.
• Passes request headers from both task-stream routes into the stream use case.
• Records stream lifecycle events, outcomes, disconnect reasons, and task identifiers.
• Adds unit coverage for trace isolation, parent continuation, lifecycle metadata, and context cleanup.

Confidence Score: 4/5

The pre-loop SSE failure path needs to be fixed before merging because task or Redis lookup errors now terminate the response without an SSE error event.

Initial task resolution, stream-tail lookup, and status reads execute outside the exception handler that serializes stream failures, allowing those errors to escape through the response wrapper as a generic server failure.

Files Needing Attention: agentex/src/domain/use_cases/streams_use_case.py

Important Files Changed

Filename	Overview
agentex/src/domain/use_cases/streams_use_case.py	Adds isolated stream tracing and lifecycle metadata, but leaves initial task and repository operations outside the SSE error-frame handler.
agentex/src/api/routes/tasks.py	Passes request headers from both SSE endpoints to the stream use case for ingress trace continuation.
agentex/tests/unit/use_cases/test_streams_use_case_tracing.py	Covers tracing lifecycle and context cleanup but does not exercise failures during pre-loop stream setup.

Sequence Diagram

sequenceDiagram
    participant Client
    participant Route as Task stream route
    participant Stream as StreamsUseCase
    participant OTel as OpenTelemetry
    participant Repo as Stream repository

    Client->>Route: GET task event stream\noptional traceparent
    Route->>Stream: "stream_task_events(carrier=headers)"
    Stream->>OTel: Extract parent from isolated Context
    Stream->>OTel: Start and attach stream span
    Stream->>Repo: Read task events
    loop Stream lifetime
        Repo-->>Stream: Event or idle cycle
        Stream-->>Client: SSE event or keepalive
    end
    Stream->>OTel: Record outcome and close event
    Stream->>OTel: End span and detach context

Loading

Prompt To Fix All With AI

### Issue 1
agentex/src/domain/use_cases/streams_use_case.py:151
**Setup failures escape the SSE stream**

When task lookup, Redis tail lookup, or the initial status read fails before the inner handler begins, the exception now exits the generator instead of producing the established SSE error event. The response wrapper then raises `StreamResponseError`, so clients receive a broken stream and a generic 500 rather than the stream error frame.

---

For each issue above, determine whether it is valid and should be fixed. If so, fix it directly.

_{Reviews (1): Last reviewed commit: "fix(observability): isolate SSE stream s..." | Re-trigger Greptile}

Greptile also left 1 inline comment on this PR.

[greptile-apps]

Setup failures escape the SSE stream

When task lookup, Redis tail lookup, or the initial status read fails before the inner handler begins, the exception now exits the generator instead of producing the established SSE error event. The response wrapper then raises StreamResponseError, so clients receive a broken stream and a generic 500 rather than the stream error frame.

Prompt To Fix With AI

This is a comment left during a code review.
Path: agentex/src/domain/use_cases/streams_use_case.py
Line: 151

Comment:
**Setup failures escape the SSE stream**

When task lookup, Redis tail lookup, or the initial status read fails before the inner handler begins, the exception now exits the generator instead of producing the established SSE error event. The response wrapper then raises `StreamResponseError`, so clients receive a broken stream and a generic 500 rather than the stream error frame.

---

For each issue above, determine whether it is valid and should be fixed. If so, fix it directly.