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