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