fix(aicore): prevent content filter bypass in multi-module and streaming modes

src/sap_cloud_sdk/aicore/completion.py

@@ -40,9 +40,15 @@
         ...
 
 These are intentionally thin — every other keyword argument is forwarded
-verbatim to ``litellm.completion`` / ``litellm.acompletion``, including
-``stream=True`` (the streaming iterator is returned unchanged). No
-telemetry is recorded here; the wrappers fire on every LLM call and a
+verbatim to ``litellm.completion`` / ``litellm.acompletion``. When output
+filtering is active, ``stream=True`` is not supported — the filtering
+transport patch raises :class:`ContentFilteredError` upfront before the
+request is sent. This applies universally to all callers (including
+``ChatLiteLLM`` and LangGraph agents). Input filtering works with streaming
+— filter rejections surface as :class:`ContentFilteredError` through the
+stream iterator regardless.
+
+No telemetry is recorded here; the wrappers fire on every LLM call and a
 counter at this level would be both noisy and uninformative about
 adoption of the SDK.
 """

src/sap_cloud_sdk/aicore/filtering/_patch.py

@@ -26,6 +26,24 @@
 back into :class:`ContentFilteredError` before it reaches caller code; the
 :func:`_parse_input_filter_error` helper in :mod:`.filters` does the JSON
 parsing.
+
+Streaming + output filtering
+----------------------------
+The SAP AI Core Orchestration v2 API supports output filtering with streaming
+— the server filters each chunk and signals a rejection via an in-stream error
+event (no ``[DONE]``).  LiteLLM surfaces this as a ``ValueError`` raised from
+inside its streaming iterator, **not** via ``transform_response``.
+
+The SDK cannot intercept that iterator-level error universally: agent
+frameworks (``ChatLiteLLM``, LangGraph, etc.) call ``litellm.completion``
+directly and never pass through ``sap_cloud_sdk.aicore.completion``, so the
+``ValueError`` would surface as an untyped exception rather than
+:class:`ContentFilteredError`.
+
+Until the SDK wraps the streaming iterator for all call paths,
+``transform_request`` raises :class:`ContentFilteredError` upfront when both
+streaming and output filtering are active. This guarantee is universal because
+``transform_request`` runs at the LiteLLM transport layer for all callers.
 """
 
 from __future__ import annotations
@@ -58,6 +76,9 @@ def transform_request(
         litellm_params: dict,
         headers: dict,
     ) -> dict:
+        # Capture stream flag before super() pops it from optional_params.
+        is_streaming = bool(optional_params.get("stream"))
+
         body = super().transform_request(
             model=model,
             messages=messages,
@@ -73,10 +94,24 @@ def transform_request(
         if not filtering_dict:
             return body
 
+        # Streaming + output filtering: the API supports it, but mid-stream
+        # filter rejections surface as ValueError inside LiteLLM's iterator —
+        # a path the SDK cannot intercept universally (agent frameworks call
+        # litellm directly, bypassing sap_cloud_sdk.aicore.completion).
+        # Block upfront until iterator wrapping is implemented for all callers.
+        if _active_cfg.output_filtering is not None and (
+            is_streaming or body.get("config", {}).get("stream", {}).get("enabled")
+        ):
+            raise ContentFilteredError(
+                direction="output",
+                details={},
+                request_id=None,
+            )
+
         modules = body["config"]["modules"]
         if isinstance(modules, list):
-            # Fallback mode (list of configs) — inject into primary config only.
-            modules[0]["filtering"] = filtering_dict
+            for m in modules:
+                m["filtering"] = filtering_dict
         else:
             modules["filtering"] = filtering_dict
 

src/sap_cloud_sdk/aicore/filtering/exceptions.py

@@ -20,8 +20,12 @@ class ContentFilteredError(OrchestrationError):
     Attributes:
         direction: ``"input"`` if the user prompt was rejected (HTTP 4xx,
             ``error.location`` startswith ``"Filtering Module - Input Filter"``),
-            or ``"output"`` if the model response was rejected (HTTP 200,
-            ``finish_reason == "content_filter"``).
+            or ``"output"`` if the model response was rejected. In non-streaming
+            responses this is signalled by ``finish_reason == "content_filter"``
+            (HTTP 200). When output filtering is active, streaming is blocked
+            upfront by the SDK (raised before the request is sent) because
+            mid-stream filter rejections cannot yet be intercepted uniformly
+            for all callers.
         details: Severity scalars from ``intermediate_results.{input,output}_filtering.data``.
             Safe to log; does NOT include raw prompt or completion content.
         request_id: ``error.request_id`` (input) or top-level ``request_id`` (output).

src/sap_cloud_sdk/aicore/user-guide.md

@@ -360,10 +360,17 @@ except ContentFilteredError as e:
     return "Your request was blocked by content safety policy."
 ```
 
-The wrapper forwards every argument verbatim to `litellm.completion`
-(including `stream=True`), and only intercepts the wrapped
-`APIConnectionError` shape that LiteLLM produces for input-filter
-rejections. All other exceptions surface unchanged.
+The wrapper forwards every argument verbatim to `litellm.completion`,
+including `stream=True`. When output filtering is active, `stream=True` is
+not currently supported by the SDK — a `ContentFilteredError(direction="output")`
+is raised before the request is sent. The SAP AI Core API does support output
+filtering with streaming, but mid-stream filter rejections surface as an untyped
+error inside LiteLLM's iterator and cannot yet be intercepted uniformly for all
+callers (agent frameworks such as `ChatLiteLLM` and LangGraph call `litellm`
+directly, bypassing the SDK's exception translation). Input filtering works
+with streaming — filter rejections surface as `ContentFilteredError` regardless.
+
+All other exceptions surface unchanged.
 
 `ContentFilteredError` exposes three attributes — `direction`, `details`,
 `request_id`. The `details` field contains severity scalars from the server,
@@ -444,21 +451,31 @@ print(response.choices[0].message.content)
 
 ### Streaming Responses
 
+`stream=True` is supported when no output filtering is configured. If output
+filtering is active, streaming is not currently supported by the SDK —
+`ContentFilteredError(direction="output")` is raised before the request is
+sent. The underlying API supports output filtering with streaming, but the SDK
+cannot yet translate mid-stream filter rejections into `ContentFilteredError`
+uniformly for all callers (agent frameworks call `litellm` directly and would
+receive an untyped error). Input-only filtering works with streaming.
+
 ```python
-from sap_cloud_sdk.aicore import set_aicore_config
-from litellm import completion
+from sap_cloud_sdk.aicore import ContentFilteredError, completion, set_aicore_config
 
 set_aicore_config()
 
-response = completion(
-    model="sap/gpt-4",
-    messages=[{"role": "user", "content": "Tell me a story"}],
-    stream=True,
-)
+try:
+    response = completion(
+        model="sap/gpt-4",
+        messages=[{"role": "user", "content": "Tell me a story"}],
+        stream=True,
+    )
 
-for chunk in response:
-    if chunk.choices[0].delta.content:
-        print(chunk.choices[0].delta.content, end="")
+    for chunk in response:
+        if chunk.choices[0].delta.content:
+            print(chunk.choices[0].delta.content, end="")
+except ContentFilteredError as e:
+    print(f"Blocked ({e.direction}): {e.details}")
 ```
 
 ### Using Different Models

tests/aicore/filtering/unit/test_filtering.py

@@ -133,3 +133,60 @@ def test_disable_when_never_enabled(self, monkeypatch):
         disable_filtering()
         assert litellm.GenAIHubOrchestrationConfig is _ORIGINAL_CONFIG
         assert _filters_mod._active_cfg is None
+
+
+class TestTransformRequestMultiModule:
+    """Regression tests: filtering must be injected into every fallback module, not just modules[0]."""
+
+    def _make_cfg(self) -> ContentFiltering:
+        return ContentFiltering(
+            input_filtering=InputFiltering(filters=[AzureContentFilter()])
+        )
+
+    def _call_transform_request(self, modules_value):
+        """Call the filtering injection logic directly, bypassing the LiteLLM base class."""
+        from unittest.mock import patch
+
+        body = {"config": {"modules": modules_value}}
+        instance = FilteringOrchestrationConfig()
+
+        with patch(
+            "sap_cloud_sdk.aicore.filtering._patch.GenAIHubOrchestrationConfig.transform_request",
+            return_value=body,
+        ):
+            return instance.transform_request(
+                model="sap/x",
+                messages=[],
+                optional_params={},
+                litellm_params={},
+                headers={},
+            )
+
+    def test_single_module_dict_gets_filtering(self, monkeypatch):
+        _clear_aicore_env(monkeypatch)
+        set_filtering(self._make_cfg())
+        modules = {"model_id": "primary-llm"}
+        result = self._call_transform_request(modules)
+        assert "filtering" in result["config"]["modules"]
+
+    def test_multi_module_list_all_get_filtering(self, monkeypatch):
+        _clear_aicore_env(monkeypatch)
+        set_filtering(self._make_cfg())
+        modules = [
+            {"model_id": "primary-llm"},
+            {"model_id": "fallback-llm-1"},
+            {"model_id": "fallback-llm-2"},
+        ]
+        result = self._call_transform_request(modules)
+        for i, m in enumerate(result["config"]["modules"]):
+            assert "filtering" in m, f"modules[{i}] missing filtering block"
+
+    def test_multi_module_filtering_dict_is_identical_across_modules(self, monkeypatch):
+        _clear_aicore_env(monkeypatch)
+        set_filtering(self._make_cfg())
+        modules = [{"model_id": "a"}, {"model_id": "b"}]
+        result = self._call_transform_request(modules)
+        assert (
+            result["config"]["modules"][0]["filtering"]
+            == result["config"]["modules"][1]["filtering"]
+        )

tests/aicore/unit/test_completion.py

@@ -23,7 +23,7 @@
 
 import asyncio
 import json
-from unittest.mock import MagicMock, call, patch
+from unittest.mock import MagicMock, patch
 
 import litellm
 import pytest
@@ -275,6 +275,162 @@ def test_non_auth_error_does_not_trigger_reload(self):
         mock_reload.assert_not_called()
 
 
+# ---------------------------------------------------------------------------
+# Streaming guard — stream=True must not bypass output filtering
+# ---------------------------------------------------------------------------
+
+
+class TestStreamingGuard:
+    """stream=True is blocked upfront when output filtering is active.
+    Input-only filtering: stream=True passes through unmodified.
+    """
+
+    @pytest.fixture(autouse=True)
+    def restore_filtering(self):
+        from sap_cloud_sdk.aicore.filtering._patch import _install
+        _install(None)
+        yield
+        _install(None)
+
+    def _activate_input_only(self):
+        from sap_cloud_sdk.aicore.filtering import (
+            AzureContentFilter, ContentFiltering, InputFiltering, set_filtering,
+        )
+        set_filtering(ContentFiltering(
+            input_filtering=InputFiltering(filters=[AzureContentFilter()])
+        ))
+
+    def _activate_output(self):
+        from sap_cloud_sdk.aicore.filtering import (
+            AzureContentFilter, ContentFiltering, OutputFiltering, set_filtering,
+        )
+        set_filtering(ContentFiltering(
+            output_filtering=OutputFiltering(filters=[AzureContentFilter()])
+        ))
+
+    def _activate_input_and_output(self):
+        from sap_cloud_sdk.aicore.filtering import (
+            AzureContentFilter, ContentFiltering, InputFiltering, OutputFiltering,
+            set_filtering,
+        )
+        set_filtering(ContentFiltering(
+            input_filtering=InputFiltering(filters=[AzureContentFilter()]),
+            output_filtering=OutputFiltering(filters=[AzureContentFilter()]),
+        ))
+
+    # --- input-only or no filtering: stream=True allowed --------------------
+
+    def test_no_filtering_stream_allowed(self):
+        sentinel = object()
+        with patch("sap_cloud_sdk.aicore.completion.litellm.completion", return_value=sentinel):
+            result = completion(model="sap/x", messages=[], stream=True)
+        assert result is sentinel
+
+    def test_input_only_filtering_stream_allowed(self):
+        self._activate_input_only()
+        sentinel = object()
+        with patch("sap_cloud_sdk.aicore.completion.litellm.completion", return_value=sentinel):
+            result = completion(model="sap/x", messages=[], stream=True)
+        assert result is sentinel
+
+    def test_acompletion_input_only_stream_allowed(self):
+        sentinel = object()
+
+        async def _fake_acompletion(*_, **__):  # noqa: S7503
+            return sentinel
+
+        with patch("sap_cloud_sdk.aicore.completion.litellm.acompletion", side_effect=_fake_acompletion):
+            result = asyncio.run(acompletion(model="sap/x", messages=[], stream=True))
+        assert result is sentinel
+
+    # --- output filtering + stream=True: blocked upfront --------------------
+
+    def test_output_filter_stream_blocked(self):
+        from sap_cloud_sdk.aicore.filtering.exceptions import ContentFilteredError
+        from sap_cloud_sdk.aicore.filtering._patch import FilteringOrchestrationConfig
+        from unittest.mock import patch as _patch
+
+        self._activate_output()
+
+        # stream=True is in optional_params — LiteLLM pops it before calling
+        # transform_request, so the body returned by super() never has
+        # body["config"]["stream"]. We must read it from optional_params first.
+        body = {"config": {"modules": {"model_id": "x"}}}
+        instance = FilteringOrchestrationConfig()
+
+        with _patch(
+            "sap_cloud_sdk.aicore.filtering._patch.GenAIHubOrchestrationConfig.transform_request",
+            return_value=body,
+        ):
+            with pytest.raises(ContentFilteredError) as ei:
+                instance.transform_request(
+                    model="sap/x", messages=[], optional_params={"stream": True},
+                    litellm_params={}, headers={},
+                )
+        assert ei.value.direction == "output"
+
+    def test_input_and_output_filter_stream_blocked(self):
+        from sap_cloud_sdk.aicore.filtering.exceptions import ContentFilteredError
+        from sap_cloud_sdk.aicore.filtering._patch import FilteringOrchestrationConfig
+        from unittest.mock import patch as _patch
+
+        self._activate_input_and_output()
+
+        body = {"config": {"modules": {"model_id": "x"}}}
+        instance = FilteringOrchestrationConfig()
+
+        with _patch(
+            "sap_cloud_sdk.aicore.filtering._patch.GenAIHubOrchestrationConfig.transform_request",
+            return_value=body,
+        ):
+            with pytest.raises(ContentFilteredError) as ei:
+                instance.transform_request(
+                    model="sap/x", messages=[], optional_params={"stream": True},
+                    litellm_params={}, headers={},
+                )
+        assert ei.value.direction == "output"
+
+    def test_output_filter_no_stream_not_blocked(self):
+        from sap_cloud_sdk.aicore.filtering._patch import FilteringOrchestrationConfig
+        from unittest.mock import patch as _patch
+
+        self._activate_output()
+
+        body = {"config": {"modules": {"model_id": "x"}}}
+        instance = FilteringOrchestrationConfig()
+
+        with _patch(
+            "sap_cloud_sdk.aicore.filtering._patch.GenAIHubOrchestrationConfig.transform_request",
+            return_value=body,
+        ):
+            result = instance.transform_request(
+                model="sap/x", messages=[], optional_params={},
+                litellm_params={}, headers={},
+            )
+        assert "filtering" in result["config"]["modules"]
+
+    def test_acompletion_output_filter_stream_blocked(self):
+        from sap_cloud_sdk.aicore.filtering.exceptions import ContentFilteredError
+        from sap_cloud_sdk.aicore.filtering._patch import FilteringOrchestrationConfig
+        from unittest.mock import patch as _patch
+
+        self._activate_output()
+
+        body = {"config": {"modules": {"model_id": "x"}}}
+        instance = FilteringOrchestrationConfig()
+
+        with _patch(
+            "sap_cloud_sdk.aicore.filtering._patch.GenAIHubOrchestrationConfig.transform_request",
+            return_value=body,
+        ):
+            with pytest.raises(ContentFilteredError) as ei:
+                instance.transform_request(
+                    model="sap/x", messages=[], optional_params={"stream": True},
+                    litellm_params={}, headers={},
+                )
+        assert ei.value.direction == "output"
+
+
 # ---------------------------------------------------------------------------
 # Reactive reload on AuthenticationError — async
 # ---------------------------------------------------------------------------