crewAI/tests/test_task_guardrails.py

from unittest.mock import Mock, patch

import pytest

from crewai import Agent, Task
from crewai.events.event_bus import crewai_event_bus
from crewai.events.event_types import (
    LLMGuardrailCompletedEvent,
    LLMGuardrailStartedEvent,
)
from crewai.llm import LLM
from crewai.tasks.hallucination_guardrail import HallucinationGuardrail
from crewai.tasks.llm_guardrail import LLMGuardrail
from crewai.tasks.task_output import TaskOutput


def test_task_without_guardrail():
    """Test that tasks work normally without guardrails (backward compatibility)."""
    agent = Mock()
    agent.role = "test_agent"
    agent.execute_task.return_value = "test result"
    agent.crew = None

    task = Task(description="Test task", expected_output="Output")

    result = task.execute_sync(agent=agent)
    assert isinstance(result, TaskOutput)
    assert result.raw == "test result"


def test_task_with_successful_guardrail_func():
    """Test that successful guardrail validation passes transformed result."""

    def guardrail(result: TaskOutput):
        return (True, result.raw.upper())

    agent = Mock()
    agent.role = "test_agent"
    agent.execute_task.return_value = "test result"
    agent.crew = None

    task = Task(description="Test task", expected_output="Output", guardrail=guardrail)

    result = task.execute_sync(agent=agent)
    assert isinstance(result, TaskOutput)
    assert result.raw == "TEST RESULT"


def test_task_with_failing_guardrail():
    """Test that failing guardrail triggers retry with error context."""

    def guardrail(result: TaskOutput):
        return (False, "Invalid format")

    agent = Mock()
    agent.role = "test_agent"
    agent.execute_task.side_effect = ["bad result", "good result"]
    agent.crew = None

    task = Task(
        description="Test task",
        expected_output="Output",
        guardrail=guardrail,
        guardrail_max_retries=1,
    )

    # First execution fails guardrail, second succeeds
    agent.execute_task.side_effect = ["bad result", "good result"]
    with pytest.raises(Exception) as exc_info:
        task.execute_sync(agent=agent)

    assert "Task failed guardrail validation" in str(exc_info.value)
    assert task.retry_count == 1


def test_task_with_guardrail_retries():
    """Test that guardrail respects max_retries configuration."""

    def guardrail(result: TaskOutput):
        return (False, "Invalid format")

    agent = Mock()
    agent.role = "test_agent"
    agent.execute_task.return_value = "bad result"
    agent.crew = None

    task = Task(
        description="Test task",
        expected_output="Output",
        guardrail=guardrail,
        guardrail_max_retries=2,
    )

    with pytest.raises(Exception) as exc_info:
        task.execute_sync(agent=agent)

    assert task.retry_count == 2
    assert "Task failed guardrail validation after 2 retries" in str(exc_info.value)
    assert "Invalid format" in str(exc_info.value)


def test_guardrail_error_in_context():
    """Test that guardrail error is passed in context for retry."""

    def guardrail(result: TaskOutput):
        return (False, "Expected JSON, got string")

    agent = Mock()
    agent.role = "test_agent"
    agent.crew = None

    task = Task(
        description="Test task",
        expected_output="Output",
        guardrail=guardrail,
        guardrail_max_retries=1,
    )

    # Mock execute_task to succeed on second attempt
    first_call = True

    def execute_task(task, context, tools):
        nonlocal first_call
        if first_call:
            first_call = False
            return "invalid"
        return '{"valid": "json"}'

    agent.execute_task.side_effect = execute_task

    with pytest.raises(Exception) as exc_info:
        task.execute_sync(agent=agent)

    assert "Task failed guardrail validation" in str(exc_info.value)
    assert "Expected JSON, got string" in str(exc_info.value)


@pytest.fixture
def sample_agent():
    return Agent(role="Test Agent", goal="Test Goal", backstory="Test Backstory")


@pytest.fixture
def task_output():
    return TaskOutput(
        raw="""
        Lorem Ipsum is simply dummy text of the printing and typesetting industry. Lorem Ipsum has been the industry's standard dummy text ever
        """,
        description="Test task",
        expected_output="Output",
        agent="Test Agent",
    )


@pytest.mark.vcr(filter_headers=["authorization"])
def test_task_guardrail_process_output(task_output):
    guardrail = LLMGuardrail(
        description="Ensure the result has less than 10 words", llm=LLM(model="gpt-4o")
    )

    result = guardrail(task_output)
    assert result[0] is False

    assert "exceeding the guardrail limit of fewer than" in result[1].lower()

    guardrail = LLMGuardrail(
        description="Ensure the result has less than 500 words", llm=LLM(model="gpt-4o")
    )

    result = guardrail(task_output)
    assert result[0] is True
    assert result[1] == task_output.raw


@pytest.mark.vcr(filter_headers=["authorization"])
def test_guardrail_emits_events(sample_agent):
    started_guardrail = []
    completed_guardrail = []

    task = Task(
        description="Gather information about available books on the First World War",
        agent=sample_agent,
        expected_output="A list of available books on the First World War",
        guardrail="Ensure the authors are from Italy",
    )

    with crewai_event_bus.scoped_handlers():

        @crewai_event_bus.on(LLMGuardrailStartedEvent)
        def handle_guardrail_started(source, event):
            assert source == task
            started_guardrail.append(
                {"guardrail": event.guardrail, "retry_count": event.retry_count}
            )

        @crewai_event_bus.on(LLMGuardrailCompletedEvent)
        def handle_guardrail_completed(source, event):
            assert source == task
            completed_guardrail.append(
                {
                    "success": event.success,
                    "result": event.result,
                    "error": event.error,
                    "retry_count": event.retry_count,
                }
            )

        result = task.execute_sync(agent=sample_agent)

        def custom_guardrail(result: TaskOutput):
            return (True, "good result from callable function")

        task = Task(
            description="Test task",
            expected_output="Output",
            guardrail=custom_guardrail,
        )

        task.execute_sync(agent=sample_agent)

        expected_started_events = [
            {"guardrail": "Ensure the authors are from Italy", "retry_count": 0},
            {"guardrail": "Ensure the authors are from Italy", "retry_count": 1},
            {
                "guardrail": """def custom_guardrail(result: TaskOutput):
            return (True, "good result from callable function")""",
                "retry_count": 0,
            },
        ]

        expected_completed_events = [
            {
                "success": False,
                "result": None,
                "error": "The task result does not comply with the guardrail because none of "
                "the listed authors are from Italy. All authors mentioned are from "
                "different countries, including Germany, the UK, the USA, and others, "
                "which violates the requirement that authors must be Italian.",
                "retry_count": 0,
            },
            {"success": True, "result": result.raw, "error": None, "retry_count": 1},
            {
                "success": True,
                "result": "good result from callable function",
                "error": None,
                "retry_count": 0,
            },
        ]
        assert started_guardrail == expected_started_events
        assert completed_guardrail == expected_completed_events


@pytest.mark.vcr(filter_headers=["authorization"])
def test_guardrail_when_an_error_occurs(sample_agent, task_output):
    with (
        patch(
            "crewai.Agent.kickoff",
            side_effect=Exception("Unexpected error"),
        ),
        pytest.raises(
            Exception,
            match="Error while validating the task output: Unexpected error",
        ),
    ):
        task = Task(
            description="Gather information about available books on the First World War",
            agent=sample_agent,
            expected_output="A list of available books on the First World War",
            guardrail="Ensure the authors are from Italy",
            guardrail_max_retries=0,
        )
        task.execute_sync(agent=sample_agent)


def test_hallucination_guardrail_integration():
    """Test that HallucinationGuardrail integrates properly with the task system."""
    agent = Mock()
    agent.role = "test_agent"
    agent.execute_task.return_value = "test result"
    agent.crew = None

    mock_llm = Mock(spec=LLM)
    guardrail = HallucinationGuardrail(
        context="Test reference context for validation", llm=mock_llm, threshold=8.0
    )

    task = Task(
        description="Test task with hallucination guardrail",
        expected_output="Valid output",
        guardrail=guardrail,
    )

    result = task.execute_sync(agent=agent)
    assert isinstance(result, TaskOutput)
    assert result.raw == "test result"


def test_hallucination_guardrail_description_in_events():
    """Test that HallucinationGuardrail description appears correctly in events."""
    mock_llm = Mock(spec=LLM)
    guardrail = HallucinationGuardrail(context="Test context", llm=mock_llm)

    assert guardrail.description == "HallucinationGuardrail (no-op)"

    event = LLMGuardrailStartedEvent(guardrail=guardrail, retry_count=0)
    assert event.guardrail == "HallucinationGuardrail (no-op)"