feat(core): wire StallAction dispatch into ReactAgent and runtime (#401) (#425)

* feat(core): wire StallAction dispatch into ReactAgent and runtime (#401)

- Add StallDetectedError exception for RETRY stall recovery path
- Add stall_action parameter to ReactAgent (default: BLOCKER)
- Wire configurable StallAction dispatch in react loop stall check:
  RETRY raises StallDetectedError, FAIL returns FAILED, BLOCKER creates blocker
- Catch StallDetectedError in execute_agent() with 1 retry attempt
- Add --stall-action CLI flag to work start and work batch run
- Thread stall_action through conductor and subprocess calls
- 24 tests covering all new integration points

* fix: remove unused threading import from test file

* fix: honor stall_action during verification and emit correct failure reason

- Verification stall check now dispatches via StallAction (RETRY/FAIL/BLOCKER)
- FAILED status from react loop correctly reports stall_detected vs max_iterations

---------

Co-authored-by: Test User <test@example.com>

Author: frankbria

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

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",
 ]

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,15 +1262,15 @@ 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:
             supervisor = get_supervisor(workspace)
             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,15 +1550,15 @@ 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:
         supervisor = get_supervisor(workspace)
         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

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...")

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,

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.