Merge pull request #34166 from BerriAI/litellm_lit_4562_weekly_anomaly_load_test

test(e2e): add weekly session-anomaly load test against real providers

.github/workflows/weekly_load_anomaly.yml

@@ -0,0 +1,81 @@
+name: "Weekly Load Anomaly Check"
+
+on:
+  schedule:
+    - cron: "0 12 * * 6"
+  workflow_dispatch:
+
+permissions:
+  contents: read
+
+jobs:
+  weekly-load-anomaly:
+    if: github.event_name != 'schedule' || github.repository == 'BerriAI/litellm'
+    runs-on: ubuntu-latest
+    timeout-minutes: 45
+    services:
+      postgres:
+        image: postgres:16.6
+        env:
+          POSTGRES_USER: llmproxy
+          POSTGRES_PASSWORD: dbpassword9090
+          POSTGRES_DB: litellm
+        ports:
+          - 5432:5432
+        options: >-
+          --health-cmd "pg_isready -U llmproxy"
+          --health-interval 5s
+          --health-timeout 5s
+          --health-retries 10
+    env:
+      DATABASE_URL: postgresql://llmproxy:dbpassword9090@localhost:5432/litellm
+      LITELLM_MASTER_KEY: sk-weekly-anomaly-check
+      ANTHROPIC_API_KEY: ${{ secrets.ANTHROPIC_API_KEY }}
+      AWS_BEARER_TOKEN_BEDROCK: ${{ secrets.AWS_BEARER_TOKEN_BEDROCK }}
+    steps:
+      - uses: actions/checkout@08eba0b27e820071cde6df949e0beb9ba4906955 # v4.3.0
+        with:
+          persist-credentials: false
+
+      - name: Set up Python
+        uses: actions/setup-python@a26af69be951a213d495a4c3e4e4022e16d87065 # v5.6.0
+        with:
+          python-version: "3.12"
+
+      - name: Set up uv
+        uses: ./.github/actions/setup-uv-with-retries
+        with:
+          version: "0.10.9"
+
+      - name: Install dependencies
+        run: |
+          .github/scripts/uv_sync_with_retries.sh --frozen --group ci --group proxy-dev --extra proxy
+
+      - name: Generate Prisma client
+        env:
+          PRISMA_BINARY_CACHE_DIR: ${{ runner.temp }}/prisma-cache
+        run: |
+          uv run --no-sync prisma generate --schema litellm/proxy/schema.prisma
+
+      - name: Start the proxy
+        run: |
+          nohup uv run --no-sync litellm --config tests/e2e/load/weekly_anomaly_config.yml --port 4000 > proxy.log 2>&1 &
+          for _ in $(seq 1 90); do
+            if curl -fs http://localhost:4000/health/liveliness > /dev/null; then
+              exit 0
+            fi
+            sleep 2
+          done
+          echo "proxy never became live"
+          tail -n 100 proxy.log
+          exit 1
+
+      - name: Run the weekly session anomaly test
+        env:
+          E2E_WEEKLY_ANOMALY: "1"
+        run: |
+          uv run --no-sync pytest tests/e2e/load/test_weekly_session_anomaly_e2e.py -v --tb=short -rA
+
+      - name: Show proxy log on failure
+        if: failure()
+        run: tail -n 300 proxy.log

tests/e2e/CLAUDE.md

@@ -18,7 +18,7 @@ Each subdirectory under `tests/e2e/` is one suite, scoped to an endpoint family
 - `logging/` - logging-integration delivery (datadog and friends)
 - `security/` - secret handling and log-leak protection
 - `router/` - routing and reliability behavior (fallbacks, cooldowns)
-- `load/` - throughput/performance under concurrency: drives real concurrent traffic through the whole stack with Locust and asserts a throughput SLO; marked `load` so the parent conftest collects it last and it never perturbs latency-sensitive suites
+- `load/` - throughput/performance under concurrency: drives real concurrent traffic through the whole stack with Locust and asserts a throughput SLO; marked `load` so the parent conftest collects it last and it never perturbs latency-sensitive suites. Also home of the weekly session-anomaly test (`test_weekly_session_anomaly_e2e.py`): Claude Code-shaped multi-turn sessions against real providers with ceilings on error rate, cache read/write, turn time, and spend; additionally marked `weekly` and deselected unless `E2E_WEEKLY_ANOMALY` is set, because it spends real provider money (driven by `.github/workflows/weekly_load_anomaly.yml`)
 - `other/` - the holding-pen suite for the `other.*` registry cluster with no home of its own yet: the master-key auth gate and the process-lifecycle health probes (liveness, public readiness, authenticated readiness diagnostics). Promote a cluster out once it is large/stable enough for its own suite
 - `gateway/` - proxy configuration only (`litellm-config.yml`); no tests
 - `claude_code/` - the Claude Code compatibility matrix: drives the real `claude` CLI (and HTTP probes) against a proxy for each feature x provider cell, reporting tagged-union outcomes via the `compat_result` fixture; ships its own driver/builder/publisher plus `_*_unit_tests/` trees. The HTTP probes ride the shared transport (`ProxyClient.count_tokens` / `ProxyClient.messages`); the CLI-driving path stays bespoke
@@ -132,7 +132,7 @@ reliability.<behavior>.<variant>.<assertion>
   behavior  : fallback | retry | cooldown | timeout | routing | cache | circuit_breaker | perf
   variant   : <trigger>   5xx | context_window | content_policy | 429 | timeout
               <strategy>  simple_shuffle | usage_based | latency_based | cost_based | least_busy
-              <dimension> latency | throughput   (perf only; SLO/threshold assertion, not binary)
+              <dimension> latency | throughput | session_anomaly   (perf only; SLO/threshold assertion, not binary)
   assertion : routes_to_fallback | succeeds_within_retries | picks_under_tpm | returns_cached
               | trips_then_recovers | under_slo
   e.g.  reliability.fallback.context_window.routes_to_fallback     exercised_on=[chat_completions]

tests/e2e/conftest.py

@@ -43,6 +43,10 @@ def pytest_configure(config: pytest.Config) -> None:
         "markers",
         "load: heavy throughput/load test; collected last so it never perturbs latency-sensitive suites",
     )
+    config.addinivalue_line(
+        "markers",
+        "weekly: real-provider anomaly load test that spends real money; deselected unless E2E_WEEKLY_ANOMALY is set",
+    )
 
 
 def pytest_collection_modifyitems(items: list[pytest.Item]) -> None:

tests/e2e/coverage_registry/reliability.yaml

@@ -24,3 +24,4 @@
 - {id: reliability.timeout.request_timeout.exceeds_deadline, module: reliability, tier: P1, behavior: timeout, variant: request_timeout, assertions: [exceeds_deadline], exercised_on: [chat_completions, messages], source: "litellm/router.py:545-551", rationale: "Per-request timeout raises Timeout"}
 - {id: reliability.timeout.stream_timeout.exceeds_deadline, module: reliability, tier: P1, behavior: timeout, variant: stream_timeout, assertions: [exceeds_deadline], exercised_on: [chat_completions], source: "litellm/router.py:551", rationale: "Streaming chunk-delivery timeout"}
 - {id: reliability.perf.throughput.under_slo, module: reliability, tier: P1, behavior: perf, variant: throughput, assertions: [under_slo], exercised_on: [chat_completions, messages], source: grammar, rationale: "Throughput SLO under load"}
+- {id: reliability.perf.session_anomaly.under_slo, module: reliability, tier: P1, behavior: perf, variant: session_anomaly, assertions: [under_slo], exercised_on: [messages], source: grammar, rationale: "Weekly Claude Code-shaped multi-turn session load against real providers; ceilings on error rate, warm-turn cache read/write, p95 turn time, and gateway-recorded spend (LIT-4562)"}

tests/e2e/e2e_config.py

@@ -78,6 +78,24 @@ LOAD_DURATION_SECONDS = float(os.environ.get("E2E_LOAD_DURATION_SECONDS", "60"))
 LOAD_MIN_RPS = float(os.environ.get("E2E_LOAD_MIN_RPS", "355"))
 LOAD_MAX_FAILURE_RATIO = float(os.environ.get("E2E_LOAD_MAX_FAILURE_RATIO", "0.01"))
 
+WEEKLY_ANOMALY_OPT_IN_ENV = "E2E_WEEKLY_ANOMALY"
+ANOMALY_SESSIONS = int(os.environ.get("E2E_ANOMALY_SESSIONS", "6"))
+ANOMALY_TURNS_PER_SESSION = int(os.environ.get("E2E_ANOMALY_TURNS_PER_SESSION", "6"))
+ANOMALY_TURN_ATTEMPTS = int(os.environ.get("E2E_ANOMALY_TURN_ATTEMPTS", "3"))
+ANOMALY_MAX_ERROR_RATIO = float(os.environ.get("E2E_ANOMALY_MAX_ERROR_RATIO", "0.05"))
+ANOMALY_MIN_WARM_CACHE_READ_SHARE = float(
+    os.environ.get("E2E_ANOMALY_MIN_WARM_CACHE_READ_SHARE", "0.65")
+)
+ANOMALY_MAX_P95_TURN_SECONDS = float(
+    os.environ.get("E2E_ANOMALY_MAX_P95_TURN_SECONDS", "30")
+)
+ANOMALY_MAX_KEY_SPEND_USD = float(
+    os.environ.get("E2E_ANOMALY_MAX_KEY_SPEND_USD", "0.60")
+)
+ANOMALY_SPEND_SETTLE_SECONDS = float(
+    os.environ.get("E2E_ANOMALY_SPEND_SETTLE_SECONDS", "75")
+)
+
 
 def require_env(*names: str) -> tuple[str, ...]:
     """Return the non-empty values for each env name, or hard-fail naming which are missing.

tests/e2e/load/conftest.py

@@ -1,10 +1,12 @@
 from __future__ import annotations
 
+import os
 from collections.abc import Iterator
 
 import pytest
 from requests import RequestException
 
+from e2e_config import WEEKLY_ANOMALY_OPT_IN_ENV
 from e2e_http import NoBody, Success
 from load_client import LoadClient, build_client
 from load_constants import LOAD_MODEL
@@ -18,6 +20,22 @@ LOAD_MODEL_PARAMS = LiteLLMParamsBody(
 )
 
 
+def pytest_collection_modifyitems(
+    config: pytest.Config, items: list[pytest.Item]
+) -> None:
+    if os.environ.get(WEEKLY_ANOMALY_OPT_IN_ENV):
+        return
+    deselected = [
+        item for item in items if item.get_closest_marker("weekly") is not None
+    ]
+    if not deselected:
+        return
+    config.hook.pytest_deselected(items=deselected)
+    items[:] = [
+        item for item in items if item.get_closest_marker("weekly") is None
+    ]
+
+
 @pytest.fixture(scope="session")
 def client(proxy: ProxyClient) -> LoadClient:
     return build_client(proxy)
@@ -33,10 +51,8 @@ def _model_is_servable(proxy: ProxyClient, model_name: str) -> bool:
     return isinstance(result, Success) and any(entry.id == model_name for entry in result.data.data)
 
 
-@pytest.fixture(scope="session", autouse=True)
-def _ensure_load_model(  # pyright: ignore[reportUnusedFunction]  # pytest autouse session fixture, wired by name
-    client: LoadClient,
-) -> Iterator[None]:
+@pytest.fixture(scope="session")
+def ensure_load_model(client: LoadClient) -> Iterator[None]:
     proxy = client.proxy
     if _model_is_servable(proxy, LOAD_MODEL):
         yield
@@ -60,7 +76,9 @@ def _ensure_load_model(  # pyright: ignore[reportUnusedFunction]  # pytest autou
 
 
 @pytest.fixture
-def load_key(resources: ResourceManager, client: LoadClient) -> str:
+def load_key(
+    resources: ResourceManager, client: LoadClient, ensure_load_model: None
+) -> str:
     key = client.proxy.generate_key(KeyGenerateBody(models=[LOAD_MODEL], user_id="e2e-load"))
     resources.defer(lambda: client.proxy.delete_key(key))
     return key

tests/e2e/load/session_anomaly.py

@@ -0,0 +1,299 @@
+from __future__ import annotations
+
+import time
+from collections.abc import Callable
+from concurrent.futures import ThreadPoolExecutor
+from dataclasses import dataclass
+
+from pydantic import BaseModel
+
+from e2e_config import unique_marker
+from e2e_http import Result, Success
+from models import CacheControl, RichMessage, TextBlock
+from transport import Transport
+
+
+class SessionMessagesRequest(BaseModel):
+    model: str
+    max_tokens: int = 128
+    system: list[TextBlock]
+    messages: list[RichMessage]
+
+
+class SessionUsage(BaseModel):
+    input_tokens: int = 0
+    output_tokens: int = 0
+    cache_creation_input_tokens: int = 0
+    cache_read_input_tokens: int = 0
+
+
+class SessionContentBlock(BaseModel):
+    type: str | None = None
+    text: str | None = None
+
+
+class SessionMessagesResponse(BaseModel):
+    content: list[SessionContentBlock] = []
+    usage: SessionUsage = SessionUsage()
+
+    @property
+    def text(self) -> str:
+        return "".join(block.text or "" for block in self.content)
+
+
+@dataclass(frozen=True, slots=True)
+class TurnMetric:
+    turn_index: int
+    ok: bool
+    latency_seconds: float
+    uncached_input_tokens: int
+    cache_read_tokens: int
+    cache_creation_tokens: int
+    failure: str | None
+
+
+@dataclass(frozen=True, slots=True)
+class AnomalyReport:
+    planned_turns: int
+    attempted_turns: int
+    failed_turns: int
+    warm_turns: int
+    warm_uncached_input_tokens: int
+    warm_cache_read_tokens: int
+    warm_cache_creation_tokens: int
+    p95_turn_seconds: float
+
+    @property
+    def error_ratio(self) -> float:
+        return self.failed_turns / self.planned_turns if self.planned_turns else 1.0
+
+    @property
+    def warm_cache_read_share(self) -> float:
+        billed = (
+            self.warm_uncached_input_tokens
+            + self.warm_cache_read_tokens
+            + self.warm_cache_creation_tokens
+        )
+        return self.warm_cache_read_tokens / billed if billed else 0.0
+
+
+def _system_prefix_block(marker: str) -> TextBlock:
+    text = " ".join(
+        f"Project context paragraph {index} for session {marker}." for index in range(300)
+    )
+    return TextBlock(text=text, cache_control=CacheControl())
+
+
+def _user_turn_text(marker: str, turn_index: int) -> str:
+    notes = " ".join(
+        f"Working note {index} of turn {turn_index} in session {marker}."
+        for index in range(80)
+    )
+    return f"Reply with one short sentence.\n{notes}"
+
+
+def _reminder_turn() -> RichMessage:
+    return RichMessage(
+        role="system",
+        content=[
+            TextBlock(
+                text="<system-reminder>Keep the answer to one short sentence.</system-reminder>"
+            )
+        ],
+    )
+
+
+def _without_cache_control(message: RichMessage) -> RichMessage:
+    return RichMessage(
+        role=message.role,
+        content=[TextBlock(text=block.text) for block in message.content],
+    )
+
+
+RETRY_BACKOFF_SECONDS = 2.0
+
+
+def retried(
+    call: Callable[[], Result[SessionMessagesResponse]],
+    attempts: int,
+    backoff_seconds: float = RETRY_BACKOFF_SECONDS,
+    sleep: Callable[[float], None] = time.sleep,
+) -> Result[SessionMessagesResponse]:
+    result = call()
+    if isinstance(result, Success) or attempts <= 1:
+        return result
+    sleep(backoff_seconds)
+    return retried(call, attempts - 1, backoff_seconds, sleep)
+
+
+def _metric(
+    result: Result[SessionMessagesResponse], turn_index: int, latency_seconds: float
+) -> TurnMetric:
+    if isinstance(result, Success):
+        usage = result.data.usage
+        return TurnMetric(
+            turn_index=turn_index,
+            ok=True,
+            latency_seconds=latency_seconds,
+            uncached_input_tokens=usage.input_tokens,
+            cache_read_tokens=usage.cache_read_input_tokens,
+            cache_creation_tokens=usage.cache_creation_input_tokens,
+            failure=None,
+        )
+    return TurnMetric(
+        turn_index=turn_index,
+        ok=False,
+        latency_seconds=latency_seconds,
+        uncached_input_tokens=0,
+        cache_read_tokens=0,
+        cache_creation_tokens=0,
+        failure=repr(result),
+    )
+
+
+def _drive_turns(
+    transport: Transport,
+    key: str,
+    model: str,
+    marker: str,
+    system_block: TextBlock,
+    history: tuple[RichMessage, ...],
+    turn_index: int,
+    remaining_turns: int,
+    attempts_per_turn: int,
+) -> tuple[TurnMetric, ...]:
+    if remaining_turns == 0:
+        return ()
+    user_turn = RichMessage(
+        role="user",
+        content=[
+            TextBlock(
+                text=_user_turn_text(marker, turn_index), cache_control=CacheControl()
+            )
+        ],
+    )
+    started = time.monotonic()
+    result = retried(
+        lambda: transport.post(
+            "/v1/messages",
+            headers=transport.bearer(key),
+            json=SessionMessagesRequest(
+                model=model,
+                system=[system_block],
+                messages=[*history, user_turn],
+            ),
+            response_type=SessionMessagesResponse,
+        ),
+        attempts_per_turn,
+    )
+    turn = _metric(result, turn_index, time.monotonic() - started)
+    if not isinstance(result, Success):
+        return (turn,)
+    assistant_turn = RichMessage(
+        role="assistant", content=[TextBlock(text=result.data.text or "Understood.")]
+    )
+    return (
+        turn,
+        *_drive_turns(
+            transport,
+            key,
+            model,
+            marker,
+            system_block,
+            (
+                *history,
+                _without_cache_control(user_turn),
+                _reminder_turn(),
+                assistant_turn,
+            ),
+            turn_index + 1,
+            remaining_turns - 1,
+            attempts_per_turn,
+        ),
+    )
+
+
+def run_session(
+    transport: Transport, key: str, model: str, turns: int, attempts_per_turn: int
+) -> tuple[TurnMetric, ...]:
+    marker = unique_marker()
+    return _drive_turns(
+        transport,
+        key,
+        model,
+        marker,
+        _system_prefix_block(marker),
+        (),
+        1,
+        turns,
+        attempts_per_turn,
+    )
+
+
+def run_concurrent_sessions(
+    transport: Transport,
+    key: str,
+    model: str,
+    sessions: int,
+    turns_per_session: int,
+    attempts_per_turn: int,
+) -> tuple[TurnMetric, ...]:
+    with ThreadPoolExecutor(max_workers=sessions) as pool:
+        futures = [
+            pool.submit(
+                run_session, transport, key, model, turns_per_session, attempts_per_turn
+            )
+            for _ in range(sessions)
+        ]
+        return tuple(turn for future in futures for turn in future.result())
+
+
+def settled_spend(
+    read_spend: Callable[[], float],
+    poll_interval: float,
+    settle_seconds: float,
+    timeout_seconds: float,
+    now: Callable[[], float] = time.monotonic,
+    sleep: Callable[[float], None] = time.sleep,
+) -> float:
+    deadline = now() + timeout_seconds + settle_seconds
+
+    def settle(previous: float, stable_since: float) -> float:
+        current = read_spend()
+        observed = now()
+        since = stable_since if current == previous else observed
+        if current > 0 and observed - since >= settle_seconds:
+            return current
+        if observed >= deadline:
+            raise AssertionError(
+                f"key spend never held a stable non-zero value for {settle_seconds}s "
+                f"within {timeout_seconds + settle_seconds}s (last read {current}); "
+                f"spend stopped being recorded, which is itself a spend anomaly"
+            )
+        sleep(poll_interval)
+        return settle(current, since)
+
+    return settle(-1.0, now())
+
+
+def _p95(latencies: tuple[float, ...]) -> float:
+    if not latencies:
+        return 0.0
+    ranked = sorted(latencies)
+    return ranked[max(0, -(-len(ranked) * 95 // 100) - 1)]
+
+
+def summarize(turns: tuple[TurnMetric, ...], planned_turns: int) -> AnomalyReport:
+    warm = tuple(turn for turn in turns if turn.ok and turn.turn_index >= 2)
+    return AnomalyReport(
+        planned_turns=planned_turns,
+        attempted_turns=len(turns),
+        failed_turns=planned_turns - sum(1 for turn in turns if turn.ok),
+        warm_turns=len(warm),
+        warm_uncached_input_tokens=sum(turn.uncached_input_tokens for turn in warm),
+        warm_cache_read_tokens=sum(turn.cache_read_tokens for turn in warm),
+        warm_cache_creation_tokens=sum(turn.cache_creation_tokens for turn in warm),
+        p95_turn_seconds=_p95(
+            tuple(turn.latency_seconds for turn in turns if turn.ok)
+        ),
+    )

tests/e2e/load/test_session_anomaly.py

@@ -0,0 +1,141 @@
+from __future__ import annotations
+
+from itertools import count, repeat
+
+import pytest
+
+from e2e_http import NetworkError, Success
+from session_anomaly import (
+    SessionMessagesResponse,
+    TurnMetric,
+    retried,
+    settled_spend,
+    summarize,
+)
+
+
+def _ok_turn(turn_index: int) -> TurnMetric:
+    return TurnMetric(
+        turn_index=turn_index,
+        ok=True,
+        latency_seconds=1.0,
+        uncached_input_tokens=10,
+        cache_read_tokens=100,
+        cache_creation_tokens=5,
+        failure=None,
+    )
+
+
+def _failed_turn(turn_index: int) -> TurnMetric:
+    return TurnMetric(
+        turn_index=turn_index,
+        ok=False,
+        latency_seconds=1.0,
+        uncached_input_tokens=0,
+        cache_read_tokens=0,
+        cache_creation_tokens=0,
+        failure="NetworkError()",
+    )
+
+
+class TestSummarizePlannedTurns:
+    def test_session_aborted_on_first_turn_counts_all_its_planned_turns_as_failed(
+        self,
+    ) -> None:
+        completed_session = tuple(_ok_turn(index) for index in range(1, 7))
+        aborted_session = (_failed_turn(1),)
+
+        report = summarize((*completed_session, *aborted_session), planned_turns=12)
+
+        assert report.attempted_turns == 7
+        assert report.failed_turns == 6
+        assert report.error_ratio == 0.5
+
+    def test_all_planned_turns_completing_reports_zero_failures(self) -> None:
+        report = summarize(
+            tuple(_ok_turn(index) for index in range(1, 7)), planned_turns=6
+        )
+
+        assert report.failed_turns == 0
+        assert report.error_ratio == 0.0
+
+
+class TestRetried:
+    def test_transient_failures_then_success_returns_the_success(self) -> None:
+        outcome = Success(data=SessionMessagesResponse())
+        calls = iter(
+            (NetworkError(message="overloaded"), NetworkError(message="overloaded"), outcome)
+        )
+
+        result = retried(lambda: next(calls), attempts=3, sleep=lambda _: None)
+
+        assert result is outcome
+
+    def test_exhausted_attempts_return_the_last_failure(self) -> None:
+        last_attempt = NetworkError(message="still overloaded")
+        never_reached = NetworkError(message="a fourth attempt would break the budget")
+        calls = iter(
+            (NetworkError(message="overloaded"), last_attempt, never_reached)
+        )
+
+        result = retried(lambda: next(calls), attempts=2, sleep=lambda _: None)
+
+        assert result is last_attempt
+        assert next(calls) is never_reached
+
+    def test_first_try_success_never_sleeps(self) -> None:
+        def sleep_means_retry(_: float) -> None:
+            raise AssertionError("slept after a successful attempt")
+
+        result = retried(
+            lambda: Success(data=SessionMessagesResponse()),
+            attempts=3,
+            sleep=sleep_means_retry,
+        )
+
+        assert isinstance(result, Success)
+
+
+class TestSettledSpend:
+    def test_partial_total_between_batch_flushes_is_not_accepted_as_final(self) -> None:
+        reads = iter((0.1, 0.1, 0.1, 0.35, 0.35, 0.35, 0.35, 0.35))
+        ticks = count(0.0, 2.5)
+
+        spend = settled_spend(
+            lambda: next(reads),
+            poll_interval=5.0,
+            settle_seconds=10.0,
+            timeout_seconds=100.0,
+            now=lambda: next(ticks),
+            sleep=lambda _: None,
+        )
+
+        assert spend == 0.35
+
+    def test_spend_that_never_stabilizes_raises(self) -> None:
+        reads = (0.1 * step for step in count(1))
+        ticks = count(0.0, 2.5)
+
+        with pytest.raises(AssertionError, match="spend anomaly"):
+            settled_spend(
+                lambda: next(reads),
+                poll_interval=5.0,
+                settle_seconds=5.0,
+                timeout_seconds=10.0,
+                now=lambda: next(ticks),
+                sleep=lambda _: None,
+            )
+
+    def test_spend_that_never_becomes_nonzero_raises(self) -> None:
+        reads = repeat(0.0)
+        ticks = count(0.0, 2.5)
+
+        with pytest.raises(AssertionError, match="spend anomaly"):
+            settled_spend(
+                lambda: next(reads),
+                poll_interval=5.0,
+                settle_seconds=5.0,
+                timeout_seconds=10.0,
+                now=lambda: next(ticks),
+                sleep=lambda _: None,
+            )

tests/e2e/load/test_weekly_session_anomaly_e2e.py

@@ -0,0 +1,124 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+import pytest
+
+from e2e_config import (
+    ANOMALY_MAX_ERROR_RATIO,
+    ANOMALY_MAX_KEY_SPEND_USD,
+    ANOMALY_MAX_P95_TURN_SECONDS,
+    ANOMALY_MIN_WARM_CACHE_READ_SHARE,
+    ANOMALY_SESSIONS,
+    ANOMALY_SPEND_SETTLE_SECONDS,
+    ANOMALY_TURN_ATTEMPTS,
+    ANOMALY_TURNS_PER_SESSION,
+    unique_marker,
+)
+from lifecycle import ResourceManager
+from load_client import LoadClient
+from models import KeyGenerateBody, LiteLLMParamsBody
+from proxy_client import ProxyClient
+from session_anomaly import run_concurrent_sessions, settled_spend, summarize
+
+pytestmark = [pytest.mark.e2e, pytest.mark.load, pytest.mark.weekly]
+
+
+@dataclass(frozen=True, slots=True)
+class AnomalyRoute:
+    route_id: str
+    params: LiteLLMParamsBody
+
+
+ANOMALY_ROUTES = (
+    AnomalyRoute(
+        route_id="anthropic",
+        params=LiteLLMParamsBody(model="anthropic/claude-sonnet-5"),
+    ),
+    AnomalyRoute(
+        route_id="bedrock_invoke",
+        params=LiteLLMParamsBody(
+            model="bedrock/invoke/us.anthropic.claude-sonnet-5",
+            aws_region_name="us-east-1",
+        ),
+    ),
+)
+
+
+def _route_id(route: AnomalyRoute) -> str:
+    return route.route_id
+
+
+def _settled_key_spend(proxy: ProxyClient, key: str) -> float:
+    return settled_spend(
+        lambda: proxy.key_info(key).spend or 0.0,
+        proxy.poll_interval,
+        ANOMALY_SPEND_SETTLE_SECONDS,
+        proxy.poll_timeout,
+    )
+
+
+class TestWeeklySessionAnomaly:
+    @pytest.mark.covers("reliability.perf.session_anomaly.under_slo")
+    @pytest.mark.parametrize("route", ANOMALY_ROUTES, ids=_route_id)
+    def test_session_load_stays_within_baselines(
+        self, client: LoadClient, resources: ResourceManager, route: AnomalyRoute
+    ) -> None:
+        model_name = f"weekly-anomaly-{route.route_id}-{unique_marker()}"
+        model_id = client.proxy.create_model(model_name, route.params)
+        resources.defer(lambda: client.proxy.delete_model(model_id))
+        key = client.proxy.generate_key(
+            KeyGenerateBody(models=[model_name], key_alias=model_name)
+        )
+        resources.defer(lambda: client.proxy.delete_key(key))
+
+        turns = run_concurrent_sessions(
+            client.proxy.transport,
+            key,
+            model_name,
+            ANOMALY_SESSIONS,
+            ANOMALY_TURNS_PER_SESSION,
+            ANOMALY_TURN_ATTEMPTS,
+        )
+        report = summarize(turns, ANOMALY_SESSIONS * ANOMALY_TURNS_PER_SESSION)
+        failures = tuple(turn.failure for turn in turns if turn.failure)
+        print(f"{route.route_id} anomaly report: {report}")
+
+        assert report.error_ratio <= ANOMALY_MAX_ERROR_RATIO, (
+            f"{route.route_id}: {report.failed_turns}/{report.planned_turns} planned "
+            f"turns failed or never ran because their session aborted "
+            f"({report.error_ratio:.1%} > {ANOMALY_MAX_ERROR_RATIO:.1%} allowed); "
+            f"error rate is anomalously high. Failures: {failures}"
+        )
+        assert report.warm_turns > 0, (
+            f"{route.route_id}: no session got past its first turn, so cache and "
+            f"latency baselines have nothing to read. Failures: {failures}"
+        )
+        assert report.warm_cache_read_share >= ANOMALY_MIN_WARM_CACHE_READ_SHARE, (
+            f"{route.route_id}: warm turns read only {report.warm_cache_read_share:.1%} "
+            f"of billed input tokens from the prompt cache "
+            f"(read={report.warm_cache_read_tokens}, "
+            f"creation={report.warm_cache_creation_tokens}, "
+            f"uncached={report.warm_uncached_input_tokens}), below the "
+            f"{ANOMALY_MIN_WARM_CACHE_READ_SHARE:.0%} floor; the cached prefix is "
+            f"being invalidated between turns (the mid-conversation-system cache "
+            f"collapse signature) or caching stopped working"
+        )
+        assert report.warm_cache_creation_tokens > 0, (
+            f"{route.route_id}: warm turns wrote 0 cache-creation tokens across "
+            f"{report.warm_turns} turns; the moving cache breakpoint stopped writing "
+            f"new prefix increments"
+        )
+        assert report.p95_turn_seconds <= ANOMALY_MAX_P95_TURN_SECONDS, (
+            f"{route.route_id}: p95 turn time {report.p95_turn_seconds:.1f}s exceeds "
+            f"the {ANOMALY_MAX_P95_TURN_SECONDS:.0f}s ceiling under "
+            f"{ANOMALY_SESSIONS} concurrent sessions; turn times are anomalously slow"
+        )
+
+        spend = _settled_key_spend(client.proxy, key)
+        assert spend <= ANOMALY_MAX_KEY_SPEND_USD, (
+            f"{route.route_id}: gateway recorded ${spend:.4f} for "
+            f"{report.attempted_turns} turns, above the "
+            f"${ANOMALY_MAX_KEY_SPEND_USD} ceiling; spend per session is "
+            f"anomalously high (cache regressions surface here as 2-3x spend)"
+        )

tests/e2e/load/weekly_anomaly_config.yml

@@ -0,0 +1,3 @@
+general_settings:
+  master_key: os.environ/LITELLM_MASTER_KEY
+  store_model_in_db: true

tests/e2e/pytest.ini

@@ -6,3 +6,4 @@ addopts = --strict-markers --strict-config
 markers =
     e2e: live test that requires a running proxy and real provider keys
     load: heavy throughput/load test; collected last so it never perturbs latency-sensitive suites
+    weekly: real-provider anomaly load test that spends real money; deselected unless E2E_WEEKLY_ANOMALY is set