Coverage for little_loops / fsm / executor.py: 13%
628 statements
« prev ^ index » next coverage.py v7.12.0, created at 2026-05-28 13:07 -0500
1"""FSM Executor - Runtime engine for FSM loop execution.
3This module provides the execution engine that runs FSM loops:
4- Executes actions (shell commands or slash commands)
5- Evaluates results using appropriate evaluators
6- Routes to next states based on verdicts
7- Tracks iteration count and enforces limits
8- Manages captured variables and context
9"""
11from __future__ import annotations
13import json
14import random
15import subprocess
16import threading
17import time
18from collections.abc import Callable
19from dataclasses import dataclass
20from datetime import UTC, datetime
21from pathlib import Path
22from typing import Any
24from little_loops.fsm.evaluators import (
25 EvaluationResult,
26 evaluate,
27 evaluate_exit_code,
28 evaluate_llm_structured,
29 evaluate_mcp_result,
30)
31from little_loops.fsm.handoff_handler import HandoffHandler
32from little_loops.fsm.interpolation import (
33 InterpolationContext,
34 InterpolationError,
35 interpolate,
36 interpolate_dict,
37)
38from little_loops.fsm.rate_limit_circuit import RateLimitCircuit
39from little_loops.fsm.runners import (
40 ActionRunner,
41 DefaultActionRunner,
42 SimulationActionRunner, # noqa: F401 — re-exported for backward compatibility
43 _now_ms,
44)
45from little_loops.fsm.schema import FSMLoop, StateConfig
46from little_loops.fsm.signal_detector import DetectedSignal, SignalDetector
47from little_loops.fsm.stall_detector import Stall, StallDetector
48from little_loops.fsm.types import ActionResult, Evaluator, EventCallback, ExecutionResult
49from little_loops.issue_lifecycle import FailureType, classify_failure
50from little_loops.session_log import get_current_session_jsonl
52# Maximum number of per-state rate-limit retries before emitting rate_limit_exhausted.
53_DEFAULT_RATE_LIMIT_RETRIES: int = 3
54# Base backoff in seconds; actual sleep = base * 2^(attempt-1) + uniform(0, base).
55_DEFAULT_RATE_LIMIT_BACKOFF_BASE: int = 30
56# Total wall-clock budget (seconds) across short + long tiers before routing to
57# on_rate_limit_exhausted. Mirrors RateLimitsConfig.max_wait_seconds default (6h).
58_DEFAULT_RATE_LIMIT_MAX_WAIT_SECONDS: int = 21600
59# Long-wait tier ladder (seconds), walked once the short-tier budget is spent.
60# Mirrors RateLimitsConfig.long_wait_ladder default.
61_DEFAULT_RATE_LIMIT_LONG_WAIT_LADDER: list[int] = [300, 900, 1800, 3600]
62# Event name emitted when the stall detector fires on N consecutive
63# identical (state, exit_code, verdict) transitions. See FEAT-1637.
64STALL_DETECTED_EVENT: str = "stall_detected"
65# Event name emitted when rate-limit retries are exhausted.
66RATE_LIMIT_EXHAUSTED_EVENT: str = "rate_limit_exhausted"
67# Event name emitted when consecutive rate-limit exhaustions reach the storm threshold.
68RATE_LIMIT_STORM_EVENT: str = "rate_limit_storm"
69# Event name emitted every ~60s during a long-wait rate-limit sleep so UIs can show live progress.
70RATE_LIMIT_WAITING_EVENT: str = "rate_limit_waiting"
71# Interval (seconds) between rate_limit_waiting heartbeat emissions during long-wait sleeps.
72_RATE_LIMIT_HEARTBEAT_INTERVAL: float = 60.0
73# Number of consecutive rate_limit_exhausted events that constitute a storm.
74_RATE_LIMIT_STORM_THRESHOLD: int = 3
75# Progressive throttle defaults: calls 1..normal_max pass through, at warn_max emit warning,
76# at hard_max route to on_throttle_hard, beyond hard_max hard-stop.
77_DEFAULT_THROTTLE_NORMAL_MAX: int = 3
78_DEFAULT_THROTTLE_WARN_MAX: int = 8
79_DEFAULT_THROTTLE_HARD_MAX: int = 12
80# Event names for progressive tool-call throttling within a single state visit.
81THROTTLE_WARN_EVENT: str = "throttle_warn"
82THROTTLE_HARD_EVENT: str = "throttle_hard"
83THROTTLE_STOP_EVENT: str = "throttle_stop"
84# Action types that consume LLM quota and are gated by the shared circuit breaker.
85# `_action_mode()` collapses both to "prompt"; the frozenset documents intent.
86LLM_ACTION_TYPES: frozenset[str] = frozenset({"slash_command", "prompt"})
87# Maximum per-state API server error retries before falling through to normal routing.
88_DEFAULT_API_ERROR_RETRIES: int = 2
89# Flat backoff in seconds between API server error retries (no exponential ladder).
90_DEFAULT_API_ERROR_BACKOFF: int = 30
93def _iso_now() -> str:
94 """Get current time as ISO 8601 string."""
95 return datetime.now(UTC).isoformat()
98@dataclass
99class RouteContext:
100 """Context passed to before_route / after_route interceptors."""
102 state_name: str
103 state: StateConfig
104 verdict: str
105 action_result: ActionResult | None
106 eval_result: EvaluationResult | None
107 ctx: InterpolationContext
108 iteration: int
111@dataclass
112class RouteDecision:
113 """Returned by before_route to redirect or veto a routing transition.
115 Return semantics for before_route:
116 None (implicit) → passthrough, routing proceeds normally
117 RouteDecision("state") → redirect, bypass _route() and use "state" directly
118 RouteDecision(None) → veto, _execute_state() returns None → _finish("error")
119 """
121 next_state: str | None # str → redirect; None → veto
124class FSMExecutor:
125 """Execute an FSM loop.
127 The executor runs an FSM from its initial state until:
128 - A terminal state is reached
129 - max_iterations is exceeded
130 - A timeout occurs
131 - A shutdown signal is received
132 - An unrecoverable error occurs
134 Events are emitted via the callback for observability.
135 """
137 def __init__(
138 self,
139 fsm: FSMLoop,
140 event_callback: EventCallback | None = None,
141 action_runner: ActionRunner | None = None,
142 signal_detector: SignalDetector | None = None,
143 handoff_handler: HandoffHandler | None = None,
144 loops_dir: Path | None = None,
145 circuit: RateLimitCircuit | None = None,
146 ):
147 """Initialize the executor.
149 Args:
150 fsm: The FSM loop to execute
151 event_callback: Optional callback for events
152 action_runner: Optional custom action runner (for testing)
153 signal_detector: Optional signal detector for output parsing
154 handoff_handler: Optional handler for handoff signals
155 loops_dir: Base directory for resolving sub-loop references
156 circuit: Optional shared rate-limit circuit breaker for 429 coordination
157 """
158 self.fsm = fsm
159 self.event_callback = event_callback or (lambda _: None)
160 self.action_runner: ActionRunner = action_runner or DefaultActionRunner()
161 self.signal_detector = signal_detector
162 self.handoff_handler = handoff_handler
163 self.loops_dir = loops_dir
164 self._circuit = circuit
166 # Runtime state
167 self.current_state = fsm.initial
168 self.iteration = 0
169 self.captured: dict[str, dict[str, Any]] = {}
170 self.prev_result: dict[str, Any] | None = None
171 self.started_at = ""
172 self.start_time_ms = 0
173 self.elapsed_offset_ms = (
174 0 # milliseconds from segments before current run (set by PersistentExecutor on resume)
175 )
177 # Shutdown flag for graceful signal handling
178 self._shutdown_requested = False
180 # Currently running MCP subprocess (set by _run_subprocess, cleared in finally).
181 # Enables external shutdown code to kill the process on SIGTERM.
182 self._current_process: subprocess.Popen[str] | None = None
184 # Pending handoff signal (set by _run_action, checked by main loop)
185 self._pending_handoff: DetectedSignal | None = None
187 # Pending error payload from FATAL_ERROR signal (set by _run_action, checked by main loop)
188 self._pending_error: str | None = None
190 # Per-state retry tracking for max_retries support.
191 # _retry_counts[state_name] = number of consecutive re-entries into that state.
192 # Incremented each time we enter the same state as the previous iteration.
193 # Reset when a different state is entered, or after retry exhaustion.
194 self._retry_counts: dict[str, int] = {}
195 # State entered in the previous iteration (None on first iteration or after resume).
196 self._prev_state: str | None = None
198 # BUG-1226: true between emitting a `route` event and entering the
199 # target state. Gates the "flush one pending shell state on timeout"
200 # behavior so we only flush when there is actually a pending state.
201 self._just_routed: bool = False
203 # ENH-1631: true once the on_max_iterations summary state has been
204 # dispatched. Prevents the cap guard from re-triggering before the
205 # summary state completes.
206 self._summary_state_executed: bool = False
208 # Per-state rate-limit retry tracking (parallel to _retry_counts).
209 # _rate_limit_retries[state_name] = dict-of-record:
210 # {
211 # "short_retries": int, # attempts in short-burst tier
212 # "long_retries": int, # attempts in long-wait tier
213 # "total_wait_seconds": float,
214 # "first_seen_at": float | None, # epoch timestamp of first 429
215 # }
216 # Incremented inside _handle_rate_limit on each detected rate-limit response.
217 # Reset when the state completes without a rate-limit, or after exhaustion.
218 self._rate_limit_retries: dict[str, dict[str, Any]] = {}
220 # Consecutive rate_limit_exhausted emissions across all states. Reset on any
221 # successful non-rate-limited state transition. When this reaches
222 # _RATE_LIMIT_STORM_THRESHOLD, a RATE_LIMIT_STORM event is emitted.
223 self._consecutive_rate_limit_exhaustions: int = 0
225 # Per-state API server error retry tracking (parallel to _rate_limit_retries).
226 # _api_error_retries[state_name] = {"retries": int, "total_wait": float}
227 # Reset when the state completes without a server error, or after exhaustion.
228 self._api_error_retries: dict[str, dict[str, Any]] = {}
230 # Per-state tool-call throttle counter. Counts successive action executions within
231 # a single continuous state visit. Reset on state exit; NOT serialized to LoopState
232 # (throttle counts measure instantaneous visit-level activity, not cumulative retries).
233 self._throttle_counts: dict[str, int] = {}
235 # Per-edge revisit counter for cycle detection.
236 # _edge_revisit_counts["from_state->to_state"] = number of times that edge has fired.
237 # When any edge exceeds max_edge_revisits, the loop terminates with cycle_detected.
238 self._edge_revisit_counts: dict[str, int] = {}
240 # Stall detector for repeated (state, exit_code, verdict) triples.
241 # Enabled via fsm.circuit.repeated_failure (FEAT-1637); None when not configured.
242 self._stall_detector: StallDetector | None = None
243 if fsm.circuit is not None and fsm.circuit.repeated_failure is not None:
244 self._stall_detector = StallDetector(window=fsm.circuit.repeated_failure.window)
245 # Set by _execute_state when the detector fires with on_repeated_failure="abort";
246 # checked by run() to terminate via _finish("stall_detected", ...).
247 self._pending_stall_abort: Stall | None = None
249 # Nesting depth for sub-loop event forwarding (0 = top-level, 1+ = sub-loop).
250 # Set by the parent executor when constructing child executors.
251 self._depth: int = 0
253 # Extension hook registries — populated by wire_extensions()
254 self._contributed_actions: dict[str, ActionRunner] = {}
255 self._contributed_evaluators: dict[str, Evaluator] = {}
256 self._interceptors: list[Any] = []
258 def request_shutdown(self) -> None:
259 """Request graceful shutdown of the executor.
261 Sets a flag that will be checked at the start of each iteration,
262 allowing the loop to exit cleanly after the current state completes.
263 """
264 self._shutdown_requested = True
266 def run(self) -> ExecutionResult:
267 """Execute the FSM until terminal state or limits reached.
269 Returns:
270 ExecutionResult with final state and execution metadata
271 """
272 self.started_at = _iso_now()
273 self.start_time_ms = _now_ms()
275 self._emit("loop_start", {"loop": self.fsm.name})
277 try:
278 while True:
279 # Check shutdown request (signal handling)
280 if self._shutdown_requested:
281 return self._finish("signal")
283 # Check iteration limit
284 if self.iteration >= self.fsm.max_iterations:
285 if self.fsm.on_max_iterations is not None and not self._summary_state_executed:
286 self._emit(
287 "max_iterations_summary",
288 {
289 "summary_state": self.fsm.on_max_iterations,
290 "iterations": self.iteration,
291 },
292 )
293 self._summary_state_executed = True
294 self.current_state = self.fsm.on_max_iterations
295 # Fall through — let the summary state run in this iteration.
296 # (do not `continue`; that would re-trigger the cap check
297 # before the state executes since self.iteration is unchanged.)
298 else:
299 return self._finish("max_iterations")
301 # Check timeout
302 if self.fsm.timeout:
303 elapsed = _now_ms() - self.start_time_ms + self.elapsed_offset_ms
304 if elapsed > self.fsm.timeout * 1000:
305 # BUG-1226: if timeout fires in the race window between
306 # a `route` event and `state_enter`, flush one pending
307 # shell-action state before honoring the timeout so its
308 # side effect (e.g. copying a handshake flag) is not
309 # silently lost. Bounded to shell actions — slash
310 # commands and sub-loops would violate the timeout
311 # budget. Single-step: no cascade.
312 if self._just_routed:
313 pending = self.fsm.states.get(self.current_state)
314 if (
315 pending is not None
316 and not pending.terminal
317 and pending.loop is None
318 and pending.action is not None
319 and self._action_mode(pending) == "shell"
320 ):
321 self._flush_pending_shell_state(pending)
322 return self._finish("timeout")
324 # Get current state config
325 state_config = self.fsm.states[self.current_state]
327 # Update per-state retry tracking based on transition from previous iteration.
328 # If re-entering the same state consecutively, increment retry count.
329 # If entering a different state, clear the previous state's retry count.
330 if self._prev_state is not None:
331 if self.current_state == self._prev_state:
332 self._retry_counts[self.current_state] = (
333 self._retry_counts.get(self.current_state, 0) + 1
334 )
335 else:
336 self._retry_counts.pop(self._prev_state, None)
337 self._throttle_counts.pop(self._prev_state, None)
339 # Check terminal
340 if state_config.terminal:
341 # Handle maintain mode - restart loop instead of terminating
342 if self.fsm.maintain:
343 self.iteration += 1
344 maintain_target = state_config.on_maintain or self.fsm.initial
345 self._emit(
346 "route",
347 {
348 "from": self.current_state,
349 "to": maintain_target,
350 "reason": "maintain",
351 },
352 )
353 self._prev_state = self.current_state
354 self.current_state = maintain_target
355 self._just_routed = True
356 continue
357 # ENH-1631: if we arrived here via the on_max_iterations summary
358 # state, preserve terminated_by="max_iterations" so audit tooling
359 # and PersistentExecutor see "interrupted" rather than "completed".
360 if self._summary_state_executed:
361 return self._finish("max_iterations")
362 return self._finish("terminal")
364 # Check per-state retry limit. If the consecutive re-entry count exceeds
365 # max_retries, skip execution and route to on_retry_exhausted instead.
366 if state_config.max_retries is not None:
367 retry_count = self._retry_counts.get(self.current_state, 0)
368 if retry_count > state_config.max_retries:
369 # on_retry_exhausted is guaranteed non-None by validation when
370 # max_retries is set, but we fall back to an error if misconfigured.
371 exhausted_state: str = state_config.on_retry_exhausted or ""
372 if not exhausted_state:
373 return self._finish(
374 "error",
375 error=f"State '{self.current_state}' exceeded max_retries "
376 "but on_retry_exhausted is not set",
377 )
378 self._emit(
379 "retry_exhausted",
380 {
381 "state": self.current_state,
382 "retries": retry_count,
383 "next": exhausted_state,
384 },
385 )
386 self._retry_counts.pop(self.current_state, None)
387 self._prev_state = self.current_state
388 self.current_state = exhausted_state
389 continue
391 self.iteration += 1
392 self._just_routed = False
393 self._emit(
394 "state_enter",
395 {
396 "state": self.current_state,
397 "iteration": self.iteration,
398 },
399 )
401 # Execute state
402 next_state = self._execute_state(state_config)
404 # Check for pending error signal (FATAL_ERROR)
405 if self._pending_error is not None:
406 return self._finish("error", error=self._pending_error)
408 # Check for pending stall abort (FEAT-1637). The detector
409 # fired with on_repeated_failure="abort" inside _execute_state;
410 # terminate cleanly via _finish (mirrors the cycle_detected
411 # guard below at lines 397-416).
412 if self._pending_stall_abort is not None:
413 stall = self._pending_stall_abort
414 s_state, s_exit, s_verdict = stall.triple
415 return self._finish(
416 "stall_detected",
417 error=(
418 f"Stall detected: state '{s_state}' produced "
419 f"(exit_code={s_exit}, verdict='{s_verdict}') "
420 f"for {stall.count} consecutive iterations"
421 ),
422 )
424 # Check for pending handoff signal
425 if self._pending_handoff:
426 return self._handle_handoff(self._pending_handoff)
428 # Handle maintain mode
429 if next_state is None and self.fsm.maintain:
430 next_state = state_config.on_maintain or self.fsm.initial
432 # SIGKILL in _execute_state sets shutdown flag and returns None
433 if next_state is None and self._shutdown_requested:
434 return self._finish("signal")
436 if next_state is None:
437 return self._finish("error", error="No valid transition")
439 # At this point next_state is guaranteed to be str
440 resolved_next: str = next_state
442 self._emit(
443 "route",
444 {
445 "from": self.current_state,
446 "to": resolved_next,
447 },
448 )
450 # Per-edge revisit tracking for cycle detection.
451 edge_key = f"{self.current_state}->{resolved_next}"
452 self._edge_revisit_counts[edge_key] = self._edge_revisit_counts.get(edge_key, 0) + 1
453 if self._edge_revisit_counts[edge_key] > self.fsm.max_edge_revisits:
454 self._emit(
455 "cycle_detected",
456 {
457 "edge": edge_key,
458 "from": self.current_state,
459 "to": resolved_next,
460 "count": self._edge_revisit_counts[edge_key],
461 "max": self.fsm.max_edge_revisits,
462 },
463 )
464 return self._finish(
465 "cycle_detected",
466 error=f"Cycle detected: edge {edge_key} traversed "
467 f"{self._edge_revisit_counts[edge_key]} times "
468 f"(limit: {self.fsm.max_edge_revisits})",
469 )
471 self._prev_state = self.current_state
472 self.current_state = resolved_next
473 self._just_routed = True
475 # Interruptible backoff sleep between iterations
476 if self.fsm.backoff and self.fsm.backoff > 0:
477 deadline = time.time() + self.fsm.backoff
478 while time.time() < deadline:
479 if self._shutdown_requested:
480 break
481 time.sleep(min(0.1, deadline - time.time()))
483 except InterpolationError as exc:
484 return self._finish(
485 "error",
486 error=(
487 f"Missing context variable in state '{self.current_state}': {exc}. "
488 f"Run with: ll-loop run {self.fsm.name} --context KEY=VALUE"
489 ),
490 )
491 except Exception as exc:
492 return self._finish("error", error=str(exc))
494 def _execute_sub_loop(self, state: StateConfig, ctx: InterpolationContext) -> str | None:
495 """Execute a sub-loop state by loading and running a child FSM.
497 Args:
498 state: The state configuration with loop field set
499 ctx: Interpolation context for routing
501 Returns:
502 Next state name based on child loop verdict, or None
503 """
504 from little_loops.cli.loop._helpers import resolve_loop_path
505 from little_loops.fsm.validation import load_and_validate
507 assert state.loop is not None # guarded by caller
508 loop_name = interpolate(state.loop, ctx)
509 loop_path = resolve_loop_path(loop_name, self.loops_dir or Path(".loops"))
510 child_fsm, _ = load_and_validate(loop_path)
512 # Bind child context: explicit with: bindings take precedence over legacy passthrough
513 if state.with_:
514 from little_loops.fsm.interpolation import interpolate_dict
516 resolved = interpolate_dict(state.with_, ctx)
517 # Apply declared defaults for unbound optional parameters
518 for param_name, param_spec in child_fsm.parameters.items():
519 if (
520 param_name not in resolved
521 and not param_spec.required
522 and param_spec.default is not None
523 ):
524 resolved[param_name] = param_spec.default
525 # Runtime check: required parameters must be present after interpolation
526 for param_name, param_spec in child_fsm.parameters.items():
527 if param_spec.required and param_name not in resolved:
528 raise ValueError(
529 f"Sub-loop '{state.loop}' requires parameter '{param_name}' "
530 f"but it is not bound in 'with'"
531 )
532 # Merge: child's own context block provides base; with: bindings override
533 child_fsm.context = {**child_fsm.context, **resolved}
534 elif state.context_passthrough:
535 # Extract .output strings from capture result dicts so ${context.key} resolves
536 # to the plain output string (e.g. "ENH-123") rather than the full capture object.
537 captured_as_context = {
538 k: v["output"] if isinstance(v, dict) and "exit_code" in v else v
539 for k, v in self.captured.items()
540 }
541 child_fsm.context = {**self.fsm.context, **captured_as_context, **child_fsm.context}
543 depth = self._depth + 1
544 child_events: list[dict] = []
546 def _sub_event_callback(event: dict) -> None:
547 child_events.append(event)
548 # Only inject depth if not already set by a deeper nested sub-loop
549 if "depth" not in event:
550 self.event_callback({**event, "depth": depth})
551 else:
552 self.event_callback(event)
554 child_executor = FSMExecutor(
555 child_fsm,
556 action_runner=self.action_runner,
557 loops_dir=self.loops_dir,
558 event_callback=_sub_event_callback,
559 circuit=self._circuit,
560 )
561 child_executor._depth = depth # propagate depth for further nesting
563 # Clamp child timeout to parent's remaining wall-clock budget so a slow sub-loop
564 # can't silently consume the parent's deadline with no recourse for the parent FSM.
565 if self.fsm.timeout:
566 elapsed_ms = _now_ms() - self.start_time_ms + self.elapsed_offset_ms
567 remaining_s = max(1, int((self.fsm.timeout * 1000 - elapsed_ms) // 1000))
568 if child_fsm.timeout is None or child_fsm.timeout > remaining_s:
569 child_fsm.timeout = remaining_s
571 child_result = child_executor.run()
573 # Capture child event stream as a JSON-lines string if the state declares a capture key
574 if state.capture and child_events:
575 import json as _json
577 self.captured[state.capture] = {
578 "output": "\n".join(_json.dumps(e) for e in child_events),
579 "exit_code": None,
580 }
582 # Merge child captures back into parent under the state name
583 if (state.context_passthrough or state.with_) and child_executor.captured:
584 self.captured[self.current_state] = child_executor.captured
586 # Route based on child termination reason and terminal state name
587 if child_result.terminated_by == "terminal":
588 if child_result.final_state == "done":
589 return interpolate(state.on_yes, ctx) if state.on_yes else None
590 else:
591 # Reached a non-done terminal (e.g. "failed") → failure
592 return interpolate(state.on_no, ctx) if state.on_no else None
593 elif child_result.terminated_by == "error":
594 # Runtime child failure (not a YAML load error)
595 if state.on_error:
596 return interpolate(state.on_error, ctx)
597 return interpolate(state.on_no, ctx) if state.on_no else None
598 else:
599 # max_iterations, timeout, signal — all are failure
600 return interpolate(state.on_no, ctx) if state.on_no else None
602 def _execute_learning_state(self, state: StateConfig, ctx: InterpolationContext) -> str | None:
603 """Execute a FEAT-1283 ``type: learning`` state.
605 Iterates ``state.learning.targets`` in order. For each target:
606 1. Look up its record in the learning-tests registry (ENH-1282).
607 2. If proven → continue.
608 3. If refuted → emit ``learning_target_refuted`` + ``learning_blocked``
609 and route to ``on_blocked`` (preferred) or ``on_no``.
610 4. If missing or stale → emit ``learning_target_stale`` and invoke
611 ``/ll:explore-api <target>`` via the executor's action_runner;
612 re-check the registry; repeat up to ``max_retries`` times before
613 emitting ``learning_blocked`` (reason ``retries_exhausted``) and
614 routing to ``on_blocked``/``on_no``.
616 When every target ends up proven, emit ``learning_complete`` and route
617 to ``on_yes``. Returns the resolved next-state name (or ``None`` when no
618 route is configured for the terminal verdict, mirroring ``_route``).
619 """
620 from little_loops.learning_tests import check_learning_test
622 assert state.learning is not None # guarded by caller
624 def _blocked_target(reason: str, target: str) -> str | None:
625 self._emit(
626 "learning_blocked",
627 {"state": self.current_state, "target": target, "reason": reason},
628 )
629 route = state.on_blocked or state.on_no
630 return interpolate(route, ctx) if route else None
632 for target in state.learning.targets:
633 record = check_learning_test(target)
635 attempts = 0
636 while record is None or record.status == "stale":
637 if attempts >= state.learning.max_retries:
638 return _blocked_target("retries_exhausted", target)
640 if record is None:
641 self._emit(
642 "learning_target_stale",
643 {"state": self.current_state, "target": target, "cause": "missing"},
644 )
645 else:
646 self._emit(
647 "learning_target_stale",
648 {"state": self.current_state, "target": target, "cause": "stale"},
649 )
651 self._emit(
652 "learning_explore_invoked",
653 {"state": self.current_state, "target": target, "attempt": attempts + 1},
654 )
655 self._run_action(f"/ll:explore-api {target}", state, ctx)
656 attempts += 1
657 record = check_learning_test(target)
659 if record.status == "refuted":
660 self._emit(
661 "learning_target_refuted",
662 {"state": self.current_state, "target": target},
663 )
664 return _blocked_target("refuted", target)
666 self._emit(
667 "learning_target_proven",
668 {"state": self.current_state, "target": target},
669 )
671 self._emit(
672 "learning_complete",
673 {"state": self.current_state, "targets": list(state.learning.targets)},
674 )
675 return interpolate(state.on_yes, ctx) if state.on_yes else None
677 def _compute_progress_fingerprint(self, ctx: InterpolationContext) -> tuple[object, ...] | None:
678 """Return an (mtime, size) fingerprint for configured progress_paths, or None.
680 Called just before stall_detector.record() so that file changes made by
681 intermediate next:-only states are visible to the detector. Returns None
682 when no progress_paths are configured (preserves existing semantics).
683 """
684 if self.fsm.circuit is None or self.fsm.circuit.repeated_failure is None:
685 return None
686 paths = self.fsm.circuit.repeated_failure.progress_paths
687 if not paths:
688 return None
689 entries: list[tuple[float, int]] = []
690 for raw_path in paths:
691 try:
692 resolved = interpolate(raw_path, ctx)
693 except Exception:
694 continue
695 p = Path(resolved)
696 if p.exists():
697 st = p.stat()
698 entries.append((st.st_mtime, st.st_size))
699 else:
700 entries.append((0.0, 0))
701 return tuple(entries) if entries else None
703 def _check_throttle(self, state: StateConfig, state_name: str) -> str | None:
704 """Increment the per-state tool-call counter and enforce throttle thresholds.
706 Called after every action execution within a state visit. Returns the forced
707 next-state name when the hard threshold is reached, or None when execution
708 should continue normally (warn events are emitted but do not redirect).
710 Sets self._pending_error and returns "__STOP__" when the call count exceeds
711 hard_max with no on_throttle_hard target — the caller must propagate this as
712 a None return from _execute_state so the main loop detects _pending_error.
713 """
714 count = self._throttle_counts.get(state_name, 0) + 1
715 self._throttle_counts[state_name] = count
717 throttle = state.throttle
718 normal_max = (
719 throttle.normal_max
720 if (throttle and throttle.normal_max is not None)
721 else _DEFAULT_THROTTLE_NORMAL_MAX
722 )
723 warn_max = (
724 throttle.warn_max
725 if (throttle and throttle.warn_max is not None)
726 else _DEFAULT_THROTTLE_WARN_MAX
727 )
728 hard_max = (
729 throttle.hard_max
730 if (throttle and throttle.hard_max is not None)
731 else _DEFAULT_THROTTLE_HARD_MAX
732 )
734 if count == warn_max:
735 self._emit(
736 THROTTLE_WARN_EVENT,
737 {
738 "state": state_name,
739 "count": count,
740 "normal_max": normal_max,
741 "warn_max": warn_max,
742 "hard_max": hard_max,
743 },
744 )
746 # States with type="learning" (FEAT-1283) are exempt from hard_max — they
747 # legitimately make N calls per visit (one per unproven target).
748 if state.type == "learning":
749 return None
751 if count == hard_max:
752 next_target = state.on_throttle_hard or state.on_error
753 self._emit(
754 THROTTLE_HARD_EVENT,
755 {"state": state_name, "count": count, "hard_max": hard_max, "next": next_target},
756 )
757 return next_target
759 if count > hard_max:
760 self._emit(
761 THROTTLE_STOP_EVENT,
762 {"state": state_name, "count": count, "hard_max": hard_max},
763 )
764 self._pending_error = (
765 f"Throttle stop: state '{state_name}' exceeded hard_max={hard_max} "
766 "tool calls with no on_throttle_hard target"
767 )
768 return "__STOP__"
770 return None
772 def _execute_state(self, state: StateConfig) -> str | None:
773 """Execute a single state and return next state name.
775 Args:
776 state: The state configuration to execute
778 Returns:
779 Next state name, or None if no valid transition
780 """
781 # Build interpolation context
782 ctx = self._build_context()
784 # Dispatch to sub-loop handler if this is a sub-loop state
785 if state.loop is not None:
786 try:
787 return self._execute_sub_loop(state, ctx)
788 except (FileNotFoundError, ValueError):
789 if state.on_error:
790 return interpolate(state.on_error, ctx)
791 raise
793 # FEAT-1283: dispatch to learning-state handler when both type="learning"
794 # AND a LearningConfig is present. The bare `type="learning"` marker
795 # (used pre-FEAT-1283 only as a throttle hard_max exemption hint, see
796 # ThrottleConfig docstring) falls through to normal action execution.
797 if state.type == "learning" and state.learning is not None:
798 return self._execute_learning_state(state, ctx)
800 # Handle unconditional transition
801 if state.next:
802 if state.action:
803 self._maybe_wait_for_circuit(state)
804 result, routed = self._run_action_or_route(state, ctx)
805 if routed is not None:
806 return routed
807 throttle_next = self._check_throttle(state, self.current_state)
808 if throttle_next == "__STOP__":
809 return None
810 if throttle_next is not None:
811 return throttle_next
812 assert result is not None
813 self.prev_result = {
814 "output": result.output,
815 "exit_code": result.exit_code,
816 "state": self.current_state,
817 }
818 if result.exit_code is not None and result.exit_code < 0:
819 # Process killed by signal — do not silently advance via next
820 if state.on_error:
821 return interpolate(state.on_error, ctx)
822 self.request_shutdown()
823 return None
824 # Non-zero exit: if on_error is defined, treat next as success path only
825 if result.exit_code != 0 and state.on_error:
826 return interpolate(state.on_error, ctx)
827 return interpolate(state.next, ctx)
829 # Execute action if present
830 action_result = None
831 if state.action:
832 self._maybe_wait_for_circuit(state)
833 action_result, routed = self._run_action_or_route(state, ctx)
834 if routed is not None:
835 return routed
836 throttle_next = self._check_throttle(state, self.current_state)
837 if throttle_next == "__STOP__":
838 return None
839 if throttle_next is not None:
840 return throttle_next
842 # Evaluate
843 eval_result = self._evaluate(state, action_result, ctx)
844 self.prev_result = {
845 "output": action_result.output if action_result else "",
846 "exit_code": action_result.exit_code if action_result else 0,
847 "state": self.current_state,
848 }
850 # Update context with result for routing interpolation
851 if eval_result:
852 ctx.result = {
853 "verdict": eval_result.verdict,
854 "details": eval_result.details,
855 }
857 # Route based on verdict
858 verdict = eval_result.verdict if eval_result else "yes"
860 # Stall detection (FEAT-1637). Record this transition's triple and
861 # check whether the last `window` triples are identical. On stall,
862 # either abort (set _pending_stall_abort for run() to catch) or
863 # override next_state to the configured recovery target.
864 stall_route_target: str | None = None
865 if self._stall_detector is not None:
866 stall_exit_code = action_result.exit_code if action_result else 0
867 stall_fingerprint = self._compute_progress_fingerprint(ctx)
868 self._stall_detector.record(
869 self.current_state, stall_exit_code, verdict, stall_fingerprint
870 )
871 stall = self._stall_detector.check()
872 if stall is not None:
873 assert self.fsm.circuit is not None
874 assert self.fsm.circuit.repeated_failure is not None
875 cfg_action = self.fsm.circuit.repeated_failure.on_repeated_failure
876 self._emit(
877 STALL_DETECTED_EVENT,
878 {
879 "state": self.current_state,
880 "exit_code": stall_exit_code,
881 "verdict": verdict,
882 "consecutive": stall.count,
883 "action": "abort" if cfg_action == "abort" else f"route:{cfg_action}",
884 },
885 )
886 if cfg_action == "abort":
887 self._pending_stall_abort = stall
888 return None
889 # Route to recovery target; bypass _route() entirely so the
890 # eval verdict does not pull us elsewhere first.
891 stall_route_target = cfg_action
893 route_ctx = RouteContext(
894 state_name=self.current_state,
895 state=state,
896 verdict=verdict,
897 action_result=action_result,
898 eval_result=eval_result,
899 ctx=ctx,
900 iteration=self.iteration,
901 )
902 # 429 / rate-limit detection — runs before interceptors so an in-place retry
903 # returns early without dispatching to registered before_route hooks.
904 if action_result is not None:
905 _combined = (action_result.output or "") + "\n" + (action_result.stderr or "")
906 _failure_type, _reason = classify_failure(_combined, action_result.exit_code)
907 if _failure_type == FailureType.TRANSIENT and (
908 "rate limit" in _reason.lower() or "quota" in _reason.lower()
909 ):
910 _handled, _target = self._handle_rate_limit(state, route_ctx.state_name)
911 if _handled:
912 return _target
913 elif _failure_type == FailureType.TRANSIENT and "api server error" in _reason.lower():
914 _handled, _target = self._handle_api_error(state, route_ctx.state_name)
915 if _handled:
916 return _target
917 # exhausted — fall through to normal verdict routing
918 else:
919 # Not rate-limited or server-error: reset counters so future transients start fresh.
920 self._rate_limit_retries.pop(route_ctx.state_name, None)
921 self._consecutive_rate_limit_exhaustions = 0
922 self._api_error_retries.pop(route_ctx.state_name, None)
924 # Stall-route override: if the detector elected to route to a recovery
925 # state, honor it now (bypass interceptors and _route) so the
926 # configured target wins over the eval verdict.
927 if stall_route_target is not None:
928 return stall_route_target
930 for interceptor in self._interceptors:
931 if hasattr(interceptor, "before_route"):
932 decision = interceptor.before_route(route_ctx)
933 if isinstance(decision, RouteDecision):
934 if decision.next_state is None:
935 return None # veto
936 return decision.next_state # redirect — bypass _route()
937 next_state = self._route(state, verdict, ctx)
938 for interceptor in self._interceptors:
939 if hasattr(interceptor, "after_route"):
940 interceptor.after_route(route_ctx)
941 return next_state
943 def _run_action(
944 self,
945 action_template: str,
946 state: StateConfig,
947 ctx: InterpolationContext,
948 ) -> ActionResult:
949 """Execute action and optionally capture result.
951 Args:
952 action_template: Action string (may contain variables)
953 state: State configuration
954 ctx: Interpolation context
956 Returns:
957 ActionResult with output and exit code
958 """
959 action = interpolate(action_template, ctx)
960 action_mode = self._action_mode(state)
962 self._emit("action_start", {"action": action, "is_prompt": action_mode == "prompt"})
964 def _on_line(line: str) -> None:
965 self._emit("action_output", {"line": line})
967 if action_mode == "mcp_tool":
968 # Direct MCP tool call — bypass action_runner entirely
969 interpolated_params = interpolate_dict(state.params, ctx) if state.params else {}
970 cmd = ["mcp-call", action, json.dumps(interpolated_params)]
971 result = self._run_subprocess(
972 cmd,
973 timeout=state.timeout or self.fsm.default_timeout or 30,
974 on_output_line=_on_line,
975 )
976 elif action_mode == "contributed":
977 assert (
978 state.action_type is not None
979 ) # guaranteed by _action_mode returning "contributed"
980 runner = self._contributed_actions[state.action_type]
981 result = runner.run(
982 action,
983 timeout=state.timeout or self.fsm.default_timeout or 3600,
984 is_slash_command=False,
985 on_output_line=_on_line,
986 )
987 else:
988 result = self.action_runner.run(
989 action,
990 timeout=state.timeout or self.fsm.default_timeout or 3600,
991 is_slash_command=action_mode == "prompt",
992 on_output_line=_on_line,
993 agent=state.agent if action_mode == "prompt" else None,
994 tools=state.tools if action_mode == "prompt" else None,
995 )
997 preview = result.output[-2000:].strip() if result.output else None
998 payload: dict[str, Any] = {
999 "exit_code": result.exit_code,
1000 "duration_ms": result.duration_ms,
1001 "output_preview": preview,
1002 "is_prompt": action_mode == "prompt",
1003 }
1004 if action_mode == "prompt":
1005 session_jsonl = get_current_session_jsonl()
1006 payload["session_jsonl"] = str(session_jsonl) if session_jsonl else None
1007 self._emit("action_complete", payload)
1009 # Capture if requested
1010 if state.capture:
1011 self.captured[state.capture] = {
1012 "output": result.output.rstrip("\n\r"),
1013 "stderr": result.stderr,
1014 "exit_code": result.exit_code,
1015 "duration_ms": result.duration_ms,
1016 }
1018 # Check for signals in output
1019 if self.signal_detector:
1020 signal = self.signal_detector.detect_first(result.output)
1021 if signal:
1022 if signal.signal_type == "handoff":
1023 self._pending_handoff = signal
1024 elif signal.signal_type == "error":
1025 self._pending_error = signal.payload
1026 elif signal.signal_type == "stop":
1027 self.request_shutdown()
1029 return result
1031 def _run_subprocess(
1032 self,
1033 cmd: list[str],
1034 timeout: int,
1035 on_output_line: Any | None = None,
1036 ) -> ActionResult:
1037 """Run a subprocess directly and return ActionResult.
1039 Follows the same Popen + stderr-drain-thread pattern as DefaultActionRunner.
1041 Args:
1042 cmd: Command and arguments to execute
1043 timeout: Timeout in seconds
1044 on_output_line: Optional callback for each stdout line
1046 Returns:
1047 ActionResult with output, stderr, exit_code, duration_ms
1048 """
1049 start = _now_ms()
1050 process = subprocess.Popen(
1051 cmd,
1052 stdout=subprocess.PIPE,
1053 stderr=subprocess.PIPE,
1054 text=True,
1055 )
1056 self._current_process = process
1057 output_chunks: list[str] = []
1058 stderr_chunks: list[str] = []
1060 def _drain_stderr() -> None:
1061 assert process.stderr is not None
1062 for line in process.stderr:
1063 stderr_chunks.append(line)
1065 stderr_thread = threading.Thread(target=_drain_stderr, daemon=True)
1066 stderr_thread.start()
1068 try:
1069 for line in process.stdout: # type: ignore[union-attr]
1070 output_chunks.append(line)
1071 if on_output_line:
1072 on_output_line(line.rstrip())
1073 process.wait(timeout=timeout)
1074 except subprocess.TimeoutExpired:
1075 process.kill()
1076 process.wait()
1077 stderr_thread.join(timeout=5)
1078 return ActionResult(
1079 output="".join(output_chunks),
1080 stderr="".join(stderr_chunks) or "MCP call timed out",
1081 exit_code=124,
1082 duration_ms=timeout * 1000,
1083 )
1084 finally:
1085 self._current_process = None
1086 stderr_thread.join(timeout=5)
1087 return ActionResult(
1088 output="".join(output_chunks),
1089 stderr="".join(stderr_chunks),
1090 exit_code=process.returncode,
1091 duration_ms=_now_ms() - start,
1092 )
1094 def _evaluate(
1095 self,
1096 state: StateConfig,
1097 action_result: ActionResult | None,
1098 ctx: InterpolationContext,
1099 ) -> EvaluationResult | None:
1100 """Evaluate action result.
1102 Args:
1103 state: State configuration
1104 action_result: Result from action execution (may be None)
1105 ctx: Interpolation context
1107 Returns:
1108 EvaluationResult, or None if no evaluation needed
1109 """
1110 if state.evaluate is None:
1111 # Default evaluation based on action type
1112 if action_result:
1113 action_mode = self._action_mode(state)
1115 if action_mode == "mcp_tool":
1116 # MCP tool call: use mcp_result evaluator
1117 result = evaluate_mcp_result(action_result.output, action_result.exit_code)
1118 elif action_mode == "prompt":
1119 # Slash command or prompt: use LLM evaluation
1120 if not self.fsm.llm.enabled:
1121 result = EvaluationResult(
1122 verdict="error",
1123 details={"error": "LLM evaluation disabled via --no-llm"},
1124 )
1125 else:
1126 result = evaluate_llm_structured(
1127 action_result.output,
1128 model=self.fsm.llm.model,
1129 max_tokens=self.fsm.llm.max_tokens,
1130 timeout=self.fsm.llm.timeout,
1131 )
1132 else:
1133 # Shell command: use exit code
1134 result = evaluate_exit_code(action_result.exit_code)
1136 self._emit(
1137 "evaluate",
1138 {
1139 "type": "default",
1140 "verdict": result.verdict,
1141 **result.details,
1142 },
1143 )
1144 return result
1145 return None
1147 # Explicit evaluation config
1148 raw_output = action_result.output if action_result else ""
1149 if state.evaluate.source:
1150 try:
1151 eval_input = interpolate(state.evaluate.source, ctx)
1152 except InterpolationError:
1153 eval_input = raw_output
1154 else:
1155 eval_input = raw_output
1157 if state.evaluate.type in self._contributed_evaluators:
1158 result = self._contributed_evaluators[state.evaluate.type](
1159 state.evaluate,
1160 eval_input,
1161 action_result.exit_code if action_result else 0,
1162 ctx,
1163 )
1164 elif state.evaluate.type == "llm_structured" and not self.fsm.llm.enabled:
1165 result = EvaluationResult(
1166 verdict="error",
1167 details={"error": "LLM evaluation disabled via --no-llm"},
1168 )
1169 else:
1170 result = evaluate(
1171 config=state.evaluate,
1172 output=eval_input,
1173 exit_code=action_result.exit_code if action_result else 0,
1174 context=ctx,
1175 )
1177 self._emit(
1178 "evaluate",
1179 {
1180 "type": state.evaluate.type,
1181 "verdict": result.verdict,
1182 **result.details,
1183 },
1184 )
1186 return result
1188 def _route(
1189 self,
1190 state: StateConfig,
1191 verdict: str,
1192 ctx: InterpolationContext,
1193 ) -> str | None:
1194 """Determine next state from verdict.
1196 Resolution order (from design doc):
1197 1. next (unconditional) - handled before this method
1198 2. route (full routing table)
1199 3. on_success/on_failure/on_error (shorthand)
1200 4. terminal - handled in main loop
1201 5. error
1203 Args:
1204 state: State configuration
1205 verdict: Verdict string from evaluation
1206 ctx: Interpolation context
1208 Returns:
1209 Next state name, or None if no valid route
1210 """
1211 if state.route:
1212 routes = state.route.routes
1213 if verdict in routes:
1214 return self._resolve_route(routes[verdict], ctx)
1215 if state.route.default:
1216 return self._resolve_route(state.route.default, ctx)
1217 if verdict == "error" and state.route.error:
1218 return self._resolve_route(state.route.error, ctx)
1219 return None
1221 # Shorthand routing
1222 if verdict == "yes" and state.on_yes:
1223 return self._resolve_route(state.on_yes, ctx)
1224 if verdict == "no" and state.on_no:
1225 return self._resolve_route(state.on_no, ctx)
1226 if verdict == "error" and state.on_error:
1227 return self._resolve_route(state.on_error, ctx)
1228 if verdict == "partial" and state.on_partial:
1229 return self._resolve_route(state.on_partial, ctx)
1230 if verdict == "blocked" and state.on_blocked:
1231 return self._resolve_route(state.on_blocked, ctx)
1233 # Dynamic on_<verdict> shorthands from extra_routes
1234 if verdict in state.extra_routes:
1235 return self._resolve_route(state.extra_routes[verdict], ctx)
1237 return None
1239 def _resolve_route(self, route: str, ctx: InterpolationContext) -> str:
1240 """Resolve route target, handling special tokens.
1242 Args:
1243 route: Route target string
1244 ctx: Interpolation context
1246 Returns:
1247 Resolved state name
1248 """
1249 if route == "$current":
1250 return self.current_state
1251 return interpolate(route, ctx)
1253 def _flush_pending_shell_state(self, state: StateConfig) -> None:
1254 """Execute a pending shell-action state's action before honoring a
1255 wall-clock timeout. BUG-1226: closes the narrow race between emitting
1256 a `route` event and `state_enter` so handshake states (e.g. autodev's
1257 ``copy_broke_down``) do not silently drop their side effect when the
1258 timeout fires in that window. Single-step: we run the action but do
1259 not follow its routing — ``final_state`` stays as the flushed state.
1260 """
1261 assert state.action is not None # guarded by caller
1262 self.iteration += 1
1263 self._just_routed = False
1264 self._emit(
1265 "state_enter",
1266 {
1267 "state": self.current_state,
1268 "iteration": self.iteration,
1269 "flushed": True,
1270 },
1271 )
1272 ctx = self._build_context()
1273 try:
1274 self._run_action(state.action, state, ctx)
1275 except Exception:
1276 # Deliberately swallow — the timeout is being honored regardless
1277 # of whether the flushed action succeeded.
1278 pass
1280 def _action_mode(self, state: StateConfig) -> str:
1281 """Return execution mode for the state: 'prompt', 'shell', or 'mcp_tool'."""
1282 if state.action_type == "mcp_tool":
1283 return "mcp_tool"
1284 if state.action_type in ("prompt", "slash_command"):
1285 return "prompt"
1286 if state.action_type == "shell":
1287 return "shell"
1288 if state.action_type in self._contributed_actions:
1289 return "contributed"
1290 # Heuristic: / prefix = slash_command (prompt mode)
1291 if state.action is not None and state.action.startswith("/"):
1292 return "prompt"
1293 return "shell"
1295 def _run_action_or_route(
1296 self, state: StateConfig, ctx: InterpolationContext
1297 ) -> tuple[ActionResult | None, str | None]:
1298 """Run the state action, routing unhandled exceptions to on_error.
1300 Returns (action_result, routed_target). ``routed_target`` is a non-None
1301 next-state string only when an exception was raised AND ``state.on_error``
1302 is defined; in that case ``action_result`` is None. When no on_error is
1303 set, the exception is re-raised for the top-level ``run()`` handler.
1304 """
1305 assert state.action is not None # caller-guarded
1306 try:
1307 return self._run_action(state.action, state, ctx), None
1308 except Exception as exc:
1309 if state.on_error:
1310 self._emit(
1311 "action_error",
1312 {
1313 "state": self.current_state,
1314 "error": str(exc),
1315 "route": "on_error",
1316 },
1317 )
1318 return None, interpolate(state.on_error, ctx)
1319 raise
1321 def _build_context(self) -> InterpolationContext:
1322 """Build interpolation context for current state.
1324 Returns:
1325 InterpolationContext with all runtime values
1326 """
1327 return InterpolationContext(
1328 context=self.fsm.context,
1329 captured=self.captured,
1330 prev=self.prev_result,
1331 result=None,
1332 state_name=self.current_state,
1333 iteration=self.iteration,
1334 loop_name=self.fsm.name,
1335 started_at=self.started_at,
1336 elapsed_ms=_now_ms() - self.start_time_ms + self.elapsed_offset_ms,
1337 )
1339 def _emit(self, event: str, data: dict[str, Any]) -> None:
1340 """Emit an event via the callback."""
1341 self.event_callback(
1342 {
1343 "event": event,
1344 "ts": _iso_now(),
1345 **data,
1346 }
1347 )
1349 def _handle_rate_limit(self, state: StateConfig, state_name: str) -> tuple[bool, str | None]:
1350 """Handle a detected 429/rate-limit action outcome.
1352 Implements the two-tier retry ladder:
1353 1. Short-burst tier: up to ``max_rate_limit_retries`` attempts with
1354 exponential backoff (``rate_limit_backoff_base_seconds * 2^n + jitter``).
1355 2. Long-wait tier: walks ``rate_limit_long_wait_ladder`` with index capped
1356 at the last entry, accumulating ``total_wait_seconds``.
1358 Routes to ``on_rate_limit_exhausted`` (falling back to ``on_error``) only
1359 once ``total_wait_seconds >= rate_limit_max_wait_seconds``. Emits
1360 ``rate_limit_exhausted`` on routing (including tier counters) and
1361 ``rate_limit_storm`` when consecutive exhaustions reach the threshold.
1363 Returns:
1364 (handled, target). ``handled=True`` means the caller should return
1365 ``target`` directly (in-place retry uses ``state_name``; exhaustion
1366 uses the routed target). ``handled=False`` should not occur for the
1367 current rate-limit classification path but is reserved for future
1368 extensions.
1369 """
1370 _short_max = (
1371 state.max_rate_limit_retries
1372 if state.max_rate_limit_retries is not None
1373 else _DEFAULT_RATE_LIMIT_RETRIES
1374 )
1375 _backoff_base = (
1376 state.rate_limit_backoff_base_seconds
1377 if state.rate_limit_backoff_base_seconds is not None
1378 else _DEFAULT_RATE_LIMIT_BACKOFF_BASE
1379 )
1380 _max_wait = (
1381 state.rate_limit_max_wait_seconds
1382 if state.rate_limit_max_wait_seconds is not None
1383 else _DEFAULT_RATE_LIMIT_MAX_WAIT_SECONDS
1384 )
1385 _ladder = (
1386 state.rate_limit_long_wait_ladder
1387 if state.rate_limit_long_wait_ladder is not None
1388 else _DEFAULT_RATE_LIMIT_LONG_WAIT_LADDER
1389 )
1391 record = self._rate_limit_retries.get(state_name)
1392 if record is None:
1393 record = {
1394 "short_retries": 0,
1395 "long_retries": 0,
1396 "total_wait_seconds": 0.0,
1397 "first_seen_at": time.time(),
1398 }
1399 self._rate_limit_retries[state_name] = record
1401 short_retries = int(record.get("short_retries", 0))
1402 long_retries = int(record.get("long_retries", 0))
1403 total_wait = float(record.get("total_wait_seconds", 0.0))
1405 if short_retries < _short_max:
1406 # Short-burst tier — exponential backoff with jitter. Budget is not
1407 # checked here; short-tier always advances to long-wait on exhaustion.
1408 short_retries += 1
1409 record["short_retries"] = short_retries
1410 _sleep = _backoff_base * (2 ** (short_retries - 1)) + random.uniform(0, _backoff_base)
1411 if self._circuit is not None:
1412 self._circuit.record_rate_limit(_sleep)
1413 total_wait += self._interruptible_sleep(_sleep)
1414 record["total_wait_seconds"] = total_wait
1415 return True, state_name # retry in place
1417 # Long-wait tier — walk ladder with capped index.
1418 long_retries += 1
1419 record["long_retries"] = long_retries
1420 _idx = min(long_retries - 1, len(_ladder) - 1)
1421 _wait = float(_ladder[_idx])
1422 if self._circuit is not None:
1423 self._circuit.record_rate_limit(_wait)
1424 _tier_start = time.time()
1425 _deadline = _tier_start + _wait
1426 _total_wait_before_tier = total_wait
1427 total_wait += self._interruptible_sleep(
1428 _wait,
1429 on_heartbeat=lambda elapsed: self._emit(
1430 RATE_LIMIT_WAITING_EVENT,
1431 {
1432 "state": state_name,
1433 "elapsed_seconds": elapsed,
1434 "next_attempt_at": _deadline,
1435 "total_waited_seconds": _total_wait_before_tier + elapsed,
1436 "budget_seconds": _max_wait,
1437 "tier": "long_wait",
1438 },
1439 ),
1440 )
1441 record["total_wait_seconds"] = total_wait
1442 if total_wait >= _max_wait:
1443 return True, self._exhaust_rate_limit(state, state_name, record)
1444 return True, state_name # retry in place
1446 def _maybe_wait_for_circuit(self, state: StateConfig) -> None:
1447 """Pre-action circuit-breaker check: sleep until shared 429 recovery.
1449 Skips quietly when no circuit is injected, when the state's action is not
1450 an LLM-quota consumer, or when the circuit has no active recovery window.
1451 """
1452 if self._circuit is None:
1453 return
1454 if self._action_mode(state) != "prompt":
1455 return
1456 recovery = self._circuit.get_estimated_recovery()
1457 if recovery is None:
1458 return
1459 wait = recovery - time.time()
1460 if wait > 0:
1461 self._interruptible_sleep(wait)
1463 def _interruptible_sleep(
1464 self,
1465 duration: float,
1466 on_heartbeat: Callable[[float], None] | None = None,
1467 ) -> float:
1468 """Sleep for up to ``duration`` seconds in 100ms ticks, exiting promptly
1469 on ``_shutdown_requested``. Returns the actual elapsed seconds so callers
1470 can accumulate wall-clock time spent in rate-limit waits.
1472 If ``on_heartbeat`` is provided, it is invoked with the elapsed seconds
1473 roughly every ``_RATE_LIMIT_HEARTBEAT_INTERVAL`` seconds so UIs can show
1474 live progress during long waits. The short-tier call site intentionally
1475 omits the callback to preserve backward-compatible silent behavior.
1476 """
1477 if duration <= 0:
1478 return 0.0
1479 _start = time.time()
1480 _deadline = _start + duration
1481 last_heartbeat = _start
1482 while time.time() < _deadline:
1483 if self._shutdown_requested:
1484 break
1485 time.sleep(min(0.1, _deadline - time.time()))
1486 if on_heartbeat is not None:
1487 _now = time.time()
1488 if _now - last_heartbeat >= _RATE_LIMIT_HEARTBEAT_INTERVAL:
1489 on_heartbeat(_now - _start)
1490 last_heartbeat = _now
1491 return time.time() - _start
1493 def _exhaust_rate_limit(
1494 self, state: StateConfig, state_name: str, record: dict[str, Any]
1495 ) -> str | None:
1496 """Finalize rate-limit exhaustion: emit event, storm detection, and
1497 return the routed target. Pops the per-state record and is called only
1498 once the wall-clock budget is spent.
1499 """
1500 self._rate_limit_retries.pop(state_name, None)
1501 target = state.on_rate_limit_exhausted or state.on_error
1502 self._emit(
1503 RATE_LIMIT_EXHAUSTED_EVENT,
1504 {
1505 "state": state_name,
1506 "retries": int(record.get("short_retries", 0)) + int(record.get("long_retries", 0)),
1507 "short_retries": int(record.get("short_retries", 0)),
1508 "long_retries": int(record.get("long_retries", 0)),
1509 "total_wait_seconds": float(record.get("total_wait_seconds", 0.0)),
1510 "next": target,
1511 },
1512 )
1513 self._consecutive_rate_limit_exhaustions += 1
1514 if self._consecutive_rate_limit_exhaustions >= _RATE_LIMIT_STORM_THRESHOLD:
1515 self._emit(
1516 RATE_LIMIT_STORM_EVENT,
1517 {
1518 "state": state_name,
1519 "count": self._consecutive_rate_limit_exhaustions,
1520 },
1521 )
1522 return target
1524 def _handle_api_error(self, state: StateConfig, state_name: str) -> tuple[bool, str | None]:
1525 """Handle a detected API server error with short-burst flat backoff.
1527 Unlike ``_handle_rate_limit``, uses a flat backoff with no long-wait tier and
1528 falls through to normal FSM routing after ``_DEFAULT_API_ERROR_RETRIES`` attempts
1529 so transient infrastructure hiccups don't permanently misdirect the loop.
1531 Returns:
1532 ``(True, state_name)`` to retry the state in place, or
1533 ``(False, None)`` when the retry budget is exhausted (caller falls
1534 through to normal verdict routing).
1535 """
1536 record = self._api_error_retries.setdefault(state_name, {"retries": 0, "total_wait": 0.0})
1537 if record["retries"] >= _DEFAULT_API_ERROR_RETRIES:
1538 self._api_error_retries.pop(state_name, None)
1539 self._emit("api_error_exhausted", {"state": state_name, "retries": record["retries"]})
1540 return False, None
1541 record["retries"] += 1
1542 slept = self._interruptible_sleep(_DEFAULT_API_ERROR_BACKOFF)
1543 record["total_wait"] += slept
1544 self._emit(
1545 "api_error_retry",
1546 {
1547 "state": state_name,
1548 "attempt": record["retries"],
1549 "backoff": _DEFAULT_API_ERROR_BACKOFF,
1550 },
1551 )
1552 return True, state_name
1554 def _finish(self, terminated_by: str, error: str | None = None) -> ExecutionResult:
1555 """Finalize execution and return result."""
1556 self._emit(
1557 "loop_complete",
1558 {
1559 "final_state": self.current_state,
1560 "iterations": self.iteration,
1561 "terminated_by": terminated_by,
1562 },
1563 )
1565 return ExecutionResult(
1566 final_state=self.current_state,
1567 iterations=self.iteration,
1568 terminated_by=terminated_by,
1569 duration_ms=_now_ms() - self.start_time_ms + self.elapsed_offset_ms,
1570 captured=self.captured,
1571 error=error,
1572 )
1574 def _handle_handoff(self, signal: DetectedSignal) -> ExecutionResult:
1575 """Handle a detected handoff signal.
1577 Emits a handoff_detected event and optionally invokes the handoff handler.
1579 Args:
1580 signal: The detected handoff signal
1582 Returns:
1583 ExecutionResult with handoff information
1584 """
1585 self._emit(
1586 "handoff_detected",
1587 {
1588 "state": self.current_state,
1589 "iteration": self.iteration,
1590 "continuation": signal.payload,
1591 },
1592 )
1594 # Invoke handler if configured
1595 if self.handoff_handler:
1596 result = self.handoff_handler.handle(self.fsm.name, signal.payload)
1597 if result.spawned_process is not None:
1598 self._emit(
1599 "handoff_spawned",
1600 {
1601 "pid": result.spawned_process.pid,
1602 "state": self.current_state,
1603 },
1604 )
1606 return ExecutionResult(
1607 final_state=self.current_state,
1608 iterations=self.iteration,
1609 terminated_by="handoff",
1610 duration_ms=_now_ms() - self.start_time_ms + self.elapsed_offset_ms,
1611 captured=self.captured,
1612 handoff=True,
1613 continuation_prompt=signal.payload,
1614 )