From 7edf5b36cfd25666fab30f451b599b6fec524566 Mon Sep 17 00:00:00 2001
From: mateo-berri <277851410+mateo-berri@users.noreply.github.com>
Date: Mon, 31 Aug 2026 16:01:39 -0700
Subject: [PATCH 1/8] fix(guardrails): deliver modify_response block as valid
 SSE on streaming chat and Responses

A guardrail modify_response verdict on a streaming request only produced a
proper replacement on /v1/messages: the chat completions and Responses API
translations had no build_block_sse_chunks, so the ModifyResponseException
re-raised and surfaced as an in-stream 500 error frame (or a whole-request
500 in buffered mode) instead of the documented 200 replacement.

Implement build_block_sse_chunks for both OpenAI translations: chat emits a
content delta plus a finish_reason content_filter chunk with real usage;
Responses emits the typed event sequence (standalone via
build_synthetic_response_events pre-stream, or an output-item continuation
under the in-progress response id mid-stream) ending in response.completed.
---
 basedpyright-code-budget.json                 |   2 +-
 .../chat/guardrail_translation/handler.py     |  10 +-
 .../guardrail_translation/base_translation.py |   5 +-
 .../base_llm/guardrail_translation/utils.py   |  55 ++++
 .../chat/guardrail_translation/handler.py     | 117 +++++++-
 .../guardrail_translation/handler.py          | 204 +++++++++++++-
 litellm/responses/streaming_iterator.py       |   8 +-
 .../test_responses_hooks.py                   |   2 +-
 .../test_openai_guardrail_handler.py          |  51 ++++
 ...test_openai_responses_guardrail_handler.py |  63 +++++
 .../test_openai_streaming_block.py            | 265 ++++++++++++++++++
 type-discipline-budget.json                   |   4 +-
 12 files changed, 769 insertions(+), 17 deletions(-)
 create mode 100644 tests/test_litellm/proxy/guardrails/guardrail_hooks/unified_guardrails/test_openai_streaming_block.py

diff --git a/basedpyright-code-budget.json b/basedpyright-code-budget.json
index 83969d8dedf..88a612fde98 100644
--- a/basedpyright-code-budget.json
+++ b/basedpyright-code-budget.json
@@ -117,7 +117,7 @@
     "limit": 117
   },
   "reportUnnecessaryComparison": {
-    "limit": 697
+    "limit": 696
   },
   "reportUnnecessaryContains": {
     "limit": 5
diff --git a/litellm/llms/anthropic/chat/guardrail_translation/handler.py b/litellm/llms/anthropic/chat/guardrail_translation/handler.py
index b9ca18c7843..21390e5ddb4 100644
--- a/litellm/llms/anthropic/chat/guardrail_translation/handler.py
+++ b/litellm/llms/anthropic/chat/guardrail_translation/handler.py
@@ -144,7 +144,7 @@ class AnthropicMessagesHandler(BaseTranslation):
         self,
         exc: "ModifyResponseException",
         stream_started: bool = False,
-        responses_so_far: list[object] | None = None,
+        responses_so_far: Sequence[object] | None = None,
     ) -> list[bytes]:
         """
         Build an Anthropic SSE sequence delivering the guardrail block message
@@ -162,7 +162,7 @@ class AnthropicMessagesHandler(BaseTranslation):
           would make Anthropic clients reject the stream.
         """
         if stream_started:
-            return self._block_continuation_chunks(exc, responses_so_far or [])
+            return self._block_continuation_chunks(exc, responses_so_far or ())
         return self._standalone_block_chunks(exc)
 
     def _standalone_block_chunks(self, exc: "ModifyResponseException") -> list[bytes]:
@@ -187,7 +187,9 @@ class AnthropicMessagesHandler(BaseTranslation):
         )
         return list(FakeAnthropicMessagesStreamIterator(response=block_response))
 
-    def _block_continuation_chunks(self, exc: "ModifyResponseException", responses_so_far: list[object]) -> list[bytes]:
+    def _block_continuation_chunks(
+        self, exc: "ModifyResponseException", responses_so_far: Sequence[object]
+    ) -> list[bytes]:
         """Continue an already-started message: close the open content block,
         append the block message as a new text block, then end the message --
         without a second message_start."""
@@ -237,7 +239,7 @@ class AnthropicMessagesHandler(BaseTranslation):
 
     @staticmethod
     def _content_block_state(
-        responses_so_far: list[object],
+        responses_so_far: Sequence[object],
     ) -> tuple[int | None, int | None]:
         """From the SSE chunks already sent to the client, return (open
         content-block index or None, highest content-block index seen or None).
diff --git a/litellm/llms/base_llm/guardrail_translation/base_translation.py b/litellm/llms/base_llm/guardrail_translation/base_translation.py
index ba96ab3dc99..247d9bc7c40 100644
--- a/litellm/llms/base_llm/guardrail_translation/base_translation.py
+++ b/litellm/llms/base_llm/guardrail_translation/base_translation.py
@@ -1,4 +1,5 @@
 from abc import ABC, abstractmethod
+from collections.abc import Sequence
 from dataclasses import dataclass, field
 from typing import TYPE_CHECKING, Any, Final, Optional
 
@@ -127,8 +128,8 @@ class BaseTranslation(ABC):
         self,
         exc: "ModifyResponseException",
         stream_started: bool = False,
-        responses_so_far: list[Any] | None = None,
-    ) -> list[bytes] | None:
+        responses_so_far: Sequence[Any] | None = None,
+    ) -> Sequence[bytes] | None:
         """
         Build the streaming chunks that deliver a guardrail block message and
         cleanly terminate the stream in this provider's wire format.
diff --git a/litellm/llms/base_llm/guardrail_translation/utils.py b/litellm/llms/base_llm/guardrail_translation/utils.py
index f09ee210e6c..d30cdcecff5 100644
--- a/litellm/llms/base_llm/guardrail_translation/utils.py
+++ b/litellm/llms/base_llm/guardrail_translation/utils.py
@@ -124,6 +124,61 @@ def blocked_responses_api_usage(original_response: object) -> ResponseAPIUsage:
     )
 
 
+def stream_item_field(item: object, field: str) -> object | None:
+    if isinstance(item, dict):
+        return item.get(field)
+    return getattr(item, field, None)
+
+
+def blocked_chat_stream_usage(original_response: object) -> tuple[int, int]:
+    """
+    ``(prompt_tokens, completion_tokens)`` for a synthetic guardrail-blocked
+    chat completions stream.
+
+    A mid-stream block carries the chunks received so far as a list; real usage
+    rides on the final chunk when the upstream sent one
+    (``stream_options.include_usage``). Non-list originals defer to
+    ``blocked_response_usage``.
+    """
+    if not isinstance(original_response, list):
+        usage: Final = blocked_response_usage(original_response)
+        return usage.get("input_tokens", 0), usage.get("output_tokens", 0)
+    usage_obj: Final = next(
+        (
+            chunk_usage
+            for item in reversed(original_response)
+            if (chunk_usage := stream_item_field(item, "usage")) is not None
+        ),
+        None,
+    )
+    return (
+        _usage_tokens(usage_obj, "prompt_tokens", "input_tokens"),
+        _usage_tokens(usage_obj, "completion_tokens", "output_tokens"),
+    )
+
+
+def blocked_responses_stream_usage(original_response: object) -> ResponseAPIUsage:
+    """
+    ``ResponseAPIUsage`` for a synthetic guardrail-blocked /v1/responses stream.
+
+    A mid-stream block carries the events received so far as a list; real usage
+    rides on the ``response.completed`` event's response when the upstream sent
+    one. Non-list originals defer to ``blocked_responses_api_usage``.
+    """
+    if not isinstance(original_response, list):
+        return blocked_responses_api_usage(original_response)
+    completed: Final = next(
+        (
+            response
+            for item in reversed(original_response)
+            if str(stream_item_field(item, "type") or "") == "response.completed"
+            and (response := stream_item_field(item, "response")) is not None
+        ),
+        None,
+    )
+    return blocked_responses_api_usage(completed)
+
+
 def effective_skip_system_message_for_guardrail(guardrail_to_apply: Any) -> bool:
     per: Final = getattr(guardrail_to_apply, "skip_system_message_in_guardrail", None)
     if per is not None:
diff --git a/litellm/llms/openai/chat/guardrail_translation/handler.py b/litellm/llms/openai/chat/guardrail_translation/handler.py
index 54673c77f80..31a36822c89 100644
--- a/litellm/llms/openai/chat/guardrail_translation/handler.py
+++ b/litellm/llms/openai/chat/guardrail_translation/handler.py
@@ -14,8 +14,14 @@ Pattern Overview:
 This pattern can be replicated for other message formats (e.g., Anthropic).
 """
 
+import json
+import time
+import uuid
+from collections.abc import Sequence
 from typing import TYPE_CHECKING, Any, Final, Union, cast
 
+from typing_extensions import NotRequired, ReadOnly, TypedDict
+
 import litellm
 from litellm._logging import verbose_proxy_logger
 from litellm.llms.base_llm.guardrail_translation.base_translation import (
@@ -23,6 +29,7 @@ from litellm.llms.base_llm.guardrail_translation.base_translation import (
     StreamTransformSink,
 )
 from litellm.llms.base_llm.guardrail_translation.utils import (
+    blocked_chat_stream_usage,
     effective_scan_only_tool_results_for_guardrail,
     effective_skip_system_message_for_guardrail,
     effective_skip_tool_message_for_guardrail,
@@ -31,6 +38,7 @@ from litellm.llms.base_llm.guardrail_translation.utils import (
     openai_tool_name,
     role_out_of_guardrail_scope,
     scoped_structured_message_indices,
+    stream_item_field,
 )
 from litellm.main import stream_chunk_builder
 from litellm.types.llms.openai import AllMessageValues, ChatCompletionToolParam
@@ -46,7 +54,10 @@ from litellm.types.utils import (
 )
 
 if TYPE_CHECKING:
-    from litellm.integrations.custom_guardrail import CustomGuardrail
+    from litellm.integrations.custom_guardrail import (
+        CustomGuardrail,
+        ModifyResponseException,
+    )
     from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
 
 
@@ -1000,3 +1011,107 @@ class OpenAIChatCompletionsHandler(BaseTranslation):
                                 else:
                                     # Subsequent chunks - clear the text
                                     content_item["text"] = ""
+
+    def build_block_sse_chunks(
+        self,
+        exc: "ModifyResponseException",
+        stream_started: bool = False,
+        responses_so_far: Sequence[object] | None = None,
+    ) -> Sequence[bytes]:
+        """
+        Build OpenAI chat-completions SSE chunks that deliver the guardrail
+        block message and terminate the stream cleanly, mirroring the
+        non-streaming block response: ``finish_reason`` ``content_filter`` plus
+        the real usage the upstream call consumed.
+
+        - ``stream_started`` False (buffered / pre-stream): nothing has been
+          sent, so open a standalone completion with a ``role`` delta.
+        - ``stream_started`` True (sampling / mid-stream): chunks already
+          reached the client, so continue the in-progress completion (reuse its
+          id/created/model, content-only delta).
+
+        The proxy's data generator appends ``data: [DONE]`` itself.
+        """
+        chunk_id, created, model = _blocked_stream_identity(exc, responses_so_far or ())
+        prompt_tokens, completion_tokens = blocked_chat_stream_usage(exc.original_response)
+        continuation_delta: Final[_BlockedChunkDelta] = {"content": exc.message}
+        standalone_delta: Final[_BlockedChunkDelta] = {"role": "assistant", "content": exc.message}
+        message_chunk: Final[_BlockedChunk] = {
+            "id": chunk_id,
+            "object": "chat.completion.chunk",
+            "created": created,
+            "model": model,
+            "choices": (
+                {
+                    "index": 0,
+                    "delta": continuation_delta if stream_started else standalone_delta,
+                    "finish_reason": None,
+                },
+            ),
+        }
+        final_chunk: Final[_BlockedChunk] = {
+            "id": chunk_id,
+            "object": "chat.completion.chunk",
+            "created": created,
+            "model": model,
+            "choices": ({"index": 0, "delta": {}, "finish_reason": "content_filter"},),
+            "usage": {
+                "prompt_tokens": prompt_tokens,
+                "completion_tokens": completion_tokens,
+                "total_tokens": prompt_tokens + completion_tokens,
+            },
+        }
+        return _chat_sse_chunk(message_chunk), _chat_sse_chunk(final_chunk)
+
+
+class _BlockedChunkDelta(TypedDict, total=False):
+    role: ReadOnly[str]
+    content: ReadOnly[str]
+
+
+class _BlockedChunkChoice(TypedDict):
+    index: ReadOnly[int]
+    delta: ReadOnly[_BlockedChunkDelta]
+    finish_reason: ReadOnly[str | None]
+
+
+class _BlockedChunkUsage(TypedDict):
+    prompt_tokens: ReadOnly[int]
+    completion_tokens: ReadOnly[int]
+    total_tokens: ReadOnly[int]
+
+
+class _BlockedChunk(TypedDict):
+    id: ReadOnly[str]
+    object: ReadOnly[str]
+    created: ReadOnly[int]
+    model: ReadOnly[str]
+    choices: ReadOnly[tuple[_BlockedChunkChoice, ...]]
+    usage: NotRequired[ReadOnly[_BlockedChunkUsage]]
+
+
+def _chat_sse_chunk(payload: _BlockedChunk) -> bytes:
+    return f"data: {json.dumps(payload)}\n\n".encode()
+
+
+def _blocked_stream_identity(
+    exc: "ModifyResponseException", responses_so_far: Sequence[object]
+) -> tuple[str, int, str]:
+    identified: Final = next(
+        (
+            (chunk_id, item)
+            for item in responses_so_far
+            if isinstance(chunk_id := stream_item_field(item, "id"), str) and chunk_id
+        ),
+        None,
+    )
+    if identified is None:
+        return f"chatcmpl-{uuid.uuid4()}", int(time.time()), exc.model
+    chunk_id, source = identified
+    created: Final = stream_item_field(source, "created")
+    model: Final = stream_item_field(source, "model")
+    return (
+        chunk_id,
+        created if isinstance(created, int) else int(time.time()),
+        model if isinstance(model, str) and model else exc.model,
+    )
diff --git a/litellm/llms/openai/responses/guardrail_translation/handler.py b/litellm/llms/openai/responses/guardrail_translation/handler.py
index 7c5d8ac99ad..9a58899eccd 100644
--- a/litellm/llms/openai/responses/guardrail_translation/handler.py
+++ b/litellm/llms/openai/responses/guardrail_translation/handler.py
@@ -28,6 +28,8 @@ Output: response.output is List[GenericResponseOutputItem] where each has:
     - text: str
 """
 
+import time
+import uuid
 from collections.abc import Sequence
 from typing import TYPE_CHECKING, Any, Final, Union, cast
 
@@ -41,15 +43,31 @@ from litellm.completion_extras.litellm_responses_transformation.transformation i
     OpenAiResponsesToChatCompletionStreamIterator,
 )
 from litellm.llms.base_llm.guardrail_translation.base_translation import BaseTranslation
+from litellm.llms.base_llm.guardrail_translation.utils import (
+    blocked_responses_stream_usage,
+    stream_item_field,
+)
 from litellm.responses.litellm_completion_transformation.transformation import (
     LiteLLMCompletionResponsesConfig,
 )
 from litellm.types.llms.openai import (
     AllMessageValues,
+    BaseLiteLLMOpenAIResponseObject,
     ChatCompletionToolCallChunk,
     ChatCompletionToolParam,
+    ContentPartAddedEvent,
+    ContentPartDoneEvent,
+    ContentPartDonePartOutputText,
     OpenAIMcpServerTool,
+    OutputItemAddedEvent,
+    OutputItemDoneEvent,
+    OutputTextDeltaEvent,
+    OutputTextDoneEvent,
+    ResponseAPIUsage,
+    ResponseCompletedEvent,
+    ResponsesAPIResponse,
     ResponsesAPIStreamEvents,
+    ResponsesAPIStreamingResponse,
 )
 from litellm.types.responses.main import (
     GenericResponseOutputItem,
@@ -59,11 +77,13 @@ from litellm.types.responses.main import (
 from litellm.types.utils import GenericGuardrailAPIInputs
 
 if TYPE_CHECKING:
-    from litellm.integrations.custom_guardrail import CustomGuardrail
+    from litellm.integrations.custom_guardrail import (
+        CustomGuardrail,
+        ModifyResponseException,
+    )
     from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
     from litellm.proxy._types import UserAPIKeyAuth
     from litellm.types.llms.openai import ResponseInputParam
-    from litellm.types.utils import ResponsesAPIResponse
 
 
 class ResponseOutputEnvelope(TypedDict, total=False):
@@ -802,3 +822,183 @@ class OpenAIResponsesHandler(BaseTranslation):
                         content[content_idx]["text"] = guardrail_response
                     elif hasattr(content[content_idx], "text"):
                         content[content_idx].text = guardrail_response
+
+    def build_block_sse_chunks(
+        self,
+        exc: "ModifyResponseException",
+        stream_started: bool = False,
+        responses_so_far: Sequence[object] | None = None,
+    ) -> Sequence[bytes]:
+        """
+        Build Responses API SSE events that deliver the guardrail block message
+        and terminate the stream cleanly, mirroring the non-streaming block
+        response: a completed response whose only output is the violation text,
+        with the real usage the upstream call consumed.
+
+        - ``stream_started`` False (buffered / pre-stream): nothing has been
+          sent, so emit the full synthetic sequence (``response.created``
+          through ``response.completed``).
+        - ``stream_started`` True (sampling / mid-stream): events already
+          reached the client, so continue the in-progress response: deliver the
+          block message as a new output item under the same response id and
+          close with a ``response.completed`` carrying only the replacement
+          item.
+
+        The proxy's data generator appends ``data: [DONE]`` itself.
+        """
+        events: Final = (
+            self._block_continuation_events(exc, responses_so_far or ())
+            if stream_started
+            else self._standalone_block_events(exc)
+        )
+        return tuple(
+            f"data: {event.model_dump_json(exclude_none=True, exclude_unset=True)}\n\n".encode() for event in events
+        )
+
+    @staticmethod
+    def _standalone_block_events(exc: "ModifyResponseException") -> Sequence[ResponsesAPIStreamingResponse]:
+        from litellm.responses.streaming_iterator import build_synthetic_response_events
+
+        return build_synthetic_response_events(
+            transformed=_blocked_response(exc, response_id=f"resp_{uuid.uuid4()}", model=exc.model),
+            logging_obj=None,
+            chunk_size=max(len(exc.message), 1),
+        )
+
+    @staticmethod
+    def _block_continuation_events(
+        exc: "ModifyResponseException", responses_so_far: Sequence[object]
+    ) -> Sequence[ResponsesAPIStreamingResponse]:
+        response_id, model, output_index = _continuation_identity(exc, responses_so_far)
+        item: Final = _blocked_output_item(exc)
+        item_id: Final = item["id"]
+        item_model: Final = BaseLiteLLMOpenAIResponseObject.model_validate(item)
+        part: Final[_BlockedContentPart] = {"type": "output_text", "text": exc.message, "annotations": ()}
+        done_part: Final[_BlockedDoneContentPart] = {
+            "type": "output_text",
+            "text": exc.message,
+            "annotations": (),
+            "logprobs": None,
+        }
+        return (
+            OutputItemAddedEvent(
+                type=ResponsesAPIStreamEvents.OUTPUT_ITEM_ADDED,
+                output_index=output_index,
+                item=item_model,
+            ),
+            ContentPartAddedEvent(
+                type=ResponsesAPIStreamEvents.CONTENT_PART_ADDED,
+                item_id=item_id,
+                output_index=output_index,
+                content_index=0,
+                part=BaseLiteLLMOpenAIResponseObject.model_validate(part),
+            ),
+            OutputTextDeltaEvent(
+                type=ResponsesAPIStreamEvents.OUTPUT_TEXT_DELTA,
+                item_id=item_id,
+                output_index=output_index,
+                content_index=0,
+                delta=exc.message,
+            ),
+            OutputTextDoneEvent(
+                type=ResponsesAPIStreamEvents.OUTPUT_TEXT_DONE,
+                item_id=item_id,
+                output_index=output_index,
+                content_index=0,
+                text=exc.message,
+            ),
+            ContentPartDoneEvent(
+                type=ResponsesAPIStreamEvents.CONTENT_PART_DONE,
+                item_id=item_id,
+                output_index=output_index,
+                content_index=0,
+                part=ContentPartDonePartOutputText.model_validate(done_part),
+            ),
+            OutputItemDoneEvent(
+                type=ResponsesAPIStreamEvents.OUTPUT_ITEM_DONE,
+                output_index=output_index,
+                item=item_model,
+            ),
+            ResponseCompletedEvent(
+                type=ResponsesAPIStreamEvents.RESPONSE_COMPLETED,
+                response=_blocked_response(exc, response_id=response_id, model=model, output_item=item),
+            ),
+        )
+
+
+class _BlockedContentPart(TypedDict):
+    type: ReadOnly[str]
+    text: ReadOnly[str]
+    annotations: ReadOnly[tuple[object, ...]]
+
+
+class _BlockedDoneContentPart(TypedDict):
+    type: ReadOnly[str]
+    text: ReadOnly[str]
+    annotations: ReadOnly[tuple[object, ...]]
+    logprobs: ReadOnly[None]
+
+
+class _BlockedOutputItem(TypedDict):
+    type: ReadOnly[str]
+    id: ReadOnly[str]
+    status: ReadOnly[str]
+    role: ReadOnly[str]
+    content: ReadOnly[tuple[_BlockedContentPart, ...]]
+
+
+class _BlockedResponsePayload(TypedDict):
+    id: ReadOnly[str]
+    object: ReadOnly[str]
+    created_at: ReadOnly[int]
+    model: ReadOnly[str]
+    output: ReadOnly[tuple[_BlockedOutputItem, ...]]
+    status: ReadOnly[str]
+    usage: ReadOnly[ResponseAPIUsage]
+
+
+def _blocked_output_item(exc: "ModifyResponseException") -> _BlockedOutputItem:
+    item: Final[_BlockedOutputItem] = {
+        "type": "message",
+        "id": f"msg_{uuid.uuid4()}",
+        "status": "completed",
+        "role": "assistant",
+        "content": ({"type": "output_text", "text": exc.message, "annotations": ()},),
+    }
+    return item
+
+
+def _blocked_response(
+    exc: "ModifyResponseException",
+    response_id: str,
+    model: str,
+    output_item: _BlockedOutputItem | None = None,
+) -> ResponsesAPIResponse:
+    payload: Final[_BlockedResponsePayload] = {
+        "id": response_id,
+        "object": "response",
+        "created_at": int(time.time()),
+        "model": model,
+        "output": (output_item if output_item is not None else _blocked_output_item(exc),),
+        "status": "completed",
+        "usage": blocked_responses_stream_usage(exc.original_response),
+    }
+    return ResponsesAPIResponse.model_validate(payload)
+
+
+def _continuation_identity(exc: "ModifyResponseException", responses_so_far: Sequence[object]) -> tuple[str, str, int]:
+    responses: Final = tuple(
+        response for item in responses_so_far if (response := stream_item_field(item, "response")) is not None
+    )
+    response_id: Final = next(
+        (rid for response in responses if isinstance(rid := stream_item_field(response, "id"), str) and rid),
+        f"resp_{uuid.uuid4()}",
+    )
+    model: Final = next(
+        (m for response in responses if isinstance(m := stream_item_field(response, "model"), str) and m),
+        exc.model,
+    )
+    indices: Final = tuple(
+        index for item in responses_so_far if isinstance(index := stream_item_field(item, "output_index"), int)
+    )
+    return response_id, model, max(indices) + 1 if indices else 0
diff --git a/litellm/responses/streaming_iterator.py b/litellm/responses/streaming_iterator.py
index 63adf950142..4fd47c50b1e 100644
--- a/litellm/responses/streaming_iterator.py
+++ b/litellm/responses/streaming_iterator.py
@@ -969,7 +969,7 @@ class MockResponsesAPIStreamingIterator(BaseResponsesAPIStreamingIterator):
         transformed: ResponsesAPIResponse,
         logging_obj: LiteLLMLoggingObj,
     ) -> None:
-        self._events: list[ResponsesAPIStreamingResponse] = _build_synthetic_response_events(
+        self._events: Sequence[ResponsesAPIStreamingResponse] = build_synthetic_response_events(
             transformed=transformed,
             logging_obj=logging_obj,
             chunk_size=self.CHUNK_SIZE,
@@ -1036,7 +1036,7 @@ class CachedResponsesAPIStreamingIterator(BaseResponsesAPIStreamingIterator):
         transformed: ResponsesAPIResponse,
         logging_obj: LiteLLMLoggingObj,
     ) -> None:
-        self._events = _build_synthetic_response_events(
+        self._events = build_synthetic_response_events(
             transformed=transformed,
             logging_obj=logging_obj,
             chunk_size=MockResponsesAPIStreamingIterator.CHUNK_SIZE,
@@ -1218,10 +1218,10 @@ def _add_text_like_part_events(
         )
 
 
-def _build_synthetic_response_events(
+def build_synthetic_response_events(
     *,
     transformed: ResponsesAPIResponse,
-    logging_obj: LiteLLMLoggingObj,
+    logging_obj: LiteLLMLoggingObj | None,
     chunk_size: int,
 ) -> list[ResponsesAPIStreamingResponse]:
     openai_types: Final = _get_openai_response_types()
diff --git a/tests/llm_responses_api_testing/test_responses_hooks.py b/tests/llm_responses_api_testing/test_responses_hooks.py
index 66dbb29dba5..a86752c0172 100644
--- a/tests/llm_responses_api_testing/test_responses_hooks.py
+++ b/tests/llm_responses_api_testing/test_responses_hooks.py
@@ -841,7 +841,7 @@ def test_build_synthetic_response_events_covers_annotations_function_calls_and_r
     )
 
     try:
-        events = streaming_module._build_synthetic_response_events(
+        events = streaming_module.build_synthetic_response_events(
             transformed=transformed,
             logging_obj=logging_obj,
             chunk_size=5,
diff --git a/tests/test_litellm/llms/openai/chat/guardrail_translation/test_openai_guardrail_handler.py b/tests/test_litellm/llms/openai/chat/guardrail_translation/test_openai_guardrail_handler.py
index a29e0be4655..8ad0026d1ab 100644
--- a/tests/test_litellm/llms/openai/chat/guardrail_translation/test_openai_guardrail_handler.py
+++ b/tests/test_litellm/llms/openai/chat/guardrail_translation/test_openai_guardrail_handler.py
@@ -1559,3 +1559,54 @@ class TestScanOnlyToolResults:
         assert data["messages"][3]["content"] == "page says [BLOCKED] here"
         assert data["messages"][3]["tool_call_id"] == "call_1"
         assert data["messages"][4]["content"] == "and then?"
+
+
+class TestBuildBlockSseChunks:
+    """build_block_sse_chunks turns a streaming ModifyResponseException into 200 SSE chunks"""
+
+    def _exc(self, original_response=None):
+        from litellm.exceptions import ModifyResponseException
+
+        return ModifyResponseException(
+            message="Blocked by policy.",
+            model="gpt-5.4-mini",
+            request_data={},
+            guardrail_name="test",
+            original_response=original_response,
+        )
+
+    def _payloads(self, chunks):
+        return [json.loads(chunk.decode().removeprefix("data: ").strip()) for chunk in chunks]
+
+    def test_standalone_block_uses_fresh_identity_and_zero_usage(self):
+        handler = OpenAIChatCompletionsHandler()
+        first, final = self._payloads(handler.build_block_sse_chunks(self._exc(), stream_started=False))
+        assert first["id"].startswith("chatcmpl-")
+        assert first["model"] == "gpt-5.4-mini"
+        assert first["choices"][0]["delta"] == {"role": "assistant", "content": "Blocked by policy."}
+        assert first["choices"][0]["finish_reason"] is None
+        assert final["choices"][0]["delta"] == {}
+        assert final["choices"][0]["finish_reason"] == "content_filter"
+        assert final["usage"] == {"prompt_tokens": 0, "completion_tokens": 0, "total_tokens": 0}
+
+    def test_continuation_reuses_stream_identity_and_real_usage(self):
+        handler = OpenAIChatCompletionsHandler()
+        yielded = [
+            {"id": "chatcmpl-live", "created": 1724900000, "model": "gpt-5.4-mini-2026-01-01"},
+        ]
+        original = yielded + [
+            {"id": "chatcmpl-live", "usage": {"prompt_tokens": 11, "completion_tokens": 5}},
+        ]
+        first, final = self._payloads(
+            handler.build_block_sse_chunks(
+                self._exc(original_response=original), stream_started=True, responses_so_far=yielded
+            )
+        )
+        assert (first["id"], first["created"], first["model"]) == (
+            "chatcmpl-live",
+            1724900000,
+            "gpt-5.4-mini-2026-01-01",
+        )
+        assert first["choices"][0]["delta"] == {"content": "Blocked by policy."}
+        assert final["id"] == "chatcmpl-live"
+        assert final["usage"] == {"prompt_tokens": 11, "completion_tokens": 5, "total_tokens": 16}
diff --git a/tests/test_litellm/llms/openai/responses/test_openai_responses_guardrail_handler.py b/tests/test_litellm/llms/openai/responses/test_openai_responses_guardrail_handler.py
index 447175b09a6..cd0e7f29933 100644
--- a/tests/test_litellm/llms/openai/responses/test_openai_responses_guardrail_handler.py
+++ b/tests/test_litellm/llms/openai/responses/test_openai_responses_guardrail_handler.py
@@ -1229,3 +1229,66 @@ class TestOpenAIResponsesHandlerToolInjection:
         names = [t.get("name") for t in result["tools"]]
         assert "get_weather" in names
         assert "injected_tool" in names
+
+
+class TestBuildBlockSseChunks:
+    """build_block_sse_chunks turns a streaming ModifyResponseException into 200 SSE events"""
+
+    def _exc(self, original_response=None):
+        from litellm.exceptions import ModifyResponseException
+
+        return ModifyResponseException(
+            message="Blocked by policy.",
+            model="gpt-5.4-mini",
+            request_data={},
+            guardrail_name="test",
+            original_response=original_response,
+        )
+
+    def _payloads(self, chunks):
+        import json
+
+        return [json.loads(chunk.decode().removeprefix("data: ").strip()) for chunk in chunks]
+
+    def test_standalone_block_emits_complete_synthetic_stream(self):
+        handler = OpenAIResponsesHandler()
+        payloads = self._payloads(handler.build_block_sse_chunks(self._exc(), stream_started=False))
+        types = [payload["type"] for payload in payloads]
+        assert types[0] == "response.created"
+        assert types[-1] == "response.completed"
+        completed = payloads[-1]["response"]
+        assert completed["id"].startswith("resp_")
+        assert completed["model"] == "gpt-5.4-mini"
+        assert completed["output"][0]["content"][0]["text"] == "Blocked by policy."
+
+    def test_continuation_appends_item_at_next_output_index_with_real_usage(self):
+        handler = OpenAIResponsesHandler()
+        yielded = [
+            {"type": "response.created", "response": {"id": "resp_live", "model": "gpt-5.4-mini-2026-01-01"}},
+            {"type": "response.output_item.added", "output_index": 2, "item": {"id": "msg_orig"}},
+        ]
+        original = yielded + [
+            {
+                "type": "response.completed",
+                "response": {
+                    "id": "resp_live",
+                    "model": "gpt-5.4-mini-2026-01-01",
+                    "output": [],
+                    "usage": {"input_tokens": 7, "output_tokens": 21, "total_tokens": 28},
+                },
+            }
+        ]
+        payloads = self._payloads(
+            handler.build_block_sse_chunks(
+                self._exc(original_response=original), stream_started=True, responses_so_far=yielded
+            )
+        )
+        types = [payload["type"] for payload in payloads]
+        assert "response.created" not in types
+        assert types[0] == "response.output_item.added"
+        assert payloads[0]["output_index"] == 3
+        completed = payloads[-1]["response"]
+        assert completed["id"] == "resp_live"
+        assert completed["model"] == "gpt-5.4-mini-2026-01-01"
+        assert completed["output"][0]["content"][0]["text"] == "Blocked by policy."
+        assert completed["usage"] == {"input_tokens": 7, "output_tokens": 21, "total_tokens": 28}
diff --git a/tests/test_litellm/proxy/guardrails/guardrail_hooks/unified_guardrails/test_openai_streaming_block.py b/tests/test_litellm/proxy/guardrails/guardrail_hooks/unified_guardrails/test_openai_streaming_block.py
new file mode 100644
index 00000000000..4d5581290ba
--- /dev/null
+++ b/tests/test_litellm/proxy/guardrails/guardrail_hooks/unified_guardrails/test_openai_streaming_block.py
@@ -0,0 +1,265 @@
+"""
+Regression tests for blocking an OpenAI-format streaming response from the
+unified guardrail post-call streaming iterator hook.
+
+When a guardrail's ``apply_guardrail`` raises ``ModifyResponseException``
+while (or at the end of) a chat completions or Responses API stream is being
+relayed, the hook must emit a well-formed SSE termination sequence carrying
+the block message - NOT a bare ``data: {"error": ...}`` blob that surfaces as
+an HTTP 500 error frame and truncates the stream.
+"""
+
+import json
+from typing import Any, AsyncGenerator, List, Literal, Optional
+
+import pytest
+
+from litellm.integrations.custom_guardrail import (
+    CustomGuardrail,
+    ModifyResponseException,
+)
+from litellm.proxy._types import UserAPIKeyAuth
+from litellm.proxy.guardrails.guardrail_hooks.unified_guardrail.unified_guardrail import (
+    UnifiedLLMGuardrails,
+)
+from litellm.types.utils import (
+    Delta,
+    GenericGuardrailAPIInputs,
+    ModelResponseStream,
+    StreamingChoices,
+)
+
+BLOCK_MESSAGE = "This response was replaced by policy."
+
+
+class _BlockingGuardrail(CustomGuardrail):
+    """Mock guardrail that always blocks response scans by raising ModifyResponseException."""
+
+    async def apply_guardrail(
+        self,
+        inputs: GenericGuardrailAPIInputs,
+        request_data: dict,
+        input_type: Literal["request", "response"],
+        logging_obj: Optional[Any] = None,
+    ) -> GenericGuardrailAPIInputs:
+        raise ModifyResponseException(
+            message=BLOCK_MESSAGE,
+            model="gpt-5.4-mini",
+            request_data=request_data,
+            guardrail_name=self.guardrail_name,
+        )
+
+
+def _chat_chunk(delta: Delta, finish_reason: Optional[str] = None) -> ModelResponseStream:
+    return ModelResponseStream(
+        id="chatcmpl-live",
+        created=1724900000,
+        model="gpt-5.4-mini",
+        choices=[StreamingChoices(index=0, delta=delta, finish_reason=finish_reason)],
+    )
+
+
+async def _chat_stream(end: bool) -> AsyncGenerator[ModelResponseStream, None]:
+    yield _chat_chunk(Delta(role="assistant", content="This "))
+    for text in ["is ", "the ", "original ", "answer."]:
+        yield _chat_chunk(Delta(content=text))
+    if end:
+        yield _chat_chunk(Delta(), finish_reason="stop")
+
+
+async def _responses_stream(end: bool) -> AsyncGenerator[dict, None]:
+    original_text = "This is the original answer."
+    response_envelope = {"id": "resp_live", "model": "gpt-5.4-mini", "status": "in_progress", "output": []}
+    yield {"type": "response.created", "response": response_envelope}
+    yield {"type": "response.in_progress", "response": response_envelope}
+    yield {
+        "type": "response.output_item.added",
+        "output_index": 0,
+        "item": {"id": "msg_orig", "type": "message", "role": "assistant", "content": []},
+    }
+    yield {
+        "type": "response.content_part.added",
+        "item_id": "msg_orig",
+        "output_index": 0,
+        "content_index": 0,
+        "part": {"type": "output_text", "text": "", "annotations": []},
+    }
+    for delta in ["This ", "is ", "the ", "original ", "answer."]:
+        yield {
+            "type": "response.output_text.delta",
+            "item_id": "msg_orig",
+            "output_index": 0,
+            "content_index": 0,
+            "delta": delta,
+        }
+    yield {
+        "type": "response.output_text.done",
+        "item_id": "msg_orig",
+        "output_index": 0,
+        "content_index": 0,
+        "text": original_text,
+    }
+    if end:
+        yield {
+            "type": "response.completed",
+            "response": {
+                "id": "resp_live",
+                "model": "gpt-5.4-mini",
+                "status": "completed",
+                "output": [
+                    {
+                        "id": "msg_orig",
+                        "type": "message",
+                        "role": "assistant",
+                        "status": "completed",
+                        "content": [{"type": "output_text", "text": original_text, "annotations": []}],
+                    }
+                ],
+                "usage": {"input_tokens": 7, "output_tokens": 21, "total_tokens": 28},
+            },
+        }
+
+
+async def _run_hook(
+    route: str,
+    stream: AsyncGenerator[Any, None],
+    sampling_rate: int = 1,
+    end_of_stream_only: bool = False,
+    buffer_until_moderated: bool = False,
+) -> List[Any]:
+    guardrail = _BlockingGuardrail(guardrail_name="test-blocking-guardrail", event_hook="post_call")
+    guardrail.streaming_sampling_rate = sampling_rate
+    guardrail.streaming_end_of_stream_only = end_of_stream_only
+    guardrail.streaming_buffer_until_moderated = buffer_until_moderated
+
+    unified_guardrail = UnifiedLLMGuardrails()
+    user_api_key_dict = UserAPIKeyAuth(api_key="test", request_route=route)
+    request_data = {
+        "messages": [{"role": "user", "content": "hi"}],
+        "guardrail_to_apply": guardrail,
+        "metadata": {"guardrails": ["test-blocking-guardrail"]},
+    }
+
+    collected: List[Any] = []
+    async for chunk in unified_guardrail.async_post_call_streaming_iterator_hook(
+        user_api_key_dict=user_api_key_dict,
+        response=stream,
+        request_data=request_data,
+    ):
+        collected.append(chunk)
+    return collected
+
+
+def _sse_payloads(collected: List[Any]) -> List[dict]:
+    payloads = []
+    for chunk in collected:
+        if not isinstance(chunk, bytes):
+            continue
+        for block in chunk.decode().split("\n\n"):
+            for line in block.strip().split("\n"):
+                if line.startswith("data:"):
+                    payloads.append(json.loads(line[len("data:") :].strip()))
+    return payloads
+
+
+def _assert_no_error_frame(collected: List[Any]) -> None:
+    raw = "".join(chunk.decode() for chunk in collected if isinstance(chunk, bytes))
+    assert '"error"' not in raw, f"unexpected error blob in stream: {raw!r}"
+
+
+@pytest.mark.asyncio
+async def test_chat_pre_stream_block_emits_standalone_completion():
+    """Block on the first chunk: a standalone completion opens with a role delta
+    and ends with finish_reason content_filter."""
+    collected = await _run_hook("/v1/chat/completions", _chat_stream(end=False))
+    _assert_no_error_frame(collected)
+    payloads = _sse_payloads(collected)
+    assert payloads, "no block SSE chunks were emitted"
+    assert payloads[0]["choices"][0]["delta"] == {"role": "assistant", "content": BLOCK_MESSAGE}
+    assert payloads[-1]["choices"][0]["finish_reason"] == "content_filter"
+
+
+@pytest.mark.asyncio
+async def test_chat_mid_stream_block_continues_the_completion():
+    """Regression for the LIT-6496 500 error frame: after chunks were already
+    forwarded, the block continues the same completion id and terminates with
+    finish_reason content_filter instead of raising into an error blob."""
+    collected = await _run_hook("/v1/chat/completions", _chat_stream(end=False), sampling_rate=5)
+    _assert_no_error_frame(collected)
+    forwarded = [chunk for chunk in collected if isinstance(chunk, ModelResponseStream)]
+    assert forwarded, "original chunks should have streamed before the block"
+    payloads = _sse_payloads(collected)
+    assert payloads, "no block SSE chunks were emitted"
+    assert all(payload["id"] == "chatcmpl-live" for payload in payloads), (
+        "block chunks must continue the in-progress completion, not start a new one"
+    )
+    assert payloads[0]["choices"][0]["delta"] == {"content": BLOCK_MESSAGE}
+    assert payloads[-1]["choices"][0]["finish_reason"] == "content_filter"
+
+
+@pytest.mark.asyncio
+async def test_chat_end_of_stream_block_terminates_cleanly():
+    collected = await _run_hook("/v1/chat/completions", _chat_stream(end=True), end_of_stream_only=True)
+    _assert_no_error_frame(collected)
+    payloads = _sse_payloads(collected)
+    assert BLOCK_MESSAGE in json.dumps(payloads)
+    assert payloads[-1]["choices"][0]["finish_reason"] == "content_filter"
+
+
+@pytest.mark.asyncio
+async def test_responses_buffered_block_emits_full_event_sequence():
+    """Buffered moderation blocks before anything streams: a complete synthetic
+    Responses stream from response.created through response.completed carrying
+    the block message, with the original content never released."""
+    collected = await _run_hook("/v1/responses", _responses_stream(end=True), buffer_until_moderated=True)
+    _assert_no_error_frame(collected)
+    assert not [chunk for chunk in collected if isinstance(chunk, dict)], (
+        "buffered original chunks must never be released after a block"
+    )
+    payloads = _sse_payloads(collected)
+    event_types = [payload["type"] for payload in payloads]
+    assert event_types[0] == "response.created"
+    assert "response.output_text.delta" in event_types
+    assert event_types[-1] == "response.completed"
+    completed = payloads[-1]["response"]
+    assert completed["status"] == "completed"
+    assert completed["output"][0]["content"][0]["text"] == BLOCK_MESSAGE
+    assert completed["usage"] == {"input_tokens": 7, "output_tokens": 21, "total_tokens": 28}
+    assert "original answer" not in json.dumps(payloads)
+
+
+@pytest.mark.asyncio
+async def test_responses_mid_stream_block_continues_the_response():
+    """Regression for the LIT-6496 500 error frame: after events were already
+    forwarded, the block appends a new output item under the same response id
+    and closes with response.completed - never a second response.created."""
+    collected = await _run_hook("/v1/responses", _responses_stream(end=False))
+    _assert_no_error_frame(collected)
+    forwarded_types = [chunk["type"] for chunk in collected if isinstance(chunk, dict)]
+    assert "response.created" in forwarded_types, "original events should have streamed before the block"
+    payloads = _sse_payloads(collected)
+    assert payloads, "no block SSE chunks were emitted"
+    block_types = [payload["type"] for payload in payloads]
+    assert "response.created" not in block_types, "a mid-stream block must not restart the response"
+    assert block_types[0] == "response.output_item.added"
+    assert block_types[-1] == "response.completed"
+    assert payloads[0]["output_index"] == 1, "the block item must continue after the original output item"
+    completed = payloads[-1]["response"]
+    assert completed["id"] == "resp_live"
+    assert completed["output"][0]["content"][0]["text"] == BLOCK_MESSAGE
+
+
+@pytest.mark.asyncio
+async def test_responses_end_of_stream_block_reports_original_usage():
+    collected = await _run_hook("/v1/responses", _responses_stream(end=True), end_of_stream_only=True)
+    _assert_no_error_frame(collected)
+    forwarded_types = [chunk["type"] for chunk in collected if isinstance(chunk, dict)]
+    assert "response.completed" not in forwarded_types, (
+        "the original terminal event must be withheld and replaced by the block sequence"
+    )
+    payloads = _sse_payloads(collected)
+    completed = payloads[-1]["response"]
+    assert payloads[-1]["type"] == "response.completed"
+    assert completed["id"] == "resp_live"
+    assert completed["output"][0]["content"][0]["text"] == BLOCK_MESSAGE
+    assert completed["usage"] == {"input_tokens": 7, "output_tokens": 21, "total_tokens": 28}
diff --git a/type-discipline-budget.json b/type-discipline-budget.json
index 7365cec4fdd..f17d0b7af4a 100644
--- a/type-discipline-budget.json
+++ b/type-discipline-budget.json
@@ -1,9 +1,9 @@
 {
   "LIT001": {
-    "limit": 22704
+    "limit": 22698
   },
   "LIT002": {
-    "limit": 26854
+    "limit": 26853
   },
   "LIT003": {
     "limit": 269

From 31a9f7e6ad932b0adbc24f7666178aa4c1071984 Mon Sep 17 00:00:00 2001
From: mateo-berri <277851410+mateo-berri@users.noreply.github.com>
Date: Mon, 31 Aug 2026 16:34:21 -0700
Subject: [PATCH 2/8] test(guardrails): type streaming-block test helpers and
 drop mutable accumulators

---
 .../test_openai_streaming_block.py            | 50 ++++++++++---------
 1 file changed, 27 insertions(+), 23 deletions(-)

diff --git a/tests/test_litellm/proxy/guardrails/guardrail_hooks/unified_guardrails/test_openai_streaming_block.py b/tests/test_litellm/proxy/guardrails/guardrail_hooks/unified_guardrails/test_openai_streaming_block.py
index 4d5581290ba..99172ec3be6 100644
--- a/tests/test_litellm/proxy/guardrails/guardrail_hooks/unified_guardrails/test_openai_streaming_block.py
+++ b/tests/test_litellm/proxy/guardrails/guardrail_hooks/unified_guardrails/test_openai_streaming_block.py
@@ -10,7 +10,7 @@ an HTTP 500 error frame and truncates the stream.
 """
 
 import json
-from typing import Any, AsyncGenerator, List, Literal, Optional
+from typing import Any, AsyncGenerator, Dict, Literal, Optional, Tuple, Union
 
 import pytest
 
@@ -31,6 +31,9 @@ from litellm.types.utils import (
 
 BLOCK_MESSAGE = "This response was replaced by policy."
 
+JsonPayload = Dict[str, object]
+StreamChunk = Union[ModelResponseStream, JsonPayload, bytes]
+
 
 class _BlockingGuardrail(CustomGuardrail):
     """Mock guardrail that always blocks response scans by raising ModifyResponseException."""
@@ -67,7 +70,7 @@ async def _chat_stream(end: bool) -> AsyncGenerator[ModelResponseStream, None]:
         yield _chat_chunk(Delta(), finish_reason="stop")
 
 
-async def _responses_stream(end: bool) -> AsyncGenerator[dict, None]:
+async def _responses_stream(end: bool) -> AsyncGenerator[JsonPayload, None]:
     original_text = "This is the original answer."
     response_envelope = {"id": "resp_live", "model": "gpt-5.4-mini", "status": "in_progress", "output": []}
     yield {"type": "response.created", "response": response_envelope}
@@ -122,11 +125,11 @@ async def _responses_stream(end: bool) -> AsyncGenerator[dict, None]:
 
 async def _run_hook(
     route: str,
-    stream: AsyncGenerator[Any, None],
+    stream: AsyncGenerator[Union[ModelResponseStream, JsonPayload], None],
     sampling_rate: int = 1,
     end_of_stream_only: bool = False,
     buffer_until_moderated: bool = False,
-) -> List[Any]:
+) -> Tuple[StreamChunk, ...]:
     guardrail = _BlockingGuardrail(guardrail_name="test-blocking-guardrail", event_hook="post_call")
     guardrail.streaming_sampling_rate = sampling_rate
     guardrail.streaming_end_of_stream_only = end_of_stream_only
@@ -140,29 +143,30 @@ async def _run_hook(
         "metadata": {"guardrails": ["test-blocking-guardrail"]},
     }
 
-    collected: List[Any] = []
-    async for chunk in unified_guardrail.async_post_call_streaming_iterator_hook(
-        user_api_key_dict=user_api_key_dict,
-        response=stream,
-        request_data=request_data,
-    ):
-        collected.append(chunk)
-    return collected
+    return tuple(
+        [
+            chunk
+            async for chunk in unified_guardrail.async_post_call_streaming_iterator_hook(
+                user_api_key_dict=user_api_key_dict,
+                response=stream,
+                request_data=request_data,
+            )
+        ]
+    )
 
 
-def _sse_payloads(collected: List[Any]) -> List[dict]:
-    payloads = []
-    for chunk in collected:
-        if not isinstance(chunk, bytes):
-            continue
-        for block in chunk.decode().split("\n\n"):
-            for line in block.strip().split("\n"):
-                if line.startswith("data:"):
-                    payloads.append(json.loads(line[len("data:") :].strip()))
-    return payloads
+def _sse_payloads(collected: Tuple[StreamChunk, ...]) -> Tuple[JsonPayload, ...]:
+    return tuple(
+        json.loads(line[len("data:") :].strip())
+        for chunk in collected
+        if isinstance(chunk, bytes)
+        for block in chunk.decode().split("\n\n")
+        for line in block.strip().split("\n")
+        if line.startswith("data:")
+    )
 
 
-def _assert_no_error_frame(collected: List[Any]) -> None:
+def _assert_no_error_frame(collected: Tuple[StreamChunk, ...]) -> None:
     raw = "".join(chunk.decode() for chunk in collected if isinstance(chunk, bytes))
     assert '"error"' not in raw, f"unexpected error blob in stream: {raw!r}"
 

From 78b57fb427de52a3fc0f10ba5c20005f46725a6f Mon Sep 17 00:00:00 2001
From: mateo-berri <277851410+mateo-berri@users.noreply.github.com>
Date: Mon, 31 Aug 2026 17:05:45 -0700
Subject: [PATCH 3/8] fix(guardrails): withhold chat finish chunk in
 end_of_stream_only mode and close open Responses items before a mid-stream
 block

---
 .../chat/guardrail_translation/handler.py     |  22 +++
 .../guardrail_translation/handler.py          | 133 +++++++++++++++++-
 .../test_openai_guardrail_handler.py          |  33 +++++
 ...test_openai_responses_guardrail_handler.py |  26 +++-
 .../test_openai_streaming_block.py            |  74 ++++++++--
 5 files changed, 273 insertions(+), 15 deletions(-)

diff --git a/litellm/llms/openai/chat/guardrail_translation/handler.py b/litellm/llms/openai/chat/guardrail_translation/handler.py
index 05716e65137..55b27947faa 100644
--- a/litellm/llms/openai/chat/guardrail_translation/handler.py
+++ b/litellm/llms/openai/chat/guardrail_translation/handler.py
@@ -1015,6 +1015,21 @@ class OpenAIChatCompletionsHandler(BaseTranslation):
                                     # Subsequent chunks - clear the text
                                     content_item["text"] = ""
 
+    def _check_streaming_has_ended(self, responses_so_far: Sequence[object]) -> bool:
+        """
+        True once any relayed chunk carries a non-null ``finish_reason``.
+
+        The unified guardrail's ``end_of_stream_only`` streaming path probes
+        this via ``hasattr`` to withhold the terminal chunks until
+        end-of-stream moderation runs, so a block can replace the finish
+        instead of trailing after a ``finish_reason`` the client already saw.
+        """
+        return any(
+            stream_item_field(choice, "finish_reason") is not None
+            for item in responses_so_far
+            for choice in _stream_chunk_choices(item)
+        )
+
     def build_block_sse_chunks(
         self,
         exc: "ModifyResponseException",
@@ -1097,6 +1112,13 @@ def _chat_sse_chunk(payload: _BlockedChunk) -> bytes:
     return f"data: {json.dumps(payload)}\n\n".encode()
 
 
+def _stream_chunk_choices(item: object) -> Sequence[object]:
+    choices: Final = stream_item_field(item, "choices")
+    if isinstance(choices, Sequence) and not isinstance(choices, (str, bytes)):
+        return choices
+    return ()
+
+
 def _blocked_stream_identity(
     exc: "ModifyResponseException", responses_so_far: Sequence[object]
 ) -> tuple[str, int, str]:
diff --git a/litellm/llms/openai/responses/guardrail_translation/handler.py b/litellm/llms/openai/responses/guardrail_translation/handler.py
index 811c4feeacc..cf01242f151 100644
--- a/litellm/llms/openai/responses/guardrail_translation/handler.py
+++ b/litellm/llms/openai/responses/guardrail_translation/handler.py
@@ -31,6 +31,7 @@ Output: response.output is List[GenericResponseOutputItem] where each has:
 import time
 import uuid
 from collections.abc import Sequence
+from dataclasses import dataclass
 from typing import TYPE_CHECKING, Any, Final, Union, cast
 
 from openai.types.responses.response_function_tool_call import ResponseFunctionToolCall
@@ -874,10 +875,10 @@ class OpenAIResponsesHandler(BaseTranslation):
           sent, so emit the full synthetic sequence (``response.created``
           through ``response.completed``).
         - ``stream_started`` True (sampling / mid-stream): events already
-          reached the client, so continue the in-progress response: deliver the
-          block message as a new output item under the same response id and
-          close with a ``response.completed`` carrying only the replacement
-          item.
+          reached the client, so continue the in-progress response: close the
+          output item still open on the wire, deliver the block message as a
+          new output item under the same response id, and close with a
+          ``response.completed`` carrying only the replacement item.
 
         The proxy's data generator appends ``data: [DONE]`` itself.
         """
@@ -887,7 +888,8 @@ class OpenAIResponsesHandler(BaseTranslation):
             else self._standalone_block_events(exc)
         )
         return tuple(
-            f"data: {event.model_dump_json(exclude_none=True, exclude_unset=True)}\n\n".encode() for event in events
+            f"data: {event.model_dump_json(exclude_none=True, exclude_unset=True, serialize_as_any=True)}\n\n".encode()
+            for event in events
         )
 
     @staticmethod
@@ -915,6 +917,7 @@ class OpenAIResponsesHandler(BaseTranslation):
             "logprobs": None,
         }
         return (
+            *_open_item_closing_events(responses_so_far),
             OutputItemAddedEvent(
                 type=ResponsesAPIStreamEvents.OUTPUT_ITEM_ADDED,
                 output_index=output_index,
@@ -1036,3 +1039,123 @@ def _continuation_identity(exc: "ModifyResponseException", responses_so_far: Seq
         index for item in responses_so_far if isinstance(index := stream_item_field(item, "output_index"), int)
     )
     return response_id, model, max(indices) + 1 if indices else 0
+
+
+@dataclass(frozen=True, slots=True)
+class _OpenItemState:
+    item_id: str
+    item_type: str
+    role: str
+    output_index: int
+    content_index: int
+    text: str
+    part_open: bool
+
+
+def _open_item_state(responses_so_far: Sequence[object]) -> _OpenItemState | None:
+    typed: Final = tuple((str(stream_item_field(event, "type") or ""), event) for event in responses_so_far)
+    added: Final = tuple(
+        (added_index, stream_item_field(event, "item"))
+        for event_type, event in typed
+        if event_type == "response.output_item.added"
+        and isinstance(added_index := stream_item_field(event, "output_index"), int)
+    )
+    done_indices: Final = frozenset(
+        done_index
+        for event_type, event in typed
+        if event_type == "response.output_item.done"
+        and isinstance(done_index := stream_item_field(event, "output_index"), int)
+    )
+    open_added: Final = tuple((index, payload) for index, payload in added if index not in done_indices)
+    if not open_added:
+        return None
+    output_index, item_payload = open_added[-1]
+    item_id: Final = stream_item_field(item_payload, "id") if item_payload is not None else None
+    if not isinstance(item_id, str) or not item_id:
+        return None
+    raw_type: Final = stream_item_field(item_payload, "type")
+    raw_role: Final = stream_item_field(item_payload, "role")
+    part_added: Final = tuple(
+        part_index
+        for event_type, event in typed
+        if event_type == "response.content_part.added"
+        and stream_item_field(event, "item_id") == item_id
+        and isinstance(part_index := stream_item_field(event, "content_index"), int)
+    )
+    part_done: Final = frozenset(
+        part_done_index
+        for event_type, event in typed
+        if event_type == "response.content_part.done"
+        and stream_item_field(event, "item_id") == item_id
+        and isinstance(part_done_index := stream_item_field(event, "content_index"), int)
+    )
+    open_parts: Final = tuple(index for index in part_added if index not in part_done)
+    text: Final = "".join(
+        delta
+        for event_type, event in typed
+        if event_type == "response.output_text.delta"
+        and stream_item_field(event, "item_id") == item_id
+        and isinstance(delta := stream_item_field(event, "delta"), str)
+    )
+    return _OpenItemState(
+        item_id=item_id,
+        item_type=raw_type if isinstance(raw_type, str) and raw_type else "message",
+        role=raw_role if isinstance(raw_role, str) and raw_role else "assistant",
+        output_index=output_index,
+        content_index=open_parts[-1] if open_parts else 0,
+        text=text,
+        part_open=bool(open_parts),
+    )
+
+
+def _open_item_closing_events(responses_so_far: Sequence[object]) -> Sequence[ResponsesAPIStreamingResponse]:
+    """Close the output item still in progress on the relayed stream before the
+    block item is appended: strict Responses clients reject a
+    ``response.completed`` that arrives while an earlier ``output_item.added``
+    was never closed. The closing text is exactly what the client has received
+    for that item so far."""
+    open_item: Final = _open_item_state(responses_so_far)
+    if open_item is None:
+        return ()
+    partial_part: Final[_BlockedContentPart] = {
+        "type": "output_text",
+        "text": open_item.text,
+        "annotations": (),
+    }
+    closed_payload: Final[_BlockedItemPayload] = {
+        "type": open_item.item_type,
+        "id": open_item.item_id,
+        "status": "completed",
+        "role": open_item.role,
+        "content": (partial_part,),
+    }
+    item_done: Final = OutputItemDoneEvent(
+        type=ResponsesAPIStreamEvents.OUTPUT_ITEM_DONE,
+        output_index=open_item.output_index,
+        item=GenericResponseOutputItem.model_validate(closed_payload),
+    )
+    if not open_item.part_open:
+        return (item_done,)
+    partial_done_part: Final[_BlockedDoneContentPart] = {
+        "type": "output_text",
+        "text": open_item.text,
+        "annotations": (),
+        "logprobs": None,
+    }
+    return (
+        OutputTextDoneEvent(
+            type=ResponsesAPIStreamEvents.OUTPUT_TEXT_DONE,
+            item_id=open_item.item_id,
+            output_index=open_item.output_index,
+            content_index=open_item.content_index,
+            text=open_item.text,
+        ),
+        ContentPartDoneEvent(
+            type=ResponsesAPIStreamEvents.CONTENT_PART_DONE,
+            item_id=open_item.item_id,
+            output_index=open_item.output_index,
+            content_index=open_item.content_index,
+            part=ContentPartDonePartOutputText.model_validate(partial_done_part),
+        ),
+        item_done,
+    )
diff --git a/tests/test_litellm/llms/openai/chat/guardrail_translation/test_openai_guardrail_handler.py b/tests/test_litellm/llms/openai/chat/guardrail_translation/test_openai_guardrail_handler.py
index 8ad0026d1ab..7dd6065063a 100644
--- a/tests/test_litellm/llms/openai/chat/guardrail_translation/test_openai_guardrail_handler.py
+++ b/tests/test_litellm/llms/openai/chat/guardrail_translation/test_openai_guardrail_handler.py
@@ -1610,3 +1610,36 @@ class TestBuildBlockSseChunks:
         assert first["choices"][0]["delta"] == {"content": "Blocked by policy."}
         assert final["id"] == "chatcmpl-live"
         assert final["usage"] == {"prompt_tokens": 11, "completion_tokens": 5, "total_tokens": 16}
+
+
+class TestCheckStreamingHasEnded:
+    """_check_streaming_has_ended lets end_of_stream_only withhold the finish chunk until moderation"""
+
+    def test_empty_and_content_only_chunks_are_not_ended(self):
+        handler = OpenAIChatCompletionsHandler()
+        assert handler._check_streaming_has_ended([]) is False
+        content_only = [
+            {"id": "chatcmpl-live", "choices": [{"index": 0, "delta": {"content": "hi"}, "finish_reason": None}]},
+            {"id": "chatcmpl-live", "choices": []},
+            {"id": "chatcmpl-live", "usage": {"prompt_tokens": 1, "completion_tokens": 1}},
+        ]
+        assert handler._check_streaming_has_ended(content_only) is False
+
+    def test_dict_finish_chunk_marks_stream_ended(self):
+        handler = OpenAIChatCompletionsHandler()
+        chunks = [
+            {"id": "chatcmpl-live", "choices": [{"index": 0, "delta": {"content": "hi"}, "finish_reason": None}]},
+            {"id": "chatcmpl-live", "choices": [{"index": 0, "delta": {}, "finish_reason": "stop"}]},
+        ]
+        assert handler._check_streaming_has_ended(chunks) is True
+
+    def test_object_finish_chunk_marks_stream_ended(self):
+        from litellm.types.utils import Delta, ModelResponseStream, StreamingChoices
+
+        handler = OpenAIChatCompletionsHandler()
+        chunks = [
+            ModelResponseStream(
+                choices=[StreamingChoices(index=0, delta=Delta(content=None), finish_reason="stop")]
+            )
+        ]
+        assert handler._check_streaming_has_ended(chunks) is True
diff --git a/tests/test_litellm/llms/openai/responses/test_openai_responses_guardrail_handler.py b/tests/test_litellm/llms/openai/responses/test_openai_responses_guardrail_handler.py
index cd0e7f29933..1ae4e7a699b 100644
--- a/tests/test_litellm/llms/openai/responses/test_openai_responses_guardrail_handler.py
+++ b/tests/test_litellm/llms/openai/responses/test_openai_responses_guardrail_handler.py
@@ -1285,10 +1285,32 @@ class TestBuildBlockSseChunks:
         )
         types = [payload["type"] for payload in payloads]
         assert "response.created" not in types
-        assert types[0] == "response.output_item.added"
-        assert payloads[0]["output_index"] == 3
+        assert types[0] == "response.output_item.done"
+        assert payloads[0]["output_index"] == 2
+        assert payloads[0]["item"]["id"] == "msg_orig"
+        assert payloads[0]["item"]["status"] == "completed"
+        assert types[1] == "response.output_item.added"
+        assert payloads[1]["output_index"] == 3
         completed = payloads[-1]["response"]
         assert completed["id"] == "resp_live"
         assert completed["model"] == "gpt-5.4-mini-2026-01-01"
         assert completed["output"][0]["content"][0]["text"] == "Blocked by policy."
         assert completed["usage"] == {"input_tokens": 7, "output_tokens": 21, "total_tokens": 28}
+
+    def test_continuation_without_open_item_emits_no_closing_events(self):
+        handler = OpenAIResponsesHandler()
+        yielded = [
+            {"type": "response.created", "response": {"id": "resp_live", "model": "gpt-5.4-mini"}},
+            {"type": "response.in_progress", "response": {"id": "resp_live"}},
+        ]
+        payloads = self._payloads(
+            handler.build_block_sse_chunks(
+                self._exc(original_response=yielded), stream_started=True, responses_so_far=yielded
+            )
+        )
+        types = [payload["type"] for payload in payloads]
+        assert types[0] == "response.output_item.added"
+        assert types[-1] == "response.completed"
+        dones = [payload for payload in payloads if payload["type"] == "response.output_item.done"]
+        assert len(dones) == 1
+        assert dones[0]["item"]["content"][0]["text"] == "Blocked by policy."
diff --git a/tests/test_litellm/proxy/guardrails/guardrail_hooks/unified_guardrails/test_openai_streaming_block.py b/tests/test_litellm/proxy/guardrails/guardrail_hooks/unified_guardrails/test_openai_streaming_block.py
index 99172ec3be6..42bdf41bc88 100644
--- a/tests/test_litellm/proxy/guardrails/guardrail_hooks/unified_guardrails/test_openai_streaming_block.py
+++ b/tests/test_litellm/proxy/guardrails/guardrail_hooks/unified_guardrails/test_openai_streaming_block.py
@@ -53,6 +53,19 @@ class _BlockingGuardrail(CustomGuardrail):
         )
 
 
+class _PassingGuardrail(CustomGuardrail):
+    """Mock guardrail that always lets response scans through unchanged."""
+
+    async def apply_guardrail(
+        self,
+        inputs: GenericGuardrailAPIInputs,
+        request_data: dict,
+        input_type: Literal["request", "response"],
+        logging_obj: Optional[Any] = None,
+    ) -> GenericGuardrailAPIInputs:
+        return inputs
+
+
 def _chat_chunk(delta: Delta, finish_reason: Optional[str] = None) -> ModelResponseStream:
     return ModelResponseStream(
         id="chatcmpl-live",
@@ -129,8 +142,13 @@ async def _run_hook(
     sampling_rate: int = 1,
     end_of_stream_only: bool = False,
     buffer_until_moderated: bool = False,
+    blocks: bool = True,
 ) -> Tuple[StreamChunk, ...]:
-    guardrail = _BlockingGuardrail(guardrail_name="test-blocking-guardrail", event_hook="post_call")
+    guardrail = (
+        _BlockingGuardrail(guardrail_name="test-blocking-guardrail", event_hook="post_call")
+        if blocks
+        else _PassingGuardrail(guardrail_name="test-passing-guardrail", event_hook="post_call")
+    )
     guardrail.streaming_sampling_rate = sampling_rate
     guardrail.streaming_end_of_stream_only = end_of_stream_only
     guardrail.streaming_buffer_until_moderated = buffer_until_moderated
@@ -140,7 +158,7 @@ async def _run_hook(
     request_data = {
         "messages": [{"role": "user", "content": "hi"}],
         "guardrail_to_apply": guardrail,
-        "metadata": {"guardrails": ["test-blocking-guardrail"]},
+        "metadata": {"guardrails": [guardrail.guardrail_name]},
     }
 
     return tuple(
@@ -203,13 +221,38 @@ async def test_chat_mid_stream_block_continues_the_completion():
 
 @pytest.mark.asyncio
 async def test_chat_end_of_stream_block_terminates_cleanly():
+    """Regression for bugbot's finish-ordering finding: in end_of_stream_only
+    mode the original finish chunk must be withheld until moderation decides,
+    so a block's content_filter finish is the only stream terminator a client
+    ever sees - never policy text trailing after finish_reason stop."""
     collected = await _run_hook("/v1/chat/completions", _chat_stream(end=True), end_of_stream_only=True)
     _assert_no_error_frame(collected)
+    forwarded = [chunk for chunk in collected if isinstance(chunk, ModelResponseStream)]
+    assert forwarded, "content chunks still stream to the client before end-of-stream moderation"
+    assert all(choice.finish_reason is None for chunk in forwarded for choice in chunk.choices), (
+        "the original finish chunk must be withheld until moderation decides"
+    )
     payloads = _sse_payloads(collected)
     assert BLOCK_MESSAGE in json.dumps(payloads)
     assert payloads[-1]["choices"][0]["finish_reason"] == "content_filter"
 
 
+@pytest.mark.asyncio
+async def test_chat_end_of_stream_pass_releases_withheld_finish_chunk():
+    """When end-of-stream moderation passes, the withheld finish chunk is
+    released so a clean stream still terminates normally."""
+    collected = await _run_hook(
+        "/v1/chat/completions", _chat_stream(end=True), end_of_stream_only=True, blocks=False
+    )
+    assert not [chunk for chunk in collected if isinstance(chunk, bytes)], (
+        "a clean stream must carry no synthetic block frames"
+    )
+    forwarded = [chunk for chunk in collected if isinstance(chunk, ModelResponseStream)]
+    finish_reasons = [choice.finish_reason for chunk in forwarded for choice in chunk.choices]
+    assert finish_reasons[-1] == "stop", "the withheld finish chunk must be released after moderation passes"
+    assert all(reason is None for reason in finish_reasons[:-1])
+
+
 @pytest.mark.asyncio
 async def test_responses_buffered_block_emits_full_event_sequence():
     """Buffered moderation blocks before anything streams: a complete synthetic
@@ -234,20 +277,35 @@ async def test_responses_buffered_block_emits_full_event_sequence():
 
 @pytest.mark.asyncio
 async def test_responses_mid_stream_block_continues_the_response():
-    """Regression for the LIT-6496 500 error frame: after events were already
-    forwarded, the block appends a new output item under the same response id
-    and closes with response.completed - never a second response.created."""
+    """Regression for the LIT-6496 500 error frame and bugbot's unclosed-item
+    finding: after events were already forwarded, the block first closes the
+    output item still open on the wire, then appends the replacement item under
+    the same response id, and closes with response.completed - never a second
+    response.created and never a completed response with an item left open."""
     collected = await _run_hook("/v1/responses", _responses_stream(end=False))
     _assert_no_error_frame(collected)
-    forwarded_types = [chunk["type"] for chunk in collected if isinstance(chunk, dict)]
+    forwarded = [chunk for chunk in collected if isinstance(chunk, dict)]
+    forwarded_types = [chunk["type"] for chunk in forwarded]
     assert "response.created" in forwarded_types, "original events should have streamed before the block"
     payloads = _sse_payloads(collected)
     assert payloads, "no block SSE chunks were emitted"
     block_types = [payload["type"] for payload in payloads]
     assert "response.created" not in block_types, "a mid-stream block must not restart the response"
-    assert block_types[0] == "response.output_item.added"
     assert block_types[-1] == "response.completed"
-    assert payloads[0]["output_index"] == 1, "the block item must continue after the original output item"
+
+    all_events = forwarded + list(payloads)
+    opened = sorted(event["output_index"] for event in all_events if event["type"] == "response.output_item.added")
+    closed = sorted(event["output_index"] for event in all_events if event["type"] == "response.output_item.done")
+    assert opened == closed, "every output item opened on the stream must be closed before response.completed"
+    original_done_position = block_types.index("response.output_item.done")
+    block_item_position = block_types.index("response.output_item.added")
+    assert original_done_position < block_item_position, (
+        "the in-progress original item must be closed before the block item is appended"
+    )
+    assert payloads[original_done_position]["item"]["id"] == "msg_orig"
+    assert payloads[block_item_position]["output_index"] == 1, (
+        "the block item must continue after the original output item"
+    )
     completed = payloads[-1]["response"]
     assert completed["id"] == "resp_live"
     assert completed["output"][0]["content"][0]["text"] == BLOCK_MESSAGE

From 38825cf9c62eeb3c374b87a4ea9b2c31cc9f19cc Mon Sep 17 00:00:00 2001
From: mateo-berri <277851410+mateo-berri@users.noreply.github.com>
Date: Mon, 31 Aug 2026 17:19:55 -0700
Subject: [PATCH 4/8] fix(guardrails): match Responses stream event types by
 value so enum-typed events close the open item

---
 .../guardrail_translation/handler.py          |  2 +-
 ...test_openai_responses_guardrail_handler.py | 59 +++++++++++++++++++
 2 files changed, 60 insertions(+), 1 deletion(-)

diff --git a/litellm/llms/openai/responses/guardrail_translation/handler.py b/litellm/llms/openai/responses/guardrail_translation/handler.py
index cf01242f151..6295df1dbfa 100644
--- a/litellm/llms/openai/responses/guardrail_translation/handler.py
+++ b/litellm/llms/openai/responses/guardrail_translation/handler.py
@@ -1053,7 +1053,7 @@ class _OpenItemState:
 
 
 def _open_item_state(responses_so_far: Sequence[object]) -> _OpenItemState | None:
-    typed: Final = tuple((str(stream_item_field(event, "type") or ""), event) for event in responses_so_far)
+    typed: Final = tuple((stream_item_field(event, "type"), event) for event in responses_so_far)
     added: Final = tuple(
         (added_index, stream_item_field(event, "item"))
         for event_type, event in typed
diff --git a/tests/test_litellm/llms/openai/responses/test_openai_responses_guardrail_handler.py b/tests/test_litellm/llms/openai/responses/test_openai_responses_guardrail_handler.py
index 1ae4e7a699b..d6e56f0faf1 100644
--- a/tests/test_litellm/llms/openai/responses/test_openai_responses_guardrail_handler.py
+++ b/tests/test_litellm/llms/openai/responses/test_openai_responses_guardrail_handler.py
@@ -1297,6 +1297,65 @@ class TestBuildBlockSseChunks:
         assert completed["output"][0]["content"][0]["text"] == "Blocked by policy."
         assert completed["usage"] == {"input_tokens": 7, "output_tokens": 21, "total_tokens": 28}
 
+    def test_continuation_closes_open_item_given_pydantic_events_with_enum_types(self):
+        from litellm.types.llms.openai import (
+            BaseLiteLLMOpenAIResponseObject,
+            ContentPartAddedEvent,
+            OutputItemAddedEvent,
+            OutputTextDeltaEvent,
+            ResponsesAPIStreamEvents,
+        )
+
+        handler = OpenAIResponsesHandler()
+        open_item = GenericResponseOutputItem.model_validate(
+            {"type": "message", "id": "msg_live", "status": "in_progress", "role": "assistant", "content": []}
+        )
+        yielded = [
+            OutputItemAddedEvent(
+                type=ResponsesAPIStreamEvents.OUTPUT_ITEM_ADDED, output_index=0, item=open_item
+            ),
+            ContentPartAddedEvent(
+                type=ResponsesAPIStreamEvents.CONTENT_PART_ADDED,
+                item_id="msg_live",
+                output_index=0,
+                content_index=0,
+                part=BaseLiteLLMOpenAIResponseObject.model_validate(
+                    {"type": "output_text", "text": "", "annotations": []}
+                ),
+            ),
+            OutputTextDeltaEvent(
+                type=ResponsesAPIStreamEvents.OUTPUT_TEXT_DELTA,
+                item_id="msg_live",
+                output_index=0,
+                content_index=0,
+                delta="partial ",
+            ),
+            OutputTextDeltaEvent(
+                type=ResponsesAPIStreamEvents.OUTPUT_TEXT_DELTA,
+                item_id="msg_live",
+                output_index=0,
+                content_index=0,
+                delta="text",
+            ),
+        ]
+        payloads = self._payloads(
+            handler.build_block_sse_chunks(
+                self._exc(original_response=yielded), stream_started=True, responses_so_far=yielded
+            )
+        )
+        types = [payload["type"] for payload in payloads]
+        assert types[:3] == [
+            "response.output_text.done",
+            "response.content_part.done",
+            "response.output_item.done",
+        ]
+        assert payloads[0]["text"] == "partial text"
+        assert payloads[2]["item"]["id"] == "msg_live"
+        assert payloads[2]["item"]["status"] == "completed"
+        assert payloads[2]["item"]["content"][0]["text"] == "partial text"
+        assert types[3] == "response.output_item.added"
+        assert payloads[3]["output_index"] == 1
+
     def test_continuation_without_open_item_emits_no_closing_events(self):
         handler = OpenAIResponsesHandler()
         yielded = [

From 329654765a9d07a3f8d15abc6a86040044668473 Mon Sep 17 00:00:00 2001
From: mateo-berri <277851410+mateo-berri@users.noreply.github.com>
Date: Mon, 31 Aug 2026 17:31:59 -0700
Subject: [PATCH 5/8] test(guardrails): update chat eos block tests for
 finish-chunk withholding

---
 .../guardrails/guardrail_hooks/test_bedrock_guardrails.py  | 7 +++++--
 .../unified_guardrails/test_unified_guardrail.py           | 3 ++-
 2 files changed, 7 insertions(+), 3 deletions(-)

diff --git a/tests/test_litellm/proxy/guardrails/guardrail_hooks/test_bedrock_guardrails.py b/tests/test_litellm/proxy/guardrails/guardrail_hooks/test_bedrock_guardrails.py
index 235a5c0c09b..bacaa1c3e22 100644
--- a/tests/test_litellm/proxy/guardrails/guardrail_hooks/test_bedrock_guardrails.py
+++ b/tests/test_litellm/proxy/guardrails/guardrail_hooks/test_bedrock_guardrails.py
@@ -5524,7 +5524,9 @@ async def test_streaming_end_of_stream_block_emits_error_frame_instead_of_trunca
     """Regression for PR #38722: a topicPolicy DENY caught by the end-of-stream
     scan used to raise after SSE headers were flushed, so the client saw a
     silently truncated stream. The unified hook must emit the chat in-stream
-    error frame instead."""
+    error frame instead. The finish chunk is withheld while the end-of-stream
+    scan runs, so on a block it is dropped rather than relayed before the
+    frame."""
     from litellm.llms import load_guardrail_translation_mappings
     from litellm.proxy.guardrails.guardrail_hooks.unified_guardrail import (
         unified_guardrail as unified_module,
@@ -5582,8 +5584,9 @@ async def test_streaming_end_of_stream_block_emits_error_frame_instead_of_trunca
     finally:
         unified_module.endpoint_guardrail_translation_mappings = None
 
-    assert len(out) == 3
+    assert len(out) == 2
     assert isinstance(out[0], ModelResponseStream)
+    assert out[0].choices[0].finish_reason is None
     frame = out[-1]
     assert isinstance(frame, bytes)
     payload = json.loads(frame.decode()[len("data: ") :])
diff --git a/tests/test_litellm/proxy/guardrails/guardrail_hooks/unified_guardrails/test_unified_guardrail.py b/tests/test_litellm/proxy/guardrails/guardrail_hooks/unified_guardrails/test_unified_guardrail.py
index 0b32558a00a..8cad1c634a9 100644
--- a/tests/test_litellm/proxy/guardrails/guardrail_hooks/unified_guardrails/test_unified_guardrail.py
+++ b/tests/test_litellm/proxy/guardrails/guardrail_hooks/unified_guardrails/test_unified_guardrail.py
@@ -1844,7 +1844,8 @@ class TestStreamingHttpErrorFrames:
 
         out = await _drive_stream(UnifiedLLMGuardrails(), guardrail, chunks)
 
-        assert out[:2] == chunks
+        assert out[0] == chunks[0]
+        assert chunks[1] not in out
         frame = out[-1]
         assert isinstance(frame, bytes)
         text = frame.decode()

From 4567fc784c6fe43acd41a6db32c834465b8d653d Mon Sep 17 00:00:00 2001
From: mateo-berri <277851410+mateo-berri@users.noreply.github.com>
Date: Tue, 1 Sep 2026 12:45:23 -0700
Subject: [PATCH 6/8] fix: close non-message open items as incomplete when a
 stream is blocked

---
 .../guardrail_translation/handler.py          | 38 +++++++++++++++---
 ...test_openai_responses_guardrail_handler.py | 40 +++++++++++++++++++
 2 files changed, 73 insertions(+), 5 deletions(-)

diff --git a/litellm/llms/openai/responses/guardrail_translation/handler.py b/litellm/llms/openai/responses/guardrail_translation/handler.py
index 6295df1dbfa..14759f6475e 100644
--- a/litellm/llms/openai/responses/guardrail_translation/handler.py
+++ b/litellm/llms/openai/responses/guardrail_translation/handler.py
@@ -30,13 +30,14 @@ Output: response.output is List[GenericResponseOutputItem] where each has:
 
 import time
 import uuid
-from collections.abc import Sequence
+from collections.abc import Mapping, Sequence
 from dataclasses import dataclass
+from types import MappingProxyType
 from typing import TYPE_CHECKING, Any, Final, Union, cast
 
 from openai.types.responses.response_function_tool_call import ResponseFunctionToolCall
 from openai.types.responses.tool_param import FunctionToolParam
-from pydantic import BaseModel
+from pydantic import BaseModel, TypeAdapter
 from typing_extensions import ReadOnly, TypedDict
 
 from litellm._logging import verbose_proxy_logger
@@ -1050,6 +1051,7 @@ class _OpenItemState:
     content_index: int
     text: str
     part_open: bool
+    payload: object
 
 
 def _open_item_state(responses_so_far: Sequence[object]) -> _OpenItemState | None:
@@ -1070,7 +1072,9 @@ def _open_item_state(responses_so_far: Sequence[object]) -> _OpenItemState | Non
     if not open_added:
         return None
     output_index, item_payload = open_added[-1]
-    item_id: Final = stream_item_field(item_payload, "id") if item_payload is not None else None
+    if item_payload is None:
+        return None
+    item_id: Final = stream_item_field(item_payload, "id")
     if not isinstance(item_id, str) or not item_id:
         return None
     raw_type: Final = stream_item_field(item_payload, "type")
@@ -1105,18 +1109,42 @@ def _open_item_state(responses_so_far: Sequence[object]) -> _OpenItemState | Non
         content_index=open_parts[-1] if open_parts else 0,
         text=text,
         part_open=bool(open_parts),
+        payload=item_payload,
     )
 
 
+_item_fields_adapter: Final = TypeAdapter(Mapping[str, object])
+_no_item_fields: Final[Mapping[str, object]] = MappingProxyType({})
+
+
+def _incomplete_item_fields(payload: object) -> Mapping[str, object]:
+    raw: Final = payload.model_dump() if isinstance(payload, BaseModel) else payload
+    if not isinstance(raw, dict):
+        return _no_item_fields
+    return _item_fields_adapter.validate_python(raw)
+
+
 def _open_item_closing_events(responses_so_far: Sequence[object]) -> Sequence[ResponsesAPIStreamingResponse]:
     """Close the output item still in progress on the relayed stream before the
     block item is appended: strict Responses clients reject a
     ``response.completed`` that arrives while an earlier ``output_item.added``
-    was never closed. The closing text is exactly what the client has received
-    for that item so far."""
+    was never closed. A message item closes ``completed`` with exactly the text
+    the client has received so far; any other item type (a function call the
+    guardrail rejected, for instance) closes ``incomplete`` so the synthetic
+    done event can never authorize acting on it."""
     open_item: Final = _open_item_state(responses_so_far)
     if open_item is None:
         return ()
+    if open_item.item_type != "message":
+        return (
+            OutputItemDoneEvent(
+                type=ResponsesAPIStreamEvents.OUTPUT_ITEM_DONE,
+                output_index=open_item.output_index,
+                item=BaseLiteLLMOpenAIResponseObject.model_validate(
+                    MappingProxyType({**_incomplete_item_fields(open_item.payload), "status": "incomplete"})
+                ),
+            ),
+        )
     partial_part: Final[_BlockedContentPart] = {
         "type": "output_text",
         "text": open_item.text,
diff --git a/tests/test_litellm/llms/openai/responses/test_openai_responses_guardrail_handler.py b/tests/test_litellm/llms/openai/responses/test_openai_responses_guardrail_handler.py
index d6e56f0faf1..49f4e9df7c9 100644
--- a/tests/test_litellm/llms/openai/responses/test_openai_responses_guardrail_handler.py
+++ b/tests/test_litellm/llms/openai/responses/test_openai_responses_guardrail_handler.py
@@ -1356,6 +1356,46 @@ class TestBuildBlockSseChunks:
         assert types[3] == "response.output_item.added"
         assert payloads[3]["output_index"] == 1
 
+    def test_continuation_closes_open_function_call_as_incomplete(self):
+        handler = OpenAIResponsesHandler()
+        yielded = [
+            {"type": "response.created", "response": {"id": "resp_live", "model": "gpt-5.4-mini"}},
+            {
+                "type": "response.output_item.added",
+                "output_index": 0,
+                "item": {
+                    "id": "fc_live",
+                    "type": "function_call",
+                    "status": "in_progress",
+                    "call_id": "call_1",
+                    "name": "run_payment",
+                    "arguments": "",
+                },
+            },
+            {
+                "type": "response.function_call_arguments.delta",
+                "item_id": "fc_live",
+                "output_index": 0,
+                "delta": '{"amount": 100}',
+            },
+        ]
+        payloads = self._payloads(
+            handler.build_block_sse_chunks(
+                self._exc(original_response=yielded), stream_started=True, responses_so_far=yielded
+            )
+        )
+        types = [payload["type"] for payload in payloads]
+        assert types[0] == "response.output_item.done"
+        closed = payloads[0]["item"]
+        assert closed["id"] == "fc_live"
+        assert closed["type"] == "function_call"
+        assert closed["status"] == "incomplete"
+        assert closed["name"] == "run_payment"
+        assert "content" not in closed
+        assert types[1] == "response.output_item.added"
+        assert payloads[1]["output_index"] == 1
+        assert types[-1] == "response.completed"
+
     def test_continuation_without_open_item_emits_no_closing_events(self):
         handler = OpenAIResponsesHandler()
         yielded = [

From 8c7fe00d80a9118baf5d2b8f87d24dc53644a8fd Mon Sep 17 00:00:00 2001
From: mateo-berri <277851410+mateo-berri@users.noreply.github.com>
Date: Tue, 1 Sep 2026 17:37:07 -0700
Subject: [PATCH 7/8] fix: compare stream event types by equality so typed
 completed events keep their usage

---
 .../base_llm/guardrail_translation/utils.py   |  2 +-
 ...test_openai_responses_guardrail_handler.py | 32 +++++++++++++++++++
 2 files changed, 33 insertions(+), 1 deletion(-)

diff --git a/litellm/llms/base_llm/guardrail_translation/utils.py b/litellm/llms/base_llm/guardrail_translation/utils.py
index d30cdcecff5..9b6f9c47105 100644
--- a/litellm/llms/base_llm/guardrail_translation/utils.py
+++ b/litellm/llms/base_llm/guardrail_translation/utils.py
@@ -171,7 +171,7 @@ def blocked_responses_stream_usage(original_response: object) -> ResponseAPIUsag
         (
             response
             for item in reversed(original_response)
-            if str(stream_item_field(item, "type") or "") == "response.completed"
+            if stream_item_field(item, "type") == "response.completed"
             and (response := stream_item_field(item, "response")) is not None
         ),
         None,
diff --git a/tests/test_litellm/llms/openai/responses/test_openai_responses_guardrail_handler.py b/tests/test_litellm/llms/openai/responses/test_openai_responses_guardrail_handler.py
index d68ca0fdfb6..315b6948bd8 100644
--- a/tests/test_litellm/llms/openai/responses/test_openai_responses_guardrail_handler.py
+++ b/tests/test_litellm/llms/openai/responses/test_openai_responses_guardrail_handler.py
@@ -1389,6 +1389,38 @@ class TestBuildBlockSseChunks:
         assert completed["output"][0]["content"][0]["text"] == "Blocked by policy."
         assert completed["usage"] == {"input_tokens": 7, "output_tokens": 21, "total_tokens": 28}
 
+    def test_continuation_reads_usage_from_typed_completed_event(self):
+        from litellm.types.llms.openai import (
+            ResponseCompletedEvent,
+            ResponsesAPIResponse,
+            ResponsesAPIStreamEvents,
+        )
+
+        handler = OpenAIResponsesHandler()
+        original = [
+            ResponseCompletedEvent(
+                type=ResponsesAPIStreamEvents.RESPONSE_COMPLETED,
+                response=ResponsesAPIResponse.model_validate(
+                    {
+                        "id": "resp_live",
+                        "created_at": 1,
+                        "model": "gpt-5.4-mini",
+                        "output": [],
+                        "usage": {"input_tokens": 7, "output_tokens": 21, "total_tokens": 28},
+                    }
+                ),
+            )
+        ]
+        payloads = self._payloads(
+            handler.build_block_sse_chunks(
+                self._exc(original_response=original), stream_started=True, responses_so_far=[]
+            )
+        )
+        completed = payloads[-1]["response"]
+        assert completed["usage"]["input_tokens"] == 7
+        assert completed["usage"]["output_tokens"] == 21
+        assert completed["usage"]["total_tokens"] == 28
+
     def test_continuation_closes_open_item_given_pydantic_events_with_enum_types(self):
         from litellm.types.llms.openai import (
             BaseLiteLLMOpenAIResponseObject,

From e2c3f51c46aa2a11a930b6c19bc28e4144a38a1e Mon Sep 17 00:00:00 2001
From: James Liounis <jamesliounisgithub@gmail.com>
Date: Wed, 2 Sep 2026 00:46:46 -0400
Subject: [PATCH 8/8] fix(search): forward search-tool params through the
 router, complete Parallel AI v1 param mapping (#37883)
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

* fix(search): forward search-tool params through the router, complete Parallel AI v1 param mapping

SearchAPIRouter dropped every parameter configured on a search tool, forwarding
only per-request kwargs. Any tool-level setting (mode, max_results, ...) was
silently lost on the way to the adapter, for every search provider.

Also completes the Parallel AI v1 search surface: after_date, fetch_policy,
location and include_domains now nest under advanced_settings instead of being
sent as unknown top-level fields, responses preserve search_id / session_id /
warnings / raw excerpts, and search cost is derived from the request mode and
the provider's reported usage rather than a single flat rate.

* fix(parallel_ai): stop a caller from pricing its own search request

`_parallel_ai_usage` carries the provider's reported usage into cost
calculation. It was only written when the response contained a usage block, so
a caller could pass `_parallel_ai_usage=[{"name": "sku_search", "count": 0}]`
and, whenever the provider omitted usage, bill $0.00 instead of $0.005 — the
value also reached the upstream request body as an unknown field.

The key is now stripped from inbound params and written unconditionally from
the parsed response, so only the provider can populate it.

* fix(parallel_ai): price fast search mode correctly

* test(parallel_ai): fake search at HTTP boundary

* fix(parallel_ai): tolerate null search result fields

---------

Co-authored-by: khushishelat <shelatkhushi@gmail.com>
---
 .../parallel_ai/search/cost_calculator.py     |  90 +++++++++
 .../llms/parallel_ai/search/transformation.py | 143 +++++++++----
 ...odel_prices_and_context_window_backup.json |  14 +-
 litellm/router_utils/search_api_router.py     |  12 +-
 litellm/search/cost_calculator.py             |  35 +++-
 model_prices_and_context_window.json          |  14 +-
 .../test_router_helper_utils.py               |   3 +
 .../parallel_ai/test_parallel_ai_search.py    | 190 +++++++++++++++--
 .../test_parallel_ai_search_gateway.py        | 191 ++++++++++++++++++
 9 files changed, 637 insertions(+), 55 deletions(-)
 create mode 100644 litellm/llms/parallel_ai/search/cost_calculator.py
 create mode 100644 tests/test_litellm/llms/parallel_ai/test_parallel_ai_search_gateway.py

diff --git a/litellm/llms/parallel_ai/search/cost_calculator.py b/litellm/llms/parallel_ai/search/cost_calculator.py
new file mode 100644
index 00000000000..809cd280cc8
--- /dev/null
+++ b/litellm/llms/parallel_ai/search/cost_calculator.py
@@ -0,0 +1,90 @@
+from collections.abc import Mapping, Sequence
+from types import MappingProxyType
+from typing import Final
+
+from pydantic import TypeAdapter, ValidationError
+
+from litellm.utils import get_model_info
+
+PARALLEL_AI_DEFAULT_RESULTS: Final = 10
+PARALLEL_AI_ADDITIONAL_RESULT_COST: Final = 0.001
+PARALLEL_AI_USAGE_PARAM: Final = "_parallel_ai_usage"
+PARALLEL_AI_STANDARD_SEARCH_MODEL: Final = "parallel_ai/search"
+PARALLEL_AI_FAST_SEARCH_MODEL: Final = "parallel_ai/search-fast"
+PARALLEL_AI_TURBO_SEARCH_MODEL: Final = "parallel_ai/search-turbo"
+PARALLEL_AI_PRICING_MODEL_BY_MODE: Final[Mapping[str, str]] = MappingProxyType(
+    {
+        "fast": PARALLEL_AI_FAST_SEARCH_MODEL,
+        "turbo": PARALLEL_AI_TURBO_SEARCH_MODEL,
+    }
+)
+ADVANCED_SETTINGS_ADAPTER: Final[TypeAdapter[Mapping[str, object]]] = TypeAdapter(Mapping[str, object])
+
+
+def _non_negative_int(value: object) -> int | None:
+    if isinstance(value, bool) or not isinstance(value, int) or value < 0:
+        return None
+    return value
+
+
+def _usage_count(usage: Sequence[Mapping[str, object]], sku: str) -> int | None:
+    counts: Final = tuple(
+        count
+        for item in usage
+        if item.get("name") == sku
+        if (count := _non_negative_int(item.get("count"))) is not None
+    )
+    return sum(counts) if counts else None
+
+
+def _effective_mode(optional_params: Mapping[str, object]) -> str:
+    mode: Final = optional_params.get("mode")
+    if isinstance(mode, str):
+        return mode
+
+    processor: Final = optional_params.get("processor")
+    if processor == "pro":
+        return "advanced"
+    return "basic"
+
+
+def _effective_max_results(optional_params: Mapping[str, object]) -> int:
+    try:
+        advanced_settings: Final = ADVANCED_SETTINGS_ADAPTER.validate_python(optional_params.get("advanced_settings"))
+        advanced_max_results: Final = _non_negative_int(advanced_settings.get("max_results"))
+        if advanced_max_results is not None:
+            return advanced_max_results
+    except ValidationError:
+        pass
+
+    max_results: Final = _non_negative_int(optional_params.get("max_results"))
+    return max_results if max_results is not None else PARALLEL_AI_DEFAULT_RESULTS
+
+
+def _request_cost(mode: str) -> float:
+    pricing_model: Final = PARALLEL_AI_PRICING_MODEL_BY_MODE.get(mode, PARALLEL_AI_STANDARD_SEARCH_MODEL)
+    model_info: Final = get_model_info(model=pricing_model, custom_llm_provider="parallel_ai")
+    return float(model_info.get("input_cost_per_query") or 0.0)
+
+
+def _additional_results(
+    optional_params: Mapping[str, object],
+    usage: Sequence[Mapping[str, object]] | None,
+) -> int:
+    usage_count: Final = _usage_count(usage, "sku_search_additional_results") if usage is not None else None
+    if usage_count is not None:
+        return usage_count
+    if usage is not None:
+        return 0
+    return max(_effective_max_results(optional_params) - PARALLEL_AI_DEFAULT_RESULTS, 0)
+
+
+def parallel_ai_search_cost(
+    optional_params: Mapping[str, object],
+    usage: Sequence[Mapping[str, object]] | None,
+) -> float:
+    request_cost: Final = _request_cost(_effective_mode(optional_params))
+    request_count_from_usage: Final = _usage_count(usage, "sku_search") if usage is not None else None
+    request_count: Final = request_count_from_usage if request_count_from_usage is not None else 1
+    additional_results: Final = _additional_results(optional_params, usage)
+    return request_count * request_cost + additional_results * PARALLEL_AI_ADDITIONAL_RESULT_COST
diff --git a/litellm/llms/parallel_ai/search/transformation.py b/litellm/llms/parallel_ai/search/transformation.py
index ea21d1153fe..bde7b7b86db 100644
--- a/litellm/llms/parallel_ai/search/transformation.py
+++ b/litellm/llms/parallel_ai/search/transformation.py
@@ -4,9 +4,13 @@ Calls Parallel AI's /v1/search endpoint to search the web.
 Parallel AI API Reference: https://docs.parallel.ai/api-reference/search/search
 """
 
+from collections.abc import Mapping, Sequence
+from types import MappingProxyType
 from typing import Final, TypedDict
 
 import httpx
+from pydantic import BaseModel, ConfigDict
+from typing_extensions import ReadOnly
 
 from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
 from litellm.llms.base_llm.search.transformation import (
@@ -14,9 +18,29 @@ from litellm.llms.base_llm.search.transformation import (
     SearchResponse,
     SearchResult,
 )
+from litellm.llms.parallel_ai.search.cost_calculator import PARALLEL_AI_USAGE_PARAM
 from litellm.secret_managers.main import get_secret_str
 
 
+class _ParallelAIV1SearchResult(BaseModel):
+    model_config = ConfigDict(extra="ignore")
+
+    url: str | None = None
+    title: str | None = None
+    publish_date: str | None = None
+    excerpts: Sequence[str] | None = None
+
+
+class _ParallelAIV1SearchResponse(BaseModel):
+    model_config = ConfigDict(extra="ignore")
+
+    search_id: str | None = None
+    session_id: str | None = None
+    results: Sequence[_ParallelAIV1SearchResult] = ()
+    usage: Sequence[Mapping[str, object]] | None = None
+    warnings: Sequence[Mapping[str, object]] | None = None
+
+
 class _ParallelAISourcePolicy(TypedDict, total=False):
     include_domains: list[str]
     exclude_domains: list[str]
@@ -27,10 +51,16 @@ class _ParallelAIExcerptSettings(TypedDict, total=False):
     max_chars_per_result: int
 
 
+class _ParallelAIFetchPolicy(TypedDict, total=False):
+    max_age_seconds: ReadOnly[int]
+    timeout_seconds: ReadOnly[float]
+    disable_cache_fallback: ReadOnly[bool]
+
+
 class _ParallelAIAdvancedSettings(TypedDict, total=False):
     source_policy: _ParallelAISourcePolicy
     excerpt_settings: _ParallelAIExcerptSettings
-    fetch_policy: dict
+    fetch_policy: _ParallelAIFetchPolicy
     location: str
     max_results: int
 
@@ -43,14 +73,14 @@ class ParallelAISearchRequest(TypedDict, total=False):
 
     search_queries: list[str]  # Required - at least one keyword search query
     objective: str  # Optional - natural-language description of search goal
-    mode: str  # Optional - 'turbo', 'basic', or 'advanced' (default 'advanced')
+    mode: str  # Optional - 'turbo', 'fast', 'basic', or 'advanced' (default 'advanced')
     max_chars_total: int  # Optional - upper bound on total excerpt characters
     session_id: str  # Optional - tracks calls across search/extract requests
     client_model: str  # Optional - model consuming the results
     advanced_settings: _ParallelAIAdvancedSettings
 
 
-LEGACY_PROCESSOR_TO_MODE: Final = {"base": "basic", "pro": "advanced"}
+LEGACY_PROCESSOR_TO_MODE: Final = MappingProxyType({"base": "basic", "pro": "advanced"})
 
 
 class ParallelAISearchConfig(BaseSearchConfig):
@@ -67,16 +97,16 @@ class ParallelAISearchConfig(BaseSearchConfig):
         api_base: str | None = None,
         **kwargs,
     ) -> dict:
-        api_key = self.resolve_server_api_key(
+        resolved_api_key: Final = self.resolve_server_api_key(
             caller_api_key=api_key,
             caller_api_base=api_base,
             key_env_vars=("PARALLEL_AI_API_KEY", "PARALLEL_API_KEY"),
             base_env_var="PARALLEL_AI_API_BASE",
             default_api_base=self.PARALLEL_AI_API_BASE,
         )
-        if not api_key:
+        if not resolved_api_key:
             raise ValueError("PARALLEL_API_KEY is not set. Set `PARALLEL_API_KEY` environment variable.")
-        headers["x-api-key"] = api_key
+        headers["x-api-key"] = resolved_api_key
         headers["Content-Type"] = "application/json"
         return headers
 
@@ -87,13 +117,12 @@ class ParallelAISearchConfig(BaseSearchConfig):
         data: dict | list[dict] | None = None,
         **kwargs,
     ) -> str:
-        api_base = api_base or get_secret_str("PARALLEL_AI_API_BASE") or self.PARALLEL_AI_API_BASE
+        resolved_api_base: Final = api_base or get_secret_str("PARALLEL_AI_API_BASE") or self.PARALLEL_AI_API_BASE
 
-        api_base = api_base.rstrip("/")
-        if not api_base.endswith("/v1/search"):
-            api_base = f"{api_base.removesuffix('/v1')}/v1/search"
-
-        return api_base
+        trimmed: Final = resolved_api_base.rstrip("/")
+        if trimmed.endswith("/v1/search"):
+            return trimmed
+        return f"{trimmed.removesuffix('/v1')}/v1/search"
 
     def transform_search_request(
         self,
@@ -109,14 +138,17 @@ class ParallelAISearchConfig(BaseSearchConfig):
                 - If string: maps to `search_queries` (single item) and `objective`
                 - If list: maps to `search_queries` (keyword queries)
             optional_params: Optional parameters for the request
-                - mode: Search mode ('turbo', 'basic', 'advanced'); defaults to 'basic'
+                - mode: Search mode ('turbo', 'fast', 'basic', 'advanced'); defaults to 'basic'
                 - processor: Legacy v1beta param; 'base' maps to mode 'basic', 'pro' to 'advanced'
                 - max_results: Maximum number of search results -> `advanced_settings.max_results`
-                - search_domain_filter: Domains to include -> `advanced_settings.source_policy.include_domains`
+                - search_domain_filter / include_domains: Domains to include -> `advanced_settings.source_policy.include_domains`
                 - exclude_domains: Domains to exclude -> `advanced_settings.source_policy.exclude_domains`
-                - country: ISO 3166-1 alpha-2 code -> `advanced_settings.location`
+                - after_date: RFC 3339 date (YYYY-MM-DD) -> `advanced_settings.source_policy.after_date`
+                - country / location: ISO 3166-1 alpha-2 code -> `advanced_settings.location`
                 - max_chars_per_result: -> `advanced_settings.excerpt_settings.max_chars_per_result`
-                - Any other params are passed through to the request body as-is
+                - fetch_policy: Cache vs live-fetch policy -> `advanced_settings.fetch_policy`
+                - Any other params (objective, max_chars_total, session_id, client_model, ...)
+                  are passed through to the request body as-is
 
         Returns:
             Dict with request data following the v1 search request spec
@@ -137,7 +169,7 @@ class ParallelAISearchConfig(BaseSearchConfig):
             mode = LEGACY_PROCESSOR_TO_MODE.get(processor, processor)
         # the v1 API defaults to 'advanced' when mode is omitted; default to 'basic'
         # instead to keep v1beta's default tier (processor 'base') and litellm's
-        # $0.004/query cost map entry for `parallel_ai/search` accurate
+        # cost map entry for `parallel_ai/search` accurate
         request_data["mode"] = mode or "basic"
 
         advanced_settings: Final[_ParallelAIAdvancedSettings] = {}
@@ -148,17 +180,29 @@ class ParallelAISearchConfig(BaseSearchConfig):
         if "country" in params:
             advanced_settings["location"] = params.pop("country")
 
+        if "location" in params:
+            advanced_settings["location"] = params.pop("location")
+
         if "max_chars_per_result" in params:
             advanced_settings["excerpt_settings"] = {"max_chars_per_result": params.pop("max_chars_per_result")}
 
+        if "fetch_policy" in params:
+            advanced_settings["fetch_policy"] = params.pop("fetch_policy")
+
         source_policy: Final[_ParallelAISourcePolicy] = {}
 
         if "search_domain_filter" in params:
             source_policy["include_domains"] = params.pop("search_domain_filter")
 
+        if "include_domains" in params:
+            source_policy["include_domains"] = params.pop("include_domains")
+
         if "exclude_domains" in params:
             source_policy["exclude_domains"] = params.pop("exclude_domains")
 
+        if "after_date" in params:
+            source_policy["after_date"] = params.pop("after_date")
+
         if source_policy:
             advanced_settings["source_policy"] = source_policy
 
@@ -170,9 +214,11 @@ class ParallelAISearchConfig(BaseSearchConfig):
         # unified-spec param with no v1 equivalent
         params.pop("max_tokens_per_page", None)
 
-        result_data: Final[dict] = dict(request_data)
-        result_data.update(params)
-        return result_data
+        # reserved for the provider's own reported usage, which prices the request;
+        # a caller-supplied value would otherwise set its own cost
+        params.pop(PARALLEL_AI_USAGE_PARAM, None)
+
+        return {**request_data, **params}
 
     def transform_search_response(
         self,
@@ -186,26 +232,49 @@ class ParallelAISearchConfig(BaseSearchConfig):
         Parallel AI -> LiteLLM mappings:
         - results[].title -> SearchResult.title
         - results[].url -> SearchResult.url
-        - results[].excerpts (array) -> SearchResult.snippet (joined string)
+        - results[].excerpts (array) -> SearchResult.snippet (joined string); the raw
+          array is preserved as an extra `excerpts` field on each result
         - results[].publish_date -> SearchResult.date
+        - search_id / session_id / warnings are preserved as extra fields on the
+          response; usage is preserved as `parallel_usage` (the `usage` name is
+          reserved for LiteLLM's token-usage object)
         """
-        response_json: Final = raw_response.json()
+        parsed: Final = _ParallelAIV1SearchResponse.model_validate(raw_response.json())
 
-        results: Final = []
-        for result in response_json.get("results", []):
-            excerpts = result.get("excerpts") or []
-            snippet = " ... ".join(excerpts) if excerpts else ""
+        # written unconditionally: leaving a caller-supplied value in place when the
+        # provider reports no usage would let the caller price its own request
+        logging_obj.optional_params = {
+            **logging_obj.optional_params,
+            PARALLEL_AI_USAGE_PARAM: parsed.usage,
+        }
 
-            search_result = SearchResult(
-                title=result.get("title") or "",
-                url=result.get("url") or "",
-                snippet=snippet,
-                date=result.get("publish_date"),
-                last_updated=None,
+        results: Final = tuple(
+            SearchResult.model_validate(
+                MappingProxyType(
+                    {
+                        "title": result.title or "",
+                        "url": result.url or "",
+                        "snippet": " ... ".join(result.excerpts or ()),
+                        "date": result.publish_date,
+                        "last_updated": None,
+                        "excerpts": result.excerpts or (),
+                    }
+                )
             )
-            results.append(search_result)
-
-        return SearchResponse(
-            results=results,
-            object="search",
+            for result in parsed.results
         )
+
+        extra_fields: Final = MappingProxyType(
+            {
+                key: value
+                for key, value in (
+                    ("search_id", parsed.search_id),
+                    ("session_id", parsed.session_id),
+                    ("parallel_usage", parsed.usage),
+                    ("warnings", parsed.warnings),
+                )
+                if value is not None
+            }
+        )
+
+        return SearchResponse.model_validate(MappingProxyType({"results": results, "object": "search", **extra_fields}))
diff --git a/litellm/model_prices_and_context_window_backup.json b/litellm/model_prices_and_context_window_backup.json
index 55618d9f772..a3cfb300ea6 100644
--- a/litellm/model_prices_and_context_window_backup.json
+++ b/litellm/model_prices_and_context_window_backup.json
@@ -38556,12 +38556,22 @@
         "source": "https://cloud.google.com/vertex-ai/generative-ai/docs/learn/models#foundation_models"
     },
     "parallel_ai/search": {
-        "input_cost_per_query": 0.004,
+        "input_cost_per_query": 0.005,
+        "litellm_provider": "parallel_ai",
+        "mode": "search"
+    },
+    "parallel_ai/search-fast": {
+        "input_cost_per_query": 0.001,
         "litellm_provider": "parallel_ai",
         "mode": "search"
     },
     "parallel_ai/search-pro": {
-        "input_cost_per_query": 0.009,
+        "input_cost_per_query": 0.005,
+        "litellm_provider": "parallel_ai",
+        "mode": "search"
+    },
+    "parallel_ai/search-turbo": {
+        "input_cost_per_query": 0.001,
         "litellm_provider": "parallel_ai",
         "mode": "search"
     },
diff --git a/litellm/router_utils/search_api_router.py b/litellm/router_utils/search_api_router.py
index d96defbbcd6..ab5ef5853c9 100644
--- a/litellm/router_utils/search_api_router.py
+++ b/litellm/router_utils/search_api_router.py
@@ -9,6 +9,7 @@ import random
 import traceback
 from collections.abc import Callable
 from functools import partial
+from types import MappingProxyType
 from typing import Any, Final
 
 from litellm._logging import verbose_router_logger
@@ -214,6 +215,15 @@ class SearchAPIRouter:
             api_key, api_base = SearchAPIRouter._resolve_search_provider_credentials(
                 tool_litellm_params=litellm_params,
             )
+            protected_params: Final = frozenset(("search_provider", "api_key", "api_base"))
+            search_params: Final = MappingProxyType(
+                {
+                    key: value
+                    for params in (litellm_params, kwargs)
+                    for key, value in params.items()
+                    if key not in protected_params and value is not None
+                }
+            )
 
             verbose_router_logger.debug("Selected search tool with provider: %s", search_provider)
 
@@ -222,7 +232,7 @@ class SearchAPIRouter:
                 search_provider=search_provider,
                 api_key=api_key,
                 api_base=api_base,
-                **kwargs,
+                **search_params,
             )
 
             return response
diff --git a/litellm/search/cost_calculator.py b/litellm/search/cost_calculator.py
index 84461115e8e..21f27075e0f 100644
--- a/litellm/search/cost_calculator.py
+++ b/litellm/search/cost_calculator.py
@@ -2,16 +2,37 @@
 Cost calculation for search providers.
 """
 
+from collections.abc import Mapping
+from types import MappingProxyType
 from typing import Final
 
+from pydantic import TypeAdapter, ValidationError
+
 from litellm.utils import get_model_info
 
+PROVIDER_USAGE_ADAPTER: Final[TypeAdapter[tuple[Mapping[str, object], ...]]] = TypeAdapter(
+    tuple[Mapping[str, object], ...]
+)
+EMPTY_OPTIONAL_PARAMS: Final[Mapping[str, object]] = MappingProxyType({})
+
+
+def _provider_usage(
+    optional_params: Mapping[str, object] | None,
+    usage_param: str,
+) -> tuple[Mapping[str, object], ...] | None:
+    params: Final = optional_params if optional_params is not None else EMPTY_OPTIONAL_PARAMS
+    raw_usage: Final[object] = params.get(usage_param)
+    try:
+        return PROVIDER_USAGE_ADAPTER.validate_python(raw_usage)
+    except ValidationError:
+        return None
+
 
 def search_provider_cost_per_query(
     model: str,
     custom_llm_provider: str | None = None,
     number_of_queries: int = 1,
-    optional_params: dict | None = None,
+    optional_params: Mapping[str, object] | None = None,
 ) -> tuple[float, float]:
     """
     Calculate cost for search-only providers.
@@ -28,6 +49,18 @@ def search_provider_cost_per_query(
     Returns:
         Tuple of (input_cost, output_cost) where output_cost is always 0.0
     """
+    if custom_llm_provider == "parallel_ai":
+        from litellm.llms.parallel_ai.search.cost_calculator import (
+            PARALLEL_AI_USAGE_PARAM,
+            parallel_ai_search_cost,
+        )
+
+        input_cost: Final = parallel_ai_search_cost(
+            optional_params=optional_params if optional_params is not None else EMPTY_OPTIONAL_PARAMS,
+            usage=_provider_usage(optional_params, PARALLEL_AI_USAGE_PARAM),
+        )
+        return (input_cost, 0.0)
+
     model_info: Final = get_model_info(model=model, custom_llm_provider=custom_llm_provider)
 
     # Check for tiered pricing (e.g., Exa AI based on max_results)
diff --git a/model_prices_and_context_window.json b/model_prices_and_context_window.json
index 55618d9f772..a3cfb300ea6 100644
--- a/model_prices_and_context_window.json
+++ b/model_prices_and_context_window.json
@@ -38556,12 +38556,22 @@
         "source": "https://cloud.google.com/vertex-ai/generative-ai/docs/learn/models#foundation_models"
     },
     "parallel_ai/search": {
-        "input_cost_per_query": 0.004,
+        "input_cost_per_query": 0.005,
+        "litellm_provider": "parallel_ai",
+        "mode": "search"
+    },
+    "parallel_ai/search-fast": {
+        "input_cost_per_query": 0.001,
         "litellm_provider": "parallel_ai",
         "mode": "search"
     },
     "parallel_ai/search-pro": {
-        "input_cost_per_query": 0.009,
+        "input_cost_per_query": 0.005,
+        "litellm_provider": "parallel_ai",
+        "mode": "search"
+    },
+    "parallel_ai/search-turbo": {
+        "input_cost_per_query": 0.001,
         "litellm_provider": "parallel_ai",
         "mode": "search"
     },
diff --git a/tests/router_unit_tests/test_router_helper_utils.py b/tests/router_unit_tests/test_router_helper_utils.py
index dcd2e9edf7b..7dbac243d55 100644
--- a/tests/router_unit_tests/test_router_helper_utils.py
+++ b/tests/router_unit_tests/test_router_helper_utils.py
@@ -2294,6 +2294,7 @@ def search_tools():
                 "search_provider": "perplexity",
                 "api_key": "test-api-key",
                 "api_base": "https://api.perplexity.ai",
+                "mode": "turbo",
             },
         },
         {
@@ -2302,6 +2303,7 @@ def search_tools():
                 "search_provider": "perplexity",
                 "api_key": "test-api-key-2",
                 "api_base": "https://api.perplexity.ai",
+                "mode": "turbo",
             },
         },
     ]
@@ -2393,6 +2395,7 @@ async def test_asearch_with_fallbacks_helper(search_tools):
         assert "search_provider" in kwargs
         assert kwargs["search_provider"] == "perplexity"
         assert "api_key" in kwargs
+        assert kwargs["mode"] == "turbo"
         assert kwargs["query"] == "helper test query"
         return mock_response
 
diff --git a/tests/test_litellm/llms/parallel_ai/test_parallel_ai_search.py b/tests/test_litellm/llms/parallel_ai/test_parallel_ai_search.py
index 8a9ae4dae6d..62b4d003b45 100644
--- a/tests/test_litellm/llms/parallel_ai/test_parallel_ai_search.py
+++ b/tests/test_litellm/llms/parallel_ai/test_parallel_ai_search.py
@@ -2,6 +2,7 @@
 Tests for Parallel AI Search API integration (v1 endpoint).
 """
 
+import json
 from unittest.mock import AsyncMock, MagicMock, patch
 
 import pytest
@@ -30,13 +31,41 @@ MOCK_V1_RESPONSE = {
 }
 
 
-def _mock_response():
+def _mock_response(payload=None):
     mock_response = MagicMock()
     mock_response.status_code = 200
-    mock_response.json.return_value = MOCK_V1_RESPONSE
+    mock_response.json.return_value = payload if payload is not None else MOCK_V1_RESPONSE
     return mock_response
 
 
+@pytest.fixture
+def httpx_transport(monkeypatch):
+    monkeypatch.setattr(  # test-quality-ok: respx needs HTTPX enabled to fake the provider HTTP boundary.
+        litellm,
+        "disable_aiohttp_transport",
+        True,
+    )
+    litellm.in_memory_llm_clients_cache.flush_cache()
+    yield
+    litellm.in_memory_llm_clients_cache.flush_cache()
+
+
+@pytest.fixture
+def bundled_cost_map(monkeypatch):
+    """Price lookups against the bundled cost map.
+
+    litellm caches model-info lookups, so swapping ``model_cost`` only takes
+    effect once those caches are invalidated -- on the way in and back out.
+    """
+    from litellm.utils import _invalidate_model_cost_lowercase_map
+
+    monkeypatch.setattr(litellm, "model_cost", litellm.get_model_cost_map(url=""))
+    _invalidate_model_cost_lowercase_map()
+    yield
+    monkeypatch.undo()
+    _invalidate_model_cost_lowercase_map()
+
+
 class TestParallelAISearch:
     @pytest.fixture(autouse=True)
     def _set_api_key(self, monkeypatch):
@@ -135,9 +164,7 @@ class TestParallelAISearch:
             json_data = mock_post.call_args.kwargs.get("json")
             assert json_data["mode"] == "basic"
 
-    @pytest.mark.parametrize(
-        "processor,expected_mode", [("base", "basic"), ("pro", "advanced")]
-    )
+    @pytest.mark.parametrize("processor,expected_mode", [("base", "basic"), ("pro", "advanced")])
     @pytest.mark.asyncio
     async def test_legacy_processor_maps_to_mode(self, processor, expected_mode):
         with patch(
@@ -222,9 +249,7 @@ class TestParallelAISearch:
                 "arxiv.org",
                 "nature.com",
             ]
-            assert advanced_settings["source_policy"]["exclude_domains"] == [
-                "reddit.com"
-            ]
+            assert advanced_settings["source_policy"]["exclude_domains"] == ["reddit.com"]
             assert advanced_settings["excerpt_settings"]["max_chars_per_result"] == 1500
 
             assert "max_results" not in json_data
@@ -306,10 +331,7 @@ class TestParallelAISearch:
             )
 
             call_args = mock_post.call_args
-            assert (
-                call_args.kwargs["url"]
-                == "https://proxy.internal.example.com/v1/search"
-            )
+            assert call_args.kwargs["url"] == "https://proxy.internal.example.com/v1/search"
 
     @pytest.mark.asyncio
     async def test_caller_api_base_without_key_is_refused(self, monkeypatch):
@@ -338,3 +360,147 @@ class TestParallelAISearch:
                 query="AI developments",
                 search_provider="parallel_ai",
             )
+
+    @pytest.mark.asyncio
+    async def test_flat_source_and_fetch_params_nest_under_advanced_settings(self, respx_mock, httpx_transport):
+        route = respx_mock.post("https://api.parallel.ai/v1/search").respond(json=MOCK_V1_RESPONSE)
+
+        await litellm.asearch(
+            query="AI developments",
+            search_provider="parallel_ai",
+            objective="find peer-reviewed AI research",
+            include_domains=["arxiv.org"],
+            after_date="2026-01-01",
+            location="gb",
+            fetch_policy={"max_age_seconds": 600, "disable_cache_fallback": True},
+            client_model="claude-fable-5",
+        )
+
+        json_data = json.loads(route.calls[0].request.content)
+        assert json_data["objective"] == "find peer-reviewed AI research"
+        assert json_data["client_model"] == "claude-fable-5"
+
+        advanced_settings = json_data["advanced_settings"]
+        assert advanced_settings["location"] == "gb"
+        assert advanced_settings["fetch_policy"] == {
+            "max_age_seconds": 600,
+            "disable_cache_fallback": True,
+        }
+        assert advanced_settings["source_policy"]["include_domains"] == ["arxiv.org"]
+        assert advanced_settings["source_policy"]["after_date"] == "2026-01-01"
+
+        assert "include_domains" not in json_data
+        assert "after_date" not in json_data
+        assert "location" not in json_data
+        assert "fetch_policy" not in json_data
+
+    @pytest.mark.asyncio
+    async def test_response_preserves_raw_parallel_fields(self, respx_mock, httpx_transport):
+        respx_mock.post("https://api.parallel.ai/v1/search").respond(json=MOCK_V1_RESPONSE)
+
+        response = await litellm.asearch(
+            query="AI developments",
+            search_provider="parallel_ai",
+        )
+
+        dumped = response.model_dump()
+        assert dumped["search_id"] == "search_abc123"
+        assert dumped["session_id"] == "session_xyz"
+        assert dumped["parallel_usage"] == [{"name": "search_advanced", "count": 1}]
+
+        first = response.results[0].model_dump()
+        assert first["excerpts"] == ["First excerpt.", "Second excerpt."]
+
+    @pytest.mark.asyncio
+    async def test_response_normalizes_null_result_fields(self, respx_mock, httpx_transport):
+        response_payload = {
+            **MOCK_V1_RESPONSE,
+            "results": [{"url": None, "title": None, "publish_date": None, "excerpts": None}],
+        }
+        respx_mock.post("https://api.parallel.ai/v1/search").respond(json=response_payload)
+
+        response = await litellm.asearch(
+            query="AI developments",
+            search_provider="parallel_ai",
+        )
+
+        assert len(response.results) == 1
+        result = response.results[0]
+        assert result.url == ""
+        assert result.title == ""
+        assert result.snippet == ""
+        assert result.date is None
+        assert result.model_dump()["excerpts"] == ()
+
+    @pytest.mark.parametrize(
+        "mode,usage,max_results,expected_cost",
+        [
+            ("turbo", [{"name": "sku_search", "count": 1}], None, 0.001),
+            ("fast", [{"name": "sku_search", "count": 1}], None, 0.001),
+            ("basic", [{"name": "sku_search", "count": 1}], None, 0.005),
+            ("advanced", [{"name": "sku_search", "count": 1}], None, 0.005),
+            (
+                "basic",
+                [
+                    {"name": "sku_search", "count": 1},
+                    {"name": "sku_search_additional_results", "count": 2},
+                ],
+                20,
+                0.007,
+            ),
+            ("basic", None, 20, 0.015),
+        ],
+    )
+    @pytest.mark.asyncio
+    async def test_search_cost_uses_mode_and_provider_usage(
+        self, mode, usage, max_results, expected_cost, bundled_cost_map, respx_mock, httpx_transport
+    ):
+        response_payload = {**MOCK_V1_RESPONSE, "usage": usage}
+        respx_mock.post("https://api.parallel.ai/v1/search").respond(json=response_payload)
+
+        response = await litellm.asearch(
+            query="AI developments",
+            search_provider="parallel_ai",
+            mode=mode,
+            max_results=max_results,
+        )
+
+        assert response._hidden_params["response_cost"] == pytest.approx(expected_cost)
+
+    @pytest.mark.asyncio
+    async def test_search_cost_treats_keyword_queries_as_one_request(
+        self, bundled_cost_map, respx_mock, httpx_transport
+    ):
+        response_payload = {
+            **MOCK_V1_RESPONSE,
+            "usage": [{"name": "sku_search", "count": 1}],
+        }
+        respx_mock.post("https://api.parallel.ai/v1/search").respond(json=response_payload)
+
+        response = await litellm.asearch(
+            query=["AI developments", "machine learning trends"],
+            search_provider="parallel_ai",
+            mode="basic",
+        )
+
+        assert response._hidden_params["response_cost"] == pytest.approx(0.005)
+
+    @pytest.mark.asyncio
+    async def test_caller_cannot_supply_provider_usage(self, bundled_cost_map, respx_mock, httpx_transport):
+        """`_parallel_ai_usage` prices the request, so a caller must not be able to set it.
+
+        The provider reports no usage here, which is the case where a caller-supplied
+        value would otherwise survive into the cost calculation.
+        """
+        response_payload = {k: v for k, v in MOCK_V1_RESPONSE.items() if k != "usage"}
+        route = respx_mock.post("https://api.parallel.ai/v1/search").respond(json=response_payload)
+
+        response = await litellm.asearch(
+            query="AI developments",
+            search_provider="parallel_ai",
+            mode="basic",
+            _parallel_ai_usage=[{"name": "sku_search", "count": 0}],
+        )
+
+        assert response._hidden_params["response_cost"] == pytest.approx(0.005)
+        assert "_parallel_ai_usage" not in json.loads(route.calls[0].request.content)
diff --git a/tests/test_litellm/llms/parallel_ai/test_parallel_ai_search_gateway.py b/tests/test_litellm/llms/parallel_ai/test_parallel_ai_search_gateway.py
new file mode 100644
index 00000000000..72c69fc622c
--- /dev/null
+++ b/tests/test_litellm/llms/parallel_ai/test_parallel_ai_search_gateway.py
@@ -0,0 +1,191 @@
+"""Gateway coverage for Parallel AI Search."""
+
+from __future__ import annotations
+
+from collections.abc import Iterator
+from typing import Final
+from unittest.mock import AsyncMock
+
+import httpx
+import pytest
+from fastapi.testclient import TestClient
+
+import litellm
+from litellm import Router
+from litellm.integrations.websearch_interception.handler import (
+    WebSearchInterceptionLogger,
+)
+from litellm.llms.custom_httpx.http_handler import AsyncHTTPHandler
+from litellm.proxy import proxy_server
+from litellm.proxy._types import UserAPIKeyAuth
+from litellm.proxy.auth.user_api_key_auth import user_api_key_auth
+from litellm.types.utils import LlmProviders
+
+PARALLEL_SEARCH_URL: Final = "https://api.parallel.ai/v1/search"
+
+
+@pytest.fixture
+def client() -> TestClient:
+    return TestClient(proxy_server.app, raise_server_exceptions=False)
+
+
+@pytest.fixture
+def auth_as() -> Iterator[None]:
+    async def _authorized_request() -> UserAPIKeyAuth:
+        return UserAPIKeyAuth(
+            api_key="hashed-sk-test",
+            user_id="parallel-test-user",
+        )
+
+    previous: Final = proxy_server.app.dependency_overrides.get(user_api_key_auth)
+    proxy_server.app.dependency_overrides[user_api_key_auth] = _authorized_request
+    try:
+        yield
+    finally:
+        if previous is None:
+            proxy_server.app.dependency_overrides.pop(user_api_key_auth, None)
+        else:
+            proxy_server.app.dependency_overrides[user_api_key_auth] = previous
+
+
+def _parallel_search_body() -> dict[str, object]:
+    return {
+        "search_id": "search_parallel_gateway",
+        "results": [
+            {
+                "url": "https://example.com/parallel",
+                "title": "Parallel result",
+                "publish_date": "2026-08-13",
+                "excerpts": ["First excerpt", "Second excerpt"],
+            }
+        ],
+        "usage": [{"name": "sku_search", "count": 1}],
+    }
+
+
+def _parallel_router(mode: str = "turbo") -> Router:
+    return Router(
+        model_list=[],
+        search_tools=[
+            {
+                "search_tool_name": "parallel-search",
+                "litellm_params": {
+                    "search_provider": "parallel_ai",
+                    "api_key": "parallel-search-key",
+                    "mode": mode,
+                },
+            }
+        ],
+        num_retries=0,
+    )
+
+
+def _mock_async_post(
+    monkeypatch,
+    *,
+    url: str,
+    response_body: dict[str, object],
+) -> AsyncMock:
+    response = httpx.Response(
+        status_code=200,
+        json=response_body,
+        request=httpx.Request("POST", url),
+    )
+    mock_post = AsyncMock(return_value=response)
+    monkeypatch.setattr(AsyncHTTPHandler, "post", mock_post)
+    return mock_post
+
+
+def test_parallel_search_gateway_route(client, auth_as, monkeypatch):
+    """The named search route selects its configured Parallel Search tool.
+
+    The tool-level `mode` must survive the router hop, so the upstream request
+    is sent as `turbo` rather than falling back to the adapter default.
+    """
+    monkeypatch.setattr(litellm, "model_cost", litellm.get_model_cost_map(url=""))
+    monkeypatch.setattr(proxy_server, "llm_router", _parallel_router())
+    mock_post = _mock_async_post(
+        monkeypatch,
+        url=PARALLEL_SEARCH_URL,
+        response_body=_parallel_search_body(),
+    )
+
+    response = client.post(
+        "/v1/search/parallel-search",
+        json={"query": "Parallel AI news", "max_results": 3},
+    )
+
+    assert response.status_code == 200, response.text
+    assert response.json()["results"] == [
+        {
+            "title": "Parallel result",
+            "url": "https://example.com/parallel",
+            "snippet": "First excerpt ... Second excerpt",
+            "date": "2026-08-13",
+            "last_updated": None,
+            "excerpts": ["First excerpt", "Second excerpt"],
+        }
+    ]
+
+    request_kwargs = mock_post.await_args.kwargs
+    assert request_kwargs["url"] == PARALLEL_SEARCH_URL
+    assert request_kwargs["headers"]["x-api-key"] == "parallel-search-key"
+    assert request_kwargs["json"] == {
+        "objective": "Parallel AI news",
+        "search_queries": ["Parallel AI news"],
+        "mode": "turbo",
+        "advanced_settings": {"max_results": 3},
+    }
+
+
+@pytest.mark.asyncio
+async def test_web_search_interception_executes_parallel_search(monkeypatch):
+    """An intercepted web-search call uses the configured Parallel Search tool."""
+    monkeypatch.setattr(litellm, "model_cost", litellm.get_model_cost_map(url=""))
+    monkeypatch.setattr(proxy_server, "llm_router", _parallel_router(mode="fast"))
+    mock_post = _mock_async_post(
+        monkeypatch,
+        url=PARALLEL_SEARCH_URL,
+        response_body=_parallel_search_body(),
+    )
+    logger = WebSearchInterceptionLogger(
+        enabled_providers=[LlmProviders.OPENAI],
+        search_tool_name="parallel-search",
+    )
+
+    plan = await logger.async_build_responses_agentic_loop_plan(
+        tools={
+            "tool_calls": [
+                {
+                    "id": "fc_parallel",
+                    "call_id": "fc_parallel",
+                    "type": "function_call",
+                    "name": "litellm_web_search",
+                    "arguments": '{"query":"Parallel AI news"}',
+                    "input": {"query": "Parallel AI news"},
+                }
+            ]
+        },
+        model="gpt-5",
+        messages=[{"role": "user", "content": "Research Parallel"}],
+        response=None,
+        optional_params={"tools": [{"type": "function", "name": "litellm_web_search"}]},
+        logging_obj=None,
+        stream=False,
+        kwargs={"custom_llm_provider": "openai"},
+    )
+
+    assert plan.run_agentic_loop is True
+    assert plan.request_patch is not None
+    assert plan.request_patch.messages[-1] == {
+        "type": "function_call_output",
+        "call_id": "fc_parallel",
+        "output": (
+            "Title: Parallel result\nURL: https://example.com/parallel\nSnippet: First excerpt ... Second excerpt"
+        ),
+    }
+
+    request_kwargs = mock_post.await_args.kwargs
+    assert request_kwargs["url"] == PARALLEL_SEARCH_URL
+    assert request_kwargs["headers"]["x-api-key"] == "parallel-search-key"
+    assert request_kwargs["json"]["mode"] == "fast"