Merge remote-tracking branch 'origin/main' into litellm_project_all_team_models_403

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

# Conflicts:
#	tests/test_litellm/proxy/auth/test_auth_checks.py

.circleci/scripts/unit_selection.sh

@@ -89,6 +89,7 @@ legacy_paths() {
     proxy-db-auth-checks)
       echo tests/unit/proxy/auth/test_auth_checks.py
       echo tests/unit/proxy/auth/test_user_api_key_auth.py
+      echo tests/unit/proxy/test_credential_slot_registry.py
       echo tests/unit/proxy/test_deprecated_key_grace_period.py ;;
     proxy-db-budgets)
       echo tests/unit/proxy/auth/test_default_end_user_budget_simple.py
@@ -106,6 +107,7 @@ legacy_paths() {
       echo tests/unit/proxy/test_update_spend.py
       echo tests/unit/skills/test_skills_db.py ;;
     proxy-db-endpoints-and-responses)
+      echo tests/unit/proxy/engine
       echo tests/unit/proxy/auth/test_models_fallback_endpoint.py
       echo tests/unit/proxy/common_utils/test_check_batch_cost.py
       echo tests/unit/proxy/common_utils/test_check_responses_cost.py
@@ -147,7 +149,10 @@ legacy_paths() {
       echo tests/unit/proxy/test_proxy_server.py ;;
     proxy-db-proxy-utils) echo tests/unit/proxy/test_proxy_utils.py ;;
     proxy-extras) echo tests/unit/litellm_proxy_extras ;;
-    proxy-infra) echo tests/unit/gateway ;;
+    proxy-infra)
+      echo tests/unit/gateway
+      echo tests/unit/proxy/management_endpoints/test_roi_calculator_endpoints.py
+      echo tests/unit/proxy/roi_calculator ;;
     responses-caching-types)
       find tests/unit/responses -name 'test_*.py' -not -path 'tests/unit/responses/mcp/*'
       echo tests/unit/types ;;

.github/workflows/lens-worker.yml

@@ -0,0 +1,54 @@
+name: Lens Worker Image
+
+on:
+  pull_request:
+    branches: [main, litellm_oss_branch, "litellm_**"]
+    paths:
+      - deploy/lens/**
+      - litellm/proxy/engine/**
+      - .github/workflows/lens-worker.yml
+  push:
+    branches: [main, litellm_agent_engine]
+    paths:
+      - deploy/lens/**
+      - litellm/proxy/engine/**
+      - .github/workflows/lens-worker.yml
+  workflow_dispatch:
+
+permissions:
+  contents: read
+
+concurrency:
+  group: ${{ github.workflow }}-${{ github.event.pull_request.number || github.ref }}
+  cancel-in-progress: true
+
+jobs:
+  lens-worker-image:
+    permissions:
+      contents: read
+      packages: write
+    runs-on: ubuntu-latest
+    timeout-minutes: 10
+    steps:
+      - uses: actions/checkout@08eba0b27e820071cde6df949e0beb9ba4906955 # v4.3.0
+        with:
+          persist-credentials: false
+      - name: Build Lens worker
+        run: docker build -f deploy/lens/Dockerfile -t lens-worker:${{ github.sha }} .
+      - name: Verify standalone imports with a read-only filesystem
+        run: >-
+          docker run --rm --network none --read-only --cap-drop ALL
+          --security-opt no-new-privileges --entrypoint python
+          lens-worker:${{ github.sha }}
+          -c 'import os; import engine.worker; assert os.getuid() == 65532'
+      - name: Publish versioned Lens worker
+        if: github.event_name != 'pull_request' && github.repository == 'BerriAI/litellm'
+        env:
+          REGISTRY_TOKEN: ${{ secrets.GITHUB_TOKEN }}
+          REGISTRY_USER: ${{ github.actor }}
+          IMAGE: ghcr.io/berriai/litellm-lens-worker:sha-${{ github.sha }}
+        run: |
+          printf '%s' "$REGISTRY_TOKEN" | docker login ghcr.io -u "$REGISTRY_USER" --password-stdin
+          docker tag lens-worker:${{ github.sha }} "$IMAGE"
+          docker push "$IMAGE"
+          printf 'Lens worker image: `%s`\n' "$IMAGE" >> "$GITHUB_STEP_SUMMARY"

.github/workflows/test-postgres.yml

@@ -24,6 +24,7 @@ jobs:
     timeout-minutes: ${{ matrix.job-timeout-minutes }}
     permissions:
       contents: read
+      id-token: write
 
     services:
       postgres:
@@ -134,9 +135,19 @@ jobs:
         env:
           TEST_PATH: ${{ matrix.test-path }}
           WORKERS: ${{ matrix.workers }}
+          PYTEST_ADDOPTS: ${{ matrix.shard == 'proxy-behavior' && '--cov=litellm/proxy/engine --cov-report=xml:coverage-lens-postgres.xml' || '' }}
         run: |
           if [ "${WORKERS}" = "0" ]; then
             uv run --no-sync pytest ${TEST_PATH:?} -vv --tb=short --durations=10
           else
             uv run --no-sync pytest ${TEST_PATH:?} -vv --tb=short --durations=10 -n "${WORKERS}"
           fi
+
+      - name: Upload Lens database coverage
+        if: steps.changes.outputs.decision != 'skip' && matrix.shard == 'proxy-behavior' && !cancelled()
+        uses: codecov/codecov-action@75cd11691c0faa626561e295848008c8a7dddffe # v5.5.4
+        with:
+          use_oidc: true
+          files: coverage-lens-postgres.xml
+          flags: lens-postgres
+          fail_ci_if_error: true

.github/workflows/test-unit.yml

@@ -79,7 +79,9 @@ jobs:
 
           - shard: integrations
             artifact-name: integrations
-            test-path: ""
+            test-path: >-
+              tests/test_litellm/integrations
+              tests/test_litellm/tracing
             unit-flag: integrations
             workers: 2
             reruns: 3

backend/routes/allowlist.py

@@ -81,6 +81,8 @@ BACKEND_PATH_PREFIXES: tuple[str, ...] = (
     # Spend / analytics
     "/spend/",
     "/analytics/",
+    "/engine/",
+    "/v1/traces",
     "/global/",
     "/user_agent",
     "/usage/",
@@ -144,6 +146,7 @@ BACKEND_EXACT_PATHS: frozenset[str] = frozenset(
     {
         "/",
         "/routes",
+        "/engine",
         "/openapi.json",
         "/docs",
         "/docs/oauth2-redirect",

basedpyright-code-budget.json

@@ -99,7 +99,7 @@
     "limit": 0
   },
   "reportUnknownArgumentType": {
-    "limit": 44358
+    "limit": 44802
   },
   "reportUnknownLambdaType": {
     "limit": 109

cookbook/misc/config.yaml

@@ -24,7 +24,7 @@ model_list:
   - model_name: sagemaker-completion-model
     litellm_params:
       model: sagemaker/berri-benchmarking-Llama-2-70b-chat-hf-4
-      input_cost_per_second: 0.000420  
+      cost_per_second: 0.000420
   - model_name: text-embedding-ada-002
     litellm_params: 
       model: azure/azure-embedding-model

cookbook/misc/test_responses_api.py

@@ -12,7 +12,7 @@ def encode_image(image_path):
 
 
 # Path to your image
-image_path = "litellm/proxy/logo.jpg"
+image_path = "litellm/proxy/logo.png"
 
 # Getting the Base64 string
 base64_image = encode_image(image_path)
@@ -27,7 +27,7 @@ response = client.responses.create(
                 {"type": "input_text", "text": "what color is the image"},
                 {
                     "type": "input_image",
-                    "image_url": f"data:image/jpeg;base64,{base64_image}",
+                    "image_url": f"data:image/png;base64,{base64_image}",
                 },
             ],
         }

deploy/lens/Dockerfile

@@ -0,0 +1,6 @@
+FROM python:3.12-slim
+WORKDIR /app
+RUN pip install --no-cache-dir httpx==0.28.1 pydantic==2.11.7
+COPY litellm/proxy/engine/__init__.py litellm/proxy/engine/models.py litellm/proxy/engine/analysis.py litellm/proxy/engine/worker.py /app/engine/
+USER 65532:65532
+CMD ["python", "-m", "engine.worker"]

deploy/lens/Dockerfile.dockerignore

@@ -0,0 +1,8 @@
+**
+!litellm/
+!litellm/proxy/
+!litellm/proxy/engine/
+!litellm/proxy/engine/__init__.py
+!litellm/proxy/engine/models.py
+!litellm/proxy/engine/analysis.py
+!litellm/proxy/engine/worker.py

deploy/lens/README.md

@@ -0,0 +1,59 @@
+# Lens worker
+
+Lens reviews recorded activity and saves evidence-linked findings in the LiteLLM dashboard under Observability, Lens (`/ui/lens/`)
+
+## Start a worker
+
+Upgrade your existing LiteLLM proxy to a release that includes Lens with PostgreSQL, agent tracing (`general_settings.tracing: {store: clickhouse}`), and ClickHouse configured through `CLICKHOUSE_URL` and a separate SELECT-only `CLICKHOUSE_READER_URL`. Enable the ClickHouse callback and request/response logging to analyze LLM requests. Lens can only inspect content you actually retain
+
+In Lens, click **Connect worker**, then **Generate setup command**. The LiteLLM address is filled in for you; change it only if the server running Docker needs a different network address. Copy the command and run it on your server. The dialog changes to **Worker connected** when the container checks in
+
+The command already contains the compatible worker image and one worker token. No separate API key, source checkout, environment file, or second LiteLLM deployment is needed. Keep the command private because it includes the token. The LiteLLM release provides the dashboard and APIs; the container only runs background analysis
+
+The dashboard and Compose file pin a verified worker image by digest. The image uses Linux amd64, and the generated command selects that platform. Worker image releases are independent of proxy releases: update the pinned image when changing their API contract. CI also publishes immutable commit tags for reproducible builds
+
+For deployments managed with Compose, download `compose.yaml` and provide `LITELLM_URL` and `LENS_WORKER_TOKEN` in an environment file. Its default image is already selected:
+
+```bash
+docker compose --env-file /path/to/lens.env -f compose.yaml up -d
+```
+
+Developers can build locally with `LENS_WORKER_IMAGE=litellm-lens-worker:local docker compose -f deploy/lens/compose.yaml -f deploy/lens/compose.build.yaml up -d --build`
+
+The worker needs outbound HTTPS access to LiteLLM. It needs no inbound ports, provider keys, direct database access, or GPU. The proxy calls your selected model through its configured router; trace content reaches that model provider. Use a model with JSON output support and known token prices. One worker handles one scan at a time and can serve multiple lenses. For more throughput, start another worker with a separate credential
+
+V1 setup, manual runs, feedback, and worker credentials are restricted to proxy administrators. Admin viewers can inspect results. Worker credentials can serve the administrator’s lenses. Revoke it in the connection dialog when retiring a worker. Redeploy the worker alongside proxy upgrades so their API versions match
+
+## Configure a lens
+
+Choose agent runs, individual LLM requests, or both. The matching-activity preview updates as you choose an application (the recorded OpenTelemetry service.name) or, for request activity, a LiteLLM model group and add metadata conditions. It shows run names, timestamps, and trace IDs; open a run to inspect its original steps before starting analysis. Suggestions come from up to 100 recent executions and may not include every recorded attribute. You can enter other exact keys and values. Leave service and filters blank for all activity your account can access. Filters are exact key/value matches, combined with AND. Trace filters match span or resource attributes on the same span. Request filters match logged metadata, including caller metadata stored under `requester_metadata`; `tag=value` matches request tags. `swarm=research` works only if your instrumentation records that attribute
+
+Write a few questions, give context about a successful run, choose a model, and set the monthly limit and sample size. Choose an initial history window from 1 hour to 30 days, in hours or days. Creation queues the first scan over that window. New lenses run once by default; opt into background monitoring for a custom interval from 1 minute to 7 days, entered in minutes, hours, or days. **Analyze now** checks activity since the last successful scan; **Recheck the last 24 hours** revisits recent history. The runs API accepts `lookback_hours` from 1 to 720 for other historical windows
+
+Pausing stops future scheduled scans; cancel the active scan separately if needed. The worker polls every 10 seconds; creating a lens or clicking Analyze now queues a scan, and due schedules are queued when the worker polls. Scans for the same lens never overlap, and its next interval starts after completion. Closing the browser does not stop the worker. Configuration edits apply to the next scan. A running scan retains its settings and selected execution IDs across retries
+
+## Read the results
+
+Needs attention shows issues, highest priority first. Patterns contains useful trends and successful behavior that may not need a fix. Each finding starts with a short explanation and a next step when useful. Expand the limitations for uncertainty and counterexamples. Evidence is grouped by run and collapsed until you need it; each quote opens the original step
+
+The Runs tab lists the actual sample frozen for the latest scan. Linked-run counts on findings include cited counterexamples, so they are not failure counts. The Scans tab shows history and coverage. Existing findings retain their original wording; the shorter summaries apply to new analysis
+
+## What a scan does
+
+The proxy selects newly received or updated executions with a two-minute settling period and a five-minute overlap. Older rows without receipt timestamps use execution end time. Overlapping scans do not increment a finding's occurrence count for the same execution ID
+
+A trace is spans sharing a trace ID within one team, not an automatically reconstructed conversation session. Requests are individual LLM calls. When both sources are enabled, requests correlated to a recorded span by response ID are excluded to reduce double counting
+
+The worker screens a deterministic sample, at most the configured 1–500 executions. For each execution it reads up to 160 spans, with 8,000 characters per span section, and splits these into model calls. It consolidates observations across batches, then investigates at most 10 candidate patterns using up to five model turns each. The dashboard shows these three stages, completed work counts, and elapsed time; progress is based on the selected sample, not every eligible execution. The investigator can read more original content from the selected executions. It has no shell, browsing, code-editing, or production-action tools
+
+Each model response must match a bounded JSON schema. A malformed response gets one repair attempt through the same budget controls; repeated invalid output fails the scan. Both the worker and proxy validate quoted evidence. Findings retain exact quotes and open the source trace or request. Resolve a finding after a fix, or dismiss it with a reason. A resolved finding reopens when new execution IDs support the same pattern; dismissed findings remain dismissed
+
+Coverage distinguishes eligible, sampled, reviewed, partial, and unassessable executions. Findings describe observations in the sample, not population-wide success rates or proven causes. A root span does not prove that a trace contains every expected span. Long, missing, redacted, or expired content limits the conclusions
+
+## Operations and limits
+
+PostgreSQL stores configurations, findings and the latest 50 jobs. Workers claim jobs with optimistic concurrency and a five-minute lease, renewed every 30 seconds. A disconnected job can be reclaimed up to three times. Cancellation stops subsequent work; a model call already in flight may finish and incur cost
+
+Before every model call, Lens reserves a conservative amount against the monthly lens budget. Successful calls reconcile to reported cost where pricing is available. Interrupted calls retain their reservation because the provider may have charged. A scan stops when the next reservation would exceed the limit, so it can stop with some budget remaining. Lens budgets are separate from virtual-key budgets; analysis calls use the proxy router directly
+
+V1 requires ClickHouse for both sources. It does not reconstruct sessions from unrelated trace IDs, guarantee exhaustive reviews, cache all per-execution observations across scans, or automatically fix agent code. Trace contents can change as late spans arrive, even though a job's selected IDs are fixed. Findings should be reviewed by a person before acting on them

deploy/lens/compose.build.yaml

@@ -0,0 +1,6 @@
+services:
+  lens-worker:
+    build:
+      context: ../..
+      dockerfile: deploy/lens/Dockerfile
+    image: litellm-lens-worker:local

deploy/lens/compose.yaml

@@ -0,0 +1,10 @@
+services:
+  lens-worker:
+    image: ${LENS_WORKER_IMAGE:-ghcr.io/berriai/litellm-lens-worker@sha256:47445afedfb6de2ae37a3a246ea1c939196bfd365436a880ab96ecf5f42b2342}
+    environment:
+      LITELLM_URL: ${LITELLM_URL:?Set the URL reachable from this container}
+      LENS_WORKER_TOKEN: ${LENS_WORKER_TOKEN:?Create a worker credential in the Lens UI}
+    restart: unless-stopped
+    read_only: true
+    cap_drop: [ALL]
+    security_opt: [no-new-privileges:true]

docker/Dockerfile.non_root

@@ -103,7 +103,7 @@ ENV LITELLM_NON_ROOT=true
 
 RUN mkdir -p /var/lib/litellm/ui /var/lib/litellm/assets && \
     cp -r /app/litellm/proxy/_experimental/out/. /var/lib/litellm/ui/ && \
-    cp /app/litellm/proxy/logo.jpg /var/lib/litellm/assets/logo.jpg && \
+    cp /app/litellm/proxy/logo.png /var/lib/litellm/assets/logo.png && \
     touch /var/lib/litellm/ui/.litellm_ui_ready
 
 RUN --mount=type=cache,target=/app/.cache/uv,id=litellm-uv-cache \

docker/docker-compose.tracing.yml

@@ -0,0 +1,62 @@
+name: litellm-tracing
+
+services:
+  litellm:
+    build:
+      context: ..
+      target: runtime
+    command: ["--config", "/app/tracing-config.yaml", "--port", "4000"]
+    environment:
+      LITELLM_MASTER_KEY: local-tracing-master-key
+      LITELLM_SALT_KEY: sk-local-tracing-salt-key
+      DATABASE_URL: postgresql://litellm:litellm@db:5432/litellm
+      STORE_MODEL_IN_DB: "True"
+      CLICKHOUSE_URL: http://default:local-tracing@clickhouse:8123
+      CLICKHOUSE_READER_URL: http://default:local-tracing@clickhouse:8123
+      CLICKHOUSE_DATABASE: litellm
+      OPENAI_API_KEY: ${OPENAI_API_KEY:-}
+    volumes:
+      - ./tracing-config.yaml:/app/tracing-config.yaml:ro
+    ports:
+      - "127.0.0.1:4002:4000"
+    depends_on:
+      db:
+        condition: service_healthy
+      clickhouse:
+        condition: service_healthy
+
+  db:
+    image: postgres:16
+    environment:
+      POSTGRES_DB: litellm
+      POSTGRES_USER: litellm
+      POSTGRES_PASSWORD: litellm
+    volumes:
+      - postgres_data:/var/lib/postgresql/data
+    ports:
+      - "127.0.0.1:15432:5432"
+    healthcheck:
+      test: ["CMD-SHELL", "pg_isready -U litellm -d litellm"]
+      interval: 5s
+      timeout: 5s
+      retries: 10
+
+  clickhouse:
+    image: clickhouse/clickhouse-server:26.9.6.6
+    environment:
+      CLICKHOUSE_USER: default
+      CLICKHOUSE_PASSWORD: local-tracing
+      CLICKHOUSE_DEFAULT_ACCESS_MANAGEMENT: "1"
+    volumes:
+      - clickhouse_data:/var/lib/clickhouse
+    ports:
+      - "127.0.0.1:18123:8123"
+    healthcheck:
+      test: ["CMD", "clickhouse-client", "--user", "default", "--password", "local-tracing", "--query", "SELECT 1"]
+      interval: 5s
+      timeout: 5s
+      retries: 20
+
+volumes:
+  postgres_data:
+  clickhouse_data:

docker/tracing-config.yaml

@@ -0,0 +1,10 @@
+model_list:
+  - model_name: gpt-6.1-sol
+    litellm_params:
+      model: openai/gpt-6.1-sol
+      api_key: os.environ/OPENAI_API_KEY
+
+general_settings:
+  master_key: os.environ/LITELLM_MASTER_KEY
+  tracing:
+    store: clickhouse

enterprise/litellm_enterprise/proxy/management_endpoints/project_endpoints.py

@@ -22,10 +22,8 @@ from litellm._uuid import uuid
 from litellm.proxy._types import *
 from litellm.proxy.auth.auth_checks import delete_cached_project_object
 from litellm.proxy.auth.user_api_key_auth import user_api_key_auth
-from litellm.proxy.management_endpoints.common_utils import (
-    _is_user_team_admin,  # pyright: ignore[reportPrivateUsage]  # shared owner of team-admin membership
-    _set_object_metadata_field,
-)
+from litellm.proxy.management.teams.access import is_team_admin
+from litellm.proxy.management_endpoints.common_utils import _set_object_metadata_field
 from litellm.proxy.management_endpoints.team_admin_field_permissions import team_admin_may_manage_projects
 from litellm.proxy.management_helpers.utils import (
     management_endpoint_wrapper,
@@ -117,7 +115,7 @@ async def _check_user_permission_for_project(
         return False
 
     team: Final = LiteLLM_TeamTable.model_validate(team_row.model_dump())
-    return _is_user_team_admin(user_api_key_dict, team) or user_api_key_dict.user_id in (team.admins or [])
+    return is_team_admin(user_api_key_dict, team) or user_api_key_dict.user_id in (team.admins or [])
 
 
 async def _validate_team_exists(

enterprise/pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "litellm-enterprise"
-version = "0.1.71"
+version = "0.1.72"
 description = "Package for LiteLLM Enterprise features"
 readme = "README.md"
 requires-python = ">=3.9"
@@ -26,7 +26,7 @@ required-version = ">=0.10.9"
 module-root = ""
 
 [tool.commitizen]
-version = "0.1.71"
+version = "0.1.72"
 version_files = [
     "pyproject.toml:^version",
     "../pyproject.toml:litellm-enterprise==",

gateway/routes/allowlist.py

@@ -73,6 +73,7 @@ GATEWAY_PATH_PREFIXES: tuple[str, ...] = (
     "/v1/containers",
     "/containers",
     "/v1/evals",
+    "/v1/traces",
     "/v1/memory",
     "/queue/chat/",
     # Google data plane (v1beta is the Google AI Studio version)

litellm-proxy-extras/litellm_proxy_extras/migrations/20260921190000_agent_identity/migration.sql

@@ -0,0 +1,97 @@
+-- AlterTable
+ALTER TABLE "LiteLLM_AgentsTable" ADD COLUMN IF NOT EXISTS "enabled" BOOLEAN NOT NULL DEFAULT true,
+ADD COLUMN IF NOT EXISTS "execution_mode" TEXT NOT NULL DEFAULT 'autonomous',
+ADD COLUMN IF NOT EXISTS "identity_managed" BOOLEAN NOT NULL DEFAULT false;
+
+-- AlterTable
+ALTER TABLE "LiteLLM_SpendLogs" ADD COLUMN IF NOT EXISTS "billing_agent_id" TEXT;
+
+-- CreateTable
+CREATE TABLE IF NOT EXISTS "LiteLLM_AgentIdentity" (
+    "agent_id" TEXT NOT NULL,
+    "active" BOOLEAN NOT NULL DEFAULT true,
+    "provider" TEXT NOT NULL,
+    "issuer" TEXT NOT NULL,
+    "tenant_id" TEXT NOT NULL,
+    "client_id" TEXT NOT NULL,
+    "service_principal_id" TEXT,
+    "required_roles" TEXT[] DEFAULT ARRAY[]::TEXT[],
+    "required_scopes" TEXT[] DEFAULT ARRAY['user_impersonation']::TEXT[],
+    "revision" TEXT NOT NULL,
+    "last_authenticated_at" TIMESTAMP(3),
+
+    CONSTRAINT "LiteLLM_AgentIdentity_pkey" PRIMARY KEY ("agent_id")
+);
+
+-- CreateTable
+CREATE TABLE IF NOT EXISTS "LiteLLM_RetiredAgentIdentity" (
+    "binding_id" TEXT NOT NULL,
+    "agent_id" TEXT,
+    "provider" TEXT NOT NULL,
+    "issuer" TEXT NOT NULL,
+    "tenant_id" TEXT NOT NULL,
+    "client_id" TEXT NOT NULL,
+
+    CONSTRAINT "LiteLLM_RetiredAgentIdentity_pkey" PRIMARY KEY ("binding_id")
+);
+
+-- CreateTable
+CREATE TABLE IF NOT EXISTS "LiteLLM_RetiredAgent" (
+    "original_agent_id" TEXT NOT NULL,
+    "retired_at" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
+
+    CONSTRAINT "LiteLLM_RetiredAgent_pkey" PRIMARY KEY ("original_agent_id")
+);
+
+-- CreateTable
+CREATE TABLE IF NOT EXISTS "LiteLLM_VerifiedSubject" (
+    "subject_id" TEXT NOT NULL,
+    "issuer" TEXT NOT NULL,
+    "tenant_id" TEXT NOT NULL,
+    "oid" TEXT NOT NULL,
+    "kind" TEXT NOT NULL DEFAULT 'human',
+    "user_id" TEXT,
+    "verified_via" TEXT NOT NULL DEFAULT 'sso_interactive',
+    "verified_at" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
+
+    CONSTRAINT "LiteLLM_VerifiedSubject_pkey" PRIMARY KEY ("subject_id")
+);
+
+-- CreateIndex
+CREATE UNIQUE INDEX IF NOT EXISTS "LiteLLM_AgentIdentity_provider_tenant_id_client_id_key" ON "LiteLLM_AgentIdentity"("provider", "tenant_id", "client_id");
+
+-- CreateIndex
+CREATE UNIQUE INDEX IF NOT EXISTS "LiteLLM_AgentIdentity_issuer_service_principal_id_key" ON "LiteLLM_AgentIdentity"("issuer", "service_principal_id");
+
+-- CreateIndex
+CREATE UNIQUE INDEX IF NOT EXISTS "LiteLLM_RetiredAgentIdentity_provider_tenant_id_client_id_key" ON "LiteLLM_RetiredAgentIdentity"("provider", "tenant_id", "client_id");
+
+-- CreateIndex
+CREATE INDEX IF NOT EXISTS "LiteLLM_VerifiedSubject_user_id_idx" ON "LiteLLM_VerifiedSubject"("user_id");
+
+-- CreateIndex
+CREATE UNIQUE INDEX IF NOT EXISTS "LiteLLM_VerifiedSubject_issuer_tenant_id_oid_key" ON "LiteLLM_VerifiedSubject"("issuer", "tenant_id", "oid");
+
+-- AddForeignKey
+DO $$
+BEGIN
+    IF NOT EXISTS (SELECT 1 FROM pg_constraint WHERE conname = 'LiteLLM_AgentIdentity_agent_id_fkey') THEN
+        ALTER TABLE "LiteLLM_AgentIdentity" ADD CONSTRAINT "LiteLLM_AgentIdentity_agent_id_fkey" FOREIGN KEY ("agent_id") REFERENCES "LiteLLM_AgentsTable"("agent_id") ON DELETE CASCADE ON UPDATE CASCADE;
+    END IF;
+END $$;
+
+-- AddForeignKey
+DO $$
+BEGIN
+    IF NOT EXISTS (SELECT 1 FROM pg_constraint WHERE conname = 'LiteLLM_RetiredAgentIdentity_agent_id_fkey') THEN
+        ALTER TABLE "LiteLLM_RetiredAgentIdentity" ADD CONSTRAINT "LiteLLM_RetiredAgentIdentity_agent_id_fkey" FOREIGN KEY ("agent_id") REFERENCES "LiteLLM_AgentsTable"("agent_id") ON DELETE SET NULL ON UPDATE CASCADE;
+    END IF;
+END $$;
+
+-- AddForeignKey
+DO $$
+BEGIN
+    IF NOT EXISTS (SELECT 1 FROM pg_constraint WHERE conname = 'LiteLLM_VerifiedSubject_user_id_fkey') THEN
+        ALTER TABLE "LiteLLM_VerifiedSubject" ADD CONSTRAINT "LiteLLM_VerifiedSubject_user_id_fkey" FOREIGN KEY ("user_id") REFERENCES "LiteLLM_UserTable"("user_id") ON DELETE CASCADE ON UPDATE CASCADE;
+    END IF;
+END $$;

litellm-proxy-extras/litellm_proxy_extras/migrations/20260930000000_agent_engine/migration.sql

@@ -0,0 +1,10 @@
+CREATE TABLE IF NOT EXISTS "LiteLLM_Engine" (
+    "id" TEXT NOT NULL PRIMARY KEY,
+    "version" INTEGER NOT NULL DEFAULT 0,
+    "data" JSONB NOT NULL
+);
+CREATE TABLE IF NOT EXISTS "LiteLLM_EngineWorker" (
+    "id" TEXT NOT NULL PRIMARY KEY,
+    "token_hash" TEXT NOT NULL UNIQUE,
+    "data" JSONB NOT NULL
+);

litellm-proxy-extras/litellm_proxy_extras/schema.prisma

@@ -78,6 +78,11 @@ model LiteLLM_AgentsTable {
   object_permission_id String?
   object_permission LiteLLM_ObjectPermissionTable? @relation(fields: [object_permission_id], references: [object_permission_id])
   spend      Float    @default(0.0)
+  identity_managed Boolean @default(false)
+  enabled Boolean @default(true)
+  execution_mode String @default("autonomous")
+  identity LiteLLM_AgentIdentity?
+  retired_identities LiteLLM_RetiredAgentIdentity[]
   tpm_limit  Int?
   rpm_limit  Int?
   session_tpm_limit Int?
@@ -88,6 +93,56 @@ model LiteLLM_AgentsTable {
   updated_by String
 }
 
+model LiteLLM_AgentIdentity {
+  agent_id String @id
+  active Boolean @default(true)
+  agent LiteLLM_AgentsTable @relation(fields: [agent_id], references: [agent_id], onDelete: Cascade)
+  provider String
+  issuer String
+  tenant_id String
+  client_id String
+  service_principal_id String?
+  required_roles String[] @default([])
+  required_scopes String[] @default(["user_impersonation"])
+  revision String @default(uuid())
+  last_authenticated_at DateTime?
+  @@unique([provider, tenant_id, client_id])
+  @@unique([issuer, service_principal_id])
+}
+
+model LiteLLM_RetiredAgentIdentity {
+  binding_id String @id @default(uuid())
+  agent_id String?
+  agent LiteLLM_AgentsTable? @relation(fields: [agent_id], references: [agent_id], onDelete: SetNull)
+  provider String
+  issuer String
+  tenant_id String
+  client_id String
+  @@unique([provider, tenant_id, client_id])
+}
+
+model LiteLLM_RetiredAgent {
+  original_agent_id String @id
+  retired_at DateTime @default(now())
+}
+
+model LiteLLM_VerifiedSubject {
+  subject_id String @id @default(uuid())
+  issuer String
+  tenant_id String
+  oid String
+  kind String @default("human")
+  user_id String?
+  user LiteLLM_UserTable? @relation(fields: [user_id], references: [user_id], onDelete: Cascade)
+  verified_via String @default("sso_interactive")
+  verified_at DateTime @default(now())
+  @@unique([issuer, tenant_id, oid])
+  @@index([user_id])
+}
+
+
+
+
 model LiteLLM_OrganizationTable {
 		organization_id String @id @default(uuid())
     organization_alias  String
@@ -241,6 +296,7 @@ model LiteLLM_DeletedTeamTable {
 
 // Track spend, rate limit, budget Users
 model LiteLLM_UserTable {
+    verified_subjects LiteLLM_VerifiedSubject[]
 		user_id    String @id
     user_alias String? 
     team_id    String?
@@ -675,6 +731,7 @@ model LiteLLM_SpendLogs {
   session_id          String?
   status              String?
   mcp_namespaced_tool_name String? 
+  billing_agent_id    String?
   agent_id            String?
   proxy_server_request Json?     @default("{}")
   litellm_call_id     String?
@@ -1837,3 +1894,15 @@ model LiteLLM_WorkflowMessage {
   @@unique([run_id, sequence_number])
   @@index([run_id])
 }
+
+model LiteLLM_Engine {
+  id      String @id
+  version Int @default(0)
+  data    Json
+}
+
+model LiteLLM_EngineWorker {
+  id         String @id
+  token_hash String @unique
+  data       Json
+}

litellm-proxy-extras/pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "litellm-proxy-extras"
-version = "0.4.102"
+version = "0.4.103"
 description = "Additional files for the LiteLLM Proxy. Reduces the size of the main litellm package."
 readme = "README.md"
 requires-python = ">=3.9"
@@ -26,7 +26,7 @@ required-version = ">=0.10.9"
 module-root = ""
 
 [tool.commitizen]
-version = "0.4.102"
+version = "0.4.103"
 version_files = [
     "pyproject.toml:^version",
     "../pyproject.toml:litellm-proxy-extras==",

litellm-rust/Cargo.lock

@@ -1274,6 +1274,18 @@ version = "0.4.33"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "6e8ccc4ea9f6acc32d102c0f6d471d11d913ad15f20c04de743374861fa1d414"
 
+[[package]]
+name = "const-hex"
+version = "1.19.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "0e59eef12462b0f9b0a3620219be5d639afd79fe39dff0a42c3997061f9298b4"
+dependencies = [
+ "cfg-if",
+ "cpufeatures 0.2.17",
+ "proptest",
+ "serde_core",
+]
+
 [[package]]
 name = "const-oid"
 version = "0.9.6"
@@ -2372,9 +2384,9 @@ dependencies = [
  "http-body-util",
  "hyper 1.10.1",
  "lazy_static",
- "opentelemetry",
+ "opentelemetry 0.32.0",
  "opentelemetry-semantic-conventions",
- "opentelemetry_sdk",
+ "opentelemetry_sdk 0.32.1",
  "percent-encoding",
  "pin-project",
  "prost",
@@ -3619,7 +3631,6 @@ dependencies = [
  "litellm-auth",
  "litellm-host",
  "litellm-host-python",
- "litellm-types",
  "proptest",
  "pyo3",
  "rstest",
@@ -3658,9 +3669,9 @@ dependencies = [
  "litellm-host-native",
  "litellm-http",
  "litellm-llms",
+ "litellm-llms-types",
  "litellm-secrets",
  "litellm-tracing",
- "litellm-types",
  "mime_guess",
  "moka",
  "rand 0.8.7",
@@ -3688,13 +3699,12 @@ name = "litellm-core-utils"
 version = "0.1.0"
 dependencies = [
  "fancy-regex 0.19.2",
+ "litellm-llms-types",
  "litellm-tracing",
- "litellm-types",
  "rstest",
  "serde",
  "serde_json",
  "serde_path_to_error",
- "serde_with",
  "strum",
  "thiserror 2.0.19",
  "url",
@@ -3824,9 +3834,9 @@ dependencies = [
  "litellm-host-http",
  "litellm-http",
  "litellm-llms",
+ "litellm-llms-types",
  "litellm-router",
  "litellm-secrets",
- "litellm-types",
  "rstest",
  "serde",
  "serde_json",
@@ -3999,9 +4009,9 @@ dependencies = [
  "litellm-framing",
  "litellm-host",
  "litellm-http",
+ "litellm-llms-types",
  "litellm-python-compat",
  "litellm-secrets",
- "litellm-types",
  "reqwest 0.12.28",
  "rstest",
  "serde",
@@ -4015,13 +4025,26 @@ dependencies = [
  "url",
 ]
 
+[[package]]
+name = "litellm-llms-types"
+version = "0.1.0"
+dependencies = [
+ "macro_rules_attribute",
+ "rstest",
+ "schemars 1.2.2",
+ "serde",
+ "serde_json",
+ "serde_with",
+ "strum",
+]
+
 [[package]]
 name = "litellm-model-catalog"
 version = "0.1.0"
 dependencies = [
  "indexmap 2.14.0",
  "jsonschema",
- "litellm-types",
+ "litellm-llms-types",
  "rstest",
  "schemars 1.2.2",
  "serde",
@@ -4059,12 +4082,13 @@ dependencies = [
  "litellm-host-python",
  "litellm-http",
  "litellm-llms",
+ "litellm-llms-types",
  "litellm-secrets",
  "litellm-secrets-aws",
  "litellm-secrets-types",
  "litellm-token-counter",
+ "litellm-traces",
  "litellm-tracing",
- "litellm-types",
  "pyo3",
  "pyo3-async-runtimes",
  "qdrant-client",
@@ -4340,6 +4364,26 @@ dependencies = [
  "tiktoken-rs",
 ]
 
+[[package]]
+name = "litellm-traces"
+version = "0.1.0"
+dependencies = [
+ "base64 0.22.1",
+ "flate2",
+ "litellm-http",
+ "opentelemetry-proto",
+ "prost",
+ "rstest",
+ "serde",
+ "serde_json",
+ "sha2 0.10.9",
+ "testcontainers-modules",
+ "thiserror 2.0.19",
+ "time",
+ "tokio",
+ "url",
+]
+
 [[package]]
 name = "litellm-tracing"
 version = "0.1.0"
@@ -4354,17 +4398,6 @@ dependencies = [
  "tracing-subscriber",
 ]
 
-[[package]]
-name = "litellm-types"
-version = "0.1.0"
-dependencies = [
- "rstest",
- "schemars 1.2.2",
- "serde",
- "serde_json",
- "strum",
-]
-
 [[package]]
 name = "litemap"
 version = "0.8.2"
@@ -4760,6 +4793,33 @@ dependencies = [
  "tracing",
 ]
 
+[[package]]
+name = "opentelemetry"
+version = "0.33.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "6cdb0b1b267eb9db3331b434ed9ddab10d50e280a9adf9d13e5233e2002b61b5"
+dependencies = [
+ "futures-core",
+ "futures-sink",
+ "js-sys",
+ "pin-project-lite",
+ "thiserror 2.0.19",
+]
+
+[[package]]
+name = "opentelemetry-proto"
+version = "0.33.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "25da1ac11a0aeccf38d7f77ee0348715adaf8340f65ad46c94a02c6b20e2f65d"
+dependencies = [
+ "base64 0.22.1",
+ "const-hex",
+ "opentelemetry 0.33.0",
+ "opentelemetry_sdk 0.33.0",
+ "prost",
+ "serde",
+]
+
 [[package]]
 name = "opentelemetry-semantic-conventions"
 version = "0.32.1"
@@ -4775,7 +4835,23 @@ dependencies = [
  "futures-channel",
  "futures-executor",
  "futures-util",
- "opentelemetry",
+ "opentelemetry 0.32.0",
+ "percent-encoding",
+ "portable-atomic",
+ "rand 0.9.5",
+ "thiserror 2.0.19",
+]
+
+[[package]]
+name = "opentelemetry_sdk"
+version = "0.33.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "cb39533d9d1c912123efd7d41d7e0c29d16917b60ce15b4c8d87cb1af7f67520"
+dependencies = [
+ "futures-channel",
+ "futures-executor",
+ "futures-util",
+ "opentelemetry 0.33.0",
  "percent-encoding",
  "portable-atomic",
  "rand 0.9.5",
@@ -5704,6 +5780,7 @@ checksum = "16a1cfa75cc186dd73d5818e510e042e40927bccc9c236b061cea97e1eb08029"
 dependencies = [
  "base64 0.23.1",
  "bytes",
+ "encoding_rs",
  "futures-core",
  "futures-util",
  "h2 0.4.15",
@@ -5715,6 +5792,7 @@ dependencies = [
  "hyper-util",
  "js-sys",
  "log",
+ "mime",
  "percent-encoding",
  "pin-project-lite",
  "quinn",
@@ -6945,6 +7023,7 @@ dependencies = [
  "memchr",
  "parse-display",
  "pin-project-lite",
+ "reqwest 0.13.5",
  "serde",
  "serde_json",
  "serde_with",
@@ -7505,7 +7584,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "adbc64cba7137545b8044cb1fe9814f7aacf3c6b5f9b45be8bb5db538befdb26"
 dependencies = [
  "js-sys",
- "opentelemetry",
+ "opentelemetry 0.32.0",
  "tracing",
  "tracing-core",
  "tracing-subscriber",

litellm-rust/Cargo.toml

@@ -12,6 +12,7 @@ repository = "https://github.com/BerriAI/litellm"
 litellm-config = { path = "crates/config" }
 litellm-router = { path = "crates/router" }
 litellm-tracing = { path = "crates/tracing" }
+litellm-traces = { path = "crates/traces" }
 litellm-core = { path = "crates/core" }
 litellm-gateway-mcp = { path = "crates/gateway-mcp" }
 litellm-gateway = { path = "crates/gateway" }
@@ -39,7 +40,7 @@ litellm-secrets-azure = { path = "crates/secrets-azure" }
 litellm-secrets-cyberark = { path = "crates/secrets-cyberark" }
 litellm-http = { path = "crates/http" }
 litellm-llms = { path = "crates/llms" }
-litellm-types = { path = "crates/types" }
+litellm-llms-types = { path = "crates/llms-types" }
 litellm-core-utils = { path = "crates/core-utils" }
 litellm-db = { path = "crates/db" }
 litellm-db-testing = { path = "crates/db-testing" }
@@ -74,6 +75,7 @@ proptest = "1.7.0"
 pyo3 = "0.29.2"
 pyo3-async-runtimes = { version = "0.29.0", features = ["tokio-runtime"] }
 rand = "0.8"
+macro_rules_attribute = "0.2.3"
 schemars = "1"
 reqwest = { version = "0.12", default-features = false, features = ["json", "multipart", "rustls-tls", "http2", "stream"] }
 qdrant-client = { version = "1.19.0", default-features = false }

litellm-rust/crates/cache-response/AGENTS.md

@@ -24,6 +24,6 @@ Keep unary caching independent of stream-only methods. Store streams only after
 
 Test each contract in its owner: storage capabilities in backend tests, envelopes and freshness here, reuse and replay in core, Python callback and fallback behavior at the bridge, and HTTP behavior at the gateway. Run backend contract checks and Python response-codec fixtures before exposing a new backend
 
-`ScopedCache` requires an explicit shared or isolated scope at construction. `CacheOptions` has no default sharing policy. Callers may override policy per invocation without replacing the attached service. Versioned native envelopes reject incompatible API surfaces and versions as misses; this envelope is distinct from the legacy Python response codec
+`ScopedCache` requires an explicit shared or isolated scope at construction. Per-call `CachePolicy` controls reads, writes, expiry, and freshness without replacing the attached scope or service. `CacheOptions` binds that policy to an explicit scope for storage requests and has no default sharing policy. Versioned native envelopes reject incompatible API surfaces and versions as misses; this envelope is distinct from the legacy Python response codec
 
 Response storage is not the source of budget or rate-limit coordination dependencies. Keep counters, reservations, and atomic admission operations out of `ResponseCacheService`, including when both services happen to use Redis

litellm-rust/crates/cache-response/src/lib.rs

@@ -17,6 +17,6 @@ pub use exact::{ConnectionProbe, ExactResponseCache};
 pub use response::{ResponseCache, ResponseCacheRequest};
 
 pub use service::{
-    CacheOptions, CacheScope, ResponseCacheConfig, ResponseCacheService, ResponseEnvelope,
-    ScopedCache,
+    CacheOptions, CachePolicy, CacheScope, ResponseCacheConfig, ResponseCacheService,
+    ResponseEnvelope, ScopedCache,
 };

litellm-rust/crates/cache-response/src/service.rs

@@ -73,32 +73,35 @@ pub enum CacheScope {
     Isolated(String),
 }
 
-#[derive(Clone)]
-pub struct CacheOptions {
+#[derive(Clone, Copy, Default)]
+pub struct CachePolicy {
     pub caching: Option<bool>,
     pub no_cache: bool,
     pub no_store: bool,
     pub ttl: Option<Duration>,
     pub max_age: Option<Duration>,
+}
+
+impl CachePolicy {
+    pub fn enabled(&self) -> bool {
+        self.caching != Some(false) && !(self.no_cache && self.no_store)
+    }
+}
+
+#[derive(Clone)]
+pub struct CacheOptions {
+    pub policy: CachePolicy,
     pub scope: CacheScope,
 }
 
 impl CacheOptions {
     pub fn new(scope: CacheScope) -> Self {
         Self {
-            caching: None,
-            no_cache: false,
-            no_store: false,
-            ttl: None,
-            max_age: None,
+            policy: CachePolicy::default(),
             scope,
         }
     }
 
-    pub fn enabled(&self) -> bool {
-        self.caching != Some(false) && !(self.no_cache && self.no_store)
-    }
-
     pub fn request(self, namespace: &str, surface: &str, mut input: Value) -> ResponseCacheRequest {
         input.sort_all_objects();
         let scope = match self.scope {
@@ -128,13 +131,15 @@ impl CacheOptions {
                 supported_call_type: true,
                 native_backend: true,
                 default_on: true,
-                caching: self.caching,
-                no_cache: self.no_cache,
-                no_store: self.no_store,
+                caching: self.policy.caching,
+                no_cache: self.policy.no_cache,
+                no_store: self.policy.no_store,
                 ..Default::default()
             },
-            context: ExactCacheContext { ttl: self.ttl },
-            max_age: self.max_age,
+            context: ExactCacheContext {
+                ttl: self.policy.ttl,
+            },
+            max_age: self.policy.max_age,
         }
     }
 }
@@ -171,7 +176,10 @@ impl ScopedCache {
         Self { service, scope }
     }
 
-    pub fn options(&self, overrides: Option<CacheOptions>) -> CacheOptions {
-        overrides.unwrap_or_else(|| CacheOptions::new(self.scope.clone()))
+    pub fn options(&self, policy: Option<CachePolicy>) -> CacheOptions {
+        CacheOptions {
+            policy: policy.unwrap_or_default(),
+            scope: self.scope.clone(),
+        }
     }
 }

litellm-rust/crates/cache-response/tests/service.rs

@@ -129,11 +129,20 @@ async fn isolated_policy_controls_actual_entry_reuse(
     #[case] first: &str,
     #[case] second: &str,
     #[case] hit: bool,
+    #[values(false, true)] override_policy: bool,
 ) {
-    use litellm_cache_response::{CacheOptions, CacheScope};
-    let service = ResponseCache::new(Arc::new(InMemoryCache::<CacheEntry>::default()));
-    let request =
-        |scope| CacheOptions::new(scope).request("test", "messages", json!({"prompt":"hello"}));
+    use litellm_cache_response::{CachePolicy, CacheScope, ScopedCache};
+    let service = Arc::new(ResponseCache::new(Arc::new(
+        InMemoryCache::<CacheEntry>::default(),
+    )));
+    let request = |scope| {
+        ScopedCache::new(service.clone(), scope)
+            .options(override_policy.then_some(CachePolicy {
+                ttl: Some(Duration::from_secs(30)),
+                ..CachePolicy::default()
+            }))
+            .request("test", "messages", json!({"prompt":"hello"}))
+    };
     service
         .async_store(
             &request(CacheScope::Isolated(first.into())),

litellm-rust/crates/callbacks-legacy-python/AGENTS.md

@@ -7,6 +7,7 @@
   - The enum only shrinks: when Rust owns a subsystem, delete its group rather than adding a Rust path beside it
   - Calling a user's own callback directly is permanent Python surface and gets its own type outside `LegacyPython`
 - `PublicCall` is the caller's call as `Logging` sees it: the positional arguments, the keyword view as the call rewrites it (setup, deployment hook, preflight) and the bound request object backing omitted keywords; shared bridge composition hands it to `LegacyLogging`; routes use the neutral call boundary
+- `LoggingOperation` selects legacy logging entrypoints and response handling. It belongs here rather than in shared inference data contracts
 - `setup` reuses a `Logging` passed as `litellm_logging_obj` (the proxy and Router) and otherwise builds one through `function_setup`; which callbacks run is `Logging`'s decision, never this crate's
 - Callbacks receive the caller's own objects and may mutate them; this crate alone carries that obligation
   - Retain complete boundary arguments, opaque values, aliases, omitted/default distinctions and deliberate copies; preserve the deployment-hook kwargs view

litellm-rust/crates/callbacks-legacy-python/Cargo.toml

@@ -6,7 +6,6 @@ license.workspace = true
 repository.workspace = true
 
 [dependencies]
-litellm-types.workspace = true
 litellm-host.workspace = true
 litellm-host-python.workspace = true
 

litellm-rust/crates/callbacks-legacy-python/src/adapter.rs

@@ -2,8 +2,8 @@
 //! raises is answered with the same `Logging` calls, in the same order, as the Python
 //! `@client` path makes them.
 
+use crate::LoggingOperation;
 use litellm_host_python::PythonOwned;
-use litellm_types::Operation;
 
 use litellm_host::{
     interceptors::{RawResponse, RequestContext, WireRequest},
@@ -45,7 +45,7 @@ struct LoggedRequest {
 }
 
 pub struct LegacyLogging {
-    operation: Operation,
+    operation: LoggingOperation,
     call: PublicCall,
     logger: Option<PythonLogger>,
     start: Py<PyAny>,
@@ -68,7 +68,12 @@ fn is_cancellation(py: Python<'_>, error: &PyErr) -> bool {
 }
 
 impl LegacyLogging {
-    pub fn new(py: Python<'_>, operation: Operation, call: PublicCall, asynchronous: bool) -> Self {
+    pub fn new(
+        py: Python<'_>,
+        operation: LoggingOperation,
+        call: PublicCall,
+        asynchronous: bool,
+    ) -> Self {
         Self {
             operation,
             call,
@@ -87,32 +92,34 @@ impl LegacyLogging {
 
     fn call_type(&self) -> &'static str {
         match (self.operation, self.asynchronous) {
-            (Operation::Completion, false) => "completion",
-            (Operation::Completion, true) => "acompletion",
-            (Operation::Responses, false) => "responses",
-            (Operation::Responses, true) => "aresponses",
-            (Operation::Messages, _) => "anthropic_messages",
-            (Operation::Ocr, false) => "ocr",
-            (Operation::Ocr, true) => "aocr",
+            (LoggingOperation::Completion, false) => "completion",
+            (LoggingOperation::Completion, true) => "acompletion",
+            (LoggingOperation::Responses, false) => "responses",
+            (LoggingOperation::Responses, true) => "aresponses",
+            (LoggingOperation::Messages, _) => "anthropic_messages",
+            (LoggingOperation::Ocr, false) => "ocr",
+            (LoggingOperation::Ocr, true) => "aocr",
         }
     }
 
     fn input_description(&self) -> &'static str {
         match self.operation {
-            Operation::Completion => "Chat completions",
-            Operation::Responses => "Responses",
-            Operation::Messages => "Messages",
-            Operation::Ocr => "OCR document processing",
+            LoggingOperation::Completion => "Chat completions",
+            LoggingOperation::Responses => "Responses",
+            LoggingOperation::Messages => "Messages",
+            LoggingOperation::Ocr => "OCR document processing",
         }
     }
 
     fn stream_billing(&self) -> Option<PassThroughStream> {
         match self.operation {
-            Operation::Messages => Some(PassThroughStream {
+            LoggingOperation::Messages => Some(PassThroughStream {
                 url_route: "/v1/messages",
                 endpoint_type: "anthropic",
             }),
-            Operation::Completion | Operation::Responses | Operation::Ocr => None,
+            LoggingOperation::Completion | LoggingOperation::Responses | LoggingOperation::Ocr => {
+                None
+            }
         }
     }
 
@@ -643,16 +650,16 @@ kwargs = {'logger': logger, 'document': document}
     }
 
     #[rstest]
-    #[case::sync_completion(litellm_types::Operation::Completion, false, "completion")]
-    #[case::async_completion(litellm_types::Operation::Completion, true, "acompletion")]
-    #[case::sync_responses(litellm_types::Operation::Responses, false, "responses")]
-    #[case::async_responses(litellm_types::Operation::Responses, true, "aresponses")]
-    #[case::sync_messages(litellm_types::Operation::Messages, false, "anthropic_messages")]
-    #[case::async_messages(litellm_types::Operation::Messages, true, "anthropic_messages")]
-    #[case::sync_ocr(litellm_types::Operation::Ocr, false, "ocr")]
-    #[case::async_ocr(litellm_types::Operation::Ocr, true, "aocr")]
+    #[case::sync_completion(crate::LoggingOperation::Completion, false, "completion")]
+    #[case::async_completion(crate::LoggingOperation::Completion, true, "acompletion")]
+    #[case::sync_responses(crate::LoggingOperation::Responses, false, "responses")]
+    #[case::async_responses(crate::LoggingOperation::Responses, true, "aresponses")]
+    #[case::sync_messages(crate::LoggingOperation::Messages, false, "anthropic_messages")]
+    #[case::async_messages(crate::LoggingOperation::Messages, true, "anthropic_messages")]
+    #[case::sync_ocr(crate::LoggingOperation::Ocr, false, "ocr")]
+    #[case::async_ocr(crate::LoggingOperation::Ocr, true, "aocr")]
     fn operation_selects_the_legacy_setup_and_deployment_hook_contract(
-        #[case] operation: litellm_types::Operation,
+        #[case] operation: crate::LoggingOperation,
         #[case] asynchronous: bool,
         #[case] expected: &str,
     ) {
@@ -1088,12 +1095,12 @@ check = lambda: None
     }
 
     #[rstest]
-    #[case::completion(litellm_types::Operation::Completion, "Chat completions")]
-    #[case::responses(litellm_types::Operation::Responses, "Responses")]
-    #[case::messages(litellm_types::Operation::Messages, "Messages")]
-    #[case::ocr(litellm_types::Operation::Ocr, "OCR document processing")]
+    #[case::completion(crate::LoggingOperation::Completion, "Chat completions")]
+    #[case::responses(crate::LoggingOperation::Responses, "Responses")]
+    #[case::messages(crate::LoggingOperation::Messages, "Messages")]
+    #[case::ocr(crate::LoggingOperation::Ocr, "OCR document processing")]
     fn prepared_arguments_replace_the_legacy_view_without_losing_callback_aliases(
-        #[case] operation: litellm_types::Operation,
+        #[case] operation: crate::LoggingOperation,
         #[case] description: &str,
     ) {
         Python::initialize();
@@ -1763,7 +1770,7 @@ assert logger.calls[1][1] is response
         Python::attach(|py| {
             let locals = namespace(py, c"first = b'first'\nlast = b'last'\nresponse = None");
             let mut logging = LegacyLogging {
-                operation: litellm_types::Operation::Messages,
+                operation: crate::LoggingOperation::Messages,
                 ..logged(py, &locals, true)
             };
             logging

litellm-rust/crates/callbacks-legacy-python/src/lib.rs

@@ -20,5 +20,13 @@ pub(crate) use callbacks::{LegacyCallbacks, is_internal_call};
 pub(crate) use logger::{DeploymentHooks, PythonLogger, finalize, setup};
 pub use mapping::{CallBoundary, CallbackMapping, Dispatch, callback_mappings};
 
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub enum LoggingOperation {
+    Completion,
+    Responses,
+    Messages,
+    Ocr,
+}
+
 #[cfg(test)]
 mod test_support;

litellm-rust/crates/callbacks-legacy-python/src/test_support.rs

@@ -189,5 +189,5 @@ pub(crate) fn legacy_call(
         .map(|kwargs| kwargs.cast_into::<PyDict>().unwrap())
         .unwrap_or_else(|| PyDict::new(py));
     let call = PublicCall::capture(&request, &PyTuple::empty(py), &kwargs).unwrap();
-    LegacyLogging::new(py, litellm_types::Operation::Ocr, call, asynchronous)
+    LegacyLogging::new(py, crate::LoggingOperation::Ocr, call, asynchronous)
 }

litellm-rust/crates/core-utils/Cargo.toml

@@ -8,11 +8,10 @@ repository.workspace = true
 [dependencies]
 fancy-regex.workspace = true
 litellm-tracing.workspace = true
-litellm-types.workspace = true
+litellm-llms-types.workspace = true
 serde.workspace = true
 serde_json.workspace = true
 serde_path_to_error = "0.1"
-serde_with.workspace = true
 strum.workspace = true
 thiserror.workspace = true
 url.workspace = true

litellm-rust/crates/core-utils/src/core_helpers.rs

@@ -2,7 +2,7 @@
 
 use std::time::{SystemTime, UNIX_EPOCH};
 
-use litellm_types::utils::{ChatCompletionsUsage, PromptTokensDetails};
+use litellm_llms_types::formats::chat_completions::{ChatCompletionsUsage, PromptTokensDetails};
 
 /// OpenAI finish reasons, mirroring Python's `_FINISH_REASON_MAP` for the
 /// reasons the providers on this route can emit. Python warns and falls back to

litellm-rust/crates/core-utils/src/get_provider_specific_headers.rs

@@ -1,4 +1,4 @@
-use litellm_types::utils::{ProviderSpecificHeader, ProviderSpecificHeaders};
+use litellm_llms_types::headers::{ProviderSpecificHeader, ProviderSpecificHeaders};
 use serde_json::{Map, Value};
 
 pub fn get_provider_specific_headers(

litellm-rust/crates/core-utils/src/prompt_templates/factory.rs

@@ -10,7 +10,7 @@
 //! `_bedrock_converse_messages_pt` for the text-only surface this route
 //! accepts; anything richer is declined upstream by the capability gate.
 
-use litellm_types::llms::openai::{ChatMessage, ChatMessageContent};
+use litellm_llms_types::formats::chat_completions::{ChatMessage, ChatMessageContent};
 use strum::IntoStaticStr;
 
 pub const EMPTY_TEXT_PLACEHOLDER: &str =

litellm-rust/crates/core-utils/src/serde_compat.rs

@@ -1,12 +1,3 @@
-use serde::{
-    Deserializer,
-    de::{Error, Visitor},
-};
-use serde_with::DeserializeAs;
-
-pub struct LaxI64;
-pub struct FiniteF64;
-
 pub fn parse_str_bool(value: &str) -> Option<bool> {
     let token = value.trim_matches(|character: char| {
         character.is_whitespace() || matches!(character, '\u{1c}'..='\u{1f}')
@@ -22,129 +13,12 @@ pub fn parse_redis_bool(value: &str) -> bool {
     value == "1" || value.eq_ignore_ascii_case("true") || value.eq_ignore_ascii_case("yes")
 }
 
-impl<'de> DeserializeAs<'de, i64> for LaxI64 {
-    fn deserialize_as<D: Deserializer<'de>>(deserializer: D) -> Result<i64, D::Error> {
-        deserializer.deserialize_any(Self)
-    }
-}
-
-impl<'de> Visitor<'de> for LaxI64 {
-    type Value = i64;
-
-    fn expecting(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
-        formatter.write_str("an integer in the i64 range")
-    }
-
-    fn visit_i64<E: Error>(self, value: i64) -> Result<i64, E> {
-        Ok(value)
-    }
-
-    fn visit_u64<E: Error>(self, value: u64) -> Result<i64, E> {
-        i64::try_from(value).map_err(E::custom)
-    }
-
-    fn visit_f64<E: Error>(self, value: f64) -> Result<i64, E> {
-        integral_float(value).ok_or_else(|| E::custom("expected an integer in the i64 range"))
-    }
-
-    fn visit_str<E: Error>(self, value: &str) -> Result<i64, E> {
-        integer_string(value.trim())
-            .ok_or_else(|| E::custom("expected an integer in the i64 range"))
-    }
-
-    fn visit_bool<E: Error>(self, value: bool) -> Result<i64, E> {
-        Ok(i64::from(value))
-    }
-}
-
-impl<'de> DeserializeAs<'de, f64> for FiniteF64 {
-    fn deserialize_as<D: Deserializer<'de>>(deserializer: D) -> Result<f64, D::Error> {
-        deserializer.deserialize_any(Self)
-    }
-}
-
-impl<'de> Visitor<'de> for FiniteF64 {
-    type Value = f64;
-
-    fn expecting(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
-        formatter.write_str("a finite number")
-    }
-
-    fn visit_i64<E: Error>(self, value: i64) -> Result<f64, E> {
-        Ok(value as f64)
-    }
-
-    fn visit_u64<E: Error>(self, value: u64) -> Result<f64, E> {
-        Ok(value as f64)
-    }
-
-    fn visit_f64<E: Error>(self, value: f64) -> Result<f64, E> {
-        value
-            .is_finite()
-            .then_some(value)
-            .ok_or_else(|| E::custom("expected a finite number"))
-    }
-
-    fn visit_str<E: Error>(self, value: &str) -> Result<f64, E> {
-        self.visit_f64(value.trim().parse::<f64>().map_err(E::custom)?)
-    }
-
-    fn visit_bool<E: Error>(self, value: bool) -> Result<f64, E> {
-        Ok(f64::from(value))
-    }
-}
-
-fn integer_string(value: &str) -> Option<i64> {
-    let integer = match value.split_once('.') {
-        Some((integer, fraction)) => {
-            if fraction.is_empty() || !fraction.bytes().all(|byte| byte == b'0') {
-                return None;
-            }
-            integer
-        }
-        None => value,
-    };
-    if integer.starts_with('_') || integer.ends_with('_') || integer.contains("__") {
-        return None;
-    }
-    let digits = integer.strip_prefix(['+', '-']).unwrap_or(integer);
-    if digits.is_empty()
-        || digits.starts_with('_')
-        || !digits
-            .bytes()
-            .all(|byte| byte.is_ascii_digit() || byte == b'_')
-    {
-        return None;
-    }
-    integer.replace('_', "").parse().ok()
-}
-
-fn integral_float(value: f64) -> Option<i64> {
-    (value.is_finite()
-        && value.fract() == 0.0
-        && value >= i64::MIN as f64
-        && value < -(i64::MIN as f64))
-        .then_some(value as i64)
-}
-
 #[cfg(test)]
 mod tests {
     use rstest::rstest;
-    use serde::{Deserialize, Serialize};
-    use serde_json::json;
-    use serde_with::serde_as;
 
     use super::*;
 
-    #[serde_as]
-    #[derive(Debug, Deserialize, Serialize, PartialEq)]
-    struct Numbers {
-        #[serde_as(deserialize_as = "Option<Vec<LaxI64>>")]
-        integers: Option<Vec<i64>>,
-        #[serde_as(deserialize_as = "Option<FiniteF64>")]
-        float: Option<f64>,
-    }
-
     #[rstest]
     #[case::trimmed_true(" True ", Some(true))]
     #[case::control_whitespace_true("\u{1c}TRUE\u{1f}", Some(true))]
@@ -160,73 +34,4 @@ mod tests {
     ) {
         assert_eq!(parse_str_bool(input), expected, "{input:?}");
     }
-
-    #[test]
-    fn adapters_compose_and_serialize_as_numbers() {
-        let numbers: Numbers = serde_json::from_value(json!({
-            "integers": ["9007199254740993.0", "1_000", " +2.000 ", 3.0, true],
-            "float": " 1.5 "
-        }))
-        .unwrap();
-        assert_eq!(
-            serde_json::to_value(numbers).unwrap(),
-            json!({
-                "integers": [9_007_199_254_740_993_i64, 1000, 2, 3, 1], "float": 1.5
-            })
-        );
-        for input in [json!({}), json!({"integers": null, "float": null})] {
-            assert_eq!(
-                serde_json::from_value::<Numbers>(input).unwrap(),
-                Numbers {
-                    integers: None,
-                    float: None,
-                }
-            );
-        }
-    }
-
-    #[test]
-    fn integer_bounds_and_invalid_values_are_checked() {
-        for input in [
-            json!(i64::MIN),
-            json!(i64::MAX),
-            json!(i64::MAX.to_string()),
-        ] {
-            assert!(serde_json::from_value::<Numbers>(json!({"integers": [input]})).is_ok());
-        }
-        for input in [
-            json!(u64::MAX),
-            json!(9_223_372_036_854_775_808_u64),
-            json!(9_223_372_036_854_775_808.0),
-            json!("-9223372036854775809"),
-            json!("1.0000000000000001"),
-            json!("1e3"),
-            json!("2."),
-            json!(".0"),
-            json!("_2"),
-            json!("2__0"),
-            json!(2.5),
-            json!(null),
-            json!({}),
-        ] {
-            assert!(serde_json::from_value::<Numbers>(json!({"integers": [input]})).is_err());
-        }
-    }
-
-    #[test]
-    fn floats_reject_nonfinite_and_invalid_values() {
-        for input in [
-            json!("NaN"),
-            json!("inf"),
-            json!("-inf"),
-            json!("1e999"),
-            json!([]),
-        ] {
-            assert!(serde_json::from_value::<Numbers>(json!({"float": input})).is_err());
-        }
-        for (input, expected) in [(json!(2), 2.0), (json!(2.5), 2.5), (json!(true), 1.0)] {
-            let numbers: Numbers = serde_json::from_value(json!({"float": input})).unwrap();
-            assert_eq!(numbers.float, Some(expected));
-        }
-    }
 }

litellm-rust/crates/core/AGENTS.md

@@ -10,11 +10,11 @@ Responses WebSocket sessions remain separate from the HTTP call driver because a
 
 ## Crate layering
 
-For Messages, Responses, Chat Completions, OCR, and other API formats, `core/src/<format>/` owns orchestration. Shared API data contracts belong in `litellm-types`, adapter contracts and shared transformation machinery in `llms/src/base_llm/<format>/`, and provider policy in `llms/src/<provider>/<format>/`. A repeated format directory name does not imply interchangeable responsibilities. Select concrete adapters here, then invoke their contracts instead of applying one provider's policy to every call. Route types describe call envelopes and execution state, not duplicate public payload schemas
+For Messages, Responses, Chat Completions, OCR, and other API formats, `core/src/<format>/` owns orchestration. Shared API data contracts belong in `litellm-llms-types`, adapter contracts and shared transformation machinery in `llms/src/base_llm/<format>/`, and provider policy in `llms/src/<provider>/<format>/`. A repeated format directory name does not imply interchangeable responsibilities. Select concrete adapters here, then invoke their contracts instead of applying one provider's policy to every call. Route types describe call envelopes and execution state, not duplicate public payload schemas
 
-Each crate mirrors one top-level Python package, so a Rust path reads as its Python path with the crate name in place of the package directory. Dependencies only point down:
+Crates separate API data, transformations, transport, and orchestration. Python package names identify counterparts, not ownership. Dependencies only point down:
 
-- `litellm-types` mirrors `litellm/types/`: pure serde data, no I/O
+- `litellm-llms-types` owns shared inference API contracts, grouped by format: pure serde data and shape validation, no I/O
 - `litellm-core-utils` mirrors `litellm/litellm_core_utils/`: pure helpers (provider resolution, prompt factory, call arguments, settings lookup and layer merge), no network I/O
 - `litellm-http` is Rust-only and route-neutral: settings resolution, the pooled `reqwest` clients, TLS, proxies, the SSRF-safe media fetcher, request and header helpers, and transport errors. Python's `litellm/llms/custom_httpx/` is split by responsibility instead of mirrored: its transport half lives here, its OCR handler in `litellm-llms`
 - `litellm-llms` mirrors `litellm/llms/`: `base_llm/<api>/transformation.rs`, `<provider>/<api>/transformation.rs`, and `base_llm/ocr/handler.rs` (the OCR request handler)
@@ -36,7 +36,9 @@ Not here: serving HTTP (axum routes, extractors), config file reading, rollout s
 
 ## Response caching and accounting boundary
 
-Attach a `litellm_cache_response::ScopedCache` with `route.with_cache(cache)`. Cached and uncached routes use the same `execute` and `machine` methods. `CallOptions` carries per-call cache overrides and observation; attaching a service does not change the execution contract
+Attach a `litellm_cache_response::ScopedCache` with `route.with_cache(cache)`. Cached and uncached routes use the same `execute` and `machine` methods. `CallOptions` carries a scope-free `CachePolicy` and observation; per-call policy never replaces the attached scope or service
+
+Messages groups per-call dependencies in `CallContext` and explicitly sequences cache lookup, provider execution, result acceptance, and cache storage. Provider transport does not own cache orchestration. Stream capture remains in the shared cache implementation
 
 Core owns request identity, typed response reconstruction and stream capture/replay. `cache-response` owns cache policy, namespacing, scope encoding, versioned envelopes and freshness. The SDK explicitly chooses shared scope. The gateway derives isolated scope from authenticated identity before attaching its service
 

litellm-rust/crates/core/Cargo.toml

@@ -11,7 +11,7 @@ litellm-cache-response.workspace = true
 litellm-framing.workspace = true
 tokio-util = { version = "0.7", features = ["codec"] }
 litellm-secrets.workspace = true
-litellm-types.workspace = true
+litellm-llms-types.workspace = true
 litellm-core-utils.workspace = true
 litellm-host.workspace = true
 bytes.workspace = true

litellm-rust/crates/core/src/caching.rs

@@ -1,5 +1,6 @@
 use std::{
     future::Future,
+    marker::PhantomData,
     sync::Arc,
     time::{Duration, SystemTime, UNIX_EPOCH},
 };
@@ -7,7 +8,8 @@ use std::{
 use bytes::{Bytes, BytesMut};
 use futures_util::{StreamExt, TryStreamExt, stream};
 use litellm_cache_response::{
-    CacheOptions, ResponseCacheRequest, ResponseCacheService, ResponseEnvelope, cache_key,
+    CacheOptions, CachePolicy, ResponseCacheRequest, ResponseCacheService, ResponseEnvelope,
+    ScopedCache, cache_key,
 };
 use litellm_host::{
     call::{CallOutput, OutputOf},
@@ -77,7 +79,7 @@ impl CacheSession {
         options: Option<CacheOptions>,
         request: &CacheRequest,
     ) -> Option<Self> {
-        let options = options.filter(CacheOptions::enabled)?;
+        let options = options.filter(|options| options.policy.enabled())?;
         let service = service?;
         let input = request.input.clone();
         let request = options.request(&service.config().namespace, P::SURFACE, input);
@@ -198,81 +200,137 @@ where
     let identity = request.identity.clone();
     crate::diagnostic::provider(&identity.model, &identity.provider);
     let session = CacheSession::prepare::<P>(cache, options, &request);
-    let hit = match &session {
-        Some(session) => session.lookup::<P>().await.and_then(|entry| {
-            let output = match entry {
-                CachedOutput::Response(response) => Some(CallOutput::Complete(response)),
-                CachedOutput::Stream(data) => P::replay(Bytes::from(data)),
-            };
-            output.map(|output| (output, cache_key(&session.request.key)))
-        }),
-        None => None,
+    let cache = CallCache::<P> {
+        session,
+        protocol: PhantomData,
     };
+    let hit = cache.lookup().await;
     let (output, source) = match hit {
-        Some((output, key)) => (output, ResultSource::Cache { key }),
+        Some(hit) => hit,
         None => (provider().await?, ResultSource::Provider),
     };
-    let from_provider = source == ResultSource::Provider;
     publish(
         ExecutionFacts {
             provider: identity,
-            source,
+            source: source.clone(),
         },
         interceptors,
         observers,
     )
     .await?;
-    let Some(session) =
-        session.filter(|session| from_provider && session.request.controls.writes())
-    else {
-        return Ok(output);
-    };
-    match output {
-        CallOutput::Complete(response) => {
-            session.store_response::<P>(&response).await;
-            Ok(CallOutput::Complete(response))
-        }
-        CallOutput::Stream { head, chunks } => {
-            let captured = stream::try_unfold(
-                (chunks, Some(Vec::<u8>::new()), session),
-                |(mut chunks, captured, session)| async move {
-                    match chunks.try_next().await? {
-                        Some(chunk) => {
-                            let captured = captured.and_then(|mut data| {
-                                let bytes = P::bytes(&chunk);
-                                if data.len().saturating_add(bytes.len())
-                                    > session.service.config().max_entry_bytes
-                                {
-                                    return None;
-                                }
-                                data.extend_from_slice(bytes);
-                                Some(data)
-                            });
-                            Ok(Some((chunk, (chunks, captured, session))))
-                        }
-                        None => {
-                            if let Some(data) = captured
-                                && let Ok(text) = String::from_utf8(data)
-                                && successful_stream(&text, P::TERMINAL_EVENT)
-                                && let Ok(entry) = serde_json::to_value(ResponseEnvelope::new(
-                                    P::SURFACE,
-                                    CachedOutput::<Value>::Stream(text),
-                                ))
-                            {
-                                session.store(entry).await;
-                            }
-                            Ok::<_, RouteError>(None)
-                        }
-                    }
-                },
-            )
-            .boxed();
-            Ok(CallOutput::Stream {
-                head,
-                chunks: captured,
+    Ok(cache.finish(output, &source).await)
+}
+
+pub(crate) struct CallCache<P> {
+    session: Option<CacheSession>,
+    protocol: PhantomData<P>,
+}
+
+impl<P: StreamCachable> CallCache<P> {
+    pub(crate) fn from_wire(
+        cache: Option<&ScopedCache>,
+        policy: CachePolicy,
+        identity: &ProviderIdentity,
+        wire: &WireRequest,
+    ) -> Self {
+        let session = cache.and_then(|cache| {
+            if !policy.enabled() {
+                return None;
+            }
+            let options = cache.options(Some(policy));
+            let request = CacheRequest::from_wire(identity.clone(), Some(wire));
+            Some(CacheSession {
+                request: options.request(
+                    &cache.service.config().namespace,
+                    P::SURFACE,
+                    request.input,
+                ),
+                service: cache.service.clone(),
             })
+        });
+        Self {
+            session,
+            protocol: PhantomData,
         }
     }
+
+    pub(crate) async fn lookup(&self) -> Option<(OutputOf<P>, ResultSource)>
+    where
+        P::Response: DeserializeOwned,
+    {
+        let session = self.session.as_ref()?;
+        let output = match session.lookup::<P>().await? {
+            CachedOutput::Response(response) => CallOutput::Complete(response),
+            CachedOutput::Stream(data) => P::replay(Bytes::from(data))?,
+        };
+        Some((
+            output,
+            ResultSource::Cache {
+                key: cache_key(&session.request.key),
+            },
+        ))
+    }
+
+    pub(crate) async fn finish(self, output: OutputOf<P>, source: &ResultSource) -> OutputOf<P>
+    where
+        P::Response: Serialize,
+    {
+        let Some(session) = self.session.filter(|session| {
+            *source == ResultSource::Provider && session.request.controls.writes()
+        }) else {
+            return output;
+        };
+        match output {
+            CallOutput::Complete(response) => {
+                session.store_response::<P>(&response).await;
+                CallOutput::Complete(response)
+            }
+            CallOutput::Stream { head, chunks } => CallOutput::Stream {
+                head,
+                chunks: capture_stream::<P>(chunks, session),
+            },
+        }
+    }
+}
+
+fn capture_stream<P: StreamCachable>(
+    chunks: futures_util::stream::BoxStream<'static, Result<P::Chunk, RouteError>>,
+    session: CacheSession,
+) -> futures_util::stream::BoxStream<'static, Result<P::Chunk, RouteError>> {
+    stream::try_unfold(
+        (chunks, Some(Vec::<u8>::new()), session),
+        |(mut chunks, captured, session)| async move {
+            match chunks.try_next().await? {
+                Some(chunk) => {
+                    let captured = captured.and_then(|mut data| {
+                        let bytes = P::bytes(&chunk);
+                        if data.len().saturating_add(bytes.len())
+                            > session.service.config().max_entry_bytes
+                        {
+                            return None;
+                        }
+                        data.extend_from_slice(bytes);
+                        Some(data)
+                    });
+                    Ok(Some((chunk, (chunks, captured, session))))
+                }
+                None => {
+                    if let Some(data) = captured
+                        && let Ok(text) = String::from_utf8(data)
+                        && successful_stream(&text, P::TERMINAL_EVENT)
+                        && let Ok(entry) = serde_json::to_value(ResponseEnvelope::new(
+                            P::SURFACE,
+                            CachedOutput::<Value>::Stream(text),
+                        ))
+                    {
+                        session.store(entry).await;
+                    }
+                    Ok::<_, RouteError>(None)
+                }
+            }
+        },
+    )
+    .boxed()
 }
 
 fn now() -> Duration {

litellm-rust/crates/core/src/chat_completions/handler.rs

@@ -1,5 +1,4 @@
-use litellm_host::lifecycle::ExecutionEvent;
-use litellm_host::observation::ObservationSender;
+use litellm_host::{lifecycle::ExecutionEvent, observation::ObservationSender};
 use std::time::Duration;
 
 use litellm_auth::AuthServices;
@@ -9,7 +8,7 @@ use litellm_llms::base_llm::{
     auth::{Authenticated, resolve_auth},
     chat::transformation::ProviderChatResponseData,
 };
-use litellm_types::utils::ChatCompletionsResponse;
+use litellm_llms_types::formats::chat_completions::ChatCompletionsResponse;
 use serde_json::Value;
 
 use super::Error;
@@ -23,7 +22,7 @@ pub(super) async fn execute(
     auth: &AuthServices,
     request: ProviderChatCompletionsRequest,
     cache: Option<litellm_cache_response::ScopedCache>,
-    cache_options: Option<litellm_cache_response::CacheOptions>,
+    cache_options: Option<litellm_cache_response::CachePolicy>,
     interceptors: &impl Interceptors<Error>,
     observers: Option<&ObservationSender>,
 ) -> Result<ChatCompletionsResponse, Error> {

litellm-rust/crates/core/src/chat_completions/mod.rs

@@ -5,7 +5,7 @@ pub use crate::error::RouteError as Error;
 mod common_utils;
 pub(crate) mod handler;
 mod prepare;
-use litellm_types::utils::ChatCompletionsResponse;
+use litellm_llms_types::formats::chat_completions::ChatCompletionsResponse;
 use prepare::{prepare_provider_request, resolve_request};
 
 use crate::chat_completions::types::ChatCompletionsRequest;
@@ -67,7 +67,7 @@ impl ChatCompletionsRoute {
     async fn run(
         &self,
         request: ChatCompletionsRequest<'_>,
-        cache_options: Option<litellm_cache_response::CacheOptions>,
+        cache_options: Option<litellm_cache_response::CachePolicy>,
         interceptors: &impl litellm_host::interceptors::Interceptors<Error>,
         observers: Option<&ObservationSender>,
     ) -> Result<ChatCompletionsResponse, Error> {

litellm-rust/crates/core/src/chat_completions/prepare.rs

@@ -2,8 +2,8 @@ use litellm_auth::SecretValue;
 use litellm_core_utils::settings::Lookup;
 use litellm_http::request::with_default_headers;
 use litellm_llms::base_llm::{auth::ValidatedEnvironment, chat::transformation::BaseConfig};
+use litellm_llms_types::formats::chat_completions::ChatMessage;
 use litellm_secrets::source::Secrets;
-use litellm_types::llms::openai::ChatMessage;
 use serde_json::Value;
 
 use super::{

litellm-rust/crates/core/src/chat_completions/route.rs

@@ -5,7 +5,7 @@ use litellm_host::{
     call::{CallOutput, HostedMachine, hosted_call},
     protocol::Protocol,
 };
-use litellm_types::utils::ChatCompletionsResponse;
+use litellm_llms_types::formats::chat_completions::ChatCompletionsResponse;
 
 use super::{
     ChatCompletionsRoute, Error,
@@ -55,7 +55,7 @@ impl ChatCompletionsRoute {
     pub(super) async fn run_call(
         &self,
         call: ChatCompletionsCall,
-        cache_options: Option<litellm_cache_response::CacheOptions>,
+        cache_options: Option<litellm_cache_response::CachePolicy>,
         interceptors: &impl litellm_host::interceptors::Interceptors<Error>,
         observers: Option<&ObservationSender>,
     ) -> Result<ChatCompletionsResponse, Error> {

litellm-rust/crates/core/src/chat_completions/types.rs

@@ -3,7 +3,7 @@ use std::time::Duration;
 
 use litellm_auth::SecretValue;
 use litellm_llms::base_llm::{auth::ValidatedEnvironment, chat::transformation::BaseConfig};
-use litellm_types::llms::openai::ChatMessage;
+use litellm_llms_types::formats::chat_completions::ChatMessage;
 use serde_json::{Map, Value};
 
 /// A `/chat/completions` call as it crosses into the core.

litellm-rust/crates/core/src/context.rs

@@ -0,0 +1,48 @@
+use litellm_cache_response::CachePolicy;
+use litellm_host::{
+    interceptors::{ExecutionFacts, Interceptors, RawResponse},
+    lifecycle::{CallEvent, ExecutionEvent},
+    observation::ObservationSender,
+};
+
+use crate::{CallOptions, RouteError};
+
+pub(crate) struct CallContext<'a, I> {
+    pub interceptors: &'a I,
+    pub observers: Option<ObservationSender>,
+    pub cache: CachePolicy,
+}
+
+impl<'a, I: Interceptors<RouteError>> CallContext<'a, I> {
+    pub fn new(interceptors: &'a I, options: CallOptions) -> Self {
+        Self {
+            interceptors,
+            observers: options.observers,
+            cache: options.cache.unwrap_or_default(),
+        }
+    }
+
+    pub async fn result_ready(&self, facts: ExecutionFacts) -> Result<(), RouteError> {
+        if let Some(observers) = &self.observers {
+            observers.emit(CallEvent::Execution(ExecutionEvent::ResultReady {
+                facts: facts.clone(),
+            }));
+        }
+        self.interceptors.result_ready(facts).await
+    }
+
+    pub async fn response_received(&self, body: &str) -> Result<(), RouteError> {
+        let raw = RawResponse {
+            body: body.to_owned(),
+        };
+        if let Some(observers) = &self.observers {
+            observers.emit(CallEvent::Execution(
+                ExecutionEvent::ProviderResponseReceived { raw: raw.clone() },
+            ));
+        }
+        self.interceptors
+            .after_provider_response(raw)
+            .await
+            .map_err(RouteError::post_call)
+    }
+}

litellm-rust/crates/core/src/lib.rs

@@ -1,3 +1,4 @@
+mod context;
 mod diagnostic;
 
 pub mod audio_transcription;
@@ -16,7 +17,7 @@ pub use error::RouteError;
 
 #[derive(Clone, Default)]
 pub struct CallOptions {
-    pub cache: Option<litellm_cache_response::CacheOptions>,
+    pub cache: Option<litellm_cache_response::CachePolicy>,
     pub observers: Option<litellm_host::observation::ObservationSender>,
 }
 
@@ -29,8 +30,8 @@ impl From<Option<litellm_host::observation::ObservationSender>> for CallOptions
     }
 }
 
-impl From<litellm_cache_response::CacheOptions> for CallOptions {
-    fn from(cache: litellm_cache_response::CacheOptions) -> Self {
+impl From<litellm_cache_response::CachePolicy> for CallOptions {
+    fn from(cache: litellm_cache_response::CachePolicy) -> Self {
         Self {
             cache: Some(cache),
             observers: None,

litellm-rust/crates/core/src/messages/AGENTS.md

@@ -1,4 +1,4 @@
-This directory owns provider-independent Messages call orchestration: the entrypoint, call envelopes, provider selection, credential resolution, transport coordination, hooks, and stream lifecycle. Shared API data contracts belong in `litellm-types::messages`, adapter contracts and execution inputs in `llms/src/base_llm/messages`, and provider implementations in `llms/src/<provider>/messages`
+This directory owns provider-independent Messages call orchestration: the entrypoint, call envelopes, provider selection, credential resolution, transport coordination, hooks, and stream lifecycle. Shared API data contracts belong in `litellm-llms-types::formats::messages`, adapter contracts and execution inputs in `llms/src/base_llm/messages`, and provider implementations in `llms/src/<provider>/messages`
 
 Select concrete provider adapters and invoke their contracts. Delegate authentication policy, beta selection, payload rewriting, and response interpretation to those adapters. Keep provider policy out of request preparation and transport handlers. Calling a concrete provider helper for every provider is still a policy dependency
 

litellm-rust/crates/core/src/messages/common_utils.rs

@@ -3,7 +3,7 @@ pub(super) use litellm_http::request::truncate_error_body;
 use litellm_llms::{
     anthropic::messages::transformation::ANTHROPIC_MESSAGES_CONFIG,
     azure_ai::messages::transformation::AZURE_ANTHROPIC_MESSAGES_CONFIG,
-    base_llm::messages::transformation::BaseAnthropicMessagesConfig,
+    base_llm::messages::transformation::BaseMessagesConfig,
     bedrock::messages::invoke_transformations::anthropic_claude3_transformation::BEDROCK_ANTHROPIC_MESSAGES_CONFIG,
 };
 use serde_json::{Map, Value};
@@ -30,7 +30,7 @@ impl MessagesProvider {
         .into()
     }
 
-    pub(crate) fn config(self) -> &'static dyn BaseAnthropicMessagesConfig {
+    pub(crate) fn config(self) -> &'static dyn BaseMessagesConfig {
         match self {
             Self::Anthropic => &ANTHROPIC_MESSAGES_CONFIG,
             Self::AzureAi => &AZURE_ANTHROPIC_MESSAGES_CONFIG,

litellm-rust/crates/core/src/messages/handler.rs

@@ -1,119 +1,145 @@
-use litellm_host::lifecycle::ExecutionEvent;
-use litellm_host::observation::ObservationSender;
 use std::time::Duration;
 
 use bytes::Bytes;
 use futures_util::{StreamExt, TryStreamExt, stream::BoxStream};
-use litellm_auth::AuthServices;
-use litellm_host::interceptors::{Interceptors, RawResponse, RequestContext, WireRequest};
+use litellm_host::interceptors::{Interceptors, ProviderIdentity, RequestContext, WireRequest};
 use litellm_http::transport::Error as TransportError;
 use litellm_llms::base_llm::{
     auth::{Authenticated, resolve_auth},
     messages::{
         streaming::{ByteStream, StreamDecoder, encode_anthropic_sse},
-        transformation::BaseAnthropicMessagesConfig,
+        transformation::BaseMessagesConfig,
     },
 };
+use litellm_llms_types::formats::messages::MessagesResponse;
 use litellm_tracing::ByteChunk;
-use litellm_types::llms::anthropic_messages::anthropic_response::AnthropicMessagesResponse;
 use serde_json::Value;
 
 use super::{
-    Error, MessagesResponse, common_utils::truncate_error_body, prepare::ProviderMessagesRequest,
+    Error, MessagesCallResponse, MessagesRoute, common_utils::truncate_error_body,
+    prepare::ProviderMessagesRequest,
 };
-use crate::{constants::MESSAGES_TIMEOUT_SECS, outbound::outbound_request};
+use crate::{constants::MESSAGES_TIMEOUT_SECS, context::CallContext, outbound::outbound_request};
 
-pub(super) async fn execute(
-    http: &litellm_http::Client,
-    auth: &AuthServices,
-    request: ProviderMessagesRequest,
-    cache: Option<litellm_cache_response::ScopedCache>,
-    cache_options: Option<litellm_cache_response::CacheOptions>,
-    interceptors: &impl Interceptors<Error>,
-    observers: Option<&ObservationSender>,
-) -> Result<MessagesResponse, Error> {
-    let ProviderMessagesRequest {
-        provider,
-        url,
-        body,
-        environment,
-        timeout,
-        api_key,
-    } = request;
-    let stream = body.params.stream == Some(true);
-    let context = RequestContext {
-        model: body.model.clone(),
-        custom_llm_provider: provider.as_str().to_string(),
-        optional_params: serde_json::to_value(&body.params).map_err(serialize_failure)?,
-        secret_fields: Vec::new(),
-        api_key,
-    };
-    let authenticated = resolve_auth(auth, environment, &|key| std::env::var(key).ok()).await?;
-    let identity = litellm_host::interceptors::ProviderIdentity {
-        model: context.model.clone(),
-        provider: context.custom_llm_provider.clone(),
-    };
-    let wire = interceptors
-        .before_provider_request(
-            WireRequest {
-                url,
-                headers: authenticated.headers,
-                body: serde_json::to_value(&body).map_err(serialize_failure)?,
-            },
-            context,
-        )
-        .await?;
-    let cache = cache.filter(|_| authenticated.signer.is_none());
-    let cache_request =
-        crate::caching::CacheRequest::from_wire(identity, cache.as_ref().map(|_| &wire));
-    crate::caching::execute_streaming::<super::route::Messages, _, _>(
-        cache_request,
-        cache.as_ref().map(|cache| cache.service.clone()),
-        cache.as_ref().map(|cache| cache.options(cache_options)),
-        interceptors,
-        observers,
-        || async move {
-            let provider_name = provider.as_str();
-            log_request_body(provider_name, stream, &wire.body);
-            let response = send(
-                http,
-                Authenticated {
-                    headers: wire.headers,
-                    signer: authenticated.signer,
+pub(super) struct ProviderCall {
+    pub identity: ProviderIdentity,
+    pub wire: WireRequest,
+    provider: super::common_utils::MessagesProvider,
+    signer: Option<litellm_auth_aws::SigV4Signer>,
+    timeout: Option<Duration>,
+    stream: bool,
+}
+
+impl ProviderCall {
+    pub fn cacheable(&self) -> bool {
+        self.signer.is_none()
+    }
+}
+
+impl MessagesRoute {
+    pub(super) async fn prepare_outbound(
+        &self,
+        request: ProviderMessagesRequest,
+        context: &CallContext<'_, impl Interceptors<Error>>,
+    ) -> Result<ProviderCall, Error> {
+        let ProviderMessagesRequest {
+            provider,
+            url,
+            body,
+            environment,
+            timeout,
+            api_key,
+        } = request;
+        let request_context = RequestContext {
+            model: body.model.clone(),
+            custom_llm_provider: provider.as_str().to_string(),
+            optional_params: serde_json::to_value(&body.params).map_err(serialize_failure)?,
+            secret_fields: Vec::new(),
+            api_key,
+        };
+        let authenticated =
+            resolve_auth(&self.auth, environment, &|key| std::env::var(key).ok()).await?;
+        let identity = ProviderIdentity {
+            model: request_context.model.clone(),
+            provider: request_context.custom_llm_provider.clone(),
+        };
+        let wire = context
+            .interceptors
+            .before_provider_request(
+                WireRequest {
+                    url,
+                    headers: authenticated.headers,
+                    body: serde_json::to_value(&body).map_err(serialize_failure)?,
                 },
-                &wire.url,
-                &wire.body,
-                timeout,
+                request_context,
             )
             .await?;
-            if !response.status().is_success() {
-                return Err(provider_error(response).await);
-            }
-            let config = provider.config();
-            if stream {
-                return Ok(streaming_response(
-                    response,
-                    config.stream_decoder(),
-                    provider_name,
+        let stream = match wire.body.get("stream") {
+            None | Some(Value::Null) => false,
+            Some(Value::Bool(stream)) => *stream,
+            Some(value) => {
+                return Err(Error::InvalidRequest(
+                    litellm_llms::ErrorDetail::InvalidValue {
+                        field: "stream",
+                        expected: "a boolean",
+                        actual: value.clone(),
+                    },
                 ));
             }
-            let text = response.text().await.map_err(network)?;
-            log_response_body(&text);
-            let raw = RawResponse { body: text.clone() };
-            if let Some(observers) = observers {
-                observers.emit(litellm_host::lifecycle::CallEvent::Execution(
-                    ExecutionEvent::ProviderResponseReceived { raw: raw.clone() },
-                ));
-            }
-            interceptors
-                .after_provider_response(raw)
-                .await
-                .map_err(Error::post_call)?;
-            decode_response(config, &body.model, &text)
-                .map(|message| MessagesResponse::Complete(Box::new(message)))
-        },
-    )
-    .await
+        };
+        Ok(ProviderCall {
+            identity,
+            wire,
+            provider,
+            signer: authenticated.signer,
+            timeout,
+            stream,
+        })
+    }
+
+    pub(super) async fn call_provider(
+        &self,
+        request: ProviderCall,
+        context: &CallContext<'_, impl Interceptors<Error>>,
+    ) -> Result<MessagesCallResponse, Error> {
+        let ProviderCall {
+            identity,
+            wire,
+            provider,
+            signer,
+            timeout,
+            stream,
+        } = request;
+        let provider_name = provider.as_str();
+        log_request_body(provider_name, stream, &wire.body);
+        let response = send(
+            &self.http,
+            Authenticated {
+                headers: wire.headers,
+                signer,
+            },
+            &wire.url,
+            &wire.body,
+            timeout,
+        )
+        .await?;
+        if !response.status().is_success() {
+            return Err(provider_error(response).await);
+        }
+        let config = provider.config();
+        if stream {
+            return Ok(streaming_response(
+                response,
+                config.stream_decoder(),
+                provider_name,
+            ));
+        }
+        let text = response.text().await.map_err(network)?;
+        log_response_body(&text);
+        context.response_received(&text).await?;
+        decode_response(config, &identity.model, &text)
+            .map(|message| MessagesCallResponse::Complete(Box::new(message)))
+    }
 }
 
 fn serialize_failure(err: serde_json::Error) -> Error {
@@ -158,10 +184,10 @@ async fn provider_error(response: reqwest::Response) -> Error {
 }
 
 fn decode_response(
-    config: &dyn BaseAnthropicMessagesConfig,
+    config: &dyn BaseMessagesConfig,
     model: &str,
     text: &str,
-) -> Result<AnthropicMessagesResponse, Error> {
+) -> Result<MessagesResponse, Error> {
     let response = serde_json::from_str(text).map_err(|err| {
         Error::InvalidResponse(litellm_llms::ErrorDetail::invalid(
             "messages response JSON",
@@ -177,7 +203,7 @@ fn streaming_response(
     response: reqwest::Response,
     decoder: Option<StreamDecoder>,
     provider: &'static str,
-) -> MessagesResponse {
+) -> MessagesCallResponse {
     let headers = response
         .headers()
         .iter()
@@ -194,7 +220,7 @@ fn streaming_response(
         .boxed(),
         Some(decode) => decoded_chunks(response, decode, provider),
     };
-    MessagesResponse::Stream {
+    MessagesCallResponse::Stream {
         head: super::route::MessagesStreamHead { headers },
         chunks,
     }
@@ -268,7 +294,7 @@ mod tests {
             .send()
             .await
             .unwrap();
-        let MessagesResponse::Stream { mut chunks, .. } =
+        let MessagesCallResponse::Stream { mut chunks, .. } =
             streaming_response(response, Some(anthropic_sse_event_stream), "test")
         else {
             panic!("a streaming response returns chunks");

litellm-rust/crates/core/src/messages/mod.rs

@@ -1,16 +1,19 @@
-use litellm_host::observation::ObservationSender;
 mod common_utils;
 mod handler;
 mod prepare;
 pub mod route;
 mod types;
 
+use futures_util::FutureExt;
 use litellm_auth::AuthServices;
+use litellm_host::interceptors::{ExecutionFacts, Interceptors, ResultSource};
+
+use crate::{caching::CallCache, context::CallContext};
 use litellm_secrets::source::SecretSource;
 use std::sync::Arc;
 
 pub use crate::error::RouteError as Error;
-pub use types::{MessagesCall, MessagesResponse, MessagesShaping, messages_body};
+pub use types::{MessagesCall, MessagesCallResponse, MessagesShaping, messages_body};
 
 #[derive(Clone)]
 pub struct MessagesRoute {
@@ -20,74 +23,18 @@ pub struct MessagesRoute {
     cache: Option<litellm_cache_response::ScopedCache>,
 }
 
-#[must_use]
-#[derive(Clone, Default)]
-pub struct MessagesRouteBuilder<Http = (), Auth = (), Secrets = ()> {
-    http: Http,
-    auth: Auth,
-    secrets: Secrets,
-    cache: Option<litellm_cache_response::ScopedCache>,
-}
-
-impl<Http, Auth, Secrets> MessagesRouteBuilder<Http, Auth, Secrets> {
-    pub fn with_http(
-        self,
-        http: litellm_http::Client,
-    ) -> MessagesRouteBuilder<litellm_http::Client, Auth, Secrets> {
-        MessagesRouteBuilder {
-            http,
-            auth: self.auth,
-            secrets: self.secrets,
-            cache: self.cache,
-        }
-    }
-
-    pub fn with_auth(
-        self,
-        auth: Arc<AuthServices>,
-    ) -> MessagesRouteBuilder<Http, Arc<AuthServices>, Secrets> {
-        MessagesRouteBuilder {
-            http: self.http,
-            auth,
-            secrets: self.secrets,
-            cache: self.cache,
-        }
-    }
-
-    pub fn with_secrets(
-        self,
-        secrets: Arc<dyn SecretSource>,
-    ) -> MessagesRouteBuilder<Http, Auth, Arc<dyn SecretSource>> {
-        MessagesRouteBuilder {
-            http: self.http,
-            auth: self.auth,
-            secrets,
-            cache: self.cache,
-        }
-    }
-
-    pub fn with_cache(self, cache: litellm_cache_response::ScopedCache) -> Self {
-        Self {
-            cache: Some(cache),
-            ..self
-        }
-    }
-}
-
-impl MessagesRouteBuilder<litellm_http::Client, Arc<AuthServices>, Arc<dyn SecretSource>> {
-    pub fn build(self) -> MessagesRoute {
-        MessagesRoute {
-            http: self.http,
-            auth: self.auth,
-            secrets: self.secrets,
-            cache: self.cache,
-        }
-    }
-}
-
 impl MessagesRoute {
-    pub fn builder() -> MessagesRouteBuilder {
-        MessagesRouteBuilder::default()
+    pub fn new(
+        http: litellm_http::Client,
+        auth: Arc<AuthServices>,
+        secrets: Arc<dyn SecretSource>,
+    ) -> Self {
+        Self {
+            http,
+            auth,
+            secrets,
+            cache: None,
+        }
     }
 
     #[must_use]
@@ -103,16 +50,10 @@ impl MessagesRoute {
         call: MessagesCall,
         interceptors: &impl litellm_host::interceptors::Interceptors<Error>,
         options: impl Into<crate::CallOptions>,
-    ) -> Result<MessagesResponse, Error> {
-        let crate::CallOptions {
-            cache: cache_options,
-            observers,
-        } = options.into();
-        litellm_host::lifecycle::observe_call(
-            observers.clone(),
-            self.run(call, cache_options, interceptors, observers.as_ref()),
-        )
-        .await
+    ) -> Result<MessagesCallResponse, Error> {
+        let context = CallContext::new(interceptors, options.into());
+        litellm_host::lifecycle::observe_call(context.observers.clone(), self.run(call, context))
+            .await
     }
 
     #[tracing::instrument(name = "litellm.route", skip_all, fields(
@@ -126,36 +67,34 @@ impl MessagesRoute {
     async fn run(
         &self,
         call: MessagesCall,
-        cache_options: Option<litellm_cache_response::CacheOptions>,
-        interceptors: &impl litellm_host::interceptors::Interceptors<Error>,
-        observers: Option<&ObservationSender>,
-    ) -> Result<MessagesResponse, Error> {
+        context: CallContext<'_, impl Interceptors<Error>>,
+    ) -> Result<MessagesCallResponse, Error> {
         crate::diagnostic::call(async {
-            self.run_provider(call, cache_options, interceptors, observers)
-                .await
+            let prepared = prepare::prepare(call, self.secrets.as_ref()).await?;
+            crate::diagnostic::provider(&prepared.body.model, prepared.provider.as_str());
+            let request = self.prepare_outbound(prepared, &context).boxed().await?;
+            let cache = CallCache::<route::Messages>::from_wire(
+                self.cache.as_ref().filter(|_| request.cacheable()),
+                context.cache,
+                &request.identity,
+                &request.wire,
+            );
+            let identity = request.identity.clone();
+            let (output, source) = match cache.lookup().await {
+                Some(hit) => hit,
+                None => (
+                    self.call_provider(request, &context).await?,
+                    ResultSource::Provider,
+                ),
+            };
+            context
+                .result_ready(ExecutionFacts {
+                    provider: identity,
+                    source: source.clone(),
+                })
+                .await?;
+            Ok(cache.finish(output, &source).await)
         })
         .await
     }
-
-    async fn run_provider(
-        &self,
-        call: MessagesCall,
-        cache_options: Option<litellm_cache_response::CacheOptions>,
-        interceptors: &impl litellm_host::interceptors::Interceptors<Error>,
-        observers: Option<&ObservationSender>,
-    ) -> Result<MessagesResponse, Error> {
-        let request = prepare::prepare(call, self.secrets.as_ref()).await?;
-        crate::diagnostic::provider(&request.body.model, request.provider.as_str());
-        let execute: futures_util::future::BoxFuture<'_, Result<MessagesResponse, Error>> =
-            Box::pin(handler::execute(
-                &self.http,
-                &self.auth,
-                request,
-                self.cache.clone(),
-                cache_options,
-                interceptors,
-                observers,
-            ));
-        execute.await
-    }
 }

litellm-rust/crates/core/src/messages/prepare.rs

@@ -9,8 +9,8 @@ use litellm_http::request::with_default_headers;
 use litellm_llms::base_llm::{
     auth::ValidatedEnvironment, messages::context::MessagesTransformContext,
 };
+use litellm_llms_types::formats::messages::MessagesRequest;
 use litellm_secrets::source::SecretSource;
-use litellm_types::llms::anthropic_messages::anthropic_request::AnthropicMessagesRequest;
 
 use super::{
     Error, MessagesCall,
@@ -27,7 +27,7 @@ struct ResolvedProvider {
 pub(super) struct ProviderMessagesRequest {
     pub(super) provider: MessagesProvider,
     pub(super) url: String,
-    pub(super) body: AnthropicMessagesRequest,
+    pub(super) body: MessagesRequest,
     pub(super) environment: ValidatedEnvironment,
     pub(super) timeout: Option<Duration>,
     /// The caller's own credential, reported to the host beside the wire request.
@@ -79,7 +79,7 @@ fn prepare_provider_request(
     let env_lookup = |key: &str| secrets.get(key);
 
     let sanitized = config.shape_request(
-        AnthropicMessagesRequest { model, ..body },
+        MessagesRequest { model, ..body },
         shaping.reasoning_auto_summary,
     )?;
     let trimmed = without_additional_drop_params(sanitized, &shaping.additional_drop_params)?;
@@ -124,9 +124,9 @@ fn prepare_provider_request(
 }
 
 fn without_additional_drop_params(
-    request: AnthropicMessagesRequest,
+    request: MessagesRequest,
     paths: &[String],
-) -> Result<AnthropicMessagesRequest, Error> {
+) -> Result<MessagesRequest, Error> {
     if paths.is_empty() {
         return Ok(request);
     }
@@ -134,7 +134,7 @@ fn without_additional_drop_params(
     let trimmed = paths
         .iter()
         .fold(params, |params, path| delete_nested_value(params, path));
-    Ok(AnthropicMessagesRequest {
+    Ok(MessagesRequest {
         params: serde_json::from_value(trimmed).map_err(invalid_request)?,
         ..request
     })
@@ -143,7 +143,7 @@ fn without_additional_drop_params(
 #[cfg(test)]
 mod tests {
     use litellm_llms::base_llm::auth::resolve_auth;
-    use litellm_types::utils::ProviderSpecificHeaders;
+    use litellm_llms_types::headers::ProviderSpecificHeaders;
     use rstest::{fixture, rstest};
     use serde_json::{Map, Value, json};
 
@@ -155,7 +155,7 @@ mod tests {
         MessagesShaping::default()
     }
 
-    fn body(value: Value) -> AnthropicMessagesRequest {
+    fn body(value: Value) -> MessagesRequest {
         serde_json::from_value(value).unwrap()
     }
 

litellm-rust/crates/core/src/messages/route.rs

@@ -5,11 +5,11 @@ use litellm_host::{
     call::{HostedCompletion, HostedMachine, hosted_call},
     protocol::Protocol,
 };
-use litellm_types::llms::anthropic_messages::anthropic_response::AnthropicMessagesResponse;
+use litellm_llms_types::formats::messages::MessagesResponse;
 
 use super::{Error, MessagesCall};
 
-pub type MessagesOutput = HostedCompletion<Box<AnthropicMessagesResponse>>;
+pub type MessagesOutput = HostedCompletion<Box<MessagesResponse>>;
 
 /// The upstream response as the caller sees it at stream hand-off, before any chunk.
 pub struct MessagesStreamHead {
@@ -19,7 +19,7 @@ pub struct MessagesStreamHead {
 pub struct Messages;
 
 impl Protocol for Messages {
-    type Response = Box<AnthropicMessagesResponse>;
+    type Response = Box<MessagesResponse>;
     type Error = Error;
     type Request = MessagesCall;
     type HostCall = Infallible;
@@ -43,8 +43,14 @@ impl super::MessagesRoute {
             request,
             observers,
             move |call, _, interceptors, observers| async move {
-                self.run(call, cache_options, &interceptors, observers.as_ref())
-                    .await
+                let context = crate::context::CallContext::new(
+                    &interceptors,
+                    crate::CallOptions {
+                        cache: cache_options,
+                        observers,
+                    },
+                );
+                self.run(call, context).await
             },
         )
     }

litellm-rust/crates/core/src/messages/types.rs

@@ -2,12 +2,10 @@ use std::time::Duration;
 
 use bytes::Bytes;
 use litellm_host::call::CallOutput;
-use litellm_llms::base_llm::messages::context::MessagesModelCapabilities as AnthropicModelCapabilities;
-use litellm_types::{
-    llms::anthropic_messages::{
-        anthropic_request::AnthropicMessagesRequest, anthropic_response::AnthropicMessagesResponse,
-    },
-    utils::ProviderSpecificHeaders,
+use litellm_llms::base_llm::messages::context::MessagesModelCapabilities;
+use litellm_llms_types::{
+    formats::messages::{MessagesRequest, MessagesResponse},
+    headers::ProviderSpecificHeaders,
 };
 use serde::{Deserialize, Serialize};
 use serde_json::{Map, Value};
@@ -15,7 +13,7 @@ use serde_json::{Map, Value};
 use super::Error;
 
 pub struct MessagesCall {
-    pub body: AnthropicMessagesRequest,
+    pub body: MessagesRequest,
     pub api_key: Option<String>,
     pub api_base: Option<String>,
     pub custom_llm_provider: Option<String>,
@@ -25,7 +23,7 @@ pub struct MessagesCall {
     pub shaping: MessagesShaping,
 }
 
-pub fn messages_body(body: Map<String, Value>) -> Result<AnthropicMessagesRequest, Error> {
+pub fn messages_body(body: Map<String, Value>) -> Result<MessagesRequest, Error> {
     serde_json::from_value(Value::Object(body)).map_err(invalid_request)
 }
 
@@ -33,13 +31,13 @@ pub(super) fn invalid_request(err: serde_json::Error) -> Error {
     Error::InvalidRequest(format!("invalid Anthropic messages request: {err}").into())
 }
 
-pub type MessagesResponse =
-    CallOutput<Box<AnthropicMessagesResponse>, super::route::MessagesStreamHead, Bytes, Error>;
+pub type MessagesCallResponse =
+    CallOutput<Box<MessagesResponse>, super::route::MessagesStreamHead, Bytes, Error>;
 
 #[derive(Clone, Debug, Default, PartialEq, Serialize, Deserialize)]
 pub struct MessagesShaping {
     #[serde(default)]
-    pub capabilities: AnthropicModelCapabilities,
+    pub capabilities: MessagesModelCapabilities,
     #[serde(default)]
     pub drop_params: bool,
     #[serde(default)]
@@ -76,9 +74,9 @@ mod tests {
     #[case::partial_capabilities(
         json!({"capabilities": {"supports_reasoning": true}}),
         MessagesShaping {
-            capabilities: AnthropicModelCapabilities {
+            capabilities: MessagesModelCapabilities {
                 supports_reasoning: true,
-                ..AnthropicModelCapabilities::default()
+                ..MessagesModelCapabilities::default()
             },
             ..MessagesShaping::default()
         },
@@ -100,7 +98,7 @@ mod tests {
             "additional_drop_params": ["metadata.user_id", "thinking"]
         }),
         MessagesShaping {
-            capabilities: AnthropicModelCapabilities {
+            capabilities: MessagesModelCapabilities {
                 supports_reasoning: true,
                 supports_adaptive_thinking: true,
                 thinking_always_on: false,

litellm-rust/crates/core/src/ocr/client.rs

@@ -2,9 +2,8 @@ use litellm_host::observation::ObservationSender;
 use std::sync::Arc;
 
 use litellm_host::interceptors::Interceptors;
-use litellm_llms::base_llm::ocr::{
-    error::Error, handler::OcrClient, transformation::LiteLLMOcrResponse,
-};
+use litellm_llms::base_llm::ocr::{error::Error, handler::OcrClient};
+use litellm_llms_types::formats::ocr::LiteLLMOcrResponse;
 
 use super::{
     handler::perform_ocr_request,

litellm-rust/crates/core/src/ocr/document.rs

@@ -1,10 +1,8 @@
 use std::{collections::BTreeMap as Map, io::Read, path::Path};
 
 use base64::{Engine, engine::general_purpose::STANDARD};
-use litellm_llms::base_llm::ocr::{
-    error::Error,
-    transformation::{OCR_INLINE_MAX_BYTES, OcrDocument},
-};
+use litellm_llms::base_llm::ocr::{error::Error, transformation::OCR_INLINE_MAX_BYTES};
+use litellm_llms_types::formats::ocr::OcrDocument;
 
 use crate::ocr::types::OcrDocumentInput;
 

litellm-rust/crates/core/src/ocr/handler.rs

@@ -1,12 +1,12 @@
 use futures_util::future::BoxFuture;
 use litellm_host::interceptors::{Interceptors, RawResponse, RequestContext, WireRequest};
-use litellm_host::lifecycle::ExecutionEvent;
-use litellm_host::observation::ObservationSender;
+use litellm_host::{lifecycle::ExecutionEvent, observation::ObservationSender};
 use litellm_llms::base_llm::ocr::{
     error::Error,
     handler::{CallHooks, OcrClient},
-    transformation::{LiteLLMOcrResponse, PreparedOcrRequest},
+    transformation::PreparedOcrRequest,
 };
+use litellm_llms_types::formats::ocr::LiteLLMOcrResponse;
 use serde_json::Value;
 
 use super::{arguments::is_secret_param, prepare::prepare_request, provider_config::OcrConfigKind};

litellm-rust/crates/core/src/ocr/prepare.rs

@@ -85,12 +85,13 @@ mod tests {
         base_llm::ocr::{
             error::Error,
             handler::{CallHooks, OcrClient},
-            transformation::{BaseOcrConfig, OcrResponseFormat},
+            transformation::BaseOcrConfig,
         },
         cohere::ocr::transformation::CohereParseConfig,
         mistral::ocr::transformation::MistralOcrConfig,
         vertex_ai::ocr::transformation::VertexAiOcrConfig,
     };
+    use litellm_llms_types::formats::ocr::OcrResponseFormat;
     use serde_json::{Value, json};
 
     use super::*;

litellm-rust/crates/core/src/ocr/provider_config.rs

@@ -14,8 +14,7 @@ use litellm_llms::{
         error::Error,
         handler::{self, CallHooks, OcrClient},
         transformation::{
-            BaseOcrConfig, LiteLLMOcrResponse, OcrCredentialInputs, OcrDocument, OcrResponseFormat,
-            PreparedOcrRequest, ResolvedOcrCredentials,
+            BaseOcrConfig, OcrCredentialInputs, PreparedOcrRequest, ResolvedOcrCredentials,
         },
     },
     cohere::ocr::transformation::CohereParseConfig,
@@ -25,6 +24,7 @@ use litellm_llms::{
         deepseek_transformation::VertexAIDeepSeekOCRConfig, transformation::VertexAiOcrConfig,
     },
 };
+use litellm_llms_types::formats::ocr::{LiteLLMOcrResponse, OcrDocument, OcrResponseFormat};
 
 macro_rules! with_config {
     ($kind:expr, $config:ident => $body:expr) => {

litellm-rust/crates/core/src/ocr/route.rs

@@ -6,7 +6,8 @@ use litellm_host::{
     protocol::Protocol,
     protocol::Reply,
 };
-use litellm_llms::base_llm::ocr::{error::Error, transformation::LiteLLMOcrResponse};
+use litellm_llms::base_llm::ocr::error::Error;
+use litellm_llms_types::formats::ocr::LiteLLMOcrResponse;
 
 use crate::ocr::types::{LiteLLMOcrRequest, OcrDocumentInput};
 

litellm-rust/crates/core/src/ocr/types.rs

@@ -5,10 +5,9 @@ use litellm_auth::{InputSource, SecretValue, TokenProviderHandle};
 use litellm_core_utils::call_arguments::CallArguments;
 use litellm_llms::base_llm::ocr::{
     error::Error,
-    transformation::{
-        OcrCredentialInputs, OcrDocument, OcrResponseFormat, OcrTransportConfig, response_format,
-    },
+    transformation::{OcrCredentialInputs, OcrTransportConfig, response_format},
 };
+use litellm_llms_types::formats::ocr::{OcrDocument, OcrResponseFormat};
 use serde_json::{Map, Value};
 
 use super::provider_config::{OcrConfigKind, resolve_provider_config};
@@ -222,7 +221,7 @@ mod tests {
     use super::*;
 
     fn document() -> OcrDocument {
-        OcrDocument::try_from(
+        serde_json::from_value(
             json!({"type":"document_url","document_url":"data:application/pdf;base64,YWJj"}),
         )
         .unwrap()

litellm-rust/crates/core/src/ocr/wire.rs

@@ -1,10 +1,8 @@
 use std::{collections::BTreeMap, time::Duration};
 
 use litellm_auth::{InputSource, SecretValue};
-use litellm_llms::base_llm::ocr::{
-    error::Error,
-    transformation::{OcrDocument, decode_request_value},
-};
+use litellm_llms::base_llm::ocr::{error::Error, transformation::decode_request_value};
+use litellm_llms_types::formats::ocr::OcrDocument;
 use serde::Deserialize;
 use serde_json::{Map, Value};
 

litellm-rust/crates/core/src/responses/handler.rs

@@ -16,7 +16,7 @@ pub(super) async fn execute(
     auth: &litellm_auth::AuthServices,
     request: ProviderResponsesRequest,
     cache: Option<litellm_cache_response::ScopedCache>,
-    cache_options: Option<litellm_cache_response::CacheOptions>,
+    cache_options: Option<litellm_cache_response::CachePolicy>,
     interceptors: &impl Interceptors<Error>,
     observers: Option<&ObservationSender>,
 ) -> Result<ResponsesOutput, Error> {

litellm-rust/crates/core/src/responses/mod.rs

@@ -71,7 +71,7 @@ impl ResponsesRoute {
     async fn run(
         &self,
         call: ResponsesCall,
-        cache_options: Option<litellm_cache_response::CacheOptions>,
+        cache_options: Option<litellm_cache_response::CachePolicy>,
         interceptors: &impl litellm_host::interceptors::Interceptors<Error>,
         observers: Option<&ObservationSender>,
     ) -> Result<ResponsesOutput, Error> {
@@ -85,7 +85,7 @@ impl ResponsesRoute {
     async fn run_provider(
         &self,
         call: ResponsesCall,
-        cache_options: Option<litellm_cache_response::CacheOptions>,
+        cache_options: Option<litellm_cache_response::CachePolicy>,
         interceptors: &impl Interceptors<Error>,
         observers: Option<&ObservationSender>,
     ) -> Result<ResponsesOutput, Error> {

litellm-rust/crates/core/src/responses/route.rs

@@ -5,7 +5,7 @@ use litellm_host::{
     call::{HostedMachine, hosted_call},
     protocol::Protocol,
 };
-use litellm_types::responses::main::ResponsesApiResponse;
+use litellm_llms_types::formats::responses::ResponsesApiResponse;
 
 use super::{
     Error, ResponsesRoute,

litellm-rust/crates/core/src/responses/types.rs

@@ -5,7 +5,7 @@ use litellm_host::call::CallOutput;
 use litellm_llms::base_llm::{
     auth::ValidatedEnvironment, responses::transformation::BaseResponsesApiConfig,
 };
-use litellm_types::responses::main::ResponsesApiResponse;
+use litellm_llms_types::formats::responses::ResponsesApiResponse;
 use serde_json::{Map, Value};
 
 use super::Error;

litellm-rust/crates/core/src/responses/websocket.rs

@@ -2,7 +2,7 @@ use std::{collections::HashMap, sync::Arc, time::Duration};
 
 use futures_util::{SinkExt, StreamExt};
 use litellm_http::websocket::{UpstreamWebSocket, connect_upstream};
-use litellm_types::responses::streaming_websocket::ResponsesWsEventType;
+use litellm_llms_types::formats::responses::streaming_websocket::ResponsesWsEventType;
 use tokio::sync::Mutex;
 use tokio_tungstenite::tungstenite::{
     Message,

litellm-rust/crates/core/tests/caching.rs

@@ -12,8 +12,8 @@ use bytes::Bytes;
 use futures_util::{StreamExt, TryStreamExt, stream};
 use litellm_cache_memory::InMemoryCache;
 use litellm_cache_response::{
-    CacheOptions, CacheScope, ResponseCache, ResponseCacheConfig, ResponseCacheService,
-    ResponseEnvelope,
+    CacheOptions, CachePolicy, CacheScope, ResponseCache, ResponseCacheConfig,
+    ResponseCacheService, ResponseEnvelope,
 };
 use litellm_core::{
     RouteError,
@@ -117,9 +117,9 @@ async fn call(
 
 #[rstest]
 #[case::normal(CacheOptions::new(CacheScope::Shared), true, true)]
-#[case::no_cache(CacheOptions { no_cache: true, ..CacheOptions::new(CacheScope::Shared) }, false, true)]
-#[case::no_store(CacheOptions { no_store: true, ..CacheOptions::new(CacheScope::Shared) }, true, false)]
-#[case::disabled(CacheOptions { caching: Some(false), ..CacheOptions::new(CacheScope::Shared) }, false, false)]
+#[case::no_cache(CacheOptions { policy: CachePolicy { no_cache: true, ..CachePolicy::default() }, ..CacheOptions::new(CacheScope::Shared) }, false, true)]
+#[case::no_store(CacheOptions { policy: CachePolicy { no_store: true, ..CachePolicy::default() }, ..CacheOptions::new(CacheScope::Shared) }, true, false)]
+#[case::disabled(CacheOptions { policy: CachePolicy { caching: Some(false), ..CachePolicy::default() }, ..CacheOptions::new(CacheScope::Shared) }, false, false)]
 #[tokio::test]
 async fn cache_controls_apply_to_both_reads_and_writes(
     cache: Arc<dyn ResponseCacheService>,
@@ -411,7 +411,7 @@ async fn responses_refetches_instead_of_deserializing_another_api_response(
     #[case] poisoned: Value,
 ) {
     use litellm_core::responses::route::Responses;
-    use litellm_types::responses::main::ResponsesApiResponse;
+    use litellm_llms_types::formats::responses::ResponsesApiResponse;
 
     let cache: Arc<dyn ResponseCacheService> = Arc::new(InvalidEntryCache(
         ResponseCache::new(Arc::new(InMemoryCache::default())),
@@ -461,7 +461,7 @@ async fn messages_cache_identity_includes_provider_native_parameters(
     #[case] changed: Value,
 ) {
     use litellm_core::messages::route::Messages;
-    use litellm_types::llms::anthropic_messages::anthropic_response::AnthropicMessagesResponse;
+    use litellm_llms_types::formats::messages::MessagesResponse;
 
     let calls = AtomicUsize::new(0);
     for (value, expected_call) in [(original.clone(), 0), (changed, 1), (original, 0)] {
@@ -487,7 +487,7 @@ async fn messages_cache_identity_includes_provider_native_parameters(
                 None,
                 || async {
                     let call = calls.fetch_add(1, Ordering::SeqCst);
-                    Ok(Box::new(serde_json::from_value::<AnthropicMessagesResponse>(json!({
+                    Ok(Box::new(serde_json::from_value::<MessagesResponse>(json!({
                     "id":call.to_string(), "type":"message", "role":"assistant", "model":"test",
                     "content":[{"type":"text","text":format!("answer {call}")}],
                     "stop_reason":"end_turn", "stop_sequence":null
@@ -723,9 +723,9 @@ async fn unary_call(
 
 #[rstest]
 #[case::normal(CacheOptions::new(CacheScope::Shared), true, true)]
-#[case::no_cache(CacheOptions { no_cache: true, ..CacheOptions::new(CacheScope::Shared) }, false, true)]
-#[case::no_store(CacheOptions { no_store: true, ..CacheOptions::new(CacheScope::Shared) }, true, false)]
-#[case::disabled(CacheOptions { caching: Some(false), ..CacheOptions::new(CacheScope::Shared) }, false, false)]
+#[case::no_cache(CacheOptions { policy: CachePolicy { no_cache: true, ..CachePolicy::default() }, ..CacheOptions::new(CacheScope::Shared) }, false, true)]
+#[case::no_store(CacheOptions { policy: CachePolicy { no_store: true, ..CachePolicy::default() }, ..CacheOptions::new(CacheScope::Shared) }, true, false)]
+#[case::disabled(CacheOptions { policy: CachePolicy { caching: Some(false), ..CachePolicy::default() }, ..CacheOptions::new(CacheScope::Shared) }, false, false)]
 #[tokio::test]
 async fn unary_cache_controls_do_not_change_the_shared_service(
     cache: Arc<dyn ResponseCacheService>,
@@ -843,7 +843,7 @@ async fn responses_cache_only_reuses_completed_responses(
     #[case] expected_calls: usize,
 ) {
     use litellm_core::responses::route::Responses;
-    use litellm_types::responses::main::ResponsesApiResponse;
+    use litellm_llms_types::formats::responses::ResponsesApiResponse;
 
     let calls = AtomicUsize::new(0);
     for _ in 0..2 {

litellm-rust/crates/core/tests/chat_completions.rs

@@ -7,7 +7,7 @@ use std::time::Duration;
 
 use litellm_core::chat_completions::{Error, types::ChatCompletionsRequest};
 use litellm_http::transport::Error as TransportError;
-use litellm_types::utils::ChatCompletionsResponse;
+use litellm_llms_types::formats::chat_completions::ChatCompletionsResponse;
 use rstest::{fixture, rstest};
 use serde_json::{Map, Value, json};
 use wiremock::ResponseTemplate;

litellm-rust/crates/core/tests/messages/host.rs

@@ -1,9 +1,9 @@
 use litellm_host::lifecycle::ExecutionEvent;
 use std::sync::Mutex;
 
-use litellm_core::messages::route::Messages;
+use litellm_core::messages::{MessagesCallResponse, route::Messages};
 use litellm_host::{
-    interceptors::{RequestContext, WireRequest},
+    interceptors::{ExecutionFacts, RequestContext, ResultSource, WireRequest},
     lifecycle::CallEvent,
 };
 use litellm_llms::base_llm::messages::context::MessagesModelCapabilities as AnthropicModelCapabilities;
@@ -20,6 +20,8 @@ struct RecordingHost {
     rewrite: Rewrite,
     events: super::support::Observations,
     optional_params: Mutex<Vec<Value>>,
+    facts: Mutex<Vec<ExecutionFacts>>,
+    reject_result: bool,
 }
 
 impl RecordingHost {
@@ -29,6 +31,8 @@ impl RecordingHost {
             rewrite,
             events: super::support::Observations::default(),
             optional_params: Mutex::new(Vec::new()),
+            facts: Mutex::new(Vec::new()),
+            reject_result: false,
         }
     }
 
@@ -73,6 +77,14 @@ impl litellm_host::lifecycle::CallObserver for RecordingHost {
 impl litellm_host::interceptors::Interceptors<<Messages as litellm_host::protocol::Protocol>::Error>
     for RecordingHost
 {
+    async fn result_ready(&self, facts: ExecutionFacts) -> Result<(), Error> {
+        self.facts.lock().unwrap().push(facts);
+        if self.reject_result {
+            return Err(Error::Unsupported("result rejected"));
+        }
+        Ok(())
+    }
+
     async fn before_provider_request(
         &self,
         wire: WireRequest,
@@ -98,6 +110,91 @@ impl litellm_host::interceptors::Interceptors<<Messages as litellm_host::protoco
     }
 }
 
+#[rstest]
+#[case::native_unary(false, false)]
+#[case::native_stream(true, false)]
+#[case::hosted_unary(false, true)]
+#[case::hosted_stream(true, true)]
+#[tokio::test]
+async fn rejected_results_are_not_delivered_or_cached(
+    call: MessagesCall,
+    #[case] streaming: bool,
+    #[case] hosted: bool,
+) {
+    use futures_util::TryStreamExt;
+    use litellm_cache_memory::InMemoryCache;
+    use litellm_cache_response::{CacheScope, ResponseCache, ScopedCache};
+
+    let response = if streaming {
+        ResponseTemplate::new(200).set_body_raw(
+            "event: message_stop\ndata: {\"type\":\"message_stop\"}\n\n",
+            "text/event-stream",
+        )
+    } else {
+        message_response()
+    };
+    let upstream = upstream([response.clone(), response]).await;
+    let route = messages_route(no_secrets()).with_cache(ScopedCache::new(
+        Arc::new(ResponseCache::new(Arc::new(InMemoryCache::new(
+            Some(100),
+            Some(Duration::from_secs(60)),
+        )))),
+        CacheScope::Shared,
+    ));
+    for (reject, expected_requests, cached) in [
+        (true, 1, false),
+        (false, 2, false),
+        (true, 2, true),
+        (false, 2, true),
+    ] {
+        let request = authenticated(
+            with_fields(
+                MessagesCall {
+                    body: call.body.clone(),
+                    ..super::call()
+                },
+                json!({"stream": streaming}),
+            ),
+            upstream.uri(),
+        );
+        let host = RecordingHost {
+            reject_result: reject,
+            ..RecordingHost::passthrough(request)
+        };
+        let result = if hosted {
+            litellm_host_native::in_process::run_hosted(
+                route.clone().machine(host.request().unwrap(), None),
+                host.runtime(),
+            )
+            .await
+            .map(|_| ())
+        } else {
+            match route.execute(host.request().unwrap(), &host, None).await {
+                Ok(MessagesCallResponse::Complete(_)) => Ok(()),
+                Ok(MessagesCallResponse::Stream { chunks, .. }) => {
+                    chunks.try_collect::<Vec<_>>().await.map(|_| ())
+                }
+                Err(error) => Err(error),
+            }
+        };
+        assert_eq!(
+            result,
+            if reject {
+                Err(Error::Unsupported("result rejected"))
+            } else {
+                Ok(())
+            }
+        );
+        assert_eq!(received(&upstream).await.len(), expected_requests);
+        let facts = host.facts.lock().unwrap();
+        assert_eq!(facts.len(), 1);
+        assert_eq!(
+            matches!(facts[0].source, ResultSource::Cache { .. }),
+            cached
+        );
+    }
+}
+
 async fn run_through(host: &RecordingHost) -> Result<MessagesOutput, Error> {
     litellm_host_native::in_process::run_hosted(
         machine(Arc::new(RecordingSecrets::empty()))(host.request()?),
@@ -143,6 +240,81 @@ async fn what_before_send_returns_is_what_the_provider_receives(call: MessagesCa
     assert_eq!(request.header("x-api-key"), Some("sk-ant"));
 }
 
+#[rstest]
+#[case::enable(false, json!(true), Some(true))]
+#[case::disable(true, json!(false), Some(false))]
+#[case::null(true, Value::Null, Some(false))]
+#[case::invalid(false, json!("true"), None)]
+#[tokio::test]
+async fn response_mode_follows_the_intercepted_request(
+    call: MessagesCall,
+    traces: TraceCapture,
+    #[case] original_stream: bool,
+    #[case] rewritten_stream: Value,
+    #[case] expected_stream: Option<bool>,
+) {
+    use futures_util::TryStreamExt;
+
+    let sse = "event: message_stop\ndata: {\"type\":\"message_stop\"}\n\n";
+    let response = if expected_stream == Some(true) {
+        ResponseTemplate::new(200).set_body_raw(sse, "text/event-stream")
+    } else {
+        message_response()
+    };
+    let upstream = upstream([response]).await;
+    let rewrite = rewritten_stream.clone();
+    let host = RecordingHost::new(
+        authenticated(
+            with_fields(call, json!({"stream": original_stream})),
+            upstream.uri(),
+        ),
+        Box::new(move |wire| {
+            let mut body = wire.body;
+            body["stream"] = rewrite.clone();
+            Ok(WireRequest { body, ..wire })
+        }),
+    );
+    let result = traces
+        .logger()
+        .instrument(async {
+            let output = messages_route(no_secrets())
+                .execute(host.request()?, &host, None)
+                .await?;
+            match output {
+                MessagesCallResponse::Stream { chunks, .. } => {
+                    assert_eq!(expected_stream, Some(true));
+                    assert_eq!(
+                        chunks.try_collect::<Vec<_>>().await?.concat(),
+                        sse.as_bytes()
+                    );
+                }
+                MessagesCallResponse::Complete(message) => {
+                    assert_eq!(expected_stream, Some(false));
+                    assert_eq!(*message, serde_json::from_value(message_body()).unwrap());
+                }
+            }
+            Ok::<_, Error>(())
+        })
+        .await;
+
+    let summaries = traces.summaries("litellm.route");
+    assert_eq!(summaries.len(), 1);
+    let Some(expected_stream) = expected_stream else {
+        assert!(matches!(result, Err(Error::InvalidRequest(_))));
+        assert!(received(&upstream).await.is_empty());
+        assert_eq!(summaries[0]["outcome"], "failure");
+        return;
+    };
+    result.unwrap();
+    assert_eq!(
+        only_request(&upstream).await.json()["stream"],
+        rewritten_stream
+    );
+    assert_eq!(host.raw_responses().len(), usize::from(!expected_stream));
+    assert_eq!(summaries[0]["stream"], expected_stream);
+    assert_eq!(summaries[0]["outcome"], "success");
+}
+
 #[rstest]
 #[tokio::test]
 async fn a_before_send_failure_never_sends(call: MessagesCall) {

litellm-rust/crates/core/tests/messages/main.rs

@@ -8,10 +8,8 @@ use litellm_core::messages::{
     route::{Messages, MessagesMachine, MessagesOutput},
 };
 use litellm_http::{HttpSettings, Resolution};
+use litellm_llms_types::formats::messages::{MessagesRequest, MessagesResponse};
 use litellm_secrets::source::SecretSource;
-use litellm_types::llms::anthropic_messages::{
-    anthropic_request::AnthropicMessagesRequest, anthropic_response::AnthropicMessagesResponse,
-};
 use rstest::fixture;
 use serde_json::{Map, Value, json};
 use wiremock::ResponseTemplate;
@@ -35,7 +33,7 @@ fn object(value: Value) -> Map<String, Value> {
     map
 }
 
-fn body(value: Value) -> AnthropicMessagesRequest {
+fn body(value: Value) -> MessagesRequest {
     serde_json::from_value(value).unwrap()
 }
 
@@ -116,7 +114,7 @@ async fn run(call: MessagesCall) -> Result<MessagesOutput, Error> {
     run_with(Arc::new(RecordingSecrets::empty()), call).await
 }
 
-async fn run_message(call: MessagesCall) -> AnthropicMessagesResponse {
+async fn run_message(call: MessagesCall) -> MessagesResponse {
     match run(call).await.expect("messages call succeeds") {
         MessagesOutput::Complete(message) => *message,
         MessagesOutput::StreamEnded | MessagesOutput::Detached => {