From c074b0500239533aea7b33eb6484a6150dfe3955 Mon Sep 17 00:00:00 2001 From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Sat, 30 May 2026 08:33:26 +0000 Subject: [PATCH] fix: enforce guardrail re-validation on retry and reject negative max_retries MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Fixes #5979 — guardrails are now enforced as hard constraints. Two bugs fixed: 1. Multiple guardrails skipped re-validation of earlier guardrails after retry. When guardrail N failed and the agent retried, the new output was only checked against guardrail N onward. Earlier guardrails (0..N-1) were not re-evaluated, allowing retry outputs to silently violate them. Fix: refactored _invoke_guardrail_function into _run_guardrails which runs ALL guardrails from the beginning on each retry attempt. 2. Negative guardrail_max_retries bypassed guardrails entirely. Setting guardrail_max_retries to a negative value made the guardrail loop execute zero iterations, letting output pass without validation. Fix: added ge=0 constraint to the guardrail_max_retries field. Co-Authored-By: João --- lib/crewai/src/crewai/task.py | 429 +++++++++++++---------- lib/crewai/tests/test_task_guardrails.py | 201 ++++++++++- 2 files changed, 439 insertions(+), 191 deletions(-) diff --git a/lib/crewai/src/crewai/task.py b/lib/crewai/src/crewai/task.py index c63cfe866..84a071b0c 100644 --- a/lib/crewai/src/crewai/task.py +++ b/lib/crewai/src/crewai/task.py @@ -271,7 +271,9 @@ class Task(BaseModel): description="[DEPRECATED] Maximum number of retries when guardrail fails. Use guardrail_max_retries instead. Will be removed in v1.0.0", ) guardrail_max_retries: int = Field( - default=3, description="Maximum number of retries when guardrail fails" + default=3, + description="Maximum number of retries when guardrail fails", + ge=0, ) retry_count: int = Field(default=0, description="Current number of retries") start_time: datetime.datetime | None = Field( @@ -700,23 +702,11 @@ class Task(BaseModel): messages=agent.last_messages, # type: ignore[attr-defined] ) - if self._guardrails: - for idx, guardrail in enumerate(self._guardrails): - task_output = await self._ainvoke_guardrail_function( - task_output=task_output, - agent=agent, - tools=tools, - guardrail=guardrail, - guardrail_index=idx, - ) - - if self._guardrail: - task_output = await self._ainvoke_guardrail_function( - task_output=task_output, - agent=agent, - tools=tools, - guardrail=self._guardrail, - ) + task_output = await self._arun_guardrails( + task_output=task_output, + agent=agent, + tools=tools, + ) self.output = task_output self.end_time = datetime.datetime.now() @@ -825,23 +815,11 @@ class Task(BaseModel): messages=agent.last_messages, # type: ignore[attr-defined] ) - if self._guardrails: - for idx, guardrail in enumerate(self._guardrails): - task_output = self._invoke_guardrail_function( - task_output=task_output, - agent=agent, - tools=tools, - guardrail=guardrail, - guardrail_index=idx, - ) - - if self._guardrail: - task_output = self._invoke_guardrail_function( - task_output=task_output, - agent=agent, - tools=tools, - guardrail=self._guardrail, - ) + task_output = self._run_guardrails( + task_output=task_output, + agent=agent, + tools=tools, + ) self.output = task_output self.end_time = datetime.datetime.now() @@ -1243,221 +1221,296 @@ Follow these guidelines: """ return self.security_config.fingerprint - def _invoke_guardrail_function( + def _collect_guardrails(self) -> list[tuple[int | None, GuardrailCallable]]: + """Collect all guardrails into a single indexed list. + + Returns a list of ``(index, callable)`` pairs. When ``_guardrails`` + is populated the index is the position in that list; for the legacy + single ``_guardrail`` the index is ``None``. + """ + guardrails: list[tuple[int | None, GuardrailCallable]] = [] + if self._guardrails: + for idx, g in enumerate(self._guardrails): + guardrails.append((idx, g)) + elif self._guardrail: + guardrails.append((None, self._guardrail)) + return guardrails + + def _apply_guardrail_result( + self, + task_output: TaskOutput, + guardrail_result: Any, + ) -> TaskOutput: + """Apply a successful guardrail result to the task output (sync).""" + if guardrail_result.result is None: + raise Exception( + "Task guardrail returned None as result. This is not allowed." + ) + + if isinstance(guardrail_result.result, str): + task_output.raw = guardrail_result.result + pydantic_output, json_output = self._export_output( + guardrail_result.result + ) + task_output.pydantic = pydantic_output + task_output.json_dict = json_output + elif isinstance(guardrail_result.result, TaskOutput): + task_output = guardrail_result.result + + return task_output + + async def _aapply_guardrail_result( + self, + task_output: TaskOutput, + guardrail_result: Any, + ) -> TaskOutput: + """Apply a successful guardrail result to the task output (async).""" + if guardrail_result.result is None: + raise Exception( + "Task guardrail returned None as result. This is not allowed." + ) + + if isinstance(guardrail_result.result, str): + task_output.raw = guardrail_result.result + pydantic_output, json_output = await self._aexport_output( + guardrail_result.result + ) + task_output.pydantic = pydantic_output + task_output.json_dict = json_output + elif isinstance(guardrail_result.result, TaskOutput): + task_output = guardrail_result.result + + return task_output + + def _rebuild_task_output( + self, + result: Any, + agent: BaseAgent, + ) -> TaskOutput: + """Build a new TaskOutput from an agent execution result (sync).""" + if isinstance(result, BaseModel): + raw = result.model_dump_json() + if self.output_pydantic: + pydantic_output = result + json_output = None + elif self.output_json: + pydantic_output = None + json_output = result.model_dump() + else: + pydantic_output = None + json_output = None + else: + raw = result + pydantic_output, json_output = self._export_output(result) + + return TaskOutput( + name=self.name or self.description, + description=self.description, + expected_output=self.expected_output, + raw=raw, + pydantic=pydantic_output, + json_dict=json_output, + agent=agent.role, + output_format=self._get_output_format(), + messages=agent.last_messages, # type: ignore[attr-defined] + ) + + async def _arebuild_task_output( + self, + result: Any, + agent: BaseAgent, + ) -> TaskOutput: + """Build a new TaskOutput from an agent execution result (async).""" + if isinstance(result, BaseModel): + raw = result.model_dump_json() + if self.output_pydantic: + pydantic_output = result + json_output = None + elif self.output_json: + pydantic_output = None + json_output = result.model_dump() + else: + pydantic_output = None + json_output = None + else: + raw = result + pydantic_output, json_output = await self._aexport_output(result) + + return TaskOutput( + name=self.name or self.description, + description=self.description, + expected_output=self.expected_output, + raw=raw, + pydantic=pydantic_output, + json_dict=json_output, + agent=agent.role, + output_format=self._get_output_format(), + messages=agent.last_messages, # type: ignore[attr-defined] + ) + + def _run_guardrails( self, task_output: TaskOutput, agent: BaseAgent, tools: list[BaseTool], - guardrail: GuardrailCallable | None, - guardrail_index: int | None = None, ) -> TaskOutput: - if not guardrail: + """Run all guardrails with proper enforcement. + + When any guardrail fails and a retry is triggered, ALL guardrails are + re-evaluated from the beginning on the new output. This prevents + earlier guardrails from being silently bypassed by retry outputs. + """ + guardrails = self._collect_guardrails() + if not guardrails: return task_output - if guardrail_index is not None: - current_retry_count = self._guardrail_retry_counts.get(guardrail_index, 0) - else: - current_retry_count = self.retry_count + max_retries = self.guardrail_max_retries + retry_count = 0 - max_attempts = self.guardrail_max_retries + 1 + while True: + failed = False + failed_guardrail_index: int | None = None + failed_error: str | None = None - for attempt in range(max_attempts): - guardrail_result = process_guardrail( - output=task_output, - guardrail=guardrail, - retry_count=current_retry_count, - event_source=self, - from_task=self, - from_agent=agent, - ) + for guardrail_index, guardrail in guardrails: + guardrail_result = process_guardrail( + output=task_output, + guardrail=guardrail, + retry_count=retry_count, + event_source=self, + from_task=self, + from_agent=agent, + ) - if guardrail_result.success: - if guardrail_result.result is None: - raise Exception( - "Task guardrail returned None as result. This is not allowed." + if guardrail_result.success: + task_output = self._apply_guardrail_result( + task_output, guardrail_result ) + continue - if isinstance(guardrail_result.result, str): - task_output.raw = guardrail_result.result - pydantic_output, json_output = self._export_output( - guardrail_result.result - ) - task_output.pydantic = pydantic_output - task_output.json_dict = json_output - elif isinstance(guardrail_result.result, TaskOutput): - task_output = guardrail_result.result + failed = True + failed_guardrail_index = guardrail_index + failed_error = guardrail_result.error + break + if not failed: return task_output - if attempt >= self.guardrail_max_retries: + if retry_count >= max_retries: guardrail_name = ( - f"guardrail {guardrail_index}" - if guardrail_index is not None + f"guardrail {failed_guardrail_index}" + if failed_guardrail_index is not None else "guardrail" ) raise Exception( - f"Task failed {guardrail_name} validation after {self.guardrail_max_retries} retries. " - f"Last error: {guardrail_result.error}" + f"Task failed {guardrail_name} validation after " + f"{max_retries} retries. Last error: {failed_error}" ) - if guardrail_index is not None: - current_retry_count += 1 - self._guardrail_retry_counts[guardrail_index] = current_retry_count - else: - self.retry_count += 1 - current_retry_count = self.retry_count + retry_count += 1 + self.retry_count = retry_count + if failed_guardrail_index is not None: + self._guardrail_retry_counts[failed_guardrail_index] = ( + self._guardrail_retry_counts.get(failed_guardrail_index, 0) + 1 + ) context = I18N_DEFAULT.errors("validation_error").format( - guardrail_result_error=guardrail_result.error, + guardrail_result_error=failed_error, task_output=task_output.raw, ) if agent and agent.verbose: + max_attempts = max_retries + 1 PRINTER.print( - content=f"Guardrail {guardrail_index if guardrail_index is not None else ''} blocked (attempt {attempt + 1}/{max_attempts}), retrying due to: {guardrail_result.error}\n", + content=( + f"Guardrail {failed_guardrail_index if failed_guardrail_index is not None else ''} " + f"blocked (attempt {retry_count}/{max_attempts}), " + f"retrying due to: {failed_error}\n" + ), color="yellow", ) result = agent.execute_task( - task=self, - context=context, - tools=tools, + task=self, context=context, tools=tools ) + task_output = self._rebuild_task_output(result, agent) - if isinstance(result, BaseModel): - raw = result.model_dump_json() - if self.output_pydantic: - pydantic_output = result - json_output = None - elif self.output_json: - pydantic_output = None - json_output = result.model_dump() - else: - pydantic_output = None - json_output = None - else: - raw = result - pydantic_output, json_output = self._export_output(result) - - task_output = TaskOutput( - name=self.name or self.description, - description=self.description, - expected_output=self.expected_output, - raw=raw, - pydantic=pydantic_output, - json_dict=json_output, - agent=agent.role, - output_format=self._get_output_format(), - messages=agent.last_messages, # type: ignore[attr-defined] - ) - - return task_output - - async def _ainvoke_guardrail_function( + async def _arun_guardrails( self, task_output: TaskOutput, agent: BaseAgent, tools: list[BaseTool], - guardrail: GuardrailCallable | None, - guardrail_index: int | None = None, ) -> TaskOutput: - """Invoke the guardrail function asynchronously.""" - if not guardrail: + """Async version of ``_run_guardrails``.""" + guardrails = self._collect_guardrails() + if not guardrails: return task_output - if guardrail_index is not None: - current_retry_count = self._guardrail_retry_counts.get(guardrail_index, 0) - else: - current_retry_count = self.retry_count + max_retries = self.guardrail_max_retries + retry_count = 0 - max_attempts = self.guardrail_max_retries + 1 + while True: + failed = False + failed_guardrail_index: int | None = None + failed_error: str | None = None - for attempt in range(max_attempts): - guardrail_result = process_guardrail( - output=task_output, - guardrail=guardrail, - retry_count=current_retry_count, - event_source=self, - from_task=self, - from_agent=agent, - ) + for guardrail_index, guardrail in guardrails: + guardrail_result = process_guardrail( + output=task_output, + guardrail=guardrail, + retry_count=retry_count, + event_source=self, + from_task=self, + from_agent=agent, + ) - if guardrail_result.success: - if guardrail_result.result is None: - raise Exception( - "Task guardrail returned None as result. This is not allowed." + if guardrail_result.success: + task_output = await self._aapply_guardrail_result( + task_output, guardrail_result ) + continue - if isinstance(guardrail_result.result, str): - task_output.raw = guardrail_result.result - pydantic_output, json_output = await self._aexport_output( - guardrail_result.result - ) - task_output.pydantic = pydantic_output - task_output.json_dict = json_output - elif isinstance(guardrail_result.result, TaskOutput): - task_output = guardrail_result.result + failed = True + failed_guardrail_index = guardrail_index + failed_error = guardrail_result.error + break + if not failed: return task_output - if attempt >= self.guardrail_max_retries: + if retry_count >= max_retries: guardrail_name = ( - f"guardrail {guardrail_index}" - if guardrail_index is not None + f"guardrail {failed_guardrail_index}" + if failed_guardrail_index is not None else "guardrail" ) raise Exception( - f"Task failed {guardrail_name} validation after {self.guardrail_max_retries} retries. " - f"Last error: {guardrail_result.error}" + f"Task failed {guardrail_name} validation after " + f"{max_retries} retries. Last error: {failed_error}" ) - if guardrail_index is not None: - current_retry_count += 1 - self._guardrail_retry_counts[guardrail_index] = current_retry_count - else: - self.retry_count += 1 - current_retry_count = self.retry_count + retry_count += 1 + self.retry_count = retry_count + if failed_guardrail_index is not None: + self._guardrail_retry_counts[failed_guardrail_index] = ( + self._guardrail_retry_counts.get(failed_guardrail_index, 0) + 1 + ) context = I18N_DEFAULT.errors("validation_error").format( - guardrail_result_error=guardrail_result.error, + guardrail_result_error=failed_error, task_output=task_output.raw, ) if agent and agent.verbose: + max_attempts = max_retries + 1 PRINTER.print( - content=f"Guardrail {guardrail_index if guardrail_index is not None else ''} blocked (attempt {attempt + 1}/{max_attempts}), retrying due to: {guardrail_result.error}\n", + content=( + f"Guardrail {failed_guardrail_index if failed_guardrail_index is not None else ''} " + f"blocked (attempt {retry_count}/{max_attempts}), " + f"retrying due to: {failed_error}\n" + ), color="yellow", ) result = await agent.aexecute_task( - task=self, - context=context, - tools=tools, + task=self, context=context, tools=tools ) - - if isinstance(result, BaseModel): - raw = result.model_dump_json() - if self.output_pydantic: - pydantic_output = result - json_output = None - elif self.output_json: - pydantic_output = None - json_output = result.model_dump() - else: - pydantic_output = None - json_output = None - else: - raw = result - pydantic_output, json_output = await self._aexport_output(result) - - task_output = TaskOutput( - name=self.name or self.description, - description=self.description, - expected_output=self.expected_output, - raw=raw, - pydantic=pydantic_output, - json_dict=json_output, - agent=agent.role, - output_format=self._get_output_format(), - messages=agent.last_messages, # type: ignore[attr-defined] - ) - - return task_output + task_output = await self._arebuild_task_output(result, agent) diff --git a/lib/crewai/tests/test_task_guardrails.py b/lib/crewai/tests/test_task_guardrails.py index 1f3cbd60a..53a33678d 100644 --- a/lib/crewai/tests/test_task_guardrails.py +++ b/lib/crewai/tests/test_task_guardrails.py @@ -480,7 +480,12 @@ def test_multiple_guardrails_with_mixed_string_and_taskoutput(): def test_multiple_guardrails_with_retry_on_middle_guardrail(): - """Test that retry works correctly when a middle guardrail fails.""" + """Test that retry works correctly when a middle guardrail fails. + + When the second guardrail fails, the task is re-executed and ALL + guardrails are re-evaluated from the beginning on the new output. + This ensures earlier guardrails are not silently bypassed. + """ call_count = {"first": 0, "second": 0, "third": 0} @@ -516,10 +521,14 @@ def test_multiple_guardrails_with_retry_on_middle_guardrail(): result = task.execute_sync(agent=agent) assert task._guardrail_retry_counts.get(1, 0) == 1 - assert call_count["first"] == 1 + # first_guardrail is called twice: once in the initial pass and + # once after the retry restarts all guardrails from the beginning. + assert call_count["first"] == 2 assert call_count["second"] == 2 assert call_count["third"] == 1 assert "Second(2)" in result.raw + # Verify first guardrail is also properly applied on the retried output + assert "First(2)" in result.raw def test_multiple_guardrails_with_max_retries_exceeded(): @@ -774,8 +783,194 @@ def test_per_guardrail_independent_retry_tracking(): assert task._guardrail_retry_counts.get(1, 0) == 1 assert task._guardrail_retry_counts.get(2, 0) == 0 - assert call_counts["g1"] == 3 + # g1 is called 4 times: twice failing, once succeeding (attempt 3), + # then once more when g2's failure triggers a full restart. + assert call_counts["g1"] == 4 assert call_counts["g2"] == 2 assert call_counts["g3"] == 1 assert "G3(1)" in result.raw + assert "G1(4)" in result.raw + + +def test_retry_revalidates_earlier_guardrails(): + """Test that when a later guardrail fails and triggers a retry, the new + output is re-validated against ALL guardrails from the beginning. + + This is the core fix for issue #5979: guardrails must be enforced as + constraints, not suggestions. A retry output that violates an earlier + guardrail must be caught. + """ + + def length_guardrail(result: TaskOutput) -> tuple[bool, str]: + """Require output to be at least 20 characters.""" + if len(result.raw) < 20: + return (False, f"Output too short ({len(result.raw)} chars, need 20+)") + return (True, result.raw) + + def no_digits_guardrail(result: TaskOutput) -> tuple[bool, str]: + """Reject output containing digits.""" + if any(c.isdigit() for c in result.raw): + return (False, "Output must not contain digits") + return (True, result.raw) + + call_count = 0 + + def mock_execute_task(*args, **kwargs): + nonlocal call_count + call_count += 1 + if call_count == 1: + # First attempt: long enough but contains digits → passes length, + # fails no_digits + return "this text has digits 123 in it" + if call_count == 2: + # Second attempt (after no_digits retry): short, no digits → + # would pass no_digits but must fail length on re-validation + return "no digits here" + # Third attempt: long enough, no digits → passes both + return "this is a valid long text without any numbers" + + agent = Mock() + agent.role = "revalidation_agent" + agent.execute_task = mock_execute_task + agent.crew = None + agent.last_messages = [] + + task = create_smart_task( + description="Test re-validation of earlier guardrails", + expected_output="Valid text", + guardrails=[length_guardrail, no_digits_guardrail], + guardrail_max_retries=3, + ) + + result = task.execute_sync(agent=agent) + # The final output must satisfy BOTH guardrails + assert len(result.raw) >= 20 + assert not any(c.isdigit() for c in result.raw) + assert result.raw == "this is a valid long text without any numbers" + # Three task executions: original + two retries + assert call_count == 3 + assert task.retry_count == 2 + + +def test_negative_guardrail_max_retries_rejected(): + """Test that negative guardrail_max_retries raises a validation error.""" + with pytest.raises(Exception): + create_smart_task( + description="Test negative max retries", + expected_output="Should fail", + guardrail_max_retries=-1, + ) + + +def test_zero_guardrail_max_retries_no_retry(): + """Test that guardrail_max_retries=0 means no retries — fail immediately.""" + + def failing_guardrail(result: TaskOutput) -> tuple[bool, str]: + return (False, "Always fails") + + agent = Mock() + agent.role = "no_retry_agent" + agent.execute_task.return_value = "test" + agent.crew = None + agent.last_messages = [] + + task = create_smart_task( + description="Test zero max retries", + expected_output="Output", + guardrail=failing_guardrail, + guardrail_max_retries=0, + ) + + with pytest.raises(Exception) as exc_info: + task.execute_sync(agent=agent) + + assert "after 0 retries" in str(exc_info.value) + # Agent should not have been asked to retry + assert agent.execute_task.call_count == 1 + + +def test_single_guardrail_retry_still_works(): + """Test that single guardrail retry logic is preserved in the new + _run_guardrails implementation.""" + + call_count = 0 + + def guardrail(result: TaskOutput) -> tuple[bool, str]: + nonlocal call_count + call_count += 1 + if call_count <= 2: + return (False, f"Attempt {call_count} failed") + return (True, result.raw.upper()) + + agent = Mock() + agent.role = "single_retry_agent" + agent.execute_task.return_value = "hello world" + agent.crew = None + agent.last_messages = [] + + task = create_smart_task( + description="Test single guardrail retry", + expected_output="Output", + guardrail=guardrail, + guardrail_max_retries=3, + ) + + result = task.execute_sync(agent=agent) + assert result.raw == "HELLO WORLD" + assert call_count == 3 + assert task.retry_count == 2 + + +def test_retry_restart_catches_earlier_guardrail_violation(): + """Regression test: ensure that after a retry triggered by guardrail N, + the new output is checked by guardrails 0..N-1 as well. + + Scenario: + g0 = length check (>= 10 chars) + g1 = no-uppercase check + Original output: 'HELLO WORLD LONG TEXT' → passes g0, fails g1 → retry + Retry output: 'short' → must fail g0 (re-validation catches it) → retry again + Final output: 'this works fine' → passes both + """ + + call_count = 0 + + def length_check(result: TaskOutput) -> tuple[bool, str]: + if len(result.raw) < 10: + return (False, "Too short") + return (True, result.raw) + + def no_uppercase(result: TaskOutput) -> tuple[bool, str]: + if result.raw != result.raw.lower(): + return (False, "Must be lowercase") + return (True, result.raw) + + def mock_execute(*args, **kwargs): + nonlocal call_count + call_count += 1 + if call_count == 1: + return "HELLO WORLD LONG TEXT" + if call_count == 2: + return "short" + return "this works fine" + + agent = Mock() + agent.role = "regression_agent" + agent.execute_task = mock_execute + agent.crew = None + agent.last_messages = [] + + task = create_smart_task( + description="Regression test", + expected_output="Valid text", + guardrails=[length_check, no_uppercase], + guardrail_max_retries=3, + ) + + result = task.execute_sync(agent=agent) + assert result.raw == "this works fine" + assert call_count == 3 + # retry 1 triggered by g1 (no_uppercase), retry 2 triggered by g0 (length_check) + assert task._guardrail_retry_counts.get(0, 0) == 1 + assert task._guardrail_retry_counts.get(1, 0) == 1