feat(server): classify guarded projection outcomes

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

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."""
 
         ...
 

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,

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,84 @@ async def dispatch(value):
     assert dispatched == [request]
 
 
+@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",
+        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():
+    """Hide the provider response when output projection fails."""
+
+    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():
+    """Preserve a response that does not contain content to check."""
+
+    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():
+    """Reject an input projection without content to check."""
+
+    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."""

tests/server/experimental/test_http_kernel.py

@@ -26,14 +26,20 @@
 from nemoguardrails.server.experimental._buffered_kernel import (
     OperationBlocked,
     OperationCheckFailed,
+    OperationModificationUnsupported,
+    OperationProjectionFailed,
 )
 from nemoguardrails.server.experimental._content_checker import (
     ContentAllowed,
     ContentBlocked,
     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,81 @@ 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():
+    """Render input projection failures without provider 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():
+    """Preserve a provider error that has no output content to check."""
+
+    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):