Coverage for little_loops / cli / loop / _helpers.py: 24%
317 statements
« prev ^ index » next coverage.py v7.12.0, created at 2026-04-11 23:20 -0500
1"""Shared helpers for ll-loop CLI subcommands."""
3from __future__ import annotations
5import argparse
6import signal
7import subprocess
8import sys
9import time
10from pathlib import Path
11from types import FrameType
12from typing import TYPE_CHECKING, Any
14from little_loops.cli.output import colorize, terminal_width
15from little_loops.logger import Logger
17if TYPE_CHECKING:
18 from little_loops.fsm.schema import FSMLoop
20# Exit code mapping for terminated_by values
21EXIT_CODES: dict[str, int] = {
22 "terminal": 0,
23 "signal": 0,
24 "handoff": 0,
25 "max_iterations": 1,
26 "timeout": 1,
27}
29# Module-level shutdown state for signal handling
30_loop_shutdown_requested: bool = False
31_loop_executor: Any = None
32_loop_pid_file: Path | None = None
33_using_alt_screen: bool = False
36def _loop_signal_handler(signum: int, frame: FrameType | None) -> None:
37 """Handle shutdown signals gracefully for ll-loop.
39 First signal: Set shutdown flag for graceful exit after current state.
40 Second signal: Force immediate exit.
41 """
42 global _loop_shutdown_requested, _using_alt_screen
43 if _loop_shutdown_requested:
44 # Second signal - force exit
45 if _loop_pid_file is not None:
46 _loop_pid_file.unlink(missing_ok=True)
47 if _using_alt_screen:
48 print("\033[?1049l", end="", file=sys.stderr, flush=True)
49 print("\nForce shutdown requested", file=sys.stderr)
50 sys.exit(1)
51 _loop_shutdown_requested = True
52 print("\nShutdown requested, will exit after current state...", file=sys.stderr)
53 if _loop_executor is not None:
54 _loop_executor.request_shutdown()
55 # Kill any child subprocess currently blocking in the action runner
56 inner = getattr(_loop_executor, "_executor", None)
57 if inner is not None:
58 runner = getattr(inner, "action_runner", None)
59 if runner is not None:
60 proc = getattr(runner, "_current_process", None)
61 if proc is not None:
62 proc.kill()
63 # Also kill MCP subprocesses tracked directly on FSMExecutor (_run_subprocess path)
64 fsm_proc = getattr(inner, "_current_process", None)
65 if fsm_proc is not None:
66 fsm_proc.kill()
69def register_loop_signal_handlers(executor: Any, pid_file: Path | None = None) -> None:
70 """Register SIGINT/SIGTERM handlers for graceful loop shutdown.
72 Sets up signal handling so that Ctrl-C triggers a graceful shutdown
73 (calls executor.request_shutdown()) rather than raising KeyboardInterrupt.
74 A second Ctrl-C forces immediate exit with PID file cleanup.
76 Args:
77 executor: The PersistentExecutor instance to request shutdown on.
78 pid_file: Optional path to PID file to clean up on forced exit.
79 """
80 global _loop_shutdown_requested, _loop_executor, _loop_pid_file
81 _loop_shutdown_requested = False
82 _loop_executor = executor
83 _loop_pid_file = pid_file
84 signal.signal(signal.SIGINT, _loop_signal_handler)
85 signal.signal(signal.SIGTERM, _loop_signal_handler)
88def get_builtin_loops_dir() -> Path:
89 """Get the path to built-in loops bundled with the plugin."""
90 return Path(__file__).parent.parent.parent / "loops"
93def resolve_loop_path(name_or_path: str, loops_dir: Path) -> Path:
94 """Resolve loop name to file path."""
95 path = Path(name_or_path)
96 if path.exists():
97 return path
99 # Try <loops_dir>/<name>.fsm.yaml first (compiled FSM)
100 fsm_path = loops_dir / f"{name_or_path}.fsm.yaml"
101 if fsm_path.exists():
102 return fsm_path
104 # Fall back to <loops_dir>/<name>.yaml
105 loops_path = loops_dir / f"{name_or_path}.yaml"
106 if loops_path.exists():
107 return loops_path
109 # Fall back to built-in loops from plugin directory
110 builtin_path = get_builtin_loops_dir() / f"{name_or_path}.yaml"
111 if builtin_path.exists():
112 return builtin_path
114 raise FileNotFoundError(f"Loop not found: {name_or_path}")
117def load_loop(name_or_path: str, loops_dir: Path, logger: Logger) -> FSMLoop:
118 """Load and validate a loop.
120 Raises:
121 FileNotFoundError: If loop not found.
122 ValueError: If loop is invalid.
123 """
124 from little_loops.fsm.validation import load_and_validate
126 path = resolve_loop_path(name_or_path, loops_dir)
127 fsm, _ = load_and_validate(path)
128 return fsm
131def load_loop_with_spec(
132 name_or_path: str, loops_dir: Path, logger: Logger
133) -> tuple[FSMLoop, dict[str, Any]]:
134 """Load a loop and return both the FSMLoop and raw spec dict.
136 Used by commands that need access to raw YAML fields (e.g., description).
138 Raises:
139 FileNotFoundError: If loop not found.
140 ValueError: If loop is invalid.
141 """
142 import yaml
144 from little_loops.fsm.validation import load_and_validate
146 path = resolve_loop_path(name_or_path, loops_dir)
148 with open(path) as f:
149 spec = yaml.safe_load(f)
151 fsm, _ = load_and_validate(path)
152 return fsm, spec
155def print_execution_plan(fsm: FSMLoop) -> None:
156 """Print dry-run execution plan."""
157 tw = terminal_width()
158 print(colorize(f"Execution plan for: {fsm.name}", "1"))
159 print()
160 print("States:")
161 for name, state in fsm.states.items():
162 terminal_marker = colorize(" [TERMINAL]", "32") if state.terminal else ""
163 print(f" {colorize(f'[{name}]', '1')}{terminal_marker}")
164 if state.action:
165 if state.action_type == "prompt":
166 lines = state.action.strip().splitlines()
167 preview = "\n ".join(lines[:3])
168 if len(lines) > 3 or len(state.action) > 200:
169 preview += " ..."
170 print(f" action: |\n {preview}")
171 else:
172 max_action = tw - 16
173 action_display = (
174 state.action[:max_action] + "..."
175 if len(state.action) > max_action
176 else state.action
177 )
178 print(f" action: {action_display}")
179 if state.evaluate:
180 print(f" evaluate: {state.evaluate.type}")
181 if state.on_yes:
182 print(f" on_yes {colorize('->', '2')} {colorize(state.on_yes, '2')}")
183 if state.on_no:
184 print(f" on_no {colorize('->', '2')} {colorize(state.on_no, '2')}")
185 if state.on_error:
186 print(f" on_error {colorize('->', '2')} {colorize(state.on_error, '2')}")
187 if state.next:
188 print(f" next {colorize('->', '2')} {colorize(state.next, '2')}")
189 if state.route:
190 print(" route:")
191 for verdict, target in state.route.routes.items():
192 print(f" {verdict} {colorize('->', '2')} {colorize(target, '2')}")
193 if state.route.default:
194 print(f" _ {colorize('->', '2')} {colorize(state.route.default, '2')}")
195 print()
196 print(f"Initial state: {fsm.initial}")
197 print(f"Max iterations: {fsm.max_iterations}")
198 if fsm.timeout:
199 print(f"Timeout: {fsm.timeout}s")
200 if fsm.context:
201 print("Context:")
202 for key, value in fsm.context.items():
203 print(f" {key}: {value!r}")
206def run_background(
207 loop_name: str, args: argparse.Namespace, loops_dir: Path, subcommand: str = "run"
208) -> int:
209 """Launch loop as a detached background process.
211 Spawns a new process with start_new_session=True that re-executes
212 the loop with --foreground-internal. The parent writes the PID file
213 and returns immediately.
215 Args:
216 subcommand: The ll-loop subcommand to spawn ("run" or "resume").
218 Returns:
219 Exit code (0 = launched successfully).
220 """
221 running_dir = loops_dir / ".running"
222 running_dir.mkdir(parents=True, exist_ok=True)
224 pid_file = running_dir / f"{loop_name}.pid"
225 log_file = running_dir / f"{loop_name}.log"
227 # Build re-exec command with --foreground-internal instead of --background
228 cmd = [
229 sys.executable,
230 "-m",
231 "little_loops.cli.loop",
232 subcommand,
233 loop_name,
234 "--foreground-internal",
235 ]
237 # Forward relevant args
238 max_iter = getattr(args, "max_iterations", None)
239 if max_iter:
240 cmd.extend(["--max-iterations", str(max_iter)])
241 if getattr(args, "no_llm", False):
242 cmd.append("--no-llm")
243 llm_model = getattr(args, "llm_model", None)
244 if llm_model:
245 cmd.extend(["--llm-model", llm_model])
246 if getattr(args, "verbose", False):
247 cmd.append("--verbose")
248 if getattr(args, "show_diagrams", False):
249 cmd.append("--show-diagrams")
250 if getattr(args, "quiet", False):
251 cmd.append("--quiet")
252 if getattr(args, "queue", False):
253 cmd.append("--queue")
254 for kv in getattr(args, "context", None) or []:
255 cmd.extend(["--context", kv])
256 delay = getattr(args, "delay", None)
257 if delay is not None:
258 cmd.extend(["--delay", str(delay)])
259 handoff_threshold = getattr(args, "handoff_threshold", None)
260 if handoff_threshold is not None:
261 cmd.extend(["--handoff-threshold", str(handoff_threshold)])
262 context_limit = getattr(args, "context_limit", None)
263 if context_limit is not None:
264 cmd.extend(["--context-limit", str(context_limit)])
266 with open(log_file, "w") as log_fh:
267 process = subprocess.Popen(
268 cmd,
269 start_new_session=True,
270 stdout=log_fh,
271 stderr=log_fh,
272 stdin=subprocess.DEVNULL,
273 )
275 pid_file.write_text(str(process.pid))
276 print(
277 f"Loop {colorize(loop_name, '1')} started in background (PID: {colorize(str(process.pid), '2')})"
278 )
279 print(f" Log: {colorize(str(log_file), '2')}")
280 print(f" Status: {colorize(f'll-loop status {loop_name}', '2')}")
281 print(f" Stop: {colorize(f'll-loop stop {loop_name}', '2')}")
282 return 0
285def run_foreground(
286 executor: Any,
287 fsm: FSMLoop,
288 args: argparse.Namespace,
289 highlight_color: str = "32",
290 edge_label_colors: dict[str, str] | None = None,
291 badges: dict[str, str] | None = None,
292) -> int:
293 """Run loop with progress display.
295 Args:
296 highlight_color: ANSI SGR code for the active FSM state highlight in verbose mode.
297 edge_label_colors: Optional label→SGR-code mapping for transition edge labels.
298 badges: Optional glyph-key→string mapping for state type badges in FSM diagrams.
300 Returns:
301 Exit code (0 = success).
302 """
303 quiet = getattr(args, "quiet", False)
304 verbose = getattr(args, "verbose", False)
305 show_diagrams = getattr(args, "show_diagrams", False)
306 clear_screen = getattr(args, "clear", False)
307 if not quiet:
308 print(f"Running loop: {colorize(fsm.name, '1')}")
309 print(f"Max iterations: {colorize(str(fsm.max_iterations), '2')}")
310 print()
312 current_iteration = [0] # Use list to allow mutation in closure
313 last_state_at_depth: dict[int, str] = {} # Track last known state per nesting depth
314 child_fsm_stack: dict[int, FSMLoop | None] = {} # Active child FSM per depth
315 loop_start_time = time.monotonic()
317 def display_progress(event: dict) -> None:
318 """Display progress for events."""
319 event_type = event.get("event")
320 depth = event.get("depth", 0)
321 indent = " " * depth
322 tw = terminal_width()
323 max_line = tw - 8 - len(indent)
325 if event_type == "state_enter":
326 current_iteration[0] = event.get("iteration", 0)
327 state = event.get("state", "")
328 if not quiet:
329 elapsed_int = int(time.monotonic() - loop_start_time)
330 if elapsed_int < 60:
331 elapsed_str = f"{elapsed_int}s"
332 else:
333 elapsed_str = f"{elapsed_int // 60}m {elapsed_int % 60}s"
334 if clear_screen and sys.stdout.isatty() and depth == 0:
335 print("\033[2J\033[H", end="", flush=True)
336 # Update last-known state at this depth and clear stale deeper entries
337 last_state_at_depth[depth] = state
338 for k in [k for k in last_state_at_depth if k > depth]:
339 del last_state_at_depth[k]
340 # Load child FSM for the current state at this depth
341 parent_at_depth = fsm if depth == 0 else child_fsm_stack.get(depth - 1)
342 if parent_at_depth is not None and state in parent_at_depth.states:
343 fsm_state = parent_at_depth.states[state]
344 if fsm_state.loop is not None:
345 try:
346 child_fsm_stack[depth] = load_loop(
347 fsm_state.loop, executor.loops_dir, Logger()
348 )
349 except (FileNotFoundError, ValueError):
350 pass # leave child_fsm_stack[depth] unchanged on failure
351 else:
352 child_fsm_stack[depth] = None
353 else:
354 child_fsm_stack[depth] = None
355 # Clear stale deeper child FSM entries
356 for k in [k for k in child_fsm_stack if k > depth]:
357 del child_fsm_stack[k]
358 if show_diagrams:
359 from little_loops.cli.loop.layout import _render_fsm_diagram
361 diagram = _render_fsm_diagram(
362 fsm,
363 highlight_state=last_state_at_depth.get(0),
364 highlight_color=highlight_color,
365 edge_label_colors=edge_label_colors,
366 badges=badges,
367 )
368 header_text = f"== loop: {fsm.name} "
369 header = header_text + "=" * max(0, tw - len(header_text))
370 print(header, flush=True)
371 print(diagram, flush=True)
372 for d, child_fsm_at_d in sorted(child_fsm_stack.items()):
373 if child_fsm_at_d is not None and (d + 1) in last_state_at_depth:
374 child_name = child_fsm_at_d.name
375 separator_text = f"\u2500\u2500 sub-loop: {child_name} "
376 separator = separator_text + "\u2500" * max(0, tw - len(separator_text))
377 print(separator, flush=True)
378 child_diagram = _render_fsm_diagram(
379 child_fsm_at_d,
380 highlight_state=last_state_at_depth.get(d + 1),
381 highlight_color=highlight_color,
382 edge_label_colors=edge_label_colors,
383 badges=badges,
384 )
385 print(child_diagram, flush=True)
386 if not quiet:
387 print(
388 f"{indent}[{current_iteration[0]}/{fsm.max_iterations}] {colorize(state, '1')} ({colorize(elapsed_str, '2')})",
389 end="",
390 flush=True,
391 )
393 elif event_type == "action_start":
394 if not quiet:
395 action = event.get("action", "")
396 is_prompt = event.get("is_prompt", False)
397 if is_prompt:
398 lines = action.strip().splitlines()
399 line_count = len(lines)
400 prompt_badge = "\u2726" # ✦
401 print(
402 f"{indent} -> {colorize(prompt_badge, '2')} {colorize(f'({line_count} lines)', '2')}",
403 flush=True,
404 )
405 show_count = line_count if verbose else min(5, line_count)
406 for line in lines[:show_count]:
407 display = line[:max_line] + "..." if len(line) > max_line else line
408 print(f"{indent} {display}", flush=True)
409 if line_count > show_count:
410 print(
411 f"{indent} ... ({line_count - show_count} more lines)", flush=True
412 )
413 else:
414 action_display = action[:max_line] + "..." if len(action) > max_line else action
415 print(f"{indent} -> {colorize(action_display, '2')}", flush=True)
417 elif event_type == "action_output":
418 if not quiet and verbose:
419 line = event.get("line", "")
420 if line.strip():
421 display = line[:max_line] + "..." if len(line) > max_line else line
422 print(f"{indent} {display}", flush=True)
424 elif event_type == "action_complete":
425 if not quiet:
426 duration_ms = event.get("duration_ms", 0)
427 exit_code = event.get("exit_code", 0)
428 output_preview = event.get("output_preview")
429 is_prompt = event.get("is_prompt", False)
430 duration_sec = duration_ms / 1000
431 if duration_sec < 60:
432 duration_str = f"{duration_sec:.1f}s"
433 else:
434 minutes = int(duration_sec // 60)
435 seconds = duration_sec % 60
436 duration_str = f"{minutes}m {seconds:.0f}s"
437 parts = [f"{indent} ({colorize(duration_str, '2')})"]
438 if exit_code == 124:
439 parts.append(colorize("timed out", "38;5;208"))
440 elif exit_code != 0:
441 parts.append(colorize(f"exit: {exit_code}", "38;5;208"))
442 print(" ".join(parts), flush=True)
443 # Skip output preview for prompt states (already streamed) and in verbose mode
444 # (lines already shown via action_output events). In non-verbose mode, show
445 # a tail summary for shell states.
446 if output_preview and not is_prompt and not verbose:
447 lines = [ln for ln in output_preview.splitlines() if ln.strip()]
448 show_lines = lines[-8:] if lines else []
449 for line in show_lines:
450 display = line[:max_line] + "..." if len(line) > max_line else line
451 print(f"{indent} {display}", flush=True)
453 elif event_type == "evaluate":
454 if not quiet:
455 verdict = event.get("verdict", "")
456 confidence = event.get("confidence")
457 reason = event.get("reason", "")
458 error = event.get("error", "")
459 if verdict in ("yes", "target", "progress"):
460 symbol = colorize("\u2713", "32") # green checkmark
461 verdict_colored = colorize(verdict, "32")
462 else:
463 symbol = colorize("\u2717", "38;5;208") # orange x mark
464 verdict_colored = (
465 colorize(verdict, "38;5;208")
466 if verdict in ("no", "error")
467 else colorize(verdict, "2")
468 )
469 # Build verdict line
470 if error and verdict == "error":
471 verdict_line = f"{symbol} {verdict_colored}: {error}"
472 elif confidence is not None:
473 verdict_line = (
474 f"{symbol} {verdict_colored} {colorize(f'({confidence:.2f})', '2')}"
475 )
476 else:
477 verdict_line = f"{symbol} {verdict_colored}"
478 print(f"{indent} {verdict_line}", flush=True)
479 # Show raw_preview for error verdicts to aid diagnosis
480 raw_preview = event.get("raw_preview", "")
481 if raw_preview and verdict == "error":
482 print(f"{indent} raw: {raw_preview[:200]}", flush=True)
483 # Show reason on a second line if present (and not already shown as error)
484 if reason and not (error and verdict == "error"):
485 reason_display = reason[:300] + "..." if len(reason) > 300 else reason
486 print(f"{indent} {reason_display}", flush=True)
488 elif event_type == "route":
489 if not quiet:
490 to_state = event.get("to", "")
491 print(f"{indent} {colorize('->', '2')} {colorize(to_state, '1')}", flush=True)
493 # Wire progress display via the EventBus on PersistentExecutor
494 if not quiet or show_diagrams:
495 if hasattr(executor, "event_bus"):
496 executor.event_bus.register(display_progress)
497 else:
498 executor._on_event = display_progress
500 # Enter alternate screen buffer when showing diagrams with clear to prevent
501 # scrollback contamination from diagrams taller than the terminal height.
502 global _using_alt_screen
503 if show_diagrams and clear_screen and sys.stdout.isatty():
504 _using_alt_screen = True
505 print("\033[?1049h\033[H", end="", flush=True)
507 try:
508 result = executor.run()
509 finally:
510 if _using_alt_screen:
511 print("\033[?1049l", end="", flush=True)
512 _using_alt_screen = False
514 if not quiet:
515 print()
516 duration_sec = result.duration_ms / 1000
517 if duration_sec < 60:
518 duration_str = f"{duration_sec:.1f}s"
519 else:
520 minutes = int(duration_sec // 60)
521 seconds = duration_sec % 60
522 duration_str = f"{minutes}m {seconds:.0f}s"
523 if result.terminated_by == "terminal":
524 state_colored = colorize(result.final_state, "32")
525 else:
526 state_colored = colorize(result.final_state, "38;5;208")
527 print(f"Loop completed: {state_colored} ({result.iterations} iterations, {duration_str})")
529 return EXIT_CODES.get(result.terminated_by, 1)