Coverage for little_loops / fsm / runners.py: 42%
114 statements
« prev ^ index » next coverage.py v7.12.0, created at 2026-06-08 15:34 -0500
1"""FSM action runner implementations.
3Provides the protocol and concrete implementations for action execution:
4- ActionRunner: Protocol defining the runner interface
5- DefaultActionRunner: Subprocess-based runner for real execution
6- SimulationActionRunner: Interactive/scenario-based runner for testing and dry-runs
7"""
9from __future__ import annotations
11import subprocess
12import sys
13import threading
14import time
15from collections.abc import Callable
16from dataclasses import dataclass, field
17from typing import Protocol
19from little_loops.fsm.types import ActionResult
20from little_loops.subprocess_utils import (
21 DetailedUsageCallback,
22 TokenUsage,
23 UsageCallback,
24 run_claude_command,
25)
28def _now_ms() -> int:
29 """Get current time in milliseconds."""
30 return int(time.time() * 1000)
33class ActionRunner(Protocol):
34 """Protocol for action execution."""
36 def run(
37 self,
38 action: str,
39 timeout: int,
40 is_slash_command: bool,
41 on_output_line: Callable[[str], None] | None = None,
42 agent: str | None = None,
43 tools: list[str] | None = None,
44 on_usage: UsageCallback | None = None,
45 on_usage_detailed: DetailedUsageCallback | None = None,
46 ) -> ActionResult:
47 """Execute an action and return the result.
49 Args:
50 action: The command to execute
51 timeout: Timeout in seconds
52 is_slash_command: True if this is a slash command (starts with /)
53 on_output_line: Optional callback invoked for each output line
54 agent: Optional agent name to pass as --agent to Claude CLI (prompt-mode only)
55 tools: Optional list of tool names to pass as --tools CSV to Claude CLI (prompt-mode only)
56 on_usage: Optional callback invoked with (input_tokens, output_tokens) on completion
57 on_usage_detailed: Optional callback invoked with a TokenUsage dataclass on completion
59 Returns:
60 ActionResult with output, stderr, exit_code, duration_ms
61 """
62 ...
65class DefaultActionRunner:
66 """Execute actions via subprocess or Claude CLI."""
68 def __init__(self) -> None:
69 self._current_process: subprocess.Popen[str] | None = None
71 def run(
72 self,
73 action: str,
74 timeout: int,
75 is_slash_command: bool,
76 on_output_line: Callable[[str], None] | None = None,
77 agent: str | None = None,
78 tools: list[str] | None = None,
79 on_usage: UsageCallback | None = None,
80 on_usage_detailed: DetailedUsageCallback | None = None,
81 ) -> ActionResult:
82 """Execute action and return result, streaming output line by line.
84 Args:
85 action: The command to execute
86 timeout: Timeout in seconds
87 is_slash_command: True if action starts with /
88 on_output_line: Optional callback invoked for each stdout line
89 agent: Optional agent name to pass as --agent to Claude CLI (prompt-mode only)
90 tools: Optional list of tool names to pass as --tools CSV (prompt-mode only)
91 on_usage: Optional callback invoked with (input_tokens, output_tokens) on completion
92 on_usage_detailed: Optional callback invoked with a TokenUsage dataclass on completion
94 Returns:
95 ActionResult with execution details
96 """
97 start = _now_ms()
99 if is_slash_command:
100 # Execute via Claude CLI using run_claude_command() so that the
101 # subprocess loads the full plugin/tool context (including deferred
102 # tools like Skill). The old --no-session-persistence path prevented
103 # ToolSearch from resolving deferred tool schemas (BUG-946).
104 def _stream_cb(line: str, is_stderr: bool) -> None:
105 if not is_stderr and on_output_line:
106 on_output_line(line)
108 def _on_proc_start(p: subprocess.Popen[str]) -> None:
109 self._current_process = p
111 def _on_proc_end(p: subprocess.Popen[str]) -> None:
112 self._current_process = None
114 collected_usage: list[TokenUsage] = []
116 def _collect_usage(u: TokenUsage) -> None:
117 collected_usage.append(u)
118 if on_usage_detailed:
119 on_usage_detailed(u)
121 try:
122 completed = run_claude_command(
123 command=action,
124 timeout=timeout,
125 stream_callback=_stream_cb,
126 on_process_start=_on_proc_start,
127 on_process_end=_on_proc_end,
128 agent=agent,
129 tools=tools,
130 on_usage=on_usage,
131 on_usage_detailed=_collect_usage,
132 )
133 except subprocess.TimeoutExpired:
134 return ActionResult(
135 output="",
136 stderr="Action timed out",
137 exit_code=124,
138 duration_ms=timeout * 1000,
139 )
140 return ActionResult(
141 output=completed.stdout,
142 stderr=completed.stderr,
143 exit_code=completed.returncode,
144 duration_ms=_now_ms() - start,
145 usage_events=collected_usage,
146 )
148 # Shell command
149 cmd = ["bash", "-c", action]
150 process = subprocess.Popen(
151 cmd,
152 stdout=subprocess.PIPE,
153 stderr=subprocess.PIPE,
154 text=True,
155 )
156 self._current_process = process
157 output_chunks: list[str] = []
158 stderr_chunks: list[str] = []
160 def _drain_stderr() -> None:
161 assert process.stderr is not None
162 for line in process.stderr:
163 stderr_chunks.append(line)
165 stderr_thread = threading.Thread(target=_drain_stderr, daemon=True)
166 stderr_thread.start()
168 try:
169 for line in process.stdout: # type: ignore[union-attr]
170 output_chunks.append(line)
171 if on_output_line:
172 on_output_line(line.rstrip())
173 process.wait(timeout=timeout)
174 except subprocess.TimeoutExpired:
175 process.kill()
176 process.wait()
177 stderr_thread.join(timeout=5)
178 return ActionResult(
179 output="".join(output_chunks),
180 stderr="".join(stderr_chunks) or "Action timed out",
181 exit_code=124,
182 duration_ms=timeout * 1000,
183 )
184 finally:
185 self._current_process = None
186 stderr_thread.join(timeout=5)
187 stderr = "".join(stderr_chunks)
188 return ActionResult(
189 output="".join(output_chunks),
190 stderr=stderr,
191 exit_code=process.returncode,
192 duration_ms=_now_ms() - start,
193 )
196@dataclass
197class SimulationActionRunner:
198 """Action runner for simulation mode - prompts user instead of executing.
200 This runner allows users to trace through FSM logic without executing
201 real commands. It can either prompt interactively for results or use
202 predefined scenarios.
204 Attributes:
205 scenario: Predefined result pattern ("all-pass", "all-fail", "first-fail", "alternating")
206 call_count: Number of actions simulated so far
207 calls: List of all actions that would have been executed
208 """
210 scenario: str | None = None
211 call_count: int = 0
212 calls: list[str] = field(default_factory=list)
214 def run(
215 self,
216 action: str,
217 timeout: int,
218 is_slash_command: bool,
219 on_output_line: Callable[[str], None] | None = None,
220 agent: str | None = None,
221 tools: list[str] | None = None,
222 on_usage: UsageCallback | None = None,
223 on_usage_detailed: DetailedUsageCallback | None = None,
224 ) -> ActionResult:
225 """Prompt user for simulated result instead of executing.
227 Args:
228 action: The command that would be executed
229 timeout: Timeout (ignored in simulation)
230 is_slash_command: Whether this is a slash command
231 on_output_line: Ignored in simulation
232 agent: Ignored in simulation
233 tools: Ignored in simulation
234 on_usage: Ignored in simulation
235 on_usage_detailed: Ignored in simulation
237 Returns:
238 ActionResult with simulated exit code
239 """
240 del timeout, on_output_line, agent, tools, on_usage # unused in simulation
241 self.calls.append(action)
242 self.call_count += 1
244 cmd_type = "slash command" if is_slash_command else "shell command"
245 print(f" [SIMULATED] Would execute ({cmd_type}): {action}")
247 if self.scenario:
248 exit_code = self._scenario_result()
249 scenario_label = {
250 "all-pass": "Success (scenario: all-pass)",
251 "all-fail": "Failure (scenario: all-fail)",
252 "all-error": "Error (scenario: all-error)",
253 "first-fail": "Failure" if self.call_count == 1 else "Success",
254 "alternating": "Failure" if self.call_count % 2 == 1 else "Success",
255 }.get(self.scenario, "Success")
256 print(f" [AUTO] Result: {scenario_label}")
257 else:
258 exit_code = self._prompt_result()
260 return ActionResult(
261 output=f"[simulated output for: {action}]",
262 stderr="",
263 exit_code=exit_code,
264 duration_ms=0,
265 )
267 def _scenario_result(self) -> int:
268 """Return exit code based on scenario pattern.
270 Returns:
271 0 for success, 1 for failure, 2 for error based on scenario logic
272 """
273 if self.scenario == "all-pass":
274 return 0
275 elif self.scenario == "all-fail":
276 return 1
277 elif self.scenario == "all-error":
278 return 2
279 elif self.scenario == "first-fail":
280 # First call fails, rest pass
281 return 1 if self.call_count == 1 else 0
282 elif self.scenario == "alternating":
283 # Odd calls fail, even calls pass
284 return 1 if self.call_count % 2 == 1 else 0
285 return 0
287 def _prompt_result(self) -> int:
288 """Prompt user for simulated exit code.
290 Returns:
291 Exit code based on user selection
292 """
293 print()
294 print(" ? What should the simulated result be?")
295 print(" 1) Success (exit 0) [default]")
296 print(" 2) Failure (exit 1)")
297 print(" 3) Error (exit 2)")
299 while True:
300 try:
301 sys.stdout.write(" > ")
302 sys.stdout.flush()
303 choice = sys.stdin.readline().strip()
304 if choice in ("1", ""):
305 return 0
306 elif choice == "2":
307 return 1
308 elif choice == "3":
309 return 2
310 print(" Invalid choice. Enter 1, 2, or 3.")
311 except (EOFError, KeyboardInterrupt):
312 print()
313 return 0