Coverage for little_loops / fsm / runners.py: 54%
131 statements
« prev ^ index » next coverage.py v7.12.0, created at 2026-06-16 13:12 -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 selectors
12import subprocess
13import sys
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 _kill_process_group,
25 run_claude_command,
26)
29def _now_ms() -> int:
30 """Get current time in milliseconds."""
31 return int(time.time() * 1000)
34class ActionRunner(Protocol):
35 """Protocol for action execution."""
37 def run(
38 self,
39 action: str,
40 timeout: int,
41 is_slash_command: bool,
42 on_output_line: Callable[[str], None] | None = None,
43 agent: str | None = None,
44 tools: list[str] | None = None,
45 on_usage: UsageCallback | None = None,
46 on_usage_detailed: DetailedUsageCallback | None = None,
47 model: str | None = None,
48 ) -> ActionResult:
49 """Execute an action and return the result.
51 Args:
52 action: The command to execute
53 timeout: Timeout in seconds
54 is_slash_command: True if this is a slash command (starts with /)
55 on_output_line: Optional callback invoked for each output line
56 agent: Optional agent name to pass as --agent to Claude CLI (prompt-mode only)
57 tools: Optional list of tool names to pass as --tools CSV to Claude CLI (prompt-mode only)
58 on_usage: Optional callback invoked with (input_tokens, output_tokens) on completion
59 on_usage_detailed: Optional callback invoked with a TokenUsage dataclass on completion
61 Returns:
62 ActionResult with output, stderr, exit_code, duration_ms
63 """
64 ...
67class DefaultActionRunner:
68 """Execute actions via subprocess or Claude CLI."""
70 def __init__(self) -> None:
71 self._current_process: subprocess.Popen[str] | None = None
73 def run(
74 self,
75 action: str,
76 timeout: int,
77 is_slash_command: bool,
78 on_output_line: Callable[[str], None] | None = None,
79 agent: str | None = None,
80 tools: list[str] | None = None,
81 on_usage: UsageCallback | None = None,
82 on_usage_detailed: DetailedUsageCallback | None = None,
83 model: str | None = None,
84 ) -> ActionResult:
85 """Execute action and return result, streaming output line by line.
87 Args:
88 action: The command to execute
89 timeout: Timeout in seconds
90 is_slash_command: True if action starts with /
91 on_output_line: Optional callback invoked for each stdout line
92 agent: Optional agent name to pass as --agent to Claude CLI (prompt-mode only)
93 tools: Optional list of tool names to pass as --tools CSV (prompt-mode only)
94 on_usage: Optional callback invoked with (input_tokens, output_tokens) on completion
95 on_usage_detailed: Optional callback invoked with a TokenUsage dataclass on completion
96 model: Optional model override to pass as --model to Claude CLI (prompt-mode only)
98 Returns:
99 ActionResult with execution details
100 """
101 start = _now_ms()
103 if is_slash_command:
104 # Execute via Claude CLI using run_claude_command() so that the
105 # subprocess loads the full plugin/tool context (including deferred
106 # tools like Skill). The old --no-session-persistence path prevented
107 # ToolSearch from resolving deferred tool schemas (BUG-946).
108 def _stream_cb(line: str, is_stderr: bool) -> None:
109 if not is_stderr and on_output_line:
110 on_output_line(line)
112 def _on_proc_start(p: subprocess.Popen[str]) -> None:
113 self._current_process = p
115 def _on_proc_end(p: subprocess.Popen[str]) -> None:
116 self._current_process = None
118 collected_usage: list[TokenUsage] = []
120 def _collect_usage(u: TokenUsage) -> None:
121 collected_usage.append(u)
122 if on_usage_detailed:
123 on_usage_detailed(u)
125 try:
126 completed = run_claude_command(
127 command=action,
128 timeout=timeout,
129 stream_callback=_stream_cb,
130 on_process_start=_on_proc_start,
131 on_process_end=_on_proc_end,
132 agent=agent,
133 tools=tools,
134 on_usage=on_usage,
135 on_usage_detailed=_collect_usage,
136 model=model,
137 )
138 except subprocess.TimeoutExpired:
139 return ActionResult(
140 output="",
141 stderr="Action timed out",
142 exit_code=124,
143 duration_ms=timeout * 1000,
144 )
145 return ActionResult(
146 output=completed.stdout,
147 stderr=completed.stderr,
148 exit_code=completed.returncode,
149 duration_ms=_now_ms() - start,
150 usage_events=collected_usage,
151 )
153 # Shell command — selector-based I/O with wall-clock timeout enforcement.
154 # Uses selectors.DefaultSelector to read stdout and stderr concurrently
155 # with bounded polling (sel.select(timeout=1.0)), checking the wall-clock
156 # deadline before each read. This closes the timeout dead-zone in the old
157 # `for line in process.stdout` pattern which blocked indefinitely when a
158 # shell process hung before producing any output.
159 cmd = ["bash", "-c", action]
160 process = subprocess.Popen(
161 cmd,
162 stdout=subprocess.PIPE,
163 stderr=subprocess.PIPE,
164 text=True,
165 )
166 self._current_process = process
167 deadline = time.time() + timeout
169 output_chunks: list[str] = []
170 stderr_chunks: list[str] = []
172 sel = selectors.DefaultSelector()
173 if process.stdout is not None:
174 sel.register(process.stdout, selectors.EVENT_READ, data="stdout")
175 if process.stderr is not None:
176 sel.register(process.stderr, selectors.EVENT_READ, data="stderr")
178 timed_out = False
179 try:
180 while sel.get_map():
181 # Bounded poll — never block longer than 1 second
182 remaining = deadline - time.time()
183 if remaining <= 0:
184 timed_out = True
185 break
186 poll_timeout = min(1.0, remaining)
187 ready = sel.select(timeout=poll_timeout)
188 if not ready:
189 # No pipes ready within poll window — loop re-checks deadline
190 continue
191 for key, _mask in ready:
192 line = key.fileobj.readline() # type: ignore[union-attr]
193 if line:
194 if key.data == "stdout":
195 output_chunks.append(line)
196 if on_output_line:
197 on_output_line(line.rstrip())
198 else:
199 stderr_chunks.append(line)
200 else:
201 # EOF on this pipe — unregister it
202 sel.unregister(key.fileobj)
203 finally:
204 sel.close()
205 self._current_process = None
207 if timed_out:
208 _kill_process_group(process)
209 # Drain any remaining output after the kill
210 try:
211 process.wait(timeout=5)
212 except subprocess.TimeoutExpired:
213 process.kill()
214 process.wait()
215 return ActionResult(
216 output="".join(output_chunks),
217 stderr="".join(stderr_chunks) or "Action timed out",
218 exit_code=124,
219 duration_ms=timeout * 1000,
220 )
222 process.wait(timeout=5)
223 return ActionResult(
224 output="".join(output_chunks),
225 stderr="".join(stderr_chunks),
226 exit_code=process.returncode,
227 duration_ms=_now_ms() - start,
228 )
231@dataclass
232class SimulationActionRunner:
233 """Action runner for simulation mode - prompts user instead of executing.
235 This runner allows users to trace through FSM logic without executing
236 real commands. It can either prompt interactively for results or use
237 predefined scenarios.
239 Attributes:
240 scenario: Predefined result pattern ("all-pass", "all-fail", "first-fail", "alternating")
241 call_count: Number of actions simulated so far
242 calls: List of all actions that would have been executed
243 """
245 scenario: str | None = None
246 call_count: int = 0
247 calls: list[str] = field(default_factory=list)
249 def run(
250 self,
251 action: str,
252 timeout: int,
253 is_slash_command: bool,
254 on_output_line: Callable[[str], None] | None = None,
255 agent: str | None = None,
256 tools: list[str] | None = None,
257 on_usage: UsageCallback | None = None,
258 on_usage_detailed: DetailedUsageCallback | None = None,
259 model: str | None = None,
260 ) -> ActionResult:
261 """Prompt user for simulated result instead of executing.
263 Args:
264 action: The command that would be executed
265 timeout: Timeout (ignored in simulation)
266 is_slash_command: Whether this is a slash command
267 on_output_line: Ignored in simulation
268 agent: Ignored in simulation
269 tools: Ignored in simulation
270 on_usage: Ignored in simulation
271 on_usage_detailed: Ignored in simulation
272 model: Ignored in simulation
274 Returns:
275 ActionResult with simulated exit code
276 """
277 del timeout, on_output_line, agent, tools, on_usage, model # unused in simulation
278 self.calls.append(action)
279 self.call_count += 1
281 cmd_type = "slash command" if is_slash_command else "shell command"
282 print(f" [SIMULATED] Would execute ({cmd_type}): {action}")
284 if self.scenario:
285 exit_code = self._scenario_result()
286 scenario_label = {
287 "all-pass": "Success (scenario: all-pass)",
288 "all-fail": "Failure (scenario: all-fail)",
289 "all-error": "Error (scenario: all-error)",
290 "first-fail": "Failure" if self.call_count == 1 else "Success",
291 "alternating": "Failure" if self.call_count % 2 == 1 else "Success",
292 }.get(self.scenario, "Success")
293 print(f" [AUTO] Result: {scenario_label}")
294 else:
295 exit_code = self._prompt_result()
297 return ActionResult(
298 output=f"[simulated output for: {action}]",
299 stderr="",
300 exit_code=exit_code,
301 duration_ms=0,
302 )
304 def _scenario_result(self) -> int:
305 """Return exit code based on scenario pattern.
307 Returns:
308 0 for success, 1 for failure, 2 for error based on scenario logic
309 """
310 if self.scenario == "all-pass":
311 return 0
312 elif self.scenario == "all-fail":
313 return 1
314 elif self.scenario == "all-error":
315 return 2
316 elif self.scenario == "first-fail":
317 # First call fails, rest pass
318 return 1 if self.call_count == 1 else 0
319 elif self.scenario == "alternating":
320 # Odd calls fail, even calls pass
321 return 1 if self.call_count % 2 == 1 else 0
322 return 0
324 def _prompt_result(self) -> int:
325 """Prompt user for simulated exit code.
327 Returns:
328 Exit code based on user selection
329 """
330 print()
331 print(" ? What should the simulated result be?")
332 print(" 1) Success (exit 0) [default]")
333 print(" 2) Failure (exit 1)")
334 print(" 3) Error (exit 2)")
336 while True:
337 try:
338 sys.stdout.write(" > ")
339 sys.stdout.flush()
340 choice = sys.stdin.readline().strip()
341 if choice in ("1", ""):
342 return 0
343 elif choice == "2":
344 return 1
345 elif choice == "3":
346 return 2
347 print(" Invalid choice. Enter 1, 2, or 3.")
348 except (EOFError, KeyboardInterrupt):
349 print()
350 return 0