Merge branch 'litellm_internal_staging' (last green: 6c30b4331) into litellm_add_spendlogs_api_key_startTime_index

.circleci/config.yml

@@ -430,7 +430,7 @@ jobs:
               --verbose \
               --command="tr ' ' '\\n' | awk '/\\.py/ {print; next} {sub(/\\.[A-Z][^.]*$/, \"\"); gsub(/\\./, \"/\"); print \$0 \".py\"}' | xargs uv run --no-sync python -m pytest \
                 -vv \
-                --cov=./litellm \
+                --cov=./litellm --cov=./enterprise/litellm_enterprise \
                 --cov-report=xml \
                 --junitxml=test-results/junit.xml \
                 --durations=20 \
@@ -504,7 +504,7 @@ jobs:
               --verbose \
               --command="tr ' ' '\\n' | awk '/\\.py/ {print; next} {sub(/\\.[A-Z][^.]*$/, \"\"); gsub(/\\./, \"/\"); print \$0 \".py\"}' | xargs uv run --no-sync python -m pytest \
                 -vv \
-                --cov=./litellm \
+                --cov=./litellm --cov=./enterprise/litellm_enterprise \
                 --cov-report=xml \
                 --junitxml=test-results/junit.xml \
                 --durations=20 \
@@ -631,7 +631,7 @@ jobs:
               --verbose \
               --command="tr ' ' '\\n' | awk '/\\.py/ {print; next} {sub(/\\.[A-Z][^.]*$/, \"\"); gsub(/\\./, \"/\"); print \$0 \".py\"}' | xargs uv run --no-sync python -m pytest \
                 -v -x \
-                --cov=./litellm --cov-report=xml \
+                --cov=./litellm --cov=./enterprise/litellm_enterprise --cov-report=xml \
                 --junitxml=test-results/junit.xml \
                 --durations=5 \
                 -n 2"
@@ -651,126 +651,6 @@ jobs:
             - auth_ui_unit_tests_coverage.xml
             - auth_ui_unit_tests_coverage
 
-  proxy_behavior_tests:
-    docker:
-      - *python312_image
-      - image: cimg/postgres:16.0@sha256:b125148bc76e8e8eee5eb3ad6020a3a14110a14e8192f1c645128afebe2e2f84
-        environment:
-          POSTGRES_USER: postgres
-          POSTGRES_PASSWORD: postgres
-          POSTGRES_DB: litellm_test
-    working_directory: ~/project
-    environment:
-      DATABASE_URL: "postgresql://postgres:postgres@localhost:5432/litellm_test"
-    steps:
-      - checkout
-      - skip_if_unrelated_changes
-      - setup_google_dns
-      - install_uv
-      - install_rust
-      - run:
-          name: Install Dependencies
-          command: |
-            uv sync --frozen --all-groups --all-extras --python 3.12
-      - wait_for_service:
-          url: tcp://localhost:5432
-          timeout: "60"
-      - run:
-          name: Seed DB schema via prisma db push
-          command: |
-            uv run --no-sync prisma db push --schema litellm/proxy/schema.prisma --accept-data-loss
-      - run:
-          name: Generate Prisma Client
-          command: uv run --no-sync python -m prisma generate
-      - run:
-          name: Run proxy management behavior tests
-          command: |
-            mkdir -p test-results
-            uv run --no-sync python -m pytest tests/proxy_behavior \
-              -v --junitxml=test-results/junit.xml --durations=10
-          no_output_timeout: 15m
-      - store_test_results:
-          path: test-results
-
-  proxy_security_tests:
-    docker:
-      - *python312_image
-      - image: cimg/postgres:16.0@sha256:b125148bc76e8e8eee5eb3ad6020a3a14110a14e8192f1c645128afebe2e2f84
-        environment:
-          POSTGRES_USER: postgres
-          POSTGRES_PASSWORD: postgres
-          POSTGRES_DB: litellm_test
-    working_directory: ~/project
-    environment:
-      DATABASE_URL: "postgresql://postgres:postgres@localhost:5432/litellm_test"
-    steps:
-      - checkout
-      - skip_if_unrelated_changes
-      - setup_google_dns
-      - install_uv
-      - install_rust
-      - run:
-          name: Install Dependencies
-          command: |
-            uv sync --frozen --all-groups --all-extras --python 3.12
-      - wait_for_service:
-          url: tcp://localhost:5432
-          timeout: "60"
-      - run:
-          name: Seed DB schema via prisma db push
-          command: |
-            uv run --no-sync prisma db push --schema litellm/proxy/schema.prisma --accept-data-loss
-      - run:
-          name: Generate Prisma Client
-          command: uv run --no-sync python -m prisma generate
-      - run:
-          name: Run proxy security tests
-          command: |
-            mkdir -p test-results
-            uv run --no-sync python -m pytest tests/proxy_security_tests \
-              -v --junitxml=test-results/junit.xml --durations=10
-          no_output_timeout: 15m
-      - store_test_results:
-          path: test-results
-
-  schema_migration_check:
-    docker:
-      - *python312_image
-      - image: cimg/postgres:16.0@sha256:b125148bc76e8e8eee5eb3ad6020a3a14110a14e8192f1c645128afebe2e2f84
-        environment:
-          POSTGRES_USER: postgres
-          POSTGRES_PASSWORD: postgres
-          POSTGRES_DB: litellm_test
-    working_directory: ~/project
-    environment:
-      # An empty database; the test applies every committed migration itself.
-      DATABASE_URL: "postgresql://postgres:postgres@localhost:5432/litellm_test"
-    steps:
-      - checkout
-      - skip_if_unrelated_changes
-      - setup_google_dns
-      - install_uv
-      - install_rust
-      - run:
-          name: Install Dependencies
-          command: |
-            uv sync --frozen --all-groups --all-extras --python 3.12
-      - wait_for_service:
-          url: tcp://localhost:5432
-          timeout: "60"
-      - run:
-          name: Generate Prisma Client
-          command: uv run --no-sync python -m prisma generate
-      - run:
-          name: Check schema.prisma is in sync with committed migrations
-          command: |
-            mkdir -p test-results
-            uv run --no-sync python -m pytest tests/proxy_migration_tests \
-              -v --junitxml=test-results/junit.xml --durations=10
-          no_output_timeout: 15m
-      - store_test_results:
-          path: test-results
-
   litellm_router_testing: # Runs all tests with the "router" keyword
     docker:
       - *python312_image
@@ -858,7 +738,7 @@ jobs:
               --verbose \
               --command="tr ' ' '\\n' | awk '/\\.py/ {print; next} {sub(/\\.[A-Z][^.]*$/, \"\"); gsub(/\\./, \"/\"); print \$0 \".py\"}' | xargs uv run --no-sync python -m pytest \
                 -v -x \
-                --cov=./litellm --cov-report=xml \
+                --cov=./litellm --cov=./enterprise/litellm_enterprise --cov-report=xml \
                 --junitxml=test-results/junit.xml \
                 --durations=5 \
                 -n 4"
@@ -985,7 +865,7 @@ jobs:
               --verbose \
               --command="tr ' ' '\\n' | awk '/\\.py/ {print; next} {sub(/\\.[A-Z][^.]*$/, \"\"); gsub(/\\./, \"/\"); print \$0 \".py\"}' | xargs uv run --no-sync python -m pytest \
                 -vv \
-                --cov=./litellm --cov-report=xml \
+                --cov=./litellm --cov=./enterprise/litellm_enterprise --cov-report=xml \
                 --junitxml=test-results/junit.xml \
                 --durations=20 \
                 -n 4 \
@@ -1030,7 +910,7 @@ jobs:
               --verbose \
               --command="tr ' ' '\\n' | awk '/\\.py/ {print; next} {sub(/\\.[A-Z][^.]*$/, \"\"); gsub(/\\./, \"/\"); print \$0 \".py\"}' | xargs uv run --no-sync python -m pytest \
                 -vv -x -s \
-                --cov=./litellm --cov-report=xml \
+                --cov=./litellm --cov=./enterprise/litellm_enterprise --cov-report=xml \
                 --junitxml=test-results/junit.xml \
                 --durations=5"
           no_output_timeout: 15m
@@ -1074,7 +954,7 @@ jobs:
               --verbose \
               --command="tr ' ' '\\n' | awk '/\\.py/ {print; next} {sub(/\\.[A-Z][^.]*$/, \"\"); gsub(/\\./, \"/\"); print \$0 \".py\"}' | xargs uv run --no-sync python -m pytest \
                 -vv \
-                --cov=./litellm --cov-report=xml \
+                --cov=./litellm --cov=./enterprise/litellm_enterprise --cov-report=xml \
                 --junitxml=test-results/junit.xml \
                 --durations=5 \
                 -n 2 \
@@ -1120,7 +1000,7 @@ jobs:
               --verbose \
               --command="tr ' ' '\\n' | awk '/\\.py/ {print; next} {sub(/\\.[A-Z][^.]*$/, \"\"); gsub(/\\./, \"/\"); print \$0 \".py\"}' | xargs uv run --no-sync python -m pytest \
                 -vv -x -s \
-                --cov=./litellm --cov-report=xml \
+                --cov=./litellm --cov=./enterprise/litellm_enterprise --cov-report=xml \
                 --junitxml=test-results/junit.xml \
                 --durations=5 \
                 --retries 3 --retry-delay 5"
@@ -1211,7 +1091,7 @@ jobs:
               --verbose \
               --command="tr ' ' '\\n' | awk '/\\.py/ {print; next} {sub(/\\.[A-Z][^.]*$/, \"\"); gsub(/\\./, \"/\"); print \$0 \".py\"}' | xargs uv run --no-sync python -m pytest \
                 -vv -x \
-                --cov=./litellm --cov-report=xml \
+                --cov=./litellm --cov=./enterprise/litellm_enterprise --cov-report=xml \
                 --junitxml=test-results/junit.xml \
                 --durations=5 \
                 -n 4"
@@ -1255,7 +1135,7 @@ jobs:
               --verbose \
               --command="tr ' ' '\\n' | awk '/\\.py/ {print; next} {sub(/\\.[A-Z][^.]*$/, \"\"); gsub(/\\./, \"/\"); print \$0 \".py\"}' | xargs uv run --no-sync python -m pytest \
                 -vv -x \
-                --cov=./litellm --cov-report=xml \
+                --cov=./litellm --cov=./enterprise/litellm_enterprise --cov-report=xml \
                 --junitxml=test-results/junit.xml \
                 --durations=5 \
                 -n 4"
@@ -1333,7 +1213,7 @@ jobs:
               --verbose \
               --command="tr ' ' '\\n' | awk '/\\.py/ {print; next} {sub(/\\.[A-Z][^.]*$/, \"\"); gsub(/\\./, \"/\"); print \$0 \".py\"}' | xargs uv run --no-sync python -m pytest \
                 -vv -x -s \
-                --cov=./litellm --cov-report=xml \
+                --cov=./litellm --cov=./enterprise/litellm_enterprise --cov-report=xml \
                 --junitxml=test-results/junit.xml \
                 --durations=5 \
                 -n 2"
@@ -1377,7 +1257,7 @@ jobs:
               --verbose \
               --command="tr ' ' '\\n' | awk '/\\.py/ {print; next} {sub(/\\.[A-Z][^.]*$/, \"\"); gsub(/\\./, \"/\"); print \$0 \".py\"}' | xargs uv run --no-sync python -m pytest \
                 -vv -x -s \
-                --cov=./litellm --cov-report=xml \
+                --cov=./litellm --cov=./enterprise/litellm_enterprise --cov-report=xml \
                 --junitxml=test-results/junit.xml \
                 --durations=5 \
                 -n 2"
@@ -1422,7 +1302,7 @@ jobs:
               --verbose \
               --command="tr ' ' '\\n' | awk '/\\.py/ {print; next} {sub(/\\.[A-Z][^.]*$/, \"\"); gsub(/\\./, \"/\"); print \$0 \".py\"}' | xargs uv run --no-sync python -m pytest \
                 -vv -x \
-                --cov=./litellm --cov-report=xml \
+                --cov=./litellm --cov=./enterprise/litellm_enterprise --cov-report=xml \
                 --junitxml=test-results/junit.xml \
                 --durations=5 \
                 -n 4"
@@ -1501,7 +1381,7 @@ jobs:
               --verbose \
               --command="tr ' ' '\\n' | awk '/\\.py/ {print; next} {sub(/\\.[A-Z][^.]*$/, \"\"); gsub(/\\./, \"/\"); print \$0 \".py\"}' | xargs uv run --no-sync python -m pytest \
                 -vv \
-                --cov=./litellm --cov-report=xml \
+                --cov=./litellm --cov=./enterprise/litellm_enterprise --cov-report=xml \
                 -n 4 \
                 --junitxml=test-results/junit.xml \
                 --durations=5 \
@@ -1546,7 +1426,7 @@ jobs:
               --verbose \
               --command="tr ' ' '\\n' | awk '/\\.py/ {print; next} {sub(/\\.[A-Z][^.]*$/, \"\"); gsub(/\\./, \"/\"); print \$0 \".py\"}' | xargs uv run --no-sync python -m pytest \
                 -vv -x -s \
-                --cov=./litellm --cov-report=xml \
+                --cov=./litellm --cov=./enterprise/litellm_enterprise --cov-report=xml \
                 --junitxml=test-results/junit.xml \
                 --durations=5"
           no_output_timeout: 15m
@@ -1599,7 +1479,7 @@ jobs:
               --verbose \
               --command="tr ' ' '\\n' | awk '/\\.py/ {print; next} {sub(/\\.[A-Z][^.]*$/, \"\"); gsub(/\\./, \"/\"); print \$0 \".py\"}' | xargs uv run --no-sync python -m pytest \
                 -vv -x -s \
-                --cov=./litellm --cov-report=xml \
+                --cov=./litellm --cov=./enterprise/litellm_enterprise --cov-report=xml \
                 --junitxml=test-results/junit.xml \
                 --durations=5 -n 2 \
                 --reruns 2 --reruns-delay 1"
@@ -3105,12 +2985,6 @@ workflows:
           filters: *main_branches
       - auth_ui_unit_tests:
           filters: *main_branches
-      - proxy_behavior_tests:
-          filters: *main_branches
-      - proxy_security_tests:
-          filters: *main_branches
-      - schema_migration_check:
-          filters: *main_branches
       - build_docker_database_image:
           filters: *main_branches
       - e2e_ui_testing:

.github/workflows/_test-unit-base.yml

@@ -149,7 +149,7 @@ jobs:
               --reruns "${RERUNS}" \
               --reruns-delay 1 \
               --durations=20 \
-              --cov=./litellm \
+              --cov=./litellm --cov=./enterprise/litellm_enterprise \
               --cov-report=xml:coverage.xml \
               --cov-config=pyproject.toml
           else
@@ -161,7 +161,7 @@ jobs:
               --reruns-delay 1 \
               --dist="${DIST}" \
               --durations=20 \
-              --cov=./litellm \
+              --cov=./litellm --cov=./enterprise/litellm_enterprise \
               --cov-report=xml:coverage.xml \
               --cov-config=pyproject.toml
           fi

.github/workflows/test-linting.yml

@@ -67,6 +67,17 @@ jobs:
         with:
           version: "0.10.9"
 
+      - name: Cache uv dependencies
+        if: steps.changes.outputs.decision != 'skip'
+        uses: actions/cache@0057852bfaa89a56745cba8c7296529d2fc39830 # v4.3.0
+        with:
+          path: |
+            ~/.cache/uv
+            .venv
+          key: ${{ runner.os }}-uv-lint-${{ hashFiles('uv.lock') }}
+          restore-keys: |
+            ${{ runner.os }}-uv-lint-
+
       - name: Clean Python cache
         if: steps.changes.outputs.decision != 'skip'
         run: |

.github/workflows/test-mcp.yml

@@ -60,4 +60,4 @@ jobs:
       - name: Run MCP tests
         if: steps.changes.outputs.decision != 'skip'
         run: |
-          uv run --no-sync pytest tests/mcp_tests -x -vv -n 4 --cov=./litellm --cov-report=xml --durations=5
+          uv run --no-sync pytest tests/mcp_tests -x -vv -n 4 --cov=./litellm --cov=./enterprise/litellm_enterprise --cov-report=xml --durations=5

.github/workflows/test-postgres.yml

@@ -0,0 +1,145 @@
+name: "Postgres Tests"
+
+on:
+  pull_request:
+    branches:
+      - main
+      - litellm_internal_staging
+      - litellm_oss_staging
+      - "litellm_**"
+  push:
+    branches:
+      - main
+      - litellm_internal_staging
+  workflow_dispatch:
+
+permissions:
+  contents: read
+
+concurrency:
+  group: ${{ github.workflow }}-${{ github.event.pull_request.number || github.sha }}
+  cancel-in-progress: ${{ github.event_name == 'pull_request' }}
+
+jobs:
+  postgres:
+    name: ${{ matrix.shard }}
+    runs-on: ubuntu-latest
+    timeout-minutes: ${{ matrix.job-timeout-minutes }}
+    permissions:
+      contents: read
+
+    services:
+      postgres:
+        image: postgres:16@sha256:e17e86066e5ef83e0952a9347f5c792b7ece00972e2aa787a6986f471b3dd3d5
+        env:
+          POSTGRES_USER: postgres
+          POSTGRES_PASSWORD: postgres
+          POSTGRES_DB: litellm_test
+        ports:
+          - 5432:5432
+        options: >-
+          --health-cmd pg_isready
+          --health-interval 10s
+          --health-timeout 5s
+          --health-retries 10
+
+    strategy:
+      fail-fast: false
+      matrix:
+        include:
+          - shard: proxy-behavior
+            test-path: "tests/proxy_behavior"
+            seed: db-push
+            workers: 0
+            timeout-minutes: 25
+            job-timeout-minutes: 50
+
+          - shard: proxy-security
+            test-path: "tests/proxy_security_tests"
+            seed: db-push
+            workers: 0
+            timeout-minutes: 15
+            job-timeout-minutes: 40
+
+          - shard: schema-migration
+            test-path: "tests/proxy_migration_tests"
+            seed: none
+            workers: 0
+            timeout-minutes: 20
+            job-timeout-minutes: 45
+
+    env:
+      DATABASE_URL: "postgresql://postgres:postgres@localhost:5432/litellm_test"
+
+    steps:
+      - uses: actions/checkout@08eba0b27e820071cde6df949e0beb9ba4906955 # v4.3.0
+        timeout-minutes: 3
+        with:
+          persist-credentials: false
+
+      - name: Detect relevant changes
+        id: changes
+        timeout-minutes: 2
+        uses: ./.github/actions/detect-changes
+
+      - name: Set up Python
+        if: steps.changes.outputs.decision != 'skip'
+        timeout-minutes: 3
+        uses: actions/setup-python@a26af69be951a213d495a4c3e4e4022e16d87065 # v5.6.0
+        with:
+          python-version: "3.12"
+
+      - name: Set up uv
+        if: steps.changes.outputs.decision != 'skip'
+        timeout-minutes: 3
+        uses: ./.github/actions/setup-uv-with-retries
+        with:
+          version: "0.10.9"
+
+      - name: Cache uv dependencies
+        if: steps.changes.outputs.decision != 'skip'
+        timeout-minutes: 5
+        uses: actions/cache@0057852bfaa89a56745cba8c7296529d2fc39830 # v4.3.0
+        with:
+          path: |
+            ~/.cache/uv
+            .venv
+          key: ${{ runner.os }}-uv-postgres-${{ hashFiles('uv.lock') }}
+          restore-keys: |
+            ${{ runner.os }}-uv-postgres-
+
+      - name: Install dependencies
+        if: steps.changes.outputs.decision != 'skip'
+        timeout-minutes: 12
+        run: |
+          .github/scripts/uv_sync_with_retries.sh --frozen --all-groups --all-extras
+
+      - name: Cache Prisma binaries
+        if: steps.changes.outputs.decision != 'skip'
+        timeout-minutes: 3
+        uses: ./.github/actions/cache-prisma-binaries
+
+      - name: Generate Prisma client
+        if: steps.changes.outputs.decision != 'skip'
+        timeout-minutes: 5
+        run: |
+          uv run --no-sync prisma generate --schema litellm/proxy/schema.prisma
+
+      - name: Seed database schema
+        if: steps.changes.outputs.decision != 'skip' && matrix.seed != 'none'
+        timeout-minutes: 10
+        run: |
+          uv run --no-sync prisma db push --schema litellm/proxy/schema.prisma --accept-data-loss
+
+      - name: Run tests
+        if: steps.changes.outputs.decision != 'skip'
+        timeout-minutes: ${{ matrix.timeout-minutes }}
+        env:
+          TEST_PATH: ${{ matrix.test-path }}
+          WORKERS: ${{ matrix.workers }}
+        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

litellm/integrations/callback_configs.json

@@ -294,12 +294,18 @@
     "id": "newrelic",
     "displayName": "New Relic",
     "logo": "newrelic.png",
-    "supports_key_team_logging": false,
+    "supports_key_team_logging": true,
     "dynamic_params": {
-      "NEW_RELIC_AI_MONITORING_RECORD_CONTENT_ENABLED": {
+      "newrelic_api_key": {
+        "type": "password",
+        "ui_name": "New Relic Ingest License Key",
+        "description": "Per-team ingest (license) key. Team traces export to this key's New Relic account over OTLP.",
+        "required": false
+      },
+      "newrelic_region": {
         "type": "text",
-        "ui_name": "Record AI Content (default: true)",
-        "description": "Whether to record AI message content. Set to false to disable.",
+        "ui_name": "New Relic Region (us or eu)",
+        "description": "Data center region for this team's account. Defaults to us.",
         "required": false
       }
     },

litellm/integrations/otel/emitter.py

@@ -156,6 +156,12 @@ class SpanEmitter:
             links=list(links) if links else None,
         )
 
+    def mark_emitted(self, dedup_key: str | None, role: SpanRole) -> None:
+        """Register a span emitted outside :meth:`emit` (the boundary-opened
+        LLM-call span closed via :meth:`finish_span`) so a later :meth:`emit`
+        for the same ``(dedup_key, role)`` deduplicates against it."""
+        self._seen(dedup_key, role)
+
     def _seen(self, dedup_key: str | None, role: SpanRole) -> bool:
         """Return True once a ``(dedup_key, role)`` pair has been emitted.
 

litellm/integrations/otel/logger.py

@@ -484,10 +484,15 @@ class OpenTelemetryV2(CustomLogger):
         # ``pop`` is the dedup: this method runs from both the success and failure
         # paths, and whichever fires first removes the carrier and closes the span.
         carrier: Final = self._open_llm_calls.pop(call_id, None) if call_id else None
-        if carrier is None:
+        # A missing carrier does not always mean nothing happened: a team/key-scoped
+        # logger is a success/failure callback only, so ``pre_call`` never reaches it
+        # and no carrier exists. The payload plus the request-level provider-handoff
+        # stamp (``upstream_started``) is the affirmative signal of a real call; a
+        # gate rejection carries ``is_no_upstream_call`` and gets no span.
+        if carrier is None and (call.is_no_upstream_call or not call.upstream_started or call.payload is None):
             return None
         try:
-            return self._finish_carrier(carrier, call, end_time)
+            return self._finish_carrier(carrier, call, start_time, end_time)
         finally:
             # After the span has ended, so a release-triggered provider shutdown
             # force-flushes it out rather than racing its enqueue.
@@ -497,8 +502,11 @@ class OpenTelemetryV2(CustomLogger):
         """Remember an in-flight LLM call, evicting the oldest if over budget.
 
         A call that opens but never closes (a stream that only fires stream
-        events) would linger otherwise; the evicted span is simply dropped
-        (never exported).
+        events) would linger otherwise. Eviction only drops the boundary carrier,
+        not the call: if that call later closes as a real completed call, it still
+        emits through the deferred branch in ``_close_llm_call`` (the same path a
+        team/key-scoped logger uses, since it never opens a carrier), deduplicated
+        by call id. Only a call that is evicted and never closes goes unexported.
         """
         self._open_llm_calls[call_id] = carrier
         if len(self._open_llm_calls) > _OPEN_CALLS_MAX:
@@ -512,15 +520,20 @@ class OpenTelemetryV2(CustomLogger):
 
     def _finish_carrier(
         self,
-        carrier: _LLMCallSpan,
+        carrier: "_LLMCallSpan | None",
         call: LLMCallEvent,
+        start_time: datetime | float | None,
         end_time: datetime | float | None,
     ) -> Span | None:
         payload: Final = call.payload
+        call_id: Final = call.call_id
         if payload is None:
-            if carrier.span is not None:
+            if carrier is not None and carrier.span is not None:
                 # Opened at the boundary but the payload never materialized — end
-                # it (named provisionally) so it isn't leaked as an open span.
+                # it (named provisionally) so it isn't leaked as an open span, and
+                # register the dedup marker so a later payload-carrying close for
+                # the same call id cannot re-emit through the deferred branch.
+                self._emitter.mark_emitted(call_id, SpanRole.LLM_CALL)
                 carrier.span.end(end_time=to_ns(end_time))
             return None
         data: Final = LLMCallSpanData.from_standard_logging_payload(
@@ -529,10 +542,13 @@ class OpenTelemetryV2(CustomLogger):
             time_to_first_chunk_seconds=call.time_to_first_chunk_seconds,
         )
         end_time_ns: Final = to_ns(end_time)
-        if carrier.span is not None:
+        if carrier is not None and carrier.span is not None:
             # Born at the boundary: stamp attributes from the typed payload, set
             # status, and end it. Its parent (the server span) was captured at
-            # creation from real ambient context.
+            # creation from real ambient context. Register the dedup marker so a
+            # second close for the same call id (success then failure on one
+            # logging object) cannot re-emit through the deferred branch.
+            self._emitter.mark_emitted(call_id, SpanRole.LLM_CALL)
             self._emitter.finish_span(SpanRole.LLM_CALL, carrier.span, data, end_time_ns=end_time_ns)
             return carrier.span
         # Deferred: ``pre_call`` saw no recordable parent, so create the span now.
@@ -549,7 +565,7 @@ class OpenTelemetryV2(CustomLogger):
                 SpanRole.LLM_CALL,
                 data,
                 parent_context=(set_span_in_context(INVALID_SPAN, parent_ctx) if route.detached else parent_ctx),
-                start_time_ns=carrier.start_time_ns,
+                start_time_ns=(carrier.start_time_ns if carrier is not None else to_ns(start_time)),
                 end_time_ns=end_time_ns,
                 tracer=route.tracer,
                 links=_request_trace_links(parent_ctx) if route.detached else None,

litellm/integrations/otel/model/config.py

@@ -39,6 +39,7 @@ class ExporterOwner(str, Enum):
     WEAVE_OTEL = "weave_otel"
     LEVO = "levo"
     AGENTOPS = "agentops"
+    NEWRELIC = "newrelic"
 
 
 class _OTelV2Flag(BaseSettings):
@@ -97,6 +98,15 @@ class ExporterSpec(BaseModel):
             "auto (Simple for console/in_memory, Batch otherwise)."
         ),
     )
+    requires_headers: bool = Field(
+        default=False,
+        description=(
+            "Skip this exporter when no headers are resolved. For destinations "
+            "that reject unauthenticated exports (e.g. New Relic), a spec kept "
+            "only as the per-request credential-stamping target would otherwise "
+            "export keyless traffic and produce a 4xx for every span batch."
+        ),
+    )
 
 
 class OpenTelemetryV2Config(BaseSettings):

litellm/integrations/otel/model/metadata.py

@@ -203,6 +203,11 @@ class LLMCallEvent:
     # True for synthetic proxy-gate logs (auth / rate-limit rejections): they fire
     # the ``pre_call`` hook but never made an upstream call, so they get no span.
     is_no_upstream_call: bool
+    # True once the request handed off to a provider (``pre_call`` stamped
+    # ``api_call_start_time``). The affirmative signal that an LLM call was
+    # actually attempted — router pre-call rejections, SDK failures before the
+    # provider handoff, and standalone guardrail runs all lack it.
+    upstream_started: bool
     # A best-effort ``"{operation} {model}"`` name known at ``pre_call`` time. The
     # span is renamed from the typed payload at close (``finish_span``); this only
     # needs to be reasonable for a span that never gets closed (a leak).
@@ -221,6 +226,7 @@ class LLMCallEvent:
             dynamic_params=kwargs.get("standard_callback_dynamic_params"),
             auth_metadata=auth_metadata(payload, kwargs),
             is_no_upstream_call=bool(kwargs.get(LITELLM_LOGGING_NO_UPSTREAM_LLM_CALL)),
+            upstream_started=kwargs.get("api_call_start_time") is not None,
             provisional_span_name=f"{operation.value} {model}".strip(),
             time_to_first_chunk_seconds=time_to_first_chunk_seconds(kwargs),
         )

litellm/integrations/otel/plumbing/providers.py

@@ -436,6 +436,8 @@ def build_tracer_provider(
     # ``config._normalize`` guarantees at least one spec (it folds the top-level
     # ``exporter``/``endpoint``/``headers`` fields in when ``exporters`` is empty).
     for spec in config.exporters:
+        if spec.requires_headers and not spec.headers:
+            continue
         exp = _exporter_from_spec(spec)
         provider.add_span_processor(
             _processor_for(

litellm/integrations/otel/plumbing/routing.py

@@ -28,6 +28,7 @@ from litellm.integrations.otel.plumbing.providers import (
     get_tracer,
 )
 from litellm.integrations.otel.presets import (
+    dynamic_otlp_endpoint,
     dynamic_otlp_headers,
     project_routing_headers,
 )
@@ -129,7 +130,9 @@ class TenantTracerCache:
         # thread-pool workers concurrently with the event loop, so cache
         # updates, span counts, and retirement must be atomic.
         self._lock: Final = threading.Lock()
-        self._providers: OrderedDict[tuple[_HeaderItems, _HeaderItems], TracerProvider] = OrderedDict()
+        self._providers: OrderedDict[tuple[_HeaderItems, _HeaderItems, str | None], TracerProvider] = (
+            OrderedDict()  # mutable-ok: bounded LRU; eviction needs in-place ordered mutation
+        )
         self._open_span_counts: dict[TracerProvider, int] = {}  # mutable-ok: live refcount state
         # Oldest-first so an overflow of draining providers sheds the stalest.
         self._retired: OrderedDict[TracerProvider, None] = OrderedDict()  # mutable-ok: draining evicted providers
@@ -182,12 +185,16 @@ class TenantTracerCache:
         project_headers: Final = self._project_headers(auth_metadata)
         if not credential_headers and not project_headers:
             return TenantRoute(tracer=default, detached=False)
+        # A fixed per-integration region endpoint (New Relic us/eu), never a
+        # caller-supplied host; ``None`` keeps the preset's own endpoint.
+        endpoint: Final = dynamic_otlp_endpoint(self._callback_name, dynamic_params)
         cache_key: Final = (
             tuple(sorted(credential_headers.items())),
             tuple(sorted(project_headers.items())),
+            endpoint,
         )
         with self._lock:
-            provider: Final = self._cached_provider_locked(cache_key, credential_headers, project_headers)
+            provider: Final = self._cached_provider_locked(cache_key, credential_headers, project_headers, endpoint)
             self._open_span_counts[provider] = self._open_span_counts.get(provider, 0) + 1
             evicted: Final = self._evicted_on_overflow_locked()
         if evicted is not None:
@@ -200,15 +207,16 @@ class TenantTracerCache:
 
     def _cached_provider_locked(
         self,
-        cache_key: tuple[_HeaderItems, _HeaderItems],
+        cache_key: tuple[_HeaderItems, _HeaderItems, str | None],
         credential_headers: Mapping[str, str],
         project_headers: Mapping[str, str],
+        endpoint: str | None,
     ) -> TracerProvider:
         cached: Final = self._providers.get(cache_key)
         if cached is not None:
             self._providers.move_to_end(cache_key)
             return cached
-        built: Final = build_tracer_provider(self._routed_config(credential_headers, project_headers))
+        built: Final = build_tracer_provider(self._routed_config(credential_headers, project_headers, endpoint))
         self._providers[cache_key] = built
         return built
 
@@ -257,6 +265,7 @@ class TenantTracerCache:
         self,
         credential_headers: Mapping[str, str],
         project_headers: Mapping[str, str],
+        endpoint: str | None = None,
     ) -> OpenTelemetryV2Config:
         """Clone the config, rewriting headers on the callback's own exporter.
 
@@ -272,7 +281,8 @@ class TenantTracerCache:
         ``Authorization``), which must survive routing to a project.
         """
         exporters: Final = [
-            self._routed_exporter(spec, credential_headers, project_headers) for spec in self._config.exporters
+            self._routed_exporter(spec, credential_headers, project_headers, endpoint)
+            for spec in self._config.exporters
         ]
         return self._config.model_copy(update={"exporters": exporters})
 
@@ -281,6 +291,7 @@ class TenantTracerCache:
         spec: ExporterSpec,
         credential_headers: Mapping[str, str],
         project_headers: Mapping[str, str],
+        endpoint: str | None = None,
     ) -> ExporterSpec:
         kind: Final = spec.kind.lower()
         if spec.owner != self._callback_name or kind in _NON_OTLP_KINDS:
@@ -291,4 +302,10 @@ class TenantTracerCache:
             if project_headers and kind not in _GRPC_KINDS
             else base
         )
-        return spec if routed == spec.headers else spec.model_copy(update={"headers": routed})
+        update: Final = {  # mutable-ok: model_copy(update=...) requires a plain dict
+            field: value
+            for field, value in (("headers", routed), ("endpoint", endpoint))
+            if (field == "headers" and routed != spec.headers)
+            or (field == "endpoint" and endpoint is not None and endpoint != spec.endpoint)
+        }
+        return spec if not update else spec.model_copy(update=update)

litellm/integrations/otel/presets/__init__.py

@@ -21,6 +21,11 @@ from litellm.integrations.otel.presets.langfuse import (
 )
 from litellm.integrations.otel.presets.langtrace import langtrace_preset
 from litellm.integrations.otel.presets.levo import levo_preset
+from litellm.integrations.otel.presets.newrelic import (
+    newrelic_dynamic_endpoint,
+    newrelic_dynamic_headers,
+    newrelic_preset,
+)
 from litellm.integrations.otel.presets.phoenix import (
     phoenix_preset,
     phoenix_project_headers,
@@ -30,25 +35,45 @@ from litellm.types.utils import StandardCallbackDynamicParams
 
 #: Callback name → preset. The ``Preset`` annotation makes mypy verify every
 #: registered value matches the preset interface.
-PRESET_BY_CALLBACK: Final[dict[str, Preset]] = {
-    "agentops": agentops_preset,
-    "arize": arize_preset,
-    "arize_phoenix": phoenix_preset,
-    "langfuse_otel": langfuse_preset,
-    "langtrace": langtrace_preset,
-    "levo": levo_preset,
-    "weave_otel": weave_preset,
-}
+PRESET_BY_CALLBACK: Final[Mapping[str, Preset]] = MappingProxyType(
+    {
+        "agentops": agentops_preset,
+        "arize": arize_preset,
+        "arize_phoenix": phoenix_preset,
+        "langfuse_otel": langfuse_preset,
+        "langtrace": langtrace_preset,
+        "levo": levo_preset,
+        "newrelic": newrelic_preset,
+        "weave_otel": weave_preset,
+    }
+)
 
 #: Callback name → per-request OTLP header builder (team/key multi-tenant
 #: routing). Only integrations that support dynamic credentials appear here —
 #: Arize-Phoenix/Langtrace/Levo/AgentOps don't, so they use the logger's
 #: default tracer.
-DYNAMIC_HEADERS_BY_CALLBACK: Final[dict[str, Callable[[StandardCallbackDynamicParams], dict[str, str]]]] = {
-    "arize": arize_dynamic_headers,
-    "langfuse_otel": langfuse_dynamic_headers,
-    "weave_otel": weave_dynamic_headers,
-}
+DYNAMIC_HEADERS_BY_CALLBACK: Final[Mapping[str, Callable[[StandardCallbackDynamicParams], dict[str, str]]]] = (
+    MappingProxyType(
+        {
+            "arize": arize_dynamic_headers,
+            "langfuse_otel": langfuse_dynamic_headers,
+            "newrelic": newrelic_dynamic_headers,
+            "weave_otel": weave_dynamic_headers,
+        }
+    )
+)
+
+#: Callback name → per-request OTLP endpoint resolver. Only integrations whose
+#: destination host varies per tenant (from a fixed region table, never a
+#: caller-supplied URL) appear here; for everyone else the preset's endpoint is
+#: authoritative.
+DYNAMIC_ENDPOINT_BY_CALLBACK: Final[Mapping[str, Callable[[StandardCallbackDynamicParams], str | None]]] = (
+    MappingProxyType(
+        {
+            "newrelic": newrelic_dynamic_endpoint,
+        }
+    )
+)
 
 
 #: Callback name → per-request *routing* header builder, sourced from the key/team
@@ -98,17 +123,34 @@ def project_routing_headers(
     return builder(auth_metadata)
 
 
+def dynamic_otlp_endpoint(
+    callback_name: str | None,
+    dynamic_params: StandardCallbackDynamicParams | None,
+) -> str | None:
+    """Per-request OTLP endpoint for ``callback_name``, or ``None`` if N/A.
+
+    ``None`` means "keep the preset's own endpoint".
+    """
+    resolver: Final = DYNAMIC_ENDPOINT_BY_CALLBACK.get(callback_name or "")
+    if resolver is None or not dynamic_params:
+        return None
+    return resolver(dynamic_params)
+
+
 __all__ = [
+    "DYNAMIC_ENDPOINT_BY_CALLBACK",
     "DYNAMIC_HEADERS_BY_CALLBACK",
     "PRESET_BY_CALLBACK",
     "PROJECT_HEADERS_BY_CALLBACK",
     "Preset",
     "agentops_preset",
     "arize_preset",
+    "dynamic_otlp_endpoint",
     "dynamic_otlp_headers",
     "langfuse_preset",
     "langtrace_preset",
     "levo_preset",
+    "newrelic_preset",
     "phoenix_preset",
     "project_routing_headers",
     "weave_preset",

litellm/integrations/otel/presets/newrelic.py

@@ -0,0 +1,104 @@
+"""New Relic preset — OTLP/HTTP exporter to New Relic + GenAI vocabulary."""
+
+from collections.abc import Mapping
+from types import MappingProxyType
+from typing import Final
+
+from pydantic import Field
+from pydantic_settings import BaseSettings, SettingsConfigDict
+
+from litellm._logging import verbose_logger
+from litellm.integrations.otel.model.config import (
+    ExporterOwner,
+    ExporterSpec,
+    OpenTelemetryV2Config,
+)
+from litellm.integrations.otel.presets.utils import ensure_mappers
+from litellm.types.utils import StandardCallbackDynamicParams
+
+#: Region -> OTLP base endpoint. A fixed table by design: team config picks a
+#: region enum rather than a free-form endpoint, so callback vars can never
+#: redirect telemetry to an arbitrary host.
+NEWRELIC_OTLP_ENDPOINT_BY_REGION: Final[Mapping[str, str]] = MappingProxyType(
+    {
+        "us": "https://otlp.nr-data.net",
+        "eu": "https://otlp.eu01.nr-data.net",
+    }
+)
+
+_DEFAULT_REGION: Final = "us"
+
+
+class _NewRelicSettings(BaseSettings):
+    model_config = SettingsConfigDict(case_sensitive=False, extra="ignore")
+
+    # The same env vars the agent-based integration documents; the key is the
+    # operator-level fallback for traffic without team credentials, the region
+    # picks that fallback's data center, and the record-content flag keeps its
+    # documented meaning when the OTel path replaces the agent.
+    license_key: str | None = Field(default=None, validation_alias="NEW_RELIC_LICENSE_KEY")
+    region: str | None = Field(default=None, validation_alias="NEW_RELIC_REGION")
+    record_content: bool | None = Field(default=None, validation_alias="NEW_RELIC_AI_MONITORING_RECORD_CONTENT_ENABLED")
+
+
+def newrelic_preset(
+    *,
+    config_overrides: OpenTelemetryV2Config | None = None,
+) -> OpenTelemetryV2Config:
+    settings: Final = _NewRelicSettings()
+    base: Final = config_overrides or OpenTelemetryV2Config()
+    endpoint: Final = NEWRELIC_OTLP_ENDPOINT_BY_REGION.get(
+        (settings.region or _DEFAULT_REGION).lower(), NEWRELIC_OTLP_ENDPOINT_BY_REGION[_DEFAULT_REGION]
+    )
+    return base.model_copy(
+        update={
+            "exporters": [
+                *base.exporters,
+                ExporterSpec(
+                    kind="otlp_http",
+                    endpoint=endpoint,
+                    headers=(f"api-key={settings.license_key}" if settings.license_key else None),
+                    owner=ExporterOwner.NEWRELIC,
+                    requires_headers=True,
+                ),
+            ],
+            # New Relic ingests the OTLP GenAI semantic conventions natively.
+            "mapper_names": ensure_mappers(base.mapper_names, "genai"),
+            **(
+                {"capture_message_content": ("span_only" if settings.record_content else "no_content")}
+                if settings.record_content is not None
+                else {}
+            ),
+        }
+    )
+
+
+def newrelic_dynamic_headers(params: StandardCallbackDynamicParams) -> dict[str, str]:
+    """Per-request New Relic OTLP headers from team/key dynamic params."""
+    api_key: Final = params.get("newrelic_api_key")
+    return {header: value for header, value in (("api-key", api_key),) if value}
+
+
+def newrelic_dynamic_endpoint(params: StandardCallbackDynamicParams) -> str:
+    """Per-request OTLP endpoint for the team's ``newrelic_region``.
+
+    Always the team's own region endpoint, defaulting to US when the team left
+    the region unset. It never falls through to the preset's endpoint, which
+    follows the operator's ``NEW_RELIC_REGION`` env; a team that saved only its
+    ingest key must not inherit the operator's region and have its US-account
+    spans rejected by an EU-configured default (or vice versa). An unknown
+    region likewise resolves to the documented US default rather than a guess.
+    """
+    region: Final = params.get("newrelic_region")
+    default_endpoint: Final = NEWRELIC_OTLP_ENDPOINT_BY_REGION[_DEFAULT_REGION]
+    if not region:
+        return default_endpoint
+    endpoint: Final = NEWRELIC_OTLP_ENDPOINT_BY_REGION.get(region.lower())
+    if endpoint is None:
+        verbose_logger.warning(
+            "New Relic: unknown newrelic_region %r; supported regions: %s. Using the default (US) endpoint.",
+            region,
+            ", ".join(sorted(NEWRELIC_OTLP_ENDPOINT_BY_REGION)),
+        )
+        return default_endpoint
+    return endpoint

litellm/litellm_core_utils/initialize_dynamic_callback_params.py

@@ -46,7 +46,7 @@ def validate_no_callback_env_reference(param: str, value: object, *, source: str
 
 
 # Hardcoded list of supported callback params to avoid runtime inspection issues with TypedDict
-_supported_callback_params: Final = [
+_supported_callback_params: Final[tuple[str, ...]] = (
     "langfuse_public_key",
     "langfuse_secret",
     "langfuse_secret_key",
@@ -72,8 +72,10 @@ _supported_callback_params: Final = [
     "dd_site",
     "dd_agent_host",
     "dd_agent_port",
+    "newrelic_api_key",
+    "newrelic_region",
     "turn_off_message_logging",
-]
+)
 
 _request_blocked_callback_params: Final = frozenset(
     {
@@ -83,6 +85,20 @@ _request_blocked_callback_params: Final = frozenset(
         "dd_site",
         "dd_agent_host",
         "dd_agent_port",
+        "newrelic_api_key",
+        "newrelic_region",
+    }
+)
+
+# Request-blocked params that must still reach ``standard_callback_dynamic_params``
+# when the proxy itself stamped them from admin-configured team/key callback
+# settings (the trusted-vars channel). The OTel per-tenant tracer routing reads
+# ``standard_callback_dynamic_params``, so without this overlay a blocked param
+# could never drive routing at all.
+_trusted_overlay_callback_params: Final = frozenset(
+    {
+        "newrelic_api_key",
+        "newrelic_region",
     }
 )
 
@@ -121,7 +137,9 @@ def initialize_standard_callback_dynamic_params(
             if param in kwargs:
                 _param_value = kwargs.get(param)
                 validate_no_callback_env_reference(param, _param_value, source="request body")
-                standard_callback_dynamic_params[param] = _param_value
+                standard_callback_dynamic_params[param] = (  # pyright: ignore[reportGeneralTypeIssues]  # several supported params predate their StandardCallbackDynamicParams fields
+                    _param_value
+                )
 
         for slot_label, metadata in iter_client_callback_metadata_dicts(kwargs):
             for param in _supported_callback_params:
@@ -130,6 +148,12 @@ def initialize_standard_callback_dynamic_params(
                 if param not in standard_callback_dynamic_params and param in metadata:
                     _param_value = metadata.get(param)
                     validate_no_callback_env_reference(param, _param_value, source=slot_label)
-                    standard_callback_dynamic_params[param] = _param_value
+                    standard_callback_dynamic_params[param] = (  # pyright: ignore[reportGeneralTypeIssues]  # several supported params predate their StandardCallbackDynamicParams fields
+                        _param_value
+                    )
+
+        for param, trusted_value in get_trusted_callback_params(kwargs):
+            if param in _trusted_overlay_callback_params:
+                standard_callback_dynamic_params[param] = trusted_value
 
     return standard_callback_dynamic_params

litellm/litellm_core_utils/litellm_logging.py

@@ -4503,6 +4503,9 @@ def _init_custom_logger_compatible_class(
             _in_memory_loggers.append(gitlab_logger)
             return gitlab_logger
         elif logging_integration == "newrelic":
+            _v2 = _maybe_construct_otel_v2("newrelic", _in_memory_loggers)
+            if _v2 is not None:
+                return _v2
             for callback in _in_memory_loggers:
                 if isinstance(callback, NewRelicLogger):
                     return callback
@@ -4789,7 +4792,11 @@ def get_custom_logger_compatible_class(
                 if isinstance(callback, SMTPEmailLogger):
                     return callback
         elif logging_integration == "newrelic":
+            from litellm.integrations.otel.logger import OpenTelemetryV2
+
             for callback in _in_memory_loggers:
+                if isinstance(callback, OpenTelemetryV2) and callback.callback_name == "newrelic":
+                    return callback
                 if isinstance(callback, NewRelicLogger):
                     return callback
         return None

litellm/proxy/common_utils/callback_config_validation.py

@@ -0,0 +1,77 @@
+"""Save-time validation of team/key logging configs the runtime cannot honor.
+
+Team callbacks arrive as a single ``AddTeamCallback``, key callbacks arrive as a
+``logging`` list inside the key metadata, so both shapes funnel into the same
+per-integration checks here.
+"""
+
+from collections.abc import Mapping, Sequence
+from types import MappingProxyType
+from typing import Final
+
+_NEWRELIC_CALLBACK: Final = "newrelic"
+_NEWRELIC_VAR_PREFIX: Final = "newrelic_"
+
+
+def callback_config_error(callback_name: str | None, callback_vars: Mapping[str, str] | None) -> str | None:
+    if callback_name != _NEWRELIC_CALLBACK or not callback_vars:
+        return None
+    return _newrelic_config_error(callback_vars)
+
+
+def logging_metadata_config_error(metadata: Mapping[str, object] | None) -> str | None:
+    """Validate every ``logging`` entry of a team/key metadata payload."""
+    if not metadata:
+        return None
+    entries: Final = metadata.get("logging")
+    if not isinstance(entries, Sequence) or isinstance(entries, (str, bytes)):
+        return None
+    return next(
+        (error for error in (_logging_entry_error(entry) for entry in entries) if error is not None),
+        None,
+    )
+
+
+def _logging_entry_error(entry: object) -> str | None:
+    if not isinstance(entry, Mapping):
+        return None
+    callback_name: Final = entry.get("callback_name")
+    callback_vars: Final = entry.get("callback_vars")
+    if not isinstance(callback_name, str) or not isinstance(callback_vars, Mapping):
+        return None
+    return callback_config_error(
+        callback_name,
+        MappingProxyType({str(key): str(value) for key, value in callback_vars.items()}),
+    )
+
+
+def _newrelic_config_error(callback_vars: Mapping[str, str]) -> str | None:
+    """Per-team New Relic routing runs on the OTel v2 path only.
+
+    Accepting the config with the flag off would silently ship the team's traffic
+    through the operator's env-configured agent instead of the team's account. A
+    region outside the fixed table, or a region without a key, would likewise be
+    accepted and then silently ignored or misrouted at request time.
+    """
+    if not any(key.startswith(_NEWRELIC_VAR_PREFIX) for key in callback_vars):
+        return None
+
+    from litellm.integrations.otel.model.config import is_otel_v2_enabled
+    from litellm.integrations.otel.presets.newrelic import NEWRELIC_OTLP_ENDPOINT_BY_REGION
+
+    if not is_otel_v2_enabled():
+        return "Per-team New Relic routing requires the proxy to run with LITELLM_OTEL_V2=true."
+
+    region: Final = callback_vars.get("newrelic_region")
+    if region is not None and region.lower() not in NEWRELIC_OTLP_ENDPOINT_BY_REGION:
+        return (
+            f"Unknown newrelic_region {region!r}. "
+            f"Supported regions: {', '.join(sorted(NEWRELIC_OTLP_ENDPOINT_BY_REGION))}."
+        )
+
+    # ``callback_vars`` values are str()-coerced upstream, so a JSON ``null`` key
+    # arrives as the literal ``"None"``; treat that and the empty string as absent.
+    api_key: Final = callback_vars.get("newrelic_api_key")
+    if region is not None and (not api_key or api_key == "None"):
+        return "newrelic_region requires newrelic_api_key; the region rides the team's own key."
+    return None

litellm/proxy/litellm_pre_call_utils.py

@@ -754,6 +754,12 @@ def convert_key_logging_metadata_to_callback(
             team_callback_settings_obj.callbacks.append(data.callback_name)
 
     for var, value in data.callback_vars.items():
+        # New Relic routing reads these from the trusted-vars overlay with no
+        # callback-name check, so scope them to the newrelic entry: a team that
+        # put newrelic_* under a different callback never asked for New Relic and
+        # must not export to it.
+        if var.startswith("newrelic_") and data.callback_name != "newrelic":
+            continue
         if team_callback_settings_obj.callback_vars is None:
             team_callback_settings_obj.callback_vars = {}
         team_callback_settings_obj.callback_vars[var] = str(value)

litellm/proxy/management_endpoints/key_management_endpoints.py

@@ -62,6 +62,7 @@ from litellm.proxy.auth.auth_utils import (
     enforce_output_token_estimates_are_admin_only,
 )
 from litellm.proxy.auth.user_api_key_auth import user_api_key_auth
+from litellm.proxy.common_utils.callback_config_validation import logging_metadata_config_error
 from litellm.proxy.common_utils.callback_utils import (
     decrypt_callback_vars,
     encrypt_callback_vars,
@@ -553,6 +554,17 @@ def key_generation_check(
         return _personal_key_generation_check(user_api_key_dict=user_api_key_dict, data=data)
 
 
+def raise_on_invalid_key_logging_config(metadata: Mapping[str, object] | None) -> None:
+    """Key-level logging writes go through key metadata, not /team/callback.
+
+    Without this the same New Relic config the team endpoint rejects would be
+    accepted here and then silently ignored or misrouted at request time.
+    """
+    error: Final = logging_metadata_config_error(metadata)
+    if error is not None:
+        raise HTTPException(status_code=400, detail={"error": error})  # mutable-ok: FastAPI detail contract
+
+
 def common_key_access_checks(
     user_api_key_dict: UserAPIKeyAuth,
     data: GenerateKeyRequest | UpdateKeyRequest,
@@ -891,6 +903,7 @@ async def _common_key_generation_helper(
     )
 
     validate_budget_duration(data.budget_duration)
+    raise_on_invalid_key_logging_config(data.metadata)
 
     if data.throttle_on_budget_exceeded is True and user_api_key_dict.user_role != LitellmUserRoles.PROXY_ADMIN.value:
         raise HTTPException(
@@ -1992,6 +2005,8 @@ def prepare_metadata_fields(data: BaseModel, non_default_values: dict, existing_
     """
     Check LiteLLM_ManagementEndpoint_MetadataFields (proxy/_types.py) for fields that are allowed to be updated
     """
+    raise_on_invalid_key_logging_config(non_default_values.get("metadata"))
+
     if "metadata" not in non_default_values:  # allow user to set metadata to none
         non_default_values["metadata"] = existing_metadata.copy()
 

litellm/proxy/management_endpoints/team_callback_endpoints.py

@@ -28,6 +28,7 @@ from litellm.proxy._types import (
     UserAPIKeyAuth,
 )
 from litellm.proxy.auth.user_api_key_auth import user_api_key_auth
+from litellm.proxy.common_utils.callback_config_validation import callback_config_error
 from litellm.proxy.common_utils.callback_utils import (
     _CALLBACK_VAR_ENCRYPTED_PREFIX,
     decrypt_callback_vars,
@@ -51,6 +52,16 @@ router: Final = APIRouter()
 _CALLBACK_VARS_REDACTED: Final = "***REDACTED***"
 
 
+def _callback_config_error(message: str) -> HTTPException:
+    return HTTPException(status_code=400, detail={"error": message})  # mutable-ok: FastAPI detail contract
+
+
+def _validate_team_callback(data: "AddTeamCallback") -> None:
+    error: Final = callback_config_error(data.callback_name, data.callback_vars)
+    if error is not None:
+        raise _callback_config_error(error)
+
+
 def _redact_callback_secrets(metadata: Any) -> Any:
     """Strip secret values out of a team-metadata snapshot before audit logging.
 
@@ -304,6 +315,8 @@ async def add_team_callbacks(
             user_api_key_dict=user_api_key_dict,
         )
 
+        _validate_team_callback(data)
+
         # store team callback settings in metadata
         team_metadata = _existing_team.metadata
         team_callback_settings: list[dict] = team_metadata.get("logging")  # will be dict of type AddTeamCallback

litellm/types/utils.py

@@ -3300,6 +3300,11 @@ class StandardCallbackDynamicParams(TypedDict, total=False):
     dd_agent_host: str | None
     dd_agent_port: str | None
 
+    # New Relic dynamic params (proxy-stamped team/key callback vars only;
+    # request-supplied values are blocked)
+    newrelic_api_key: str | None  # writable-ok: initialize_standard_callback_dynamic_params assigns into the dict
+    newrelic_region: str | None  # writable-ok: initialize_standard_callback_dynamic_params assigns into the dict
+
     # Logging settings
     turn_off_message_logging: bool | None  # when true will not log messages
     litellm_disabled_callbacks: list[str] | None

scripts/check_test_quality.py

@@ -48,6 +48,10 @@ TQ006   A `pytest.skip` reached only when a credential-shaped environment variab
         deliberate branch. The gate follows one local or module-level binding, which is
         the `key = os.getenv(...)` then `if not key: pytest.skip(...)` shape most of
         these use.
+TQ008   A `patch(...)` whose target is a `litellm.` internal. Patching the SDK's own
+        functions pins the test to the current wiring instead of the behaviour, and it
+        is the idiom the suite reaches for instead of faking the HTTP boundary. Mocking
+        a third-party client, a transport, or anything outside `litellm.` is untouched.
 TQ007   A module global that a conftest saves before every test and restores after it.
         The save/restore list is a hand-maintained inventory of the leaks the suite
         already knows about, so it is allowed to shrink and never to grow: a new entry
@@ -102,11 +106,13 @@ from __future__ import annotations
 
 import ast
 import io
+import os
 import re
 import sys
 import tokenize
 from collections.abc import Iterable, Iterator, Mapping, Sequence
 from dataclasses import dataclass
+from multiprocessing import Pool
 from pathlib import Path
 from types import MappingProxyType
 from typing import Final, NamedTuple
@@ -467,6 +473,67 @@ def iter_global_mutation_violations(path: Path, tree: ast.Module) -> Iterator[Vi
                 )
 
 
+def _is_sdk_internal(dotted: str) -> bool:
+    return dotted == SDK_MODULE or dotted.startswith(f"{SDK_MODULE}.")
+
+
+def _sdk_import_bindings(tree: ast.Module) -> Iterator[tuple[str, str]]:
+    """(local name, dotted path) for every import that binds something under `litellm`."""
+    for node in ast.walk(tree):
+        if isinstance(node, ast.Import):
+            yield from (
+                (alias.asname, alias.name) if alias.asname else (root, root)
+                for alias in node.names
+                if _is_sdk_internal(alias.name)
+                for root in (alias.name.partition(".")[0],)
+            )
+        elif isinstance(node, ast.ImportFrom) and node.module and _is_sdk_internal(node.module):
+            yield from ((alias.asname or alias.name, f"{node.module}.{alias.name}") for alias in node.names)
+
+
+def _sdk_aliases(tree: ast.Module) -> Mapping[str, str]:
+    """Local names bound to something under `litellm`, mapped to the path they stand for.
+
+    `from litellm.llms.openai.chat import handler` then `patch.object(handler.X, ...)`
+    reaches the same internal as the dotted string form and has to read the same way.
+    """
+    return MappingProxyType({name: dotted for name, dotted in _sdk_import_bindings(tree)})
+
+
+def _resolved(dotted: str, aliases: Mapping[str, str]) -> str:
+    root, _, rest = dotted.partition(".")
+    base: Final = aliases.get(root, root)
+    return f"{base}.{rest}" if rest else base
+
+
+def _patch_targets(call: ast.Call, aliases: Mapping[str, str]) -> Iterator[str]:
+    """What a patch installer is replacing: the dotted string it names, or the
+    attribute chain handed to `patch.object` / `patch.dict`, resolved through the
+    module's imports so a locally bound SDK object reads as its full path."""
+    for first in call.args[:1]:
+        if isinstance(first, ast.Constant) and isinstance(first.value, str):
+            yield first.value
+        elif dotted := _dotted_name(first):
+            yield _resolved(dotted, aliases)
+
+
+def iter_internal_patch_violations(path: Path, tree: ast.Module) -> Iterator[Violation]:
+    aliases: Final = _sdk_aliases(tree)
+    for node in ast.walk(tree):
+        if not (isinstance(node, ast.Call) and _is_patch_installer(_dotted_name(node.func))):
+            continue
+        for target in _patch_targets(node, aliases):
+            if _is_sdk_internal(target):
+                yield Violation(
+                    path,
+                    node.lineno,
+                    "TQ008",
+                    f"patches `{target}`, an SDK internal, so the test is pinned to how the code is "
+                    "wired rather than what it does; fake the HTTP boundary (respx / MockTransport) "
+                    f"or inject the collaborator (suppress: `# {SUPPRESSION_TOKEN}: <reason>`)",
+                )
+
+
 def _environ_keys(node: ast.AST) -> Iterator[str]:
     for inner in ast.walk(node):
         if isinstance(inner, ast.Call) and _dotted_name(inner.func) in ENVIRON_READERS:
@@ -678,6 +745,7 @@ def check_file(path: Path) -> tuple[Violation, ...]:
             *iter_global_mutation_violations(path, tree),
             *iter_credential_skip_violations(path, tree),
             *iter_conftest_inventory_violations(path, tree),
+            *iter_internal_patch_violations(path, tree),
         )
         if violation.line not in skip
     )
@@ -692,13 +760,36 @@ def collect_paths(raw: Iterable[str]) -> Iterator[Path]:
             yield candidate
 
 
+PARALLEL_MIN_PATHS = 200
+MAX_WORKERS = 8
+
+
+def _worker_count(path_count: int) -> int:
+    """1 when the run is too small to repay process startup, else one worker per
+    core up to MAX_WORKERS."""
+    if path_count < PARALLEL_MIN_PATHS:
+        return 1
+    return max(1, min(os.cpu_count() or 1, MAX_WORKERS))
+
+
+def scan_paths(paths: Sequence[Path]) -> tuple[Violation, ...]:
+    """check_file over every path. Pure per-file work, so it fans out across
+    processes; callers sort, which is what keeps output order stable."""
+    workers = _worker_count(len(paths))
+    if workers == 1:
+        return tuple(v for path in paths for v in check_file(path))
+    with Pool(workers) as pool:
+        return tuple(v for found in pool.imap_unordered(check_file, paths, chunksize=32) for v in found)
+
+
 def main(argv: Sequence[str]) -> int:
     paths: Final = tuple(a for a in argv if not a.startswith("-"))
     if not paths:
         print("usage: check_test_quality.py <files-or-dirs>...", file=sys.stderr)
         return 2
 
-    violations: Final = sorted(v for path in collect_paths(paths) for v in check_file(path))
+    targets: Final = tuple(collect_paths(paths))
+    violations: Final = sorted(scan_paths(targets))
     for violation in violations:
         print(violation.render())
 

scripts/check_type_discipline.py

@@ -114,10 +114,12 @@ from __future__ import annotations
  
 import ast
 import io
+import os
 import re
 import sys
 import tokenize
 from dataclasses import dataclass
+from multiprocessing import Pool
 from pathlib import Path
 from collections.abc import Iterable, Iterator, Mapping, Sequence
 from typing import NamedTuple
@@ -1070,13 +1072,36 @@ def collect_paths(raw: Iterable[str]) -> Iterator[Path]:
             yield p
  
  
+PARALLEL_MIN_PATHS = 200
+MAX_WORKERS = 8
+
+
+def _worker_count(path_count: int) -> int:
+    """1 when the run is too small to repay process startup, else one worker per
+    core up to MAX_WORKERS."""
+    if path_count < PARALLEL_MIN_PATHS:
+        return 1
+    return max(1, min(os.cpu_count() or 1, MAX_WORKERS))
+
+
+def scan_paths(paths: Sequence[Path]) -> tuple[Violation, ...]:
+    """check_file over every path. Pure per-file work, so it fans out across
+    processes; callers sort, which is what keeps output order stable."""
+    workers = _worker_count(len(paths))
+    if workers == 1:
+        return tuple(v for path in paths for v in check_file(path))
+    with Pool(workers) as pool:
+        return tuple(v for found in pool.imap_unordered(check_file, paths, chunksize=32) for v in found)
+
+
 def main(argv: Sequence[str]) -> int:
     paths = tuple(a for a in argv if not a.startswith("-"))
     if not paths:
         print("usage: check_type_discipline.py <files-or-dirs>...", file=sys.stderr)
         return 2
  
-    violations = sorted(v for path in collect_paths(paths) for v in check_file(path))
+    targets = tuple(collect_paths(paths))
+    violations = sorted(scan_paths(targets))
     for v in violations:
         print(v.render())
  

test-quality-budget.json

@@ -19,5 +19,8 @@
   },
   "TQ007": {
     "limit": 117
+  },
+  "TQ008": {
+    "limit": 11139
   }
 }

tests/test_litellm/integrations/otel/test_otel_v2_dynamic.py

@@ -7,6 +7,7 @@ from opentelemetry.trace import NoOpTracer
 
 from litellm.integrations.otel.model.config import ExporterSpec, OpenTelemetryV2Config
 from litellm.integrations.otel.presets import (
+    dynamic_otlp_endpoint,
     dynamic_otlp_headers,
     project_routing_headers,
 )
@@ -23,31 +24,23 @@ def _cache(callback_name, exporters=None):
 
 
 def test_arize_dynamic_headers():
-    headers = dynamic_otlp_headers(
-        "arize", {"arize_space_id": "S", "arize_api_key": "K"}
-    )
+    headers = dynamic_otlp_headers("arize", {"arize_space_id": "S", "arize_api_key": "K"})
     assert headers == {"arize-space-id": "S", "api_key": "K"}
 
 
 def test_arize_space_key_overrides_space_id():
-    headers = dynamic_otlp_headers(
-        "arize", {"arize_space_id": "S", "arize_space_key": "SK"}
-    )
+    headers = dynamic_otlp_headers("arize", {"arize_space_id": "S", "arize_space_key": "SK"})
     assert headers == {"arize-space-id": "SK"}
 
 
 def test_langfuse_dynamic_headers_need_both_keys():
     assert dynamic_otlp_headers("langfuse_otel", {"langfuse_public_key": "pk"}) is None
-    headers = dynamic_otlp_headers(
-        "langfuse_otel", {"langfuse_public_key": "pk", "langfuse_secret_key": "sk"}
-    )
+    headers = dynamic_otlp_headers("langfuse_otel", {"langfuse_public_key": "pk", "langfuse_secret_key": "sk"})
     assert headers is not None and "Authorization" in headers
 
 
 def test_langfuse_dynamic_headers_carry_v4_ingestion_version():
-    headers = dynamic_otlp_headers(
-        "langfuse_otel", {"langfuse_public_key": "pk", "langfuse_secret_key": "sk"}
-    )
+    headers = dynamic_otlp_headers("langfuse_otel", {"langfuse_public_key": "pk", "langfuse_secret_key": "sk"})
     expected_auth = "Basic " + base64.b64encode(b"pk:sk").decode()
     assert headers == {
         "Authorization": expected_auth,
@@ -56,9 +49,7 @@ def test_langfuse_dynamic_headers_carry_v4_ingestion_version():
 
 
 def test_weave_dynamic_headers():
-    headers = dynamic_otlp_headers(
-        "weave_otel", {"wandb_api_key": "w", "weave_project_id": "p"}
-    )
+    headers = dynamic_otlp_headers("weave_otel", {"wandb_api_key": "w", "weave_project_id": "p"})
     assert headers is not None
     assert "Authorization" in headers and headers["project_id"] == "p"
 
@@ -100,9 +91,7 @@ def test_provider_cache_is_bounded_and_evicts_lru(monkeypatch):
 
     monkeypatch.setattr(routing_mod, "_MAX_CACHED_PROVIDERS", 2)
     shut_down = []
-    monkeypatch.setattr(
-        routing_mod, "_shutdown_provider", lambda p: shut_down.append(p)
-    )
+    monkeypatch.setattr(routing_mod, "_shutdown_provider", lambda p: shut_down.append(p))
 
     cache = _cache("arize")
     default = NoOpTracer()
@@ -179,9 +168,7 @@ def test_dynamic_headers_do_not_leak_to_other_owners_exporter():
             ),
         ],
     )
-    new_cfg = cache._routed_config(
-        {"arize-space-id": "TEAMX", "api_key": "TEAMX_KEY"}, {}
-    )
+    new_cfg = cache._routed_config({"arize-space-id": "TEAMX", "api_key": "TEAMX_KEY"}, {})
     by_owner = {e.owner: e.headers for e in new_cfg.exporters}
     assert by_owner["arize"] == "arize-space-id=TEAMX,api_key=TEAMX_KEY"
     assert by_owner[None] == "x=base-collector"
@@ -206,16 +193,14 @@ def _phoenix_cache(kind="otlp_http"):
 
 
 def test_phoenix_project_headers_precedence_and_blanks():
-    assert project_routing_headers(
-        "arize_phoenix", {"phoenix_project_name": "team-proj"}
-    ) == {"x-project-name": "team-proj"}
+    assert project_routing_headers("arize_phoenix", {"phoenix_project_name": "team-proj"}) == {
+        "x-project-name": "team-proj"
+    }
     assert project_routing_headers(
         "arize_phoenix",
         {"phoenix_project_name_override": "override", "phoenix_project_name": "base"},
     ) == {"x-project-name": "override"}
-    assert (
-        project_routing_headers("arize_phoenix", {"phoenix_project_name": "   "}) == {}
-    )
+    assert project_routing_headers("arize_phoenix", {"phoenix_project_name": "   "}) == {}
     assert project_routing_headers("arize_phoenix", None) == {}
     # Only Phoenix participates in project routing.
     assert project_routing_headers("arize", {"phoenix_project_name": "p"}) == {}
@@ -286,10 +271,7 @@ def test_client_dynamic_params_cannot_choose_phoenix_project():
     cache = _phoenix_cache()
     default = NoOpTracer()
     assert cache.route_for(default, {"phoenix_project_name": "attacker"}).tracer is default
-    assert (
-        cache.route_for(default, {"phoenix_project_name_override": "attacker"}).tracer
-        is default
-    )
+    assert cache.route_for(default, {"phoenix_project_name_override": "attacker"}).tracer is default
     assert cache._providers == {}
 
 
@@ -321,9 +303,7 @@ def test_eviction_defers_shutdown_while_a_span_is_open(monkeypatch):
 
     monkeypatch.setattr(routing_mod, "_MAX_CACHED_PROVIDERS", 1)
     shut_down = []
-    monkeypatch.setattr(
-        routing_mod, "_shutdown_provider", lambda p: shut_down.append(p)
-    )
+    monkeypatch.setattr(routing_mod, "_shutdown_provider", lambda p: shut_down.append(p))
     cache = _cache("arize")
     default = NoOpTracer()
 
@@ -347,18 +327,13 @@ def test_retired_providers_are_capped(monkeypatch):
     monkeypatch.setattr(routing_mod, "_MAX_CACHED_PROVIDERS", 1)
     monkeypatch.setattr(routing_mod, "_MAX_RETIRED_PROVIDERS", 2)
     shut_down = []
-    monkeypatch.setattr(
-        routing_mod, "_shutdown_provider", lambda p: shut_down.append(p)
-    )
+    monkeypatch.setattr(routing_mod, "_shutdown_provider", lambda p: shut_down.append(p))
     cache = _cache("arize")
     default = NoOpTracer()
 
     # Every route stays held (no release), so each one evicts and retires its
     # predecessor instead of shutting it down.
-    routes = [
-        cache.route_for(default, {"arize_space_id": str(i), "arize_api_key": "K"})
-        for i in range(5)
-    ]
+    routes = [cache.route_for(default, {"arize_space_id": str(i), "arize_api_key": "K"}) for i in range(5)]
 
     assert len(cache._providers) == 1
     assert len(cache._retired) == 2  # capped, not one retiree per open call
@@ -374,11 +349,112 @@ def test_release_without_eviction_keeps_provider_alive(monkeypatch):
     from litellm.integrations.otel.plumbing import routing as routing_mod
 
     shut_down = []
-    monkeypatch.setattr(
-        routing_mod, "_shutdown_provider", lambda p: shut_down.append(p)
-    )
+    monkeypatch.setattr(routing_mod, "_shutdown_provider", lambda p: shut_down.append(p))
     cache = _cache("arize")
     route = cache.route_for(NoOpTracer(), {"arize_space_id": "A", "arize_api_key": "K"})
     cache.release(route.provider)
     assert shut_down == []  # still cached, never retired
     cache.release(None)  # default-route release is a no-op
+
+
+# --- New Relic: per-team api-key header + fixed-table region endpoint --- #
+
+
+def test_newrelic_dynamic_headers():
+    assert dynamic_otlp_headers("newrelic", {"newrelic_api_key": "NRAL-KEY"}) == {"api-key": "NRAL-KEY"}
+    assert dynamic_otlp_headers("newrelic", {"newrelic_region": "eu"}) is None
+
+
+def test_newrelic_dynamic_endpoint_resolves_from_fixed_table():
+    from litellm.integrations.otel.presets import dynamic_otlp_endpoint
+
+    assert dynamic_otlp_endpoint("newrelic", {"newrelic_region": "eu"}) == "https://otlp.eu01.nr-data.net"
+    assert dynamic_otlp_endpoint("newrelic", {"newrelic_region": "US"}) == "https://otlp.nr-data.net"
+    # A key-only team (no region) resolves to the fixed US default deterministically,
+    # never the operator's NEW_RELIC_REGION-configured preset endpoint.
+    assert dynamic_otlp_endpoint("newrelic", {"newrelic_api_key": "k"}) == "https://otlp.nr-data.net"
+    # An unknown region also resolves to the documented US default, not a guess.
+    assert dynamic_otlp_endpoint("newrelic", {"newrelic_region": "mars"}) == "https://otlp.nr-data.net"
+    # Callbacks without an endpoint resolver keep their preset endpoint.
+    assert dynamic_otlp_endpoint("arize", {"newrelic_region": "eu"}) is None
+
+
+def test_newrelic_endpoint_stamped_onto_owned_exporter_only():
+    cache = _cache(
+        "newrelic",
+        exporters=[
+            ExporterSpec(
+                kind="otlp_http",
+                endpoint="http://self-hosted-collector:4318",
+                headers="x=base-collector",
+                owner=None,
+            ),
+            ExporterSpec(
+                kind="otlp_http",
+                endpoint="https://otlp.nr-data.net",
+                owner="newrelic",
+                requires_headers=True,
+            ),
+        ],
+    )
+    new_cfg = cache._routed_config({"api-key": "TEAM-EU-KEY"}, {}, "https://otlp.eu01.nr-data.net")
+    by_owner = {e.owner: e for e in new_cfg.exporters}
+    assert by_owner["newrelic"].endpoint == "https://otlp.eu01.nr-data.net"
+    assert by_owner["newrelic"].headers == "api-key=TEAM-EU-KEY"
+    assert by_owner[None].endpoint == "http://self-hosted-collector:4318"
+    assert by_owner[None].headers == "x=base-collector"
+
+
+def test_newrelic_provider_cached_per_key_and_region():
+    cache = _cache(
+        "newrelic",
+        exporters=[ExporterSpec(kind="in_memory"), ExporterSpec(kind="otlp_http", owner="newrelic")],
+    )
+    default = NoOpTracer()
+    cache.route_for(default, {"newrelic_api_key": "K1", "newrelic_region": "us"})
+    cache.route_for(default, {"newrelic_api_key": "K1", "newrelic_region": "us"})
+    assert len(cache._providers) == 1
+    # Same key, different region → distinct provider (distinct endpoint).
+    cache.route_for(default, {"newrelic_api_key": "K1", "newrelic_region": "eu"})
+    assert len(cache._providers) == 2
+    cache.route_for(default, {"newrelic_api_key": "K2", "newrelic_region": "eu"})
+    assert len(cache._providers) == 3
+
+
+def test_requires_headers_spec_skipped_without_headers():
+    from litellm.integrations.otel.plumbing.providers import build_tracer_provider
+
+    cfg = OpenTelemetryV2Config(
+        exporters=[ExporterSpec(kind="otlp_http", endpoint="https://otlp.nr-data.net", requires_headers=True)]
+    )
+    provider = build_tracer_provider(cfg)
+    processors = provider._active_span_processor._span_processors
+    # Only the baggage processor: the keyless spec must not export (New Relic
+    # rejects unauthenticated posts with a 4xx per span batch).
+    assert [type(p).__name__ for p in processors] == ["LiteLLMBaggageSpanProcessor"]
+
+    keyed = OpenTelemetryV2Config(
+        exporters=[
+            ExporterSpec(
+                kind="otlp_http", endpoint="https://otlp.nr-data.net", headers="api-key=k", requires_headers=True
+            )
+        ]
+    )
+    keyed_provider = build_tracer_provider(keyed)
+    assert len(keyed_provider._active_span_processor._span_processors) == 2
+
+
+def test_newrelic_key_only_team_routes_to_us_not_operator_region(monkeypatch):
+    """A team that saves a key but no region must export to the fixed US default,
+    independent of the operator's NEW_RELIC_REGION, so its spans are never
+    silently dropped by an operator-configured region its key does not match."""
+    monkeypatch.setenv("NEW_RELIC_REGION", "eu")
+    cache = _cache(
+        "newrelic",
+        exporters=[ExporterSpec(kind="otlp_http", endpoint="https://otlp.eu01.nr-data.net", owner="newrelic")],
+    )
+    new_cfg = cache._routed_config(
+        {"api-key": "US-KEY"}, {}, dynamic_otlp_endpoint("newrelic", {"newrelic_api_key": "US-KEY"})
+    )
+    owned = next(e for e in new_cfg.exporters if e.owner == "newrelic")
+    assert owned.endpoint == "https://otlp.nr-data.net"

tests/test_litellm/integrations/otel/test_otel_v2_logger.py

@@ -341,6 +341,35 @@ def test_idempotent_on_repeat_callback():
     assert len(exporter.get_finished_spans()) == 1
 
 
+def test_evicted_carrier_completed_call_emits_one_deferred_span():
+    """Eviction over the concurrency budget drops only the boundary carrier, not
+    the call. When an evicted call later closes as a real completed call
+    (``upstream_started``, payload present) it still emits exactly one span
+    through the deferred branch, and a second close for the same id dedups. Only
+    an evicted call that never closes goes unexported."""
+    logger, exporter = _logger()
+    kwargs = {**_kwargs(), "api_call_start_time": datetime(2026, 5, 26, 12, 0, 0, tzinfo=timezone.utc)}
+    logger.log_pre_api_call(model="gpt-4o", messages=[], kwargs=kwargs)
+    assert "call_1" in logger._open_llm_calls
+
+    # Evict exactly as ``_store_open_call`` does over budget: drop the oldest
+    # carrier and release its routed provider.
+    _, evicted = logger._open_llm_calls.popitem(last=False)
+    logger._release_carrier(evicted)
+    assert not logger._open_llm_calls
+    assert exporter.get_finished_spans() == ()  # the evicted boundary span is never exported
+
+    asyncio.run(logger.async_log_success_event(kwargs, None, None, None))
+    spans = exporter.get_finished_spans()
+    assert len(spans) == 1, "the evicted call's real close re-emits one deferred span, not zero"
+    assert spans[0].name == "chat gpt-4o"
+    assert spans[0].attributes[LiteLLM.CALL_ID] == "call_1"
+
+    # Success-then-failure on one logging object: the deferred branch dedups by id.
+    asyncio.run(logger.async_log_failure_event(kwargs, None, None, None))
+    assert len(exporter.get_finished_spans()) == 1, "second close for the same id must not duplicate"
+
+
 # --------------------------------------------------------------------------- #
 #  MCP tool-call spans
 # --------------------------------------------------------------------------- #
@@ -2522,3 +2551,149 @@ def test_deferred_pre_call_does_not_churn_tenant_cache(monkeypatch):
     server.end()
     headers_b = next(h for h in captured if "proj-b" in h)
     assert [s.name for s in captured[headers_b].get_finished_spans()] == ["chat gpt-4o"]
+
+
+# --- New Relic team-scoped deferred emit + dedup (no pre_call carrier) --- #
+
+
+def test_no_span_when_request_never_reached_upstream():
+    """A request rejected before the upstream call — at the auth/budget gate, or
+    blocked by a pre-call guardrail — carries the ``no upstream call`` marker
+    (stamped in ``proxy/utils.py`` before its handlers fire), so the failure log
+    produces no phantom CLIENT span even though a payload exists."""
+    from litellm.constants import LITELLM_LOGGING_NO_UPSTREAM_LLM_CALL
+
+    logger, exporter = _logger()
+    payload = _payload(
+        status="failure",
+        error_information={"error_class": "ProxyException", "error_code": "401"},
+    )
+    kwargs = _kwargs(payload=payload)
+    kwargs[LITELLM_LOGGING_NO_UPSTREAM_LLM_CALL] = True
+    # No log_pre_api_call: the call never started.
+    asyncio.run(logger.async_log_failure_event(kwargs, None, None, None))
+    assert exporter.get_finished_spans() == ()  # no phantom LLM span
+
+
+def test_success_without_pre_call_emits_deferred_span():
+    """A team/key-scoped logger is registered as a success callback only, so
+    ``pre_call`` never reaches it and no carrier exists. A completed call (it has
+    its payload, no ``no upstream call`` marker) must still get its span — the
+    deferred branch — or team-scoped destinations receive nothing at all."""
+    logger, exporter = _logger()
+    # No log_pre_api_call: this logger never receives the input hook. The
+    # request-level provider-handoff stamp is present (pre_call ran globally).
+    asyncio.run(
+        logger.async_log_success_event({**_kwargs(), "api_call_start_time": 100.0}, None, 100.0, 101.5)
+    )
+    spans = exporter.get_finished_spans()
+    assert len(spans) == 1
+    assert spans[0].attributes.get("gen_ai.operation.name")
+    # Start time comes from the callback's start_time, not a bogus zero.
+    assert spans[0].start_time == 100_000_000_000
+    assert spans[0].end_time == 101_500_000_000
+
+
+def test_no_carrier_and_no_payload_is_noop():
+    logger, exporter = _logger()
+    asyncio.run(
+        logger.async_log_success_event({"litellm_params": {}}, None, None, None)
+    )
+    assert exporter.get_finished_spans() == ()
+
+
+def test_second_close_for_same_call_does_not_duplicate_span():
+    """Success and failure can both fire on one logging object for the same call
+    id. The first close pops the carrier and finishes the boundary span; the
+    second must dedup against it, not fabricate a duplicate through the
+    deferred branch."""
+    logger, exporter = _logger()
+    kwargs = {**_kwargs(), "api_call_start_time": 100.0}
+    # Boundary open: the span is born at pre_call under a live server span and
+    # closed via finish_span, which never passes through emit()'s dedup.
+    server = logger._emitter.start_span(SpanRole.PROXY_REQUEST, LITELLM_PROXY_REQUEST_SPAN_NAME)
+    _emit_llm(logger, kwargs, ambient=server)
+    server.end()
+    llm_before = [s for s in exporter.get_finished_spans() if s.name.startswith("chat")]
+    assert len(llm_before) == 1
+    # Second close: carrier already popped, payload still present, handoff stamped.
+    asyncio.run(logger.async_log_failure_event(kwargs, None, None, None))
+    llm_after = [s for s in exporter.get_finished_spans() if s.name.startswith("chat")]
+    assert len(llm_after) == 1
+
+
+def test_failure_without_pre_call_emits_deferred_error_span():
+    """A team-scoped logger registered as a failure callback only still gets an
+    ERROR span for a real provider failure (payload present, no marker)."""
+    from opentelemetry.trace import StatusCode
+
+    logger, exporter = _logger()
+    payload = _payload(
+        status="failure",
+        error_information={"error_class": "RateLimitError", "error_code": "429"},
+    )
+    asyncio.run(
+        logger.async_log_failure_event(
+            {**_kwargs(payload=payload), "api_call_start_time": 100.0}, None, None, None
+        )
+    )
+    spans = exporter.get_finished_spans()
+    assert len(spans) == 1
+    assert spans[0].status.status_code == StatusCode.ERROR
+
+
+def test_boundary_open_with_no_payload_ends_provisional_span():
+    """Opened at pre_call but the payload never materialized: the boundary span
+    is ended provisionally (no payload attributes) rather than leaked open."""
+    logger, exporter = _logger()
+    kwargs = _kwargs()
+    server = logger._emitter.start_span(SpanRole.PROXY_REQUEST, LITELLM_PROXY_REQUEST_SPAN_NAME)
+    with trace.use_span(server, end_on_exit=False):
+        logger.log_pre_api_call(model="gpt-4o", messages=[], kwargs=kwargs)
+    asyncio.run(
+        logger.async_log_success_event(
+            {**kwargs, "standard_logging_object": None, "litellm_call_id": "call_1"}, None, None, None
+        )
+    )
+    server.end()
+    llm_spans = [s for s in exporter.get_finished_spans() if s.name.startswith("chat")]
+    assert len(llm_spans) == 1
+    assert "gen_ai.usage.input_tokens" not in llm_spans[0].attributes
+
+
+def test_failure_before_provider_handoff_emits_nothing():
+    """A failure event whose request never handed off to a provider (router
+    pre-call rejection, SDK error before the call, standalone guardrail run)
+    has a payload but no ``api_call_start_time``; without a carrier it must not
+    fabricate an LLM-call span."""
+    logger, exporter = _logger()
+    payload = _payload(
+        status="failure",
+        error_information={"error_class": "RateLimitError", "error_code": "429"},
+    )
+    asyncio.run(logger.async_log_failure_event(_kwargs(payload=payload), None, None, None))
+    assert exporter.get_finished_spans() == ()
+
+
+def test_provisional_close_then_payload_close_does_not_duplicate():
+    """Streaming shape: the success close arrives with no assembled payload (the
+    boundary span is ended provisionally), then the failure close arrives with a
+    payload for the same call id. Exactly one exported span."""
+    logger, exporter = _logger()
+    kwargs = {**_kwargs(), "api_call_start_time": 100.0}
+    payload = kwargs["standard_logging_object"]
+    server = logger._emitter.start_span(SpanRole.PROXY_REQUEST, LITELLM_PROXY_REQUEST_SPAN_NAME)
+    with trace.use_span(server, end_on_exit=False):
+        logger.log_pre_api_call(model="gpt-4o", messages=[], kwargs=kwargs)
+    asyncio.run(
+        logger.async_log_success_event(
+            {**kwargs, "standard_logging_object": None, "litellm_call_id": payload["litellm_call_id"]},
+            None,
+            None,
+            None,
+        )
+    )
+    asyncio.run(logger.async_log_failure_event(kwargs, None, None, None))
+    server.end()
+    llm_spans = [s for s in exporter.get_finished_spans() if s.name.startswith("chat")]
+    assert len(llm_spans) == 1

tests/test_litellm/integrations/otel/test_otel_v2_presets.py

@@ -160,3 +160,55 @@ def test_agentops_endpoint_points_at_live_host():
     # so a typo or stale domain can never ship again.
     assert _AGENTOPS_ENDPOINT == "https://otlp.agentops.ai/v1/traces"
     assert "agentops.cloud" not in _AGENTOPS_ENDPOINT
+
+
+def test_newrelic_preset_reads_env_license_key(monkeypatch):
+    monkeypatch.setenv("NEW_RELIC_LICENSE_KEY", "env-license-key")
+    from litellm.integrations.otel.model.config import ExporterOwner
+    from litellm.integrations.otel.presets.newrelic import newrelic_preset
+
+    cfg = newrelic_preset()
+    spec = next(e for e in cfg.exporters if e.owner == ExporterOwner.NEWRELIC)
+    assert spec.kind == "otlp_http"
+    assert spec.endpoint == "https://otlp.nr-data.net"
+    assert spec.headers == "api-key=env-license-key"
+    assert spec.requires_headers is True
+    assert "genai" in cfg.mapper_names
+
+
+def test_newrelic_preset_without_key_still_contributes_owned_spec(monkeypatch):
+    # The owned spec is the stamping target for per-team credentials, so it must
+    # exist even with no operator env key; requires_headers keeps the keyless
+    # copy from ever exporting.
+    monkeypatch.delenv("NEW_RELIC_LICENSE_KEY", raising=False)
+    from litellm.integrations.otel.model.config import ExporterOwner
+    from litellm.integrations.otel.presets.newrelic import newrelic_preset
+
+    cfg = newrelic_preset()
+    spec = next(e for e in cfg.exporters if e.owner == ExporterOwner.NEWRELIC)
+    assert spec.headers is None
+    assert spec.requires_headers is True
+
+
+def test_newrelic_preset_operator_region_and_content_knob(monkeypatch):
+    monkeypatch.setenv("NEW_RELIC_LICENSE_KEY", "env-license-key")
+    monkeypatch.setenv("NEW_RELIC_REGION", "EU")
+    monkeypatch.setenv("NEW_RELIC_AI_MONITORING_RECORD_CONTENT_ENABLED", "true")
+    from litellm.integrations.otel.model.config import ExporterOwner
+    from litellm.integrations.otel.presets.newrelic import newrelic_preset
+
+    cfg = newrelic_preset()
+    spec = next(e for e in cfg.exporters if e.owner == ExporterOwner.NEWRELIC)
+    assert spec.endpoint == "https://otlp.eu01.nr-data.net"
+    assert cfg.capture_span_content is True
+
+    monkeypatch.setenv("NEW_RELIC_AI_MONITORING_RECORD_CONTENT_ENABLED", "false")
+    assert newrelic_preset().capture_span_content is False
+
+
+def test_newrelic_preset_unset_content_knob_keeps_default(monkeypatch):
+    monkeypatch.delenv("NEW_RELIC_AI_MONITORING_RECORD_CONTENT_ENABLED", raising=False)
+    monkeypatch.delenv("OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT", raising=False)
+    from litellm.integrations.otel.presets.newrelic import newrelic_preset
+
+    assert newrelic_preset().capture_span_content is False

tests/test_litellm/litellm_core_utils/test_initialize_dynamic_callback_params.py

@@ -187,3 +187,49 @@ def test_empty_kwargs_returns_empty_params():
 
     params = initialize_standard_callback_dynamic_params({})
     assert dict(params) == {}
+
+
+def test_newrelic_callback_params_are_not_extracted_from_request_kwargs():
+    kwargs = {
+        "newrelic_api_key": "caller-key",
+        "metadata": {"newrelic_api_key": "caller-key-2", "newrelic_region": "eu"},
+        "litellm_params": {"metadata": {"newrelic_region": "eu"}},
+    }
+
+    params = initialize_standard_callback_dynamic_params(kwargs)
+
+    assert params.get("newrelic_api_key") is None
+    assert params.get("newrelic_region") is None
+
+
+def test_newrelic_trusted_vars_overlay_reaches_standard_params():
+    from litellm.types.utils import TRUSTED_CALLBACK_VARS_FIELD
+
+    kwargs = {
+        # A caller-supplied copy must lose to the proxy-stamped trusted value.
+        "newrelic_api_key": "caller-key",
+        TRUSTED_CALLBACK_VARS_FIELD: {
+            "newrelic_api_key": "team-key",
+            "newrelic_region": "eu",
+            # Non-overlay trusted vars must not be copied by the overlay.
+            "langfuse_public_key": "pk-team",
+        },
+    }
+
+    params = initialize_standard_callback_dynamic_params(kwargs)
+
+    assert params.get("newrelic_api_key") == "team-key"
+    assert params.get("newrelic_region") == "eu"
+    assert params.get("langfuse_public_key") is None
+
+
+def test_trusted_vars_overlay_uses_shared_parser_semantics():
+    # The overlay rides get_trusted_callback_params, the same parser the
+    # datadog handler consumes, so values are str()-coerced identically.
+    from litellm.types.utils import TRUSTED_CALLBACK_VARS_FIELD
+
+    params = initialize_standard_callback_dynamic_params(
+        {TRUSTED_CALLBACK_VARS_FIELD: {"newrelic_api_key": 12345}}
+    )
+
+    assert params.get("newrelic_api_key") == "12345"

tests/test_litellm/litellm_core_utils/test_litellm_logging.py

@@ -5200,3 +5200,81 @@ def test_prompt_hooks_skip_prompt_managers_when_no_prompt_id(logging_obj, tmp_pa
             )
         for hook in [cb for cb in litellm.callbacks if isinstance(cb, VectorStorePreCallHook)]:
             litellm.logging_callback_manager.remove_callback_from_list_by_object(litellm.callbacks, hook)
+def test_newrelic_dispatch_prefers_otel_v2_when_flag_on(monkeypatch):
+    """With LITELLM_OTEL_V2 on, the "newrelic" callback builds the OTel v2
+    logger (per-team credential routing); with the flag off (default) it keeps
+    the legacy agent-based logger, so existing deployments are untouched."""
+    from litellm.integrations.otel.logger import OpenTelemetryV2
+    from litellm.integrations.otel.model.config import is_otel_v2_enabled
+    from litellm.litellm_core_utils import litellm_logging as logging_module
+
+    logging_module._in_memory_loggers.clear()
+    monkeypatch.setenv("LITELLM_OTEL_V2", "true")
+    is_otel_v2_enabled.cache_clear()
+    try:
+        v2_logger = logging_module._init_custom_logger_compatible_class(
+            logging_integration="newrelic",
+            internal_usage_cache=None,
+            llm_router=None,
+            custom_logger_init_args={},
+        )
+        assert isinstance(v2_logger, OpenTelemetryV2)
+        assert v2_logger.callback_name == "newrelic"
+        # Same name resolves to the same instance, not a second logger.
+        again = logging_module._init_custom_logger_compatible_class(
+            logging_integration="newrelic",
+            internal_usage_cache=None,
+            llm_router=None,
+            custom_logger_init_args={},
+        )
+        assert again is v2_logger
+    finally:
+        logging_module._in_memory_loggers.clear()
+        monkeypatch.delenv("LITELLM_OTEL_V2", raising=False)
+        is_otel_v2_enabled.cache_clear()
+
+
+def test_newrelic_dispatch_keeps_legacy_agent_when_flag_off(monkeypatch):
+    from litellm.integrations.newrelic import NewRelicLogger
+    from litellm.integrations.otel.model.config import is_otel_v2_enabled
+    from litellm.litellm_core_utils import litellm_logging as logging_module
+
+    logging_module._in_memory_loggers.clear()
+    monkeypatch.delenv("LITELLM_OTEL_V2", raising=False)
+    is_otel_v2_enabled.cache_clear()
+    try:
+        legacy = logging_module._init_custom_logger_compatible_class(
+            logging_integration="newrelic",
+            internal_usage_cache=None,
+            llm_router=None,
+            custom_logger_init_args={},
+        )
+        assert isinstance(legacy, NewRelicLogger)
+    finally:
+        logging_module._in_memory_loggers.clear()
+        is_otel_v2_enabled.cache_clear()
+
+
+def test_get_custom_logger_compatible_class_finds_v2_newrelic(monkeypatch):
+    """Under LITELLM_OTEL_V2 the "newrelic" instance is an OpenTelemetryV2; the
+    cached-lookup must find it or hook resolution (post-call failure/success
+    hooks) silently skips the callback."""
+    from litellm.integrations.otel.model.config import is_otel_v2_enabled
+    from litellm.litellm_core_utils import litellm_logging as logging_module
+
+    logging_module._in_memory_loggers.clear()
+    monkeypatch.setenv("LITELLM_OTEL_V2", "true")
+    is_otel_v2_enabled.cache_clear()
+    try:
+        created = logging_module._init_custom_logger_compatible_class(
+            logging_integration="newrelic",
+            internal_usage_cache=None,
+            llm_router=None,
+            custom_logger_init_args={},
+        )
+        found = logging_module.get_custom_logger_compatible_class("newrelic")
+        assert found is created
+    finally:
+        logging_module._in_memory_loggers.clear()
+        monkeypatch.delenv("LITELLM_OTEL_V2", raising=False)
+        is_otel_v2_enabled.cache_clear()

tests/test_litellm/proxy/management_endpoints/test_callback_management_endpoints.py

@@ -264,3 +264,148 @@ class TestCallbackManagementEndpoints:
         assert galileo_config["displayName"] == "Galileo"
         assert "GALILEO_API_KEY" in galileo_config["dynamic_params"]
         assert "GALILEO_PROJECT_ID" in galileo_config["dynamic_params"]
+
+
+class TestNewRelicCallbackConfig:
+    def test_newrelic_entry_supports_team_logging_with_dynamic_params(self):
+        client = TestClient(app)
+        response = client.get("/callbacks/configs", headers={"Authorization": "Bearer sk-1234"})
+        assert response.status_code == 200
+        newrelic = next(
+            (config for config in response.json() if config.get("id") == "newrelic"),
+            None,
+        )
+        assert newrelic is not None
+        assert newrelic["supports_key_team_logging"] is True
+        params = newrelic["dynamic_params"]
+        assert params["newrelic_api_key"]["type"] == "password"
+        assert "newrelic_region" in params
+        # The operator-only agent env flag must not appear as a team-configurable
+        # field: it is not a StandardCallbackDynamicParams key and would be rejected.
+        assert "NEW_RELIC_AI_MONITORING_RECORD_CONTENT_ENABLED" not in params
+
+
+class TestNewRelicTeamCallbackValidation:
+    def _data(self, callback_vars):
+        from litellm.proxy._types import AddTeamCallback
+
+        return AddTeamCallback(callback_name="newrelic", callback_type="success", callback_vars=callback_vars)
+
+    def test_rejects_when_otel_v2_off(self, monkeypatch):
+        from fastapi import HTTPException
+
+        from litellm.integrations.otel.model.config import is_otel_v2_enabled
+        from litellm.proxy.management_endpoints.team_callback_endpoints import _validate_team_callback
+
+        monkeypatch.delenv("LITELLM_OTEL_V2", raising=False)
+        is_otel_v2_enabled.cache_clear()
+        try:
+            with pytest.raises(HTTPException) as exc:
+                _validate_team_callback(self._data({"newrelic_api_key": "k"}))
+            assert "LITELLM_OTEL_V2" in str(exc.value.detail)
+        finally:
+            is_otel_v2_enabled.cache_clear()
+
+    def test_rejects_unknown_region_and_region_without_key(self, monkeypatch):
+        from fastapi import HTTPException
+
+        from litellm.integrations.otel.model.config import is_otel_v2_enabled
+        from litellm.proxy.management_endpoints.team_callback_endpoints import _validate_team_callback
+
+        monkeypatch.setenv("LITELLM_OTEL_V2", "true")
+        is_otel_v2_enabled.cache_clear()
+        try:
+            with pytest.raises(HTTPException) as exc:
+                _validate_team_callback(self._data({"newrelic_api_key": "k", "newrelic_region": "mars"}))
+            assert "Unknown newrelic_region" in str(exc.value.detail)
+            with pytest.raises(HTTPException) as exc:
+                _validate_team_callback(self._data({"newrelic_region": "eu"}))
+            assert "requires newrelic_api_key" in str(exc.value.detail)
+            _validate_team_callback(self._data({"newrelic_api_key": "k", "newrelic_region": "EU"}))
+            # A JSON-null key is str()-coerced to "None" upstream; it must not
+            # slip past the region-requires-key guard.
+            with pytest.raises(HTTPException) as exc:
+                _validate_team_callback(self._data({"newrelic_api_key": None, "newrelic_region": "eu"}))
+            assert "requires newrelic_api_key" in str(exc.value.detail)
+        finally:
+            is_otel_v2_enabled.cache_clear()
+
+    def test_ignores_other_callbacks_and_bare_newrelic(self, monkeypatch):
+        from litellm.integrations.otel.model.config import is_otel_v2_enabled
+        from litellm.proxy._types import AddTeamCallback
+        from litellm.proxy.management_endpoints.team_callback_endpoints import _validate_team_callback
+
+        monkeypatch.delenv("LITELLM_OTEL_V2", raising=False)
+        is_otel_v2_enabled.cache_clear()
+        try:
+            assert _validate_team_callback(self._data({})) is None
+            assert (
+                _validate_team_callback(
+                    AddTeamCallback(
+                        callback_name="langfuse",
+                        callback_type="success",
+                        callback_vars={"langfuse_public_key": "pk", "langfuse_secret_key": "sk"},
+                    )
+                )
+                is None
+            )
+        finally:
+            is_otel_v2_enabled.cache_clear()
+
+
+class TestNewRelicKeyLoggingValidation:
+    """Key-level logging is written through key metadata, not /team/callback."""
+
+    def _metadata(self, callback_vars):
+        return {"logging": [{"callback_name": "newrelic", "callback_type": "success", "callback_vars": callback_vars}]}
+
+    def test_rejects_same_configs_as_team_endpoint(self, monkeypatch):
+        from fastapi import HTTPException
+
+        from litellm.integrations.otel.model.config import is_otel_v2_enabled
+        from litellm.proxy.management_endpoints.key_management_endpoints import (
+            raise_on_invalid_key_logging_config,
+        )
+
+        monkeypatch.delenv("LITELLM_OTEL_V2", raising=False)
+        is_otel_v2_enabled.cache_clear()
+        try:
+            with pytest.raises(HTTPException) as exc:
+                raise_on_invalid_key_logging_config(self._metadata({"newrelic_api_key": "k"}))
+            assert "LITELLM_OTEL_V2" in str(exc.value.detail)
+
+            monkeypatch.setenv("LITELLM_OTEL_V2", "true")
+            is_otel_v2_enabled.cache_clear()
+            with pytest.raises(HTTPException) as exc:
+                raise_on_invalid_key_logging_config(
+                    self._metadata({"newrelic_api_key": "k", "newrelic_region": "mars"})
+                )
+            assert "Unknown newrelic_region" in str(exc.value.detail)
+            with pytest.raises(HTTPException) as exc:
+                raise_on_invalid_key_logging_config(self._metadata({"newrelic_region": "eu"}))
+            assert "requires newrelic_api_key" in str(exc.value.detail)
+            raise_on_invalid_key_logging_config(self._metadata({"newrelic_api_key": "k", "newrelic_region": "EU"}))
+        finally:
+            is_otel_v2_enabled.cache_clear()
+
+    def test_ignores_metadata_without_newrelic_logging(self, monkeypatch):
+        from litellm.integrations.otel.model.config import is_otel_v2_enabled
+        from litellm.proxy.management_endpoints.key_management_endpoints import (
+            raise_on_invalid_key_logging_config,
+        )
+
+        monkeypatch.delenv("LITELLM_OTEL_V2", raising=False)
+        is_otel_v2_enabled.cache_clear()
+        try:
+            assert raise_on_invalid_key_logging_config(None) is None
+            assert raise_on_invalid_key_logging_config({"logging": "not-a-list"}) is None
+            assert raise_on_invalid_key_logging_config({"tags": ["a"]}) is None
+            assert raise_on_invalid_key_logging_config(self._metadata({})) is None
+            assert (
+                raise_on_invalid_key_logging_config(
+                    {"logging": [{"callback_name": "langfuse", "callback_vars": {"langfuse_public_key": "pk"}}]}
+                )
+                is None
+            )
+        finally:
+            is_otel_v2_enabled.cache_clear()

tests/test_litellm/proxy/test_litellm_pre_call_utils.py

@@ -7329,3 +7329,92 @@ def test_vertex_sends_exactly_one_authorization_header():
     vertex_request_headers.update(forwarded)
 
     assert _authorization_values(vertex_request_headers) == [GOOGLE_ACCESS_TOKEN]
+@pytest.mark.asyncio
+async def test_newrelic_team_callback_vars_reach_trusted_field():
+    """A key with a newrelic team callback stamps its vars into the proxy-owned
+    trusted field, and a caller-supplied newrelic_api_key in the body is
+    stripped rather than merged."""
+    key_with_newrelic_callback = UserAPIKeyAuth(
+        api_key="hashed-key",
+        metadata={
+            "logging": [
+                {
+                    "callback_name": "newrelic",
+                    "callback_type": "success",
+                    "callback_vars": {"newrelic_api_key": "team-nr-key", "newrelic_region": "eu"},
+                }
+            ]
+        },
+    )
+    data = {
+        "model": "gpt-3.5-turbo",
+        "messages": [{"role": "user", "content": "hello"}],
+        "newrelic_api_key": "attacker-key",
+    }
+
+    updated = await add_litellm_data_to_request(
+        data=data,
+        request=_callback_credential_request_mock(),
+        user_api_key_dict=key_with_newrelic_callback,
+        proxy_config=MagicMock(),
+        general_settings={},
+        version="test-version",
+    )
+
+    assert updated[TRUSTED_CALLBACK_VARS_FIELD] == {
+        "newrelic_api_key": "team-nr-key",
+        "newrelic_region": "eu",
+    }
+    assert updated["success_callback"] == ["newrelic"]
+
+    from litellm.litellm_core_utils.initialize_dynamic_callback_params import (
+        initialize_standard_callback_dynamic_params,
+    )
+
+    params = initialize_standard_callback_dynamic_params(updated)
+    assert params.get("newrelic_api_key") == "team-nr-key"
+    assert params.get("newrelic_region") == "eu"
+
+    from litellm.integrations.otel.presets import dynamic_otlp_endpoint, dynamic_otlp_headers
+
+    assert dynamic_otlp_headers("newrelic", params) == {"api-key": "team-nr-key"}
+    assert dynamic_otlp_endpoint("newrelic", params) == "https://otlp.eu01.nr-data.net"
+
+    from litellm.utils import get_non_default_completion_params
+
+    forwarded = get_non_default_completion_params(updated)
+    assert not any(param.startswith("newrelic_") for param in forwarded)
+    assert TRUSTED_CALLBACK_VARS_FIELD not in forwarded
+
+
+def test_newrelic_vars_scoped_to_newrelic_callback_entry():
+    """New Relic routing reads these vars from the trusted overlay with no
+    callback-name check, so a team that puts newrelic_* under a different
+    callback's vars must not have them enter the shared bag (and so never
+    exports to New Relic). Vars under a real newrelic entry are kept."""
+    from litellm.proxy._types import AddTeamCallback
+    from litellm.proxy.litellm_pre_call_utils import convert_key_logging_metadata_to_callback
+
+    smuggled = convert_key_logging_metadata_to_callback(
+        AddTeamCallback(
+            callback_name="langfuse",
+            callback_type="success",
+            callback_vars={
+                "langfuse_public_key": "pk",
+                "newrelic_api_key": "SMUGGLED",
+                "newrelic_region": "eu",
+            },
+        ),
+        None,
+    )
+    assert smuggled.callback_vars == {"langfuse_public_key": "pk"}
+
+    legit = convert_key_logging_metadata_to_callback(
+        AddTeamCallback(
+            callback_name="newrelic",
+            callback_type="success",
+            callback_vars={"newrelic_api_key": "REAL", "newrelic_region": "us"},
+        ),
+        None,
+    )
+    assert legit.callback_vars == {"newrelic_api_key": "REAL", "newrelic_region": "us"}

tests/test_litellm/test_check_test_quality.py

@@ -8,9 +8,13 @@ in the test body.
 """
 
 import importlib.util
+import os
+import subprocess
 import sys
 from pathlib import Path
 
+import pytest
+
 _REPO_ROOT = Path(__file__).resolve().parents[2]
 _MODULE_PATH = _REPO_ROOT / "scripts" / "check_test_quality.py"
 _spec = importlib.util.spec_from_file_location("check_test_quality", _MODULE_PATH)
@@ -183,7 +187,7 @@ def test_mock_echo_is_flagged(tmp_path):
         "        run()\n"
         "    mock_completion.assert_called_once()\n"
     )
-    assert _codes(tmp_path, source) == ["TQ002"]
+    assert _codes(tmp_path, source) == ["TQ002", "TQ008"]
 
 
 def test_call_args_inspection_is_mock_echo(tmp_path):
@@ -196,7 +200,7 @@ def test_call_args_inspection_is_mock_echo(tmp_path):
         "        run()\n"
         "    assert mock_completion.call_args[1]['model'] == 'gpt-4o'\n"
     )
-    assert _codes(tmp_path, source) == ["TQ002"]
+    assert _codes(tmp_path, source) == ["TQ002", "TQ008"]
 
 
 def test_patch_decorator_counts_as_installing_a_patch(tmp_path):
@@ -209,7 +213,7 @@ def test_patch_decorator_counts_as_installing_a_patch(tmp_path):
         "    run()\n"
         "    mock_completion.assert_called_once()\n"
     )
-    assert _codes(tmp_path, source) == ["TQ002"]
+    assert _codes(tmp_path, source) == ["TQ002", "TQ008"]
 
 
 def test_patching_but_asserting_the_output_is_not_mock_echo(tmp_path):
@@ -223,7 +227,7 @@ def test_patching_but_asserting_the_output_is_not_mock_echo(tmp_path):
         "    mock_completion.assert_called_once()\n"
         "    assert result.choices[0].message.content == 'pong'\n"
     )
-    assert _codes(tmp_path, source) == []
+    assert _codes(tmp_path, source) == ["TQ008"]
 
 
 def test_asserting_without_patching_is_not_mock_echo(tmp_path):
@@ -240,7 +244,7 @@ def test_a_test_with_no_assertions_is_tq001_not_tq002(tmp_path):
         "    with patch('litellm.completion'):\n"
         "        run()\n"
     )
-    assert _codes(tmp_path, source) == ["TQ001"]
+    assert _codes(tmp_path, source) == ["TQ001", "TQ008"]
 
 
 def test_sys_path_insert_is_flagged(tmp_path):
@@ -548,3 +552,188 @@ def test_the_read_may_sit_a_statement_above_the_store(tmp_path):
 def test_a_loop_storing_under_a_key_that_is_not_the_loop_variable_is_not_an_inventory(tmp_path):
     source = _HELPER_DICT_CONFTEST.replace("state[attr] =", 'state["fixed"] =')
     assert [v.code for v in checker.check_file(_written(tmp_path, source))] == []
+
+
+def test_patching_an_sdk_function_by_string_is_flagged(tmp_path):
+    source = 'from unittest.mock import patch\n\n\n@patch("litellm.completion")\ndef test_x(m):\n    assert m\n'
+    assert "TQ008" in _codes(tmp_path, source)
+
+
+def test_patching_a_deep_sdk_path_is_flagged(tmp_path):
+    source = (
+        "from unittest.mock import patch\n\n\n"
+        "def test_x():\n"
+        '    with patch("litellm.llms.openai.chat.handler.OpenAIChatCompletion.completion"):\n'
+        "        assert True\n"
+    )
+    assert "TQ008" in _codes(tmp_path, source)
+
+
+def test_patch_object_rooted_at_the_sdk_is_flagged(tmp_path):
+    source = (
+        "import litellm\nfrom unittest.mock import patch\n\n\n"
+        "def test_x():\n"
+        '    with patch.object(litellm, "api_key", "x"):\n'
+        "        assert True\n"
+    )
+    assert "TQ008" in _codes(tmp_path, source)
+
+
+def test_patch_object_on_a_from_imported_sdk_module_is_flagged(tmp_path):
+    source = (
+        "from litellm.llms.openai.chat import handler\nfrom unittest.mock import patch\n\n\n"
+        "def test_x():\n"
+        '    with patch.object(handler.OpenAIChatCompletion, "completion"):\n'
+        "        assert True\n"
+    )
+    assert "TQ008" in _codes(tmp_path, source)
+
+
+def test_patch_object_on_an_aliased_sdk_module_is_flagged(tmp_path):
+    source = (
+        "import litellm.llms.openai.chat.handler as oai\nfrom unittest.mock import patch\n\n\n"
+        "def test_x():\n"
+        '    with patch.object(oai.OpenAIChatCompletion, "completion"):\n'
+        "        assert True\n"
+    )
+    assert "TQ008" in _codes(tmp_path, source)
+
+
+def test_patch_object_on_a_renamed_sdk_symbol_is_flagged(tmp_path):
+    source = (
+        "from litellm.utils import get_llm_provider as glp\nfrom unittest.mock import patch\n\n\n"
+        "def test_x():\n"
+        '    with patch.object(glp, "__wrapped__"):\n'
+        "        assert True\n"
+    )
+    assert "TQ008" in _codes(tmp_path, source)
+
+
+def test_the_reported_target_is_the_resolved_sdk_path(tmp_path):
+    source = (
+        "from litellm.llms.openai.chat import handler\nfrom unittest.mock import patch\n\n\n"
+        "def test_x():\n"
+        '    with patch.object(handler.OpenAIChatCompletion, "completion"):\n'
+        "        assert True\n"
+    )
+    reported = [v.message for v in checker.check_file(_written(tmp_path, source)) if v.code == "TQ008"]
+    assert reported
+    assert "litellm.llms.openai.chat.handler.OpenAIChatCompletion" in reported[0]
+
+
+def test_patch_object_on_a_from_imported_third_party_is_not_flagged(tmp_path):
+    source = (
+        "from openai import OpenAI\nfrom unittest.mock import patch\n\n\n"
+        "def test_x():\n"
+        '    with patch.object(OpenAI, "chat"):\n'
+        "        assert True\n"
+    )
+    assert "TQ008" not in _codes(tmp_path, source)
+
+
+def test_a_local_name_with_no_sdk_import_behind_it_is_not_flagged(tmp_path):
+    source = (
+        "from unittest.mock import patch\n\n\n"
+        "def test_x(handler):\n"
+        '    with patch.object(handler, "completion"):\n'
+        "        assert True\n"
+    )
+    assert "TQ008" not in _codes(tmp_path, source)
+
+
+def test_mocking_a_third_party_client_is_not_flagged(tmp_path):
+    source = (
+        "from unittest.mock import patch\n\n\n"
+        "def test_x():\n"
+        '    with patch("openai.OpenAI.chat"):\n'
+        "        assert True\n"
+    )
+    assert "TQ008" not in _codes(tmp_path, source)
+
+
+def test_mocking_the_http_transport_is_not_flagged(tmp_path):
+    source = (
+        "from unittest.mock import patch\n\n\n"
+        "def test_x():\n"
+        '    with patch("httpx.AsyncClient.send"):\n'
+        "        assert True\n"
+    )
+    assert "TQ008" not in _codes(tmp_path, source)
+
+
+def test_a_name_merely_starting_with_litellm_is_not_the_sdk(tmp_path):
+    source = (
+        "from unittest.mock import patch\n\n\n"
+        "def test_x():\n"
+        '    with patch("litellm_enterprise.thing.go"):\n'
+        "        assert True\n"
+    )
+    assert "TQ008" not in _codes(tmp_path, source)
+
+
+def test_an_sdk_patch_can_be_suppressed(tmp_path):
+    source = (
+        "from unittest.mock import patch\n\n\n"
+        "def test_x():\n"
+        '    with patch("litellm.completion"):  # test-quality-ok: pinning the router seam\n'
+        "        assert True\n"
+    )
+    assert "TQ008" not in _codes(tmp_path, source)
+
+
+_FANS_OUT = checker._worker_count(checker.PARALLEL_MIN_PATHS) > 1
+_SERIAL_ONLY = "one usable core, so scan_paths stays serial and there is no fan-out to compare"
+
+
+def _corpus(tmp_path: Path, count: int) -> tuple[Path, ...]:
+    for index in range(count):
+        (tmp_path / f"test_gen_{index}.py").write_text(
+            f"def test_flagged_{index}():\n    compute()\n\n\ndef test_clean_{index}():\n    assert compute() == {index}\n",
+            encoding="utf-8",
+        )
+    return tuple(sorted(tmp_path.rglob("*.py")))
+
+
+def _run_checker(target: Path) -> list[str]:
+    completed = subprocess.run(
+        [sys.executable, str(_MODULE_PATH), str(target)],
+        capture_output=True, text=True, timeout=300,
+    )
+    return completed.stdout.splitlines()
+
+
+def test_worker_count_stays_serial_below_the_threshold():
+    assert checker._worker_count(checker.PARALLEL_MIN_PATHS - 1) == 1
+
+
+def test_worker_count_fans_out_at_the_threshold():
+    assert checker._worker_count(checker.PARALLEL_MIN_PATHS) == max(
+        1, min(os.cpu_count() or 1, checker.MAX_WORKERS)
+    )
+
+
+def test_worker_count_never_exceeds_the_cap():
+    assert checker._worker_count(100_000) <= checker.MAX_WORKERS
+
+
+def test_scan_paths_below_the_threshold_returns_every_violation(tmp_path):
+    paths = _corpus(tmp_path, 3)
+    assert checker._worker_count(len(paths)) == 1
+    assert [v.code for v in checker.scan_paths(paths)] == ["TQ001"] * 3
+
+
+@pytest.mark.skipif(not _FANS_OUT, reason=_SERIAL_ONLY)
+def test_a_fanned_out_run_reports_exactly_what_a_serial_run_reports(tmp_path):
+    paths = _corpus(tmp_path, checker.PARALLEL_MIN_PATHS + 5)
+    serial = [v.render() for v in sorted(v for path in paths for v in checker.check_file(path))]
+    assert serial, "corpus must produce violations or the comparison proves nothing"
+    assert _run_checker(tmp_path) == serial
+
+
+@pytest.mark.skipif(not _FANS_OUT, reason=_SERIAL_ONLY)
+def test_a_fanned_out_run_reports_each_generated_file_exactly_once(tmp_path):
+    paths = _corpus(tmp_path, checker.PARALLEL_MIN_PATHS + 5)
+    reported = _run_checker(tmp_path)
+    assert len(reported) == len(paths)
+    assert len({line.split(":")[0] for line in reported}) == len(paths)
+    assert all(" TQ001 " in line for line in reported)

tests/test_litellm/test_check_type_discipline.py

@@ -8,10 +8,14 @@ a test fail. The comment-scanner cases are the regression for the readline path:
 
 import importlib.util
 import json
+import os
 import re
+import subprocess
 import sys
 from pathlib import Path
 
+import pytest
+
 _REPO_ROOT = Path(__file__).resolve().parents[2]
 _MODULE_PATH = _REPO_ROOT / "scripts" / "check_type_discipline.py"
 _spec = importlib.util.spec_from_file_location("check_type_discipline", _MODULE_PATH)
@@ -695,3 +699,61 @@ def test_budget_covers_exactly_the_checker_rules():
     for spec in budget.values():
         assert isinstance(spec["limit"], int)
         assert spec["limit"] >= 0
+
+
+_FANS_OUT = checker._worker_count(checker.PARALLEL_MIN_PATHS) > 1
+_SERIAL_ONLY = "one usable core, so scan_paths stays serial and there is no fan-out to compare"
+
+
+def _corpus(tmp_path: Path, count: int) -> tuple[Path, ...]:
+    for index in range(count):
+        (tmp_path / f"mod_{index}.py").write_text(
+            f"def build_{index}(items: list[int]) -> None:\n    return None\n",
+            encoding="utf-8",
+        )
+    return tuple(sorted(tmp_path.rglob("*.py")))
+
+
+def _run_checker(target: Path) -> list[str]:
+    completed = subprocess.run(
+        [sys.executable, str(_MODULE_PATH), str(target)],
+        capture_output=True, text=True, timeout=300,
+    )
+    return completed.stdout.splitlines()
+
+
+def test_worker_count_stays_serial_below_the_threshold():
+    assert checker._worker_count(checker.PARALLEL_MIN_PATHS - 1) == 1
+
+
+def test_worker_count_fans_out_at_the_threshold():
+    assert checker._worker_count(checker.PARALLEL_MIN_PATHS) == max(
+        1, min(os.cpu_count() or 1, checker.MAX_WORKERS)
+    )
+
+
+def test_worker_count_never_exceeds_the_cap():
+    assert checker._worker_count(100_000) <= checker.MAX_WORKERS
+
+
+def test_scan_paths_below_the_threshold_returns_every_violation(tmp_path):
+    paths = _corpus(tmp_path, 3)
+    assert checker._worker_count(len(paths)) == 1
+    found = checker.scan_paths(paths)
+    assert found and len({v.path for v in found}) == 3
+
+
+@pytest.mark.skipif(not _FANS_OUT, reason=_SERIAL_ONLY)
+def test_a_fanned_out_run_reports_exactly_what_a_serial_run_reports(tmp_path):
+    paths = _corpus(tmp_path, checker.PARALLEL_MIN_PATHS + 5)
+    serial = [v.render() for v in sorted(v for path in paths for v in checker.check_file(path))]
+    assert serial, "corpus must produce violations or the comparison proves nothing"
+    assert _run_checker(tmp_path) == serial
+
+
+@pytest.mark.skipif(not _FANS_OUT, reason=_SERIAL_ONLY)
+def test_a_fanned_out_run_reports_each_generated_file_exactly_once(tmp_path):
+    paths = _corpus(tmp_path, checker.PARALLEL_MIN_PATHS + 5)
+    reported = _run_checker(tmp_path)
+    assert reported
+    assert len({line.split(":")[0] for line in reported}) == len(paths)

tests/test_litellm/test_test_quality_gate.py

@@ -144,5 +144,5 @@ def test_the_shipped_budget_covers_every_rule_the_checker_can_emit():
     import json
 
     budget = json.loads((_REPO_ROOT / "test-quality-budget.json").read_text())
-    assert set(budget) == {"TQ001", "TQ002", "TQ003", "TQ004", "TQ005", "TQ006", "TQ007"}
+    assert set(budget) == {"TQ001", "TQ002", "TQ003", "TQ004", "TQ005", "TQ006", "TQ007", "TQ008"}
     assert all(spec["limit"] >= 0 for spec in budget.values())

ui/litellm-dashboard/src/components/callback_info_helpers.tsx

@@ -6,6 +6,7 @@ import galileoLogo from "../../public/assets/logos/galileo.ico";
 import lagoLogo from "../../public/assets/logos/lago.svg";
 import langfuseLogo from "../../public/assets/logos/langfuse.png";
 import langsmithLogo from "../../public/assets/logos/langsmith.png";
+import newrelicLogo from "../../public/assets/logos/newrelic.png";
 import openmeterLogo from "../../public/assets/logos/openmeter.png";
 import otelLogo from "../../public/assets/logos/otel.png";
 
@@ -77,6 +78,17 @@ export const CALLBACK_CONFIGS: CallbackConfig[] = [
     },
     description: "Datadog Logging Integration",
   },
+  {
+    id: "newrelic",
+    displayName: "New Relic",
+    logo: newrelicLogo.src,
+    supports_key_team_logging: true,
+    dynamic_params: {
+      newrelic_api_key: "password",
+      newrelic_region: "text",
+    },
+    description: "New Relic Logging Integration",
+  },
   {
     id: "lago",
     displayName: "Lago",