fix(server): emit terminal SSE frames for rail execution failures

nemoguardrails/llm/clients/_errors.py

@@ -162,16 +162,19 @@ def build_error_payload(
 # looks like an OpenAI error object cannot truncate a stream or skip output
 # rails. ``downstream_error`` carries an upstream HTTP status;
 # ``generation_error`` is a status-less generation failure;
-# ``guardrails_violation`` is emitted when a rail blocks.
+# ``guardrails_violation`` is emitted when a rail blocks;
+# ``internal_error`` is emitted when a rail fails to execute.
 GENERATION_ERROR_TYPE = "generation_error"
 DOWNSTREAM_ERROR_TYPE = "downstream_error"
 GUARDRAILS_VIOLATION_TYPE = "guardrails_violation"
+INTERNAL_ERROR_TYPE = "internal_error"
 
 STREAM_ERROR_TYPES = frozenset(
     {
         GENERATION_ERROR_TYPE,
         DOWNSTREAM_ERROR_TYPE,
         GUARDRAILS_VIOLATION_TYPE,
+        INTERNAL_ERROR_TYPE,
     }
 )
 

nemoguardrails/server/api.py

@@ -23,13 +23,13 @@
 import time
 import uuid
 from contextlib import asynccontextmanager
-from typing import Any, AsyncIterator, Callable, List, Literal, Optional, Union
+from typing import Any, AsyncIterator, Callable, List, Optional, Union
 
 import httpx
 from fastapi import FastAPI, HTTPException, Request
 from fastapi.exceptions import RequestValidationError
 from fastapi.middleware.cors import CORSMiddleware
-from pydantic import BaseModel, ValidationError
+from pydantic import BaseModel, ValidationError, field_validator
 from starlette.exceptions import HTTPException as StarletteHTTPException
 from starlette.middleware.exceptions import ExceptionMiddleware
 from starlette.responses import JSONResponse, RedirectResponse, StreamingResponse
@@ -53,6 +53,7 @@
 from nemoguardrails.http.errors import HTTPClientError
 from nemoguardrails.llm.call import _prepend_think_tags
 from nemoguardrails.llm.clients._errors import (
+    STREAM_ERROR_TYPES,
     build_error_payload,
     build_streaming_error_payload,
     normalize_error_status,
@@ -539,10 +540,17 @@ class ChunkErrorMetadata(BaseModel):
     message: str
     # Only the internal stream markers count as a terminal error frame; see
     # ChunkError below.
-    type: Literal["generation_error", "downstream_error", "guardrails_violation"]
+    type: str
     param: Optional[str] = None
     code: Union[str, int, None] = None
 
+    @field_validator("type")
+    @classmethod
+    def validate_error_type(cls, value: str) -> str:
+        if value not in STREAM_ERROR_TYPES:
+            raise ValueError("Not an internal stream error type")
+        return value
+
 
 class ChunkError(BaseModel):
     """A terminal error frame pushed into the stream by the guardrails runtime.

tests/guardrails/test_iorails_streaming.py

@@ -990,6 +990,12 @@ def test_error_json_string(self):
     def test_metadata_frame_with_error(self):
         assert _is_stream_error_chunk({"text": '{"error": {"type": "generation_error"}}'}) is True
 
+    @pytest.mark.parametrize("include_metadata", [False, True])
+    def test_internal_rail_error_is_terminal(self, include_metadata):
+        error = '{"error": {"type": "internal_error", "code": "rail_execution_failure"}}'
+        chunk = {"text": error} if include_metadata else error
+        assert _is_stream_error_chunk(chunk) is True
+
     def test_non_error_json_string(self):
         assert _is_stream_error_chunk('{"tool_calls": []}') is False
 

tests/server/test_api.py

@@ -25,7 +25,8 @@
 pytest.importorskip("openai", reason="openai is required for server tests")
 from fastapi.testclient import TestClient
 
-from nemoguardrails import RailsConfig
+from nemoguardrails import Guardrails, RailsConfig
+from nemoguardrails.actions import action
 from nemoguardrails.exceptions import InvalidModelConfigurationError
 from nemoguardrails.guardrails.model_engine import ModelEngine
 from nemoguardrails.llm.models.openai_chat import OpenAIChatModel
@@ -1412,6 +1413,97 @@ async def test_openai_sse_format_basic_chunks():
     assert done == "data: [DONE]\n\n"
 
 
+@pytest.mark.asyncio
+@pytest.mark.parametrize("as_dict", [False, True])
+async def test_streaming_rail_execution_failure_is_terminal(as_dict):
+    error = {
+        "error": {
+            "message": "Internal error in failing safety check rail: action failed",
+            "type": "internal_error",
+            "param": "failing safety check",
+            "code": "rail_execution_failure",
+        }
+    }
+    stream = _create_test_stream([error if as_dict else json.dumps(error), "Unchecked content"])
+
+    frames = [frame async for frame in _format_streaming_response(stream, model_name="test-model")]
+
+    assert len(frames) == 2
+    assert json.loads(frames[0][6:]) == error
+    assert frames[1] == "data: [DONE]\n\n"
+
+
+@pytest.mark.parametrize("use_iorails", [False, True], ids=["legacy", "iorails-fallback"])
+@pytest.mark.parametrize("parallel", [False, True], ids=["sequential", "parallel"])
+@pytest.mark.parametrize("invalid_outcome", [False, True], ids=["action-failure", "invalid-outcome"])
+@pytest.mark.parametrize("stream_first", [False, True], ids=["rails-first", "stream-first"])
+def test_streaming_output_rail_failure_returns_error_frame(
+    monkeypatch, use_iorails, parallel, invalid_outcome, stream_first
+):
+    @action(is_system_action=True)
+    def failing_rail_action(**params):
+        if invalid_outcome:
+            return False
+        raise RuntimeError("Action execution failed")
+
+    config = RailsConfig.from_content(
+        config={
+            "models": [],
+            "passthrough": True,
+            "rails": {
+                "output": {
+                    "parallel": parallel,
+                    "flows": ["failing safety check"],
+                    "streaming": {"enabled": True, "stream_first": stream_first, "chunk_size": 4},
+                }
+            },
+        },
+        colang_content="""
+        define subflow failing safety check
+          execute failing_rail_action
+        """,
+    )
+    rails = Guardrails(config, llm=FakeLLMModel(responses=["Unchecked model content"]), use_iorails=use_iorails)
+    assert isinstance(rails.rails_engine, LLMRails)
+    rails.register_action(failing_rail_action)
+    monkeypatch.setattr(api, "_get_rails", AsyncMock(return_value=rails))
+
+    response = client.post(
+        "/v1/chat/completions",
+        json={
+            "model": "test-model",
+            "messages": [{"role": "user", "content": "Hello"}],
+            "stream": True,
+            "guardrails": {"config_id": "with_custom_llm"},
+        },
+    )
+
+    assert response.status_code == 200
+    assert response.headers["content-type"].startswith("text/event-stream")
+    frames = [line[6:] for line in response.text.splitlines() if line.startswith("data: ")]
+    assert frames[-1] == "[DONE]"
+    events = [json.loads(frame) for frame in frames[:-1]]
+    content_events = events[:-1]
+    if stream_first:
+        assert content_events
+        assert "".join(event["choices"][0]["delta"]["content"] for event in content_events) == "Unchecked model content"
+    else:
+        assert content_events == []
+    error_message = (
+        "Rail action must return RailOutcome, got bool"
+        if invalid_outcome
+        else "Action failing_rail_action failed with status: failed"
+    )
+    assert events[-1] == {
+        "error": {
+            "message": f"Internal error in failing safety check rail: {error_message}",
+            "type": "internal_error",
+            "param": "failing safety check",
+            "code": "rail_execution_failure",
+        }
+    }
+
+
 @pytest.mark.asyncio
 async def test_openai_sse_format_with_model_name():
     """Test that model name is properly included in the response."""