Coverage for little_loops / cli / loop / _helpers.py: 31%
867 statements
« prev ^ index » next coverage.py v7.12.0, created at 2026-06-16 13:12 -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_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
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_iterations}] "
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):
573 pass
574 else:
575 self.child_fsm_stack[depth] = None
576 else:
577 self.child_fsm_stack[depth] = None
578 # Clear stale deeper child FSM entries
579 for k in [k for k in self.child_fsm_stack if k > depth]:
580 del self.child_fsm_stack[k]
582 if self.in_pinned_mode:
583 state0 = self.last_state_at_depth.get(0)
584 if state0 is None:
585 state0 = state
586 self._redraw_pinned(state0)
587 elif self.show_diagrams:
588 from little_loops.cli.loop.layout import (
589 _collect_edges,
590 _filter_main_path_graph,
591 _render_fsm_diagram,
592 )
594 assert self.facets is not None
596 # Find deepest active loop — show only that one.
597 active_fsm_diag = self.fsm
598 active_highlight = self.last_state_at_depth.get(0)
599 active_depth_diag = 0
600 for d in sorted(self.child_fsm_stack.keys()):
601 child_at_d = self.child_fsm_stack[d]
602 if child_at_d is not None and (d + 1) in self.last_state_at_depth:
603 active_fsm_diag = child_at_d
604 active_highlight = self.last_state_at_depth.get(d + 1)
605 active_depth_diag = d + 1
607 # Fall back to full scope when the highlighted state is hidden in main scope.
608 active_scope = self.facets.scope
609 fallback_note: str | None = None
610 if active_scope == "main" and active_highlight is not None:
611 _filtered_edges, active_reachable = _filter_main_path_graph(
612 active_fsm_diag, _collect_edges(active_fsm_diag)
613 )
614 if active_highlight not in active_reachable:
615 active_scope = "full"
616 fallback_note = (
617 f"(showing full diagram: active state "
618 f"{active_highlight!r} is off the main path)"
619 )
620 diagram = _render_fsm_diagram(
621 active_fsm_diag,
622 highlight_state=active_highlight,
623 highlight_color=self.highlight_color,
624 edge_label_colors=self.edge_label_colors,
625 badges=self.badges,
626 mode=active_scope,
627 suppress_labels=not self.facets.edge_labels,
628 title_only=self.facets.state_detail == "title",
629 )
630 # Header: breadcrumb shows immediate parent when inside a sub-loop.
631 if active_depth_diag > 0:
632 imm_parent_name = (
633 self.fsm.name
634 if active_depth_diag == 1
635 else (self.child_fsm_stack.get(active_depth_diag - 2) or self.fsm).name
636 )
637 imm_parent_state = self.last_state_at_depth.get(active_depth_diag - 1, "")
638 header_text = (
639 f"== loop: {active_fsm_diag.name} ({imm_parent_name} › {imm_parent_state}) "
640 )
641 else:
642 header_text = f"== loop: {self.fsm.name} "
643 header = header_text + "=" * max(0, tw - len(header_text))
644 print(header, flush=True)
645 if fallback_note is not None:
646 print(fallback_note, flush=True)
647 for key, value in _artifact_lines(self.fsm, self.loop_path):
648 print(f" {key}: {colorize(value, '2')}", flush=True)
649 if self.model is not None:
650 print(f" model: {colorize(self.model, '2')}", flush=True)
651 print(diagram, flush=True)
652 # In pinned mode the iteration line is part of the pinned pane;
653 # only print it inline for non-pinned paths.
654 if not self.quiet and not self.in_pinned_mode:
655 print(
656 f"{indent}[{self.current_iteration[0]}/{self.fsm.max_iterations}] {colorize(state, '1')} ({colorize(elapsed_str, '2')})",
657 end="",
658 flush=True,
659 )
661 elif event_type == "action_start":
662 if not self.quiet:
663 action = event.get("action", "")
664 is_prompt = event.get("is_prompt", False)
665 if is_prompt:
666 lines = action.strip().splitlines()
667 line_count = len(lines)
668 prompt_badge = "✦" # ✦
669 if self.verbose:
670 print(
671 f"{indent} -> {colorize(prompt_badge, '2')} {colorize(f'({line_count} lines)', '2')}",
672 flush=True,
673 )
674 for line in lines:
675 print(f"{indent} {line}", flush=True)
676 else:
677 first_line = lines[0] if lines else ""
678 preview = first_line[:60] + "..." if len(first_line) > 60 else first_line
679 print(
680 f"{indent} -> {colorize(prompt_badge, '2')} {colorize(preview, '2')}",
681 flush=True,
682 )
683 else:
684 if self.verbose:
685 action_display = action
686 else:
687 action_display = (
688 action[:max_line] + "..." if len(action) > max_line else action
689 )
690 print(f"{indent} -> {colorize(action_display, '2')}", flush=True)
692 elif event_type == "action_output":
693 if not self.quiet:
694 line = event.get("line", "")
695 if line.strip():
696 print(f"{indent} {line}", flush=True)
698 elif event_type == "action_complete":
699 actual_model = event.get("model")
700 if actual_model:
701 self.model = actual_model
702 if not self.quiet:
703 duration_ms = event.get("duration_ms", 0)
704 exit_code = event.get("exit_code", 0)
705 duration_sec = duration_ms / 1000
706 if duration_sec < 60:
707 duration_str = f"{duration_sec:.1f}s"
708 else:
709 minutes = int(duration_sec // 60)
710 seconds = duration_sec % 60
711 duration_str = f"{minutes}m {seconds:.0f}s"
712 parts = [f"{indent} ({colorize(duration_str, '2')})"]
713 if exit_code == 124:
714 parts.append(colorize("timed out", "38;5;208"))
715 elif exit_code != 0:
716 parts.append(colorize(f"exit: {exit_code}", "38;5;208"))
717 print(" ".join(parts), flush=True)
719 elif event_type == "baseline_complete":
720 if not self.quiet:
721 h_ms = event.get("harness_duration_ms", 0)
722 b_ms = event.get("baseline_duration_ms", 0)
723 h_tok = event.get("harness_tokens", 0)
724 b_tok = event.get("baseline_tokens", 0)
725 print(
726 f"{indent} baseline: {colorize(f'{b_ms / 1000:.1f}s', '2')}, "
727 f"{colorize(str(b_tok), '2')} tokens | "
728 f"harness: {colorize(f'{h_ms / 1000:.1f}s', '2')}, "
729 f"{colorize(str(h_tok), '2')} tokens",
730 flush=True,
731 )
733 elif event_type == "evaluate":
734 if not self.quiet:
735 verdict = event.get("verdict", "")
736 confidence = event.get("confidence")
737 reason = event.get("reason", "")
738 error = event.get("error", "")
739 _elc = self.edge_label_colors or {}
740 if verdict in ("yes", "target", "progress"):
741 _vc = _elc.get("yes", "32")
742 symbol = colorize("✓", _vc)
743 verdict_colored = colorize(verdict, _vc)
744 elif verdict == "no":
745 _vc = _elc.get("no", "38;5;208")
746 symbol = colorize("✗", _vc)
747 verdict_colored = colorize(verdict, _vc)
748 elif verdict == "error":
749 _vc = _elc.get("error", "38;5;208")
750 symbol = colorize("✗", _vc)
751 verdict_colored = colorize(verdict, _vc)
752 else:
753 symbol = colorize("✗", "38;5;208")
754 verdict_colored = colorize(verdict, "2")
755 # Build verdict line
756 if error and verdict == "error":
757 verdict_line = f"{symbol} {verdict_colored}: {error}"
758 elif confidence is not None:
759 verdict_line = (
760 f"{symbol} {verdict_colored} {colorize(f'({confidence:.2f})', '2')}"
761 )
762 else:
763 verdict_line = f"{symbol} {verdict_colored}"
764 print(f"{indent} {verdict_line}", flush=True)
765 # Show raw_preview for error verdicts to aid diagnosis
766 raw_preview = event.get("raw_preview", "")
767 if raw_preview and verdict == "error":
768 if self.verbose:
769 sub_lines = raw_preview.splitlines() or [""]
770 first, rest = sub_lines[0], sub_lines[1:]
771 print(f"{indent} raw: {first}", flush=True)
772 for sub in rest:
773 print(f"{indent} {sub}", flush=True)
774 else:
775 print(f"{indent} raw: {raw_preview[:200]}", flush=True)
776 # Show reason on a second line if present (and not already shown as error)
777 if reason and not (error and verdict == "error"):
778 if self.verbose:
779 for sub in reason.splitlines() or [""]:
780 print(f"{indent} {sub}", flush=True)
781 else:
782 reason_display = reason[:300] + "..." if len(reason) > 300 else reason
783 print(f"{indent} {reason_display}", flush=True)
785 elif event_type == "route":
786 if not self.quiet:
787 to_state = event.get("to", "")
788 print(
789 f"{indent} {colorize('->', '2')} {colorize(to_state, '1')}",
790 flush=True,
791 )
793 elif event_type == "max_iterations_summary":
794 if not self.quiet:
795 summary_state = event.get("summary_state", "")
796 iters = event.get("iterations", 0)
797 msg = f"iteration cap reached ({iters}); running summary state '{summary_state}'"
798 print(f"{indent} {colorize(msg, '38;5;208')}", flush=True)
800 elif event_type == "stall_detected":
801 if not self.quiet:
802 state = event.get("state", "")
803 exit_code = event.get("exit_code", 0)
804 verdict = event.get("verdict", "")
805 consecutive = event.get("consecutive", 0)
806 action = event.get("action", "abort")
807 triple = f"(exit_code={exit_code}, verdict='{verdict}')"
808 msg = (
809 f"stall_detected: state '{state}' produced {triple} "
810 f"for {consecutive} consecutive iterations -> {action}"
811 )
812 print(f"{indent} {colorize(msg, '38;5;208')}", flush=True)
815def get_builtin_loops_dir() -> Path:
816 """Get the path to built-in loops bundled with the plugin."""
817 return Path(__file__).parent.parent.parent / "loops"
820def _artifact_lines(fsm: FSMLoop, loop_path: Path | None) -> list[tuple[str, str]]:
821 """Extract path-like context values from *fsm* for display in artifact headers.
823 Returns a list of ``(key, value)`` pairs where *value* is a non-empty string
824 that starts with ``.``, ``/``, or ``~``, or contains ``/``, and does not
825 contain ``${`` (unresolved template expression). When *loop_path* is not
826 ``None``, the first entry is always ``("loop", str(loop_path))``.
827 """
828 pairs: list[tuple[str, str]] = []
829 if loop_path is not None:
830 pairs.append(("loop", str(loop_path)))
831 context: dict[str, Any] = getattr(fsm, "context", None) or {}
832 for key, value in context.items():
833 if not isinstance(value, str) or not value:
834 continue
835 if "${" in value:
836 continue
837 if value.startswith(".") or value.startswith("/") or value.startswith("~") or "/" in value:
838 pairs.append((key, value))
839 return pairs
842def resolve_loop_path(name_or_path: str, loops_dir: Path) -> Path:
843 """Resolve loop name to file path."""
844 path = Path(name_or_path)
845 if path.exists():
846 return path
848 # Try <loops_dir>/<name>.fsm.yaml first (compiled FSM)
849 fsm_path = loops_dir / f"{name_or_path}.fsm.yaml"
850 if fsm_path.exists():
851 return fsm_path
853 # Fall back to <loops_dir>/<name>.yaml
854 loops_path = loops_dir / f"{name_or_path}.yaml"
855 if loops_path.exists():
856 return loops_path
858 # Fall back to built-in loops from plugin directory
859 builtin_path = get_builtin_loops_dir() / f"{name_or_path}.yaml"
860 if builtin_path.exists():
861 return builtin_path
863 raise FileNotFoundError(f"Loop not found: {name_or_path}")
866def load_loop(name_or_path: str, loops_dir: Path, logger: Logger) -> FSMLoop:
867 """Load and validate a loop.
869 Raises:
870 FileNotFoundError: If loop not found.
871 ValueError: If loop is invalid.
872 """
873 from little_loops.fsm.validation import load_and_validate
875 path = resolve_loop_path(name_or_path, loops_dir)
876 fsm, _ = load_and_validate(path)
877 return fsm
880def load_loop_with_spec(
881 name_or_path: str, loops_dir: Path, logger: Logger
882) -> tuple[FSMLoop, dict[str, Any]]:
883 """Load a loop and return both the FSMLoop and raw spec dict.
885 Used by commands that need access to raw YAML fields (e.g., description).
887 Raises:
888 FileNotFoundError: If loop not found.
889 ValueError: If loop is invalid.
890 """
891 import yaml
893 from little_loops.fsm.validation import load_and_validate
895 path = resolve_loop_path(name_or_path, loops_dir)
897 with open(path) as f:
898 spec = yaml.safe_load(f)
900 fsm, _ = load_and_validate(path)
901 return fsm, spec
904def print_execution_plan(fsm: FSMLoop, edge_label_colors: dict[str, str] | None = None) -> None:
905 """Print dry-run execution plan."""
906 _elc = edge_label_colors or {}
907 _yes_color = _elc.get("yes", "32")
908 tw = terminal_width()
909 print(colorize(f"Execution plan for: {fsm.name}", "1"))
910 print()
911 print("States:")
912 for name, state in fsm.states.items():
913 terminal_marker = colorize(" [TERMINAL]", _yes_color) if state.terminal else ""
914 print(f" {colorize(f'[{name}]', '1')}{terminal_marker}")
915 if state.action:
916 if state.action_type == "prompt":
917 lines = state.action.strip().splitlines()
918 preview = "\n ".join(lines[:3])
919 if len(lines) > 3 or len(state.action) > 200:
920 preview += " ..."
921 print(f" action: |\n {preview}")
922 else:
923 max_action = tw - 16
924 action_display = (
925 state.action[:max_action] + "..."
926 if len(state.action) > max_action
927 else state.action
928 )
929 print(f" action: {action_display}")
930 if state.evaluate:
931 print(f" evaluate: {state.evaluate.type}")
932 if state.on_yes:
933 print(f" on_yes {colorize('->', '2')} {colorize(state.on_yes, '2')}")
934 if state.on_no:
935 print(f" on_no {colorize('->', '2')} {colorize(state.on_no, '2')}")
936 if state.on_error:
937 print(f" on_error {colorize('->', '2')} {colorize(state.on_error, '2')}")
938 if state.next:
939 print(f" next {colorize('->', '2')} {colorize(state.next, '2')}")
940 if state.route:
941 print(" route:")
942 for verdict, target in state.route.routes.items():
943 print(f" {verdict} {colorize('->', '2')} {colorize(target, '2')}")
944 if state.route.default:
945 print(f" _ {colorize('->', '2')} {colorize(state.route.default, '2')}")
946 print()
947 print(f"Initial state: {fsm.initial}")
948 print(f"Max iterations: {fsm.max_iterations}")
949 if fsm.timeout:
950 print(f"Timeout: {fsm.timeout}s")
951 if fsm.context:
952 print("Context:")
953 for key, value in fsm.context.items():
954 print(f" {key}: {value!r}")
957def _make_instance_id(loop_name: str) -> str:
958 """Generate a unique instance ID for a loop run."""
959 return f"{loop_name}-{datetime.now().strftime('%Y%m%dT%H%M%S')}"
962def run_background(
963 loop_name: str,
964 args: argparse.Namespace,
965 loops_dir: Path,
966 subcommand: str = "run",
967 instance_id: str | None = None,
968) -> int:
969 """Launch loop as a detached background process.
971 Spawns a new process with start_new_session=True that re-executes
972 the loop with --foreground-internal. The parent writes the PID file
973 and returns immediately.
975 Args:
976 subcommand: The ll-loop subcommand to spawn ("run" or "resume").
977 instance_id: Pre-resolved instance ID. When provided, skips
978 _make_instance_id() allocation. Used by cmd_resume() to pass
979 an already-discovered resumable instance.
981 Returns:
982 Exit code (0 = launched successfully).
983 """
984 running_dir = loops_dir / ".running"
985 running_dir.mkdir(parents=True, exist_ok=True)
987 # Pre-flight scope conflict check — detect conflicts before spawning the child
988 # so the user gets immediate feedback instead of a silent child failure.
989 logger = Logger()
990 try:
991 fsm = load_loop(loop_name, loops_dir, logger)
992 except (FileNotFoundError, ValueError) as e:
993 print(f"Error loading loop '{loop_name}': {e}", file=sys.stderr)
994 return 1
996 lock_manager = LockManager(loops_dir)
997 # Build context for scope resolution: YAML defaults + CLI --context overrides.
998 # CLI --context is only forwarded to the child process (line ~1018); we parse
999 # it locally so resolve_scope() can use it for the pre-flight conflict check.
1000 scope_context = dict(fsm.context)
1001 for kv in getattr(args, "context", None) or []:
1002 key, _, value = kv.partition("=")
1003 scope_context[key.strip()] = value.strip()
1004 scope = resolve_scope(fsm.scope or ["."], scope_context)
1005 conflict = lock_manager.find_conflict(scope)
1006 if conflict and not getattr(args, "queue", False) and not getattr(args, "no_lock", False):
1007 print(f"Scope conflict with running loop: {conflict.loop_name}", file=sys.stderr)
1008 print(f" Conflicting scope: {conflict.scope}", file=sys.stderr)
1009 print(" Use --queue to wait for it to finish", file=sys.stderr)
1010 return 1
1012 if instance_id is None:
1013 instance_id = _make_instance_id(loop_name)
1014 pid_file = running_dir / f"{instance_id}.pid"
1015 log_file = running_dir / f"{instance_id}.log"
1017 # Build re-exec command with --foreground-internal instead of --background
1018 cmd = [
1019 sys.executable,
1020 "-m",
1021 "little_loops.cli.loop",
1022 subcommand,
1023 loop_name,
1024 ]
1025 input_val = getattr(args, "input", None)
1026 if input_val is not None:
1027 cmd.append(input_val)
1028 cmd.append("--foreground-internal")
1029 cmd.extend(["--instance-id", instance_id])
1031 # Forward relevant args
1032 max_iter = getattr(args, "max_iterations", None)
1033 if max_iter:
1034 cmd.extend(["--max-iterations", str(max_iter)])
1035 if getattr(args, "no_llm", False):
1036 cmd.append("--no-llm")
1037 llm_model = getattr(args, "llm_model", None)
1038 if llm_model:
1039 cmd.extend(["--llm-model", llm_model])
1040 if getattr(args, "verbose", False):
1041 cmd.append("--verbose")
1042 show_diagrams_raw = getattr(args, "show_diagrams", None)
1043 if show_diagrams_raw is not None:
1044 if show_diagrams_raw is True:
1045 cmd.append("--show-diagrams")
1046 else:
1047 cmd.extend(["--show-diagrams", show_diagrams_raw])
1048 diagram_edge_labels = getattr(args, "diagram_edge_labels", None)
1049 if diagram_edge_labels is not None:
1050 cmd.extend(["--diagram-edge-labels", diagram_edge_labels])
1051 diagram_state_detail = getattr(args, "diagram_state_detail", None)
1052 if diagram_state_detail is not None:
1053 cmd.extend(["--diagram-state-detail", diagram_state_detail])
1054 diagram_scope = getattr(args, "diagram_scope", None)
1055 if diagram_scope is not None:
1056 cmd.extend(["--diagram-scope", diagram_scope])
1057 if getattr(args, "quiet", False):
1058 cmd.append("--quiet")
1059 if getattr(args, "queue", False):
1060 cmd.append("--queue")
1061 if getattr(args, "no_lock", False):
1062 cmd.append("--no-lock")
1063 for kv in getattr(args, "context", None) or []:
1064 cmd.extend(["--context", kv])
1065 program_md = getattr(args, "program_md", None)
1066 if program_md is not None:
1067 cmd.extend(["--program-md", str(program_md)])
1068 delay = getattr(args, "delay", None)
1069 if delay is not None:
1070 cmd.extend(["--delay", str(delay)])
1071 handoff_threshold = getattr(args, "handoff_threshold", None)
1072 if handoff_threshold is not None:
1073 cmd.extend(["--handoff-threshold", str(handoff_threshold)])
1074 context_limit = getattr(args, "context_limit", None)
1075 if context_limit is not None:
1076 cmd.extend(["--context-limit", str(context_limit)])
1077 if getattr(args, "baseline", False):
1078 cmd.append("--baseline")
1079 baseline_skill = getattr(args, "baseline_skill", None)
1080 if baseline_skill is not None:
1081 cmd.extend(["--baseline-skill", baseline_skill])
1082 items = getattr(args, "items", None)
1083 if items is not None:
1084 cmd.extend(["--items", str(items)])
1085 if getattr(args, "cross_host", False):
1086 cmd.append("--cross-host")
1088 log_file.parent.mkdir(parents=True, exist_ok=True)
1089 with open(log_file, "w") as log_fh:
1090 process = subprocess.Popen(
1091 cmd,
1092 start_new_session=True,
1093 stdout=log_fh,
1094 stderr=log_fh,
1095 stdin=subprocess.DEVNULL,
1096 )
1098 pid_file.write_text(str(process.pid))
1099 print(
1100 f"Loop {colorize(loop_name, '1')} started in background (PID: {colorize(str(process.pid), '2')})"
1101 )
1102 print(f" Log: {colorize(str(log_file), '2')}")
1103 print(f" Status: {colorize(f'll-loop status {loop_name}', '2')}")
1104 print(f" Stop: {colorize(f'll-loop stop {loop_name}', '2')}")
1105 return 0
1108def run_foreground(
1109 executor: Any,
1110 fsm: FSMLoop,
1111 args: argparse.Namespace,
1112 highlight_color: str = "32",
1113 edge_label_colors: dict[str, str] | None = None,
1114 badges: dict[str, str] | None = None,
1115 mode: str = "run",
1116 instance_id: str | None = None,
1117 running_dir: Path | None = None,
1118 loop_path: Path | None = None,
1119 model: str | None = None,
1120) -> int:
1121 """Run loop with progress display.
1123 Args:
1124 highlight_color: ANSI SGR code for the active FSM state highlight in verbose mode.
1125 edge_label_colors: Optional label→SGR-code mapping for transition edge labels.
1126 badges: Optional glyph-key→string mapping for state type badges in FSM diagrams.
1127 mode: ``"run"`` (default) calls ``executor.run()``; ``"resume"`` calls
1128 ``executor.resume()`` so a resumed loop reuses the same display-wiring
1129 path as a fresh run (BUG-1645). In ``"resume"`` mode a ``None`` result
1130 from ``executor.resume()`` is treated as "nothing to resume": a warning
1131 is logged and exit code 1 is returned before any alt-screen sequences
1132 are emitted.
1134 instance_id: When provided and not a background-spawned child
1135 (``foreground_internal=False``), stdout/stderr are teed to
1136 ``{running_dir}/{instance_id}.log`` with ANSI sequences stripped.
1137 running_dir: Directory for the log file. Defaults to ``.running``
1138 alongside the loops directory.
1139 Returns:
1140 Exit code (0 = success).
1141 """
1142 if mode not in ("run", "resume"):
1143 raise ValueError(f"run_foreground: invalid mode {mode!r}; expected 'run' or 'resume'")
1145 _orig_stdout = sys.stdout
1146 _orig_stderr = sys.stderr
1147 _log_fh: Any = None
1148 if instance_id is not None and not getattr(args, "foreground_internal", False):
1149 _log_path = (running_dir or Path(".running")) / f"{instance_id}.log"
1150 _log_path.parent.mkdir(parents=True, exist_ok=True)
1151 _log_fh = open(_log_path, "w")
1152 sys.stdout = _TeeWriter(_orig_stdout, _log_fh) # type: ignore[assignment]
1153 sys.stderr = _TeeWriter(_orig_stderr, _log_fh) # type: ignore[assignment]
1155 try:
1156 # Create the state feed renderer — encapsulates display state and event handling.
1157 renderer = StateFeedRenderer(
1158 fsm,
1159 args,
1160 highlight_color=highlight_color,
1161 edge_label_colors=edge_label_colors,
1162 badges=badges,
1163 loops_dir=getattr(executor, "loops_dir", Path(".")),
1164 loop_path=loop_path,
1165 model=model,
1166 )
1167 if not renderer.quiet:
1168 print(f"Running loop: {colorize(fsm.name, '1')}")
1169 print(f"Max iterations: {colorize(str(fsm.max_iterations), '2')}")
1170 for key, value in _artifact_lines(fsm, loop_path):
1171 print(f" {key}: {colorize(value, '2')}")
1172 if model is not None:
1173 print(f" model: {colorize(model, '2')}")
1174 print()
1176 # Wire progress display via the EventBus on PersistentExecutor
1177 if not renderer.quiet or renderer.show_diagrams:
1178 if hasattr(executor, "event_bus"):
1179 executor.event_bus.register(renderer.handle_event)
1180 else:
1181 executor._on_event = renderer.handle_event
1183 # Capture the last failure-relevant message from the event stream so the
1184 # completion summary can surface *why* a run failed. This is the only reliable
1185 # source: on on_error / exception routes the executor leaves prev_result and
1186 # captured empty, and alt-screen teardown or non-verbose mode hide the live
1187 # output. action_error carries interpolation/exception reasons; a non-zero
1188 # action_complete carries the failing state's stdout.
1189 _failure_capture: dict[str, str] = {}
1191 def _capture_failure(event: dict[str, Any]) -> None:
1192 ev = event.get("event")
1193 if ev == "action_error" and event.get("error"):
1194 _failure_capture["error"] = str(event["error"])
1195 elif ev == "action_complete" and event.get("exit_code") not in (0, None):
1196 out = event.get("output") or event.get("output_preview") or ""
1197 if out:
1198 _failure_capture["output"] = str(out)
1200 if not renderer.quiet:
1201 if hasattr(executor, "event_bus"):
1202 executor.event_bus.register(_capture_failure)
1203 else:
1204 _prev_capture_cb = executor._on_event
1206 def _chained_capture(event: dict[str, Any]) -> None:
1207 if _prev_capture_cb:
1208 _prev_capture_cb(event)
1209 _capture_failure(event)
1211 executor._on_event = _chained_capture
1213 # Wire follow mode — streams history-formatted events independently of quiet
1214 if getattr(args, "follow", False):
1215 from little_loops.cli.loop.info import _format_history_event
1217 tw = terminal_width()
1218 _verbose = renderer.verbose
1220 def _follow_callback(event: dict[str, Any]) -> None:
1221 line = _format_history_event(event, verbose=_verbose, width=tw)
1222 if line is not None:
1223 print(line, flush=True)
1225 if hasattr(executor, "event_bus"):
1226 executor.event_bus.register(_follow_callback)
1227 else:
1228 _prev_on_event = executor._on_event
1230 def _chained(event: dict[str, Any]) -> None:
1231 if _prev_on_event:
1232 _prev_on_event(event)
1233 _follow_callback(event)
1235 executor._on_event = _chained
1237 # Enter alternate screen buffer when showing diagrams with clear to prevent
1238 # scrollback contamination from diagrams taller than the terminal height.
1239 global _using_alt_screen
1240 if renderer.show_diagrams and renderer.clear_screen and sys.stdout.isatty():
1241 _using_alt_screen = True
1242 print("\033[?1049h\033[H", end="", flush=True)
1243 _install_sigwinch_handler()
1245 try:
1246 if mode == "resume":
1247 result = executor.resume()
1248 # "Nothing to resume" path: no run actually executed, so don't fall
1249 # through to completion-line formatting. Exit cleanly with code 1.
1250 if result is None:
1251 Logger().warning(f"Nothing to resume for: {fsm.name}")
1252 return 1
1253 else:
1254 result = executor.run()
1255 finally:
1256 # Remember whether we were in the alt-screen so the summary block can
1257 # decide whether the failing state's output still needs re-printing
1258 # (alt-screen teardown wipes it from scrollback).
1259 _was_alt_screen = _using_alt_screen
1260 if _using_alt_screen:
1261 # Reset DECSTBM scroll region BEFORE exiting alt-screen, otherwise
1262 # the main buffer is left with a restricted scroll region (one of
1263 # the Success Metrics for ENH-1642).
1264 print("\033[r", end="", flush=True)
1265 print("\033[?1049l", end="", flush=True)
1266 _using_alt_screen = False
1267 _restore_sigwinch_handler()
1269 if not renderer.quiet:
1270 print()
1271 duration_sec = result.duration_ms / 1000
1272 if duration_sec < 60:
1273 duration_str = f"{duration_sec:.1f}s"
1274 else:
1275 minutes = int(duration_sec // 60)
1276 seconds = duration_sec % 60
1277 duration_str = f"{minutes}m {seconds:.0f}s"
1278 # A terminal state whose name is not "done" represents failure (the
1279 # established convention; see sub-FSM routing in executor.py). Colour
1280 # only genuine success green so a `failed` terminal doesn't read as a pass.
1281 _is_success = result.terminated_by in ("terminal", "signal", "handoff") and not (
1282 result.terminated_by == "terminal" and result.final_state != "done"
1283 )
1284 if _is_success:
1285 state_colored = colorize(result.final_state, "32")
1286 else:
1287 state_colored = colorize(result.final_state, "38;5;208")
1289 # Surface the failing state's output as the failure reason. It is otherwise
1290 # invisible: the alt-screen wipes it on teardown, and in non-verbose mode the
1291 # per-state stdout is never printed inline at all. Skip only the verbose
1292 # non-alt-screen case, where the live renderer already echoed it.
1293 if not _is_success and (_was_alt_screen or not renderer.verbose):
1294 reason_text = (
1295 _failure_capture.get("error")
1296 or _failure_capture.get("output")
1297 or result.error
1298 or ""
1299 ).strip()
1300 if reason_text:
1301 print()
1302 print(colorize("Failure reason:", "1"))
1303 for _line in reason_text.splitlines()[-40:]: # cap to bound scrollback
1304 print(colorize("│ " + _line, "2"))
1306 completion_prefix = "Resumed and completed" if mode == "resume" else "Loop completed"
1307 rejection_count = 0
1308 for _t in getattr(getattr(executor, "event_bus", None), "_transports", []):
1309 if hasattr(_t, "get_stats"):
1310 rejection_count += _t.get_stats().get("client_rejections", 0)
1311 suffix = f", {rejection_count} client rejections" if rejection_count > 0 else ""
1313 # Print per-state token/cost table if usage data was collected
1314 run_dir = fsm.context.get("run_dir", "")
1315 if run_dir:
1316 try:
1317 _print_usage_summary(Path(run_dir) / "usage.jsonl")
1318 except Exception:
1319 pass # Non-fatal: display failure shouldn't block exit
1321 print(
1322 f"{completion_prefix}: {state_colored} ({result.iterations} iterations, {duration_str}{suffix})"
1323 )
1325 # FEAT-1822: Print A/B summary if baseline was enabled
1326 if run_dir:
1327 ab_path = Path(run_dir) / "ab.json"
1328 if ab_path.exists():
1329 try:
1330 _print_ab_summary(ab_path)
1331 except Exception:
1332 pass # Non-fatal: display failure shouldn't block exit
1334 # ENH-2086: Cross-host validation when --cross-host was requested
1335 baseline_ctx = fsm.context.get("_baseline") or {}
1336 if baseline_ctx.get("cross_host"):
1337 loop_name = getattr(args, "loop", "")
1338 try:
1339 _run_cross_host_validation(
1340 args,
1341 loop_path,
1342 Path(run_dir),
1343 ab_path,
1344 loop_name,
1345 )
1346 except Exception:
1347 pass # Non-fatal
1349 return EXIT_CODES.get(result.terminated_by, 1)
1350 finally:
1351 sys.stdout = _orig_stdout
1352 sys.stderr = _orig_stderr
1353 if _log_fh is not None:
1354 _log_fh.close()
1357def _print_usage_summary(usage_path: Path) -> None:
1358 """Print per-state token usage summary from usage.jsonl.
1360 Args:
1361 usage_path: Path to usage.jsonl written by PersistentExecutor
1362 """
1363 from collections import defaultdict
1365 if not usage_path.exists():
1366 return
1367 lines = usage_path.read_text(encoding="utf-8").splitlines()
1368 if not lines:
1369 return
1371 per_state: dict[str, dict[str, Any]] = defaultdict(
1372 lambda: {
1373 "invocations": 0,
1374 "input": 0,
1375 "output": 0,
1376 "cache_read": 0,
1377 "cache_creation": 0,
1378 "model": "unknown",
1379 "est_cost": 0.0,
1380 "has_unknown_model": False,
1381 }
1382 )
1383 for raw in lines:
1384 try:
1385 row = json.loads(raw)
1386 except json.JSONDecodeError:
1387 continue
1388 state = row.get("state", "unknown")
1389 model = row.get("model", "unknown")
1390 inp = row.get("input_tokens", 0)
1391 out = row.get("output_tokens", 0)
1392 cr = row.get("cache_read_tokens", 0)
1393 cc = row.get("cache_creation_tokens", 0)
1394 bucket = per_state[state]
1395 bucket["invocations"] += 1
1396 bucket["input"] += inp
1397 bucket["output"] += out
1398 bucket["cache_read"] += cr
1399 bucket["cache_creation"] += cc
1400 bucket["model"] = model
1401 cost = estimate_cost_usd(model, inp, out, cr, cc)
1402 if cost is None:
1403 bucket["has_unknown_model"] = True
1404 else:
1405 bucket["est_cost"] += cost
1407 if not per_state:
1408 return
1410 print()
1411 print(f"{'state':<24} {'invoc':>5} {'input':>8} {'output':>8} {'cache':>8} {'est_cost':>10}")
1412 print("-" * 68)
1413 for state, b in sorted(per_state.items()):
1414 cache = b["cache_read"] + b["cache_creation"]
1415 cost_str = f"${b['est_cost']:.4f}" if not b["has_unknown_model"] else "n/a"
1416 print(
1417 f"{state:<24} {b['invocations']:>5} {b['input']:>8} {b['output']:>8} {cache:>8} {cost_str:>10}"
1418 )
1419 print()
1422def _print_ab_summary(ab_path: Path) -> None:
1423 """Print A/B comparison summary from ab.json.
1425 Args:
1426 ab_path: Path to ab.json file written by the executor
1427 """
1428 from little_loops.ab_writer import read_ab_json
1429 from little_loops.stats import wilson_ci
1431 results = read_ab_json(str(ab_path.parent))
1432 if results is None or not results.per_item:
1433 return
1435 n = len(results.per_item)
1436 harness_pct = results.harness_pass_rate * 100
1437 baseline_pct = results.baseline_pass_rate * 100
1438 delta_pct = results.delta * 100
1440 k_harness = sum(1 for item in results.per_item if item.get("harness_pass", False))
1441 k_baseline = sum(1 for item in results.per_item if item.get("baseline_pass", False))
1442 h_lo, h_hi = wilson_ci(k_harness, n)
1443 b_lo, b_hi = wilson_ci(k_baseline, n)
1445 tokens_ratio = (
1446 results.median_tokens_harness / results.median_tokens_baseline
1447 if results.median_tokens_baseline > 0
1448 else 0
1449 )
1450 dur_ratio = (
1451 results.median_duration_harness / results.median_duration_baseline
1452 if results.median_duration_baseline > 0
1453 else 0
1454 )
1456 def _fmt_dur(ms: float) -> str:
1457 if ms < 1000:
1458 return f"{ms:.0f}ms"
1459 elif ms < 60000:
1460 return f"{ms / 1000:.1f}s"
1461 else:
1462 return f"{ms / 60000:.1f}m"
1464 tokens_dir = "+" if tokens_ratio > 1 else "-"
1465 dur_dir = "+" if dur_ratio > 1 else "-"
1467 print()
1468 print(f"A/B Summary (n={n})")
1469 print(f" Harness pass-rate: {harness_pct:.0f}% [{h_lo:.2f}, {h_hi:.2f}]")
1470 print(f" Baseline pass-rate: {baseline_pct:.0f}% [{b_lo:.2f}, {b_hi:.2f}]")
1471 print(f" Delta: {delta_pct:+.0f}%")
1472 print()
1473 print(
1474 f" Median tokens: harness={results.median_tokens_harness} "
1475 f"baseline={results.median_tokens_baseline} "
1476 f"({tokens_dir}{abs(tokens_ratio - 1) * 100:.0f}%)"
1477 )
1478 print(
1479 f" Median duration: harness={_fmt_dur(results.median_duration_harness)} "
1480 f"baseline={_fmt_dur(results.median_duration_baseline)} "
1481 f"({dur_dir}{abs(dur_ratio - 1) * 100:.0f}%)"
1482 )
1484 # Verdict line
1485 quality_verdict = (
1486 "harness wins on quality"
1487 if results.delta > 0
1488 else "baseline wins on quality"
1489 if results.delta < 0
1490 else "no quality difference"
1491 )
1492 cost_verdict = (
1493 f"costs ~{abs(tokens_ratio - 1) * 100:.0f}% more tokens"
1494 if tokens_ratio > 1
1495 else (
1496 f"costs ~{abs(tokens_ratio - 1) * 100:.0f}% fewer tokens"
1497 if tokens_ratio < 1
1498 else "same token cost"
1499 )
1500 )
1501 print(f" Verdict: {quality_verdict}, {cost_verdict}")
1502 print()
1503 print(f"Per-item: {ab_path}")
1506def _run_cross_host_validation(
1507 args: argparse.Namespace,
1508 loop_path: Path | None,
1509 primary_run_dir: Path,
1510 primary_ab_path: Path,
1511 loop_name: str,
1512) -> None:
1513 """Re-run the loop on a second available host and print a cross-host comparison.
1515 Identifies a second host via _PROBE_ORDER, runs the same baseline trial with
1516 LL_HOST_CLI overridden, then calls _print_cross_host_table if both ab.json
1517 files are available.
1518 """
1519 import os
1520 import shutil
1522 from little_loops.host_runner import _PROBE_ORDER, HostNotConfigured, resolve_host
1524 # Identify the current (primary) host
1525 try:
1526 primary_host = resolve_host().name
1527 except HostNotConfigured:
1528 primary_host = None
1530 # Find a second available host different from the primary
1531 second_host: str | None = None
1532 for host_name, binary in _PROBE_ORDER:
1533 if host_name == primary_host:
1534 continue
1535 if shutil.which(binary) is not None:
1536 second_host = host_name
1537 break
1539 if second_host is None:
1540 print("\nCross-host: only one host available — skipping cross-host validation.")
1541 return
1543 print(f"\nCross-host: running {loop_name!r} on {second_host}...")
1545 # Build the second-host command
1546 cmd = ["ll-loop", "run", "--baseline"]
1547 baseline_skill = getattr(args, "baseline_skill", None)
1548 if baseline_skill is not None:
1549 cmd.extend(["--baseline-skill", baseline_skill])
1550 items = getattr(args, "items", None)
1551 if items is not None:
1552 cmd.extend(["--items", str(items)])
1553 cmd.append(loop_name)
1555 env = dict(os.environ)
1556 env["LL_HOST_CLI"] = second_host
1558 before = time.time()
1559 result = subprocess.run(cmd, env=env)
1561 if result.returncode != 0:
1562 print(
1563 f"\nCross-host: second-host run ({second_host}) failed "
1564 f"(exit {result.returncode}) — no comparison available."
1565 )
1566 return
1568 # Find the second run's ab.json: newest file under the same runs directory
1569 # that appeared after the second run started and differs from the primary.
1570 runs_dir = primary_run_dir.parent
1571 candidates = sorted(
1572 runs_dir.glob("*/ab.json"),
1573 key=lambda p: p.stat().st_mtime,
1574 reverse=True,
1575 )
1576 second_ab_path = next(
1577 (p for p in candidates if p != primary_ab_path and p.stat().st_mtime >= before),
1578 None,
1579 )
1581 if second_ab_path is None:
1582 print(
1583 "\nCross-host: second-host run completed but no ab.json found — "
1584 "no comparison available."
1585 )
1586 return
1588 from little_loops.ab_writer import read_ab_json
1590 primary_results = read_ab_json(str(primary_run_dir))
1591 second_results = read_ab_json(str(second_ab_path.parent))
1593 if primary_results is None or second_results is None:
1594 return
1596 _print_cross_host_table(primary_host or "primary", primary_results, second_host, second_results)
1599def _print_cross_host_table(
1600 host1: str,
1601 results1: Any,
1602 host2: str,
1603 results2: Any,
1604) -> None:
1605 """Print a cross-host pass-rate comparison table with Wilson 95% CIs."""
1606 from little_loops.stats import wilson_ci
1608 def _host_stats(results: Any) -> tuple[int, int, float, float, float]:
1609 n = len(results.per_item)
1610 k = sum(1 for item in results.per_item if item.get("harness_pass", False))
1611 rate = results.harness_pass_rate * 100
1612 lo, hi = wilson_ci(k, n) if n > 0 else (0.0, 0.0)
1613 return n, k, rate, lo, hi
1615 n1, _k1, rate1, lo1, hi1 = _host_stats(results1)
1616 n2, _k2, rate2, lo2, hi2 = _host_stats(results2)
1618 print()
1619 print("Cross-host Comparison")
1620 print(f" {'Host':<20} {'Pass rate':>10} {'95% CI':>18} {'n':>5}")
1621 print(f" {'-' * 20} {'-' * 10} {'-' * 18} {'-' * 5}")
1622 print(f" {host1:<20} {rate1:>9.0f}% [{lo1:.2f}, {hi1:.2f}] {n1:>5}")
1623 print(f" {host2:<20} {rate2:>9.0f}% [{lo2:.2f}, {hi2:.2f}] {n2:>5}")
1625 # Warn when the harness-vs-baseline ordering reverses between hosts
1626 host1_harness_wins = results1.delta > 0
1627 host2_harness_wins = results2.delta > 0
1628 if host1_harness_wins != host2_harness_wins:
1629 winner1 = "harness" if host1_harness_wins else "baseline"
1630 winner2 = "harness" if host2_harness_wins else "baseline"
1631 print()
1632 print(
1633 f" ⚠ Ordering reversal: {host1} shows {winner1} ahead, "
1634 f"{host2} shows {winner2} ahead. "
1635 "Improvement may be host-specific."
1636 )
1637 print()