Coverage for little_loops / cli / loop / _helpers.py: 0%
603 statements
« prev ^ index » next coverage.py v7.12.0, created at 2026-05-28 13:07 -0500
1"""Shared helpers for ll-loop CLI subcommands."""
3from __future__ import annotations
5import argparse
6import json
7import re
8import signal
9import subprocess
10import sys
11import time
12from datetime import datetime
13from pathlib import Path
14from types import FrameType
15from typing import TYPE_CHECKING, Any
17from little_loops.cli.loop.diagram_modes import (
18 TOPOLOGY_TO_DETAIL,
19 DiagramFacets,
20 resolve_facets,
21)
22from little_loops.cli.output import colorize, terminal_size, terminal_width
23from little_loops.fsm.concurrency import _process_alive
24from little_loops.logger import Logger
26if TYPE_CHECKING:
27 from little_loops.fsm.schema import FSMLoop
29# Exit code mapping for terminated_by values
30EXIT_CODES: dict[str, int] = {
31 "terminal": 0,
32 "signal": 0,
33 "handoff": 0,
34 "max_iterations": 1,
35 "timeout": 1,
36 "cycle_detected": 1,
37 "stall_detected": 1,
38}
40# Minimum number of action-output rows reserved beneath the pinned pane in
41# alt-screen mode. When the pinned pane plus this margin would exceed the
42# terminal height, the layout falls back to a more compact diagram variant.
43MIN_ACTION_ROWS = 6
45# Module-level shutdown state for signal handling
46_loop_shutdown_requested: bool = False
47_loop_executor: Any = None
48_loop_pid_file: Path | None = None
49_using_alt_screen: bool = False
50# Set by SIGWINCH handler when the terminal is resized; consumed by the
51# display_progress callback to trigger a pinned-pane redraw on the next event.
52_needs_redraw: bool = False
53# Previous SIGWINCH handler, stashed when we install our own so it can be
54# restored in run_foreground's finally block. ``None`` means "not installed".
55_original_sigwinch: Any = None
58_ANSI_RE = re.compile(r"\x1b\[[0-9;]*[mABCDEFGHJKSTfhilmnprsu]")
61class _TeeWriter:
62 """Wraps a stream, writing to both the original and a log file (ANSI stripped on log)."""
64 def __init__(self, stream: Any, log_fh: Any) -> None:
65 self._stream = stream
66 self._log_fh = log_fh
68 def write(self, data: str) -> int:
69 n = self._stream.write(data)
70 self._log_fh.write(_ANSI_RE.sub("", data))
71 return n
73 def flush(self) -> None:
74 self._stream.flush()
75 self._log_fh.flush()
77 def __getattr__(self, name: str) -> Any:
78 return getattr(self._stream, name)
81def _loop_signal_handler(signum: int, frame: FrameType | None) -> None:
82 """Handle shutdown signals gracefully for ll-loop.
84 First signal: Set shutdown flag for graceful exit after current state.
85 Second signal: Force immediate exit.
86 """
87 global _loop_shutdown_requested, _using_alt_screen
88 if _loop_shutdown_requested:
89 # Second signal - force exit
90 if _loop_pid_file is not None:
91 _loop_pid_file.unlink(missing_ok=True)
92 if _using_alt_screen:
93 # Reset the DECSTBM scroll region BEFORE exiting the alt screen,
94 # otherwise the main buffer is left with a restricted scroll
95 # region — visible to the user as `seq 1 50` failing to scroll
96 # past the previous pinned-pane height.
97 print("\033[r", end="", file=sys.stderr, flush=True)
98 print("\033[?1049l", end="", file=sys.stderr, flush=True)
99 print(colorize("\nForce shutdown requested", "38;5;208"), file=sys.stderr)
100 sys.exit(1)
101 _loop_shutdown_requested = True
102 print(colorize("\nShutdown requested, will exit after current state...", "33"), file=sys.stderr)
103 if _loop_executor is not None:
104 _loop_executor.request_shutdown()
105 # Kill any child subprocess currently blocking in the action runner
106 inner = getattr(_loop_executor, "_executor", None)
107 if inner is not None:
108 runner = getattr(inner, "action_runner", None)
109 if runner is not None:
110 proc = getattr(runner, "_current_process", None)
111 if proc is not None:
112 proc.kill()
113 # Also kill MCP subprocesses tracked directly on FSMExecutor (_run_subprocess path)
114 fsm_proc = getattr(inner, "_current_process", None)
115 if fsm_proc is not None:
116 fsm_proc.kill()
119def _is_earliest_waiter(entry_id: str, queue_dir: Path) -> bool:
120 """Return True if entry_id is the earliest-enqueued waiter in queue_dir.
122 Returns True when this waiter is first or the queue is empty/unreadable,
123 allowing it to proceed with acquire(). Non-first waiters return False and
124 should back off to yield to the earlier waiter (ENH-1332).
126 Stale entries (dead PIDs) are removed on the fly so orphaned queue files
127 from crashed processes do not block live waiters indefinitely (BUG-1360).
128 """
129 if not queue_dir.exists():
130 return True
131 entries: list[dict] = []
132 for f in queue_dir.glob("*.json"):
133 try:
134 with open(f) as fh:
135 data = json.load(fh)
136 pid = data.get("context", {}).get("pid")
137 if pid is not None and not _process_alive(pid):
138 f.unlink(missing_ok=True)
139 continue
140 entries.append(data)
141 except (json.JSONDecodeError, KeyError, FileNotFoundError, OSError):
142 continue
143 if not entries:
144 return True
145 entries.sort(key=lambda d: d.get("enqueuedAt", ""))
146 return entries[0].get("id") == entry_id
149def register_loop_signal_handlers(executor: Any, pid_file: Path | None = None) -> None:
150 """Register SIGINT/SIGTERM handlers for graceful loop shutdown.
152 Sets up signal handling so that Ctrl-C triggers a graceful shutdown
153 (calls executor.request_shutdown()) rather than raising KeyboardInterrupt.
154 A second Ctrl-C forces immediate exit with PID file cleanup.
156 Args:
157 executor: The PersistentExecutor instance to request shutdown on.
158 pid_file: Optional path to PID file to clean up on forced exit.
159 """
160 global _loop_shutdown_requested, _loop_executor, _loop_pid_file
161 _loop_shutdown_requested = False
162 _loop_executor = executor
163 _loop_pid_file = pid_file
164 signal.signal(signal.SIGINT, _loop_signal_handler)
165 signal.signal(signal.SIGTERM, _loop_signal_handler)
168# ---------------------------------------------------------------------------
169# SIGWINCH handler (alt-screen pinned-pane mode only)
170# ---------------------------------------------------------------------------
173def _sigwinch_handler(signum: int, frame: FrameType | None) -> None:
174 """Mark the pinned pane as needing a redraw after a terminal resize.
176 The handler does the minimum amount of work safe to perform in a signal
177 context: it just sets a flag. ``display_progress`` consumes the flag
178 before processing the next FSM event and triggers the actual redraw.
179 """
180 global _needs_redraw
181 _needs_redraw = True
184def _install_sigwinch_handler() -> None:
185 """Install ``_sigwinch_handler`` for SIGWINCH, stashing the prior handler.
187 Idempotent — re-calling while installed is a no-op so we never leak the
188 chain by overwriting our own stash. No-op on platforms without SIGWINCH
189 (e.g. Windows).
190 """
191 global _original_sigwinch
192 if not hasattr(signal, "SIGWINCH"):
193 return
194 if _original_sigwinch is not None:
195 return
196 _original_sigwinch = signal.signal(signal.SIGWINCH, _sigwinch_handler)
199def _restore_sigwinch_handler() -> None:
200 """Restore the SIGWINCH handler stashed by ``_install_sigwinch_handler``.
202 Safe to call when no handler was installed. After this returns,
203 ``_original_sigwinch`` is reset to ``None`` so the install is repeatable.
204 """
205 global _original_sigwinch, _needs_redraw
206 if not hasattr(signal, "SIGWINCH"):
207 _original_sigwinch = None
208 _needs_redraw = False
209 return
210 if _original_sigwinch is None:
211 return
212 signal.signal(signal.SIGWINCH, _original_sigwinch)
213 _original_sigwinch = None
214 _needs_redraw = False
217# ---------------------------------------------------------------------------
218# Pinned-pane layout (alt-screen mode)
219# ---------------------------------------------------------------------------
222def _count_display_lines(text: str) -> int:
223 """Return the number of terminal rows a string occupies.
225 Treats each ``\\n`` as a row boundary; a trailing newline is *not*
226 counted as an extra empty row. ANSI escape sequences are assumed not to
227 contain newlines (true for SGR / cursor / scroll-region codes).
228 """
229 if not text:
230 return 0
231 return text.count("\n") + (0 if text.endswith("\n") else 1)
234def _choose_pinned_layout(
235 rows: int,
236 variants: list[str],
237 min_action_rows: int = MIN_ACTION_ROWS,
238) -> tuple[str, int]:
239 """Pick the most detailed pinned-pane variant that leaves room for action output.
241 ``variants`` is an ordered list from most-detailed to least-detailed
242 (e.g. ``[full, neighborhood, single_line]``). Returns
243 ``(pinned_str, line_count)`` for the first variant whose height plus
244 ``min_action_rows`` fits within ``rows``. If none fit, returns the
245 last (smallest) variant unchanged — a degenerate terminal still gets
246 *something* pinned.
247 """
248 last_text = ""
249 last_h = 0
250 for variant in variants:
251 last_text = variant
252 last_h = _count_display_lines(variant)
253 if last_h + min_action_rows <= rows:
254 return variant, last_h
255 return last_text, last_h
258def _render_single_line_status(fsm: FSMLoop, active_state: str | None) -> str:
259 """Render the single-line fallback: ``fsm: <preds> → [<active>] → <succs>``."""
260 from little_loops.cli.loop.layout import _collect_edges
262 if active_state is None or active_state not in fsm.states:
263 return f"fsm: · → [{active_state or '?'}] → ·"
264 edges = _collect_edges(fsm)
265 preds = sorted({s for (s, t, _lbl) in edges if t == active_state and s != active_state})
266 succs = sorted({t for (s, t, _lbl) in edges if s == active_state and t != active_state})
267 preds_s = ",".join(preds) if preds else "·"
268 succs_s = ",".join(succs) if succs else "·"
269 return f"fsm: {preds_s} → [{active_state}] → {succs_s}"
272def _build_pinned_pane(
273 detail: str,
274 fsm: FSMLoop,
275 parent_highlight: str | None,
276 child_fsm_stack: dict[int, FSMLoop | None],
277 last_state_at_depth: dict[int, str],
278 iteration_line: str,
279 cols: int,
280 *,
281 facets: DiagramFacets,
282 highlight_color: str,
283 edge_label_colors: dict[str, str] | None,
284 badges: dict[str, str] | None,
285 prev_highlight: str | None = None,
286 prev_state_at_depth: dict[int, str] | None = None,
287) -> str:
288 """Compose the pinned pane (header + diagram(s) + state line + separator).
290 ``detail`` selects the diagram variant: ``"full"`` (layered
291 ``_render_fsm_diagram``), ``"neighborhood"`` (1-hop pred/active/succ), or
292 ``"single"`` (one-line ``fsm:`` status). The returned string is intended
293 to be printed with ``flush=True`` and is terminated by a horizontal
294 separator (no trailing newline).
295 """
296 from little_loops.cli.loop.layout import (
297 _collect_edges,
298 _filter_main_path_graph,
299 _render_fsm_diagram,
300 _render_neighborhood_diagram,
301 )
303 verbose = facets.scope == "full" and facets.state_detail == "full"
305 def _render_one(target: FSMLoop, highlight: str | None, prev: str | None) -> str:
306 if detail == "single":
307 return _render_single_line_status(target, highlight)
308 if detail == "neighborhood":
309 return _render_neighborhood_diagram(
310 target,
311 highlight or target.initial,
312 edge_label_colors=edge_label_colors,
313 badges=badges,
314 highlight_color=highlight_color,
315 mode=facets.scope,
316 prev_state=prev,
317 )
318 # "full" (layered)
319 scope = facets.scope
320 if scope == "main" and highlight is not None:
321 _filtered_edges, reachable = _filter_main_path_graph(target, _collect_edges(target))
322 if highlight not in reachable:
323 scope = "full"
324 return _render_fsm_diagram(
325 target,
326 verbose=verbose,
327 highlight_state=highlight,
328 highlight_color=highlight_color,
329 edge_label_colors=edge_label_colors,
330 badges=badges,
331 mode=scope,
332 suppress_labels=not facets.edge_labels,
333 title_only=facets.state_detail == "title",
334 )
336 prev_map = prev_state_at_depth or {}
337 lines: list[str] = []
339 # Find deepest active loop — show only that one instead of stacking all levels.
340 active_fsm = fsm
341 active_state = parent_highlight
342 active_prev = prev_highlight
343 active_depth = 0
344 for d in sorted(child_fsm_stack.keys()):
345 child = child_fsm_stack[d]
346 if child is not None and (d + 1) in last_state_at_depth:
347 active_fsm = child
348 active_state = last_state_at_depth.get(d + 1)
349 active_prev = prev_map.get(d + 1)
350 active_depth = d + 1
352 # Header: breadcrumb shows immediate parent when inside a sub-loop.
353 if active_depth > 0:
354 imm_parent_name = (
355 fsm.name if active_depth == 1 else (child_fsm_stack.get(active_depth - 2) or fsm).name
356 )
357 imm_parent_state = last_state_at_depth.get(active_depth - 1, "")
358 header_text = f"== loop: {active_fsm.name} ({imm_parent_name} › {imm_parent_state}) "
359 else:
360 header_text = f"== loop: {fsm.name} "
361 lines.append(header_text + "=" * max(0, cols - len(header_text)))
362 diagram = _render_one(active_fsm, active_state, active_prev)
363 if diagram:
364 lines.extend(diagram.split("\n"))
366 lines.append(iteration_line)
367 lines.append("─" * cols)
368 return "\n".join(lines)
371def _render_pinned_pane(
372 fsm: FSMLoop,
373 parent_highlight: str | None,
374 child_fsm_stack: dict[int, FSMLoop | None],
375 last_state_at_depth: dict[int, str],
376 iteration_line: str,
377 *,
378 facets: DiagramFacets,
379 highlight_color: str,
380 edge_label_colors: dict[str, str] | None,
381 badges: dict[str, str] | None,
382 min_action_rows: int = MIN_ACTION_ROWS,
383 prev_state_at_depth: dict[int, str] | None = None,
384) -> int:
385 """Render the pinned pane to stdout and set the scroll region beneath it.
387 Performs (in order): reset scroll region, clear+home cursor, build all
388 pinned-pane variants, pick the largest that fits, print it, set the
389 DECSTBM scroll region to start one row below the pinned pane, and
390 position the cursor at the top of that scroll region. Returns the
391 pinned-pane height in rows so the caller can track it across events.
392 """
393 cols, rows = terminal_size()
394 # 1. Reset any existing scroll region so the clear covers the full screen.
395 print("\033[r", end="", flush=True)
396 # 2. Clear + cursor home.
397 print("\033[2J\033[H", end="", flush=True)
399 prev_map = prev_state_at_depth or {}
401 def _build(detail: str) -> str:
402 return _build_pinned_pane(
403 detail,
404 fsm,
405 parent_highlight,
406 child_fsm_stack,
407 last_state_at_depth,
408 iteration_line,
409 cols,
410 facets=facets,
411 highlight_color=highlight_color,
412 edge_label_colors=edge_label_colors,
413 badges=badges,
414 prev_highlight=prev_map.get(0),
415 prev_state_at_depth=prev_map,
416 )
418 # Build the fallback ladder based on facets source and topology.
419 # Explicit topology (source="topology"): render exactly once, no degradation.
420 # Preset/default with layered topology: skip neighborhood — layered→neighborhood
421 # looks like a broken diagram and confuses users. Go straight to single-line.
422 # Explicit neighborhood topology: neighborhood→single.
423 topo_detail = TOPOLOGY_TO_DETAIL[facets.topology]
424 if facets.source == "topology":
425 variants = [_build(topo_detail)]
426 elif topo_detail == "full":
427 variants = [_build("full"), _build("single")]
428 elif topo_detail == "neighborhood":
429 variants = [_build("neighborhood"), _build("single")]
430 else: # inline / single
431 variants = [_build("single")]
433 pinned, pinned_height = _choose_pinned_layout(
434 rows,
435 variants,
436 min_action_rows=min_action_rows,
437 )
438 print(pinned, flush=True)
440 # Guard against degenerate terminals: scroll region must have at least
441 # 1 row beneath the pinned pane. If pinned_height >= rows, skip the
442 # scroll-region setup entirely — output will append normally and the
443 # caller can rely on the alt-screen exit to clean up.
444 if pinned_height < rows:
445 # DECSTBM uses 1-indexed inclusive rows.
446 print(f"\033[{pinned_height + 1};{rows}r", end="", flush=True)
447 # Move cursor to the top of the scroll region.
448 print(f"\033[{pinned_height + 1};1H", end="", flush=True)
449 return pinned_height
452class StateFeedRenderer:
453 """Renders loop-state events as terminal output for foreground runs and monitor attach.
455 Extracted from ``run_foreground()`` so both the foreground run path and the
456 ``cmd_monitor`` attach path (FEAT-1764) can share the same rendering logic.
457 """
459 def __init__(
460 self,
461 fsm: FSMLoop,
462 args: argparse.Namespace,
463 highlight_color: str = "32",
464 edge_label_colors: dict[str, str] | None = None,
465 badges: dict[str, str] | None = None,
466 loops_dir: Path | None = None,
467 ) -> None:
468 self.fsm = fsm
469 self.args = args
470 self.highlight_color = highlight_color
471 self.edge_label_colors = edge_label_colors
472 self.badges = badges
473 self.loops_dir = loops_dir or Path(".")
475 # Derived from args
476 self.quiet: bool = getattr(args, "quiet", False)
477 self.verbose: bool = getattr(args, "verbose", False)
478 self.facets: DiagramFacets | None = resolve_facets(args)
479 self.show_diagrams: bool = self.facets is not None
480 self.clear_screen: bool = getattr(args, "clear", False)
481 self.in_pinned_mode: bool = (
482 self.show_diagrams and self.clear_screen and sys.stdout.isatty()
483 )
485 # Mutable state (was closure-captured in run_foreground)
486 self.current_iteration: list[int] = [0]
487 self.last_state_at_depth: dict[int, str] = {}
488 self.prev_state_at_depth: dict[int, str] = {}
489 self.child_fsm_stack: dict[int, FSMLoop | None] = {}
490 self.pinned_height: list[int] = [0]
491 self.loop_start_time: float = time.monotonic()
493 def _elapsed_str(self) -> str:
494 elapsed_int = int(time.monotonic() - self.loop_start_time)
495 if elapsed_int < 60:
496 return f"{elapsed_int}s"
497 return f"{elapsed_int // 60}m {elapsed_int % 60}s"
499 def _redraw_pinned(self, state0: str) -> None:
500 """Redraw the pinned pane in place using the current depth-0 state."""
501 assert self.facets is not None
502 iter_line = (
503 f"[{self.current_iteration[0]}/{self.fsm.max_iterations}] "
504 f"{colorize(state0, '1')} ({colorize(self._elapsed_str(), '2')})"
505 )
506 self.pinned_height[0] = _render_pinned_pane(
507 self.fsm,
508 state0,
509 self.child_fsm_stack,
510 self.last_state_at_depth,
511 iter_line,
512 facets=self.facets,
513 highlight_color=self.highlight_color,
514 edge_label_colors=self.edge_label_colors,
515 badges=self.badges,
516 prev_state_at_depth=self.prev_state_at_depth,
517 )
519 def handle_event(self, event: dict) -> None:
520 """Display progress for events."""
521 global _needs_redraw
522 # SIGWINCH redraw: terminal was resized; re-render the pinned pane
523 # before processing the next event so the layout matches the new size.
524 if _needs_redraw and self.in_pinned_mode and 0 in self.last_state_at_depth:
525 self._redraw_pinned(self.last_state_at_depth[0])
526 _needs_redraw = False
528 event_type = event.get("event")
529 depth = event.get("depth", 0)
530 indent = " " * depth
531 tw = terminal_width()
532 max_line = tw - 8 - len(indent)
534 if event_type == "state_enter":
535 self.current_iteration[0] = event.get("iteration", 0)
536 state = event.get("state", "")
537 elapsed_str = self._elapsed_str() if not self.quiet else ""
538 # Non-pinned --clear path keeps the bare full-screen clear.
539 if (
540 self.clear_screen
541 and sys.stdout.isatty()
542 and depth == 0
543 and not self.in_pinned_mode
544 ):
545 print("\033[2J\033[H", end="", flush=True)
546 # Update last-known state at this depth and clear stale deeper entries.
547 old_state = self.last_state_at_depth.get(depth)
548 if old_state is not None and old_state != state:
549 self.prev_state_at_depth[depth] = old_state
550 self.last_state_at_depth[depth] = state
551 for k in [k for k in self.last_state_at_depth if k > depth]:
552 del self.last_state_at_depth[k]
553 self.prev_state_at_depth.pop(k, None)
554 # Load child FSM for the current state at this depth
555 parent_at_depth = self.fsm if depth == 0 else self.child_fsm_stack.get(depth - 1)
556 if parent_at_depth is not None and state in parent_at_depth.states:
557 fsm_state = parent_at_depth.states[state]
558 if fsm_state.loop is not None:
559 try:
560 self.child_fsm_stack[depth] = load_loop(
561 fsm_state.loop, self.loops_dir, Logger()
562 )
563 except (FileNotFoundError, ValueError):
564 pass
565 else:
566 self.child_fsm_stack[depth] = None
567 else:
568 self.child_fsm_stack[depth] = None
569 # Clear stale deeper child FSM entries
570 for k in [k for k in self.child_fsm_stack if k > depth]:
571 del self.child_fsm_stack[k]
573 if self.in_pinned_mode:
574 state0 = self.last_state_at_depth.get(0)
575 if state0 is None:
576 state0 = state
577 self._redraw_pinned(state0)
578 elif self.show_diagrams:
579 from little_loops.cli.loop.layout import (
580 _collect_edges,
581 _filter_main_path_graph,
582 _render_fsm_diagram,
583 )
585 assert self.facets is not None
587 # Find deepest active loop — show only that one.
588 active_fsm_diag = self.fsm
589 active_highlight = self.last_state_at_depth.get(0)
590 active_depth_diag = 0
591 for d in sorted(self.child_fsm_stack.keys()):
592 child_at_d = self.child_fsm_stack[d]
593 if child_at_d is not None and (d + 1) in self.last_state_at_depth:
594 active_fsm_diag = child_at_d
595 active_highlight = self.last_state_at_depth.get(d + 1)
596 active_depth_diag = d + 1
598 # Fall back to full scope when the highlighted state is hidden in main scope.
599 active_scope = self.facets.scope
600 fallback_note: str | None = None
601 if active_scope == "main" and active_highlight is not None:
602 _filtered_edges, active_reachable = _filter_main_path_graph(
603 active_fsm_diag, _collect_edges(active_fsm_diag)
604 )
605 if active_highlight not in active_reachable:
606 active_scope = "full"
607 fallback_note = (
608 f"(showing full diagram: active state "
609 f"{active_highlight!r} is off the main path)"
610 )
611 diagram = _render_fsm_diagram(
612 active_fsm_diag,
613 highlight_state=active_highlight,
614 highlight_color=self.highlight_color,
615 edge_label_colors=self.edge_label_colors,
616 badges=self.badges,
617 mode=active_scope,
618 suppress_labels=not self.facets.edge_labels,
619 title_only=self.facets.state_detail == "title",
620 )
621 # Header: breadcrumb shows immediate parent when inside a sub-loop.
622 if active_depth_diag > 0:
623 imm_parent_name = (
624 self.fsm.name
625 if active_depth_diag == 1
626 else (self.child_fsm_stack.get(active_depth_diag - 2) or self.fsm).name
627 )
628 imm_parent_state = self.last_state_at_depth.get(active_depth_diag - 1, "")
629 header_text = f"== loop: {active_fsm_diag.name} ({imm_parent_name} › {imm_parent_state}) "
630 else:
631 header_text = f"== loop: {self.fsm.name} "
632 header = header_text + "=" * max(0, tw - len(header_text))
633 print(header, flush=True)
634 if fallback_note is not None:
635 print(fallback_note, flush=True)
636 print(diagram, flush=True)
637 # In pinned mode the iteration line is part of the pinned pane;
638 # only print it inline for non-pinned paths.
639 if not self.quiet and not self.in_pinned_mode:
640 print(
641 f"{indent}[{self.current_iteration[0]}/{self.fsm.max_iterations}] {colorize(state, '1')} ({colorize(elapsed_str, '2')})",
642 end="",
643 flush=True,
644 )
646 elif event_type == "action_start":
647 if not self.quiet:
648 action = event.get("action", "")
649 is_prompt = event.get("is_prompt", False)
650 if is_prompt:
651 lines = action.strip().splitlines()
652 line_count = len(lines)
653 prompt_badge = "✦" # ✦
654 if self.verbose:
655 print(
656 f"{indent} -> {colorize(prompt_badge, '2')} {colorize(f'({line_count} lines)', '2')}",
657 flush=True,
658 )
659 for line in lines:
660 print(f"{indent} {line}", flush=True)
661 else:
662 first_line = lines[0] if lines else ""
663 preview = (
664 first_line[:60] + "..." if len(first_line) > 60 else first_line
665 )
666 print(
667 f"{indent} -> {colorize(prompt_badge, '2')} {colorize(preview, '2')}",
668 flush=True,
669 )
670 else:
671 if self.verbose:
672 action_display = action
673 else:
674 action_display = (
675 action[:max_line] + "..." if len(action) > max_line else action
676 )
677 print(f"{indent} -> {colorize(action_display, '2')}", flush=True)
679 elif event_type == "action_output":
680 if not self.quiet:
681 line = event.get("line", "")
682 if line.strip():
683 print(f"{indent} {line}", flush=True)
685 elif event_type == "action_complete":
686 if not self.quiet:
687 duration_ms = event.get("duration_ms", 0)
688 exit_code = event.get("exit_code", 0)
689 duration_sec = duration_ms / 1000
690 if duration_sec < 60:
691 duration_str = f"{duration_sec:.1f}s"
692 else:
693 minutes = int(duration_sec // 60)
694 seconds = duration_sec % 60
695 duration_str = f"{minutes}m {seconds:.0f}s"
696 parts = [f"{indent} ({colorize(duration_str, '2')})"]
697 if exit_code == 124:
698 parts.append(colorize("timed out", "38;5;208"))
699 elif exit_code != 0:
700 parts.append(colorize(f"exit: {exit_code}", "38;5;208"))
701 print(" ".join(parts), flush=True)
703 elif event_type == "evaluate":
704 if not self.quiet:
705 verdict = event.get("verdict", "")
706 confidence = event.get("confidence")
707 reason = event.get("reason", "")
708 error = event.get("error", "")
709 _elc = self.edge_label_colors or {}
710 if verdict in ("yes", "target", "progress"):
711 _vc = _elc.get("yes", "32")
712 symbol = colorize("✓", _vc)
713 verdict_colored = colorize(verdict, _vc)
714 elif verdict == "no":
715 _vc = _elc.get("no", "38;5;208")
716 symbol = colorize("✗", _vc)
717 verdict_colored = colorize(verdict, _vc)
718 elif verdict == "error":
719 _vc = _elc.get("error", "38;5;208")
720 symbol = colorize("✗", _vc)
721 verdict_colored = colorize(verdict, _vc)
722 else:
723 symbol = colorize("✗", "38;5;208")
724 verdict_colored = colorize(verdict, "2")
725 # Build verdict line
726 if error and verdict == "error":
727 verdict_line = f"{symbol} {verdict_colored}: {error}"
728 elif confidence is not None:
729 verdict_line = (
730 f"{symbol} {verdict_colored} {colorize(f'({confidence:.2f})', '2')}"
731 )
732 else:
733 verdict_line = f"{symbol} {verdict_colored}"
734 print(f"{indent} {verdict_line}", flush=True)
735 # Show raw_preview for error verdicts to aid diagnosis
736 raw_preview = event.get("raw_preview", "")
737 if raw_preview and verdict == "error":
738 if self.verbose:
739 sub_lines = raw_preview.splitlines() or [""]
740 first, rest = sub_lines[0], sub_lines[1:]
741 print(f"{indent} raw: {first}", flush=True)
742 for sub in rest:
743 print(f"{indent} {sub}", flush=True)
744 else:
745 print(f"{indent} raw: {raw_preview[:200]}", flush=True)
746 # Show reason on a second line if present (and not already shown as error)
747 if reason and not (error and verdict == "error"):
748 if self.verbose:
749 for sub in reason.splitlines() or [""]:
750 print(f"{indent} {sub}", flush=True)
751 else:
752 reason_display = reason[:300] + "..." if len(reason) > 300 else reason
753 print(f"{indent} {reason_display}", flush=True)
755 elif event_type == "route":
756 if not self.quiet:
757 to_state = event.get("to", "")
758 print(
759 f"{indent} {colorize('->', '2')} {colorize(to_state, '1')}",
760 flush=True,
761 )
763 elif event_type == "max_iterations_summary":
764 if not self.quiet:
765 summary_state = event.get("summary_state", "")
766 iters = event.get("iterations", 0)
767 msg = (
768 f"iteration cap reached ({iters}); running summary state '{summary_state}'"
769 )
770 print(f"{indent} {colorize(msg, '38;5;208')}", flush=True)
772 elif event_type == "stall_detected":
773 if not self.quiet:
774 state = event.get("state", "")
775 exit_code = event.get("exit_code", 0)
776 verdict = event.get("verdict", "")
777 consecutive = event.get("consecutive", 0)
778 action = event.get("action", "abort")
779 triple = f"(exit_code={exit_code}, verdict='{verdict}')"
780 msg = (
781 f"stall_detected: state '{state}' produced {triple} "
782 f"for {consecutive} consecutive iterations -> {action}"
783 )
784 print(f"{indent} {colorize(msg, '38;5;208')}", flush=True)
787def get_builtin_loops_dir() -> Path:
788 """Get the path to built-in loops bundled with the plugin."""
789 return Path(__file__).parent.parent.parent / "loops"
792def resolve_loop_path(name_or_path: str, loops_dir: Path) -> Path:
793 """Resolve loop name to file path."""
794 path = Path(name_or_path)
795 if path.exists():
796 return path
798 # Try <loops_dir>/<name>.fsm.yaml first (compiled FSM)
799 fsm_path = loops_dir / f"{name_or_path}.fsm.yaml"
800 if fsm_path.exists():
801 return fsm_path
803 # Fall back to <loops_dir>/<name>.yaml
804 loops_path = loops_dir / f"{name_or_path}.yaml"
805 if loops_path.exists():
806 return loops_path
808 # Fall back to built-in loops from plugin directory
809 builtin_path = get_builtin_loops_dir() / f"{name_or_path}.yaml"
810 if builtin_path.exists():
811 return builtin_path
813 raise FileNotFoundError(f"Loop not found: {name_or_path}")
816def load_loop(name_or_path: str, loops_dir: Path, logger: Logger) -> FSMLoop:
817 """Load and validate a loop.
819 Raises:
820 FileNotFoundError: If loop not found.
821 ValueError: If loop is invalid.
822 """
823 from little_loops.fsm.validation import load_and_validate
825 path = resolve_loop_path(name_or_path, loops_dir)
826 fsm, _ = load_and_validate(path)
827 return fsm
830def load_loop_with_spec(
831 name_or_path: str, loops_dir: Path, logger: Logger
832) -> tuple[FSMLoop, dict[str, Any]]:
833 """Load a loop and return both the FSMLoop and raw spec dict.
835 Used by commands that need access to raw YAML fields (e.g., description).
837 Raises:
838 FileNotFoundError: If loop not found.
839 ValueError: If loop is invalid.
840 """
841 import yaml
843 from little_loops.fsm.validation import load_and_validate
845 path = resolve_loop_path(name_or_path, loops_dir)
847 with open(path) as f:
848 spec = yaml.safe_load(f)
850 fsm, _ = load_and_validate(path)
851 return fsm, spec
854def print_execution_plan(fsm: FSMLoop, edge_label_colors: dict[str, str] | None = None) -> None:
855 """Print dry-run execution plan."""
856 _elc = edge_label_colors or {}
857 _yes_color = _elc.get("yes", "32")
858 tw = terminal_width()
859 print(colorize(f"Execution plan for: {fsm.name}", "1"))
860 print()
861 print("States:")
862 for name, state in fsm.states.items():
863 terminal_marker = colorize(" [TERMINAL]", _yes_color) if state.terminal else ""
864 print(f" {colorize(f'[{name}]', '1')}{terminal_marker}")
865 if state.action:
866 if state.action_type == "prompt":
867 lines = state.action.strip().splitlines()
868 preview = "\n ".join(lines[:3])
869 if len(lines) > 3 or len(state.action) > 200:
870 preview += " ..."
871 print(f" action: |\n {preview}")
872 else:
873 max_action = tw - 16
874 action_display = (
875 state.action[:max_action] + "..."
876 if len(state.action) > max_action
877 else state.action
878 )
879 print(f" action: {action_display}")
880 if state.evaluate:
881 print(f" evaluate: {state.evaluate.type}")
882 if state.on_yes:
883 print(f" on_yes {colorize('->', '2')} {colorize(state.on_yes, '2')}")
884 if state.on_no:
885 print(f" on_no {colorize('->', '2')} {colorize(state.on_no, '2')}")
886 if state.on_error:
887 print(f" on_error {colorize('->', '2')} {colorize(state.on_error, '2')}")
888 if state.next:
889 print(f" next {colorize('->', '2')} {colorize(state.next, '2')}")
890 if state.route:
891 print(" route:")
892 for verdict, target in state.route.routes.items():
893 print(f" {verdict} {colorize('->', '2')} {colorize(target, '2')}")
894 if state.route.default:
895 print(f" _ {colorize('->', '2')} {colorize(state.route.default, '2')}")
896 print()
897 print(f"Initial state: {fsm.initial}")
898 print(f"Max iterations: {fsm.max_iterations}")
899 if fsm.timeout:
900 print(f"Timeout: {fsm.timeout}s")
901 if fsm.context:
902 print("Context:")
903 for key, value in fsm.context.items():
904 print(f" {key}: {value!r}")
907def _make_instance_id(loop_name: str) -> str:
908 """Generate a unique instance ID for a loop run."""
909 return f"{loop_name}-{datetime.now().strftime('%Y%m%dT%H%M%S')}"
912def run_background(
913 loop_name: str, args: argparse.Namespace, loops_dir: Path, subcommand: str = "run"
914) -> int:
915 """Launch loop as a detached background process.
917 Spawns a new process with start_new_session=True that re-executes
918 the loop with --foreground-internal. The parent writes the PID file
919 and returns immediately.
921 Args:
922 subcommand: The ll-loop subcommand to spawn ("run" or "resume").
924 Returns:
925 Exit code (0 = launched successfully).
926 """
927 running_dir = loops_dir / ".running"
928 running_dir.mkdir(parents=True, exist_ok=True)
930 instance_id = _make_instance_id(loop_name)
931 pid_file = running_dir / f"{instance_id}.pid"
932 log_file = running_dir / f"{instance_id}.log"
934 # Build re-exec command with --foreground-internal instead of --background
935 cmd = [
936 sys.executable,
937 "-m",
938 "little_loops.cli.loop",
939 subcommand,
940 loop_name,
941 ]
942 input_val = getattr(args, "input", None)
943 if input_val is not None:
944 cmd.append(input_val)
945 cmd.append("--foreground-internal")
946 cmd.extend(["--instance-id", instance_id])
948 # Forward relevant args
949 max_iter = getattr(args, "max_iterations", None)
950 if max_iter:
951 cmd.extend(["--max-iterations", str(max_iter)])
952 if getattr(args, "no_llm", False):
953 cmd.append("--no-llm")
954 llm_model = getattr(args, "llm_model", None)
955 if llm_model:
956 cmd.extend(["--llm-model", llm_model])
957 if getattr(args, "verbose", False):
958 cmd.append("--verbose")
959 show_diagrams_raw = getattr(args, "show_diagrams", None)
960 if show_diagrams_raw is not None:
961 if show_diagrams_raw is True:
962 cmd.append("--show-diagrams")
963 else:
964 cmd.extend(["--show-diagrams", show_diagrams_raw])
965 diagram_edge_labels = getattr(args, "diagram_edge_labels", None)
966 if diagram_edge_labels is not None:
967 cmd.extend(["--diagram-edge-labels", diagram_edge_labels])
968 diagram_state_detail = getattr(args, "diagram_state_detail", None)
969 if diagram_state_detail is not None:
970 cmd.extend(["--diagram-state-detail", diagram_state_detail])
971 diagram_scope = getattr(args, "diagram_scope", None)
972 if diagram_scope is not None:
973 cmd.extend(["--diagram-scope", diagram_scope])
974 if getattr(args, "quiet", False):
975 cmd.append("--quiet")
976 if getattr(args, "queue", False):
977 cmd.append("--queue")
978 for kv in getattr(args, "context", None) or []:
979 cmd.extend(["--context", kv])
980 program_md = getattr(args, "program_md", None)
981 if program_md is not None:
982 cmd.extend(["--program-md", str(program_md)])
983 delay = getattr(args, "delay", None)
984 if delay is not None:
985 cmd.extend(["--delay", str(delay)])
986 handoff_threshold = getattr(args, "handoff_threshold", None)
987 if handoff_threshold is not None:
988 cmd.extend(["--handoff-threshold", str(handoff_threshold)])
989 context_limit = getattr(args, "context_limit", None)
990 if context_limit is not None:
991 cmd.extend(["--context-limit", str(context_limit)])
993 with open(log_file, "w") as log_fh:
994 process = subprocess.Popen(
995 cmd,
996 start_new_session=True,
997 stdout=log_fh,
998 stderr=log_fh,
999 stdin=subprocess.DEVNULL,
1000 )
1002 pid_file.write_text(str(process.pid))
1003 print(
1004 f"Loop {colorize(loop_name, '1')} started in background (PID: {colorize(str(process.pid), '2')})"
1005 )
1006 print(f" Log: {colorize(str(log_file), '2')}")
1007 print(f" Status: {colorize(f'll-loop status {loop_name}', '2')}")
1008 print(f" Stop: {colorize(f'll-loop stop {loop_name}', '2')}")
1009 return 0
1012def run_foreground(
1013 executor: Any,
1014 fsm: FSMLoop,
1015 args: argparse.Namespace,
1016 highlight_color: str = "32",
1017 edge_label_colors: dict[str, str] | None = None,
1018 badges: dict[str, str] | None = None,
1019 mode: str = "run",
1020 instance_id: str | None = None,
1021 running_dir: Path | None = None,
1022) -> int:
1023 """Run loop with progress display.
1025 Args:
1026 highlight_color: ANSI SGR code for the active FSM state highlight in verbose mode.
1027 edge_label_colors: Optional label→SGR-code mapping for transition edge labels.
1028 badges: Optional glyph-key→string mapping for state type badges in FSM diagrams.
1029 mode: ``"run"`` (default) calls ``executor.run()``; ``"resume"`` calls
1030 ``executor.resume()`` so a resumed loop reuses the same display-wiring
1031 path as a fresh run (BUG-1645). In ``"resume"`` mode a ``None`` result
1032 from ``executor.resume()`` is treated as "nothing to resume": a warning
1033 is logged and exit code 1 is returned before any alt-screen sequences
1034 are emitted.
1036 instance_id: When provided and not a background-spawned child
1037 (``foreground_internal=False``), stdout/stderr are teed to
1038 ``{running_dir}/{instance_id}.log`` with ANSI sequences stripped.
1039 running_dir: Directory for the log file. Defaults to ``.running``
1040 alongside the loops directory.
1041 Returns:
1042 Exit code (0 = success).
1043 """
1044 if mode not in ("run", "resume"):
1045 raise ValueError(f"run_foreground: invalid mode {mode!r}; expected 'run' or 'resume'")
1047 _orig_stdout = sys.stdout
1048 _orig_stderr = sys.stderr
1049 _log_fh: Any = None
1050 if instance_id is not None and not getattr(args, "foreground_internal", False):
1051 _log_path = (running_dir or Path(".running")) / f"{instance_id}.log"
1052 _log_fh = open(_log_path, "w")
1053 sys.stdout = _TeeWriter(_orig_stdout, _log_fh) # type: ignore[assignment]
1054 sys.stderr = _TeeWriter(_orig_stderr, _log_fh) # type: ignore[assignment]
1056 try:
1057 # Create the state feed renderer — encapsulates display state and event handling.
1058 renderer = StateFeedRenderer(
1059 fsm,
1060 args,
1061 highlight_color=highlight_color,
1062 edge_label_colors=edge_label_colors,
1063 badges=badges,
1064 loops_dir=getattr(executor, "loops_dir", Path(".")),
1065 )
1066 if not renderer.quiet:
1067 print(f"Running loop: {colorize(fsm.name, '1')}")
1068 print(f"Max iterations: {colorize(str(fsm.max_iterations), '2')}")
1069 print()
1071 # Wire progress display via the EventBus on PersistentExecutor
1072 if not renderer.quiet or renderer.show_diagrams:
1073 if hasattr(executor, "event_bus"):
1074 executor.event_bus.register(renderer.handle_event)
1075 else:
1076 executor._on_event = renderer.handle_event
1078 # Wire follow mode — streams history-formatted events independently of quiet
1079 if getattr(args, "follow", False):
1080 from little_loops.cli.loop.info import _format_history_event
1082 tw = terminal_width()
1083 _verbose = renderer.verbose
1085 def _follow_callback(event: dict[str, Any]) -> None:
1086 line = _format_history_event(event, verbose=_verbose, width=tw)
1087 if line is not None:
1088 print(line, flush=True)
1090 if hasattr(executor, "event_bus"):
1091 executor.event_bus.register(_follow_callback)
1092 else:
1093 _prev_on_event = executor._on_event
1095 def _chained(event: dict[str, Any]) -> None:
1096 if _prev_on_event:
1097 _prev_on_event(event)
1098 _follow_callback(event)
1100 executor._on_event = _chained
1102 # Enter alternate screen buffer when showing diagrams with clear to prevent
1103 # scrollback contamination from diagrams taller than the terminal height.
1104 global _using_alt_screen
1105 if renderer.show_diagrams and renderer.clear_screen and sys.stdout.isatty():
1106 _using_alt_screen = True
1107 print("\033[?1049h\033[H", end="", flush=True)
1108 _install_sigwinch_handler()
1110 try:
1111 if mode == "resume":
1112 result = executor.resume()
1113 # "Nothing to resume" path: no run actually executed, so don't fall
1114 # through to completion-line formatting. Exit cleanly with code 1.
1115 if result is None:
1116 Logger().warning(f"Nothing to resume for: {fsm.name}")
1117 return 1
1118 else:
1119 result = executor.run()
1120 finally:
1121 if _using_alt_screen:
1122 # Reset DECSTBM scroll region BEFORE exiting alt-screen, otherwise
1123 # the main buffer is left with a restricted scroll region (one of
1124 # the Success Metrics for ENH-1642).
1125 print("\033[r", end="", flush=True)
1126 print("\033[?1049l", end="", flush=True)
1127 _using_alt_screen = False
1128 _restore_sigwinch_handler()
1130 if not renderer.quiet:
1131 print()
1132 duration_sec = result.duration_ms / 1000
1133 if duration_sec < 60:
1134 duration_str = f"{duration_sec:.1f}s"
1135 else:
1136 minutes = int(duration_sec // 60)
1137 seconds = duration_sec % 60
1138 duration_str = f"{minutes}m {seconds:.0f}s"
1139 if result.terminated_by == "terminal":
1140 state_colored = colorize(result.final_state, "32")
1141 else:
1142 state_colored = colorize(result.final_state, "38;5;208")
1143 completion_prefix = "Resumed and completed" if mode == "resume" else "Loop completed"
1144 rejection_count = 0
1145 for _t in getattr(getattr(executor, "event_bus", None), "_transports", []):
1146 if hasattr(_t, "get_stats"):
1147 rejection_count += _t.get_stats().get("client_rejections", 0)
1148 suffix = f", {rejection_count} client rejections" if rejection_count > 0 else ""
1149 print(
1150 f"{completion_prefix}: {state_colored} ({result.iterations} iterations, {duration_str}{suffix})"
1151 )
1153 return EXIT_CODES.get(result.terminated_by, 1)
1154 finally:
1155 sys.stdout = _orig_stdout
1156 sys.stderr = _orig_stderr
1157 if _log_fh is not None:
1158 _log_fh.close()