test(e2e): extend the negative-spend guard to bedrock

Add bedrock (bedrock/us.anthropic.claude-haiku-4-5) to the negative-spend
sweep, the surface BerriAI/litellm#25846's negative streaming-cost bug
actually lived on, and register it in the suite's driver_models with AWS
creds rather than a single api_key. Rework the sweep constants into a small
frozen _SweepModel(call, row_marker) so a friendly bedrock alias whose spend
row resolves to bedrock/us.anthropic... still matches; the two claude markers
are prefix-qualified so a bedrock-anthropic row can't stand in for
anthropic-direct or vice versa.

tests/e2e/coverage_registry/quota_management.yaml

@@ -33,4 +33,4 @@
 - {id: quota_management.spend_tracking.failure.writes_failure_row, module: quota_management, tier: P1, behavior: spend_tracking, variant: failure, assertions: [writes_failure_row], exercised_on: [chat_completions], source: "proxy/spend_tracking/spend_log_error_logger.py", rationale: "A failed call writes a failure-status spend row"}
 - {id: quota_management.spend_tracking.spend_calculate.returns_cost, module: quota_management, tier: P2, behavior: spend_tracking, variant: spend_calculate, assertions: [returns_cost], exercised_on: [spend_calculate], source: "proxy/spend_tracking/spend_management_endpoints.py", rationale: "/spend/calculate prices a hypothetical request at nonzero cost"}
 - {id: quota_management.spend_tracking.pagination.keeps_total, module: quota_management, tier: P2, behavior: spend_tracking, variant: pagination, assertions: [keeps_total], exercised_on: [chat_completions], source: "proxy/spend_tracking/spend_management_endpoints.py", rationale: "Spend-logs v2 pagination caps page size without losing the total"}
-- {id: quota_management.spend_tracking.non_negative.never_negative, module: quota_management, tier: P1, behavior: spend_tracking, variant: non_negative, assertions: [never_negative], exercised_on: [chat_completions, embeddings], source: "litellm_core_utils/llm_cost_calc/utils.py", rationale: "No provider writes a negative spend row across streaming/non-streaming chat and embeddings; derived cache-token counts must clamp at zero (BerriAI/litellm#25846)"}
+- {id: quota_management.spend_tracking.non_negative.never_negative, module: quota_management, tier: P1, behavior: spend_tracking, variant: non_negative, assertions: [never_negative], exercised_on: [chat_completions, embeddings], source: "litellm_core_utils/llm_cost_calc/utils.py", rationale: "No provider (gemini/anthropic/openai/bedrock) writes a negative spend row across streaming/non-streaming chat and embeddings; derived cache-token counts must clamp at zero (BerriAI/litellm#25846)"}

tests/e2e/quota_management/spend_tracking/SPEND_TRACKING_COVERAGE_MATRIX.md

@@ -69,7 +69,7 @@ proxy + SpendLogs rows. Status: `covered` / `partial` / `gap`.
 | `test_tag_spend_matches_sum_of_tagged_logs` | `/spend/tags` SUM/COUNT == tagged rows |
 | `test_end_user_spend_attributed_on_row` | `end_user` attributed + costed |
 | `test_each_model_on_a_shared_key_gets_its_own_row` | per-model/provider rows, correct model + cost, distinct request_ids matching response id |
-| `test_no_provider_logs_negative_spend` | no provider writes spend < 0 across streaming/non-streaming chat + embeddings; each provider still logs a positive row (non-vacuous) |
+| `test_no_provider_logs_negative_spend` | no provider (gemini/anthropic/openai/bedrock) writes spend < 0 across streaming/non-streaming chat + embeddings; each provider still logs a positive row (non-vacuous) |
 | `test_failure_call_writes_failure_status_row` | failed call -> `status=failure`, `spend=0` |
 | `test_spend_calculate_returns_nonzero_cost` | cost-map smoke (no batch wait) |
 | `test_spend_logs_endpoint_returns_spend` | `/spend/logs` returns 200 + the key's spend, never a 5xx (intermittent-500 regression) |

tests/e2e/quota_management/spend_tracking/conftest.py

@@ -5,13 +5,14 @@ live in the parent tests/e2e/conftest.py. SpendClient exposes the shared Gateway
 (GatewayProvider), so the `resources` fixture cleans up keys and customers this
 suite creates.
 
-The suite drives real calls through three deployments. On the stage gateway they
-are baked into the proxy config; on a local dev proxy they usually are not, so
-`driver_models` registers whichever are missing via /model/new and deletes only
-the ones it created, never a config-baked deployment. Each registration carries
-the provider key from the test runner's env when set (so a local proxy whose
-container env lacks the key still works); otherwise it falls back to an
-os.environ reference resolved from the proxy's own env, the stage convention.
+The suite drives real calls through four deployments (gemini, anthropic, bedrock,
+openai embeddings). On the stage gateway they are baked into the proxy config; on a
+local dev proxy they usually are not, so `driver_models` registers whichever are
+missing via /model/new and deletes only the ones it created, never a config-baked
+deployment. Each registration carries the provider credential from the test runner's
+env when set (so a local proxy whose container env lacks it still works); otherwise
+it falls back to an os.environ reference resolved from the proxy's own env, the stage
+convention. Bedrock is credentialed with AWS keys rather than a single api_key.
 """
 
 import os
@@ -23,17 +24,37 @@ from models import LiteLLMParamsBody
 from spend_e2e_client import SpendClient, build_client
 
 
-def _driver_params(provider_model: str, env_var: str) -> LiteLLMParamsBody:
+def _env_ref(name: str) -> str:
+    """The runner's value when set (so a local proxy whose container env lacks it
+    still works), else an os.environ reference the proxy resolves from its own env."""
+    return os.environ.get(name) or f"os.environ/{name}"
+
+
+def _api_key_params(provider_model: str, env_var: str) -> LiteLLMParamsBody:
+    return LiteLLMParamsBody(model=provider_model, api_key=_env_ref(env_var))
+
+
+def _bedrock_params(provider_model: str) -> LiteLLMParamsBody:
+    """Bedrock authenticates with AWS creds, not a single api_key."""
     return LiteLLMParamsBody(
         model=provider_model,
-        api_key=os.environ.get(env_var) or f"os.environ/{env_var}",
+        aws_access_key_id=_env_ref("AWS_ACCESS_KEY_ID"),
+        aws_secret_access_key=_env_ref("AWS_SECRET_ACCESS_KEY"),
+        aws_region_name=os.environ.get("AWS_REGION_NAME") or _env_ref("AWS_REGION"),
     )
 
 
-DRIVER_MODELS: tuple[tuple[str, str, str], ...] = (
-    ("gemini-2.5-flash", "gemini/gemini-2.5-flash", "GEMINI_API_KEY"),
-    ("claude-haiku-4-5", "anthropic/claude-haiku-4-5", "ANTHROPIC_API_KEY"),
-    ("openai-text-embedding-3-small", "openai/text-embedding-3-small", "OPENAI_API_KEY"),
+DRIVER_MODELS: tuple[tuple[str, LiteLLMParamsBody], ...] = (
+    ("gemini-2.5-flash", _api_key_params("gemini/gemini-2.5-flash", "GEMINI_API_KEY")),
+    ("claude-haiku-4-5", _api_key_params("anthropic/claude-haiku-4-5", "ANTHROPIC_API_KEY")),
+    (
+        "bedrock-claude-haiku-4-5",
+        _bedrock_params("bedrock/us.anthropic.claude-haiku-4-5-20251001-v1:0"),
+    ),
+    (
+        "openai-text-embedding-3-small",
+        _api_key_params("openai/text-embedding-3-small", "OPENAI_API_KEY"),
+    ),
 )
 
 
@@ -46,8 +67,8 @@ def client() -> SpendClient:
 def driver_models(client: SpendClient) -> Iterator[None]:
     existing = frozenset(entry.model_name for entry in client.gateway.model_info())
     created = tuple(
-        client.gateway.create_model(name, _driver_params(provider_model, env_var))
-        for name, provider_model, env_var in DRIVER_MODELS
+        client.gateway.create_model(name, params)
+        for name, params in DRIVER_MODELS
         if name not in existing
     )
     yield

tests/e2e/quota_management/spend_tracking/test_spend_tracking_e2e.py

@@ -18,6 +18,7 @@ fails the test; a pricing or token-count drift does not.
 import time
 from collections.abc import Callable
 from concurrent.futures import ThreadPoolExecutor
+from dataclasses import dataclass
 
 import pytest
 
@@ -430,13 +431,30 @@ def test_each_model_on_a_shared_key_gets_its_own_row(
         ), f"claude row request_id {claude_row.request_id} != response id {claude.id}"
 
 
-# Chat providers this negative-spend guard sweeps when the proxy exposes them. Two
-# are guaranteed by the suite's driver_models (gemini, anthropic); gpt-5.5 (openai
-# chat) is only wired on the stage/docker gateways, so it's swept when present.
-# Adding a provider here widens the guard by one line.
-_NEG_GUARD_CHAT_MODELS: tuple[str, ...] = ("gemini-2.5-flash", "claude-haiku-4-5", "gpt-5.5")
-_NEG_GUARD_EMBEDDING_MODEL = "openai-text-embedding-3-small"
-_NEG_GUARD_EMBEDDING_ROW_MARKER = "text-embedding-3-small"
+@dataclass(frozen=True, slots=True)
+class _SweepModel:
+    """A deployment the negative-spend guard drives. `call` is the model_name sent to
+    the proxy; `row_marker` is the substring the resulting spend row's `model` must
+    contain (rows carry the resolved provider/model, so bedrock's alias shows up as
+    `bedrock/us.anthropic...`). The two claude markers are prefix-qualified so a
+    bedrock-anthropic row can't stand in for anthropic-direct, or vice versa."""
+
+    call: str
+    row_marker: str
+
+
+# Chat providers this guard sweeps when the proxy exposes them, each non-streaming and
+# streaming: gemini, anthropic, openai, and bedrock - the surface #25846's negative
+# streaming-cost bug lived on. gemini/anthropic/bedrock are registered by the suite's
+# driver_models; gpt-5.5 is only wired on the stage/docker gateways. Adding a provider
+# is one line.
+_CHAT_SWEEP: tuple[_SweepModel, ...] = (
+    _SweepModel("gemini-2.5-flash", "gemini-2.5-flash"),
+    _SweepModel("claude-haiku-4-5", "anthropic/claude-haiku-4-5"),
+    _SweepModel("gpt-5.5", "gpt-5.5"),
+    _SweepModel("bedrock-claude-haiku-4-5", "us.anthropic.claude-haiku-4-5"),
+)
+_EMBEDDING_SWEEP = _SweepModel("openai-text-embedding-3-small", "text-embedding-3-small")
 
 
 @pytest.mark.covers("quota_management.spend_tracking.non_negative.never_negative")
@@ -447,50 +465,53 @@ def test_no_provider_logs_negative_spend(
     provider the proxy exposes, both non-streaming and streaming, plus an embedding;
     then every logged row is asserted non-negative. A negative cost is a real billing
     bug: cache-token accounting that lets a derived token count fall below zero
-    (BerriAI/litellm#25846) surfaces here as spend < 0, and the streaming leg is
-    included on purpose because that is the path that regression rode in on. The
-    per-provider positive-row check keeps the guard non-vacuous - a pipeline that
-    silently logs 0 (or drops the row) fails instead of sliding past a bare `>= 0`."""
+    (BerriAI/litellm#25846, a bedrock-anthropic streaming regression) surfaces here as
+    spend < 0, and both bedrock and the streaming leg are exercised on purpose because
+    that is the path that regression rode in on. The per-provider positive-row check
+    keeps the guard non-vacuous - a pipeline that silently logs 0 (or drops the row)
+    fails instead of sliding past a bare `>= 0`."""
     present = frozenset(entry.model_name for entry in client.gateway.model_info())
-    chat_models = tuple(m for m in _NEG_GUARD_CHAT_MODELS if m in present)
-    assert len(chat_models) >= 2, (
+    chat = tuple(m for m in _CHAT_SWEEP if m.call in present)
+    assert len(chat) >= 2, (
         f"need >=2 chat providers for a cross-provider guard; "
         f"proxy exposes {sorted(present)}"
     )
-    assert _NEG_GUARD_EMBEDDING_MODEL in present, (
-        f"embedding deployment {_NEG_GUARD_EMBEDDING_MODEL!r} not registered; "
+    assert _EMBEDDING_SWEEP.call in present, (
+        f"embedding deployment {_EMBEDDING_SWEEP.call!r} not registered; "
         f"proxy exposes {sorted(present)}"
     )
 
-    for model in chat_models:
+    for model in chat:
         _ = unwrap(
-            client.chat(scoped_key, model, f"one word {unique_marker()}", max_tokens=16)
+            client.chat(
+                scoped_key, model.call, f"one word {unique_marker()}", max_tokens=16
+            )
         )
         stream = client.chat_stream(
-            scoped_key, model, f"count to three {unique_marker()}", max_tokens=32
+            scoped_key, model.call, f"count to three {unique_marker()}", max_tokens=32
         )
         assert stream.ok, (
-            f"{model} stream failed (status {stream.status_code}): {stream.body[:300]}"
+            f"{model.call} stream failed (status {stream.status_code}): {stream.body[:300]}"
         )
     _ = unwrap(
         client.embed(
-            scoped_key, _NEG_GUARD_EMBEDDING_MODEL, f"vectorize {unique_marker()}"
+            scoped_key, _EMBEDDING_SWEEP.call, f"vectorize {unique_marker()}"
         )
     )
 
-    expected_markers = (*chat_models, _NEG_GUARD_EMBEDDING_ROW_MARKER)
+    expected = (*chat, _EMBEDDING_SWEEP)
 
     def every_provider_costed(rows: list[SpendLogRow]) -> bool:
         return all(
-            any(marker in (r.model or "") and (r.spend or 0) > 0 for r in rows)
-            for marker in expected_markers
+            any(m.row_marker in (r.model or "") and (r.spend or 0) > 0 for r in rows)
+            for m in expected
         )
 
     # min_rows waits for the streaming rows too (non-stream + stream per chat model,
     # plus the embedding), so the streaming spend is actually observed and checked.
     rows = client.poll_logs_for_key(
         scoped_key,
-        min_rows=2 * len(chat_models) + 1,
+        min_rows=2 * len(chat) + 1,
         predicate=every_provider_costed,
     )
 
@@ -499,10 +520,10 @@ def test_no_provider_logs_negative_spend(
         f"provider logged negative spend (billing regression): {_summarize(negative)}"
     )
 
-    for marker in expected_markers:
+    for m in expected:
         assert any(
-            marker in (r.model or "") and (r.spend or 0) > 0 for r in rows
-        ), f"no positive spend row for {marker!r}; guard would be vacuous: {_summarize(rows)}"
+            m.row_marker in (r.model or "") and (r.spend or 0) > 0 for r in rows
+        ), f"no positive spend row for {m.row_marker!r}; guard would be vacuous: {_summarize(rows)}"
 
 
 @pytest.mark.covers("quota_management.spend_tracking.failure.writes_failure_row")