mirror of
https://github.com/crewAIInc/crewAI.git
synced 2026-07-06 23:49:24 +00:00
fix: enforce guardrail re-validation on retry and reject negative max_retries
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 <joao@crewai.com>
This commit is contained in:
Devin AI
@@ -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)
|
||||
|
||||
@@ -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
|
||||
|
||||
Reference in New Issue
Block a user