test(e2e): otel trace completeness on streaming /v1/messages (LIT-3787)

tests/e2e/logging/logging_client.py

@@ -77,11 +77,12 @@
 
 
 class ResponsesRequestBody(BaseModel):
-    """OpenAI Responses API /v1/responses request (non-streaming)."""
+    """OpenAI Responses API /v1/responses request."""
 
     model: str
     input: str
     max_output_tokens: int
+    stream: bool | None = None
 
 
 class TeamCallbackBody(BaseModel):
@@ -464,30 +465,43 @@ def chat_raw(
             json=body,
         )
 
-    def messages_raw(self, key: str, model: str, text: str, *, max_tokens: int = 16) -> StreamingResponse:
-        """Non-streaming POST /v1/messages (Anthropic-native body): raw outcome
-        judged by status/body/headers, for tests that need x-litellm-call-id."""
+    def messages_raw(
+        self, key: str, model: str, text: str, *, max_tokens: int = 16, stream: bool = False
+    ) -> StreamingResponse:
+        """POST /v1/messages (Anthropic-native body): raw outcome judged by
+        status/body/headers, for tests that need x-litellm-call-id. With
+        ``stream=True`` the SSE body is consumed and its events counted."""
+        body = AnthropicMessagesBody(
+            model=model,
+            max_tokens=max_tokens,
+            messages=[ChatMessage(role="user", content=text)],
+            stream=True if stream else None,
+        )
+        if stream:
+            return self.gateway.transport.stream(
+                "/v1/messages", headers=self.gateway.transport.bearer(key), json=body
+            )
         return self.gateway.transport.send(
-            "/v1/messages",
-            headers=self.gateway.transport.bearer(key),
-            json=AnthropicMessagesBody(
-                model=model,
-                max_tokens=max_tokens,
-                messages=[ChatMessage(role="user", content=text)],
-            ),
+            "/v1/messages", headers=self.gateway.transport.bearer(key), json=body
         )
 
     def responses_raw(
-        self, key: str, model: str, text: str, *, max_output_tokens: int = 64
+        self, key: str, model: str, text: str, *, max_output_tokens: int = 64, stream: bool = False
     ) -> StreamingResponse:
-        """Non-streaming POST /v1/responses (OpenAI Responses API): raw outcome
-        judged by status/body/headers, for tests that need x-litellm-call-id.
+        """POST /v1/responses (OpenAI Responses API): raw outcome judged by
+        status/body/headers, for tests that need x-litellm-call-id.
         max_output_tokens caps reasoning-model output cost; a capped response is
-        still a 200 and still exports the trace."""
+        still a 200 and still exports the trace. With ``stream=True`` the SSE
+        body is consumed and its events counted."""
+        body = ResponsesRequestBody(
+            model=model, input=text, max_output_tokens=max_output_tokens, stream=True if stream else None
+        )
+        if stream:
+            return self.gateway.transport.stream(
+                "/v1/responses", headers=self.gateway.transport.bearer(key), json=body
+            )
         return self.gateway.transport.send(
-            "/v1/responses",
-            headers=self.gateway.transport.bearer(key),
-            json=ResponsesRequestBody(model=model, input=text, max_output_tokens=max_output_tokens),
+            "/v1/responses", headers=self.gateway.transport.bearer(key), json=body
         )
 
     def scrape_metrics(self) -> str:

tests/e2e/logging/test_otel_trace_e2e.py

@@ -96,7 +96,9 @@ def _chain_reaches(span_id: str, root_id: str, trace: JaegerTrace) -> bool:
     return False
 
 
-def _assert_complete_trace(hits: list[JaegerTrace], *, route: str, genai_span: str) -> None:
+def _assert_complete_trace(
+    hits: list[JaegerTrace], *, route: str, genai_span: str, require_cost_span: bool = True
+) -> None:
     """The enforced behavior: the destination holds exactly one trace for the
     call, rooted at the SERVER span, with auth/db/cost children and the gen-AI
     span all connected into that one tree - no dangling parent references."""
@@ -135,7 +137,8 @@ def _assert_complete_trace(hits: list[JaegerTrace], *, route: str, genai_span: s
     assert any(name.startswith(DB_SPAN_PREFIX) for name in names), (
         f"no db ('{DB_SPAN_PREFIX}*') span in the trace; spans: {names}"
     )
-    assert COST_SPAN in names, f"cost write span {COST_SPAN!r} missing; spans: {names}"
+    if require_cost_span:
+        assert COST_SPAN in names, f"cost write span {COST_SPAN!r} missing; spans: {names}"
 
     genai = next((span for span in trace.spans if span.operation_name == genai_span), None)
     assert genai is not None, f"gen-AI span {genai_span!r} missing; spans: {names}"
@@ -146,8 +149,9 @@ def _assert_complete_trace(hits: list[JaegerTrace], *, route: str, genai_span: s
     )
 
 
-def _settled_names(*, route: str, genai_span: str) -> set[str]:
-    return {f"POST {route}", f"auth {route}", COST_SPAN, genai_span}
+def _settled_names(*, route: str, genai_span: str, require_cost_span: bool = True) -> set[str]:
+    names = {f"POST {route}", f"auth {route}", genai_span}
+    return (names | {COST_SPAN}) if require_cost_span else names
 
 
 def _tag(span: JaegerSpan, key: str) -> str | int | float | bool | None:
@@ -315,3 +319,100 @@ def test_chat_completions_stream_exports_complete_trace(
             "the gen-AI span must record litellm.request.streaming=true; its absence means "
             "the stream flag was dropped before the model call"
         )
+
+    @pytest.mark.covers("logging.otel.stream.exports_metric", exercised_on=["messages"])
+    def test_messages_stream_exports_complete_trace(
+        self, client: LoggingClient, otel_reader: OtelReader, resources: ResourceManager
+    ) -> None:
+        """One successful STREAMED /v1/messages call must export ONE complete
+        OTEL trace: a single root SERVER span with auth/db/cost children and
+        the gen-AI CLIENT span connected back to the root.
+
+        Same streaming lifecycle risk as the chat surface (the gen-AI span
+        closes from the stream-consumption path), so the same stream-specific
+        assertions apply: the response actually streamed, exactly ONE gen-AI
+        span exists for the call, and the span records
+        litellm.request.streaming=true.
+        """
+        route = "/v1/messages"
+        _assert_otel_destination_configured(client)
+
+        key = client.key_with_alias(f"otel-stream-messages-{unique_marker()}", models=[MODEL])
+        resources.defer(lambda: client.delete_key(key))
+
+        marker = unique_marker()
+        outcome = _first_ok(
+            client,
+            lambda: client.messages_raw(key, MODEL, f"reply with one word {marker}", max_tokens=16, stream=True),
+        )
+        assert outcome.call_id is not None, "success response must carry x-litellm-call-id"
+        assert outcome.is_streaming, f"response must be an event stream, got content-type {outcome.content_type!r}"
+        assert outcome.chunks > 0, "the stream must deliver at least one event"
+
+        genai_span = f"chat {MODEL}"
+        hits = otel_reader.poll_traces_for_call(
+            call_id=outcome.call_id,
+            settled_names=_settled_names(route=route, genai_span=genai_span),
+            settled_prefixes={DB_SPAN_PREFIX},
+        )
+        _assert_complete_trace(hits, route=route, genai_span=genai_span)
+
+        genai_spans = [span for span in hits[0].spans if span.operation_name == genai_span]
+        assert len(genai_spans) == 1, (
+            f"a streamed call must produce exactly ONE gen-AI span, got {len(genai_spans)}; "
+            f"spans: {hits[0].span_names()}"
+        )
+        assert _tag(genai_spans[0], "litellm.request.streaming") is True, (
+            "the gen-AI span must record litellm.request.streaming=true; its absence means "
+            "the stream flag was dropped before the model call"
+        )
+
+    @pytest.mark.covers("logging.otel.stream.exports_metric", exercised_on=["responses"])
+    def test_responses_stream_exports_complete_trace(
+        self, client: LoggingClient, otel_reader: OtelReader, resources: ResourceManager
+    ) -> None:
+        """One successful STREAMED /v1/responses call must export ONE complete
+        OTEL trace: a single root SERVER span with auth/db/cost children and
+        the gen-AI CLIENT span connected back to the root, plus the stream
+        actually delivering events and exactly ONE gen-AI span for the call.
+
+        Two knowingly relaxed assertions on this surface, both verified against
+        live traces and both tracked in LIT-4428: the responses route
+        does not stamp litellm.request.streaming on the gen-AI span, and the
+        spend write for a streamed responses call records spend correctly but
+        emits no batch_write_to_db cost span (streamed chat/messages and
+        non-streamed responses all emit it). Streaming is instead proven from
+        the response side (event-stream content type, chunks consumed). When
+        the product closes either gap, tighten this test to match the sibling
+        assertions.
+        """
+        route = "/v1/responses"
+        _assert_otel_destination_configured(client)
+
+        key = client.key_with_alias(
+            f"otel-stream-responses-{unique_marker()}", models=[CHEAP_OPENAI_MODEL]
+        )
+        resources.defer(lambda: client.delete_key(key))
+
+        marker = unique_marker()
+        outcome = _first_ok(
+            client,
+            lambda: client.responses_raw(key, CHEAP_OPENAI_MODEL, f"reply with one word {marker}", stream=True),
+        )
+        assert outcome.call_id is not None, "success response must carry x-litellm-call-id"
+        assert outcome.is_streaming, f"response must be an event stream, got content-type {outcome.content_type!r}"
+        assert outcome.chunks > 0, "the stream must deliver at least one event"
+
+        genai_span = f"chat {CHEAP_OPENAI_MODEL}"
+        hits = otel_reader.poll_traces_for_call(
+            call_id=outcome.call_id,
+            settled_names=_settled_names(route=route, genai_span=genai_span, require_cost_span=False),
+            settled_prefixes={DB_SPAN_PREFIX},
+        )
+        _assert_complete_trace(hits, route=route, genai_span=genai_span, require_cost_span=False)
+
+        genai_spans = [span for span in hits[0].spans if span.operation_name == genai_span]
+        assert len(genai_spans) == 1, (
+            f"a streamed call must produce exactly ONE gen-AI span, got {len(genai_spans)}; "
+            f"spans: {hits[0].span_names()}"
+        )

tests/e2e/models.py

@@ -153,6 +153,7 @@ class AnthropicMessagesBody(BaseModel):
     model: str
     messages: list[ChatMessage]
     max_tokens: int
+    stream: bool | None = None
 
 
 class AnthropicMessagesResponse(BaseModel):