fix(websearch_interception): keep intercepted searches under the parent request's session and trace (#41711)

* fix(websearch_interception): propagate parent session/trace ids into intercepted searches

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

* fix(websearch_interception): let parent correlation win over configured search params and type test params

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

* test(e2e): bill an intercepted web search under the parent request session

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

* test(e2e): drop unrelated reformatting from the websearch session harness change

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

* ci(e2e): keep the websearch interception session suite out of the stage-mirror gate

The stage-mirror stack runs no websearch_interception callback or search tool, so the suite is deselected there and the changed-tests gate fails on a file that executed nothing

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

* ci(e2e): run the websearch interception session suite on the stage-mirror stack

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

* docs(e2e): leave CONTRIBUTING.md untouched

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

---------

Co-authored-by: shivam <shivam@berri.ai>
Co-authored-by: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

litellm/integrations/websearch_interception/handler.py

@@ -10,6 +10,8 @@ import asyncio
 import math
 import uuid
 from collections.abc import AsyncIterator, Mapping, Sequence
+from dataclasses import dataclass
+from types import MappingProxyType
 from typing import TYPE_CHECKING, Any, Final, Literal, TypedDict, TypeVar, cast
 
 from typing_extensions import Never, ReadOnly
@@ -196,6 +198,46 @@ class _AcompletionNamedParams(TypedDict, total=False):
 
 _NO_ACREATE_NAMED: Final[_AcreateNamedParams] = {}
 _NO_ASEARCH_NAMED: Final[_AsearchNamedParams] = {}
+
+
+def _as_str_mapping(value: object) -> Mapping[str, object] | None:
+    return value if isinstance(value, Mapping) else None  # pyright: ignore[reportUnknownVariableType]  # str-keyed request metadata is not narrowable from object
+
+
+@dataclass(frozen=True, slots=True)
+class _ParentRequestCorrelation:
+    """Correlation ids of the LLM request that triggered an intercepted search, so the search's
+    own spend log row and traces land under the same session/trace instead of a fresh one."""
+
+    session_id: str | None
+    trace_id: str | None
+    parent_request_id: str | None
+    parent_otel_span: object | None
+
+    def as_search_metadata(self) -> Mapping[str, object]:
+        return MappingProxyType(
+            {
+                key: value
+                for key, value in (
+                    ("session_id", self.session_id),
+                    ("trace_id", self.trace_id),
+                    ("parent_request_id", self.parent_request_id),
+                    ("litellm_parent_otel_span", self.parent_otel_span),
+                )
+                if value is not None
+            }
+        )
+
+    def as_search_kwargs(self) -> Mapping[str, str]:
+        return MappingProxyType(
+            {
+                key: value
+                for key, value in (("litellm_session_id", self.session_id), ("litellm_trace_id", self.trace_id))
+                if value is not None
+            }
+        )
+
+
 _NO_ACOMPLETION_NAMED: Final[_AcompletionNamedParams] = {}
 
 
@@ -1527,15 +1569,17 @@ class WebSearchInterceptionLogger(CustomLogger):
                 "WebSearchInterception: Executing search for '%s' using provider '%s'", query, search_provider
             )
             user_api_key_auth: Final = self._get_user_api_key_auth_from_kwargs(kwargs)
+            parent_correlation: Final = self._get_parent_request_correlation(kwargs, user_api_key_auth)
             search_metadata: Final = (
                 None
                 if user_api_key_auth is None
                 else self._build_search_request_metadata(
                     user_api_key_auth=user_api_key_auth,
                     search_tool_name=search_tool_name,
+                    parent_correlation=parent_correlation,
                 )
             )
-            search_kwargs: Final = {
+            configured_search_kwargs: Final = {
                 key: value
                 for key, value in search_litellm_params.items()
                 if key != "search_provider" and value is not None
@@ -1549,8 +1593,11 @@ class WebSearchInterceptionLogger(CustomLogger):
                 if rich_queries:
                     query_arg = rich_queries
                 rich_objective = rich.get("objective")
-                if rich_objective and "objective" not in search_kwargs:
-                    search_kwargs["objective"] = rich_objective
+                if rich_objective and "objective" not in configured_search_kwargs:
+                    configured_search_kwargs["objective"] = rich_objective
+            search_kwargs: Final = MappingProxyType(
+                {**configured_search_kwargs, **parent_correlation.as_search_kwargs()}
+            )
             result: Final = (
                 await litellm.asearch(
                     query=query_arg, search_provider=search_provider, **_NO_ASEARCH_NAMED, **search_kwargs
@@ -1624,6 +1671,7 @@ class WebSearchInterceptionLogger(CustomLogger):
     def _build_search_request_metadata(
         user_api_key_auth: "UserAPIKeyAuth",
         search_tool_name: str | None,
+        parent_correlation: _ParentRequestCorrelation,
     ) -> Mapping[str, object]:
         """
         Spend-tracking metadata for the intercepted search, so its provider cost is logged
@@ -1637,11 +1685,50 @@ class WebSearchInterceptionLogger(CustomLogger):
         )
         return {  # mutable-ok: litellm's metadata channel is a plain dict its logging path reads and enriches
             **user_api_key_metadata,
+            **parent_correlation.as_search_metadata(),
             "model_group": search_tool_name,
             "user_api_key": user_api_key_auth.api_key,
             "user_api_key_auth": user_api_key_auth,
         }
 
+    @staticmethod
+    def _get_parent_request_correlation(
+        kwargs: Mapping[str, object] | None,
+        user_api_key_auth: "UserAPIKeyAuth | None",
+    ) -> _ParentRequestCorrelation:
+        """Read the originating request's ids from the hook kwargs, which are either the raw call
+        kwargs (metadata/litellm_metadata at top level) or a logging payload (under litellm_params)."""
+        if not kwargs:
+            return _ParentRequestCorrelation(None, None, None, None)
+        litellm_params: Final = _as_str_mapping(kwargs.get("litellm_params"))
+        scopes: Final[tuple[Mapping[str, object], ...]] = (
+            (kwargs,) if litellm_params is None else (kwargs, litellm_params)
+        )
+        metadatas: Final[tuple[Mapping[str, object], ...]] = tuple(
+            metadata
+            for scope in scopes
+            for metadata_key in ("metadata", "litellm_metadata")
+            if (metadata := _as_str_mapping(scope.get(metadata_key))) is not None
+        )
+
+        def first_str(scope_key: str | None, metadata_key: str | None) -> str | None:
+            candidates: Final[tuple[object, ...]] = (
+                *(scope.get(scope_key) for scope in scopes if scope_key is not None),
+                *(metadata.get(metadata_key) for metadata in metadatas if metadata_key is not None),
+            )
+            return next((value for value in candidates if isinstance(value, str) and value), None)
+
+        parent_otel_span: Final[object | None] = next(
+            (span for metadata in metadatas if (span := metadata.get("litellm_parent_otel_span")) is not None),
+            None if user_api_key_auth is None else user_api_key_auth.parent_otel_span,
+        )
+        return _ParentRequestCorrelation(
+            session_id=first_str("litellm_session_id", "session_id"),
+            trace_id=first_str("litellm_trace_id", "trace_id"),
+            parent_request_id=first_str("litellm_call_id", None),
+            parent_otel_span=parent_otel_span,
+        )
+
     @staticmethod
     def _selected_search_tool_name(search_tool: Mapping[str, object] | None) -> str | None:
         if search_tool is None:

tests/e2e/AGENTS.md

@@ -210,6 +210,7 @@ quota_management.<behavior>.<variant>.<assertion>
               <spend_tracking> chat_completions | stream | messages_bridge | embeddings
                                | cache_hit | key_rollup | concurrent_burst | tags | end_user
                                | per_model | failure | spend_calculate | pagination | key_attribution
+                               | websearch_interception
   assertion : blocks_over_limit | resets_after_window | headers_report_remaining | picks_under_tpm
               | blocks_then_resets | resets_windows_independently | alerts_without_blocking
               | isolates_per_model | isolates_per_member | isolates_per_group | enforced_across_keys
@@ -217,7 +218,7 @@ quota_management.<behavior>.<variant>.<assertion>
               | matches_sum_of_logs | loses_no_spend | attributes_spend | writes_own_rows
               | writes_failure_row | returns_cost | keeps_total | joins_key | reports_alias_and_email
               | health_rows_keep_service_account | retrieve_batch_cost_joins_retrieving_key
-              | poller_batch_cost_joins_creating_key
+              | poller_batch_cost_joins_creating_key | bills_under_request_session
   e.g.  quota_management.ratelimit.rpm.blocks_over_limit           exercised_on=[chat_completions, messages]
         quota_management.budget.key.blocks_over_limit              exercised_on=[chat_completions]
 ```

tests/e2e/coverage_registry/quota_management.yaml

@@ -58,6 +58,7 @@
 - {id: quota_management.spend_tracking.service_tier.bills_tier_rates, module: quota_management, tier: P1, behavior: spend_tracking, variant: service_tier, assertions: [bills_tier_rates], exercised_on: [chat_completions], source: "cost_calculator.py", rationale: "A priority service_tier call bills input, output, and reasoning at the deployment's *_priority rates and records the tier on the row (#35923, #35925)"}
 - {id: quota_management.spend_tracking.cost_headers.additive_components, module: quota_management, tier: P1, behavior: spend_tracking, variant: cost_headers, assertions: [additive_components], exercised_on: [chat_completions], source: "proxy/common_request_processing.py", rationale: "The x-litellm-response-cost-* component headers sum to the total, input covers only fresh tokens, and reasoning stays a subset of output (#36965)"}
 - {id: quota_management.spend_tracking.passthrough_stream.injects_usage_cost, module: quota_management, tier: P1, behavior: spend_tracking, variant: passthrough_stream, assertions: [injects_usage_cost], exercised_on: [openai_passthrough], source: "proxy/pass_through_endpoints/streaming_handler.py", rationale: "With include_cost_in_streaming_usage on, the /openai passthrough's final streaming usage frame carries the proxy-computed cost (#36503). Uncovered: the flag is only settable in litellm_settings, and the shared e2e stack does not turn it on yet"}
+- {id: quota_management.spend_tracking.websearch_interception.bills_under_request_session, module: quota_management, tier: P1, behavior: spend_tracking, variant: websearch_interception, assertions: [bills_under_request_session], exercised_on: [messages], source: "integrations/websearch_interception/handler.py", fail_before_fix: proven, rationale: "A web_search server tool the proxy intercepts into litellm.asearch writes its own asearch spend row, and that row carries the parent request's session_id so the session view counts the search and its cost next to the turn that triggered it (LIT-8063)"}
 - {id: quota_management.spend_tracking.key_attribution.joins_key, module: quota_management, tier: P1, behavior: spend_tracking, variant: key_attribution, assertions: [joins_key], exercised_on: [chat_completions, messages, responses, embeddings, batches, files, google_native, rust_control_plane], source: "proxy/spend_tracking/spend_tracking_utils.py", rationale: "Every spend row a virtual key writes across chat, queued chat, messages, responses, embeddings, the Gemini passthrough, file upload, batch create, and a replayed callback log carries api_key equal to the key's token hash and the key alias, the join the usage APIs depend on; a re-hashed token shows up as an unattributed key-hash-* row (#39568, #39572)"}
 - {id: quota_management.spend_tracking.key_attribution.reports_alias_and_email, module: quota_management, tier: P1, behavior: spend_tracking, variant: key_attribution, assertions: [reports_alias_and_email], exercised_on: [chat_completions, messages, responses, embeddings, batches, files, google_native, rust_control_plane], source: "proxy/management_endpoints/internal_user_endpoints.py", rationale: "/spend/logs?api_key= returns every one of the key's rows with its alias and /user/daily/activity aggregates them under the key's token with key_alias and user_email; /spend/logs carries no email field, so the email is asserted on daily activity only"}
 - {id: quota_management.spend_tracking.key_attribution.health_rows_keep_service_account, module: quota_management, tier: P1, behavior: spend_tracking, variant: key_attribution, assertions: [health_rows_keep_service_account], exercised_on: [chat_completions], source: "proxy/health_check.py", rationale: "A /health probe's spend row stays keyed by the literal litellm-internal-health-check service account rather than a hash of it, so health spend never appears as an unattributed key"}

tests/e2e/e2e_http.py

@@ -49,6 +49,7 @@ class AnthropicHeaders(AuthHeaders):
     on its own internal calls."""
 
     anthropic_version: str = Field(default="2023-06-01", alias="anthropic-version")
+    x_litellm_session_id: str | None = Field(default=None, serialization_alias="x-litellm-session-id")
 
 
 class PartialBody(BaseModel):

tests/e2e/gateway/stage_mirror_ci_config.yml

@@ -32,7 +32,10 @@ litellm_settings:
       - host: 127.0.0.1
         port: 6379
     ssl: true
-  callbacks: ["arize_phoenix", "datadog", "smtp_email", "prometheus", "otel"]
+  callbacks: ["arize_phoenix", "datadog", "smtp_email", "prometheus", "otel", "websearch_interception"]
+  websearch_interception_params:
+    enabled_providers: ["bedrock"]
+    search_tool_name: e2e-search
   require_auth_for_metrics_endpoint: false
 
 router_settings:
@@ -65,6 +68,12 @@ model_list:
       model: openai/text-embedding-3-small
       api_key: os.environ/OPENAI_API_KEY
 
+search_tools:
+  - search_tool_name: e2e-search
+    litellm_params:
+      search_provider: perplexity
+      api_key: os.environ/PERPLEXITY_API_KEY
+
 files_settings:
   - custom_llm_provider: openai
     api_key: os.environ/OPENAI_API_KEY

tests/e2e/models.py

@@ -949,6 +949,7 @@ class SpendLogRow(BaseModel):
     completion_tokens: int | None = None
     total_tokens: int | None = None
     request_tags: list[str] | None = None
+    session_id: str | None = None
     metadata: SpendLogMetadata | None = None
     proxy_server_request: JsonValue = None
     response: JsonValue = None
@@ -985,6 +986,15 @@ class SpendLogsPageParams(BaseModel):
     api_key: str | None = None
 
 
+class SessionSpendLogsParams(BaseModel):
+    """Query for /spend/logs/session/ui, the session view the Admin UI logs page
+    opens: every row whose session_id equals the given one, newest first."""
+
+    session_id: str
+    page: int = 1
+    page_size: int = 100
+
+
 class SpendLogsPage(BaseModel):
     data: list[SpendLogRow] = []
     total: int

tests/e2e/proxy_client.py

@@ -87,6 +87,7 @@ from models import (
     RouterSettingsResponse,
     SearchToolCreateBody,
     SearchToolCreateResponse,
+    SessionSpendLogsParams,
     SpendLogRow,
     SpendLogs,
     SpendLogsPage,
@@ -1004,19 +1005,26 @@ class ProxyClient:
             response_type=CountTokensResponse,
         )
 
-    def messages(self, key: str, body: AnthropicMessagesBody) -> Result[AnthropicMessagesResponse]:
+    def messages(
+        self, key: str, body: AnthropicMessagesBody, *, session_id: str | None = None
+    ) -> Result[AnthropicMessagesResponse]:
         """POST /v1/messages (Anthropic-native). The response is either the
         Anthropic-shape passthrough (`content`) or the OpenAI-normalized shape
-        (`choices`); AnthropicMessagesResponse models both."""
+        (`choices`); AnthropicMessagesResponse models both. `session_id` goes out
+        as the `x-litellm-session-id` header, the way Claude Code sends it through
+        ANTHROPIC_CUSTOM_HEADERS, so every spend row the call produces shares it."""
         return self.transport.post(
             "/v1/messages",
-            headers=self._anthropic_headers(key),
+            headers=self._anthropic_headers(key, session_id=session_id),
             json=body,
             response_type=AnthropicMessagesResponse,
         )
 
-    def _anthropic_headers(self, key: str) -> AnthropicHeaders:
-        return AnthropicHeaders(authorization=self.transport.bearer(key).authorization)
+    def _anthropic_headers(self, key: str, *, session_id: str | None = None) -> AnthropicHeaders:
+        return AnthropicHeaders(
+            authorization=self.transport.bearer(key).authorization,
+            x_litellm_session_id=session_id,
+        )
 
     # ---- spend read-back ------------------------------------------------
 
@@ -1060,6 +1068,27 @@ class ProxyClient:
     ) -> list[SpendLogRow]:
         return self._poll(lambda: self.spend_logs(SpendLogsParams(api_key=key)), min_rows, predicate)
 
+    def session_spend_logs(self, session_id: str) -> list[SpendLogRow]:
+        """GET /spend/logs/session/ui, the per-session view the Admin UI logs page
+        opens when a session id is clicked."""
+        return unwrap(
+            self.transport.get(
+                "/spend/logs/session/ui",
+                headers=self.management_headers(),
+                params=SessionSpendLogsParams(session_id=session_id),
+                response_type=SpendLogsPage,
+            )
+        ).data
+
+    def poll_logs_for_session(
+        self,
+        session_id: str,
+        *,
+        min_rows: int = 1,
+        predicate: RowsPredicate | None = None,
+    ) -> list[SpendLogRow]:
+        return self._poll(lambda: self.session_spend_logs(session_id), min_rows, predicate)
+
     def poll_logs_for_request_id(
         self,
         request_id: str,

tests/e2e/quota_management/spend_tracking/test_websearch_interception_session_e2e.py

@@ -0,0 +1,102 @@
+"""Intercepted web searches are billed under the LLM request's session.
+
+The websearch_interception callback turns an Anthropic ``web_search`` server tool
+into a ``litellm.asearch`` call against a configured search tool, so each search is
+its own spend row (call_type ``asearch``) next to the ``anthropic_messages`` row for
+the turn that asked for it. Claude Code and the Admin UI group spend by
+``session_id``, so the search row has to carry the same session as the turn that
+triggered it; before the fix it landed under a session of its own and the session
+view under-counted both requests and spend (LIT-8063).
+
+Needs a proxy booted with the callback and a real search backend, which
+``gateway/stage_mirror_ci_config.yml`` carries as the ``e2e-search`` Perplexity tool.
+"""
+
+from typing import Final
+
+import pytest
+from e2e_config import unique_marker
+from e2e_http import unwrap
+from lifecycle import ResourceManager
+from models import (
+    AnthropicMessagesBody,
+    AnthropicWebSearchTool,
+    ChatMessage,
+    LiteLLMParamsBody,
+    SpendLogRow,
+)
+from proxy_client import ProxyClient
+
+pytestmark = pytest.mark.e2e
+
+BEDROCK_INVOKE_BACKEND: Final = "bedrock/invoke/us.anthropic.claude-haiku-4-5-20251001-v1:0"
+SEARCH_CALL_TYPE: Final = "asearch"
+
+
+def _has_search_row(rows: list[SpendLogRow]) -> bool:
+    return any(row.call_type == SEARCH_CALL_TYPE for row in rows)
+
+
+class TestWebSearchInterceptionSession:
+    @pytest.mark.covers(
+        "quota_management.spend_tracking.websearch_interception.bills_under_request_session",
+        exercised_on=("messages",),
+    )
+    def test_intercepted_search_is_billed_under_the_request_session(
+        self, proxy: ProxyClient, resources: ResourceManager
+    ) -> None:
+        """One /v1/messages turn that runs an intercepted web search must produce an
+        ``asearch`` spend row in the same session as its ``anthropic_messages`` row,
+        billed separately and with its own request id."""
+        marker: Final = unique_marker()
+        model: Final = f"e2e-websearch-session-{marker}"
+        model_id: Final = proxy.create_model(
+            model, LiteLLMParamsBody(model=BEDROCK_INVOKE_BACKEND, aws_region_name="us-east-1")
+        )
+        resources.defer(lambda: proxy.delete_model(model_id))
+        key: Final = resources.key(models=[model])
+        session_id: Final = f"e2e-websearch-session-{marker}"
+
+        response: Final = unwrap(
+            proxy.messages(
+                key,
+                AnthropicMessagesBody(
+                    model=model,
+                    max_tokens=512,
+                    tools=[AnthropicWebSearchTool(type="web_search_20250305", name="web_search", max_uses=1)],
+                    messages=[
+                        ChatMessage(
+                            role="user",
+                            content=f"Use web search to find one recent news headline about Anthropic ({marker}).",
+                        )
+                    ],
+                ),
+                session_id=session_id,
+            )
+        )
+        block_types: Final = tuple(block.type for block in response.content or ())
+        assert "web_search_tool_result" in block_types, (
+            f"precondition: the turn never ran an intercepted search, so there is no search row to attribute. "
+            f"blocks={block_types}"
+        )
+
+        rows: Final = proxy.poll_logs_for_session(session_id, min_rows=2, predicate=_has_search_row)
+        by_call_type: Final = {row.call_type or "" for row in rows}
+        assert SEARCH_CALL_TYPE in by_call_type, (
+            f"session {session_id} has no {SEARCH_CALL_TYPE} row, so the intercepted search was billed under a "
+            f"different session and the session view misses its cost. call_types={sorted(by_call_type)} "
+            f"rows={[(row.call_type, row.request_id, row.spend) for row in rows]}"
+        )
+        search_rows: Final = tuple(row for row in rows if row.call_type == SEARCH_CALL_TYPE)
+        turn_rows: Final = tuple(row for row in rows if row.call_type != SEARCH_CALL_TYPE)
+        assert turn_rows, f"session {session_id} carries only search rows: {rows!r}"
+        assert all(row.session_id == session_id for row in rows), (
+            f"session view returned rows outside {session_id}: {[row.session_id for row in rows]}"
+        )
+        assert all((row.spend or 0.0) > 0 for row in search_rows), (
+            f"an intercepted search must stay a separately billed row: {[row.spend for row in search_rows]}"
+        )
+        assert {row.request_id for row in search_rows}.isdisjoint({row.request_id for row in turn_rows}), (
+            "a search row reused its parent turn's request_id instead of keeping its own: "
+            f"{[(row.call_type, row.request_id) for row in rows]}"
+        )

tests/test_litellm/integrations/websearch_interception/test_websearch_interception_handler.py

@@ -287,6 +287,162 @@ async def test_execute_search_attributes_spend_to_the_calling_key(monkeypatch):
     )
 
 
+def _perplexity_router() -> MagicMock:
+    router = MagicMock()
+    router.search_tools = [
+        {
+            "search_tool_name": "perplexity-sonar-pro",
+            "litellm_params": {"search_provider": "perplexity", "api_key": "fake-key"},
+        }
+    ]
+    return router
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    "parent_kwargs",
+    [
+        pytest.param(
+            {
+                "litellm_call_id": "parent-call-1",
+                "litellm_trace_id": "trace-abc",
+                "litellm_session_id": "session-abc",
+                "metadata": {
+                    "user_api_key_auth": UserAPIKeyAuth(api_key="hashed-sk-1234"),
+                    "session_id": "session-abc",
+                },
+            },
+            id="chat-completions-call-kwargs",
+        ),
+        pytest.param(
+            {
+                "litellm_call_id": "parent-call-1",
+                "litellm_metadata": {
+                    "user_api_key_auth": UserAPIKeyAuth(api_key="hashed-sk-1234"),
+                    "session_id": "session-abc",
+                    "trace_id": "trace-abc",
+                },
+            },
+            id="anthropic-messages-litellm-metadata",
+        ),
+        pytest.param(
+            {
+                "litellm_params": {
+                    "litellm_call_id": "parent-call-1",
+                    "litellm_trace_id": "trace-abc",
+                    "metadata": {"user_api_key_auth": UserAPIKeyAuth(api_key="hashed-sk-1234")},
+                    "litellm_metadata": {"session_id": "session-abc"},
+                }
+            },
+            id="logging-payload-with-both-metadata-keys",
+        ),
+    ],
+)
+async def test_execute_search_inherits_parent_request_session_and_trace(
+    monkeypatch: pytest.MonkeyPatch, parent_kwargs: dict[str, object]
+):
+    """The intercepted asearch is billed as its own call but must land in the parent request's
+    session and trace, otherwise every search shows up as a separate one-call session in SpendLogs."""
+    import litellm
+    from litellm.proxy import proxy_server
+    from litellm.proxy.spend_tracking.spend_tracking_utils import _get_session_id_for_spend_log
+
+    logger = WebSearchInterceptionLogger(enabled_providers=["bedrock"], search_tool_name="perplexity-sonar-pro")
+    mock_asearch = AsyncMock(return_value=SearchResponse(object="search", results=[]))
+    monkeypatch.setattr(proxy_server, "llm_router", _perplexity_router())
+    monkeypatch.setattr(litellm, "asearch", mock_asearch)
+
+    await logger._execute_search("what is litellm", kwargs=parent_kwargs)
+
+    forwarded = mock_asearch.await_args.kwargs
+    assert forwarded["litellm_session_id"] == "session-abc"
+    assert forwarded["litellm_trace_id"] == "trace-abc"
+    assert forwarded["litellm_metadata"]["session_id"] == "session-abc"
+    assert forwarded["litellm_metadata"]["trace_id"] == "trace-abc"
+    assert forwarded["litellm_metadata"]["parent_request_id"] == "parent-call-1"
+    assert forwarded["litellm_metadata"]["user_api_key"] == "hashed-sk-1234"
+    assert forwarded["litellm_metadata"]["model_group"] == "perplexity-sonar-pro"
+    assert "litellm_call_id" not in forwarded
+    assert (
+        _get_session_id_for_spend_log(
+            kwargs={"litellm_trace_id": forwarded["litellm_trace_id"]},
+            metadata=forwarded["litellm_metadata"],
+            standard_logging_payload=None,
+            omit_when_missing=True,
+        )
+        == "session-abc"
+    )
+
+
+@pytest.mark.asyncio
+async def test_execute_search_forwards_parent_otel_span_from_key_auth(monkeypatch: pytest.MonkeyPatch):
+    import litellm
+    from litellm.proxy import proxy_server
+
+    logger = WebSearchInterceptionLogger(enabled_providers=["bedrock"], search_tool_name="perplexity-sonar-pro")
+    mock_asearch = AsyncMock(return_value=SearchResponse(object="search", results=[]))
+    monkeypatch.setattr(proxy_server, "llm_router", _perplexity_router())
+    monkeypatch.setattr(litellm, "asearch", mock_asearch)
+    parent_span = object()
+
+    await logger._execute_search(
+        "what is litellm",
+        kwargs={"metadata": {"user_api_key_auth": UserAPIKeyAuth(api_key="sk", parent_otel_span=parent_span)}},
+    )
+
+    assert mock_asearch.await_args.kwargs["litellm_metadata"]["litellm_parent_otel_span"] is parent_span
+
+
+@pytest.mark.asyncio
+async def test_execute_search_without_parent_session_does_not_invent_one(monkeypatch: pytest.MonkeyPatch):
+    """A parent request with no session/trace must not stamp empty correlation keys on the search."""
+    import litellm
+    from litellm.proxy import proxy_server
+
+    logger = WebSearchInterceptionLogger(enabled_providers=["bedrock"], search_tool_name="perplexity-sonar-pro")
+    mock_asearch = AsyncMock(return_value=SearchResponse(object="search", results=[]))
+    monkeypatch.setattr(proxy_server, "llm_router", _perplexity_router())
+    monkeypatch.setattr(litellm, "asearch", mock_asearch)
+
+    await logger._execute_search(
+        "what is litellm",
+        kwargs={"litellm_params": {"metadata": {"user_api_key_auth": UserAPIKeyAuth(api_key="sk"), "prompt": "x"}}},
+    )
+
+    forwarded = mock_asearch.await_args.kwargs
+    assert "litellm_session_id" not in forwarded
+    assert "litellm_trace_id" not in forwarded
+    assert not {"session_id", "trace_id", "parent_request_id", "prompt"} & forwarded["litellm_metadata"].keys()
+
+
+@pytest.mark.asyncio
+async def test_concurrent_searches_keep_their_own_parent_session(monkeypatch: pytest.MonkeyPatch):
+    import asyncio
+
+    import litellm
+    from litellm.proxy import proxy_server
+
+    logger = WebSearchInterceptionLogger(enabled_providers=["bedrock"], search_tool_name="perplexity-sonar-pro")
+    mock_asearch = AsyncMock(return_value=SearchResponse(object="search", results=[]))
+    monkeypatch.setattr(proxy_server, "llm_router", _perplexity_router())
+    monkeypatch.setattr(litellm, "asearch", mock_asearch)
+
+    await asyncio.gather(
+        *(
+            logger._execute_search(
+                f"query {i}",
+                kwargs={"metadata": {"user_api_key_auth": UserAPIKeyAuth(api_key="sk"), "session_id": f"session-{i}"}},
+            )
+            for i in range(5)
+        )
+    )
+
+    seen = {
+        call.kwargs["query"]: call.kwargs["litellm_metadata"]["session_id"] for call in mock_asearch.await_args_list
+    }
+    assert seen == {f"query {i}": f"session-{i}" for i in range(5)}
+
+
 @pytest.mark.asyncio
 async def test_execute_search_without_proxy_auth_context_stays_sdk_only(monkeypatch):
     """SDK callers have no key to attribute the search to, so no proxy metadata is invented."""