Coverage for little_loops / fsm / executor.py: 14%
548 statements
« prev ^ index » next coverage.py v7.12.0, created at 2026-05-11 00:29 -0500
1"""FSM Executor - Runtime engine for FSM loop execution.
3This module provides the execution engine that runs FSM loops:
4- Executes actions (shell commands or slash commands)
5- Evaluates results using appropriate evaluators
6- Routes to next states based on verdicts
7- Tracks iteration count and enforces limits
8- Manages captured variables and context
9"""
11from __future__ import annotations
13import json
14import random
15import subprocess
16import threading
17import time
18from collections.abc import Callable
19from dataclasses import dataclass
20from datetime import UTC, datetime
21from pathlib import Path
22from typing import Any
24from little_loops.fsm.evaluators import (
25 EvaluationResult,
26 evaluate,
27 evaluate_exit_code,
28 evaluate_llm_structured,
29 evaluate_mcp_result,
30)
31from little_loops.fsm.handoff_handler import HandoffHandler
32from little_loops.fsm.interpolation import (
33 InterpolationContext,
34 InterpolationError,
35 interpolate,
36 interpolate_dict,
37)
38from little_loops.fsm.rate_limit_circuit import RateLimitCircuit
39from little_loops.fsm.runners import (
40 ActionRunner,
41 DefaultActionRunner,
42 SimulationActionRunner, # noqa: F401 — re-exported for backward compatibility
43 _now_ms,
44)
45from little_loops.fsm.schema import FSMLoop, StateConfig
46from little_loops.fsm.signal_detector import DetectedSignal, SignalDetector
47from little_loops.fsm.types import ActionResult, Evaluator, EventCallback, ExecutionResult
48from little_loops.issue_lifecycle import FailureType, classify_failure
49from little_loops.session_log import get_current_session_jsonl
51# Maximum number of per-state rate-limit retries before emitting rate_limit_exhausted.
52_DEFAULT_RATE_LIMIT_RETRIES: int = 3
53# Base backoff in seconds; actual sleep = base * 2^(attempt-1) + uniform(0, base).
54_DEFAULT_RATE_LIMIT_BACKOFF_BASE: int = 30
55# Total wall-clock budget (seconds) across short + long tiers before routing to
56# on_rate_limit_exhausted. Mirrors RateLimitsConfig.max_wait_seconds default (6h).
57_DEFAULT_RATE_LIMIT_MAX_WAIT_SECONDS: int = 21600
58# Long-wait tier ladder (seconds), walked once the short-tier budget is spent.
59# Mirrors RateLimitsConfig.long_wait_ladder default.
60_DEFAULT_RATE_LIMIT_LONG_WAIT_LADDER: list[int] = [300, 900, 1800, 3600]
61# Event name emitted when rate-limit retries are exhausted.
62RATE_LIMIT_EXHAUSTED_EVENT: str = "rate_limit_exhausted"
63# Event name emitted when consecutive rate-limit exhaustions reach the storm threshold.
64RATE_LIMIT_STORM_EVENT: str = "rate_limit_storm"
65# Event name emitted every ~60s during a long-wait rate-limit sleep so UIs can show live progress.
66RATE_LIMIT_WAITING_EVENT: str = "rate_limit_waiting"
67# Interval (seconds) between rate_limit_waiting heartbeat emissions during long-wait sleeps.
68_RATE_LIMIT_HEARTBEAT_INTERVAL: float = 60.0
69# Number of consecutive rate_limit_exhausted events that constitute a storm.
70_RATE_LIMIT_STORM_THRESHOLD: int = 3
71# Progressive throttle defaults: calls 1..normal_max pass through, at warn_max emit warning,
72# at hard_max route to on_throttle_hard, beyond hard_max hard-stop.
73_DEFAULT_THROTTLE_NORMAL_MAX: int = 3
74_DEFAULT_THROTTLE_WARN_MAX: int = 8
75_DEFAULT_THROTTLE_HARD_MAX: int = 12
76# Event names for progressive tool-call throttling within a single state visit.
77THROTTLE_WARN_EVENT: str = "throttle_warn"
78THROTTLE_HARD_EVENT: str = "throttle_hard"
79THROTTLE_STOP_EVENT: str = "throttle_stop"
80# Action types that consume LLM quota and are gated by the shared circuit breaker.
81# `_action_mode()` collapses both to "prompt"; the frozenset documents intent.
82LLM_ACTION_TYPES: frozenset[str] = frozenset({"slash_command", "prompt"})
83# Maximum per-state API server error retries before falling through to normal routing.
84_DEFAULT_API_ERROR_RETRIES: int = 2
85# Flat backoff in seconds between API server error retries (no exponential ladder).
86_DEFAULT_API_ERROR_BACKOFF: int = 30
89def _iso_now() -> str:
90 """Get current time as ISO 8601 string."""
91 return datetime.now(UTC).isoformat()
94@dataclass
95class RouteContext:
96 """Context passed to before_route / after_route interceptors."""
98 state_name: str
99 state: StateConfig
100 verdict: str
101 action_result: ActionResult | None
102 eval_result: EvaluationResult | None
103 ctx: InterpolationContext
104 iteration: int
107@dataclass
108class RouteDecision:
109 """Returned by before_route to redirect or veto a routing transition.
111 Return semantics for before_route:
112 None (implicit) → passthrough, routing proceeds normally
113 RouteDecision("state") → redirect, bypass _route() and use "state" directly
114 RouteDecision(None) → veto, _execute_state() returns None → _finish("error")
115 """
117 next_state: str | None # str → redirect; None → veto
120class FSMExecutor:
121 """Execute an FSM loop.
123 The executor runs an FSM from its initial state until:
124 - A terminal state is reached
125 - max_iterations is exceeded
126 - A timeout occurs
127 - A shutdown signal is received
128 - An unrecoverable error occurs
130 Events are emitted via the callback for observability.
131 """
133 def __init__(
134 self,
135 fsm: FSMLoop,
136 event_callback: EventCallback | None = None,
137 action_runner: ActionRunner | None = None,
138 signal_detector: SignalDetector | None = None,
139 handoff_handler: HandoffHandler | None = None,
140 loops_dir: Path | None = None,
141 circuit: RateLimitCircuit | None = None,
142 ):
143 """Initialize the executor.
145 Args:
146 fsm: The FSM loop to execute
147 event_callback: Optional callback for events
148 action_runner: Optional custom action runner (for testing)
149 signal_detector: Optional signal detector for output parsing
150 handoff_handler: Optional handler for handoff signals
151 loops_dir: Base directory for resolving sub-loop references
152 circuit: Optional shared rate-limit circuit breaker for 429 coordination
153 """
154 self.fsm = fsm
155 self.event_callback = event_callback or (lambda _: None)
156 self.action_runner: ActionRunner = action_runner or DefaultActionRunner()
157 self.signal_detector = signal_detector
158 self.handoff_handler = handoff_handler
159 self.loops_dir = loops_dir
160 self._circuit = circuit
162 # Runtime state
163 self.current_state = fsm.initial
164 self.iteration = 0
165 self.captured: dict[str, dict[str, Any]] = {}
166 self.prev_result: dict[str, Any] | None = None
167 self.started_at = ""
168 self.start_time_ms = 0
169 self.elapsed_offset_ms = (
170 0 # milliseconds from segments before current run (set by PersistentExecutor on resume)
171 )
173 # Shutdown flag for graceful signal handling
174 self._shutdown_requested = False
176 # Currently running MCP subprocess (set by _run_subprocess, cleared in finally).
177 # Enables external shutdown code to kill the process on SIGTERM.
178 self._current_process: subprocess.Popen[str] | None = None
180 # Pending handoff signal (set by _run_action, checked by main loop)
181 self._pending_handoff: DetectedSignal | None = None
183 # Pending error payload from FATAL_ERROR signal (set by _run_action, checked by main loop)
184 self._pending_error: str | None = None
186 # Per-state retry tracking for max_retries support.
187 # _retry_counts[state_name] = number of consecutive re-entries into that state.
188 # Incremented each time we enter the same state as the previous iteration.
189 # Reset when a different state is entered, or after retry exhaustion.
190 self._retry_counts: dict[str, int] = {}
191 # State entered in the previous iteration (None on first iteration or after resume).
192 self._prev_state: str | None = None
194 # BUG-1226: true between emitting a `route` event and entering the
195 # target state. Gates the "flush one pending shell state on timeout"
196 # behavior so we only flush when there is actually a pending state.
197 self._just_routed: bool = False
199 # Per-state rate-limit retry tracking (parallel to _retry_counts).
200 # _rate_limit_retries[state_name] = dict-of-record:
201 # {
202 # "short_retries": int, # attempts in short-burst tier
203 # "long_retries": int, # attempts in long-wait tier
204 # "total_wait_seconds": float,
205 # "first_seen_at": float | None, # epoch timestamp of first 429
206 # }
207 # Incremented inside _handle_rate_limit on each detected rate-limit response.
208 # Reset when the state completes without a rate-limit, or after exhaustion.
209 self._rate_limit_retries: dict[str, dict[str, Any]] = {}
211 # Consecutive rate_limit_exhausted emissions across all states. Reset on any
212 # successful non-rate-limited state transition. When this reaches
213 # _RATE_LIMIT_STORM_THRESHOLD, a RATE_LIMIT_STORM event is emitted.
214 self._consecutive_rate_limit_exhaustions: int = 0
216 # Per-state API server error retry tracking (parallel to _rate_limit_retries).
217 # _api_error_retries[state_name] = {"retries": int, "total_wait": float}
218 # Reset when the state completes without a server error, or after exhaustion.
219 self._api_error_retries: dict[str, dict[str, Any]] = {}
221 # Per-state tool-call throttle counter. Counts successive action executions within
222 # a single continuous state visit. Reset on state exit; NOT serialized to LoopState
223 # (throttle counts measure instantaneous visit-level activity, not cumulative retries).
224 self._throttle_counts: dict[str, int] = {}
226 # Per-edge revisit counter for cycle detection.
227 # _edge_revisit_counts["from_state->to_state"] = number of times that edge has fired.
228 # When any edge exceeds max_edge_revisits, the loop terminates with cycle_detected.
229 self._edge_revisit_counts: dict[str, int] = {}
231 # Nesting depth for sub-loop event forwarding (0 = top-level, 1+ = sub-loop).
232 # Set by the parent executor when constructing child executors.
233 self._depth: int = 0
235 # Extension hook registries — populated by wire_extensions()
236 self._contributed_actions: dict[str, ActionRunner] = {}
237 self._contributed_evaluators: dict[str, Evaluator] = {}
238 self._interceptors: list[Any] = []
240 def request_shutdown(self) -> None:
241 """Request graceful shutdown of the executor.
243 Sets a flag that will be checked at the start of each iteration,
244 allowing the loop to exit cleanly after the current state completes.
245 """
246 self._shutdown_requested = True
248 def run(self) -> ExecutionResult:
249 """Execute the FSM until terminal state or limits reached.
251 Returns:
252 ExecutionResult with final state and execution metadata
253 """
254 self.started_at = _iso_now()
255 self.start_time_ms = _now_ms()
257 self._emit("loop_start", {"loop": self.fsm.name})
259 try:
260 while True:
261 # Check shutdown request (signal handling)
262 if self._shutdown_requested:
263 return self._finish("signal")
265 # Check iteration limit
266 if self.iteration >= self.fsm.max_iterations:
267 return self._finish("max_iterations")
269 # Check timeout
270 if self.fsm.timeout:
271 elapsed = _now_ms() - self.start_time_ms + self.elapsed_offset_ms
272 if elapsed > self.fsm.timeout * 1000:
273 # BUG-1226: if timeout fires in the race window between
274 # a `route` event and `state_enter`, flush one pending
275 # shell-action state before honoring the timeout so its
276 # side effect (e.g. copying a handshake flag) is not
277 # silently lost. Bounded to shell actions — slash
278 # commands and sub-loops would violate the timeout
279 # budget. Single-step: no cascade.
280 if self._just_routed:
281 pending = self.fsm.states.get(self.current_state)
282 if (
283 pending is not None
284 and not pending.terminal
285 and pending.loop is None
286 and pending.action is not None
287 and self._action_mode(pending) == "shell"
288 ):
289 self._flush_pending_shell_state(pending)
290 return self._finish("timeout")
292 # Get current state config
293 state_config = self.fsm.states[self.current_state]
295 # Update per-state retry tracking based on transition from previous iteration.
296 # If re-entering the same state consecutively, increment retry count.
297 # If entering a different state, clear the previous state's retry count.
298 if self._prev_state is not None:
299 if self.current_state == self._prev_state:
300 self._retry_counts[self.current_state] = (
301 self._retry_counts.get(self.current_state, 0) + 1
302 )
303 else:
304 self._retry_counts.pop(self._prev_state, None)
305 self._throttle_counts.pop(self._prev_state, None)
307 # Check terminal
308 if state_config.terminal:
309 # Handle maintain mode - restart loop instead of terminating
310 if self.fsm.maintain:
311 self.iteration += 1
312 maintain_target = state_config.on_maintain or self.fsm.initial
313 self._emit(
314 "route",
315 {
316 "from": self.current_state,
317 "to": maintain_target,
318 "reason": "maintain",
319 },
320 )
321 self._prev_state = self.current_state
322 self.current_state = maintain_target
323 self._just_routed = True
324 continue
325 return self._finish("terminal")
327 # Check per-state retry limit. If the consecutive re-entry count exceeds
328 # max_retries, skip execution and route to on_retry_exhausted instead.
329 if state_config.max_retries is not None:
330 retry_count = self._retry_counts.get(self.current_state, 0)
331 if retry_count > state_config.max_retries:
332 # on_retry_exhausted is guaranteed non-None by validation when
333 # max_retries is set, but we fall back to an error if misconfigured.
334 exhausted_state: str = state_config.on_retry_exhausted or ""
335 if not exhausted_state:
336 return self._finish(
337 "error",
338 error=f"State '{self.current_state}' exceeded max_retries "
339 "but on_retry_exhausted is not set",
340 )
341 self._emit(
342 "retry_exhausted",
343 {
344 "state": self.current_state,
345 "retries": retry_count,
346 "next": exhausted_state,
347 },
348 )
349 self._retry_counts.pop(self.current_state, None)
350 self._prev_state = self.current_state
351 self.current_state = exhausted_state
352 continue
354 self.iteration += 1
355 self._just_routed = False
356 self._emit(
357 "state_enter",
358 {
359 "state": self.current_state,
360 "iteration": self.iteration,
361 },
362 )
364 # Execute state
365 next_state = self._execute_state(state_config)
367 # Check for pending error signal (FATAL_ERROR)
368 if self._pending_error is not None:
369 return self._finish("error", error=self._pending_error)
371 # Check for pending handoff signal
372 if self._pending_handoff:
373 return self._handle_handoff(self._pending_handoff)
375 # Handle maintain mode
376 if next_state is None and self.fsm.maintain:
377 next_state = state_config.on_maintain or self.fsm.initial
379 # SIGKILL in _execute_state sets shutdown flag and returns None
380 if next_state is None and self._shutdown_requested:
381 return self._finish("signal")
383 if next_state is None:
384 return self._finish("error", error="No valid transition")
386 # At this point next_state is guaranteed to be str
387 resolved_next: str = next_state
389 self._emit(
390 "route",
391 {
392 "from": self.current_state,
393 "to": resolved_next,
394 },
395 )
397 # Per-edge revisit tracking for cycle detection.
398 edge_key = f"{self.current_state}->{resolved_next}"
399 self._edge_revisit_counts[edge_key] = (
400 self._edge_revisit_counts.get(edge_key, 0) + 1
401 )
402 if self._edge_revisit_counts[edge_key] > self.fsm.max_edge_revisits:
403 self._emit("cycle_detected", {
404 "edge": edge_key,
405 "from": self.current_state,
406 "to": resolved_next,
407 "count": self._edge_revisit_counts[edge_key],
408 "max": self.fsm.max_edge_revisits,
409 })
410 return self._finish(
411 "cycle_detected",
412 error=f"Cycle detected: edge {edge_key} traversed "
413 f"{self._edge_revisit_counts[edge_key]} times "
414 f"(limit: {self.fsm.max_edge_revisits})",
415 )
417 self._prev_state = self.current_state
418 self.current_state = resolved_next
419 self._just_routed = True
421 # Interruptible backoff sleep between iterations
422 if self.fsm.backoff and self.fsm.backoff > 0:
423 deadline = time.time() + self.fsm.backoff
424 while time.time() < deadline:
425 if self._shutdown_requested:
426 break
427 time.sleep(min(0.1, deadline - time.time()))
429 except InterpolationError as exc:
430 return self._finish(
431 "error",
432 error=(
433 f"Missing context variable in state '{self.current_state}': {exc}. "
434 f"Run with: ll-loop run {self.fsm.name} --context KEY=VALUE"
435 ),
436 )
437 except Exception as exc:
438 return self._finish("error", error=str(exc))
440 def _execute_sub_loop(self, state: StateConfig, ctx: InterpolationContext) -> str | None:
441 """Execute a sub-loop state by loading and running a child FSM.
443 Args:
444 state: The state configuration with loop field set
445 ctx: Interpolation context for routing
447 Returns:
448 Next state name based on child loop verdict, or None
449 """
450 from little_loops.cli.loop._helpers import resolve_loop_path
451 from little_loops.fsm.validation import load_and_validate
453 assert state.loop is not None # guarded by caller
454 loop_name = interpolate(state.loop, ctx)
455 loop_path = resolve_loop_path(loop_name, self.loops_dir or Path(".loops"))
456 child_fsm, _ = load_and_validate(loop_path)
458 # Bind child context: explicit with: bindings take precedence over legacy passthrough
459 if state.with_:
460 from little_loops.fsm.interpolation import interpolate_dict
462 resolved = interpolate_dict(state.with_, ctx)
463 # Apply declared defaults for unbound optional parameters
464 for param_name, param_spec in child_fsm.parameters.items():
465 if (
466 param_name not in resolved
467 and not param_spec.required
468 and param_spec.default is not None
469 ):
470 resolved[param_name] = param_spec.default
471 # Runtime check: required parameters must be present after interpolation
472 for param_name, param_spec in child_fsm.parameters.items():
473 if param_spec.required and param_name not in resolved:
474 raise ValueError(
475 f"Sub-loop '{state.loop}' requires parameter '{param_name}' "
476 f"but it is not bound in 'with'"
477 )
478 # Merge: child's own context block provides base; with: bindings override
479 child_fsm.context = {**child_fsm.context, **resolved}
480 elif state.context_passthrough:
481 # Extract .output strings from capture result dicts so ${context.key} resolves
482 # to the plain output string (e.g. "ENH-123") rather than the full capture object.
483 captured_as_context = {
484 k: v["output"] if isinstance(v, dict) and "exit_code" in v else v
485 for k, v in self.captured.items()
486 }
487 child_fsm.context = {**self.fsm.context, **captured_as_context, **child_fsm.context}
489 depth = self._depth + 1
490 child_events: list[dict] = []
492 def _sub_event_callback(event: dict) -> None:
493 child_events.append(event)
494 # Only inject depth if not already set by a deeper nested sub-loop
495 if "depth" not in event:
496 self.event_callback({**event, "depth": depth})
497 else:
498 self.event_callback(event)
500 child_executor = FSMExecutor(
501 child_fsm,
502 action_runner=self.action_runner,
503 loops_dir=self.loops_dir,
504 event_callback=_sub_event_callback,
505 circuit=self._circuit,
506 )
507 child_executor._depth = depth # propagate depth for further nesting
509 # Clamp child timeout to parent's remaining wall-clock budget so a slow sub-loop
510 # can't silently consume the parent's deadline with no recourse for the parent FSM.
511 if self.fsm.timeout:
512 elapsed_ms = _now_ms() - self.start_time_ms + self.elapsed_offset_ms
513 remaining_s = max(1, int((self.fsm.timeout * 1000 - elapsed_ms) // 1000))
514 if child_fsm.timeout is None or child_fsm.timeout > remaining_s:
515 child_fsm.timeout = remaining_s
517 child_result = child_executor.run()
519 # Capture child event stream as a JSON-lines string if the state declares a capture key
520 if state.capture and child_events:
521 import json as _json
523 self.captured[state.capture] = {
524 "output": "\n".join(_json.dumps(e) for e in child_events),
525 "exit_code": None,
526 }
528 # Merge child captures back into parent under the state name
529 if (state.context_passthrough or state.with_) and child_executor.captured:
530 self.captured[self.current_state] = child_executor.captured
532 # Route based on child termination reason and terminal state name
533 if child_result.terminated_by == "terminal":
534 if child_result.final_state == "done":
535 return interpolate(state.on_yes, ctx) if state.on_yes else None
536 else:
537 # Reached a non-done terminal (e.g. "failed") → failure
538 return interpolate(state.on_no, ctx) if state.on_no else None
539 elif child_result.terminated_by == "error":
540 # Runtime child failure (not a YAML load error)
541 if state.on_error:
542 return interpolate(state.on_error, ctx)
543 return interpolate(state.on_no, ctx) if state.on_no else None
544 else:
545 # max_iterations, timeout, signal — all are failure
546 return interpolate(state.on_no, ctx) if state.on_no else None
548 def _check_throttle(self, state: StateConfig, state_name: str) -> str | None:
549 """Increment the per-state tool-call counter and enforce throttle thresholds.
551 Called after every action execution within a state visit. Returns the forced
552 next-state name when the hard threshold is reached, or None when execution
553 should continue normally (warn events are emitted but do not redirect).
555 Sets self._pending_error and returns "__STOP__" when the call count exceeds
556 hard_max with no on_throttle_hard target — the caller must propagate this as
557 a None return from _execute_state so the main loop detects _pending_error.
558 """
559 count = self._throttle_counts.get(state_name, 0) + 1
560 self._throttle_counts[state_name] = count
562 throttle = state.throttle
563 normal_max = (
564 throttle.normal_max
565 if (throttle and throttle.normal_max is not None)
566 else _DEFAULT_THROTTLE_NORMAL_MAX
567 )
568 warn_max = (
569 throttle.warn_max
570 if (throttle and throttle.warn_max is not None)
571 else _DEFAULT_THROTTLE_WARN_MAX
572 )
573 hard_max = (
574 throttle.hard_max
575 if (throttle and throttle.hard_max is not None)
576 else _DEFAULT_THROTTLE_HARD_MAX
577 )
579 if count == warn_max:
580 self._emit(
581 THROTTLE_WARN_EVENT,
582 {
583 "state": state_name,
584 "count": count,
585 "normal_max": normal_max,
586 "warn_max": warn_max,
587 "hard_max": hard_max,
588 },
589 )
591 # States with type="learning" (FEAT-1283) are exempt from hard_max — they
592 # legitimately make N calls per visit (one per unproven target).
593 if state.type == "learning":
594 return None
596 if count == hard_max:
597 next_target = state.on_throttle_hard or state.on_error
598 self._emit(
599 THROTTLE_HARD_EVENT,
600 {"state": state_name, "count": count, "hard_max": hard_max, "next": next_target},
601 )
602 return next_target
604 if count > hard_max:
605 self._emit(
606 THROTTLE_STOP_EVENT,
607 {"state": state_name, "count": count, "hard_max": hard_max},
608 )
609 self._pending_error = (
610 f"Throttle stop: state '{state_name}' exceeded hard_max={hard_max} "
611 "tool calls with no on_throttle_hard target"
612 )
613 return "__STOP__"
615 return None
617 def _execute_state(self, state: StateConfig) -> str | None:
618 """Execute a single state and return next state name.
620 Args:
621 state: The state configuration to execute
623 Returns:
624 Next state name, or None if no valid transition
625 """
626 # Build interpolation context
627 ctx = self._build_context()
629 # Dispatch to sub-loop handler if this is a sub-loop state
630 if state.loop is not None:
631 try:
632 return self._execute_sub_loop(state, ctx)
633 except (FileNotFoundError, ValueError):
634 if state.on_error:
635 return interpolate(state.on_error, ctx)
636 raise
638 # Handle unconditional transition
639 if state.next:
640 if state.action:
641 self._maybe_wait_for_circuit(state)
642 result, routed = self._run_action_or_route(state, ctx)
643 if routed is not None:
644 return routed
645 throttle_next = self._check_throttle(state, self.current_state)
646 if throttle_next == "__STOP__":
647 return None
648 if throttle_next is not None:
649 return throttle_next
650 assert result is not None
651 self.prev_result = {
652 "output": result.output,
653 "exit_code": result.exit_code,
654 "state": self.current_state,
655 }
656 if result.exit_code is not None and result.exit_code < 0:
657 # Process killed by signal — do not silently advance via next
658 if state.on_error:
659 return interpolate(state.on_error, ctx)
660 self.request_shutdown()
661 return None
662 # Non-zero exit: if on_error is defined, treat next as success path only
663 if result.exit_code != 0 and state.on_error:
664 return interpolate(state.on_error, ctx)
665 return interpolate(state.next, ctx)
667 # Execute action if present
668 action_result = None
669 if state.action:
670 self._maybe_wait_for_circuit(state)
671 action_result, routed = self._run_action_or_route(state, ctx)
672 if routed is not None:
673 return routed
674 throttle_next = self._check_throttle(state, self.current_state)
675 if throttle_next == "__STOP__":
676 return None
677 if throttle_next is not None:
678 return throttle_next
680 # Evaluate
681 eval_result = self._evaluate(state, action_result, ctx)
682 self.prev_result = {
683 "output": action_result.output if action_result else "",
684 "exit_code": action_result.exit_code if action_result else 0,
685 "state": self.current_state,
686 }
688 # Update context with result for routing interpolation
689 if eval_result:
690 ctx.result = {
691 "verdict": eval_result.verdict,
692 "details": eval_result.details,
693 }
695 # Route based on verdict
696 verdict = eval_result.verdict if eval_result else "yes"
697 route_ctx = RouteContext(
698 state_name=self.current_state,
699 state=state,
700 verdict=verdict,
701 action_result=action_result,
702 eval_result=eval_result,
703 ctx=ctx,
704 iteration=self.iteration,
705 )
706 # 429 / rate-limit detection — runs before interceptors so an in-place retry
707 # returns early without dispatching to registered before_route hooks.
708 if action_result is not None:
709 _combined = (action_result.output or "") + "\n" + (action_result.stderr or "")
710 _failure_type, _reason = classify_failure(_combined, action_result.exit_code)
711 if _failure_type == FailureType.TRANSIENT and (
712 "rate limit" in _reason.lower() or "quota" in _reason.lower()
713 ):
714 _handled, _target = self._handle_rate_limit(state, route_ctx.state_name)
715 if _handled:
716 return _target
717 elif _failure_type == FailureType.TRANSIENT and "api server error" in _reason.lower():
718 _handled, _target = self._handle_api_error(state, route_ctx.state_name)
719 if _handled:
720 return _target
721 # exhausted — fall through to normal verdict routing
722 else:
723 # Not rate-limited or server-error: reset counters so future transients start fresh.
724 self._rate_limit_retries.pop(route_ctx.state_name, None)
725 self._consecutive_rate_limit_exhaustions = 0
726 self._api_error_retries.pop(route_ctx.state_name, None)
728 for interceptor in self._interceptors:
729 if hasattr(interceptor, "before_route"):
730 decision = interceptor.before_route(route_ctx)
731 if isinstance(decision, RouteDecision):
732 if decision.next_state is None:
733 return None # veto
734 return decision.next_state # redirect — bypass _route()
735 next_state = self._route(state, verdict, ctx)
736 for interceptor in self._interceptors:
737 if hasattr(interceptor, "after_route"):
738 interceptor.after_route(route_ctx)
739 return next_state
741 def _run_action(
742 self,
743 action_template: str,
744 state: StateConfig,
745 ctx: InterpolationContext,
746 ) -> ActionResult:
747 """Execute action and optionally capture result.
749 Args:
750 action_template: Action string (may contain variables)
751 state: State configuration
752 ctx: Interpolation context
754 Returns:
755 ActionResult with output and exit code
756 """
757 action = interpolate(action_template, ctx)
758 action_mode = self._action_mode(state)
760 self._emit("action_start", {"action": action, "is_prompt": action_mode == "prompt"})
762 def _on_line(line: str) -> None:
763 self._emit("action_output", {"line": line})
765 if action_mode == "mcp_tool":
766 # Direct MCP tool call — bypass action_runner entirely
767 interpolated_params = interpolate_dict(state.params, ctx) if state.params else {}
768 cmd = ["mcp-call", action, json.dumps(interpolated_params)]
769 result = self._run_subprocess(
770 cmd,
771 timeout=state.timeout or self.fsm.default_timeout or 30,
772 on_output_line=_on_line,
773 )
774 elif action_mode == "contributed":
775 assert (
776 state.action_type is not None
777 ) # guaranteed by _action_mode returning "contributed"
778 runner = self._contributed_actions[state.action_type]
779 result = runner.run(
780 action,
781 timeout=state.timeout or self.fsm.default_timeout or 3600,
782 is_slash_command=False,
783 on_output_line=_on_line,
784 )
785 else:
786 result = self.action_runner.run(
787 action,
788 timeout=state.timeout or self.fsm.default_timeout or 3600,
789 is_slash_command=action_mode == "prompt",
790 on_output_line=_on_line,
791 agent=state.agent if action_mode == "prompt" else None,
792 tools=state.tools if action_mode == "prompt" else None,
793 )
795 preview = result.output[-2000:].strip() if result.output else None
796 payload: dict[str, Any] = {
797 "exit_code": result.exit_code,
798 "duration_ms": result.duration_ms,
799 "output_preview": preview,
800 "is_prompt": action_mode == "prompt",
801 }
802 if action_mode == "prompt":
803 session_jsonl = get_current_session_jsonl()
804 payload["session_jsonl"] = str(session_jsonl) if session_jsonl else None
805 self._emit("action_complete", payload)
807 # Capture if requested
808 if state.capture:
809 self.captured[state.capture] = {
810 "output": result.output.rstrip("\n\r"),
811 "stderr": result.stderr,
812 "exit_code": result.exit_code,
813 "duration_ms": result.duration_ms,
814 }
816 # Check for signals in output
817 if self.signal_detector:
818 signal = self.signal_detector.detect_first(result.output)
819 if signal:
820 if signal.signal_type == "handoff":
821 self._pending_handoff = signal
822 elif signal.signal_type == "error":
823 self._pending_error = signal.payload
824 elif signal.signal_type == "stop":
825 self.request_shutdown()
827 return result
829 def _run_subprocess(
830 self,
831 cmd: list[str],
832 timeout: int,
833 on_output_line: Any | None = None,
834 ) -> ActionResult:
835 """Run a subprocess directly and return ActionResult.
837 Follows the same Popen + stderr-drain-thread pattern as DefaultActionRunner.
839 Args:
840 cmd: Command and arguments to execute
841 timeout: Timeout in seconds
842 on_output_line: Optional callback for each stdout line
844 Returns:
845 ActionResult with output, stderr, exit_code, duration_ms
846 """
847 start = _now_ms()
848 process = subprocess.Popen(
849 cmd,
850 stdout=subprocess.PIPE,
851 stderr=subprocess.PIPE,
852 text=True,
853 )
854 self._current_process = process
855 output_chunks: list[str] = []
856 stderr_chunks: list[str] = []
858 def _drain_stderr() -> None:
859 assert process.stderr is not None
860 for line in process.stderr:
861 stderr_chunks.append(line)
863 stderr_thread = threading.Thread(target=_drain_stderr, daemon=True)
864 stderr_thread.start()
866 try:
867 for line in process.stdout: # type: ignore[union-attr]
868 output_chunks.append(line)
869 if on_output_line:
870 on_output_line(line.rstrip())
871 process.wait(timeout=timeout)
872 except subprocess.TimeoutExpired:
873 process.kill()
874 process.wait()
875 stderr_thread.join(timeout=5)
876 return ActionResult(
877 output="".join(output_chunks),
878 stderr="".join(stderr_chunks) or "MCP call timed out",
879 exit_code=124,
880 duration_ms=timeout * 1000,
881 )
882 finally:
883 self._current_process = None
884 stderr_thread.join(timeout=5)
885 return ActionResult(
886 output="".join(output_chunks),
887 stderr="".join(stderr_chunks),
888 exit_code=process.returncode,
889 duration_ms=_now_ms() - start,
890 )
892 def _evaluate(
893 self,
894 state: StateConfig,
895 action_result: ActionResult | None,
896 ctx: InterpolationContext,
897 ) -> EvaluationResult | None:
898 """Evaluate action result.
900 Args:
901 state: State configuration
902 action_result: Result from action execution (may be None)
903 ctx: Interpolation context
905 Returns:
906 EvaluationResult, or None if no evaluation needed
907 """
908 if state.evaluate is None:
909 # Default evaluation based on action type
910 if action_result:
911 action_mode = self._action_mode(state)
913 if action_mode == "mcp_tool":
914 # MCP tool call: use mcp_result evaluator
915 result = evaluate_mcp_result(action_result.output, action_result.exit_code)
916 elif action_mode == "prompt":
917 # Slash command or prompt: use LLM evaluation
918 if not self.fsm.llm.enabled:
919 result = EvaluationResult(
920 verdict="error",
921 details={"error": "LLM evaluation disabled via --no-llm"},
922 )
923 else:
924 result = evaluate_llm_structured(
925 action_result.output,
926 model=self.fsm.llm.model,
927 max_tokens=self.fsm.llm.max_tokens,
928 timeout=self.fsm.llm.timeout,
929 )
930 else:
931 # Shell command: use exit code
932 result = evaluate_exit_code(action_result.exit_code)
934 self._emit(
935 "evaluate",
936 {
937 "type": "default",
938 "verdict": result.verdict,
939 **result.details,
940 },
941 )
942 return result
943 return None
945 # Explicit evaluation config
946 raw_output = action_result.output if action_result else ""
947 if state.evaluate.source:
948 try:
949 eval_input = interpolate(state.evaluate.source, ctx)
950 except InterpolationError:
951 eval_input = raw_output
952 else:
953 eval_input = raw_output
955 if state.evaluate.type in self._contributed_evaluators:
956 result = self._contributed_evaluators[state.evaluate.type](
957 state.evaluate,
958 eval_input,
959 action_result.exit_code if action_result else 0,
960 ctx,
961 )
962 elif state.evaluate.type == "llm_structured" and not self.fsm.llm.enabled:
963 result = EvaluationResult(
964 verdict="error",
965 details={"error": "LLM evaluation disabled via --no-llm"},
966 )
967 else:
968 result = evaluate(
969 config=state.evaluate,
970 output=eval_input,
971 exit_code=action_result.exit_code if action_result else 0,
972 context=ctx,
973 )
975 self._emit(
976 "evaluate",
977 {
978 "type": state.evaluate.type,
979 "verdict": result.verdict,
980 **result.details,
981 },
982 )
984 return result
986 def _route(
987 self,
988 state: StateConfig,
989 verdict: str,
990 ctx: InterpolationContext,
991 ) -> str | None:
992 """Determine next state from verdict.
994 Resolution order (from design doc):
995 1. next (unconditional) - handled before this method
996 2. route (full routing table)
997 3. on_success/on_failure/on_error (shorthand)
998 4. terminal - handled in main loop
999 5. error
1001 Args:
1002 state: State configuration
1003 verdict: Verdict string from evaluation
1004 ctx: Interpolation context
1006 Returns:
1007 Next state name, or None if no valid route
1008 """
1009 if state.route:
1010 routes = state.route.routes
1011 if verdict in routes:
1012 return self._resolve_route(routes[verdict], ctx)
1013 if state.route.default:
1014 return self._resolve_route(state.route.default, ctx)
1015 if verdict == "error" and state.route.error:
1016 return self._resolve_route(state.route.error, ctx)
1017 return None
1019 # Shorthand routing
1020 if verdict == "yes" and state.on_yes:
1021 return self._resolve_route(state.on_yes, ctx)
1022 if verdict == "no" and state.on_no:
1023 return self._resolve_route(state.on_no, ctx)
1024 if verdict == "error" and state.on_error:
1025 return self._resolve_route(state.on_error, ctx)
1026 if verdict == "partial" and state.on_partial:
1027 return self._resolve_route(state.on_partial, ctx)
1028 if verdict == "blocked" and state.on_blocked:
1029 return self._resolve_route(state.on_blocked, ctx)
1031 # Dynamic on_<verdict> shorthands from extra_routes
1032 if verdict in state.extra_routes:
1033 return self._resolve_route(state.extra_routes[verdict], ctx)
1035 return None
1037 def _resolve_route(self, route: str, ctx: InterpolationContext) -> str:
1038 """Resolve route target, handling special tokens.
1040 Args:
1041 route: Route target string
1042 ctx: Interpolation context
1044 Returns:
1045 Resolved state name
1046 """
1047 if route == "$current":
1048 return self.current_state
1049 return interpolate(route, ctx)
1051 def _flush_pending_shell_state(self, state: StateConfig) -> None:
1052 """Execute a pending shell-action state's action before honoring a
1053 wall-clock timeout. BUG-1226: closes the narrow race between emitting
1054 a `route` event and `state_enter` so handshake states (e.g. autodev's
1055 ``copy_broke_down``) do not silently drop their side effect when the
1056 timeout fires in that window. Single-step: we run the action but do
1057 not follow its routing — ``final_state`` stays as the flushed state.
1058 """
1059 assert state.action is not None # guarded by caller
1060 self.iteration += 1
1061 self._just_routed = False
1062 self._emit(
1063 "state_enter",
1064 {
1065 "state": self.current_state,
1066 "iteration": self.iteration,
1067 "flushed": True,
1068 },
1069 )
1070 ctx = self._build_context()
1071 try:
1072 self._run_action(state.action, state, ctx)
1073 except Exception:
1074 # Deliberately swallow — the timeout is being honored regardless
1075 # of whether the flushed action succeeded.
1076 pass
1078 def _action_mode(self, state: StateConfig) -> str:
1079 """Return execution mode for the state: 'prompt', 'shell', or 'mcp_tool'."""
1080 if state.action_type == "mcp_tool":
1081 return "mcp_tool"
1082 if state.action_type in ("prompt", "slash_command"):
1083 return "prompt"
1084 if state.action_type == "shell":
1085 return "shell"
1086 if state.action_type in self._contributed_actions:
1087 return "contributed"
1088 # Heuristic: / prefix = slash_command (prompt mode)
1089 if state.action is not None and state.action.startswith("/"):
1090 return "prompt"
1091 return "shell"
1093 def _run_action_or_route(
1094 self, state: StateConfig, ctx: InterpolationContext
1095 ) -> tuple[ActionResult | None, str | None]:
1096 """Run the state action, routing unhandled exceptions to on_error.
1098 Returns (action_result, routed_target). ``routed_target`` is a non-None
1099 next-state string only when an exception was raised AND ``state.on_error``
1100 is defined; in that case ``action_result`` is None. When no on_error is
1101 set, the exception is re-raised for the top-level ``run()`` handler.
1102 """
1103 assert state.action is not None # caller-guarded
1104 try:
1105 return self._run_action(state.action, state, ctx), None
1106 except Exception as exc:
1107 if state.on_error:
1108 self._emit(
1109 "action_error",
1110 {
1111 "state": self.current_state,
1112 "error": str(exc),
1113 "route": "on_error",
1114 },
1115 )
1116 return None, interpolate(state.on_error, ctx)
1117 raise
1119 def _build_context(self) -> InterpolationContext:
1120 """Build interpolation context for current state.
1122 Returns:
1123 InterpolationContext with all runtime values
1124 """
1125 return InterpolationContext(
1126 context=self.fsm.context,
1127 captured=self.captured,
1128 prev=self.prev_result,
1129 result=None,
1130 state_name=self.current_state,
1131 iteration=self.iteration,
1132 loop_name=self.fsm.name,
1133 started_at=self.started_at,
1134 elapsed_ms=_now_ms() - self.start_time_ms + self.elapsed_offset_ms,
1135 )
1137 def _emit(self, event: str, data: dict[str, Any]) -> None:
1138 """Emit an event via the callback."""
1139 self.event_callback(
1140 {
1141 "event": event,
1142 "ts": _iso_now(),
1143 **data,
1144 }
1145 )
1147 def _handle_rate_limit(self, state: StateConfig, state_name: str) -> tuple[bool, str | None]:
1148 """Handle a detected 429/rate-limit action outcome.
1150 Implements the two-tier retry ladder:
1151 1. Short-burst tier: up to ``max_rate_limit_retries`` attempts with
1152 exponential backoff (``rate_limit_backoff_base_seconds * 2^n + jitter``).
1153 2. Long-wait tier: walks ``rate_limit_long_wait_ladder`` with index capped
1154 at the last entry, accumulating ``total_wait_seconds``.
1156 Routes to ``on_rate_limit_exhausted`` (falling back to ``on_error``) only
1157 once ``total_wait_seconds >= rate_limit_max_wait_seconds``. Emits
1158 ``rate_limit_exhausted`` on routing (including tier counters) and
1159 ``rate_limit_storm`` when consecutive exhaustions reach the threshold.
1161 Returns:
1162 (handled, target). ``handled=True`` means the caller should return
1163 ``target`` directly (in-place retry uses ``state_name``; exhaustion
1164 uses the routed target). ``handled=False`` should not occur for the
1165 current rate-limit classification path but is reserved for future
1166 extensions.
1167 """
1168 _short_max = (
1169 state.max_rate_limit_retries
1170 if state.max_rate_limit_retries is not None
1171 else _DEFAULT_RATE_LIMIT_RETRIES
1172 )
1173 _backoff_base = (
1174 state.rate_limit_backoff_base_seconds
1175 if state.rate_limit_backoff_base_seconds is not None
1176 else _DEFAULT_RATE_LIMIT_BACKOFF_BASE
1177 )
1178 _max_wait = (
1179 state.rate_limit_max_wait_seconds
1180 if state.rate_limit_max_wait_seconds is not None
1181 else _DEFAULT_RATE_LIMIT_MAX_WAIT_SECONDS
1182 )
1183 _ladder = (
1184 state.rate_limit_long_wait_ladder
1185 if state.rate_limit_long_wait_ladder is not None
1186 else _DEFAULT_RATE_LIMIT_LONG_WAIT_LADDER
1187 )
1189 record = self._rate_limit_retries.get(state_name)
1190 if record is None:
1191 record = {
1192 "short_retries": 0,
1193 "long_retries": 0,
1194 "total_wait_seconds": 0.0,
1195 "first_seen_at": time.time(),
1196 }
1197 self._rate_limit_retries[state_name] = record
1199 short_retries = int(record.get("short_retries", 0))
1200 long_retries = int(record.get("long_retries", 0))
1201 total_wait = float(record.get("total_wait_seconds", 0.0))
1203 if short_retries < _short_max:
1204 # Short-burst tier — exponential backoff with jitter. Budget is not
1205 # checked here; short-tier always advances to long-wait on exhaustion.
1206 short_retries += 1
1207 record["short_retries"] = short_retries
1208 _sleep = _backoff_base * (2 ** (short_retries - 1)) + random.uniform(0, _backoff_base)
1209 if self._circuit is not None:
1210 self._circuit.record_rate_limit(_sleep)
1211 total_wait += self._interruptible_sleep(_sleep)
1212 record["total_wait_seconds"] = total_wait
1213 return True, state_name # retry in place
1215 # Long-wait tier — walk ladder with capped index.
1216 long_retries += 1
1217 record["long_retries"] = long_retries
1218 _idx = min(long_retries - 1, len(_ladder) - 1)
1219 _wait = float(_ladder[_idx])
1220 if self._circuit is not None:
1221 self._circuit.record_rate_limit(_wait)
1222 _tier_start = time.time()
1223 _deadline = _tier_start + _wait
1224 _total_wait_before_tier = total_wait
1225 total_wait += self._interruptible_sleep(
1226 _wait,
1227 on_heartbeat=lambda elapsed: self._emit(
1228 RATE_LIMIT_WAITING_EVENT,
1229 {
1230 "state": state_name,
1231 "elapsed_seconds": elapsed,
1232 "next_attempt_at": _deadline,
1233 "total_waited_seconds": _total_wait_before_tier + elapsed,
1234 "budget_seconds": _max_wait,
1235 "tier": "long_wait",
1236 },
1237 ),
1238 )
1239 record["total_wait_seconds"] = total_wait
1240 if total_wait >= _max_wait:
1241 return True, self._exhaust_rate_limit(state, state_name, record)
1242 return True, state_name # retry in place
1244 def _maybe_wait_for_circuit(self, state: StateConfig) -> None:
1245 """Pre-action circuit-breaker check: sleep until shared 429 recovery.
1247 Skips quietly when no circuit is injected, when the state's action is not
1248 an LLM-quota consumer, or when the circuit has no active recovery window.
1249 """
1250 if self._circuit is None:
1251 return
1252 if self._action_mode(state) != "prompt":
1253 return
1254 recovery = self._circuit.get_estimated_recovery()
1255 if recovery is None:
1256 return
1257 wait = recovery - time.time()
1258 if wait > 0:
1259 self._interruptible_sleep(wait)
1261 def _interruptible_sleep(
1262 self,
1263 duration: float,
1264 on_heartbeat: Callable[[float], None] | None = None,
1265 ) -> float:
1266 """Sleep for up to ``duration`` seconds in 100ms ticks, exiting promptly
1267 on ``_shutdown_requested``. Returns the actual elapsed seconds so callers
1268 can accumulate wall-clock time spent in rate-limit waits.
1270 If ``on_heartbeat`` is provided, it is invoked with the elapsed seconds
1271 roughly every ``_RATE_LIMIT_HEARTBEAT_INTERVAL`` seconds so UIs can show
1272 live progress during long waits. The short-tier call site intentionally
1273 omits the callback to preserve backward-compatible silent behavior.
1274 """
1275 if duration <= 0:
1276 return 0.0
1277 _start = time.time()
1278 _deadline = _start + duration
1279 last_heartbeat = _start
1280 while time.time() < _deadline:
1281 if self._shutdown_requested:
1282 break
1283 time.sleep(min(0.1, _deadline - time.time()))
1284 if on_heartbeat is not None:
1285 _now = time.time()
1286 if _now - last_heartbeat >= _RATE_LIMIT_HEARTBEAT_INTERVAL:
1287 on_heartbeat(_now - _start)
1288 last_heartbeat = _now
1289 return time.time() - _start
1291 def _exhaust_rate_limit(
1292 self, state: StateConfig, state_name: str, record: dict[str, Any]
1293 ) -> str | None:
1294 """Finalize rate-limit exhaustion: emit event, storm detection, and
1295 return the routed target. Pops the per-state record and is called only
1296 once the wall-clock budget is spent.
1297 """
1298 self._rate_limit_retries.pop(state_name, None)
1299 target = state.on_rate_limit_exhausted or state.on_error
1300 self._emit(
1301 RATE_LIMIT_EXHAUSTED_EVENT,
1302 {
1303 "state": state_name,
1304 "retries": int(record.get("short_retries", 0)) + int(record.get("long_retries", 0)),
1305 "short_retries": int(record.get("short_retries", 0)),
1306 "long_retries": int(record.get("long_retries", 0)),
1307 "total_wait_seconds": float(record.get("total_wait_seconds", 0.0)),
1308 "next": target,
1309 },
1310 )
1311 self._consecutive_rate_limit_exhaustions += 1
1312 if self._consecutive_rate_limit_exhaustions >= _RATE_LIMIT_STORM_THRESHOLD:
1313 self._emit(
1314 RATE_LIMIT_STORM_EVENT,
1315 {
1316 "state": state_name,
1317 "count": self._consecutive_rate_limit_exhaustions,
1318 },
1319 )
1320 return target
1322 def _handle_api_error(self, state: StateConfig, state_name: str) -> tuple[bool, str | None]:
1323 """Handle a detected API server error with short-burst flat backoff.
1325 Unlike ``_handle_rate_limit``, uses a flat backoff with no long-wait tier and
1326 falls through to normal FSM routing after ``_DEFAULT_API_ERROR_RETRIES`` attempts
1327 so transient infrastructure hiccups don't permanently misdirect the loop.
1329 Returns:
1330 ``(True, state_name)`` to retry the state in place, or
1331 ``(False, None)`` when the retry budget is exhausted (caller falls
1332 through to normal verdict routing).
1333 """
1334 record = self._api_error_retries.setdefault(state_name, {"retries": 0, "total_wait": 0.0})
1335 if record["retries"] >= _DEFAULT_API_ERROR_RETRIES:
1336 self._api_error_retries.pop(state_name, None)
1337 self._emit("api_error_exhausted", {"state": state_name, "retries": record["retries"]})
1338 return False, None
1339 record["retries"] += 1
1340 slept = self._interruptible_sleep(_DEFAULT_API_ERROR_BACKOFF)
1341 record["total_wait"] += slept
1342 self._emit(
1343 "api_error_retry",
1344 {
1345 "state": state_name,
1346 "attempt": record["retries"],
1347 "backoff": _DEFAULT_API_ERROR_BACKOFF,
1348 },
1349 )
1350 return True, state_name
1352 def _finish(self, terminated_by: str, error: str | None = None) -> ExecutionResult:
1353 """Finalize execution and return result."""
1354 self._emit(
1355 "loop_complete",
1356 {
1357 "final_state": self.current_state,
1358 "iterations": self.iteration,
1359 "terminated_by": terminated_by,
1360 },
1361 )
1363 return ExecutionResult(
1364 final_state=self.current_state,
1365 iterations=self.iteration,
1366 terminated_by=terminated_by,
1367 duration_ms=_now_ms() - self.start_time_ms + self.elapsed_offset_ms,
1368 captured=self.captured,
1369 error=error,
1370 )
1372 def _handle_handoff(self, signal: DetectedSignal) -> ExecutionResult:
1373 """Handle a detected handoff signal.
1375 Emits a handoff_detected event and optionally invokes the handoff handler.
1377 Args:
1378 signal: The detected handoff signal
1380 Returns:
1381 ExecutionResult with handoff information
1382 """
1383 self._emit(
1384 "handoff_detected",
1385 {
1386 "state": self.current_state,
1387 "iteration": self.iteration,
1388 "continuation": signal.payload,
1389 },
1390 )
1392 # Invoke handler if configured
1393 if self.handoff_handler:
1394 result = self.handoff_handler.handle(self.fsm.name, signal.payload)
1395 if result.spawned_process is not None:
1396 self._emit(
1397 "handoff_spawned",
1398 {
1399 "pid": result.spawned_process.pid,
1400 "state": self.current_state,
1401 },
1402 )
1404 return ExecutionResult(
1405 final_state=self.current_state,
1406 iterations=self.iteration,
1407 terminated_by="handoff",
1408 duration_ms=_now_ms() - self.start_time_ms + self.elapsed_offset_ms,
1409 captured=self.captured,
1410 handoff=True,
1411 continuation_prompt=signal.payload,
1412 )