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.

Testing

Local trace validation - Exercised the real stream_task_events code against a local OpenTelemetry collector to confirm the per-stream span is emitted with the right lineage and lifecycle.

What was run

• Added a temporary traces pipeline to the local collector config (otlp receiver → batch → debug exporter) so spans sent to :4317 print to the collector's stdout.
• Ran a throwaway host-side script that installs a TracerProvider + OTLP exporter (the local app process installs none), then calls stream_task_events twice through it: once with no inbound traceparent, once with a fixed W3C traceparent. Both invocations use a fake repo/task, so no Redis/Postgres needed.
• Read the spans back with docker logs agentex-otel-collector.
• Both artifacts are local-only and are not part of this change.
• output

cynthia.wang@SCMJMH9C4MX2W agentex % docker logs --since 40s agentex-otel-collector 2>&1 | tail -80
     -> server.address: Str(agentex-postgres)
     -> server.port: Int(5432)
     -> db.namespace: Str(agentex)
     -> deployment.environment: Str(development)
StartTimestamp: 2026-08-04 00:00:46.032029327 +0000 UTC
Timestamp: 2026-08-04 18:46:13.005977041 +0000 UTC
Value: 30
	{"kind": "exporter", "data_type": "metrics", "name": "debug"}
2026-08-04T18:46:29.476Z	info	TracesExporter	{"kind": "exporter", "data_type": "traces", "name": "debug", "resource spans": 1, "spans": 2}
2026-08-04T18:46:29.476Z	info	ResourceSpans #0
Resource SchemaURL: 
Resource attributes:
     -> telemetry.sdk.language: Str(python)
     -> telemetry.sdk.name: Str(opentelemetry)
     -> telemetry.sdk.version: Str(1.39.1)
     -> service.name: Str(unknown_service)
ScopeSpans #0
ScopeSpans SchemaURL: 
InstrumentationScope agentex.task_stream 
Span #0
    Trace ID       : a8570b702f8e9bf556ae9312ef52d5b4
    Parent ID      : 
    ID             : 2f61e8c10d846335
    Name           : stream task events
    Kind           : Server
    Start time     : 2026-08-04 18:46:20.65019 +0000 UTC
    End time       : 2026-08-04 18:46:20.671546 +0000 UTC
    Status code    : Unset
    Status message : 
Attributes:
     -> task.id: Str(task-123)
     -> stream.outcome: Str(completed)
     -> disconnect.reason: Str(already_terminal)
Events:
SpanEvent #0
     -> Name: open
     -> Timestamp: 2026-08-04 18:46:20.650199 +0000 UTC
     -> DroppedAttributesCount: 0
SpanEvent #1
     -> Name: first-event
     -> Timestamp: 2026-08-04 18:46:20.650381 +0000 UTC
     -> DroppedAttributesCount: 0
SpanEvent #2
     -> Name: close
     -> Timestamp: 2026-08-04 18:46:20.671544 +0000 UTC
     -> DroppedAttributesCount: 0
     -> Attributes::
          -> stream.outcome: Str(completed)
          -> disconnect.reason: Str(already_terminal)
Span #1
    Trace ID       : 4bf92f3577b34da6a3ce929d0e0e4736
    Parent ID      : 00f067aa0ba902b7
    ID             : ead43fd6edc6c9ba
    Name           : stream task events
    Kind           : Server
    Start time     : 2026-08-04 18:46:20.671629 +0000 UTC
    End time       : 2026-08-04 18:46:20.692803 +0000 UTC
    Status code    : Unset
    Status message : 
Attributes:
     -> task.id: Str(task-123)
     -> stream.outcome: Str(completed)
     -> disconnect.reason: Str(already_terminal)
Events:
SpanEvent #0
     -> Name: open
     -> Timestamp: 2026-08-04 18:46:20.671633 +0000 UTC
     -> DroppedAttributesCount: 0
SpanEvent #1
     -> Name: first-event
     -> Timestamp: 2026-08-04 18:46:20.671649 +0000 UTC
     -> DroppedAttributesCount: 0
SpanEvent #2
     -> Name: close
     -> Timestamp: 2026-08-04 18:46:20.692801 +0000 UTC
     -> DroppedAttributesCount: 0
     -> Attributes::
          -> stream.outcome: Str(completed)
          -> disconnect.reason: Str(already_terminal)
	{"kind": "exporter", "data_type": "traces", "name": "debug"}
cynthia.wang@SCMJMH9C4MX2W agentex % 

What it shows — two stream task events spans (Kind: Server):

• Span 0 (no traceparent) — empty Parent ID and its own Trace ID: a fresh root, confirming the fallback when no inbound context is present.
• Span 1 (with traceparent) — Trace ID equals the inbound trace and Parent ID equals the inbound span id, with a new span id of its own: the stream continues the caller's trace as a child rather than starting a new one.
  • Both carry the lifecycle events open → first-event → close and the attributes task.id, stream.outcome, disconnect.reason, verifying the per-stream span shape.

The context-isolation behavior (a stream must not inherit a leftover ambient span from an unrelated request) can't be reproduced by this out-of-process harness and is covered by the
unit tests in tests/unit/use_cases/test_streams_use_case_tracing.py.

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

The PR isolates each SSE task-event stream in its own OpenTelemetry span and continues valid inbound W3C trace context.

• Passes request headers from both task stream routes into the stream use case.
• Adds stream lifecycle events, outcome attributes, exception recording, and context cleanup.
• Moves task and stream setup into the SSE error-handling boundary.
• Adds unit coverage for context isolation, parent continuation, lifecycle metadata, disconnect handling, cleanup, and setup failures.

Confidence Score: 5/5

The PR appears safe to merge.

The previously reported setup failures are now handled inside the stream recovery boundary and produce an SSE error frame, with no blocking failure remaining.

Important Files Changed

Filename	Overview
agentex/src/domain/use_cases/streams_use_case.py	Introduces isolated per-stream tracing and lifecycle telemetry while moving the previously reported setup operations inside the SSE error-frame handler.
agentex/src/api/routes/tasks.py	Passes inbound request headers from both SSE routes to the stream use case for W3C trace-context continuation.
agentex/tests/unit/use_cases/test_streams_use_case_tracing.py	Covers trace isolation and continuation, lifecycle attributes, generator closure, context detachment, and setup-failure recovery.

Sequence Diagram

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

    Client->>Route: "GET /tasks/.../stream<br/>optional traceparent"
    Route->>Stream: "stream_task_events(carrier=headers)"
    Stream->>OTel: Extract using isolated Context
    Stream->>OTel: Start and attach per-stream span
    Stream->>Repo: Resolve task and stream cursor
    alt Setup succeeds
        Stream-->>Client: connected and event frames
    else Setup fails
        Stream->>OTel: Record exception and error status
        Stream-->>Client: SSE error frame
    end
    Stream->>OTel: Add close metadata, end span, detach context

Loading

_{Reviews (5): Last reviewed commit: "Merge branch 'main' into cynthiawang/agx..." | Re-trigger Greptile}

[harvhan]

This docstring references an internal ticket ID (line 3). This repo is public, so internal tracker IDs shouldn't land in anything that gets pushed, per the repo guidelines in CLAUDE.md. Suggest rewording to something like "These lock in the SSE trace context bleed fixes:". A follow up commit plus the eventual squash merge keeps main clean; if you want it fully scrubbed from the branch history too, amend and force push with lease.

[harvhan]

nit: the test name promises the span ends exactly once, but the body only asserts the context was detached. Adding a _only_stream_span(span_exporter) call here would cover the other half.

[stephen-wang24]

lg