diff --git a/lib/crewai/src/crewai/utilities/reasoning_handler.py b/lib/crewai/src/crewai/utilities/reasoning_handler.py index 1028a3f3de..5ddef4aad1 100644 --- a/lib/crewai/src/crewai/utilities/reasoning_handler.py +++ b/lib/crewai/src/crewai/utilities/reasoning_handler.py @@ -4,6 +4,7 @@ import json import logging +import re from typing import TYPE_CHECKING, Any, Final, Literal, cast from pydantic import BaseModel, Field @@ -27,6 +28,36 @@ from crewai.task import Task +_READY_MARKER: Final[re.Pattern[str]] = re.compile( + r"\bNOT\s+READY\b|\bREADY\b", re.IGNORECASE +) + + +def _detect_plan_ready(text: str) -> bool: + """Detect whether a text reasoning plan concludes with READY. + + The reasoning prompts instruct the model to conclude with "READY" or + "NOT READY", but models (especially on the text-parsing path) often emit a + bare ``READY`` rather than the full "READY: I am ready to execute the task." + sentence, which the previous exact-substring check missed (#6204). + + Matches the LAST readiness marker so a "not ready" mentioned earlier in the + plan doesn't override a final "READY", and so "NOT READY" wins over the bare + "READY" it contains. + + Args: + text: The raw model response. + + Returns: + True if the concluding marker is READY, False for NOT READY or if no + marker is present. + """ + markers = _READY_MARKER.findall(text) + if not markers: + return False + return "NOT" not in markers[-1].upper() + + class ReasoningPlan(BaseModel): """Model representing a reasoning plan for a task.""" @@ -409,7 +440,7 @@ def _create_reasoning_plan( return ( response_str, [], - "READY: I am ready to execute the task." in response_str, + _detect_plan_ready(response_str), ) except Exception as e: @@ -433,7 +464,7 @@ def _create_reasoning_plan( return ( fallback_str, [], - "READY: I am ready to execute the task." in fallback_str, + _detect_plan_ready(fallback_str), ) except Exception as inner_e: self.logger.error(f"Error during fallback text parsing: {inner_e!s}") @@ -593,7 +624,7 @@ def _parse_planning_response(response: str) -> tuple[str, bool]: return "No plan was generated.", False plan = response - ready = "READY: I am ready to execute the task." in response + ready = _detect_plan_ready(response) return plan, ready diff --git a/lib/crewai/tests/utilities/test_structured_planning.py b/lib/crewai/tests/utilities/test_structured_planning.py index 1e36c6de97..8f03737b0c 100644 --- a/lib/crewai/tests/utilities/test_structured_planning.py +++ b/lib/crewai/tests/utilities/test_structured_planning.py @@ -17,9 +17,52 @@ FUNCTION_SCHEMA, AgentReasoning, ReasoningPlan, + _detect_plan_ready, ) +class TestDetectPlanReady: + """Readiness detection must accept a bare READY marker, not only the full + 'READY: I am ready to execute the task.' sentence (#6204).""" + + def test_bare_ready_marker_is_ready(self): + assert _detect_plan_ready("1. do X\n2. do Y\n\n---\n\nREADY") is True + + def test_full_sentence_still_ready(self): + assert _detect_plan_ready("plan...\nREADY: I am ready to execute the task.") + + def test_not_ready_marker_is_not_ready(self): + assert ( + _detect_plan_ready("plan...\n\nNOT READY: I need more detail.") is False + ) + + def test_last_marker_wins_over_mid_text_not_ready(self): + text = "Step 1 is not ready to run yet without input.\n\nREADY" + assert _detect_plan_ready(text) is True + + def test_case_insensitive(self): + assert _detect_plan_ready("plan...\n\nready") is True + assert _detect_plan_ready("plan...\n\nnot ready") is False + + def test_no_marker_defaults_to_not_ready(self): + assert _detect_plan_ready("just a plan with no verdict") is False + + def test_ready_substring_inside_words_is_ignored(self): + # "already" contains "ready" but is not a READY marker (word boundaries). + assert _detect_plan_ready("I am already prepared") is False + assert _detect_plan_ready("The steps are already documented") is False + + def test_trailing_already_does_not_flip_not_ready(self): + # A trailing "already" must not override a concluding NOT READY. + assert ( + _detect_plan_ready("NOT READY. I already have the tools.") is False + ) + + def test_parse_planning_response_detects_bare_ready(self): + _plan, ready = AgentReasoning._parse_planning_response("my plan\n\nREADY") + assert ready is True + + class TestFunctionSchema: """Tests for the FUNCTION_SCHEMA used in structured planning."""