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