Coverage for little_loops / fsm / runners.py: 18%
133 statements
« prev ^ index » next coverage.py v7.12.0, created at 2026-06-26 17:38 -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 except Exception as exc:
146 return ActionResult(
147 output="",
148 stderr=f"Action failed: {exc}",
149 exit_code=1,
150 duration_ms=_now_ms() - start,
151 )
152 return ActionResult(
153 output=completed.stdout,
154 stderr=completed.stderr,
155 exit_code=completed.returncode,
156 duration_ms=_now_ms() - start,
157 usage_events=collected_usage,
158 )
160 # Shell command — selector-based I/O with wall-clock timeout enforcement.
161 # Uses selectors.DefaultSelector to read stdout and stderr concurrently
162 # with bounded polling (sel.select(timeout=1.0)), checking the wall-clock
163 # deadline before each read. This closes the timeout dead-zone in the old
164 # `for line in process.stdout` pattern which blocked indefinitely when a
165 # shell process hung before producing any output.
166 cmd = ["bash", "-c", action]
167 process = subprocess.Popen(
168 cmd,
169 stdout=subprocess.PIPE,
170 stderr=subprocess.PIPE,
171 text=True,
172 )
173 self._current_process = process
174 deadline = time.time() + timeout
176 output_chunks: list[str] = []
177 stderr_chunks: list[str] = []
179 sel = selectors.DefaultSelector()
180 if process.stdout is not None:
181 sel.register(process.stdout, selectors.EVENT_READ, data="stdout")
182 if process.stderr is not None:
183 sel.register(process.stderr, selectors.EVENT_READ, data="stderr")
185 timed_out = False
186 try:
187 while sel.get_map():
188 # Bounded poll — never block longer than 1 second
189 remaining = deadline - time.time()
190 if remaining <= 0:
191 timed_out = True
192 break
193 poll_timeout = min(1.0, remaining)
194 ready = sel.select(timeout=poll_timeout)
195 if not ready:
196 # No pipes ready within poll window — loop re-checks deadline
197 continue
198 for key, _mask in ready:
199 line = key.fileobj.readline() # type: ignore[union-attr]
200 if line:
201 if key.data == "stdout":
202 output_chunks.append(line)
203 if on_output_line:
204 on_output_line(line.rstrip())
205 else:
206 stderr_chunks.append(line)
207 else:
208 # EOF on this pipe — unregister it
209 sel.unregister(key.fileobj)
210 finally:
211 sel.close()
212 self._current_process = None
214 if timed_out:
215 _kill_process_group(process)
216 # Drain any remaining output after the kill
217 try:
218 process.wait(timeout=5)
219 except subprocess.TimeoutExpired:
220 process.kill()
221 process.wait()
222 return ActionResult(
223 output="".join(output_chunks),
224 stderr="".join(stderr_chunks) or "Action timed out",
225 exit_code=124,
226 duration_ms=timeout * 1000,
227 )
229 process.wait(timeout=5)
230 return ActionResult(
231 output="".join(output_chunks),
232 stderr="".join(stderr_chunks),
233 exit_code=process.returncode,
234 duration_ms=_now_ms() - start,
235 )
238@dataclass
239class SimulationActionRunner:
240 """Action runner for simulation mode - prompts user instead of executing.
242 This runner allows users to trace through FSM logic without executing
243 real commands. It can either prompt interactively for results or use
244 predefined scenarios.
246 Attributes:
247 scenario: Predefined result pattern ("all-pass", "all-fail", "first-fail", "alternating")
248 call_count: Number of actions simulated so far
249 calls: List of all actions that would have been executed
250 """
252 scenario: str | None = None
253 call_count: int = 0
254 calls: list[str] = field(default_factory=list)
256 def run(
257 self,
258 action: str,
259 timeout: int,
260 is_slash_command: bool,
261 on_output_line: Callable[[str], None] | None = None,
262 agent: str | None = None,
263 tools: list[str] | None = None,
264 on_usage: UsageCallback | None = None,
265 on_usage_detailed: DetailedUsageCallback | None = None,
266 model: str | None = None,
267 ) -> ActionResult:
268 """Prompt user for simulated result instead of executing.
270 Args:
271 action: The command that would be executed
272 timeout: Timeout (ignored in simulation)
273 is_slash_command: Whether this is a slash command
274 on_output_line: Ignored in simulation
275 agent: Ignored in simulation
276 tools: Ignored in simulation
277 on_usage: Ignored in simulation
278 on_usage_detailed: Ignored in simulation
279 model: Ignored in simulation
281 Returns:
282 ActionResult with simulated exit code
283 """
284 del timeout, on_output_line, agent, tools, on_usage, model # unused in simulation
285 self.calls.append(action)
286 self.call_count += 1
288 cmd_type = "slash command" if is_slash_command else "shell command"
289 print(f" [SIMULATED] Would execute ({cmd_type}): {action}")
291 if self.scenario:
292 exit_code = self._scenario_result()
293 scenario_label = {
294 "all-pass": "Success (scenario: all-pass)",
295 "all-fail": "Failure (scenario: all-fail)",
296 "all-error": "Error (scenario: all-error)",
297 "first-fail": "Failure" if self.call_count == 1 else "Success",
298 "alternating": "Failure" if self.call_count % 2 == 1 else "Success",
299 }.get(self.scenario, "Success")
300 print(f" [AUTO] Result: {scenario_label}")
301 else:
302 exit_code = self._prompt_result()
304 return ActionResult(
305 output=f"[simulated output for: {action}]",
306 stderr="",
307 exit_code=exit_code,
308 duration_ms=0,
309 )
311 def _scenario_result(self) -> int:
312 """Return exit code based on scenario pattern.
314 Returns:
315 0 for success, 1 for failure, 2 for error based on scenario logic
316 """
317 if self.scenario == "all-pass":
318 return 0
319 elif self.scenario == "all-fail":
320 return 1
321 elif self.scenario == "all-error":
322 return 2
323 elif self.scenario == "first-fail":
324 # First call fails, rest pass
325 return 1 if self.call_count == 1 else 0
326 elif self.scenario == "alternating":
327 # Odd calls fail, even calls pass
328 return 1 if self.call_count % 2 == 1 else 0
329 return 0
331 def _prompt_result(self) -> int:
332 """Prompt user for simulated exit code.
334 Returns:
335 Exit code based on user selection
336 """
337 print()
338 print(" ? What should the simulated result be?")
339 print(" 1) Success (exit 0) [default]")
340 print(" 2) Failure (exit 1)")
341 print(" 3) Error (exit 2)")
343 while True:
344 try:
345 sys.stdout.write(" > ")
346 sys.stdout.flush()
347 choice = sys.stdin.readline().strip()
348 if choice in ("1", ""):
349 return 0
350 elif choice == "2":
351 return 1
352 elif choice == "3":
353 return 2
354 print(" Invalid choice. Enter 1, 2, or 3.")
355 except (EOFError, KeyboardInterrupt):
356 print()
357 return 0