fix(streaming): fail closed when rail actions fail

Signed-off-by: Pouyanpi <13303554+Pouyanpi@users.noreply.github.com>

Author: Pouyanpi

nemoguardrails/rails/llm/llmrails.py

@@ -2015,9 +2015,27 @@ def _prepare_params(
                         action_params=action_params,
                     )
 
-                    result = await self.runtime.action_dispatcher.execute_action(action_name, params)
+                    try:
+                        result, status = await self.runtime.action_dispatcher.execute_action(action_name, params)
+                    except Exception:
+                        log.exception("Action %s failed during sequential streaming", action_name)
+                        result, status = None, "failed"
                     self._explain_info = self._ensure_explain_info()
 
+                    if status != "success":
+                        error_message = f"Action {action_name} failed with status: {status}"
+                        log.error(error_message)
+                        error_data = {
+                            "error": {
+                                "message": f"Internal error in {flow_id} rail: {error_message}",
+                                "type": "internal_error",
+                                "param": flow_id,
+                                "code": "rail_execution_failure",
+                            }
+                        }
+                        yield json.dumps(error_data)
+                        return
+
                     action_func = self.runtime.action_dispatcher.get_action(action_name)
 
                     # Use the mapping to decide if the result indicates blocked content.

tests/test_streaming_internal_errors.py

@@ -23,6 +23,7 @@
 
 from nemoguardrails import RailsConfig
 from nemoguardrails.actions import action
+from nemoguardrails.exceptions import LLMCallException
 from nemoguardrails.imports import check_optional_dependency
 from tests.utils import TestChat
 
@@ -109,6 +110,67 @@ def failing_rail_action(**params):
     assert error["error"]["param"] == "failing safety check"
 
 
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    "action_error",
+    [
+        pytest.param(RuntimeError("Action execution failed"), id="runtime-error"),
+        pytest.param(LLMCallException(RuntimeError("LLM call failed")), id="llm-call-error"),
+    ],
+)
+async def test_sequential_streaming_action_execution_failure(action_error: Exception):
+    """Test fail-closed behavior in sequential streaming when a rail action fails."""
+
+    @action(is_system_action=True)
+    def failing_rail_action(**params):
+        raise action_error
+
+    config = RailsConfig.from_content(
+        config={
+            "models": [{"type": "main", "engine": "openai", "model": "gpt-3.5-turbo"}],
+            "rails": {
+                "output": {
+                    "parallel": False,
+                    "flows": ["failing safety check"],
+                    "streaming": {
+                        "enabled": True,
+                        "chunk_size": 4,
+                    },
+                }
+            },
+        },
+        colang_content="""
+        define user express greeting
+          "hi"
+        define flow
+          user express greeting
+          bot express greeting
+
+        define subflow failing safety check
+          execute failing_rail_action
+        """,
+    )
+
+    llm_completions = [
+        'bot express greeting\n  "Hello there! How can I help you?"',
+    ]
+
+    chat = TestChat(config, llm_completions=llm_completions, streaming=True)
+    chat.app.register_action(failing_rail_action)
+
+    chunks = await collect_streaming_chunks(chat.app.stream_async(messages=[{"role": "user", "content": "Hi!"}]))
+
+    internal_error_chunks = find_internal_error_chunks(chunks)
+    assert len(internal_error_chunks) == 1
+    error = internal_error_chunks[0]["error"]
+    assert error == {
+        "message": "Internal error in failing safety check rail: Action failing_rail_action failed with status: failed",
+        "type": "internal_error",
+        "param": "failing safety check",
+        "code": "rail_execution_failure",
+    }
+
+
 @pytest.mark.asyncio
 async def test_streaming_internal_error_format():
     """Test that streaming internal errors have the correct format."""