From ac6cf1fce6411acb7e23b2a257c9a126d81e8539 Mon Sep 17 00:00:00 2001 From: Yucheng Zhu Date: Tue, 14 Jul 2026 10:17:24 -0700 Subject: [PATCH 1/3] test(e2e): OTEL trace completeness on streaming /v1/messages Covers the messages surface of logging.otel.stream.exports_metric: same tree contract as the non-streaming tests plus the stream-specific assertions (the response actually streamed, exactly one gen-AI span, and the span records litellm.request.streaming=true). Adds a stream field to the shared AnthropicMessagesBody and a stream mode to messages_raw --- tests/e2e/logging/logging_client.py | 27 +++++++++----- tests/e2e/logging/test_otel_trace_e2e.py | 47 ++++++++++++++++++++++++ tests/e2e/models.py | 1 + 3 files changed, 65 insertions(+), 10 deletions(-) diff --git a/tests/e2e/logging/logging_client.py b/tests/e2e/logging/logging_client.py index bffdf71ed80..a70a4a6aec2 100644 --- a/tests/e2e/logging/logging_client.py +++ b/tests/e2e/logging/logging_client.py @@ -464,17 +464,24 @@ 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=stream or 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( diff --git a/tests/e2e/logging/test_otel_trace_e2e.py b/tests/e2e/logging/test_otel_trace_e2e.py index 7478402da1a..ebfdbe5fa48 100644 --- a/tests/e2e/logging/test_otel_trace_e2e.py +++ b/tests/e2e/logging/test_otel_trace_e2e.py @@ -315,3 +315,50 @@ 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" + ) diff --git a/tests/e2e/models.py b/tests/e2e/models.py index c8fc6c1ad4d..b82275ecde3 100644 --- a/tests/e2e/models.py +++ b/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): From 183bfb1a0542d079405ea714ac77de04db6c6b66 Mon Sep 17 00:00:00 2001 From: Yucheng Zhu Date: Tue, 14 Jul 2026 13:36:48 -0700 Subject: [PATCH 2/3] test(e2e): make the stream field coercion explicit per review --- tests/e2e/logging/logging_client.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/e2e/logging/logging_client.py b/tests/e2e/logging/logging_client.py index a70a4a6aec2..3959eab163d 100644 --- a/tests/e2e/logging/logging_client.py +++ b/tests/e2e/logging/logging_client.py @@ -474,7 +474,7 @@ def messages_raw( model=model, max_tokens=max_tokens, messages=[ChatMessage(role="user", content=text)], - stream=stream or None, + stream=True if stream else None, ) if stream: return self.gateway.transport.stream( From a712e86a846a201295d79196382bedfca69bce53 Mon Sep 17 00:00:00 2001 From: yucheng-berri Date: Tue, 14 Jul 2026 15:42:10 -0700 Subject: [PATCH 3/3] test(e2e): otel trace completeness on streaming /v1/responses (LIT-3787) (#33262) * test(e2e): OTEL trace completeness on streaming /v1/responses Covers the responses surface of logging.otel.stream.exports_metric: same tree contract plus the stream-side assertions (event-stream content type, chunks consumed, exactly one gen-AI span). Two assertions are knowingly relaxed on this surface, both verified against live traces and 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. Adds a stream mode to responses_raw * test(e2e): parenthesize the settle-names ternary and make stream coercion explicit per review --- tests/e2e/logging/logging_client.py | 23 ++++++--- tests/e2e/logging/test_otel_trace_e2e.py | 62 ++++++++++++++++++++++-- 2 files changed, 73 insertions(+), 12 deletions(-) diff --git a/tests/e2e/logging/logging_client.py b/tests/e2e/logging/logging_client.py index 3959eab163d..e37d6175705 100644 --- a/tests/e2e/logging/logging_client.py +++ b/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): @@ -485,16 +486,22 @@ def messages_raw( ) 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: diff --git a/tests/e2e/logging/test_otel_trace_e2e.py b/tests/e2e/logging/test_otel_trace_e2e.py index ebfdbe5fa48..434b3c4f3ed 100644 --- a/tests/e2e/logging/test_otel_trace_e2e.py +++ b/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: @@ -362,3 +366,53 @@ def test_messages_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=["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()}" + )