Coverage for little_loops / fsm / executor.py: 11%
753 statements
« prev ^ index » next coverage.py v7.12.0, created at 2026-06-04 12:21 -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_blind_comparator,
28 evaluate_exit_code,
29 evaluate_llm_structured,
30 evaluate_mcp_result,
31)
32from little_loops.fsm.handoff_handler import HandoffHandler
33from little_loops.fsm.interpolation import (
34 InterpolationContext,
35 InterpolationError,
36 interpolate,
37 interpolate_dict,
38)
39from little_loops.fsm.rate_limit_circuit import RateLimitCircuit
40from little_loops.fsm.runners import (
41 ActionRunner,
42 DefaultActionRunner,
43 SimulationActionRunner, # noqa: F401 — re-exported for backward compatibility
44 _now_ms,
45)
46from little_loops.fsm.schema import FSMLoop, StateConfig
47from little_loops.fsm.signal_detector import DetectedSignal, SignalDetector
48from little_loops.fsm.stall_detector import Stall, StallDetector
49from little_loops.fsm.types import ActionResult, Evaluator, EventCallback, ExecutionResult
50from little_loops.issue_lifecycle import FailureType, classify_failure
51from little_loops.session_log import get_current_session_jsonl
52from little_loops.subprocess_utils import (
53 UsageCallback,
54 run_claude_command,
55)
57# Maximum number of per-state rate-limit retries before emitting rate_limit_exhausted.
58_DEFAULT_RATE_LIMIT_RETRIES: int = 3
59# Base backoff in seconds; actual sleep = base * 2^(attempt-1) + uniform(0, base).
60_DEFAULT_RATE_LIMIT_BACKOFF_BASE: int = 30
61# Total wall-clock budget (seconds) across short + long tiers before routing to
62# on_rate_limit_exhausted. Mirrors RateLimitsConfig.max_wait_seconds default (6h).
63_DEFAULT_RATE_LIMIT_MAX_WAIT_SECONDS: int = 21600
64# Long-wait tier ladder (seconds), walked once the short-tier budget is spent.
65# Mirrors RateLimitsConfig.long_wait_ladder default.
66_DEFAULT_RATE_LIMIT_LONG_WAIT_LADDER: list[int] = [300, 900, 1800, 3600]
67# Event name emitted when the stall detector fires on N consecutive
68# identical (state, exit_code, verdict) transitions. See FEAT-1637.
69STALL_DETECTED_EVENT: str = "stall_detected"
70# Event name emitted when rate-limit retries are exhausted.
71RATE_LIMIT_EXHAUSTED_EVENT: str = "rate_limit_exhausted"
72# Event name emitted when consecutive rate-limit exhaustions reach the storm threshold.
73RATE_LIMIT_STORM_EVENT: str = "rate_limit_storm"
74# Event name emitted every ~60s during a long-wait rate-limit sleep so UIs can show live progress.
75RATE_LIMIT_WAITING_EVENT: str = "rate_limit_waiting"
76# Interval (seconds) between rate_limit_waiting heartbeat emissions during long-wait sleeps.
77_RATE_LIMIT_HEARTBEAT_INTERVAL: float = 60.0
78# Number of consecutive rate_limit_exhausted events that constitute a storm.
79_RATE_LIMIT_STORM_THRESHOLD: int = 3
80# Progressive throttle defaults: calls 1..normal_max pass through, at warn_max emit warning,
81# at hard_max route to on_throttle_hard, beyond hard_max hard-stop.
82_DEFAULT_THROTTLE_NORMAL_MAX: int = 3
83_DEFAULT_THROTTLE_WARN_MAX: int = 8
84_DEFAULT_THROTTLE_HARD_MAX: int = 12
85# Event names for progressive tool-call throttling within a single state visit.
86THROTTLE_WARN_EVENT: str = "throttle_warn"
87THROTTLE_HARD_EVENT: str = "throttle_hard"
88THROTTLE_STOP_EVENT: str = "throttle_stop"
89# Action types that consume LLM quota and are gated by the shared circuit breaker.
90# `_action_mode()` collapses both to "prompt"; the frozenset documents intent.
91LLM_ACTION_TYPES: frozenset[str] = frozenset({"slash_command", "prompt"})
92# Maximum per-state API server error retries before falling through to normal routing.
93_DEFAULT_API_ERROR_RETRIES: int = 2
94# Flat backoff in seconds between API server error retries (no exponential ladder).
95_DEFAULT_API_ERROR_BACKOFF: int = 30
98def _iso_now() -> str:
99 """Get current time as ISO 8601 string."""
100 return datetime.now(UTC).isoformat()
103@dataclass
104class RouteContext:
105 """Context passed to before_route / after_route interceptors."""
107 state_name: str
108 state: StateConfig
109 verdict: str
110 action_result: ActionResult | None
111 eval_result: EvaluationResult | None
112 ctx: InterpolationContext
113 iteration: int
116@dataclass
117class RouteDecision:
118 """Returned by before_route to redirect or veto a routing transition.
120 Return semantics for before_route:
121 None (implicit) → passthrough, routing proceeds normally
122 RouteDecision("state") → redirect, bypass _route() and use "state" directly
123 RouteDecision(None) → veto, _execute_state() returns None → _finish("error")
124 """
126 next_state: str | None # str → redirect; None → veto
129class FSMExecutor:
130 """Execute an FSM loop.
132 The executor runs an FSM from its initial state until:
133 - A terminal state is reached
134 - max_iterations is exceeded
135 - A timeout occurs
136 - A shutdown signal is received
137 - An unrecoverable error occurs
139 Events are emitted via the callback for observability.
140 """
142 def __init__(
143 self,
144 fsm: FSMLoop,
145 event_callback: EventCallback | None = None,
146 action_runner: ActionRunner | None = None,
147 signal_detector: SignalDetector | None = None,
148 handoff_handler: HandoffHandler | None = None,
149 loops_dir: Path | None = None,
150 circuit: RateLimitCircuit | None = None,
151 ):
152 """Initialize the executor.
154 Args:
155 fsm: The FSM loop to execute
156 event_callback: Optional callback for events
157 action_runner: Optional custom action runner (for testing)
158 signal_detector: Optional signal detector for output parsing
159 handoff_handler: Optional handler for handoff signals
160 loops_dir: Base directory for resolving sub-loop references
161 circuit: Optional shared rate-limit circuit breaker for 429 coordination
162 """
163 self.fsm = fsm
164 self.event_callback = event_callback or (lambda _: None)
165 self.action_runner: ActionRunner = action_runner or DefaultActionRunner()
166 self.signal_detector = signal_detector
167 self.handoff_handler = handoff_handler
168 self.loops_dir = loops_dir
169 self._circuit = circuit
171 # Runtime state
172 self.current_state = fsm.initial
173 self.iteration = 0
174 self.captured: dict[str, dict[str, Any]] = {}
175 self.messages: list[str] = []
176 self.prev_result: dict[str, Any] | None = None
177 self.started_at = ""
178 self.start_time_ms = 0
179 self.elapsed_offset_ms = (
180 0 # milliseconds from segments before current run (set by PersistentExecutor on resume)
181 )
183 # Shutdown flag for graceful signal handling
184 self._shutdown_requested = False
186 # Currently running MCP subprocess (set by _run_subprocess, cleared in finally).
187 # Enables external shutdown code to kill the process on SIGTERM.
188 self._current_process: subprocess.Popen[str] | None = None
190 # Pending handoff signal (set by _run_action, checked by main loop)
191 self._pending_handoff: DetectedSignal | None = None
193 # Pending error payload from FATAL_ERROR signal (set by _run_action, checked by main loop)
194 self._pending_error: str | None = None
196 # Per-state retry tracking for max_retries support.
197 # _retry_counts[state_name] = number of consecutive re-entries into that state.
198 # Incremented each time we enter the same state as the previous iteration.
199 # Reset when a different state is entered, or after retry exhaustion.
200 self._retry_counts: dict[str, int] = {}
201 # State entered in the previous iteration (None on first iteration or after resume).
202 self._prev_state: str | None = None
204 # BUG-1226: true between emitting a `route` event and entering the
205 # target state. Gates the "flush one pending shell state on timeout"
206 # behavior so we only flush when there is actually a pending state.
207 self._just_routed: bool = False
209 # ENH-1631: true once the on_max_iterations summary state has been
210 # dispatched. Prevents the cap guard from re-triggering before the
211 # summary state completes.
212 self._summary_state_executed: bool = False
214 # FEAT-1822: Per-item A/B comparison results accumulated during baseline
215 # execution. Populated by _execute_with_baseline(), written to ab.json
216 # by _finish().
217 self._ab_results: list[dict[str, Any]] = []
218 self._ab_item_index: int = 0
220 # Per-state rate-limit retry tracking (parallel to _retry_counts).
221 # _rate_limit_retries[state_name] = dict-of-record:
222 # {
223 # "short_retries": int, # attempts in short-burst tier
224 # "long_retries": int, # attempts in long-wait tier
225 # "total_wait_seconds": float,
226 # "first_seen_at": float | None, # epoch timestamp of first 429
227 # }
228 # Incremented inside _handle_rate_limit on each detected rate-limit response.
229 # Reset when the state completes without a rate-limit, or after exhaustion.
230 self._rate_limit_retries: dict[str, dict[str, Any]] = {}
232 # Consecutive rate_limit_exhausted emissions across all states. Reset on any
233 # successful non-rate-limited state transition. When this reaches
234 # _RATE_LIMIT_STORM_THRESHOLD, a RATE_LIMIT_STORM event is emitted.
235 self._consecutive_rate_limit_exhaustions: int = 0
237 # Per-state API server error retry tracking (parallel to _rate_limit_retries).
238 # _api_error_retries[state_name] = {"retries": int, "total_wait": float}
239 # Reset when the state completes without a server error, or after exhaustion.
240 self._api_error_retries: dict[str, dict[str, Any]] = {}
242 # Per-state tool-call throttle counter. Counts successive action executions within
243 # a single continuous state visit. Reset on state exit; NOT serialized to LoopState
244 # (throttle counts measure instantaneous visit-level activity, not cumulative retries).
245 self._throttle_counts: dict[str, int] = {}
247 # Per-edge revisit counter for cycle detection.
248 # _edge_revisit_counts["from_state->to_state"] = number of times that edge has fired.
249 # When any edge exceeds max_edge_revisits, the loop terminates with cycle_detected.
250 self._edge_revisit_counts: dict[str, int] = {}
252 # Stall detector for repeated (state, exit_code, verdict) triples.
253 # Enabled via fsm.circuit.repeated_failure (FEAT-1637); None when not configured.
254 self._stall_detector: StallDetector | None = None
255 if fsm.circuit is not None and fsm.circuit.repeated_failure is not None:
256 self._stall_detector = StallDetector(window=fsm.circuit.repeated_failure.window)
257 # Set by _execute_state when the detector fires with on_repeated_failure="abort";
258 # checked by run() to terminate via _finish("stall_detected", ...).
259 self._pending_stall_abort: Stall | None = None
261 # Nesting depth for sub-loop event forwarding (0 = top-level, 1+ = sub-loop).
262 # Set by the parent executor when constructing child executors.
263 self._depth: int = 0
265 # Extension hook registries — populated by wire_extensions()
266 self._contributed_actions: dict[str, ActionRunner] = {}
267 self._contributed_evaluators: dict[str, Evaluator] = {}
268 self._interceptors: list[Any] = []
270 def request_shutdown(self) -> None:
271 """Request graceful shutdown of the executor.
273 Sets a flag that will be checked at the start of each iteration,
274 allowing the loop to exit cleanly after the current state completes.
275 """
276 self._shutdown_requested = True
278 def run(self) -> ExecutionResult:
279 """Execute the FSM until terminal state or limits reached.
281 Returns:
282 ExecutionResult with final state and execution metadata
283 """
284 self.started_at = _iso_now()
285 self.start_time_ms = _now_ms()
287 self._emit("loop_start", {"loop": self.fsm.name})
289 try:
290 while True:
291 # Check shutdown request (signal handling)
292 if self._shutdown_requested:
293 return self._finish("signal")
295 # Check iteration limit
296 if self.iteration >= self.fsm.max_iterations:
297 if self.fsm.on_max_iterations is not None and not self._summary_state_executed:
298 self._emit(
299 "max_iterations_summary",
300 {
301 "summary_state": self.fsm.on_max_iterations,
302 "iterations": self.iteration,
303 },
304 )
305 self._summary_state_executed = True
306 self.current_state = self.fsm.on_max_iterations
307 # Fall through — let the summary state run in this iteration.
308 # (do not `continue`; that would re-trigger the cap check
309 # before the state executes since self.iteration is unchanged.)
310 else:
311 return self._finish("max_iterations")
313 # Check timeout
314 if self.fsm.timeout:
315 elapsed = _now_ms() - self.start_time_ms + self.elapsed_offset_ms
316 if elapsed > self.fsm.timeout * 1000:
317 # BUG-1226: if timeout fires in the race window between
318 # a `route` event and `state_enter`, flush one pending
319 # shell-action state before honoring the timeout so its
320 # side effect (e.g. copying a handshake flag) is not
321 # silently lost. Bounded to shell actions — slash
322 # commands and sub-loops would violate the timeout
323 # budget. Single-step: no cascade.
324 if self._just_routed:
325 pending = self.fsm.states.get(self.current_state)
326 if (
327 pending is not None
328 and not pending.terminal
329 and pending.loop is None
330 and pending.action is not None
331 and self._action_mode(pending) == "shell"
332 ):
333 self._flush_pending_shell_state(pending)
334 return self._finish("timeout")
336 # Get current state config
337 state_config = self.fsm.states[self.current_state]
339 # Update per-state retry tracking based on transition from previous iteration.
340 # If re-entering the same state consecutively, increment retry count.
341 # If entering a different state, clear the previous state's retry count.
342 if self._prev_state is not None:
343 if self.current_state == self._prev_state:
344 self._retry_counts[self.current_state] = (
345 self._retry_counts.get(self.current_state, 0) + 1
346 )
347 else:
348 self._retry_counts.pop(self._prev_state, None)
349 self._throttle_counts.pop(self._prev_state, None)
351 # Check terminal
352 if state_config.terminal:
353 # Handle maintain mode - restart loop instead of terminating
354 if self.fsm.maintain:
355 self.iteration += 1
356 maintain_target = state_config.on_maintain or self.fsm.initial
357 self._emit(
358 "route",
359 {
360 "from": self.current_state,
361 "to": maintain_target,
362 "reason": "maintain",
363 },
364 )
365 self._prev_state = self.current_state
366 self.current_state = maintain_target
367 self._just_routed = True
368 continue
369 # ENH-1631: if we arrived here via the on_max_iterations summary
370 # state, preserve terminated_by="max_iterations" so audit tooling
371 # and PersistentExecutor see "interrupted" rather than "completed".
372 if self._summary_state_executed:
373 return self._finish("max_iterations")
374 return self._finish("terminal")
376 # Check per-state retry limit. If the consecutive re-entry count exceeds
377 # max_retries, skip execution and route to on_retry_exhausted instead.
378 if state_config.max_retries is not None:
379 retry_count = self._retry_counts.get(self.current_state, 0)
380 if retry_count > state_config.max_retries:
381 # on_retry_exhausted is guaranteed non-None by validation when
382 # max_retries is set, but we fall back to an error if misconfigured.
383 exhausted_state: str = state_config.on_retry_exhausted or ""
384 if not exhausted_state:
385 return self._finish(
386 "error",
387 error=f"State '{self.current_state}' exceeded max_retries "
388 "but on_retry_exhausted is not set",
389 )
390 self._emit(
391 "retry_exhausted",
392 {
393 "state": self.current_state,
394 "retries": retry_count,
395 "next": exhausted_state,
396 },
397 )
398 self._retry_counts.pop(self.current_state, None)
399 self._prev_state = self.current_state
400 self.current_state = exhausted_state
401 continue
403 self.iteration += 1
404 self._just_routed = False
405 self._emit(
406 "state_enter",
407 {
408 "state": self.current_state,
409 "iteration": self.iteration,
410 },
411 )
413 # Execute state
414 next_state = self._execute_state(state_config)
416 # Check for pending error signal (FATAL_ERROR)
417 if self._pending_error is not None:
418 return self._finish("error", error=self._pending_error)
420 # Check for pending stall abort (FEAT-1637). The detector
421 # fired with on_repeated_failure="abort" inside _execute_state;
422 # terminate cleanly via _finish (mirrors the cycle_detected
423 # guard below at lines 397-416).
424 if self._pending_stall_abort is not None:
425 stall = self._pending_stall_abort
426 s_state, s_exit, s_verdict = stall.triple
427 return self._finish(
428 "stall_detected",
429 error=(
430 f"Stall detected: state '{s_state}' produced "
431 f"(exit_code={s_exit}, verdict='{s_verdict}') "
432 f"for {stall.count} consecutive iterations"
433 ),
434 )
436 # Check for pending handoff signal
437 if self._pending_handoff:
438 return self._handle_handoff(self._pending_handoff)
440 # Handle maintain mode
441 if next_state is None and self.fsm.maintain:
442 next_state = state_config.on_maintain or self.fsm.initial
444 # SIGKILL in _execute_state sets shutdown flag and returns None
445 if next_state is None and self._shutdown_requested:
446 return self._finish("signal")
448 if next_state is None:
449 return self._finish("error", error="No valid transition")
451 # At this point next_state is guaranteed to be str
452 resolved_next: str = next_state
454 self._emit(
455 "route",
456 {
457 "from": self.current_state,
458 "to": resolved_next,
459 },
460 )
462 # Per-edge revisit tracking for cycle detection.
463 edge_key = f"{self.current_state}->{resolved_next}"
464 self._edge_revisit_counts[edge_key] = self._edge_revisit_counts.get(edge_key, 0) + 1
465 if self._edge_revisit_counts[edge_key] > self.fsm.max_edge_revisits:
466 self._emit(
467 "cycle_detected",
468 {
469 "edge": edge_key,
470 "from": self.current_state,
471 "to": resolved_next,
472 "count": self._edge_revisit_counts[edge_key],
473 "max": self.fsm.max_edge_revisits,
474 },
475 )
476 return self._finish(
477 "cycle_detected",
478 error=f"Cycle detected: edge {edge_key} traversed "
479 f"{self._edge_revisit_counts[edge_key]} times "
480 f"(limit: {self.fsm.max_edge_revisits})",
481 )
483 self._prev_state = self.current_state
484 self.current_state = resolved_next
485 self._just_routed = True
487 # Interruptible backoff sleep between iterations
488 if self.fsm.backoff and self.fsm.backoff > 0:
489 deadline = time.time() + self.fsm.backoff
490 while time.time() < deadline:
491 if self._shutdown_requested:
492 break
493 time.sleep(min(0.1, deadline - time.time()))
495 except InterpolationError as exc:
496 return self._finish(
497 "error",
498 error=(
499 f"Missing context variable in state '{self.current_state}': {exc}. "
500 f"Run with: ll-loop run {self.fsm.name} --context KEY=VALUE"
501 ),
502 )
503 except Exception as exc:
504 return self._finish("error", error=str(exc))
506 def _execute_sub_loop(self, state: StateConfig, ctx: InterpolationContext) -> str | None:
507 """Execute a sub-loop state by loading and running a child FSM.
509 Args:
510 state: The state configuration with loop field set
511 ctx: Interpolation context for routing
513 Returns:
514 Next state name based on child loop verdict, or None
515 """
516 from little_loops.cli.loop._helpers import resolve_loop_path
517 from little_loops.fsm.validation import load_and_validate
519 assert state.loop is not None # guarded by caller
520 loop_name = interpolate(state.loop, ctx)
521 loop_path = resolve_loop_path(loop_name, self.loops_dir or Path(".loops"))
522 child_fsm, _ = load_and_validate(loop_path)
524 # Bind child context: explicit with: bindings take precedence over legacy passthrough
525 if state.with_:
526 from little_loops.fsm.interpolation import interpolate_dict
528 resolved = interpolate_dict(state.with_, ctx)
529 # Apply declared defaults for unbound optional parameters
530 for param_name, param_spec in child_fsm.parameters.items():
531 if (
532 param_name not in resolved
533 and not param_spec.required
534 and param_spec.default is not None
535 ):
536 resolved[param_name] = param_spec.default
537 # Runtime check: required parameters must be present after interpolation
538 for param_name, param_spec in child_fsm.parameters.items():
539 if param_spec.required and param_name not in resolved:
540 raise ValueError(
541 f"Sub-loop '{state.loop}' requires parameter '{param_name}' "
542 f"but it is not bound in 'with'"
543 )
544 # Merge: child's own context block provides base; with: bindings override
545 child_fsm.context = {**child_fsm.context, **resolved}
546 elif state.context_passthrough:
547 # Extract .output strings from capture result dicts so ${context.key} resolves
548 # to the plain output string (e.g. "ENH-123") rather than the full capture object.
549 captured_as_context = {
550 k: v["output"] if isinstance(v, dict) and "exit_code" in v else v
551 for k, v in self.captured.items()
552 }
553 child_fsm.context = {**self.fsm.context, **captured_as_context, **child_fsm.context}
555 depth = self._depth + 1
556 child_events: list[dict] = []
558 def _sub_event_callback(event: dict) -> None:
559 child_events.append(event)
560 # Only inject depth if not already set by a deeper nested sub-loop
561 if "depth" not in event:
562 self.event_callback({**event, "depth": depth})
563 else:
564 self.event_callback(event)
566 child_executor = FSMExecutor(
567 child_fsm,
568 action_runner=self.action_runner,
569 loops_dir=self.loops_dir,
570 event_callback=_sub_event_callback,
571 circuit=self._circuit,
572 )
573 child_executor._depth = depth # propagate depth for further nesting
575 # Clamp child timeout to parent's remaining wall-clock budget so a slow sub-loop
576 # can't silently consume the parent's deadline with no recourse for the parent FSM.
577 if self.fsm.timeout:
578 elapsed_ms = _now_ms() - self.start_time_ms + self.elapsed_offset_ms
579 remaining_s = max(1, int((self.fsm.timeout * 1000 - elapsed_ms) // 1000))
580 if child_fsm.timeout is None or child_fsm.timeout > remaining_s:
581 child_fsm.timeout = remaining_s
583 child_result = child_executor.run()
585 # Capture child event stream as a JSON-lines string if the state declares a capture key
586 if state.capture and child_events:
587 import json as _json
589 self.captured[state.capture] = {
590 "output": "\n".join(_json.dumps(e) for e in child_events),
591 "exit_code": None,
592 }
594 # Merge child captures back into parent under the state name
595 if (state.context_passthrough or state.with_) and child_executor.captured:
596 self.captured[self.current_state] = child_executor.captured
598 # Route based on child termination reason and terminal state name
599 if child_result.terminated_by == "terminal":
600 if child_result.final_state == "done":
601 return interpolate(state.on_yes, ctx) if state.on_yes else None
602 else:
603 # Reached a non-done terminal (e.g. "failed") → failure
604 return interpolate(state.on_no, ctx) if state.on_no else None
605 elif child_result.terminated_by == "error":
606 # Runtime child failure (not a YAML load error)
607 if state.on_error:
608 return interpolate(state.on_error, ctx)
609 return interpolate(state.on_no, ctx) if state.on_no else None
610 else:
611 # max_iterations, timeout, signal — all are failure
612 return interpolate(state.on_no, ctx) if state.on_no else None
614 def _execute_learning_state(self, state: StateConfig, ctx: InterpolationContext) -> str | None:
615 """Execute a FEAT-1283 ``type: learning`` state.
617 Resolves the target list at runtime: if ``learning.targets_csv`` is set,
618 it is interpolated and CSV-split; otherwise ``learning.targets`` is used
619 directly. The retry limit is resolved similarly: ``learning.max_retries_expr``
620 (if set) is interpolated and int()-cast; otherwise ``learning.max_retries``
621 (default 2) is used.
623 For each target:
624 1. Look up its record in the learning-tests registry (ENH-1282).
625 2. If proven → continue.
626 3. If refuted → emit ``learning_target_refuted`` + ``learning_blocked``
627 and route to ``on_blocked`` (preferred) or ``on_no``.
628 4. If missing or stale → emit ``learning_target_stale`` and invoke
629 ``/ll:explore-api <target>`` via the executor's action_runner;
630 re-check the registry; repeat up to ``max_retries`` times before
631 emitting ``learning_blocked`` (reason ``retries_exhausted``) and
632 routing to ``on_blocked``/``on_no``.
634 When every target ends up proven, emit ``learning_complete`` and route
635 to ``on_yes``. Returns the resolved next-state name (or ``None`` when no
636 route is configured for the terminal verdict, mirroring ``_route``).
637 """
638 from little_loops.learning_tests import check_learning_test
640 assert state.learning is not None # guarded by caller
642 # Resolve target list at runtime (ENH-1741: targets_csv support).
643 if state.learning.targets_csv is not None:
644 raw_csv = interpolate(state.learning.targets_csv, ctx)
645 targets = [t.strip() for t in raw_csv.split(",") if t.strip()]
646 else:
647 targets = list(state.learning.targets)
649 # Resolve retry limit at runtime (ENH-1741: max_retries_expr support).
650 if state.learning.max_retries_expr is not None:
651 max_retries = int(interpolate(state.learning.max_retries_expr, ctx))
652 else:
653 max_retries = state.learning.max_retries
655 def _blocked_target(reason: str, target: str) -> str | None:
656 self._emit(
657 "learning_blocked",
658 {"state": self.current_state, "target": target, "reason": reason},
659 )
660 route = state.on_blocked or state.on_no
661 return interpolate(route, ctx) if route else None
663 for target in targets:
664 record = check_learning_test(target)
666 attempts = 0
667 while record is None or record.status == "stale":
668 if attempts >= max_retries:
669 return _blocked_target("retries_exhausted", target)
671 if record is None:
672 self._emit(
673 "learning_target_stale",
674 {"state": self.current_state, "target": target, "cause": "missing"},
675 )
676 else:
677 self._emit(
678 "learning_target_stale",
679 {"state": self.current_state, "target": target, "cause": "stale"},
680 )
682 self._emit(
683 "learning_explore_invoked",
684 {"state": self.current_state, "target": target, "attempt": attempts + 1},
685 )
686 self._run_action(f"/ll:explore-api {target}", state, ctx)
687 attempts += 1
688 record = check_learning_test(target)
690 if record.status == "refuted":
691 self._emit(
692 "learning_target_refuted",
693 {"state": self.current_state, "target": target},
694 )
695 return _blocked_target("refuted", target)
697 self._emit(
698 "learning_target_proven",
699 {"state": self.current_state, "target": target},
700 )
702 self._emit(
703 "learning_complete",
704 {"state": self.current_state, "targets": targets},
705 )
706 return interpolate(state.on_yes, ctx) if state.on_yes else None
708 def _compute_progress_fingerprint(self, ctx: InterpolationContext) -> tuple[object, ...] | None:
709 """Return an (mtime, size) fingerprint for configured progress_paths, or None.
711 Called just before stall_detector.record() so that file changes made by
712 intermediate next:-only states are visible to the detector. Returns None
713 when no progress_paths are configured (preserves existing semantics).
715 Paths listed in exclude_paths (BUG-1767) are resolved and removed before
716 building the fingerprint tuple so that a loop's internal bookkeeping files
717 cannot reset the stall window on every cycle.
718 """
719 if self.fsm.circuit is None or self.fsm.circuit.repeated_failure is None:
720 return None
721 rf = self.fsm.circuit.repeated_failure
722 paths = rf.progress_paths
723 if not paths:
724 return None
726 excluded: set[str] = set()
727 for raw_excl in rf.exclude_paths:
728 try:
729 excluded.add(interpolate(raw_excl, ctx))
730 except Exception:
731 pass
733 entries: list[tuple[float, int]] = []
734 for raw_path in paths:
735 try:
736 resolved = interpolate(raw_path, ctx)
737 except Exception:
738 continue
739 if resolved in excluded:
740 continue
741 p = Path(resolved)
742 if p.exists():
743 st = p.stat()
744 entries.append((st.st_mtime, st.st_size))
745 else:
746 entries.append((0.0, 0))
747 return tuple(entries) if entries else None
749 def _check_throttle(self, state: StateConfig, state_name: str) -> str | None:
750 """Increment the per-state tool-call counter and enforce throttle thresholds.
752 Called after every action execution within a state visit. Returns the forced
753 next-state name when the hard threshold is reached, or None when execution
754 should continue normally (warn events are emitted but do not redirect).
756 Sets self._pending_error and returns "__STOP__" when the call count exceeds
757 hard_max with no on_throttle_hard target — the caller must propagate this as
758 a None return from _execute_state so the main loop detects _pending_error.
759 """
760 count = self._throttle_counts.get(state_name, 0) + 1
761 self._throttle_counts[state_name] = count
763 throttle = state.throttle
764 normal_max = (
765 throttle.normal_max
766 if (throttle and throttle.normal_max is not None)
767 else _DEFAULT_THROTTLE_NORMAL_MAX
768 )
769 warn_max = (
770 throttle.warn_max
771 if (throttle and throttle.warn_max is not None)
772 else _DEFAULT_THROTTLE_WARN_MAX
773 )
774 hard_max = (
775 throttle.hard_max
776 if (throttle and throttle.hard_max is not None)
777 else _DEFAULT_THROTTLE_HARD_MAX
778 )
780 if count == warn_max:
781 self._emit(
782 THROTTLE_WARN_EVENT,
783 {
784 "state": state_name,
785 "count": count,
786 "normal_max": normal_max,
787 "warn_max": warn_max,
788 "hard_max": hard_max,
789 },
790 )
792 # States with type="learning" (FEAT-1283) are exempt from hard_max — they
793 # legitimately make N calls per visit (one per unproven target).
794 if state.type == "learning":
795 return None
797 if count == hard_max:
798 next_target = state.on_throttle_hard or state.on_error
799 self._emit(
800 THROTTLE_HARD_EVENT,
801 {"state": state_name, "count": count, "hard_max": hard_max, "next": next_target},
802 )
803 return next_target
805 if count > hard_max:
806 self._emit(
807 THROTTLE_STOP_EVENT,
808 {"state": state_name, "count": count, "hard_max": hard_max},
809 )
810 self._pending_error = (
811 f"Throttle stop: state '{state_name}' exceeded hard_max={hard_max} "
812 "tool calls with no on_throttle_hard target"
813 )
814 return "__STOP__"
816 return None
818 def _execute_state(self, state: StateConfig) -> str | None:
819 """Execute a single state and return next state name.
821 Args:
822 state: The state configuration to execute
824 Returns:
825 Next state name, or None if no valid transition
826 """
827 # Build interpolation context
828 ctx = self._build_context()
830 # Dispatch to sub-loop handler if this is a sub-loop state
831 if state.loop is not None:
832 try:
833 return self._execute_sub_loop(state, ctx)
834 except (FileNotFoundError, ValueError):
835 if state.on_error:
836 return interpolate(state.on_error, ctx)
837 raise
839 # FEAT-1283: dispatch to learning-state handler when both type="learning"
840 # AND a LearningConfig is present. The bare `type="learning"` marker
841 # (used pre-FEAT-1283 only as a throttle hard_max exemption hint, see
842 # ThrottleConfig docstring) falls through to normal action execution.
843 if state.type == "learning" and state.learning is not None:
844 return self._execute_learning_state(state, ctx)
846 # Handle unconditional transition
847 if state.next:
848 if state.action:
849 self._maybe_wait_for_circuit(state)
850 result, routed = self._run_action_or_route(state, ctx)
851 if routed is not None:
852 return routed
853 throttle_next = self._check_throttle(state, self.current_state)
854 if throttle_next == "__STOP__":
855 return None
856 if throttle_next is not None:
857 return throttle_next
858 assert result is not None
859 self.prev_result = {
860 "output": result.output,
861 "exit_code": result.exit_code,
862 "state": self.current_state,
863 }
864 if result.exit_code is not None and result.exit_code < 0:
865 # Process killed by signal — do not silently advance via next
866 if state.on_error:
867 return interpolate(state.on_error, ctx)
868 self.request_shutdown()
869 return None
870 # Non-zero exit: if on_error is defined, treat next as success path only
871 if result.exit_code != 0 and state.on_error:
872 error_target = interpolate(state.on_error, ctx)
873 if (
874 state.retryable_exit_codes is not None
875 and result.exit_code is not None
876 and result.exit_code not in state.retryable_exit_codes
877 ):
878 # Non-retryable exit code — bypass retry, route to
879 # on_retry_exhausted (if set) or on_error directly.
880 if state.on_retry_exhausted:
881 return state.on_retry_exhausted
882 return error_target
883 return error_target
884 return interpolate(state.next, ctx)
886 # Execute action if present
887 action_result = None
888 if state.action and self._action_mode(state) != "contract":
889 self._maybe_wait_for_circuit(state)
890 baseline_cfg = ctx.context.get("_baseline")
891 if baseline_cfg and isinstance(baseline_cfg, dict) and baseline_cfg.get("enabled"):
892 # Baseline mode: spawn parallel arms, harness drives routing
893 action_result, routed = self._execute_with_baseline(state, ctx, baseline_cfg)
894 if routed is not None:
895 return routed
896 else:
897 action_result, routed = self._run_action_or_route(state, ctx)
898 if routed is not None:
899 return routed
900 throttle_next = self._check_throttle(state, self.current_state)
901 if throttle_next == "__STOP__":
902 return None
903 if throttle_next is not None:
904 return throttle_next
906 # Evaluate
907 eval_result = self._evaluate(state, action_result, ctx)
908 self.prev_result = {
909 "output": action_result.output if action_result else "",
910 "exit_code": action_result.exit_code if action_result else 0,
911 "state": self.current_state,
912 }
914 # Update context with result for routing interpolation
915 if eval_result:
916 ctx.result = {
917 "verdict": eval_result.verdict,
918 "details": eval_result.details,
919 }
921 # Route based on verdict
922 verdict = eval_result.verdict if eval_result else "yes"
924 # Stall detection (FEAT-1637). Record this transition's triple and
925 # check whether the last `window` triples are identical. On stall,
926 # either abort (set _pending_stall_abort for run() to catch) or
927 # override next_state to the configured recovery target.
928 stall_route_target: str | None = None
929 if self._stall_detector is not None:
930 stall_exit_code = action_result.exit_code if action_result else 0
931 stall_fingerprint = self._compute_progress_fingerprint(ctx)
932 self._stall_detector.record(
933 self.current_state, stall_exit_code, verdict, stall_fingerprint
934 )
935 stall = self._stall_detector.check()
936 if stall is not None:
937 assert self.fsm.circuit is not None
938 assert self.fsm.circuit.repeated_failure is not None
939 cfg_action = self.fsm.circuit.repeated_failure.on_repeated_failure
940 self._emit(
941 STALL_DETECTED_EVENT,
942 {
943 "state": self.current_state,
944 "exit_code": stall_exit_code,
945 "verdict": verdict,
946 "consecutive": stall.count,
947 "action": "abort" if cfg_action == "abort" else f"route:{cfg_action}",
948 },
949 )
950 if cfg_action == "abort":
951 self._pending_stall_abort = stall
952 return None
953 # Route to recovery target; bypass _route() entirely so the
954 # eval verdict does not pull us elsewhere first.
955 stall_route_target = cfg_action
957 route_ctx = RouteContext(
958 state_name=self.current_state,
959 state=state,
960 verdict=verdict,
961 action_result=action_result,
962 eval_result=eval_result,
963 ctx=ctx,
964 iteration=self.iteration,
965 )
966 # 429 / rate-limit detection — runs before interceptors so an in-place retry
967 # returns early without dispatching to registered before_route hooks.
968 if action_result is not None:
969 _combined = (action_result.output or "") + "\n" + (action_result.stderr or "")
970 _failure_type, _reason = classify_failure(_combined, action_result.exit_code)
971 if _failure_type == FailureType.TRANSIENT and (
972 "rate limit" in _reason.lower() or "quota" in _reason.lower()
973 ):
974 _handled, _target = self._handle_rate_limit(state, route_ctx.state_name)
975 if _handled:
976 return _target
977 elif _failure_type == FailureType.TRANSIENT and "api server error" in _reason.lower():
978 _handled, _target = self._handle_api_error(state, route_ctx.state_name)
979 if _handled:
980 return _target
981 # exhausted — fall through to normal verdict routing
982 else:
983 # Not rate-limited or server-error: reset counters so future transients start fresh.
984 self._rate_limit_retries.pop(route_ctx.state_name, None)
985 self._consecutive_rate_limit_exhaustions = 0
986 self._api_error_retries.pop(route_ctx.state_name, None)
988 # Stall-route override: if the detector elected to route to a recovery
989 # state, honor it now (bypass interceptors and _route) so the
990 # configured target wins over the eval verdict.
991 if stall_route_target is not None:
992 return stall_route_target
994 # Non-retryable exit code filter for eval-based error routing. When a
995 # state uses retryable_exit_codes and the action fails with a code that
996 # is NOT retryable, bypass the normal error route (which may be a
997 # self-retry) and go directly to on_retry_exhausted (or on_error).
998 if (
999 verdict == "error"
1000 and state.retryable_exit_codes is not None
1001 and action_result is not None
1002 and action_result.exit_code is not None
1003 and action_result.exit_code not in state.retryable_exit_codes
1004 ):
1005 if state.on_retry_exhausted:
1006 return state.on_retry_exhausted
1007 if state.on_error:
1008 return interpolate(state.on_error, ctx)
1010 for interceptor in self._interceptors:
1011 if hasattr(interceptor, "before_route"):
1012 decision = interceptor.before_route(route_ctx)
1013 if isinstance(decision, RouteDecision):
1014 if decision.next_state is None:
1015 return None # veto
1016 return decision.next_state # redirect — bypass _route()
1017 next_state = self._route(state, verdict, ctx)
1018 for interceptor in self._interceptors:
1019 if hasattr(interceptor, "after_route"):
1020 interceptor.after_route(route_ctx)
1021 return next_state
1023 def _run_action(
1024 self,
1025 action_template: str,
1026 state: StateConfig,
1027 ctx: InterpolationContext,
1028 on_usage: UsageCallback | None = None,
1029 ) -> ActionResult:
1030 """Execute action and optionally capture result.
1032 Args:
1033 action_template: Action string (may contain variables)
1034 state: State configuration
1035 ctx: Interpolation context
1036 on_usage: Optional callback invoked with (input_tokens, output_tokens) on completion
1038 Returns:
1039 ActionResult with output and exit code
1040 """
1041 action = interpolate(action_template, ctx)
1042 action_mode = self._action_mode(state)
1044 self._emit("action_start", {"action": action, "is_prompt": action_mode == "prompt"})
1046 def _on_line(line: str) -> None:
1047 self._emit("action_output", {"line": line})
1049 if action_mode == "mcp_tool":
1050 # Direct MCP tool call — bypass action_runner entirely
1051 interpolated_params = interpolate_dict(state.params, ctx) if state.params else {}
1052 cmd = ["mcp-call", action, json.dumps(interpolated_params)]
1053 result = self._run_subprocess(
1054 cmd,
1055 timeout=state.timeout or self.fsm.default_timeout or 30,
1056 on_output_line=_on_line,
1057 )
1058 elif action_mode == "contributed":
1059 assert (
1060 state.action_type is not None
1061 ) # guaranteed by _action_mode returning "contributed"
1062 runner = self._contributed_actions[state.action_type]
1063 result = runner.run(
1064 action,
1065 timeout=state.timeout or self.fsm.default_timeout or 3600,
1066 is_slash_command=False,
1067 on_output_line=_on_line,
1068 on_usage=on_usage,
1069 )
1070 else:
1071 result = self.action_runner.run(
1072 action,
1073 timeout=state.timeout or self.fsm.default_timeout or 3600,
1074 is_slash_command=action_mode == "prompt",
1075 on_output_line=_on_line,
1076 agent=state.agent if action_mode == "prompt" else None,
1077 tools=state.tools if action_mode == "prompt" else None,
1078 on_usage=on_usage,
1079 )
1081 preview = result.output[-2000:].strip() if result.output else None
1082 payload: dict[str, Any] = {
1083 "exit_code": result.exit_code,
1084 "duration_ms": result.duration_ms,
1085 "output_preview": preview,
1086 "is_prompt": action_mode == "prompt",
1087 }
1088 if action_mode == "prompt":
1089 session_jsonl = get_current_session_jsonl()
1090 payload["session_jsonl"] = str(session_jsonl) if session_jsonl else None
1091 # Aggregate token usage from host-CLI invocations (prompt / slash_command only)
1092 if result.usage_events:
1093 total_input = sum(u.input_tokens for u in result.usage_events)
1094 total_output = sum(u.output_tokens for u in result.usage_events)
1095 total_cache_read = sum(u.cache_read_tokens for u in result.usage_events)
1096 total_cache_creation = sum(u.cache_creation_tokens for u in result.usage_events)
1097 model = result.usage_events[-1].model
1098 payload["input_tokens"] = total_input
1099 payload["output_tokens"] = total_output
1100 payload["cache_read_tokens"] = total_cache_read
1101 payload["cache_creation_tokens"] = total_cache_creation
1102 payload["model"] = model
1103 self._emit("action_complete", payload)
1105 # Capture if requested
1106 if state.capture:
1107 self.captured[state.capture] = {
1108 "output": result.output.rstrip("\n\r"),
1109 "stderr": result.stderr,
1110 "exit_code": result.exit_code,
1111 "duration_ms": result.duration_ms,
1112 }
1114 # Append to shared messages log if requested
1115 if state.append_to_messages:
1116 post_ctx = self._build_context()
1117 message = interpolate(state.append_to_messages, post_ctx)
1118 self.messages.append(message)
1119 self._emit(
1120 "messages_append",
1121 {"message": message, "state": self.current_state},
1122 )
1124 # Check for signals in output
1125 if self.signal_detector:
1126 signal = self.signal_detector.detect_first(result.output)
1127 if signal:
1128 if signal.signal_type == "handoff":
1129 self._pending_handoff = signal
1130 elif signal.signal_type == "error":
1131 self._pending_error = signal.payload
1132 elif signal.signal_type == "stop":
1133 self.request_shutdown()
1135 return result
1137 def _run_subprocess(
1138 self,
1139 cmd: list[str],
1140 timeout: int,
1141 on_output_line: Any | None = None,
1142 ) -> ActionResult:
1143 """Run a subprocess directly and return ActionResult.
1145 Follows the same Popen + stderr-drain-thread pattern as DefaultActionRunner.
1147 Args:
1148 cmd: Command and arguments to execute
1149 timeout: Timeout in seconds
1150 on_output_line: Optional callback for each stdout line
1152 Returns:
1153 ActionResult with output, stderr, exit_code, duration_ms
1154 """
1155 start = _now_ms()
1156 process = subprocess.Popen(
1157 cmd,
1158 stdout=subprocess.PIPE,
1159 stderr=subprocess.PIPE,
1160 text=True,
1161 )
1162 self._current_process = process
1163 output_chunks: list[str] = []
1164 stderr_chunks: list[str] = []
1166 def _drain_stderr() -> None:
1167 assert process.stderr is not None
1168 for line in process.stderr:
1169 stderr_chunks.append(line)
1171 stderr_thread = threading.Thread(target=_drain_stderr, daemon=True)
1172 stderr_thread.start()
1174 try:
1175 for line in process.stdout: # type: ignore[union-attr]
1176 output_chunks.append(line)
1177 if on_output_line:
1178 on_output_line(line.rstrip())
1179 process.wait(timeout=timeout)
1180 except subprocess.TimeoutExpired:
1181 process.kill()
1182 process.wait()
1183 stderr_thread.join(timeout=5)
1184 return ActionResult(
1185 output="".join(output_chunks),
1186 stderr="".join(stderr_chunks) or "MCP call timed out",
1187 exit_code=124,
1188 duration_ms=timeout * 1000,
1189 )
1190 finally:
1191 self._current_process = None
1192 stderr_thread.join(timeout=5)
1193 return ActionResult(
1194 output="".join(output_chunks),
1195 stderr="".join(stderr_chunks),
1196 exit_code=process.returncode,
1197 duration_ms=_now_ms() - start,
1198 )
1200 def _evaluate(
1201 self,
1202 state: StateConfig,
1203 action_result: ActionResult | None,
1204 ctx: InterpolationContext,
1205 ) -> EvaluationResult | None:
1206 """Evaluate action result.
1208 Args:
1209 state: State configuration
1210 action_result: Result from action execution (may be None)
1211 ctx: Interpolation context
1213 Returns:
1214 EvaluationResult, or None if no evaluation needed
1215 """
1216 if state.evaluate is None:
1217 # Default evaluation based on action type
1218 if action_result:
1219 action_mode = self._action_mode(state)
1221 if action_mode == "mcp_tool":
1222 # MCP tool call: use mcp_result evaluator
1223 result = evaluate_mcp_result(action_result.output, action_result.exit_code)
1224 elif action_mode == "prompt":
1225 # Slash command or prompt: use LLM evaluation
1226 if not self.fsm.llm.enabled:
1227 result = EvaluationResult(
1228 verdict="error",
1229 details={"error": "LLM evaluation disabled via --no-llm"},
1230 )
1231 else:
1232 result = evaluate_llm_structured(
1233 action_result.output,
1234 model=self.fsm.llm.model,
1235 max_tokens=self.fsm.llm.max_tokens,
1236 timeout=self.fsm.llm.timeout,
1237 )
1238 else:
1239 # Shell command: use exit code
1240 result = evaluate_exit_code(action_result.exit_code)
1242 self._emit(
1243 "evaluate",
1244 {
1245 "type": "default",
1246 "verdict": result.verdict,
1247 **result.details,
1248 },
1249 )
1250 return result
1251 return None
1253 # Explicit evaluation config
1254 raw_output = action_result.output if action_result else ""
1255 if state.evaluate.source:
1256 try:
1257 eval_input = interpolate(state.evaluate.source, ctx)
1258 except InterpolationError:
1259 eval_input = raw_output
1260 else:
1261 eval_input = raw_output
1263 if state.evaluate.type in self._contributed_evaluators:
1264 result = self._contributed_evaluators[state.evaluate.type](
1265 state.evaluate,
1266 eval_input,
1267 action_result.exit_code if action_result else 0,
1268 ctx,
1269 )
1270 elif state.evaluate.type == "llm_structured" and not self.fsm.llm.enabled:
1271 result = EvaluationResult(
1272 verdict="error",
1273 details={"error": "LLM evaluation disabled via --no-llm"},
1274 )
1275 else:
1276 result = evaluate(
1277 config=state.evaluate,
1278 output=eval_input,
1279 exit_code=action_result.exit_code if action_result else 0,
1280 context=ctx,
1281 )
1283 self._emit(
1284 "evaluate",
1285 {
1286 "type": state.evaluate.type,
1287 "verdict": result.verdict,
1288 **result.details,
1289 },
1290 )
1292 return result
1294 def _route(
1295 self,
1296 state: StateConfig,
1297 verdict: str,
1298 ctx: InterpolationContext,
1299 ) -> str | None:
1300 """Determine next state from verdict.
1302 Resolution order (from design doc):
1303 1. next (unconditional) - handled before this method
1304 2. route (full routing table)
1305 3. on_success/on_failure/on_error (shorthand)
1306 4. terminal - handled in main loop
1307 5. error
1309 Args:
1310 state: State configuration
1311 verdict: Verdict string from evaluation
1312 ctx: Interpolation context
1314 Returns:
1315 Next state name, or None if no valid route
1316 """
1317 if state.route:
1318 routes = state.route.routes
1319 if verdict in routes:
1320 return self._resolve_route(routes[verdict], ctx)
1321 if state.route.default:
1322 return self._resolve_route(state.route.default, ctx)
1323 if verdict == "error" and state.route.error:
1324 return self._resolve_route(state.route.error, ctx)
1325 if verdict == "no" and state.route.error:
1326 return self._resolve_route(state.route.error, ctx)
1327 return None
1329 # Shorthand routing
1330 if verdict == "yes" and state.on_yes:
1331 return self._resolve_route(state.on_yes, ctx)
1332 if verdict == "no" and state.on_no:
1333 return self._resolve_route(state.on_no, ctx)
1334 if verdict == "no" and not state.on_no and state.on_error:
1335 return self._resolve_route(state.on_error, ctx)
1336 if verdict == "error" and state.on_error:
1337 return self._resolve_route(state.on_error, ctx)
1338 if verdict == "partial" and state.on_partial:
1339 return self._resolve_route(state.on_partial, ctx)
1340 if verdict == "blocked" and state.on_blocked:
1341 return self._resolve_route(state.on_blocked, ctx)
1343 # Dynamic on_<verdict> shorthands from extra_routes
1344 if verdict in state.extra_routes:
1345 return self._resolve_route(state.extra_routes[verdict], ctx)
1347 return None
1349 def _resolve_route(self, route: str, ctx: InterpolationContext) -> str:
1350 """Resolve route target, handling special tokens.
1352 Args:
1353 route: Route target string
1354 ctx: Interpolation context
1356 Returns:
1357 Resolved state name
1358 """
1359 if route == "$current":
1360 return self.current_state
1361 return interpolate(route, ctx)
1363 def _flush_pending_shell_state(self, state: StateConfig) -> None:
1364 """Execute a pending shell-action state's action before honoring a
1365 wall-clock timeout. BUG-1226: closes the narrow race between emitting
1366 a `route` event and `state_enter` so handshake states (e.g. autodev's
1367 ``copy_broke_down``) do not silently drop their side effect when the
1368 timeout fires in that window. Single-step: we run the action but do
1369 not follow its routing — ``final_state`` stays as the flushed state.
1370 """
1371 assert state.action is not None # guarded by caller
1372 self.iteration += 1
1373 self._just_routed = False
1374 self._emit(
1375 "state_enter",
1376 {
1377 "state": self.current_state,
1378 "iteration": self.iteration,
1379 "flushed": True,
1380 },
1381 )
1382 ctx = self._build_context()
1383 try:
1384 self._run_action(state.action, state, ctx)
1385 except Exception:
1386 # Deliberately swallow — the timeout is being honored regardless
1387 # of whether the flushed action succeeded.
1388 pass
1390 def _action_mode(self, state: StateConfig) -> str:
1391 """Return execution mode for the state: 'prompt', 'shell', 'mcp_tool', or 'contract'."""
1392 if state.action_type == "contract":
1393 return "contract"
1394 if state.action_type == "mcp_tool":
1395 return "mcp_tool"
1396 if state.action_type in ("prompt", "slash_command"):
1397 return "prompt"
1398 if state.action_type == "shell":
1399 return "shell"
1400 if state.action_type in self._contributed_actions:
1401 return "contributed"
1402 # Heuristic: / prefix = slash_command (prompt mode)
1403 if state.action is not None and state.action.startswith("/"):
1404 return "prompt"
1405 return "shell"
1407 def _execute_with_baseline(
1408 self,
1409 state: StateConfig,
1410 ctx: InterpolationContext,
1411 baseline_cfg: dict[str, Any],
1412 ) -> tuple[ActionResult | None, str | None]:
1413 """Execute harness arm + baseline arm in parallel.
1415 The harness arm drives FSM routing; the baseline arm runs a single-shot
1416 skill invocation with no eval gates for data collection only.
1418 Returns (action_result, routed_target) where action_result is from the
1419 harness arm and routed_target is None (routing happens in _execute_state).
1420 """
1421 from concurrent.futures import ThreadPoolExecutor
1423 harness_tokens: list[tuple[int, int]] = []
1424 baseline_tokens: list[tuple[int, int]] = []
1426 def _on_harness_usage(input_tokens: int, output_tokens: int) -> None:
1427 harness_tokens.append((input_tokens, output_tokens))
1429 def _on_baseline_usage(input_tokens: int, output_tokens: int) -> None:
1430 baseline_tokens.append((input_tokens, output_tokens))
1432 # Determine baseline skill: use --baseline-skill override or extract from action
1433 baseline_skill_name = baseline_cfg.get("skill")
1434 if baseline_skill_name is None:
1435 action_text = interpolate(state.action, ctx) # type: ignore[arg-type]
1436 baseline_skill_name = _extract_skill_from_action(action_text)
1438 assert state.action is not None # caller-guarded by `if state.action:`
1439 with ThreadPoolExecutor(max_workers=2) as pool:
1440 harness_future = pool.submit(
1441 self._run_action, state.action, state, ctx, _on_harness_usage
1442 )
1443 baseline_future = pool.submit(
1444 self._run_baseline_arm, baseline_skill_name, state, _on_baseline_usage
1445 )
1446 harness_result: ActionResult = harness_future.result()
1447 baseline_result: ActionResult = baseline_future.result()
1449 harness_total_tokens = sum(t[0] + t[1] for t in harness_tokens)
1450 baseline_total_tokens = sum(t[0] + t[1] for t in baseline_tokens)
1452 self._emit(
1453 "baseline_complete",
1454 {
1455 "harness_duration_ms": harness_result.duration_ms,
1456 "baseline_duration_ms": baseline_result.duration_ms,
1457 "harness_tokens": harness_total_tokens,
1458 "baseline_tokens": baseline_total_tokens,
1459 },
1460 )
1462 # FEAT-1822: Run blind comparison and accumulate per-item results.
1463 # The blind comparator is non-fatal: if it errors, we record a
1464 # degraded result but let the harness routing proceed unchanged.
1465 try:
1466 comparison = evaluate_blind_comparator(
1467 output_harness=harness_result.output,
1468 output_baseline=baseline_result.output,
1469 )
1470 item: dict[str, Any] = {
1471 "index": self._ab_item_index,
1472 "harness_pass": comparison.get("harness_pass", False),
1473 "baseline_pass": comparison.get("baseline_pass", False),
1474 "harness_tokens": harness_total_tokens,
1475 "baseline_tokens": baseline_total_tokens,
1476 "harness_duration_ms": harness_result.duration_ms,
1477 "baseline_duration_ms": baseline_result.duration_ms,
1478 "confidence": comparison.get("confidence", 0.0),
1479 "reason": comparison.get("reason", ""),
1480 }
1481 self._ab_results.append(item)
1482 self._ab_item_index += 1
1483 self._emit("ab_comparison", {**item, "raw": comparison.get("raw", {})})
1484 except Exception:
1485 # Blind evaluation failure is non-fatal — record degraded result
1486 item = {
1487 "index": self._ab_item_index,
1488 "harness_pass": False,
1489 "baseline_pass": False,
1490 "harness_tokens": harness_total_tokens,
1491 "baseline_tokens": baseline_total_tokens,
1492 "harness_duration_ms": harness_result.duration_ms,
1493 "baseline_duration_ms": baseline_result.duration_ms,
1494 "confidence": 0.0,
1495 "reason": "Blind evaluation failed",
1496 "error": "evaluation_error",
1497 }
1498 self._ab_results.append(item)
1499 self._ab_item_index += 1
1501 return harness_result, None
1503 def _run_baseline_arm(
1504 self,
1505 skill_command: str,
1506 state: StateConfig,
1507 on_usage: UsageCallback | None = None,
1508 ) -> ActionResult:
1509 """Run a single-shot baseline skill invocation with no eval gates.
1511 Args:
1512 skill_command: Slash command for the baseline skill (e.g., "/ll:some-skill")
1513 state: State configuration (for timeout)
1514 on_usage: Optional callback for token capture
1516 Returns:
1517 ActionResult with output, exit code, and duration
1518 """
1519 start = _now_ms()
1520 timeout = state.timeout or self.fsm.default_timeout or 3600
1521 try:
1522 completed = run_claude_command(
1523 command=skill_command,
1524 timeout=timeout,
1525 on_usage=on_usage,
1526 )
1527 return ActionResult(
1528 output=completed.stdout,
1529 stderr=completed.stderr,
1530 exit_code=completed.returncode,
1531 duration_ms=_now_ms() - start,
1532 )
1533 except subprocess.TimeoutExpired:
1534 return ActionResult(
1535 output="",
1536 stderr="Baseline action timed out",
1537 exit_code=124,
1538 duration_ms=timeout * 1000,
1539 )
1541 def _run_action_or_route(
1542 self, state: StateConfig, ctx: InterpolationContext
1543 ) -> tuple[ActionResult | None, str | None]:
1544 """Run the state action, routing unhandled exceptions to on_error.
1546 Returns (action_result, routed_target). ``routed_target`` is a non-None
1547 next-state string only when an exception was raised AND ``state.on_error``
1548 is defined; in that case ``action_result`` is None. When no on_error is
1549 set, the exception is re-raised for the top-level ``run()`` handler.
1550 """
1551 assert state.action is not None # caller-guarded
1552 try:
1553 return self._run_action(state.action, state, ctx), None
1554 except Exception as exc:
1555 if state.on_error:
1556 self._emit(
1557 "action_error",
1558 {
1559 "state": self.current_state,
1560 "error": str(exc),
1561 "route": "on_error",
1562 },
1563 )
1564 return None, interpolate(state.on_error, ctx)
1565 raise
1567 def _build_context(self) -> InterpolationContext:
1568 """Build interpolation context for current state.
1570 Returns:
1571 InterpolationContext with all runtime values
1572 """
1573 from little_loops.fsm.interpolation import interpolate_dict
1575 ctx = InterpolationContext(
1576 context=self.fsm.context,
1577 captured=self.captured,
1578 prev=self.prev_result,
1579 result=None,
1580 state_name=self.current_state,
1581 iteration=self.iteration,
1582 loop_name=self.fsm.name,
1583 started_at=self.started_at,
1584 elapsed_ms=_now_ms() - self.start_time_ms + self.elapsed_offset_ms,
1585 messages=list(self.messages),
1586 )
1588 # Populate param namespace from fragment bindings (if this state uses a fragment)
1589 state = self.fsm.states.get(self.current_state)
1590 if state is not None and (state.fragment_bindings or state.fragment_parameters):
1591 resolved = interpolate_dict(state.fragment_bindings, ctx)
1592 # Apply declared defaults for unbound optional parameters
1593 for param_name, param_spec in state.fragment_parameters.items():
1594 if (
1595 param_name not in resolved
1596 and not param_spec.required
1597 and param_spec.default is not None
1598 ):
1599 resolved[param_name] = param_spec.default
1600 # Runtime enforcement: required parameters must be bound
1601 for param_name, param_spec in state.fragment_parameters.items():
1602 if param_spec.required and param_name not in resolved:
1603 raise ValueError(
1604 f"Fragment '{state.fragment_name}' requires parameter '{param_name}' "
1605 f"but it is not bound in 'with'"
1606 )
1607 ctx.param = resolved
1609 return ctx
1611 def _emit(self, event: str, data: dict[str, Any]) -> None:
1612 """Emit an event via the callback."""
1613 self.event_callback(
1614 {
1615 "event": event,
1616 "ts": _iso_now(),
1617 **data,
1618 }
1619 )
1621 def _handle_rate_limit(self, state: StateConfig, state_name: str) -> tuple[bool, str | None]:
1622 """Handle a detected 429/rate-limit action outcome.
1624 Implements the two-tier retry ladder:
1625 1. Short-burst tier: up to ``max_rate_limit_retries`` attempts with
1626 exponential backoff (``rate_limit_backoff_base_seconds * 2^n + jitter``).
1627 2. Long-wait tier: walks ``rate_limit_long_wait_ladder`` with index capped
1628 at the last entry, accumulating ``total_wait_seconds``.
1630 Routes to ``on_rate_limit_exhausted`` (falling back to ``on_error``) only
1631 once ``total_wait_seconds >= rate_limit_max_wait_seconds``. Emits
1632 ``rate_limit_exhausted`` on routing (including tier counters) and
1633 ``rate_limit_storm`` when consecutive exhaustions reach the threshold.
1635 Returns:
1636 (handled, target). ``handled=True`` means the caller should return
1637 ``target`` directly (in-place retry uses ``state_name``; exhaustion
1638 uses the routed target). ``handled=False`` should not occur for the
1639 current rate-limit classification path but is reserved for future
1640 extensions.
1641 """
1642 _short_max = (
1643 state.max_rate_limit_retries
1644 if state.max_rate_limit_retries is not None
1645 else _DEFAULT_RATE_LIMIT_RETRIES
1646 )
1647 _backoff_base = (
1648 state.rate_limit_backoff_base_seconds
1649 if state.rate_limit_backoff_base_seconds is not None
1650 else _DEFAULT_RATE_LIMIT_BACKOFF_BASE
1651 )
1652 _max_wait = (
1653 state.rate_limit_max_wait_seconds
1654 if state.rate_limit_max_wait_seconds is not None
1655 else _DEFAULT_RATE_LIMIT_MAX_WAIT_SECONDS
1656 )
1657 _ladder = (
1658 state.rate_limit_long_wait_ladder
1659 if state.rate_limit_long_wait_ladder is not None
1660 else _DEFAULT_RATE_LIMIT_LONG_WAIT_LADDER
1661 )
1663 record = self._rate_limit_retries.get(state_name)
1664 if record is None:
1665 record = {
1666 "short_retries": 0,
1667 "long_retries": 0,
1668 "total_wait_seconds": 0.0,
1669 "first_seen_at": time.time(),
1670 }
1671 self._rate_limit_retries[state_name] = record
1673 short_retries = int(record.get("short_retries", 0))
1674 long_retries = int(record.get("long_retries", 0))
1675 total_wait = float(record.get("total_wait_seconds", 0.0))
1677 if short_retries < _short_max:
1678 # Short-burst tier — exponential backoff with jitter. Budget is not
1679 # checked here; short-tier always advances to long-wait on exhaustion.
1680 short_retries += 1
1681 record["short_retries"] = short_retries
1682 _sleep = _backoff_base * (2 ** (short_retries - 1)) + random.uniform(0, _backoff_base)
1683 if self._circuit is not None:
1684 self._circuit.record_rate_limit(_sleep)
1685 total_wait += self._interruptible_sleep(_sleep)
1686 record["total_wait_seconds"] = total_wait
1687 return True, state_name # retry in place
1689 # Long-wait tier — walk ladder with capped index.
1690 long_retries += 1
1691 record["long_retries"] = long_retries
1692 _idx = min(long_retries - 1, len(_ladder) - 1)
1693 _wait = float(_ladder[_idx])
1694 if self._circuit is not None:
1695 self._circuit.record_rate_limit(_wait)
1696 _tier_start = time.time()
1697 _deadline = _tier_start + _wait
1698 _total_wait_before_tier = total_wait
1699 total_wait += self._interruptible_sleep(
1700 _wait,
1701 on_heartbeat=lambda elapsed: self._emit(
1702 RATE_LIMIT_WAITING_EVENT,
1703 {
1704 "state": state_name,
1705 "elapsed_seconds": elapsed,
1706 "next_attempt_at": _deadline,
1707 "total_waited_seconds": _total_wait_before_tier + elapsed,
1708 "budget_seconds": _max_wait,
1709 "tier": "long_wait",
1710 },
1711 ),
1712 )
1713 record["total_wait_seconds"] = total_wait
1714 if total_wait >= _max_wait:
1715 return True, self._exhaust_rate_limit(state, state_name, record)
1716 return True, state_name # retry in place
1718 def _maybe_wait_for_circuit(self, state: StateConfig) -> None:
1719 """Pre-action circuit-breaker check: sleep until shared 429 recovery.
1721 Skips quietly when no circuit is injected, when the state's action is not
1722 an LLM-quota consumer, or when the circuit has no active recovery window.
1723 """
1724 if self._circuit is None:
1725 return
1726 if self._action_mode(state) != "prompt":
1727 return
1728 recovery = self._circuit.get_estimated_recovery()
1729 if recovery is None:
1730 return
1731 wait = recovery - time.time()
1732 if wait > 0:
1733 self._interruptible_sleep(wait)
1735 def _interruptible_sleep(
1736 self,
1737 duration: float,
1738 on_heartbeat: Callable[[float], None] | None = None,
1739 ) -> float:
1740 """Sleep for up to ``duration`` seconds in 100ms ticks, exiting promptly
1741 on ``_shutdown_requested``. Returns the actual elapsed seconds so callers
1742 can accumulate wall-clock time spent in rate-limit waits.
1744 If ``on_heartbeat`` is provided, it is invoked with the elapsed seconds
1745 roughly every ``_RATE_LIMIT_HEARTBEAT_INTERVAL`` seconds so UIs can show
1746 live progress during long waits. The short-tier call site intentionally
1747 omits the callback to preserve backward-compatible silent behavior.
1748 """
1749 if duration <= 0:
1750 return 0.0
1751 _start = time.time()
1752 _deadline = _start + duration
1753 last_heartbeat = _start
1754 while time.time() < _deadline:
1755 if self._shutdown_requested:
1756 break
1757 time.sleep(min(0.1, _deadline - time.time()))
1758 if on_heartbeat is not None:
1759 _now = time.time()
1760 if _now - last_heartbeat >= _RATE_LIMIT_HEARTBEAT_INTERVAL:
1761 on_heartbeat(_now - _start)
1762 last_heartbeat = _now
1763 return time.time() - _start
1765 def _exhaust_rate_limit(
1766 self, state: StateConfig, state_name: str, record: dict[str, Any]
1767 ) -> str | None:
1768 """Finalize rate-limit exhaustion: emit event, storm detection, and
1769 return the routed target. Pops the per-state record and is called only
1770 once the wall-clock budget is spent.
1771 """
1772 self._rate_limit_retries.pop(state_name, None)
1773 target = state.on_rate_limit_exhausted or state.on_error
1774 self._emit(
1775 RATE_LIMIT_EXHAUSTED_EVENT,
1776 {
1777 "state": state_name,
1778 "retries": int(record.get("short_retries", 0)) + int(record.get("long_retries", 0)),
1779 "short_retries": int(record.get("short_retries", 0)),
1780 "long_retries": int(record.get("long_retries", 0)),
1781 "total_wait_seconds": float(record.get("total_wait_seconds", 0.0)),
1782 "next": target,
1783 },
1784 )
1785 self._consecutive_rate_limit_exhaustions += 1
1786 if self._consecutive_rate_limit_exhaustions >= _RATE_LIMIT_STORM_THRESHOLD:
1787 self._emit(
1788 RATE_LIMIT_STORM_EVENT,
1789 {
1790 "state": state_name,
1791 "count": self._consecutive_rate_limit_exhaustions,
1792 },
1793 )
1794 return target
1796 def _handle_api_error(self, state: StateConfig, state_name: str) -> tuple[bool, str | None]:
1797 """Handle a detected API server error with short-burst flat backoff.
1799 Unlike ``_handle_rate_limit``, uses a flat backoff with no long-wait tier and
1800 falls through to normal FSM routing after ``_DEFAULT_API_ERROR_RETRIES`` attempts
1801 so transient infrastructure hiccups don't permanently misdirect the loop.
1803 Returns:
1804 ``(True, state_name)`` to retry the state in place, or
1805 ``(False, None)`` when the retry budget is exhausted (caller falls
1806 through to normal verdict routing).
1807 """
1808 record = self._api_error_retries.setdefault(state_name, {"retries": 0, "total_wait": 0.0})
1809 if record["retries"] >= _DEFAULT_API_ERROR_RETRIES:
1810 self._api_error_retries.pop(state_name, None)
1811 self._emit("api_error_exhausted", {"state": state_name, "retries": record["retries"]})
1812 return False, None
1813 record["retries"] += 1
1814 slept = self._interruptible_sleep(_DEFAULT_API_ERROR_BACKOFF)
1815 record["total_wait"] += slept
1816 self._emit(
1817 "api_error_retry",
1818 {
1819 "state": state_name,
1820 "attempt": record["retries"],
1821 "backoff": _DEFAULT_API_ERROR_BACKOFF,
1822 },
1823 )
1824 return True, state_name
1826 def _finish(self, terminated_by: str, error: str | None = None) -> ExecutionResult:
1827 """Finalize execution and return result."""
1828 self._emit(
1829 "loop_complete",
1830 {
1831 "final_state": self.current_state,
1832 "iterations": self.iteration,
1833 "terminated_by": terminated_by,
1834 },
1835 )
1837 # FEAT-1822: Write ab.json if baseline comparison results exist
1838 if self._ab_results:
1839 try:
1840 from little_loops.ab_writer import calculate_ab_summary, write_ab_json
1842 run_dir = self.fsm.context.get("run_dir", "")
1843 if run_dir:
1844 summary = calculate_ab_summary(self._ab_results)
1845 write_ab_json(summary, run_dir)
1846 self._emit(
1847 "ab_summary",
1848 {
1849 "harness_pass_rate": summary.harness_pass_rate,
1850 "baseline_pass_rate": summary.baseline_pass_rate,
1851 "delta": summary.delta,
1852 "item_count": len(summary.per_item),
1853 },
1854 )
1855 except Exception:
1856 pass # Non-fatal: loop still completes
1858 return ExecutionResult(
1859 final_state=self.current_state,
1860 iterations=self.iteration,
1861 terminated_by=terminated_by,
1862 duration_ms=_now_ms() - self.start_time_ms + self.elapsed_offset_ms,
1863 captured=self.captured,
1864 error=error,
1865 messages=list(self.messages),
1866 )
1868 def _handle_handoff(self, signal: DetectedSignal) -> ExecutionResult:
1869 """Handle a detected handoff signal.
1871 Emits a handoff_detected event and optionally invokes the handoff handler.
1873 Args:
1874 signal: The detected handoff signal
1876 Returns:
1877 ExecutionResult with handoff information
1878 """
1879 self._emit(
1880 "handoff_detected",
1881 {
1882 "state": self.current_state,
1883 "iteration": self.iteration,
1884 "continuation": signal.payload,
1885 },
1886 )
1888 # Invoke handler if configured
1889 if self.handoff_handler:
1890 result = self.handoff_handler.handle(self.fsm.name, signal.payload)
1891 if result.spawned_process is not None:
1892 self._emit(
1893 "handoff_spawned",
1894 {
1895 "pid": result.spawned_process.pid,
1896 "state": self.current_state,
1897 },
1898 )
1900 return ExecutionResult(
1901 final_state=self.current_state,
1902 iterations=self.iteration,
1903 terminated_by="handoff",
1904 duration_ms=_now_ms() - self.start_time_ms + self.elapsed_offset_ms,
1905 captured=self.captured,
1906 handoff=True,
1907 continuation_prompt=signal.payload,
1908 )
1911def _extract_skill_from_action(action: str) -> str:
1912 """Extract the base skill name from an action string.
1914 For slash commands like "/ll:some-skill --arg value", returns "/ll:some-skill".
1915 For other actions, returns the action as-is.
1916 """
1917 stripped = action.strip()
1918 if stripped.startswith("/"):
1919 parts = stripped.split(maxsplit=1)
1920 return parts[0]
1921 return stripped