Coverage for little_loops / fsm / executor.py: 44%
755 statements
« prev ^ index » next coverage.py v7.12.0, created at 2026-06-08 15:34 -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 # Runner-managed runtime invariants must survive explicit `with:` binding. run_dir is
547 # injected into the parent context by the runner (cli/loop/run.py) and every loop
548 # assumes its presence (writes goals.json, batch-plan.json, etc.). The
549 # context_passthrough branch inherits it via **self.fsm.context; the with: branch
550 # must re-inject it explicitly or the child's first os.makedirs('${context.run_dir}')
551 # -> os.makedirs('') crashes. setdefault keeps an explicit `with: run_dir:` override
552 # winning if a caller ever sets one.
553 if "run_dir" in self.fsm.context:
554 child_fsm.context.setdefault("run_dir", self.fsm.context["run_dir"])
555 elif state.context_passthrough:
556 # Extract .output strings from capture result dicts so ${context.key} resolves
557 # to the plain output string (e.g. "ENH-123") rather than the full capture object.
558 captured_as_context = {
559 k: v["output"] if isinstance(v, dict) and "exit_code" in v else v
560 for k, v in self.captured.items()
561 }
562 child_fsm.context = {**self.fsm.context, **captured_as_context, **child_fsm.context}
564 depth = self._depth + 1
565 child_events: list[dict] = []
567 def _sub_event_callback(event: dict) -> None:
568 child_events.append(event)
569 # Only inject depth if not already set by a deeper nested sub-loop
570 if "depth" not in event:
571 self.event_callback({**event, "depth": depth})
572 else:
573 self.event_callback(event)
575 child_executor = FSMExecutor(
576 child_fsm,
577 action_runner=self.action_runner,
578 loops_dir=self.loops_dir,
579 event_callback=_sub_event_callback,
580 circuit=self._circuit,
581 )
582 child_executor._depth = depth # propagate depth for further nesting
584 # Clamp child timeout to parent's remaining wall-clock budget so a slow sub-loop
585 # can't silently consume the parent's deadline with no recourse for the parent FSM.
586 if self.fsm.timeout:
587 elapsed_ms = _now_ms() - self.start_time_ms + self.elapsed_offset_ms
588 remaining_s = max(1, int((self.fsm.timeout * 1000 - elapsed_ms) // 1000))
589 if child_fsm.timeout is None or child_fsm.timeout > remaining_s:
590 child_fsm.timeout = remaining_s
592 child_result = child_executor.run()
594 # Capture child event stream as a JSON-lines string if the state declares a capture key
595 if state.capture and child_events:
596 import json as _json
598 self.captured[state.capture] = {
599 "output": "\n".join(_json.dumps(e) for e in child_events),
600 "exit_code": None,
601 }
603 # Merge child captures back into parent under the state name
604 if (state.context_passthrough or state.with_) and child_executor.captured:
605 self.captured[self.current_state] = child_executor.captured
607 # Route based on child termination reason and terminal state name
608 if child_result.terminated_by == "terminal":
609 if child_result.final_state == "done":
610 return interpolate(state.on_yes, ctx) if state.on_yes else None
611 else:
612 # Reached a non-done terminal (e.g. "failed") → failure
613 return interpolate(state.on_no, ctx) if state.on_no else None
614 elif child_result.terminated_by == "error":
615 # Runtime child failure (not a YAML load error)
616 if state.on_error:
617 return interpolate(state.on_error, ctx)
618 return interpolate(state.on_no, ctx) if state.on_no else None
619 else:
620 # max_iterations, timeout, signal — all are failure
621 return interpolate(state.on_no, ctx) if state.on_no else None
623 def _execute_learning_state(self, state: StateConfig, ctx: InterpolationContext) -> str | None:
624 """Execute a FEAT-1283 ``type: learning`` state.
626 Resolves the target list at runtime: if ``learning.targets_csv`` is set,
627 it is interpolated and CSV-split; otherwise ``learning.targets`` is used
628 directly. The retry limit is resolved similarly: ``learning.max_retries_expr``
629 (if set) is interpolated and int()-cast; otherwise ``learning.max_retries``
630 (default 2) is used.
632 For each target:
633 1. Look up its record in the learning-tests registry (ENH-1282).
634 2. If proven → continue.
635 3. If refuted → emit ``learning_target_refuted`` + ``learning_blocked``
636 and route to ``on_blocked`` (preferred) or ``on_no``.
637 4. If missing or stale → emit ``learning_target_stale`` and invoke
638 ``/ll:explore-api <target>`` via the executor's action_runner;
639 re-check the registry; repeat up to ``max_retries`` times before
640 emitting ``learning_blocked`` (reason ``retries_exhausted``) and
641 routing to ``on_blocked``/``on_no``.
643 When every target ends up proven, emit ``learning_complete`` and route
644 to ``on_yes``. Returns the resolved next-state name (or ``None`` when no
645 route is configured for the terminal verdict, mirroring ``_route``).
646 """
647 from little_loops.learning_tests import check_learning_test
649 assert state.learning is not None # guarded by caller
651 # Resolve target list at runtime (ENH-1741: targets_csv support).
652 if state.learning.targets_csv is not None:
653 raw_csv = interpolate(state.learning.targets_csv, ctx)
654 targets = [t.strip() for t in raw_csv.split(",") if t.strip()]
655 else:
656 targets = list(state.learning.targets)
658 # Resolve retry limit at runtime (ENH-1741: max_retries_expr support).
659 if state.learning.max_retries_expr is not None:
660 max_retries = int(interpolate(state.learning.max_retries_expr, ctx))
661 else:
662 max_retries = state.learning.max_retries
664 def _blocked_target(reason: str, target: str) -> str | None:
665 self._emit(
666 "learning_blocked",
667 {"state": self.current_state, "target": target, "reason": reason},
668 )
669 route = state.on_blocked or state.on_no
670 return interpolate(route, ctx) if route else None
672 for target in targets:
673 record = check_learning_test(target)
675 attempts = 0
676 while record is None or record.status == "stale":
677 if attempts >= max_retries:
678 return _blocked_target("retries_exhausted", target)
680 if record is None:
681 self._emit(
682 "learning_target_stale",
683 {"state": self.current_state, "target": target, "cause": "missing"},
684 )
685 else:
686 self._emit(
687 "learning_target_stale",
688 {"state": self.current_state, "target": target, "cause": "stale"},
689 )
691 self._emit(
692 "learning_explore_invoked",
693 {"state": self.current_state, "target": target, "attempt": attempts + 1},
694 )
695 self._run_action(f"/ll:explore-api {target}", state, ctx)
696 attempts += 1
697 record = check_learning_test(target)
699 if record.status == "refuted":
700 self._emit(
701 "learning_target_refuted",
702 {"state": self.current_state, "target": target},
703 )
704 return _blocked_target("refuted", target)
706 self._emit(
707 "learning_target_proven",
708 {"state": self.current_state, "target": target},
709 )
711 self._emit(
712 "learning_complete",
713 {"state": self.current_state, "targets": targets},
714 )
715 return interpolate(state.on_yes, ctx) if state.on_yes else None
717 def _compute_progress_fingerprint(self, ctx: InterpolationContext) -> tuple[object, ...] | None:
718 """Return an (mtime, size) fingerprint for configured progress_paths, or None.
720 Called just before stall_detector.record() so that file changes made by
721 intermediate next:-only states are visible to the detector. Returns None
722 when no progress_paths are configured (preserves existing semantics).
724 Paths listed in exclude_paths (BUG-1767) are resolved and removed before
725 building the fingerprint tuple so that a loop's internal bookkeeping files
726 cannot reset the stall window on every cycle.
727 """
728 if self.fsm.circuit is None or self.fsm.circuit.repeated_failure is None:
729 return None
730 rf = self.fsm.circuit.repeated_failure
731 paths = rf.progress_paths
732 if not paths:
733 return None
735 excluded: set[str] = set()
736 for raw_excl in rf.exclude_paths:
737 try:
738 excluded.add(interpolate(raw_excl, ctx))
739 except Exception:
740 pass
742 entries: list[tuple[float, int]] = []
743 for raw_path in paths:
744 try:
745 resolved = interpolate(raw_path, ctx)
746 except Exception:
747 continue
748 if resolved in excluded:
749 continue
750 p = Path(resolved)
751 if p.exists():
752 st = p.stat()
753 entries.append((st.st_mtime, st.st_size))
754 else:
755 entries.append((0.0, 0))
756 return tuple(entries) if entries else None
758 def _check_throttle(self, state: StateConfig, state_name: str) -> str | None:
759 """Increment the per-state tool-call counter and enforce throttle thresholds.
761 Called after every action execution within a state visit. Returns the forced
762 next-state name when the hard threshold is reached, or None when execution
763 should continue normally (warn events are emitted but do not redirect).
765 Sets self._pending_error and returns "__STOP__" when the call count exceeds
766 hard_max with no on_throttle_hard target — the caller must propagate this as
767 a None return from _execute_state so the main loop detects _pending_error.
768 """
769 count = self._throttle_counts.get(state_name, 0) + 1
770 self._throttle_counts[state_name] = count
772 throttle = state.throttle
773 normal_max = (
774 throttle.normal_max
775 if (throttle and throttle.normal_max is not None)
776 else _DEFAULT_THROTTLE_NORMAL_MAX
777 )
778 warn_max = (
779 throttle.warn_max
780 if (throttle and throttle.warn_max is not None)
781 else _DEFAULT_THROTTLE_WARN_MAX
782 )
783 hard_max = (
784 throttle.hard_max
785 if (throttle and throttle.hard_max is not None)
786 else _DEFAULT_THROTTLE_HARD_MAX
787 )
789 if count == warn_max:
790 self._emit(
791 THROTTLE_WARN_EVENT,
792 {
793 "state": state_name,
794 "count": count,
795 "normal_max": normal_max,
796 "warn_max": warn_max,
797 "hard_max": hard_max,
798 },
799 )
801 # States with type="learning" (FEAT-1283) are exempt from hard_max — they
802 # legitimately make N calls per visit (one per unproven target).
803 if state.type == "learning":
804 return None
806 if count == hard_max:
807 next_target = state.on_throttle_hard or state.on_error
808 self._emit(
809 THROTTLE_HARD_EVENT,
810 {"state": state_name, "count": count, "hard_max": hard_max, "next": next_target},
811 )
812 return next_target
814 if count > hard_max:
815 self._emit(
816 THROTTLE_STOP_EVENT,
817 {"state": state_name, "count": count, "hard_max": hard_max},
818 )
819 self._pending_error = (
820 f"Throttle stop: state '{state_name}' exceeded hard_max={hard_max} "
821 "tool calls with no on_throttle_hard target"
822 )
823 return "__STOP__"
825 return None
827 def _execute_state(self, state: StateConfig) -> str | None:
828 """Execute a single state and return next state name.
830 Args:
831 state: The state configuration to execute
833 Returns:
834 Next state name, or None if no valid transition
835 """
836 # Build interpolation context
837 ctx = self._build_context()
839 # Dispatch to sub-loop handler if this is a sub-loop state
840 if state.loop is not None:
841 try:
842 return self._execute_sub_loop(state, ctx)
843 except (FileNotFoundError, ValueError):
844 if state.on_error:
845 return interpolate(state.on_error, ctx)
846 raise
848 # FEAT-1283: dispatch to learning-state handler when both type="learning"
849 # AND a LearningConfig is present. The bare `type="learning"` marker
850 # (used pre-FEAT-1283 only as a throttle hard_max exemption hint, see
851 # ThrottleConfig docstring) falls through to normal action execution.
852 if state.type == "learning" and state.learning is not None:
853 return self._execute_learning_state(state, ctx)
855 # Handle unconditional transition
856 if state.next:
857 if state.action:
858 self._maybe_wait_for_circuit(state)
859 result, routed = self._run_action_or_route(state, ctx)
860 if routed is not None:
861 return routed
862 throttle_next = self._check_throttle(state, self.current_state)
863 if throttle_next == "__STOP__":
864 return None
865 if throttle_next is not None:
866 return throttle_next
867 assert result is not None
868 self.prev_result = {
869 "output": result.output,
870 "exit_code": result.exit_code,
871 "state": self.current_state,
872 }
873 if result.exit_code is not None and result.exit_code < 0:
874 # Process killed by signal — do not silently advance via next
875 if state.on_error:
876 return interpolate(state.on_error, ctx)
877 self.request_shutdown()
878 return None
879 # Non-zero exit: if on_error is defined, treat next as success path only
880 if result.exit_code != 0 and state.on_error:
881 error_target = interpolate(state.on_error, ctx)
882 if (
883 state.retryable_exit_codes is not None
884 and result.exit_code is not None
885 and result.exit_code not in state.retryable_exit_codes
886 ):
887 # Non-retryable exit code — bypass retry, route to
888 # on_retry_exhausted (if set) or on_error directly.
889 if state.on_retry_exhausted:
890 return state.on_retry_exhausted
891 return error_target
892 return error_target
893 return interpolate(state.next, ctx)
895 # Execute action if present
896 action_result = None
897 if state.action and self._action_mode(state) != "contract":
898 self._maybe_wait_for_circuit(state)
899 baseline_cfg = ctx.context.get("_baseline")
900 if baseline_cfg and isinstance(baseline_cfg, dict) and baseline_cfg.get("enabled"):
901 # Baseline mode: spawn parallel arms, harness drives routing
902 action_result, routed = self._execute_with_baseline(state, ctx, baseline_cfg)
903 if routed is not None:
904 return routed
905 else:
906 action_result, routed = self._run_action_or_route(state, ctx)
907 if routed is not None:
908 return routed
909 throttle_next = self._check_throttle(state, self.current_state)
910 if throttle_next == "__STOP__":
911 return None
912 if throttle_next is not None:
913 return throttle_next
915 # Evaluate
916 eval_result = self._evaluate(state, action_result, ctx)
917 self.prev_result = {
918 "output": action_result.output if action_result else "",
919 "exit_code": action_result.exit_code if action_result else 0,
920 "state": self.current_state,
921 }
923 # Update context with result for routing interpolation
924 if eval_result:
925 ctx.result = {
926 "verdict": eval_result.verdict,
927 "details": eval_result.details,
928 }
930 # Route based on verdict
931 verdict = eval_result.verdict if eval_result else "yes"
933 # Stall detection (FEAT-1637). Record this transition's triple and
934 # check whether the last `window` triples are identical. On stall,
935 # either abort (set _pending_stall_abort for run() to catch) or
936 # override next_state to the configured recovery target.
937 stall_route_target: str | None = None
938 if self._stall_detector is not None:
939 stall_exit_code = action_result.exit_code if action_result else 0
940 stall_fingerprint = self._compute_progress_fingerprint(ctx)
941 self._stall_detector.record(
942 self.current_state, stall_exit_code, verdict, stall_fingerprint
943 )
944 stall = self._stall_detector.check()
945 if stall is not None:
946 assert self.fsm.circuit is not None
947 assert self.fsm.circuit.repeated_failure is not None
948 cfg_action = self.fsm.circuit.repeated_failure.on_repeated_failure
949 self._emit(
950 STALL_DETECTED_EVENT,
951 {
952 "state": self.current_state,
953 "exit_code": stall_exit_code,
954 "verdict": verdict,
955 "consecutive": stall.count,
956 "action": "abort" if cfg_action == "abort" else f"route:{cfg_action}",
957 },
958 )
959 if cfg_action == "abort":
960 self._pending_stall_abort = stall
961 return None
962 # Route to recovery target; bypass _route() entirely so the
963 # eval verdict does not pull us elsewhere first.
964 stall_route_target = cfg_action
966 route_ctx = RouteContext(
967 state_name=self.current_state,
968 state=state,
969 verdict=verdict,
970 action_result=action_result,
971 eval_result=eval_result,
972 ctx=ctx,
973 iteration=self.iteration,
974 )
975 # 429 / rate-limit detection — runs before interceptors so an in-place retry
976 # returns early without dispatching to registered before_route hooks.
977 if action_result is not None:
978 _combined = (action_result.output or "") + "\n" + (action_result.stderr or "")
979 _failure_type, _reason = classify_failure(_combined, action_result.exit_code)
980 if _failure_type == FailureType.TRANSIENT and (
981 "rate limit" in _reason.lower() or "quota" in _reason.lower()
982 ):
983 _handled, _target = self._handle_rate_limit(state, route_ctx.state_name)
984 if _handled:
985 return _target
986 elif _failure_type == FailureType.TRANSIENT and "api server error" in _reason.lower():
987 _handled, _target = self._handle_api_error(state, route_ctx.state_name)
988 if _handled:
989 return _target
990 # exhausted — fall through to normal verdict routing
991 else:
992 # Not rate-limited or server-error: reset counters so future transients start fresh.
993 self._rate_limit_retries.pop(route_ctx.state_name, None)
994 self._consecutive_rate_limit_exhaustions = 0
995 self._api_error_retries.pop(route_ctx.state_name, None)
997 # Stall-route override: if the detector elected to route to a recovery
998 # state, honor it now (bypass interceptors and _route) so the
999 # configured target wins over the eval verdict.
1000 if stall_route_target is not None:
1001 return stall_route_target
1003 # Non-retryable exit code filter for eval-based error routing. When a
1004 # state uses retryable_exit_codes and the action fails with a code that
1005 # is NOT retryable, bypass the normal error route (which may be a
1006 # self-retry) and go directly to on_retry_exhausted (or on_error).
1007 if (
1008 verdict == "error"
1009 and state.retryable_exit_codes is not None
1010 and action_result is not None
1011 and action_result.exit_code is not None
1012 and action_result.exit_code not in state.retryable_exit_codes
1013 ):
1014 if state.on_retry_exhausted:
1015 return state.on_retry_exhausted
1016 if state.on_error:
1017 return interpolate(state.on_error, ctx)
1019 for interceptor in self._interceptors:
1020 if hasattr(interceptor, "before_route"):
1021 decision = interceptor.before_route(route_ctx)
1022 if isinstance(decision, RouteDecision):
1023 if decision.next_state is None:
1024 return None # veto
1025 return decision.next_state # redirect — bypass _route()
1026 next_state = self._route(state, verdict, ctx)
1027 for interceptor in self._interceptors:
1028 if hasattr(interceptor, "after_route"):
1029 interceptor.after_route(route_ctx)
1030 return next_state
1032 def _run_action(
1033 self,
1034 action_template: str,
1035 state: StateConfig,
1036 ctx: InterpolationContext,
1037 on_usage: UsageCallback | None = None,
1038 ) -> ActionResult:
1039 """Execute action and optionally capture result.
1041 Args:
1042 action_template: Action string (may contain variables)
1043 state: State configuration
1044 ctx: Interpolation context
1045 on_usage: Optional callback invoked with (input_tokens, output_tokens) on completion
1047 Returns:
1048 ActionResult with output and exit code
1049 """
1050 action = interpolate(action_template, ctx)
1051 action_mode = self._action_mode(state)
1053 self._emit("action_start", {"action": action, "is_prompt": action_mode == "prompt"})
1055 def _on_line(line: str) -> None:
1056 self._emit("action_output", {"line": line})
1058 if action_mode == "mcp_tool":
1059 # Direct MCP tool call — bypass action_runner entirely
1060 interpolated_params = interpolate_dict(state.params, ctx) if state.params else {}
1061 cmd = ["mcp-call", action, json.dumps(interpolated_params)]
1062 result = self._run_subprocess(
1063 cmd,
1064 timeout=state.timeout or self.fsm.default_timeout or 30,
1065 on_output_line=_on_line,
1066 )
1067 elif action_mode == "contributed":
1068 assert (
1069 state.action_type is not None
1070 ) # guaranteed by _action_mode returning "contributed"
1071 runner = self._contributed_actions[state.action_type]
1072 result = runner.run(
1073 action,
1074 timeout=state.timeout or self.fsm.default_timeout or 3600,
1075 is_slash_command=False,
1076 on_output_line=_on_line,
1077 on_usage=on_usage,
1078 )
1079 else:
1080 result = self.action_runner.run(
1081 action,
1082 timeout=state.timeout or self.fsm.default_timeout or 3600,
1083 is_slash_command=action_mode == "prompt",
1084 on_output_line=_on_line,
1085 agent=state.agent if action_mode == "prompt" else None,
1086 tools=state.tools if action_mode == "prompt" else None,
1087 on_usage=on_usage,
1088 )
1090 preview = result.output[-2000:].strip() if result.output else None
1091 payload: dict[str, Any] = {
1092 "exit_code": result.exit_code,
1093 "duration_ms": result.duration_ms,
1094 "output_preview": preview,
1095 "is_prompt": action_mode == "prompt",
1096 }
1097 if action_mode == "prompt":
1098 session_jsonl = get_current_session_jsonl()
1099 payload["session_jsonl"] = str(session_jsonl) if session_jsonl else None
1100 # Aggregate token usage from host-CLI invocations (prompt / slash_command only)
1101 if result.usage_events:
1102 total_input = sum(u.input_tokens for u in result.usage_events)
1103 total_output = sum(u.output_tokens for u in result.usage_events)
1104 total_cache_read = sum(u.cache_read_tokens for u in result.usage_events)
1105 total_cache_creation = sum(u.cache_creation_tokens for u in result.usage_events)
1106 model = result.usage_events[-1].model
1107 payload["input_tokens"] = total_input
1108 payload["output_tokens"] = total_output
1109 payload["cache_read_tokens"] = total_cache_read
1110 payload["cache_creation_tokens"] = total_cache_creation
1111 payload["model"] = model
1112 self._emit("action_complete", payload)
1114 # Capture if requested
1115 if state.capture:
1116 self.captured[state.capture] = {
1117 "output": result.output.rstrip("\n\r"),
1118 "stderr": result.stderr,
1119 "exit_code": result.exit_code,
1120 "duration_ms": result.duration_ms,
1121 }
1123 # Append to shared messages log if requested
1124 if state.append_to_messages:
1125 post_ctx = self._build_context()
1126 message = interpolate(state.append_to_messages, post_ctx)
1127 self.messages.append(message)
1128 self._emit(
1129 "messages_append",
1130 {"message": message, "state": self.current_state},
1131 )
1133 # Check for signals in output
1134 if self.signal_detector:
1135 signal = self.signal_detector.detect_first(result.output)
1136 if signal:
1137 if signal.signal_type == "handoff":
1138 self._pending_handoff = signal
1139 elif signal.signal_type == "error":
1140 self._pending_error = signal.payload
1141 elif signal.signal_type == "stop":
1142 self.request_shutdown()
1144 return result
1146 def _run_subprocess(
1147 self,
1148 cmd: list[str],
1149 timeout: int,
1150 on_output_line: Any | None = None,
1151 ) -> ActionResult:
1152 """Run a subprocess directly and return ActionResult.
1154 Follows the same Popen + stderr-drain-thread pattern as DefaultActionRunner.
1156 Args:
1157 cmd: Command and arguments to execute
1158 timeout: Timeout in seconds
1159 on_output_line: Optional callback for each stdout line
1161 Returns:
1162 ActionResult with output, stderr, exit_code, duration_ms
1163 """
1164 start = _now_ms()
1165 process = subprocess.Popen(
1166 cmd,
1167 stdout=subprocess.PIPE,
1168 stderr=subprocess.PIPE,
1169 text=True,
1170 )
1171 self._current_process = process
1172 output_chunks: list[str] = []
1173 stderr_chunks: list[str] = []
1175 def _drain_stderr() -> None:
1176 assert process.stderr is not None
1177 for line in process.stderr:
1178 stderr_chunks.append(line)
1180 stderr_thread = threading.Thread(target=_drain_stderr, daemon=True)
1181 stderr_thread.start()
1183 try:
1184 for line in process.stdout: # type: ignore[union-attr]
1185 output_chunks.append(line)
1186 if on_output_line:
1187 on_output_line(line.rstrip())
1188 process.wait(timeout=timeout)
1189 except subprocess.TimeoutExpired:
1190 process.kill()
1191 process.wait()
1192 stderr_thread.join(timeout=5)
1193 return ActionResult(
1194 output="".join(output_chunks),
1195 stderr="".join(stderr_chunks) or "MCP call timed out",
1196 exit_code=124,
1197 duration_ms=timeout * 1000,
1198 )
1199 finally:
1200 self._current_process = None
1201 stderr_thread.join(timeout=5)
1202 return ActionResult(
1203 output="".join(output_chunks),
1204 stderr="".join(stderr_chunks),
1205 exit_code=process.returncode,
1206 duration_ms=_now_ms() - start,
1207 )
1209 def _evaluate(
1210 self,
1211 state: StateConfig,
1212 action_result: ActionResult | None,
1213 ctx: InterpolationContext,
1214 ) -> EvaluationResult | None:
1215 """Evaluate action result.
1217 Args:
1218 state: State configuration
1219 action_result: Result from action execution (may be None)
1220 ctx: Interpolation context
1222 Returns:
1223 EvaluationResult, or None if no evaluation needed
1224 """
1225 if state.evaluate is None:
1226 # Default evaluation based on action type
1227 if action_result:
1228 action_mode = self._action_mode(state)
1230 if action_mode == "mcp_tool":
1231 # MCP tool call: use mcp_result evaluator
1232 result = evaluate_mcp_result(action_result.output, action_result.exit_code)
1233 elif action_mode == "prompt":
1234 # Slash command or prompt: use LLM evaluation
1235 if not self.fsm.llm.enabled:
1236 result = EvaluationResult(
1237 verdict="error",
1238 details={"error": "LLM evaluation disabled via --no-llm"},
1239 )
1240 else:
1241 result = evaluate_llm_structured(
1242 action_result.output,
1243 model=self.fsm.llm.model,
1244 max_tokens=self.fsm.llm.max_tokens,
1245 timeout=self.fsm.llm.timeout,
1246 )
1247 else:
1248 # Shell command: use exit code
1249 result = evaluate_exit_code(action_result.exit_code)
1251 self._emit(
1252 "evaluate",
1253 {
1254 "type": "default",
1255 "verdict": result.verdict,
1256 **result.details,
1257 },
1258 )
1259 return result
1260 return None
1262 # Explicit evaluation config
1263 raw_output = action_result.output if action_result else ""
1264 if state.evaluate.source:
1265 try:
1266 eval_input = interpolate(state.evaluate.source, ctx)
1267 except InterpolationError:
1268 eval_input = raw_output
1269 else:
1270 eval_input = raw_output
1272 if state.evaluate.type in self._contributed_evaluators:
1273 result = self._contributed_evaluators[state.evaluate.type](
1274 state.evaluate,
1275 eval_input,
1276 action_result.exit_code if action_result else 0,
1277 ctx,
1278 )
1279 elif state.evaluate.type == "llm_structured" and not self.fsm.llm.enabled:
1280 result = EvaluationResult(
1281 verdict="error",
1282 details={"error": "LLM evaluation disabled via --no-llm"},
1283 )
1284 else:
1285 result = evaluate(
1286 config=state.evaluate,
1287 output=eval_input,
1288 exit_code=action_result.exit_code if action_result else 0,
1289 context=ctx,
1290 )
1292 self._emit(
1293 "evaluate",
1294 {
1295 "type": state.evaluate.type,
1296 "verdict": result.verdict,
1297 **result.details,
1298 },
1299 )
1301 return result
1303 def _route(
1304 self,
1305 state: StateConfig,
1306 verdict: str,
1307 ctx: InterpolationContext,
1308 ) -> str | None:
1309 """Determine next state from verdict.
1311 Resolution order (from design doc):
1312 1. next (unconditional) - handled before this method
1313 2. route (full routing table)
1314 3. on_success/on_failure/on_error (shorthand)
1315 4. terminal - handled in main loop
1316 5. error
1318 Args:
1319 state: State configuration
1320 verdict: Verdict string from evaluation
1321 ctx: Interpolation context
1323 Returns:
1324 Next state name, or None if no valid route
1325 """
1326 if state.route:
1327 routes = state.route.routes
1328 if verdict in routes:
1329 return self._resolve_route(routes[verdict], ctx)
1330 if state.route.default:
1331 return self._resolve_route(state.route.default, ctx)
1332 if verdict == "error" and state.route.error:
1333 return self._resolve_route(state.route.error, ctx)
1334 if verdict == "no" and state.route.error:
1335 return self._resolve_route(state.route.error, ctx)
1336 return None
1338 # Shorthand routing
1339 if verdict == "yes" and state.on_yes:
1340 return self._resolve_route(state.on_yes, ctx)
1341 if verdict == "no" and state.on_no:
1342 return self._resolve_route(state.on_no, ctx)
1343 if verdict == "no" and not state.on_no and state.on_error:
1344 return self._resolve_route(state.on_error, ctx)
1345 if verdict == "error" and state.on_error:
1346 return self._resolve_route(state.on_error, ctx)
1347 if verdict == "partial" and state.on_partial:
1348 return self._resolve_route(state.on_partial, ctx)
1349 if verdict == "blocked" and state.on_blocked:
1350 return self._resolve_route(state.on_blocked, ctx)
1352 # Dynamic on_<verdict> shorthands from extra_routes
1353 if verdict in state.extra_routes:
1354 return self._resolve_route(state.extra_routes[verdict], ctx)
1356 return None
1358 def _resolve_route(self, route: str, ctx: InterpolationContext) -> str:
1359 """Resolve route target, handling special tokens.
1361 Args:
1362 route: Route target string
1363 ctx: Interpolation context
1365 Returns:
1366 Resolved state name
1367 """
1368 if route == "$current":
1369 return self.current_state
1370 return interpolate(route, ctx)
1372 def _flush_pending_shell_state(self, state: StateConfig) -> None:
1373 """Execute a pending shell-action state's action before honoring a
1374 wall-clock timeout. BUG-1226: closes the narrow race between emitting
1375 a `route` event and `state_enter` so handshake states (e.g. autodev's
1376 ``copy_broke_down``) do not silently drop their side effect when the
1377 timeout fires in that window. Single-step: we run the action but do
1378 not follow its routing — ``final_state`` stays as the flushed state.
1379 """
1380 assert state.action is not None # guarded by caller
1381 self.iteration += 1
1382 self._just_routed = False
1383 self._emit(
1384 "state_enter",
1385 {
1386 "state": self.current_state,
1387 "iteration": self.iteration,
1388 "flushed": True,
1389 },
1390 )
1391 ctx = self._build_context()
1392 try:
1393 self._run_action(state.action, state, ctx)
1394 except Exception:
1395 # Deliberately swallow — the timeout is being honored regardless
1396 # of whether the flushed action succeeded.
1397 pass
1399 def _action_mode(self, state: StateConfig) -> str:
1400 """Return execution mode for the state: 'prompt', 'shell', 'mcp_tool', or 'contract'."""
1401 if state.action_type == "contract":
1402 return "contract"
1403 if state.action_type == "mcp_tool":
1404 return "mcp_tool"
1405 if state.action_type in ("prompt", "slash_command"):
1406 return "prompt"
1407 if state.action_type == "shell":
1408 return "shell"
1409 if state.action_type in self._contributed_actions:
1410 return "contributed"
1411 # Heuristic: / prefix = slash_command (prompt mode)
1412 if state.action is not None and state.action.startswith("/"):
1413 return "prompt"
1414 return "shell"
1416 def _execute_with_baseline(
1417 self,
1418 state: StateConfig,
1419 ctx: InterpolationContext,
1420 baseline_cfg: dict[str, Any],
1421 ) -> tuple[ActionResult | None, str | None]:
1422 """Execute harness arm + baseline arm in parallel.
1424 The harness arm drives FSM routing; the baseline arm runs a single-shot
1425 skill invocation with no eval gates for data collection only.
1427 Returns (action_result, routed_target) where action_result is from the
1428 harness arm and routed_target is None (routing happens in _execute_state).
1429 """
1430 from concurrent.futures import ThreadPoolExecutor
1432 harness_tokens: list[tuple[int, int]] = []
1433 baseline_tokens: list[tuple[int, int]] = []
1435 def _on_harness_usage(input_tokens: int, output_tokens: int) -> None:
1436 harness_tokens.append((input_tokens, output_tokens))
1438 def _on_baseline_usage(input_tokens: int, output_tokens: int) -> None:
1439 baseline_tokens.append((input_tokens, output_tokens))
1441 # Determine baseline skill: use --baseline-skill override or extract from action
1442 baseline_skill_name = baseline_cfg.get("skill")
1443 if baseline_skill_name is None:
1444 action_text = interpolate(state.action, ctx) # type: ignore[arg-type]
1445 baseline_skill_name = _extract_skill_from_action(action_text)
1447 assert state.action is not None # caller-guarded by `if state.action:`
1448 with ThreadPoolExecutor(max_workers=2) as pool:
1449 harness_future = pool.submit(
1450 self._run_action, state.action, state, ctx, _on_harness_usage
1451 )
1452 baseline_future = pool.submit(
1453 self._run_baseline_arm, baseline_skill_name, state, _on_baseline_usage
1454 )
1455 harness_result: ActionResult = harness_future.result()
1456 baseline_result: ActionResult = baseline_future.result()
1458 harness_total_tokens = sum(t[0] + t[1] for t in harness_tokens)
1459 baseline_total_tokens = sum(t[0] + t[1] for t in baseline_tokens)
1461 self._emit(
1462 "baseline_complete",
1463 {
1464 "harness_duration_ms": harness_result.duration_ms,
1465 "baseline_duration_ms": baseline_result.duration_ms,
1466 "harness_tokens": harness_total_tokens,
1467 "baseline_tokens": baseline_total_tokens,
1468 },
1469 )
1471 # FEAT-1822: Run blind comparison and accumulate per-item results.
1472 # The blind comparator is non-fatal: if it errors, we record a
1473 # degraded result but let the harness routing proceed unchanged.
1474 try:
1475 comparison = evaluate_blind_comparator(
1476 output_harness=harness_result.output,
1477 output_baseline=baseline_result.output,
1478 )
1479 item: dict[str, Any] = {
1480 "index": self._ab_item_index,
1481 "harness_pass": comparison.get("harness_pass", False),
1482 "baseline_pass": comparison.get("baseline_pass", False),
1483 "harness_tokens": harness_total_tokens,
1484 "baseline_tokens": baseline_total_tokens,
1485 "harness_duration_ms": harness_result.duration_ms,
1486 "baseline_duration_ms": baseline_result.duration_ms,
1487 "confidence": comparison.get("confidence", 0.0),
1488 "reason": comparison.get("reason", ""),
1489 }
1490 self._ab_results.append(item)
1491 self._ab_item_index += 1
1492 self._emit("ab_comparison", {**item, "raw": comparison.get("raw", {})})
1493 except Exception:
1494 # Blind evaluation failure is non-fatal — record degraded result
1495 item = {
1496 "index": self._ab_item_index,
1497 "harness_pass": False,
1498 "baseline_pass": False,
1499 "harness_tokens": harness_total_tokens,
1500 "baseline_tokens": baseline_total_tokens,
1501 "harness_duration_ms": harness_result.duration_ms,
1502 "baseline_duration_ms": baseline_result.duration_ms,
1503 "confidence": 0.0,
1504 "reason": "Blind evaluation failed",
1505 "error": "evaluation_error",
1506 }
1507 self._ab_results.append(item)
1508 self._ab_item_index += 1
1510 return harness_result, None
1512 def _run_baseline_arm(
1513 self,
1514 skill_command: str,
1515 state: StateConfig,
1516 on_usage: UsageCallback | None = None,
1517 ) -> ActionResult:
1518 """Run a single-shot baseline skill invocation with no eval gates.
1520 Args:
1521 skill_command: Slash command for the baseline skill (e.g., "/ll:some-skill")
1522 state: State configuration (for timeout)
1523 on_usage: Optional callback for token capture
1525 Returns:
1526 ActionResult with output, exit code, and duration
1527 """
1528 start = _now_ms()
1529 timeout = state.timeout or self.fsm.default_timeout or 3600
1530 try:
1531 completed = run_claude_command(
1532 command=skill_command,
1533 timeout=timeout,
1534 on_usage=on_usage,
1535 )
1536 return ActionResult(
1537 output=completed.stdout,
1538 stderr=completed.stderr,
1539 exit_code=completed.returncode,
1540 duration_ms=_now_ms() - start,
1541 )
1542 except subprocess.TimeoutExpired:
1543 return ActionResult(
1544 output="",
1545 stderr="Baseline action timed out",
1546 exit_code=124,
1547 duration_ms=timeout * 1000,
1548 )
1550 def _run_action_or_route(
1551 self, state: StateConfig, ctx: InterpolationContext
1552 ) -> tuple[ActionResult | None, str | None]:
1553 """Run the state action, routing unhandled exceptions to on_error.
1555 Returns (action_result, routed_target). ``routed_target`` is a non-None
1556 next-state string only when an exception was raised AND ``state.on_error``
1557 is defined; in that case ``action_result`` is None. When no on_error is
1558 set, the exception is re-raised for the top-level ``run()`` handler.
1559 """
1560 assert state.action is not None # caller-guarded
1561 try:
1562 return self._run_action(state.action, state, ctx), None
1563 except Exception as exc:
1564 if state.on_error:
1565 self._emit(
1566 "action_error",
1567 {
1568 "state": self.current_state,
1569 "error": str(exc),
1570 "route": "on_error",
1571 },
1572 )
1573 return None, interpolate(state.on_error, ctx)
1574 raise
1576 def _build_context(self) -> InterpolationContext:
1577 """Build interpolation context for current state.
1579 Returns:
1580 InterpolationContext with all runtime values
1581 """
1582 from little_loops.fsm.interpolation import interpolate_dict
1584 ctx = InterpolationContext(
1585 context=self.fsm.context,
1586 captured=self.captured,
1587 prev=self.prev_result,
1588 result=None,
1589 state_name=self.current_state,
1590 iteration=self.iteration,
1591 loop_name=self.fsm.name,
1592 started_at=self.started_at,
1593 elapsed_ms=_now_ms() - self.start_time_ms + self.elapsed_offset_ms,
1594 messages=list(self.messages),
1595 )
1597 # Populate param namespace from fragment bindings (if this state uses a fragment)
1598 state = self.fsm.states.get(self.current_state)
1599 if state is not None and (state.fragment_bindings or state.fragment_parameters):
1600 resolved = interpolate_dict(state.fragment_bindings, ctx)
1601 # Apply declared defaults for unbound optional parameters
1602 for param_name, param_spec in state.fragment_parameters.items():
1603 if (
1604 param_name not in resolved
1605 and not param_spec.required
1606 and param_spec.default is not None
1607 ):
1608 resolved[param_name] = param_spec.default
1609 # Runtime enforcement: required parameters must be bound
1610 for param_name, param_spec in state.fragment_parameters.items():
1611 if param_spec.required and param_name not in resolved:
1612 raise ValueError(
1613 f"Fragment '{state.fragment_name}' requires parameter '{param_name}' "
1614 f"but it is not bound in 'with'"
1615 )
1616 ctx.param = resolved
1618 return ctx
1620 def _emit(self, event: str, data: dict[str, Any]) -> None:
1621 """Emit an event via the callback."""
1622 self.event_callback(
1623 {
1624 "event": event,
1625 "ts": _iso_now(),
1626 **data,
1627 }
1628 )
1630 def _handle_rate_limit(self, state: StateConfig, state_name: str) -> tuple[bool, str | None]:
1631 """Handle a detected 429/rate-limit action outcome.
1633 Implements the two-tier retry ladder:
1634 1. Short-burst tier: up to ``max_rate_limit_retries`` attempts with
1635 exponential backoff (``rate_limit_backoff_base_seconds * 2^n + jitter``).
1636 2. Long-wait tier: walks ``rate_limit_long_wait_ladder`` with index capped
1637 at the last entry, accumulating ``total_wait_seconds``.
1639 Routes to ``on_rate_limit_exhausted`` (falling back to ``on_error``) only
1640 once ``total_wait_seconds >= rate_limit_max_wait_seconds``. Emits
1641 ``rate_limit_exhausted`` on routing (including tier counters) and
1642 ``rate_limit_storm`` when consecutive exhaustions reach the threshold.
1644 Returns:
1645 (handled, target). ``handled=True`` means the caller should return
1646 ``target`` directly (in-place retry uses ``state_name``; exhaustion
1647 uses the routed target). ``handled=False`` should not occur for the
1648 current rate-limit classification path but is reserved for future
1649 extensions.
1650 """
1651 _short_max = (
1652 state.max_rate_limit_retries
1653 if state.max_rate_limit_retries is not None
1654 else _DEFAULT_RATE_LIMIT_RETRIES
1655 )
1656 _backoff_base = (
1657 state.rate_limit_backoff_base_seconds
1658 if state.rate_limit_backoff_base_seconds is not None
1659 else _DEFAULT_RATE_LIMIT_BACKOFF_BASE
1660 )
1661 _max_wait = (
1662 state.rate_limit_max_wait_seconds
1663 if state.rate_limit_max_wait_seconds is not None
1664 else _DEFAULT_RATE_LIMIT_MAX_WAIT_SECONDS
1665 )
1666 _ladder = (
1667 state.rate_limit_long_wait_ladder
1668 if state.rate_limit_long_wait_ladder is not None
1669 else _DEFAULT_RATE_LIMIT_LONG_WAIT_LADDER
1670 )
1672 record = self._rate_limit_retries.get(state_name)
1673 if record is None:
1674 record = {
1675 "short_retries": 0,
1676 "long_retries": 0,
1677 "total_wait_seconds": 0.0,
1678 "first_seen_at": time.time(),
1679 }
1680 self._rate_limit_retries[state_name] = record
1682 short_retries = int(record.get("short_retries", 0))
1683 long_retries = int(record.get("long_retries", 0))
1684 total_wait = float(record.get("total_wait_seconds", 0.0))
1686 if short_retries < _short_max:
1687 # Short-burst tier — exponential backoff with jitter. Budget is not
1688 # checked here; short-tier always advances to long-wait on exhaustion.
1689 short_retries += 1
1690 record["short_retries"] = short_retries
1691 _sleep = _backoff_base * (2 ** (short_retries - 1)) + random.uniform(0, _backoff_base)
1692 if self._circuit is not None:
1693 self._circuit.record_rate_limit(_sleep)
1694 total_wait += self._interruptible_sleep(_sleep)
1695 record["total_wait_seconds"] = total_wait
1696 return True, state_name # retry in place
1698 # Long-wait tier — walk ladder with capped index.
1699 long_retries += 1
1700 record["long_retries"] = long_retries
1701 _idx = min(long_retries - 1, len(_ladder) - 1)
1702 _wait = float(_ladder[_idx])
1703 if self._circuit is not None:
1704 self._circuit.record_rate_limit(_wait)
1705 _tier_start = time.time()
1706 _deadline = _tier_start + _wait
1707 _total_wait_before_tier = total_wait
1708 total_wait += self._interruptible_sleep(
1709 _wait,
1710 on_heartbeat=lambda elapsed: self._emit(
1711 RATE_LIMIT_WAITING_EVENT,
1712 {
1713 "state": state_name,
1714 "elapsed_seconds": elapsed,
1715 "next_attempt_at": _deadline,
1716 "total_waited_seconds": _total_wait_before_tier + elapsed,
1717 "budget_seconds": _max_wait,
1718 "tier": "long_wait",
1719 },
1720 ),
1721 )
1722 record["total_wait_seconds"] = total_wait
1723 if total_wait >= _max_wait:
1724 return True, self._exhaust_rate_limit(state, state_name, record)
1725 return True, state_name # retry in place
1727 def _maybe_wait_for_circuit(self, state: StateConfig) -> None:
1728 """Pre-action circuit-breaker check: sleep until shared 429 recovery.
1730 Skips quietly when no circuit is injected, when the state's action is not
1731 an LLM-quota consumer, or when the circuit has no active recovery window.
1732 """
1733 if self._circuit is None:
1734 return
1735 if self._action_mode(state) != "prompt":
1736 return
1737 recovery = self._circuit.get_estimated_recovery()
1738 if recovery is None:
1739 return
1740 wait = recovery - time.time()
1741 if wait > 0:
1742 self._interruptible_sleep(wait)
1744 def _interruptible_sleep(
1745 self,
1746 duration: float,
1747 on_heartbeat: Callable[[float], None] | None = None,
1748 ) -> float:
1749 """Sleep for up to ``duration`` seconds in 100ms ticks, exiting promptly
1750 on ``_shutdown_requested``. Returns the actual elapsed seconds so callers
1751 can accumulate wall-clock time spent in rate-limit waits.
1753 If ``on_heartbeat`` is provided, it is invoked with the elapsed seconds
1754 roughly every ``_RATE_LIMIT_HEARTBEAT_INTERVAL`` seconds so UIs can show
1755 live progress during long waits. The short-tier call site intentionally
1756 omits the callback to preserve backward-compatible silent behavior.
1757 """
1758 if duration <= 0:
1759 return 0.0
1760 _start = time.time()
1761 _deadline = _start + duration
1762 last_heartbeat = _start
1763 while time.time() < _deadline:
1764 if self._shutdown_requested:
1765 break
1766 time.sleep(min(0.1, _deadline - time.time()))
1767 if on_heartbeat is not None:
1768 _now = time.time()
1769 if _now - last_heartbeat >= _RATE_LIMIT_HEARTBEAT_INTERVAL:
1770 on_heartbeat(_now - _start)
1771 last_heartbeat = _now
1772 return time.time() - _start
1774 def _exhaust_rate_limit(
1775 self, state: StateConfig, state_name: str, record: dict[str, Any]
1776 ) -> str | None:
1777 """Finalize rate-limit exhaustion: emit event, storm detection, and
1778 return the routed target. Pops the per-state record and is called only
1779 once the wall-clock budget is spent.
1780 """
1781 self._rate_limit_retries.pop(state_name, None)
1782 target = state.on_rate_limit_exhausted or state.on_error
1783 self._emit(
1784 RATE_LIMIT_EXHAUSTED_EVENT,
1785 {
1786 "state": state_name,
1787 "retries": int(record.get("short_retries", 0)) + int(record.get("long_retries", 0)),
1788 "short_retries": int(record.get("short_retries", 0)),
1789 "long_retries": int(record.get("long_retries", 0)),
1790 "total_wait_seconds": float(record.get("total_wait_seconds", 0.0)),
1791 "next": target,
1792 },
1793 )
1794 self._consecutive_rate_limit_exhaustions += 1
1795 if self._consecutive_rate_limit_exhaustions >= _RATE_LIMIT_STORM_THRESHOLD:
1796 self._emit(
1797 RATE_LIMIT_STORM_EVENT,
1798 {
1799 "state": state_name,
1800 "count": self._consecutive_rate_limit_exhaustions,
1801 },
1802 )
1803 return target
1805 def _handle_api_error(self, state: StateConfig, state_name: str) -> tuple[bool, str | None]:
1806 """Handle a detected API server error with short-burst flat backoff.
1808 Unlike ``_handle_rate_limit``, uses a flat backoff with no long-wait tier and
1809 falls through to normal FSM routing after ``_DEFAULT_API_ERROR_RETRIES`` attempts
1810 so transient infrastructure hiccups don't permanently misdirect the loop.
1812 Returns:
1813 ``(True, state_name)`` to retry the state in place, or
1814 ``(False, None)`` when the retry budget is exhausted (caller falls
1815 through to normal verdict routing).
1816 """
1817 record = self._api_error_retries.setdefault(state_name, {"retries": 0, "total_wait": 0.0})
1818 if record["retries"] >= _DEFAULT_API_ERROR_RETRIES:
1819 self._api_error_retries.pop(state_name, None)
1820 self._emit("api_error_exhausted", {"state": state_name, "retries": record["retries"]})
1821 return False, None
1822 record["retries"] += 1
1823 slept = self._interruptible_sleep(_DEFAULT_API_ERROR_BACKOFF)
1824 record["total_wait"] += slept
1825 self._emit(
1826 "api_error_retry",
1827 {
1828 "state": state_name,
1829 "attempt": record["retries"],
1830 "backoff": _DEFAULT_API_ERROR_BACKOFF,
1831 },
1832 )
1833 return True, state_name
1835 def _finish(self, terminated_by: str, error: str | None = None) -> ExecutionResult:
1836 """Finalize execution and return result."""
1837 self._emit(
1838 "loop_complete",
1839 {
1840 "final_state": self.current_state,
1841 "iterations": self.iteration,
1842 "terminated_by": terminated_by,
1843 },
1844 )
1846 # FEAT-1822: Write ab.json if baseline comparison results exist
1847 if self._ab_results:
1848 try:
1849 from little_loops.ab_writer import calculate_ab_summary, write_ab_json
1851 run_dir = self.fsm.context.get("run_dir", "")
1852 if run_dir:
1853 summary = calculate_ab_summary(self._ab_results)
1854 write_ab_json(summary, run_dir)
1855 self._emit(
1856 "ab_summary",
1857 {
1858 "harness_pass_rate": summary.harness_pass_rate,
1859 "baseline_pass_rate": summary.baseline_pass_rate,
1860 "delta": summary.delta,
1861 "item_count": len(summary.per_item),
1862 },
1863 )
1864 except Exception:
1865 pass # Non-fatal: loop still completes
1867 return ExecutionResult(
1868 final_state=self.current_state,
1869 iterations=self.iteration,
1870 terminated_by=terminated_by,
1871 duration_ms=_now_ms() - self.start_time_ms + self.elapsed_offset_ms,
1872 captured=self.captured,
1873 error=error,
1874 messages=list(self.messages),
1875 )
1877 def _handle_handoff(self, signal: DetectedSignal) -> ExecutionResult:
1878 """Handle a detected handoff signal.
1880 Emits a handoff_detected event and optionally invokes the handoff handler.
1882 Args:
1883 signal: The detected handoff signal
1885 Returns:
1886 ExecutionResult with handoff information
1887 """
1888 self._emit(
1889 "handoff_detected",
1890 {
1891 "state": self.current_state,
1892 "iteration": self.iteration,
1893 "continuation": signal.payload,
1894 },
1895 )
1897 # Invoke handler if configured
1898 if self.handoff_handler:
1899 result = self.handoff_handler.handle(self.fsm.name, signal.payload)
1900 if result.spawned_process is not None:
1901 self._emit(
1902 "handoff_spawned",
1903 {
1904 "pid": result.spawned_process.pid,
1905 "state": self.current_state,
1906 },
1907 )
1909 return ExecutionResult(
1910 final_state=self.current_state,
1911 iterations=self.iteration,
1912 terminated_by="handoff",
1913 duration_ms=_now_ms() - self.start_time_ms + self.elapsed_offset_ms,
1914 captured=self.captured,
1915 handoff=True,
1916 continuation_prompt=signal.payload,
1917 )
1920def _extract_skill_from_action(action: str) -> str:
1921 """Extract the base skill name from an action string.
1923 For slash commands like "/ll:some-skill --arg value", returns "/ll:some-skill".
1924 For other actions, returns the action as-is.
1925 """
1926 stripped = action.strip()
1927 if stripped.startswith("/"):
1928 parts = stripped.split(maxsplit=1)
1929 return parts[0]
1930 return stripped