Coverage for little_loops / fsm / executor.py: 10%
849 statements
« prev ^ index » next coverage.py v7.12.0, created at 2026-06-29 00:54 -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 selectors
16import subprocess
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 _kill_process_group,
55 run_claude_command,
56)
58# Maximum number of per-state rate-limit retries before emitting rate_limit_exhausted.
59_DEFAULT_RATE_LIMIT_RETRIES: int = 3
60# Base backoff in seconds; actual sleep = base * 2^(attempt-1) + uniform(0, base).
61_DEFAULT_RATE_LIMIT_BACKOFF_BASE: int = 30
62# Total wall-clock budget (seconds) across short + long tiers before routing to
63# on_rate_limit_exhausted. Mirrors RateLimitsConfig.max_wait_seconds default (6h).
64_DEFAULT_RATE_LIMIT_MAX_WAIT_SECONDS: int = 21600
65# Long-wait tier ladder (seconds), walked once the short-tier budget is spent.
66# Mirrors RateLimitsConfig.long_wait_ladder default.
67_DEFAULT_RATE_LIMIT_LONG_WAIT_LADDER: list[int] = [300, 900, 1800, 3600]
68# Event name emitted when the stall detector fires on N consecutive
69# identical (state, exit_code, verdict) transitions. See FEAT-1637.
70STALL_DETECTED_EVENT: str = "stall_detected"
71# Event name emitted when rate-limit retries are exhausted.
72RATE_LIMIT_EXHAUSTED_EVENT: str = "rate_limit_exhausted"
73# Event name emitted when consecutive rate-limit exhaustions reach the storm threshold.
74RATE_LIMIT_STORM_EVENT: str = "rate_limit_storm"
75# Event name emitted every ~60s during a long-wait rate-limit sleep so UIs can show live progress.
76RATE_LIMIT_WAITING_EVENT: str = "rate_limit_waiting"
77# Interval (seconds) between rate_limit_waiting heartbeat emissions during long-wait sleeps.
78_RATE_LIMIT_HEARTBEAT_INTERVAL: float = 60.0
79# Number of consecutive rate_limit_exhausted events that constitute a storm.
80_RATE_LIMIT_STORM_THRESHOLD: int = 3
81# Progressive throttle defaults: calls 1..normal_max pass through, at warn_max emit warning,
82# at hard_max route to on_throttle_hard, beyond hard_max hard-stop.
83_DEFAULT_THROTTLE_NORMAL_MAX: int = 3
84_DEFAULT_THROTTLE_WARN_MAX: int = 8
85_DEFAULT_THROTTLE_HARD_MAX: int = 12
86# Event names for progressive tool-call throttling within a single state visit.
87THROTTLE_WARN_EVENT: str = "throttle_warn"
88THROTTLE_HARD_EVENT: str = "throttle_hard"
89THROTTLE_STOP_EVENT: str = "throttle_stop"
90# Action types that consume LLM quota and are gated by the shared circuit breaker.
91# `_action_mode()` collapses both to "prompt"; the frozenset documents intent.
92LLM_ACTION_TYPES: frozenset[str] = frozenset({"slash_command", "prompt"})
93# Maximum per-state API server error retries before falling through to normal routing.
94_DEFAULT_API_ERROR_RETRIES: int = 2
95# Flat backoff in seconds between API server error retries (no exponential ladder).
96_DEFAULT_API_ERROR_BACKOFF: int = 30
99def _iso_now() -> str:
100 """Get current time as ISO 8601 string."""
101 return datetime.now(UTC).isoformat()
104@dataclass
105class RouteContext:
106 """Context passed to before_route / after_route interceptors."""
108 state_name: str
109 state: StateConfig
110 verdict: str
111 action_result: ActionResult | None
112 eval_result: EvaluationResult | None
113 ctx: InterpolationContext
114 iteration: int
117@dataclass
118class RouteDecision:
119 """Returned by before_route to redirect or veto a routing transition.
121 Return semantics for before_route:
122 None (implicit) → passthrough, routing proceeds normally
123 RouteDecision("state") → redirect, bypass _route() and use "state" directly
124 RouteDecision(None) → veto, _execute_state() returns None → _finish("error")
125 """
127 next_state: str | None # str → redirect; None → veto
130class FSMExecutor:
131 """Execute an FSM loop.
133 The executor runs an FSM from its initial state until:
134 - A terminal state is reached
135 - max_iterations is exceeded
136 - A timeout occurs
137 - A shutdown signal is received
138 - An unrecoverable error occurs
140 Events are emitted via the callback for observability.
141 """
143 def __init__(
144 self,
145 fsm: FSMLoop,
146 event_callback: EventCallback | None = None,
147 action_runner: ActionRunner | None = None,
148 signal_detector: SignalDetector | None = None,
149 handoff_handler: HandoffHandler | None = None,
150 loops_dir: Path | None = None,
151 circuit: RateLimitCircuit | None = None,
152 run_model: str | None = None,
153 ):
154 """Initialize the executor.
156 Args:
157 fsm: The FSM loop to execute
158 event_callback: Optional callback for events
159 action_runner: Optional custom action runner (for testing)
160 signal_detector: Optional signal detector for output parsing
161 handoff_handler: Optional handler for handoff signals
162 loops_dir: Base directory for resolving sub-loop references
163 circuit: Optional shared rate-limit circuit breaker for 429 coordination
164 run_model: Run-level default model for host-CLI action states. Per-state
165 StateConfig.model overrides this value.
166 """
167 self.fsm = fsm
168 self.event_callback = event_callback or (lambda _: None)
169 self.action_runner: ActionRunner = action_runner or DefaultActionRunner()
170 self.signal_detector = signal_detector
171 self.handoff_handler = handoff_handler
172 self.loops_dir = loops_dir
173 self._circuit = circuit
174 self.run_model = run_model
176 # Runtime state
177 self.current_state = fsm.initial
178 self.iteration = 0
179 self.captured: dict[str, dict[str, Any]] = {}
180 self.messages: list[str] = []
181 self.prev_result: dict[str, Any] | None = None
182 self.started_at = ""
183 self.start_time_ms = 0
184 self.elapsed_offset_ms = (
185 0 # milliseconds from segments before current run (set by PersistentExecutor on resume)
186 )
188 # Shutdown flag for graceful signal handling
189 self._shutdown_requested = False
191 # Currently running MCP subprocess (set by _run_subprocess, cleared in finally).
192 # Enables external shutdown code to kill the process on SIGTERM.
193 self._current_process: subprocess.Popen[str] | None = None
195 # Pending handoff signal (set by _run_action, checked by main loop)
196 self._pending_handoff: DetectedSignal | None = None
198 # Pending error payload from FATAL_ERROR signal (set by _run_action, checked by main loop)
199 self._pending_error: str | None = None
201 # Per-state retry tracking for max_retries support.
202 # _retry_counts[state_name] = number of consecutive re-entries into that state.
203 # Incremented each time we enter the same state as the previous iteration.
204 # Reset when a different state is entered, or after retry exhaustion.
205 self._retry_counts: dict[str, int] = {}
206 # State entered in the previous iteration (None on first iteration or after resume).
207 self._prev_state: str | None = None
209 # BUG-1226 / BUG-158: true between emitting a `route` event and
210 # entering the target state. Gates the "flush one pending state on
211 # timeout / max_iterations" behavior so we only flush when there is
212 # actually a pending state.
213 self._just_routed: bool = False
215 # ENH-1631 / BUG-158: true once the on_max_steps summary state has been
216 # dispatched. Prevents the step-cap guard from re-triggering before the
217 # summary state completes. Also gates the terminal-check short-circuit so
218 # the handler executes at least once before terminating (BUG-158 fix).
219 self._summary_state_executed: bool = False
221 # BUG-2204: full-pass counter (incremented on each maintain-mode restart)
222 # and flag for the on_max_iterations iteration-cap summary hook.
223 self._iteration_count: int = 0
224 self._iteration_summary_executed: bool = False
226 # FEAT-1822: Per-item A/B comparison results accumulated during baseline
227 # execution. Populated by _execute_with_baseline(), written to ab.json
228 # by _finish().
229 self._ab_results: list[dict[str, Any]] = []
230 self._ab_item_index: int = 0
232 # Per-state rate-limit retry tracking (parallel to _retry_counts).
233 # _rate_limit_retries[state_name] = dict-of-record:
234 # {
235 # "short_retries": int, # attempts in short-burst tier
236 # "long_retries": int, # attempts in long-wait tier
237 # "total_wait_seconds": float,
238 # "first_seen_at": float | None, # epoch timestamp of first 429
239 # }
240 # Incremented inside _handle_rate_limit on each detected rate-limit response.
241 # Reset when the state completes without a rate-limit, or after exhaustion.
242 self._rate_limit_retries: dict[str, dict[str, Any]] = {}
244 # States currently mid rate-limit in-place retry. Populated by _route_next_state
245 # when _handle_rate_limit returns an in-place target; consumed (discarded) by the
246 # retry-counting block on the next same-state re-entry. Exempts infrastructure
247 # pauses from the max_retries budget (BUG-2065 Fix 2).
248 self._rate_limit_in_flight: set[str] = set()
250 # Consecutive rate_limit_exhausted emissions across all states. Reset on any
251 # successful non-rate-limited state transition. When this reaches
252 # _RATE_LIMIT_STORM_THRESHOLD, a RATE_LIMIT_STORM event is emitted.
253 self._consecutive_rate_limit_exhaustions: int = 0
255 # Per-state API server error retry tracking (parallel to _rate_limit_retries).
256 # _api_error_retries[state_name] = {"retries": int, "total_wait": float}
257 # Reset when the state completes without a server error, or after exhaustion.
258 self._api_error_retries: dict[str, dict[str, Any]] = {}
260 # Per-state tool-call throttle counter. Counts successive action executions within
261 # a single continuous state visit. Reset on state exit; NOT serialized to LoopState
262 # (throttle counts measure instantaneous visit-level activity, not cumulative retries).
263 self._throttle_counts: dict[str, int] = {}
265 # Per-edge revisit counter for cycle detection.
266 # _edge_revisit_counts["from_state->to_state"] = number of times that edge has fired.
267 # When any edge exceeds max_edge_revisits, the loop terminates with cycle_detected.
268 self._edge_revisit_counts: dict[str, int] = {}
270 # Stall detector for repeated (state, exit_code, verdict) triples.
271 # Enabled via fsm.circuit.repeated_failure (FEAT-1637); None when not configured.
272 self._stall_detector: StallDetector | None = None
273 if fsm.circuit is not None and fsm.circuit.repeated_failure is not None:
274 self._stall_detector = StallDetector(window=fsm.circuit.repeated_failure.window)
275 # Set by _execute_state when the detector fires with on_repeated_failure="abort";
276 # checked by run() to terminate via _finish("stall_detected", ...).
277 self._pending_stall_abort: Stall | None = None
279 # Recurrent-window counter (ENH-2245). Tracks total occurrences of each
280 # (state, exit_code, verdict) triple across the run (non-consecutive).
281 # Enabled only when circuit.repeated_failure.recurrent_window is set.
282 self._recurrent_counts: dict[tuple[str, int, str], int] = {}
284 # Nesting depth for sub-loop event forwarding (0 = top-level, 1+ = sub-loop).
285 # Set by the parent executor when constructing child executors.
286 self._depth: int = 0
288 # Extension hook registries — populated by wire_extensions()
289 self._contributed_actions: dict[str, ActionRunner] = {}
290 self._contributed_evaluators: dict[str, Evaluator] = {}
291 self._interceptors: list[Any] = []
293 def request_shutdown(self) -> None:
294 """Request graceful shutdown of the executor.
296 Sets a flag that will be checked at the start of each iteration,
297 allowing the loop to exit cleanly after the current state completes.
298 """
299 self._shutdown_requested = True
301 def run(self) -> ExecutionResult:
302 """Execute the FSM until terminal state or limits reached.
304 Returns:
305 ExecutionResult with final state and execution metadata
306 """
307 self.started_at = _iso_now()
308 self.start_time_ms = _now_ms()
310 self._emit("loop_start", {"loop": self.fsm.name})
312 try:
313 while True:
314 # Check shutdown request (signal handling)
315 if self._shutdown_requested:
316 return self._finish("signal")
318 # Check step limit (max_steps): caps individual state executions.
319 if self.iteration >= self.fsm.max_steps:
320 if self.fsm.on_max_steps is not None and not self._summary_state_executed:
321 # BUG-158: if we just routed from a sub-loop state
322 # (e.g. its on_error fired at the step cap),
323 # flush one pending non-sub-loop state before
324 # entering the summary handler so its action isn't
325 # silently lost. Mirrors the BUG-1226 timeout guard.
326 # Single-step: no cascade. Gate: only flush when the
327 # route originated from a sub-loop (checked via
328 # _prev_state.loop). Normal self-routes are not
329 # sub-loop–originated and their action already executed
330 # — flushing them would double-count the step.
331 # Gate uses `pending.action is not None` (broader than
332 # the timeout guard's shell-only gate) so both
333 # shell-action and prompt-action intermediate states
334 # are flushed.
335 if self._just_routed:
336 prev_config = self.fsm.states.get(self._prev_state or "")
337 if prev_config is not None and prev_config.loop is not None:
338 pending = self.fsm.states.get(self.current_state)
339 if (
340 pending is not None
341 and not pending.terminal
342 and pending.loop is None
343 and pending.action is not None
344 ):
345 self._flush_pending_shell_state(pending)
346 self._emit(
347 "max_steps_summary",
348 {
349 "summary_state": self.fsm.on_max_steps,
350 "iterations": self.iteration,
351 },
352 )
353 self._summary_state_executed = True
354 self.current_state = self.fsm.on_max_steps
355 # Fall through — let the summary state run in this iteration.
356 # (do not `continue`; that would re-trigger the cap check
357 # before the state executes since self.iteration is unchanged.)
358 else:
359 return self._finish("max_steps")
361 # Check iteration limit (max_iterations): caps full loop passes
362 # (maintain-mode restarts). Only active when max_iterations is set.
363 if (
364 self.fsm.max_iterations is not None
365 and self._iteration_count >= self.fsm.max_iterations
366 ):
367 if (
368 self.fsm.on_max_iterations is not None
369 and not self._iteration_summary_executed
370 ):
371 self._emit(
372 "max_iterations_reached_summary",
373 {
374 "summary_state": self.fsm.on_max_iterations,
375 "iteration_count": self._iteration_count,
376 },
377 )
378 self._iteration_summary_executed = True
379 self.current_state = self.fsm.on_max_iterations
380 # Fall through — let the summary state run.
381 else:
382 return self._finish("max_iterations_reached")
384 # Check timeout
385 if self.fsm.timeout:
386 elapsed = _now_ms() - self.start_time_ms + self.elapsed_offset_ms
387 if elapsed > self.fsm.timeout * 1000:
388 # BUG-1226: if timeout fires in the race window between
389 # a `route` event and `state_enter`, flush one pending
390 # shell-action state before honoring the timeout so its
391 # side effect (e.g. copying a handshake flag) is not
392 # silently lost. Bounded to shell actions — slash
393 # commands and sub-loops would violate the timeout
394 # budget. Single-step: no cascade.
395 if self._just_routed:
396 pending = self.fsm.states.get(self.current_state)
397 if (
398 pending is not None
399 and not pending.terminal
400 and pending.loop is None
401 and pending.action is not None
402 and self._action_mode(pending) == "shell"
403 ):
404 self._flush_pending_shell_state(pending)
405 return self._finish("timeout")
407 # Get current state config
408 state_config = self.fsm.states[self.current_state]
410 # Update per-state retry tracking based on transition from previous iteration.
411 # If re-entering the same state consecutively, increment retry count.
412 # If entering a different state, clear the previous state's retry count.
413 if self._prev_state is not None:
414 if self.current_state == self._prev_state:
415 # Rate-limit in-place retries are infrastructure pauses, not action
416 # failures — exempt them from max_retries budget (BUG-2065 Fix 2).
417 if self.current_state not in self._rate_limit_in_flight:
418 self._retry_counts[self.current_state] = (
419 self._retry_counts.get(self.current_state, 0) + 1
420 )
421 self._rate_limit_in_flight.discard(self.current_state)
422 else:
423 self._retry_counts.pop(self._prev_state, None)
424 self._throttle_counts.pop(self._prev_state, None)
425 self._rate_limit_in_flight.discard(self._prev_state)
427 # Check terminal
428 if state_config.terminal:
429 # Handle maintain mode - restart loop instead of terminating
430 if self.fsm.maintain:
431 self.iteration += 1
432 self._iteration_count += 1 # BUG-2204: full-pass counter
433 maintain_target = state_config.on_maintain or self.fsm.initial
434 self._emit(
435 "route",
436 {
437 "from": self.current_state,
438 "to": maintain_target,
439 "reason": "maintain",
440 },
441 )
442 self._prev_state = self.current_state
443 self.current_state = maintain_target
444 self._just_routed = True
445 continue
446 # ENH-1631 / BUG-2204: if we arrived here via a cap summary state,
447 # preserve the correct terminated_by so audit tooling and
448 # PersistentExecutor see "interrupted" rather than "completed".
449 if self._summary_state_executed:
450 # BUG-158: when current_state IS the on_max_steps handler,
451 # it hasn't executed yet — let it run once before terminating.
452 # The flag prevents the cap guard from re-triggering; it
453 # shouldn't prevent the handler from executing its action.
454 if self.current_state != self.fsm.on_max_steps:
455 return self._finish("max_steps")
456 # Fall through — execute the step-cap handler's action.
457 elif self._iteration_summary_executed:
458 if self.current_state != self.fsm.on_max_iterations:
459 return self._finish("max_iterations_reached")
460 # Fall through — execute the iteration-cap handler's action.
461 else:
462 return self._finish("terminal")
464 # Check per-state retry limit. If the consecutive re-entry count exceeds
465 # max_retries, skip execution and route to on_retry_exhausted instead.
466 if state_config.max_retries is not None:
467 retry_count = self._retry_counts.get(self.current_state, 0)
468 if retry_count > state_config.max_retries:
469 # on_retry_exhausted is guaranteed non-None by validation when
470 # max_retries is set, but we fall back to an error if misconfigured.
471 exhausted_state: str = state_config.on_retry_exhausted or ""
472 if not exhausted_state:
473 return self._finish(
474 "error",
475 error=f"State '{self.current_state}' exceeded max_retries "
476 "but on_retry_exhausted is not set",
477 )
478 self._emit(
479 "retry_exhausted",
480 {
481 "state": self.current_state,
482 "retries": retry_count,
483 "next": exhausted_state,
484 },
485 )
486 self._retry_counts.pop(self.current_state, None)
487 self._prev_state = self.current_state
488 self.current_state = exhausted_state
489 continue
491 self.iteration += 1
492 self._just_routed = False
493 self._emit(
494 "state_enter",
495 {
496 "state": self.current_state,
497 "iteration": self.iteration,
498 "iteration_count": self._iteration_count,
499 },
500 )
502 # Execute state
503 next_state = self._execute_state(state_config)
505 # Check for pending error signal (FATAL_ERROR)
506 if self._pending_error is not None:
507 return self._finish("error", error=self._pending_error)
509 # Check for pending stall abort (FEAT-1637). The detector
510 # fired with on_repeated_failure="abort" inside _execute_state;
511 # terminate cleanly via _finish (mirrors the cycle_detected
512 # guard below at lines 397-416).
513 if self._pending_stall_abort is not None:
514 stall = self._pending_stall_abort
515 s_state, s_exit, s_verdict = stall.triple
516 return self._finish(
517 "stall_detected",
518 error=(
519 f"Stall detected: state '{s_state}' produced "
520 f"(exit_code={s_exit}, verdict='{s_verdict}') "
521 f"for {stall.count} consecutive iterations"
522 ),
523 )
525 # Check for pending handoff signal
526 if self._pending_handoff:
527 return self._handle_handoff(self._pending_handoff)
529 # Handle maintain mode
530 if next_state is None and self.fsm.maintain:
531 next_state = state_config.on_maintain or self.fsm.initial
533 # SIGKILL in _execute_state sets shutdown flag and returns None
534 if next_state is None and self._shutdown_requested:
535 return self._finish("signal")
537 if next_state is None:
538 # BUG-158 / BUG-2204: if a cap summary handler executed and
539 # returned None (no routing targets — typical for terminal
540 # summary states), terminate with the appropriate reason.
541 if self._summary_state_executed:
542 return self._finish("max_steps")
543 if self._iteration_summary_executed:
544 return self._finish("max_iterations_reached")
545 return self._finish("error", error="No valid transition")
547 # At this point next_state is guaranteed to be str
548 resolved_next: str = next_state
550 self._emit(
551 "route",
552 {
553 "from": self.current_state,
554 "to": resolved_next,
555 },
556 )
558 # Per-edge revisit tracking for cycle detection.
559 edge_key = f"{self.current_state}->{resolved_next}"
560 self._edge_revisit_counts[edge_key] = self._edge_revisit_counts.get(edge_key, 0) + 1
561 if self._edge_revisit_counts[edge_key] > self.fsm.max_edge_revisits:
562 self._emit(
563 "cycle_detected",
564 {
565 "edge": edge_key,
566 "from": self.current_state,
567 "to": resolved_next,
568 "count": self._edge_revisit_counts[edge_key],
569 "max": self.fsm.max_edge_revisits,
570 },
571 )
572 return self._finish(
573 "cycle_detected",
574 error=f"Cycle detected: edge {edge_key} traversed "
575 f"{self._edge_revisit_counts[edge_key]} times "
576 f"(limit: {self.fsm.max_edge_revisits})",
577 )
579 self._prev_state = self.current_state
580 self.current_state = resolved_next
581 self._just_routed = True
583 # Interruptible backoff sleep between iterations
584 if self.fsm.backoff and self.fsm.backoff > 0:
585 deadline = time.time() + self.fsm.backoff
586 while time.time() < deadline:
587 if self._shutdown_requested:
588 break
589 time.sleep(min(0.1, deadline - time.time()))
591 except InterpolationError as exc:
592 return self._finish(
593 "error",
594 error=(
595 f"Missing context variable in state '{self.current_state}': {exc}. "
596 f"Run with: ll-loop run {self.fsm.name} --context KEY=VALUE"
597 ),
598 )
599 except Exception as exc:
600 return self._finish("error", error=str(exc))
602 def _execute_sub_loop(self, state: StateConfig, ctx: InterpolationContext) -> str | None:
603 """Execute a sub-loop state by loading and running a child FSM.
605 Args:
606 state: The state configuration with loop field set
607 ctx: Interpolation context for routing
609 Returns:
610 Next state name based on child loop verdict, or None
611 """
612 from little_loops.cli.loop._helpers import resolve_loop_path
613 from little_loops.fsm.validation import load_and_validate
615 assert state.loop is not None # guarded by caller
617 # Simulation mode: stub dispatch instead of loading/running the real child FSM.
618 # Mirrors the ENH-1164 treatment of parallel: states — no side effects, no real child
619 # execution. Dynamic loop names that can't resolve in simulation yield a display label
620 # from the raw template rather than aborting with InterpolationError.
621 if isinstance(self.action_runner, SimulationActionRunner):
622 try:
623 display_name = interpolate(state.loop, ctx)
624 except InterpolationError:
625 display_name = state.loop # raw template as label when context not populated
626 sim_result = self.action_runner.run(
627 action=f"[sub-loop: {display_name}]",
628 timeout=0,
629 is_slash_command=False,
630 )
631 if sim_result.exit_code == 0:
632 return interpolate(state.on_yes, ctx) if state.on_yes else None
633 elif sim_result.exit_code == 2:
634 if state.on_error:
635 return interpolate(state.on_error, ctx)
636 return interpolate(state.on_no, ctx) if state.on_no else None
637 else:
638 return interpolate(state.on_no, ctx) if state.on_no else None
640 loop_name = interpolate(state.loop, ctx)
641 loop_path = resolve_loop_path(loop_name, self.loops_dir or Path(".loops"))
642 child_fsm, _ = load_and_validate(loop_path)
644 # Bind child context: explicit with: bindings take precedence over legacy passthrough
645 if state.with_:
646 from little_loops.fsm.interpolation import interpolate_dict
648 resolved = interpolate_dict(state.with_, ctx)
649 # Apply declared defaults for unbound optional parameters
650 for param_name, param_spec in child_fsm.parameters.items():
651 if (
652 param_name not in resolved
653 and not param_spec.required
654 and param_spec.default is not None
655 ):
656 resolved[param_name] = param_spec.default
657 # Runtime check: required parameters must be present after interpolation
658 for param_name, param_spec in child_fsm.parameters.items():
659 if param_spec.required and param_name not in resolved:
660 raise ValueError(
661 f"Sub-loop '{state.loop}' requires parameter '{param_name}' "
662 f"but it is not bound in 'with'"
663 )
664 # Merge: child's own context block provides base; with: bindings override
665 child_fsm.context = {**child_fsm.context, **resolved}
666 # Runner-managed runtime invariants must survive explicit `with:` binding. run_dir is
667 # injected into the parent context by the runner (cli/loop/run.py) and every loop
668 # assumes its presence (writes goals.json, batch-plan.json, etc.). The
669 # context_passthrough branch inherits it via **self.fsm.context; the with: branch
670 # must re-inject it explicitly or the child's first os.makedirs('${context.run_dir}')
671 # -> os.makedirs('') crashes. setdefault keeps an explicit `with: run_dir:` override
672 # winning if a caller ever sets one.
673 if "run_dir" in self.fsm.context:
674 child_fsm.context.setdefault("run_dir", self.fsm.context["run_dir"])
675 elif state.context_passthrough:
676 # Extract .output strings from capture result dicts so ${context.key} resolves
677 # to the plain output string (e.g. "ENH-123") rather than the full capture object.
678 captured_as_context = {
679 k: v["output"] if isinstance(v, dict) and "exit_code" in v else v
680 for k, v in self.captured.items()
681 }
682 child_fsm.context = {**self.fsm.context, **captured_as_context, **child_fsm.context}
684 depth = self._depth + 1
685 child_events: list[dict] = []
687 def _sub_event_callback(event: dict) -> None:
688 child_events.append(event)
689 # Only inject depth if not already set by a deeper nested sub-loop
690 if "depth" not in event:
691 self.event_callback({**event, "depth": depth})
692 else:
693 self.event_callback(event)
695 child_executor = FSMExecutor(
696 child_fsm,
697 action_runner=self.action_runner,
698 loops_dir=self.loops_dir,
699 event_callback=_sub_event_callback,
700 circuit=self._circuit,
701 )
702 child_executor._depth = depth # propagate depth for further nesting
704 # Clamp child timeout to parent's remaining wall-clock budget so a slow sub-loop
705 # can't silently consume the parent's deadline with no recourse for the parent FSM.
706 if self.fsm.timeout:
707 elapsed_ms = _now_ms() - self.start_time_ms + self.elapsed_offset_ms
708 remaining_s = max(1, int((self.fsm.timeout * 1000 - elapsed_ms) // 1000))
709 if child_fsm.timeout is None or child_fsm.timeout > remaining_s:
710 child_fsm.timeout = remaining_s
712 child_result = child_executor.run()
714 # Capture child event stream as a JSON-lines string if the state declares a capture key
715 if state.capture and child_events:
716 import json as _json
718 self.captured[state.capture] = {
719 "output": "\n".join(_json.dumps(e) for e in child_events),
720 "exit_code": None,
721 }
723 # Merge child captures back into parent under the state name
724 if (state.context_passthrough or state.with_) and child_executor.captured:
725 self.captured[self.current_state] = child_executor.captured
727 # Route based on child termination reason and terminal state name
728 if child_result.terminated_by == "terminal":
729 if child_result.final_state == "done":
730 return interpolate(state.on_yes, ctx) if state.on_yes else None
731 else:
732 # Reached a non-done terminal (e.g. "failed") → failure
733 return interpolate(state.on_no, ctx) if state.on_no else None
734 elif child_result.terminated_by == "error":
735 # Runtime child failure (not a YAML load error)
736 if child_result.error:
737 self.captured.setdefault(self.current_state, {})["error"] = child_result.error
738 if state.on_error:
739 return interpolate(state.on_error, ctx)
740 return interpolate(state.on_no, ctx) if state.on_no else None
741 else:
742 # max_iterations, timeout, signal — all are failure
743 return interpolate(state.on_no, ctx) if state.on_no else None
745 def _execute_learning_state(self, state: StateConfig, ctx: InterpolationContext) -> str | None:
746 """Execute a FEAT-1283 ``type: learning`` state.
748 Resolves the target list at runtime: if ``learning.targets_csv`` is set,
749 it is interpolated and CSV-split; otherwise ``learning.targets`` is used
750 directly. The retry limit is resolved similarly: ``learning.max_retries_expr``
751 (if set) is interpolated and int()-cast; otherwise ``learning.max_retries``
752 (default 2) is used.
754 For each target:
755 1. Look up its record in the learning-tests registry (ENH-1282).
756 2. If proven → continue.
757 3. If refuted → emit ``learning_target_refuted`` + ``learning_blocked``
758 and route to ``on_blocked`` (preferred) or ``on_no``.
759 4. If missing or stale → emit ``learning_target_stale`` and invoke
760 ``/ll:explore-api <target>`` via the executor's action_runner;
761 re-check the registry; repeat up to ``max_retries`` times before
762 emitting ``learning_blocked`` (reason ``retries_exhausted``) and
763 routing to ``on_blocked``/``on_no``.
765 When every target ends up proven, emit ``learning_complete`` and route
766 to ``on_yes``. Returns the resolved next-state name (or ``None`` when no
767 route is configured for the terminal verdict, mirroring ``_route``).
768 """
769 from little_loops.learning_tests import check_learning_test
770 from little_loops.learning_tests.gate import is_record_stale
772 assert state.learning is not None # guarded by caller
774 # Resolve target list at runtime (ENH-1741: targets_csv support).
775 if state.learning.targets_csv is not None:
776 raw_csv = interpolate(state.learning.targets_csv, ctx)
777 targets = [t.strip() for t in raw_csv.split(",") if t.strip()]
778 else:
779 targets = list(state.learning.targets)
781 # Resolve retry limit at runtime (ENH-1741: max_retries_expr support).
782 if state.learning.max_retries_expr is not None:
783 max_retries = int(interpolate(state.learning.max_retries_expr, ctx))
784 else:
785 max_retries = state.learning.max_retries
787 # Load learning_tests config once per state execution (ENH-2208).
788 # Staleness check is only active when learning_tests.enabled is True.
789 _lt_staleness_enabled = False
790 _lt_stale_days = 30
791 try:
792 from little_loops.config.core import BRConfig as _BRConfig
794 _lt_cfg = _BRConfig(Path.cwd()).learning_tests
795 if _lt_cfg.enabled:
796 _lt_staleness_enabled = True
797 _lt_stale_days = _lt_cfg.stale_after_days
798 except Exception:
799 pass
801 def _blocked_target(reason: str, target: str) -> str | None:
802 self._emit(
803 "learning_blocked",
804 {"state": self.current_state, "target": target, "reason": reason},
805 )
806 route = state.on_blocked or state.on_no
807 return interpolate(route, ctx) if route else None
809 for target in targets:
810 record = check_learning_test(target)
811 # Treat a date-stale proven record as absent so the retry path
812 # re-proves it (ENH-2208). Guard behind enabled to avoid breaking
813 # tests and projects that haven't opted into learning_tests.
814 if (
815 _lt_staleness_enabled
816 and record is not None
817 and record.status == "proven"
818 and is_record_stale(record, _lt_stale_days)
819 ):
820 record = None
822 attempts = 0
823 while record is None or record.status == "stale":
824 if attempts >= max_retries:
825 return _blocked_target("retries_exhausted", target)
827 if record is None:
828 self._emit(
829 "learning_target_stale",
830 {"state": self.current_state, "target": target, "cause": "missing"},
831 )
832 else:
833 self._emit(
834 "learning_target_stale",
835 {"state": self.current_state, "target": target, "cause": "stale"},
836 )
838 self._emit(
839 "learning_explore_invoked",
840 {"state": self.current_state, "target": target, "attempt": attempts + 1},
841 )
842 self._run_action(f"/ll:explore-api {target}", state, ctx)
843 attempts += 1
844 record = check_learning_test(target)
846 if record.status == "refuted":
847 self._emit(
848 "learning_target_refuted",
849 {"state": self.current_state, "target": target},
850 )
851 return _blocked_target("refuted", target)
853 self._emit(
854 "learning_target_proven",
855 {"state": self.current_state, "target": target},
856 )
858 self._emit(
859 "learning_complete",
860 {"state": self.current_state, "targets": targets},
861 )
862 return interpolate(state.on_yes, ctx) if state.on_yes else None
864 def _compute_progress_fingerprint(self, ctx: InterpolationContext) -> tuple[object, ...] | None:
865 """Return an (mtime, size) fingerprint for configured progress_paths, or None.
867 Called just before stall_detector.record() so that file changes made by
868 intermediate next:-only states are visible to the detector. Returns None
869 when no progress_paths are configured (preserves existing semantics).
871 Paths listed in exclude_paths (BUG-1767) are resolved and removed before
872 building the fingerprint tuple so that a loop's internal bookkeeping files
873 cannot reset the stall window on every cycle.
874 """
875 if self.fsm.circuit is None or self.fsm.circuit.repeated_failure is None:
876 return None
877 rf = self.fsm.circuit.repeated_failure
878 paths = rf.progress_paths
879 if not paths:
880 return None
882 excluded: set[str] = set()
883 for raw_excl in rf.exclude_paths:
884 try:
885 excluded.add(interpolate(raw_excl, ctx))
886 except Exception:
887 pass
889 entries: list[tuple[float, int]] = []
890 for raw_path in paths:
891 try:
892 resolved = interpolate(raw_path, ctx)
893 except Exception:
894 continue
895 if resolved in excluded:
896 continue
897 p = Path(resolved)
898 if p.exists():
899 st = p.stat()
900 entries.append((st.st_mtime, st.st_size))
901 else:
902 entries.append((0.0, 0))
903 return tuple(entries) if entries else None
905 def _check_throttle(self, state: StateConfig, state_name: str) -> str | None:
906 """Increment the per-state tool-call counter and enforce throttle thresholds.
908 Called after every action execution within a state visit. Returns the forced
909 next-state name when the hard threshold is reached, or None when execution
910 should continue normally (warn events are emitted but do not redirect).
912 Sets self._pending_error and returns "__STOP__" when the call count exceeds
913 hard_max with no on_throttle_hard target — the caller must propagate this as
914 a None return from _execute_state so the main loop detects _pending_error.
915 """
916 count = self._throttle_counts.get(state_name, 0) + 1
917 self._throttle_counts[state_name] = count
919 throttle = state.throttle
920 normal_max = (
921 throttle.normal_max
922 if (throttle and throttle.normal_max is not None)
923 else _DEFAULT_THROTTLE_NORMAL_MAX
924 )
925 warn_max = (
926 throttle.warn_max
927 if (throttle and throttle.warn_max is not None)
928 else _DEFAULT_THROTTLE_WARN_MAX
929 )
930 hard_max = (
931 throttle.hard_max
932 if (throttle and throttle.hard_max is not None)
933 else _DEFAULT_THROTTLE_HARD_MAX
934 )
936 if count == warn_max:
937 self._emit(
938 THROTTLE_WARN_EVENT,
939 {
940 "state": state_name,
941 "count": count,
942 "normal_max": normal_max,
943 "warn_max": warn_max,
944 "hard_max": hard_max,
945 },
946 )
948 # States with type="learning" (FEAT-1283) are exempt from hard_max — they
949 # legitimately make N calls per visit (one per unproven target).
950 if state.type == "learning":
951 return None
953 if count == hard_max:
954 next_target = state.on_throttle_hard or state.on_error
955 self._emit(
956 THROTTLE_HARD_EVENT,
957 {"state": state_name, "count": count, "hard_max": hard_max, "next": next_target},
958 )
959 return next_target
961 if count > hard_max:
962 self._emit(
963 THROTTLE_STOP_EVENT,
964 {"state": state_name, "count": count, "hard_max": hard_max},
965 )
966 self._pending_error = (
967 f"Throttle stop: state '{state_name}' exceeded hard_max={hard_max} "
968 "tool calls with no on_throttle_hard target"
969 )
970 return "__STOP__"
972 return None
974 def _execute_state(self, state: StateConfig) -> str | None:
975 """Execute a single state and return next state name.
977 Args:
978 state: The state configuration to execute
980 Returns:
981 Next state name, or None if no valid transition
982 """
983 # Build interpolation context
984 ctx = self._build_context()
986 # Dispatch to sub-loop handler if this is a sub-loop state
987 if state.loop is not None:
988 try:
989 return self._execute_sub_loop(state, ctx)
990 except (FileNotFoundError, ValueError, InterpolationError):
991 if state.on_error:
992 return interpolate(state.on_error, ctx)
993 raise
995 # FEAT-1283: dispatch to learning-state handler when both type="learning"
996 # AND a LearningConfig is present. The bare `type="learning"` marker
997 # (used pre-FEAT-1283 only as a throttle hard_max exemption hint, see
998 # ThrottleConfig docstring) falls through to normal action execution.
999 if state.type == "learning" and state.learning is not None:
1000 return self._execute_learning_state(state, ctx)
1002 # Handle unconditional transition
1003 if state.next:
1004 if state.action:
1005 self._maybe_wait_for_circuit(state)
1006 result, routed = self._run_action_or_route(state, ctx)
1007 if routed is not None:
1008 return routed
1009 throttle_next = self._check_throttle(state, self.current_state)
1010 if throttle_next == "__STOP__":
1011 return None
1012 if throttle_next is not None:
1013 return throttle_next
1014 assert result is not None
1015 self.prev_result = {
1016 "output": result.output,
1017 "exit_code": result.exit_code,
1018 "state": self.current_state,
1019 }
1020 if result.exit_code is not None and result.exit_code < 0:
1021 # Process killed by signal — do not silently advance via next
1022 if state.on_error:
1023 return interpolate(state.on_error, ctx)
1024 self.request_shutdown()
1025 return None
1026 # Non-zero exit: if on_error is defined, treat next as success path only
1027 if result.exit_code != 0 and state.on_error:
1028 error_target = interpolate(state.on_error, ctx)
1029 if (
1030 state.retryable_exit_codes is not None
1031 and result.exit_code is not None
1032 and result.exit_code not in state.retryable_exit_codes
1033 ):
1034 # Non-retryable exit code — bypass retry, route to
1035 # on_retry_exhausted (if set) or on_error directly.
1036 if state.on_retry_exhausted:
1037 return state.on_retry_exhausted
1038 return error_target
1039 return error_target
1040 return interpolate(state.next, ctx)
1042 # Execute action if present
1043 action_result = None
1044 if state.action and self._action_mode(state) != "contract":
1045 self._maybe_wait_for_circuit(state)
1046 baseline_cfg = ctx.context.get("_baseline")
1047 if baseline_cfg and isinstance(baseline_cfg, dict) and baseline_cfg.get("enabled"):
1048 # Baseline mode: spawn parallel arms, harness drives routing
1049 action_result, routed = self._execute_with_baseline(state, ctx, baseline_cfg)
1050 if routed is not None:
1051 return routed
1052 else:
1053 action_result, routed = self._run_action_or_route(state, ctx)
1054 if routed is not None:
1055 return routed
1056 throttle_next = self._check_throttle(state, self.current_state)
1057 if throttle_next == "__STOP__":
1058 return None
1059 if throttle_next is not None:
1060 return throttle_next
1062 # Evaluate
1063 eval_result = self._evaluate(state, action_result, ctx)
1064 self.prev_result = {
1065 "output": action_result.output if action_result else "",
1066 "exit_code": action_result.exit_code if action_result else 0,
1067 "state": self.current_state,
1068 }
1070 # Update context with result for routing interpolation
1071 if eval_result:
1072 ctx.result = {
1073 "verdict": eval_result.verdict,
1074 "details": eval_result.details,
1075 }
1077 # Route based on verdict
1078 verdict = eval_result.verdict if eval_result else "yes"
1080 # Stall detection (FEAT-1637 + ENH-2245). Record this transition's triple and
1081 # check whether the last `window` triples are identical (consecutive), or
1082 # whether the triple has been seen recurrent_window times total (non-consecutive).
1083 # On stall, either abort (set _pending_stall_abort for run() to catch) or
1084 # override next_state to the configured recovery target.
1085 stall_exit_code = action_result.exit_code if action_result else 0
1086 stall_route_target: str | None = None
1087 if self._stall_detector is not None:
1088 stall_fingerprint = self._compute_progress_fingerprint(ctx)
1089 self._stall_detector.record(
1090 self.current_state, stall_exit_code, verdict, stall_fingerprint
1091 )
1092 stall = self._stall_detector.check()
1093 if stall is not None:
1094 assert self.fsm.circuit is not None
1095 assert self.fsm.circuit.repeated_failure is not None
1096 cfg_action = self.fsm.circuit.repeated_failure.on_repeated_failure
1097 self._emit(
1098 STALL_DETECTED_EVENT,
1099 {
1100 "state": self.current_state,
1101 "exit_code": stall_exit_code,
1102 "verdict": verdict,
1103 "consecutive": stall.count,
1104 "action": "abort" if cfg_action == "abort" else f"route:{cfg_action}",
1105 },
1106 )
1107 if cfg_action == "abort":
1108 self._pending_stall_abort = stall
1109 return None
1110 # Route to recovery target; bypass _route() entirely so the
1111 # eval verdict does not pull us elsewhere first.
1112 stall_route_target = cfg_action
1114 # Recurrent-window check (ENH-2245): fire when the same (state, exit_code, verdict)
1115 # triple has been seen recurrent_window times total in this run (non-consecutive).
1116 # Only fires if the consecutive detector did not already route this iteration.
1117 if (
1118 stall_route_target is None
1119 and self.fsm.circuit is not None
1120 and self.fsm.circuit.repeated_failure is not None
1121 and self.fsm.circuit.repeated_failure.recurrent_window is not None
1122 ):
1123 rw_triple = (self.current_state, stall_exit_code, verdict)
1124 self._recurrent_counts[rw_triple] = self._recurrent_counts.get(rw_triple, 0) + 1
1125 rw_count = self._recurrent_counts[rw_triple]
1126 if rw_count >= self.fsm.circuit.repeated_failure.recurrent_window:
1127 cfg_action = self.fsm.circuit.repeated_failure.on_repeated_failure
1128 self._emit(
1129 STALL_DETECTED_EVENT,
1130 {
1131 "state": self.current_state,
1132 "exit_code": stall_exit_code,
1133 "verdict": verdict,
1134 "recurrent": rw_count,
1135 "action": "abort" if cfg_action == "abort" else f"route:{cfg_action}",
1136 },
1137 )
1138 if cfg_action == "abort":
1139 self._pending_stall_abort = Stall(triple=rw_triple, count=rw_count)
1140 return None
1141 stall_route_target = cfg_action
1143 route_ctx = RouteContext(
1144 state_name=self.current_state,
1145 state=state,
1146 verdict=verdict,
1147 action_result=action_result,
1148 eval_result=eval_result,
1149 ctx=ctx,
1150 iteration=self.iteration,
1151 )
1152 # 429 / rate-limit detection — runs before interceptors so an in-place retry
1153 # returns early without dispatching to registered before_route hooks.
1154 # BUG-2065 Fix 1: guard on exit_code != 0 so that successful actions whose
1155 # output incidentally contains "rate limit" text are never intercepted.
1156 if action_result is not None:
1157 _combined = (action_result.output or "") + "\n" + (action_result.stderr or "")
1158 _failure_type, _reason = classify_failure(_combined, action_result.exit_code)
1159 if (
1160 action_result.exit_code != 0
1161 and _failure_type == FailureType.TRANSIENT
1162 and ("rate limit" in _reason.lower() or "quota" in _reason.lower())
1163 ):
1164 _handled, _target = self._handle_rate_limit(state, route_ctx.state_name)
1165 if _handled:
1166 self._rate_limit_in_flight.add(route_ctx.state_name)
1167 return _target
1168 elif (
1169 action_result.exit_code != 0
1170 and _failure_type == FailureType.TRANSIENT
1171 and "api server error" in _reason.lower()
1172 ):
1173 _handled, _target = self._handle_api_error(state, route_ctx.state_name)
1174 if _handled:
1175 return _target
1176 # exhausted — fall through to normal verdict routing
1177 elif action_result.exit_code != 0 and _failure_type == FailureType.NON_RECOVERABLE:
1178 # Auth/credential failure — abort immediately, do not retry.
1179 # Route to on_error if defined, otherwise let normal verdict routing handle it.
1180 _on_error = state.on_error
1181 if _on_error:
1182 return _on_error
1183 # No on_error: fall through to verdict routing (will produce non-zero exit verdict)
1184 else:
1185 # Not rate-limited or server-error (or exit_code=0): reset counters so
1186 # future transients start fresh.
1187 self._rate_limit_retries.pop(route_ctx.state_name, None)
1188 self._consecutive_rate_limit_exhaustions = 0
1189 self._api_error_retries.pop(route_ctx.state_name, None)
1191 # Stall-route override: if the detector elected to route to a recovery
1192 # state, honor it now (bypass interceptors and _route) so the
1193 # configured target wins over the eval verdict.
1194 if stall_route_target is not None:
1195 return stall_route_target
1197 # Non-retryable exit code filter for eval-based error routing. When a
1198 # state uses retryable_exit_codes and the action fails with a code that
1199 # is NOT retryable, bypass the normal error route (which may be a
1200 # self-retry) and go directly to on_retry_exhausted (or on_error).
1201 if (
1202 verdict == "error"
1203 and state.retryable_exit_codes is not None
1204 and action_result is not None
1205 and action_result.exit_code is not None
1206 and action_result.exit_code not in state.retryable_exit_codes
1207 ):
1208 if state.on_retry_exhausted:
1209 return state.on_retry_exhausted
1210 if state.on_error:
1211 return interpolate(state.on_error, ctx)
1213 for interceptor in self._interceptors:
1214 if hasattr(interceptor, "before_route"):
1215 decision = interceptor.before_route(route_ctx)
1216 if isinstance(decision, RouteDecision):
1217 if decision.next_state is None:
1218 return None # veto
1219 return decision.next_state # redirect — bypass _route()
1220 next_state = self._route(state, verdict, ctx)
1221 for interceptor in self._interceptors:
1222 if hasattr(interceptor, "after_route"):
1223 interceptor.after_route(route_ctx)
1224 return next_state
1226 def _run_action(
1227 self,
1228 action_template: str,
1229 state: StateConfig,
1230 ctx: InterpolationContext,
1231 on_usage: UsageCallback | None = None,
1232 ) -> ActionResult:
1233 """Execute action and optionally capture result.
1235 Args:
1236 action_template: Action string (may contain variables)
1237 state: State configuration
1238 ctx: Interpolation context
1239 on_usage: Optional callback invoked with (input_tokens, output_tokens) on completion
1241 Returns:
1242 ActionResult with output and exit code
1243 """
1244 action = interpolate(action_template, ctx)
1245 action_mode = self._action_mode(state)
1247 self._emit("action_start", {"action": action, "is_prompt": action_mode == "prompt"})
1249 def _on_line(line: str) -> None:
1250 self._emit("action_output", {"line": line})
1252 if action_mode == "mcp_tool":
1253 # Direct MCP tool call — bypass action_runner entirely
1254 interpolated_params = interpolate_dict(state.params, ctx) if state.params else {}
1255 cmd = ["mcp-call", action, json.dumps(interpolated_params)]
1256 result = self._run_subprocess(
1257 cmd,
1258 timeout=state.timeout or self.fsm.default_timeout or 30,
1259 on_output_line=_on_line,
1260 )
1261 elif action_mode == "contributed":
1262 assert (
1263 state.action_type is not None
1264 ) # guaranteed by _action_mode returning "contributed"
1265 runner = self._contributed_actions[state.action_type]
1266 result = runner.run(
1267 action,
1268 timeout=state.timeout or self.fsm.default_timeout or 3600,
1269 is_slash_command=False,
1270 on_output_line=_on_line,
1271 on_usage=on_usage,
1272 )
1273 else:
1274 result = self.action_runner.run(
1275 action,
1276 timeout=state.timeout or self.fsm.default_timeout or 3600,
1277 is_slash_command=action_mode == "prompt",
1278 on_output_line=_on_line,
1279 agent=state.agent if action_mode == "prompt" else None,
1280 tools=state.tools if action_mode == "prompt" else None,
1281 on_usage=on_usage,
1282 model=(state.model or self.run_model) if action_mode == "prompt" else None,
1283 )
1285 preview = result.output[-2000:].strip() if result.output else None
1286 payload: dict[str, Any] = {
1287 "exit_code": result.exit_code,
1288 "duration_ms": result.duration_ms,
1289 "output_preview": preview,
1290 "is_prompt": action_mode == "prompt",
1291 }
1292 if action_mode == "prompt":
1293 session_jsonl = get_current_session_jsonl()
1294 payload["session_jsonl"] = str(session_jsonl) if session_jsonl else None
1295 # Aggregate token usage from host-CLI invocations (prompt / slash_command only)
1296 if result.usage_events:
1297 total_input = sum(u.input_tokens for u in result.usage_events)
1298 total_output = sum(u.output_tokens for u in result.usage_events)
1299 total_cache_read = sum(u.cache_read_tokens for u in result.usage_events)
1300 total_cache_creation = sum(u.cache_creation_tokens for u in result.usage_events)
1301 model = result.usage_events[-1].model
1302 payload["input_tokens"] = total_input
1303 payload["output_tokens"] = total_output
1304 payload["cache_read_tokens"] = total_cache_read
1305 payload["cache_creation_tokens"] = total_cache_creation
1306 payload["model"] = model
1307 self._emit("action_complete", payload)
1309 # Capture if requested
1310 if state.capture:
1311 self.captured[state.capture] = {
1312 "output": result.output.rstrip("\n\r"),
1313 "stderr": result.stderr,
1314 "exit_code": result.exit_code,
1315 "duration_ms": result.duration_ms,
1316 }
1318 # Append to shared messages log if requested
1319 if state.append_to_messages:
1320 post_ctx = self._build_context()
1321 message = interpolate(state.append_to_messages, post_ctx)
1322 self.messages.append(message)
1323 self._emit(
1324 "messages_append",
1325 {"message": message, "state": self.current_state},
1326 )
1328 # Check for signals in output
1329 if self.signal_detector:
1330 signal = self.signal_detector.detect_first(result.output)
1331 if signal:
1332 if signal.signal_type == "handoff":
1333 self._pending_handoff = signal
1334 elif signal.signal_type == "error":
1335 self._pending_error = signal.payload
1336 elif signal.signal_type == "stop":
1337 self.request_shutdown()
1339 return result
1341 def _run_subprocess(
1342 self,
1343 cmd: list[str],
1344 timeout: int,
1345 on_output_line: Any | None = None,
1346 ) -> ActionResult:
1347 """Run a subprocess directly and return ActionResult.
1349 Uses selector-based I/O with wall-clock timeout enforcement so the
1350 timeout is honoured even when the process hangs before producing output.
1351 """
1352 start = _now_ms()
1353 process = subprocess.Popen(
1354 cmd,
1355 stdout=subprocess.PIPE,
1356 stderr=subprocess.PIPE,
1357 text=True,
1358 )
1359 self._current_process = process
1360 deadline = time.time() + timeout
1362 output_chunks: list[str] = []
1363 stderr_chunks: list[str] = []
1365 sel = selectors.DefaultSelector()
1366 if process.stdout is not None:
1367 sel.register(process.stdout, selectors.EVENT_READ, data="stdout")
1368 if process.stderr is not None:
1369 sel.register(process.stderr, selectors.EVENT_READ, data="stderr")
1371 timed_out = False
1372 try:
1373 while sel.get_map():
1374 remaining = deadline - time.time()
1375 if remaining <= 0:
1376 timed_out = True
1377 break
1378 poll_timeout = min(1.0, remaining)
1379 ready = sel.select(timeout=poll_timeout)
1380 if not ready:
1381 continue
1382 for key, _mask in ready:
1383 line = key.fileobj.readline() # type: ignore[union-attr]
1384 if line:
1385 if key.data == "stdout":
1386 output_chunks.append(line)
1387 if on_output_line:
1388 on_output_line(line.rstrip())
1389 else:
1390 stderr_chunks.append(line)
1391 else:
1392 sel.unregister(key.fileobj)
1393 finally:
1394 sel.close()
1395 self._current_process = None
1397 if timed_out:
1398 _kill_process_group(process)
1399 try:
1400 process.wait(timeout=5)
1401 except subprocess.TimeoutExpired:
1402 process.kill()
1403 process.wait()
1404 return ActionResult(
1405 output="".join(output_chunks),
1406 stderr="".join(stderr_chunks) or "MCP call timed out",
1407 exit_code=124,
1408 duration_ms=timeout * 1000,
1409 )
1411 process.wait(timeout=5)
1412 return ActionResult(
1413 output="".join(output_chunks),
1414 stderr="".join(stderr_chunks),
1415 exit_code=process.returncode,
1416 duration_ms=_now_ms() - start,
1417 )
1419 def _evaluate(
1420 self,
1421 state: StateConfig,
1422 action_result: ActionResult | None,
1423 ctx: InterpolationContext,
1424 ) -> EvaluationResult | None:
1425 """Evaluate action result.
1427 Args:
1428 state: State configuration
1429 action_result: Result from action execution (may be None)
1430 ctx: Interpolation context
1432 Returns:
1433 EvaluationResult, or None if no evaluation needed
1434 """
1435 if state.evaluate is None:
1436 # Default evaluation based on action type
1437 if action_result:
1438 action_mode = self._action_mode(state)
1440 if action_mode == "mcp_tool":
1441 # MCP tool call: use mcp_result evaluator
1442 result = evaluate_mcp_result(action_result.output, action_result.exit_code)
1443 elif action_mode == "prompt":
1444 # Slash command or prompt: use LLM evaluation
1445 if not self.fsm.llm.enabled:
1446 result = EvaluationResult(
1447 verdict="error",
1448 details={"error": "LLM evaluation disabled via --no-llm"},
1449 )
1450 else:
1451 result = evaluate_llm_structured(
1452 action_result.output,
1453 model=self.fsm.llm.model,
1454 max_tokens=self.fsm.llm.max_tokens,
1455 timeout=self.fsm.llm.timeout,
1456 )
1457 else:
1458 # Shell command: use exit code
1459 result = evaluate_exit_code(action_result.exit_code)
1461 self._emit(
1462 "evaluate",
1463 {
1464 "type": "default",
1465 "verdict": result.verdict,
1466 **result.details,
1467 },
1468 )
1469 return result
1470 return None
1472 # Explicit evaluation config
1473 raw_output = action_result.output if action_result else ""
1474 if state.evaluate.source:
1475 try:
1476 eval_input = interpolate(state.evaluate.source, ctx)
1477 except InterpolationError:
1478 eval_input = raw_output
1479 else:
1480 eval_input = raw_output
1482 if state.evaluate.type in self._contributed_evaluators:
1483 result = self._contributed_evaluators[state.evaluate.type](
1484 state.evaluate,
1485 eval_input,
1486 action_result.exit_code if action_result else 0,
1487 ctx,
1488 )
1489 elif state.evaluate.type == "llm_structured" and not self.fsm.llm.enabled:
1490 result = EvaluationResult(
1491 verdict="error",
1492 details={"error": "LLM evaluation disabled via --no-llm"},
1493 )
1494 else:
1495 result = evaluate(
1496 config=state.evaluate,
1497 output=eval_input,
1498 exit_code=action_result.exit_code if action_result else 0,
1499 context=ctx,
1500 )
1502 self._emit(
1503 "evaluate",
1504 {
1505 "type": state.evaluate.type,
1506 "verdict": result.verdict,
1507 **result.details,
1508 },
1509 )
1511 return result
1513 def _route(
1514 self,
1515 state: StateConfig,
1516 verdict: str,
1517 ctx: InterpolationContext,
1518 ) -> str | None:
1519 """Determine next state from verdict.
1521 Resolution order (from design doc):
1522 1. next (unconditional) - handled before this method
1523 2. route (full routing table)
1524 3. on_success/on_failure/on_error (shorthand)
1525 4. terminal - handled in main loop
1526 5. error
1528 Args:
1529 state: State configuration
1530 verdict: Verdict string from evaluation
1531 ctx: Interpolation context
1533 Returns:
1534 Next state name, or None if no valid route
1535 """
1536 if state.route:
1537 routes = state.route.routes
1538 if verdict in routes:
1539 return self._resolve_route(routes[verdict], ctx)
1540 if state.route.default:
1541 return self._resolve_route(state.route.default, ctx)
1542 if verdict == "error" and state.route.error:
1543 return self._resolve_route(state.route.error, ctx)
1544 if verdict == "no" and state.route.error:
1545 return self._resolve_route(state.route.error, ctx)
1546 return None
1548 # Shorthand routing
1549 if verdict == "yes" and state.on_yes:
1550 return self._resolve_route(state.on_yes, ctx)
1551 if verdict == "no" and state.on_no:
1552 return self._resolve_route(state.on_no, ctx)
1553 if verdict == "no" and not state.on_no and state.on_error:
1554 return self._resolve_route(state.on_error, ctx)
1555 if verdict == "error" and state.on_error:
1556 return self._resolve_route(state.on_error, ctx)
1557 if verdict == "partial" and state.on_partial:
1558 return self._resolve_route(state.on_partial, ctx)
1559 if verdict == "blocked" and state.on_blocked:
1560 return self._resolve_route(state.on_blocked, ctx)
1562 # Dynamic on_<verdict> shorthands from extra_routes
1563 if verdict in state.extra_routes:
1564 return self._resolve_route(state.extra_routes[verdict], ctx)
1566 return None
1568 def _resolve_route(self, route: str, ctx: InterpolationContext) -> str:
1569 """Resolve route target, handling special tokens.
1571 Args:
1572 route: Route target string
1573 ctx: Interpolation context
1575 Returns:
1576 Resolved state name
1577 """
1578 if route == "$current":
1579 return self.current_state
1580 return interpolate(route, ctx)
1582 def _flush_pending_shell_state(self, state: StateConfig) -> None:
1583 """Execute a pending shell-action state's action before honoring a
1584 wall-clock timeout. BUG-1226: closes the narrow race between emitting
1585 a `route` event and `state_enter` so handshake states (e.g. autodev's
1586 ``copy_broke_down``) do not silently drop their side effect when the
1587 timeout fires in that window. Single-step: we run the action but do
1588 not follow its routing — ``final_state`` stays as the flushed state.
1589 """
1590 assert state.action is not None # guarded by caller
1591 self.iteration += 1
1592 self._just_routed = False
1593 self._emit(
1594 "state_enter",
1595 {
1596 "state": self.current_state,
1597 "iteration": self.iteration,
1598 "iteration_count": self._iteration_count,
1599 "flushed": True,
1600 },
1601 )
1602 ctx = self._build_context()
1603 try:
1604 self._run_action(state.action, state, ctx)
1605 except Exception:
1606 # Deliberately swallow — the timeout is being honored regardless
1607 # of whether the flushed action succeeded.
1608 pass
1610 def _action_mode(self, state: StateConfig) -> str:
1611 """Return execution mode for the state: 'prompt', 'shell', 'mcp_tool', or 'contract'."""
1612 if state.action_type == "contract":
1613 return "contract"
1614 if state.action_type == "mcp_tool":
1615 return "mcp_tool"
1616 if state.action_type in ("prompt", "slash_command"):
1617 return "prompt"
1618 if state.action_type == "shell":
1619 return "shell"
1620 if state.action_type in self._contributed_actions:
1621 return "contributed"
1622 # Heuristic: / prefix = slash_command (prompt mode)
1623 if state.action is not None and state.action.startswith("/"):
1624 return "prompt"
1625 return "shell"
1627 def _execute_with_baseline(
1628 self,
1629 state: StateConfig,
1630 ctx: InterpolationContext,
1631 baseline_cfg: dict[str, Any],
1632 ) -> tuple[ActionResult | None, str | None]:
1633 """Execute harness arm + baseline arm in parallel.
1635 The harness arm drives FSM routing; the baseline arm runs a single-shot
1636 skill invocation with no eval gates for data collection only.
1638 Returns (action_result, routed_target) where action_result is from the
1639 harness arm and routed_target is None (routing happens in _execute_state).
1640 """
1641 from concurrent.futures import ThreadPoolExecutor
1643 harness_tokens: list[tuple[int, int]] = []
1644 baseline_tokens: list[tuple[int, int]] = []
1646 def _on_harness_usage(input_tokens: int, output_tokens: int) -> None:
1647 harness_tokens.append((input_tokens, output_tokens))
1649 def _on_baseline_usage(input_tokens: int, output_tokens: int) -> None:
1650 baseline_tokens.append((input_tokens, output_tokens))
1652 # Determine baseline skill: use --baseline-skill override or extract from action
1653 baseline_skill_name = baseline_cfg.get("skill")
1654 if baseline_skill_name is None:
1655 action_text = interpolate(state.action, ctx) # type: ignore[arg-type]
1656 baseline_skill_name = _extract_skill_from_action(action_text)
1658 assert state.action is not None # caller-guarded by `if state.action:`
1659 with ThreadPoolExecutor(max_workers=2) as pool:
1660 harness_future = pool.submit(
1661 self._run_action, state.action, state, ctx, _on_harness_usage
1662 )
1663 baseline_future = pool.submit(
1664 self._run_baseline_arm, baseline_skill_name, state, _on_baseline_usage
1665 )
1666 harness_result: ActionResult = harness_future.result()
1667 baseline_result: ActionResult = baseline_future.result()
1669 harness_total_tokens = sum(t[0] + t[1] for t in harness_tokens)
1670 baseline_total_tokens = sum(t[0] + t[1] for t in baseline_tokens)
1672 self._emit(
1673 "baseline_complete",
1674 {
1675 "harness_duration_ms": harness_result.duration_ms,
1676 "baseline_duration_ms": baseline_result.duration_ms,
1677 "harness_tokens": harness_total_tokens,
1678 "baseline_tokens": baseline_total_tokens,
1679 },
1680 )
1682 # FEAT-1822: Run blind comparison and accumulate per-item results.
1683 # The blind comparator is non-fatal: if it errors, we record a
1684 # degraded result but let the harness routing proceed unchanged.
1685 try:
1686 comparison = evaluate_blind_comparator(
1687 output_harness=harness_result.output,
1688 output_baseline=baseline_result.output,
1689 )
1690 item: dict[str, Any] = {
1691 "index": self._ab_item_index,
1692 "harness_pass": comparison.get("harness_pass", False),
1693 "baseline_pass": comparison.get("baseline_pass", False),
1694 "harness_tokens": harness_total_tokens,
1695 "baseline_tokens": baseline_total_tokens,
1696 "harness_duration_ms": harness_result.duration_ms,
1697 "baseline_duration_ms": baseline_result.duration_ms,
1698 "confidence": comparison.get("confidence", 0.0),
1699 "reason": comparison.get("reason", ""),
1700 }
1701 self._ab_results.append(item)
1702 self._ab_item_index += 1
1703 self._emit("ab_comparison", {**item, "raw": comparison.get("raw", {})})
1704 except Exception:
1705 # Blind evaluation failure is non-fatal — record degraded result
1706 item = {
1707 "index": self._ab_item_index,
1708 "harness_pass": False,
1709 "baseline_pass": False,
1710 "harness_tokens": harness_total_tokens,
1711 "baseline_tokens": baseline_total_tokens,
1712 "harness_duration_ms": harness_result.duration_ms,
1713 "baseline_duration_ms": baseline_result.duration_ms,
1714 "confidence": 0.0,
1715 "reason": "Blind evaluation failed",
1716 "error": "evaluation_error",
1717 }
1718 self._ab_results.append(item)
1719 self._ab_item_index += 1
1721 return harness_result, None
1723 def _run_baseline_arm(
1724 self,
1725 skill_command: str,
1726 state: StateConfig,
1727 on_usage: UsageCallback | None = None,
1728 ) -> ActionResult:
1729 """Run a single-shot baseline skill invocation with no eval gates.
1731 Args:
1732 skill_command: Slash command for the baseline skill (e.g., "/ll:some-skill")
1733 state: State configuration (for timeout)
1734 on_usage: Optional callback for token capture
1736 Returns:
1737 ActionResult with output, exit code, and duration
1738 """
1739 start = _now_ms()
1740 timeout = state.timeout or self.fsm.default_timeout or 3600
1741 try:
1742 completed = run_claude_command(
1743 command=skill_command,
1744 timeout=timeout,
1745 on_usage=on_usage,
1746 )
1747 return ActionResult(
1748 output=completed.stdout,
1749 stderr=completed.stderr,
1750 exit_code=completed.returncode,
1751 duration_ms=_now_ms() - start,
1752 )
1753 except subprocess.TimeoutExpired:
1754 return ActionResult(
1755 output="",
1756 stderr="Baseline action timed out",
1757 exit_code=124,
1758 duration_ms=timeout * 1000,
1759 )
1761 def _run_action_or_route(
1762 self, state: StateConfig, ctx: InterpolationContext
1763 ) -> tuple[ActionResult | None, str | None]:
1764 """Run the state action, routing unhandled exceptions to on_error.
1766 Returns (action_result, routed_target). ``routed_target`` is a non-None
1767 next-state string only when an exception was raised AND ``state.on_error``
1768 is defined; in that case ``action_result`` is None. When no on_error is
1769 set, the exception is re-raised for the top-level ``run()`` handler.
1770 """
1771 assert state.action is not None # caller-guarded
1772 try:
1773 return self._run_action(state.action, state, ctx), None
1774 except Exception as exc:
1775 if state.on_error:
1776 self._emit(
1777 "action_error",
1778 {
1779 "state": self.current_state,
1780 "error": str(exc),
1781 "route": "on_error",
1782 },
1783 )
1784 return None, interpolate(state.on_error, ctx)
1785 raise
1787 def _build_context(self) -> InterpolationContext:
1788 """Build interpolation context for current state.
1790 Returns:
1791 InterpolationContext with all runtime values
1792 """
1793 from little_loops.fsm.interpolation import interpolate_dict
1795 ctx = InterpolationContext(
1796 context=self.fsm.context,
1797 captured=self.captured,
1798 prev=self.prev_result,
1799 result=None,
1800 state_name=self.current_state,
1801 iteration=self.iteration,
1802 loop_name=self.fsm.name,
1803 started_at=self.started_at,
1804 elapsed_ms=_now_ms() - self.start_time_ms + self.elapsed_offset_ms,
1805 messages=list(self.messages),
1806 )
1808 # Populate param namespace from fragment bindings (if this state uses a fragment)
1809 state = self.fsm.states.get(self.current_state)
1810 if state is not None and (state.fragment_bindings or state.fragment_parameters):
1811 resolved = interpolate_dict(state.fragment_bindings, ctx)
1812 # Apply declared defaults for unbound optional parameters
1813 for param_name, param_spec in state.fragment_parameters.items():
1814 if (
1815 param_name not in resolved
1816 and not param_spec.required
1817 and param_spec.default is not None
1818 ):
1819 resolved[param_name] = param_spec.default
1820 # Runtime enforcement: required parameters must be bound
1821 for param_name, param_spec in state.fragment_parameters.items():
1822 if param_spec.required and param_name not in resolved:
1823 raise ValueError(
1824 f"Fragment '{state.fragment_name}' requires parameter '{param_name}' "
1825 f"but it is not bound in 'with'"
1826 )
1827 ctx.param = resolved
1829 return ctx
1831 def _emit(self, event: str, data: dict[str, Any]) -> None:
1832 """Emit an event via the callback."""
1833 self.event_callback(
1834 {
1835 "event": event,
1836 "ts": _iso_now(),
1837 **data,
1838 }
1839 )
1841 def _handle_rate_limit(self, state: StateConfig, state_name: str) -> tuple[bool, str | None]:
1842 """Handle a detected 429/rate-limit action outcome.
1844 Implements the two-tier retry ladder:
1845 1. Short-burst tier: up to ``max_rate_limit_retries`` attempts with
1846 exponential backoff (``rate_limit_backoff_base_seconds * 2^n + jitter``).
1847 2. Long-wait tier: walks ``rate_limit_long_wait_ladder`` with index capped
1848 at the last entry, accumulating ``total_wait_seconds``.
1850 Routes to ``on_rate_limit_exhausted`` (falling back to ``on_error``) only
1851 once ``total_wait_seconds >= rate_limit_max_wait_seconds``. Emits
1852 ``rate_limit_exhausted`` on routing (including tier counters) and
1853 ``rate_limit_storm`` when consecutive exhaustions reach the threshold.
1855 Returns:
1856 (handled, target). ``handled=True`` means the caller should return
1857 ``target`` directly (in-place retry uses ``state_name``; exhaustion
1858 uses the routed target). ``handled=False`` should not occur for the
1859 current rate-limit classification path but is reserved for future
1860 extensions.
1861 """
1862 _short_max = (
1863 state.max_rate_limit_retries
1864 if state.max_rate_limit_retries is not None
1865 else _DEFAULT_RATE_LIMIT_RETRIES
1866 )
1867 _backoff_base = (
1868 state.rate_limit_backoff_base_seconds
1869 if state.rate_limit_backoff_base_seconds is not None
1870 else _DEFAULT_RATE_LIMIT_BACKOFF_BASE
1871 )
1872 _max_wait = (
1873 state.rate_limit_max_wait_seconds
1874 if state.rate_limit_max_wait_seconds is not None
1875 else _DEFAULT_RATE_LIMIT_MAX_WAIT_SECONDS
1876 )
1877 _ladder = (
1878 state.rate_limit_long_wait_ladder
1879 if state.rate_limit_long_wait_ladder is not None
1880 else _DEFAULT_RATE_LIMIT_LONG_WAIT_LADDER
1881 )
1883 record = self._rate_limit_retries.get(state_name)
1884 if record is None:
1885 record = {
1886 "short_retries": 0,
1887 "long_retries": 0,
1888 "total_wait_seconds": 0.0,
1889 "first_seen_at": time.time(),
1890 }
1891 self._rate_limit_retries[state_name] = record
1893 short_retries = int(record.get("short_retries", 0))
1894 long_retries = int(record.get("long_retries", 0))
1895 total_wait = float(record.get("total_wait_seconds", 0.0))
1897 if short_retries < _short_max:
1898 # Short-burst tier — exponential backoff with jitter. Budget is not
1899 # checked here; short-tier always advances to long-wait on exhaustion.
1900 short_retries += 1
1901 record["short_retries"] = short_retries
1902 _sleep = _backoff_base * (2 ** (short_retries - 1)) + random.uniform(0, _backoff_base)
1903 if self._circuit is not None:
1904 self._circuit.record_rate_limit(_sleep)
1905 total_wait += self._interruptible_sleep(_sleep)
1906 record["total_wait_seconds"] = total_wait
1907 return True, state_name # retry in place
1909 # Long-wait tier — walk ladder with capped index.
1910 long_retries += 1
1911 record["long_retries"] = long_retries
1912 _idx = min(long_retries - 1, len(_ladder) - 1)
1913 _wait = float(_ladder[_idx])
1914 if self._circuit is not None:
1915 self._circuit.record_rate_limit(_wait)
1916 _tier_start = time.time()
1917 _deadline = _tier_start + _wait
1918 _total_wait_before_tier = total_wait
1919 total_wait += self._interruptible_sleep(
1920 _wait,
1921 on_heartbeat=lambda elapsed: self._emit(
1922 RATE_LIMIT_WAITING_EVENT,
1923 {
1924 "state": state_name,
1925 "elapsed_seconds": elapsed,
1926 "next_attempt_at": _deadline,
1927 "total_waited_seconds": _total_wait_before_tier + elapsed,
1928 "budget_seconds": _max_wait,
1929 "tier": "long_wait",
1930 },
1931 ),
1932 )
1933 record["total_wait_seconds"] = total_wait
1934 if total_wait >= _max_wait:
1935 return True, self._exhaust_rate_limit(state, state_name, record)
1936 return True, state_name # retry in place
1938 def _maybe_wait_for_circuit(self, state: StateConfig) -> None:
1939 """Pre-action circuit-breaker check: sleep until shared 429 recovery.
1941 Skips quietly when no circuit is injected, when the state's action is not
1942 an LLM-quota consumer, or when the circuit has no active recovery window.
1943 """
1944 if self._circuit is None:
1945 return
1946 if self._action_mode(state) != "prompt":
1947 return
1948 recovery = self._circuit.get_estimated_recovery()
1949 if recovery is None:
1950 return
1951 wait = recovery - time.time()
1952 if wait > 0:
1953 self._interruptible_sleep(wait)
1955 def _interruptible_sleep(
1956 self,
1957 duration: float,
1958 on_heartbeat: Callable[[float], None] | None = None,
1959 ) -> float:
1960 """Sleep for up to ``duration`` seconds in 100ms ticks, exiting promptly
1961 on ``_shutdown_requested``. Returns the actual elapsed seconds so callers
1962 can accumulate wall-clock time spent in rate-limit waits.
1964 If ``on_heartbeat`` is provided, it is invoked with the elapsed seconds
1965 roughly every ``_RATE_LIMIT_HEARTBEAT_INTERVAL`` seconds so UIs can show
1966 live progress during long waits. The short-tier call site intentionally
1967 omits the callback to preserve backward-compatible silent behavior.
1968 """
1969 if duration <= 0:
1970 return 0.0
1971 _start = time.time()
1972 _deadline = _start + duration
1973 last_heartbeat = _start
1974 while time.time() < _deadline:
1975 if self._shutdown_requested:
1976 break
1977 time.sleep(min(0.1, _deadline - time.time()))
1978 if on_heartbeat is not None:
1979 _now = time.time()
1980 if _now - last_heartbeat >= _RATE_LIMIT_HEARTBEAT_INTERVAL:
1981 on_heartbeat(_now - _start)
1982 last_heartbeat = _now
1983 return time.time() - _start
1985 def _exhaust_rate_limit(
1986 self, state: StateConfig, state_name: str, record: dict[str, Any]
1987 ) -> str | None:
1988 """Finalize rate-limit exhaustion: emit event, storm detection, and
1989 return the routed target. Pops the per-state record and is called only
1990 once the wall-clock budget is spent.
1991 """
1992 self._rate_limit_retries.pop(state_name, None)
1993 target = state.on_rate_limit_exhausted or state.on_error
1994 self._emit(
1995 RATE_LIMIT_EXHAUSTED_EVENT,
1996 {
1997 "state": state_name,
1998 "retries": int(record.get("short_retries", 0)) + int(record.get("long_retries", 0)),
1999 "short_retries": int(record.get("short_retries", 0)),
2000 "long_retries": int(record.get("long_retries", 0)),
2001 "total_wait_seconds": float(record.get("total_wait_seconds", 0.0)),
2002 "next": target,
2003 },
2004 )
2005 self._consecutive_rate_limit_exhaustions += 1
2006 if self._consecutive_rate_limit_exhaustions >= _RATE_LIMIT_STORM_THRESHOLD:
2007 self._emit(
2008 RATE_LIMIT_STORM_EVENT,
2009 {
2010 "state": state_name,
2011 "count": self._consecutive_rate_limit_exhaustions,
2012 },
2013 )
2014 return target
2016 def _handle_api_error(self, state: StateConfig, state_name: str) -> tuple[bool, str | None]:
2017 """Handle a detected API server error with short-burst flat backoff.
2019 Unlike ``_handle_rate_limit``, uses a flat backoff with no long-wait tier and
2020 falls through to normal FSM routing after ``_DEFAULT_API_ERROR_RETRIES`` attempts
2021 so transient infrastructure hiccups don't permanently misdirect the loop.
2023 Returns:
2024 ``(True, state_name)`` to retry the state in place, or
2025 ``(False, None)`` when the retry budget is exhausted (caller falls
2026 through to normal verdict routing).
2027 """
2028 record = self._api_error_retries.setdefault(state_name, {"retries": 0, "total_wait": 0.0})
2029 if record["retries"] >= _DEFAULT_API_ERROR_RETRIES:
2030 self._api_error_retries.pop(state_name, None)
2031 self._emit("api_error_exhausted", {"state": state_name, "retries": record["retries"]})
2032 return False, None
2033 record["retries"] += 1
2034 slept = self._interruptible_sleep(_DEFAULT_API_ERROR_BACKOFF)
2035 record["total_wait"] += slept
2036 self._emit(
2037 "api_error_retry",
2038 {
2039 "state": state_name,
2040 "attempt": record["retries"],
2041 "backoff": _DEFAULT_API_ERROR_BACKOFF,
2042 },
2043 )
2044 return True, state_name
2046 def _finish(self, terminated_by: str, error: str | None = None) -> ExecutionResult:
2047 """Finalize execution and return result."""
2048 payload: dict[str, Any] = {
2049 "final_state": self.current_state,
2050 "iterations": self.iteration,
2051 "terminated_by": terminated_by,
2052 }
2053 if error is not None:
2054 payload["error"] = error
2055 self._emit("loop_complete", payload)
2057 # FEAT-1822: Write ab.json if baseline comparison results exist
2058 if self._ab_results:
2059 try:
2060 from little_loops.ab_writer import calculate_ab_summary, write_ab_json
2062 run_dir = self.fsm.context.get("run_dir", "")
2063 if run_dir:
2064 summary = calculate_ab_summary(self._ab_results)
2065 write_ab_json(summary, run_dir)
2066 self._emit(
2067 "ab_summary",
2068 {
2069 "harness_pass_rate": summary.harness_pass_rate,
2070 "baseline_pass_rate": summary.baseline_pass_rate,
2071 "delta": summary.delta,
2072 "item_count": len(summary.per_item),
2073 },
2074 )
2075 except Exception:
2076 pass # Non-fatal: loop still completes
2078 return ExecutionResult(
2079 final_state=self.current_state,
2080 iterations=self.iteration,
2081 terminated_by=terminated_by,
2082 duration_ms=_now_ms() - self.start_time_ms + self.elapsed_offset_ms,
2083 captured=self.captured,
2084 error=error,
2085 messages=list(self.messages),
2086 )
2088 def _handle_handoff(self, signal: DetectedSignal) -> ExecutionResult:
2089 """Handle a detected handoff signal.
2091 Emits a handoff_detected event and optionally invokes the handoff handler.
2093 Args:
2094 signal: The detected handoff signal
2096 Returns:
2097 ExecutionResult with handoff information
2098 """
2099 self._emit(
2100 "handoff_detected",
2101 {
2102 "state": self.current_state,
2103 "iteration": self.iteration,
2104 "continuation": signal.payload,
2105 },
2106 )
2108 # Invoke handler if configured
2109 if self.handoff_handler:
2110 result = self.handoff_handler.handle(self.fsm.name, signal.payload)
2111 if result.spawned_process is not None:
2112 self._emit(
2113 "handoff_spawned",
2114 {
2115 "pid": result.spawned_process.pid,
2116 "state": self.current_state,
2117 },
2118 )
2120 return ExecutionResult(
2121 final_state=self.current_state,
2122 iterations=self.iteration,
2123 terminated_by="handoff",
2124 duration_ms=_now_ms() - self.start_time_ms + self.elapsed_offset_ms,
2125 captured=self.captured,
2126 handoff=True,
2127 continuation_prompt=signal.payload,
2128 )
2131def _extract_skill_from_action(action: str) -> str:
2132 """Extract the base skill name from an action string.
2134 For slash commands like "/ll:some-skill --arg value", returns "/ll:some-skill".
2135 For other actions, returns the action as-is.
2136 """
2137 stripped = action.strip()
2138 if stripped.startswith("/"):
2139 parts = stripped.split(maxsplit=1)
2140 return parts[0]
2141 return stripped