refactor(library): support canonical conversation history in topic safety

nemoguardrails/library/self_check/input_check/actions.py

@@ -43,6 +43,9 @@ async def self_check_input(
     llms: Dict[str, LLMModel],
     llm_task_manager: LLMTaskManager,
     context: Optional[dict] = None,
+    # LLMRails injects Colang events only to recover variants for legacy 1.0
+    # flows, including parallel rails. Manifests bind `variant` instead so
+    # engine-neutral callers do not provide or synthesize events.
     events: Optional[List[dict]] = None,
     llm: Optional[LLMModel] = None,
     config: Optional[RailsConfig] = None,

nemoguardrails/library/self_check/output_check/actions.py

@@ -43,6 +43,9 @@ async def self_check_output(
     llms: Dict[str, LLMModel],
     llm_task_manager: LLMTaskManager,
     context: Optional[dict] = None,
+    # LLMRails injects Colang events only to recover variants for legacy 1.0
+    # flows, including parallel rails. Manifests bind `variant` instead so
+    # engine-neutral callers do not provide or synthesize events.
     events: Optional[List[dict]] = None,
     llm: Optional[LLMModel] = None,
     config: Optional[RailsConfig] = None,

nemoguardrails/library/topic_safety/actions.py

@@ -14,7 +14,8 @@
 # limitations under the License.
 
 import logging
-from typing import Dict, List, Optional
+from collections.abc import Mapping, Sequence
+from typing import Any, Dict, List, Optional
 
 from nemoguardrails.actions.actions import action
 from nemoguardrails.actions.llm.utils import llm_call
@@ -50,8 +51,12 @@ async def topic_safety_check_input(
     llm_task_manager: LLMTaskManager,
     model_name: Optional[str] = None,
     context: Optional[dict] = None,
+    # LLMRails injects Colang events only as a compatibility source of
+    # conversation history. Manifests omit events because engine-neutral
+    # executors provide canonical `messages` and must not synthesize events.
     events: Optional[List[dict]] = None,
     model_caches: Optional[Dict[str, CacheInterface]] = None,
+    messages: Optional[Sequence[Mapping[str, Any]]] = None,
     **kwargs,
 ) -> RailOutcome:
     _MAX_TOKENS = TOPIC_SAFETY_MAX_TOKENS
@@ -61,10 +66,12 @@ async def topic_safety_check_input(
         user_input = context.get("user_message", "")
         model_name = model_name or context.get("model", None)
 
-    if events is not None:
+    if messages is not None:
+        conversation_history = [dict(message) for message in messages]
+    else:
         # convert InternalEvent objects to dictionary format for compatibility with to_chat_messages
         dict_events = []
-        for event in events:
+        for event in events or []:
             if hasattr(event, "name") and hasattr(event, "arguments"):
                 dict_event = {"type": event.name}
                 dict_event.update(event.arguments)
@@ -108,21 +115,27 @@ async def topic_safety_check_input(
 
     max_tokens = max_tokens or _MAX_TOKENS
 
-    messages = []
-    messages.append({"type": "system", "content": system_prompt})
-    messages.extend(conversation_history)
-    messages.append({"type": "user", "content": user_input})
+    prompt_messages = []
+    prompt_messages.append({"type": "system", "content": system_prompt})
+    prompt_messages.extend(conversation_history)
+    if messages is None:
+        prompt_messages.append({"type": "user", "content": user_input})
 
     cache = model_caches.get(model_name) if model_caches else None
 
     if cache:
-        cache_key = create_normalized_cache_key(messages)
+        cache_key = create_normalized_cache_key(prompt_messages)
         cached_result = get_from_cache_and_restore_stats(cache, cache_key)
         if cached_result is not None:
             log.debug(f"Topic safety cache hit for model '{model_name}'")
             return cached_result
 
-    response = await llm_call(llm, messages, stop=stop, llm_params={"temperature": TOPIC_SAFETY_TEMPERATURE})
+    response = await llm_call(
+        llm,
+        prompt_messages,
+        stop=stop,
+        llm_params={"temperature": TOPIC_SAFETY_TEMPERATURE},
+    )
     result = response.content
 
     if result.lower().strip() == "off-topic":
@@ -133,7 +146,7 @@ async def topic_safety_check_input(
     final_result = RailOutcome.allow() if on_topic else RailOutcome.block()
 
     if cache:
-        cache_key = create_normalized_cache_key(messages)
+        cache_key = create_normalized_cache_key(prompt_messages)
         cache_entry: CacheEntry = {
             "result": final_result,
             "llm_stats": extract_llm_stats_for_cache(),

tests/test_multiple_self_check_rails.py

@@ -887,10 +887,17 @@ def test_get_self_check_task_from_rail_resolves_custom_and_default_tasks():
     )
 
 
-def test_resolve_self_check_task_prefers_explicit_task():
+def test_resolve_self_check_task_prefers_explicit_task_over_legacy_context_and_events():
     task = _resolve_input_task(
         variant="check_harmful",
         context={"triggered_input_rail": "self check input $variant=check_off_topic"},
+        events=[
+            {
+                "type": "start_flow",
+                "flow_id": SELF_CHECK_INPUT_FLOW,
+                "params": {"variant": "check_legacy_event"},
+            }
+        ],
     )
 
     assert task == "check_harmful"

tests/test_topic_safety_actions.py

@@ -0,0 +1,135 @@
+# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+
+from nemoguardrails.library.topic_safety.actions import (
+    TOPIC_SAFETY_OUTPUT_RESTRICTION,
+    topic_safety_check_input,
+)
+from nemoguardrails.types import LLMResponse
+
+
+@pytest.fixture
+def task_manager():
+    manager = MagicMock()
+    manager.render_task_prompt.return_value = "Stay on topic."
+    manager.get_stop_tokens.return_value = []
+    manager.get_max_tokens.return_value = 10
+    return manager
+
+
+async def _run_topic_safety(task_manager, **kwargs):
+    with patch(
+        "nemoguardrails.library.topic_safety.actions.llm_call",
+        new_callable=AsyncMock,
+    ) as mock_llm_call:
+        mock_llm_call.return_value = LLMResponse(content="on-topic")
+        result = await topic_safety_check_input(
+            llms={"topic_control": "topic model"},
+            llm_task_manager=task_manager,
+            model_name="topic_control",
+            context={"user_message": "current question"},
+            **kwargs,
+        )
+
+    return result, mock_llm_call
+
+
+@pytest.mark.asyncio
+async def test_canonical_messages_are_authoritative_and_preserve_metadata(task_manager):
+    canonical_messages = [
+        {"role": "user", "content": "earlier question"},
+        {
+            "role": "assistant",
+            "content": "earlier answer",
+            "provider_metadata": {"request_id": "response-1"},
+        },
+        {
+            "role": "user",
+            "content": [{"type": "text", "text": "current question"}],
+            "name": "customer",
+        },
+    ]
+
+    result, mock_llm_call = await _run_topic_safety(
+        task_manager,
+        messages=canonical_messages,
+        events=[{"type": "UserMessage", "text": "ignored event"}],
+    )
+
+    assert result.is_blocked is False
+    assert mock_llm_call.await_args.args[1] == [
+        {
+            "type": "system",
+            "content": f"Stay on topic.\n\n{TOPIC_SAFETY_OUTPUT_RESTRICTION}",
+        },
+        *canonical_messages,
+    ]
+
+
+@pytest.mark.asyncio
+async def test_legacy_events_preserve_conversation_order_and_append_current_input(task_manager):
+    result, mock_llm_call = await _run_topic_safety(
+        task_manager,
+        events=[
+            {"type": "UserMessage", "text": "earlier question"},
+            {"type": "StartUtteranceBotAction", "script": "earlier answer"},
+        ],
+    )
+
+    assert result.is_blocked is False
+    assert mock_llm_call.await_args.args[1] == [
+        {
+            "type": "system",
+            "content": f"Stay on topic.\n\n{TOPIC_SAFETY_OUTPUT_RESTRICTION}",
+        },
+        {"role": "user", "content": "earlier question"},
+        {"role": "assistant", "content": "earlier answer"},
+        {"type": "user", "content": "current question"},
+    ]
+
+
+@pytest.mark.asyncio
+async def test_absent_history_still_checks_current_input(task_manager):
+    result, mock_llm_call = await _run_topic_safety(task_manager)
+
+    assert result.is_blocked is False
+    assert mock_llm_call.await_args.args[1] == [
+        {
+            "type": "system",
+            "content": f"Stay on topic.\n\n{TOPIC_SAFETY_OUTPUT_RESTRICTION}",
+        },
+        {"type": "user", "content": "current question"},
+    ]
+
+
+@pytest.mark.asyncio
+async def test_empty_canonical_messages_take_precedence_over_legacy_history(task_manager):
+    result, mock_llm_call = await _run_topic_safety(
+        task_manager,
+        messages=[],
+        events=[{"type": "UserMessage", "text": "ignored event"}],
+    )
+
+    assert result.is_blocked is False
+    assert mock_llm_call.await_args.args[1] == [
+        {
+            "type": "system",
+            "content": f"Stay on topic.\n\n{TOPIC_SAFETY_OUTPUT_RESTRICTION}",
+        }
+    ]

tests/test_topic_safety_railoutcome_flow.py

@@ -23,6 +23,7 @@
 """
 
 import textwrap
+from unittest.mock import AsyncMock
 
 from nemoguardrails import RailsConfig
 from tests.utils import FakeLLMModel, TestChat
@@ -69,3 +70,24 @@ def test_on_topic_input_passes_through_railoutcome():
     chat = _chat_with_verdict("on-topic")
     response = chat.app.generate(messages=[{"role": "user", "content": "a relevant question"}])
     assert response["content"] == "Hello! How can I help you?"
+
+
+def test_multiturn_history_reaches_topic_safety_once_and_in_order():
+    chat = _chat_with_verdict("on-topic")
+    topic_model = chat.app.runtime.registered_action_params["llms"]["topic_control"]
+    topic_model.generate_async = AsyncMock(wraps=topic_model.generate_async)
+
+    chat.app.generate(
+        messages=[
+            {"role": "user", "content": "earlier question"},
+            {"role": "assistant", "content": "earlier answer"},
+            {"role": "user", "content": "current question"},
+        ]
+    )
+
+    prompt = topic_model.generate_async.await_args.args[0]
+    assert [(message.role, message.content) for message in prompt[1:]] == [
+        ("user", "earlier question"),
+        ("assistant", "earlier answer"),
+        ("user", "current question"),
+    ]