Coverage for little_loops / fsm / executor.py: 14%
334 statements
« prev ^ index » next coverage.py v7.12.0, created at 2026-04-11 23:20 -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 subprocess
15import threading
16import time
17from dataclasses import dataclass
18from datetime import UTC, datetime
19from pathlib import Path
20from typing import Any
22from little_loops.fsm.evaluators import (
23 EvaluationResult,
24 evaluate,
25 evaluate_exit_code,
26 evaluate_llm_structured,
27 evaluate_mcp_result,
28)
29from little_loops.fsm.handoff_handler import HandoffHandler
30from little_loops.fsm.interpolation import (
31 InterpolationContext,
32 InterpolationError,
33 interpolate,
34 interpolate_dict,
35)
36from little_loops.fsm.runners import (
37 ActionRunner,
38 DefaultActionRunner,
39 SimulationActionRunner, # noqa: F401 — re-exported for backward compatibility
40 _now_ms,
41)
42from little_loops.fsm.schema import FSMLoop, StateConfig
43from little_loops.fsm.signal_detector import DetectedSignal, SignalDetector
44from little_loops.fsm.types import ActionResult, Evaluator, EventCallback, ExecutionResult
45from little_loops.session_log import get_current_session_jsonl
48def _iso_now() -> str:
49 """Get current time as ISO 8601 string."""
50 return datetime.now(UTC).isoformat()
53@dataclass
54class RouteContext:
55 """Context passed to before_route / after_route interceptors."""
57 state_name: str
58 state: StateConfig
59 verdict: str
60 action_result: ActionResult | None
61 eval_result: EvaluationResult | None
62 ctx: InterpolationContext
63 iteration: int
66@dataclass
67class RouteDecision:
68 """Returned by before_route to redirect or veto a routing transition.
70 Return semantics for before_route:
71 None (implicit) → passthrough, routing proceeds normally
72 RouteDecision("state") → redirect, bypass _route() and use "state" directly
73 RouteDecision(None) → veto, _execute_state() returns None → _finish("error")
74 """
76 next_state: str | None # str → redirect; None → veto
79class FSMExecutor:
80 """Execute an FSM loop.
82 The executor runs an FSM from its initial state until:
83 - A terminal state is reached
84 - max_iterations is exceeded
85 - A timeout occurs
86 - A shutdown signal is received
87 - An unrecoverable error occurs
89 Events are emitted via the callback for observability.
90 """
92 def __init__(
93 self,
94 fsm: FSMLoop,
95 event_callback: EventCallback | None = None,
96 action_runner: ActionRunner | None = None,
97 signal_detector: SignalDetector | None = None,
98 handoff_handler: HandoffHandler | None = None,
99 loops_dir: Path | None = None,
100 ):
101 """Initialize the executor.
103 Args:
104 fsm: The FSM loop to execute
105 event_callback: Optional callback for events
106 action_runner: Optional custom action runner (for testing)
107 signal_detector: Optional signal detector for output parsing
108 handoff_handler: Optional handler for handoff signals
109 loops_dir: Base directory for resolving sub-loop references
110 """
111 self.fsm = fsm
112 self.event_callback = event_callback or (lambda _: None)
113 self.action_runner: ActionRunner = action_runner or DefaultActionRunner()
114 self.signal_detector = signal_detector
115 self.handoff_handler = handoff_handler
116 self.loops_dir = loops_dir
118 # Runtime state
119 self.current_state = fsm.initial
120 self.iteration = 0
121 self.captured: dict[str, dict[str, Any]] = {}
122 self.prev_result: dict[str, Any] | None = None
123 self.started_at = ""
124 self.start_time_ms = 0
125 self.elapsed_offset_ms = (
126 0 # milliseconds from segments before current run (set by PersistentExecutor on resume)
127 )
129 # Shutdown flag for graceful signal handling
130 self._shutdown_requested = False
132 # Currently running MCP subprocess (set by _run_subprocess, cleared in finally).
133 # Enables external shutdown code to kill the process on SIGTERM.
134 self._current_process: subprocess.Popen[str] | None = None
136 # Pending handoff signal (set by _run_action, checked by main loop)
137 self._pending_handoff: DetectedSignal | None = None
139 # Pending error payload from FATAL_ERROR signal (set by _run_action, checked by main loop)
140 self._pending_error: str | None = None
142 # Per-state retry tracking for max_retries support.
143 # _retry_counts[state_name] = number of consecutive re-entries into that state.
144 # Incremented each time we enter the same state as the previous iteration.
145 # Reset when a different state is entered, or after retry exhaustion.
146 self._retry_counts: dict[str, int] = {}
147 # State entered in the previous iteration (None on first iteration or after resume).
148 self._prev_state: str | None = None
150 # Nesting depth for sub-loop event forwarding (0 = top-level, 1+ = sub-loop).
151 # Set by the parent executor when constructing child executors.
152 self._depth: int = 0
154 # Extension hook registries — populated by wire_extensions()
155 self._contributed_actions: dict[str, ActionRunner] = {}
156 self._contributed_evaluators: dict[str, Evaluator] = {}
157 self._interceptors: list[Any] = []
159 def request_shutdown(self) -> None:
160 """Request graceful shutdown of the executor.
162 Sets a flag that will be checked at the start of each iteration,
163 allowing the loop to exit cleanly after the current state completes.
164 """
165 self._shutdown_requested = True
167 def run(self) -> ExecutionResult:
168 """Execute the FSM until terminal state or limits reached.
170 Returns:
171 ExecutionResult with final state and execution metadata
172 """
173 self.started_at = _iso_now()
174 self.start_time_ms = _now_ms()
176 self._emit("loop_start", {"loop": self.fsm.name})
178 try:
179 while True:
180 # Check shutdown request (signal handling)
181 if self._shutdown_requested:
182 return self._finish("signal")
184 # Check iteration limit
185 if self.iteration >= self.fsm.max_iterations:
186 return self._finish("max_iterations")
188 # Check timeout
189 if self.fsm.timeout:
190 elapsed = _now_ms() - self.start_time_ms + self.elapsed_offset_ms
191 if elapsed > self.fsm.timeout * 1000:
192 return self._finish("timeout")
194 # Get current state config
195 state_config = self.fsm.states[self.current_state]
197 # Update per-state retry tracking based on transition from previous iteration.
198 # If re-entering the same state consecutively, increment retry count.
199 # If entering a different state, clear the previous state's retry count.
200 if self._prev_state is not None:
201 if self.current_state == self._prev_state:
202 self._retry_counts[self.current_state] = (
203 self._retry_counts.get(self.current_state, 0) + 1
204 )
205 else:
206 self._retry_counts.pop(self._prev_state, None)
208 # Check terminal
209 if state_config.terminal:
210 # Handle maintain mode - restart loop instead of terminating
211 if self.fsm.maintain:
212 self.iteration += 1
213 maintain_target = state_config.on_maintain or self.fsm.initial
214 self._emit(
215 "route",
216 {
217 "from": self.current_state,
218 "to": maintain_target,
219 "reason": "maintain",
220 },
221 )
222 self._prev_state = self.current_state
223 self.current_state = maintain_target
224 continue
225 return self._finish("terminal")
227 # Check per-state retry limit. If the consecutive re-entry count exceeds
228 # max_retries, skip execution and route to on_retry_exhausted instead.
229 if state_config.max_retries is not None:
230 retry_count = self._retry_counts.get(self.current_state, 0)
231 if retry_count > state_config.max_retries:
232 # on_retry_exhausted is guaranteed non-None by validation when
233 # max_retries is set, but we fall back to an error if misconfigured.
234 exhausted_state: str = state_config.on_retry_exhausted or ""
235 if not exhausted_state:
236 return self._finish(
237 "error",
238 error=f"State '{self.current_state}' exceeded max_retries "
239 "but on_retry_exhausted is not set",
240 )
241 self._emit(
242 "retry_exhausted",
243 {
244 "state": self.current_state,
245 "retries": retry_count,
246 "next": exhausted_state,
247 },
248 )
249 self._retry_counts.pop(self.current_state, None)
250 self._prev_state = self.current_state
251 self.current_state = exhausted_state
252 continue
254 self.iteration += 1
255 self._emit(
256 "state_enter",
257 {
258 "state": self.current_state,
259 "iteration": self.iteration,
260 },
261 )
263 # Execute state
264 next_state = self._execute_state(state_config)
266 # Check for pending error signal (FATAL_ERROR)
267 if self._pending_error is not None:
268 return self._finish("error", error=self._pending_error)
270 # Check for pending handoff signal
271 if self._pending_handoff:
272 return self._handle_handoff(self._pending_handoff)
274 # Handle maintain mode
275 if next_state is None and self.fsm.maintain:
276 next_state = state_config.on_maintain or self.fsm.initial
278 # SIGKILL in _execute_state sets shutdown flag and returns None
279 if next_state is None and self._shutdown_requested:
280 return self._finish("signal")
282 if next_state is None:
283 return self._finish("error", error="No valid transition")
285 # At this point next_state is guaranteed to be str
286 resolved_next: str = next_state
288 self._emit(
289 "route",
290 {
291 "from": self.current_state,
292 "to": resolved_next,
293 },
294 )
296 self._prev_state = self.current_state
297 self.current_state = resolved_next
299 # Interruptible backoff sleep between iterations
300 if self.fsm.backoff and self.fsm.backoff > 0:
301 deadline = time.time() + self.fsm.backoff
302 while time.time() < deadline:
303 if self._shutdown_requested:
304 break
305 time.sleep(min(0.1, deadline - time.time()))
307 except InterpolationError as exc:
308 return self._finish(
309 "error",
310 error=(
311 f"Missing context variable in state '{self.current_state}': {exc}. "
312 f"Run with: ll-loop run {self.fsm.name} --context KEY=VALUE"
313 ),
314 )
315 except Exception as exc:
316 return self._finish("error", error=str(exc))
318 def _execute_sub_loop(self, state: StateConfig, ctx: InterpolationContext) -> str | None:
319 """Execute a sub-loop state by loading and running a child FSM.
321 Args:
322 state: The state configuration with loop field set
323 ctx: Interpolation context for routing
325 Returns:
326 Next state name based on child loop verdict, or None
327 """
328 from little_loops.cli.loop._helpers import resolve_loop_path
329 from little_loops.fsm.validation import load_and_validate
331 assert state.loop is not None # guarded by caller
332 loop_path = resolve_loop_path(state.loop, self.loops_dir or Path(".loops"))
333 child_fsm, _ = load_and_validate(loop_path)
335 # Pass parent context to child if requested
336 if state.context_passthrough:
337 # Extract .output strings from capture result dicts so ${context.key} resolves
338 # to the plain output string (e.g. "ENH-123") rather than the full capture object.
339 captured_as_context = {
340 k: v["output"] if isinstance(v, dict) and "exit_code" in v else v
341 for k, v in self.captured.items()
342 }
343 child_fsm.context = {**self.fsm.context, **captured_as_context, **child_fsm.context}
345 depth = self._depth + 1
347 def _sub_event_callback(event: dict) -> None:
348 # Only inject depth if not already set by a deeper nested sub-loop
349 if "depth" not in event:
350 self.event_callback({**event, "depth": depth})
351 else:
352 self.event_callback(event)
354 child_executor = FSMExecutor(
355 child_fsm,
356 action_runner=self.action_runner,
357 loops_dir=self.loops_dir,
358 event_callback=_sub_event_callback,
359 )
360 child_executor._depth = depth # propagate depth for further nesting
361 child_result = child_executor.run()
363 # Merge child captures back into parent under the state name
364 if state.context_passthrough and child_executor.captured:
365 self.captured[self.current_state] = child_executor.captured
367 # Route based on child termination reason and terminal state name
368 if child_result.terminated_by == "terminal":
369 if child_result.final_state == "done":
370 return interpolate(state.on_yes, ctx) if state.on_yes else None
371 else:
372 # Reached a non-done terminal (e.g. "failed") → failure
373 return interpolate(state.on_no, ctx) if state.on_no else None
374 elif child_result.terminated_by == "error":
375 # Runtime child failure (not a YAML load error)
376 if state.on_error:
377 return interpolate(state.on_error, ctx)
378 return interpolate(state.on_no, ctx) if state.on_no else None
379 else:
380 # max_iterations, timeout, signal — all are failure
381 return interpolate(state.on_no, ctx) if state.on_no else None
383 def _execute_state(self, state: StateConfig) -> str | None:
384 """Execute a single state and return next state name.
386 Args:
387 state: The state configuration to execute
389 Returns:
390 Next state name, or None if no valid transition
391 """
392 # Build interpolation context
393 ctx = self._build_context()
395 # Dispatch to sub-loop handler if this is a sub-loop state
396 if state.loop is not None:
397 try:
398 return self._execute_sub_loop(state, ctx)
399 except (FileNotFoundError, ValueError):
400 if state.on_error:
401 return interpolate(state.on_error, ctx)
402 raise
404 # Handle unconditional transition
405 if state.next:
406 if state.action:
407 result = self._run_action(state.action, state, ctx)
408 self.prev_result = {
409 "output": result.output,
410 "exit_code": result.exit_code,
411 "state": self.current_state,
412 }
413 if result.exit_code is not None and result.exit_code < 0:
414 # Process killed by signal — do not silently advance via next
415 if state.on_error:
416 return interpolate(state.on_error, ctx)
417 self.request_shutdown()
418 return None
419 # Non-zero exit: if on_error is defined, treat next as success path only
420 if result.exit_code != 0 and state.on_error:
421 return interpolate(state.on_error, ctx)
422 return interpolate(state.next, ctx)
424 # Execute action if present
425 action_result = None
426 if state.action:
427 action_result = self._run_action(state.action, state, ctx)
429 # Evaluate
430 eval_result = self._evaluate(state, action_result, ctx)
431 self.prev_result = {
432 "output": action_result.output if action_result else "",
433 "exit_code": action_result.exit_code if action_result else 0,
434 "state": self.current_state,
435 }
437 # Update context with result for routing interpolation
438 if eval_result:
439 ctx.result = {
440 "verdict": eval_result.verdict,
441 "details": eval_result.details,
442 }
444 # Route based on verdict
445 verdict = eval_result.verdict if eval_result else "yes"
446 route_ctx = RouteContext(
447 state_name=self.current_state,
448 state=state,
449 verdict=verdict,
450 action_result=action_result,
451 eval_result=eval_result,
452 ctx=ctx,
453 iteration=self.iteration,
454 )
455 for interceptor in self._interceptors:
456 if hasattr(interceptor, "before_route"):
457 decision = interceptor.before_route(route_ctx)
458 if isinstance(decision, RouteDecision):
459 if decision.next_state is None:
460 return None # veto
461 return decision.next_state # redirect — bypass _route()
462 next_state = self._route(state, verdict, ctx)
463 for interceptor in self._interceptors:
464 if hasattr(interceptor, "after_route"):
465 interceptor.after_route(route_ctx)
466 return next_state
468 def _run_action(
469 self,
470 action_template: str,
471 state: StateConfig,
472 ctx: InterpolationContext,
473 ) -> ActionResult:
474 """Execute action and optionally capture result.
476 Args:
477 action_template: Action string (may contain variables)
478 state: State configuration
479 ctx: Interpolation context
481 Returns:
482 ActionResult with output and exit code
483 """
484 action = interpolate(action_template, ctx)
485 action_mode = self._action_mode(state)
487 self._emit("action_start", {"action": action, "is_prompt": action_mode == "prompt"})
489 def _on_line(line: str) -> None:
490 self._emit("action_output", {"line": line})
492 if action_mode == "mcp_tool":
493 # Direct MCP tool call — bypass action_runner entirely
494 interpolated_params = interpolate_dict(state.params, ctx) if state.params else {}
495 cmd = ["mcp-call", action, json.dumps(interpolated_params)]
496 result = self._run_subprocess(
497 cmd,
498 timeout=state.timeout or self.fsm.default_timeout or 30,
499 on_output_line=_on_line,
500 )
501 elif action_mode == "contributed":
502 assert (
503 state.action_type is not None
504 ) # guaranteed by _action_mode returning "contributed"
505 runner = self._contributed_actions[state.action_type]
506 result = runner.run(
507 action,
508 timeout=state.timeout or self.fsm.default_timeout or 3600,
509 is_slash_command=False,
510 on_output_line=_on_line,
511 )
512 else:
513 result = self.action_runner.run(
514 action,
515 timeout=state.timeout or self.fsm.default_timeout or 3600,
516 is_slash_command=action_mode == "prompt",
517 on_output_line=_on_line,
518 agent=state.agent if action_mode == "prompt" else None,
519 tools=state.tools if action_mode == "prompt" else None,
520 )
522 preview = result.output[-2000:].strip() if result.output else None
523 payload: dict[str, Any] = {
524 "exit_code": result.exit_code,
525 "duration_ms": result.duration_ms,
526 "output_preview": preview,
527 "is_prompt": action_mode == "prompt",
528 }
529 if action_mode == "prompt":
530 session_jsonl = get_current_session_jsonl()
531 payload["session_jsonl"] = str(session_jsonl) if session_jsonl else None
532 self._emit("action_complete", payload)
534 # Capture if requested
535 if state.capture:
536 self.captured[state.capture] = {
537 "output": result.output.rstrip("\n\r"),
538 "stderr": result.stderr,
539 "exit_code": result.exit_code,
540 "duration_ms": result.duration_ms,
541 }
543 # Check for signals in output
544 if self.signal_detector:
545 signal = self.signal_detector.detect_first(result.output)
546 if signal:
547 if signal.signal_type == "handoff":
548 self._pending_handoff = signal
549 elif signal.signal_type == "error":
550 self._pending_error = signal.payload
551 elif signal.signal_type == "stop":
552 self.request_shutdown()
554 return result
556 def _run_subprocess(
557 self,
558 cmd: list[str],
559 timeout: int,
560 on_output_line: Any | None = None,
561 ) -> ActionResult:
562 """Run a subprocess directly and return ActionResult.
564 Follows the same Popen + stderr-drain-thread pattern as DefaultActionRunner.
566 Args:
567 cmd: Command and arguments to execute
568 timeout: Timeout in seconds
569 on_output_line: Optional callback for each stdout line
571 Returns:
572 ActionResult with output, stderr, exit_code, duration_ms
573 """
574 start = _now_ms()
575 process = subprocess.Popen(
576 cmd,
577 stdout=subprocess.PIPE,
578 stderr=subprocess.PIPE,
579 text=True,
580 )
581 self._current_process = process
582 output_chunks: list[str] = []
583 stderr_chunks: list[str] = []
585 def _drain_stderr() -> None:
586 assert process.stderr is not None
587 for line in process.stderr:
588 stderr_chunks.append(line)
590 stderr_thread = threading.Thread(target=_drain_stderr, daemon=True)
591 stderr_thread.start()
593 try:
594 for line in process.stdout: # type: ignore[union-attr]
595 output_chunks.append(line)
596 if on_output_line:
597 on_output_line(line.rstrip())
598 process.wait(timeout=timeout)
599 except subprocess.TimeoutExpired:
600 process.kill()
601 process.wait()
602 stderr_thread.join(timeout=5)
603 return ActionResult(
604 output="".join(output_chunks),
605 stderr="".join(stderr_chunks) or "MCP call timed out",
606 exit_code=124,
607 duration_ms=timeout * 1000,
608 )
609 finally:
610 self._current_process = None
611 stderr_thread.join(timeout=5)
612 return ActionResult(
613 output="".join(output_chunks),
614 stderr="".join(stderr_chunks),
615 exit_code=process.returncode,
616 duration_ms=_now_ms() - start,
617 )
619 def _evaluate(
620 self,
621 state: StateConfig,
622 action_result: ActionResult | None,
623 ctx: InterpolationContext,
624 ) -> EvaluationResult | None:
625 """Evaluate action result.
627 Args:
628 state: State configuration
629 action_result: Result from action execution (may be None)
630 ctx: Interpolation context
632 Returns:
633 EvaluationResult, or None if no evaluation needed
634 """
635 if state.evaluate is None:
636 # Default evaluation based on action type
637 if action_result:
638 action_mode = self._action_mode(state)
640 if action_mode == "mcp_tool":
641 # MCP tool call: use mcp_result evaluator
642 result = evaluate_mcp_result(action_result.output, action_result.exit_code)
643 elif action_mode == "prompt":
644 # Slash command or prompt: use LLM evaluation
645 if not self.fsm.llm.enabled:
646 result = EvaluationResult(
647 verdict="error",
648 details={"error": "LLM evaluation disabled via --no-llm"},
649 )
650 else:
651 result = evaluate_llm_structured(
652 action_result.output,
653 model=self.fsm.llm.model,
654 max_tokens=self.fsm.llm.max_tokens,
655 timeout=self.fsm.llm.timeout,
656 )
657 else:
658 # Shell command: use exit code
659 result = evaluate_exit_code(action_result.exit_code)
661 self._emit(
662 "evaluate",
663 {
664 "type": "default",
665 "verdict": result.verdict,
666 **result.details,
667 },
668 )
669 return result
670 return None
672 # Explicit evaluation config
673 raw_output = action_result.output if action_result else ""
674 if state.evaluate.source:
675 try:
676 eval_input = interpolate(state.evaluate.source, ctx)
677 except InterpolationError:
678 eval_input = raw_output
679 else:
680 eval_input = raw_output
682 if state.evaluate.type in self._contributed_evaluators:
683 result = self._contributed_evaluators[state.evaluate.type](
684 state.evaluate,
685 eval_input,
686 action_result.exit_code if action_result else 0,
687 ctx,
688 )
689 elif state.evaluate.type == "llm_structured" and not self.fsm.llm.enabled:
690 result = EvaluationResult(
691 verdict="error",
692 details={"error": "LLM evaluation disabled via --no-llm"},
693 )
694 else:
695 result = evaluate(
696 config=state.evaluate,
697 output=eval_input,
698 exit_code=action_result.exit_code if action_result else 0,
699 context=ctx,
700 )
702 self._emit(
703 "evaluate",
704 {
705 "type": state.evaluate.type,
706 "verdict": result.verdict,
707 **result.details,
708 },
709 )
711 return result
713 def _route(
714 self,
715 state: StateConfig,
716 verdict: str,
717 ctx: InterpolationContext,
718 ) -> str | None:
719 """Determine next state from verdict.
721 Resolution order (from design doc):
722 1. next (unconditional) - handled before this method
723 2. route (full routing table)
724 3. on_success/on_failure/on_error (shorthand)
725 4. terminal - handled in main loop
726 5. error
728 Args:
729 state: State configuration
730 verdict: Verdict string from evaluation
731 ctx: Interpolation context
733 Returns:
734 Next state name, or None if no valid route
735 """
736 if state.route:
737 routes = state.route.routes
738 if verdict in routes:
739 return self._resolve_route(routes[verdict], ctx)
740 if state.route.default:
741 return self._resolve_route(state.route.default, ctx)
742 if verdict == "error" and state.route.error:
743 return self._resolve_route(state.route.error, ctx)
744 return None
746 # Shorthand routing
747 if verdict == "yes" and state.on_yes:
748 return self._resolve_route(state.on_yes, ctx)
749 if verdict == "no" and state.on_no:
750 return self._resolve_route(state.on_no, ctx)
751 if verdict == "error" and state.on_error:
752 return self._resolve_route(state.on_error, ctx)
753 if verdict == "partial" and state.on_partial:
754 return self._resolve_route(state.on_partial, ctx)
755 if verdict == "blocked" and state.on_blocked:
756 return self._resolve_route(state.on_blocked, ctx)
758 # Dynamic on_<verdict> shorthands from extra_routes
759 if verdict in state.extra_routes:
760 return self._resolve_route(state.extra_routes[verdict], ctx)
762 return None
764 def _resolve_route(self, route: str, ctx: InterpolationContext) -> str:
765 """Resolve route target, handling special tokens.
767 Args:
768 route: Route target string
769 ctx: Interpolation context
771 Returns:
772 Resolved state name
773 """
774 if route == "$current":
775 return self.current_state
776 return interpolate(route, ctx)
778 def _action_mode(self, state: StateConfig) -> str:
779 """Return execution mode for the state: 'prompt', 'shell', or 'mcp_tool'."""
780 if state.action_type == "mcp_tool":
781 return "mcp_tool"
782 if state.action_type in ("prompt", "slash_command"):
783 return "prompt"
784 if state.action_type == "shell":
785 return "shell"
786 if state.action_type in self._contributed_actions:
787 return "contributed"
788 # Heuristic: / prefix = slash_command (prompt mode)
789 if state.action is not None and state.action.startswith("/"):
790 return "prompt"
791 return "shell"
793 def _build_context(self) -> InterpolationContext:
794 """Build interpolation context for current state.
796 Returns:
797 InterpolationContext with all runtime values
798 """
799 return InterpolationContext(
800 context=self.fsm.context,
801 captured=self.captured,
802 prev=self.prev_result,
803 result=None,
804 state_name=self.current_state,
805 iteration=self.iteration,
806 loop_name=self.fsm.name,
807 started_at=self.started_at,
808 elapsed_ms=_now_ms() - self.start_time_ms + self.elapsed_offset_ms,
809 )
811 def _emit(self, event: str, data: dict[str, Any]) -> None:
812 """Emit an event via the callback."""
813 self.event_callback(
814 {
815 "event": event,
816 "ts": _iso_now(),
817 **data,
818 }
819 )
821 def _finish(self, terminated_by: str, error: str | None = None) -> ExecutionResult:
822 """Finalize execution and return result."""
823 self._emit(
824 "loop_complete",
825 {
826 "final_state": self.current_state,
827 "iterations": self.iteration,
828 "terminated_by": terminated_by,
829 },
830 )
832 return ExecutionResult(
833 final_state=self.current_state,
834 iterations=self.iteration,
835 terminated_by=terminated_by,
836 duration_ms=_now_ms() - self.start_time_ms + self.elapsed_offset_ms,
837 captured=self.captured,
838 error=error,
839 )
841 def _handle_handoff(self, signal: DetectedSignal) -> ExecutionResult:
842 """Handle a detected handoff signal.
844 Emits a handoff_detected event and optionally invokes the handoff handler.
846 Args:
847 signal: The detected handoff signal
849 Returns:
850 ExecutionResult with handoff information
851 """
852 self._emit(
853 "handoff_detected",
854 {
855 "state": self.current_state,
856 "iteration": self.iteration,
857 "continuation": signal.payload,
858 },
859 )
861 # Invoke handler if configured
862 if self.handoff_handler:
863 result = self.handoff_handler.handle(self.fsm.name, signal.payload)
864 if result.spawned_process is not None:
865 self._emit(
866 "handoff_spawned",
867 {
868 "pid": result.spawned_process.pid,
869 "state": self.current_state,
870 },
871 )
873 return ExecutionResult(
874 final_state=self.current_state,
875 iterations=self.iteration,
876 terminated_by="handoff",
877 duration_ms=_now_ms() - self.start_time_ms + self.elapsed_offset_ms,
878 captured=self.captured,
879 handoff=True,
880 continuation_prompt=signal.payload,
881 )