Coverage for little_loops / cli / loop / _helpers.py: 7%
884 statements
« prev ^ index » next coverage.py v7.12.0, created at 2026-06-26 17:38 -0500
1"""Shared helpers for ll-loop CLI subcommands."""
3from __future__ import annotations
5import argparse
6import json
7import signal
8import subprocess
9import sys
10import time
11from datetime import datetime
12from pathlib import Path
13from types import FrameType
14from typing import TYPE_CHECKING, Any
16from little_loops.cli.loop.diagram_modes import (
17 TOPOLOGY_TO_DETAIL,
18 DiagramFacets,
19 resolve_facets,
20)
21from little_loops.cli.output import colorize, strip_ansi, terminal_size, terminal_width
22from little_loops.fsm.concurrency import LockManager, _process_alive, resolve_scope
23from little_loops.logger import Logger
24from little_loops.pricing import estimate_cost_usd
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_steps": 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
58class _TeeWriter:
59 """Wraps a stream, writing to both the original and a log file (ANSI stripped on log)."""
61 def __init__(self, stream: Any, log_fh: Any) -> None:
62 self._stream = stream
63 self._log_fh = log_fh
65 def write(self, data: str) -> int:
66 n = self._stream.write(data)
67 self._log_fh.write(strip_ansi(data))
68 return n
70 def flush(self) -> None:
71 self._stream.flush()
72 self._log_fh.flush()
74 def __getattr__(self, name: str) -> Any:
75 return getattr(self._stream, name)
78def _loop_signal_handler(signum: int, frame: FrameType | None) -> None:
79 """Handle shutdown signals gracefully for ll-loop.
81 First signal: Set shutdown flag for graceful exit after current state.
82 Second signal: Force immediate exit.
83 """
84 global _loop_shutdown_requested, _using_alt_screen
85 if _loop_shutdown_requested:
86 # Second signal - force exit
87 if _loop_pid_file is not None:
88 _loop_pid_file.unlink(missing_ok=True)
89 if _using_alt_screen:
90 # Reset the DECSTBM scroll region BEFORE exiting the alt screen,
91 # otherwise the main buffer is left with a restricted scroll
92 # region — visible to the user as `seq 1 50` failing to scroll
93 # past the previous pinned-pane height.
94 print("\033[r", end="", file=sys.stderr, flush=True)
95 print("\033[?1049l", end="", file=sys.stderr, flush=True)
96 print(colorize("\nForce shutdown requested", "38;5;208"), file=sys.stderr)
97 sys.exit(1)
98 _loop_shutdown_requested = True
99 print(colorize("\nShutdown requested, will exit after current state...", "33"), file=sys.stderr)
100 if _loop_executor is not None:
101 _loop_executor.request_shutdown()
102 # Kill any child subprocess currently blocking in the action runner
103 inner = getattr(_loop_executor, "_executor", None)
104 if inner is not None:
105 runner = getattr(inner, "action_runner", None)
106 if runner is not None:
107 proc = getattr(runner, "_current_process", None)
108 if proc is not None:
109 proc.kill()
110 # Also kill MCP subprocesses tracked directly on FSMExecutor (_run_subprocess path)
111 fsm_proc = getattr(inner, "_current_process", None)
112 if fsm_proc is not None:
113 fsm_proc.kill()
116def _is_earliest_waiter(entry_id: str, queue_dir: Path) -> bool:
117 """Return True if entry_id is the earliest-enqueued waiter in queue_dir.
119 Returns True when this waiter is first or the queue is empty/unreadable,
120 allowing it to proceed with acquire(). Non-first waiters return False and
121 should back off to yield to the earlier waiter (ENH-1332).
123 Stale entries (dead PIDs) are removed on the fly so orphaned queue files
124 from crashed processes do not block live waiters indefinitely (BUG-1360).
125 """
126 if not queue_dir.exists():
127 return True
128 entries: list[dict] = []
129 for f in queue_dir.glob("*.json"):
130 try:
131 with open(f) as fh:
132 data = json.load(fh)
133 pid = data.get("context", {}).get("pid")
134 if pid is not None and not _process_alive(pid):
135 f.unlink(missing_ok=True)
136 continue
137 entries.append(data)
138 except (json.JSONDecodeError, KeyError, FileNotFoundError, OSError):
139 continue
140 if not entries:
141 return True
142 entries.sort(key=lambda d: d.get("enqueuedAt", ""))
143 return entries[0].get("id") == entry_id
146def register_loop_signal_handlers(executor: Any, pid_file: Path | None = None) -> None:
147 """Register SIGINT/SIGTERM handlers for graceful loop shutdown.
149 Sets up signal handling so that Ctrl-C triggers a graceful shutdown
150 (calls executor.request_shutdown()) rather than raising KeyboardInterrupt.
151 A second Ctrl-C forces immediate exit with PID file cleanup.
153 Args:
154 executor: The PersistentExecutor instance to request shutdown on.
155 pid_file: Optional path to PID file to clean up on forced exit.
156 """
157 global _loop_shutdown_requested, _loop_executor, _loop_pid_file
158 _loop_shutdown_requested = False
159 _loop_executor = executor
160 _loop_pid_file = pid_file
161 signal.signal(signal.SIGINT, _loop_signal_handler)
162 signal.signal(signal.SIGTERM, _loop_signal_handler)
165# ---------------------------------------------------------------------------
166# SIGWINCH handler (alt-screen pinned-pane mode only)
167# ---------------------------------------------------------------------------
170def _sigwinch_handler(signum: int, frame: FrameType | None) -> None:
171 """Mark the pinned pane as needing a redraw after a terminal resize.
173 The handler does the minimum amount of work safe to perform in a signal
174 context: it just sets a flag. ``display_progress`` consumes the flag
175 before processing the next FSM event and triggers the actual redraw.
176 """
177 global _needs_redraw
178 _needs_redraw = True
181def _install_sigwinch_handler() -> None:
182 """Install ``_sigwinch_handler`` for SIGWINCH, stashing the prior handler.
184 Idempotent — re-calling while installed is a no-op so we never leak the
185 chain by overwriting our own stash. No-op on platforms without SIGWINCH
186 (e.g. Windows).
187 """
188 global _original_sigwinch
189 if not hasattr(signal, "SIGWINCH"):
190 return
191 if _original_sigwinch is not None:
192 return
193 _original_sigwinch = signal.signal(signal.SIGWINCH, _sigwinch_handler)
196def _restore_sigwinch_handler() -> None:
197 """Restore the SIGWINCH handler stashed by ``_install_sigwinch_handler``.
199 Safe to call when no handler was installed. After this returns,
200 ``_original_sigwinch`` is reset to ``None`` so the install is repeatable.
201 """
202 global _original_sigwinch, _needs_redraw
203 if not hasattr(signal, "SIGWINCH"):
204 _original_sigwinch = None
205 _needs_redraw = False
206 return
207 if _original_sigwinch is None:
208 return
209 signal.signal(signal.SIGWINCH, _original_sigwinch)
210 _original_sigwinch = None
211 _needs_redraw = False
214# ---------------------------------------------------------------------------
215# Pinned-pane layout (alt-screen mode)
216# ---------------------------------------------------------------------------
219def _count_display_lines(text: str) -> int:
220 """Return the number of terminal rows a string occupies.
222 Treats each ``\\n`` as a row boundary; a trailing newline is *not*
223 counted as an extra empty row. ANSI escape sequences are assumed not to
224 contain newlines (true for SGR / cursor / scroll-region codes).
225 """
226 if not text:
227 return 0
228 return text.count("\n") + (0 if text.endswith("\n") else 1)
231def _choose_pinned_layout(
232 rows: int,
233 variants: list[str],
234 min_action_rows: int = MIN_ACTION_ROWS,
235) -> tuple[str, int]:
236 """Pick the most detailed pinned-pane variant that leaves room for action output.
238 ``variants`` is an ordered list from most-detailed to least-detailed
239 (e.g. ``[full, neighborhood, single_line]``). Returns
240 ``(pinned_str, line_count)`` for the first variant whose height plus
241 ``min_action_rows`` fits within ``rows``. If none fit, returns the
242 last (smallest) variant unchanged — a degenerate terminal still gets
243 *something* pinned.
244 """
245 last_text = ""
246 last_h = 0
247 for variant in variants:
248 last_text = variant
249 last_h = _count_display_lines(variant)
250 if last_h + min_action_rows <= rows:
251 return variant, last_h
252 return last_text, last_h
255def _render_single_line_status(fsm: FSMLoop, active_state: str | None) -> str:
256 """Render the single-line fallback: ``fsm: <preds> → [<active>] → <succs>``."""
257 from little_loops.cli.loop.layout import _collect_edges
259 if active_state is None or active_state not in fsm.states:
260 return f"fsm: · → [{active_state or '?'}] → ·"
261 edges = _collect_edges(fsm)
262 preds = sorted({s for (s, t, _lbl) in edges if t == active_state and s != active_state})
263 succs = sorted({t for (s, t, _lbl) in edges if s == active_state and t != active_state})
264 preds_s = ",".join(preds) if preds else "·"
265 succs_s = ",".join(succs) if succs else "·"
266 return f"fsm: {preds_s} → [{active_state}] → {succs_s}"
269def _build_pinned_pane(
270 detail: str,
271 fsm: FSMLoop,
272 parent_highlight: str | None,
273 child_fsm_stack: dict[int, FSMLoop | None],
274 last_state_at_depth: dict[int, str],
275 iteration_line: str,
276 cols: int,
277 *,
278 facets: DiagramFacets,
279 highlight_color: str,
280 edge_label_colors: dict[str, str] | None,
281 badges: dict[str, str] | None,
282 prev_highlight: str | None = None,
283 prev_state_at_depth: dict[int, str] | None = None,
284 loop_path: Path | None = None,
285 model: str | None = None,
286) -> str:
287 """Compose the pinned pane (header + diagram(s) + state line + separator).
289 ``detail`` selects the diagram variant: ``"full"`` (layered
290 ``_render_fsm_diagram``), ``"neighborhood"`` (1-hop pred/active/succ), or
291 ``"single"`` (one-line ``fsm:`` status). The returned string is intended
292 to be printed with ``flush=True`` and is terminated by a horizontal
293 separator (no trailing newline).
294 """
295 from little_loops.cli.loop.layout import (
296 _collect_edges,
297 _filter_main_path_graph,
298 _render_fsm_diagram,
299 _render_neighborhood_diagram,
300 )
302 verbose = facets.scope == "full" and facets.state_detail == "full"
304 def _render_one(target: FSMLoop, highlight: str | None, prev: str | None) -> str:
305 if detail == "single":
306 return _render_single_line_status(target, highlight)
307 if detail == "neighborhood":
308 return _render_neighborhood_diagram(
309 target,
310 highlight or target.initial,
311 edge_label_colors=edge_label_colors,
312 badges=badges,
313 highlight_color=highlight_color,
314 mode=facets.scope,
315 prev_state=prev,
316 )
317 # "full" (layered)
318 scope = facets.scope
319 if scope == "main" and highlight is not None:
320 _filtered_edges, reachable = _filter_main_path_graph(target, _collect_edges(target))
321 if highlight not in reachable:
322 scope = "full"
323 return _render_fsm_diagram(
324 target,
325 verbose=verbose,
326 highlight_state=highlight,
327 highlight_color=highlight_color,
328 edge_label_colors=edge_label_colors,
329 badges=badges,
330 mode=scope,
331 suppress_labels=not facets.edge_labels,
332 title_only=facets.state_detail == "title",
333 )
335 prev_map = prev_state_at_depth or {}
336 lines: list[str] = []
338 # Find deepest active loop — show only that one instead of stacking all levels.
339 active_fsm = fsm
340 active_state = parent_highlight
341 active_prev = prev_highlight
342 active_depth = 0
343 for d in sorted(child_fsm_stack.keys()):
344 child = child_fsm_stack[d]
345 if child is not None and (d + 1) in last_state_at_depth:
346 active_fsm = child
347 active_state = last_state_at_depth.get(d + 1)
348 active_prev = prev_map.get(d + 1)
349 active_depth = d + 1
351 # Header: breadcrumb shows immediate parent when inside a sub-loop.
352 if active_depth > 0:
353 imm_parent_name = (
354 fsm.name if active_depth == 1 else (child_fsm_stack.get(active_depth - 2) or fsm).name
355 )
356 imm_parent_state = last_state_at_depth.get(active_depth - 1, "")
357 header_text = f"== loop: {active_fsm.name} ({imm_parent_name} › {imm_parent_state}) "
358 else:
359 header_text = f"== loop: {fsm.name} "
360 lines.append(header_text + "=" * max(0, cols - len(header_text)))
361 for key, value in _artifact_lines(fsm, loop_path):
362 lines.append(f" {key}: {colorize(value, '2')}")
363 if model is not None:
364 lines.append(f" model: {colorize(model, '2')}")
365 diagram = _render_one(active_fsm, active_state, active_prev)
366 if diagram:
367 lines.extend(diagram.split("\n"))
369 lines.append(iteration_line)
370 lines.append("─" * cols)
371 return "\n".join(lines)
374def _render_pinned_pane(
375 fsm: FSMLoop,
376 parent_highlight: str | None,
377 child_fsm_stack: dict[int, FSMLoop | None],
378 last_state_at_depth: dict[int, str],
379 iteration_line: str,
380 *,
381 facets: DiagramFacets,
382 highlight_color: str,
383 edge_label_colors: dict[str, str] | None,
384 badges: dict[str, str] | None,
385 min_action_rows: int = MIN_ACTION_ROWS,
386 prev_state_at_depth: dict[int, str] | None = None,
387 loop_path: Path | None = None,
388 model: str | None = None,
389) -> int:
390 """Render the pinned pane to stdout and set the scroll region beneath it.
392 Performs (in order): reset scroll region, clear+home cursor, build all
393 pinned-pane variants, pick the largest that fits, print it, set the
394 DECSTBM scroll region to start one row below the pinned pane, and
395 position the cursor at the top of that scroll region. Returns the
396 pinned-pane height in rows so the caller can track it across events.
397 """
398 cols, rows = terminal_size()
399 # 1. Reset any existing scroll region so the clear covers the full screen.
400 print("\033[r", end="", flush=True)
401 # 2. Clear + cursor home.
402 print("\033[2J\033[H", end="", flush=True)
404 prev_map = prev_state_at_depth or {}
406 def _build(detail: str) -> str:
407 return _build_pinned_pane(
408 detail,
409 fsm,
410 parent_highlight,
411 child_fsm_stack,
412 last_state_at_depth,
413 iteration_line,
414 cols,
415 facets=facets,
416 highlight_color=highlight_color,
417 edge_label_colors=edge_label_colors,
418 badges=badges,
419 prev_highlight=prev_map.get(0),
420 prev_state_at_depth=prev_map,
421 loop_path=loop_path,
422 model=model,
423 )
425 # Build the fallback ladder based on facets source and topology.
426 # Explicit topology (source="topology"): render exactly once, no degradation.
427 # Preset/default with layered topology: full → neighborhood → single.
428 # Explicit neighborhood topology: neighborhood→single.
429 topo_detail = TOPOLOGY_TO_DETAIL[facets.topology]
430 if facets.source == "topology":
431 variants = [_build(topo_detail)]
432 elif topo_detail == "full":
433 variants = [_build("full"), _build("neighborhood"), _build("single")]
434 elif topo_detail == "neighborhood":
435 variants = [_build("neighborhood"), _build("single")]
436 else: # inline / single
437 variants = [_build("single")]
439 # compact presets (clean/slim) set state_detail="title" and already sacrifice
440 # action body lines; tolerate fewer rows below the diagram to avoid the
441 # single-line fsm: fallback on larger loops.
442 effective_min_action_rows = 3 if facets.state_detail == "title" else min_action_rows
443 pinned, pinned_height = _choose_pinned_layout(
444 rows,
445 variants,
446 min_action_rows=effective_min_action_rows,
447 )
448 print(pinned, flush=True)
450 # Guard against degenerate terminals: scroll region must have at least
451 # 1 row beneath the pinned pane. If pinned_height >= rows, skip the
452 # scroll-region setup entirely — output will append normally and the
453 # caller can rely on the alt-screen exit to clean up.
454 if pinned_height < rows:
455 # DECSTBM uses 1-indexed inclusive rows.
456 print(f"\033[{pinned_height + 1};{rows}r", end="", flush=True)
457 # Move cursor to the top of the scroll region.
458 print(f"\033[{pinned_height + 1};1H", end="", flush=True)
459 return pinned_height
462class StateFeedRenderer:
463 """Renders loop-state events as terminal output for foreground runs and monitor attach.
465 Extracted from ``run_foreground()`` so both the foreground run path and the
466 ``cmd_monitor`` attach path (FEAT-1764) can share the same rendering logic.
467 """
469 def __init__(
470 self,
471 fsm: FSMLoop,
472 args: argparse.Namespace,
473 highlight_color: str = "32",
474 edge_label_colors: dict[str, str] | None = None,
475 badges: dict[str, str] | None = None,
476 loops_dir: Path | None = None,
477 loop_path: Path | None = None,
478 model: str | None = None,
479 ) -> None:
480 self.fsm = fsm
481 self.args = args
482 self.highlight_color = highlight_color
483 self.edge_label_colors = edge_label_colors
484 self.badges = badges
485 self.loops_dir = loops_dir or Path(".")
486 self.loop_path = loop_path
487 self.model = model
489 # Derived from args
490 self.quiet: bool = getattr(args, "quiet", False)
491 self.verbose: bool = getattr(args, "verbose", False)
492 self.facets: DiagramFacets | None = resolve_facets(args)
493 self.show_diagrams: bool = self.facets is not None
494 self.clear_screen: bool = getattr(args, "clear", False)
495 self.in_pinned_mode: bool = self.show_diagrams and self.clear_screen and sys.stdout.isatty()
497 # Mutable state (was closure-captured in run_foreground)
498 self.current_iteration: list[int] = [0]
499 self.last_state_at_depth: dict[int, str] = {}
500 self.prev_state_at_depth: dict[int, str] = {}
501 self.child_fsm_stack: dict[int, FSMLoop | None] = {}
502 self.pinned_height: list[int] = [0]
503 self.loop_start_time: float = time.monotonic()
505 def _elapsed_str(self) -> str:
506 elapsed_int = int(time.monotonic() - self.loop_start_time)
507 if elapsed_int < 60:
508 return f"{elapsed_int}s"
509 return f"{elapsed_int // 60}m {elapsed_int % 60}s"
511 def _redraw_pinned(self, state0: str) -> None:
512 """Redraw the pinned pane in place using the current depth-0 state."""
513 assert self.facets is not None
514 iter_line = (
515 f"[{self.current_iteration[0]}/{self.fsm.max_steps}] "
516 f"{colorize(state0, '1')} ({colorize(self._elapsed_str(), '2')})"
517 )
518 self.pinned_height[0] = _render_pinned_pane(
519 self.fsm,
520 state0,
521 self.child_fsm_stack,
522 self.last_state_at_depth,
523 iter_line,
524 facets=self.facets,
525 highlight_color=self.highlight_color,
526 edge_label_colors=self.edge_label_colors,
527 badges=self.badges,
528 prev_state_at_depth=self.prev_state_at_depth,
529 loop_path=self.loop_path,
530 model=self.model,
531 )
533 def handle_event(self, event: dict) -> None:
534 """Display progress for events."""
535 global _needs_redraw
536 # SIGWINCH redraw: terminal was resized; re-render the pinned pane
537 # before processing the next event so the layout matches the new size.
538 if _needs_redraw and self.in_pinned_mode and 0 in self.last_state_at_depth:
539 self._redraw_pinned(self.last_state_at_depth[0])
540 _needs_redraw = False
542 event_type = event.get("event")
543 depth = event.get("depth", 0)
544 indent = " " * depth
545 tw = terminal_width()
546 max_line = tw - 8 - len(indent)
548 if event_type == "state_enter":
549 self.current_iteration[0] = event.get("iteration", 0)
550 state = event.get("state", "")
551 elapsed_str = self._elapsed_str() if not self.quiet else ""
552 # Non-pinned --clear path keeps the bare full-screen clear.
553 if self.clear_screen and sys.stdout.isatty() and depth == 0 and not self.in_pinned_mode:
554 print("\033[2J\033[H", end="", flush=True)
555 # Update last-known state at this depth and clear stale deeper entries.
556 old_state = self.last_state_at_depth.get(depth)
557 if old_state is not None and old_state != state:
558 self.prev_state_at_depth[depth] = old_state
559 self.last_state_at_depth[depth] = state
560 for k in [k for k in self.last_state_at_depth if k > depth]:
561 del self.last_state_at_depth[k]
562 self.prev_state_at_depth.pop(k, None)
563 # Load child FSM for the current state at this depth
564 parent_at_depth = self.fsm if depth == 0 else self.child_fsm_stack.get(depth - 1)
565 if parent_at_depth is not None and state in parent_at_depth.states:
566 fsm_state = parent_at_depth.states[state]
567 if fsm_state.loop is not None:
568 try:
569 self.child_fsm_stack[depth] = load_loop(
570 fsm_state.loop, self.loops_dir, Logger()
571 )
572 except (FileNotFoundError, ValueError) as e:
573 Logger().warning(
574 f"Could not load child loop '{fsm_state.loop}' for state '{state}': {e}"
575 )
576 self.child_fsm_stack[depth] = None
577 else:
578 self.child_fsm_stack[depth] = None
579 else:
580 self.child_fsm_stack[depth] = None
581 # Clear stale deeper child FSM entries
582 for k in [k for k in self.child_fsm_stack if k > depth]:
583 del self.child_fsm_stack[k]
585 if self.in_pinned_mode:
586 state0 = self.last_state_at_depth.get(0)
587 if state0 is None:
588 state0 = state
589 self._redraw_pinned(state0)
590 elif self.show_diagrams:
591 from little_loops.cli.loop.layout import (
592 _collect_edges,
593 _filter_main_path_graph,
594 _render_fsm_diagram,
595 )
597 assert self.facets is not None
599 # Find deepest active loop — show only that one.
600 active_fsm_diag = self.fsm
601 active_highlight = self.last_state_at_depth.get(0)
602 active_depth_diag = 0
603 for d in sorted(self.child_fsm_stack.keys()):
604 child_at_d = self.child_fsm_stack[d]
605 if child_at_d is not None and (d + 1) in self.last_state_at_depth:
606 active_fsm_diag = child_at_d
607 active_highlight = self.last_state_at_depth.get(d + 1)
608 active_depth_diag = d + 1
610 # Fall back to full scope when the highlighted state is hidden in main scope.
611 active_scope = self.facets.scope
612 fallback_note: str | None = None
613 if active_scope == "main" and active_highlight is not None:
614 _filtered_edges, active_reachable = _filter_main_path_graph(
615 active_fsm_diag, _collect_edges(active_fsm_diag)
616 )
617 if active_highlight not in active_reachable:
618 active_scope = "full"
619 fallback_note = (
620 f"(showing full diagram: active state "
621 f"{active_highlight!r} is off the main path)"
622 )
623 diagram = _render_fsm_diagram(
624 active_fsm_diag,
625 highlight_state=active_highlight,
626 highlight_color=self.highlight_color,
627 edge_label_colors=self.edge_label_colors,
628 badges=self.badges,
629 mode=active_scope,
630 suppress_labels=not self.facets.edge_labels,
631 title_only=self.facets.state_detail == "title",
632 )
633 # Header: breadcrumb shows immediate parent when inside a sub-loop.
634 if active_depth_diag > 0:
635 imm_parent_name = (
636 self.fsm.name
637 if active_depth_diag == 1
638 else (self.child_fsm_stack.get(active_depth_diag - 2) or self.fsm).name
639 )
640 imm_parent_state = self.last_state_at_depth.get(active_depth_diag - 1, "")
641 header_text = (
642 f"== loop: {active_fsm_diag.name} ({imm_parent_name} › {imm_parent_state}) "
643 )
644 else:
645 header_text = f"== loop: {self.fsm.name} "
646 header = header_text + "=" * max(0, tw - len(header_text))
647 print(header, flush=True)
648 if fallback_note is not None:
649 print(fallback_note, flush=True)
650 for key, value in _artifact_lines(self.fsm, self.loop_path):
651 print(f" {key}: {colorize(value, '2')}", flush=True)
652 if self.model is not None:
653 print(f" model: {colorize(self.model, '2')}", flush=True)
654 print(diagram, flush=True)
655 # In pinned mode the iteration line is part of the pinned pane;
656 # only print it inline for non-pinned paths.
657 if not self.quiet and not self.in_pinned_mode:
658 print(
659 f"{indent}[{self.current_iteration[0]}/{self.fsm.max_steps}] {colorize(state, '1')} ({colorize(elapsed_str, '2')})",
660 end="",
661 flush=True,
662 )
664 elif event_type == "action_start":
665 if not self.quiet:
666 action = event.get("action", "")
667 is_prompt = event.get("is_prompt", False)
668 if is_prompt:
669 lines = action.strip().splitlines()
670 line_count = len(lines)
671 prompt_badge = "✦" # ✦
672 if self.verbose:
673 print(
674 f"{indent} -> {colorize(prompt_badge, '2')} {colorize(f'({line_count} lines)', '2')}",
675 flush=True,
676 )
677 for line in lines:
678 print(f"{indent} {line}", flush=True)
679 else:
680 first_line = lines[0] if lines else ""
681 preview = first_line[:60] + "..." if len(first_line) > 60 else first_line
682 print(
683 f"{indent} -> {colorize(prompt_badge, '2')} {colorize(preview, '2')}",
684 flush=True,
685 )
686 else:
687 if self.verbose:
688 action_display = action
689 else:
690 action_display = (
691 action[:max_line] + "..." if len(action) > max_line else action
692 )
693 print(f"{indent} -> {colorize(action_display, '2')}", flush=True)
695 elif event_type == "action_output":
696 if not self.quiet:
697 line = event.get("line", "")
698 if line.strip():
699 print(f"{indent} {line}", flush=True)
701 elif event_type == "action_complete":
702 actual_model = event.get("model")
703 if actual_model:
704 self.model = actual_model
705 if not self.quiet:
706 duration_ms = event.get("duration_ms", 0)
707 exit_code = event.get("exit_code", 0)
708 duration_sec = duration_ms / 1000
709 if duration_sec < 60:
710 duration_str = f"{duration_sec:.1f}s"
711 else:
712 minutes = int(duration_sec // 60)
713 seconds = duration_sec % 60
714 duration_str = f"{minutes}m {seconds:.0f}s"
715 parts = [f"{indent} ({colorize(duration_str, '2')})"]
716 if exit_code == 124:
717 parts.append(colorize("timed out", "38;5;208"))
718 elif exit_code != 0:
719 parts.append(colorize(f"exit: {exit_code}", "38;5;208"))
720 print(" ".join(parts), flush=True)
722 elif event_type == "baseline_complete":
723 if not self.quiet:
724 h_ms = event.get("harness_duration_ms", 0)
725 b_ms = event.get("baseline_duration_ms", 0)
726 h_tok = event.get("harness_tokens", 0)
727 b_tok = event.get("baseline_tokens", 0)
728 print(
729 f"{indent} baseline: {colorize(f'{b_ms / 1000:.1f}s', '2')}, "
730 f"{colorize(str(b_tok), '2')} tokens | "
731 f"harness: {colorize(f'{h_ms / 1000:.1f}s', '2')}, "
732 f"{colorize(str(h_tok), '2')} tokens",
733 flush=True,
734 )
736 elif event_type == "evaluate":
737 if not self.quiet:
738 verdict = event.get("verdict", "")
739 confidence = event.get("confidence")
740 reason = event.get("reason", "")
741 error = event.get("error", "")
742 _elc = self.edge_label_colors or {}
743 if verdict in ("yes", "target", "progress"):
744 _vc = _elc.get("yes", "32")
745 symbol = colorize("✓", _vc)
746 verdict_colored = colorize(verdict, _vc)
747 elif verdict == "no":
748 _vc = _elc.get("no", "38;5;208")
749 symbol = colorize("✗", _vc)
750 verdict_colored = colorize(verdict, _vc)
751 elif verdict == "error":
752 _vc = _elc.get("error", "38;5;208")
753 symbol = colorize("✗", _vc)
754 verdict_colored = colorize(verdict, _vc)
755 else:
756 symbol = colorize("✗", "38;5;208")
757 verdict_colored = colorize(verdict, "2")
758 # Build verdict line
759 if error and verdict == "error":
760 verdict_line = f"{symbol} {verdict_colored}: {error}"
761 elif confidence is not None:
762 verdict_line = (
763 f"{symbol} {verdict_colored} {colorize(f'({confidence:.2f})', '2')}"
764 )
765 else:
766 verdict_line = f"{symbol} {verdict_colored}"
767 print(f"{indent} {verdict_line}", flush=True)
768 # Show raw_preview for error verdicts to aid diagnosis
769 raw_preview = event.get("raw_preview", "")
770 if raw_preview and verdict == "error":
771 if self.verbose:
772 sub_lines = raw_preview.splitlines() or [""]
773 first, rest = sub_lines[0], sub_lines[1:]
774 print(f"{indent} raw: {first}", flush=True)
775 for sub in rest:
776 print(f"{indent} {sub}", flush=True)
777 else:
778 print(f"{indent} raw: {raw_preview[:200]}", flush=True)
779 # Show reason on a second line if present (and not already shown as error)
780 if reason and not (error and verdict == "error"):
781 if self.verbose:
782 for sub in reason.splitlines() or [""]:
783 print(f"{indent} {sub}", flush=True)
784 else:
785 reason_display = reason[:300] + "..." if len(reason) > 300 else reason
786 print(f"{indent} {reason_display}", flush=True)
788 elif event_type == "route":
789 if not self.quiet:
790 to_state = event.get("to", "")
791 print(
792 f"{indent} {colorize('->', '2')} {colorize(to_state, '1')}",
793 flush=True,
794 )
796 elif event_type == "max_steps_summary":
797 if not self.quiet:
798 summary_state = event.get("summary_state", "")
799 iters = event.get("iterations", 0)
800 msg = f"step cap reached ({iters}); running summary state '{summary_state}'"
801 print(f"{indent} {colorize(msg, '38;5;208')}", flush=True)
803 elif event_type == "max_iterations_reached_summary":
804 if not self.quiet:
805 summary_state = event.get("summary_state", "")
806 iteration_count = event.get("iteration_count", 0)
807 msg = f"iteration cap reached ({iteration_count}); running summary state '{summary_state}'"
808 print(f"{indent} {colorize(msg, '38;5;208')}", flush=True)
810 elif event_type == "stall_detected":
811 if not self.quiet:
812 state = event.get("state", "")
813 exit_code = event.get("exit_code", 0)
814 verdict = event.get("verdict", "")
815 consecutive = event.get("consecutive", 0)
816 action = event.get("action", "abort")
817 triple = f"(exit_code={exit_code}, verdict='{verdict}')"
818 msg = (
819 f"stall_detected: state '{state}' produced {triple} "
820 f"for {consecutive} consecutive iterations -> {action}"
821 )
822 print(f"{indent} {colorize(msg, '38;5;208')}", flush=True)
825def get_builtin_loops_dir() -> Path:
826 """Get the path to built-in loops bundled with the plugin."""
827 return Path(__file__).parent.parent.parent / "loops"
830def _artifact_lines(fsm: FSMLoop, loop_path: Path | None) -> list[tuple[str, str]]:
831 """Extract path-like context values from *fsm* for display in artifact headers.
833 Returns a list of ``(key, value)`` pairs where *value* is a non-empty string
834 that starts with ``.``, ``/``, or ``~``, or contains ``/``, and does not
835 contain ``${`` (unresolved template expression). When *loop_path* is not
836 ``None``, the first entry is always ``("loop", str(loop_path))``.
837 """
838 pairs: list[tuple[str, str]] = []
839 if loop_path is not None:
840 pairs.append(("loop", str(loop_path)))
841 context: dict[str, Any] = getattr(fsm, "context", None) or {}
842 for key, value in context.items():
843 if not isinstance(value, str) or not value:
844 continue
845 if "${" in value:
846 continue
847 if value.startswith(".") or value.startswith("/") or value.startswith("~") or "/" in value:
848 pairs.append((key, value))
849 return pairs
852def resolve_loop_path(name_or_path: str, loops_dir: Path) -> Path:
853 """Resolve loop name to file path."""
854 path = Path(name_or_path)
855 if path.exists():
856 return path
858 # Try <loops_dir>/<name>.fsm.yaml first (compiled FSM)
859 fsm_path = loops_dir / f"{name_or_path}.fsm.yaml"
860 if fsm_path.exists():
861 return fsm_path
863 # Fall back to <loops_dir>/<name>.yaml
864 loops_path = loops_dir / f"{name_or_path}.yaml"
865 if loops_path.exists():
866 return loops_path
868 # Fall back to built-in loops from plugin directory
869 builtin_path = get_builtin_loops_dir() / f"{name_or_path}.yaml"
870 if builtin_path.exists():
871 return builtin_path
873 raise FileNotFoundError(f"Loop not found: {name_or_path}")
876def load_loop(name_or_path: str, loops_dir: Path, logger: Logger) -> FSMLoop:
877 """Load and validate a loop.
879 Raises:
880 FileNotFoundError: If loop not found.
881 ValueError: If loop is invalid.
882 """
883 from little_loops.fsm.validation import load_and_validate
885 path = resolve_loop_path(name_or_path, loops_dir)
886 fsm, _ = load_and_validate(path)
887 return fsm
890def load_loop_with_spec(
891 name_or_path: str, loops_dir: Path, logger: Logger
892) -> tuple[FSMLoop, dict[str, Any]]:
893 """Load a loop and return both the FSMLoop and raw spec dict.
895 Used by commands that need access to raw YAML fields (e.g., description).
897 Raises:
898 FileNotFoundError: If loop not found.
899 ValueError: If loop is invalid.
900 """
901 import yaml
903 from little_loops.fsm.validation import load_and_validate
905 path = resolve_loop_path(name_or_path, loops_dir)
907 with open(path) as f:
908 spec = yaml.safe_load(f)
910 fsm, _ = load_and_validate(path)
911 return fsm, spec
914def print_execution_plan(fsm: FSMLoop, edge_label_colors: dict[str, str] | None = None) -> None:
915 """Print dry-run execution plan."""
916 _elc = edge_label_colors or {}
917 _yes_color = _elc.get("yes", "32")
918 tw = terminal_width()
919 print(colorize(f"Execution plan for: {fsm.name}", "1"))
920 print()
921 print("States:")
922 for name, state in fsm.states.items():
923 terminal_marker = colorize(" [TERMINAL]", _yes_color) if state.terminal else ""
924 print(f" {colorize(f'[{name}]', '1')}{terminal_marker}")
925 if state.action:
926 if state.action_type == "prompt":
927 lines = state.action.strip().splitlines()
928 preview = "\n ".join(lines[:3])
929 if len(lines) > 3 or len(state.action) > 200:
930 preview += " ..."
931 print(f" action: |\n {preview}")
932 else:
933 max_action = tw - 16
934 action_display = (
935 state.action[:max_action] + "..."
936 if len(state.action) > max_action
937 else state.action
938 )
939 print(f" action: {action_display}")
940 if state.evaluate:
941 print(f" evaluate: {state.evaluate.type}")
942 if state.on_yes:
943 print(f" on_yes {colorize('->', '2')} {colorize(state.on_yes, '2')}")
944 if state.on_no:
945 print(f" on_no {colorize('->', '2')} {colorize(state.on_no, '2')}")
946 if state.on_error:
947 print(f" on_error {colorize('->', '2')} {colorize(state.on_error, '2')}")
948 if state.next:
949 print(f" next {colorize('->', '2')} {colorize(state.next, '2')}")
950 if state.route:
951 print(" route:")
952 for verdict, target in state.route.routes.items():
953 print(f" {verdict} {colorize('->', '2')} {colorize(target, '2')}")
954 if state.route.default:
955 print(f" _ {colorize('->', '2')} {colorize(state.route.default, '2')}")
956 print()
957 print(f"Initial state: {fsm.initial}")
958 print(f"Max steps: {fsm.max_steps}")
959 if fsm.max_iterations is not None:
960 print(f"Max iterations: {fsm.max_iterations}")
961 if fsm.timeout:
962 print(f"Timeout: {fsm.timeout}s")
963 if fsm.context:
964 print("Context:")
965 for key, value in fsm.context.items():
966 print(f" {key}: {value!r}")
969def _make_instance_id(loop_name: str) -> str:
970 """Generate a unique instance ID for a loop run."""
971 return f"{loop_name}-{datetime.now().strftime('%Y%m%dT%H%M%S')}"
974def run_background(
975 loop_name: str,
976 args: argparse.Namespace,
977 loops_dir: Path,
978 subcommand: str = "run",
979 instance_id: str | None = None,
980) -> int:
981 """Launch loop as a detached background process.
983 Spawns a new process with start_new_session=True that re-executes
984 the loop with --foreground-internal. The parent writes the PID file
985 and returns immediately.
987 Args:
988 subcommand: The ll-loop subcommand to spawn ("run" or "resume").
989 instance_id: Pre-resolved instance ID. When provided, skips
990 _make_instance_id() allocation. Used by cmd_resume() to pass
991 an already-discovered resumable instance.
993 Returns:
994 Exit code (0 = launched successfully).
995 """
996 running_dir = loops_dir / ".running"
997 running_dir.mkdir(parents=True, exist_ok=True)
999 # Pre-flight scope conflict check — detect conflicts before spawning the child
1000 # so the user gets immediate feedback instead of a silent child failure.
1001 logger = Logger()
1002 try:
1003 fsm = load_loop(loop_name, loops_dir, logger)
1004 except (FileNotFoundError, ValueError) as e:
1005 print(f"Error loading loop '{loop_name}': {e}", file=sys.stderr)
1006 return 1
1008 lock_manager = LockManager(loops_dir)
1009 # Build context for scope resolution: YAML defaults + CLI --context overrides.
1010 # CLI --context is only forwarded to the child process (line ~1018); we parse
1011 # it locally so resolve_scope() can use it for the pre-flight conflict check.
1012 scope_context = dict(fsm.context)
1013 for kv in getattr(args, "context", None) or []:
1014 key, _, value = kv.partition("=")
1015 scope_context[key.strip()] = value.strip()
1016 scope = resolve_scope(fsm.scope or ["."], scope_context)
1017 conflict = lock_manager.find_conflict(scope)
1018 if conflict and not getattr(args, "queue", False) and not getattr(args, "no_lock", False):
1019 print(f"Scope conflict with running loop: {conflict.loop_name}", file=sys.stderr)
1020 print(f" Conflicting scope: {conflict.scope}", file=sys.stderr)
1021 print(" Use --queue to wait for it to finish", file=sys.stderr)
1022 return 1
1024 if instance_id is None:
1025 instance_id = _make_instance_id(loop_name)
1026 pid_file = running_dir / f"{instance_id}.pid"
1027 log_file = running_dir / f"{instance_id}.log"
1029 # Build re-exec command with --foreground-internal instead of --background
1030 cmd = [
1031 sys.executable,
1032 "-m",
1033 "little_loops.cli.loop",
1034 subcommand,
1035 loop_name,
1036 ]
1037 input_val = getattr(args, "input", None)
1038 if input_val is not None:
1039 cmd.append(input_val)
1040 cmd.append("--foreground-internal")
1041 cmd.extend(["--instance-id", instance_id])
1043 # Forward relevant args
1044 max_steps = getattr(args, "max_steps", None)
1045 if max_steps:
1046 cmd.extend(["--max-steps", str(max_steps)])
1047 max_iter = getattr(args, "max_iterations", None)
1048 if max_iter:
1049 cmd.extend(["--max-iterations", str(max_iter)])
1050 if getattr(args, "no_llm", False):
1051 cmd.append("--no-llm")
1052 run_model = getattr(args, "run_model", None)
1053 if run_model:
1054 cmd.extend(["--model", run_model])
1055 llm_model = getattr(args, "llm_model", None)
1056 if llm_model:
1057 cmd.extend(["--llm-model", llm_model])
1058 if getattr(args, "verbose", False):
1059 cmd.append("--verbose")
1060 show_diagrams_raw = getattr(args, "show_diagrams", None)
1061 if show_diagrams_raw is not None:
1062 if show_diagrams_raw is True:
1063 cmd.append("--show-diagrams")
1064 else:
1065 cmd.extend(["--show-diagrams", show_diagrams_raw])
1066 diagram_edge_labels = getattr(args, "diagram_edge_labels", None)
1067 if diagram_edge_labels is not None:
1068 cmd.extend(["--diagram-edge-labels", diagram_edge_labels])
1069 diagram_state_detail = getattr(args, "diagram_state_detail", None)
1070 if diagram_state_detail is not None:
1071 cmd.extend(["--diagram-state-detail", diagram_state_detail])
1072 diagram_scope = getattr(args, "diagram_scope", None)
1073 if diagram_scope is not None:
1074 cmd.extend(["--diagram-scope", diagram_scope])
1075 if getattr(args, "quiet", False):
1076 cmd.append("--quiet")
1077 if getattr(args, "queue", False):
1078 cmd.append("--queue")
1079 if getattr(args, "no_lock", False):
1080 cmd.append("--no-lock")
1081 for kv in getattr(args, "context", None) or []:
1082 cmd.extend(["--context", kv])
1083 program_md = getattr(args, "program_md", None)
1084 if program_md is not None:
1085 cmd.extend(["--program-md", str(program_md)])
1086 delay = getattr(args, "delay", None)
1087 if delay is not None:
1088 cmd.extend(["--delay", str(delay)])
1089 handoff_threshold = getattr(args, "handoff_threshold", None)
1090 if handoff_threshold is not None:
1091 cmd.extend(["--handoff-threshold", str(handoff_threshold)])
1092 context_limit = getattr(args, "context_limit", None)
1093 if context_limit is not None:
1094 cmd.extend(["--context-limit", str(context_limit)])
1095 if getattr(args, "baseline", False):
1096 cmd.append("--baseline")
1097 baseline_skill = getattr(args, "baseline_skill", None)
1098 if baseline_skill is not None:
1099 cmd.extend(["--baseline-skill", baseline_skill])
1100 items = getattr(args, "items", None)
1101 if items is not None:
1102 cmd.extend(["--items", str(items)])
1103 if getattr(args, "cross_host", False):
1104 cmd.append("--cross-host")
1106 log_file.parent.mkdir(parents=True, exist_ok=True)
1107 with open(log_file, "w") as log_fh:
1108 process = subprocess.Popen(
1109 cmd,
1110 start_new_session=True,
1111 stdout=log_fh,
1112 stderr=log_fh,
1113 stdin=subprocess.DEVNULL,
1114 )
1116 pid_file.write_text(str(process.pid))
1117 print(
1118 f"Loop {colorize(loop_name, '1')} started in background (PID: {colorize(str(process.pid), '2')})"
1119 )
1120 print(f" Log: {colorize(str(log_file), '2')}")
1121 print(f" Status: {colorize(f'll-loop status {loop_name}', '2')}")
1122 print(f" Stop: {colorize(f'll-loop stop {loop_name}', '2')}")
1123 return 0
1126def run_foreground(
1127 executor: Any,
1128 fsm: FSMLoop,
1129 args: argparse.Namespace,
1130 highlight_color: str = "32",
1131 edge_label_colors: dict[str, str] | None = None,
1132 badges: dict[str, str] | None = None,
1133 mode: str = "run",
1134 instance_id: str | None = None,
1135 running_dir: Path | None = None,
1136 loop_path: Path | None = None,
1137 model: str | None = None,
1138) -> int:
1139 """Run loop with progress display.
1141 Args:
1142 highlight_color: ANSI SGR code for the active FSM state highlight in verbose mode.
1143 edge_label_colors: Optional label→SGR-code mapping for transition edge labels.
1144 badges: Optional glyph-key→string mapping for state type badges in FSM diagrams.
1145 mode: ``"run"`` (default) calls ``executor.run()``; ``"resume"`` calls
1146 ``executor.resume()`` so a resumed loop reuses the same display-wiring
1147 path as a fresh run (BUG-1645). In ``"resume"`` mode a ``None`` result
1148 from ``executor.resume()`` is treated as "nothing to resume": a warning
1149 is logged and exit code 1 is returned before any alt-screen sequences
1150 are emitted.
1152 instance_id: When provided and not a background-spawned child
1153 (``foreground_internal=False``), stdout/stderr are teed to
1154 ``{running_dir}/{instance_id}.log`` with ANSI sequences stripped.
1155 running_dir: Directory for the log file. Defaults to ``.running``
1156 alongside the loops directory.
1157 Returns:
1158 Exit code (0 = success).
1159 """
1160 if mode not in ("run", "resume"):
1161 raise ValueError(f"run_foreground: invalid mode {mode!r}; expected 'run' or 'resume'")
1163 _orig_stdout = sys.stdout
1164 _orig_stderr = sys.stderr
1165 _log_fh: Any = None
1166 if instance_id is not None and not getattr(args, "foreground_internal", False):
1167 _log_path = (running_dir or Path(".running")) / f"{instance_id}.log"
1168 _log_path.parent.mkdir(parents=True, exist_ok=True)
1169 _log_fh = open(_log_path, "w")
1170 sys.stdout = _TeeWriter(_orig_stdout, _log_fh) # type: ignore[assignment]
1171 sys.stderr = _TeeWriter(_orig_stderr, _log_fh) # type: ignore[assignment]
1173 try:
1174 # Create the state feed renderer — encapsulates display state and event handling.
1175 renderer = StateFeedRenderer(
1176 fsm,
1177 args,
1178 highlight_color=highlight_color,
1179 edge_label_colors=edge_label_colors,
1180 badges=badges,
1181 loops_dir=getattr(executor, "loops_dir", Path(".")),
1182 loop_path=loop_path,
1183 model=model,
1184 )
1185 if not renderer.quiet:
1186 print(f"Running loop: {colorize(fsm.name, '1')}")
1187 print(f"Max steps: {colorize(str(fsm.max_steps), '2')}")
1188 if fsm.max_iterations is not None:
1189 print(f"Max iterations: {colorize(str(fsm.max_iterations), '2')}")
1190 for key, value in _artifact_lines(fsm, loop_path):
1191 print(f" {key}: {colorize(value, '2')}")
1192 if model is not None:
1193 print(f" model: {colorize(model, '2')}")
1194 print()
1196 # Wire progress display via the EventBus on PersistentExecutor
1197 if not renderer.quiet or renderer.show_diagrams:
1198 if hasattr(executor, "event_bus"):
1199 executor.event_bus.register(renderer.handle_event)
1200 else:
1201 executor._on_event = renderer.handle_event
1203 # Capture the last failure-relevant message from the event stream so the
1204 # completion summary can surface *why* a run failed. This is the only reliable
1205 # source: on on_error / exception routes the executor leaves prev_result and
1206 # captured empty, and alt-screen teardown or non-verbose mode hide the live
1207 # output. action_error carries interpolation/exception reasons; a non-zero
1208 # action_complete carries the failing state's stdout.
1209 _failure_capture: dict[str, str] = {}
1211 def _capture_failure(event: dict[str, Any]) -> None:
1212 ev = event.get("event")
1213 if ev == "action_error" and event.get("error"):
1214 _failure_capture["error"] = str(event["error"])
1215 elif ev == "action_complete" and event.get("exit_code") not in (0, None):
1216 out = event.get("output") or event.get("output_preview") or ""
1217 if out:
1218 _failure_capture["output"] = str(out)
1220 if not renderer.quiet:
1221 if hasattr(executor, "event_bus"):
1222 executor.event_bus.register(_capture_failure)
1223 else:
1224 _prev_capture_cb = executor._on_event
1226 def _chained_capture(event: dict[str, Any]) -> None:
1227 if _prev_capture_cb:
1228 _prev_capture_cb(event)
1229 _capture_failure(event)
1231 executor._on_event = _chained_capture
1233 # Wire follow mode — streams history-formatted events independently of quiet
1234 if getattr(args, "follow", False):
1235 from little_loops.cli.loop.info import _format_history_event
1237 tw = terminal_width()
1238 _verbose = renderer.verbose
1240 def _follow_callback(event: dict[str, Any]) -> None:
1241 line = _format_history_event(event, verbose=_verbose, width=tw)
1242 if line is not None:
1243 print(line, flush=True)
1245 if hasattr(executor, "event_bus"):
1246 executor.event_bus.register(_follow_callback)
1247 else:
1248 _prev_on_event = executor._on_event
1250 def _chained(event: dict[str, Any]) -> None:
1251 if _prev_on_event:
1252 _prev_on_event(event)
1253 _follow_callback(event)
1255 executor._on_event = _chained
1257 # Enter alternate screen buffer when showing diagrams with clear to prevent
1258 # scrollback contamination from diagrams taller than the terminal height.
1259 global _using_alt_screen
1260 if renderer.show_diagrams and renderer.clear_screen and sys.stdout.isatty():
1261 _using_alt_screen = True
1262 print("\033[?1049h\033[H", end="", flush=True)
1263 _install_sigwinch_handler()
1265 try:
1266 if mode == "resume":
1267 result = executor.resume()
1268 # "Nothing to resume" path: no run actually executed, so don't fall
1269 # through to completion-line formatting. Exit cleanly with code 1.
1270 if result is None:
1271 Logger().warning(f"Nothing to resume for: {fsm.name}")
1272 return 1
1273 else:
1274 result = executor.run()
1275 finally:
1276 # Remember whether we were in the alt-screen so the summary block can
1277 # decide whether the failing state's output still needs re-printing
1278 # (alt-screen teardown wipes it from scrollback).
1279 _was_alt_screen = _using_alt_screen
1280 if _using_alt_screen:
1281 # Reset DECSTBM scroll region BEFORE exiting alt-screen, otherwise
1282 # the main buffer is left with a restricted scroll region (one of
1283 # the Success Metrics for ENH-1642).
1284 print("\033[r", end="", flush=True)
1285 print("\033[?1049l", end="", flush=True)
1286 _using_alt_screen = False
1287 _restore_sigwinch_handler()
1289 if not renderer.quiet:
1290 print()
1291 duration_sec = result.duration_ms / 1000
1292 if duration_sec < 60:
1293 duration_str = f"{duration_sec:.1f}s"
1294 else:
1295 minutes = int(duration_sec // 60)
1296 seconds = duration_sec % 60
1297 duration_str = f"{minutes}m {seconds:.0f}s"
1298 # A terminal state whose name is not "done" represents failure (the
1299 # established convention; see sub-FSM routing in executor.py). Colour
1300 # only genuine success green so a `failed` terminal doesn't read as a pass.
1301 _is_success = result.terminated_by in ("terminal", "signal", "handoff") and not (
1302 result.terminated_by == "terminal" and result.final_state != "done"
1303 )
1304 if _is_success:
1305 state_colored = colorize(result.final_state, "32")
1306 else:
1307 state_colored = colorize(result.final_state, "38;5;208")
1309 # Surface the failing state's output as the failure reason. It is otherwise
1310 # invisible: the alt-screen wipes it on teardown, and in non-verbose mode the
1311 # per-state stdout is never printed inline at all. Skip only the verbose
1312 # non-alt-screen case, where the live renderer already echoed it.
1313 if not _is_success and (_was_alt_screen or not renderer.verbose):
1314 reason_text = (
1315 _failure_capture.get("error")
1316 or _failure_capture.get("output")
1317 or result.error
1318 or ""
1319 ).strip()
1320 if reason_text:
1321 print()
1322 print(colorize("Failure reason:", "1"))
1323 for _line in reason_text.splitlines()[-40:]: # cap to bound scrollback
1324 print(colorize("│ " + _line, "2"))
1326 completion_prefix = "Resumed and completed" if mode == "resume" else "Loop completed"
1327 rejection_count = 0
1328 for _t in getattr(getattr(executor, "event_bus", None), "_transports", []):
1329 if hasattr(_t, "get_stats"):
1330 rejection_count += _t.get_stats().get("client_rejections", 0)
1331 suffix = f", {rejection_count} client rejections" if rejection_count > 0 else ""
1333 # Print per-state token/cost table if usage data was collected
1334 run_dir = fsm.context.get("run_dir", "")
1335 if run_dir:
1336 try:
1337 _print_usage_summary(Path(run_dir) / "usage.jsonl")
1338 except Exception:
1339 pass # Non-fatal: display failure shouldn't block exit
1341 print(
1342 f"{completion_prefix}: {state_colored} ({result.iterations} iterations, {duration_str}{suffix})"
1343 )
1345 # FEAT-1822: Print A/B summary if baseline was enabled
1346 if run_dir:
1347 ab_path = Path(run_dir) / "ab.json"
1348 if ab_path.exists():
1349 try:
1350 _print_ab_summary(ab_path)
1351 except Exception:
1352 pass # Non-fatal: display failure shouldn't block exit
1354 # ENH-2086: Cross-host validation when --cross-host was requested
1355 baseline_ctx = fsm.context.get("_baseline") or {}
1356 if baseline_ctx.get("cross_host"):
1357 loop_name = getattr(args, "loop", "")
1358 try:
1359 _run_cross_host_validation(
1360 args,
1361 loop_path,
1362 Path(run_dir),
1363 ab_path,
1364 loop_name,
1365 )
1366 except Exception:
1367 pass # Non-fatal
1369 return EXIT_CODES.get(result.terminated_by, 1)
1370 finally:
1371 sys.stdout = _orig_stdout
1372 sys.stderr = _orig_stderr
1373 if _log_fh is not None:
1374 _log_fh.close()
1377def _print_usage_summary(usage_path: Path) -> None:
1378 """Print per-state token usage summary from usage.jsonl.
1380 Args:
1381 usage_path: Path to usage.jsonl written by PersistentExecutor
1382 """
1383 from collections import defaultdict
1385 if not usage_path.exists():
1386 return
1387 lines = usage_path.read_text(encoding="utf-8").splitlines()
1388 if not lines:
1389 return
1391 per_state: dict[str, dict[str, Any]] = defaultdict(
1392 lambda: {
1393 "invocations": 0,
1394 "input": 0,
1395 "output": 0,
1396 "cache_read": 0,
1397 "cache_creation": 0,
1398 "model": "unknown",
1399 "est_cost": 0.0,
1400 "has_unknown_model": False,
1401 }
1402 )
1403 for raw in lines:
1404 try:
1405 row = json.loads(raw)
1406 except json.JSONDecodeError:
1407 continue
1408 state = row.get("state", "unknown")
1409 model = row.get("model", "unknown")
1410 inp = row.get("input_tokens", 0)
1411 out = row.get("output_tokens", 0)
1412 cr = row.get("cache_read_tokens", 0)
1413 cc = row.get("cache_creation_tokens", 0)
1414 bucket = per_state[state]
1415 bucket["invocations"] += 1
1416 bucket["input"] += inp
1417 bucket["output"] += out
1418 bucket["cache_read"] += cr
1419 bucket["cache_creation"] += cc
1420 bucket["model"] = model
1421 cost = estimate_cost_usd(model, inp, out, cr, cc)
1422 if cost is None:
1423 bucket["has_unknown_model"] = True
1424 else:
1425 bucket["est_cost"] += cost
1427 if not per_state:
1428 return
1430 print()
1431 print(f"{'state':<24} {'invoc':>5} {'input':>8} {'output':>8} {'cache':>8} {'est_cost':>10}")
1432 print("-" * 68)
1433 for state, b in sorted(per_state.items()):
1434 cache = b["cache_read"] + b["cache_creation"]
1435 cost_str = f"${b['est_cost']:.4f}" if not b["has_unknown_model"] else "n/a"
1436 print(
1437 f"{state:<24} {b['invocations']:>5} {b['input']:>8} {b['output']:>8} {cache:>8} {cost_str:>10}"
1438 )
1439 print()
1442def _print_ab_summary(ab_path: Path) -> None:
1443 """Print A/B comparison summary from ab.json.
1445 Args:
1446 ab_path: Path to ab.json file written by the executor
1447 """
1448 from little_loops.ab_writer import read_ab_json
1449 from little_loops.stats import wilson_ci
1451 results = read_ab_json(str(ab_path.parent))
1452 if results is None or not results.per_item:
1453 return
1455 n = len(results.per_item)
1456 harness_pct = results.harness_pass_rate * 100
1457 baseline_pct = results.baseline_pass_rate * 100
1458 delta_pct = results.delta * 100
1460 k_harness = sum(1 for item in results.per_item if item.get("harness_pass", False))
1461 k_baseline = sum(1 for item in results.per_item if item.get("baseline_pass", False))
1462 h_lo, h_hi = wilson_ci(k_harness, n)
1463 b_lo, b_hi = wilson_ci(k_baseline, n)
1465 tokens_ratio = (
1466 results.median_tokens_harness / results.median_tokens_baseline
1467 if results.median_tokens_baseline > 0
1468 else 0
1469 )
1470 dur_ratio = (
1471 results.median_duration_harness / results.median_duration_baseline
1472 if results.median_duration_baseline > 0
1473 else 0
1474 )
1476 def _fmt_dur(ms: float) -> str:
1477 if ms < 1000:
1478 return f"{ms:.0f}ms"
1479 elif ms < 60000:
1480 return f"{ms / 1000:.1f}s"
1481 else:
1482 return f"{ms / 60000:.1f}m"
1484 tokens_dir = "+" if tokens_ratio > 1 else "-"
1485 dur_dir = "+" if dur_ratio > 1 else "-"
1487 print()
1488 print(f"A/B Summary (n={n})")
1489 print(f" Harness pass-rate: {harness_pct:.0f}% [{h_lo:.2f}, {h_hi:.2f}]")
1490 print(f" Baseline pass-rate: {baseline_pct:.0f}% [{b_lo:.2f}, {b_hi:.2f}]")
1491 print(f" Delta: {delta_pct:+.0f}%")
1492 print()
1493 print(
1494 f" Median tokens: harness={results.median_tokens_harness} "
1495 f"baseline={results.median_tokens_baseline} "
1496 f"({tokens_dir}{abs(tokens_ratio - 1) * 100:.0f}%)"
1497 )
1498 print(
1499 f" Median duration: harness={_fmt_dur(results.median_duration_harness)} "
1500 f"baseline={_fmt_dur(results.median_duration_baseline)} "
1501 f"({dur_dir}{abs(dur_ratio - 1) * 100:.0f}%)"
1502 )
1504 # Verdict line
1505 quality_verdict = (
1506 "harness wins on quality"
1507 if results.delta > 0
1508 else "baseline wins on quality"
1509 if results.delta < 0
1510 else "no quality difference"
1511 )
1512 cost_verdict = (
1513 f"costs ~{abs(tokens_ratio - 1) * 100:.0f}% more tokens"
1514 if tokens_ratio > 1
1515 else (
1516 f"costs ~{abs(tokens_ratio - 1) * 100:.0f}% fewer tokens"
1517 if tokens_ratio < 1
1518 else "same token cost"
1519 )
1520 )
1521 print(f" Verdict: {quality_verdict}, {cost_verdict}")
1522 print()
1523 print(f"Per-item: {ab_path}")
1526def _run_cross_host_validation(
1527 args: argparse.Namespace,
1528 loop_path: Path | None,
1529 primary_run_dir: Path,
1530 primary_ab_path: Path,
1531 loop_name: str,
1532) -> None:
1533 """Re-run the loop on a second available host and print a cross-host comparison.
1535 Identifies a second host via _PROBE_ORDER, runs the same baseline trial with
1536 LL_HOST_CLI overridden, then calls _print_cross_host_table if both ab.json
1537 files are available.
1538 """
1539 import os
1540 import shutil
1542 from little_loops.host_runner import _PROBE_ORDER, HostNotConfigured, resolve_host
1544 # Identify the current (primary) host
1545 try:
1546 primary_host = resolve_host().name
1547 except HostNotConfigured:
1548 primary_host = None
1550 # Find a second available host different from the primary
1551 second_host: str | None = None
1552 for host_name, binary in _PROBE_ORDER:
1553 if host_name == primary_host:
1554 continue
1555 if shutil.which(binary) is not None:
1556 second_host = host_name
1557 break
1559 if second_host is None:
1560 print("\nCross-host: only one host available — skipping cross-host validation.")
1561 return
1563 print(f"\nCross-host: running {loop_name!r} on {second_host}...")
1565 # Build the second-host command
1566 cmd = ["ll-loop", "run", "--baseline"]
1567 baseline_skill = getattr(args, "baseline_skill", None)
1568 if baseline_skill is not None:
1569 cmd.extend(["--baseline-skill", baseline_skill])
1570 items = getattr(args, "items", None)
1571 if items is not None:
1572 cmd.extend(["--items", str(items)])
1573 cmd.append(loop_name)
1575 env = dict(os.environ)
1576 env["LL_HOST_CLI"] = second_host
1578 before = time.time()
1579 result = subprocess.run(cmd, env=env)
1581 if result.returncode != 0:
1582 print(
1583 f"\nCross-host: second-host run ({second_host}) failed "
1584 f"(exit {result.returncode}) — no comparison available."
1585 )
1586 return
1588 # Find the second run's ab.json: newest file under the same runs directory
1589 # that appeared after the second run started and differs from the primary.
1590 runs_dir = primary_run_dir.parent
1591 candidates = sorted(
1592 runs_dir.glob("*/ab.json"),
1593 key=lambda p: p.stat().st_mtime,
1594 reverse=True,
1595 )
1596 second_ab_path = next(
1597 (p for p in candidates if p != primary_ab_path and p.stat().st_mtime >= before),
1598 None,
1599 )
1601 if second_ab_path is None:
1602 print(
1603 "\nCross-host: second-host run completed but no ab.json found — "
1604 "no comparison available."
1605 )
1606 return
1608 from little_loops.ab_writer import read_ab_json
1610 primary_results = read_ab_json(str(primary_run_dir))
1611 second_results = read_ab_json(str(second_ab_path.parent))
1613 if primary_results is None or second_results is None:
1614 return
1616 _print_cross_host_table(primary_host or "primary", primary_results, second_host, second_results)
1619def _print_cross_host_table(
1620 host1: str,
1621 results1: Any,
1622 host2: str,
1623 results2: Any,
1624) -> None:
1625 """Print a cross-host pass-rate comparison table with Wilson 95% CIs."""
1626 from little_loops.stats import wilson_ci
1628 def _host_stats(results: Any) -> tuple[int, int, float, float, float]:
1629 n = len(results.per_item)
1630 k = sum(1 for item in results.per_item if item.get("harness_pass", False))
1631 rate = results.harness_pass_rate * 100
1632 lo, hi = wilson_ci(k, n) if n > 0 else (0.0, 0.0)
1633 return n, k, rate, lo, hi
1635 n1, _k1, rate1, lo1, hi1 = _host_stats(results1)
1636 n2, _k2, rate2, lo2, hi2 = _host_stats(results2)
1638 print()
1639 print("Cross-host Comparison")
1640 print(f" {'Host':<20} {'Pass rate':>10} {'95% CI':>18} {'n':>5}")
1641 print(f" {'-' * 20} {'-' * 10} {'-' * 18} {'-' * 5}")
1642 print(f" {host1:<20} {rate1:>9.0f}% [{lo1:.2f}, {hi1:.2f}] {n1:>5}")
1643 print(f" {host2:<20} {rate2:>9.0f}% [{lo2:.2f}, {hi2:.2f}] {n2:>5}")
1645 # Warn when the harness-vs-baseline ordering reverses between hosts
1646 host1_harness_wins = results1.delta > 0
1647 host2_harness_wins = results2.delta > 0
1648 if host1_harness_wins != host2_harness_wins:
1649 winner1 = "harness" if host1_harness_wins else "baseline"
1650 winner2 = "harness" if host2_harness_wins else "baseline"
1651 print()
1652 print(
1653 f" ⚠ Ordering reversal: {host1} shows {winner1} ahead, "
1654 f"{host2} shows {winner2} ahead. "
1655 "Improvement may be host-specific."
1656 )
1657 print()