test: e2e staging leftovers

litellm/router_strategy/complexity_router/complexity_router.py

@@ -56,11 +56,13 @@ class TierClassification(BaseModel):
 
 _CLASSIFICATION_PROMPT_TEMPLATE = """Classify the complexity of the following user request into exactly one tier.
 
+Judge the intellectual difficulty of answering correctly, not how short the request is.
+
 Tiers:
-- SIMPLE: factual lookups, greetings, short direct questions with no reasoning or code involved.
-- MEDIUM: everyday requests needing some explanation or minor code/technical content.
-- COMPLEX: requests involving non-trivial code, architecture, or multi-step technical work.
-- REASONING: requests explicitly requiring step-by-step reasoning, analysis, or weighing tradeoffs.
+- SIMPLE: greetings, chitchat, or factual lookups with a short known answer. Do not use SIMPLE for unsolved problems, proofs, deep theory, multi-step analysis, or non-trivial code, even if the request is only one sentence.
+- MEDIUM: everyday requests that need some explanation, light reasoning, or minor code/technical content.
+- COMPLEX: non-trivial code, architecture, multi-step technical work, or specialized domain depth.
+- REASONING: open-ended analysis, proofs, famous hard problems, step-by-step reasoning, tradeoffs, or anything where a correct answer requires careful thought rather than a quick lookup.
 
 {system_context}Request:
 {prompt}"""

tests/e2e/docker-compose.yml

@@ -1,41 +1,5 @@
 # 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:
@@ -129,15 +93,16 @@ 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
+      # Real DataDog delivery (no local sink): the key comes from the
+      # environment - the cluster's secret manager injects it, locally
+      # tests/e2e/.env provides it. Tests read delivery back via the DataDog
+      # Logs Search API (DD_APP_KEY, test-side only - see logging/datadog_reader.py).
+      DD_API_KEY: ${DD_API_KEY:-}
+      DD_SITE: ${DD_SITE:-datadoghq.com}
       LITELLM_OTEL_V2: "true"
       PHOENIX_COLLECTOR_HTTP_ENDPOINT: http://jaeger:4318/v1/traces
       PHOENIX_API_KEY: local-jaeger-noauth
@@ -198,19 +163,3 @@ 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

@@ -10,20 +10,21 @@
 PROXY_BASE_URL = os.environ.get("LITELLM_PROXY_URL", "http://localhost:4000").rstrip("/")
 MASTER_KEY = os.environ.get("LITELLM_MASTER_KEY", "sk-1234")
 
-# Control-plane (management/admin) base URL. In a split control-plane/data-plane
-# deployment the LLM data plane (PROXY_BASE_URL: /chat, /embeddings, native
-# passthrough) and the management API (keys, users, teams, orgs, budgets, spend,
-# model info, /openapi.json) are served by *different* services. The suite drives
-# both through one Transport that routes by path (see transport.SplitTransport).
-# Defaults to PROXY_BASE_URL so a monolithic proxy serving everything on one URL
-# behaves exactly as before.
+# Control-plane (management/admin) base URL. Defaults to PROXY_BASE_URL so a
+# single path-routing host (stage ALB, compose monolith) works for both planes.
+# Set LITELLM_CONTROL_PLANE_URL only when management is a different base than
+# the LLM host and you are not going through an ingress that path-routes.
 CONTROL_PLANE_BASE_URL = os.environ.get(
     "LITELLM_CONTROL_PLANE_URL", PROXY_BASE_URL
 ).rstrip("/")
 
 UI_USERNAME = os.environ.get("E2E_UI_USERNAME", "admin")
 UI_PASSWORD = os.environ.get("E2E_UI_PASSWORD", MASTER_KEY)
 
+# Dashboard base for playwright. Defaults to PROXY_BASE_URL so one ALB/monolith
+# host covers /ui as well. Override E2E_UI_BASE_URL only if the UI is elsewhere.
+UI_BASE_URL = os.environ.get("E2E_UI_BASE_URL", PROXY_BASE_URL).rstrip("/")
+
 CHEAP_ANTHROPIC_MODEL = os.environ.get("E2E_CHEAP_ANTHROPIC_MODEL", "claude-haiku-4-5")
 CHEAP_OPENAI_MODEL = os.environ.get("E2E_CHEAP_OPENAI_MODEL", "gpt-5.5")
 
@@ -32,9 +33,19 @@
 # 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("/")
+# Real-DataDog read-back (no local sink - destination fakes cannot be deployed
+# on the cluster): the proxy delivers with DD_API_KEY as in production, and the
+# tests read ingested events back through the DataDog Logs Search API, which
+# additionally needs an application key. On the cluster the secret manager
+# injects both; locally tests/e2e/.env provides them.
+DD_SITE = os.environ.get("DD_SITE", "datadoghq.com").strip()
+DD_API_KEY = os.environ.get("DD_API_KEY", "").strip()
+DD_APP_KEY = os.environ.get("DD_APP_KEY", "").strip()
+# After the first event is searchable, keep watching this long for a late
+# duplicate before the exactly-one assertion: real-DataDog ingestion jitter can
+# make one call's two events searchable tens of seconds apart, and a duplicate
+# that surfaces late IS the bug (LIT-4447), so one poll interval is not enough.
+DD_SETTLE_SECONDS = float(os.environ.get("E2E_DD_SETTLE_SECONDS", "30"))
 
 # 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.

tests/e2e/logging/conftest.py

@@ -11,7 +11,7 @@
 import pytest
 
 from logging_client import LangfuseCreds, LoggingClient, build_logging_client, load_langfuse_creds
-from datadog_sink import DdSinkReader, build_dd_sink_reader
+from datadog_reader import DdLogsReader, build_dd_logs_reader
 from otel_client import OtelReader, build_otel_reader
 
 
@@ -37,9 +37,10 @@ def otel_reader() -> OtelReader:
 
 
 @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()
+def dd_logs() -> DdLogsReader:
+    """Read-back client for the real DataDog Logs Search API (keys from the
+    secret manager on the cluster, tests/e2e/.env locally)."""
+    return build_dd_logs_reader()
 
 
 @pytest.fixture

tests/e2e/logging/datadog_reader.py

@@ -0,0 +1,140 @@
+"""Read-back for the DataDog logging tests against the real DataDog Logs
+Search API.
+
+Delivery is judged on what DataDog itself ingested: the proxy ships logs with
+DD_API_KEY exactly as in production (no base-URL override, no local sink), and
+the tests search the ingested events back with POST /api/v2/logs/events/search,
+authenticated with the same DD_API_KEY plus a DD_APP_KEY application key. On
+the cluster the secret manager injects both keys; locally tests/e2e/.env
+provides them. Missing keys or a failed search call are hard failures, 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, Field
+
+from e2e_config import (
+    DD_API_KEY,
+    DD_APP_KEY,
+    DD_SETTLE_SECONDS,
+    DD_SITE,
+    POLL_INTERVAL,
+    POLL_TIMEOUT,
+)
+from e2e_http import URL, Headers, Success, post
+
+
+class _DdAuthHeaders(Headers):
+    api_key: str = Field(serialization_alias="DD-API-KEY")
+    app_key: str = Field(serialization_alias="DD-APPLICATION-KEY")
+
+
+class _SearchFilter(BaseModel):
+    query: str
+    #: Wide enough to cover a full suite run plus DataDog's ingestion lag;
+    #: markers are unique per test, so a wide window cannot match foreign events.
+    from_: str = Field(default="now-30m", serialization_alias="from")
+    to: str = "now"
+
+
+class _SearchPage(BaseModel):
+    limit: int = 100
+
+
+class _SearchRequest(BaseModel):
+    filter: _SearchFilter
+    page: _SearchPage = _SearchPage()
+    sort: str = "timestamp"
+
+
+class DdLogEvent(BaseModel):
+    """One ingested log event as the search API returns it: the indexed
+    envelope (service/status/tags) plus ``attributes`` - DataDog's parse of the
+    JSON message the integration shipped, i.e. the StandardLoggingPayload
+    fields."""
+
+    model_config = ConfigDict(extra="ignore")
+
+    service: str | None = None
+    status: str | None = None
+    tags: list[str] = []
+    attributes: dict[str, object] = {}
+
+
+class _SearchEvent(BaseModel):
+    model_config = ConfigDict(extra="ignore")
+
+    attributes: DdLogEvent
+
+
+class _SearchResponse(BaseModel):
+    model_config = ConfigDict(extra="ignore")
+
+    data: list[_SearchEvent] = []
+
+
+@dataclass(frozen=True, slots=True)
+class DdLogsReader:
+    site: str
+    api_key: str
+    app_key: str
+
+    def events_for_marker(self, marker: str) -> list[DdLogEvent]:
+        """Every ingested event matching the marker (full-text, exact phrase).
+        More than one hit for one call IS the duplicate-delivery bug, so this
+        never collapses to a single event."""
+        result = post(
+            URL(f"https://api.{self.site}/api/v2/logs/events/search"),
+            headers=_DdAuthHeaders(api_key=self.api_key, app_key=self.app_key),
+            json=_SearchRequest(filter=_SearchFilter(query=f'"{marker}"')),
+            response_type=_SearchResponse,
+            timeout=30.0,
+        )
+        match result:
+            case Success(data=page):
+                return [event.attributes for event in page.data]
+            case failure:
+                pytest.fail(f"DataDog Logs Search API at api.{self.site} failed: {failure}")
+
+    def poll_events_for_marker(self, marker: str) -> list[DdLogEvent]:
+        """Poll until at least one matching event is searchable (the callback
+        flushes in periodic batches and DataDog ingestion adds seconds of lag),
+        then keep re-reading for DD_SETTLE_SECONDS so a late duplicate cannot
+        hide from the exactly-one assertion - real-DataDog jitter can surface
+        one call's two events tens of seconds apart. 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:
+                return self._settled_events_for_marker(marker, events)
+            time.sleep(POLL_INTERVAL)
+        return self.events_for_marker(marker)
+
+    def _settled_events_for_marker(
+        self, marker: str, events: list[DdLogEvent]
+    ) -> list[DdLogEvent]:
+        """Re-read at every poll interval until the settle window closes; a
+        duplicate ends the watch early because more waiting cannot clear it."""
+        settle_deadline = time.monotonic() + DD_SETTLE_SECONDS
+        while time.monotonic() < settle_deadline:
+            time.sleep(POLL_INTERVAL)
+            events = self.events_for_marker(marker)
+            if len(events) > 1:
+                return events
+        return events
+
+
+def build_dd_logs_reader() -> DdLogsReader:
+    if not DD_API_KEY or not DD_APP_KEY:
+        pytest.fail(
+            "DD_API_KEY and DD_APP_KEY must be set: the DataDog tests deliver to and "
+            "read back from the real DataDog API (on the cluster the secret manager "
+            "injects them; locally set them in tests/e2e/.env)"
+        )
+    return DdLogsReader(site=DD_SITE, api_key=DD_API_KEY, app_key=DD_APP_KEY)