diff --git a/codeframe/cli/app.py b/codeframe/cli/app.py index be2abad6..f5cec6f4 100644 --- a/codeframe/cli/app.py +++ b/codeframe/cli/app.py @@ -2003,6 +2003,11 @@ def work_start( "--stall-timeout", help="Seconds without a tool call before agent is considered stalled (0 = disabled)", ), + stall_action: str = typer.Option( + "blocker", + "--stall-action", + help="Recovery action on stall: 'blocker' (default), 'retry', or 'fail'", + ), ) -> None: """Start working on a task. @@ -2068,6 +2073,7 @@ def work_start( state = runtime.execute_agent( workspace, run, dry_run=dry_run, debug=debug, verbose=verbose, engine=engine, stall_timeout_s=stall_timeout, + stall_action=stall_action, ) if state.status == AgentStatus.COMPLETED: @@ -2874,6 +2880,11 @@ def batch_run( "--stall-timeout", help="Seconds without a tool call before agent is considered stalled (0 = disabled)", ), + stall_action: str = typer.Option( + "blocker", + "--stall-action", + help="Recovery action on stall: 'blocker' (default), 'retry', or 'fail'", + ), ) -> None: """Execute multiple tasks in batch. @@ -2972,6 +2983,7 @@ def batch_run( max_retries=max_retries, engine=engine, stall_timeout_s=stall_timeout, + stall_action=stall_action, ) # Show summary diff --git a/codeframe/core/__init__.py b/codeframe/core/__init__.py index b4bf77db..80a60a6b 100644 --- a/codeframe/core/__init__.py +++ b/codeframe/core/__init__.py @@ -3,7 +3,7 @@ from codeframe.core.project import Project from codeframe.core.config import Config from codeframe.core.models import Task, TaskStatus, AgentMaturity -from codeframe.core.stall_detector import StallAction, StallDetector +from codeframe.core.stall_detector import StallAction, StallDetectedError, StallDetector __all__ = [ "Project", @@ -12,5 +12,6 @@ "TaskStatus", "AgentMaturity", "StallAction", + "StallDetectedError", "StallDetector", ] diff --git a/codeframe/core/conductor.py b/codeframe/core/conductor.py index 1894ef63..1c8dfccd 100644 --- a/codeframe/core/conductor.py +++ b/codeframe/core/conductor.py @@ -458,6 +458,7 @@ class BatchRun: results: dict[str, str] = field(default_factory=dict) engine: str = "react" stall_timeout_s: int = 300 + stall_action: str = "blocker" def start_batch( @@ -471,6 +472,7 @@ def start_batch( on_event: Optional[Callable[[str, dict], None]] = None, engine: str = "react", stall_timeout_s: int = 300, + stall_action: str = "blocker", ) -> BatchRun: """Start a batch execution of multiple tasks. @@ -518,6 +520,7 @@ def start_batch( results={}, engine=engine, stall_timeout_s=stall_timeout_s, + stall_action=stall_action, ) # Save to database @@ -924,7 +927,7 @@ def _execute_serial_resume( on_event("batch_task_started", {"task_id": task_id, "position": i + 1, "is_retry": True}) # Execute task via subprocess - result_status = _execute_task_subprocess(workspace, task_id, batch.id, engine=batch.engine, stall_timeout_s=batch.stall_timeout_s) + result_status = _execute_task_subprocess(workspace, task_id, batch.id, engine=batch.engine, stall_timeout_s=batch.stall_timeout_s, stall_action=batch.stall_action) # Record result (overwrites previous result) batch.results[task_id] = result_status @@ -1092,7 +1095,7 @@ def _execute_retries( print(f" Previous: {previous_status}") # Execute task - result_status = _execute_task_subprocess(workspace, task_id, batch.id, engine=batch.engine, stall_timeout_s=batch.stall_timeout_s) + result_status = _execute_task_subprocess(workspace, task_id, batch.id, engine=batch.engine, stall_timeout_s=batch.stall_timeout_s, stall_action=batch.stall_action) # Update result batch.results[task_id] = result_status @@ -1259,7 +1262,7 @@ def _execute_serial( on_event("batch_task_started", {"task_id": task_id, "position": i + 1}) # Execute task via subprocess - result_status = _execute_task_subprocess(workspace, task_id, batch.id, engine=batch.engine, stall_timeout_s=batch.stall_timeout_s) + result_status = _execute_task_subprocess(workspace, task_id, batch.id, engine=batch.engine, stall_timeout_s=batch.stall_timeout_s, stall_action=batch.stall_action) # If task is BLOCKED, try supervisor resolution if result_status == RunStatus.BLOCKED.value: @@ -1267,7 +1270,7 @@ def _execute_serial( if supervisor.try_resolve_blocked_task(task_id): # Supervisor resolved the blocker - retry the task print(" [Supervisor] Retrying task after auto-resolution...") - result_status = _execute_task_subprocess(workspace, task_id, batch.id, engine=batch.engine, stall_timeout_s=batch.stall_timeout_s) + result_status = _execute_task_subprocess(workspace, task_id, batch.id, engine=batch.engine, stall_timeout_s=batch.stall_timeout_s, stall_action=batch.stall_action) # Record result batch.results[task_id] = result_status @@ -1547,7 +1550,7 @@ def _execute_single_task( on_event("batch_task_started", {"task_id": task_id, "position": position}) # Execute task via subprocess - result_status = _execute_task_subprocess(workspace, task_id, batch.id, engine=batch.engine, stall_timeout_s=batch.stall_timeout_s) + result_status = _execute_task_subprocess(workspace, task_id, batch.id, engine=batch.engine, stall_timeout_s=batch.stall_timeout_s, stall_action=batch.stall_action) # If task is BLOCKED, try supervisor resolution if result_status == RunStatus.BLOCKED.value: @@ -1555,7 +1558,7 @@ def _execute_single_task( if supervisor.try_resolve_blocked_task(task_id): # Supervisor resolved the blocker - retry the task print(" [Supervisor] Retrying task after auto-resolution...") - result_status = _execute_task_subprocess(workspace, task_id, batch.id, engine=batch.engine, stall_timeout_s=batch.stall_timeout_s) + result_status = _execute_task_subprocess(workspace, task_id, batch.id, engine=batch.engine, stall_timeout_s=batch.stall_timeout_s, stall_action=batch.stall_action) # Record result batch.results[task_id] = result_status @@ -1646,7 +1649,7 @@ def execute_task(task_id: str) -> tuple[str, str]: on_event("batch_task_started", {"task_id": task_id, "parallel": True}) # Execute via subprocess - result_status = _execute_task_subprocess(workspace, task_id, batch.id, engine=batch.engine, stall_timeout_s=batch.stall_timeout_s) + result_status = _execute_task_subprocess(workspace, task_id, batch.id, engine=batch.engine, stall_timeout_s=batch.stall_timeout_s, stall_action=batch.stall_action) # Record result (thread-safe due to GIL for simple dict operations) batch.results[task_id] = result_status @@ -1704,6 +1707,7 @@ def _execute_task_subprocess( batch_id: Optional[str] = None, engine: str = "react", stall_timeout_s: int = 300, + stall_action: str = "blocker", ) -> str: """Execute a single task via subprocess. @@ -1715,6 +1719,7 @@ def _execute_task_subprocess( batch_id: Optional batch ID for process tracking (enables force stop) engine: Agent engine to use ("plan" or "react") stall_timeout_s: Stall detection timeout in seconds (0 = disabled) + stall_action: Recovery action on stall ("blocker", "retry", or "fail") Returns: RunStatus value string (COMPLETED, FAILED, BLOCKED) @@ -1725,6 +1730,7 @@ def _execute_task_subprocess( "work", "start", task_id, "--execute", "--engine", engine, "--stall-timeout", str(stall_timeout_s), + "--stall-action", stall_action, ] process = None diff --git a/codeframe/core/react_agent.py b/codeframe/core/react_agent.py index 7994973a..73d66c43 100644 --- a/codeframe/core/react_agent.py +++ b/codeframe/core/react_agent.py @@ -23,6 +23,7 @@ from codeframe.core.context import ContextLoader, TaskContext from codeframe.core.events import EventType from codeframe.core.fix_tracker import EscalationDecision, FixAttemptTracker, FixOutcome +from codeframe.core.stall_detector import StallAction, StallDetectedError from codeframe.core.stall_monitor import StallEvent, StallMonitor from codeframe.core.models import AgentPhase, CompletionEvent, ErrorEvent, ProgressEvent from codeframe.core.quick_fixes import apply_quick_fix, find_quick_fix @@ -110,6 +111,7 @@ def __init__( max_iterations: int = 30, max_verification_retries: int = 5, stall_timeout_s: float = 300, + stall_action: StallAction = StallAction.BLOCKER, event_publisher: Optional[EventPublisher] = None, dry_run: bool = False, verbose: bool = False, @@ -123,6 +125,7 @@ def __init__( self.max_iterations = max_iterations self.max_verification_retries = max_verification_retries self._stall_timeout_s = stall_timeout_s + self._stall_action = stall_action self.event_publisher = event_publisher self.dry_run = dry_run self.verbose = verbose @@ -205,11 +208,12 @@ def run(self, task_id: str) -> AgentStatus: try: status = self._react_loop(system_prompt) if status == AgentStatus.FAILED: + reason = "stall_detected" if self._stall_triggered.is_set() else "max_iterations_reached" self._emit(EventType.AGENT_FAILED, { "task_id": task_id, - "reason": "max_iterations_reached", + "reason": reason, }) - self._emit_stream_error(task_id, "max_iterations_reached") + self._emit_stream_error(task_id, reason) return status if status == AgentStatus.BLOCKED: @@ -244,6 +248,9 @@ def run(self, task_id: str) -> AgentStatus: return AgentStatus.FAILED finally: self._stall_monitor.stop() + except StallDetectedError: + self._stall_monitor.stop() + raise # Let runtime handle retry except Exception: logger.exception("ReactAgent.run() failed for task %s", task_id) self._emit(EventType.AGENT_FAILED, { @@ -283,16 +290,30 @@ def _react_loop(self, system_prompt: str) -> AgentStatus: # Check for stall before each iteration if self._stall_triggered.is_set(): stall_ctx = "" + elapsed_s = 0.0 if self._stall_event: + elapsed_s = self._stall_event.elapsed_s stall_ctx = ( - f"Agent stalled: no tool call for {self._stall_event.elapsed_s:.0f}s " + f"Agent stalled: no tool call for {elapsed_s:.0f}s " f"(timeout: {self._stall_event.stall_timeout_s}s)" ) - self._create_text_blocker( - stall_ctx or "Agent stalled with no tool activity", - "stall_detected", - ) - return AgentStatus.BLOCKED + + if self._stall_action == StallAction.RETRY: + raise StallDetectedError( + elapsed_s=elapsed_s, + iterations=iterations, + last_tool=recent_tool_signatures[-1][0] if recent_tool_signatures else "", + ) + elif self._stall_action == StallAction.FAIL: + self._verbose_print(f"[ReactAgent] Stall → FAILED: {stall_ctx}") + return AgentStatus.FAILED + else: + # StallAction.BLOCKER (default) + self._create_text_blocker( + stall_ctx or "Agent stalled with no tool activity", + "stall_detected", + ) + return AgentStatus.BLOCKED self._verbose_print(f"[ReactAgent] Iteration {iterations + 1}/{self.max_iterations}") self._emit(EventType.AGENT_ITERATION_STARTED, { @@ -475,6 +496,13 @@ def _run_final_verification( for attempt in range(1 + self.max_verification_retries): if self._stall_triggered.is_set(): + if self._stall_action == StallAction.RETRY: + raise StallDetectedError( + elapsed_s=self._stall_event.elapsed_s if self._stall_event else 0, + iterations=0, + ) + elif self._stall_action == StallAction.FAIL: + return (False, "stall_failed") return (False, "stall_detected") self._verbose_print("[ReactAgent] Running final verification...") diff --git a/codeframe/core/runtime.py b/codeframe/core/runtime.py index 917eda73..fe377a9b 100644 --- a/codeframe/core/runtime.py +++ b/codeframe/core/runtime.py @@ -598,6 +598,7 @@ def execute_agent( event_publisher: Optional["EventPublisher"] = None, engine: str = "react", stall_timeout_s: int = 300, + stall_action: str = "blocker", ) -> "AgentState": """Execute a task using the agent orchestrator. @@ -673,22 +674,52 @@ def on_agent_event(event_type: str, data: dict) -> None: # ReactAgent has a simpler interface — it handles its own # retries and verification internally. from codeframe.core.react_agent import ReactAgent + from codeframe.core.stall_detector import StallAction, StallDetectedError - react_agent = ReactAgent( - workspace=workspace, - llm_provider=provider, - stall_timeout_s=stall_timeout_s, - event_publisher=event_publisher, - dry_run=dry_run, - verbose=verbose, - on_event=on_agent_event, - debug=debug, - output_logger=output_logger, - fix_coordinator=fix_coordinator, - ) - react_status = react_agent.run(run.task_id) - # Wrap AgentStatus enum into AgentState dataclass for compatibility - state = AgentState(status=react_status) + resolved_action = StallAction(stall_action) + + def _build_react_agent() -> ReactAgent: + return ReactAgent( + workspace=workspace, + llm_provider=provider, + stall_timeout_s=stall_timeout_s, + stall_action=resolved_action, + event_publisher=event_publisher, + dry_run=dry_run, + verbose=verbose, + on_event=on_agent_event, + debug=debug, + output_logger=output_logger, + fix_coordinator=fix_coordinator, + ) + + max_stall_retries = 1 + for stall_attempt in range(1 + max_stall_retries): + try: + react_agent = _build_react_agent() + react_status = react_agent.run(run.task_id) + # Wrap AgentStatus enum into AgentState dataclass for compatibility + state = AgentState(status=react_status) + break + except StallDetectedError as exc: + run_logger.warning( + LogCategory.AGENT_ACTION, + f"Stall detected (attempt {stall_attempt + 1}): {exc}", + {"elapsed_s": exc.elapsed_s, "iterations": exc.iterations}, + ) + if stall_attempt >= max_stall_retries: + run_logger.error( + LogCategory.AGENT_ACTION, + "Max stall retries exceeded, failing task", + {}, + ) + state = AgentState(status=AgentStatus.FAILED) + break + run_logger.info( + LogCategory.AGENT_ACTION, + "Retrying after stall", + {"attempt": stall_attempt + 2}, + ) else: agent = Agent( workspace=workspace, diff --git a/codeframe/core/stall_detector.py b/codeframe/core/stall_detector.py index f5375da3..a93e2b54 100644 --- a/codeframe/core/stall_detector.py +++ b/codeframe/core/stall_detector.py @@ -24,6 +24,23 @@ class StallAction(str, Enum): FAIL = "fail" +class StallDetectedError(Exception): + """Raised when a stall is detected with RETRY action. + + Propagates up to ``execute_agent()`` so the runtime can re-invoke + the agent with context about why the previous attempt stalled. + """ + + def __init__(self, elapsed_s: float, iterations: int, last_tool: str = "") -> None: + self.elapsed_s = elapsed_s + self.iterations = iterations + self.last_tool = last_tool + super().__init__( + f"Agent stalled after {elapsed_s:.0f}s " + f"(iterations={iterations}, last_tool={last_tool!r})" + ) + + class StallDetector: """Tracks elapsed time since the last recorded activity. diff --git a/tests/core/test_stall_detector.py b/tests/core/test_stall_detector.py index 7f4e10cb..e63654e0 100644 --- a/tests/core/test_stall_detector.py +++ b/tests/core/test_stall_detector.py @@ -1,4 +1,4 @@ -"""Tests for StallDetector and StallAction. +"""Tests for StallDetector, StallAction, and StallDetectedError. Tests cover: - StallDetector time tracking and threshold detection @@ -6,13 +6,18 @@ - Disabled detection (timeout <= 0) - elapsed_since_activity_ms() accuracy - StallAction enum values and str inheritance +- StallDetectedError exception attributes +- ReactAgent stall_action parameter integration +- execute_agent stall_action parameter """ +import inspect import time +from unittest.mock import MagicMock import pytest -from codeframe.core.stall_detector import StallAction, StallDetector +from codeframe.core.stall_detector import StallAction, StallDetectedError, StallDetector pytestmark = pytest.mark.v2 @@ -94,3 +99,90 @@ def test_very_short_timeout_stalls_quickly(self, short_detector): # Deterministic: backdate activity to guarantee stall short_detector._last_activity = time.monotonic() - 1.0 assert short_detector.is_stalled() is True + + +class TestStallDetectedError: + """Test StallDetectedError exception.""" + + def test_attributes(self): + err = StallDetectedError(elapsed_s=305.2, iterations=7, last_tool="run_tests") + assert err.elapsed_s == 305.2 + assert err.iterations == 7 + assert err.last_tool == "run_tests" + + def test_is_exception(self): + err = StallDetectedError(elapsed_s=100, iterations=3) + assert isinstance(err, Exception) + + def test_message_format(self): + err = StallDetectedError(elapsed_s=300, iterations=5, last_tool="edit_file") + assert "300" in str(err) + assert "edit_file" in str(err) + + def test_default_last_tool(self): + err = StallDetectedError(elapsed_s=100, iterations=2) + assert err.last_tool == "" + + +class TestReactAgentStallAction: + """Test ReactAgent stall_action parameter.""" + + def test_accepts_stall_action_param(self): + from codeframe.core.react_agent import ReactAgent + mock_provider = MagicMock() + mock_workspace = MagicMock() + agent = ReactAgent( + workspace=mock_workspace, + llm_provider=mock_provider, + stall_action=StallAction.RETRY, + ) + assert agent._stall_action == StallAction.RETRY + + def test_default_stall_action_is_blocker(self): + from codeframe.core.react_agent import ReactAgent + mock_provider = MagicMock() + mock_workspace = MagicMock() + agent = ReactAgent( + workspace=mock_workspace, + llm_provider=mock_provider, + ) + assert agent._stall_action == StallAction.BLOCKER + + def test_stall_action_fail(self): + from codeframe.core.react_agent import ReactAgent + mock_provider = MagicMock() + mock_workspace = MagicMock() + agent = ReactAgent( + workspace=mock_workspace, + llm_provider=mock_provider, + stall_action=StallAction.FAIL, + ) + assert agent._stall_action == StallAction.FAIL + + +class TestExecuteAgentStallAction: + """Test that execute_agent accepts stall_action parameter.""" + + def test_has_stall_action_param(self): + from codeframe.core.runtime import execute_agent + sig = inspect.signature(execute_agent) + assert "stall_action" in sig.parameters + assert sig.parameters["stall_action"].default == "blocker" + + +class TestConductorStallAction: + """Test stall_action threading through conductor.""" + + def test_batch_run_has_stall_action(self): + from codeframe.core.conductor import BatchRun + # Verify the field exists with correct default + import dataclasses + fields = {f.name: f for f in dataclasses.fields(BatchRun)} + assert "stall_action" in fields + assert fields["stall_action"].default == "blocker" + + def test_start_batch_accepts_stall_action(self): + from codeframe.core.conductor import start_batch + sig = inspect.signature(start_batch) + assert "stall_action" in sig.parameters + assert sig.parameters["stall_action"].default == "blocker"