Merge remote-tracking branch 'origin/litellm_internal_staging' into litellm_fix_agent_mcp_grants

basedpyright-code-budget.json

@@ -117,7 +117,7 @@
     "limit": 111
   },
   "reportUnnecessaryComparison": {
-    "limit": 695
+    "limit": 692
   },
   "reportUnnecessaryContains": {
     "limit": 5

litellm/llms/base_llm/guardrail_translation/base_translation.py

@@ -155,8 +155,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.

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 stream_item_field(item, "type") == "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:

litellm/llms/openai/chat/guardrail_translation/handler.py

@@ -14,9 +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 (
@@ -24,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,
@@ -32,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
@@ -49,7 +56,10 @@ from litellm.types.utils import (
 if TYPE_CHECKING:
     from fastapi import HTTPException
 
-    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
 
@@ -1005,3 +1015,129 @@ class OpenAIChatCompletionsHandler(BaseTranslation):
                                 else:
                                     # 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",
+        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 _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]:
+    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,
+    )

litellm/llms/openai/responses/guardrail_translation/handler.py

@@ -28,12 +28,16 @@ Output: response.output is List[GenericResponseOutputItem] where each has:
     - text: str
 """
 
-from collections.abc import Sequence
+import time
+import uuid
+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
@@ -41,17 +45,33 @@ 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,
     ErrorEvent,
     ErrorEventError,
     OpenAIMcpServerTool,
+    OutputItemAddedEvent,
+    OutputItemDoneEvent,
+    OutputTextDeltaEvent,
+    OutputTextDoneEvent,
+    ResponseAPIUsage,
+    ResponseCompletedEvent,
+    ResponsesAPIResponse,
     ResponsesAPIStreamEvents,
+    ResponsesAPIStreamingResponse,
 )
 from litellm.types.responses.main import (
     GenericResponseOutputItem,
@@ -63,11 +83,13 @@ from litellm.types.utils import GenericGuardrailAPIInputs
 if TYPE_CHECKING:
     from fastapi import HTTPException
 
-    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):
@@ -865,3 +887,331 @@ 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: 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.
+        """
+        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, serialize_as_any=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
+        part: Final[_BlockedContentPart] = {"type": "output_text", "text": exc.message, "annotations": ()}
+        done_part: Final[_BlockedDoneContentPart] = {
+            "type": "output_text",
+            "text": exc.message,
+            "annotations": (),
+            "logprobs": None,
+        }
+        return (
+            *_open_item_closing_events(responses_so_far),
+            OutputItemAddedEvent(
+                type=ResponsesAPIStreamEvents.OUTPUT_ITEM_ADDED,
+                output_index=output_index,
+                item=item,
+            ),
+            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,
+            ),
+            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 _BlockedItemPayload(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[GenericResponseOutputItem, ...]]
+    status: ReadOnly[str]
+    usage: ReadOnly[ResponseAPIUsage]
+
+
+def _blocked_output_item(exc: "ModifyResponseException") -> GenericResponseOutputItem:
+    payload: Final[_BlockedItemPayload] = {
+        "type": "message",
+        "id": f"msg_{uuid.uuid4()}",
+        "status": "completed",
+        "role": "assistant",
+        "content": ({"type": "output_text", "text": exc.message, "annotations": ()},),
+    }
+    return GenericResponseOutputItem.model_validate(payload)
+
+
+def _blocked_response(
+    exc: "ModifyResponseException",
+    response_id: str,
+    model: str,
+    output_item: GenericResponseOutputItem | 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
+
+
+@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
+    payload: object
+
+
+def _open_item_state(responses_so_far: Sequence[object]) -> _OpenItemState | None:
+    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
+        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]
+    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")
+    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),
+        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. 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,
+        "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,
+    )

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

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}))

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"
     },

litellm/responses/streaming_iterator.py

@@ -1023,7 +1023,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,
@@ -1090,7 +1090,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,
@@ -1274,10 +1274,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()

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

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)

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"
     },

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,

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
 

tests/test_litellm/llms/openai/chat/guardrail_translation/test_openai_guardrail_handler.py

@@ -1559,3 +1559,87 @@ 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}
+
+
+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

tests/test_litellm/llms/openai/responses/test_openai_responses_guardrail_handler.py

@@ -1321,3 +1321,219 @@ 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.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_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,
+            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_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 = [
+            {"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."

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)

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"

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: ") :])

tests/test_litellm/proxy/guardrails/guardrail_hooks/unified_guardrails/test_openai_streaming_block.py

@@ -0,0 +1,327 @@
+"""
+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, Dict, Literal, Optional, Tuple, Union
+
+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."
+
+JsonPayload = Dict[str, object]
+StreamChunk = Union[ModelResponseStream, JsonPayload, bytes]
+
+
+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,
+        )
+
+
+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",
+        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[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}
+    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[Union[ModelResponseStream, JsonPayload], None],
+    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")
+        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
+
+    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": [guardrail.guardrail_name]},
+    }
+
+    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: 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: 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}"
+
+
+@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():
+    """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
+    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 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 = [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[-1] == "response.completed"
+
+    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
+
+
+@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}

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()

type-discipline-budget.json

@@ -1,9 +1,9 @@
 {
   "LIT001": {
-    "limit": 22365
+    "limit": 22362
   },
   "LIT002": {
-    "limit": 26777
+    "limit": 26774
   },
   "LIT003": {
     "limit": 269
@@ -27,7 +27,7 @@
     "limit": 0
   },
   "LIT010": {
-    "limit": 16507
+    "limit": 16499
   },
   "LIT011": {
     "limit": 5535