From 431555792e1541d1beab0b372d74c69cc4c8d64a Mon Sep 17 00:00:00 2001 From: Pouyanpi <13303554+Pouyanpi@users.noreply.github.com> Date: Wed, 28 Oct 2026 10:24:01 +0100 Subject: [PATCH 1/2] fix(server): emit terminal SSE frames for rail execution failures Signed-off-by: Pouyanpi <13303554+Pouyanpi@users.noreply.github.com> --- nemoguardrails/server/api.py | 2 +- tests/server/test_api.py | 85 +++++++++++++++++++++++++++++++++++- 2 files changed, 85 insertions(+), 2 deletions(-) diff --git a/nemoguardrails/server/api.py b/nemoguardrails/server/api.py index 9aefe4b732..3db48b3b58 100644 --- a/nemoguardrails/server/api.py +++ b/nemoguardrails/server/api.py @@ -539,7 +539,7 @@ 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: Literal["generation_error", "downstream_error", "guardrails_violation", "internal_error"] param: Optional[str] = None code: Union[str, int, None] = None diff --git a/tests/server/test_api.py b/tests/server/test_api.py index 1b805db595..70962c7385 100644 --- a/tests/server/test_api.py +++ b/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,88 @@ 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"]) +def test_streaming_output_rail_failure_returns_error_frame(monkeypatch, use_iorails, parallel, invalid_outcome): + @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": False, "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 len(frames) == 2 + assert frames[1] == "[DONE]" + error_message = ( + "Rail action must return RailOutcome, got bool" + if invalid_outcome + else "Action failing_rail_action failed with status: failed" + ) + assert json.loads(frames[0]) == { + "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.""" From 012f999aa8b702066451d7cc422625ff3e932255 Mon Sep 17 00:00:00 2001 From: Pouyanpi <13303554+Pouyanpi@users.noreply.github.com> Date: Thu, 29 Oct 2026 08:53:12 +0100 Subject: [PATCH 2/2] fix(streaming): share terminal error types and cover stream-first failures Signed-off-by: Pouyanpi <13303554+Pouyanpi@users.noreply.github.com> --- nemoguardrails/llm/clients/_errors.py | 5 ++++- nemoguardrails/server/api.py | 14 +++++++++++--- tests/guardrails/test_iorails_streaming.py | 6 ++++++ tests/server/test_api.py | 19 ++++++++++++++----- 4 files changed, 35 insertions(+), 9 deletions(-) diff --git a/nemoguardrails/llm/clients/_errors.py b/nemoguardrails/llm/clients/_errors.py index 988e875fd0..0d0d8de651 100644 --- a/nemoguardrails/llm/clients/_errors.py +++ b/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, } ) diff --git a/nemoguardrails/server/api.py b/nemoguardrails/server/api.py index 3db48b3b58..3f1c4fa610 100644 --- a/nemoguardrails/server/api.py +++ b/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", "internal_error"] + 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. diff --git a/tests/guardrails/test_iorails_streaming.py b/tests/guardrails/test_iorails_streaming.py index b89668383c..63ceb148b2 100644 --- a/tests/guardrails/test_iorails_streaming.py +++ b/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 diff --git a/tests/server/test_api.py b/tests/server/test_api.py index 70962c7385..22f5cc051c 100644 --- a/tests/server/test_api.py +++ b/tests/server/test_api.py @@ -1436,7 +1436,10 @@ async def test_streaming_rail_execution_failure_is_terminal(as_dict): @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"]) -def test_streaming_output_rail_failure_returns_error_frame(monkeypatch, use_iorails, parallel, 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: @@ -1451,7 +1454,7 @@ def failing_rail_action(**params): "output": { "parallel": parallel, "flows": ["failing safety check"], - "streaming": {"enabled": True, "stream_first": False, "chunk_size": 4}, + "streaming": {"enabled": True, "stream_first": stream_first, "chunk_size": 4}, } }, }, @@ -1478,14 +1481,20 @@ def failing_rail_action(**params): 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 len(frames) == 2 - assert frames[1] == "[DONE]" + 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 json.loads(frames[0]) == { + assert events[-1] == { "error": { "message": f"Internal error in failing safety check rail: {error_message}", "type": "internal_error",