test(e2e): datadog log delivery for successful chat, messages, and responses (LIT-4447) (#33415)

* test(e2e): datadog log delivery for successful chat, messages, and responses

Covers logging.datadog.success.exports_metric on all three routes: one
successful non-streaming call must reach the DataDog logs intake as exactly
one log event whose StandardLoggingPayload message carries the model group,
real token counts, and a response cost equal to the x-litellm-response-cost
header of the same response. Delivery is judged at the intake: the compose
stack gains a dd-sink service recording every batch the datadog callback
ships via the DD_BASE_URL testing override, and a typed reader replays it.

Writing these caught a live product bug: /v1/messages double-logs every
success (two byte-identical events per call), filed as LIT-4447; the messages
test tolerates byte-identical duplicates of the one event until it lands,
while a second differing event still fails

* test(e2e): address review findings on the datadog delivery suite

Consolidates the fresh-key first_ok helper into logging_client now that the
otel PR it mirrored has merged (both test files use the shared copy), moves
intake batch parsing into a helper so no path can leave the batch unbound,
and gives the sink's /health endpoint a truthful text/plain content type

* test(e2e): tolerate same-logical-event duplicates by call id, not byte identity

A clean LIT-4447 repro showed the duplicated payload is built twice and can
mint a fresh synthetic completion id per emission, arriving as two separate
intake POSTs with the same litellm_call_id and identical substantive fields.
Byte-identity was therefore a flaky criterion; duplicates now qualify only
when they share the call id, call type, model group, tokens, and cost, and a
second differing event still fails

* test(e2e): assert the scenario strictly; the messages test is the LIT-4447 regression pin

Per review direction the tests now assert exactly what the scenario promises:
exactly one DataDog log event per successful call, on every route. The
/v1/messages test therefore fails on current code against the known
double-log (LIT-4447) and is its regression pin; it goes green when the fix
lands. The duplicate-tolerance machinery is removed

* Simplify docstrings for DataDog log tests

Removed redundant phrasing about cost cross-checking in docstrings.

* Update test_datadog_log_e2e.py

tests/e2e/coverage_registry/logging.yaml

@@ -5,7 +5,7 @@
 - {id: logging.s3.success.writes_object, module: logging, tier: P0, event: success, assertions: [writes_object], exercised_on: [chat_completions, messages, embeddings], source: "integrations/s3_v2.py", rationale: "Primary audit trail; batch flush no-drop"}
 - {id: logging.s3.failure.writes_object, module: logging, tier: P0, event: failure, assertions: [writes_object], exercised_on: [chat_completions, messages], source: "integrations/s3_v2.py", rationale: "Failed calls persisted for compliance"}
 - {id: logging.gcs_bucket.success.writes_object, module: logging, tier: P0, event: success, assertions: [writes_object], exercised_on: [chat_completions, messages, embeddings], source: "integrations/gcs_bucket/gcs_bucket.py", rationale: "GCS parallel to S3"}
-- {id: logging.datadog.success.exports_metric, module: logging, tier: P0, event: success, assertions: [exports_metric], exercised_on: [chat_completions, messages, embeddings], source: "integrations/datadog/datadog.py", rationale: "Powers dashboards/alerts; cardinality regressions common"}
+- {id: logging.datadog.success.exports_metric, module: logging, tier: P0, event: success, assertions: [exports_metric], exercised_on: [chat_completions, messages, responses, embeddings], source: "integrations/datadog/datadog.py", rationale: "Powers dashboards/alerts; cardinality regressions common"}
 - {id: logging.datadog.failure.exports_metric, module: logging, tier: P0, event: failure, assertions: [exports_metric], exercised_on: [chat_completions], source: "integrations/datadog/datadog.py", rationale: "Failure metrics for alerting/SLO"}
 - {id: logging.prometheus.success.exports_metric, module: logging, tier: P0, event: success, assertions: [exports_metric], exercised_on: [chat_completions, messages, embeddings], source: "integrations/prometheus.py", rationale: "Standard OSS metrics; per-key cardinality (existing e2e)"}
 - {id: logging.otel.success.exports_metric, module: logging, tier: P0, event: success, assertions: [exports_metric], exercised_on: [chat_completions, messages, responses, embeddings], source: "integrations/otel/logger.py", rationale: "OTEL spans on every call path"}

tests/e2e/docker-compose.yml

@@ -1,5 +1,41 @@
 # local setup to run e2e tests
 configs:
+  dd_sink_script:
+    content: |
+      # Minimal DataDog logs-intake sink for the logging suite: records every
+      # POST (gunzipping the compressed batches the integration sends) and
+      # replays them as JSON on GET /requests so tests can assert delivery.
+      import gzip, json
+      from http.server import BaseHTTPRequestHandler, HTTPServer
+
+      REQUESTS = []
+
+      class Handler(BaseHTTPRequestHandler):
+          def do_POST(self):
+              body = self.rfile.read(int(self.headers.get("Content-Length", 0)))
+              if self.headers.get("Content-Encoding") == "gzip":
+                  body = gzip.decompress(body)
+              REQUESTS.append({"path": self.path, "body": body.decode("utf-8", "replace")})
+              self.send_response(202)
+              self.end_headers()
+              self.wfile.write(b"{}")
+
+          def do_GET(self):
+              self.send_response(200)
+              if self.path == "/health":
+                  self.send_header("Content-Type", "text/plain")
+                  self.end_headers()
+                  self.wfile.write(b"ok")
+                  return
+              self.send_header("Content-Type", "application/json")
+              self.end_headers()
+              self.wfile.write(json.dumps({"requests": REQUESTS}).encode())
+
+          def log_message(self, *args):
+              pass
+
+      HTTPServer(("0.0.0.0", 8080), Handler).serve_forever()
+
   litellm_config:
     content: |
       general_settings:
@@ -23,7 +59,7 @@ configs:
         # (PHOENIX_COLLECTOR_HTTP_ENDPOINT below points it at the jaeger service),
         # so gen-AI spans export through a preset-owned provider - the code path
         # where trace splits actually happen - with no cloud credentials needed.
-        callbacks: ["arize_phoenix"]
+        callbacks: ["arize_phoenix", "datadog"]
 
       router_settings:
         routing_strategy: simple-shuffle
@@ -93,10 +129,15 @@ services:
         condition: service_healthy
       jaeger:
         condition: service_healthy
+      dd-sink:
+        condition: service_healthy
     env_file: .env
     environment:
       LITELLM_MASTER_KEY: sk-1234
       STORE_MODEL_IN_DB: "True"
+      DD_API_KEY: local-sink-noauth
+      DD_SITE: datadoghq.com
+      DD_BASE_URL: http://dd-sink:8080
       LITELLM_OTEL_V2: "true"
       PHOENIX_COLLECTOR_HTTP_ENDPOINT: http://jaeger:4318/v1/traces
       PHOENIX_API_KEY: local-jaeger-noauth
@@ -155,3 +196,19 @@ services:
       interval: 3s
       timeout: 3s
       retries: 20
+
+# throwaway DataDog logs-intake sink (records POSTs, replays on GET /requests;
+# see E2E_DD_SINK_URL)
+  dd-sink:
+    image: python:3.12-alpine
+    command: ["python", "/sink.py"]
+    configs:
+      - source: dd_sink_script
+        target: /sink.py
+    ports:
+      - "9915:8080"
+    healthcheck:
+      test: ["CMD", "wget", "-qO-", "http://127.0.0.1:8080/health"]
+      interval: 3s
+      timeout: 3s
+      retries: 20

tests/e2e/e2e_config.py

@@ -32,6 +32,10 @@ CHEAP_OPENAI_MODEL = os.environ.get("E2E_CHEAP_OPENAI_MODEL", "gpt-5.5")
 # read exported spans back through it.
 OTEL_QUERY_URL = os.environ.get("E2E_OTEL_QUERY_URL", "http://localhost:16686").rstrip("/")
 
+# Query URL of the compose stack's DataDog logs-intake sink (the `dd-sink`
+# service records every intake POST and replays them on GET /requests).
+DD_SINK_URL = os.environ.get("E2E_DD_SINK_URL", "http://localhost:9915").rstrip("/")
+
 # Writes on the proxy are eventually consistent (e.g. spend rows flush on
 # proxy_batch_write_at, ~60s). Read-backs poll to this deadline, never sleep-once.
 POLL_TIMEOUT = float(os.environ.get("E2E_POLL_TIMEOUT", "120"))

tests/e2e/logging/conftest.py

@@ -11,6 +11,7 @@ import os
 import pytest
 
 from logging_client import LangfuseCreds, LoggingClient, build_logging_client, load_langfuse_creds
+from datadog_sink import DdSinkReader, build_dd_sink_reader
 from otel_client import OtelReader, build_otel_reader
 
 
@@ -35,6 +36,12 @@ def otel_reader() -> OtelReader:
     return build_otel_reader()
 
 
+@pytest.fixture(scope="session")
+def dd_sink() -> DdSinkReader:
+    """Read-back client for the compose stack's DataDog logs-intake sink."""
+    return build_dd_sink_reader()
+
+
 @pytest.fixture
 def datadog_creds() -> None:
     """Require Datadog shipping credentials. Hard-fail when absent; never skip."""

tests/e2e/logging/datadog_sink.py

@@ -0,0 +1,103 @@
+"""Read-back for the DataDog logging tests: typed models over the compose
+stack's dd-sink service, which records every logs-intake POST the datadog
+callback sends (gunzipped) and replays them as JSON.
+
+Delivery is judged on what the sink actually received, mirroring how the OTEL
+tests read Jaeger; a failed sink query is a hard failure, never an empty
+result. External reads go through ``e2e_http``.
+"""
+
+from __future__ import annotations
+
+import time
+from dataclasses import dataclass
+
+import pytest
+from pydantic import BaseModel, ConfigDict, TypeAdapter, ValidationError
+
+from e2e_config import DD_SINK_URL, POLL_INTERVAL, POLL_TIMEOUT
+from e2e_http import URL, NoBody, Success, get
+
+
+class DdSinkRequest(BaseModel):
+    model_config = ConfigDict(extra="ignore")
+
+    path: str
+    body: str
+
+
+class DdSinkRequests(BaseModel):
+    model_config = ConfigDict(extra="ignore")
+
+    requests: list[DdSinkRequest] = []
+
+
+class DdLogEvent(BaseModel):
+    model_config = ConfigDict(extra="ignore")
+
+    message: str
+    ddsource: str | None = None
+    service: str | None = None
+    status: str | None = None
+
+
+_EVENT_BATCH: TypeAdapter[list[DdLogEvent]] = TypeAdapter(list[DdLogEvent])
+
+
+def _parse_batch(request: DdSinkRequest) -> list[DdLogEvent]:
+    """The intake accepts an array of events or a single event object."""
+    try:
+        return _EVENT_BATCH.validate_json(request.body)
+    except ValidationError:
+        try:
+            return [DdLogEvent.model_validate_json(request.body)]
+        except ValidationError:
+            pytest.fail(f"dd-sink recorded a non-log body on {request.path}: {request.body[:200]}")
+
+
+@dataclass(frozen=True, slots=True)
+class DdSinkReader:
+    sink_url: str
+
+    def _recorded_requests(self) -> list[DdSinkRequest]:
+        result = get(
+            URL(f"{self.sink_url}/requests"),
+            headers=NoBody(),
+            params=NoBody(),
+            response_type=DdSinkRequests,
+            timeout=30.0,
+        )
+        match result:
+            case Success(data=page):
+                return page.requests
+            case failure:
+                pytest.fail(f"dd-sink query at {self.sink_url} failed: {failure}")
+
+    def events_for_marker(self, marker: str) -> list[DdLogEvent]:
+        """Every log event across every recorded intake batch whose message
+        carries the marker. More than one hit for one call IS the
+        duplicate-delivery bug, so this never collapses to a single event."""
+        events: list[DdLogEvent] = []
+        for request in self._recorded_requests():
+            if "/api/v2/logs" not in request.path:
+                continue
+            events.extend(event for event in _parse_batch(request) if marker in event.message)
+        return events
+
+    def poll_events_for_marker(self, marker: str) -> list[DdLogEvent]:
+        """Poll until at least one matching event lands (the callback flushes
+        in periodic batches), then re-read after one more interval so a late
+        duplicate cannot hide from the exactly-one assertion. At the deadline
+        the last result is returned as-is."""
+        deadline = time.monotonic() + POLL_TIMEOUT
+        while time.monotonic() < deadline:
+            events = self.events_for_marker(marker)
+            if events:
+                time.sleep(POLL_INTERVAL)
+                return self.events_for_marker(marker)
+            time.sleep(POLL_INTERVAL)
+        return self.events_for_marker(marker)
+
+
+def build_dd_sink_reader() -> DdSinkReader:
+    return DdSinkReader(sink_url=DD_SINK_URL)

tests/e2e/logging/logging_client.py

@@ -18,7 +18,7 @@ import json
 import os
 import time
 from dataclasses import dataclass
-from typing import Literal
+from typing import Callable, Literal
 
 import pytest
 from pydantic import BaseModel, ConfigDict, Field, JsonValue, TypeAdapter, ValidationError
@@ -28,6 +28,7 @@ from e2e_gateway import Gateway, build_gateway
 from e2e_http import (
     URL,
     AuthHeaders,
+    require_successful_call,
     NoBody,
     StreamingResponse,
     Success,
@@ -617,5 +618,20 @@ class LoggingClient:
         return self.list_langfuse_observations(creds, trace_id=gen.trace_id) or [gen]
 
 
+def first_ok(client: LoggingClient, send: Callable[[], StreamingResponse]) -> StreamingResponse:
+    """First successful call on a fresh key. A fresh key may briefly 401 until
+    the data plane's auth cache picks it up, so retry on 401 to a deadline; a
+    401 is rejected before the LLM call, so it cannot contaminate delivery or
+    trace assertions. Any other failure is behavior under test and fails hard."""
+    deadline = time.monotonic() + client.gateway.poll_timeout
+    while True:
+        outcome = send()
+        if outcome.ok:
+            return outcome
+        if outcome.status_code != 401 or time.monotonic() >= deadline:
+            require_successful_call(outcome)
+        time.sleep(client.gateway.poll_interval)
+
+
 def build_logging_client() -> LoggingClient:
     return LoggingClient(gateway=build_gateway())

tests/e2e/logging/test_datadog_log_e2e.py

@@ -0,0 +1,162 @@
+"""Live e2e: DataDog log delivery for successful non-streaming calls.
+
+Covers logging.datadog.success.exports_metric: one successful call on each
+route must reach the DataDog logs intake as EXACTLY ONE log event whose
+message (the StandardLoggingPayload) carries the model, the token counts, and
+the response cost. Delivery is judged on what the intake actually received:
+the compose stack's dd-sink service records every batch the datadog callback
+ships (DD_BASE_URL override) and the tests read it back, so a dropped event, a
+duplicated event, or a payload missing the cost all fail here.
+
+Both halves of the contract are asserted: the recorded state (the proxy
+reports the DataDogLogger callback active via /health/readiness/details) and
+the enforced behavior (the event at the intake, with the cost cross-checked
+exactly against the x-litellm-response-cost header of the very response the
+caller received).
+"""
+
+from __future__ import annotations
+
+import pytest
+from pydantic import BaseModel, ConfigDict
+
+from datadog_sink import DdLogEvent, DdSinkReader
+from e2e_config import CHEAP_ANTHROPIC_MODEL, CHEAP_OPENAI_MODEL, unique_marker
+from e2e_http import NoBody, StreamingResponse
+from lifecycle import ResourceManager
+from logging_client import LoggingClient, first_ok
+
+pytestmark = pytest.mark.e2e
+
+#: The active DataDog callback's name in /health/readiness/details success_callbacks.
+DD_LOGGER_NAME = "DataDogLogger"
+
+
+class _DdMessagePayload(BaseModel):
+    """The fields of the StandardLoggingPayload the scenario pins."""
+
+    model_config = ConfigDict(extra="ignore")
+
+    model_group: str
+    total_tokens: int
+    response_cost: float
+    status: str
+    call_type: str
+
+
+def _assert_datadog_configured(client: LoggingClient) -> None:
+    """Recorded state: the proxy reports the DataDog callback among its active
+    callbacks, so a missing destination config fails here, before any
+    delivery-based assertion can time out confusingly."""
+    result = client.gateway.probe("/health/readiness/details", params=NoBody())
+    assert result.status_code == 200, (
+        f"/health/readiness/details must answer 200, got {result.status_code}: {result.body[:300]}"
+    )
+    assert DD_LOGGER_NAME in result.body, (
+        f"the proxy must report the {DD_LOGGER_NAME} callback active "
+        f"(callbacks + DD_* env in the compose config); got: {result.body[:400]}"
+    )
+
+
+def _assert_exactly_one_event(
+    events: list[DdLogEvent], *, model_group: str, call_type: str, outcome: StreamingResponse
+) -> None:
+    """The enforced behavior: the intake holds exactly one event for the call,
+    sourced from litellm, whose payload names the model group and call type,
+    counts real tokens, and carries the same cost the response header reported."""
+    assert events, "no DataDog log event for this call reached the intake within the deadline"
+    assert len(events) == 1, (
+        f"expected exactly ONE DataDog log event for the call, got {len(events)} - "
+        "more than one event for one call is the duplicate-delivery bug (see LIT-4447 "
+        "for the currently known /v1/messages instance)"
+    )
+    event = events[0]
+    assert event.ddsource == "litellm", f"event ddsource must be litellm, got {event.ddsource!r}"
+    assert event.status == "info", f"success events ship at status info, got {event.status!r}"
+
+    payload = _DdMessagePayload.model_validate_json(event.message)
+    assert payload.status == "success", f"payload status must be success, got {payload.status!r}"
+    assert payload.model_group == model_group, (
+        f"payload model_group must be {model_group!r}, got {payload.model_group!r}"
+    )
+    assert payload.call_type == call_type, (
+        f"payload call_type must be {call_type!r}, got {payload.call_type!r}"
+    )
+    assert payload.total_tokens > 0, f"payload must count real tokens, got {payload.total_tokens}"
+    assert outcome.response_cost is not None and outcome.response_cost > 0, (
+        f"the response must report x-litellm-response-cost, got {outcome.response_cost!r}"
+    )
+    assert abs(payload.response_cost - outcome.response_cost) < 1e-12, (
+        f"payload response_cost {payload.response_cost} must equal the response header "
+        f"cost {outcome.response_cost}"
+    )
+
+
+class TestDataDogLogDelivery:
+    @pytest.mark.covers("logging.datadog.success.exports_metric", exercised_on=["chat_completions"])
+    def test_chat_completions_emits_one_log_event(
+        self, client: LoggingClient, dd_sink: DdSinkReader, resources: ResourceManager
+    ) -> None:
+        """One successful non-streaming /chat/completions call must reach the
+        DataDog logs intake as exactly one log event whose payload carries the
+        model, the token counts, and the response cost."""
+        _assert_datadog_configured(client)
+
+        key = client.key_with_alias(f"dd-chat-{unique_marker()}", models=[CHEAP_ANTHROPIC_MODEL])
+        resources.defer(lambda: client.delete_key(key))
+
+        marker = unique_marker()
+        outcome = first_ok(
+            client,
+            lambda: client.chat_raw(key, CHEAP_ANTHROPIC_MODEL, f"reply with one word {marker}", max_tokens=16),
+        )
+        events = dd_sink.poll_events_for_marker(marker)
+        _assert_exactly_one_event(
+            events, model_group=CHEAP_ANTHROPIC_MODEL, call_type="acompletion", outcome=outcome
+        )
+
+    @pytest.mark.covers("logging.datadog.success.exports_metric", exercised_on=["messages"])
+    def test_messages_emits_one_log_event(
+        self, client: LoggingClient, dd_sink: DdSinkReader, resources: ResourceManager
+    ) -> None:
+        """One successful non-streaming /v1/messages call must reach the
+        DataDog logs intake as exactly one log event whose payload carries the
+        model, the token counts, and the response cost.
+
+        This currently fails on the known /v1/messages double-log (LIT-4447); it goes green when the fix lands."""
+        _assert_datadog_configured(client)
+
+        key = client.key_with_alias(f"dd-messages-{unique_marker()}", models=[CHEAP_ANTHROPIC_MODEL])
+        resources.defer(lambda: client.delete_key(key))
+
+        marker = unique_marker()
+        outcome = first_ok(
+            client,
+            lambda: client.messages_raw(key, CHEAP_ANTHROPIC_MODEL, f"reply with one word {marker}", max_tokens=16),
+        )
+        events = dd_sink.poll_events_for_marker(marker)
+        _assert_exactly_one_event(
+            events, model_group=CHEAP_ANTHROPIC_MODEL, call_type="anthropic_messages", outcome=outcome
+        )
+
+    @pytest.mark.covers("logging.datadog.success.exports_metric", exercised_on=["responses"])
+    def test_responses_emits_one_log_event(
+        self, client: LoggingClient, dd_sink: DdSinkReader, resources: ResourceManager
+    ) -> None:
+        """One successful non-streaming /v1/responses call must reach the
+        DataDog logs intake as exactly one log event whose payload carries the
+        model, the token counts, and the response cost."""
+        _assert_datadog_configured(client)
+
+        key = client.key_with_alias(f"dd-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}"),
+        )
+        events = dd_sink.poll_events_for_marker(marker)
+        _assert_exactly_one_event(
+            events, model_group=CHEAP_OPENAI_MODEL, call_type="aresponses", outcome=outcome
+        )

tests/e2e/logging/test_otel_trace_e2e.py

@@ -18,15 +18,14 @@ destination's own query API - never proxy-side "export succeeded" logs).
 from __future__ import annotations
 
 import time
-from collections.abc import Callable
 
 import pytest
 from pydantic import BaseModel, ConfigDict, ValidationError
 
 from e2e_config import CHEAP_ANTHROPIC_MODEL, CHEAP_OPENAI_MODEL, unique_marker
-from e2e_http import NoBody, StreamingResponse, require_successful_call
+from e2e_http import NoBody
 from lifecycle import ResourceManager
-from logging_client import INVALID_UPSTREAM_API_KEY, LoggingClient
+from logging_client import INVALID_UPSTREAM_API_KEY, LoggingClient, first_ok
 from models import LiteLLMParamsBody
 from otel_client import JaegerSpan, JaegerTrace, OtelReader
 
@@ -60,22 +59,6 @@ def _assert_otel_destination_configured(client: LoggingClient) -> None:
     )
 
 
-def _first_ok(client: LoggingClient, send: Callable[[], StreamingResponse]) -> StreamingResponse:
-    """First successful call on a fresh key. A fresh key may briefly 401 until
-    the data plane's auth cache picks it up, so retry on 401 to a deadline; a
-    401 is rejected before the LLM call so it exports no gen-AI span and cannot
-    contaminate the trace assertions. Any other failure is behavior under test
-    and fails hard."""
-    deadline = time.monotonic() + client.gateway.poll_timeout
-    while True:
-        outcome = send()
-        if outcome.ok:
-            return outcome
-        if outcome.status_code != 401 or time.monotonic() >= deadline:
-            require_successful_call(outcome)
-        time.sleep(client.gateway.poll_interval)
-
-
 def _parent_ids(span_id: str, trace: JaegerTrace) -> list[str]:
     span = next(s for s in trace.spans if s.span_id == span_id)
     return [ref.span_id for ref in span.references if ref.ref_type == "CHILD_OF"]
@@ -253,7 +236,7 @@ class TestOtelTraceCompleteness:
         resources.defer(lambda: client.delete_key(key))
 
         marker = unique_marker()
-        outcome = _first_ok(
+        outcome = first_ok(
             client, lambda: client.chat_raw(key, MODEL, f"reply with one word {marker}", max_tokens=16)
         )
         assert outcome.call_id is not None, "success response must carry x-litellm-call-id"
@@ -286,7 +269,7 @@ class TestOtelTraceCompleteness:
         resources.defer(lambda: client.delete_key(key))
 
         marker = unique_marker()
-        outcome = _first_ok(
+        outcome = first_ok(
             client, lambda: client.messages_raw(key, MODEL, f"reply with one word {marker}", max_tokens=16)
         )
         assert outcome.call_id is not None, "success response must carry x-litellm-call-id"
@@ -319,7 +302,7 @@ class TestOtelTraceCompleteness:
         resources.defer(lambda: client.delete_key(key))
 
         marker = unique_marker()
-        outcome = _first_ok(
+        outcome = first_ok(
             client,
             lambda: client.responses_raw(key, CHEAP_OPENAI_MODEL, f"reply with one word {marker}"),
         )
@@ -360,7 +343,7 @@ class TestOtelTraceCompleteness:
         resources.defer(lambda: client.delete_key(key))
 
         marker = unique_marker()
-        outcome = _first_ok(
+        outcome = first_ok(
             client,
             lambda: client.chat_raw(key, MODEL, f"reply with one word {marker}", stream=True, max_tokens=16),
         )
@@ -416,7 +399,7 @@ class TestOtelTraceCompleteness:
         resources.defer(lambda: client.delete_key(key))
 
         marker = unique_marker()
-        outcome = _first_ok(
+        outcome = first_ok(
             client,
             lambda: client.messages_raw(key, MODEL, f"reply with one word {marker}", max_tokens=16, stream=True),
         )
@@ -474,7 +457,7 @@ class TestOtelTraceCompleteness:
         resources.defer(lambda: client.delete_key(key))
 
         marker = unique_marker()
-        outcome = _first_ok(
+        outcome = first_ok(
             client,
             lambda: client.responses_raw(key, CHEAP_OPENAI_MODEL, f"reply with one word {marker}", stream=True),
         )