Coverage for little_loops / cli / loop / _helpers.py: 8%
753 statements
« prev ^ index » next coverage.py v7.12.0, created at 2026-06-04 12:21 -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: skip neighborhood — layered→neighborhood
428 # looks like a broken diagram and confuses users. Go straight to single-line.
429 # Explicit neighborhood topology: neighborhood→single.
430 topo_detail = TOPOLOGY_TO_DETAIL[facets.topology]
431 if facets.source == "topology":
432 variants = [_build(topo_detail)]
433 elif topo_detail == "full":
434 variants = [_build("full"), _build("single")]
435 elif topo_detail == "neighborhood":
436 variants = [_build("neighborhood"), _build("single")]
437 else: # inline / single
438 variants = [_build("single")]
440 # compact presets (clean/slim) set state_detail="title" and already sacrifice
441 # action body lines; tolerate fewer rows below the diagram to avoid the
442 # single-line fsm: fallback on larger loops.
443 effective_min_action_rows = 3 if facets.state_detail == "title" else min_action_rows
444 pinned, pinned_height = _choose_pinned_layout(
445 rows,
446 variants,
447 min_action_rows=effective_min_action_rows,
448 )
449 print(pinned, flush=True)
451 # Guard against degenerate terminals: scroll region must have at least
452 # 1 row beneath the pinned pane. If pinned_height >= rows, skip the
453 # scroll-region setup entirely — output will append normally and the
454 # caller can rely on the alt-screen exit to clean up.
455 if pinned_height < rows:
456 # DECSTBM uses 1-indexed inclusive rows.
457 print(f"\033[{pinned_height + 1};{rows}r", end="", flush=True)
458 # Move cursor to the top of the scroll region.
459 print(f"\033[{pinned_height + 1};1H", end="", flush=True)
460 return pinned_height
463class StateFeedRenderer:
464 """Renders loop-state events as terminal output for foreground runs and monitor attach.
466 Extracted from ``run_foreground()`` so both the foreground run path and the
467 ``cmd_monitor`` attach path (FEAT-1764) can share the same rendering logic.
468 """
470 def __init__(
471 self,
472 fsm: FSMLoop,
473 args: argparse.Namespace,
474 highlight_color: str = "32",
475 edge_label_colors: dict[str, str] | None = None,
476 badges: dict[str, str] | None = None,
477 loops_dir: Path | None = None,
478 loop_path: Path | None = None,
479 model: str | None = None,
480 ) -> None:
481 self.fsm = fsm
482 self.args = args
483 self.highlight_color = highlight_color
484 self.edge_label_colors = edge_label_colors
485 self.badges = badges
486 self.loops_dir = loops_dir or Path(".")
487 self.loop_path = loop_path
488 self.model = model
490 # Derived from args
491 self.quiet: bool = getattr(args, "quiet", False)
492 self.verbose: bool = getattr(args, "verbose", False)
493 self.facets: DiagramFacets | None = resolve_facets(args)
494 self.show_diagrams: bool = self.facets is not None
495 self.clear_screen: bool = getattr(args, "clear", False)
496 self.in_pinned_mode: bool = self.show_diagrams and self.clear_screen and sys.stdout.isatty()
498 # Mutable state (was closure-captured in run_foreground)
499 self.current_iteration: list[int] = [0]
500 self.last_state_at_depth: dict[int, str] = {}
501 self.prev_state_at_depth: dict[int, str] = {}
502 self.child_fsm_stack: dict[int, FSMLoop | None] = {}
503 self.pinned_height: list[int] = [0]
504 self.loop_start_time: float = time.monotonic()
506 def _elapsed_str(self) -> str:
507 elapsed_int = int(time.monotonic() - self.loop_start_time)
508 if elapsed_int < 60:
509 return f"{elapsed_int}s"
510 return f"{elapsed_int // 60}m {elapsed_int % 60}s"
512 def _redraw_pinned(self, state0: str) -> None:
513 """Redraw the pinned pane in place using the current depth-0 state."""
514 assert self.facets is not None
515 iter_line = (
516 f"[{self.current_iteration[0]}/{self.fsm.max_iterations}] "
517 f"{colorize(state0, '1')} ({colorize(self._elapsed_str(), '2')})"
518 )
519 self.pinned_height[0] = _render_pinned_pane(
520 self.fsm,
521 state0,
522 self.child_fsm_stack,
523 self.last_state_at_depth,
524 iter_line,
525 facets=self.facets,
526 highlight_color=self.highlight_color,
527 edge_label_colors=self.edge_label_colors,
528 badges=self.badges,
529 prev_state_at_depth=self.prev_state_at_depth,
530 loop_path=self.loop_path,
531 model=self.model,
532 )
534 def handle_event(self, event: dict) -> None:
535 """Display progress for events."""
536 global _needs_redraw
537 # SIGWINCH redraw: terminal was resized; re-render the pinned pane
538 # before processing the next event so the layout matches the new size.
539 if _needs_redraw and self.in_pinned_mode and 0 in self.last_state_at_depth:
540 self._redraw_pinned(self.last_state_at_depth[0])
541 _needs_redraw = False
543 event_type = event.get("event")
544 depth = event.get("depth", 0)
545 indent = " " * depth
546 tw = terminal_width()
547 max_line = tw - 8 - len(indent)
549 if event_type == "state_enter":
550 self.current_iteration[0] = event.get("iteration", 0)
551 state = event.get("state", "")
552 elapsed_str = self._elapsed_str() if not self.quiet else ""
553 # Non-pinned --clear path keeps the bare full-screen clear.
554 if self.clear_screen and sys.stdout.isatty() and depth == 0 and not self.in_pinned_mode:
555 print("\033[2J\033[H", end="", flush=True)
556 # Update last-known state at this depth and clear stale deeper entries.
557 old_state = self.last_state_at_depth.get(depth)
558 if old_state is not None and old_state != state:
559 self.prev_state_at_depth[depth] = old_state
560 self.last_state_at_depth[depth] = state
561 for k in [k for k in self.last_state_at_depth if k > depth]:
562 del self.last_state_at_depth[k]
563 self.prev_state_at_depth.pop(k, None)
564 # Load child FSM for the current state at this depth
565 parent_at_depth = self.fsm if depth == 0 else self.child_fsm_stack.get(depth - 1)
566 if parent_at_depth is not None and state in parent_at_depth.states:
567 fsm_state = parent_at_depth.states[state]
568 if fsm_state.loop is not None:
569 try:
570 self.child_fsm_stack[depth] = load_loop(
571 fsm_state.loop, self.loops_dir, Logger()
572 )
573 except (FileNotFoundError, ValueError):
574 pass
575 else:
576 self.child_fsm_stack[depth] = None
577 else:
578 self.child_fsm_stack[depth] = None
579 # Clear stale deeper child FSM entries
580 for k in [k for k in self.child_fsm_stack if k > depth]:
581 del self.child_fsm_stack[k]
583 if self.in_pinned_mode:
584 state0 = self.last_state_at_depth.get(0)
585 if state0 is None:
586 state0 = state
587 self._redraw_pinned(state0)
588 elif self.show_diagrams:
589 from little_loops.cli.loop.layout import (
590 _collect_edges,
591 _filter_main_path_graph,
592 _render_fsm_diagram,
593 )
595 assert self.facets is not None
597 # Find deepest active loop — show only that one.
598 active_fsm_diag = self.fsm
599 active_highlight = self.last_state_at_depth.get(0)
600 active_depth_diag = 0
601 for d in sorted(self.child_fsm_stack.keys()):
602 child_at_d = self.child_fsm_stack[d]
603 if child_at_d is not None and (d + 1) in self.last_state_at_depth:
604 active_fsm_diag = child_at_d
605 active_highlight = self.last_state_at_depth.get(d + 1)
606 active_depth_diag = d + 1
608 # Fall back to full scope when the highlighted state is hidden in main scope.
609 active_scope = self.facets.scope
610 fallback_note: str | None = None
611 if active_scope == "main" and active_highlight is not None:
612 _filtered_edges, active_reachable = _filter_main_path_graph(
613 active_fsm_diag, _collect_edges(active_fsm_diag)
614 )
615 if active_highlight not in active_reachable:
616 active_scope = "full"
617 fallback_note = (
618 f"(showing full diagram: active state "
619 f"{active_highlight!r} is off the main path)"
620 )
621 diagram = _render_fsm_diagram(
622 active_fsm_diag,
623 highlight_state=active_highlight,
624 highlight_color=self.highlight_color,
625 edge_label_colors=self.edge_label_colors,
626 badges=self.badges,
627 mode=active_scope,
628 suppress_labels=not self.facets.edge_labels,
629 title_only=self.facets.state_detail == "title",
630 )
631 # Header: breadcrumb shows immediate parent when inside a sub-loop.
632 if active_depth_diag > 0:
633 imm_parent_name = (
634 self.fsm.name
635 if active_depth_diag == 1
636 else (self.child_fsm_stack.get(active_depth_diag - 2) or self.fsm).name
637 )
638 imm_parent_state = self.last_state_at_depth.get(active_depth_diag - 1, "")
639 header_text = (
640 f"== loop: {active_fsm_diag.name} ({imm_parent_name} › {imm_parent_state}) "
641 )
642 else:
643 header_text = f"== loop: {self.fsm.name} "
644 header = header_text + "=" * max(0, tw - len(header_text))
645 print(header, flush=True)
646 if fallback_note is not None:
647 print(fallback_note, flush=True)
648 for key, value in _artifact_lines(self.fsm, self.loop_path):
649 print(f" {key}: {colorize(value, '2')}", flush=True)
650 if self.model is not None:
651 print(f" model: {colorize(self.model, '2')}", flush=True)
652 print(diagram, flush=True)
653 # In pinned mode the iteration line is part of the pinned pane;
654 # only print it inline for non-pinned paths.
655 if not self.quiet and not self.in_pinned_mode:
656 print(
657 f"{indent}[{self.current_iteration[0]}/{self.fsm.max_iterations}] {colorize(state, '1')} ({colorize(elapsed_str, '2')})",
658 end="",
659 flush=True,
660 )
662 elif event_type == "action_start":
663 if not self.quiet:
664 action = event.get("action", "")
665 is_prompt = event.get("is_prompt", False)
666 if is_prompt:
667 lines = action.strip().splitlines()
668 line_count = len(lines)
669 prompt_badge = "✦" # ✦
670 if self.verbose:
671 print(
672 f"{indent} -> {colorize(prompt_badge, '2')} {colorize(f'({line_count} lines)', '2')}",
673 flush=True,
674 )
675 for line in lines:
676 print(f"{indent} {line}", flush=True)
677 else:
678 first_line = lines[0] if lines else ""
679 preview = first_line[:60] + "..." if len(first_line) > 60 else first_line
680 print(
681 f"{indent} -> {colorize(prompt_badge, '2')} {colorize(preview, '2')}",
682 flush=True,
683 )
684 else:
685 if self.verbose:
686 action_display = action
687 else:
688 action_display = (
689 action[:max_line] + "..." if len(action) > max_line else action
690 )
691 print(f"{indent} -> {colorize(action_display, '2')}", flush=True)
693 elif event_type == "action_output":
694 if not self.quiet:
695 line = event.get("line", "")
696 if line.strip():
697 print(f"{indent} {line}", flush=True)
699 elif event_type == "action_complete":
700 if not self.quiet:
701 duration_ms = event.get("duration_ms", 0)
702 exit_code = event.get("exit_code", 0)
703 duration_sec = duration_ms / 1000
704 if duration_sec < 60:
705 duration_str = f"{duration_sec:.1f}s"
706 else:
707 minutes = int(duration_sec // 60)
708 seconds = duration_sec % 60
709 duration_str = f"{minutes}m {seconds:.0f}s"
710 parts = [f"{indent} ({colorize(duration_str, '2')})"]
711 if exit_code == 124:
712 parts.append(colorize("timed out", "38;5;208"))
713 elif exit_code != 0:
714 parts.append(colorize(f"exit: {exit_code}", "38;5;208"))
715 print(" ".join(parts), flush=True)
717 elif event_type == "baseline_complete":
718 if not self.quiet:
719 h_ms = event.get("harness_duration_ms", 0)
720 b_ms = event.get("baseline_duration_ms", 0)
721 h_tok = event.get("harness_tokens", 0)
722 b_tok = event.get("baseline_tokens", 0)
723 print(
724 f"{indent} baseline: {colorize(f'{b_ms / 1000:.1f}s', '2')}, "
725 f"{colorize(str(b_tok), '2')} tokens | "
726 f"harness: {colorize(f'{h_ms / 1000:.1f}s', '2')}, "
727 f"{colorize(str(h_tok), '2')} tokens",
728 flush=True,
729 )
731 elif event_type == "evaluate":
732 if not self.quiet:
733 verdict = event.get("verdict", "")
734 confidence = event.get("confidence")
735 reason = event.get("reason", "")
736 error = event.get("error", "")
737 _elc = self.edge_label_colors or {}
738 if verdict in ("yes", "target", "progress"):
739 _vc = _elc.get("yes", "32")
740 symbol = colorize("✓", _vc)
741 verdict_colored = colorize(verdict, _vc)
742 elif verdict == "no":
743 _vc = _elc.get("no", "38;5;208")
744 symbol = colorize("✗", _vc)
745 verdict_colored = colorize(verdict, _vc)
746 elif verdict == "error":
747 _vc = _elc.get("error", "38;5;208")
748 symbol = colorize("✗", _vc)
749 verdict_colored = colorize(verdict, _vc)
750 else:
751 symbol = colorize("✗", "38;5;208")
752 verdict_colored = colorize(verdict, "2")
753 # Build verdict line
754 if error and verdict == "error":
755 verdict_line = f"{symbol} {verdict_colored}: {error}"
756 elif confidence is not None:
757 verdict_line = (
758 f"{symbol} {verdict_colored} {colorize(f'({confidence:.2f})', '2')}"
759 )
760 else:
761 verdict_line = f"{symbol} {verdict_colored}"
762 print(f"{indent} {verdict_line}", flush=True)
763 # Show raw_preview for error verdicts to aid diagnosis
764 raw_preview = event.get("raw_preview", "")
765 if raw_preview and verdict == "error":
766 if self.verbose:
767 sub_lines = raw_preview.splitlines() or [""]
768 first, rest = sub_lines[0], sub_lines[1:]
769 print(f"{indent} raw: {first}", flush=True)
770 for sub in rest:
771 print(f"{indent} {sub}", flush=True)
772 else:
773 print(f"{indent} raw: {raw_preview[:200]}", flush=True)
774 # Show reason on a second line if present (and not already shown as error)
775 if reason and not (error and verdict == "error"):
776 if self.verbose:
777 for sub in reason.splitlines() or [""]:
778 print(f"{indent} {sub}", flush=True)
779 else:
780 reason_display = reason[:300] + "..." if len(reason) > 300 else reason
781 print(f"{indent} {reason_display}", flush=True)
783 elif event_type == "route":
784 if not self.quiet:
785 to_state = event.get("to", "")
786 print(
787 f"{indent} {colorize('->', '2')} {colorize(to_state, '1')}",
788 flush=True,
789 )
791 elif event_type == "max_iterations_summary":
792 if not self.quiet:
793 summary_state = event.get("summary_state", "")
794 iters = event.get("iterations", 0)
795 msg = f"iteration cap reached ({iters}); running summary state '{summary_state}'"
796 print(f"{indent} {colorize(msg, '38;5;208')}", flush=True)
798 elif event_type == "stall_detected":
799 if not self.quiet:
800 state = event.get("state", "")
801 exit_code = event.get("exit_code", 0)
802 verdict = event.get("verdict", "")
803 consecutive = event.get("consecutive", 0)
804 action = event.get("action", "abort")
805 triple = f"(exit_code={exit_code}, verdict='{verdict}')"
806 msg = (
807 f"stall_detected: state '{state}' produced {triple} "
808 f"for {consecutive} consecutive iterations -> {action}"
809 )
810 print(f"{indent} {colorize(msg, '38;5;208')}", flush=True)
813def get_builtin_loops_dir() -> Path:
814 """Get the path to built-in loops bundled with the plugin."""
815 return Path(__file__).parent.parent.parent / "loops"
818def _artifact_lines(fsm: FSMLoop, loop_path: Path | None) -> list[tuple[str, str]]:
819 """Extract path-like context values from *fsm* for display in artifact headers.
821 Returns a list of ``(key, value)`` pairs where *value* is a non-empty string
822 that starts with ``.``, ``/``, or ``~``, or contains ``/``, and does not
823 contain ``${`` (unresolved template expression). When *loop_path* is not
824 ``None``, the first entry is always ``("loop", str(loop_path))``.
825 """
826 pairs: list[tuple[str, str]] = []
827 if loop_path is not None:
828 pairs.append(("loop", str(loop_path)))
829 context: dict[str, Any] = getattr(fsm, "context", None) or {}
830 for key, value in context.items():
831 if not isinstance(value, str) or not value:
832 continue
833 if "${" in value:
834 continue
835 if value.startswith(".") or value.startswith("/") or value.startswith("~") or "/" in value:
836 pairs.append((key, value))
837 return pairs
840def resolve_loop_path(name_or_path: str, loops_dir: Path) -> Path:
841 """Resolve loop name to file path."""
842 path = Path(name_or_path)
843 if path.exists():
844 return path
846 # Try <loops_dir>/<name>.fsm.yaml first (compiled FSM)
847 fsm_path = loops_dir / f"{name_or_path}.fsm.yaml"
848 if fsm_path.exists():
849 return fsm_path
851 # Fall back to <loops_dir>/<name>.yaml
852 loops_path = loops_dir / f"{name_or_path}.yaml"
853 if loops_path.exists():
854 return loops_path
856 # Fall back to built-in loops from plugin directory
857 builtin_path = get_builtin_loops_dir() / f"{name_or_path}.yaml"
858 if builtin_path.exists():
859 return builtin_path
861 raise FileNotFoundError(f"Loop not found: {name_or_path}")
864def load_loop(name_or_path: str, loops_dir: Path, logger: Logger) -> FSMLoop:
865 """Load and validate a loop.
867 Raises:
868 FileNotFoundError: If loop not found.
869 ValueError: If loop is invalid.
870 """
871 from little_loops.fsm.validation import load_and_validate
873 path = resolve_loop_path(name_or_path, loops_dir)
874 fsm, _ = load_and_validate(path)
875 return fsm
878def load_loop_with_spec(
879 name_or_path: str, loops_dir: Path, logger: Logger
880) -> tuple[FSMLoop, dict[str, Any]]:
881 """Load a loop and return both the FSMLoop and raw spec dict.
883 Used by commands that need access to raw YAML fields (e.g., description).
885 Raises:
886 FileNotFoundError: If loop not found.
887 ValueError: If loop is invalid.
888 """
889 import yaml
891 from little_loops.fsm.validation import load_and_validate
893 path = resolve_loop_path(name_or_path, loops_dir)
895 with open(path) as f:
896 spec = yaml.safe_load(f)
898 fsm, _ = load_and_validate(path)
899 return fsm, spec
902def print_execution_plan(fsm: FSMLoop, edge_label_colors: dict[str, str] | None = None) -> None:
903 """Print dry-run execution plan."""
904 _elc = edge_label_colors or {}
905 _yes_color = _elc.get("yes", "32")
906 tw = terminal_width()
907 print(colorize(f"Execution plan for: {fsm.name}", "1"))
908 print()
909 print("States:")
910 for name, state in fsm.states.items():
911 terminal_marker = colorize(" [TERMINAL]", _yes_color) if state.terminal else ""
912 print(f" {colorize(f'[{name}]', '1')}{terminal_marker}")
913 if state.action:
914 if state.action_type == "prompt":
915 lines = state.action.strip().splitlines()
916 preview = "\n ".join(lines[:3])
917 if len(lines) > 3 or len(state.action) > 200:
918 preview += " ..."
919 print(f" action: |\n {preview}")
920 else:
921 max_action = tw - 16
922 action_display = (
923 state.action[:max_action] + "..."
924 if len(state.action) > max_action
925 else state.action
926 )
927 print(f" action: {action_display}")
928 if state.evaluate:
929 print(f" evaluate: {state.evaluate.type}")
930 if state.on_yes:
931 print(f" on_yes {colorize('->', '2')} {colorize(state.on_yes, '2')}")
932 if state.on_no:
933 print(f" on_no {colorize('->', '2')} {colorize(state.on_no, '2')}")
934 if state.on_error:
935 print(f" on_error {colorize('->', '2')} {colorize(state.on_error, '2')}")
936 if state.next:
937 print(f" next {colorize('->', '2')} {colorize(state.next, '2')}")
938 if state.route:
939 print(" route:")
940 for verdict, target in state.route.routes.items():
941 print(f" {verdict} {colorize('->', '2')} {colorize(target, '2')}")
942 if state.route.default:
943 print(f" _ {colorize('->', '2')} {colorize(state.route.default, '2')}")
944 print()
945 print(f"Initial state: {fsm.initial}")
946 print(f"Max iterations: {fsm.max_iterations}")
947 if fsm.timeout:
948 print(f"Timeout: {fsm.timeout}s")
949 if fsm.context:
950 print("Context:")
951 for key, value in fsm.context.items():
952 print(f" {key}: {value!r}")
955def _make_instance_id(loop_name: str) -> str:
956 """Generate a unique instance ID for a loop run."""
957 return f"{loop_name}-{datetime.now().strftime('%Y%m%dT%H%M%S')}"
960def run_background(
961 loop_name: str,
962 args: argparse.Namespace,
963 loops_dir: Path,
964 subcommand: str = "run",
965 instance_id: str | None = None,
966) -> int:
967 """Launch loop as a detached background process.
969 Spawns a new process with start_new_session=True that re-executes
970 the loop with --foreground-internal. The parent writes the PID file
971 and returns immediately.
973 Args:
974 subcommand: The ll-loop subcommand to spawn ("run" or "resume").
975 instance_id: Pre-resolved instance ID. When provided, skips
976 _make_instance_id() allocation. Used by cmd_resume() to pass
977 an already-discovered resumable instance.
979 Returns:
980 Exit code (0 = launched successfully).
981 """
982 running_dir = loops_dir / ".running"
983 running_dir.mkdir(parents=True, exist_ok=True)
985 # Pre-flight scope conflict check — detect conflicts before spawning the child
986 # so the user gets immediate feedback instead of a silent child failure.
987 logger = Logger()
988 try:
989 fsm = load_loop(loop_name, loops_dir, logger)
990 except (FileNotFoundError, ValueError) as e:
991 print(f"Error loading loop '{loop_name}': {e}", file=sys.stderr)
992 return 1
994 lock_manager = LockManager(loops_dir)
995 # Build context for scope resolution: YAML defaults + CLI --context overrides.
996 # CLI --context is only forwarded to the child process (line ~1018); we parse
997 # it locally so resolve_scope() can use it for the pre-flight conflict check.
998 scope_context = dict(fsm.context)
999 for kv in getattr(args, "context", None) or []:
1000 key, _, value = kv.partition("=")
1001 scope_context[key.strip()] = value.strip()
1002 scope = resolve_scope(fsm.scope or ["."], scope_context)
1003 conflict = lock_manager.find_conflict(scope)
1004 if conflict and not getattr(args, "queue", False) and not getattr(args, "no_lock", False):
1005 print(f"Scope conflict with running loop: {conflict.loop_name}", file=sys.stderr)
1006 print(f" Conflicting scope: {conflict.scope}", file=sys.stderr)
1007 print(" Use --queue to wait for it to finish", file=sys.stderr)
1008 return 1
1010 if instance_id is None:
1011 instance_id = _make_instance_id(loop_name)
1012 pid_file = running_dir / f"{instance_id}.pid"
1013 log_file = running_dir / f"{instance_id}.log"
1015 # Build re-exec command with --foreground-internal instead of --background
1016 cmd = [
1017 sys.executable,
1018 "-m",
1019 "little_loops.cli.loop",
1020 subcommand,
1021 loop_name,
1022 ]
1023 input_val = getattr(args, "input", None)
1024 if input_val is not None:
1025 cmd.append(input_val)
1026 cmd.append("--foreground-internal")
1027 cmd.extend(["--instance-id", instance_id])
1029 # Forward relevant args
1030 max_iter = getattr(args, "max_iterations", None)
1031 if max_iter:
1032 cmd.extend(["--max-iterations", str(max_iter)])
1033 if getattr(args, "no_llm", False):
1034 cmd.append("--no-llm")
1035 llm_model = getattr(args, "llm_model", None)
1036 if llm_model:
1037 cmd.extend(["--llm-model", llm_model])
1038 if getattr(args, "verbose", False):
1039 cmd.append("--verbose")
1040 show_diagrams_raw = getattr(args, "show_diagrams", None)
1041 if show_diagrams_raw is not None:
1042 if show_diagrams_raw is True:
1043 cmd.append("--show-diagrams")
1044 else:
1045 cmd.extend(["--show-diagrams", show_diagrams_raw])
1046 diagram_edge_labels = getattr(args, "diagram_edge_labels", None)
1047 if diagram_edge_labels is not None:
1048 cmd.extend(["--diagram-edge-labels", diagram_edge_labels])
1049 diagram_state_detail = getattr(args, "diagram_state_detail", None)
1050 if diagram_state_detail is not None:
1051 cmd.extend(["--diagram-state-detail", diagram_state_detail])
1052 diagram_scope = getattr(args, "diagram_scope", None)
1053 if diagram_scope is not None:
1054 cmd.extend(["--diagram-scope", diagram_scope])
1055 if getattr(args, "quiet", False):
1056 cmd.append("--quiet")
1057 if getattr(args, "queue", False):
1058 cmd.append("--queue")
1059 if getattr(args, "no_lock", False):
1060 cmd.append("--no-lock")
1061 for kv in getattr(args, "context", None) or []:
1062 cmd.extend(["--context", kv])
1063 program_md = getattr(args, "program_md", None)
1064 if program_md is not None:
1065 cmd.extend(["--program-md", str(program_md)])
1066 delay = getattr(args, "delay", None)
1067 if delay is not None:
1068 cmd.extend(["--delay", str(delay)])
1069 handoff_threshold = getattr(args, "handoff_threshold", None)
1070 if handoff_threshold is not None:
1071 cmd.extend(["--handoff-threshold", str(handoff_threshold)])
1072 context_limit = getattr(args, "context_limit", None)
1073 if context_limit is not None:
1074 cmd.extend(["--context-limit", str(context_limit)])
1075 if getattr(args, "baseline", False):
1076 cmd.append("--baseline")
1077 baseline_skill = getattr(args, "baseline_skill", None)
1078 if baseline_skill is not None:
1079 cmd.extend(["--baseline-skill", baseline_skill])
1080 items = getattr(args, "items", None)
1081 if items is not None:
1082 cmd.extend(["--items", str(items)])
1084 log_file.parent.mkdir(parents=True, exist_ok=True)
1085 with open(log_file, "w") as log_fh:
1086 process = subprocess.Popen(
1087 cmd,
1088 start_new_session=True,
1089 stdout=log_fh,
1090 stderr=log_fh,
1091 stdin=subprocess.DEVNULL,
1092 )
1094 pid_file.write_text(str(process.pid))
1095 print(
1096 f"Loop {colorize(loop_name, '1')} started in background (PID: {colorize(str(process.pid), '2')})"
1097 )
1098 print(f" Log: {colorize(str(log_file), '2')}")
1099 print(f" Status: {colorize(f'll-loop status {loop_name}', '2')}")
1100 print(f" Stop: {colorize(f'll-loop stop {loop_name}', '2')}")
1101 return 0
1104def run_foreground(
1105 executor: Any,
1106 fsm: FSMLoop,
1107 args: argparse.Namespace,
1108 highlight_color: str = "32",
1109 edge_label_colors: dict[str, str] | None = None,
1110 badges: dict[str, str] | None = None,
1111 mode: str = "run",
1112 instance_id: str | None = None,
1113 running_dir: Path | None = None,
1114 loop_path: Path | None = None,
1115 model: str | None = None,
1116) -> int:
1117 """Run loop with progress display.
1119 Args:
1120 highlight_color: ANSI SGR code for the active FSM state highlight in verbose mode.
1121 edge_label_colors: Optional label→SGR-code mapping for transition edge labels.
1122 badges: Optional glyph-key→string mapping for state type badges in FSM diagrams.
1123 mode: ``"run"`` (default) calls ``executor.run()``; ``"resume"`` calls
1124 ``executor.resume()`` so a resumed loop reuses the same display-wiring
1125 path as a fresh run (BUG-1645). In ``"resume"`` mode a ``None`` result
1126 from ``executor.resume()`` is treated as "nothing to resume": a warning
1127 is logged and exit code 1 is returned before any alt-screen sequences
1128 are emitted.
1130 instance_id: When provided and not a background-spawned child
1131 (``foreground_internal=False``), stdout/stderr are teed to
1132 ``{running_dir}/{instance_id}.log`` with ANSI sequences stripped.
1133 running_dir: Directory for the log file. Defaults to ``.running``
1134 alongside the loops directory.
1135 Returns:
1136 Exit code (0 = success).
1137 """
1138 if mode not in ("run", "resume"):
1139 raise ValueError(f"run_foreground: invalid mode {mode!r}; expected 'run' or 'resume'")
1141 _orig_stdout = sys.stdout
1142 _orig_stderr = sys.stderr
1143 _log_fh: Any = None
1144 if instance_id is not None and not getattr(args, "foreground_internal", False):
1145 _log_path = (running_dir or Path(".running")) / f"{instance_id}.log"
1146 _log_path.parent.mkdir(parents=True, exist_ok=True)
1147 _log_fh = open(_log_path, "w")
1148 sys.stdout = _TeeWriter(_orig_stdout, _log_fh) # type: ignore[assignment]
1149 sys.stderr = _TeeWriter(_orig_stderr, _log_fh) # type: ignore[assignment]
1151 try:
1152 # Create the state feed renderer — encapsulates display state and event handling.
1153 renderer = StateFeedRenderer(
1154 fsm,
1155 args,
1156 highlight_color=highlight_color,
1157 edge_label_colors=edge_label_colors,
1158 badges=badges,
1159 loops_dir=getattr(executor, "loops_dir", Path(".")),
1160 loop_path=loop_path,
1161 model=model,
1162 )
1163 if not renderer.quiet:
1164 print(f"Running loop: {colorize(fsm.name, '1')}")
1165 print(f"Max iterations: {colorize(str(fsm.max_iterations), '2')}")
1166 for key, value in _artifact_lines(fsm, loop_path):
1167 print(f" {key}: {colorize(value, '2')}")
1168 if model is not None:
1169 print(f" model: {colorize(model, '2')}")
1170 print()
1172 # Wire progress display via the EventBus on PersistentExecutor
1173 if not renderer.quiet or renderer.show_diagrams:
1174 if hasattr(executor, "event_bus"):
1175 executor.event_bus.register(renderer.handle_event)
1176 else:
1177 executor._on_event = renderer.handle_event
1179 # Wire follow mode — streams history-formatted events independently of quiet
1180 if getattr(args, "follow", False):
1181 from little_loops.cli.loop.info import _format_history_event
1183 tw = terminal_width()
1184 _verbose = renderer.verbose
1186 def _follow_callback(event: dict[str, Any]) -> None:
1187 line = _format_history_event(event, verbose=_verbose, width=tw)
1188 if line is not None:
1189 print(line, flush=True)
1191 if hasattr(executor, "event_bus"):
1192 executor.event_bus.register(_follow_callback)
1193 else:
1194 _prev_on_event = executor._on_event
1196 def _chained(event: dict[str, Any]) -> None:
1197 if _prev_on_event:
1198 _prev_on_event(event)
1199 _follow_callback(event)
1201 executor._on_event = _chained
1203 # Enter alternate screen buffer when showing diagrams with clear to prevent
1204 # scrollback contamination from diagrams taller than the terminal height.
1205 global _using_alt_screen
1206 if renderer.show_diagrams and renderer.clear_screen and sys.stdout.isatty():
1207 _using_alt_screen = True
1208 print("\033[?1049h\033[H", end="", flush=True)
1209 _install_sigwinch_handler()
1211 try:
1212 if mode == "resume":
1213 result = executor.resume()
1214 # "Nothing to resume" path: no run actually executed, so don't fall
1215 # through to completion-line formatting. Exit cleanly with code 1.
1216 if result is None:
1217 Logger().warning(f"Nothing to resume for: {fsm.name}")
1218 return 1
1219 else:
1220 result = executor.run()
1221 finally:
1222 if _using_alt_screen:
1223 # Reset DECSTBM scroll region BEFORE exiting alt-screen, otherwise
1224 # the main buffer is left with a restricted scroll region (one of
1225 # the Success Metrics for ENH-1642).
1226 print("\033[r", end="", flush=True)
1227 print("\033[?1049l", end="", flush=True)
1228 _using_alt_screen = False
1229 _restore_sigwinch_handler()
1231 if not renderer.quiet:
1232 print()
1233 duration_sec = result.duration_ms / 1000
1234 if duration_sec < 60:
1235 duration_str = f"{duration_sec:.1f}s"
1236 else:
1237 minutes = int(duration_sec // 60)
1238 seconds = duration_sec % 60
1239 duration_str = f"{minutes}m {seconds:.0f}s"
1240 if result.terminated_by == "terminal":
1241 state_colored = colorize(result.final_state, "32")
1242 else:
1243 state_colored = colorize(result.final_state, "38;5;208")
1244 completion_prefix = "Resumed and completed" if mode == "resume" else "Loop completed"
1245 rejection_count = 0
1246 for _t in getattr(getattr(executor, "event_bus", None), "_transports", []):
1247 if hasattr(_t, "get_stats"):
1248 rejection_count += _t.get_stats().get("client_rejections", 0)
1249 suffix = f", {rejection_count} client rejections" if rejection_count > 0 else ""
1251 # Print per-state token/cost table if usage data was collected
1252 run_dir = fsm.context.get("run_dir", "")
1253 if run_dir:
1254 try:
1255 _print_usage_summary(Path(run_dir) / "usage.jsonl")
1256 except Exception:
1257 pass # Non-fatal: display failure shouldn't block exit
1259 print(
1260 f"{completion_prefix}: {state_colored} ({result.iterations} iterations, {duration_str}{suffix})"
1261 )
1263 # FEAT-1822: Print A/B summary if baseline was enabled
1264 if run_dir:
1265 ab_path = Path(run_dir) / "ab.json"
1266 if ab_path.exists():
1267 try:
1268 _print_ab_summary(ab_path)
1269 except Exception:
1270 pass # Non-fatal: display failure shouldn't block exit
1272 return EXIT_CODES.get(result.terminated_by, 1)
1273 finally:
1274 sys.stdout = _orig_stdout
1275 sys.stderr = _orig_stderr
1276 if _log_fh is not None:
1277 _log_fh.close()
1280def _print_usage_summary(usage_path: Path) -> None:
1281 """Print per-state token usage summary from usage.jsonl.
1283 Args:
1284 usage_path: Path to usage.jsonl written by PersistentExecutor
1285 """
1286 from collections import defaultdict
1288 if not usage_path.exists():
1289 return
1290 lines = usage_path.read_text(encoding="utf-8").splitlines()
1291 if not lines:
1292 return
1294 per_state: dict[str, dict[str, Any]] = defaultdict(
1295 lambda: {
1296 "invocations": 0,
1297 "input": 0,
1298 "output": 0,
1299 "cache_read": 0,
1300 "cache_creation": 0,
1301 "model": "unknown",
1302 "est_cost": 0.0,
1303 "has_unknown_model": False,
1304 }
1305 )
1306 for raw in lines:
1307 try:
1308 row = json.loads(raw)
1309 except json.JSONDecodeError:
1310 continue
1311 state = row.get("state", "unknown")
1312 model = row.get("model", "unknown")
1313 inp = row.get("input_tokens", 0)
1314 out = row.get("output_tokens", 0)
1315 cr = row.get("cache_read_tokens", 0)
1316 cc = row.get("cache_creation_tokens", 0)
1317 bucket = per_state[state]
1318 bucket["invocations"] += 1
1319 bucket["input"] += inp
1320 bucket["output"] += out
1321 bucket["cache_read"] += cr
1322 bucket["cache_creation"] += cc
1323 bucket["model"] = model
1324 cost = estimate_cost_usd(model, inp, out, cr, cc)
1325 if cost is None:
1326 bucket["has_unknown_model"] = True
1327 else:
1328 bucket["est_cost"] += cost
1330 if not per_state:
1331 return
1333 print()
1334 print(f"{'state':<24} {'invoc':>5} {'input':>8} {'output':>8} {'cache':>8} {'est_cost':>10}")
1335 print("-" * 68)
1336 for state, b in sorted(per_state.items()):
1337 cache = b["cache_read"] + b["cache_creation"]
1338 cost_str = f"${b['est_cost']:.4f}" if not b["has_unknown_model"] else "n/a"
1339 print(
1340 f"{state:<24} {b['invocations']:>5} {b['input']:>8} {b['output']:>8} {cache:>8} {cost_str:>10}"
1341 )
1342 print()
1345def _print_ab_summary(ab_path: Path) -> None:
1346 """Print A/B comparison summary from ab.json.
1348 Args:
1349 ab_path: Path to ab.json file written by the executor
1350 """
1351 from little_loops.ab_writer import read_ab_json
1353 results = read_ab_json(str(ab_path.parent))
1354 if results is None or not results.per_item:
1355 return
1357 n = len(results.per_item)
1358 harness_pct = results.harness_pass_rate * 100
1359 baseline_pct = results.baseline_pass_rate * 100
1360 delta_pct = results.delta * 100
1362 tokens_ratio = (
1363 results.median_tokens_harness / results.median_tokens_baseline
1364 if results.median_tokens_baseline > 0
1365 else 0
1366 )
1367 dur_ratio = (
1368 results.median_duration_harness / results.median_duration_baseline
1369 if results.median_duration_baseline > 0
1370 else 0
1371 )
1373 def _fmt_dur(ms: float) -> str:
1374 if ms < 1000:
1375 return f"{ms:.0f}ms"
1376 elif ms < 60000:
1377 return f"{ms / 1000:.1f}s"
1378 else:
1379 return f"{ms / 60000:.1f}m"
1381 tokens_dir = "+" if tokens_ratio > 1 else "-"
1382 dur_dir = "+" if dur_ratio > 1 else "-"
1384 print()
1385 print(f"A/B Summary (n={n})")
1386 print(f" Harness pass-rate: {harness_pct:.0f}%")
1387 print(f" Baseline pass-rate: {baseline_pct:.0f}%")
1388 print(f" Delta: {delta_pct:+.0f}%")
1389 print()
1390 print(
1391 f" Median tokens: harness={results.median_tokens_harness} "
1392 f"baseline={results.median_tokens_baseline} "
1393 f"({tokens_dir}{abs(tokens_ratio - 1) * 100:.0f}%)"
1394 )
1395 print(
1396 f" Median duration: harness={_fmt_dur(results.median_duration_harness)} "
1397 f"baseline={_fmt_dur(results.median_duration_baseline)} "
1398 f"({dur_dir}{abs(dur_ratio - 1) * 100:.0f}%)"
1399 )
1401 # Verdict line
1402 quality_verdict = (
1403 "harness wins on quality"
1404 if results.delta > 0
1405 else "baseline wins on quality"
1406 if results.delta < 0
1407 else "no quality difference"
1408 )
1409 cost_verdict = (
1410 f"costs ~{abs(tokens_ratio - 1) * 100:.0f}% more tokens"
1411 if tokens_ratio > 1
1412 else (
1413 f"costs ~{abs(tokens_ratio - 1) * 100:.0f}% fewer tokens"
1414 if tokens_ratio < 1
1415 else "same token cost"
1416 )
1417 )
1418 print(f" Verdict: {quality_verdict}, {cost_verdict}")
1419 print()
1420 print(f"Per-item: {ab_path}")