From 32dfc7acecd986074d8f5a3fcffb375a89371cb3 Mon Sep 17 00:00:00 2001 From: Pouyanpi <13303554+Pouyanpi@users.noreply.github.com> Date: Fri, 25 Sep 2026 14:51:07 +0200 Subject: [PATCH 1/2] feat(server): classify guarded projection outcomes Signed-off-by: Pouyanpi <13303554+Pouyanpi@users.noreply.github.com> --- .../server/experimental/_buffered_kernel.py | 45 ++++++++-- .../server/experimental/_guarded_operation.py | 13 ++- .../server/experimental/_http_kernel.py | 3 + .../experimental/test_buffered_kernel.py | 84 ++++++++++++++++- tests/server/experimental/test_http_kernel.py | 90 ++++++++++++++++++- 5 files changed, 224 insertions(+), 11 deletions(-) diff --git a/nemoguardrails/server/experimental/_buffered_kernel.py b/nemoguardrails/server/experimental/_buffered_kernel.py index 8f7553c9d4..e912f257a2 100644 --- a/nemoguardrails/server/experimental/_buffered_kernel.py +++ b/nemoguardrails/server/experimental/_buffered_kernel.py @@ -32,7 +32,12 @@ validate_content_check_decision, validate_content_checker, ) -from nemoguardrails.server.experimental._guarded_operation import BufferedGuardedOperation, GuardedMessageProjection +from nemoguardrails.server.experimental._guarded_operation import ( + BufferedGuardedOperation, + ContentInspectionNotApplicable, + GuardedMessageProjection, + UnsupportedGuardedPayload, +) from nemoguardrails.server.experimental.provider.types import GuardedMessage RequestT = TypeVar("RequestT") @@ -78,16 +83,26 @@ class OperationModificationUnsupported: failure: UnsupportedContentModification +@dataclass(frozen=True, slots=True) +class OperationProjectionFailed: + """Stop an operation when a provider payload cannot be inspected safely.""" + + stage: InspectionStage + failure: UnsupportedGuardedPayload + + def _project_message( projection: GuardedMessageProjection[PayloadT], payload: PayloadT, expected_role: Literal["user", "assistant"], -) -> GuardedMessage: - """Project and validate one message selected for content checking.""" +) -> GuardedMessage | ContentInspectionNotApplicable: + """Project and validate one result for content checking.""" message = projection(payload) + if isinstance(message, ContentInspectionNotApplicable): + return message if not isinstance(message, GuardedMessage): - raise TypeError("A guarded operation projection must return GuardedMessage.") + raise TypeError("A guarded operation projection must return GuardedMessage or ContentInspectionNotApplicable.") if message.role != expected_role: raise ValueError(f"A guarded operation {expected_role} projection returned role {message.role!r}.") return message @@ -136,13 +151,24 @@ async def execute_buffered_operation( checker: ContentChecker | _ResolvedContentChecker, request: RequestT, dispatch: Callable[[RequestT], Awaitable[ResponseT]], -) -> OperationCompleted[ResponseT] | OperationBlocked | OperationCheckFailed | OperationModificationUnsupported: +) -> ( + OperationCompleted[ResponseT] + | OperationBlocked + | OperationCheckFailed + | OperationModificationUnsupported + | OperationProjectionFailed +): """Execute one buffered operation with exactly one statically bound checker.""" validated = checker if isinstance(checker, _ResolvedContentChecker) else validate_content_checker(checker) validated_checker = validated.checker policy = validated.policy - input_message = _project_message(operation.input_projection, request, "user") + try: + input_message = _project_message(operation.input_projection, request, "user") + except UnsupportedGuardedPayload as failure: + return OperationProjectionFailed(InspectionStage.INPUT, failure) + if isinstance(input_message, ContentInspectionNotApplicable): + raise TypeError("A guarded operation input projection cannot be inapplicable.") if policy.inspect_input: stopped = await _run_check( @@ -155,7 +181,12 @@ async def execute_buffered_operation( response = await dispatch(request) if policy.inspect_output: - output_message = _project_message(operation.output_projection, response, "assistant") + try: + output_message = _project_message(operation.output_projection, response, "assistant") + except UnsupportedGuardedPayload as failure: + return OperationProjectionFailed(InspectionStage.OUTPUT, failure) + if isinstance(output_message, ContentInspectionNotApplicable): + return OperationCompleted(response) stopped = await _run_check( InspectionStage.OUTPUT, lambda: validated_checker.check_output(OutputContentCheck(input_message, output_message.content)), diff --git a/nemoguardrails/server/experimental/_guarded_operation.py b/nemoguardrails/server/experimental/_guarded_operation.py index 6da4679b90..1aaba3978c 100644 --- a/nemoguardrails/server/experimental/_guarded_operation.py +++ b/nemoguardrails/server/experimental/_guarded_operation.py @@ -26,11 +26,20 @@ ResponseT = TypeVar("ResponseT") +class UnsupportedGuardedPayload(ValueError): + """Report a provider payload outside an operation's guarded projection.""" + + +@dataclass(frozen=True, slots=True) +class ContentInspectionNotApplicable: + """Declare that one provider response does not contain guarded output.""" + + class GuardedMessageProjection(Protocol[PayloadT]): """Extract one message to check from a provider-owned value.""" - def __call__(self, payload: PayloadT) -> GuardedMessage: - """Return the message selected from one payload.""" + def __call__(self, payload: PayloadT) -> GuardedMessage | ContentInspectionNotApplicable: + """Return the message selected from one payload, when present.""" ... diff --git a/nemoguardrails/server/experimental/_http_kernel.py b/nemoguardrails/server/experimental/_http_kernel.py index 9b2571c250..d178888806 100644 --- a/nemoguardrails/server/experimental/_http_kernel.py +++ b/nemoguardrails/server/experimental/_http_kernel.py @@ -31,6 +31,7 @@ OperationCheckFailed, OperationCompleted, OperationModificationUnsupported, + OperationProjectionFailed, execute_buffered_operation, ) from nemoguardrails.server.experimental._content_checker import ( @@ -185,6 +186,7 @@ class GuardedHttpOperation: OperationBlocked | OperationCheckFailed | OperationModificationUnsupported + | OperationProjectionFailed | HttpOperationFailed | HttpRouteRejected ], @@ -414,6 +416,7 @@ def _render_failure( failure: OperationBlocked | OperationCheckFailed | OperationModificationUnsupported + | OperationProjectionFailed | HttpOperationFailed | HttpRouteRejected, render_outcome: OutcomeRenderer, diff --git a/tests/server/experimental/test_buffered_kernel.py b/tests/server/experimental/test_buffered_kernel.py index 5e1f19f9ce..1a1448327d 100644 --- a/tests/server/experimental/test_buffered_kernel.py +++ b/tests/server/experimental/test_buffered_kernel.py @@ -25,6 +25,7 @@ OperationCheckFailed, OperationCompleted, OperationModificationUnsupported, + OperationProjectionFailed, execute_buffered_operation, ) from nemoguardrails.server.experimental._content_checker import ( @@ -33,7 +34,11 @@ ContentCheckFailed, ContentInspectionPolicy, ) -from nemoguardrails.server.experimental._guarded_operation import BufferedGuardedOperation +from nemoguardrails.server.experimental._guarded_operation import ( + BufferedGuardedOperation, + ContentInspectionNotApplicable, + UnsupportedGuardedPayload, +) from nemoguardrails.server.experimental.provider.types import GuardedMessage @@ -88,6 +93,13 @@ async def check_output(self, check): return self.output_decision +def projection_raising(failure): + def project(_payload): + raise failure + + return project + + @pytest.fixture def operation(): return BufferedGuardedOperation( @@ -309,6 +321,76 @@ async def dispatch(value): assert dispatched == [request] +@pytest.mark.asyncio +async def test_unsupported_input_payload_fails_before_dispatch(): + failure = UnsupportedGuardedPayload("unsupported request") + operation = BufferedGuardedOperation( + name="test.unsupported_input", + input_projection=projection_raising(failure), + output_projection=lambda response: GuardedMessage("assistant", response.text), + ) + + async def dispatch(_request): + pytest.fail("an unsupported input must not be dispatched") + + result = await execute_buffered_operation(operation, StaticChecker(), Request("question"), dispatch) + + assert result == OperationProjectionFailed(InspectionStage.INPUT, failure) + + +@pytest.mark.asyncio +async def test_unsupported_output_payload_hides_provider_response(): + failure = UnsupportedGuardedPayload("unsupported response") + response = Response("answer") + operation = BufferedGuardedOperation( + name="test.unsupported_output", + input_projection=lambda request: GuardedMessage("user", request.text), + output_projection=projection_raising(failure), + ) + + async def dispatch(_request): + return response + + result = await execute_buffered_operation(operation, StaticChecker(), Request("question"), dispatch) + + assert result == OperationProjectionFailed(InspectionStage.OUTPUT, failure) + assert not hasattr(result, "response") + + +@pytest.mark.asyncio +async def test_inapplicable_output_inspection_preserves_provider_response(): + response = Response("provider error") + checker = StaticChecker() + operation = BufferedGuardedOperation( + name="test.inapplicable_output", + input_projection=lambda request: GuardedMessage("user", request.text), + output_projection=lambda _response: ContentInspectionNotApplicable(), + ) + + async def dispatch(_request): + return response + + result = await execute_buffered_operation(operation, checker, Request("question"), dispatch) + + assert result == OperationCompleted(response) + assert [call if isinstance(call, str) else call[0] for call in checker.calls] == ["policy", "input"] + + +@pytest.mark.asyncio +async def test_input_inspection_cannot_be_declared_inapplicable(): + operation = BufferedGuardedOperation( + name="test.inapplicable_input", + input_projection=lambda _request: ContentInspectionNotApplicable(), + output_projection=lambda response: GuardedMessage("assistant", response.text), + ) + + async def dispatch(_request): + pytest.fail("an inapplicable input must not be dispatched") + + with pytest.raises(TypeError, match="input projection cannot be inapplicable"): + await execute_buffered_operation(operation, StaticChecker(), Request("question"), dispatch) + + @pytest.mark.asyncio async def test_unknown_checker_decision_fails_closed_before_dispatch(operation): """Convert an unknown input decision into a failed check.""" diff --git a/tests/server/experimental/test_http_kernel.py b/tests/server/experimental/test_http_kernel.py index 01c08fcbd1..76800db43a 100644 --- a/tests/server/experimental/test_http_kernel.py +++ b/tests/server/experimental/test_http_kernel.py @@ -26,6 +26,8 @@ from nemoguardrails.server.experimental._buffered_kernel import ( OperationBlocked, OperationCheckFailed, + OperationModificationUnsupported, + OperationProjectionFailed, ) from nemoguardrails.server.experimental._content_checker import ( ContentAllowed, @@ -33,7 +35,11 @@ ContentCheckFailed, ContentInspectionPolicy, ) -from nemoguardrails.server.experimental._guarded_operation import BufferedGuardedOperation +from nemoguardrails.server.experimental._guarded_operation import ( + BufferedGuardedOperation, + ContentInspectionNotApplicable, + UnsupportedGuardedPayload, +) from nemoguardrails.server.experimental._http_kernel import ( BufferedHttpResponse, GuardedHttpOperation, @@ -82,6 +88,13 @@ def project_response(response): return GuardedMessage("assistant", payload["output"]) +def projection_raising(failure): + def project(_payload): + raise failure + + return project + + def render_test_outcome(outcome): if isinstance(outcome, HttpRouteRejected): status_codes = { @@ -98,6 +111,10 @@ def render_test_outcome(outcome): HttpFailureKind.RESPONSE_BODY_TOO_LARGE: 502, } return BufferedHttpResponse(status_codes[outcome.kind], (), outcome.kind.value.encode()) + if isinstance(outcome, OperationProjectionFailed): + return BufferedHttpResponse(422, (), str(outcome.failure).encode()) + if isinstance(outcome, OperationModificationUnsupported): + return BufferedHttpResponse(422, (), b"replacement_not_supported") if isinstance(outcome, OperationBlocked): body = f"{outcome.stage.value}:{outcome.decision.message}".encode() return BufferedHttpResponse(400, ((b"content-type", b"text/plain"),), body) @@ -608,6 +625,77 @@ def test_guarded_http_path_rejects_path_spanning_parameters(): GuardedOperationPath("/v1/{rest:path}") +@pytest.mark.asyncio +async def test_projection_failure_is_rendered_without_dispatch(): + dispatched = [] + operation = GuardedHttpOperation( + operation_path=GuardedOperationPath("/v1/projected"), + operation=BufferedGuardedOperation( + name="test.projected", + input_projection=projection_raising(UnsupportedGuardedPayload("unsupported")), + output_projection=project_response, + ), + ) + + async def dispatch(request): + dispatched.append(request) + raise AssertionError("an unsupported input must not be dispatched") + + app = FastAPI() + app.include_router( + create_http_proxy_router( + operations=[operation], + checker=StaticChecker(), + dispatch=dispatch, + render_outcome=render_test_outcome, + ) + ) + async with httpx.AsyncClient(transport=httpx.ASGITransport(app=app), base_url="http://proxy.test") as client: + response = await client.post("/v1/projected", json={"input": "question"}) + + assert response.status_code == 422 + assert response.content == b"unsupported" + assert dispatched == [] + + +@pytest.mark.asyncio +async def test_inapplicable_output_inspection_preserves_provider_error(): + checker = StaticChecker() + provider_response = BufferedHttpResponse( + 429, + ((b"content-type", b"application/json"), (b"x-request-id", b"provider-id")), + b'{"error":{"message":"rate limited"}}', + ) + operation = GuardedHttpOperation( + operation_path=GuardedOperationPath("/v1/projected"), + operation=BufferedGuardedOperation( + name="test.provider_error", + input_projection=project_request, + output_projection=lambda _response: ContentInspectionNotApplicable(), + ), + ) + + async def dispatch(_request): + return provider_response + + app = FastAPI() + app.include_router( + create_http_proxy_router( + operations=[operation], + checker=checker, + dispatch=dispatch, + render_outcome=render_test_outcome, + ) + ) + async with httpx.AsyncClient(transport=httpx.ASGITransport(app=app), base_url="http://proxy.test") as client: + response = await client.post("/v1/projected", json={"input": "question"}) + + assert response.status_code == 429 + assert response.content == provider_response.body + assert response.headers["x-request-id"] == "provider-id" + assert [call[0] for call in checker.calls if not isinstance(call, str)] == ["input"] + + @pytest.mark.asyncio @pytest.mark.parametrize("content_length", ["invalid", "-1", "+2", " 2", "2,2"]) async def test_invalid_content_length_is_rendered_without_dispatch(guarded_operation, content_length): From 2121b0a8cfe8e683f09e3185255d1dabd8a9019d Mon Sep 17 00:00:00 2001 From: Pouyanpi <13303554+Pouyanpi@users.noreply.github.com> Date: Sat, 26 Sep 2026 08:51:26 +0200 Subject: [PATCH 2/2] docs(server): describe projection outcome tests Signed-off-by: Pouyanpi <13303554+Pouyanpi@users.noreply.github.com> --- tests/server/experimental/test_buffered_kernel.py | 8 ++++++++ tests/server/experimental/test_http_kernel.py | 4 ++++ 2 files changed, 12 insertions(+) diff --git a/tests/server/experimental/test_buffered_kernel.py b/tests/server/experimental/test_buffered_kernel.py index 1a1448327d..7e5722366f 100644 --- a/tests/server/experimental/test_buffered_kernel.py +++ b/tests/server/experimental/test_buffered_kernel.py @@ -323,6 +323,8 @@ async def dispatch(value): @pytest.mark.asyncio async def test_unsupported_input_payload_fails_before_dispatch(): + """Return an input projection failure before provider dispatch.""" + failure = UnsupportedGuardedPayload("unsupported request") operation = BufferedGuardedOperation( name="test.unsupported_input", @@ -340,6 +342,8 @@ async def dispatch(_request): @pytest.mark.asyncio async def test_unsupported_output_payload_hides_provider_response(): + """Hide the provider response when output projection fails.""" + failure = UnsupportedGuardedPayload("unsupported response") response = Response("answer") operation = BufferedGuardedOperation( @@ -359,6 +363,8 @@ async def dispatch(_request): @pytest.mark.asyncio async def test_inapplicable_output_inspection_preserves_provider_response(): + """Preserve a response that does not contain content to check.""" + response = Response("provider error") checker = StaticChecker() operation = BufferedGuardedOperation( @@ -378,6 +384,8 @@ async def dispatch(_request): @pytest.mark.asyncio async def test_input_inspection_cannot_be_declared_inapplicable(): + """Reject an input projection without content to check.""" + operation = BufferedGuardedOperation( name="test.inapplicable_input", input_projection=lambda _request: ContentInspectionNotApplicable(), diff --git a/tests/server/experimental/test_http_kernel.py b/tests/server/experimental/test_http_kernel.py index 76800db43a..204823b257 100644 --- a/tests/server/experimental/test_http_kernel.py +++ b/tests/server/experimental/test_http_kernel.py @@ -627,6 +627,8 @@ def test_guarded_http_path_rejects_path_spanning_parameters(): @pytest.mark.asyncio async def test_projection_failure_is_rendered_without_dispatch(): + """Render input projection failures without provider dispatch.""" + dispatched = [] operation = GuardedHttpOperation( operation_path=GuardedOperationPath("/v1/projected"), @@ -660,6 +662,8 @@ async def dispatch(request): @pytest.mark.asyncio async def test_inapplicable_output_inspection_preserves_provider_error(): + """Preserve a provider error that has no output content to check.""" + checker = StaticChecker() provider_response = BufferedHttpResponse( 429,