diff --git a/lib/crewai/src/crewai/agents/step_executor.py b/lib/crewai/src/crewai/agents/step_executor.py index 81238f473..bc6a963f8 100644 --- a/lib/crewai/src/crewai/agents/step_executor.py +++ b/lib/crewai/src/crewai/agents/step_executor.py @@ -43,7 +43,7 @@ from crewai.utilities.agent_utils import ( setup_native_tools, ) from crewai.utilities.i18n import I18N_DEFAULT -from crewai.utilities.planning_types import TodoItem +from crewai.utilities.planning_types import StepOutcome, TodoItem from crewai.utilities.step_execution_context import StepExecutionContext, StepResult from crewai.utilities.string_utils import sanitize_tool_name from crewai.utilities.tool_utils import execute_tool_and_check_finality @@ -60,6 +60,21 @@ if TYPE_CHECKING: from crewai.tools.structured_tool import CrewStructuredTool +class _StepExecutionTerminatedError(Exception): + """Signal a bounded step termination while preserving partial results.""" + + def __init__( + self, + outcome: StepOutcome, + result: str, + reason: str, + ) -> None: + super().__init__(reason) + self.outcome = outcome + self.result = result + self.reason = reason + + class StepExecutor: """Executes a SINGLE todo item using direct-action execution. @@ -175,11 +190,20 @@ class StepExecutor: elapsed = time.monotonic() - start_time return StepResult( - success=True, + outcome="completed", result=result_text, tool_calls_made=tool_calls_made, execution_time=elapsed, ) + except _StepExecutionTerminatedError as termination: + elapsed = time.monotonic() - start_time + return StepResult( + outcome=termination.outcome, + result=termination.result, + termination_reason=termination.reason, + tool_calls_made=tool_calls_made, + execution_time=elapsed, + ) except Exception as e: if self._use_native_tools and is_native_tool_calling_unsupported_error(e): try: @@ -213,19 +237,28 @@ class StepExecutor: self._validate_expected_tool_usage(todo, tool_calls_made) elapsed = time.monotonic() - start_time return StepResult( - success=True, + outcome="completed", result=result_text, tool_calls_made=tool_calls_made, execution_time=elapsed, ) + except _StepExecutionTerminatedError as termination: + elapsed = time.monotonic() - start_time + return StepResult( + outcome=termination.outcome, + result=termination.result, + termination_reason=termination.reason, + tool_calls_made=tool_calls_made, + execution_time=elapsed, + ) except Exception as fallback_error: e = fallback_error elapsed = time.monotonic() - start_time return StepResult( - success=False, + outcome="failed", result="", - error=str(e), + termination_reason=str(e), tool_calls_made=tool_calls_made, execution_time=elapsed, ) @@ -337,7 +370,14 @@ class StepExecutor: if step_timeout and start_time: elapsed = time.monotonic() - start_time if elapsed >= step_timeout: - return last_tool_result or f"Step timed out after {elapsed:.0f}s" + reason = ( + f"Step timed out after {elapsed:.0f}s (limit: {step_timeout}s)." + ) + raise _StepExecutionTerminatedError( + "timed_out", + last_tool_result, + reason, + ) answer = self.llm.call( messages, callbacks=self.callbacks, @@ -364,7 +404,12 @@ class StepExecutor: return answer_str - return last_tool_result + raise _StepExecutionTerminatedError( + "iteration_exhausted", + last_tool_result, + f"Step reached max_step_iterations ({max_step_iterations}) " + "before producing a final answer.", + ) def _execute_text_tool_with_events( self, formatted: AgentAction, todo: TodoItem @@ -547,10 +592,13 @@ class StepExecutor: if step_timeout and start_time: elapsed = time.monotonic() - start_time if elapsed >= step_timeout: - return ( - "\n\n".join(accumulated_results) - if accumulated_results - else f"Step timed out after {elapsed:.0f}s" + reason = ( + f"Step timed out after {elapsed:.0f}s (limit: {step_timeout}s)." + ) + raise _StepExecutionTerminatedError( + "timed_out", + "\n\n".join(accumulated_results), + reason, ) answer = self.llm.call( messages, @@ -575,7 +623,12 @@ class StepExecutor: return str(answer) - return "\n".join(filter(None, accumulated_results)) + raise _StepExecutionTerminatedError( + "iteration_exhausted", + "\n".join(filter(None, accumulated_results)), + f"Step reached max_step_iterations ({max_step_iterations}) " + "before producing a final answer.", + ) def _execute_native_tool_calls( self, diff --git a/lib/crewai/src/crewai/events/types/observation_events.py b/lib/crewai/src/crewai/events/types/observation_events.py index 5f1ef6ea0..829905267 100644 --- a/lib/crewai/src/crewai/events/types/observation_events.py +++ b/lib/crewai/src/crewai/events/types/observation_events.py @@ -8,6 +8,7 @@ continuation, refinement, or replanning. from typing import Any, Literal from crewai.events.base_events import BaseEvent +from crewai.utilities.planning_types import StepOutcome class ObservationEvent(BaseEvent): @@ -54,6 +55,8 @@ class PlanStepCompletedEvent(PlanStepEvent): type: Literal["plan_step_completed"] = "plan_step_completed" success: bool = True + outcome: StepOutcome = "completed" + termination_reason: str | None = None result: str | None = None error: str | None = None diff --git a/lib/crewai/src/crewai/experimental/agent_executor.py b/lib/crewai/src/crewai/experimental/agent_executor.py index 96308f9c9..b2a47d277 100644 --- a/lib/crewai/src/crewai/experimental/agent_executor.py +++ b/lib/crewai/src/crewai/experimental/agent_executor.py @@ -103,6 +103,7 @@ from crewai.utilities.i18n import I18N_DEFAULT from crewai.utilities.planning_types import ( PlanStep, StepObservation, + StepOutcome, TodoItem, TodoList, ) @@ -417,6 +418,8 @@ class AgentExecutor(Flow[AgentExecutorState], BaseAgentExecutor): todo: TodoItem, *, success: bool, + outcome: StepOutcome, + termination_reason: str | None = None, result: str | None = None, error: str | None = None, ) -> None: @@ -429,6 +432,8 @@ class AgentExecutor(Flow[AgentExecutorState], BaseAgentExecutor): step_description=todo.description, tool_to_use=todo.tool_to_use, success=success, + outcome=outcome, + termination_reason=termination_reason, result=result, error=error, from_task=self.task, @@ -454,22 +459,44 @@ class AgentExecutor(Flow[AgentExecutorState], BaseAgentExecutor): self.state.todos.mark_completed(step_number, result=result) todo = self.state.todos.get_by_step_number(step_number) if todo and previous_status != "completed": - self._emit_plan_step_completed(todo, success=True, result=result) + self._emit_plan_step_completed( + todo, + success=True, + outcome="completed", + result=result, + ) def _mark_todo_failed( self, step_number: int, result: str | None = None, error: str | None = None, + outcome: StepOutcome | None = None, + termination_reason: str | None = None, ) -> None: todo = self.state.todos.get_by_step_number(step_number) previous_status = todo.status if todo else None - self.state.todos.mark_failed(step_number, result=result) + final_outcome = outcome or ( + todo.outcome + if todo and todo.outcome not in (None, "completed") + else "failed" + ) + final_reason = ( + termination_reason or error or (todo.termination_reason if todo else None) + ) + self.state.todos.mark_failed( + step_number, + result=result, + outcome=final_outcome, + termination_reason=final_reason, + ) todo = self.state.todos.get_by_step_number(step_number) if todo and previous_status != "failed": self._emit_plan_step_completed( todo, success=False, + outcome=final_outcome, + termination_reason=final_reason, result=result, error=error, ) @@ -578,17 +605,23 @@ class AgentExecutor(Flow[AgentExecutorState], BaseAgentExecutor): """Observe a completed step via LLM or a lightweight heuristic.""" from crewai.agents.planner_observer import PlannerObserver + if step_success is None: + step_success = self._step_success_from_log(completed_step.step_number) + if self._should_observe_steps(): observer = self._ensure_planner_observer() - return observer.observe( + observation = observer.observe( completed_step=completed_step, result=result, all_completed=all_completed, remaining_todos=remaining_todos, ) + if step_success is False and observation.step_completed_successfully: + return observation.model_copy( + update={"step_completed_successfully": False} + ) + return observation - if step_success is None: - step_success = self._step_success_from_log(completed_step.step_number) if step_success is None: step_success = True @@ -1129,12 +1162,16 @@ class AgentExecutor(Flow[AgentExecutorState], BaseAgentExecutor): # Do NOT mark completed here — observation logic decides current.result = result.result + current.outcome = result.outcome + current.termination_reason = result.termination_reason self.state.execution_log.append( { "type": "step_execution", "step_number": current.step_number, "success": result.success, + "outcome": result.outcome, + "termination_reason": result.termination_reason, "result_preview": result.result[:200] if result.result else "", "error": result.error, "tool_calls": result.tool_calls_made, @@ -1251,10 +1288,14 @@ class AgentExecutor(Flow[AgentExecutorState], BaseAgentExecutor): if isinstance(item, BaseException): error_msg = f"Error: {item!s}" todo.result = error_msg + todo.outcome = "failed" + todo.termination_reason = error_msg self._mark_todo_failed( todo.step_number, result=error_msg, error=error_msg, + outcome="failed", + termination_reason=error_msg, ) if self.agent.verbose: PRINTER.print( @@ -1265,12 +1306,16 @@ class AgentExecutor(Flow[AgentExecutorState], BaseAgentExecutor): _returned_todo, result = item step_result = cast(StepResult, result) todo.result = step_result.result + todo.outcome = step_result.outcome + todo.termination_reason = step_result.termination_reason self.state.execution_log.append( { "type": "step_execution", "step_number": todo.step_number, "success": step_result.success, + "outcome": step_result.outcome, + "termination_reason": step_result.termination_reason, "result_preview": step_result.result[:200] if step_result.result else "", diff --git a/lib/crewai/src/crewai/lite_agent_output.py b/lib/crewai/src/crewai/lite_agent_output.py index 4b4bec446..7e0b6de1a 100644 --- a/lib/crewai/src/crewai/lite_agent_output.py +++ b/lib/crewai/src/crewai/lite_agent_output.py @@ -7,7 +7,7 @@ from typing import Any from pydantic import BaseModel, Field from crewai.types.usage_metrics import UsageMetrics -from crewai.utilities.planning_types import TodoItem +from crewai.utilities.planning_types import StepOutcome, TodoItem from crewai.utilities.types import LLMMessage @@ -23,6 +23,13 @@ class TodoExecutionResult(BaseModel): result: str | None = Field( default=None, description="Result or error message from execution" ) + outcome: StepOutcome | None = Field( + default=None, description="How execution of the step terminated" + ) + termination_reason: str | None = Field( + default=None, + description="Human-readable reason for non-completed step termination", + ) depends_on: list[int] = Field( default_factory=list, description="Step numbers this depended on" ) @@ -82,6 +89,8 @@ class LiteAgentOutput(BaseModel): tool_used=item.tool_to_use, status=item.status, result=item.result, + outcome=item.outcome, + termination_reason=item.termination_reason, depends_on=item.depends_on, ) for item in todo_items diff --git a/lib/crewai/src/crewai/utilities/planning_types.py b/lib/crewai/src/crewai/utilities/planning_types.py index 1fbafb31a..367a6b636 100644 --- a/lib/crewai/src/crewai/utilities/planning_types.py +++ b/lib/crewai/src/crewai/utilities/planning_types.py @@ -9,6 +9,7 @@ from pydantic import BaseModel, Field, field_validator TodoStatus = Literal["pending", "running", "completed", "failed"] +StepOutcome = Literal["completed", "timed_out", "iteration_exhausted", "failed"] class PlanStep(BaseModel): @@ -40,6 +41,14 @@ class TodoItem(BaseModel): result: str | None = Field( default=None, description="Result after completion, if any" ) + outcome: StepOutcome | None = Field( + default=None, + description="How execution of this step terminated, once available", + ) + termination_reason: str | None = Field( + default=None, + description="Human-readable reason for non-completed step termination", + ) class TodoList(BaseModel): @@ -98,14 +107,25 @@ class TodoList(BaseModel): item = self.get_by_step_number(step_number) if item: item.status = "completed" + item.outcome = "completed" + item.termination_reason = None if result is not None: item.result = result - def mark_failed(self, step_number: int, result: str | None = None) -> None: + def mark_failed( + self, + step_number: int, + result: str | None = None, + *, + outcome: StepOutcome = "failed", + termination_reason: str | None = None, + ) -> None: """Mark a todo as failed by step number.""" item = self.get_by_step_number(step_number) if item: item.status = "failed" + item.outcome = outcome + item.termination_reason = termination_reason if result is not None: item.result = result diff --git a/lib/crewai/src/crewai/utilities/step_execution_context.py b/lib/crewai/src/crewai/utilities/step_execution_context.py index 39e3acd92..46f72c349 100644 --- a/lib/crewai/src/crewai/utilities/step_execution_context.py +++ b/lib/crewai/src/crewai/utilities/step_execution_context.py @@ -9,6 +9,8 @@ from __future__ import annotations from dataclasses import dataclass, field +from crewai.utilities.planning_types import StepOutcome + @dataclass(frozen=True) class StepExecutionContext: @@ -50,15 +52,25 @@ class StepResult: to subsequent steps or the Planner. Attributes: - success: Whether the step completed successfully. + outcome: How execution of the step terminated. result: The final output string from the step. - error: Error message if the step failed (None on success). + termination_reason: Human-readable reason for a non-completed outcome. tool_calls_made: List of tool names invoked (for debugging/logging only). execution_time: Wall-clock time in seconds for the step execution. """ - success: bool + outcome: StepOutcome result: str - error: str | None = None + termination_reason: str | None = None tool_calls_made: list[str] = field(default_factory=list) execution_time: float = 0.0 + + @property + def success(self) -> bool: + """Whether the step reached the completed outcome.""" + return self.outcome == "completed" + + @property + def error(self) -> str | None: + """Backward-compatible alias for the termination reason.""" + return self.termination_reason diff --git a/lib/crewai/tests/agents/test_agent_executor.py b/lib/crewai/tests/agents/test_agent_executor.py index 80bd5fe85..ac2969aa5 100644 --- a/lib/crewai/tests/agents/test_agent_executor.py +++ b/lib/crewai/tests/agents/test_agent_executor.py @@ -812,9 +812,106 @@ class TestStepExecutorCriticalFixes: result = step_executor.execute(todo, context) assert result.success is False + assert result.outcome == "failed" assert result.error is not None assert "Expected tool 'count_words' was not called" in result.error + def test_step_executor_marks_text_iteration_exhaustion_as_failure( + self, step_executor + ): + """A tool result at the iteration limit is partial, not successful.""" + todo = TodoItem( + step_number=1, + description="Count words in input text.", + tool_to_use="count_words", + ) + context = StepExecutionContext(task_description="task", task_goal="goal") + action = AgentAction( + thought="Need a tool", + tool="count_words", + tool_input='{"text":"hello world"}', + text="Action: count_words", + ) + + with ( + patch( + "crewai.agents.step_executor.process_llm_response", + return_value=action, + ), + patch.object( + step_executor, + "_execute_text_tool_with_events", + return_value="2", + ), + ): + step_executor.llm.call.return_value = action.text + result = step_executor.execute( + todo, + context, + max_step_iterations=1, + ) + + assert result.success is False + assert result.outcome == "iteration_exhausted" + assert result.result == "2" + assert result.termination_reason is not None + assert "max_step_iterations (1)" in result.termination_reason + + def test_step_executor_marks_native_iteration_exhaustion_as_failure( + self, step_executor + ): + """Native tool output at the iteration limit remains a partial result.""" + step_executor._use_native_tools = True + step_executor._openai_tools = [ + {"type": "function", "function": {"name": "count_words"}} + ] + todo = TodoItem(step_number=1, description="Count words") + context = StepExecutionContext(task_description="task", task_goal="goal") + tool_calls = [object()] + step_executor.llm.call.return_value = tool_calls + + with ( + patch("crewai.agents.step_executor.is_tool_call_list", return_value=True), + patch.object( + step_executor, + "_execute_native_tool_calls", + return_value="2", + ), + ): + result = step_executor.execute( + todo, + context, + max_step_iterations=1, + ) + + assert result.success is False + assert result.outcome == "iteration_exhausted" + assert result.result == "2" + assert result.termination_reason is not None + assert "max_step_iterations (1)" in result.termination_reason + + def test_step_executor_marks_timeout_as_failure(self, step_executor): + """Reaching the wall-clock limit returns an explicit timeout outcome.""" + step_executor._use_native_tools = False + todo = TodoItem(step_number=1, description="Count words") + context = StepExecutionContext(task_description="task", task_goal="goal") + + with patch( + "crewai.agents.step_executor.time.monotonic", + side_effect=[0.5, 2.0, 2.0], + ): + result = step_executor.execute( + todo, + context, + step_timeout=1, + ) + + assert result.success is False + assert result.outcome == "timed_out" + assert result.result == "" + assert result.termination_reason == "Step timed out after 2s (limit: 1s)." + step_executor.llm.call.assert_not_called() + def test_step_executor_text_tool_emits_usage_events(self, step_executor): """Text-parsed tool execution should emit started and finished events.""" started_events: list[ToolUsageStartedEvent] = [] @@ -929,6 +1026,8 @@ class TestStepExecutorCriticalFixes: assert started_events[0].tool_to_use == "search" assert len(completed_events) == 1 assert completed_events[0].success is True + assert completed_events[0].outcome == "completed" + assert completed_events[0].termination_reason is None assert completed_events[0].step_number == 1 assert completed_events[0].result == "Found release" @@ -953,16 +1052,44 @@ class TestStepExecutorCriticalFixes: 1, result="Error: no result", error="No result", + outcome="timed_out", + termination_reason="Step timed out after 30s.", ) crewai_event_bus.flush() assert todo.status == "failed" assert len(completed_events) == 1 assert completed_events[0].success is False + assert completed_events[0].outcome == "timed_out" + assert completed_events[0].termination_reason == "Step timed out after 30s." assert completed_events[0].step_number == 1 assert completed_events[0].result == "Error: no result" assert completed_events[0].error == "No result" + def test_lite_agent_output_includes_step_termination_details(self): + """Standalone agent output exposes how each planned step terminated.""" + from crewai.lite_agent_output import LiteAgentOutput + + todos = LiteAgentOutput.from_todo_items( + [ + TodoItem( + step_number=1, + description="Run bounded work", + status="failed", + result="partial result", + outcome="iteration_exhausted", + termination_reason=( + "Step reached max_step_iterations (1) before producing " + "a final answer." + ), + ) + ] + ) + + assert todos[0].outcome == "iteration_exhausted" + assert todos[0].termination_reason is not None + assert "max_step_iterations (1)" in todos[0].termination_reason + @patch("crewai.experimental.agent_executor.handle_output_parser_exception") def test_recover_from_parser_error( self, mock_handle_exception, mock_dependencies @@ -1849,6 +1976,40 @@ class TestReasoningEffort: executor._ensure_planner_observer.assert_not_called() assert observation.step_completed_successfully is True + def test_observer_cannot_override_failed_step_execution(self): + """An optimistic observer cannot turn an execution failure into success.""" + from crewai.agent.planning_config import PlanningConfig + from crewai.experimental.agent_executor import AgentExecutor + from crewai.utilities.planning_types import StepObservation, TodoItem + + executor = Mock(spec=AgentExecutor) + executor.agent = Mock() + executor.agent.planning_config = PlanningConfig(reasoning_effort="medium") + executor._should_observe_steps = ( + AgentExecutor._should_observe_steps.__get__(executor) + ) + executor._observe_completed_step = ( + AgentExecutor._observe_completed_step.__get__(executor) + ) + executor._ensure_planner_observer.return_value.observe.return_value = ( + StepObservation( + step_completed_successfully=True, + key_information_learned="Partial data is usable", + remaining_plan_still_valid=True, + ) + ) + todo = TodoItem(step_number=1, description="Run bounded work") + + observation = executor._observe_completed_step( + completed_step=todo, + result="partial result", + all_completed=[], + remaining_todos=[], + step_success=False, + ) + + assert observation.step_completed_successfully is False + @pytest.mark.vcr() def test_reasoning_effort_low_skips_decide_and_replan(self): """Low effort: heuristic observe, no decide/replan/refine LLM pipeline.