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()}" + )